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

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