diff --git a/modules/nats/cmd/mesh-nats-snapshot/bussnap.go b/modules/nats/cmd/mesh-nats-snapshot/bussnap.go index e4d8a8f59..444dd213f 100644 --- a/modules/nats/cmd/mesh-nats-snapshot/bussnap.go +++ b/modules/nats/cmd/mesh-nats-snapshot/bussnap.go @@ -14,6 +14,8 @@ var ( StreamNames = bussnap.StreamNames request = bussnap.Request truncate = bussnap.Truncate + RestoreAll = bussnap.RestoreAll + heldTo = bussnap.HeldTo ) const ( diff --git a/modules/nats/cmd/mesh-nats-snapshot/restore.go b/modules/nats/cmd/mesh-nats-snapshot/restore.go index 5d5866094..f98bbc536 100644 --- a/modules/nats/cmd/mesh-nats-snapshot/restore.go +++ b/modules/nats/cmd/mesh-nats-snapshot/restore.go @@ -4,10 +4,8 @@ import ( "context" "crypto/rand" "encoding/hex" - "encoding/json" "errors" "fmt" - "io" "net" "os" "os/exec" @@ -20,118 +18,6 @@ import ( "nats-tools/internal/busarchive" ) -// restoreChunk is what is sent at a time; the server answers each before the next goes. -const restoreChunk = 64 * 1024 - -// RestoreAll puts every stream of the archive into the server nc reaches, one at a time, and holds -// each restored stream's state to the manifest. **Nothing is restored over anything**: a server that -// already holds any of the streams is refused before the first is sent. -func RestoreAll(ctx context.Context, nc *nats.Conn, a *busarchive.Archive, log func(string, ...any)) error { - var present []string - for _, s := range a.Manifest.Streams { - err := request(ctx, nc, apiStreamInfo+s.Name, nil, nil) - var api *apiError - switch { - case err == nil: - present = append(present, s.Name) - case errors.As(err, &api) && api.Code == 404: - default: - return err - } - } - if len(present) > 0 { - return fmt.Errorf("the server already holds %s; nothing is restored over a live stream — restore into a new "+ - "store (--into) and swap it in with the bus stopped", strings.Join(present, ", ")) - } - 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, at least %d message(s) to sequence %d, in %.2fs", s.Name, s.Messages, s.LastSeq, 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 { - cctx, cancel := context.WithTimeout(ctx, 30*time.Second) - reply, rerr := nc.RequestWithContext(cctx, resp.DeliverSubject, buf[:n]) - cancel() - 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 - } - } - // The end: an empty message, answered once the server has rebuilt the stream from what it was - // sent — which for a large stream takes a while. - fctx, cancel := context.WithTimeout(ctx, 30*time.Minute) - defer cancel() - final, err := nc.RequestWithContext(fctx, 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 - } - return heldTo(created.State, s) -} - -// heldTo is whether a restored stream holds at least what its snapshot said it held. -// -// **At least, not exactly — found by the restore test, not assumed.** The state the server answers a -// snapshot request with is taken when the snapshot starts; the stream goes on taking messages while -// its blocks are read out, and a block read a moment later carries some of what was appended in that -// moment. So a stream written during its snapshot restores with a later last sequence than the -// manifest says (90075 where it said 90024, with a writer publishing every 2ms through it), and of -// the messages in that tail some may be there and some not. Every message up to the manifest's last -// sequence is there, exactly; that is the snapshot's promise, and what the test holds it to message -// by message. What must never happen is a restored stream that ends before the manifest said it did. -func heldTo(restored streamState, s busarchive.StreamEntry) error { - if restored.LastSeq < s.LastSeq || (s.Messages > 0 && restored.Messages == 0) { - return fmt.Errorf("restored with %d message(s) up to sequence %d; the snapshot held %d up to %d", - restored.Messages, restored.LastSeq, s.Messages, s.LastSeq) - } - return nil -} - // RestoreOffline fills a new store directory with the archive's streams, by a server of this // program's own: nats-server, the same binary the bus runs, on loopback, with the bus's one account // and a user that exists only for this run. It is stopped when the streams are in. What it leaves diff --git a/modules/nats/internal/bussnap/restore.go b/modules/nats/internal/bussnap/restore.go new file mode 100644 index 000000000..484ddf22e --- /dev/null +++ b/modules/nats/internal/bussnap/restore.go @@ -0,0 +1,126 @@ +package bussnap + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "strings" + "time" + + "github.com/nats-io/nats.go" + "nats-tools/internal/busarchive" +) + +// restoreChunk is what is sent at a time; the server answers each before the next goes. +const restoreChunk = 64 * 1024 + +// RestoreAll puts every stream of the archive into the server nc reaches, one at a time, and holds +// each restored stream's state to the manifest. **Nothing is restored over anything**: a server that +// already holds any of the streams is refused before the first is sent. +func RestoreAll(ctx context.Context, nc *nats.Conn, a *busarchive.Archive, log func(string, ...any)) error { + var present []string + for _, s := range a.Manifest.Streams { + err := Request(ctx, nc, APIStreamInfo+s.Name, nil, nil) + var api *APIError + switch { + case err == nil: + present = append(present, s.Name) + case errors.As(err, &api) && api.Code == 404: + default: + return err + } + } + if len(present) > 0 { + return fmt.Errorf("the server already holds %s; nothing is restored over a live stream — restore into a new "+ + "store (--into) and swap it in with the bus stopped", strings.Join(present, ", ")) + } + 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, at least %d message(s) to sequence %d, in %.2fs", s.Name, s.Messages, s.LastSeq, 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 { + cctx, cancel := context.WithTimeout(ctx, 30*time.Second) + reply, rerr := nc.RequestWithContext(cctx, resp.DeliverSubject, buf[:n]) + cancel() + 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 + } + } + // The end: an empty message, answered once the server has rebuilt the stream from what it was + // sent — which for a large stream takes a while. + fctx, cancel := context.WithTimeout(ctx, 30*time.Minute) + defer cancel() + final, err := nc.RequestWithContext(fctx, 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 + } + return HeldTo(created.State, s) +} + +// HeldTo is whether a restored stream holds at least what its snapshot said it held. +// +// **At least, not exactly — found by the restore test, not assumed.** The state the server answers a +// snapshot request with is taken when the snapshot starts; the stream goes on taking messages while +// its blocks are read out, and a block read a moment later carries some of what was appended in that +// moment. So a stream written during its snapshot restores with a later last sequence than the +// manifest says (90075 where it said 90024, with a writer publishing every 2ms through it), and of +// the messages in that tail some may be there and some not. Every message up to the manifest's last +// sequence is there, exactly; that is the snapshot's promise, and what the test holds it to message +// by message. What must never happen is a restored stream that ends before the manifest said it did. +func HeldTo(restored StreamState, s busarchive.StreamEntry) error { + if restored.LastSeq < s.LastSeq || (s.Messages > 0 && restored.Messages == 0) { + return fmt.Errorf("restored with %d message(s) up to sequence %d; the snapshot held %d up to %d", + restored.Messages, restored.LastSeq, s.Messages, s.LastSeq) + } + return nil +}