GR3: a kept seat event's stream is made from the definition's keep object: its age (0 for none), its cap, 90 days and 16 MiB where unnamed, dropping its oldest at the cap

Builds: design 47 G §396, §331, §260
This commit is contained in:
2026-11-11 14:41:42 +01:00
parent 4a61381ff1
commit 9efc402a06
4 changed files with 26 additions and 19 deletions
@@ -13,14 +13,20 @@ import (
// 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.KeepLastPerSubject}, {Name: "composed"}}},
{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.KeepStream, MaxAgeDays: 7, MaxMiB: 64}}},
{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)
running, owed := kept[1].StreamConfig(), kept[0].StreamConfig()
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]))
keyed := BucketStreamConfig(Bucket{Module: "m", Name: "presence", History: 5, TTLSeconds: 60, KeyTTL: true})
@@ -51,13 +57,15 @@ func declaredStreamsAreCreatedAsTheControllerAssertsThem(t *testing.T) {
{"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 two", len(kept), 2},
{"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"},
+12 -13
View File
@@ -27,8 +27,9 @@ type KeptEvent struct {
Stream string
// Filter is the subject that carries every instance of the event, a part of its subject `*`.
Filter string
// Keep is seatdef.KeepLastPerSubject or seatdef.KeepStream; MaxAgeDays and MaxMiB bound the latter.
Keep string
// Last says the stream keeps the last message of each subject; MaxAgeDays (0 for no age limit) and MaxMiB bound it
// as the event's keep says, its defaults 90 days and 16 MiB (design 47 G §396, §331).
Last bool
MaxAgeDays int
MaxMiB int
Why string
@@ -39,11 +40,11 @@ func KeptEventsOf(defs map[string]seatdef.Definition) []KeptEvent {
var out []KeptEvent
for _, d := range defs {
for _, e := range d.Events {
if e.Keep == "" {
if e.Keep == nil {
continue
}
out = append(out, KeptEvent{Seat: d.Seat, Event: e.Name, Stream: seatdef.KeptStream(d.Seat, e),
Filter: seatdef.EventSubject(d.Seat, e.Local()), Keep: e.Keep, MaxAgeDays: e.MaxAgeDays, MaxMiB: e.MaxMiB,
Filter: seatdef.EventSubject(d.Seat, e.Local()), Last: e.KeptLastPerSubject(), MaxAgeDays: e.KeptAgeDays(), MaxMiB: e.KeptMiB(),
Why: "the seat " + d.Seat + "'s event " + e.Name + ", kept apart from the shared events stream (design 47): " + keptHow(e)})
}
}
@@ -52,18 +53,18 @@ func KeptEventsOf(defs map[string]seatdef.Definition) []KeptEvent {
}
func keptHow(e seatdef.Message) string {
if e.Keep == seatdef.KeepLastPerSubject {
if e.KeptLastPerSubject() {
return "the last message of each subject, no age limit"
}
return "every message within its age and size"
}
// LastPerSubject says a hearing consumer starts from the last message of each subject, never from now.
func (k KeptEvent) LastPerSubject() bool { return k.Keep == seatdef.KeepLastPerSubject }
func (k KeptEvent) LastPerSubject() bool { return k.Last }
// StreamConfig is the kept stream as it is created: sourced from the shared events stream by the event's subject, with
// its own limits. Last per subject holds one message a subject and no age; the other form its age and size, a full stream
// refusing rather than dropping what it holds.
// its own limits as the event's keep says (design 47 G §396): its age (none for 0), its cap, and one message a subject
// when it keeps the last of each. A stream at its cap drops its oldest message (§331).
func (k KeptEvent) StreamConfig() jetstream.StreamConfig {
cfg := jetstream.StreamConfig{
Name: k.Stream,
@@ -71,14 +72,12 @@ func (k KeptEvent) StreamConfig() jetstream.StreamConfig {
Storage: jetstream.FileStorage,
Replicas: 1,
Sources: []*jetstream.StreamSource{{Name: EventsStream, FilterSubject: k.Filter}},
MaxAge: time.Duration(k.MaxAgeDays) * 24 * time.Hour,
MaxBytes: int64(k.MaxMiB) << 20,
}
if k.LastPerSubject() {
if k.Last {
cfg.MaxMsgsPerSubject = 1
return cfg
}
cfg.MaxAge = time.Duration(k.MaxAgeDays) * 24 * time.Hour
cfg.MaxBytes = int64(k.MaxMiB) << 20
cfg.Discard = jetstream.DiscardNew
return cfg
}
File diff suppressed because one or more lines are too long
+1 -1
View File
@@ -68,7 +68,7 @@ func createKept(c Creator, done *creations, kept []buscompose.KeptEvent, held ma
continue
}
cfg := k.StreamConfig()
digest := fmt.Sprintf("%v", cfg.Sources[0].FilterSubject) + k.Keep + fmt.Sprint(k.MaxAgeDays, k.MaxMiB)
digest := fmt.Sprintf("%v", cfg.Sources[0].FilterSubject) + fmt.Sprint(k.Last, k.MaxAgeDays, k.MaxMiB)
if !force && done.madeAt(k.Stream) == digest {
notMade = append(notMade, k.Stream+": made at this definition and gone since; grants_assert makes it again")
continue