From 97481ee750329a97a3a08b4f15ffadacae1166c3 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 24 Jul 2026 11:58:38 +0000
Subject: [PATCH 1/3] agent: per-destination sync workers + rclone transfer
 timeouts (#160)

The scheduler was a single serial worker with no per-run time bound: a
hung rclone froze every destination, every volume, the peer syncs, and
the index cadences until an operator killed the process by hand (F25).

Split sync execution off the tick goroutine into a per-destination
dispatcher. The concurrency unit is the destination: at most one sync
per destination is in flight (two volumes targeting the same endpoint
serialise on it), different destinations run concurrently, and an
overall semaphore (default 4) bounds total parallelism. A slow or wedged
destination is now confined to its own worker and can never delay the
tick loop, index runs, peer syncs, or LAN pairs. The per-pair
`skipped reason="in-flight sync run"` discipline and the pre-sync index
ordering are preserved.

Bound every automatic transfer two ways. rclone carries explicit
--contimeout/--timeout on each invocation. A squirrel-side no-progress
guard (stallGuard) watches rclone's per-second stats and, if
transferred+checked+bytes stops advancing for StallTimeout (default
10m), kills the child and fails the run with a diagnosable "stalled"
error that composes with #157's stderr capture. Checks count as
progress so a long verification pass is never mistaken for a stall; a
slow-but-moving transfer of a large volume is never killed. The guard is
off for the foreground CLI (a human can interrupt) and set by the agent.

In-flight visibility is already per-pair via the runs table (the TUI's
active-runs block); the timeouts make those rows truthful by failing a
stuck run instead of leaving it "running" forever.

No schema change.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014Skm2Qm6wWUB7dD4CpPTz7
---
 agent/dispatcher.go      | 148 ++++++++++++++++++++++++++++
 agent/dispatcher_test.go | 128 ++++++++++++++++++++++++
 agent/scheduler.go       |  61 +++++-------
 agent/scheduler_test.go  |  19 +++-
 cmd/squirrel/agent.go    |   4 +
 sync/rclone.go           | 207 +++++++++++++++++++++++++++++++++++----
 sync/rclone_test.go      |  59 +++++++++--
 7 files changed, 560 insertions(+), 66 deletions(-)
 create mode 100644 agent/dispatcher.go
 create mode 100644 agent/dispatcher_test.go

diff --git a/agent/dispatcher.go b/agent/dispatcher.go
new file mode 100644
index 0000000..70c1053
--- /dev/null
+++ b/agent/dispatcher.go
@@ -0,0 +1,148 @@
+package agent
+
+import (
+	"context"
+	"log/slog"
+	"sync"
+	"time"
+
+	"github.com/mbertschler/squirrel/config"
+	"github.com/mbertschler/squirrel/store"
+)
+
+// defaultMaxParallelSyncs bounds how many automatic syncs run at once
+// across all destinations. Per-destination serialisation already caps
+// parallelism at the number of configured destinations; this is the
+// overall ceiling so a host with many destinations can't spawn one rclone
+// per destination simultaneously and thrash its uplink and CPU. Four
+// covers the reference household — a NAS pushing photos/docs/media to
+// cloudbox + s3archive + kopia-mirror + htpc — without a LAN pair ever
+// queueing behind a slow offsite, while still bounding a pathological
+// config.
+const defaultMaxParallelSyncs = 4
+
+// syncDispatcher runs (volume, destination) syncs off the scheduler's tick
+// goroutine so a slow or wedged destination can never delay the tick loop,
+// index runs, peer syncs, or syncs to other destinations (#160, F25). The
+// concurrency unit is the destination: at most one sync per destination is
+// in flight at a time (inflight), and an overall semaphore (sem) bounds how
+// many run concurrently across destinations.
+//
+// The pre-sync index ordering the scheduler guarantees is unaffected: the
+// tick body still runs a volume's index to completion before it hands that
+// volume's due syncs to the dispatcher.
+type syncDispatcher struct {
+	run    SyncRunner
+	logger *slog.Logger
+	now    func() time.Time
+
+	sem chan struct{}
+
+	mu       sync.Mutex
+	inflight map[string]bool
+	wg       sync.WaitGroup
+}
+
+func newSyncDispatcher(run SyncRunner, logger *slog.Logger, now func() time.Time, maxParallel int) *syncDispatcher {
+	if maxParallel <= 0 {
+		maxParallel = defaultMaxParallelSyncs
+	}
+	return &syncDispatcher{
+		run:      run,
+		logger:   logger,
+		now:      now,
+		sem:      make(chan struct{}, maxParallel),
+		inflight: make(map[string]bool),
+	}
+}
+
+// dispatch launches the sync for (vol, destName) on destName's worker,
+// unless a sync to that destination is already in flight (a per-pair skip
+// log, matching the CLI wording) or no runner is configured. It never
+// blocks on the transfer: the rclone call runs in a goroutine bounded by
+// the semaphore, so the caller — the tick loop — returns at once.
+func (d *syncDispatcher) dispatch(ctx context.Context, vol *config.Volume, destName string) {
+	if d.run == nil {
+		d.logger.Info("scheduler.skipped",
+			"kind", "sync", "volume", vol.Name, "destination", destName,
+			"reason", "sync runner not configured")
+		return
+	}
+	if !d.claim(destName) {
+		d.logger.Info("scheduler.skipped",
+			"kind", "sync", "volume", vol.Name, "destination", destName,
+			"reason", "in-flight sync run")
+		return
+	}
+	d.logger.Info("scheduler.kicked",
+		"kind", "sync", "volume", vol.Name, "destination", destName)
+	d.wg.Add(1)
+	go d.runOne(ctx, vol, destName)
+}
+
+// runOne is the per-destination worker body: take an overall slot (or bail
+// if the scheduler is shutting down), run the sync, and always release both
+// the slot and the destination claim.
+func (d *syncDispatcher) runOne(ctx context.Context, vol *config.Volume, destName string) {
+	defer d.wg.Done()
+	defer d.release(destName)
+	select {
+	case d.sem <- struct{}{}:
+		defer func() { <-d.sem }()
+	case <-ctx.Done():
+		return
+	}
+	if ctx.Err() != nil {
+		return
+	}
+	d.execute(ctx, vol, destName)
+}
+
+// execute invokes the sync runner and emits the finished/error logs,
+// mirroring the scheduler's kicked/finished/error discipline. A sync that
+// stalls is failed by the runner's transfer timeout (rclone is killed);
+// its diagnosable error lands here and in the run row.
+func (d *syncDispatcher) execute(ctx context.Context, vol *config.Volume, destName string) {
+	start := d.now()
+	rep := d.run(ctx, vol, destName)
+	duration := d.now().Sub(start)
+	status := rep.Status
+	if status == "" {
+		status = store.RunStatusFailed
+	}
+	d.logger.Info("scheduler.finished",
+		"kind", "sync", "volume", vol.Name, "destination", destName,
+		"run_id", rep.RunID, "status", status,
+		"duration_ms", duration.Milliseconds(),
+	)
+	if rep.Err != nil {
+		d.logger.Error("scheduler.error",
+			"kind", "sync", "volume", vol.Name, "destination", destName,
+			"run_id", rep.RunID, "err", rep.Err.Error())
+	}
+}
+
+// claim marks destName in flight, returning false when a sync to it is
+// already running. release clears the mark.
+func (d *syncDispatcher) claim(destName string) bool {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	if d.inflight[destName] {
+		return false
+	}
+	d.inflight[destName] = true
+	return true
+}
+
+func (d *syncDispatcher) release(destName string) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	delete(d.inflight, destName)
+}
+
+// wait blocks until every dispatched sync has finished. The scheduler
+// calls it during shutdown, after the run context is cancelled (which
+// kills any in-flight rclone child), so it returns promptly.
+func (d *syncDispatcher) wait() {
+	d.wg.Wait()
+}
diff --git a/agent/dispatcher_test.go b/agent/dispatcher_test.go
new file mode 100644
index 0000000..c520775
--- /dev/null
+++ b/agent/dispatcher_test.go
@@ -0,0 +1,128 @@
+package agent
+
+import (
+	"bytes"
+	"context"
+	"log/slog"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/mbertschler/squirrel/config"
+	"github.com/mbertschler/squirrel/store"
+)
+
+func discardLogger() *slog.Logger { return slog.New(slog.DiscardHandler) }
+
+// TestSyncDispatcherRunsDestinationsConcurrently proves the core F25 fix:
+// two destinations run at the same time, so a slow offsite can never delay
+// a LAN pair. Both runner invocations must be in flight before either is
+// released.
+func TestSyncDispatcherRunsDestinationsConcurrently(t *testing.T) {
+	var active, maxActive atomic.Int32
+	entered := make(chan struct{}, 2)
+	release := make(chan struct{})
+	run := func(ctx context.Context, vol *config.Volume, destName string) SyncRunReport {
+		n := active.Add(1)
+		for {
+			old := maxActive.Load()
+			if n <= old || maxActive.CompareAndSwap(old, n) {
+				break
+			}
+		}
+		entered <- struct{}{}
+		<-release
+		active.Add(-1)
+		return SyncRunReport{Status: store.RunStatusSuccess}
+	}
+	d := newSyncDispatcher(run, discardLogger(), time.Now, 4)
+	vol := &config.Volume{Name: "photos"}
+	d.dispatch(context.Background(), vol, "cloudbox")
+	d.dispatch(context.Background(), vol, "htpc")
+	<-entered // both must be running before we release either
+	<-entered
+	close(release)
+	d.wait()
+	if got := maxActive.Load(); got != 2 {
+		t.Fatalf("max concurrent syncs = %d; want 2 (different destinations run concurrently)", got)
+	}
+}
+
+// TestSyncDispatcherOneInFlightPerDestination proves the concurrency unit
+// is the destination: a second volume targeting a destination that already
+// has a sync in flight is skipped with the per-pair "in-flight sync run"
+// log, and its runner never fires while the first is running.
+func TestSyncDispatcherOneInFlightPerDestination(t *testing.T) {
+	var calls atomic.Int32
+	entered := make(chan struct{}, 1)
+	release := make(chan struct{})
+	run := func(ctx context.Context, vol *config.Volume, destName string) SyncRunReport {
+		calls.Add(1)
+		entered <- struct{}{}
+		<-release
+		return SyncRunReport{Status: store.RunStatusSuccess}
+	}
+	buf := &bytes.Buffer{}
+	logger := slog.New(slog.NewTextHandler(buf, nil))
+	d := newSyncDispatcher(run, logger, time.Now, 4)
+
+	d.dispatch(context.Background(), &config.Volume{Name: "photos"}, "cloudbox")
+	<-entered // the first sync has claimed cloudbox and is running
+	d.dispatch(context.Background(), &config.Volume{Name: "docs"}, "cloudbox")
+
+	if got := calls.Load(); got != 1 {
+		t.Fatalf("runner called %d times; want 1 (second dispatch to a busy destination must skip)", got)
+	}
+	log := buf.String()
+	if !strings.Contains(log, `reason="in-flight sync run"`) || !strings.Contains(log, "volume=docs") {
+		t.Fatalf("expected a per-pair in-flight skip for docs→cloudbox, got:\n%s", log)
+	}
+	close(release)
+	d.wait()
+}
+
+// TestSyncDispatcherBoundsOverallParallelism proves the semaphore ceiling:
+// with a single slot, a second destination waits for the first to free the
+// slot rather than running immediately.
+func TestSyncDispatcherBoundsOverallParallelism(t *testing.T) {
+	var bCalled atomic.Bool
+	aEntered := make(chan struct{})
+	release := make(chan struct{})
+	run := func(ctx context.Context, vol *config.Volume, destName string) SyncRunReport {
+		if destName == "a" {
+			close(aEntered)
+			<-release
+		} else {
+			bCalled.Store(true)
+		}
+		return SyncRunReport{Status: store.RunStatusSuccess}
+	}
+	d := newSyncDispatcher(run, discardLogger(), time.Now, 1)
+	vol := &config.Volume{Name: "v"}
+	d.dispatch(context.Background(), vol, "a")
+	<-aEntered // a holds the only slot
+	d.dispatch(context.Background(), vol, "b")
+	time.Sleep(20 * time.Millisecond) // give b's worker time to reach the slot wait
+	if bCalled.Load() {
+		t.Fatal("b ran while the single parallelism slot was held by a")
+	}
+	close(release)
+	d.wait()
+	if !bCalled.Load() {
+		t.Fatal("b never ran after the slot was freed")
+	}
+}
+
+// TestSyncDispatcherNilRunnerSkips covers the index-only host: no runner
+// wired, so a dispatch logs a skip and starts no worker.
+func TestSyncDispatcherNilRunnerSkips(t *testing.T) {
+	buf := &bytes.Buffer{}
+	logger := slog.New(slog.NewTextHandler(buf, nil))
+	d := newSyncDispatcher(nil, logger, time.Now, 4)
+	d.dispatch(context.Background(), &config.Volume{Name: "v"}, "dest")
+	d.wait()
+	if !strings.Contains(buf.String(), "sync runner not configured") {
+		t.Fatalf("expected a nil-runner skip log, got:\n%s", buf.String())
+	}
+}
diff --git a/agent/scheduler.go b/agent/scheduler.go
index 17d3883..0394688 100644
--- a/agent/scheduler.go
+++ b/agent/scheduler.go
@@ -30,7 +30,7 @@ const DefaultSchedulerTick = 30 * time.Second
 type scheduler struct {
 	store     *store.Store
 	volumes   map[string]*config.Volume
-	syncRun   SyncRunner
+	dispatch  *syncDispatcher
 	logger    *slog.Logger
 	locks     lockHolder
 	tickEvery time.Duration
@@ -60,7 +60,7 @@ func newScheduler(srv *Server, tickEvery time.Duration, now func() time.Time) *s
 	return &scheduler{
 		store:     srv.store,
 		volumes:   srv.cfg.Volumes,
-		syncRun:   srv.cfg.SyncRunner,
+		dispatch:  newSyncDispatcher(srv.cfg.SyncRunner, srv.cfg.Logger, now, defaultMaxParallelSyncs),
 		logger:    srv.cfg.Logger,
 		locks:     srv.router,
 		tickEvery: tickEvery,
@@ -103,9 +103,12 @@ func (s *scheduler) run(ctx context.Context) {
 	for {
 		select {
 		case <-ctx.Done():
-			// Drain in-flight hooks before returning so Serve's shutdown
-			// wait doesn't race a hook goroutine writing its outcome.
+			// Drain in-flight hooks and per-destination sync workers before
+			// returning so Serve's shutdown wait doesn't race a goroutine
+			// writing its outcome. ctx is already cancelled, so any rclone
+			// child has been killed and the workers return promptly.
 			s.hooks.wait()
+			s.dispatch.wait()
 			return
 		case <-t.C:
 			s.tick(ctx)
@@ -460,17 +463,21 @@ func indexRunStatus(rep index.Report, err error) (string, bool) {
 	return store.RunStatusSuccess, true
 }
 
-// runSync kicks one (volume, destination) sync via the configured
-// SyncRunner. The pre-check against the runs table is the in-flight
-// detection the issue specifies; the SyncRunner's downstream call
-// (sync.RunPair) adds its own atomic BeginSyncRunIfClear gate, so a
-// race that sneaks past the pre-check still produces a clean skip
-// rather than a duplicate run.
+// runSync evaluates the per-pair in-flight gate, then hands the sync to
+// the per-destination dispatcher (#160). The DB check catches a sync of
+// this exact pair already running — a concurrent CLI `squirrel sync`, or a
+// stale row not yet reaped — producing the same clean per-pair skip the
+// serial scheduler did; the dispatcher additionally serialises different
+// volumes that target the same destination and confines a slow transfer to
+// its own worker. The SyncRunner's downstream call (sync.RunPair) adds its
+// own atomic BeginSyncRunIfClear gate, so a race that sneaks past the
+// pre-check still produces a clean skip rather than a duplicate run.
 //
-// A nil SyncRunner surfaces as a clean skip rather than an error so
-// an agent running pure index-only schedules can omit the sync wiring
-// (and the rclone dependency that comes with it) without the
-// scheduler logging at error level on each tick.
+// Dispatch is non-blocking: a slow or wedged destination is confined to its
+// own worker and never stalls the tick loop or syncs to other destinations,
+// which is the freeze F25 recorded. A nil SyncRunner surfaces as a clean
+// skip inside the dispatcher so an agent running pure index-only schedules
+// can omit the sync wiring without logging at error level each tick.
 func (s *scheduler) runSync(ctx context.Context, vol *config.Volume, volumeID int64, destName string) {
 	running, err := s.store.HasRunningRun(ctx, store.RunKindSync, volumeID, destName)
 	if err != nil {
@@ -485,29 +492,5 @@ func (s *scheduler) runSync(ctx context.Context, vol *config.Volume, volumeID in
 			"reason", "in-flight sync run")
 		return
 	}
-	if s.syncRun == nil {
-		s.logger.Info("scheduler.skipped",
-			"kind", "sync", "volume", vol.Name, "destination", destName,
-			"reason", "sync runner not configured")
-		return
-	}
-	s.logger.Info("scheduler.kicked",
-		"kind", "sync", "volume", vol.Name, "destination", destName)
-	start := s.now()
-	rep := s.syncRun(ctx, vol, destName)
-	duration := s.now().Sub(start)
-	status := rep.Status
-	if status == "" {
-		status = store.RunStatusFailed
-	}
-	s.logger.Info("scheduler.finished",
-		"kind", "sync", "volume", vol.Name, "destination", destName,
-		"run_id", rep.RunID, "status", status,
-		"duration_ms", duration.Milliseconds(),
-	)
-	if rep.Err != nil {
-		s.logger.Error("scheduler.error",
-			"kind", "sync", "volume", vol.Name, "destination", destName,
-			"run_id", rep.RunID, "err", rep.Err.Error())
-	}
+	s.dispatch.dispatch(ctx, vol, destName)
 }
diff --git a/agent/scheduler_test.go b/agent/scheduler_test.go
index 9a4a8d6..2d03553 100644
--- a/agent/scheduler_test.go
+++ b/agent/scheduler_test.go
@@ -174,7 +174,7 @@ func (f *schedulerFixture) scheduler() *scheduler {
 	return &scheduler{
 		store:     f.srv.store,
 		volumes:   f.srv.cfg.Volumes,
-		syncRun:   f.srv.cfg.SyncRunner,
+		dispatch:  newSyncDispatcher(f.srv.cfg.SyncRunner, f.srv.cfg.Logger, f.clock.Now, defaultMaxParallelSyncs),
 		logger:    f.srv.cfg.Logger,
 		locks:     f.srv.router,
 		tickEvery: time.Second,
@@ -310,6 +310,7 @@ func TestSchedulerSyncRunsPreSyncIndexFirst(t *testing.T) {
 	ctx := context.Background()
 
 	sch.tick(ctx)
+	sch.dispatch.wait() // syncs run on per-destination workers; drain them
 
 	runs, err := f.store.ListRuns(ctx, store.ListRunsOpts{})
 	if err != nil {
@@ -437,6 +438,7 @@ func TestSchedulerSkipsWhenInFlightSyncRun(t *testing.T) {
 
 	sch := f.scheduler()
 	sch.tick(context.Background())
+	sch.dispatch.wait()
 
 	if got := len(f.syncLog.Calls()); got != 0 {
 		t.Fatalf("sync invoked %d times despite running sync row; want 0", got)
@@ -517,6 +519,7 @@ func TestSchedulerCadenceUsesLastFinished(t *testing.T) {
 
 	// Tick 1: nothing prior → fires.
 	sch.tick(ctx)
+	sch.dispatch.wait()
 	if got := len(f.syncLog.Calls()); got != 1 {
 		t.Fatalf("after tick 1: sync calls = %d; want 1", got)
 	}
@@ -524,6 +527,7 @@ func TestSchedulerCadenceUsesLastFinished(t *testing.T) {
 	// Advance 30min, still under cadence → does not fire.
 	f.clock.Add(30 * time.Minute)
 	sch.tick(ctx)
+	sch.dispatch.wait()
 	if got := len(f.syncLog.Calls()); got != 1 {
 		t.Fatalf("after tick 2 (under cadence): sync calls = %d; want 1", got)
 	}
@@ -531,6 +535,7 @@ func TestSchedulerCadenceUsesLastFinished(t *testing.T) {
 	// Advance another hour, past cadence → fires again.
 	f.clock.Add(time.Hour)
 	sch.tick(ctx)
+	sch.dispatch.wait()
 	if got := len(f.syncLog.Calls()); got != 2 {
 		t.Fatalf("after tick 3 (over cadence): sync calls = %d; want 2", got)
 	}
@@ -562,6 +567,7 @@ func TestSchedulerFailedSyncStillConsumesCadence(t *testing.T) {
 	ctx := context.Background()
 
 	sch.tick(ctx)
+	sch.dispatch.wait()
 	if calls.Load() != 1 {
 		t.Fatalf("tick 1 sync calls = %d; want 1", calls.Load())
 	}
@@ -569,6 +575,7 @@ func TestSchedulerFailedSyncStillConsumesCadence(t *testing.T) {
 	// Within cadence → no re-fire even though the prior run failed.
 	f.clock.Add(10 * time.Minute)
 	sch.tick(ctx)
+	sch.dispatch.wait()
 	if calls.Load() != 1 {
 		t.Fatalf("under-cadence tick after failure called sync %d times; want 1 (no special retry)",
 			calls.Load())
@@ -577,6 +584,7 @@ func TestSchedulerFailedSyncStillConsumesCadence(t *testing.T) {
 	// Past cadence → re-fires (regardless of prior failure).
 	f.clock.Add(time.Hour)
 	sch.tick(ctx)
+	sch.dispatch.wait()
 	if calls.Load() != 2 {
 		t.Fatalf("past-cadence tick after failure called sync %d times; want 2", calls.Load())
 	}
@@ -598,6 +606,7 @@ func TestSchedulerBothCadencesShareIndexRun(t *testing.T) {
 	ctx := context.Background()
 
 	sch.tick(ctx)
+	sch.dispatch.wait()
 
 	runs, _ := f.store.ListRuns(ctx, store.ListRunsOpts{})
 	var indexCount int
@@ -627,11 +636,13 @@ func TestSchedulerStandaloneIndexAfterSyncCadenceWaits(t *testing.T) {
 	ctx := context.Background()
 
 	sch.tick(ctx) // fires sync + pre-sync index.
+	sch.dispatch.wait()
 
 	// 10min later → under index_every from the pre-sync index, no
 	// standalone fires.
 	f.clock.Add(10 * time.Minute)
 	sch.tick(ctx)
+	sch.dispatch.wait()
 	runs, _ := f.store.ListRuns(ctx, store.ListRunsOpts{})
 	var indexCount int
 	for _, r := range runs {
@@ -647,6 +658,7 @@ func TestSchedulerStandaloneIndexAfterSyncCadenceWaits(t *testing.T) {
 	// due yet because sync_every is 1h).
 	f.clock.Add(10 * time.Minute) // total 20min since pre-sync.
 	sch.tick(ctx)
+	sch.dispatch.wait()
 	runs, _ = f.store.ListRuns(ctx, store.ListRunsOpts{})
 	indexCount = 0
 	for _, r := range runs {
@@ -792,9 +804,12 @@ func TestSchedulerDropsSyncWhenRunnerNil(t *testing.T) {
 	})
 	f.seedFile()
 	sch := f.scheduler()
-	sch.syncRun = nil
+	// A dispatcher with no runner is the index-only-host case (no rclone
+	// wired): sync-triggering paths log a skip and create no sync rows.
+	sch.dispatch = newSyncDispatcher(nil, f.srv.cfg.Logger, f.clock.Now, defaultMaxParallelSyncs)
 
 	sch.tick(context.Background())
+	sch.dispatch.wait()
 
 	if !f.containsLogLine("scheduler.skipped", "kind=sync", "sync runner not configured") {
 		t.Fatalf("missing nil-runner skip log:\n%s", f.logBuf.String())
diff --git a/cmd/squirrel/agent.go b/cmd/squirrel/agent.go
index ad18923..872c5df 100644
--- a/cmd/squirrel/agent.go
+++ b/cmd/squirrel/agent.go
@@ -169,6 +169,10 @@ func resolveSchedulerRclone(cmd *cobra.Command, cfg *config.Config) (*sync.Rclon
 	if err != nil {
 		return nil, fmt.Errorf("scheduler needs rclone for scheduled syncs: %w", err)
 	}
+	// Bound every automatic transfer by the no-progress guard so a wedged
+	// endpoint fails its own run instead of hanging forever (#160, F25).
+	// Foreground `squirrel sync` leaves this unset — a human can interrupt.
+	rcl.StallTimeout = sync.DefaultStallTimeout
 	pairs, err := sync.PairsFor(cfg, "", "")
 	if err != nil {
 		return nil, fmt.Errorf("scheduler rclone preflight: %w", err)
diff --git a/sync/rclone.go b/sync/rclone.go
index 9a4ab86..8995da5 100644
--- a/sync/rclone.go
+++ b/sync/rclone.go
@@ -21,6 +21,8 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
+	"time"
 
 	"github.com/mbertschler/squirrel/config"
 	"github.com/mbertschler/squirrel/runevents"
@@ -31,6 +33,26 @@ import (
 // --hash blake3 flag used by sync would be rejected by rclone.
 var MinRcloneVersion = Version{Major: 1, Minor: 66}
 
+// DefaultStallTimeout is the no-progress bound the agent scheduler applies
+// to every automatic rclone transfer (see Rclone.StallTimeout). Ten
+// minutes is comfortably longer than the longest legitimate gap between
+// progress reports — hashing one very large file for a --checksum
+// comparison surfaces no transfer/check advance until it completes — while
+// bounding a wedged endpoint far below the "indefinite" hang F25 recorded.
+// A dark endpoint usually fails sooner via rclone's own --contimeout /
+// --timeout; the guard covers the live-but-stuck case those don't.
+const DefaultStallTimeout = 10 * time.Minute
+
+// rcloneConnectTimeout and rcloneIOTimeout are passed as --contimeout and
+// --timeout on every streaming transfer so a dead or unresponsive endpoint
+// self-terminates instead of hanging. They match rclone's own defaults but
+// are set explicitly so the bound is guaranteed applied and lives in one
+// tunable place (#160, F25).
+const (
+	rcloneConnectTimeout = "1m"
+	rcloneIOTimeout      = "5m"
+)
+
 // Rclone is a configured rclone wrapper. Binary is the resolved executable
 // path; Config is the path to the squirrel-managed rclone.conf written by
 // WriteRcloneConfig. All Run invocations pass --config Config so the user's
@@ -38,6 +60,17 @@ var MinRcloneVersion = Version{Major: 1, Minor: 66}
 type Rclone struct {
 	Binary string
 	Config string
+	// StallTimeout, when > 0, bounds a streaming transfer (RunWithProgress)
+	// by progress rather than by total wall-clock: if rclone reports no
+	// advance in transferred+checked+bytes for this long, the child is
+	// killed and the run fails with a diagnosable "stalled" error. It is the
+	// squirrel-side backstop for a connection that stays alive but stops
+	// making progress — the storage-full S3 hang of F25 that tripped neither
+	// --contimeout nor --timeout. Zero (the default, used by the foreground
+	// CLI where a human can interrupt) disables the guard; the agent
+	// scheduler sets it to DefaultStallTimeout so an unattended cadence can
+	// never wedge forever.
+	StallTimeout time.Duration
 }
 
 // Find locates the rclone binary on PATH. The returned Rclone has Config
@@ -308,23 +341,18 @@ func (r *Rclone) RunWithProgress(ctx context.Context, onProgress func(runevents.
 	if r.Config == "" {
 		return RunResult{}, errors.New("rclone wrapper: Config not set (call WriteRcloneConfig first)")
 	}
-	full := append([]string{
-		"--config", r.Config,
-		"--use-json-log",
-		"--log-level", "INFO",
-		"--stats", "1s",
-	}, args...)
-	cmd := exec.CommandContext(ctx, r.Binary, full...)
-	stderr, err := cmd.StderrPipe()
-	if err != nil {
-		return RunResult{}, fmt.Errorf("rclone stderr pipe: %w", err)
+	// A per-run cancel backs the no-progress guard: rclone's own timeouts
+	// bound a dead connection, the guard bounds a live-but-wedged one (F25).
+	runCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	var guard *stallGuard
+	if r.StallTimeout > 0 {
+		guard = newStallGuard(runCtx, r.StallTimeout, cancel)
 	}
-	stdout, err := cmd.StdoutPipe()
+
+	cmd, stderr, stdout, err := r.startRclone(runCtx, args)
 	if err != nil {
-		return RunResult{}, fmt.Errorf("rclone stdout pipe: %w", err)
-	}
-	if err := cmd.Start(); err != nil {
-		return RunResult{}, fmt.Errorf("rclone start: %w", err)
+		return RunResult{}, err
 	}
 
 	var (
@@ -334,7 +362,7 @@ func (r *Rclone) RunWithProgress(ctx context.Context, onProgress func(runevents.
 	wg.Add(2)
 	go func() {
 		defer wg.Done()
-		parseJSONLog(stderr, &result, onProgress)
+		parseJSONLog(stderr, &result, onProgress, guard.advance)
 	}()
 	go func() {
 		defer wg.Done()
@@ -345,13 +373,136 @@ func (r *Rclone) RunWithProgress(ctx context.Context, onProgress func(runevents.
 	wg.Wait()
 
 	waitErr := cmd.Wait()
+	cancel()
+	guard.wait()
 	if waitErr != nil {
 		result.FatalError = result.Errors == 0
-		return result, rcloneExitError(waitErr, result.Stderr)
+		return result, rcloneRunError(waitErr, result.Stderr, guard, r.StallTimeout)
 	}
 	return result, nil
 }
 
+// startRclone assembles the full argument list — config, JSON logging,
+// per-second stats, and the connect/IO timeouts every transfer carries
+// (#160, F25) — and starts the child on runCtx, returning its stderr and
+// stdout pipes. runCtx backs both the caller's cancellation and the stall
+// guard's kill.
+func (r *Rclone) startRclone(runCtx context.Context, args []string) (*exec.Cmd, io.Reader, io.Reader, error) {
+	full := append([]string{
+		"--config", r.Config,
+		"--use-json-log",
+		"--log-level", "INFO",
+		"--stats", "1s",
+		"--contimeout", rcloneConnectTimeout,
+		"--timeout", rcloneIOTimeout,
+	}, args...)
+	cmd := exec.CommandContext(runCtx, r.Binary, full...)
+	stderr, err := cmd.StderrPipe()
+	if err != nil {
+		return nil, nil, nil, fmt.Errorf("rclone stderr pipe: %w", err)
+	}
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return nil, nil, nil, fmt.Errorf("rclone stdout pipe: %w", err)
+	}
+	if err := cmd.Start(); err != nil {
+		return nil, nil, nil, fmt.Errorf("rclone start: %w", err)
+	}
+	return cmd, stderr, stdout, nil
+}
+
+// stallGuard kills a running rclone child when a streaming transfer stops
+// making progress for longer than timeout. rclone's --contimeout/--timeout
+// bound a dead connection; the guard is the squirrel-side backstop for a
+// connection that stays alive yet stops advancing — the storage-full hang
+// of F25 that tripped neither. It is driven off rclone's periodic --stats
+// events (via notifyAdvance): any growth in transferred+checked+bytes
+// resets the timer, so a slow-but-moving transfer of a large volume is
+// never killed while a wedged one is.
+type stallGuard struct {
+	timeout time.Duration
+	cancel  context.CancelFunc
+	poke    chan struct{}
+	done    chan struct{}
+	fired   atomic.Bool
+}
+
+func newStallGuard(ctx context.Context, timeout time.Duration, cancel context.CancelFunc) *stallGuard {
+	g := &stallGuard{
+		timeout: timeout,
+		cancel:  cancel,
+		poke:    make(chan struct{}, 1),
+		done:    make(chan struct{}),
+	}
+	go g.watch(ctx)
+	return g
+}
+
+// advance signals progress from the stats-reader goroutine. Non-blocking:
+// a pending poke already carries the reset, so a dropped signal is
+// harmless. Safe on a nil guard so the disabled path needs no branch.
+func (g *stallGuard) advance() {
+	if g == nil {
+		return
+	}
+	select {
+	case g.poke <- struct{}{}:
+	default:
+	}
+}
+
+// watch resets its timer on every progress poke and, when the timer
+// expires with no intervening progress, records the stall and cancels the
+// run context — which makes exec.CommandContext kill the rclone child. It
+// returns when ctx is done (the run finished or was cancelled elsewhere).
+func (g *stallGuard) watch(ctx context.Context) {
+	defer close(g.done)
+	timer := time.NewTimer(g.timeout)
+	defer timer.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-g.poke:
+			if !timer.Stop() {
+				select {
+				case <-timer.C:
+				default:
+				}
+			}
+			timer.Reset(g.timeout)
+		case <-timer.C:
+			g.fired.Store(true)
+			g.cancel()
+			return
+		}
+	}
+}
+
+// wait blocks until the watch goroutine has returned so the caller can read
+// fired without a race. Safe on a nil guard.
+func (g *stallGuard) wait() {
+	if g == nil {
+		return
+	}
+	<-g.done
+}
+
+// rcloneRunError builds the error for a non-zero rclone exit. When the
+// stall guard fired, the exit is a kill squirrel caused, so we say so
+// plainly (folding in the captured stderr tail, #157) rather than
+// surfacing a bare "signal: killed"; otherwise it defers to
+// rcloneExitError.
+func rcloneRunError(waitErr error, stderrTail string, guard *stallGuard, timeout time.Duration) error {
+	if guard != nil && guard.fired.Load() {
+		if tail := strings.TrimSpace(stderrTail); tail != "" {
+			return fmt.Errorf("rclone stalled: no progress for %s, killed: %w: %s", timeout, waitErr, tail)
+		}
+		return fmt.Errorf("rclone stalled: no progress for %s, killed: %w", timeout, waitErr)
+	}
+	return rcloneExitError(waitErr, stderrTail)
+}
+
 // rcloneExitError wraps a non-zero rclone exit, folding in the captured
 // stderr tail when present so the scheduler log and the run row carry the
 // actual reason (auth, host key, path) rather than a bare "exit status 1"
@@ -554,9 +705,10 @@ func isErrorLevel(level string) bool {
 // that never reaches the JSON log still carries its reason (#157,
 // F6/F15). onProgress, if non-nil, is invoked once per stats event so
 // callers can drive a live UI.
-func parseJSONLog(r io.Reader, result *RunResult, onProgress func(runevents.Progress)) {
+func parseJSONLog(r io.Reader, result *RunResult, onProgress func(runevents.Progress), onAdvance func()) {
 	scanner := bufio.NewScanner(r)
 	scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
+	var lastProgressTotal int64
 	for scanner.Scan() {
 		line := scanner.Bytes()
 		if len(line) == 0 {
@@ -579,6 +731,7 @@ func parseJSONLog(r io.Reader, result *RunResult, onProgress func(runevents.Prog
 		}
 		if ev.Stats != nil {
 			applyStatsEvent(result, ev.Stats, onProgress)
+			notifyAdvance(result, &lastProgressTotal, onAdvance)
 			continue
 		}
 		if isErrorLevel(ev.Level) && !isRetrySummary(ev.Msg) {
@@ -596,6 +749,24 @@ func parseJSONLog(r io.Reader, result *RunResult, onProgress func(runevents.Prog
 	}
 }
 
+// notifyAdvance pokes the stall guard whenever the running total of
+// transferred+checked+bytes has grown since the last stats event, so a
+// transfer that is slow but genuinely moving keeps resetting the guard's
+// timer while a wedged one — rclone still emitting per-second stats, but
+// with flat counters — lets it fire. Checks count as progress so a long
+// verification pass over an already-populated destination is not mistaken
+// for a stall.
+func notifyAdvance(result *RunResult, last *int64, onAdvance func()) {
+	if onAdvance == nil {
+		return
+	}
+	total := result.Transferred + result.Checked + result.Bytes
+	if total > *last {
+		*last = total
+		onAdvance()
+	}
+}
+
 // applyStatsEvent folds one rclone stats event into result and drives the
 // optional progress callback. Errors is taken as the running maximum so a
 // per-attempt stats line that reports fewer errors than a prior one (a
diff --git a/sync/rclone_test.go b/sync/rclone_test.go
index efb52e1..0b252ac 100644
--- a/sync/rclone_test.go
+++ b/sync/rclone_test.go
@@ -7,6 +7,7 @@ import (
 	"os/exec"
 	"path/filepath"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -14,6 +15,51 @@ import (
 	"github.com/mbertschler/squirrel/runevents"
 )
 
+// TestStallGuardFiresWithoutProgress proves the no-progress guard cancels
+// the run — and records that it fired — when no advance arrives within the
+// timeout. This is the F25 wedge: rclone alive but transferring nothing.
+func TestStallGuardFiresWithoutProgress(t *testing.T) {
+	var cancelled atomic.Bool
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	g := newStallGuard(ctx, 30*time.Millisecond, func() {
+		cancelled.Store(true)
+		cancel()
+	})
+	g.wait() // blocks until watch returns; with no poke it fires first
+	if !g.fired.Load() {
+		t.Fatal("guard did not fire after the stall timeout elapsed with no progress")
+	}
+	if !cancelled.Load() {
+		t.Fatal("guard fired but never cancelled the run context")
+	}
+}
+
+// TestStallGuardResetsOnProgress proves a transfer that keeps advancing is
+// never killed: pokes arriving well inside the timeout keep resetting it,
+// even across a span longer than a single timeout window.
+func TestStallGuardResetsOnProgress(t *testing.T) {
+	var cancelled atomic.Bool
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	g := newStallGuard(ctx, 100*time.Millisecond, func() {
+		cancelled.Store(true)
+		cancel()
+	})
+	for i := 0; i < 8; i++ { // 160ms total, gaps of 20ms << the 100ms bound
+		time.Sleep(20 * time.Millisecond)
+		g.advance()
+	}
+	if g.fired.Load() {
+		t.Fatal("guard fired while progress was still arriving")
+	}
+	cancel() // finishing the run stops the guard cleanly
+	g.wait()
+	if cancelled.Load() {
+		t.Fatal("guard cancelled the run despite steady progress")
+	}
+}
+
 // requireRclone skips the test if rclone is not on PATH. The wrapper tests
 // exercise the real binary against a local-filesystem destination; if a
 // developer wants to run them they need rclone installed.
@@ -58,7 +104,7 @@ func TestParseJSONLogCapturesObjectlessErrors(t *testing.T) {
 		`{"stats":{"errors":1,"fatalError":true,"totalTransfers":0,"totalChecks":0,"bytes":0}}`,
 	}, "\n")
 	var r RunResult
-	parseJSONLog(strings.NewReader(stream), &r, nil)
+	parseJSONLog(strings.NewReader(stream), &r, nil, nil)
 
 	if len(r.FailedFiles) != 2 {
 		t.Fatalf("FailedFiles = %+v, want 2 (auth + reading; retry summaries dropped)", r.FailedFiles)
@@ -85,7 +131,7 @@ func TestParseJSONLogCapturesFatalLevelAndRawStderr(t *testing.T) {
 		`{"stats":{"errors":0,"fatalError":true,"totalTransfers":0,"totalChecks":0,"bytes":0}}`,
 	}, "\n")
 	var r RunResult
-	parseJSONLog(strings.NewReader(stream), &r, nil)
+	parseJSONLog(strings.NewReader(stream), &r, nil, nil)
 
 	if len(r.FailedFiles) != 1 || !strings.Contains(r.FailedFiles[0].Message, "NoCredentialProviders") {
 		t.Fatalf("FailedFiles = %+v, want the fatal-level message captured", r.FailedFiles)
@@ -105,8 +151,7 @@ func TestParseJSONLogStderrBounded(t *testing.T) {
 		lines = append(lines, fmt.Sprintf("stderr line %04d", i))
 	}
 	var r RunResult
-	parseJSONLog(strings.NewReader(strings.Join(lines, "\n")), &r, nil)
-
+	parseJSONLog(strings.NewReader(strings.Join(lines, "\n")), &r, nil, nil)
 	if len(r.Stderr) == 0 {
 		t.Fatal("Stderr empty, want the tail of the stream")
 	}
@@ -153,7 +198,7 @@ func TestParseJSONLogDetectsHashFallback(t *testing.T) {
 		`{"stats":{"errors":0,"fatalError":false,"totalTransfers":2,"totalChecks":0,"bytes":10}}`,
 	}, "\n")
 	var r RunResult
-	parseJSONLog(strings.NewReader(stream), &r, nil)
+	parseJSONLog(strings.NewReader(stream), &r, nil, nil)
 
 	if !r.HashFallback {
 		t.Fatalf("HashFallback = false, want true (no-common-hash notice should be detected)")
@@ -168,7 +213,7 @@ func TestParseJSONLogDetectsHashFallback(t *testing.T) {
 func TestParseJSONLogNoFalseHashFallback(t *testing.T) {
 	stream := `{"stats":{"errors":0,"fatalError":false,"totalTransfers":2,"totalChecks":1,"bytes":10}}`
 	var r RunResult
-	parseJSONLog(strings.NewReader(stream), &r, nil)
+	parseJSONLog(strings.NewReader(stream), &r, nil, nil)
 	if r.HashFallback {
 		t.Fatalf("HashFallback = true on a clean run, want false")
 	}
@@ -187,7 +232,7 @@ func TestParseJSONLogEmitsProgressWithByteTotal(t *testing.T) {
 	var events []runevents.Progress
 	parseJSONLog(strings.NewReader(stream), &r, func(p runevents.Progress) {
 		events = append(events, p)
-	})
+	}, nil)
 	if len(events) != 2 {
 		t.Fatalf("progress events = %d, want 2", len(events))
 	}

From 62722ebb7eed3ccbb1f38872eae153771eb0fb9f Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 24 Jul 2026 12:19:59 +0000
Subject: [PATCH 2/3] agent: real per-destination FIFO queue; loosen
 stall-guard test timing
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Address Copilot review on #181.

Fix 1 (starvation): the dispatcher skipped a sync when its destination was
already busy. Because the tick iterates volumes by name, two volumes due to
the same destination on one tick meant the later one was skipped every tick
forever — never a run row, never consuming its cadence, just skip-log spam.
In the reference setup this is the common case (photos and docs both push
to cloudbox / s3archive / kopia-mirror), and it defeated the per-destination
QUEUES that #160 option 4b specified. Replace skip-on-busy with a real
per-destination FIFO queue: an idle destination starts its worker at once, a
busy one takes the pair onto its queue (deduped so re-evaluation can't queue
it twice) and runs it when the current transfer finishes. At most one sync
per destination, overall parallelism still bounded, pre-sync-index ordering
and per-pair semantics preserved. The genuine same-pair in-flight skip stays
(the DB HasRunningRun check in the scheduler).

Fix 2/3 (flaky tests): the stallGuard tests used tight real-time bounds.
Loosen them — the fire test waits for the guard rather than racing a
deadline; the reset test drives advance() off a ticker at 20x the bound's
resolution across a span twice the stall window, so scheduler jitter can't
fake a stall.

Tests: TestSyncDispatcherSerializesSameDestination (two volumes, one
destination: both run, FIFO, max one concurrent) and
TestSchedulerTwoVolumesSameDestinationBothRun (two volumes due to the same
destination on one tick both produce a sync run — neither starved).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014Skm2Qm6wWUB7dD4CpPTz7
---
 agent/dispatcher.go      | 152 ++++++++++++++++++++++++---------------
 agent/dispatcher_test.go |  65 +++++++++++------
 agent/scheduler_test.go  |  48 +++++++++++--
 sync/rclone_test.go      |  28 ++++++--
 4 files changed, 202 insertions(+), 91 deletions(-)

diff --git a/agent/dispatcher.go b/agent/dispatcher.go
index 70c1053..b8f0cf8 100644
--- a/agent/dispatcher.go
+++ b/agent/dispatcher.go
@@ -23,10 +23,15 @@ const defaultMaxParallelSyncs = 4
 
 // syncDispatcher runs (volume, destination) syncs off the scheduler's tick
 // goroutine so a slow or wedged destination can never delay the tick loop,
-// index runs, peer syncs, or syncs to other destinations (#160, F25). The
-// concurrency unit is the destination: at most one sync per destination is
-// in flight at a time (inflight), and an overall semaphore (sem) bounds how
-// many run concurrently across destinations.
+// index runs, peer syncs, or syncs to other destinations (#160, F25).
+//
+// The concurrency unit is the destination: each destination has a FIFO
+// queue drained by a single worker, so at most one sync per destination is
+// in flight while a second volume due to the same destination (photos and
+// docs both push to cloudbox in the reference setup) waits its turn instead
+// of being skipped and starved — issue #160 option 4b is per-destination
+// queues, not skip-on-busy. An overall semaphore bounds how many
+// destinations transfer at once.
 //
 // The pre-sync index ordering the scheduler guarantees is unaffected: the
 // tick body still runs a volume's index to completion before it hands that
@@ -38,9 +43,20 @@ type syncDispatcher struct {
 
 	sem chan struct{}
 
-	mu       sync.Mutex
-	inflight map[string]bool
-	wg       sync.WaitGroup
+	mu    sync.Mutex
+	dests map[string]*destQueue
+	wg    sync.WaitGroup
+}
+
+// destQueue is one destination's serial pipeline: at most one sync active,
+// the rest waiting in FIFO order. queued holds every volume name active or
+// pending on this destination so a re-dispatch of a pair already in the
+// pipeline — the scheduler re-evaluates a still-unfinished pair on every
+// tick — is a quiet no-op rather than a duplicate enqueue.
+type destQueue struct {
+	pending []*config.Volume
+	queued  map[string]bool
+	active  bool
 }
 
 func newSyncDispatcher(run SyncRunner, logger *slog.Logger, now func() time.Time, maxParallel int) *syncDispatcher {
@@ -48,19 +64,20 @@ func newSyncDispatcher(run SyncRunner, logger *slog.Logger, now func() time.Time
 		maxParallel = defaultMaxParallelSyncs
 	}
 	return &syncDispatcher{
-		run:      run,
-		logger:   logger,
-		now:      now,
-		sem:      make(chan struct{}, maxParallel),
-		inflight: make(map[string]bool),
+		run:    run,
+		logger: logger,
+		now:    now,
+		sem:    make(chan struct{}, maxParallel),
+		dests:  make(map[string]*destQueue),
 	}
 }
 
-// dispatch launches the sync for (vol, destName) on destName's worker,
-// unless a sync to that destination is already in flight (a per-pair skip
-// log, matching the CLI wording) or no runner is configured. It never
-// blocks on the transfer: the rclone call runs in a goroutine bounded by
-// the semaphore, so the caller — the tick loop — returns at once.
+// dispatch enqueues the sync for (vol, destName) on destName's FIFO queue.
+// An idle destination starts its worker immediately; a busy one takes the
+// pair onto its queue to run when the current transfer finishes (deduped so
+// re-evaluation on later ticks can't queue the same pair twice). It never
+// blocks on the transfer — the tick loop returns at once, so a slow
+// destination cannot delay any other.
 func (d *syncDispatcher) dispatch(ctx context.Context, vol *config.Volume, destName string) {
 	if d.run == nil {
 		d.logger.Info("scheduler.skipped",
@@ -68,24 +85,67 @@ func (d *syncDispatcher) dispatch(ctx context.Context, vol *config.Volume, destN
 			"reason", "sync runner not configured")
 		return
 	}
-	if !d.claim(destName) {
-		d.logger.Info("scheduler.skipped",
-			"kind", "sync", "volume", vol.Name, "destination", destName,
-			"reason", "in-flight sync run")
+	d.mu.Lock()
+	q := d.dests[destName]
+	if q == nil {
+		q = &destQueue{queued: make(map[string]bool)}
+		d.dests[destName] = q
+	}
+	if q.queued[vol.Name] {
+		d.mu.Unlock() // already active or queued for this destination
 		return
 	}
-	d.logger.Info("scheduler.kicked",
-		"kind", "sync", "volume", vol.Name, "destination", destName)
+	q.queued[vol.Name] = true
+	if q.active {
+		q.pending = append(q.pending, vol)
+		d.mu.Unlock()
+		return
+	}
+	q.active = true
+	d.mu.Unlock()
 	d.wg.Add(1)
-	go d.runOne(ctx, vol, destName)
+	go d.worker(ctx, destName, vol)
 }
 
-// runOne is the per-destination worker body: take an overall slot (or bail
-// if the scheduler is shutting down), run the sync, and always release both
-// the slot and the destination claim.
-func (d *syncDispatcher) runOne(ctx context.Context, vol *config.Volume, destName string) {
+// worker drains destName's queue one sync at a time, so the destination
+// never has two transfers in flight. It exits — freeing the destination —
+// when the queue empties; on shutdown (ctx cancelled) it drains without
+// running, leaving the unfinished pairs to be re-evaluated next start.
+func (d *syncDispatcher) worker(ctx context.Context, destName string, vol *config.Volume) {
 	defer d.wg.Done()
-	defer d.release(destName)
+	for vol != nil {
+		if ctx.Err() == nil {
+			d.runOne(ctx, vol, destName)
+		}
+		vol = d.next(destName, vol.Name)
+	}
+}
+
+// next removes the just-finished volume from destName's pipeline and returns
+// the next queued volume, or nil (dropping the now-idle destination) when
+// none remain.
+func (d *syncDispatcher) next(destName, done string) *config.Volume {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	q := d.dests[destName]
+	if q == nil {
+		return nil
+	}
+	delete(q.queued, done)
+	if len(q.pending) == 0 {
+		delete(d.dests, destName)
+		return nil
+	}
+	next := q.pending[0]
+	q.pending = q.pending[1:]
+	return next
+}
+
+// runOne takes an overall parallelism slot (or bails on shutdown), runs the
+// sync, and emits the kicked/finished/error logs. A sync that stalls is
+// failed by the runner's transfer timeout (rclone is killed); its
+// diagnosable error lands here and in the run row.
+func (d *syncDispatcher) runOne(ctx context.Context, vol *config.Volume, destName string) {
 	select {
 	case d.sem <- struct{}{}:
 		defer func() { <-d.sem }()
@@ -95,14 +155,8 @@ func (d *syncDispatcher) runOne(ctx context.Context, vol *config.Volume, destNam
 	if ctx.Err() != nil {
 		return
 	}
-	d.execute(ctx, vol, destName)
-}
-
-// execute invokes the sync runner and emits the finished/error logs,
-// mirroring the scheduler's kicked/finished/error discipline. A sync that
-// stalls is failed by the runner's transfer timeout (rclone is killed);
-// its diagnosable error lands here and in the run row.
-func (d *syncDispatcher) execute(ctx context.Context, vol *config.Volume, destName string) {
+	d.logger.Info("scheduler.kicked",
+		"kind", "sync", "volume", vol.Name, "destination", destName)
 	start := d.now()
 	rep := d.run(ctx, vol, destName)
 	duration := d.now().Sub(start)
@@ -122,27 +176,9 @@ func (d *syncDispatcher) execute(ctx context.Context, vol *config.Volume, destNa
 	}
 }
 
-// claim marks destName in flight, returning false when a sync to it is
-// already running. release clears the mark.
-func (d *syncDispatcher) claim(destName string) bool {
-	d.mu.Lock()
-	defer d.mu.Unlock()
-	if d.inflight[destName] {
-		return false
-	}
-	d.inflight[destName] = true
-	return true
-}
-
-func (d *syncDispatcher) release(destName string) {
-	d.mu.Lock()
-	defer d.mu.Unlock()
-	delete(d.inflight, destName)
-}
-
-// wait blocks until every dispatched sync has finished. The scheduler
-// calls it during shutdown, after the run context is cancelled (which
-// kills any in-flight rclone child), so it returns promptly.
+// wait blocks until every destination worker has drained. The scheduler
+// calls it during shutdown, after the run context is cancelled (which kills
+// any in-flight rclone child), so it returns promptly.
 func (d *syncDispatcher) wait() {
 	d.wg.Wait()
 }
diff --git a/agent/dispatcher_test.go b/agent/dispatcher_test.go
index c520775..c795341 100644
--- a/agent/dispatcher_test.go
+++ b/agent/dispatcher_test.go
@@ -49,37 +49,60 @@ func TestSyncDispatcherRunsDestinationsConcurrently(t *testing.T) {
 	}
 }
 
-// TestSyncDispatcherOneInFlightPerDestination proves the concurrency unit
-// is the destination: a second volume targeting a destination that already
-// has a sync in flight is skipped with the per-pair "in-flight sync run"
-// log, and its runner never fires while the first is running.
-func TestSyncDispatcherOneInFlightPerDestination(t *testing.T) {
-	var calls atomic.Int32
-	entered := make(chan struct{}, 1)
-	release := make(chan struct{})
+// TestSyncDispatcherSerializesSameDestination proves the concurrency unit
+// is the destination: two volumes due to the same destination on one tick
+// both run — the second is queued behind the first, never starved — while
+// at most one is ever in flight against that destination.
+func TestSyncDispatcherSerializesSameDestination(t *testing.T) {
+	var active, maxActive atomic.Int32
+	var photosRan, docsRan atomic.Bool
+	entered := make(chan string, 2)
+	release := make(chan struct{}) // one send per sync, in completion order
 	run := func(ctx context.Context, vol *config.Volume, destName string) SyncRunReport {
-		calls.Add(1)
-		entered <- struct{}{}
+		n := active.Add(1)
+		for {
+			old := maxActive.Load()
+			if n <= old || maxActive.CompareAndSwap(old, n) {
+				break
+			}
+		}
+		switch vol.Name {
+		case "photos":
+			photosRan.Store(true)
+		case "docs":
+			docsRan.Store(true)
+		}
+		entered <- vol.Name
 		<-release
+		active.Add(-1)
 		return SyncRunReport{Status: store.RunStatusSuccess}
 	}
-	buf := &bytes.Buffer{}
-	logger := slog.New(slog.NewTextHandler(buf, nil))
-	d := newSyncDispatcher(run, logger, time.Now, 4)
-
+	d := newSyncDispatcher(run, discardLogger(), time.Now, 4)
 	d.dispatch(context.Background(), &config.Volume{Name: "photos"}, "cloudbox")
-	<-entered // the first sync has claimed cloudbox and is running
 	d.dispatch(context.Background(), &config.Volume{Name: "docs"}, "cloudbox")
 
-	if got := calls.Load(); got != 1 {
-		t.Fatalf("runner called %d times; want 1 (second dispatch to a busy destination must skip)", got)
+	first := <-entered // FIFO: photos was dispatched first
+	if first != "photos" {
+		t.Fatalf("first sync = %q; want photos (FIFO order)", first)
 	}
-	log := buf.String()
-	if !strings.Contains(log, `reason="in-flight sync run"`) || !strings.Contains(log, "volume=docs") {
-		t.Fatalf("expected a per-pair in-flight skip for docs→cloudbox, got:\n%s", log)
+	time.Sleep(20 * time.Millisecond) // give a wrongly-concurrent docs a chance to start
+	if got := active.Load(); got != 1 {
+		t.Fatalf("concurrent syncs to one destination = %d; want 1 (serialized)", got)
 	}
-	close(release)
+	release <- struct{}{} // let photos finish; the worker then dequeues docs
+	if second := <-entered; second != "docs" {
+		t.Fatalf("second sync = %q; want docs (queued, not starved)", second)
+	}
+	release <- struct{}{} // let docs finish
 	d.wait()
+
+	if !photosRan.Load() || !docsRan.Load() {
+		t.Fatalf("both volumes must run to the shared destination; photos=%v docs=%v",
+			photosRan.Load(), docsRan.Load())
+	}
+	if got := maxActive.Load(); got != 1 {
+		t.Fatalf("max concurrent syncs to one destination = %d; want 1", got)
+	}
 }
 
 // TestSyncDispatcherBoundsOverallParallelism proves the semaphore ceiling:
diff --git a/agent/scheduler_test.go b/agent/scheduler_test.go
index 2d03553..ff46705 100644
--- a/agent/scheduler_test.go
+++ b/agent/scheduler_test.go
@@ -124,10 +124,14 @@ func (f *fakeSyncRunner) Calls() []syncCall {
 // + fake sync runner. The scheduler's tick method drives behaviour
 // directly; the goroutine-driven run() loop is only used in the
 // integration test below.
-func newSchedulerFixture(t *testing.T, volumeCfg *config.Volume) *schedulerFixture {
+func newSchedulerFixture(t *testing.T, vols ...*config.Volume) *schedulerFixture {
 	t.Helper()
-	if volumeCfg.Path == "" {
-		volumeCfg.Path = t.TempDir()
+	volumes := make(map[string]*config.Volume, len(vols))
+	for _, v := range vols {
+		if v.Path == "" {
+			v.Path = t.TempDir()
+		}
+		volumes[v.Name] = v
 	}
 	dbPath := filepath.Join(t.TempDir(), "index.db")
 	s, err := store.Open(dbPath)
@@ -144,7 +148,7 @@ func newSchedulerFixture(t *testing.T, volumeCfg *config.Volume) *schedulerFixtu
 		Listen:     "127.0.0.1:0",
 		Token:      "tok",
 		Version:    "test",
-		Volumes:    map[string]*config.Volume{volumeCfg.Name: volumeCfg},
+		Volumes:    volumes,
 		Logger:     logger,
 		SyncRunner: syncRunner.Runner(),
 	}, s)
@@ -194,7 +198,6 @@ func (f *schedulerFixture) seedFile() {
 		if err := os.WriteFile(p, []byte("x"), 0o644); err != nil {
 			f.t.Fatalf("write %s: %v", p, err)
 		}
-		return
 	}
 }
 
@@ -468,6 +471,41 @@ func TestSchedulerSkipsWhenInFlightSyncRun(t *testing.T) {
 	}
 }
 
+// TestSchedulerTwoVolumesSameDestinationBothRun guards the starvation trap
+// (#160): the tick iterates volumes by name, so if two volumes are due to
+// the same destination on one tick the earlier one wins the destination.
+// The dispatcher must queue the later one behind it and run it — not skip
+// it every tick forever. "aaa" sorts before "zzz"; both must produce a sync
+// run to "backup".
+func TestSchedulerTwoVolumesSameDestinationBothRun(t *testing.T) {
+	f := newSchedulerFixture(t,
+		&config.Volume{Name: "aaa", SyncTo: []string{"backup"}, SyncEvery: time.Hour},
+		&config.Volume{Name: "zzz", SyncTo: []string{"backup"}, SyncEvery: time.Hour},
+	)
+	f.seedFile()
+	sch := f.scheduler()
+
+	sch.tick(context.Background())
+	sch.dispatch.wait()
+
+	var aaaRan, zzzRan bool
+	for _, c := range f.syncLog.Calls() {
+		if c.Destination != "backup" {
+			continue
+		}
+		switch c.Volume {
+		case "aaa":
+			aaaRan = true
+		case "zzz":
+			zzzRan = true
+		}
+	}
+	if !aaaRan || !zzzRan {
+		t.Fatalf("both volumes must sync to the shared destination; aaa=%v zzz=%v (calls=%+v)",
+			aaaRan, zzzRan, f.syncLog.Calls())
+	}
+}
+
 // TestSchedulerSkipsWhenVolumeLockHeld pins the coordination with the
 // peer-sync surface: an incoming /v1/sync session holds the per-volume
 // router lock for the duration of the session; a scheduler tick during
diff --git a/sync/rclone_test.go b/sync/rclone_test.go
index 0b252ac..340eb2e 100644
--- a/sync/rclone_test.go
+++ b/sync/rclone_test.go
@@ -18,11 +18,14 @@ import (
 // TestStallGuardFiresWithoutProgress proves the no-progress guard cancels
 // the run — and records that it fired — when no advance arrives within the
 // timeout. This is the F25 wedge: rclone alive but transferring nothing.
+// The test waits for the guard rather than racing a deadline, so the exact
+// timeout is not timing-sensitive; a generous value keeps it robust under
+// load.
 func TestStallGuardFiresWithoutProgress(t *testing.T) {
 	var cancelled atomic.Bool
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	g := newStallGuard(ctx, 30*time.Millisecond, func() {
+	g := newStallGuard(ctx, 100*time.Millisecond, func() {
 		cancelled.Store(true)
 		cancel()
 	})
@@ -36,19 +39,30 @@ func TestStallGuardFiresWithoutProgress(t *testing.T) {
 }
 
 // TestStallGuardResetsOnProgress proves a transfer that keeps advancing is
-// never killed: pokes arriving well inside the timeout keep resetting it,
-// even across a span longer than a single timeout window.
+// never killed: pokes driven off a ticker at a fraction of the stall window
+// keep resetting it across a span twice as long as one window, so only a
+// working reset explains the guard staying quiet. The wide ratio between
+// the stall timeout and the poke interval keeps scheduler jitter from
+// faking a stall.
 func TestStallGuardResetsOnProgress(t *testing.T) {
 	var cancelled atomic.Bool
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	g := newStallGuard(ctx, 100*time.Millisecond, func() {
+	const stall = 400 * time.Millisecond
+	g := newStallGuard(ctx, stall, func() {
 		cancelled.Store(true)
 		cancel()
 	})
-	for i := 0; i < 8; i++ { // 160ms total, gaps of 20ms << the 100ms bound
-		time.Sleep(20 * time.Millisecond)
-		g.advance()
+	tick := time.NewTicker(stall / 20) // pokes at 20x the resolution of the bound
+	defer tick.Stop()
+	deadline := time.After(2 * stall) // outlast a single window, so a reset is required
+	for done := false; !done; {
+		select {
+		case <-tick.C:
+			g.advance()
+		case <-deadline:
+			done = true
+		}
 	}
 	if g.fired.Load() {
 		t.Fatal("guard fired while progress was still arriving")

From c0029c8531bc2708cfc630314e11777ac489428a Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 24 Jul 2026 18:58:32 +0000
Subject: [PATCH 3/3] agent: skip redundant pre-sync index for an in-flight
 sync (#160)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Async dispatch (this PR) freed the tick loop to keep firing every 30s
while a dispatched sync runs in the background. But maybeRunSync ran the
pre-sync index before the per-pair in-flight check, and syncDue stays
true for the whole duration of an in-flight sync — the finished-run
watermark only advances when the sync completes. So a slow multi-hour
push made the scheduler re-index the source on every tick, burning I/O
and flooding the never-pruned runs audit trail with a kind='index' row
per tick. The old serial scheduler never hit this: the tick blocked
inside the sync, so it could not re-evaluate until the sync finished.

Filter each volume's due destinations down to the pairs that are not
already syncing before running the pre-sync index, and skip the index
entirely when none remain (there is no new push for it to precede). The
in-flight signal reads from two sources that compose to close every
window: the dispatcher's in-memory queue (set the instant a pair is
enqueued, before its run row is visible) and the runs table
(authoritative once the row exists, and the signal that also catches a
stale row or a concurrent CLI `squirrel sync`). The ordering invariant
holds — a volume syncing to one in-flight dest and one free dest still
indexes once and dispatches the free one — and the per-pair
scheduler.skipped log is unchanged.

runSync becomes syncInFlight (a bool predicate, no dispatch); maybeRunSync
is decomposed into dueSyncDests/dispatchableSyncDests to stay small.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014Skm2Qm6wWUB7dD4CpPTz7
---
 agent/dispatcher.go     |  16 ++++
 agent/scheduler.go      | 125 ++++++++++++++++++----------
 agent/scheduler_test.go | 177 +++++++++++++++++++++++++++++++++++-----
 3 files changed, 254 insertions(+), 64 deletions(-)

diff --git a/agent/dispatcher.go b/agent/dispatcher.go
index b8f0cf8..9f99a57 100644
--- a/agent/dispatcher.go
+++ b/agent/dispatcher.go
@@ -121,6 +121,22 @@ func (d *syncDispatcher) worker(ctx context.Context, destName string, vol *confi
 	}
 }
 
+// inFlight reports whether (destName, volName) is currently active or queued
+// on destName's pipeline. The scheduler consults it alongside the runs table
+// so a pair whose dispatch from an earlier tick is still working — but whose
+// run row is not yet visible — is recognised as in flight and not dragged
+// through a redundant pre-sync index every tick (#160). Safe for concurrent
+// use; a nil-runner dispatcher (no dests) always reports false.
+func (d *syncDispatcher) inFlight(destName, volName string) bool {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	q := d.dests[destName]
+	if q == nil {
+		return false
+	}
+	return q.queued[volName]
+}
+
 // next removes the just-finished volume from destName's pipeline and returns
 // the next queued volume, or nil (dropping the now-idle destination) when
 // none remain.
diff --git a/agent/scheduler.go b/agent/scheduler.go
index 0394688..8524404 100644
--- a/agent/scheduler.go
+++ b/agent/scheduler.go
@@ -234,42 +234,87 @@ func (s *scheduler) resolveVolume(ctx context.Context, name, absPath string) (st
 	return created, nil
 }
 
-// maybeRunSync evaluates the sync_every cadence and, if any destination
-// is due, runs a pre-sync index and then the per-destination syncs.
-// Returns true when the scheduler took any sync-related action (kicked,
-// skipped, or errored) so the caller can suppress a redundant
-// standalone-index check.
+// maybeRunSync evaluates the sync_every cadence and, for the destinations
+// that are due and not already syncing, runs a pre-sync index and then hands
+// each pair to the per-destination dispatcher. Returns true when the
+// scheduler took any sync-related action (kicked, skipped, or errored) so the
+// caller can suppress a redundant standalone-index check.
 //
 // Invariant: a scheduled sync always runs an index immediately before
-// pushing. If the pre-sync index fails or is skipped, no syncs run
-// this tick — the next tick re-evaluates.
+// pushing. If the pre-sync index fails or is skipped, no syncs run this tick
+// — the next tick re-evaluates.
+//
+// A due destination whose (volume, destination) sync from an earlier tick is
+// still in flight is dropped before the pre-sync index runs. Dispatch is
+// asynchronous (#160), so without this the tick loop would re-index the source
+// on every 30s tick for the entire duration of a slow multi-hour push —
+// burning I/O and flooding the never-pruned runs audit trail with a
+// kind='index' row per tick. When every due destination is already syncing,
+// the pre-sync index is skipped entirely: there is no new push for it to
+// precede, and the in-flight pushes were each already preceded by their own.
 func (s *scheduler) maybeRunSync(ctx context.Context, vol *config.Volume, volumeID int64) bool {
 	if vol.SyncEvery == 0 {
 		return false
 	}
-	var dueDests []string
+	dueDests := s.dueSyncDests(ctx, vol, volumeID)
+	if len(dueDests) == 0 {
+		return false
+	}
+	dispatchable := s.dispatchableSyncDests(ctx, vol, volumeID, dueDests)
+	if len(dispatchable) == 0 {
+		// Every due destination is already syncing (each skip logged per
+		// pair). Report the sync action so the standalone-index branch stays
+		// suppressed — otherwise its own re-index would reintroduce the exact
+		// per-tick flood this guard exists to prevent.
+		return true
+	}
+	if !s.maybeRunIndex(ctx, vol, volumeID, "pre-sync", 0) {
+		return true
+	}
+	for _, destName := range dispatchable {
+		s.dispatch.dispatch(ctx, vol, destName)
+	}
+	return true
+}
+
+// dueSyncDests returns the destinations on the volume's sync_to whose
+// sync_every cadence has elapsed. A per-destination lookup error is logged and
+// that destination is dropped from this tick's evaluation (the next tick
+// re-evaluates), never aborting the others.
+func (s *scheduler) dueSyncDests(ctx context.Context, vol *config.Volume, volumeID int64) []string {
+	var due []string
 	for _, destName := range vol.SyncTo {
-		due, err := s.syncDue(ctx, volumeID, destName, vol.SyncEvery)
+		ok, err := s.syncDue(ctx, volumeID, destName, vol.SyncEvery)
 		if err != nil {
 			s.logger.Error("scheduler.error",
 				"kind", "sync", "volume", vol.Name, "destination", destName,
 				"err", err.Error())
 			continue
 		}
-		if due {
-			dueDests = append(dueDests, destName)
+		if ok {
+			due = append(due, destName)
 		}
 	}
-	if len(dueDests) == 0 {
-		return false
-	}
-	if !s.maybeRunIndex(ctx, vol, volumeID, "pre-sync", 0) {
-		return true
-	}
+	return due
+}
+
+// dispatchableSyncDests filters dueDests to the pairs the dispatcher can start
+// now, dropping any whose sync is already in flight and logging the same
+// per-pair scheduler.skipped the serial scheduler emitted — so an operator
+// still sees why a due destination isn't moving, it just no longer drags a
+// redundant pre-sync index along with it.
+func (s *scheduler) dispatchableSyncDests(ctx context.Context, vol *config.Volume, volumeID int64, dueDests []string) []string {
+	var dispatchable []string
 	for _, destName := range dueDests {
-		s.runSync(ctx, vol, volumeID, destName)
+		if s.syncInFlight(ctx, vol, volumeID, destName) {
+			s.logger.Info("scheduler.skipped",
+				"kind", "sync", "volume", vol.Name, "destination", destName,
+				"reason", "in-flight sync run")
+			continue
+		}
+		dispatchable = append(dispatchable, destName)
 	}
-	return true
+	return dispatchable
 }
 
 // syncDue computes (now - last_finished) ≥ cadence for the (volume,
@@ -463,34 +508,28 @@ func indexRunStatus(rep index.Report, err error) (string, bool) {
 	return store.RunStatusSuccess, true
 }
 
-// runSync evaluates the per-pair in-flight gate, then hands the sync to
-// the per-destination dispatcher (#160). The DB check catches a sync of
-// this exact pair already running — a concurrent CLI `squirrel sync`, or a
-// stale row not yet reaped — producing the same clean per-pair skip the
-// serial scheduler did; the dispatcher additionally serialises different
-// volumes that target the same destination and confines a slow transfer to
-// its own worker. The SyncRunner's downstream call (sync.RunPair) adds its
-// own atomic BeginSyncRunIfClear gate, so a race that sneaks past the
-// pre-check still produces a clean skip rather than a duplicate run.
-//
-// Dispatch is non-blocking: a slow or wedged destination is confined to its
-// own worker and never stalls the tick loop or syncs to other destinations,
-// which is the freeze F25 recorded. A nil SyncRunner surfaces as a clean
-// skip inside the dispatcher so an agent running pure index-only schedules
-// can omit the sync wiring without logging at error level each tick.
-func (s *scheduler) runSync(ctx context.Context, vol *config.Volume, volumeID int64, destName string) {
+// syncInFlight reports whether a sync of this exact (volume, destination) pair
+// is already active — a stale row, a concurrent CLI `squirrel sync`, or (the
+// common case) a dispatch from an earlier tick still working. It consults two
+// sources so neither a not-yet-visible run row nor a just-drained queue slot
+// can fool it: the dispatcher's in-memory queue (set the instant a pair is
+// enqueued, before its run row exists) and the runs table (authoritative once
+// the row is written, and the signal that also catches a stale row or a
+// concurrent CLI sync). The downstream dispatch and sync.RunPair's atomic
+// BeginSyncRunIfClear gate remain the final guard against a duplicate run, so
+// this check only needs to be good enough to suppress a redundant pre-sync
+// index. A store error is treated as in-flight so a transient DB hiccup skips
+// this tick rather than racing a duplicate push.
+func (s *scheduler) syncInFlight(ctx context.Context, vol *config.Volume, volumeID int64, destName string) bool {
+	if s.dispatch.inFlight(destName, vol.Name) {
+		return true
+	}
 	running, err := s.store.HasRunningRun(ctx, store.RunKindSync, volumeID, destName)
 	if err != nil {
 		s.logger.Error("scheduler.error",
 			"kind", "sync", "volume", vol.Name, "destination", destName,
 			"err", err.Error())
-		return
-	}
-	if running {
-		s.logger.Info("scheduler.skipped",
-			"kind", "sync", "volume", vol.Name, "destination", destName,
-			"reason", "in-flight sync run")
-		return
+		return true
 	}
-	s.dispatch.dispatch(ctx, vol, destName)
+	return running
 }
diff --git a/agent/scheduler_test.go b/agent/scheduler_test.go
index ff46705..617a9a9 100644
--- a/agent/scheduler_test.go
+++ b/agent/scheduler_test.go
@@ -416,9 +416,12 @@ func TestSchedulerSkipsWhenInFlightIndexRun(t *testing.T) {
 	}
 }
 
-// TestSchedulerSkipsWhenInFlightSyncRun plants a stale running sync
-// row and asserts the scheduler skips the sync (but, per the
-// invariant, the pre-sync index still ran in this tick). We use a
+// TestSchedulerSkipsWhenInFlightSyncRun plants a stale running sync row and
+// asserts the scheduler skips the sync — and, since that pair is the volume's
+// only due destination, skips the pre-sync index too (#160). Re-indexing the
+// source on every tick for the whole duration of an in-flight push would burn
+// I/O and flood the never-pruned runs audit trail with a kind='index' row per
+// tick; the in-flight push was already preceded by its own index. We use a
 // fake sync runner so we can verify it never gets invoked.
 func TestSchedulerSkipsWhenInFlightSyncRun(t *testing.T) {
 	f := newSchedulerFixture(t, &config.Volume{
@@ -449,25 +452,12 @@ func TestSchedulerSkipsWhenInFlightSyncRun(t *testing.T) {
 	if !f.containsLogLine("scheduler.skipped", "kind=sync", "in-flight") {
 		t.Fatalf("missing in-flight sync skip log:\n%s", f.logBuf.String())
 	}
-	// Pre-sync index still ran — the invariant says sync always
-	// indexes first, and the only thing skipped here is the sync
-	// itself.
+	// No new pre-sync index: with the sole due destination already in
+	// flight, there is no new push for an index to precede. Only the planted
+	// 'running' sync row should exist.
 	runs, _ := f.store.ListRuns(context.Background(), store.ListRunsOpts{})
-	var sawIndex, sawSync bool
-	for _, r := range runs {
-		switch r.Kind {
-		case store.RunKindIndex:
-			sawIndex = true
-		case store.RunKindSync:
-			sawSync = true
-		}
-	}
-	if !sawIndex {
-		t.Fatalf("expected a pre-sync index row even when sync is gated: %+v", runs)
-	}
-	if !sawSync {
-		// The planted 'running' sync row counts as a sync; just sanity.
-		t.Fatalf("expected the planted running sync row to remain: %+v", runs)
+	if len(runs) != 1 || runs[0].Kind != store.RunKindSync || runs[0].Status != store.RunStatusRunning {
+		t.Fatalf("runs=%+v; want only the planted running sync row (no redundant pre-sync index)", runs)
 	}
 }
 
@@ -579,6 +569,151 @@ func TestSchedulerCadenceUsesLastFinished(t *testing.T) {
 	}
 }
 
+// waitUntil polls cond until it returns true or a short deadline passes,
+// failing the test on timeout. It synchronises with the dispatcher's async
+// workers without a fixed sleep, so a test can wait for one destination's
+// queue to drain while another destination's sync is still (deliberately)
+// blocked.
+func waitUntil(t *testing.T, cond func() bool, what string) {
+	t.Helper()
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		if cond() {
+			return
+		}
+		time.Sleep(time.Millisecond)
+	}
+	t.Fatalf("timed out waiting for %s", what)
+}
+
+// TestSchedulerInFlightSyncNotReindexedEachTick is the #160 regression guard:
+// once a sync is dispatched and stays in flight across ticks, the scheduler
+// must NOT re-run that volume's pre-sync index every tick — otherwise a slow
+// multi-hour push would flood the never-pruned runs audit trail with a
+// kind='index' row per 30s tick. The pre-sync index runs exactly once (on the
+// tick that dispatches the sync), while a different, not-in-flight
+// (volume, destination) keeps indexing and dispatching normally on the very
+// ticks the wedged one is skipped — proving the guard is per-pair, not global.
+func TestSchedulerInFlightSyncNotReindexedEachTick(t *testing.T) {
+	f := newSchedulerFixture(t,
+		&config.Volume{Name: "photos", SyncTo: []string{"cloudbox"}, SyncEvery: time.Hour},
+		&config.Volume{Name: "docs", SyncTo: []string{"backup"}, SyncEvery: time.Hour},
+	)
+	f.seedFile()
+
+	release := make(chan struct{})
+	photosInFlight := make(chan struct{}, 1)
+	docsFinished := make(chan struct{}, 8)
+	f.syncLog.runFn = func(ctx context.Context, vol *config.Volume, destName string) SyncRunReport {
+		v, err := f.store.GetVolumeByName(ctx, vol.Name)
+		if err != nil {
+			return SyncRunReport{Err: err}
+		}
+		id, blocker, err := f.store.BeginSyncRunIfClear(ctx, store.SyncRunSpec{
+			VolumeID: v.ID, Destination: destName,
+		})
+		if err != nil {
+			return SyncRunReport{Err: err}
+		}
+		if blocker != nil {
+			return SyncRunReport{RunID: blocker.ID, Err: fmt.Errorf("already running")}
+		}
+		if vol.Name == "photos" {
+			// Announce the running row exists, then block — keeping the
+			// photos->cloudbox sync in flight across the following ticks
+			// without ever finishing it (the F25 wedge, minus the kill).
+			photosInFlight <- struct{}{}
+			<-release
+		}
+		if err := f.store.FinishRun(ctx, id, store.RunStatusSuccess, "", 0); err != nil {
+			return SyncRunReport{RunID: id, Err: err}
+		}
+		if vol.Name == "docs" {
+			docsFinished <- struct{}{}
+		}
+		return SyncRunReport{RunID: id, Status: store.RunStatusSuccess}
+	}
+
+	sch := f.scheduler()
+	ctx := context.Background()
+
+	countIndex := func(volName string) int {
+		t.Helper()
+		v, err := f.store.GetVolumeByName(ctx, volName)
+		if err != nil {
+			t.Fatalf("GetVolumeByName(%s): %v", volName, err)
+		}
+		runs, err := f.store.ListRuns(ctx, store.ListRunsOpts{})
+		if err != nil {
+			t.Fatalf("ListRuns: %v", err)
+		}
+		n := 0
+		for _, r := range runs {
+			if r.Kind == store.RunKindIndex && r.VolumeID.Valid && r.VolumeID.Int64 == v.ID {
+				n++
+			}
+		}
+		return n
+	}
+	// drainDocs waits for docs' just-dispatched sync to finish and its queue
+	// slot to clear, so the next tick sees docs as not-in-flight while photos
+	// stays wedged.
+	drainDocs := func(tick int) {
+		<-docsFinished
+		waitUntil(t, func() bool { return !sch.dispatch.inFlight("backup", "docs") },
+			fmt.Sprintf("docs queue to drain after tick %d", tick))
+	}
+
+	// Tick 1: both due, neither in flight -> each indexes once and dispatches.
+	// photos' sync blocks (stays running); docs' sync finishes.
+	sch.tick(ctx)
+	<-photosInFlight
+	drainDocs(1)
+	if got := countIndex("photos"); got != 1 {
+		t.Fatalf("after tick 1: photos index runs = %d; want 1", got)
+	}
+	if got := countIndex("docs"); got != 1 {
+		t.Fatalf("after tick 1: docs index runs = %d; want 1", got)
+	}
+
+	// Ticks 2 and 3: photos is still in flight -> skipped with NO new
+	// pre-sync index. docs comes due again each hour and must keep indexing +
+	// dispatching normally on the very ticks photos is skipped.
+	for tick := 2; tick <= 3; tick++ {
+		f.clock.Add(time.Hour + time.Minute)
+		sch.tick(ctx)
+		drainDocs(tick)
+		if got := countIndex("photos"); got != 1 {
+			t.Fatalf("after tick %d: photos index runs = %d; want 1 (in-flight sync must not be re-indexed)", tick, got)
+		}
+		if got := countIndex("docs"); got != tick {
+			t.Fatalf("after tick %d: docs index runs = %d; want %d (not-in-flight volume indexes each due tick)", tick, got, tick)
+		}
+	}
+
+	var photosCalls, docsCalls int
+	for _, c := range f.syncLog.Calls() {
+		switch c.Volume {
+		case "photos":
+			photosCalls++
+		case "docs":
+			docsCalls++
+		}
+	}
+	if photosCalls != 1 {
+		t.Fatalf("photos sync calls = %d; want 1 (dispatched once, then skipped while in flight)", photosCalls)
+	}
+	if docsCalls != 3 {
+		t.Fatalf("docs sync calls = %d; want 3 (once per due tick)", docsCalls)
+	}
+	if !f.containsLogLine("scheduler.skipped", "kind=sync", "volume=photos", "in-flight") {
+		t.Fatalf("missing in-flight skip log for photos:\n%s", f.logBuf.String())
+	}
+
+	close(release)
+	sch.dispatch.wait()
+}
+
 // TestSchedulerFailedSyncStillConsumesCadence guards the failure
 // policy ("Failed runs aren't specially retried — the next tick
 // re-evaluates"). After a failed sync, the next within-cadence tick