diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index db89bb59..345cb0f2 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -74,8 +74,8 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15 t.Fatal(err) } } - for name, subject := range map[string]string{"TAKE_IN": "mesh.stream.mesh-delivery-indexer.event.take-in.>", - "INDEXER_EVENTS": "mesh.mod.mesh-delivery-indexer.event.>"} { + // The take-in events are the indexer's ordinary events (ADR 0633 §5), on the path manifest-accepted takes. + for name, subject := range map[string]string{"INDEXER_EVENTS": "mesh.mod.mesh-delivery-indexer.event.>"} { if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: name, Subjects: []string{subject}}); err != nil { t.Fatal(err) } @@ -164,18 +164,34 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15 return ack.Sequence } m1Key := takein.PendingKey(merged(m1)) - takeIns, err := js.Stream(ctx, "TAKE_IN") + events, err := js.Stream(ctx, "INDEXER_EVENTS") if err != nil { t.Fatal(err) } + // saidAll is every take-in event of kind said under key so far, oldest first. + saidAll := func(kind, key string) []takein.TakeIn { + event, _ := takein.TakeInEvent(kind) + info, err := events.Info(ctx) + if err != nil { + t.Fatal(err) + } + var out []takein.TakeIn + for seq := info.State.FirstSeq; seq <= info.State.LastSeq && seq > 0; seq++ { + msg, err := events.GetMsg(ctx, seq) + if err != nil || msg.Subject != "mesh.mod."+ix+".event."+event { + continue + } + if ti, err := takein.DecodeTakeIn(msg.Data); err == nil && ti.Key == key { + out = append(out, ti) + } + } + return out + } // said is the last take-in event under kind and key, within 20 s. said := func(kind, key string) (takein.TakeIn, bool) { - event, _ := takein.TakeInEvent(kind, key) for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { - if msg, err := takeIns.GetLastMsgForSubject(ctx, process.ModuleStreamSubject(ix, event)); err == nil { - if ti, err := takein.DecodeTakeIn(msg.Data); err == nil { - return ti, true - } + if all := saidAll(kind, key); len(all) > 0 { + return all[len(all)-1], true } } return takein.TakeIn{}, false @@ -258,7 +274,7 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15 } } if len(w) != 2 || w[0].Key != m1Key || w[0].WaitsFor != named || !slices.Contains(w[0].Holds, "a") || !slices.Contains(w[0].Holds, ix) || - w[1].Key != groupKey || w[1].What != takein.TakeInGroup || len(w[1].Merges) != 1 || !strings.HasSuffix(w[1].Said, "pending."+groupKey) { + w[1].Key != groupKey || w[1].What != takein.TakeInGroup || len(w[1].Merges) != 1 || w[1].Said != takein.TakeInPendingEvent { t.Fatalf("indexer_pending = %+v, want the revision %s waiting for %s holding a and the indexer, then the group gathered", w, m1Key, named) } @@ -309,8 +325,7 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15 t.Errorf("after the group's end: modules.c %v, modules.k %v, indexer_pending %+v: want both taken in and nothing waiting", hasC, hasK, w) } // One take-in: neither member was said as a revision of its own. - single, _ := takein.TakeInEvent(takein.TakeInPending, g1Key) - if _, err := takeIns.GetLastMsgForSubject(ctx, process.ModuleStreamSubject(ix, single)); err == nil { + if len(saidAll(takein.TakeInPending, g1Key)) != 0 { t.Errorf("the group's first member %s was said pending as a revision of its own", g1Key) } } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go index 3526751a..3d887748 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go @@ -64,11 +64,11 @@ func sayAccepted(js jetstream.JetStream, cred process.Account, logf func(string, } } -// sayTakeIn publishes a take-in event on the indexer's own stream take-in (novox/hq ADR 0633 §5): -// `mesh.stream..event.take-in..`. A publish that fails is the caller's to say. +// sayTakeIn says a take-in event (novox/hq ADR 0633 §5) as the indexer's ordinary event, on the path its other events +// take: mesh.mod..event.take-in., the wait's key in the body. A publish that fails is the caller's to say. func sayTakeIn(js jetstream.JetStream, cred process.Account) func(context.Context, takein.TakeIn) error { return func(ctx context.Context, t takein.TakeIn) error { - event, err := takein.TakeInEvent(t.Kind, t.Key) + event, err := takein.TakeInEvent(t.Kind) if err != nil { return err } @@ -76,7 +76,7 @@ func sayTakeIn(js jetstream.JetStream, cred process.Account) func(context.Contex if err != nil { return err } - msg := nats.NewMsg(process.ModuleStreamSubject(cred.Module, event)) + msg := nats.NewMsg(cred.EventSubject(event)) msg.Header.Set("x-source", cred.Module) msg.Header.Set("x-node", cred.Node) msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano)) diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 3eca3a36..b3a15633 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -815,7 +815,7 @@ func waitOf(entries []PendingEntry, what, behind string) takein.PendingWait { } } if e.Value.Said != "" && e.Value.SaidAt.After(pw.SaidAt) { - pw.Said, pw.SaidAt = takein.TakeInStream+"."+e.Value.SaidKind+"."+e.Value.Said, e.Value.SaidAt + pw.Said, pw.SaidAt = "take-in."+e.Value.SaidKind, e.Value.SaidAt } } slices.Sort(pw.Holds) diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index 966f599f..5a5396ce 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -385,7 +385,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) } } if len(answer.Waits) != 1 || answer.Waits[0].Key != takein.PendingKey(2) || !slices.Equal(answer.Waits[0].Holds, []string{"avahi", delivery.IndexerModule}) || - answer.Waits[0].Said != "take-in.pending."+takein.PendingKey(2) || len(answer.Waits[0].Keys) != 2 { + answer.Waits[0].Said != takein.TakeInPendingEvent || len(answer.Waits[0].Keys) != 2 { t.Fatalf("indexer_pending = %+v", answer) } b.said = nil diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index ace5d0ed..5ce577c3 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -104,8 +104,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // 1126: placement is the seat mesh-placement's (ADR 0567): the indexer holds no placement state. ADR 0578 §1: it says // a refused manifest as manifest-refused, and (issue 1966 D3, 1966/AC10) a newly accepted one asking for bus state as // manifest-accepted. - if !reflect.DeepEqual(m.Emits, []string{"manifest-refused", "manifest-accepted"}) { - t.Errorf("emits = %v, want manifest-refused and manifest-accepted", m.Emits) + if !reflect.DeepEqual(m.Emits, []string{"manifest-refused", "manifest-accepted", takein.TakeInPendingEvent, takein.TakeInTakenEvent}) { + t.Errorf("emits = %v, want manifest-refused, manifest-accepted and the two take-in events", m.Emits) } // 1126 note 76: heard keeps the last 64 FORGE messages the indexer handled, for indexer_heard. if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: delivery.PackagesState, History: delivery.PackagesHistory}, @@ -146,10 +146,10 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build !slices.Equal(m.Claims[0].Serves, []string{"modules", "module", "provides", "dependents"}) { t.Errorf("claims = %+v, want mesh-catalog with its four verbs", m.Claims) } - // 1585/T11 (ADR 0633 §1, §5): its one stream, take-in, an old streams entry (the indexer is foundation, ADR 0595 §11), - // carries the take-in events. - if want := []streamDecl{{Name: takein.TakeInStream, Events: []string{"take-in.**"}, MaxAgeDays: 7}}; !reflect.DeepEqual(m.Streams, want) { - t.Errorf("streams = %+v, want %+v", m.Streams, want) + // 1585/T11 (ADR 0633 §1, §5): the take-in events are its ordinary events, with no stream of its own: a stream is + // outside the bus's predefined forms, and the manifest validator refuses it. + if len(m.Streams) != 0 || !slices.Contains(m.Emits, takein.TakeInPendingEvent) || !slices.Contains(m.Emits, takein.TakeInTakenEvent) { + t.Errorf("streams = %+v, emits = %v: want no stream, and both take-in events emitted", m.Streams, m.Emits) } // No other bus section, no other seat (ADR 0437 §4). for _, k := range []string{"logs", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} { diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index fafd9bcc..10bb8227 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -44,12 +44,11 @@ { "name": "graph", "history": 5 }, { "name": "pending", "history": 1 } ], - "streams": [ - { "name": "take-in", "events": ["take-in.**"], "max-age-days": 7 } - ], "emits": [ "manifest-refused", - "manifest-accepted" + "manifest-accepted", + "take-in.pending", + "take-in.taken" ], "reads": [ "mesh-build-go.toolchains", diff --git a/modules/mesh-delivery-indexer/takein/takein.go b/modules/mesh-delivery-indexer/takein/takein.go index dc53626c..31fbfb8e 100644 --- a/modules/mesh-delivery-indexer/takein/takein.go +++ b/modules/mesh-delivery-indexer/takein/takein.go @@ -49,14 +49,17 @@ func decode(data []byte, into any) error { return nil } -// The take-in events (novox/hq ADR 0633 §1, §5): the indexer publishes them on its own stream, which keeps the last -// message per subject, `take-in..`. A wait starts with `take-in.pending` and ends with `take-in.taken`, -// whoever ended it (the named indexer, a revert, a hand take-in or a drop). The stream's name is the indexer's manifest's; -// these names are its events, relative to it. +// The take-in events (novox/hq ADR 0633 §1, §5): the indexer's ordinary events, `take-in.pending` and `take-in.taken`, on +// the path its other events take (the module's event subject; its manifest's emits), with no stream of its own. A wait +// starts with `take-in.pending` and ends with `take-in.taken`, whoever ended it (the named indexer, a revert, a hand +// take-in or a drop); the wait's key is in the body (TakeIn.Key), since an emitted event names one event and a key is +// no event name. A reader that starts later asks the waits of indexer_pending, not a replay. const ( - TakeInStream = "take-in" TakeInPending = "pending" TakeInTaken = "taken" + // TakeInPendingEvent and TakeInTakenEvent are the two events' names, relative to the indexer. + TakeInPendingEvent = "take-in." + TakeInPending + TakeInTakenEvent = "take-in." + TakeInTaken ) // What waits in a take-in event: a source revision for the indexer build it names, or a delivery group for its end. @@ -80,27 +83,24 @@ func GroupTakeInKey(group string) string { }, group) } -// TakeInEvent is the name of a take-in event of the given kind and key: `take-in..`. -func TakeInEvent(kind, key string) (string, error) { +// TakeInEvent is the name of the take-in event of the given kind: `take-in.`. +func TakeInEvent(kind string) (string, error) { if kind != TakeInPending && kind != TakeInTaken { return "", fmt.Errorf("delivery: the take-in kind %q is not pending or taken", kind) } - if err := checkTakeInKey(key); err != nil { - return "", err - } - return TakeInStream + "." + kind + "." + key, nil + return "take-in." + kind, nil } -// ParseTakeInEvent is the inverse of TakeInEvent. -func ParseTakeInEvent(event string) (kind, key string, err error) { - parts := strings.SplitN(event, ".", 3) - if len(parts) != 3 || parts[0] != TakeInStream { - return "", "", fmt.Errorf("delivery: %q is not a take-in event, take-in..", event) +// ParseTakeInEvent is the inverse of TakeInEvent: the kind of a take-in event's name. +func ParseTakeInEvent(event string) (kind string, err error) { + k, ok := strings.CutPrefix(event, "take-in.") + if !ok { + return "", fmt.Errorf("delivery: %q is not a take-in event, take-in.", event) } - if _, err := TakeInEvent(parts[1], parts[2]); err != nil { - return "", "", fmt.Errorf("delivery: %q is not a take-in event: %w", event, err) + if _, err := TakeInEvent(k); err != nil { + return "", fmt.Errorf("delivery: %q is not a take-in event: %w", event, err) } - return parts[1], parts[2], nil + return k, nil } func checkTakeInKey(key string) error { @@ -136,7 +136,10 @@ type TakeIn struct { } func (t TakeIn) check() error { - if _, err := TakeInEvent(t.Kind, t.Key); err != nil { + if _, err := TakeInEvent(t.Kind); err != nil { + return err + } + if err := checkTakeInKey(t.Key); err != nil { return err } if len(t.Merges) == 0 { @@ -213,7 +216,7 @@ func SameTakeIn(a, b TakeIn) bool { return reflect.DeepEqual(a, b) } // merge's key, waiting for the indexer build WaitsFor (empty while it is worked out), or a group being gathered under // its group key, with Members every approved member and Merges those held so far. A revision that waits behind a // group being gathered names that group's key in Behind. Keys are the pending keys of its merges, in Merges' order (indexer_drop_pending drops one). Holds are the modules it holds back: every module of its -// merges' trees. Said is the last take-in event said of it (`take-in..`) and SaidAt when. +// merges' trees. Said is the last take-in event said of it (`take-in.`, of its Key) and SaidAt when. type PendingWait struct { Key string `json:"key"` What string `json:"what"` diff --git a/modules/mesh-delivery-indexer/takein/takein_test.go b/modules/mesh-delivery-indexer/takein/takein_test.go index a37bc4c6..f9a5d6cd 100644 --- a/modules/mesh-delivery-indexer/takein/takein_test.go +++ b/modules/mesh-delivery-indexer/takein/takein_test.go @@ -25,21 +25,21 @@ func sampleTakeIns(since time.Time) []TakeIn { // TestTakeInEvents holds the indexer's take-in contract, moved from mesh-sdk go/delivery with its tests (1585/T2, T12). func TestTakeInEvents(t *testing.T) { - // The take-in events, `take-in..`, a group's key with its dots and no subject wildcard. + // The take-in events, `take-in.`; a group's key with its dots and no subject wildcard, in the body. if PendingKey(42) != "m.00000000000000000042" || GroupTakeInKey("feat/a.b *>") != "g.feat/a.b___" { t.Errorf("take-in keys %q %q", PendingKey(42), GroupTakeInKey("feat/a.b *>")) } - for _, c := range []struct{ kind, key string }{{TakeInPending, PendingKey(7)}, {TakeInTaken, GroupTakeInKey("feat/a.b")}} { - ev, err := TakeInEvent(c.kind, c.key) - if err != nil || ev != "take-in."+c.kind+"."+c.key { + for kind, want := range map[string]string{TakeInPending: TakeInPendingEvent, TakeInTaken: TakeInTakenEvent} { + ev, err := TakeInEvent(kind) + if err != nil || ev != want || ev != "take-in."+kind { t.Fatalf("take-in event = %q, %v", ev, err) } - if k, key, err := ParseTakeInEvent(ev); err != nil || k != c.kind || key != c.key { - t.Errorf("parse take-in event %q = %q %q %v", ev, k, key, err) + if k, err := ParseTakeInEvent(ev); err != nil || k != kind { + t.Errorf("parse take-in event %q = %q %v", ev, k, err) } } - for _, bad := range []string{"", "take-in.pending", "take-in.held.m.1", "take-in.taken.x.1", "take-in.taken.m..1", "forge.taken.m.1"} { - if _, _, err := ParseTakeInEvent(bad); err == nil { + for _, bad := range []string{"", "take-in", "take-in.held", "take-in.pending.m.1", "forge.taken"} { + if _, err := ParseTakeInEvent(bad); err == nil { t.Errorf("parse take-in event %q accepted", bad) } } @@ -74,6 +74,7 @@ func TestTakeInEvents(t *testing.T) { {"take-in without merges", "no merge", 0, func(t *TakeIn) { t.Merges = nil }}, {"take-in without since", "since", 0, func(t *TakeIn) { t.Since = time.Time{} }}, {"a merge of no repository", "owner", 0, func(t *TakeIn) { t.Merges = []Merge{{Repository: "nope", Merge: "m"}} }}, + {"a key that is neither a merge's nor a group's", "neither", 0, func(t *TakeIn) { t.Key = "x.1" }}, } { ti := sampleTakeIns(since)[c.i] c.mut(&ti) diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go index 6be59ed0..443b6455 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -24,7 +24,6 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" - "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/provisions" @@ -378,7 +377,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro dead.ages() dead.check() provisioned(ctx, js, w, logger.Printf) - // The indexer's stream take-in (ADR 0633 §5): a revision or a group it holds back is a wait, ended by take-in.taken. The + // The indexer's take-in events (ADR 0633 §5): a revision or a group it holds back is a wait, ended by take-in.taken. The // durable consumer _ is made by the controller from this module's consumes, and it is durable: what it // acknowledged before a restart is not heard again. So the waits the indexer holds now are asked of it first, through // its tool indexer_pending (this module's invokes), each with its own since; the consumer then hears what follows. @@ -388,10 +387,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro seedTakeIn(w, calls.Call, logger.Printf) calls.Close() } - switch tc, err := cred.EventConsumer(ctx, js, catalogue.StreamBusName(takein.TakeInStream)); { + // The take-in events are the indexer's ordinary events (its emits), heard on the shared events stream through this + // module's durable consumer, which its consumes filter to them. + switch tc, err := cred.EventConsumer(ctx, js, eventsStream); { case err != nil: logger.Printf("the consumer %s_%s on %s does not exist yet, so no take-in wait is said until the watchman starts again: %v", - cred.Node, cred.Module, catalogue.StreamBusName(takein.TakeInStream), err) + cred.Node, cred.Module, eventsStream, err) default: // The replay has caught up when a message says none is pending after it, or when the consumer had none pending at its // bind (the stream holds nothing it has not acknowledged): the waits past their bound are said then, not before. @@ -473,6 +474,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } } +// eventsStream is the shared stream of the modules' ordinary events, as the bus names it. +const eventsStream = "EVENTS" + // pendingTool is the indexer's tool that answers the waits it holds (ADR 0633 §5), as the view asks it. const pendingTool = delivery.IndexerModule + ".indexer_pending" @@ -491,7 +495,7 @@ func seedTakeIn(w *watch.Watchman, call func(key string, body any) (json.RawMess logf("take-in: %s cannot be asked (%v), so a wait the indexer held before this start is said only once a take-in event of it is heard", pendingTool, err) } -// takeInHeard hands one event of a take-in stream to the watchman, by its name relative to the stream, and acknowledges it; +// takeInHeard hands one take-in event of the indexer to the watchman, by its name relative to the indexer, and acknowledges it; // one that is not a take-in event is said and acknowledged, since hearing it again would not read it. func takeInHeard(w *watch.Watchman, m jetstream.Msg, logf func(string, ...any)) { _, event, ok := strings.Cut(m.Subject(), ".event.") diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index f6802626..e03c00ec 100644 --- a/modules/mesh-delivery-watchman/internal/watch/findings_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/findings_test.go @@ -622,7 +622,7 @@ const ( func takeInEvent(t *testing.T, kind, key string) string { t.Helper() - e, err := takein.TakeInEvent(kind, key) + e, err := takein.TakeInEvent(kind) if err != nil { t.Fatal(err) } diff --git a/modules/mesh-delivery-watchman/internal/watch/takein.go b/modules/mesh-delivery-watchman/internal/watch/takein.go index 9bb48246..3c3465ff 100644 --- a/modules/mesh-delivery-watchman/internal/watch/takein.go +++ b/modules/mesh-delivery-watchman/internal/watch/takein.go @@ -24,11 +24,11 @@ const ( // takeInRow is the signals table's row of a take-in wait. const takeInRow = "take-in" -// TakeIn hears one event of the indexer's stream `take-in`, by its name relative to the stream (`take-in..`): -// a pending one starts the wait for its key, whose bound runs from the event's since, and a taken one ends it. A group's +// TakeIn hears one take-in event of the indexer, by its name relative to the indexer (`take-in.`), the wait's key +// in its body: a pending one starts the wait for its key, whose bound runs from the event's since, and a taken one ends it. A group's // taken event ends the group's wait only: the revision it becomes is its own key and starts its own, longer wait. func (w *Watchman) TakeIn(event string, body []byte) { - kind, key, err := takein.ParseTakeInEvent(event) + kind, err := takein.ParseTakeInEvent(event) if err != nil { w.say("a take-in event is left: %v", err) return @@ -38,6 +38,11 @@ func (w *Watchman) TakeIn(event string, body []byte) { w.say("the body of %s cannot be read, left: %v", event, err) return } + if t.Kind != kind { + w.say("the body of %s says the kind %q, left", event, t.Kind) + return + } + key := t.Key w.mu.Lock() defer w.mu.Unlock() w.takeIn(kind, key, t) diff --git a/modules/mesh-delivery-watchman/manifest_test.go b/modules/mesh-delivery-watchman/manifest_test.go index 342caead..b6c85b07 100644 --- a/modules/mesh-delivery-watchman/manifest_test.go +++ b/modules/mesh-delivery-watchman/manifest_test.go @@ -97,7 +97,7 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6 t.Errorf("reads = %v", reads) } // 1585/T20: it hears the indexer's stream take-in, whole, through a durable consumer the controller makes from consumes. - if !reflect.DeepEqual(m.Consumes, []string{"mesh-delivery-indexer.take-in.**"}) { + if !reflect.DeepEqual(m.Consumes, []string{"mesh-delivery-indexer.take-in.pending", "mesh-delivery-indexer.take-in.taken"}) { t.Errorf("consumes = %v", m.Consumes) } // The waits the indexer holds at the watchman's start are asked of its tool (ADR 0633 §5), a verb edge as the view's. diff --git a/modules/mesh-delivery-watchman/module.json b/modules/mesh-delivery-watchman/module.json index 7cdc94d5..ae654b78 100644 --- a/modules/mesh-delivery-watchman/module.json +++ b/modules/mesh-delivery-watchman/module.json @@ -27,7 +27,8 @@ "node-engine.have" ], "consumes": [ - "mesh-delivery-indexer.take-in.**" + "mesh-delivery-indexer.take-in.pending", + "mesh-delivery-indexer.take-in.taken" ], "invokes": [ "mesh-delivery-indexer.indexer_pending"