From 2134806396c169acf5ac082aa29500793f689f95 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 06:19:55 +0100 Subject: [PATCH] A-233 (sdk): HearStream reads a stream kept by age whole from its first message with FromFirst At every start an ordered consumer, made for the one read, hands every message from the first, then the durable is followed, so a part that keeps what an age-kept seat event said (the assigner's version-released and version-delivered) knows it again after a restart. Builds: design 47 G A-233, A-189, A-181 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- go/process/event_hub_integration_test.go | 81 ++++++++++++++++++++++++ go/process/seatstream.go | 36 +++++++++-- 2 files changed, 110 insertions(+), 7 deletions(-) diff --git a/go/process/event_hub_integration_test.go b/go/process/event_hub_integration_test.go index 73e84576..5597cfea 100644 --- a/go/process/event_hub_integration_test.go +++ b/go/process/event_hub_integration_test.go @@ -4,6 +4,7 @@ package process_test import ( "context" + "strings" "testing" "time" @@ -192,9 +193,89 @@ func subAModuleHearsAnotherModulesStreamKeptLastPerSubject(t *testing.T) { } } +// A-233: a module hears a stream kept by age with FromFirst: at every start the stream is read whole from its first message, though +// its durable began after them, then the durable is followed; without FromFirst only the durable is. +func subAModuleHearsAStreamKeptByAgeFromItsFirstMessage(t *testing.T) { + nc, err := nats.Connect(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer nc.Close() + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + kept, subjects := seatdef.KeptStream("mesh-catalog", seatdef.Message{Name: "version-released"}), []string{"mesh.seat.mesh-catalog.event.version-released"} + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: kept, Subjects: subjects, MaxAge: 90 * 24 * time.Hour}); err != nil { + t.Fatal(err) + } + for _, v := range []string{"web 1.0.0", "web 1.1.0", "api 2.0.0"} { + if _, err := js.Publish(ctx, subjects[0], []byte(v)); err != nil { + t.Fatal(err) + } + } + cred := process.Account{Node: "ace", Module: "assigner"} + durable := cred.SeatEventConsumerConfig(subjects, false) + durable.DeliverPolicy = jetstream.DeliverNewPolicy + if _, err := js.CreateOrUpdateConsumer(ctx, kept, durable); err != nil { + t.Fatal(err) + } + got := make(chan string, 8) + hear := func(m jetstream.Msg) { got <- string(m.Data()); _ = m.Ack() } + var reads []string + cc, err := cred.HearSeatEvents(ctx, js, kept, hear, t.Logf, process.FromFirst(), process.OnRead(func(s string) { reads = append(reads, s) })) + if err != nil { + t.Fatal(err) + } + var heard []string + for len(heard) < 3 { + select { + case h := <-got: + heard = append(heard, h) + case <-ctx.Done(): + t.Fatalf("heard %v, want every message from the first", heard) + } + } + if strings.Join(heard, ",") != "web 1.0.0,web 1.1.0,api 2.0.0" || len(reads) != 2 { + t.Errorf("heard %v with reads %v, want every message in order, read whole", heard, reads) + } + if _, err := js.Publish(ctx, subjects[0], []byte("web 1.2.0")); err != nil { + t.Fatal(err) + } + select { + case h := <-got: + if h != "web 1.2.0" { + t.Errorf("the durable heard %s", h) + } + case <-ctx.Done(): + t.Fatal("the durable was not followed after the read") + } + cc.Stop() + // Without FromFirst, a stream kept by age is not read: the durable alone, past every message, hands the next one first. + cc, err = cred.HearSeatEvents(ctx, js, kept, hear, t.Logf) + if err != nil { + t.Fatal(err) + } + defer cc.Stop() + if _, err := js.Publish(ctx, subjects[0], []byte("next")); err != nil { + t.Fatal(err) + } + select { + case h := <-got: + if h != "next" { + t.Errorf("without FromFirst %s was handed before the next message", h) + } + case <-ctx.Done(): + t.Fatal("the durable was not followed") + } +} + // TestTheHubBindsALateConsumerAndSaysItIsLive runs the tests of this package that were merged to keep the module within its test budget (design 47 G §419), each as a subtest. func TestTheHubBindsALateConsumerAndSaysItIsLive(t *testing.T) { t.Run("TheHubBindsALateConsumerAndSaysItIsLive", subTheHubBindsALateConsumerAndSaysItIsLive) t.Run("AModuleHearsASeatEventKeptOnItsOwnStream", subAModuleHearsASeatEventKeptOnItsOwnStream) t.Run("AModuleHearsAnotherModulesStreamKeptLastPerSubject", subAModuleHearsAnotherModulesStreamKeptLastPerSubject) + t.Run("AModuleHearsAStreamKeptByAgeFromItsFirstMessage", subAModuleHearsAStreamKeptByAgeFromItsFirstMessage) } diff --git a/go/process/seatstream.go b/go/process/seatstream.go index 9dffa21f..47518d47 100644 --- a/go/process/seatstream.go +++ b/go/process/seatstream.go @@ -46,17 +46,26 @@ const ( // HearOption is an option of HearSeatEvents. type HearOption func(*hearOptions) -type hearOptions struct{ onRead func(state string) } +type hearOptions struct { + onRead func(state string) + fromFirst bool +} // 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 } } +// FromFirst is HearStream's form for a stream kept by age and cap, not last per subject (design 47 G A-233, A-189): at every +// start the stream is read whole from its first message through an ordered consumer made for that one read, on the subjects of +// the durable, and only then is the durable followed. A message the durable has not acknowledged yet is so handed twice, once +// by the read and once by the durable: the handler takes a message again as it took it. +func FromFirst() HearOption { return func(o *hearOptions) { o.fromFirst = true } } + // 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 +// subject (or, from DeliverAll, from the first message, FromFirst) 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}) +func readWhole(ctx context.Context, js jetstream.JetStream, stream string, subjects []string, handle func(jetstream.Msg), from jetstream.DeliverPolicy) error { + cons, err := js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{FilterSubjects: subjects, DeliverPolicy: from}) if err != nil { return err } @@ -100,9 +109,22 @@ func (a Account) HearSeatEvents(ctx context.Context, js jetstream.JetStream, str return a.HearStream(ctx, js, stream, handle, logf, opts...) } +// wholeRead is how a start read begins: from the last of each subject for a stream kept last per subject, from the first +// message for FromFirst's, and none for another stream. +func wholeRead(maxPerSubject int64, fromFirst bool) (jetstream.DeliverPolicy, bool) { + switch { + case maxPerSubject == 1: + return jetstream.DeliverLastPerSubjectPolicy, true + case fromFirst: + return jetstream.DeliverAllPolicy, true + } + return 0, false +} + // HearStream is HearSeatEvents for the stream of a module's own events (a manifest's `streams` entry, named as // catalogue.StreamBusName names it), which the module is granted through its `consumes`: the same one rule, a stream the bus's own -// info says is kept last per subject (`last-per-subject`) read whole first, then the durable followed (design 47 G §556, §703, A-181). +// info says is kept last per subject (`last-per-subject`) read whole first, then the durable followed (design 47 G §556, §703, A-181); +// with FromFirst, a stream kept by age is read whole from its first message the same way (A-233). func (a Account) HearStream(ctx context.Context, js jetstream.JetStream, stream string, handle func(jetstream.Msg), logf func(string, ...any), opts ...HearOption) (jetstream.ConsumeContext, error) { if handle == nil { return nil, fmt.Errorf("hearing %s needs a handler", stream) @@ -122,7 +144,7 @@ func (a Account) HearStream(ctx context.Context, js jetstream.JetStream, stream if err != nil { return nil, fmt.Errorf("the stream %s is not read: %w", stream, err) } - if st.CachedInfo().Config.MaxMsgsPerSubject == 1 { + if from, whole := wholeRead(st.CachedInfo().Config.MaxMsgsPerSubject, o.fromFirst); whole { if o.onRead != nil { o.onRead(ReadWaiting) } @@ -131,7 +153,7 @@ func (a Account) HearStream(ctx context.Context, js jetstream.JetStream, stream if cfg.FilterSubject != "" { subjects = append(subjects, cfg.FilterSubject) } - if err := readWhole(ctx, js, stream, subjects, handle); err != nil { + if err := readWhole(ctx, js, stream, subjects, handle, from); err != nil { return nil, fmt.Errorf("the start read of %s is not made: %w", stream, err) } if o.onRead != nil {