The restic holder copied JetStream's store while the server wrote it; such a copy may not restore. The nats image now carries mesh-nats-snapshot, run by the declared dump under the module's own bus account (snapshot API only): every stream one at a time, flow-controlled, into one tar with a manifest of counts, sequences and checksums. Restore builds a new store beside the live one with the bus's own server; a person swaps it in. Proven against throwaway nats 2.11 servers being written to during the snapshot.
163 lines
4.2 KiB
Go
163 lines
4.2 KiB
Go
package main
|
|
|
|
import (
|
|
"archive/tar"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"os"
|
|
"path/filepath"
|
|
)
|
|
|
|
// Archive is a snapshot read back: its manifest, and each file beside it unpacked into a scratch
|
|
// directory of its own, removed by Close.
|
|
type Archive struct {
|
|
Manifest Manifest
|
|
dir string
|
|
files map[string]string
|
|
}
|
|
|
|
// ReadArchive unpacks a snapshot. Only the files the manifest names are kept, and only under names
|
|
// that are one stream's directory: an archive is data, and a path in it is never trusted to say where
|
|
// on the machine a file goes.
|
|
func ReadArchive(r io.Reader, scratch string) (*Archive, error) {
|
|
dir, err := os.MkdirTemp(scratch, "mesh-nats-restore-*")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
a := &Archive{dir: dir, files: map[string]string{}}
|
|
tr := tar.NewReader(r)
|
|
first := true
|
|
for {
|
|
hdr, err := tr.Next()
|
|
if err == io.EOF {
|
|
break
|
|
}
|
|
if err != nil {
|
|
a.Close()
|
|
return nil, fmt.Errorf("the archive does not read: %w", err)
|
|
}
|
|
if first {
|
|
if hdr.Name != ManifestName {
|
|
a.Close()
|
|
return nil, fmt.Errorf("the archive starts with %q, not its manifest: it is not a snapshot of the bus", hdr.Name)
|
|
}
|
|
raw, err := io.ReadAll(io.LimitReader(tr, 64<<20))
|
|
if err != nil {
|
|
a.Close()
|
|
return nil, err
|
|
}
|
|
if err := json.Unmarshal(raw, &a.Manifest); err != nil {
|
|
a.Close()
|
|
return nil, fmt.Errorf("its manifest: %w", err)
|
|
}
|
|
if a.Manifest.Format != 1 {
|
|
a.Close()
|
|
return nil, fmt.Errorf("its manifest is format %d; this program reads format 1", a.Manifest.Format)
|
|
}
|
|
first = false
|
|
continue
|
|
}
|
|
if !a.named(hdr.Name) {
|
|
a.Close()
|
|
return nil, fmt.Errorf("the archive holds %q, which its manifest does not name", hdr.Name)
|
|
}
|
|
local := filepath.Join(dir, fmt.Sprintf("%04d", len(a.files)))
|
|
f, err := os.OpenFile(local, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
|
|
if err != nil {
|
|
a.Close()
|
|
return nil, err
|
|
}
|
|
_, err = io.Copy(f, tr)
|
|
f.Close()
|
|
if err != nil {
|
|
a.Close()
|
|
return nil, err
|
|
}
|
|
a.files[hdr.Name] = local
|
|
}
|
|
if first {
|
|
a.Close()
|
|
return nil, errors.New("the archive is empty")
|
|
}
|
|
for _, s := range a.Manifest.Streams {
|
|
if !validName(s.Name) || s.Meta != "streams/"+s.Name+"/backup.json" || s.Archive != "streams/"+s.Name+"/stream.tar.s2" {
|
|
a.Close()
|
|
return nil, fmt.Errorf("the manifest names a stream %q at paths that are not its own", s.Name)
|
|
}
|
|
}
|
|
return a, nil
|
|
}
|
|
|
|
func (a *Archive) named(name string) bool {
|
|
for _, s := range a.Manifest.Streams {
|
|
if name == s.Meta || name == s.Archive {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// Close removes the unpacked files.
|
|
func (a *Archive) Close() { os.RemoveAll(a.dir) }
|
|
|
|
// Verify holds every file to the manifest's size and checksum, and reads every stream's archive to
|
|
// its end.
|
|
func (a *Archive) Verify() error {
|
|
for _, s := range a.Manifest.Streams {
|
|
for _, f := range []struct{ name, sum string }{{s.Meta, s.MetaSHA256}, {s.Archive, s.ArchiveSHA256}} {
|
|
local, ok := a.files[f.name]
|
|
if !ok {
|
|
return fmt.Errorf("%s is named in the manifest and missing from the archive", f.name)
|
|
}
|
|
got, size, err := sha256Of(local)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if got != f.sum {
|
|
return fmt.Errorf("%s does not match the checksum its manifest gives", f.name)
|
|
}
|
|
if f.name == s.Archive && size != s.ArchiveBytes {
|
|
return fmt.Errorf("%s is %d bytes; its manifest says %d", f.name, size, s.ArchiveBytes)
|
|
}
|
|
}
|
|
af, err := os.Open(a.files[s.Archive])
|
|
if err != nil {
|
|
return err
|
|
}
|
|
err = readsThrough(af)
|
|
af.Close()
|
|
if err != nil {
|
|
return fmt.Errorf("%s does not read: %w", s.Archive, err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func sha256Of(path string) (string, int64, error) {
|
|
f, err := os.Open(path)
|
|
if err != nil {
|
|
return "", 0, err
|
|
}
|
|
defer f.Close()
|
|
h := sha256.New()
|
|
n, err := io.Copy(h, f)
|
|
if err != nil {
|
|
return "", 0, err
|
|
}
|
|
return hex.EncodeToString(h.Sum(nil)), n, nil
|
|
}
|
|
|
|
// open is one stream's meta and archive.
|
|
func (a *Archive) open(s StreamEntry) (meta []byte, archive *os.File, err error) {
|
|
meta, err = os.ReadFile(a.files[s.Meta])
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
archive, err = os.Open(a.files[s.Archive])
|
|
return meta, archive, err
|
|
}
|