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)
mesh/delivery-group group feat/take-in rejected: composed together: the checker's verdict at the members' heads (38623984d6ad): mesh-build-go
mesh/delivery closed: a newer head of the same pull request
mesh/check failed (2): mesh-build-go: the indexer's judge at the group's mesh-delivery head 7b7096f37ea76e1…; and 1 more | 7b7096f37ea7 on ae31b66a9e33

- 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
This commit is contained in:
2026-11-05 00:28:56 +01:00
parent 4e4601a1eb
commit 7b7096f37e
20 changed files with 402 additions and 152 deletions
@@ -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)
@@ -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
}
@@ -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)
@@ -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)
}
@@ -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])
}
}
}
@@ -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)
@@ -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])
}
}
}
@@ -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
@@ -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.<key>`.
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
}
@@ -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 {
@@ -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)
@@ -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
}
@@ -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)
}
}
@@ -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"))
@@ -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.
@@ -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 <node>_<module> 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 <node>_<module> 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)) {
@@ -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))
@@ -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)
@@ -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)
@@ -29,6 +29,9 @@
"consumes": [
"mesh-delivery-indexer.take-in.**"
],
"invokes": [
"mesh-delivery-indexer.indexer_pending"
],
"tools": [
"watchman_findings"
],