BKP4, BKP5: the backup holder says each item's start and end and gives way to a restore of the recovery module

Builds: design 47 G §554, §556, §563, A-150
This commit is contained in:
2026-11-12 05:05:54 +01:00
parent f8160922c9
commit d708998a53
7 changed files with 576 additions and 7 deletions
+19 -2
View File
@@ -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)
@@ -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]
}
}
}
+186
View File
@@ -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
})
}
@@ -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)
}
}
+16
View File
@@ -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)
+1 -1
View File
@@ -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
+12 -4
View File
@@ -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"
}
},
{