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

indexer: the seed asks steps of each artifact's build seat from the newest trunk entry

Browse Source 
The seat list's build seat per artifact (language or kind), a kept call
<seat>.steps with {module, artifact, fingerprint, fetch}, and the finding
call-unanswered (scope call, id <seat>.steps) after five minutes, cleared on an answer.

Builds: design 47 G §704

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-11 22:56:06 +01:00
parent 0f62358fb0
commit c0e2621bba
7 changed files with 311 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go
+6 -2
View File
@@ -452,8 +452,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.ProposalsBucket, "no change of a registration waits for its warrant", logger.Printf)},
routerRecordUnread, ix, logger.Printf)
ix.Registry = registry
ix.SeedModules = func(context.Context) ([]index.SeedModule, error) {
return nil, errors.New("the first registrations are not composed by this build: what the build seats answer steps with is not read yet (design question)")
ix.SeedModules = func(ctx context.Context) ([]index.SeedModule, error) {
r := seatReader.Reader()
if r == nil {
return nil, errors.New("the seat list is not read yet: the seed waits for it")
}
return ix.SeedModulesFrom(ctx, r.Definitions(), stepsVia(calls.CallKept))
}
// Which build this program is, fixed here once, before it hears any merge and never asked again (index.FixOwnBuild): the
// node-engine names the successor while it replaces this process, and an old process that learnt it later would take
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/steps.go
+28
View File
@@ -0,0 +1,28 @@
package main
import (
"context"
"encoding/json"
"fmt"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
"mesh-delivery-indexer/internal/index"
)
// stepsVia asks a build seat's verb steps through call (process.Conn.CallKept, ADR 0668 §1) and reads its answer (design
// 47 G §704); the seats' verbs have one form, so nodebuildgo's contract reads each seat's answer.
func stepsVia(call toolCaller) index.StepsCaller {
return func(_ context.Context, seat string, in nodebuildgo.StepsInput) (nodebuildgo.StepsAnswer, error) {
verb := "seat:" + seat + "." + index.StepsVerb
raw, err := call(verb, in)
if err != nil {
return nodebuildgo.StepsAnswer{}, err
}
var a nodebuildgo.StepsAnswer
if err := json.Unmarshal(raw, &a); err != nil {
return a, fmt.Errorf("the answer of %s is not its steps: %w", verb, err)
}
return a, nil
}
}
modules/mesh-delivery-indexer/internal/index/candidates.go
+1
View File
@@ -149,6 +149,7 @@ type Indexer struct {
previews map[string]meshdeliverychecker.PreviewsByRepositoryHeadTrunkValue // by previews key (previews.go)
toolchains map[string]map[string]nodebuildgo.Toolchain // language, then node (toolchains.go)
buildModules map[string]BuildModule // by language: the build modules whose toolchains are taken in (buildmodules.go)
stepsSince map[string]time.Time // by build seat: since when its steps is unanswered (seed_steps.go)
tried map[string]bool // repository, commit and directory a registered module's manifest was tried at (registered.go)
unread map[string]bool // registered modules whose last read the forge could not answer, read again by RetryUnread
moving map[string]forgemsg.Forge // by modules key: a merge naming a module another repository holds, waiting for that trunk to drop it (moves.go)
modules/mesh-delivery-indexer/internal/index/seed_steps.go
+178
View File
@@ -0,0 +1,178 @@
package index
import (
"context"
"fmt"
"maps"
"path"
"slices"
"time"
mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
"git.novox.be/novox/mesh-sdk/go/seatdef"
)
// StepsVerb is the verb of each build seat the seed asks (design 47 G §704).
const StepsVerb = "steps"
// StepsCaller asks the build seat seat its verb steps (a kept call, ADR 0668 §1): the answer, or why none came.
type StepsCaller func(ctx context.Context, seat string, in nodebuildgo.StepsInput) (nodebuildgo.StepsAnswer, error)
// ModuleManifest is the file name of a module's manifest in its directory.
const ModuleManifest = "module.json"
// buildSeatOf is the seat whose acting rule `build` has `on` matching the artifact's language or kind (design 47 G §704,
// §203, §324), read from the seat list defs as BuildModulesOf reads it; none, or more than one, is said.
func buildSeatOf(defs map[string]seatdef.Definition, a meshcatalog.Artifact) (string, error) {
var seats []string
for _, name := range slices.Sorted(maps.Keys(defs)) {
for _, r := range defs[name].Acts {
if r.Step != seatdef.StepBuild {
continue
}
if r.ActsOn(a.Language) || (r.On == seatdef.OnKindImage && a.Kind == "image") {
seats = append(seats, name)
break
}
}
}
switch len(seats) {
case 0:
return "", fmt.Errorf("no seat builds the artifact %s (language %q, kind %q)", a.Ref, a.Language, a.Kind)
case 1:
return seats[0], nil
}
return "", fmt.Errorf("the seats %v each build the artifact %s", seats, a.Ref)
}
// SeedModulesFrom reads the modules the newest trunk revision holds for the seed (design 47 G §704, §8): for each artifact of
// each module the seat list's build seat answers steps with {module, artifact, fingerprint, fetch} from the module's trunk
// entry, the fetch its repository, commit, directory and the artifact's named paths. A seat that does not answer leaves
// its artifact unanswered, which holds the seed; its call-unanswered finding (scope call, id <seat>.steps) is written once
// it is unanswered five minutes and cleared when it answers.
func (ix *Indexer) SeedModulesFrom(ctx context.Context, defs map[string]seatdef.Definition, steps StepsCaller) ([]SeedModule, error) {
if ix.Revisions == nil {
return nil, fmt.Errorf("the catalog graph is not kept here: no trunk revision to seed from")
}
rev, found, err := newestTrunk(ctx, ix.Revisions)
if err != nil {
return nil, err
}
if !found {
return nil, fmt.Errorf("the indexer keeps no trunk revision yet: the seed waits for its first")
}
_, entries, err := ReadRevision(ctx, ix.Revisions, fmt.Sprintf("trunk.%d", rev.Revision))
if err != nil {
return nil, err
}
seatOf := map[string]string{} // artifact reference -> its build seat, all found before any call is made
for _, e := range entries {
for _, a := range e.Artifacts {
seat, err := buildSeatOf(defs, a)
if err != nil {
return nil, fmt.Errorf("%s: %w", e.Module, err)
}
seatOf[a.Ref] = seat
}
}
answered := map[string]bool{}
unanswered := map[string]error{}
var out []SeedModule
for _, e := range entries {
m := SeedModule{Module: e.Module, Origin: e.Repository, Path: path.Join(e.Path, ModuleManifest)}
for _, a := range e.Artifacts {
seat := seatOf[a.Ref]
name := artifactName(a.Ref)
sa := SeedArtifact{Name: name, Seat: seat}
in := nodebuildgo.StepsInput{Module: e.Module, Artifact: nodebuildgo.DetectedStepsByModuleArtifactKey(e.Module, name), Fingerprint: e.Fp,
Fetch: &nodebuildgo.Fetch{Repository: e.Repository, Commit: e.Commit, Dir: e.Path, Paths: namedPaths(e, name)}}
ans, err := steps(ctx, seat, in)
switch {
case err != nil:
unanswered[seat] = err
case ans.Error != "":
unanswered[seat] = fmt.Errorf("%s", ans.Error)
default:
sa.Answered = true
answered[seat] = true
for _, s := range ans.Steps {
sa.Steps = append(sa.Steps, s.Step)
}
}
m.Artifacts = append(m.Artifacts, sa)
}
out = append(out, m)
}
ix.stepsFindings(ctx, answered, unanswered)
return out, nil
}
// artifactName is the artifact's name in an artifact reference <module>:<artifact>.
func artifactName(ref string) string {
for i := len(ref) - 1; i >= 0; i-- {
if ref[i] == ':' {
return ref[i+1:]
}
}
return ref
}
// stepsFindings writes call-unanswered for each seat unanswered for five minutes and clears it for each that answered.
func (ix *Indexer) stepsFindings(ctx context.Context, answered map[string]bool, unanswered map[string]error) {
if ix.Findings == nil {
return
}
ix.mu.Lock()
if ix.stepsSince == nil {
ix.stepsSince = map[string]time.Time{}
}
now := ix.now()
var raise []string
for seat := range unanswered {
if _, ok := ix.stepsSince[seat]; !ok {
ix.stepsSince[seat] = now
}
if now.Sub(ix.stepsSince[seat]) >= mcc.CallUnansweredAfter {
raise = append(raise, seat)
}
}
for seat := range answered {
if _, bad := unanswered[seat]; !bad {
delete(ix.stepsSince, seat)
}
}
since := maps.Clone(ix.stepsSince)
ix.mu.Unlock()
for _, seat := range slices.Sorted(slices.Values(raise)) {
id := seat + "." + StepsVerb
f := ix.observed(mcc.ScopeCall, id, mcc.KindCallUnanswered,
fmt.Sprintf("the call %s is unanswered since %s", id, since[seat].UTC().Format(time.RFC3339)),
"A call whose answer is kept is not answered",
fmt.Sprintf("The seed asks %s of every artifact its build seat builds; it is not answered since %s (%v), so the first registrations wait for it.", id, since[seat].UTC().Format(time.RFC3339), unanswered[seat]),
"indexer."+mcc.KindCallUnanswered)
if err := ix.Findings.Write(ctx, f); err != nil {
ix.logf("seed: the finding %s is not written: %v", id, err)
}
}
for seat := range answered {
if _, bad := unanswered[seat]; bad {
continue
}
if err := ix.clearFinding(ctx, mcc.ScopeCall, seat+"."+StepsVerb, mcc.KindCallUnanswered, "the seat answers steps"); err != nil {
ix.logf("seed: the finding %s.%s is not cleared: %v", seat, StepsVerb, err)
}
}
}
// namedPaths are the named paths the entry keeps for the artifact, by its name or its reference.
func namedPaths(e meshcatalog.CatalogEntry, name string) []string {
if e.Inputs == nil {
return nil
}
if p, ok := e.Inputs.Paths[name]; ok {
return p
}
return e.Inputs.Paths[ArtifactRefOf(e.Module, name)]
}
modules/mesh-delivery-indexer/internal/index/seed_steps_test.go
+93
View File
@@ -0,0 +1,93 @@
package index
import (
"context"
"errors"
"strings"
"testing"
"time"
mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
"git.novox.be/novox/mesh-sdk/go/seatdef"
)
// Design 47 G §704: the seed asks `steps` of the build seat whose `build` rule matches each artifact, with {module,
// artifact, fingerprint, fetch} from the newest trunk entry; a seat that does not answer holds the seed and, after five
// minutes, is the finding call-unanswered (scope call, id <seat>.steps), cleared when it answers.
func TestTheSeedAsksStepsOfTheBuildSeatsOfTheTrunkEntries(t *testing.T) {
ctx := context.Background()
now := time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC)
w := newFindings()
ix := &Indexer{Findings: w, Revisions: newMemObjects(1 << 20), Now: func() time.Time { return now }}
fp := "src2:" + strings.Repeat("a", 64)
entries := map[string]meshcatalog.CatalogEntry{"avahi": {Module: "avahi", Repository: "novox/r", Commit: "c1", Path: "mods/avahi", Fp: fp, Manifest: map[string]any{},
Artifacts: []meshcatalog.Artifact{{Ref: "avahi:program", Language: "go"}, {Ref: "avahi:image", Language: "go", Kind: "image"}},
Inputs: &meshcatalog.EntryInputs{Paths: map[string][]string{"program": {"cmd"}}}}}
rev, objects, err := CatalogRevisionOf(meshcatalog.CatalogRevision{Revision: 3, Kind: "trunk"}, entries)
if err != nil {
t.Fatal(err)
}
if err := KeepRevision(ctx, ix.Revisions, rev, objects); err != nil {
t.Fatal(err)
}
defs := map[string]seatdef.Definition{
"node-build-go": {Acts: []seatdef.ActingRule{{Step: seatdef.StepBuild, On: "go"}}},
"node-build-image": {Acts: []seatdef.ActingRule{{Step: seatdef.StepBuild, On: seatdef.OnKindImage}}},
}
// the image artifact is built by both rules: the language's and the kind's
if _, err := ix.SeedModulesFrom(ctx, defs, nil); err == nil || !strings.Contains(err.Error(), "each build") {
t.Fatalf("two seats building one artifact: %v", err)
}
e := entries["avahi"]
e.Artifacts[1].Language = "dockerfile"
entries["avahi"] = e
rev, objects, _ = CatalogRevisionOf(meshcatalog.CatalogRevision{Revision: 4, Kind: "trunk"}, entries)
if err := KeepRevision(ctx, ix.Revisions, rev, objects); err != nil {
t.Fatal(err)
}
var got []nodebuildgo.StepsInput
goUp := false
call := func(_ context.Context, seat string, in nodebuildgo.StepsInput) (nodebuildgo.StepsAnswer, error) {
got = append(got, in)
if seat == "node-build-go" && !goUp {
return nodebuildgo.StepsAnswer{}, errors.New("no responders")
}
return nodebuildgo.StepsAnswer{Steps: []nodebuildgo.DetectedStep{{Step: "unit"}}}, nil
}
mods, err := ix.SeedModulesFrom(ctx, defs, call)
if err != nil {
t.Fatal(err)
}
if len(mods) != 1 || mods[0].Module != "avahi" || mods[0].Origin != "novox/r" || mods[0].Path != "mods/avahi/module.json" || len(mods[0].Artifacts) != 2 {
t.Fatalf("the seed's modules: %+v", mods)
}
if a := mods[0].Artifacts[0]; a.Answered || a.Seat != "node-build-go" {
t.Errorf("the go artifact is %+v, want unanswered by node-build-go", a)
}
if a := mods[0].Artifacts[1]; !a.Answered || a.Seat != "node-build-image" || len(a.Steps) != 1 {
t.Errorf("the image artifact is %+v", a)
}
in := got[0]
if in.Module != "avahi" || in.Artifact != "avahi.program" || in.Fingerprint != fp || in.Fetch == nil || in.Fetch.Repository != "novox/r" || in.Fetch.Commit != "c1" || in.Fetch.Dir != "mods/avahi" || len(in.Fetch.Paths) != 1 {
t.Errorf("the steps input is %+v", in)
}
if len(w.Held()) != 0 {
t.Fatal("a call unanswered for no time wrote its finding")
}
now = now.Add(5 * time.Minute)
if _, err := ix.SeedModulesFrom(ctx, defs, call); err != nil {
t.Fatal(err)
}
if h := w.Held(); len(h) != 1 || h[0].Scope != mcc.ScopeCall || h[0].ID != "node-build-go.steps" || h[0].Kind != mcc.KindCallUnanswered {
t.Fatalf("the findings after five minutes: %+v", h)
}
goUp = true
if _, err := ix.SeedModulesFrom(ctx, defs, call); err != nil {
t.Fatal(err)
}
if len(w.Held()) != 0 {
t.Errorf("the finding stands after the seat answered: %+v", w.Held())
}
}
modules/mesh-delivery-indexer/manifest_test.go
+1 -1
View File
@@ -172,7 +172,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-hand-acts.hand-act", "seat:mesh-watchdogs.presence"}; !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-checker.checker_tries", "mesh-delivery-forge-bridge.bridge_merged_message", "seat:node-engine.composition", "seat:mesh-delivery.source", "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"}; !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
modules/mesh-delivery-indexer/module.json
+4 -1
View File
@@ -149,7 +149,10 @@
"seat:node-engine.composition",
"seat:mesh-delivery.source",
"seat:mesh-hand-acts.hand-act",
"seat:mesh-watchdogs.presence"
"seat:mesh-watchdogs.presence",
"seat:node-build-go.steps",
"seat:node-build-node.steps",
"seat:node-build-image.steps"
],
"own-secrets": {
"broker": "${dir:state}/broker"