the translator reads nothing for a record it has already encoded

ProjectIncrementally derived the board's patches and every studied
form's patches for every record, and then looked up the per-LT
translation cache and discarded all of it on a hit. One board read plus
one read per observed form, per record, to build a message that was
never encoded. Fifty observers was fifty-one reads to produce nothing.

The cache is consulted first now. A hit advances the cursors from the
entry's own stamps, which is where that fact already lived: a stamp is
where a form stood, and knowing it never required reading it.

That alone is incorrect. snap advances only from msg.Patches, is what
Encode renders transitions against, and is returned as
IncrementalProjection.Form and read back as the next warm start. Skip
the derivation and the board freezes, the next uncached record renders
against it, and the stale value is inherited by every later pass.
TestCatchUpReplaysCachedPrefixSnapshot caught exactly that on the first
run, which is why it exists.

So the board is caught up in ONE range read spanning everything the
cached prefix skipped, at the first record that actually needs it,
rather than N reads that were only ever folding into a value. Where the
board stands is now carried explicitly (FormVersionOfSnapshot) instead
of being assumed equal to LastFormVersion, because after a run of hits
it is not.

Ephemeral arias carry their patches on the record and have no accessor
to catch up from, so a hit folds the inline ones. They are already
decoded; it costs no read.

  ColdWalkWarmCache            794us -> 292us   1.93MB -> 655KB
  ColdWalkWarmCache8Observed  1369us -> 514us   2.36MB -> 758KB
  ColdWalkColdCache (control)  1.39ms -> 1.39ms        unchanged

The control is the point: a pass with no cache hits does not move, so
the win is the discarded work and nothing else.

Author: gluck

internal/provider/projection.go

@@ -23,6 +23,10 @@ type IncrementalProjection[T any] struct {
 	// study half of the cursor stamp), and must survive a warm start for
 	// the same reason.
 	LastStudyVersions map[string]uint64
+	// FormVersionOfSnapshot is where Form actually stands, which is not
+	// LastFormVersion once a run of cached entries has been skipped. Both
+	// must survive a warm start or the next pass folds the wrong span.
+	FormVersionOfSnapshot uint64
 }
 
 type ProjectionStats struct {
@@ -65,7 +69,7 @@ type ProjectionConfig[T any] struct {
 func ProjectIncrementally[T any](config ProjectionConfig[T]) (*IncrementalProjection[T], ProjectionStats, error) {
 	state := config.Initial
 	snap := form.Snapshot{}
-	var lastForm uint64
+	var lastForm, snapAt uint64
 	lastStudy := map[string]uint64{}
 
 	// A warm start reads from the watermark; a cold one reads everything.
@@ -88,6 +92,7 @@ func ProjectIncrementally[T any](config ProjectionConfig[T]) (*IncrementalProjec
 		state = previous.State
 		snap = previous.Form
 		lastForm = previous.LastFormVersion
+		snapAt = previous.FormVersionOfSnapshot
 		for k, v := range previous.LastStudyVersions {
 			lastStudy[k] = v
 		}
@@ -104,7 +109,52 @@ func ProjectIncrementally[T any](config ProjectionConfig[T]) (*IncrementalProjec
 		if msg.Role == message.RoleGenesis {
 			continue
 		}
+
+		// THE CACHE FIRST. A record whose bytes are already encoded needs no
+		// patches: they exist only to be rendered, and the rendering is what
+		// was cached. Deriving them and then discarding them cost one board
+		// read plus one read per observed form, per record, on every cold
+		// pass over a warm cache -- which is what a fingerprint bump or a
+		// rejected watermark produces.
+		//
+		// The cursors still advance, from the entry's own stamps. A stamp is
+		// where a form stood, and knowing that never required reading it.
+		if cached, ok := lookupCached(config, entry); ok {
+			stats.Cached++
+			lastForm = maxU64(lastForm, entry.FormChannelVersion)
+			for fid, upTo := range entry.StudyVersions {
+				lastStudy[fid] = maxU64(lastStudy[fid], upTo)
+			}
+			// An EPHEMERAL aria has no accessor and carries its patches on
+			// the record itself, so there is nothing to catch up from later.
+			// They are already decoded here; applying them costs no read.
+			if config.Form == nil {
+				for _, patch := range msg.Patches {
+					snap = snap.Apply(patch)
+				}
+			}
+			if len(cached) > 0 {
+				state = config.Append(state, cached, entry.LT)
+			}
+			continue
+		}
+
+		// A MISS, so the board must be real. Skipping the derivation above
+		// left `snap` behind wherever the last encoded record put it, and
+		// `snap` is what Encode renders transitions against. Catch it up in
+		// ONE range read spanning everything skipped, rather than the N reads
+		// the per-record derivation used to do.
 		if config.Form != nil {
+			// Only when something was actually skipped. The guard is not an
+			// optimisation of a nil result: it is one read per record that a
+			// segment-backed accessor would otherwise perform to be told
+			// nothing, and this loop runs once per record forever.
+			if snapAt < lastForm {
+				for _, p := range config.Form.PatchesBetween(snapAt, lastForm) {
+					snap = snap.Apply(p)
+				}
+				snapAt = lastForm
+			}
 			// (after, upTo]: the previous entry's mark and this one's. Absolute,
 			// so a warm start renders exactly the same patches a cold walk would.
 			msg.Patches = config.Form.PatchesBetween(lastForm, entry.FormChannelVersion)
@@ -140,35 +190,21 @@ func ProjectIncrementally[T any](config ProjectionConfig[T]) (*IncrementalProjec
 			}
 		}
 
-		var encoded []json.RawMessage
-		if config.Cache != nil {
-			if cached, ok := config.Cache.Lookup(entry.LT); ok &&
-				(cached.Fingerprint == "" || cached.Fingerprint == config.Fingerprint) &&
-				len(cached.Payload) > 0 {
-				encoded = cached.Payload
-				stats.Cached++
+		encoded, err := config.Encode(msg, snap)
+		if err != nil {
+			if config.ReportEncodeError != nil {
+				config.ReportEncodeError(entry.LT, err)
 			}
+			return nil, stats, err
 		}
-		if encoded == nil {
-			var err error
-			encoded, err = config.Encode(msg, snap)
-			if err != nil {
-				if config.ReportEncodeError != nil {
-					config.ReportEncodeError(entry.LT, err)
-				}
-				return nil, stats, err
-			} else {
-				stats.Encoded++
-				if config.Cache != nil && len(encoded) > 0 {
-					_, err = config.Cache.Append(store.Entry[[]json.RawMessage]{
-						FigaroLT:    entry.LT,
-						Payload:     encoded,
-						Fingerprint: config.Fingerprint,
-					})
-					if err != nil && config.HandleCacheError != nil {
-						config.HandleCacheError(entry.LT, err)
-					}
-				}
+		stats.Encoded++
+		if config.Cache != nil && len(encoded) > 0 {
+			if _, cerr := config.Cache.Append(store.Entry[[]json.RawMessage]{
+				FigaroLT:    entry.LT,
+				Payload:     encoded,
+				Fingerprint: config.Fingerprint,
+			}); cerr != nil && config.HandleCacheError != nil {
+				config.HandleCacheError(entry.LT, cerr)
 			}
 		}
 		if len(encoded) > 0 {
@@ -177,6 +213,7 @@ func ProjectIncrementally[T any](config ProjectionConfig[T]) (*IncrementalProjec
 		for _, patch := range msg.Patches {
 			snap = snap.Apply(patch)
 		}
+		snapAt = maxU64(snapAt, entry.FormChannelVersion)
 	}
 
 	// lastLT must be the tail of the WHOLE log, not of the suffix we walked:
@@ -194,9 +231,34 @@ func ProjectIncrementally[T any](config ProjectionConfig[T]) (*IncrementalProjec
 		LastLT:            lastLT,
 		LastFormVersion:   lastForm,
 		LastStudyVersions: lastStudy,
+
+		FormVersionOfSnapshot: snapAt,
 	}, stats, nil
 }
 
+// lookupCached answers whether this entry is already encoded under the
+// current fingerprint.
+func lookupCached[T any](config ProjectionConfig[T], entry store.Entry[message.Message]) ([]json.RawMessage, bool) {
+	if config.Cache == nil {
+		return nil, false
+	}
+	cached, ok := config.Cache.Lookup(entry.LT)
+	if !ok || len(cached.Payload) == 0 {
+		return nil, false
+	}
+	if cached.Fingerprint != "" && cached.Fingerprint != config.Fingerprint {
+		return nil, false
+	}
+	return cached.Payload, true
+}
+
+func maxU64(a, b uint64) uint64 {
+	if b > a {
+		return b
+	}
+	return a
+}
+
 type EncodedMessages struct {
 	PerMessage   [][]json.RawMessage
 	LogicalTimes []uint64

internal/provider/projection_reads_test.go

@@ -0,0 +1,203 @@
+package provider_test
+
+import (
+	"encoding/json"
+	"fmt"
+	"testing"
+
+	"github.com/jack-work/figaro/internal/form"
+	"github.com/jack-work/figaro/internal/message"
+	"github.com/jack-work/figaro/internal/provider"
+	"github.com/jack-work/figaro/internal/store"
+)
+
+// countingForm records what the projection actually asked the store for.
+// The counter is the artifact, not the timing: a duration decays with the
+// machine, a read count states the shape and fails when the shape changes.
+type countingForm struct {
+	patches []message.Patch
+	calls   int
+	spanned int // total versions covered by the ranges asked for
+}
+
+func (f *countingForm) PatchesBetween(after, upTo uint64) []message.Patch {
+	f.calls++
+	if upTo > uint64(len(f.patches)) {
+		upTo = uint64(len(f.patches))
+	}
+	if after >= upTo {
+		return nil
+	}
+	f.spanned += int(upTo - after)
+	return f.patches[after:upTo]
+}
+
+// projectOnce builds a log of n entries, each advancing the board by one
+// patch, and pre-populates the translation cache for the first `cached` of
+// them. Returns the accessor so the caller can read the counters.
+func projectOnce(t testing.TB, n, cached, observed int) (*countingForm, provider.ProjectionStats) {
+	t.Helper()
+
+	board := make([]message.Patch, n)
+	for i := range board {
+		board[i] = message.Patch{Set: map[string]json.RawMessage{
+			"mode": json.RawMessage(fmt.Sprintf(`"v%d"`, i)),
+		}}
+	}
+	acc := &countingForm{patches: board}
+
+	studies := map[string]provider.Form{}
+	studyAccs := map[string]*countingForm{}
+	for i := 0; i < observed; i++ {
+		id := fmt.Sprintf("@obs%02d", i)
+		a := &countingForm{patches: board}
+		studies[id], studyAccs[id] = a, a
+	}
+
+	log := store.NewMemLog[message.Message]()
+	cache := store.NewMemLog[[]json.RawMessage]()
+	const fingerprint = "fp"
+	for i := 0; i < n; i++ {
+		versions := map[string]uint64{}
+		for id := range studies {
+			versions[id] = uint64(i + 1)
+		}
+		e, err := log.Append(store.Entry[message.Message]{
+			Payload:            message.Message{Role: message.RoleInput},
+			FormChannelVersion: uint64(i + 1),
+			StudyVersions:      versions,
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if i < cached {
+			if _, err := cache.Append(store.Entry[[]json.RawMessage]{
+				FigaroLT:    e.LT,
+				Payload:     []json.RawMessage{json.RawMessage(`{"cached":true}`)},
+				Fingerprint: fingerprint,
+			}); err != nil {
+				t.Fatal(err)
+			}
+		}
+	}
+
+	_, stats, err := provider.ProjectIncrementally(provider.ProjectionConfig[int]{
+		Log:         log,
+		Cache:       cache,
+		Form:        acc,
+		Studies:     studies,
+		Fingerprint: fingerprint,
+		Encode: func(m message.Message, _ form.Snapshot) ([]json.RawMessage, error) {
+			return []json.RawMessage{json.RawMessage(`{}`)}, nil
+		},
+		Append: func(s int, _ []json.RawMessage, _ uint64) int { return s + 1 },
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, a := range studyAccs {
+		acc.calls += a.calls
+	}
+	return acc, stats
+}
+
+// A record whose bytes are cached costs no read, at any layer. This is the
+// whole change, stated as an assertion rather than as a benchmark.
+func TestCachedRecordsCostNoReads(t *testing.T) {
+	acc, stats := projectOnce(t, 100, 100, 8)
+	if stats.Cached != 100 || stats.Encoded != 0 {
+		t.Fatalf("want 100 cached and 0 encoded, got %d and %d", stats.Cached, stats.Encoded)
+	}
+	if acc.calls != 0 {
+		t.Fatalf("a fully cached pass read the board %d times; it must read nothing", acc.calls)
+	}
+}
+
+// The reads a partly cached pass does make are bounded by the misses, not by
+// the length of the log. The board catch-up over the cached prefix is ONE
+// range, whatever the prefix length.
+func TestReadsScaleWithMissesNotLength(t *testing.T) {
+	short, _ := projectOnce(t, 100, 90, 0)
+	long, _ := projectOnce(t, 1000, 990, 0)
+
+	if short.calls != long.calls {
+		t.Fatalf("reads tracked log length: %d at 100 entries, %d at 1000, with 10 misses either way",
+			short.calls, long.calls)
+	}
+	// One catch-up over the skipped prefix, then one per miss.
+	if want := 1 + 10; short.calls != want {
+		t.Fatalf("want %d reads for 10 misses, got %d", want, short.calls)
+	}
+}
+
+// Studied forms are read only for records that are actually encoded. They
+// feed nothing but the rendering, so a cache hit must not touch them: that
+// cost was one read per observed form per record, and fifty observers was
+// fifty-one reads to produce nothing.
+func TestObservedFormsAreNotReadForCachedRecords(t *testing.T) {
+	none, _ := projectOnce(t, 100, 100, 50)
+	if none.calls != 0 {
+		t.Fatalf("50 observed forms cost %d reads over a fully cached pass", none.calls)
+	}
+}
+
+// The board still has to be correct at the first uncached record, which is
+// what the catch-up buys and what a naive cache-first ordering breaks.
+func TestSnapshotIsCurrentAtTheFirstMiss(t *testing.T) {
+	board := []message.Patch{
+		{Set: map[string]json.RawMessage{"mode": json.RawMessage(`"old"`)}},
+		{Set: map[string]json.RawMessage{"mode": json.RawMessage(`"new"`)}},
+	}
+	acc := &countingForm{patches: board}
+
+	log := store.NewMemLog[message.Message]()
+	cache := store.NewMemLog[[]json.RawMessage]()
+	const fingerprint = "fp"
+	for i := 0; i < 2; i++ {
+		e, err := log.Append(store.Entry[message.Message]{
+			Payload:            message.Message{Role: message.RoleInput},
+			FormChannelVersion: uint64(i + 1),
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if i == 0 {
+			if _, err := cache.Append(store.Entry[[]json.RawMessage]{
+				FigaroLT: e.LT, Payload: []json.RawMessage{json.RawMessage(`{}`)}, Fingerprint: fingerprint,
+			}); err != nil {
+				t.Fatal(err)
+			}
+		}
+	}
+
+	var seen form.Snapshot
+	if _, _, err := provider.ProjectIncrementally(provider.ProjectionConfig[int]{
+		Log: log, Cache: cache, Form: acc, Fingerprint: fingerprint,
+		Encode: func(_ message.Message, snap form.Snapshot) ([]json.RawMessage, error) {
+			seen = snap
+			return []json.RawMessage{json.RawMessage(`{}`)}, nil
+		},
+		Append: func(s int, _ []json.RawMessage, _ uint64) int { return s + 1 },
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	got, ok := seen.Get("mode")
+	if !ok || string(got) != `"old"` {
+		t.Fatalf("the encode after a cached record saw mode=%s; the skipped patch was never folded", got)
+	}
+}
+
+func benchProject(b *testing.B, n, cached, observed int) {
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		projectOnce(b, n, cached, observed)
+	}
+}
+
+// The shape a fingerprint bump or a rejected watermark produces: a cold walk
+// over a log whose per-LT bytes are still valid.
+func BenchmarkColdWalkWarmCache(b *testing.B)          { benchProject(b, 500, 500, 0) }
+func BenchmarkColdWalkWarmCache8Observed(b *testing.B) { benchProject(b, 500, 500, 8) }
+func BenchmarkColdWalkColdCache(b *testing.B)          { benchProject(b, 500, 0, 8) }