BKP6: the put-off wait for a restore is a bound key of restic's state put-off; its expiry raises backup-waits and ends the wait
The record backed-up reads is the key `<module>.<item>` of the module's own state put-off (key-ttl), bounded by the setting put-off-bound (minutes, default 60); the bus removing it is heard as the expiry, which writes the finding backup-waits through the held writer over the module's findings and ends the wait, so the night never holds its lock without a bound. The file put-off.json and its writes are gone. The manifest requires put-off, findings and finding-events and hears findings-asked. Builds: design 47 G §561, §563, §557, §302, A-155, A-77 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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"}}}
|
||||
|
||||
@@ -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):
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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 `<module>.<item>` 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
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
${config:put-off-bound}
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user