novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

A-194: indexer_runs reads the state runs and the run candidates, indexer_run_drop drops one by hand

Browse Source 
The drop is a hand act recorded through mesh-hand-acts before it is asked; the indexer, the state's owner, drops the
run's candidate and its key.

Builds: design 47 G §64; A-194

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-12 05:57:11 +01:00
parent 0c6c5d20d6
commit c2950e332f
7 changed files with 253 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go
+4
View File
@@ -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})
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go
+4
View File
@@ -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 != "" {
modules/mesh-delivery-indexer/internal/index/runcandidates_test.go
+58
View File
@@ -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)
}
}
modules/mesh-delivery-indexer/internal/index/runs_tools.go
+179
View File
@@ -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-<id>).
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-<id>) 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-<id> 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-<id>, 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
}},
}
}
modules/mesh-delivery-indexer/internal/index/takein.go
+3 -1
View File
@@ -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-<id>, 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
modules/mesh-delivery-indexer/manifest_test.go
+2 -2
View File
@@ -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.
modules/mesh-delivery-indexer/module.json
+3 -1
View File
@@ -73,7 +73,9 @@
"indexer_inputs",
"indexer_pins",
"indexer_unpin",
"indexer_consumer"
"indexer_consumer",
"indexer_runs",
"indexer_run_drop"
],
"consumes": [
"go-package-registry.published",