From 211e09554232f508e12078a95c633b978467b045 Mon Sep 17 00:00:00 2001
From: Quadrubo <71718414+Quadrubo@users.noreply.github.com>
Date: Thu, 16 Apr 2026 21:40:12 +0200
Subject: [PATCH] fix: prevent duplicate target forwards under concurrent
 uploads

---
 server/internal/server/server.go      | 66 +++++++++++++--------------
 server/internal/server/server_test.go | 24 ++++++++++
 2 files changed, 57 insertions(+), 33 deletions(-)

diff --git a/server/internal/server/server.go b/server/internal/server/server.go
index 7a23065..f3c3bf7 100644
--- a/server/internal/server/server.go
+++ b/server/internal/server/server.go
@@ -86,6 +86,21 @@ func (s *Server) handleUpload(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// Validate target exists for device
+	t, ok := s.targets[deviceID]
+	if !ok {
+		slog.Error("no target configured", "device", deviceID)
+		http.Error(w, "no target for device", http.StatusBadRequest)
+		return
+	}
+
+	// Validate source format
+	sourceFormat := r.Header.Get("X-Source-Format")
+	if sourceFormat == "" {
+		http.Error(w, "missing X-Source-Format header", http.StatusBadRequest)
+		return
+	}
+
 	// Parse multipart form
 	if err := r.ParseMultipartForm(32 << 20); err != nil {
 		http.Error(w, "invalid multipart form", http.StatusBadRequest)
@@ -104,36 +119,34 @@ func (s *Server) handleUpload(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// Deduplicate
+	// Deduplicate - claim the hash in a transaction before forwarding to
+	// the target. On failure the transaction is rolled back so retries work.
 	hash := sha256.Sum256(data)
 	hashHex := hex.EncodeToString(hash[:])
 
-	var exists bool
-	err = s.db.QueryRow("SELECT EXISTS(SELECT 1 FROM uploads WHERE sha256 = ?)", hashHex).Scan(&exists)
+	tx, err := s.db.Begin()
 	if err != nil {
 		slog.Error("database error", "error", err)
 		http.Error(w, "internal error", http.StatusInternalServerError)
 		return
 	}
-	if exists {
-		slog.Info("duplicate", "file", header.Filename, "sha256", hashHex[:12], "client", client.ID)
-		w.WriteHeader(http.StatusOK)
-		_, _ = fmt.Fprintln(w, "duplicate")
-		return
-	}
+	defer func() { _ = tx.Rollback() }()
 
-	// Forward to target
-	t, ok := s.targets[deviceID]
-	if !ok {
-		slog.Error("no target configured", "device", deviceID)
-		http.Error(w, "no target for device", http.StatusBadRequest)
+	result, err := tx.Exec(
+		"INSERT OR IGNORE INTO uploads (sha256, device_id, client_id, filename, uploaded_at) VALUES (?, ?, ?, ?, ?)",
+		hashHex, deviceID, client.ID, header.Filename, time.Now().UTC(),
+	)
+	if err != nil {
+		slog.Error("database error", "error", err)
+		http.Error(w, "internal error", http.StatusInternalServerError)
 		return
 	}
 
-	// Convert format if needed
-	sourceFormat := r.Header.Get("X-Source-Format")
-	if sourceFormat == "" {
-		http.Error(w, "missing X-Source-Format header", http.StatusBadRequest)
+	rows, _ := result.RowsAffected()
+	if rows == 0 {
+		slog.Info("duplicate", "file", header.Filename, "sha256", hashHex[:12], "client", client.ID)
+		w.WriteHeader(http.StatusOK)
+		_, _ = fmt.Fprintln(w, "duplicate")
 		return
 	}
 
@@ -165,25 +178,12 @@ func (s *Server) handleUpload(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// Record successful upload
-	result, err := s.db.Exec(
-		"INSERT OR IGNORE INTO uploads (sha256, device_id, client_id, filename, uploaded_at) VALUES (?, ?, ?, ?, ?)",
-		hashHex, deviceID, client.ID, header.Filename, time.Now().UTC(),
-	)
-	if err != nil {
-		slog.Error("failed to record upload", "error", err)
+	if err := tx.Commit(); err != nil {
+		slog.Error("failed to commit upload", "error", err)
 		http.Error(w, "internal error", http.StatusInternalServerError)
 		return
 	}
 
-	rows, _ := result.RowsAffected()
-	if rows == 0 {
-		slog.Info("duplicate (concurrent)", "source", header.Filename, "file", newFilename, "sha256", hashHex[:12], "client", client.ID)
-		w.WriteHeader(http.StatusOK)
-		_, _ = fmt.Fprintln(w, "duplicate")
-		return
-	}
-
 	slog.Info("uploaded", "source", header.Filename, "file", newFilename, "sha256", hashHex[:12], "client", client.ID, "device", deviceID)
 	w.WriteHeader(http.StatusCreated)
 	_, _ = fmt.Fprintln(w, "uploaded")
diff --git a/server/internal/server/server_test.go b/server/internal/server/server_test.go
index d0a5714..4c58910 100644
--- a/server/internal/server/server_test.go
+++ b/server/internal/server/server_test.go
@@ -139,6 +139,30 @@ func TestUpload_TargetFailure(t *testing.T) {
 	assert.Equal(t, http.StatusBadGateway, rec.Code)
 }
 
+func TestUpload_RetryAfterTargetFailure(t *testing.T) {
+	mock := &mockTarget{err: fmt.Errorf("connection refused")}
+	db, err := sql.Open("sqlite", ":memory:")
+	require.NoError(t, err)
+	require.NoError(t, InitDB(db))
+	t.Cleanup(func() { _ = db.Close() })
+
+	cfg := &config.Config{
+		Clients: []config.Client{
+			{ID: "test-client", Token: "valid-token", AllowedDeviceIDs: []string{"dev-1"}},
+		},
+	}
+	srv := New(cfg, db, map[string]target.Target{"dev-1": mock})
+
+	// First attempt fails
+	rec1 := serveUpload(srv, "valid-token", "dev-1", "gpx_1.1", "track.gpx", validGPX)
+	require.Equal(t, http.StatusBadGateway, rec1.Code)
+
+	// Fix the target and retry - should succeed, not be treated as duplicate
+	mock.err = nil
+	rec2 := serveUpload(srv, "valid-token", "dev-1", "gpx_1.1", "track.gpx", validGPX)
+	assert.Equal(t, http.StatusCreated, rec2.Code)
+}
+
 func TestUpload_NoTargetForDevice(t *testing.T) {
 	srv := setupTestServer(t, nil)
 	rec := serveUpload(srv, "limited-token", "dev-2", "gpx_1.1", "track.gpx", validGPX)