novox/mesh-catalog
9
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source 
Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
jschoubben
2026-11-11 12:05:54 +01:00
parent d64e9f3818
commit bbd9f638a0
3 changed files with 128 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/nats/cmd/mesh-nats-snapshot/bussnap.go
+2
View File
@@ -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
-114
View File
@@ -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
+126
View File
@@ -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
}