diff --git a/internal/broker/modstream_work_queue_test.go b/internal/broker/modstream_work_queue_test.go new file mode 100644 index 00000000..ea7107b0 --- /dev/null +++ b/internal/broker/modstream_work_queue_test.go @@ -0,0 +1,38 @@ +package broker + +import "testing" + +// mailSendsStream is mesh-mail's future `mail-sends` stream (novox/hq ADR 0436 §7, §8), as the bus holds it: +// work-queue retention, a 24-hour duplicate window, kept one day and at most 64 MiB. +var mailSendsStream = ModuleStream{Module: "mesh-mail", Name: "MAIL_SENDS", + Events: []string{"mail.send.approved"}, MaxAgeDays: 1, MaxMiB: 64, WorkQueue: true, Duplicates: 24} + +// 1177/AC1: a work-queue stream is composed with work-queue retention, refusing a new message when full +// (DiscardNew), a 24-hour duplicate window, a one-day maximum age and a 64 MiB cap. +func TestAWorkQueueStreamIsComposedToRefuseWhenFull(t *testing.T) { + s := mailSendsStream.Stream() + if s.Retention != RetentionWorkQueue { + t.Fatalf("retention: %v, want %v", s.Retention, RetentionWorkQueue) + } + if !s.DiscardNew { + t.Fatal("a work-queue stream did not refuse a new message when full") + } + if s.DuplicatesSeconds != 24*60*60 { + t.Fatalf("duplicate window: %d seconds, want %d (24 hours)", s.DuplicatesSeconds, 24*60*60) + } + if s.MaxAge != 24*60*60 { + t.Fatalf("max age: %d seconds, want %d (1 day)", s.MaxAge, 24*60*60) + } + if s.MaxBytes != 64<<20 { + t.Fatalf("max bytes: %d, want %d (64 MiB)", s.MaxBytes, 64<<20) + } +} + +// 1177/AC3: a stream with no work-queue clause composes exactly as main does: limits retention, dropping its +// oldest, never refusing a new message. +func TestANonWorkQueueStreamStillDropsItsOldest(t *testing.T) { + s := mailStream.Stream() + if s.Retention != RetentionLimits || s.DiscardNew || s.DuplicatesSeconds != 0 { + t.Fatalf("composed as %+v; want limits retention, never discarding new, no duplicate window", s) + } +} diff --git a/internal/broker/modstreams.go b/internal/broker/modstreams.go index fc584b67..33e78271 100644 --- a/internal/broker/modstreams.go +++ b/internal/broker/modstreams.go @@ -49,6 +49,12 @@ type ModuleStream struct { MaxMiB int // Restricted says a trusted config value of the module grants who hears it (ADR 0363 §4). Restricted bool + // WorkQueue says this stream holds ADR 0436 §6's work-queue retention: a message is removed once a + // consumer acknowledges it, and the bus refuses a new one when full rather than dropping its oldest. + WorkQueue bool + // Duplicates is the work-queue's duplicate window in hours (novox/hq ADR 0436 §6); meaningless unless + // WorkQueue. + Duplicates int } // Subjects are the subjects the stream takes. @@ -70,7 +76,11 @@ func (s ModuleStream) Stream() Stream { if s.Restricted { why += "; heard only by the consumers its trusted config value grants" } - return Stream{ + if s.WorkQueue { + why += "; a work-queue (novox/hq ADR 0436 §6): a message is removed once its one consumer " + + "acknowledges it, and a full stream refuses a new one rather than dropping what it holds" + } + out := Stream{ Name: s.Name, Subjects: s.Subjects(), Retention: RetentionLimits, @@ -80,6 +90,12 @@ func (s ModuleStream) Stream() Stream { // Said on the bus, so a dead letter kept from it ages out with it (novox/hq ADR 0376 §3). Restricted: s.Restricted, } + if s.WorkQueue { + out.Retention = RetentionWorkQueue + out.DiscardNew = true + out.DuplicatesSeconds = s.Duplicates * 60 * 60 + } + return out } // ModuleStreams is every declared module stream as the controller asserts it, in name order. diff --git a/internal/catalogue/stream_work_queue_test.go b/internal/catalogue/stream_work_queue_test.go new file mode 100644 index 00000000..f37d6bfe --- /dev/null +++ b/internal/catalogue/stream_work_queue_test.go @@ -0,0 +1,66 @@ +package catalogue + +import ( + "strings" + "testing" +) + +// A module stream may declare retention: work-queue (novox/hq ADR 0436 §6): a message is removed once a +// consumer acknowledges it, the bus refusing a new one rather than dropping what it holds. The fixture is +// mesh-mail's future `mail-sends` stream (ADR 0436 §7, §8), copied from the module.json shape at +// /home/jochen/projects/novox/mesh-catalog/modules/mesh-mail/module.json and extended with retention, +// max-deliveries and duplicates as the ADR's clause 6 describes. +const mailSendsStream = `[{"name":"mail-sends","events":["mail.send.approved"],"max-age-days":1,"max-mib":64,` + + `"retention":"work-queue","max-deliveries":5,"duplicates":24}]` + +// 1177/AC1: a well-formed work-queue stream is accepted and read back whole. +func TestAWorkQueueStreamIsDeclaredAndAccepted(t *testing.T) { + m, err := ParseManifest([]byte(mailManifest(t, mailSendsStream, ""))) + if err != nil { + t.Fatalf("a well-formed work-queue stream was refused: %v", err) + } + s := m.Streams[0] + if s.Retention != StreamRetentionWorkQueue || s.MaxDeliveries != 5 || s.Duplicates != 24 || + s.Days() != 1 || s.Cap() != 64 { + t.Fatalf("the stream read as %+v", s) + } +} + +// 1177/AC2: a stream whose work-queue values are out of range, contradictory, or carries a grant is refused +// naming the stream and the rule. +func TestAWorkQueueStreamsValuesAreBounded(t *testing.T) { + for name, c := range map[string]struct{ streams, says string }{ + "max-deliveries 0": {`[{"name":"mail-sends","events":["mail.send.approved"],"retention":"work-queue",` + + `"max-deliveries":0,"duplicates":24}]`, "between 1 and 10"}, + "max-deliveries 11": {`[{"name":"mail-sends","events":["mail.send.approved"],"retention":"work-queue",` + + `"max-deliveries":11,"duplicates":24}]`, "between 1 and 10"}, + "duplicates longer than its maximum age": {`[{"name":"mail-sends","events":["mail.send.approved"],` + + `"max-age-days":1,"retention":"work-queue","max-deliveries":5,"duplicates":25}]`, "longer than its maximum age"}, + "an unknown retention": {`[{"name":"mail-sends","events":["mail.send.approved"],"retention":"sometimes"}]`, + "retention"}, + "max-deliveries without retention work-queue": {`[{"name":"mail-sends","events":["mail.send.approved"],` + + `"max-deliveries":5}]`, "without retention"}, + "duplicates without retention work-queue": {`[{"name":"mail-sends","events":["mail.send.approved"],` + + `"duplicates":24}]`, "without retention"}, + "a work-queue stream carrying a grant": {`[{"name":"mail-sends","events":["mail.send.approved"],` + + `"retention":"work-queue","max-deliveries":5,` + + `"grant":{"config":"grants","events":"mail-sends.{key}.**"}}]`, "carrying a grant"}, + } { + _, err := ParseManifest([]byte(mailManifest(t, c.streams, ""))) + if err == nil || !strings.Contains(err.Error(), "mail-sends") || !strings.Contains(err.Error(), c.says) { + t.Errorf("%s: %v; want a refusal naming the stream and %q", name, err, c.says) + } + } +} + +// 1177/AC3: a stream with no retention composes as main does — limits retention, dropping its oldest — proven +// at the broker, where the bus's retention policy is decided (internal/broker/modstreams_test.go). +func TestAStreamWithNoRetentionReadsAsLimits(t *testing.T) { + m, err := ParseManifest([]byte(mailManifest(t, mailStream, ""))) + if err != nil { + t.Fatal(err) + } + if got := m.Streams[0].Retention; got != "" { + t.Fatalf("a stream declaring no retention read %q", got) + } +} diff --git a/internal/catalogue/streams.go b/internal/catalogue/streams.go index 012178b6..62c1eb1a 100644 --- a/internal/catalogue/streams.go +++ b/internal/catalogue/streams.go @@ -30,6 +30,18 @@ const ( // GrantKey is the name in a grant's pattern that each key of the config value fills. const GrantKey = "{key}" +// StreamRetentionWorkQueue is a stream's declared retention (novox/hq ADR 0436 §6): a message is removed +// once a consumer acknowledges it, and the bus refuses a new one rather than dropping what it holds. Empty +// is the mesh's own limits retention — kept until it ages or the stream fills, dropping its oldest. +const StreamRetentionWorkQueue = "work-queue" + +// StreamLeastDeliveries and StreamMostDeliveries bound a work-queue stream's max-deliveries (novox/hq ADR +// 0436 §6): how many times a message is handed over before it is given up as a dead letter. +const ( + StreamLeastDeliveries = 1 + StreamMostDeliveries = 10 +) + // ReservedStreamPrefixes are the starts of the stream names the mesh derives, which no module's stream // may take: a seat's work queue, a log, a bucket, an object store, and the dead letters. var ReservedStreamPrefixes = []string{"SEAT_", "LOG_", "KV_", "OBJ_", "DEAD_"} @@ -48,6 +60,15 @@ type StreamDeclaration struct { MaxMiB int `json:"max-mib,omitempty"` // Grant restricts it, when said. Grant *StreamGrant `json:"grant,omitempty"` + // Retention is how the stream decides what to keep when it fills (novox/hq ADR 0436 §6): empty for the + // mesh's own limits retention (dropping its oldest), or StreamRetentionWorkQueue. + Retention string `json:"retention,omitempty"` + // MaxDeliveries bounds how many times a work-queue message is handed over, 1 to 10; refused without + // retention work-queue (novox/hq ADR 0436 §6). + MaxDeliveries int `json:"max-deliveries,omitempty"` + // Duplicates is a work-queue stream's duplicate window in hours, at most its maximum age; refused + // without retention work-queue (novox/hq ADR 0436 §6). + Duplicates int `json:"duplicates,omitempty"` } // StreamGrant says which trusted config value of the module grants a consumer a part of the stream, and @@ -67,7 +88,8 @@ func (s *StreamDeclaration) UnmarshalJSON(raw []byte) error { dec := json.NewDecoder(bytes.NewReader(raw)) dec.DisallowUnknownFields() if err := dec.Decode(&p); err != nil { - return fmt.Errorf("a stream is {name, events, max-age-days, max-mib, grant}: %w", typedUnknown(err)) + return fmt.Errorf("a stream is {name, events, max-age-days, max-mib, grant, retention, max-deliveries, "+ + "duplicates}: %w", typedUnknown(err)) } *s = StreamDeclaration(p) return nil @@ -163,6 +185,44 @@ func StreamProblems(m Manifest) []string { problems = append(problems, fmt.Sprintf("%s caps stream %q at %d MiB; a stream holds between %d and %d "+ "(novox/hq ADR 0363)", m.Module, s.Name, s.MaxMiB, StreamLeastMiB, StreamMostMiB)) } + // A stream may declare retention: work-queue (novox/hq ADR 0436 §6): a message removed once a + // consumer acknowledges it, the bus refusing a new one when full rather than dropping its oldest, with + // max-deliveries (1 to 10) and a duplicate window (duplicates, hours, at most its maximum age) -- both + // meaningless, and refused, without it. A work-queue stream is read by its one consumer, never granted. + switch s.Retention { + case "", StreamRetentionWorkQueue: + default: + problems = append(problems, fmt.Sprintf( + "%s's stream %q declares retention %q; a stream's retention is %q or nothing (novox/hq ADR 0436 §6)", + m.Module, s.Name, s.Retention, StreamRetentionWorkQueue)) + } + switch { + case s.Retention == StreamRetentionWorkQueue && + (s.MaxDeliveries < StreamLeastDeliveries || s.MaxDeliveries > StreamMostDeliveries): + problems = append(problems, fmt.Sprintf( + "%s's stream %q declares max-deliveries %d; a work-queue stream's max-deliveries is between %d "+ + "and %d (novox/hq ADR 0436 §6)", m.Module, s.Name, s.MaxDeliveries, StreamLeastDeliveries, + StreamMostDeliveries)) + case s.Retention != StreamRetentionWorkQueue && s.MaxDeliveries != 0: + problems = append(problems, fmt.Sprintf( + "%s's stream %q declares max-deliveries %d without retention %q (novox/hq ADR 0436 §6)", + m.Module, s.Name, s.MaxDeliveries, StreamRetentionWorkQueue)) + } + switch { + case s.Duplicates != 0 && s.Retention != StreamRetentionWorkQueue: + problems = append(problems, fmt.Sprintf( + "%s's stream %q declares duplicates %d hours without retention %q (novox/hq ADR 0436 §6)", + m.Module, s.Name, s.Duplicates, StreamRetentionWorkQueue)) + case s.Duplicates != 0 && s.Duplicates > s.Days()*24: + problems = append(problems, fmt.Sprintf( + "%s's stream %q keeps duplicates %d hours, longer than its maximum age of %d days (novox/hq "+ + "ADR 0436 §6)", m.Module, s.Name, s.Duplicates, s.Days())) + } + if s.Retention == StreamRetentionWorkQueue && s.Grant != nil { + problems = append(problems, fmt.Sprintf( + "%s's stream %q declares retention %q carrying a grant; a work-queue stream is read by its "+ + "module's one consumer, never granted (novox/hq ADR 0436 §6)", m.Module, s.Name, s.Retention)) + } if len(s.Events) == 0 { problems = append(problems, fmt.Sprintf("%s declares stream %q with no events", m.Module, s.Name)) } diff --git a/internal/inventory/busrecords.go b/internal/inventory/busrecords.go index e8b78364..380fa149 100644 --- a/internal/inventory/busrecords.go +++ b/internal/inventory/busrecords.go @@ -422,7 +422,8 @@ func streamsOf(m catalogue.Manifest) []broker.ModuleStream { events = append(events, broker.WirePattern(e)) } out = append(out, broker.ModuleStream{Module: m.Module, Name: s.BusName(), Events: events, - MaxAgeDays: s.Days(), MaxMiB: s.Cap(), Restricted: s.Grant != nil}) + MaxAgeDays: s.Days(), MaxMiB: s.Cap(), Restricted: s.Grant != nil, + WorkQueue: s.Retention == catalogue.StreamRetentionWorkQueue, Duplicates: s.Duplicates}) } return out } diff --git a/internal/inventory/streams_test.go b/internal/inventory/streams_test.go index dbddf979..b79df9bf 100644 --- a/internal/inventory/streams_test.go +++ b/internal/inventory/streams_test.go @@ -43,3 +43,25 @@ func TestWhatAModuleHearsIsComposedOntoTheStreamsSubjects(t *testing.T) { t.Fatalf("the provider's stream: %+v", got) } } + +// 1177/AC1: a catalogue stream declaring retention work-queue is carried into the broker's ModuleStream as +// WorkQueue, with its duplicate window. +func TestAWorkQueueStreamIsCarriedIntoTheBrokersModuleStream(t *testing.T) { + sender := catalogue.Manifest{Module: "mesh-mail", Streams: []catalogue.StreamDeclaration{{Name: "mail-sends", + Events: []string{"mail.send.approved"}, MaxAgeDays: 1, MaxMiB: 64, + Retention: catalogue.StreamRetentionWorkQueue, MaxDeliveries: 5, Duplicates: 24}}} + got := streamsOf(sender) + if len(got) != 1 || got[0].Name != "MAIL_SENDS" || !got[0].WorkQueue || got[0].Duplicates != 24 { + t.Fatalf("the sender's stream: %+v", got) + } +} + +// 1177/AC3: a catalogue stream with no retention is carried into the broker as not a work queue. +func TestANonWorkQueueStreamIsCarriedAsNotAWorkQueue(t *testing.T) { + plain := catalogue.Manifest{Module: "feed", Streams: []catalogue.StreamDeclaration{{Name: "items", + Events: []string{"item.**"}}}} + got := streamsOf(plain) + if len(got) != 1 || got[0].WorkQueue || got[0].Duplicates != 0 { + t.Fatalf("the plain stream: %+v", got) + } +}