From e2c5397b5dd19dea22b38f634d7d9f70c0aece69 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Fri, 4 Sep 2026 08:08:46 +1000 Subject: [PATCH] feat: gateway-side instance record cache (Phase 2) (#992) * feat: add read-only instance record cache (Phase 2) Adds spinifex/instancecache: a revision-fenced snapshot+watch informer over the instance record KV space, with a per-account index and OTel metrics. Wired up and running in awsgw at startup, but not yet consulted by any read path. * fix: cover the fence with real syncs, bound decode-failure cardinality, expose degraded Adds end-to-end fencing tests that drive freshSync/periodicResync against the embedded NATS server via a test-only postSnapshotHook seam, since drainBuffered unit tests alone don't exercise the buffer-before-snapshot ordering. Drops the per-key attribute from the decode_failures counter to keep it O(1) time series; the key is still logged where the decode happens. Adds a Degraded accessor set on sync failure and cleared on the next success, per D7, without changing what List or Get answer. * refactor: replace instancecache's buffering fence with ResumeFromRevision The watch/snapshot fence no longer needs a side buffer, watcher modes, or an overflow failure mode. Every sync now snapshots first, then opens a watcher with kvstore.Bucket.WatchFrom (ResumeFromRevision), and replays onto a private candidate map until the end-of-replay marker arrives. Only a fully caught-up candidate is swapped in as the live map, together with its watcher becoming active; the previous watcher is stopped and joined only after, so two watchers never apply to the same map. kvstore.Bucket gains WatchFrom as a sibling to Watch, opened through the same withKV wrapper, for a caller that needs to replay from a snapshot's high-water mark rather than an UpdatesOnly live feed. * fix: cover the fence with production entry points, not a reimplementation The four fence tests recomputed hw+1 and called kvstore.WatchFrom/replay themselves instead of going through syncOnce, so they verified nats.go's resume behavior (already covered by TestResumeFromRevision_...) rather than the cache's own revision handling. A mutation inside syncOnce (hw+1 -> hw+1_000_000) passed all four. Reinstate a postSnapshotHook seam (nil in production) so the initial-sync pair can call syncOnce/installSync directly and the periodic-resync pair can call periodicResync directly, landing the write inside syncOnce's internal snapshot-to-watch window without duplicating it. The same mutation now fails all four; reverting passes all four. * feat: count the records a resumed watch replays, so the fence is observable live --- spinifex/instancecache/cache.go | 546 +++++++++++++++ spinifex/instancecache/cache_test.go | 637 ++++++++++++++++++ spinifex/instancecache/metrics.go | 115 ++++ .../resume_from_revision_test.go | 76 +++ spinifex/kvstore/bucket.go | 24 + spinifex/services/awsgw/awsgw.go | 10 + 6 files changed, 1408 insertions(+) create mode 100644 spinifex/instancecache/cache.go create mode 100644 spinifex/instancecache/cache_test.go create mode 100644 spinifex/instancecache/metrics.go create mode 100644 spinifex/instancecache/resume_from_revision_test.go diff --git a/spinifex/instancecache/cache.go b/spinifex/instancecache/cache.go new file mode 100644 index 000000000..a03633e13 --- /dev/null +++ b/spinifex/instancecache/cache.go @@ -0,0 +1,546 @@ +// Package instancecache is a read-only informer over the live instance record +// space: a cache of *vm.VM projections kept current by a JetStream KV watch +// and backstopped by a periodic whole-set resync. It answers List and Get +// from memory; nothing here fans out over NATS. +// +// Entries are immutable. An update replaces a map entry's pointer; nothing +// mutates a *vm.VM in place, and a caller must not either. +// +// The watch and a snapshot are fenced against each other by sequencing, not +// buffering: every sync — initial, periodic, or a dead watcher's replacement +// — takes a Snapshot first, then opens a new watcher that resumes from just +// past the snapshot's high-water mark and replays onto a private candidate +// map until it catches up. Only once that candidate is complete does it +// become the live map, at the same moment the new watcher becomes the active +// one; the old watcher, if there was one, is stopped and joined only after. +// Because the candidate is a map nobody else can see until that swap, a +// failed sync costs nothing: the live map and the previously active watcher +// were never touched. +// +// A record that will not decode never removes an entry: the cache only ever +// removes a key because the record space said it is gone, never because one +// read of it failed. +package instancecache + +import ( + "context" + "encoding/json" + "errors" + "log/slog" + "strings" + "sync" + "sync/atomic" + "time" + + instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance" + "github.com/mulgadc/spinifex/spinifex/kvstore" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/mulgadc/spinifex/spinifex/vm" + "github.com/nats-io/nats.go/jetstream" +) + +// Defaults for Config. +const ( + // DefaultResyncInterval is the periodic resync cadence: the completeness + // backstop for a delete or TTL expiry the watch missed. + DefaultResyncInterval = 30 * time.Second + + // DefaultRetryInterval is how long a failed sync waits before trying + // again, so an unreachable bucket is retried without spinning. + DefaultRetryInterval = 5 * time.Second +) + +// Config configures the KV bucket a Cache watches and its clocks. Bucket and +// Prefix are supplied by the caller so this package never has to know the +// daemon's bucket name or key-space constants. +type Config struct { + // Bucket names and sizes the KV bucket to open. The caller resolves the + // name and replica count, matching how every other reconcile loop in + // awsgw opens the same bucket. + Bucket kvstore.Config + + // Prefix is the key prefix identifying an instance record, e.g. "i.". + // The watch and snapshot filter is Prefix+"*", and an entry's map key is + // its record key with Prefix trimmed. + Prefix string + + ResyncInterval time.Duration + RetryInterval time.Duration +} + +// liveWatcher is one KeyWatcher plus the context that owns it and the +// goroutine draining it once it is live. +type liveWatcher struct { + kw jetstream.KeyWatcher + ctx context.Context + cancel context.CancelFunc + done chan struct{} +} + +// Cache is the informer: a live map of instance ID to *vm.VM, an account +// index over it, and the goroutine that keeps both current. Construct with +// New and start with Run. +type Cache struct { + store *kvstore.Store[vm.InstanceRecord] + prefix string + filter string + + resyncInterval time.Duration + retryInterval time.Duration + + mu sync.RWMutex + entries map[string]*vm.VM + index map[string]map[string]struct{} + ready bool + + watcherMu sync.Mutex + active *liveWatcher + + watcherLost chan struct{} + lastResync atomic.Int64 // unix nanos of the last successful sync, 0 = never + degraded atomic.Bool // true from a failed sync until the next one succeeds + + metrics *cacheMetrics + + // postSnapshotHook, when set, runs inside syncOnce after snapshotCandidate + // returns and before the resuming watcher is opened. Nil in production; a + // test seam for landing a write deterministically inside the window a + // resuming watcher has to close, which nothing else can reach because it + // is internal to syncOnce. + postSnapshotHook func() +} + +// New returns a Cache over the bucket cfg describes. Call Run to start it; +// until the first sync completes, List reports the cache not ready. +func New(js jetstream.JetStream, cfg Config) *Cache { + resync := cfg.ResyncInterval + if resync <= 0 { + resync = DefaultResyncInterval + } + retry := cfg.RetryInterval + if retry <= 0 { + retry = DefaultRetryInterval + } + c := &Cache{ + store: kvstore.New[vm.InstanceRecord](js, cfg.Bucket), + prefix: cfg.Prefix, + filter: cfg.Prefix + "*", + resyncInterval: resync, + retryInterval: retry, + entries: map[string]*vm.VM{}, + index: map[string]map[string]struct{}{}, + watcherLost: make(chan struct{}, 1), + } + c.metrics = newCacheMetrics(c) + return c +} + +// Run drives the informer until ctx is cancelled: the initial sync, then the +// periodic resync and watcher-replacement loop. Intended to run in its own +// goroutine, matching the other reconcile loops awsgw starts. +func (c *Cache) Run(ctx context.Context) { + lw := c.initialSync(ctx) + if lw == nil { + return + } + + ticker := time.NewTicker(c.resyncInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + active := c.getActive() + if active != nil { + active.cancel() + _ = active.kw.Stop() + } + return + case <-c.watcherLost: + active := c.getActive() + if active != nil { + c.replaceWatcher(ctx, active) + } + case <-ticker.C: + c.periodicResync(ctx) + } + } +} + +// Ready reports whether the initial whole-set sync has completed. Before it +// has, List and Get answer from an empty or partial view and must never be +// read as proof an instance does not exist. +func (c *Cache) Ready() bool { + c.mu.RLock() + defer c.mu.RUnlock() + return c.ready +} + +// Degraded reports whether the most recent sync attempt (initial, periodic, +// or watcher-replacement) failed. The cache keeps serving its last-known +// contents regardless, but a cache that has been degraded for a while cannot +// be trusted about absence. Not consulted by List or Get in this phase. +func (c *Cache) Degraded() bool { + return c.degraded.Load() +} + +// List returns the cached instances visible to accountID and whether the +// cache is ready to be believed about absence. IsInstanceVisibleToCaller is +// applied after the index lookup as defence in depth: the index is a +// performance structure, not the authorisation boundary. +func (c *Cache) List(_ context.Context, accountID string) ([]*vm.VM, bool) { + c.mu.RLock() + defer c.mu.RUnlock() + ids := c.index[indexKeyFor(accountID)] + out := make([]*vm.VM, 0, len(ids)) + for id := range ids { + v, ok := c.entries[id] + if !ok || !instance.IsInstanceVisibleToCaller(accountID, v) { + continue + } + out = append(out, v) + } + return out, c.ready +} + +// Get returns one instance by ID, consulting KV directly on a cache miss. A +// nil instance with a nil error means the record space has no record of it. +func (c *Cache) Get(ctx context.Context, instanceID string) (*vm.VM, error) { + c.mu.RLock() + v, ok := c.entries[instanceID] + c.mu.RUnlock() + if ok { + return v, nil + } + + rec, _, err := c.store.Get(ctx, c.prefix+instanceID) + if err != nil { + if errors.Is(err, kvstore.ErrNotFound) { + return nil, nil + } + return nil, err + } + return vm.VMFromRecord(rec), nil +} + +// getActive and setActive guard the watcher pointer separately from the map, +// so a List or Get never blocks behind a watcher transition. +func (c *Cache) getActive() *liveWatcher { + c.watcherMu.Lock() + defer c.watcherMu.Unlock() + return c.active +} + +func (c *Cache) setActive(lw *liveWatcher) { + c.watcherMu.Lock() + c.active = lw + c.watcherMu.Unlock() +} + +func (c *Cache) markResynced() { + c.lastResync.Store(time.Now().UnixNano()) + c.degraded.Store(false) +} + +// resyncAge reports how long ago the last successful sync completed. ok is +// false when no sync has ever succeeded, which the resync-age metric skips +// rather than reporting a misleading zero. +func (c *Cache) resyncAge() (time.Duration, bool) { + n := c.lastResync.Load() + if n == 0 { + return 0, false + } + return time.Since(time.Unix(0, n)), true +} + +// entriesByState groups the live map by raw instance state, for the +// entries-by-state gauge. +func (c *Cache) entriesByState() map[string]int64 { + c.mu.RLock() + defer c.mu.RUnlock() + out := make(map[string]int64, len(c.entries)) + for _, v := range c.entries { + out[string(v.Status)]++ + } + return out +} + +// sleep waits out the retry interval, reporting false if ctx ended first. +func (c *Cache) sleep(ctx context.Context) bool { + select { + case <-ctx.Done(): + return false + case <-time.After(c.retryInterval): + return true + } +} + +// syncFailed records a failed sync attempt: the metric, the degraded flag, +// and a warning log, shared by every sync path so a failure looks the same +// regardless of which one hit it. +func (c *Cache) syncFailed(ctx context.Context, err error) { + c.metrics.resyncFailed() + c.degraded.Store(true) + slog.WarnContext(ctx, "instancecache: sync failed", "err", err) +} + +// drain applies every event lw's watcher delivers, live, once the sync that +// installed lw has already caught it up: a channel that closes while lw's +// context is still live means the connection dropped out from under it, +// which is the signal Run replaces the watcher on; a channel that closes +// because the context was cancelled is an intentional teardown and reports +// nothing. An event delivered after lw has been superseded as the active +// watcher is dropped rather than applied: the watcher that replaced it is +// independently live-subscribed and already covers it, so nothing is lost, +// and this keeps it true that only the active watcher ever writes to the +// live map. +func (c *Cache) drain(lw *liveWatcher) { + defer close(lw.done) + for { + select { + case <-lw.ctx.Done(): + return + case entry, ok := <-lw.kw.Updates(): + if !ok { + if lw.ctx.Err() == nil && c.getActive() == lw { + select { + case c.watcherLost <- struct{}{}: + default: + } + } + return + } + if entry == nil { + continue + } + if c.getActive() != lw { + continue + } + c.applyLive(entry) + } + } +} + +func (c *Cache) applyLive(entry jetstream.KeyValueEntry) { + c.mu.Lock() + defer c.mu.Unlock() + c.apply(c.entries, c.index, entry) +} + +// apply decodes one watch entry into entries/index. A put that will not +// decode leaves whatever is already at that key untouched and counts a +// decode failure; a delete or purge always removes the key. Absence is +// established only by the record space saying so, never by a decode failure. +func (c *Cache) apply(entries map[string]*vm.VM, index map[string]map[string]struct{}, entry jetstream.KeyValueEntry) { + id := strings.TrimPrefix(entry.Key(), c.prefix) + switch entry.Operation() { + case jetstream.KeyValueDelete, jetstream.KeyValuePurge: + removeFrom(entries, index, id) + default: + var rec vm.InstanceRecord + if err := json.Unmarshal(entry.Value(), &rec); err != nil { + c.metrics.decodeFailure() + slog.Warn("instancecache: undecodable record, keeping previous value", + "key", entry.Key(), "err", err) + return + } + putInto(entries, index, id, vm.VMFromRecord(&rec)) + } +} + +// snapshotCandidate reads the whole record space through one consumer, +// mutually consistent per kvstore.Store.Snapshot, and builds a fresh +// entries/index pair from it. +func (c *Cache) snapshotCandidate(ctx context.Context) (map[string]*vm.VM, map[string]map[string]struct{}, uint64, error) { + items, highWater, err := c.store.Snapshot(ctx, c.filter) + if err != nil { + return nil, nil, 0, err + } + entries := make(map[string]*vm.VM, len(items)) + index := make(map[string]map[string]struct{}) + for i := range items { + id := strings.TrimPrefix(items[i].Key, c.prefix) + putInto(entries, index, id, vm.VMFromRecord(&items[i].Value)) + } + return entries, index, highWater, nil +} + +// replay applies every event kw delivers onto entries/index until the +// end-of-replay marker (nil) arrives: the candidate then reflects everything +// between the snapshot's high-water mark and the moment it caught up, which +// is the whole fence — no buffer, no mode, just sequencing. +func (c *Cache) replay(ctx context.Context, kw jetstream.KeyWatcher, entries map[string]*vm.VM, index map[string]map[string]struct{}) (int, error) { + applied := 0 + for { + select { + case <-ctx.Done(): + return applied, ctx.Err() + case entry, ok := <-kw.Updates(): + if !ok { + return applied, errors.New("instancecache: watcher closed before catching up") + } + if entry == nil { + return applied, nil + } + c.apply(entries, index, entry) + applied++ + } + } +} + +// syncOnce runs the whole build algorithm once: snapshot the record space, +// open a watcher that resumes from just past the snapshot's high-water mark, +// and replay onto the candidate until it catches up. It touches neither the +// live map nor the active watcher — installSync does that, and only once an +// attempt fully succeeds, so a failed attempt has cost nothing. +func (c *Cache) syncOnce(ctx context.Context) (*liveWatcher, map[string]*vm.VM, map[string]map[string]struct{}, error) { + entries, index, hw, err := c.snapshotCandidate(ctx) + if err != nil { + return nil, nil, nil, err + } + if c.postSnapshotHook != nil { + c.postSnapshotHook() + } + + watchCtx, cancel := context.WithCancel(ctx) + kw, err := c.store.WatchFrom(watchCtx, c.filter, hw+1) + if err != nil { + cancel() + return nil, nil, nil, err + } + + replayed, err := c.replay(watchCtx, kw, entries, index) + if err != nil { + cancel() + _ = kw.Stop() + return nil, nil, nil, err + } + c.metrics.replayApplied(replayed) + slog.Debug("instancecache: sync caught up", "snapshot_entries", len(entries), + "high_water", hw, "replayed", replayed) + + return &liveWatcher{kw: kw, ctx: watchCtx, cancel: cancel, done: make(chan struct{})}, entries, index, nil +} + +// installSync swaps entries/index in as the live map and lw in as the active +// watcher together, starts lw's live-apply goroutine, then stops and joins +// oldLw. In that order, so there is never a moment where two watchers both +// apply to the live map: oldLw, while it was still active, only ever applied +// to the map installSync is now replacing, and it stops being active in the +// same step that replacement happens. +func (c *Cache) installSync(lw *liveWatcher, entries map[string]*vm.VM, index map[string]map[string]struct{}, oldLw *liveWatcher) { + c.mu.Lock() + c.entries, c.index = entries, index + c.ready = true + c.mu.Unlock() + c.setActive(lw) + go c.drain(lw) + + if oldLw != nil { + oldLw.cancel() + _ = oldLw.kw.Stop() + <-oldLw.done + } +} + +// resyncUntilSuccess retries syncOnce against the retry interval until it +// succeeds or ctx ends, installing the result over oldLw. Used by the +// initial sync and watcher replacement, which both must block until a live +// watcher exists again; a periodic resync tries only once per tick instead. +func (c *Cache) resyncUntilSuccess(ctx context.Context, oldLw *liveWatcher) *liveWatcher { + for { + lw, entries, index, err := c.syncOnce(ctx) + if err != nil { + c.syncFailed(ctx, err) + if !c.sleep(ctx) { + return nil + } + continue + } + c.installSync(lw, entries, index, oldLw) + c.markResynced() + return lw + } +} + +// initialSync is resyncUntilSuccess with no old watcher to fence against, +// plus making the result the active watcher and recording the first +// successful resync. +func (c *Cache) initialSync(ctx context.Context) *liveWatcher { + return c.resyncUntilSuccess(ctx, nil) +} + +// periodicResync tries the build algorithm once per tick. The active watcher +// keeps applying to the live map for the whole attempt, since installSync is +// the only thing that ever changes what "the live map" is, so a failure +// costs nothing but a discarded candidate; the next tick tries again rather +// than retrying in a loop here. +func (c *Cache) periodicResync(ctx context.Context) { + oldLw := c.getActive() + if oldLw == nil { + return + } + lw, entries, index, err := c.syncOnce(ctx) + if err != nil { + c.syncFailed(ctx, err) + return + } + c.installSync(lw, entries, index, oldLw) + c.markResynced() +} + +// replaceWatcher retires a dead watcher and installs a fresh one under the +// same build algorithm, retrying against the retry interval until it +// succeeds or ctx ends. +func (c *Cache) replaceWatcher(ctx context.Context, oldLw *liveWatcher) { + lw := c.resyncUntilSuccess(ctx, oldLw) + if lw != nil { + c.metrics.watchReconnected() + } +} + +// indexKeyFor is the account index's bucket for v's owner: Global for an +// empty owner, so a legacy pre-account record files under the same bucket a +// platform-managed record does, and never under every account at once. +func indexKeyFor(accountID string) string { + if accountID == "" { + return utils.GlobalAccountID + } + return accountID +} + +func putInto(entries map[string]*vm.VM, index map[string]map[string]struct{}, id string, v *vm.VM) { + if old, ok := entries[id]; ok { + removeFromIndex(index, indexKeyFor(old.AccountID), id) + } + entries[id] = v + addToIndex(index, indexKeyFor(v.AccountID), id) +} + +func removeFrom(entries map[string]*vm.VM, index map[string]map[string]struct{}, id string) { + old, ok := entries[id] + if !ok { + return + } + delete(entries, id) + removeFromIndex(index, indexKeyFor(old.AccountID), id) +} + +func addToIndex(index map[string]map[string]struct{}, key, id string) { + set, ok := index[key] + if !ok { + set = make(map[string]struct{}) + index[key] = set + } + set[id] = struct{}{} +} + +func removeFromIndex(index map[string]map[string]struct{}, key, id string) { + set, ok := index[key] + if !ok { + return + } + delete(set, id) + if len(set) == 0 { + delete(index, key) + } +} diff --git a/spinifex/instancecache/cache_test.go b/spinifex/instancecache/cache_test.go new file mode 100644 index 000000000..afc4b1aab --- /dev/null +++ b/spinifex/instancecache/cache_test.go @@ -0,0 +1,637 @@ +// internal watcher and lifecycle state directly (there is no exported +// surface for "one live watcher" or "the candidate before install"), and the +// account-index defence-in-depth test deliberately mis-files a record by +// writing straight into the unexported maps. +// +//test:in-package — the fencing and lifecycle tests assert on the watcher's +package instancecache + +import ( + "context" + "encoding/json" + "fmt" + "sync" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/kvstore" + "github.com/mulgadc/spinifex/spinifex/resource" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/mulgadc/spinifex/spinifex/vm" + "github.com/nats-io/nats.go/jetstream" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" +) + +const ( + testPrefix = "i." + acctA = "111122223333" + acctB = "444455556666" +) + +// newTestCache starts an embedded JetStream server, opens a KV bucket for it +// and returns a Cache over that bucket plus the raw handles a test needs to +// drive the underlying key space directly. +func newTestCache(t *testing.T, mutate ...func(*Config)) (*Cache, jetstream.KeyValue, *cacheConn) { + t.Helper() + _, nc, js := testutil.StartTestJetStream(t) + + bucket := fmt.Sprintf("instances-%d", time.Now().UnixNano()) + kv, err := js.CreateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: bucket, History: 1}) + require.NoError(t, err) + + cfg := Config{ + Bucket: kvstore.Config{Name: bucket, History: 1, Replicas: 1}, + Prefix: testPrefix, + RetryInterval: 20 * time.Millisecond, + } + for _, m := range mutate { + m(&cfg) + } + return New(js, cfg), kv, &cacheConn{nc: nc, js: js} +} + +// cacheConn bundles the connection handles a test occasionally needs beyond +// the Cache and its bucket, named to avoid importing nats.go just for a type. +type cacheConn struct { + nc interface{ NumSubscriptions() int } + js jetstream.JetStream +} + +func testRecord(id, accountID string, state vm.InstanceState) *vm.InstanceRecord { + return &vm.InstanceRecord{ + Metadata: resource.Metadata{Name: id, AccountID: accountID}, + Spec: vm.InstanceSpec{InstanceType: "m7i.large"}, + Status: vm.InstanceStatus{Status: state, LastNode: "node-1"}, + } +} + +func putRecord(t *testing.T, kv jetstream.KeyValue, id string, rec *vm.InstanceRecord) uint64 { + t.Helper() + data, err := json.Marshal(rec) + require.NoError(t, err) + rev, err := kv.Put(context.Background(), testPrefix+id, data) + require.NoError(t, err) + return rev +} + +func putGarbage(t *testing.T, kv jetstream.KeyValue, id string) uint64 { + t.Helper() + rev, err := kv.Put(context.Background(), testPrefix+id, []byte("not-json")) + require.NoError(t, err) + return rev +} + +func idsOf(list []*vm.VM) []string { + out := make([]string, len(list)) + for i, v := range list { + out[i] = v.ID + } + return out +} + +func waitFor(t *testing.T, timeout time.Duration, cond func() bool) { + t.Helper() + deadline := time.Now().Add(timeout) + for { + if cond() { + return + } + if time.Now().After(deadline) { + t.Fatalf("condition not met within %s", timeout) + } + time.Sleep(5 * time.Millisecond) + } +} + +func waitForReady(t *testing.T, c *Cache) { + t.Helper() + waitFor(t, 2*time.Second, c.Ready) +} + +func waitForListLen(t *testing.T, c *Cache, accountID string, n int) []*vm.VM { + t.Helper() + var list []*vm.VM + waitFor(t, 2*time.Second, func() bool { + list, _ = c.List(context.Background(), accountID) + return len(list) == n + }) + return list +} + +// Process-wide manual metric reader, installed once per test binary so the +// resync-failure and decode-failure counters can be observed. Cumulative for +// the life of the binary, so tests read a before/after delta. +var ( + metricsReaderOnce sync.Once + metricsReader *sdkmetric.ManualReader +) + +func installMetricsReader(t *testing.T) { + t.Helper() + metricsReaderOnce.Do(func() { + metricsReader = sdkmetric.NewManualReader() + otel.SetMeterProvider(sdkmetric.NewMeterProvider(sdkmetric.WithReader(metricsReader))) + }) +} + +func counterSum(t *testing.T, name string) int64 { + t.Helper() + var rm metricdata.ResourceMetrics + require.NoError(t, metricsReader.Collect(context.Background(), &rm)) + var total int64 + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + if m.Name != name { + continue + } + sum, ok := m.Data.(metricdata.Sum[int64]) + if !ok { + continue + } + for _, dp := range sum.DataPoints { + total += dp.Value + } + } + } + return total +} + +// --- Unit, in instancecache ------------------------------------------------ + +func TestList_PutAppearsAfterWatchDelivers(t *testing.T) { + c, kv, _ := newTestCache(t) + ctx := t.Context() + go c.Run(ctx) + waitForReady(t, c) + + putRecord(t, kv, "i-1", testRecord("i-1", acctA, vm.StateRunning)) + + list := waitForListLen(t, c, acctA, 1) + require.Equal(t, "i-1", list[0].ID) +} + +func TestList_DeleteAndPurgeRemove(t *testing.T) { + c, kv, _ := newTestCache(t) + putRecord(t, kv, "i-1", testRecord("i-1", acctA, vm.StateRunning)) + putRecord(t, kv, "i-2", testRecord("i-2", acctA, vm.StateRunning)) + + ctx := t.Context() + go c.Run(ctx) + waitForReady(t, c) + waitForListLen(t, c, acctA, 2) + + require.NoError(t, kv.Delete(context.Background(), testPrefix+"i-1")) + list := waitForListLen(t, c, acctA, 1) + require.Equal(t, "i-2", list[0].ID) + + require.NoError(t, kv.Purge(context.Background(), testPrefix+"i-2")) + waitForListLen(t, c, acctA, 0) +} + +func TestPeriodicResync_RemovesEntryTheWatchNeverToldItAbout(t *testing.T) { + c, kv, _ := newTestCache(t) + putRecord(t, kv, "i-real", testRecord("i-real", acctA, vm.StateRunning)) + + ctx := t.Context() + go c.Run(ctx) + waitForReady(t, c) + waitForListLen(t, c, acctA, 1) + + // Simulate exactly what a per-key queue cannot cover: a key is gone from + // the record space, but nothing was watching when it went, so the live + // map still shows it. + c.mu.Lock() + putInto(c.entries, c.index, "i-ghost", vm.VMFromRecord(testRecord("i-ghost", acctA, vm.StateRunning))) + c.mu.Unlock() + waitForListLen(t, c, acctA, 2) + + c.periodicResync(ctx) + + list := waitForListLen(t, c, acctA, 1) + require.Equal(t, "i-real", list[0].ID) +} + +func TestSync_LeaksNothingOnRepeatedSnapshotFailure(t *testing.T) { + c, kv, conn := newTestCache(t) + putGarbage(t, kv, "i-bad") + + ctx := t.Context() + baseline := conn.nc.NumSubscriptions() + go c.Run(ctx) + + // The initial sync's snapshot fails on the garbage record every retry, so + // the replay watcher is never opened: nothing persists to leak. Snapshot + // itself briefly opens and closes its own probe watcher on every attempt, + // so the assertion is that the count always settles back to baseline, + // not that it never blips. + for range 10 { + time.Sleep(20 * time.Millisecond) + waitFor(t, time.Second, func() bool { return conn.nc.NumSubscriptions() == baseline }) + } + require.False(t, c.Ready()) +} + +func TestPeriodicResync_TTLExpiredRecordRemovedWithNoWatchEvent(t *testing.T) { + _, _, js := testutil.StartTestJetStream(t) + bucket := fmt.Sprintf("instances-ttl-%d", time.Now().UnixNano()) + kv, err := js.CreateKeyValue(context.Background(), + jetstream.KeyValueConfig{Bucket: bucket, History: 1, TTL: 200 * time.Millisecond}) + require.NoError(t, err) + + c := New(js, Config{ + Bucket: kvstore.Config{Name: bucket, History: 1, Replicas: 1}, + Prefix: testPrefix, + RetryInterval: 20 * time.Millisecond, + }) + putRecord(t, kv, "i-ttl", testRecord("i-ttl", acctA, vm.StateTerminated)) + + ctx := t.Context() + go c.Run(ctx) + waitForReady(t, c) + waitForListLen(t, c, acctA, 1) + + time.Sleep(500 * time.Millisecond) // past the bucket TTL; no delete event is ever delivered + + c.mu.RLock() + _, stillThere := c.entries["i-ttl"] + c.mu.RUnlock() + require.True(t, stillThere, "the watch alone must not have removed a TTL-expired key") + + c.periodicResync(ctx) + waitForListLen(t, c, acctA, 0) +} + +func TestPeriodicResync_UndecodableRecordMarksDegraded(t *testing.T) { + installMetricsReader(t) + c, kv, _ := newTestCache(t) + putRecord(t, kv, "i-1", testRecord("i-1", acctA, vm.StateRunning)) + + ctx := t.Context() + go c.Run(ctx) + waitForReady(t, c) + waitForListLen(t, c, acctA, 1) + + before := counterSum(t, "mulga.instancecache.resync_failures") + beforeAge, hadAge := c.resyncAge() + require.True(t, hadAge) + + putGarbage(t, kv, "i-bad") + time.Sleep(50 * time.Millisecond) + + c.periodicResync(ctx) + + require.Equal(t, before+1, counterSum(t, "mulga.instancecache.resync_failures")) + require.True(t, c.Degraded(), "a failed resync must mark the cache degraded") + + list, ready := c.List(ctx, acctA) + require.True(t, ready) + require.Len(t, list, 1) + require.Equal(t, "i-1", list[0].ID) + + afterAge, hadAge2 := c.resyncAge() + require.True(t, hadAge2) + require.GreaterOrEqual(t, afterAge, beforeAge, "a failed resync must not refresh the last-successful-resync mark") +} + +func TestDegraded_ClearedByNextSuccessfulResync(t *testing.T) { + c, kv, _ := newTestCache(t) + putRecord(t, kv, "i-1", testRecord("i-1", acctA, vm.StateRunning)) + + ctx := t.Context() + go c.Run(ctx) + waitForReady(t, c) + waitForListLen(t, c, acctA, 1) + require.False(t, c.Degraded(), "a cache that has never failed a sync must not report degraded") + + putGarbage(t, kv, "i-bad") + time.Sleep(50 * time.Millisecond) + c.periodicResync(ctx) + require.True(t, c.Degraded()) + + require.NoError(t, kv.Delete(context.Background(), testPrefix+"i-bad")) + time.Sleep(50 * time.Millisecond) + c.periodicResync(ctx) + require.False(t, c.Degraded(), "the next successful resync must clear degraded") +} + +func TestWatch_UndecodableUpdateLeavesExistingEntry(t *testing.T) { + installMetricsReader(t) + c, kv, _ := newTestCache(t) + putRecord(t, kv, "i-1", testRecord("i-1", acctA, vm.StateRunning)) + + ctx := t.Context() + go c.Run(ctx) + waitForReady(t, c) + waitForListLen(t, c, acctA, 1) + + before := counterSum(t, "mulga.instancecache.decode_failures") + putGarbage(t, kv, "i-1") + waitFor(t, 2*time.Second, func() bool { + return counterSum(t, "mulga.instancecache.decode_failures") > before + }) + + list, ready := c.List(ctx, acctA) + require.True(t, ready) + require.Len(t, list, 1) + require.Equal(t, vm.StateRunning, list[0].Status, + "an undecodable watch update must not remove or corrupt the existing entry") +} + +func TestList_NotReadyBeforeFirstSync(t *testing.T) { + c, _, _ := newTestCache(t) + list, ready := c.List(context.Background(), acctA) + require.False(t, ready) + require.Empty(t, list) +} + +// --- Fencing: replay closes the snapshot-to-watch window -------------------- +// +// The window the fence has to cover — after Snapshot has already read the +// record space, before the resuming watcher opens — is internal to syncOnce, +// so postSnapshotHook is the seam that lands a write inside it. Each test +// drives the real production path (syncOnce/installSync or periodicResync) +// rather than reimplementing the window's steps itself, so a wrong revision +// or a dropped WatchFrom call in production is what these catch, not +// whatever a test independently recomputes. + +func TestFence_PutDuringInitialSnapshot_SurvivesInstall(t *testing.T) { + c, kv, _ := newTestCache(t) + putRecord(t, kv, "i-seed", testRecord("i-seed", acctA, vm.StateRunning)) + + c.postSnapshotHook = func() { + putRecord(t, kv, "i-new", testRecord("i-new", acctA, vm.StateRunning)) + } + + lw, entries, index, err := c.syncOnce(context.Background()) + require.NoError(t, err) + c.installSync(lw, entries, index, nil) + + list, ready := c.List(context.Background(), acctA) + require.True(t, ready) + require.ElementsMatch(t, []string{"i-seed", "i-new"}, idsOf(list), + "a put landing after the snapshot but before the watcher opened must be replayed") +} + +func TestFence_DeleteDuringInitialSnapshot_RemovedFromInstall(t *testing.T) { + c, kv, _ := newTestCache(t) + putRecord(t, kv, "i-seed", testRecord("i-seed", acctA, vm.StateRunning)) + putRecord(t, kv, "i-doomed", testRecord("i-doomed", acctA, vm.StateRunning)) + + c.postSnapshotHook = func() { + require.NoError(t, kv.Delete(context.Background(), testPrefix+"i-doomed")) + } + + lw, entries, index, err := c.syncOnce(context.Background()) + require.NoError(t, err) + c.installSync(lw, entries, index, nil) + + list, ready := c.List(context.Background(), acctA) + require.True(t, ready) + require.Equal(t, []string{"i-seed"}, idsOf(list), + "a delete landing after the snapshot but before the watcher opened must not survive as the snapshot's stale copy") +} + +func TestFence_PutDuringPeriodicResync_SurvivesInstall(t *testing.T) { + c, kv, _ := newTestCache(t) + putRecord(t, kv, "i-seed", testRecord("i-seed", acctA, vm.StateRunning)) + + ctx := t.Context() + go c.Run(ctx) + waitForReady(t, c) + waitForListLen(t, c, acctA, 1) + + c.postSnapshotHook = func() { + putRecord(t, kv, "i-new", testRecord("i-new", acctA, vm.StateRunning)) + } + c.periodicResync(ctx) + c.postSnapshotHook = nil + + list, ready := c.List(ctx, acctA) + require.True(t, ready) + require.ElementsMatch(t, []string{"i-seed", "i-new"}, idsOf(list), + "a put landing after a periodic resync's snapshot but before the new watcher opened must be replayed") +} + +func TestFence_DeleteDuringPeriodicResync_RemovedFromInstall(t *testing.T) { + c, kv, _ := newTestCache(t) + putRecord(t, kv, "i-seed", testRecord("i-seed", acctA, vm.StateRunning)) + putRecord(t, kv, "i-doomed", testRecord("i-doomed", acctA, vm.StateRunning)) + + ctx := t.Context() + go c.Run(ctx) + waitForReady(t, c) + waitForListLen(t, c, acctA, 2) + + c.postSnapshotHook = func() { + require.NoError(t, kv.Delete(context.Background(), testPrefix+"i-doomed")) + } + c.periodicResync(ctx) + c.postSnapshotHook = nil + + list, ready := c.List(ctx, acctA) + require.True(t, ready) + require.Equal(t, []string{"i-seed"}, idsOf(list), + "a delete landing after a periodic resync's snapshot but before the new watcher opened must not be overwritten by the snapshot's stale copy") +} + +// --- Lifecycle --------------------------------------------------------------- + +func TestPeriodicResync_FailureLeavesLiveWatcherAndMapServing(t *testing.T) { + c, kv, conn := newTestCache(t) + putRecord(t, kv, "i-1", testRecord("i-1", acctA, vm.StateRunning)) + + ctx := t.Context() + go c.Run(ctx) + waitForReady(t, c) + waitForListLen(t, c, acctA, 1) + + baseline := conn.nc.NumSubscriptions() + putGarbage(t, kv, "i-bad") + + for range 10 { + c.periodicResync(ctx) + list, ready := c.List(ctx, acctA) + require.True(t, ready) + require.Len(t, list, 1) + require.Equal(t, "i-1", list[0].ID) + require.Equal(t, baseline, conn.nc.NumSubscriptions(), + "a failed periodic resync must not leave a stray subscription behind") + } +} + +func TestPeriodicResync_FailureLosesNothing(t *testing.T) { + c, kv, _ := newTestCache(t) + putRecord(t, kv, "i-seed", testRecord("i-seed", acctA, vm.StateRunning)) + + ctx := t.Context() + go c.Run(ctx) + waitForReady(t, c) + waitForListLen(t, c, acctA, 1) + + putGarbage(t, kv, "i-bad") // every future resync now fails against the same bad key + + for i := range 10 { + id := fmt.Sprintf("i-live-%d", i) + putRecord(t, kv, id, testRecord(id, acctA, vm.StateRunning)) + waitFor(t, 2*time.Second, func() bool { + list, _ := c.List(ctx, acctA) + for _, v := range list { + if v.ID == id { + return true + } + } + return false + }) + + c.periodicResync(ctx) + + list, ready := c.List(ctx, acctA) + require.True(t, ready) + require.Contains(t, idsOf(list), id) + require.Contains(t, idsOf(list), "i-seed") + } + + require.NoError(t, kv.Delete(context.Background(), testPrefix+"i-seed")) + waitFor(t, 2*time.Second, func() bool { + list, _ := c.List(ctx, acctA) + for _, v := range list { + if v.ID == "i-seed" { + return false + } + } + return true + }) +} + +func TestPeriodicResync_TwentyConsecutiveSuccessesOneWatcherEach(t *testing.T) { + c, kv, conn := newTestCache(t) + putRecord(t, kv, "i-seed", testRecord("i-seed", acctA, vm.StateRunning)) + + ctx := t.Context() + go c.Run(ctx) + waitForReady(t, c) + waitForListLen(t, c, acctA, 1) + + baseline := conn.nc.NumSubscriptions() + + // Each periodic resync opens a fresh watcher and retires the last one, so + // the subscription count stays flat even though the watcher identity + // changes every time. + prev := c.getActive() + for range 20 { + c.periodicResync(ctx) + require.Equal(t, baseline, conn.nc.NumSubscriptions()) + cur := c.getActive() + require.NotSame(t, prev, cur, "each successful periodic resync installs a new watcher") + prev = cur + } + + list, ready := c.List(ctx, acctA) + require.True(t, ready) + require.Equal(t, []string{"i-seed"}, idsOf(list)) +} + +func TestReplaceWatcher_OldStoppedAndJoinedBeforeNewGoesLive(t *testing.T) { + c, kv, conn := newTestCache(t) + putRecord(t, kv, "i-1", testRecord("i-1", acctA, vm.StateRunning)) + + ctx := t.Context() + go c.Run(ctx) + waitForReady(t, c) + waitForListLen(t, c, acctA, 1) + + oldLw := c.getActive() + baseline := conn.nc.NumSubscriptions() + + require.NoError(t, oldLw.kw.Stop()) // simulate the connection dropping this watcher + + waitFor(t, 2*time.Second, func() bool { + select { + case <-oldLw.done: + default: + return false + } + next := c.getActive() + return next != nil && next != oldLw + }) + + select { + case <-oldLw.done: + default: + t.Fatal("old watcher must be joined once replacement completes") + } + + waitFor(t, 2*time.Second, func() bool { return conn.nc.NumSubscriptions() == baseline }) + + list, ready := c.List(ctx, acctA) + require.True(t, ready) + require.Len(t, list, 1) + require.Equal(t, "i-1", list[0].ID) + + putRecord(t, kv, "i-2", testRecord("i-2", acctA, vm.StateRunning)) + waitForListLen(t, c, acctA, 2) +} + +// --- The account index ------------------------------------------------------ + +func TestList_ReturnsOnlyThatAccountsInstances(t *testing.T) { + c, kv, _ := newTestCache(t) + putRecord(t, kv, "i-a", testRecord("i-a", acctA, vm.StateRunning)) + putRecord(t, kv, "i-b", testRecord("i-b", acctB, vm.StateRunning)) + + ctx := t.Context() + go c.Run(ctx) + waitForReady(t, c) + waitForListLen(t, c, acctA, 1) + waitForListLen(t, c, acctB, 1) + + listA, _ := c.List(ctx, acctA) + require.Equal(t, []string{"i-a"}, idsOf(listA)) + + listB, _ := c.List(ctx, acctB) + require.Equal(t, []string{"i-b"}, idsOf(listB)) +} + +func TestList_GlobalDoesNotSeeOtherAccountsInstances(t *testing.T) { + c, kv, _ := newTestCache(t) + putRecord(t, kv, "i-cust", testRecord("i-cust", acctA, vm.StateRunning)) + putRecord(t, kv, "i-global", testRecord("i-global", utils.GlobalAccountID, vm.StateRunning)) + putRecord(t, kv, "i-legacy", testRecord("i-legacy", "", vm.StateRunning)) + + ctx := t.Context() + go c.Run(ctx) + waitForReady(t, c) + waitFor(t, 2*time.Second, func() bool { + c.mu.RLock() + defer c.mu.RUnlock() + return len(c.entries) == 3 + }) + + listGlobal, ready := c.List(ctx, utils.GlobalAccountID) + require.True(t, ready) + require.ElementsMatch(t, []string{"i-global", "i-legacy"}, idsOf(listGlobal)) + + listCust, _ := c.List(ctx, acctA) + require.Equal(t, []string{"i-cust"}, idsOf(listCust)) +} + +func TestList_VisibilityAppliedAfterIndexLookup(t *testing.T) { + c, _, _ := newTestCache(t) + + // Deliberately mis-file: the index says i-misfiled belongs to acctA, but + // the record itself is owned by acctB. IsInstanceVisibleToCaller must + // still reject it at the List seam. + c.mu.Lock() + c.entries["i-misfiled"] = vm.VMFromRecord(testRecord("i-misfiled", acctB, vm.StateRunning)) + addToIndex(c.index, acctA, "i-misfiled") + c.ready = true + c.mu.Unlock() + + list, ready := c.List(context.Background(), acctA) + require.True(t, ready) + require.Empty(t, list) +} diff --git a/spinifex/instancecache/metrics.go b/spinifex/instancecache/metrics.go new file mode 100644 index 000000000..dc8027adb --- /dev/null +++ b/spinifex/instancecache/metrics.go @@ -0,0 +1,115 @@ +package instancecache + +import ( + "context" + "log/slog" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" +) + +// meterName is the instrumentation scope for every instancecache metric. +const meterName = "github.com/mulgadc/spinifex/spinifex/instancecache" + +// cacheMetrics holds one Cache's instruments. Registered per instance, not as +// a package-level singleton, because the gauges read that instance's own +// state through a callback closure. +type cacheMetrics struct { + resyncFailures metric.Int64Counter + watchReconnects metric.Int64Counter + decodeFailures metric.Int64Counter + replayEvents metric.Int64Counter +} + +// newCacheMetrics registers c's instruments against the global meter +// provider. A registration failure is handled by otel.Handle and leaves the +// affected instrument nil; every recorder below tolerates a nil instrument. +func newCacheMetrics(c *Cache) *cacheMetrics { + meter := otel.Meter(meterName) + m := &cacheMetrics{} + + var err error + m.resyncFailures, err = meter.Int64Counter("mulga.instancecache.resync_failures", + metric.WithDescription("Count of failed initial, periodic or watcher-replacement syncs."), + metric.WithUnit("{failure}")) + if err != nil { + otel.Handle(err) + } + m.watchReconnects, err = meter.Int64Counter("mulga.instancecache.watch_reconnects", + metric.WithDescription("Count of times the KV watch was replaced after dying."), + metric.WithUnit("{reconnect}")) + if err != nil { + otel.Handle(err) + } + m.replayEvents, err = meter.Int64Counter("mulga.instancecache.replay_events", + metric.WithDescription("Count of records the snapshot missed and the resumed watch replayed before the cache went live."), + metric.WithUnit("{record}")) + if err != nil { + otel.Handle(err) + } + m.decodeFailures, err = meter.Int64Counter("mulga.instancecache.decode_failures", + metric.WithDescription("Count of instance records that failed to decode. The offending key is logged, not attributed, to bound cardinality."), + metric.WithUnit("{failure}")) + if err != nil { + otel.Handle(err) + } + + if _, err := meter.Int64ObservableGauge("mulga.instancecache.entries", + metric.WithDescription("Cached instance count by raw instance state."), + metric.WithUnit("{instance}"), + metric.WithInt64Callback(func(_ context.Context, o metric.Int64Observer) error { + for state, n := range c.entriesByState() { + o.Observe(n, metric.WithAttributeSet(attribute.NewSet(attribute.String("state", state)))) + } + return nil + })); err != nil { + otel.Handle(err) + } + + if _, err := meter.Float64ObservableGauge("mulga.instancecache.resync_age_seconds", + metric.WithDescription("Age of the last successful resync. Absent until the first one completes."), + metric.WithUnit("s"), + metric.WithFloat64Callback(func(_ context.Context, o metric.Float64Observer) error { + if age, ok := c.resyncAge(); ok { + o.Observe(age.Seconds()) + } + return nil + })); err != nil { + otel.Handle(err) + } + + return m +} + +func (m *cacheMetrics) resyncFailed() { + if m.resyncFailures == nil { + return + } + m.resyncFailures.Add(context.Background(), 1) +} + +func (m *cacheMetrics) watchReconnected() { + if m.watchReconnects == nil { + return + } + m.watchReconnects.Add(context.Background(), 1) +} + +// replayApplied records how many records the resumed watch had to replay. +// A zero count is recorded too: it is the difference between a quiet cluster +// and an instrument that never fired. +func (m *cacheMetrics) replayApplied(n int) { + if m.replayEvents == nil { + return + } + m.replayEvents.Add(context.Background(), int64(n)) +} + +func (m *cacheMetrics) decodeFailure() { + if m.decodeFailures == nil { + slog.Warn("instancecache: decode failure counter unavailable") + return + } + m.decodeFailures.Add(context.Background(), 1) +} diff --git a/spinifex/instancecache/resume_from_revision_test.go b/spinifex/instancecache/resume_from_revision_test.go new file mode 100644 index 000000000..e3f1e9864 --- /dev/null +++ b/spinifex/instancecache/resume_from_revision_test.go @@ -0,0 +1,76 @@ +package instancecache_test + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/kvstore" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/mulgadc/spinifex/spinifex/vm" + "github.com/nats-io/nats.go/jetstream" + "github.com/stretchr/testify/require" +) + +// TestResumeFromRevision_ReplaysWhatTheSnapshotMissed documents the server +// behaviour the whole Cache design rests on: a watcher opened AFTER a +// snapshot, resuming from just past the snapshot's high-water mark, replays +// exactly the events the snapshot missed — even though nothing was watching +// while they happened — and nothing the snapshot already covered. If a +// nats.go upgrade ever changes this, this test fails loudly and points +// straight at the cause, rather than the cache's own tests failing in a way +// that looks like a cache bug. +func TestResumeFromRevision_ReplaysWhatTheSnapshotMissed(t *testing.T) { + _, _, js := testutil.StartTestJetStream(t) + ctx := context.Background() + + store := kvstore.New[vm.InstanceRecord](js, kvstore.Config{Name: "resume-bucket", History: 1, Replicas: 1}) + + rec := func(id string) *vm.InstanceRecord { + r := &vm.InstanceRecord{} + r.Metadata.Name = id + return r + } + + // Three keys exist before the snapshot. + for i := range 3 { + require.NoError(t, store.Set(ctx, fmt.Sprintf("i.pre-%d", i), rec(fmt.Sprintf("pre-%d", i)))) + } + + items, highWater, err := store.Snapshot(ctx, "i.*") + require.NoError(t, err) + require.Len(t, items, 3, "snapshot should see the three pre-existing keys") + + // These land AFTER the snapshot and BEFORE any watcher exists. This is + // exactly the window a buffering fence would otherwise have to cover. + require.NoError(t, store.Set(ctx, "i.post-1", rec("post-1"))) + require.NoError(t, store.Delete(ctx, "i.pre-0")) + + // Open the watcher only now, resuming from just past the snapshot. + kw, err := js.KeyValue(ctx, "resume-bucket") + require.NoError(t, err) + w, err := kw.Watch(ctx, "i.*", jetstream.ResumeFromRevision(highWater+1)) + require.NoError(t, err) + defer func() { _ = w.Stop() }() + + seen := map[string]jetstream.KeyValueOp{} + timeout := time.After(10 * time.Second) + for { + select { + case e := <-w.Updates(): + if e == nil { + // End-of-replay marker: everything the resume owed us has arrived. + require.Equal(t, jetstream.KeyValuePut, seen["i.post-1"], + "a put made after the snapshot, before the watcher existed, must be replayed") + require.Equal(t, jetstream.KeyValueDelete, seen["i.pre-0"], + "a delete made after the snapshot, before the watcher existed, must be replayed") + require.NotContains(t, seen, "i.pre-1", "keys already in the snapshot must not be replayed") + return + } + seen[e.Key()] = e.Operation() + case <-timeout: + t.Fatalf("no end-of-replay marker within 10s; saw %v", seen) + } + } +} diff --git a/spinifex/kvstore/bucket.go b/spinifex/kvstore/bucket.go index 6aac39f6c..31094b240 100644 --- a/spinifex/kvstore/bucket.go +++ b/spinifex/kvstore/bucket.go @@ -271,6 +271,30 @@ func (b *Bucket) Watch(ctx context.Context, filter string) (jetstream.KeyWatcher return w, nil } +// WatchFrom returns a watcher over the keys matching filter that resumes +// from revision: it replays every value at or after revision, followed by +// the end-of-replay marker, before turning into an ordinary live feed. It +// takes no UpdatesOnly, unlike Watch, because the whole point is to replay +// what a Snapshot's high-water mark missed — a caller rebuilding a cache +// from a snapshot uses this to close that window without a buffering fence. +func (b *Bucket) WatchFrom(ctx context.Context, filter string, revision uint64) (jetstream.KeyWatcher, error) { + var w jetstream.KeyWatcher + // Wrapped inside the closure, so a bucket that could not be opened at all + // reports the configured reason rather than a watch failure on top of it. + err := b.withKV(ctx, func(kv jetstream.KeyValue) error { + var err error + w, err = kv.Watch(ctx, filter, jetstream.ResumeFromRevision(revision)) + if err != nil { + return fmt.Errorf("kvstore: watch %s %s: %w", b.cfg.Name, filter, err) + } + return nil + }) + if err != nil { + return nil, 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/services/awsgw/awsgw.go b/spinifex/services/awsgw/awsgw.go index 7ac328f42..53f204fab 100644 --- a/spinifex/services/awsgw/awsgw.go +++ b/spinifex/services/awsgw/awsgw.go @@ -30,6 +30,7 @@ import ( handlers_ochrevector "github.com/mulgadc/spinifex/spinifex/handlers/ochrevector" handlers_quota "github.com/mulgadc/spinifex/spinifex/handlers/quota" handlers_sts "github.com/mulgadc/spinifex/spinifex/handlers/sts" + "github.com/mulgadc/spinifex/spinifex/instancecache" "github.com/mulgadc/spinifex/spinifex/kvstore" "github.com/mulgadc/spinifex/spinifex/network/reconcile" "github.com/mulgadc/spinifex/spinifex/objectstore" @@ -570,6 +571,15 @@ func launchService(config *config.ClusterConfig) error { // disabled. go gw.Quota.RunBedrockRPMSync(janitorCtx, js, len(config.Nodes)) + // Instance cache: a read-only informer over the live instance record space, + // started now so it is warm well before anything reads it. Nothing consults + // it yet; the describe path still serves from the fan-out and KV. + instanceCache := instancecache.New(js, instancecache.Config{ + Bucket: kvstore.Config{Name: daemon.InstanceStateBucket, History: 1, Replicas: len(config.Nodes)}, + Prefix: daemon.InstanceRecordPrefix, + }) + go instanceCache.Run(janitorCtx) + handler := gw.SetupRoutes() // Load TLS certificate