Keep builds on the bus under their module, so no lookup reads the bucket whole (issue 1121, ADR 0427)

Milestone 5 moved 8,599 build records onto the bus under bare build ids, each
carrying its module's whole manifest. Every build lookup then read the whole
bucket, and a delivery plan's send did that per module per machine while it
held the plans and the machine's declaration: the send of build-agent never
finished, and every other act and the status waited behind it.

Builds now live under <module>.build.<id> (requests under req.<id>, id.<id>
names a build's module), with pointer keys newest, worked, held,
commit.<sha> and source.<reuse key> that move only to a newer build and are
read as hints, repaired when wrong. Manifests are kept once in the same
bucket under manifest.sha256-<hex>, so no grant the running bus lacks is
needed. The old keys are laid out once at start, idempotently, and left
unread. The calls a plan look makes on the silent clocks while it holds the
plans are bounded, since the vendored kv.Keys ignores its context.

internal/chain/backend.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"sort"
+	"strings"
 	"sync"
 	"time"
 )
@@ -33,6 +34,9 @@ type Backend interface {
 	Keys(ctx context.Context) ([]string, error)
 	// Watch sends every key's current value, then a Caught entry, then each later write, until the context ends.
 	Watch(ctx context.Context) (<-chan Entry, error)
+	// WatchFiltered is Watch of the keys a subject pattern matches (`*` one token, `>` the rest), the filter kept by the
+	// server: a read of one module's builds reads that module's keys and no other (novox/hq issue 1121).
+	WatchFiltered(ctx context.Context, filter string) (<-chan Entry, error)
 	// History is a key's earlier values and its current one, oldest first, at most the bucket keeps.
 	History(ctx context.Context, key string) ([]Entry, error)
 }
@@ -58,6 +62,7 @@ type Memory struct {
 	values   map[string][]Entry
 	revision uint64
 	watchers []*watcher
+	filters  map[string]int
 	// Now is the clock the entries are dated by; nil is time.Now.
 	Now func() time.Time
 	// writes counts the creates and updates taken.
@@ -155,7 +160,9 @@ func (m *Memory) write(key string, value []byte) uint64 {
 	}
 	m.values[key] = h
 	for _, w := range m.watchers {
-		w.push(clone(e))
+		if w.match(key) {
+			w.push(clone(e))
+		}
 	}
 	return e.Revision
 }
@@ -178,9 +185,10 @@ func (m *Memory) Keys(_ context.Context) ([]string, error) {
 
 // watcher is one watch of a Memory: an unbounded queue, so a watch that is slow is behind and never misses a write.
 type watcher struct {
-	mu   sync.Mutex
-	q    []Entry
-	wake chan struct{}
+	match func(string) bool
+	mu    sync.Mutex
+	q     []Entry
+	wake  chan struct{}
 }
 
 func (w *watcher) push(e Entry) {
@@ -205,18 +213,54 @@ func (w *watcher) pop() (Entry, bool) {
 }
 
 func (m *Memory) Watch(ctx context.Context) (<-chan Entry, error) {
+	return m.WatchFiltered(ctx, ">")
+}
+
+// SubjectMatches is whether a key matches a subject pattern: `*` stands for one token, `>` for one or more at the end.
+func SubjectMatches(pattern, key string) bool {
+	p, k := strings.Split(pattern, "."), strings.Split(key, ".")
+	for n, t := range p {
+		if t == ">" {
+			return len(k) > n
+		}
+		if n >= len(k) || (t != "*" && t != k[n]) {
+			return false
+		}
+	}
+	return len(p) == len(k)
+}
+
+// Filtered counts the filtered watches a Memory took, by filter, for a test that says what a read touched.
+func (m *Memory) Filtered() map[string]int {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	out := map[string]int{}
+	for k, v := range m.filters {
+		out[k] = v
+	}
+	return out
+}
+
+func (m *Memory) WatchFiltered(ctx context.Context, filter string) (<-chan Entry, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	if m.fail != nil {
 		return nil, m.fail
 	}
-	w := &watcher{wake: make(chan struct{}, 1)}
+	w := &watcher{wake: make(chan struct{}, 1), match: func(k string) bool { return SubjectMatches(filter, k) }}
+	if m.filters == nil {
+		m.filters = map[string]int{}
+	}
+	m.filters[filter]++
 	keys := make([]string, 0, len(m.values))
 	for k := range m.values {
 		keys = append(keys, k)
 	}
 	sort.Strings(keys)
 	for _, k := range keys {
+		if !w.match(k) {
+			continue
+		}
 		if e, held := m.last(k); held {
 			w.q = append(w.q, clone(e))
 		}

internal/chain/bus.go

@@ -38,7 +38,7 @@ func OnTheBus(ctx context.Context, conn *nats.Conn, holds Holds) (*Chain, error)
 	if err != nil {
 		return nil, err
 	}
-	return New(builds, rollouts, assignments, holds), nil
+	return New(builds, rollouts, assignments, holds).WithManifests(builds), nil
 }
 
 // busBucket is a Backend on a key-value bucket. It calls no delete and no purge: the type has none to call (ADR 0408 §4).
@@ -85,7 +85,17 @@ func (b busBucket) Keys(ctx context.Context) ([]string, error) {
 }
 
 func (b busBucket) Watch(ctx context.Context) (<-chan Entry, error) {
-	w, err := b.kv.WatchAll(ctx, jetstream.IgnoreDeletes())
+	return b.WatchFiltered(ctx, ">")
+}
+
+func (b busBucket) WatchFiltered(ctx context.Context, filter string) (<-chan Entry, error) {
+	var w jetstream.KeyWatcher
+	var err error
+	if filter == ">" {
+		w, err = b.kv.WatchAll(ctx, jetstream.IgnoreDeletes())
+	} else {
+		w, err = b.kv.WatchFiltered(ctx, []string{filter}, jetstream.IgnoreDeletes())
+	}
 	if err != nil {
 		return nil, err
 	}

internal/chain/chain.go

@@ -128,6 +128,60 @@ func (s Bucket[T]) All(ctx context.Context) (map[string]Kept[T], error) {
 	}
 }
 
+// Filtered is every record whose key matches a subject pattern, by key, read in one pass of a watch the server filters.
+func (s Bucket[T]) Filtered(ctx context.Context, filter string) (map[string]Kept[T], error) {
+	ctx, stop := context.WithCancel(ctx)
+	defer stop()
+	w, err := s.b.WatchFiltered(ctx, filter)
+	if err != nil {
+		return nil, err
+	}
+	out := map[string]Kept[T]{}
+	for {
+		select {
+		case <-ctx.Done():
+			return nil, fmt.Errorf("reading %s: %w", filter, ctx.Err())
+		case e, open := <-w:
+			if !open {
+				return nil, errors.New("the bucket stopped answering before it was read whole")
+			}
+			if e.Caught {
+				return out, nil
+			}
+			k, err := decode[T](e)
+			if err != nil {
+				return nil, err
+			}
+			out[e.Key] = k
+		}
+	}
+}
+
+// FilteredRaw is Filtered for keys that hold bytes of their own (a pointer holds a build's id), by key.
+func (s Bucket[T]) FilteredRaw(ctx context.Context, filter string) (map[string]Entry, error) {
+	ctx, stop := context.WithCancel(ctx)
+	defer stop()
+	w, err := s.b.WatchFiltered(ctx, filter)
+	if err != nil {
+		return nil, err
+	}
+	out := map[string]Entry{}
+	for {
+		select {
+		case <-ctx.Done():
+			return nil, fmt.Errorf("reading %s: %w", filter, ctx.Err())
+		case e, open := <-w:
+			if !open {
+				return nil, errors.New("the bucket stopped answering before it was read whole")
+			}
+			if e.Caught {
+				return out, nil
+			}
+			out[e.Key] = e
+		}
+	}
+}
+
 // History is a key's earlier records and its current one, oldest first.
 func (s Bucket[T]) History(ctx context.Context, key string) ([]Kept[T], error) {
 	all, err := s.b.History(ctx, key)
@@ -182,10 +236,22 @@ type Chain struct {
 	Rollouts    Bucket[Rollout]
 	Assignments Bucket[Assignment]
 
+	// Manifests holds each build's manifest once, under the digest of its bytes (ManifestKey); Kept says it is on the bus. A
+	// chain without it keeps the manifest in the build's record, as the tests' chains do.
+	Manifests Bucket[json.RawMessage]
+	Kept      bool
+
 	rollouts Backend
 	holds    Holds
 }
 
+// WithManifests gives the chain its manifests bucket.
+func (c *Chain) WithManifests(b Backend) *Chain {
+	c.Manifests = Bucket[json.RawMessage]{b: b, holds: c.holds}
+	c.Kept = true
+	return c
+}
+
 // RolloutsOf is the rollouts bucket typed as the controller's plan, which carries more than the chain's Rollout names:
 // the same keys and values, read and written as the plan.
 func RolloutsOf[T any](c *Chain) Bucket[T] { return Bucket[T]{c.rollouts, c.holds} }
@@ -206,5 +272,6 @@ func InMemory(holds Holds) *Chain {
 	if holds == nil {
 		holds = func(context.Context) error { return nil }
 	}
-	return New(NewMemory(), NewMemory(), NewMemory(), holds)
+	builds := NewMemory()
+	return New(builds, NewMemory(), NewMemory(), holds).WithManifests(builds)
 }

internal/chain/keys.go

@@ -0,0 +1,97 @@
+package chain
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"slices"
+	"strings"
+)
+
+// The builds bucket's keys (novox/hq issue 1121). A build is under its module, so a read of one module is a read of the keys
+// `<module>.>` and no other, and a lookup that has what it looks up by is one Get of a pointer key, whose value is a build's id.
+//
+//	<module>.build.<id>        a build's record, its outcome (a records-only read is the filter `<module>.build.>`)
+//	req.<id>                   a request, and a build that knew no module
+//	id.<id>                    the module of the build with that id (the pointer a lookup by id alone reads)
+//	<module>.newest            the newest build of the module, of any state
+//	<module>.worked            the newest successful build of the module
+//	<module>.held              the build standing for the module's newest successful build (issue 280)
+//	<module>.commit.<sha>      the newest successful build from that commit: a CommitPointer
+//	<module>.source.<key>      the newest successful build with that reuse key (ADR 0350 §1), spelled by SpellReuse
+//	manifest.sha256-<hex>      a manifest, once, by the digest of its bytes
+//
+// **A pointer is a hint** (issue 1121): the record is written first, then each pointer by compare-and-set; a reader Gets the
+// record a pointer names and checks it, and when the pointer is missing or wrong reads the module's records, repairs the
+// pointers and says so.
+//
+// No module's name contains a dot. A key is never removed (ADR 0408 §4); the bare `<id>` keys of before the move stay, and are
+// never read once the kind `builds-layout` is moved.
+const (
+	RequestPrefix = "req"
+	IDPrefix      = "id"
+)
+
+// BuildKey is a build's record under its module; a build that knew none is the request's key.
+func BuildKey(module, id string) string {
+	if module == "" {
+		return RequestKey(id)
+	}
+	return module + ".build." + id
+}
+
+func RequestKey(id string) string         { return RequestPrefix + "." + id }
+func IDKey(id string) string              { return IDPrefix + "." + id }
+func NewestKey(module string) string      { return module + ".newest" }
+func WorkedKey(module string) string      { return module + ".worked" }
+func HeldKey(module string) string        { return module + ".held" }
+func CommitKey(module, sha string) string { return module + ".commit." + sha }
+func SourceKey(module, fp string) string  { return module + ".source." + fp }
+func ModuleFilter(module string) string   { return module + ".build.>" }
+
+// ManifestPrefix is where a build's manifest is kept, in the builds bucket itself: a new bucket needs a grant the bus gets
+// only at a later push, and the push needs the builds. No module is named so.
+const ManifestPrefix = "manifest"
+
+// ReservedModules are the names that are not a module's: a key under one is not a build's record.
+var ReservedModules = []string{ManifestPrefix, RequestPrefix, IDPrefix}
+
+// ManifestKey is where a manifest is kept: `manifest.sha256-<hex of its bytes>`.
+func ManifestKey(raw []byte) string {
+	sum := sha256.Sum256(raw)
+	return ManifestPrefix + ".sha256-" + hex.EncodeToString(sum[:])
+}
+
+// CommitPointer is the value of `<module>.commit.<sha>`: the build, and what the surveys of every module's commits read
+// without a Get of each build: its source fingerprint and what it would put on a machine.
+type CommitPointer struct {
+	Build       string `json:"build"`
+	Source      string `json:"source,omitempty"`
+	Fingerprint string `json:"fingerprint,omitempty"`
+}
+
+// IsRecordKey is whether a key of the builds bucket is a build's record under a module: `<module>.build.<id>`.
+func IsRecordKey(key string) bool {
+	parts := strings.Split(key, ".")
+	return len(parts) == 3 && parts[1] == "build" && !slices.Contains(ReservedModules, parts[0])
+}
+
+// SpellReuse is a reuse key (`v1:<hex>`) as a part of a key the bus takes: every character it does not take becomes `-`.
+func SpellReuse(key string) string {
+	return strings.Map(func(r rune) rune {
+		if r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == '-' || r == '_' {
+			return r
+		}
+		return '-'
+	}, key)
+}
+
+// ModuleOf is the module a record key is under.
+func ModuleOf(key string) string { m, _, _ := strings.Cut(key, "."); return m }
+
+// DecodeKept is a watched entry as a typed record, for a read that took the entries of several kinds and decodes those
+// that hold a record.
+func DecodeKept[T any](e Entry) (Kept[T], error) { return decode[T](e) }
+
+// AsAnswered is the state of a request record whose outcome has been heard: the record keeps the request and says which
+// build answered it, so a read of the requests needs no read of the builds.
+const BuildAnswered = "answered"

internal/chain/keys_test.go

@@ -0,0 +1,38 @@
+package chain
+
+import (
+	"context"
+	"testing"
+)
+
+func TestSubjectsMatchAsTheBusFiltersThem(t *testing.T) {
+	for _, c := range []struct {
+		pattern, key string
+		want         bool
+	}{
+		{"alpha.>", "alpha.build-1", true}, {"alpha.>", "alpha.commit.c1", true}, {"alpha.>", "beta.build-1", false},
+		{"alpha.>", "alpha", false}, {"*.newest", "alpha.newest", true}, {"*.newest", "alpha.commit.newest", false},
+		{"*", "build-1", true}, {"*", "alpha.build-1", false}, {">", "a.b.c", true}, {"req.*", "req.build-1", true},
+	} {
+		if got := SubjectMatches(c.pattern, c.key); got != c.want {
+			t.Errorf("SubjectMatches(%q, %q) = %v", c.pattern, c.key, got)
+		}
+	}
+}
+
+func TestOnlyTheKeysOfAFilterAreWatched(t *testing.T) {
+	ctx := context.Background()
+	c := InMemory(nil)
+	for _, k := range []string{"alpha.build.b1", "alpha.build.b2", "beta.build.b3", "alpha.newest"} {
+		if _, err := c.Builds.CreateRaw(ctx, k, []byte(`{"id":"x"}`)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	got, err := c.Builds.FilteredRaw(ctx, ModuleFilter("alpha"))
+	if err != nil || len(got) != 2 {
+		t.Fatalf("alpha.> gave %v: %v", got, err)
+	}
+	if !IsRecordKey("alpha.build.b1") || IsRecordKey("alpha.newest") || IsRecordKey("alpha.commit.c") || IsRecordKey("manifest.build.x") || IsRecordKey("req.b1") || IsRecordKey("id.b1") {
+		t.Error("IsRecordKey tells a record from a pointer wrongly")
+	}
+}

internal/chain/records.go

@@ -87,15 +87,17 @@ type Request struct {
 // Build is one attempt, whichever way it went, kept under its id. It names its module, its repository and commit, its
 // reuse key (novox/hq ADR 0350) and what it published (ADR 0408 §1).
 type Build struct {
-	ID                string           `json:"id"`
-	State             string           `json:"state,omitempty"`
-	Repository        string           `json:"repository"`
-	Ref               string           `json:"ref,omitempty"`
-	Module            string           `json:"module,omitempty"`
-	Commit            string           `json:"commit,omitempty"`
-	On                string           `json:"on,omitempty"`
-	Path              string           `json:"path,omitempty"`
-	Manifest          json.RawMessage  `json:"manifest,omitempty"`
+	ID         string          `json:"id"`
+	State      string          `json:"state,omitempty"`
+	Repository string          `json:"repository"`
+	Ref        string          `json:"ref,omitempty"`
+	Module     string          `json:"module,omitempty"`
+	Commit     string          `json:"commit,omitempty"`
+	On         string          `json:"on,omitempty"`
+	Path       string          `json:"path,omitempty"`
+	Manifest   json.RawMessage `json:"manifest,omitempty"`
+	// ManifestDigest names the manifest in the manifests bucket (ManifestKey); a record that has it keeps no Manifest.
+	ManifestDigest    string           `json:"manifest-digest,omitempty"`
 	Against           []string         `json:"against,omitempty"`
 	Mirrored          []string         `json:"mirrored,omitempty"`
 	Read              []ReadRepository `json:"read,omitempty"`