From 7b7096f37ea76e19b5b6d98e43d5a637a6d4d006 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 00:28:56 +0100 Subject: [PATCH] Take-in, review round 1: no claim left behind, no sleep at the start, waits seeded at a restart, ends asked readable (issue 1585, ADR 0633) - A take-in releases its claims (compare-and-set, only while the claim is still its own) on a lost claim, on a claim it cannot read back, and on a merge it fails to take in: a claim left behind refused every later take-in until a restart. T5 row with a store that fails once, both ways. - The start step waits for nothing: with the build it runs unknown it says so and the forge consumer starts; a merged message left unacknowledged asks the build again (clause 1). T10 row: the start answers at once and takes nothing in. - A toolchain report evaluates pending again (ToolchainReported), since the build a revision names moves with the toolchain. T7 row: the revision is taken in once the report makes it name the running build. - The group's end is told by its whole subject (process.ModuleStreamSubject), so a repository named forge/group is never taken for one; row in the indexer's handle test. - The watchman asks the indexer's waits through indexer_pending at its start (invokes), before its durable consumer, which does not hear again what it acknowledged before a restart; the comment that said the whole stream is heard is corrected. W2 row: a wait seeded within its bound is raised past it. - bridge_groups names each group's last end ask, its who, why and when, and the bridge's answer (state end-asks). - The check's words for the indexer's rules: the trunk's manifest validator. - The bridge's end drops the merged-state test its merge commit already implies. - Test budget: groupask_test folded into groups_test, and the K1 mesh/check test into the owed status one: 804 unit tests (was 806). Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn --- .../internal/check/examiner_test.go | 16 +-- .../internal/check/manifests_judge.go | 25 ++--- .../mesh-delivery-forge-bridge-tools/main.go | 26 +++-- .../main_test.go | 13 ++- .../groupask_test.go | 64 ------------ .../cmd/mesh-delivery-forge-bridge/groups.go | 4 +- .../mesh-delivery-forge-bridge/groups_test.go | 46 ++++++++- .../internal/bridge/acts_test.go | 32 +++--- .../internal/groups/groups.go | 19 +++- .../cmd/mesh-delivery-indexer/main.go | 26 ++--- .../cmd/mesh-delivery-indexer/main_test.go | 7 ++ .../internal/index/index_test.go | 11 +++ .../internal/index/pending.go | 84 +++++++++++++--- .../internal/index/pending_test.go | 98 ++++++++++++++++++- .../internal/index/toolchains.go | 14 ++- .../cmd/mesh-delivery-watchman/main.go | 30 +++++- .../internal/watch/findings_test.go | 12 +++ .../internal/watch/takein.go | 19 ++++ .../mesh-delivery-watchman/manifest_test.go | 5 + modules/mesh-delivery-watchman/module.json | 3 + 20 files changed, 402 insertions(+), 152 deletions(-) delete mode 100644 modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groupask_test.go diff --git a/modules/mesh-delivery-checker/internal/check/examiner_test.go b/modules/mesh-delivery-checker/internal/check/examiner_test.go index fca34153..5cf259ff 100644 --- a/modules/mesh-delivery-checker/internal/check/examiner_test.go +++ b/modules/mesh-delivery-checker/internal/check/examiner_test.go @@ -1035,18 +1035,18 @@ func TestTheIndexersJudgeJudgesTheChangesManifests(t *testing.T) { // mail's tool while mail awaits web's database: both left out, so neither is refused (ADR 0614 §3, §6). cycle string trunkCycle bool - // indexer makes web the holder of the seat mesh-catalog, and the trunk's judge answers for it: "held" with no verdict, + // indexer makes web the holder of the seat mesh-catalog, and the trunk's manifest validator answers for it: "held" with no verdict, // "aside" with a field it does not know; ownJudge has the group change the judge, so the manifest is also judged by the // trunk's, in a second ask (ADR 0633 clause 6, 1585/T22, T23) indexer string ownJudge bool }{ {name: "accepted"}, - {name: "the indexer accepted by the trunk's judge", indexer: "ok"}, - {name: "the indexer held by the trunk's judge", indexer: "held", fails: "the indexer web, the holder of the seat mesh-catalog, would never run: the trunk's judge at mesh-delivery's trunk " + strings.Repeat("d", 40) + " does not accept its manifest: the judge holds it, with no verdict yet"}, - {name: "the indexer with a field the trunk's judge does not know", indexer: "aside", fails: "the judge does not know its fields new_field, so it waits"}, - {name: "the indexer refused by the trunk's judge", indexer: "refused", refused: "web has no id", fails: "does not accept its manifest: web has no id"}, - {name: "the indexer under a changed judge is judged by the trunk's as well", indexer: "aside", ownJudge: true, fails: "the judge does not know its fields new_field, so it waits"}, + {name: "the indexer accepted by the trunk's manifest validator", indexer: "ok"}, + {name: "the indexer held by the trunk's manifest validator", indexer: "held", fails: "the indexer web, the holder of the seat mesh-catalog, would never run: the trunk's manifest validator at mesh-delivery's trunk " + strings.Repeat("d", 40) + " does not accept its manifest: the manifest validator holds it, with no verdict yet"}, + {name: "the indexer with a field the trunk's manifest validator does not know", indexer: "aside", fails: "the manifest validator does not know its fields new_field, so it waits"}, + {name: "the indexer refused by the trunk's manifest validator", indexer: "refused", refused: "web has no id", fails: "does not accept its manifest: web has no id"}, + {name: "the indexer under a changed judge is judged by the trunk's as well", indexer: "aside", ownJudge: true, fails: "the manifest validator does not know its fields new_field, so it waits"}, {name: "a touched manifest the judge holds", indexer: "heldplain", fails: "holds the manifest of web: it gave no verdict"}, {name: "an untouched manifest's problem the trunk had", untouched: "mail says no credential form", trunkMail: "mail says no credential form"}, {name: "a problem the change causes in an untouched manifest", untouched: "mail reads a state web no longer keeps", @@ -1122,9 +1122,9 @@ func TestTheIndexersJudgeJudgesTheChangesManifests(t *testing.T) { cs := caseOf(t, p) v := b.judge(t, cs).Verdict if row.indexer == "ok" && !slices.ContainsFunc(cs.notes, func(n string) bool { - return strings.Contains(n, "the new indexer's manifest of web is accepted by the trunk's judge") + return strings.Contains(n, "the new indexer's manifest of web is accepted by the trunk's manifest validator") }) { - t.Errorf("%s: the case says %v, want the new indexer's manifest said accepted by the trunk's judge", row.name, cs.notes) + t.Errorf("%s: the case says %v, want the new indexer's manifest said accepted by the trunk's manifest validator", row.name, cs.notes) } if row.ownJudge && (len(judged) != 2 || judged[0].Fetch["commit"] != head1 || judged[1].Fetch["commit"] != trunk) { t.Fatalf("%s: judged %+v, want the head's judge and then the trunk's", row.name, judged) diff --git a/modules/mesh-delivery-checker/internal/check/manifests_judge.go b/modules/mesh-delivery-checker/internal/check/manifests_judge.go index c8320df3..c34e8308 100644 --- a/modules/mesh-delivery-checker/internal/check/manifests_judge.go +++ b/modules/mesh-delivery-checker/internal/check/manifests_judge.go @@ -206,12 +206,13 @@ func (st *state) runJudge(ctx context.Context, req composer.Request, touched []s // never by its name (novox/hq ADR 0633 clause 6, ADR 0601 §4). const CatalogSeat = "mesh-catalog" -// refuseUnmovableIndexer fails the check when the trunk's judge, the one the running parts hold, would refuse or hold the -// change's new indexer manifest: such an indexer would never run, and every revision behind it would wait for ever -// (ADR 0633 clause 6). The indexer is the touched module whose manifest claims the seat mesh-catalog. When the change -// judges with its own head's judge (it changes the judge), the manifest is judged by the trunk's as well. The verdicts -// read are the trunk's judge's: its change case, where refused, held (not accepted with no reason) or with a field set -// aside (a field the running judge does not know waits, ADR 0601 §3) each fail it, naming the module and the problems. +// refuseUnmovableIndexer fails the check when the trunk's manifest validator, the one the running parts hold, would +// refuse or hold the change's new indexer manifest: such an indexer would never run, and every revision behind it would +// wait for ever (ADR 0633 clause 6). The indexer is the touched module whose manifest claims the seat mesh-catalog. +// When the change judges with its own head's manifest validator (it changes it), the manifest is judged by the trunk's +// as well. The verdicts read are the trunk's manifest validator's: its change case, where refused, held (not accepted +// with no reason) or with a field set aside (a field the running manifest validator does not know waits, ADR 0601 §3) +// each fail it, naming the module and the problems. func (st *state) refuseUnmovableIndexer(ctx context.Context, req composer.Request, touched []string, cv composer.CaseVerdict, judgedAt string) *delivery.Verdict { if st.j.Read == nil { return nil @@ -237,7 +238,7 @@ func (st *state) refuseUnmovableIndexer(ctx context.Context, req composer.Reques trunk = st.j.Trunks.Heads()[DeliveryRepository] } if trunk == "" { - return fail(goModule, "the new indexer cannot be judged by the trunk's judge: the trunk head of "+DeliveryRepository+" is not known") + return fail(goModule, "the new indexer cannot be judged by the trunk's manifest validator: the trunk head of "+DeliveryRepository+" is not known") } var v *delivery.Verdict judgedAt = "mesh-delivery's trunk " + trunk @@ -249,19 +250,19 @@ func (st *state) refuseUnmovableIndexer(ctx context.Context, req composer.Reques var problems []string switch { case !ok: - problems = append(problems, "the judge gave no verdict") + problems = append(problems, "the manifest validator gave no verdict") case r.Refused != "": problems = append(problems, strings.Split(r.Refused, "\n")...) case !r.Accepted: - problems = append(problems, "the judge holds it, with no verdict yet") + problems = append(problems, "the manifest validator holds it, with no verdict yet") case r.SetAside != "": - problems = append(problems, "the judge does not know its fields "+r.SetAside+", so it waits") + problems = append(problems, "the manifest validator does not know its fields "+r.SetAside+", so it waits") } if len(problems) > 0 { - return fail(goModule, fmt.Sprintf("the indexer %s, the holder of the seat %s, would never run: the trunk's judge at %s does not accept its manifest: %s", + return fail(goModule, fmt.Sprintf("the indexer %s, the holder of the seat %s, would never run: the trunk's manifest validator at %s does not accept its manifest: %s", indexer, CatalogSeat, judgedAt, strings.Join(problems, "; "))) } - st.c.note(fmt.Sprintf("the new indexer's manifest of %s is accepted by the trunk's judge at %s", indexer, judgedAt)) + st.c.note(fmt.Sprintf("the new indexer's manifest of %s is accepted by the trunk's manifest validator at %s", indexer, judgedAt)) return nil } diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main.go index f3638d1e..86faa05a 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main.go @@ -37,7 +37,8 @@ func Tools(ask asker, now func() time.Time, groupsRead func() ([]groups.Row, []s Description: "The delivery groups the gate holds (more than one member), each with its members' merge state and whether each " + "merged message is written on the forge stream, and whether the group's end (forge.group.merged) is written: ended, " + "or open with why (gathered, or merged and its end not written yet). An open group with every member merged is " + - "what bridge_group_end_again says the end of. A group's name narrows it to that group; open lists only open ones.", + "what bridge_group_end_again says the end of. Each group names its last end ask (who, why, when) and the bridge's " + + "answer to it, from the state end-asks. A group's name narrows it to that group; open lists only open ones.", Input: map[string]any{"type": "object", "properties": map[string]any{ "group": map[string]any{"type": "string", "description": "only this group, by the name the gate keys it"}, "open": map[string]any{"type": "boolean", "description": "only the groups whose end is not written"}}}, @@ -80,8 +81,9 @@ func Tools(ask asker, now func() time.Time, groupsRead func() ([]groups.Row, []s } } -// read is the groups the gate holds with what the bridge kept of them. -func read(merges, performed stdio.KeptState) ([]groups.Row, []string, error) { +// read is the groups the gate holds with what the bridge kept of them: their ends in performed, and the last ask of +// each in end-asks with the bridge's answer. +func read(merges, performed, asks stdio.KeptState) ([]groups.Row, []string, error) { keys, err := merges.Keys() if err != nil { return nil, nil, fmt.Errorf("the gate's merges cannot be listed: %w", err) @@ -110,7 +112,17 @@ func read(merges, performed stdio.KeptState) ([]groups.Row, []string, error) { } } } - rows, unread := groups.Rows(raws, published, ended) + akeys, err := asks.Keys() + if err != nil { + return nil, nil, fmt.Errorf("the bridge's state %s cannot be listed: %w", groups.EndAsksState, err) + } + askRaws := map[string][]byte{} + for _, k := range akeys { + if e, err := asks.Get(k); err == nil && e != nil { + askRaws[k] = e.Value + } + } + rows, unread := groups.Rows(raws, published, ended, askRaws) return rows, unread, nil } @@ -164,9 +176,9 @@ func askEnd(s stdio.KeptState, bound time.Duration) asker { } func main() { - merges, performed := stdio.State(delivery.GateModule+".merges"), stdio.State(groups.PerformedState) - tools := Tools(askEnd(stdio.State(groups.EndAsksState), 25*time.Second), time.Now, - func() ([]groups.Row, []string, error) { return read(merges, performed) }) + merges, performed, asks := stdio.State(delivery.GateModule+".merges"), stdio.State(groups.PerformedState), stdio.State(groups.EndAsksState) + tools := Tools(askEnd(asks, 25*time.Second), time.Now, + func() ([]groups.Row, []string, error) { return read(merges, performed, asks) }) if err := stdio.Serve("", tools); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-forge-bridge-tools] %v\n", err) os.Exit(1) diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go index cdfe4250..2d1b9e6f 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go @@ -34,6 +34,10 @@ func TestTheBridgeToolsReadTheGroupsAndAskTheEndAgain(t *testing.T) { for _, c := range []string{c1, c2} { performed[groups.PublishedKey("novox/a", 1, c)], performed[groups.PublishedKey("novox/b", 2, c)] = true, true } + // The state end-asks keeps the last ask of the group ready, answered, and one that does not read. + answeredAsk, _ := json.Marshal(groups.EndAsk{V: 1, Group: "ready", Why: "lost", By: "lead@shanks", At: time.Date(2026, 11, 4, 22, 0, 0, 0, time.UTC), + Written: true, Answer: "the end of the group is written again", Answered: time.Date(2026, 11, 4, 22, 0, 1, 0, time.UTC)}) + asks := map[string][]byte{"ready": answeredAsk, "gathered": []byte("{")} var asked groups.EndAsk var askedKey string ask := func(key string, value []byte) (groups.EndAsk, error) { @@ -45,7 +49,7 @@ func TestTheBridgeToolsReadTheGroupsAndAskTheEndAgain(t *testing.T) { } tools := Tools(ask, func() time.Time { return time.Date(2026, 11, 4, 23, 0, 0, 0, time.UTC) }, func() ([]groups.Row, []string, error) { - rows, unread := groups.Rows(merges, performed, map[string]uint64{"ended": 7}) + rows, unread := groups.Rows(merges, performed, map[string]uint64{"ended": 7}, asks) return rows, unread, nil }) call := func(name string, args map[string]any) any { @@ -74,8 +78,8 @@ func TestTheBridgeToolsReadTheGroupsAndAskTheEndAgain(t *testing.T) { for _, r := range out.Groups { m[r.Group] = r } - if len(args) == 0 && (len(out.Unread) != 1 || out.Unread[0] != "bad") { - t.Errorf("unread = %v, want bad", out.Unread) + if len(args) == 0 && strings.Join(out.Unread, " ") != "bad end-asks.gathered" { + t.Errorf("unread = %v, want bad and end-asks.gathered", out.Unread) } return m } @@ -86,6 +90,9 @@ func TestTheBridgeToolsReadTheGroupsAndAskTheEndAgain(t *testing.T) { if r := all["ready"]; !r.Open || !strings.Contains(r.Words, "bridge_group_end_again") || !r.Members[1].Published { t.Errorf("ready = %+v, want open, merged and published, its end the thing to say", r) } + if a := all["ready"].LastEndAsk; a == nil || a.By != "lead@shanks" || a.Why != "lost" || !a.Written || a.Answer == "" || all["ended"].LastEndAsk != nil { + t.Errorf("ready's last end ask = %+v, ended's %+v: want ready's ask and its answer, none for ended", a, all["ended"].LastEndAsk) + } if r := all["gathered"]; !r.Open || !strings.Contains(r.Words, "gathered") || r.Members[1].State != string(delivery.MergeMerging) { t.Errorf("gathered = %+v", r) } diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groupask_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groupask_test.go deleted file mode 100644 index de6ded88..00000000 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groupask_test.go +++ /dev/null @@ -1,64 +0,0 @@ -package main - -import ( - "bytes" - "context" - "log" - "strings" - "testing" - "time" - - "git.novox.be/novox/mesh-sdk/go/delivery" - - "mesh-delivery-forge-bridge/internal/groups" -) - -// 1585/T19 (ADR 0633 §4): an ask to say a group's end again is answered by the bridge from the gate's merges, with the ask's -// time on the message id; not written while a member is not merged, with the reason; not for a group the gate does not hold; -// and the keys of performed the tools read are the ones the process writes. -func TestAnEndAskSaysTheGroupsEndAgainOrWhyNot(t *testing.T) { - ctx := context.Background() - h1, h2 := strings.Repeat("1", 40), strings.Repeat("2", 40) - c1, c2 := strings.Repeat("c", 40), strings.Repeat("d", 40) - pull := func(repo string, n int, head string, state delivery.MergeState, commit string) delivery.MergePull { - return delivery.MergePull{Head: delivery.Head{Repository: repo, Number: n, Head: head}, State: state, Commit: commit} - } - merge := func(second delivery.MergeState, commit2 string) []byte { - v, err := delivery.EncodeMerge(delivery.Merge{Group: "feat/take-in", Ask: "ask", Approved: map[string]string{"novox/mesh-sdk@37": h1, "novox/mesh-delivery@9": h2}, - Pulls: []delivery.MergePull{pull("novox/mesh-sdk", 37, h1, delivery.MergeMerged, c1), pull("novox/mesh-delivery", 9, h2, second, commit2)}}) - if err != nil { - t.Fatal(err) - } - return v - } - ask := groups.EndAsk{V: 1, Group: "feat/take-in", By: "operator@shanks", Why: "the indexer never took it in", At: time.Date(2026, 11, 4, 23, 0, 0, 123, time.UTC)} - for _, tc := range []struct { - name string - gate map[string][]byte - published bool - written bool - answer string - }{ - {"merged and published: written again", map[string][]byte{"feat/take-in": merge(delivery.MergeMerged, c2)}, true, true, "written again"}, - {"a member's message not written", map[string][]byte{"feat/take-in": merge(delivery.MergeMerged, c2)}, false, false, "not written"}, - {"a member not merged", map[string][]byte{"feat/take-in": merge(delivery.MergeMerging, "")}, true, false, "not written"}, - {"no merge at the gate", map[string][]byte{}, true, false, "holds no merge"}, - } { - pub, done := &idPub{}, memRevisions{} - if tc.published { - done[groups.PublishedKey("novox/mesh-sdk", 37, c1)], done[groups.PublishedKey("novox/mesh-delivery", 9, c2)] = 1, 1 - } - g := &grouper{send: pub.send, done: done, logger: log.New(&bytes.Buffer{}, "", 0), - merges: func(_ context.Context, key string) ([]byte, uint64, bool, error) { - v, ok := tc.gate[key] - return v, 4, ok, nil - }} - got := g.answerEnd(ctx, "feat/take-in", ask, ask.At.Add(time.Second)) - if got.Written != tc.written || !strings.Contains(got.Answer, tc.answer) || !got.Done() || len(pub.ids) != map[bool]int{true: 1}[tc.written] { - t.Errorf("%s: %+v, ids %v", tc.name, got, pub.ids) - } - if tc.written && pub.ids[0] != groups.EndID("feat/take-in", c2, groups.AgainStamp(ask.At)) { - t.Errorf("%s: the id %s does not carry the ask's time", tc.name, pub.ids[0]) - } - } -} diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups.go index f00c6688..82b1eb12 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups.go @@ -169,8 +169,8 @@ func (g *grouper) end(ctx context.Context, key string, rev uint64, value []byte, at := slices.IndexFunc(m.Pulls, func(p delivery.MergePull) bool { return delivery.SameRepository(p.Repository, member.Repository) && p.Number == member.Number }) - if at < 0 || m.Pulls[at].State != delivery.MergeMerged || m.Pulls[at].Commit == "" { - return false // not every member is merged: the group stays gathered + if at < 0 || m.Pulls[at].Commit == "" { + return false // not every member is merged (a merge commit is written only for a merged pull): the group stays gathered } if _, ok, err := g.done.Get(ctx, groupkeys.PublishedKey(member.Repository, member.Number, m.Pulls[at].Commit)); err != nil || !ok { return false // its merged message is not written yet: said when it is (published) diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups_test.go index 17c57b22..a81c2c41 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups_test.go @@ -6,6 +6,7 @@ import ( "log" "strings" "testing" + "time" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -24,7 +25,7 @@ func (p *idPub) send(_ context.Context, subject string, data []byte, id string) return nil } -func TestAGroupsMembersNameItAndItsEndIsSaidOnce(t *testing.T) { // 1585/T16, T17 (ADR 0633 §4) +func TestAGroupsMembersNameItAndItsEndIsSaidOnce(t *testing.T) { // 1585/T16, T17, T19 (ADR 0633 §4) ctx := context.Background() h1, h2 := strings.Repeat("1", 40), strings.Repeat("2", 40) c1, c2 := strings.Repeat("c", 40), strings.Repeat("d", 40) @@ -94,4 +95,47 @@ func TestAGroupsMembersNameItAndItsEndIsSaidOnce(t *testing.T) { // 1585/T16, T1 if !g.end(ctx, "feat/take-in", 10, gate["feat/take-in"], "20261104T2300") || pub.ids[1] == pub.ids[0] { t.Errorf("the end again by hand: ids %v", pub.ids) } + // 1585/T19: an ask to say a group's end again is answered by the bridge from the gate's merges, with the ask's time on + // the message id; not written while a member is not merged, with the reason; not for a group the gate does not hold. + pull := func(repo string, n int, head string, state delivery.MergeState, commit string) delivery.MergePull { + return delivery.MergePull{Head: delivery.Head{Repository: repo, Number: n, Head: head}, State: state, Commit: commit} + } + askMerge := func(second delivery.MergeState, commit2 string) []byte { + v, err := delivery.EncodeMerge(delivery.Merge{Group: "feat/take-in", Ask: "ask", Approved: map[string]string{"novox/mesh-sdk@37": h1, "novox/mesh-delivery@9": h2}, + Pulls: []delivery.MergePull{pull("novox/mesh-sdk", 37, h1, delivery.MergeMerged, c1), pull("novox/mesh-delivery", 9, h2, second, commit2)}}) + if err != nil { + t.Fatal(err) + } + return v + } + ask := groupkeys.EndAsk{V: 1, Group: "feat/take-in", By: "operator@shanks", Why: "the indexer never took it in", At: time.Date(2026, 11, 4, 23, 0, 0, 123, time.UTC)} + for _, tc := range []struct { + name string + gate map[string][]byte + published bool + written bool + answer string + }{ + {"merged and published: written again", map[string][]byte{"feat/take-in": askMerge(delivery.MergeMerged, c2)}, true, true, "written again"}, + {"a member's message not written", map[string][]byte{"feat/take-in": askMerge(delivery.MergeMerged, c2)}, false, false, "not written"}, + {"a member not merged", map[string][]byte{"feat/take-in": askMerge(delivery.MergeMerging, "")}, true, false, "not written"}, + {"no merge at the gate", map[string][]byte{}, true, false, "holds no merge"}, + } { + pub, done := &idPub{}, memRevisions{} + if tc.published { + done[groupkeys.PublishedKey("novox/mesh-sdk", 37, c1)], done[groupkeys.PublishedKey("novox/mesh-delivery", 9, c2)] = 1, 1 + } + g := &grouper{send: pub.send, done: done, logger: log.New(&bytes.Buffer{}, "", 0), + merges: func(_ context.Context, key string) ([]byte, uint64, bool, error) { + v, ok := tc.gate[key] + return v, 4, ok, nil + }} + got := g.answerEnd(ctx, "feat/take-in", ask, ask.At.Add(time.Second)) + if got.Written != tc.written || !strings.Contains(got.Answer, tc.answer) || !got.Done() || len(pub.ids) != map[bool]int{true: 1}[tc.written] { + t.Errorf("%s: %+v, ids %v", tc.name, got, pub.ids) + } + if tc.written && pub.ids[0] != groupkeys.EndID("feat/take-in", c2, groupkeys.AgainStamp(ask.At)) { + t.Errorf("%s: the id %s does not carry the ask's time", tc.name, pub.ids[0]) + } + } } diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/acts_test.go b/modules/mesh-delivery-forge-bridge/internal/bridge/acts_test.go index 26fa4f41..c9b9fe2d 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/acts_test.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/acts_test.go @@ -28,7 +28,7 @@ func owedValue(t *testing.T, repository string, number int, act string, o delive return key, b } -func TestOwedStatusIsACommitStatusCall(t *testing.T) { // 1341/AC1 +func TestOwedStatusIsACommitStatusCall(t *testing.T) { // 1341/AC1, K1 key, v := owedValue(t, "novox/mesh-delivery", 12, "status=mesh/merge-gate", delivery.Owed{Act: delivery.OwedStatus, Context: "mesh/merge-gate", State: "pending", Text: "waiting", Target: "https://forge.example/novox/mesh-delivery/pulls/12"}) a, err := OwedAct(key, v) @@ -43,6 +43,19 @@ func TestOwedStatusIsACommitStatusCall(t *testing.T) { // 1341/AC1 if a.Repository != "novox/mesh-delivery" || a.Number != 12 || a.Head != headA || a.Name != "status=mesh/merge-gate" { t.Errorf("act = %+v", a) } + // R4e findings K1 and R2-1 (issue 1301 S4; ADR 0503 §3, ADR 0521): no tool writes mesh/check, since a tool is told no + // caller and node-tools invokes every tool; the forge module writes it from the checker's owed state, and the bridge + // performs no mesh/check act. + ck, cv := checkerKV(t, delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://view.example/checks/g.1"}) + if a, err := OwedAct(ck, cv); err == nil || !strings.Contains(err.Error(), "mesh-delivery-checker.owed") { + t.Fatalf("OwedAct(mesh/check) = %+v, %v; want refused, naming where the forge module reads it", a, err) + } + // Nor through the gate's own owed bucket: the context is refused whichever key carries it. + gk, gv := owedValue(t, "novox/mesh-delivery", 12, "status=mesh/check", + delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://x.example"}) + if a, err := OwedAct(gk, gv); err == nil || !strings.Contains(err.Error(), "mesh-delivery-checker.owed") { + t.Fatalf("OwedAct(gate's mesh/check) = %+v, %v; want refused", a, err) + } } // checkerKV is the status the checker owes, as it writes it (mesh-sdk go/delivery CheckerOwedKey, EncodeCheckerOwed). @@ -59,23 +72,6 @@ func checkerKV(t *testing.T, o delivery.Owed) (string, []byte) { return key, v } -// R4e findings K1 and R2-1 (issue 1301 S4; ADR 0503 §3, ADR 0521): no tool writes mesh/check, since a tool is told no -// caller and node-tools invokes every tool; the forge module writes it from the checker's owed state, and the bridge -// performs no mesh/check act. -func TestOwedMeshCheckIsNoToolCall(t *testing.T) { // K1 - key, v := checkerKV(t, delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://view.example/checks/g.1"}) - a, err := OwedAct(key, v) - if err == nil || !strings.Contains(err.Error(), "mesh-delivery-checker.owed") { - t.Fatalf("OwedAct(mesh/check) = %+v, %v; want refused, naming where the forge module reads it", a, err) - } - // Nor through the gate's own owed bucket: the context is refused whichever key carries it. - gk, gv := owedValue(t, "novox/mesh-delivery", 12, "status=mesh/check", - delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://x.example"}) - if a, err := OwedAct(gk, gv); err == nil || !strings.Contains(err.Error(), "mesh-delivery-checker.owed") { - t.Fatalf("OwedAct(gate's mesh/check) = %+v, %v; want refused", a, err) - } -} - func TestOwedViewNoteAndCommentCalls(t *testing.T) { // 1341/AC2 cases := []struct { act string diff --git a/modules/mesh-delivery-forge-bridge/internal/groups/groups.go b/modules/mesh-delivery-forge-bridge/internal/groups/groups.go index 006e1703..bd09dbb7 100644 --- a/modules/mesh-delivery-forge-bridge/internal/groups/groups.go +++ b/modules/mesh-delivery-forge-bridge/internal/groups/groups.go @@ -102,6 +102,9 @@ type Row struct { EndedAt uint64 `json:"ended-at-revision,omitempty"` Open bool `json:"open"` Words string `json:"words"` + // LastEndAsk is the last ask to say the group's end again (bridge_group_end_again), its who, why and when, and the + // bridge's answer once it read it; absent when none was asked. The state end-asks keeps the last per group. + LastEndAsk *EndAsk `json:"last-end-ask,omitempty"` } // RowMember is one approved member of a group: its merge's state at the gate and whether its merged message is written. @@ -116,8 +119,18 @@ type RowMember struct { // Rows are the groups the gate's merges hold with more than one member, each with what the bridge kept: ended (its end // written, the revision it was worked out from) or open (some member is not merged or its merged message not written, or its // end not written). merges is the gate's merges by key, performed the bridge's performed keys by key; unread are the gate's -// keys that do not read. -func Rows(merges map[string][]byte, performed map[string]bool, endedAt map[string]uint64) (rows []Row, unread []string) { +// keys that do not read. asks are the end-asks values by the same key; one that does not read is named unread as +// `end-asks.`. +func Rows(merges map[string][]byte, performed map[string]bool, endedAt map[string]uint64, asks map[string][]byte) (rows []Row, unread []string) { + lastAsk := map[string]*EndAsk{} + for key, raw := range asks { + a, err := DecodeEndAsk(raw) + if err != nil { + unread = append(unread, EndAsksState+"."+key) + continue + } + lastAsk[key] = &a + } for key, raw := range merges { m, err := delivery.DecodeMerge(raw) if err != nil { @@ -140,6 +153,7 @@ func Rows(merges map[string][]byte, performed map[string]bool, endedAt map[strin allMerged = false } r.EndedAt, r.Ended = endedAt[key] + r.LastEndAsk = lastAsk[key] r.Open = !r.Ended switch { case r.Ended: @@ -152,6 +166,7 @@ func Rows(merges map[string][]byte, performed map[string]bool, endedAt map[strin rows = append(rows, r) } sortRows(rows) + sort.Strings(unread) return rows, unread } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index ce9d794b..3c24a297 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -16,7 +16,6 @@ import ( "log" "os" "os/signal" - "strings" "sync/atomic" "syscall" "time" @@ -155,6 +154,10 @@ type delivered interface { Metadata() (*jetstream.MsgMetadata, error) } +// groupEndSubject is the whole subject of a delivery group's end on the forge bridge's stream: compared exactly, so a +// repository named forge/group, whose merged message ends the same way, is never taken for one. +var groupEndSubject = process.ModuleStreamSubject(delivery.ForgeBridgeModule, forgemsg.ForgeGroupEndEvent) + // handleGroupEnd takes a group's end in (index.EndGroup): a body that does not read is terminated and said; an end // whose members are not all held yet, or while the build this program runs is not known, is left for redelivery. func handleGroupEnd(ctx context.Context, m delivered, ix *index.Indexer, h *index.Heard, logger *log.Logger) { @@ -198,7 +201,7 @@ func handle(ctx context.Context, m delivered, ix *index.Indexer, heard index.Sto } }() // A delivery group's end (novox/hq ADR 0633 §4) has a body of its own: told apart by its name, before the decode. - if strings.HasSuffix(m.Subject(), "."+forgemsg.ForgeGroupEndEvent) { + if m.Subject() == groupEndSubject { handleGroupEnd(ctx, m, ix, &h, logger) return } @@ -409,7 +412,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if !gate.Toolchain(module, key, value, deleted) { return nil } - return ix.Recompute(ctx) + // The modules written again, and pending evaluated again (ADR 0633 §1): the build a revision names moves with + // the toolchain. + return ix.ToolchainReported(ctx) }, recompute: ix.Recompute}, logger) if err := tools.start(); err != nil { @@ -500,18 +505,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro defer pc.Stop() } // The pending revisions that name this build are taken in now, oldest first, before the consumer on the forge bridge's - // stream starts (ADR 0633 §2); the build this program runs is asked until it is known, said every minute. A hand - // take-in (indexer_take_in_pending) is answered meanwhile: the takeins watch runs already. - if _, words, err := ix.TakeInAtStart(ctx, func(ctx context.Context) bool { - // A context's bound between two asks of the node-engine, no timer (ADR 0437 §8), as askTakeIn bounds its wait. - again, cancel := context.WithTimeout(ctx, time.Minute) - defer cancel() - <-again.Done() - return ctx.Err() == nil - }); err != nil { + // stream starts (ADR 0633 §2). When the build this program runs is not known yet, nothing waits for it: the consumer + // starts, and a merged message left unacknowledged asks the build again at its redelivery (clause 1). A hand take-in + // (indexer_take_in_pending) is answered at any time: the takeins watch runs already. + if _, words, err := ix.TakeInAtStart(ctx); err != nil { return fmt.Errorf("the take-in at start: %w", err) } else if words != "" { - logger.Printf("take-in at start: still waiting: %s", words) + logger.Printf("take-in at start: %s", words) } cons, err := cred.EventConsumer(ctx, js, forgeStream) if err != nil { diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go index c4c3672c..b63c77ed 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go @@ -121,6 +121,13 @@ func TestAMessageThatIsNoForgeMessageIsTerminatedAndLogged(t *testing.T) { // 12 t.Errorf("heard of a group's end = %+v", h) } } + // The repository forge/group's merged message ends as a group's end does: it is a forge message all the same. + logs.Reset() + m = &fakeMsg{subject: "mesh.stream.mesh-delivery-forge-bridge.event.forge.forge.group.merged", data: []byte(`{"v":1,"group":"feat/x","members":[{"repository":"novox/mesh-sdk","number":3,"merge":"` + head + `"}]}`)} + handle(context.Background(), m, &index.Indexer{Store: st, Builds: index.NewBuilds()}, heard, log.New(&logs, "", 0)) + if !strings.Contains(logs.String(), "not a forge message") { + t.Errorf("forge/group's merged message read as a group's end: %q", logs.String()) + } } func TestTheSourceMainKeyIsTakenAtItsRevision(t *testing.T) { // 1344/AC1 (the watch's side: key, revision, value) diff --git a/modules/mesh-delivery-indexer/internal/index/index_test.go b/modules/mesh-delivery-indexer/internal/index/index_test.go index a29adba7..8c0cff64 100644 --- a/modules/mesh-delivery-indexer/internal/index/index_test.go +++ b/modules/mesh-delivery-indexer/internal/index/index_test.go @@ -26,6 +26,7 @@ type memStore struct { next uint64 writes int beforeWrite func() + failWrite func(key string) error // a write's own error, asked before each Create and Update } func newMem() *memStore { return &memStore{vals: map[string][]byte{}, revs: map[string]uint64{}} } @@ -59,6 +60,11 @@ func (m *memStore) put(key string, value []byte) { } func (m *memStore) Create(_ context.Context, key string, value []byte) error { m.hook() + if f := m.failWrite; f != nil { + if err := f(key); err != nil { + return err + } + } if _, ok := m.vals[key]; ok { return ErrConflict } @@ -68,6 +74,11 @@ func (m *memStore) Create(_ context.Context, key string, value []byte) error { } func (m *memStore) Update(_ context.Context, key string, value []byte, rev uint64) error { m.hook() + if f := m.failWrite; f != nil { + if err := f(key); err != nil { + return err + } + } if _, ok := m.vals[key]; !ok || m.revs[key] != rev { return ErrConflict } diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 19e91f1b..3eca3a36 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -396,21 +396,20 @@ func (ix *Indexer) takenAlready(ctx context.Context, m forgemsg.GroupMerge) (boo } // TakeInAtStart is the take-in at the start (ADR 0633 §2), before the consumer on the forge bridge's stream starts: -// it asks the build this program runs until it is known (wait says when to ask again, false ends it), then evaluates -// pending, taking in the whole revisions that name this build; a group still being gathered stays, and a revision -// naming yet another build is held for that one. A claim left by an indexer that ran before is void here. -func (ix *Indexer) TakeInAtStart(ctx context.Context, wait func(context.Context) bool) (Applied, string, error) { +// it evaluates pending once with the build this program runs, taking in the whole revisions that name this build; a +// group still being gathered stays, and a revision naming yet another build is held for that one. A claim left by an +// indexer that ran before is void here. When the build is not known (the node-engine did not answer), it waits for +// nothing: it says so and the consumer starts all the same; a merged message is then left unacknowledged until the +// build is known (clause 1), and its redelivery, or a toolchain report (ToolchainReported), asks again. No timer. +func (ix *Indexer) TakeInAtStart(ctx context.Context) (Applied, string, error) { if ix.Pending == nil || !ix.pendingOnBus(ctx) { // Nothing can wait in a state not on the bus: the start asks no build and does not wait for one. return Applied{}, "the pending state is not on the bus yet: nothing waits", nil } own := ix.ownBuildNow(ctx) - for own == "" { - ix.logf("take-in: the build this program runs is not known yet, so nothing pending is taken in; asked again") - if !wait(ctx) { - return Applied{}, "", ErrOwnBuildUnknown - } - own = ix.ownBuildNow(ctx) + if own == "" { + ix.logf("take-in: the build this program runs is not known at its start, so nothing pending is taken in now; the forge consumer starts, and a merged message left unacknowledged asks it again") + return Applied{}, "the build this program runs is not known yet: nothing pending is taken in at the start", nil } ix.pendingMu.Lock() defer ix.pendingMu.Unlock() @@ -421,6 +420,29 @@ func (ix *Indexer) TakeInAtStart(ctx context.Context, wait func(context.Context) return a, words, err } +// PendingAgain evaluates pending again after something a fingerprint is computed from changed, a toolchain report +// (ADR 0633 §1, §3): a revision that now names the running build is taken in, and one that names another build is held +// for that one. It does nothing while the pending state is not on the bus or the build this program runs is not known. +func (ix *Indexer) PendingAgain(ctx context.Context) error { + if ix.Pending == nil || !ix.pendingOnBus(ctx) { + return nil + } + own := ix.ownBuildNow(ctx) + if own == "" { + return nil + } + ix.pendingMu.Lock() + defer ix.pendingMu.Unlock() + _, words, err := ix.evaluate(ctx, own, false) + if errors.Is(err, process.ErrNotYet) { + return nil + } + if err == nil && words != "" { + ix.logf("take-in: evaluated again after a toolchain report: %s", words) + } + return err +} + // evaluate works out what pending's ready merges name, together (ADR 0633 §3), and takes them in or holds them. func (ix *Indexer) evaluate(ctx context.Context, own string, start bool) (Applied, string, error) { entries, err := ReadPending(ctx, ix.Pending, ix.logf) @@ -547,8 +569,9 @@ func (ix *Indexer) computeOver(ctx context.Context, ready []PendingEntry) (strin return m.FP2, d, nil } -// claim marks the entries as taken in by this take-in, by compare-and-set on the revisions read; a lost one leaves -// every claim made and answers false, and nothing is taken in. +// claim marks the entries as taken in by this take-in, by compare-and-set on the revisions read. A lost one (another +// take-in wrote the key, or the store failed) releases every claim this take-in made and answers false: nothing is +// taken in, and no claim of this one is left behind. func (ix *Indexer) claim(ctx context.Context, entries []PendingEntry, own, by string) ([]PendingEntry, bool) { id := fmt.Sprintf("%s at %s", own, ix.now().Format(time.RFC3339Nano)) if by != "" { @@ -564,18 +587,52 @@ func (ix *Indexer) claim(ctx context.Context, entries []PendingEntry, own, by st } if err != nil { ix.logf("take-in: the claim of %s.%s is lost (%v): another take-in took it, so this one takes nothing in", PendingState, e.Key, err) + ix.release(ctx, claimed) return nil, false } + e.Value = p _, rev, _, err := ix.Pending.Get(ctx, e.Key) if err != nil { + ix.logf("take-in: the claim of %s.%s cannot be read back (%v), so this take-in takes nothing in", PendingState, e.Key, err) + ix.release(ctx, append(claimed, e)) return nil, false } - e.Revision, e.Value = rev, p + e.Revision = rev claimed = append(claimed, e) } return claimed, true } +// release clears this take-in's claim on each entry, so a later merge, a restart or a hand take-in takes it in. It +// reads the key again and clears it by compare-and-set only while the claim is still this take-in's: a key another +// take-in wrote since, or removed, is left to it. A release that fails is said; the start of the next indexer voids +// the claim (TakeInAtStart). +func (ix *Indexer) release(ctx context.Context, claimed []PendingEntry) { + for _, e := range claimed { + value, rev, found, err := ix.Pending.Get(ctx, e.Key) + if err == nil && !found { + continue + } + var cur Pending + if err == nil { + err = json.Unmarshal(value, &cur) + } + if err == nil && cur.Taking != e.Value.Taking { + continue + } + if err == nil { + cur.Taking = "" + var data []byte + if data, err = encodePending(cur); err == nil { + err = ix.Pending.Update(ctx, e.Key, data, rev) + } + } + if err != nil { + ix.logf("take-in: the claim of %s.%s is not released (%v): the next indexer's start voids it", PendingState, e.Key, err) + } + } +} + // takeInWhole takes the entries in with this indexer, as merges are taken in (ApplyMessage), oldest first, then // removes them from pending and says each wait they were said under taken. by names a hand act ("" for none). func (ix *Indexer) takeInWhole(ctx context.Context, entries []PendingEntry, own, by string) (Applied, error) { @@ -600,6 +657,7 @@ func (ix *Indexer) takeInWholeBy(ctx context.Context, entries []PendingEntry, ow continue } if err != nil { + ix.release(ctx, claimed) return applied, fmt.Errorf("taking %s (%s#%d) in: %w", e.Key, e.Forge.Repository, e.Forge.Number, err) } } diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index 51d7157f..966f599f 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -118,7 +118,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) // Before the pending state is on the bus nothing waits: the start asks no build, and the merge is taken in at once, // as before ADR 0633, though the build is unknown (TestIndexerWhole found the start waiting on it for ever). b.ix.Pending = &memStore{unbound: process.ErrNotYet} - if _, _, err := b.ix.TakeInAtStart(ctx, func(context.Context) bool { t.Fatal("the start waits for the build"); return false }); err != nil { + if _, _, err := b.ix.TakeInAtStart(ctx); err != nil { t.Fatal(err) } if _, _, err := b.ix.Arrive(ctx, 1, b.merge(20, sha('b'), "tree-avahi-2", "")); err != nil || b.commit(t, "avahi") != sha('b') { @@ -183,7 +183,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) next.toolchains = b.ix.toolchains asked := 0 next.OwnBuild = func(context.Context) (string, error) { - if asked++; asked < 2 { + if asked++; asked < 3 { // the hand ask's and the first start's return "", errors.New("not yet") } return named, nil @@ -191,10 +191,15 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) if r := next.TakeInPendingByHand(ctx, "m.nothing", "lead@shanks", "a test"); r.Stored || !strings.Contains(r.Answer, "nothing waits") { t.Errorf("a hand ask while the own build is unknown: %+v", r) } - if _, _, err := next.TakeInAtStart(ctx, func(context.Context) bool { return true }); err != nil { + // The first start, its build not known, waits for nothing (no sleep, no timer): it answers at once, takes nothing + // in, and the forge consumer starts after it. + if _, words, err := next.TakeInAtStart(ctx); err != nil || asked != 2 || len(b.keys()) != 2 || !strings.Contains(words, "not known") { + t.Fatalf("a start with its build unknown: %q %v, asked %d, pending %v: want it to answer at once and take nothing in", words, err, asked, b.keys()) + } + if _, _, err := next.TakeInAtStart(ctx); err != nil { t.Fatal(err) } - if asked < 2 || !reflect.DeepEqual(b.keys(), []string{takein.PendingKey(5)}) || b.commit(t, "avahi") != sha('c') { + if asked < 3 || !reflect.DeepEqual(b.keys(), []string{takein.PendingKey(5)}) || b.commit(t, "avahi") != sha('c') { t.Errorf("start: asked %d, pending %v, avahi at %s: want the revision taken in, the group left", asked, b.keys(), b.commit(t, "avahi")) } }) @@ -275,13 +280,96 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) data, _ := json.Marshal(p) b.pending.put(takein.PendingKey(2), data) } - if _, _, err := b.ix.TakeInAtStart(ctx, func(context.Context) bool { return false }); err != nil { + if _, _, err := b.ix.TakeInAtStart(ctx); err != nil { t.Fatal(err) } if b.commit(t, "avahi") == sha('c') || len(b.keys()) != 1 || !strings.Contains(b.logs.String(), "is lost") { t.Errorf("avahi at %s, pending %v: want nothing taken in by the loser, and it said", b.commit(t, "avahi"), b.keys()) } }) + t.Run("T5 a take-in whose store fails once leaves no claim behind, and is taken in after", func(t *testing.T) { + claims := func(b *takeInBed) []string { + var out []string + for _, k := range b.keys() { + var p Pending + _ = json.Unmarshal(b.pending.vals[k], &p) + if p.Taking != "" { + out = append(out, k) + } + } + return out + } + once := func(match func(key string) bool) func(string) error { + done := false + return func(key string) error { + if !done && match(key) { + done = true + return errors.New("the store failed once") + } + return nil + } + } + for _, c := range []struct { + name string + fail func(b *takeInBed) + }{ + // The second merge's claim fails: the first's, already made, is released. + {"a claim lost", func(b *takeInBed) { + b.pending.failWrite = once(func(k string) bool { return k == takein.PendingKey(3) }) + }}, + // The take-in itself fails on a module's write after both claims: both are released. + {"a take-in that fails", func(b *takeInBed) { b.mods.failWrite = once(func(string) bool { return true }) }}, + } { + b := newTakeInBed(t) + b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2")) + b.arrive(t, 3, b.merge(22, sha('d'), "tree-avahi-4", "tree-indexer-2")) + named := b.fp(t, delivery.IndexerModule) + b.ix.ownMu.Lock() + b.ix.ownFP = named + b.ix.ownMu.Unlock() + c.fail(b) + _, _, _ = b.ix.TakeInAtStart(ctx) + if got := claims(b); len(got) != 0 || len(b.keys()) != 2 { + t.Fatalf("%s: pending %v, claimed %v: want both pending and no claim left", c.name, b.keys(), got) + } + // The next take-in, not at a start, is not refused by a claim left behind: it takes both in. + if _, _, err := b.ix.Arrive(ctx, 4, b.merge(23, sha('e'), "tree-avahi-5", "tree-indexer-2")); err != nil { + t.Fatal(err) + } + if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('e') { + t.Errorf("%s: after: pending %v, avahi at %s: want all taken in", c.name, b.keys(), b.commit(t, "avahi")) + } + } + }) + t.Run("T7 a toolchain report evaluates pending again: the revision then names the running build and is taken in", func(t *testing.T) { + b := newTakeInBed(t) + b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2")) + entries, err := ReadPending(ctx, b.pending, nil) + if err != nil { + t.Fatal(err) + } + // The build the revision names under the next toolchain, worked out on a dry indexer; this one runs it. + x := b.ix.dry().ix + for _, node := range []string{"novox", "ace"} { + key, _ := delivery.ToolchainKey(node, "go") + x.PutToolchain("mesh-build-go", key, toolchainValue(t, "go", "go1.27.2", "1"), false) + } + next, _, err := x.computeOver(ctx, WaitsOf(entries).Ready) + if err != nil || next == "" || next == b.said[0].WaitsFor { + t.Fatalf("the build under the next toolchain: %q %v (the revision names %s)", next, err, b.said[0].WaitsFor) + } + b.ix.ownMu.Lock() + b.ix.ownFP = next + b.ix.ownMu.Unlock() + reportToolchain(t, b.ix, "mesh-build-go", "novox", "go", "go1.27.2") + if len(b.keys()) != 1 { + t.Fatalf("one node's report: pending %v, want the revision still held (the nodes disagree)", b.keys()) + } + reportToolchain(t, b.ix, "mesh-build-go", "ace", "go", "go1.27.2") + if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('c') { + t.Errorf("after the toolchain report: pending %v, avahi at %s: want the revision taken in", b.keys(), b.commit(t, "avahi")) + } + }) t.Run("T12 T13 indexer_pending names the wait; a hand take-in and a drop say what they ended", func(t *testing.T) { b := newTakeInBed(t) b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2")) diff --git a/modules/mesh-delivery-indexer/internal/index/toolchains.go b/modules/mesh-delivery-indexer/internal/index/toolchains.go index 8218c5f7..18452c86 100644 --- a/modules/mesh-delivery-indexer/internal/index/toolchains.go +++ b/modules/mesh-delivery-indexer/internal/index/toolchains.go @@ -78,12 +78,22 @@ func (ix *Indexer) PutToolchain(module, key string, value []byte, deleted bool) } // ApplyToolchain takes one change of a build module's toolchains in (PutToolchain) and, when the language's toolchain -// changed, writes the modules again that it moves (Recompute). The same version and recipe seen again writes nothing. +// changed, writes the modules again that it moves and evaluates pending again (ToolchainReported). The same version and +// recipe seen again writes nothing. func (ix *Indexer) ApplyToolchain(ctx context.Context, module, key string, value []byte, deleted bool) error { if !ix.PutToolchain(module, key, value, deleted) { return nil } - return ix.Recompute(ctx) + return ix.ToolchainReported(ctx) +} + +// ToolchainReported is what a changed toolchain moves: the modules written again (Recompute), then pending evaluated +// again (PendingAgain), since the indexer build a pending revision names is a fingerprint the toolchain is part of. +func (ix *Indexer) ToolchainReported(ctx context.Context) error { + if err := ix.Recompute(ctx); err != nil { + return err + } + return ix.PendingAgain(ctx) } // agreed is the toolchain of each language every reporting node agrees on, or why there is none. diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go index 9e280190..6be59ed0 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -7,6 +7,7 @@ package main import ( "context" + "encoding/json" "errors" "fmt" "io" @@ -378,8 +379,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro dead.check() provisioned(ctx, js, w, logger.Printf) // The indexer's stream take-in (ADR 0633 §5): a revision or a group it holds back is a wait, ended by take-in.taken. The - // durable consumer _ is made by the controller from this module's consumes; the whole stream is heard at - // the start, so a wait that began while the watchman was away is read with its own since. + // durable consumer _ is made by the controller from this module's consumes, and it is durable: what it + // acknowledged before a restart is not heard again. So the waits the indexer holds now are asked of it first, through + // its tool indexer_pending (this module's invokes), each with its own since; the consumer then hears what follows. + if calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls")); err != nil { + logger.Printf("take-in: no connection to ask %s (%v), so a wait the indexer held before this start is said only once a take-in event of it is heard", pendingTool, err) + } else { + seedTakeIn(w, calls.Call, logger.Printf) + calls.Close() + } switch tc, err := cred.EventConsumer(ctx, js, catalogue.StreamBusName(takein.TakeInStream)); { case err != nil: logger.Printf("the consumer %s_%s on %s does not exist yet, so no take-in wait is said until the watchman starts again: %v", @@ -465,6 +473,24 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } } +// pendingTool is the indexer's tool that answers the waits it holds (ADR 0633 §5), as the view asks it. +const pendingTool = delivery.IndexerModule + ".indexer_pending" + +// seedTakeIn asks the indexer the waits it holds now and starts each (Watchman.SeedTakeIn). An indexer that does not +// answer is said: the waits are then known only from the events heard after. +func seedTakeIn(w *watch.Watchman, call func(key string, body any) (json.RawMessage, error), logf func(string, ...any)) { + raw, err := call(pendingTool, map[string]any{}) + if err == nil { + var a takein.PendingAnswer + if a, err = takein.DecodePendingAnswer(raw); err == nil { + w.SeedTakeIn(a) + logf("take-in: %d waits the indexer holds are watched from their own since (%s)", len(a.Waits), pendingTool) + return + } + } + logf("take-in: %s cannot be asked (%v), so a wait the indexer held before this start is said only once a take-in event of it is heard", pendingTool, err) +} + // takeInHeard hands one event of a take-in stream to the watchman, by its name relative to the stream, and acknowledges it; // one that is not a take-in event is said and acknowledged, since hearing it again would not read it. func takeInHeard(w *watch.Watchman, m jetstream.Msg, logf func(string, ...any)) { diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index 8f81f1a1..f6802626 100644 --- a/modules/mesh-delivery-watchman/internal/watch/findings_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/findings_test.go @@ -94,6 +94,18 @@ func TestTheWatchmansFindings(t *testing.T) { t.Fatalf("a wait that ended in the replay wrote %d findings and left pulses %v", o.findingWrites, o.pulses) } }, map[string]string{}}, + {"ADR 0633 §5 (1585/T21): a wait the indexer holds at a restart within its bound is seeded from indexer_pending and raised past it", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + // The restarted watchman's durable consumer hears nothing of the pending event it acknowledged before; the wait + // began ten minutes ago, and the indexer still holds it. + since := c.t.Add(-10 * time.Minute) + w.SeedTakeIn(takein.PendingAnswer{Waits: []takein.PendingWait{{Key: revKey, What: takein.TakeInRevision, Keys: []string{revKey}, + Merges: []takein.Merge{{Repository: "novox/a", Number: 1, Merge: "c0ffee"}}, WaitsFor: "src2:x", Since: since}}}) + w.TakeInCaughtUp() + if len(o.findings) != 0 || o.ttls[signalKey("take-in", revKey)] != TakeInRevisionBound-10*time.Minute { + t.Fatalf("within its bound: findings %v, ttl %v; want none yet and the rest of the bound", o.findings, o.ttls[signalKey("take-in", revKey)]) + } + expire(w, c, signalKey("take-in", revKey), TakeInRevisionBound-10*time.Minute) + }, map[string]string{FindingKey(ScopeDelivery, revKey, KindTakeInWaits): Warning}}, {"ADR 0633 §5 (1585/T21): a drop clears the wait, and the merges left under it said pending again with the same since are raised from it", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { w.TakeInCaughtUp() // a live message, heard after the replay w.TakeIn(takeInEvent(t, takein.TakeInPending, revKey), takeInBody(t, takein.TakeInPending, revKey, t0)) diff --git a/modules/mesh-delivery-watchman/internal/watch/takein.go b/modules/mesh-delivery-watchman/internal/watch/takein.go index 4d3e4276..9bb48246 100644 --- a/modules/mesh-delivery-watchman/internal/watch/takein.go +++ b/modules/mesh-delivery-watchman/internal/watch/takein.go @@ -40,6 +40,25 @@ func (w *Watchman) TakeIn(event string, body []byte) { } w.mu.Lock() defer w.mu.Unlock() + w.takeIn(kind, key, t) +} + +// SeedTakeIn starts, at the watchman's start and before its consumer on the take-in stream, the wait of every revision +// and group the indexer holds now, as indexer_pending answers them (ADR 0633 §5): the consumer is durable, so the +// pending events it acknowledged before a restart are not heard again, and a wait within its bound would otherwise be +// lost. Each wait's bound runs from its own since, as a pending event's does; the stream's events that follow replace +// or end it. +func (w *Watchman) SeedTakeIn(a takein.PendingAnswer) { + w.mu.Lock() + defer w.mu.Unlock() + for _, wait := range a.Waits { + w.takeIn(takein.TakeInPending, wait.Key, takein.TakeIn{Kind: takein.TakeInPending, Key: wait.Key, What: wait.What, + Merges: wait.Merges, Group: wait.Group, Members: wait.Members, WaitsFor: wait.WaitsFor, Since: wait.Since}) + } +} + +// takeIn is one take-in event of key, w.mu held. +func (w *Watchman) takeIn(kind, key string, t takein.TakeIn) { finding := FindingKey(ScopeDelivery, key, KindTakeInWaits) if kind == takein.TakeInTaken { delete(w.takeInHeld, key) diff --git a/modules/mesh-delivery-watchman/manifest_test.go b/modules/mesh-delivery-watchman/manifest_test.go index 6a54308e..342caead 100644 --- a/modules/mesh-delivery-watchman/manifest_test.go +++ b/modules/mesh-delivery-watchman/manifest_test.go @@ -52,6 +52,7 @@ type manifest struct { Slug string `json:"slug"` Reads []string `json:"reads"` Consumes []string `json:"consumes"` + Invokes []string `json:"invokes"` State []state `json:"state"` OwnSecrets map[string]string `json:"own-secrets"` SecretsOwner string `json:"secrets-owner"` @@ -99,6 +100,10 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6 if !reflect.DeepEqual(m.Consumes, []string{"mesh-delivery-indexer.take-in.**"}) { t.Errorf("consumes = %v", m.Consumes) } + // The waits the indexer holds at the watchman's start are asked of its tool (ADR 0633 §5), a verb edge as the view's. + if !reflect.DeepEqual(m.Invokes, []string{"mesh-delivery-indexer.indexer_pending"}) { + t.Errorf("invokes = %v", m.Invokes) + } // 1952/WM1: the watchman holds the seat mesh-watchdogs at mesh scope, its first seat (novox/hq ADR 0564 §1, §4). if !reflect.DeepEqual(m.Claims, []claim{{Name: "mesh-watchdogs", Scope: "mesh", Serves: []string{"signals", "stuck", "presence", "dead-letters", "resync"}}}) { t.Errorf("claims = %+v", m.Claims) diff --git a/modules/mesh-delivery-watchman/module.json b/modules/mesh-delivery-watchman/module.json index 2de43956..7cdc94d5 100644 --- a/modules/mesh-delivery-watchman/module.json +++ b/modules/mesh-delivery-watchman/module.json @@ -29,6 +29,9 @@ "consumes": [ "mesh-delivery-indexer.take-in.**" ], + "invokes": [ + "mesh-delivery-indexer.indexer_pending" + ], "tools": [ "watchman_findings" ],