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}, } }