Let a module stream declare work-queue retention (issue 1177)

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

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.

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

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

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
 }

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