From f907be2c1477b17bd98d54f834ecc11b58b18adf Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 20:56:12 +0100 Subject: [PATCH] The view derives per repository up to which merge every node runs it, keeps it as its state reach and says each reached head, so a fix on the machines reads delivered (issue 1280, ADR 0556 clauses 2 and 3) --- .../cmd/mesh-delivery-view-tools/main.go | 15 +- .../cmd/mesh-delivery-view-tools/main_test.go | 13 + .../cmd/mesh-delivery-view/main.go | 47 ++- .../mesh-delivery-view/internal/view/reach.go | 313 ++++++++++++++++++ .../internal/view/reach_test.go | 140 ++++++++ .../mesh-delivery-view/internal/view/read.go | 32 ++ .../mesh-delivery-view/internal/view/view.go | 72 +++- .../internal/view/view_test.go | 10 +- modules/mesh-delivery-view/manifest_test.go | 9 +- modules/mesh-delivery-view/module.json | 7 +- 10 files changed, 628 insertions(+), 30 deletions(-) create mode 100644 modules/mesh-delivery-view/internal/view/reach.go create mode 100644 modules/mesh-delivery-view/internal/view/reach_test.go diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go index c48e580..04a9279 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go @@ -19,6 +19,18 @@ import ( // Tools are the read tools over a modules state, and build_usage over the build modules' states. func Tools(r view.Reader, u view.UsageReader) []stdio.Tool { return []stdio.Tool{ + {Name: view.ReachReadTool, + Description: "Whether a merge is on the machines: per repository, the reached head (the newest merge every node runs a build " + + "containing, as the view derives it from the indexer's modules, the wants and the haves; a merge at or before it is on the " + + "machines), what holds the next one (a module on a node, the indexer's lag), and per module its fingerprint, the nodes " + + "that want it, its changes (fingerprint, the merge it first appeared at, its source revision), its floor and the nodes " + + "where it is not settled or not healthy (novox/hq ADR 0556). Give a repository (owner/name) for one, none for every one. Read-only.", + Input: map[string]any{"type": "object", "properties": map[string]any{ + "repository": map[string]any{"type": "string", "description": "the repository, owner/name; none for every repository"}}}, + Run: func(args map[string]any) (any, error) { + repository, _ := args["repository"].(string) + return r.Reach(repository) + }}, {Name: view.BuildUsageTool, Description: "How the build nodes are used, and whether a node is eligible to build, per node (every node holding a build module or named by a capacity): each build seat it holds (node-build-go, -node, -image) with its capacity " + "(mesh-config's resolved value, or the modules' default) and pause, and whether it is eligible to build and why not; the asks " + @@ -180,7 +192,8 @@ func read(state stdio.KeptState) view.StateRead { func main() { r := view.Reader{State: read(stdio.State("mesh-delivery-indexer.modules")), Builds: read(stdio.State("mesh-delivery-builder.builds")), CandidatesState: read(stdio.State("mesh-delivery-indexer.candidates")), - WantsState: read(stdio.State("mesh-delivery-assigner.wants")), DifferencesState: read(stdio.State("mesh-delivery-assigner.differences")), HavesState: read(stdio.State("node-engine.have"))} + WantsState: read(stdio.State("mesh-delivery-assigner.wants")), DifferencesState: read(stdio.State("mesh-delivery-assigner.differences")), HavesState: read(stdio.State("node-engine.have")), + ReachState: read(stdio.State("mesh-delivery-view.reach"))} u := view.UsageReader{Haves: r.HavesState, Resolved: read(stdio.State("mesh-config.resolved")), Tries: read(stdio.State("mesh-delivery-checker.tries")), Steps: map[string]view.StateRead{}, Pauses: map[string]view.StateRead{}, Backlog: backlogOf(socket())} for _, lang := range delivery.AskLanguages { diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go index cd98d08..94da531 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go @@ -71,6 +71,19 @@ func TestModuleToolAnswersTheStoredValue(t *testing.T) { if _, err := run(t, r, view.ModuleReadTool, map[string]any{"module": ""}); err == nil { t.Errorf("an empty name was answered") } + // 1280 (ADR 0556): view_reach answers the view's reach of one repository as stored, or of every one, and says a + // repository it holds none of. + const reach = `{"v":1,"repository":"novox/web","reached":"c1","reached_src":12}` + r.ReachState = fake{"novox/web": reach}.reader().State + if out, err := run(t, r, view.ReachReadTool, map[string]any{"repository": "novox/Web"}); err != nil || string(out.(json.RawMessage)) != reach { + t.Errorf("view_reach novox/Web = %v, %v", out, err) + } + if out, err := run(t, r, view.ReachReadTool, map[string]any{}); err != nil || !strings.Contains(fmt.Sprint(out), "novox/web") { + t.Errorf("view_reach of every repository = %v, %v", out, err) + } + if _, err := run(t, r, view.ReachReadTool, map[string]any{"repository": "novox/gone"}); err == nil || !strings.Contains(err.Error(), "no reach of novox/gone") { + t.Errorf("a repository with no reach: %v", err) + } } // 1562/AC1: indexer_modules answers every key as stored, and with none2 only those that say why they have no FP2. diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go index 4ae4f95..d8e64e6 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go @@ -144,13 +144,27 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } }) model.SetDeliveredEvent(func(e view.DeliveredEvent) { - if err := emitEvent(ctx, js, cred, DeliveredEventType, e, fmt.Sprintf("delivered.%d", e.Revision)); err != nil { + id := fmt.Sprintf("delivered.%d", e.Revision) + if e.Repository != "" { // a repository's reached head (ADR 0556 clause 3) + key, _ := delivery.RepositoryKey(e.Repository) + id = fmt.Sprintf("reached.%s.%d", key, e.Revision) + } + if err := emitEvent(ctx, js, cred, DeliveredEventType, e, id); err != nil { logger.Printf("the delivery of source revision %d could not be said: %v", e.Revision, err) } }) + model.SetReachWriter(func(w view.ReachWrite) { + if err := putReach(ctx, js, w); err != nil { + logger.Printf("the reach of %s could not be written (written again at its next change): %v", w.Key, err) + } + }) model.SetBacklog(view.BacklogOver(js, 5*time.Second)) - for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket, delivery.CandidatesBucket, delivery.ModulesBucket, delivery.AsksBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket, delivery.AssignerWantsBucket, delivery.SourceBucket, delivery.ControllerComposedBucket, delivery.AssignerAsksBucket, delivery.ViewAnswersBucket} { - go follow(ctx, js, bucket, model, logger) + for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket, delivery.CandidatesBucket, delivery.ModulesBucket, delivery.AsksBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket, delivery.AssignerWantsBucket, delivery.SourceBucket, delivery.ControllerComposedBucket, delivery.AssignerAsksBucket, delivery.ViewAnswersBucket, delivery.TakenBucket} { + go follow(ctx, js, bucket, false, model, logger) + } + // The modules' and the source's history give each fingerprint's first commit and each head's source revision (ADR 0556). + for _, bucket := range []string{delivery.ModulesBucket, delivery.SourceBucket} { + go follow(ctx, js, bucket, true, model, logger) } select { case <-ctx.Done(): @@ -181,11 +195,21 @@ func (w kvWriter) Put(key string, value []byte) error { return err } +// putReach writes a reach value under its key in the view's own state reach (ADR 0556 clause 2), its only writer. +func putReach(ctx context.Context, js jetstream.JetStream, w view.ReachWrite) error { + kv, err := js.KeyValue(ctx, delivery.ReachBucket) + if err != nil { + return fmt.Errorf("the bucket %s: %w", delivery.ReachBucket, err) + } + _, err = kv.Put(ctx, w.Key, w.Value) + return err +} + // follow watches a bucket with its last values first and then each change, for as long as ctx lives: a bucket that is // not there yet, or a watch that ends, is tried again. -func follow(ctx context.Context, js jetstream.JetStream, bucket string, model *view.Model, logger *log.Logger) { +func follow(ctx context.Context, js jetstream.JetStream, bucket string, history bool, model *view.Model, logger *log.Logger) { for ctx.Err() == nil { - if err := watchOnce(ctx, js, bucket, model); err != nil && ctx.Err() == nil { + if err := watchOnce(ctx, js, bucket, history, model); err != nil && ctx.Err() == nil { logger.Printf("cannot watch %s, trying again: %v", bucket, err) } select { @@ -195,12 +219,17 @@ func follow(ctx context.Context, js jetstream.JetStream, bucket string, model *v } } -func watchOnce(ctx context.Context, js jetstream.JetStream, bucket string, model *view.Model) error { +func watchOnce(ctx context.Context, js jetstream.JetStream, bucket string, history bool, model *view.Model) error { kv, err := js.KeyValue(ctx, bucket) if err != nil { return err } - w, err := kv.WatchAll(ctx) + var opts []jetstream.WatchOpt + initial := bucket + if history { + opts, initial = append(opts, jetstream.IncludeHistory()), view.ReachHistory(bucket) + } + w, err := kv.WatchAll(ctx, opts...) if err != nil { return err } @@ -214,11 +243,11 @@ func watchOnce(ctx context.Context, js jetstream.JetStream, bucket string, model return errors.New("the watch ended") } if e == nil { // the last values have all been delivered - model.Initial(bucket) + model.Initial(initial) continue } model.Apply(view.Entry{Bucket: bucket, Key: e.Key(), Value: e.Value(), At: e.Created(), - Deleted: e.Operation() != jetstream.KeyValuePut, Revision: e.Revision()}) + Deleted: e.Operation() != jetstream.KeyValuePut, Revision: e.Revision(), History: history}) } } } diff --git a/modules/mesh-delivery-view/internal/view/reach.go b/modules/mesh-delivery-view/internal/view/reach.go new file mode 100644 index 0000000..4a353ae --- /dev/null +++ b/modules/mesh-delivery-view/internal/view/reach.go @@ -0,0 +1,313 @@ +package view + +import ( + "fmt" + "maps" + "slices" + "strings" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// How far each repository reached (novox/hq ADR 0556 clauses 2 and 3, issue 1280). The view keeps, per repository the gate's +// source names, how far each of its modules reached on the nodes that want it, and derives the repository's reached head: the +// newest head of the repository in the source's history, not past the merge the indexer took in, at which every wanted module's +// floor is at or after its last change at or before that head. It writes that as its state reach and says each new reached +// head as its event delivered, per repository. Nothing here records a merge as delivered (ADR 0429 §1): a merge at or before +// the reached head is on the machines, derived by whoever reads it. + +// ReachHistory names the watch of a bucket with its history, as Initial is told it: the modules' and the source's history +// give each fingerprint's first commit and each head's source revision. +func ReachHistory(bucket string) string { return bucket + "#history" } + +// reachBuckets are the watches the derivation waits for before it writes or says anything, so a start does not write a reach +// derived from half its inputs. +var reachBuckets = []string{delivery.ModulesBucket, delivery.AssignerWantsBucket, delivery.HaveBucket, delivery.SourceBucket, + delivery.TakenBucket, ReachHistory(delivery.ModulesBucket), ReachHistory(delivery.SourceBucket)} + +// ReachWrite is a reach value to write under its key in the view's state reach. +type ReachWrite struct { + Key string + Value []byte +} + +// SetReachWriter sets where the view's reach values go; nothing is written until it is set. +func (m *Model) SetReachWriter(put func(ReachWrite)) { + m.mu.Lock() + defer m.mu.Unlock() + m.reachPut = put +} + +// applyHistory takes one revision of a watch with history: of the indexer's modules (each module's values in order) and of +// the gate's source (each revision's heads). +func (m *Model) applyHistory(e Entry) { + switch e.Bucket { + case delivery.ModulesBucket: + if e.Deleted { + delete(m.modHist, e.Key) + return + } + mod, err := delivery.DecodeModule(e.Value) + if err != nil || mod.FP2 == "" || mod.Commit == "" { + return + } + h := m.modHist[e.Key] + if n := len(h); n > 0 && h[n-1].FP2 == mod.FP2 && h[n-1].Commit == mod.Commit { + if mod.Src != 0 { + h[n-1].Src = mod.Src + } + return + } + h = append(h, mod) + if len(h) > 2*delivery.MostChanges { + h = h[len(h)-delivery.MostChanges:] + } + m.modHist[e.Key] = h + case delivery.SourceBucket: + if e.Key != delivery.SourceMainKey || e.Deleted || e.Revision == 0 { + return + } + src, err := delivery.DecodeSource(e.Value) + if err != nil { + return + } + m.srcHist[e.Revision] = src.Heads + } +} + +// headAt is one head of a repository and the first source revision that named it. +type headAt struct { + commit string + rev uint64 +} + +// headsOf are the heads of a repository in the source's history, oldest first, each with the first revision naming it. +func (m *Model) headsOf(repository string) []headAt { + var out []headAt + for _, rev := range slices.Sorted(maps.Keys(m.srcHist)) { + for r, head := range m.srcHist[rev] { + if !delivery.SameRepository(r, repository) || head == "" { + continue + } + if n := len(out); n == 0 || !sameCommit(out[n-1].commit, head) { + out = append(out, headAt{commit: head, rev: rev}) + } + } + } + return out +} + +// revOf is the first source revision naming a commit as the repository's head, or 0 when the history holds none. +func revOf(heads []headAt, commit string) uint64 { + for _, h := range heads { + if sameCommit(h.commit, commit) { + return h.rev + } + } + return 0 +} + +// sameCommit is whether two spellings of a commit are one (either may be abbreviated to at least seven characters). +func sameCommit(a, b string) bool { + a, b = strings.ToLower(a), strings.ToLower(b) + if len(a) > len(b) { + a, b = b, a + } + return len(a) >= 7 && strings.HasPrefix(b, a) +} + +// noBound is a module's bound when its floor is its newest change: it holds no head back. +const noBound = ^uint64(0) + +// deriveReach is the reach of one repository from the model as it stands. +func (m *Model) deriveReach(repository string) delivery.Reach { + r := delivery.Reach{Repository: delivery.RepositoryName(repository), Modules: map[string]delivery.ModuleReach{}} + heads := m.headsOf(repository) + key, _ := delivery.RepositoryKey(repository) + if t, ok := m.taken[key]; ok { + r.Indexed = t.Merge + } + bound := noBound + var holds []string + for _, name := range slices.Sorted(maps.Keys(m.modules)) { + me := m.modules[name] + if me.err != "" || me.value.FP2 == "" || !delivery.SameRepository(me.value.Repository, repository) { + continue + } + mr, b, words := m.moduleReach(name, me.value, heads) + r.Modules[name] = mr + if b < bound { + bound = b + } + if b != noBound { + holds = append(holds, words) + } + } + if len(r.Modules) == 0 { + r.Modules = nil + } + indexed := uint64(0) + if r.Indexed != "" { + indexed = revOf(heads, r.Indexed) + } + switch { + case r.Indexed == "": + holds = append(holds, "the indexer has taken in no merge of "+r.Repository) + case indexed == 0: + holds = append(holds, fmt.Sprintf("the merge the indexer took in, %.12s, is not a head of %s in the source's history", r.Indexed, r.Repository)) + } + for _, h := range heads { + if h.rev <= indexed && h.rev < bound { + r.Reached, r.ReachedSrc = h.commit, h.rev + } + } + if n := len(heads); n > 0 && !sameCommit(heads[n-1].commit, r.Reached) { + if indexed != 0 && indexed < heads[n-1].rev { + holds = append(holds, fmt.Sprintf("the indexer has not yet taken in %.12s", heads[n-1].commit)) + } + slices.Sort(holds) + r.Holds = slices.Compact(holds) + } + return r +} + +// moduleReach is how far one module reached, the source revision it holds the reached head before (noBound when none), and +// what holds it, in words. A node counts for the fingerprint its running have runs, healthy or not (ADR 0556 clause 2). +func (m *Model) moduleReach(name string, mod delivery.Module, heads []headAt) (delivery.ModuleReach, uint64, string) { + mr := delivery.ModuleReach{FP: mod.FP2} + hist := m.modHist[name] + if n := len(hist); n == 0 || hist[n-1].FP2 != mod.FP2 || hist[n-1].Commit != mod.Commit { + hist = append(slices.Clone(hist), mod) // the value watched now is the newest, whatever the history watch has given + } + for i := len(hist) - 1; i >= 0 && len(mr.Changes) < delivery.MostChanges; i-- { + c := delivery.Change{Commit: hist[i].Commit, FP: hist[i].FP2, Src: hist[i].Src} + if c.Src == 0 { + c.Src = revOf(heads, c.Commit) + } + if n := len(mr.Changes); n > 0 && mr.Changes[n-1].FP == c.FP { + mr.Changes[n-1] = c // the same fingerprint again, back to back: its first appearance is the older commit + continue + } + mr.Changes = append(mr.Changes, c) + } + var waiting []string + floor, floorNode, unset := -1, "", "" + for _, key := range slices.Sorted(maps.Keys(m.wants)) { + node, module := splitKey(key) + w := m.wants[key] + if module != name || w.err != "" || w.value.Removed || w.value.FP == "" { + continue + } + mr.Nodes++ + h, ok := m.haves[key] + switch { + case !ok || h.err != "" || h.have.FP == "": + waiting = append(waiting, fmt.Sprintf("%s on %s runs nothing yet", name, node)) + unset = waiting[len(waiting)-1] + continue + case !runs(h.have.Phase): + waiting = append(waiting, fmt.Sprintf("%s on %s is %s at %s, not running", name, node, h.have.Phase, short(h.have.FP))) + unset = waiting[len(waiting)-1] + continue + case h.have.FP != w.value.FP: + waiting = append(waiting, fmt.Sprintf("%s on %s runs %s, not yet the %s it wants", name, node, short(h.have.FP), short(w.value.FP))) + case h.have.Health == delivery.HealthFailed: + waiting = append(waiting, fmt.Sprintf("%s on %s runs %s and its health check failed", name, node, short(h.have.FP))) + case h.have.Health != delivery.HealthOK: + waiting = append(waiting, fmt.Sprintf("%s on %s runs %s, not yet healthy", name, node, short(h.have.FP))) + } + i := slices.IndexFunc(mr.Changes, func(c delivery.Change) bool { return c.FP == h.have.FP }) + if i < 0 { + unset = fmt.Sprintf("%s on %s runs %s, which is not among its %d known changes", name, node, short(h.have.FP), len(mr.Changes)) + waiting = append(waiting, unset) + continue + } + if i > floor { + floor, floorNode = i, node + } + } + mr.Waiting = strings.Join(waiting, "; ") + switch { + case mr.Nodes == 0: + return mr, noBound, "" + case unset != "": + return mr, 0, unset + } + mr.Floor = mr.Changes[floor].Commit + if floor == 0 { + return mr, noBound, "" + } + next := mr.Changes[floor-1] + words := fmt.Sprintf("%s on %s runs the build of %.12s, not yet that of %.12s", name, floorNode, mr.Floor, next.Commit) + if next.Src == 0 { + // The next change's order is not known: the head is held at the floor's own revision, or before every head when that is + // not known either. + if own := mr.Changes[floor].Src; own != 0 { + return mr, own + 1, words + } + return mr, 0, words + } + return mr, next.Src, words +} + +// runs is whether a have in this phase runs the fingerprint it names: running, being checked after its apply, or held after a +// failed backup, which keeps running what it has (ADR 0539 clause 16). A have backing up or applying, failed or removed runs +// nothing the derivation can count. +func runs(p delivery.Phase) bool { + return p == delivery.PhaseRunning || p == delivery.PhaseChecking || p == delivery.PhaseHeld +} + +// short is a fingerprint's first twelve characters, after its form. +func short(fp string) string { + fp = strings.TrimPrefix(fp, delivery.SourcePrefix2) + if len(fp) > 12 { + fp = fp[:12] + } + return fp +} + +// reachReady is whether every watch the derivation reads handed over what stood at its start. +func (m *Model) reachReady() bool { + for _, b := range reachBuckets { + if !m.initial[b] { + return false + } + } + return m.source != nil && m.source.err == "" +} + +// reachNow derives the reach of every repository the source names, and answers the values that changed and the reached +// heads not yet said. It is called with the lock held. +func (m *Model) reachNow() ([]ReachWrite, []DeliveredEvent) { + if (m.reachPut == nil && m.deliveredEvent == nil) || !m.reachReady() { + return nil, nil + } + var writes []ReachWrite + var events []DeliveredEvent + for _, repository := range slices.Sorted(maps.Keys(m.source.heads)) { + key, err := delivery.RepositoryKey(repository) + if err != nil { + continue + } + r := m.deriveReach(repository) + if prev, ok := m.reach[key]; !ok || !delivery.SameReach(prev, r) { + if data, err := delivery.EncodeReach(r); err == nil { + m.reach[key] = r + writes = append(writes, ReachWrite{Key: key, Value: data}) + } + } + if r.Reached != "" && r.ReachedSrc > m.reachSaid[key] { + m.reachSaid[key] = r.ReachedSrc + events = append(events, DeliveredEvent{Revision: r.ReachedSrc, Heads: map[string]string{r.Repository: r.Reached}, + Repository: r.Repository}) + } + } + return writes, events +} + +// Reach is the reach of every repository as the view derived it last, by key. +func (m *Model) Reach() map[string]delivery.Reach { + m.mu.Lock() + defer m.mu.Unlock() + return maps.Clone(m.reach) +} diff --git a/modules/mesh-delivery-view/internal/view/reach_test.go b/modules/mesh-delivery-view/internal/view/reach_test.go new file mode 100644 index 0000000..bc6db3c --- /dev/null +++ b/modules/mesh-delivery-view/internal/view/reach_test.go @@ -0,0 +1,140 @@ +package view + +import ( + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// 1280 (ADR 0556 clauses 2 and 3): the view derives, per repository, how far each module reached and the reached head, writes +// it when it differs, and says each new reached head once as its event delivered. +func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) { + const web, sdk = "novox/web", "novox/sdk" + commitA, commitB, commitC := strings.Repeat("a", 40), strings.Repeat("b", 40), strings.Repeat("c", 40) + commitS, commitX := strings.Repeat("5", 40), strings.Repeat("e", 40) + const fpMail = "src2:3333333333333333333333333333333333333333333333333333333333333333" + mod := func(module, fp, commit string, src uint64, history bool) Entry { + v, err := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: strings.Repeat("a", 64), Repository: web, Commit: commit, Src: src, Path: module}) + if err != nil { + t.Fatal(err) + } + return Entry{Bucket: delivery.ModulesBucket, Key: module, Value: v, At: t0, History: history} + } + src := func(rev uint64, webHead string) Entry { + v, err := delivery.EncodeSource(delivery.Source{Group: "g", Heads: map[string]string{web: webHead, sdk: commitS}}) + if err != nil { + t.Fatal(err) + } + return Entry{Bucket: delivery.SourceBucket, Key: delivery.SourceMainKey, Value: v, At: t0, Revision: rev, History: true} + } + taken := func(repository, merge string) Entry { + key, _ := delivery.RepositoryKey(repository) + v, err := delivery.EncodeTaken(delivery.Taken{Repository: repository, Merge: merge, At: t0}) + if err != nil { + t.Fatal(err) + } + return Entry{Bucket: delivery.TakenBucket, Key: key, Value: v, At: t0} + } + haveIn := func(node, module, fp string, phase delivery.Phase, health delivery.Health) Entry { + e := haveFP(t, node, module, fp, health) + v, _ := delivery.EncodeHave(delivery.Have{FP: fp, Phase: phase, Health: health}) + e.Value = v + return e + } + // web changed at A (fpOld) and at C (fpNew); mail first appeared at A. The source named A, B, then C as web's head. + world := func(t *testing.T) []Entry { + cur, _ := delivery.DecodeSource(src(12, commitC).Value) + current := Entry{Bucket: delivery.SourceBucket, Key: delivery.SourceMainKey, Revision: 12, At: t0} + current.Value, _ = delivery.EncodeSource(cur) + return []Entry{src(10, commitA), src(11, commitB), src(12, commitC), current, + mod("web", fpOld, commitA, 10, true), mod("web", fpNew, commitC, 12, true), mod("mail", fpMail, commitA, 0, true), + mod("web", fpNew, commitC, 12, false), mod("mail", fpMail, commitA, 0, false), + taken(web, commitC), taken(sdk, commitS), + wantE(t, "ace", "web", delivery.Want{FP: fpNew}), haveFP(t, "ace", "web", fpNew, delivery.HealthOK), + wantE(t, "g14", "web", delivery.Want{FP: fpNew}), haveFP(t, "g14", "web", fpNew, delivery.HealthOK), + wantE(t, "ace", "mail", delivery.Want{FP: fpMail}), haveFP(t, "ace", "mail", fpMail, delivery.HealthOK)} + } + for _, tc := range []struct { + name string + then []Entry + reached string // web's reached head; "" none + holds string // words one of web's holds carries; "" none + waiting string // words web's module web waits with; "" none + }{ + {name: "every node runs the newest build", reached: commitC}, + {name: "a node lagging stops the head before the change it lacks", reached: commitB, holds: "web on g14", + then: []Entry{haveFP(t, "g14", "web", fpOld, delivery.HealthOK)}, waiting: "not yet the"}, + {name: "a want held back at an older build holds the head as a lag does", reached: commitB, holds: "web on g14 runs the build of", + then: []Entry{wantE(t, "g14", "web", delivery.Want{FP: fpOld, Newest: fpNew, Rule: 2}), haveFP(t, "g14", "web", fpOld, delivery.HealthOK)}}, + {name: "a failing health check is said and holds nothing", reached: commitC, waiting: "health check failed", + then: []Entry{haveFP(t, "g14", "web", fpNew, delivery.HealthFailed)}}, + {name: "a node with no running have leaves no reached head", reached: "", holds: "not running", waiting: "not running", + then: []Entry{haveIn("g14", "web", fpNew, delivery.PhaseApplying, delivery.HealthUnknown)}}, + {name: "a module no node wants holds nothing", reached: commitC, + then: []Entry{wantE(t, "ace", "mail", delivery.Want{Removed: true})}}, + {name: "the indexer not yet having taken the head in", reached: commitB, holds: "has not yet taken in", + then: []Entry{taken(web, commitB)}}, + {name: "a change whose source revision is not known yet holds no head the source named", reached: commitC, + then: []Entry{mod("web", "src2:"+strings.Repeat("9", 64), commitX, 0, true), mod("web", "src2:"+strings.Repeat("9", 64), commitX, 0, false)}}, + } { + t.Run(tc.name, func(t *testing.T) { + m := New() + var written []ReachWrite + var said []DeliveredEvent + m.SetReachWriter(func(w ReachWrite) { written = append(written, w) }) + m.SetDeliveredEvent(func(e DeliveredEvent) { + if e.Repository != "" { + said = append(said, e) + } + }) + fill(m, world(t)...) + if len(written) != 0 { + t.Fatalf("written before every watch handed over: %v", written) + } + for _, b := range reachBuckets { + m.Initial(b) + } + fill(m, tc.then...) + key, _ := delivery.RepositoryKey(web) + r := m.Reach()[key] + if r.Reached != tc.reached { + t.Errorf("reached = %.8s, want %.8s (holds %v)", r.Reached, tc.reached, r.Holds) + } + if got := strings.Join(r.Holds, "; "); (tc.holds == "") != (got == "") || !strings.Contains(got, tc.holds) { + t.Errorf("holds = %q, want words with %q", got, tc.holds) + } + if got := r.Modules["web"].Waiting; (tc.waiting == "") != (got == "") || !strings.Contains(got, tc.waiting) { + t.Errorf("web waits with %q, want words with %q", got, tc.waiting) + } + // A library reaches the head the indexer took in, with no module. + sk, _ := delivery.RepositoryKey(sdk) + if lib := m.Reach()[sk]; lib.Reached != commitS || len(lib.Modules) != 0 { + t.Errorf("the library's reach = %+v", lib) + } + // Each key is written once with its last value; a reached head is said once, for its repository alone. + last := map[string]delivery.Reach{} + for _, w := range written { + v, err := delivery.DecodeReach(w.Value) + if err != nil { + t.Fatalf("written %s: %v", w.Key, err) + } + last[w.Key] = v + } + if !delivery.SameReach(last[key], r) { + t.Errorf("the last value written is %+v, not the reach derived %+v", last[key], r) + } + n := len(said) + fill(m, tc.then...) + fill(m, world(t)[len(world(t))-1]) + if len(said) != n && tc.reached == commitC { + t.Errorf("said again: %+v", said[n:]) + } + for _, e := range said { + if len(e.Heads) != 1 || e.Group != "" || e.Revision == 0 { + t.Errorf("a reached head said as %+v", e) + } + } + }) + } +} diff --git a/modules/mesh-delivery-view/internal/view/read.go b/modules/mesh-delivery-view/internal/view/read.go index 31be53d..c4047ab 100644 --- a/modules/mesh-delivery-view/internal/view/read.go +++ b/modules/mesh-delivery-view/internal/view/read.go @@ -39,6 +39,38 @@ type Reader struct { WantsState StateRead // the assigner's wants and holds DifferencesState StateRead // the assigner's differences HavesState StateRead // the node-engine's haves + ReachState StateRead // the view's own reach (ADR 0556) +} + +// ReachReadTool reads the view's reach: per repository, how far each module reached and the reached head (ADR 0556). +const ReachReadTool = "view_reach" + +// Reach is the stored reach of one repository, or of every repository when none is named, as the view wrote it. +func (r Reader) Reach(repository string) (any, error) { + if r.ReachState.Get == nil || r.ReachState.Keys == nil { + return nil, fmt.Errorf("the reach state cannot be read here") + } + if repository = strings.TrimSpace(repository); repository != "" { + key, err := delivery.RepositoryKey(repository) + if err != nil { + return nil, err + } + raw, err := r.ReachState.Get(key) + if err != nil { + return nil, err + } + if len(raw) == 0 { + return nil, fmt.Errorf("no reach of %s: the view has derived none (the gate's source does not name it, or the view waits for a watch)", repository) + } + return raw, nil + } + keys, err := r.ReachState.Keys() + if err != nil { + return nil, err + } + sort.Strings(keys) + values, err := r.ReachState.values(keys) + return map[string]any{"reach": values}, err } // Module is the stored value of modules., exactly as the indexer wrote it. diff --git a/modules/mesh-delivery-view/internal/view/view.go b/modules/mesh-delivery-view/internal/view/view.go index 8e3ecff..73795b8 100644 --- a/modules/mesh-delivery-view/internal/view/view.go +++ b/modules/mesh-delivery-view/internal/view/view.go @@ -23,6 +23,8 @@ type Entry struct { Deleted bool // Revision is the bucket revision of the change, when the watch gives one (the source revision is the gate's source key's). Revision uint64 + // History says the change comes from a watch with history (ReachHistory): it feeds only the history the reach reads. + History bool } // Row is one module on one node. @@ -240,6 +242,13 @@ type Model struct { // the watchman's buckets, by key presence map[string]presenceEntry stuck map[string]stuckEntry + // how far each repository reached (reach.go, ADR 0556) + modHist map[string][]delivery.Module // the indexer's values of each module, oldest first + srcHist map[uint64]map[string]string // the source's heads, by revision + taken map[string]delivery.Taken // the indexer's taken, by repository key + reach map[string]delivery.Reach // the reach derived last, by repository key + reachSaid map[string]uint64 // the reached head's source revision said last, by repository key + reachPut func(ReachWrite) } type presenceEntry struct { @@ -348,21 +357,41 @@ type nodeEntry struct { func New() *Model { return &Model{assigns: map[string]assignment{}, builds: map[string]buildEntry{}, groups: map[string]groupEntry{}, modules: map[string]moduleEntry{}, keys: map[string]keyEntry{}, asks: map[string]askRecord{}, wants: map[string]wantEntry{}, told: map[string]bool{}, initial: map[string]bool{}, composed: map[string]composedEntry{}, aasks: map[string]assignerAskRecord{}, chosen: map[string]delivery.ViewAnswer{}, answers: map[string]string{}, answerOf: map[string]delivery.Answer{}, holds: map[string]heldEntry{}, newest: map[string]string{}, haves: map[string]haveEntry{}, nodes: map[string]nodeEntry{}, - presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{}} + presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{}, + modHist: map[string][]delivery.Module{}, srcHist: map[uint64]map[string]string{}, taken: map[string]delivery.Taken{}, + reach: map[string]delivery.Reach{}, reachSaid: map[string]uint64{}} } // Apply takes one change of a key. A value it cannot read is kept as an error on its row, never guessed. When the change makes // the source revision delivered, the event of it is said once (SetDeliveredEvent), after the model's lock is released. func (m *Model) Apply(e Entry) { m.mu.Lock() - m.apply(e) - event := m.deliveredNow() - say := m.deliveredEvent - motion, sayMotion := m.motionNow(e), m.motionEvent + var event *DeliveredEvent + var motion []MotionEvent + if e.History { + m.applyHistory(e) + } else { + m.apply(e) + event = m.deliveredNow() + motion = m.motionNow(e) + } + writes, reached := m.reachNow() + say, put := m.deliveredEvent, m.reachPut + sayMotion := m.motionEvent m.mu.Unlock() if event != nil && say != nil { say(*event) } + if put != nil { + for _, w := range writes { + put(w) + } + } + if say != nil { + for _, ev := range reached { + say(ev) + } + } if sayMotion != nil { for _, ev := range motion { sayMotion(ev) @@ -373,10 +402,15 @@ func (m *Model) Apply(e Entry) { // DeliveredEvent is what the view says when the source revision is delivered (event `delivered` of the view, issue 1906 D9, ADR // 0539 §17): the revision of the gate's source key, the group that moved it and every repository's merged head at it. A // consumer takes every earlier revision as covered by it, and takes the same revision twice as once (after a restart of the view). +// +// The same event says a repository's reached head (ADR 0556 clause 3): Repository names it, Heads holds its head alone, Revision +// is the source revision that named that head and Group is empty; a consumer takes every earlier merge of that repository as on +// the machines. type DeliveredEvent struct { - Revision uint64 `json:"revision"` - Group string `json:"group"` - Heads map[string]string `json:"heads"` + Revision uint64 `json:"revision"` + Group string `json:"group"` + Heads map[string]string `json:"heads"` + Repository string `json:"repository,omitempty"` } // SetDeliveredEvent sets where the event of a delivered source revision goes; nothing is said until it is set. @@ -512,6 +546,16 @@ func (m *Model) apply(e Entry) { } d, err := delivery.DecodeComposedDigest(e.Value) m.composed[e.Key] = composedEntry{value: d, err: errText(err)} + case delivery.TakenBucket: + if e.Deleted { + delete(m.taken, e.Key) + return + } + if t, err := delivery.DecodeTaken(e.Value); err == nil { + m.taken[e.Key] = t + } else { + delete(m.taken, e.Key) + } case delivery.SourceBucket: if e.Key != delivery.SourceMainKey { return @@ -1214,10 +1258,22 @@ func (m *Model) Initial(bucket string) { } } say := m.motionEvent + writes, reached := m.reachNow() + put, sayDelivered := m.reachPut, m.deliveredEvent m.mu.Unlock() for _, ev := range out { say(ev) } + if put != nil { + for _, w := range writes { + put(w) + } + } + if sayDelivered != nil { + for _, ev := range reached { + sayDelivered(ev) + } + } } // motionOf is the events to say now about the node and module of a model key, from their want, have and stuck key. diff --git a/modules/mesh-delivery-view/internal/view/view_test.go b/modules/mesh-delivery-view/internal/view/view_test.go index d6ce64e..d228cab 100644 --- a/modules/mesh-delivery-view/internal/view/view_test.go +++ b/modules/mesh-delivery-view/internal/view/view_test.go @@ -110,7 +110,7 @@ func TestWaitsForTheNodeAndForANewerBuild(t *testing.T) { // 1134/AC2 } } -func TestLeftBehindIsNotRunning(t *testing.T) { // 1134/AC3 +func TestLeftBehindIsNotRunningAndPulledInIs(t *testing.T) { // 1134/AC3 (both halves) for name, extra := range map[string]string{"ended": `,"ended":"2026-10-18T11:30:00Z"`, "state": `,"state":"pending"`} { m := New() fill(m, base(t)...) @@ -131,13 +131,11 @@ func TestLeftBehindIsNotRunning(t *testing.T) { // 1134/AC3 if find(t, s, "a", "w").Status != StatusLeftBehind || s.Counted() != 4 { t.Fatalf("a have with no assignment: %+v counted %d", find(t, s, "a", "w"), s.Counted()) } -} - -func TestPulledInIsRunning(t *testing.T) { // 1134/AC3 - m := New() + // A module pulled in by another stands: shown and counted, with what required it. + m = New() fill(m, base(t)...) fill(m, assign("b", "z", "b3", `,"pulled-in":true,"required-by":"x"`)) - s := m.Snapshot() + s = m.Snapshot() r := find(t, s, "b", "z") if r.Status != StatusAssigned || r.RequiredBy != "x" || r.Have != "b3" || s.Counted() != 4 { t.Fatalf("pulled-in row = %+v counted %d, want shown and counted with required-by x", r, s.Counted()) diff --git a/modules/mesh-delivery-view/manifest_test.go b/modules/mesh-delivery-view/manifest_test.go index 6d41a31..0a5b6b8 100644 --- a/modules/mesh-delivery-view/manifest_test.go +++ b/modules/mesh-delivery-view/manifest_test.go @@ -84,7 +84,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 if m.Module != "mesh-delivery-view" { t.Errorf("module = %q", m.Module) } - want := []string{"mesh-build-go.pauses", "mesh-build-go.steps", "mesh-build-image.pauses", "mesh-build-image.steps", "mesh-build-node.pauses", "mesh-build-node.steps", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.asks", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9 + want := []string{"mesh-build-go.pauses", "mesh-build-go.steps", "mesh-build-image.pauses", "mesh-build-image.steps", "mesh-build-node.pauses", "mesh-build-node.steps", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.asks", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.taken", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9 got := slices.Clone(m.Reads) slices.Sort(got) if !reflect.DeepEqual(got, want) { @@ -103,8 +103,9 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 if len(m.Uses) != 0 { t.Errorf("uses = %v, want none", m.Uses) } - // The one state the view writes: the operator's answers to the asks that need no proof (ADR 0528 clause 6, 1297/B9), which the teller reads. - if want := []state{{Name: "answers", History: 5}}; !reflect.DeepEqual(m.State, want) { + // The states the view writes: the operator's answers to the asks that need no proof (ADR 0528 clause 6, 1297/B9), which the + // teller reads, and how far each repository reached (ADR 0556 clause 2, issue 1280), which the gate reads. + if want := []state{{Name: "answers", History: 5}, {Name: "reach", History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 { @@ -115,7 +116,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 } // The two read tools of the indexer's modules (issue 1562), carried by a bundle the runtime launches; the indexer // itself serves no tool (ADR 0437 §4). Carrying them gives the tool runner read access to every bucket the view reads. - if want := []string{"indexer_module", "indexer_modules", "indexer_candidate", "indexer_candidates", "builder_build", "builder_builds", "assigner_wants", "assigner_want", "assigner_differences", "node_haves", "build_usage"}; !reflect.DeepEqual(m.Tools, want) { + if want := []string{"indexer_module", "indexer_modules", "indexer_candidate", "indexer_candidates", "builder_build", "builder_builds", "assigner_wants", "assigner_want", "assigner_differences", "node_haves", "build_usage", "view_reach"}; !reflect.DeepEqual(m.Tools, want) { t.Errorf("tools = %v, want %v", m.Tools, want) } for _, a := range m.Build.Artifacts { diff --git a/modules/mesh-delivery-view/module.json b/modules/mesh-delivery-view/module.json index 9e10b47..551ff18 100644 --- a/modules/mesh-delivery-view/module.json +++ b/modules/mesh-delivery-view/module.json @@ -14,7 +14,8 @@ "assigner_want", "assigner_differences", "node_haves", - "build_usage" + "build_usage", + "view_reach" ], "emits": [ "delivered", @@ -23,7 +24,8 @@ "stuck" ], "state": [ - { "name": "answers", "history": 5 } + { "name": "answers", "history": 5 }, + { "name": "reach", "history": 1 } ], "reads": [ "mesh-controller.assignments", @@ -45,6 +47,7 @@ "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", + "mesh-delivery-indexer.taken", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck",