diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/asks.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/asks.go index 6fbe713..2ceefe7 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/asks.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/asks.go @@ -119,11 +119,11 @@ func (p *passer) composeAtHead(ctx context.Context, m assign.Mesh, ch assign.Cha if h := ch.Members[repository]; h != "" { commits, member = []string{h}, true } else { - commits = p.w.runningCommits(ctx, kvGetter(p.kvs[delivery.BuildsBucket]), module) + commits = p.w.runningCommits(module) } if len(commits) == 0 { - notJudged = append(notJudged, "not judged whether "+module+" reads the group's manifests: no build of it is assigned with a "+ - "commit on record, and the group holds no member of "+repository) + notJudged = append(notJudged, "not judged whether "+module+" reads the group's manifests: the builder's log holds no done "+ + "trunk build of it, and the group holds no member of "+repository) continue } for _, commit := range commits { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compose_integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compose_integration_test.go index 8bc0662..43030fa 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compose_integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compose_integration_test.go @@ -106,7 +106,7 @@ func TestAGroupWithAnSDKMemberIsComposedByTheComposerAtItsHead(t *testing.T) { b.putJSON(t, provider.AssignmentsBucket, "g14.web", []byte(`{"node":"g14","module":"web","build":"build-1"}`)) running := strings.Repeat("3", 40) b.putJSON(t, provider.AssignmentsBucket, "novox.mesh-controller", []byte(`{"node":"novox","module":"mesh-controller","build":"c-1"}`)) - b.putJSON(t, delivery.BuildsBucket, delivery.BuildKey("mesh-controller", "c-1"), []byte(`{"id":"c-1","repository":"novox/mesh-controller","commit":"`+running+`"}`)) + b.built(t, "mesh-controller", "novox/mesh-controller", running) sdkHead := strings.Repeat("4", 40) heads := []delivery.Head{{Repository: "novox/mesh-sdk", Number: 80, Head: sdkHead}, {Repository: "novox/web", Number: 7, Head: strings.Repeat("2", 40)}} hash, _ := delivery.HeadsHash(heads) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index 234b56a..5cce388 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -21,6 +21,7 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/announce" "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/compose/judge" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -82,6 +83,11 @@ func startBed(t *testing.T, leaveOut string) *bed { } b.buckets[name] = kv } + // The builder's module log, as the runtime raises it (ADR 0297): the running commits are read from it (ADR 0571 clause 1). + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: buscompose.LogStreamName(delivery.BuilderModule, delivery.BuilderLog), + Subjects: []string{buscompose.LogSubject(delivery.BuilderModule, delivery.BuilderLog) + ".>"}, AllowDirect: true}); err != nil { + t.Fatal(err) + } if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}); err != nil { t.Fatal(err) } @@ -177,8 +183,8 @@ func saysWhatItKeeps(t *testing.T, b *bed, done <-chan error) { // 1297/B2b: a bucket it reads that the bus does not hold stops the start, naming the bucket. func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) { for _, missing := range readBuckets { - if missing == delivery.PlacementBucket || sourceOfGrants(missing) { - continue // it starts without it, below (the grants' source: main_test.go) + if startsWithout(missing) { + continue // it starts without it (the placement and the grants' source: main_test.go; the puts: below) } b := startBed(t, missing) select { @@ -264,6 +270,19 @@ func (b *bed) putJSON(t *testing.T, bucket, key string, value []byte) { } } +// built appends the end of a done trunk build of module at commit to the builder's log, as the builder does. +func (b *bed) built(t *testing.T, module, repository, commit string) { + t.Helper() + raw, err := delivery.EncodeBuildLogEntry(delivery.BuildLogEntry{Event: delivery.BuildEndedEvent, Module: module, FP: delivery.SourcePrefix2 + strings.Repeat("c", 64), + Repository: repository, Directory: delivery.RootPath, Commit: commit, Node: "novox", At: time.Now().UTC(), Outcome: delivery.BuildDone}) + if err != nil { + t.Fatal(err) + } + if _, err := b.js.Publish(b.ctx, buscompose.LogSubject(delivery.BuilderModule, delivery.BuilderLog)+"."+module, raw); err != nil { + t.Fatal(err) + } +} + // compose says the controller composed module for node at fp, with the digest d (a letter repeated) and the judgement of the // move to it, in mesh-controller_composed. func (b *bed) compose(t *testing.T, node, module, fp string, d byte, judgement delivery.Judgement) string { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 737cee7..a3fd2f7 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -22,6 +22,7 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/announce" "git.novox.be/novox/mesh-sdk/go/asks" "git.novox.be/novox/mesh-sdk/go/buscompose" @@ -58,7 +59,6 @@ func main() { var readBuckets = []string{ buscompose.ResolvedBucket, provider.AssignmentsBucket, - delivery.BuildsBucket, delivery.ControllerGrantsBucket, delivery.PartGrantsBucket, buscompose.LeaseBucket, @@ -76,6 +76,10 @@ var readBuckets = []string{ keptBucket, } +// readLogs are the logs of other parts the assigner reads, each named in its manifest's reads: the builder's log, whose +// newest done trunk build of a module names the commit a reader of the manifests runs at (novox/hq ADR 0571 clause 1). +var readLogs = []string{delivery.BuildLogRead} + // takeoverBucket is the condition keeper's take-over key's bucket (novox/hq ADR 0564 §9): whether the assigner's said // conditions go straight to the router or only as findings (issue 1952 WM6). var takeoverBucket = health.Bucket(health.KeeperModule, health.TakeoverState) @@ -204,6 +208,17 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any return out, nil } +// startsWithout says the assigner starts without the bucket when it cannot be opened (open says why for each): the seat's +// states and those new with a delivery of their own, bound once made, and the two that choose the grants' source. +func startsWithout(name string) bool { + switch name { + case delivery.SeatHoldersBucket, keptBucket, provisions.ProvisionsBucket, assign.PoliciesBucket, assign.FindingsBucket, takeoverBucket, + delivery.AssignerPutsBucket: + return true + } + return slices.Contains(seatBuckets, name) || sourceOfGrants(name) +} + // sourceOfGrants says the bucket is one of the two that choose the grants' source, the grants part's grants and the // controller's lease: neither stops the assigner when it is not there (delivery.GrantSource). func sourceOfGrants(name string) bool { @@ -239,7 +254,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro thisNode = account.Node logger.Printf("started on the account %s, reading %d buckets", account.User, len(kvs)) findings := findingsOf(ctx, kvs, logger.Printf) - return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, logger) + buildLog := func(handle func(stdio.LogEntry) error) error { + return said.ReadLog(delivery.BuilderModule, delivery.BuilderLog).Watch("", stdio.LogFromAll, handle) + } + return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, buildLog, logger) } // findingsOf is where the assigner's said conditions go as findings (issue 1952 WM6): its findings bucket, and the @@ -341,7 +359,7 @@ var watched = []string{ // follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after // each change; changes that come during a pass make one more. It returns when ctx ends or a watch ends. -func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, kvs buckets, say func(asks.Condition) error, findings assign.Findings, bus caller, logger *log.Logger) error { +func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, kvs buckets, say func(asks.Condition) error, findings assign.Findings, bus caller, buildLog logWatch, logger *log.Logger) error { w := newWorld() dirty := make(chan struct{}, 1) wake := func() { @@ -350,6 +368,22 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, default: } } + // The builder's log (ADR 0571 clause 1): every entry it keeps before the first pass, then each new one as it is appended; + // a module's newer done trunk build makes a pass, as a composition waiting on its commit is judged then. Where it cannot be + // read, a composition names no running reader of the manifests until the assigner starts again, and it is said. + if buildLog != nil { + var first atomic.Bool + if err := buildLog(func(le stdio.LogEntry) error { + if w.putBuilt(le.Value) && first.Load() { + wake() + } + return nil + }); err != nil { + logger.Printf("%s is not read (%v): no composition is judged against a running reader of the manifests until the assigner starts again", + delivery.BuildLogRead, err) + } + first.Store(true) + } errs := make(chan error, len(watched)+2) settled := make(chan struct{}, len(watched)+2) watch := func(name string, opts ...jetstream.WatchOpt) error { @@ -433,7 +467,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, } started := 0 for _, name := range watched { - if kvs[name] == nil && (slices.Contains(seatBuckets, name) || sourceOfGrants(name) || name == provider.AssignmentsBucket || name == assign.PoliciesBucket || name == provisions.ProvisionsBucket || name == keptBucket || name == delivery.SeatHoldersBucket) { + if kvs[name] == nil && (startsWithout(name) || name == provider.AssignmentsBucket) { continue // not on the bus at this start (open said so) } started++ diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go index d2a1a83..54f98ec 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go @@ -42,8 +42,16 @@ func TestOpensWhatItsManifestNames(t *testing.T) { } var reads []string for _, r := range m.Reads { + if slices.Contains(readLogs, r) { + continue // a log, read through its own reader (ADR 0571 clause 1), not a bucket + } reads = append(reads, strings.Replace(r, ".", "_", 1)) } + for _, l := range readLogs { + if !slices.Contains(m.Reads, l) { + t.Errorf("the program reads the log %s, the manifest does not name it", l) + } + } var states []string for _, s := range m.State { states = append(states, m.Module+"_"+s.Name) @@ -112,7 +120,7 @@ func TestOpensWhatItsManifestNames(t *testing.T) { kvs[delivery.ControllerComposedBucket].(*watchKV).entries = []kvEntry{{key: ckey, value: cd}} kvs[delivery.AssignerPutsBucket].(*watchKV).closed = true kvs[delivery.AssignerWantsBucket].(*watchKV).wrote = wrote - go follow(ctx, nil, nil, kvs, func(asks.Condition) error { return nil }, assign.Findings{}, nil, log.New(io.Discard, "", 0)) + go follow(ctx, nil, nil, kvs, func(asks.Condition) error { return nil }, assign.Findings{}, nil, nil, log.New(io.Discard, "", 0)) select { case key := <-wrote: if key != "want.ace.node" { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index 175bf7a..1acdc2b 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -13,6 +13,7 @@ import ( "github.com/nats-io/nats.go/jetstream" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/compose" @@ -51,6 +52,8 @@ type world struct { changes map[string]uint64 // meta is, by bucket and key, the revision and the time of the value held now, where the watch said them (putEntry). meta map[string]entryMeta + // trunkBuilt is, by module, its newest done trunk build in the builder's log (putBuilt). + trunkBuilt map[string]delivery.BuildLogEntry } // entryMeta is a value's revision in its bucket and when it was written. @@ -68,7 +71,7 @@ type parsed struct { func newWorld() *world { return &world{values: map[string]map[string][]byte{}, manifests: map[string]parsed{}, healthy: map[string][]string{}, changes: map[string]uint64{}, - meta: map[string]entryMeta{}} + meta: map[string]entryMeta{}, trunkBuilt: map[string]delivery.BuildLogEntry{}} } // putEntry is put, keeping the value's revision and the time it was written. @@ -666,32 +669,35 @@ func (w *world) changed(bucket, key string) uint64 { return w.changes[bucket+"|"+key] } -// runningCommits is, of every open assignment of module, the commit of the build it names, read from the controller's -// build record (`mesh-controller_builds` `.build.`), distinct and sorted: the commits a reader of the -// manifests runs at (novox/hq ADR 0482 clause 5, issue 1623). A build whose record names no full commit is left out. -func (w *world) runningCommits(ctx context.Context, get getter, module string) []string { - _, all := w.assigned() - seen := map[string]bool{} - for _, a := range all { - if a.Module != module || a.Build == "" { - continue - } - value, err := get(ctx, delivery.BuildKey(module, a.Build)) - if err != nil { - continue - } - var rec delivery.BuildRecord - if json.Unmarshal(value, &rec) != nil || len(rec.Commit) != 40 { - continue - } - seen[rec.Commit] = true +// logWatch follows the builder's log: handle each entry it keeps, then each new one (process.LogReader.Watch). +type logWatch func(handle func(stdio.LogEntry) error) error + +// putBuilt takes one entry of the builder's log in: a done trunk build's end (no ref) naming a full commit, newer than the +// module's last, becomes the module's running commit. It answers whether the commit changed. +func (w *world) putBuilt(raw []byte) bool { + e, err := delivery.DecodeBuildLogEntry(raw) + if err != nil || e.Event != delivery.BuildEndedEvent || e.Outcome != delivery.BuildDone || e.Ref != "" || len(e.Commit) != 40 { + return false } - out := make([]string, 0, len(seen)) - for c := range seen { - out = append(out, c) + w.mu.Lock() + defer w.mu.Unlock() + have, ok := w.trunkBuilt[e.Module] + if ok && !e.At.After(have.At) { + return false } - sort.Strings(out) - return out + w.trunkBuilt[e.Module] = e + return !ok || have.Commit != e.Commit +} + +// runningCommits is the commit of module's newest done trunk build in the builder's log (ADR 0571 clause 1): the commit a +// reader of the manifests runs at (novox/hq ADR 0482 clause 5, issue 1623); none when the log holds no such build. +func (w *world) runningCommits(module string) []string { + w.mu.Lock() + defer w.mu.Unlock() + if e, ok := w.trunkBuilt[module]; ok { + return []string{e.Commit} + } + return nil } // keptBucket is mesh-config's `kept`: the approved held values of open pull requests (ADR 0565 §5). diff --git a/modules/mesh-delivery-assigner/go.mod b/modules/mesh-delivery-assigner/go.mod index 0744ac9..df09256 100644 --- a/modules/mesh-delivery-assigner/go.mod +++ b/modules/mesh-delivery-assigner/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-assigner go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029222821-bb993dd60641 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-assigner/go.sum b/modules/mesh-delivery-assigner/go.sum index ecec070..07e0ff7 100644 --- a/modules/mesh-delivery-assigner/go.sum +++ b/modules/mesh-delivery-assigner/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121 h1:OsA1v6n736a7RqEV8Nkip8pf5ILMbzj0PWuKqcBOhK0= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029222821-bb993dd60641 h1:nP/jrjm2Qt1wdmd1Bc+E6c4+PobAzW5o64SPArE+jaw= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029222821-bb993dd60641/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index b939b72..f722280 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -33,8 +33,8 @@ func TestManifest(t *testing.T) { } reads := slices.Clone(m.Reads) slices.Sort(reads) - want := []string{"mesh-build-go.steps", "mesh-conditions.takeover", "mesh-config.kept", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-controller.grants", "mesh-controller.lease", - "mesh-delivery-builder.builds", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", + want := []string{"mesh-build-go.steps", "mesh-conditions.takeover", "mesh-config.kept", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.composed", "mesh-controller.grants", "mesh-controller.lease", + "mesh-delivery-builder.build-log", "mesh-delivery-builder.builds", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-grants.grants", "mesh-provisioner.provisions", "node-engine.have"} if !reflect.DeepEqual(reads, want) { t.Errorf("reads = %v, want exactly %v", reads, want) diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index 9dae8b6..91ad85a 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -31,10 +31,10 @@ "mesh-config.kept", "mesh-config.resolved", "mesh-controller.assignments", - "mesh-controller.builds", "mesh-controller.composed", "mesh-controller.grants", "mesh-controller.lease", + "mesh-delivery-builder.build-log", "mesh-delivery-builder.builds", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests",