diff --git a/projection/emitter.go b/projection/emitter.go new file mode 100644 index 0000000..604781b --- /dev/null +++ b/projection/emitter.go @@ -0,0 +1,154 @@ +package projection + +import ( + "errors" + "strconv" +) + +// Emitter is the producer half of the seam: it frames a full projection as a +// begin → chunk* → end snapshot stream (and per-batch deltas) and hands each +// frame to a caller-supplied publish function. It is transport-agnostic — it +// owns only the framing (the Meta* keys, chunking, epoch tagging); the caller +// wires PublishFunc to its bus (rabbit.PublishEvent, etc.) and EncodeFunc to +// its wire schema. This is the symmetric counterpart to Store/Decoder. +// +// One authoritative producer per projection: receivers' Stores assemble their +// own .bin from this stream. Two producers publishing the same projection would +// make receivers thrash between epochs — don't. +// +// catalog's finder writeradmin predates this and keeps its own framing for now; +// new domains (tracking popularity/suggestions) use this. + +// PublishFunc publishes one framed event: meta carries the framing keys +// (snapshot phase, epoch, seq, count, tenant) already set; payload is the +// chunk/delta body. The caller may add its own keys to meta before sending +// (e.g. a country tag). A non-nil error aborts the current snapshot stream. +type PublishFunc func(meta map[string]string, payload []byte) error + +// EncodeFunc serialises a batch of wire values into a chunk/delta payload — the +// inverse of Decoder, and the one place the domain wire schema lives on the +// producer side. It MUST encode a nil/empty slice as a payload the matching +// Decoder reads back as zero updates (begin/end frames carry no entries). +// JSONEncode satisfies this for the generic Update[W] schema. +type EncodeFunc[W any] func([]Update[W]) ([]byte, error) + +// ErrEmptySnapshot is returned by Snapshot when handed no entries. An empty +// snapshot would swap a receiver's base to nothing; refusing it mirrors the +// catalog producer, which never propagates an empty catalog. +var ErrEmptySnapshot = errors.New("projection: refusing to emit an empty snapshot (would wipe receivers)") + +// DefaultChunkSize is the entries-per-chunk used when WithChunkSize isn't set. +// Mirrors catalog's projectionSnapshotChunk. +const DefaultChunkSize = 500 + +// Emitter frames and publishes a projection of wire type W. +type Emitter[W any] struct { + encode EncodeFunc[W] + publish PublishFunc + tenant string + chunkSize int +} + +// emitterConfig is set via options then copied onto the typed Emitter (options +// can't be generic over W, so they carry the un-typed knobs). +type emitterConfig struct { + tenant string + chunkSize int +} + +// WithTenant tags every frame with a tenant namespace (lands in MetaTenant and, +// for snapshots, in the receiver's base header). +func WithTenant(tenant string) func(*emitterConfig) { + return func(c *emitterConfig) { c.tenant = tenant } +} + +// WithChunkSize overrides the entries-per-chunk (default DefaultChunkSize). +// Bounds per-message size; smaller = more frames. +func WithChunkSize(n int) func(*emitterConfig) { + return func(c *emitterConfig) { + if n > 0 { + c.chunkSize = n + } + } +} + +// NewEmitter builds an emitter. encode and publish are REQUIRED. Options: +// WithTenant, WithChunkSize. +func NewEmitter[W any](encode EncodeFunc[W], publish PublishFunc, opts ...func(*emitterConfig)) *Emitter[W] { + cfg := emitterConfig{chunkSize: DefaultChunkSize} + for _, o := range opts { + o(&cfg) + } + return &Emitter[W]{ + encode: encode, + publish: publish, + tenant: cfg.tenant, + chunkSize: cfg.chunkSize, + } +} + +// Snapshot publishes the full entry set as a begin → chunk* → end stream tagged +// with epoch (a monotonic Unix-ms stamp the caller provides). Returns the +// number of chunk frames emitted. ErrEmptySnapshot if items is empty; a +// publish/encode error aborts the stream (receivers keep their previous base) +// and is returned. The same epoch must later tag any Delta built on this base. +func (e *Emitter[W]) Snapshot(epoch int64, items []Update[W]) (chunks int, err error) { + if len(items) == 0 { + return 0, ErrEmptySnapshot + } + if err := e.frame(epoch, SnapshotBegin, 0, len(items), nil); err != nil { + return 0, err + } + seq := 0 + for start := 0; start < len(items); start += e.chunkSize { + end := start + e.chunkSize + if end > len(items) { + end = len(items) + } + if err := e.frame(epoch, SnapshotChunk, seq, 0, items[start:end]); err != nil { + return seq, err + } + seq++ + } + if err := e.frame(epoch, SnapshotEnd, seq, len(items), nil); err != nil { + return seq, err + } + return seq, nil +} + +// Delta publishes one delta batch tagged with the base epoch it builds on. +// Entries with Deleted=true tombstone the key in receivers' overlays. A +// receiver ignores a delta whose epoch predates the base it holds. Empty batch +// is a no-op (no error). +func (e *Emitter[W]) Delta(epoch int64, items []Update[W]) error { + if len(items) == 0 { + return nil + } + return e.frame(epoch, "", 0, 0, items) +} + +// frame encodes items and publishes one frame. phase "" ⇒ delta (no snapshot +// phase key). begin/end carry MetaCount; chunks/deltas carry the entries. +func (e *Emitter[W]) frame(epoch int64, phase string, seq, count int, items []Update[W]) error { + // Always encode — even begin/end with no entries. encode(nil) must yield a + // payload the Decoder reads as zero updates (see EncodeFunc), so a + // framing-unaware decode never sees an empty/undecodable body. + payload, err := e.encode(items) + if err != nil { + return err + } + meta := map[string]string{ + MetaEpoch: strconv.FormatInt(epoch, 10), + } + if e.tenant != "" { + meta[MetaTenant] = e.tenant + } + if phase != "" { + meta[MetaSnapshotPhase] = phase // presence marks a snapshot frame vs a delta + meta[MetaSeq] = strconv.Itoa(seq) + } + if phase == SnapshotBegin || phase == SnapshotEnd { + meta[MetaCount] = strconv.Itoa(count) + } + return e.publish(meta, payload) +} diff --git a/projection/emitter_test.go b/projection/emitter_test.go new file mode 100644 index 0000000..df25be3 --- /dev/null +++ b/projection/emitter_test.go @@ -0,0 +1,187 @@ +package projection + +import ( + "errors" + "fmt" + "sort" + "testing" +) + +// busToStore wires an Emitter's frames straight into a Store via HandleFrame — +// the in-process equivalent of publishing to the bus and a receiver consuming. +// Returns a PublishFunc the emitter calls per frame. +func busToStore[W, T any](t *testing.T, s *Store[W, T]) PublishFunc { + t.Helper() + return func(meta map[string]string, payload []byte) error { + return s.HandleFrame(meta, payload) + } +} + +// newIdentityStore builds a Store where wire == stored (W == T == tProj), using +// the generic JSON codec. opts forwards e.g. WithKeyHashing. +func newIdentityStore(t *testing.T, opts ...StoreOption) *Store[tProj, tProj] { + t.Helper() + s, err := NewStore[tProj, tProj](t.TempDir(), "emit.bin", + JSONDecode[tProj], func(p tProj) tProj { return p }, nil, opts...) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = s.Close() }) + return s +} + +func ups(items ...tProj) []Update[tProj] { + out := make([]Update[tProj], len(items)) + for i, it := range items { + out[i] = Update[tProj]{Key: it.SKU, Value: it} + } + return out +} + +// TestEmitter_SnapshotRoundTripsIntoStore — a full snapshot emitted frame-by-frame +// reassembles in the receiver's Store, served by Get. +func TestEmitter_SnapshotRoundTripsIntoStore(t *testing.T) { + s := newIdentityStore(t) + e := NewEmitter[tProj](JSONEncode[tProj], busToStore(t, s), WithTenant("se"), WithChunkSize(2)) + + items := ups( + makeProjection("A", 100), makeProjection("B", 200), + makeProjection("C", 300), makeProjection("D", 400), makeProjection("E", 500), + ) + chunks, err := e.Snapshot(1000, items) + if err != nil { + t.Fatalf("Snapshot: %v", err) + } + if chunks != 3 { // 5 items / chunkSize 2 = ceil = 3 + t.Errorf("chunks = %d, want 3", chunks) + } + if ep, base, _ := s.Stats(); ep != 1000 || base != 5 { + t.Fatalf("stats = epoch %d base %d, want 1000/5", ep, base) + } + for _, k := range []string{"A", "C", "E"} { + if _, ok := s.Get(k); !ok { + t.Errorf("Get(%q) miss after snapshot", k) + } + } +} + +// TestEmitter_DeltaAfterSnapshot — a same-epoch delta lands in the overlay and +// wins over the base; a tombstone hides the key. +func TestEmitter_DeltaAfterSnapshot(t *testing.T) { + s := newIdentityStore(t) + e := NewEmitter[tProj](JSONEncode[tProj], busToStore(t, s)) + + if _, err := e.Snapshot(10, ups(makeProjection("A", 100), makeProjection("B", 200))); err != nil { + t.Fatal(err) + } + // upsert A to a new price + delete B, same epoch. + if err := e.Delta(10, []Update[tProj]{ + {Key: "A", Value: makeProjection("A", 999)}, + {Key: "B", Deleted: true}, + }); err != nil { + t.Fatal(err) + } + if got, _ := s.Get("A"); got.PriceIncVat != 999 { + t.Errorf("A = %d, want overlay 999", got.PriceIncVat) + } + if _, ok := s.Get("B"); ok { + t.Error("B should be tombstoned") + } + if !s.IsDeleted("B") { + t.Error("IsDeleted(B) should be true") + } +} + +// TestEmitter_EmptySnapshotRefused — emitting nothing must not wipe receivers. +func TestEmitter_EmptySnapshotRefused(t *testing.T) { + var published int + e := NewEmitter[tProj](JSONEncode[tProj], func(map[string]string, []byte) error { + published++ + return nil + }) + if _, err := e.Snapshot(1, nil); !errors.Is(err, ErrEmptySnapshot) { + t.Fatalf("empty Snapshot err = %v, want ErrEmptySnapshot", err) + } + if published != 0 { + t.Errorf("published %d frames for an empty snapshot, want 0", published) + } +} + +// TestEmitter_PublishErrorAbortsStream — a publish failure mid-stream stops the +// emitter and the receiver keeps its previous base. +func TestEmitter_PublishErrorAbortsStream(t *testing.T) { + s := newIdentityStore(t) + // Seed a good base at epoch 5. + good := NewEmitter[tProj](JSONEncode[tProj], busToStore(t, s)) + if _, err := good.Snapshot(5, ups(makeProjection("OLD", 1))); err != nil { + t.Fatal(err) + } + + // New emitter whose publish fails on the first chunk of the next snapshot. + calls := 0 + flaky := NewEmitter[tProj](JSONEncode[tProj], func(meta map[string]string, payload []byte) error { + calls++ + if meta[MetaSnapshotPhase] == SnapshotChunk { + return fmt.Errorf("boom") + } + return s.HandleFrame(meta, payload) + }) + if _, err := flaky.Snapshot(9, ups(makeProjection("NEW", 2))); err == nil { + t.Fatal("expected Snapshot to surface the publish error") + } + // Base stays at the old epoch; OLD still served, NEW never applied. + if ep := s.Epoch(); ep != 5 { + t.Errorf("epoch = %d, want 5 (aborted snapshot didn't swap)", ep) + } + if _, ok := s.Get("OLD"); !ok { + t.Error("OLD should survive the aborted rebuild") + } + if _, ok := s.Get("NEW"); ok { + t.Error("NEW must not be visible from an aborted snapshot") + } +} + +// TestEmitter_HashedKeysEndToEnd — long keys flow producer→consumer in hashed +// mode: Get by original key works, and Scan yields digests with the original +// recoverable from the value. Ranked top-N (the popularity use case) works. +func TestEmitter_HashedKeysEndToEnd(t *testing.T) { + s := newIdentityStore(t, WithKeyHashing()) + e := NewEmitter[tProj](JSONEncode[tProj], busToStore(t, s)) + + // Keys that blow past 32 bytes (query strings), values carry the original + + // a score in PriceIncVat to rank by. + mk := func(q string, score int64) tProj { return tProj{SKU: q, PriceIncVat: score} } + q1 := "the quick brown fox jumps over the lazy dog (query one)" + q2 := "another unbounded length search query that a user typed" + q3 := "short-ish but still over thirty-two bytes for sure yes" + if _, err := e.Snapshot(1, ups(mk(q1, 30), mk(q2, 10), mk(q3, 20))); err != nil { + t.Fatal(err) + } + + // Point lookup by the ORIGINAL key. + if got, ok := s.Get(q1); !ok || got.PriceIncVat != 30 { + t.Fatalf("Get(q1) = %+v ok=%v, want score 30", got, ok) + } + + // Ranked enumeration: collect via Scan (keys are digests), sort by score, + // recover the original query from the value. Top-N popularity in a nutshell. + type row struct { + query string + score int64 + } + var rows []row + s.Scan(func(key string, v tProj) bool { + if len(key) != 32 { + t.Errorf("hashed Scan key len = %d, want 32", len(key)) + } + rows = append(rows, row{v.SKU, v.PriceIncVat}) + return true + }) + if len(rows) != 3 { + t.Fatalf("scanned %d rows, want 3", len(rows)) + } + sort.Slice(rows, func(i, j int) bool { return rows[i].score > rows[j].score }) + if rows[0].query != q1 || rows[1].query != q3 || rows[2].query != q2 { + t.Fatalf("ranked order wrong: %+v", rows) + } +} diff --git a/projection/framing.go b/projection/framing.go index c5d11f4..5392d95 100644 --- a/projection/framing.go +++ b/projection/framing.go @@ -1,5 +1,7 @@ package projection +import "encoding/json" + // Bus framing contract (shared by producer + consumer). // // A projection stream carries two channels distinguished only by the event's @@ -55,3 +57,24 @@ type Transform[W, T any] func(W) T // 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) + +// JSONDecode is a ready-made Decoder for domains whose wire schema is just a +// JSON array of Update[W] (the generic shape — Key/Value/Deleted). Catalog does +// NOT use this (its wire is a flat Projection for back-compat); identity +// consumers like tracking popularity can, pairing it with JSONEncode on the +// emitter side. Marshal errors propagate so HandleFrame can NACK. +func JSONDecode[W any](payload []byte) ([]Update[W], error) { + var out []Update[W] + if err := json.Unmarshal(payload, &out); err != nil { + return nil, err + } + return out, nil +} + +// JSONEncode is the inverse of JSONDecode — the EncodeFunc an Emitter uses when +// the wire schema is a JSON array of Update[W]. A nil/empty slice marshals to a +// payload JSONDecode reads back as zero updates, which is what begin/end frames +// need. +func JSONEncode[W any](items []Update[W]) ([]byte, error) { + return json.Marshal(items) +} diff --git a/projection/hashing_test.go b/projection/hashing_test.go new file mode 100644 index 0000000..36a1520 --- /dev/null +++ b/projection/hashing_test.go @@ -0,0 +1,114 @@ +package projection + +import ( + "path/filepath" + "strings" + "testing" +) + +// TestKeyHashing_LongKeyRoundTrips — a key well over the 32-byte raw ceiling +// (a query string / page path) is accepted in hashed mode and Get finds it. +func TestKeyHashing_LongKeyRoundTrips(t *testing.T) { + dir := t.TempDir() + longKey := "/kategori/" + strings.Repeat("hem-hushall-tradgard/", 5) + "?q=" + strings.Repeat("x", 80) + if len(longKey) <= maxSKULen { + t.Fatalf("test bug: key is only %d bytes, not over the ceiling", len(longKey)) + } + + w := NewSnapshotWriter[tProj]().WithKeyHashing() + // The value carries the original key — the id/sku stays the link even though + // the index key is a digest. + if err := w.Append(longKey, tProj{SKU: longKey, PriceIncVat: 42}); err != nil { + t.Fatalf("Append long key in hashed mode: %v", err) + } + if _, err := w.Write(dir, "hashed.bin"); err != nil { + t.Fatal(err) + } + + r, err := ReadSnapshotFile[tProj](filepath.Join(dir, "hashed.bin")) + if err != nil { + t.Fatal(err) + } + defer r.Close() + + got, ok := r.Get(longKey) + if !ok { + t.Fatal("Get(longKey) miss — hashed lookup did not match the stored digest") + } + if got.PriceIncVat != 42 || got.SKU != longKey { + t.Fatalf("got %+v, want price 42 and original key in value", got) + } + // Scan yields the digest, not the original — caller recovers it from the value. + r.Scan(func(key string, v tProj) bool { + if key == longKey { + t.Error("Scan key should be the digest in hashed mode, got the original") + } + if len(key) != 32 { + t.Errorf("digest key len = %d, want 32 hex chars", len(key)) + } + if v.SKU != longKey { + t.Errorf("value lost the original key: %q", v.SKU) + } + return true + }) +} + +// TestKeyHashing_RawModeStillRejectsLongKeys — default (raw) mode is unchanged: +// an over-long key still errors, so catalog's contract is untouched. +func TestKeyHashing_RawModeStillRejectsLongKeys(t *testing.T) { + w := NewSnapshotWriter[tProj]() // raw + long := strings.Repeat("x", 33) + if err := w.Append(long, tProj{SKU: long}); err == nil { + t.Fatal("raw mode should still reject a 33-byte key") + } +} + +// TestKeyHashing_RawFileHeaderUnchanged — a raw-mode file sets keyEnc=0 at +// offset 12, i.e. byte-identical header to a pre-hashing v2 file. Guards the +// back-compat claim for catalog's existing .bin files. +func TestKeyHashing_RawFileHeaderUnchanged(t *testing.T) { + dir := t.TempDir() + w := NewSnapshotWriter[tProj]() + if err := appendP(w, makeProjection("A", 1)); err != nil { + t.Fatal(err) + } + if _, err := w.Write(dir, "raw.bin"); err != nil { + t.Fatal(err) + } + r, err := ReadSnapshotFile[tProj](filepath.Join(dir, "raw.bin")) + if err != nil { + t.Fatal(err) + } + defer r.Close() + if r.raw.keyEnc != keyEncRaw { + t.Fatalf("raw file keyEnc = %d, want 0", r.raw.keyEnc) + } + if _, ok := r.Get("A"); !ok { + t.Error("raw-mode Get still works") + } +} + +// TestKeyHashing_UnknownEncodingRejected — a header carrying an unknown key +// encoding is rejected rather than silently mis-looked-up. +func TestKeyHashing_UnknownEncodingRejected(t *testing.T) { + dir := t.TempDir() + w := NewSnapshotWriter[tProj]() + if err := appendP(w, makeProjection("A", 1)); err != nil { + t.Fatal(err) + } + path := filepath.Join(dir, "tamper.bin") + if _, err := w.Write(dir, "tamper.bin"); err != nil { + t.Fatal(err) + } + data, err := os2ReadFile(path) + if err != nil { + t.Fatal(err) + } + data[12] = 0x09 // keyEnc = 9, unknown + if err := os2WriteFile(path, data, 0o644); err != nil { + t.Fatal(err) + } + if _, err := ReadSnapshotFile[tProj](path); err == nil { + t.Fatal("unknown key encoding should be rejected") + } +} diff --git a/projection/helpers_test.go b/projection/helpers_test.go index 167c34c..9df1703 100644 --- a/projection/helpers_test.go +++ b/projection/helpers_test.go @@ -26,6 +26,10 @@ func os2Stat(path string) (os.FileInfo, error) { return os.Stat(path) } +func os2ReadFile(path string) ([]byte, error) { + return os.ReadFile(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 { diff --git a/projection/snapshot.go b/projection/snapshot.go index f7daba4..482aa7b 100644 --- a/projection/snapshot.go +++ b/projection/snapshot.go @@ -40,7 +40,9 @@ package projection import ( + "crypto/sha256" "encoding/binary" + "encoding/hex" "encoding/json" "errors" "fmt" @@ -60,16 +62,40 @@ const ( recordSize = 32 + 8 + 4 // 44 bytes per record ) +// Key encoding (stored in the formerly-reserved uint32 at header offset 12). +// 0 = raw, which is byte-identical to a pre-hashing v2 file — so catalog's +// existing .bin files keep parsing unchanged. 1 = hashed: keys are stored as +// hex(sha256(key)[:16]) — 32 ASCII chars, so an arbitrarily long key (page +// path, query string) fits the fixed 32-byte slot. The original key is NOT +// recoverable from the index in hashed mode — the consumer carries it inside +// the value blob (the id/sku stays the link). A whole file is one mode: Get +// applies the same transform the Writer used, read from the header. +const ( + keyEncRaw uint32 = 0 + keyEncHashed uint32 = 1 +) + 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) + ErrSnapshotKeyEnc = errors.New("projection: unknown key encoding in snapshot header") + ErrSKUTooLong = fmt.Errorf("projection: key longer than %d bytes (enable key hashing for unbounded keys)", maxSKULen) ErrTenantTooLong = fmt.Errorf("projection: tenant name longer than %d bytes", maxTenantLen) ) +// hashKey maps an arbitrary-length key to a fixed 32-char hex digest for the +// hashed key encoding. 128 bits (truncated SHA-256) — collision-negligible at +// the cardinalities this store targets (tens of thousands of keys). Hex output +// is pure ASCII, so it never contains the NUL bytes the padded comparator +// treats as terminators. +func hashKey(key string) string { + sum := sha256.Sum256([]byte(key)) + return hex.EncodeToString(sum[:16]) +} + // 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 @@ -79,6 +105,7 @@ var ( // Single-shot: append() all, then write() once (atomic rename). type rawSnapshotWriter struct { tenant string + keyEnc uint32 blobs []byte records []snapshotRecord dirty bool @@ -92,14 +119,18 @@ type snapshotRecord struct { BlobLength uint32 } -// append adds a pre-encoded blob under key. ErrSKUTooLong if key exceeds the -// on-disk width. +// append adds a pre-encoded blob under key. In raw mode, ErrSKUTooLong if key +// exceeds the on-disk width; in hashed mode the key is digested to a fixed +// 32-char slot first, so any length is accepted. func (w *rawSnapshotWriter) append(sku string, blob []byte) error { - if len(sku) > maxSKULen { + key := sku + if w.keyEnc == keyEncHashed { + key = hashKey(sku) + } else if len(sku) > maxSKULen { return ErrSKUTooLong } var rec snapshotRecord - copy(rec.SKU[:], sku) + copy(rec.SKU[:], key) rec.BlobOffset = uint64(headerSize + len(w.blobs)) rec.BlobLength = uint32(len(blob)) @@ -130,11 +161,23 @@ func (w *SnapshotWriter[T]) WithTenant(tenant string) *SnapshotWriter[T] { 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. +// WithKeyHashing switches this writer to the hashed key encoding: keys are +// stored as a fixed 32-char digest, so arbitrarily long keys (page paths, query +// strings) are accepted. The original key is NOT recoverable from the index — +// carry it inside the value if you need it back (see Scan). A reader opening the +// file applies the same hashing on Get automatically (the mode is in the +// header). Set this before any Append. +func (w *SnapshotWriter[T]) WithKeyHashing() *SnapshotWriter[T] { + w.raw.keyEnc = keyEncHashed + return w +} + +// Append encodes v and indexes it under key. In raw mode returns ErrSKUTooLong +// if key exceeds the on-disk width; in hashed mode any length is accepted. +// 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 { + if w.raw.keyEnc == keyEncRaw && len(sku) > maxSKULen { return ErrSKUTooLong } jb, err := json.Marshal(v) @@ -190,7 +233,7 @@ func (w *rawSnapshotWriter) write(dir, name string) (finalPath string, err error 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.PutUint32(hdr[12:16], w.keyEnc) // 0 = raw (byte-identical to pre-hashing v2) binary.LittleEndian.PutUint64(hdr[16:24], uint64(len(w.records))) binary.LittleEndian.PutUint64(hdr[24:32], blobStart) binary.LittleEndian.PutUint64(hdr[32:40], recordStart) @@ -282,6 +325,7 @@ type rawSnapshotReader struct { count uint64 blobStart uint64 recordStart uint64 + keyEnc uint32 // raw | hashed; applied to the query key in getRaw Tenant string // populated from header at Open time; "" if not set closeOnce sync.Once @@ -329,12 +373,16 @@ func (r *SnapshotReader[T]) Get(sku string) (T, bool) { 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. +// Scan walks every (key, value) in stored-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. +// +// In hashed key mode the `key` passed to fn is the DIGEST, not the original — +// recover the original from the value (that's why the consumer carries the +// id/sku in the value blob). func (r *SnapshotReader[T]) Scan(fn func(key string, v T) bool) { r.raw.scan(func(key string, blob []byte) bool { var v T @@ -384,6 +432,12 @@ func newRawSnapshotReader(data []byte) (*rawSnapshotReader, error) { if version != snapshotVersion { return nil, ErrSnapshotVersion } + keyEnc := binary.LittleEndian.Uint32(data[12:16]) + if keyEnc != keyEncRaw && keyEnc != keyEncHashed { + // Reject rather than mis-look-up: an unknown encoding means Get can't + // reproduce the key transform the writer used. + return nil, fmt.Errorf("%w: %d", ErrSnapshotKeyEnc, keyEnc) + } count := binary.LittleEndian.Uint64(data[16:24]) blobStart := binary.LittleEndian.Uint64(data[24:32]) recordStart := binary.LittleEndian.Uint64(data[32:40]) @@ -435,6 +489,7 @@ func newRawSnapshotReader(data []byte) (*rawSnapshotReader, error) { count: count, blobStart: blobStart, recordStart: recordStart, + keyEnc: keyEnc, Tenant: tenant, }, nil } @@ -461,6 +516,9 @@ func (r *rawSnapshotReader) getRaw(sku string) ([]byte, bool) { if r.count == 0 || sku == "" { return nil, false } + if r.keyEnc == keyEncHashed { + sku = hashKey(sku) // same transform the writer applied + } skuBytes := []byte(sku) if len(skuBytes) > maxSKULen { return nil, false diff --git a/projection/store.go b/projection/store.go index 45bcfbe..8ffc63f 100644 --- a/projection/store.go +++ b/projection/store.go @@ -45,9 +45,11 @@ type Store[W, T any] struct { fallback Fallback[W] holder *SnapshotHolder[T] // mmap base; atomic-swap + drain-grace + hashKeys bool // build the base + dedup Scan under the hashed key encoding + mu sync.RWMutex epoch int64 // epoch of the live base - overlay map[string]overlayEntry[T] // same-epoch deltas layered on the base + overlay map[string]overlayEntry[T] // same-epoch deltas layered on the base (keyed by ORIGINAL key) // in-progress snapshot assembly (one epoch at a time) building *SnapshotWriter[T] @@ -60,13 +62,24 @@ type overlayEntry[T any] struct { deleted bool } +// StoreOption configures a Store at construction. +type StoreOption func(*storeConfig) + +type storeConfig struct{ hashKeys bool } + +// WithKeyHashing makes the store build its base under the hashed key encoding, +// so unbounded keys (page paths, query strings) fit the fixed-width index. Get +// still takes the original key (it is hashed internally); Scan yields digests +// (recover the original from the value). Match this to how the producer emits. +func WithKeyHashing() StoreOption { return func(c *storeConfig) { c.hashKeys = true } } + // 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) { +func NewStore[W, T any](dir, name string, decode Decoder[W], transform Transform[W, T], fallback Fallback[W], opts ...StoreOption) (*Store[W, T], error) { if dir == "" || name == "" { return nil, fmt.Errorf("projection: Store needs dir and name") } @@ -76,12 +89,17 @@ func NewStore[W, T any](dir, name string, decode Decoder[W], transform Transform if transform == nil { return nil, fmt.Errorf("projection: Store needs a transform (declare which facts this consumer stores)") } + var cfg storeConfig + for _, o := range opts { + o(&cfg) + } s := &Store[W, T]{ dir: dir, name: name, decode: decode, transform: transform, fallback: fallback, + hashKeys: cfg.hashKeys, holder: NewSnapshotHolder[T](0), overlay: make(map[string]overlayEntry[T]), } @@ -127,6 +145,9 @@ func (s *Store[W, T]) beginSnapshot(epoch int64, tenant string) error { // 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) + if s.hashKeys { + s.building.WithKeyHashing() + } s.buildingEpoch = epoch s.buildingSeen = 0 return nil @@ -271,13 +292,23 @@ func (s *Store[W, T]) Get(sku string) (T, bool) { func (s *Store[W, T]) Scan(fn func(key string, v T) bool) { s.mu.RLock() defer s.mu.RUnlock() + // The base (holder) is keyed by stored key — a digest in hashed mode. To + // dedup overlay against it and emit one consistent key space, project the + // overlay's original keys through the same encoding. + storeKey := func(k string) string { + if s.hashKeys { + return hashKey(k) + } + return k + } seen := make(map[string]struct{}, len(s.overlay)) for k, e := range s.overlay { - seen[k] = struct{}{} + sk := storeKey(k) + seen[sk] = struct{}{} if e.deleted { continue } - if !fn(k, e.val) { + if !fn(sk, e.val) { return } }