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 }