mesh-recovery: take a restore point, list the points, expire them, one act at a time
Builds: design 47 G §516, §551, §554, §560, §564, §567, §517 Accept writes the hand-act log first, answers accepted/busy/error at once and writes the act before it answers; take writes each stream under a temporary name, its digests on the bus before ended; the cap is refused before anything is written; expiry never removes the newest point. Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -0,0 +1,214 @@
|
||||
package recovery_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
seat "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery"
|
||||
|
||||
"mesh-recovery/internal/recovery"
|
||||
)
|
||||
|
||||
// fakeBus holds streams as strings; a snapshot file is "snap:<name>:<content>".
|
||||
type fakeBus struct {
|
||||
mu sync.Mutex
|
||||
streams map[string]string
|
||||
consumers map[string][]seat.Consumer
|
||||
setBack []seat.Consumer
|
||||
failRestoreOnce bool
|
||||
}
|
||||
|
||||
func (b *fakeBus) Objects(_ context.Context, names []string) ([]recovery.Object, error) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
var out []recovery.Object
|
||||
for _, n := range names {
|
||||
c, ok := b.streams[n]
|
||||
out = append(out, recovery.Object{Stream: n, Exists: ok, Bytes: int64(len(c)), Consumers: b.consumers[n]})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (b *fakeBus) Take(_ context.Context, s string, w io.Writer) error {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
c, ok := b.streams[s]
|
||||
if !ok {
|
||||
return errors.New("no stream " + s)
|
||||
}
|
||||
_, err := io.WriteString(w, "snap:"+s+":"+c)
|
||||
return err
|
||||
}
|
||||
|
||||
func (b *fakeBus) Delete(_ context.Context, s string) error {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
delete(b.streams, s)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *fakeBus) Restore(_ context.Context, file string) error {
|
||||
raw, err := os.ReadFile(file)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
if b.failRestoreOnce && len(raw) > 0 && b.streams != nil {
|
||||
if string(raw[:5]) == "snap:" && !b.restoredOnce(raw) {
|
||||
return errors.New("the server refused the archive")
|
||||
}
|
||||
}
|
||||
var name, content string
|
||||
rest := string(raw[5:])
|
||||
for i := range rest {
|
||||
if rest[i] == ':' {
|
||||
name, content = rest[:i], rest[i+1:]
|
||||
break
|
||||
}
|
||||
}
|
||||
if _, live := b.streams[name]; live {
|
||||
return errors.New("stream name already in use")
|
||||
}
|
||||
b.streams[name] = content
|
||||
return nil
|
||||
}
|
||||
|
||||
var once sync.Map
|
||||
|
||||
func (b *fakeBus) restoredOnce(raw []byte) bool {
|
||||
_, loaded := once.LoadOrStore(string(raw), true)
|
||||
return loaded
|
||||
}
|
||||
|
||||
func (b *fakeBus) SetConsumers(_ context.Context, cs []seat.Consumer) error {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
b.setBack = append(b.setBack, cs...)
|
||||
return nil
|
||||
}
|
||||
|
||||
type said struct {
|
||||
Event string
|
||||
Body any
|
||||
Parts []string
|
||||
}
|
||||
|
||||
type fakeSeats struct {
|
||||
mu sync.Mutex
|
||||
events []said
|
||||
handActs []string
|
||||
handFail error
|
||||
streams []string
|
||||
backup string
|
||||
}
|
||||
|
||||
func (s *fakeSeats) Say(event string, body any, parts ...string) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.events = append(s.events, said{event, body, parts})
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *fakeSeats) HandAct(caller, asker, what, why string) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.handFail != nil {
|
||||
return s.handFail
|
||||
}
|
||||
s.handActs = append(s.handActs, caller+"|"+asker+"|"+what+"|"+why)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *fakeSeats) Streams(context.Context, string, string, string) ([]string, error) {
|
||||
return s.streams, nil
|
||||
}
|
||||
|
||||
func (s *fakeSeats) Backup(context.Context, string) (string, error) {
|
||||
if s.backup == "" {
|
||||
return "", errors.New("no repository snapshot holds it")
|
||||
}
|
||||
return s.backup, nil
|
||||
}
|
||||
|
||||
func (s *fakeSeats) named(event string) []said {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
var out []said
|
||||
for _, e := range s.events {
|
||||
if e.Event == event {
|
||||
out = append(out, e)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
type memDigests struct {
|
||||
mu sync.Mutex
|
||||
m map[string][]byte
|
||||
}
|
||||
|
||||
func (d *memDigests) Put(k string, v []byte) error {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
d.m[k] = v
|
||||
return nil
|
||||
}
|
||||
func (d *memDigests) Get(k string) ([]byte, bool, error) {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
v, ok := d.m[k]
|
||||
return v, ok, nil
|
||||
}
|
||||
func (d *memDigests) Delete(k string) error {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
delete(d.m, k)
|
||||
return nil
|
||||
}
|
||||
|
||||
type noDump struct{}
|
||||
|
||||
func (noDump) Wait(context.Context, func(string)) error { return nil }
|
||||
|
||||
type rig struct {
|
||||
e *recovery.Engine
|
||||
b *fakeBus
|
||||
s *fakeSeats
|
||||
d *memDigests
|
||||
t time.Time
|
||||
}
|
||||
|
||||
func newRig(t *testing.T) *rig {
|
||||
t.Helper()
|
||||
r := &rig{b: &fakeBus{streams: map[string]string{"app_notes": "hello", "KV_app_seen": "world"}, consumers: map[string][]seat.Consumer{}},
|
||||
s: &fakeSeats{streams: []string{"app_notes", "KV_app_seen"}}, d: &memDigests{m: map[string][]byte{}}, t: time.Date(2026, 11, 12, 1, 0, 0, 0, time.UTC)}
|
||||
r.b.consumers["app_notes"] = []seat.Consumer{{Name: "c1", Stream: "app_notes", Sequence: 7}}
|
||||
r.e = &recovery.Engine{Dir: filepath.Join(t.TempDir(), "points-dir"), Bus: r.b, Seats: r.s, Digests: r.d, Dump: noDump{},
|
||||
Settings: recovery.Settings{Cap: 1 << 20, KeepDays: 90}, Now: func() time.Time { return r.t }}
|
||||
return r
|
||||
}
|
||||
|
||||
func (r *rig) do(a recovery.Act) recovery.Answer {
|
||||
ans := r.e.Accept(a)
|
||||
r.e.Wait()
|
||||
return ans
|
||||
}
|
||||
|
||||
var errTest = errors.New("the log is unreachable")
|
||||
|
||||
type gatedBus struct {
|
||||
*fakeBus
|
||||
gate chan struct{}
|
||||
}
|
||||
|
||||
func (g *gatedBus) Objects(ctx context.Context, n []string) ([]recovery.Object, error) {
|
||||
<-g.gate
|
||||
return g.fakeBus.Objects(ctx, n)
|
||||
}
|
||||
@@ -0,0 +1,221 @@
|
||||
package recovery
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
seat "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery"
|
||||
"git.novox.be/novox/mesh-sdk/go/keys"
|
||||
)
|
||||
|
||||
const pointFile = "point.json"
|
||||
|
||||
// run is the work of one accepted act; it ends the act with its event under the act's id (§560).
|
||||
func (e *Engine) run(ctx context.Context, a *Act) {
|
||||
var err error
|
||||
switch a.Kind {
|
||||
case KindTake:
|
||||
err = e.take(ctx, a)
|
||||
case KindExpire:
|
||||
e.expire(a.ID)
|
||||
default:
|
||||
err = e.endedFailed(a, fmt.Errorf("no act of kind %q", a.Kind))
|
||||
}
|
||||
if err != nil {
|
||||
e.logf("act %s ended failed: %v", a.ID, err)
|
||||
}
|
||||
e.markEnded(a)
|
||||
}
|
||||
|
||||
// say publishes an event and logs the failure; an event lost is never silent.
|
||||
func (e *Engine) say(event string, body any, parts ...string) {
|
||||
if err := e.Seats.Say(event, body, parts...); err != nil {
|
||||
e.logf("the event %s was not said: %v", event, err)
|
||||
}
|
||||
}
|
||||
|
||||
func (e *Engine) endedFailed(a *Act, why error) error {
|
||||
e.say(seat.EventEnded, seat.EndedEvent{Act: a.ID, At: e.now(), Module: a.Module, Kept: a.Kept, Form: a.Form,
|
||||
Fingerprint: a.Fingerprint, Snapshot: a.Snapshot, Outcome: seat.EndedEventOutcomeFailed, Error: why.Error(), Why: cancelWhy(a, why)},
|
||||
a.Module, a.Kept)
|
||||
return why
|
||||
}
|
||||
|
||||
// cancelWhy is the hand act's words for a cancel (A-117).
|
||||
func cancelWhy(a *Act, why error) string {
|
||||
if why != nil && why.Error() == ErrCanceled {
|
||||
return a.Why
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// filled is the fill of the points directory without the temporary names of acts still writing.
|
||||
func (e *Engine) filled() int64 {
|
||||
return dirSize(filepath.Join(e.Dir, "points")) + dirSize(filepath.Join(e.Dir, "aside"))
|
||||
}
|
||||
|
||||
// take is `snapshot-kept` and the hand verb `take` (§516, §564, §567): one stream at a time under a temporary name,
|
||||
// renamed whole, its digests on the bus before `ended` is said.
|
||||
func (e *Engine) take(ctx context.Context, a *Act) error {
|
||||
dir, err := e.pointDir(a.Module, a.Kept, a.Form, a.Fingerprint)
|
||||
if err != nil {
|
||||
return e.endedFailed(a, err)
|
||||
}
|
||||
names, err := e.Seats.Streams(ctx, a.Module, a.Kept, a.Fingerprint)
|
||||
if err != nil || len(names) == 0 {
|
||||
if err == nil {
|
||||
err = fmt.Errorf("%s's %s has no bus name at %s", a.Module, a.Kept, a.Fingerprint)
|
||||
}
|
||||
return e.endedFailed(a, err)
|
||||
}
|
||||
e.setStep(a, "measuring", "")
|
||||
objs, err := e.Bus.Objects(ctx, names)
|
||||
if err != nil {
|
||||
return e.endedFailed(a, err)
|
||||
}
|
||||
var need int64
|
||||
for _, o := range objs {
|
||||
need += o.Bytes
|
||||
}
|
||||
if fill := e.filled(); fill+need > e.Settings.Cap {
|
||||
return e.endedFailed(a, fmt.Errorf("%s: the cap is %d bytes, the point needs %d and the directory holds %d", ErrCap, e.Settings.Cap, need, fill))
|
||||
}
|
||||
tmp := filepath.Join(filepath.Dir(dir), ".tmp-"+a.ID)
|
||||
_ = os.RemoveAll(tmp)
|
||||
if err := os.MkdirAll(tmp, 0o700); err != nil {
|
||||
return e.endedFailed(a, err)
|
||||
}
|
||||
defer os.RemoveAll(tmp)
|
||||
p := seat.Point{At: e.now(), CatchUp: a.CatchUp, Fingerprint: a.Fingerprint, Form: a.Form, Kept: a.Kept, Module: a.Module,
|
||||
Snapshot: strings.TrimPrefix(dir, filepath.Join(e.Dir, "points")+string(filepath.Separator))}
|
||||
e.setStep(a, "taking", "")
|
||||
for _, o := range objs {
|
||||
if !o.Exists {
|
||||
continue
|
||||
}
|
||||
file := filepath.Join(tmp, keys.EscapeName(o.Stream)+".snap")
|
||||
sum, n, err := e.takeOne(ctx, o.Stream, file)
|
||||
if err != nil {
|
||||
return e.endedFailed(a, err)
|
||||
}
|
||||
p.Digests = append(p.Digests, seat.Digest{Name: o.Stream, Digest: sum})
|
||||
p.Size += n
|
||||
p.Consumers = append(p.Consumers, o.Consumers...)
|
||||
if e.filled()+p.Size > e.Settings.Cap {
|
||||
return e.endedFailed(a, fmt.Errorf("%s: the stream grew while it was taken and passes the cap of %d bytes", ErrCap, e.Settings.Cap))
|
||||
}
|
||||
}
|
||||
if err := writeJSON(filepath.Join(tmp, pointFile), p); err != nil {
|
||||
return e.endedFailed(a, err)
|
||||
}
|
||||
_ = os.RemoveAll(dir)
|
||||
if err := os.Rename(tmp, dir); err != nil {
|
||||
return e.endedFailed(a, err)
|
||||
}
|
||||
raw, _ := jsonMarshal(p.Digests)
|
||||
if err := e.Digests.Put(keys.EscapeName(p.Snapshot), raw); err != nil {
|
||||
_ = os.RemoveAll(dir)
|
||||
return e.endedFailed(a, fmt.Errorf("the digests are not on the bus: %w", err))
|
||||
}
|
||||
e.say(seat.EventEnded, seat.EndedEvent{Act: a.ID, At: e.now(), Module: a.Module, Kept: a.Kept, Form: a.Form,
|
||||
Fingerprint: a.Fingerprint, Snapshot: p.Snapshot, Outcome: seat.EndedEventOutcomeDone, Consumers: p.Consumers,
|
||||
Digests: p.Digests, Size: p.Size, CatchUp: p.CatchUp}, a.Module, a.Kept)
|
||||
e.expire(a.ID)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *Engine) takeOne(ctx context.Context, stream, file string) (string, int64, error) {
|
||||
f, err := os.Create(file)
|
||||
if err != nil {
|
||||
return "", 0, err
|
||||
}
|
||||
if err := e.Bus.Take(ctx, stream, f); err != nil {
|
||||
f.Close()
|
||||
return "", 0, fmt.Errorf("stream %s: %w", stream, err)
|
||||
}
|
||||
if err := f.Close(); err != nil {
|
||||
return "", 0, err
|
||||
}
|
||||
return digestOf(file)
|
||||
}
|
||||
|
||||
func digestOf(file string) (string, int64, error) {
|
||||
f, err := os.Open(file)
|
||||
if err != nil {
|
||||
return "", 0, err
|
||||
}
|
||||
defer f.Close()
|
||||
h := sha256.New()
|
||||
n, err := io.Copy(h, f)
|
||||
return "sha256:" + hex.EncodeToString(h.Sum(nil)), n, err
|
||||
}
|
||||
|
||||
// Points lists every restore point in the directory, newest first (§517): the one place they are kept.
|
||||
func (e *Engine) Points() ([]seat.Point, error) {
|
||||
var out []seat.Point
|
||||
root := filepath.Join(e.Dir, "points")
|
||||
err := filepath.WalkDir(root, func(path string, d os.DirEntry, err error) error {
|
||||
if err != nil {
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
if d.IsDir() && strings.HasPrefix(d.Name(), ".") {
|
||||
return filepath.SkipDir
|
||||
}
|
||||
if !d.IsDir() && d.Name() == pointFile {
|
||||
var p seat.Point
|
||||
if err := readJSON(path, &p); err != nil {
|
||||
return fmt.Errorf("%s: %w", path, err)
|
||||
}
|
||||
out = append(out, p)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].At.After(out[j].At) })
|
||||
return out, err
|
||||
}
|
||||
|
||||
// expire is §551's removal: a point 90 days after its `at` (the setting), never the newest of its kept data, each removal
|
||||
// said as point-removed and the run as expiry-ended, also when it removes nothing (§560). It runs inside an act.
|
||||
func (e *Engine) expire(act string) {
|
||||
pts, err := e.Points()
|
||||
if err != nil {
|
||||
e.logf("expiry: %v", err)
|
||||
}
|
||||
newest := map[string]string{}
|
||||
for _, p := range pts { // newest first
|
||||
k := p.Module + "\x00" + p.Kept
|
||||
if _, ok := newest[k]; !ok {
|
||||
newest[k] = p.Snapshot
|
||||
}
|
||||
}
|
||||
cutoff := e.now().AddDate(0, 0, -e.Settings.KeepDays)
|
||||
removed := int64(0)
|
||||
for _, p := range pts {
|
||||
if !p.At.Before(cutoff) || newest[p.Module+"\x00"+p.Kept] == p.Snapshot {
|
||||
continue
|
||||
}
|
||||
if err := os.RemoveAll(filepath.Join(e.Dir, "points", p.Snapshot)); err != nil {
|
||||
e.logf("the point %s is not removed: %v", p.Snapshot, err)
|
||||
continue
|
||||
}
|
||||
_ = e.Digests.Delete(keys.EscapeName(p.Snapshot))
|
||||
if err := e.Seats.HandAct(Module, "", "point-removed "+p.Snapshot, "older than the keep days"); err != nil {
|
||||
e.logf("the removal of %s is not in the hand-act log: %v", p.Snapshot, err)
|
||||
}
|
||||
e.say(seat.EventPointRemoved, seat.PointRemovedEvent{Act: act, At: e.now(), Module: p.Module, Kept: p.Kept, Form: p.Form,
|
||||
Fingerprint: p.Fingerprint, Snapshot: p.Snapshot}, p.Module, p.Kept)
|
||||
removed++
|
||||
}
|
||||
e.say(seat.EventExpiryEnded, seat.ExpiryEndedEvent{Act: act, At: e.now(), Removed: removed})
|
||||
}
|
||||
@@ -0,0 +1,121 @@
|
||||
package recovery_test
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
seat "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery"
|
||||
|
||||
"mesh-recovery/internal/recovery"
|
||||
)
|
||||
|
||||
func takeAct() recovery.Act {
|
||||
return recovery.Act{Kind: recovery.KindTake, Module: "app", Kept: "notes", Form: 2, Fingerprint: "fp1", Why: "form change", Caller: "node.engine"}
|
||||
}
|
||||
|
||||
// A point is taken stream by stream under a temporary name, renamed whole, its digests put on the bus before `ended` is
|
||||
// said under the act's id with the point whole, and the act is written in the hand-act log with the caller the bus proved
|
||||
// (design 47 G §516, §554, §560, §567).
|
||||
func TestATakenPointIsListedWithItsDigestsAndEndedUnderItsAct(t *testing.T) {
|
||||
r := newRig(t)
|
||||
ans := r.do(takeAct())
|
||||
if ans.Accepted == "" {
|
||||
t.Fatalf("answer %+v", ans)
|
||||
}
|
||||
ended := r.s.named(seat.EventEnded)
|
||||
if len(ended) != 1 {
|
||||
t.Fatalf("events %+v", r.s.events)
|
||||
}
|
||||
ev := ended[0].Body.(seat.EndedEvent)
|
||||
if ev.Act != ans.Accepted || ev.Outcome != "done" || len(ev.Digests) != 2 || ev.Size == 0 || ev.Snapshot != "app/notes/2-fp1" {
|
||||
t.Errorf("ended %+v", ev)
|
||||
}
|
||||
if got := strings.Join(ended[0].Parts, "."); got != "app.notes" {
|
||||
t.Errorf("parts %q", got)
|
||||
}
|
||||
if len(ev.Consumers) != 1 || ev.Consumers[0].Sequence != 7 {
|
||||
t.Errorf("consumers %+v", ev.Consumers)
|
||||
}
|
||||
pts, err := r.e.Points()
|
||||
if err != nil || len(pts) != 1 || pts[0].Snapshot != ev.Snapshot {
|
||||
t.Fatalf("points %+v %v", pts, err)
|
||||
}
|
||||
if _, ok, _ := r.d.Get("app/notes/2-fp1"); !ok {
|
||||
t.Errorf("no digests on the bus: %v", r.d.m)
|
||||
}
|
||||
if len(r.s.handActs) != 1 || !strings.HasPrefix(r.s.handActs[0], "node.engine||take app notes|form change") {
|
||||
t.Errorf("hand acts %v", r.s.handActs)
|
||||
}
|
||||
}
|
||||
|
||||
// One act at a time and no queue: a second ask while one runs is answered busy with the running act (§554).
|
||||
func TestASecondAskWhileOneRunsIsAnsweredBusy(t *testing.T) {
|
||||
r := newRig(t)
|
||||
gate := make(chan struct{})
|
||||
r.e.Dump = nil
|
||||
r.e.Bus = &gatedBus{fakeBus: r.b, gate: gate}
|
||||
first := r.e.Accept(takeAct())
|
||||
second := r.e.Accept(takeAct())
|
||||
close(gate)
|
||||
r.e.Wait()
|
||||
if first.Accepted == "" || second.Busy != first.Accepted || second.Accepted != "" || second.Error != "" {
|
||||
t.Errorf("first %+v second %+v", first, second)
|
||||
}
|
||||
}
|
||||
|
||||
// An act whose hand-act log write fails does nothing and answers an error naming the log (§554).
|
||||
func TestAnActWhoseHandActLogWriteFailsDoesNothing(t *testing.T) {
|
||||
r := newRig(t)
|
||||
r.s.handFail = errTest
|
||||
ans := r.do(takeAct())
|
||||
if !strings.Contains(ans.Error, "hand-act log") || ans.Accepted != "" || len(r.s.events) != 0 {
|
||||
t.Errorf("answer %+v events %+v", ans, r.s.events)
|
||||
}
|
||||
if pts, _ := r.e.Points(); len(pts) != 0 {
|
||||
t.Errorf("points %+v", pts)
|
||||
}
|
||||
}
|
||||
|
||||
// An act that would pass the cap ends failed naming it, before anything is written (§564).
|
||||
func TestAnActThatWouldPassTheCapEndsFailedBeforeItWritesAnything(t *testing.T) {
|
||||
r := newRig(t)
|
||||
r.e.Settings.Cap = 4
|
||||
r.do(takeAct())
|
||||
ev := r.s.named(seat.EventEnded)[0].Body.(seat.EndedEvent)
|
||||
if ev.Outcome != "failed" || !strings.HasPrefix(ev.Error, recovery.ErrCap) {
|
||||
t.Errorf("ended %+v", ev)
|
||||
}
|
||||
if pts, _ := r.e.Points(); len(pts) != 0 || len(r.d.m) != 0 {
|
||||
t.Errorf("points %+v digests %v", pts, r.d.m)
|
||||
}
|
||||
}
|
||||
|
||||
// A point older than the keep days is removed and said, the newest of its kept data stays however old, and the run says
|
||||
// expiry-ended with the count, 0 included (§551, §560).
|
||||
func TestExpiryRemovesOldPointsButNeverTheNewest(t *testing.T) {
|
||||
r := newRig(t)
|
||||
r.t = time.Date(2026, 6, 1, 0, 0, 0, 0, time.UTC)
|
||||
r.do(takeAct())
|
||||
a := takeAct()
|
||||
a.Fingerprint, a.Form = "fp2", 3
|
||||
r.t = time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC)
|
||||
r.do(a)
|
||||
r.s.events = nil
|
||||
r.t = time.Date(2026, 11, 12, 0, 0, 0, 0, time.UTC)
|
||||
r.do(recovery.Act{Kind: recovery.KindExpire, Why: "by hand", Caller: "tools"})
|
||||
pts, _ := r.e.Points()
|
||||
if len(pts) != 1 || pts[0].Fingerprint != "fp2" {
|
||||
t.Fatalf("points %+v", pts)
|
||||
}
|
||||
rem := r.s.named(seat.EventPointRemoved)
|
||||
if len(rem) != 1 || rem[0].Body.(seat.PointRemovedEvent).Fingerprint != "fp1" {
|
||||
t.Errorf("removed %+v", rem)
|
||||
}
|
||||
if ex := r.s.named(seat.EventExpiryEnded); len(ex) != 1 || ex[0].Body.(seat.ExpiryEndedEvent).Removed != 1 {
|
||||
t.Errorf("expiry %+v", ex)
|
||||
}
|
||||
if _, ok, _ := r.d.Get("app/notes/2-fp1"); ok {
|
||||
t.Errorf("its digests stay on the bus")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,270 @@
|
||||
// Package recovery is the recovery module's engine (design 47 G §554, §516): it takes, sets aside and restores a
|
||||
// module's kept data on the live bus, one act at a time, each accepted act written in its directory before the caller is
|
||||
// answered, the result said as an event under the act's id (§560). Every dependence on the bus, the seat list, the
|
||||
// hand-act log and the backup holder is an interface here, so the one whole test runs the engine against the real bus
|
||||
// and a unit test against a fake.
|
||||
package recovery
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
seat "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery"
|
||||
"git.novox.be/novox/mesh-sdk/go/keys"
|
||||
)
|
||||
|
||||
// Module and Seat are the module's name and the seat it claims.
|
||||
const (
|
||||
Module = "mesh-recovery"
|
||||
Seat = seat.Seat
|
||||
)
|
||||
|
||||
// The kinds of act, one per verb that starts one.
|
||||
const (
|
||||
KindTake = "take"
|
||||
KindRestore = "restore"
|
||||
KindAside = "aside"
|
||||
KindDiscardAside = "discard-aside"
|
||||
KindRestoreAside = "restore-aside"
|
||||
KindExpire = "expire-points"
|
||||
)
|
||||
|
||||
// Errors an act ends with: the words the node agent's table reads (§560).
|
||||
const (
|
||||
ErrCap = "cap"
|
||||
ErrStopped = "stopped"
|
||||
ErrCanceled = "cancelled"
|
||||
ErrMissing = "restore point missing"
|
||||
ErrDamaged = "restore point damaged"
|
||||
)
|
||||
|
||||
// Object is one bus object of a kept data: a stream, by its name, with what the bus states of it.
|
||||
type Object struct {
|
||||
Stream string
|
||||
Exists bool
|
||||
Bytes int64
|
||||
Consumers []seat.Consumer
|
||||
}
|
||||
|
||||
// Bus is the bus as the module's account sees it (§554): a stream's snapshot and restore, its deletion, the
|
||||
// durable consumers' positions.
|
||||
type Bus interface {
|
||||
// Objects states each stream of names: whether it exists, its bytes, its durable consumers' acknowledged sequences.
|
||||
Objects(ctx context.Context, streams []string) ([]Object, error)
|
||||
// Take writes one stream's snapshot file to w (ADR 0235 §10's pacing and §11's bounds are the implementation's).
|
||||
Take(ctx context.Context, stream string, w io.Writer) error
|
||||
// Delete removes a stream; one that is not there is no error.
|
||||
Delete(ctx context.Context, stream string) error
|
||||
// Restore puts the stream of a snapshot file back under its own name, where no stream of that name exists.
|
||||
Restore(ctx context.Context, file string) error
|
||||
// SetConsumers creates each durable consumer at its recorded sequence.
|
||||
SetConsumers(ctx context.Context, cs []seat.Consumer) error
|
||||
}
|
||||
|
||||
// Seats is what the engine asks of the rest of the mesh.
|
||||
type Seats interface {
|
||||
// Say publishes one event of the seat under its subject parts (§560); an event of no parts has none.
|
||||
Say(event string, body any, parts ...string) error
|
||||
// HandAct writes the act in the hand-act log first (§554): the account the bus proved, who asked, what, why.
|
||||
HandAct(caller, asker, what, why string) error
|
||||
// Streams are the bus names of a module's kept data at a fingerprint, from its catalog graph entry (§554 Bus names).
|
||||
Streams(ctx context.Context, module, kept, fingerprint string) ([]string, error)
|
||||
// Backup brings the points directory of the module's item back beside its path through the backup holder and
|
||||
// answers where (§516); an error when no repository snapshot holds it.
|
||||
Backup(ctx context.Context, module string) (dir string, err error)
|
||||
}
|
||||
|
||||
// Digests keeps each point's digests on the bus, apart from its bytes (§567): the state digests.
|
||||
type Digests interface {
|
||||
Put(key string, v []byte) error
|
||||
Get(key string) ([]byte, bool, error)
|
||||
Delete(key string) error
|
||||
}
|
||||
|
||||
// Dump is what the act waits on while the bus's item is dumped (§563).
|
||||
type Dump interface {
|
||||
// Wait returns once no dump of the bus's item runs; it calls waiting(on) once if it has to wait.
|
||||
Wait(ctx context.Context, waiting func(on string)) error
|
||||
}
|
||||
|
||||
// Settings are the bounds a whole test sets in seconds (§557, §564, §633).
|
||||
type Settings struct {
|
||||
Cap int64
|
||||
KeepDays int
|
||||
}
|
||||
|
||||
// Engine is the recovery module's one engine.
|
||||
type Engine struct {
|
||||
Dir string
|
||||
Bus Bus
|
||||
Seats Seats
|
||||
Digests Digests
|
||||
Dump Dump
|
||||
Settings Settings
|
||||
Now func() time.Time
|
||||
Logf func(string, ...any)
|
||||
|
||||
mu sync.Mutex
|
||||
running *Act
|
||||
cancel context.CancelFunc
|
||||
wg sync.WaitGroup
|
||||
seq int
|
||||
}
|
||||
|
||||
// Act is one accepted act, written in the directory before its caller is answered (§554).
|
||||
type Act struct {
|
||||
ID string `json:"act"`
|
||||
Kind string `json:"kind"`
|
||||
Module string `json:"module,omitempty"`
|
||||
Kept string `json:"kept,omitempty"`
|
||||
Form int64 `json:"form,omitempty"`
|
||||
Fingerprint string `json:"fingerprint,omitempty"`
|
||||
Snapshot string `json:"snapshot,omitempty"`
|
||||
AsideID string `json:"aside-id,omitempty"`
|
||||
FromBackup bool `json:"from-backup,omitempty"`
|
||||
CatchUp *seat.CatchUp `json:"catch-up,omitempty"`
|
||||
Why string `json:"why"`
|
||||
Caller string `json:"caller"`
|
||||
Asker string `json:"asker,omitempty"`
|
||||
Since time.Time `json:"since"`
|
||||
Step string `json:"step,omitempty"`
|
||||
WaitsOn string `json:"waits-on,omitempty"`
|
||||
Ended bool `json:"ended,omitempty"`
|
||||
}
|
||||
|
||||
// Answer is the one form every acting verb answers (§691): exactly one field set.
|
||||
type Answer struct {
|
||||
Accepted string `json:"accepted,omitempty"`
|
||||
Busy string `json:"busy,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
func (e *Engine) now() time.Time {
|
||||
if e.Now != nil {
|
||||
return e.Now().UTC()
|
||||
}
|
||||
return time.Now().UTC()
|
||||
}
|
||||
|
||||
func (e *Engine) logf(f string, a ...any) {
|
||||
if e.Logf != nil {
|
||||
e.Logf(f, a...)
|
||||
}
|
||||
}
|
||||
|
||||
// Tokens are escaped as a key's name is (§73); a token that is empty, `.` or `..` is refused.
|
||||
func token(s string) (string, error) {
|
||||
if s == "" || s == "." || s == ".." {
|
||||
return "", fmt.Errorf("%q is not a name", s)
|
||||
}
|
||||
return keys.EscapeName(s), nil
|
||||
}
|
||||
|
||||
func (e *Engine) pointDir(module, kept string, form int64, fp string) (string, error) {
|
||||
m, err := token(module)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
k, err := token(kept)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
f, err := token(fp)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return filepath.Join(e.Dir, "points", m, k, fmt.Sprintf("%d-%s", form, f)), nil
|
||||
}
|
||||
|
||||
func writeJSON(path string, v any) error {
|
||||
raw, err := json.MarshalIndent(v, "", " ")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tmp := path + ".partial"
|
||||
if err := os.WriteFile(tmp, raw, 0o600); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.Rename(tmp, path)
|
||||
}
|
||||
|
||||
func readJSON(path string, v any) error {
|
||||
raw, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return json.Unmarshal(raw, v)
|
||||
}
|
||||
|
||||
func dirSize(dir string) int64 {
|
||||
var n int64
|
||||
_ = filepath.WalkDir(dir, func(_ string, d os.DirEntry, err error) error {
|
||||
if err == nil && !d.IsDir() {
|
||||
if info, err := d.Info(); err == nil {
|
||||
n += info.Size()
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return n
|
||||
}
|
||||
|
||||
var errBusy = errors.New("busy")
|
||||
|
||||
// Accept is the one door of every acting verb (§554, §682): it writes the act in the hand-act log first (a failed
|
||||
// write does nothing and answers an error naming the log), then answers busy with the running act's id, or writes the
|
||||
// act in its directory and answers accepted; the work runs on, and its result is the event.
|
||||
func (e *Engine) Accept(a Act) Answer {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
if e.running != nil {
|
||||
return Answer{Busy: e.running.ID}
|
||||
}
|
||||
if err := e.Seats.HandAct(a.Caller, a.Asker, a.Kind+" "+a.Module+" "+a.Kept, a.Why); err != nil {
|
||||
return Answer{Error: "the hand-act log was not written, so nothing was done: " + err.Error()}
|
||||
}
|
||||
e.seq++
|
||||
a.ID = fmt.Sprintf("act-%s-%d", e.now().Format("20060102T150405Z"), e.seq)
|
||||
a.Since = e.now()
|
||||
if err := os.MkdirAll(filepath.Join(e.Dir, "acts"), 0o700); err != nil {
|
||||
return Answer{Error: err.Error()}
|
||||
}
|
||||
if err := writeJSON(filepath.Join(e.Dir, "acts", a.ID+".json"), a); err != nil {
|
||||
return Answer{Error: "the act cannot be written: " + err.Error()}
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
e.running, e.cancel = &a, cancel
|
||||
e.wg.Add(1)
|
||||
go func() {
|
||||
defer e.wg.Done()
|
||||
e.run(ctx, &a)
|
||||
e.mu.Lock()
|
||||
e.running, e.cancel = nil, nil
|
||||
e.mu.Unlock()
|
||||
cancel()
|
||||
}()
|
||||
return Answer{Accepted: a.ID}
|
||||
}
|
||||
|
||||
// Wait returns once the running act, if any, has ended (a whole test's and a stop's join).
|
||||
func (e *Engine) Wait() { e.wg.Wait() }
|
||||
|
||||
func (e *Engine) setStep(a *Act, step, waitsOn string) {
|
||||
e.mu.Lock()
|
||||
a.Step, a.WaitsOn = step, waitsOn
|
||||
e.mu.Unlock()
|
||||
}
|
||||
|
||||
func (e *Engine) markEnded(a *Act) {
|
||||
a.Ended = true
|
||||
_ = writeJSON(filepath.Join(e.Dir, "acts", a.ID+".json"), a)
|
||||
}
|
||||
|
||||
func jsonMarshal(v any) ([]byte, error) { return json.Marshal(v) }
|
||||
Reference in New Issue
Block a user