internal/broker/consumer_from_now_test.go

@@ -88,3 +88,133 @@ func TestAConsumerMadeFromNowNeverReplaysTheStreamsHistory(t *testing.T) {
 		t.Fatal("a work queue's consumer was reset")
 	}
 }
+
+// pendingOf polls a consumer's pending count until it is want or a second has passed, and returns what it saw last.
+func pendingOf(js *JetStream, stream, name string, want uint64) uint64 {
+	var got uint64
+	for deadline := time.Now().Add(time.Second); ; time.Sleep(20 * time.Millisecond) {
+		if info, err := js.js.ConsumerInfo(stream, name); err == nil {
+			got = info.NumPending
+		}
+		if got == want || time.Now().After(deadline) {
+			return got
+		}
+	}
+}
+
+// 1279/AC2 (ADR 0443 §1): on a bus whose EVENTS holds earlier events, the consumer the controller makes for a module
+// delivers none of them and delivers one published after its making; the same on another module's stream.
+func TestAModulesConsumerDeliversNothingOfTheStreamsPast(t *testing.T) { // 1279/AC2
+	js := aLiveBus(t)
+	if err := AssertMeshStreams(js); err != nil {
+		t.Fatal(err)
+	}
+	if err := js.Context().PurgeStream(EventsStream); err != nil {
+		t.Fatal(err)
+	}
+	const subject = "mesh.mod.shop.event.order.placed"
+	for i := 0; i < 3; i++ {
+		if _, err := js.Context().Publish(subject, []byte(fmt.Sprint(i))); err != nil {
+			t.Fatal(err)
+		}
+	}
+	c, ok := ConsumerFor(Principal{Kind: KindModule, Node: "one", Module: "audit", Consumes: []string{"shop.order.placed"}, PasswordHash: "x"})
+	if !ok {
+		t.Fatal("no consumer derived")
+	}
+	t.Cleanup(func() { _ = js.Context().DeleteConsumer(c.Stream, c.Name) })
+	if err := js.EnsureConsumer(c); err != nil {
+		t.Fatal(err)
+	}
+	if got := pendingOf(js, c.Stream, c.Name, 0); got != 0 {
+		t.Fatalf("a module's consumer made after 3 events holds %d of them", got)
+	}
+	if _, err := js.Context().Publish(subject, []byte("after")); err != nil {
+		t.Fatal(err)
+	}
+	if got := pendingOf(js, c.Stream, c.Name, 1); got != 1 {
+		t.Fatalf("the event published after its making is not pending: %d", got)
+	}
+
+	// Another module's stream.
+	s := mailStream.Stream()
+	if err := js.EnsureStream(s); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = js.js.DeleteStream(s.Name) })
+	for _, m := range []string{"added", "changed"} {
+		if _, err := js.js.Publish(ModuleStreamSubject("mesh-mail", "mail.mesh.message."+m), []byte("{}")); err != nil {
+			t.Fatal(err)
+		}
+	}
+	reader := Principal{Kind: KindModule, Node: "ace", Module: "reader", Hears: []Heard{mailHeard}}
+	h := heardConsumer(reader, mailHeard)
+	t.Cleanup(func() { _ = js.js.DeleteConsumer(h.Stream, h.Name) })
+	if err := js.EnsureConsumer(h); err != nil {
+		t.Fatal(err)
+	}
+	if got := pendingOf(js, h.Stream, h.Name, 0); got != 0 {
+		t.Fatalf("a module's consumer on another module's stream holds %d of its past", got)
+	}
+	if _, err := js.js.Publish(ModuleStreamSubject("mesh-mail", "mail.mesh.message.added"), []byte("{}")); err != nil {
+		t.Fatal(err)
+	}
+	if got := pendingOf(js, h.Stream, h.Name, 1); got != 1 {
+		t.Fatalf("the event published after its making is not pending on the module's stream: %d", got)
+	}
+}
+
+// 1279/AC3 (ADR 0443 §2): a consumer that exists, made from the start, asserted again as the controller now derives it,
+// keeps its deliver policy and delivers its next message from where it was.
+func TestAnExistingConsumerKeepsItsPositionWhenAssertedAgain(t *testing.T) { // 1279/AC3
+	js := aLiveBus(t)
+	if err := AssertMeshStreams(js); err != nil {
+		t.Fatal(err)
+	}
+	if err := js.Context().PurgeStream(EventsStream); err != nil {
+		t.Fatal(err)
+	}
+	const subject = "mesh.mod.shop.event.order.placed"
+	for i := 0; i < 3; i++ {
+		if _, err := js.Context().Publish(subject, []byte(fmt.Sprint(i))); err != nil {
+			t.Fatal(err)
+		}
+	}
+	c, _ := ConsumerFor(Principal{Kind: KindModule, Node: "one", Module: "audit", Consumes: []string{"shop.order.placed"}, PasswordHash: "x"})
+	t.Cleanup(func() { _ = js.Context().DeleteConsumer(c.Stream, c.Name) })
+	old := c
+	old.FromNow = false // as the controller made it before ADR 0443
+	if err := js.EnsureConsumer(old); err != nil {
+		t.Fatal(err)
+	}
+	next := func() string {
+		t.Helper()
+		sub, err := js.Context().PullSubscribe("", c.Name, nats.Bind(c.Stream, c.Name))
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer func() { _ = sub.Unsubscribe() }()
+		msgs, err := sub.Fetch(1, nats.MaxWait(2*time.Second))
+		if err != nil || len(msgs) != 1 {
+			t.Fatalf("fetch: %v (%d messages)", err, len(msgs))
+		}
+		_ = msgs[0].Ack()
+		return string(msgs[0].Data)
+	}
+	if got := next(); got != "0" {
+		t.Fatalf("the first message is %q", got)
+	}
+	if err := js.EnsureConsumer(c); err != nil { // now derived from now
+		t.Fatal(err)
+	}
+	info, err := js.Context().ConsumerInfo(c.Stream, c.Name)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if info.Config.DeliverPolicy != nats.DeliverAllPolicy {
+		t.Fatalf("asserting moved the deliver policy to %v", info.Config.DeliverPolicy)
+	}
+	if got := next(); got != "1" {
+		t.Fatalf("after asserting again the next message is %q, want 1: the position moved", got)
+	}
+}

internal/broker/derived.go

@@ -171,6 +171,10 @@ func ConsumerFor(p Principal) (Consumer, bool) {
 		Filters:        filters,
 		AckWaitSeconds: 30,
 		MaxDeliver:     5,
+		// From now, as the controller's own consumers have been since issue 248: a module made on a stream that
+		// keeps a week replays none of it. What a part needs of the present it reads from its owner
+		// (novox/hq ADR 0443 §1, §3). A consumer that exists keeps its place whatever this says.
+		FromNow: true,
 		Why: "what " + p.Module + " declared it consumes; after max-deliver it gives an event up, and the " +
 			"controller keeps it in DEAD_LETTERS until a person delivers it again or drops it",
 	}, true

internal/broker/derived_test.go

@@ -180,3 +180,35 @@ func TestAHoldersWorkerIsPulledByEveryHolder(t *testing.T) {
 		t.Fatalf("the worker is %s on %s; one per seat, shared by its holders", c.Name, c.Stream)
 	}
 }
+
+// 1279/AC1 (ADR 0443 §1): every module principal's durable consumer on a stream that keeps its history is made from
+// now, the operator-channel holder's consumer of condition events among them; a seat's worker and a module's own work
+// queue are not, because work queued before its holder came is work to do.
+func TestAModulesConsumerIsMadeFromNow(t *testing.T) { // 1279/AC1
+	for name, p := range map[string]Principal{
+		"consumes": {Kind: KindModule, Node: "one", Module: "audit", Consumes: []string{"shop.order.placed"}, PasswordHash: "x"},
+		"watches": {Kind: KindModule, Node: "one", Module: "mesh-catalog", PasswordHash: "x",
+			Watches: []Seat{{Name: "mesh-build-machine", Emits: []string{"built"}}}},
+		// The operator-channel holder (issue 549): a module that holds a seat and consumes the condition events.
+		"a seat holder consuming conditions": {Kind: KindModule, Node: "one", Module: "messenger", PasswordHash: "x",
+			Consumes: []string{"mesh-controller.condition.raised", "mesh-controller.condition.cleared"}},
+	} {
+		c, ok := ConsumerFor(p)
+		if !ok {
+			t.Fatalf("%s: no consumer derived", name)
+		}
+		if !c.FromNow {
+			t.Errorf("%s: the consumer %s on %s is made from the start of the stream", name, c.Name, c.Stream)
+		}
+	}
+	if c := heardConsumer(Principal{Kind: KindModule, Node: "ace", Module: "reader", Hears: []Heard{mailHeard}}, mailHeard); !c.FromNow {
+		t.Errorf("a module's consumer on another module's stream is made from the start of %s", c.Stream)
+	}
+	if c, _ := HolderConsumerFor("one", "telegram", telegramSeat()); c.FromNow {
+		t.Error("a seat's worker is made from now: work queued before its holder came would be lost")
+	}
+	q := QueueConsumer(Principal{Kind: KindModule, Node: "novox", Module: "mesh-mail"}, QueueOwner{Process: "send", Stream: "MAIL_SENDS", MaxDeliver: 5})
+	if q.FromNow {
+		t.Error("a module's own work queue is made from now: queued sends would be lost")
+	}
+}

internal/broker/modstreams.go

@@ -223,6 +223,8 @@ func heardConsumer(p Principal, h Heard) Consumer {
 		Filters:        filters,
 		AckWaitSeconds: 30,
 		MaxDeliver:     5,
+		// From now, like a module's consumer on EVENTS (novox/hq ADR 0443 §1).
+		FromNow: true,
 		Why: "what " + p.Module + " consumes of " + h.Emitter + "'s stream " + h.Stream + " and is granted " +
 			"(novox/hq ADR 0363); after max-deliver it gives an event up, and the controller keeps it in " +
 			"DEAD_LETTERS until a person delivers it again or drops it",

internal/broker/modstreams_live_test.go

@@ -42,6 +42,16 @@ func TestAModuleStreamOnARealBus(t *testing.T) {
 	if err := js.EnsureConsumer(c); err != nil {
 		t.Fatal(err)
 	}
+	// Made from now (novox/hq ADR 0443 §1): the two mails published above are its past, not its news.
+	time.Sleep(50 * time.Millisecond)
+	if info, _ := js.js.ConsumerInfo("MAIL", c.Name); info.NumPending != 0 {
+		t.Fatalf("the reader made after two mails has %d of them pending, want 0", info.NumPending)
+	}
+	for _, account := range []string{"mesh", "work"} {
+		if _, err := js.js.Publish(ModuleStreamSubject("mesh-mail", "mail."+account+".message.added"), []byte("{}")); err != nil {
+			t.Fatalf("publishing %s's mail: %v", account, err)
+		}
+	}
 	time.Sleep(50 * time.Millisecond)
 	if info, _ := js.js.ConsumerInfo("MAIL", c.Name); info.NumPending != 1 {
 		t.Fatalf("the reader granted `mesh` has %d pending, want 1", info.NumPending)

internal/broker/raise_live_test.go

@@ -143,18 +143,11 @@ func TestRaisingAMeshRolesWorkQueue(t *testing.T) {
 	}
 }
 
-// **A consumer created after the fact still sees what came before it**, which is why the mesh needs no
+// **A module's consumer created after the fact starts from now** (novox/hq ADR 0443 §1, which changes what
-// catch-up at all on this bus (novox/hq 04-ISSUES/050).
+// 04-ISSUES/050 relied on): the builds announced before the catalogue's consumer existed are not delivered to it
-//
+// as if they had just happened. What the catalogue needs of the present it reads from its owner (ADR 0443 §3);
-// On the bus the mesh runs on today a queue receives only what is published after it is bound, so
+// an event only wakes it. Asked of a real server, because the whole rule rests on it.
-// everything built before the catalogue existed was announced to nobody — and on a fresh mesh that is
+func TestAModulesConsumerCreatedAfterwardsStartsFromNow(t *testing.T) {
-// always the foundation, because those are the things the catalogue needed in order to exist. A whole
-// mechanism was built for it: the catalogue asks, the controller re-publishes.
-//
-// A stream is a log and a consumer is a position in it. A consumer created later starts at the
-// beginning by default, so the builds are simply there. Asked of a real server rather than assumed,
-// because the whole decision about whether to keep that mechanism rests on it.
-func TestAConsumerCreatedAfterwardsStillSeesWhatCameBefore(t *testing.T) {
 	js := aLiveBus(t)
 	if err := AssertMeshStreams(js); err != nil {
 		t.Fatal(err)
@@ -187,9 +180,14 @@ func TestAConsumerCreatedAfterwardsStillSeesWhatCameBefore(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if info.NumPending != uint64(len(built)) {
+	if info.NumPending != 0 {
-		t.Fatalf("a consumer created after %d builds has %d waiting for it — if this is 0 the mesh "+
+		t.Fatalf("a consumer created after %d builds has %d waiting for it; a module's consumer starts from now "+
-			"does need a catch-up after all, and the reasoning for deleting it is wrong",
+			"(novox/hq ADR 0443 §1)", len(built), info.NumPending)
-			len(built), info.NumPending)
+	}
+	if _, err := js.Context().Publish("mesh.seat.mesh-build-machine.event.built", []byte(`{"module":"after"}`)); err != nil {
+		t.Fatal(err)
+	}
+	if got := pendingOf(js, c.Stream, c.Name, 1); got != 1 {
+		t.Fatalf("a build announced after the consumer was made is not waiting for it: %d", got)
 	}
 }

internal/catalogue/consumes_start_test.go

@@ -0,0 +1,24 @@
+package catalogue
+
+import (
+	"strings"
+	"testing"
+)
+
+// 1279/AC4 (ADR 0443 §4): no module is given a way to ask where its consumer starts. A consumes entry is an event's
+// name; one that carries a key asking for the stream's history is refused by name, so the safe start is the only one.
+func TestAConsumesEntryAskingWhereItStartsIsRefused(t *testing.T) { // 1279/AC4
+	for _, key := range []string{"from", "start", "deliver"} {
+		raw := `{"module":"audit","version":"1","consumes":[{"event":"shop.order.placed","` + key + `":"beginning"}]}`
+		_, err := ParseManifest([]byte(raw))
+		if err == nil {
+			t.Fatalf("a consumes entry with %q was accepted", key)
+		}
+		if !strings.Contains(err.Error(), `"`+key+`"`) || !strings.Contains(err.Error(), "consumes") {
+			t.Errorf("the refusal for %q does not name the key and the field: %v", key, err)
+		}
+	}
+	if _, err := ParseManifest([]byte(`{"module":"audit","version":"1","consumes":["shop.order.placed"]}`)); err != nil {
+		t.Errorf("a plain consumes entry was refused: %v", err)
+	}
+}

internal/catalogue/manifest.go

@@ -1466,6 +1466,9 @@ func (m *Manifest) UnmarshalJSON(raw []byte) error {
 		// Beside `settings` it is refused by name (MixedFormProblems), never merged into it.
 		delete(keys, ConfigField)
 	}
+	if err := consumesEntryProblem(keys["consumes"]); err != nil {
+		return err
+	}
 	plain := map[string]string{}
 	many := map[string]map[string]string{}
 	if secrets, ok := keys["secrets"]; ok && string(secrets) != "null" {
@@ -1611,6 +1614,32 @@ func (m *Manifest) UnmarshalJSON(raw []byte) error {
 	return nil
 }
 
+// consumesEntryProblem refuses a consumes entry that is more than an event's name. **No manifest word asks where a
+// consumer starts** (novox/hq ADR 0443 §4): the controller makes every module's consumer from now, and the first module
+// that needs a stream's history brings a decision that adds the word. An entry carrying a key such as from, start or
+// deliver would otherwise be read as a bare "cannot unmarshal object", which says nothing of why.
+func consumesEntryProblem(raw json.RawMessage) error {
+	var entries []json.RawMessage
+	if len(raw) == 0 || json.Unmarshal(raw, &entries) != nil {
+		return nil // not an array: the ordinary decoding says so
+	}
+	for i, e := range entries {
+		var obj map[string]json.RawMessage
+		if json.Unmarshal(e, &obj) != nil {
+			continue
+		}
+		keys := make([]string, 0, len(obj))
+		for k := range obj {
+			keys = append(keys, strconv.Quote(k))
+		}
+		sort.Strings(keys)
+		return fmt.Errorf("consumes entry %d carries the key %s: an entry is an event's name alone, and no word asks where a "+
+			"consumer starts, because the controller makes every module's consumer from now (novox/hq ADR 0443 §4)",
+			i, strings.Join(keys, ", "))
+	}
+	return nil
+}
+
 // MarshalJSON writes `secrets` and `contributes` back in the shape they were read: single values,
 // and objects of local names; and its config in the form it was read in (config.go).
 func (m Manifest) MarshalJSON() ([]byte, error) {