From 141b34d29db9e792092ada4986e893a2b411202c Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 21:52:41 +0100 Subject: [PATCH] =?UTF-8?q?G=20=C2=A7641:=20HearSeatEvents=20reads=20a=20s?= =?UTF-8?q?tream=20kept=20last=20per=20subject=20whole=20at=20its=20start?= =?UTF-8?q?=20through=20an=20ordered=20consumer,=20then=20follows=20the=20?= =?UTF-8?q?durable,=20and=20says=20waiting=20then=20read?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §641, §556, A-19 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- go/process/event_hub_integration_test.go | 24 ++++++- go/process/seatstream.go | 80 +++++++++++++++++++++++- 2 files changed, 101 insertions(+), 3 deletions(-) diff --git a/go/process/event_hub_integration_test.go b/go/process/event_hub_integration_test.go index 5463abe4..f3bcd924 100644 --- a/go/process/event_hub_integration_test.go +++ b/go/process/event_hub_integration_test.go @@ -100,10 +100,15 @@ func TestAModuleHearsASeatEventKeptOnItsOwnStream(t *testing.T) { if _, err := cred.HearSeatEvents(ctx, js, kept, hear, t.Logf); err == nil { t.Fatal("no consumer was made, and hearing said nothing") } - if _, err := js.CreateOrUpdateConsumer(ctx, kept, cred.SeatEventConsumerConfig(subjects, true)); err != nil { + // The durable has gone past every message (it starts from now), so only the start read, an ordered consumer on the stream + // kept last per subject, gives the last of each subject (design 47 G §641, A-19). + durable := cred.SeatEventConsumerConfig(subjects, true) + durable.DeliverPolicy = jetstream.DeliverNewPolicy + if _, err := js.CreateOrUpdateConsumer(ctx, kept, durable); err != nil { t.Fatal(err) } - cc, err := cred.HearSeatEvents(ctx, js, kept, hear, t.Logf) + var reads []string + cc, err := cred.HearSeatEvents(ctx, js, kept, hear, t.Logf, process.OnRead(func(s string) { reads = append(reads, s) })) if err != nil { t.Fatal(err) } @@ -120,4 +125,19 @@ func TestAModuleHearsASeatEventKeptOnItsOwnStream(t *testing.T) { if !heard["mesh.seat.node-engine.event.running.ace.mesh-tools=fp2"] || !heard["mesh.seat.node-engine.event.running.g14.mesh-tools=fp3"] { t.Errorf("heard %v, want the last of each subject", heard) } + if len(reads) != 2 || reads[0] != process.ReadWaiting || reads[1] != process.ReadDone { + t.Errorf("the read said %v, want waiting then read", reads) + } + // then the durable is followed + if _, err := js.Publish(ctx, "mesh.seat.node-engine.event.running.ace.mesh-tools", []byte("fp4")); err != nil { + t.Fatal(err) + } + select { + case h := <-got: + if h != "mesh.seat.node-engine.event.running.ace.mesh-tools=fp4" { + t.Errorf("the durable heard %s", h) + } + case <-ctx.Done(): + t.Fatal("the durable was not followed after the read") + } } diff --git a/go/process/seatstream.go b/go/process/seatstream.go index 10836869..7da0ebc8 100644 --- a/go/process/seatstream.go +++ b/go/process/seatstream.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "strings" + "sync" "github.com/nats-io/nats.go/jetstream" @@ -35,11 +36,63 @@ func (a Account) SeatEventConsumerConfig(subjects []string, lastPerSubject bool) return cfg } +// What HearSeatEvents says of its start read through OnRead (design 47 G §641, A-19): `waiting` when it begins the read of +// a stream kept last per subject, `read` when the read has ended and the durable is followed. +const ( + ReadWaiting = "waiting" + ReadDone = "read" +) + +// HearOption is an option of HearSeatEvents. +type HearOption func(*hearOptions) + +type hearOptions struct{ onRead func(state string) } + +// OnRead is the callback HearSeatEvents calls with ReadWaiting and then ReadDone around its start read; the module wires it +// to its first views (design 47 G A-19), so this package imports nothing of views. +func OnRead(f func(state string)) HearOption { return func(o *hearOptions) { o.onRead = f } } + +// readWhole hands handle every message an ordered consumer, made on stream for the one read, delivers from the last of each +// subject filtered to subjects, until none is left pending (design 47 G §556): it ends when the message it delivered has +// none after it, or at once when the consumer, made, holds none. +func readWhole(ctx context.Context, js jetstream.JetStream, stream string, subjects []string, handle func(jetstream.Msg)) error { + cons, err := js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{FilterSubjects: subjects, DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy}) + if err != nil { + return err + } + last := make(chan struct{}) + var once sync.Once + cc, err := cons.Consume(func(m jetstream.Msg) { + handle(m) + if md, err := m.Metadata(); err == nil && md.NumPending == 0 { + once.Do(func() { close(last) }) + } + }) + if err != nil { + return err + } + defer cc.Stop() + if info := cons.CachedInfo(); info != nil && info.NumPending == 0 && info.Delivered.Consumer == 0 { + return nil + } + select { + case <-last: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + // HearSeatEvents binds the module's durable on the stream (a contract's Stream, as it stands) and pulls it, handing each // message to handle, which acknowledges it; every error the bus reports while hearing is said through logf. Like // EventHub it makes neither the stream nor the consumer: one the bus does not hold yet is this call's error, naming // both, and a later call binds it. The caller stops the returned context. -func (a Account) HearSeatEvents(ctx context.Context, js jetstream.JetStream, stream string, handle func(jetstream.Msg), logf func(string, ...any)) (jetstream.ConsumeContext, error) { +// +// A stream the bus's own info says is kept last per subject is first read whole (design 47 G §641, §556, A-19): through an +// ordered consumer made for that one read, on the subjects of the durable, to the last of each, its messages going to the +// same handle; only then is the durable followed, so until the read ends the part hears none of it. OnRead says `waiting` +// before the read and `read` after it. +func (a Account) HearSeatEvents(ctx context.Context, js jetstream.JetStream, stream string, handle func(jetstream.Msg), logf func(string, ...any), opts ...HearOption) (jetstream.ConsumeContext, error) { if !keptStreamName(stream) { return nil, fmt.Errorf("%q is not the bus's name of a seat event's stream: a contract's Stream is %s_, "+ "with no dot (seatdef.KeptStream)", stream, seatdef.KeptStreamPrefix) @@ -50,10 +103,35 @@ func (a Account) HearSeatEvents(ctx context.Context, js jetstream.JetStream, str if logf == nil { logf = func(string, ...any) {} } + var o hearOptions + for _, f := range opts { + f(&o) + } cons, err := a.EventConsumer(ctx, js, stream) if err != nil { return nil, fmt.Errorf("the consumer %s on %s is not there: %w", a.EventConsumerName(), stream, err) } + st, err := js.Stream(ctx, stream) + if err != nil { + return nil, fmt.Errorf("the stream %s is not read: %w", stream, err) + } + if st.CachedInfo().Config.MaxMsgsPerSubject == 1 { + if o.onRead != nil { + o.onRead(ReadWaiting) + } + cfg := cons.CachedInfo().Config + subjects := cfg.FilterSubjects + if cfg.FilterSubject != "" { + subjects = append(subjects, cfg.FilterSubject) + } + if err := readWhole(ctx, js, stream, subjects, handle); err != nil { + return nil, fmt.Errorf("the start read of %s is not made: %w", stream, err) + } + if o.onRead != nil { + o.onRead(ReadDone) + } + logf("events: %s is read whole at its start", stream) + } cc, err := HearEvents(cons, handle, logf) if err != nil { return nil, fmt.Errorf("the consumer %s on %s is not heard: %w", a.EventConsumerName(), stream, err)