diff --git a/modules/nats/cmd/nats-tools/keep.go b/modules/nats/cmd/nats-tools/keep.go index cc2966f9..36f0ce15 100644 --- a/modules/nats/cmd/nats-tools/keep.go +++ b/modules/nats/cmd/nats-tools/keep.go @@ -128,6 +128,7 @@ func streamEmitter(js jetstream.JetStream, a process.Account) func(*keeper.Sayin if subjects == nil { subjects = EventSubjects(a, s.Event) } + s.Said = nil for i, subject := range subjects { msg := nats.NewMsg(subject) msg.Header.Set("x-event-id", s.ID) @@ -143,6 +144,7 @@ func streamEmitter(js jetstream.JetStream, a process.Account) func(*keeper.Sayin s.Left = append([]string(nil), subjects[i:]...) return fmt.Errorf("saying %s on %s: %w", s.Event, subject, err) } + s.Said = append(s.Said, subject) } s.Left = nil return nil @@ -267,6 +269,9 @@ func hearer(js jetstream.JetStream, a process.Account, logf func(string, ...any) } } +// FillEvery is how often the keeper's fill pass runs: a minute (a test's shorter bound beside). +var FillEvery = time.Minute + // 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) @@ -278,7 +283,7 @@ var DecidedSubject = oc.SubjectDecided(keeper.Module) // 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) + tick := time.NewTicker(FillEvery) defer tick.Stop() k.FillPass() var wait time.Duration @@ -335,10 +340,18 @@ func Keep(ctx context.Context, k *keeper.Keeper, hear func(context.Context, func break } logf("manifest-accepted cannot be heard yet (%v): asked again in %s", err, backoff) - select { - case <-ctx.Done(): - return ctx.Err() - case <-time.After(backoff): + // The fill pass runs on while the consumer is bound again (review round 2, running F1). + bound := time.After(backoff) + binding: + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-tick.C: + k.FillPass() + case <-bound: + break binding + } } } if requests != nil { diff --git a/modules/nats/cmd/nats-tools/keeper_integration_test.go b/modules/nats/cmd/nats-tools/keeper_integration_test.go index 55648353..7dd1356d 100644 --- a/modules/nats/cmd/nats-tools/keeper_integration_test.go +++ b/modules/nats/cmd/nats-tools/keeper_integration_test.go @@ -153,7 +153,7 @@ func TestTheBusModuleRunsWhole(t *testing.T) { t.Fatal(err) } defer pconn.Close() - for _, st := range []string{keeper.StateAsked, keeper.StateRequests} { + for _, st := range []string{keeper.StateAsked, keeper.StateRequests, keeper.StateSpent} { if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "nats_" + st, History: 5}); err != nil { t.Fatal(err) } @@ -252,6 +252,22 @@ func TestTheBusModuleRunsWhole(t *testing.T) { stopDown := run(down) eventually(t, "a fill pass with no first view", func() bool { return !down.Status().Pass.IsZero() && down.Status().FirstView.Answered.IsZero() }) stopDown() + // And while its consumer of manifest-accepted is bound again on its backoff (review round 2, running F1). + was := FillEvery + FillEvery = 20 * time.Millisecond + defer func() { FillEvery = was }() + deaf := newKeeper() + kctx, stopDeaf := context.WithCancel(ctx) + done := make(chan struct{}) + go func() { + _ = Keep(kctx, deaf, func(context.Context, func(string, []byte) error) error { return errors.New("no consumer yet") }, nil, t.Logf) + close(done) + }() + eventually(t, "the first view answered", func() bool { return !deaf.Status().FirstView.Answered.IsZero() }) + first := deaf.Status().Pass + eventually(t, "a fill pass while the consumer is bound again", func() bool { return deaf.Status().Pass.After(first) }) + stopDeaf() + <-done }) k := newKeeper() stop := run(k) diff --git a/modules/nats/cmd/nats-tools/manifest_test.go b/modules/nats/cmd/nats-tools/manifest_test.go index 8a3245d2..632e74c6 100644 --- a/modules/nats/cmd/nats-tools/manifest_test.go +++ b/modules/nats/cmd/nats-tools/manifest_test.go @@ -174,7 +174,8 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) { MaxMiB int `json:"max-mib"` } `json:"streams"` State []struct { - Name string `json:"name"` + Name string `json:"name"` + ReadOnly bool `json:"read-only"` } `json:"state"` } if err := json.Unmarshal(raw, &nm); err != nil { @@ -203,12 +204,17 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) { t.Errorf("emits lists %s, which its stream carries: an event is named in one place (ADR 0363)", e) } } - var states []string + var states, readOnly []string for _, s := range nm.State { states = append(states, s.Name) + if s.ReadOnly { + readOnly = append(readOnly, s.Name) + } } - if !slices.Equal(states, []string{"asked", "requests"}) { - t.Errorf("state is %v: asked and requests, old sections (ADR 0595 §10)", states) + // spent, the warrants the keeper spent, is written by the keeper alone and only read by the tools (review round 2, + // security F2: the tool runner's account, which writes requests, may never delete a spend record). + if !slices.Equal(states, []string{"asked", "requests", "spent"}) || !slices.Equal(readOnly, []string{"spent"}) { + t.Errorf("state is %v, read-only %v: asked and requests, old sections (ADR 0595 §10), and spent, read-only", states, readOnly) } if _, ok := nm.OwnSecrets["broker-keeper"]; !ok { t.Error("own-secrets lack broker-keeper, the keeper's own account (ADR 0436 §1)") @@ -217,9 +223,9 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) { bus, _ := keeper["bus"].(map[string]any) health, _ := keeper["health"].(map[string]any) if keeper["type"] != "process" || keeper["artifact"] != "tools" || fmt.Sprint(keeper["run"]) != "[./nats-tools keep]" || - fmt.Sprint(bus["state"]) != "[asked requests]" || + fmt.Sprint(bus["state"]) != "[asked requests spent]" || health["kind"] != "tool" || health["tool"] != "nats_keeper_health" { - t.Errorf("the keeper is %v: a process of the tools bundle, run as keep, keeping asked and requests, judged by nats_keeper_health", keeper) + t.Errorf("the keeper is %v: a process of the tools bundle, run as keep, keeping asked, requests and spent, judged by nats_keeper_health", keeper) } // Its bus declares keeps, the one key more ADR 0626 §1 allows it, read through the SDK as the grant composition reads it. var whole catalogue.Manifest diff --git a/modules/nats/cmd/nats-tools/requests.go b/modules/nats/cmd/nats-tools/requests.go index 6470cf3b..37ddfeb1 100644 --- a/modules/nats/cmd/nats-tools/requests.go +++ b/modules/nats/cmd/nats-tools/requests.go @@ -31,10 +31,10 @@ type requestStore struct { 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 + // Spent reads the keeper's spend records, its state spent, which this bundle reads and never writes. + Spent func() ([]keeper.Acted, 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 @@ -64,12 +64,21 @@ func RequestsInState() requestStore { }, Keys: s.Keys, Delete: s.Delete, - Raw: func(key string) ([]byte, error) { - e, err := s.Get(key) - if err != nil || e == nil { + Spent: func() ([]keeper.Acted, error) { + spent := stdio.State(keeper.StateSpent) + keys, err := spent.Keys() + if err != nil { return nil, err } - return e.Value, nil + sort.Strings(keys) + out := []keeper.Acted{} + for _, key := range keys { + var a keeper.Acted + if e, err := spent.Get(key); err == nil && e != nil && json.Unmarshal(e.Value, &a) == nil { + out = append(out, a) + } + } + return out, 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() }, @@ -249,7 +258,7 @@ func awaitEnd(rs requestStore, r keeper.Request, tool string) any { // 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.", + 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 (the keeper's state spent, which only it writes: 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 != "" { @@ -279,14 +288,12 @@ func requestsTool(rs requestStore) stdio.Tool { } sort.Strings(keys) var waiting, ended []keeper.Request - spent := []keeper.Acted{} + spent, err := rs.Spent() + if err != nil { + return nil, fmt.Errorf("the warrants spent cannot be read: %w", err) + } 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) diff --git a/modules/nats/cmd/nats-tools/requests_test.go b/modules/nats/cmd/nats-tools/requests_test.go index 557bdf00..80771a22 100644 --- a/modules/nats/cmd/nats-tools/requests_test.go +++ b/modules/nats/cmd/nats-tools/requests_test.go @@ -43,10 +43,12 @@ func TestAPersonsRequestIsHandedToTheKeeperAsTheProvedCaller(t *testing.T) { for k := range held { out = append(out, k) } - return append(out, keeper.ActedKey("bus-old")), nil + return out, 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 }, + Spent: func() ([]keeper.Acted, error) { + return []keeper.Acted{{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 }, diff --git a/modules/nats/internal/keeper/health.go b/modules/nats/internal/keeper/health.go index 44fc3016..aa96d985 100644 --- a/modules/nats/internal/keeper/health.go +++ b/modules/nats/internal/keeper/health.go @@ -34,7 +34,13 @@ func HealthOf(s Status, held map[string]Asked, pending int, now time.Time) Healt case !s.Pass.IsZero(): h.Pass = s.Pass.Format(time.RFC3339) if now.Sub(s.Pass) > Bound { - why = append(why, fmt.Sprintf("its last fill pass is %s old, more than %s", now.Sub(s.Pass).Round(time.Second), Bound)) + w := fmt.Sprintf("its last fill pass is %s old, more than %s", now.Sub(s.Pass).Round(time.Second), Bound) + if len(s.Unsaid)+len(s.UnsaidConditions) > 0 { + // Its status may be what is not written: what it held unsaid then may be lost at a restart. + w += fmt.Sprintf(", so its status may no longer be written: the %d word(s) it held unsaid then, and any since, are kept only in its memory and may be lost at a restart", + len(s.Unsaid)+len(s.UnsaidConditions)) + } + why = append(why, w) } } // Its first view unanswered past the bound since its start, not at once: the indexer answers within it. diff --git a/modules/nats/internal/keeper/keeper.go b/modules/nats/internal/keeper/keeper.go index 1d52ca63..3908dfff 100644 --- a/modules/nats/internal/keeper/keeper.go +++ b/modules/nats/internal/keeper/keeper.go @@ -143,10 +143,12 @@ type Status struct { // 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"` + Event string `json:"event"` + ID string `json:"id"` + Body Event `json:"body"` + Left []string `json:"left,omitempty"` + // Said are the subjects this try said it on, in order: the seat's subject first. + Said []string `json:"said,omitempty"` Error string `json:"error,omitempty"` At time.Time `json:"at"` } @@ -223,6 +225,8 @@ type Keeper struct { mu sync.Mutex status Status viewed bool + // unwritten says the condition that the status cannot be written, with words unsaid, is raised. + unwritten bool } func (k *Keeper) now() time.Time { @@ -274,9 +278,46 @@ func (k *Keeper) setStatus(f func(*Status)) { k.mu.Unlock() if k.Store != nil { raw, _ := json.Marshal(s) - if _, err := k.Store.Put(StateAsked, KeyStatus, raw); err != nil { + _, err := k.Store.Put(StateAsked, KeyStatus, raw) + if err != nil { k.logf("the keeper's status cannot be written to its own %s: %v", StateAsked, err) } + k.statusWritten(s, err) + } +} + +// statusWritten raises, once, the keeper's own condition when its status cannot be written while it holds words the bus +// did not take: those are then kept only in memory, and a restart loses them (review round 2, seams note 3 and health +// note 4: named, not only a log). The next write that stands clears it. +func (k *Keeper) statusWritten(s Status, err error) { + if k.Say == nil { + return + } + n := len(s.Unsaid) + len(s.UnsaidConditions) + k.mu.Lock() + raised := k.unwritten + switch { + case err != nil && n > 0 && !raised: + k.unwritten = true + case err == nil && raised: + k.unwritten = false + default: + k.mu.Unlock() + return + } + k.mu.Unlock() + c := oc.Condition{Key: ConditionKey(conditionUnwritten, Module, "keeper"), Subject: oc.ConditionSubject{Scope: "module", ID: Module}} + if err != nil { + needs := "a look at why the keeper cannot write its own state " + StateAsked + c.State, c.Severity, c.Headline, c.Needs = occ.StateRaised, occ.SeverityWarning, "The bus module may lose words it could not say", needs + c.Summary = fmt.Sprintf("the keeper holds %d word(s) the bus did not take and cannot write its status (%v): they are kept only in memory, and a restart of the keeper loses them", n, err) + c.Explanation = occ.NeedsYou + needs + ". " + c.Summary + c.Resolved = "cleared by the bus module once its status is written again" + } else { + c.State, c.Resolved = occ.StateCleared, "the keeper's status is written again, with every word it still has to say" + } + if e := k.Say(c); e != nil { + k.logf("the condition %s (%s) could not be said: %v", c.Key, c.State, e) } } @@ -548,6 +589,8 @@ const ( conditionDrifted = "drifted" conditionOrphaned = "orphaned" conditionRefused = "refused" + // conditionUnwritten is the keeper's own: its status, with words unsaid, cannot be written. + conditionUnwritten = "status-unwritten" ) // ConditionKey is the key of one of the keeper's conditions: `nats.._`. @@ -631,18 +674,26 @@ func (k *Keeper) emit(event string, p Plan, revision uint64, why string, asked, } } -// sayNow says one event and answers whether the bus took it whole; nothing is kept to say again (its caller says it -// again itself). +// sayNow says one event and answers an error when the bus took none of it, so its caller says it again itself. One said +// on the seat's subject and not on its history is said: what is left is kept with its one id and said again at the next +// fill pass on the subjects left, never said again with a new id (review round 2, health note 3). 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 + err := k.Emit(&s) + if err == nil { + return nil } - return nil + if len(s.Said) > 0 { + s.Error = err.Error() + k.setStatus(func(st *Status) { st.Unsaid = append(st.Unsaid, s) }) + k.logf("the event %s of %s %s was said on %v and not on %v: the rest said again at the next fill pass: %v", event, p.Module, p.Local, s.Said, s.Left, err) + return nil + } + k.logf("the event %s of %s %s could not be said: %v", event, p.Module, p.Local, err) + return err } func (k *Keeper) saying(event string, p Plan, revision uint64, why string, asked, held map[string]Settings) Saying { diff --git a/modules/nats/internal/keeper/keeper_test.go b/modules/nats/internal/keeper/keeper_test.go index cef3c843..b9ad0c6c 100644 --- a/modules/nats/internal/keeper/keeper_test.go +++ b/modules/nats/internal/keeper/keeper_test.go @@ -78,13 +78,17 @@ func (b *fakeBus) Snapshot(_ context.Context, streams []string) (string, error) // memStore is a key-value state in memory. type memStore struct { - mu sync.Mutex - kv map[string][]byte - revs map[string]uint64 - rev uint64 + // refuse refuses a Put of `/`. + refuse map[string]error + mu sync.Mutex + kv map[string][]byte + revs map[string]uint64 + rev uint64 } -func newMemStore() *memStore { return &memStore{kv: map[string][]byte{}, revs: map[string]uint64{}} } +func newMemStore() *memStore { + return &memStore{kv: map[string][]byte{}, revs: map[string]uint64{}, refuse: map[string]error{}} +} func (s *memStore) Get(state, key string) ([]byte, uint64, bool, error) { s.mu.Lock() @@ -96,6 +100,9 @@ func (s *memStore) Get(state, key string) ([]byte, uint64, bool, error) { func (s *memStore) Put(state, key string, value []byte) (uint64, error) { s.mu.Lock() defer s.mu.Unlock() + if err := s.refuse[state+"/"+key]; err != nil { + return 0, err + } s.rev++ s.kv[state+"/"+key], s.revs[state+"/"+key] = value, s.rev return s.rev, nil @@ -139,17 +146,26 @@ type said struct { bodies []Event ids []string conds []string - // down refuses every event and condition while set, as a bus that does not take them. - down bool + // down refuses every event and condition while set, as a bus that does not take them; partly takes an event on the + // seat's subject and refuses its history. + down, partly bool + // left are the subjects each try was asked to say an event on (nil: every one). + left [][]string } func (s *said) emit(e *Saying) error { s.mu.Lock() defer s.mu.Unlock() + s.left = append(s.left, e.Left) if s.down { - e.Left = []string{"seat", "history"} + e.Left, e.Said = []string{"seat", "history"}, nil return errors.New("no responders") } + if s.partly && e.Left == nil { + e.Left, e.Said = []string{"history"}, []string{"seat"} + s.ids = append(s.ids, e.ID) + return errors.New("the history stream is full") + } s.events = append(s.events, e.Event+" "+e.Body.Module+" "+e.Body.Local) s.bodies = append(s.bodies, e.Body) s.ids = append(s.ids, e.ID) @@ -602,8 +618,17 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) { 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) + if raw, _, found, _ := store.Get(StateSpent, "bus-5"); !found || !strings.Contains(string(raw), `"request":"r5"`) { + t.Errorf("the spend record is %s (%v)", raw, found) + } + // The spend record is the process's own state, never in requests, which the tools write: every key of requests + // gone, a third request on the same warrant is still refused (review round 2, security F2). + keys, _ := store.Keys(StateRequests) + for _, key := range keys { + _ = store.Delete(StateRequests, key) + } + if got := put("r6b"); !strings.Contains(got.Error, "already acted") || len(bus.deleted) != 1 { + t.Errorf("with requests wiped, the spent warrant: %+v, deleted %v", got, bus.deleted) } }}, // An ask that ended unanswered ends its request, said, instead of a request that waits until its key ages out; and a @@ -666,6 +691,48 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) { t.Errorf("after the pass: unsaid %+v %+v, drafts %+v", st.Unsaid, st.UnsaidConditions, k.Held()["m"].Made["drafts"]) } }}, + // Review round 2 (health note 3): created said on the seat's subject and refused by its history is created, and its + // history is said again at the fill pass with the same id, on the history alone. + {"created said on the seat and not its history is said again on what is left", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) { + out.partly = true + k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts}}) + st := k.Status() + if m := k.Held()["m"].Made["drafts"]; !m.Created || len(st.Unsaid) != 1 || fmt.Sprint(st.Unsaid[0].Left) != "[history]" { + t.Fatalf("made %+v, unsaid %+v: created, its history kept to say", m, st.Unsaid) + } + id := st.Unsaid[0].ID + out.partly = false + k.FillPass() + if last := out.left[len(out.left)-1]; fmt.Sprint(last) != "[history]" || out.ids[len(out.ids)-1] != id || len(k.Status().Unsaid) != 0 { + t.Errorf("said again on %v with id %s (want [history], %s); unsaid %+v", last, out.ids[len(out.ids)-1], id, k.Status().Unsaid) + } + if n := strings.Count(fmt.Sprint(out.ids), id); n != 2 { + t.Errorf("the id %s was used %d times: once on the seat, once on the history", id, n) + } + }}, + // Review round 2 (seams note 3, health note 4): words unsaid with a status that cannot be written raise the keeper's + // own condition, once, cleared when the status is written again; the health names what a restart would lose. + {"a status that cannot be written with words unsaid is said", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) { + store := k.Store.(*memStore) + k.FillPass() + store.refuse[StateAsked+"/"+KeyStatus] = errors.New("the bucket is full") + k.Emit = func(e *Saying) error { e.Left = []string{"seat", "history"}; return errors.New("no responders") } + k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts, entry("raw", catalogue.StorageStream, "", nil)}}) + k.FillPass() + raisedN := strings.Count(fmt.Sprint(out.conds), "raised nats.status-unwritten.nats_keeper") + if raisedN != 1 { + t.Errorf("said %v: status-unwritten raised once", out.conds) + } + if h := HealthOf(k.Status(), nil, 0, k.now().Add(4*time.Minute)); !strings.Contains(h.Why, "may be lost at a restart") { + t.Errorf("health %+v: names what a restart loses", h) + } + delete(store.refuse, StateAsked+"/"+KeyStatus) + k.Emit = out.emit + k.FillPass() + if !strings.Contains(fmt.Sprint(out.conds), "cleared nats.status-unwritten.nats_keeper") { + t.Errorf("said %v: cleared once the status is written", out.conds) + } + }}, // CAT-N2b: nats_first_view_again asks the first view at once, by hand, recorded as a hand act. {"the first view asked again by hand", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) { store := k.Store.(*memStore) diff --git a/modules/nats/internal/keeper/requests.go b/modules/nats/internal/keeper/requests.go index b388e718..81cc08cc 100644 --- a/modules/nats/internal/keeper/requests.go +++ b/modules/nats/internal/keeper/requests.go @@ -55,12 +55,15 @@ type Request struct { // 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. +// ActedPrefix started the spend record's key while it was kept in StateRequests (`acted.`); a key of that form +// still met there is no request and is skipped. const ActedPrefix = "acted." -// ActedKey is the key spending the warrant on ask. -func ActedKey(ask string) string { return ActedPrefix + ask } +// StateSpent is the keeper process's own state of the warrants it spent, one key per ask id, claimed by compare-and-set +// before a recreate or a drop deletes anything (design 25 § 12.3: a warrant acts once). Only the process writes it: the +// tools bundle, which writes StateRequests as the tool runner's account, may read it and never delete a spend record +// (review round 2, security F2). +const StateSpent = "spent" // Acted is the value of an acted key: which request spent the warrant, on what. type Acted struct { @@ -267,7 +270,7 @@ func (k *Keeper) warranted(r *Request) error { // 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) + _, err := k.Store.Create(StateSpent, r.Ask, raw) if err == nil { return nil } @@ -275,7 +278,7 @@ func (k *Keeper) spend(r *Request) error { 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 { + if got, _, found, _ := k.Store.Get(StateSpent, 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", diff --git a/modules/nats/module.json b/modules/nats/module.json index 02b1320c..a066dcda 100644 --- a/modules/nats/module.json +++ b/modules/nats/module.json @@ -89,6 +89,12 @@ "name": "requests", "history": 5, "ttl-seconds": 86400 + }, + { + "name": "spent", + "history": 1, + "ttl-seconds": 86400, + "read-only": true } ], "listens": [ @@ -245,7 +251,8 @@ "bus": { "state": [ "asked", - "requests" + "requests", + "spent" ], "asks": true, "keeps": true,