platform/catalog: generic ProjectionStore[T] consumer + race-free holder; finish C1
- Generic SnapshotWriter[T]/SnapshotReader[T] over a value-agnostic raw layer; SnapshotHolder[T] now RWMutex-guards munmap vs in-flight Get (race-free swap, TestSnapshotHolder_SwapRace) — the drain grace is now just close-batching. - ProjectionStore[T]: snapshot begin|chunk|end assembly → atomic swap, epoch-gated delta overlay, tombstones, atomic epoch sidecar (ordered before the swap), cold-start reopen, mandatory ProjectionTransform[T]. Future-epoch deltas are dropped by contract (recover via next snapshot), not buffered. - Shared bus-framing constants so producer + consumer agree. - Bump slask-finder / go-cart-actor / backoffice pointers; refresh status.md + catalog-projection.md to the finished state. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
matsandClaude Opus 4.8 committed
1 parent
5a3ffcb6bc
commit
8b454bccac
8 files changed
+959
-170
No files matched your search
@@ -0,0 +1,323 @@
|
||||
package catalog
|
||||
|
||||
// ProjectionStore — the consumer half of the projection seam (see
|
||||
// docs/catalog-projection.md). A commerce service (cart, checkout, order, UCP)
|
||||
// feeds it the catalog.projection_published frames it receives off the bus and
|
||||
// reads catalog facts back by SKU. The store owns a LOCAL read-only projection:
|
||||
//
|
||||
// - A full snapshot stream (begin → chunk* → end, one MetaEpoch) is assembled
|
||||
// into the store's OWN .bin (SnapshotWriter), atomically swapped in as the
|
||||
// mmap base (SnapshotHolder). The base is the cold-start substrate — it
|
||||
// survives a restart (reopened from disk), so the consumer serves reads
|
||||
// immediately without waiting for a bus replay.
|
||||
// - Per-batch deltas of the SAME epoch are applied to an in-memory overlay on
|
||||
// top of the base (overlay wins; a delete leaves a tombstone that shadows the
|
||||
// base and the fallback). The next snapshot folds the overlay into a new base.
|
||||
//
|
||||
// Read order on Get: overlay → mmap base → optional caller fallback (e.g. an
|
||||
// HTTP fetch from the reader) — last-resort only. The access pattern is point
|
||||
// lookup by SKU; nothing enumerates the catalog.
|
||||
//
|
||||
// No shared volume: each consumer builds its own base from the bus. Concurrency:
|
||||
// HandleFrame is single-writer (drive it from one delivery goroutine); Get is
|
||||
// safe from many readers.
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// ProjectionFallback resolves an SKU the local store doesn't have yet (cold
|
||||
// start before the first snapshot, or an item added since the last base). It
|
||||
// returns a raw Projection — the store applies the SAME Transform to it, so a
|
||||
// fallback hit is shaped identically to bus-fed data. Return (_, false) to
|
||||
// report a genuine miss.
|
||||
type ProjectionFallback func(sku string) (Projection, bool)
|
||||
|
||||
// ProjectionTransform projects the wire Projection down to the consumer's own
|
||||
// stored type T. It is MANDATORY: it is the one place a consumer declares which
|
||||
// catalog facts it actually needs (and any derivation, e.g. tax_class → rate),
|
||||
// so its local .bin holds only that — smaller, and read-ready with no per-Get
|
||||
// conversion. Every value entering the store — snapshot, delta, and fallback —
|
||||
// passes through it exactly once.
|
||||
type ProjectionTransform[T any] func(Projection) T
|
||||
|
||||
// ProjectionStore is a consumer's local, cold-start-ready projection cache of T.
|
||||
type ProjectionStore[T any] struct {
|
||||
dir, name string
|
||||
transform ProjectionTransform[T]
|
||||
fallback ProjectionFallback
|
||||
holder *SnapshotHolder[T] // mmap base; atomic-swap + drain-grace
|
||||
|
||||
mu sync.RWMutex
|
||||
epoch int64 // epoch of the live base
|
||||
overlay map[string]overlayEntry[T] // same-epoch deltas layered on the base
|
||||
|
||||
// in-progress snapshot assembly (one epoch at a time)
|
||||
building *SnapshotWriter[T]
|
||||
buildingEpoch int64
|
||||
buildingSeen int
|
||||
}
|
||||
|
||||
type overlayEntry[T any] struct {
|
||||
val T
|
||||
deleted bool
|
||||
}
|
||||
|
||||
// NewProjectionStore opens a store of T backed by dir/name (its own .bin).
|
||||
// transform is REQUIRED — it defines the consumer's projection shape (see
|
||||
// ProjectionTransform). If a base from a previous run is present it is reopened
|
||||
// immediately (cold-start ready); its epoch is read from the sidecar so
|
||||
// same-epoch deltas can resume. fallback may be nil. Close() when done.
|
||||
func NewProjectionStore[T any](dir, name string, transform ProjectionTransform[T], fallback ProjectionFallback) (*ProjectionStore[T], error) {
|
||||
if dir == "" || name == "" {
|
||||
return nil, fmt.Errorf("catalog: ProjectionStore needs dir and name")
|
||||
}
|
||||
if transform == nil {
|
||||
return nil, fmt.Errorf("catalog: ProjectionStore needs a transform (declare which catalog facts this consumer stores)")
|
||||
}
|
||||
s := &ProjectionStore[T]{
|
||||
dir: dir,
|
||||
name: name,
|
||||
transform: transform,
|
||||
fallback: fallback,
|
||||
holder: NewSnapshotHolder[T](0),
|
||||
overlay: make(map[string]overlayEntry[T]),
|
||||
}
|
||||
path := filepath.Join(dir, name)
|
||||
if err := s.holder.Open(path); err == nil {
|
||||
s.epoch = readEpochSidecar(path)
|
||||
log.Printf("catalog: ProjectionStore reopened base %s (epoch=%d, %d entries)", path, s.epoch, s.holder.Len())
|
||||
} else if !os.IsNotExist(err) {
|
||||
// A corrupt/partial base on disk shouldn't crash boot — start empty and
|
||||
// wait for the next snapshot; reads fall back meanwhile.
|
||||
log.Printf("catalog: ProjectionStore ignoring unreadable base %s: %v", path, err)
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
// HandleFrame applies one catalog.projection_published event (snapshot frame or
|
||||
// delta), routed by its Meta. Drive this from a single delivery goroutine. A
|
||||
// returned error means the payload was undecodable (the caller may NACK); a
|
||||
// frame that simply doesn't apply (stale epoch, wrong assembly state) returns
|
||||
// nil so the delivery is acked.
|
||||
func (s *ProjectionStore[T]) HandleFrame(meta map[string]string, payload []byte) error {
|
||||
epoch, _ := strconv.ParseInt(meta[MetaEpoch], 10, 64)
|
||||
switch meta[MetaSnapshotPhase] {
|
||||
case SnapshotBegin:
|
||||
return s.beginSnapshot(epoch, meta[MetaTenant])
|
||||
case SnapshotChunk:
|
||||
return s.chunkSnapshot(epoch, payload)
|
||||
case SnapshotEnd:
|
||||
return s.endSnapshot(epoch, meta)
|
||||
case "":
|
||||
return s.applyDelta(epoch, payload)
|
||||
default:
|
||||
return nil // unknown phase — ignore forward-compatibly
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ProjectionStore[T]) beginSnapshot(epoch int64, tenant string) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if epoch <= s.epoch {
|
||||
return nil // we already hold this base or newer
|
||||
}
|
||||
// Start (or restart) assembly for this epoch. A new begin supersedes any
|
||||
// half-built snapshot (e.g. a producer that aborted mid-stream).
|
||||
s.building = NewSnapshotWriter[T]().WithTenant(tenant)
|
||||
s.buildingEpoch = epoch
|
||||
s.buildingSeen = 0
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *ProjectionStore[T]) chunkSnapshot(epoch int64, payload []byte) error {
|
||||
ups, err := DecodeUpdates(payload)
|
||||
if err != nil {
|
||||
return fmt.Errorf("catalog: decode snapshot chunk: %w", err)
|
||||
}
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.building == nil || epoch != s.buildingEpoch {
|
||||
return nil // not assembling this epoch
|
||||
}
|
||||
for _, u := range ups {
|
||||
if u.Deleted || u.SKU == "" {
|
||||
continue // a snapshot carries live items only
|
||||
}
|
||||
// Transform applied here — the base stores the consumer's shape, not the
|
||||
// full Projection.
|
||||
if err := s.building.Append(u.SKU, s.transform(u.Projection)); err != nil {
|
||||
log.Printf("catalog: ProjectionStore skip %q in snapshot: %v", u.SKU, err)
|
||||
continue
|
||||
}
|
||||
s.buildingSeen++
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *ProjectionStore[T]) endSnapshot(epoch int64, meta map[string]string) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.building == nil || epoch != s.buildingEpoch {
|
||||
return nil
|
||||
}
|
||||
w := s.building
|
||||
s.building = nil
|
||||
|
||||
// Completeness guard: if begin/end advertised a count and we appended fewer
|
||||
// (a lost chunk), DON'T swap — keep the old base and wait for the next
|
||||
// snapshot. Serving an incomplete catalog silently is worse than stale.
|
||||
if cs := meta[MetaCount]; cs != "" {
|
||||
if want, perr := strconv.Atoi(cs); perr == nil && want != s.buildingSeen {
|
||||
log.Printf("catalog: ProjectionStore snapshot epoch=%d incomplete (%d/%d) — discarding, keeping base epoch=%d",
|
||||
epoch, s.buildingSeen, want, s.epoch)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
path, err := w.Write(s.dir, s.name)
|
||||
if err != nil {
|
||||
return fmt.Errorf("catalog: write snapshot base: %w", err)
|
||||
}
|
||||
// Persist the epoch sidecar (atomic) BEFORE swapping the in-memory reader, so
|
||||
// a reopened store's epoch always matches the .bin already on disk (w.Write
|
||||
// renamed the new file into place above). Writing it after Reload would leave
|
||||
// a crash window where the new .bin carries a stale/absent epoch — the store
|
||||
// would then mislabel its base and drop the current epoch's deltas on restart.
|
||||
writeEpochSidecar(path, epoch)
|
||||
if err := s.holder.Reload(path); err != nil {
|
||||
return fmt.Errorf("catalog: reload snapshot base: %w", err)
|
||||
}
|
||||
s.epoch = epoch
|
||||
s.overlay = make(map[string]overlayEntry[T]) // base now includes everything ≤ epoch
|
||||
log.Printf("catalog: ProjectionStore base swapped epoch=%d (%d entries)", epoch, s.buildingSeen)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *ProjectionStore[T]) applyDelta(epoch int64, payload []byte) error {
|
||||
ups, err := DecodeUpdates(payload)
|
||||
if err != nil {
|
||||
return fmt.Errorf("catalog: decode delta: %w", err)
|
||||
}
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
switch {
|
||||
case s.building != nil:
|
||||
// Under FIFO delivery a delta won't interleave a snapshot stream; if it
|
||||
// does, the snapshot is authoritative for its epoch, so drop the delta.
|
||||
return nil
|
||||
case epoch < s.epoch:
|
||||
return nil // stale — predates our base
|
||||
case epoch > s.epoch:
|
||||
// CONTRACT: a future-epoch delta is DROPPED, not buffered. Under FIFO
|
||||
// per-queue delivery begin(N) always precedes delta(N), so this only
|
||||
// happens after a reconnect that missed the whole snapshot — which a
|
||||
// buffer couldn't reassemble anyway. Recovery is the next snapshot of
|
||||
// this epoch; the consumer's fallback covers reads until then.
|
||||
return nil
|
||||
}
|
||||
for _, u := range ups {
|
||||
if u.SKU == "" {
|
||||
continue
|
||||
}
|
||||
if u.Deleted {
|
||||
s.overlay[u.SKU] = overlayEntry[T]{deleted: true}
|
||||
continue
|
||||
}
|
||||
// Transform applied here too — overlay holds the consumer's shape.
|
||||
s.overlay[u.SKU] = overlayEntry[T]{val: s.transform(u.Projection)}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Get resolves a value by SKU: overlay (incl. tombstone) → mmap base →
|
||||
// fallback. A tombstoned SKU returns (zero, false) and never reaches the
|
||||
// fallback — the catalog explicitly removed it. A fallback hit is run through
|
||||
// the same Transform, so it's shaped like everything else.
|
||||
func (s *ProjectionStore[T]) Get(sku string) (T, bool) {
|
||||
var zero T
|
||||
s.mu.RLock()
|
||||
e, ok := s.overlay[sku]
|
||||
s.mu.RUnlock()
|
||||
if ok {
|
||||
if e.deleted {
|
||||
return zero, false
|
||||
}
|
||||
return e.val, true
|
||||
}
|
||||
if v, ok := s.holder.Get(sku); ok {
|
||||
return v, true
|
||||
}
|
||||
if s.fallback != nil {
|
||||
if p, ok := s.fallback(sku); ok {
|
||||
return s.transform(p), true
|
||||
}
|
||||
}
|
||||
return zero, false
|
||||
}
|
||||
|
||||
// IsDeleted reports whether the overlay holds a tombstone for sku — a delete
|
||||
// seen on the bus since the current base. Lets a caller skip a downstream fetch
|
||||
// for an SKU the catalog just removed. After a snapshot folds the delete into a
|
||||
// new base the SKU is simply absent (IsDeleted → false); "absent" is then the
|
||||
// caller's fetch/miss path, which a deleted item resolves as a downstream 404.
|
||||
func (s *ProjectionStore[T]) IsDeleted(sku string) bool {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
e, ok := s.overlay[sku]
|
||||
return ok && e.deleted
|
||||
}
|
||||
|
||||
// Epoch returns the epoch of the current base (0 if none yet).
|
||||
func (s *ProjectionStore[T]) Epoch() int64 {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
return s.epoch
|
||||
}
|
||||
|
||||
// Stats returns observability counters (base size, overlay size, epoch).
|
||||
func (s *ProjectionStore[T]) Stats() (epoch int64, baseCount, overlayCount int) {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
return s.epoch, s.holder.Len(), len(s.overlay)
|
||||
}
|
||||
|
||||
// Close releases the mmap base.
|
||||
func (s *ProjectionStore[T]) Close() error { return s.holder.Close() }
|
||||
|
||||
// ── epoch sidecar ────────────────────────────────────────────────────────────
|
||||
// The .bin header has no room for the epoch, so it's persisted next to it as
|
||||
// "<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 — a reader either sees the
|
||||
// old epoch or the new one, never garbage.
|
||||
func writeEpochSidecar(binPath string, epoch int64) {
|
||||
p := epochSidecarPath(binPath)
|
||||
tmp := p + ".tmp"
|
||||
if err := os.WriteFile(tmp, []byte(strconv.FormatInt(epoch, 10)), 0o644); err != nil {
|
||||
log.Printf("catalog: ProjectionStore epoch sidecar write: %v", err)
|
||||
return
|
||||
}
|
||||
if err := os.Rename(tmp, p); err != nil {
|
||||
log.Printf("catalog: ProjectionStore epoch sidecar rename: %v", err)
|
||||
_ = os.Remove(tmp)
|
||||
}
|
||||
}
|
||||
|
||||
func readEpochSidecar(binPath string) int64 {
|
||||
b, err := os.ReadFile(epochSidecarPath(binPath))
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
n, _ := strconv.ParseInt(strings.TrimSpace(string(b)), 10, 64)
|
||||
return n
|
||||
}
|
||||
Reference in new issue
Block a user