Merge branch 'feat/47-G-fix9-recovery' into feat/47-G-join10
This commit is contained in:
@@ -0,0 +1,87 @@
|
||||
package buscalls
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
// SeatChunks is the send of one seat verb's answer file in chunks (design 47 G §681, A-64, A-139), a verb whose seat
|
||||
// definition says `chunks: true`: the holder's process sends the nth chunk on mesh.seat.<seat>.chunk.<verb>.<call>.<n>,
|
||||
// and the caller acknowledges it on mesh.seat.<seat>.ack.<verb>.<call>.<n> within SeatAckWithin before the next is sent.
|
||||
// `call` is the 128 random bits the caller chose. The holder sends with it and the caller receives with it: one form.
|
||||
type SeatChunks struct {
|
||||
Seat, Verb, Call string
|
||||
}
|
||||
|
||||
// SeatChunkBytes is at most what one chunk carries: 512 KiB.
|
||||
const SeatChunkBytes = 512 << 10
|
||||
|
||||
// SeatAckWithin is how long the holder waits for a chunk's acknowledgement, and so how long a caller waits for the next chunk.
|
||||
const SeatAckWithin = 30 * time.Second
|
||||
|
||||
// SeatChunk is the body of one chunk: its number from 1, how many there are, and its bytes.
|
||||
type SeatChunk struct {
|
||||
N int64 `json:"n"`
|
||||
Of int64 `json:"of"`
|
||||
Data []byte `json:"data"`
|
||||
}
|
||||
|
||||
// Chunk is the subject of the nth chunk.
|
||||
func (c SeatChunks) Chunk(n int64) string {
|
||||
return "mesh.seat." + c.Seat + ".chunk." + c.Verb + "." + c.Call + "." + strconv.FormatInt(n, 10)
|
||||
}
|
||||
|
||||
// Ack is the subject the nth chunk is acknowledged on.
|
||||
func (c SeatChunks) Ack(n int64) string {
|
||||
return "mesh.seat." + c.Seat + ".ack." + c.Verb + "." + c.Call + "." + strconv.FormatInt(n, 10)
|
||||
}
|
||||
|
||||
// Chunks is every chunk subject of this call: what a caller subscribes to before it asks the verb.
|
||||
func (c SeatChunks) Chunks() string {
|
||||
return "mesh.seat." + c.Seat + ".chunk." + c.Verb + "." + c.Call + ".*"
|
||||
}
|
||||
|
||||
// SeatAcks is every acknowledgement of a verb's sends: what the holder subscribes to.
|
||||
func SeatAcks(seat, verb string) string {
|
||||
return "mesh.seat." + seat + ".ack." + verb + ".*.*"
|
||||
}
|
||||
|
||||
// ErrSeatChunkMissed is a chunk that did not come within its bound: the holder ended the send.
|
||||
var ErrSeatChunkMissed = errors.New("a chunk did not come within its bound, so the send ended")
|
||||
|
||||
// Receive reads `of` chunks in order, writes each one's bytes to w and acknowledges it before the next is sent. next
|
||||
// yields the subject and body of the next chunk heard on Chunks(), or ErrSeatChunkMissed at its bound (SeatAckWithin);
|
||||
// ack publishes on a subject. A chunk out of order, of another count or on another subject ends the receive, and so the
|
||||
// send, since it is never acknowledged. It answers the bytes written.
|
||||
func (c SeatChunks) Receive(of int64, next func() (subject string, body []byte, err error), ack func(subject string) error, w io.Writer) (int64, error) {
|
||||
var written int64
|
||||
for n := int64(1); n <= of; n++ {
|
||||
subject, body, err := next()
|
||||
if err != nil {
|
||||
return written, fmt.Errorf("chunk %d of %d: %w", n, of, err)
|
||||
}
|
||||
if subject != c.Chunk(n) {
|
||||
return written, fmt.Errorf("chunk %d of %d came on %s, not %s", n, of, subject, c.Chunk(n))
|
||||
}
|
||||
var ch SeatChunk
|
||||
if err := json.Unmarshal(body, &ch); err != nil {
|
||||
return written, fmt.Errorf("chunk %d of %d does not read: %w", n, of, err)
|
||||
}
|
||||
if ch.N != n || ch.Of != of {
|
||||
return written, fmt.Errorf("chunk %d of %d says it is %d of %d", n, of, ch.N, ch.Of)
|
||||
}
|
||||
k, err := w.Write(ch.Data)
|
||||
written += int64(k)
|
||||
if err != nil {
|
||||
return written, fmt.Errorf("chunk %d of %d cannot be written: %w", n, of, err)
|
||||
}
|
||||
if err := ack(c.Ack(n)); err != nil {
|
||||
return written, fmt.Errorf("chunk %d of %d cannot be acknowledged: %w", n, of, err)
|
||||
}
|
||||
}
|
||||
return written, nil
|
||||
}
|
||||
@@ -0,0 +1,72 @@
|
||||
package buscalls
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// TestSeatChunksReceive: the chunks of a send are written in order and each acknowledged before the next; a missed,
|
||||
// misplaced or miscounted chunk ends the receive unacknowledged (design 47 G §681, A-139).
|
||||
func TestSeatChunksReceive(t *testing.T) {
|
||||
c := SeatChunks{Seat: "mesh-broker", Verb: "snapshot-get", Call: "0123456789abcdef0123456789abcdef"}
|
||||
if c.Chunk(2) != "mesh.seat.mesh-broker.chunk.snapshot-get.0123456789abcdef0123456789abcdef.2" ||
|
||||
c.Ack(2) != "mesh.seat.mesh-broker.ack.snapshot-get.0123456789abcdef0123456789abcdef.2" ||
|
||||
c.Chunks() != "mesh.seat.mesh-broker.chunk.snapshot-get.0123456789abcdef0123456789abcdef.*" ||
|
||||
SeatAcks("mesh-broker", "snapshot-get") != "mesh.seat.mesh-broker.ack.snapshot-get.*.*" {
|
||||
t.Fatalf("subjects: %s %s %s", c.Chunk(2), c.Ack(2), c.Chunks())
|
||||
}
|
||||
type sent struct {
|
||||
subject string
|
||||
body SeatChunk
|
||||
missed bool
|
||||
}
|
||||
good := []sent{{subject: c.Chunk(1), body: SeatChunk{N: 1, Of: 3, Data: []byte("ab")}},
|
||||
{subject: c.Chunk(2), body: SeatChunk{N: 2, Of: 3, Data: []byte("cd")}},
|
||||
{subject: c.Chunk(3), body: SeatChunk{N: 3, Of: 3, Data: []byte("e")}}}
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
sends []sent
|
||||
want string
|
||||
acks int
|
||||
err string
|
||||
}{
|
||||
{name: "three chunks", sends: good, want: "abcde", acks: 3},
|
||||
{name: "a missed chunk", sends: append(good[:1:1], sent{missed: true}), want: "ab", acks: 1, err: "chunk 2 of 3"},
|
||||
{name: "out of order", sends: []sent{good[1]}, acks: 0, err: "came on"},
|
||||
{name: "another count", sends: []sent{{subject: c.Chunk(1), body: SeatChunk{N: 1, Of: 2, Data: []byte("x")}}}, err: "says it is 1 of 2"},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
i, acks := 0, 0
|
||||
next := func() (string, []byte, error) {
|
||||
if i >= len(tc.sends) || tc.sends[i].missed {
|
||||
return "", nil, ErrSeatChunkMissed
|
||||
}
|
||||
s := tc.sends[i]
|
||||
i++
|
||||
body, _ := json.Marshal(s.body)
|
||||
return s.subject, body, nil
|
||||
}
|
||||
ack := func(subject string) error {
|
||||
acks++
|
||||
if subject != c.Ack(int64(acks)) {
|
||||
t.Fatalf("ack %d on %s", acks, subject)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
var out bytes.Buffer
|
||||
n, err := c.Receive(3, next, ack, &out)
|
||||
if tc.err == "" && err != nil || tc.err != "" && (err == nil || !strings.Contains(err.Error(), tc.err)) {
|
||||
t.Fatalf("err %v, want %q", err, tc.err)
|
||||
}
|
||||
if tc.name == "a missed chunk" && !errors.Is(err, ErrSeatChunkMissed) {
|
||||
t.Fatalf("a missed chunk is not ErrSeatChunkMissed: %v", err)
|
||||
}
|
||||
if out.String() != tc.want || n != int64(len(tc.want)) || acks != tc.acks {
|
||||
t.Fatalf("wrote %q (%d), %d acks; want %q, %d acks", out.String(), n, acks, tc.want, tc.acks)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user