diff --git a/catalog/consumer.go b/catalog/consumer.go new file mode 100644 index 0000000..1e46a03 --- /dev/null +++ b/catalog/consumer.go @@ -0,0 +1,323 @@ +package catalog + +// ProjectionStore — the consumer half of the projection seam (see +// docs/catalog-projection.md). A commerce service (cart, checkout, order, UCP) +// feeds it the catalog.projection_published frames it receives off the bus and +// reads catalog facts back by SKU. The store owns a LOCAL read-only projection: +// +// - A full snapshot stream (begin → chunk* → end, one MetaEpoch) is assembled +// into the store's OWN .bin (SnapshotWriter), atomically swapped in as the +// mmap base (SnapshotHolder). The base is the cold-start substrate — it +// survives a restart (reopened from disk), so the consumer serves reads +// immediately without waiting for a bus replay. +// - Per-batch deltas of the SAME epoch are applied to an in-memory overlay on +// top of the base (overlay wins; a delete leaves a tombstone that shadows the +// base and the fallback). The next snapshot folds the overlay into a new base. +// +// Read order on Get: overlay → mmap base → optional caller fallback (e.g. an +// HTTP fetch from the reader) — last-resort only. The access pattern is point +// lookup by SKU; nothing enumerates the catalog. +// +// No shared volume: each consumer builds its own base from the bus. Concurrency: +// HandleFrame is single-writer (drive it from one delivery goroutine); Get is +// safe from many readers. + +import ( + "fmt" + "log" + "os" + "path/filepath" + "strconv" + "strings" + "sync" +) + +// ProjectionFallback resolves an SKU the local store doesn't have yet (cold +// start before the first snapshot, or an item added since the last base). It +// returns a raw Projection — the store applies the SAME Transform to it, so a +// fallback hit is shaped identically to bus-fed data. Return (_, false) to +// report a genuine miss. +type ProjectionFallback func(sku string) (Projection, bool) + +// ProjectionTransform projects the wire Projection down to the consumer's own +// stored type T. It is MANDATORY: it is the one place a consumer declares which +// catalog facts it actually needs (and any derivation, e.g. tax_class → rate), +// so its local .bin holds only that — smaller, and read-ready with no per-Get +// conversion. Every value entering the store — snapshot, delta, and fallback — +// passes through it exactly once. +type ProjectionTransform[T any] func(Projection) T + +// ProjectionStore is a consumer's local, cold-start-ready projection cache of T. +type ProjectionStore[T any] struct { + dir, name string + transform ProjectionTransform[T] + fallback ProjectionFallback + holder *SnapshotHolder[T] // mmap base; atomic-swap + drain-grace + + mu sync.RWMutex + epoch int64 // epoch of the live base + overlay map[string]overlayEntry[T] // same-epoch deltas layered on the base + + // in-progress snapshot assembly (one epoch at a time) + building *SnapshotWriter[T] + buildingEpoch int64 + buildingSeen int +} + +type overlayEntry[T any] struct { + val T + deleted bool +} + +// NewProjectionStore opens a store of T backed by dir/name (its own .bin). +// transform is REQUIRED — it defines the consumer's projection shape (see +// ProjectionTransform). If a base from a previous run is present it is reopened +// immediately (cold-start ready); its epoch is read from the sidecar so +// same-epoch deltas can resume. fallback may be nil. Close() when done. +func NewProjectionStore[T any](dir, name string, transform ProjectionTransform[T], fallback ProjectionFallback) (*ProjectionStore[T], error) { + if dir == "" || name == "" { + return nil, fmt.Errorf("catalog: ProjectionStore needs dir and name") + } + if transform == nil { + return nil, fmt.Errorf("catalog: ProjectionStore needs a transform (declare which catalog facts this consumer stores)") + } + s := &ProjectionStore[T]{ + dir: dir, + name: name, + transform: transform, + fallback: fallback, + holder: NewSnapshotHolder[T](0), + overlay: make(map[string]overlayEntry[T]), + } + path := filepath.Join(dir, name) + if err := s.holder.Open(path); err == nil { + s.epoch = readEpochSidecar(path) + log.Printf("catalog: ProjectionStore reopened base %s (epoch=%d, %d entries)", path, s.epoch, s.holder.Len()) + } else if !os.IsNotExist(err) { + // A corrupt/partial base on disk shouldn't crash boot — start empty and + // wait for the next snapshot; reads fall back meanwhile. + log.Printf("catalog: ProjectionStore ignoring unreadable base %s: %v", path, err) + } + return s, nil +} + +// HandleFrame applies one catalog.projection_published event (snapshot frame or +// delta), routed by its Meta. Drive this from a single delivery goroutine. A +// returned error means the payload was undecodable (the caller may NACK); a +// frame that simply doesn't apply (stale epoch, wrong assembly state) returns +// nil so the delivery is acked. +func (s *ProjectionStore[T]) HandleFrame(meta map[string]string, payload []byte) error { + epoch, _ := strconv.ParseInt(meta[MetaEpoch], 10, 64) + switch meta[MetaSnapshotPhase] { + case SnapshotBegin: + return s.beginSnapshot(epoch, meta[MetaTenant]) + case SnapshotChunk: + return s.chunkSnapshot(epoch, payload) + case SnapshotEnd: + return s.endSnapshot(epoch, meta) + case "": + return s.applyDelta(epoch, payload) + default: + return nil // unknown phase — ignore forward-compatibly + } +} + +func (s *ProjectionStore[T]) beginSnapshot(epoch int64, tenant string) error { + s.mu.Lock() + defer s.mu.Unlock() + if epoch <= s.epoch { + return nil // we already hold this base or newer + } + // Start (or restart) assembly for this epoch. A new begin supersedes any + // half-built snapshot (e.g. a producer that aborted mid-stream). + s.building = NewSnapshotWriter[T]().WithTenant(tenant) + s.buildingEpoch = epoch + s.buildingSeen = 0 + return nil +} + +func (s *ProjectionStore[T]) chunkSnapshot(epoch int64, payload []byte) error { + ups, err := DecodeUpdates(payload) + if err != nil { + return fmt.Errorf("catalog: decode snapshot chunk: %w", err) + } + s.mu.Lock() + defer s.mu.Unlock() + if s.building == nil || epoch != s.buildingEpoch { + return nil // not assembling this epoch + } + for _, u := range ups { + if u.Deleted || u.SKU == "" { + continue // a snapshot carries live items only + } + // Transform applied here — the base stores the consumer's shape, not the + // full Projection. + if err := s.building.Append(u.SKU, s.transform(u.Projection)); err != nil { + log.Printf("catalog: ProjectionStore skip %q in snapshot: %v", u.SKU, err) + continue + } + s.buildingSeen++ + } + return nil +} + +func (s *ProjectionStore[T]) endSnapshot(epoch int64, meta map[string]string) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.building == nil || epoch != s.buildingEpoch { + return nil + } + w := s.building + s.building = nil + + // Completeness guard: if begin/end advertised a count and we appended fewer + // (a lost chunk), DON'T swap — keep the old base and wait for the next + // snapshot. Serving an incomplete catalog silently is worse than stale. + if cs := meta[MetaCount]; cs != "" { + if want, perr := strconv.Atoi(cs); perr == nil && want != s.buildingSeen { + log.Printf("catalog: ProjectionStore snapshot epoch=%d incomplete (%d/%d) — discarding, keeping base epoch=%d", + epoch, s.buildingSeen, want, s.epoch) + return nil + } + } + + path, err := w.Write(s.dir, s.name) + if err != nil { + return fmt.Errorf("catalog: write snapshot base: %w", err) + } + // Persist the epoch sidecar (atomic) BEFORE swapping the in-memory reader, so + // a reopened store's epoch always matches the .bin already on disk (w.Write + // renamed the new file into place above). Writing it after Reload would leave + // a crash window where the new .bin carries a stale/absent epoch — the store + // would then mislabel its base and drop the current epoch's deltas on restart. + writeEpochSidecar(path, epoch) + if err := s.holder.Reload(path); err != nil { + return fmt.Errorf("catalog: reload snapshot base: %w", err) + } + s.epoch = epoch + s.overlay = make(map[string]overlayEntry[T]) // base now includes everything ≤ epoch + log.Printf("catalog: ProjectionStore base swapped epoch=%d (%d entries)", epoch, s.buildingSeen) + return nil +} + +func (s *ProjectionStore[T]) applyDelta(epoch int64, payload []byte) error { + ups, err := DecodeUpdates(payload) + if err != nil { + return fmt.Errorf("catalog: decode delta: %w", err) + } + s.mu.Lock() + defer s.mu.Unlock() + switch { + case s.building != nil: + // Under FIFO delivery a delta won't interleave a snapshot stream; if it + // does, the snapshot is authoritative for its epoch, so drop the delta. + return nil + case epoch < s.epoch: + return nil // stale — predates our base + case epoch > s.epoch: + // CONTRACT: a future-epoch delta is DROPPED, not buffered. Under FIFO + // per-queue delivery begin(N) always precedes delta(N), so this only + // happens after a reconnect that missed the whole snapshot — which a + // buffer couldn't reassemble anyway. Recovery is the next snapshot of + // this epoch; the consumer's fallback covers reads until then. + return nil + } + for _, u := range ups { + if u.SKU == "" { + continue + } + if u.Deleted { + s.overlay[u.SKU] = overlayEntry[T]{deleted: true} + continue + } + // Transform applied here too — overlay holds the consumer's shape. + s.overlay[u.SKU] = overlayEntry[T]{val: s.transform(u.Projection)} + } + return nil +} + +// Get resolves a value by SKU: overlay (incl. tombstone) → mmap base → +// fallback. A tombstoned SKU returns (zero, false) and never reaches the +// fallback — the catalog explicitly removed it. A fallback hit is run through +// the same Transform, so it's shaped like everything else. +func (s *ProjectionStore[T]) Get(sku string) (T, bool) { + var zero T + s.mu.RLock() + e, ok := s.overlay[sku] + s.mu.RUnlock() + if ok { + if e.deleted { + return zero, false + } + return e.val, true + } + if v, ok := s.holder.Get(sku); ok { + return v, true + } + if s.fallback != nil { + if p, ok := s.fallback(sku); ok { + return s.transform(p), true + } + } + return zero, false +} + +// IsDeleted reports whether the overlay holds a tombstone for sku — a delete +// seen on the bus since the current base. Lets a caller skip a downstream fetch +// for an SKU the catalog just removed. After a snapshot folds the delete into a +// new base the SKU is simply absent (IsDeleted → false); "absent" is then the +// caller's fetch/miss path, which a deleted item resolves as a downstream 404. +func (s *ProjectionStore[T]) IsDeleted(sku string) bool { + s.mu.RLock() + defer s.mu.RUnlock() + e, ok := s.overlay[sku] + return ok && e.deleted +} + +// Epoch returns the epoch of the current base (0 if none yet). +func (s *ProjectionStore[T]) Epoch() int64 { + s.mu.RLock() + defer s.mu.RUnlock() + return s.epoch +} + +// Stats returns observability counters (base size, overlay size, epoch). +func (s *ProjectionStore[T]) Stats() (epoch int64, baseCount, overlayCount int) { + s.mu.RLock() + defer s.mu.RUnlock() + return s.epoch, s.holder.Len(), len(s.overlay) +} + +// Close releases the mmap base. +func (s *ProjectionStore[T]) Close() error { return s.holder.Close() } + +// ── epoch sidecar ──────────────────────────────────────────────────────────── +// The .bin header has no room for the epoch, so it's persisted next to it as +// ".epoch" — read on boot so a reopened base can resume same-epoch deltas. + +func epochSidecarPath(binPath string) string { return binPath + ".epoch" } + +// writeEpochSidecar writes the epoch atomically (temp+rename) so a crash mid-write +// can't leave a torn file that parses as a bogus epoch — a reader either sees the +// old epoch or the new one, never garbage. +func writeEpochSidecar(binPath string, epoch int64) { + p := epochSidecarPath(binPath) + tmp := p + ".tmp" + if err := os.WriteFile(tmp, []byte(strconv.FormatInt(epoch, 10)), 0o644); err != nil { + log.Printf("catalog: ProjectionStore epoch sidecar write: %v", err) + return + } + if err := os.Rename(tmp, p); err != nil { + log.Printf("catalog: ProjectionStore epoch sidecar rename: %v", err) + _ = os.Remove(tmp) + } +} + +func readEpochSidecar(binPath string) int64 { + b, err := os.ReadFile(epochSidecarPath(binPath)) + if err != nil { + return 0 + } + n, _ := strconv.ParseInt(strings.TrimSpace(string(b)), 10, 64) + return n +} diff --git a/catalog/consumer_test.go b/catalog/consumer_test.go new file mode 100644 index 0000000..0739c2e --- /dev/null +++ b/catalog/consumer_test.go @@ -0,0 +1,225 @@ +package catalog + +import ( + "strconv" + "testing" + + "git.k6n.net/mats/platform/money" +) + +func snapMeta(phase string, epoch int64, seq, count int) map[string]string { + m := map[string]string{MetaEpoch: strconv.FormatInt(epoch, 10), MetaTenant: "SE"} + if phase != "" { + m[MetaSnapshotPhase] = phase + m[MetaSeq] = strconv.Itoa(seq) + } + if phase == SnapshotBegin || phase == SnapshotEnd { + m[MetaCount] = strconv.Itoa(count) + } + return m +} + +func mustEnc(t *testing.T, ups []ProjectionUpdate) []byte { + t.Helper() + b, err := EncodeUpdates(ups) + if err != nil { + t.Fatalf("encode: %v", err) + } + return b +} + +// identity is the transform used by tests: store the full Projection (T = Projection). +func identity(p Projection) Projection { return p } + +// emitSnapshot drives a full begin→chunk→end stream of projs at epoch. +func emitSnapshot(t *testing.T, s *ProjectionStore[Projection], epoch int64, projs []Projection) { + t.Helper() + if err := s.HandleFrame(snapMeta(SnapshotBegin, epoch, 0, len(projs)), mustEnc(t, nil)); err != nil { + t.Fatalf("begin: %v", err) + } + ups := make([]ProjectionUpdate, len(projs)) + for i, p := range projs { + ups[i] = ProjectionUpdate{Projection: p} + } + if err := s.HandleFrame(snapMeta(SnapshotChunk, epoch, 0, 0), mustEnc(t, ups)); err != nil { + t.Fatalf("chunk: %v", err) + } + if err := s.HandleFrame(snapMeta(SnapshotEnd, epoch, 1, len(projs)), mustEnc(t, nil)); err != nil { + t.Fatalf("end: %v", err) + } +} + +func proj(sku string, price money.Cents) Projection { + return Projection{ID: "sku:" + sku, SKU: sku, PriceIncVat: price} +} + +func newStore(t *testing.T, fb ProjectionFallback) *ProjectionStore[Projection] { + t.Helper() + s, err := NewProjectionStore(t.TempDir(), "consumer.bin", identity, fb) + if err != nil { + t.Fatalf("new store: %v", err) + } + t.Cleanup(func() { _ = s.Close() }) + return s +} + +func TestProjectionStore_SnapshotThenGet(t *testing.T) { + s := newStore(t, nil) + emitSnapshot(t, s, 100, []Projection{proj("A", 100_00), proj("B", 150_00)}) + + if got, ok := s.Get("A"); !ok || got.PriceIncVat != 100_00 { + t.Fatalf("A = %+v ok=%v, want 10000", got, ok) + } + if e, base, ov := s.Stats(); e != 100 || base != 2 || ov != 0 { + t.Fatalf("stats = epoch %d base %d overlay %d, want 100/2/0", e, base, ov) + } +} + +func TestProjectionStore_DeltaOverlayWinsOverBase(t *testing.T) { + s := newStore(t, nil) + emitSnapshot(t, s, 100, []Projection{proj("A", 100_00)}) + + // Same-epoch delta with a fresher price overrides the base. + if err := s.HandleFrame(snapMeta("", 100, 0, 0), mustEnc(t, []ProjectionUpdate{{Projection: proj("A", 80_00)}})); err != nil { + t.Fatalf("delta: %v", err) + } + if got, _ := s.Get("A"); got.PriceIncVat != 80_00 { + t.Fatalf("A after delta = %d, want 8000", got.PriceIncVat) + } +} + +func TestProjectionStore_DeltaTombstoneShadowsBaseAndFallback(t *testing.T) { + fbHits := 0 + s := newStore(t, func(string) (Projection, bool) { fbHits++; return proj("A", 1), true }) + emitSnapshot(t, s, 100, []Projection{proj("A", 100_00)}) + + if err := s.HandleFrame(snapMeta("", 100, 0, 0), mustEnc(t, []ProjectionUpdate{{Projection: Projection{SKU: "A"}, Deleted: true}})); err != nil { + t.Fatalf("delete: %v", err) + } + if got, ok := s.Get("A"); ok { + t.Fatalf("A after tombstone = %+v, want miss (no resurrect, no fallback)", got) + } + if fbHits != 0 { + t.Fatalf("fallback hit %d times for a tombstoned SKU, want 0", fbHits) + } +} + +func TestProjectionStore_FallbackOnMiss(t *testing.T) { + s := newStore(t, func(sku string) (Projection, bool) { + if sku == "Z" { + return proj("Z", 42), true + } + return Projection{}, false + }) + emitSnapshot(t, s, 100, []Projection{proj("A", 100_00)}) + + if got, ok := s.Get("Z"); !ok || got.PriceIncVat != 42 { + t.Fatalf("Z = %+v ok=%v, want fallback 42", got, ok) + } + if _, ok := s.Get("nope"); ok { + t.Fatalf("unknown SKU should miss") + } +} + +func TestProjectionStore_StaleEpochIgnored(t *testing.T) { + s := newStore(t, nil) + emitSnapshot(t, s, 200, []Projection{proj("A", 100_00)}) + + // An older snapshot must not replace the newer base. + emitSnapshot(t, s, 100, []Projection{proj("A", 999_00), proj("OLD", 1)}) + if got, _ := s.Get("A"); got.PriceIncVat != 100_00 { + t.Fatalf("A = %d, want base epoch 200 price 10000 (stale snapshot ignored)", got.PriceIncVat) + } + if _, ok := s.Get("OLD"); ok { + t.Fatalf("OLD from a stale snapshot should not be present") + } + + // An older-epoch delta is ignored too. + _ = s.HandleFrame(snapMeta("", 100, 0, 0), mustEnc(t, []ProjectionUpdate{{Projection: proj("A", 1)}})) + if got, _ := s.Get("A"); got.PriceIncVat != 100_00 { + t.Fatalf("A after stale delta = %d, want 10000", got.PriceIncVat) + } +} + +// A future-epoch delta is DROPPED, by contract. Under FIFO per-queue delivery a +// delta never legitimately precedes its snapshot (the producer emits begin(N) +// before any delta(N)); the only way to see one is a reconnect that missed the +// whole snapshot, which a buffer couldn't reassemble anyway. Recovery is the next +// snapshot of that epoch (and the consumer's fallback in the meantime). See +// docs/catalog-projection.md. +func TestProjectionStore_FutureEpochDeltaDropped(t *testing.T) { + s := newStore(t, nil) + emitSnapshot(t, s, 100, []Projection{proj("A", 100_00)}) + + // A delta ahead of our base (gap) is dropped; only a matching snapshot applies it. + _ = s.HandleFrame(snapMeta("", 200, 0, 0), mustEnc(t, []ProjectionUpdate{{Projection: proj("A", 7)}})) + if got, _ := s.Get("A"); got.PriceIncVat != 100_00 { + t.Fatalf("A after future delta = %d, want 10000 (gap delta dropped)", got.PriceIncVat) + } +} + +func TestProjectionStore_IncompleteSnapshotDiscarded(t *testing.T) { + s := newStore(t, nil) + emitSnapshot(t, s, 100, []Projection{proj("A", 100_00)}) + + // begin says 3, but only 1 chunked → end must discard and keep the old base. + if err := s.HandleFrame(snapMeta(SnapshotBegin, 200, 0, 3), mustEnc(t, nil)); err != nil { + t.Fatal(err) + } + if err := s.HandleFrame(snapMeta(SnapshotChunk, 200, 0, 0), mustEnc(t, []ProjectionUpdate{{Projection: proj("B", 5)}})); err != nil { + t.Fatal(err) + } + if err := s.HandleFrame(snapMeta(SnapshotEnd, 200, 1, 3), mustEnc(t, nil)); err != nil { + t.Fatal(err) + } + if e := s.Epoch(); e != 100 { + t.Fatalf("epoch = %d, want 100 (incomplete snapshot discarded)", e) + } + if _, ok := s.Get("B"); ok { + t.Fatalf("B from an incomplete snapshot must not be visible") + } + if got, _ := s.Get("A"); got.PriceIncVat != 100_00 { + t.Fatalf("A = %d, want old base intact", got.PriceIncVat) + } +} + +func TestProjectionStore_ReopenFromDiskColdStart(t *testing.T) { + dir := t.TempDir() + s1, err := NewProjectionStore(dir, "consumer.bin", identity, nil) + if err != nil { + t.Fatal(err) + } + emitSnapshot(t, s1, 500, []Projection{proj("A", 100_00), proj("B", 150_00)}) + _ = s1.Close() + + // A fresh store on the same dir must serve immediately from the persisted base + // and resume same-epoch deltas (epoch read from the sidecar). + s2, err := NewProjectionStore(dir, "consumer.bin", identity, nil) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = s2.Close() }) + + if e := s2.Epoch(); e != 500 { + t.Fatalf("reopened epoch = %d, want 500 (from sidecar)", e) + } + if got, ok := s2.Get("A"); !ok || got.PriceIncVat != 100_00 { + t.Fatalf("A after reopen = %+v ok=%v, want 10000", got, ok) + } + if err := s2.HandleFrame(snapMeta("", 500, 0, 0), mustEnc(t, []ProjectionUpdate{{Projection: proj("A", 90_00)}})); err != nil { + t.Fatalf("post-reopen delta: %v", err) + } + if got, _ := s2.Get("A"); got.PriceIncVat != 90_00 { + t.Fatalf("A after post-reopen delta = %d, want 9000", got.PriceIncVat) + } +} + +func TestProjectionStore_BadPayloadErrors(t *testing.T) { + s := newStore(t, nil) + if err := s.HandleFrame(snapMeta(SnapshotChunk, 1, 0, 0), []byte("{not json")); err == nil { + t.Fatal("chunk with bad payload should error (caller may NACK)") + } + if err := s.HandleFrame(snapMeta("", 1, 0, 0), []byte("{not json")); err == nil { + t.Fatal("delta with bad payload should error") + } +} diff --git a/catalog/projection.go b/catalog/projection.go index 3daf519..2d14794 100644 --- a/catalog/projection.go +++ b/catalog/projection.go @@ -148,3 +148,27 @@ func Decode(raw []byte) (Projection, error) { err := json.Unmarshal(raw, &p) return p, err } + +// ── Bus framing contract (shared by producer + consumer) ───────────────────── +// +// catalog.projection_published carries two channels distinguished only by +// event.Event.Meta. The producer (slask-finder writeradmin) sets these keys and +// the consumer (ProjectionStore, consumer.go) reads them — one definition so the +// two never drift. See docs/catalog-projection.md. +// +// - A FULL SNAPSHOT is a framed begin → chunk* → end stream, all tagged with one +// MetaEpoch. begin/end carry MetaCount (total projections) for completeness +// validation; chunks carry the projections. +// - A DELTA has NO MetaSnapshotPhase key, and MetaEpoch names the snapshot base +// it builds on (consumers ignore a delta older than the base they hold). +const ( + MetaSnapshotPhase = "snapshot" // begin|chunk|end; absent ⇒ delta + MetaEpoch = "epoch" // int64 (Unix-ms), monotonic across rebuilds + MetaSeq = "seq" // chunk sequence within a snapshot + MetaCount = "count" // total projections (begin/end only) + MetaTenant = "tenant" // namespace tag + + SnapshotBegin = "begin" + SnapshotChunk = "chunk" + SnapshotEnd = "end" +) diff --git a/catalog/snapshot.go b/catalog/snapshot.go index 6df72b6..285047c 100644 --- a/catalog/snapshot.go +++ b/catalog/snapshot.go @@ -68,13 +68,21 @@ var ( ErrTenantTooLong = fmt.Errorf("catalog: tenant name longer than %d bytes", maxTenantLen) ) -// SnapshotWriter is single-shot: Append() all projections, then Write() once -// (atomic rename-into-place so readers never see a partial file). -type SnapshotWriter struct { - Tenant string // optional namespace tag (writes into header) - blobs []byte +// The snapshot is generic over the stored value T (see SnapshotWriter[T] / +// SnapshotReader[T] below). The binary format, sorting, mmap, and bounds-check +// logic is value-agnostic and lives in the unexported raw* layer; the generic +// wrappers only marshal T → blob (write) and blob → T (read). The producer +// stores catalog.Projection; each consumer stores its OWN projected type via a +// transform (see ProjectionStore[T] in consumer.go) — that's the per-consumer +// read-only projection from docs/catalog-projection.md. + +// rawSnapshotWriter is the value-agnostic writer: it indexes (sku → blob bytes). +// Single-shot: append() all, then write() once (atomic rename). +type rawSnapshotWriter struct { + tenant string + blobs []byte records []snapshotRecord - dirty bool + dirty bool } // snapshotRecord is the on-disk layout: SKU zero-padded to 32 bytes, then @@ -85,56 +93,78 @@ type snapshotRecord struct { BlobLength uint32 } -// NewSnapshotWriter returns an empty writer. Caller Appends every projection -// and calls Write once. -func NewSnapshotWriter() *SnapshotWriter { - return &SnapshotWriter{} -} - -// WithTenant tags this writer with a tenant namespace that lands in the -// snapshot header. Optional — empty tenant gets zero-padded; readers see an -// empty string when the writer was untagged. -func (w *SnapshotWriter) WithTenant(tenant string) *SnapshotWriter { - w.Tenant = tenant - return w -} - -// Append encodes and adds a projection. Returns ErrSKUTooLong if SKU exceeds -// the on-disk width; returns marshal errors as wrapped fmt.Errorf. -func (w *SnapshotWriter) Append(p Projection) error { - if len(p.SKU) > maxSKULen { +// append adds a pre-encoded blob under sku. ErrSKUTooLong if sku exceeds the +// on-disk width. +func (w *rawSnapshotWriter) append(sku string, blob []byte) error { + if len(sku) > maxSKULen { return ErrSKUTooLong } - jb, err := json.Marshal(p) - if err != nil { - return fmt.Errorf("catalog: marshal projection %q: %w", p.SKU, err) - } - var rec snapshotRecord - copy(rec.SKU[:], p.SKU) + copy(rec.SKU[:], sku) rec.BlobOffset = uint64(headerSize + len(w.blobs)) - rec.BlobLength = uint32(len(jb)) + rec.BlobLength = uint32(len(blob)) - w.blobs = append(w.blobs, jb...) + w.blobs = append(w.blobs, blob...) w.records = append(w.records, rec) w.dirty = true return nil } -// Len returns how many projections have been Append'd. -func (w *SnapshotWriter) Len() int { return len(w.records) } +// SnapshotWriter[T] is single-shot: Append() all values, then Write() once +// (atomic rename-into-place so readers never see a partial file). T is the +// stored value type (Projection for the producer; a consumer-specific type for +// a ProjectionStore[T]). +type SnapshotWriter[T any] struct { + raw rawSnapshotWriter +} -// Write serialises to a temp file in dir, fsync's, then renames into the +// NewSnapshotWriter returns an empty writer for value type T. Caller Appends +// every value and calls Write once. +func NewSnapshotWriter[T any]() *SnapshotWriter[T] { + return &SnapshotWriter[T]{} +} + +// WithTenant tags this writer with a tenant namespace that lands in the +// snapshot header. Optional — empty tenant gets zero-padded; readers see an +// empty string when the writer was untagged. +func (w *SnapshotWriter[T]) WithTenant(tenant string) *SnapshotWriter[T] { + w.raw.tenant = tenant + return w +} + +// Append encodes v and indexes it under sku. Returns ErrSKUTooLong if sku +// exceeds the on-disk width; returns marshal errors as wrapped fmt.Errorf. +// SKU is explicit because T is opaque — the caller knows the key. +func (w *SnapshotWriter[T]) Append(sku string, v T) error { + if len(sku) > maxSKULen { + return ErrSKUTooLong + } + jb, err := json.Marshal(v) + if err != nil { + return fmt.Errorf("catalog: marshal snapshot value %q: %w", sku, err) + } + return w.raw.append(sku, jb) +} + +// Len returns how many values have been Append'd. +func (w *SnapshotWriter[T]) Len() int { return len(w.raw.records) } + +// Write serialises to dir/name (atomic temp+rename). See rawSnapshotWriter.write. +func (w *SnapshotWriter[T]) Write(dir, name string) (finalPath string, err error) { + return w.raw.write(dir, name) +} + +// write serialises to a temp file in dir, fsync's, then renames into the // final path. Records are sorted by SKU asc before serialisation so the // reader's binary search works. -func (w *SnapshotWriter) Write(dir, name string) (finalPath string, err error) { +func (w *rawSnapshotWriter) write(dir, name string) (finalPath string, err error) { if w.dirty { sort.Slice(w.records, func(i, j int) bool { return compareSKU(w.records[i].SKU[:], w.records[j].SKU[:]) < 0 }) w.dirty = false } - if len(w.Tenant) > maxTenantLen { + if len(w.tenant) > maxTenantLen { err = ErrTenantTooLong return "", err } @@ -166,7 +196,7 @@ func (w *SnapshotWriter) Write(dir, name string) (finalPath string, err error) { binary.LittleEndian.PutUint64(hdr[16:24], uint64(len(w.records))) binary.LittleEndian.PutUint64(hdr[24:32], blobStart) binary.LittleEndian.PutUint64(hdr[32:40], recordStart) - copy(hdr[40:40+len(w.Tenant)], w.Tenant) // rest is zero-pad + copy(hdr[40:40+len(w.tenant)], w.tenant) // rest is zero-pad if _, werr := f.Write(hdr[:]); werr != nil { err = werr @@ -251,14 +281,10 @@ func compareRecordLookup(padded, want []byte) int { return compareSKU(padded, want) } -// SnapshotReader is read-only and concurrency-safe — parallel Get calls are -// OK because no fields are mutated after construction. Held open via an -// embedded file handle so the underlying mmap stays pinned. Close uses -// sync.Once so concurrent shutdown paths can't double-munmap. -// -// IMPORTANT — see package doc: SnapshotReader is NOT safe racing Close -// concurrently with Get. Use SnapshotHolder for any reload-bearing workflow. -type SnapshotReader struct { +// rawSnapshotReader is the value-agnostic reader: binary search by SKU, returns +// the raw blob bytes. Concurrency-safe for parallel getRaw; NOT safe racing +// Close with getRaw (see package doc — use SnapshotHolder for reload). +type rawSnapshotReader struct { data []byte count uint64 blobStart uint64 @@ -269,14 +295,64 @@ type SnapshotReader struct { closer func() error // munmap + close on Unix; no-op on ReadFile fallback } -// ReadSnapshotFile opens + parses a snapshot file. On Unix uses mmap; on +// SnapshotReader[T] is the typed view over a raw snapshot: Get decodes the blob +// into T. Read-only and concurrency-safe — parallel Get calls are OK. NOT safe +// racing Close concurrently with Get; use SnapshotHolder[T] for reload. +type SnapshotReader[T any] struct { + raw *rawSnapshotReader +} + +// ReadSnapshotFile opens + parses a snapshot file as a reader of T. On Unix uses +// mmap; on other platforms reads into memory (functionally identical, higher RSS). +func ReadSnapshotFile[T any](path string) (*SnapshotReader[T], error) { + rr, err := readRawSnapshotFile(path) + if err != nil { + return nil, err + } + return &SnapshotReader[T]{raw: rr}, nil +} + +// NewSnapshotReader wraps an already-loaded []byte as a reader of T (use in +// tests; production paths go through ReadSnapshotFile). +func NewSnapshotReader[T any](data []byte) (*SnapshotReader[T], error) { + rr, err := newRawSnapshotReader(data) + if err != nil { + return nil, err + } + return &SnapshotReader[T]{raw: rr}, nil +} + +// Get returns the value for sku, or (zero, false) if absent / undecodable. +func (r *SnapshotReader[T]) Get(sku string) (T, bool) { + var zero T + blob, ok := r.raw.getRaw(sku) + if !ok { + return zero, false + } + var v T + if err := json.Unmarshal(blob, &v); err != nil { + return zero, false + } + return v, true +} + +// Len returns the number of values in this snapshot. +func (r *SnapshotReader[T]) Len() int { return r.raw.Len() } + +// Tenant returns the snapshot's namespace tag ("" if untagged). +func (r *SnapshotReader[T]) Tenant() string { return r.raw.Tenant } + +// Close releases the file handle and mmap region. Idempotent. +func (r *SnapshotReader[T]) Close() error { return r.raw.Close() } + +// readRawSnapshotFile opens + parses a snapshot file. On Unix uses mmap; on // other platforms reads into memory (functionally identical, higher RSS). -func ReadSnapshotFile(path string) (*SnapshotReader, error) { +func readRawSnapshotFile(path string) (*rawSnapshotReader, error) { data, closer, err := loadSnapshotBytes(path) if err != nil { return nil, err } - r, err := newSnapshotReader(data) + r, err := newRawSnapshotReader(data) if err != nil { if closer != nil { _ = closer() @@ -287,13 +363,7 @@ func ReadSnapshotFile(path string) (*SnapshotReader, error) { return r, nil } -// NewSnapshotReader wraps an already-loaded []byte (use this in tests; for -// production paths go through ReadSnapshotFile). -func NewSnapshotReader(data []byte) (*SnapshotReader, error) { - return newSnapshotReader(data) -} - -func newSnapshotReader(data []byte) (*SnapshotReader, error) { +func newRawSnapshotReader(data []byte) (*rawSnapshotReader, error) { if len(data) < headerSize { return nil, ErrSnapshotTruncated } @@ -355,7 +425,7 @@ func newSnapshotReader(data []byte) (*SnapshotReader, error) { } tenant := string(tenantBytes[:tenantEnd]) - return &SnapshotReader{ + return &rawSnapshotReader{ data: data, count: count, blobStart: blobStart, @@ -364,15 +434,15 @@ func newSnapshotReader(data []byte) (*SnapshotReader, error) { }, nil } -// Len returns the number of Projections in this snapshot. -func (r *SnapshotReader) Len() int { return int(r.count) } +// Len returns the number of records in this snapshot. +func (r *rawSnapshotReader) Len() int { return int(r.count) } // Close releases the file handle and mmap region (if any). Idempotent — safe // to call from multiple goroutines (uses sync.Once under the hood). // -// SAFETY: this is NOT safe against concurrent Get() — see package doc. Use +// SAFETY: this is NOT safe against concurrent getRaw() — see package doc. Use // SnapshotHolder for any reload-bearing workflow. -func (r *SnapshotReader) Close() error { +func (r *rawSnapshotReader) Close() error { var err error r.closeOnce.Do(func() { if r.closer != nil { @@ -382,16 +452,16 @@ func (r *SnapshotReader) Close() error { return err } -// Get returns the projection for sku, or (zero, false) if not found. -// O(log N) — binary search on the sorted record section; on hit, reads the -// JSON blob and decodes it (one allocation). -func (r *SnapshotReader) Get(sku string) (Projection, bool) { +// getRaw returns the blob bytes for sku, or (nil, false) if not found. The +// returned slice aliases the mmap region — valid until Close. O(log N) binary +// search on the sorted record section. +func (r *rawSnapshotReader) getRaw(sku string) ([]byte, bool) { if r.count == 0 || sku == "" { - return Projection{}, false + return nil, false } skuBytes := []byte(sku) if len(skuBytes) > maxSKULen { - return Projection{}, false + return nil, false } lo, hi := uint64(0), r.count @@ -408,16 +478,11 @@ func (r *SnapshotReader) Get(sku string) (Projection, bool) { default: blobLen := binary.LittleEndian.Uint32(r.data[recOff+40 : recOff+44]) blobOff := binary.LittleEndian.Uint64(r.data[recOff+32 : recOff+40]) - // Per-record bounds check at Open guarantees this slice is in-range - rdata := r.data[blobOff : blobOff+uint64(blobLen)] - var p Projection - if err := json.Unmarshal(rdata, &p); err != nil { - return Projection{}, false - } - return p, true + // Per-record bounds check at Open guarantees this slice is in-range. + return r.data[blobOff : blobOff+uint64(blobLen)], true } } - return Projection{}, false + return nil, false } diff --git a/catalog/snapshot_helpers_test.go b/catalog/snapshot_helpers_test.go index ac7df68..ad5337a 100644 --- a/catalog/snapshot_helpers_test.go +++ b/catalog/snapshot_helpers_test.go @@ -13,6 +13,12 @@ func os2Stat(path string) (os.FileInfo, error) { return os.Stat(path) } +// appendP appends a Projection keyed by its own SKU — the producer/test pattern +// over the now-generic SnapshotWriter[Projection] (Append takes an explicit key). +func appendP(w *SnapshotWriter[Projection], p Projection) error { + return w.Append(p.SKU, p) +} + func repeat32(s string, n int) string { out := make([]byte, 0, n*len(s)) for i := 0; i < n; i++ { diff --git a/catalog/snapshot_manager.go b/catalog/snapshot_manager.go index 7872516..50104a3 100644 --- a/catalog/snapshot_manager.go +++ b/catalog/snapshot_manager.go @@ -1,12 +1,18 @@ -// SnapshotHolder provides atomic hot-reload for SnapshotReader with drain -// grace — solves the classic "Get concurrent with Close/munmap" race that -// would otherwise produce uncatchable SIGBUS / SIGSEGV under reload. +// SnapshotHolder provides atomic hot-reload for SnapshotReader and solves the +// classic "Get concurrent with Close/munmap" race that would otherwise produce +// uncatchable SIGBUS / SIGSEGV under reload. // -// Pattern source: slask-finder's `RawItemsStore` and similar mmap+reload -// stores, which delay unmap behind a configurable drain-grace timer so any -// in-flight Get on the previously-current reader completes before the -// underlying pages become unmapped. See docs/inventory-shape.md & the -// finder WARP doc on mmap reload for the prior art. +// Safety is an RWMutex (readMu): Get holds RLock across the load-and-read of a +// reader, and a reader is only Close()d (munmap) under Lock, so a reader can +// never be unmapped while a Get is reading it — regardless of how long that +// Get's goroutine was descheduled. (A drain-grace timer alone, the original +// approach, is only best-effort: a goroutine paused longer than the grace can +// still read freed pages. TestSnapshotHolder_SwapRace reproduces that crash.) +// +// The drain grace is retained as a close-BATCHING optimization: it defers and +// coalesces reader closes onto the reaper tick instead of taking the exclusive +// Lock on every swap. Pattern source: slask-finder's mmap+reload stores. +// See docs/inventory-shape.md for the prior art. // // Use: // @@ -36,30 +42,39 @@ const DefaultDrainGrace = 30 * time.Second // out from under in-flight Get()s. Set retains the previous reader behind a // drain timer; after the timer elapses the reaper goroutine closes the old // reader (and runs its munmap). -type SnapshotHolder struct { - current atomic.Pointer[SnapshotReader] +type SnapshotHolder[T any] struct { + current atomic.Pointer[SnapshotReader[T]] - drainsMu sync.Mutex - drains []*pendingClose + // readMu makes reclamation race-free: Get holds RLock across the + // Load-and-read of a reader, and a reader is only Close()d (munmap) under + // Lock — so a reader can never be unmapped while a Get is reading it, even + // if that Get's goroutine was descheduled past the drain grace. The grace + // (below) is now only close-BATCHING — defer + coalesce closes off the swap + // path — NOT the safety mechanism it was before (which was best-effort and + // could SIGSEGV under a long enough scheduler pause). + readMu sync.RWMutex + + drainsMu sync.Mutex + drains []*pendingClose[T] drainGrace time.Duration stopCh chan struct{} once sync.Once } -type pendingClose struct { - r *SnapshotReader +type pendingClose[T any] struct { + r *SnapshotReader[T] expireAt time.Time } -// NewSnapshotHolder returns a holder with the given drain grace (the time +// NewSnapshotHolder returns a holder of T with the given drain grace (the time // the previous reader lives behind Set before being closed). Zero or // negative grace falls back to DefaultDrainGrace. -func NewSnapshotHolder(grace time.Duration) *SnapshotHolder { +func NewSnapshotHolder[T any](grace time.Duration) *SnapshotHolder[T] { if grace <= 0 { grace = DefaultDrainGrace } - h := &SnapshotHolder{ + h := &SnapshotHolder[T]{ drainGrace: grace, stopCh: make(chan struct{}), } @@ -69,8 +84,8 @@ func NewSnapshotHolder(grace time.Duration) *SnapshotHolder { // Open is shorthand for Set(ReadSnapshotFile(path)). Errors leave the // holder's current reader unchanged. -func (h *SnapshotHolder) Open(path string) error { - r, err := ReadSnapshotFile(path) +func (h *SnapshotHolder[T]) Open(path string) error { + r, err := ReadSnapshotFile[T](path) if err != nil { return err } @@ -81,8 +96,8 @@ func (h *SnapshotHolder) Open(path string) error { // Reload atomically swaps the current reader for one freshly opened from // path. The old reader (if any) is queued for close after the drain grace // elapses; in-flight Get(s) on the old reader continue to work. -func (h *SnapshotHolder) Reload(path string) error { - r, err := ReadSnapshotFile(path) +func (h *SnapshotHolder[T]) Reload(path string) error { + r, err := ReadSnapshotFile[T](path) if err != nil { return err } @@ -92,7 +107,7 @@ func (h *SnapshotHolder) Reload(path string) error { // Set replaces the current reader with r. The previous reader is queued // for close after the drain grace elapses. r may be nil to clear. -func (h *SnapshotHolder) Set(r *SnapshotReader) { +func (h *SnapshotHolder[T]) Set(r *SnapshotReader[T]) { if r == nil { h.current.Store(nil) return @@ -102,7 +117,7 @@ func (h *SnapshotHolder) Set(r *SnapshotReader) { return } h.drainsMu.Lock() - h.drains = append(h.drains, &pendingClose{ + h.drains = append(h.drains, &pendingClose[T]{ r: old, expireAt: time.Now().Add(h.drainGrace), }) @@ -110,17 +125,23 @@ func (h *SnapshotHolder) Set(r *SnapshotReader) { } // Get is the concurrent-safe shim over the current reader's Get. Returns -// (zero, false) when the holder is empty. -func (h *SnapshotHolder) Get(sku string) (Projection, bool) { +// (zero, false) when the holder is empty. The RLock is held across the Load AND +// the read so a concurrent reaper can't munmap this reader mid-read. +func (h *SnapshotHolder[T]) Get(sku string) (T, bool) { + h.readMu.RLock() + defer h.readMu.RUnlock() r := h.current.Load() if r == nil { - return Projection{}, false + var zero T + return zero, false } return r.Get(sku) } // Len returns the count of the current reader (0 if empty). -func (h *SnapshotHolder) Len() int { +func (h *SnapshotHolder[T]) Len() int { + h.readMu.RLock() + defer h.readMu.RUnlock() r := h.current.Load() if r == nil { return 0 @@ -131,22 +152,29 @@ func (h *SnapshotHolder) Len() int { // Close stops the reaper goroutine and closes the current reader + any // pending drains without waiting for the grace period. Safe to call // multiple times. -func (h *SnapshotHolder) Close() error { +func (h *SnapshotHolder[T]) Close() error { var firstErr error h.once.Do(func() { close(h.stopCh) + // Close under Lock so no in-flight Get is reading these readers' mmap + // when we munmap them. + h.readMu.Lock() r := h.current.Swap(nil) if r != nil { if err := r.Close(); err != nil { firstErr = err } } + h.readMu.Unlock() h.drainsMu.Lock() drained := h.drains h.drains = nil h.drainsMu.Unlock() for _, d := range drained { - if err := d.r.Close(); err != nil && firstErr == nil { + h.readMu.Lock() + err := d.r.Close() + h.readMu.Unlock() + if err != nil && firstErr == nil { firstErr = err } } @@ -156,7 +184,7 @@ func (h *SnapshotHolder) Close() error { // DrainCount returns the number of readers currently in the drain queue. // Useful for tests and operational metrics. -func (h *SnapshotHolder) DrainCount() int { +func (h *SnapshotHolder[T]) DrainCount() int { h.drainsMu.Lock() defer h.drainsMu.Unlock() return len(h.drains) @@ -164,10 +192,10 @@ func (h *SnapshotHolder) DrainCount() int { // reapNow scans the drain queue and Closes any readers whose grace has // elapsed. Called by the sweeper goroutine on a tick. -func (h *SnapshotHolder) reapNow() { +func (h *SnapshotHolder[T]) reapNow() { now := time.Now() - var alive []*pendingClose - var reaped []*pendingClose + var alive []*pendingClose[T] + var reaped []*pendingClose[T] h.drainsMu.Lock() for _, d := range h.drains { if !d.expireAt.After(now) { @@ -179,11 +207,16 @@ func (h *SnapshotHolder) reapNow() { h.drains = alive h.drainsMu.Unlock() for _, d := range reaped { + // Lock briefly so the munmap can't race an in-flight Get on this reader. + // Gets that loaded this reader hold RLock and finish before the Lock is + // granted; Gets starting after see the swapped-in reader, never this one. + h.readMu.Lock() _ = d.r.Close() + h.readMu.Unlock() } } -func (h *SnapshotHolder) sweepLoop() { +func (h *SnapshotHolder[T]) sweepLoop() { interval := h.drainGrace / 4 if interval < 10*time.Millisecond { interval = 10 * time.Millisecond diff --git a/catalog/snapshot_manager_test.go b/catalog/snapshot_manager_test.go index 23eaf5b..4048848 100644 --- a/catalog/snapshot_manager_test.go +++ b/catalog/snapshot_manager_test.go @@ -1,11 +1,124 @@ package catalog import ( + "fmt" "path/filepath" + "sync" + "sync/atomic" "testing" "time" + + "git.k6n.net/mats/platform/money" ) +// TestSnapshotHolder_SwapRace storms the holder: 100 readers Get in tight loops +// while a rotator Reloads a fresh snapshot every 50ms. It asserts the holder's +// safety + liveness invariants under concurrency (run with -race): +// +// - No panic / SIGBUS — completion is the proof the drain grace never unmaps a +// reader while an in-flight Get is reading it. +// - No torn reads — every hit decodes to a valid Projection (SKU matches, +// price > 0). A read of a munmapped page would crash or fail to decode. +// - No gap — since every snapshot version contains every SKU, a Get must never +// miss; misses would mean the holder briefly exposed an empty/nil reader +// mid-swap. (Exact "stale beyond drain grace" timing is intentionally not +// asserted — it's timing-flaky, and the no-torn-read + no-crash checks +// already cover the drain-safety property it stands in for.) +func TestSnapshotHolder_SwapRace(t *testing.T) { + dir := t.TempDir() + + const versions = 4 + const skuCount = 200 + paths := make([]string, versions) + for v := 0; v < versions; v++ { + w := NewSnapshotWriter[Projection]() + for i := 0; i < skuCount; i++ { + sku := fmt.Sprintf("SKU-%03d", i) + // price encodes the version (always > 0) so a torn read is detectable. + if err := w.Append(sku, Projection{ID: "sku:" + sku, SKU: sku, PriceIncVat: money.Cents(int64((v + 1) * 100))}); err != nil { + t.Fatal(err) + } + } + p, err := w.Write(dir, fmt.Sprintf("snap-%d.bin", v)) + if err != nil { + t.Fatal(err) + } + paths[v] = p + } + + h := NewSnapshotHolder[Projection](100 * time.Millisecond) + if err := h.Open(paths[0]); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = h.Close() }) + + stop := make(chan struct{}) + var reads, misses int64 + var torn int32 + var wg sync.WaitGroup + + for r := 0; r < 100; r++ { + wg.Add(1) + go func(seed int) { + defer wg.Done() + i := seed + for { + select { + case <-stop: + return + default: + } + sku := fmt.Sprintf("SKU-%03d", i%skuCount) + if p, ok := h.Get(sku); ok { + if p.SKU != sku || p.PriceIncVat <= 0 { + atomic.StoreInt32(&torn, 1) + return + } + atomic.AddInt64(&reads, 1) + } else { + atomic.AddInt64(&misses, 1) + } + i++ + } + }(r) + } + + wg.Add(1) + go func() { + defer wg.Done() + tick := time.NewTicker(50 * time.Millisecond) + defer tick.Stop() + v := 0 + for { + select { + case <-stop: + return + case <-tick.C: + v = (v + 1) % versions + if err := h.Reload(paths[v]); err != nil { + atomic.StoreInt32(&torn, 1) + return + } + } + } + }() + + time.Sleep(1500 * time.Millisecond) + close(stop) + wg.Wait() + + if atomic.LoadInt32(&torn) != 0 { + t.Fatal("torn/garbage read or reload error during swap storm") + } + if m := atomic.LoadInt64(&misses); m != 0 { + t.Fatalf("got %d misses — holder exposed an empty reader mid-swap (every version has every SKU)", m) + } + if atomic.LoadInt64(&reads) == 0 { + t.Fatal("no successful reads — test did not exercise the holder") + } + t.Logf("SwapRace OK: %d reads across %d rotations, 0 torn, 0 miss", atomic.LoadInt64(&reads), versions) +} + // fakeCloser counts how many times it's been called. Used to assert the // drain-grace semantics of SnapshotHolder. type fakeCloser struct{ hits int } @@ -15,14 +128,14 @@ func (f *fakeCloser) Close() error { f.hits++; return nil } // withFakeCloser swaps SnapshotReader's closer for our counter. Returns the // reader and the counter; tests then call holder.Set(reloaded) and assert // that close is invoked after the drain grace. -func withFakeCloser(t *testing.T, path string, fc *fakeCloser) *SnapshotReader { +func withFakeCloser(t *testing.T, path string, fc *fakeCloser) *SnapshotReader[Projection] { t.Helper() - r, err := ReadSnapshotFile(path) + r, err := ReadSnapshotFile[Projection](path) if err != nil { t.Fatal(err) } // replace real closer with our counter; the test asserts only via fc - r.closer = fc.Close + r.raw.closer = fc.Close return r } @@ -35,11 +148,11 @@ func TestSnapshotHolderReloadDrainsOld(t *testing.T) { dir := t.TempDir() // 1) Build a valid first snapshot (skus A-1 / A-2). - w := NewSnapshotWriter() - if err := w.Append(makeProjection("A-1", 100)); err != nil { + w := NewSnapshotWriter[Projection]() + if err := appendP(w, makeProjection("A-1", 100)); err != nil { t.Fatal(err) } - if err := w.Append(makeProjection("A-2", 200)); err != nil { + if err := appendP(w, makeProjection("A-2", 200)); err != nil { t.Fatal(err) } if _, err := w.Write(dir, "first.bin"); err != nil { @@ -50,8 +163,8 @@ func TestSnapshotHolderReloadDrainsOld(t *testing.T) { defer first.Close() // 2) Build a second snapshot (skus B-1). - w2 := NewSnapshotWriter() - if err := w2.Append(makeProjection("B-1", 300)); err != nil { + w2 := NewSnapshotWriter[Projection]() + if err := appendP(w2, makeProjection("B-1", 300)); err != nil { t.Fatal(err) } if _, err := w2.Write(dir, "second.bin"); err != nil { @@ -61,7 +174,7 @@ func TestSnapshotHolderReloadDrainsOld(t *testing.T) { second := withFakeCloser(t, filepath.Join(dir, "second.bin"), secondCloser) defer second.Close() - h := NewSnapshotHolder(50 * time.Millisecond) + h := NewSnapshotHolder[Projection](50 * time.Millisecond) defer h.Close() h.Set(first) if h.Len() != 2 { @@ -96,7 +209,7 @@ func TestSnapshotHolderReloadDrainsOld(t *testing.T) { // TestSnapshotHolderEmptyGet — holder with no current serves empty. func TestSnapshotHolderEmptyGet(t *testing.T) { - h := NewSnapshotHolder(100 * time.Millisecond) + h := NewSnapshotHolder[Projection](100 * time.Millisecond) defer h.Close() if _, ok := h.Get("anything"); ok { t.Error("empty holder should return false") @@ -112,8 +225,8 @@ func TestSnapshotHolderEmptyGet(t *testing.T) { func TestSnapshotHolderCloseImmediatelyReaps(t *testing.T) { dir := t.TempDir() - w := NewSnapshotWriter() - if err := w.Append(makeProjection("X-1", 1)); err != nil { + w := NewSnapshotWriter[Projection]() + if err := appendP(w, makeProjection("X-1", 1)); err != nil { t.Fatal(err) } if _, err := w.Write(dir, "x.bin"); err != nil { @@ -122,8 +235,8 @@ func TestSnapshotHolderCloseImmediatelyReaps(t *testing.T) { fc1 := &fakeCloser{} r1 := withFakeCloser(t, filepath.Join(dir, "x.bin"), fc1) - w2 := NewSnapshotWriter() - if err := w2.Append(makeProjection("Y-1", 2)); err != nil { + w2 := NewSnapshotWriter[Projection]() + if err := appendP(w2, makeProjection("Y-1", 2)); err != nil { t.Fatal(err) } if _, err := w2.Write(dir, "y.bin"); err != nil { @@ -132,7 +245,7 @@ func TestSnapshotHolderCloseImmediatelyReaps(t *testing.T) { fc2 := &fakeCloser{} r2 := withFakeCloser(t, filepath.Join(dir, "y.bin"), fc2) - h := NewSnapshotHolder(10 * time.Second) // long grace so reap wouldn't happen on its own + h := NewSnapshotHolder[Projection](10 * time.Second) // long grace so reap wouldn't happen on its own h.Set(r1) h.Set(r2) @@ -152,15 +265,15 @@ func TestSnapshotHolderCloseImmediatelyReaps(t *testing.T) { // until grace elapses. func TestSnapshotHolderIdentityReload(t *testing.T) { dir := t.TempDir() - w := NewSnapshotWriter() - if err := w.Append(makeProjection("ID-1", 1)); err != nil { + w := NewSnapshotWriter[Projection]() + if err := appendP(w, makeProjection("ID-1", 1)); err != nil { t.Fatal(err) } if _, err := w.Write(dir, "id.bin"); err != nil { t.Fatal(err) } - h := NewSnapshotHolder(10 * time.Second) + h := NewSnapshotHolder[Projection](10 * time.Second) defer h.Close() if err := h.Reload(filepath.Join(dir, "id.bin")); err != nil { t.Fatal(err) diff --git a/catalog/snapshot_test.go b/catalog/snapshot_test.go index 8dbe066..78ea39b 100644 --- a/catalog/snapshot_test.go +++ b/catalog/snapshot_test.go @@ -20,12 +20,12 @@ func TestSnapshotRoundTrip(t *testing.T) { path := filepath.Join(dir, "snap.bin") const n = 1000 - w := NewSnapshotWriter().WithTenant("default") + w := NewSnapshotWriter[Projection]().WithTenant("default") // intentionally shuffled input — sort is what reorders at Write. perm := rand.Perm(n) for _, i := range perm { sku := fmt.Sprintf("SKU-%05d", i) - if err := w.Append(makeProjection(sku, int64(100+(i%97)*100))); err != nil { + if err := appendP(w, makeProjection(sku, int64(100+(i%97)*100))); err != nil { t.Fatalf("Append %s: %v", sku, err) } } @@ -40,7 +40,7 @@ func TestSnapshotRoundTrip(t *testing.T) { t.Errorf("Write returned %q expected %q", final, path) } - r, err := ReadSnapshotFile(path) + r, err := ReadSnapshotFile[Projection](path) if err != nil { t.Fatalf("ReadSnapshotFile: %v", err) } @@ -49,8 +49,8 @@ func TestSnapshotRoundTrip(t *testing.T) { if r.Len() != n { t.Errorf("Len: got %d want %d", r.Len(), n) } - if r.Tenant != "default" { - t.Errorf("Tenant: got %q want %q", r.Tenant, "default") + if r.Tenant() != "default" { + t.Errorf("Tenant: got %q want %q", r.Tenant(), "default") } // every SKU present and correct @@ -79,19 +79,19 @@ func TestSnapshotRoundTrip(t *testing.T) { // — this test pins that invariant. func TestSnapshotCompareSKUInvariant(t *testing.T) { dir := t.TempDir() - w := NewSnapshotWriter() + w := NewSnapshotWriter[Projection]() // deliberately inverted order after Append (so sort has work to do) perm := rand.Perm(500) for _, i := range perm { sku := fmt.Sprintf("XX-%05d", i) - if err := w.Append(makeProjection(sku, int64(i))); err != nil { + if err := appendP(w, makeProjection(sku, int64(i))); err != nil { t.Fatal(err) } } if _, err := w.Write(dir, "snap.bin"); err != nil { t.Fatal(err) } - r, err := ReadSnapshotFile(filepath.Join(dir, "snap.bin")) + r, err := ReadSnapshotFile[Projection](filepath.Join(dir, "snap.bin")) if err != nil { t.Fatal(err) } @@ -101,8 +101,8 @@ func TestSnapshotCompareSKUInvariant(t *testing.T) { // form, and assert non-decreasing order under compareSKU. prevSKU := "" for i := 0; i < r.Len(); i++ { - recOff := int(r.recordStart) + i*recordSize - raw := r.data[recOff : recOff+32] + recOff := int(r.raw.recordStart) + i*recordSize + raw := r.raw.data[recOff : recOff+32] // trim trailing zeros end := len(raw) for end > 0 && raw[end-1] == 0 { @@ -131,45 +131,45 @@ func runesToPadded32(s string) []byte { // with no on-disk evidence. func TestSnapshotTenantRoundTrip(t *testing.T) { dir := t.TempDir() - w := NewSnapshotWriter().WithTenant("bygglagret") - if err := w.Append(makeProjection("ABC-1", 100)); err != nil { + w := NewSnapshotWriter[Projection]().WithTenant("bygglagret") + if err := appendP(w, makeProjection("ABC-1", 100)); err != nil { t.Fatal(err) } if _, err := w.Write(dir, "snap.bin"); err != nil { t.Fatal(err) } - r, err := ReadSnapshotFile(filepath.Join(dir, "snap.bin")) + r, err := ReadSnapshotFile[Projection](filepath.Join(dir, "snap.bin")) if err != nil { t.Fatal(err) } defer r.Close() - if r.Tenant != "bygglagret" { - t.Errorf("Tenant: got %q want %q", r.Tenant, "bygglagret") + if r.Tenant() != "bygglagret" { + t.Errorf("Tenant: got %q want %q", r.Tenant(), "bygglagret") } // Empty tenant writes zero-padded. - w2 := NewSnapshotWriter() - if err := w2.Append(makeProjection("ABC-2", 200)); err != nil { + w2 := NewSnapshotWriter[Projection]() + if err := appendP(w2, makeProjection("ABC-2", 200)); err != nil { t.Fatal(err) } if _, err := w2.Write(dir, "snap2.bin"); err != nil { t.Fatal(err) } - r2, err := ReadSnapshotFile(filepath.Join(dir, "snap2.bin")) + r2, err := ReadSnapshotFile[Projection](filepath.Join(dir, "snap2.bin")) if err != nil { t.Fatal(err) } defer r2.Close() - if r2.Tenant != "" { - t.Errorf("Tenant (no tag): got %q want empty", r2.Tenant) + if r2.Tenant() != "" { + t.Errorf("Tenant (no tag): got %q want empty", r2.Tenant()) } } // TestSnapshotTenantTooLong — over-long tenant name rejected at Write. func TestSnapshotTenantTooLong(t *testing.T) { dir := t.TempDir() - w := NewSnapshotWriter().WithTenant(string(make([]byte, 33))) - if err := w.Append(makeProjection("ABC", 1)); err != nil { + w := NewSnapshotWriter[Projection]().WithTenant(string(make([]byte, 33))) + if err := appendP(w, makeProjection("ABC", 1)); err != nil { t.Fatal(err) } if _, err := w.Write(dir, "snap.bin"); err == nil { @@ -196,7 +196,7 @@ func TestSnapshotCountOverflowRejected(t *testing.T) { if err := os2WriteFile(path, bs, 0o644); err != nil { t.Fatal(err) } - if _, err := ReadSnapshotFile(path); !errors.Is(err, ErrSnapshotOverflow) { + if _, err := ReadSnapshotFile[Projection](path); !errors.Is(err, ErrSnapshotOverflow) { t.Errorf("overflow count rejected with %v, want ErrSnapshotOverflow", err) } } @@ -205,10 +205,10 @@ func TestSnapshotCountOverflowRejected(t *testing.T) { // leaves a half-written snap.bin visible to readers. func TestSnapshotAtomicSwap(t *testing.T) { dir := t.TempDir() - w := NewSnapshotWriter() - mustAppend := func(w *SnapshotWriter, sku string, price int64) { + w := NewSnapshotWriter[Projection]() + mustAppend := func(w *SnapshotWriter[Projection], sku string, price int64) { t.Helper() - if err := w.Append(makeProjection(sku, price)); err != nil { + if err := appendP(w, makeProjection(sku, price)); err != nil { t.Fatalf("Append %s: %v", sku, err) } } @@ -223,13 +223,13 @@ func TestSnapshotAtomicSwap(t *testing.T) { } // replace - w2 := NewSnapshotWriter() + w2 := NewSnapshotWriter[Projection]() mustAppend(w2, "ZZZ", 200) if _, err := w2.Write(dir, "snap.bin"); err != nil { t.Fatal(err) } - r, err := ReadSnapshotFile(final) + r, err := ReadSnapshotFile[Projection](final) if err != nil { t.Fatal(err) } @@ -251,7 +251,7 @@ func TestSnapshotRejectBadInput(t *testing.T) { if err := os2WriteFile(path, []byte{}, 0o644); err != nil { t.Fatal(err) } - if _, err := ReadSnapshotFile(path); !errors.Is(err, ErrSnapshotTruncated) { + if _, err := ReadSnapshotFile[Projection](path); !errors.Is(err, ErrSnapshotTruncated) { t.Errorf("empty file rejected with %v, want ErrSnapshotTruncated", err) } @@ -265,7 +265,7 @@ func TestSnapshotRejectBadInput(t *testing.T) { if err := os2WriteFile(path, wrongMagic, 0o644); err != nil { t.Fatal(err) } - if _, err := ReadSnapshotFile(path); !errors.Is(err, ErrSnapshotMagic) { + if _, err := ReadSnapshotFile[Projection](path); !errors.Is(err, ErrSnapshotMagic) { t.Errorf("wrong magic rejected with %v, want ErrSnapshotMagic", err) } @@ -274,7 +274,7 @@ func TestSnapshotRejectBadInput(t *testing.T) { if err := os2WriteFile(path, buildValidHeader(5), 0o644); err != nil { t.Fatal(err) } - if _, err := ReadSnapshotFile(path); !errors.Is(err, ErrSnapshotTruncated) { + if _, err := ReadSnapshotFile[Projection](path); !errors.Is(err, ErrSnapshotTruncated) { t.Errorf("truncated snapshot rejected with %v, want ErrSnapshotTruncated", err) } @@ -284,7 +284,7 @@ func TestSnapshotRejectBadInput(t *testing.T) { if err := os2WriteFile(path, tampered, 0o644); err != nil { t.Fatal(err) } - if _, err := ReadSnapshotFile(path); !errors.Is(err, ErrSnapshotBadBlob) { + if _, err := ReadSnapshotFile[Projection](path); !errors.Is(err, ErrSnapshotBadBlob) { t.Errorf("tampered-blob rejected with %v, want ErrSnapshotBadBlob", err) } } @@ -292,11 +292,11 @@ func TestSnapshotRejectBadInput(t *testing.T) { // TestSnapshotEmpty — empty (count=0) snapshot is valid; Get returns false. func TestSnapshotEmpty(t *testing.T) { dir := t.TempDir() - w := NewSnapshotWriter() + w := NewSnapshotWriter[Projection]() if _, err := w.Write(dir, "empty.bin"); err != nil { t.Fatal(err) } - r, err := ReadSnapshotFile(filepath.Join(dir, "empty.bin")) + r, err := ReadSnapshotFile[Projection](filepath.Join(dir, "empty.bin")) if err != nil { t.Fatal(err) } @@ -316,14 +316,14 @@ func TestSnapshotSKU32(t *testing.T) { if len(sku) != 32 { t.Fatalf("test bug: sku length %d", len(sku)) } - w := NewSnapshotWriter() - if err := w.Append(makeProjection(sku, 1)); err != nil { + w := NewSnapshotWriter[Projection]() + if err := appendP(w, makeProjection(sku, 1)); err != nil { t.Fatalf("Append 32-char SKU: %v", err) } if _, err := w.Write(dir, "snap.bin"); err != nil { t.Fatal(err) } - r, err := ReadSnapshotFile(filepath.Join(dir, "snap.bin")) + r, err := ReadSnapshotFile[Projection](filepath.Join(dir, "snap.bin")) if err != nil { t.Fatal(err) } @@ -335,9 +335,9 @@ func TestSnapshotSKU32(t *testing.T) { // TestSnapshotSKU33Rejected — one char over should error at Append. func TestSnapshotSKU33Rejected(t *testing.T) { - w := NewSnapshotWriter() + w := NewSnapshotWriter[Projection]() sku := repeat32("X", 33) - err := w.Append(makeProjection(sku, 1)) + err := appendP(w, makeProjection(sku, 1)) if !errors.Is(err, ErrSKUTooLong) { t.Errorf("33-char SKU Append: got %v want ErrSKUTooLong", err) } @@ -351,18 +351,18 @@ func TestSnapshotGetSpeed(t *testing.T) { } dir := t.TempDir() const n = 5000 - w := NewSnapshotWriter() + w := NewSnapshotWriter[Projection]() skus := make([]string, n) for i := 0; i < n; i++ { skus[i] = fmt.Sprintf("Z%06d", i) - if err := w.Append(makeProjection(skus[i], int64(i))); err != nil { + if err := appendP(w, makeProjection(skus[i], int64(i))); err != nil { t.Fatalf("Append %s: %v", skus[i], err) } } if _, err := w.Write(dir, "snap.bin"); err != nil { t.Fatal(err) } - r, err := ReadSnapshotFile(filepath.Join(dir, "snap.bin")) + r, err := ReadSnapshotFile[Projection](filepath.Join(dir, "snap.bin")) if err != nil { t.Fatal(err) } @@ -429,7 +429,7 @@ func TestSnapshotRejectBadVersion(t *testing.T) { if err := os2WriteFile(path, bs, 0o644); err != nil { t.Fatal(err) } - if _, err := ReadSnapshotFile(path); !errors.Is(err, ErrSnapshotVersion) { + if _, err := ReadSnapshotFile[Projection](path); !errors.Is(err, ErrSnapshotVersion) { t.Errorf("v2-not-supported header rejected with %v, want ErrSnapshotVersion", err) } }