From 425134784a002d35188f3a8cdc4da268c4356888 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 18 Oct 2026 21:15:48 +0200 Subject: [PATCH 1/8] A machine silent past the bound at a tier between the first and the last is named and left as it was, as at the last tier (issue 1160) --- cmd/mesh-controller/release_plan.go | 104 ++++++++- .../silent_mid_tier_integration_test.go | 202 ++++++++++++++++++ cmd/mesh-controller/silent_mid_tier_test.go | 85 ++++++++ testdata/store-tests | 1 + 4 files changed, 384 insertions(+), 8 deletions(-) create mode 100644 cmd/mesh-controller/silent_mid_tier_integration_test.go create mode 100644 cmd/mesh-controller/silent_mid_tier_test.go diff --git a/cmd/mesh-controller/release_plan.go b/cmd/mesh-controller/release_plan.go index a2a28ce1..c61bf214 100644 --- a/cmd/mesh-controller/release_plan.go +++ b/cmd/mesh-controller/release_plan.go @@ -250,6 +250,68 @@ func appliedEach(module string, since func(node string) time.Time, running []str return len(waiting) == 0, waiting } +// gateNeed is a module the next tier waits for: the machines running it, and from when each must have reported. +type gateNeed struct { + module string + running []string + since func(node string) time.Time +} + +// silentNodes is the machines a walk sent and heard nothing from past ApplySilentAfter, on whichever tier it sent +// them (novox/hq ADR 0360 §5, issue 1160): the one rule both the gates between tiers and the walk's end read, through +// walkReports. A machine that reports on its send is not silent from then on. +func silentNodes(p inventory.Plan, reports []inventory.Reported, now time.Time, prov inventory.ProvLookup) map[string]bool { + _, _, silent := walkReports(p, reports, now, prov) + out := map[string]bool{} + for _, n := range silent { + out[n] = true + } + return out +} + +// withoutSilent is the machines of running that are not silent: a later tier is not sent to a machine the walk heard +// nothing from, which keeps its want of the plan's builds for the send it gets when it is heard again (ADR 0360 §5). +func withoutSilent(running []string, silent map[string]bool) []string { + return slices.DeleteFunc(slices.Clone(running), func(n string) bool { return silent[n] }) +} + +// settleTierGate is the wait between tiers on the reports of the machines running what the next tier needs: what is +// still waited for ("module on a, b"), the silent machines left out of the wait, sorted, and whether every machine the +// gate waits on is silent, so that nothing it needs was delivered anywhere (issue 1160). +func settleTierGate(needs []gateNeed, reports []inventory.Reported, silent map[string]bool) (waiting, quiet []string, lost bool) { + all, silentAll := 0, 0 + seenQuiet := map[string]bool{} + seenNode := map[string]bool{} + for _, need := range needs { + for _, n := range need.running { + if !seenNode[n] { + seenNode[n] = true + all++ + if silent[n] { + silentAll++ + } + } + } + _, on := appliedEach(need.module, need.since, need.running, reports) + var still []string + for _, n := range on { + if silent[n] { + if !seenQuiet[n] { + seenQuiet[n] = true + quiet = append(quiet, n) + } + continue + } + still = append(still, n) + } + if len(still) > 0 { + waiting = append(waiting, fmt.Sprintf("%s on %s", need.module, strings.Join(still, ", "))) + } + } + sort.Strings(quiet) + return waiting, quiet, all > 0 && silentAll == all +} + func applied(module string, builtAt time.Time, running []string, reports []inventory.Reported) (bool, []string) { at := map[string]*time.Time{} for _, r := range reports { @@ -869,6 +931,13 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, // being behind. Now every module of the tier whose first machine is the same is sent there in one // send, judged there by one gate — kept on the first of them, the others gated by it — and the rest // of the tier's machines are sent once, together, when the gates have passed. + // **A machine silent past the bound is left as it was, on every tier** (novox/hq ADR 0360 §5, issue 1160): the + // walk's end names it, and the gates between tiers and the sends of the tiers after do not wait for it or go to it. + allReports, err := inv.LastReports(ctx) + if err != nil { + return false, err + } + silent := silentNodes(*p, allReports, time.Now().UTC(), provLookup(ctx)) var pending []string firstOn := map[string][]string{} runningOf := map[string][]string{} @@ -903,6 +972,9 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, if goes != nil { running = slices.DeleteFunc(running, func(n string) bool { return !goes[n] }) } + if state.FirstAt == nil { + running = withoutSilent(running, silent) + } runningOf[m] = running // **A build whose source is unchanged is never a move** (novox/hq issue 280): made from the source // of the build every machine running it was sent, it is registered with that build's artifacts — @@ -1096,13 +1168,14 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, return changed, nil } // And wait for what the next tier needs running. + leftAsWas := "" needed := gates(*p, edges, rollsOut) if len(needed) > 0 { reports, err := inv.LastReports(ctx) if err != nil { return false, err } - var waiting []string + var needs []gateNeed for _, m := range needed { running, err := inv.Running(ctx, m) if err != nil { @@ -1133,14 +1206,12 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, for _, r := range reports { byNode[r.Node] = r } - if ok, on := appliedEach(m, func(node string) time.Time { + needs = append(needs, gateNeed{m, running, func(node string) time.Time { if s := byNode[node].Sent; s != nil { return *s } return time.Time{} - }, running, reports); !ok { - waiting = append(waiting, fmt.Sprintf("%s on %s", m, strings.Join(on, ", "))) - } + }}) continue } // **Each machine from its own send** (novox/hq issue 256). With one machine first (ADR 0218) @@ -1160,9 +1231,23 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, } return at } - if ok, on := appliedEach(m, sinceFor, running, reports); !ok { - waiting = append(waiting, fmt.Sprintf("%s on %s", m, strings.Join(on, ", "))) - } + needs = append(needs, gateNeed{m, running, sinceFor}) + } + waiting, quiet, lost := settleTierGate(needs, reports, silent) + if lost { + // Every machine the next tier waits on is silent: nothing it needs was delivered anywhere, so the plan does + // not go on as if it had been (issue 1160). + p.State = inventory.PlanFailed + p.Note = fmt.Sprintf("tier %d built; every machine it waits on said nothing on its send within %s: %s", + p.Tier, applySilentAfter, strings.Join(quiet, ", ")) + fmt.Printf("%s: %s\n", p.ID, p.Note) + return true, nil + } + if len(quiet) > 0 { + fmt.Printf("%s: tier %d: %s said nothing on its send within %s and is left as it was; the plan goes on without it\n", + p.ID, p.Tier, strings.Join(quiet, ", "), applySilentAfter) + leftAsWas = fmt.Sprintf("%s said nothing on its send within %s and is left as it was", strings.Join(quiet, ", "), + applySilentAfter) } if len(waiting) > 0 { note := "tier " + fmt.Sprint(p.Tier) + " built; waiting for " + strings.Join(waiting, "; ") + " to be applied" @@ -1204,6 +1289,9 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, carryUserList(ctx, open, p) } } + if end == "" { + end = leftAsWas + } p.Tier++ p.State = inventory.PlanBuilding p.Note = end diff --git a/cmd/mesh-controller/silent_mid_tier_integration_test.go b/cmd/mesh-controller/silent_mid_tier_integration_test.go new file mode 100644 index 00000000..645c908b --- /dev/null +++ b/cmd/mesh-controller/silent_mid_tier_integration_test.go @@ -0,0 +1,202 @@ +//go:build integration + +package main + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/catalogue" + "github.com/novox/mesh-controller/internal/inventory" + "github.com/novox/mesh-controller/internal/link" + "github.com/novox/mesh-controller/internal/testbus" +) + +// 1160/AC5, live, with the store and a real bus and no ticker: a plan of two tiers whose first tier has a machine that +// never reports, and whose second tier waits on the first (the plan's own order), passes its first tier after the bound, +// the silent key's expiry the only trigger, and sends its second tier to the machine that reported alone (issue 1160). +func TestAPlanPassesATierWhoseMachineNeverReportsAfterTheBoundAndSendsTheNextToTheOthers(t *testing.T) { + open := aMesh(t) + ctx := t.Context() + inv := open.inventory + + // A quarter of an hour of silence is two seconds here. + wasSilent := applySilentAfter + applySilentAfter = 2 * time.Second + t.Cleanup(func() { applySilentAfter = wasSilent }) + + js, err := broker.Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(js.Close) + if err := broker.AssertMeshStreams(js); err != nil { + t.Fatal(err) + } + if err := js.EnsureControllerBuckets(); err != nil { + t.Fatal(err) + } + legacy, err := js.Conn().JetStream() + if err != nil { + t.Fatal(err) + } + checkEvents = link.OverNATS{Conn: js.Conn(), JS: legacy} + t.Cleanup(func() { checkEvents = nil }) + sentHeard := make(chan *nats.Msg, 32) + sub, err := js.Conn().ChanSubscribe(link.SeatEventSubject(link.MeshControllerSeat, link.KeySent), sentHeard) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + wasSaved := inventory.PlanSaved + inventory.PlanSaved = servingPlanSaved(ctx, checkEvents, open) + t.Cleanup(func() { inventory.PlanSaved = wasSaved }) + + // The build machine and the machines, as far as the plan meets them: an ask is counted, a send is recorded and kept. + var mu sync.Mutex + var asks []string + var sends [][]string + wasAsk := askABuild + askABuild = func(_ context.Context, source buildSource, _, _ string) (string, error) { + mu.Lock() + defer mu.Unlock() + id := link.NewBuildID(time.Now().Add(time.Duration(len(asks)) * time.Millisecond)) + asks = append(asks, source.Repository+"#"+id) + return id, nil + } + t.Cleanup(func() { askABuild = wasAsk }) + digests := 0 + wasSend := sendRollout + sendRollout = func(ctx context.Context, open *stores, names []string) ([]string, error) { + mu.Lock() + defer mu.Unlock() + sends = append(sends, append([]string(nil), names...)) + current, err := open.inventory.CurrentBuilds(ctx) + if err != nil { + return nil, err + } + builds := map[string]string{} + for _, m := range []string{"a", "b"} { + builds[m] = current[m].Commit + } + for _, node := range names { + digests++ + if err := open.inventory.RecordSent(ctx, nodeID(t, open, node), fmt.Sprintf("d-%s-%d", node, digests), builds); err != nil { + return nil, err + } + } + return names, nil + } + t.Cleanup(func() { sendRollout = wasSend }) + + register := func(m, commit string, asked time.Time) { + t.Helper() + if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: m, Version: commit}, + inventory.Source{Repository: "novox/" + m, Seat: "git", Path: "modules/" + m, BuiltFrom: commit, Head: commit, + Asked: asked}); err != nil { + t.Fatal(err) + } + } + record := func(id, m, commit string, at time.Time) { + t.Helper() + b := inventory.Build{ID: id, Module: m, Commit: commit, Repository: "novox/" + m, Path: "modules/" + m, Asked: at, At: at} + b.Manifest, _ = json.Marshal(catalogue.Manifest{Module: m, Version: commit}) + if err := inv.RecordBuild(ctx, b); err != nil { + t.Fatal(err) + } + } + old := time.Now().Add(-2 * time.Hour) + for _, m := range []string{"a", "b"} { + record("build-"+m+"-1", m, "c1", old) + register(m, "c1", old) + for _, n := range []string{"anchor", "laptop"} { + if _, err := inv.Assign(ctx, n, m); err != nil { + t.Fatal(err) + } + } + } + for _, n := range []string{"anchor", "laptop"} { + if err := inv.RecordSent(ctx, nodeID(t, open, n), "d-"+n+"-c1", map[string]string{"a": "c1", "b": "c1"}); err != nil { + t.Fatal(err) + } + } + record("build-a-2", "a", "c2", time.Now().Add(-time.Minute)) + register("a", "c2", time.Now().Add(-time.Minute)) + + // A machine's report on what it was sent, heard as the controller hears it. + listener := nudgingListener{Enrolment: link.Enrolment{Inventory: inv}} + seq := int64(0) + reports := func(node string) { + t.Helper() + last, found, err := inv.SentTo(ctx, node) + if err != nil || !found { + t.Fatalf("%s was sent nothing: %v", node, err) + } + seq++ + body, _ := json.Marshal(map[string]any{"node": node, "applied": []string{"x.server"}, "declared": last.Digest, + "report_sequence": seq}) + var report link.Report + if err := json.Unmarshal(body, &report); err != nil { + t.Fatal(err) + } + if _, err := listener.Heard(ctx, report); err != nil { + t.Fatal(err) + } + } + plan := func() inventory.Plan { + t.Helper() + p, err := inv.PlanByID(ctx, "plan-1160") + if err != nil { + t.Fatal(err) + } + return p + } + + built := time.Now().UTC() + p := inventory.Plan{ID: "plan-1160", Repository: "novox/mesh-catalog", Branch: "main", Commit: "c2", Created: time.Now().UTC(), + State: inventory.PlanBuilding, Tiers: [][]string{{"a"}, {"b"}}, + Delivery: &inventory.PlanDelivery{Order: []inventory.Edge{{From: "b", To: "a", Kind: edgeChangeOrder}}}, + Modules: map[string]*inventory.PlanModule{ + "a": {State: "built", BuiltAt: &built, Commit: "c2", Build: "build-a-2"}, + "b": {}, + }} + first, stopFirst := context.WithCancel(ctx) + defer stopFirst() + startRolloutEvents(first, open, js) + if err := inv.SavePlan(ctx, &p); err != nil { + t.Fatal(err) + } + eventually(t, "tier 0 sent to both machines", func() bool { + mu.Lock() + defer mu.Unlock() + return len(sends) == 1 + }) + reports("anchor") + time.Sleep(time.Second) + if got := plan().Tier; got != 0 { + t.Fatalf("1160/AC5: the plan went on to tier %d before the bound, with laptop not heard from", got) + } + eventually(t, "tier 1 asked once laptop's key expired", func() bool { return plan().Tier == 1 }) + if note := plan().Note; !strings.Contains(note, "laptop") || strings.Contains(note, "anchor") { + t.Errorf("1160/AC5: the plan says %q; want laptop named silent and not anchor", note) + } + b := plan().Modules["b"] + askedAt, _ := link.BuildAskedAt(b.Build) + record(b.Build, "b", "c2", askedAt) + register("b", "c2", askedAt) + planBuilt(ctx, open, "b", "c2", "", askedAt, b.Build) + eventually(t, "tier 1 sent to anchor alone", func() bool { + mu.Lock() + defer mu.Unlock() + return len(sends) == 2 && strings.Join(sends[1], ",") == "anchor" + }) +} diff --git a/cmd/mesh-controller/silent_mid_tier_test.go b/cmd/mesh-controller/silent_mid_tier_test.go new file mode 100644 index 00000000..f0d64c99 --- /dev/null +++ b/cmd/mesh-controller/silent_mid_tier_test.go @@ -0,0 +1,85 @@ +package main + +import ( + "slices" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/inventory" +) + +// A machine silent past the bound at a tier between the first and the last is named silent and left as it was, as the +// last tier's is (novox/hq ADR 0282 §1, ADR 0360 §4-5, issue 1160): one rule, walkReports, read by the walk's end and by +// the gates between tiers. + +func silentWalk(now time.Time) (inventory.Plan, []inventory.Reported) { + sent := now.Add(-time.Minute) + long := now.Add(-inventory.ApplySilentAfter - time.Minute) + at := now.Add(-10 * time.Second) + p := inventory.Plan{Tier: 0, Tiers: [][]string{{"engine"}, {"app"}}, Modules: map[string]*inventory.PlanModule{ + "engine": {Rest: map[string]inventory.SentDeclaration{ + "anchor": {Digest: "d-anchor", At: sent}, + "g14": {Digest: "d-g14", At: long}, + "fresh": {Digest: "d-fresh", At: sent}}}, + "app": {}}} + reports := []inventory.Reported{{Node: "anchor", Outcome: inventory.OutcomeApplied, At: &at, Declared: "d-anchor"}} + return p, reports +} + +func needsOf(running ...string) []gateNeed { + return []gateNeed{{module: "engine", running: running, since: func(string) time.Time { return time.Time{} }}} +} + +// 1160/AC1: at a middle tier the machine silent past the bound is named, not waited for, and the plan goes on with the +// machines that reported; one still inside the bound is waited for. +func TestAMiddleTierGateNamesTheSilentMachineAndGoesOnWithThoseThatReported(t *testing.T) { + now := time.Now().UTC() + p, reports := silentWalk(now) + silent := silentNodes(p, reports, now, nil) + waiting, quiet, lost := settleTierGate(needsOf("anchor", "g14"), reports, silent) + if len(waiting) != 0 || !slices.Equal(quiet, []string{"g14"}) || lost { + t.Fatalf("1160/AC1: waiting %v, silent %v, lost %v; want g14 named silent and nothing waited for", waiting, quiet, lost) + } + waiting, quiet, _ = settleTierGate(needsOf("anchor", "g14", "fresh"), reports, silent) + if len(waiting) != 1 || waiting[0] != "engine on fresh" || !slices.Equal(quiet, []string{"g14"}) { + t.Fatalf("1160/AC1: a machine inside the bound is still waited for: waiting %v, silent %v", waiting, quiet) + } +} + +// 1160/AC2: the walk's end and the gate read the same silent machines. +func TestTheGateAndTheWalksEndShareOneRuleOfSilence(t *testing.T) { + now := time.Now().UTC() + p, reports := silentWalk(now) + _, _, ended := walkReports(p, reports, now, nil) + got := silentNodes(p, reports, now, nil) + if len(got) != len(ended) || !got["g14"] || !slices.Equal(ended, []string{"g14"}) { + t.Fatalf("1160/AC2: the gate reads %v silent, the walk's end %v", got, ended) + } +} + +// 1160/AC3: a later tier is not sent to a machine silent. +func TestALaterTierIsNotSentToASilentMachine(t *testing.T) { + got := withoutSilent([]string{"anchor", "g14"}, map[string]bool{"g14": true}) + if !slices.Equal(got, []string{"anchor"}) { + t.Fatalf("1160/AC3: a later tier goes to %v; want anchor alone", got) + } + in := []string{"anchor", "g14"} + withoutSilent(in, map[string]bool{"g14": true}) + if !slices.Equal(in, []string{"anchor", "g14"}) { + t.Fatalf("1160/AC3: the machines running it were changed: %v", in) + } +} + +// 1160/AC4: every machine the gate waits on silent: the plan is lost, not gone on as if delivered. +func TestAGateWhoseEveryMachineIsSilentIsLost(t *testing.T) { + now := time.Now().UTC() + p, reports := silentWalk(now) + reports = nil + p.Modules["engine"].Rest = map[string]inventory.SentDeclaration{ + "g14": {Digest: "d", At: now.Add(-time.Hour)}, "laptop": {Digest: "d", At: now.Add(-time.Hour)}} + silent := silentNodes(p, reports, now, nil) + _, quiet, lost := settleTierGate(needsOf("g14", "laptop"), reports, silent) + if !lost || !slices.Equal(quiet, []string{"g14", "laptop"}) { + t.Fatalf("1160/AC4: lost %v, silent %v; want both named and the gate lost", lost, quiet) + } +} diff --git a/testdata/store-tests b/testdata/store-tests index 4bf968a0..970640e6 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -163,3 +163,4 @@ ./internal/inventory TestASendEmptiesTheColumnSoALaterCopyLeavesAClearedKeyBare ./internal/inventory TestWhetherAMachineReadsModulesIsItsLatestWord ./cmd/mesh-controller TestAComposedDeclarationCarriesModulesAndBuildsOnlyForANodeThatReadsThem +./cmd/mesh-controller TestAPlanPassesATierWhoseMachineNeverReportsAfterTheBoundAndSendsTheNextToTheOthers From 1ec27e6ba18f1e3e1b59bbf907237b2ca45a9602 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 18 Oct 2026 21:24:07 +0200 Subject: [PATCH 2/8] Read tier 1's build and the note once the plan has reached them in the silent middle tier test (issue 1160) --- .../silent_mid_tier_integration_test.go | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/cmd/mesh-controller/silent_mid_tier_integration_test.go b/cmd/mesh-controller/silent_mid_tier_integration_test.go index 645c908b..ac0a465e 100644 --- a/cmd/mesh-controller/silent_mid_tier_integration_test.go +++ b/cmd/mesh-controller/silent_mid_tier_integration_test.go @@ -185,10 +185,20 @@ func TestAPlanPassesATierWhoseMachineNeverReportsAfterTheBoundAndSendsTheNextToT if got := plan().Tier; got != 0 { t.Fatalf("1160/AC5: the plan went on to tier %d before the bound, with laptop not heard from", got) } - eventually(t, "tier 1 asked once laptop's key expired", func() bool { return plan().Tier == 1 }) - if note := plan().Note; !strings.Contains(note, "laptop") || strings.Contains(note, "anchor") { - t.Errorf("1160/AC5: the plan says %q; want laptop named silent and not anchor", note) + // The note of the move to tier 1 is read in the step that sees it: later steps of the tier say their own. + var moved string + eventually(t, "tier 1 passed once laptop's key expired", func() bool { + if q := plan(); q.Tier == 1 { + moved = q.Note + return true + } + return false + }) + if !strings.Contains(moved, "laptop") || strings.Contains(moved, "anchor") { + t.Errorf("1160/AC5: the plan says %q; want laptop named silent and not anchor", moved) } + // The ask for tier 1's build is the next step, and records its build id on the module. + eventually(t, "tier 1 asked", func() bool { m := plan().Modules["b"]; return m != nil && m.Build != "" }) b := plan().Modules["b"] askedAt, _ := link.BuildAskedAt(b.Build) record(b.Build, "b", "c2", askedAt) From 75a27ec1a4bc855c9481f208c5896a2957dddadc Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 18 Oct 2026 23:51:56 +0200 Subject: [PATCH 3/8] The have bucket takes a node's whole facts, as large as a chain value, never cut (issue 1162) A node's own key carries its profile and was refused at 4 KiB on novox and g14 (nats: message size exceeds maximum allowed). --- internal/broker/controller_buckets.go | 6 ++++-- internal/broker/have_bucket_test.go | 9 +++++++++ 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/internal/broker/controller_buckets.go b/internal/broker/controller_buckets.go index 29ce9439..e08eba93 100644 --- a/internal/broker/controller_buckets.go +++ b/internal/broker/controller_buckets.go @@ -449,7 +449,9 @@ func EnsureLeaseBucket(ctx context.Context, js jetstream.JetStream) error { } // HaveBucketConfig is the options of the have bucket (novox/hq ADR 0429 §1): delivery.HaveHistory revisions of each -// key, no age, a value under a kilobyte (a have is JSON of about 500 bytes), 64 MiB on disk like the assignments. +// key, no age, 64 MiB on disk like the assignments. A module's have is JSON of about 500 bytes, but a node's own key +// carries its whole facts (its profile among them), several kilobytes on a desktop: a value is as large as the chain's +// buckets take, since a value is never cut (novox/hq ADR 0408 §6, issue 1162). func HaveBucketConfig() jetstream.KeyValueConfig { return jetstream.KeyValueConfig{ Bucket: delivery.HaveBucket, @@ -457,7 +459,7 @@ func HaveBucketConfig() jetstream.KeyValueConfig { "own facts) (novox/hq ADR 0429, design 47): written by each machine's node-engine under its own node's keys " + "alone, never by the controller; read by the delivery view and the watchman", History: delivery.HaveHistory, - MaxValueSize: 4 << 10, + MaxValueSize: ChainValueBytes, MaxBytes: 64 << 20, Storage: jetstream.FileStorage, } diff --git a/internal/broker/have_bucket_test.go b/internal/broker/have_bucket_test.go index cb584a68..076bf9f5 100644 --- a/internal/broker/have_bucket_test.go +++ b/internal/broker/have_bucket_test.go @@ -104,3 +104,12 @@ func TestAModuleReadingHaveAssignmentsAndBuildsReadsAndWritesNone(t *testing.T) } } } + +// A node's own key carries its whole facts (its profile among them), several kilobytes on a desktop: the have +// bucket takes any value the chain's buckets take, since a value is never cut (novox/hq ADR 0408 §6, issue 1162). +// 1162/AC1 +func TestTheHaveBucketTakesAValueAsLargeAsTheChains(t *testing.T) { + if got := HaveBucketConfig().MaxValueSize; got != ChainValueBytes { + t.Fatalf("the have bucket takes values up to %d bytes, the chain's up to %d", got, ChainValueBytes) + } +} From 19f15d8ea0d077e8d5a25038bfc6bff5705ca852 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 00:11:52 +0200 Subject: [PATCH 4/8] A merge on the operator's Approve into a repository the mesh builds nothing from asks the first build of the modules it holds (issue 1146, AC6, AC7) --- cmd/mesh-controller/batches.go | 5 + cmd/mesh-controller/first_builds.go | 98 +++++++++++++++++++ .../first_builds_integration_test.go | 81 +++++++++++++++ cmd/mesh-controller/first_builds_test.go | 77 +++++++++++++++ cmd/mesh-controller/moved.go | 14 ++- cmd/mesh-controller/upgrades.go | 7 +- testdata/store-tests | 1 + 7 files changed, 280 insertions(+), 3 deletions(-) create mode 100644 cmd/mesh-controller/first_builds.go create mode 100644 cmd/mesh-controller/first_builds_integration_test.go create mode 100644 cmd/mesh-controller/first_builds_test.go diff --git a/cmd/mesh-controller/batches.go b/cmd/mesh-controller/batches.go index 1acbe182..b97b7507 100644 --- a/cmd/mesh-controller/batches.go +++ b/cmd/mesh-controller/batches.go @@ -176,6 +176,11 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T repository, m.Base, m.Commit, already) return nil } + if already == 0 { + if done, err := f.firstBuilds(ctx, m, entries, read, now); err != nil || done { + return err + } + } fmt.Print(nothingReadsIt(repository, m)) return nil } diff --git a/cmd/mesh-controller/first_builds.go b/cmd/mesh-controller/first_builds.go new file mode 100644 index 00000000..58ff68a4 --- /dev/null +++ b/cmd/mesh-controller/first_builds.go @@ -0,0 +1,98 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "github.com/novox/mesh-controller/internal/inventory" + "github.com/novox/mesh-controller/internal/link" +) + +// firstBuildWait is how long after the forge made a merge the controller keeps hearing it again for mesh-delivery's word +// on it, which may be on its way when the forge's announcement is heard (novox/hq ADR 0432). +const firstBuildWait = 2 * time.Minute + +// firstBuilds asks the first build of every module a merge adds to a repository no registered module is built from, when +// the merge was announced as made on the operator's Approve of that head (novox/hq ADR 0432, issue 1146): the modules' +// directories are the ones the announcement says hold a module.json at the merge commit, as touchedBy reads them. The +// merge is kept, so the build's outcome is judged by the rule that registers it (mayRegisterFrom, approvedMove). Without +// the Approve nothing is asked, and it says why. Answers whether the merge was dealt with here. +func (f following) firstBuilds(ctx context.Context, m link.SourceMoved, entries []inventory.Entry, + read map[string][]inventory.ReadRepository, now time.Time) (bool, error) { + inv := f.open.inventory + repository := m.Owner + "/" + m.Repo + added, why, err := firstBuildsOf(ctx, inv, m, entries, read, now) + switch { + case errors.Is(err, errFirstNotApproved): + fmt.Printf("%s merged into %s (%.8s) adds %s, and no build is asked: %v\n", repository, m.Base, m.Commit, + strings.Join(addedDirs(m, entries, read), ", "), err) + return true, nil + case err != nil: + return true, notNow(err) + case len(added) == 0: + return false, nil + } + merged, err := time.Parse(time.RFC3339Nano, m.MergedAt) + if err != nil { + merged = now + } + event, err := json.Marshal(keptMerge{SourceMoved: m}) + if err != nil { + return true, err + } + if _, err := inv.AddMerge(ctx, inventory.BatchedMerge{Repository: repository, Branch: m.Base, Commit: m.Commit, + Merged: merged.UTC(), Heard: now, Event: event}); err != nil { + return true, notNow(err) + } + fmt.Printf("%s merged into %s (%.8s) on the operator's Approve (%s): the mesh builds nothing from it yet, so the "+ + "first build of %s is asked\n", repository, m.Base, m.Commit, why, strings.Join(added, ", ")) + askBuilds(ctx, inv, m, buildSource{Repository: repository, Seat: gitSeat}, added) + return true, nil +} + +// errFirstNotApproved says a merge into a repository the mesh builds nothing from is not one the operator approved. +var errFirstNotApproved = errors.New("it is no merge made on the operator's Approve of its head, so a hand `build` " + + "registers a module from a repository the mesh does not build yet") + +// addedDirs are the module directories the merge adds to a repository no registered module is built from, none where +// the merge is not into the trunk or some module is built from the repository already. +func addedDirs(m link.SourceMoved, entries []inventory.Entry, read map[string][]inventory.ReadRepository) []string { + if m.Base != "main" { + return nil + } + for _, e := range entries { + if !e.Provided && sameRepository(e.Source.Repository, m) { + return nil + } + } + _, added, _ := touchedBy(nil, entries, m, read) + return added +} + +// firstBuildsOf is the directories whose first build a merge asks, and the warrant it was made on; none when the merge +// is no first merge into a repository the mesh builds nothing from. errFirstNotApproved when it is, and the operator's +// Approve of its head is not on record, a try-again error while mesh-delivery's word may still be on its way. +func firstBuildsOf(ctx context.Context, inv catalogueReader, m link.SourceMoved, entries []inventory.Entry, + read map[string][]inventory.ReadRepository, now time.Time) ([]string, string, error) { + added := addedDirs(m, entries, read) + if len(added) == 0 { + return nil, "", nil + } + why, err := approvedAnnounced(ctx, inv, m.Owner, m.Repo, keptMerge{SourceMoved: m}) + if err == nil { + return added, why, nil + } + merged, perr := time.Parse(time.RFC3339Nano, m.MergedAt) + if errors.Is(err, errNotAnnounced) && perr == nil && now.Sub(merged) < firstBuildWait { + return nil, "", fmt.Errorf("%w: %w", link.ErrTryAgain, err) + } + var nope error = fmt.Errorf("%w (%v)", errFirstNotApproved, err) + if inventory.Unreachable(err) { + nope = err + } + return nil, "", nope +} diff --git a/cmd/mesh-controller/first_builds_integration_test.go b/cmd/mesh-controller/first_builds_integration_test.go new file mode 100644 index 00000000..f3b51e3e --- /dev/null +++ b/cmd/mesh-controller/first_builds_integration_test.go @@ -0,0 +1,81 @@ +//go:build integration + +package main + +import ( + "context" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/asks" + "github.com/novox/mesh-controller/internal/inventory" + "github.com/novox/mesh-controller/internal/link" +) + +// 1146/AC7: mesh-delivery #1 merged on the operator's Approve into a repository the mesh builds nothing from. The forge's +// announcement, heard as the controller hears it (SourceMoved, hearMerge), asks the first build of the module the merge +// commit holds; the build's outcome registers it; and the module is in the catalogue. Without the Approve nothing is +// asked. Store-backed: skips without one. +func TestAMergeOnTheApproveIntoANewRepositoryAsksTheFirstBuildAndItsOutcomeRegistersTheModule(t *testing.T) { + const head, commit = "a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1", "0177ea7a00000000" + for _, approved := range []bool{true, false} { + open := aMesh(t) + ctx := t.Context() + inv := open.inventory + asked := asksWithPaths(t) + + a := aDeliveryMergeOf(t, head, map[string]string{"novox/mesh-delivery@a1a1a1a1a1a1": head}) + a.merge.Number = 1 + wasKept, wasAsk := keptDeliveryMerge, routerAskOf + keptDeliveryMerge = func(context.Context, string, string, int, string) (deliveryMerge, bool, error) { + return a.merge, approved, nil + } + routerAskOf = func(context.Context, catalogueReader, string) (*asks.Ask, *asks.Warrant, error) { + return &a.ask, &a.warrant, nil + } + t.Cleanup(func() { keptDeliveryMerge, routerAskOf = wasKept, wasAsk }) + + m := link.SourceMoved{Owner: "novox", Repo: "mesh-delivery", Base: "main", HeadSHA: head, Commit: commit, Number: 1, + MergedAt: time.Now().Add(-time.Hour).UTC().Format(time.RFC3339Nano), Paths: []string{"module.json", "main.go"}, + ModuleDirs: []string{"."}, ModuleDirsSaid: true} + if err := (following{open: open}).SourceMoved(ctx, m); err != nil { + t.Fatalf("approved %v: %v", approved, err) + } + if !approved { + if len(*asked) != 0 { + t.Fatalf("1146/AC7: a merge with no Approve asked %v", *asked) + } + continue + } + want := [3]string{"novox/mesh-delivery", "", "main"} + if len(*asked) != 1 || (*asked)[0] != want { + t.Fatalf("1146/AC7: the merge asked %v, want %v", *asked, want) + } + build := onTrunk("b-", "novox/mesh-delivery", "git", "", map[string]any{"module": "mesh-delivery-view", "version": "1"}) + build.Commit = commit + if _, _, err := takeIn(ctx, inv, build); err != nil { + t.Fatalf("1146/AC7: the outcome of the first build was refused: %v", err) + } + src, err := inv.SourceOf(ctx, "mesh-delivery-view") + if err != nil || src.Repository != "novox/mesh-delivery" { + t.Fatalf("1146/AC7: the module is built from %q (%v)", src.Repository, err) + } + if _, known := inventoryHolds(t, inv, "mesh-delivery-view"); !known { + t.Fatal("1146/AC7: the module is not in the catalogue") + } + } +} + +func inventoryHolds(t *testing.T, inv *inventory.Inventory, module string) (inventory.Entry, bool) { + t.Helper() + entries, err := inv.Catalogued(t.Context()) + if err != nil { + t.Fatal(err) + } + for _, e := range entries { + if e.Manifest.Module == module { + return e, true + } + } + return inventory.Entry{}, false +} diff --git a/cmd/mesh-controller/first_builds_test.go b/cmd/mesh-controller/first_builds_test.go new file mode 100644 index 00000000..d49ec572 --- /dev/null +++ b/cmd/mesh-controller/first_builds_test.go @@ -0,0 +1,77 @@ +package main + +import ( + "context" + "errors" + "slices" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/asks" + "github.com/novox/mesh-controller/internal/inventory" + "github.com/novox/mesh-controller/internal/link" +) + +// 1146/AC6: the first merge into a repository no registered module is built from asks the first build of every module +// the merge commit holds when it was made on the operator's Approve of that head, and asks nothing without it. +func TestAFirstMergeIntoANewRepositoryAsksItsModulesBuildsOnlyOnTheApprove(t *testing.T) { + const head = "a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1" + now := time.Date(2026, 10, 19, 0, 6, 0, 0, time.UTC) + mergedAt := now.Add(-time.Hour).Format(time.RFC3339Nano) + heard := link.SourceMoved{Owner: "novox", Repo: "mesh-delivery", Base: "main", HeadSHA: head, Commit: "0177ea7a", + Number: 1, MergedAt: mergedAt, ModuleDirsSaid: true, ModuleDirs: []string{"watchman", "view"}, + Paths: []string{"view/module.json", "watchman/module.json", "README.md"}} + built := inventory.Entry{Source: inventory.Source{Repository: "novox/mesh-catalog", Seat: "git"}} + for _, c := range []struct { + what string + change func(*anApprovedMerge, *link.SourceMoved, *[]inventory.Entry) + unkept bool + dirs []string + refused bool + retry bool + }{ + {what: "the Approve of the merged head", dirs: []string{"view", "watchman"}}, + {what: "a Decline", refused: true, change: func(a *anApprovedMerge, _ *link.SourceMoved, _ *[]inventory.Entry) { + a.warrant.Option, a.warrant.Label, a.merge.Option = "decline", "Decline", "decline" + }}, + {what: "no word of mesh-delivery on a merge made long ago", unkept: true, refused: true}, + {what: "no word yet on a merge just made", unkept: true, retry: true, change: func(_ *anApprovedMerge, m *link.SourceMoved, _ *[]inventory.Entry) { + m.MergedAt = now.Add(-time.Second).Format(time.RFC3339Nano) + }}, + {what: "a module already built from the repository", change: func(_ *anApprovedMerge, _ *link.SourceMoved, e *[]inventory.Entry) { + *e = append(*e, inventory.Entry{Source: inventory.Source{Repository: "novox/mesh-delivery", Seat: "git"}}) + }}, + {what: "a merge into another branch", change: func(_ *anApprovedMerge, m *link.SourceMoved, _ *[]inventory.Entry) { m.Base = "feat" }}, + {what: "a merge that holds no module.json", change: func(_ *anApprovedMerge, m *link.SourceMoved, _ *[]inventory.Entry) { + m.ModuleDirs, m.Paths = nil, []string{"README.md"} + }}, + } { + t.Run(c.what, func(t *testing.T) { + a := aDeliveryMergeOf(t, head, map[string]string{"novox/mesh-delivery@a1a1a1a1a1a1": head}) + a.merge.Number = 1 + m, entries := heard, []inventory.Entry{built} + if c.change != nil { + c.change(&a, &m, &entries) + } + wasKept, wasAsk := keptDeliveryMerge, routerAskOf + t.Cleanup(func() { keptDeliveryMerge, routerAskOf = wasKept, wasAsk }) + keptDeliveryMerge = func(context.Context, string, string, int, string) (deliveryMerge, bool, error) { + return a.merge, !c.unkept, nil + } + routerAskOf = func(context.Context, catalogueReader, string) (*asks.Ask, *asks.Warrant, error) { + return &a.ask, &a.warrant, nil + } + dirs, _, err := firstBuildsOf(t.Context(), nil, m, entries, nil, now) + switch { + case c.retry && !errors.Is(err, link.ErrTryAgain): + t.Fatalf("1146/AC6: %s: want a try-again, got %v", c.what, err) + case c.refused && !errors.Is(err, errFirstNotApproved): + t.Fatalf("1146/AC6: %s: want the Approve missing, got %v (dirs %v)", c.what, err, dirs) + case !c.refused && !c.retry && err != nil: + t.Fatalf("1146/AC6: %s: %v", c.what, err) + case !slices.Equal(dirs, c.dirs): + t.Fatalf("1146/AC6: %s: first builds asked of %v, want %v", c.what, dirs, c.dirs) + } + }) + } +} diff --git a/cmd/mesh-controller/moved.go b/cmd/mesh-controller/moved.go index 456f166b..0344feb8 100644 --- a/cmd/mesh-controller/moved.go +++ b/cmd/mesh-controller/moved.go @@ -236,7 +236,17 @@ func approvedMove(ctx context.Context, inv registrations, forms *sourceForms, bu if !known { return "", fmt.Errorf("the controller kept no merge of %s/%s at %.8s", owner, name, r.Commit) } - k := announcedOf(merged) + return approvedAnnounced(ctx, inv, owner, name, announcedOf(merged)) +} + +// errNotAnnounced says mesh-delivery's word on a merge is not kept (yet): the merge may have been made by someone else, +// or its word may still be on its way. +var errNotAnnounced = errors.New("mesh-delivery announced no such merge") + +// approvedAnnounced is the warrant of the merge the forge announced as k: mesh-delivery announced it, and the ask it made +// it on holds the operator's Approve of that head. Why it holds, or why not. Shared by the approved move of a known +// module and the first build of a new repository's modules (ADR 0420 §7, ADR 0432). +func approvedAnnounced(ctx context.Context, inv catalogueReader, owner, name string, k keptMerge) (string, error) { if k.Number <= 0 || k.HeadSHA == "" { return "", errors.New("the forge's announcement of the merge names no pull request or head") } @@ -245,7 +255,7 @@ func approvedMove(ctx context.Context, inv registrations, forms *sourceForms, bu return "", err } if !kept { - return "", fmt.Errorf("mesh-delivery announced no merge of %s/%s#%d at %.8s", owner, name, k.Number, k.HeadSHA) + return "", fmt.Errorf("%w: %s/%s#%d at %.8s", errNotAnnounced, owner, name, k.Number, k.HeadSHA) } ask, w, err := routerAskOf(ctx, inv, d.AskID) if err != nil { diff --git a/cmd/mesh-controller/upgrades.go b/cmd/mesh-controller/upgrades.go index 94d25d4d..95be4d23 100644 --- a/cmd/mesh-controller/upgrades.go +++ b/cmd/mesh-controller/upgrades.go @@ -406,7 +406,12 @@ func askNewModules(ctx context.Context, inv *inventory.Inventory, m link.SourceM if len(added) == 0 || len(from) == 0 { return nil } - source := buildSource{Repository: from[0].Source.Repository, Seat: from[0].Source.Seat} + return askBuilds(ctx, inv, m, buildSource{Repository: from[0].Source.Repository, Seat: from[0].Source.Seat}, added) +} + +// askBuilds asks the first build of each directory of a merge's repository, as askNewModules does, from the source given. +func askBuilds(ctx context.Context, inv *inventory.Inventory, m link.SourceMoved, source buildSource, + added []string) []string { var asked []string for _, dir := range added { path := dir diff --git a/testdata/store-tests b/testdata/store-tests index 970640e6..3ba3497a 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -164,3 +164,4 @@ ./internal/inventory TestWhetherAMachineReadsModulesIsItsLatestWord ./cmd/mesh-controller TestAComposedDeclarationCarriesModulesAndBuildsOnlyForANodeThatReadsThem ./cmd/mesh-controller TestAPlanPassesATierWhoseMachineNeverReportsAfterTheBoundAndSendsTheNextToTheOthers +./cmd/mesh-controller TestAMergeOnTheApproveIntoANewRepositoryAsksTheFirstBuildAndItsOutcomeRegistersTheModule From 4e12fc19b7fe983a4540d050bdef309c539ec8fb Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 00:16:36 +0200 Subject: [PATCH 5/8] A first merge into a new repository waits kept for mesh-delivery's word, and the word's arrival asks the builds, with no timer; AC7 on the bus in both orders (issue 1146) --- cmd/mesh-controller/batches.go | 3 + cmd/mesh-controller/first_builds.go | 117 +++++++++--- .../first_builds_integration_test.go | 177 +++++++++++++----- cmd/mesh-controller/first_builds_test.go | 15 +- cmd/mesh-controller/moved.go | 14 +- cmd/mesh-controller/push.go | 3 +- internal/inventory/batches.go | 10 + internal/inventory/chain_rollouts.go | 16 ++ internal/inventory/waiting_merges_test.go | 33 ++++ 9 files changed, 301 insertions(+), 87 deletions(-) create mode 100644 internal/inventory/waiting_merges_test.go diff --git a/cmd/mesh-controller/batches.go b/cmd/mesh-controller/batches.go index b97b7507..20240682 100644 --- a/cmd/mesh-controller/batches.go +++ b/cmd/mesh-controller/batches.go @@ -747,6 +747,9 @@ func carriedOf(merges []inventory.BatchedMerge) ([]inventory.PlanCommit, []inven type keptMerge struct { link.SourceMoved Moves []string `json:"moves,omitempty"` + // FirstBuild says the merge is a first merge into a repository the mesh builds nothing from, kept to ask the first + // build of its modules once mesh-delivery's word on it is kept (novox/hq ADR 0432, issue 1146). + FirstBuild bool `json:"first_build,omitempty"` } // announcedOf is a kept merge as the forge announced it, and what it moved; empty where the record says none. diff --git a/cmd/mesh-controller/first_builds.go b/cmd/mesh-controller/first_builds.go index 58ff68a4..31624282 100644 --- a/cmd/mesh-controller/first_builds.go +++ b/cmd/mesh-controller/first_builds.go @@ -12,46 +12,115 @@ import ( "github.com/novox/mesh-controller/internal/link" ) -// firstBuildWait is how long after the forge made a merge the controller keeps hearing it again for mesh-delivery's word -// on it, which may be on its way when the forge's announcement is heard (novox/hq ADR 0432). -const firstBuildWait = 2 * time.Minute - // firstBuilds asks the first build of every module a merge adds to a repository no registered module is built from, when // the merge was announced as made on the operator's Approve of that head (novox/hq ADR 0432, issue 1146): the modules' // directories are the ones the announcement says hold a module.json at the merge commit, as touchedBy reads them. The -// merge is kept, so the build's outcome is judged by the rule that registers it (mayRegisterFrom, approvedMove). Without -// the Approve nothing is asked, and it says why. Answers whether the merge was dealt with here. +// merge is kept, so the build's outcome is judged by the rule that registers it (mayRegisterFrom, approvedMove). +// +// **Complete in either order, with no timer** (ADR 0426): when mesh-delivery's word is kept first, it is found here; when +// the forge's merge is heard first, the merge is kept as a first merge waiting for the word, and keepingDeliveryMerges +// runs this again when the word is kept (resumeFirstBuilds). A word that says no (a Decline, another head) asks nothing and +// says why. Answers whether the merge was dealt with here. func (f following) firstBuilds(ctx context.Context, m link.SourceMoved, entries []inventory.Entry, read map[string][]inventory.ReadRepository, now time.Time) (bool, error) { inv := f.open.inventory repository := m.Owner + "/" + m.Repo - added, why, err := firstBuildsOf(ctx, inv, m, entries, read, now) + added, why, err := firstBuildsOf(ctx, inv, m, entries, read) switch { case errors.Is(err, errFirstNotApproved): fmt.Printf("%s merged into %s (%.8s) adds %s, and no build is asked: %v\n", repository, m.Base, m.Commit, strings.Join(addedDirs(m, entries, read), ", "), err) return true, nil + case errors.Is(err, errNotAnnounced): + // Kept, waiting for mesh-delivery's word: a restart keeps it, and the word's arrival asks the builds. + if kerr := keepFirstMerge(ctx, inv, m, now, true); kerr != nil { + return true, notNow(kerr) + } + fmt.Printf("%s merged into %s (%.8s) adds %s; it is kept, and the first build is asked when mesh-delivery "+ + "announces the merge on the operator's Approve\n", repository, m.Base, m.Commit, + strings.Join(addedDirs(m, entries, read), ", ")) + return true, nil case err != nil: return true, notNow(err) case len(added) == 0: return false, nil } + if err := askFirstBuilds(ctx, inv, m, now, added, why); err != nil { + return true, notNow(err) + } + return true, nil +} + +// keepFirstMerge keeps a first merge as the controller heard it, waiting (no plan answers it). +func keepFirstMerge(ctx context.Context, inv *inventory.Inventory, m link.SourceMoved, now time.Time, waitingForWord bool) error { merged, err := time.Parse(time.RFC3339Nano, m.MergedAt) if err != nil { merged = now } - event, err := json.Marshal(keptMerge{SourceMoved: m}) + event, err := json.Marshal(keptMerge{SourceMoved: m, FirstBuild: waitingForWord}) if err != nil { - return true, err + return err } - if _, err := inv.AddMerge(ctx, inventory.BatchedMerge{Repository: repository, Branch: m.Base, Commit: m.Commit, - Merged: merged.UTC(), Heard: now, Event: event}); err != nil { - return true, notNow(err) + _, err = inv.AddMerge(ctx, inventory.BatchedMerge{Repository: m.Owner + "/" + m.Repo, Branch: m.Base, Commit: m.Commit, + Merged: merged.UTC(), Heard: now, Event: event}) + return err +} + +// askFirstBuilds keeps the merge, asks the builds, and marks the merge as no longer waiting for them. +func askFirstBuilds(ctx context.Context, inv *inventory.Inventory, m link.SourceMoved, now time.Time, added []string, + why string) error { + repository := m.Owner + "/" + m.Repo + if err := keepFirstMerge(ctx, inv, m, now, false); err != nil { + return err } fmt.Printf("%s merged into %s (%.8s) on the operator's Approve (%s): the mesh builds nothing from it yet, so the "+ "first build of %s is asked\n", repository, m.Base, m.Commit, why, strings.Join(added, ", ")) askBuilds(ctx, inv, m, buildSource{Repository: repository, Seat: gitSeat}, added) - return true, nil + // Answered: it no longer waits, so the word's arrival again asks nothing twice. + return inv.AnswerMerge(ctx, repository, m.Commit, inventory.WaitingRecord, false) +} + +// resumeFirstBuilds runs firstBuilds for the first merges waiting for the word mesh-delivery just announced for a pull +// request's head: the other order of the same two events. +func (f following) resumeFirstBuilds(ctx context.Context, d deliveryMerge) error { + actingOnMerges.Lock() + defer actingOnMerges.Unlock() + inv := f.open.inventory + waiting, err := inv.WaitingMerges(ctx) + if err != nil { + return notNow(err) + } + for _, w := range waiting { + k := announcedOf(w) + if !k.FirstBuild || !strings.EqualFold(k.Owner+"/"+k.Repo, d.Owner+"/"+d.Repo) || k.Number != d.Number || + k.HeadSHA != d.HeadSHA { + continue + } + entries, err := inv.Catalogued(ctx) + if err != nil { + return notNow(err) + } + read, err := readForPlanning(ctx, inv) + if err != nil { + return notNow(err) + } + added, why, err := firstBuildsOf(ctx, inv, k.SourceMoved, entries, read) + switch { + case errors.Is(err, errFirstNotApproved): + fmt.Printf("%s/%s merged into %s (%.8s) was waiting for mesh-delivery's word, and no build is asked: %v\n", + k.Owner, k.Repo, k.Base, k.Commit, err) + if aerr := inv.AnswerMerge(ctx, w.Repository, w.Commit, inventory.WaitingRecord, false); aerr != nil { + return notNow(aerr) + } + case err != nil: + return notNow(err) + case len(added) > 0: + if err := askFirstBuilds(ctx, inv, k.SourceMoved, time.Now().UTC(), added, why); err != nil { + return notNow(err) + } + } + } + return nil } // errFirstNotApproved says a merge into a repository the mesh builds nothing from is not one the operator approved. @@ -74,25 +143,21 @@ func addedDirs(m link.SourceMoved, entries []inventory.Entry, read map[string][] } // firstBuildsOf is the directories whose first build a merge asks, and the warrant it was made on; none when the merge -// is no first merge into a repository the mesh builds nothing from. errFirstNotApproved when it is, and the operator's -// Approve of its head is not on record, a try-again error while mesh-delivery's word may still be on its way. +// is no first merge into a repository the mesh builds nothing from. errFirstNotApproved when it is and the word on record +// is no Approve of its head; errNotAnnounced when mesh-delivery's word is not kept yet. func firstBuildsOf(ctx context.Context, inv catalogueReader, m link.SourceMoved, entries []inventory.Entry, - read map[string][]inventory.ReadRepository, now time.Time) ([]string, string, error) { + read map[string][]inventory.ReadRepository) ([]string, string, error) { added := addedDirs(m, entries, read) if len(added) == 0 { return nil, "", nil } why, err := approvedAnnounced(ctx, inv, m.Owner, m.Repo, keptMerge{SourceMoved: m}) - if err == nil { + switch { + case err == nil: return added, why, nil + case errors.Is(err, errWordNotAnApprove): + return nil, "", fmt.Errorf("%w (%v)", errFirstNotApproved, err) } - merged, perr := time.Parse(time.RFC3339Nano, m.MergedAt) - if errors.Is(err, errNotAnnounced) && perr == nil && now.Sub(merged) < firstBuildWait { - return nil, "", fmt.Errorf("%w: %w", link.ErrTryAgain, err) - } - var nope error = fmt.Errorf("%w (%v)", errFirstNotApproved, err) - if inventory.Unreachable(err) { - nope = err - } - return nil, "", nope + // Not announced yet, or the bus or the store could not be asked: the caller waits or tries again, and asks nothing. + return nil, "", err } diff --git a/cmd/mesh-controller/first_builds_integration_test.go b/cmd/mesh-controller/first_builds_integration_test.go index f3b51e3e..c1b4adb4 100644 --- a/cmd/mesh-controller/first_builds_integration_test.go +++ b/cmd/mesh-controller/first_builds_integration_test.go @@ -4,65 +4,144 @@ package main import ( "context" + "encoding/json" "testing" "time" - "git.novox.be/novox/mesh-sdk/go/asks" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/catalogue" "github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/link" + "github.com/novox/mesh-controller/internal/testbus" ) -// 1146/AC7: mesh-delivery #1 merged on the operator's Approve into a repository the mesh builds nothing from. The forge's -// announcement, heard as the controller hears it (SourceMoved, hearMerge), asks the first build of the module the merge -// commit holds; the build's outcome registers it; and the module is in the catalogue. Without the Approve nothing is -// asked. Store-backed: skips without one. +// 1146/AC7, whole against a real bus and the store, in both orders of the two events (novox/hq ADR 0432, issue 1146): +// mesh-delivery #7 merged on the operator's Approve into a repository the mesh builds nothing from. The forge's merge, +// heard as the controller hears it (SourceMoved), and mesh-delivery's word, published on the subject the controller +// consumes with the router's record of the ask in its bucket, come in either order; each way the first build of the +// module the merge commit holds is asked exactly once, its outcome registers the module, and it is in the catalogue. +// Without the Approve in the record nothing is asked. Store-backed: skips without one. func TestAMergeOnTheApproveIntoANewRepositoryAsksTheFirstBuildAndItsOutcomeRegistersTheModule(t *testing.T) { const head, commit = "a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1", "0177ea7a00000000" - for _, approved := range []bool{true, false} { - open := aMesh(t) - ctx := t.Context() - inv := open.inventory - asked := asksWithPaths(t) - - a := aDeliveryMergeOf(t, head, map[string]string{"novox/mesh-delivery@a1a1a1a1a1a1": head}) - a.merge.Number = 1 - wasKept, wasAsk := keptDeliveryMerge, routerAskOf - keptDeliveryMerge = func(context.Context, string, string, int, string) (deliveryMerge, bool, error) { - return a.merge, approved, nil - } - routerAskOf = func(context.Context, catalogueReader, string) (*asks.Ask, *asks.Warrant, error) { - return &a.ask, &a.warrant, nil - } - t.Cleanup(func() { keptDeliveryMerge, routerAskOf = wasKept, wasAsk }) - - m := link.SourceMoved{Owner: "novox", Repo: "mesh-delivery", Base: "main", HeadSHA: head, Commit: commit, Number: 1, - MergedAt: time.Now().Add(-time.Hour).UTC().Format(time.RFC3339Nano), Paths: []string{"module.json", "main.go"}, - ModuleDirs: []string{"."}, ModuleDirsSaid: true} - if err := (following{open: open}).SourceMoved(ctx, m); err != nil { - t.Fatalf("approved %v: %v", approved, err) - } - if !approved { - if len(*asked) != 0 { - t.Fatalf("1146/AC7: a merge with no Approve asked %v", *asked) + for _, c := range []struct { + what string + wordFirst bool + approved bool // the router's record holds the operator's Approve of the merged head + wantRegiter bool + }{ + {what: "the forge's merge before mesh-delivery's word", approved: true, wantRegiter: true}, + {what: "mesh-delivery's word before the forge's merge", wordFirst: true, approved: true, wantRegiter: true}, + {what: "a word whose record holds no Approve", wordFirst: true}, + } { + t.Run(c.what, func(t *testing.T) { + open := aMesh(t) + ctx := t.Context() + inv := open.inventory + asked := asksWithPaths(t) + if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "operator-channel", Version: "1", + DefinesSeats: []catalogue.SeatDeclaration{{Name: broker.AsksSeat, Records: []string{"asks"}}}}, + inventory.Source{Repository: "novox/mesh-catalog", Seat: "git", Path: "modules/operator-channel", BuiltFrom: "c0ffee"}); err != nil { + t.Fatal(err) } - continue - } - want := [3]string{"novox/mesh-delivery", "", "main"} - if len(*asked) != 1 || (*asked)[0] != want { - t.Fatalf("1146/AC7: the merge asked %v, want %v", *asked, want) - } - build := onTrunk("b-", "novox/mesh-delivery", "git", "", map[string]any{"module": "mesh-delivery-view", "version": "1"}) - build.Commit = commit - if _, _, err := takeIn(ctx, inv, build); err != nil { - t.Fatalf("1146/AC7: the outcome of the first build was refused: %v", err) - } - src, err := inv.SourceOf(ctx, "mesh-delivery-view") - if err != nil || src.Repository != "novox/mesh-delivery" { - t.Fatalf("1146/AC7: the module is built from %q (%v)", src.Repository, err) - } - if _, known := inventoryHolds(t, inv, "mesh-delivery-view"); !known { - t.Fatal("1146/AC7: the module is not in the catalogue") - } + js, err := broker.Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(js.Close) + if err := broker.AssertMeshStreams(js); err != nil { + t.Fatal(err) + } + if err := js.EnsureControllerBuckets(); err != nil { + t.Fatal(err) + } + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + records, err := api.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: broker.BucketName("operator-channel", "asks")}) + if err != nil { + t.Fatal(err) + } + a := aDeliveryMergeOf(t, head, map[string]string{"novox/mesh-delivery@a1a1a1a1a1a1": head}) + if c.approved { + record, _ := json.Marshal(map[string]any{"state": "closed", "ask": a.ask, "warrant": a.warrant}) + if _, err := records.Put(ctx, theDelivery+"."+a.merge.AskID, record); err != nil { + t.Fatal(err) + } + } + + following := following{open: open} + server := link.ConnectNats(js, nil, noListener{}) + if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream, kept: following.resumeFirstBuilds}); err != nil { + t.Fatal(err) + } + if err := broker.AssertServingConsumers(js); err != nil { + t.Fatal(err) + } + serving, stop := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { defer close(done); _ = server.Serve(serving) }() + t.Cleanup(func() { stop(); <-done }) + servingBus.Store(js) + t.Cleanup(func() { servingBus.Store(nil) }) + + word := func() { + t.Helper() + body, _ := json.Marshal(a.merge) + if _, err := js.Context().Publish(broker.DeliveryMergedSubject, body); err != nil { + t.Fatal(err) + } + becomes(t, 15*time.Second, "the word being kept", func() bool { + _, kept, err := keptDeliveryMerge(ctx, "novox", "mesh-delivery", 7, head) + return err == nil && kept + }) + time.Sleep(500 * time.Millisecond) // the keeper's hook runs as the word is kept + } + forge := func() { + t.Helper() + m := link.SourceMoved{Owner: "novox", Repo: "mesh-delivery", Base: "main", HeadSHA: head, Commit: commit, Number: 7, + MergedAt: time.Now().Add(-time.Second).UTC().Format(time.RFC3339Nano), Paths: []string{"module.json", "main.go"}, + ModuleDirs: []string{"."}, ModuleDirsSaid: true} + if err := following.SourceMoved(ctx, m); err != nil { + t.Fatal(err) + } + } + if c.wordFirst { + word() + forge() + } else { + forge() + if len(*asked) != 0 { + t.Fatalf("1146/AC7: the merge asked %v before mesh-delivery's word was kept", *asked) + } + word() + } + if !c.wantRegiter { + if len(*asked) != 0 { + t.Fatalf("1146/AC7: a merge whose record holds no Approve asked %v", *asked) + } + return + } + want := [3]string{"novox/mesh-delivery", "", "main"} + becomes(t, 10*time.Second, "the first build asked", func() bool { return len(*asked) >= 1 }) + if len(*asked) != 1 || (*asked)[0] != want { + t.Fatalf("1146/AC7: the merge asked %v, want %v once", *asked, want) + } + build := onTrunk("b-", "novox/mesh-delivery", "git", "", map[string]any{"module": "mesh-delivery-view", "version": "1"}) + build.Commit = commit + if _, _, err := takeIn(ctx, inv, build); err != nil { + t.Fatalf("1146/AC7: the outcome of the first build was refused: %v", err) + } + src, err := inv.SourceOf(ctx, "mesh-delivery-view") + if err != nil || src.Repository != "novox/mesh-delivery" { + t.Fatalf("1146/AC7: the module is built from %q (%v)", src.Repository, err) + } + if _, known := inventoryHolds(t, inv, "mesh-delivery-view"); !known { + t.Fatal("1146/AC7: the module is not in the catalogue") + } + }) } } diff --git a/cmd/mesh-controller/first_builds_test.go b/cmd/mesh-controller/first_builds_test.go index d49ec572..fa22e3ed 100644 --- a/cmd/mesh-controller/first_builds_test.go +++ b/cmd/mesh-controller/first_builds_test.go @@ -28,16 +28,13 @@ func TestAFirstMergeIntoANewRepositoryAsksItsModulesBuildsOnlyOnTheApprove(t *te unkept bool dirs []string refused bool - retry bool + waits bool }{ {what: "the Approve of the merged head", dirs: []string{"view", "watchman"}}, {what: "a Decline", refused: true, change: func(a *anApprovedMerge, _ *link.SourceMoved, _ *[]inventory.Entry) { a.warrant.Option, a.warrant.Label, a.merge.Option = "decline", "Decline", "decline" }}, - {what: "no word of mesh-delivery on a merge made long ago", unkept: true, refused: true}, - {what: "no word yet on a merge just made", unkept: true, retry: true, change: func(_ *anApprovedMerge, m *link.SourceMoved, _ *[]inventory.Entry) { - m.MergedAt = now.Add(-time.Second).Format(time.RFC3339Nano) - }}, + {what: "no word of mesh-delivery yet, however long ago the merge was made", unkept: true, waits: true}, {what: "a module already built from the repository", change: func(_ *anApprovedMerge, _ *link.SourceMoved, e *[]inventory.Entry) { *e = append(*e, inventory.Entry{Source: inventory.Source{Repository: "novox/mesh-delivery", Seat: "git"}}) }}, @@ -61,13 +58,13 @@ func TestAFirstMergeIntoANewRepositoryAsksItsModulesBuildsOnlyOnTheApprove(t *te routerAskOf = func(context.Context, catalogueReader, string) (*asks.Ask, *asks.Warrant, error) { return &a.ask, &a.warrant, nil } - dirs, _, err := firstBuildsOf(t.Context(), nil, m, entries, nil, now) + dirs, _, err := firstBuildsOf(t.Context(), nil, m, entries, nil) switch { - case c.retry && !errors.Is(err, link.ErrTryAgain): - t.Fatalf("1146/AC6: %s: want a try-again, got %v", c.what, err) + case c.waits && !errors.Is(err, errNotAnnounced): + t.Fatalf("1146/AC6: %s: want a wait for the word, got %v", c.what, err) case c.refused && !errors.Is(err, errFirstNotApproved): t.Fatalf("1146/AC6: %s: want the Approve missing, got %v (dirs %v)", c.what, err, dirs) - case !c.refused && !c.retry && err != nil: + case !c.refused && !c.waits && err != nil: t.Fatalf("1146/AC6: %s: %v", c.what, err) case !slices.Equal(dirs, c.dirs): t.Fatalf("1146/AC6: %s: first builds asked of %v, want %v", c.what, dirs, c.dirs) diff --git a/cmd/mesh-controller/moved.go b/cmd/mesh-controller/moved.go index 0344feb8..b0a5276e 100644 --- a/cmd/mesh-controller/moved.go +++ b/cmd/mesh-controller/moved.go @@ -115,6 +115,9 @@ func movedOnApprove(d deliveryMerge, ask *asks.Ask, w *asks.Warrant) (string, er // keepingDeliveryMerges keeps each merge mesh-delivery announces in the controller's bucket for them. type keepingDeliveryMerges struct { js func() *broker.JetStream + // kept, when set, is told each merge once it is kept: a first merge waiting for the word asks its builds then + // (novox/hq ADR 0432, issue 1146). + kept func(ctx context.Context, d deliveryMerge) error } func (k keepingDeliveryMerges) KeepDeliveryMerge(ctx context.Context, body []byte) error { @@ -137,6 +140,9 @@ func (k keepingDeliveryMerges) KeepDeliveryMerge(ctx context.Context, body []byt if _, err := kv.Put(ctx, deliveryMergeKey(d.Owner, d.Repo, d.Number, d.HeadSHA), body); err != nil { return fmt.Errorf("%w: %w", link.ErrTryAgain, err) } + if k.kept != nil { + return k.kept(ctx, d) + } return nil } @@ -239,6 +245,10 @@ func approvedMove(ctx context.Context, inv registrations, forms *sourceForms, bu return approvedAnnounced(ctx, inv, owner, name, announcedOf(merged)) } +// errWordNotAnApprove says the merge cannot be shown to be the operator's Approve of its head: its announcement names no +// pull request, or the word on record is not that Approve. +var errWordNotAnApprove = errors.New("the word on record is no Approve of the head merged") + // errNotAnnounced says mesh-delivery's word on a merge is not kept (yet): the merge may have been made by someone else, // or its word may still be on its way. var errNotAnnounced = errors.New("mesh-delivery announced no such merge") @@ -248,7 +258,7 @@ var errNotAnnounced = errors.New("mesh-delivery announced no such merge") // module and the first build of a new repository's modules (ADR 0420 §7, ADR 0432). func approvedAnnounced(ctx context.Context, inv catalogueReader, owner, name string, k keptMerge) (string, error) { if k.Number <= 0 || k.HeadSHA == "" { - return "", errors.New("the forge's announcement of the merge names no pull request or head") + return "", fmt.Errorf("%w: the forge's announcement of the merge names no pull request or head", errWordNotAnApprove) } d, kept, err := keptDeliveryMerge(ctx, owner, name, k.Number, k.HeadSHA) if err != nil { @@ -263,7 +273,7 @@ func approvedAnnounced(ctx context.Context, inv catalogueReader, owner, name str } why, err := movedOnApprove(d, ask, w) if err != nil { - return "", err + return "", fmt.Errorf("%w: %w", errWordNotAnApprove, err) } return fmt.Sprintf("%s/%s#%d merged at %.8s on %s", owner, name, k.Number, k.HeadSHA, why), nil } diff --git a/cmd/mesh-controller/push.go b/cmd/mesh-controller/push.go index 2ae089e6..0b5fc231 100644 --- a/cmd/mesh-controller/push.go +++ b/cmd/mesh-controller/push.go @@ -209,7 +209,8 @@ func serve(ctx context.Context) (err error) { } // And the merges mesh-delivery made, kept a week with the ask each was made on: what registers a module that // moved between repositories on the operator's Approve (novox/hq ADR 0420 §7, issue 1025). - if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream}); err != nil { + if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream, + kept: following{open: open}.resumeFirstBuilds}); err != nil { return err } // No timer reads the merges back (novox/hq ADR 0407 decision 3, issue 973): the forge's event and the durable diff --git a/internal/inventory/batches.go b/internal/inventory/batches.go index 294c5f00..e870a676 100644 --- a/internal/inventory/batches.go +++ b/internal/inventory/batches.go @@ -84,6 +84,16 @@ func (i *Inventory) AloneMerges(ctx context.Context) ([]BatchedMerge, error) { return i.chainAloneMerges(ctx, c) } +// WaitingMerges is every merge no plan answers that is not walked alone: kept, and waiting for something else to +// happen to it (a first merge into a repository, waiting for mesh-delivery's word on it, issue 1146). +func (i *Inventory) WaitingMerges(ctx context.Context) ([]BatchedMerge, error) { + c, err := i.theChain(ctx) + if err != nil { + return nil, err + } + return i.chainWaitingMerges(ctx, c) +} + // AnswerMerge moves a merge to the plan that answers it now; an empty plan with alone leaves it waiting to // be walked on its own commit. func (i *Inventory) AnswerMerge(ctx context.Context, repository, commit, plan string, alone bool) error { diff --git a/internal/inventory/chain_rollouts.go b/internal/inventory/chain_rollouts.go index b7a9e839..6c242b6d 100644 --- a/internal/inventory/chain_rollouts.go +++ b/internal/inventory/chain_rollouts.go @@ -452,6 +452,22 @@ func (i *Inventory) chainAloneMerges(ctx context.Context, c *chain.Chain) ([]Bat return out, nil } +// chainWaitingMerges is every merge no plan answers that waits and is not walked alone, the newest first. +func (i *Inventory) chainWaitingMerges(ctx context.Context, c *chain.Chain) ([]BatchedMerge, error) { + w, err := i.chainWinners(ctx, c) + if err != nil { + return nil, err + } + var out []BatchedMerge + for _, h := range w { + if h.m.Waiting && !h.m.Alone { + out = append(out, batchedOf(h)) + } + } + sortMerges(out, true) + return out, nil +} + // pendingMerge is a merge added to a plan that is not saved yet: carried until that plan's save writes it. type pendingMerge struct { plan string diff --git a/internal/inventory/waiting_merges_test.go b/internal/inventory/waiting_merges_test.go new file mode 100644 index 00000000..4601e937 --- /dev/null +++ b/internal/inventory/waiting_merges_test.go @@ -0,0 +1,33 @@ +package inventory + +import ( + "context" + "testing" + "time" +) + +// A merge no plan answers waits, and is listed as waiting until it is answered where it is kept (novox/hq issue 1146): +// what a first merge into a new repository does while it waits for mesh-delivery's word. +func TestAMergeNoPlanAnswersIsListedWaitingUntilItIsAnswered(t *testing.T) { + inv, c := anInventoryOnTheChain(t) + ctx := context.Background() + now := time.Now().UTC() + if added, err := inv.chainAddMerge(ctx, c, BatchedMerge{Repository: "novox/mesh-delivery", Branch: "main", Commit: "c1", + Merged: now, Heard: now}); err != nil || !added { + t.Fatalf("added %v, %v", added, err) + } + waiting, err := inv.chainWaitingMerges(ctx, c) + if err != nil || len(waiting) != 1 || waiting[0].Commit != "c1" { + t.Fatalf("waiting %+v, %v", waiting, err) + } + move := []MergeMove{{Repository: "novox/mesh-delivery", Commit: "c1", Plan: WaitingRecord}} + if err := inv.chainSavePlans(ctx, c, nil, move); err != nil { + t.Fatal(err) + } + if waiting, err = inv.chainWaitingMerges(ctx, c); err != nil || len(waiting) != 0 { + t.Fatalf("answered, it still waits: %+v, %v", waiting, err) + } + if _, known, err := inv.chainMergeOf(ctx, c, "novox/mesh-delivery", "c1"); err != nil || !known { + t.Fatalf("an answered merge is no longer kept: %v %v", known, err) + } +} From 39ae7de6d30edc2537cf1e0c7bcfb0218aece0ed Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 00:19:03 +0200 Subject: [PATCH 6/8] Say a first merge that waits past ten minutes for mesh-delivery's word, with probe D17 (issue 1146, AC8) --- cmd/mesh-controller/doctor.go | 5 +++ cmd/mesh-controller/first_builds.go | 49 ++++++++++++++++++++++++ cmd/mesh-controller/first_builds_test.go | 32 ++++++++++++++++ 3 files changed, 86 insertions(+) diff --git a/cmd/mesh-controller/doctor.go b/cmd/mesh-controller/doctor.go index 4358b135..d9e436c4 100644 --- a/cmd/mesh-controller/doctor.go +++ b/cmd/mesh-controller/doctor.go @@ -120,6 +120,11 @@ var probeRegistry = []probe{ {ID: probeDeliveriesID, Asserts: "no delivery is held past its state's bound unsaid: mesh-delivery's " + "`stalled`, each with the transition its table lets healer H2 take", From: "ADR 0239", Kind: kindDeliveryStalled, Phase: 3, run: probeDeliveries}, + // A first merge into a repository the mesh builds nothing from, kept waiting for mesh-delivery's word on the + // operator's Approve (novox/hq ADR 0432, issue 1146): said past a bound, never only in a log line (ADR 0426 §3, §8). + {ID: probeFirstBuildsID, Asserts: "no first merge into a repository the mesh builds nothing from waits more than ten " + + "minutes for mesh-delivery's announcement of the operator's Approve of its head", From: "ADR 0432, issue 1146", + Kind: kindFirstBuildUnasked, Phase: 3, run: probeFirstBuilds}, // The delivery budgets (novox/hq ADR 0282 decision 7): the newest delivery of each class within its budget, // read from mesh-delivery's `times`; a measurement said, never a delivery held. {ID: probeBudgetsID, Asserts: "the newest delivery of a leaf module ran on every machine within three minutes of " + diff --git a/cmd/mesh-controller/first_builds.go b/cmd/mesh-controller/first_builds.go index 31624282..aeda9d21 100644 --- a/cmd/mesh-controller/first_builds.go +++ b/cmd/mesh-controller/first_builds.go @@ -8,6 +8,7 @@ import ( "strings" "time" + "github.com/novox/mesh-controller/internal/conditions" "github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/link" ) @@ -161,3 +162,51 @@ func firstBuildsOf(ctx context.Context, inv catalogueReader, m link.SourceMoved, // Not announced yet, or the bus or the store could not be asked: the caller waits or tries again, and asks nothing. return nil, "", err } + +// probeFirstBuildsID is the probe that reads the first merges waiting for mesh-delivery's word. +const probeFirstBuildsID = "D17" + +// kindFirstBuildUnasked is what a first merge waiting too long for mesh-delivery's word raises. +const kindFirstBuildUnasked = "first-build-unasked" + +// firstBuildWaitBound is how long a first merge waits for mesh-delivery's announcement of the operator's Approve before +// it is said: a bound in the sense of ADR 0426 §3, not a delay. Nothing waits on it: the word's arrival asks the builds +// at once, and the merge is said, never dropped, once it is past this (issue 1146). +const firstBuildWaitBound = 10 * time.Minute + +// firstBuildObservations are the conditions of the first merges waiting past the bound, one for each; a merge answered is +// no longer listed as waiting, so its condition clears with the next look. +func firstBuildObservations(waiting []inventory.BatchedMerge, now time.Time) []conditions.Observation { + var out []conditions.Observation + for _, w := range waiting { + k := announcedOf(w) + if !k.FirstBuild || now.Sub(w.Heard) <= firstBuildWaitBound { + continue + } + repository := k.Owner + "/" + k.Repo + for_ := now.Sub(w.Heard).Round(time.Minute) + out = append(out, conditions.Observation{Scope: conditions.ScopeMerge, + ID: fmt.Sprintf("%s_%.8s", repository, k.Commit), Kind: kindFirstBuildUnasked, Severity: conditions.Warning, + Summary: fmt.Sprintf("the merge %.8s of %s adds modules the mesh builds nothing from yet, and has waited %s for "+ + "mesh-delivery's announcement of the operator's Approve of its head: no build is asked", k.Commit, repository, for_), + Said: fmt.Sprintf("%s at %s waits for mesh-delivery's word on pull request %d (head %.8s)", repository, k.Commit, + k.Number, k.HeadSHA), + Headline: fmt.Sprintf("The first build of %s is not asked: no Approve was announced", repository), + Explanation: fmt.Sprintf("The merge %.8s into %s adds modules the mesh builds nothing from yet. It registers them "+ + "only when mesh-delivery announces that it merged the pull request on your Approve of that head, and no such "+ + "announcement has come in %s. The merge is kept, and the builds are asked the moment the announcement is kept.", + k.Commit, repository, for_), + Resolved: "mesh-delivery announces the merge, or a hand `build` registers the modules", + }) + } + return out +} + +// probeFirstBuilds is D17: no first merge waits past its bound for mesh-delivery's word unsaid. +func probeFirstBuilds(ctx context.Context, d *doctor) ([]conditions.Observation, error) { + waiting, err := d.open.inventory.WaitingMerges(ctx) + if err != nil { + return nil, err + } + return firstBuildObservations(waiting, time.Now().UTC()), nil +} diff --git a/cmd/mesh-controller/first_builds_test.go b/cmd/mesh-controller/first_builds_test.go index fa22e3ed..7a6b2d14 100644 --- a/cmd/mesh-controller/first_builds_test.go +++ b/cmd/mesh-controller/first_builds_test.go @@ -2,12 +2,15 @@ package main import ( "context" + "encoding/json" "errors" "slices" + "strings" "testing" "time" "git.novox.be/novox/mesh-sdk/go/asks" + "github.com/novox/mesh-controller/internal/conditions" "github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/link" ) @@ -72,3 +75,32 @@ func TestAFirstMergeIntoANewRepositoryAsksItsModulesBuildsOnlyOnTheApprove(t *te }) } } + +// 1146/AC8: a first merge waiting past the bound for mesh-delivery's word is said, naming the repository, the commit and +// the missing word; one inside the bound, one that is no first merge, and one answered (no longer waiting) are not. +func TestAFirstMergeWaitingPastItsBoundIsSaidAndClearsWhenAnswered(t *testing.T) { + now := time.Date(2026, 10, 19, 1, 0, 0, 0, time.UTC) + waiting := func(commit string, first bool, heard time.Time) inventory.BatchedMerge { + event, _ := json.Marshal(keptMerge{SourceMoved: link.SourceMoved{Owner: "novox", Repo: "mesh-delivery", Base: "main", + Commit: commit, Number: 1, HeadSHA: "a1a1a1a1a1a1"}, FirstBuild: first}) + return inventory.BatchedMerge{Repository: "novox/mesh-delivery", Branch: "main", Commit: commit, Heard: heard, Event: event} + } + got := firstBuildObservations([]inventory.BatchedMerge{ + waiting("0177ea7a00", true, now.Add(-firstBuildWaitBound-time.Minute)), + waiting("1111111100", true, now.Add(-time.Minute)), + waiting("2222222200", false, now.Add(-time.Hour)), + }, now) + if len(got) != 1 { + t.Fatalf("1146/AC8: %d conditions raised, want the one merge past the bound: %+v", len(got), got) + } + o := got[0] + if o.Kind != kindFirstBuildUnasked || o.Scope != conditions.ScopeMerge || o.Severity != conditions.Warning || + !strings.Contains(o.Summary, "novox/mesh-delivery") || !strings.Contains(o.Summary, "0177ea7a") || + !strings.Contains(o.Summary, "announcement of the operator's Approve") { + t.Fatalf("1146/AC8: the condition is %+v", o) + } + // Answered, the merge is no longer listed as waiting: nothing is observed, and the condition clears. + if cleared := firstBuildObservations(nil, now); len(cleared) != 0 { + t.Fatalf("1146/AC8: an answered merge still raises %+v", cleared) + } +} From a08498b24023d84fba9efd35fd2d2705f9ae5add Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 00:20:20 +0200 Subject: [PATCH 7/8] Name the waiting first merge's probe D18: designs 06 and 54 already name probe D17 (issue 1146) --- cmd/mesh-controller/first_builds.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cmd/mesh-controller/first_builds.go b/cmd/mesh-controller/first_builds.go index aeda9d21..64c877de 100644 --- a/cmd/mesh-controller/first_builds.go +++ b/cmd/mesh-controller/first_builds.go @@ -164,7 +164,7 @@ func firstBuildsOf(ctx context.Context, inv catalogueReader, m link.SourceMoved, } // probeFirstBuildsID is the probe that reads the first merges waiting for mesh-delivery's word. -const probeFirstBuildsID = "D17" +const probeFirstBuildsID = "D18" // kindFirstBuildUnasked is what a first merge waiting too long for mesh-delivery's word raises. const kindFirstBuildUnasked = "first-build-unasked" From e621e15d2f5dec15da0358cfada5acfd7d257cbd Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 00:20:31 +0200 Subject: [PATCH 8/8] Its comment names D18 too (issue 1146) --- cmd/mesh-controller/first_builds.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cmd/mesh-controller/first_builds.go b/cmd/mesh-controller/first_builds.go index 64c877de..7283c44f 100644 --- a/cmd/mesh-controller/first_builds.go +++ b/cmd/mesh-controller/first_builds.go @@ -202,7 +202,7 @@ func firstBuildObservations(waiting []inventory.BatchedMerge, now time.Time) []c return out } -// probeFirstBuilds is D17: no first merge waits past its bound for mesh-delivery's word unsaid. +// probeFirstBuilds is D18: no first merge waits past its bound for mesh-delivery's word unsaid. func probeFirstBuilds(ctx context.Context, d *doctor) ([]conditions.Observation, error) { waiting, err := d.open.inventory.WaitingMerges(ctx) if err != nil {