Merge fix/1121-holds-released: delivery 2 on the builds' new layout (issue 1121)

Delivery 2 deleted the move and chain_moved, so the layout of the builds under their modules is made by the controller's start (LayBuildsOut), recorded by the key layout.done in the builds bucket, retried every half minute with an urgent condition while it fails; the chain on the bus is refused until it exists. Its source readers use the module and pointer reads.

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,104 @@
+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, LayoutPrefix}
+
+// LayoutPrefix is the name of the key that says the builds are laid out under their modules: LayoutMarkerKey, written once by
+// the controller's start after it has laid the old keys out. Nothing reads the builds before it exists.
+const LayoutPrefix = "layout"
+
+// LayoutMarkerKey is the key whose existence says the layout is made.
+const LayoutMarkerKey = LayoutPrefix + ".done"
+
+// 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"`