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 32069148..166ab4d1 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 @@ -12,6 +12,7 @@ import ( "path/filepath" "slices" "strings" + "sync" "testing" "time" @@ -30,6 +31,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/provisions" "mesh-delivery-assigner/internal/judge" + "git.novox.be/novox/mesh-delivery/sdk/enginereads" + "mesh-delivery-assigner/internal/assign" ) @@ -92,7 +95,14 @@ func startBed(t *testing.T, leaveOut string) *bed { 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 { + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}); err != nil { + t.Fatal(err) + } + // The assigner's durable consumer on EVENTS, as the grants part makes it from the manifest's consumes: a node-engine seat's + // event names the engine's machine last, so the filter is the event with every machine (mesh-grants buscompose + // machineEvent; the grants part's own test holds the derivation). + if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-assigner", + FilterSubjects: []string{enginereads.SubjectPrefix + "*"}, AckPolicy: jetstream.AckExplicitPolicy}); err != nil { t.Fatal(err) } // The operator-channel seat's stream of conditions, which the assigner says its conditions on. @@ -490,6 +500,116 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) { if w := b.wantNow(t, "ace", "app", "the consumer's want once its provision is provided", func(w delivery.Want) bool { return w.FP == fpOf('3') }); w.FP != fpOf('3') { t.Errorf("app's want = %+v", w) } + + // ADR 0618 §1, §3, §4, ADR 0624 §2: ace's engine answers the node-engine seat's manifest-reads; a build written only in a + // form it does not state is held on ace, naming the form, and moves once the engine says manifest-reads-changed on its + // subject and states the form; no timer moves it. g14's engine answers nothing and has no record: read as the baseline. + stated := catalogue.ManifestReads() + var statedMu sync.Mutex + nc, err := nats.Connect(b.srv.ClientURL()) + if err != nil { + t.Fatal(err) + } + defer nc.Close() + if _, err := nc.Subscribe("mesh.seat.node-engine.tool.manifest-reads.ace", func(m *nats.Msg) { + statedMu.Lock() + answer, _ := delivery.EncodeManifestReadsAnswer(delivery.ManifestReadsAnswer{Node: "ace", ManifestReads: stated, StatedAt: time.Now()}) + statedMu.Unlock() + reply, _ := json.Marshal(map[string]any{"result": json.RawMessage(answer), "node": "ace"}) + _ = m.Respond(reply) + }); err != nil { + t.Fatal(err) + } + say := func(why string) { + body, _ := delivery.EncodeManifestReadsChanged(delivery.ManifestReadsChanged{Node: "ace", ManifestReads: stated, At: time.Now(), Why: why}) + if _, err := b.js.Publish(b.ctx, enginereads.SubjectPrefix+"ace", body); err != nil { + t.Fatal(err) + } + } + say(delivery.ReadsStarted) + formsOld, formsNew := fpOf('5'), fpOf('6') + formsDigest := b.manifest(t, "forms", `{"module":"forms","version":"1"}`) + formsAt := func(fp string) { + mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: formsDigest, Repository: "novox/forms", Commit: strings.Repeat("3", 40), Path: delivery.RootPath}) + b.putJSON(t, delivery.ModulesBucket, "forms", mod) + } + for _, fp := range []string{formsOld, formsNew} { + build := delivery.BuilderBuild{State: delivery.BuildDone, By: "builder.novox.1", Step: "pushed", Manifest: formsDigest} + if fp == formsNew { + build.Needs = []string{"form:new"} + } + built, err := delivery.EncodeBuilderBuild(build) + if err != nil { + t.Fatal(err) + } + bkey, _ := delivery.BuilderBuildKey("forms", fp) + b.putJSON(t, delivery.BuilderBuildsBucket, bkey, built) + gkey, _ := delivery.PartGrantKey("forms", fp) + grant, _ := delivery.EncodeGrant(delivery.Grant{}) + b.putJSON(t, delivery.PartGrantsBucket, gkey, grant) + } + runsFormsOld, _ := delivery.EncodeHave(delivery.Have{FP: formsOld, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}) + b.putJSON(t, delivery.HaveBucket, "ace.forms", runsFormsOld) + b.place(t, "ace", "forms") + formsAt(formsOld) + b.wantNow(t, "ace", "forms", "the want at what ace runs, a build in the first form", func(w delivery.Want) bool { return w.FP == formsOld }) + formsAt(formsNew) + statement := func(node string, ok func(enginereads.Statement) bool) enginereads.Statement { + var last enginereads.Statement + for deadline := time.Now().Add(15 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if e, err := b.buckets[engineReadsBucket].Get(b.ctx, node); err == nil { + if last, err = enginereads.Decode(e.Value()); err == nil && ok(last) { + return last + } + } + } + t.Fatalf("%s's engine reads are %+v", node, last) + return last + } + statement("ace", func(s enginereads.Statement) bool { return s.From == enginereads.FromVerb && s.Answered }) + if s := statement("g14", func(s enginereads.Statement) bool { return !s.Answered }); s.From != enginereads.FromNone { + t.Errorf("g14, whose engine answers nothing and which has no record, reads %+v", s) + } + if w := b.wantNow(t, "ace", "forms", "the want held on ace, naming the form", func(w delivery.Want) bool { + return w.Newest == formsNew && strings.Contains(w.Words, "form:new") + }); w.FP != formsOld || w.Rule != delivery.RuleComposed { + t.Errorf("forms' want = %+v, want it held at what ace runs under rule 7", w) + } + statedMu.Lock() + stated = append(slices.Clone(stated), "form:new") + statedMu.Unlock() + say(delivery.ReadsStarted) + b.wantNow(t, "ace", "forms", "the want moved once ace's engine said it reads the form", func(w delivery.Want) bool { return w.FP == formsNew }) + // assigner_engine_reads asks ace's engine again through the state: the assigner answers there with what it states now. + again, err := enginereads.AskAgainOf(bedState{b, engineReadsBucket}, "ace", "n1.ada", "a person asks", time.Now(), 15*time.Second) + if err != nil || again.From != enginereads.FromVerb || !slices.Contains(again.ManifestReads, "form:new") || again.AskAgain != nil { + t.Errorf("the ask again of ace answered %+v, %v", again, err) + } +} + +// bedState is one of the bed's buckets as the engine reads' tool reads it. +type bedState struct { + b *bed + bucket string +} + +func (s bedState) Keys() ([]string, error) { return s.b.buckets[s.bucket].Keys(s.b.ctx) } +func (s bedState) Get(key string) ([]byte, uint64, bool, error) { + e, err := s.b.buckets[s.bucket].Get(s.b.ctx, key) + if err != nil { + return nil, 0, false, nil + } + return e.Value(), e.Revision(), true, nil +} +func (s bedState) Put(key string, value []byte) (uint64, error) { return s.b.buckets[s.bucket].Put(s.b.ctx, key, value) } +func (s bedState) Delete(key string) error { return s.b.buckets[s.bucket].Delete(s.b.ctx, key) } +func (s bedState) Await(key string, after uint64, done func([]byte) bool, bound time.Duration) ([]byte, error) { + for deadline := time.Now().Add(bound); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if v, rev, found, _ := s.Get(key); found && rev > after && done(v) { + return v, nil + } + } + return nil, fmt.Errorf("no answer at %s within %s", key, bound) } // 1297/B9 (the assigner's half, ADR 0528 clause 6): started whole, the assigner holds a fingerprint whose health check 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 ec894b3b..679199e8 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -28,6 +28,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/provisions" + "git.novox.be/novox/mesh-delivery/sdk/enginereads" "git.novox.be/novox/mesh-delivery/sdk/graph" "mesh-delivery-assigner/internal/assign" @@ -75,7 +76,15 @@ var readFamilies = []string{delivery.ComposedRead} // (ADR 0567 §1), named after the assigner as every module state is. 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} + delivery.PlacementCandidatesBucket, delivery.PlacementPulsesBucket, assign.FindingsBucket, assign.PoliciesBucket, delivery.SeatHoldersBucket, + engineReadsBucket} + +// engineReadsState is the assigner's state of what each node's engine states it reads (ADR 0618 §3), one key ``, as +// the verb manifest-reads last answered it or the fallback read it: what assigner_engine_reads lists and asks again. +const engineReadsState = "engine-reads" + +// engineReadsBucket is that state's bucket. +const engineReadsBucket = delivery.AssignerModule + "_" + engineReadsState // seatBuckets are the seat's states, new with it: until the controller has made them from the manifest and the bus grants // them, the assigner starts without them and binds each once the bus says it was created (issue 1790's kind). @@ -99,6 +108,7 @@ var keptStates = []announce.State{ {Name: health.FindingsState, Readable: true}, {Name: delivery.AssignerPoliciesState, Readable: true}, {Name: delivery.SeatHoldersState, Readable: true}, + {Name: engineReadsState, Readable: true}, } // buckets is every bucket the assigner opened, by name. @@ -156,6 +166,12 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any logf("the bucket %s cannot be opened (%v): no want waits for a verb it invokes until it is bound once the bus says it was made", name, err) continue } + if name == engineReadsBucket && err != nil { + // New with ADR 0618 §3: without it the statements the verb answers are kept in memory only, and + // assigner_engine_reads lists nothing. + logf("the bucket %s cannot be opened (%v): what each node's engine reads is kept in memory only until it starts again", name, err) + continue + } if name == delivery.PartGrantsBucket && err != nil { // The grants part's grants (ADR 0571 clause 9): without them no module counts as granted until it is bound. logf("the bucket %s cannot be opened (%v): no grant is read until it is bound once the bus says it was made", name, err) @@ -174,7 +190,7 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any func startsWithout(name string) bool { switch name { case delivery.SeatHoldersBucket, provisions.ProvisionsBucket, assign.PoliciesBucket, assign.FindingsBucket, - delivery.AssignerPutsBucket, delivery.SourceBucket, delivery.MergesBucket, graph.Bucket: + delivery.AssignerPutsBucket, delivery.SourceBucket, delivery.MergesBucket, graph.Bucket, engineReadsBucket: return true } return slices.Contains(seatBuckets, name) || name == delivery.PartGrantsBucket @@ -260,6 +276,7 @@ var watched = []string{ delivery.SourceBucket, delivery.MergesBucket, graph.Bucket, + engineReadsBucket, } // follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after @@ -275,6 +292,16 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, } errs := make(chan error, len(watched)+2) settled := make(chan struct{}, len(watched)+2) + // What each node's engine reads (ADR 0618 §3, §4): asked through the seat's verb, kept in the state engine-reads, read + // from the node's record where the engine does not answer (the fallback, said in the journal). + reads := &enginereads.Keeper{Fallback: w.recordReads, Changed: wake, Logf: logger.Printf} + if bus != nil { + reads.Call = bus.CallOn + } + if kv := kvs[engineReadsBucket]; kv != nil { + reads.Put = func(node string, v []byte) error { _, err := kv.Put(ctx, node, v); return err } + reads.Delete = func(node string) error { return kv.Delete(ctx, node) } + } watch := func(name string, opts ...jetstream.WatchOpt) error { kw, err := kvs[name].WatchAll(ctx, opts...) if err != nil { @@ -308,6 +335,10 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, if e.Operation() == jetstream.KeyValuePut { value = e.Value() } + if name == engineReadsBucket { + reads.Heard(e.Key(), value, value == nil) // it wakes a pass when a statement changed + continue + } w.putEntry(name, e.Key(), value, e.Revision(), e.Created()) wake() } @@ -405,7 +436,14 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, case <-settled: } } - p := &passer{kvs: kvs, w: w, logf: logger.Printf, say: say, findings: findings, said: map[string]bool{}, answeredAt: map[string]uint64{}} + p := &passer{kvs: kvs, w: w, logf: logger.Printf, say: say, findings: findings, said: map[string]bool{}, answeredAt: map[string]uint64{}, + reads: reads} + // The node-engine seat's manifest-reads-changed, for one node (ADR 0618 §4): that node is asked again, with no timer. + // Heard on the assigner's durable consumer on EVENTS, which the grants part makes from the manifest's consumes; where it + // is not there yet, a node is asked again only when it comes back online. + if js != nil { + hearReadsChanged(ctx, js, reads, logger.Printf) + } if bus != nil && kvs[delivery.AssignerPutsBucket] != nil { p.acts = newPutActs(bus, wake, logger.Printf) } @@ -470,3 +508,37 @@ func sayVia(ctx context.Context, js jetstream.JetStream) func(asks.Condition) er return err } } + +// hearReadsChanged hears the node-engine seat's event manifest-reads-changed on the assigner's durable consumer on EVENTS, +// and asks the node it names again. Any other event the consumer hands is acknowledged and left. +func hearReadsChanged(ctx context.Context, js jetstream.JetStream, reads *enginereads.Keeper, logf func(string, ...any)) { + account := process.Account{Node: thisNode, Module: delivery.AssignerModule} + cons, err := account.EventConsumer(ctx, js, process.EventsStream) + if err != nil { + logf("the consumer %s on %s does not exist yet, so %s is not heard until the assigner starts again (a node is asked again when it comes back online): %v", + account.EventConsumerName(), process.EventsStream, enginereads.Event, err) + return + } + cc, err := process.HearEvents(cons, func(m jetstream.Msg) { + if err := m.Ack(); err != nil { + logf("acknowledging %s: %v", m.Subject(), err) + } + node, ours, err := enginereads.NodeOf(m.Subject(), m.Data()) + switch { + case !ours: + case err != nil: + logf("%s holds what is not %s, and %s is asked again all the same: %v", m.Subject(), enginereads.Event, node, err) + reads.AskNode(node, enginereads.Event+" that does not read") + default: + reads.AskNode(node, enginereads.Event) + } + }, logf) + if err != nil { + logf("%s is not heard: %v", enginereads.Event, err) + return + } + go func() { + <-ctx.Done() + cc.Stop() + }() +} 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 cc520b40..02b3dc23 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -15,6 +15,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/provisions" + "git.novox.be/novox/mesh-delivery/sdk/enginereads" + "mesh-delivery-assigner/internal/assign" ) @@ -68,6 +70,40 @@ type passer struct { // saidWaits is the last said of the seat's wait for its states (sayWaits). saidWaits string intended map[string]map[string]map[string]bool + // reads keeps what each node's engine states it reads, asked through the node-engine seat's verb (ADR 0618 §3); nil reads + // each node's record, as the rules did before the verb. online is, by node, whether the last pass saw it online. + reads *enginereads.Keeper + online map[string]bool +} + +// engineReads sets what the rules read of each node's engine (ADR 0618 §3, §4): the statement the verb manifest-reads +// answered, or its record where the verb was not answered (the fallback) or not asked yet. A node is asked when the +// assigner first hears of it, and again when it comes back online, since an engine built before the event says none. +func (p *passer) engineReads(n assign.Nodes) { + if p.reads == nil { + return + } + known := p.w.knownNodes() + first := p.online == nil + for _, node := range known { + switch { + case first: + p.reads.AskNode(node, "the assigner started") + case !p.reads.Known(node): + p.reads.AskNode(node, "first heard of") + case !first && n.Online[node] && !p.online[node]: + p.reads.AskNode(node, "back online") + } + } + p.online = maps.Clone(n.Online) + if p.online == nil { + p.online = map[string]bool{} + } + for node, reads := range p.reads.Reads(known) { + if p.reads.Known(node) { + n.Reads[node] = reads + } + } } func (p *passer) clock() time.Time { @@ -120,6 +156,7 @@ func (p *passer) wants(ctx context.Context) { } } nodes := p.w.nodes(manifests, assigned) + p.engineReads(nodes) nodesOf := map[string][]string{} for node, modules := range assigned { for _, m := range modules { @@ -162,7 +199,7 @@ func (p *passer) wants(ctx context.Context) { for name := range names { fp := trunk[name].FP2 name := name - m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Needs: trunk[name].Needs, Chosen: chosenOf(chosen, name), Provisions: provisionsOf(name), Answering: answering(manifests, nodes), Offered: offered(manifests), Nodes: nodesOf[name], Now: time.Now(), Puts: puts[name], + m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Needs: trunk[name].Needs, Chosen: chosenOf(chosen, name), Provisions: provisionsOf(name), Answering: answering(manifests, nodes), Offered: offered(manifests), Nodes: nodesOf[name], Now: time.Now(), Puts: puts[name], FormNeeds: p.w.buildNeeds(name, fp), Self: func(node string) bool { return placed[node] }, Composed: func(node, fp string) (delivery.ComposedDigest, bool) { return p.w.composed(node, name, fp) // a placed node's own: the digest its own have names (ADR 0571 §6) 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 ec5f1bc0..d87a9bfa 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 @@ -14,9 +14,11 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/busretry" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-delivery/sdk/enginereads" "git.novox.be/novox/mesh-delivery/sdk/graph" "mesh-delivery-assigner/internal/assign" @@ -166,6 +168,35 @@ func TestAnAnswerCountsOnlyForTheAskingItAnswered(t *testing.T) { if !slices.Equal(n.Reads["ace"], []string{"word:module", "word:tool-runner"}) || len(n.Reads["g14"]) != 0 || len(n.Reads["novox"]) != 0 { t.Errorf("the nodes' reads are %v", n.Reads) } + // ADR 0618 §3: with the verb asked, a node's reads are what its engine answered; where the engine answers it does not + // serve the verb, its record is read once in its place (the fallback, issue 774); a node asked again when it comes + // back online, since an engine built before the event says none. + asked := map[string]int{} + call := func(node, key string, _ any) (json.RawMessage, error) { + asked[node]++ + if node == "g14" || key != enginereads.Invoke { + return nil, &busretry.Answered{Words: "the seat node-engine serves no verb manifest-reads"} + } + return delivery.EncodeManifestReadsAnswer(delivery.ManifestReadsAnswer{Node: node, ManifestReads: []string{"form:new"}, StatedAt: time.Now()}) + } + p := &passer{w: w, reads: &enginereads.Keeper{Call: call, Fallback: w.recordReads, Sync: true}} + on, _ := delivery.EncodePresence(delivery.Presence{Online: false}) + w.put(delivery.PresenceBucket, "ace", on) + n = w.nodes(nil, nil) + p.engineReads(n) + if !slices.Equal(n.Reads["ace"], []string{"form:new"}) || len(n.Reads["g14"]) != 0 || asked["ace"] != 1 { + t.Errorf("with the verb the nodes' reads are %v (asked %v)", n.Reads, asked) + } + if s := p.reads.Statements(); s["ace"].From != enginereads.FromVerb || s["g14"].From != enginereads.FromRecord { + t.Errorf("the statements are %+v", s) + } + p.engineReads(w.nodes(nil, nil)) + on, _ = delivery.EncodePresence(delivery.Presence{Online: true}) + w.put(delivery.PresenceBucket, "ace", on) + p.engineReads(w.nodes(nil, nil)) + if asked["ace"] != 2 { + t.Errorf("ace, back online, was asked %d times, want 2", asked["ace"]) + } }) } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go index 2ed4d101..ddf992b2 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go @@ -14,6 +14,8 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/delivery" + + "git.novox.be/novox/mesh-delivery/sdk/enginereads" ) // The assigner's tools (novox/hq ADR 0546), served through the tool runner from its bundle: a process never serves (ADR @@ -21,6 +23,9 @@ import ( // every step as the event put. They are the module's own tools; the seat mesh-placement's verbs are served beside them // (seat_tools.go, ADR 0567 §10). +// engineReadsTool lists what each node's engine reads, as the assigner uses it, and asks one node's engine again. +const engineReadsTool = "assigner_engine_reads" + // putsState is the assigner's state of puts, as the runtime names a module's own state. const putsState = "puts" @@ -358,6 +363,8 @@ func serveTools() error { }) wants.watched = stdio.NewWatchedStates() // the large states read whole from one watched copy (ADR 0580 clause 3) tools = append(tools, wants.tools()...) + // What each node's engine reads, as the assigner uses it (ADR 0618 §3): its own state engine-reads, asked again through it. + tools = append(tools, enginereads.Tool(engineReadsTool, "the assigner", enginereads.KeptState(stdio.State(engineReadsState)), false, time.Now)) return stdio.Serve("", tools) } 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 f37724e7..0cd66950 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -3,6 +3,7 @@ package main import ( "context" "errors" + "maps" "slices" "sort" "sync" @@ -420,6 +421,42 @@ func (w *world) built(module, fp string) bool { return err == nil && b.State == delivery.BuildDone } +// buildNeeds are the reference forms the builder's done build of module at fp is written in (the build key's needs, ADR +// 0618 §1): nil when it is not done or names none. +func (w *world) buildNeeds(module, fp string) []string { + key, err := delivery.BuilderBuildKey(module, fp) + if err != nil { + return nil + } + b, err := delivery.DecodeBuilderBuild(w.value(delivery.BuilderBuildsBucket, key)) + if err != nil || b.State != delivery.BuildDone { + return nil + } + return b.Needs +} + +// recordReads is what node's record `.node` states its engine reads (ADR 0614 §4), and whether the assigner holds +// a record of it: the fallback where the engine does not answer the verb manifest-reads (enginereads.FromRecord). +func (w *world) recordReads(node string) ([]string, bool) { + r, ok := w.records()[node] + return r.ManifestReads, ok +} + +// knownNodes is every node the assigner hears of: placed, with a record, or with a presence. +func (w *world) knownNodes() []string { + set := map[string]bool{} + for node := range w.placedNodes() { + set[node] = true + } + for node := range w.records() { + set[node] = true + } + for node := range w.online() { + set[node] = true + } + return slices.Sorted(maps.Keys(set)) +} + // composed is the composition of module for node: a placed node composes its own, and its composition is the digest its own // have names (ADR 0571 §6); an unplaced node has none, so it gets no want (and the pass says so, unplacedFindings). The // controller's copy (mesh-controller_composed) is read no more (issue 774). diff --git a/modules/mesh-delivery-assigner/go.mod b/modules/mesh-delivery-assigner/go.mod index 777c0c97..2ce41a81 100644 --- a/modules/mesh-delivery-assigner/go.mod +++ b/modules/mesh-delivery-assigner/go.mod @@ -24,3 +24,5 @@ require ( require git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000 replace git.novox.be/novox/mesh-delivery/sdk => ../../sdk + +replace git.novox.be/novox/mesh-sdk/go => ../../../0618-sdk/go diff --git a/modules/mesh-delivery-assigner/go.sum b/modules/mesh-delivery-assigner/go.sum index 7679a5dc..98dc72c4 100644 --- a/modules/mesh-delivery-assigner/go.sum +++ b/modules/mesh-delivery-assigner/go.sum @@ -1,5 +1,3 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8 h1:XEbIVImp6yGjA91rp0iZBoDdL5KfamjOuDLwWvqh0hc= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8/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/internal/assign/wants.go b/modules/mesh-delivery-assigner/internal/assign/wants.go index 5057439e..f029f074 100644 --- a/modules/mesh-delivery-assigner/internal/assign/wants.go +++ b/modules/mesh-delivery-assigner/internal/assign/wants.go @@ -12,6 +12,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/provisions" + "git.novox.be/novox/mesh-delivery/sdk/forms" "git.novox.be/novox/mesh-delivery/sdk/graph" ) @@ -80,6 +81,10 @@ type Module struct { // Needs are the reader capabilities the manifest at FP needs (the indexer's modules value, catalogue.ManifestNeeds, ADR // 0614 §4): a node's want moves only while that node's engine states each of them. Nil: none. Needs []string + // FormNeeds are the reference forms the module's build at FP is written in (`form:`, the builder's build key's + // needs, ADR 0618 §1): a node's want is held only while that node's engine reads none of them (ADR 0624 §2). Nil: the + // build names none, and holds nothing. + FormNeeds []string // Self says whether node composes the module itself: the catalogue's placement names the node (ADR 0547 §2). It is the one // source of a want's Self: every want the rules write or keep is derived from it (selfDerived), so a want written Self while // the node was placed does not outlive the placement (issue 1126). Nil: no node does. @@ -179,8 +184,14 @@ func holds(m Module, node string, n Nodes, ordering bool) (bool, int, string) { // manifest needs that the engine does not state, read as the baseline when it states none. It is held under rule 7 (the // composition), since no rule of its own is numbered in this release, with words naming the capability; it moves once the // engine states it. The node-engine's own new build is held only by a word of its own manifest no engine states. +// +// A build's reference forms hold the want only when the engine reads none of them (ADR 0624 §2): every engine reads a +// first form, so a build written in one is never held by its forms; the words name the forms it reads none of. func (m Module) readerLacking(node string, n Nodes) (string, bool) { missing := catalogue.MissingManifestReads(m.Needs, n.Reads[node]) + if len(missing) == 0 && forms.Current.Held(m.FormNeeds, n.Reads[node]) { + missing = slices.Clone(m.FormNeeds) + } if len(missing) == 0 { return "", false } diff --git a/modules/mesh-delivery-assigner/internal/assign/wants_test.go b/modules/mesh-delivery-assigner/internal/assign/wants_test.go index 5e61414d..a3ce7e58 100644 --- a/modules/mesh-delivery-assigner/internal/assign/wants_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/wants_test.go @@ -122,21 +122,33 @@ func TestEachRuleHoldsTheWantBack(t *testing.T) { needs []string reads map[string][]string held bool + // forms are the build's reference forms (ADR 0624 §2), and names what the hold's words name. + forms []string + names string }{ - {"a word the baseline lacks, on engines stating nothing", "mail", []string{"word:module", "word:tool-runner"}, nil, true}, + {"a word the baseline lacks, on engines stating nothing", "mail", []string{"word:module", "word:tool-runner"}, nil, true, nil, "word:tool-runner"}, {"the same, once both engines state it", "mail", []string{"word:module", "word:tool-runner"}, - map[string][]string{"ace": catalogue.ManifestReads(), "g14": catalogue.ManifestReads()}, false}, - {"the baseline's words on engines stating nothing", "mail", []string{"word:claims.serves", "seat:mesh-catalog", "form:config"}, nil, false}, - {"the node-engine's new build, whose contract has more words than its manifest uses", "mesh-host", hostNeeds, nil, false}, + map[string][]string{"ace": catalogue.ManifestReads(), "g14": catalogue.ManifestReads()}, false, nil, ""}, + {"the baseline's words on engines stating nothing", "mail", []string{"word:claims.serves", "seat:mesh-catalog", "form:config"}, nil, false, nil, ""}, + {"the node-engine's new build, whose contract has more words than its manifest uses", "mesh-host", hostNeeds, nil, false, nil, ""}, + // ADR 0624 §2, with form:new a form no engine reads unless it states it and artifact-store the first form. + {"g14 back stating {old}, a build carrying {old,new}: moves", "mail", nil, map[string][]string{"g14": {"word:module"}}, false, + []string{"form:artifact-store", "form:new"}, ""}, + {"an engine stating nothing, a build carrying the baseline's form: moves", "mail", nil, nil, false, []string{"form:artifact-store"}, ""}, + {"a build carrying {new} only: held, naming the form", "mail", nil, map[string][]string{"g14": {"word:module"}}, true, + []string{"form:new"}, "form:new"}, + {"the same, once g14's engine states it", "mail", nil, map[string][]string{"g14": {"form:new"}}, false, []string{"form:new"}, ""}, + {"a manifest needing an unstated word, its build in the first form: held, naming the word", "mail", []string{"word:tool-runner"}, nil, true, + []string{"form:artifact-store"}, "word:tool-runner"}, } { m, n := ready() - m.Name, m.Needs, n.Reads = c.name, c.needs, c.reads + m.Name, m.Needs, n.Reads, m.FormNeeds = c.name, c.needs, c.reads, c.forms n.Haves["ace"][c.name], n.Haves["g14"][c.name] = n.Haves["ace"]["mail"], n.Haves["g14"]["mail"] out := Wants(m, n, wantedOld(), nil) if moved := out.Wants["g14"].FP == fNew; moved == c.held { t.Errorf("%s: g14's want %+v, held by %+v", c.why, out.Wants["g14"], out.Rules["g14"]) } - if r := out.Rules["g14"]; c.held && (r.Rule != delivery.RuleComposed || !strings.Contains(r.Words, "word:tool-runner") || + if r := out.Rules["g14"]; c.held && (r.Rule != delivery.RuleComposed || !strings.Contains(r.Words, c.names) || !strings.Contains(r.Words, "node-engine of g14")) { t.Errorf("%s: held by %+v, want rule 7 naming the capability", c.why, r) } diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 087b34cd..0fdfc72d 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -47,7 +47,8 @@ func TestManifest(t *testing.T) { states[s.Name] = s.History } 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) { + "placement": 64, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5, + "engine-reads": 1}; !reflect.DeepEqual(states, want) { t.Errorf("state = %v, want %v", states, want) } if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 { @@ -78,8 +79,9 @@ func TestManifest(t *testing.T) { if seat, _ := catalogue.SeatNamed(catalogue.PlacementSeat); catalogue.CanHold(m, seat) != nil { t.Errorf("the assigner cannot hold the seat: %v", catalogue.CanHold(m, seat)) } - if !slices.Equal(m.Tools, []string{"assigner_put", "assigner_release", "assigner_puts"}) { - t.Errorf("the assigner serves tools %v, want its three put tools", m.Tools) + // ADR 0618 §3: and assigner_engine_reads, which lists and asks again what each node's engine reads. + if !slices.Equal(m.Tools, []string{"assigner_put", "assigner_release", "assigner_puts", "assigner_engine_reads"}) { + t.Errorf("the assigner serves tools %v, want its three put tools and assigner_engine_reads", m.Tools) } if len(m.Streams) != 1 || m.Streams[0].Name != delivery.PlacementStream || m.Streams[0].MaxAgeDays != delivery.PlacementStreamDays || !slices.Equal(slices.Sorted(slices.Values(m.Streams[0].Events)), slices.Sorted(slices.Values(delivery.PlacementEvents))) { @@ -87,9 +89,14 @@ func TestManifest(t *testing.T) { } if want := []string{ "seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", "seat:mesh-hand-acts.warranted", // ADR 0565 §9, §11: the seat alone - "seat:node-backup.backed-up", "seat:node-backup.now"}; !slices.Equal(slices.Sorted(slices.Values(m.Invokes)), want) { + "seat:node-backup.backed-up", "seat:node-backup.now", + "seat:node-engine.manifest-reads"}; !slices.Equal(slices.Sorted(slices.Values(m.Invokes)), want) { // ADR 0618 §3 t.Errorf("invokes = %v, want %v", m.Invokes, want) } + // ADR 0618 §4: it hears each node's engine say its reads changed. + if !slices.Equal(m.Consumes, []string{"node-engine.manifest-reads-changed"}) { + t.Errorf("consumes = %v", m.Consumes) + } if !slices.Equal(m.Emits, []string{"put"}) { t.Errorf("emits = %v, want put", m.Emits) } diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index 7f94deaa..392718f2 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -45,7 +45,11 @@ "seat:mesh-hand-acts.hand-acts", "seat:mesh-hand-acts.warranted", "seat:node-backup.backed-up", - "seat:node-backup.now" + "seat:node-backup.now", + "seat:node-engine.manifest-reads" + ], + "consumes": [ + "node-engine.manifest-reads-changed" ], "emits": [ "put" @@ -53,7 +57,8 @@ "tools": [ "assigner_put", "assigner_release", - "assigner_puts" + "assigner_puts", + "assigner_engine_reads" ], "state": [ { "name": "placement", "history": 64 }, @@ -66,7 +71,8 @@ { "name": "wants", "history": 20 }, { "name": "findings", "history": 5 }, { "name": "policies", "history": 20 }, - { "name": "seat-holders", "history": 5 } + { "name": "seat-holders", "history": 5 }, + { "name": "engine-reads", "history": 1 } ], "streams": [ {