diff --git a/cmd/mesh-host/backup_event_test.go b/cmd/mesh-host/backup_event_test.go new file mode 100644 index 0000000..0a773cc --- /dev/null +++ b/cmd/mesh-host/backup_event_test.go @@ -0,0 +1,42 @@ +package main + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" +) + +func backupEvent(node, module string, ok bool, at time.Time, why string) []byte { + raw, _ := json.Marshal(map[string]any{"node": node, "module": module, "ok": ok, "at": at, "error": why}) + return raw +} + +// A backup is done when the holder says so for this node and module, held on its words when it says it failed, and held +// when it says nothing within the bound; another node's or module's event, or one from before the ask, is not the answer +// (R6-A A5, novox/hq ADR 0429 §14). +func TestABackupEndsWhenItsEventSaysSo(t *testing.T) { + since := time.Now() + run := func(within time.Duration, events ...[]byte) (done bool, words string) { + ch := make(chan []byte, len(events)) + for _, e := range events { + ch <- e + } + got := awaitBackup(context.Background(), ch, "n1", "db", since, within) + return got.Done, got.Words + } + if done, _ := run(time.Second, + backupEvent("n2", "db", true, since, ""), backupEvent("n1", "web", true, since, ""), + backupEvent("n1", "db", false, since.Add(-time.Hour), "an old night"), + backupEvent("n1", "db", true, since.Add(time.Second), "")); !done { + t.Error("the event of this node and module was not the answer") + } + if done, words := run(time.Second, backupEvent("n1", "db", false, since, "the repository is locked")); done || + !strings.Contains(words, "repository is locked") { + t.Errorf("a failed backup: done=%v words=%q", done, words) + } + if done, words := run(50 * time.Millisecond); done || !strings.Contains(words, "did not say it was done") { + t.Errorf("silence within the bound: done=%v words=%q", done, words) + } +} diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index afa82ad..a0dc79c 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -89,52 +89,61 @@ func shorten(s string) string { return s[:most-1] + "…" } -// backupWithin is how long a backup is given from its ask to say it is done (ADR 0429 §14), and backupPollEvery how -// soon the holder is asked again how it went while it works. -var ( - backupWithin = 10 * time.Minute - backupPollEvery = 5 * time.Second -) +// backupWithin is how long a backup is given from its ask to say it is done (ADR 0429 §14). +var backupWithin = 10 * time.Minute -// backupNode asks the holder of node-backup on this node to back a module up now, and waits for it to say how it went: -// the holder answers when it has started, and its `backed-up` says the module's last night. +// backedUpEvent is where the holder of node-backup says a backup taken by hand ended, also one that could not start. +const backedUpEvent = "mesh.mod.restic.event.backed-up" + +// backupNode asks the holder of node-backup on this node to back a module up now, and waits for its event to say how it +// went: the holder answers `now` when it has started, and says `backed-up` when the backup ends. func backupNode(session *link.Session) func(ctx context.Context, module string) agent.Backup { return func(ctx context.Context, module string) agent.Backup { + // Heard before the ask, so an event that comes at once is not lost. + events, stop, err := session.Hear(backedUpEvent) + if err != nil { + return agent.Backup{Words: "the end of the backup cannot be heard: " + err.Error()} + } + defer stop() asked := time.Now() if _, err := session.AskSeat(ctx, "node-backup", "now", map[string]any{"module": module}, 30*time.Second); err != nil { return agent.Backup{Words: err.Error()} } - deadline := asked.Add(backupWithin) - for { - select { - case <-ctx.Done(): - return agent.Backup{Words: "the node agent was stopped while the backup ran"} - case <-time.After(backupPollEvery): + return awaitBackup(ctx, events, session.Membership.Node, module, asked, backupWithin) + } +} + +// awaitBackup waits for the event that says the backup of a module on this node ended, within the bound: done when it +// says ok, held on its words when it does not, and held when it says nothing within the bound. An event of another node +// or module, or from before the ask, is not the answer. +func awaitBackup(ctx context.Context, events <-chan []byte, node, module string, since time.Time, within time.Duration) agent.Backup { + bound := time.NewTimer(within) + defer bound.Stop() + for { + select { + case <-ctx.Done(): + return agent.Backup{Words: "the node agent was stopped while the backup ran"} + case <-bound.C: + return agent.Backup{Words: fmt.Sprintf("the backup did not say it was done within %s", within)} + case raw := <-events: + var e struct { + Node string `json:"node"` + Module string `json:"module"` + OK bool `json:"ok"` + At time.Time `json:"at"` + Error string `json:"error"` } - raw, err := session.AskSeat(ctx, "node-backup", "backed-up", map[string]any{"module": module}, 10*time.Second) - if err == nil { - var read []struct { - Module string `json:"module"` - LastNight *struct { - OK bool `json:"ok"` - At time.Time `json:"at"` - Error string `json:"error"` - } `json:"lastNight"` - } - if json.Unmarshal(raw, &read) == nil { - for _, m := range read { - if m.Module == module && m.LastNight != nil && m.LastNight.At.After(asked) { - if m.LastNight.OK { - return agent.Backup{Done: true} - } - return agent.Backup{Words: "the backup failed: " + m.LastNight.Error} - } - } - } + if json.Unmarshal(raw, &e) != nil || e.Node != node || e.Module != module || e.At.Before(since.Add(-time.Second)) { + continue } - if time.Now().After(deadline) { - return agent.Backup{Words: fmt.Sprintf("the backup did not say it was done within %s", backupWithin)} + if e.OK { + return agent.Backup{Done: true} } + words := "the backup failed" + if e.Error != "" { + words += ": " + e.Error + } + return agent.Backup{Words: words} } } } diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index df7f1d9..f35a722 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -230,32 +230,19 @@ func TestNatsAWantAndItsCompositionAreAppliedWithinASecond(t *testing.T) { // module (nothing applied, the have says held with the backup's words); when its want then says skip, it is applied. func TestNatsADataModuleAsksItsBackupFirstAndIsHeldWhenItFails(t *testing.T) { b := newItBed(t) - backupPollEvery = 20 * time.Millisecond var mu sync.Mutex var asked []string var onDiskWhenAsked string - var nightOK bool - var at time.Time if _, err := b.conn.Subscribe("mesh.seat.node-backup.tool.now."+itNode, func(m *nats.Msg) { mu.Lock() asked = append(asked, "now") onDiskWhenAsked = b.file("db") - at = time.Now() mu.Unlock() _ = m.Respond([]byte(`{"result":{"started":"db"}}`)) - }); err != nil { - t.Fatal(err) - } - if _, err := b.conn.Subscribe("mesh.seat.node-backup.tool.backed-up."+itNode, func(m *nats.Msg) { - mu.Lock() - defer mu.Unlock() - if at.IsZero() { - _ = m.Respond([]byte(`{"result":[]}`)) - return - } - night, _ := json.Marshal(map[string]any{"module": "db", "lastNight": map[string]any{"ok": nightOK, - "at": time.Now().UTC(), "error": "restic: the repository is locked"}}) - _ = m.Respond([]byte(`{"result":[` + string(night) + `]}`)) + // The holder says the backup ended on its event. + ev, _ := json.Marshal(map[string]any{"node": itNode, "module": "db", "ok": false, "at": time.Now().UTC(), + "error": "restic: the repository is locked"}) + _ = b.conn.Publish(backedUpEvent, ev) }); err != nil { t.Fatal(err) } diff --git a/internal/link/watch.go b/internal/link/watch.go index 8b1c123..0a78e6e 100644 --- a/internal/link/watch.go +++ b/internal/link/watch.go @@ -127,3 +127,50 @@ func (q *Session) WhenLinked(ctx context.Context) (context.Context, context.Canc } } } + +// EventBus is a link that can hear an event on a subject of core NATS: the node agent waits for a backup's end with it, +// and never polls. +type EventBus interface { + // Hear yields the payloads published on the subject until stop is called or the link ends. + Hear(subject string) (events <-chan []byte, stop func(), err error) +} + +func (b OverNATS) Hear(subject string) (<-chan []byte, func(), error) { + out := make(chan []byte, 16) + sub, err := b.Conn.Subscribe(subject, func(m *nats.Msg) { + select { + case out <- m.Data: + default: + } + }) + if err != nil { + return nil, nil, err + } + if err := b.Conn.Flush(); err != nil { + _ = sub.Unsubscribe() + return nil, nil, err + } + return out, func() { _ = sub.Unsubscribe() }, nil +} + +func (l *natsLink) Hear(subject string) (<-chan []byte, func(), error) { + return OverNATS{Conn: l.conn, JS: l.js}.Hear(subject) +} + +var ( + _ EventBus = OverNATS{} + _ EventBus = (*natsLink)(nil) +) + +// Hear listens for an event on the link open now, from before the caller asks for what it waits on. +func (q *Session) Hear(subject string) (<-chan []byte, func(), error) { + q.init() + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + h, ok := bus.(EventBus) + if bus == nil || !ok { + return nil, nil, ErrNoLink + } + return h.Hear(subject) +} diff --git a/module.json b/module.json index d92aa6c..fcb6b77 100644 --- a/module.json +++ b/module.json @@ -14,6 +14,9 @@ } ] }, + "consumes": [ + "restic.backed-up" + ], "config": { "operator-signers": { "type": "string", diff --git a/packaging/operator_keys_test.go b/packaging/operator_keys_test.go index 723cd2b..5935047 100644 --- a/packaging/operator_keys_test.go +++ b/packaging/operator_keys_test.go @@ -52,3 +52,20 @@ func TestTheManifestDeclaresTheOperatorKeysFileFedByASetting(t *testing.T) { } t.Fatal("module.json declares no operator-keys file") } + +// The node agent hears the end of a backup on the event of node-backup's holder, so the manifest consumes it. +func TestTheManifestConsumesTheBackupEvent(t *testing.T) { + raw, err := os.ReadFile("../module.json") + if err != nil { + t.Fatal(err) + } + var m struct { + Consumes []string `json:"consumes"` + } + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + if len(m.Consumes) != 1 || m.Consumes[0] != "restic.backed-up" { + t.Fatalf("consumes %v", m.Consumes) + } +}