diff --git a/modules/mesh-recovery/cmd/mesh-recovery/main.go b/modules/mesh-recovery/cmd/mesh-recovery/main.go index d85dc1369..edeb28a9d 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/main.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/main.go @@ -139,11 +139,20 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if _, err := account.HearSeatEvents(ctx, js, meshbroker.StreamSnapshotRestored, restored.hear, logger.Printf); err != nil { return err } + // The module's event hub: the binding of its finding stream and the seat list register on it before it starts. + hub := process.NewEventHub(logger.Printf) + // The module's finding stream is its `finding-events` entry, asked of the seat mesh-broker's show (A-274, §529): the + // writer says its events under that binding's subjects. + binding := process.NewBinding(meshbroker.Seat, account.Module, conn.CallKept, logger.Printf, views.FindingEventsLocal) + if err := binding.Register(hub); err != nil { + return err + } + binding.Ask(views.FindingEventsLocal) // The module's findings go through mesh-sdk's held writer over its own state findings (§561), started before the // engine's start, so what the start writes is held and written. writer, err := views.NewHeldWriter(views.WriterConfig[meshconditions.Finding, meshconditions.FindingCleared, meshconditions.FindingsStated, meshconditions.FindingsAskedEvent]{ Asked: meshconditions.SubjectFindingsAsked, Module: account.Module, Node: account.Node, State: conn.State("findings"), - Kinds: recovery.FindingKinds, JS: js, Conn: nc, Logf: logger.Printf}) + Kinds: recovery.FindingKinds, JS: js, Conn: nc, Bindings: binding, Logf: logger.Printf}) if err != nil { return err } @@ -171,7 +180,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } // The seat list, read over the connection and heard on the module's event hub (seatlist.Part), which the seat's verbs check // their callers against (A-179, A-186): registered before the hub starts, read again once its events are live. - hub := process.NewEventHub(logger.Printf) seats := seatlist.NewPart("recovery", nil) seats.Follow(seatlist.ProcessBus(conn.Call, hub), nil, logger.Printf) if err := hub.StartUntilBound(ctx, js, account, nil, seats.EventsLive); err != nil { diff --git a/modules/mesh-recovery/cmd/mesh-recovery/whole_integration_test.go b/modules/mesh-recovery/cmd/mesh-recovery/whole_integration_test.go index 99b05d39c..0aad2a95a 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/whole_integration_test.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/whole_integration_test.go @@ -85,6 +85,7 @@ func TestRecoveryWhole(t *testing.T) { for _, s := range []jetstream.StreamConfig{ {Name: "RECOVERY_EVENTS", Subjects: []string{"mesh.seat.mesh-recovery.event.>"}}, {Name: "RECOVERY_FINDINGS", Subjects: []string{"mesh.stream.mesh-recovery.event.>"}}, + {Name: "RECOVERY_FINDING_EVENTS", Subjects: []string{"mesh.bus.mesh-recovery.finding-events.>"}}, {Name: "FINDINGS_ASKED", Subjects: []string{meshconditions.SubjectFindingsAsked}}, } { if _, err := js.CreateStream(ctx, s); err != nil { @@ -169,6 +170,11 @@ func TestRecoveryWhole(t *testing.T) { answer(m, seatlist.Answer{Seat: in.Seat, Found: in.Seat == seat.Seat, Entry: &entry, Revision: 1}) listedOnce.Do(func() { close(listed) }) }) + // The seat mesh-broker's show names the module's `finding-events` entry (A-274): its finding events go under that subject. + sub("mesh.seat.mesh-broker.call.show.*", func(m *nats.Msg) { + answer(m, map[string]any{"binding": map[string]any{"local": "finding-events", "storage": "stream", "template": "events", + "stream": "RECOVERY_FINDING_EVENTS", "subject": "mesh.bus.mesh-recovery.finding-events"}}) + }) keeper := &standKeeper{} restored := func(request, act, outcome string) { b, _ := json.Marshal(meshbroker.SnapshotRestoredEvent{Request: request, Act: act, Outcome: outcome, Error: "cancelled"}) @@ -499,6 +505,24 @@ func TestRecoveryWhole(t *testing.T) { t.Errorf("restored %v, live %q", ev, live()) } }) + + t.Run("the finding events go under the module's finding-events binding, none on the old stream (A-274, §529)", func(t *testing.T) { + st, err := js.Stream(ctx, "RECOVERY_FINDING_EVENTS") + if err != nil { + t.Fatal(err) + } + info, err := st.Info(ctx) + if err != nil || info.State.Msgs == 0 { + t.Fatalf("the finding stream holds no event (%v): %+v", err, info) + } + old, err := js.Stream(ctx, "RECOVERY_FINDINGS") + if err != nil { + t.Fatal(err) + } + if oi, err := old.Info(ctx); err != nil || oi.State.Msgs != 0 { + t.Errorf("the old stream holds %d finding events (%v)", oi.State.Msgs, err) + } + }) } type testLog struct{ t *testing.T } diff --git a/modules/mesh-recovery/module.json b/modules/mesh-recovery/module.json index 28978de32..9ad3e4fa9 100644 --- a/modules/mesh-recovery/module.json +++ b/modules/mesh-recovery/module.json @@ -46,6 +46,7 @@ "seat:mesh-catalog.entry", "seat:mesh-module-assigner.assignments", "seat:mesh-hand-acts.hand-act", + "seat:mesh-broker.show", "seat:mesh-broker.snapshot-get", "seat:mesh-broker.snapshot-restore", "seat:mesh-broker.snapshot-restore-cancel", @@ -267,6 +268,7 @@ "seat:mesh-catalog.entry", "seat:mesh-module-assigner.assignments", "seat:mesh-hand-acts.hand-act", + "seat:mesh-broker.show", "seat:mesh-broker.snapshot-get", "seat:mesh-broker.snapshot-restore", "seat:mesh-broker.snapshot-restore-cancel", diff --git a/modules/restic/cmd/restic-backups/itembus.go b/modules/restic/cmd/restic-backups/itembus.go index cd0d95a41..43a30b01e 100644 --- a/modules/restic/cmd/restic-backups/itembus.go +++ b/modules/restic/cmd/restic-backups/itembus.go @@ -86,9 +86,13 @@ func (l *liveBus) PutOffs() PutOffState { return stateOfPutOffs{l.conn.State(Put // Findings starts the module's held finding writer over its state `findings` (§561), as the recovery module does (§461): started // before the night's first backup, so what it writes is held and written. It answers how to close it. func (l *liveBus) Findings(ctx context.Context, logf func(string, ...any)) (Findings, func(), error) { + // The module's finding stream is its `finding-events` entry, asked of the seat mesh-broker's show once (A-274, §529); the + // night's process lives one night, so it hears no recreated of it. Until it answers, the writer holds its events. + binding := process.NewBinding(BrokerSeat, l.account.Module, l.conn.CallKept, logf, views.FindingEventsLocal) + binding.Ask(views.FindingEventsLocal) w, err := views.NewHeldWriter(views.WriterConfig[meshconditions.Finding, meshconditions.FindingCleared, meshconditions.FindingsStated, meshconditions.FindingsAskedEvent]{ Asked: meshconditions.SubjectFindingsAsked, Module: l.account.Module, Node: l.account.Node, State: l.conn.State("findings"), - Kinds: []string{FindingBackupWaits}, JS: l.js, Conn: l.nc, Logf: logf}) + Kinds: []string{FindingBackupWaits}, JS: l.js, Conn: l.nc, Bindings: binding, Logf: logf}) if err != nil { return nil, nil, err } diff --git a/modules/restic/cmd/restic-backups/live_integration_test.go b/modules/restic/cmd/restic-backups/live_integration_test.go index 87d16daae..44712bb24 100644 --- a/modules/restic/cmd/restic-backups/live_integration_test.go +++ b/modules/restic/cmd/restic-backups/live_integration_test.go @@ -145,6 +145,7 @@ func TestTheNightSaysItsItemsAndGivesWayToARestoreOnTheBus(t *testing.T) { } } for _, s := range []jetstream.StreamConfig{{Name: "RESTIC_FINDINGS", Subjects: []string{"mesh.stream.restic.event.>"}}, + {Name: "RESTIC_FINDING_EVENTS", Subjects: []string{"mesh.bus.restic.finding-events.>"}}, {Name: "FINDINGS_ASKED", Subjects: []string{meshconditions.SubjectFindingsAsked}}} { if _, err := js.CreateStream(ctx, s); err != nil { t.Fatal(err) @@ -158,6 +159,14 @@ func TestTheNightSaysItsItemsAndGivesWayToARestoreOnTheBus(t *testing.T) { }); err != nil { t.Fatal(err) } + // The seat mesh-broker's show names the module's `finding-events` entry (A-274). + if _, err := nc.Subscribe("mesh.seat.mesh-broker.call.show.*", func(msg *nats.Msg) { + body, _ := json.Marshal(map[string]any{"result": map[string]any{"binding": map[string]any{"local": "finding-events", "storage": "stream", + "template": "events", "stream": "RESTIC_FINDING_EVENTS", "subject": "mesh.bus.restic.finding-events"}}, "node": "ace"}) + _ = msg.Respond(body) + }); err != nil { + t.Fatal(err) + } findings, closeFindings, err := live.Findings(ctx, t.Logf) if err != nil { t.Fatal(err) @@ -241,4 +250,14 @@ func TestTheNightSaysItsItemsAndGivesWayToARestoreOnTheBus(t *testing.T) { if err != nil || waits != 1 { t.Fatalf("the findings kept: %v, %v", ks, err) } + // A-274: the finding events went under the finding-events binding, none on the old stream. + for stream, want := range map[string]bool{"RESTIC_FINDING_EVENTS": true, "RESTIC_FINDINGS": false} { + st, err := js.Stream(ctx, stream) + if err != nil { + t.Fatal(err) + } + if info, err := st.Info(ctx); err != nil || (info.State.Msgs > 0) != want { + t.Fatalf("%s holds %d finding events (%v), want some: %v", stream, info.State.Msgs, err, want) + } + } } diff --git a/modules/restic/module.json b/modules/restic/module.json index 6482eb39e..e46097bd4 100644 --- a/modules/restic/module.json +++ b/modules/restic/module.json @@ -16,6 +16,9 @@ ] } ], + "invokes": [ + "seat:mesh-broker.show" + ], "consumes": [ "mesh-recovery.restore-started", "mesh-recovery.restored",