Merge branch 'feat/47-G-fix11-b5g6' into feat/47-G-join11

This commit is contained in:
2026-11-12 10:37:21 +01:00
22 changed files with 836 additions and 29 deletions
@@ -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.<queue>.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.<queue>.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
}
@@ -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"}
}
@@ -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
@@ -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,22 @@ 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, 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) {
raw, _ := json.Marshal(map[string]any{"result": map[string]any{"runs": map[string]string{"fp": *running.Load()}}, "node": "n1"})
@@ -271,6 +291,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 +430,112 @@ 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()) != 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)
}
}
// 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
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 +798,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).
@@ -1444,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,
@@ -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,
@@ -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
@@ -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()
@@ -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
}
@@ -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)
@@ -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)
}
@@ -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
}
@@ -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{})
@@ -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.<sequence>, 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
}}}
}
@@ -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.<key>`, and the
// value. key is a revision's wait key, m.<sequence>, 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.<sequence>, 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.<take-in|drop>.<key>`, 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.
@@ -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.
@@ -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 `<seat>.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.
@@ -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)
}
}
@@ -0,0 +1,340 @@
package index
import (
"context"
"encoding/json"
"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"
"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
}
// 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 base := ready[newestOfRevision(ready)].Forge.Base; base != "" {
return base
}
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), ""
}
// 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
}
@@ -0,0 +1,113 @@
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)
}
// 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 {
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())
}
}
@@ -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
@@ -202,8 +204,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.
+3 -1
View File
@@ -75,7 +75,8 @@
"indexer_unpin",
"indexer_consumer",
"indexer_runs",
"indexer_run_drop"
"indexer_run_drop",
"indexer_validate_again"
],
"consumes": [
"go-package-registry.published",
@@ -174,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"
+20 -2
View File
@@ -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"`
@@ -172,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.<sequence>, 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 == "" {
@@ -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.
@@ -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},
}
}