bus5 B2 (§681): the keeper restores a snapshot in place (verify, names live, 60-minute bound, cancel, partial stream removed, snapshot-restored); requests keeps records 91 days (A-53)

Builds: design 47 G §681, ADR 0675, lead answers A-45, A-53

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

@@ -118,7 +118,11 @@ func (s processStore) Create(state, key string, value []byte) (uint64, error) {
 // keeps it once (the two publishes are not one act: a failed second one is kept and said again, never silent).
 func streamEmitter(js jetstream.JetStream, a process.Account) func(*keeper.Saying) error {
 	return func(s *keeper.Saying) error {
-		data, err := json.Marshal(s.Body)
+		var body any = s.Body
+		if s.Restored != nil {
+			body = s.Restored
+		}
+		data, err := json.Marshal(body)
 		if err != nil {
 			return err
 		}
@@ -132,6 +136,10 @@ func streamEmitter(js jetstream.JetStream, a process.Account) func(*keeper.Sayin
 		subjects := s.Left
 		if subjects == nil {
 			subjects = EventSubjects(a, s.Event)
+			if s.Restored != nil {
+				// snapshot-restored has its own subject, under the request and the act (§681).
+				subjects = []string{meshbroker.SubjectSnapshotRestored(s.Restored.Request, s.Restored.Act)}
+			}
 		}
 		s.Said = nil
 		for i, subject := range subjects {
@@ -244,7 +252,8 @@ func runKeeper(ctx context.Context, getenv func(string) string, logOut io.Writer
 	defer nc.Close()
 	recorder := meshhandactsclient.Recorder{Ask: conn.Call}
 	k := &keeper.Keeper{
-		Bus:   keeper.JetStream{JS: js, Snap: bussnap.Into(nc, getenv("MESH_NATS_SNAPSHOTS_DIR"))},
+		Bus: keeper.JetStream{JS: js, Snap: bussnap.Into(nc, getenv("MESH_NATS_SNAPSHOTS_DIR")), NC: nc,
+			Dir: getenv("MESH_NATS_SNAPSHOTS_DIR"), Log: logger.Printf},
 		Store: processStore{conn}, Emit: streamEmitter(js, account), Say: sayer(js), View: firstView(conn),
 		Act: func(what, why, cause string) error {
 			_, err := recorder.Record(meshhandactsclient.Request{What: what, Why: why, Cause: cause})
@@ -303,6 +312,7 @@ func Keep(ctx context.Context, k *keeper.Keeper, hear func(context.Context, func
 	defer tick.Stop()
 	k.FillPass()
 	k.SnapshotsAtStart()
+	k.EndCutOffRestores()
 	var wait time.Duration
 	for {
 		next, err := k.AskFirstView(wait)

modules/nats/internal/bussnap/restore.go

@@ -0,0 +1,158 @@
+package bussnap
+
+import (
+	"archive/tar"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"time"
+
+	"nats-tools/internal/busarchive"
+
+	"github.com/nats-io/nats.go"
+)
+
+// The keeper's restore in place of a snapshot it took (design 47 G §681): each stream of the file put back under its own
+// name, only where no stream of that name is live.
+
+// APIStreamDelete is the subject the keeper removes its own partial stream with.
+const APIStreamDelete = "$JS.API.STREAM.DELETE."
+
+const restoreChunk = 64 * 1024
+
+// ManifestOf reads the manifest of a snapshot file: the streams it holds, without unpacking them.
+func ManifestOf(path string) (*busarchive.Manifest, error) {
+	f, err := os.Open(path)
+	if err != nil {
+		return nil, err
+	}
+	defer f.Close()
+	tr := tar.NewReader(f)
+	hdr, err := tr.Next()
+	if err != nil {
+		return nil, fmt.Errorf("the snapshot does not read: %w", err)
+	}
+	if hdr.Name != busarchive.ManifestName {
+		return nil, fmt.Errorf("the snapshot starts with %q, not its manifest", hdr.Name)
+	}
+	raw, err := io.ReadAll(io.LimitReader(tr, 64<<20))
+	if err != nil {
+		return nil, err
+	}
+	var m busarchive.Manifest
+	if err := json.Unmarshal(raw, &m); err != nil {
+		return nil, fmt.Errorf("its manifest: %w", err)
+	}
+	return &m, nil
+}
+
+// Live says whether a stream of that name exists.
+func Live(ctx context.Context, nc *nats.Conn, name string) (bool, error) {
+	err := Request(ctx, nc, APIStreamInfo+name, nil, nil)
+	var api *APIError
+	switch {
+	case err == nil:
+		return true, nil
+	case errors.As(err, &api) && api.Code == 404:
+		return false, nil
+	}
+	return false, err
+}
+
+// DeleteStream removes a stream; one that is not there is no error.
+func DeleteStream(ctx context.Context, nc *nats.Conn, name string) error {
+	err := Request(ctx, nc, APIStreamDelete+name, nil, nil)
+	var api *APIError
+	if errors.As(err, &api) && api.Code == 404 {
+		return nil
+	}
+	return err
+}
+
+// RestoreFile puts every stream of the snapshot file back under its own name, each sent in chunks the server answers; it
+// ends with the context's bound. A stream already there is the caller's to have refused beforehand.
+func RestoreFile(ctx context.Context, nc *nats.Conn, path, scratch string, log func(string, ...any)) error {
+	f, err := os.Open(path)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	a, err := busarchive.ReadArchive(f, scratch)
+	if err != nil {
+		return err
+	}
+	defer a.Close()
+	if err := a.Verify(); err != nil {
+		return err
+	}
+	for _, s := range a.Manifest.Streams {
+		started := time.Now()
+		if err := restoreStream(ctx, nc, a, s); err != nil {
+			return fmt.Errorf("stream %s: %w", s.Name, err)
+		}
+		log("%s: restored in %.2fs", s.Name, time.Since(started).Seconds())
+	}
+	return nil
+}
+
+func restoreStream(ctx context.Context, nc *nats.Conn, a *busarchive.Archive, s busarchive.StreamEntry) error {
+	meta, archive, err := a.Open(s)
+	if err != nil {
+		return err
+	}
+	defer archive.Close()
+	var req map[string]json.RawMessage
+	if err := json.Unmarshal(meta, &req); err != nil || req["config"] == nil {
+		return errors.New("its meta is not a stream's configuration and state")
+	}
+	var resp struct {
+		DeliverSubject string `json:"deliver_subject"`
+	}
+	if err := Request(ctx, nc, APIStreamRestore+s.Name, req, &resp); err != nil {
+		return err
+	}
+	if resp.DeliverSubject == "" {
+		return errors.New("the server gave nowhere to send the archive")
+	}
+	buf := make([]byte, restoreChunk)
+	for {
+		n, err := archive.Read(buf)
+		if n > 0 {
+			reply, rerr := nc.RequestWithContext(ctx, resp.DeliverSubject, buf[:n])
+			if rerr != nil {
+				return fmt.Errorf("sending the archive: %w", rerr)
+			}
+			if len(reply.Data) > 0 {
+				return fmt.Errorf("the server refused the archive: %s", Truncate(reply.Data))
+			}
+		}
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			return err
+		}
+	}
+	final, err := nc.RequestWithContext(ctx, resp.DeliverSubject, nil)
+	if err != nil {
+		return fmt.Errorf("finishing the restore: %w", err)
+	}
+	var created struct {
+		Error *APIError   `json:"error"`
+		State StreamState `json:"state"`
+	}
+	if err := json.Unmarshal(final.Data, &created); err != nil {
+		return fmt.Errorf("the server's last answer is not JSON: %q", Truncate(final.Data))
+	}
+	if created.Error != nil {
+		return created.Error
+	}
+	if created.State.LastSeq < s.LastSeq || (s.Messages > 0 && created.State.Messages == 0) {
+		return fmt.Errorf("restored with %d message(s) up to sequence %d; the snapshot held %d up to %d",
+			created.State.Messages, created.State.LastSeq, s.Messages, s.LastSeq)
+	}
+	return nil
+}

modules/nats/internal/keeper/jetstream.go

@@ -7,7 +7,10 @@ import (
 	"strings"
 	"time"
 
+	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
+
+	"nats-tools/internal/bussnap"
 )
 
 // JetStream is the bus as the keeper's own account reaches it (Bus): each object a stream, a key-value bucket or an
@@ -16,6 +19,10 @@ type JetStream struct {
 	JS jetstream.JetStream
 	// Snap takes a snapshot of streams and says where it is kept (bussnap through the keeper's own connection).
 	Snap func(ctx context.Context, streams []string) (string, error)
+	// NC and Dir are the connection and the scratch directory the keeper's restore in place of a snapshot uses (design 47 G §681).
+	NC  *nats.Conn
+	Dir string
+	Log func(string, ...any)
 }
 
 // Held reads the object's stream.
@@ -178,3 +185,35 @@ func (b JetStream) Restricted(ctx context.Context, streams []string) (bool, erro
 	}
 	return false, nil
 }
+
+// SnapshotStreams reads the streams a snapshot file holds from its manifest.
+func (b JetStream) SnapshotStreams(path string) ([]string, error) {
+	m, err := bussnap.ManifestOf(path)
+	if err != nil {
+		return nil, err
+	}
+	var out []string
+	for _, s := range m.Streams {
+		out = append(out, s.Name)
+	}
+	return out, nil
+}
+
+// Live says whether a stream of that name exists.
+func (b JetStream) Live(ctx context.Context, name string) (bool, error) {
+	return bussnap.Live(ctx, b.NC, name)
+}
+
+// Restore puts the snapshot's streams back under their own names.
+func (b JetStream) Restore(ctx context.Context, path string) error {
+	log := b.Log
+	if log == nil {
+		log = func(string, ...any) {}
+	}
+	return bussnap.RestoreFile(ctx, b.NC, path, b.Dir, log)
+}
+
+// DeleteStream removes the stream by its name.
+func (b JetStream) DeleteStream(ctx context.Context, name string) error {
+	return bussnap.DeleteStream(ctx, b.NC, name)
+}

modules/nats/internal/keeper/keeper.go

@@ -151,6 +151,9 @@ type Saying struct {
 	Said  []string  `json:"said,omitempty"`
 	Error string    `json:"error,omitempty"`
 	At    time.Time `json:"at"`
+	// Restored, when set, is the body of the event snapshot-restored, said on its own subject (request and act) and not on
+	// the entry events' ones.
+	Restored *SnapshotRestored `json:"restored,omitempty"`
 }
 
 // FirstViewStatus is what the keeper holds of its first view.
@@ -183,6 +186,13 @@ type Bus interface {
 	// Restricted says whether any of the streams is one ADR 0436 §3 lets no other account hear (its metadata says it), so
 	// a snapshot of them is never handed out (design 47 G §681).
 	Restricted(ctx context.Context, streams []string) (bool, error)
+	// SnapshotStreams are the streams a snapshot file holds; Live says whether a stream of that name exists.
+	SnapshotStreams(path string) ([]string, error)
+	Live(ctx context.Context, name string) (bool, error)
+	// Restore puts every stream of the snapshot file back under its own name, and DeleteStream removes one by its name (the
+	// keeper's own partial stream of a restore that did not end).
+	Restore(ctx context.Context, path string) error
+	DeleteStream(ctx context.Context, name string) error
 }
 
 // Fill is how much one object holds: its bytes and messages.
@@ -237,6 +247,8 @@ type Keeper struct {
 	unwritten bool
 	// filling is whether the snapshots directory was last judged past 80 % of its cap, fillJudged whether it is to be judged afresh (the keeper's start).
 	filling, fillUnjudged bool
+	// restoring is the snapshot restore that runs, at most one.
+	restoring *restoreRun
 }
 
 func (k *Keeper) now() time.Time {

modules/nats/internal/keeper/keeper_test.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"os"
 	"sort"
 	"strings"
 	"sync"
@@ -30,6 +31,45 @@ type fakeBus struct {
 	snaps   [][]string
 	// restricted names the streams whose metadata says they are restricted.
 	restricted map[string]bool
+	// snapStreams are the streams each snapshot file holds; restored the files restored; gate holds a restore until it is
+	// closed or its context ends; deletedStreams the streams removed by name.
+	snapStreams    map[string][]string
+	restored       []string
+	gate           chan struct{}
+	deletedStreams []string
+}
+
+func (b *fakeBus) SnapshotStreams(path string) ([]string, error) {
+	return b.snapStreams[path], nil
+}
+
+func (b *fakeBus) Live(_ context.Context, name string) (bool, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	_, ok := b.held[name]
+	return ok, nil
+}
+
+func (b *fakeBus) Restore(ctx context.Context, path string) error {
+	b.mu.Lock()
+	gate, err := b.gate, b.refuse["restore"]
+	b.restored = append(b.restored, path)
+	b.mu.Unlock()
+	if gate != nil {
+		select {
+		case <-gate:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+	return err
+}
+
+func (b *fakeBus) DeleteStream(_ context.Context, name string) error {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.deletedStreams = append(b.deletedStreams, name)
+	return nil
 }
 
 func newFakeBus() *fakeBus {
@@ -95,9 +135,13 @@ func (b *fakeBus) Restricted(_ context.Context, streams []string) (bool, error)
 type memFiles struct {
 	size    map[string]int64
 	removed []string
+	missing map[string]bool
 }
 
 func (f *memFiles) Digest(path string) (int64, string, error) {
+	if f.missing[path] {
+		return 0, "", os.ErrNotExist
+	}
 	n, ok := f.size[path]
 	if !ok {
 		n = 100
@@ -185,11 +229,21 @@ type said struct {
 	down, partly bool
 	// left are the subjects each try was asked to say an event on (nil: every one).
 	left [][]string
+	// restored are the snapshot-restored events said, and sig is told of each.
+	restored []string
+	sig      chan struct{}
 }
 
 func (s *said) emit(e *Saying) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
+	if e.Restored != nil {
+		s.restored = append(s.restored, fmt.Sprintf("%s %s %s %s", e.Restored.Request, e.Restored.Act, e.Restored.Outcome, e.Restored.Error))
+		if s.sig != nil {
+			s.sig <- struct{}{}
+		}
+		return nil
+	}
 	s.left = append(s.left, e.Left)
 	if s.down {
 		e.Left, e.Said = []string{"seat", "history"}, nil
@@ -692,6 +746,90 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) {
 				t.Error("a snapshot of before G was removed by age")
 			}
 		}},
+		// Design 47 G §681: the keeper restores a snapshot it took in place: it checks the file against its digest, refuses a
+		// name that is live, restores within a bound, one at a time, cancels, removes its own partial stream of one that failed
+		// and says snapshot-restored under the request and the act; a restore a restart cut off is ended so at the start.
+		{"a snapshot is restored in place, verified, refused while its names are live, cancelled, and its failure removes the partial stream", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
+			store, files := k.Store.(*memStore), k.Files.(*memFiles)
+			out.sig = make(chan struct{}, 8)
+			path := "/snapshots/KV_m_drafts.tar"
+			bus.snapStreams = map[string][]string{path: {"KV_m_drafts"}}
+			files.missing = map[string]bool{"/snapshots/gone.tar": true}
+			digest := "sha256:" + hex.EncodeToString([]byte(path))
+			put := func(r Request) {
+				raw, _ := json.Marshal(r)
+				_, _ = store.Put(StateRequests, r.ID, raw)
+			}
+			get := func(id string) Request {
+				raw, _, _, _ := store.Get(StateRequests, id)
+				var got Request
+				_ = json.Unmarshal(raw, &got)
+				return got
+			}
+			ask := func(r Request) Request {
+				r.Caller = "person.jo"
+				put(r)
+				k.ActOnRequests()
+				return get(r.ID)
+			}
+			put(Request{ID: "r20", Verb: VerbDrop, Path: path, Size: 100, Digest: digest, Protected: true, At: k.now(), Done: k.now()})
+			put(Request{ID: "r21", Verb: VerbDrop, Path: path, Size: 100, Digest: "sha256:bad", At: k.now(), Done: k.now()})
+			put(Request{ID: "r22", Verb: VerbDrop, Path: "/snapshots/gone.tar", Size: 100, Digest: digest, At: k.now(), Done: k.now()})
+			put(Request{ID: "r23", Verb: VerbDrop, Snapshot: "/before/g.tar", At: k.now(), Done: k.now()})
+			for _, row := range []struct{ id, of, want string }{{"x1", "r21", "snapshot damaged"}, {"x2", "r22", "snapshot missing"},
+				{"x3", "r23", "snapshot taken before milestone G"}, {"x4", "r99", "no snapshot of the request r99"}} {
+				if got := ask(Request{ID: row.id, Verb: VerbSnapshotRestore, Of: row.of, Why: "w"}); got.Outcome != "error" || !strings.Contains(got.Error, row.want) {
+					t.Errorf("%s: ended %+v, want %q", row.of, got, row.want)
+				}
+			}
+			if got := ask(Request{ID: "v1", Verb: VerbSnapshotRestore, Of: "r20", Verify: true, Why: "w"}); got.Outcome != "verified" || len(bus.restored) != 0 {
+				t.Errorf("verify: %+v, restored %v", got, bus.restored)
+			}
+			bus.held["KV_m_drafts"] = Settings{}
+			if got := ask(Request{ID: "n1", Verb: VerbSnapshotRestore, Of: "r20", Why: "w"}); got.Outcome != "error" || !strings.Contains(got.Error, "names live") || len(bus.restored) != 0 {
+				t.Errorf("a live name: %+v, restored %v", got, bus.restored)
+			}
+			delete(bus.held, "KV_m_drafts")
+			// A restore that ends well says so under the request and the act.
+			if got := ask(Request{ID: "a1", Verb: VerbSnapshotRestore, Of: "r20", Why: "w"}); got.Outcome != "accepted" || got.ActID != "a1" {
+				t.Fatalf("accepted: %+v", got)
+			}
+			<-out.sig
+			if fmt.Sprint(out.restored) != "[r20 a1 done ]" || fmt.Sprint(bus.restored) != "["+path+"]" || len(bus.deletedStreams) != 0 || get("a1").Result != "done" {
+				t.Errorf("a restore: said %v, restored %v, removed %v, record %+v", out.restored, bus.restored, bus.deletedStreams, get("a1"))
+			}
+			// One that fails removes the keeper's partial stream, then says failed.
+			bus.refuse["restore"] = errors.New("boom")
+			ask(Request{ID: "a2", Verb: VerbSnapshotRestore, Of: "r20", Why: "w"})
+			<-out.sig
+			if got := out.restored[1]; got != "r20 a2 failed boom" || fmt.Sprint(bus.deletedStreams) != "[KV_m_drafts]" {
+				t.Errorf("a failed restore: said %q, removed %v", got, bus.deletedStreams)
+			}
+			// One at a time, and a running one cancelled by its act.
+			delete(bus.refuse, "restore")
+			bus.gate = make(chan struct{})
+			ask(Request{ID: "a3", Verb: VerbSnapshotRestore, Of: "r20", Why: "w"})
+			if got := ask(Request{ID: "a4", Verb: VerbSnapshotRestore, Of: "r20", Why: "w"}); got.Outcome != "busy" || got.ActID != "a3" {
+				t.Errorf("a second restore: %+v", got)
+			}
+			if got := ask(Request{ID: "c1", Verb: VerbSnapshotRestoreCancel, Of: "a3", Why: "w"}); got.Outcome != "accepted" || got.ActID != "a3" {
+				t.Errorf("cancel: %+v", got)
+			}
+			<-out.sig
+			if got := out.restored[2]; got != "r20 a3 failed cancelled" || len(bus.deletedStreams) != 2 {
+				t.Errorf("a cancelled restore: said %q, removed %v", got, bus.deletedStreams)
+			}
+			if got := ask(Request{ID: "c2", Verb: VerbSnapshotRestoreCancel, Of: "a3", Why: "w"}); got.Outcome != "error" {
+				t.Errorf("cancel of an ended act: %+v", got)
+			}
+			// A restore a restart cut off is ended at the keeper's start.
+			put(Request{ID: "a5", Verb: VerbSnapshotRestore, Of: "r20", Outcome: "accepted", Done: k.now()})
+			k.EndCutOffRestores()
+			<-out.sig
+			if got := out.restored[3]; !strings.HasPrefix(got, "r20 a5 failed cut off") || get("a5").Result != "failed" || len(bus.deletedStreams) != 3 {
+				t.Errorf("a cut-off restore: said %q, record %+v, removed %v", got, get("a5"), bus.deletedStreams)
+			}
+		}},
 		{"a recreate rebuilds a drifted entry after a snapshot", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
 			store := k.Store.(*memStore)
 			k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{log}})

modules/nats/internal/keeper/requests.go

@@ -51,6 +51,14 @@ type Request struct {
 	Size      int64  `json:"size,omitempty"`
 	Digest    string `json:"digest,omitempty"`
 	Protected bool   `json:"protected,omitempty"`
+	// Of is the keeper's snapshot a snapshot-restore or snapshot-remove names, by its request; Verify checks only; ActID is the
+	// restore's act (the id of the running one for a busy answer); Ended is when the restore ended, and Result how: done or failed
+	// (design 47 G §681).
+	Of     string    `json:"of,omitempty"`
+	Verify bool      `json:"verify,omitempty"`
+	ActID  string    `json:"act-id,omitempty"`
+	Ended  time.Time `json:"ended,omitempty"`
+	Result string    `json:"result,omitempty"`
 	// Waiting is why it waits, while it waits: the operator's warrant, or the keeper's first view.
 	Waiting string `json:"waiting,omitempty"`
 	// Unrecorded is why the hand act of what it did could not be recorded, when it could not: said, never only logged.
@@ -144,6 +152,12 @@ func (k *Keeper) act(r *Request) (string, error) {
 		}
 		return fmt.Sprintf("the first view answered at revision %d", s.Revision), nil
 	}
+	switch r.Verb {
+	case VerbSnapshotRestore:
+		return k.startRestore(r)
+	case VerbSnapshotRestoreCancel:
+		return k.cancelRestore(r)
+	}
 	if !k.Viewed() {
 		return "", fmt.Errorf("%w: the keeper holds no first view yet, so it acts on nothing", ErrNoWarrant)
 	}

modules/nats/internal/keeper/restore.go

@@ -0,0 +1,222 @@
+package keeper
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"time"
+
+	"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
+)
+
+// The keeper's restore in place of a snapshot it took (design 47 G §681, ADR 0675; the bus module hands the work through
+// the state requests, A-45): the recovery module asks the seat mesh-broker's snapshot-restore for a protected snapshot, which
+// is never handed out. The keeper checks the file against its recorded digest, refuses while a live stream of the snapshot's
+// names exists, and restores them under their own names within RestoreBound, one restore at a time. It says the outcome on
+// the event snapshot-restored under the request and the act.
+
+// Verbs a request names for the snapshot restore.
+const (
+	VerbSnapshotRestore       = meshbroker.VerbSnapshotRestore
+	VerbSnapshotRestoreCancel = meshbroker.VerbSnapshotRestoreCancel
+)
+
+// RestoreBound is the longest a restore runs: past it the keeper ends it as failed.
+const RestoreBound = 60 * time.Minute
+
+// SnapshotRestored is the body of the event snapshot-restored.
+type SnapshotRestored = meshbroker.SnapshotRestoredEvent
+
+// restoreRun is the restore that runs: its act and how to end it.
+type restoreRun struct {
+	act       string
+	request   string
+	streams   []string
+	cancel    context.CancelFunc
+	cancelled bool
+}
+
+// snapshotOf reads the recorded snapshot of request id: the refusals §681 words, or the record.
+func (k *Keeper) snapshotOf(id string) (Request, error) {
+	raw, _, found, err := k.Store.Get(StateRequests, id)
+	if err != nil {
+		return Request{}, fmt.Errorf("the snapshot of the request %s cannot be read: %w", id, err)
+	}
+	var rec Request
+	if !found || json.Unmarshal(raw, &rec) != nil || rec.Verb == "" {
+		return Request{}, fmt.Errorf("no snapshot of the request %s is held", id)
+	}
+	if !rec.Recorded() {
+		return Request{}, fmt.Errorf("snapshot taken before milestone G")
+	}
+	return rec, nil
+}
+
+// checked reads the snapshot file against its record: missing, or damaged.
+func (k *Keeper) checked(rec Request) error {
+	size, digest, err := k.files().Digest(rec.Path)
+	if errors.Is(err, os.ErrNotExist) {
+		return fmt.Errorf("snapshot missing")
+	}
+	if err != nil {
+		return fmt.Errorf("the snapshot %s cannot be read: %w", rec.Path, err)
+	}
+	if digest != rec.Digest || size != rec.Size {
+		return fmt.Errorf("snapshot damaged: request %s", rec.ID)
+	}
+	return nil
+}
+
+// startRestore answers a snapshot-restore at once: accepted with the act's id (the request's own id), busy with the id of the
+// running one, or an error; verify says the file holds and restores nothing.
+func (k *Keeper) startRestore(r *Request) (string, error) {
+	rec, err := k.snapshotOf(r.Of)
+	if err != nil {
+		return "error", err
+	}
+	if err := k.checked(rec); err != nil {
+		return "error", err
+	}
+	if r.Verify {
+		return "verified", nil
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), RestoreBound)
+	streams, err := k.Bus.SnapshotStreams(rec.Path)
+	if err != nil {
+		cancel()
+		return "error", fmt.Errorf("the snapshot's streams cannot be read: %w", err)
+	}
+	for _, name := range streams {
+		live, err := k.Bus.Live(ctx, name)
+		if err != nil {
+			cancel()
+			return "error", fmt.Errorf("whether %s is live cannot be read: %w", name, err)
+		}
+		if live {
+			cancel()
+			return "error", fmt.Errorf("names live: the stream %s exists, and a restore goes over nothing", name)
+		}
+	}
+	k.mu.Lock()
+	if run := k.restoring; run != nil {
+		k.mu.Unlock()
+		cancel()
+		r.ActID = run.act
+		return "busy", nil
+	}
+	run := &restoreRun{act: r.ID, request: rec.ID, streams: streams, cancel: cancel}
+	k.restoring = run
+	k.mu.Unlock()
+	r.ActID = run.act
+	k.record(r, fmt.Sprintf("restores the snapshot of the request %s (%s) in place", rec.ID, rec.Path))
+	go k.restore(ctx, run, rec, r.Caller, r.Why)
+	return "accepted", nil
+}
+
+// restore runs the restore to its end. A restore that failed, passed its bound or was cancelled removes the keeper's own
+// partial stream (it held no live data, names live having come first) and says snapshot-restored failed before the keeper
+// takes another restore.
+func (k *Keeper) restore(ctx context.Context, run *restoreRun, rec Request, caller, why string) {
+	err := k.Bus.Restore(ctx, rec.Path)
+	k.mu.Lock()
+	cancelled := run.cancelled
+	k.mu.Unlock()
+	switch {
+	case cancelled:
+		err = fmt.Errorf("cancelled")
+	case errors.Is(ctx.Err(), context.DeadlineExceeded):
+		err = fmt.Errorf("not finished within %s: %v", RestoreBound, err)
+	}
+	run.cancel()
+	k.endRestore(run.request, run.act, run.streams, err)
+	k.mu.Lock()
+	if k.restoring == run {
+		k.restoring = nil
+	}
+	k.mu.Unlock()
+}
+
+// endRestore ends the act: a failure removes the partial streams first, then the event says it and the request's record
+// holds how it ended.
+func (k *Keeper) endRestore(request, act string, streams []string, failure error) {
+	result, text := meshbroker.SnapshotRestoredEventOutcomeDone, ""
+	if failure != nil {
+		result, text = meshbroker.SnapshotRestoredEventOutcomeFailed, failure.Error()
+		// Removing the partial stream does not depend on the restore's own context, which has ended.
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+		for _, name := range streams {
+			if err := k.Bus.DeleteStream(ctx, name); err != nil {
+				text += fmt.Sprintf("; the partial stream %s could not be removed: %v", name, err)
+				k.logf("the partial stream %s of the restore %s cannot be removed: %v", name, act, err)
+			}
+		}
+		cancel()
+	}
+	k.logf("the restore %s of the snapshot of the request %s ended %s %s", act, request, result, text)
+	if raw, _, found, err := k.Store.Get(StateRequests, act); err == nil && found {
+		var r Request
+		if json.Unmarshal(raw, &r) == nil {
+			r.Ended, r.Result = k.now(), result
+			if text != "" {
+				r.Error = text
+			}
+			data, _ := json.Marshal(r)
+			if _, err := k.Store.Put(StateRequests, act, data); err != nil {
+				k.logf("the end of the restore %s cannot be written: %v", act, err)
+			}
+		}
+	}
+	if k.Emit == nil {
+		return
+	}
+	s := Saying{Event: meshbroker.EventSnapshotRestored, ID: EventID(), At: k.now(),
+		Restored: &SnapshotRestored{Request: request, Act: act, Outcome: result, Error: text}}
+	if err := k.Emit(&s); err != nil {
+		s.Error = err.Error()
+		k.setStatus(func(st *Status) { st.Unsaid = append(st.Unsaid, s) })
+		k.logf("the event %s of the restore %s could not be said, said again at the next fill pass: %v", s.Event, act, err)
+	}
+}
+
+// cancelRestore cancels the restore of the act r.Of names: accepted with its id, or an error when it has ended.
+func (k *Keeper) cancelRestore(r *Request) (string, error) {
+	k.mu.Lock()
+	run := k.restoring
+	if run == nil || run.act != r.Of {
+		k.mu.Unlock()
+		return "error", fmt.Errorf("the restore %s has ended or is none of this keeper's running", r.Of)
+	}
+	run.cancelled = true
+	k.mu.Unlock()
+	run.cancel()
+	r.ActID = run.act
+	k.record(r, fmt.Sprintf("cancelled the restore %s", run.act))
+	return "accepted", nil
+}
+
+// EndCutOffRestores ends, at the keeper's start, every restore a restart cut off: its partial stream removed and
+// snapshot-restored said failed under its act.
+func (k *Keeper) EndCutOffRestores() {
+	keys, err := k.Store.Keys(StateRequests)
+	if err != nil {
+		k.logf("the restores a restart cut off cannot be looked for: %v", err)
+		return
+	}
+	for _, key := range keys {
+		raw, _, found, err := k.Store.Get(StateRequests, key)
+		if err != nil || !found {
+			continue
+		}
+		var r Request
+		if json.Unmarshal(raw, &r) != nil || r.Verb != VerbSnapshotRestore || r.Outcome != "accepted" || !r.Ended.IsZero() {
+			continue
+		}
+		var streams []string
+		if rec, err := k.snapshotOf(r.Of); err == nil {
+			streams, _ = k.Bus.SnapshotStreams(rec.Path)
+		}
+		k.endRestore(r.Of, r.ID, streams, fmt.Errorf("cut off by the keeper's restart"))
+	}
+}

modules/nats/module.json

@@ -91,7 +91,7 @@
     {
       "name": "requests",
       "history": 5,
-      "ttl-seconds": 86400
+      "ttl-seconds": 7862400
     },
     {
       "name": "spent",