A-125: the restore of one stream lives in mesh-sdk's bussnap beside its snapshot

The bus module's keeper restores a protected snapshot in place and the recovery module restores its points: one copy, in mesh-sdk, imported by both. The snapshot takes the bus module's newest form (an entry's file named to the microsecond). Builds: design 47 G §655, A-125 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
2026-11-12 01:26:38 +01:00
parent 9545fad65b
commit d025dfd409
3 changed files with 222 additions and 1 deletions
@@ -0,0 +1,158 @@
+package bussnap
+
+import (
+	"archive/tar"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/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
+}
@@ -0,0 +1,63 @@
+//go:build integration
+
+package bussnap_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/bussnap"
+	"git.novox.be/novox/mesh-sdk/go/testbus"
+
+	"github.com/nats-io/nats.go"
+)
+
+// TestOneStreamIsTakenRemovedAndRestored: the take and restore of one stream is mesh-sdk's (design 47 G §655, A-125):
+// a stream's snapshot taken with Into, the stream deleted, and RestoreFile puts it back under its own name with its
+// messages; a second restore over the live name is refused by the server.
+func TestOneStreamIsTakenRemovedAndRestored(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	nc, err := nats.Connect(testbus.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer nc.Close()
+	js, err := nc.JetStream()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := js.AddStream(&nats.StreamConfig{Name: "KEPT_one", Subjects: []string{"kept.one.>"}}); err != nil {
+		t.Fatal(err)
+	}
+	for _, m := range []string{"a", "b", "c"} {
+		if _, err := js.Publish("kept.one."+m, []byte(m)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	file, err := bussnap.Into(nc, t.TempDir())(ctx, []string{"KEPT_one"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	m, err := bussnap.ManifestOf(file)
+	if err != nil || len(m.Streams) != 1 || m.Streams[0].Name != "KEPT_one" {
+		t.Fatalf("the manifest says %+v, %v; want the one stream KEPT_one", m, err)
+	}
+	if err := bussnap.DeleteStream(ctx, nc, "KEPT_one"); err != nil {
+		t.Fatal(err)
+	}
+	if live, err := bussnap.Live(ctx, nc, "KEPT_one"); err != nil || live {
+		t.Fatalf("live after its deletion: %v, %v", live, err)
+	}
+	if err := bussnap.RestoreFile(ctx, nc, file, t.TempDir(), t.Logf); err != nil {
+		t.Fatal(err)
+	}
+	info, err := js.StreamInfo("KEPT_one")
+	if err != nil || info.State.Msgs != 3 || info.State.LastSeq != 3 {
+		t.Fatalf("restored as %+v, %v; want 3 messages up to sequence 3", info, err)
+	}
+	if err := bussnap.RestoreFile(ctx, nc, file, t.TempDir(), t.Logf); err == nil {
+		t.Fatal("a restore over the live name was taken")
+	}
+}
@@ -415,7 +415,7 @@ func Into(nc *nats.Conn, dir string) func(ctx context.Context, streams []string)
 		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")
+		name := filepath.Join(dir, "entry-"+strings.Join(streams, "+")+"-"+time.Now().UTC().Format("20060102T150405.000000Z")+".tar")
 		f, err := os.CreateTemp(dir, ".entry-*.partial")
 		if err != nil {
 			return "", err