From 4bcb6f86a42e04fab571678b833c0b49dad8abbb Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 20:54:27 +0100 Subject: [PATCH] R-3: W1.12: one stream name MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A seat event's stream has one bus name, seatdef.KeptStream's, which the generated contract carries as Stream: HearSeatEvents takes it as it stands and refuses a dotted or unprefixed name by name, so no part makes the name a second time (SEAT_SEAT_...). Builds: design 47 G ยง260 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- go/process/event_hub_integration_test.go | 7 +++-- go/process/event_hub_test.go | 13 +++++++-- go/process/seatstream.go | 37 +++++++++++++----------- 3 files changed, 34 insertions(+), 23 deletions(-) diff --git a/go/process/event_hub_integration_test.go b/go/process/event_hub_integration_test.go index 00ee817f..5463abe4 100644 --- a/go/process/event_hub_integration_test.go +++ b/go/process/event_hub_integration_test.go @@ -11,6 +11,7 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/seatdef" "git.novox.be/novox/mesh-sdk/go/testbus" ) @@ -80,8 +81,8 @@ func TestAModuleHearsASeatEventKeptOnItsOwnStream(t *testing.T) { } ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() - kept, subjects := "node-engine.running", []string{"mesh.seat.node-engine.event.running.>"} - if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: process.SeatStreamName(kept), Subjects: subjects, MaxMsgsPerSubject: 1}); err != nil { + kept, subjects := seatdef.KeptStream("node-engine", seatdef.Message{Name: "running"}), []string{"mesh.seat.node-engine.event.running.>"} + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: kept, Subjects: subjects, MaxMsgsPerSubject: 1}); err != nil { t.Fatal(err) } for _, m := range []struct{ subject, fp string }{ @@ -99,7 +100,7 @@ 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, process.SeatStreamName(kept), cred.SeatEventConsumerConfig(subjects, true)); err != nil { + if _, err := js.CreateOrUpdateConsumer(ctx, kept, cred.SeatEventConsumerConfig(subjects, true)); err != nil { t.Fatal(err) } cc, err := cred.HearSeatEvents(ctx, js, kept, hear, t.Logf) diff --git a/go/process/event_hub_test.go b/go/process/event_hub_test.go index 07acd8d8..16797254 100644 --- a/go/process/event_hub_test.go +++ b/go/process/event_hub_test.go @@ -1,6 +1,8 @@ package process import ( + "context" + "strings" "testing" "github.com/nats-io/nats.go/jetstream" @@ -36,10 +38,15 @@ func TestTheHubMatchesSubjectsAndRefusesOverlappingPatterns(t *testing.T) { t.Fatal(err) } // W1.12: a seat's event kept in a stream of its own is heard on one bus name and one durable, the maker's and the - // hearer's alike; an event kept last per subject is delivered from the last of each subject. + // hearer's alike; an event kept last per subject is delivered from the last of each subject. The name is the + // contract's Stream as it stands (seatdef.KeptStream), never made again here: a dotted or unprefixed name is + // refused by name before the bus is asked. a := Account{Node: "ace", Module: "mesh-grants"} - if got := SeatStreamName("node-engine.running"); got != "SEAT_node-engine_running" { - t.Errorf("the bus's name of node-engine.running is %q", got) + for _, stream := range []string{"node-engine.running", "SEAT_node-engine.running", "SEAT_", "EVENTS"} { + if _, err := a.HearSeatEvents(context.Background(), nil, stream, func(jetstream.Msg) {}, nil); err == nil || + !strings.Contains(err.Error(), "is not the bus's name of a seat event's stream") { + t.Errorf("hearing %q: %v", stream, err) + } } cfg := a.SeatEventConsumerConfig([]string{"mesh.seat.node-engine.event.running.>"}, true) if cfg.Durable != "ace_mesh-grants" || cfg.DeliverPolicy != jetstream.DeliverLastPerSubjectPolicy || diff --git a/go/process/seatstream.go b/go/process/seatstream.go index c1295dab..10836869 100644 --- a/go/process/seatstream.go +++ b/go/process/seatstream.go @@ -6,21 +6,21 @@ import ( "strings" "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/seatdef" ) // A seat's event kept in a stream of its own (seatdef's `keep`: the last message of each subject, or a stream with an age // and a cap), not on the shared events stream (novox/hq design 47: the node-engine seat's `running..`, the -// seat mesh-delivery's `owed`). A generated contract names that stream as `.` (Stream, seatdef.KeptStream); -// these are the one rule, for the part that makes the stream and its consumers (the grants part, from each module's -// `consumes`) and for every part that hears it, of the stream's name on the bus and the module's durable on it. +// seat mesh-delivery's `owed`). Its name on the bus is seatdef.KeptStream's, `SEAT__`, which a generated +// contract carries as Stream: one name in one place. These are the one rule, for the part that makes the stream's +// consumers (the grants part, from each module's `consumes`) and for every part that hears it, of the module's durable +// on it. -// SeatStreamPrefix starts the bus's name of every stream a seat's event is kept in apart from its family. -const SeatStreamPrefix = "SEAT_" - -// SeatStreamName is the bus's name of the stream a contract names kept (`.`): the bus takes no dot in a -// stream's name, so `SEAT__`, each dot an underscore (`node-engine.running` is `SEAT_node-engine_running`). -func SeatStreamName(kept string) string { - return SeatStreamPrefix + strings.ReplaceAll(kept, ".", "_") +// keptStreamName is whether stream is a name seatdef.KeptStream gives: the bus's, with its prefix and no dot. +func keptStreamName(stream string) bool { + return strings.HasPrefix(stream, seatdef.KeptStreamPrefix) && len(stream) > len(seatdef.KeptStreamPrefix) && + !strings.Contains(stream, ".") } // SeatEventConsumerConfig is the module's durable on a seat event's own stream as its maker makes it: _, as @@ -35,26 +35,29 @@ func (a Account) SeatEventConsumerConfig(subjects []string, lastPerSubject bool) return cfg } -// HearSeatEvents binds the module's durable on the stream of kept (a contract's Stream) and pulls it, handing each +// 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, kept string, handle func(jetstream.Msg), logf func(string, ...any)) (jetstream.ConsumeContext, error) { +func (a Account) HearSeatEvents(ctx context.Context, js jetstream.JetStream, stream string, handle func(jetstream.Msg), logf func(string, ...any)) (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) + } if handle == nil { - return nil, fmt.Errorf("hearing %s needs a handler", kept) + return nil, fmt.Errorf("hearing %s needs a handler", stream) } if logf == nil { logf = func(string, ...any) {} } - stream := SeatStreamName(kept) cons, err := a.EventConsumer(ctx, js, stream) if err != nil { - return nil, fmt.Errorf("the consumer %s on %s (the seat's stream %s) is not there: %w", a.EventConsumerName(), stream, kept, err) + return nil, fmt.Errorf("the consumer %s on %s is not there: %w", a.EventConsumerName(), stream, err) } cc, err := HearEvents(cons, handle, logf) if err != nil { - return nil, fmt.Errorf("the consumer %s on %s (the seat's stream %s) is not heard: %w", a.EventConsumerName(), stream, kept, err) + return nil, fmt.Errorf("the consumer %s on %s is not heard: %w", a.EventConsumerName(), stream, err) } - logf("events: the consumer %s on %s (the seat's stream %s) is heard", a.EventConsumerName(), stream, kept) + logf("events: the consumer %s on %s is heard", a.EventConsumerName(), stream) return cc, nil }