mesh-recovery: the hand verb restore takes a keeper's snapshot by request

An unprotected snapshot is received through snapshot-get into
incoming/<request>, refused before the first chunk past the cap, checked
against its digest before any aside, and restored by the module; a
protected one is verified, the running data set aside and restored in
place by the keeper, waited on by snapshot-restored within
snapshot-restore-wait-minutes, cancelled at the bound. The bus module
sends its chunks with mesh-sdk's form.

Builds: design 47 G §681, §682, A-137, A-138, A-139, A-140

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h

modules/mesh-recovery/cmd/mesh-recovery/keeper.go

@@ -0,0 +1,151 @@
+package main
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"io"
+	"sync"
+	"time"
+
+	"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
+	"git.novox.be/novox/mesh-sdk/go/buscalls"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"mesh-recovery/internal/recovery"
+)
+
+// liveKeeper is the bus module's keeper reached by the seat mesh-broker's verbs (design 47 G §681): snapshot-get's chunks
+// received in mesh-sdk's form (A-139), snapshot-restore and its cancel called, and snapshot-restored heard.
+type liveKeeper struct {
+	nc       *nats.Conn
+	call     Call
+	restored *restoredView
+}
+
+func (k liveKeeper) verb(verb string, in, out any) error {
+	raw, err := k.call("seat:"+meshbroker.Seat+"."+verb, in)
+	if err != nil {
+		return fmt.Errorf("%s's %s: %w", meshbroker.Seat, verb, err)
+	}
+	if err := json.Unmarshal(raw, out); err != nil {
+		return fmt.Errorf("%s's %s answered what does not read: %w", meshbroker.Seat, verb, err)
+	}
+	return nil
+}
+
+// Get subscribes to the chunks of a call of its own before it asks snapshot-get, so none is missed, then receives them in
+// order, acknowledging each (§681, A-139).
+func (k liveKeeper) Get(ctx context.Context, request string, fits func(meshbroker.SnapshotGetAnswer) error, w io.Writer) (meshbroker.SnapshotGetAnswer, error) {
+	var bits [16]byte
+	if _, err := rand.Read(bits[:]); err != nil {
+		return meshbroker.SnapshotGetAnswer{}, err
+	}
+	send := buscalls.SeatChunks{Seat: meshbroker.Seat, Verb: meshbroker.VerbSnapshotGet, Call: hex.EncodeToString(bits[:])}
+	chunks := make(chan *nats.Msg, 4)
+	sub, err := k.nc.ChanSubscribe(send.Chunks(), chunks)
+	if err != nil {
+		return meshbroker.SnapshotGetAnswer{}, fmt.Errorf("the chunks of snapshot-get cannot be heard: %w", err)
+	}
+	defer func() { _ = sub.Unsubscribe() }()
+	var ans meshbroker.SnapshotGetAnswer
+	if err := k.verb(meshbroker.VerbSnapshotGet, meshbroker.SnapshotGetInput{Call: send.Call, Request: request}, &ans); err != nil {
+		return ans, err
+	}
+	if ans.Error != "" {
+		return ans, nil
+	}
+	if err := fits(ans); err != nil {
+		return ans, err
+	}
+	if ans.Size > 0 && ans.Chunks == 0 {
+		return ans, fmt.Errorf("the keeper answered the request %s's %d bytes with no chunk: this account is granted no chunk subjects of snapshot-get", request, ans.Size)
+	}
+	next := func() (string, []byte, error) {
+		bound := time.NewTimer(buscalls.SeatAckWithin)
+		defer bound.Stop()
+		select {
+		case m := <-chunks:
+			return m.Subject, m.Data, nil
+		case <-bound.C:
+			return "", nil, buscalls.ErrSeatChunkMissed
+		case <-ctx.Done():
+			return "", nil, ctx.Err()
+		}
+	}
+	ack := func(subject string) error {
+		_, err := k.nc.Request(subject, []byte("{}"), buscalls.SeatAckWithin)
+		return err
+	}
+	if _, err := send.Receive(ans.Chunks, next, ack, w); err != nil {
+		return ans, fmt.Errorf("the snapshot of the request %s: %w", request, err)
+	}
+	return ans, nil
+}
+
+func (k liveKeeper) Restore(_ context.Context, request string, verify bool, why string) (meshbroker.SnapshotRestoreAnswer, error) {
+	var ans meshbroker.SnapshotRestoreAnswer
+	err := k.verb(meshbroker.VerbSnapshotRestore, meshbroker.SnapshotRestoreInput{Request: request, Verify: verify, Why: why}, &ans)
+	return ans, err
+}
+
+func (k liveKeeper) CancelRestore(_ context.Context, act, why string) (meshbroker.SnapshotRestoreCancelAnswer, error) {
+	var ans meshbroker.SnapshotRestoreCancelAnswer
+	err := k.verb(meshbroker.VerbSnapshotRestoreCancel, meshbroker.SnapshotRestoreCancelInput{Act: act, Why: why}, &ans)
+	return ans, err
+}
+
+func (k liveKeeper) Restored(ctx context.Context, request, act string) (meshbroker.SnapshotRestoredEvent, bool) {
+	return k.restored.wait(ctx, request, act)
+}
+
+var _ recovery.Keeper = liveKeeper{}
+
+// restoredView is the keeper's snapshot-restored as heard, each request and act kept (§681): read as §556 has it, the last
+// of each subject first, so an outcome said before a wait began wakes it too.
+type restoredView struct {
+	mu      sync.Mutex
+	by      map[string]meshbroker.SnapshotRestoredEvent
+	changed chan struct{}
+}
+
+func newRestoredView() *restoredView {
+	return &restoredView{by: map[string]meshbroker.SnapshotRestoredEvent{}, changed: make(chan struct{})}
+}
+
+func (v *restoredView) hear(m jetstream.Msg) {
+	var ev meshbroker.SnapshotRestoredEvent
+	if err := json.Unmarshal(m.Data(), &ev); err == nil && ev.Request != "" && ev.Act != "" {
+		v.take(ev)
+	}
+	_ = m.Ack()
+}
+
+func (v *restoredView) take(ev meshbroker.SnapshotRestoredEvent) {
+	v.mu.Lock()
+	v.by[ev.Request+"\x00"+ev.Act] = ev
+	close(v.changed)
+	v.changed = make(chan struct{})
+	v.mu.Unlock()
+}
+
+func (v *restoredView) wait(ctx context.Context, request, act string) (meshbroker.SnapshotRestoredEvent, bool) {
+	for {
+		v.mu.Lock()
+		ev, ok := v.by[request+"\x00"+act]
+		changed := v.changed
+		v.mu.Unlock()
+		if ok {
+			return ev, true
+		}
+		select {
+		case <-changed:
+		case <-ctx.Done():
+			return meshbroker.SnapshotRestoredEvent{}, false
+		}
+	}
+}

modules/mesh-recovery/cmd/mesh-recovery/main.go

@@ -35,6 +35,7 @@ const (
 	envKeepDays = "MESH_RECOVERY_KEEP_DAYS_FILE"
 	envMark     = "MESH_RECOVERY_FILLING_MARK_FILE"
 	envDumpWait = "MESH_RECOVERY_DUMP_WAIT_FILE"
+	envRestore  = "MESH_RECOVERY_RESTORE_WAIT_FILE"
 )
 
 // The settings' defaults: the design's values (§551, §564).
@@ -96,6 +97,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err != nil {
 		return err
 	}
+	restoreWait, err := readMinutes(getenv(envRestore), recovery.DefaultRestoreWait)
+	if err != nil {
+		return err
+	}
 	account, err := process.ReadBrokerAccount(getenv)
 	if err != nil {
 		return err
@@ -123,6 +128,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			return err
 		}
 	}
+	// The keeper's snapshot-restored, each request and act, for a restore in place (§681).
+	restored := newRestoredView()
+	if _, err := account.HearSeatEvents(ctx, js, meshbroker.StreamSnapshotRestored, restored.hear, logger.Printf); err != nil {
+		return err
+	}
 	// The module's findings go through mesh-sdk's held writer over its own state findings (§561), started before the
 	// engine's start, so what the start writes is held and written.
 	writer, err := views.NewHeldWriter(views.WriterConfig[meshconditions.Finding, meshconditions.FindingCleared, meshconditions.FindingsStated, meshconditions.FindingsAskedEvent]{
@@ -143,7 +153,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			node: account.Node, points: dir, running: running},
 		Digests:  digests{conn.State("digests")},
 		Dump:     dump,
-		Settings: recovery.Settings{Cap: cap, KeepDays: int(keepDays), FillingMark: int(mark), DumpWait: dumpWait},
+		Keeper:   liveKeeper{nc: nc, call: conn.Call, restored: restored},
+		Settings: recovery.Settings{Cap: cap, KeepDays: int(keepDays), FillingMark: int(mark), DumpWait: dumpWait, RestoreWait: restoreWait},
 		Findings: writer,
 		Node:     account.Node,
 		Logf:     logger.Printf,

modules/mesh-recovery/cmd/mesh-recovery/main_test.go

@@ -1,10 +1,12 @@
 package main
 
 import (
+	"context"
 	"strings"
 	"testing"
 	"time"
 
+	"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
 	"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
 	"git.novox.be/novox/mesh-sdk/go/process"
@@ -48,6 +50,27 @@ func TestWhatTheProcessReads(t *testing.T) {
 			t.Errorf("two fingerprints answered %v", err)
 		}
 	})
+	t.Run("the keeper's snapshot-restored of a request and act (§681)", func(t *testing.T) {
+		v := newRestoredView()
+		v.take(meshbroker.SnapshotRestoredEvent{Request: "req-1", Act: "act-0", Outcome: "failed"})
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+		if _, ok := v.wait(ctx, "req-1", "act-1"); ok {
+			t.Error("another act's outcome woke the wait")
+		}
+		woken := make(chan meshbroker.SnapshotRestoredEvent, 1)
+		go func() {
+			ev, _ := v.wait(context.Background(), "req-1", "act-1")
+			woken <- ev
+		}()
+		v.take(meshbroker.SnapshotRestoredEvent{Request: "req-1", Act: "act-1", Outcome: "done"})
+		if ev := <-woken; ev.Outcome != "done" {
+			t.Errorf("woken by %+v", ev)
+		}
+		if ev, ok := v.wait(ctx, "req-1", "act-1"); !ok || ev.Outcome != "done" {
+			t.Errorf("an outcome said before the wait is not taken: %+v %v", ev, ok)
+		}
+	})
 	t.Run("a dump of the bus module runs (§563)", func(t *testing.T) {
 		d := newDumpView(nil, nil)
 		start := time.Date(2026, 11, 12, 2, 0, 0, 0, time.UTC)

modules/mesh-recovery/files/snapshot-restore-wait-minutes

@@ -0,0 +1 @@
+${config:snapshot-restore-wait-minutes}

modules/mesh-recovery/internal/recovery/control.go

@@ -134,6 +134,8 @@ func (e *Engine) Start(ctx context.Context) []UnrestoredAside {
 		}
 		return nil
 	})
+	// A keeper's snapshot a cut-off act left half or whole received is deleted (§681).
+	_ = os.RemoveAll(e.incoming())
 	_, recs, _ := e.Asides()
 	for i := range recs {
 		r := &recs[i]

modules/mesh-recovery/internal/recovery/keeper.go

@@ -0,0 +1,191 @@
+package recovery
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"time"
+
+	"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
+)
+
+// Keeper is the bus module's keeper as the seat mesh-broker answers it (design 47 G §681): its snapshot taken before a
+// `drop` or `recreate`, sent by snapshot-get, or restored in place by snapshot-restore when it holds a protected stream.
+type Keeper interface {
+	// Get asks snapshot-get of the request; when its answer names a file and fits accepts it, it receives the chunks into
+	// w. A refusal of the keeper is the answer's Error, no error of Get.
+	Get(ctx context.Context, request string, fits func(meshbroker.SnapshotGetAnswer) error, w io.Writer) (meshbroker.SnapshotGetAnswer, error)
+	// Restore asks snapshot-restore; with verify the keeper checks the file and restores nothing.
+	Restore(ctx context.Context, request string, verify bool, why string) (meshbroker.SnapshotRestoreAnswer, error)
+	// CancelRestore asks snapshot-restore-cancel of a running restore.
+	CancelRestore(ctx context.Context, act, why string) (meshbroker.SnapshotRestoreCancelAnswer, error)
+	// Restored answers the keeper's snapshot-restored of that request and act once heard (read as §556 has it, so one said
+	// before the wait counts), or ok false when ctx ends first.
+	Restored(ctx context.Context, request, act string) (ev meshbroker.SnapshotRestoredEvent, ok bool)
+}
+
+// The keeper's word for a snapshot it never hands out (§681).
+const snapshotProtected = "snapshot protected"
+
+// ErrUnanswered is the end of a restore in place the keeper answered neither by its outcome nor by the cancel's (§681).
+const ErrUnanswered = "snapshot-restore unanswered"
+
+// DefaultRestoreWait is how long a restore in place is waited on: past the keeper's own 60 minutes (§681, A-138).
+const DefaultRestoreWait = 90 * time.Minute
+
+func (e *Engine) restoreWait() time.Duration {
+	if e.Settings.RestoreWait > 0 {
+		return e.Settings.RestoreWait
+	}
+	return DefaultRestoreWait
+}
+
+func (e *Engine) incoming() string { return filepath.Join(e.Dir, "incoming") }
+
+// restoreRequest is the hand verb `restore` with `request`, a keeper's snapshot (§681 Its use, §682): an unprotected one is
+// read through snapshot-get into incoming/<request>, refused before the first chunk when it would pass the cap, checked
+// against its digest before any aside, and restored by the module itself as a point is (§516), the file deleted when the
+// act ends; a protected one is restored in place by the keeper (restoreInPlace).
+func (e *Engine) restoreRequest(ctx context.Context, a *Act) error {
+	if e.Keeper == nil {
+		return e.endedRestored(a, errors.New("no keeper is reachable, so no snapshot of it is restored"), nil, false)
+	}
+	e.setStep(a, "receiving the keeper's snapshot", "")
+	name, err := token(a.Request)
+	if err != nil {
+		return e.endedRestored(a, err, nil, false)
+	}
+	if err := os.MkdirAll(e.incoming(), 0o700); err != nil {
+		return e.endedRestored(a, err, nil, false)
+	}
+	file := filepath.Join(e.incoming(), name)
+	defer os.Remove(file)
+	f, err := os.Create(file)
+	if err != nil {
+		return e.endedRestored(a, err, nil, false)
+	}
+	ans, err := e.Keeper.Get(ctx, a.Request, func(ans meshbroker.SnapshotGetAnswer) error {
+		if fill := e.filled(); fill+ans.Size > e.Settings.Cap {
+			return fmt.Errorf("%s: the cap is %d bytes, the snapshot of the request %s needs %d and the directory holds %d",
+				ErrCap, e.Settings.Cap, a.Request, ans.Size, fill)
+		}
+		return nil
+	}, f)
+	if cerr := f.Close(); err == nil {
+		err = cerr
+	}
+	if err != nil {
+		return e.endedRestored(a, err, nil, false)
+	}
+	if ans.Error == snapshotProtected {
+		_ = os.Remove(file)
+		return e.restoreInPlace(ctx, a)
+	}
+	if ans.Error != "" {
+		return e.endedRestored(a, errors.New(ans.Error), nil, false)
+	}
+	// The bytes are checked against the keeper's digest before any aside is taken (§567).
+	if got, n, err := digestOf(file); err != nil || got != ans.Digest || n != ans.Size {
+		return e.endedRestored(a, fmt.Errorf("snapshot damaged: the request %s's bytes received do not match its digest", a.Request), nil, false)
+	}
+	rec, err := e.setAside(ctx, a, []string{ans.Stream}, true)
+	if err != nil {
+		return e.endedRestored(a, err, nil, false)
+	}
+	e.setStep(a, "restoring", "")
+	if err := e.Bus.Restore(ctx, file); err != nil {
+		err = fmt.Errorf("stream %s: %w", ans.Stream, err)
+		if back := e.restoreFromAside(ctx, rec, true); back != nil {
+			err = fmt.Errorf("%w; and the aside is not put back: %v", err, back)
+		}
+		return e.endedRestored(a, err, asidePtr(rec), false)
+	}
+	rec.Step = stepRestored
+	_ = writeJSON(filepath.Join(e.asideDirOf(rec), asideFile), rec)
+	return e.endedRestored(a, nil, asidePtr(rec), false)
+}
+
+// restoreInPlace is a protected snapshot's restore (§681, O-7): snapshot-restore with verify first, then the live data of
+// the restore's module and kept data set aside, its names from the catalog entry of the fingerprint the node runs
+// (A-137), then snapshot-restore, waited on by snapshot-restored of that request and the act it was answered for at most
+// snapshot-restore-wait-minutes (A-138). Its `failed` puts the aside back. At the bound it asks snapshot-restore-cancel and
+// waits one more bound for that act's outcome (A-140); when none comes it puts nothing back and ends failed with
+// `snapshot-restore unanswered`, the aside left at `removed` for aside-unrestored (§561). A keeper answering busy ends the
+// act failed before any aside, naming the keeper's running act (A-140).
+func (e *Engine) restoreInPlace(ctx context.Context, a *Act) error {
+	e.setStep(a, "verifying the keeper's snapshot", "")
+	v, err := e.Keeper.Restore(ctx, a.Request, true, a.Why)
+	if err == nil {
+		err = keeperAnswer(v)
+	}
+	if err != nil {
+		return e.endedRestored(a, err, nil, false)
+	}
+	fp, err := e.Seats.RunningFingerprint(ctx, a.Module)
+	if err != nil {
+		return e.endedRestored(a, err, nil, false)
+	}
+	a.Fingerprint = fp
+	names, err := e.Seats.Streams(ctx, a.Module, a.Kept, fp)
+	if err != nil {
+		return e.endedRestored(a, err, nil, false)
+	}
+	rec, err := e.setAside(ctx, a, names, true)
+	if err != nil {
+		return e.endedRestored(a, err, nil, false)
+	}
+	putBack := func(why error) error {
+		if back := e.restoreFromAside(ctx, rec, true); back != nil {
+			why = fmt.Errorf("%w; and the aside is not put back: %v", why, back)
+		}
+		return e.endedRestored(a, why, asidePtr(rec), false)
+	}
+	e.setStep(a, "restoring in place", "the keeper's snapshot-restored")
+	r, err := e.Keeper.Restore(ctx, a.Request, false, a.Why)
+	if err == nil {
+		err = keeperAnswer(r)
+	}
+	if err != nil {
+		return putBack(err)
+	}
+	act := r.Accepted
+	ev, ok := e.restoredWithin(ctx, a.Request, act)
+	if !ok {
+		e.setStep(a, "cancelling the keeper's restore", "the keeper's snapshot-restored after the cancel")
+		why := fmt.Sprintf("the recovery module's wait of %s on the restore in place passed (act %s)", e.restoreWait(), a.ID)
+		if c, err := e.Keeper.CancelRestore(context.Background(), act, why); err != nil || c.Error != "" {
+			e.logf("the cancel of the keeper's restore %s is not accepted (%v%s); its outcome is waited on once more", act, err, c.Error)
+		}
+		if ev, ok = e.restoredWithin(context.Background(), a.Request, act); !ok {
+			return e.endedRestored(a, errors.New(ErrUnanswered), asidePtr(rec), false)
+		}
+	}
+	if ev.Outcome != meshbroker.SnapshotRestoredEventOutcomeDone {
+		return putBack(fmt.Errorf("the keeper's restore in place failed: %s", ev.Error))
+	}
+	rec.Step = stepRestored
+	_ = writeJSON(filepath.Join(e.asideDirOf(rec), asideFile), rec)
+	return e.endedRestored(a, nil, asidePtr(rec), false)
+}
+
+func (e *Engine) restoredWithin(ctx context.Context, request, act string) (meshbroker.SnapshotRestoredEvent, bool) {
+	wctx, cancel := context.WithTimeout(ctx, e.restoreWait())
+	defer cancel()
+	return e.Keeper.Restored(wctx, request, act)
+}
+
+// keeperAnswer reads snapshot-restore's one form (§691): accepted, or an error naming the keeper's refusal or its running act.
+func keeperAnswer(r meshbroker.SnapshotRestoreAnswer) error {
+	switch {
+	case r.Error != "":
+		return errors.New(r.Error)
+	case r.Busy != "":
+		return fmt.Errorf("busy: the keeper restores its act %s", r.Busy)
+	case r.Accepted == "":
+		return errors.New("the keeper answered snapshot-restore with nothing")
+	}
+	return nil
+}

modules/mesh-recovery/internal/recovery/keeper_test.go

@@ -0,0 +1,206 @@
+package recovery_test
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
+
+	"mesh-recovery/internal/recovery"
+)
+
+// fakeKeeper is the bus module's keeper holding one snapshot of app_notes (the fake bus's file form). in place: how its
+// restore in place ends: "done", "failed", "silent" (it says nothing until cancelled, then failed) or "deaf" (never).
+type fakeKeeper struct {
+	mu        sync.Mutex
+	b         *fakeBus
+	file      string
+	protected bool
+	busy      string
+	damaged   bool
+	inPlace   string
+	asked     []string
+	wrote     int
+	cancelled chan struct{}
+}
+
+func newKeeper(b *fakeBus) *fakeKeeper {
+	return &fakeKeeper{b: b, file: "snap:app_notes:as dropped", inPlace: "done", cancelled: make(chan struct{})}
+}
+
+func (k *fakeKeeper) note(s string) {
+	k.mu.Lock()
+	k.asked = append(k.asked, s)
+	k.mu.Unlock()
+}
+
+func (k *fakeKeeper) Get(_ context.Context, request string, fits func(meshbroker.SnapshotGetAnswer) error, w io.Writer) (meshbroker.SnapshotGetAnswer, error) {
+	k.note("get " + request)
+	if k.protected {
+		return meshbroker.SnapshotGetAnswer{Error: "snapshot protected"}, nil
+	}
+	sum := sha256.Sum256([]byte(k.file))
+	ans := meshbroker.SnapshotGetAnswer{Request: request, Module: "app", Local: "notes", Stream: "app_notes", Size: int64(len(k.file)),
+		Digest: "sha256:" + hex.EncodeToString(sum[:]), Chunks: 1}
+	if err := fits(ans); err != nil {
+		return ans, err
+	}
+	body := k.file
+	if k.damaged {
+		body = "snap:app_notes:altered on its way"
+	}
+	n, err := io.WriteString(w, body)
+	k.wrote += n
+	return ans, err
+}
+
+func (k *fakeKeeper) Restore(_ context.Context, request string, verify bool, _ string) (meshbroker.SnapshotRestoreAnswer, error) {
+	if verify {
+		k.note("verify " + request)
+		return meshbroker.SnapshotRestoreAnswer{Accepted: request}, nil
+	}
+	k.note("restore " + request)
+	if k.busy != "" {
+		return meshbroker.SnapshotRestoreAnswer{Busy: k.busy}, nil
+	}
+	return meshbroker.SnapshotRestoreAnswer{Accepted: "keeper-act-1"}, nil
+}
+
+func (k *fakeKeeper) CancelRestore(_ context.Context, act, _ string) (meshbroker.SnapshotRestoreCancelAnswer, error) {
+	k.note("cancel " + act)
+	close(k.cancelled)
+	return meshbroker.SnapshotRestoreCancelAnswer{Accepted: act}, nil
+}
+
+func (k *fakeKeeper) Restored(ctx context.Context, request, act string) (meshbroker.SnapshotRestoredEvent, bool) {
+	ev := meshbroker.SnapshotRestoredEvent{Request: request, Act: act, Outcome: "failed", Error: "the server refused the archive"}
+	switch k.inPlace {
+	case "done":
+		k.b.mu.Lock()
+		if _, live := k.b.streams["app_notes"]; live {
+			k.b.mu.Unlock()
+			return meshbroker.SnapshotRestoredEvent{Request: request, Act: act, Outcome: "failed", Error: "names live"}, true
+		}
+		k.b.streams["app_notes"] = "as dropped"
+		k.b.mu.Unlock()
+		ev.Outcome, ev.Error = "done", ""
+		return ev, true
+	case "failed":
+		return ev, true
+	case "silent":
+		select {
+		case <-k.cancelled:
+			ev.Error = "cancelled"
+			return ev, true
+		case <-ctx.Done():
+			return ev, false
+		}
+	}
+	<-ctx.Done()
+	return ev, false
+}
+
+func requestRig(t *testing.T) (*rig, *fakeKeeper) {
+	r := newRig(t)
+	k := newKeeper(r.b)
+	r.e.Keeper = k
+	r.e.Settings.RestoreWait = 20 * time.Millisecond
+	return r, k
+}
+
+func requestAct() recovery.Act {
+	return recovery.Act{Kind: recovery.KindRestore, Module: "app", Kept: "notes", Request: "req-1", Why: "test the drop's snapshot", Caller: "tool-runner", Asker: "op@node"}
+}
+
+// A keeper's snapshot by request is read through snapshot-get into incoming/<request>, refused before the first chunk past
+// the cap, checked against its digest before any aside, restored by the module itself, and its file deleted at the end
+// (design 47 G §681 Its use, §682).
+func TestARestoreByRequestReadsTheKeepersSnapshot(t *testing.T) {
+	for _, tc := range []struct {
+		name    string
+		cap     int64
+		damaged bool
+		err     string
+	}{
+		{name: "received and restored", cap: 1 << 20},
+		{name: "altered on its way", cap: 1 << 20, damaged: true, err: "snapshot damaged"},
+		{name: "past the cap", cap: 10, err: "cap:"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			r, k := requestRig(t)
+			r.e.Settings.Cap, k.damaged = tc.cap, tc.damaged
+			ans := r.do(requestAct())
+			ev := lastRestored(t, r)
+			if ev.Act != ans.Accepted || !strings.HasPrefix(ev.Error, tc.err) || (tc.err == "") != (ev.Outcome == "done") {
+				t.Fatalf("restored %+v, want error %q", ev, tc.err)
+			}
+			if left, _ := os.ReadDir(filepath.Join(r.e.Dir, "incoming")); len(left) != 0 {
+				t.Errorf("incoming holds %v after the act", left)
+			}
+			kp, _ := r.e.KeptPoints("app", "notes")
+			if tc.err != "" {
+				if r.b.streams["app_notes"] != "hello" || len(kp.Asides) != 0 {
+					t.Errorf("a refused restore touched the live data: %v, asides %+v", r.b.streams, kp.Asides)
+				}
+				if tc.name == "past the cap" && k.wrote != 0 {
+					t.Errorf("%d bytes received past the cap", k.wrote)
+				}
+				return
+			}
+			if r.b.streams["app_notes"] != "as dropped" || r.b.streams["KV_app_seen"] != "world" || len(kp.Asides) != 1 ||
+				ev.Aside == nil || len(ev.Aside.Names) != 1 || ev.Aside.Names[0] != "app_notes" {
+				t.Errorf("streams %v, asides %+v, event aside %+v", r.b.streams, kp.Asides, ev.Aside)
+			}
+		})
+	}
+}
+
+// A protected snapshot is verified, the running data of the module and kept data set aside (A-137), and restored in place
+// by the keeper; its failure puts the aside back; at the bound the module cancels first and puts the aside back on the
+// cancel's outcome, and with no outcome at all ends `snapshot-restore unanswered`, the aside left unrestored; a busy keeper
+// ends the act before any aside (§681, A-138, A-140).
+func TestAProtectedSnapshotIsRestoredInPlaceByTheKeeper(t *testing.T) {
+	for _, tc := range []struct {
+		inPlace, busy string
+		err           string
+		asked         string
+		live          string
+		unrestored    int
+	}{
+		{inPlace: "done", asked: "get req-1|verify req-1|restore req-1", live: "as dropped"},
+		{inPlace: "failed", asked: "get req-1|verify req-1|restore req-1", live: "hello", err: "the keeper's restore in place failed"},
+		{inPlace: "silent", asked: "get req-1|verify req-1|restore req-1|cancel keeper-act-1", live: "hello", err: "the keeper's restore in place failed: cancelled"},
+		{inPlace: "deaf", asked: "get req-1|verify req-1|restore req-1|cancel keeper-act-1", err: recovery.ErrUnanswered, unrestored: 1},
+		{inPlace: "done", busy: "keeper-act-0", asked: "get req-1|verify req-1|restore req-1", live: "hello", err: "busy: the keeper restores its act keeper-act-0"},
+	} {
+		t.Run(tc.inPlace+tc.busy, func(t *testing.T) {
+			r, k := requestRig(t)
+			k.protected, k.inPlace, k.busy = true, tc.inPlace, tc.busy
+			r.do(requestAct())
+			ev := lastRestored(t, r)
+			if !strings.HasPrefix(ev.Error, tc.err) || (tc.err == "") != (ev.Outcome == "done") || ev.Fingerprint != "fp-run" {
+				t.Fatalf("restored %+v, want error %q", ev, tc.err)
+			}
+			if got := strings.Join(k.asked, "|"); got != tc.asked {
+				t.Errorf("asked %s, want %s", got, tc.asked)
+			}
+			if r.b.streams["app_notes"] != tc.live {
+				t.Errorf("app_notes is %q, want %q", r.b.streams["app_notes"], tc.live)
+			}
+			if tc.live == "hello" && r.b.streams["KV_app_seen"] != "world" {
+				t.Errorf("the second name of the running data is not put back: %v", r.b.streams)
+			}
+			if n := len(r.e.Unrestored()); n != tc.unrestored {
+				t.Errorf("%d unrestored asides, want %d", n, tc.unrestored)
+			}
+		})
+	}
+}

modules/mesh-recovery/internal/recovery/manifest_test.go

@@ -70,11 +70,17 @@ func TestTheManifestSaysWhatTheModuleIs(t *testing.T) {
 			t.Errorf("no data item %q", id)
 		}
 	}
-	for k, want := range map[string]float64{"points-cap": 10 << 30, "points-filling-mark": 80, "point-keep-days": 90, "dump-wait-minutes": 60} {
+	for k, want := range map[string]float64{"points-cap": 10 << 30, "points-filling-mark": 80, "point-keep-days": 90, "dump-wait-minutes": 60, "snapshot-restore-wait-minutes": 90} {
 		if m.Config[k].Default != want {
 			t.Errorf("setting %s defaults to %v, want %v", k, m.Config[k].Default, want)
 		}
 	}
+	// The keeper's snapshot verbs the recovery module asks, the cancel at its bound among them (§681).
+	for _, inv := range []string{"seat:mesh-broker.snapshot-get", "seat:mesh-broker.snapshot-restore", "seat:mesh-broker.snapshot-restore-cancel"} {
+		if !slices.Contains(m.Invokes, inv) {
+			t.Errorf("does not invoke %s", inv)
+		}
+	}
 	for _, r := range m.Resources {
 		if r.Type != "process" {
 			continue

modules/mesh-recovery/internal/recovery/recovery.go

@@ -104,15 +104,20 @@ type Settings struct {
 	FillingMark int
 	// DumpWait is how long an act waits on the dump before recovery-waits is written (dump-wait-minutes, 60 unless set).
 	DumpWait time.Duration
+	// RestoreWait is how long a restore in place is waited on, and once more after its cancel (snapshot-restore-wait-minutes,
+	// 90 unless set; §681, A-138, A-140).
+	RestoreWait time.Duration
 }
 
 // Engine is the recovery module's one engine.
 type Engine struct {
-	Dir      string
-	Bus      Bus
-	Seats    Seats
-	Digests  Digests
-	Dump     Dump
+	Dir     string
+	Bus     Bus
+	Seats   Seats
+	Digests Digests
+	Dump    Dump
+	// Keeper is the bus module's keeper, whose snapshots the hand verb restore takes by request (§681).
+	Keeper   Keeper
 	Settings Settings
 	// Findings is the module's finding writer (§561); Node the machine it runs on, the id of points-filling.
 	Findings Findings
@@ -130,24 +135,26 @@ type Engine struct {
 
 // 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"`
-	Error       string        `json:"error,omitempty"`
+	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"`
+	// Request is a keeper's snapshot by its request, the hand verb restore's other source (§681, §682).
+	Request    string        `json:"request,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"`
+	Error      string        `json:"error,omitempty"`
 }
 
 // Answer is the one form every acting verb answers (§691): exactly one field set.

modules/mesh-recovery/internal/recovery/restore.go

@@ -70,7 +70,7 @@ func (e *Engine) cancelledWhy(err error) string {
 	return ""
 }
 
-// restore is `restore-kept` and the hand verb `restore` with a point (§516, §554, §563, §567): it says restore-started,
+// restore is `restore-kept` and the hand verb `restore` with a point (or, past the dump, a keeper's snapshot by request) (§516, §554, §563, §567): it says restore-started,
 // waits for the dump of the bus's item, finds the point (or has the backup holder bring it back), checks its digests
 // against the bus, sets the live data aside whole and checked, removes it, restores the point under its names and
 // sets each consumer to its recorded sequence; a step after the removal that fails puts the aside back.
@@ -91,6 +91,9 @@ func (e *Engine) restore(ctx context.Context, a *Act) error {
 			return e.endedRestored(a, err, nil, false)
 		}
 	}
+	if a.Request != "" {
+		return e.restoreRequest(ctx, a)
+	}
 	e.setStep(a, "finding the point", "")
 	pointPath := filepath.Join(e.Dir, "points", filepath.FromSlash(a.Snapshot))
 	if _, err := os.Stat(filepath.Join(pointPath, pointFile)); err != nil {

modules/mesh-recovery/internal/recovery/verbs.go

@@ -46,10 +46,10 @@ func (e *Engine) Handlers() map[string]process.SeatHandler {
 		if (rs.Snapshot == "") == (rs.Request == "") {
 			return Act{}, errors.New("restore takes exactly one of snapshot and request")
 		}
-		if rs.Request != "" {
-			return Act{}, errors.New("the restore of a keeper's snapshot by request is not built yet (design 47 G §681)")
+		if rs.Request != "" && rs.FromBackup {
+			return Act{}, errors.New("from-backup is taken only with snapshot, a restore point (design 47 G §682)")
 		}
-		return Act{Kind: KindRestore, Module: rs.Module, Kept: rs.Kept, Snapshot: rs.Snapshot, FromBackup: rs.FromBackup, Why: rs.Why, Asker: rs.Asker}, nil
+		return Act{Kind: KindRestore, Module: rs.Module, Kept: rs.Kept, Snapshot: rs.Snapshot, Request: rs.Request, FromBackup: rs.FromBackup, Why: rs.Why, Asker: rs.Asker}, nil
 	})
 	h[seat.VerbAside] = actOf(e, func(ad seat.AsideInput) (Act, error) {
 		return Act{Kind: KindAside, Module: ad.Module, Kept: ad.Kept, Why: ad.Why, Asker: ad.Asker}, nil

modules/mesh-recovery/module.json

@@ -45,6 +45,7 @@
     "seat:mesh-hand-acts.hand-act",
     "seat:mesh-broker.snapshot-get",
     "seat:mesh-broker.snapshot-restore",
+    "seat:mesh-broker.snapshot-restore-cancel",
     "seat:node-backup.restore",
     "seat:mesh-recovery.kept-points",
     "seat:mesh-recovery.restore-aside",
@@ -113,6 +114,14 @@
       "maximum": 10080,
       "trusted": false,
       "default": 60
+    },
+    "snapshot-restore-wait-minutes": {
+      "type": "number",
+      "description": "minutes a restore in place of a protected keeper's snapshot is waited on, and once more after its cancel (design 47 G section 681, A-138, A-140).",
+      "minimum": 1,
+      "maximum": 10080,
+      "trusted": false,
+      "default": 90
     }
   },
   "data": {
@@ -185,6 +194,14 @@
       "mode": "0600",
       "template": "files/dump-wait-minutes"
     },
+    {
+      "id": "restore-wait",
+      "type": "file",
+      "path": "${dir:state}/snapshot-restore-wait-minutes",
+      "owner": "mesh-recovery",
+      "mode": "0600",
+      "template": "files/snapshot-restore-wait-minutes"
+    },
     {
       "id": "recovery",
       "type": "process",
@@ -199,7 +216,8 @@
         "MESH_RECOVERY_POINTS_CAP_FILE": "${dir:state}/points-cap",
         "MESH_RECOVERY_FILLING_MARK_FILE": "${dir:state}/points-filling-mark",
         "MESH_RECOVERY_KEEP_DAYS_FILE": "${dir:state}/point-keep-days",
-        "MESH_RECOVERY_DUMP_WAIT_FILE": "${dir:state}/dump-wait-minutes"
+        "MESH_RECOVERY_DUMP_WAIT_FILE": "${dir:state}/dump-wait-minutes",
+        "MESH_RECOVERY_RESTORE_WAIT_FILE": "${dir:state}/snapshot-restore-wait-minutes"
       },
       "user": "mesh-recovery",
       "serves": [
@@ -222,6 +240,7 @@
           "seat:mesh-hand-acts.hand-act",
           "seat:mesh-broker.snapshot-get",
           "seat:mesh-broker.snapshot-restore",
+          "seat:mesh-broker.snapshot-restore-cancel",
           "seat:node-backup.restore"
         ]
       },

modules/nats/cmd/nats-tools/snapshots.go

@@ -19,6 +19,7 @@ import (
 
 	"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
 	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/buscalls"
 	"git.novox.be/novox/mesh-sdk/go/process"
 
 	"nats-tools/internal/keeper"
@@ -30,34 +31,32 @@ import (
 // snapshot-restore, snapshot-restore-cancel and snapshot-remove are acted on by the keeper at once, with no hand-over through the
 // state requests. The tools bundle keeps nats_snapshots, which lists them.
 
-// ChunkSize is at most what one chunk carries: 512 KiB.
-const ChunkSize = 512 << 10
-
-// AckWithin is how long the next chunk waits for its acknowledgement.
-const AckWithin = 30 * time.Second
+// The form of the chunks is mesh-sdk's (buscalls.SeatChunks, A-139): the caller receives with the same one.
+const (
+	// ChunkSize is at most what one chunk carries: 512 KiB.
+	ChunkSize = buscalls.SeatChunkBytes
+	// AckWithin is how long the next chunk waits for its acknowledgement.
+	AckWithin = buscalls.SeatAckWithin
+)
 
 // callToken is the shape of the subject token a caller chooses for a send.
 var callToken = regexp.MustCompile(`^[A-Za-z0-9_-]{16,64}$`)
 
-// ChunkSubject and AckSubject are the subjects of the nth chunk of a send and of its acknowledgement.
-func ChunkSubject(call string, n int64) string {
-	return "mesh.seat." + meshbroker.Seat + ".chunk." + meshbroker.VerbSnapshotGet + "." + call + "." + strconv.FormatInt(n, 10)
+func sendOf(call string) buscalls.SeatChunks {
+	return buscalls.SeatChunks{Seat: meshbroker.Seat, Verb: meshbroker.VerbSnapshotGet, Call: call}
 }
 
+// ChunkSubject and AckSubject are the subjects of the nth chunk of a send and of its acknowledgement.
+func ChunkSubject(call string, n int64) string { return sendOf(call).Chunk(n) }
+
 // AckSubject is the subject the caller acknowledges the nth chunk on.
-func AckSubject(call string, n int64) string {
-	return "mesh.seat." + meshbroker.Seat + ".ack." + meshbroker.VerbSnapshotGet + "." + call + "." + strconv.FormatInt(n, 10)
-}
+func AckSubject(call string, n int64) string { return sendOf(call).Ack(n) }
 
 // AckPattern is every acknowledgement of snapshot-get.
-const AckPattern = "mesh.seat.mesh-broker.ack.snapshot-get.*.*"
+var AckPattern = buscalls.SeatAcks(meshbroker.Seat, meshbroker.VerbSnapshotGet)
 
 // Chunk is the body of one chunk: its number, how many there are, and its bytes.
-type Chunk struct {
-	N    int64  `json:"n"`
-	Of   int64  `json:"of"`
-	Data []byte `json:"data"`
-}
+type Chunk = buscalls.SeatChunk
 
 // snapshotWorld is what the keeper's snapshot verbs reach: the file, the process's own publish and subscription, and the bus's users.
 type snapshotWorld struct {