feat(sync): add guarded fingerprint path lock and syncthing progress

Author: acmore

internal/cli/syncthing.go

@@ -8,7 +8,9 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"log/slog"
 	"net/http"
+	"net/url"
 	"os"
 	"os/exec"
 	"os/signal"
@@ -127,6 +129,9 @@ func runSyncthingSync(cmd *cobra.Command, opts *Options, cfg *config.DevEnvironm
 	fmt.Fprintf(cmd.OutOrStdout(), "Remote folder: %s\n", pair.Remote)
 	fmt.Fprintf(cmd.OutOrStdout(), "Local binary: %s\n", localBinary)
 	fmt.Fprintln(cmd.OutOrStdout(), "Press Ctrl+C to stop sync tunnel and local syncthing.")
+	progressCtx, stopProgress := context.WithCancel(context.Background())
+	defer stopProgress()
+	go runSyncthingProgressReporter(progressCtx, cmd.OutOrStdout(), localBase, localKey, remoteBase, remoteKey, folderID, localID, remoteID)
 
 	sigCh := make(chan os.Signal, 1)
 	signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM)
@@ -289,6 +294,49 @@ func syncthingDeviceID(ctx context.Context, base, key string) (string, error) {
 	return payload.MyID, nil
 }
 
+func runSyncthingProgressReporter(ctx context.Context, out io.Writer, localBase, localKey, remoteBase, remoteKey, folderID, localID, remoteID string) {
+	ticker := time.NewTicker(5 * time.Second)
+	defer ticker.Stop()
+	emit := func() {
+		upPct, upNeed, err := syncthingCompletion(ctx, localBase, localKey, folderID, remoteID)
+		if err != nil {
+			slog.Debug("syncthing progress read failed", "side", "local", "error", err)
+			return
+		}
+		downPct, downNeed, err := syncthingCompletion(ctx, remoteBase, remoteKey, folderID, localID)
+		if err != nil {
+			slog.Debug("syncthing progress read failed", "side", "remote", "error", err)
+			return
+		}
+		fmt.Fprintf(out, "Syncthing progress: up %.1f%% (need=%dB), down %.1f%% (need=%dB)\n", upPct, upNeed, downPct, downNeed)
+	}
+	emit()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			emit()
+		}
+	}
+}
+
+func syncthingCompletion(ctx context.Context, base, key, folderID, deviceID string) (float64, int64, error) {
+	path := fmt.Sprintf("/rest/db/completion?folder=%s&device=%s", url.QueryEscape(folderID), url.QueryEscape(deviceID))
+	body, err := syncthingAPIRequestWithContext(ctx, http.MethodGet, base, key, path, nil, "")
+	if err != nil {
+		return 0, 0, err
+	}
+	var payload struct {
+		Completion float64 `json:"completion"`
+		NeedBytes  int64   `json:"needBytes"`
+	}
+	if err := json.Unmarshal(body, &payload); err != nil {
+		return 0, 0, err
+	}
+	return payload.Completion, payload.NeedBytes, nil
+}
+
 func folderTypesForMode(mode string) (local, remote string) {
 	switch mode {
 	case "up":
@@ -407,7 +455,11 @@ func syncthingPost(base, key, path string, body []byte) error {
 }
 
 func syncthingAPIRequest(method, base, key, path string, body []byte, contentType string) ([]byte, error) {
-	req, err := http.NewRequest(method, base+path, bytes.NewReader(body))
+	return syncthingAPIRequestWithContext(context.Background(), method, base, key, path, body, contentType)
+}
+
+func syncthingAPIRequestWithContext(ctx context.Context, method, base, key, path string, body []byte, contentType string) ([]byte, error) {
+	req, err := http.NewRequestWithContext(ctx, method, base+path, bytes.NewReader(body))
 	if err != nil {
 		return nil, err
 	}

internal/cli/syncthing_test.go

@@ -1,6 +1,11 @@
 package cli
 
-import "testing"
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
 
 func TestSyncthingObjectArray(t *testing.T) {
 	arr, err := syncthingObjectArray(map[string]any{"devices": []any{map[string]any{"deviceID": "abc"}}}, "devices")
@@ -25,3 +30,30 @@ func TestSyncthingObjectMapRejectsWrongType(t *testing.T) {
 		t.Fatal("expected type error")
 	}
 }
+
+func TestSyncthingCompletion(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/rest/db/completion" {
+			t.Fatalf("unexpected path %s", r.URL.Path)
+		}
+		if got := r.URL.Query().Get("folder"); got != "okdev-sess" {
+			t.Fatalf("unexpected folder query %q", got)
+		}
+		if got := r.URL.Query().Get("device"); got != "REMOTEID" {
+			t.Fatalf("unexpected device query %q", got)
+		}
+		if got := r.Header.Get("X-API-Key"); got != "k" {
+			t.Fatalf("unexpected API key %q", got)
+		}
+		_, _ = w.Write([]byte(`{"completion":99.5,"needBytes":1234}`))
+	}))
+	defer srv.Close()
+
+	pct, need, err := syncthingCompletion(context.Background(), srv.URL, "k", "okdev-sess", "REMOTEID")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if pct != 99.5 || need != 1234 {
+		t.Fatalf("unexpected completion values pct=%v need=%d", pct, need)
+	}
+}

internal/sync/native.go

@@ -50,6 +50,8 @@ var uploadFingerprintCache = struct {
 	m: map[string]uploadFingerprintEntry{},
 }
 
+var uploadPathLocks sync.Map
+
 func ParsePairs(configured []string, defaultRemote string) ([]Pair, error) {
 	if len(configured) == 0 {
 		return []Pair{{Local: ".", Remote: defaultRemote}}, nil
@@ -200,43 +202,47 @@ func syncUpPath(ctx context.Context, k Client, namespace, pod, podInstance strin
 
 	cacheKey := uploadFingerprintCacheKey(namespace, pod, p, excludes)
 	if !st.IsDir() {
-		fingerprint := fmt.Sprintf("file:%d:%d", st.Size(), st.ModTime().UnixNano())
-		if !force && uploadFingerprintMatches(cacheKey, podInstance, fingerprint) {
-			return pathStats{Skipped: true}, nil
-		}
-		if err := k.CopyToPod(ctx, namespace, absLocal, pod, p.Remote); err != nil {
-			return pathStats{}, err
-		}
-		setUploadFingerprint(cacheKey, podInstance, fingerprint)
-		return pathStats{UploadBytes: st.Size()}, nil
+		return withUploadKeyLock(cacheKey, func() (pathStats, error) {
+			fingerprint := fmt.Sprintf("file:%d:%d", st.Size(), st.ModTime().UnixNano())
+			if !force && uploadFingerprintMatches(cacheKey, podInstance, fingerprint) {
+				return pathStats{Skipped: true}, nil
+			}
+			if err := k.CopyToPod(ctx, namespace, absLocal, pod, p.Remote); err != nil {
+				return pathStats{}, err
+			}
+			setUploadFingerprint(cacheKey, podInstance, fingerprint)
+			return pathStats{UploadBytes: st.Size()}, nil
+		})
 	}
 
 	fingerprint, err := localDirFingerprint(absLocal, excludes)
 	if err != nil {
 		return pathStats{}, err
 	}
-	if !force && uploadFingerprintMatches(cacheKey, podInstance, fingerprint) {
-		return pathStats{Skipped: true}, nil
-	}
+	return withUploadKeyLock(cacheKey, func() (pathStats, error) {
+		if !force && uploadFingerprintMatches(cacheKey, podInstance, fingerprint) {
+			return pathStats{Skipped: true}, nil
+		}
 
-	stream, waitTar, err := startLocalTarStream(ctx, absLocal, excludes)
-	if err != nil {
-		return pathStats{}, err
-	}
-	defer stream.Close()
+		stream, waitTar, err := startLocalTarStream(ctx, absLocal, excludes)
+		if err != nil {
+			return pathStats{}, err
+		}
+		defer stream.Close()
 
-	countingStream := &countingReader{Reader: stream}
-	if err := k.ExtractTarToPod(ctx, namespace, pod, p.Remote, countingStream); err != nil {
-		if tarErr := waitTar(); tarErr != nil {
-			return pathStats{}, tarErr
+		countingStream := &countingReader{Reader: stream}
+		if err := k.ExtractTarToPod(ctx, namespace, pod, p.Remote, countingStream); err != nil {
+			if tarErr := waitTar(); tarErr != nil {
+				return pathStats{}, tarErr
+			}
+			return pathStats{}, err
 		}
-		return pathStats{}, err
-	}
-	if err := waitTar(); err != nil {
-		return pathStats{}, err
-	}
-	setUploadFingerprint(cacheKey, podInstance, fingerprint)
-	return pathStats{UploadBytes: countingStream.BytesRead()}, nil
+		if err := waitTar(); err != nil {
+			return pathStats{}, err
+		}
+		setUploadFingerprint(cacheKey, podInstance, fingerprint)
+		return pathStats{UploadBytes: countingStream.BytesRead()}, nil
+	})
 }
 
 func syncDownPath(ctx context.Context, k Client, namespace, pod string, p Pair, excludes []string) (pathStats, error) {
@@ -436,6 +442,14 @@ func uploadFingerprintCacheLen() int {
 	return len(uploadFingerprintCache.m)
 }
 
+func withUploadKeyLock(key string, fn func() (pathStats, error)) (pathStats, error) {
+	v, _ := uploadPathLocks.LoadOrStore(key, &sync.Mutex{})
+	mu := v.(*sync.Mutex)
+	mu.Lock()
+	defer mu.Unlock()
+	return fn()
+}
+
 func localDirFingerprint(root string, excludes []string) (string, error) {
 	var files int64
 	var bytes int64

internal/sync/native_test.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"sync"
 	"testing"
 	"time"
 )
@@ -157,3 +158,64 @@ func TestRunOnceWithOptionsForceBypassesFingerprint(t *testing.T) {
 		t.Fatalf("expected force sync to bypass cache, skipped=%d", stats.SkippedPaths)
 	}
 }
+
+type countingCopyClient struct {
+	mu    sync.Mutex
+	calls int
+}
+
+func (c *countingCopyClient) CopyToPod(context.Context, string, string, string, string) error {
+	c.mu.Lock()
+	c.calls++
+	c.mu.Unlock()
+	time.Sleep(50 * time.Millisecond)
+	return nil
+}
+func (c *countingCopyClient) CopyFromPod(context.Context, string, string, string, string) error {
+	return nil
+}
+func (c *countingCopyClient) ExtractTarToPod(context.Context, string, string, string, io.Reader) error {
+	return nil
+}
+func (c *countingCopyClient) StreamFromPod(context.Context, string, string, string, io.Writer) error {
+	return nil
+}
+func (c *countingCopyClient) ExecSh(context.Context, string, string, string) ([]byte, error) {
+	return []byte("pod-instance"), nil
+}
+
+func TestRunOnceWithReportConcurrentUploadsShareFingerprintLock(t *testing.T) {
+	ResetUploadFingerprintCache()
+	tmp := t.TempDir()
+	local := filepath.Join(tmp, "file.txt")
+	if err := os.WriteFile(local, []byte("hello"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+
+	client := &countingCopyClient{}
+	pairs := []Pair{{Local: local, Remote: "/workspace/file.txt"}}
+
+	var wg sync.WaitGroup
+	errCh := make(chan error, 2)
+	run := func() {
+		defer wg.Done()
+		_, err := RunOnceWithReport(context.Background(), "up", client, "ns", "pod", pairs, nil)
+		errCh <- err
+	}
+	wg.Add(2)
+	go run()
+	go run()
+	wg.Wait()
+	close(errCh)
+	for err := range errCh {
+		if err != nil {
+			t.Fatalf("unexpected sync error: %v", err)
+		}
+	}
+
+	client.mu.Lock()
+	defer client.mu.Unlock()
+	if client.calls != 1 {
+		t.Fatalf("expected one upload call with lock-protected fingerprint cache, got %d", client.calls)
+	}
+}