From 136fa8b2a896d004edde7ae49d9ae3e9fc68a4a0 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 09:04:43 +0100 Subject: [PATCH] Say how a declared bucket and module stream are created, so the grants part makes the same stream the controller does BucketStreamConfig and ModuleStreamConfig are the stream configurations the controller's CreateOrUpdateKeyValue (BucketConfig) and StreamConfigFor assert at its start, pure, for the grants part's create of a bucket or stream a manifest declares and the bus lacks. A log already has Log.Config. Never used for an update. Issue: 1790 --- go/buscompose/create.go | 88 ++++++++++++++++++++++++++++++++++++ go/buscompose/create_test.go | 46 +++++++++++++++++++ 2 files changed, 134 insertions(+) create mode 100644 go/buscompose/create.go create mode 100644 go/buscompose/create_test.go diff --git a/go/buscompose/create.go b/go/buscompose/create.go new file mode 100644 index 0000000..17adb72 --- /dev/null +++ b/go/buscompose/create.go @@ -0,0 +1,88 @@ +package buscompose + +import ( + "time" + + "github.com/nats-io/nats.go/jetstream" +) + +// The streams a manifest declares, as the server is asked to create them (novox/hq ADR 0547 §5, issue 1790): the +// grants part creates each one the bus does not hold yet, with what the controller asserts at its start, so the two +// make the same stream. Pure. Only a create is ever asked with these: a stream the bus holds is never updated by the +// grants part (CreateStreamGrants grants no update). + +// BucketStreamConfig is a module's bucket as the stream the controller's CreateOrUpdateKeyValue makes of its options +// (mesh-controller internal/broker BucketConfig, with no marker age held): the client library's key-value shape +// (rollup, no delete, direct reads, discard new, a duplicate window of two minutes or the age when shorter) with the +// mesh's caps, file storage and one replica. +func BucketStreamConfig(b Bucket) jetstream.StreamConfig { + history := b.History + if history < 1 { + history = 1 + } + age := time.Duration(b.TTLSeconds) * time.Second + dupes := 2 * time.Minute + if age > 0 && age < dupes { + dupes = age + } + cfg := jetstream.StreamConfig{ + Name: "KV_" + b.Bucket(), + Description: b.Why(), + Subjects: []string{"$KV." + b.Bucket() + ".>"}, + MaxMsgsPerSubject: int64(history), + MaxBytes: StateMaxBytes, + MaxAge: age, + MaxMsgSize: StateMaxValueBytes, + Storage: jetstream.FileStorage, + Replicas: 1, + AllowRollup: true, + DenyDelete: true, + Duplicates: dupes, + MaxMsgs: -1, + MaxConsumers: -1, + AllowDirect: true, + Discard: jetstream.DiscardNew, + } + if b.KeyTTL { + cfg.AllowMsgTTL, cfg.SubjectDeleteMarkerTTL = true, KeyTTLMarkerTTL + } + return cfg +} + +// ModuleStreamConfig is a module's own stream as the controller asserts it (mesh-controller internal/broker +// StreamConfigFor of ModuleStream.Stream). +func ModuleStreamConfig(m ModuleStream) jetstream.StreamConfig { + s := m.Stream() + cfg := jetstream.StreamConfig{ + Name: s.Name, + Subjects: s.Subjects, + Retention: jetstream.LimitsPolicy, + MaxAge: time.Duration(s.MaxAge) * time.Second, + MaxMsgsPerSubject: int64(s.MaxMsgsPerSubject), + Description: s.Why, + AllowDirect: s.Direct, + AllowMsgTTL: s.AllowMsgTTL, + } + if s.Retention == RetentionWorkQueue { + cfg.Retention = jetstream.WorkQueuePolicy + } + if s.Restricted || s.WorkQueue { + cfg.Metadata = map[string]string{} + if s.Restricted { + cfg.Metadata[RestrictedMetadata] = "true" + } + if s.WorkQueue { + cfg.Metadata[WorkQueueMetadata] = "true" + } + } + if s.MaxBytes > 0 { + cfg.MaxBytes = s.MaxBytes + } + if s.DiscardNew { + cfg.Discard = jetstream.DiscardNew + } + if s.DuplicatesSeconds > 0 { + cfg.Duplicates = time.Duration(s.DuplicatesSeconds) * time.Second + } + return cfg +} diff --git a/go/buscompose/create_test.go b/go/buscompose/create_test.go new file mode 100644 index 0000000..68cd04b --- /dev/null +++ b/go/buscompose/create_test.go @@ -0,0 +1,46 @@ +package buscompose + +import ( + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" +) + +// Defends issue 1790 (ADR 0547 §5): the grants part creates a declared bucket and stream as the controller asserts them. +func TestDeclaredStreamsAreCreatedAsTheControllerAssertsThem(t *testing.T) { + plain := BucketStreamConfig(Bucket{Module: "mesh-vault", Name: "bus-accounts"}) + keyed := BucketStreamConfig(Bucket{Module: "m", Name: "presence", History: 5, TTLSeconds: 60, KeyTTL: true}) + queue := ModuleStreamConfig(ModuleStream{Module: "b", Name: "ASKS", Events: []string{"ask"}, MaxAgeDays: 7, MaxMiB: 32, + WorkQueue: true, Duplicates: 1}) + for _, row := range []struct { + what string + got, want any + }{ + {"bucket name", plain.Name, "KV_mesh-vault_bus-accounts"}, + {"bucket subjects", plain.Subjects[0], "$KV.mesh-vault_bus-accounts.>"}, + {"bucket history defaults to one", plain.MaxMsgsPerSubject, int64(1)}, + {"bucket cap", plain.MaxBytes, int64(StateMaxBytes)}, + {"bucket value cap", plain.MaxMsgSize, int32(StateMaxValueBytes)}, + {"bucket duplicate window", plain.Duplicates, 2 * time.Minute}, + {"bucket discards new", plain.Discard, jetstream.DiscardNew}, + {"bucket direct and rollup, no delete", plain.AllowDirect && plain.AllowRollup && plain.DenyDelete, true}, + {"bucket without key ttl", plain.AllowMsgTTL, false}, + {"history", keyed.MaxMsgsPerSubject, int64(5)}, + {"age", keyed.MaxAge, time.Minute}, + {"duplicate window under the age", keyed.Duplicates, time.Minute}, + {"key ttl markers", keyed.SubjectDeleteMarkerTTL, KeyTTLMarkerTTL}, + {"key ttl", keyed.AllowMsgTTL, true}, + {"queue retention", queue.Retention, jetstream.WorkQueuePolicy}, + {"queue subjects", queue.Subjects[0], "mesh.stream.b.event.ask"}, + {"queue age", queue.MaxAge, 7 * 24 * time.Hour}, + {"queue cap", queue.MaxBytes, int64(32 << 20)}, + {"queue discards new", queue.Discard, jetstream.DiscardNew}, + {"queue duplicates", queue.Duplicates, time.Hour}, + {"queue metadata", queue.Metadata[WorkQueueMetadata] + queue.Metadata[RestrictedMetadata], "truetrue"}, + } { + if row.got != row.want { + t.Errorf("%s: %v, want %v", row.what, row.got, row.want) + } + } +}