store: ReadPage on a raw log reads the page, not the log

xwalLog.ReadPage was readPage(l.Read(), ...): it decoded every record in the
channel to return n of them. Every caller asking for "the last record" was
walking the whole conversation, while the comments beside them said it cost
one read: the dormant reader's tail probe (on every read of a dormant aria),
the agent's, the IR door's pages, the composed cache's seed and bracket reads
of an ancestor. The aria's own log is a tree log in front of the same channel
and delegates ReadPage to it, so it paid the same.

The channel is addressed by its own LT, so the window is arithmetic on the
bounds: backward from before-1 until n records are found, forward from
max(first, from). A record that does not read is skipped as Read skipped it,
so the page reaches past it exactly as the slice did. total is the channel's
extent, as Len reports it and as the tree log already returned.

TestReadPageByPositionMatchesTheSlice asks every (from, before, n) shape of
every aria in random fork chains, raw and through the aria's log, against
readPage over Read(); two off-by-one breakages were canaried and caught.
BenchmarkReadPage: a tail read is ~18-20 us at 300 records and at 3000.

TurnAtLT and the lineage probe keep their positional reads; their comments no
longer cite this bug.

Author: gluck

internal/angelus/lineage_bench_test.go

@@ -22,12 +22,12 @@ import (
 // long conversation, then a CHAIN of forks, each forked again after its own
 // turn, the leaf read against its parent. Two shapes:
 //
-//   head      each link a head fork of an abandoned parent: the base sits past
-//             the end of the parent's log, which used to mean a walk of that
-//             log from LT 1 for every link, unmemoized, on every hop.
-//   interior  each link cut at a turn the parent has: the base is a real
-//             record, which the old memo did remember, so this shape isolates
-//             the other cost: listing every trunk in the store per read.
+//	head      each link a head fork of an abandoned parent: the base sits past
+//	          the end of the parent's log, which used to mean a walk of that
+//	          log from LT 1 for every link, unmemoized, on every hop.
+//	interior  each link cut at a turn the parent has: the base is a real
+//	          record, which the old memo did remember, so this shape isolates
+//	          the other cost: listing every trunk in the store per read.
 //
 // unrelated is how many other arias share the store, which is what the
 // listing paid for.

internal/angelus/lineage_handlers.go

@@ -53,9 +53,7 @@ func (a *Angelus) baseTurnOf(parent string, l store.Link) uint64 {
 	}
 	if l.BaseTurnKnown {
 		// Recorded past the end. The parent's tail says whether it has
-		// written at the base since; until it has, the record stands. The
-		// tail is a positional read, not ReadPage, which on a raw node log
-		// materializes the whole log to return one record.
+		// written at the base since; until it has, the record stands.
 		if tail, ok := log.PeekTail(); !ok || tail.LT < l.BaseLT {
 			return l.BaseTurn
 		}

internal/store/forkturn_test.go

@@ -369,3 +369,71 @@ func onlySidecar(t *testing.T, root string) string {
 	}
 	return found[0]
 }
+
+// READPAGE BY POSITION IS THE SAME PAGE AS THE WHOLE-LOG SLICE. The raw log's
+// ReadPage used to decode every record and slice; it reads only the window
+// now. Every (from, before, n) shape is asked of every aria in random fork
+// chains (whose prefixes are read through their parents), raw and through the
+// aria's own log, and compared with readPage over Read().
+func TestReadPageByPositionMatchesTheSlice(t *testing.T) {
+	for seed := int64(1); seed <= 12; seed++ {
+		rng := rand.New(rand.NewSource(seed))
+		be, cur, log := newConversation(t)
+		chain := []string{cur}
+		turn := uint64(0)
+		for hop := 0; hop < 3; hop++ {
+			for i := 0; i < 2+rng.Intn(5); i++ {
+				turn++
+				put(t, log, ask(turn, "q"))
+				put(t, log, answer(turn, "a"))
+			}
+			at := uint64(0)
+			if rng.Intn(2) == 0 {
+				tail, _ := log.ReadPage(0, 1<<62, 1)
+				at = 1 + uint64(rng.Int63n(int64(tail[0].LT)))
+			}
+			child, _, err := be.ForkWith(cur, at, birth("c"))
+			if err != nil {
+				t.Fatal(err)
+			}
+			cur = child
+			chain = append(chain, cur)
+			log, _ = be.OpenFigIR(cur)
+		}
+		for _, id := range chain {
+			aria, _ := be.OpenFigIR(id)
+			for name, l := range map[string]Log[message.Message]{"raw": be.OpenNode(id), "aria": aria} {
+				all := l.Read()
+				if len(all) == 0 {
+					continue
+				}
+				top := all[len(all)-1].LT
+				for q := 0; q < 200; q++ {
+					from, before, n := uint64(0), uint64(0), rng.Intn(6)
+					switch rng.Intn(4) {
+					case 0:
+						before = 1 + uint64(rng.Int63n(int64(top+3)))
+					case 1:
+						before = ^uint64(0)
+					case 2:
+						from = uint64(rng.Int63n(int64(top + 3)))
+					case 3:
+						from = 0
+					}
+					got, _ := l.ReadPage(from, before, n)
+					want, _ := readPage(all, from, before, n)
+					if len(got) != len(want) {
+						t.Fatalf("seed %d %s %s ReadPage(%d, %d, %d): %d entries, the slice has %d",
+							seed, name, id, from, before, n, len(got), len(want))
+					}
+					for i := range got {
+						if got[i].LT != want[i].LT || got[i].Payload.TurnID != want[i].Payload.TurnID {
+							t.Fatalf("seed %d %s %s ReadPage(%d, %d, %d)[%d]: LT %d turn %d, the slice has LT %d turn %d",
+								seed, name, id, from, before, n, i, got[i].LT, got[i].Payload.TurnID, want[i].LT, want[i].Payload.TurnID)
+						}
+					}
+				}
+			}
+		}
+	}
+}

internal/store/readpage_bench_test.go

@@ -0,0 +1,56 @@
+package store
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/jack-work/figaro/api/message"
+)
+
+// READPAGE, by the shapes its callers actually ask. The tail probe (0, ^0, 1)
+// is what the dormant reader runs on every read of an aria and the agent runs
+// on its own log; a tail page (0, ^0, 32) is a backward page of history; a
+// forward page from the middle is the composed cache filling an LT bracket of
+// an ancestor. raw is the node log the composed cache opens for an ancestor;
+// aria is the aria's own log, the tree log in front of the same channel.
+//
+// Compiles against the tree before the positional ReadPage, so benchstat
+// compares like with like.
+func BenchmarkReadPage(b *testing.B) {
+	prose := strings.Repeat("a sentence of ordinary length in a reply. ", 24) // ~1 KB
+	for _, history := range []int{300, 3000} {
+		be, id, log := newConversation(b)
+		for t := uint64(1); t <= uint64(history/3); t++ {
+			put(b, log, ask(t, "question"))
+			put(b, log, answer(t, prose))
+			put(b, log, answer(t, prose))
+		}
+		tail, _ := log.ReadPage(0, ^uint64(0), 1)
+		mid := tail[0].LT / 2
+		aria, _ := be.OpenFigIR(id)
+		for _, src := range []struct {
+			name string
+			log  Log[message.Message]
+		}{{"raw", be.OpenNode(id)}, {"aria", aria}} {
+			for _, q := range []struct {
+				name         string
+				from, before uint64
+				n            int
+			}{
+				{"tail-1", 0, ^uint64(0), 1},
+				{"tail-32", 0, ^uint64(0), 32},
+				{"forward-32", mid, 0, 32},
+			} {
+				b.Run(fmt.Sprintf("log=%s/query=%s/history=%d", src.name, q.name, history), func(b *testing.B) {
+					b.ReportAllocs()
+					for b.Loop() {
+						if page, _ := src.log.ReadPage(q.from, q.before, q.n); len(page) == 0 {
+							b.Fatal("empty page")
+						}
+					}
+				})
+			}
+		}
+	}
+}

internal/store/turnat.go

@@ -22,9 +22,8 @@ import (
 // It used to be a walk from LT 1 whenever the base was past the end of the
 // parent's log, which is every head fork, on every hop between two arias.
 //
-// ONLY INDEXED READS. PeekTail and ReadFrom address the channel by position;
-// ReadPage, on a raw node log, materializes the whole log and slices it, so
-// the backward read here is built from ReadFrom windows and never from it.
+// ONLY POSITIONAL READS: PeekTail and ReadFrom windows, each of which reads
+// the records it returns and nothing else.
 //
 // Were the invariant ever broken (an unstamped question below lt), the answer
 // would be the stamp BELOW that question: a smaller turn than the truth. For

internal/store/xwal_log.go

@@ -286,8 +286,75 @@ func (l *xwalLog[T]) ReadFrom(lt uint64, n int) []Entry[T] {
 	return out
 }
 
+// ReadPage is readPage's contract read by POSITION: only the records the page
+// returns are read and decoded.
+//
+// It used to be readPage(l.Read(), ...), which decoded every record in the
+// channel to return n of them. Every caller that asked for "the last record"
+// (the dormant reader's tail probe, the agent's, the composed cache's seed for
+// an ancestor range) was walking the whole conversation, while the comments
+// beside them said it cost one read. The channel is addressed by its own LT
+// (see ReadFrom), so the window is arithmetic on the bounds.
+//
+// total is the channel's extent, as Len reports it: the number the tree log
+// in front of this already returns. It counted DECODABLE records before, which
+// is the same number unless a record fails to decode, and counting those meant
+// decoding all of them.
 func (l *xwalLog[T]) ReadPage(from, before uint64, n int) ([]Entry[T], int) {
-	return readPage(l.Read(), from, before, n)
+	var out []Entry[T]
+	total := 0
+	_ = l.openOnce(func(xw *xwal.XWAL) error {
+		first, last, ok := channelBounds(xw, l.channel)
+		if !ok {
+			return nil
+		}
+		total = int(last-first) + 1
+		read := func(lt uint64) (Entry[T], bool) {
+			r, err := xw.ReadAt(l.channel, lt)
+			if err != nil {
+				return Entry[T]{}, false
+			}
+			return decodeRecord[T](r)
+		}
+		if before > 0 {
+			if n <= 0 {
+				return nil
+			}
+			// Backward from before-1, until n are found or the channel ends.
+			// A record that does not read is skipped, as Read skips it, so the
+			// page reaches past it exactly as the whole-log slice did.
+			top := last
+			if before-1 < top {
+				top = before - 1
+			}
+			for lt := top; lt >= first && len(out) < n; lt-- {
+				if e, ok := read(lt); ok {
+					out = append(out, e)
+				}
+				if lt == 0 {
+					break
+				}
+			}
+			for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 {
+				out[i], out[j] = out[j], out[i]
+			}
+			return nil
+		}
+		start := first
+		if from > start {
+			start = from
+		}
+		for lt := start; lt <= last && (n <= 0 || len(out) < n); lt++ {
+			if e, ok := read(lt); ok {
+				out = append(out, e)
+			}
+		}
+		return nil
+	})
+	if out == nil && before > 0 && n > 0 {
+		out = []Entry[T]{}
+	}
+	return out, total
 }
 
 // TailSnapshot is the last n entries, ascending, read from the tail