From 1448f85b152ee3ccd1b260b19b30e9492f8868a2 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 09:15:11 +0100 Subject: [PATCH 1/6] A-241: the validation's seats are the revision's definitions with the merges' laid over MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A revision that touches no seat definition was judged against no seats at all; it is now judged against the whole seat list, the trunk's definitions standing in for the revision's until A-234 keeps them. Builds: design 47 G §712, A-241, A-234 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../internal/index/validate.go | 51 +++++++++++++++-- .../internal/index/validate_test.go | 56 +++++++++++++++++++ 2 files changed, 101 insertions(+), 6 deletions(-) diff --git a/modules/mesh-delivery-indexer/internal/index/validate.go b/modules/mesh-delivery-indexer/internal/index/validate.go index a394aa1e..103c3e6a 100644 --- a/modules/mesh-delivery-indexer/internal/index/validate.go +++ b/modules/mesh-delivery-indexer/internal/index/validate.go @@ -122,23 +122,62 @@ func placeKey(repository, path string) string { return strings.ToLower(trunk.RepositoryName(repository)) + " " + path } -// revisionDefinitions are the seat definitions the merges' trees carry, raw (§712 `seats`). +// revisionDefinitions are the validation's seat definitions, raw (§712 `seats`, A-241): the branch's revision's seat +// definitions with the definitions the merges' trees carry laid over them, so a revision that touches no definition is still +// judged against the whole seat list. A revision's own definitions are not kept yet (A-234), so the revision's are the +// trunk's, read from the state seat-definitions as the verb seat-definitions answers them. func (ix *Indexer) revisionDefinitions(ctx context.Context, ready []PendingEntry) ([]json.RawMessage, error) { - defs, err := ix.revisionSeatDefinitions(ctx, ready) + defs, err := ix.trunkDefinitions(ctx) if err != nil { return nil, err } - var out []json.RawMessage - for _, seat := range slices.Sorted(maps.Keys(defs)) { - raw, err := json.Marshal(defs[seat]) + merged, err := ix.revisionSeatDefinitions(ctx, ready) + if err != nil { + return nil, err + } + for seat, d := range merged { + raw, err := json.Marshal(d) if err != nil { return nil, fmt.Errorf("the definition of %s: %w", seat, err) } - out = append(out, raw) + defs[seat] = raw + } + var out []json.RawMessage + for _, seat := range slices.Sorted(maps.Keys(defs)) { + out = append(out, defs[seat]) } return out, nil } +// trunkDefinitions are the trunk's seat definitions of the state seat-definitions, raw, by seat: every `.trunk` entry +// that holds a definition; none where the indexer keeps no such state. +func (ix *Indexer) trunkDefinitions(ctx context.Context) (map[string]json.RawMessage, error) { + defs := map[string]json.RawMessage{} + if ix.SeatDefinitions == nil { + return defs, nil + } + keys, err := ix.SeatDefinitions.Keys(ctx) + if err != nil { + return nil, fmt.Errorf("the %s state cannot be listed: %w", SeatDefinitionsState, err) + } + for _, key := range keys { + seat, ok := strings.CutSuffix(key, "."+SourceTrunk) + if !ok || strings.HasPrefix(key, outOfStepPrefix) { + continue + } + v, _, found, err := ix.SeatDefinitions.Get(ctx, key) + if err != nil { + return nil, fmt.Errorf("%s.%s cannot be read: %w", SeatDefinitionsState, key, err) + } + var e SeatDefinitionEntry + if !found || json.Unmarshal(v, &e) != nil || e.Digest == "" || len(e.Definition) == 0 { + continue // an entry that holds no definition is no seat of the list + } + defs[seat] = e.Definition + } + return defs, nil +} + // validationJob is the run of the manifest validator for the request (§712): the program cmd/mesh-judge under the Go module // the indexer entry's Go artifact names, at the entry's path, built at head, the revision's head of the entry's repository; no // heads and no arguments; the input the request's bytes, served by the seat mesh-catalog's verb input. diff --git a/modules/mesh-delivery-indexer/internal/index/validate_test.go b/modules/mesh-delivery-indexer/internal/index/validate_test.go index d99a3358..5926cc31 100644 --- a/modules/mesh-delivery-indexer/internal/index/validate_test.go +++ b/modules/mesh-delivery-indexer/internal/index/validate_test.go @@ -1,11 +1,14 @@ package index import ( + "context" "encoding/json" + "reflect" "strings" "testing" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/composer" @@ -72,3 +75,56 @@ func TestValidationRefusesWhatTheRevisionCauses(t *testing.T) { t.Errorf("no indexer entry: %v", err) } } + +// The validation's seats are the branch's revision's seat definitions with the merges' changed ones laid over them (A-241): a +// revision that touches no definition is judged against the whole seat list, and a merge's definition replaces its seat's. +func TestValidationSeatsAreTheRevisionsWithTheMergesOver(t *testing.T) { + ctx := context.Background() + defs := newMem() + for seat, raw := range map[string]string{"node-engine": `{"seat":"node-engine","verbs":[{"name":"old"}]}`, "git": `{"seat":"git"}`} { + v, _ := json.Marshal(SeatDefinitionEntry{Seat: seat, Source: SourceTrunk, Digest: "d-" + seat, Definition: json.RawMessage(raw)}) + defs.put(SeatDefinitionKey(seat, SourceTrunk, "", 0), v) + } + v, _ := json.Marshal(SeatDefinitionEntry{Seat: "pulled", Source: SourcePull, Pull: 3, Digest: "d", Definition: json.RawMessage(`{"seat":"pulled"}`)}) + defs.put(SeatDefinitionKey("pulled", SourcePull, "r", 3), v) + ix := &Indexer{SeatDefinitions: defs} + seatsOf := func(raws []json.RawMessage) map[string]string { + out := map[string]string{} + for _, raw := range raws { + var d struct { + Seat string `json:"seat"` + Verbs []struct { + Name string `json:"name"` + } `json:"verbs"` + } + if err := json.Unmarshal(raw, &d); err != nil { + t.Fatal(err) + } + out[d.Seat] = "" + if len(d.Verbs) > 0 { + out[d.Seat] = d.Verbs[0].Name + } + } + return out + } + got, err := ix.revisionDefinitions(ctx, nil) + if err != nil || !reflect.DeepEqual(seatsOf(got), map[string]string{"node-engine": "old", "git": ""}) { + t.Errorf("a revision touching no definition: %v (%v), want the trunk's whole list and no pull request's", seatsOf(got), err) + } + const contracts = "modules/seat-contracts" + f := forgemsg.Forge{Repository: "r", Merge: strings.Repeat("c", 40), Manifests: map[string]string{ + contracts: `{"module":"seat-contracts","version":"1.0","build":{"artifacts":[{"name":"go-seats","kind":"package","language":"go","paths":["go"],"series":"0.1","generated-from":"."}]}}`}} + ix.Files = func(_ context.Context, _, _ string, dirs []string) (map[string]map[string]string, error) { + return map[string]map[string]string{dirs[0]: {"node-engine/seat.json": "1"}}, nil + } + ix.ReadFile = func(_ context.Context, _, _, p string) ([]byte, error) { + if p == contracts+"/seats/node-engine/seat.json" { + return []byte(`{"format":1,"seat":"node-engine","scope":"node","grain":"seat-list","decision":"d","domain":"Core","verbs":[{"name":"new"}]}`), nil + } + return nil, nil + } + got, err = ix.revisionDefinitions(ctx, []PendingEntry{{Forge: f}}) + if err != nil || !reflect.DeepEqual(seatsOf(got), map[string]string{"node-engine": "new", "git": ""}) { + t.Errorf("a merge changing a definition: %v (%v), want its definition over the trunk's and the rest kept", seatsOf(got), err) + } +} From ffc4b888591a113cd03284701334488e831f2e77 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 09:21:11 +0100 Subject: [PATCH 2/6] A-248: buildrun.Asker puts under the asker its subject names MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The indexer's validation ask rides accept.. under its own name (§712, A-232); an asker that names none still puts under the builder's. Builds: design 47 G A-248, A-247, §712 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- modules/mesh-delivery-builder/buildrun/asks.go | 10 +++++++--- modules/mesh-delivery-builder/buildrun/asks_test.go | 5 +++++ 2 files changed, 12 insertions(+), 3 deletions(-) diff --git a/modules/mesh-delivery-builder/buildrun/asks.go b/modules/mesh-delivery-builder/buildrun/asks.go index 26cc3e2c..d65d845e 100644 --- a/modules/mesh-delivery-builder/buildrun/asks.go +++ b/modules/mesh-delivery-builder/buildrun/asks.go @@ -1,6 +1,7 @@ package buildrun import ( + "cmp" "context" "encoding/json" "errors" @@ -34,6 +35,9 @@ type AskWaiter interface { // Asker puts asks and waits for their steps. type Asker struct { + // Name is the asker the subject names, the caller's seat name (design 47 G A-248, A-163): empty puts under the builder's + // name until the asker rename after the join. + Name string Publisher AskPublisher Waiter AskWaiter // Seats finds the build module of an ask's language (design 47 G A-82). @@ -173,8 +177,8 @@ func (k Asker) dropped(a buildstep.StepAsk, why string) { // PutTry is how long one publish of an ask waits for its stream's answer before it is made again (Put). var PutTry = 10 * time.Second -// Put puts a on its stream: the seat's accept subject of its language and kind, under the builder's name as asker (design 47 G -// §481, §600: `accept..builder`). The put is a kept call (novox/hq ADR 0668 +// Put puts a on its stream: the seat's accept subject of its language and kind, under the asker's Name, else the builder's +// name (design 47 G §481, §600, A-248: `accept..builder`). The put is a kept call (novox/hq ADR 0668 // §1, §2; issue 1966, issue 1748 note 27): it carries the ask's id as its message id, so the stream takes it once, and it // is made again through every transport fault (a deadline, no responders) within busretry.Bound, each try bounded by // PutTry, until ctx ends. A put that still meets one is an error opening `transport fault:` (TransportFault), so the key it @@ -184,7 +188,7 @@ func (k Asker) Put(ctx context.Context, a buildstep.StepAsk) error { if err != nil { return err } - subject, err := buildstep.AskSubjectBy(a.Language, a.Kind, buildstep.AskerBuilder) + subject, err := buildstep.AskSubjectBy(a.Language, a.Kind, cmp.Or(k.Name, buildstep.AskerBuilder)) if err != nil { return err } diff --git a/modules/mesh-delivery-builder/buildrun/asks_test.go b/modules/mesh-delivery-builder/buildrun/asks_test.go index fac93dbd..5f847802 100644 --- a/modules/mesh-delivery-builder/buildrun/asks_test.go +++ b/modules/mesh-delivery-builder/buildrun/asks_test.go @@ -44,6 +44,11 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) { if len(subjects.seen) != 1 || subjects.seen[0] != "mesh.seat.node-build-go.accept."+goAsk("x").Kind+".builder" { t.Fatalf("the ask was put on %v", subjects.seen) } + // A-248: an asker that names itself puts under its own name. + byName := Asker{Name: "indexer", Seats: testSeats(), Publisher: subjects, Waiter: f} + if err := byName.Put(context.Background(), goAsk("mvl0frv18d1959")); err != nil || subjects.seen[len(subjects.seen)-1] != "mesh.seat.node-build-go.accept."+goAsk("x").Kind+".indexer" { + t.Fatalf("a named asker put on %v (%v)", subjects.seen, err) + } f.Answer = func(a buildstep.StepAsk) buildstep.Step { return buildstep.Step{ID: a.ID, Node: "shanks", State: buildstep.StepRefused, Reason: "go 1.27 installed"} } From b29d7b89c295ec10c46d0a4231b4e9301f36f3db Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 09:27:52 +0100 Subject: [PATCH 3/6] =?UTF-8?q?=C2=A7713:=20a=20pending=20revision=20names?= =?UTF-8?q?=20its=20validation=20ask=20and=20resumes=20on=20its=20end?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The take-in puts the validation's run ask and ends there; the revision's newest pending value keeps validation {ask, asked, input, seq, try, failed, accepted}. The build seat's step-ended of that ask resumes it: accepted takes it in as before and is never asked again, refused keeps it pending with refused, and an end that could not validate keeps it pending with failed, said in take-in.pending, never a refusal and never passed. An end of a replaced ask is dropped, written in the journal. The answer is read from the event's own output or output object. The wiring in the indexer's program comes with the whole-module case (IX5). Builds: design 47 G §713, A-232, A-247, §712 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../internal/index/candidates.go | 6 + .../internal/index/pending.go | 9 + .../internal/index/pending_test.go | 2 +- .../internal/index/validate.go | 4 +- .../internal/index/validation_wait.go | 266 ++++++++++++++++++ .../internal/index/validation_wait_test.go | 87 ++++++ .../mesh-delivery-indexer/takein/takein.go | 18 ++ 7 files changed, 389 insertions(+), 3 deletions(-) create mode 100644 modules/mesh-delivery-indexer/internal/index/validation_wait.go create mode 100644 modules/mesh-delivery-indexer/internal/index/validation_wait_test.go diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index d55ddff0..378ae6c7 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -222,6 +222,12 @@ type Indexer struct { Pending Store // SayTakeIn publishes a take-in event (take-in.pending, take-in.taken; ADR 0633 §5); nil says none. SayTakeIn func(ctx context.Context, t takein.TakeIn) error + // ValidationAsks puts the validation's run ask (design 47 G §713, A-247: buildrun.Asker, named as the indexer's asker, + // A-248); nil validates nothing, and a revision is taken in as before. + ValidationAsks ValidationAsks + // ReadStepOutput reads a step-ended's output object, where the build seat keeps an answer too large for the event + // (A-247); nil reads none, and such an answer could not validate. + ReadStepOutput func(ctx context.Context, object string) ([]byte, error) // Runs is the state runs: the runs on request the indexer holds a candidate for (runcandidates.go, A-174); nil keeps none. Runs Store runMu sync.Mutex // one run event at a time (runcandidates.go) diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index ae1aee31..15a81ecf 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -98,6 +98,8 @@ type Pending struct { // Held is why a ready merge is not taken in although its revision names this build: objects full (HeldObjectsFull), // until room is said (RoomAgain). Held string `json:"held,omitempty"` + // Validation is the revision's validation (design 47 G §713, A-247), on the revision's newest value alone. + Validation *takein.Validation `json:"validation,omitempty"` } // PendingEntry is one key of the pending state as read: its key, revision, value and the merged message decoded. For a @@ -773,6 +775,9 @@ func (ix *Indexer) evaluate(ctx context.Context, own string, start bool) (Applie if len(lost) > 0 { return Applied{}, ix.waitingLine(w), ix.holdLost(ctx, w.Ready, lost, reasons, unread) } + if wait, words, err := ix.validationGate(ctx, ready, own); wait || err != nil { + return Applied{}, words, err + } names, dry, err := ix.computeOver(ctx, ready) if err != nil { return Applied{}, "", err @@ -904,6 +909,7 @@ func (ix *Indexer) dry() *dryRun { x := &Indexer{Manifests: ix.Manifests, Packages: ix.Packages, JudgeID: ix.JudgeID, Files: ix.Files, GoSum: ix.GoSum, GoMod: ix.GoMod, Now: ix.Now, Builds: ix.Builds, Named: ix.Named, Trunk: ix.Trunk, Read: ix.Read, Contains: ix.Contains, ReadOpen: nil, OwnBuild: nil, Message: ix.Message, SayRefused: nil, SayAccepted: nil, SayTakeIn: nil, Pending: nil, + ValidationAsks: nil, ReadStepOutput: nil, // a dry run asks no validation (§713) // The registry milestone's fields: what a take-in would write on the bus stays out of a dry run, so the states of the // requirements, findings, seat definitions, holds and their bounds are not overlaid, nothing is said, and no hold is // weighed (Holds and Counted nil hold nothing); ReadFile only reads. @@ -1332,6 +1338,9 @@ func waitOf(entries []PendingEntry, what, behind string) takein.PendingWait { if e.Value.Refused != "" { pw.Refused = e.Value.Refused } + if e.Value.Validation != nil { + pw.Validation = e.Value.Validation + } if e.Value.Said != "" && e.Value.SaidAt.After(pw.SaidAt) { pw.Said, pw.SaidAt = "take-in."+e.Value.SaidKind, e.Value.SaidAt } diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index a3e072a5..04c22a84 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -1131,7 +1131,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) for _, f := range []string{"Store", "Modules", "Manifests", "Predefined", "ConfigCandidates", "Taken", "Verdicts", "OldSections", "Unjudged", "JudgeID", "SayRefused", "SayAccepted", "Packages", "Graph", "Files", "GoSum", "GoMod", "Now", "Builds", "Named", "Trunk", "Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Message", "Pending", "SayTakeIn", - "Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "Runs", "OwnBuildState", "Heard", "Placement", "RunningWhole", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "NamedSets", "Registry", "SeedModules", "Versions", "SayVersionReleased", "SayVersionsFilled", "Inputs", "InputObjects", "BuildEdges", "Pins", "PinsPending"} { + "Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "Runs", "OwnBuildState", "Heard", "Placement", "RunningWhole", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "NamedSets", "Registry", "SeedModules", "Versions", "SayVersionReleased", "SayVersionsFilled", "Inputs", "InputObjects", "BuildEdges", "Pins", "PinsPending", "ValidationAsks", "ReadStepOutput"} { named[f] = true } ty := reflect.TypeOf(Indexer{}) diff --git a/modules/mesh-delivery-indexer/internal/index/validate.go b/modules/mesh-delivery-indexer/internal/index/validate.go index 103c3e6a..0fbf3bdc 100644 --- a/modules/mesh-delivery-indexer/internal/index/validate.go +++ b/modules/mesh-delivery-indexer/internal/index/validate.go @@ -25,9 +25,9 @@ const ( // validationProgram is the manifest validator's program, by its directory under the indexer entry's Go module: the indexer's // own code, so no repository or module name is fixed here (ADR 0666 §5). validationProgram = "cmd/mesh-judge" - // validationAsker is the asker the ask is put under. A-232 names the indexer's seat name mesh-catalog; the ask code still + // ValidationAsker is the asker the ask is put under (buildrun.Asker.Name, A-248). A-232 names the indexer's seat name mesh-catalog; the ask code still // names askers by their nicknames until the asker rename after the join (A-163), so this is `indexer` until then. - validationAsker = "indexer" + ValidationAsker = "indexer" // validationCatalogSeat is the seat the indexer entry claims. validationCatalogSeat = "mesh-catalog" // The cases of the request, as the manifest validator names them. diff --git a/modules/mesh-delivery-indexer/internal/index/validation_wait.go b/modules/mesh-delivery-indexer/internal/index/validation_wait.go new file mode 100644 index 00000000..f5fc3f5b --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/validation_wait.go @@ -0,0 +1,266 @@ +package index + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/composer" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" +) + +// The pending revision's wait for its validation (design 47 G §713, A-232, A-247): the take-in puts the run ask of §712, writes +// `validation` on the revision's newest pending value and ends there; the build seat's step-ended of that ask, heard through the +// indexer's own durable consumer, resumes it: accepted takes the revision in as before, refused keeps it pending with `refused`, +// and an end that could not validate keeps it pending with `failed`, asked again by a later merge of the branch or by hand. + +// ValidationAsks puts a run ask on the build seat (buildrun.Asker). +type ValidationAsks interface { + Put(ctx context.Context, a buildstep.StepAsk) error +} + +// newestOfRevision is the index in ready of the revision's newest merge, the value that carries `validation` (A-247). +func newestOfRevision(ready []PendingEntry) int { + n := 0 + for i, e := range ready { + if e.Value.Seq > ready[n].Value.Seq { + n = i + } + } + return n +} + +func branchOfRevision(ready []PendingEntry) string { + if ref := ready[newestOfRevision(ready)].Value.Ref; ref != nil && ref.Branch != "" { + return ref.Branch + } + return TrunkBranch +} + +// validationGate says whether the ready revision waits for its validation (§713): it goes on to the take-in once its newest +// value's validation was accepted at that value's sequence; it waits while its ask is out, while it is refused, and while it +// could not validate; otherwise it puts a new ask, the next try, and waits. Without an asker nothing is validated. +func (ix *Indexer) validationGate(ctx context.Context, ready []PendingEntry, own string) (bool, string, error) { + if ix.ValidationAsks == nil { + return false, "", nil + } + newest := ready[newestOfRevision(ready)] + seq := newest.Value.Seq + if v := newest.Value.Validation; v != nil && v.Seq == seq { + switch { + case v.Accepted == seq: + return false, "", nil + case v.Failed != "": + return true, fmt.Sprintf("the validation of the revision at %d could not validate: %s", seq, v.Failed), nil + case newest.Value.Refused != "": + return true, fmt.Sprintf("the revision at %d is refused: %s", seq, newest.Value.Refused), nil + case v.Ask != "": + return true, fmt.Sprintf("the revision at %d waits for its validation, the ask %s", seq, v.Ask), nil + } + } + return true, "", ix.askValidation(ctx, ready, own) +} + +// askValidation puts the revision's validation ask (§712) and writes it on its newest value, the try after every earlier try +// of the revision; an older value's validation goes (A-247: on the newest alone). A request or job that cannot be made could not +// validate: it is written as `failed`, said, and no ask is put. +func (ix *Indexer) askValidation(ctx context.Context, ready []PendingEntry, own string) error { + n := newestOfRevision(ready) + newest := ready[n] + try := 0 + for _, e := range ready { + if v := e.Value.Validation; v != nil && v.Try > try { + try = v.Try + } + } + v := &takein.Validation{Seq: newest.Value.Seq, Try: try + 1, Asked: ix.now()} + branch := branchOfRevision(ready) + ask := buildstep.StepAsk{ID: buildstep.NewID(), Kind: buildstep.AskCheck, Language: buildstep.AskGo, Asker: ValidationAsker, At: ix.now().UTC()} + req, entry, err := ix.validationRequest(ctx, branch, ready) + var raw []byte + if err == nil { + raw, err = json.Marshal(req) + } + var job buildstep.RunJob + if err == nil { + job, err = validationJob(entry, revisionHead(ready, entry.Repository, entry.Commit), branch, v.Seq, raw) + } + if err == nil { + v.Input, err = ix.KeepInput(ctx, ask.ID, raw) + } + if err != nil { + v.Failed = "the validation could not be asked: " + err.Error() + } else { + v.Ask, ask.Run = ask.ID, &job + } + if err := ix.writeValidation(ctx, ready, n, v, own); err != nil { + return err + } + if v.Failed != "" { + ix.sayValidation(ctx, ready, v) + return nil + } + if err := ix.ValidationAsks.Put(ctx, ask); err != nil { + return fmt.Errorf("the validation ask %s of the revision at %d could not be put: %w", ask.ID, v.Seq, err) + } + ix.logf("take-in: the revision at %d is validated by the ask %s (try %d); it waits for its end", v.Seq, ask.ID, v.Try) + return nil +} + +// revisionHead is the revision's head of repository: the newest merge of it the revision covers, else the commit its entry stands at. +func revisionHead(ready []PendingEntry, repository, commit string) string { + var seq uint64 + head := commit + for _, e := range ready { + if e.Forge.Merge != "" && strings.EqualFold(e.Forge.Repository, repository) && e.Value.Seq >= seq { + seq, head = e.Value.Seq, e.Forge.Merge + } + } + return head +} + +// writeValidation writes v on ready[n] and takes it from every other value, each by compare-and-set; a new validation clears the +// value's refusal. A value lost under it is an error: the evaluation is made again. +func (ix *Indexer) writeValidation(ctx context.Context, ready []PendingEntry, n int, v *takein.Validation, own string) error { + for i, e := range ready { + if i != n && e.Value.Validation == nil { + continue + } + p := e.Value + p.Validation, p.LastWriter = nil, own + if i == n { + p.Validation, p.Refused = v, "" + } + data, err := encodePending(p) + if err != nil { + return err + } + if err := ix.Pending.Update(ctx, e.Key, data, e.Revision); err != nil { + return fmt.Errorf("%s.%s: %w", PendingState, e.Key, err) + } + } + return nil +} + +// sayValidation says the revision's wait in take-in.pending with its refusal or why it could not validate. +func (ix *Indexer) sayValidation(ctx context.Context, ready []PendingEntry, v *takein.Validation) { + t := revisionWait(takein.TakeInPending, ready, "") + t.Failed = v.Failed + t.Refused = ready[newestOfRevision(ready)].Value.Refused + ix.sayTakeIn(ctx, t) +} + +// errNoValidationAnswer is an end whose answer is no verdict of both cases. +var errNoValidationAnswer = errors.New("no answer") + +// HearValidationEnded takes in the end of a validation ask (§713), heard on the build seat's step-ended through the indexer's own +// durable consumer: only an end of the indexer's asks outside a pipeline; an end whose ask no pending value names now is dropped, +// written in the journal. The answer is read from the event's own output or output object (A-247). +func (ix *Indexer) HearValidationEnded(ctx context.Context, e buildstep.StepEnded) error { + if e.Asker != ValidationAsker || e.Pipeline != "" || e.Ask == "" || ix.Pending == nil || !ix.pendingOnBus(ctx) { + return nil + } + own := ix.ownBuildNow(ctx) + ix.pendingMu.Lock() + defer ix.pendingMu.Unlock() + entries, err := ReadPending(ctx, ix.Pending, ix.logf) + if err != nil { + return err + } + ready := WaitsOf(entries).Ready + n := -1 + for i, p := range ready { + if p.Value.Validation != nil && p.Value.Validation.Ask == e.Ask { + n = i + } + } + if n < 0 { + ix.logf("take-in: the end of the validation ask %s (%s on %s) is dropped: no pending revision names that ask now", e.Ask, e.Outcome, e.Node) + return nil + } + ready, lost, _, _ := ix.loadNewest(ctx, ready) + if len(lost) > 0 { + return fmt.Errorf("the revision the validation ask %s ended for holds a merge that cannot be read", e.Ask) + } + n = newestOfRevision(ready) + v := *ready[n].Value.Validation + refused, failed := ix.validationAnswer(ctx, ready, e, v) + p := ready[n].Value + switch { + case failed != "": + v.Failed = failed + case refused != "": + p.Refused = refused + default: + v.Accepted = v.Seq + } + p.Validation, p.LastWriter = &v, own + data, err := encodePending(p) + if err != nil { + return err + } + if err := ix.Pending.Update(ctx, ready[n].Key, data, ready[n].Revision); err != nil { + return fmt.Errorf("%s.%s: %w", PendingState, ready[n].Key, err) + } + if v.Accepted == 0 { + ready[n].Value = p + ix.sayValidation(ctx, ready, &v) + ix.logf("take-in: the validation ask %s of the revision at %d ended: refused %q, could not validate %q", e.Ask, v.Seq, refused, failed) + return nil + } + ix.logf("take-in: the validation ask %s accepted the revision at %d", e.Ask, v.Seq) + if own == "" { + return nil // the take-in waits for the build this program runs, as every take-in does + } + _, _, err = ix.evaluate(ctx, own, false) + return err +} + +// validationAnswer reads the end's answer: the refusal of the revision, or why it could not validate (the step did not pass, the +// program exited other than 0, its answer is no verdict of both cases, or its request is no longer the one asked). +func (ix *Indexer) validationAnswer(ctx context.Context, ready []PendingEntry, e buildstep.StepEnded, v takein.Validation) (string, string) { + if e.Outcome != buildstep.OutcomePassed { + return "", strings.TrimSpace(e.Outcome + ": " + e.Said) + } + out := []byte(e.Output) + if len(out) == 0 && e.OutputObject != "" { + if ix.ReadStepOutput == nil { + return "", "the answer is in the output object " + e.OutputObject + ", which this indexer does not read" + } + var err error + if out, err = ix.ReadStepOutput(ctx, e.OutputObject); err != nil { + return "", "the output object " + e.OutputObject + " could not be read: " + err.Error() + } + } + var answer buildstep.RunAnswer + if err := json.Unmarshal(out, &answer); err != nil { + return "", fmt.Sprintf("the run's answer is no run answer: %v", err) + } + if answer.Exit != 0 { + return "", fmt.Sprintf("the manifest validator exited %d: %s", answer.Exit, answer.Said) + } + var verdict composer.Verdict + if err := json.Unmarshal(answer.Output, &verdict); err != nil { + return "", fmt.Sprintf("%v: the manifest validator's output is no verdict: %v", errNoValidationAnswer, err) + } + req, _, err := ix.validationRequest(ctx, branchOfRevision(ready), ready) + var raw []byte + if err == nil { + raw, err = json.Marshal(req) + } + if err != nil { + return "", "the request cannot be made again to read the verdict: " + err.Error() + } + if runDigest(raw) != v.Input { + return "", "the revision's request is no longer the one the ask " + v.Ask + " was asked with" + } + refused, err := ValidationRefusal(req, verdict) + if err != nil { + return "", err.Error() + } + return RefusedWords(refused), "" +} diff --git a/modules/mesh-delivery-indexer/internal/index/validation_wait_test.go b/modules/mesh-delivery-indexer/internal/index/validation_wait_test.go new file mode 100644 index 00000000..5d344a47 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/validation_wait_test.go @@ -0,0 +1,87 @@ +package index + +import ( + "context" + "encoding/json" + "strings" + "testing" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/composer" +) + +type askRecorder struct{ asks []buildstep.StepAsk } + +func (r *askRecorder) Put(_ context.Context, a buildstep.StepAsk) error { + r.asks = append(r.asks, a) + return nil +} + +// Design 47 G §713, A-247: a pending revision names its ask on its newest value and resumes on that ask's end; an end of a +// replaced ask is dropped; a run that could not validate is no refusal and never passed; the start asks nothing again while +// the ask is out. +func TestAPendingRevisionWaitsForItsValidation(t *testing.T) { + ctx := context.Background() + b := newTakeInBed(t) + asks := &askRecorder{} + b.ix.ValidationAsks, b.ix.Inputs, b.ix.Revisions = asks, newMem(), newMemObjects(0) + // No entry of the revision claims the seat mesh-catalog: the validation could not be asked, so it is failed, said, and + // nothing is taken in. + b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "")) + entries := readPending(t, b) + if len(entries) != 1 || entries[0].Value.Validation == nil || !strings.Contains(entries[0].Value.Validation.Failed, "claims the seat mesh-catalog") || len(asks.asks) != 0 { + t.Fatalf("pending %+v, asks %d: want m.2 pending with failed and no ask", entries, len(asks.asks)) + } + if last := b.said[len(b.said)-1]; last.Kind != takein.TakeInPending || last.Failed == "" || last.Refused != "" { + t.Errorf("said %+v, want take-in.pending naming failed, never a refusal", last) + } + if w := PendingWaits(entries); len(w) != 1 || w[0].Validation == nil || w[0].Validation.Try != 1 { + t.Errorf("indexer_pending %+v, want the validation whole", w) + } + // An ask out: the newest value names it; the start asks nothing new while it is out. + loaded, _, _, _ := b.ix.loadNewest(ctx, entries) + req, _, err := b.ix.validationRequest(ctx, TrunkBranch, loaded) + if err != nil { + t.Fatal(err) + } + raw, _ := json.Marshal(req) + setValidation := func(v takein.Validation) { + e := readPending(t, b)[0] + p := e.Value + p.Validation = &v + data, _ := encodePending(p) + if err := b.pending.Update(ctx, e.Key, data, e.Revision); err != nil { + t.Fatal(err) + } + } + setValidation(takein.Validation{Ask: "mvl0frv18d1957", Input: runDigest(raw), Seq: 2, Try: 2}) + if _, _, err := b.ix.TakeInAtStart(ctx); err != nil || len(asks.asks) != 0 || len(b.keys()) != 1 { + t.Fatalf("the start: %v, asks %d, pending %v: want it waiting for its ask", err, len(asks.asks), b.keys()) + } + end := func(ask, outcome, said string, answer any) buildstep.StepEnded { + out, _ := json.Marshal(answer) + return buildstep.StepEnded{Ask: ask, ID: ask, Asker: ValidationAsker, Node: "ace", Outcome: outcome, Said: said, Output: string(out)} + } + verdict, _ := json.Marshal(composer.Verdict{Cases: []composer.CaseVerdict{{Name: "change"}, {Name: "trunk"}}}) + accepted := buildstep.RunAnswer{Built: true, Exit: 0, Output: verdict} + // The end of a replaced ask is dropped, written in the journal. + if err := b.ix.HearValidationEnded(ctx, end("mvl0frv18d1950", buildstep.OutcomePassed, "", accepted)); err != nil || len(b.keys()) != 1 || !strings.Contains(b.logs.String(), "is dropped") { + t.Fatalf("a replaced ask's end: %v, pending %v", err, b.keys()) + } + // could not run: pending with failed, never passed. + if err := b.ix.HearValidationEnded(ctx, end("mvl0frv18d1957", buildstep.OutcomeCouldNotRun, "no build node", nil)); err != nil { + t.Fatal(err) + } + if v := readPending(t, b)[0].Value.Validation; v == nil || v.Failed != "could not run: no build node" || v.Accepted != 0 { + t.Fatalf("after could not run: %+v", v) + } + // Asked again (by hand or by a later merge) and accepted: the revision is taken in, and its keys go. + setValidation(takein.Validation{Ask: "mvl0frv18d1958", Input: runDigest(raw), Seq: 2, Try: 3}) + if err := b.ix.HearValidationEnded(ctx, end("mvl0frv18d1958", buildstep.OutcomePassed, "", accepted)); err != nil { + t.Fatal(err) + } + if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('c') { + t.Fatalf("accepted: pending %v %+v, avahi at %s: want it taken in\n%s", b.keys(), readPending(t, b)[0].Value, b.commit(t, "avahi"), b.logs.String()) + } +} diff --git a/modules/mesh-delivery-indexer/takein/takein.go b/modules/mesh-delivery-indexer/takein/takein.go index cc3570d5..203872b0 100644 --- a/modules/mesh-delivery-indexer/takein/takein.go +++ b/modules/mesh-delivery-indexer/takein/takein.go @@ -141,6 +141,9 @@ type TakeIn struct { // the seat definitions the revision carries (ADR 0633 §1), the manifest validator's problems. The revision waits for // an indexer whose manifest is accepted, and says why here, as it says what it lost. Refused string `json:"refused,omitempty"` + // Failed is why the revision's validation could not validate (design 47 G §713): the build seat's words, never a refusal + // and never passed; it is asked again by a later merge of the branch and by indexer_validate_again. + Failed string `json:"failed,omitempty"` // Unknown is why the indexer holds a merge it does not take in although no indexer build is named: it cannot learn the build // it runs itself (ADR 0633 §1), so Since is when it began not to know. A pending revision names WaitsFor or Unknown. Unknown string `json:"own-build-unknown,omitempty"` @@ -248,6 +251,21 @@ type PendingWait struct { Unread []string `json:"unread,omitempty"` // Refused is why the manifest of the indexer it names was refused (TakeIn.Refused). Refused string `json:"refused,omitempty"` + // Validation is the revision's validation whole, as its newest pending value keeps it (design 47 G §713, A-247). + Validation *Validation `json:"validation,omitempty"` +} + +// Validation is a pending revision's validation (design 47 G §712, §713, A-232, A-247), kept on the revision's newest pending +// value alone: the run ask's id, when it was asked, its input's digest, the sequence of the newest merge it covers and the try +// number for it, why it could not validate, and the sequence it was accepted at (never asked again once set). +type Validation struct { + Ask string `json:"ask,omitempty"` + Asked time.Time `json:"asked,omitzero"` + Input string `json:"input,omitempty"` + Seq uint64 `json:"seq"` + Try int `json:"try"` + Failed string `json:"failed,omitempty"` + Accepted uint64 `json:"accepted,omitempty"` } // PendingAnswer is indexer_pending's answer: every wait, oldest first, and the pending keys that do not read. From 63a35b88176baa060c203c1cc9d767ff943e92c6 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 09:35:28 +0100 Subject: [PATCH 4/6] B5g6-1: indexer_validate_again asks a revision that could not validate again MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The tool (class act) records its hand act through mesh-hand-acts first and asks the indexer through takeins; the indexer asks the revision's validation under its next try and answers the validation's state. A revision accepted, refused or whose ask is out is asked nothing. Builds: design 47 G §713 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-indexer-tools/main.go | 4 ++ .../cmd/mesh-delivery-indexer/takein.go | 4 ++ .../internal/index/pending_tools.go | 31 ++++++++ .../internal/index/takein.go | 17 +++++ .../internal/index/validation_wait.go | 72 +++++++++++++++++++ .../internal/index/validation_wait_test.go | 26 +++++++ .../mesh-delivery-indexer/manifest_test.go | 5 +- modules/mesh-delivery-indexer/module.json | 3 +- 8 files changed, 159 insertions(+), 3 deletions(-) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go index c767b78b..a584fa9b 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go @@ -316,6 +316,10 @@ func main() { })...) tools = append(tools, viewTools(os.Getenv, trunk.State, time.Now)...) tools = append(tools, index.PendingTools(keptRead{stdio.State(index.PendingState)}, keptRead{stdio.State(index.OwnBuildStateName)}, asks, time.Now)...) + tools = append(tools, index.ValidationTools(asks, time.Now, func(what, why string) error { + _, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: index.ValidateAgainTool, Verb: "validate-again"}) + return err + })...) tools = append(tools, index.FingerprintsTools(asks, time.Now, func(what, why string) error { _, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: index.FingerprintsAgainTool, Verb: "fingerprints-again"}) return err diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go index e72897a0..dde180b0 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go @@ -28,6 +28,7 @@ type takerIn interface { NamedDrop(ctx context.Context, fingerprint, member, by, why string) index.TakenIn Unpin(ctx context.Context, revision, record, by, why string) index.TakenIn DropRunByHand(ctx context.Context, run, by, why string) index.TakenIn + ValidateAgainByHand(ctx context.Context, key, by, why string) index.TakenIn RegistryAct(ctx context.Context, a index.TakeInAsk) index.TakenIn } @@ -69,6 +70,9 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream } else if a.Merge != "" { asked, r = "hand take-in of a merge", ix.TakeInMerge(ctx, a.MergeRepository, a.Merge, a.Sequence, a.By, a.Why) a.Module = r.Module + } else if a.Pending != "" && a.ValidateAgain { + asked, r = "validation asked again", ix.ValidateAgainByHand(ctx, a.Pending, a.By, a.Why) + a.Module = a.Pending } else if a.Pending != "" && a.Drop { asked, r = "drop of a pending merge", ix.DropPending(ctx, a.Pending, a.By, a.Why, a.Abandon) a.Module = a.Pending diff --git a/modules/mesh-delivery-indexer/internal/index/pending_tools.go b/modules/mesh-delivery-indexer/internal/index/pending_tools.go index 366654eb..c27abaa7 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_tools.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_tools.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "strings" "time" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" @@ -111,3 +112,33 @@ func PendingTools(pending, ownBuild Store, ask func(key string, value []byte) (T Run: merge}, } } + +// ValidationTools is indexer_validate_again (class act, design 47 G §713): the tools bundle writes an ask in takeins and the +// indexer asks the revision's validation again; record says the hand act in the hand-act log before it is asked. +func ValidationTools(ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time, record func(what, why string) error) []stdio.Tool { + return []stdio.Tool{{Name: ValidateAgainTool, + Description: "Ask again the validation of a pending revision that could not validate (design 47 G §712, §713): the manifest " + + "validator's run on the build seat node-build-go ended could not run, exited other than 0, or answered not both cases. " + + "The revision is asked under its next try; the answer says the validation's state (the ask, its try, accepted, refused, " + + "or why it could not validate). A revision accepted, refused or whose ask is out is asked nothing: a refusal is asked " + + "again only by a later merge of its branch. A hand act, recorded through mesh-hand-acts before it is asked. " + + "indexer_pending names the keys and shows the validation whole.", + Input: map[string]any{"type": "object", "required": []string{"key", "why"}, "properties": map[string]any{ + "key": map[string]any{"type": "string", "description": "the revision's wait key, m., as indexer_pending names it"}, + "why": map[string]any{"type": "string", "description": "why it is asked again, in plain words"}}}, + Run: func(args map[string]any) (any, error) { + str := func(k string) string { s, _ := args[k].(string); return s } + key, value, err := NewValidateAgainAsk(str("key"), str("why"), stdio.CallerOf(args), now()) + if err != nil { + return nil, err + } + if err := record("asked the validation of the pending revision "+strings.TrimSpace(str("key"))+" again by hand", strings.TrimSpace(str("why"))); err != nil { + return nil, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so nothing was asked: %w", err) + } + a, err := ask(key, value) + if err != nil { + return nil, err + } + return a.Result, nil + }}} +} diff --git a/modules/mesh-delivery-indexer/internal/index/takein.go b/modules/mesh-delivery-indexer/internal/index/takein.go index 7e581059..632a35fb 100644 --- a/modules/mesh-delivery-indexer/internal/index/takein.go +++ b/modules/mesh-delivery-indexer/internal/index/takein.go @@ -52,6 +52,9 @@ type TakeInAsk struct { // Abandon, with Drop, drops a pending merge no later merge covers (ADR 0633 §5 asks a later merge; issue 1966, ADR // 0668 §5): the hand act says why it is abandoned all the same. Abandon bool `json:"abandon,omitempty"` + // ValidateAgain, with Pending, asks the validation of the revision under that key again (indexer_validate_again, design 47 + // G §713): one that could not validate. + ValidateAgain bool `json:"validate-again,omitempty"` // Merge asks the merge of MergeRepository at this commit to be taken in by hand (indexer_take_in_merge): its merged message // is asked of the forge bridge and run through the take-in as a delivered one, under its stream sequence Sequence. Merge string `json:"merge,omitempty"` @@ -165,6 +168,20 @@ func NewMergeAsk(repository, merge string, sequence uint64, why, by string, at t return "merge." + merge, data, err } +// NewValidateAgainAsk is the value indexer_validate_again writes: its key in takeins, `pending.validate-again.`, and the +// value. key is a revision's wait key, m., as indexer_pending names it. +func NewValidateAgainAsk(key, why, by string, at time.Time) (string, []byte, error) { + key = strings.TrimSpace(key) + if !strings.HasPrefix(key, "m.") { + return "", nil, errors.New("name the revision whose validation is asked again by its key, m., as indexer_pending names it") + } + if strings.TrimSpace(why) == "" { + return "", nil, errors.New("say why (why): a hand act is recorded through mesh-hand-acts") + } + data, err := json.Marshal(TakeInAsk{V: 1, Pending: key, ValidateAgain: true, Why: strings.TrimSpace(why), By: by, At: at.UTC()}) + return "pending.validate-again." + key, data, err +} + // NewPendingAsk is the value indexer_take_in_pending (drop false) or indexer_drop_pending (drop true) writes: its key // in takeins, `pending..`, and the value. key is a wait's key as indexer_pending names it, for a // drop one pending merge's; abandon drops one no later merge covers. diff --git a/modules/mesh-delivery-indexer/internal/index/validation_wait.go b/modules/mesh-delivery-indexer/internal/index/validation_wait.go index f5fc3f5b..f24e27a1 100644 --- a/modules/mesh-delivery-indexer/internal/index/validation_wait.go +++ b/modules/mesh-delivery-indexer/internal/index/validation_wait.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "strings" + "time" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/composer" @@ -264,3 +265,74 @@ func (ix *Indexer) validationAnswer(ctx context.Context, ready []PendingEntry, e } return RefusedWords(refused), "" } + +// ValidateAgainTool asks the validation of a revision that could not validate again by hand (§713). +const ValidateAgainTool = "indexer_validate_again" + +// validationWords is the validation's state as the hand ask answers it. +func validationWords(v *takein.Validation, refused string) string { + switch { + case v == nil: + return "no validation is asked yet" + case v.Accepted != 0: + return fmt.Sprintf("accepted at %d (try %d, the ask %s)", v.Accepted, v.Try, v.Ask) + case v.Failed != "": + return fmt.Sprintf("could not validate at %d (try %d): %s", v.Seq, v.Try, v.Failed) + case refused != "": + return fmt.Sprintf("refused at %d (try %d): %s", v.Seq, v.Try, refused) + default: + return fmt.Sprintf("the ask %s is out since %s (try %d at %d)", v.Ask, v.Asked.UTC().Format(time.RFC3339), v.Try, v.Seq) + } +} + +// ValidateAgainByHand is indexer_validate_again answered (§713): the ready revision under key (its oldest merge's m. key, as +// indexer_pending names it) whose validation could not validate is asked again, the next try; its answer says the ask's state. +// A revision accepted, refused or whose ask is out is asked nothing: a refusal is asked again only by a later merge of its branch. +func (ix *Indexer) ValidateAgainByHand(ctx context.Context, key, by, why string) TakenIn { + r := TakenIn{Module: key} + if ix.Pending == nil || ix.ValidationAsks == nil { + r.Answer = "this indexer asks no validation: nothing asked" + return r + } + own := ix.ownBuildNow(ctx) + ix.pendingMu.Lock() + defer ix.pendingMu.Unlock() + entries, err := ReadPending(ctx, ix.Pending, ix.logf) + if err != nil { + r.Answer = err.Error() + return r + } + ready := WaitsOf(entries).Ready + if len(ready) == 0 || ready[0].Key != key { + r.Answer = fmt.Sprintf("no revision waits under %s: indexer_pending names the waits", key) + return r + } + ready, lost, reasons, _ := ix.loadNewest(ctx, ready) + if len(lost) > 0 { + r.Answer = fmt.Sprintf("nothing asked: the message of %s is not read (%s)", lost[0].Key, reasons[lost[0].Key]) + return r + } + newest := ready[newestOfRevision(ready)] + v := newest.Value.Validation + if v == nil || v.Seq != newest.Value.Seq || v.Failed == "" { + r.Answer = "nothing asked: the validation is " + validationWords(v, newest.Value.Refused) + "; only one that could not validate is asked again by hand" + return r + } + ix.logf("take-in: the validation of %s is asked again by hand by %s: %s", key, by, why) + if err := ix.askValidation(ctx, ready, own); err != nil { + r.Answer = "the validation was not asked again: " + err.Error() + return r + } + after, err := ReadPending(ctx, ix.Pending, ix.logf) + if err == nil { + ready = WaitsOf(after).Ready + } + if err != nil || len(ready) == 0 { + r.Answer = "asked again; its state is not read back" + return r + } + n := ready[newestOfRevision(ready)] + r.Stored = n.Value.Validation != nil && n.Value.Validation.Failed == "" + r.Answer = fmt.Sprintf("asked again by %s (%s): the validation is %s", by, why, validationWords(n.Value.Validation, n.Value.Refused)) + return r +} diff --git a/modules/mesh-delivery-indexer/internal/index/validation_wait_test.go b/modules/mesh-delivery-indexer/internal/index/validation_wait_test.go index 5d344a47..bc749e59 100644 --- a/modules/mesh-delivery-indexer/internal/index/validation_wait_test.go +++ b/modules/mesh-delivery-indexer/internal/index/validation_wait_test.go @@ -76,6 +76,32 @@ func TestAPendingRevisionWaitsForItsValidation(t *testing.T) { if v := readPending(t, b)[0].Value.Validation; v == nil || v.Failed != "could not run: no build node" || v.Accepted != 0 { t.Fatalf("after could not run: %+v", v) } + // indexer_validate_again (§713): its hand act recorded first, then the next try asked; the answer says the ask's state. + var recorded []string + tools := ValidationTools(func(key string, value []byte) (TakeInAsk, error) { + a, err := DecodeTakeIn(value) + if err != nil || key != "pending.validate-again."+takein.PendingKey(2) || !a.ValidateAgain { + t.Fatalf("asked %s %+v %v", key, a, err) + } + r := b.ix.ValidateAgainByHand(ctx, a.Pending, a.By, a.Why) + a.Result = &r + return a, nil + }, b.ix.now, func(what, why string) error { recorded = append(recorded, what); return nil }) + if len(tools) != 1 || tools[0].Name != ValidateAgainTool { + t.Fatalf("tools = %v", tools) + } + if _, err := tools[0].Run(map[string]any{"key": takein.PendingKey(2)}); err == nil || len(recorded) != 0 { + t.Fatalf("asked again with no why: %v, %v", err, recorded) + } + got, err := tools[0].Run(map[string]any{"key": takein.PendingKey(2), "why": "the build node is back"}) + if v := readPending(t, b)[0].Value.Validation; err != nil || len(recorded) != 1 || v == nil || v.Try != 3 || v.Failed == "could not run: no build node" || + !strings.Contains(got.(*TakenIn).Answer, "try 3") { + t.Fatalf("asked again: %v, recorded %v, validation %+v, answer %+v: want the try after 2 asked", err, recorded, v, got) + } + setValidation(takein.Validation{Ask: "mvl0frv18d1959", Input: runDigest(raw), Seq: 2, Try: 3}) + if r := b.ix.ValidateAgainByHand(ctx, takein.PendingKey(2), "lead", "again"); r.Stored || !strings.Contains(r.Answer, "is out") || readPending(t, b)[0].Value.Validation.Try != 3 { + t.Fatalf("an ask out was asked again: %+v", r) + } // Asked again (by hand or by a later merge) and accepted: the revision is taken in, and its keys go. setValidation(takein.Validation{Ask: "mvl0frv18d1958", Input: runDigest(raw), Seq: 2, Try: 3}) if err := b.ix.HearValidationEnded(ctx, end("mvl0frv18d1958", buildstep.OutcomePassed, "", accepted)); err != nil { diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index d7da242e..5dfaa46b 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -202,8 +202,9 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // milestone 47/G W4.12 (design 47, the indexer's tools): the catalog graph revisions read and steered. index.CatalogRevisionTool, index.CatalogRevisionsTool, index.CandidateGraphsTool, index.RecomputeTool, index.DropCandidateGraphTool, index.EntryIndexTool, index.NamedDropTool, // design 47 G §675 - index.ProposalsTool, index.ProposalDropTool, index.SeedAgainTool, index.InputsTool, index.PinsTool, index.UnpinTool, index.ConsumerTool, index.RunsTool, index.RunDropTool) // design 47 G §667, §8, §579, §64 (A-194) - if !reflect.DeepEqual(m.Tools, want) || len(want) != 49 || slices.Contains(m.Tools, "modules") { + index.ProposalsTool, index.ProposalDropTool, index.SeedAgainTool, index.InputsTool, index.PinsTool, index.UnpinTool, index.ConsumerTool, index.RunsTool, index.RunDropTool, // design 47 G §667, §8, §579, §64 (A-194) + index.ValidateAgainTool) // design 47 G §713 + if !reflect.DeepEqual(m.Tools, want) || len(want) != 50 || slices.Contains(m.Tools, "modules") { t.Errorf("tools = %v, want %v", m.Tools, want) } // Design 47 G §2, §9: every version of every registration is kept with no age limit, 64 MiB, on the old streams section. diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 6041744f..1255de8a 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -75,7 +75,8 @@ "indexer_unpin", "indexer_consumer", "indexer_runs", - "indexer_run_drop" + "indexer_run_drop", + "indexer_validate_again" ], "consumes": [ "go-package-registry.published", From 722d3a03b7af944157a9bc4ab2596e97a714aabf Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 09:50:37 +0100 Subject: [PATCH 5/6] B5g6-2a: a Go bundle's entry carries the Go module its go.mod names MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The validation's program is read under the Go module the indexer entry's Go artifact names; the indexer's artifacts are bundles, which carried no package, so every validation could not be asked. Builds: design 47 G §676, §712, A-263 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-delivery-indexer/integration_test.go | 99 ++++++++++++++++++- .../cmd/mesh-delivery-indexer/main.go | 4 +- .../cmd/mesh-delivery-indexer/versions.go | 11 +++ .../internal/index/artifacts.go | 13 ++- .../internal/index/catalog_graph_test.go | 7 +- .../mesh-delivery-indexer/takein/takein.go | 4 +- .../takein/takein_test.go | 3 + 7 files changed, 128 insertions(+), 13 deletions(-) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index 67660906..48eee1a9 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -22,6 +22,7 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo" sdkstep "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/composer" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" @@ -88,6 +89,10 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15 // The forge bridge's tool bridge_merged_message (issue 1585): answers the merged message published for a merge, by // repository@merge, as the bridge does; a restarted indexer holds none in memory and asks it. var bridgeMessages sync.Map + // The build seat node-build-go's stand-in (design 47 G §712, §713): it hears the indexer's validation asks and ends each with + // the manifest validator's verdict; refusing names, by the commit the job runs at, the manifest a revision's change refuses. + var validations *standIn + var refusing sync.Map js, ctx, candidates, _, restart := startIndexerRun(t, func(ctx context.Context, builds buildLog) { // The builder's log entries of two builds' ends (ADR 0571 clause 1), as the builder appends them. builds.end(t, "novox/mesh-delivery", "a", "modules/a") @@ -115,7 +120,12 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15 if _, err := tc.Put(ctx, key, value); err != nil { t.Fatal(err) } - seedBuildModules(t, ctx, js, nil) + // The trunk holds the indexer's entry, its Go artifact naming its package (§676): the validation's program is read from it (§712). + seedBuildModules(t, ctx, js, map[string]meshcatalog.CatalogEntry{index.Module: {V: index.CatalogEntryLayout, Module: index.Module, + Repository: "novox/mesh-delivery", Path: "modules/" + index.Module, Commit: "4444444444444444444444444444444444444444", None: sdkstep.ToolchainNoReport, + Artifacts: []meshcatalog.Artifact{{Kind: "bundle", Language: "go", Ref: index.Module + ":program", + Package: &meshcatalog.ArtifactPackage{Ecosystem: "go", Name: index.Module}}}}}) + validations = standInBuildSeat(t, ctx, js, &refusing) // The node-engine's composition on n1 (process.Conn.Call's subjects): the build of the indexer that runs there. answer := func(m *nats.Msg) { raw, _ := json.Marshal(map[string]any{"result": map[string]any{"runs": map[string]string{"fp": *running.Load()}}, "node": "n1"}) @@ -271,6 +281,13 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15 t.Fatalf("take-in.pending.%s = %+v (said %v) within 20 s, want the revision waiting for another build than %s", m1Key, held, ok, first) } named := held.WaitsFor + // IX5 (§712): the revision was validated first, by one run ask whose job names the indexer entry's repository, path and + // the revision's head, and no heads; accepted, it is never asked again while it waits for the named indexer. + if asks := validations.asked(); len(asks) != 1 || asks[0].Run == nil || asks[0].Kind != sdkstep.AskCheck || + asks[0].Run.Program.Repository != m1.Repository || asks[0].Run.Program.Dir != "modules/"+ix || asks[0].Run.Program.Commit != m1.Merge || + len(asks[0].Run.Heads) != 0 || !strings.HasSuffix(asks[0].Run.Program.Package, "/cmd/mesh-judge") { + t.Fatalf("validation asks = %+v, want one run of the indexer's manifest validator at %s", asks, m1.Merge) + } get := func(bucket, key string) ([]byte, bool) { t.Helper() if bucket == index.ModulesState { // the view of the trunk, from the newest catalog graph revision @@ -403,6 +420,81 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15 t.Errorf("the group's first member %s was said pending as a revision of its own", g1Key) } + // IX5 (§712, §713): a merge whose changed manifest gains a problem the trunk case lacks stays pending, refused naming it, + // said in take-in.pending; nothing of it is taken in. + m3 := forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: "novox/mesh-delivery", Branch: "feat/y", Number: 13, Head: h1, + Merge: "3333333333333333333333333333333333333333", Root: "r0", Trees: map[string]string{"modules/a": "tree-a-3"}, + Manifests: map[string]string{"modules/a": `{"module":"a","version":"2.0"}`}} + refusing.Store(m3.Merge, "a") + m3Key := takein.PendingKey(merged(m3)) + if refused, ok := said(takein.TakeInPending, m3Key); !ok || !strings.Contains(refused.Refused, "a: the stand-in's problem") || refused.Failed != "" { + t.Fatalf("take-in.pending.%s = %+v (said %v) within 20 s, want the revision refused naming a's problem", m3Key, refused, ok) + } + if w = waits(); len(w) != 1 || w[0].Key != m3Key || w[0].Validation == nil || w[0].Validation.Accepted != 0 || len(validations.asked()) != 2 { + t.Fatalf("indexer_pending = %+v, asks %d: want the refused revision pending with its validation", w, len(validations.asked())) + } + if raw, ok := get(index.ModulesState, "a"); !ok || strings.Contains(string(raw), m3.Merge) { + t.Errorf("modules.a = %s: the refused revision was taken in", raw) + } + +} + +// standIn is the build seat node-build-go as the indexer's validation meets it: the asks it heard, oldest first. +type standIn struct { + mu sync.Mutex + asks []sdkstep.StepAsk +} + +func (s *standIn) asked() []sdkstep.StepAsk { + s.mu.Lock() + defer s.mu.Unlock() + return slices.Clone(s.asks) +} + +// standInBuildSeat keeps the seat's check queue as a stream and hears the indexer's asks on it (accept.check.indexer): each is +// ended at once with step-ended on the seat's family, its answer inline (A-254) the manifest validator's verdict over both cases, +// the change case refusing the manifest refusing names for the job's commit. +func standInBuildSeat(t *testing.T, ctx context.Context, js jetstream.JetStream, refusing *sync.Map) *standIn { + t.Helper() + subject, err := sdkstep.AskSubjectBy(sdkstep.AskGo, sdkstep.AskCheck, index.ValidationAsker) + if err != nil { + t.Fatal(err) + } + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "SEAT_NODE_BUILD_GO_CHECK", Subjects: []string{"mesh.seat.node-build-go.accept.check.>"}}); err != nil { + t.Fatal(err) + } + // The indexer's state inputs and its objects, where the validation's request is kept for the build seat's holder (§354). + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.InputsBucket, History: 1}); err != nil { + t.Fatal(err) + } + if _, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: catalogue.BusName(index.Module, catalogue.StateObjectsName(index.InputsState))}); err != nil { + t.Fatal(err) + } + s := &standIn{} + if _, err := js.Conn().Subscribe(subject, func(m *nats.Msg) { + a, err := sdkstep.DecodeStepAsk(m.Data) + if err != nil || a.Run == nil { + t.Errorf("the stand-in build seat heard what is no run ask: %v", err) + return + } + s.mu.Lock() + s.asks = append(s.asks, a) + s.mu.Unlock() + change := composer.CaseVerdict{Name: "change", Manifests: map[string]composer.ManifestRead{}} + if module, ok := refusing.Load(a.Run.Program.Commit); ok { + change.Manifests[module.(string)] = composer.ManifestRead{Refused: "the stand-in's problem"} + } + verdict, _ := json.Marshal(composer.Verdict{V: 1, Cases: []composer.CaseVerdict{change, {Name: judge.TrunkCase, Manifests: map[string]composer.ManifestRead{}}}}) + answer, _ := json.Marshal(sdkstep.RunAnswer{Built: true, Exit: 0, Output: verdict}) + end, _ := json.Marshal(sdkstep.StepEnded{V: 1, ID: a.ID, Ask: a.ID, Asker: index.ValidationAsker, Node: "n1", Outcome: sdkstep.OutcomePassed, + Output: string(answer), At: time.Now().UTC()}) + if _, err := js.Publish(ctx, nodebuildgo.SubjectStepEnded("n1"), end); err != nil { + t.Errorf("the stand-in build seat could not say step-ended: %v", err) + } + }); err != nil { + t.Fatal(err) + } + return s } // jsSource is a bucket as a views.Source. @@ -665,7 +757,7 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before . // The shared events stream: the indexer's ordinary events and the assigner's views events (ADR 0635 §5). if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.mesh-catalog.event.>", "mesh.seat.mesh-delivery.event.>", "mesh.seat.mesh-delivery-checker.event.>", "mesh.seat.mesh-conditions.event.>", "mesh.seat.operator-channel.event.>", "mesh.seat.go-package-registry.event.>", "mesh.seat.npm-package-registry.event.>", - "mesh.seat.mesh-module-assigner.event.>"}}); err != nil { // not the seats kept in streams of their own (wanted and running, above) + "mesh.seat.mesh-module-assigner.event.>", "mesh.seat.node-build-go.event.>"}}); err != nil { // not the seats kept in streams of their own (wanted and running, above) t.Fatal(err) } // The indexer's own stream of its findings' events (design 47 G §383), which its finding writer says into (A-30). @@ -691,7 +783,7 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before . start := func() { runCtx, cancel := context.WithCancel(ctx) done := make(chan error, 1) - go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() + go func() { done <- run(runCtx, func(k string) string { return env[k] }, os.Stderr) }() stop = func() { cancel(); <-done } } start() @@ -1639,3 +1731,4 @@ func TestIndexerTakesInAWatchedBranchOtherThanTheFirstIntoItsOwnRevision(t *test t.Errorf("modules.avahi = %s, want the trunk's modules left", e.Value()) } } +var _ = io.Discard 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 cd82847e..f38ec4c3 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -476,7 +476,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The take-in step (ADR 0633): its state pending and its stream take-in, both this delivery's, bound when they appear; // until then every merge is taken in at once, as before. Pending: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.PendingBucket, "no merge is held for the indexer its revision names: each is taken in at once", logger.Printf)}, - SayTakeIn: sayTakeIn(js, cred)} + SayTakeIn: sayTakeIn(js, cred), + // The validation's run ask (design 47 G §713, A-247, A-248), put under the indexer's own asker. + ValidationAsks: buildrun.Asker{Name: index.ValidationAsker, Publisher: buildrun.JetStreamAsks{JS: js}}} // The registrations and the changes that wait for their warrant (design 47 G §2, §667): their states are bound when they // appear; a warrant is heard on the router's decided event for this module, below. registry := newRegistry(js, conn, cred, calls.Call, diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/versions.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/versions.go index cb5dc56d..1c473918 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/versions.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/versions.go @@ -7,6 +7,8 @@ import ( "sync/atomic" "time" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" + "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -121,6 +123,15 @@ func heardStepEnded(ctx context.Context, ix *index.Indexer, m jetstream.Msg, log _ = m.Ack() return } + // The end of a validation ask resumes its pending revision (design 47 G §713); its own fault is given again by the bus. + var end buildstep.StepEnded + if err := json.Unmarshal(m.Data(), &end); err == nil { + if err := ix.HearValidationEnded(ctx, end); err != nil { + logf("take-in: the end of the validation ask %s is not taken in, so the bus gives it again: %v", end.Ask, err) + _ = m.Nak() + return + } + } if err := ix.HearInputEnded(ctx, e); err != nil { logf("inputs: the input of the ask %s could not be deleted, so the bus gives it again: %v", e.Ask, err) _ = m.Nak() diff --git a/modules/mesh-delivery-indexer/internal/index/artifacts.go b/modules/mesh-delivery-indexer/internal/index/artifacts.go index e12e23e1..e48da124 100644 --- a/modules/mesh-delivery-indexer/internal/index/artifacts.go +++ b/modules/mesh-delivery-indexer/internal/index/artifacts.go @@ -13,7 +13,7 @@ import ( // Each entry carries its artifacts' package names (design 47 G §676, §491, §478, §504, §412): at take-in the indexer reads // each artifact's package name from the revision's own files at the artifact's path, the module of its go.mod for a Go -// package, the name of its package.json for an npm package, the module's and the artifact's names joined by / for an image artifact (§694), and writes it into +// package and for a Go bundle (A-263), the name of its package.json for an npm package, the module's and the artifact's names joined by / for an image artifact (§694), and writes it into // the entry's artifact as package, {ecosystem, name}. The manifest validator judges §491 from these fields. // Package ecosystems of an artifact's package (§676). @@ -47,13 +47,16 @@ func (ix *Indexer) withArtifacts(ctx context.Context, e meshcatalog.CatalogEntry // packageOf is the package name of artifact a of the entry e: read through the forge at the entry's commit, kept per // repository, commit, path and artifact so a later entry of the same commit reads nothing; nil for an artifact that names -// none (a program, an archive) or whose name cannot be read now. +// none (a program or an archive not built from a Go module) or whose name cannot be read now. func (ix *Indexer) packageOf(ctx context.Context, e meshcatalog.CatalogEntry, a catalogue.Artifact) *meshcatalog.ArtifactPackage { - switch a.Kind { - case catalogue.ArtifactPackage: - case catalogue.ArtifactImage: + switch { + case a.Kind == catalogue.ArtifactPackage: + case a.Kind == catalogue.ArtifactImage: // An image's package name is its module's name and its artifact's joined by `/`, read from no file (§694). return &meshcatalog.ArtifactPackage{Ecosystem: EcosystemImage, Name: e.Module + "/" + a.Name} + case a.Language == "go": + // A Go artifact of any kind, a bundle too, is built from the Go module its go.mod names (design 47 G A-263): the manifest + // validator's program is read under it (§712). default: return nil } diff --git a/modules/mesh-delivery-indexer/internal/index/catalog_graph_test.go b/modules/mesh-delivery-indexer/internal/index/catalog_graph_test.go index d1066aee..645ff914 100644 --- a/modules/mesh-delivery-indexer/internal/index/catalog_graph_test.go +++ b/modules/mesh-delivery-indexer/internal/index/catalog_graph_test.go @@ -33,11 +33,12 @@ func TestAnEntryCarriesItsManifestWholeAndTheDigestOfItsBytes(t *testing.T) { } // Design 47 G §676: at take-in the entry's artifacts carry the package name read from the revision's own files at the -// artifact's path: the module of a go.mod, the name of a package.json. +// artifact's path: the module of a go.mod, the name of a package.json; a Go bundle the module of its go.mod (A-263). func TestAnEntryCarriesItsArtifactsPackageNames(t *testing.T) { ix := &Indexer{ReadFile: func(_ context.Context, repository, sha, path string) ([]byte, error) { files := map[string]string{ "mods/m/go/go.mod": "module git.example/m/go\n\ngo 1.26\n", + "mods/m/go.mod": "module git.example/m\n", "mods/m/js/package.json": `{"name":"@example/m"}`, } if repository != "r" || sha != "c1" { @@ -53,6 +54,7 @@ func TestAnEntryCarriesItsArtifactsPackageNames(t *testing.T) { {Name: "jspkg", Kind: catalogue.ArtifactPackage, Language: "typescript", Paths: []string{"js"}}, {Name: "img", Kind: catalogue.ArtifactImage}, {Name: "prog", Kind: "bundle", Language: "go"}, + {Name: "web", Kind: "bundle", Language: "typescript"}, }}} e := ix.withArtifacts(context.Background(), meshcatalog.CatalogEntry{Module: "m", Repository: "r", Path: "mods/m", Commit: "c1"}, m) got := map[string]string{} @@ -63,7 +65,8 @@ func TestAnEntryCarriesItsArtifactsPackageNames(t *testing.T) { } got[a.Ref] = name } - want := map[string]string{"m:gopkg": "go git.example/m/go", "m:jspkg": "npm @example/m", "m:img": "image m/img", "m:prog": "none"} + want := map[string]string{"m:gopkg": "go git.example/m/go", "m:jspkg": "npm @example/m", "m:img": "image m/img", + "m:prog": "go git.example/m", "m:web": "none"} // A-263: a Go bundle carries the Go module its go.mod names if !reflect.DeepEqual(got, want) { t.Errorf("artifacts' packages %v, want %v (§676)", got, want) } diff --git a/modules/mesh-delivery-indexer/takein/takein.go b/modules/mesh-delivery-indexer/takein/takein.go index 203872b0..056b521f 100644 --- a/modules/mesh-delivery-indexer/takein/takein.go +++ b/modules/mesh-delivery-indexer/takein/takein.go @@ -175,8 +175,8 @@ func (t TakeIn) check() error { if !strings.HasPrefix(t.Key, "m.") { return fmt.Errorf("delivery: a revision's take-in key is its oldest merge's m., not %q", t.Key) } - if t.Kind == TakeInPending && t.WaitsFor == "" && t.Unknown == "" && len(t.Lost) == 0 { - return fmt.Errorf("delivery: a pending revision %s names the indexer build it waits for, why the indexer does not know its own, or the merges it lost", t.Key) + if t.Kind == TakeInPending && t.WaitsFor == "" && t.Unknown == "" && len(t.Lost) == 0 && t.Refused == "" && t.Failed == "" { + return fmt.Errorf("delivery: a pending revision %s names the indexer build it waits for, why the indexer does not know its own, the merges it lost, its refusal or why it could not validate", t.Key) } case TakeInGroup: if t.Group == "" { diff --git a/modules/mesh-delivery-indexer/takein/takein_test.go b/modules/mesh-delivery-indexer/takein/takein_test.go index 8d971c64..f69c1e49 100644 --- a/modules/mesh-delivery-indexer/takein/takein_test.go +++ b/modules/mesh-delivery-indexer/takein/takein_test.go @@ -22,6 +22,9 @@ func sampleTakeIns(since time.Time) []TakeIn { {Kind: TakeInTaken, Key: PendingKey(12), What: TakeInRevision, Merges: m, WaitsFor: "src2:abc", Since: since, By: "lead-47@shanks", Why: "the named indexer can never run"}, {Kind: TakeInPending, Key: PendingKey(13), What: TakeInRevision, Merges: m, Unknown: "the engine does not answer", Since: since}, // the indexer cannot learn its own build + // design 47 G §713: a revision refused by its validation, and one that could not validate. + {Kind: TakeInPending, Key: PendingKey(14), What: TakeInRevision, Merges: m, Refused: "a: names no version", Since: since}, + {Kind: TakeInPending, Key: PendingKey(15), What: TakeInRevision, Merges: m, Failed: "could not run: no build node", Since: since}, } } From fb62b14512637c6c4e5635ba3c12ce7037fd1ac0 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 09:50:37 +0100 Subject: [PATCH 6/6] B5g6-2: the indexer validates every pending revision before its take-in MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The program puts the validation's run ask on node-build-go's check queue under its own asker and hears its end on step-ended; the manifest invokes seat:node-build-go.accept.check. The revision validated is the branch its merges landed on (the merged message's base, else the trunk), never the pull request's branch. take-in.pending may name a refusal or why the revision could not validate instead of the build it waits for. The whole case runs a stand-in build seat: an accepted revision is taken in as before, by one ask whose job names the indexer entry's repository, path and the revision's head and no heads; a merge whose changed manifest gains a problem stays pending, refused naming it. Builds: design 47 G §712, §713, §714, A-231, A-232, A-247, A-254 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-delivery-indexer/integration_test.go | 61 ++++++++++++++++--- .../internal/index/validation_wait.go | 6 +- .../mesh-delivery-indexer/manifest_test.go | 4 +- modules/mesh-delivery-indexer/module.json | 1 + 4 files changed, 59 insertions(+), 13 deletions(-) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index 48eee1a9..a995b7a3 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -120,11 +120,21 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15 if _, err := tc.Put(ctx, key, value); err != nil { t.Fatal(err) } - // The trunk holds the indexer's entry, its Go artifact naming its package (§676): the validation's program is read from it (§712). - seedBuildModules(t, ctx, js, map[string]meshcatalog.CatalogEntry{index.Module: {V: index.CatalogEntryLayout, Module: index.Module, - Repository: "novox/mesh-delivery", Path: "modules/" + index.Module, Commit: "4444444444444444444444444444444444444444", None: sdkstep.ToolchainNoReport, - Artifacts: []meshcatalog.Artifact{{Kind: "bundle", Language: "go", Ref: index.Module + ":program", - Package: &meshcatalog.ArtifactPackage{Ecosystem: "go", Name: index.Module}}}}}) + // The trunk holds the indexer's entry, taken in at an earlier commit with its own manifest; the seat git answers its go.mod, so + // its Go artifacts carry the Go module the validation's program is read under (§712, §676, A-263). + seedBuildModules(t, ctx, js, map[string]meshcatalog.CatalogEntry{index.Module: trunkIndexerEntry(t, ctx, js)}) + if _, err := js.Conn().Subscribe("mesh.seat.git.tool.file", func(m *nats.Msg) { + var a git.FileInput + _ = json.Unmarshal(m.Data, &a) + answer := git.FileAnswer{Path: a.Path, Ref: a.Ref, Absent: true} + if a.Path == "modules/"+index.Module+"/go.mod" { + answer = git.FileAnswer{Path: a.Path, Ref: a.Ref, Content: "module " + index.Module + "\n"} + } + raw, _ := json.Marshal(map[string]any{"result": answer, "node": "n0"}) + _ = m.Respond(raw) + }); err != nil { + t.Fatal(err) + } validations = standInBuildSeat(t, ctx, js, &refusing) // The node-engine's composition on n1 (process.Conn.Call's subjects): the build of the indexer that runs there. answer := func(m *nats.Msg) { @@ -430,8 +440,8 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15 if refused, ok := said(takein.TakeInPending, m3Key); !ok || !strings.Contains(refused.Refused, "a: the stand-in's problem") || refused.Failed != "" { t.Fatalf("take-in.pending.%s = %+v (said %v) within 20 s, want the revision refused naming a's problem", m3Key, refused, ok) } - if w = waits(); len(w) != 1 || w[0].Key != m3Key || w[0].Validation == nil || w[0].Validation.Accepted != 0 || len(validations.asked()) != 2 { - t.Fatalf("indexer_pending = %+v, asks %d: want the refused revision pending with its validation", w, len(validations.asked())) + if w = waits(); len(w) != 1 || w[0].Key != m3Key || w[0].Validation == nil || w[0].Validation.Accepted != 0 || len(validations.asked()) != 3 { + t.Fatalf("indexer_pending = %+v, asks %d: want the refused revision pending with its validation, asked once after m1 and the group", w, len(validations.asked())) } if raw, ok := get(index.ModulesState, "a"); !ok || strings.Contains(string(raw), m3.Merge) { t.Errorf("modules.a = %s: the refused revision was taken in", raw) @@ -439,6 +449,37 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15 } +// trunkIndexerEntry keeps the indexer's own manifest by its digest, accepted, and answers its trunk entry at an earlier commit. +func trunkIndexerEntry(t *testing.T, ctx context.Context, js jetstream.JetStream) meshcatalog.CatalogEntry { + t.Helper() + raw, err := os.ReadFile("../../module.json") + if err != nil { + t.Fatal(err) + } + var m catalogue.Manifest + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + digest := meshcatalogclient.ManifestDigest(raw) + manifests, err := js.KeyValue(ctx, index.ManifestsBucket) + if err != nil { + t.Fatal(err) + } + mkey, _ := meshcatalogclient.ManifestsKey(digest) + mdata, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: index.Module, Manifest: string(raw)}) + if _, err := manifests.Put(ctx, mkey, mdata); err != nil { + t.Fatal(err) + } + held := meshcatalog.ModulesByModuleValue{Repository: "novox/mesh-delivery", Commit: "4444444444444444444444444444444444444444", Path: "modules/" + index.Module, + Manifest: digest, None2: sdkstep.ToolchainNoReport} + e := index.CatalogEntryOf(index.Module, held, m, string(raw), graph.Node{None: held.None2}) + for _, a := range m.Build.Artifacts { + e.Artifacts = append(e.Artifacts, meshcatalog.Artifact{Kind: a.Kind, Language: a.Language, Ref: index.ArtifactRefOf(index.Module, a.Name), + Package: &meshcatalog.ArtifactPackage{Ecosystem: index.EcosystemGo, Name: index.Module}}) + } + return e +} + // standIn is the build seat node-build-go as the indexer's validation meets it: the asks it heard, oldest first. type standIn struct { mu sync.Mutex @@ -783,7 +824,7 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before . start := func() { runCtx, cancel := context.WithCancel(ctx) done := make(chan error, 1) - go func() { done <- run(runCtx, func(k string) string { return env[k] }, os.Stderr) }() + go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() stop = func() { cancel(); <-done } } start() @@ -1536,7 +1577,8 @@ func serveAssignerAssignments(t *testing.T, ctx context.Context, js jetstream.Je t.Fatal(err) } if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-indexer", AckPolicy: jetstream.AckExplicitPolicy, - FilterSubjects: []string{"mesh.mod." + assignerHolderModule + ".event.views.>", "mesh.seat.mesh-delivery.event.>"}}); err != nil { + // and the build seat's step-ended the manifest's consumes names (design 47 G §713: the end of a validation ask). + FilterSubjects: []string{"mesh.mod." + assignerHolderModule + ".event.views.>", "mesh.seat.mesh-delivery.event.>", nodebuildgo.SubjectStepEndedAll}}); err != nil { t.Fatal(err) } em := views.Emitter{Owner: assignerHolderModule, Subject: process.Account{Node: "n1", Module: assignerHolderModule}.EventSubject, @@ -1731,4 +1773,3 @@ func TestIndexerTakesInAWatchedBranchOtherThanTheFirstIntoItsOwnRevision(t *test t.Errorf("modules.avahi = %s, want the trunk's modules left", e.Value()) } } -var _ = io.Discard diff --git a/modules/mesh-delivery-indexer/internal/index/validation_wait.go b/modules/mesh-delivery-indexer/internal/index/validation_wait.go index f24e27a1..5e9c7f1e 100644 --- a/modules/mesh-delivery-indexer/internal/index/validation_wait.go +++ b/modules/mesh-delivery-indexer/internal/index/validation_wait.go @@ -35,9 +35,11 @@ func newestOfRevision(ready []PendingEntry) int { return n } +// branchOfRevision is the watched branch the revision's merges landed on, its newest merged message's base (design 47 G §714); +// a message that names none is the trunk's. A pending value's own branch is its pull request's, never the branch validated. func branchOfRevision(ready []PendingEntry) string { - if ref := ready[newestOfRevision(ready)].Value.Ref; ref != nil && ref.Branch != "" { - return ref.Branch + if base := ready[newestOfRevision(ready)].Forge.Base; base != "" { + return base } return TrunkBranch } diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 5dfaa46b..defaa435 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -179,7 +179,9 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // Three invokes: the file of a registered module's manifest and lock files (registered.go), the forge's trees (the named paths' objects, since the indexer is the one reader of a manifest's paths: // ADR 0499 clauses 2 and 3, ADR 0495 clause 1; and the module directories of the trunk, issue 1900) and whether a // commit is in another (issue 1900). - if want := []string{"seat:git.trees", "seat:git.file", "seat:git.contains", "seat:git.open-pulls", "seat:git.branch", "mesh-delivery-forge-bridge.bridge_merged_message", "seat:node-engine.composition", "seat:mesh-delivery.source", "seat:mesh-module-assigner.assignments", "seat:mesh-module-assigner.placement", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence", "seat:node-build-go.steps", "seat:node-build-node.steps", "seat:node-build-image.steps", "mesh-delivery-builder.builder_versions"}; !reflect.DeepEqual(m.Invokes, want) { + if want := []string{"seat:git.trees", "seat:git.file", "seat:git.contains", "seat:git.open-pulls", "seat:git.branch", "mesh-delivery-forge-bridge.bridge_merged_message", "seat:node-engine.composition", "seat:mesh-delivery.source", "seat:mesh-module-assigner.assignments", "seat:mesh-module-assigner.placement", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence", "seat:node-build-go.steps", + "seat:node-build-go.accept.check", // design 47 G A-231: the validation's run ask (§712) on the check queue + "seat:node-build-node.steps", "seat:node-build-image.steps", "mesh-delivery-builder.builder_versions"}; !reflect.DeepEqual(m.Invokes, want) { t.Errorf("invokes = %v, want %v", m.Invokes, want) } // 1945: indexer_take_in, served by its tools bundle through the takeins state; the process serves none. The same bundle diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 1255de8a..26fdb649 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -175,6 +175,7 @@ "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence", "seat:node-build-go.steps", + "seat:node-build-go.accept.check", "seat:node-build-node.steps", "seat:node-build-image.steps", "mesh-delivery-builder.builder_versions"