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 4a13173e..1509c9c2 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 @@ -304,6 +304,10 @@ func main() { _, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: index.UnpinTool, Verb: "unpin"}) return err })...) + tools = append(tools, index.RunsTools(keptRead{stdio.State(index.RunsState)}, keptRead{stdio.State(index.CandidatesState)}, asks, time.Now, func(what, why string) error { + _, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: index.RunDropTool, Verb: "run-drop"}) + return err + })...) tools = append(tools, index.ConsumerTools(keptRead{stdio.State(process.ConsumersState)})...) tools = append(tools, index.RegistrationTools(keptRead{stdio.State(index.ProposalsState)}, asks, time.Now, func(cause, verb, what, why string) error { _, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: cause, Verb: verb}) 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 f876df51..e72897a0 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go @@ -27,6 +27,7 @@ type takerIn interface { DropCandidateGraph(ctx context.Context, group, heads, by, why string) index.TakenIn NamedDrop(ctx context.Context, fingerprint, member, by, why string) index.TakenIn Unpin(ctx context.Context, revision, record, by, why string) index.TakenIn + DropRunByHand(ctx context.Context, run, by, why string) index.TakenIn RegistryAct(ctx context.Context, a index.TakeInAsk) index.TakenIn } @@ -60,6 +61,9 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream } else if a.UnpinRevision != "" { asked, r = "unpin of a record from a catalog graph revision", ix.Unpin(ctx, a.UnpinRevision, a.UnpinRecord, a.By, a.Why) a.Module = r.Module + } else if a.RunDrop != "" { + asked, r = "drop of a run on request", ix.DropRunByHand(ctx, a.RunDrop, a.By, a.Why) + a.Module = r.Module } else if a.Fingerprints { asked, r = "computing of fingerprints again", ix.FingerprintsAgain(ctx, a.Module) } else if a.Merge != "" { diff --git a/modules/mesh-delivery-indexer/internal/index/runcandidates_test.go b/modules/mesh-delivery-indexer/internal/index/runcandidates_test.go index dfd17632..2f6c4cfa 100644 --- a/modules/mesh-delivery-indexer/internal/index/runcandidates_test.go +++ b/modules/mesh-delivery-indexer/internal/index/runcandidates_test.go @@ -4,7 +4,9 @@ import ( "context" "encoding/json" "errors" + "slices" "testing" + "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/seats/meshcatalog" @@ -128,3 +130,59 @@ func TestARunsCandidateIsBuiltOnItsStartAndDroppedWhenEveryMembersPipelineEnded( } }) } + +// Design 47 G §64, A-194: the state runs and the run candidates are read by indexer_runs and a run is dropped by hand by +// indexer_run_drop, its hand act written first. +func TestTheRunsAreReadAndDroppedByHand(t *testing.T) { + ctx := context.Background() + cands, runs := newMem(), newMem() + ix := &Indexer{Store: cands, Runs: runs, Now: func() time.Time { return time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC) }} + put := func(s *memStore, key string, v any) { + t.Helper() + data, _ := json.Marshal(v) + if err := s.Create(ctx, key, data); err != nil { + t.Fatal(err) + } + } + const id = "0a1b2c3d4e5f" + put(runs, "run-"+id, Run{V: 1, Run: id, Members: []meshpipelinerunner.Member{{Module: "avahi", Commit: "c1"}, {Module: "netcheck", Commit: "c2"}}, Ended: []string{"avahi"}}) + for _, g := range []string{"run-" + id, "run-ffffffffffff"} { + key, _ := meshcatalogclient.CandidatesKey(g) + put(cands, key, map[string]string{"group": g}) + } + read, err := ReadRuns(ctx, runs, cands) + if err != nil || len(read.Runs) != 1 || read.Runs[0].Run != id || !read.Runs[0].Candidate || !slices.Equal(read.Runs[0].Waiting, []string{"netcheck"}) || !slices.Equal(read.Orphans, []string{"run-ffffffffffff"}) { + t.Fatalf("runs = %+v, %v, want the run waiting for netcheck with its candidate and one orphan candidate", read, err) + } + var recorded []string + var asked []string + tools := RunsTools(runs, cands, func(key string, value []byte) (TakeInAsk, error) { + asked = append(asked, key) + a, err := DecodeTakeIn(value) + if err != nil { + return a, err + } + a.Result = &TakenIn{} + *a.Result = ix.DropRunByHand(ctx, a.RunDrop, a.By, a.Why) + return a, nil + }, ix.now, func(what, why string) error { recorded = append(recorded, what); return nil }) + if len(tools) != 2 || tools[0].Name != RunsTool || tools[1].Name != RunDropTool { + t.Fatalf("tools = %v", tools) + } + if _, err := tools[1].Run(map[string]any{"run": id}); err == nil || len(recorded) != 0 { + t.Fatalf("a drop with no why was done: %v, %v", err, recorded) + } + if _, err := tools[1].Run(map[string]any{"run": id, "why": "its pipelines never end"}); err != nil || len(recorded) != 1 || len(asked) != 1 { + t.Fatalf("drop: %v, recorded %v, asked %v: want the hand act recorded, then the ask", err, recorded, asked) + } + read, _ = ReadRuns(ctx, runs, cands) + if len(read.Runs) != 0 || len(read.Orphans) != 1 { + t.Errorf("after the drop: %+v, want the run and its candidate gone and the orphan left", read) + } + if got := ix.DropRunByHand(ctx, "ffffffffffff", "lead", "an orphan"); !got.Stored { + t.Errorf("an orphan candidate is not dropped: %+v", got) + } + if got := ix.DropRunByHand(ctx, id, "lead", "again"); got.Stored { + t.Errorf("a run already dropped was dropped again: %+v", got) + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/runs_tools.go b/modules/mesh-delivery-indexer/internal/index/runs_tools.go new file mode 100644 index 00000000..7135bc3c --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/runs_tools.go @@ -0,0 +1,179 @@ +package index + +// The tools of the state runs and the run candidates (design 47 G §64; A-174, A-194): indexer_runs reads every run the indexer +// holds a candidate for, and indexer_run_drop drops one by hand, its hand act written first. + +import ( + "context" + "encoding/json" + "errors" + "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/seats/meshpipelinerunner" + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +const ( + // RunsTool reads the runs; RunDropTool drops one by hand. + RunsTool = "indexer_runs" + RunDropTool = "indexer_run_drop" +) + +// RunRead is one run of the state runs as indexer_runs lists it. +type RunRead struct { + // Run is the run's id as the runner says it, Group the candidate's group and the key of the state (run-). + Run string `json:"run"` + Group string `json:"group"` + Members []meshpipelinerunner.Member `json:"members,omitempty"` + // Ended are the modules whose pipeline of the run has ended, Waiting the members whose has not. + Ended []string `json:"ended,omitempty"` + Waiting []string `json:"waiting,omitempty"` + At time.Time `json:"at"` + // Candidate says the run's candidate is in the state candidates. + Candidate bool `json:"candidate"` +} + +// RunsRead is the answer of indexer_runs: the runs the state holds, and the run candidates no run of the state holds. +type RunsRead struct { + Runs []RunRead `json:"runs"` + // Orphans are the groups of the candidates that are a run's (run-) while the state runs holds no such run. + Orphans []string `json:"orphans,omitempty"` +} + +// ReadRuns reads the state runs and the run candidates of the state candidates; neither is written. +func ReadRuns(ctx context.Context, runs, candidates Store) (RunsRead, error) { + out := RunsRead{Runs: []RunRead{}} + if runs == nil { + return out, fmt.Errorf("the state %s is not bound", RunsState) + } + held := map[string]bool{} // the candidates' keys + if candidates != nil { + ks, err := candidates.Keys(ctx) + if err != nil { + return out, fmt.Errorf("listing %s: %w", CandidatesState, err) + } + for _, k := range ks { + held[k] = true + } + } + names, err := runs.Keys(ctx) + if err != nil { + return out, fmt.Errorf("listing %s: %w", RunsState, err) + } + slices.Sort(names) + known := map[string]bool{} + for _, name := range names { + value, _, found, err := runs.Get(ctx, name) + if err != nil { + return out, fmt.Errorf("%s.%s: %w", RunsState, name, err) + } + var r Run + if !found || json.Unmarshal(value, &r) != nil { + continue + } + key, _ := meshcatalogclient.CandidatesKey(name) + known[key] = true + read := RunRead{Run: r.Run, Group: name, Members: r.Members, Ended: r.Ended, At: r.At, Candidate: held[key]} + for _, m := range r.Members { + if !slices.Contains(r.Ended, m.Module) { + read.Waiting = append(read.Waiting, m.Module) + } + } + out.Runs = append(out.Runs, read) + } + for k := range held { + if group, err := meshcatalogclient.ParseCandidatesKey(k); err == nil && strings.HasPrefix(group, "run-") && !known[k] { + out.Orphans = append(out.Orphans, group) + } + } + slices.Sort(out.Orphans) + return out, nil +} + +// runGroupOf is the group of a run the tool names: the runner's id, or run- as indexer_runs lists it. +func runGroupOf(run string) (string, error) { + return RunGroup(strings.TrimPrefix(strings.TrimSpace(run), "run-")) +} + +// DropRunByHand is indexer_run_drop answered: the run's candidate and its key in the state runs are removed, as they are when +// every member's pipeline has ended. A run no state or candidate holds drops nothing. +func (ix *Indexer) DropRunByHand(ctx context.Context, run, by, why string) TakenIn { + r := TakenIn{Module: "run " + run} + group, err := runGroupOf(run) + if err != nil { + r.Answer = err.Error() + ": nothing dropped" + return r + } + if ix.Runs == nil { + r.Answer = "the indexer keeps no state " + RunsState + ": nothing dropped" + return r + } + ix.runMu.Lock() + defer ix.runMu.Unlock() + read, err := ReadRuns(ctx, ix.Runs, ix.Store) + if err != nil { + r.Answer = "the runs are not read: " + err.Error() + return r + } + if !slices.ContainsFunc(read.Runs, func(x RunRead) bool { return x.Group == group }) && !slices.Contains(read.Orphans, group) { + r.Answer = "no run " + group + " is held by the state " + RunsState + " or the candidates: nothing dropped" + return r + } + if err := ix.dropRun(ctx, group); err != nil { + r.Answer = group + " could not be dropped: " + err.Error() + ix.logf("%s", r.Answer) + return r + } + r.Stored = true + r.Answer = fmt.Sprintf("the run %s and its candidate are dropped by %s (%s)", group, by, why) + ix.logf("%s", r.Answer) + return r +} + +// RunsTools are indexer_runs and indexer_run_drop (class act): the tools bundle reads the states and writes an ask in takeins, +// and the indexer, the state's owner, drops the run; record says the hand act in the hand-act log before it is asked. +func RunsTools(runs, candidates Store, ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time, record func(what, why string) error) []stdio.Tool { + return []stdio.Tool{ + {Name: RunsTool, + Description: "Every run on request the indexer holds a candidate for, from its state runs (design 47 G §64, A-174): the run's id and " + + "group, its members (module, commit, branch), the modules whose pipeline of the run has ended and those still waiting, " + + "when it was last written, and whether its candidate is in the state candidates; and the run candidates no run of the state " + + "holds (orphans). indexer_run_drop drops one.", + Input: map[string]any{"type": "object", "properties": map[string]any{}}, + Run: func(map[string]any) (any, error) { + return ReadRuns(context.Background(), runs, candidates) + }}, + {Name: RunDropTool, + Description: "Drop one run on request of the indexer: its candidate and its key in the state runs, as the end of its last member's " + + "pipeline does (design 47 G §64, A-194), for a run whose pipelines never end. A hand act, recorded through mesh-hand-acts " + + "before it is asked. indexer_runs names the runs.", + Input: map[string]any{"type": "object", "required": []string{"run", "why"}, "properties": map[string]any{ + "run": map[string]any{"type": "string", "description": "the run's id, or its group run-, as indexer_runs names it"}, + "why": map[string]any{"type": "string", "description": "why it is dropped, in plain words"}}}, + Run: func(args map[string]any) (any, error) { + str := func(k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) } + run, why := str("run"), str("why") + group, err := runGroupOf(run) + if err != nil || why == "" { + return nil, errors.New("name the run as indexer_runs names it (run) and say why (why): a drop is a hand act, recorded through mesh-hand-acts") + } + data, err := json.Marshal(TakeInAsk{V: 1, RunDrop: group, Why: why, By: stdio.CallerOf(args), At: now().UTC()}) + if err != nil { + return nil, err + } + if err := record("dropped the run "+group+" and its candidate by hand", why); err != nil { + return nil, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so nothing was dropped: %w", err) + } + a, err := ask("run-drop."+keys.EscapeName(group), data) + if err != nil { + return nil, err + } + return a.Result, nil + }}, + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/takein.go b/modules/mesh-delivery-indexer/internal/index/takein.go index 6e1ed4b8..7e581059 100644 --- a/modules/mesh-delivery-indexer/internal/index/takein.go +++ b/modules/mesh-delivery-indexer/internal/index/takein.go @@ -73,6 +73,8 @@ type TakeInAsk struct { UnpinRevision string `json:"unpin-revision,omitempty"` UnpinRecord string `json:"unpin-record,omitempty"` NamedMember string `json:"named-member,omitempty"` + // RunDrop asks one run on request, by its group run-, to be dropped with its candidate (indexer_run_drop, runs_tools.go, A-194). + RunDrop string `json:"run-drop,omitempty"` // Registry asks one of the registrations' acts (registrations.go): a trusted change of one, the verb registrations, // indexer_proposal_drop or indexer_seed_again (design 47 G §3, §4, §8, §667). Registry *RegistryAct `json:"registry,omitempty"` @@ -94,7 +96,7 @@ 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.Fingerprints && !a.CatchUp && !a.Recompute && a.CandidateGroup == "" && a.NamedFingerprint == "" && a.UnpinRevision == "" && a.Registry == nil { + if a.Module == "" && a.Repository == "" && a.Pending == "" && a.Merge == "" && !a.Fingerprints && !a.CatchUp && !a.Recompute && a.CandidateGroup == "" && a.NamedFingerprint == "" && a.UnpinRevision == "" && a.RunDrop == "" && 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/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index d1997018..2eba8cea 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -202,8 +202,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // milestone 47/G W4.12 (design 47, the indexer's tools): the catalog graph revisions read and steered. index.CatalogRevisionTool, index.CatalogRevisionsTool, index.CandidateGraphsTool, index.RecomputeTool, index.DropCandidateGraphTool, index.EntryIndexTool, index.NamedDropTool, // design 47 G §675 - index.ProposalsTool, index.ProposalDropTool, index.SeedAgainTool, index.InputsTool, index.PinsTool, index.UnpinTool, index.ConsumerTool) // design 47 G §667, §8, §579 - if !reflect.DeepEqual(m.Tools, want) || len(want) != 47 || slices.Contains(m.Tools, "modules") { + index.ProposalsTool, index.ProposalDropTool, index.SeedAgainTool, index.InputsTool, index.PinsTool, index.UnpinTool, index.ConsumerTool, index.RunsTool, index.RunDropTool) // design 47 G §667, §8, §579, §64 (A-194) + if !reflect.DeepEqual(m.Tools, want) || len(want) != 49 || 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 1e7d2c6f..6041744f 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -73,7 +73,9 @@ "indexer_inputs", "indexer_pins", "indexer_unpin", - "indexer_consumer" + "indexer_consumer", + "indexer_runs", + "indexer_run_drop" ], "consumes": [ "go-package-registry.published",