diff --git a/spinifex/kvstore/bucket.go b/spinifex/kvstore/bucket.go index 457553e55..5df9686e1 100644 --- a/spinifex/kvstore/bucket.go +++ b/spinifex/kvstore/bucket.go @@ -6,6 +6,7 @@ package kvstore import ( "context" "errors" + "fmt" "sync" "time" @@ -46,6 +47,10 @@ func NewOpenBucket(kv jetstream.KeyValue, cfg Config) *Bucket { return &Bucket{kv: kv, cfg: cfg} } +// Name returns the configured bucket name, for callers holding several buckets +// that need to tell them apart in a map or a log line. +func (b *Bucket) Name() string { return b.cfg.Name } + // Configured reports whether the bucket has anything to open against, for the // callers whose absent-KV path is a legitimate fallback rather than an error. func (b *Bucket) Configured() bool { @@ -75,6 +80,25 @@ func (b *Bucket) KV(ctx context.Context) (jetstream.KeyValue, error) { return kv, nil } +// Watch returns a watcher over the keys matching filter, which takes NATS +// subject wildcards ("node.*", "lb.*", ">" for everything). +// +// Updates only: the caller reconciles once at startup, so replaying every +// existing key would fire a redundant pass, and replaying them again on every +// re-establish would defeat the debounce. The cost is that a disconnect gap is +// invisible, which is what the caller's periodic resync exists to cover. +func (b *Bucket) Watch(ctx context.Context, filter string) (jetstream.KeyWatcher, error) { + kv, err := b.KV(ctx) + if err != nil { + return nil, err + } + w, err := kv.Watch(ctx, filter, jetstream.UpdatesOnly()) + if err != nil { + return nil, fmt.Errorf("kvstore: watch %s %s: %w", b.cfg.Name, filter, err) + } + return w, nil +} + // open picks the kvutil helper matching the configured TTL and replica count. func (b *Bucket) open(ctx context.Context) (jetstream.KeyValue, error) { switch { diff --git a/spinifex/reconciler/reconciler.go b/spinifex/reconciler/reconciler.go new file mode 100644 index 000000000..a4cdb9457 --- /dev/null +++ b/spinifex/reconciler/reconciler.go @@ -0,0 +1,295 @@ +// Package reconciler runs a reconcile function on change rather than on a +// 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. +package reconciler + +import ( + "context" + "log/slog" + "time" + + "github.com/mulgadc/spinifex/spinifex/kvstore" + "github.com/mulgadc/spinifex/spinifex/otelsetup" + "github.com/nats-io/nats.go/jetstream" +) + +// Defaults for Config. The resync is the correctness backstop, so it is slow; +// the debounce only has to outlast a multi-key write, so it is short. +const ( + DefaultResync = 5 * time.Minute + DefaultDebounce = 250 * time.Millisecond + // maxDebounce bounds how long a continuous stream of updates can defer a + // pass, so a busy bucket still reconciles rather than starving. + maxDebounce = 2 * time.Second + // retryWatch is how long to wait before re-opening a watcher that could not + // be established, so a bucket that is briefly unreachable is retried + // without spinning. + retryWatch = 5 * time.Second +) + +// Source names the buckets to watch and the keys within them that matter. +type Source interface { + // Buckets is re-evaluated on every resync, so a bucket that appears after + // startup — a new account's, say — is picked up without a restart. + Buckets(ctx context.Context) ([]*kvstore.Bucket, error) + // Filter is a NATS subject filter over key names, e.g. "node.*". + Filter() string +} + +// Config describes one reconcile loop. +type Config struct { + // Name identifies the loop in logs. + Name string + // 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 func(ctx context.Context) error + // 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 + // Debounce is how long to wait for a burst to settle. Zero means + // DefaultDebounce. + Debounce time.Duration +} + +// fixDefaults fills in the zero-valued durations. +func (c *Config) fixDefaults() { + if c.Resync <= 0 { + c.Resync = DefaultResync + } + if c.Debounce <= 0 { + c.Debounce = DefaultDebounce + } +} + +// staticSource is one fixed bucket, for the callers whose bucket set does not +// change while the process runs. +type staticSource struct { + bucket *kvstore.Bucket + filter string +} + +func (s staticSource) Buckets(context.Context) ([]*kvstore.Bucket, error) { + return []*kvstore.Bucket{s.bucket}, nil +} + +func (s staticSource) Filter() string { return s.filter } + +// Fixed returns a Source over one bucket whose identity is known at startup. +func Fixed(bucket *kvstore.Bucket, filter string) Source { + return staticSource{bucket: bucket, filter: filter} +} + +var _ Source = staticSource{} + +// Run reconciles once, then on every change and every resync until ctx is done. +// It returns only when ctx is cancelled. +func Run(ctx context.Context, cfg Config) { + cfg.fixDefaults() + if cfg.Reconcile == nil { + 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. + changes := make(chan struct{}, 1) + w := &watchSet{cfg: cfg, changes: changes} + defer w.stop() + + // Watches go up before the first pass, not after: a change landing between + // the two would otherwise be invisible until the next resync, because the + // pass that would have seen it ran before the watch existed. + w.resync(ctx) + pass(ctx, cfg, "startup") + + resync := time.NewTicker(cfg.Resync) + defer resync.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-resync.C: + // Re-enumerating before the pass means a bucket that appeared since + // the last resync is watched from here on, and the pass that + // follows covers whatever it already held. + w.resync(ctx) + pass(ctx, cfg, "resync") + case <-changes: + if settle(ctx, changes, cfg.Debounce) { + pass(ctx, cfg, "change") + } + } + } +} + +// 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 +// false only when ctx ended first. +func settle(ctx context.Context, changes <-chan struct{}, debounce time.Duration) bool { + quiet := time.NewTimer(debounce) + defer quiet.Stop() + limit := time.NewTimer(maxDebounce) + defer limit.Stop() + for { + select { + case <-ctx.Done(): + return false + case <-limit.C: + return true + case <-quiet.C: + return true + case <-changes: + if !quiet.Stop() { + <-quiet.C + } + quiet.Reset(debounce) + } + } +} + +// pass runs one reconcile, logging rather than returning a failure: the loop +// outlives any single pass, and the next resync retries. +func pass(ctx context.Context, cfg Config, trigger string) { + start := time.Now() + if err := cfg.Reconcile(ctx); err != nil { + slog.WarnContext(ctx, "reconcile pass failed, retrying next cycle", + "loop", cfg.Name, "trigger", trigger, "duration_ms", otelsetup.Millis(time.Since(start)), "err", err) + return + } + slog.DebugContext(ctx, "reconcile pass complete", + "loop", cfg.Name, "trigger", trigger, "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{} + active map[string]*watcher +} + +// watcher is one bucket's watcher plus the goroutine draining it. +type watcher struct { + kw jetstream.KeyWatcher + cancel context.CancelFunc +} + +// resync opens a watcher for every bucket a Source now names and drops those it +// no longer does. A bucket that cannot be watched is left out and retried on +// the next resync, so one unreachable bucket does not stop the others. +func (w *watchSet) resync(ctx context.Context) { + if w.active == nil { + w.active = map[string]*watcher{} + } + wanted := map[string]struct{}{} + for _, src := range w.cfg.Sources { + buckets, err := src.Buckets(ctx) + if err != nil { + // Leave the existing watchers in place: a failed enumeration is not + // evidence that the buckets went away. + slog.WarnContext(ctx, "reconcile: enumerate watch buckets failed", + "loop", w.cfg.Name, "err", err) + for name := range w.active { + wanted[name] = struct{}{} + } + continue + } + for _, bucket := range buckets { + name := bucket.Name() + wanted[name] = struct{}{} + if _, ok := w.active[name]; ok { + continue + } + if watch := w.open(ctx, bucket, src.Filter()); watch != nil { + w.active[name] = watch + } + } + } + for name, watch := range w.active { + if _, ok := wanted[name]; !ok { + watch.stop() + delete(w.active, name) + } + } +} + +// open establishes one watcher and starts draining it. A nil return means the +// bucket could not be watched this cycle. +func (w *watchSet) open(ctx context.Context, bucket *kvstore.Bucket, filter string) *watcher { + watchCtx, cancel := context.WithCancel(ctx) + kw, err := bucket.Watch(watchCtx, filter) + if err != nil { + cancel() + slog.WarnContext(ctx, "reconcile: watch unavailable, falling back to resync until it recovers", + "loop", w.cfg.Name, "bucket", bucket.Name(), "filter", filter, "err", err) + return nil + } + watch := &watcher{kw: kw, cancel: cancel} + go w.drain(watchCtx, bucket, filter, watch) + return watch +} + +// drain forwards updates until the watcher's channel closes, then re-opens it. +// A closed channel means the connection dropped, so re-establishment is itself +// a change signal: UpdatesOnly hides whatever happened during the gap. +func (w *watchSet) drain(ctx context.Context, bucket *kvstore.Bucket, filter string, watch *watcher) { + for { + for update := range watch.kw.Updates() { + // A nil update marks the end of the initial replay, not a change. + if update == nil { + continue + } + w.signal() + } + if ctx.Err() != nil { + return + } + slog.InfoContext(ctx, "reconcile: watch dropped, re-establishing", + "loop", w.cfg.Name, "bucket", bucket.Name()) + kw, err := bucket.Watch(ctx, filter) + if err != nil { + slog.WarnContext(ctx, "reconcile: re-establish watch failed", + "loop", w.cfg.Name, "bucket", bucket.Name(), "err", err) + select { + case <-ctx.Done(): + return + case <-time.After(retryWatch): + } + continue + } + watch.kw = kw + 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. +func (w *watchSet) signal() { + select { + case w.changes <- struct{}{}: + default: + } +} + +// stop tears down every watcher. +func (w *watchSet) stop() { + for name, watch := range w.active { + watch.stop() + delete(w.active, name) + } +} + +// stop cancels the draining goroutine and closes the underlying watcher. +// Cancelling already tears the subscription down, so Stop routinely reports an +// invalid subscription on the way out; that is the expected shutdown path. +func (w *watcher) stop() { + w.cancel() + if err := w.kw.Stop(); err != nil { + slog.Debug("reconcile: stopping watcher", "err", err) + } +} diff --git a/spinifex/reconciler/reconciler_test.go b/spinifex/reconciler/reconciler_test.go new file mode 100644 index 000000000..17fb62375 --- /dev/null +++ b/spinifex/reconciler/reconciler_test.go @@ -0,0 +1,362 @@ +package reconciler_test + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/kvstore" + "github.com/mulgadc/spinifex/spinifex/reconciler" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// record is a stand-in for whatever the watched bucket holds; the loop never +// decodes it, only notices that it changed. +type record struct { + Name string `json:"name"` +} + +// passes counts reconcile calls and lets a test wait for the count to reach a +// value, so no test sleeps for a fixed duration hoping a pass happened. +type passes struct { + mu sync.Mutex + cond *sync.Cond + count int +} + +func newPasses() *passes { + p := &passes{} + p.cond = sync.NewCond(&p.mu) + return p +} + +func (p *passes) record() { + p.mu.Lock() + defer p.mu.Unlock() + p.count++ + p.cond.Broadcast() +} + +func (p *passes) now() int { + p.mu.Lock() + defer p.mu.Unlock() + return p.count +} + +// waitFor blocks until at least n passes have run, failing the test rather than +// hanging if they do not. +func (p *passes) waitFor(t *testing.T, n int, why string) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + p.mu.Lock() + defer p.mu.Unlock() + for p.count < n { + p.mu.Unlock() + if time.Now().After(deadline) { + p.mu.Lock() + t.Fatalf("timed out waiting for %d passes (%s); saw %d", n, why, p.count) + } + time.Sleep(5 * time.Millisecond) + p.mu.Lock() + } +} + +// stillAt asserts no further pass ran during a short settling window, for the +// tests whose point is that something did NOT trigger a pass. +func (p *passes) stillAt(t *testing.T, n int, why string) { + t.Helper() + time.Sleep(300 * time.Millisecond) + assert.Equal(t, n, p.now(), why) +} + +// newBucket starts an embedded JetStream server and returns a store over a +// bucket on it, plus the connection so a test can drop it. +func newBucket(t *testing.T, name string) (*kvstore.Store[record], *nats.Conn) { + t.Helper() + _, nc, _ := testutil.StartTestJetStream(t) + return kvstore.New[record](testutil.NewJetStream(t, nc), kvstore.Config{ + Name: name, + History: 1, + }), nc +} + +// run starts the loop under a context the test cancels on cleanup. +func run(t *testing.T, cfg reconciler.Config) { + t.Helper() + ctx, cancel := context.WithCancel(t.Context()) + done := make(chan struct{}) + go func() { + defer close(done) + reconciler.Run(ctx, cfg) + }() + t.Cleanup(func() { + cancel() + <-done + }) +} + +func TestRun_ReconcilesOnceAtStartup(t *testing.T) { + store, _ := newBucket(t, "reconciler-startup") + p := newPasses() + + run(t, reconciler.Config{ + Name: "startup", + Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, + Reconcile: func(context.Context) error { p.record(); return nil }, + Resync: time.Hour, + }) + + p.waitFor(t, 1, "the startup pass") + p.stillAt(t, 1, "an idle bucket must not produce further passes") +} + +func TestRun_AnUpdateTriggersAPass(t *testing.T) { + store, _ := newBucket(t, "reconciler-update") + p := newPasses() + + run(t, reconciler.Config{ + Name: "update", + Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, + Reconcile: func(context.Context) error { p.record(); return nil }, + Resync: time.Hour, + }) + p.waitFor(t, 1, "the startup pass") + + require.NoError(t, store.Set(t.Context(), "one", &record{Name: "one"})) + p.waitFor(t, 2, "the pass triggered by the write") +} + +// TestRun_ABurstCollapsesIntoOnePass is the whole point of the debounce: a +// multi-key write is one logical change, and a full-recompute reconcile gains +// nothing from running once per key. +func TestRun_ABurstCollapsesIntoOnePass(t *testing.T) { + store, _ := newBucket(t, "reconciler-burst") + p := newPasses() + + run(t, reconciler.Config{ + Name: "burst", + Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, + Reconcile: func(context.Context) error { p.record(); return nil }, + Resync: time.Hour, + Debounce: 200 * time.Millisecond, + }) + p.waitFor(t, 1, "the startup pass") + + for _, key := range []string{"a", "b", "c", "d", "e", "f", "g", "h"} { + require.NoError(t, store.Set(t.Context(), key, &record{Name: key})) + } + + p.waitFor(t, 2, "the pass covering the burst") + p.stillAt(t, 2, "eight writes in one burst must produce one pass, not eight") +} + +func TestRun_ResyncFiresWithNoWatchTraffic(t *testing.T) { + store, _ := newBucket(t, "reconciler-resync") + p := newPasses() + + run(t, reconciler.Config{ + Name: "resync", + Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, + Reconcile: func(context.Context) error { p.record(); return nil }, + Resync: 150 * time.Millisecond, + }) + + p.waitFor(t, 4, "repeated resync passes against a bucket nobody is writing to") +} + +// TestRun_NoSourcesStillReconcilesOnTheResync pins the degenerate case: a loop +// with nothing watchable is the pre-watch behaviour, not a dead loop. +func TestRun_NoSourcesStillReconcilesOnTheResync(t *testing.T) { + p := newPasses() + + run(t, reconciler.Config{ + Name: "no-sources", + Reconcile: func(context.Context) error { p.record(); return nil }, + Resync: 150 * time.Millisecond, + }) + + p.waitFor(t, 3, "resync passes with no sources configured") +} + +// TestRun_AFailedPassDoesNotStopTheLoop covers the reason Reconcile's error is +// logged rather than returned: the loop outlives any single pass. +func TestRun_AFailedPassDoesNotStopTheLoop(t *testing.T) { + store, _ := newBucket(t, "reconciler-failure") + p := newPasses() + var calls atomic.Int64 + + run(t, reconciler.Config{ + Name: "failure", + Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, ">")}, + Reconcile: func(context.Context) error { + p.record() + if calls.Add(1) == 1 { + return assert.AnError + } + return nil + }, + Resync: 150 * time.Millisecond, + }) + + p.waitFor(t, 3, "passes after the first one failed") +} + +// TestRun_AFilterExcludesOtherKeys stops an unrelated write to the same bucket +// waking a loop that does not care about it — the ELBv2 bucket holds listeners +// and rules alongside the load balancers DNS reconciles on. +func TestRun_AFilterExcludesOtherKeys(t *testing.T) { + store, _ := newBucket(t, "reconciler-filter") + p := newPasses() + + run(t, reconciler.Config{ + Name: "filter", + Sources: []reconciler.Source{reconciler.Fixed(store.Bucket, "lb.*")}, + Reconcile: func(context.Context) error { p.record(); return nil }, + Resync: time.Hour, + }) + p.waitFor(t, 1, "the startup pass") + + require.NoError(t, store.Set(t.Context(), "rule.one", &record{Name: "rule"})) + p.stillAt(t, 1, "a write outside the filter must not trigger a pass") + + require.NoError(t, store.Set(t.Context(), "lb.one", &record{Name: "lb"})) + p.waitFor(t, 2, "the pass triggered by a write inside the filter") +} + +// dynamicSource is the per-account bucket case: the set of buckets is not known +// at startup and is re-read on every resync. +type dynamicSource struct { + mu sync.Mutex + buckets []*kvstore.Bucket +} + +func (s *dynamicSource) Buckets(context.Context) ([]*kvstore.Bucket, error) { + s.mu.Lock() + defer s.mu.Unlock() + return append([]*kvstore.Bucket(nil), s.buckets...), nil +} + +func (s *dynamicSource) Filter() string { return ">" } + +func (s *dynamicSource) add(b *kvstore.Bucket) { + s.mu.Lock() + defer s.mu.Unlock() + s.buckets = append(s.buckets, b) +} + +// TestRun_ABucketAppearingAfterStartupIsPickedUp is the per-account bucket +// case: EKS and RDS create a bucket per account, and JetStream has no +// bucket-created event, so discovery rides on the resync. +func TestRun_ABucketAppearingAfterStartupIsPickedUp(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + js := testutil.NewJetStream(t, nc) + first := kvstore.New[record](js, kvstore.Config{Name: "reconciler-acct-a", History: 1}) + second := kvstore.New[record](js, kvstore.Config{Name: "reconciler-acct-b", History: 1}) + + src := &dynamicSource{} + src.add(first.Bucket) + p := newPasses() + + run(t, reconciler.Config{ + Name: "dynamic", + Sources: []reconciler.Source{src}, + Reconcile: func(context.Context) error { p.record(); return nil }, + Resync: 200 * time.Millisecond, + }) + p.waitFor(t, 1, "the startup pass") + + // Force the second bucket into existence, then publish it to the source. + require.NoError(t, second.Set(t.Context(), "seed", &record{Name: "seed"})) + src.add(second.Bucket) + + // One resync to discover it, then a write it must now notice. + before := p.now() + p.waitFor(t, before+2, "resync passes covering the discovery of the new bucket") + + before = p.now() + require.NoError(t, second.Set(t.Context(), "later", &record{Name: "later"})) + p.waitFor(t, before+1, "a pass triggered by the newly discovered bucket") +} + +// TestRun_AnUnwatchableSourceStillResyncs covers the fallback: a bucket with no +// JetStream behind it cannot be watched, and the loop must degrade to the timer +// rather than stop. +func TestRun_AnUnwatchableSourceStillResyncs(t *testing.T) { + unwatchable := kvstore.NewBucket(nil, kvstore.Config{ + Name: "reconciler-unwatchable", + Missing: "no JetStream configured", + }) + p := newPasses() + + run(t, reconciler.Config{ + Name: "unwatchable", + Sources: []reconciler.Source{reconciler.Fixed(unwatchable, ">")}, + Reconcile: func(context.Context) error { p.record(); return nil }, + Resync: 150 * time.Millisecond, + }) + + p.waitFor(t, 3, "resync passes despite the watch never establishing") +} + +func TestFixed_ReportsItsBucketAndFilter(t *testing.T) { + bucket := kvstore.NewBucket(nil, kvstore.Config{Name: "b"}) + src := reconciler.Fixed(bucket, "node.*") + + buckets, err := src.Buckets(t.Context()) + require.NoError(t, err) + require.Len(t, buckets, 1) + assert.Equal(t, "b", buckets[0].Name()) + assert.Equal(t, "node.*", src.Filter()) +} + +func TestBucket_WatchOnANilJetStreamReportsTheConfiguredMessage(t *testing.T) { + bucket := kvstore.NewBucket(nil, kvstore.Config{ + Name: "b", + Missing: "watch test: no JetStream client configured", + }) + + _, err := bucket.Watch(t.Context(), ">") + require.EqualError(t, err, "watch test: no JetStream client configured") +} + +// TestBucket_WatchDeliversUpdatesOnly pins UpdatesOnly at the kvstore seam: a +// watcher established over a bucket that already holds keys must not replay +// them, or every re-establish would fire a redundant pass per existing key. +func TestBucket_WatchDeliversUpdatesOnly(t *testing.T) { + store, _ := newBucket(t, "reconciler-updatesonly") + require.NoError(t, store.Set(t.Context(), "existing", &record{Name: "existing"})) + + watcher, err := store.Watch(t.Context(), ">") + require.NoError(t, err) + t.Cleanup(func() { _ = watcher.Stop() }) + + require.NoError(t, store.Set(t.Context(), "fresh", &record{Name: "fresh"})) + + key := waitForUpdate(t, watcher) + assert.Equal(t, "fresh", key, "a pre-existing key must not be replayed") +} + +// waitForUpdate returns the key of the next real update, skipping the nil that +// marks the end of the initial replay. +func waitForUpdate(t *testing.T, watcher jetstream.KeyWatcher) string { + t.Helper() + deadline := time.After(10 * time.Second) + for { + select { + case update := <-watcher.Updates(): + if update == nil { + continue + } + return update.Key() + case <-deadline: + t.Fatal("timed out waiting for a watch update") + return "" + } + } +}