agent: per-destination scheduler workers + transfer timeouts (#160)

agent/dispatcher.go

@@ -0,0 +1,200 @@
+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: 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
+// 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
+	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 {
+	if maxParallel <= 0 {
+		maxParallel = defaultMaxParallelSyncs
+	}
+	return &syncDispatcher{
+		run:    run,
+		logger: logger,
+		now:    now,
+		sem:    make(chan struct{}, maxParallel),
+		dests:  make(map[string]*destQueue),
+	}
+}
+
+// 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",
+			"kind", "sync", "volume", vol.Name, "destination", destName,
+			"reason", "sync runner not configured")
+		return
+	}
+	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
+	}
+	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.worker(ctx, destName, vol)
+}
+
+// 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()
+	for vol != nil {
+		if ctx.Err() == nil {
+			d.runOne(ctx, vol, destName)
+		}
+		vol = d.next(destName, vol.Name)
+	}
+}
+
+// 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.
+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 }()
+	case <-ctx.Done():
+		return
+	}
+	if ctx.Err() != nil {
+		return
+	}
+	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)
+	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())
+	}
+}
+
+// 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()
+}

agent/dispatcher_test.go

@@ -0,0 +1,151 @@
+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)
+	}
+}
+
+// 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 {
+		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}
+	}
+	d := newSyncDispatcher(run, discardLogger(), time.Now, 4)
+	d.dispatch(context.Background(), &config.Volume{Name: "photos"}, "cloudbox")
+	d.dispatch(context.Background(), &config.Volume{Name: "docs"}, "cloudbox")
+
+	first := <-entered // FIFO: photos was dispatched first
+	if first != "photos" {
+		t.Fatalf("first sync = %q; want photos (FIFO order)", first)
+	}
+	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)
+	}
+	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:
+// 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())
+	}
+}