diff --git a/modules/restic/cmd/restic-backups/backups.go b/modules/restic/cmd/restic-backups/backups.go index 1ac10e8bf..14cbfd915 100644 --- a/modules/restic/cmd/restic-backups/backups.go +++ b/modules/restic/cmd/restic-backups/backups.go @@ -212,6 +212,11 @@ type Backups struct { // 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) + // PutOffs is the module's own state `put-off`, whose keys are the items put off for a restore; PutOffBound is the setting + // put-off-bound; Findings is the finding writer of the night's process (putoff.go). Each nil where the process has none. + PutOffs PutOffState + PutOffBound time.Duration + Findings Findings // 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) diff --git a/modules/restic/cmd/restic-backups/itembus.go b/modules/restic/cmd/restic-backups/itembus.go index 6961f2689..316a4b071 100644 --- a/modules/restic/cmd/restic-backups/itembus.go +++ b/modules/restic/cmd/restic-backups/itembus.go @@ -6,14 +6,20 @@ package main import ( "context" "encoding/json" + "errors" "fmt" "strings" + "sync/atomic" "time" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/keys" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/seatlist" + "git.novox.be/novox/mesh-sdk/go/views" + "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" ) @@ -38,41 +44,65 @@ func itemEventSubject(event, node, module, item string) string { // liveBus is ItemBus over the module's own account. type liveBus struct { - conn *process.Conn - js jetstream.JetStream - node string + conn *process.Conn + nc *nats.Conn + js jetstream.JetStream + node string + account process.Account } -// 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) { +// connectItems opens the bus as the module's own account when the process was given its credential (MESH_BROKER_FILE); nil, nil +// when it was not. +func connectItems(getenv func(string) string) (*liveBus, error) { if strings.TrimSpace(getenv(process.BrokerFileEnv)) == "" { - return nil, nil, func() {}, nil + return nil, nil } account, err := process.ReadBrokerAccount(getenv) if err != nil { - return nil, nil, nil, err + return nil, err } conn, err := process.Connect(account) if err != nil { - return nil, nil, nil, err + return nil, err } nc, js, err := account.Dial() if err != nil { conn.Close() - return nil, nil, nil, err + return 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 + return &liveBus{conn: conn, nc: nc, js: js, node: account.Node, account: account}, nil } -func (l liveBus) Say(event string, ref ItemRef, body map[string]any) error { +// Broker answers how to find the holder of the seat mesh-broker. +func (l *liveBus) Broker() (string, error) { return seatlist.HolderOf(l.conn.Call, BrokerSeat) } + +// Close ends both connections. +func (l *liveBus) Close() { l.nc.Close(); l.conn.Close() } + +// PutOffs is the module's own state `put-off`, its keys with their own time to live (A-155). +func (l *liveBus) PutOffs() PutOffState { return stateOfPutOffs{l.conn.State(PutOffStateName)} } + +// 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) { + 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}) + if err != nil { + return nil, nil, err + } + if err := w.Start(ctx); err != nil { + return nil, nil, err + } + return w, w.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 { +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 { @@ -105,7 +135,7 @@ func atOf(data []byte) time.Time { return b.At } -func (l liveBus) Unended() ([]ItemRef, error) { +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{} @@ -141,7 +171,7 @@ func (l liveBus) Unended() ([]ItemRef, error) { return out, nil } -func (l liveBus) Restoring(module string) (string, error) { +func (l *liveBus) Restoring(module string) (string, error) { ctx, cancel := context.WithTimeout(context.Background(), 2*readWait) defer cancel() type act struct { @@ -183,7 +213,7 @@ func (l liveBus) Restoring(module string) (string, error) { return running[0], nil } -func (l liveBus) WaitRestored(ctx context.Context, want string) error { +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 { @@ -213,3 +243,66 @@ func sortStrings(s []string) { } } } + +// PutOffStateName is the module's own state of the items put off for a restore, whose keys have a time to live (key-ttl). +const PutOffStateName = "put-off" + +// putOffBucket is the state's calls this module makes on its bucket. +type putOffBucket interface { + CreateTTL(key string, value any, ttl time.Duration) (uint64, error) + Watch(key string, handle func(stdio.StateChange) error) error + Delete(key string) error + Keys() ([]string, error) + Get(key string) (*stdio.StateEntry, error) +} + +// errBoundEnded ends a bound key's watch once the key is gone. +var errBoundEnded = errors.New("the bound key is removed") + +// stateOfPutOffs times the wait for a restore by a bound key of the state put-off (A-77, A-155): the key is watched before it is +// written, so its expiry is never missed, and the watch ends with the key's removal. +type stateOfPutOffs struct{ s putOffBucket } + +func (p stateOfPutOffs) Hold(key string, rec PutOff, ttl time.Duration, expired func()) (func(), error) { + var ended atomic.Bool + if err := p.s.Watch(key, func(c stdio.StateChange) error { + if c.Op == "put" || c.Current { + return nil + } + if !ended.Swap(true) { + expired() + } + return errBoundEnded + }); err != nil { + return nil, err + } + if _, err := p.s.CreateTTL(key, rec, ttl); err != nil { + ended.Store(true) + _ = p.s.Delete(key) + return nil, err + } + return func() { + if !ended.Swap(true) { + _ = p.s.Delete(key) + } + }, nil +} + +func (p stateOfPutOffs) All() (map[string]PutOff, error) { + names, err := p.s.Keys() + if err != nil { + return nil, err + } + out := map[string]PutOff{} + for _, k := range names { + e, err := p.s.Get(k) + if err != nil || e == nil { + continue + } + var rec PutOff + if json.Unmarshal(e.Value, &rec) == nil { + out[k] = rec + } + } + return out, nil +} diff --git a/modules/restic/cmd/restic-backups/items.go b/modules/restic/cmd/restic-backups/items.go index 23c0e059b..cbe87f2a1 100644 --- a/modules/restic/cmd/restic-backups/items.go +++ b/modules/restic/cmd/restic-backups/items.go @@ -11,6 +11,7 @@ import ( "fmt" "sort" "strings" + "sync" ) // The seat node-backup's events of an item, and the seat of the holder whose items are the bus's. @@ -142,6 +143,7 @@ func (b *Backups) beginItems(ctx context.Context, module string) (func(err error if err := started(); err != nil { return nothing, fmt.Errorf("the start of %s's items could not be said: %w", module, err) } + b.clearWaits(items) if b.BrokerModule == nil { return finish, nil } @@ -166,15 +168,36 @@ func (b *Backups) beginItems(ctx context.Context, module string) (func(err error 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) } - b.keepPutOff(items, act) - if err := b.Bus.WaitRestored(ctx, act); err != nil { - b.clearPutOff(module) - return nothing, fmt.Errorf("the end of the restore %s could not be waited for: %w", act, err) + wctx, stop := context.WithCancel(ctx) + var expiredMu sync.Mutex + expired := false + release, err := b.keepPutOff(items, act, func(ref ItemRef) { + expiredMu.Lock() + expired = true + expiredMu.Unlock() + b.writeWaits(ref, act) + stop() + }) + if err != nil { + stop() + return nothing, err + } + waitErr := b.Bus.WaitRestored(wctx, act) + release() + stop() + expiredMu.Lock() + ran := expired + expiredMu.Unlock() + if ran { + return nothing, fmt.Errorf("%w: the restore %s", errPutOffExpired, act) + } + if waitErr != nil { + return nothing, fmt.Errorf("the end of the restore %s could not be waited for: %w", act, waitErr) } - b.clearPutOff(module) if err := started(); err != nil { return nothing, fmt.Errorf("the start of %s's items could not be said: %w", module, err) } + b.clearWaits(items) } } diff --git a/modules/restic/cmd/restic-backups/items_test.go b/modules/restic/cmd/restic-backups/items_test.go index f5eb5d14a..aabae9fb1 100644 --- a/modules/restic/cmd/restic-backups/items_test.go +++ b/modules/restic/cmd/restic-backups/items_test.go @@ -4,9 +4,14 @@ import ( "context" "errors" "fmt" + "os" + "path/filepath" "reflect" + "strings" "testing" "time" + + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" ) // itemBusFake records every call, in order, as the backup holder makes them. @@ -15,8 +20,9 @@ type itemBusFake struct { restoring []string // what each read of the recovery module's restores answers, in turn unended []ItemRef runFailure error - // waiting runs while the holder waits for a restore's restored. + // waiting runs while the holder waits for a restore's restored; hold makes the wait last until its context ends. waiting func() + hold bool } func (f *itemBusFake) Say(event string, ref ItemRef, body map[string]any) error { @@ -46,14 +52,62 @@ func (f *itemBusFake) Restoring(module string) (string, error) { f.restoring = f.restoring[1:] return act, nil } -func (f *itemBusFake) WaitRestored(_ context.Context, act string) error { +func (f *itemBusFake) WaitRestored(ctx context.Context, act string) error { f.log = append(f.log, "restored "+act) if f.waiting != nil { f.waiting() } + if f.hold { + <-ctx.Done() + return ctx.Err() + } return nil } +// putOffFake is the state put-off: its keys with the time to live each was written with; expire plays the bus removing one. +type putOffFake struct { + keys map[string]PutOff + ttl map[string]time.Duration + expired map[string]func() +} + +func newPutOffFake() *putOffFake { + return &putOffFake{keys: map[string]PutOff{}, ttl: map[string]time.Duration{}, expired: map[string]func(){}} +} + +func (f *putOffFake) Hold(key string, rec PutOff, ttl time.Duration, expired func()) (func(), error) { + f.keys[key], f.ttl[key], f.expired[key] = rec, ttl, expired + return func() { delete(f.keys, key); delete(f.expired, key) }, nil +} +func (f *putOffFake) All() (map[string]PutOff, error) { return f.keys, nil } +func (f *putOffFake) expire(key string) { + cb := f.expired[key] + delete(f.keys, key) + delete(f.expired, key) + cb() +} + +// findingsFake holds the findings written and not cleared. +type findingsFake struct { + held map[string]meshconditions.Finding +} + +func (f *findingsFake) Write(_ context.Context, x meshconditions.Finding) error { + f.held[x.Scope+"."+x.ID+"."+x.Kind] = x + return nil +} +func (f *findingsFake) Clear(_ context.Context, scope, id, kind, _ string) error { + delete(f.held, scope+"."+id+"."+kind) + return nil +} +func (f *findingsFake) Held() []meshconditions.Finding { + var out []meshconditions.Finding + for _, x := range f.held { + out = append(out, x) + } + return out +} + 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) }, @@ -107,10 +161,15 @@ func TestEveryItemOfABackedUpModuleSaysItsStartAndItsEnd(t *testing.T) { func TestTheBackupOfTheBusGivesWayToARestoreThatRuns(t *testing.T) { bus := &itemBusFake{restoring: []string{"act-7"}} b := itemsBackups(t, bus, "postgres", nil) + state := newPutOffFake() + b.PutOffs, b.PutOffBound = state, 45*time.Minute // While it waits, `backed-up` lists the item put off with the restore's act and since when (BKP8, §302, §664). bus.waiting = func() { got, err := b.BackedUp(context.Background(), "postgres") must(t, err) + if rec, ok := state.keys["postgres.store"]; !ok || rec.Restore != "act-7" || state.ttl["postgres.store"] != 45*time.Minute { + t.Errorf("the put-off is not a key of the state put-off with the bound as its time to live: %+v %v", state.keys, state.ttl) + } var store *ItemReport for i := range got[0].Data { if got[0].Data[i].ID == "store" { @@ -142,6 +201,52 @@ func TestTheBackupOfTheBusGivesWayToARestoreThatRuns(t *testing.T) { } } +// Design 47 G §561, §563, A-155 (BKP6): a wait for a restore past put-off-bound, the key of the state put-off expiring, writes +// the finding backup-waits for the item (scope the node), ends the wait (the backup fails, naming the bound) and holds the night's +// lock no longer; the finding is cleared when the item starts again. +func TestAWaitPastItsBoundWritesBackupWaitsAndEndsTheWait(t *testing.T) { + bus := &itemBusFake{restoring: []string{"act-7"}, hold: true} + b := itemsBackups(t, bus, "postgres", nil) + state, findings := newPutOffFake(), &findingsFake{held: map[string]meshconditions.Finding{}} + b.PutOffs, b.Findings = state, findings + bus.waiting = func() { state.expire("postgres.dumps") } + n := b.one(context.Background(), declaredOf(t, b, "postgres"), false) + if n.OK || !strings.Contains(n.Error, "put-off-bound") || !strings.Contains(n.Error, "act-7") { + t.Fatalf("the night %+v", n) + } + f, ok := findings.held["ace.postgres=2edumps.backup-waits"] + if !ok || len(findings.held) != 1 || f.Scope != "ace" || f.Kind != FindingBackupWaits || !strings.Contains(f.Summary, "act-7") { + t.Fatalf("the findings held: %+v", findings.held) + } + if len(state.keys) != 0 { + t.Errorf("the keys left: %+v", state.keys) + } + // The next night: the item starts again and the finding is cleared. + bus.restoring, bus.hold, bus.waiting = nil, false, nil + if n := b.one(context.Background(), declaredOf(t, b, "postgres"), false); !n.OK || len(findings.held) != 0 { + t.Fatalf("the night %+v, findings %+v", n, findings.held) + } +} + +// Design 47 G §561 (A-155): the setting put-off-bound is minutes in the file the manifest places; sixty when none is given. +func TestPutOffBoundReadsItsSettingFile(t *testing.T) { + file := filepath.Join(t.TempDir(), "put-off-bound") + for text, want := range map[string]time.Duration{"15\n": 15 * time.Minute, "": DefaultPutOffBound} { + must(t, os.WriteFile(file, []byte(text), 0o600)) + got, err := putOffBoundFrom(func(string) string { return file }) + if err != nil || got != want { + t.Errorf("%q gave %v, %v", text, got, err) + } + } + if got, _ := putOffBoundFrom(func(string) string { return "" }); got != DefaultPutOffBound { + t.Errorf("no file gave %v", got) + } + must(t, os.WriteFile(file, []byte("soon"), 0o600)) + if _, err := putOffBoundFrom(func(string) string { return file }); err == nil { + t.Error("a bound that is no number was taken") + } +} + // 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"}}} diff --git a/modules/restic/cmd/restic-backups/live_integration_test.go b/modules/restic/cmd/restic-backups/live_integration_test.go new file mode 100644 index 000000000..6044481e2 --- /dev/null +++ b/modules/restic/cmd/restic-backups/live_integration_test.go @@ -0,0 +1,76 @@ +//go:build integration + +package main + +// The module's live path on a bus of this test alone (design 47 G §561, §563, A-150, A-155): the night's own broker account. + +import ( + "context" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/testbus" +) + +// A put-off is a key of the state put-off with a time to live of its own: it is listed while it lasts, and the bus removing it +// at the end of its time is heard as the expiry, once; a key released by the holder is heard as nothing. +func TestAPutOffKeyIsListedAndItsExpiryIsHeardFromTheBus(t *testing.T) { + url := testbus.URL(t) + nc, err := nats.Connect(url) + if err != nil { + t.Fatal(err) + } + defer nc.Close() + js, _ := jetstream.New(nc) + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "restic_" + PutOffStateName, LimitMarkerTTL: time.Second}); err != nil { + t.Fatal(err) + } + conn, err := process.Connect(process.Account{URL: url, Node: "ace", Module: "restic", User: "ace.restic", Password: "unused"}) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + st := stateOfPutOffs{conn.State(PutOffStateName)} + now := time.Date(2026, 11, 12, 3, 0, 0, 0, time.UTC) + expired := make(chan string, 4) + if _, err := st.Hold("postgres.store", PutOff{Restore: "act-7", Since: now}, 2*time.Second, func() { expired <- "store" }); err != nil { + t.Fatal(err) + } + release, err := st.Hold("postgres.dumps", PutOff{Restore: "act-7", Since: now}, time.Hour, func() { expired <- "dumps" }) + if err != nil { + t.Fatal(err) + } + all, err := st.All() + if err != nil || len(all) != 2 || all["postgres.store"].Restore != "act-7" || !all["postgres.store"].Since.Equal(now) { + t.Fatalf("the keys %+v, %v", all, err) + } + select { + case who := <-expired: + if who != "store" { + t.Fatalf("%s expired", who) + } + case <-time.After(30 * time.Second): + t.Fatal("the bus did not remove the key at the end of its time") + } + release() + for { + all, err = st.All() + if err != nil { + t.Fatal(err) + } + if len(all) == 0 { + break + } + select { + case who := <-expired: + t.Fatalf("%s expired after its release", who) + case <-time.After(100 * time.Millisecond): + } + } +} diff --git a/modules/restic/cmd/restic-backups/main.go b/modules/restic/cmd/restic-backups/main.go index d8a7244a9..a117dd6ab 100644 --- a/modules/restic/cmd/restic-backups/main.go +++ b/modules/restic/cmd/restic-backups/main.go @@ -50,13 +50,30 @@ func main() { // 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) + live, 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 + if live != nil { + defer live.Close() + b.Bus, b.BrokerModule, b.PutOffs = live, live.Broker, live.PutOffs() + bound, err := putOffBoundFrom(os.Getenv) + if err != nil { + say("%v", err) + os.Exit(1) + } + b.PutOffBound = bound + if way == "night" { + findings, closeFindings, err := live.Findings(context.Background(), say) + if err != nil { + say("the finding writer cannot be started: %v", err) + os.Exit(1) + } + defer closeFindings() + b.Findings = findings + } + } // Every item this node said started and never ended is ended failed, stopped (§563). if err := b.StopStale(); err != nil { say("%v", err) diff --git a/modules/restic/cmd/restic-backups/putoff.go b/modules/restic/cmd/restic-backups/putoff.go index 13b3d1b67..776b617c2 100644 --- a/modules/restic/cmd/restic-backups/putoff.go +++ b/modules/restic/cmd/restic-backups/putoff.go @@ -1,13 +1,21 @@ package main -// An item put off for a restore (design 47 G §563, §302, §664): `backed-up` lists it with the restore's act and since when. The -// night's process keeps it in the module's state directory while the item waits, and the tools' runtime reads it there. +// An item put off for a restore (design 47 G §563, §302, §664, A-155): `backed-up` lists it with the restore's act and since +// when. The wait is a bound key `.` in the module's own state `put-off`, with a time to live of its own (the setting +// put-off-bound): the key is the record `backed-up` reads, its expiry is heard as the bus's own change, raises the finding +// `backup-waits` and ends the wait. No file and no timer keeps it. import ( - "encoding/json" + "context" + "errors" + "fmt" "os" - "path/filepath" + "strconv" + "strings" "time" + + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + "git.novox.be/novox/mesh-sdk/go/keys" ) // PutOff says an item gave way to a restore: the act of that restore and since when. @@ -16,69 +24,162 @@ type PutOff struct { Since time.Time `json:"since"` } -type putOffItem struct { - Module string `json:"module"` - Item string `json:"item"` - PutOff +// FindingBackupWaits is the backup holder's finding of an item put off past its bound (§561). +const FindingBackupWaits = "backup-waits" + +// DefaultPutOffBound is the setting put-off-bound's default (§561, §557): sixty minutes. +const DefaultPutOffBound = 60 * time.Minute + +// PutOffState is the module's own state `put-off`, whose keys live a time of their own. +type PutOffState interface { + // Hold writes the key with its record and time to live, watched before it is written; expired is called once when the bus + // removes the key at the end of its time. release removes the key and ends the watch, calling nothing. + Hold(key string, rec PutOff, ttl time.Duration, expired func()) (release func(), err error) + // All reads every key with its record. + All() (map[string]PutOff, error) } -func (b *Backups) putOffFile() string { return filepath.Join(b.Where.State, "put-off.json") } +// Findings is the backup holder's finding writer (mesh-sdk's held writer). +type Findings interface { + Write(ctx context.Context, f meshconditions.Finding) error + Clear(ctx context.Context, scope, id, kind, why string) error + Held() []meshconditions.Finding +} -func (b *Backups) readPutOff() []putOffItem { - raw, err := os.ReadFile(b.putOffFile()) - if err != nil { - return nil +func putOffKey(ref ItemRef) string { + return keys.EscapeName(ref.Module) + "." + keys.EscapeName(ref.Item) +} + +func putOffRef(key string) (ItemRef, bool) { + m, i, ok := strings.Cut(key, ".") + if !ok { + return ItemRef{}, false } - var all []putOffItem - _ = json.Unmarshal(raw, &all) - return all + module, e1 := keys.UnescapeName(m) + item, e2 := keys.UnescapeName(i) + return ItemRef{Module: module, Item: item}, e1 == nil && e2 == nil } -func (b *Backups) writePutOff(all []putOffItem) { - if len(all) == 0 { - if err := os.Remove(b.putOffFile()); err != nil && !os.IsNotExist(err) { - b.Say("the items put off could not be cleared: %v", err) +func (b *Backups) putOffBound() time.Duration { + if b.PutOffBound > 0 { + return b.PutOffBound + } + return DefaultPutOffBound +} + +// errPutOffExpired ends the wait for a restore past the bound. +var errPutOffExpired = errors.New("the restore did not end within the bound put-off-bound") + +// keepPutOff records a module's items as put off for a restore, since now, each a bound key; expired is called once when the +// first of them is removed by the bus at the end of its time. It answers how to let the keys go. +func (b *Backups) keepPutOff(items []ItemRef, act string, expired func(ItemRef)) (func(), error) { + if b.PutOffs == nil { + return func() {}, nil + } + var releases []func() + release := func() { + for _, r := range releases { + r() } + } + for _, ref := range items { + ref := ref + r, err := b.PutOffs.Hold(putOffKey(ref), PutOff{Restore: act, Since: b.Now().UTC()}, b.putOffBound(), func() { expired(ref) }) + if err != nil { + release() + return nil, fmt.Errorf("the put-off of %s.%s could not be kept: %w", ref.Module, ref.Item, err) + } + releases = append(releases, r) + } + return release, nil +} + +// writeWaits says the finding backup-waits for an item put off past its bound, naming the restore's act (§561). +func (b *Backups) writeWaits(ref ItemRef, act string) { + if b.Findings == nil { return } - raw, _ := json.MarshalIndent(all, "", " ") - if err := os.WriteFile(b.putOffFile(), append(raw, '\n'), 0o600); err != nil { - b.Say("the items put off could not be recorded: %v", err) + f := meshconditions.Finding{Scope: b.Node, ID: keys.EscapeName(ref.Module + "." + ref.Item), Kind: FindingBackupWaits, + Subject: meshconditions.Subject{Scope: "module", ID: "restic"}, + Severity: meshconditions.FindingSeverityWarning, Headline: "A backup waits on a restore", + Summary: fmt.Sprintf("the backup of %s.%s was put off for the restore %s of the recovery module and that restore has not ended within %s", ref.Module, ref.Item, act, b.putOffBound()), + Explanation: fmt.Sprintf("the item %s.%s gave way to the restore %s at its start and waits for its restored", ref.Module, ref.Item, act), + Needs: "the restore ended or cancelled in the recovery module (recovery_cancel); the backup then starts again at the next night", + Source: "backups", Observer: "restic", At: b.Now().UTC()} + if err := b.Findings.Write(context.Background(), f); err != nil { + b.Say("the finding %s of %s.%s could not be written: %v", FindingBackupWaits, ref.Module, ref.Item, err) } } -// keepPutOff records a module's items as put off for a restore, since now. -func (b *Backups) keepPutOff(items []ItemRef, act string) { - all := b.readPutOff() - for _, ref := range items { - all = append(all, putOffItem{Module: ref.Module, Item: ref.Item, PutOff: PutOff{Restore: act, Since: b.Now().UTC()}}) +// clearWaits clears the finding backup-waits of the items that started again. +func (b *Backups) clearWaits(items []ItemRef) { + if b.Findings == nil { + return } - b.writePutOff(all) -} - -// clearPutOff forgets a module's items put off: they started again. -func (b *Backups) clearPutOff(module string) { - var keep []putOffItem - for _, p := range b.readPutOff() { - if p.Module != module { - keep = append(keep, p) + held := map[string]bool{} + for _, f := range b.Findings.Held() { + if f.Kind == FindingBackupWaits { + held[f.Scope+"\x00"+f.ID] = true + } + } + for _, ref := range items { + id := keys.EscapeName(ref.Module + "." + ref.Item) + if !held[b.Node+"\x00"+id] { + continue + } + if err := b.Findings.Clear(context.Background(), b.Node, id, FindingBackupWaits, "the item started again"); err != nil { + b.Say("the finding %s of %s.%s could not be cleared: %v", FindingBackupWaits, ref.Module, ref.Item, err) } } - b.writePutOff(keep) } -// withPutOff says on a module's item reports which items are put off. +// withPutOff says on a module's item reports which items are put off, read from the state's keys. func (b *Backups) withPutOff(module string, reports []ItemReport) []ItemReport { - for _, p := range b.readPutOff() { - if p.Module != module { + if b.PutOffs == nil { + return reports + } + all, err := b.PutOffs.All() + if err != nil { + b.Say("the items put off could not be read: %v", err) + return reports + } + for key, rec := range all { + ref, ok := putOffRef(key) + if !ok || ref.Module != module { continue } for i := range reports { - if reports[i].ID == p.Item { - put := p.PutOff + if reports[i].ID == ref.Item { + put := rec reports[i].PutOff = &put } } } return reports } + +// PutOffBoundEnv names the file of the setting put-off-bound, in minutes: the time the wait for a restore may last before +// backup-waits is written and the wait ends (§561, §557). +const PutOffBoundEnv = "MESH_BACKUP_PUT_OFF_BOUND_FILE" + +// putOffBoundFrom reads the setting from its file; it is the default (sixty minutes) when the process is given no file or an +// empty one, and refused when it does not read as a number of minutes above zero. +func putOffBoundFrom(getenv func(string) string) (time.Duration, error) { + path := strings.TrimSpace(getenv(PutOffBoundEnv)) + if path == "" { + return DefaultPutOffBound, nil + } + raw, err := os.ReadFile(path) + if err != nil { + return 0, fmt.Errorf("the setting put-off-bound cannot be read: %w", err) + } + text := strings.TrimSpace(string(raw)) + if text == "" { + return DefaultPutOffBound, nil + } + minutes, err := strconv.ParseFloat(text, 64) + if err != nil || minutes <= 0 { + return 0, fmt.Errorf("the setting put-off-bound is %q, not a number of minutes above zero", text) + } + return time.Duration(minutes * float64(time.Minute)), nil +} diff --git a/modules/restic/cmd/restic-backups/templates_test.go b/modules/restic/cmd/restic-backups/templates_test.go index 7d0718b57..e0c416421 100644 --- a/modules/restic/cmd/restic-backups/templates_test.go +++ b/modules/restic/cmd/restic-backups/templates_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "os" "path/filepath" + "strings" "testing" ) @@ -62,3 +63,58 @@ func TestEveryFileTakesItsTextFromATemplate(t *testing.T) { t.Errorf("the build packs files/ into %d archives, not one", packs) } } + +// Design 47 G §561, A-155: the put-off wait is a bound key of a state of the module's own with a key-ttl, the finding backup-waits +// goes through the findings template and its events, the setting put-off-bound is a file the night reads, and the module hears +// the findings asked. +func TestThePutOffBoundAndItsStatesAreInTheManifest(t *testing.T) { + raw, err := os.ReadFile(filepath.Join(moduleDir, "module.json")) + if err != nil { + t.Fatal(err) + } + var m struct { + Consumes []string `json:"consumes"` + Requires []struct { + Name, Template string + Options map[string]any + } `json:"requires"` + Config map[string]map[string]any `json:"config"` + Resources []map[string]any `json:"resources"` + } + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + got := map[string]bool{} + for _, r := range m.Requires { + got[r.Name+"/"+r.Template] = true + if r.Name == "put-off" && r.Options["key-ttl"] != true { + t.Errorf("the state put-off has no key-ttl: %v", r.Options) + } + } + for _, want := range []string{"put-off/state", "findings/state", "finding-events/events"} { + if !got[want] { + t.Errorf("the manifest requires no %s: %v", want, got) + } + } + if m.Config["put-off-bound"]["default"] != float64(60) { + t.Errorf("the setting put-off-bound is %v", m.Config["put-off-bound"]) + } + heard := false + for _, c := range m.Consumes { + heard = heard || c == "mesh-conditions.findings-asked" + } + if !heard { + t.Error("the module does not hear mesh-conditions.findings-asked") + } + for _, r := range m.Resources { + if r["id"] == "night" { + env, _ := r["env"].(map[string]any) + if env[PutOffBoundEnv] != "${dir:state}/put-off-bound" { + t.Errorf("the night is not given the file of put-off-bound: %v", env) + } + } + if r["id"] == "put-off-bound" && strings.TrimSpace(templateText(t, r)) != "${config:put-off-bound}" { + t.Errorf("the file of put-off-bound does not hold its setting") + } + } +} diff --git a/modules/restic/files/put-off-bound b/modules/restic/files/put-off-bound new file mode 100644 index 000000000..c33d7f57e --- /dev/null +++ b/modules/restic/files/put-off-bound @@ -0,0 +1 @@ +${config:put-off-bound} diff --git a/modules/restic/go.mod b/modules/restic/go.mod index b764ec582..fe82d5eb6 100644 --- a/modules/restic/go.mod +++ b/modules/restic/go.mod @@ -2,7 +2,10 @@ module restic go 1.26.0 -require git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000 +require ( + git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000 +) require ( github.com/klauspost/compress v1.20.1 // indirect diff --git a/modules/restic/module.json b/modules/restic/module.json index 671b2f9c1..6482eb39e 100644 --- a/modules/restic/module.json +++ b/modules/restic/module.json @@ -18,7 +18,8 @@ ], "consumes": [ "mesh-recovery.restore-started", - "mesh-recovery.restored" + "mesh-recovery.restored", + "mesh-conditions.findings-asked" ], "emits": [ "backed-up" @@ -81,6 +82,14 @@ "maximum": 9999, "trusted": true, "default": 7 + }, + "put-off-bound": { + "type": "number", + "description": "minutes an item put off for a restore of the recovery module is waited for before the finding backup-waits is written and the wait ends, so the night never holds its lock without a bound; timed by a bound key of the state put-off (design 47 G sections 561, 557, A-155).", + "minimum": 1, + "maximum": 10080, + "trusted": false, + "default": 60 } }, "resources": [ @@ -119,6 +128,13 @@ "trusted": true, "template": "files/retention.conf" }, + { + "id": "put-off-bound", + "type": "file", + "path": "${dir:state}/put-off-bound", + "mode": "0600", + "template": "files/put-off-bound" + }, { "id": "tool", "type": "package", @@ -145,7 +161,8 @@ "MESH_BACKUP_PASSWORD_FILE": "${dir:state}/repository.secret", "MESH_BACKUP_STATE": "${dir:state}", "MESH_BACKUP_DATA": "${dir:state}/data.conf", - "MESH_BROKER_FILE": "${dir:state}/broker" + "MESH_BROKER_FILE": "${dir:state}/broker", + "MESH_BACKUP_PUT_OFF_BOUND_FILE": "${dir:state}/put-off-bound" }, "health": { "kind": "unit" @@ -167,7 +184,8 @@ "MESH_BACKUP_PASSWORD_FILE": "${dir:state}/repository.secret", "MESH_BACKUP_STATE": "${dir:state}", "MESH_BACKUP_DATA": "${dir:state}/data.conf", - "MESH_BROKER_FILE": "${dir:state}/broker" + "MESH_BROKER_FILE": "${dir:state}/broker", + "MESH_BACKUP_PUT_OFF_BOUND_FILE": "${dir:state}/put-off-bound" }, "health": { "kind": "tool", @@ -197,7 +215,8 @@ "MESH_BACKUP_PASSWORD_FILE": "${dir:state}/repository.secret", "MESH_BACKUP_STATE": "${dir:state}", "MESH_BACKUP_DATA": "${dir:state}/data.conf", - "MESH_BROKER_FILE": "${dir:state}/broker" + "MESH_BROKER_FILE": "${dir:state}/broker", + "MESH_BACKUP_PUT_OFF_BOUND_FILE": "${dir:state}/put-off-bound" } }, { @@ -212,5 +231,34 @@ ], "state": [ "good" + ], + "requires": [ + { + "name": "put-off", + "storage": "key-value", + "template": "state", + "options": { + "key-ttl": true + } + }, + { + "name": "findings", + "storage": "key-value", + "template": "state" + }, + { + "name": "finding-events", + "storage": "stream", + "template": "events", + "options": { + "events": [ + "finding-written.*", + "finding-cleared.*", + "findings-stated.*" + ], + "max-age-days": 7, + "max-mib": 16 + } + } ] }