From d627400cf1e28279956009b1bab531ffeb1d9e58 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 22:45:56 +0100 Subject: [PATCH] =?UTF-8?q?bus5=20B2=20(=C2=A7681):=20the=20keeper=20resto?= =?UTF-8?q?res=20a=20snapshot=20in=20place=20(verify,=20names=20live,=2060?= =?UTF-8?q?-minute=20bound,=20cancel,=20partial=20stream=20removed,=20snap?= =?UTF-8?q?shot-restored);=20requests=20keeps=20records=2091=20days=20(A-5?= =?UTF-8?q?3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §681, ADR 0675, lead answers A-45, A-53 --- modules/nats/cmd/nats-tools/keep.go | 14 +- modules/nats/internal/bussnap/restore.go | 158 ++++++++++++++ modules/nats/internal/keeper/jetstream.go | 39 ++++ modules/nats/internal/keeper/keeper.go | 12 ++ modules/nats/internal/keeper/keeper_test.go | 138 ++++++++++++ modules/nats/internal/keeper/requests.go | 14 ++ modules/nats/internal/keeper/restore.go | 222 ++++++++++++++++++++ modules/nats/module.json | 2 +- 8 files changed, 596 insertions(+), 3 deletions(-) create mode 100644 modules/nats/internal/bussnap/restore.go create mode 100644 modules/nats/internal/keeper/restore.go diff --git a/modules/nats/cmd/nats-tools/keep.go b/modules/nats/cmd/nats-tools/keep.go index a073a95d5..1247d8e44 100644 --- a/modules/nats/cmd/nats-tools/keep.go +++ b/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) diff --git a/modules/nats/internal/bussnap/restore.go b/modules/nats/internal/bussnap/restore.go new file mode 100644 index 000000000..030f8d7b1 --- /dev/null +++ b/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 +} diff --git a/modules/nats/internal/keeper/jetstream.go b/modules/nats/internal/keeper/jetstream.go index fbb94b3f6..5ff2d51c0 100644 --- a/modules/nats/internal/keeper/jetstream.go +++ b/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) +} diff --git a/modules/nats/internal/keeper/keeper.go b/modules/nats/internal/keeper/keeper.go index 7e9ba93d9..bb3747d27 100644 --- a/modules/nats/internal/keeper/keeper.go +++ b/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 { diff --git a/modules/nats/internal/keeper/keeper_test.go b/modules/nats/internal/keeper/keeper_test.go index 99943317c..a498b5f2d 100644 --- a/modules/nats/internal/keeper/keeper_test.go +++ b/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}}) diff --git a/modules/nats/internal/keeper/requests.go b/modules/nats/internal/keeper/requests.go index 692b98b20..ace02a3ac 100644 --- a/modules/nats/internal/keeper/requests.go +++ b/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) } diff --git a/modules/nats/internal/keeper/restore.go b/modules/nats/internal/keeper/restore.go new file mode 100644 index 000000000..4ada2e3f9 --- /dev/null +++ b/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")) + } +} diff --git a/modules/nats/module.json b/modules/nats/module.json index b9bba9c08..8021aa955 100644 --- a/modules/nats/module.json +++ b/modules/nats/module.json @@ -91,7 +91,7 @@ { "name": "requests", "history": 5, - "ttl-seconds": 86400 + "ttl-seconds": 7862400 }, { "name": "spent",