blocklog: look a block up by index, 51us down to 0.5us on a full log

tools/blocklog/bench_test.go

@@ -33,6 +33,17 @@ func BenchmarkSummaries(b *testing.B) {
 	}
 }
 
+// BenchmarkSummary is the lookup execute_step does for one block, over a full log.
+func BenchmarkSummary(b *testing.B) {
+	l := New()
+	fill(b, l, DefaultMaxEntries)
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_, _ = l.Summary("db-maintenance", "v1", "step_3", "precondition", "state_3")
+	}
+}
+
 // BenchmarkStats is the run and rule totals over the same log.
 func BenchmarkStats(b *testing.B) {
 	l := New()
@@ -65,3 +76,15 @@ func BenchmarkRecordToFile(b *testing.B) {
 		}
 	}
 }
+
+// BenchmarkSummaryByScan is the same lookup done the way Summary did it before
+// the index, for comparison.
+func BenchmarkSummaryByScan(b *testing.B) {
+	l := New()
+	fill(b, l, DefaultMaxEntries)
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_, _ = summaryByScan(l, "db-maintenance", "v1", "step_3", "precondition", "state_3")
+	}
+}

tools/blocklog/blocklog.go

@@ -81,6 +81,18 @@ func (e Entry) key() key {
 	return key{e.Kind, e.Workflow, e.Version, e.TraceID, e.Step, e.BlockedBy, e.MissingState}
 }
 
+// sumKey is what Summary looks a block up by.
+type sumKey struct {
+	workflow, version string
+	step              verify.StepID
+	blockedBy         string
+	missing           verify.State
+}
+
+func (e *Entry) sumKey() sumKey {
+	return sumKey{e.Workflow, e.Version, e.Step, e.BlockedBy, e.MissingState}
+}
+
 // Option configures a Log.
 type Option func(*Log)
 
@@ -112,9 +124,15 @@ func WithClock(now func() time.Time) Option {
 // Log holds recorded blocks, in memory and optionally in an append-only JSON
 // lines file. It is safe for concurrent use.
 type Log struct {
-	mu         sync.Mutex
-	entries    []Entry
-	seen       map[key]struct{}
+	mu      sync.Mutex
+	entries []Entry
+	seen    map[key]struct{}
+	// base is the sequence number of entries[0]; an entry's sequence number is
+	// base plus its position, and never changes while it is kept. bySum lists,
+	// in order, the sequence numbers of the block entries for each sumKey, so
+	// Summary reads the few that match instead of scanning every entry.
+	base       uint64
+	bySum      map[sumKey][]uint64
 	file       *os.File
 	path       string
 	compactErr error
@@ -126,6 +144,7 @@ type Log struct {
 func newLog(opts []Option) *Log {
 	l := &Log{
 		seen:       make(map[key]struct{}),
+		bySum:      make(map[sumKey][]uint64),
 		ttl:        DefaultTTL,
 		maxEntries: DefaultMaxEntries,
 		now:        time.Now,
@@ -234,15 +253,30 @@ func (l *Log) rewrite(path string) error {
 }
 
 func (l *Log) add(e Entry) {
+	if e.Kind == "" {
+		sk := e.sumKey()
+		l.bySum[sk] = append(l.bySum[sk], l.base+uint64(len(l.entries)))
+	}
 	l.entries = append(l.entries, e)
 	l.seen[e.key()] = struct{}{}
 }
 
-// trim drops the oldest entries until the cap holds.
+// trim drops the oldest entries until the cap holds. A dropped block entry has
+// the lowest sequence number of its sumKey, so it is the first in that list.
 func (l *Log) trim() {
 	for len(l.entries) > l.maxEntries {
-		delete(l.seen, l.entries[0].key())
+		old := &l.entries[0]
+		delete(l.seen, old.key())
+		if old.Kind == "" {
+			sk := old.sumKey()
+			if seqs := l.bySum[sk][1:]; len(seqs) > 0 {
+				l.bySum[sk] = seqs
+			} else {
+				delete(l.bySum, sk)
+			}
+		}
 		l.entries = l.entries[1:]
+		l.base++
 	}
 }
 
@@ -499,10 +533,9 @@ func (l *Log) Summary(workflow, version string, step verify.StepID, blockedBy st
 	defer l.mu.Unlock()
 
 	cutoff := l.now().Add(-l.ttl)
-	for i := range l.entries {
-		e := &l.entries[i]
-		if e.Kind != "" || e.Workflow != workflow || e.Version != version || e.Step != step ||
-			e.BlockedBy != blockedBy || e.MissingState != missing || e.At.Before(cutoff) {
+	for _, seq := range l.bySum[sumKey{workflow, version, step, blockedBy, missing}] {
+		e := &l.entries[seq-l.base]
+		if e.At.Before(cutoff) {
 			continue
 		}
 		if !ok {

tools/blocklog/blocklog_test.go

@@ -1,8 +1,11 @@
 package blocklog
 
 import (
+	"fmt"
+	"math/rand"
 	"os"
 	"path/filepath"
+	"reflect"
 	"runtime"
 	"strings"
 	"sync"
@@ -251,3 +254,77 @@ func TestOpenSucceedsWhenTheFileCannotBeCompacted(t *testing.T) {
 		t.Error("the failed compaction should be reported")
 	}
 }
+
+// summaryByScan is Summary as it was before the index: every entry is checked.
+func summaryByScan(l *Log, workflow, version string, step verify.StepID, blockedBy string, missing verify.State) (s Summary, ok bool) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	cutoff := l.now().Add(-l.ttl)
+	for i := range l.entries {
+		e := &l.entries[i]
+		if e.Kind != "" || e.Workflow != workflow || e.Version != version || e.Step != step ||
+			e.BlockedBy != blockedBy || e.MissingState != missing || e.At.Before(cutoff) {
+			continue
+		}
+		if !ok {
+			s = Summary{Step: e.Step, BlockedBy: e.BlockedBy, MissingState: e.MissingState}
+			ok = true
+		}
+		s.Runs++
+		if !e.At.Before(s.LastSeen) {
+			s.LastSeen = e.At
+			s.EstablishedBy = e.EstablishedBy
+		}
+	}
+	return s, ok
+}
+
+// TestSummaryIndexAgreesWithAFullScan records blocks, runs and recoveries
+// through trims and expiry, and checks every lookup against the scan.
+func TestSummaryIndexAgreesWithAFullScan(t *testing.T) {
+	now := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+	l := New(WithMaxEntries(40), WithTTL(10*time.Hour), WithClock(func() time.Time { return now }))
+	rng := rand.New(rand.NewSource(1))
+	steps := []verify.StepID{"a", "b", "c"}
+	states := []verify.State{"x", "y"}
+	for i := 0; i < 600; i++ {
+		now = now.Add(time.Duration(rng.Intn(40)) * time.Minute)
+		switch rng.Intn(4) {
+		case 0:
+			_, _ = l.RecordRun("w", "v1", fmt.Sprintf("t%d", rng.Intn(30)))
+		case 1:
+			_, _ = l.RecordRecovery("w", "v1", fmt.Sprintf("t%d", rng.Intn(30)), steps[rng.Intn(3)])
+		default:
+			_, _ = l.Record(Entry{
+				Workflow: "w", Version: fmt.Sprintf("v%d", 1+rng.Intn(2)), TraceID: fmt.Sprintf("t%d", rng.Intn(30)),
+				Step: steps[rng.Intn(3)], BlockedBy: "precondition", MissingState: states[rng.Intn(2)],
+				EstablishedBy: []verify.StepID{steps[rng.Intn(3)]},
+			})
+		}
+		for _, v := range []string{"v1", "v2"} {
+			for _, st := range steps {
+				for _, ms := range states {
+					got, gok := l.Summary("w", v, st, "precondition", ms)
+					want, wok := summaryByScan(l, "w", v, st, "precondition", ms)
+					if gok != wok || !reflect.DeepEqual(got, want) {
+						t.Fatalf("step %d, %s/%s/%s: index %+v %v, scan %+v %v", i, v, st, ms, got, gok, want, wok)
+					}
+				}
+			}
+		}
+	}
+	// Nothing is left behind for entries that were trimmed away.
+	listed := 0
+	for _, seqs := range l.bySum {
+		listed += len(seqs)
+	}
+	blocks := 0
+	for _, e := range l.entries {
+		if e.Kind == "" {
+			blocks++
+		}
+	}
+	if listed != blocks {
+		t.Fatalf("index lists %d entries, log holds %d blocks", listed, blocks)
+	}
+}