diff --git a/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/candidate.go b/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/candidate.go index e609af0b..27a90dc8 100644 --- a/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/candidate.go +++ b/modules/mesh-delivery-contracts/go/clients/meshcatalogclient/candidate.go @@ -62,10 +62,11 @@ func ParseCandidatesKey(key string) (string, error) { return group, nil } -var tryGroupOK = regexp.MustCompile(`^try-[0-9]{1,20}$`) +var tryGroupOK = regexp.MustCompile(`^(try-[0-9]{1,20}|run-[A-Za-z0-9_-]{1,64})$`) -// IsTryGroup says a candidates key is the group of a whole try, try-, which is also the try's id: the indexer makes the -// candidate of a try's group from its previews, and the checker judges it by its try, never as a pull request's check. +// IsTryGroup says a candidates key is the group of a whole try, try-, which is also the try's id, or of a run on +// request, run- (design 47 G A-160): the indexer makes the candidate of a run from the runner's run-started, and the +// checker judges neither as a pull request's check. func IsTryGroup(key string) bool { return tryGroupOK.MatchString(key) } // NoneNoPreview is why a candidate's touched module has no fingerprint of the new form yet: its merge preview is not 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 e1736091..9248d6e3 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 @@ -309,7 +309,6 @@ func main() { _, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: index.FingerprintsAgainTool, Verb: "fingerprints-again"}) return err }, 10*time.Second)...) - tools = append(tools, index.TryCandidateTools(keptRead{stdio.State("candidates")}, keptRead{stdio.State(index.TryRecoveryState)}, asks, time.Now)...) tools = append(tools, configReadersTool(stdio.NewWatchedStates(), trunk.State("modules")), busStateTool(trunk.State), requirementsTool(keptSource), packageUsersTool(trunk.State), seatDefinitionTool(keptSource), outOfStepTool(keptSource)) if err := stdio.Serve("", tools); err != nil { 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 21b60028..ddf6699a 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 @@ -33,9 +33,10 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/git" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdeliverychecker" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" @@ -1111,86 +1112,98 @@ func TestIndexerWritesAGoModuleAgainWhenTheBuildModuleReportsItsToolchain(t *tes } } -// TestIndexerGivesACandidateItsFingerprintWhenTheCheckerCreatesThePreviewsBucket: the indexer starts with the stream PREVIEWS -// and its durable consumer (made by the controller from the checker's declaration and the indexer's consumes) and no bucket -// mesh-delivery-checker_previews; a pull request of mesh-catalog touching avahi makes the candidate with the reason -// 'no merge preview is computed'; the checker then makes its bucket, writes the preview of that head onto the trunk head -// the source names, and says so on its stream; the candidate's avahi gets, within 20 s and with no restart, the reason -// computed from the preview (the toolchain is not reported here, so 'toolchain not yet reported'). -func TestIndexerGivesACandidateItsFingerprintWhenTheCheckerCreatesThePreviewsBucket(t *testing.T) { // 1445/AC4 +// TestIndexerBuildsARunsCandidateAndDropsItWhenEveryMembersPipelineEnded (design 47 G §64, §67 test 64, §190; A-160, A-174): +// the indexer starts with the runner's streams run-started and pipeline-ended and its durables on them, its state runs and a +// registration of avahi; the runner says run-started of run 9 with avahi at a commit on a branch no registration watches; +// within 20 s candidates.run-9 holds one pull of number 0 at that commit, avahi's manifest read through the seat git; the +// indexer is restarted, the runner says avahi's pipeline of run 9 ended, and the candidate and the run's key go. +func TestIndexerBuildsARunsCandidateAndDropsItWhenEveryMembersPipelineEnded(t *testing.T) { manifest, err := os.ReadFile("../../internal/index/testdata/catalogue-a67f0196-avahi.module.json") if err != nil { t.Fatal(err) } const goSum = "3b5028f0c1e9bde04dd766abd63eb5931bc516bae695cf46f00abcd8510b0381" - const trunk, head = "5555555555555555555555555555555555555555", "0123456789abcdef0123456789abcdef01234567" - js, ctx, candidates, _ := startIndexer(t, func(ctx context.Context, builds buildLog) { - builds.end(t, "novox/mesh-catalog", "avahi", "modules/avahi") - }, func(ctx context.Context, js jetstream.JetStream) { - if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: PreviewsStream, Subjects: []string{previewWrittenSubject}}); err != nil { - t.Fatal(err) - } - if _, err := js.CreateOrUpdateConsumer(ctx, PreviewsStream, jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-indexer", AckPolicy: jetstream.AckExplicitPolicy}); err != nil { - t.Fatal(err) - } - seedBuildModules(t, ctx, js, nil) - // The seat mesh-delivery's verb source, as the gate serves it (design 47 G §115): main names the catalogue's trunk head. - answer := func(m *nats.Msg) { - raw, _ := json.Marshal(map[string]any{"result": meshdelivery.SourceAnswer{Branch: "main", Group: "g", Revision: 5, - Repositories: []meshdelivery.SourceHead{{Repository: "novox/mesh-catalog", Head: trunk, Trunk: "main"}}}, "node": "n1"}) - _ = m.Respond(raw) - } - for _, subject := range []string{"mesh.seat.mesh-delivery.call.source.>", "mesh.seat.mesh-delivery.tool.source"} { - if _, err := js.Conn().Subscribe(subject, answer); err != nil { + const commit = "0123456789abcdef0123456789abcdef01234567" + var runs jetstream.KeyValue + js, ctx, candidates, _, restart := startIndexerRun(t, func(context.Context, buildLog) {}, func(ctx context.Context, js jetstream.JetStream) { + for stream, subject := range map[string]string{meshpipelinerunner.StreamRunStarted: meshpipelinerunner.SubjectRunStartedAll, + meshpipelinerunner.StreamPipelineEnded: meshpipelinerunner.SubjectPipelineEndedAll} { + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{subject}}); err != nil { + t.Fatal(err) + } + if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-indexer", AckPolicy: jetstream.AckExplicitPolicy, + FilterSubjects: []string{subject}}); err != nil { t.Fatal(err) } } + var err error + if runs, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.RunsBucket, History: 1}); err != nil { + t.Fatal(err) + } + regs, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.RegistrationsBucket, History: 1}) + if err != nil { + t.Fatal(err) + } + reg, _ := json.Marshal(meshcatalog.Registration{V: 1, Module: "avahi", Origin: "novox/mesh-catalog", Path: "modules/avahi", Branches: []string{"main"}}) + if _, err := regs.Create(ctx, index.RegistrationKey("avahi"), reg); err != nil { + t.Fatal(err) + } + seedBuildModules(t, ctx, js, nil) + // The seat git as the forge module serves it: avahi's directory and its manifest at the run's commit. + reply := func(m *nats.Msg, result any) { + raw, _ := json.Marshal(map[string]any{"result": result, "node": "n0"}) + _ = m.Respond(raw) + } + if _, err := js.Conn().Subscribe("mesh.seat.git.tool.trees", func(m *nats.Msg) { + reply(m, git.TreesAnswer{Root: "r0", Commit: commit, Trees: map[string]string{"modules/avahi": "newtree"}}) + }); err != nil { + t.Fatal(err) + } + if _, err := js.Conn().Subscribe("mesh.seat.git.tool.file", func(m *nats.Msg) { + var a git.FileInput + _ = json.Unmarshal(m.Data, &a) + switch a.Path { + case "modules/avahi/module.json": + reply(m, git.FileAnswer{Path: a.Path, Ref: a.Ref, Content: string(manifest)}) + case "modules/avahi/go.sum": + reply(m, git.FileAnswer{Path: a.Path, Ref: a.Ref, Content: goSum}) + default: + reply(m, git.FileAnswer{Path: a.Path, Ref: a.Ref, Absent: true}) + } + }); err != nil { + t.Fatal(err) + } }) - pull, err := forgemsg.EncodeForge(forgemsg.Forge{Kind: forgemsg.ForgePull, Repository: "novox/mesh-catalog", Branch: "feat/x", Number: 301, Head: head, Action: forgemsg.PullUpdated, - Dirs: []string{"modules/avahi"}, DirsKnown: true, Manifests: map[string]string{"modules/avahi": string(manifest)}}) - if err != nil { + started, _ := json.Marshal(meshpipelinerunner.RunStarted{V: 1, Run: "9", By: "lead@n1", Members: []meshpipelinerunner.Member{{Module: "avahi", Commit: commit, Branch: "feature/1966-x"}}}) + if _, err := js.Publish(ctx, meshpipelinerunner.SubjectRunStarted("avahi", "feature-1966-x", commit), started); err != nil { t.Fatal(err) } - if _, err := js.Publish(ctx, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-catalog.pull", pull); err != nil { - t.Fatal(err) - } - none2 := func(want string) string { - var got string - for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { - if e, err := candidates.Get(ctx, "feat/x"); err == nil { - if c, err := index.DecodeCandidate(e.Value()); err == nil && len(c.Pulls) == 1 { - got = c.Pulls[0].Fps["avahi"].None2 - if got == want { - break - } - } + key, _ := meshcatalogclient.CandidatesKey("run-9") + var c meshcatalog.CandidatesByGroupValue + for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if e, err := candidates.Get(ctx, key); err == nil { + if c, err = index.DecodeCandidate(e.Value()); err == nil { + break } } - return got } - if got := none2("no merge preview is computed"); got != "no merge preview is computed" { - t.Fatalf("none2 = %q within 20 s of the pull request, want 'no merge preview is computed'", got) + if len(c.Pulls) != 1 || c.Pulls[0].Number != 0 || c.Pulls[0].Head != commit || c.Pulls[0].Fps["avahi"].Manifest != meshcatalogclient.ManifestDigest(manifest) { + t.Fatalf("candidates.run-9 = %+v within 20 s of run-started, want avahi at the run's commit with its manifest", c) } - // The checker's bucket is made only now, as when the checker is assigned after the indexer started. - kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.PreviewsBucket, History: index.PreviewsHistory}) - if err != nil { + restart() + ended, _ := json.Marshal(meshpipelinerunner.PipelineEnded{V: 1, Pipeline: "feature-1966-x.0.avahi", Module: "avahi", Branch: "feature/1966-x", Commit: commit, + Outcome: "passed", Steps: []meshpipelinerunner.StepResult{}, Run: "9"}) + if _, err := js.Publish(ctx, meshpipelinerunner.SubjectPipelineEnded("avahi", "feature-1966-x", commit), ended); err != nil { t.Fatal(err) } - pv := meshdeliverychecker.PreviewsByRepositoryHeadTrunkValue{Repository: "novox/mesh-catalog", Head: head, Trunk: trunk, Merge: "6666666666666666666666666666666666666666", - Modules: map[string]meshdeliverychecker.PreviewsByRepositoryHeadTrunkValueModulesValue{"modules/avahi": {Tree: "newtree", Manifest: meshcatalogclient.ManifestDigest(manifest), Locks: map[string]string{"go.sum": goSum}}}} - key, _ := index.PreviewKey(pv.Repository, pv.Head, pv.Trunk) - data, err := index.EncodePreview(pv) - if err != nil { - t.Fatal(err) + gone := false + for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline) && !gone; time.Sleep(50 * time.Millisecond) { + _, cerr := candidates.Get(ctx, key) + _, rerr := runs.Get(ctx, "run-9") + gone = errors.Is(cerr, jetstream.ErrKeyNotFound) && errors.Is(rerr, jetstream.ErrKeyNotFound) } - if _, err := kv.Create(ctx, key, data); err != nil { - t.Fatal(err) - } - if _, err := js.Publish(ctx, previewWrittenSubject, []byte(`{"key":"`+key+`"}`)); err != nil { - t.Fatal(err) - } - if got := none2(sdkstep.ToolchainNoReport); got != sdkstep.ToolchainNoReport { - t.Errorf("none2 = %q within 20 s of the preview, want %q (computed from the preview)", got, sdkstep.ToolchainNoReport) + if !gone { + t.Error("the run's candidate or its runs key stays 20 s after its member's pipeline ended across a restart") } } 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 70451ace..5603ae66 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -407,7 +407,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The manifests a take-in wrote and no judge pass judged yet (issue 1966): a state this delivery brings, bound when it // appears; until then a take-in's new manifests are kept for their judging in its memory alone. unjudged := process.AwaitBucket(ctx, js, conn, index.UnjudgedBucket, "a take-in's new manifests are kept for their judging in its memory alone", logger.Printf) - tryRecovery := process.AwaitBucket(ctx, js, conn, index.TryRecoveryBucket, "how the recovery of the open tries went is not kept", logger.Printf) + // The runs on request the indexer holds a candidate for (A-174): a state this delivery brings, bound when it appears. + runs := process.AwaitBucket(ctx, js, conn, index.RunsBucket, "no run on request's candidate is written or dropped", logger.Printf) findings := process.AwaitBucket(ctx, js, conn, findingsBucket, "no finding of the indexer is written", logger.Printf) // The indexer's findings are the seat mesh-conditions' generated finding, written through mesh-sdk's held finding writer // (design 47 G §689, A-29): it holds what is written before the state is bound, and starts once the state is. @@ -444,7 +445,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro Predefined: awaitedStore{b: predefined}, ConfigCandidates: awaitedStore{b: configCandidates}, Logf: logger.Printf, Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call), ReadOpen: openVia(calls.Call), Packages: awaitedStore{b: packages, js: js, name: index.PackagesBucket}, Files: filesVia(calls.Call), GoSum: goSumVia(calls.Call), GoMod: goModVia(calls.Call), - OwnBuild: ownBuildVia(calls.Call, cred.Node), OpenTries: openTriesVia(calls.Call), OwnBuildState: awaitedStore{b: ownBuildState}, Heard: heardStore, Placement: placementVia(calls.CallKept), RunningWhole: runningWholeVia(js), TryRecovery: awaitedStore{b: tryRecovery}, Graph: graph, + OwnBuild: ownBuildVia(calls.Call, cred.Node), OwnBuildState: awaitedStore{b: ownBuildState}, Heard: heardStore, Placement: placementVia(calls.CallKept), RunningWhole: runningWholeVia(js), Runs: awaitedStore{b: runs}, Graph: graph, SeatDefinitions: awaitedStore{b: seatDefs}, ReadFile: readFileVia(calls.Call), SaySeatDefinition: saySeats, Message: messageVia(calls.CallKept), Revisions: revisions, EntryIndex: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.EntryIndexBucket, "no entry is found by its fingerprint", logger.Printf)}, // Message a kept call: a read, called again through a transport fault (ADR 0668 §1) // Each catalog revision may bring another holder of a build seat: the build modules are found again (buildmodules.go). @@ -665,17 +666,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro buildModules.hearWith(func(stream, module string) { hear(stream, module+"'s toolchain reports are not heard until the indexer starts again", func(m jetstream.Msg) { tools.Reported(m) }) }) - // The consumer on the checker's stream exists before the open tries are listed and is heard after (ADR 0638 §3): an event - // said between the call and the hearing waits in it, and a try that started or ended meanwhile is found either way. - var checksCons jetstream.Consumer - _ = notHeard.Bind(PreviewsStream, "the checker's previews and tries are not heard until the indexer starts again", func() error { - c, err := cred.EventConsumer(ctx, js, PreviewsStream) - if err != nil { - return fmt.Errorf("the consumer %s_%s on %s does not exist yet: %w", cred.Node, cred.Module, PreviewsStream, err) - } - checksCons = c - return nil - }) // The versions each package's series holds (DLV-I5): the registry seats' published, on the // indexer's durable consumer of the shared events stream. Where it is not there yet, no series moves until the indexer // starts again, said. @@ -723,6 +713,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return nil }) hearNamed(ctx, js, cred, notHeard, named, logger.Printf) + hearRuns(ctx, js, cred, notHeard, ix, logger.Printf) // The checker's verdicts carry a pull request's out-of-step files (DLV-I20): on the indexer's durable consumer of the // checker's stream asks. Where it is not there yet, no pull request's out-of-step files are kept until the indexer // starts again, said. @@ -741,17 +732,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } else if words != "" { logger.Printf("take-in at start: %s", words) } - // The open tries are found again through the checker's tool: a try event missed while the indexer was down would leave a - // try with no candidate, or a candidate with no try. Not fatal: the events come, and a preview written writes the candidates. - // A recovery that fails is said (journal, and indexer_try_candidates) and the events are heard all the same. - _ = ix.RecoverTries(ctx) - if checksCons != nil { - pc, err := process.HearEvents(checksCons, func(m jetstream.Msg) { previews.heardChecks(m) }, logger.Printf) - if err != nil { - return err - } - defer pc.Stop() - } cons, err := cred.EventConsumer(ctx, js, forgeStream) if err != nil { // Essential: the forge's merges are what the indexer indexes. diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named.go index c99ab795..41d4a90f 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named.go @@ -6,7 +6,6 @@ import ( "encoding/hex" "encoding/json" "fmt" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-checker/checkermsg" "path" "slices" "strings" @@ -261,31 +260,3 @@ func openVia(call toolCaller) func(ctx context.Context) ([]forgemsg.OpenPull, er return out, nil } } - -// openTriesTool is the checker's tool that lists the tries that wait or run (mesh-delivery-checker checker_tries). -const openTriesTool = "mesh-delivery-checker.checker_tries" - -// openTriesVia lists the open tries through the checker's tool, never its state (ADR 0596). -func openTriesVia(call toolCaller) func(ctx context.Context) ([]checkermsg.Open, error) { - return func(context.Context) ([]checkermsg.Open, error) { - raw, err := call(openTriesTool, map[string]any{}) - if err != nil { - return nil, err - } - // No try open is an answer: the checker's list is empty, or null in an older checker (issue 1921). An answer without the - // key is refused below, never read as none. - var o map[string]json.RawMessage - if err := json.Unmarshal(raw, &o); err != nil { - return nil, fmt.Errorf("the answer of %s is not {tries}: %v", openTriesTool, err) - } - list, ok := o["tries"] - if !ok { - return nil, fmt.Errorf("the answer of %s is not {tries}: no tries key", openTriesTool) - } - var tries []checkermsg.Open - if err := json.Unmarshal(list, &tries); err != nil { - return nil, fmt.Errorf("the answer of %s is not {tries}: %v", openTriesTool, err) - } - return tries, nil - } -} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named_test.go index 980e96bd..52f3b9c4 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named_test.go @@ -4,7 +4,9 @@ import ( "context" "encoding/json" "errors" + "fmt" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/git" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" "mesh-delivery-indexer/internal/index" "slices" "strings" @@ -127,23 +129,30 @@ func TestTheTrunkIsTheDirectoriesTheForgeListsTrees(t *testing.T) { // 1900/AC3 if _, err := ownBuildVia(none, "n1")(context.Background()); err == nil { t.Error("an engine that says nothing runs gave a build") } - // Issue 1921: the checker's list of open tries; none open is an answer, whether [] or null. + // Design 47 G §64, A-160, A-174: the runner's run-started and pipeline-ended reach the indexer; one that is no such body, + // or a run that can never be taken in, is terminated, any other failure is left for redelivery, and a taken one is acked. for _, row := range []struct { - name, answer string - want int - refused bool + name, body string + ended bool + err error + acked, termd int + started, endedOf string }{ - {"one open", `{"tries":[{"id":"try-1","state":"running","whole":true,"members":[{"repository":"novox/a","head":"h","trunk":"t"}]}]}`, 1, false}, - {"an empty list", `{"tries":[]}`, 0, false}, - {"null, as an older checker answers none", `{"tries":null}`, 0, false}, - {"no tries key", `{}`, 0, true}, - {"not json", `nope`, 0, true}, - {"tries of another shape", `{"tries":{"a":1}}`, 0, true}, + {name: "a run started", body: `{"v":1,"run":"7","members":[{"module":"a","commit":"c"}],"by":"s"}`, acked: 1, started: "7"}, + {name: "a run's pipeline ended", body: `{"v":1,"pipeline":"p","module":"a","branch":"b","commit":"c","outcome":"passed","steps":[],"run":"7"}`, ended: true, acked: 1, endedOf: "7"}, + {name: "not json", body: `nope`, termd: 1}, + {name: "a run never taken in", body: `{"v":1,"run":"7","members":[]}`, err: fmt.Errorf("%w: no member", index.ErrUnreadable), termd: 1, started: "7"}, + {name: "a bus fault", body: `{"v":1,"run":"7","members":[{"module":"a","commit":"c"}]}`, err: errors.New("the bus did not answer"), started: "7"}, } { - call := func(string, any) (json.RawMessage, error) { return json.RawMessage(row.answer), nil } - got, err := openTriesVia(call)(context.Background()) - if (err != nil) != row.refused || len(got) != row.want { - t.Errorf("%s: %d tries, error %v; want %d, refused %v", row.name, len(got), err, row.want, row.refused) + ix := &fakeRuns{err: row.err} + m := &fakeMsg{subject: "x", data: []byte(row.body)} + if row.ended { + heardPipelineEnded(context.Background(), ix, m, t.Logf) + } else { + heardRunStarted(context.Background(), ix, m, t.Logf) + } + if m.acked != row.acked || m.termd != row.termd || ix.started != row.started || ix.ended != row.endedOf { + t.Errorf("%s: acked %d, terminated %d, started %q, ended %q", row.name, m.acked, m.termd, ix.started, ix.ended) } } } @@ -177,3 +186,19 @@ func TestReadViaReadsManifestsAndLocks(t *testing.T) { t.Fatalf("read = %+v", got) } } + +// fakeRuns records the run events handed to it and answers err. +type fakeRuns struct { + err error + started, ended string +} + +func (f *fakeRuns) StartRun(_ context.Context, e meshpipelinerunner.RunStarted) error { + f.started = e.Run + return f.err +} + +func (f *fakeRuns) EndRunPipeline(_ context.Context, e meshpipelinerunner.PipelineEnded) error { + f.ended = e.Run + return f.err +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go index a38c80b3..766e7270 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go @@ -4,30 +4,19 @@ import ( "context" "errors" "log" - "strings" "sync" "git.novox.be/novox/mesh-sdk/go/process" "github.com/nats-io/nats.go/jetstream" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-checker/checkermsg" - "mesh-delivery-indexer/internal/index" ) -// PreviewsStream is the stream the checker carries its events on, as the bus names it (declared by the -// checker as the stream checks, heard by this module through consumes; novox/hq ADR 0484 clause 3). -const PreviewsStream = "CHECKS" - -// previewWrittenSubject is the subject of the checker's preview.written event. -const previewWrittenSubject = "mesh.stream." + index.CheckerModule + ".event.preview.written" - // previewWatch reads the merge previews the checker writes (mesh-delivery-checker_previews) and follows them: every change // writes the candidates' fingerprints again (index.Indexer.ApplyPreview). The checker's bucket is made when it is assigned, -// after the indexer started, and the indexer holds no timer (ADR 0437 §8): it starts with the bucket absent, and the -// checker says each time it wrote a preview (preview.written, handled by Written); the bucket is then watched and settled -// as the others, with no restart. +// after the indexer started, and the indexer holds no timer (ADR 0437 §8) and hears no preview.written (design 47 G A-160): a +// bucket absent at the start is not watched until the indexer starts again. type previewWatch struct { ctx context.Context js bucketOpener @@ -53,16 +42,16 @@ func (w *previewWatch) ensure() error { kv, err := w.js.KeyValue(w.ctx, index.PreviewsBucket) switch { case errors.Is(err, jetstream.ErrBucketNotFound): - w.logger.Printf("the bucket %s does not exist yet: it is watched when the checker says it wrote a preview", index.PreviewsBucket) + w.logger.Printf("the bucket %s does not exist yet: it is watched from the indexer's next start", index.PreviewsBucket) return nil case err != nil: // Refused or not answered (its grant not on the bus yet, issue 1945): said, and opened again at the checker's next preview. - w.logger.Printf("the bucket %s cannot be read (%v): it is watched when the checker says it wrote a preview", index.PreviewsBucket, err) + w.logger.Printf("the bucket %s cannot be read (%v): it is watched from the indexer's next start", index.PreviewsBucket, err) return nil } watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.ix.PutPreview(e.Key(), e.Value(), !opOf(e)) }) if err != nil { - process.NotWatched(w.logger.Printf, index.PreviewsBucket, err, "it is watched when the checker says it wrote a preview") + process.NotWatched(w.logger.Printf, index.PreviewsBucket, err, "it is watched from the indexer's next start") return nil } w.watched = true @@ -73,64 +62,5 @@ func (w *previewWatch) ensure() error { } }) }() - if err := w.ix.RecomputeCandidates(w.ctx); err != nil { - return err - } - return w.ix.WriteTryCandidates(w.ctx) -} - -// Written takes one preview.written event of the checker in: its bucket is watched if it is not. A subject that is not the -// checker's is terminated; a failure stays unacknowledged, and the server delivers it again. -func (w *previewWatch) Written(m delivered) { - if !strings.HasPrefix(m.Subject(), previewWrittenSubject) { - w.logger.Printf("%s: not the checker's preview.written, dropped", m.Subject()) - _ = m.Term() - return - } - if err := w.ensure(); err != nil { - w.logger.Printf("%s: left for redelivery: %v", m.Subject(), err) - return - } - _ = m.Ack() -} - -// tryEventSubject is the subject of an event of the checker's stream checks about a try. -func tryEventSubject(event string) string { - return "mesh.stream." + index.CheckerModule + ".event." + event -} - -// heardChecks takes one event of the checker's stream in: a preview written (Written), or a whole try's start or end, from -// which the candidate of its group is written or deleted (index.Indexer.StartTry, EndTry). A failure stays unacknowledged, and -// the server delivers it again; a body that is no try event is terminated. -func (w *previewWatch) heardChecks(m delivered) { - var event string - switch m.Subject() { - case tryEventSubject(checkermsg.Started): - event = checkermsg.Started - case tryEventSubject(checkermsg.Ended): - event = checkermsg.Ended - default: - w.Written(m) - return - } - e, err := checkermsg.Decode(event, m.Data()) - if err != nil { - w.logger.Printf("%s: not a try event, dropped: %v", m.Subject(), err) - _ = m.Term() - return - } - if event == checkermsg.Started { - var ms []index.TryMember - for _, x := range e.Members { - ms = append(ms, index.TryMember{Repository: x.Repository, Head: x.Head, Trunk: x.Trunk}) - } - err = w.ix.StartTry(w.ctx, e.ID, ms) - } else { - err = w.ix.EndTry(w.ctx, e.ID) - } - if err != nil { - w.logger.Printf("%s of %s: left for redelivery: %v", event, e.ID, err) - return - } - _ = m.Ack() + return w.ix.RecomputeCandidates(w.ctx) } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/runs.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/runs.go new file mode 100644 index 00000000..6c6bdbfb --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/runs.go @@ -0,0 +1,77 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + "git.novox.be/novox/mesh-sdk/go/process" + + "mesh-delivery-indexer/internal/index" +) + +// runTaker is what the indexer does with the runner's events of a run on request (index.Indexer.StartRun, EndRunPipeline). +type runTaker interface { + StartRun(ctx context.Context, e meshpipelinerunner.RunStarted) error + EndRunPipeline(ctx context.Context, e meshpipelinerunner.PipelineEnded) error +} + +// hearRuns hears the runner's run-started and pipeline-ended (design 47 G §64, §190; A-160, A-174), each on its own stream, on +// the indexer's durable: a run's candidate is written on its start and dropped once every member's pipeline has ended. A stream +// the bus does not hold yet is the finding family-not-heard, and no run's candidate is written or dropped meanwhile. +func hearRuns(ctx context.Context, js jetstream.JetStream, cred process.Account, notHeard binder, ix runTaker, logf func(string, ...any)) { + for _, s := range []struct { + stream string + each func(jetstream.Msg) + }{ + {meshpipelinerunner.StreamRunStarted, func(m jetstream.Msg) { heardRunStarted(ctx, ix, m, logf) }}, + {meshpipelinerunner.StreamPipelineEnded, func(m jetstream.Msg) { heardPipelineEnded(ctx, ix, m, logf) }}, + } { + _ = notHeard.Bind(s.stream, "no run on request's candidate is written or dropped until the indexer starts again", func() error { + cc, err := cred.HearSeatEvents(ctx, js, s.stream, s.each, logf) + if err != nil { + return err + } + go func() { <-ctx.Done(); cc.Stop() }() + return nil + }) + } +} + +// heardRunStarted takes one run-started in. A body that is no run-started, or a run that can never be taken in, is said and +// terminated; any other failure stays unacknowledged, and the server delivers it again. +func heardRunStarted(ctx context.Context, ix runTaker, m delivered, logf func(string, ...any)) { + var e meshpipelinerunner.RunStarted + if err := json.Unmarshal(m.Data(), &e); err != nil { + logf("%s: not a run-started, dropped: %v", m.Subject(), err) + _ = m.Term() + return + } + settleRun(m, ix.StartRun(ctx, e), "run-started of "+e.Run, logf) +} + +// heardPipelineEnded takes one pipeline-ended in: of a run's pipeline, its module is kept as ended. +func heardPipelineEnded(ctx context.Context, ix runTaker, m delivered, logf func(string, ...any)) { + var e meshpipelinerunner.PipelineEnded + if err := json.Unmarshal(m.Data(), &e); err != nil { + logf("%s: not a pipeline-ended, dropped: %v", m.Subject(), err) + _ = m.Term() + return + } + settleRun(m, ix.EndRunPipeline(ctx, e), "pipeline-ended of "+e.Module+" in the run "+e.Run, logf) +} + +func settleRun(m delivered, err error, what string, logf func(string, ...any)) { + switch { + case err == nil: + _ = m.Ack() + case errors.Is(err, index.ErrUnreadable): + logf("%s: never taken in, dropped: %v", what, err) + _ = m.Term() + default: + logf("%s: left for redelivery: %v", what, 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 552b9afb..6e2709f1 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go @@ -22,8 +22,6 @@ type takerIn interface { TakeInPendingByHand(ctx context.Context, key, by, why string) index.TakenIn DropPending(ctx context.Context, key, by, why string, abandon bool) index.TakenIn TakeInMerge(ctx context.Context, repository, merge string, seq uint64, by, why string) index.TakenIn - DropTryCandidate(ctx context.Context, id, by, why string) index.TakenIn - RecoverTriesByHand(ctx context.Context, by, why string) index.TakenIn FingerprintsAgain(ctx context.Context, module string) index.TakenIn RecomputeByHand(ctx context.Context, by, why string) index.TakenIn DropCandidateGraph(ctx context.Context, group, heads, by, why string) index.TakenIn @@ -60,12 +58,6 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream a.Module = r.Module } else if a.Fingerprints { asked, r = "computing of fingerprints again", ix.FingerprintsAgain(ctx, a.Module) - } else if a.RecoverTries { - asked, r = "recovery of the tries", ix.RecoverTriesByHand(ctx, a.By, a.Why) - a.Module = "tries" - } else if a.TryCandidate != "" && a.Drop { - asked, r = "drop of a try's candidate", ix.DropTryCandidate(ctx, a.TryCandidate, a.By, a.Why) - a.Module = a.TryCandidate } 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 diff --git a/modules/mesh-delivery-indexer/go.mod b/modules/mesh-delivery-indexer/go.mod index 9c4b866d..70c6453e 100644 --- a/modules/mesh-delivery-indexer/go.mod +++ b/modules/mesh-delivery-indexer/go.mod @@ -15,6 +15,7 @@ require ( git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdeliverychecker v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/npmpackageregistry v0.0.0-00010101000000-000000000000 @@ -41,7 +42,6 @@ require ( replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun => ../mesh-delivery-builder/buildrun require ( - git.novox.be/novox/mesh-delivery/modules/mesh-delivery-checker/checkermsg v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk v0.0.0-00010101000000-000000000000 @@ -53,8 +53,6 @@ replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk => ../mesh-de replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein => ./takein -replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-checker/checkermsg => ../mesh-delivery-checker/checkermsg - replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg => ../mesh-delivery-forge-bridge/forgemsg replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds => ../mesh-delivery-builder/builds diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index cce9eba5..d5bf7ef4 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -15,7 +15,6 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-checker/checkermsg" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" @@ -215,14 +214,10 @@ 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 - // OpenTries lists the tries that wait or run, from the checker's tool (checker_tries); nil recovers none (trycandidates.go). - OpenTries func(ctx context.Context) ([]checkermsg.Open, error) - // TryRecovery keeps how the last recovery of the open tries went (trycandidates.go); nil keeps none. - TryRecovery Store - tryMu sync.Mutex // one try event or recovery at a time (trycandidates.go) - recoveryFailed atomic.Bool // the last recovery of the open tries failed (trycandidates.go) - tries map[string][]TryMember // the whole tries that wait or run, by try id (trycandidates.go); guarded by mu - pendingMu sync.Mutex // one take-in step at a time: a merge, a group's end, the start and a hand act (pending.go) + // 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) + pendingMu sync.Mutex // one take-in step at a time: a merge, a group's end, the start and a hand act (pending.go) } func (ix *Indexer) logf(format string, args ...any) { @@ -384,7 +379,7 @@ func (ix *Indexer) applyOpen(ctx context.Context, f forgemsg.Forge) error { } for _, key := range keys { if meshcatalogclient.IsTryGroup(key) { - continue // a try's candidate is no pull request's: the checker's try owns it (trycandidates.go) + continue // a run's candidate is no pull request's: its run owns it (runcandidates.go) } err := write(ctx, ix.Store, key, func(value []byte, found bool) ([]byte, bool, error) { if !found { diff --git a/modules/mesh-delivery-indexer/internal/index/control_records.go b/modules/mesh-delivery-indexer/internal/index/control_records.go index c13e960b..b5688b42 100644 --- a/modules/mesh-delivery-indexer/internal/index/control_records.go +++ b/modules/mesh-delivery-indexer/internal/index/control_records.go @@ -237,7 +237,7 @@ func (ix *Indexer) configCandidatesAfterMerge(ctx context.Context, f forgemsg.Fo var errs []error for _, g := range groups { if meshcatalogclient.IsTryGroup(g) { - continue // a try's candidate holds no pull request's config declarations (trycandidates.go) + continue // a run's candidate holds no pull request's config declarations (runcandidates.go) } raw, _, found, err := ix.Store.Get(ctx, g) if err != nil || !found { diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index b7261c76..a732a6ff 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", "OpenTries", "TryRecovery", "OwnBuildState", "Heard", "Placement", "RunningWhole", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "NamedSets", "Registry", "SeedModules", "Versions", "SayVersionReleased", "SayVersionsFilled", "Inputs", "BuildEdges"} { + "Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "Runs", "OwnBuildState", "Heard", "Placement", "RunningWhole", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "NamedSets", "Registry", "SeedModules", "Versions", "SayVersionReleased", "SayVersionsFilled", "Inputs", "BuildEdges"} { named[f] = true } ty := reflect.TypeOf(Indexer{}) diff --git a/modules/mesh-delivery-indexer/internal/index/pending_tools.go b/modules/mesh-delivery-indexer/internal/index/pending_tools.go index 5557a43a..366654eb 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_tools.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_tools.go @@ -4,12 +4,8 @@ import ( "context" "encoding/json" "fmt" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" - "strings" "time" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" stdio "git.novox.be/novox/mesh-sdk/go" ) @@ -115,94 +111,3 @@ func PendingTools(pending, ownBuild Store, ask func(key string, value []byte) (T Run: merge}, } } - -// TryCandidateTools are the tools of a group try's candidate (issue 1921, trycandidates.go): indexer_try_candidates reads the -// candidates the indexer holds for tries (the try, its heads, each member's modules and whether each has its fingerprint), -// and indexer_drop_try_candidate drops one by an ask in takeins. candidates is the candidates state as the tools bundle reads it. -func TryCandidateTools(candidates, recovery Store, ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time) []stdio.Tool { - return []stdio.Tool{ - {Name: TryCandidatesTool, - Description: "The candidates the indexer holds for group tries (a try of several repositories at exact commits, mesh-delivery-checker try): " + - "per try its heads hash, each member's repository and commit, the modules the try touches and, per module, its manifest digest, its " + - "fingerprint or why it has none yet (a merge preview not yet written). The checker's manifest validator and the node-engines' " + - "compose-candidate read this candidate. Read-only; an empty answer means no whole try waits or runs.", - Input: map[string]any{"type": "object", "properties": map[string]any{}}, - Run: func(map[string]any) (any, error) { - ctx := context.Background() - keys, err := candidates.Keys(ctx) - if err != nil { - return nil, err - } - var out []meshcatalog.CandidatesByGroupValue - for _, k := range keys { - if !meshcatalogclient.IsTryGroup(k) { - continue - } - v, _, found, err := candidates.Get(ctx, k) - if err != nil || !found { - continue - } - if c, err := DecodeCandidate(v); err == nil { - out = append(out, c) - } - } - // How the recovery of the open tries went: loud when it is not done (since when, why). - said := map[string]any{"done": false, "why": "no recovery has been written yet"} - if v, _, found, err := recovery.Get(ctx, TryRecoveryKey); err == nil && found { - var r TryRecovery - if json.Unmarshal(v, &r) == nil { - said = map[string]any{"done": r.Done} - if !r.Done { - said["why"], said["since"] = r.Why, r.Since - said["says"] = "recovery not done, since " + r.Since.Format(time.RFC3339) + ": " + r.Why - } - } - } - return map[string]any{"tries": out, "recovery": said}, nil - }}, - {Name: RecoverTriesTool, - Description: "Run the recovery of the open tries now: the indexer lists them through the checker's checker_tries, writes the candidate of each " + - "whole try that waits or runs and deletes the candidate of a try no longer listed. It runs by itself at the indexer's start; when it failed " + - "(indexer_try_candidates says recovery not done, since when and why) this runs it again. A hand act: who and why are said in the journal.", - Input: map[string]any{"type": "object", "required": []string{"why"}, "properties": map[string]any{ - "why": map[string]any{"type": "string", "description": "why it is run now, in plain words"}}}, - Run: func(args map[string]any) (any, error) { - why, _ := args["why"].(string) - if strings.TrimSpace(why) == "" { - return nil, fmt.Errorf("say why: a hand act is said in the journal") - } - data, err := json.Marshal(TakeInAsk{V: 1, RecoverTries: true, Why: strings.TrimSpace(why), By: stdio.CallerOf(args), At: now().UTC()}) - if err != nil { - return nil, err - } - a, err := ask("tries.recover."+now().UTC().Format("20060102T150405.000000000"), data) - if err != nil { - return nil, err - } - return a.Result, nil - }}, - {Name: DropTryCandidateTool, - Description: "Drop a group try's candidate by hand (the indexer deletes it itself when the try ends): for a candidate left for a try that " + - "is gone, or to make the indexer write it again from the previews at the try's next change. A hand act: who and why are said in the journal.", - Input: map[string]any{"type": "object", "required": []string{"try", "why"}, "properties": map[string]any{ - "try": map[string]any{"type": "string", "description": "the try's id, try-"}, - "why": map[string]any{"type": "string", "description": "why it is dropped, in plain words"}}}, - Run: func(args map[string]any) (any, error) { - id, _ := args["try"].(string) - why, _ := args["why"].(string) - id = strings.TrimSpace(id) - if !meshcatalogclient.IsTryGroup(id) || strings.TrimSpace(why) == "" { - return nil, fmt.Errorf("name the try (try-) and say why") - } - data, err := json.Marshal(TakeInAsk{V: 1, TryCandidate: id, Drop: true, Why: strings.TrimSpace(why), By: stdio.CallerOf(args), At: now().UTC()}) - if err != nil { - return nil, err - } - a, err := ask("try-candidate.drop."+id, data) - if err != nil { - return nil, err - } - return a.Result, nil - }}, - } -} diff --git a/modules/mesh-delivery-indexer/internal/index/previews.go b/modules/mesh-delivery-indexer/internal/index/previews.go index 14d0fd53..172f327b 100644 --- a/modules/mesh-delivery-indexer/internal/index/previews.go +++ b/modules/mesh-delivery-indexer/internal/index/previews.go @@ -77,10 +77,7 @@ func (ix *Indexer) ApplyPreview(ctx context.Context, key string, value []byte, d if !ix.PutPreview(key, value, deleted) { return nil } - if err := ix.RecomputeCandidates(ctx); err != nil { - return err - } - return ix.WriteTryCandidates(ctx) // a try that waited for this preview gets its candidate + return ix.RecomputeCandidates(ctx) } // RecomputeCandidates writes again every candidate whose pull requests' fingerprints, from the previews held, are no longer diff --git a/modules/mesh-delivery-indexer/internal/index/previews_test.go b/modules/mesh-delivery-indexer/internal/index/previews_test.go index 03766078..8f1b3a5c 100644 --- a/modules/mesh-delivery-indexer/internal/index/previews_test.go +++ b/modules/mesh-delivery-indexer/internal/index/previews_test.go @@ -7,15 +7,11 @@ import ( "fmt" "maps" "os" - "runtime" - "slices" "strings" "testing" - "time" sdkstep "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-checker/checkermsg" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdeliverychecker" @@ -229,221 +225,6 @@ func TestAPreviewHeardBeforeItsPullRequestGivesTheFingerprintWhenTheCandidateIsW if got := fpOf(t, cands); got.Fp2 != "" || got.None2 != meshcatalogclient.NoneNoPreview { t.Errorf("a new head: %+v, want no preview", got) } - // Issue 1921: a whole group try has no pull request, yet the indexer writes its candidate from the checker's try events and - // the previews its fetch wrote (number 0, the heads hash of its members), keeps it apart from the pull requests' candidates, - // and deletes it when the try ends. Its manifests are judged with the seat definitions its members carry, in one locked - // pass, whether the contracts module's manifest changed or only its seat.json did (the wait is bounded: a lock held twice - // would show as a failure here, not as a stuck run). - for _, row := range []struct { - name string - unchanged bool // the contracts module's module.json is the trunk's: only its definitions changed - }{ - {name: "a changed manifest that carries definitions"}, - {name: "an unchanged manifest that carries definitions", unchanged: true}, - } { - t.Run("a group try's candidate: "+row.name, func(t *testing.T) { - ix, mods, cands := previewBed(t, true) - ctx := context.Background() - ix.Verdicts = newMem() - var changed map[string]any - if err := json.Unmarshal([]byte(catalogueManifest(t, "avahi")), &changed); err != nil { - t.Fatal(err) - } - changed["build"] = map[string]any{"artifacts": []any{map[string]any{"name": "go-seats", "kind": "package", "language": "go", "paths": []any{"go"}, "generated-from": "defs"}}} - raw, _ := json.Marshal(changed) - pv := avahiPreview(t, trunkNow, h1) - pm := meshdeliverychecker.PreviewsByRepositoryHeadTrunkValueModulesValue{Tree: "tree-avahi", Manifest: meshcatalogclient.ManifestDigest(raw), Locks: map[string]string{"go.sum": avahiGoSum}} - if row.unchanged { - m, found := moduleOf(t, mods, "avahi") - _ = found - pm.Manifest = m.Manifest - key, _ := meshcatalogclient.ManifestsKey(m.Manifest) - data, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: "avahi", Manifest: string(raw)}) - if _, _, _, err := ix.Manifests.Get(ctx, key); err != nil { - t.Fatal(err) - } - _ = ix.Manifests.(*memStore).Update(ctx, key, data, ix.Manifests.(*memStore).revs[key]) - } - pv.Modules["modules/avahi"] = pm - ix.Read = func(_ context.Context, _, _, dir string) (RepoRead, error) { - return RepoRead{Manifests: map[string]string{dir: string(raw)}}, nil - } - listed := 0 - ix.Files = func(_ context.Context, _, commit string, dirs []string) (map[string]map[string]string, error) { - listed++ - if commit != h1 || !slices.Equal(dirs, []string{"modules/avahi/defs/seats"}) { - t.Errorf("listed %v at %s, want the member's head", dirs, commit) - } - return map[string]map[string]string{dirs[0]: {"node-engine/seat.json": "1"}}, nil - } - ix.ReadFile = func(_ context.Context, _, _, p string) ([]byte, error) { - switch p { - case "modules/avahi/defs/seats/node-engine/seat.json": - return []byte(`{"format":1,"seat":"node-engine","scope":"node","grain":"seat-list","decision":"d","domain":"Core","verbs":[{"name":"try-verb"}]}`), nil - } - return nil, nil - } - started := `{"id":"try-7"}` - _ = started - members := []TryMember{{Repository: catalogueRepo, Head: h1, Trunk: trunkNow}} - held := func() []string { k, _ := cands.Keys(ctx); return k } - if err := ix.StartTry(ctx, "try-7", members); err != nil || len(held()) != 0 { - t.Fatalf("a try before its preview: %v, candidates %v", err, held()) - } - done := make(chan struct{}) - go func() { defer close(done); applyPreview(t, ix, pv) }() - select { - case <-done: - case <-time.After(10 * time.Second): - panic("writing the try's candidate hangs: the seat lock is held twice (a locked pass cannot be left by t.Fatal: the lock stays held)") - } - if listed == 0 { - t.Error("the member's definitions were not read") - } - key, _ := meshcatalogclient.CandidatesKey("try-7") - rawc, _, found, _ := cands.Get(ctx, key) - if !found { - t.Fatalf("no candidate for the try after its preview: %v", held()) - } - c, err := DecodeCandidate(rawc) - pull := meshcatalog.CandidatesByGroupValuePullsItem{} - if len(c.Pulls) == 1 { - pull = c.Pulls[0] - } - if row.unchanged { - if err != nil || c.Group != "try-7" || len(pull.Modules) != 0 { - t.Fatalf("candidate = %+v, %v, want the try's group with no module touched", c, err) - } - } else if err != nil || c.Group != "try-7" || pull.Number != 0 || pull.Head != h1 || pull.Fps["avahi"].Manifest == "" { // no fingerprint asserted: this mesh-sdk pin does not read generated-from yet - t.Fatalf("candidate = %+v, %v, want one pull of number 0 with avahi's manifest", c, err) - } - if err := ix.Apply(ctx, forgemsg.Forge{Kind: forgemsg.ForgeOpen}); err != nil || len(held()) != 1 { - t.Errorf("the open message left %v (%v), want the try's candidate alone", held(), err) - } - // The tools: one reads the candidates held for tries, one drops by an ask the indexer answers. - asks := func(key string, value []byte) (TakeInAsk, error) { - a, err := DecodeTakeIn(value) - r := ix.DropTryCandidate(ctx, a.TryCandidate, a.By, a.Why) - a.Result = &r - return a, err - } - recovery := newMem() - ix.TryRecovery = recovery - tools := TryCandidateTools(cands, recovery, asks, time.Now) - if out, err := tools[0].Run(map[string]any{}); err != nil || len(out.(map[string]any)["tries"].([]meshcatalog.CandidatesByGroupValue)) != 1 { - t.Errorf("%s = %v, %v, want the try's candidate", tools[0].Name, out, err) - } - if _, err := tools[2].Run(map[string]any{"try": "feat/x", "why": "w"}); err == nil { - t.Error("a branch was dropped as a try") - } - if out, err := tools[2].Run(map[string]any{"try": "try-7", "why": "by hand"}); err != nil || !out.(*TakenIn).Stored || len(held()) != 0 { - t.Errorf("drop = %v, %v, candidates %v", out, err, held()) - } - // The start recovered from the checker's tool: an open whole try is written again, one the checker no longer - // lists is deleted, and a try narrowed to a level is none. - ix.OpenTries = func(context.Context) ([]checkermsg.Open, error) { - return []checkermsg.Open{{ID: "try-7", Whole: true, Members: []checkermsg.Member{{Repository: catalogueRepo, Head: h1, Trunk: trunkNow}}}, - {ID: "try-8", Whole: false, Members: []checkermsg.Member{{Repository: catalogueRepo, Head: h2, Trunk: trunkNow}}}}, nil - } - if err := ix.RecoverTries(ctx); err != nil || len(held()) != 1 { - t.Fatalf("recovered: %v, candidates %v, want try-7's alone", err, held()) - } - ix.OpenTries = func(context.Context) ([]checkermsg.Open, error) { return nil, nil } - if err := ix.RecoverTries(ctx); err != nil || len(held()) != 0 { - t.Errorf("recovered with no open try: %v, candidates %v, want none", err, held()) - } - // A recovery whose call fails is loud and deletes nothing: the candidate stays, indexer_try_candidates says recovery not - // done since when and why, and the next try event runs the recovery again, which clears it; the tool runs it by hand. - if err := ix.StartTry(ctx, "try-7", members); err != nil || len(held()) != 1 { - t.Fatalf("started: %v, candidates %v", err, held()) - } - ix.OpenTries = func(context.Context) ([]checkermsg.Open, error) { - return nil, errors.New("checker_tries does not answer") - } - if err := ix.RecoverTries(ctx); err == nil || len(held()) != 1 { - t.Fatalf("a failed call: %v, candidates %v, want the error and the candidate kept", err, held()) - } - if out, err := tools[0].Run(map[string]any{}); err != nil || !strings.Contains(out.(map[string]any)["recovery"].(map[string]any)["says"].(string), "recovery not done, since") || - !strings.Contains(out.(map[string]any)["recovery"].(map[string]any)["says"].(string), "checker_tries does not answer") { - t.Errorf("%s = %v, %v, want recovery not done with since and why", tools[0].Name, out, err) - } - ix.OpenTries = func(context.Context) ([]checkermsg.Open, error) { return nil, nil } - if err := ix.StartTry(ctx, "try-11", members); err != nil || len(held()) != 1 || held()[0] != "try-11" { - t.Fatalf("a try event after a failed recovery: %v, candidates %v, want the recovery run (try-7 not listed, gone) and try-11 written", err, held()) - } - if out, _ := tools[0].Run(map[string]any{}); out.(map[string]any)["recovery"].(map[string]any)["done"] != true { - t.Errorf("recovery still said not done after a success: %v", out) - } - ix.OpenTries = func(context.Context) ([]checkermsg.Open, error) { return nil, errors.New("down") } - recoverAsk := func(key string, value []byte) (TakeInAsk, error) { - a, err := DecodeTakeIn(value) - r := ix.RecoverTriesByHand(ctx, a.By, a.Why) - a.Result = &r - return a, err - } - byHand := TryCandidateTools(cands, recovery, recoverAsk, time.Now) - if out, err := byHand[1].Run(map[string]any{"why": "the checker was replaced"}); err != nil || out.(*TakenIn).Stored || !strings.Contains(out.(*TakenIn).Answer, "recovery not done") { - t.Errorf("%s with the checker down = %v, %v", byHand[1].Name, out, err) - } - ix.OpenTries = func(context.Context) ([]checkermsg.Open, error) { return nil, nil } - if out, err := byHand[1].Run(map[string]any{"why": "the checker is back"}); err != nil || !out.(*TakenIn).Stored || len(held()) != 0 { - t.Errorf("%s = %v, %v, candidates %v, want it recovered (try-11 not listed)", byHand[1].Name, out, err, held()) - } - if err := ix.StartTry(ctx, "try-7", members); err != nil { - t.Fatal(err) - } - // ADR 0638 §3, §4: the consumer exists before the call, so what happens between the call and the hearing is heard after - // it. A try the call listed and whose start is then heard gets one candidate; a try started after the list gets one; a - // try that ended during the call, listed as open, leaves none once its end is heard. - ix.OpenTries = func(context.Context) ([]checkermsg.Open, error) { - return []checkermsg.Open{{ID: "try-7", Whole: true, Members: []checkermsg.Member{{Repository: catalogueRepo, Head: h1, Trunk: trunkNow}}}, - {ID: "try-9", Whole: true, Members: []checkermsg.Member{{Repository: catalogueRepo, Head: h1, Trunk: trunkNow}}}}, nil - } - if err := ix.RecoverTries(ctx); err != nil { - t.Fatal(err) - } - for _, step := range []struct { - what string - do func() error - want int - }{ - {"the start heard after the call, of a try the call listed", func() error { return ix.StartTry(ctx, "try-7", members) }, 2}, - {"a start after the list", func() error { return ix.StartTry(ctx, "try-10", members) }, 3}, - {"the end heard of a try ended during the call", func() error { return ix.EndTry(ctx, "try-9") }, 2}, - } { - if err := step.do(); err != nil || len(held()) != step.want { - t.Fatalf("%s: %v, candidates %v, want %d", step.what, err, held(), step.want) - } - } - for _, id := range []string{"try-10"} { - if err := ix.EndTry(ctx, id); err != nil { - t.Fatal(err) - } - } - if err := ix.StartTry(ctx, "try-7", members); err != nil || len(held()) != 1 { - t.Fatalf("started again: %v, candidates %v", err, held()) - } - if err := ix.EndTry(ctx, "try-7"); err != nil || len(held()) != 0 { - t.Errorf("an ended try: %v, candidates %v, want none", err, held()) - } - // The hand recovery never overlaps a try event: a start heard while the call to the checker is out writes its candidate - // before or after the sweep, never between the call and the sweep (which would delete it, the try not being listed). - startDone := make(chan error, 1) - ix.OpenTries = func(context.Context) ([]checkermsg.Open, error) { - go func() { startDone <- ix.StartTry(ctx, "try-12", members) }() - for i := 0; i < 2000 && len(held()) == 0; i++ { - runtime.Gosched() // lets the start write if nothing holds it back - } - return nil, nil - } - if r := ix.RecoverTriesByHand(ctx, "lead", "a test"); !r.Stored { - t.Fatalf("the hand recovery = %+v", r) - } - if err := <-startDone; err != nil || len(held()) != 1 || held()[0] != "try-12" { - t.Errorf("a start heard during the hand recovery: %v, candidates %v, want try-12 kept", err, held()) - } - }) - } // Issue 1921: a head whose tree lacks the directory asked is asked once, however many passes read it, until the head changes; // any other failure (the forge does not answer) is asked again at every pass and said once per head. for _, row := range []struct { diff --git a/modules/mesh-delivery-indexer/internal/index/runcandidates.go b/modules/mesh-delivery-indexer/internal/index/runcandidates.go new file mode 100644 index 00000000..8a66b035 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/runcandidates.go @@ -0,0 +1,296 @@ +package index + +// **A run on request's candidate** (design 47 G §64, §190, §321; A-160, A-174): the runner says `run-started` with the run's +// members, each a module at a named commit on a branch; the indexer builds the run's candidate from it, `candidates.run-`, +// and drops it once every member's pipeline of that run has ended (`pipeline-ended` naming the run). Each member maps to its +// repository and directory by its registration, and its manifest and fingerprint come from the same take-in the indexer +// computes for a watched branch (its files read through the seat git), over the revision view of the member's branch when its +// registration watches that branch, else over the trunk's (A-93). No checker preview is heard and no checker tool is asked. +// +// What the indexer knows of each run is kept in its state runs (key run-: the members and the modules whose pipeline +// ended), written on run-started and on each of the run's pipeline-ended and removed with the candidate, so a restart between +// two members' ends loses nothing (A-174). A pipeline-ended heard before its run-started is kept all the same: the run-started +// that follows finds it. + +import ( + "context" + "encoding/json" + "fmt" + "slices" + "strings" + "time" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/trunk" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// RunsState is the indexer's state of the runs on request it holds a candidate for (A-174): the bucket +// mesh-delivery-indexer_runs, one key per run, run-. +const ( + RunsState = "runs" + RunsBucket = Module + "_" + RunsState +) + +// Run is one value of the state runs: the run's members as run-started named them, and the modules whose pipeline of the +// run has ended. +type Run struct { + V int `json:"v"` + Run string `json:"run"` + Members []meshpipelinerunner.Member `json:"members,omitempty"` + Ended []string `json:"ended,omitempty"` + At time.Time `json:"at"` +} + +// RunGroup is the candidate's group, and the runs key, of a run: run- (A-160). +func RunGroup(run string) (string, error) { + g := "run-" + strings.TrimSpace(run) + if strings.TrimSpace(run) == "" || !meshcatalogclient.IsTryGroup(g) { + return "", fmt.Errorf("%w: %q is no run's id", ErrUnreadable, run) + } + return g, nil +} + +// done says every member's pipeline of the run has ended; a run whose run-started is not heard yet is not done. +func (r Run) done() bool { + if len(r.Members) == 0 { + return false + } + for _, m := range r.Members { + if !slices.Contains(r.Ended, m.Module) { + return false + } + } + return true +} + +// StartRun takes a run on request in, as the runner said it started: its members are kept in the state runs and its +// candidate is written; a run every member of which has ended already (its ends heard first) is dropped instead. +func (ix *Indexer) StartRun(ctx context.Context, e meshpipelinerunner.RunStarted) error { + group, err := RunGroup(e.Run) + if err != nil { + return err + } + if len(e.Members) == 0 { + return fmt.Errorf("%w: the run %s names no member", ErrUnreadable, e.Run) + } + ix.runMu.Lock() + defer ix.runMu.Unlock() + r, err := ix.updateRun(ctx, group, func(r *Run) { r.Members = slices.Clone(e.Members) }) + if err != nil { + return err + } + if r.done() { + return ix.dropRun(ctx, group) + } + return ix.writeRunCandidate(ctx, group, r.Members) +} + +// EndRunPipeline takes one pipeline-ended in: of a run's pipeline, its module is kept as ended, and once every member's has +// ended the candidate and the run's key are removed. A pipeline of no run is none of this. +func (ix *Indexer) EndRunPipeline(ctx context.Context, e meshpipelinerunner.PipelineEnded) error { + if e.Run == "" { + return nil + } + group, err := RunGroup(e.Run) + if err != nil { + return err + } + ix.runMu.Lock() + defer ix.runMu.Unlock() + r, err := ix.updateRun(ctx, group, func(r *Run) { + if e.Module != "" && !slices.Contains(r.Ended, e.Module) { + r.Ended = append(r.Ended, e.Module) + slices.Sort(r.Ended) + } + }) + if err != nil { + return err + } + if !r.done() { + return nil + } + return ix.dropRun(ctx, group) +} + +// updateRun reads the run's key, changes it and writes it back, made when absent. +func (ix *Indexer) updateRun(ctx context.Context, group string, change func(*Run)) (Run, error) { + if ix.Runs == nil { + return Run{}, fmt.Errorf("the state %s is not bound, so the run %s is not kept", RunsState, group) + } + var out Run + err := write(ctx, ix.Runs, group, func(value []byte, found bool) ([]byte, bool, error) { + r := Run{V: 1, Run: strings.TrimPrefix(group, "run-")} + if found { + if err := json.Unmarshal(value, &r); err != nil { + return nil, false, fmt.Errorf("%s.%s holds what is not a run: %w", RunsState, group, err) + } + } + change(&r) + r.At = ix.now().UTC() + out = r + data, err := json.Marshal(r) + return data, false, err + }) + return out, err +} + +// dropRun removes the run's candidate, then its key. +func (ix *Indexer) dropRun(ctx context.Context, group string) error { + if ix.Store != nil { + key, err := meshcatalogclient.CandidatesKey(group) + if err != nil { + return err + } + if err := write(ctx, ix.Store, key, func(_ []byte, found bool) ([]byte, bool, error) { return nil, found, nil }); err != nil { + return fmt.Errorf("the candidate of %s is not dropped: %w", group, err) + } + } + if err := write(ctx, ix.Runs, group, func(_ []byte, found bool) ([]byte, bool, error) { return nil, found, nil }); err != nil { + return fmt.Errorf("the run %s is not removed: %w", group, err) + } + ix.logf("the run %s: every member's pipeline ended, its candidate is dropped", group) + return nil +} + +// RunBase is the branch whose revision view a member's take-in starts from (A-174): the member's own branch when its +// registration watches it, else the trunk. +func RunBase(reg meshcatalog.Registration, branch string) string { + if branch != "" && slices.Contains(reg.Branches, branch) { + return branch + } + return TrunkBranch +} + +// runMember is one member placed: its registration's repository and directory, and the branch its take-in starts from. +type runMember struct { + meshpipelinerunner.Member + repository, dir, base string +} + +// writeRunCandidate writes the run's candidate: one pull of number 0 per member, its module's manifest and fingerprint as +// the take-in of its commit over its base's revision view computes them, its manifests judged with the seat definitions +// the members carry, as a revision's are. +func (ix *Indexer) writeRunCandidate(ctx context.Context, group string, members []meshpipelinerunner.Member) error { + if ix.Store == nil || ix.Read == nil { + return nil + } + if ix.Registry == nil || ix.Registry.Registrations == nil || ix.Revisions == nil { + return fmt.Errorf("the run %s: the indexer holds no registrations or no revisions, so its members cannot be taken in", group) + } + var placed []runMember + for _, m := range members { + reg, found, err := ix.Registry.Registration(ctx, m.Module) + if err != nil { + return fmt.Errorf("the run %s: the registration of %s: %w", group, m.Module, err) + } + if !found { + return fmt.Errorf("%w: the run %s: %s has no registration, so its repository is not known", ErrUnreadable, group, m.Module) + } + dir := strings.Trim(reg.Path, "/") + if dir == meshcatalogclient.RootPath { + dir = "" + } + placed = append(placed, runMember{Member: m, repository: trunk.RepositoryName(reg.Origin), dir: dir, base: RunBase(reg, m.Branch)}) + } + slices.SortFunc(placed, func(a, b runMember) int { + return strings.Compare(a.repository+"\x00"+a.Module, b.repository+"\x00"+b.Module) + }) + heads := make([]meshdeliveryclient.Head, len(placed)) + for i, m := range placed { + heads[i] = meshdeliveryclient.Head{Repository: m.repository, Head: m.Commit} + } + hash, err := meshdeliveryclient.HeadsHash(heads) + if err != nil { + return err + } + pulls := make([]meshcatalog.CandidatesByGroupValuePullsItem, len(placed)) + var entries []PendingEntry + over := map[string]string{} + // One take-in per base branch: the members on one base are taken in together, so one member sees another's. + bases := map[string]*dryRun{} + for i, m := range placed { + d, ok := bases[m.base] + if !ok { + view, nodes, _, err := BranchView(ctx, ix.Revisions, m.base) + if err != nil { + return fmt.Errorf("the run %s: the revision view of %s: %w", group, m.base, err) + } + d = ix.dry() + d.ix.Modules, d.ix.Graph, d.ix.Taken = view, nodes, newOverlay(nil) + d.ix.Logf = func(format string, args ...any) { + ix.logf("the run "+group+", computed and not written: "+format, args...) + } + bases[m.base] = d + } + read, err := ix.Read(ctx, m.repository, m.Commit, m.dir) + if err != nil { + return fmt.Errorf("the run %s: reading %s of %s at %.12s: %w", group, m.Module, m.repository, m.Commit, err) + } + raw := read.Manifests[m.dir] + if raw == "" { + return fmt.Errorf("%w: the run %s: %s at %.12s holds no manifest at %q", ErrUnreadable, group, m.repository, m.Commit, m.dir) + } + if name := manifestModule(raw); name != m.Module { + return fmt.Errorf("%w: the run %s: the manifest at %q of %s at %.12s names %q, not %s", ErrUnreadable, group, m.dir, m.repository, m.Commit, name, m.Module) + } + f := forgemsg.Forge{V: keys.Version, Kind: forgemsg.ForgeMerged, Repository: m.repository, Merge: m.Commit, Root: read.Root, + Trees: map[string]string{m.dir: read.Trees[m.dir]}, Manifests: map[string]string{m.dir: raw}} + if l := read.Locks[m.dir]; len(l) > 0 { + f.Locks = map[string]map[string]string{m.dir: l} + } + if p := read.Packages[m.dir]; len(p) > 0 { + f.Packages = map[string][]string{m.dir: p} + } + if _, err := d.ix.applyManifests(ctx, f, false); err != nil { + return fmt.Errorf("the run %s: taking %s in: %w", group, m.Module, err) + } + key, err := meshcatalogclient.ModulesKey(m.Module) + if err != nil { + return err + } + value, _, found, err := d.ix.Modules.Get(ctx, key) + if err != nil { + return err + } + if !found { + return fmt.Errorf("the run %s: the take-in of %s at %.12s wrote no module (the journal says why)", group, m.Module, m.Commit) + } + mv, err := meshcatalogclient.DecodeModule(value) + if err != nil { + return err + } + pulls[i] = meshcatalog.CandidatesByGroupValuePullsItem{Repository: m.repository, Head: m.Commit, Modules: []string{m.Module}, + Fps: map[string]meshcatalog.CandidatesByGroupValuePullsItemFpsValue{m.Module: {Manifest: mv.Manifest, Fp2: mv.Fp2, None2: mv.None2, + Packages: mv.Packages}}} + over[m.Module] = mv.Manifest + entries = append(entries, PendingEntry{Forge: forgemsg.Forge{Repository: m.repository, Merge: m.Commit, Manifests: map[string]string{m.dir: raw}}}) + } + seats, err := ix.revisionSeats(ctx, entries) + if err != nil { + return fmt.Errorf("the run %s: reading the seat definitions of its members: %w", group, err) + } + ix.judgeDigestsWith(ctx, seats, over, "the run "+group) + c := meshcatalog.CandidatesByGroupValue{V: keys.Version, Group: group, Heads: hash, Pulls: pulls} + c.Fpset = meshcatalogclient.FingerprintSet(c.Pulls) + if err := ix.refuseDisagreement(c); err != nil { + return err + } + data, err := EncodeCandidate(c) + if err != nil { + return err + } + key, err := meshcatalogclient.CandidatesKey(group) + if err != nil { + return err + } + if err := write(ctx, ix.Store, key, func(_ []byte, _ bool) ([]byte, bool, error) { return data, false, nil }); err != nil { + return err + } + ix.logf("the run %s: its candidate of %d members is written", group, len(placed)) + return nil +} diff --git a/modules/mesh-delivery-indexer/internal/index/runcandidates_test.go b/modules/mesh-delivery-indexer/internal/index/runcandidates_test.go new file mode 100644 index 00000000..dfd17632 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/runcandidates_test.go @@ -0,0 +1,130 @@ +package index + +import ( + "context" + "encoding/json" + "errors" + "testing" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" +) + +// Design 47 G §64, §190, A-160, A-174: a run on request's candidate is built from the runner's run-started, each member taken +// in from its registration's repository at its commit (read through the seat git), and dropped once every member's pipeline of +// the run has ended; what the indexer knows of the run is kept in its state runs, so a restart between two ends loses nothing. +func TestARunsCandidateIsBuiltOnItsStartAndDroppedWhenEveryMembersPipelineEnded(t *testing.T) { + const c1, c2 = "1111111111111111111111111111111111111111", "2222222222222222222222222222222222222222" + bed := func(t *testing.T) (*Indexer, *memStore, *memStore, *int) { + t.Helper() + ix, _, _ := manifestsIndexer(t) + cands, runs := newMem(), newMem() + ix.Store, ix.Runs, ix.Revisions = cands, runs, newMemObjects(0) + ix.Registry = &Registry{Registrations: MemoryStore()} + for _, r := range []meshcatalog.Registration{ + {Module: "avahi", Origin: catalogueRepo, Path: "modules/avahi", Branches: []string{"main"}}, + {Module: "netcheck", Origin: "novox/mesh-networking", Path: "modules/netcheck", Branches: []string{"main", "beta"}}, + } { + data, _ := json.Marshal(r) + if err := ix.Registry.Registrations.Create(context.Background(), RegistrationKey(r.Module), data); err != nil { + t.Fatal(err) + } + } + reads := 0 + ix.Read = func(_ context.Context, repository, commit, dir string) (RepoRead, error) { + reads++ + module := map[string]string{"modules/avahi": "avahi", "modules/netcheck": "netcheck"}[dir] + return RepoRead{Root: "r", Trees: map[string]string{dir: "tree-" + commit[:4]}, Manifests: map[string]string{dir: catalogueManifest(t, module)}, + Locks: map[string]map[string]string{dir: {"go.sum": avahiGoSum}}}, nil + } + return ix, cands, runs, &reads + } + ctx := context.Background() + started := meshpipelinerunner.RunStarted{V: 1, Run: "7", By: "lead@shanks", Members: []meshpipelinerunner.Member{ + {Module: "avahi", Commit: c1, Branch: "feature/1966-x"}, {Module: "netcheck", Commit: c2, Branch: "beta"}}} + ended := func(module string) meshpipelinerunner.PipelineEnded { + return meshpipelinerunner.PipelineEnded{V: 1, Pipeline: "p", Module: module, Branch: "b", Commit: "c", Outcome: "passed", Run: "7"} + } + key, _ := meshcatalogclient.CandidatesKey("run-7") + t.Run("built on its start, one pull per member, and dropped with the run after a restart", func(t *testing.T) { + ix, cands, runs, reads := bed(t) + if err := ix.StartRun(ctx, started); err != nil { + t.Fatal(err) + } + raw, _, found, _ := cands.Get(ctx, key) + c, err := DecodeCandidate(raw) + if !found || err != nil || c.Group != "run-7" || len(c.Pulls) != 2 || *reads != 2 { + t.Fatalf("candidate %+v (%v, found %v, %d reads), want run-7 with two pulls read through the seat git", c, err, found, *reads) + } + for _, p := range c.Pulls { + m := p.Modules[0] + if p.Number != 0 || len(p.Modules) != 1 || p.Fps[m].Manifest == "" || (p.Fps[m].Fp2 == "" && p.Fps[m].None2 == "") || + (m == "avahi") != (p.Repository == catalogueRepo && p.Head == c1) { + t.Errorf("pull %+v, want number 0, its registration's repository at its commit, a manifest and a fingerprint or why none", p) + } + } + if err := ix.EndRunPipeline(ctx, ended("avahi")); err != nil { + t.Fatal(err) + } + // A restart: a fresh indexer over the same states. + again := &Indexer{Store: cands, Runs: runs, Revisions: ix.Revisions, Registry: ix.Registry, Read: ix.Read} + if _, _, found, _ := cands.Get(ctx, key); !found { + t.Fatal("the candidate went before its last member's pipeline ended") + } + if err := again.EndRunPipeline(ctx, ended("netcheck")); err != nil { + t.Fatal(err) + } + if _, _, found, _ := cands.Get(ctx, key); found { + t.Error("the candidate stays after every member's pipeline ended") + } + if k, _ := runs.Keys(ctx); len(k) != 0 { + t.Errorf("runs holds %v after the run's end, want none", k) + } + }) + t.Run("ends heard before the start drop it at the start", func(t *testing.T) { + ix, cands, runs, _ := bed(t) + for _, m := range []string{"avahi", "netcheck"} { + if err := ix.EndRunPipeline(ctx, ended(m)); err != nil { + t.Fatal(err) + } + } + if err := ix.StartRun(ctx, started); err != nil { + t.Fatal(err) + } + k1, _ := cands.Keys(ctx) + k2, _ := runs.Keys(ctx) + if len(k1)+len(k2) != 0 { + t.Errorf("candidates %v, runs %v, want none", k1, k2) + } + }) + t.Run("a member with no registration is never taken in; a pipeline of no run is none of this", func(t *testing.T) { + ix, cands, _, _ := bed(t) + lost := started + lost.Members = []meshpipelinerunner.Member{{Module: "gone", Commit: c1}} + if err := ix.StartRun(ctx, lost); !errors.Is(err, ErrUnreadable) { + t.Errorf("a member with no registration: %v, want unreadable", err) + } + if err := ix.EndRunPipeline(ctx, meshpipelinerunner.PipelineEnded{Module: "avahi"}); err != nil { + t.Error(err) + } + if k, _ := cands.Keys(ctx); len(k) != 0 { + t.Errorf("candidates %v, want none", k) + } + }) + t.Run("a member's take-in starts from its own branch only when its registration watches it", func(t *testing.T) { + reg := meshcatalog.Registration{Branches: []string{"main", "beta"}} + for branch, want := range map[string]string{"beta": "beta", "main": TrunkBranch, "feature/1966-x": TrunkBranch, "": TrunkBranch} { + if got := RunBase(reg, branch); got != want { + t.Errorf("RunBase(%q) = %q, want %q", branch, got, want) + } + } + }) + t.Run("run- is a run's group beside try-, never a pull request's", func(t *testing.T) { + for g, want := range map[string]bool{"run-7": true, "run-a1-b": true, "try-3": true, "run-": false, "run-a.b": false, "feat/x": false} { + if meshcatalogclient.IsTryGroup(g) != want { + t.Errorf("IsTryGroup(%q) = %v, want %v", g, !want, want) + } + } + }) +} diff --git a/modules/mesh-delivery-indexer/internal/index/takein.go b/modules/mesh-delivery-indexer/internal/index/takein.go index f007fddb..74aa1eb2 100644 --- a/modules/mesh-delivery-indexer/internal/index/takein.go +++ b/modules/mesh-delivery-indexer/internal/index/takein.go @@ -57,13 +57,9 @@ type TakeInAsk struct { Merge string `json:"merge,omitempty"` MergeRepository string `json:"merge-repository,omitempty"` Sequence uint64 `json:"sequence,omitempty"` - // TryCandidate asks the candidate of this try's group to be dropped (indexer_drop_try_candidate), trycandidates.go. - TryCandidate string `json:"try-candidate,omitempty"` // Fingerprints asks the fingerprints of Module ("all": every module) to be computed again now, from the published // packages as they are read (indexer_fingerprints_again, fingerprints.go). Fingerprints bool `json:"fingerprints,omitempty"` - // RecoverTries asks the open tries' candidates to be recovered now (indexer_recover_tries). - RecoverTries bool `json:"recover-tries,omitempty"` // Recompute asks the catalog graph to be computed again now and compared with the newest kept revision (indexer_recompute). Recompute bool `json:"recompute,omitempty"` // CandidateGroup and CandidateHeads, with Drop, ask that candidate's catalog half to be dropped (indexer_drop_candidate_graph). @@ -95,8 +91,8 @@ func DecodeTakeIn(data []byte) (TakeInAsk, error) { if a.V != 1 { return a, fmt.Errorf("a takeins value of version %d, not 1", a.V) } - if a.Module == "" && a.Repository == "" && a.Pending == "" && a.Merge == "" && a.TryCandidate == "" && !a.RecoverTries && !a.Fingerprints && !a.CatchUp && !a.Recompute && a.CandidateGroup == "" && a.NamedFingerprint == "" && a.Registry == nil { - return a, errors.New("a takeins value naming no module, repository, pending wait, merge or try") + if a.Module == "" && a.Repository == "" && a.Pending == "" && a.Merge == "" && !a.Fingerprints && !a.CatchUp && !a.Recompute && a.CandidateGroup == "" && a.NamedFingerprint == "" && a.Registry == nil { + return a, errors.New("a takeins value naming no module, repository, pending wait or merge") } return a, nil } diff --git a/modules/mesh-delivery-indexer/internal/index/trycandidates.go b/modules/mesh-delivery-indexer/internal/index/trycandidates.go deleted file mode 100644 index 9c00c6d0..00000000 --- a/modules/mesh-delivery-indexer/internal/index/trycandidates.go +++ /dev/null @@ -1,437 +0,0 @@ -package index - -// **A group try's candidate** (issue 1921; novox/hq ADR 0593 §7): a try of several repositories at exact commits has no pull -// request, so the forge bridge sends the indexer no message for it; yet the node-engine's compose-candidate and the checker's -// manifest validator read the indexer's candidate of a group. The indexer therefore watches the checker's state tries (the -// checker keeps it; this module reads it) and, for a try that is not narrowed to a level or to some tests, writes -// `candidates.`: one pull per member (number 0), its modules and manifests taken from the merge preview the try's -// fetch wrote, its manifests kept and judged with the seat definitions the members carry, as a take-in judges a revision's. -// The fingerprints come with the previews (RecomputeCandidates). The candidate is deleted when the try ends. `try-` -// is the reserved group name of a try: the bridge's open message never takes it out, and the checker never checks it. - -import ( - "context" - "encoding/json" - "errors" - "fmt" - "path" - "slices" - "strings" - "time" - - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdeliverychecker" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" - "git.novox.be/novox/mesh-sdk/go/keys" -) - -// TryCandidateTools: the tools that read, drop and recover a try's candidate. -const ( - TryCandidatesTool = "indexer_try_candidates" - DropTryCandidateTool = "indexer_drop_try_candidate" - RecoverTriesTool = "indexer_recover_tries" - // TryRecoveryState is the indexer's state of how the last recovery of the open tries went: the bucket - // mesh-delivery-indexer_try-recovery, one key (TryRecoveryKey). - TryRecoveryState = "try-recovery" - TryRecoveryBucket = Module + "_" + TryRecoveryState - TryRecoveryKey = "state" -) - -// TryRecovery is the value of the try-recovery state: whether the candidates of the open tries were recovered at the -// indexer's start or by hand, and when it began to fail, why. -type TryRecovery struct { - V int `json:"v"` - Done bool `json:"done"` - Since time.Time `json:"since,omitzero"` - Why string `json:"why,omitempty"` - At time.Time `json:"at"` -} - -// TryMember is one repository of a try: its exact commit and the trunk commit it is merged onto. -type TryMember struct{ Repository, Head, Trunk string } - -// StartTry takes a whole try in, as the checker said it started (checkermsg.Started): its members at their commits are kept, and -// its candidate is written once every member's preview is held. The checker's state is never read (ADR 0596). -func (ix *Indexer) StartTry(ctx context.Context, id string, members []TryMember) error { - ix.tryMu.Lock() - defer ix.tryMu.Unlock() - ix.retryRecovery(ctx) - return ix.startTry(ctx, id, members) -} - -// retryRecovery runs the recovery again on a try event when the last one failed: the failure stays said until one succeeds. -func (ix *Indexer) retryRecovery(ctx context.Context) { - if ix.recoveryFailed.Load() && ix.OpenTries != nil { - _ = ix.recoverNow(ctx) - } -} - -func (ix *Indexer) startTry(ctx context.Context, id string, members []TryMember) error { - if !meshcatalogclient.IsTryGroup(id) || len(members) == 0 { - return fmt.Errorf("%q with %d members is no try to start", id, len(members)) - } - ix.mu.Lock() - if ix.tries == nil { - ix.tries = map[string][]TryMember{} - } - ix.tries[id] = slices.Clone(members) - ix.mu.Unlock() - return ix.writeTryCandidate(ctx, id) -} - -// EndTry takes a try out, as the checker said it ended (checkermsg.Ended), and deletes its candidate. -func (ix *Indexer) EndTry(ctx context.Context, id string) error { - ix.tryMu.Lock() - defer ix.tryMu.Unlock() - ix.retryRecovery(ctx) - return ix.endTry(ctx, id) -} - -func (ix *Indexer) endTry(ctx context.Context, id string) error { - ix.mu.Lock() - delete(ix.tries, id) - ix.mu.Unlock() - if !meshcatalogclient.IsTryGroup(id) { - return nil - } - return ix.dropTryCandidate(ctx, id) -} - -// RecoverTries is, at the indexer's start, what the events it missed would have said: the candidates of the tries that wait -// or run now are written, and the candidate of a try the checker no longer lists is deleted. The checker's tool lists them -// (OpenTries); its bucket is not read. Nothing to recover from when no part is wired to ask. -// -// A recovery that fails is said loudly and kept (the try-recovery state, read by indexer_try_candidates) until one succeeds: -// the events are still heard after it, and indexer_recover_tries runs it again by hand. -// -// A recovery and a try event never overlap (tryMu): a sweep that deleted the candidate a start event wrote after the call to the -// checker would leave a running try with none. -func (ix *Indexer) RecoverTries(ctx context.Context) error { - ix.tryMu.Lock() - defer ix.tryMu.Unlock() - return ix.recoverNow(ctx) -} - -func (ix *Indexer) recoverNow(ctx context.Context) error { - if ix.OpenTries == nil || ix.Store == nil { - return nil - } - err := ix.recoverTries(ctx) - ix.recoveryFailed.Store(err != nil) - ix.noteRecovery(ctx, err) - return err -} - -// noteRecovery writes how a recovery went: failing keeps the time it began to fail, and the journal says it each time. -func (ix *Indexer) noteRecovery(ctx context.Context, err error) { - now := ix.now() - r := TryRecovery{V: 1, Done: err == nil, At: now.UTC()} - if err != nil { - r.Why, r.Since = err.Error(), now.UTC() - ix.logf("RECOVERY OF THE TRIES NOT DONE: %v (the try candidates may be stale or missing until indexer_recover_tries or the next start succeeds)", err) - } - if ix.TryRecovery == nil { - return - } - for range writeTries { - cur, rev, found, gerr := ix.TryRecovery.Get(ctx, TryRecoveryKey) - if gerr != nil { - ix.logf("the try-recovery state cannot be read: %v", gerr) - return - } - if err != nil && found { - var was TryRecovery - if json.Unmarshal(cur, &was) == nil && !was.Done && !was.Since.IsZero() { - r.Since = was.Since - } - } - data, _ := json.Marshal(r) - var werr error - if found { - werr = ix.TryRecovery.Update(ctx, TryRecoveryKey, data, rev) - } else { - werr = ix.TryRecovery.Create(ctx, TryRecoveryKey, data) - } - if werr == nil { - return - } - if !errors.Is(werr, ErrConflict) { - ix.logf("the try-recovery state cannot be written: %v", werr) - return - } - } -} - -// RecoverTriesByHand runs the recovery now (indexer_recover_tries), a hand act said in the journal. -func (ix *Indexer) RecoverTriesByHand(ctx context.Context, by, why string) TakenIn { - r := TakenIn{Module: "tries"} - ix.logf("recovery of the tries asked by %s (%s)", by, why) - if ix.OpenTries == nil { - r.Answer = "the indexer is wired to list no open try: nothing to recover from" - return r - } - if err := ix.RecoverTries(ctx); err != nil { - r.Answer = "recovery not done: " + err.Error() - return r - } - r.Stored, r.Answer = true, "the candidates of the open tries are recovered" - return r -} - -func (ix *Indexer) recoverTries(ctx context.Context) error { - open, err := ix.OpenTries(ctx) - if err != nil { - return fmt.Errorf("the open tries cannot be listed, so a missed try event is not recovered: %w", err) - } - live := map[string]bool{} - for _, o := range open { - if !o.Whole || len(o.Members) == 0 { - continue - } - live[o.ID] = true - var ms []TryMember - for _, m := range o.Members { - ms = append(ms, TryMember{m.Repository, m.Head, m.Trunk}) - } - if err := ix.startTry(ctx, o.ID, ms); err != nil { - return err - } - } - keys, err := ix.Store.Keys(ctx) - if err != nil { - return err - } - for _, k := range keys { - if meshcatalogclient.IsTryGroup(k) && !live[k] { - if err := ix.endTry(ctx, k); err != nil { - return err - } - } - } - return nil -} - -// WriteTryCandidates writes the candidate of every live try that has none yet and whose previews are all there: after a -// preview was written, the try that waited for it. -func (ix *Indexer) WriteTryCandidates(ctx context.Context) error { - ix.mu.Lock() - ids := make([]string, 0, len(ix.tries)) - for id := range ix.tries { - ids = append(ids, id) - } - ix.mu.Unlock() - slices.Sort(ids) - for _, id := range ids { - if err := ix.writeTryCandidate(ctx, id); err != nil { - return err - } - } - return nil -} - -func (ix *Indexer) dropTryCandidate(ctx context.Context, id string) error { - if ix.Store == nil { - return nil - } - key, err := meshcatalogclient.CandidatesKey(id) - if err != nil { - return err - } - return write(ctx, ix.Store, key, func(_ []byte, found bool) ([]byte, bool, error) { return nil, found, nil }) -} - -// DropTryCandidate deletes a try's candidate by hand (indexer_drop_try_candidate); a live try gets it again at its next -// change. -func (ix *Indexer) DropTryCandidate(ctx context.Context, id, by, why string) TakenIn { - r := TakenIn{Module: id} - switch { - case !meshcatalogclient.IsTryGroup(id): - r.Answer = id + " is no try's group, try-: nothing dropped" - default: - if err := ix.dropTryCandidate(ctx, id); err != nil { - r.Answer = "the candidate of " + id + " could not be dropped: " + err.Error() - } else { - r.Stored, r.Answer = true, "the candidate of "+id+" is dropped by "+by+" ("+why+")" - ix.logf("%s", r.Answer) - } - } - return r -} - -// writeTryCandidate writes the candidate of a live try when it has none at its members' heads and every member's merge -// preview is held. -func (ix *Indexer) writeTryCandidate(ctx context.Context, id string) error { - if ix.Store == nil || ix.Read == nil { - return nil - } - ix.mu.Lock() - members := slices.Clone(ix.tries[id]) - ix.mu.Unlock() - if len(members) == 0 { - return nil - } - key, err := meshcatalogclient.CandidatesKey(id) - if err != nil { - return err - } - slices.SortFunc(members, func(a, b TryMember) int { return strings.Compare(a.Repository, b.Repository) }) - heads := make([]meshdeliveryclient.Head, len(members)) - for i, m := range members { - heads[i] = meshdeliveryclient.Head{Repository: m.Repository, Head: m.Head} - } - hash, err := meshdeliveryclient.HeadsHash(heads) - if err != nil { - return err - } - if value, _, found, err := ix.Store.Get(ctx, key); err == nil && found { - if c, err := DecodeCandidate(value); err == nil && c.Heads == hash { - return nil // fingerprints follow the previews (RecomputeCandidates) - } - } - previews := make([]meshdeliverychecker.PreviewsByRepositoryHeadTrunkValue, len(members)) - for i, m := range members { - pk, err := PreviewKey(m.Repository, m.Head, m.Trunk) - if err != nil { - return fmt.Errorf("the try %s: %w", id, err) - } - ix.mu.Lock() - pv, ok := ix.previews[pk] - ix.mu.Unlock() - if !ok { - return nil // its fetch has not written the preview yet: this is called again when it does - } - previews[i] = pv - } - held, err := ix.heldModules(ctx) - if err != nil { - return err - } - pulls := make([]meshcatalog.CandidatesByGroupValuePullsItem, len(members)) - var entries []PendingEntry - over := map[string]string{} - for i, m := range members { - pull := meshcatalog.CandidatesByGroupValuePullsItem{Repository: heads[i].Repository, Head: heads[i].Head, Modules: []string{}} - // defs are every module directory's manifest of the member at its head, the unchanged ones too: a contracts module whose - // seat.json changed while its module.json did not carries definitions all the same (ADR 0636 §4). - defs := map[string]string{} - for _, dir := range slices.Sorted(mapKeys(previews[i].Modules)) { - pm := previews[i].Modules[dir] - name, unchanged := "", false - for n, h := range held { - if strings.EqualFold(h.Repository, m.Repository) && h.Path == forgeDir(dir) { - name, unchanged = n, h.Manifest == pm.Manifest - break - } - } - if unchanged { - if raw := ix.keptManifest(ctx, pm.Manifest); raw != "" { - defs[dir] = raw - } - continue // the manifest the trunk holds: the try does not touch the module - } - rr, err := ix.Read(ctx, m.Repository, m.Head, forgeDir(dir)) - if err != nil { - return fmt.Errorf("the try %s: reading the manifest of %s at %.12s: %w", id, dir, m.Head, err) - } - raw := rr.Manifests[forgeDir(dir)] - if raw == "" { - ix.logf("the try %s: %s holds no manifest at %.12s, so no module of it is taken", id, dir, m.Head) - continue - } - module := manifestModule(raw) - if module == "" { - module = name - } - if module == "" { - module = path.Base(strings.Trim(dir, "/.")) - } - digest, err := ix.keepManifest(ctx, module, raw) - if err != nil { - return err - } - if digest != pm.Manifest { - ix.logf("the try %s: the manifest of %s read at %.12s is not the one its preview names, left", id, dir, m.Head) - continue - } - if pull.Fps == nil { - pull.Fps = map[string]meshcatalog.CandidatesByGroupValuePullsItemFpsValue{} - } - pull.Modules = append(pull.Modules, module) - pull.Fps[module] = meshcatalog.CandidatesByGroupValuePullsItemFpsValue{Manifest: digest, None2: meshcatalogclient.NoneNoPreview} - defs[dir] = raw - over[module] = digest - } - pulls[i] = pull - entries = append(entries, PendingEntry{Forge: forgemsg.Forge{Repository: m.Repository, Merge: m.Head, Manifests: defs}}) - } - // The manifests are judged with the seat definitions the members carry (seatsof.go, ADR 0636 §4), in one pass that takes the - // seats. Their verdicts are the verdicts state's, keyed by the manifest's digest, as a pull request's candidate's are - // (judgeDigests in ApplyMessage): nothing reads them but the engines' composition of that digest and the module - // value of a trunk module with the same digest. - seats, err := ix.revisionSeats(ctx, entries) - if err != nil { - return fmt.Errorf("the try %s: reading the seat definitions of its members: %w", id, err) - } - ix.judgeDigestsWith(ctx, seats, over, "the try "+id) - c := meshcatalog.CandidatesByGroupValue{V: keys.Version, Group: id, Heads: hash, Pulls: pulls} - c.Fpset = meshcatalogclient.FingerprintSet(c.Pulls) - ix.withPreviews(ctx, &c) - if err := ix.refuseDisagreement(c); err != nil { - return err - } - data, err := EncodeCandidate(c) - if err != nil { - return err - } - return write(ctx, ix.Store, key, func(_ []byte, _ bool) ([]byte, bool, error) { return data, false, nil }) -} - -// keptManifest is the text of the manifest the indexer keeps under a digest, or empty. -func (ix *Indexer) keptManifest(ctx context.Context, digest string) string { - if ix.Manifests == nil || digest == "" { - return "" - } - key, err := meshcatalogclient.ManifestsKey(digest) - if err != nil { - return "" - } - value, _, found, err := ix.Manifests.Get(ctx, key) - if err != nil || !found { - return "" - } - m, err := meshcatalogclient.DecodeManifest(value) - if err != nil { - return "" - } - return m.Manifest -} - -func mapKeys[V any](m map[string]V) func(func(string) bool) { - return func(yield func(string) bool) { - for k := range m { - if !yield(k) { - return - } - } - } -} - -// forgeDir is a preview's directory as the forge reader spells it: the root as "". -func forgeDir(dir string) string { - if dir == "." || dir == "/" { - return "" - } - return strings.Trim(dir, "/") -} - -// heldAt says the trunk holds a module of the repository in that directory. -func heldAt(held map[string]meshcatalog.ModulesByModuleValue, repository, dir string) bool { - for _, h := range held { - if strings.EqualFold(h.Repository, repository) && h.Path == forgeDir(dir) { - return true - } - } - return false -} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 56b7eeda..6ea9b9cc 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -126,7 +126,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build t.Errorf("uses = %v, invokes = %v, want the seat operator-channel used and seat:mesh-module-assigner.assignments invoked", m.Uses, m.Invokes) } // 1126 note 76: heard keeps the last 64 FORGE messages the indexer handled, for indexer_heard. - if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: "try-recovery", History: 1}, {Name: "own-build", History: 1}, {Name: index.PackagesState, History: index.PackagesHistory}, + if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: index.RunsState, History: 1}, {Name: "own-build", History: 1}, {Name: index.PackagesState, History: index.PackagesHistory}, {Name: views.MarksState, History: 1}, {Name: index.FindingsState, History: 5}, {Name: views.StatusState, History: 1}, {Name: index.RequirementsState, History: 5}, {Name: index.SeatDefinitionsState, History: 5}, {Name: index.PendingState, History: 1}, {Name: index.UnjudgedState, History: 1}, {Name: index.EntryIndexState, History: 1, MaxMiB: 8}, {Name: index.NamedState, History: 1, MaxMiB: 4}, // design 47 G §695 @@ -166,8 +166,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // internal/broker/state.go:62-93). // The registry seats' published fill each (§413: no consume of version.numbered) package's series (DLV-I5, design 51 *The versions*). if want := []string{"go-package-registry.published", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-conditions.findings-asked", "mesh-delivery-assigner.views.assignments", - "mesh-delivery-checker.preview.written", "mesh-delivery-checker.try.ended", "mesh-delivery-checker.try.started", "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**", - "mesh-delivery.source-moved", "mesh-wants.version-delivered", "mesh-wants.wait-ended", "mesh-wants.wanted", "node-build-go.step-ended", "node-engine.running", "npm-package-registry.published", "operator-channel.decided"}; !reflect.DeepEqual(sorted(m.Consumes), want) { + "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**", + "mesh-delivery.source-moved", "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.run-started", "mesh-wants.version-delivered", "mesh-wants.wait-ended", "mesh-wants.wanted", "node-build-go.step-ended", "node-engine.running", "npm-package-registry.published", "operator-channel.decided"}; !reflect.DeepEqual(sorted(m.Consumes), want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" { @@ -179,7 +179,7 @@ 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{"mesh-delivery-assigner.assignments", "seat:git.trees", "seat:git.file", "seat:git.contains", "seat:git.open-pulls", "seat:git.branch", "mesh-delivery-checker.checker_tries", "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{"mesh-delivery-assigner.assignments", "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) { 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 @@ -198,12 +198,12 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build } want = append(want, "indexer_first_views", "indexer_first_view_again", index.ConfigReadersTool, index.BusStateTool, index.RequirementsTool, index.PackageUsersTool, index.SayAcceptedTool, // DLV-F3r, DLV-I10, DLV-I11, DLV-I7, DLV-I8, DLV-I9c index.SeatDefinitionTool, index.SeatDefinitionsAgainTool, index.OutOfStepTool, // DLV-I19 - index.PendingTool, index.TakeInPendingTool, index.DropPendingTool, index.TryCandidatesTool, index.DropTryCandidateTool, index.RecoverTriesTool, index.TakeInMergeTool, index.FingerprintsAgainTool, // issue 1966; 1585/T12, T13 (ADR 0633 §5) + index.PendingTool, index.TakeInPendingTool, index.DropPendingTool, index.TakeInMergeTool, index.FingerprintsAgainTool, // issue 1966; 1585/T12, T13 (ADR 0633 §5) // 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) // design 47 G §667, §8 - if !reflect.DeepEqual(m.Tools, want) || len(want) != 47 || slices.Contains(m.Tools, "modules") { + if !reflect.DeepEqual(m.Tools, want) || len(want) != 44 || 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 4f6e4770..30b23dc5 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -58,9 +58,6 @@ "indexer_pending", "indexer_take_in_pending", "indexer_drop_pending", - "indexer_try_candidates", - "indexer_drop_try_candidate", - "indexer_recover_tries", "indexer_take_in_merge", "indexer_fingerprints_again", "indexer_catalog_revision", @@ -81,9 +78,6 @@ "mesh-build-node.toolchain.reported", "mesh-conditions.findings-asked", "mesh-delivery-assigner.views.assignments", - "mesh-delivery-checker.preview.written", - "mesh-delivery-checker.try.ended", - "mesh-delivery-checker.try.started", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-wants.version-delivered", @@ -93,6 +87,8 @@ "mesh-delivery-forge-bridge.forge.**", "mesh-delivery.source-moved", "mesh-wants.wanted", + "mesh-pipeline-runner.run-started", + "mesh-pipeline-runner.pipeline-ended", "node-engine.running", "operator-channel.decided", "npm-package-registry.published" @@ -107,7 +103,7 @@ { "name": "takeins", "history": 1 }, { "name": "verdicts", "history": 5 }, { "name": "oldsections", "history": 1 }, - { "name": "try-recovery", "history": 1 }, + { "name": "runs", "history": 1 }, { "name": "own-build", "history": 1 }, { "name": "packages", "history": 5 }, { "name": "view-marks", "history": 1 }, @@ -160,7 +156,6 @@ "seat:git.contains", "seat:git.open-pulls", "seat:git.branch", - "mesh-delivery-checker.checker_tries", "mesh-delivery-forge-bridge.bridge_merged_message", "seat:node-engine.composition", "seat:mesh-delivery.source",