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

IX31: the indexer follows the seat mesh-module-assigner's assignments through its holder, no assigner module named

Browse Source 
The holder is asked through mesh-sdk's seatlist.HolderOf, its first view and views family followed as the builder's and the
checker's are; the module consumes the seat's views.assignments and invokes the seat's verb assignments alone.

Builds: design 47 G §7, §428, §556, A-119, A-126, A-131
Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-12 05:35:57 +01:00
parent 9a28de8ab0
commit 820f7c2351
6 changed files with 66 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go
+25 -9
View File
@@ -1307,7 +1307,7 @@ func TestIndexerCountsOnlyThePlacedBuildNodesToolchains(t *testing.T) {
}
}
// The seat mesh-module-assigner's assignments, as the assigner writes it (ADR 0567): n1 places mesh-build-go, n2 nothing.
if assignments, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.AssignmentsBucket, History: index.AssignmentsHistory}); err != nil {
if assignments, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: assignerHolderModule + "_" + meshmoduleassigner.StateAssignments, History: 5}); err != nil {
t.Fatal(err)
}
place(ctx, t, "n1", "mesh-build-go")
@@ -1386,32 +1386,48 @@ func TestIndexerCountsOnlyThePlacedBuildNodesToolchains(t *testing.T) {
// assigner's runtime started together with the indexer's, answering after the indexer asked (DLV-F5r).
var assignerLate chan struct{}
// assignerHolderModule is the module holding the seat mesh-module-assigner in these tests: the seat list names it, the indexer
// follows it (design 47 G A-119, A-126, A-131) and no code of the indexer holds its name.
const assignerHolderModule = "acme-assigner"
// serveAssignerAssignments is the assigner as the owner of its assignments (ADR 0615, ADR 0621), with the real owner's side:
// its first-view tool assignments answered from the bucket at its mark (the bucket made empty where a test made none),
// and its views family said by the views emitter onto ASSIGNER_VIEWS, which the indexer's durable consumer reads.
func serveAssignerAssignments(t *testing.T, ctx context.Context, js jetstream.JetStream, nc *nats.Conn) {
t.Helper()
assignments, err := js.KeyValue(ctx, index.AssignmentsBucket)
assignments, err := js.KeyValue(ctx, assignerHolderModule+"_"+meshmoduleassigner.StateAssignments)
if err != nil {
if assignments, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.AssignmentsBucket, History: index.AssignmentsHistory}); err != nil {
if assignments, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: assignerHolderModule + "_" + meshmoduleassigner.StateAssignments, History: 5}); err != nil {
t.Fatal(err)
}
}
marks, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.AssignerModule + "_" + views.MarksState, History: 1})
marks, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: assignerHolderModule + "_" + views.MarksState, History: 1})
if err != nil {
t.Fatal(err)
}
if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-indexer", AckPolicy: jetstream.AckExplicitPolicy,
FilterSubjects: []string{"mesh.mod." + index.AssignerModule + ".event.views.>"}}); err != nil {
FilterSubjects: []string{"mesh.mod." + assignerHolderModule + ".event.views.>"}}); err != nil {
t.Fatal(err)
}
em := views.Emitter{Owner: index.AssignerModule, Subject: process.Account{Node: "n1", Module: index.AssignerModule}.EventSubject,
em := views.Emitter{Owner: assignerHolderModule, Subject: process.Account{Node: "n1", Module: assignerHolderModule}.EventSubject,
Publish: views.JetStreamPublish(js), Marks: marks}
if err := em.Start(ctx, index.AssignmentsState, assignments, nil); err != nil {
if err := em.Start(ctx, meshmoduleassigner.StateAssignments, assignments, nil); err != nil {
t.Fatal(err)
}
late := assignerLate
sub, err := nc.Subscribe("mesh.mod.mesh-delivery-assigner.call.assignments.>", func(m *nats.Msg) {
// The seat's verb assignments with seats true answers who holds each seat (mesh-sdk's seatlist.HolderOf, A-126).
holder := func(m *nats.Msg) {
raw, _ := json.Marshal(map[string]any{"result": map[string]any{"seats": []map[string]any{{"seat": meshmoduleassigner.Seat, "holders": []map[string]string{{"module": assignerHolderModule}}}}}, "node": "n0"})
_ = m.Respond(raw)
}
for _, subject := range []string{"mesh.seat." + meshmoduleassigner.Seat + ".call.assignments.>", "mesh.seat." + meshmoduleassigner.Seat + ".tool.assignments"} {
hs, err := nc.Subscribe(subject, holder)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = hs.Unsubscribe() })
}
sub, err := nc.Subscribe("mesh.mod."+assignerHolderModule+".call.assignments.>", func(m *nats.Msg) {
if late != nil { // its runtime comes up after the indexer started (DLV-F5r)
select {
case <-late:
@@ -1419,7 +1435,7 @@ func serveAssignerAssignments(t *testing.T, ctx context.Context, js jetstream.Je
return
}
}
a, err := views.FirstView(index.AssignerModule, index.AssignmentsState, jsSource{ctx, assignments}, jsSource{ctx, marks}, "", time.Now())
a, err := views.FirstView(assignerHolderModule, meshmoduleassigner.StateAssignments, jsSource{ctx, assignments}, jsSource{ctx, marks}, "", time.Now())
out := map[string]any{"result": a, "node": "n0"}
if err != nil {
out = map[string]any{"error": err.Error()}
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go
+2 -2
View File
@@ -528,14 +528,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return nil
}}
buildModules.refresh()
// The assignments is the assigner's: read through its first view and its views family, never its bucket (DLV-F1r).
// The assignments is the seat mesh-module-assigner's: read through its holder's first view and views family, never its bucket (DLV-F1r).
assignmentsErrs := make(chan error)
// The indexer's one pull of its durable on the shared events stream (process.EventHub): every handler is registered
// before it starts.
hub := process.NewEventHub(logger.Printf)
reader := &views.ProcessReader{Module: index.Module, Call: calls.Call, Logf: logger.Printf, Hub: hub,
Findings: func() views.WireWriter { return writer }}
followAssignments(ctx, reader, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "indexer_first_views reads nothing and indexer_first_view_again is not heard", logger.Printf), cred, js, gate, func(ctx context.Context) {
followAssignments(ctx, reader, calls.Call, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "indexer_first_views reads nothing and indexer_first_view_again is not heard", logger.Printf), cred, js, gate, func(ctx context.Context) {
if err := ix.Recompute(ctx); err != nil {
logger.Printf("writing the modules again after the assignments of a build module changed: %v", err)
}
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go
+35 -30
View File
@@ -12,9 +12,11 @@ import (
"github.com/nats-io/nats.go/jetstream"
mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
"git.novox.be/novox/mesh-sdk/go/keys"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-sdk/go/seatlist"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/heard"
mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
@@ -110,39 +112,42 @@ const graphState = graph.State
// where indexer_first_view_again asks.
var firstViewsBucket = index.Module + "_" + views.StatusState
// assignmentsTool is the assigner's first-view tool of its assignments, as the indexer's invokes names it (ADR 0621 §2).
const assignmentsTool = index.AssignerModule + "." + index.AssignmentsState
// followAssignments reads the seat mesh-module-assigner's assignments through the assigner's first view and then its views
// family, never its bucket (ADR 0615 §1, DLV-F1r), into gate: the module's views reader follows it (design 47 G §532, A-26),
// which asks the first view, hears the views family and raises and clears the finding of a first view unanswered through the
// indexer's finding writer. Each key is taken as it changes, settled once answered, with changed told whenever what the gate
// passes on changed. It waits for the first answer at most views.Bound, so a normal start, the two owners coming up together,
// computes the fingerprints once; past it the indexer goes on, the gate passing every report, and raises its finding naming the
// assigner and assignments until the answer clears it.
func followAssignments(ctx context.Context, reader *views.ProcessReader, status *process.AwaitedBucket, cred process.Account, js jetstream.JetStream,
// followAssignments reads the seat mesh-module-assigner's assignments through its holder's first view and then its views family,
// never its bucket (ADR 0615 §1, DLV-F1r), into gate: the holder is the module the seat list names for the seat, asked through
// mesh-sdk's seatlist.HolderOf (design 47 G A-119, A-126, A-131), so no module is named and a new holder is followed with no change
// of code. The module's views reader follows it (design 47 G §532, A-26), which asks the first view, hears the views family and
// raises and clears the finding of a first view unanswered through the indexer's finding writer. Each key is taken as it changes,
// settled once answered, with changed told whenever what the gate passes on changed. It waits for the holder and the first answer
// at most views.Bound, so a normal start, the two owners coming up together, computes the fingerprints once; past it the indexer
// goes on, the gate passing every report, and the holder is asked again until it answers.
func followAssignments(ctx context.Context, reader *views.ProcessReader, call seatlist.HolderCaller, status *process.AwaitedBucket, cred process.Account, js jetstream.JetStream,
gate *buildrun.ToolchainGate, changed func(context.Context), logf func(string, ...any)) {
var settled atomic.Bool // the first view answered: the gate's nodes are the counted ones from now on
pic, err := reader.Follow(index.AssignerModule, index.AssignmentsState, assignmentsTool)
if err != nil {
logf("the assignments of the assigner is not followed: %v", err)
return
}
f := pic.Follower()
keep := f.Changed // the picture's own, kept
f.Changed = func(key string, value json.RawMessage, deleted bool, at time.Time) {
if keep != nil {
keep(key, value, deleted, at)
}
if gate.Assignments(key, value, deleted, at) {
changed(ctx)
}
if settled.Load() && gate.OnPlaced != nil {
gate.OnPlaced() // the counted nodes may have changed (ADR 0624 §1)
}
}
answered := make(chan struct{})
go func() {
holder, ok := buildrun.AssignerHolder(ctx, call, 5*time.Second, buildrun.SleepFor, logf)
if !ok {
return
}
names := buildrun.AssignmentsOf(holder)
pic, err := reader.Follow(names.Holder, mma.StateAssignments, names.Tool)
if err != nil {
logf("the assignments of the seat %s is not followed: %v", mma.Seat, err)
return
}
f := pic.Follower()
keep := f.Changed // the picture's own, kept
f.Changed = func(key string, value json.RawMessage, deleted bool, at time.Time) {
if keep != nil {
keep(key, value, deleted, at)
}
if gate.Assignments(key, value, deleted, at) {
changed(ctx)
}
if settled.Load() && gate.OnPlaced != nil {
gate.OnPlaced() // the counted nodes may have changed (ADR 0624 §1)
}
}
select {
case <-f.Ready():
case <-ctx.Done():
@@ -163,7 +168,7 @@ func followAssignments(ctx context.Context, reader *views.ProcessReader, status
select {
case <-answered:
case <-wait.Done():
logf("the assigner has not answered assignments within %s: the indexer goes on, every node's toolchain report counting until it answers", views.Bound)
logf("the seat %s has not answered assignments within %s: the indexer goes on, every node's toolchain report counting until it answers", mma.Seat, views.Bound)
}
}
modules/mesh-delivery-indexer/internal/index/names.go
-4
View File
@@ -26,12 +26,8 @@ const (
TakenBucket = Module + "_" + TakenState
PredefinedListKey = "list"
ForgeBridgeModule = "mesh-delivery-forge-bridge"
AssignerModule = "mesh-delivery-assigner"
GateModule = "mesh-delivery-gate"
CheckerModule = "mesh-delivery-checker"
AssignmentsState = "assignments"
AssignmentsBucket = AssignerModule + "_" + AssignmentsState
AssignmentsHistory = 5
PreviewsBucket = CheckerModule + "_previews"
PreviewsHistory = 5
toolchainsState = "toolchains"
modules/mesh-delivery-indexer/manifest_test.go
+3 -4
View File
@@ -165,9 +165,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
// "Reads FORGE" is a consumes of the indexer's stream: the broker grants only buckets by reads (mesh-control
// internal/broker/state.go:62-93).
// The registry seats' published fill each (§413: no consume of version.numbered) package's series (DLV-I5, design 51 *The versions*).
if want := []string{"go-package-registry.published", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-conditions.findings-asked", "mesh-delivery-assigner.views.assignments",
"mesh-delivery-checker.preview.written", "mesh-delivery-checker.try.ended", "mesh-delivery-checker.try.started", "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**",
"mesh-delivery.source-moved", "mesh-wants.version-delivered", "mesh-wants.wait-ended", "mesh-wants.wanted", "node-build-go.step-ended", "node-engine.running", "npm-package-registry.published", "operator-channel.decided"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
if want := []string{"go-package-registry.published", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-conditions.findings-asked", "mesh-delivery-checker.preview.written", "mesh-delivery-checker.try.ended", "mesh-delivery-checker.try.started", "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**",
"mesh-delivery.source-moved", "mesh-wants.version-delivered", "mesh-wants.wait-ended", "mesh-wants.wanted", "node-build-go.step-ended", "node-engine.running", "npm-package-registry.published", "operator-channel.decided", "seat:mesh-module-assigner.views.assignments"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
t.Errorf("consumes = %v, want %v", m.Consumes, want)
}
if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" {
@@ -179,7 +178,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
// Three invokes: the file of a registered module's manifest and lock files (registered.go), the forge's trees (the named paths' objects, since the indexer is the one reader of a manifest's paths:
// ADR 0499 clauses 2 and 3, ADR 0495 clause 1; and the module directories of the trunk, issue 1900) and whether a
// commit is in another (issue 1900).
if want := []string{"mesh-delivery-assigner.assignments", "seat:git.trees", "seat:git.file", "seat:git.contains", "seat:git.open-pulls", "seat:git.branch", "mesh-delivery-checker.checker_tries", "mesh-delivery-forge-bridge.bridge_merged_message", "seat:node-engine.composition", "seat:mesh-delivery.source", "seat:mesh-module-assigner.assignments", "seat:mesh-module-assigner.placement", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence", "seat:node-build-go.steps", "seat:node-build-node.steps", "seat:node-build-image.steps", "mesh-delivery-builder.builder_versions"}; !reflect.DeepEqual(m.Invokes, want) {
if want := []string{"seat:git.trees", "seat:git.file", "seat:git.contains", "seat:git.open-pulls", "seat:git.branch", "mesh-delivery-checker.checker_tries", "mesh-delivery-forge-bridge.bridge_merged_message", "seat:node-engine.composition", "seat:mesh-delivery.source", "seat:mesh-module-assigner.assignments", "seat:mesh-module-assigner.placement", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence", "seat:node-build-go.steps", "seat:node-build-node.steps", "seat:node-build-image.steps", "mesh-delivery-builder.builder_versions"}; !reflect.DeepEqual(m.Invokes, want) {
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
+1 -2
View File
@@ -80,7 +80,7 @@
"mesh-build-go.toolchain.reported",
"mesh-build-node.toolchain.reported",
"mesh-conditions.findings-asked",
"mesh-delivery-assigner.views.assignments",
"seat:mesh-module-assigner.views.assignments",
"mesh-delivery-checker.preview.written",
"mesh-delivery-checker.try.ended",
"mesh-delivery-checker.try.started",
@@ -154,7 +154,6 @@
"mesh-delivery-checker.previews"
],
"invokes": [
"mesh-delivery-assigner.assignments",
"seat:git.trees",
"seat:git.file",
"seat:git.contains",