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

nats: the keeper wakes on decided, spends a warrant once, and says again what the bus did not take

Browse Source 
Review round 1 of the registry milestone (design 25 § 12.2, § 12.3, § 12.7):
- the fill pass runs from the start, while the first view is asked;
- recreate and drop wake on the operator channel's decided on the keeper's durable consumer, never a minute's tick;
- a warrant is spent by compare-and-set (acted.<ask id> in requests) before anything is deleted; a second request on it
  is refused; a request whose ask ended unanswered is ended, said; the ask is raised before the request is handed;
- created is marked only once said; 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 its one id, on the subjects left (the double publish);
- the health judges the first view past its bound and never skips an unread consumer backlog;
- an unrecordable hand act is said on the request; nats_requests reads what waits, what ended and each warrant spent,
  and drops a waiting request by hand; a tool waits for its request's end on one watch of requests;
- the whole-module test reads the warrant from the router's record under a process connection.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
jschoubben
2026-11-04 17:19:54 +01:00
parent 1443adc4f0
commit f735392cb9
10 changed files with 754 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/nats/cmd/nats-tools/keep.go
+100 -65
View File
@@ -2,9 +2,8 @@ package main
import (
"context"
"crypto/rand"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"log"
@@ -32,9 +31,10 @@ import (
)
// The keeper (`nats-tools keep`): the bus module's process on its own account (own secret broker-keeper, ADR 0436 §1),
// started by the node-engine as the manifest's process keeper. It asks the indexer's first view once, then hears
// manifest-accepted on its durable consumer, makes what each asks for, reads the fill every minute, and acts on the
// persons' requests its tools put in its own state requests (requests.go). It serves nothing (ADR 0436 §3).
// started by the node-engine as the manifest's process keeper. It reads the fill every minute from its start, asks the
// indexer's first view once, then hears manifest-accepted and the operator's decided on its durable consumer, makes what
// each verdict asks for, and acts on the persons' requests its tools put in its own state requests (requests.go) as they
// are written or answered. It serves nothing (ADR 0436 §3).
// AcceptedOf reads a manifest-accepted body: mesh-sdk's delivery.ManifestAccepted, with the indexer's revision of the
// verdict when the event carries it (a field mesh-sdk is asked to add; 0 until then).
@@ -58,6 +58,25 @@ var AcceptedSubject = "mesh.mod." + keeper.FirstViewOwner + ".event." + delivery
// The router's records of asks, read under this module's own name (ADR 0259 §3), as the mail sender reads them.
const asksBucket = "messenger_asks"
// WarrantFrom reads the operator's warrant on one of the bus module's asks from the router's own record of it, under the
// keeper's own account (read: process.Conn.Ask): the ask and its warrant, or nil while the operator has not answered.
func WarrantFrom(read func(bucket, id string) (*stdio.StateEntry, error)) func(id string) (*oc.Ask, *oc.Warrant, error) {
return func(id string) (*oc.Ask, *oc.Warrant, error) {
e, err := read(asksBucket, id)
if err != nil || e == nil {
return nil, nil, err
}
var rec struct {
Ask *oc.Ask `json:"ask"`
Warrant *oc.Warrant `json:"warrant"`
}
if err := json.Unmarshal(e.Value, &rec); err != nil {
return nil, nil, err
}
return rec.Ask, rec.Warrant, nil
}
}
// processStore is the keeper's own state through its process connection.
type processStore struct{ c *process.Conn }
@@ -77,34 +96,55 @@ func (s processStore) Delete(state, key string) error { return s.c.State(state).
func (s processStore) Keys(state string) ([]string, error) { return s.c.State(state).Keys() }
// streamEmitter says an event of the seat mesh-broker, with the SDK's headers, twice under one event id (the lead's ruling,
func (s processStore) Create(state, key string, value []byte) (uint64, error) {
rev, err := s.c.State(state).Create(key, json.RawMessage(value))
if errors.Is(err, stdio.ErrChanged) {
return 0, keeper.ErrExists
}
return rev, err
}
// streamEmitter says an event of the seat mesh-broker, with the SDK's headers, under its one event id (the lead's ruling,
// registry milestone, 13:0x; design 25 § 12.3): on the seat's own subject `mesh.seat.mesh-broker.event.<event>`, which the
// shared events stream keeps 7 days and every consumer of `mesh-broker.<event>` hears, like any seat's event; then on the
// bus module's own stream (its old `streams` section, 30 days, 256 MiB), which its tool `history` reads. The seat's subject
// first: a reader hears it before the history holds it, never the other way round.
func streamEmitter(js jetstream.JetStream, a process.Account) func(string, keeper.Event) error {
return func(event string, body keeper.Event) error {
data, err := json.Marshal(body)
// first: a reader hears it before the history holds it, never the other way round. It says it on s.Left (every subject
// when nil) and leaves in s.Left what the bus did not take, so the keeper says that again with the same id and the bus
// keeps it once (the two publishes are not one act: a failed second one is kept and said again, never silent).
func streamEmitter(js jetstream.JetStream, a process.Account) func(*keeper.Saying) error {
return func(s *keeper.Saying) error {
data, err := json.Marshal(s.Body)
if err != nil {
return err
}
id := eventID()
at := time.Now().UTC().Format(time.RFC3339Nano)
for _, subject := range EventSubjects(a, event) {
if s.ID == "" {
s.ID = keeper.EventID()
}
at := s.At
if at.IsZero() {
at = time.Now().UTC()
}
subjects := s.Left
if subjects == nil {
subjects = EventSubjects(a, s.Event)
}
for i, subject := range subjects {
msg := nats.NewMsg(subject)
msg.Header.Set("x-event-id", id)
msg.Header.Set("x-event-id", s.ID)
msg.Header.Set("x-source", a.Module)
msg.Header.Set("x-node", a.Node)
msg.Header.Set("x-time", at)
msg.Header.Set("x-time", at.UTC().Format(time.RFC3339Nano))
msg.Header.Set("content-type", "application/json")
msg.Data = data
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
_, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(id))
_, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(s.ID))
cancel()
if err != nil {
return fmt.Errorf("saying %s on %s: %w", event, subject, err)
s.Left = append([]string(nil), subjects[i:]...)
return fmt.Errorf("saying %s on %s: %w", s.Event, subject, err)
}
}
s.Left = nil
return nil
}
}
@@ -115,12 +155,6 @@ func EventSubjects(a process.Account, event string) []string {
return []string{seatdef.EventSubject(catalogue.BrokerSeat, event), a.StreamSubject(event)}
}
func eventID() string {
var b [16]byte
_, _ = rand.Read(b[:])
return hex.EncodeToString(b[:])
}
// sayer raises and clears the keeper's conditions through the condition keeper, as every module says them.
func sayer(js jetstream.JetStream) func(oc.Condition) error {
return func(c oc.Condition) error {
@@ -201,22 +235,9 @@ func runKeeper(ctx context.Context, getenv func(string) string, logOut io.Writer
_, err := recorder.Record(handacts.Request{What: what, Why: why, Cause: cause})
return err
},
Warrant: func(id string) (*oc.Ask, *oc.Warrant, error) {
e, err := conn.Ask(asksBucket, id)
if err != nil || e == nil {
return nil, nil, err
}
var rec struct {
Ask *oc.Ask `json:"ask"`
Warrant *oc.Warrant `json:"warrant"`
}
if err := json.Unmarshal(e.Value, &rec); err != nil {
return nil, nil, err
}
return rec.Ask, rec.Warrant, nil
},
Logf: logger.Printf,
Raise: percent(getenv, "MESH_NATS_FULL_RAISE", 75), Clear: percent(getenv, "MESH_NATS_FULL_CLEAR", 70),
Warrant: WarrantFrom(conn.Ask),
Logf: logger.Printf,
Raise: percent(getenv, "MESH_NATS_FULL_RAISE", 75), Clear: percent(getenv, "MESH_NATS_FULL_CLEAR", 70),
}
logger.Printf("started on the account %s: asking %s.%s for the first view", account.User, keeper.FirstViewOwner, keeper.FirstViewVerb)
return Keep(ctx, k, hearer(js, account, logger.Printf), requestsOf(conn), logger.Printf)
@@ -246,11 +267,20 @@ func hearer(js jetstream.JetStream, a process.Account, logf func(string, ...any)
}
}
// Keep is the keeper's loop, apart from the bus so a test drives it: the first view asked until it answers, on its
// backoff; then hear, which applies each manifest-accepted; a fill pass every minute; the waiting entries made again
// on the same backoff; and the persons' requests acted on as they come. It returns when ctx ends.
// DecidedSubject is where the operator channel says the operator's answer to one of the bus module's own asks (ADR
// 0259 §6; ADR 0620 §2), heard on the keeper's durable consumer: the warrant a recreate or a drop waits for.
var DecidedSubject = oc.SubjectDecided(keeper.Module)
// Keep is the keeper's loop, apart from the bus so a test drives it: a fill pass at once and every minute, from its
// start, so the store's one guard runs while the first view is still asked (design 25 § 12.3); the first view asked
// until it answers, on its backoff; then hear, which applies each manifest-accepted and wakes the persons' requests on
// the operator channel's decided for the bus module's own asks; the waiting entries made again on the same backoff; and
// the persons' requests acted on as they are written or answered, never on a timer. It returns when ctx ends.
func Keep(ctx context.Context, k *keeper.Keeper, hear func(context.Context, func(string, []byte) error) error,
requests func(context.Context, func()) error, logf func(string, ...any)) error {
tick := time.NewTicker(time.Minute)
defer tick.Stop()
k.FillPass()
var wait time.Duration
for {
next, err := k.AskFirstView(wait)
@@ -259,22 +289,36 @@ func Keep(ctx context.Context, k *keeper.Keeper, hear func(context.Context, func
}
logf("the first view did not answer (%v): nothing is created or orphaned until it does; asked again in %s", err, next)
wait = next
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(next):
again := time.After(next)
waiting:
for {
select {
case <-ctx.Done():
return ctx.Err()
case <-tick.C:
k.FillPass()
case <-again:
break waiting
}
}
}
logf("the first view answered at revision %d", k.Status().FirstView.Revision)
heard := make(chan struct{}, 1)
poke := func() {
select {
case heard <- struct{}{}:
default:
wake := func(c chan struct{}) func() {
return func() {
select {
case c <- struct{}{}:
default:
}
}
}
heard, asked := make(chan struct{}, 1), make(chan struct{}, 1)
for backoff := 10 * time.Second; ; backoff = min(backoff*2, 5*time.Minute) {
err := hear(ctx, func(subject string, body []byte) error {
if subject == DecidedSubject {
// The operator answered one of the bus module's asks: the waiting requests are read again now.
wake(asked)()
return nil
}
if subject != AcceptedSubject && !strings.HasSuffix(subject, ".event."+delivery.ManifestAcceptedEvent) {
return nil
}
@@ -284,7 +328,7 @@ func Keep(ctx context.Context, k *keeper.Keeper, hear func(context.Context, func
return nil
}
k.Heard(a)
poke()
wake(heard)()
return nil
})
if err == nil {
@@ -297,22 +341,15 @@ func Keep(ctx context.Context, k *keeper.Keeper, hear func(context.Context, func
case <-time.After(backoff):
}
}
asked := make(chan struct{}, 1)
if requests != nil {
if err := requests(ctx, func() {
select {
case asked <- struct{}{}:
default:
}
}); err != nil {
logf("the persons' requests cannot be watched (%v): acted on at each fill pass", err)
if err := requests(ctx, wake(asked)); err != nil {
logf("the persons' requests cannot be watched (%v): acted on as the operator answers", err)
}
}
tick := time.NewTicker(time.Minute)
defer tick.Stop()
// What waited while the keeper was down, or before its first view, is acted on now.
k.ActOnRequests()
retry := 10 * time.Second
var again <-chan time.Time
k.FillPass()
for {
if len(k.Waiting()) > 0 && again == nil {
again = time.After(retry)
@@ -322,8 +359,6 @@ func Keep(ctx context.Context, k *keeper.Keeper, hear func(context.Context, func
return ctx.Err()
case <-tick.C:
k.FillPass()
// A recreate or a drop waits for the operator's warrant: read again once a minute, as an asker reads its asks.
k.ActOnRequests()
case <-asked:
k.ActOnRequests()
case <-heard:
modules/nats/cmd/nats-tools/keeper_integration_test.go
+69 -13
View File
@@ -67,6 +67,18 @@ func (s kvStore) Delete(state, key string) error {
return kv.Delete(context.Background(), key)
}
func (s kvStore) Create(state, key string, value []byte) (uint64, error) {
kv, err := s.kv(state)
if err != nil {
return 0, err
}
rev, err := kv.Create(context.Background(), key, value)
if errors.Is(err, jetstream.ErrKeyExists) {
return 0, keeper.ErrExists
}
return rev, err
}
func (s kvStore) Keys(state string) ([]string, error) {
kv, err := s.kv(state)
if err != nil {
@@ -124,10 +136,23 @@ func TestTheBusModuleRunsWhole(t *testing.T) {
t.Fatal(err)
}
}
// The keeper's consumer hears the indexer's verdicts and the operator channel's decided for the bus module's own
// asks, as the grants part composes it from its consumes and the seats it uses.
if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: account.QueueConsumerName(),
FilterSubject: AcceptedSubject, AckPolicy: jetstream.AckExplicitPolicy, MaxDeliver: 5}); err != nil {
FilterSubjects: []string{AcceptedSubject, DecidedSubject}, AckPolicy: jetstream.AckExplicitPolicy, MaxDeliver: 5}); err != nil {
t.Fatal(err)
}
// The router's own record of the asks, which the keeper reads under its own process account (process.Conn.Ask), never
// a stand-in: an ask's record is `<asker>.<ask id>`, as the router keeps it.
asksKV, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: asksBucket, History: 5})
if err != nil {
t.Fatal(err)
}
pconn, err := process.Connect(process.Account{URL: s.ClientURL(), Node: account.Node, Module: account.Module, User: account.User, Password: "p"})
if err != nil {
t.Fatal(err)
}
defer pconn.Close()
for _, st := range []string{keeper.StateAsked, keeper.StateRequests} {
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "nats_" + st, History: 5}); err != nil {
t.Fatal(err)
@@ -138,8 +163,6 @@ func TestTheBusModuleRunsWhole(t *testing.T) {
var mu sync.Mutex
view := keeper.FirstView{Revision: 1}
var acts []string
var ask *oc.Ask
var warrant *oc.Warrant
newKeeper := func() *keeper.Keeper {
return &keeper.Keeper{Bus: keeper.JetStream{JS: js, Snap: bussnap.Into(nc, snapshots)}, Store: store,
Emit: streamEmitter(js, account), Say: sayer(js),
@@ -150,7 +173,7 @@ func TestTheBusModuleRunsWhole(t *testing.T) {
acts = append(acts, cause+": "+what)
return nil
},
Warrant: func(string) (*oc.Ask, *oc.Warrant, error) { mu.Lock(); defer mu.Unlock(); return ask, warrant, nil },
Warrant: WarrantFrom(pconn.Ask),
Logf: t.Logf}
}
run := func(k *keeper.Keeper) func() {
@@ -221,6 +244,15 @@ func TestTheBusModuleRunsWhole(t *testing.T) {
state := entry("drafts", "key-value", "state", nil)
log := entry("log", "stream", "log", map[string]string{"max-mib": "1"})
// Review round 1 (running F1): the fill pass runs from the start, while the first view is still asked, so the store's
// one guard never waits on the indexer.
t.Run("the fill pass runs while the first view is asked", func(t *testing.T) {
down := newKeeper()
down.View = func() (keeper.FirstView, error) { return keeper.FirstView{}, errors.New("no responders") }
stopDown := run(down)
eventually(t, "a fill pass with no first view", func() bool { return !down.Status().Pass.IsZero() && down.Status().FirstView.Answered.IsZero() })
stopDown()
})
k := newKeeper()
stop := run(k)
t.Run("a verdict, then created, with its binding", func(t *testing.T) {
@@ -317,23 +349,35 @@ func TestTheBusModuleRunsWhole(t *testing.T) {
}
put(keeper.Request{ID: "r1", Verb: keeper.VerbDrop, Module: "m", Local: "drafts", Caller: "person.jo", Why: "unused"})
eventually(t, "the drop with no ask refused", func() bool { return strings.Contains(get("r1").Error, "operator's warrant") })
r := keeper.Request{ID: "r2", Verb: keeper.VerbDrop, Module: "m", Local: "drafts", Caller: "person.jo", Why: "unused"}
r := keeper.Request{ID: "r2", Verb: keeper.VerbDrop, Module: "m", Local: "drafts", Caller: "person.jo", Why: "unused", At: time.Now()}
a, _ := askFor(r, time.Now())
r.Ask = a.ID
mu.Lock()
ask = &a
mu.Unlock()
record := func(w *oc.Warrant) {
raw, _ := json.Marshal(map[string]any{"asker": keeper.Module, "ask": a, "state": "open", "warrant": w})
if _, err := asksKV.Put(ctx, keeper.Module+"."+a.ID, raw); err != nil {
t.Fatal(err)
}
}
record(nil)
put(r)
k.ActOnRequests()
eventually(t, "the request read, waiting for the operator", func() bool { return get("r2").Waiting != "" })
if _, ok := held("KV_m_drafts"); !ok || !get("r2").Done.IsZero() {
t.Fatal("dropped before the operator chose")
}
mu.Lock()
warrant = &oc.Warrant{Ask: a.ID, Asker: keeper.Module, Outcome: occ.OutcomeChosen, By: &oc.Person{Who: occ.Operator},
// The operator chooses: the router records the warrant and says decided; the keeper acts on decided, on its
// durable consumer, with no timer between and no call of its pass by the test (seams F1).
w := &oc.Warrant{Ask: a.ID, Asker: keeper.Module, Outcome: occ.OutcomeChosen, By: &oc.Person{Who: occ.Operator},
Option: "go", Label: "Remove", Level: occ.Destroy, At: occ.FormatTime(time.Now()), AskDigest: occ.Digest(a)}
mu.Unlock()
k.ActOnRequests()
record(w)
decided := time.Now()
body, _ := json.Marshal(w)
if _, err := js.Publish(ctx, DecidedSubject, body); err != nil {
t.Fatal(err)
}
eventually(t, "dropped m drafts", saidHas("dropped m drafts"))
if since := time.Since(decided); since > 5*time.Second {
t.Errorf("dropped %s after decided: within seconds", since)
}
if _, ok := held("KV_m_drafts"); ok {
t.Error("KV_m_drafts is still on the bus")
}
@@ -348,6 +392,18 @@ func TestTheBusModuleRunsWhole(t *testing.T) {
if !strings.Contains(recorded, "nats_drop: dropped drafts of m") {
t.Errorf("recorded %s: the drop as a hand act", recorded)
}
// The same warrant handed in a second request is refused and deletes nothing (security F2): drafts asked for and
// orphaned again, then a request naming the spent ask.
accepted("m", 9, state, entry("log", "stream", "log", map[string]string{"max-mib": "4"}))
eventually(t, "m drafts made again", func() bool { _, ok := held("KV_m_drafts"); return ok })
accepted("m", 10, entry("log", "stream", "log", map[string]string{"max-mib": "4"}))
eventually(t, "m drafts orphaned again", func() bool { return k.Held()["m"].Made["drafts"].Orphaned })
again := keeper.Request{ID: "r2b", Verb: keeper.VerbDrop, Module: "m", Local: "drafts", Caller: "person.jo", Why: "again", Ask: a.ID, At: time.Now()}
put(again)
eventually(t, "the second request on the spent warrant refused", func() bool { return strings.Contains(get("r2b").Error, "already acted") })
if _, ok := held("KV_m_drafts"); !ok {
t.Error("the spent warrant deleted KV_m_drafts a second time")
}
})
t.Run("nats_first_view_again asks at once and records the hand act", func(t *testing.T) {
raw, _ := json.Marshal(keeper.Request{ID: "r3", Verb: keeper.VerbFirstViewAgain, Why: "a verdict thought missed", Caller: "person.jo"})
modules/nats/cmd/nats-tools/requests.go
+172 -18
View File
@@ -5,12 +5,16 @@ import (
"encoding/hex"
"encoding/json"
"fmt"
"slices"
"sort"
"strings"
"sync"
"time"
occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient"
oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/handacts"
"nats-tools/internal/keeper"
)
@@ -23,16 +27,31 @@ import (
// requestStore is the keeper's state requests, as this bundle reaches it.
type requestStore struct {
Put func(key string, r keeper.Request) error
Get func(key string) (*keeper.Request, error)
Ask func(a oc.Ask) error
Now func() time.Time
Wait func(d time.Duration)
Put func(key string, r keeper.Request) error
Get func(key string) (*keeper.Request, error)
Keys func() ([]string, error)
Delete func(key string) error
// Raw reads one key's value as it is (an acted key is no request).
Raw func(key string) ([]byte, error)
Ask func(a oc.Ask) error
Now func() time.Time
// Ended answers the request's end once the keeper wrote it, or nil once within passed: heard on the state's watch,
// never read again on a timer.
Ended func(id string, within time.Duration) *keeper.Request
// Record records a hand act through the seat mesh-hand-acts.
Record Recorded
}
// RequestsInState is the store through the runtime that launched this bundle.
// RequestsInState is the store through the runtime that launched this bundle: one watch of the state requests for the
// bundle's life (the runtime's watches last as long as the process), handing each end to the call waiting for it.
func RequestsInState() requestStore {
s := stdio.State(keeper.StateRequests)
ends := newEnds(func(handle func(key string, value []byte)) error {
return s.Watch("", func(c stdio.StateChange) error {
handle(c.Key, c.Value)
return nil
})
})
return requestStore{
Put: func(key string, r keeper.Request) error { _, err := s.Put(key, r); return err },
Get: func(key string) (*keeper.Request, error) {
@@ -43,9 +62,72 @@ func RequestsInState() requestStore {
var r keeper.Request
return &r, json.Unmarshal(e.Value, &r)
},
Ask: func(a oc.Ask) error { return stdio.SeatPublish(oc.AcceptSubjectAsk(keeper.Module), a, a.ID) },
Now: func() time.Time { return time.Now().UTC() },
Wait: time.Sleep,
Keys: s.Keys,
Delete: s.Delete,
Raw: func(key string) ([]byte, error) {
e, err := s.Get(key)
if err != nil || e == nil {
return nil, err
}
return e.Value, nil
},
Ask: func(a oc.Ask) error { return stdio.SeatPublish(oc.AcceptSubjectAsk(keeper.Module), a, a.ID) },
Now: func() time.Time { return time.Now().UTC() },
Ended: ends.wait,
Record: RecordedBySeat,
}
}
// ends hands a request's end to the call waiting for it, from one watch of the state started at the first wait.
type ends struct {
watch func(handle func(key string, value []byte)) error
mu sync.Mutex
started bool
waiting map[string][]chan *keeper.Request
}
func newEnds(watch func(handle func(key string, value []byte)) error) *ends {
return &ends{watch: watch, waiting: map[string][]chan *keeper.Request{}}
}
func (e *ends) heard(key string, value []byte) {
var r keeper.Request
if json.Unmarshal(value, &r) != nil || r.Done.IsZero() {
return
}
e.mu.Lock()
chans := e.waiting[key]
delete(e.waiting, key)
e.mu.Unlock()
for _, c := range chans {
c <- &r
}
}
// wait waits for id's end at most within: nil when it did not come, or the watch could not be taken.
func (e *ends) wait(id string, within time.Duration) *keeper.Request {
c := make(chan *keeper.Request, 1)
e.mu.Lock()
e.waiting[id] = append(e.waiting[id], c)
start := !e.started
e.started = true
e.mu.Unlock()
if start {
if err := e.watch(e.heard); err != nil {
e.mu.Lock()
e.started = false
e.mu.Unlock()
return nil
}
}
select {
case r := <-c:
return r
case <-time.After(within):
e.mu.Lock()
e.waiting[id] = slices.DeleteFunc(e.waiting[id], func(w chan *keeper.Request) bool { return w == c })
e.mu.Unlock()
return nil
}
}
@@ -63,7 +145,7 @@ func askFor(r keeper.Request, now time.Time) (oc.Ask, error) {
}
what := map[string]string{keeper.VerbDrop: "remove", keeper.VerbRecreate: "rebuild"}[r.Verb]
name := keeper.BusName(r.Module, r.Local)
return oc.Ask{ID: "bus-" + r.ID, Who: occ.Operator, Expires: occ.FormatTime(now.Add(24 * time.Hour)),
return oc.Ask{ID: "bus-" + r.ID, Who: occ.Operator, Expires: occ.FormatTime(now.Add(keeper.AskLasts)),
OnExpiry: "nothing is removed and the entry stays as it is",
About: keeper.ConditionKey(map[string]string{keeper.VerbDrop: "orphaned", keeper.VerbRecreate: "drifted"}[r.Verb], r.Module, r.Local),
Headline: fmt.Sprintf("%s's kept data %s: %s it?", r.Module, r.Local, what),
@@ -100,12 +182,14 @@ func requestTools(rs requestStore) []stdio.Tool {
return nil, err
}
r.Ask = ask.ID
if err := rs.Put(r.ID, r); err != nil {
return nil, fmt.Errorf("the request cannot be handed to the keeper: %w", err)
}
// The ask first: a request is never handed for an ask that was not raised (an ask with no request acts on
// nothing, since the keeper acts only on a request).
if err := rs.Ask(ask); err != nil {
return nil, fmt.Errorf("the operator's ask cannot be raised, so nothing will be removed: %w", err)
}
if err := rs.Put(r.ID, r); err != nil {
return nil, fmt.Errorf("the request cannot be handed to the keeper, so nothing will be removed whatever the operator answers: %w", err)
}
return map[string]any{"request": r.ID, "ask": ask.ID, "state": "the operator is asked; the keeper acts once they approve, after a snapshot",
"read": fmt.Sprintf("nats_%s with request %s", verb, r.ID)}, nil
}
@@ -137,6 +221,7 @@ func requestTools(rs requestStore) []stdio.Tool {
return awaitEnd(rs, r, "nats_first_view_again"), nil
}}
return []stdio.Tool{
requestsTool(rs),
firstView,
tool(keeper.VerbCreate, "Create one entry of a module's kept data on the bus now, as its accepted manifest asks for it (design 25 § 12.3): handed to the bus module's keeper, which alone may create, and recorded as a hand act."),
tool(keeper.VerbApply, "Apply what a module's manifest asks of one entry where the bus holds less (a larger cap, a longer age): only a difference that grows; handed to the keeper and recorded as a hand act."),
@@ -145,14 +230,83 @@ func requestTools(rs requestStore) []stdio.Tool {
}
}
// awaitEnd reads a request's end back: the keeper acts on its watch of requests at once, so the end is read for at most
// half a minute, and otherwise how to read it later is answered.
// awaitEnd waits for a request's end, which the keeper writes as soon as its watch of requests hands it the request;
// past half a minute it answers how to read the end later.
func awaitEnd(rs requestStore, r keeper.Request, tool string) any {
for wait := 100 * time.Millisecond; wait < 20*time.Second; wait *= 2 {
rs.Wait(wait)
if got, err := rs.Get(r.ID); err == nil && got != nil && !got.Done.IsZero() {
if rs.Ended != nil {
if got := rs.Ended(r.ID, 30*time.Second); got != nil {
return got
}
}
if got, err := rs.Get(r.ID); err == nil && got != nil && !got.Done.IsZero() {
return got
}
return map[string]any{"request": r.ID, "state": "handed to the keeper, not ended yet", "read": fmt.Sprintf("%s with request %s", tool, r.ID)}
}
// requestsTool reads the keeper's state requests (design 25 § 12.3; review round 1, tools F3): every request with what
// waits, what ended and why, and each warrant spent; and drops one request not ended yet by hand, recorded as a hand act,
// so a request nobody will answer leaves by a person's act and never only by its key ageing out.
func requestsTool(rs requestStore) stdio.Tool {
return stdio.Tool{Name: "nats_requests",
Description: "The persons' requests the bus module's tools handed its keeper (nats_create, nats_apply, nats_recreate, nats_drop, nats_first_view_again): each one waiting (and for what: the operator's warrant on its ask, or the keeper's first view) or ended, with its outcome or why not, and each warrant spent (acted.<ask id>, a warrant acts once). With drop and why, removes one request not ended yet, recorded as a hand act; the operator's ask it raised then acts on nothing.",
Input: map[string]any{"drop": str("a request not ended yet, to remove by hand"), "why": str("why, in plain words: kept with the act")},
Run: func(a map[string]any) (any, error) {
if id := strings.TrimSpace(arg(a, "drop")); id != "" {
why := strings.TrimSpace(arg(a, "why"))
if why == "" {
return nil, fmt.Errorf("nats_requests drop says why")
}
r, err := rs.Get(id)
if err != nil || r == nil || strings.HasPrefix(id, keeper.ActedPrefix) {
return nil, fmt.Errorf("no request %s is held (%v)", id, err)
}
if !r.Done.IsZero() {
return nil, fmt.Errorf("the request %s ended at %s (%s): only a request that waits is dropped", id, r.Done.Format(time.RFC3339), r.Outcome)
}
if err := rs.Record(handacts.Request{What: fmt.Sprintf("dropped the waiting request %s (%s of %s's %s, ask %s)", id, r.Verb, r.Module, r.Local, r.Ask),
Why: why, Cause: "nats_requests"}); err != nil {
return nil, fmt.Errorf("the hand act cannot be recorded, so the request stays: %w", err)
}
if err := rs.Delete(id); err != nil {
return nil, err
}
return map[string]any{"dropped": id, "ask": r.Ask, "state": "the keeper acts on it no more; the operator's ask it raised acts on nothing"}, nil
}
keys, err := rs.Keys()
if err != nil {
return nil, err
}
sort.Strings(keys)
var waiting, ended []keeper.Request
spent := []keeper.Acted{}
for _, key := range keys {
if strings.HasPrefix(key, keeper.ActedPrefix) {
raw, err := rs.Raw(key)
var acted keeper.Acted
if err == nil && json.Unmarshal(raw, &acted) == nil {
spent = append(spent, acted)
}
continue
}
r, err := rs.Get(key)
if err != nil || r == nil {
continue
}
if r.Done.IsZero() {
waiting = append(waiting, *r)
} else {
ended = append(ended, *r)
}
}
return map[string]any{"waiting": orEmpty(waiting), "ended": orEmpty(ended), "warrants-spent": spent,
"why-waiting": "a recreate or a drop waits for the operator's warrant on its ask, acted on once they answer, and ends unanswered when its ask ends; any request waits while the keeper holds no first view"}, nil
}}
}
func orEmpty(rs []keeper.Request) []keeper.Request {
if rs == nil {
return []keeper.Request{}
}
return rs
}
modules/nats/cmd/nats-tools/requests_test.go
+77 -3
View File
@@ -1,6 +1,8 @@
package main
import (
"errors"
"runtime"
"strings"
"testing"
"time"
@@ -8,6 +10,7 @@ import (
occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient"
oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/handacts"
"nats-tools/internal/keeper"
)
@@ -18,6 +21,7 @@ import (
func TestAPersonsRequestIsHandedToTheKeeperAsTheProvedCaller(t *testing.T) {
held := map[string]keeper.Request{}
var raised []oc.Ask
var acts []handacts.Request
rs := requestStore{
Put: func(key string, r keeper.Request) error {
if r.Verb == keeper.VerbCreate {
@@ -34,9 +38,18 @@ func TestAPersonsRequestIsHandedToTheKeeperAsTheProvedCaller(t *testing.T) {
}
return &r, nil
},
Ask: func(a oc.Ask) error { raised = append(raised, a); return nil },
Now: func() time.Time { return time.Date(2026, 11, 4, 12, 0, 0, 0, time.UTC) },
Wait: func(time.Duration) {},
Keys: func() ([]string, error) {
var out []string
for k := range held {
out = append(out, k)
}
return append(out, keeper.ActedKey("bus-old")), nil
},
Delete: func(key string) error { delete(held, key); return nil },
Raw: func(string) ([]byte, error) { return []byte(`{"ask":"bus-old","request":"r0","verb":"drop"}`), nil },
Ask: func(a oc.Ask) error { raised = append(raised, a); return nil },
Now: func() time.Time { return time.Date(2026, 11, 4, 12, 0, 0, 0, time.UTC) },
Record: func(q handacts.Request) error { acts = append(acts, q); return nil },
}
tools := map[string]stdio.Tool{}
for _, tl := range requestTools(rs) {
@@ -68,4 +81,65 @@ func TestAPersonsRequestIsHandedToTheKeeperAsTheProvedCaller(t *testing.T) {
if got, _ := tools["nats_drop"].Run(map[string]any{"request": r.ID}); got.(*keeper.Request).ID != r.ID {
t.Errorf("reading the request back answered %+v", got)
}
// Review round 1 (tools F3): nats_requests reads what waits, what ended and each warrant spent, and drops a waiting
// request by hand, recorded; an ended one is not dropped.
out, err = tools["nats_requests"].Run(map[string]any{})
if err != nil {
t.Fatal(err)
}
list := out.(map[string]any)
if w, e, spent := list["waiting"].([]keeper.Request), list["ended"].([]keeper.Request), list["warrants-spent"].([]keeper.Acted); len(w) != 1 ||
w[0].ID != r.ID || len(e) != 1 || len(spent) != 1 || spent[0].Ask != "bus-old" {
t.Errorf("nats_requests answered %+v: the drop waiting, the create ended, one warrant spent", list)
}
if _, err := tools["nats_requests"].Run(map[string]any{"drop": r.ID}); err == nil || !strings.Contains(err.Error(), "says why") {
t.Errorf("a drop by hand without why: %v", err)
}
for id, r := range held {
if r.Verb == keeper.VerbCreate {
if _, err := tools["nats_requests"].Run(map[string]any{"drop": id, "why": "x"}); err == nil || !strings.Contains(err.Error(), "only a request that waits") {
t.Errorf("dropping an ended request: %v", err)
}
}
}
if _, err := tools["nats_requests"].Run(map[string]any{"drop": r.ID, "why": "nobody will answer"}); err != nil || len(acts) != 1 ||
!strings.Contains(acts[0].What, "dropped the waiting request "+r.ID) {
t.Errorf("dropped by hand (%v), recorded %+v", err, acts)
}
if _, still := held[r.ID]; still {
t.Error("the dropped request is still held")
}
// The ask first: an ask that cannot be raised hands no request to the keeper.
before := len(held)
rs.Ask = func(oc.Ask) error { return errors.New("no responders") }
tools = map[string]stdio.Tool{}
for _, tl := range requestTools(rs) {
tools[tl.Name] = tl
}
if _, err := tools["nats_recreate"].Run(map[string]any{"module": "m", "local": "log", "why": "drifted"}); err == nil || len(held) != before {
t.Errorf("an ask not raised: %v, %d requests held (was %d)", err, len(held), before)
}
// A request's end is heard on the one watch of the state, handed to the call that waits for it.
var handle func(string, []byte)
watches := 0
e := newEnds(func(h func(string, []byte)) error { watches++; handle = h; return nil })
got := make(chan *keeper.Request, 1)
go func() { got <- e.wait("r1", time.Minute) }()
for {
e.mu.Lock()
ready := len(e.waiting["r1"]) == 1 && handle != nil
e.mu.Unlock()
if ready {
break
}
runtime.Gosched()
}
handle("r1", []byte(`{"id":"r1"}`))
handle("r1", []byte(`{"id":"r1","done":"2026-11-04T12:00:01Z","outcome":"created"}`))
if r := <-got; r == nil || r.Outcome != "created" || watches != 1 {
t.Errorf("the end heard %+v on %d watch(es): created, on one", r, watches)
}
if r := e.wait("r2", time.Millisecond); r != nil || watches != 1 {
t.Errorf("an end not written: %+v on %d watch(es)", r, watches)
}
}
modules/nats/internal/keeper/fill.go
+2
View File
@@ -11,6 +11,8 @@ import (
// FillPass reads the fill of every entry that refuses when full (ADR 0595 §6), raising `full` at the raise mark and
// clearing it only below the clear mark, so an entry at the mark does not flap. It is the keeper's one periodic read.
func (k *Keeper) FillPass() {
// What the bus did not take before is said again first (design 25 § 12.2).
k.SayUnsaid()
raise, clear := k.marks()
keys, err := k.Store.Keys(StateAsked)
if err != nil {
modules/nats/internal/keeper/health.go
+17 -2
View File
@@ -37,16 +37,31 @@ func HealthOf(s Status, held map[string]Asked, pending int, now time.Time) Healt
why = append(why, fmt.Sprintf("its last fill pass is %s old, more than %s", now.Sub(s.Pass).Round(time.Second), Bound))
}
}
if s.FirstView.Answered.IsZero() {
// Its first view unanswered past the bound since its start, not at once: the indexer answers within it.
if s.FirstView.Answered.IsZero() && (s.Started.IsZero() || now.Sub(s.Started) > Bound) {
w := fmt.Sprintf("its first view, %s.%s, has not answered", s.FirstView.Owner, s.FirstView.Verb)
if s.FirstView.Error != "" {
w += ": " + s.FirstView.Error
}
why = append(why, w)
}
if pending > 0 && (s.Heard.IsZero() || now.Sub(s.Heard) > Bound) {
switch {
case pending < 0:
// Never skipped silently: what waits on its consumer is not known, so nothing says it is answered.
why = append(why, "what waits on its consumer cannot be read from the bus's monitoring")
case pending > 0 && (s.Heard.IsZero() || now.Sub(s.Heard) > Bound):
why = append(why, fmt.Sprintf("%d accepted verdict(s) wait on its consumer unanswered for more than %s", pending, Bound))
}
if n := len(s.Unsaid) + len(s.UnsaidConditions); n > 0 {
var what []string
for _, u := range s.Unsaid {
what = append(what, fmt.Sprintf("%s of %s %s (%s)", u.Event, u.Body.Module, u.Body.Local, u.Error))
}
for _, key := range sortedKeys(s.UnsaidConditions) {
what = append(what, "the condition "+key+" ("+s.UnsaidConditions[key].State+")")
}
why = append(why, fmt.Sprintf("%d word(s) the bus did not take wait to be said again at the next fill pass: %s", n, strings.Join(what, "; ")))
}
for _, module := range sortedKeys(held) {
for _, local := range sortedKeys(held[module].Made) {
m := held[module].Made[local]
modules/nats/internal/keeper/keeper.go
+108 -11
View File
@@ -2,6 +2,8 @@ package keeper
import (
"context"
"crypto/rand"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
@@ -130,6 +132,23 @@ type Status struct {
// Heard is when the last manifest-accepted was heard, and Unanswered since when one has waited unhandled.
Heard time.Time `json:"heard,omitempty"`
Unanswered time.Time `json:"unanswered,omitempty"`
// Unsaid are the events the bus did not take, kept with their id and the subjects still to say them on, and
// UnsaidConditions the conditions the condition keeper did not take, by key: each said again at every fill pass, and
// named by the keeper's health until it is (design 25 § 12.2: a failed say is never only logged).
Unsaid []Saying `json:"unsaid,omitempty"`
UnsaidConditions map[string]oc.Condition `json:"unsaid-conditions,omitempty"`
}
// Saying is one event of the seat mesh-broker as the keeper says it: its name, its one event id (the same on every
// subject and every try, so the bus keeps it once), its body, and Left, the subjects it is still to be said on (nil:
// every one). Emit says it on Left, and leaves in Left what the bus did not take.
type Saying struct {
Event string `json:"event"`
ID string `json:"id"`
Body Event `json:"body"`
Left []string `json:"left,omitempty"`
Error string `json:"error,omitempty"`
At time.Time `json:"at"`
}
// FirstViewStatus is what the keeper holds of its first view.
@@ -177,14 +196,19 @@ type Store interface {
Put(state, key string, value []byte) (uint64, error)
Delete(state, key string) error
Keys(state string) ([]string, error)
// Create writes a key only when it has none: ErrExists when it has one (a compare-and-set at revision 0).
Create(state, key string, value []byte) (uint64, error)
}
// ErrExists is a Create of a key that has a value.
var ErrExists = errors.New("the key has a value already")
// Keeper is the bus module's keeper.
type Keeper struct {
Bus Bus
Store Store
// Emit says one of the bus module's events; Say raises or clears a condition through the condition keeper.
Emit func(event string, body Event) error
// Emit says one of the bus module's events (Saying); Say raises or clears a condition through the condition keeper.
Emit func(s *Saying) error
Say func(oc.Condition) error
// View asks the indexer's indexer_bus_state.
View func() (FirstView, error)
@@ -423,8 +447,13 @@ func (k *Keeper) ensure(held *Asked, m *Made) {
k.clear(conditionRefused, held.Module, m.Plan.Local)
}
if !m.Created {
// Marked created only once `created` is said whole: the grants part binds on it, so one the bus did not take is
// said again on the backoff (Waiting), never marked made and forgotten.
if err := k.sayNow(EventCreated, m.Plan, held.Revision, "", asked, holds); err != nil {
m.Waiting = fmt.Sprintf("its created could not be said yet: %v", err)
return
}
m.Created = true
k.emit(EventCreated, m.Plan, held.Revision, "", asked, holds)
}
drift := strings.Join(drifts, "; ")
switch {
@@ -535,9 +564,7 @@ func (k *Keeper) raise(kind, module, local, severity, headline, needs, explanati
Summary: explanation,
Explanation: occ.NeedsYou + needs + ". " + explanation,
Resolved: "cleared by the bus module once it no longer holds"}
if err := k.Say(c); err != nil {
k.logf("the condition %s cannot be raised: %v", c.Key, err)
}
k.sayCondition(c)
}
func (k *Keeper) clear(kind, module, local string) {
@@ -545,15 +572,80 @@ func (k *Keeper) clear(kind, module, local string) {
return
}
c := oc.Condition{Key: ConditionKey(kind, module, local), Subject: oc.ConditionSubject{Scope: "module", ID: module}, State: occ.StateCleared}
if err := k.Say(c); err != nil {
k.logf("the condition %s cannot be cleared: %v", c.Key, err)
k.sayCondition(c)
}
// sayCondition says c; one the condition keeper did not take is kept in the keeper's status, replacing an older word on
// its key, and said again at the next fill pass.
func (k *Keeper) sayCondition(c oc.Condition) {
err := k.Say(c)
k.setStatus(func(s *Status) {
if err == nil {
delete(s.UnsaidConditions, c.Key)
return
}
if s.UnsaidConditions == nil {
s.UnsaidConditions = map[string]oc.Condition{}
}
s.UnsaidConditions[c.Key] = c
})
if err != nil {
k.logf("the condition %s (%s) could not be said, said again at the next fill pass: %v", c.Key, c.State, err)
}
}
// SayUnsaid says again every event and condition the bus did not take (the fill pass calls it): an event on the
// subjects it is still to be said on, with its one id.
func (k *Keeper) SayUnsaid() {
s := k.Status()
var still []Saying
for _, u := range s.Unsaid {
u := u
if err := k.Emit(&u); err != nil {
u.Error = err.Error()
still = append(still, u)
k.logf("the event %s of %s %s (%s) could not be said again: %v", u.Event, u.Body.Module, u.Body.Local, u.ID, err)
}
}
for _, key := range sortedKeys(s.UnsaidConditions) {
k.sayCondition(s.UnsaidConditions[key])
}
k.setStatus(func(st *Status) {
// Kept: what was not said now, and what was added meanwhile.
added := st.Unsaid[min(len(s.Unsaid), len(st.Unsaid)):]
st.Unsaid = append(still, added...)
})
}
// emit says one event; one the bus did not take whole is kept in the keeper's status and said again, on the subjects
// left, at the next fill pass.
func (k *Keeper) emit(event string, p Plan, revision uint64, why string, asked, held map[string]Settings) {
if k.Emit == nil {
return
}
s := k.saying(event, p, revision, why, asked, held)
if err := k.Emit(&s); err != nil {
s.Error = err.Error()
k.setStatus(func(st *Status) { st.Unsaid = append(st.Unsaid, s) })
k.logf("the event %s of %s %s could not be said, said again at the next fill pass: %v", event, p.Module, p.Local, err)
}
}
// sayNow says one event and answers whether the bus took it whole; nothing is kept to say again (its caller says it
// again itself).
func (k *Keeper) sayNow(event string, p Plan, revision uint64, why string, asked, held map[string]Settings) error {
if k.Emit == nil {
return nil
}
s := k.saying(event, p, revision, why, asked, held)
if err := k.Emit(&s); err != nil {
k.logf("the event %s of %s %s could not be said: %v", event, p.Module, p.Local, err)
return err
}
return nil
}
func (k *Keeper) saying(event string, p Plan, revision uint64, why string, asked, held map[string]Settings) Saying {
body := Event{V: 1, Module: p.Module, Local: p.Local, Storage: p.Storage, Template: p.Template, Options: p.Entry.Options,
Configuration: p.Entry.Configuration, BusName: BusName(p.Module, p.Local), Binding: p.Binding,
Asked: asked, Held: held, Why: why, Revision: revision}
@@ -561,9 +653,14 @@ func (k *Keeper) emit(event string, p Plan, revision uint64, why string, asked,
// A dropped entry's names only: the binding to remove.
body.Binding.Types = nil
}
if err := k.Emit(event, body); err != nil {
k.logf("the event %s of %s %s could not be said: %v", event, p.Module, p.Local, err)
}
return Saying{Event: event, ID: EventID(), Body: body, At: k.now()}
}
// EventID is a new event id: sixteen random bytes in hex.
func EventID() string {
var b [16]byte
_, _ = rand.Read(b[:])
return hex.EncodeToString(b[:])
}
func (k *Keeper) asked(module string) (Asked, uint64, bool) {
modules/nats/internal/keeper/keeper_test.go
+130 -5
View File
@@ -101,6 +101,17 @@ func (s *memStore) Put(state, key string, value []byte) (uint64, error) {
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()
@@ -126,20 +137,32 @@ 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.
down bool
}
func (s *said) emit(event string, body Event) error {
func (s *said) emit(e *Saying) error {
s.mu.Lock()
defer s.mu.Unlock()
s.events = append(s.events, event+" "+body.Module+" "+body.Local)
s.bodies = append(s.bodies, body)
if s.down {
e.Left = []string{"seat", "history"}
return errors.New("no responders")
}
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
}
@@ -451,8 +474,17 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) {
if next, err := k.AskFirstView(10 * time.Second); err == nil || next != 20*time.Second {
t.Fatalf("asked again in %s (%v): on the backoff", next, err)
}
if h := HealthOf(k.Status(), nil, 0, at); h.Healthy || !strings.Contains(h.Why, "mesh-delivery-indexer.indexer_bus_state") {
t.Errorf("health %+v: unhealthy, naming the verb", h)
// 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-delivery-indexer.indexer_bus_state") {
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
@@ -541,6 +573,99 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) {
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) != 1 {
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) != 1 {
t.Errorf("the second request on the same warrant: %+v, deleted %v", got, bus.deleted)
}
if raw, _, found, _ := store.Get(StateRequests, ActedKey("bus-5")); !found || !strings.Contains(string(raw), `"request":"r5"`) {
t.Errorf("the acted key is %s (%v)", raw, found)
}
}},
// 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(-time.Hour)})
_, _ = 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"])
}
}},
// 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)
modules/nats/internal/keeper/requests.go
+77
View File
@@ -44,6 +44,32 @@ type Request struct {
Outcome string `json:"outcome,omitempty"`
Error string `json:"error,omitempty"`
Snapshot string `json:"snapshot,omitempty"`
// Waiting is why it waits, while it waits: the operator's warrant, or the keeper's first view.
Waiting string `json:"waiting,omitempty"`
// Unrecorded is why the hand act of what it did could not be recorded, when it could not: said, never only logged.
Unrecorded string `json:"unrecorded,omitempty"`
}
// AskLasts is how long the operator's ask for a recreate or a drop waits for an answer: shorter than the day the state
// requests keeps a key, so a request whose ask ended unanswered is ended by the keeper, saying so, before its key ages
// out, and the warrant's acted key (ActedKey) outlives the ask it spends.
const AskLasts = 12 * time.Hour
// ActedPrefix starts the key in StateRequests that spends one warrant, `acted.<ask id>`, claimed by compare-and-set
// before a recreate or a drop deletes anything (design 25 § 12.3: a warrant acts once). A request id holds no dot.
const ActedPrefix = "acted."
// ActedKey is the key spending the warrant on ask.
func ActedKey(ask string) string { return ActedPrefix + ask }
// Acted is the value of an acted key: which request spent the warrant, on what.
type Acted struct {
Ask string `json:"ask"`
Request string `json:"request"`
Verb string `json:"verb"`
Module string `json:"module"`
Local string `json:"local"`
At time.Time `json:"at"`
}
// Act is what an option of the operator's ask binds for a recreate or a drop: the verb and the entry (occ.ActDigest).
@@ -62,6 +88,9 @@ func (k *Keeper) ActOnRequests() {
return
}
for _, key := range keys {
if strings.HasPrefix(key, ActedPrefix) {
continue
}
raw, _, found, err := k.Store.Get(StateRequests, key)
if err != nil || !found {
continue
@@ -72,8 +101,17 @@ func (k *Keeper) ActOnRequests() {
}
outcome, err := k.act(&r)
if errors.Is(err, ErrNoWarrant) {
// Said on the request while it waits, once (its write wakes the watch, which then finds the same words).
if r.Waiting != err.Error() {
r.Waiting = err.Error()
data, _ := json.Marshal(r)
if _, err := k.Store.Put(StateRequests, key, data); err != nil {
k.logf("why the request %s waits cannot be written: %v", key, err)
}
}
continue
}
r.Waiting = ""
r.Done, r.Outcome = k.now(), outcome
if err != nil {
r.Error = err.Error()
@@ -151,6 +189,10 @@ func (k *Keeper) act(r *Request) (string, error) {
return "nothing done", fmt.Errorf("the snapshot before %s failed, so nothing was removed: %w", r.Verb, err)
}
r.Snapshot = path
// The warrant is spent before anything is deleted: of two requests naming one ask, only the one whose claim stands acts.
if err := k.spend(r); err != nil {
return "nothing done", err
}
for _, o := range m.Plan.Objects {
if err := k.Bus.Delete(ctx, o); err != nil {
return "partly done", fmt.Errorf("%s could not be removed after its snapshot at %s: %w", o.Stream(), path, err)
@@ -194,6 +236,21 @@ func (k *Keeper) warranted(r *Request) error {
if err != nil {
return fmt.Errorf("%w: the router's record of the ask %s cannot be read now: %v", ErrNoWarrant, r.Ask, err)
}
// An ask that ended unanswered ends its request, said: never a request that waits until its key ages out. And a
// warrant is not acted on once its ask ended, so a request handed after its acted key aged out acts on nothing.
var expires time.Time
if !r.At.IsZero() {
expires = r.At.Add(AskLasts)
}
if ask != nil {
if t, err := occ.Time(ask.Expires); err == nil && !t.IsZero() {
expires = t
}
}
if !expires.IsZero() && k.now().After(expires) {
return fmt.Errorf("the operator's ask %s ended at %s with no warrant acted on, so nothing is removed: ask again with a new request",
r.Ask, expires.UTC().Format(time.RFC3339))
}
if ask == nil || w == nil {
return ErrNoWarrant
}
@@ -207,11 +264,31 @@ func (k *Keeper) warranted(r *Request) error {
return occ.Performs(o, r.Act())
}
// spend claims the warrant of r's ask by compare-and-set; a warrant already spent is refused, naming what it did.
func (k *Keeper) spend(r *Request) error {
raw, _ := json.Marshal(Acted{Ask: r.Ask, Request: r.ID, Verb: r.Verb, Module: r.Module, Local: r.Local, At: k.now()})
_, err := k.Store.Create(StateRequests, ActedKey(r.Ask), raw)
if err == nil {
return nil
}
if !errors.Is(err, ErrExists) {
return fmt.Errorf("the warrant of the ask %s cannot be spent now, so nothing was removed: %w", r.Ask, err)
}
var a Acted
if got, _, found, _ := k.Store.Get(StateRequests, ActedKey(r.Ask)); found {
_ = json.Unmarshal(got, &a)
}
return fmt.Errorf("the operator's warrant on the ask %s already acted (%s of %s's %s, request %s, at %s): a warrant acts once, so nothing was removed",
r.Ask, a.Verb, a.Module, a.Local, a.Request, a.At.Format(time.RFC3339))
}
// record records what a request did as a hand act; one that cannot be recorded is said on the request's end.
func (k *Keeper) record(r *Request, what string) {
if k.Act == nil {
return
}
if err := k.Act(what+", asked by "+r.Caller, r.Why, "nats_"+strings.ReplaceAll(r.Verb, "-", "_")); err != nil {
r.Unrecorded = err.Error()
k.logf("the hand act %q could not be recorded: %v", what, err)
}
}
modules/nats/module.json
+2 -1
View File
@@ -121,7 +121,8 @@
"nats_recreate",
"nats_drop",
"nats_keeper_health",
"nats_first_view_again"
"nats_first_view_again",
"nats_requests"
],
"guards": [
8222