From c7959153157599ecf16e0905f3fa285d61820501 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 18 Oct 2026 13:32:31 +0200 Subject: [PATCH] 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 .build. (requests under req., id. names a build's module), with pointer keys newest, worked, held, commit. and source. 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-, 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. --- cmd/mesh-controller/plan_looks.go | 46 + cmd/mesh-controller/plan_looks_test.go | 29 + internal/chain/backend.go | 54 +- internal/chain/bus.go | 14 +- internal/chain/chain.go | 69 +- internal/chain/keys.go | 97 ++ internal/chain/keys_test.go | 38 + internal/chain/records.go | 20 +- internal/inventory/builds.go | 8 +- internal/inventory/builds_test.go | 2 +- internal/inventory/chain.go | 60 +- internal/inventory/chain_builds.go | 876 ++++++++++++++++-- .../inventory/chain_builds_layout_test.go | 398 ++++++++ internal/inventory/chain_move.go | 40 +- internal/inventory/chain_move_test.go | 13 +- internal/inventory/chain_requests.go | 61 +- internal/inventory/fortest.go | 2 +- internal/inventory/source.go | 29 + 18 files changed, 1700 insertions(+), 156 deletions(-) create mode 100644 internal/chain/keys.go create mode 100644 internal/chain/keys_test.go create mode 100644 internal/inventory/chain_builds_layout_test.go diff --git a/cmd/mesh-controller/plan_looks.go b/cmd/mesh-controller/plan_looks.go index 1bfdbf7d..6ccf3467 100644 --- a/cmd/mesh-controller/plan_looks.go +++ b/cmd/mesh-controller/plan_looks.go @@ -412,6 +412,7 @@ func keepSilentKeys(ctx context.Context, c silentClock, want map[silentWait]time byTimer() return nil } + c = boundedClock{c} held, err := c.Held(ctx) if err != nil { byTimer() @@ -435,6 +436,51 @@ func keepSilentKeys(ctx context.Context, c silentClock, want map[silentWait]time return first } +// silentCallBound is how long one call to the silent clock may take while the look holds the plans (novox/hq issue 1121). +// The vendored key listing reads its watcher without the context, so a consumer lost when the broker reloaded its user +// list left the look waiting for good, with the plans and every node held. +var silentCallBound = 20 * time.Second + +// boundedClock is a clock whose every call is answered within silentCallBound, or fails: a call that never returns is +// left behind on its own goroutine. +type boundedClock struct{ silentClock } + +func boundedCall[T any](ctx context.Context, what string, call func(context.Context) (T, error)) (T, error) { + ctx, cancel := context.WithTimeout(ctx, silentCallBound) + defer cancel() + type answer struct { + v T + err error + } + done := make(chan answer, 1) + go func() { v, err := call(ctx); done <- answer{v, err} }() + select { + case a := <-done: + return a.v, a.err + case <-ctx.Done(): + var zero T + return zero, fmt.Errorf("%s: no answer from the bus within %s: %w", what, silentCallBound, ctx.Err()) + } +} + +func (b boundedClock) Arm(ctx context.Context, w silentWait, ttl time.Duration, sent time.Time) error { + _, err := boundedCall(ctx, "arming a wait", func(ctx context.Context) (struct{}, error) { + return struct{}{}, b.silentClock.Arm(ctx, w, ttl, sent) + }) + return err +} + +func (b boundedClock) Held(ctx context.Context) (map[silentWait]time.Time, error) { + return boundedCall(ctx, "listing the waits", b.silentClock.Held) +} + +func (b boundedClock) Reported(ctx context.Context, w silentWait) error { + _, err := boundedCall(ctx, "dropping a wait", func(ctx context.Context) (struct{}, error) { + return struct{}{}, b.silentClock.Reported(ctx, w) + }) + return err +} + var silentTimers = struct { mu sync.Mutex timers map[silentWait]time.Time diff --git a/cmd/mesh-controller/plan_looks_test.go b/cmd/mesh-controller/plan_looks_test.go index be81219a..5b1c8970 100644 --- a/cmd/mesh-controller/plan_looks_test.go +++ b/cmd/mesh-controller/plan_looks_test.go @@ -179,3 +179,32 @@ func TestKeysThatCannotBeReadAreWatchedByTimersAndTheLookFails(t *testing.T) { // A plan moved by a process the serving controller did not spawn — a verb run at the control node's terminal — is // looked at: the controller follows its own plan-moved (issue 974, round 1). + +// A clock whose bus answers nothing, ever: a consumer lost when the broker reloaded its user list, a key listing that +// never reaches its end. +type hungClock struct{ silentClock } + +func (hungClock) Held(context.Context) (map[silentWait]time.Time, error) { + select {} // never answers, whatever the context says (the vendored Keys reads its watcher without it) +} + +// issue 1121: the look holds the plans while it keeps the silent keys, so a key listing that never ends held the plans +// and every node for good. Every call to the clock has a bound, and the wait is looked at by a timer of its own. +func TestAClockThatNeverAnswersDoesNotHoldTheLookForever(t *testing.T) { + was := silentCallBound + silentCallBound = 50 * time.Millisecond + t.Cleanup(func() { silentCallBound = was }) + w := silentWait{Plan: "plan-hung", Tier: 0, Node: "ace"} + sent := time.Now().UTC() + t.Cleanup(func() { silentTimers.mu.Lock(); delete(silentTimers.timers, w); silentTimers.mu.Unlock() }) + done := make(chan error, 1) + go func() { done <- keepSilentKeys(t.Context(), hungClock{}, map[silentWait]time.Time{w: sent}, sent) }() + select { + case err := <-done: + if err == nil { + t.Fatal("a clock that never answered was reported as fine") + } + case <-time.After(5 * time.Second): + t.Fatal("the look is still waiting for a clock that never answers") + } +} diff --git a/internal/chain/backend.go b/internal/chain/backend.go index b99f3673..8d446473 100644 --- a/internal/chain/backend.go +++ b/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)) } diff --git a/internal/chain/bus.go b/internal/chain/bus.go index c94f6373..5000788f 100644 --- a/internal/chain/bus.go +++ b/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 } diff --git a/internal/chain/chain.go b/internal/chain/chain.go index bb3f6862..7de4e0f1 100644 --- a/internal/chain/chain.go +++ b/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) } diff --git a/internal/chain/keys.go b/internal/chain/keys.go new file mode 100644 index 00000000..0c956d72 --- /dev/null +++ b/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 +// `.>` 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. +// +// .build. a build's record, its outcome (a records-only read is the filter `.build.>`) +// req. a request, and a build that knew no module +// id. the module of the build with that id (the pointer a lookup by id alone reads) +// .newest the newest build of the module, of any state +// .worked the newest successful build of the module +// .held the build standing for the module's newest successful build (issue 280) +// .commit. the newest successful build from that commit: a CommitPointer +// .source. the newest successful build with that reuse key (ADR 0350 §1), spelled by SpellReuse +// manifest.sha256- 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 `` 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-`. +func ManifestKey(raw []byte) string { + sum := sha256.Sum256(raw) + return ManifestPrefix + ".sha256-" + hex.EncodeToString(sum[:]) +} + +// CommitPointer is the value of `.commit.`: 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: `.build.`. +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:`) 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" diff --git a/internal/chain/keys_test.go b/internal/chain/keys_test.go new file mode 100644 index 00000000..274cc2ea --- /dev/null +++ b/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") + } +} diff --git a/internal/chain/records.go b/internal/chain/records.go index 709aa833..ceecabbb 100644 --- a/internal/chain/records.go +++ b/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"` diff --git a/internal/inventory/builds.go b/internal/inventory/builds.go index 1fa1c78d..4d7b58ef 100644 --- a/internal/inventory/builds.go +++ b/internal/inventory/builds.go @@ -22,9 +22,11 @@ import ( // Build is one attempt, whichever way it went. type Build struct { - ID string - Repository string - Ref string + ID string + // ManifestDigest names the manifest in the manifests bucket, when a chain keeps it apart from the record (issue 1121). + ManifestDigest string + Repository string + Ref string // Module is empty for a build that failed before it knew what it was building. Module string Commit string diff --git a/internal/inventory/builds_test.go b/internal/inventory/builds_test.go index e0f53b83..bfdbd6fd 100644 --- a/internal/inventory/builds_test.go +++ b/internal/inventory/builds_test.go @@ -216,7 +216,7 @@ func TestABuildsBuildSourceComesBackOverWhatItRead(t *testing.T) { if got := read["build-agent"]; len(got) != 1 || got[0].Paths != nil || got[0].Own { t.Fatalf("a build that said no build source reads as %+v", got) } - kept, found, err := inv.chain.Builds.Get(ctx, "said") + kept, found, err := inv.chain.Builds.Get(ctx, "route-proxy.build.said") if err != nil || !found { t.Fatalf("the build is not on the bus: %v %v", found, err) } diff --git a/internal/inventory/chain.go b/internal/inventory/chain.go index 148d22da..cf83baa4 100644 --- a/internal/inventory/chain.go +++ b/internal/inventory/chain.go @@ -89,29 +89,19 @@ func (i *Inventory) attached() *chain.Chain { // ChainReads says whether a kind of the chain is read from its bucket: chain_moved names it. It answers the chain to read // through, and ErrNoChain when the kind is moved and none is attached. func (i *Inventory) ChainReads(ctx context.Context, kind string) (*chain.Chain, bool, error) { - i.moved.mu.Lock() - known := i.moved.kind[kind] - i.moved.mu.Unlock() - if !known { - var moved bool - err := i.store.Pool().QueryRow(ctx, `select counts ? $1 from chain_moved`, kind).Scan(&moved) - var pgErr *pgconn.PgError - if errors.Is(err, pgx.ErrNoRows) || (errors.As(err, &pgErr) && pgErr.Code == "42P01") { - // No row, or a store older than the migration that makes the table: nothing moved. - return nil, false, nil - } + moved, err := i.kindMoved(ctx, kind) + if err != nil || !moved { + return nil, false, err + } + if kind == KindBuilds { + // **The builds are read under their modules** (issue 1121): until the layout is made, a read would find nothing. + laid, err := i.kindMoved(ctx, KindBuildsLayout) if err != nil { - return nil, false, fmt.Errorf("whether %s moved to the bus cannot be read: %w", kind, err) + return nil, false, err } - if !moved { - return nil, false, nil + if !laid { + return nil, false, fmt.Errorf("%w: the builds are not laid out under their modules yet; the move at the controller's start does it", ErrNoChain) } - i.moved.mu.Lock() - if i.moved.kind == nil { - i.moved.kind = map[string]bool{} - } - i.moved.kind[kind] = true - i.moved.mu.Unlock() } i.moved.mu.Lock() defer i.moved.mu.Unlock() @@ -128,6 +118,36 @@ func (i *Inventory) ChainReads(ctx context.Context, kind string) (*chain.Chain, return i.chain, true, nil } +// kindMoved is whether chain_moved names a kind; once it does, always, so the store is asked until it is true. +func (i *Inventory) kindMoved(ctx context.Context, kind string) (bool, error) { + i.moved.mu.Lock() + known := i.moved.kind[kind] + i.moved.mu.Unlock() + if known { + return true, nil + } + var moved bool + err := i.store.Pool().QueryRow(ctx, `select counts ? $1 from chain_moved`, kind).Scan(&moved) + var pgErr *pgconn.PgError + if errors.Is(err, pgx.ErrNoRows) || (errors.As(err, &pgErr) && pgErr.Code == "42P01") { + // No row, or a store older than the migration that makes the table: nothing moved. + return false, nil + } + if err != nil { + return false, fmt.Errorf("whether %s moved to the bus cannot be read: %w", kind, err) + } + if !moved { + return false, nil + } + i.moved.mu.Lock() + if i.moved.kind == nil { + i.moved.kind = map[string]bool{} + } + i.moved.kind[kind] = true + i.moved.mu.Unlock() + return true, nil +} + // ChainMoved is, per kind moved, how many records the move wrote; empty before any is. func (i *Inventory) ChainMoved(ctx context.Context) (map[string]int, error) { var counts map[string]int diff --git a/internal/inventory/chain_builds.go b/internal/inventory/chain_builds.go index ececd272..06f65807 100644 --- a/internal/inventory/chain_builds.go +++ b/internal/inventory/chain_builds.go @@ -8,6 +8,7 @@ import ( "encoding/json" "errors" "fmt" + "slices" "sort" "strings" "time" @@ -23,7 +24,7 @@ import ( func toChainBuild(b Build) chain.Build { c := chain.Build{ID: b.ID, Repository: b.Repository, Ref: b.Ref, Module: b.Module, Commit: b.Commit, On: b.On, Path: b.Path, Against: b.Against, Mirrored: b.Mirrored, SourceFingerprint: b.SourceFingerprint, - Failed: b.Failed, Asked: b.Asked.UTC(), At: b.At.UTC(), ReuseKey: b.ReuseKey, Published: b.Published} + ManifestDigest: b.ManifestDigest, Failed: b.Failed, Asked: b.Asked.UTC(), At: b.At.UTC(), ReuseKey: b.ReuseKey, Published: b.Published} if len(b.Manifest) > 0 { c.Manifest = json.RawMessage(b.Manifest) } @@ -45,7 +46,7 @@ func toChainBuild(b Build) chain.Build { // fromChainBuild is a build as the mesh reads it. func fromChainBuild(c chain.Build) Build { b := Build{ID: c.ID, Repository: c.Repository, Ref: c.Ref, Module: c.Module, Commit: c.Commit, On: c.On, Path: c.Path, - Manifest: []byte(c.Manifest), Against: c.Against, Mirrored: c.Mirrored, SourceFingerprint: c.SourceFingerprint, + Manifest: []byte(c.Manifest), ManifestDigest: c.ManifestDigest, Against: c.Against, Mirrored: c.Mirrored, SourceFingerprint: c.SourceFingerprint, Failed: c.Failed, Asked: c.Asked, At: c.At, ReuseKey: c.ReuseKey, Published: c.Published} if len(b.Manifest) == 0 { b.Manifest = nil @@ -84,67 +85,499 @@ func publishedOf(made []Artifact) *chain.Published { return &out } -// chainBuildKey is a build's key. An id the bus takes no spelling of is refused by the write, loudly. -func chainBuildKey(id string) string { return id } +// legacyBuildKey is the key a build had before the layout under its module (issue 1121): its id alone. The keys stay (ADR +// 0408 §4) and nothing reads them once the kind builds-layout is moved. +func legacyBuildKey(id string) string { return id } -// chainRecordBuild keeps a build under its id. **Idempotent on the id**, as the table's insert was: a result arrives twice, -// and the first record stands. A request nobody has answered is the record in state asked (issue 1091); the outcome -// takes its place under the same key, and keeps the request it answers. +// chainRecordBuild keeps a build under its module and id. **Idempotent on the id**, as the table's insert was: a result +// arrives twice, and the first record stands. A request nobody has answered is the record `req.` in state asked (issue +// 1091); the outcome is written under its module and keeps the request it answers, and the request's record then says it was +// answered. The build's manifest goes to the manifests bucket once, and the record names its digest (issue 1121). func (i *Inventory) chainRecordBuild(ctx context.Context, c *chain.Chain, b Build) error { if b.At.IsZero() { b.At = time.Now().UTC() } - key := chainBuildKey(b.ID) + if slices.Contains(chain.ReservedModules, b.Module) { + return fmt.Errorf("build %s is not kept on the bus: %q is a name the bucket keeps for itself, not a module's", b.ID, b.Module) + } + digest, err := i.keepManifest(ctx, c, b.Manifest) + if err != nil { + return fmt.Errorf("build %s is not kept on the bus: its manifest: %w", b.ID, err) + } + key, reqKey := chain.BuildKey(b.Module, b.ID), chain.RequestKey(b.ID) for try := 0; try < casTries; try++ { next := toChainBuild(b) + if digest != "" { + next.Manifest, next.ManifestDigest = nil, digest + } cur, found, err := c.Builds.Get(ctx, key) if err != nil { return fmt.Errorf("build %s is not kept on the bus: %w", b.ID, err) } + var asked chain.Kept[chain.Build] + var askedFound bool + if key != reqKey { + if asked, askedFound, err = c.Builds.Get(ctx, reqKey); err != nil { + return fmt.Errorf("build %s is not kept on the bus: %w", b.ID, err) + } + } switch { + case !found && askedFound: + next.Request = asked.Value.Request + _, err = c.Builds.Create(ctx, key, next) + if errors.Is(err, chain.ErrExists) { + continue + } case !found: _, err = c.Builds.Create(ctx, key, next) if errors.Is(err, chain.ErrExists) { continue } - case cur.Value.State == chain.BuildAsked: + case key == reqKey && cur.Value.State == chain.BuildAsked: next.Request = cur.Value.Request _, err = c.Builds.Update(ctx, key, next, cur.Revision) if errors.Is(err, chain.ErrMoved) { continue } default: - // An outcome stands: the first record is the record. + // An outcome stands: the first record is the record. Its id pointer is made if a crash left it out. + if _, err := i.pointTo(ctx, c, chain.IDKey(b.ID), idPointer(b.Module), false); err != nil { + return err + } return i.RecordMirrored(ctx, b.ID, "", b.Mirrored) } if err != nil { return fmt.Errorf("build %s is not kept on the bus: %w", b.ID, err) } + if err := i.pointAt(ctx, c, next, chainFingerprint(fromChainBuild(next))); err != nil { + return err + } + if askedFound && asked.Value.State == chain.BuildAsked { + i.answerRequest(ctx, c, reqKey, next) + } return i.RecordMirrored(ctx, b.ID, "", b.Mirrored) } return fmt.Errorf("build %s is not kept on the bus: it kept moving under the write", b.ID) } +// idPointer is what `id.` holds: the build's module, or the request prefix for a build that knew none. +func idPointer(module string) []byte { + if module == "" { + return []byte(chain.RequestPrefix) + } + return []byte(module) +} + +// pointTo makes a pointer key hold value, by compare-and-set: made when it is absent, and, when overwrite, changed when it +// holds something else. +func (i *Inventory) pointTo(ctx context.Context, c *chain.Chain, key string, value []byte, overwrite bool) (bool, error) { + for try := 0; try < casTries; try++ { + e, found, err := c.Builds.GetRaw(ctx, key) + if err != nil { + return false, fmt.Errorf("%s is not read on the bus: %w", key, err) + } + switch { + case !found: + _, err = c.Builds.CreateRaw(ctx, key, value) + if errors.Is(err, chain.ErrExists) { + continue + } + case !overwrite || bytes.Equal(e.Value, value): + return false, nil + default: + _, err = c.Builds.UpdateRaw(ctx, key, value, e.Revision) + if errors.Is(err, chain.ErrMoved) { + continue + } + } + if err != nil { + return false, fmt.Errorf("%s is not kept on the bus: %w", key, err) + } + return true, nil + } + return false, fmt.Errorf("%s kept moving under the write", key) +} + +// pointAt moves the pointers a build written now is the answer to: its id's module, the module's newest, and for a success +// the module's worked, held, commit and source (issue 1121). +func (i *Inventory) pointAt(ctx context.Context, c *chain.Chain, rec chain.Build, fingerprint string) error { + if _, err := i.pointTo(ctx, c, chain.IDKey(rec.ID), idPointer(rec.Module), false); err != nil { + return err + } + if rec.Module == "" { + return nil + } + if _, err := i.pointIfNewer(ctx, c, chain.NewestKey(rec.Module), []byte(rec.ID), rec); err != nil { + return err + } + if rec.Failed != "" { + return nil + } + // Held: the run of builds made from one source stands on its first. The previous success is read before it is replaced; + // a build older than it, finishing late, moves neither. + hold, older := true, false + if prev, found, err := c.Builds.GetRaw(ctx, chain.WorkedKey(rec.Module)); err != nil { + return err + } else if found { + if old, found, err := c.Builds.Get(ctx, chain.BuildKey(rec.Module, pointedID(prev.Value))); err != nil { + return err + } else if found { + older = !buildNewer(rec, old.Value) + hold = rec.SourceFingerprint == "" || old.Value.SourceFingerprint != rec.SourceFingerprint + } + } + if !older { + if _, err := i.pointIfNewer(ctx, c, chain.WorkedKey(rec.Module), []byte(rec.ID), rec); err != nil { + return err + } + if _, found, err := c.Builds.GetRaw(ctx, chain.HeldKey(rec.Module)); err != nil { + return err + } else if hold || !found { + if _, err := i.pointIfNewer(ctx, c, chain.HeldKey(rec.Module), []byte(rec.ID), rec); err != nil { + return err + } + } + } + if rec.Commit != "" { + body, err := json.Marshal(chain.CommitPointer{Build: rec.ID, Source: rec.SourceFingerprint, Fingerprint: fingerprint}) + if err != nil { + return err + } + if _, err := i.pointIfNewer(ctx, c, chain.CommitKey(rec.Module, rec.Commit), body, rec); err != nil { + return err + } + } + if rec.ReuseKey != "" { + _, err := i.pointIfNewer(ctx, c, chain.SourceKey(rec.Module, chain.SpellReuse(rec.ReuseKey)), []byte(rec.ID), rec) + return err + } + return nil +} + +// buildNewer is whether a is a newer build than b in the order every "what a module currently is" question uses: the newest +// request first, then the newest record, then the id (newestRequestFirst). +func buildNewer(a, b chain.Build) bool { + x, y := fromChainBuild(a).AskedOrAt(), fromChainBuild(b).AskedOrAt() + if !x.Equal(y) { + return x.After(y) + } + if !a.At.Equal(b.At) { + return a.At.After(b.At) + } + return a.ID > b.ID +} + +// pointIfNewer makes a pointer name rec by compare-and-set on the pointer's own revision, unless the build it names now is +// newer: **a pointer follows the newest build, not the last write** (issue 1121), so an older build that finishes late +// leaves it alone. +func (i *Inventory) pointIfNewer(ctx context.Context, c *chain.Chain, key string, value []byte, rec chain.Build) (bool, error) { + for try := 0; try < casTries; try++ { + e, found, err := c.Builds.GetRaw(ctx, key) + if err != nil { + return false, fmt.Errorf("%s is not read on the bus: %w", key, err) + } + if found { + if bytes.Equal(e.Value, value) { + return false, nil + } + cur, has, err := c.Builds.Get(ctx, chain.BuildKey(rec.Module, pointedID(e.Value))) + if err != nil { + return false, err + } + if has && cur.Value.State == "" && !buildNewer(rec, cur.Value) { + return false, nil + } + _, err = c.Builds.UpdateRaw(ctx, key, value, e.Revision) + if errors.Is(err, chain.ErrMoved) { + continue + } + if err != nil { + return false, fmt.Errorf("%s is not kept on the bus: %w", key, err) + } + return true, nil + } + _, err = c.Builds.CreateRaw(ctx, key, value) + if errors.Is(err, chain.ErrExists) { + continue + } + if err != nil { + return false, fmt.Errorf("%s is not kept on the bus: %w", key, err) + } + return true, nil + } + return false, fmt.Errorf("%s kept moving under the write", key) +} + +// answerRequest says on the request's record that a build answered it. Said and not failed on: the outcome stands, and the +// request is read as unheard until a later write makes this one. +func (i *Inventory) answerRequest(ctx context.Context, c *chain.Chain, reqKey string, out chain.Build) { + for try := 0; try < casTries; try++ { + cur, found, err := c.Builds.Get(ctx, reqKey) + if err != nil || !found || cur.Value.State != chain.BuildAsked { + return + } + cur.Value.State, cur.Value.Module, cur.Value.Failed, cur.Value.At = chain.BuildAnswered, out.Module, out.Failed, out.At + if _, err := c.Builds.Update(ctx, reqKey, cur.Value, cur.Revision); !errors.Is(err, chain.ErrMoved) { + return + } + } +} + +// manifestBytes is a manifest as the bucket keeps it: its canonical form, nil when the build kept none. +func manifestBytes(raw []byte) []byte { + if t := bytes.TrimSpace(raw); len(t) == 0 || string(t) == "null" { + return nil + } + return canonicalJSON(raw) +} + +// keepManifest keeps a manifest in the manifests bucket under the digest of its canonical bytes, once, and answers the +// digest; empty when the build kept none or the chain keeps manifests in the record. +func (i *Inventory) keepManifest(ctx context.Context, c *chain.Chain, raw []byte) (string, error) { + body := manifestBytes(raw) + if body == nil || !c.Kept { + return "", nil + } + key := chain.ManifestKey(body) + if _, err := c.Manifests.CreateRaw(ctx, key, body); err != nil && !errors.Is(err, chain.ErrExists) { + return "", err + } + return key, nil +} + +// manifestOfRecord is the manifest a record names, or keeps. +func manifestOfRecord(ctx context.Context, c *chain.Chain, rec chain.Build) ([]byte, error) { + if rec.ManifestDigest == "" { + return []byte(rec.Manifest), nil + } + e, found, err := c.Manifests.GetRaw(ctx, rec.ManifestDigest) + if err != nil { + return nil, fmt.Errorf("the manifest %s of build %s cannot be read: %w", rec.ManifestDigest, rec.ID, err) + } + if !found { + return nil, fmt.Errorf("build %s names the manifest %s and the bus does not hold it", rec.ID, rec.ManifestDigest) + } + return e.Value, nil +} + // casTries is how often a write that lost a compare-and-set reads the key again and tries once more. const casTries = 8 -// chainBuilds is every build of the bucket, whole, in no order. -func chainBuilds(ctx context.Context, c *chain.Chain) ([]Build, error) { - all, err := c.Builds.All(ctx) +// buildsIn is the records of the builds bucket under the keys a filter matches, read in one pass of a watch the server +// filters; the pointers among them are left out. +func buildsIn(ctx context.Context, c *chain.Chain, filter string) ([]chain.Kept[chain.Build], error) { + raw, err := c.Builds.FilteredRaw(ctx, filter) if err != nil { return nil, fmt.Errorf("the builds cannot be read from the bus: %w", err) } - out := make([]Build, 0, len(all)) - for _, k := range all { + out := make([]chain.Kept[chain.Build], 0, len(raw)) + for key, e := range raw { + if !chain.IsRecordKey(key) && !strings.HasPrefix(key, chain.RequestPrefix+".") { + continue + } + k, err := chain.DecodeKept[chain.Build](e) + if err != nil { + return nil, err + } + out = append(out, k) + } + return out, nil +} + +func outcomesOf(kept []chain.Kept[chain.Build]) []Build { + out := make([]Build, 0, len(kept)) + for _, k := range kept { // A request nobody has answered is no build: it has no outcome to read. - if k.Value.State == chain.BuildAsked { + if k.Value.State == chain.BuildAsked || k.Value.State == chain.BuildAnswered { continue } out = append(out, fromChainBuild(k.Value)) } + return out +} + +// chainBuilds is every build of every module, whole, in no order: the modules named by their `newest` pointers, each read by +// its own keys (the manifests kept in the same bucket are never in a filter), and the builds that knew no module. +func chainBuilds(ctx context.Context, c *chain.Chain) ([]Build, error) { + pointers, err := c.Builds.FilteredRaw(ctx, "*.newest") + if err != nil { + return nil, fmt.Errorf("the builds cannot be read from the bus: %w", err) + } + var out []Build + for key := range pointers { + module := chain.ModuleOf(key) + if slices.Contains(chain.ReservedModules, module) { + continue + } + builds, err := chainModuleBuilds(ctx, c, module) + if err != nil { + return nil, err + } + out = append(out, builds...) + } + kept, err := buildsIn(ctx, c, chain.RequestPrefix+".*") + if err != nil { + return nil, err + } + return append(out, outcomesOf(kept)...), nil +} + +// chainModuleBuilds is every build of one module, in no order: the keys `.>` and no other. +func chainModuleBuilds(ctx context.Context, c *chain.Chain, module string) ([]Build, error) { + kept, err := buildsIn(ctx, c, chain.ModuleFilter(module)) + if err != nil { + return nil, err + } + return outcomesOf(kept), nil +} + +// chainBuildsOf is the builds of a module, or of every module when none is named. +func chainBuildsOf(ctx context.Context, c *chain.Chain, module string) ([]Build, error) { + if module == "" { + return chainBuilds(ctx, c) + } + return chainModuleBuilds(ctx, c, module) +} + +// pointedBuilds is, per module, the build a family of pointers (`*.newest`, `*.worked`, `*.held`) names, each read by one +// Get and checked: a survey of every module reads one record of each and not their histories. +func (i *Inventory) pointedBuilds(ctx context.Context, c *chain.Chain, filter string) (map[string]chain.Build, error) { + raw, err := c.Builds.FilteredRaw(ctx, filter) + if err != nil { + return nil, fmt.Errorf("the builds cannot be read from the bus: %w", err) + } + worked := strings.HasSuffix(filter, ".worked") + out := make(map[string]chain.Build, len(raw)) + for key := range raw { + module := chain.ModuleOf(key) + if slices.Contains(chain.ReservedModules, module) { + continue + } + rec, found, err := i.followPointer(ctx, c, module, key, func(b chain.Build) bool { return !worked || b.Failed == "" }) + if err != nil { + return nil, err + } + if found { + out[module] = rec + } + } return out, nil } +// pointedID is the build id a pointer holds: a commit pointer's build, or the id itself. +func pointedID(value []byte) string { + if len(value) > 0 && value[0] == '{' { + var p chain.CommitPointer + if json.Unmarshal(value, &p) == nil { + return p.Build + } + } + return string(value) +} + +// followPointer reads the record a pointer of a module names, and checks it: it is the module's, it is an outcome, and ok +// agrees. **A pointer is a hint** (issue 1121): when it is missing or the record is not what it was for, the module's records +// are read, the pointers made again and said, and the pointer is followed once more. +func (i *Inventory) followPointer(ctx context.Context, c *chain.Chain, module, pointer string, ok func(chain.Build) bool) (chain.Build, bool, error) { + for pass := 0; pass < 2; pass++ { + e, found, err := c.Builds.GetRaw(ctx, pointer) + if err != nil { + return chain.Build{}, false, err + } + if found { + k, has, err := c.Builds.Get(ctx, chain.BuildKey(module, pointedID(e.Value))) + if err != nil { + return chain.Build{}, false, err + } + if has && k.Value.Module == module && k.Value.State == "" && ok(k.Value) { + return k.Value, true, nil + } + } + if pass == 1 { + break + } + if err := i.repairPointers(ctx, c, module, pointer); err != nil { + return chain.Build{}, false, err + } + } + return chain.Build{}, false, nil +} + +// repairPointers reads a module's records and makes its pointers what they say, saying each it changed. +func (i *Inventory) repairPointers(ctx context.Context, c *chain.Chain, module, why string) error { + kept, err := buildsIn(ctx, c, chain.ModuleFilter(module)) + if err != nil { + return err + } + var recs []chain.Build + for _, k := range kept { + if k.Value.State == "" { + recs = append(recs, k.Value) + } + } + points, err := pointsOf(module, recs) + if err != nil { + return err + } + for _, p := range points { + wrote, err := i.pointTo(ctx, c, p.key, p.value, true) + if err != nil { + return err + } + if wrote { + fmt.Printf("builds: the pointer %s was missing or wrong (a lookup of %s) and is repaired from the records of %s\n", p.key, why, module) + } + } + return nil +} + +type pointer struct { + key string + value []byte +} + +// pointsOf are the pointers of a module as its records say: the newest, the newest success, the held, and per commit and per +// reuse key the newest success. The records are in no order. +func pointsOf(module string, recs []chain.Build) ([]pointer, error) { + if len(recs) == 0 { + return nil, nil + } + sort.SliceStable(recs, func(a, b int) bool { return buildNewer(recs[b], recs[a]) }) + points := []pointer{{chain.NewestKey(module), []byte(recs[len(recs)-1].ID)}} + var lastWorked, held *chain.Build + commits, sources := map[string]chain.Build{}, map[string]chain.Build{} + for k := range recs { + rec := recs[k] + if rec.Failed != "" { + continue + } + if held == nil || rec.SourceFingerprint == "" || lastWorked == nil || lastWorked.SourceFingerprint != rec.SourceFingerprint { + held = &recs[k] + } + lastWorked = &recs[k] + if rec.Commit != "" { + commits[rec.Commit] = rec + } + if rec.ReuseKey != "" { + sources[rec.ReuseKey] = rec + } + } + if lastWorked != nil { + points = append(points, pointer{chain.WorkedKey(module), []byte(lastWorked.ID)}, pointer{chain.HeldKey(module), []byte(held.ID)}) + } + for commit, rec := range commits { + body, err := json.Marshal(chain.CommitPointer{Build: rec.ID, Source: rec.SourceFingerprint, Fingerprint: chainFingerprint(fromChainBuild(rec))}) + if err != nil { + return nil, err + } + points = append(points, pointer{chain.CommitKey(module, commit), body}) + } + for key, rec := range sources { + points = append(points, pointer{chain.SourceKey(module, chain.SpellReuse(key)), []byte(rec.ID)}) + } + return points, nil +} + // newestRequestOrder sorts builds as newestRequestFirst does: the newest request first, then the newest record. func newestRequestOrder(builds []Build) { sort.SliceStable(builds, func(a, b int) bool { @@ -177,7 +610,7 @@ func worked(all []Build) []Build { } func (i *Inventory) chainBuildList(ctx context.Context, c *chain.Chain, module string, limit int) ([]Build, error) { - all, err := chainBuilds(ctx, c) + all, err := chainBuildsOf(ctx, c, module) if err != nil { return nil, err } @@ -200,68 +633,77 @@ func (i *Inventory) chainBuildList(ctx context.Context, c *chain.Chain, module s return out, nil } +// chainRecordOf is the record of a build known by its id alone: the pointer `id.` names its module, and the record is +// one Get more. A request no outcome has answered is no build. +func (i *Inventory) chainRecordOf(ctx context.Context, c *chain.Chain, id string) (chain.Build, bool, error) { + ptr, found, err := c.Builds.GetRaw(ctx, chain.IDKey(id)) + if err != nil { + return chain.Build{}, false, err + } + module := "" + if found && string(ptr.Value) != chain.RequestPrefix { + module = string(ptr.Value) + } + k, ok, err := c.Builds.Get(ctx, chain.BuildKey(module, id)) + if err != nil || !ok || k.Value.State == chain.BuildAsked || k.Value.State == chain.BuildAnswered { + return chain.Build{}, false, err + } + return k.Value, true, nil +} + func (i *Inventory) chainBuildByID(ctx context.Context, c *chain.Chain, id string) (Build, bool, error) { - k, ok, err := c.Builds.Get(ctx, chainBuildKey(id)) - if err != nil || !ok || k.Value.State == chain.BuildAsked { - // A request nobody has answered has no outcome to read. + rec, ok, err := i.chainRecordOf(ctx, c, id) + if err != nil || !ok { return Build{}, false, err } - b := fromChainBuild(k.Value) + b := fromChainBuild(rec) return Build{ID: b.ID, Repository: b.Repository, Ref: b.Ref, Module: b.Module, Commit: b.Commit, On: b.On, Failed: b.Failed, Made: b.Made, Path: b.Path, Asked: b.Asked, At: b.At}, true, nil } func (i *Inventory) chainBuiltAgainst(ctx context.Context, c *chain.Chain) (map[string][]string, error) { - all, err := chainBuilds(ctx, c) + worked, err := i.pointedBuilds(ctx, c, "*.worked") if err != nil { return nil, err } - ok := worked(all) - byModule(ok) out := map[string][]string{} - seen := map[string]bool{} - for _, b := range ok { - if seen[b.Module] { - continue - } - seen[b.Module] = true + for module, b := range worked { if len(b.Against) > 0 { - out[b.Module] = b.Against + out[module] = b.Against } } return out, nil } func (i *Inventory) chainReadRepositories(ctx context.Context, c *chain.Chain) (map[string][]ReadRepository, error) { - all, err := chainBuilds(ctx, c) - if err != nil { - return nil, err - } // The newest build of each module whatever its outcome: a failed one newer than the newest that worked leaves that - // one's build source stale. + // one's build source stale. Each is one record read through its pointer. type newestOf struct { at time.Time failed bool } + tried, err := i.pointedBuilds(ctx, c, "*.newest") + if err != nil { + return nil, err + } newest := map[string]newestOf{} - tried := make([]Build, 0, len(all)) - for _, b := range all { - if b.Module != "" { - tried = append(tried, b) - } + for module, rec := range tried { + b := fromChainBuild(rec) + newest[module] = newestOf{b.AskedOrAt(), b.Failed != ""} } - byModule(tried) - for _, b := range tried { - if _, seen := newest[b.Module]; !seen { - newest[b.Module] = newestOf{b.AskedOrAt(), b.Failed != ""} - } + workedRecs, err := i.pointedBuilds(ctx, c, "*.worked") + if err != nil { + return nil, err } + ok := make([]Build, 0, len(workedRecs)) + for _, rec := range workedRecs { + ok = append(ok, fromChainBuild(rec)) + } + byModule(ok) owns, err := i.moduleSources(ctx) if err != nil { return nil, err } - ok := worked(all) - byModule(ok) read := map[string][]ReadRepository{} seen := map[string]bool{} for _, b := range ok { @@ -333,12 +775,37 @@ func (i *Inventory) chainAnnounceable(ctx context.Context, c *chain.Chain) ([]Bu } seen[k] = true out = append(out, Build{ID: b.ID, Repository: b.Repository, Ref: b.Ref, Module: b.Module, Commit: b.Commit, - On: b.On, Failed: b.Failed, Made: b.Made, Path: b.Path, Manifest: b.Manifest, Against: b.Against, At: b.At}) + On: b.On, Failed: b.Failed, Made: b.Made, Path: b.Path, Manifest: b.Manifest, ManifestDigest: b.ManifestDigest, + Against: b.Against, At: b.At}) } sort.SliceStable(out, func(a, b int) bool { return out[a].At.Before(out[b].At) }) + if err := fillManifests(ctx, c, out); err != nil { + return nil, err + } return out, nil } +// fillManifests reads the manifests the builds name, each digest once. +func fillManifests(ctx context.Context, c *chain.Chain, builds []Build) error { + seen := map[string][]byte{} + for k := range builds { + d := builds[k].ManifestDigest + if d == "" || len(builds[k].Manifest) > 0 { + continue + } + raw, ok := seen[d] + if !ok { + var err error + if raw, err = manifestOfRecord(ctx, c, chain.Build{ID: builds[k].ID, ManifestDigest: d}); err != nil { + return err + } + seen[d] = raw + } + builds[k].Manifest = raw + } + return nil +} + // failedAtAGate is the builds that failed their gate. The gate's verdicts stay in their table (issue 1097). func (i *Inventory) failedAtAGate(ctx context.Context) (map[string]bool, error) { rows, err := i.store.Pool().Query(ctx, `select build from build_gate where verdict = 'failed'`) @@ -358,7 +825,7 @@ func (i *Inventory) failedAtAGate(ctx context.Context) (map[string]bool, error) } func (i *Inventory) chainSourceRows(ctx context.Context, c *chain.Chain, module string) (map[string][]sourceRow, []string, error) { - all, err := chainBuilds(ctx, c) + all, err := chainBuildsOf(ctx, c, module) if err != nil { return nil, nil, err } @@ -382,7 +849,7 @@ func (i *Inventory) chainSourceRows(ctx context.Context, c *chain.Chain, module order = append(order, b.Module) } out[b.Module] = append(out[b.Module], sourceRow{id: b.ID, commit: b.Commit, fingerprint: b.SourceFingerprint, - made: made, manifest: b.Manifest, failedGate: failed[b.ID]}) + made: made, manifest: b.Manifest, manifestDigest: b.ManifestDigest, failedGate: failed[b.ID]}) } return out, order, nil } @@ -400,56 +867,105 @@ func canonicalJSON(raw []byte) []byte { return out } -// chainFingerprint is what a build would put on a machine, hashed: its artifacts and its manifest. +// chainFingerprint is what a build would put on a machine, hashed: its artifacts and its manifest, the manifest by its +// digest (the digest of its canonical bytes, which the manifests bucket keys it by). func chainFingerprint(b Build) string { made, _ := json.Marshal(b.Made) - sum := sha256.Sum256(append(append(canonicalJSON(made), 0), canonicalJSON(b.Manifest)...)) + digest := b.ManifestDigest + if digest == "" { + if body := manifestBytes(b.Manifest); body != nil { + digest = chain.ManifestKey(body) + } + } + sum := sha256.Sum256(append(append(canonicalJSON(made), 0), digest...)) return hex.EncodeToString(sum[:]) } +// commitPointers are, per commit of a module (every module when none is named), the pointer to the newest successful build +// from it: a read of the keys `.commit.*` and no record. +func commitPointers(ctx context.Context, c *chain.Chain, module string) (map[string]map[string]chain.CommitPointer, error) { + filter := "*.commit.*" + if module != "" { + filter = module + ".commit.*" + } + raw, err := c.Builds.FilteredRaw(ctx, filter) + if err != nil { + return nil, fmt.Errorf("the builds cannot be read from the bus: %w", err) + } + out := map[string]map[string]chain.CommitPointer{} + for key, e := range raw { + var p chain.CommitPointer + if err := json.Unmarshal(e.Value, &p); err != nil { + return nil, fmt.Errorf("the pointer %s is not readable: %w", key, err) + } + m, commit, _ := strings.Cut(key, ".commit.") + if out[m] == nil { + out[m] = map[string]chain.CommitPointer{} + } + out[m][commit] = p + } + return out, nil +} + func (i *Inventory) chainBuildFingerprints(ctx context.Context, c *chain.Chain, module string) (map[string]string, error) { - all, err := chainBuilds(ctx, c) + pointers, err := commitPointers(ctx, c, module) if err != nil { return nil, err } - var ok []Build - for _, b := range all { - if b.Module == module && b.Failed == "" && b.Commit != "" { - ok = append(ok, b) - } - } - sortByRecorded(ok) - if len(ok) > 50 { - ok = ok[:50] - } out := map[string]string{} - for _, b := range ok { - if _, seen := out[b.Commit]; !seen { - out[b.Commit] = chainFingerprint(b) - } + for commit, p := range pointers[module] { + out[commit] = p.Fingerprint } return out, nil } func (i *Inventory) chainFingerprints(ctx context.Context, c *chain.Chain) (map[string]map[string]string, error) { - all, err := chainBuilds(ctx, c) + pointers, err := commitPointers(ctx, c, "") if err != nil { return nil, err } - var ok []Build - for _, b := range all { - if b.Module != "" && b.Failed == "" && b.Commit != "" { - ok = append(ok, b) + out := map[string]map[string]string{} + for m, commits := range pointers { + out[m] = map[string]string{} + for commit, p := range commits { + out[m][commit] = p.Fingerprint } } - sortByRecorded(ok) - out := map[string]map[string]string{} - for _, b := range ok { - if out[b.Module] == nil { - out[b.Module] = map[string]string{} + return out, nil +} + +// chainBuildSourceFingerprints is chainSourceFingerprints for one module: commit → fingerprint, from its commit pointers. +func (i *Inventory) chainBuildSourceFingerprints(ctx context.Context, c *chain.Chain, module string) (map[string]string, error) { + pointers, err := commitPointers(ctx, c, module) + if err != nil { + return nil, err + } + out := map[string]string{} + for commit, p := range pointers[module] { + if p.Source != "" { + out[commit] = p.Source } - if _, seen := out[b.Module][b.Commit]; !seen { - out[b.Module][b.Commit] = chainFingerprint(b) + } + return out, nil +} + +// chainSourceFingerprints is, per module and commit, the source fingerprint of the newest successful build from it that has +// one, read from the commit pointers. +func (i *Inventory) chainSourceFingerprints(ctx context.Context, c *chain.Chain) (map[string]map[string]string, error) { + pointers, err := commitPointers(ctx, c, "") + if err != nil { + return nil, err + } + out := map[string]map[string]string{} + for m, commits := range pointers { + for commit, p := range commits { + if p.Source == "" { + continue + } + if out[m] == nil { + out[m] = map[string]string{} + } + out[m][commit] = p.Source } } return out, nil @@ -465,33 +981,45 @@ func sortByRecorded(builds []Build) { }) } +// chainBuildOf is the id of the newest successful build of a module from a commit: its commit pointer, and the record it names. func (i *Inventory) chainBuildOf(ctx context.Context, c *chain.Chain, module, commit string) (string, error) { - all, err := chainBuilds(ctx, c) - if err != nil { + rec, found, err := i.followPointer(ctx, c, module, chain.CommitKey(module, commit), + func(b chain.Build) bool { return b.Failed == "" && b.Commit == commit }) + if err != nil || !found { return "", err } - var ok []Build - for _, b := range all { - if b.Module == module && b.Commit == commit && b.Failed == "" { - ok = append(ok, b) - } + return rec.ID, nil +} + +// chainBuildOfSource is the id of the newest successful build of a module with a reuse key (ADR 0350 §1): its source +// pointer, and the record it names. What a reuse asks. +func (i *Inventory) chainBuildOfSource(ctx context.Context, c *chain.Chain, module, reuseKey string) (string, error) { + rec, found, err := i.followPointer(ctx, c, module, chain.SourceKey(module, chain.SpellReuse(reuseKey)), + func(b chain.Build) bool { return b.Failed == "" && b.ReuseKey == reuseKey }) + if err != nil || !found { + return "", err } - sortByRecorded(ok) - if len(ok) == 0 { - return "", nil - } - return ok[0].ID, nil + return rec.ID, nil } // chainManifestBuild is the newest successful build of a module from a commit (or one it abbreviates) that kept a manifest. +// A whole commit is its pointer and the record; an abbreviation reads the module's builds. func (i *Inventory) chainManifestBuild(ctx context.Context, c *chain.Chain, module, commit string) (string, []byte, bool, error) { - all, err := chainBuilds(ctx, c) + kept := func(b chain.Build) bool { return b.ManifestDigest != "" || manifestBytes(b.Manifest) != nil } + if rec, found, err := i.followPointer(ctx, c, module, chain.CommitKey(module, commit), + func(b chain.Build) bool { return b.Failed == "" && b.Commit == commit }); err != nil { + return "", nil, false, err + } else if found && kept(rec) { + raw, err := manifestOfRecord(ctx, c, rec) + return rec.ID, raw, err == nil, err + } + all, err := chainModuleBuilds(ctx, c, module) if err != nil { return "", nil, false, err } var ok []Build for _, b := range all { - if b.Module == module && b.Failed == "" && len(b.Manifest) > 0 && string(bytes.TrimSpace(b.Manifest)) != "null" && + if b.Failed == "" && (b.ManifestDigest != "" || manifestBytes(b.Manifest) != nil) && (b.Commit == commit || strings.HasPrefix(b.Commit, commit)) { ok = append(ok, b) } @@ -500,16 +1028,17 @@ func (i *Inventory) chainManifestBuild(ctx context.Context, c *chain.Chain, modu if len(ok) == 0 { return "", nil, false, nil } - return ok[0].ID, ok[0].Manifest, true, nil + raw, err := manifestOfRecord(ctx, c, chain.Build{ID: ok[0].ID, Manifest: ok[0].Manifest, ManifestDigest: ok[0].ManifestDigest}) + return ok[0].ID, raw, err == nil, err } // chainManifestOf is the manifest a build was recorded with. func (i *Inventory) chainManifestOf(ctx context.Context, c *chain.Chain, id string) ([]byte, error) { - k, ok, err := c.Builds.Get(ctx, chainBuildKey(id)) + rec, ok, err := i.chainRecordOf(ctx, c, id) if err != nil || !ok { return nil, err } - return []byte(k.Value.Manifest), nil + return manifestOfRecord(ctx, c, rec) } // chainRecentMade is the artifacts and bases of the KeptBuilds newest successful builds of each registered module. @@ -568,3 +1097,156 @@ func (i *Inventory) chainEveryReferenceMade(ctx context.Context, c *chain.Chain) }) return ok, nil } + +// chainHeldMade is, per module, what the build standing for its newest successful build made. Read from the pointers: the +// module's `worked` (its newest success) and `held` (the first build of the run of one source that it stands on), one record +// of each. A build that failed its gate ends a run (issue 280), and the pointer cannot say where; then the module's builds +// are read. +func (i *Inventory) chainHeldMade(ctx context.Context, c *chain.Chain) (map[string][]Artifact, error) { + worked, err := i.pointedBuilds(ctx, c, "*.worked") + if err != nil { + return nil, err + } + held, err := i.pointedBuilds(ctx, c, "*.held") + if err != nil { + return nil, err + } + failed, err := i.failedAtAGate(ctx) + if err != nil { + return nil, err + } + out := map[string][]Artifact{} + for module, w := range worked { + stand := w + if h, ok := held[module]; ok && !failed[w.ID] { + stand = h + } + if failed[stand.ID] && stand.ID != w.ID { + rows, _, err := i.chainSourceRows(ctx, c, module) + if err != nil { + return nil, err + } + if builds := rows[module]; len(builds) > 0 { + var made []Artifact + if err := json.Unmarshal(builds[standing(builds, 0)].made, &made); err == nil { + out[module] = made + } + } + continue + } + out[module] = fromChainBuild(stand).Made + } + return out, nil +} + +// LayoutReport is what the layout of the builds bucket did. +type LayoutReport struct { + // Records is the keys of before it read, Written the keys this run made (records, pointers, manifests) and Present the + // records already under their new key. + Records, Written, Present int +} + +// KindBuildsLayout is the kind chain_moved names the layout of the builds bucket under their modules by (issue 1121). +const KindBuildsLayout = "builds-layout" + +// layoutBuilds lays the builds of the bucket out under their modules: each record of a key that is its id alone is written +// under `.` (a request, and a build that knew no module, under `req.`), its manifest moved to the manifests +// bucket, and the pointers made. **Idempotent**: a key that exists is already moved, and a pointer that holds what it should +// is not written. The old keys stay (ADR 0408 §4) and are not read again. +func (i *Inventory) layoutBuilds(ctx context.Context, c *chain.Chain, say func(string, ...any)) (LayoutReport, error) { + if say == nil { + say = func(string, ...any) {} + } + var r LayoutReport + legacy, err := c.Builds.FilteredRaw(ctx, "*") + if err != nil { + return r, fmt.Errorf("the builds of before the layout cannot be read: %w", err) + } + ids := make([]string, 0, len(legacy)) + for key := range legacy { + ids = append(ids, key) + } + sort.Strings(ids) + byModule := map[string][]chain.Build{} + manifestsDone := map[string]bool{} + for _, key := range ids { + k, err := chain.DecodeKept[chain.Build](legacy[key]) + if err != nil { + return r, err + } + rec := k.Value + r.Records++ + if r.Records%1000 == 0 { + say("laying the builds out: %d of %d records, %d keys written so far", r.Records, len(ids), r.Written) + } + // A manifest already written in this run is not written again. + digest := "" + if body := manifestBytes([]byte(rec.Manifest)); body != nil && c.Kept { + digest = chain.ManifestKey(body) + if !manifestsDone[digest] { + if _, err := c.Manifests.CreateRaw(ctx, digest, body); err == nil { + r.Written++ + } else if !errors.Is(err, chain.ErrExists) { + return r, fmt.Errorf("the manifest of %s is not kept: %w", key, err) + } + manifestsDone[digest] = true + } + } + if digest != "" { + rec.Manifest, rec.ManifestDigest = nil, digest + } + target := chain.BuildKey(rec.Module, rec.ID) + if rec.State == chain.BuildAsked { + target = chain.RequestKey(rec.ID) + } + switch _, err := c.Builds.Create(ctx, target, rec); { + case err == nil: + r.Written++ + case errors.Is(err, chain.ErrExists): + r.Present++ + default: + return r, fmt.Errorf("%s is not written under %s: %w", key, target, err) + } + if rec.State != chain.BuildAsked { + wrote, err := i.pointTo(ctx, c, chain.IDKey(rec.ID), idPointer(rec.Module), false) + if err != nil { + return r, err + } + if wrote { + r.Written++ + } + } + if rec.State != chain.BuildAsked && rec.Module != "" && rec.Request != nil { + // The request beside its outcome, so the requests are read from their own prefix. + ans := chain.Build{ID: rec.ID, State: chain.BuildAnswered, Repository: rec.Repository, Ref: rec.Ref, Path: rec.Path, + Asked: rec.Asked, Request: rec.Request, Module: rec.Module, Failed: rec.Failed, At: rec.At} + switch _, err := c.Builds.Create(ctx, chain.RequestKey(rec.ID), ans); { + case err == nil: + r.Written++ + case errors.Is(err, chain.ErrExists): + default: + return r, err + } + } + if rec.State != chain.BuildAsked && rec.Module != "" { + byModule[rec.Module] = append(byModule[rec.Module], rec) + } + } + for module, recs := range byModule { + points, err := pointsOf(module, recs) + if err != nil { + return r, err + } + for _, p := range points { + wrote, err := i.pointTo(ctx, c, p.key, p.value, true) + if err != nil { + return r, err + } + if wrote { + r.Written++ + } + } + } + say("laid out the builds under their modules: %d records read, %d keys written, %d records already there", r.Records, r.Written, r.Present) + return r, nil +} diff --git a/internal/inventory/chain_builds_layout_test.go b/internal/inventory/chain_builds_layout_test.go new file mode 100644 index 00000000..a27e8f3e --- /dev/null +++ b/internal/inventory/chain_builds_layout_test.go @@ -0,0 +1,398 @@ +package inventory + +import ( + "context" + "encoding/json" + "fmt" + "reflect" + "sort" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/chain" +) + +// issue 1121: a build is under its module, so a lookup reads the keys it needs and no others. + +type readCounting struct { + chain.Backend + gets, watches atomic.Int32 +} + +func (r *readCounting) Get(ctx context.Context, key string) (chain.Entry, bool, error) { + r.gets.Add(1) + return r.Backend.Get(ctx, key) +} + +func (r *readCounting) Watch(ctx context.Context) (<-chan chain.Entry, error) { + r.watches.Add(1) + return r.Backend.Watch(ctx) +} + +func (r *readCounting) WatchFiltered(ctx context.Context, f string) (<-chan chain.Entry, error) { + r.watches.Add(1) + return r.Backend.WatchFiltered(ctx, f) +} + +func layoutChain(builds chain.Backend) (*Inventory, *chain.Chain) { + c := chain.New(builds, chain.NewMemory(), chain.NewMemory(), func(context.Context) error { return nil }). + WithManifests(builds) + inv := &Inventory{moved: &movedKinds{}} + inv.UseChain(c) + return inv, c +} + +var layoutAt = time.Date(2026, 10, 18, 12, 0, 0, 0, time.UTC) + +func layoutBuild(module string, n int, commit, source string) Build { + return Build{ID: fmt.Sprintf("build-%d", n), Repository: "novox/" + module, Module: module, Commit: commit, + SourceFingerprint: source, ReuseKey: "v1:" + source, Made: []Artifact{{Name: module, Kind: "image", Reference: fmt.Sprintf("r/%s@sha256:%d", module, n)}}, + Manifest: []byte(fmt.Sprintf(`{"module":%q,"n":%d}`, module, n)), At: layoutAt.Add(time.Duration(n) * time.Minute)} +} + +func TestALookupOfOneModuleReadsThatModulesKeysOnly(t *testing.T) { + mem := chain.NewMemory() + inv, c := layoutChain(mem) + ctx := context.Background() + for n := 1; n <= 3; n++ { + for _, m := range []string{"alpha", "beta"} { + if err := inv.chainRecordBuild(ctx, c, layoutBuild(m, n, fmt.Sprintf("c%d", n), fmt.Sprintf("s%d", n))); err != nil { + t.Fatal(err) + } + } + } + got, err := inv.chainBuildList(ctx, c, "alpha", 10) + if err != nil || len(got) != 3 { + t.Fatalf("alpha's builds %+v: %v", got, err) + } + if want := map[string]int{"alpha.build.>": 1}; !reflect.DeepEqual(mem.Filtered(), want) { + t.Fatalf("the list of one module watched %v; want only %v", mem.Filtered(), want) + } + if _, err := inv.chainBuildFingerprints(ctx, c, "alpha"); err != nil { + t.Fatal(err) + } + for f := range mem.Filtered() { + if f != "alpha.build.>" && f != "alpha.commit.*" { + t.Errorf("a lookup of alpha watched %q", f) + } + } +} + +func TestReuseBySourceFingerprintAndTheBuildOfACommitAreOneGet(t *testing.T) { + counting := &readCounting{Backend: chain.NewMemory()} + inv, c := layoutChain(counting) + ctx := context.Background() + for n := 1; n <= 20; n++ { + if err := inv.chainRecordBuild(ctx, c, layoutBuild("alpha", n, fmt.Sprintf("c%d", n), fmt.Sprintf("s%d", n))); err != nil { + t.Fatal(err) + } + } + counting.gets.Store(0) + counting.watches.Store(0) + id, err := inv.chainBuildOfSource(ctx, c, "alpha", "v1:s7") + if err != nil || id != "build-7" { + t.Fatalf("source s7 is %q: %v", id, err) + } + if counting.gets.Load() != 2 || counting.watches.Load() != 0 { + t.Fatalf("the reuse lookup made %d gets and %d watches; want 2 (the pointer, the record it names) and 0", counting.gets.Load(), counting.watches.Load()) + } + counting.gets.Store(0) + if id, err = inv.chainBuildOf(ctx, c, "alpha", "c19"); err != nil || id != "build-19" { + t.Fatalf("commit c19 is %q: %v", id, err) + } + if counting.gets.Load() != 2 || counting.watches.Load() != 0 { + t.Fatalf("the build of a commit made %d gets and %d watches; want 2 and 0", counting.gets.Load(), counting.watches.Load()) + } +} + +func TestABuildWrittenIsFoundByCommitSourceIDAndRequest(t *testing.T) { + inv, c := layoutChain(chain.NewMemory()) + ctx := context.Background() + ask := BuildRequest{ID: "build-5", Repository: "novox/alpha", Seat: "mesh-build-machine", At: layoutAt} + if err := inv.chainRecordBuildRequest(ctx, c, ask); err != nil { + t.Fatal(err) + } + if _, found, _ := inv.chainBuildByID(ctx, c, "build-5"); found { + t.Fatal("a request nobody answered was read as a build") + } + b := layoutBuild("alpha", 5, "c5", "s5") + if err := inv.chainRecordBuild(ctx, c, b); err != nil { + t.Fatal(err) + } + if id, _ := inv.chainBuildOf(ctx, c, "alpha", "c5"); id != "build-5" { + t.Errorf("by commit: %q", id) + } + if id, _ := inv.chainBuildOfSource(ctx, c, "alpha", "v1:s5"); id != "build-5" { + t.Errorf("by source: %q", id) + } + got, found, err := inv.chainBuildByID(ctx, c, "build-5") + if err != nil || !found || got.Module != "alpha" { + t.Fatalf("by id: %+v %v %v", got, found, err) + } + // The manifest is kept once, apart from the record, and read back whole. + raw, err := inv.chainManifestOf(ctx, c, "build-5") + if err != nil || !json.Valid(raw) || string(raw) != string(canonicalJSON(b.Manifest)) { + t.Fatalf("manifest %q: %v", raw, err) + } + k, _, _ := c.Builds.Get(ctx, "alpha.build.build-5") + if len(k.Value.Manifest) != 0 || k.Value.ManifestDigest == "" { + t.Errorf("the record keeps the manifest itself: %+v", k.Value) + } + if k.Value.Request == nil { + t.Error("the outcome lost the request it answers") + } + reqs, err := inv.chainRequests(ctx, c) + if err != nil || len(reqs) != 1 || !reqs[0].Heard || reqs[0].Module != "alpha" { + t.Fatalf("requests %+v: %v", reqs, err) + } +} + +func legacyBuild(n int, module, commit string) chain.Build { + b := toChainBuild(layoutBuild(module, n, commit, "s"+commit)) + return b +} + +func keysOf(t *testing.T, mem chain.Backend) []string { + t.Helper() + keys, err := mem.Keys(context.Background()) + if err != nil { + t.Fatal(err) + } + sort.Strings(keys) + return keys +} + +// The layout of the bucket's keys is made once from the keys of before it: run again it changes nothing, the old keys stay, +// and the pointers say the newest. +func TestTheLayoutMigrationIsIdempotentAndKeepsTheOldKeys(t *testing.T) { + mem := chain.NewMemory() + inv, c := layoutChain(mem) + ctx := context.Background() + for n, m := range []string{"alpha", "alpha", "beta", "alpha"} { + old := legacyBuild(n+1, m, fmt.Sprintf("c%d", n+1)) + if _, err := c.Builds.Create(ctx, old.ID, old); err != nil { + t.Fatal(err) + } + } + failed := legacyBuild(9, "alpha", "c9") + failed.Failed = "broke" + failed.SourceFingerprint = "" + if _, err := c.Builds.Create(ctx, failed.ID, failed); err != nil { + t.Fatal(err) + } + asked := chain.Build{ID: "build-10", State: chain.BuildAsked, Repository: "novox/alpha", Asked: layoutAt, + Request: &chain.Request{Repository: "novox/alpha", At: layoutAt}} + if _, err := c.Builds.Create(ctx, asked.ID, asked); err != nil { + t.Fatal(err) + } + first, err := inv.layoutBuilds(ctx, c, nil) + if err != nil || first.Written == 0 { + t.Fatalf("first run %+v: %v", first, err) + } + keys := keysOf(t, mem) + second, err := inv.layoutBuilds(ctx, c, nil) + if err != nil || second.Written != 0 { + t.Fatalf("second run %+v: %v", second, err) + } + if again := keysOf(t, mem); !reflect.DeepEqual(keys, again) { + t.Fatalf("keys changed on the second run:\n%v\n%v", keys, again) + } + has := func(k string) bool { return containsKey(keys, k) } + for _, k := range []string{"build-1", "build-9", "alpha.build.build-1", "alpha.build.build-4", "beta.build.build-3", "alpha.build.build-9", "req.build-10", + "id.build-3", "alpha.newest", "alpha.worked", "alpha.held", "alpha.commit.c4", "alpha.source.v1-sc4", "beta.worked"} { + if !has(k) { + t.Errorf("no key %s in %v", k, keys) + } + } + if e, _, _ := mem.Get(ctx, "alpha.newest"); string(e.Value) != "build-9" { + t.Errorf("alpha.newest = %q; want the failed build-9", e.Value) + } + if e, _, _ := mem.Get(ctx, "alpha.worked"); string(e.Value) != "build-4" { + t.Errorf("alpha.worked = %q; want build-4", e.Value) + } + if id, _ := inv.chainBuildOf(ctx, c, "alpha", "c4"); id != "build-4" { + t.Errorf("by commit after the layout: %q", id) + } + // No record is read from an old key. + list, err := inv.chainBuildList(ctx, c, "alpha", 10) + if err != nil || len(list) != 4 { + t.Fatalf("alpha's builds after the layout: %d %v", len(list), err) + } +} + +func containsKey(s []string, k string) bool { + for _, x := range s { + if x == k { + return true + } + } + return false +} + +// The manifest is kept once under its digest in the same bucket, and the moved record keeps none. +func TestAMovedRecordHoldsNoManifestAndItsManifestIsUnderItsDigest(t *testing.T) { + mem := chain.NewMemory() + inv, c := layoutChain(mem) + ctx := context.Background() + old := legacyBuild(1, "alpha", "c1") + if _, err := c.Builds.Create(ctx, old.ID, old); err != nil { + t.Fatal(err) + } + if _, err := inv.layoutBuilds(ctx, c, nil); err != nil { + t.Fatal(err) + } + k, found, err := c.Builds.Get(ctx, "alpha.build.build-1") + if err != nil || !found || len(k.Value.Manifest) != 0 || k.Value.ManifestDigest == "" { + t.Fatalf("moved record %+v %v %v", k.Value, found, err) + } + if !strings.HasPrefix(k.Value.ManifestDigest, "manifest.sha256-") { + t.Errorf("the digest %q is not a key under manifest.", k.Value.ManifestDigest) + } + e, found, _ := mem.Get(ctx, k.Value.ManifestDigest) + if !found || string(e.Value) != string(canonicalJSON([]byte(old.Manifest))) { + t.Errorf("the manifest under its digest is %q (found %v)", e.Value, found) + } + if got := chain.ManifestKey(e.Value); got != k.Value.ManifestDigest { + t.Errorf("the digest does not name the bytes: %s", got) + } +} + +// A pointer is a hint: a record written without its pointers is found by reading the module's records, and the pointers are +// repaired; a pointer that names a record that is not what it was for is not followed. +func TestAMissingOrWrongPointerFallsBackToTheModulesRecordsAndIsRepaired(t *testing.T) { + mem := chain.NewMemory() + inv, c := layoutChain(mem) + ctx := context.Background() + for n := 1; n <= 3; n++ { + if err := inv.chainRecordBuild(ctx, c, layoutBuild("alpha", n, fmt.Sprintf("c%d", n), fmt.Sprintf("s%d", n))); err != nil { + t.Fatal(err) + } + } + // A record whose pointers the writer never made (a crash between the record and its pointers). + lost := toChainBuild(layoutBuild("alpha", 4, "c4", "s4")) + if _, err := c.Builds.Create(ctx, "alpha.build.build-4", lost); err != nil { + t.Fatal(err) + } + if id, err := inv.chainBuildOf(ctx, c, "alpha", "c4"); err != nil || id != "build-4" { + t.Fatalf("the build of c4 is %q: %v", id, err) + } + if e, found, _ := mem.Get(ctx, "alpha.commit.c4"); !found || pointedID(e.Value) != "build-4" { + t.Errorf("the commit pointer was not repaired: %q %v", e.Value, found) + } + if id, _ := inv.chainBuildOfSource(ctx, c, "alpha", "v1:s4"); id != "build-4" { + t.Errorf("the source pointer was not repaired: %q", id) + } + // A pointer to the wrong build is not followed. + e, _, _ := mem.Get(ctx, "alpha.commit.c2") + wrong, _ := json.Marshal(chain.CommitPointer{Build: "build-1"}) + if _, err := mem.Update(ctx, "alpha.commit.c2", wrong, e.Revision); err != nil { + t.Fatal(err) + } + if id, err := inv.chainBuildOf(ctx, c, "alpha", "c2"); err != nil || id != "build-2" { + t.Fatalf("with a wrong pointer the build of c2 is %q: %v", id, err) + } + if e, _, _ := mem.Get(ctx, "alpha.commit.c2"); pointedID(e.Value) != "build-2" { + t.Errorf("the wrong pointer stayed: %s", e.Value) + } + // A commit nothing built is none. + if id, err := inv.chainBuildOf(ctx, c, "alpha", "nope"); err != nil || id != "" { + t.Errorf("an unbuilt commit is %q: %v", id, err) + } +} + +type touchRecording struct { + chain.Backend + mu sync.Mutex + touched []string +} + +func (r *touchRecording) note(s string) { r.mu.Lock(); r.touched = append(r.touched, s); r.mu.Unlock() } +func (r *touchRecording) Get(ctx context.Context, key string) (chain.Entry, bool, error) { + r.note("get " + key) + return r.Backend.Get(ctx, key) +} +func (r *touchRecording) Watch(ctx context.Context) (<-chan chain.Entry, error) { + r.note("watch >") + return r.Backend.Watch(ctx) +} +func (r *touchRecording) WatchFiltered(ctx context.Context, f string) (<-chan chain.Entry, error) { + r.note("watch " + f) + return r.Backend.WatchFiltered(ctx, f) +} +func (r *touchRecording) Keys(ctx context.Context) ([]string, error) { + r.note("keys") + return r.Backend.Keys(ctx) +} + +// Once the layout is made, no read touches a key that is a build's id alone, or the bucket whole. +func TestAfterTheLayoutNoReadTouchesABareBuildIDKey(t *testing.T) { + rec := &touchRecording{Backend: chain.NewMemory()} + inv, c := layoutChain(rec) + ctx := context.Background() + old := legacyBuild(1, "alpha", "c1") + if _, err := c.Builds.Create(ctx, old.ID, old); err != nil { + t.Fatal(err) + } + if _, err := inv.layoutBuilds(ctx, c, nil); err != nil { + t.Fatal(err) + } + if err := inv.chainRecordBuild(ctx, c, layoutBuild("alpha", 2, "c2", "s2")); err != nil { + t.Fatal(err) + } + rec.mu.Lock() + rec.touched = nil + rec.mu.Unlock() + _, _ = inv.chainBuildList(ctx, c, "alpha", 5) + _, _ = inv.chainBuildList(ctx, c, "", 5) + _, _, _ = inv.chainBuildByID(ctx, c, "build-1") + _, _, _ = inv.chainBuildByID(ctx, c, "build-2") + _, _ = inv.chainBuildOf(ctx, c, "alpha", "c1") + _, _ = inv.chainBuildFingerprints(ctx, c, "alpha") + _, _ = inv.chainFingerprints(ctx, c) + _, _ = inv.chainManifestOf(ctx, c, "build-1") + _, _, _, _ = inv.chainManifestBuild(ctx, c, "alpha", "c1") + _, _ = chainBuilds(ctx, c) + _, _ = inv.chainRequests(ctx, c) + _, _ = inv.chainBuiltAgainst(ctx, c) + rec.mu.Lock() + defer rec.mu.Unlock() + for _, what := range rec.touched { + verb, key, _ := strings.Cut(what, " ") + if what == "keys" || what == "watch >" || what == "watch *" || (verb == "get" && !strings.Contains(key, ".")) { + t.Errorf("a read touched %q", what) + } + } +} + +// A pointer follows the newest build, not the last write: two builds overlap and the older, asked first, finishes last. +func TestAnOlderBuildFinishingLastLeavesThePointersOnTheNewer(t *testing.T) { + mem := chain.NewMemory() + inv, c := layoutChain(mem) + ctx := context.Background() + older := layoutBuild("alpha", 1, "c1", "s1") + older.Asked = layoutAt + older.At = layoutAt.Add(20 * time.Minute) // finished last + newer := layoutBuild("alpha", 2, "c1", "s2") + newer.Asked = layoutAt.Add(5 * time.Minute) + newer.At = layoutAt.Add(10 * time.Minute) + for _, b := range []Build{newer, older} { + if err := inv.chainRecordBuild(ctx, c, b); err != nil { + t.Fatal(err) + } + } + for _, key := range []string{"alpha.newest", "alpha.worked"} { + if e, _, _ := mem.Get(ctx, key); string(e.Value) != "build-2" { + t.Errorf("%s = %q; want the newer build-2", key, e.Value) + } + } + if e, _, _ := mem.Get(ctx, "alpha.commit.c1"); pointedID(e.Value) != "build-2" { + t.Errorf("the commit pointer = %s; want build-2", e.Value) + } + // The older build is still found by what only it has. + if id, _ := inv.chainBuildOfSource(ctx, c, "alpha", "v1:s1"); id != "build-1" { + t.Errorf("the older build's own reuse key finds %q", id) + } +} diff --git a/internal/inventory/chain_move.go b/internal/inventory/chain_move.go index 281ce98f..acf7bde5 100644 --- a/internal/inventory/chain_move.go +++ b/internal/inventory/chain_move.go @@ -112,7 +112,7 @@ func (i *Inventory) movers() []mover { if err != nil { return nil, 0, err } - out = append(out, moveRow{key: chainBuildKey(id), value: *byID[id], size: len(body)}) + out = append(out, moveRow{key: legacyBuildKey(id), value: *byID[id], size: len(body)}) } return out, len(builds) + len(requests), nil }, @@ -360,6 +360,12 @@ func (i *Inventory) MoveChain(ctx context.Context, say func(format string, args return out, err } out = append(out, r) + if m.kind == KindBuilds { + // The builds are laid out under their modules before anything reads them (issue 1121). + if err := i.moveBuildsLayout(ctx, say); err != nil { + return out, err + } + } } return out, nil } @@ -661,3 +667,35 @@ func settleAssignment(ctx context.Context, c *chain.Chain, row moveRow) error { } return fmt.Errorf("the assignment %s kept moving under the write", row.key) } + +// moveBuildsLayout lays the builds out under their modules once, and records the kind builds-layout in chain_moved. A key +// the old layout made stays (ADR 0408 §4). +func (i *Inventory) moveBuildsLayout(ctx context.Context, say func(string, ...any)) error { + moved, err := i.ChainMoved(ctx) + if err != nil { + return err + } + if _, done := moved[KindBuildsLayout]; done { + return nil + } + report, err := i.layoutBuilds(ctx, i.attached(), say) + if err != nil { + return err + } + counts, _ := json.Marshal(map[string]int{KindBuildsLayout: report.Records}) + largest, _ := json.Marshal(map[string]int{KindBuildsLayout: 0}) + if _, err := i.store.Pool().Exec(ctx, + `insert into chain_moved (counts, largest) values ($1, $2) + on conflict (only_one) do update + set counts = chain_moved.counts || excluded.counts, largest = chain_moved.largest || excluded.largest, + moved_at = now()`, counts, largest); err != nil { + return err + } + i.moved.mu.Lock() + if i.moved.kind == nil { + i.moved.kind = map[string]bool{} + } + i.moved.kind[KindBuildsLayout] = true + i.moved.mu.Unlock() + return nil +} diff --git a/internal/inventory/chain_move_test.go b/internal/inventory/chain_move_test.go index e72a07b9..ab3297e8 100644 --- a/internal/inventory/chain_move_test.go +++ b/internal/inventory/chain_move_test.go @@ -135,9 +135,10 @@ func TestTheMoveCopiesTheBuildRowsAndAgainWritesNothing(t *testing.T) { t.Fatal(err) } allAlready(t, again) - for _, id := range []string{"build-1", "build-2", "build-3", "build-4"} { - if history, err := inv.chain.Builds.History(ctx, chainBuildKey(id)); err != nil || len(history) != 1 { - t.Fatalf("%s has %d values: %v", id, len(history), err) + for _, key := range []string{legacyBuildKey("build-1"), legacyBuildKey("build-2"), legacyBuildKey("build-3"), + chain.BuildKey("notes", "build-4")} { + if history, err := inv.chain.Builds.History(ctx, key); err != nil || len(history) != 1 { + t.Fatalf("%s has %d values: %v", key, len(history), err) } } } @@ -319,11 +320,11 @@ func TestBuildsOnTheBusAreReadBackByEveryReader(t *testing.T) { if n := rowsInTheTable(t, inv, "build"); n != "0" { t.Fatalf("the table took %s builds", n) } - kept, found, err := inv.chain.Builds.Get(ctx, "build-1") + kept, found, err := inv.chain.Builds.Get(ctx, chain.BuildKey("notes", "build-1")) if err != nil || !found || kept.Value.ReuseKey != "rk-1" { t.Fatalf("build-1 on the bus: %+v %v %v", kept.Value, found, err) } - if history, _ := inv.chain.Builds.History(ctx, "build-1"); len(history) != 1 { + if history, _ := inv.chain.Builds.History(ctx, chain.BuildKey("notes", "build-1")); len(history) != 1 { t.Fatalf("a build recorded twice has %d values", len(history)) } list, err := inv.Builds(ctx, "notes", 10) @@ -427,7 +428,7 @@ func TestARequestIsTheBuildInStateAskedAndItsOutcomeTakesItsPlace(t *testing.T) if again, _, _ := inv.BuildRequestByID(ctx, "build-1"); again.NotAsked != "" { t.Fatalf("an answered request was marked not asked: %+v", again) } - if history, _ := inv.chain.Builds.History(ctx, "build-1"); len(history) != 3 { + if history, _ := inv.chain.Builds.History(ctx, chain.RequestKey("build-1")); len(history) != 3 { // asked, marked unknown, answered. t.Fatalf("the record has %d values", len(history)) } diff --git a/internal/inventory/chain_requests.go b/internal/inventory/chain_requests.go index 230ba9c3..808ffd08 100644 --- a/internal/inventory/chain_requests.go +++ b/internal/inventory/chain_requests.go @@ -35,7 +35,14 @@ func (i *Inventory) chainRecordBuildRequest(ctx context.Context, c *chain.Chain, if a.At.IsZero() { a.At = time.Now().UTC() } - key := chainBuildKey(a.ID) + key := chain.RequestKey(a.ID) + // An outcome under a module that came first has no request beside it: the request is kept on it, and its own record says + // it was answered (issue 1121). + if ptr, found, err := c.Builds.GetRaw(ctx, chain.IDKey(a.ID)); err != nil { + return err + } else if found && string(ptr.Value) != chain.RequestPrefix { + return i.keepAnsweredRequest(ctx, c, string(ptr.Value), a) + } for try := 0; try < casTries; try++ { _, err := c.Builds.Create(ctx, key, askedRecord(a)) if err == nil { @@ -65,8 +72,35 @@ func (i *Inventory) chainRecordBuildRequest(ctx context.Context, c *chain.Chain, return fmt.Errorf("build request %s is not kept on the bus: it kept moving under the write", a.ID) } +// keepAnsweredRequest puts a request on the outcome that answered it before the request was kept, and keeps the request's +// own record as answered. +func (i *Inventory) keepAnsweredRequest(ctx context.Context, c *chain.Chain, module string, a BuildRequest) error { + outKey := chain.BuildKey(module, a.ID) + for try := 0; try < casTries; try++ { + out, found, err := c.Builds.Get(ctx, outKey) + if err != nil || !found { + return err + } + if out.Value.Request == nil { + out.Value.Request = toRequest(a) + if _, err := c.Builds.Update(ctx, outKey, out.Value, out.Revision); errors.Is(err, chain.ErrMoved) { + continue + } else if err != nil { + return fmt.Errorf("build request %s is not kept on the bus: %w", a.ID, err) + } + } + rec := askedRecord(a) + rec.State, rec.Module, rec.Failed, rec.At = chain.BuildAnswered, module, out.Value.Failed, out.Value.At + if _, err := c.Builds.Create(ctx, chain.RequestKey(a.ID), rec); err != nil && !errors.Is(err, chain.ErrExists) { + return fmt.Errorf("build request %s is not kept on the bus: %w", a.ID, err) + } + return nil + } + return fmt.Errorf("build request %s is not kept: it kept moving under the write", a.ID) +} + func (i *Inventory) chainBuildRequestByID(ctx context.Context, c *chain.Chain, id string) (BuildRequest, bool, error) { - k, found, err := c.Builds.Get(ctx, chainBuildKey(id)) + k, found, err := c.Builds.Get(ctx, chain.RequestKey(id)) if err != nil || !found || k.Value.Request == nil { return BuildRequest{}, false, err } @@ -75,7 +109,7 @@ func (i *Inventory) chainBuildRequestByID(ctx context.Context, c *chain.Chain, i // chainMarkRequest changes a request that has no outcome yet, by compare-and-set; one answered, or none kept, is left. func (i *Inventory) chainMarkRequest(ctx context.Context, c *chain.Chain, id string, mark func(*chain.Request)) error { - key := chainBuildKey(id) + key := chain.RequestKey(id) for try := 0; try < casTries; try++ { cur, found, err := c.Builds.Get(ctx, key) if err != nil || !found || cur.Value.State != chain.BuildAsked || cur.Value.Request == nil { @@ -94,13 +128,14 @@ func (i *Inventory) chainMarkRequest(ctx context.Context, c *chain.Chain, id str // chainRequests is every request kept, newest first, each with its outcome where its build was heard. Requests older than // KeptFor are not listed: the bucket keeps them, and no reader needs them. func (i *Inventory) chainRequests(ctx context.Context, c *chain.Chain) ([]RequestOutcome, error) { - all, err := c.Builds.All(ctx) + all, err := requestRecords(ctx, c) if err != nil { - return nil, fmt.Errorf("the builds cannot be read from the bus: %w", err) + return nil, err } oldest := time.Now().Add(-KeptFor) var out []RequestOutcome - for id, k := range all { + for _, k := range all { + id := strings.TrimPrefix(k.Key, chain.RequestPrefix+".") r := k.Value.Request if r == nil || !r.At.After(oldest) { continue @@ -120,6 +155,11 @@ func (i *Inventory) chainRequests(ctx context.Context, c *chain.Chain) ([]Reques return out, nil } +// requestRecords are the records under `req.`: the requests, and the builds that knew no module. +func requestRecords(ctx context.Context, c *chain.Chain) ([]chain.Kept[chain.Build], error) { + return buildsIn(ctx, c, chain.RequestPrefix+".*") +} + // RequestsToAskAgain are the requests the move found never handed over (issue 981 note 7 Q5), not yet asked again: each is // asked once, and never twice. func (i *Inventory) RequestsToAskAgain(ctx context.Context) ([]BuildRequest, error) { @@ -127,12 +167,13 @@ func (i *Inventory) RequestsToAskAgain(ctx context.Context) ([]BuildRequest, err if err != nil || !on { return nil, err } - all, err := c.Builds.All(ctx) + all, err := requestRecords(ctx, c) if err != nil { return nil, err } var out []BuildRequest - for id, k := range all { + for _, k := range all { + id := strings.TrimPrefix(k.Key, chain.RequestPrefix+".") if k.Value.State == chain.BuildAsked && k.Value.Request != nil && k.Value.Request.Reask == chain.ReaskPending { out = append(out, fromRequest(id, *k.Value.Request)) } @@ -148,12 +189,12 @@ func (i *Inventory) ClaimToAskAgain(ctx context.Context, id string) (bool, error if err != nil || !on { return false, err } - cur, found, err := c.Builds.Get(ctx, chainBuildKey(id)) + cur, found, err := c.Builds.Get(ctx, chain.RequestKey(id)) if err != nil || !found || cur.Value.Request == nil || cur.Value.Request.Reask != chain.ReaskPending { return false, err } cur.Value.Request.Reask = chain.ReaskClaimed - if _, err := c.Builds.Update(ctx, chainBuildKey(id), cur.Value, cur.Revision); err != nil { + if _, err := c.Builds.Update(ctx, chain.RequestKey(id), cur.Value, cur.Revision); err != nil { if errors.Is(err, chain.ErrMoved) { return false, nil } diff --git a/internal/inventory/fortest.go b/internal/inventory/fortest.go index 5a2067d0..3469d78d 100644 --- a/internal/inventory/fortest.go +++ b/internal/inventory/fortest.go @@ -72,7 +72,7 @@ func ForTest(t *testing.T) *Inventory { testChains.Store(os.Getenv(store.Variable(Name)), c) t.Cleanup(func() { testChains.Delete(os.Getenv(store.Variable(Name))) }) if _, err := inv.store.Pool().Exec(t.Context(), - `insert into chain_moved (counts, largest) values ($1, $1)`, `{"builds": 0, "rollouts": 0, "assignments": 0}`); err != nil { + `insert into chain_moved (counts, largest) values ($1, $1)`, `{"builds": 0, "builds-layout": 0, "rollouts": 0, "assignments": 0}`); err != nil { t.Fatal(err) } return inv diff --git a/internal/inventory/source.go b/internal/inventory/source.go index 63bb9dbc..084905b7 100644 --- a/internal/inventory/source.go +++ b/internal/inventory/source.go @@ -3,6 +3,8 @@ package inventory import ( "context" "encoding/json" + + "github.com/novox/mesh-controller/internal/chain" ) // A build whose source is unchanged is never a move (novox/hq issue 280). @@ -26,6 +28,7 @@ type sourceRow struct { id, commit, fingerprint string made []byte manifest []byte + manifestDigest string failedGate bool } @@ -97,6 +100,15 @@ func (i *Inventory) StandingBuild(ctx context.Context, module, build string) (st for k, b := range builds { if b.id == build { s := builds[standing(builds, k)] + if len(s.manifest) == 0 && s.manifestDigest != "" { + // The manifest is kept apart from the record (issue 1121): one Get. + c, on, err := i.ChainReads(ctx, KindBuilds) + if err != nil || !on { + return "", nil, err + } + raw, err := manifestOfRecord(ctx, c, chain.Build{ID: s.id, ManifestDigest: s.manifestDigest}) + return s.id, raw, err + } return s.id, s.manifest, nil } } @@ -106,6 +118,12 @@ func (i *Inventory) StandingBuild(ctx context.Context, module, build string) (st // SourceFingerprints is, per module, per commit, the source fingerprint of the newest successful build // from it that has one: module → commit → fingerprint. A commit whose builds carry none is absent. func (i *Inventory) SourceFingerprints(ctx context.Context) (map[string]map[string]string, error) { + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return nil, err + } else if on { + // Read from the commit pointers, which carry each commit's source fingerprint: no record is read (issue 1121). + return i.chainSourceFingerprints(ctx, c) + } all, _, err := i.sourceRows(ctx, "") if err != nil { return nil, err @@ -129,6 +147,12 @@ func (i *Inventory) SourceFingerprints(ctx context.Context) (map[string]map[stri // BuildSourceFingerprints is SourceFingerprints for one module: commit → fingerprint. func (i *Inventory) BuildSourceFingerprints(ctx context.Context, module string) (map[string]string, error) { + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return nil, err + } else if on { + // This module's commit pointers only (issue 1121). + return i.chainBuildSourceFingerprints(ctx, c, module) + } all, err := i.SourceFingerprints(ctx) if err != nil { return nil, err @@ -142,6 +166,11 @@ func (i *Inventory) BuildSourceFingerprints(ctx context.Context, module string) // heldMade is, per module, what the build standing for its newest successful build made — what Held // answers (novox/hq issue 280). func (i *Inventory) heldMade(ctx context.Context) (map[string][]Artifact, error) { + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return nil, err + } else if on { + return i.chainHeldMade(ctx, c) + } all, _, err := i.sourceRows(ctx, "") if err != nil { return nil, err -- 2.54.0