novox/mesh-catalog
9
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity
Files
bf60d6a8bb1671fa7d9c5b0fbe202e8e2b20d61a
mesh-catalog/modules/nats/internal/keeper/keeper_test.go
T
jschoubben bf60d6a8bb NATS3 (rest): the bus module plans from the catalog half of every watched branch, not from manifest-accepted 
The keeper follows meshcatalogclient's branch follower and the seat list on its views reader, and plans again at each
move of a branch's view, the registrations or the seat list. Each local name is planned from the first of the module's
watched branches in its registration's order; another branch asking it with other settings is said drifted, naming
the branch. A module of the new form whose resolved seat definition has a verb or an event gets its sent store; one
including only parts, and a foundation part, none. A module in no watched branch's newest revision loses its plans
key, its entries orphans. The sum of the caps counts the old sections the entries declare. The manifest consumes the
catalog half's and the seat list's events and invokes their verbs; manifest-accepted and indexer_bus_state go.

Builds: design 47 G §344, §433, §532, §533, §537, §538, §547, §591

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
2026-11-12 05:45:49 +01:00

1273 lines
67 KiB
Go
Raw Blame History

package keeper
import (
"context"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"os"
"sort"
"strings"
"sync"
"testing"
"time"
"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"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/seatdef"
)
// fakeBus holds objects by stream, as a bus would, and refuses what a test tells it to.
type fakeBus struct {
mu sync.Mutex
held map[string]Settings
fill map[string]uint64
refuse map[string]error
created []string
updated []string
deleted []string
snaps [][]string
// restricted names the streams whose metadata says they are restricted.
restricted map[string]bool
// snapStreams are the streams each snapshot file holds; restored the files restored; gate holds a restore until it is
// closed or its context ends; deletedStreams the streams removed by name.
snapStreams map[string][]string
restored []string
gate chan struct{}
deletedStreams []string
// limits are the bus's storage limits the fake answers.
limits Limits
// reads counts the reads of an object's stream.
reads int
}
func (b *fakeBus) SnapshotStreams(path string) ([]string, error) {
return b.snapStreams[path], nil
}
func (b *fakeBus) Live(_ context.Context, name string) (bool, error) {
b.mu.Lock()
defer b.mu.Unlock()
_, ok := b.held[name]
return ok, nil
}
func (b *fakeBus) Restore(ctx context.Context, path string) error {
b.mu.Lock()
gate, err := b.gate, b.refuse["restore"]
b.restored = append(b.restored, path)
b.mu.Unlock()
if gate != nil {
select {
case <-gate:
case <-ctx.Done():
return ctx.Err()
}
}
return err
}
func (b *fakeBus) DeleteStream(_ context.Context, name string) error {
b.mu.Lock()
defer b.mu.Unlock()
b.deletedStreams = append(b.deletedStreams, name)
return nil
}
func (b *fakeBus) Limits(context.Context) (Limits, error) {
b.mu.Lock()
defer b.mu.Unlock()
return b.limits, nil
}
func newFakeBus() *fakeBus {
return &fakeBus{held: map[string]Settings{}, fill: map[string]uint64{}, refuse: map[string]error{}}
}
func (b *fakeBus) Held(_ context.Context, o Object) (Settings, Fill, bool, error) {
b.mu.Lock()
defer b.mu.Unlock()
b.reads++
s, ok := b.held[o.Stream()]
return s, Fill{Bytes: b.fill[o.Stream()]}, ok, nil
}
func (b *fakeBus) Create(_ context.Context, o Object) error {
b.mu.Lock()
defer b.mu.Unlock()
if err := b.refuse[o.Stream()]; err != nil {
return err
}
b.held[o.Stream()] = o.Settings
b.created = append(b.created, o.Stream())
return nil
}
func (b *fakeBus) Update(_ context.Context, o Object) error {
b.mu.Lock()
defer b.mu.Unlock()
b.held[o.Stream()] = o.Settings
b.updated = append(b.updated, o.Stream())
return nil
}
func (b *fakeBus) Delete(_ context.Context, o Object) error {
b.mu.Lock()
defer b.mu.Unlock()
delete(b.held, o.Stream())
b.deleted = append(b.deleted, o.Stream())
return nil
}
func (b *fakeBus) Snapshot(_ context.Context, streams []string) (string, error) {
b.mu.Lock()
defer b.mu.Unlock()
b.snaps = append(b.snaps, streams)
if err := b.refuse["snapshot"]; err != nil {
return "", err
}
return "/snapshots/" + strings.Join(streams, "+") + ".tar", nil
}
func (b *fakeBus) Restricted(_ context.Context, streams []string) (bool, error) {
b.mu.Lock()
defer b.mu.Unlock()
for _, s := range streams {
if b.restricted[s] {
return true, nil
}
}
return false, nil
}
// memFiles are snapshot files in memory: each of size, 100 bytes unless a test says so, with a digest of its path.
type memFiles struct {
size map[string]int64
removed []string
missing map[string]bool
}
func (f *memFiles) Digest(path string) (int64, string, error) {
if f.missing[path] {
return 0, "", os.ErrNotExist
}
n, ok := f.size[path]
if !ok {
n = 100
}
return n, "sha256:" + hex.EncodeToString([]byte(path)), nil
}
func (f *memFiles) Remove(path string) error {
f.removed = append(f.removed, path)
return nil
}
// memStore is a key-value state in memory.
type memStore struct {
// refuse refuses a Put of `<state>/<key>`.
refuse map[string]error
mu sync.Mutex
kv map[string][]byte
revs map[string]uint64
rev uint64
}
func newMemStore() *memStore {
return &memStore{kv: map[string][]byte{}, revs: map[string]uint64{}, refuse: map[string]error{}}
}
func (s *memStore) Get(state, key string) ([]byte, uint64, bool, error) {
s.mu.Lock()
defer s.mu.Unlock()
v, ok := s.kv[state+"/"+key]
return v, s.revs[state+"/"+key], ok, nil
}
func (s *memStore) Put(state, key string, value []byte) (uint64, error) {
s.mu.Lock()
defer s.mu.Unlock()
if err := s.refuse[state+"/"+key]; err != nil {
return 0, err
}
s.rev++
s.kv[state+"/"+key], s.revs[state+"/"+key] = value, s.rev
return s.rev, nil
}
func (s *memStore) Create(state, key string, value []byte) (uint64, error) {
s.mu.Lock()
defer s.mu.Unlock()
if _, ok := s.kv[state+"/"+key]; ok {
return 0, ErrExists
}
s.rev++
s.kv[state+"/"+key], s.revs[state+"/"+key] = value, s.rev
return s.rev, nil
}
func (s *memStore) Delete(state, key string) error {
s.mu.Lock()
defer s.mu.Unlock()
delete(s.kv, state+"/"+key)
return nil
}
func (s *memStore) Keys(state string) ([]string, error) {
s.mu.Lock()
defer s.mu.Unlock()
var out []string
for k := range s.kv {
if st, key, _ := strings.Cut(k, "/"); st == state {
out = append(out, key)
}
}
sort.Strings(out)
return out, nil
}
// said is what a keeper said: its events and its conditions, in order.
type said struct {
mu sync.Mutex
events []string
bodies []Event
ids []string
conds []string
// down refuses every event and condition while set, as a bus that does not take them; partly takes an event on the
// seat's subject and refuses its history.
down, partly bool
// left are the subjects each try was asked to say an event on (nil: every one).
left [][]string
// restored are the snapshot-restored events said, and sig is told of each.
restored []string
sig chan struct{}
}
func (s *said) emit(e *Saying) error {
s.mu.Lock()
defer s.mu.Unlock()
if e.Restored != nil {
s.restored = append(s.restored, fmt.Sprintf("%s %s %s %s", e.Restored.Request, e.Restored.Act, e.Restored.Outcome, e.Restored.Error))
if s.sig != nil {
s.sig <- struct{}{}
}
return nil
}
s.left = append(s.left, e.Left)
if s.down {
e.Left, e.Said = []string{"seat", "history"}, nil
return errors.New("no responders")
}
if s.partly && e.Left == nil {
e.Left, e.Said = []string{"history"}, []string{"seat"}
s.ids = append(s.ids, e.ID)
return errors.New("the history stream is full")
}
s.events = append(s.events, e.Event+" "+e.Body.Module+" "+e.Body.Local)
s.bodies = append(s.bodies, e.Body)
s.ids = append(s.ids, e.ID)
e.Left = nil
return nil
}
func (s *said) say(c oc.Condition) error {
s.mu.Lock()
defer s.mu.Unlock()
if s.down {
return errors.New("no responders")
}
s.conds = append(s.conds, c.State+" "+c.Key)
return nil
}
func (s *said) reset() {
s.mu.Lock()
defer s.mu.Unlock()
s.events, s.bodies, s.conds = nil, nil, nil
}
// findingLog is the finding writer's calls, as the held writer would be handed them.
type findingLog struct{ calls []string }
func (f *findingLog) Write(_ context.Context, fi meshconditions.Finding) error {
f.calls = append(f.calls, "write "+fi.Scope+"."+fi.ID+"."+fi.Kind+" "+fi.Observer)
return nil
}
func (f *findingLog) Clear(_ context.Context, scope, id, kind, _ string) error {
f.calls = append(f.calls, "clear "+scope+"."+id+"."+kind)
return nil
}
func newKeeper(view CatalogView) (*Keeper, *fakeBus, *memStore, *said) {
bus, store, out := newFakeBus(), newMemStore(), &said{}
at := time.Date(2026, 11, 4, 12, 0, 0, 0, time.UTC)
k := &Keeper{Bus: bus, Store: store, Finding: &findingLog{}, Emit: out.emit, Say: out.say, Files: &memFiles{size: map[string]int64{}},
Catalog: func() (CatalogView, error) { return view, nil },
Now: func() time.Time { return at }}
return k, bus, store, out
}
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
}
// What one entry is on the bus (design 25 § 12.1, CAT-N3): each template's objects, named `<module>_<local>`, its
// binding, and the fixed settings no option changes; and the entries the keeper refuses itself, naming why.
func TestEachTemplateIsPlannedUnderTheModulesNames(t *testing.T) {
ledger := entry("records", catalogue.StorageLedger, catalogue.TemplateLedger, map[string]string{"max-mib": "16"})
ledger.Types = []catalogue.LedgerType{{Name: "decision", Indexes: []catalogue.LedgerIndex{{Name: "by-topic", On: []string{"topic"}}}}}
bare := entry("raw", catalogue.StorageStream, "", nil)
bare.Configuration = map[string]json.RawMessage{"when-full": json.RawMessage(`"drop-oldest"`), "max-mib": json.RawMessage("8"), "max-age-days": json.RawMessage("3")}
forGood := entry("reviews", catalogue.StorageStream, "", nil)
forGood.Configuration = map[string]json.RawMessage{"when-full": json.RawMessage(`"refuse"`), "max-mib": json.RawMessage("512"), "max-age-days": json.RawMessage(`"none"`)}
for _, row := range []struct {
name string
entry catalogue.BusStateEntry
objects string
binding string
refused string
}{
{name: "state", entry: entry("drafts", catalogue.StorageKeyValue, catalogue.TemplateState, map[string]string{"history": "5"}),
objects: "key-value mesh-knowledge_drafts {64 MiB, age 0, history 5, new, size 262144}; object-store mesh-knowledge_drafts-objects {16 MiB, age 0, history 0, new, size 0}",
binding: "bucket mesh-knowledge_drafts"},
{name: "state naming objects-max-mib (§591)", entry: entry("drafts", catalogue.StorageKeyValue, catalogue.TemplateState, map[string]string{"objects-max-mib": "512"}),
objects: "key-value mesh-knowledge_drafts {64 MiB, age 0, history 1, new, size 262144}; object-store mesh-knowledge_drafts-objects {512 MiB, age 0, history 0, new, size 0}",
binding: "bucket mesh-knowledge_drafts"},
{name: "state naming objects-max-mib 0", entry: entry("drafts", catalogue.StorageKeyValue, catalogue.TemplateState, map[string]string{"objects-max-mib": "0"}),
refused: "not a size above 0"},
{name: "events, default age and cap", entry: entry("events", catalogue.StorageStream, catalogue.TemplateEvents, nil),
objects: "stream mesh-knowledge_events [mesh.bus.mesh-knowledge.events.>] {1024 MiB, age 604800, history 0, old, size 0}",
binding: "stream mesh-knowledge_events on mesh.bus.mesh-knowledge.events"},
{name: "work queue", entry: entry("jobs", catalogue.StorageStream, catalogue.TemplateWorkQueue, map[string]string{"max-mib": "4", "duplicates": "120"}),
objects: "stream mesh-knowledge_jobs [mesh.bus.mesh-knowledge.jobs.>] {4 MiB, age 0, history 0, new, size 0, work-queue, duplicates 120}",
binding: "stream mesh-knowledge_jobs on mesh.bus.mesh-knowledge.jobs"},
{name: "log", entry: entry("log", catalogue.StorageStream, catalogue.TemplateLog, nil),
objects: "stream mesh-knowledge_log [mesh.bus.mesh-knowledge.log.>] {1024 MiB, age 0, history 0, new, size 262144}",
binding: "stream mesh-knowledge_log on mesh.bus.mesh-knowledge.log"},
{name: "objects", entry: entry("files", catalogue.StorageObjectStore, catalogue.TemplateObjects, map[string]string{"max-mib": "32"}),
objects: "object-store mesh-knowledge_files {32 MiB, age 0, history 0, new, size 0}",
binding: "objects mesh-knowledge_files"},
{name: "ledger", entry: ledger,
objects: "stream mesh-knowledge_records [mesh.ledger.mesh-knowledge.records.history.>] {16 MiB, age 0, history 0, new, size 262144}; " +
"key-value mesh-knowledge_records_state {64 MiB, age 0, history 1, new, size 262144}; " +
"key-value mesh-knowledge_records_index-by-topic {64 MiB, age 0, history 1, new, size 262144}; " +
"stream mesh-knowledge_records_snapshot [mesh.ledger.mesh-knowledge.records.snapshot.>] {16 MiB, age 0, history 1, old, size 0}",
binding: "history mesh-knowledge_records on mesh.ledger.mesh-knowledge.records.history, state mesh-knowledge_records_state, indexes map[by-topic:mesh-knowledge_records_index-by-topic], snapshot mesh.ledger.mesh-knowledge.records.snapshot"},
{name: "bare stream", entry: bare,
objects: "stream mesh-knowledge_raw [mesh.bus.mesh-knowledge.raw.>] {8 MiB, age 259200, history 0, old, size 0}",
binding: "stream mesh-knowledge_raw on mesh.bus.mesh-knowledge.raw"},
{name: "a bare stream kept for good (max-age-days none, as the judge accepts it)", entry: forGood,
objects: "stream mesh-knowledge_reviews [mesh.bus.mesh-knowledge.reviews.>] {512 MiB, age 0, history 0, new, size 0}",
binding: "stream mesh-knowledge_reviews on mesh.bus.mesh-knowledge.reviews"},
{name: "a local name holding the separator", entry: entry("my_drafts", catalogue.StorageKeyValue, catalogue.TemplateState, nil),
refused: `the local name "my_drafts" is empty or holds "_"`},
{name: "a local name that is the sent store's", entry: entry("sent-objects", catalogue.StorageObjectStore, catalogue.TemplateObjects, map[string]string{"max-mib": "1"}),
refused: "the bus module's own for the sent store mesh-knowledge_sent-objects"},
{name: "objects with no cap", entry: entry("files", catalogue.StorageObjectStore, catalogue.TemplateObjects, nil),
refused: "gives no max-mib"},
{name: "a bare stream with no when-full", entry: entry("raw", catalogue.StorageStream, "", nil),
refused: "must give when-full"},
{name: "a template on the wrong storage type", entry: entry("x", catalogue.StorageKeyValue, catalogue.TemplateLog, nil),
refused: "is none the bus module makes"},
} {
t.Run(row.name, func(t *testing.T) {
p, err := PlanOf("mesh-knowledge", row.entry)
if row.refused != "" {
if err == nil || !strings.Contains(err.Error(), row.refused) {
t.Fatalf("planned %+v (%v): refused, naming %q", p, err, row.refused)
}
return
}
if err != nil {
t.Fatal(err)
}
var objs []string
for _, o := range p.Objects {
s := o.Settings
w := fmt.Sprintf("%s %s", o.Kind, o.Name)
if len(o.Subjects) > 0 {
w += fmt.Sprintf(" %v", o.Subjects)
}
w += fmt.Sprintf(" {%d MiB, age %d, history %d, %s, size %d", s.MaxBytes/MiB, s.MaxAgeSeconds, s.History, s.Discard, s.MaxMsgSize)
if s.WorkQueue {
w += fmt.Sprintf(", work-queue, duplicates %d", s.DuplicatesSeconds)
}
objs = append(objs, w+"}")
}
if got := strings.Join(objs, "; "); got != row.objects {
t.Errorf("objects\n got %s\nwant %s", got, row.objects)
}
b := p.Binding
var binding string
switch {
case b.History != "":
binding = fmt.Sprintf("history %s on %s, state %s, indexes %v, snapshot %s", b.History, b.Subject, b.State, b.Indexes, b.Snapshot)
case b.Stream != "":
binding = fmt.Sprintf("stream %s on %s", b.Stream, b.Subject)
case b.Bucket != "":
binding = "bucket " + b.Bucket
case b.ObjectStore != "":
binding = "objects " + b.ObjectStore
}
if binding != row.binding || b.Local != row.entry.Name || b.Storage != row.entry.Storage {
t.Errorf("binding %s (%+v), want %s", binding, b, row.binding)
}
})
}
}
// The keeper makes what an accepted verdict asks for, once, and says it (design 25 § 12.2, ADR 0595 §4; CAT-N2, N3):
// nothing before its first view; a replay not newer than what it holds is not applied; a refusal of its grant waits and
// is made on the backoff, never said refused; a refusal of the entry is said and raised, and cleared by a later make.
func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) {
drafts := entry("drafts", catalogue.StorageKeyValue, catalogue.TemplateState, nil)
log := entry("log", catalogue.StorageStream, catalogue.TemplateLog, nil)
for _, row := range []struct {
name string
run func(t *testing.T, k *Keeper, bus *fakeBus, out *said)
}{
{"created once, binding said, nothing before the first view", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
k.mu.Lock()
k.viewed = false
k.mu.Unlock()
k.Heard(Accepted{Module: "m", Digest: "d1", Revision: 3, BusState: []catalogue.BusStateEntry{drafts}})
if len(bus.created) != 0 {
t.Fatalf("created %v before the first view", bus.created)
}
if err := k.PlanCatalog(); err != nil {
t.Fatal(err)
}
k.Heard(Accepted{Module: "m", Digest: "d1", Revision: 3, BusState: []catalogue.BusStateEntry{drafts, log}})
k.Heard(Accepted{Module: "m", Digest: "d1", Revision: 3, BusState: []catalogue.BusStateEntry{drafts, log}})
if want := "[KV_m_drafts OBJ_m_drafts-objects m_log]"; fmt.Sprint(bus.created) != want {
t.Errorf("created %v, want %s", bus.created, want)
}
if want := "[created m drafts created m log]"; fmt.Sprint(out.events) != want {
t.Errorf("said %v, want %s", out.events, want)
}
if b := out.bodies[1]; b.Binding.Stream != "m_log" || b.Binding.Subject != "mesh.bus.m.log" || b.Revision != 3 || b.Asked["m_log"].MaxBytes == 0 {
t.Errorf("created carries %+v: the binding, the revision and the settings asked", b)
}
// In the field names of the seat mesh-broker's message said (mesh-contracts), beside binding and revision.
raw, _ := json.Marshal(out.bodies[1])
for _, w := range []string{`"module":"m"`, `"local-name":"log"`, `"storage-type":"stream"`, `"template":"log"`, `"bus-name":"m_log"`, `"binding":{`, `"revision":3`} {
if !strings.Contains(string(raw), w) {
t.Errorf("the body %s lacks %s", raw, w)
}
}
}},
{"a sender gets its sent store: 256 MiB, 7 days, refusing when full, and it is kept when a verdict asks nothing of it", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
k.Sends("m")
k.Heard(Accepted{Module: "m", Digest: "d1", Revision: 3, BusState: []catalogue.BusStateEntry{drafts}})
k.Sends("m")
if want := "[OBJ_m_sent-objects KV_m_drafts OBJ_m_drafts-objects]"; fmt.Sprint(bus.created) != want {
t.Fatalf("created %v, want %s", bus.created, want)
}
if s := bus.held["OBJ_m_sent-objects"]; s.MaxBytes != 256*MiB || s.MaxAgeSeconds != 7*24*3600 || s.Discard != DiscardNew {
t.Errorf("the sent store is %+v, want 256 MiB, 7 days, refusing when full", s)
}
if want := "[created m sent-objects created m drafts]"; fmt.Sprint(out.events) != want {
t.Errorf("said %v, want %s", out.events, want)
}
if b := out.bodies[0].Binding; b.ObjectStore != "m_sent-objects" {
t.Errorf("binding %+v names no sent store", b)
}
bus.fill["OBJ_m_sent-objects"] = 200 * uint64(MiB)
k.FillPass()
if got := out.events[len(out.events)-1]; got != "full m sent-objects" {
t.Errorf("a full sent store said %q, want full", got)
}
if held := k.Held()["m"]; held.Made[SentLocal].Orphaned || len(bus.deleted) != 0 {
t.Errorf("the sent store was orphaned or removed: %+v %v", held.Made[SentLocal], bus.deleted)
}
}},
// §344, §533, §537: a module of the new form claiming a seat whose resolved definition has a verb or an event, or
// declaring an event of its own, gets a sent store; a seat including only bodies, and a foundation part, none; a
// definition gaining a verb gives its holder one at the next plan.
{"the catalog half and the seat list plan the sent stores", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
defs := map[string]seatdef.Definition{
"verbed": {Seat: "verbed", Verbs: []seatdef.Verb{{Name: "go"}}},
"resolved": {Seat: "resolved", Includes: []string{"part"}, Verbs: []seatdef.Verb{{Name: "from-the-part"}}},
"bodies": {Seat: "bodies", Includes: []string{"shared-bodies"}},
}
k.Seats = func() map[string]seatdef.Definition { return defs }
k.Catalog = func() (CatalogView, error) {
return CatalogView{Modules: []CatalogModule{
{Accepted: Accepted{Module: "a"}, Claims: []string{"verbed"}},
{Accepted: Accepted{Module: "b"}, Claims: []string{"resolved"}},
{Accepted: Accepted{Module: "c"}, Claims: []string{"bodies"}},
{Accepted: Accepted{Module: "d"}, Claims: []string{"verbed"}, Foundation: true},
{Accepted: Accepted{Module: "e"}, EmitsOwn: true},
}}, nil
}
_ = k.PlanCatalog()
if want := "[OBJ_a_sent-objects OBJ_b_sent-objects OBJ_e_sent-objects]"; fmt.Sprint(bus.created) != want {
t.Fatalf("created %v, want %s", bus.created, want)
}
defs["bodies"] = seatdef.Definition{Seat: "bodies", Includes: []string{"shared-bodies"}, Verbs: []seatdef.Verb{{Name: "new"}}}
_ = k.PlanCatalog()
if want := "[OBJ_a_sent-objects OBJ_b_sent-objects OBJ_e_sent-objects OBJ_c_sent-objects]"; fmt.Sprint(bus.created) != want {
t.Errorf("created %v, want %s: the holder of a definition gaining a verb gets its store", bus.created, want)
}
}},
// §532: a later branch asking a local name with other settings is said drifted once, naming the branch; §547: the
// old sections' caps count in the sum.
{"a drift between branches is said once, naming the branch", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
view := CatalogView{Modules: []CatalogModule{{Accepted: Accepted{Module: "m", Revision: 4, BusState: []catalogue.BusStateEntry{drafts}}, OldBytes: 8 * MiB}},
Drifts: []Drift{{Module: "m", Local: "drafts", Branch: "feature", From: "main"}}}
k.Catalog = func() (CatalogView, error) { return view, nil }
_ = k.PlanCatalog()
_ = k.PlanCatalog()
if fmt.Sprint(out.events) != "[created m drafts drifted m drafts]" || !strings.Contains(out.bodies[1].Why, "the branch feature asks drafts") {
t.Errorf("said %v (%+v): created from main, drifted once naming feature", out.events, out.bodies)
}
if k.Status().OldBytes != 8*MiB {
t.Errorf("old sections %d: 8 MiB counted", k.Status().OldBytes)
}
}},
// §538: a module in no watched branch's newest revision has its plans key deleted, its entries then orphans.
{"a module no watched branch lists loses its plan, its entries orphans", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
k.Catalog = func() (CatalogView, error) {
return CatalogView{Modules: []CatalogModule{{Accepted: Accepted{Module: "m", Revision: 9, BusState: []catalogue.BusStateEntry{drafts}}}}}, nil
}
_ = k.PlanCatalog()
if _, ok := k.planned("m"); !ok {
t.Fatal("m's plan is not kept")
}
k.Catalog = func() (CatalogView, error) { return CatalogView{}, nil }
_ = k.PlanCatalog()
if _, ok := k.planned("m"); ok {
t.Error("m's plan is kept while no watched branch lists it")
}
if a := k.Held()["m"]; !a.Made["drafts"].Orphaned || len(bus.deleted) != 0 {
t.Errorf("held %+v, deleted %v: drafts an orphan, kept on the bus", a, bus.deleted)
}
}},
{"a grant not composed yet waits, then is made", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
bus.refuse["m_log"] = fmt.Errorf("%w: permissions violation", ErrRetry)
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{log}})
if w := k.Waiting(); len(w) != 1 || len(out.events) != 0 {
t.Fatalf("waiting %v, said %v: waits, says nothing", w, out.events)
}
delete(bus.refuse, "m_log")
k.CheckAll()
if len(k.Waiting()) != 0 || fmt.Sprint(out.events) != "[created m log]" {
t.Errorf("waiting %v, said %v: made at the next check", k.Waiting(), out.events)
}
}},
{"a refused create is said and raised, cleared by a later create", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
bus.refuse["KV_m_drafts"] = errors.New("insufficient resources")
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts}})
if fmt.Sprint(out.events) != "[refused m drafts]" || fmt.Sprint(out.conds) != "[raised nats.refused.m_drafts]" ||
!strings.Contains(out.bodies[0].Why, "insufficient resources") {
t.Fatalf("said %v %v %+v", out.events, out.conds, out.bodies)
}
delete(bus.refuse, "KV_m_drafts")
out.reset()
k.CheckAll()
if fmt.Sprint(out.events) != "[created m drafts]" || fmt.Sprint(out.conds) != "[cleared nats.refused.m_drafts]" {
t.Errorf("said %v %v: created, the refusal cleared", out.events, out.conds)
}
}},
{"an entry the keeper cannot plan is refused, naming why", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{entry("a_b", catalogue.StorageKeyValue, catalogue.TemplateState, nil)}})
if len(bus.created) != 0 || fmt.Sprint(out.events) != "[refused m a_b]" {
t.Errorf("created %v, said %v", bus.created, out.events)
}
}},
// CAT-N4: check and drift (ADR 0595 §5, §6): a difference that only grows is applied, said by nothing; any other
// is said and raised, never fixed quietly, and cleared once the bus holds what is asked again.
{"a held entry that grows is applied without a person", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{log}})
out.reset()
bigger := entry("log", catalogue.StorageStream, catalogue.TemplateLog, map[string]string{"max-mib": "2048"})
k.Heard(Accepted{Module: "m", Revision: 2, BusState: []catalogue.BusStateEntry{bigger}})
if fmt.Sprint(bus.updated) != "[m_log]" || bus.held["m_log"].MaxBytes != 2048*MiB || len(out.events)+len(out.conds) != 0 {
t.Errorf("updated %v to %d, said %v %v: grown quietly", bus.updated, bus.held["m_log"].MaxBytes, out.events, out.conds)
}
}},
{"a held entry that would shrink is a drift, said, raised and cleared", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{log}})
out.reset()
smaller := entry("log", catalogue.StorageStream, catalogue.TemplateLog, map[string]string{"max-mib": "16"})
k.Heard(Accepted{Module: "m", Revision: 2, BusState: []catalogue.BusStateEntry{smaller}})
if len(bus.updated) != 0 || fmt.Sprint(out.events) != "[drifted m log]" || fmt.Sprint(out.conds) != "[raised nats.drifted.m_log]" ||
!strings.Contains(out.bodies[0].Why, "max-bytes asked 16777216, held 1073741824") {
t.Fatalf("updated %v, said %v %v %+v: the drift said, nothing changed", bus.updated, out.events, out.conds, out.bodies)
}
if l := k.CheckAll(); len(l) != 1 || l[0].Drift == "" || len(out.events) != 1 {
t.Errorf("check answers %+v and said %v: the drift once", l, out.events)
}
k.Heard(Accepted{Module: "m", Revision: 3, BusState: []catalogue.BusStateEntry{log}})
if fmt.Sprint(out.events) != "[drifted m log drift-cleared m log]" || out.conds[len(out.conds)-1] != "cleared nats.drifted.m_log" {
t.Errorf("said %v %v: cleared once asked as held", out.events, out.conds)
}
}},
{"a plan whose caps pass the account's storage limit is refused, naming the sum and the limit, and the caps made stay (§547)", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
// drafts: 64 MiB state + 16 MiB objects = 80 MiB; log: 1024 MiB.
bus.limits = Limits{AccountBytes: 100 * MiB}
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts}})
k.Heard(Accepted{Module: "m", Revision: 2, BusState: []catalogue.BusStateEntry{drafts, log}})
m := k.Held()["m"].Made
if fmt.Sprint(bus.created) != "[KV_m_drafts OBJ_m_drafts-objects]" || !m["drafts"].Created || m["log"].Created ||
!strings.Contains(m["log"].Refused, "1104 MiB, past the account's storage limit of 100 MiB") || fmt.Sprint(out.events) != "[created m drafts refused m log]" {
t.Errorf("created %v, said %v, log %+v: the caps made stay, the plan that passes the limit refused naming both", bus.created, out.events, m["log"])
}
if CapSum(k.Held(), "", "") != 80*MiB {
t.Errorf("the sum is %d, want 80 MiB: a refused entry holds none", CapSum(k.Held(), "", ""))
}
bus.limits = Limits{}
k.CheckAll()
if !k.Held()["m"].Made["log"].Created {
t.Errorf("with no account limit nothing is refused: %+v", k.Held()["m"].Made["log"])
}
}},
{"a verdict plans again only when its entries differ, never on the manifest's digest (§538)", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
k.Heard(Accepted{Module: "m", Digest: "d1", Revision: 1, BusState: []catalogue.BusStateEntry{drafts}})
reads := bus.reads
k.Heard(Accepted{Module: "m", Digest: "d2", Revision: 2, BusState: []catalogue.BusStateEntry{drafts}})
if bus.reads != reads {
t.Errorf("the bus was read %d times more for a manifest that changed outside its entries", bus.reads-reads)
}
raw, _, found, _ := k.Store.Get(StatePlans, "m")
var p Planned
_ = json.Unmarshal(raw, &p)
if !found || p.Digest != EntriesDigest([]catalogue.BusStateEntry{drafts}) || len(p.Entries) != 1 || k.Held()["m"].Digest != "d2" || k.Held()["m"].Revision != 2 {
t.Errorf("the plan kept %+v (%v), the digest and revision follow the verdict: %+v", p, found, k.Held()["m"])
}
bigger := entry("drafts", catalogue.StorageKeyValue, catalogue.TemplateState, map[string]string{"history": "7"})
k.Heard(Accepted{Module: "m", Digest: "d3", Revision: 3, BusState: []catalogue.BusStateEntry{bigger}})
if bus.reads == reads || bus.updated == nil {
t.Errorf("a changed option plans the difference: reads %d (was %d), updated %v", bus.reads, reads, bus.updated)
}
k.Heard(Accepted{Module: "m", Digest: "d4", Revision: 4, BusState: nil})
if _, _, found, _ := k.Store.Get(StatePlans, "m"); !found {
t.Errorf("an entry made and now unasked is an orphan the keeper still holds, with its plan")
}
}},
{"a state made before its objects store existed gets the store and its binding said again (§58)", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
p, _ := PlanOf("m", drafts)
bus.held["KV_m_drafts"] = p.Objects[0].Settings
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts}})
out.reset()
bus.created = nil
delete(bus.held, "OBJ_m_drafts-objects")
k.CheckAll()
if fmt.Sprint(bus.created) != "[OBJ_m_drafts-objects]" || fmt.Sprint(out.events) != "[created m drafts]" || out.bodies[0].Binding.Objects != "m_drafts-objects" {
t.Errorf("created %v, said %v: the missing store made, its binding said again", bus.created, out.events)
}
}},
{"a bus that holds an entry the keeper never made is taken as made", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
p, _ := PlanOf("m", drafts)
bus.held["KV_m_drafts"], bus.held["OBJ_m_drafts-objects"] = p.Objects[0].Settings, p.Objects[1].Settings
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts}})
if len(bus.created) != 0 || fmt.Sprint(out.events) != "[created m drafts]" {
t.Errorf("created %v, said %v: its binding said, nothing made twice", bus.created, out.events)
}
}},
// CAT-N5: orphans (design 25 § 12.2-12.3): an empty manifest-accepted asks nothing now, so what was made is an
// orphan, said and raised, kept until a drop; asked again it is kept and its binding said again; a module missing
// from the first view (set aside to wait) keeps its key; one never made leaves nothing behind.
{"an empty verdict orphans what was made, and asking again keeps it", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts}})
out.reset()
k.Heard(Accepted{Module: "m", Revision: 2})
a, held := k.Held()["m"]
if !held || !a.Made["drafts"].Orphaned || len(bus.deleted) != 0 || fmt.Sprint(out.events) != "[orphaned m drafts]" ||
fmt.Sprint(out.conds) != "[raised nats.orphaned.m_drafts]" {
t.Fatalf("held %v %+v, deleted %v, said %v %v: orphaned, kept, raised", held, a, bus.deleted, out.events, out.conds)
}
k.Heard(Accepted{Module: "m", Revision: 3})
k.Heard(Accepted{Module: "m", Revision: 4, BusState: []catalogue.BusStateEntry{drafts}})
if fmt.Sprint(out.events) != "[orphaned m drafts created m drafts]" || out.conds[len(out.conds)-1] != "cleared nats.orphaned.m_drafts" {
t.Errorf("said %v %v: orphaned once, then created again and cleared", out.events, out.conds)
}
}},
{"a module missing from the first view keeps its key; one never made leaves none", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts}})
k.Catalog = func() (CatalogView, error) {
return CatalogView{}, errors.New("the catalog half has no first view of main yet")
}
if err := k.PlanCatalog(); err == nil {
t.Fatal("planned while a watched branch has no first view")
}
if a := k.Held()["m"]; len(a.Entries) != 1 || a.Made["drafts"].Orphaned {
t.Errorf("held %+v: kept as asked, never orphaned while a first view waits", a)
}
bus.refuse["KV_n_x"] = errors.New("no")
k.Heard(Accepted{Module: "n", Revision: 1, BusState: []catalogue.BusStateEntry{entry("x", catalogue.StorageKeyValue, catalogue.TemplateState, nil)}})
k.Heard(Accepted{Module: "n", Revision: 2})
if _, held := k.Held()["n"]; held {
t.Errorf("n's key is still held: nothing of it was made, so its empty verdict ends it")
}
}},
// CAT-N6: fill (ADR 0595 §6): an entry that refuses when full raises `full` at the raise mark and clears only
// below the clear mark; one that drops its oldest is never read; the health names the pass.
{"full at the raise mark, cleared only below the clear mark", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
events := entry("events", catalogue.StorageStream, catalogue.TemplateEvents, map[string]string{"max-mib": "1"})
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts, events}})
out.reset()
bus.fill["m_events"] = 1 * uint64(MiB)
var steps []string
for _, mib := range []uint64{40, 48, 46, 45, 44} {
bus.fill["KV_m_drafts"] = mib * uint64(MiB)
k.FillPass()
steps = append(steps, fmt.Sprintf("%d:%d", mib, len(out.events)))
}
if fmt.Sprint(steps) != "[40:0 48:1 46:1 45:1 44:2]" || fmt.Sprint(out.events) != "[full m drafts full-cleared m drafts]" ||
fmt.Sprint(out.conds) != "[raised nats.full.m_drafts cleared nats.full.m_drafts]" {
t.Errorf("said %v %v: full at 48 of 64 MiB (75%%), kept at 46 and 45 (71%%, 70%%), cleared at 44 (68%%)", out.events, out.conds)
}
h := HealthOf(k.Status(), k.Held(), 0, k.now())
if !h.Healthy || h.Pass == "none" {
t.Errorf("health %+v: healthy, naming its pass", h)
}
}},
{"the health is unhealthy past its bound, naming why", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
at := k.now()
k.FillPass()
later := at.Add(4 * time.Minute)
if h := HealthOf(k.Status(), nil, 0, later); h.Healthy || !strings.Contains(h.Why, "last fill pass is 4m0s old") {
t.Errorf("health %+v: unhealthy, its pass too old", h)
}
if h := HealthOf(k.Status(), nil, 2, later); !strings.Contains(h.Why, "2 accepted verdict(s) wait") {
t.Errorf("health %+v: names the verdicts waiting", h)
}
k.Catalog = func() (CatalogView, error) {
return CatalogView{}, errors.New("the catalog half has no first view of main yet")
}
k.mu.Lock()
k.status.FirstView.Answered = time.Time{}
k.mu.Unlock()
if err := k.PlanCatalog(); err == nil {
t.Fatal("planned with no first view")
}
// The first view unanswered is unhealthy only past the bound since the start (review round 1: health.go:41).
if h := HealthOf(k.Status(), nil, 0, at); !h.Healthy {
t.Errorf("health %+v: healthy within the bound while the first view is asked", h)
}
k.FillPass()
if h := HealthOf(k.Status(), nil, 0, later); h.Healthy || !strings.Contains(h.Why, "mesh-catalog.revision, has not answered: the catalog half has no first view of main yet") {
t.Errorf("health %+v: unhealthy past the bound, naming the verb", h)
}
// What waits on its consumer unread is never skipped silently (review round 1: health.go:47).
if h := HealthOf(k.Status(), nil, -1, at); h.Healthy || !strings.Contains(h.Why, "cannot be read") {
t.Errorf("health %+v: unhealthy when its consumer's backlog cannot be read", h)
}
}},
// CAT-N8: a person's requests: a drop removes an orphan alone, only on the operator's warrant binding exactly
// that drop, after a snapshot, said dropped and recorded; a recreate rebuilds a drifted entry the same way.
{"an ask the router refused ends its request with the router's words (§570)", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
store := k.Store.(*memStore)
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts}})
k.Heard(Accepted{Module: "m", Revision: 2, BusState: nil})
w := &oc.Warrant{Ask: "bus-1", Asker: Module, Outcome: oc.WarrantOutcomeRefused, Words: "an ask that destroys lasts at most 10m0s"}
k.Warrant = func(string) (*oc.Ask, *oc.Warrant, error) { return nil, w, nil }
raw, _ := json.Marshal(Request{ID: "rr", Verb: VerbDrop, Module: "m", Local: "drafts", Ask: "bus-1", Caller: "person.jo", At: k.now()})
_, _ = store.Put(StateRequests, "rr", raw)
k.ActOnRequests()
raw, _, _, _ = store.Get(StateRequests, "rr")
var got Request
_ = json.Unmarshal(raw, &got)
if got.Done.IsZero() || got.Waiting != "" || !strings.Contains(got.Error, "refused") || !strings.Contains(got.Error, w.Words) || len(bus.deleted) != 0 {
t.Errorf("ended %+v, deleted %v: the router's refusal and its words, nothing removed", got, bus.deleted)
}
}},
{"a drop waits for the warrant, refuses another act's, then acts after a snapshot", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
store := k.Store.(*memStore)
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts, log}})
k.Heard(Accepted{Module: "m", Revision: 2, BusState: []catalogue.BusStateEntry{log}})
var acts []string
k.Act = func(what, why, cause string) error { acts = append(acts, cause+": "+what); return nil }
ask := oc.Ask{ID: "bus-1", Who: occ.Operator, Expires: occ.FormatTime(k.now().Add(time.Hour))}
var w *oc.Warrant
k.Warrant = func(string) (*oc.Ask, *oc.Warrant, error) { return &ask, w, nil }
put := func(r Request) Request {
raw, _ := json.Marshal(r)
_, _ = store.Put(StateRequests, r.ID, raw)
k.ActOnRequests()
raw, _, _, _ = store.Get(StateRequests, r.ID)
var got Request
_ = json.Unmarshal(raw, &got)
return got
}
if got := put(Request{ID: "r0", Verb: VerbDrop, Module: "m", Local: "log", Ask: "bus-1", Caller: "person.jo"}); !strings.Contains(got.Error, "drop removes an orphan alone") {
t.Errorf("dropping an asked entry: %+v", got)
}
if got := put(Request{ID: "r1", Verb: VerbDrop, Module: "m", Local: "drafts", Ask: "bus-1", Caller: "person.jo"}); !got.Done.IsZero() || len(bus.deleted) != 0 {
t.Fatalf("acted with no warrant: %+v, deleted %v", got, bus.deleted)
}
binds := func(act occ.Act) string { d, _ := occ.ActDigest(act); return d }
choose := func(act occ.Act) {
ask.Options = []oc.Option{{ID: "go", Label: "Remove", Level: occ.Destroy, Binds: binds(act)}}
w = &oc.Warrant{Ask: "bus-1", Asker: Module, Outcome: occ.OutcomeChosen, By: &oc.Person{Who: occ.Operator},
Option: "go", Label: "Remove", Level: occ.Destroy, At: occ.FormatTime(k.now()), AskDigest: occ.Digest(ask)}
}
choose(occ.Act{"verb": "drop", "module": "m", "local": "log"})
if got := put(Request{ID: "r1", Verb: VerbDrop, Module: "m", Local: "drafts", Ask: "bus-1", Caller: "person.jo"}); !strings.Contains(got.Error, "bound") || len(bus.deleted) != 0 {
t.Fatalf("acted on a warrant for another act: %+v, deleted %v", got, bus.deleted)
}
// ADR 0626 §4: the keeper trusts nothing in the request; each of these is refused, said why, and deletes nothing.
refused := func(r Request, why string) {
t.Helper()
if got := put(r); got.Done.IsZero() || !strings.Contains(got.Error, why) || len(bus.deleted) != 0 {
t.Errorf("%s: ended %+v, deleted %v: refused for %q", r.ID, got, bus.deleted, why)
}
}
refused(Request{ID: "r1a", Verb: VerbDrop, Module: "m", Local: "drafts", Caller: "person.jo"}, "names no ask")
choose(occ.Act{"verb": "drop", "module": "m", "local": "drafts"})
w.Level, w.Label, ask.Options[0].Level = occ.Approve, "Keep", occ.Approve
w.AskDigest = occ.Digest(ask)
refused(Request{ID: "r1b", Verb: VerbDrop, Module: "m", Local: "drafts", Ask: "bus-1", Caller: "person.jo"}, "nothing is removed")
choose(occ.Act{"verb": "drop", "module": "m", "local": "drafts"})
w.Asker = "mesh-grants"
refused(Request{ID: "r1c", Verb: VerbDrop, Module: "m", Local: "drafts", Ask: "bus-1", Caller: "person.jo"}, "no warrant")
refused(Request{ID: "r1d", Verb: VerbCreate, Module: "m", Local: "drafts", Caller: "person.jo"}, "no accepted manifest of m asks for drafts")
choose(occ.Act{"verb": "drop", "module": "m", "local": "drafts"})
bus.refuse["snapshot"] = errors.New("no space")
refused(Request{ID: "r1e", Verb: VerbDrop, Module: "m", Local: "drafts", Ask: "bus-1", Caller: "person.jo"}, "the snapshot before drop failed, so nothing was removed")
delete(bus.refuse, "snapshot")
bus.snaps = nil
out.reset()
got := put(Request{ID: "r2", Verb: VerbDrop, Module: "m", Local: "drafts", Ask: "bus-1", Caller: "person.jo", Why: "unused"})
if got.Error != "" || fmt.Sprint(bus.snaps) != "[[KV_m_drafts OBJ_m_drafts-objects]]" || fmt.Sprint(bus.deleted) != "[KV_m_drafts OBJ_m_drafts-objects]" ||
fmt.Sprint(out.events) != "[dropped m drafts]" || len(acts) != 1 || !strings.Contains(acts[0], "nats_drop: dropped drafts of m") ||
!strings.Contains(acts[0], "person.jo") {
t.Errorf("ended %+v, snapshots %v, deleted %v, said %v, recorded %v", got, bus.snaps, bus.deleted, out.events, acts)
}
if _, made := k.Held()["m"].Made["drafts"]; made {
t.Error("the dropped entry is still held")
}
}},
// Design 47 G §681: the snapshot is recorded in its request (path, size, digest, protected) before the live data is
// touched; one past the cap is refused before anything is removed; the keeper alone removes one past its days, said
// in its journal; the finding snapshots-filling is raised past 80 % of the cap and cleared under it; a snapshot of
// before G (no path, no digest) is never removed by age.
{"a snapshot is recorded, held to the cap, removed after its days, and told as a finding while it fills", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
store, files := k.Store.(*memStore), k.Files.(*memFiles)
var journal []string
k.Logf = func(f string, a ...any) { journal = append(journal, fmt.Sprintf(f, a...)) }
k.SnapshotsCap = 120
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts, log}})
k.Heard(Accepted{Module: "m", Revision: 2, BusState: []catalogue.BusStateEntry{log}})
bus.restricted = map[string]bool{"KV_m_drafts": true}
binds, _ := occ.ActDigest(occ.Act{"verb": "drop", "module": "m", "local": "drafts"})
ask := oc.Ask{ID: "bus-9", Who: occ.Operator, Expires: occ.FormatTime(k.now().Add(time.Hour)), Options: []oc.Option{{ID: "go", Label: "Remove", Level: occ.Destroy, Binds: binds}}}
w := &oc.Warrant{Ask: "bus-9", Asker: Module, Outcome: occ.OutcomeChosen, By: &oc.Person{Who: occ.Operator},
Option: "go", Label: "Remove", Level: occ.Destroy, At: occ.FormatTime(k.now()), AskDigest: occ.Digest(ask)}
k.Warrant = func(string) (*oc.Ask, *oc.Warrant, error) { return &ask, w, nil }
get := func(id string) Request {
raw, _, _, _ := store.Get(StateRequests, id)
var got Request
_ = json.Unmarshal(raw, &got)
return got
}
raw, _ := json.Marshal(Request{ID: "r10", Verb: VerbDrop, Module: "m", Local: "drafts", Ask: "bus-9", Caller: "person.jo", At: k.now()})
_, _ = store.Put(StateRequests, "r10", raw)
out.reset()
k.ActOnRequests()
got := get("r10")
if got.Path != "/snapshots/KV_m_drafts+OBJ_m_drafts-objects.tar" || got.Size != 100 || !strings.HasPrefix(got.Digest, "sha256:") || !got.Protected || got.Error != "" {
t.Fatalf("the drop's record: %+v", got)
}
findings := k.Finding.(*findingLog)
if fmt.Sprint(findings.calls) != "[write module.nats.snapshots.snapshots-filling nats]" {
t.Errorf("100 of 120 bytes is 83%%, so the finding is written through the held writer: %v", findings.calls)
}
// Another snapshot of 100 bytes would pass the cap: refused, its file removed, nothing deleted from the bus.
files.size["/snapshots/m_log.tar"] = 100
err := k.recordSnapshot(context.Background(), &Request{ID: "r11", At: k.now()}, "/snapshots/m_log.tar", []string{"KV_m_log"})
if err == nil || !strings.Contains(err.Error(), "would pass the cap 120") || fmt.Sprint(files.removed) != "[/snapshots/m_log.tar]" {
t.Errorf("a snapshot past the cap: %v, removed %v", err, files.removed)
}
// A snapshot of before G stays; one past 90 days goes with its record, said, and the finding clears.
old, _ := json.Marshal(Request{ID: "r0", Verb: VerbDrop, Snapshot: "/before/g.tar", At: k.now().AddDate(0, 0, -400), Done: k.now()})
_, _ = store.Put(StateRequests, "r0", old)
base := k.now()
k.Now = func() time.Time { return base.AddDate(0, 0, 91) }
findings.calls = nil
k.ExpireSnapshots()
if _, _, found, _ := store.Get(StateRequests, "r10"); found || fmt.Sprint(files.removed) != "[/snapshots/m_log.tar /snapshots/KV_m_drafts+OBJ_m_drafts-objects.tar]" ||
len(journal) != 1 || !strings.Contains(journal[0], "past its 90 days") || fmt.Sprint(findings.calls) != "[clear module.nats.snapshots.snapshots-filling]" {
t.Errorf("after 91 days: removed %v, journal %v, finding calls %v", files.removed, journal, findings.calls)
}
if _, _, found, _ := store.Get(StateRequests, "r0"); !found {
t.Error("a snapshot of before G was removed by age")
}
}},
// Design 47 G §681: the keeper restores a snapshot it took in place: it checks the file against its digest, refuses a
// name that is live, restores within a bound, one at a time, cancels, removes its own partial stream of one that failed
// and says snapshot-restored under the request and the act; a restore a restart cut off is ended so at the start.
{"a snapshot is restored in place, verified, refused while its names are live, cancelled, and its failure removes the partial stream", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
store, files := k.Store.(*memStore), k.Files.(*memFiles)
out.sig = make(chan struct{}, 8)
path := "/snapshots/KV_m_drafts+OBJ_m_drafts-objects.tar"
bus.snapStreams = map[string][]string{path: {"KV_m_drafts"}}
files.missing = map[string]bool{"/snapshots/gone.tar": true}
digest := "sha256:" + hex.EncodeToString([]byte(path))
put := func(r Request) {
raw, _ := json.Marshal(r)
_, _ = store.Put(StateRequests, r.ID, raw)
}
get := func(id string) Request {
raw, _, _, _ := store.Get(StateRequests, id)
var got Request
_ = json.Unmarshal(raw, &got)
return got
}
ask := func(r Request) Request {
r.Caller = "person.jo"
return k.AnswerSnapshot(r)
}
put(Request{ID: "r20", Verb: VerbDrop, Path: path, Size: 100, Digest: digest, Protected: true, At: k.now(), Done: k.now()})
put(Request{ID: "r21", Verb: VerbDrop, Path: path, Size: 100, Digest: "sha256:bad", At: k.now(), Done: k.now()})
put(Request{ID: "r22", Verb: VerbDrop, Path: "/snapshots/gone.tar", Size: 100, Digest: digest, At: k.now(), Done: k.now()})
put(Request{ID: "r23", Verb: VerbDrop, Snapshot: "/before/g.tar", At: k.now(), Done: k.now()})
for _, row := range []struct{ id, of, want string }{{"x1", "r21", "snapshot damaged"}, {"x2", "r22", "snapshot missing"},
{"x3", "r23", "snapshot taken before milestone G"}, {"x4", "r99", "no snapshot of the request r99"}} {
if got := ask(Request{ID: row.id, Verb: VerbSnapshotRestore, Of: row.of, Why: "w"}); got.Outcome != "error" || !strings.Contains(got.Error, row.want) {
t.Errorf("%s: ended %+v, want %q", row.of, got, row.want)
}
}
if got := ask(Request{ID: "v1", Verb: VerbSnapshotRestore, Of: "r20", Verify: true, Why: "w"}); got.Outcome != "verified" || len(bus.restored) != 0 {
t.Errorf("verify: %+v, restored %v", got, bus.restored)
}
bus.held["KV_m_drafts"] = Settings{}
if got := ask(Request{ID: "n1", Verb: VerbSnapshotRestore, Of: "r20", Why: "w"}); got.Outcome != "error" || !strings.Contains(got.Error, "names live") || len(bus.restored) != 0 {
t.Errorf("a live name: %+v, restored %v", got, bus.restored)
}
delete(bus.held, "KV_m_drafts")
// A restore that ends well says so under the request and the act.
if got := ask(Request{ID: "a1", Verb: VerbSnapshotRestore, Of: "r20", Why: "w"}); got.Outcome != "accepted" || got.ActID != "a1" {
t.Fatalf("accepted: %+v", got)
}
<-out.sig
if fmt.Sprint(out.restored) != "[r20 a1 done ]" || fmt.Sprint(bus.restored) != "["+path+"]" || len(bus.deletedStreams) != 0 || get("a1").Result != "done" {
t.Errorf("a restore: said %v, restored %v, removed %v, record %+v", out.restored, bus.restored, bus.deletedStreams, get("a1"))
}
// One that fails removes the keeper's partial stream, then says failed.
bus.refuse["restore"] = errors.New("boom")
ask(Request{ID: "a2", Verb: VerbSnapshotRestore, Of: "r20", Why: "w"})
<-out.sig
if got := out.restored[1]; got != "r20 a2 failed boom" || fmt.Sprint(bus.deletedStreams) != "[KV_m_drafts]" {
t.Errorf("a failed restore: said %q, removed %v", got, bus.deletedStreams)
}
// One at a time, and a running one cancelled by its act.
delete(bus.refuse, "restore")
bus.gate = make(chan struct{})
ask(Request{ID: "a3", Verb: VerbSnapshotRestore, Of: "r20", Why: "w"})
if got := ask(Request{ID: "a4", Verb: VerbSnapshotRestore, Of: "r20", Why: "w"}); got.Outcome != "busy" || got.ActID != "a3" {
t.Errorf("a second restore: %+v", got)
}
if got := ask(Request{ID: "c1", Verb: VerbSnapshotRestoreCancel, Of: "a3", Why: "w"}); got.Outcome != "accepted" || got.ActID != "a3" {
t.Errorf("cancel: %+v", got)
}
<-out.sig
if got := out.restored[2]; got != "r20 a3 failed cancelled" || len(bus.deletedStreams) != 2 {
t.Errorf("a cancelled restore: said %q, removed %v", got, bus.deletedStreams)
}
if got := ask(Request{ID: "c2", Verb: VerbSnapshotRestoreCancel, Of: "a3", Why: "w"}); got.Outcome != "error" {
t.Errorf("cancel of an ended act: %+v", got)
}
// A restore written into the state requests is no request of a person's: the keeper answers the verb on the seat alone (A-107).
before := len(bus.restored)
put(Request{ID: "z1", Verb: VerbSnapshotRestore, Of: "r20", Why: "w", Caller: "person.jo"})
k.ActOnRequests()
if got := get("z1"); len(bus.restored) != before || got.Outcome == "accepted" || got.Error == "" {
t.Errorf("a restore handed through the requests: %+v, restored %v", got, bus.restored)
}
// A restore a restart cut off is ended at the keeper's start.
put(Request{ID: "a5", Verb: VerbSnapshotRestore, Of: "r20", Outcome: "accepted", Done: k.now()})
k.EndCutOffRestores()
<-out.sig
if got := out.restored[3]; !strings.HasPrefix(got, "r20 a5 failed cut off") || get("a5").Result != "failed" || len(bus.deletedStreams) != 3 {
t.Errorf("a cut-off restore: said %q, record %+v, removed %v", got, get("a5"), bus.deletedStreams)
}
}},
// Design 47 G §681: snapshot-remove removes the snapshot of one request, or every one past its days, written in the
// hand-act log first; a snapshot that cannot be recorded stays; one of before G goes only by its request.
{"snapshot-remove removes by request or all expired, after the hand act", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
store, files := k.Store.(*memStore), k.Files.(*memFiles)
var acts []string
actErr := error(nil)
k.Act = func(what, why, cause string) error { acts = append(acts, cause+": "+what); return actErr }
put := func(r Request) {
raw, _ := json.Marshal(r)
_, _ = store.Put(StateRequests, r.ID, raw)
}
get := func(id string) Request {
raw, _, _, _ := store.Get(StateRequests, id)
var got Request
_ = json.Unmarshal(raw, &got)
return got
}
old := k.now().AddDate(0, 0, -100)
soon := k.now().AddDate(0, 0, -80)
put(Request{ID: "s1", Verb: VerbDrop, Path: "/snapshots/a.tar", Size: 10, Digest: "sha256:a", At: k.now(), Done: k.now()})
put(Request{ID: "s2", Verb: VerbDrop, Path: "/snapshots/b.tar", Size: 10, Digest: "sha256:b", At: soon, Done: soon})
put(Request{ID: "s3", Verb: VerbDrop, Snapshot: "/before/g.tar", At: old, Done: old})
ask := func(r Request) Request {
r.Caller = "person.jo"
return k.AnswerSnapshot(r)
}
for _, r := range []Request{{ID: "x1", Verb: VerbSnapshotRemove, Why: "w"}, {ID: "x2", Verb: VerbSnapshotRemove, Of: "s1", AllExpired: true, Why: "w"}} {
if got := ask(r); got.Outcome != "error" || !strings.Contains(got.Error, "not both and not neither") {
t.Errorf("%s: %+v", r.ID, got)
}
}
actErr = errors.New("the log is down")
if got := ask(Request{ID: "x3", Verb: VerbSnapshotRemove, Of: "s1", Why: "w"}); got.Outcome != "error" || len(files.removed) != 0 || get("s1").Path == "" {
t.Errorf("an unrecorded hand act: %+v, removed %v", got, files.removed)
}
actErr = nil
if got := ask(Request{ID: "x4", Verb: VerbSnapshotRemove, Of: "s1", Why: "w"}); got.Outcome != "accepted" || fmt.Sprint(files.removed) != "[/snapshots/a.tar]" || get("s1").ID != "" {
t.Errorf("by request: %+v, removed %v", got, files.removed)
}
// The keeper's pass at each request removes what is past its days first, so the verb is driven apart from it.
base := k.now()
k.Now = func() time.Time { return base.AddDate(0, 0, 20) }
all := &Request{ID: "x5", Verb: VerbSnapshotRemove, AllExpired: true, Why: "w", Caller: "person.jo"}
if outcome, err := k.removeSnapshots(all); outcome != "accepted" || err != nil || all.Result != "1 removed" || get("s2").ID != "" || get("s3").ID == "" {
t.Errorf("all expired: %q %v %+v: s2 goes, s3 (before G) stays", outcome, err, all)
}
k.Now = func() time.Time { return base }
if got := ask(Request{ID: "x6", Verb: VerbSnapshotRemove, Of: "s3", Why: "w"}); got.Outcome != "accepted" || get("s3").ID != "" || files.removed[len(files.removed)-1] != "/before/g.tar" {
t.Errorf("a snapshot of before G by its request: %+v, removed %v", got, files.removed)
}
if len(acts) != 4 || !strings.Contains(acts[len(acts)-1], "asked by person.jo") {
t.Errorf("the hand acts: %v", acts)
}
}},
{"a recreate rebuilds a drifted entry after a snapshot", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
store := k.Store.(*memStore)
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{log}})
k.Heard(Accepted{Module: "m", Revision: 2, BusState: []catalogue.BusStateEntry{entry("log", catalogue.StorageStream, catalogue.TemplateLog, map[string]string{"max-mib": "16"})}})
binds, _ := occ.ActDigest(occ.Act{"verb": "recreate", "module": "m", "local": "log"})
ask := oc.Ask{ID: "bus-2", Who: occ.Operator, Expires: occ.FormatTime(k.now().Add(time.Hour)), Options: []oc.Option{{ID: "go", Label: "Rebuild", Level: occ.Destroy, Binds: binds}}}
w := &oc.Warrant{Ask: "bus-2", Asker: Module, Outcome: occ.OutcomeChosen, By: &oc.Person{Who: occ.Operator},
Option: "go", Label: "Rebuild", Level: occ.Destroy, At: occ.FormatTime(k.now()), AskDigest: occ.Digest(ask)}
k.Warrant = func(string) (*oc.Ask, *oc.Warrant, error) { return &ask, w, nil }
out.reset()
raw, _ := json.Marshal(Request{ID: "r3", Verb: VerbRecreate, Module: "m", Local: "log", Ask: "bus-2"})
_, _ = store.Put(StateRequests, "r3", raw)
k.ActOnRequests()
if bus.held["m_log"].MaxBytes != 16*MiB || fmt.Sprint(out.events) != "[recreated m log]" ||
fmt.Sprint(out.conds) != "[cleared nats.drifted.m_log]" || k.Held()["m"].Made["log"].Drift != "" {
t.Errorf("held %+v, said %v %v: rebuilt as asked, the drift cleared", bus.held["m_log"], out.events, out.conds)
}
}},
// Review round 1 (security F2; design 25 § 12.3): a warrant acts once, spent by compare-and-set before anything is
// deleted; a second request naming the same ask is refused and deletes nothing.
{"a warrant acts once", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
store := k.Store.(*memStore)
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts, log}})
k.Heard(Accepted{Module: "m", Revision: 2})
binds, _ := occ.ActDigest(occ.Act{"verb": "drop", "module": "m", "local": "drafts"})
ask := oc.Ask{ID: "bus-5", Who: occ.Operator, Expires: occ.FormatTime(k.now().Add(time.Hour)), Options: []oc.Option{{ID: "go", Label: "Remove", Level: occ.Destroy, Binds: binds}}}
w := &oc.Warrant{Ask: "bus-5", Asker: Module, Outcome: occ.OutcomeChosen, By: &oc.Person{Who: occ.Operator},
Option: "go", Label: "Remove", Level: occ.Destroy, At: occ.FormatTime(k.now()), AskDigest: occ.Digest(ask)}
k.Warrant = func(string) (*oc.Ask, *oc.Warrant, error) { return &ask, w, nil }
put := func(id string) Request {
raw, _ := json.Marshal(Request{ID: id, Verb: VerbDrop, Module: "m", Local: "drafts", Ask: "bus-5", Caller: "person.jo"})
_, _ = store.Put(StateRequests, id, raw)
k.ActOnRequests()
raw, _, _, _ = store.Get(StateRequests, id)
var got Request
_ = json.Unmarshal(raw, &got)
return got
}
if got := put("r5"); got.Error != "" || len(bus.deleted) != 2 {
t.Fatalf("the first drop: %+v, deleted %v", got, bus.deleted)
}
// The entry is asked for and orphaned again, so a second drop would remove it: the spent warrant refuses it.
k.Heard(Accepted{Module: "m", Revision: 3, BusState: []catalogue.BusStateEntry{drafts, log}})
k.Heard(Accepted{Module: "m", Revision: 4, BusState: []catalogue.BusStateEntry{log}})
if got := put("r6"); !strings.Contains(got.Error, "already acted (drop of m's drafts, request r5") || len(bus.deleted) != 2 {
t.Errorf("the second request on the same warrant: %+v, deleted %v", got, bus.deleted)
}
if raw, _, found, _ := store.Get(StateSpent, "bus-5"); !found || !strings.Contains(string(raw), `"request":"r5"`) {
t.Errorf("the spend record is %s (%v)", raw, found)
}
// The spend record is the process's own state, never in requests, which the tools write: every key of requests
// gone, a third request on the same warrant is still refused (review round 2, security F2).
keys, _ := store.Keys(StateRequests)
for _, key := range keys {
_ = store.Delete(StateRequests, key)
}
if got := put("r6b"); !strings.Contains(got.Error, "already acted") || len(bus.deleted) != 2 {
t.Errorf("with requests wiped, the spent warrant: %+v, deleted %v", got, bus.deleted)
}
}},
// An ask that ended unanswered ends its request, said, instead of a request that waits until its key ages out; and a
// hand act that cannot be recorded is said on the request's end, never only logged.
{"an ask ended unanswered ends its request; an unrecorded hand act is said", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
store := k.Store.(*memStore)
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts, log}})
k.Heard(Accepted{Module: "m", Revision: 2, BusState: []catalogue.BusStateEntry{log}})
k.Warrant = func(string) (*oc.Ask, *oc.Warrant, error) { return nil, nil, nil }
get := func(id string) Request {
raw, _, _, _ := store.Get(StateRequests, id)
var got Request
_ = json.Unmarshal(raw, &got)
return got
}
raw, _ := json.Marshal(Request{ID: "r7", Verb: VerbDrop, Module: "m", Local: "drafts", Ask: "bus-7", Caller: "person.jo", At: k.now().Add(-AskLasts / 2)})
_, _ = store.Put(StateRequests, "r7", raw)
raw, _ = json.Marshal(Request{ID: "r8", Verb: VerbDrop, Module: "m", Local: "drafts", Ask: "bus-8", Caller: "person.jo", At: k.now().Add(-AskLasts - time.Minute)})
_, _ = store.Put(StateRequests, "r8", raw)
k.ActOnRequests()
if got := get("r7"); !got.Done.IsZero() {
t.Errorf("a request whose ask still runs ended: %+v", got)
}
if got := get("r8"); got.Done.IsZero() || !strings.Contains(got.Error, "ended at") || len(bus.deleted) != 0 {
t.Errorf("a request whose ask ended unanswered: %+v, deleted %v", got, bus.deleted)
}
k.Act = func(string, string, string) error { return errors.New("the hand-act seat does not answer") }
raw, _ = json.Marshal(Request{ID: "r9", Verb: VerbCreate, Module: "m", Local: "log", Caller: "person.jo"})
_, _ = store.Put(StateRequests, "r9", raw)
k.ActOnRequests()
if got := get("r9"); got.Done.IsZero() || !strings.Contains(got.Unrecorded, "does not answer") {
t.Errorf("an unrecorded hand act: %+v", got)
}
}},
// Review round 1 (health F1): created is marked only once said, so the grants part always gets its binding; an
// event or a condition the bus did not take is kept, named by the health, and said again at the next fill pass
// with the same id.
{"what the bus did not take is said again; created marked only once said", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
out.down = true
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts, entry("raw", catalogue.StorageStream, "", nil)}})
if m := k.Held()["m"].Made["drafts"]; m.Created || !strings.Contains(m.Waiting, "its created could not be said yet") {
t.Fatalf("made %+v: not created until said, waiting", m)
}
st := k.Status()
if len(st.Unsaid) != 1 || st.Unsaid[0].Event != EventRefused || len(st.UnsaidConditions) != 1 {
t.Fatalf("unsaid %+v %+v: the refused event and its condition kept", st.Unsaid, st.UnsaidConditions)
}
if h := HealthOf(k.Status(), nil, 0, k.now()); h.Healthy || !strings.Contains(h.Why, "2 word(s) the bus did not take") {
t.Errorf("health %+v: unhealthy, naming what was not said", h)
}
id := st.Unsaid[0].ID
out.down = false
k.FillPass()
k.CheckAll()
if !strings.Contains(fmt.Sprint(out.events), "refused m raw") || !strings.Contains(fmt.Sprint(out.events), "created m drafts") ||
fmt.Sprint(out.conds) != "[raised nats.refused.m_raw]" || out.ids[0] != id {
t.Errorf("said %v (ids %v) %v: the refused said again with its id %s, its condition raised, created said", out.events, out.ids, out.conds, id)
}
if st := k.Status(); len(st.Unsaid) != 0 || len(st.UnsaidConditions) != 0 || !k.Held()["m"].Made["drafts"].Created {
t.Errorf("after the pass: unsaid %+v %+v, drafts %+v", st.Unsaid, st.UnsaidConditions, k.Held()["m"].Made["drafts"])
}
}},
// Review round 2 (health note 3): created said on the seat's subject and refused by its history is created, and its
// history is said again at the fill pass with the same id, on the history alone.
{"created said on the seat and not its history is said again on what is left", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
out.partly = true
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts}})
st := k.Status()
if m := k.Held()["m"].Made["drafts"]; !m.Created || len(st.Unsaid) != 1 || fmt.Sprint(st.Unsaid[0].Left) != "[history]" {
t.Fatalf("made %+v, unsaid %+v: created, its history kept to say", m, st.Unsaid)
}
id := st.Unsaid[0].ID
out.partly = false
k.FillPass()
if last := out.left[len(out.left)-1]; fmt.Sprint(last) != "[history]" || out.ids[len(out.ids)-1] != id || len(k.Status().Unsaid) != 0 {
t.Errorf("said again on %v with id %s (want [history], %s); unsaid %+v", last, out.ids[len(out.ids)-1], id, k.Status().Unsaid)
}
if n := strings.Count(fmt.Sprint(out.ids), id); n != 2 {
t.Errorf("the id %s was used %d times: once on the seat, once on the history", id, n)
}
}},
// Review round 2 (seams note 3, health note 4): words unsaid with a status that cannot be written raise the keeper's
// own condition, once, cleared when the status is written again; the health names what a restart would lose.
{"a status that cannot be written with words unsaid is said", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
store := k.Store.(*memStore)
k.FillPass()
store.refuse[StateAsked+"/"+KeyStatus] = errors.New("the bucket is full")
k.Emit = func(e *Saying) error { e.Left = []string{"seat", "history"}; return errors.New("no responders") }
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts, entry("raw", catalogue.StorageStream, "", nil)}})
k.FillPass()
raisedN := strings.Count(fmt.Sprint(out.conds), "raised nats.status-unwritten.nats_keeper")
if raisedN != 1 {
t.Errorf("said %v: status-unwritten raised once", out.conds)
}
if h := HealthOf(k.Status(), nil, 0, k.now().Add(4*time.Minute)); !strings.Contains(h.Why, "may be lost at a restart") {
t.Errorf("health %+v: names what a restart loses", h)
}
delete(store.refuse, StateAsked+"/"+KeyStatus)
k.Emit = out.emit
k.FillPass()
if !strings.Contains(fmt.Sprint(out.conds), "cleared nats.status-unwritten.nats_keeper") {
t.Errorf("said %v: cleared once the status is written", out.conds)
}
}},
// CAT-N2b: nats_first_view_again asks the first view at once, by hand, recorded as a hand act.
{"the first view asked again by hand", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
store := k.Store.(*memStore)
var acts []string
k.Act = func(what, why, cause string) error { acts = append(acts, cause+": "+what); return nil }
k.Catalog = func() (CatalogView, error) {
return CatalogView{Modules: []CatalogModule{{Accepted: Accepted{Module: "m", Revision: 8, BusState: []catalogue.BusStateEntry{drafts}}}}}, nil
}
again := 0
k.Again = func() error { again++; return nil }
raw, _ := json.Marshal(Request{ID: "r9", Verb: VerbFirstViewAgain, Why: "a verdict thought missed", Caller: "person.jo"})
_, _ = store.Put(StateRequests, "r9", raw)
k.ActOnRequests()
raw, _, _, _ = store.Get(StateRequests, "r9")
var got Request
_ = json.Unmarshal(raw, &got)
if got.Outcome != "the first views answered at revision 8" || fmt.Sprint(bus.created) != "[KV_m_drafts OBJ_m_drafts-objects]" || again != 1 ||
len(acts) != 1 || !strings.Contains(acts[0], "nats_first_view_again: asked the catalog half's first views (mesh-catalog.revision) again by hand") {
t.Errorf("ended %+v, created %v, recorded %v", got, bus.created, acts)
}
}},
} {
t.Run(row.name, func(t *testing.T) {
k, bus, _, out := newKeeper(CatalogView{})
if err := k.PlanCatalog(); err != nil {
t.Fatal(err)
}
row.run(t, k, bus, out)
})
}
}
func TestSends(t *testing.T) {
defs := map[string]seatdef.Definition{
"verbed": {Verbs: []seatdef.Verb{{Name: "go"}}},
"evented": {Events: []seatdef.Message{{Name: "done"}}},
"included": {Includes: []string{"part"}},
"bare": {},
}
for _, c := range []struct {
name string
claims []string
own bool
want bool
}{
{"a seat with a verb", []string{"verbed"}, false, true},
{"a seat with an event", []string{"evented"}, false, true},
{"a seat that only includes parts (§537: resolved by the seat list, including alone counts for nothing)", []string{"included"}, false, false},
{"a seat with neither", []string{"bare"}, false, false},
{"a seat the list does not hold", []string{"unknown"}, false, false},
{"an event of its own", nil, true, true},
{"nothing claimed", nil, false, false},
{"one of two claims sends", []string{"bare", "verbed"}, false, true},
} {
if got := Sends(c.claims, c.own, defs); got != c.want {
t.Errorf("%s: Sends = %v, want %v", c.name, got, c.want)
}
}
}
Reference in New Issue View Git Blame Copy Permalink