G §655: the take and restore of one stream's snapshot is shared code in mesh-sdk (go/bussnap, go/busarchive)
The bus module's bussnap moves whole, with the archive package it reads and its tests (lead answer A-2: busarchive moves with it); the bus module, the recovery module and the nightly dump import them from here, and no copy stays in mesh-catalog. Builds: design 47 G §655, A-2 (.work/47-G/dev2-ledger.md) Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -0,0 +1,162 @@
|
||||
package busarchive
|
||||
|
||||
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
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
// Package busarchive is the bus's snapshot archive: its manifest, reading one back and holding it to
|
||||
// its checksums. The snapshot program (cmd/mesh-nats-snapshot) writes and restores with it and the bus's
|
||||
// tool nats_snapshot_verify (cmd/nats-tools) checks with it, so the two never differ on what a good
|
||||
// snapshot is.
|
||||
package busarchive
|
||||
|
||||
import (
|
||||
"archive/tar"
|
||||
"io"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/klauspost/compress/s2"
|
||||
)
|
||||
|
||||
// Manifest is the first entry of an archive: what was taken, when, how long it took, and the checksum
|
||||
// of every file beside it. It is what `verify` and `restore` hold the archive to.
|
||||
type Manifest struct {
|
||||
Format int `json:"format"`
|
||||
Server string `json:"server_version"`
|
||||
Started time.Time `json:"started"`
|
||||
Finished time.Time `json:"finished"`
|
||||
Seconds float64 `json:"seconds"`
|
||||
// Messages and StreamBytes are the streams' own totals at their snapshots; ArchiveBytes what the
|
||||
// archives take, compressed.
|
||||
Messages uint64 `json:"messages"`
|
||||
StreamBytes int64 `json:"stream_bytes"`
|
||||
ArchiveBytes int64 `json:"archive_bytes"`
|
||||
Streams []StreamEntry `json:"streams"`
|
||||
// Skipped are streams that hold nothing across a restart (memory storage), said rather than
|
||||
// silently left out.
|
||||
Skipped []Skipped `json:"skipped,omitempty"`
|
||||
}
|
||||
|
||||
// StreamEntry is one stream as it was snapshotted.
|
||||
type StreamEntry struct {
|
||||
Name string `json:"name"`
|
||||
Subjects []string `json:"subjects,omitempty"`
|
||||
Retention string `json:"retention,omitempty"`
|
||||
Messages uint64 `json:"messages"`
|
||||
Bytes uint64 `json:"bytes"`
|
||||
FirstSeq uint64 `json:"first_seq"`
|
||||
LastSeq uint64 `json:"last_seq"`
|
||||
Deleted int `json:"deleted,omitempty"`
|
||||
Consumers int `json:"consumers"`
|
||||
// Meta is the server's own description of the stream (its configuration and state), in the
|
||||
// shape the restore API and the nats CLI take back: backup.json.
|
||||
Meta string `json:"meta"`
|
||||
MetaSHA256 string `json:"meta_sha256"`
|
||||
// Archive is the server's snapshot of the stream, as the server sent it: stream.tar.s2.
|
||||
Archive string `json:"archive"`
|
||||
ArchiveBytes int64 `json:"archive_bytes"`
|
||||
ArchiveSHA256 string `json:"archive_sha256"`
|
||||
Seconds float64 `json:"seconds"`
|
||||
}
|
||||
|
||||
// Skipped is a stream not snapshotted, and why.
|
||||
type Skipped struct {
|
||||
Name string `json:"name"`
|
||||
Why string `json:"why"`
|
||||
}
|
||||
|
||||
// ManifestName is the archive's first entry.
|
||||
const ManifestName = "manifest.json"
|
||||
|
||||
// ReadsThrough is whether a stream's archive decompresses and lists to its end.
|
||||
func ReadsThrough(r io.Reader) error {
|
||||
tr := tar.NewReader(s2.NewReader(r))
|
||||
for {
|
||||
_, err := tr.Next()
|
||||
if err == io.EOF {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := io.Copy(io.Discard, tr); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ValidName is a stream name that is one safe path component.
|
||||
func ValidName(name string) bool {
|
||||
return name != "" && name != "." && name != ".." && !strings.ContainsAny(name, "/\\\x00 *>")
|
||||
}
|
||||
@@ -0,0 +1,109 @@
|
||||
package busarchive
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io/fs"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// StreamReport is one stream of a verified snapshot.
|
||||
type StreamReport struct {
|
||||
Name string `json:"name"`
|
||||
LastSeq uint64 `json:"last_seq"`
|
||||
}
|
||||
|
||||
// Report is the answer of VerifyFile: which file was held to its manifest, and whether it passed.
|
||||
type Report struct {
|
||||
File string `json:"file"`
|
||||
Modified time.Time `json:"modified"`
|
||||
Size int64 `json:"size"`
|
||||
Taken *time.Time `json:"taken,omitempty"`
|
||||
Streams []StreamReport `json:"streams"`
|
||||
Skipped []Skipped `json:"skipped,omitempty"`
|
||||
Passed bool `json:"passed"`
|
||||
// Fault is the first thing wrong, empty when the snapshot passed.
|
||||
Fault string `json:"fault,omitempty"`
|
||||
}
|
||||
|
||||
// VerifyFile verifies one snapshot inside dir: the file called name, or the newest *.tar when name is
|
||||
// empty. Nothing outside dir is read and nothing is written there; a snapshot that fails is an answer
|
||||
// (Passed false, the first Fault), and only a directory without a snapshot is an error.
|
||||
func VerifyFile(dir, name string) (*Report, error) {
|
||||
path, info, err := pick(dir, name)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
r := &Report{File: filepath.Base(path), Modified: info.ModTime().UTC(), Size: info.Size(), Streams: []StreamReport{}}
|
||||
f, err := os.Open(path)
|
||||
if err != nil {
|
||||
r.Fault = err.Error()
|
||||
return r, nil
|
||||
}
|
||||
defer f.Close()
|
||||
a, err := ReadArchive(f, "")
|
||||
if err != nil {
|
||||
r.Fault = err.Error()
|
||||
return r, nil
|
||||
}
|
||||
defer a.Close()
|
||||
taken := a.Manifest.Finished
|
||||
r.Taken = &taken
|
||||
r.Skipped = a.Manifest.Skipped
|
||||
for _, s := range a.Manifest.Streams {
|
||||
r.Streams = append(r.Streams, StreamReport{Name: s.Name, LastSeq: s.LastSeq})
|
||||
}
|
||||
if err := a.Verify(); err != nil {
|
||||
r.Fault = err.Error()
|
||||
return r, nil
|
||||
}
|
||||
r.Passed = true
|
||||
return r, nil
|
||||
}
|
||||
|
||||
// pick names the regular file to verify, and nothing outside dir: a name is one path component, and a
|
||||
// symbolic link is not followed out of the directory.
|
||||
func pick(dir, name string) (string, fs.FileInfo, error) {
|
||||
if dir == "" {
|
||||
return "", nil, errors.New("the snapshots directory is not set")
|
||||
}
|
||||
if name != "" {
|
||||
if name != filepath.Base(name) || name == "." || name == ".." || strings.ContainsRune(name, 0) {
|
||||
return "", nil, fmt.Errorf("%q is not a file name inside the snapshots directory", name)
|
||||
}
|
||||
p := filepath.Join(dir, name)
|
||||
info, err := os.Lstat(p)
|
||||
if err != nil {
|
||||
return "", nil, fmt.Errorf("no snapshot %s in the snapshots directory: %w", name, err)
|
||||
}
|
||||
if !info.Mode().IsRegular() {
|
||||
return "", nil, fmt.Errorf("%s is not a regular file", name)
|
||||
}
|
||||
return p, info, nil
|
||||
}
|
||||
entries, err := os.ReadDir(dir)
|
||||
if err != nil {
|
||||
return "", nil, fmt.Errorf("the snapshots directory does not read: %w", err)
|
||||
}
|
||||
var best string
|
||||
var bestInfo fs.FileInfo
|
||||
for _, e := range entries {
|
||||
if !strings.HasSuffix(e.Name(), ".tar") {
|
||||
continue
|
||||
}
|
||||
info, err := e.Info()
|
||||
if err != nil || !info.Mode().IsRegular() {
|
||||
continue
|
||||
}
|
||||
if bestInfo == nil || info.ModTime().After(bestInfo.ModTime()) {
|
||||
best, bestInfo = e.Name(), info
|
||||
}
|
||||
}
|
||||
if bestInfo == nil {
|
||||
return "", nil, errors.New("the snapshots directory holds no .tar snapshot")
|
||||
}
|
||||
return filepath.Join(dir, best), bestInfo, nil
|
||||
}
|
||||
@@ -0,0 +1,173 @@
|
||||
package busarchive
|
||||
|
||||
import (
|
||||
"archive/tar"
|
||||
"bytes"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/klauspost/compress/s2"
|
||||
)
|
||||
|
||||
func sum(b []byte) string { h := sha256.Sum256(b); return hex.EncodeToString(h[:]) }
|
||||
|
||||
// streamArchive is a stream's archive as the server sends it: a tar of one file, s2-compressed.
|
||||
func streamArchive(t *testing.T) []byte {
|
||||
t.Helper()
|
||||
var raw bytes.Buffer
|
||||
tw := tar.NewWriter(&raw)
|
||||
body := []byte(strings.Repeat("a message of the stream\n", 50))
|
||||
if err := tw.WriteHeader(&tar.Header{Name: "msgs.blk", Mode: 0o600, Size: int64(len(body))}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tw.Write(body)
|
||||
tw.Close()
|
||||
var out bytes.Buffer
|
||||
w := s2.NewWriter(&out)
|
||||
w.Write(raw.Bytes())
|
||||
w.Close()
|
||||
return out.Bytes()
|
||||
}
|
||||
|
||||
// snapshotTar builds a snapshot of two streams; the mutations change what the manifest says or what is
|
||||
// stored, so each fault is made in the one place it is found.
|
||||
func snapshotTar(t *testing.T, mutate func(m *Manifest, files map[string][]byte)) []byte {
|
||||
t.Helper()
|
||||
files := map[string][]byte{}
|
||||
m := Manifest{Format: 1, Server: "2.11.17", Finished: time.Date(2026, 10, 22, 3, 0, 0, 0, time.UTC)}
|
||||
for i, name := range []string{"EVENTS", "KV_conditions"} {
|
||||
meta := []byte(`{"config":{"name":"` + name + `"}}`)
|
||||
arc := streamArchive(t)
|
||||
e := StreamEntry{Name: name, LastSeq: uint64(10 * (i + 1)),
|
||||
Meta: "streams/" + name + "/backup.json", MetaSHA256: sum(meta),
|
||||
Archive: "streams/" + name + "/stream.tar.s2", ArchiveBytes: int64(len(arc)), ArchiveSHA256: sum(arc)}
|
||||
m.Streams = append(m.Streams, e)
|
||||
files[e.Meta], files[e.Archive] = meta, arc
|
||||
}
|
||||
if mutate != nil {
|
||||
mutate(&m, files)
|
||||
}
|
||||
var out bytes.Buffer
|
||||
tw := tar.NewWriter(&out)
|
||||
put := func(name string, b []byte) {
|
||||
if err := tw.WriteHeader(&tar.Header{Name: name, Mode: 0o600, Size: int64(len(b))}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tw.Write(b)
|
||||
}
|
||||
raw, _ := json.Marshal(m)
|
||||
put(ManifestName, raw)
|
||||
for _, s := range m.Streams {
|
||||
put(s.Meta, files[s.Meta])
|
||||
put(s.Archive, files[s.Archive])
|
||||
}
|
||||
tw.Close()
|
||||
return out.Bytes()
|
||||
}
|
||||
|
||||
func put(t *testing.T, dir, name string, b []byte, at time.Time) {
|
||||
t.Helper()
|
||||
p := filepath.Join(dir, name)
|
||||
if err := os.WriteFile(p, b, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
os.Chtimes(p, at, at)
|
||||
}
|
||||
|
||||
func TestAGoodSnapshotPassesAndSaysItsStreams(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
good := snapshotTar(t, nil)
|
||||
put(t, dir, "bus.tar", good, time.Now())
|
||||
r, err := VerifyFile(dir, "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !r.Passed || r.Fault != "" || r.File != "bus.tar" || r.Size != int64(len(good)) {
|
||||
t.Fatalf("%+v", r)
|
||||
}
|
||||
if r.Taken == nil || !r.Taken.Equal(time.Date(2026, 10, 22, 3, 0, 0, 0, time.UTC)) {
|
||||
t.Errorf("taken %v", r.Taken)
|
||||
}
|
||||
if len(r.Streams) != 2 || r.Streams[0] != (StreamReport{"EVENTS", 10}) || r.Streams[1] != (StreamReport{"KV_conditions", 20}) {
|
||||
t.Errorf("streams %+v", r.Streams)
|
||||
}
|
||||
}
|
||||
|
||||
func TestACorruptedChecksumFailsNamingTheFile(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
put(t, dir, "bus.tar", snapshotTar(t, func(m *Manifest, files map[string][]byte) {
|
||||
m.Streams[1].ArchiveSHA256 = sum([]byte("something else"))
|
||||
}), time.Now())
|
||||
r, err := VerifyFile(dir, "bus.tar")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r.Passed || !strings.Contains(r.Fault, "streams/KV_conditions/stream.tar.s2") || !strings.Contains(r.Fault, "checksum") {
|
||||
t.Fatalf("%+v", r)
|
||||
}
|
||||
if len(r.Streams) != 2 {
|
||||
t.Errorf("a failed snapshot still says its streams: %+v", r.Streams)
|
||||
}
|
||||
}
|
||||
|
||||
func TestATruncatedSnapshotFails(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
good := snapshotTar(t, nil)
|
||||
put(t, dir, "bus.tar", good[:len(good)*2/3], time.Now())
|
||||
r, err := VerifyFile(dir, "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r.Passed || r.Fault == "" {
|
||||
t.Fatalf("a truncated snapshot passed: %+v", r)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnArchiveThatDoesNotReadToItsEndFailsEvenWithMatchingChecksums(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
put(t, dir, "bus.tar", snapshotTar(t, func(m *Manifest, files map[string][]byte) {
|
||||
bad := []byte("not an s2 stream at all")
|
||||
s := &m.Streams[0]
|
||||
files[s.Archive] = bad
|
||||
s.ArchiveSHA256, s.ArchiveBytes = sum(bad), int64(len(bad))
|
||||
}), time.Now())
|
||||
r, _ := VerifyFile(dir, "")
|
||||
if r.Passed || !strings.Contains(r.Fault, "does not read") {
|
||||
t.Fatalf("%+v", r)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheNewestTarIsVerifiedAndOnlyInsideTheDirectory(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
now := time.Now()
|
||||
put(t, dir, "old.tar", []byte("junk"), now.Add(-48*time.Hour))
|
||||
put(t, dir, "bus.tar", snapshotTar(t, nil), now)
|
||||
put(t, dir, "bus.tar.partial", []byte("half"), now.Add(time.Hour))
|
||||
r, err := VerifyFile(dir, "")
|
||||
if err != nil || r.File != "bus.tar" || !r.Passed {
|
||||
t.Fatalf("%+v %v", r, err)
|
||||
}
|
||||
if r, _ := VerifyFile(dir, "old.tar"); r.Passed || r.File != "old.tar" {
|
||||
t.Errorf("a named file is the one checked: %+v", r)
|
||||
}
|
||||
outside := t.TempDir()
|
||||
put(t, outside, "x.tar", snapshotTar(t, nil), now)
|
||||
for _, name := range []string{"../" + filepath.Base(outside) + "/x.tar", outside + "/x.tar", ".."} {
|
||||
if _, err := VerifyFile(dir, name); err == nil {
|
||||
t.Errorf("%q was read", name)
|
||||
}
|
||||
}
|
||||
os.Symlink(filepath.Join(outside, "x.tar"), filepath.Join(dir, "link.tar"))
|
||||
if _, err := VerifyFile(dir, "link.tar"); err == nil {
|
||||
t.Error("a symbolic link out of the directory was followed")
|
||||
}
|
||||
if _, err := VerifyFile(t.TempDir(), ""); err == nil {
|
||||
t.Error("an empty directory is not a snapshot")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,433 @@
|
||||
// Package bussnap takes a snapshot of the bus's streams with the server's own snapshot API (novox/hq ADR 0235): the
|
||||
// nightly dump of every stream, and the bus module's keeper's snapshot of one entry before it is rebuilt or removed
|
||||
// (design 25 § 12.3).
|
||||
package bussnap
|
||||
|
||||
import (
|
||||
"archive/tar"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/busarchive"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// The JetStream API's subjects this program speaks. Each is granted to the bus's own module and
|
||||
// nothing else is (mesh-controller internal/broker, BusSnapshotGrants); a test holds the two lists
|
||||
// equal against a real server.
|
||||
const (
|
||||
APIStreamNames = "$JS.API.STREAM.NAMES"
|
||||
APIStreamInfo = "$JS.API.STREAM.INFO."
|
||||
APIStreamSnapshot = "$JS.API.STREAM.SNAPSHOT."
|
||||
APIStreamRestore = "$JS.API.STREAM.RESTORE."
|
||||
)
|
||||
|
||||
// ChunkSize is what the server is asked to send at a time: its own default. The server sends up to
|
||||
// an 8 MiB window ahead of the acknowledgements and waits for them past it, so one stream costs the
|
||||
// server at most that much in flight, whatever its size.
|
||||
const ChunkSize = 128 * 1024
|
||||
|
||||
// SnapshotOptions bounds a snapshot.
|
||||
type SnapshotOptions struct {
|
||||
// StreamTimeout is the longest one stream may take; past it the snapshot fails, naming it.
|
||||
StreamTimeout time.Duration
|
||||
// Pause is a rest between two streams, so a night's snapshot is never one long burst.
|
||||
Pause time.Duration
|
||||
// TempDir is where one stream's archive waits until it is written out; the system's when empty.
|
||||
TempDir string
|
||||
Log func(format string, args ...any)
|
||||
// Streams, when set, are the only streams taken (the keeper's snapshot of one entry); every stream when empty.
|
||||
Streams []string
|
||||
}
|
||||
|
||||
// APIError is the error every JetStream API answer may carry.
|
||||
type APIError struct {
|
||||
Code int `json:"code"`
|
||||
ErrCode int `json:"err_code"`
|
||||
Description string `json:"description"`
|
||||
}
|
||||
|
||||
func (e *APIError) Error() string { return fmt.Sprintf("%s (%d)", e.Description, e.Code) }
|
||||
|
||||
// StreamState is what this program reads of a stream's state; the whole of it is kept as the server
|
||||
// gave it, in the stream's meta.
|
||||
type StreamState struct {
|
||||
Messages uint64 `json:"messages"`
|
||||
Bytes uint64 `json:"bytes"`
|
||||
FirstSeq uint64 `json:"first_seq"`
|
||||
LastSeq uint64 `json:"last_seq"`
|
||||
NumDelete int `json:"num_deleted"`
|
||||
Consumers int `json:"consumer_count"`
|
||||
}
|
||||
|
||||
type streamConfig struct {
|
||||
Name string `json:"name"`
|
||||
Subjects []string `json:"subjects"`
|
||||
Retention string `json:"retention"`
|
||||
Storage string `json:"storage"`
|
||||
}
|
||||
|
||||
// Request asks the JetStream API and decodes its answer, refusing an answer that carries an error.
|
||||
func Request(ctx context.Context, nc *nats.Conn, subject string, body any, into any) error {
|
||||
var payload []byte
|
||||
if body != nil {
|
||||
var err error
|
||||
if payload, err = json.Marshal(body); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
rctx, cancel := context.WithTimeout(ctx, 30*time.Second)
|
||||
defer cancel()
|
||||
msg, err := nc.RequestWithContext(rctx, subject, payload)
|
||||
if err != nil {
|
||||
if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, nats.ErrTimeout) || errors.Is(err, nats.ErrNoResponders) {
|
||||
return fmt.Errorf("%s was not answered (%v): the bus refuses a subject this user is not granted by not answering it", subject, err)
|
||||
}
|
||||
return fmt.Errorf("%s: %w", subject, err)
|
||||
}
|
||||
var wrapped struct {
|
||||
Error *APIError `json:"error"`
|
||||
}
|
||||
if err := json.Unmarshal(msg.Data, &wrapped); err != nil {
|
||||
return fmt.Errorf("%s answered something that is not JSON: %q", subject, Truncate(msg.Data))
|
||||
}
|
||||
if wrapped.Error != nil {
|
||||
return fmt.Errorf("%s: %w", subject, wrapped.Error)
|
||||
}
|
||||
if into != nil {
|
||||
return json.Unmarshal(msg.Data, into)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func Truncate(b []byte) string {
|
||||
if len(b) > 200 {
|
||||
return string(b[:200]) + "…"
|
||||
}
|
||||
return string(b)
|
||||
}
|
||||
|
||||
// StreamNames is every stream in the account, in name order, all pages of them.
|
||||
func StreamNames(ctx context.Context, nc *nats.Conn) ([]string, error) {
|
||||
var names []string
|
||||
for {
|
||||
var page struct {
|
||||
Total int `json:"total"`
|
||||
Offset int `json:"offset"`
|
||||
Limit int `json:"limit"`
|
||||
Streams []string `json:"streams"`
|
||||
}
|
||||
if err := Request(ctx, nc, APIStreamNames, map[string]int{"offset": len(names)}, &page); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
names = append(names, page.Streams...)
|
||||
if len(page.Streams) == 0 || len(names) >= page.Total {
|
||||
break
|
||||
}
|
||||
}
|
||||
sort.Strings(names)
|
||||
return names, nil
|
||||
}
|
||||
|
||||
// Snapshot writes every stream, one at a time, as a tar: the manifest first, then each stream's
|
||||
// meta and archive under streams/<name>/. Any stream failing fails the whole: a partial copy of the
|
||||
// bus written where a whole one is expected is the silent failure this exists to prevent.
|
||||
func Snapshot(ctx context.Context, nc *nats.Conn, out io.Writer, o SnapshotOptions) (*busarchive.Manifest, error) {
|
||||
if o.StreamTimeout <= 0 {
|
||||
o.StreamTimeout = 10 * time.Minute
|
||||
}
|
||||
if o.Log == nil {
|
||||
o.Log = func(string, ...any) {}
|
||||
}
|
||||
m := &busarchive.Manifest{Format: 1, Server: nc.ConnectedServerVersion(), Started: time.Now().UTC()}
|
||||
names, err := StreamNames(ctx, nc)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(o.Streams) > 0 {
|
||||
var only []string
|
||||
for _, n := range names {
|
||||
if slices.Contains(o.Streams, n) {
|
||||
only = append(only, n)
|
||||
}
|
||||
}
|
||||
if len(only) != len(o.Streams) {
|
||||
return nil, fmt.Errorf("the bus holds %v of the streams %v asked for: a snapshot missing one would pass for a whole one", only, o.Streams)
|
||||
}
|
||||
names = only
|
||||
}
|
||||
if len(names) == 0 {
|
||||
// The mesh's own streams always exist once the controller has run; a bus with none is not
|
||||
// one to call backed up.
|
||||
return nil, errors.New("the bus holds no streams; a snapshot of nothing would pass for a backup")
|
||||
}
|
||||
|
||||
type taken struct {
|
||||
entry busarchive.StreamEntry
|
||||
meta []byte
|
||||
archive *os.File
|
||||
}
|
||||
var files []taken
|
||||
defer func() {
|
||||
for _, t := range files {
|
||||
t.archive.Close()
|
||||
os.Remove(t.archive.Name())
|
||||
}
|
||||
}()
|
||||
|
||||
for i, name := range names {
|
||||
if i > 0 && o.Pause > 0 {
|
||||
select {
|
||||
case <-time.After(o.Pause):
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
}
|
||||
var info struct {
|
||||
Config streamConfig `json:"config"`
|
||||
}
|
||||
if err := Request(ctx, nc, APIStreamInfo+name, nil, &info); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if info.Config.Storage == "memory" {
|
||||
m.Skipped = append(m.Skipped, busarchive.Skipped{Name: name, Why: "memory storage: it holds nothing across a restart, and the server cannot snapshot it"})
|
||||
o.Log("%s: skipped, memory storage", name)
|
||||
continue
|
||||
}
|
||||
f, err := os.CreateTemp(o.TempDir, "mesh-nats-snapshot-*")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
files = append(files, taken{archive: f})
|
||||
sctx, cancel := context.WithTimeout(ctx, o.StreamTimeout)
|
||||
entry, meta, err := snapshotStream(sctx, nc, name, f)
|
||||
cancel()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("stream %s: %w", name, err)
|
||||
}
|
||||
files[len(files)-1].entry, files[len(files)-1].meta = entry, meta
|
||||
m.Messages += entry.Messages
|
||||
m.StreamBytes += int64(entry.Bytes)
|
||||
m.ArchiveBytes += entry.ArchiveBytes
|
||||
o.Log("%s: %d message(s), %s, last sequence %d, %d consumer(s) → %s in %.2fs",
|
||||
name, entry.Messages, Human(int64(entry.Bytes)), entry.LastSeq, entry.Consumers, Human(entry.ArchiveBytes), entry.Seconds)
|
||||
}
|
||||
m.Finished = time.Now().UTC()
|
||||
m.Seconds = m.Finished.Sub(m.Started).Seconds()
|
||||
for _, t := range files {
|
||||
m.Streams = append(m.Streams, t.entry)
|
||||
}
|
||||
|
||||
tw := tar.NewWriter(out)
|
||||
manifest, err := json.MarshalIndent(m, "", " ")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := writeEntry(tw, busarchive.ManifestName, append(manifest, '\n'), m.Finished); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, t := range files {
|
||||
if err := writeEntry(tw, t.entry.Meta, t.meta, m.Finished); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if _, err := t.archive.Seek(0, io.SeekStart); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
hdr := &tar.Header{Name: t.entry.Archive, Mode: 0o600, Size: t.entry.ArchiveBytes, ModTime: m.Finished, Format: tar.FormatPAX}
|
||||
if err := tw.WriteHeader(hdr); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if _, err := io.Copy(tw, t.archive); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
if err := tw.Close(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return m, nil
|
||||
}
|
||||
|
||||
func writeEntry(tw *tar.Writer, name string, body []byte, at time.Time) error {
|
||||
if err := tw.WriteHeader(&tar.Header{Name: name, Mode: 0o600, Size: int64(len(body)), ModTime: at, Format: tar.FormatPAX}); err != nil {
|
||||
return err
|
||||
}
|
||||
_, err := tw.Write(body)
|
||||
return err
|
||||
}
|
||||
|
||||
// snapshotStream takes one stream's snapshot into f: the server's chunks, each acknowledged as it
|
||||
// arrives so the server's window keeps moving, until the server's last, empty message. It answers
|
||||
// the stream's entry (the archive's size and checksum included) and its meta.
|
||||
//
|
||||
// The protocol is the server's (nats-server jetstream_api.go, streamSnapshot): the request names a
|
||||
// subject to deliver to; the server answers with the stream's configuration and its state at the
|
||||
// snapshot, then sends the archive in chunks whose reply subject is the acknowledgement it waits for
|
||||
// past its window — two seconds without one and it gives up with "408 No Flow Response". The end is
|
||||
// an empty message whose status header says 204, or the error that ended it.
|
||||
func snapshotStream(ctx context.Context, nc *nats.Conn, name string, f *os.File) (busarchive.StreamEntry, []byte, error) {
|
||||
started := time.Now()
|
||||
deliver := nc.NewRespInbox()
|
||||
|
||||
hash := sha256.New()
|
||||
var written int64
|
||||
var mu sync.Mutex
|
||||
var failed error
|
||||
done := make(chan struct{})
|
||||
var once sync.Once
|
||||
finish := func(err error) {
|
||||
mu.Lock()
|
||||
if failed == nil {
|
||||
failed = err
|
||||
}
|
||||
mu.Unlock()
|
||||
once.Do(func() { close(done) })
|
||||
}
|
||||
sub, err := nc.Subscribe(deliver, func(msg *nats.Msg) {
|
||||
if len(msg.Data) == 0 {
|
||||
if status := msg.Header.Get("Status"); status != "" && status != "204" {
|
||||
finish(fmt.Errorf("the server ended the snapshot: %s %s", status, msg.Header.Get("Description")))
|
||||
return
|
||||
}
|
||||
finish(nil)
|
||||
return
|
||||
}
|
||||
// Acknowledged before it is written: the chunk is already in memory, and a slow disk must not
|
||||
// read to the server as a reader that went away.
|
||||
if msg.Reply != "" {
|
||||
if err := nc.Publish(msg.Reply, nil); err != nil {
|
||||
finish(fmt.Errorf("acknowledging a chunk: %w", err))
|
||||
return
|
||||
}
|
||||
}
|
||||
if _, err := f.Write(msg.Data); err != nil {
|
||||
finish(fmt.Errorf("writing the archive: %w", err))
|
||||
return
|
||||
}
|
||||
hash.Write(msg.Data)
|
||||
written += int64(len(msg.Data))
|
||||
})
|
||||
if err != nil {
|
||||
return busarchive.StreamEntry{}, nil, err
|
||||
}
|
||||
defer sub.Unsubscribe()
|
||||
// Every chunk is held until the callback has it; the window bounds how many there can be.
|
||||
if err := sub.SetPendingLimits(-1, -1); err != nil {
|
||||
return busarchive.StreamEntry{}, nil, err
|
||||
}
|
||||
if err := nc.FlushWithContext(ctx); err != nil {
|
||||
return busarchive.StreamEntry{}, nil, err
|
||||
}
|
||||
|
||||
var resp struct {
|
||||
Config json.RawMessage `json:"config"`
|
||||
State json.RawMessage `json:"state"`
|
||||
}
|
||||
req := map[string]any{"deliver_subject": deliver, "chunk_size": ChunkSize}
|
||||
if err := Request(ctx, nc, APIStreamSnapshot+name, req, &resp); err != nil {
|
||||
return busarchive.StreamEntry{}, nil, err
|
||||
}
|
||||
select {
|
||||
case <-done:
|
||||
case <-ctx.Done():
|
||||
return busarchive.StreamEntry{}, nil, fmt.Errorf("not finished within its bound: %w", ctx.Err())
|
||||
}
|
||||
// The subscription is drained before the archive is read back, so no callback is still writing.
|
||||
_ = sub.Unsubscribe()
|
||||
mu.Lock()
|
||||
err = failed
|
||||
mu.Unlock()
|
||||
if err != nil {
|
||||
return busarchive.StreamEntry{}, nil, err
|
||||
}
|
||||
if written == 0 {
|
||||
return busarchive.StreamEntry{}, nil, errors.New("the server sent an empty archive")
|
||||
}
|
||||
if err := f.Sync(); err != nil {
|
||||
return busarchive.StreamEntry{}, nil, err
|
||||
}
|
||||
|
||||
var cfg streamConfig
|
||||
var state StreamState
|
||||
if err := json.Unmarshal(resp.Config, &cfg); err != nil {
|
||||
return busarchive.StreamEntry{}, nil, fmt.Errorf("its configuration: %w", err)
|
||||
}
|
||||
if err := json.Unmarshal(resp.State, &state); err != nil {
|
||||
return busarchive.StreamEntry{}, nil, fmt.Errorf("its state: %w", err)
|
||||
}
|
||||
// The archive reads to its end — s2 frames, then a tar — before it is called a snapshot.
|
||||
if _, err := f.Seek(0, io.SeekStart); err != nil {
|
||||
return busarchive.StreamEntry{}, nil, err
|
||||
}
|
||||
if err := busarchive.ReadsThrough(f); err != nil {
|
||||
return busarchive.StreamEntry{}, nil, fmt.Errorf("the archive the server sent does not read: %w", err)
|
||||
}
|
||||
|
||||
// The meta, in the shape the restore API takes and the nats CLI writes (backup.json): the
|
||||
// server's own configuration and state, kept as raw JSON so a field this program does not know
|
||||
// is never dropped on the way back.
|
||||
meta, err := json.MarshalIndent(map[string]json.RawMessage{"config": resp.Config, "state": resp.State}, "", " ")
|
||||
if err != nil {
|
||||
return busarchive.StreamEntry{}, nil, err
|
||||
}
|
||||
metaSum := sha256.Sum256(meta)
|
||||
return busarchive.StreamEntry{
|
||||
Name: name, Subjects: cfg.Subjects, Retention: cfg.Retention,
|
||||
Messages: state.Messages, Bytes: state.Bytes, FirstSeq: state.FirstSeq, LastSeq: state.LastSeq,
|
||||
Deleted: state.NumDelete, Consumers: state.Consumers,
|
||||
Meta: "streams/" + name + "/backup.json", MetaSHA256: hex.EncodeToString(metaSum[:]),
|
||||
Archive: "streams/" + name + "/stream.tar.s2", ArchiveBytes: written,
|
||||
ArchiveSHA256: hex.EncodeToString(hash.Sum(nil)), Seconds: time.Since(started).Seconds(),
|
||||
}, meta, nil
|
||||
}
|
||||
|
||||
// Human is a size in bytes as a person reads it: 1.5 MiB.
|
||||
func Human(n int64) string {
|
||||
const unit = 1024
|
||||
if n < unit {
|
||||
return fmt.Sprintf("%d B", n)
|
||||
}
|
||||
div, exp := int64(unit), 0
|
||||
for v := n / unit; v >= unit; v /= unit {
|
||||
div *= unit
|
||||
exp++
|
||||
}
|
||||
return fmt.Sprintf("%.1f %ciB", float64(n)/float64(div), "KMGTPE"[exp])
|
||||
}
|
||||
|
||||
// Into is the keeper's snapshot of some streams before it rebuilds or removes them (design 25 § 12.3): one tar under
|
||||
// dir, named for the streams and the time, written whole before its name is given, so a partial one is never taken
|
||||
// for it. It answers the file's path.
|
||||
func Into(nc *nats.Conn, dir string) func(ctx context.Context, streams []string) (string, error) {
|
||||
return func(ctx context.Context, streams []string) (string, error) {
|
||||
if dir == "" {
|
||||
return "", errors.New("no snapshots directory is set (MESH_NATS_SNAPSHOTS_DIR), so no snapshot is taken and nothing is removed")
|
||||
}
|
||||
name := filepath.Join(dir, "entry-"+strings.Join(streams, "+")+"-"+time.Now().UTC().Format("20060102T150405Z")+".tar")
|
||||
f, err := os.CreateTemp(dir, ".entry-*.partial")
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer os.Remove(f.Name())
|
||||
if _, err := Snapshot(ctx, nc, f, SnapshotOptions{Streams: streams, TempDir: dir}); err != nil {
|
||||
f.Close()
|
||||
return "", err
|
||||
}
|
||||
if err := f.Close(); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return name, os.Rename(f.Name(), name)
|
||||
}
|
||||
}
|
||||
@@ -3,6 +3,7 @@ module git.novox.be/novox/mesh-sdk/go
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
github.com/klauspost/compress v1.20.0
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
@@ -10,7 +11,6 @@ require (
|
||||
require (
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect
|
||||
github.com/google/go-tpm v0.9.8 // indirect
|
||||
github.com/klauspost/compress v1.20.0 // indirect
|
||||
github.com/minio/highwayhash v1.0.4 // indirect
|
||||
github.com/nats-io/jwt/v2 v2.8.1 // indirect
|
||||
github.com/nats-io/nkeys v0.4.16 // indirect
|
||||
|
||||
@@ -23,7 +23,7 @@ import (
|
||||
// Logic only some modules need is its owner module's package. A package added here is a decision to say in the pull
|
||||
// request: does every module share it, or is it a shape more than one part reads?
|
||||
var meshSDKPackages = []string{
|
||||
".", "announce", "announcetest", "buscalls", "buscompose", "busstate", "busretry", "calladdress", "catalogue",
|
||||
".", "announce", "announcetest", "busarchive", "buscalls", "buscompose", "busstate", "bussnap", "busretry", "calladdress", "catalogue",
|
||||
"declaration", "forgekey", "health", "internal/beside", "keys", "kvlimit", "membership", "motion",
|
||||
"objectlink", "outward", "pernode", "process",
|
||||
"sdkversion", "seal", "seatdef", "seatlist", "seatlist/seatlisttest", "stateobjects", "testbus", "views", "views/viewstest",
|
||||
|
||||
Reference in New Issue
Block a user