584 lines
23 KiB
Go
584 lines
23 KiB
Go
//go:build integration
|
|
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"os"
|
|
"slices"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats-server/v2/server"
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
|
|
"git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
|
|
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
|
|
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
|
"git.novox.be/novox/mesh-sdk/go/process"
|
|
"git.novox.be/novox/mesh-sdk/go/views"
|
|
|
|
"nats-tools/internal/bussnap"
|
|
"nats-tools/internal/keeper"
|
|
)
|
|
|
|
// kvStore is the keeper's own state on the real bus: one bucket per state, as the grants part creates its old sections.
|
|
type kvStore struct {
|
|
js jetstream.JetStream
|
|
}
|
|
|
|
func (s kvStore) kv(state string) (jetstream.KeyValue, error) {
|
|
return s.js.KeyValue(context.Background(), "nats_"+state)
|
|
}
|
|
|
|
func (s kvStore) Get(state, key string) ([]byte, uint64, bool, error) {
|
|
kv, err := s.kv(state)
|
|
if err != nil {
|
|
return nil, 0, false, err
|
|
}
|
|
e, err := kv.Get(context.Background(), key)
|
|
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
return nil, 0, false, nil
|
|
}
|
|
if err != nil {
|
|
return nil, 0, false, err
|
|
}
|
|
return e.Value(), e.Revision(), true, nil
|
|
}
|
|
|
|
func (s kvStore) Put(state, key string, value []byte) (uint64, error) {
|
|
kv, err := s.kv(state)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return kv.Put(context.Background(), key, value)
|
|
}
|
|
|
|
func (s kvStore) Delete(state, key string) error {
|
|
kv, err := s.kv(state)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return kv.Delete(context.Background(), key)
|
|
}
|
|
|
|
func (s kvStore) Create(state, key string, value []byte) (uint64, error) {
|
|
kv, err := s.kv(state)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
rev, err := kv.Create(context.Background(), key, value)
|
|
if errors.Is(err, jetstream.ErrKeyExists) {
|
|
return 0, keeper.ErrExists
|
|
}
|
|
return rev, err
|
|
}
|
|
|
|
func (s kvStore) Keys(state string) ([]string, error) {
|
|
kv, err := s.kv(state)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
keys, err := kv.Keys(context.Background())
|
|
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
|
return nil, nil
|
|
}
|
|
return keys, err
|
|
}
|
|
|
|
func eventually(t *testing.T, what string, ok func() bool) {
|
|
t.Helper()
|
|
deadline := time.Now().Add(10 * time.Second)
|
|
for !ok() {
|
|
if time.Now().After(deadline) {
|
|
t.Fatalf("not within 10 s: %s", what)
|
|
}
|
|
time.Sleep(20 * time.Millisecond)
|
|
}
|
|
}
|
|
|
|
// The bus module runs whole (design 25 § 12.7, "The bus module runs whole" and "The bus module's first view"; CAT-N10):
|
|
// its keeper against the real bus server, hearing manifest-accepted on its durable consumer, creating on the bus,
|
|
// saying its events on its own stream BROKER_EVENTS, keeping its state in its own buckets, and acting on a person's
|
|
// requests. One server for every case, in order, as the mesh runs them.
|
|
func TestTheBusModuleRunsWhole(t *testing.T) {
|
|
dir := t.TempDir()
|
|
s, err := server.NewServer(&server.Options{Port: -1, JetStream: true, StoreDir: dir, NoLog: true, NoSigs: true})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
go s.Start()
|
|
if !s.ReadyForConnections(5 * time.Second) {
|
|
t.Fatal("the server did not start")
|
|
}
|
|
defer s.Shutdown()
|
|
nc, err := nats.Connect(s.ClientURL())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer nc.Close()
|
|
js, _ := jetstream.New(nc)
|
|
ctx := context.Background()
|
|
account := process.Account{Node: "anchor", Module: keeper.Module, User: "anchor.nats.keeper"}
|
|
// What the grants part creates for the bus module from its manifest's old sections: EVENTS and the keeper's
|
|
// consumer on it, its own stream, and its two states.
|
|
for _, c := range []jetstream.StreamConfig{
|
|
{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}},
|
|
{Name: BrokerEventsStream, Subjects: []string{"mesh.stream.nats.event.>"}, MaxAge: 30 * 24 * time.Hour, MaxBytes: 256 << 20},
|
|
{Name: "CONDITIONS", Subjects: []string{"mesh.seat.operator-channel.accept.condition.>"}},
|
|
} {
|
|
if _, err := js.CreateStream(ctx, c); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
// The keeper's consumer hears the indexer's verdicts and the operator channel's decided for the bus module's own
|
|
// asks, as the grants part composes it from its consumes and the seats it uses.
|
|
if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: account.QueueConsumerName(),
|
|
FilterSubjects: []string{AcceptedSubject, DecidedSubject}, AckPolicy: jetstream.AckExplicitPolicy, MaxDeliver: 5}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// The router's own record of the asks, which the keeper reads under its own process account (process.Conn.Ask), never
|
|
// a stand-in: an ask's record is `<asker>.<ask id>`, as the router keeps it.
|
|
asksKV, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: asksBucket, History: 5})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
pconn, err := process.Connect(process.Account{URL: s.ClientURL(), Node: account.Node, Module: account.Module, User: account.User, Password: "p"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer pconn.Close()
|
|
for _, st := range []string{keeper.StateAsked, keeper.StateRequests, keeper.StateSpent, "findings"} {
|
|
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "nats_" + st, History: 5}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
store := kvStore{js}
|
|
snapshots := t.TempDir()
|
|
var mu sync.Mutex
|
|
view := keeper.FirstView{Revision: 1}
|
|
var acts []string
|
|
// The bus module's findings go through a held writer over its state findings, started as the keeper's process starts it.
|
|
writer, err := views.NewHeldWriter(views.WriterConfig[meshconditions.Finding, meshconditions.FindingCleared, meshconditions.FindingsStated, meshconditions.FindingsAskedEvent]{
|
|
Asked: meshconditions.SubjectFindingsAsked, Module: keeper.Module, Node: account.Node, State: pconn.State("findings"),
|
|
Kinds: []string{keeper.FindingSnapshotsFilling}, JS: js, Conn: nc, Logf: t.Logf})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := writer.Start(ctx); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer writer.Close()
|
|
newKeeper := func() *keeper.Keeper {
|
|
return &keeper.Keeper{Bus: keeper.JetStream{JS: js, Snap: bussnap.Into(nc, snapshots), NC: nc, Dir: snapshots, Log: t.Logf}, Store: store,
|
|
Finding: writer, Emit: streamEmitter(js, account), Say: sayer(js),
|
|
View: func() (keeper.FirstView, error) { mu.Lock(); defer mu.Unlock(); return view, nil },
|
|
Act: func(what, why, cause string) error {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
acts = append(acts, cause+": "+what)
|
|
return nil
|
|
},
|
|
Warrant: WarrantFrom(pconn.Ask),
|
|
Logf: t.Logf}
|
|
}
|
|
run := func(k *keeper.Keeper) func() {
|
|
kctx, stop := context.WithCancel(ctx)
|
|
done := make(chan struct{})
|
|
watch := func(c context.Context, act func()) error {
|
|
kv, _ := js.KeyValue(ctx, "nats_"+keeper.StateRequests)
|
|
w, err := kv.WatchAll(c)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
go func() {
|
|
for range w.Updates() {
|
|
act()
|
|
}
|
|
}()
|
|
return nil
|
|
}
|
|
go func() { _ = Keep(kctx, k, hearer(js, account, t.Logf), watch, t.Logf); close(done) }()
|
|
return func() { stop(); <-done }
|
|
}
|
|
accepted := func(module string, revision uint64, entries ...catalogue.BusStateEntry) {
|
|
body, _ := json.Marshal(map[string]any{"v": 1, "module": module, "digest": fmt.Sprintf("sha256:%d", revision),
|
|
"judge": "j", "revision": revision, "bus-state": entries})
|
|
if _, err := js.Publish(ctx, AcceptedSubject, body); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
// saidOn reads what one stream holds of the bus module's events, by its subjects' prefix: its own stream (its
|
|
// history), or EVENTS on the seat mesh-broker's subjects (what every consumer of mesh-broker.<event> hears).
|
|
saidOn := func(stream, prefix string) []string {
|
|
cons, err := js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{FilterSubjects: []string{prefix + ">"}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
info, _ := cons.Info(ctx)
|
|
var out []string
|
|
if info.NumPending == 0 {
|
|
return nil
|
|
}
|
|
batch, _ := cons.Fetch(int(info.NumPending), jetstream.FetchMaxWait(2*time.Second))
|
|
for m := range batch.Messages() {
|
|
var e keeper.Event
|
|
_ = json.Unmarshal(m.Data(), &e)
|
|
out = append(out, strings.TrimPrefix(m.Subject(), prefix)+" "+e.Module+" "+e.Local)
|
|
}
|
|
return out
|
|
}
|
|
said := func() []string { return saidOn(BrokerEventsStream, "mesh.stream.nats.event.") }
|
|
saidHas := func(want string) func() bool {
|
|
return func() bool {
|
|
for _, l := range said() {
|
|
if l == want {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
}
|
|
held := func(stream string) (*jetstream.StreamInfo, bool) {
|
|
st, err := js.Stream(ctx, stream)
|
|
if err != nil {
|
|
return nil, false
|
|
}
|
|
info, err := st.Info(ctx)
|
|
return info, err == nil
|
|
}
|
|
state := entry("drafts", "key-value", "state", nil)
|
|
log := entry("log", "stream", "log", map[string]string{"max-mib": "1"})
|
|
|
|
// Review round 1 (running F1): the fill pass runs from the start, while the first view is still asked, so the store's
|
|
// one guard never waits on the indexer.
|
|
t.Run("the fill pass runs while the first view is asked", func(t *testing.T) {
|
|
down := newKeeper()
|
|
down.View = func() (keeper.FirstView, error) { return keeper.FirstView{}, errors.New("no responders") }
|
|
stopDown := run(down)
|
|
eventually(t, "a fill pass with no first view", func() bool { return !down.Status().Pass.IsZero() && down.Status().FirstView.Answered.IsZero() })
|
|
stopDown()
|
|
// And while its consumer of manifest-accepted is bound again on its backoff (review round 2, running F1).
|
|
was := FillEvery
|
|
FillEvery = 20 * time.Millisecond
|
|
defer func() { FillEvery = was }()
|
|
deaf := newKeeper()
|
|
kctx, stopDeaf := context.WithCancel(ctx)
|
|
done := make(chan struct{})
|
|
go func() {
|
|
_ = Keep(kctx, deaf, func(context.Context, func(string, []byte) error) error { return errors.New("no consumer yet") }, nil, t.Logf)
|
|
close(done)
|
|
}()
|
|
eventually(t, "the first view answered", func() bool { return !deaf.Status().FirstView.Answered.IsZero() })
|
|
first := deaf.Status().Pass
|
|
eventually(t, "a fill pass while the consumer is bound again", func() bool { return deaf.Status().Pass.After(first) })
|
|
stopDeaf()
|
|
<-done
|
|
})
|
|
k := newKeeper()
|
|
stop := run(k)
|
|
t.Run("a verdict, then created, with its binding", func(t *testing.T) {
|
|
accepted("m", 2, state, log)
|
|
eventually(t, "created m drafts and m log", func() bool { return saidHas("created m drafts")() && saidHas("created m log")() })
|
|
// Said on the seat's subject as well, on the shared events stream (the lead's ruling, 13:0x).
|
|
if seat := saidOn("EVENTS", "mesh.seat.mesh-broker.event."); !slices.Contains(seat, "created m drafts") ||
|
|
!slices.Contains(seat, "created m log") {
|
|
t.Errorf("EVENTS holds %v of mesh-broker's events: created m drafts and m log, as on the bus module's own stream", seat)
|
|
}
|
|
if _, ok := held("KV_m_drafts"); !ok {
|
|
t.Error("the bus holds no KV_m_drafts")
|
|
}
|
|
if info, ok := held("m_log"); !ok || info.Config.Discard != jetstream.DiscardNew || info.Config.Subjects[0] != "mesh.bus.m.log.>" {
|
|
t.Errorf("m_log is %+v: a log refuses when full, on the module's own subjects", info)
|
|
}
|
|
})
|
|
t.Run("a restart catches up a verdict given while it was down", func(t *testing.T) {
|
|
stop()
|
|
accepted("n", 3, entry("jobs", "stream", "work-queue", nil))
|
|
k = newKeeper()
|
|
stop = run(k)
|
|
eventually(t, "created n jobs", saidHas("created n jobs"))
|
|
})
|
|
t.Run("a replay older than the first view is not applied; a set-aside module keeps its key", func(t *testing.T) {
|
|
stop()
|
|
mu.Lock()
|
|
view = keeper.FirstView{Revision: 9, Modules: []keeper.Accepted{{Module: "p", Revision: 9, BusState: []catalogue.BusStateEntry{state}}}}
|
|
mu.Unlock()
|
|
accepted("p", 4)
|
|
k = newKeeper()
|
|
stop = run(k)
|
|
eventually(t, "created p drafts", saidHas("created p drafts"))
|
|
eventually(t, "the replay heard", func() bool {
|
|
info, _ := js.Consumer(ctx, "EVENTS", account.QueueConsumerName())
|
|
i, _ := info.Info(ctx)
|
|
return i.NumPending == 0 && i.NumAckPending == 0
|
|
})
|
|
if saidHas("orphaned p drafts")() {
|
|
t.Error("the replay at revision 4 orphaned what the first view at 9 asked")
|
|
}
|
|
if _, _, found, _ := store.Get(keeper.StateAsked, "m"); !found {
|
|
t.Error("m, missing from the first view, lost its key")
|
|
}
|
|
})
|
|
t.Run("drift is said, a growth applied and the drift cleared", func(t *testing.T) {
|
|
accepted("m", 5, state, entry("log", "stream", "log", map[string]string{"max-mib": "1", "max-age-days": "2"}))
|
|
eventually(t, "drifted m log", saidHas("drifted m log"))
|
|
accepted("m", 6, state, entry("log", "stream", "log", map[string]string{"max-mib": "4"}))
|
|
eventually(t, "drift-cleared m log", saidHas("drift-cleared m log"))
|
|
if info, _ := held("m_log"); info.Config.MaxBytes != 4<<20 {
|
|
t.Errorf("m_log's cap is %d: the larger cap applied", info.Config.MaxBytes)
|
|
}
|
|
})
|
|
t.Run("a refused create hands no binding", func(t *testing.T) {
|
|
bare := entry("raw", "stream", "", nil)
|
|
accepted("m", 7, state, entry("log", "stream", "log", map[string]string{"max-mib": "4"}), bare)
|
|
eventually(t, "refused m raw", saidHas("refused m raw"))
|
|
if saidHas("created m raw")() {
|
|
t.Error("created m raw")
|
|
}
|
|
})
|
|
t.Run("each template at a small cap refuses, full raised at the raise mark and cleared below the clear mark", func(t *testing.T) {
|
|
for i := 0; ; i++ {
|
|
_, err := js.Publish(ctx, fmt.Sprintf("mesh.bus.m.log.k%d", i), make([]byte, 200<<10))
|
|
if err != nil {
|
|
break
|
|
}
|
|
if i > 40 {
|
|
t.Fatal("a 4 MiB log took more than 8 MiB")
|
|
}
|
|
}
|
|
k.FillPass()
|
|
eventually(t, "full m log", saidHas("full m log"))
|
|
st, _ := js.Stream(ctx, "m_log")
|
|
_ = st.Purge(ctx)
|
|
k.FillPass()
|
|
eventually(t, "full-cleared m log", saidHas("full-cleared m log"))
|
|
})
|
|
t.Run("an orphan is listed, and goes only by drop on the operator's warrant, after a snapshot", func(t *testing.T) {
|
|
accepted("m", 8, entry("log", "stream", "log", map[string]string{"max-mib": "4"}))
|
|
eventually(t, "orphaned m drafts", saidHas("orphaned m drafts"))
|
|
put := func(r keeper.Request) {
|
|
raw, _ := json.Marshal(r)
|
|
if _, err := store.Put(keeper.StateRequests, r.ID, raw); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
get := func(id string) keeper.Request {
|
|
raw, _, _, _ := store.Get(keeper.StateRequests, id)
|
|
var r keeper.Request
|
|
_ = json.Unmarshal(raw, &r)
|
|
return r
|
|
}
|
|
put(keeper.Request{ID: "r1", Verb: keeper.VerbDrop, Module: "m", Local: "drafts", Caller: "person.jo", Why: "unused"})
|
|
eventually(t, "the drop with no ask refused", func() bool { return strings.Contains(get("r1").Error, "operator's warrant") })
|
|
r := keeper.Request{ID: "r2", Verb: keeper.VerbDrop, Module: "m", Local: "drafts", Caller: "person.jo", Why: "unused", At: time.Now()}
|
|
a, _ := askFor(r, time.Now())
|
|
r.Ask = a.ID
|
|
record := func(w *oc.Warrant) {
|
|
raw, _ := json.Marshal(map[string]any{"asker": keeper.Module, "ask": a, "state": "open", "warrant": w})
|
|
if _, err := asksKV.Put(ctx, keeper.Module+"."+a.ID, raw); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
record(nil)
|
|
put(r)
|
|
eventually(t, "the request read, waiting for the operator", func() bool { return get("r2").Waiting != "" })
|
|
if _, ok := held("KV_m_drafts"); !ok || !get("r2").Done.IsZero() {
|
|
t.Fatal("dropped before the operator chose")
|
|
}
|
|
// The operator chooses: the router records the warrant and says decided; the keeper acts on decided, on its
|
|
// durable consumer, with no timer between and no call of its pass by the test (seams F1).
|
|
w := &oc.Warrant{Ask: a.ID, Asker: keeper.Module, Outcome: occ.OutcomeChosen, By: &oc.Person{Who: occ.Operator},
|
|
Option: "go", Label: "Remove", Level: occ.Destroy, At: occ.FormatTime(time.Now()), AskDigest: occ.Digest(a)}
|
|
record(w)
|
|
decided := time.Now()
|
|
body, _ := json.Marshal(w)
|
|
if _, err := js.Publish(ctx, DecidedSubject, body); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
eventually(t, "dropped m drafts", saidHas("dropped m drafts"))
|
|
if since := time.Since(decided); since > 5*time.Second {
|
|
t.Errorf("dropped %s after decided: within seconds", since)
|
|
}
|
|
if _, ok := held("KV_m_drafts"); ok {
|
|
t.Error("KV_m_drafts is still on the bus")
|
|
}
|
|
if p := get("r2").Snapshot; p == "" {
|
|
t.Error("no snapshot taken before the drop")
|
|
} else if st, err := os.Stat(p); err != nil || st.Size() == 0 {
|
|
t.Errorf("the snapshot %s is not there whole: %v", p, err)
|
|
}
|
|
mu.Lock()
|
|
recorded := strings.Join(acts, "\n")
|
|
mu.Unlock()
|
|
if !strings.Contains(recorded, "nats_drop: dropped drafts of m") {
|
|
t.Errorf("recorded %s: the drop as a hand act", recorded)
|
|
}
|
|
// The same warrant handed in a second request is refused and deletes nothing (security F2): drafts asked for and
|
|
// orphaned again, then a request naming the spent ask.
|
|
accepted("m", 9, state, entry("log", "stream", "log", map[string]string{"max-mib": "4"}))
|
|
eventually(t, "m drafts made again", func() bool { _, ok := held("KV_m_drafts"); return ok })
|
|
accepted("m", 10, entry("log", "stream", "log", map[string]string{"max-mib": "4"}))
|
|
eventually(t, "m drafts orphaned again", func() bool { return k.Held()["m"].Made["drafts"].Orphaned })
|
|
again := keeper.Request{ID: "r2b", Verb: keeper.VerbDrop, Module: "m", Local: "drafts", Caller: "person.jo", Why: "again", Ask: a.ID, At: time.Now()}
|
|
put(again)
|
|
eventually(t, "the second request on the spent warrant refused", func() bool { return strings.Contains(get("r2b").Error, "already acted") })
|
|
if _, ok := held("KV_m_drafts"); !ok {
|
|
t.Error("the spent warrant deleted KV_m_drafts a second time")
|
|
}
|
|
})
|
|
// Design 47 G §681: the snapshot the keeper wrote is read by the bundle's snapshot-get, sent in chunks that hold its bytes,
|
|
// restored in place by the keeper (refused while its name is live, ended by snapshot-restored) and removed by snapshot-remove.
|
|
t.Run("a snapshot is sent by the bundle, restored in place by the keeper and removed", func(t *testing.T) {
|
|
get := func(id string) keeper.Request {
|
|
raw, _, _, _ := store.Get(keeper.StateRequests, id)
|
|
var r keeper.Request
|
|
_ = json.Unmarshal(raw, &r)
|
|
return r
|
|
}
|
|
rec := get("r2")
|
|
if rec.Path == "" || rec.Digest == "" || rec.Protected {
|
|
t.Fatalf("the drop's record holds no snapshot: %+v", rec)
|
|
}
|
|
rs := requestStore{
|
|
Put: func(key string, r keeper.Request) error {
|
|
raw, _ := json.Marshal(r)
|
|
_, err := store.Put(keeper.StateRequests, key, raw)
|
|
return err
|
|
},
|
|
Get: func(key string) (*keeper.Request, error) {
|
|
if _, _, found, _ := store.Get(keeper.StateRequests, key); !found {
|
|
return nil, nil
|
|
}
|
|
r := get(key)
|
|
return &r, nil
|
|
},
|
|
Keys: func() ([]string, error) { return store.Keys(keeper.StateRequests) },
|
|
Now: func() time.Time { return time.Now().UTC() },
|
|
Ended: func(id string, within time.Duration) *keeper.Request {
|
|
for deadline := time.Now().Add(within); time.Now().Before(deadline); time.Sleep(20 * time.Millisecond) {
|
|
if r := get(id); !r.Done.IsZero() {
|
|
return &r
|
|
}
|
|
}
|
|
return nil
|
|
},
|
|
}
|
|
var chunkMu sync.Mutex
|
|
var chunks []Chunk
|
|
var ack func(string, json.RawMessage) (any, error)
|
|
world := &snapshotWorld{
|
|
Open: func(path string) (io.ReadCloser, error) { return os.Open(path) },
|
|
Granted: func(string, string) bool { return true }, Logf: t.Logf, Within: 5 * time.Second, KeepDays: 90,
|
|
Answer: func(_ string, h func(string, json.RawMessage) (any, error)) error { ack = h; return nil },
|
|
}
|
|
world.Publish = func(subject string, body any, id string) error {
|
|
c := body.(Chunk)
|
|
chunkMu.Lock()
|
|
chunks = append(chunks, c)
|
|
chunkMu.Unlock()
|
|
go func() { _, _ = ack(strings.Replace(subject, ".chunk.", ".ack.", 1), nil) }()
|
|
return nil
|
|
}
|
|
tl := map[string]stdio.Tool{}
|
|
for _, x := range snapshotTools(rs, world) {
|
|
tl[x.Name] = x
|
|
}
|
|
call := "0123456789abcdef0123456789abcdef"
|
|
out, err := tl["mesh-broker.snapshot-get"].Run(map[string]any{"request": "r2", "call": call})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if a := out.(meshbroker.SnapshotGetAnswer); a.Error != "" || a.Digest != rec.Digest || a.Chunks < 1 || a.Stream != "m_drafts" {
|
|
t.Fatalf("snapshot-get answered %+v for the keeper's record %+v", a, rec)
|
|
}
|
|
eventually(t, "the chunks sent", func() bool { chunkMu.Lock(); defer chunkMu.Unlock(); return len(chunks) > 0 })
|
|
var whole []byte
|
|
eventually(t, "the last chunk", func() bool {
|
|
chunkMu.Lock()
|
|
defer chunkMu.Unlock()
|
|
return int64(len(chunks)) == chunks[0].Of
|
|
})
|
|
chunkMu.Lock()
|
|
for _, c := range chunks {
|
|
whole = append(whole, c.Data...)
|
|
}
|
|
chunkMu.Unlock()
|
|
if sum := sha256.Sum256(whole); "sha256:"+hex.EncodeToString(sum[:]) != rec.Digest || int64(len(whole)) != rec.Size {
|
|
t.Errorf("the chunks hold %d bytes, not the snapshot the keeper wrote (%d)", len(whole), rec.Size)
|
|
}
|
|
restore := func(why string) meshbroker.SnapshotRestoreAnswer {
|
|
o, err := tl["mesh-broker.snapshot-restore"].Run(map[string]any{"request": "r2", "why": why})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return o.(meshbroker.SnapshotRestoreAnswer)
|
|
}
|
|
// KV_m_drafts is live (made again after the drop): nothing is restored over it.
|
|
if a := restore("live"); !strings.Contains(a.Error, "names live") {
|
|
t.Errorf("a restore over a live stream answered %+v", a)
|
|
}
|
|
if err := js.DeleteStream(ctx, "KV_m_drafts"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
a := restore("it was removed by mistake")
|
|
if a.Accepted == "" {
|
|
t.Fatalf("restore answered %+v", a)
|
|
}
|
|
eventually(t, "snapshot-restored on the seat's subject", func() bool {
|
|
for _, l := range saidOn("EVENTS", "mesh.seat.mesh-broker.event.snapshot-restored.") {
|
|
if strings.HasPrefix(l, "r2."+a.Accepted+" ") {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
})
|
|
if _, ok := held("KV_m_drafts"); !ok || get(a.Accepted).Result != "done" {
|
|
t.Errorf("the stream is not restored in place: held %v, the act %+v", ok, get(a.Accepted))
|
|
}
|
|
o, err := tl["mesh-broker.snapshot-remove"].Run(map[string]any{"request": "r2", "why": "no longer needed"})
|
|
if err != nil || o.(meshbroker.SnapshotRemoveAnswer).Accepted == "" {
|
|
t.Fatalf("snapshot-remove answered %+v, %v", o, err)
|
|
}
|
|
if _, err := os.Stat(rec.Path); !os.IsNotExist(err) {
|
|
t.Errorf("the snapshot file %s is still there: %v", rec.Path, err)
|
|
}
|
|
if _, _, found, _ := store.Get(keeper.StateRequests, "r2"); found {
|
|
t.Error("the snapshot's record outlived its file")
|
|
}
|
|
})
|
|
t.Run("nats_first_view_again asks at once and records the hand act", func(t *testing.T) {
|
|
raw, _ := json.Marshal(keeper.Request{ID: "r3", Verb: keeper.VerbFirstViewAgain, Why: "a verdict thought missed", Caller: "person.jo"})
|
|
_, _ = store.Put(keeper.StateRequests, "r3", raw)
|
|
eventually(t, "the first view asked again", func() bool {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
return strings.Contains(strings.Join(acts, "\n"), "nats_first_view_again")
|
|
})
|
|
})
|
|
stop()
|
|
}
|
|
|
|
func entry(name, storage, template string, options map[string]string) catalogue.BusStateEntry {
|
|
e := catalogue.BusStateEntry{Name: name, Storage: storage, Template: template}
|
|
if len(options) > 0 {
|
|
e.Options = map[string]json.RawMessage{}
|
|
for k, v := range options {
|
|
e.Options[k] = json.RawMessage(v)
|
|
}
|
|
}
|
|
return e
|
|
}
|