diff --git a/internal/broker/consumer_from_now_test.go b/internal/broker/consumer_from_now_test.go index fa64d770..0b66e105 100644 --- a/internal/broker/consumer_from_now_test.go +++ b/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) + } +} diff --git a/internal/broker/derived.go b/internal/broker/derived.go index 65c97048..6aa1eaf1 100644 --- a/internal/broker/derived.go +++ b/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 diff --git a/internal/broker/derived_test.go b/internal/broker/derived_test.go index fc541705..68808cec 100644 --- a/internal/broker/derived_test.go +++ b/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") + } +} diff --git a/internal/broker/modstreams.go b/internal/broker/modstreams.go index 3dc94a30..76ad4dac 100644 --- a/internal/broker/modstreams.go +++ b/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", diff --git a/internal/broker/modstreams_live_test.go b/internal/broker/modstreams_live_test.go index 4532d913..877aba32 100644 --- a/internal/broker/modstreams_live_test.go +++ b/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) diff --git a/internal/broker/raise_live_test.go b/internal/broker/raise_live_test.go index fee13277..52c598ce 100644 --- a/internal/broker/raise_live_test.go +++ b/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 -// catch-up at all on this bus (novox/hq 04-ISSUES/050). -// -// On the bus the mesh runs on today a queue receives only what is published after it is bound, so -// everything built before the catalogue existed was announced to nobody — and on a fresh mesh that is -// 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) { +// **A module's consumer created after the fact starts from now** (novox/hq ADR 0443 §1, which changes what +// 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); +// an event only wakes it. Asked of a real server, because the whole rule rests on it. +func TestAModulesConsumerCreatedAfterwardsStartsFromNow(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)) { - t.Fatalf("a consumer created after %d builds has %d waiting for it — if this is 0 the mesh "+ - "does need a catch-up after all, and the reasoning for deleting it is wrong", - len(built), info.NumPending) + if info.NumPending != 0 { + t.Fatalf("a consumer created after %d builds has %d waiting for it; a module's consumer starts from now "+ + "(novox/hq ADR 0443 §1)", 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) } } diff --git a/internal/catalogue/consumes_start_test.go b/internal/catalogue/consumes_start_test.go new file mode 100644 index 00000000..33747922 --- /dev/null +++ b/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) + } +} diff --git a/internal/catalogue/manifest.go b/internal/catalogue/manifest.go index 85bbb067..cd48fdf3 100644 --- a/internal/catalogue/manifest.go +++ b/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) {