feat: add a per-key path to the reconciler (#939)

The loop watches KV buckets and runs a whole-set recompute when anything in them changes, discarding the update it already holds. A caller whose work is per resource can now set ReconcileKey and be handed the key that changed, so a change to one instance costs one instance's work. Reconcile stays required. A per-key queue only converges if something behind it covers what the queue never learned: a key deleted while the watch was down is enqueued by nothing, because nobody observes an absence. Keeping the whole-set pass as the startup pass and the resync backstop is what makes that real, and it makes the change additive — a caller that sets only Reconcile behaves exactly as before. KeyFor maps an update onto the caller's unit of work, which is not always the KV key: fifty instances changing in one account are one recompute. Reporting ok=false falls back to a whole-set pass, which is the honest answer for a delete — a tombstone carries no value, so a caller whose work key lives in the value cannot name it. A dropped and re-established watch queues the whole set rather than any key, since UpdatesOnly hides the gap. One worker drains the queue, and a key taken off it is no longer pending, so a change arriving mid-reconcile is queued again rather than folded into the pass already running.
2026-10-15 17:11:30 +00:00 · 2026-08-30 09:52:46 +10:00
parent 4804c77946
commit 423c6bee26
3 changed files with 420 additions and 8 deletions
@@ -0,0 +1,56 @@
+package reconciler
+
+import "sync"
+
+// workQueue holds the keys a per-key loop has yet to reconcile, in arrival
+// order and without duplicates.
+//
+// It is not the correctness boundary. Whatever the queue drops or never learns
+// about is still covered by the resync, which reconciles the whole set: a key
+// deleted while the watch was down is never enqueued by anything, because
+// nobody observes an absence.
+type workQueue struct {
+	mu sync.Mutex
+	// order preserves arrival order; pending is the same set, for the
+	// membership test that keeps a key from being queued twice.
+	order   []string
+	pending map[string]struct{}
+	// whole records that something asked for a whole-set pass — an update that
+	// could not be attributed to a key, or a watch that dropped and lost the
+	// gap. It outranks the keys rather than joining them.
+	whole bool
+}
+
+// add queues keys that are not already waiting.
+func (q *workQueue) add(keys ...string) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	if q.pending == nil {
+		q.pending = map[string]struct{}{}
+	}
+	for _, key := range keys {
+		if _, ok := q.pending[key]; ok {
+			continue
+		}
+		q.pending[key] = struct{}{}
+		q.order = append(q.order, key)
+	}
+}
+
+// addWhole asks for a whole-set pass instead of per-key work.
+func (q *workQueue) addWhole() {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	q.whole = true
+}
+
+// take removes and returns everything queued so far. A key taken here is no
+// longer pending, so a change arriving while it is being reconciled queues it
+// again rather than being folded into the pass that is already running.
+func (q *workQueue) take() (whole bool, keys []string) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	whole, keys = q.whole, q.order
+	q.whole, q.order, q.pending = false, nil, nil
+	return whole, keys
+}
@@ -0,0 +1,263 @@
+package reconciler_test
+
+import (
+	"context"
+	"sort"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/reconciler"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// keyLog records which keys the per-key path was handed, in order, and lets a
+// test wait for a count rather than sleep and hope.
+type keyLog struct {
+	mu   sync.Mutex
+	keys []string
+	// block, when set, holds each call until it is closed, so a test can make a
+	// reconcile still be running when the next change arrives.
+	block chan struct{}
+}
+
+func newKeyLog() *keyLog { return &keyLog{} }
+
+func (l *keyLog) record(key string) {
+	l.mu.Lock()
+	block := l.block
+	l.keys = append(l.keys, key)
+	l.mu.Unlock()
+	if block != nil {
+		<-block
+	}
+}
+
+func (l *keyLog) seen() []string {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return append([]string(nil), l.keys...)
+}
+
+func (l *keyLog) waitFor(t *testing.T, n int, why string) {
+	t.Helper()
+	deadline := time.Now().Add(10 * time.Second)
+	for {
+		if len(l.seen()) >= n {
+			return
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("timed out waiting for %d key reconciles (%s); saw %v", n, why, l.seen())
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+}
+
+func (l *keyLog) stillAt(t *testing.T, n int, why string) {
+	t.Helper()
+	time.Sleep(300 * time.Millisecond)
+	assert.Len(t, l.seen(), n, why)
+}
+
+func TestRun_AnUpdateReconcilesTheKeyThatChanged(t *testing.T) {
+	store, _ := newBucket(t, "perkey-one")
+	p, keys := newPasses(), newKeyLog()
+
+	run(t, reconciler.Config{
+		Name:         "one",
+		Sources:      []reconciler.Source{reconciler.Fixed(store.Bucket, ">")},
+		Reconcile:    func(context.Context) error { p.record(); return nil },
+		ReconcileKey: func(_ context.Context, key string) error { keys.record(key); return nil },
+		Resync:       time.Hour,
+	})
+	p.waitFor(t, 1, "the startup pass")
+
+	require.NoError(t, store.Set(t.Context(), "one", &record{Name: "one"}))
+
+	keys.waitFor(t, 1, "the key that changed")
+	assert.Equal(t, []string{"one"}, keys.seen())
+	p.stillAt(t, 1, "a per-key loop must not also run the whole set on a change")
+}
+
+// The burst case the whole-set path collapses into one pass becomes one pass
+// per distinct key here, which is the entire point of the extension.
+func TestRun_ABurstReconcilesEachKeyOnce(t *testing.T) {
+	store, _ := newBucket(t, "perkey-burst")
+	p, keys := newPasses(), newKeyLog()
+
+	run(t, reconciler.Config{
+		Name:         "burst",
+		Sources:      []reconciler.Source{reconciler.Fixed(store.Bucket, ">")},
+		Reconcile:    func(context.Context) error { p.record(); return nil },
+		ReconcileKey: func(_ context.Context, key string) error { keys.record(key); return nil },
+		Resync:       time.Hour,
+		Debounce:     200 * time.Millisecond,
+	})
+	p.waitFor(t, 1, "the startup pass")
+
+	for _, key := range []string{"a", "b", "c"} {
+		require.NoError(t, store.Set(t.Context(), key, &record{Name: key}))
+	}
+
+	keys.waitFor(t, 3, "one reconcile per key in the burst")
+	got := keys.seen()
+	sort.Strings(got)
+	assert.Equal(t, []string{"a", "b", "c"}, got)
+}
+
+// Repeated writes to one key inside a debounce window are one change to that
+// key, not five: a fresh read makes the reconcile idempotent, so the four
+// earlier revisions have nothing left to say.
+func TestRun_RepeatedWritesToOneKeyReconcileItOnce(t *testing.T) {
+	store, _ := newBucket(t, "perkey-dedup")
+	p, keys := newPasses(), newKeyLog()
+
+	run(t, reconciler.Config{
+		Name:         "dedup",
+		Sources:      []reconciler.Source{reconciler.Fixed(store.Bucket, ">")},
+		Reconcile:    func(context.Context) error { p.record(); return nil },
+		ReconcileKey: func(_ context.Context, key string) error { keys.record(key); return nil },
+		Resync:       time.Hour,
+		Debounce:     200 * time.Millisecond,
+	})
+	p.waitFor(t, 1, "the startup pass")
+
+	for range 5 {
+		require.NoError(t, store.Set(t.Context(), "hot", &record{Name: "hot"}))
+	}
+
+	keys.waitFor(t, 1, "the reconcile covering the burst")
+	keys.stillAt(t, 1, "five writes to one key in one burst must reconcile it once")
+}
+
+// The case dedup must not swallow: a key taken off the queue is no longer
+// pending, so a change arriving while it is being reconciled is a new piece of
+// work rather than one the running pass is assumed to cover.
+func TestRun_AChangeDuringAKeysReconcileIsNotLost(t *testing.T) {
+	store, _ := newBucket(t, "perkey-inflight")
+	p, keys := newPasses(), newKeyLog()
+	keys.block = make(chan struct{})
+
+	run(t, reconciler.Config{
+		Name:         "inflight",
+		Sources:      []reconciler.Source{reconciler.Fixed(store.Bucket, ">")},
+		Reconcile:    func(context.Context) error { p.record(); return nil },
+		ReconcileKey: func(_ context.Context, key string) error { keys.record(key); return nil },
+		Resync:       time.Hour,
+		Debounce:     50 * time.Millisecond,
+	})
+	p.waitFor(t, 1, "the startup pass")
+
+	require.NoError(t, store.Set(t.Context(), "hot", &record{Name: "first"}))
+	keys.waitFor(t, 1, "the first reconcile, which is now blocked")
+
+	// This write lands while the reconcile above is still in flight.
+	require.NoError(t, store.Set(t.Context(), "hot", &record{Name: "second"}))
+	time.Sleep(200 * time.Millisecond)
+	close(keys.block)
+
+	keys.waitFor(t, 2, "the second reconcile, for the change that arrived mid-pass")
+}
+
+// KeyFor exists for a caller whose unit of work is coarser than a key. Two
+// instances in one account are one recompute, not two.
+func TestRun_KeyForMapsUpdatesOntoTheCallersUnitOfWork(t *testing.T) {
+	store, _ := newBucket(t, "perkey-keyfor")
+	p, keys := newPasses(), newKeyLog()
+
+	run(t, reconciler.Config{
+		Name:         "keyfor",
+		Sources:      []reconciler.Source{reconciler.Fixed(store.Bucket, ">")},
+		Reconcile:    func(context.Context) error { p.record(); return nil },
+		ReconcileKey: func(_ context.Context, key string) error { keys.record(key); return nil },
+		KeyFor: func(entry jetstream.KeyValueEntry) ([]string, bool) {
+			return []string{"account-1"}, true
+		},
+		Resync:   time.Hour,
+		Debounce: 200 * time.Millisecond,
+	})
+	p.waitFor(t, 1, "the startup pass")
+
+	for _, key := range []string{"i-1", "i-2", "i-3"} {
+		require.NoError(t, store.Set(t.Context(), key, &record{Name: key}))
+	}
+
+	keys.waitFor(t, 1, "the account the three instances belong to")
+	keys.stillAt(t, 1, "three instances in one account are one unit of work")
+	assert.Equal(t, []string{"account-1"}, keys.seen())
+}
+
+// An update that cannot be attributed — the delete case, where the work key
+// lived in a value the tombstone does not carry — falls back to the pass that
+// needs no attribution.
+func TestRun_AnUnattributableUpdateFallsBackToTheWholeSet(t *testing.T) {
+	store, _ := newBucket(t, "perkey-fallback")
+	p, keys := newPasses(), newKeyLog()
+
+	run(t, reconciler.Config{
+		Name:         "fallback",
+		Sources:      []reconciler.Source{reconciler.Fixed(store.Bucket, ">")},
+		Reconcile:    func(context.Context) error { p.record(); return nil },
+		ReconcileKey: func(_ context.Context, key string) error { keys.record(key); return nil },
+		KeyFor: func(entry jetstream.KeyValueEntry) ([]string, bool) {
+			if entry.Operation() == jetstream.KeyValuePut {
+				return []string{entry.Key()}, true
+			}
+			return nil, false
+		},
+		Resync:   time.Hour,
+		Debounce: 100 * time.Millisecond,
+	})
+	p.waitFor(t, 1, "the startup pass")
+
+	require.NoError(t, store.Set(t.Context(), "gone", &record{Name: "gone"}))
+	keys.waitFor(t, 1, "the put, which is attributable")
+
+	require.NoError(t, store.Delete(t.Context(), "gone"))
+	p.waitFor(t, 2, "the whole-set pass covering an update that could not be attributed")
+	keys.stillAt(t, 1, "the delete must not be reconciled as a key")
+}
+
+// The backstop is the reason ReconcileKey is an addition rather than a
+// replacement: a key the queue never learned about — deleted while the watch
+// was down, so nothing observed it — is still covered.
+func TestRun_TheResyncStillRunsTheWholeSet(t *testing.T) {
+	store, _ := newBucket(t, "perkey-resync")
+	p, keys := newPasses(), newKeyLog()
+
+	run(t, reconciler.Config{
+		Name:         "resync",
+		Sources:      []reconciler.Source{reconciler.Fixed(store.Bucket, ">")},
+		Reconcile:    func(context.Context) error { p.record(); return nil },
+		ReconcileKey: func(_ context.Context, key string) error { keys.record(key); return nil },
+		Resync:       150 * time.Millisecond,
+	})
+
+	p.waitFor(t, 4, "repeated whole-set resync passes with ReconcileKey set")
+	assert.Empty(t, keys.seen(), "an idle bucket produces no per-key work")
+}
+
+func TestRun_AFailedKeyDoesNotStopTheLoop(t *testing.T) {
+	store, _ := newBucket(t, "perkey-failure")
+	p, keys := newPasses(), newKeyLog()
+
+	run(t, reconciler.Config{
+		Name:      "failure",
+		Sources:   []reconciler.Source{reconciler.Fixed(store.Bucket, ">")},
+		Reconcile: func(context.Context) error { p.record(); return nil },
+		ReconcileKey: func(_ context.Context, key string) error {
+			keys.record(key)
+			return assert.AnError
+		},
+		Resync:   time.Hour,
+		Debounce: 50 * time.Millisecond,
+	})
+	p.waitFor(t, 1, "the startup pass")
+
+	require.NoError(t, store.Set(t.Context(), "first", &record{Name: "first"}))
+	keys.waitFor(t, 1, "the first key, which fails")
+	require.NoError(t, store.Set(t.Context(), "second", &record{Name: "second"}))
+	keys.waitFor(t, 2, "the next key, after the first one failed")
+}
@@ -2,6 +2,11 @@
 // timer: it watches the KV buckets the function reads, coalesces a burst of
 // updates into one pass, and falls back to a periodic resync so a gap in the
 // watch cannot leave the world unconverged.
+//
+// A caller that can act on one resource at a time also sets ReconcileKey, and
+// the key that changed is carried through to it. The whole-set Reconcile stays
+// required either way: it is the startup pass and the resync backstop, so a key
+// the queue never learns about still converges.
 package reconciler

 import (
@@ -44,8 +49,30 @@ type Config struct {
 	// Sources are watched for changes. An empty Sources is allowed: the loop
 	// then runs on the resync alone, which is the pre-watch behaviour.
 	Sources []Source
-	// Reconcile performs one pass. It is never called concurrently with itself.
+	// Reconcile performs one whole-set pass. It is never called concurrently
+	// with itself or with ReconcileKey.
+	//
+	// Required even when ReconcileKey is set, because it is the startup pass
+	// and the resync backstop. A per-key queue only converges if something
+	// behind it covers what the queue never learned: a key deleted while the
+	// watch was down is enqueued by nothing, since nobody observes an absence.
 	Reconcile func(ctx context.Context) error
+	// ReconcileKey handles one changed key, for a caller whose work is per
+	// resource rather than a recompute of the whole set. Unset means every
+	// change runs Reconcile, which is the whole-set behaviour.
+	//
+	// It is called once per distinct key in a burst, so a caller whose unit of
+	// work is coarser than a key — an account rather than an instance — maps
+	// updates onto that unit with KeyFor rather than deduplicating here.
+	ReconcileKey func(ctx context.Context, key string) error
+	// KeyFor maps a watch update onto the keys it dirties. Nil means the key
+	// that changed is the key to reconcile.
+	//
+	// Reporting ok=false means the update could not be attributed, and the
+	// change runs Reconcile instead. That is the honest answer for a delete: a
+	// tombstone carries no value, so a caller whose work key lives in the value
+	// cannot name it, and a whole-set pass needs no attribution.
+	KeyFor func(entry jetstream.KeyValueEntry) (keys []string, ok bool)
 	// Resync bounds how stale the world can get when no update arrives, and is
 	// also when Sources are re-enumerated. Zero means DefaultResync.
 	Resync time.Duration
@@ -112,13 +139,18 @@ var _ Source = dynamicSource{}
 func Run(ctx context.Context, cfg Config) {
 	cfg.fixDefaults()
 	if cfg.Reconcile == nil {
+		// Setting only ReconcileKey looks like a working loop and is not one:
+		// nothing would run the startup pass or the resync backstop.
+		slog.ErrorContext(ctx, "reconcile loop not started: Reconcile is required", "loop", cfg.Name)
 		return
 	}

 	// Buffered by one: a change arriving mid-pass sets the pending flag rather
-	// than blocking the watcher goroutine, and is served by the next pass.
+	// than blocking the watcher goroutine, and is served by the next pass. The
+	// keys themselves ride in the queue; this only says "something is waiting".
 	changes := make(chan struct{}, 1)
-	w := &watchSet{cfg: cfg, changes: changes}
+	queue := &workQueue{}
+	w := &watchSet{cfg: cfg, changes: changes, queue: queue}
 	defer w.stop()

 	// Watches go up before the first pass, not after: a change landing between
@@ -139,15 +171,40 @@ func Run(ctx context.Context, cfg Config) {
 			// the last resync is watched from here on, and the pass that
 			// follows covers whatever it already held.
 			w.resync(ctx)
+			// Whatever is queued is covered by the whole-set pass about to run,
+			// so it is dropped rather than reconciled again after it.
+			queue.take()
 			pass(ctx, cfg, "resync")
 		case <-changes:
 			if settle(ctx, changes, cfg.Debounce) {
-				pass(ctx, cfg, "change")
+				serve(ctx, cfg, queue)
 			}
 		}
 	}
 }

+// serve runs the work a settled burst produced: one whole-set pass, or one pass
+// per distinct key that changed.
+func serve(ctx context.Context, cfg Config, queue *workQueue) {
+	if cfg.ReconcileKey == nil {
+		pass(ctx, cfg, "change")
+		return
+	}
+	whole, keys := queue.take()
+	if whole {
+		// A whole-set pass covers every key that was waiting alongside it, so
+		// the keys it returned are deliberately not reconciled again.
+		pass(ctx, cfg, "change")
+		return
+	}
+	for _, key := range keys {
+		if ctx.Err() != nil {
+			return
+		}
+		keyPass(ctx, cfg, key)
+	}
+}
+
 // settle waits for a burst to stop arriving, so one multi-key write produces
 // one pass rather than one per key. It gives up waiting after maxDebounce so a
 // continuous stream of updates cannot defer the pass indefinitely. It reports
@@ -187,11 +244,26 @@ func pass(ctx context.Context, cfg Config, trigger string) {
 		"loop", cfg.Name, "trigger", trigger, "duration_ms", otelsetup.Millis(time.Since(start)))
 }

+// keyPass runs one per-key reconcile. Its failure is logged rather than
+// retried here: the resync re-reconciles the whole set, so a key that failed is
+// covered by the same backstop as a key the queue never saw.
+func keyPass(ctx context.Context, cfg Config, key string) {
+	start := time.Now()
+	if err := cfg.ReconcileKey(ctx, key); err != nil {
+		slog.WarnContext(ctx, "reconcile key failed, retrying on the next resync",
+			"loop", cfg.Name, "key", key, "duration_ms", otelsetup.Millis(time.Since(start)), "err", err)
+		return
+	}
+	slog.DebugContext(ctx, "reconcile key complete",
+		"loop", cfg.Name, "key", key, "duration_ms", otelsetup.Millis(time.Since(start)))
+}
+
 // watchSet holds one watcher per bucket currently being watched, keyed by
 // bucket name so a resync can tell an already-watched bucket from a new one.
 type watchSet struct {
 	cfg     Config
 	changes chan struct{}
+	queue   *workQueue
 	active  map[string]*watcher
 }

@@ -266,7 +338,7 @@ func (w *watchSet) drain(ctx context.Context, bucket *kvstore.Bucket, filter str
 			if update == nil {
 				continue
 			}
-			w.signal()
+			w.observe(update)
 		}
 		if ctx.Err() != nil {
 			return
@@ -285,13 +357,34 @@ func (w *watchSet) drain(ctx context.Context, bucket *kvstore.Bucket, filter str
 			continue
 		}
 		watch.kw = kw
+		// The whole set, not a key: UpdatesOnly hides what happened during the
+		// gap, so what changed is exactly what cannot be named.
+		w.queue.addWhole()
 		w.signal()
 	}
 }

-// signal reports a change without blocking: the channel's single slot already
-// means "at least one change is pending", which is all a full-recompute
-// reconcile needs to know.
+// observe queues the work an update implies and wakes the loop.
+func (w *watchSet) observe(entry jetstream.KeyValueEntry) {
+	switch {
+	case w.cfg.ReconcileKey == nil:
+		// Whole-set callers never read the key, so nothing is queued for them.
+	case w.cfg.KeyFor == nil:
+		w.queue.add(entry.Key())
+	default:
+		keys, ok := w.cfg.KeyFor(entry)
+		if !ok {
+			w.queue.addWhole()
+			break
+		}
+		w.queue.add(keys...)
+	}
+	w.signal()
+}
+
+// signal reports that work is waiting without blocking: the channel's single
+// slot already means "at least one change is pending", and for a per-key loop
+// what is pending is in the queue rather than in the channel.
 func (w *watchSet) signal() {
 	select {
 	case w.changes <- struct{}{}: