From 81d6d548aacbd948b18a9d3817a16922285e5c10 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 22:40:07 +0100 Subject: [PATCH] GR9 fix: findings-asked from the contract, the start read by HearSeatEvents, kept events known before the seat list is read MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The definitions of definedEventsOf name mesh-sdk's three finding events and the seat mesh-conditions's findings-asked (its generated contract's SubjectFindingsAsked). readRunningWhole is deleted: the part's durable on running is heard through process.Account.HearSeatEvents with process.OnRead. The bootstrapped definitions name the kept events before the seat list is read, so the part's first durable never filters running on EVENTS (a held consumer only gains filters). Builds: design 47 G §556, §641, §644, A-3, A-13, A-17, A-19 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- modules/mesh-grants/cmd/mesh-grants/main.go | 57 ++++--------------- modules/mesh-grants/internal/grants/inputs.go | 5 +- modules/mesh-grants/internal/grants/kept.go | 8 ++- 3 files changed, 21 insertions(+), 49 deletions(-) diff --git a/modules/mesh-grants/cmd/mesh-grants/main.go b/modules/mesh-grants/cmd/mesh-grants/main.go index 596087ce..b8eb059c 100644 --- a/modules/mesh-grants/cmd/mesh-grants/main.go +++ b/modules/mesh-grants/cmd/mesh-grants/main.go @@ -276,14 +276,9 @@ func hearRunning(ctx context.Context, account process.Account, js jetstream.JetS if hearingRunning.cc != nil { return nil } - // At each start the stream is read whole first, every subject's last message, through an ordered consumer made for that - // one read; the durable follows from its acknowledged position after it (design 47 G §556). Until it ends the part acts - // on none of running and grants_first_views says the stream waiting. - if err := readRunningWhole(ctx, js, f.Part, logf); err != nil { - return err - } - f.Part.RunningRead() - f.Wake() + // At each start the stream is read whole first (design 47 G §556, §641, A-19): HearSeatEvents reads a stream kept last per + // subject through an ordered consumer made for that one read, to the last of each, and follows the durable only after it. + // Until it ends the part acts on none of running and grants_first_views says the stream waiting (OnRead). cc, err := account.HearSeatEvents(ctx, js, nodeengine.StreamRunning, func(m jetstream.Msg) { changed, err := f.Part.HearRunning(m.Subject(), m.Data()) if err != nil { @@ -295,7 +290,15 @@ func hearRunning(ctx context.Context, account process.Account, js jetstream.JetS if changed { f.Wake() } - }, logf) + }, logf, process.OnRead(func(state string) { + switch state { + case process.ReadWaiting: + f.Part.ReadingRunning(nodeengine.StreamRunning) + case process.ReadDone: + f.Part.RunningRead() + f.Wake() + } + })) if err != nil { return err } @@ -323,39 +326,3 @@ func (g *seatGate) Definitions() map[string]seatdef.Definition { } return nil } - -// readRunningWhole reads every subject's last message of the node-engine's stream running through an ordered consumer, and -// keeps each in the part's state running (design 47 G §556). -func readRunningWhole(ctx context.Context, js jetstream.JetStream, p *grants.Part, logf func(string, ...any)) error { - stream, err := js.Stream(ctx, nodeengine.StreamRunning) - if err != nil { - return fmt.Errorf("the stream %s cannot be read whole: %w", nodeengine.StreamRunning, err) - } - cons, err := stream.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy}) - if err != nil { - return fmt.Errorf("the ordered consumer for the whole read of %s cannot be made: %w", nodeengine.StreamRunning, err) - } - read := 0 - for { - batch, err := cons.FetchNoWait(256) - if err != nil { - return fmt.Errorf("the whole read of %s failed: %w", nodeengine.StreamRunning, err) - } - n := 0 - for m := range batch.Messages() { - n++ - if _, err := p.HearRunning(m.Subject(), m.Data()); err != nil { - logf("running: %v, dropped", err) - } - } - if err := batch.Error(); err != nil { - return fmt.Errorf("the whole read of %s failed: %w", nodeengine.StreamRunning, err) - } - if n == 0 { - break - } - read += n - } - logf("running: the whole read ended, %d subjects kept; the durable follows from its place", read) - return nil -} diff --git a/modules/mesh-grants/internal/grants/inputs.go b/modules/mesh-grants/internal/grants/inputs.go index 9dce0414..7b78b215 100644 --- a/modules/mesh-grants/internal/grants/inputs.go +++ b/modules/mesh-grants/internal/grants/inputs.go @@ -357,7 +357,10 @@ func ReadInputs(r Reader, cache *Manifests) Inputs { Values: map[string]any{}, Gaps: append([]Gap(nil), standingGaps...)} in.SeatListUnread = gate != nil && !gate.MayAct() in.Defined = definedEventsOf(nil) - if d, ok := gate.(SeatDefinitions); ok && !in.SeatListUnread { + // The definitions the reader holds, the bootstrapped ones before the seat list is read (ADR 0636 §2), name the events kept + // on a stream of their own, so a durable made at the first pass already hears them there and never on EVENTS (a held + // consumer only gains filters, so one made with such an event's subject on EVENTS would keep it). + if d, ok := gate.(SeatDefinitions); ok { in.Kept = buscompose.KeptEventsOf(d.Definitions()) in.EventSubjects = eventSubjectsOf(d.Definitions()) in.Defined = definedEventsOf(d.Definitions()) diff --git a/modules/mesh-grants/internal/grants/kept.go b/modules/mesh-grants/internal/grants/kept.go index 9c0a03ad..42afb8bb 100644 --- a/modules/mesh-grants/internal/grants/kept.go +++ b/modules/mesh-grants/internal/grants/kept.go @@ -6,6 +6,7 @@ import ( "slices" "strings" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "git.novox.be/novox/mesh-sdk/go/seatdef" "git.novox.be/novox/mesh-sdk/go/views" @@ -106,8 +107,8 @@ func eventSubjectsOf(defs map[string]seatdef.Definition) map[string]map[string]s } // definedEventsOf is the events a `*` emitter before them may reach (catalogue.Hearing's defined, design 47 G §386, §474, -// §644): each event, by its local name, of a seat definition of the seat list, and mesh-sdk's four finding events, which -// views.FindingSubjects names and which are defined with or without the list read. +// §644): each event, by its local name, of a seat definition of the seat list, and mesh-sdk's three finding events (views.FindingSubjects) and the seat mesh-conditions's findings-asked (its generated +// contract's SubjectFindingsAsked), which are defined with or without the list read. func definedEventsOf(defs map[string]seatdef.Definition) map[string]bool { out := map[string]bool{} for _, d := range defs { @@ -116,9 +117,10 @@ func definedEventsOf(defs map[string]seatdef.Definition) map[string]bool { } } s := views.FindingSubjects("module", "node") - for _, subject := range []string{s.Written, s.Cleared, s.Stated, s.Asked} { + for _, subject := range []string{s.Written, s.Cleared, s.Stated} { out[strings.TrimSuffix(strings.TrimPrefix(subject, "mesh.stream.module.event."), ".node")] = true } + out[strings.TrimPrefix(meshconditions.SubjectFindingsAsked, "mesh.seat.mesh-conditions.event.")] = true return out }