B1 (checker): the checker's own buckets, events and asker are named from its credential's module

CheckerModule and the buckets built on it are gone; check.Bucket names a state of the credential's module, the examiner
takes the module as its asker, the tools take the served-as module, and the tools ask the indexer's read tools of the
holder of the seat mesh-catalog on each call (B2's tool addresses for the checker).

Builds: design 47 G A-296, A-210, §428

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-12 23:35:14 +01:00
parent 07c13b0b9c
commit 1db4728081
18 changed files with 118 additions and 104 deletions
@@ -101,12 +101,25 @@ func ownerRead(ask func(string, any) (json.RawMessage, error), tool string) chec
}}
}
// The indexer's read tools the tools ask for its candidates and modules (the manifest's invokes).
// The indexer's read tools the tools ask for its candidates and modules (the manifest's invokes name them under the indexer's
// module; the call goes to the holder of the seat mesh-catalog, atHolder).
const (
indexerCandidatesTool = check.IndexerModule + ".indexer_read_candidates"
indexerModulesTool = check.IndexerModule + ".indexer_read_modules"
indexerCandidatesTool = "indexer_read_candidates"
indexerModulesTool = "indexer_read_modules"
)
// atHolder asks a tool of the module that holds the seat when it is asked (design 47 G A-210, §428), as the view's tools do:
// the holder is found by the seat list for each call, and a seat with no holder is an error that says so.
func atHolder(ask func(string, any) (json.RawMessage, error), holder func(seat string) (string, bool), seat string) func(string, any) (json.RawMessage, error) {
return func(tool string, body any) (json.RawMessage, error) {
module, ok := holder(seat)
if !ok {
return nil, fmt.Errorf("the holder of %s is not found, so %s is not asked", seat, tool)
}
return ask(module+"."+tool, body)
}
}
// runsOf says which commit the mesh runs of a module from the indexer's modules state, as the checker's own record of it does
// (the commit the module's record was computed at); "" when the record is not readable.
type runsOf struct{ modules check.StateRead }
@@ -464,7 +477,7 @@ func main() {
}
return nil
}, holderOf)
r := check.Reader{VerdictState: read(stdio.State("verdicts")), KeptState: read(stdio.State("kept")), CandidateState: ownerRead(stdio.Ask, indexerCandidatesTool),
r := check.Reader{VerdictState: read(stdio.State("verdicts")), KeptState: read(stdio.State("kept")), CandidateState: ownerRead(atHolder(stdio.Ask, holderOf, meshcatalog.Seat), indexerCandidatesTool),
PreviewState: read(stdio.State("previews")), InputState: read(stdio.State(check.InputsStateName)),
StepStates: func() map[string]check.StateRead {
out := map[string]check.StateRead{}
@@ -478,7 +491,7 @@ func main() {
HaveState: read(stdio.State("node-engine.have")), StepBound: stepBound(),
CouldNotRunState: read(stdio.State(check.CouldNotRunState)), HearingState: read(stdio.State(buildrun.HearingState)),
AskState: read(stdio.State(check.AsksState)), BuildState: read(stdio.State("mesh-delivery-builder.builds"))}
r.Runs = runsOf{ownerRead(stdio.Ask, indexerModulesTool)} // the node-engine at the commit the mesh runs, as the checker reads it
r.Runs = runsOf{ownerRead(atHolder(stdio.Ask, holderOf, meshcatalog.Seat), indexerModulesTool)} // the node-engine at the commit the mesh runs, as the checker reads it
tries := stdio.State(check.TriesState)
trunks := func() (map[string]string, error) {
raw, err := stdio.Ask(check.TrunkVerb, meshcatalog.RevisionInput{})
@@ -518,7 +531,7 @@ func main() {
tools := append(append(append(append(Tools(r), TryTools(trying)...), OpenTriesTools(trying)...), SeatTryTools(trying)...), RecoveryTools(r, written)...)
// checker_first_views and checker_first_view_again: where the checker's first views of the indexer's and the assigner's
// states stand, and asking one again by hand (ADR 0615, DLV-M2).
tools = append(tools, views.StatusTools(check.CheckerModule, "checker_first_views", "checker_first_view_again", stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...)
tools = append(tools, views.StatusTools(os.Getenv(stdio.ServedAs), "checker_first_views", "checker_first_view_again", stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...)
tools = append(tools, gate.Tools()...)
tools = append(tools, SeatInputTools(r)...)
tools = append(tools, RequestTools(read(stdio.State(check.RequestsStateName)))...)
@@ -5,6 +5,7 @@ import (
"encoding/base64"
"encoding/json"
"errors"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"math/rand"
"os"
"regexp"
@@ -123,7 +124,7 @@ func TestTheToolsServedAreTheFiveReadTools(t *testing.T) {
}
}
for _, tool := range []string{indexerCandidatesTool, indexerModulesTool} {
if !slices.Contains(man.Invokes, tool) {
if !slices.Contains(man.Invokes, check.IndexerModule+"."+tool) {
t.Errorf("the manifest does not invoke %s", tool)
}
}
@@ -140,11 +141,12 @@ func TestTheToolsServedAreTheFiveReadTools(t *testing.T) {
} {
t.Run("1966: the indexer's read tool, "+c.name, func(t *testing.T) {
var asked []string
r := ownerRead(func(key string, body any) (json.RawMessage, error) {
holder := func(seat string) (string, bool) { return "acme-indexer", seat == meshcatalog.Seat } // found at the seat list's holder
r := ownerRead(atHolder(func(key string, body any) (json.RawMessage, error) {
b, _ := json.Marshal(body)
asked = append(asked, key+" "+string(b))
return json.RawMessage(c.answer), nil
}, indexerCandidatesTool)
}, holder, meshcatalog.Seat), indexerCandidatesTool)
var got string
var err error
if c.key == "" {
@@ -156,9 +158,9 @@ func TestTheToolsServedAreTheFiveReadTools(t *testing.T) {
raw, err = r.Get(c.key)
got = string(raw)
}
want := []string{indexerCandidatesTool + " {}"}
want := []string{"acme-indexer." + indexerCandidatesTool + " {}"}
if c.key != "" {
want = []string{indexerCandidatesTool + ` {"key":"` + c.key + `"}`}
want = []string{"acme-indexer." + indexerCandidatesTool + ` {"key":"` + c.key + `"}`}
}
if (err != nil) != c.failed || (!c.failed && got != c.want) || !slices.Equal(asked, want) {
t.Errorf("read %q, %v, asking %v; want %q, failed %v, asking %v", got, err, asked, c.want, c.failed, want)
@@ -14,8 +14,6 @@ import (
"github.com/nats-io/nats.go/jetstream"
"mesh-delivery-checker/internal/check"
"git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient"
nec "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
ne "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
@@ -109,10 +107,14 @@ func (p placedNodes) Placed() ([]string, error) {
// seatSandbox runs a compose job on the Go build seat's check stream, as the checker's own ask, and answers the
// composer's verdict from the step's result (buildstep.ComposeAnswer).
type seatSandbox struct{ asker buildrun.Asker }
type seatSandbox struct {
// module is the checker's own, the credential's: the asker of the compose ask.
module string
asker buildrun.Asker
}
func (s seatSandbox) Compose(ctx context.Context, job buildstep.ComposeJob) (composer.Verdict, error) {
ask := buildstep.StepAsk{ID: buildstep.NewID(), Kind: buildstep.AskCheck, Language: buildstep.AskGo, Asker: check.CheckerModule,
ask := buildstep.StepAsk{ID: buildstep.NewID(), Kind: buildstep.AskCheck, Language: buildstep.AskGo, Asker: s.module,
At: time.Now().UTC(), Compose: &job}
step, err := s.asker.Ask(ctx, ask)
if err != nil {
@@ -205,7 +205,7 @@ func askVerb(nc *nats.Conn, key string, body any) (json.RawMessage, error) {
// checker's state inputs), and counts what it answered (issue 1921, ADR 0596 §2).
func (b *bed) serveInputVerb(t *testing.T, nc *nats.Conn) {
t.Helper()
kv := b.buckets[check.InputsBucket]
kv := b.buckets[testBucket(check.InputsStateName)]
r := check.Reader{InputState: check.StateRead{Get: func(key string) (json.RawMessage, error) {
e, err := kv.Get(b.ctx, key)
if errors.Is(err, jetstream.ErrKeyNotFound) {
@@ -586,14 +586,14 @@ func startBed(t *testing.T) *bed {
for _, s := range m.Streams {
var subjects []string
for _, e := range s.Events {
subjects = append(subjects, process.ModuleStreamSubject(check.CheckerModule, strings.ReplaceAll(e, "**", ">")))
subjects = append(subjects, process.ModuleStreamSubject(testChecker, strings.ReplaceAll(e, "**", ">")))
}
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: s.BusName(), Subjects: subjects}); err != nil {
t.Fatal(err)
}
}
if !slices.ContainsFunc(m.Streams, func(s catalogue.StreamDeclaration) bool { return s.Name == "checks" }) ||
meshdeliverychecker.HolderSubjectVerdict(check.CheckerModule) != process.ModuleStreamSubject(check.CheckerModule, "verdict") {
meshdeliverychecker.HolderSubjectVerdict(testChecker) != process.ModuleStreamSubject(testChecker, "verdict") {
t.Fatalf("the checker's manifest declares the streams %+v, want checks, and asks at the definition's subjects", m.Streams)
}
// The shared events stream with the seats' and modules' events (the package registry seats' published among them), and the checker's durable consumer of it.
@@ -603,11 +603,11 @@ func startBed(t *testing.T) *bed {
if _, err := js.CreateOrUpdateConsumer(ctx, process.EventsStream, jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-checker", AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
t.Fatal(err)
}
for name, history := range map[string]int{check.CheckerVerdictsBucket: 5, check.CheckerKeptBucket: 1,
check.CheckerOwedBucket: 5, check.PreviewsBucket: 5, check.CandidatesBucket: 10,
for name, history := range map[string]int{testBucket(meshdeliverychecker.StateVerdicts): 5, testBucket(meshdeliverychecker.StateKept): 1,
testBucket(meshdeliverychecker.StateOwed): 5, testBucket(meshdeliverychecker.StatePreviews): 5, check.CandidatesBucket: 10,
check.ModulesBucket: 20, check.ManifestsBucket: 1,
check.PresenceBucket: 20, check.HaveBucket: 20, check.TriesBucket: 1,
builds.BuildsBucket(builds.BuilderModule): 1, check.InputsBucket: 1} {
check.PresenceBucket: 20, check.HaveBucket: 20, testBucket(check.TriesState): 1,
builds.BuildsBucket(builds.BuilderModule): 1, testBucket(check.InputsStateName): 1} {
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: uint8(history)})
if err != nil {
t.Fatal(err)
@@ -617,7 +617,7 @@ func startBed(t *testing.T) *bed {
// The indexer and the assigner as the owners of the states the checker follows: their read tools and views families
// (ADR 0615, DLV-M2).
for _, st := range []string{candidatesState, modulesState, manifestsState, meshcatalog.StatePackages, check.IndexerPredefinedState, seatDefinitionsState} {
viewstest.Serve(t, ctx, js, ctl, check.IndexerModule, st, check.IndexerModule+".indexer_read_"+strings.ReplaceAll(st, "-", "_"), "n1_"+check.CheckerModule)
viewstest.Serve(t, ctx, js, ctl, check.IndexerModule, st, check.IndexerModule+".indexer_read_"+strings.ReplaceAll(st, "-", "_"), "n1_"+testChecker)
}
// The catalog half's registrations and the gate's source branch (design 47 G §614), and the claimant's seat-holders view naming
// testAssigner the holder of mesh-module-assigner (§717): nothing answers the seat's verb assignments.
@@ -630,11 +630,11 @@ func startBed(t *testing.T) *bed {
}
}
}
seatHolders := viewstest.Serve(t, ctx, js, ctl, testAssigner, meshmoduleassigner.StateSeatHolders, testAssigner+"."+meshmoduleassigner.StateSeatHolders, "n1_"+check.CheckerModule)
seatHolders := viewstest.Serve(t, ctx, js, ctl, testAssigner, meshmoduleassigner.StateSeatHolders, testAssigner+"."+meshmoduleassigner.StateSeatHolders, "n1_"+testChecker)
if _, err := seatHolders.Put(ctx, meshmoduleassigner.Seat, []byte(`{"seat":"`+meshmoduleassigner.Seat+`","holders":[{"module":"`+testAssigner+`","node":"n1"}]}`)); err != nil {
t.Fatal(err)
}
viewstest.Serve(t, ctx, js, ctl, testAssigner, meshmoduleassigner.StateAssignments, testAssigner+"."+meshmoduleassigner.StateAssignments, "n1_"+check.CheckerModule)
viewstest.Serve(t, ctx, js, ctl, testAssigner, meshmoduleassigner.StateAssignments, testAssigner+"."+meshmoduleassigner.StateAssignments, "n1_"+testChecker)
// 1929/AC1: the indexer's value of the repository's module, with no TypeScript manifest, so the Go build module alone
// judges the repository (an unknown repository is judged by both, issue 1887 A12).
known, err := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Repository: repo, Commit: trunk, Path: "modules/m", None2: "the fingerprint is computed with its chain"})
@@ -655,8 +655,8 @@ func startBed(t *testing.T) *bed {
b.serveForge(t, ctl)
b.serveInputVerb(t, ctl)
b.fake = &fakeBuildModules{js: js, ctx: ctx, calls: map[string]int{}}
b.fake.serve(t, goBuildModule, b.buckets[check.CheckerKeptBucket])
b.fake.serve(t, nodeBuildModule, b.buckets[check.CheckerKeptBucket])
b.fake.serve(t, goBuildModule, b.buckets[testBucket(meshdeliverychecker.StateKept)])
b.fake.serve(t, nodeBuildModule, b.buckets[testBucket(meshdeliverychecker.StateKept)])
b.report(t, goBuildModule, "n1", "go", "go1.27.1", "1")
b.report(t, nodeBuildModule, "n1", "typescript", "tsc5", "1")
b.setTrunk(t, map[string]string{repo: trunk})
@@ -776,7 +776,7 @@ func (b *bed) verdict(t *testing.T, key string, pred func(meshdeliverychecker.Ve
t.Helper()
var last meshdeliverychecker.VerdictsByGroupHeadsValue
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
e, err := b.buckets[check.CheckerVerdictsBucket].Get(b.ctx, key)
e, err := b.buckets[testBucket(meshdeliverychecker.StateVerdicts)].Get(b.ctx, key)
if err != nil {
continue
}
@@ -801,7 +801,7 @@ func (b *bed) owed(t *testing.T, number int, head, state string) meshdeliveryche
}
var last meshdeliverychecker.OwedByRepositoryNumberHeadActValue
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
e, err := b.buckets[check.CheckerOwedBucket].Get(b.ctx, key)
e, err := b.buckets[testBucket(meshdeliverychecker.StateOwed)].Get(b.ctx, key)
if err != nil {
continue
}
@@ -918,13 +918,13 @@ func TestCheckerJudgesACandidateAndOwesItsStatus(t *testing.T) { // N6/AC5, 1300
}
// What passed is kept once under each level's digest.
for _, k := range []string{"unit." + digU, "integration." + digI} {
if _, err := b.buckets[check.CheckerKeptBucket].Get(b.ctx, k); err != nil {
if _, err := b.buckets[testBucket(meshdeliverychecker.StateKept)].Get(b.ctx, k); err != nil {
t.Errorf("kept.%s: %v", k, err)
}
}
// The preview of the head merged onto the trunk, with the tree and the manifest digest the fetch named.
pk, _ := meshdeliverycheckerclient.PreviewKey(repo, head1, trunk)
e, err := b.buckets[check.PreviewsBucket].Get(b.ctx, pk)
e, err := b.buckets[testBucket(meshdeliverychecker.StatePreviews)].Get(b.ctx, pk)
if err != nil {
t.Fatalf("previews.%s: %v", pk, err)
}
@@ -979,7 +979,7 @@ func TestACheckOrACarryOfEqualInputsRunsNothing(t *testing.T) { // 1287/AC5, 128
func keptCount(t *testing.T, b *bed) int {
t.Helper()
keys, err := b.buckets[check.CheckerKeptBucket].Keys(b.ctx)
keys, err := b.buckets[testBucket(meshdeliverychecker.StateKept)].Keys(b.ctx)
if err != nil {
return 0
}
@@ -1027,7 +1027,7 @@ func TestACheckThatCannotBeJudgedFailsNamingWhy(t *testing.T) { // 1289/AC4, 192
}
if !row.tests {
pk, _ := meshdeliverycheckerclient.PreviewKey(repo, head1, trunk)
if e, err := b.buckets[check.PreviewsBucket].Get(b.ctx, pk); err != nil {
if e, err := b.buckets[testBucket(meshdeliverychecker.StatePreviews)].Get(b.ctx, pk); err != nil {
t.Errorf("previews.%s: %v", pk, err)
} else if pv, err := meshdeliverycheckerclient.DecodePreview(e.Value()); err != nil || strings.Join(pv.Conflicts, ",") != "a.go,sub/b.go" || pv.Merge != "" {
t.Errorf("preview = %+v, %v, want the files in conflict alone", pv, err)
@@ -1050,7 +1050,7 @@ func TestAnHqPullRequestGetsAVerdictAndAStatus(t *testing.T) {
}
key, _ := check.CheckerOwedKey("novox/hq", 3, head1)
for deadline := time.Now().Add(20 * time.Second); ; time.Sleep(50 * time.Millisecond) {
if e, err := b.buckets[check.CheckerOwedBucket].Get(b.ctx, key); err == nil {
if e, err := b.buckets[testBucket(meshdeliverychecker.StateOwed)].Get(b.ctx, key); err == nil {
if o, err := check.DecodeCheckerOwed(e.Value()); err == nil && o.State == "success" {
return
}
@@ -1071,7 +1071,7 @@ func TestACheckLeftRunningIsTakenOverWhenTheCheckerStarts(t *testing.T) { // N6/
if err != nil {
t.Fatal(err)
}
if _, err := b.buckets[check.CheckerVerdictsBucket].Create(b.ctx, key, left); err != nil {
if _, err := b.buckets[testBucket(meshdeliverychecker.StateVerdicts)].Create(b.ctx, key, left); err != nil {
t.Fatal(err)
}
b.start(t, nil)
@@ -1112,7 +1112,7 @@ func TestAMemberNamingAMeshSDKCommitOffTheTrunkIsJudgedAndNotRefused(t *testing.
// DLV-C2, DLV-C6 (design 26 *Out of step*, ADR 0620): each check's end says one verdict per pull request on the stream
// asks, at the definition's subject and in its body (decoded strictly): the lone pull request's, then the group's two.
verdicts, err := b.js.OrderedConsumer(b.ctx, "ASKS", jetstream.OrderedConsumerConfig{FilterSubjects: []string{meshdeliverychecker.HolderSubjectVerdict(check.CheckerModule)}})
verdicts, err := b.js.OrderedConsumer(b.ctx, "ASKS", jetstream.OrderedConsumerConfig{FilterSubjects: []string{meshdeliverychecker.HolderSubjectVerdict(testChecker)}})
if err != nil {
t.Fatal(err)
}
@@ -1177,7 +1177,7 @@ func TestATryIsJudgedAndItsEndWrittenWithNoStatusOwed(t *testing.T) {
b := startBed(t)
asked := b.engine(t, "n1") // a whole try is composed on every placed node, as a group's check is
b.start(t, nil)
store := triesStore{b.buckets[check.TriesBucket], b.ctx}
store := triesStore{b.buckets[testBucket(check.TriesState)], b.ctx}
id, err := check.AskTry(store, map[string]string{repo: trunk}, check.TryAsk{Repository: repo, Head: head1}, time.Now())
if err != nil {
t.Fatal(err)
@@ -1208,7 +1208,7 @@ func TestATryIsJudgedAndItsEndWrittenWithNoStatusOwed(t *testing.T) {
}
for _, event := range []string{checkermsg.Started, checkermsg.Ended} { // the end is said once the try's end is written
said, err := b.js.OrderedConsumer(b.ctx, checks.CachedInfo().Config.Name, jetstream.OrderedConsumerConfig{
FilterSubjects: []string{process.ModuleStreamSubject(check.CheckerModule, event)}})
FilterSubjects: []string{process.ModuleStreamSubject(testChecker, event)}})
if err != nil {
t.Fatal(err)
}
@@ -1216,7 +1216,7 @@ func TestATryIsJudgedAndItsEndWrittenWithNoStatusOwed(t *testing.T) {
t.Errorf("the stream checks holds no %s of the try %s: %v", event, id, err)
}
}
for _, name := range []string{check.CheckerOwedBucket, check.CheckerVerdictsBucket} {
for _, name := range []string{testBucket(meshdeliverychecker.StateOwed), testBucket(meshdeliverychecker.StateVerdicts)} {
if keys, _ := b.buckets[name].Keys(b.ctx); len(keys) != 0 {
t.Errorf("a try wrote %s: %v", name, keys)
}
@@ -1296,6 +1296,11 @@ func TestTheCheckerSaysOwedReadableOnTheBus(t *testing.T) { // R2-1
b.said(t, "holds "+testAssigner+"'s assignments, the holder of the seat mesh-module-assigner")
}
// testChecker is the module the bed's credential names; the checker's own buckets and events are named from it (design 47 G A-296).
const testChecker = "mesh-delivery-checker"
func testBucket(state string) string { return check.Bucket(testChecker, state) }
const testAssigner = "acme-assigner" // a holder of the seat mesh-module-assigner no code names
// buildSeatDefinitions are the seat definitions the bed's seat mesh-catalog lists: the build seats, as the mesh's own definitions of
@@ -32,6 +32,8 @@ import (
// judgeDeps are the parts the judge of a check is made of.
type judgeDeps struct {
// module is the checker's own, the credential's (check.Examiner.Module).
module string
// inputs keeps the checking program's inputs (check.InputsStateName, served on check.InputVerb).
inputs check.Inputs
// beginBound is how long a run ask may wait for a build node to begin it; check.DefaultBeginBound when 0.
@@ -235,7 +237,7 @@ func newJudge(d judgeDeps) check.Judge {
if d.nodes != nil {
nodes = d.nodes
}
ex := &check.Examiner{Nodes: nodes, Compositions: d.compositions, Asks: d.asks, Toolchains: d.tools,
ex := &check.Examiner{Module: d.module, Nodes: nodes, Compositions: d.compositions, Asks: d.asks, Toolchains: d.tools,
Kept: check.Kept{Store: d.kept}, Previews: d.previews, Candidates: d.candidates, Modules: d.repos, Packages: d.repos, Published: d.published, KnownPackages: d.published, Forge: d.forge, Runs: d.repos, Builds: d.builds, BuildEnds: d.ends, BuildWait: d.buildWait, Announce: d.announce, Trunks: d.trunks, Manifests: d.repos, Read: d.repos, Dirs: d.repos, Predefined: d.predefined, Renames: d.renames, SeatDefinitions: d.seatDefs, SeatList: d.seats, Seats: d.buildSeats, Inputs: d.inputs, BeginBound: d.beginBound, Logf: d.logf}
if d.nodes != nil && d.builds != nil && d.repos != nil {
ex.Serving = runningServes{have: d.nodes, builds: d.builds, read: d.repos, logf: d.logf}
@@ -10,6 +10,7 @@ import (
"encoding/json"
"errors"
"fmt"
checker "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdeliverychecker"
"io"
"log"
"os"
@@ -42,8 +43,8 @@ import (
// eventSubject is the subject of an event of the checker's stream checks (the bus names the stream CHECKS): the indexer's
// watch of the previews bucket and the forge bridge's watch of the owed bucket are woken by them.
func eventSubject(event string) string {
return process.ModuleStreamSubject(check.CheckerModule, event)
func eventSubject(module, event string) string {
return process.ModuleStreamSubject(module, event)
}
const (
@@ -76,9 +77,7 @@ func durationEnv(getenv func(string) string, name string, def time.Duration) (ti
// The checker's own states of its first views (views.StatusState) and of its findings, where the condition keeper raises a
// first view unanswered past its bound (ADR 0615 §6); and the indexer's states it follows (check.CandidatesBucket,
// ModulesBucket, ManifestsBucket).
var (
firstViewsBucket = check.CheckerModule + "_" + views.StatusState
)
func firstViewsBucket(module string) string { return check.Bucket(module, views.StatusState) }
const (
candidatesState = "candidates"
@@ -134,9 +133,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
defer conn.Close()
// Its own states: a check without one of them would be unwritten, so each stops the start, naming its grant (issue 1945).
buckets := map[string]jetstream.KeyValue{}
for _, name := range []string{check.CheckerVerdictsBucket, check.CheckerKeptBucket, check.CheckerOwedBucket, check.PreviewsBucket,
check.TriesBucket} {
if buckets[name], err = process.EssentialBucket(ctx, js, "checker", name, "judge or keep a check"); err != nil {
own := func(state string) string { return check.Bucket(cred.Module, state) } // the credential's module, never a constant (A-296)
for _, state := range []string{checker.StateVerdicts, checker.StateKept, checker.StateOwed, checker.StatePreviews, check.TriesState} {
if buckets[state], err = process.EssentialBucket(ctx, js, "checker", own(state), "judge or keep a check"); err != nil {
return err
}
}
@@ -160,7 +159,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// The module has one durable consumer of EVENTS: one hub pulls it for the views reader and for the registry seats'
// published, started once both are registered.
hub := process.NewEventHub(logger.Printf)
reader := &views.ProcessReader{Module: check.CheckerModule, Call: caller.Call, Logf: logger.Printf, Hub: hub}
reader := &views.ProcessReader{Module: cred.Module, Call: caller.Call, Logf: logger.Printf, Hub: hub}
var buildsChanged buildsWatch
seatPart := seatlist.NewPart("checker", nil)
seats := seatList{reader: seatPart.Follow(seatlist.ProcessBus(caller.Call, hub), buildsChanged.Changed, logger.Printf)}
@@ -232,7 +231,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
dead := &buildrun.DeadLetters{}
// The state may not be on the bus yet when a delivery started the checker (the grants part creates it, issue 1790): it
// is bound when the bus says it was created, and the latest hearing is written then.
hearingKV := process.AwaitBucket(ctx, js, conn, check.HearingBucket, "checker_status cannot say whether DEAD_LETTERS is heard", logger.Printf)
hearingKV := process.AwaitBucket(ctx, js, conn, own(buildrun.HearingState), "checker_status cannot say whether DEAD_LETTERS is heard", logger.Printf)
hearing := buildrun.NewDeadLetterHearing(ctx, buildrun.JetStreamDeadLetters{JS: js}, []string{buildstep.AskGo, buildstep.AskNode}, dead,
logger.Printf, buildrun.SayHearing(ctx, hearingKV, logger.Printf))
defer hearing.Stop()
@@ -241,14 +240,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
heard := buildrun.HearEnds(ctx, asks, asks, buildSeats.Modules(), logger.Printf)
conn.SetReconnectHandler(func(*nats.Conn) { hearing.Again(); heard.Nudge("") })
waiter := buildrun.LostAskWaiter{Inner: heard, Dead: dead, Hearing: hearing}
asksKV := process.AwaitBucket(ctx, js, conn, check.AsksBucket, "checker_ask_again refuses every check, as it cannot tell whether its ask waits in a stream", logger.Printf)
inputsKV := process.AwaitBucket(ctx, js, conn, check.InputsBucket, "no checking program can be asked, so every check fails naming it", logger.Printf)
couldNotRunKV := process.AwaitBucket(ctx, js, conn, check.CouldNotRunBucket, "a check that could not run is not asked again when its part runs", logger.Printf)
requestsKV := process.AwaitBucket(ctx, js, conn, check.RequestsBucket, "no composition request is kept, so every composition made with the node-engine's head fails naming it", logger.Printf)
asksKV := process.AwaitBucket(ctx, js, conn, own(check.AsksState), "checker_ask_again refuses every check, as it cannot tell whether its ask waits in a stream", logger.Printf)
inputsKV := process.AwaitBucket(ctx, js, conn, own(check.InputsStateName), "no checking program can be asked, so every check fails naming it", logger.Printf)
couldNotRunKV := process.AwaitBucket(ctx, js, conn, own(check.CouldNotRunState), "a check that could not run is not asked again when its part runs", logger.Printf)
requestsKV := process.AwaitBucket(ctx, js, conn, own(check.RequestsStateName), "no composition request is kept, so every composition made with the node-engine's head fails naming it", logger.Printf)
ch := &check.Checker{
Asks: awaitedKV{asksKV},
Verdicts: kvStore{buckets[check.CheckerVerdictsBucket]},
Owed: kvStore{buckets[check.CheckerOwedBucket]},
Verdicts: kvStore{buckets[checker.StateVerdicts]},
Owed: kvStore{buckets[checker.StateOwed]},
// A check that could not run waits for its part's running, and is the finding check-could-not-run after its bound.
CouldNotRun: awaitedKV{couldNotRunKV},
Trunks: trunks,
@@ -294,22 +293,22 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
Asker: engineAsks{call: caller.CallOn},
Placed: placedNodes{kv: assignedNodes(&assignments)}.Placed,
Bound: answerBound,
Sandbox: seatSandbox{asker: buildrun.Asker{Seats: buildSeats, Publisher: asks, Waiter: waiter, Bound: stepBound, Dropper: buildrun.SeatDrops{Caller: caller}, Outputs: caller.CallFile}},
Sandbox: seatSandbox{module: cred.Module, asker: buildrun.Asker{Seats: buildSeats, Publisher: asks, Waiter: waiter, Bound: stepBound, Dropper: buildrun.SeatDrops{Caller: caller}, Outputs: caller.CallFile}},
Verdicts: verdicts.all,
// The node-engine's trunk head: a group with no node-engine member composes with the composer built there (ADR 0656 §5).
Composer: func() string { return trunks.Heads()[check.NodeEngineRepository] },
}
// The tries (ADR 0501 §5) are judged by the same judge, held by the same deliveries.
tryer := &check.Tryer{Tries: kvStore{buckets[check.TriesBucket]}, Bound: checkBound, NotTaken: stepBound, Logf: logger.Printf,
tryer := &check.Tryer{Tries: kvStore{buckets[check.TriesState]}, Bound: checkBound, NotTaken: stepBound, Logf: logger.Printf,
// A whole try's start and end are said on the stream checks, for the indexer to write and delete the candidate of its group.
Say: func(ctx context.Context, event string, body checkermsg.Event) error {
data, _ := json.Marshal(body)
_, err := js.Publish(ctx, eventSubject(event), data, jetstream.WithRetryAttempts(10), jetstream.WithRetryWait(time.Second))
_, err := js.Publish(ctx, eventSubject(cred.Module, event), data, jetstream.WithRetryAttempts(10), jetstream.WithRetryWait(time.Second))
return err
}}
announce := func(ctx context.Context, event, key string) {
body, _ := json.Marshal(map[string]string{"key": key})
if _, err := js.Publish(ctx, eventSubject(event), body); err != nil {
if _, err := js.Publish(ctx, eventSubject(cred.Module, event), body); err != nil {
logger.Printf("%s of %s was written and the part that reads it could not be told: %v", event, key, err)
}
}
@@ -319,11 +318,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// once the first views are answered, before any check starts (afterFirstViews).
published := &publishedPackages{}
compositions.Published = published
ch.Runs = repos // the node-engine a check reads beside at its running commit (ADR 0548 §6)
ch.Says = verdictSays{js: js, node: cred.Node, now: time.Now} // a pull request's verdict event, on the stream asks (design 26 *Out of step*)
ch.Runs = repos // the node-engine a check reads beside at its running commit (ADR 0548 §6)
ch.Says = verdictSays{module: cred.Module, js: js, node: cred.Node, now: time.Now} // a pull request's verdict event, on the stream asks (design 26 *Out of step*)
ch.Judge = newJudge(judgeDeps{
asks: buildrun.Asker{Seats: buildSeats, Publisher: asks, Waiter: waiter, Bound: stepBound, Dropper: buildrun.SeatDrops{Caller: caller}, Outputs: caller.CallFile}, tools: tools, trunks: trunks,
kept: kvStore{buckets[check.CheckerKeptBucket]}, previews: kvStore{buckets[check.PreviewsBucket]}, candidates: tryCandidates{candidates},
module: cred.Module,
asks: buildrun.Asker{Seats: buildSeats, Publisher: asks, Waiter: waiter, Bound: stepBound, Dropper: buildrun.SeatDrops{Caller: caller}, Outputs: caller.CallFile}, tools: tools, trunks: trunks,
kept: kvStore{buckets[checker.StateKept]}, previews: kvStore{buckets[checker.StatePreviews]}, candidates: tryCandidates{candidates},
repos: repos, published: published, builds: buildStates, ends: ends, buildWait: set.buildWait, forge: forgeReads{call: caller.Call}, logf: logger.Printf,
announce: announce, compositions: compositions, inputs: runInputs{awaitedKV{inputsKV}}, beginBound: beginBound,
nodes: openNodeNames(ctx, js, logger.Printf),
@@ -472,7 +472,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return err
}
// The tries: a queued one starts, one left running is taken over, an ended one past its keeping is removed.
tw, err := settle(ctx, buckets[check.TriesBucket], func(jetstream.KeyValueEntry) {})
tw, err := settle(ctx, buckets[check.TriesState], func(jetstream.KeyValueEntry) {})
if err != nil {
return err
}
@@ -486,7 +486,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}()
// The checker's own verdicts, updates only, in one watch: one checker_ask_again wrote asks its check again
// (Checker.VerdictWritten); a deleted key of heads a candidate holds is judged again (Checker.VerdictDeleted, issue 1733).
vw, err := buckets[check.CheckerVerdictsBucket].WatchAll(ctx, jetstream.UpdatesOnly())
vw, err := buckets[checker.StateVerdicts].WatchAll(ctx, jetstream.UpdatesOnly())
if err != nil {
return fmt.Errorf("the checker's verdicts cannot be watched, so no check can be asked again by hand and a deleted verdict would never be written again: %w", err)
}
@@ -502,7 +502,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
go func() { errs <- follow(ctx, w, each) }()
return nil
}
reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "checker_first_views reads nothing and checker_first_view_again is not heard", logger.Printf))
reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket(cred.Module), "checker_first_views reads nothing and checker_first_view_again is not heard", logger.Printf))
// Every part has registered. Where the consumer is not there yet the hub binds it as soon as it is, and the seat list is read
// again then if it is still unread.
if err := hub.StartUntilBound(ctx, js, cred, nil, seatPart.EventsLive); err != nil {
@@ -16,9 +16,11 @@ import (
// verdictSays is check.VerdictSayer on the bus: verdict published on the checker's own stream asks.
type verdictSays struct {
js jetstream.JetStream
node string
now func() time.Time
// module is the checker's own, the credential's: the stream of its events and their source.
module string
js jetstream.JetStream
node string
now func() time.Time
}
// Say publishes one verdict on the checker's own stream asks with the checker's event headers; the message id makes a
@@ -28,10 +30,10 @@ func (t verdictSays) Say(ctx context.Context, v check.VerdictEvent) error {
if err != nil {
return err
}
msg := nats.NewMsg(eventSubject(check.EventVerdict))
msg := nats.NewMsg(eventSubject(t.module, check.EventVerdict))
id := fmt.Sprintf("verdict.%s.%d.%s.%s.%x", v.Repository, v.Pull, v.Head, v.Verdict, sha256.Sum256([]byte(strings.Join(v.OutOfStep, "\n"))))
msg.Header.Set("x-event-id", id)
msg.Header.Set("x-source", check.CheckerModule)
msg.Header.Set("x-source", t.module)
msg.Header.Set("x-node", t.node)
msg.Header.Set("x-time", t.now().UTC().Format(time.RFC3339Nano))
msg.Header.Set("content-type", "application/json")
@@ -24,10 +24,9 @@ import (
// An ask is not put again under its own id instead: the stream de-duplicates an id within its window, so the copy waiting
// stays and the new one is dropped, and past the window both would run; nothing takes a put ask back (buildrun.Asker).
// AsksState is the checker's state of the asks each running check put, by verdict key (CheckAsks); AsksBucket its bucket.
// AsksState is the checker's state of the asks each running check put, by verdict key (CheckAsks).
const (
AsksState = "asks"
AsksBucket = CheckerModule + "_" + AsksState
AsksState = "asks"
)
// CheckAsks are the asks a check put, in order, as the state asks keeps them while it runs; its end deletes the key.
@@ -26,9 +26,6 @@ import (
// CouldNotRunState is the checker's own state of its checks that could not run, by verdict key.
const CouldNotRunState = "could-not-run"
// CouldNotRunBucket is the bucket of CouldNotRunState.
const CouldNotRunBucket = CheckerModule + "_" + CouldNotRunState
// CouldNotRunKind is the checker's finding of a check that could not run and was not asked again within CouldNotRunBound.
const CouldNotRunKind = "check-could-not-run"
@@ -280,7 +280,7 @@ func newBench(t testing.TB) *bench {
}
now := time.Date(2026, 10, 21, 19, 0, 0, 0, time.UTC)
b.asks = &buildrun.FakeAsks{Answer: b.answer}
b.ex = &Examiner{Asks: buildrun.Asker{Seats: testSeats(), Publisher: b.asker(nil, nil), Waiter: b.asks, Bound: 5 * time.Second}, Toolchains: goEntries("shanks"), Kept: Kept{Store: b.kept}, Previews: b.prev,
b.ex = &Examiner{Module: "acme-checker", Asks: buildrun.Asker{Seats: testSeats(), Publisher: b.asker(nil, nil), Waiter: b.asks, Bound: 5 * time.Second}, Toolchains: goEntries("shanks"), Kept: Kept{Store: b.kept}, Previews: b.prev,
Modules: langs{}, Seats: testSeats(), Trunks: &Trunks{heads: map[string]string{DeliveryRepository: trunkD}}, Inputs: memInputs{b.inputs}, Now: func() time.Time { now = now.Add(50 * time.Millisecond); return now }}
return b
}
@@ -1184,7 +1184,7 @@ func TestACheckIsOneAskAndAReJudgeGoesOnTheRecheckStream(t *testing.T) {
b := newBench(t)
b.judge(t, caseOf(t, pullOf(repoA, 7, head1)))
asks := b.asks.Asks()
if len(asks) != 1 || asks[0].Kind != buildstep.AskCheck || asks[0].Language != buildstep.AskGo || asks[0].Asker != CheckerModule {
if len(asks) != 1 || asks[0].Kind != buildstep.AskCheck || asks[0].Language != buildstep.AskGo || asks[0].Asker != "acme-checker" {
t.Fatalf("asks = %+v, want one Go check ask", asks)
}
if j := asks[0].Check; j == nil || !j.Kept || strings.Join(j.Levels, ",") != "unit,integration" || j.Fetch.Repository != repoA {
@@ -71,7 +71,9 @@ type BuildModules interface {
// every level and member runs to its end also after a failure, so the verdict names every failure (the operator,
// 2026-11-02). It judges the step's answer as it judged the tools' answers.
type Examiner struct {
Asks Asks
// Module is the checker's own module, the credential's: the asker of its asks and the owner of its buckets (design 47 G A-296).
Module string
Asks Asks
// DeclineBound is how long a placed ask may go on being declined, from its first decline, before the check fails
// saying no node with its need took it (issue 890, ADR 0552); DefaultDeclineBound when zero.
DeclineBound time.Duration
@@ -549,7 +551,7 @@ func (st *state) judgeAll(ctx context.Context) {
return false
}
if !st.failedSince(met) { // an ask that ended with the context said it already
st.fail(CheckerModule, ctx.Err().Error())
st.fail(st.j.Module, ctx.Err().Error())
}
return true
}
@@ -735,7 +737,7 @@ func (st *state) prepareAsk(ctx context.Context, t *target, merged *[]Fetched, l
if st.c.Kind == KindTrunk {
kind = buildstep.AskRecheck // a re-judge after a trunk move goes after every new head's check (ADR 0536 §4)
}
a := buildstep.StepAsk{ID: buildstep.NewID(), Kind: kind, Language: st.j.askLanguageOf(t.module), Asker: CheckerModule,
a := buildstep.StepAsk{ID: buildstep.NewID(), Kind: kind, Language: st.j.askLanguageOf(t.module), Asker: st.j.Module,
At: st.j.now().UTC(), Check: job, Needs: needs}
if a.Language == buildstep.AskGo {
a.Packages = st.packagesOf(p.Repository)
@@ -34,7 +34,3 @@ func IsLostAsk(v meshdeliverychecker.VerdictsByGroupHeadsValue) bool {
func IsNotTaken(v meshdeliverychecker.VerdictsByGroupHeadsValue) bool {
return v.State == meshdeliverycheckerclient.VerdictFail && strings.Contains(failedOf(v).Words, buildrun.NotTakenWords)
}
// HearingBucket is the checker's state hearing: whether it hears DEAD_LETTERS (buildrun.Hearing under
// buildrun.HearingKey), which checker_status says (dead_letters).
const HearingBucket = CheckerModule + "_" + buildrun.HearingState
@@ -162,7 +162,7 @@ func (st *state) runJudge(ctx context.Context, req composer.Request, touched []s
}
job := buildstep.ComposeJob{Fetch: buildstep.Fetch{Repository: DeliveryRepository, Commit: commit, Packages: versions}, Dir: JudgeDir,
Package: JudgePackage, Request: raw, Reader: "the indexer's judge at " + at}
a := buildstep.StepAsk{ID: buildstep.NewID(), Kind: buildstep.AskCheck, Language: buildstep.AskGo, Asker: CheckerModule,
a := buildstep.StepAsk{ID: buildstep.NewID(), Kind: buildstep.AskCheck, Language: buildstep.AskGo, Asker: st.j.Module,
At: st.j.now().UTC(), Compose: &job}
st.c.Progress.Set(fmt.Sprintf("the judge ask %s of the manifests of %s", a.ID, strings.Join(touched, ", ")), "")
st.c.Asked(a.ID)
@@ -11,13 +11,6 @@ import (
// own constants, each a state of a seat (the seat's state name beside the holder's module name) and each declared in its
// manifest (module.json: its own states, its `reads`); a unit row checks them against the manifest (TestTheNamesAreTheManifests).
const (
// CheckerModule is this part's module; its states verdicts, kept, owed and previews are the seat mesh-delivery-checker's.
CheckerModule = "mesh-delivery-checker"
CheckerVerdictsBucket = CheckerModule + "_" + checker.StateVerdicts
CheckerKeptBucket = CheckerModule + "_" + checker.StateKept
CheckerOwedBucket = CheckerModule + "_" + checker.StateOwed
PreviewsBucket = CheckerModule + "_" + checker.StatePreviews
// GateModule holds the seat mesh-delivery's states source and turns, which the checker reads (reads: mesh-delivery-gate.source,
// mesh-delivery-gate.turns).
GateModule = "mesh-delivery-gate"
@@ -84,3 +77,7 @@ func failedList(items []checker.VerdictsByGroupHeadsValueFailuresItem) []FailedF
}
return out
}
// Bucket is the bucket of one of the checker's own states, `<module>_<state>`, module its credential's, never a constant
// (design 47 G A-296, §428).
func Bucket(module, state string) string { return module + "_" + state }
@@ -32,7 +32,6 @@ const (
// as its manifest's invokes grants it and checks the digest (buildstep.RunInput.Read), never reading the state (ADR 0596
// §2; the operator's answer on milestone 47/B0).
InputsStateName = "inputs"
InputsBucket = CheckerModule + "_" + InputsStateName
InputVerb = meshdeliverycheckerclient.InputVerb
// CheckingProgramDir and CheckingProgramPackage are where in mesh-delivery the checking program is built.
CheckingProgramDir = "modules/mesh-delivery-checker"
@@ -101,7 +100,7 @@ func (st *state) runProgram(ctx context.Context, args []string, body []byte) (js
job := buildstep.RunJob{Program: prog, Heads: st.runHeads(), Input: in, Args: args, Reader: "the checking program at " + at}
var ids []string
for range 2 {
a := buildstep.StepAsk{ID: buildstep.NewID(), Kind: buildstep.AskCheck, Language: buildstep.AskGo, Asker: CheckerModule,
a := buildstep.StepAsk{ID: buildstep.NewID(), Kind: buildstep.AskCheck, Language: buildstep.AskGo, Asker: st.j.Module,
At: st.j.now().UTC(), Packages: st.packagesOf(DeliveryRepository), Run: &job}
ids = append(ids, a.ID)
st.c.Progress.Set(fmt.Sprintf("the run ask %s of the checking program %s", a.ID, strings.Join(args, " ")), "")
@@ -34,7 +34,6 @@ import (
const (
// RequestsStateName is the checker's state of kept requests and RequestsBucket its bucket.
RequestsStateName = "compositions"
RequestsBucket = CheckerModule + "_" + RequestsStateName
// RequestsReadTool is the tool that reads a kept request.
RequestsReadTool = "checker_composition_requests"
// RequestsDropTool is the tool that drops a check's kept requests by hand (a state that is full is freed with it).
@@ -37,8 +37,7 @@ const (
// TriesTool lists the tries that wait or run (OpenTries).
TriesTool = "checker_tries"
// TriesState is the checker's state of tries: the bucket mesh-delivery-checker_tries, keys try-<unix nanoseconds>.
TriesState = "tries"
TriesBucket = CheckerModule + "_" + TriesState
TriesState = "tries"
// MaxTries and MaxTriesWaiting are checked when a try is asked, mesh-wide: a new try is refused while MaxTries tries run
// on a build node or MaxTriesWaiting wait for one (queued, or their asks not yet taken by a node). It is no bound on
// what runs later: tries let in while waiting may later all run at once, and each node's build capacity bounds what
@@ -3,6 +3,7 @@ package manifest
import (
"bytes"
"encoding/json"
checker "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdeliverychecker"
"io/fs"
"os"
"path/filepath"
@@ -222,10 +223,9 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
if !slices.Contains(m.Consumes, check.IndexerModule+".views."+check.IndexerPredefinedState) {
t.Errorf("the indexer's state %s is followed, and not declared", check.IndexerPredefinedState)
}
for bucket, history := range map[string]int{check.CheckerVerdictsBucket: 5, check.CheckerKeptBucket: 1, check.CheckerOwedBucket: 5, check.PreviewsBucket: 5} {
own := strings.TrimPrefix(bucket, check.CheckerModule+"_")
if own == bucket || !slices.ContainsFunc(m.State, func(st state) bool { return st.Name == own && st.History == history }) {
t.Errorf("the bucket %s is not a state of the manifest at history %d", bucket, history)
for own, history := range map[string]int{checker.StateVerdicts: 5, checker.StateKept: 1, checker.StateOwed: 5, checker.StatePreviews: 5} {
if !slices.ContainsFunc(m.State, func(st state) bool { return st.Name == own && st.History == history }) {
t.Errorf("the state %s is not a state of the manifest at history %d", own, history)
}
}
}