From d711348863b6b69a5d001e57d773289484885387 Mon Sep 17 00:00:00 2001 From: matst80 Date: Sun, 28 Jun 2026 21:05:55 +0200 Subject: [PATCH] cart: back projection cache with platform/catalog.ProjectionStore MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- cmd/cart/main.go | 133 +++++++----------- cmd/cart/projection_cache_test.go | 12 +- cmd/cart/projection_overlay_test.go | 14 +- cmd/cart/projection_store_integration_test.go | 72 ++++++++++ cmd/cart/projection_testhelpers_test.go | 39 +++++ 5 files changed, 177 insertions(+), 93 deletions(-) create mode 100644 cmd/cart/projection_store_integration_test.go create mode 100644 cmd/cart/projection_testhelpers_test.go diff --git a/cmd/cart/main.go b/cmd/cart/main.go index a8ae601..64feba5 100644 --- a/cmd/cart/main.go +++ b/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 - } - if u.SKU == "" { - continue - } - c.items[u.SKU] = projectionEntry{proj: u.Projection} - upserts++ +// 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 } - 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) diff --git a/cmd/cart/projection_cache_test.go b/cmd/cart/projection_cache_test.go index 0da6925..aaf939e 100644 --- a/cmd/cart/projection_cache_test.go +++ b/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{ diff --git a/cmd/cart/projection_overlay_test.go b/cmd/cart/projection_overlay_test.go index 2b6f2e2..ba16417 100644 --- a/cmd/cart/projection_overlay_test.go +++ b/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), diff --git a/cmd/cart/projection_store_integration_test.go b/cmd/cart/projection_store_integration_test.go new file mode 100644 index 0000000..2270b6e --- /dev/null +++ b/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)") + } +} diff --git a/cmd/cart/projection_testhelpers_test.go b/cmd/cart/projection_testhelpers_test.go new file mode 100644 index 0000000..f1bf191 --- /dev/null +++ b/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 +}