Merge branch 'feat/47-G-fix11-b5b' into feat/47-G-fix11-b7
This commit is contained in:
+2
-27
@@ -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.
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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$")},
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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 */
|
||||
|
||||
@@ -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()}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user