diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/seatlist.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judgeseats.go similarity index 52% rename from modules/mesh-delivery-checker/cmd/mesh-delivery-checker/seatlist.go rename to modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judgeseats.go index 697e256a..2bc14572 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/seatlist.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judgeseats.go @@ -7,28 +7,12 @@ import ( "slices" "sync/atomic" - "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/seatdef" "git.novox.be/novox/mesh-sdk/go/seatlist" ) -// The checker reads the seat list as every part does (novox/hq ADR 0635 §1, ADR 0636): through the seat mesh-catalog's verb -// seat-definitions, its two events heard as handlers of the module's event hub, from the bootstrapped definitions until the seat -// has answered. What the reader holds is given to catalogue at every change, which the module graph is worked out with, and to the -// indexer's judge as a request's Seats, which the manifests of a change are judged against. - -// followSeatList starts the checker's reader of the seat list over the bus; the hub is started by the caller once every handler is registered. -func followSeatList(bus seatlist.Bus, logf func(string, ...any), changed func()) *seatlist.Reader { - reader, err := seatlist.Start(seatlist.Config{Bus: bus, Bootstrap: seatlist.Bootstrapped(), - OnChange: func(seats []catalogue.Seat) { catalogue.UseSeats(seats); changed() }}) - if err != nil { - logf("the checker cannot follow its seat list, so no manifest is judged until it starts again: %v", err) - } - if reader != nil { - catalogue.UseSeats(reader.Seats()) - } - return reader -} +// The checker reads the seat list through mesh-sdk's seatlist.Part, the one wiring (design 47 G §204): what the reader holds is the +// indexer's judge's Seats of a request, which the manifests of a change are judged against. // seatList is the reader as the check's SeatList: the held definitions raw, and an error while the seat has not answered. type seatList struct{ reader *seatlist.Reader } @@ -52,15 +36,6 @@ func (s seatList) Definitions() ([]json.RawMessage, error) { return out, nil } -// readIfUnread is called when the checker's hub is heard. The seat may have answered, and said that it does, before the -// checker's consumer existed, whose events start at the stream's end: the first read found it silent and nothing else reads -// again until its next event, so while the list is unread it is read now, on a goroutine of its own (ADR 0636 §1). -func readIfUnread(reader *seatlist.Reader) { - if reader != nil { - reader.ReloadIfUnread() - } -} - // buildsWatch is what is done when the build modules may have changed (design 47 G A-82): the seat list names the build seats // and the indexer's modules the holders of the seats, so a change of either may name a build module not watched yet. It is set once // the parts that watch the build modules are made, and called from then on. diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index 9cdbc8ad..d819103b 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -161,7 +161,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro hub := process.NewEventHub(logger.Printf) reader := &views.ProcessReader{Module: check.CheckerModule, Call: caller.Call, Logf: logger.Printf, Hub: hub} var buildsChanged buildsWatch - seats := seatList{reader: followSeatList(seatlist.ProcessBus(caller.Call, hub), logger.Printf, buildsChanged.Changed)} + seatPart := seatlist.NewPart("checker", nil) + seats := seatList{reader: seatPart.Follow(seatlist.ProcessBus(caller.Call, hub), buildsChanged.Changed, logger.Printf)} // The finding writer is made once the checker is (its observe reads the checker's kept checks); until then a first view // unanswered is said in the log alone. var findings atomic.Pointer[checkerFindings] @@ -503,7 +504,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro 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)) // 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, func() { readIfUnread(seats.reader) }); err != nil { + if err := hub.StartUntilBound(ctx, js, cred, nil, seatPart.EventsLive); err != nil { logger.Printf("%v, so no package registry's published is heard until the consumer is there; the checker tries again", err) } defer hub.Stop() diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/seatlist_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/seatlist_test.go deleted file mode 100644 index b4d23411..00000000 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/seatlist_test.go +++ /dev/null @@ -1,43 +0,0 @@ -package main - -import ( - "encoding/json" - "errors" - "sync/atomic" - "testing" - "time" - - "git.novox.be/novox/mesh-sdk/go/seatlist" -) - -// silentThenAnswering is a seat that is silent at the reader's first read and answers after. -type silentThenAnswering struct{ answering atomic.Bool } - -func (b *silentThenAnswering) Ask(string, any) (json.RawMessage, error) { - if !b.answering.Load() { - return nil, errors.New("not connected yet") - } - return json.RawMessage(`{"revision":1,"seats":[]}`), nil -} - -func (b *silentThenAnswering) Subscribe(string, func(string, json.RawMessage) error) error { - return nil -} - -// A seat that spoke before the checker's consumer existed is read when the hub is heard (ADR 0636 §1). -func TestTheSeatListIsReadWhenTheHubIsHeard(t *testing.T) { - bus := &silentThenAnswering{} - reader := followSeatList(bus, t.Logf, func() {}) - <-reader.Ready() - if reader.Status().Heard { - t.Fatal("a silent seat was heard") - } - bus.answering.Store(true) - readIfUnread(reader) - for deadline := time.Now().Add(5 * time.Second); !reader.Status().Heard; time.Sleep(10 * time.Millisecond) { - if time.Now().After(deadline) { - t.Fatal("the list was not read when the hub was heard") - } - } - var _ seatlist.Bus = bus -} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/seatlist_wiring_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/seatlist_wiring_test.go new file mode 100644 index 00000000..f8e31bca --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/seatlist_wiring_test.go @@ -0,0 +1,29 @@ +package main + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +// The checker wires no reader of the seat list of its own (design 47 G §204, CK9): it follows the list through mesh-sdk's +// seatlist.Part, and its code holds no seatlist.Start nor a file of its own for the list. +func TestTheCheckerWiresNoSeatListReaderOfItsOwn(t *testing.T) { + if _, err := os.Stat("seatlist.go"); err == nil { + t.Error("seatlist.go is the checker's own wiring of the seat list: mesh-sdk's seatlist.Part is the one wiring") + } + files, _ := filepath.Glob("*.go") + for _, f := range files { + if strings.HasSuffix(f, "_test.go") { + continue + } + src, err := os.ReadFile(f) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(src), "seatlist.Start(") { + t.Errorf("%s starts a seat list reader of its own: use seatlist.Part.Follow", f) + } + } +} diff --git a/modules/mesh-delivery-contracts/go/seats/meshdelivery/contract.go b/modules/mesh-delivery-contracts/go/seats/meshdelivery/contract.go index 2227b901..1d72a015 100644 --- a/modules/mesh-delivery-contracts/go/seats/meshdelivery/contract.go +++ b/modules/mesh-delivery-contracts/go/seats/meshdelivery/contract.go @@ -434,12 +434,20 @@ const ( OwedActNote = "note" ) +// PinsCatalogItem is one catalog graph revision, named by its watched branch and its number on the seat mesh-catalog (revisions are numbered per branch). +type PinsCatalogItem struct { + // the watched branch the revision belongs to + Branch string `json:"branch"` + // its number on that branch + Revision int64 `json:"revision"` +} + // Pins is the graph revisions a delivery record starts or stops pointing at. type Pins struct { // assignment graph revisions, by their number on the seat mesh-wants Assignment []int64 `json:"assignment,omitempty"` - // catalog graph revisions, by their number on the seat mesh-catalog - Catalog []int64 `json:"catalog,omitempty"` + // catalog graph revisions, by watched branch and number on the seat mesh-catalog + Catalog []PinsCatalogItem `json:"catalog,omitempty"` // the record's milestone Milestone string `json:"milestone"` // the record's id diff --git a/modules/mesh-delivery-contracts/go/seats/meshdelivery/contract_test.go b/modules/mesh-delivery-contracts/go/seats/meshdelivery/contract_test.go index 421036e9..d8af30a2 100644 --- a/modules/mesh-delivery-contracts/go/seats/meshdelivery/contract_test.go +++ b/modules/mesh-delivery-contracts/go/seats/meshdelivery/contract_test.go @@ -21,9 +21,9 @@ func TestFixtures(t *testing.T) { decode: func(b []byte) (any, error) { var v OwedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-delivery\\.event\\.owed\\.[^.*> ]+\\.[^.*> ]+$")}, {name: "owed is said, a status", subject: "mesh.seat.mesh-delivery.event.owed.novox/mesh-sdk.178", given: "{\"act\":\"status\",\"context\":\"mesh/merge-gate\",\"description\":\"waits for the operator's Approve\",\"head\":\"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\",\"id\":\"status.novox/mesh-sdk#178.aaaaaaaaaaaa.mesh/merge-gate\",\"number\":178,\"repository\":\"novox/mesh-sdk\",\"state\":\"pending\",\"target\":\"https://forge.test/x\"}", wire: "{\"act\":\"status\",\"context\":\"mesh/merge-gate\",\"description\":\"waits for the operator's Approve\",\"head\":\"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\",\"id\":\"status.novox/mesh-sdk#178.aaaaaaaaaaaa.mesh/merge-gate\",\"number\":178,\"repository\":\"novox/mesh-sdk\",\"state\":\"pending\",\"target\":\"https://forge.test/x\"}", decode: func(b []byte) (any, error) { var v OwedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-delivery\\.event\\.owed\\.[^.*> ]+\\.[^.*> ]+$")}, - {name: "pinned is said", subject: "mesh.seat.mesh-delivery.event.pinned", given: "{\"assignment\":[1204],\"catalog\":[3610],\"milestone\":\"47/G\",\"record\":\"47-G\"}", wire: "{\"assignment\":[1204],\"catalog\":[3610],\"milestone\":\"47/G\",\"record\":\"47-G\"}", + {name: "pinned is said", subject: "mesh.seat.mesh-delivery.event.pinned", given: "{\"assignment\":[1204],\"catalog\":[{\"branch\":\"main\",\"revision\":3610}],\"milestone\":\"47/G\",\"record\":\"47-G\"}", wire: "{\"assignment\":[1204],\"catalog\":[{\"branch\":\"main\",\"revision\":3610}],\"milestone\":\"47/G\",\"record\":\"47-G\"}", decode: func(b []byte) (any, error) { var v PinnedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-delivery\\.event\\.pinned$")}, - {name: "freed is said", subject: "mesh.seat.mesh-delivery.event.freed", given: "{\"assignment\":[1204],\"catalog\":[3610],\"milestone\":\"47/G\",\"record\":\"47-G\",\"why\":\"past its retention of 90 days\"}", wire: "{\"assignment\":[1204],\"catalog\":[3610],\"milestone\":\"47/G\",\"record\":\"47-G\",\"why\":\"past its retention of 90 days\"}", + {name: "freed is said", subject: "mesh.seat.mesh-delivery.event.freed", given: "{\"assignment\":[1204],\"catalog\":[{\"branch\":\"main\",\"revision\":3610}],\"milestone\":\"47/G\",\"record\":\"47-G\",\"why\":\"past its retention of 90 days\"}", wire: "{\"assignment\":[1204],\"catalog\":[{\"branch\":\"main\",\"revision\":3610}],\"milestone\":\"47/G\",\"record\":\"47-G\",\"why\":\"past its retention of 90 days\"}", decode: func(b []byte) (any, error) { var v FreedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-delivery\\.event\\.freed$")}, {name: "heads-replaced is said", subject: "mesh.seat.mesh-delivery.event.heads-replaced", given: "{\"at\":\"2026-11-09T20:05:00Z\",\"group\":\"feat/47-G\",\"head\":\"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb\",\"heads\":\"0123456789ab\",\"number\":178,\"repository\":\"novox/mesh-sdk\"}", wire: "{\"at\":\"2026-11-09T20:05:00Z\",\"group\":\"feat/47-G\",\"head\":\"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb\",\"heads\":\"0123456789ab\",\"number\":178,\"repository\":\"novox/mesh-sdk\"}", decode: func(b []byte) (any, error) { var v HeadsReplacedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-delivery\\.event\\.heads-replaced$")}, diff --git a/modules/mesh-delivery-contracts/seats/mesh-delivery/fixtures/events.json b/modules/mesh-delivery-contracts/seats/mesh-delivery/fixtures/events.json index 7d96b149..292b6061 100644 --- a/modules/mesh-delivery-contracts/seats/mesh-delivery/fixtures/events.json +++ b/modules/mesh-delivery-contracts/seats/mesh-delivery/fixtures/events.json @@ -38,7 +38,10 @@ "record": "47-G", "milestone": "47/G", "catalog": [ - 3610 + { + "branch": "main", + "revision": 3610 + } ], "assignment": [ 1204 @@ -53,7 +56,10 @@ "record": "47-G", "milestone": "47/G", "catalog": [ - 3610 + { + "branch": "main", + "revision": 3610 + } ], "assignment": [ 1204 diff --git a/modules/mesh-delivery-contracts/seats/mesh-delivery/seat.json b/modules/mesh-delivery-contracts/seats/mesh-delivery/seat.json index dcdbb60c..93f31549 100644 --- a/modules/mesh-delivery-contracts/seats/mesh-delivery/seat.json +++ b/modules/mesh-delivery-contracts/seats/mesh-delivery/seat.json @@ -1079,9 +1079,24 @@ "catalog": { "type": "array", "items": { - "type": "integer" + "type": "object", + "description": "one catalog graph revision, named by its watched branch and its number on the seat mesh-catalog (revisions are numbered per branch)", + "properties": { + "branch": { + "type": "string", + "description": "the watched branch the revision belongs to" + }, + "revision": { + "type": "integer", + "description": "its number on that branch" + } + }, + "required": [ + "branch", + "revision" + ] }, - "description": "catalog graph revisions, by their number on the seat mesh-catalog" + "description": "catalog graph revisions, by watched branch and number on the seat mesh-catalog" }, "assignment": { "type": "array", diff --git a/modules/mesh-delivery-contracts/ts/seats/meshdelivery/index.d.ts b/modules/mesh-delivery-contracts/ts/seats/meshdelivery/index.d.ts index 242203a3..64cf716b 100644 --- a/modules/mesh-delivery-contracts/ts/seats/meshdelivery/index.d.ts +++ b/modules/mesh-delivery-contracts/ts/seats/meshdelivery/index.d.ts @@ -217,12 +217,20 @@ export interface Owed { "text"?: string; } +/** one catalog graph revision, named by its watched branch and its number on the seat mesh-catalog (revisions are numbered per branch) */ +export interface PinsCatalogItem { + /** the watched branch the revision belongs to */ + "branch": string; + /** its number on that branch */ + "revision": number; +} + /** the graph revisions a delivery record starts or stops pointing at */ export interface Pins { /** assignment graph revisions, by their number on the seat mesh-wants */ "assignment"?: number[]; - /** catalog graph revisions, by their number on the seat mesh-catalog */ - "catalog"?: number[]; + /** catalog graph revisions, by watched branch and number on the seat mesh-catalog */ + "catalog"?: PinsCatalogItem[]; /** the record's milestone */ "milestone": string; /** the record's id */ diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index 21b60028..c44f37c8 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -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()} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 70451ace..1f4e3b58 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -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) } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go index eb692696..d138b4fe 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go @@ -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) } } diff --git a/modules/mesh-delivery-indexer/internal/index/names.go b/modules/mesh-delivery-indexer/internal/index/names.go index 8a905656..8ae3641d 100644 --- a/modules/mesh-delivery-indexer/internal/index/names.go +++ b/modules/mesh-delivery-indexer/internal/index/names.go @@ -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" diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 56b7eeda..853cc3db 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -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 diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 4f6e4770..b155f070 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -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",