cart: back projection cache with platform/catalog.ProjectionStore

Consumer side of the catalog-projection seam (C1):
- catalogProjectionCache now wraps ProjectionStore[catalog.Projection] (identity
  transform — the cart's overlay uses ~all fields); delivery handler routes bus
  frames via HandleFrame (snapshot begin|chunk|end + epoch-stamped deltas).
- Per-pod cold-start-ready .bin under CART_PROJECTION_DIR (pod-local, never
  shared) + IsDeleted tombstone passthrough; HTTP fetch demoted to last-resort.
- Tests: test-only Apply shim + mustCache helper keep existing cases green;
  new snapshot-framing integration test.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

cmd/cart/main.go

@@ -11,7 +11,6 @@ import (
 	"os"
 	"os/signal"
 	"strings"
-	"sync"
 	"time"
 
 	"git.k6n.net/mats/go-cart-actor/internal/ucp"
@@ -120,91 +119,60 @@ func matchesSkuAndLocation(update inventory.InventoryResult, item cart.CartItem)
 	return false
 }
 
-// catalogProjectionCache is the cart's in-process map of SKU → catalog.Projection,
-// updated by reading catalog.projection_published events off the bus. It is the
-// cart-side source of truth for catalog facts (price, tax_class, inventory_tracked
-// flag, currency, image, brand) used at add-to-cart / checkout-reserve decisions.
-// Stock state is NOT cached here — exact qty is still queried synchronously from
-// the inventory service at decision points per docs/inventory-shape.md.
+// catalogProjectionCache is the cart's per-pod, cold-start-ready catalog
+// projection cache, backed by platform/catalog.ProjectionStore. It consumes the
+// catalog.projection_published wire — a full snapshot (begin/chunk/end, framed by
+// epoch) plus incremental deltas — into its OWN local .bin and serves catalog
+// facts by SKU (price, tax_class, inventory_tracked, image, …) at add-to-cart /
+// checkout-reserve decisions. Stock is NOT cached — queried synchronously from
+// inventory per docs/inventory-shape.md.
 //
-// Concurrency: read-mostly workload justifies sync.RWMutex.
+// Clustering: the cart runs N replicas. Each pod has its OWN exclusive bus queue
+// (rabbit.ListenToPattern declares an exclusive server-named queue, so every pod
+// receives the full stream) and its OWN pod-local .bin. It's a replicated read
+// cache — no leader, no shared volume; N pods writing one file would corrupt it.
+// See docs/catalog-projection.md.
 //
-// Single-tier storage. The bus is the only writer — the cart does not separately
-// seed from a snapshot file at boot, because the access pattern is point lookup
-// of cart-active SKUs only, and coupling two sources (mmap snapshot + bus-fed
-// map) created duplicated state with subtle drift risk and an mmap-SIGBUS hazard.
-// Cold-grace: the cache may be incomplete for an SKU the cart sees before the
-// first projection_published arrives for it; the cart falls back to the existing
-// PRODUCT_BASE_URL HTTP fetch and overlays the cache fields on top of it.
-//
-// Deletes carry a tombstone in the map (not a plain delete) so a bus delete of
-// an SKU that already fits the cache rules correctly shadows it for Get /
-// IsDeleted lookups — preventing an oversight from re-fetching a deleted SKU
-// before the cache rebuilds. Tombstones have a per-pod lifetime; bus-driven,
-// so the next sweep on the same SKU flips the entry back to live.
+// The cart stores the full catalog.Projection (identity transform): its
+// add-to-cart overlay (projection_overlay.go) consumes nearly every field, so
+// there's nothing to subset. Deletes ride as tombstones in the store's overlay
+// (IsDeleted) so a removed SKU isn't re-fetched before the next snapshot folds
+// the delete into the base. Cold-grace: an SKU not yet in the base/overlay
+// resolves via the existing PRODUCT_BASE_URL HTTP fetch + overlay.
 type catalogProjectionCache struct {
-	mu    sync.RWMutex
-	items map[string]projectionEntry
+	store *catalog.ProjectionStore[catalog.Projection]
 }
 
-// projectionEntry is a cache slot: a live projection, or a tombstone (deleted)
-// that shadows the live fallback.
-type projectionEntry struct {
-	proj    catalog.Projection
-	deleted bool
-}
+// identityProjection is the cart's mandatory transform — it stores the full
+// Projection because the overlay uses almost all of it.
+func identityProjection(p catalog.Projection) catalog.Projection { return p }
 
-func newCatalogProjectionCache() *catalogProjectionCache {
-	return &catalogProjectionCache{items: make(map[string]projectionEntry, 8192)}
-}
-
-func (c *catalogProjectionCache) Apply(updates []catalog.ProjectionUpdate) (upserts, deletes int) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	for _, u := range updates {
-		if u.Deleted {
-			c.items[u.SKU] = projectionEntry{deleted: true}
-			deletes++
-			continue
+// newCatalogProjectionCache opens the per-pod store under dir. dir MUST be
+// pod-local (container fs / an emptyDir in k8s) — never a shared volume.
+func newCatalogProjectionCache(dir string) (*catalogProjectionCache, error) {
+	s, err := catalog.NewProjectionStore(dir, "cart-projection.bin", identityProjection, nil)
+	if err != nil {
+		return nil, err
 	}
-		if u.SKU == "" {
-			continue
-		}
-		c.items[u.SKU] = projectionEntry{proj: u.Projection}
-		upserts++
-	}
-	return
+	return &catalogProjectionCache{store: s}, nil
 }
 
-func (c *catalogProjectionCache) Get(sku string) (catalog.Projection, bool) {
-	c.mu.RLock()
-	e, ok := c.items[sku]
-	c.mu.RUnlock()
-	if !ok {
-		return catalog.Projection{}, false
-	}
-	if e.deleted {
-		return catalog.Projection{}, false
-	}
-	return e.proj, true
+// HandleFrame feeds one catalog.projection_published event (snapshot frame or
+// delta) into the store; framing/epoch ordering live in platform/catalog.
+func (c *catalogProjectionCache) HandleFrame(meta map[string]string, payload []byte) error {
+	return c.store.HandleFrame(meta, payload)
 }
 
-// IsDeleted reports whether a tombstone exists for sku — used by HTTP-fetch
-// call sites to skip the fallback for an SKU the bus has explicitly deleted,
-// avoiding a wasteful round-trip on items the writer has just removed.
-func (c *catalogProjectionCache) IsDeleted(sku string) bool {
-	c.mu.RLock()
-	e, ok := c.items[sku]
-	c.mu.RUnlock()
-	return ok && e.deleted
-}
+func (c *catalogProjectionCache) Get(sku string) (catalog.Projection, bool) { return c.store.Get(sku) }
 
-// Len reports the number of map entries (bus-fed upserts + tombstones). Used
-// for log gauges only.
+// IsDeleted reports whether the bus has explicitly deleted sku (overlay
+// tombstone) — call sites skip the HTTP fallback for it.
+func (c *catalogProjectionCache) IsDeleted(sku string) bool { return c.store.IsDeleted(sku) }
+
+// Len reports cached entries (base + overlay) — debug/log gauge only.
 func (c *catalogProjectionCache) Len() int {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-	return len(c.items)
+	_, base, overlay := c.store.Stats()
+	return base + overlay
 }
 
 // projectionDeliveryHandler is the AMQP delivery handler for
@@ -226,13 +194,12 @@ func projectionDeliveryHandler(cache *catalogProjectionCache) func(amqp.Delivery
 		if err := json.Unmarshal(d.Body, &ev); err != nil {
 			return fmt.Errorf("decode event envelope: %w", err)
 		}
-		updates, err := catalog.DecodeUpdates(ev.Payload)
-		if err != nil {
-			return fmt.Errorf("decode projection updates: %w", err)
+		// Route the frame (snapshot begin/chunk/end or delta) into the store;
+		// epoch ordering, snapshot assembly + atomic swap, and tombstones all live
+		// in platform/catalog.ProjectionStore.
+		if err := cache.HandleFrame(ev.Meta, ev.Payload); err != nil {
+			return fmt.Errorf("apply projection frame: %w", err)
 		}
-		upserts, deletes := cache.Apply(updates)
-		log.Printf("cart: catalog projection batch applied: upserts=%d deletes=%d total=%d eventID=%s source=%q",
-			upserts, deletes, cache.Len(), ev.ID, ev.Source)
 		return nil
 	}
 }
@@ -423,7 +390,13 @@ func main() {
 	// connection keeps the cart's mutation-feed lifecycle decoupled from a
 	// read-write-consume connection that could grow event handlers later
 	// (e.g. promotions reacting to a price drop on a watched SKU).
-	projectionCache := newCatalogProjectionCache()
+	// Pod-local store dir — container fs / an emptyDir in k8s. NEVER a shared
+	// volume: each replica owns its own .bin (replicated read cache, no leader).
+	projectionDir := config.EnvString("CART_PROJECTION_DIR", "data/projection")
+	projectionCache, err := newCatalogProjectionCache(projectionDir)
+	if err != nil {
+		log.Fatalf("Error opening catalog projection cache: %v", err)
+	}
 	if amqpUrl != "" {
 		if pConn, perr := rabbit.Dial(amqpUrl, "cart-catalog-projection"); perr != nil {
 			log.Printf("cart: catalog projection consumer DISABLED (dial failed): %v", perr)

cmd/cart/projection_cache_test.go

@@ -10,7 +10,7 @@ import (
 // TestProjectionCache_ApplyUpserts covers the happy path: a batch of N
 // projections reaches the cache via Apply, all are visible via Get.
 func TestProjectionCache_ApplyUpserts(t *testing.T) {
-	c := newCatalogProjectionCache()
+	c := mustCache(t)
 	updates := []catalog.ProjectionUpdate{
 		{Projection: catalog.Projection{ID: "id-A", SKU: "A", PriceIncVat: money.Cents(100_00), TaxClass: "standard"}},
 		{Projection: catalog.Projection{ID: "id-B", SKU: "B", PriceIncVat: money.Cents(150_00), TaxClass: "reduced"}},
@@ -48,7 +48,7 @@ func TestProjectionCache_ApplyUpserts(t *testing.T) {
 //   - IsDeleted returns true
 //   - the same entry continues to count toward Len (size accounting)
 func TestProjectionCache_DeleteTombstone(t *testing.T) {
-	c := newCatalogProjectionCache()
+	c := mustCache(t)
 	c.Apply([]catalog.ProjectionUpdate{
 		{Projection: catalog.Projection{ID: "id-A", SKU: "A", PriceIncVat: money.Cents(100_00)}},
 	})
@@ -81,7 +81,7 @@ func TestProjectionCache_DeleteTombstone(t *testing.T) {
 // clears the tombstone (live entry replaces it): Get returns the projection,
 // IsDeleted flips to false. Mirrors how the live catalog keeps flipping state.
 func TestProjectionCache_BusResurrect(t *testing.T) {
-	c := newCatalogProjectionCache()
+	c := mustCache(t)
 	c.Apply([]catalog.ProjectionUpdate{
 		{Projection: catalog.Projection{SKU: "A", PriceIncVat: money.Cents(50_00)}},
 		{Projection: catalog.Projection{SKU: "A"}, Deleted: true},
@@ -110,7 +110,7 @@ func TestProjectionCache_BusResurrect(t *testing.T) {
 // SKU is a no-op (defensive — bus should never publish "" but cheaper to log
 // + drop than to populate an ambiguous cache slot).
 func TestProjectionCache_EmptySKUIgnored(t *testing.T) {
-	c := newCatalogProjectionCache()
+	c := mustCache(t)
 	upserts, _ := c.Apply([]catalog.ProjectionUpdate{
 		{Projection: catalog.Projection{SKU: "", PriceIncVat: money.Cents(100_00)}},
 	})
@@ -125,7 +125,7 @@ func TestProjectionCache_EmptySKUIgnored(t *testing.T) {
 // TestProjectionCache_BusRaceSafe runs Apply + Get concurrently under -race
 // to lock down the lock order. Run via `go test -race ./cmd/cart/...`.
 func TestProjectionCache_BusRaceSafe(t *testing.T) {
-	c := newCatalogProjectionCache()
+	c := mustCache(t)
 	done := make(chan struct{})
 	go func() {
 		defer close(done)
@@ -151,7 +151,7 @@ func TestProjectionCache_BusRaceSafe(t *testing.T) {
 // the per-batch counters report the right split. Without this the cache could
 // drift on count semantics across batched events.
 func TestApply_MixedUpsertDelete(t *testing.T) {
-	c := newCatalogProjectionCache()
+	c := mustCache(t)
 	// Order matters: prior in-cache state for A is upserted twice (latest wins),
 	// B is tombstoned, C cold-upserted, D cold-tombstoned.
 	updates := []catalog.ProjectionUpdate{

cmd/cart/projection_overlay_test.go

@@ -142,7 +142,7 @@ func TestApplyProjectionOverlay_NilMsgSafe(t *testing.T) {
 // bypass the tombstone branch and fall through to the HTTP fetch — failing
 // the test for a wiring reason rather than the contract under test.
 func TestAddSkuToCartHandler_TombstoneReturns404(t *testing.T) {
-	idx := newCatalogProjectionCache()
+	idx := mustCache(t)
 	idx.Apply([]catalog.ProjectionUpdate{
 		{Projection: catalog.Projection{SKU: "DEL", PriceIncVat: money.Cents(10_00)}},
 		{Projection: catalog.Projection{SKU: "DEL"}, Deleted: true},
@@ -181,7 +181,7 @@ func TestAddSkuToCartHandler_TombstoneReturns404(t *testing.T) {
 // TestIsDeletedVsGet asserts the two methods don't conflict: a tombstoned SKU
 // is reported by both IsDeleted (true) and Get (miss).
 func TestIsDeletedVsGet(t *testing.T) {
-	c := newCatalogProjectionCache()
+	c := mustCache(t)
 	c.Apply([]catalog.ProjectionUpdate{
 		{Projection: catalog.Projection{SKU: "DL", PriceIncVat: money.Cents(10_00)}},
 		{Projection: catalog.Projection{SKU: "DL"}, Deleted: true},
@@ -395,7 +395,7 @@ func TestBuildItemGroups_CacheOnlySkipsHTTP(t *testing.T) {
 	prod := newFakeProductServer(t)
 	t.Setenv("PRODUCT_BASE_URL", prod.srv.URL)
 
-	idx := newCatalogProjectionCache()
+	idx := mustCache(t)
 	idx.Apply([]catalog.ProjectionUpdate{
 		{Projection: catalog.Projection{
 			SKU: "TOP",
@@ -463,7 +463,7 @@ func TestBuildItemGroups_HTTPFallbackWhenChildrenPresent(t *testing.T) {
 	prod := newFakeProductServer(t)
 	t.Setenv("PRODUCT_BASE_URL", prod.srv.URL)
 
-	idx := newCatalogProjectionCache()
+	idx := mustCache(t)
 	idx.Apply([]catalog.ProjectionUpdate{
 		{Projection: catalog.Projection{
 			SKU: "PARENT", PriceIncVat: money.Cents(99_00),
@@ -518,7 +518,7 @@ func TestAddSkuToCartHandler_CacheOnlySkipsHTTP(t *testing.T) {
 	defer failSrv.Close()
 	t.Setenv("PRODUCT_BASE_URL", failSrv.URL)
 
-	idx := newCatalogProjectionCache()
+	idx := mustCache(t)
 	idx.Apply([]catalog.ProjectionUpdate{
 		{Projection: catalog.Projection{
 			SKU: "OK", PriceIncVat: money.Cents(99_00),
@@ -550,7 +550,7 @@ func TestBuildItemGroups_HTTPFallbackWhenCacheMiss(t *testing.T) {
 	prod := newFakeProductServer(t)
 	t.Setenv("PRODUCT_BASE_URL", prod.srv.URL)
 
-	idx := newCatalogProjectionCache() // empty
+	idx := mustCache(t) // empty
 	groups := buildItemGroups(context.Background(), []Item{
 		{Sku: "MISS", Quantity: 1},
 	}, "se", idx)
@@ -575,7 +575,7 @@ func TestBuildItemGroups_HTTPFallbackWhenHasRequiredFieldsFalse(t *testing.T) {
 	prod := newFakeProductServer(t)
 	t.Setenv("PRODUCT_BASE_URL", prod.srv.URL)
 
-	idx := newCatalogProjectionCache()
+	idx := mustCache(t)
 	idx.Apply([]catalog.ProjectionUpdate{
 		{Projection: catalog.Projection{
 			SKU: "PARTIAL", PriceIncVat: money.Cents(99_00),

cmd/cart/projection_store_integration_test.go

@@ -0,0 +1,72 @@
+package main
+
+import (
+	"strconv"
+	"testing"
+
+	"git.k6n.net/mats/platform/catalog"
+	"git.k6n.net/mats/platform/money"
+)
+
+// frameMeta builds the event.Meta a producer stamps for a snapshot frame / delta.
+func frameMeta(phase string, epoch int64) map[string]string {
+	m := map[string]string{catalog.MetaEpoch: strconv.FormatInt(epoch, 10), catalog.MetaTenant: "se"}
+	if phase != "" {
+		m[catalog.MetaSnapshotPhase] = phase
+	}
+	return m
+}
+
+func enc(t *testing.T, ups []catalog.ProjectionUpdate) []byte {
+	t.Helper()
+	b, err := catalog.EncodeUpdates(ups)
+	if err != nil {
+		t.Fatalf("encode: %v", err)
+	}
+	return b
+}
+
+// TestCartCache_SnapshotThenDelta exercises the cart's cache through the real bus
+// framing: a full begin→chunk→end snapshot makes SKUs Get-able cold (no Apply),
+// then an epoch-matched delta overrides one and tombstones another.
+func TestCartCache_SnapshotThenDelta(t *testing.T) {
+	c := mustCache(t)
+
+	const epoch = int64(1000)
+	if err := c.HandleFrame(frameMeta(catalog.SnapshotBegin, epoch), enc(t, nil)); err != nil {
+		t.Fatal(err)
+	}
+	if err := c.HandleFrame(frameMeta(catalog.SnapshotChunk, epoch), enc(t, []catalog.ProjectionUpdate{
+		{Projection: catalog.Projection{ID: "sku:A", SKU: "A", PriceIncVat: money.Cents(100_00), TaxClass: "standard", ItemID: 1}},
+		{Projection: catalog.Projection{ID: "sku:B", SKU: "B", PriceIncVat: money.Cents(50_00), TaxClass: "reduced", ItemID: 2}},
+	})); err != nil {
+		t.Fatal(err)
+	}
+	// end carries the total count for the completeness guard.
+	endMeta := frameMeta(catalog.SnapshotEnd, epoch)
+	endMeta[catalog.MetaCount] = "2"
+	if err := c.HandleFrame(endMeta, enc(t, nil)); err != nil {
+		t.Fatal(err)
+	}
+
+	if got, ok := c.Get("A"); !ok || got.PriceIncVat != 100_00 {
+		t.Fatalf("A from snapshot = %+v ok=%v, want 10000", got, ok)
+	}
+
+	// Epoch-matched delta: bump A's price, delete B.
+	if err := c.HandleFrame(frameMeta("", epoch), enc(t, []catalog.ProjectionUpdate{
+		{Projection: catalog.Projection{SKU: "A", PriceIncVat: money.Cents(90_00), TaxClass: "standard", ItemID: 1}},
+		{Projection: catalog.Projection{SKU: "B"}, Deleted: true},
+	})); err != nil {
+		t.Fatal(err)
+	}
+	if got, _ := c.Get("A"); got.PriceIncVat != 90_00 {
+		t.Fatalf("A after delta = %d, want 9000", got.PriceIncVat)
+	}
+	if _, ok := c.Get("B"); ok {
+		t.Fatalf("B after delete should miss")
+	}
+	if !c.IsDeleted("B") {
+		t.Fatalf("B should be tombstoned (skip HTTP fetch)")
+	}
+}

cmd/cart/projection_testhelpers_test.go

@@ -0,0 +1,39 @@
+package main
+
+import (
+	"testing"
+
+	"git.k6n.net/mats/platform/catalog"
+)
+
+// mustCache opens a per-pod projection cache backed by a temp dir (pod-local).
+func mustCache(t *testing.T) *catalogProjectionCache {
+	t.Helper()
+	c, err := newCatalogProjectionCache(t.TempDir())
+	if err != nil {
+		t.Fatalf("newCatalogProjectionCache: %v", err)
+	}
+	t.Cleanup(func() { _ = c.store.Close() })
+	return c
+}
+
+// Apply is a TEST-ONLY convenience that routes a delta batch through the store at
+// the initial epoch (0), reproducing the pre-ProjectionStore cache API so the
+// existing cart tests keep exercising Get/IsDeleted/Len. Production code feeds the
+// store via HandleFrame with real bus framing (snapshot + epoch-stamped deltas).
+func (c *catalogProjectionCache) Apply(updates []catalog.ProjectionUpdate) (upserts, deletes int) {
+	for _, u := range updates {
+		if u.Deleted {
+			deletes++
+		} else if u.SKU != "" {
+			upserts++
+		}
+	}
+	payload, err := catalog.EncodeUpdates(updates)
+	if err != nil {
+		return 0, 0
+	}
+	// No MetaSnapshotPhase ⇒ delta; epoch 0 matches a fresh store's base epoch.
+	_ = c.store.HandleFrame(map[string]string{catalog.MetaEpoch: "0"}, payload)
+	return upserts, deletes
+}