From 7cf47c04ebf35b5249a8a8f2bc091c2150e63aa7 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 2 Nov 2026 18:44:25 +0100 Subject: [PATCH] mesh-delivery-assigner: compose nothing; judge proposals with its own judge and compare the published compositions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ADR 0593 §7, §8: the check's composition, its compose asks to the Go build seat and the state compositions go (the checker asks the engines); a proposal is judged without frombus; mesh-wants.upgrade compare reads every node's composed- through mesh-host.composed. The assigner no longer imports compose, composer, frombus or buildrun. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- .../cmd/mesh-delivery-assigner/composed.go | 73 ++++++ .../integration_test.go | 82 +------ .../cmd/mesh-delivery-assigner/main.go | 60 +---- .../cmd/mesh-delivery-assigner/main_test.go | 10 +- .../cmd/mesh-delivery-assigner/pass.go | 54 ---- .../cmd/mesh-delivery-assigner/pass_test.go | 19 +- .../cmd/mesh-delivery-assigner/seat.go | 61 +---- .../mesh-delivery-assigner/seat_tools_test.go | 14 +- .../cmd/mesh-delivery-assigner/wants_tools.go | 8 +- .../cmd/mesh-delivery-assigner/world.go | 231 +----------------- .../cmd/mesh-delivery-assigner/world_test.go | 73 ------ modules/mesh-delivery-assigner/go.mod | 2 - .../mesh-delivery-assigner/manifest_test.go | 18 +- modules/mesh-delivery-assigner/module.json | 8 +- 14 files changed, 143 insertions(+), 570 deletions(-) create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/composed.go diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/composed.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/composed.go new file mode 100644 index 0000000..60e975f --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/composed.go @@ -0,0 +1,73 @@ +package main + +import ( + "sort" + "strings" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// Every node's published compositions, read only to compare (novox/hq ADR 0593 §8): each node's engine writes its own +// `composed-` (delivery.ComposedRead, one bucket per node), and `mesh-wants.upgrade` with compare names, for every +// node, the composition its own have runs beside the newest it published of each module and of the node, so a person sees +// where they part. The assigner composes nothing and never writes these. + +// publishedRow is one composition a node published: the module ("node" for its own), the fingerprint, the digest, or why +// it was refused or cannot be read. +type publishedRow struct { + Key string `json:"key"` + Module string `json:"module"` + FP string `json:"fp,omitempty"` + Digest string `json:"digest,omitempty"` + Refused string `json:"refused,omitempty"` + Unread string `json:"unread,omitempty"` +} + +// published is, by node, every composition it published, sorted by key, from every key of the per-node state (a key +// names its node first): each value decoded and digested here, as the node-engine digests it (delivery.ComposedDigestOf). +// A value or key that does not read is named with why, never left out. +func published(values map[string][]byte) map[string][]publishedRow { + byNode := map[string]map[string][]byte{} + for key, raw := range values { + node, _, _ := strings.Cut(key, ".") + if byNode[node] == nil { + byNode[node] = map[string][]byte{} + } + byNode[node][key] = raw + } + out := map[string][]publishedRow{} + for node, keys := range byNode { + var rows []publishedRow + for key, raw := range keys { + row := publishedRow{Key: key} + _, module, fp, isNode, err := delivery.ParseComposedKey(key) + switch { + case err != nil: + row.Unread = err.Error() + case isNode: + row.Module = "node" + default: + row.Module, row.FP = module, fp + } + if err == nil { + c, derr := delivery.DecodeComposed(raw) + switch { + case derr != nil: + row.Unread = derr.Error() + case c.Refused != "": + row.Refused = c.Refused + default: + if d, err := delivery.ComposedDigestOf(c); err != nil { + row.Unread = err.Error() + } else { + row.Digest = d + } + } + } + rows = append(rows, row) + } + sort.Slice(rows, func(i, j int) bool { return rows[i].Key < rows[j].Key }) + out[node] = rows + } + return out +} 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 62c0379..554595c 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 @@ -24,10 +24,10 @@ import ( "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" "git.novox.be/novox/mesh-sdk/go/health" "git.novox.be/novox/mesh-sdk/go/process" + "mesh-delivery-assigner/internal/judge" "mesh-delivery-assigner/internal/assign" ) @@ -290,19 +290,6 @@ func (b *bed) place(t *testing.T, node, module string) { } } -// 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 node's own composition is the digest d (a letter repeated), as its have names it (`.node`): a // placed node composes its own (ADR 0571 §6), and the controller's composition is read no more. It answers the digest. func (b *bed) compose(t *testing.T, node string, d byte) string { @@ -333,62 +320,6 @@ func (b *bed) manifest(t *testing.T, module, body string) string { return digest } -// 1297/AC1: against the real bus, a fixture change touching one module on one machine is composed and answered within -// seconds under the checker's key of the check, naming the machine composed and the mesh-sdk it composed with. -func TestAChangeIsComposedOnTheBus(t *testing.T) { - b := startBed(t, "") - trunk := b.manifest(t, "web", `{"module":"web","version":"1"}`) - changed := b.manifest(t, "web", `{"module":"web","version":"2"}`) - mod, err := delivery.EncodeModule(delivery.Module{FP2: delivery.SourcePrefix2 + strings.Repeat("a", 64), Manifest: trunk, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: delivery.RootPath}) - if err != nil { - t.Fatal(err) - } - b.putJSON(t, delivery.ModulesBucket, "web", mod) - b.place(t, "g14", "web") - b.place(t, "ace", "other") - heads := []delivery.Head{{Repository: "novox/web", Number: 7, Head: strings.Repeat("2", 40)}} - hash, _ := delivery.HeadsHash(heads) - // 1927/AC1: the candidate carries the fingerprint set the SDK computes, as the indexer writes it. - pulls := []delivery.CandidatePull{{Head: heads[0], - Modules: []string{"web"}, Fps: map[string]delivery.CandidateModule{"web": {FP2: delivery.SourcePrefix2 + strings.Repeat("b", 64), Manifest: changed}}}} - cand, err := delivery.EncodeCandidate(delivery.Candidate{Group: "feat/x", HeadsHash: hash, Pulls: pulls, FPSet: delivery.FingerprintSet(pulls)}) - if err != nil { - t.Fatal(err) - } - ckey, _ := delivery.CandidatesKey("feat/x") - b.putJSON(t, delivery.CandidatesBucket, ckey, cand) - began := time.Now() - b.start(t) - key, _ := delivery.CompositionKey("feat/x", hash) - var v delivery.CompositionVerdict - for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(50 * time.Millisecond) { - if time.Now().After(deadline) { - t.Fatalf("no composition under %s after 10 s", key) - } - e, err := b.buckets[delivery.AssignerCompositionsBucket].Get(b.ctx, key) - if err != nil { - continue - } - if v, err = delivery.DecodeCompositionVerdict(e.Value()); err != nil { - t.Fatal(err) - } - break - } - if v.State != delivery.CompositionPass || len(v.Machines) != 1 || v.Machines[0] != "g14" || v.SDK == "" { - t.Fatalf("composition = %+v, want a pass on g14 alone naming its mesh-sdk", v) - } - if took := time.Since(began); took > 5*time.Second { - t.Errorf("the composition took %s", took) - } - // 1622/AC3: a group without a mesh-sdk member builds no composer: no ask is put on the Go build seat. - seat, _ := delivery.AskStream(delivery.AskGo) - if st, err := b.js.Stream(b.ctx, seat); err != nil { - t.Fatal(err) - } else if info, err := st.Info(b.ctx); err != nil || info.State.Msgs != 0 { - t.Errorf("a group without a mesh-sdk member put %d asks on the Go build seat (%v)", info.State.Msgs, err) - } -} - // get waits for a key of one of the bed's buckets and answers its value. func (b *bed) get(t *testing.T, bucket, key string) []byte { t.Helper() @@ -1121,16 +1052,17 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) { t.Error("a declined proposal wrote its key") } - // ADR 0567 §4: a node that does not answer fails the verdict once its pulse key's time to live ends, heard on the bus - // with no timer of the assigner's, naming it. + // ADR 0593 §8: a node that does not answer is named as giving no answer once its pulse key's time to live ends, heard on + // the bus with no timer of the assigner's, and the proposal is decided on the answers it has: asked, not failed. q, err = verbs.doPropose(map[string]any{"why": "again", "changes": `[{"op":"take-off","node":"laptop","module":"web"}]`}, "op") if err != nil { t.Fatal(err) } silent := q.(proposed).Proposal - failed := proposalNow(silent.ID, "failed for a silent node", func(p delivery.PlacementProposal) bool { return p.State == delivery.ProposalFailed }) - if failed.Verdict == nil || !strings.Contains(strings.Join(failed.Verdict.Fails, "; "), "laptop (offline) did not answer") { - t.Errorf("the silent node's verdict = %+v", failed.Verdict) + decided := proposalNow(silent.ID, "decided without its silent nodes", func(p delivery.PlacementProposal) bool { return p.State == delivery.ProposalAsked }) + if decided.Verdict == nil || decided.Verdict.State != delivery.VerdictPass || + !strings.Contains(strings.Join(decided.Verdict.Notes, "; "), "no answer from") || !strings.Contains(strings.Join(decided.Verdict.Notes, "; "), "laptop (offline)") { + t.Errorf("the silent node's verdict = %+v", decided.Verdict) } // 774/S59 (ADR 0587 §7): a joining node is written on the enrolment's warrant with no node to answer and no ask raised, 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 44846c5..07368f8 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -15,18 +15,14 @@ import ( "os" "os/signal" "slices" - "sync/atomic" "syscall" "time" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" - 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/delivery" "git.novox.be/novox/mesh-sdk/go/health" "git.novox.be/novox/mesh-sdk/go/process" @@ -56,31 +52,23 @@ func main() { // readBuckets are the buckets of other parts the assigner reads, each named in its manifest's reads (the manifest test // holds the two lists equal), so the controller's broker grants exactly these. var readBuckets = []string{ - buscompose.ResolvedBucket, delivery.PartGrantsBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket, - delivery.CandidatesBucket, delivery.ManifestsBucket, delivery.ModulesBucket, delivery.PresenceBucket, delivery.HaveBucket, - delivery.StepsBucket(goBuildModule), provisions.ProvisionsBucket, - 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} - -// goBuildModule is the Go build module, which takes the assigner's compose asks on its seat's check stream and writes -// their steps in its steps state (novox/hq ADR 0482 clause 4). -const goBuildModule = "mesh-build-go" +// readFamilies are the states kept one bucket per node the assigner reads, each named in its manifest's reads: every +// node's compositions, which mesh-wants.upgrade compares (novox/hq ADR 0593 §8). +var readFamilies = []string{delivery.ComposedRead} // ownBuckets are the assigner's own states, as its manifest declares them: the seat mesh-placement's three among them // (ADR 0567 §1), named after the assigner as every module state is. -var ownBuckets = []string{delivery.AssignerAsksBucket, delivery.AssignerCompositionsBucket, delivery.AssignerDifferencesBucket, +var ownBuckets = []string{delivery.AssignerAsksBucket, delivery.AssignerDifferencesBucket, delivery.AssignerPutsBucket, delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.PlacementProposalsBucket, delivery.PlacementCandidatesBucket, delivery.PlacementPulsesBucket, assign.FindingsBucket, assign.PoliciesBucket, delivery.SeatHoldersBucket} @@ -90,7 +78,7 @@ var seatBuckets = []string{delivery.PlacementBucket, delivery.PlacementProposals delivery.PlacementPulsesBucket} // keptStates is what the assigner keeps and who may read it (ADR 0495 §2-§3): its wants by the watchman and the view -// (and the node agent from release 6), its differences by the view, its compositions by the checker, its asks by the view +// (and the node agent from release 6), its differences by the view, its asks by the view // and the teller, its puts by whoever follows one (ADR 0546), and the seat mesh-placement's placement, proposals and // candidates by the node agents, the grants part, mesh-config and the view (ADR 0567 §9), and its placement by // the builder, the checker and the indexer, through mesh-sdk's toolchain gate (issue 774, ADR 0549 §3). @@ -100,7 +88,6 @@ var keptStates = []announce.State{ {Name: delivery.PlacementCandidatesState, Readable: true}, {Name: delivery.PlacementPulsesState, Readable: true}, {Name: "asks", Readable: true}, - {Name: "compositions", Readable: true}, {Name: "differences", Readable: true}, {Name: "puts", Readable: true}, {Name: "wants", Readable: true}, @@ -135,11 +122,6 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any logf("the bucket %s cannot be opened (%v): it is bound once the bus says it was made", name, err) continue } - if name == keptBucket && err != nil { - // mesh-config's held values (ADR 0565 §5): without them a composition lays none, said in the journal. - logf("the bucket %s cannot be opened (%v): no held value is laid on a composition until it is bound", name, err) - continue - } if name == provisions.ProvisionsBucket && err != nil { // The seat mesh-provisioner's state, with its own delivery (ADR 0570 §4): bound once the bus says it was made; // until then no want waits for a provision, as before. @@ -175,7 +157,7 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any // 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, + case delivery.SeatHoldersBucket, provisions.ProvisionsBucket, assign.PoliciesBucket, assign.FindingsBucket, delivery.AssignerPutsBucket: return true } @@ -211,10 +193,7 @@ 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) - 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) + return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, logger) } // findingsOf is where the assigner's said conditions go as findings (issue 1952 WM6): its findings bucket, which the @@ -246,11 +225,9 @@ func (e emitting) Emit(eventType string, body any) error { return e.Conn.Emit(ev // watched are the buckets whose every value a pass reads; the manifests are read by digest when a pass needs one, and // watched only for keys, so a manifest that arrives wakes a composition that waits for it. var watched = []string{ - buscompose.ResolvedBucket, delivery.PartGrantsBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket, - delivery.CandidatesBucket, delivery.ModulesBucket, delivery.PlacementBucket, delivery.PlacementProposalsBucket, @@ -259,18 +236,16 @@ var watched = []string{ delivery.PresenceBucket, delivery.HaveBucket, delivery.AssignerAsksBucket, - delivery.AssignerCompositionsBucket, delivery.AssignerDifferencesBucket, delivery.AssignerWantsBucket, assign.PoliciesBucket, provisions.ProvisionsBucket, - keptBucket, delivery.SeatHoldersBucket, } // 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, buildLog logWatch, 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, logger *log.Logger) error { w := newWorld() dirty := make(chan struct{}, 1) wake := func() { @@ -279,22 +254,6 @@ 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 { @@ -402,7 +361,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, } } // And the two that choose where grants are read from, with the same delivery as the grants part that writes one. - for _, name := range []string{delivery.PartGrantsBucket, assign.PoliciesBucket, provisions.ProvisionsBucket, keptBucket, delivery.SeatHoldersBucket} { + for _, name := range []string{delivery.PartGrantsBucket, assign.PoliciesBucket, provisions.ProvisionsBucket, delivery.SeatHoldersBucket} { if kvs[name] == nil && js != nil { name := name process.AwaitBucket(ctx, js, nc, name, "until then the bucket is not read", @@ -427,7 +386,6 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, } } p := &passer{kvs: kvs, w: w, logf: logger.Printf, say: say, findings: findings, said: map[string]bool{}, answeredAt: map[string]uint64{}} - p.asks = newComposeAsks(ctx, buildrun.Asker{Publisher: buildrun.JetStreamAsks{JS: js}, Waiter: buildrun.JetStreamAsks{JS: js}}, wake, logger.Printf) if bus != nil && kvs[delivery.AssignerPutsBucket] != nil { p.acts = newPutActs(bus, wake, logger.Printf) } 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 513ca42..6adca56 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 @@ -40,14 +40,14 @@ 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 + if slices.Contains(readFamilies, r) { + continue // a state kept one bucket per node, read by its prefix (ADR 0593 §6), not one bucket } reads = append(reads, strings.Replace(r, ".", "_", 1)) } - for _, l := range readLogs { + for _, l := range readFamilies { if !slices.Contains(m.Reads, l) { - t.Errorf("the program reads the log %s, the manifest does not name it", l) + t.Errorf("the program reads the per-node state %s, the manifest does not name it", l) } } var states []string @@ -113,7 +113,7 @@ func TestOpensWhatItsManifestNames(t *testing.T) { } 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, nil, log.New(io.Discard, "", 0)) + go follow(ctx, nil, nil, kvs, func(asks.Condition) error { return nil }, assign.Findings{}, 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/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index 2764e8f..e372689 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -14,7 +14,6 @@ import ( "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/provisions" - "git.novox.be/novox/mesh-sdk/go/sdkversion" "mesh-delivery-assigner/internal/assign" ) @@ -40,7 +39,6 @@ type passer struct { unplaced map[string]bool // answeredAt is, by the id of a backup ask, how many values the have had when its answer was written into the want. answeredAt map[string]uint64 - asks *composeAsks // now is the clock; nil is time.Now. now func() time.Time // neither is, by node and module, each module on neither list whose condition this process raised (issue 1945). Nil until the @@ -81,7 +79,6 @@ func (p *passer) clock() time.Time { func (p *passer) pass(ctx context.Context) { p.placementSeat(ctx) - p.compositions(ctx) p.wants(ctx) p.puts(ctx) p.nodeWants(ctx) @@ -478,57 +475,6 @@ func (p *passer) sayNeither(now, stale map[string]neitherOne, left map[string]st } } -// compositions writes the composition check's verdict of every group the indexer holds a candidate for, under the -// checker's key of that check: waits, with why, while the indexer holds what it composes from only in part. -func (p *passer) compositions(ctx context.Context) { - if p.sdk == "" { - p.sdk = sdkversion.Version() - } - written := p.w.read(delivery.AssignerCompositionsBucket) - trunk := p.w.trunk() - var mesh *assign.Mesh - seen := map[string]bool{} - if p.asks != nil { - defer p.asks.sweep(seen) - } - for key, value := range p.w.read(delivery.CandidatesBucket) { - c, err := delivery.DecodeCandidate(value) - if err != nil { - p.logf("the candidate %s cannot be read, so it is not composed: %v", key, err) - continue - } - out, err := delivery.CompositionKey(c.Group, c.HeadsHash) - if err != nil { - p.logf("the candidate %s names no composition key: %v", key, err) - continue - } - began := time.Now() - ch, waits := p.w.change(ctx, kvGetter(p.kvs[delivery.ManifestsBucket]), c, trunk) - var v delivery.CompositionVerdict - if waits != "" { - v = delivery.CompositionVerdict{Group: c.Group, Heads: c.HeadsHash, State: delivery.CompositionWaits, Words: waits} - } else { - if mesh == nil { - m := p.w.mesh(ctx, kvGetter(p.kvs[delivery.ManifestsBucket])) - mesh = &m - } - if ch.SDKHead != "" && p.asks != nil { - // A group with a mesh-sdk member is composed by the composer built at its head (ADR 0482 clauses 3 to 5). - v = p.composeAtHead(ctx, *mesh, ch, seen, began) - } else { - v = assign.Compose(*mesh, ch, p.sdk, time.Since(began)) - } - } - if before, err := delivery.DecodeCompositionVerdict(written[out]); err == nil { - before.Took, before.V = v.Took, 0 - if delivery.SameCompositionVerdict(before, v) { - continue - } - } - p.put(ctx, delivery.AssignerCompositionsBucket, out, func() ([]byte, error) { return delivery.EncodeCompositionVerdict(v) }) - } -} - // put writes one key of the assigner's own bucket; a write that fails is said and made again at the next pass. func (p *passer) put(ctx context.Context, bucket, key string, encode func() ([]byte, error)) { value, err := encode() diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index ee60a4e..df5b9dc 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -791,35 +791,18 @@ func TestARecordAskOfAHeldBackNodeSurvivesPassesAndItsYesMovesTheWant(t *testing } // 774/S26/AC1 (ADR 0549 §5, ADR 0567 §8): the seat's placement alone names where modules run: no controller state is read, so a -// module in an assignment only is no module of its node; before any placement is read nothing is placed and the check composes -// no input; placed, the check composes over the input the nodes compose from (frombus, 1126, ADR 0547 §2). +// module in an assignment only is no module of its node, and before any placement is read nothing is placed. func aPlacementAloneNamesWhereModulesRun(t *testing.T) { p := newPasser(buckets{}, time.Date(2026, 10, 26, 18, 0, 0, 0, time.UTC)) p.w.put("mesh-controller_assignments", "g14.b", []byte(`{"node":"g14","module":"b","build":"b-1"}`)) // read no more if got := p.w.assigned(); len(got) != 0 { t.Fatalf("before any placement: assigned %v", got) } - none := func(context.Context, string) ([]byte, error) { return nil, errNotThere } - if m := p.w.mesh(context.Background(), none); m.Input != nil || m.InputUnread != "" { - t.Errorf("before any placement the check composes over %+v (%q)", m.Input, m.InputUnread) - } placeOn(t, p.w, "g14", "c", "a") placeOn(t, p.w, "n2", "d") if got := p.w.assigned(); !slices.Equal(got["g14"], []string{"a", "c"}) || !slices.Equal(got["n2"], []string{"d"}) || len(got) != 2 { t.Errorf("the placement alone: assigned %v", got) } - if m := p.w.mesh(context.Background(), none); m.Input != nil || !strings.Contains(m.InputUnread, "g14") { - t.Errorf("a placed node with no record: input %+v, unread %q", m.Input, m.InputUnread) - } - placeOn(t, p.w, "g14", "a", "b", "c") - rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "g14", TunnelKey: "k"}}) - p.w.put(delivery.HaveBucket, "g14.node", rec) - rec2, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n2", TunnelKey: "k2"}}) - p.w.put(delivery.HaveBucket, "n2.node", rec2) - m := p.w.mesh(context.Background(), none) - if m.Input == nil || len(m.Input.Nodes) != 2 || !slices.Equal(m.Input.Nodes[0].Assigned, []string{"a", "b", "c"}) || m.Input.Nodes[0].Place.Key != "k" { - t.Errorf("the check's input from the bus: %+v (%q)", m.Input, m.InputUnread) - } } // 1126 (ADR 0547 §2 to §4): a node the catalogue places composes its own: its wants name the build alone and its own want diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go index e43ccd8..ceb2500 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go @@ -16,14 +16,12 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/asks" - "git.novox.be/novox/mesh-sdk/go/compose" - "git.novox.be/novox/mesh-sdk/go/compose/frombus" - "git.novox.be/novox/mesh-sdk/go/compose/judge" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/handacts" "git.novox.be/novox/mesh-sdk/go/process" "mesh-delivery-assigner/internal/assign" + "mesh-delivery-assigner/internal/judge" ) // The seat mesh-placement's holder in the assigner's process (novox/hq ADR 0567 §2-§8): each pass follows every open @@ -354,67 +352,35 @@ func moved(prop delivery.PlacementProposal, revs map[string]uint64) bool { return false } -// intended is, by proposal and the revisions it was laid over, what Intended computed, kept while this process runs. +// intendedOf is, by proposal and the revisions it was laid over, what Intended computed, kept while this process runs. func (p *passer) intendedOf(prop delivery.PlacementProposal, current map[string]delivery.Placement, cands map[string]delivery.PlacementCandidate) map[string]map[string]bool { key := fmt.Sprint(prop.ID, prop.Base) if got, ok := p.intended[key]; ok { return got } - got, _, _ := p.compute(prop, current, cands) + got, _ := p.compute(prop, current, cands) p.intended[key] = got return got } -// compute composes every node with the current placement and with the proposal laid over it (judge.Intended). -func (p *passer) compute(prop delivery.PlacementProposal, current map[string]delivery.Placement, cands map[string]delivery.PlacementCandidate) (map[string]map[string]bool, []string, []string) { +// compute is what the proposal changes on each node by its placement, and the nodes that answer it: every node with a +// placement or a record (judge.Intended, ADR 0593 §8). The holder composes nothing: each node says what its compositions +// change in its answer. +func (p *passer) compute(prop delivery.PlacementProposal, current map[string]delivery.Placement, cands map[string]delivery.PlacementCandidate) (map[string]map[string]bool, []string) { laid := map[string]*delivery.Placement{} for _, node := range prop.Nodes() { if c, ok := cands[node]; ok && !c.Reached { laid[node] = c.Placement } } - get := kvGetter(p.kvs[delivery.ManifestsBucket]) - r := worldReader{w: p.w, manifests: get} - // A read that fails is said, never dropped silently (novox/hq ADR 0578 §3); a refused manifest no longer fails it, and - // its module is held, named in the input's Refused. - var now, with *compose.Input - if in, err := frombus.Input(context.Background(), r, frombus.Mesh{}, frombus.Scope{Every: true}); err == nil { - now = &in - } else if p.logf != nil { - p.logf("proposal %s: the current composition input cannot be read, so what it changes is judged without it: %v", prop.ID, err) - } - if in, err := frombus.Input(context.Background(), r, frombus.Mesh{}, frombus.Scope{Every: true, Placements: laid}); err == nil { - with = &in - } else if p.logf != nil { - p.logf("proposal %s: the composition input with the proposal cannot be read, so what it changes is judged without it: %v", prop.ID, err) - } - if now != nil && len(now.Refused) > 0 && p.logf != nil { - var held []string - for _, m := range slices.Sorted(maps.Keys(now.Refused)) { - held = append(held, m+" ("+now.Refused[m]+")") - } - p.logf("proposal %s: held as refused by the indexer's judge: %s", prop.ID, strings.Join(held, "; ")) - } nodes := map[string]bool{} for n := range current { nodes[n] = true } - if records, err := r.Records(context.Background()); err == nil { - for n := range records { - nodes[n] = true - } + for n := range p.w.records() { + nodes[n] = true } - names := make([]string, 0, len(nodes)) - for n := range nodes { - names = append(names, n) - } - fps := map[string]string{} - for name, m := range p.w.trunk() { - if m.FP2 != "" { - fps[name] = m.FP2 - } - } - return judge.Intended(now, with, current, laid, names, fps) + return judge.Intended(current, laid, slices.Sorted(maps.Keys(nodes))) } // mustAnswer finds the nodes the proposal must hear from (§3), writes a reached candidate for each whose key it does not @@ -425,7 +391,7 @@ func (p *passer) mustAnswer(ctx context.Context, prop delivery.PlacementProposal return // the verbs write the candidates before the proposal: wait to hear them } } - intended, must, notes := p.compute(prop, current, cands) + intended, must := p.compute(prop, current, cands) p.intended[fmt.Sprint(prop.ID, prop.Base)] = intended for _, node := range must { if _, ok := cands[node]; ok { @@ -447,9 +413,6 @@ func (p *passer) mustAnswer(ctx context.Context, prop delivery.PlacementProposal } } prop.Must = must - if len(notes) > 0 { - prop.Words = strings.TrimPrefix(prop.Words+"; "+strings.Join(notes, "; "), "; ") - } p.writeProposal(ctx, prop, rev) } @@ -483,7 +446,7 @@ func (p *passer) lays(ctx context.Context) (func(string) bool, func(string) bool held[name] = true } } - records, _ := worldReader{w: p.w}.Records(ctx) + records := p.w.records() return func(m string) bool { return held[m] }, func(n string) bool { _, ok := records[n]; return ok } } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go index 64df14c..0e202c9 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go @@ -367,9 +367,16 @@ func TestTheSeatsVerbs(t *testing.T) { func wantsVerbs(t *testing.T) { fpOld, fpNew := delivery.SourcePrefix2+strings.Repeat("1", 64), delivery.SourcePrefix2+strings.Repeat("2", 64) st := memStore{} - for _, state := range []string{modulesState, manifestsState, policiesState, haveState, putsState, "findings"} { + for _, state := range []string{modulesState, manifestsState, policiesState, haveState, putsState, "findings", delivery.ComposedRead} { st[state] = &memKV{values: map[string][]byte{}} } + // ADR 0593 §6, §8: what each node published, every node's bucket as one state: anchor's own composition and its web at + // a fingerprint, and laptop's web refused there. + webKey, _ := delivery.ComposedKey("anchor", "web", fpNew) + refusedKey, _ := delivery.ComposedKey("laptop", "web", fpNew) + st[delivery.ComposedRead].values["anchor.node"], _ = delivery.EncodeComposed(delivery.Composed{Node: "anchor"}) + st[delivery.ComposedRead].values[webKey], _ = delivery.EncodeComposed(delivery.Composed{Node: "anchor", Module: "web", FP: fpNew}) + st[delivery.ComposedRead].values[refusedKey], _ = delivery.EncodeComposed(delivery.Composed{Node: "laptop", Module: "web", FP: fpNew, Refused: "a required value is unset"}) manifest := func(name, extra string) { raw := `{"module":"` + name + `","version":"1"` + extra + `}` digest := delivery.ManifestDigest([]byte(raw)) @@ -409,6 +416,11 @@ func wantsVerbs(t *testing.T) { got["composed"].(map[string]string)["anchor"] != "sha256:"+strings.Repeat("c", 64) || len(got["composed"].(map[string]string)) != 1 { t.Errorf("compare = %+v", got) } + pub, _ := out.(map[string]any)["published"].(map[string][]publishedRow) + if a, l := pub["anchor"], pub["laptop"]; len(a) != 2 || a[0].Module != "node" || a[1].Module != "web" || a[1].FP != fpNew || + !strings.HasPrefix(a[1].Digest, "sha256:") || len(l) != 1 || l[0].Refused != "a required value is unset" || l[0].Digest != "" { + t.Errorf("published = %+v (unread %v)", pub, out.(map[string]any)["published-unread"]) + } // Read by watch (issue 1953 L5f): the same answer, each watched state started once, and no get per key of the // large states (the modules, their manifests, the composed record). watches := map[string]int{} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go index febfd0b..4749b28 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go @@ -20,7 +20,7 @@ import ( // The seat mesh-wants' verbs (novox/hq ADR 0565 §7, §8, §9, §15; issue 1953 D4-D6), served through the tool runner beside the // put tools and mesh-placement's verbs. `upgrade` reads and sets a person's upgrade policy, kept in the state `policies` // (each set recorded as a hand act and said as policy-changed), and with compare lists the digest of the composition each node's -// own have names (the controller writes no policy any more, so no module's policy differs); `bus` says what a bus upgrade would do. The assigner's process reads +// own have names beside every composition each node published (ADR 0593 §8, composed.go); `bus` says what a bus upgrade would do. The assigner's process reads // `policies` when it judges a move and writes the finding policy-split while a split stands. // The states the verbs read, as the runtime names them. @@ -158,6 +158,12 @@ func (s *wantsTools) doUpgrade(args map[string]any, by string) (any, error) { } } out := map[string]any{"composed": composed} + // And what every node published (ADR 0593 §8), read only to compare. + if values, err := s.values(delivery.ComposedRead); err != nil { + out["published-unread"] = err.Error() + } else { + out["published"] = published(values) + } // The assigner's findings, its said conditions as the condition keeper reads them (§15). if found, err := s.values(health.FindingsState); err != nil { out["findings-unread"] = err.Error() 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 f80db5e..2277786 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -3,20 +3,14 @@ package main import ( "context" "errors" - "fmt" "slices" "sort" - "strings" "sync" "time" "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" - "git.novox.be/novox/mesh-sdk/go/compose/frombus" "git.novox.be/novox/mesh-sdk/go/delivery" "mesh-delivery-assigner/internal/assign" @@ -249,63 +243,11 @@ func (w *world) trunk() map[string]delivery.Module { return out } -// mesh is the mesh as the trunk holds it, for the composition check. -func (w *world) mesh(ctx context.Context, manifests getter) assign.Mesh { - m := assign.Mesh{Manifests: map[string]catalogue.Manifest{}, Unreadable: map[string]string{}, Digests: map[string]string{}} - m.Assigned = w.assigned() - for name, mod := range w.trunk() { - if mod.Manifest == "" { - continue - } - m.Digests[name] = mod.Manifest - // The indexer's verdict on it, carried in its value (ADR 0578 §3, §5): a refused module is held and named as one - // that cannot be read, and its verdict goes into the composer's input. - if v := mod.VerdictOf(name); v != nil { - if refused, _ := delivery.TakeManifestVerdict(v); refused { - if m.Verdicts == nil { - m.Verdicts = map[string]delivery.ManifestVerdict{} - } - m.Verdicts[name] = *v - m.Unreadable[name] = v.Why() - continue - } - } - p := w.manifest(ctx, manifests, mod.Manifest) - if p.err != "" { - m.Unreadable[name] = p.err - continue - } - m.Manifests[name] = p.manifest - } - // Once the catalogue places the nodes, the check composes over the input they compose from (ADR 0547 §2, ADR 0549 - // §3): one composer's input for the check and the nodes. An input that cannot be read is said, and the nodes and their - // assignments alone are composed, as before any placement. - if _, seeded := w.placed(); seeded { - in, err := frombus.Input(ctx, worldReader{w: w, manifests: manifests}, frombus.Mesh{}, frombus.Scope{Every: true}) - if err != nil { - m.InputUnread = err.Error() - } else { - m.Input = &in - } - } - return m -} - -// worldReader is what the assigner watched as frombus reads the bus: the seat's placement, the indexer's modules, each node's -// record in its own have key, the manifests by digest; the check reads no config value, own secret or credential (the -// composition it asks resolves, and stands in for what it does not carry). -type worldReader struct { - w *world - manifests getter -} - -func (r worldReader) Placements(context.Context) (map[string]delivery.Placement, error) { - return r.w.placements(), nil -} - -func (r worldReader) Records(context.Context) (map[string]delivery.NodeRecord, error) { +// records is each node's record in its own have key (`.node`): the nodes a placement proposal is answered by, +// with those placement names (ADR 0593 §8). +func (w *world) records() map[string]delivery.NodeRecord { out := map[string]delivery.NodeRecord{} - for key, value := range r.w.read(delivery.HaveBucket) { + for key, value := range w.read(delivery.HaveBucket) { node, _, isNode, err := delivery.ParseHaveKey(key) if err != nil || !isNode { continue @@ -314,136 +256,7 @@ func (r worldReader) Records(context.Context) (map[string]delivery.NodeRecord, e out[node] = *h.Record } } - return out, nil -} - -func (r worldReader) Modules(context.Context) (map[string]delivery.Module, error) { - return r.w.trunk(), nil -} - -func (r worldReader) Manifest(ctx context.Context, digest string) ([]byte, error) { - key, err := delivery.ManifestsKey(digest) - if err != nil { - return nil, err - } - raw, err := r.manifests(ctx, key) - if err != nil { - return nil, err - } - m, err := delivery.DecodeManifest(raw) - if err != nil { - return nil, err - } - return []byte(m.JSON), nil -} - -// Values is one module's config values on one node as mesh-config resolved them, each a value its source lays -// (catalogue.LaidValue): the holder composes every node a proposal touches with them (ADR 0567 §9). -func (r worldReader) Values(_ context.Context, node, module string) (map[string]any, error) { - var out map[string]any - prefix := node + "." + module + "." - for key, raw := range r.w.read(buscompose.ResolvedBucket) { - name, ok := strings.CutPrefix(key, prefix) - if !ok || strings.Contains(name, ".") { - continue - } - if v, laid := catalogue.LaidValue(raw); laid { - if out == nil { - out = map[string]any{} - } - out[name] = v - } - } - return out, nil -} - -func (worldReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow, error) { - return nil, nil -} - -func (worldReader) Credentials(context.Context, string, []compose.CredentialKey, []compose.CredentialKey) ([]compose.Credential, []compose.Issued, error) { - return nil, nil, nil -} - -// Build and Archive are read only for a node composing itself (frombus.Scope without Every): the check of every node -// composes from the manifests the indexer read, and reads no build and no archive. -func (worldReader) Build(context.Context, string, string) (*delivery.BuilderBuild, error) { - return nil, nil -} - -func (worldReader) Archive(_ context.Context, url, _ string) ([]byte, error) { - return nil, fmt.Errorf("the composition check reads no archive (%s)", url) -} - -func (r worldReader) Served(ctx context.Context, node, module string, keys []string) (map[string]any, error) { - all, err := r.Values(ctx, node, module) - var out map[string]any - for _, k := range keys { - if v, ok := all[k]; ok { - if out == nil { - out = map[string]any{} - } - out[k] = v - } - } - return out, err -} - -// change is a candidate's change for the composition check, or why the composition waits for it. -func (w *world) change(ctx context.Context, manifests getter, c delivery.Candidate, trunk map[string]delivery.Module) (assign.Change, string) { - ch := assign.Change{Group: c.Group, Heads: c.HeadsHash, Manifests: map[string]catalogue.Manifest{}, Unreadable: map[string]string{}, - Twice: map[string][]string{}, Digests: map[string]string{}, Members: map[string]string{}} - heads := map[string]string{} - for _, p := range c.Pulls { - heads[fmt.Sprintf("%s#%d", p.Repository, p.Number)] = p.Head.Head - } - ch.Held = assign.HeldOf(w.read(keptBucket), heads) - definedIn := map[string]map[string]bool{} - for _, p := range c.Pulls { - if p.Repository == assign.SDKRepository { - ch.SDKHead = p.Head.Head - } - ch.Members[p.Repository] = p.Head.Head - for _, name := range p.Modules { - fp, ok := p.Fps[name] - if !ok || fp.Manifest == "" { - if ok && fp.None2 == noModuleWords { - ch.Removed = append(ch.Removed, name) - continue - } - return ch, fmt.Sprintf("the indexer holds no manifest of %s for %s#%d yet", name, p.Repository, p.Number) - } - if fp.FP2 != "" && fp.FP2 != trunk[name].FP2 { - ch.Rebuilt++ - } - got := w.manifest(ctx, manifests, fp.Manifest) - if strings.HasPrefix(got.err, "the indexer holds no manifest") { - return ch, got.err - } - ch.Digests[name] = fp.Manifest - if got.err != "" { - ch.Unreadable[name] = got.err - continue - } - ch.Manifests[got.module] = got.manifest - if definedIn[got.module] == nil { - definedIn[got.module] = map[string]bool{} - } - definedIn[got.module][p.Repository] = true - if t, ok := trunk[got.module]; ok && t.Repository != "" && t.Repository != p.Repository { - definedIn[got.module][t.Repository] = true - } - } - } - for name, repos := range definedIn { - if len(repos) > 1 { - for r := range repos { - ch.Twice[name] = append(ch.Twice[name], r) - } - sort.Strings(ch.Twice[name]) - } - } - return ch, "" + return out } // online is, by node, whether the watchman holds the node online. @@ -594,40 +407,6 @@ func (w *world) changed(bucket, key string) uint64 { return w.changes[bucket+"|"+key] } -// 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 - } - w.mu.Lock() - defer w.mu.Unlock() - have, ok := w.trunkBuilt[e.Module] - if ok && !e.At.After(have.At) { - return false - } - 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). -var keptBucket = "mesh-config_" + delivery.MeshConfigKeptState - // ownComposed is a placed node's own composition, as the digest its own have names (`.node` of the haves, Composed): // what rule 7 reads. A node whose have names none is not composed yet. func (w *world) ownComposed(node string) (delivery.ComposedDigest, bool) { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world_test.go index eb14547..51adf59 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world_test.go @@ -2,15 +2,12 @@ package main import ( "context" - "os" "slices" "strings" "testing" "time" "git.novox.be/novox/mesh-sdk/go/delivery" - - "mesh-delivery-assigner/internal/assign" ) // fakeManifests is the indexer's manifests bucket in memory. @@ -45,76 +42,6 @@ func pull(repo string, number int, modules map[string]delivery.CandidateModule) return p } -// 1297/B3b: a candidate's change reads each module's manifest by its digest, counts the modules rebuilt, takes a -// module with no manifest in the preview as removed, a mesh-sdk member's head, and a module the trunk holds from -// another repository as defined twice. -func TestTheChangeOfACandidate(t *testing.T) { - theWorldCountsTheValuesAKeyTakes(t) - f := fakeManifests{} - web := f.add(t, "web", `{"module":"web","version":"2"}`) - mail := f.add(t, "mail", `{"module":"mail","version":"1"}`) - trunk := map[string]delivery.Module{"web": {FP2: fp('a'), Repository: "novox/web"}, "mail": {FP2: fp('c'), Repository: "novox/mail"}} - c := delivery.Candidate{Group: "g", HeadsHash: "0123456789ab", Pulls: []delivery.CandidatePull{ - pull("novox/web", 1, map[string]delivery.CandidateModule{"web": {FP2: fp('b'), Manifest: web}, "old": {None2: noModuleWords}}), - pull("novox/other", 2, map[string]delivery.CandidateModule{"mail": {FP2: fp('c'), Manifest: mail}}), - pull(assign.SDKRepository, 3, nil), - }} - ch, waits := newWorld().change(context.Background(), f.get, c, trunk) - if waits != "" { - t.Fatalf("the change waits: %s", waits) - } - if ch.Manifests["web"].Version != "2" || ch.Rebuilt != 1 || !slices.Equal(ch.Removed, []string{"old"}) || ch.SDKHead != strings.Repeat("3", 40) { - t.Errorf("change = %+v", ch) - } - if !slices.Equal(ch.Twice["mail"], []string{"novox/mail", "novox/other"}) || len(ch.Twice) != 1 { - t.Errorf("twice = %v, want mail in novox/mail and novox/other", ch.Twice) - } -} - -// 1297/B3b: a candidate whose module has no manifest digest yet, or whose manifest the indexer does not hold yet, waits, -// saying for what; a manifest that does not decode is the change's, not a wait (one with a field the assigner does not -// know is decoded with it set aside, novox/hq ADR 0578 §6). -func TestAChangeWaitsForItsManifests(t *testing.T) { - f := fakeManifests{} - bad := f.add(t, "web", `{"module":"web","version":"2","requires":"x"}`) - for _, c := range []delivery.CandidatePull{ - pull("novox/web", 1, map[string]delivery.CandidateModule{"web": {None2: "no preview yet"}}), - pull("novox/web", 1, map[string]delivery.CandidateModule{"web": {FP2: fp('b'), Manifest: strings.Repeat("9", 64)}}), - } { - _, waits := newWorld().change(context.Background(), f.get, delivery.Candidate{Group: "g", HeadsHash: "0123456789ab", Pulls: []delivery.CandidatePull{c}}, nil) - if waits == "" { - t.Errorf("a change without its manifest %+v does not wait", c.Fps) - } - } - ch, waits := newWorld().change(context.Background(), f.get, delivery.Candidate{Group: "g", HeadsHash: "0123456789ab", - Pulls: []delivery.CandidatePull{pull("novox/web", 1, map[string]delivery.CandidateModule{"web": {FP2: fp('b'), Manifest: bad}})}}, nil) - if waits != "" || ch.Unreadable["web"] == "" { - t.Errorf("an unreadable manifest: waits %q, change %+v", waits, ch) - } -} - -// A manifest of the change whose config value falls back to a predefined key (novox/hq ADR 0348 §7) is read, not -// unreadable for the catalogue's predefined.json the assigner never holds: the merge gate judges the fallback in the -// catalogue, and the indexer reads it the same way (catalogue.ParseManifestForInputs, issue 1751). A group whose -// mesh-catalog member touched pacman and xterm failed its composition on this (group check 18df81380db3). The manifest is -// xterm's module.json from mesh-catalog at 835bf22b. -func TestAManifestFallingBackToAPredefinedKeyIsReadable(t *testing.T) { - raw, err := os.ReadFile("testdata/catalogue-835bf22b-xterm.module.json") - if err != nil { - t.Fatal(err) - } - if !strings.Contains(string(raw), `"predefined"`) { - t.Fatal("the fixture names no predefined key") - } - f := fakeManifests{} - xterm := f.add(t, "xterm", string(raw)) - ch, waits := newWorld().change(context.Background(), f.get, delivery.Candidate{Group: "g", HeadsHash: "0123456789ab", - Pulls: []delivery.CandidatePull{pull("novox/mesh-catalog", 1, map[string]delivery.CandidateModule{"xterm": {FP2: fp('b'), Manifest: xterm}})}}, nil) - if waits != "" || ch.Unreadable["xterm"] != "" || ch.Manifests["xterm"].Module != "xterm" { - t.Errorf("xterm: waits %q, unreadable %q, want read", waits, ch.Unreadable["xterm"]) - } -} - // Review round 1 (items 3 and 4): the world counts the values a key takes, so a have written again after an answer is seen. func theWorldCountsTheValuesAKeyTakes(t *testing.T) { w := newWorld() diff --git a/modules/mesh-delivery-assigner/go.mod b/modules/mesh-delivery-assigner/go.mod index d0606dc..1b3d618 100644 --- a/modules/mesh-delivery-assigner/go.mod +++ b/modules/mesh-delivery-assigner/go.mod @@ -9,7 +9,6 @@ require ( ) require ( - git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun v0.0.0-00010101000000-000000000000 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect github.com/google/go-tpm v0.9.8 // indirect github.com/klauspost/compress v1.20.0 // indirect @@ -22,4 +21,3 @@ require ( golang.org/x/time v0.15.0 // indirect ) -replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun => ../mesh-delivery-builder/buildrun diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 2495a66..d5e14e0 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -33,9 +33,11 @@ func TestManifest(t *testing.T) { } reads := slices.Clone(m.Reads) slices.Sort(reads) - want := []string{"mesh-build-go.steps", "mesh-config.kept", "mesh-config.resolved", - "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"} + // ADR 0593 §8: it composes nothing, so it reads no candidate, held or resolved value, compose step or build log; it reads + // every node's published compositions to compare them. + want := []string{"mesh-delivery-builder.builds", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", + "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-grants.grants", "mesh-host.composed", + "mesh-provisioner.provisions", "node-engine.have"} if !reflect.DeepEqual(reads, want) { t.Errorf("reads = %v, want exactly %v", reads, want) } @@ -43,7 +45,7 @@ func TestManifest(t *testing.T) { for _, s := range m.State { states[s.Name] = s.History } - if want := map[string]int{"asks": 5, "compositions": 5, "differences": 64, "puts": 20, "wants": 20, + if want := map[string]int{"asks": 5, "differences": 64, "puts": 20, "wants": 20, "placement": 64, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5}; !reflect.DeepEqual(states, want) { t.Errorf("state = %v, want %v", states, want) } @@ -53,10 +55,10 @@ func TestManifest(t *testing.T) { if m.SecretsOwner != "mesh-delivery-assigner" { t.Errorf("secrets-owner = %q", m.SecretsOwner) } - // 1906/D3: it says its conditions on the operator channel; 1622: it puts its compose asks on the Go build seat, as the - // checker puts its checks. - if !slices.Equal(m.Uses, []string{"node-build-go", "operator-channel"}) { - t.Errorf("uses = %v, want exactly node-build-go and operator-channel", m.Uses) + // 1906/D3: it says its conditions on the operator channel; it asks the Go build seat nothing since the composition is + // the node-engine's (ADR 0593 §7). + if !slices.Equal(m.Uses, []string{"operator-channel"}) { + t.Errorf("uses = %v, want exactly operator-channel", m.Uses) } // ADR 0546: it serves its own tools (put, release, puts) from its bundle through the tool runner (the artifact loads the // module), says each put's step as the event put, and calls the hand-act seat and the bus node's backup holder. diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index 43f2379..3c7deb4 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -24,22 +24,17 @@ } ], "reads": [ - "mesh-build-go.steps", - "mesh-config.kept", - "mesh-config.resolved", - "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-host.composed", "mesh-provisioner.provisions", "node-engine.have" ], "uses": [ - "node-build-go", "operator-channel" ], "invokes": [ @@ -63,7 +58,6 @@ { "name": "candidates", "history": 5 }, { "name": "pulses", "key-ttl": true }, { "name": "asks", "history": 5 }, - { "name": "compositions", "history": 5 }, { "name": "differences", "history": 64 }, { "name": "puts", "history": 20 }, { "name": "wants", "history": 20 },