diff --git a/modules/nats/cmd/nats-tools/keep.go b/modules/nats/cmd/nats-tools/keep.go index c9bd2e39c..cc2966f9f 100644 --- a/modules/nats/cmd/nats-tools/keep.go +++ b/modules/nats/cmd/nats-tools/keep.go @@ -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.`, which the // shared events stream keeps 7 days and every consumer of `mesh-broker.` 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: diff --git a/modules/nats/cmd/nats-tools/keeper_integration_test.go b/modules/nats/cmd/nats-tools/keeper_integration_test.go index b60cde404..55648353e 100644 --- a/modules/nats/cmd/nats-tools/keeper_integration_test.go +++ b/modules/nats/cmd/nats-tools/keeper_integration_test.go @@ -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 `.`, 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"}) diff --git a/modules/nats/cmd/nats-tools/requests.go b/modules/nats/cmd/nats-tools/requests.go index 2f7084810..6470cf3b0 100644 --- a/modules/nats/cmd/nats-tools/requests.go +++ b/modules/nats/cmd/nats-tools/requests.go @@ -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., 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 +} diff --git a/modules/nats/cmd/nats-tools/requests_test.go b/modules/nats/cmd/nats-tools/requests_test.go index 5c4802173..557bdf001 100644 --- a/modules/nats/cmd/nats-tools/requests_test.go +++ b/modules/nats/cmd/nats-tools/requests_test.go @@ -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) + } } diff --git a/modules/nats/internal/keeper/fill.go b/modules/nats/internal/keeper/fill.go index 9a8ee9345..deee26f87 100644 --- a/modules/nats/internal/keeper/fill.go +++ b/modules/nats/internal/keeper/fill.go @@ -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 { diff --git a/modules/nats/internal/keeper/health.go b/modules/nats/internal/keeper/health.go index fb623b1cd..44fc3016e 100644 --- a/modules/nats/internal/keeper/health.go +++ b/modules/nats/internal/keeper/health.go @@ -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] diff --git a/modules/nats/internal/keeper/keeper.go b/modules/nats/internal/keeper/keeper.go index 1eb665c5a..1d52ca637 100644 --- a/modules/nats/internal/keeper/keeper.go +++ b/modules/nats/internal/keeper/keeper.go @@ -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) { diff --git a/modules/nats/internal/keeper/keeper_test.go b/modules/nats/internal/keeper/keeper_test.go index 001e3ba43..cef3c8436 100644 --- a/modules/nats/internal/keeper/keeper_test.go +++ b/modules/nats/internal/keeper/keeper_test.go @@ -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) diff --git a/modules/nats/internal/keeper/requests.go b/modules/nats/internal/keeper/requests.go index e27757a38..b388e7183 100644 --- a/modules/nats/internal/keeper/requests.go +++ b/modules/nats/internal/keeper/requests.go @@ -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.`, 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) } } diff --git a/modules/nats/module.json b/modules/nats/module.json index 5cc9edeab..02b1320c6 100644 --- a/modules/nats/module.json +++ b/modules/nats/module.json @@ -121,7 +121,8 @@ "nats_recreate", "nats_drop", "nats_keeper_health", - "nats_first_view_again" + "nats_first_view_again", + "nats_requests" ], "guards": [ 8222