Merge pull request #10 from runbear-io/feat/parallel-push-progress

perf(sync): parallel blob upload + progress bar for initial import

cmd/bdrive/cmds.go

@@ -27,6 +27,7 @@ func syncCmd() *cobra.Command {
 				return err
 			}
 			defer closeSession(sess)
+			sess.OnProgress = progressReporter()
 			res, err := sess.Cycle(cmd.Context())
 			if err != nil {
 				return err

cmd/bdrive/progress.go

@@ -0,0 +1,58 @@
+package main
+
+import (
+	"fmt"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/mattn/go-isatty"
+
+	"github.com/runbear-io/beardrive/internal/syncer"
+)
+
+// progressReporter returns a syncer.OnProgress callback that shows upload
+// progress: an in-place bar on a TTY, periodic percentage lines otherwise.
+// It's safe to call concurrently from upload workers, and renders are
+// throttled so many small blobs don't thrash the terminal. It stays silent
+// when there's nothing to upload.
+func progressReporter() func(syncer.Progress) {
+	tty := isatty.IsTerminal(os.Stderr.Fd())
+	var mu sync.Mutex
+	var lastDraw time.Time
+	lastPct := -1
+	return func(p syncer.Progress) {
+		if p.Total == 0 {
+			return
+		}
+		mu.Lock()
+		defer mu.Unlock()
+		final := p.Done >= p.Total
+		pct := p.Done * 100 / p.Total
+		if tty {
+			if !final && time.Since(lastDraw) < 100*time.Millisecond {
+				return
+			}
+			lastDraw = time.Now()
+			const w = 24
+			filled := pct * w / 100
+			bar := strings.Repeat("█", filled) + strings.Repeat("░", w-filled)
+			fmt.Fprintf(os.Stderr, "\r  uploading [%s] %3d%%  %d/%d files  %s / %s",
+				bar, pct, p.Done, p.Total, humanBytes(p.Bytes), humanBytes(p.ToBytes))
+			if final {
+				fmt.Fprintln(os.Stderr)
+			}
+			return
+		}
+		// Non-TTY: announce the total once, then a line every ~10% and at the end.
+		if lastPct < 0 {
+			fmt.Fprintf(os.Stderr, "  uploading %d files (%s)\n", p.Total, humanBytes(p.ToBytes))
+		}
+		if final || pct/10 > lastPct/10 {
+			fmt.Fprintf(os.Stderr, "  %d%%  %d/%d files  %s / %s\n",
+				pct, p.Done, p.Total, humanBytes(p.Bytes), humanBytes(p.ToBytes))
+		}
+		lastPct = pct
+	}
+}

cmd/bdrive/sync_run.go

@@ -32,6 +32,7 @@ func startSync(ctx context.Context, folder string, proj config.Project, foregrou
 	if err != nil {
 		return err
 	}
+	sess.OnProgress = progressReporter() // the initial import is the slow one
 	res, err := sess.Cycle(ctx)
 	closeSession(sess)
 	if err != nil {

go.mod

@@ -14,6 +14,7 @@ require (
 	github.com/spf13/cobra v1.10.2
 	github.com/yuin/goldmark v1.8.2
 	golang.org/x/crypto v0.51.0
+	golang.org/x/sync v0.21.0
 	google.golang.org/api v0.284.0
 	modernc.org/sqlite v1.53.0
 )
@@ -79,7 +80,6 @@ require (
 	go.opentelemetry.io/otel/trace v1.43.0 // indirect
 	golang.org/x/net v0.55.0 // indirect
 	golang.org/x/oauth2 v0.36.0 // indirect
-	golang.org/x/sync v0.21.0 // indirect
 	golang.org/x/sys v0.45.0 // indirect
 	golang.org/x/term v0.43.0 // indirect
 	golang.org/x/text v0.37.0 // indirect

internal/syncer/syncer.go

@@ -21,14 +21,30 @@ import (
 	"path"
 	"path/filepath"
 	"strings"
+	"sync/atomic"
 	"time"
 
+	"golang.org/x/sync/errgroup"
+
 	"github.com/runbear-io/beardrive/internal/config"
 	"github.com/runbear-io/beardrive/internal/journal"
 	"github.com/runbear-io/beardrive/internal/remote"
 	"github.com/runbear-io/beardrive/internal/store"
 )
 
+// pushConcurrency bounds how many blobs upload at once. The initial import of
+// many files is latency-bound on serial round-trips, so uploading in parallel
+// is the main speedup.
+const pushConcurrency = 16
+
+// Progress reports upload progress during a cycle's push phase, so the CLI can
+// draw a bar. Total/TotalBytes are set once when the push starts; Done/Bytes
+// climb as blobs finish. Nil OnProgress means no reporting (the daemon).
+type Progress struct {
+	Done, Total    int
+	Bytes, ToBytes int64
+}
+
 // Session ties a working folder to its volume store and (optionally) remote.
 type Session struct {
 	Folder  string
@@ -40,6 +56,10 @@ type Session struct {
 	// Device.Author remains the fallback identity.
 	Account config.Settings
 	Backend remote.Backend // nil = work offline
+	// OnProgress, when set, is called during push with upload progress. It may
+	// be invoked concurrently from upload workers, so it must be safe to call
+	// from multiple goroutines.
+	OnProgress func(Progress)
 }
 
 func (s *Session) mountID() string {
@@ -516,18 +536,44 @@ func (s *Session) push(ctx context.Context, myOps []journal.Op, st *store.SyncSt
 	if st.PushedOps > int64(len(myOps)) {
 		st.PushedOps = int64(len(myOps))
 	}
-	uploaded := map[string]bool{}
+	// Collect the unique, not-yet-pushed blobs to upload (deduped by content
+	// hash). The backend's Put is idempotent and already skips content that's
+	// present remotely (the hub reports it during signing), so we don't pay a
+	// separate existence round-trip per blob.
+	seen := map[string]bool{}
+	type blobJob struct {
+		blob string
+		size int64
+	}
+	var jobs []blobJob
+	var totalBytes int64
 	for _, op := range myOps[st.PushedOps:] {
-		if op.Kind != journal.KindPut || op.Blob == "" || uploaded[op.Blob] {
+		if op.Kind != journal.KindPut || op.Blob == "" || seen[op.Blob] {
 			continue
 		}
-		key := "blobs/" + op.Blob
-		ok, err := s.Backend.Exists(ctx, key)
-		if err != nil {
-			return err
+		seen[op.Blob] = true
+		jobs = append(jobs, blobJob{op.Blob, op.Size})
+		totalBytes += op.Size
+	}
+
+	var done, bytesDone int64
+	report := func() {
+		if s.OnProgress != nil {
+			s.OnProgress(Progress{
+				Done: int(atomic.LoadInt64(&done)), Total: len(jobs),
+				Bytes: atomic.LoadInt64(&bytesDone), ToBytes: totalBytes,
+			})
 		}
-		if !ok {
-			f, err := s.Store.OpenBlob(op.Blob)
+	}
+	report() // announce the total up front (0 / N)
+
+	// Upload blobs in parallel — the initial import is bound on serial
+	// round-trips, not bandwidth, so concurrency is the win.
+	g, gctx := errgroup.WithContext(ctx)
+	g.SetLimit(pushConcurrency)
+	for _, j := range jobs {
+		g.Go(func() error {
+			f, err := s.Store.OpenBlob(j.blob)
 			if err != nil {
 				return err
 			}
@@ -536,13 +582,19 @@ func (s *Session) push(ctx context.Context, myOps []journal.Op, st *store.SyncSt
 				f.Close()
 				return err
 			}
-			err = s.Backend.Put(ctx, key, f, fi.Size())
+			err = s.Backend.Put(gctx, "blobs/"+j.blob, f, fi.Size())
 			f.Close()
 			if err != nil {
 				return err
 			}
-		}
-		uploaded[op.Blob] = true
+			atomic.AddInt64(&done, 1)
+			atomic.AddInt64(&bytesDone, fi.Size())
+			report()
+			return nil
+		})
+	}
+	if err := g.Wait(); err != nil {
+		return err
 	}
 
 	jp := s.Store.JournalPath(s.Device.ID)

internal/syncer/syncer_test.go

@@ -2,9 +2,11 @@ package syncer
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -13,6 +15,40 @@ import (
 	"github.com/runbear-io/beardrive/internal/store"
 )
 
+// TestPushProgress verifies the push phase reports upload progress: the total
+// is the number of unique blobs, Done climbs to that total, and byte totals
+// are populated. (Done isn't strictly ordered across the parallel workers, so
+// we only assert it reaches the total.)
+func TestPushProgress(t *testing.T) {
+	a := newDevice(t, "deva", sharedRemote(t))
+	const n = 25
+	for i := 0; i < n; i++ {
+		write(t, a.Folder, fmt.Sprintf("f%02d.txt", i), fmt.Sprintf("unique content for file %d — pad pad pad", i))
+	}
+	var mu sync.Mutex
+	var total, maxDone int
+	var toBytes int64
+	a.OnProgress = func(p Progress) {
+		mu.Lock()
+		defer mu.Unlock()
+		total = p.Total
+		toBytes = p.ToBytes
+		if p.Done > maxDone {
+			maxDone = p.Done
+		}
+	}
+	cycle(t, a)
+	if total != n {
+		t.Fatalf("progress Total = %d, want %d", total, n)
+	}
+	if maxDone != n {
+		t.Fatalf("progress reached Done = %d, want %d", maxDone, n)
+	}
+	if toBytes == 0 {
+		t.Fatal("progress ToBytes should be > 0")
+	}
+}
+
 // newDevice simulates one device: its own folder, volume store, and identity,
 // all syncing through a shared file:// remote.
 func newDevice(t *testing.T, name string, backend remote.Backend) *Session {