jschoubben
7c461a96ae
G's grants already carry a state's max-mib onto its bucket (Bucket.MaxBytes, §469); main's second copy of that cap in BucketStreamConfig and Composed is dropped for G's, and main's tests (the declared cap row, the held bucket raised) are kept. The checker keeps G's version 0.1 (§368) with main's compositions max-mib 256. Builds: design 47 G §469, §368; A-277 (since mesh-delivery #206 the grants part carries max-mib) Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
85 lines
4.9 KiB
Go
85 lines
4.9 KiB
Go
package buscompose
|
|
|
|
import (
|
|
"fmt"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
|
)
|
|
|
|
// Defends issue 1790 (ADR 0547 §5): the grants part creates a declared bucket and stream as the controller asserts them.
|
|
func declaredStreamsAreCreatedAsTheControllerAssertsThem(t *testing.T) {
|
|
plain := BucketStreamConfig(Bucket{Module: "mesh-vault", Name: "bus-accounts"})
|
|
zero, seven, sixtyFour := 0, 7, 64
|
|
defs := map[string]seatdef.Definition{
|
|
"node-engine": {Seat: "node-engine", Events: []seatdef.Message{
|
|
{Name: "running", Subject: "running.<node>.<module>", Keep: &seatdef.Keep{Object: true, MaxAgeDays: &zero, LastPerSubject: true}}, {Name: "composed"}}},
|
|
"mesh-delivery": {Seat: "mesh-delivery", Events: []seatdef.Message{
|
|
{Name: "owed", Subject: "owed.<repository>.<number>", Keep: &seatdef.Keep{Object: true, MaxAgeDays: &seven, MaxMiB: &sixtyFour}}}},
|
|
"mesh-catalog": {Seat: "mesh-catalog", Events: []seatdef.Message{{Name: "registration-changed", Keep: &seatdef.Keep{Object: true}}}},
|
|
}
|
|
kept := KeptEventsOf(defs)
|
|
byStream := map[string]jetstream.StreamConfig{}
|
|
for _, k := range kept {
|
|
byStream[k.Stream] = k.StreamConfig()
|
|
}
|
|
running, owed, named := byStream["SEAT_node-engine_running"], byStream["SEAT_mesh-delivery_owed"], byStream["SEAT_mesh-catalog_registration-changed"]
|
|
rest, heard := SplitKept(kept, []string{"mesh-catalog.manifest-accepted", "node-engine.running.*.*", "node-engine.running"})
|
|
heardConsumerCfg := ConsumerConfig(heardConsumer(Principal{Kind: KindModule, Node: "novox", Module: "mesh-grants"}, heard[0]))
|
|
declared := BucketStreamConfig(Bucket{Module: "c", Name: "compositions", MaxMiB: 256})
|
|
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 cap as its state declares it (max-mib)", declared.MaxBytes, int64(256 << 20)},
|
|
{"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"},
|
|
{"kept events are three", len(kept), 3},
|
|
{"kept stream name", running.Name, "SEAT_node-engine_running"},
|
|
{"a kept stream takes no subject of the shared stream", len(running.Subjects), 0},
|
|
{"a kept stream sources the shared stream by the event's subject", running.Sources[0].Name + " " + running.Sources[0].FilterSubject,
|
|
"EVENTS mesh.seat.node-engine.event.running.*.*"},
|
|
{"last per subject holds one message a subject", running.MaxMsgsPerSubject, int64(1)},
|
|
{"last per subject has no age", running.MaxAge, time.Duration(0)},
|
|
{"a keep naming nothing is 90 days within 16 MiB, dropping its oldest", fmt.Sprint(named.MaxAge, named.MaxBytes, named.Discard == jetstream.DiscardOld),
|
|
fmt.Sprint(90*24*time.Hour, int64(16<<20), true)},
|
|
{"the stream kept for a week", owed.MaxAge, 7 * 24 * time.Hour},
|
|
{"the stream kept within its size", owed.MaxBytes, int64(64 << 20)},
|
|
{"its name is the bus's", owed.Name, "SEAT_mesh-delivery_owed"},
|
|
{"a kept consume leaves the shared stream's", len(rest), 1},
|
|
{"both namings of one event are one durable on its stream", heard[0].Stream + " " + fmt.Sprint(heard[0].Filters),
|
|
"SEAT_node-engine_running [mesh.seat.node-engine.event.running.*.*]"},
|
|
{"its durable starts at the last message of each subject", heardConsumerCfg.DeliverPolicy, jetstream.DeliverLastPerSubjectPolicy},
|
|
{"its durable is the module's on the node", heardConsumerCfg.Durable, "novox_mesh-grants"},
|
|
} {
|
|
if row.got != row.want {
|
|
t.Errorf("%s: %v, want %v", row.what, row.got, row.want)
|
|
}
|
|
}
|
|
}
|