NATS10: the bus module serves no restore point tool and keeps no state kept
Revertsbbd9f638,c2e017a4and b6884796; cmd/mesh-nats-snapshot/restore.go is as before them. The manifest test asserts the claim serves only the reading verbs and the keeper's state is asked, requests, spent. Builds: design 47 G §554, §555 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -14,8 +14,6 @@ var (
|
||||
StreamNames = bussnap.StreamNames
|
||||
request = bussnap.Request
|
||||
truncate = bussnap.Truncate
|
||||
RestoreAll = bussnap.RestoreAll
|
||||
heldTo = bussnap.HeldTo
|
||||
)
|
||||
|
||||
const (
|
||||
|
||||
@@ -4,8 +4,10 @@ import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"os"
|
||||
"os/exec"
|
||||
@@ -18,6 +20,118 @@ 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
|
||||
|
||||
@@ -236,10 +236,8 @@ func runKeeper(ctx context.Context, getenv func(string) string, logOut io.Writer
|
||||
defer nc.Close()
|
||||
recorder := meshhandactsclient.Recorder{Ask: conn.Call}
|
||||
k := &keeper.Keeper{
|
||||
Bus: keeper.JetStream{JS: js, Snap: bussnap.Into(nc, getenv("MESH_NATS_SNAPSHOTS_DIR"))},
|
||||
Kept: keeper.KeptJetStream{NC: nc, JS: js, Logf: logger.Printf},
|
||||
KeptDir: getenv("MESH_NATS_SNAPSHOTS_DIR"),
|
||||
Store: processStore{conn}, Emit: streamEmitter(js, account), Say: sayer(js), View: firstView(conn),
|
||||
Bus: keeper.JetStream{JS: js, Snap: bussnap.Into(nc, getenv("MESH_NATS_SNAPSHOTS_DIR"))},
|
||||
Store: processStore{conn}, Emit: streamEmitter(js, account), Say: sayer(js), View: firstView(conn),
|
||||
Act: func(what, why, cause string) error {
|
||||
_, err := recorder.Record(meshhandactsclient.Request{What: what, Why: why, Cause: cause})
|
||||
return err
|
||||
|
||||
@@ -1,153 +0,0 @@
|
||||
//go:build integration
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats-server/v2/server"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"nats-tools/internal/bussnap"
|
||||
"nats-tools/internal/keeper"
|
||||
)
|
||||
|
||||
// mapStore is the keeper's own state in memory.
|
||||
type mapStore struct {
|
||||
mu sync.Mutex
|
||||
kv map[string][]byte
|
||||
}
|
||||
|
||||
func (s *mapStore) Get(state, key string) ([]byte, uint64, bool, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
v, ok := s.kv[state+"/"+key]
|
||||
return v, 1, ok, nil
|
||||
}
|
||||
func (s *mapStore) Put(state, key string, v []byte) (uint64, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.kv[state+"/"+key] = v
|
||||
return 1, nil
|
||||
}
|
||||
func (s *mapStore) Delete(state, key string) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
delete(s.kv, state+"/"+key)
|
||||
return nil
|
||||
}
|
||||
func (s *mapStore) Keys(string) ([]string, error) { return nil, nil }
|
||||
func (s *mapStore) Create(state, key string, v []byte) (uint64, error) {
|
||||
return 0, errors.New("not used")
|
||||
}
|
||||
|
||||
// A module's kept data on the real bus is backed up and restored beside the live data (design 47, "A module's kept data on
|
||||
// the bus, backed up and restored with its delivery"): the bucket as the server's own snapshot, the module's consumer on
|
||||
// another stream at its position; after a change, a restore brings the bucket and the position back, and the live bucket
|
||||
// is kept whole in the data set aside.
|
||||
func TestAKeptDataIsRestoredBesideTheLiveDataOnTheRealBus(t *testing.T) {
|
||||
s, err := server.NewServer(&server.Options{Port: -1, JetStream: true, StoreDir: t.TempDir(), 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, err := jetstream.New(nc)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx := context.Background()
|
||||
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "idx_modules"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
forge, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "forge", Subjects: []string{"forge.>"}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for i := 0; i < 10; i++ {
|
||||
if _, err := js.Publish(ctx, "forge.merged", []byte("m")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
cons, err := forge.CreateConsumer(ctx, jetstream.ConsumerConfig{Durable: "ace_idx", AckPolicy: jetstream.AckExplicitPolicy})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
batch, err := cons.Fetch(4)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for m := range batch.Messages() {
|
||||
_ = m.DoubleAck(ctx)
|
||||
}
|
||||
if _, err := kv.PutString(ctx, "ace", "fingerprint-1"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
dir := t.TempDir()
|
||||
var acts []string
|
||||
k := &keeper.Keeper{Bus: keeper.JetStream{JS: js, Snap: bussnap.Into(nc, dir)}, Store: &mapStore{kv: map[string][]byte{}},
|
||||
Kept: keeper.KeptJetStream{NC: nc, JS: js}, KeptDir: dir,
|
||||
Act: func(what, why, cause string) error { acts = append(acts, what); return nil }}
|
||||
take := &keeper.Request{Verb: keeper.VerbSnapshotKept, Module: "idx", Local: "modules", Form: 0, Fingerprint: "abc123", CatchUp: "indexer_take_in", Why: "before form 1"}
|
||||
if _, err := k.SnapshotKept(ctx, take); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The form-1 build rewrites the kept data, and its consumer moves on.
|
||||
if _, err := kv.PutString(ctx, "ace", "fingerprint-2-form-1"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
batch, err = cons.Fetch(5)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for m := range batch.Messages() {
|
||||
_ = m.DoubleAck(ctx)
|
||||
}
|
||||
|
||||
restore := &keeper.Request{Verb: keeper.VerbRestoreKept, Module: "idx", Local: "modules", Form: 0, Fingerprint: "abc123", From: take.Dir, Why: "revert"}
|
||||
if _, err := k.RestoreKept(ctx, restore); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
kv, err = js.KeyValue(ctx, "idx_modules")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if e, err := kv.Get(ctx, "ace"); err != nil || string(e.Value()) != "fingerprint-1" {
|
||||
t.Errorf("the restored bucket holds %v (%v), want fingerprint-1", e, err)
|
||||
}
|
||||
if restore.CatchUp != "indexer_take_in" {
|
||||
t.Errorf("catch-up %q", restore.CatchUp)
|
||||
}
|
||||
forge, _ = js.Stream(ctx, "forge")
|
||||
ci, err := forge.Consumer(ctx, "ace_idx")
|
||||
if err != nil || ci.CachedInfo().Config.OptStartSeq != 5 {
|
||||
t.Errorf("the consumer is %+v (%v): it starts after the 4 messages it had acknowledged", ci.CachedInfo().Config, err)
|
||||
}
|
||||
kept, err := os.ReadDir(filepath.Join(dir, "aside"))
|
||||
if err != nil || len(kept) != 1 {
|
||||
t.Fatalf("aside holds %v (%v)", kept, err)
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(dir, "aside", kept[0].Name(), keeper.ArchiveFile)); err != nil {
|
||||
t.Error("the live data set aside is not kept whole")
|
||||
}
|
||||
if len(acts) != 2 {
|
||||
t.Errorf("hand acts %v: the restore point and the restore", acts)
|
||||
}
|
||||
}
|
||||
@@ -1,112 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
|
||||
"nats-tools/internal/keeper"
|
||||
)
|
||||
|
||||
// keptWait is how long a call to take or restore a kept data waits for the keeper's end on the watch of its requests: under
|
||||
// the runtime's own bound on a call, so a long snapshot is answered as "handed, not ended" and read again with its request.
|
||||
const keptWait = 50 * time.Second
|
||||
|
||||
func formOf(a map[string]any) int {
|
||||
f, _ := a["form"].(float64)
|
||||
return int(f)
|
||||
}
|
||||
|
||||
// keptTools are the seat mesh-broker's verbs for a module's kept data (design 47 G §405):
|
||||
// each verb hands its request to the keeper, which acts on its own account, and waits for the end on the watch of its
|
||||
// requests (never a poll). Asked again with `request`, a verb reads the end of the request it made.
|
||||
func keptTools(rs requestStore) []stdio.Tool {
|
||||
entry := func(extra map[string]any) map[string]any {
|
||||
in := map[string]any{"module": str("the module whose kept data it is, e.g. mesh-delivery-indexer"), "local-name": str("the kept data's local name, e.g. modules"),
|
||||
"form": num("the kept data's form (a manifest field; absent reads 0)"), "fingerprint": str("the fingerprint the restore point was taken before"),
|
||||
"why": str("why, in plain words: kept with the act"), "request": str("a request this verb made before: its end is read, nothing new is asked")}
|
||||
for k, v := range extra {
|
||||
in[k] = v
|
||||
}
|
||||
return in
|
||||
}
|
||||
run := func(verb string, build func(a map[string]any, r *keeper.Request) error) func(a map[string]any) (any, error) {
|
||||
return func(a map[string]any) (any, error) {
|
||||
name := "mesh-broker." + verb
|
||||
if id := arg(a, "request"); id != "" {
|
||||
r, err := rs.Get(id)
|
||||
if err != nil || r == nil {
|
||||
return nil, fmt.Errorf("no request %s is held (%v)", id, err)
|
||||
}
|
||||
if r.Done.IsZero() && rs.Ended != nil {
|
||||
if got := rs.Ended(id, keptWait); got != nil {
|
||||
return got, nil
|
||||
}
|
||||
if again, err := rs.Get(id); err == nil && again != nil {
|
||||
r = again
|
||||
}
|
||||
}
|
||||
return r, nil
|
||||
}
|
||||
r := keeper.Request{Verb: verb, Why: strings.TrimSpace(arg(a, "why")), Caller: stdio.CallerOf(a), At: rs.Now()}
|
||||
if err := build(a, &r); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if r.Why == "" {
|
||||
return nil, fmt.Errorf("%s says why", name)
|
||||
}
|
||||
r.ID = requestID(r.At)
|
||||
if err := rs.Put(r.ID, r); err != nil {
|
||||
return nil, fmt.Errorf("the request cannot be handed to the keeper: %w", err)
|
||||
}
|
||||
if rs.Ended != nil {
|
||||
if got := rs.Ended(r.ID, keptWait); got != nil {
|
||||
return got, nil
|
||||
}
|
||||
}
|
||||
return map[string]any{"request": r.ID, "state": "handed to the keeper, not ended yet", "read": fmt.Sprintf("%s with request %s", name, r.ID)}, nil
|
||||
}
|
||||
}
|
||||
point := func(a map[string]any, r *keeper.Request) error {
|
||||
r.Module, r.Local, r.Form, r.Fingerprint = arg(a, "module"), arg(a, "local-name"), formOf(a), arg(a, "fingerprint")
|
||||
if r.Module == "" || r.Local == "" || r.Fingerprint == "" {
|
||||
return fmt.Errorf("a restore point is of one kept data: a module, its local name and the fingerprint it is taken before")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return []stdio.Tool{
|
||||
{Name: "mesh-broker.snapshot-kept",
|
||||
Description: "Take the restore point of one module's kept data on the bus (milestone G, design 47): the data whole as the server's own snapshot, the position of each durable consumer the module reads with, and the catch-up tool the newer manifest declares, kept by the bus module's keeper. Answers the directory the restore point is in. Handed to the keeper, recorded as a hand act.",
|
||||
Input: entry(map[string]any{"catch-up": str("the tool a restore is followed by, from the newer manifest, kept in the restore point")}),
|
||||
Run: run(keeper.VerbSnapshotKept, func(a map[string]any, r *keeper.Request) error {
|
||||
r.CatchUp = arg(a, "catch-up")
|
||||
return point(a, r)
|
||||
})},
|
||||
{Name: "mesh-broker.restore-kept",
|
||||
Description: "Restore one module's kept data on the bus from a restore point beside the live data, never over it (design 47 G §405): the live data is snapshotted into a set-aside directory under a name of its own and then removed from the bus, the restore point's streams are put in its place and the module's other consumers are set back to their positions. With `aside-only` it sets the live data aside and restores nothing; with `discard-aside`, naming data set aside, it removes that data and restores nothing. Answers what was set aside and the catch-up tool the restore point carries. Handed to the keeper, recorded as a hand act.",
|
||||
Input: entry(map[string]any{"from": str("the directory the restore point was restored to, an absolute path"),
|
||||
"aside-only": map[string]any{"type": "boolean", "description": "set the live data aside and restore nothing"},
|
||||
"discard-aside": str("the id of data set aside: remove it and restore nothing")}),
|
||||
Run: run(keeper.VerbRestoreKept, func(a map[string]any, r *keeper.Request) error {
|
||||
if id := arg(a, "discard-aside"); id != "" {
|
||||
r.Verb, r.From = keeper.VerbDiscardAside, id
|
||||
return nil
|
||||
}
|
||||
if b, _ := a["aside-only"].(bool); b {
|
||||
r.AsideOnly = true
|
||||
r.Module, r.Local = arg(a, "module"), arg(a, "local-name")
|
||||
if r.Module == "" || r.Local == "" {
|
||||
return fmt.Errorf("aside-only is of one kept data: a module and its local name")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
r.From = arg(a, "from")
|
||||
if r.From == "" {
|
||||
return fmt.Errorf("a restore names the directory the restore point is in")
|
||||
}
|
||||
return point(a, r)
|
||||
})},
|
||||
}
|
||||
}
|
||||
@@ -1,82 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
|
||||
"nats-tools/internal/keeper"
|
||||
)
|
||||
|
||||
// The seat mesh-broker's verbs for a module's kept data hand their request to the keeper as the proved caller, wait for
|
||||
// its end on the watch of the requests, and say what a restore set aside (design 47, "A module's kept data on the bus").
|
||||
func TestTheKeptDataVerbsAreHandedToTheKeeperAndReadBack(t *testing.T) {
|
||||
held := map[string]keeper.Request{}
|
||||
rs := requestStore{
|
||||
Put: func(key string, r keeper.Request) error { held[key] = r; return nil },
|
||||
Get: func(key string) (*keeper.Request, error) {
|
||||
r, ok := held[key]
|
||||
if !ok {
|
||||
return nil, nil
|
||||
}
|
||||
return &r, nil
|
||||
},
|
||||
Now: func() time.Time { return time.Date(2026, 11, 9, 20, 0, 0, 0, time.UTC) },
|
||||
// The keeper's end, as its watch hands it.
|
||||
Ended: func(id string, within time.Duration) *keeper.Request {
|
||||
r := held[id]
|
||||
r.Done, r.Outcome, r.Dir = time.Unix(1, 0), "done", "/keeper/points/idx/modules/f0-abc"
|
||||
held[id] = r
|
||||
return &r
|
||||
},
|
||||
}
|
||||
tools := map[string]stdio.Tool{}
|
||||
for _, tl := range keptTools(rs) {
|
||||
tools[tl.Name] = tl
|
||||
}
|
||||
for _, verb := range KeptVerbs {
|
||||
if _, ok := tools["mesh-broker."+verb]; !ok {
|
||||
t.Errorf("the verb mesh-broker.%s is not served", verb)
|
||||
}
|
||||
}
|
||||
out, err := tools["mesh-broker.snapshot-kept"].Run(map[string]any{"module": "idx", "local-name": "modules", "form": float64(1),
|
||||
"fingerprint": "abc", "catch-up": "indexer_take_in", "why": "before form 1", "caller": "person.mallory"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r := out.(*keeper.Request); r.Verb != keeper.VerbSnapshotKept || r.Form != 1 || r.Fingerprint != "abc" || r.CatchUp != "indexer_take_in" ||
|
||||
r.Caller != stdio.Unproven || r.Dir == "" {
|
||||
t.Errorf("answered %+v: the request for form 1 before abc, for the caller the runtime proved, with the directory the keeper answered", r)
|
||||
}
|
||||
for name, args := range map[string]map[string]any{
|
||||
"mesh-broker.snapshot-kept": {"module": "idx", "local-name": "modules", "fingerprint": "abc"},
|
||||
"mesh-broker.restore-kept": {"module": "idx", "local-name": "modules", "fingerprint": "abc", "why": "revert"},
|
||||
"mesh-broker.restore-kept#aside-only": {"aside-only": true, "why": "x"},
|
||||
} {
|
||||
name, _, _ = strings.Cut(name, "#")
|
||||
if _, err := tools[name].Run(args); err == nil {
|
||||
t.Errorf("%s with %v was handed to the keeper", name, args)
|
||||
}
|
||||
}
|
||||
if out, err := tools["mesh-broker.restore-kept"].Run(map[string]any{"module": "idx", "local-name": "modules", "fingerprint": "abc",
|
||||
"from": "/restored/f0-abc", "why": "revert"}); err != nil || out.(*keeper.Request).From != "/restored/f0-abc" {
|
||||
t.Errorf("restore answered %+v (%v)", out, err)
|
||||
}
|
||||
// `aside-only` sets the live data aside and restores nothing; `discard-aside` removes the data set aside (§405).
|
||||
if out, err := tools["mesh-broker.restore-kept"].Run(map[string]any{"module": "idx", "local-name": "modules", "aside-only": true, "why": "revert"}); err != nil ||
|
||||
!out.(*keeper.Request).AsideOnly || out.(*keeper.Request).Verb != keeper.VerbRestoreKept {
|
||||
t.Errorf("aside-only answered %+v (%v)", out, err)
|
||||
}
|
||||
if out, err := tools["mesh-broker.restore-kept"].Run(map[string]any{"discard-aside": "idx.modules.1", "why": "operator's call"}); err != nil ||
|
||||
out.(*keeper.Request).Verb != keeper.VerbDiscardAside || out.(*keeper.Request).From != "idx.modules.1" {
|
||||
t.Errorf("discard-aside answered %+v (%v)", out, err)
|
||||
}
|
||||
}
|
||||
|
||||
func toJSON(v any) string {
|
||||
b, _ := json.Marshal(v)
|
||||
return string(b)
|
||||
}
|
||||
@@ -168,9 +168,6 @@ func main() {
|
||||
// them.
|
||||
var ReadingVerbs = []string{"list", "show", "check", "orphans"}
|
||||
|
||||
// KeptVerbs are the seat mesh-broker's verbs for a module's kept data (keptools.go): each handed to the keeper.
|
||||
var KeptVerbs = []string{"snapshot-kept", "restore-kept"}
|
||||
|
||||
// Events are the seat mesh-broker's events (ADR 0595 §6, ADR 0620 §2): said on the seat's subjects on the shared events stream
|
||||
// and kept in full on the bus module's own stream broker-events (BROKER_EVENTS), which `history` reads (keep.go).
|
||||
var Events = []string{"created", "refused", "drifted", "drift-cleared", "full", "full-cleared", "orphaned", "recreated", "dropped"}
|
||||
@@ -193,7 +190,5 @@ func allTools() []stdio.Tool {
|
||||
out = append(out, brokerTools(HeldInState, jsz, StreamHistory(LiveBus()))...)
|
||||
out = append(out, keeperHealthTool(StatusInState, HeldInState, jsz))
|
||||
out = append(out, readTools(jsz, LiveMsgGet(LiveBus()))...)
|
||||
requests := RequestsInState()
|
||||
out = append(out, requestTools(requests)...)
|
||||
return append(out, keptTools(requests)...)
|
||||
return append(out, requestTools(RequestsInState())...)
|
||||
}
|
||||
|
||||
@@ -189,8 +189,8 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
|
||||
if err := json.Unmarshal(raw, &nm); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(nm.Claims) != 1 || nm.Claims[0].Name != "mesh-broker" || !slices.Equal(nm.Claims[0].Serves, append(slices.Clone(ReadingVerbs), KeptVerbs...)) {
|
||||
t.Errorf("claims are %+v: mesh-broker, serving its reading verbs %v and the keeper-handed verbs %v alone", nm.Claims, ReadingVerbs, KeptVerbs)
|
||||
if len(nm.Claims) != 1 || nm.Claims[0].Name != "mesh-broker" || !slices.Equal(nm.Claims[0].Serves, ReadingVerbs) {
|
||||
t.Errorf("claims are %+v: mesh-broker, serving its reading verbs %v alone", nm.Claims, ReadingVerbs)
|
||||
}
|
||||
if !slices.Equal(nm.Consumes, []string{"mesh-catalog.manifest-accepted"}) {
|
||||
t.Errorf("consumes %v: the seat mesh-catalog's manifest-accepted alone (ADR 0595 §4)", nm.Consumes)
|
||||
@@ -224,8 +224,8 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
|
||||
}
|
||||
// spent, the warrants the keeper spent, is written by the keeper alone and only read by the tools (review round 2,
|
||||
// security F2: the tool runner's account, which writes requests, may never delete a spend record).
|
||||
if !slices.Equal(states, []string{"asked", "requests", "spent", "kept"}) || !slices.Equal(readOnly, []string{"spent"}) {
|
||||
t.Errorf("state is %v, read-only %v: asked and requests, old sections (ADR 0595 §10), spent, read-only, and kept, the restore points taken", states, readOnly)
|
||||
if !slices.Equal(states, []string{"asked", "requests", "spent"}) || !slices.Equal(readOnly, []string{"spent"}) {
|
||||
t.Errorf("state is %v, read-only %v: asked and requests, old sections (ADR 0595 §10), and spent, read-only", states, readOnly)
|
||||
}
|
||||
if _, ok := nm.OwnSecrets["broker-keeper"]; !ok {
|
||||
t.Error("own-secrets lack broker-keeper, the keeper's own account (ADR 0436 §1)")
|
||||
@@ -234,9 +234,9 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
|
||||
bus, _ := keeper["bus"].(map[string]any)
|
||||
health, _ := keeper["health"].(map[string]any)
|
||||
if keeper["type"] != "process" || keeper["artifact"] != "tools" || fmt.Sprint(keeper["run"]) != "[./nats-tools keep]" ||
|
||||
fmt.Sprint(bus["state"]) != "[asked requests spent kept]" ||
|
||||
fmt.Sprint(bus["state"]) != "[asked requests spent]" ||
|
||||
health["kind"] != "tool" || health["tool"] != "nats_keeper_health" {
|
||||
t.Errorf("the keeper is %v: a process of the tools bundle, run as keep, keeping asked, requests, spent and kept, judged by nats_keeper_health", keeper)
|
||||
t.Errorf("the keeper is %v: a process of the tools bundle, run as keep, keeping asked, requests and spent, judged by nats_keeper_health", keeper)
|
||||
}
|
||||
// Its bus declares keeps, the one key more ADR 0626 §1 allows it, read through the SDK as the grant composition reads it.
|
||||
var whole catalogue.Manifest
|
||||
@@ -263,7 +263,7 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
|
||||
t.Errorf("the manifest lists the tool %s, which the bundle does not serve", name)
|
||||
}
|
||||
}
|
||||
for _, verb := range append(slices.Clone(ReadingVerbs), KeptVerbs...) {
|
||||
for _, verb := range ReadingVerbs {
|
||||
if !served["mesh-broker."+verb] {
|
||||
t.Errorf("the bundle does not serve the seat verb mesh-broker.%s", verb)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user