From 905265a56b5231c49cd9cdcca96ca22d63d3c7d8 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 17:20:30 +0100 Subject: [PATCH] J25 A-293: a placed module's failed newest build said as refused in by-entries mode, on the builder's build-ended event MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reading the catalog half by fingerprint, the node-engine no longer reads the builder's builds, so #224's refused key was never written there and a reader waited on a build that will never be done. The engine now hears the builder's build.ended stream, read whole at start then followed (as the assigner hears it); the last trunk end of a placed module that ended failed is written as the refused key with the event's words, a later done end deletes it, an end naming a branch moves nothing. The manifest consumes mesh-delivery-builder.build.ended.**. Builds: design 47 G A-293 (with §421, §628), ADR 0593 §6, issue 1966 U1/U8 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- cmd/mesh-host/nodeagent.go | 24 ++++++- cmd/mesh-host/recovery_integration_test.go | 36 ++++++++++ internal/link/running.go | 81 ++++++++++++++++++++++ internal/link/session_test.go | 27 ++++++++ internal/selfcompose/composedstate.go | 52 +++++++++++++- internal/selfcompose/composedstate_test.go | 45 ++++++++++++ internal/selfcompose/selfcompose.go | 50 +++++++++++++ module.json | 1 + packaging/operator_keys_test.go | 4 +- 9 files changed, 312 insertions(+), 8 deletions(-) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 09d85816..75310ea0 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -694,6 +694,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus composerNow.Store(composer) // what the verb compose asks a pass of, this run's alone session.SetManifestOf(composer.ManifestOf) // the event running names the manifest of the build that runs go followRunning(run, session, composer, say) + go followBuildEnds(run, session, composer, say) apply.UseEngineModule(composer.EngineModule) // the guard of the engine's own trees fixes no name (§527) env.Composed, env.SelfComposed, env.Policy, env.ComposedToday = composer, true, composer.Policy, composer.ComposedToday env.Recovery = recoveryEnv(session, composer) @@ -1204,11 +1205,28 @@ func composerEnv(session *link.Session, node string, say func(string), hooks com // followRunning keeps the composer told what every node runs, from the seat node-engine's event running, for the run it is // in: the stream or the link not being there is said and tried again after a backoff, never polled. func followRunning(run context.Context, session *link.Session, composer *selfcompose.Composer, say func(string)) { + keepFollowing(run, "this node does not hear what the nodes run, and tries again: ", func(ctx context.Context) error { + return session.FollowRunning(ctx, composer.NoteRunning, composer.RunningRead) + }, composer.RunningLost, say) +} + +// followBuildEnds keeps the composer told how the builder's builds ended, from its event build.ended, for the run it is in +// (design 47 G A-293): a placed module's newest build that ended failed is said as a refusal in the composition state. +func followBuildEnds(run context.Context, session *link.Session, composer *selfcompose.Composer, say func(string)) { + keepFollowing(run, "this node does not hear how the builder's builds end, so a failed build is not said as refused, and it tries again: ", + func(ctx context.Context) error { + return session.FollowBuildEnds(ctx, composer.NoteBuildEnded, composer.BuildEndsRead) + }, composer.BuildEndsLost, say) +} + +// keepFollowing follows a stream with follow for the run, lost called each time it ends; why it ended is said once until it +// changes, and it is followed again after a backoff, never polled. +func keepFollowing(run context.Context, line string, follow func(context.Context) error, lost func(), say func(string)) { said := "" for try := 1; run.Err() == nil; try++ { began := time.Now() - err := session.FollowRunning(run, composer.NoteRunning, composer.RunningRead) - composer.RunningLost() + err := follow(run) + lost() if run.Err() != nil { return } @@ -1216,7 +1234,7 @@ func followRunning(run context.Context, session *link.Session, composer *selfcom try = 1 } if why := fmt.Sprint(err); why != said { - say("this node does not hear what the nodes run, and tries again: " + why) + say(line + why) said = why } select { diff --git a/cmd/mesh-host/recovery_integration_test.go b/cmd/mesh-host/recovery_integration_test.go index 14718427..2f665064 100644 --- a/cmd/mesh-host/recovery_integration_test.go +++ b/cmd/mesh-host/recovery_integration_test.go @@ -11,6 +11,7 @@ import ( "time" "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" "github.com/novox/mesh-host/internal/link" @@ -131,6 +132,41 @@ func TestNatsTheRecoveryStepAsksHearsBoundsAndCatchesUpOverTheBus(t *testing.T) case <-time.After(10 * time.Second): t.Fatal("the bound's end, the bus's marker, was not heard") } + + // Design 47 G A-293: over the same session, the builder's build ends the composer follows: the stream read whole first (the + // last end of each module and fingerprint, in the stream's order), then the read's end, then each end said after it. + if _, err := b.js.AddStream(&nats.StreamConfig{Name: link.BuildEndsStream, Subjects: []string{link.BuildEndsSubject}, MaxMsgsPerSubject: 1}); err != nil { + t.Fatal(err) + } + ended := func(fp, outcome, why string) { + raw, _ := json.Marshal(builds.BuildEnded{Module: "db", Fingerprint: fp, Outcome: outcome, Error: why}) + if _, err := b.js.Publish("mesh.stream.mesh-delivery-builder.event."+builds.BuildEndedEventName("db", fp), raw); err != nil { + t.Fatal(err) + } + } + ended("f1", builds.BuildFailed, "first") + ended("f1", builds.BuildFailed, "no go") + ended("f0", builds.BuildDone, "") + heard := make(chan string, 8) + follow, stopFollow := context.WithCancel(ctx) + defer stopFollow() + go func() { + _ = session.FollowBuildEnds(follow, func(e builds.BuildEnded) { heard <- e.Fingerprint + ":" + e.Outcome + ":" + e.Error }, + func() { heard <- "read" }) + }() + for i, want := range []string{"f1:failed:no go", "f0:done:", "read", "f2:failed:later"} { + if want == "f2:failed:later" { + ended("f2", builds.BuildFailed, "later") + } + select { + case got := <-heard: + if got != want { + t.Fatalf("build end %d heard %q, want %q", i, got, want) + } + case <-time.After(10 * time.Second): + t.Fatalf("build end %d, %q, was not heard", i, want) + } + } } // startSession is a session on the node's own connection, with no node agent run on it. diff --git a/internal/link/running.go b/internal/link/running.go index fa24b0d1..c1d76213 100644 --- a/internal/link/running.go +++ b/internal/link/running.go @@ -6,6 +6,8 @@ import ( "fmt" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" + "git.novox.be/novox/mesh-sdk/go/catalogue" "github.com/nats-io/nats.go" ) @@ -105,3 +107,82 @@ func (b OverNATS) HearRunning(consumer string, heard func([]byte)) (func(), erro } var _ RunningBus = OverNATS{} + +// The builder's event build.ended (design 47 G A-293, as the assigner hears it, A-181, §556): the builder's stream of build +// ends keeps the last message of each module and fingerprint, and the node reads it whole at each start, then follows it, through +// one ordered consumer it makes for the run: what ended, and a failed build's words, are known from the stream however long ago +// they were said, never from the builder's builds. The engine's manifest consumes `mesh-delivery-builder.build.ended.**`. + +// BuildEndsStream is the builder's stream of build ends as the bus names it, BuildEndsSubject every event it carries. +var ( + BuildEndsStream = catalogue.StreamBusName(builds.BuildsStream) + BuildEndsSubject = "mesh.stream.mesh-delivery-builder.event." + builds.BuildEndedEvent + ".>" +) + +// BuildEndsBus is a link that hears the builder's build ends. +type BuildEndsBus interface { + // HearBuildEnds calls heard with the last message of every subject, then readEnded once they are all read, then heard with + // each after them, until ctx ends (nil) or the stream is lost (its error). + HearBuildEnds(ctx context.Context, heard func([]byte), readEnded func()) error +} + +func hearBuildEnds(ctx context.Context, js nats.JetStreamContext, heard func([]byte), readEnded func()) error { + sub, err := js.SubscribeSync(BuildEndsSubject, nats.OrderedConsumer(), nats.DeliverLastPerSubject(), nats.BindStream(BuildEndsStream)) + if err != nil { + return fmt.Errorf("the stream %s cannot be read: %w", BuildEndsStream, err) + } + defer sub.Unsubscribe() + info, err := sub.ConsumerInfo() + if err != nil { + return fmt.Errorf("the stream %s cannot be read: %w", BuildEndsStream, err) + } + total := info.Delivered.Consumer + info.NumPending + for got := uint64(0); ; got++ { + if got == total { + readEnded() + } + msg, err := sub.NextMsgWithContext(ctx) + if err != nil { + if ctx.Err() != nil { + return nil + } + return fmt.Errorf("the stream %s is not followed any more (%d read): %w", BuildEndsStream, got, err) + } + heard(msg.Data) + } +} + +func (l *natsLink) HearBuildEnds(ctx context.Context, heard func([]byte), readEnded func()) error { + return hearBuildEnds(ctx, l.js, heard, readEnded) +} + +// HearBuildEnds reads the stream as the link's session does, over its connection. +func (b OverNATS) HearBuildEnds(ctx context.Context, heard func([]byte), readEnded func()) error { + return hearBuildEnds(ctx, b.JS, heard, readEnded) +} + +var ( + _ BuildEndsBus = (*natsLink)(nil) + _ BuildEndsBus = OverNATS{} +) + +// FollowBuildEnds hands every build end the builder said to heard: the last of each module and fingerprint first, with +// readEnded called after them (until then the caller acts on none), then each after them, until ctx ends. A body that is no +// end of a module at a fingerprint is passed over. An error is the stream or the link not being there, which the caller tries +// again. +func (q *Session) FollowBuildEnds(ctx context.Context, heard func(builds.BuildEnded), readEnded func()) error { + q.init() + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + h, ok := bus.(BuildEndsBus) + if bus == nil || !ok { + return ErrNoLink + } + return h.HearBuildEnds(ctx, func(raw []byte) { + var e builds.BuildEnded + if json.Unmarshal(raw, &e) == nil && e.Module != "" && e.Fingerprint != "" { + heard(e) + } + }, readEnded) +} diff --git a/internal/link/session_test.go b/internal/link/session_test.go index 64783c1d..21234b77 100644 --- a/internal/link/session_test.go +++ b/internal/link/session_test.go @@ -12,6 +12,7 @@ import ( "time" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "github.com/nats-io/nats.go" ) @@ -45,6 +46,17 @@ func (l *runningLink) HearRunning(_ string, heard func([]byte)) (func(), error) return func() {}, nil } +func (l *runningLink) HearBuildEnds(_ context.Context, heard func([]byte), readEnded func()) error { + for _, raw := range l.first { + heard(raw) + } + readEnded() + for _, raw := range l.after { + heard(raw) + } + return nil +} + func aMember(t *testing.T) (Membership, ed25519.PrivateKey) { t.Helper() public, private, err := ed25519.GenerateKey(nil) @@ -84,6 +96,21 @@ func TestFactsWithNoLinkOpenAreNotSaid(t *testing.T) { if err != nil || strings.Join(order, " ") != "a/x b/y read" { t.Fatalf("followed %v: %v, want a/x b/y read", order, err) } + + // Design 47 G A-293: the builder's build ends likewise, the last of each subject first, then the read's end, then the + // rest, a body that is no end of a module at a fingerprint passed over. + if err := (&Session{Membership: m}).FollowBuildEnds(context.Background(), func(builds.BuildEnded) {}, func() {}); !errors.Is(err, ErrNoLink) { + t.Fatalf("build ends followed with no link: %v", err) + } + rl.first = [][]byte{[]byte(`{"module":"x","fingerprint":"f1","outcome":"failed","error":"no go"}`), []byte(`not json`), []byte(`{"module":"x"}`)} + rl.after = [][]byte{[]byte(`{"module":"y","fingerprint":"f2","outcome":"done","branch":"b"}`)} + order = nil + err = s.FollowBuildEnds(context.Background(), func(e builds.BuildEnded) { + order = append(order, e.Module+"@"+e.Fingerprint+":"+e.Outcome+":"+e.Error+":"+e.Branch) + }, func() { order = append(order, "read") }) + if err != nil || strings.Join(order, " ") != "x@f1:failed:no go: read y@f2:done::b" { + t.Fatalf("build ends followed %v: %v", order, err) + } } // A link opening says the node's facts, off the link's own loop. diff --git a/internal/selfcompose/composedstate.go b/internal/selfcompose/composedstate.go index 8496b967..da0eec31 100644 --- a/internal/selfcompose/composedstate.go +++ b/internal/selfcompose/composedstate.go @@ -91,8 +91,19 @@ func (c *Composer) publishComposed(ctx context.Context) { func (c *Composer) newestFPs(ctx context.Context, at *published) (out map[string]string, failed map[string][2]string) { if c.byEntries() { // The newest build is the indexer's (and the builder's done key says it is built): neither is read, and the node - // follows no branch (design 47 G §421, §628), so the state holds the wanted builds alone. - return maps.Clone(at.fps), nil + // follows no branch (design 47 G §421, §628), so the state holds the wanted builds alone. A placed module's newest + // build that ended is the last trunk end the builder said of it (NoteBuildEnded): a failed one is refused (A-293). + failed = map[string][2]string{} + c.mu.Lock() + if c.endsRead { + for _, module := range at.placed { + if e, ok := c.ends[module]; ok && e.Outcome == builds.BuildFailed { + failed[module] = [2]string{e.Fingerprint, e.Error} + } + } + } + c.mu.Unlock() + return maps.Clone(at.fps), failed } out = map[string]string{} failed = map[string][2]string{} @@ -153,7 +164,10 @@ func (c *Composer) writeComposed(ctx context.Context, at *published) (wrote, del wrote++ c.saidComposed(key, values[key]) } - for _, key := range c.pastKeeping() { + for _, key := range append(c.pastKeeping(), c.endedRefusals(values)...) { + if _, kept := c.pubKept[key]; !kept { + continue // past keeping and ended both + } if err := c.env.ComposedState.Delete(ctx, key); err != nil { return wrote, deleted, fmt.Errorf("%s cannot be deleted: %w", key, err) } @@ -271,6 +285,38 @@ func (c *Composer) composedValues(ctx context.Context, at *published) (map[strin return out, nil } +// endedRefusals is, of the keys the state holds, each refusal a failed end of a module wrote (newestFPs) once that module's +// last trunk end is done, at any fingerprint (design 47 G A-293): the state never says refused for a done build. A key this +// pass writes (values), or one a want names, is not one. Under pubWrite. +func (c *Composer) endedRefusals(values map[string][]byte) []string { + c.mu.Lock() + defer c.mu.Unlock() + if !c.byEntries() || !c.endsRead { + return nil + } + var out []string + for module, last := range c.ends { + if last.Outcome != builds.BuildDone { + continue + } + for fp := range c.endFPs[module] { + key, err := engcomp.Key(c.env.Node, module, fp) + if err != nil || values[key] != nil || c.wants[module].Fp == fp { + continue + } + kept, ok := c.pubKept[key] + if !ok { + continue + } + if comp, err := engcomp.Decode(kept.Value); err == nil && comp.Refused != "" { + out = append(out, key) + } + } + } + sort.Strings(out) + return out +} + // composedKeyOf is a composition's key in the state. func composedKeyOf(comp engcomp.Composed) (string, error) { if comp.Module == "" { diff --git a/internal/selfcompose/composedstate_test.go b/internal/selfcompose/composedstate_test.go index e87f0fc5..68f9721d 100644 --- a/internal/selfcompose/composedstate_test.go +++ b/internal/selfcompose/composedstate_test.go @@ -2,6 +2,8 @@ package selfcompose import ( "context" + "encoding/json" + "errors" "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" @@ -12,6 +14,7 @@ import ( "github.com/novox/mesh-host/internal/buckets" "github.com/novox/mesh-host/internal/buckets/bucketstest" engcomp "github.com/novox/mesh-host/internal/compositions" + "maps" "slices" "strings" "sync" @@ -258,4 +261,46 @@ func TestTheNodePublishesItsCompositions(t *testing.T) { if past := keep.pastKeeping(); len(past) != 0 { t.Errorf("past keeping with a want naming the oldest: %v, want none", past) } + + // Design 47 G A-293: reading the catalog half by fingerprint, a placed module's newest build that ended is the last trunk + // build.ended heard of it; a failed one is said refused with the event's words once the first read ended, an end naming a + // branch moves nothing, a module with no end (queued or running) stays unpublished, and a later done end, at any + // fingerprint, deletes the refused key. + byEnv := b.env() + byEnv.Entry = EntryCall(func(context.Context, string, string, any) (json.RawMessage, error) { + return nil, errors.New("nothing is asked") + }) + byState := &fakeComposed{kv: map[string]ComposedEntry{}} + byEnv.ComposedState = byState + by := New(byEnv) + byAt := &published{fps: map[string]string{}, placed: []string{"broken", "waiting", "branched"}} + words := `refused kind: package of language "go"` + by.NoteBuildEnded(builds.BuildEnded{Module: "broken", Fingerprint: bfp, Outcome: builds.BuildFailed, Error: words}) + by.NoteBuildEnded(builds.BuildEnded{Module: "branched", Fingerprint: wfp, Outcome: builds.BuildFailed, Error: "no", Branch: "feature"}) + for _, step := range []struct { + name string + do func() + holds map[string]string // by key, the refusal it says + }{ + {"before the first read ended", func() {}, map[string]string{}}, + {"once the first read ended", by.BuildEndsRead, map[string]string{brokenKey: words}}, + {"a later done end", func() { + by.NoteBuildEnded(builds.BuildEnded{Module: "broken", Fingerprint: wfp, Outcome: builds.BuildDone}) + }, map[string]string{}}, + } { + step.do() + if _, _, err := by.writeComposed(ctx, byAt); err != nil { + t.Fatalf("%s: %v", step.name, err) + } + byState.mu.Lock() + got := map[string]string{} + for k, v := range byState.kv { + comp, _ := engcomp.Decode(v.Value) + got[k] = comp.Refused + } + byState.mu.Unlock() + if !maps.Equal(got, step.holds) { + t.Errorf("%s: the state holds %v, want %v", step.name, got, step.holds) + } + } } diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 67f20f13..ea0d7af1 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -204,6 +204,12 @@ type Composer struct { // node's own consumer (NoteRunning), under mu; heardRead is whether the first read of the stream ended (RunningRead). heard map[string]map[string]heardRun heardRead bool + // ends is, by module, the last build end naming no branch the builder said of it (NoteBuildEnded), its newest build that + // ended, under mu; endsRead is whether the first read of the builder's stream ended (BuildEndsRead) (design 47 G A-293). + ends map[string]builds.BuildEnded + endsRead bool + // endFPs is, by module, every fingerprint a trunk end was heard of: the keys a failed end may have written. + endFPs map[string]map[string]bool // engine is the node-engine's own module as the last composing of the node's input found it (Scope.Engine), under mu. engine string // rehand asks for a pass that hands again what the agent waits for (Waiting). @@ -2443,6 +2449,50 @@ func (c *Composer) RunningLost() { c.mu.Unlock() } +// NoteBuildEnded takes the builder's event build.ended (design 47 G A-293): of a module, the last end naming no branch (the +// trunk's) stands as its newest build that ended, a failed one said as a refusal in the composition state, a done one ending +// that refusal; an end naming a branch moves nothing. Once the first read of the stream ended, an end of a module placed here +// asks a pass. +func (c *Composer) NoteBuildEnded(e builds.BuildEnded) { + if e.Module == "" || e.Fingerprint == "" || e.Branch != "" { + return + } + c.mu.Lock() + if c.ends == nil { + c.ends, c.endFPs = map[string]builds.BuildEnded{}, map[string]map[string]bool{} + } + if c.endFPs[e.Module] == nil { + c.endFPs[e.Module] = map[string]bool{} + } + c.endFPs[e.Module][e.Fingerprint] = true + if had, ok := c.ends[e.Module]; ok && had.Fingerprint == e.Fingerprint && had.Outcome == e.Outcome && had.Error == e.Error { + c.mu.Unlock() + return + } + c.ends[e.Module] = e + ask := c.endsRead + c.mu.Unlock() + if ask { + c.askRehand() + } +} + +// BuildEndsRead says the first read of the builder's stream ended: from now the composer acts on the ends it heard, and a +// pass is asked. +func (c *Composer) BuildEndsRead() { + c.mu.Lock() + c.endsRead = true + c.mu.Unlock() + c.askRehand() +} + +// BuildEndsLost says the builder's stream is not followed any more: what was heard is not acted on until a new read ends. +func (c *Composer) BuildEndsLost() { + c.mu.Lock() + c.endsRead = false + c.mu.Unlock() +} + func (c *Composer) askRehand() { select { case c.rehand <- struct{}{}: diff --git a/module.json b/module.json index 9e262cfe..d90fe2ba 100644 --- a/module.json +++ b/module.json @@ -56,6 +56,7 @@ "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-conditions.findings-asked", + "mesh-delivery-builder.build.ended.**", "mesh-delivery-view.settled", "mesh-recovery.ended", "mesh-recovery.restored", diff --git a/packaging/operator_keys_test.go b/packaging/operator_keys_test.go index 751d5436..72fd6773 100644 --- a/packaging/operator_keys_test.go +++ b/packaging/operator_keys_test.go @@ -88,8 +88,8 @@ func TestTheManifestConsumesTheBackupEvent(t *testing.T) { if m.Config["recovery-wait-minutes"].Default != float64(90) { t.Errorf("config recovery-wait-minutes is %+v, want default 90", m.Config["recovery-wait-minutes"]) } - // The backup event, the status view's settled, and the two events of the seat that serves the seat list (ADR 0636 §4), the keeper's ask (design 47 G §387) and the seat's running (§199). - if want := []string{"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-conditions.findings-asked", "mesh-delivery-view.settled", "mesh-recovery.ended", "mesh-recovery.restored", "node-engine.running", "restic.backed-up"}; !slices.Equal(m.Consumes, want) { + // The backup event, the status view's settled, and the two events of the seat that serves the seat list (ADR 0636 §4), the keeper's ask (design 47 G §387), the seat's running (§199) and the builder's build ends (A-293). + if want := []string{"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-conditions.findings-asked", "mesh-delivery-builder.build.ended.**", "mesh-delivery-view.settled", "mesh-recovery.ended", "mesh-recovery.restored", "node-engine.running", "restic.backed-up"}; !slices.Equal(m.Consumes, want) { t.Fatalf("consumes %v, want %v", m.Consumes, want) } // 774: the acts bucket mesh-host_acts is not declared as state for now: a declared state of a module with no account