Move the restore of a bus archive into bussnap so the keeper restores a kept data too (milestone G, W14)

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ

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 (

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

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
+}