Area 5 (join 11): mesh-recovery's and restic's finding writers pass their finding-events binding
nats and distribution declare the old stream form and require no finding-events, so their writers stay on it. Builds: design 47 G A-274 (answers G-22) Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -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 {
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -16,6 +16,9 @@
|
||||
]
|
||||
}
|
||||
],
|
||||
"invokes": [
|
||||
"seat:mesh-broker.show"
|
||||
],
|
||||
"consumes": [
|
||||
"mesh-recovery.restore-started",
|
||||
"mesh-recovery.restored",
|
||||
|
||||
Reference in New Issue
Block a user