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) + } + } +}