diff --git a/modules/restic/cmd/restic-backups/backups.go b/modules/restic/cmd/restic-backups/backups.go index ff33e5036..59fd91f52 100644 --- a/modules/restic/cmd/restic-backups/backups.go +++ b/modules/restic/cmd/restic-backups/backups.go @@ -208,6 +208,10 @@ type Backups struct { // measuring run as processes of their own). Node is the machine's name, said in the events. Emit func(event string, body any) error Node string + // Bus says the seat node-backup's item events and reads the recovery module's restores (items.go); nil where this process has + // no bus account. BrokerModule is the module that holds the seat mesh-broker now, read from the seat list. + Bus ItemBus + BrokerModule func() (string, error) // Readings is every reading disk-load keeps for this machine (its state `readings`, ADR 0571 clause 3); nil where the // process has no runtime to read it. Readings func() ([]diskreadings.Reading, error) @@ -332,8 +336,21 @@ func (b *Backups) ensureRepository(ctx context.Context) error { // one is one module's night: its run lines, then one snapshot of its paths. A failure is that // module's alone. -func (b *Backups) one(ctx context.Context, d Declared, hand bool) Night { - n := Night{At: b.Now().UTC()} +func (b *Backups) one(ctx context.Context, d Declared, hand bool) (n Night) { + n = Night{At: b.Now().UTC()} + ended, err := b.beginItems(ctx, d.Module) + if err != nil { + n.Error = err.Error() + b.Say("%s: NOT backed up: %s", d.Module, n.Error) + return n + } + defer func() { + if n.OK { + ended(nil) + } else { + ended(errors.New(n.Error)) + } + }() fail := func(err error) Night { n.Error = err.Error() b.Say("%s: NOT backed up: %s", d.Module, n.Error) diff --git a/modules/restic/cmd/restic-backups/itembus.go b/modules/restic/cmd/restic-backups/itembus.go new file mode 100644 index 000000000..6961f2689 --- /dev/null +++ b/modules/restic/cmd/restic-backups/itembus.go @@ -0,0 +1,215 @@ +package main + +// The live bus of the item events (items.go): the module's own broker account, as the recovery module's process holds its own +// (design 47 G A-150: the process `night` has a bus account of its own, says its items and reads the recovery module's restores). + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/seatlist" + + "github.com/nats-io/nats.go/jetstream" +) + +// The streams and subjects of the seats' events the holder reads and says. +const ( + backupSeat = "node-backup" + recoverySeat = "mesh-recovery" + streamItemStarted = "SEAT_node-backup_item-started" + streamItemEnded = "SEAT_node-backup_item-ended" + streamRestoreStarted = "SEAT_mesh-recovery_restore-started" + streamRestored = "SEAT_mesh-recovery_restored" + readWait = 5 * time.Second +) + +func recoveryEventSubject(event, module, kept string) string { + return "mesh.seat." + recoverySeat + ".event." + event + "." + module + "." + kept +} + +func itemEventSubject(event, node, module, item string) string { + return "mesh.seat." + backupSeat + ".event." + event + "." + node + "." + module + "." + item +} + +// liveBus is ItemBus over the module's own account. +type liveBus struct { + conn *process.Conn + js jetstream.JetStream + node string +} + +// connectItems opens the bus as the module's own account when the process was given its credential (MESH_BROKER_FILE), and +// answers how to find the holder of the seat mesh-broker; nil, nil when it was not. +func connectItems(getenv func(string) string) (ItemBus, func() (string, error), func(), error) { + if strings.TrimSpace(getenv(process.BrokerFileEnv)) == "" { + return nil, nil, func() {}, nil + } + account, err := process.ReadBrokerAccount(getenv) + if err != nil { + return nil, nil, nil, err + } + conn, err := process.Connect(account) + if err != nil { + return nil, nil, nil, err + } + nc, js, err := account.Dial() + if err != nil { + conn.Close() + return nil, nil, nil, err + } + return liveBus{conn: conn, js: js, node: account.Node}, + func() (string, error) { return seatlist.HolderOf(conn.Call, BrokerSeat) }, + func() { nc.Close(); conn.Close() }, nil +} + +func (l liveBus) Say(event string, ref ItemRef, body map[string]any) error { + return l.conn.EmitSeat(backupSeat, event+"."+l.node+"."+keys.EscapeName(ref.Module)+"."+keys.EscapeName(ref.Item), body) +} + +// last reads every subject's last message of a stream through an ordered consumer made for this one read (§556). +func (l liveBus) last(ctx context.Context, stream, filter string, take func(subject string, data []byte)) error { + cons, err := l.js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{ + FilterSubjects: []string{filter}, DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy}) + if err != nil { + return fmt.Errorf("the read of %s: %w", stream, err) + } + info, err := cons.Info(ctx) + if err != nil { + return fmt.Errorf("the read of %s: %w", stream, err) + } + for left := info.NumPending; left > 0; { + m, err := cons.Next(jetstream.FetchMaxWait(readWait)) + if err != nil { + return fmt.Errorf("the read of %s: %w", stream, err) + } + take(m.Subject(), m.Data()) + meta, err := m.Metadata() + if err != nil { + return err + } + left = meta.NumPending + } + return nil +} + +func atOf(data []byte) time.Time { + var b struct { + At time.Time `json:"at"` + } + _ = json.Unmarshal(data, &b) + return b.At +} + +func (l liveBus) Unended() ([]ItemRef, error) { + ctx, cancel := context.WithTimeout(context.Background(), 2*readWait) + defer cancel() + started, ended := map[ItemRef]time.Time{}, map[ItemRef]time.Time{} + for _, c := range []struct { + stream, event string + into map[ItemRef]time.Time + }{{streamItemStarted, EventItemStarted, started}, {streamItemEnded, EventItemEnded, ended}} { + prefix := itemEventSubject(c.event, l.node, "", "") + prefix = strings.TrimSuffix(prefix, "..") + c := c + err := l.last(ctx, c.stream, itemEventSubject(c.event, l.node, "*", "*"), func(subject string, data []byte) { + module, item, ok := strings.Cut(strings.TrimPrefix(subject, prefix+"."), ".") + if !ok { + return + } + m, e1 := keys.UnescapeName(module) + i, e2 := keys.UnescapeName(item) + if e1 == nil && e2 == nil { + c.into[ItemRef{m, i}] = atOf(data) + } + }) + if err != nil { + return nil, err + } + } + var out []ItemRef + for ref, at := range started { + if end, ok := ended[ref]; !ok || end.Before(at) { + out = append(out, ref) + } + } + sortRefs(out) + return out, nil +} + +func (l liveBus) Restoring(module string) (string, error) { + ctx, cancel := context.WithTimeout(context.Background(), 2*readWait) + defer cancel() + type act struct { + id string + at time.Time + } + started, restored := map[string]act{}, map[string]act{} + for _, c := range []struct { + stream, event string + into map[string]act + }{{streamRestoreStarted, "restore-started", started}, {streamRestored, "restored", restored}} { + prefix := recoveryEventSubject(c.event, keys.EscapeName(module), "") + c := c + err := l.last(ctx, c.stream, recoveryEventSubject(c.event, keys.EscapeName(module), "*"), func(subject string, data []byte) { + var b struct { + Act string `json:"act"` + At time.Time `json:"at"` + } + if json.Unmarshal(data, &b) == nil { + c.into[strings.TrimPrefix(subject, prefix)] = act{b.Act, b.At} + } + }) + if err != nil { + return "", err + } + } + var running []string + for kept, s := range started { + if r, ok := restored[kept]; !ok || r.id != s.id { + running = append(running, s.id) + } + } + if len(running) == 0 { + return "", nil + } + if len(running) > 1 { + sortStrings(running) + } + return running[0], nil +} + +func (l liveBus) WaitRestored(ctx context.Context, want string) error { + cons, err := l.js.OrderedConsumer(ctx, streamRestored, jetstream.OrderedConsumerConfig{ + FilterSubjects: []string{"mesh.seat." + recoverySeat + ".event.restored.*.*"}, DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy}) + if err != nil { + return err + } + for { + m, err := cons.Next(jetstream.FetchMaxWait(time.Minute)) + if err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + continue + } + var b struct { + Act string `json:"act"` + } + if json.Unmarshal(m.Data(), &b) == nil && b.Act == want { + return nil + } + } +} + +func sortStrings(s []string) { + for i := 1; i < len(s); i++ { + for j := i; j > 0 && s[j] < s[j-1]; j-- { + s[j], s[j-1] = s[j-1], s[j] + } + } +} diff --git a/modules/restic/cmd/restic-backups/items.go b/modules/restic/cmd/restic-backups/items.go new file mode 100644 index 000000000..e7ba18e48 --- /dev/null +++ b/modules/restic/cmd/restic-backups/items.go @@ -0,0 +1,186 @@ +package main + +// The backup holder's item events and its give-way to a restore of the recovery module (design 47 G §554, §556, §563): each item +// of a module is said started and ended on the seat node-backup; for the items of the holder of the seat mesh-broker the holder +// says its start, then reads the recovery module's restore-started and restored once more, and when a restore runs it ends the +// item put off, naming that restore's act, and starts it again on that act's restored. Said means acknowledged. + +import ( + "context" + "errors" + "fmt" + "sort" + "strings" +) + +// The seat node-backup's events of an item, and the seat of the holder whose items are the bus's. +const ( + EventItemStarted = "item-started" + EventItemEnded = "item-ended" + BrokerSeat = "mesh-broker" +) + +// The item's outcomes on item-ended. +const ( + ItemDone = "done" + ItemFailed = "failed" + ItemPutOff = "put-off" +) + +// StoppedWhy is the error of an item said started and never ended, ended at the holder's next start. +const StoppedWhy = "stopped" + +// ItemRef is one item of a module on a node. +type ItemRef struct{ Module, Item string } + +// ItemBus is what the backup holder says and reads on the bus. +type ItemBus interface { + // Say publishes one item event, `item-started` or `item-ended`, and answers once the bus acknowledged it. + Say(event string, ref ItemRef, body map[string]any) error + // Unended are this node's items said started whose last start is newer than their last end, read last per subject (§556). + Unended() ([]ItemRef, error) + // Restoring reads the recovery module's restore-started and restored of a module's kept data once more, last per subject, and + // answers the act of a restore that runs, "" when none does. + Restoring(module string) (string, error) + // WaitRestored answers once the recovery module said restored with the act. + WaitRestored(ctx context.Context, act string) error +} + +// ItemsOf are the items of a module a backup covers: the data items it declares that are not a cache. +func (b *Backups) ItemsOf(module string) []ItemRef { + all, err := b.Items() + if err != nil { + b.Say("the data items of %s could not be read: %v", module, err) + return nil + } + var out []ItemRef + for _, it := range all[module] { + if it.Class == "cache" { + continue + } + out = append(out, ItemRef{Module: module, Item: it.ID}) + } + return out +} + +func (b *Backups) sayItem(event string, ref ItemRef, body map[string]any) error { + body["at"] = b.Now().UTC() + return b.Bus.Say(event, ref, body) +} + +// StopStale ends every item this node said started and never ended with `failed` and the error `stopped`, so no restore waits on +// a dump that no longer runs (§563). It holds the night's lock, so no backup of this module is running meanwhile. +func (b *Backups) StopStale() error { + if b.Bus == nil { + return nil + } + release, err := b.hold(nightLock) + if err != nil { + return err + } + defer release() + left, err := b.Bus.Unended() + if err != nil { + return fmt.Errorf("the items said started and never ended could not be read: %w", err) + } + var failed []string + for _, ref := range left { + if err := b.sayItem(EventItemEnded, ref, map[string]any{"outcome": ItemFailed, "error": StoppedWhy}); err != nil { + failed = append(failed, ref.Module+"."+ref.Item+": "+err.Error()) + } + } + if len(failed) > 0 { + return errors.New("items left started could not be ended: " + strings.Join(failed, "; ")) + } + return nil +} + +// beginItems says every item of a module started and, for the items of the holder of the seat mesh-broker, gives way to a restore +// that runs: its items are ended put off naming the act, wait for that restore's restored, and are started again. It answers how to +// end the items when the backup ends. A module with no item says nothing. +func (b *Backups) beginItems(ctx context.Context, module string) (func(err error), error) { + nothing := func(error) {} + if b.Bus == nil { + return nothing, nil + } + items := b.ItemsOf(module) + if len(items) == 0 { + return nothing, nil + } + started := func() error { + for _, ref := range items { + if err := b.sayItem(EventItemStarted, ref, map[string]any{}); err != nil { + return err + } + } + return nil + } + end := func(outcome string, extra map[string]any) error { + var first error + for _, ref := range items { + body := map[string]any{"outcome": outcome} + for k, v := range extra { + body[k] = v + } + if err := b.sayItem(EventItemEnded, ref, body); err != nil && first == nil { + first = err + } + } + return first + } + finish := func(err error) { + var said error + if err != nil { + said = end(ItemFailed, map[string]any{"error": err.Error()}) + } else { + said = end(ItemDone, nil) + } + if said != nil { + b.Say("the end of %s's items could not be said: %v", module, said) + } + } + if err := started(); err != nil { + return nothing, fmt.Errorf("the start of %s's items could not be said: %w", module, err) + } + if b.BrokerModule == nil { + return finish, nil + } + broker, err := b.BrokerModule() + if err != nil { + _ = end(ItemFailed, map[string]any{"error": err.Error()}) + return nothing, fmt.Errorf("who holds the seat %s is not known: %w", BrokerSeat, err) + } + if broker != module { + return finish, nil + } + for { + act, err := b.Bus.Restoring(module) + if err != nil { + _ = end(ItemFailed, map[string]any{"error": err.Error()}) + return nothing, fmt.Errorf("the recovery module's restores could not be read: %w", err) + } + if act == "" { + return finish, nil + } + b.Say("%s's backup gives way to the restore %s of the recovery module", module, act) + if err := end(ItemPutOff, map[string]any{"restore": act}); err != nil { + return nothing, fmt.Errorf("the put-off of %s's items could not be said: %w", module, err) + } + if err := b.Bus.WaitRestored(ctx, act); err != nil { + return nothing, fmt.Errorf("the end of the restore %s could not be waited for: %w", act, err) + } + if err := started(); err != nil { + return nothing, fmt.Errorf("the start of %s's items could not be said: %w", module, err) + } + } +} + +// sortRefs orders items by module, then item. +func sortRefs(refs []ItemRef) { + sort.Slice(refs, func(i, j int) bool { + if refs[i].Module != refs[j].Module { + return refs[i].Module < refs[j].Module + } + return refs[i].Item < refs[j].Item + }) +} diff --git a/modules/restic/cmd/restic-backups/items_test.go b/modules/restic/cmd/restic-backups/items_test.go new file mode 100644 index 000000000..88d022f2d --- /dev/null +++ b/modules/restic/cmd/restic-backups/items_test.go @@ -0,0 +1,127 @@ +package main + +import ( + "context" + "errors" + "fmt" + "reflect" + "testing" + "time" +) + +// itemBusFake records every call, in order, as the backup holder makes them. +type itemBusFake struct { + log []string + restoring []string // what each read of the recovery module's restores answers, in turn + unended []ItemRef + runFailure error +} + +func (f *itemBusFake) Say(event string, ref ItemRef, body map[string]any) error { + line := event + " " + ref.Module + "." + ref.Item + if o, ok := body["outcome"]; ok { + line += " " + fmt.Sprint(o) + } + if r, ok := body["restore"]; ok { + line += " restore=" + fmt.Sprint(r) + } + if e, ok := body["error"]; ok { + line += " error=" + fmt.Sprint(e) + } + if _, ok := body["at"]; !ok { + line += " NO-AT" + } + f.log = append(f.log, line) + return nil +} +func (f *itemBusFake) Unended() ([]ItemRef, error) { return f.unended, nil } +func (f *itemBusFake) Restoring(module string) (string, error) { + f.log = append(f.log, "read restores of "+module) + if len(f.restoring) == 0 { + return "", nil + } + act := f.restoring[0] + f.restoring = f.restoring[1:] + return act, nil +} +func (f *itemBusFake) WaitRestored(_ context.Context, act string) error { + f.log = append(f.log, "restored "+act) + return nil +} + +func itemsBackups(t *testing.T, bus *itemBusFake, broker string, runErr error) *Backups { + t.Helper() + return &Backups{Where: withData(t, composed, composedData), Node: "ace", Now: func() time.Time { return time.Date(2026, 11, 12, 3, 0, 0, 0, time.UTC) }, + Say: quiet, Bus: bus, BrokerModule: func() (string, error) { return broker, nil }, + Run: func(_ context.Context, name string, args ...string) (string, error) { + if name == "restic" && len(args) > 5 && args[5] == "backup" { + return "", runErr + } + return "[]", nil + }} +} + +func declaredOf(t *testing.T, b *Backups, module string) Declared { + t.Helper() + all, err := b.Declared() + must(t, err) + for _, d := range all { + if d.Module == module { + return d + } + } + t.Fatalf("%s is not declared", module) + return Declared{} +} + +// Design 47 G §563 (BKP4): every item of a backed-up module says item-started, then item-ended with its outcome and `at`; a +// cache item says nothing; a failed backup ends its items failed with why; the holder reads no restore for a module that does +// not hold the seat mesh-broker. +func TestEveryItemOfABackedUpModuleSaysItsStartAndItsEnd(t *testing.T) { + bus := &itemBusFake{} + b := itemsBackups(t, bus, "somebody-else", nil) + if n := b.one(context.Background(), declaredOf(t, b, "postgres"), false); !n.OK { + t.Fatalf("%+v", n) + } + want := []string{"item-started postgres.store", "item-started postgres.dumps", "item-ended postgres.store done", "item-ended postgres.dumps done"} + if !reflect.DeepEqual(bus.log, want) { + t.Fatalf("said %q", bus.log) + } + bus.log = nil + b = itemsBackups(t, bus, "somebody-else", errors.New("disk full")) + if n := b.one(context.Background(), declaredOf(t, b, "postgres"), false); n.OK { + t.Fatal("a failed backup was ok") + } + if len(bus.log) != 4 || bus.log[2] != "item-ended postgres.store failed error=disk full" { + t.Fatalf("said %q", bus.log) + } +} + +// Design 47 G §563 (BKP5): for the holder of mesh-broker the holder says its start, then reads the recovery module's restores; +// a restore that runs puts the items off naming its act, waits for that act's restored, starts again, and reads once more. +func TestTheBackupOfTheBusGivesWayToARestoreThatRuns(t *testing.T) { + bus := &itemBusFake{restoring: []string{"act-7"}} + b := itemsBackups(t, bus, "postgres", nil) + if n := b.one(context.Background(), declaredOf(t, b, "postgres"), false); !n.OK { + t.Fatalf("%+v", n) + } + want := []string{ + "item-started postgres.store", "item-started postgres.dumps", "read restores of postgres", + "item-ended postgres.store put-off restore=act-7", "item-ended postgres.dumps put-off restore=act-7", + "restored act-7", + "item-started postgres.store", "item-started postgres.dumps", "read restores of postgres", + "item-ended postgres.store done", "item-ended postgres.dumps done"} + if !reflect.DeepEqual(bus.log, want) { + t.Fatalf("said %q", bus.log) + } +} + +// Design 47 G §563 (BKP4): at its start the holder ends every item it said started and never ended, failed with the error stopped. +func TestItemsLeftStartedAreEndedFailedStoppedAtTheStart(t *testing.T) { + bus := &itemBusFake{unended: []ItemRef{{"postgres", "store"}}} + b := itemsBackups(t, bus, "postgres", nil) + must(t, b.StopStale()) + if !reflect.DeepEqual(bus.log, []string{"item-ended postgres.store failed error=stopped"}) { + t.Fatalf("said %q", bus.log) + } +} diff --git a/modules/restic/cmd/restic-backups/main.go b/modules/restic/cmd/restic-backups/main.go index 14579bd33..d8a7244a9 100644 --- a/modules/restic/cmd/restic-backups/main.go +++ b/modules/restic/cmd/restic-backups/main.go @@ -47,6 +47,22 @@ func main() { if len(os.Args) > 1 { way = os.Args[1] } + // The night and the tools' runtime each say their items on the bus as the module's own account, and read the recovery + // module's restores (design 47 G §563, A-150); a process given no credential says none. + if way == "night" || way == "" { + bus, broker, closeBus, err := connectItems(os.Getenv) + if err != nil { + say("the bus cannot be reached with the module's own credential: %v", err) + os.Exit(1) + } + defer closeBus() + b.Bus, b.BrokerModule = bus, broker + // Every item this node said started and never ended is ended failed, stopped (§563). + if err := b.StopStale(); err != nil { + say("%v", err) + os.Exit(1) + } + } if way == "" { b.Emit = func(event string, body any) error { return stdio.Emit(event, body) } b.Readings = busReadings(b.Node) diff --git a/modules/restic/go.mod b/modules/restic/go.mod index 84ee5a573..b764ec582 100644 --- a/modules/restic/go.mod +++ b/modules/restic/go.mod @@ -6,7 +6,7 @@ require git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000 require ( github.com/klauspost/compress v1.20.1 // indirect - github.com/nats-io/nats.go v1.54.0 // indirect + github.com/nats-io/nats.go v1.54.0 github.com/nats-io/nkeys v0.4.16 // indirect github.com/nats-io/nuid v1.0.1 // indirect golang.org/x/crypto v0.57.0 // indirect diff --git a/modules/restic/module.json b/modules/restic/module.json index 462ff7aec..671b2f9c1 100644 --- a/modules/restic/module.json +++ b/modules/restic/module.json @@ -16,6 +16,10 @@ ] } ], + "consumes": [ + "mesh-recovery.restore-started", + "mesh-recovery.restored" + ], "emits": [ "backed-up" ], @@ -25,7 +29,8 @@ "backup_deleted" ], "own-secrets": { - "repository": "${dir:state}/repository.secret" + "repository": "${dir:state}/repository.secret", + "broker": "${dir:state}/broker" }, "data": { "own": [ @@ -139,7 +144,8 @@ "MESH_BACKUP_REPOSITORY": "${dir:repository}", "MESH_BACKUP_PASSWORD_FILE": "${dir:state}/repository.secret", "MESH_BACKUP_STATE": "${dir:state}", - "MESH_BACKUP_DATA": "${dir:state}/data.conf" + "MESH_BACKUP_DATA": "${dir:state}/data.conf", + "MESH_BROKER_FILE": "${dir:state}/broker" }, "health": { "kind": "unit" @@ -160,7 +166,8 @@ "MESH_BACKUP_REPOSITORY": "${dir:repository}", "MESH_BACKUP_PASSWORD_FILE": "${dir:state}/repository.secret", "MESH_BACKUP_STATE": "${dir:state}", - "MESH_BACKUP_DATA": "${dir:state}/data.conf" + "MESH_BACKUP_DATA": "${dir:state}/data.conf", + "MESH_BROKER_FILE": "${dir:state}/broker" }, "health": { "kind": "tool", @@ -189,7 +196,8 @@ "MESH_BACKUP_REPOSITORY": "${dir:repository}", "MESH_BACKUP_PASSWORD_FILE": "${dir:state}/repository.secret", "MESH_BACKUP_STATE": "${dir:state}", - "MESH_BACKUP_DATA": "${dir:state}/data.conf" + "MESH_BACKUP_DATA": "${dir:state}/data.conf", + "MESH_BROKER_FILE": "${dir:state}/broker" } }, {