more platform

projection/doc.go

@@ -0,0 +1,27 @@
+// Package projection is the domain-neutral substrate for the "snapshot stream +
+// local index" seam: one authoritative producer publishes a framed
+// begin → chunk* → end snapshot (plus per-batch deltas) on the bus, and N
+// receivers each assemble their OWN cold-start-ready, mmap-backed .bin index
+// from it. No shared volume; each consumer is independently eventually
+// consistent.
+//
+// It was extracted from platform/catalog (which now re-exports these types for
+// its Projection wire shape) so other domains can ride the same rails without
+// depending on catalog. The pieces:
+//
+//   - SnapshotWriter[T] / SnapshotReader[T] (snapshot.go) — the value-agnostic
+//     key→blob .bin format: sorted records, binary-search Get, full-walk Scan,
+//     mmap on Unix. Generic over the stored type T.
+//   - SnapshotHolder[T] (manager.go) — race-free atomic hot-reload of a reader
+//     under concurrent Get/Scan, with a drain-grace close-batching reaper.
+//   - Store[W, T] (store.go) — the consumer state machine: snapshot assembly,
+//     epoch gating, same-epoch delta overlay (with delete tombstones), and the
+//     completeness guard. Generic over the WIRE type W (decoded by a
+//     domain-supplied Decoder) and the STORED type T (projected by a Transform).
+//   - The bus framing contract (framing.go) — Meta* keys, snapshot phases, and
+//     the decoded Update[W] shape.
+//
+// The framing is fixed and shared; the on-the-wire payload schema is the
+// domain's own concern, bridged by Decoder[W]. See
+// docs/plan-projectionstore-generalization.md for why this boundary exists.
+package projection

projection/framing.go

@@ -0,0 +1,57 @@
+package projection
+
+// Bus framing contract (shared by producer + consumer).
+//
+// A projection stream carries two channels distinguished only by the event's
+// Meta map. The producer sets these keys and the consumer reads them — one
+// definition so the two never drift.
+//
+//   - A FULL SNAPSHOT is a framed begin → chunk* → end stream, all tagged with one
+//     MetaEpoch. begin/end carry MetaCount (total entries) for completeness
+//     validation; chunks carry the entries.
+//   - 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).
+//
+// The framing is domain-neutral. What rides inside each chunk/delta payload is
+// a domain-specific wire schema, decoded by the Decoder the consumer supplies
+// (see Store / Decoder). catalog re-exports these constants for wire-compat.
+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 entries (begin/end only)
+	MetaTenant        = "tenant"   // namespace tag
+
+	SnapshotBegin = "begin"
+	SnapshotChunk = "chunk"
+	SnapshotEnd   = "end"
+)
+
+// Update is the decoded form of one entry in a snapshot chunk or a delta. Key
+// is the index key; Value is the domain wire value; Deleted marks a removal
+// (the consumer tombstones the key in its overlay, or skips it in a snapshot).
+// This is the in-memory shape the framing layer works with — the on-the-wire
+// JSON schema is the domain's own concern, bridged by a Decoder.
+type Update[W any] struct {
+	Key     string
+	Value   W
+	Deleted bool
+}
+
+// Decoder turns one chunk/delta payload (the event's Payload bytes) into
+// decoded Updates. Each domain defines its own wire schema and supplies the
+// decoder; the Store owns only the framing/assembly around it. A decode error
+// is surfaced from HandleFrame so the caller may NACK.
+type Decoder[W any] func(payload []byte) ([]Update[W], error)
+
+// Transform projects a wire value W down to the consumer's stored type T. It is
+// the one place a consumer declares which facts it actually needs (and any
+// derivation), so its local .bin holds only that. Every value entering the
+// store — snapshot, delta, and fallback — passes through it exactly once.
+type Transform[W, T any] func(W) T
+
+// Fallback resolves a key the local store doesn't have yet (cold start before
+// the first snapshot, or a key added since the last base). It returns a raw
+// wire value W — the store applies the same Transform to it. Return (_, false)
+// for a genuine miss. May be nil.
+type Fallback[W any] func(key string) (W, bool)

projection/helpers_test.go

@@ -0,0 +1,54 @@
+// Local helpers for the snapshot / holder tests — production code does not see
+// them. tProj is a stand-in stored value; the projection package is
+// value-agnostic, so the tests use a small local struct instead of depending on
+// catalog.Projection (the type these tests were extracted alongside).
+package projection
+
+import "os"
+
+// tProj mirrors the fields the ported catalog snapshot tests exercise. Prices
+// are plain int64 here (no money dependency in the generic package).
+type tProj struct {
+	ID          string `json:"id"`
+	SKU         string `json:"sku"`
+	Title       string `json:"title,omitempty"`
+	PriceIncVat int64  `json:"price_inc_vat"`
+	Currency    string `json:"currency,omitempty"`
+	TaxClass    string `json:"tax_class,omitempty"`
+	UpdatedAt   int64  `json:"updated_at,omitempty"`
+}
+
+func os2WriteFile(path string, data []byte, perm os.FileMode) error {
+	return os.WriteFile(path, data, perm)
+}
+
+func os2Stat(path string) (os.FileInfo, error) {
+	return os.Stat(path)
+}
+
+// appendP appends a tProj keyed by its own SKU — the producer/test pattern over
+// the generic SnapshotWriter (Append takes an explicit key).
+func appendP(w *SnapshotWriter[tProj], p tProj) error {
+	return w.Append(p.SKU, p)
+}
+
+// makeProjection builds a tProj fixture.
+func makeProjection(sku string, price int64) tProj {
+	return tProj{
+		ID:          "sku:" + sku,
+		SKU:         sku,
+		Title:       "Item " + sku,
+		PriceIncVat: price,
+		Currency:    "SEK",
+		TaxClass:    "standard",
+		UpdatedAt:   1,
+	}
+}
+
+func repeat32(s string, n int) string {
+	out := make([]byte, 0, n*len(s))
+	for i := 0; i < n; i++ {
+		out = append(out, s...)
+	}
+	return string(out)
+}

projection/manager.go

@@ -0,0 +1,242 @@
+// 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.
+//
+// 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.
+//
+// Use:
+//
+//	holder := projection.NewSnapshotHolder[T](30 * time.Second) // drain grace
+//	if err := holder.Open("/var/lib/catalog/bygglagret.bin"); err != nil { ... }
+//	// later, when a fresh snapshot lands:
+//	if err := holder.Reload("/var/lib/catalog/bygglagret.bin"); err != nil { ... }
+//	// pulls the value; always hits the current reader:
+//	p, ok := holder.Get(key)
+//
+// holder.Get / Scan / Len are safe for parallel use. Open and Reload are also
+// safe for parallel use. Close stops the reaper goroutine and shuts the
+// current reader + drains down.
+package projection
+
+import (
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+// DefaultDrainGrace is the default drain period for SnapshotHolder. Finder's
+// analogous reload path uses ~30s; we mirror that value.
+const DefaultDrainGrace = 30 * time.Second
+
+// SnapshotHolder swaps a SnapshotReader under load without unmapping pages
+// 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[T any] struct {
+	current atomic.Pointer[SnapshotReader[T]]
+
+	// 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.
+	readMu sync.RWMutex
+
+	drainsMu   sync.Mutex
+	drains     []*pendingClose[T]
+	drainGrace time.Duration
+
+	stopCh chan struct{}
+	once   sync.Once
+}
+
+type pendingClose[T any] struct {
+	r        *SnapshotReader[T]
+	expireAt time.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[T any](grace time.Duration) *SnapshotHolder[T] {
+	if grace <= 0 {
+		grace = DefaultDrainGrace
+	}
+	h := &SnapshotHolder[T]{
+		drainGrace: grace,
+		stopCh:     make(chan struct{}),
+	}
+	go h.sweepLoop()
+	return h
+}
+
+// Open is shorthand for Set(ReadSnapshotFile(path)). Errors leave the
+// holder's current reader unchanged.
+func (h *SnapshotHolder[T]) Open(path string) error {
+	r, err := ReadSnapshotFile[T](path)
+	if err != nil {
+		return err
+	}
+	h.Set(r)
+	return nil
+}
+
+// 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[T]) Reload(path string) error {
+	r, err := ReadSnapshotFile[T](path)
+	if err != nil {
+		return err
+	}
+	h.Set(r)
+	return nil
+}
+
+// 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[T]) Set(r *SnapshotReader[T]) {
+	if r == nil {
+		h.current.Store(nil)
+		return
+	}
+	old := h.current.Swap(r)
+	if old == nil {
+		return
+	}
+	h.drainsMu.Lock()
+	h.drains = append(h.drains, &pendingClose[T]{
+		r:        old,
+		expireAt: time.Now().Add(h.drainGrace),
+	})
+	h.drainsMu.Unlock()
+}
+
+// Get is the concurrent-safe shim over the current reader's Get. Returns
+// (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 {
+		var zero T
+		return zero, false
+	}
+	return r.Get(sku)
+}
+
+// Scan walks every (key, value) of the current reader in key-sorted order.
+// Return false from fn to stop early. The RLock is held for the whole scan so
+// the reader can't be munmapped mid-walk — keep fn cheap (collect then sort
+// outside the callback for ranked output). No-op when the holder is empty.
+func (h *SnapshotHolder[T]) Scan(fn func(key string, v T) bool) {
+	h.readMu.RLock()
+	defer h.readMu.RUnlock()
+	r := h.current.Load()
+	if r == nil {
+		return
+	}
+	r.Scan(fn)
+}
+
+// Len returns the count of the current reader (0 if empty).
+func (h *SnapshotHolder[T]) Len() int {
+	h.readMu.RLock()
+	defer h.readMu.RUnlock()
+	r := h.current.Load()
+	if r == nil {
+		return 0
+	}
+	return r.Len()
+}
+
+// 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[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 {
+			h.readMu.Lock()
+			err := d.r.Close()
+			h.readMu.Unlock()
+			if err != nil && firstErr == nil {
+				firstErr = err
+			}
+		}
+	})
+	return firstErr
+}
+
+// DrainCount returns the number of readers currently in the drain queue.
+// Useful for tests and operational metrics.
+func (h *SnapshotHolder[T]) DrainCount() int {
+	h.drainsMu.Lock()
+	defer h.drainsMu.Unlock()
+	return len(h.drains)
+}
+
+// reapNow scans the drain queue and Closes any readers whose grace has
+// elapsed. Called by the sweeper goroutine on a tick.
+func (h *SnapshotHolder[T]) reapNow() {
+	now := time.Now()
+	var alive []*pendingClose[T]
+	var reaped []*pendingClose[T]
+	h.drainsMu.Lock()
+	for _, d := range h.drains {
+		if !d.expireAt.After(now) {
+			reaped = append(reaped, d)
+		} else {
+			alive = append(alive, d)
+		}
+	}
+	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.
+		h.readMu.Lock()
+		_ = d.r.Close()
+		h.readMu.Unlock()
+	}
+}
+
+func (h *SnapshotHolder[T]) sweepLoop() {
+	interval := h.drainGrace / 4
+	if interval < 10*time.Millisecond {
+		interval = 10 * time.Millisecond
+	}
+	t := time.NewTicker(interval)
+	defer t.Stop()
+	for {
+		select {
+		case <-h.stopCh:
+			return
+		case <-t.C:
+			h.reapNow()
+		}
+	}
+}

projection/manager_test.go

@@ -0,0 +1,278 @@
+package projection
+
+import (
+	"fmt"
+	"path/filepath"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// 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 value (SKU matches, price > 0).
+//   - No gap — since every snapshot version contains every key, a Get must never
+//     miss; misses would mean the holder briefly exposed an empty/nil reader.
+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[tProj]()
+		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, tProj{ID: "sku:" + sku, SKU: sku, PriceIncVat: 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[tProj](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 key)", 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 }
+
+func (f *fakeCloser) Close() error { f.hits++; return nil }
+
+// withFakeCloser swaps SnapshotReader's closer for our counter.
+func withFakeCloser(t *testing.T, path string, fc *fakeCloser) *SnapshotReader[tProj] {
+	t.Helper()
+	r, err := ReadSnapshotFile[tProj](path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// replace real closer with our counter; the test asserts only via fc
+	r.raw.closer = fc.Close
+	return r
+}
+
+// TestSnapshotHolderReloadDrainsOld — Set swaps the current reader; the
+// previous reader is queued. Before grace: still Get-able. After grace:
+// reaped and Close fires.
+func TestSnapshotHolderReloadDrainsOld(t *testing.T) {
+	dir := t.TempDir()
+
+	// 1) Build a valid first snapshot (keys A-1 / A-2).
+	w := NewSnapshotWriter[tProj]()
+	if err := appendP(w, makeProjection("A-1", 100)); err != nil {
+		t.Fatal(err)
+	}
+	if err := appendP(w, makeProjection("A-2", 200)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := w.Write(dir, "first.bin"); err != nil {
+		t.Fatal(err)
+	}
+	firstCloser := &fakeCloser{}
+	first := withFakeCloser(t, filepath.Join(dir, "first.bin"), firstCloser)
+	defer first.Close()
+
+	// 2) Build a second snapshot (key B-1).
+	w2 := NewSnapshotWriter[tProj]()
+	if err := appendP(w2, makeProjection("B-1", 300)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := w2.Write(dir, "second.bin"); err != nil {
+		t.Fatal(err)
+	}
+	secondCloser := &fakeCloser{}
+	second := withFakeCloser(t, filepath.Join(dir, "second.bin"), secondCloser)
+	defer second.Close()
+
+	h := NewSnapshotHolder[tProj](50 * time.Millisecond)
+	defer h.Close()
+	h.Set(first)
+	if h.Len() != 2 {
+		t.Errorf("Len after first Set: got %d want 2", h.Len())
+	}
+	if _, ok := h.Get("A-1"); !ok {
+		t.Error("A-1 not found in initial state")
+	}
+
+	// 3) Swap to second — first is now queued.
+	h.Set(second)
+	if h.Len() != 1 {
+		t.Errorf("Len after second Set: got %d want 1", h.Len())
+	}
+	if firstCloser.hits != 0 {
+		t.Errorf("firstCloser hits: got %d want 0 (still in drain grace)", firstCloser.hits)
+	}
+	if h.DrainCount() != 1 {
+		t.Errorf("DrainCount: got %d want 1", h.DrainCount())
+	}
+
+	// 4) Wait past grace, force a reap tick.
+	time.Sleep(80 * time.Millisecond)
+	h.reapNow()
+	if firstCloser.hits != 1 {
+		t.Errorf("firstCloser hits: got %d want 1 (should have reaped)", firstCloser.hits)
+	}
+	if h.DrainCount() != 0 {
+		t.Errorf("DrainCount after reap: got %d want 0", h.DrainCount())
+	}
+}
+
+// TestSnapshotHolderEmptyGet — holder with no current serves empty.
+func TestSnapshotHolderEmptyGet(t *testing.T) {
+	h := NewSnapshotHolder[tProj](100 * time.Millisecond)
+	defer h.Close()
+	if _, ok := h.Get("anything"); ok {
+		t.Error("empty holder should return false")
+	}
+	if h.Len() != 0 {
+		t.Errorf("Len: got %d want 0", h.Len())
+	}
+}
+
+// TestSnapshotHolderCloseImmediatelyReaps — Close with pending drains
+// should close them now (no grace wait).
+func TestSnapshotHolderCloseImmediatelyReaps(t *testing.T) {
+	dir := t.TempDir()
+
+	w := NewSnapshotWriter[tProj]()
+	if err := appendP(w, makeProjection("X-1", 1)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := w.Write(dir, "x.bin"); err != nil {
+		t.Fatal(err)
+	}
+	fc1 := &fakeCloser{}
+	r1 := withFakeCloser(t, filepath.Join(dir, "x.bin"), fc1)
+
+	w2 := NewSnapshotWriter[tProj]()
+	if err := appendP(w2, makeProjection("Y-1", 2)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := w2.Write(dir, "y.bin"); err != nil {
+		t.Fatal(err)
+	}
+	fc2 := &fakeCloser{}
+	r2 := withFakeCloser(t, filepath.Join(dir, "y.bin"), fc2)
+
+	h := NewSnapshotHolder[tProj](10 * time.Second) // long grace so reap wouldn't happen on its own
+	h.Set(r1)
+	h.Set(r2)
+
+	if fc1.hits != 0 || fc2.hits != 0 {
+		t.Errorf("premature Close hits: fc1=%d fc2=%d", fc1.hits, fc2.hits)
+	}
+	if err := h.Close(); err != nil {
+		t.Errorf("Close: %v", err)
+	}
+	if fc1.hits != 1 || fc2.hits != 1 {
+		t.Errorf("after Close: fc1=%d fc2=%d want 1 each", fc1.hits, fc2.hits)
+	}
+}
+
+// TestSnapshotHolderIdentityReload — Open the same path twice; current reader
+// is replaced; DrainCount returns 1 (old queued).
+func TestSnapshotHolderIdentityReload(t *testing.T) {
+	dir := t.TempDir()
+	w := NewSnapshotWriter[tProj]()
+	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[tProj](10 * time.Second)
+	defer h.Close()
+	if err := h.Reload(filepath.Join(dir, "id.bin")); err != nil {
+		t.Fatal(err)
+	}
+	if err := h.Reload(filepath.Join(dir, "id.bin")); err != nil {
+		t.Fatal(err)
+	}
+	if h.DrainCount() != 1 {
+		t.Errorf("DrainCount after two reloads of same path: got %d want 1", h.DrainCount())
+	}
+	if _, ok := h.Get("ID-1"); !ok {
+		t.Error("current reader did not surface ID-1")
+	}
+}

projection/snapshot.go

@@ -0,0 +1,507 @@
+// Snapshot reader/writer — a value-agnostic, mmap-friendly key→blob store.
+//
+// This is the domain-neutral substrate extracted from platform/catalog: the
+// bus is the live channel, but it doesn't carry the historical population on
+// cold start. A snapshot file gives a consumer a fast mmap-loaded cold-start
+// substrate, so the first read doesn't have to wait for the bus to replay. The
+// bus then keeps it fresh. catalog re-exports these types for its Projection;
+// other domains (tracking popularity/suggestions) instantiate them over their
+// own value type.
+//
+// Format v2 (little-endian, fixed-width, mmap-friendly):
+//
+//	header   [72 bytes]   magic(8) version(4) reserved(4) count(8)
+//	                       blobStart(8) recordStart(8) tenant(32, zero-padded)
+//	blobs    variable     count sequential JSON-encoded value bodies, no padding
+//	records  count*44     each: key(32, zero-padded) blobOffset(8) blobLength(4);
+//	                       sorted by key via the single compareSKU total order
+//
+// Tenant field at offsets 40–71 (zero-padded) names the namespace this
+// snapshot covers. Independent key schemas can produce values into different
+// tenant files without conflict. Hot-reload across tenants stays safe under
+// SnapshotHolder (manager.go).
+//
+// Sorting + binary search gives O(log N) per-key lookup with no on-disk hash
+// table and no collisions. Keys longer than 32 bytes are rejected at write
+// time. (Domains with longer natural keys — page paths, query strings — hash
+// the key to a fixed-width digest and carry the original inside the value blob.)
+//
+// Memory model: the Reader holds one []byte for the entire file (mmap on
+// Unix, ReadFile elsewhere). Per-key Get reads at most log2(N) record slots
+// (each 44 B) and one JSON blob. Scan walks the sorted record section once.
+//
+// SAFETY: SnapshotReader alone is NOT safe across hot-reload. During reload,
+// in-flight Get()/Scan() calls on the OLD reader continue to work until disposal
+// of the old reader is finalised; calling SnapshotReader.Close() concurrently
+// with Get() on the same reader risks unmapping pages-under-read (SIGBUS).
+// Use SnapshotHolder (manager.go) for any reload-bearing setup — it gives you
+// an atomic.Pointer swap with a drain-grace (default 30s) so concurrent Gets on
+// the OLD reader finish naturally before munmap fires.
+package projection
+
+import (
+	"encoding/binary"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"math"
+	"os"
+	"path/filepath"
+	"sort"
+	"sync"
+)
+
+const (
+	snapshotMagic   uint64 = 0x3230564a53435050 // "PCSJ002" v2 (was 0x3130564a53435050 = v1)
+	snapshotVersion uint32 = 2
+	maxSKULen              = 32
+	maxTenantLen           = 32
+	headerSize             = 8 + 4 + 4 + 8 + 8 + 8 + 32 // 72 bytes
+	recordSize             = 32 + 8 + 4                 // 44 bytes per record
+)
+
+var (
+	ErrSnapshotMagic     = errors.New("projection: not a snapshot file (bad magic)")
+	ErrSnapshotVersion   = errors.New("projection: unsupported snapshot version")
+	ErrSnapshotTruncated = errors.New("projection: snapshot truncated")
+	ErrSnapshotBadBlob   = errors.New("projection: snapshot record points past file")
+	ErrSnapshotOverflow  = errors.New("projection: snapshot record-section size overflows uint64")
+	ErrSKUTooLong        = fmt.Errorf("projection: key longer than %d bytes", maxSKULen)
+	ErrTenantTooLong     = fmt.Errorf("projection: tenant name longer than %d bytes", maxTenantLen)
+)
+
+// 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).
+
+// rawSnapshotWriter is the value-agnostic writer: it indexes (key → blob bytes).
+// Single-shot: append() all, then write() once (atomic rename).
+type rawSnapshotWriter struct {
+	tenant  string
+	blobs   []byte
+	records []snapshotRecord
+	dirty   bool
+}
+
+// snapshotRecord is the on-disk layout: key zero-padded to 32 bytes, then
+// an 8-byte offset (from file start) plus a 4-byte length for the JSON blob.
+type snapshotRecord struct {
+	SKU        [32]byte
+	BlobOffset uint64
+	BlobLength uint32
+}
+
+// append adds a pre-encoded blob under key. ErrSKUTooLong if key exceeds the
+// on-disk width.
+func (w *rawSnapshotWriter) append(sku string, blob []byte) error {
+	if len(sku) > maxSKULen {
+		return ErrSKUTooLong
+	}
+	var rec snapshotRecord
+	copy(rec.SKU[:], sku)
+	rec.BlobOffset = uint64(headerSize + len(w.blobs))
+	rec.BlobLength = uint32(len(blob))
+
+	w.blobs = append(w.blobs, blob...)
+	w.records = append(w.records, rec)
+	w.dirty = true
+	return nil
+}
+
+// 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.
+type SnapshotWriter[T any] struct {
+	raw rawSnapshotWriter
+}
+
+// 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 key. Returns ErrSKUTooLong if key
+// exceeds the on-disk width; returns marshal errors as wrapped fmt.Errorf.
+// key 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("projection: 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 key asc before serialisation so the
+// reader's binary search works.
+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 {
+		err = ErrTenantTooLong
+		return "", err
+	}
+
+	if err = os.MkdirAll(dir, 0o755); err != nil {
+		return "", err
+	}
+	finalPath = filepath.Join(dir, name)
+	tmp := finalPath + ".tmp"
+
+	f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644)
+	if err != nil {
+		return "", err
+	}
+	defer func() {
+		if err != nil {
+			_ = os.Remove(tmp)
+			_ = f.Close()
+		}
+	}()
+
+	// Header: fixed 72 bytes (8 + 4 + 4 + 8 + 8 + 8 + 32).
+	blobStart := uint64(headerSize)
+	recordStart := blobStart + uint64(len(w.blobs))
+	var hdr [headerSize]byte
+	binary.LittleEndian.PutUint64(hdr[0:8], snapshotMagic)
+	binary.LittleEndian.PutUint32(hdr[8:12], snapshotVersion)
+	binary.LittleEndian.PutUint32(hdr[12:16], 0)
+	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
+
+	if _, werr := f.Write(hdr[:]); werr != nil {
+		err = werr
+		return "", err
+	}
+	if _, werr := f.Write(w.blobs); werr != nil {
+		err = werr
+		return "", err
+	}
+	recBytes := make([]byte, recordSize*len(w.records))
+	for i, r := range w.records {
+		off := i * recordSize
+		copy(recBytes[off:off+32], r.SKU[:])
+		binary.LittleEndian.PutUint64(recBytes[off+32:off+40], r.BlobOffset)
+		binary.LittleEndian.PutUint32(recBytes[off+40:off+44], r.BlobLength)
+	}
+	if _, werr := f.Write(recBytes); werr != nil {
+		err = werr
+		return "", err
+	}
+
+	if werr := f.Sync(); werr != nil {
+		err = werr
+		return "", err
+	}
+	if werr := f.Close(); werr != nil {
+		err = werr
+		return "", err
+	}
+	if werr := os.Rename(tmp, finalPath); werr != nil {
+		err = werr
+		return "", err
+	}
+	return finalPath, nil
+}
+
+// compareSKU defines the SINGLE TOTAL ORDER for keys across both Writer
+// (uses it for sort.Slice) and Reader (uses it for binary search). One
+// definition; invariant test verifies the on-disk result is sorted under
+// it. Source keys SHOULD NOT contain embedded null bytes — the padded
+// comparator treats them as terminators, which silently truncates a key.
+func compareSKU(padded, other []byte) int {
+	// padded is a fixed-size zero-padded slot; trim trailing zeros.
+	end := len(padded)
+	for end > 0 && padded[end-1] == 0 {
+		end--
+	}
+	a := padded[:end]
+	b := other
+
+	n := len(a)
+	if len(b) < n {
+		n = len(b)
+	}
+	for i := 0; i < n; i++ {
+		if a[i] != b[i] {
+			if a[i] < b[i] {
+				return -1
+			}
+			return 1
+		}
+	}
+	switch {
+	case len(a) < len(b):
+		return -1
+	case len(a) > len(b):
+		return 1
+	default:
+		return 0
+	}
+}
+
+// compareRecordLookup is compareSKU specialised for Reader lookups so the
+// argument order matches the lex order: "padded" (fixed-slot) and "want"
+// (query). compareSKU is symmetric so this is just an alias.
+func compareRecordLookup(padded, want []byte) int {
+	return compareSKU(padded, want)
+}
+
+// rawSnapshotReader is the value-agnostic reader: binary search by key, 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
+	recordStart uint64
+	Tenant      string // populated from header at Open time; "" if not set
+
+	closeOnce sync.Once
+	closer    func() error // munmap + close on Unix; no-op on ReadFile fallback
+}
+
+// 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 key, 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
+}
+
+// Scan walks every (key, value) in key-sorted order, calling fn for each.
+// Return false from fn to stop early. A value that fails to decode is skipped
+// (fn is not called for it). This is the enumeration / ranked-scan substrate:
+// callers that need top-N-by-value collect into a slice and sort in memory —
+// the on-disk order is by key, not value. NOT safe racing Close; drive it
+// through SnapshotHolder.Scan under reload.
+func (r *SnapshotReader[T]) Scan(fn func(key string, v T) bool) {
+	r.raw.scan(func(key string, blob []byte) bool {
+		var v T
+		if err := json.Unmarshal(blob, &v); err != nil {
+			return true // skip undecodable, keep going
+		}
+		return fn(key, v)
+	})
+}
+
+// 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 readRawSnapshotFile(path string) (*rawSnapshotReader, error) {
+	data, closer, err := loadSnapshotBytes(path)
+	if err != nil {
+		return nil, err
+	}
+	r, err := newRawSnapshotReader(data)
+	if err != nil {
+		if closer != nil {
+			_ = closer()
+		}
+		return nil, err
+	}
+	r.closer = closer
+	return r, nil
+}
+
+func newRawSnapshotReader(data []byte) (*rawSnapshotReader, error) {
+	if len(data) < headerSize {
+		return nil, ErrSnapshotTruncated
+	}
+	magic := binary.LittleEndian.Uint64(data[0:8])
+	if magic != snapshotMagic {
+		return nil, ErrSnapshotMagic
+	}
+	version := binary.LittleEndian.Uint32(data[8:12])
+	if version != snapshotVersion {
+		return nil, ErrSnapshotVersion
+	}
+	count := binary.LittleEndian.Uint64(data[16:24])
+	blobStart := binary.LittleEndian.Uint64(data[24:32])
+	recordStart := binary.LittleEndian.Uint64(data[32:40])
+	if blobStart != uint64(headerSize) {
+		return nil, fmt.Errorf("projection: blobStart=%d expected %d", blobStart, headerSize)
+	}
+
+	// Multiplication overflow guard: for a malformed/tampered count, the
+	// naive int(count)*recordSize in Go silently wraps on overflow,
+	// making expectedSize negative and bypassing the truncation guard.
+	// Compute as uint64 with an explicit overflow check.
+	recordBytes := uint64(recordSize) * count
+	if count > 0 && recordBytes/uint64(count) != uint64(recordSize) {
+		return nil, fmt.Errorf("%w: count*recordSize overflows (count=%d)", ErrSnapshotOverflow, count)
+	}
+	if recordStart > math.MaxUint64-recordBytes {
+		return nil, fmt.Errorf("%w: recordStart+records exceeds uint64 (recordStart=%d count=%d)",
+			ErrSnapshotOverflow, recordStart, count)
+	}
+	expectedSize := recordStart + recordBytes
+	if uint64(len(data)) < expectedSize {
+		return nil, ErrSnapshotTruncated
+	}
+
+	// Per-record blob-offset bounds check: defends against malformed /
+	// tampered files where the record section is intact but the blob
+	// pointer lies past EOF. A linear pass; cheap (44 B × count at open).
+	dataLen := uint64(len(data))
+	for i := uint64(0); i < count; i++ {
+		recOff := recordStart + i*uint64(recordSize)
+		blobLen := binary.LittleEndian.Uint32(data[recOff+40 : recOff+44])
+		blobOff := binary.LittleEndian.Uint64(data[recOff+32 : recOff+40])
+		if blobOff > dataLen || uint64(blobLen) > dataLen-blobOff {
+			return nil, fmt.Errorf("%w: record %d blobOff=%d len=%d exceeds file size %d",
+				ErrSnapshotBadBlob, i, blobOff, blobLen, dataLen)
+		}
+	}
+
+	// Tenant: read zero-padded bytes at offset 40, trim trailing zeros.
+	tenantBytes := data[40 : 40+maxTenantLen]
+	tenantEnd := len(tenantBytes)
+	for tenantEnd > 0 && tenantBytes[tenantEnd-1] == 0 {
+		tenantEnd--
+	}
+	tenant := string(tenantBytes[:tenantEnd])
+
+	return &rawSnapshotReader{
+		data:        data,
+		count:       count,
+		blobStart:   blobStart,
+		recordStart: recordStart,
+		Tenant:      tenant,
+	}, nil
+}
+
+// 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).
+func (r *rawSnapshotReader) Close() error {
+	var err error
+	r.closeOnce.Do(func() {
+		if r.closer != nil {
+			err = r.closer()
+		}
+	})
+	return err
+}
+
+// getRaw returns the blob bytes for key, 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 nil, false
+	}
+	skuBytes := []byte(sku)
+	if len(skuBytes) > maxSKULen {
+		return nil, false
+	}
+
+	lo, hi := uint64(0), r.count
+	for lo < hi {
+		mid := (lo + hi) / 2
+		recOff := int(r.recordStart) + int(mid)*recordSize
+		recSKU := r.data[recOff : recOff+32]
+		cmp := compareRecordLookup(recSKU, skuBytes)
+		switch {
+		case cmp < 0:
+			lo = mid + 1
+		case cmp > 0:
+			hi = mid
+		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.
+			return r.data[blobOff : blobOff+uint64(blobLen)], true
+		}
+	}
+	return nil, false
+}
+
+// scan walks every record in key-sorted order, yielding (key, blob). Return
+// false from fn to stop. The blob aliases the mmap region — valid until Close.
+func (r *rawSnapshotReader) scan(fn func(key string, blob []byte) bool) {
+	for i := uint64(0); i < r.count; i++ {
+		recOff := int(r.recordStart) + int(i)*recordSize
+		recSKU := r.data[recOff : recOff+32]
+		end := len(recSKU)
+		for end > 0 && recSKU[end-1] == 0 {
+			end--
+		}
+		key := string(recSKU[:end])
+		blobLen := binary.LittleEndian.Uint32(r.data[recOff+40 : recOff+44])
+		blobOff := binary.LittleEndian.Uint64(r.data[recOff+32 : recOff+40])
+		if !fn(key, r.data[blobOff:blobOff+uint64(blobLen)]) {
+			return
+		}
+	}
+}

projection/snapshot_other.go

@@ -0,0 +1,17 @@
+//go:build !unix
+
+// Non-unix fallback for SnapshotReader: ReadFile into a []byte. Functionally
+// identical to the unix mmap path — same reader code, same lookups — just
+// pays the full resident-set cost at Open. Production mmap-on-Linux/Darwin;
+// this exists for windows / plan9 / wasm builds.
+package projection
+
+import "os"
+
+func loadSnapshotBytes(path string) (data []byte, closer func() error, err error) {
+	data, err = os.ReadFile(path)
+	if err != nil {
+		return nil, nil, err
+	}
+	return data, func() error { return nil }, nil
+}

projection/snapshot_test.go

@@ -0,0 +1,478 @@
+package projection
+
+import (
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"math"
+	"math/rand"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+// TestSnapshotRoundTrip — write 1000 values in shuffled order, read back
+// and verify all 1000 are found at the right key.
+func TestSnapshotRoundTrip(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "snap.bin")
+
+	const n = 1000
+	w := NewSnapshotWriter[tProj]().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 := appendP(w, makeProjection(sku, int64(100+(i%97)*100))); err != nil {
+			t.Fatalf("Append %s: %v", sku, err)
+		}
+	}
+	if w.Len() != n {
+		t.Fatalf("Len: got %d want %d", w.Len(), n)
+	}
+	final, err := w.Write(dir, "snap.bin")
+	if err != nil {
+		t.Fatalf("Write: %v", err)
+	}
+	if final != path {
+		t.Errorf("Write returned %q expected %q", final, path)
+	}
+
+	r, err := ReadSnapshotFile[tProj](path)
+	if err != nil {
+		t.Fatalf("ReadSnapshotFile: %v", err)
+	}
+	defer r.Close()
+
+	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")
+	}
+
+	// every key present and correct
+	for _, i := range rand.Perm(n)[:50] { // probe 50 random lookups
+		sku := fmt.Sprintf("SKU-%05d", i)
+		got, ok := r.Get(sku)
+		if !ok {
+			t.Errorf("Get(%q) not found", sku)
+			continue
+		}
+		want := int64(100 + (i%97)*100)
+		if got.PriceIncVat != want {
+			t.Errorf("Get(%q) PriceIncVat: got %d want %d", sku, got.PriceIncVat, want)
+		}
+	}
+
+	// missing key
+	if _, ok := r.Get("NOPE-99999"); ok {
+		t.Error("Get on missing key should return ok=false")
+	}
+}
+
+// TestSnapshotScanWalksAllInKeyOrder — Scan visits every record exactly once in
+// ascending key order, and early-return stops the walk.
+func TestSnapshotScanWalksAllInKeyOrder(t *testing.T) {
+	dir := t.TempDir()
+	const n = 300
+	w := NewSnapshotWriter[tProj]()
+	for _, i := range rand.Perm(n) {
+		sku := fmt.Sprintf("K-%04d", i)
+		if err := appendP(w, makeProjection(sku, int64(i))); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := w.Write(dir, "scan.bin"); err != nil {
+		t.Fatal(err)
+	}
+	r, err := ReadSnapshotFile[tProj](filepath.Join(dir, "scan.bin"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer r.Close()
+
+	var keys []string
+	r.Scan(func(key string, v tProj) bool {
+		if v.SKU != key {
+			t.Errorf("Scan value/key mismatch: key=%q value.SKU=%q", key, v.SKU)
+		}
+		keys = append(keys, key)
+		return true
+	})
+	if len(keys) != n {
+		t.Fatalf("Scan visited %d records, want %d", len(keys), n)
+	}
+	for i := 1; i < len(keys); i++ {
+		if keys[i-1] >= keys[i] {
+			t.Fatalf("Scan not ascending at %d: %q >= %q", i, keys[i-1], keys[i])
+		}
+	}
+
+	// early-return stops the walk
+	count := 0
+	r.Scan(func(string, tProj) bool {
+		count++
+		return count < 5
+	})
+	if count != 5 {
+		t.Errorf("Scan early-return visited %d, want 5", count)
+	}
+}
+
+// TestSnapshotCompareSKUInvariant — verify the on-disk record section is
+// sorted under the same compareSKU total order that BinarySearch uses.
+func TestSnapshotCompareSKUInvariant(t *testing.T) {
+	dir := t.TempDir()
+	w := NewSnapshotWriter[tProj]()
+	// 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 := 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[tProj](filepath.Join(dir, "snap.bin"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer r.Close()
+
+	// Walk every record slot, harvest key bytes from the pad-then-trim
+	// form, and assert non-decreasing order under compareSKU.
+	prevSKU := ""
+	for i := 0; i < r.Len(); i++ {
+		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 {
+			end--
+		}
+		cur := string(raw[:end])
+		if i > 0 {
+			if compareSKU(runesToPadded32(prevSKU), []byte(cur)) > 0 {
+				t.Errorf("record section out of order at i=%d: prev=%q cur=%q", i, prevSKU, cur)
+			}
+		}
+		prevSKU = cur
+	}
+}
+
+// runesToPadded32 zero-pads s into a 32-byte slot for compareSKU in the
+// invariant test.
+func runesToPadded32(s string) []byte {
+	out := make([]byte, 32)
+	copy(out, s)
+	return out
+}
+
+// TestSnapshotTenantRoundTrip — Writer.WithTenant lands in the header and
+// Reader surfaces it.
+func TestSnapshotTenantRoundTrip(t *testing.T) {
+	dir := t.TempDir()
+	w := NewSnapshotWriter[tProj]().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[tProj](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")
+	}
+
+	// Empty tenant writes zero-padded.
+	w2 := NewSnapshotWriter[tProj]()
+	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[tProj](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())
+	}
+}
+
+// TestSnapshotTenantTooLong — over-long tenant name rejected at Write.
+func TestSnapshotTenantTooLong(t *testing.T) {
+	dir := t.TempDir()
+	w := NewSnapshotWriter[tProj]().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 {
+		t.Error("expected error for over-long tenant")
+	} else if !errors.Is(err, ErrTenantTooLong) {
+		t.Errorf("got %v want ErrTenantTooLong", err)
+	}
+}
+
+// TestSnapshotCountOverflowRejected — tampered count that would cause
+// recordStart + records*recordSize to overflow uint64 must be rejected
+// with ErrSnapshotOverflow.
+func TestSnapshotCountOverflowRejected(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "over.bin")
+	bs := make([]byte, headerSize)
+	binary.LittleEndian.PutUint64(bs[0:8], snapshotMagic)
+	binary.LittleEndian.PutUint32(bs[8:12], snapshotVersion)
+	binary.LittleEndian.PutUint32(bs[12:16], 0)
+	// count that overflows when multiplied by recordSize=44
+	binary.LittleEndian.PutUint64(bs[16:24], math.MaxUint64/uint64(recordSize)+1)
+	binary.LittleEndian.PutUint64(bs[24:32], uint64(headerSize))
+	binary.LittleEndian.PutUint64(bs[32:40], uint64(headerSize))
+	if err := os2WriteFile(path, bs, 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := ReadSnapshotFile[tProj](path); !errors.Is(err, ErrSnapshotOverflow) {
+		t.Errorf("overflow count rejected with %v, want ErrSnapshotOverflow", err)
+	}
+}
+
+// TestSnapshotAtomicSwap — verify the temporary-then-rename discipline never
+// leaves a half-written snap.bin visible to readers.
+func TestSnapshotAtomicSwap(t *testing.T) {
+	dir := t.TempDir()
+	w := NewSnapshotWriter[tProj]()
+	mustAppend := func(w *SnapshotWriter[tProj], sku string, price int64) {
+		t.Helper()
+		if err := appendP(w, makeProjection(sku, price)); err != nil {
+			t.Fatalf("Append %s: %v", sku, err)
+		}
+	}
+	mustAppend(w, "AAA", 100)
+	final, err := w.Write(dir, "snap.bin")
+	if err != nil {
+		t.Fatal(err)
+	}
+	// tmp must not exist anymore
+	if _, err := os2Stat(final + ".tmp"); err == nil {
+		t.Errorf(".tmp still present after Write")
+	}
+
+	// replace
+	w2 := NewSnapshotWriter[tProj]()
+	mustAppend(w2, "ZZZ", 200)
+	if _, err := w2.Write(dir, "snap.bin"); err != nil {
+		t.Fatal(err)
+	}
+
+	r, err := ReadSnapshotFile[tProj](final)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer r.Close()
+	if _, ok := r.Get("ZZZ"); !ok {
+		t.Error("second-snapshot Get(ZZZ) missing")
+	}
+	if _, ok := r.Get("AAA"); ok {
+		t.Error("first-snapshot residue Get(AAA) found — atomic swap bug")
+	}
+}
+
+// TestSnapshotRejectBadInput — magic, version, truncation, blob-offset OOB.
+func TestSnapshotRejectBadInput(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "bad.bin")
+
+	// Empty file: truncated.
+	if err := os2WriteFile(path, []byte{}, 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := ReadSnapshotFile[tProj](path); !errors.Is(err, ErrSnapshotTruncated) {
+		t.Errorf("empty file rejected with %v, want ErrSnapshotTruncated", err)
+	}
+
+	// Wrong magic. The fixture must be at least headerSize bytes long
+	// (v2 header is 72 B), otherwise we'd hit ErrSnapshotTruncated
+	// before the magic check fires.
+	wrongMagic := make([]byte, headerSize+1)
+	for i := range wrongMagic {
+		wrongMagic[i] = 0xff
+	}
+	if err := os2WriteFile(path, wrongMagic, 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := ReadSnapshotFile[tProj](path); !errors.Is(err, ErrSnapshotMagic) {
+		t.Errorf("wrong magic rejected with %v, want ErrSnapshotMagic", err)
+	}
+
+	// Truncated mid-record. A valid header claiming 5 records but with no
+	// record section.
+	if err := os2WriteFile(path, buildValidHeader(5), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := ReadSnapshotFile[tProj](path); !errors.Is(err, ErrSnapshotTruncated) {
+		t.Errorf("truncated snapshot rejected with %v, want ErrSnapshotTruncated", err)
+	}
+
+	// Tampered blob offsets: valid header, valid record section with a
+	// blob pointer that lies past EOF. Should reject with ErrSnapshotBadBlob.
+	tampered := buildTamperedHeader()
+	if err := os2WriteFile(path, tampered, 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := ReadSnapshotFile[tProj](path); !errors.Is(err, ErrSnapshotBadBlob) {
+		t.Errorf("tampered-blob rejected with %v, want ErrSnapshotBadBlob", err)
+	}
+}
+
+// TestSnapshotEmpty — empty (count=0) snapshot is valid; Get returns false.
+func TestSnapshotEmpty(t *testing.T) {
+	dir := t.TempDir()
+	w := NewSnapshotWriter[tProj]()
+	if _, err := w.Write(dir, "empty.bin"); err != nil {
+		t.Fatal(err)
+	}
+	r, err := ReadSnapshotFile[tProj](filepath.Join(dir, "empty.bin"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer r.Close()
+	if r.Len() != 0 {
+		t.Errorf("Len: got %d want 0", r.Len())
+	}
+	if _, ok := r.Get("anything"); ok {
+		t.Error("empty snapshot should never return ok=true")
+	}
+}
+
+// TestSnapshotSKU32 — key exactly 32 chars (the boundary case) must succeed.
+func TestSnapshotSKU32(t *testing.T) {
+	dir := t.TempDir()
+	sku := "SKU-" + repeat32("X", 32-4) // 4 + 28 = 32 chars
+	if len(sku) != 32 {
+		t.Fatalf("test bug: sku length %d", len(sku))
+	}
+	w := NewSnapshotWriter[tProj]()
+	if err := appendP(w, makeProjection(sku, 1)); err != nil {
+		t.Fatalf("Append 32-char key: %v", err)
+	}
+	if _, err := w.Write(dir, "snap.bin"); err != nil {
+		t.Fatal(err)
+	}
+	r, err := ReadSnapshotFile[tProj](filepath.Join(dir, "snap.bin"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer r.Close()
+	if _, ok := r.Get(sku); !ok {
+		t.Error("32-char key Get returned false")
+	}
+}
+
+// TestSnapshotSKU33Rejected — one char over should error at Append.
+func TestSnapshotSKU33Rejected(t *testing.T) {
+	w := NewSnapshotWriter[tProj]()
+	sku := repeat32("X", 33)
+	err := appendP(w, makeProjection(sku, 1))
+	if !errors.Is(err, ErrSKUTooLong) {
+		t.Errorf("33-char key Append: got %v want ErrSKUTooLong", err)
+	}
+}
+
+// TestSnapshotGetSpeed — probe thousands of random key lookups.
+func TestSnapshotGetSpeed(t *testing.T) {
+	if testing.Short() {
+		t.Skip("short mode")
+	}
+	dir := t.TempDir()
+	const n = 5000
+	w := NewSnapshotWriter[tProj]()
+	skus := make([]string, n)
+	for i := 0; i < n; i++ {
+		skus[i] = fmt.Sprintf("Z%06d", i)
+		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[tProj](filepath.Join(dir, "snap.bin"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer r.Close()
+
+	start := time.Now()
+	hits, misses := 0, 0
+	for _, i := range rand.Perm(n) {
+		if _, ok := r.Get(skus[i]); ok {
+			hits++
+		} else {
+			misses++
+		}
+	}
+	elapsed := time.Since(start)
+	t.Logf("100%% hit Get on %d records: %d hits, %d misses, %v (%v/lookup)",
+		n, hits, misses, elapsed, elapsed/time.Duration(n))
+	if misses != 0 {
+		t.Errorf("expected zero misses on full-key set, got %d", misses)
+	}
+}
+
+// buildValidHeader synthesises a header byte slice with the given Count
+// and zero-blob / zero-record section, used to exercise the truncation
+// branch in newSnapshotReader.
+func buildValidHeader(count uint64) []byte {
+	bs := make([]byte, headerSize)
+	binary.LittleEndian.PutUint64(bs[0:8], snapshotMagic)
+	binary.LittleEndian.PutUint32(bs[8:12], snapshotVersion)
+	binary.LittleEndian.PutUint32(bs[12:16], 0)
+	binary.LittleEndian.PutUint64(bs[16:24], count)
+	binary.LittleEndian.PutUint64(bs[24:32], uint64(headerSize)) // blobStart
+	binary.LittleEndian.PutUint64(bs[32:40], uint64(headerSize)) // recordStart (truncated)
+	return bs
+}
+
+// buildTamperedHeader returns a buffer with a valid header claiming 1 record
+// AND its 44-byte record section, but the record's blob pointer lies past
+// EOF. Used to exercise ErrSnapshotBadBlob.
+func buildTamperedHeader() []byte {
+	bs := make([]byte, headerSize+recordSize)
+	binary.LittleEndian.PutUint64(bs[0:8], snapshotMagic)
+	binary.LittleEndian.PutUint32(bs[8:12], snapshotVersion)
+	binary.LittleEndian.PutUint32(bs[12:16], 0)
+	binary.LittleEndian.PutUint64(bs[16:24], 1)
+	binary.LittleEndian.PutUint64(bs[24:32], uint64(headerSize))
+	binary.LittleEndian.PutUint64(bs[32:40], uint64(headerSize))
+	binary.LittleEndian.PutUint64(bs[headerSize+32:headerSize+40], ^uint64(0))
+	binary.LittleEndian.PutUint32(bs[headerSize+40:headerSize+44], ^uint32(0))
+	return bs
+}
+
+// TestSnapshotRejectBadVersion — pin the ErrSnapshotVersion rejection path.
+func TestSnapshotRejectBadVersion(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "badver.bin")
+	bs := make([]byte, headerSize)
+	binary.LittleEndian.PutUint64(bs[0:8], snapshotMagic)
+	binary.LittleEndian.PutUint32(bs[8:12], snapshotVersion+1)
+	binary.LittleEndian.PutUint32(bs[12:16], 0)
+	binary.LittleEndian.PutUint64(bs[16:24], 0)
+	binary.LittleEndian.PutUint64(bs[24:32], uint64(headerSize))
+	binary.LittleEndian.PutUint64(bs[32:40], uint64(headerSize))
+	if err := os2WriteFile(path, bs, 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := ReadSnapshotFile[tProj](path); !errors.Is(err, ErrSnapshotVersion) {
+		t.Errorf("v2-not-supported header rejected with %v, want ErrSnapshotVersion", err)
+	}
+}

projection/snapshot_unix.go

@@ -0,0 +1,61 @@
+//go:build unix
+
+// mmap-on-unix loader for SnapshotReader. The file handle stays alive for
+// the lifetime of the mmap region (the kernel needs it pinned). Closer
+// munmaps AND closes the fd. Uses MADV_RANDOM because the dominant access
+// pattern is binary-search-by-key — touching random record slots, not
+// streaming the file.
+package projection
+
+import (
+	"fmt"
+	"os"
+
+	"golang.org/x/sys/unix"
+)
+
+// loadSnapshotBytes returns the file contents (mmap'd) + a closer that munmaps
+// and closes the fd.
+func loadSnapshotBytes(path string) (data []byte, closer func() error, err error) {
+	f, err := os.Open(path)
+	if err != nil {
+		return nil, nil, err
+	}
+	info, err := f.Stat()
+	if err != nil {
+		_ = f.Close()
+		return nil, nil, err
+	}
+	size := info.Size()
+	if size == 0 {
+		_ = f.Close()
+		return []byte{}, func() error { return nil }, nil
+	}
+	if size < int64(headerSize) {
+		_ = f.Close()
+		return nil, nil, ErrSnapshotTruncated
+	}
+
+	data, mmapErr := unix.Mmap(int(f.Fd()), 0, int(size), unix.PROT_READ, unix.MAP_SHARED)
+	if mmapErr != nil {
+		_ = f.Close()
+		return nil, nil, fmt.Errorf("projection: mmap %s: %w", path, mmapErr)
+	}
+
+	// Binary-search by key = random page access; MADV_RANDOM tells the
+	// kernel NOT to do read-ahead — saves I/O on cold layers.
+	if maErr := unix.Madvise(data, unix.MADV_RANDOM); maErr != nil {
+		// non-fatal; degrade gracefully
+		_ = maErr
+	}
+
+	closer = func() error {
+		mErr := unix.Munmap(data)
+		cErr := f.Close()
+		if mErr != nil {
+			return mErr
+		}
+		return cErr
+	}
+	return data, closer, nil
+}

projection/store.go

@@ -0,0 +1,346 @@
+package projection
+
+// Store — the consumer half of the projection seam. A service feeds it the
+// snapshot/delta frames it receives off the bus and reads facts back by key.
+// 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. Scan walks
+// overlay + base (overlay wins, tombstones hidden) for enumeration / ranked
+// output.
+//
+// The store is generic over the WIRE value W (what rides on the bus, decoded by
+// the supplied Decoder) and the STORED value T (what the Transform projects W
+// down to, held in the .bin). For an identity consumer W == T with an identity
+// transform; catalog uses W = catalog.Projection.
+//
+// No shared volume: each consumer builds its own base from the bus. Concurrency:
+// HandleFrame is single-writer (drive it from one delivery goroutine); Get/Scan
+// are safe from many readers.
+
+import (
+	"fmt"
+	"log"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+)
+
+// Store is a consumer's local, cold-start-ready projection cache of T, fed from
+// a bus stream of wire values W.
+type Store[W, T any] struct {
+	dir, name string
+	decode    Decoder[W]
+	transform Transform[W, T]
+	fallback  Fallback[W]
+	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
+}
+
+// NewStore opens a store backed by dir/name (its own .bin). decode and
+// transform are REQUIRED: decode parses the domain wire schema from a frame
+// payload; transform projects each wire value to the stored shape. 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 NewStore[W, T any](dir, name string, decode Decoder[W], transform Transform[W, T], fallback Fallback[W]) (*Store[W, T], error) {
+	if dir == "" || name == "" {
+		return nil, fmt.Errorf("projection: Store needs dir and name")
+	}
+	if decode == nil {
+		return nil, fmt.Errorf("projection: Store needs a decoder (declare the domain wire schema)")
+	}
+	if transform == nil {
+		return nil, fmt.Errorf("projection: Store needs a transform (declare which facts this consumer stores)")
+	}
+	s := &Store[W, T]{
+		dir:       dir,
+		name:      name,
+		decode:    decode,
+		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("projection: Store 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("projection: Store ignoring unreadable base %s: %v", path, err)
+	}
+	return s, nil
+}
+
+// HandleFrame applies one projection 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 *Store[W, 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 *Store[W, 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 *Store[W, T]) chunkSnapshot(epoch int64, payload []byte) error {
+	ups, err := s.decode(payload)
+	if err != nil {
+		return fmt.Errorf("projection: 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.Key == "" {
+			continue // a snapshot carries live entries only
+		}
+		// Transform applied here — the base stores the consumer's shape.
+		if err := s.building.Append(u.Key, s.transform(u.Value)); err != nil {
+			log.Printf("projection: Store skip %q in snapshot: %v", u.Key, err)
+			continue
+		}
+		s.buildingSeen++
+	}
+	return nil
+}
+
+func (s *Store[W, 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 projection silently is worse than stale.
+	if cs := meta[MetaCount]; cs != "" {
+		if want, perr := strconv.Atoi(cs); perr == nil && want != s.buildingSeen {
+			log.Printf("projection: Store 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("projection: 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.
+	writeEpochSidecar(path, epoch)
+	if err := s.holder.Reload(path); err != nil {
+		return fmt.Errorf("projection: reload snapshot base: %w", err)
+	}
+	s.epoch = epoch
+	s.overlay = make(map[string]overlayEntry[T]) // base now includes everything ≤ epoch
+	log.Printf("projection: Store snapshot applied — base swapped epoch=%d (%d entries)", epoch, s.buildingSeen)
+	return nil
+}
+
+func (s *Store[W, T]) applyDelta(epoch int64, payload []byte) error {
+	ups, err := s.decode(payload)
+	if err != nil {
+		return fmt.Errorf("projection: 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
+	}
+	var upserts, deletes int
+	for _, u := range ups {
+		if u.Key == "" {
+			continue
+		}
+		if u.Deleted {
+			s.overlay[u.Key] = overlayEntry[T]{deleted: true}
+			deletes++
+			continue
+		}
+		// Transform applied here too — overlay holds the consumer's shape.
+		s.overlay[u.Key] = overlayEntry[T]{val: s.transform(u.Value)}
+		upserts++
+	}
+	if upserts+deletes > 0 {
+		log.Printf("projection: Store delta applied epoch=%d (+%d/-%d, overlay=%d)",
+			epoch, upserts, deletes, len(s.overlay))
+	}
+	return nil
+}
+
+// Get resolves a value by key: overlay (incl. tombstone) → mmap base →
+// fallback. A tombstoned key returns (zero, false) and never reaches the
+// fallback. A fallback hit is run through the same Transform.
+func (s *Store[W, 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 w, ok := s.fallback(sku); ok {
+			return s.transform(w), true
+		}
+	}
+	return zero, false
+}
+
+// Scan walks every live (key, value) — overlay first (overlay wins; tombstones
+// hidden), then the mmap base minus any key the overlay already covered. Return
+// false from fn to stop early. The fallback is NOT consulted (it resolves
+// single misses, not enumeration). Callers that need top-N-by-value collect and
+// sort in memory — on-disk order is by key. Held under RLock for the whole
+// walk, so it blocks a concurrent snapshot swap; keep fn cheap.
+func (s *Store[W, T]) Scan(fn func(key string, v T) bool) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	seen := make(map[string]struct{}, len(s.overlay))
+	for k, e := range s.overlay {
+		seen[k] = struct{}{}
+		if e.deleted {
+			continue
+		}
+		if !fn(k, e.val) {
+			return
+		}
+	}
+	s.holder.Scan(func(k string, v T) bool {
+		if _, ok := seen[k]; ok {
+			return true // overlay already emitted (or tombstoned) this key
+		}
+		return fn(k, v)
+	})
+}
+
+// IsDeleted reports whether the overlay holds a tombstone for key — a delete
+// seen on the bus since the current base.
+func (s *Store[W, 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 *Store[W, T]) Epoch() int64 {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.epoch
+}
+
+// Stats returns observability counters (base size, overlay size, epoch).
+func (s *Store[W, 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 *Store[W, 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
+// "<name>.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.
+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("projection: Store epoch sidecar write: %v", err)
+		return
+	}
+	if err := os.Rename(tmp, p); err != nil {
+		log.Printf("projection: Store 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
+}