From 81fa009bf447a75f70360b88b47498e4f8f9a9a5 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 10:19:02 +0100 Subject: [PATCH] =?UTF-8?q?The=20node=20seat's=20acts=20and=20the=20engine?= =?UTF-8?q?'s=20own=20asks=20are=20the=20node-engine's=20declared=20state?= =?UTF-8?q?=20acts=20on=20the=20bus:=20one=20key=20each=20under=20its=20ow?= =?UTF-8?q?n=20node's=20keys,=20written=20by=20compare-and-set,=20one=20as?= =?UTF-8?q?k=20open=20per=20path=20made=20by=20a=20create;=20acts=20reads?= =?UTF-8?q?=20one=20by=20its=20id=20(issue=20774,=20ADR=200563=20=C2=A72,?= =?UTF-8?q?=20=C2=A73)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/mesh-host/engineseat.go | 2 +- cmd/mesh-host/libraries_test.go | 2 +- internal/link/access_act_test.go | 13 +- internal/link/engineseat.go | 122 +++++++++++ internal/link/handacts.go | 304 ++++++++++++++-------------- internal/link/handover_test.go | 44 +++- internal/link/setuid_search_test.go | 2 +- module.json | 6 + packaging/operator_keys_test.go | 9 + 9 files changed, 344 insertions(+), 160 deletions(-) diff --git a/cmd/mesh-host/engineseat.go b/cmd/mesh-host/engineseat.go index eed00f2..4c11ece 100644 --- a/cmd/mesh-host/engineseat.go +++ b/cmd/mesh-host/engineseat.go @@ -241,7 +241,7 @@ func keepWaiting(vs []link.ResourceHealth) []string { // engineActs is the node seat this engine holds (ADR 0563): judged and performed by the same code as the controller's // signed asks, its asks to the operator over the session's link, its acts kept beside the state. func engineActs(opts options, node string, queue *link.Session, search *accounts.SetuidCache) *link.HandActs { - state := &link.FileState{Path: filepath.Join(filepath.Dir(opts.state), link.HandActsFile)} + state := link.BusState{Session: queue, Node: node} h := &link.HandActs{Node: node, State: state, JudgeHandOver: handOverJudgeFrom(opts.state), HandOver: handOverFrom(opts.state), JudgeAccess: accessJudgeFrom(opts.state), AccessAct: accessActFrom(opts.state, opts.timeout), diff --git a/cmd/mesh-host/libraries_test.go b/cmd/mesh-host/libraries_test.go index b171db6..c753209 100644 --- a/cmd/mesh-host/libraries_test.go +++ b/cmd/mesh-host/libraries_test.go @@ -59,7 +59,7 @@ func TestALibrarySaysItsWaitWithoutWordsToParse(t *testing.T) { // open, nor after it ended; and a library waiting for its disk is not asked. t.Run("the home engine asks the take once", func(t *testing.T) { var asked []asks.Ask - state := &link.FileState{Path: filepath.Join(t.TempDir(), link.HandActsFile)} + state := &link.MemState{} h := &link.HandActs{Node: "n1", State: state, JudgeAccess: func(act, access string) (link.Judged, error) { return link.Judged{Subject: access, Path: root, Sets: "the group mesh-media", Disk: "0:0 755"}, nil diff --git a/internal/link/access_act_test.go b/internal/link/access_act_test.go index ee5c7e8..82bf596 100644 --- a/internal/link/access_act_test.go +++ b/internal/link/access_act_test.go @@ -3,7 +3,6 @@ package link import ( "crypto/ed25519" "encoding/json" - "path/filepath" "strings" "testing" "time" @@ -117,6 +116,13 @@ func TestTheTakeActRunsOnlyOnAWarrant(t *testing.T) { if len(*asked) != 1 || *done != 0 { t.Fatalf("the call asked %d and performed %d", len(*asked), *done) } + // The act is one key, and the open ask one key made by a create (ADR 0563 §3). + if acts, _ := h.State.Keys("act."); len(acts) != 1 || acts[0] != "act."+(*asked)[0].ID { + t.Fatalf("the act is kept as %v", acts) + } + if open, _ := h.State.Keys("open."); len(open) != 1 || !strings.HasPrefix(open[0], "open.hand-over.") { + t.Fatalf("the open ask is kept as %v", open) + } if _, err := h.Call(EngineCall{Verb: ActHandOver, Caller: "x", Args: map[string]any{"path": "/srv/x"}}); err == nil { t.Fatal("a second ask of the same path was asked while one is open") } @@ -135,6 +141,9 @@ func TestTheTakeActRunsOnlyOnAWarrant(t *testing.T) { if *done != row.done || len(acts) != 1 || acts[0].State != row.state { t.Fatalf("performed %d (want %d), acts %+v (want %s)", *done, row.done, acts, row.state) } + if open, _ := h.State.Keys("open."); (len(open) == 1) != (row.state == HandAsked) { + t.Fatalf("an ask %s leaves its open key as %v", row.state, open) + } if row.state == HandDone && (acts[0].By != "operator" || acts[0].Via == "" || acts[0].Ask == "") { t.Fatalf("the act does not record the warrant: %+v", acts[0]) } @@ -157,7 +166,7 @@ func anEngine(t *testing.T) (*HandActs, *[]asks.Ask, *int) { t.Helper() var asked []asks.Ask done := 0 - state := &FileState{Path: filepath.Join(t.TempDir(), HandActsFile)} + state := &MemState{} h := &HandActs{Node: "n1", State: state, HandOver: func(path, by string) (string, error) { done++; return path + " handed over by " + by, nil }} h.Asker = &asker.Client{Module: asks.EngineAsker("n1"), State: state, diff --git a/internal/link/engineseat.go b/internal/link/engineseat.go index 7b902bc..258e3f5 100644 --- a/internal/link/engineseat.go +++ b/internal/link/engineseat.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "sort" "strings" "time" @@ -258,3 +259,124 @@ func (q *Session) HeardServed(seat, verb string) (bool, error) { } return asker.HeardServed(seat, verb) } + +// ---- the engine's acts in the bus's own storage (ADR 0563 §2, §3) ------------------------------------------------- + +// actsStore is this node's keys of ActsBucket over the link open now: bound, never created (the grants part creates +// it, as the node-engine's declared state). +func (l *natsLink) actsStore() (nats.KeyValue, error) { + kv, err := l.js.KeyValue(ActsBucket) + if err != nil { + return nil, fmt.Errorf("the node-engine's state %s cannot be bound (it is created from the node-engine's "+ + "manifest; asked again once it is): %w", ActsBucket, err) + } + return kv, nil +} + +// BusState is the engine's ActStore: its own node's keys of ActsBucket, `.`, over whichever link the session +// has open; with none open every read and write is refused, and nothing is kept anywhere else. +type BusState struct { + Session *Session + Node string +} + +func (b BusState) kv() (nats.KeyValue, error) { + b.Session.init() + b.Session.mu.Lock() + bus := b.Session.bus + b.Session.mu.Unlock() + l, ok := bus.(interface{ actsStore() (nats.KeyValue, error) }) + if bus == nil || !ok { + return nil, fmt.Errorf("%w: this machine has no link to the bus open now, where its acts are kept", ErrNotAsked) + } + return l.actsStore() +} + +func (b BusState) key(k string) string { return b.Node + "." + k } + +// changed is a write that lost its compare-and-set, as stdio.ErrChanged. +func changed(err error) error { + var api *nats.APIError + if errors.Is(err, nats.ErrKeyExists) || (errors.As(err, &api) && api.ErrorCode == nats.JSErrCodeStreamWrongLastSequence) { + return stdio.ErrChanged + } + return err +} + +func (b BusState) Get(key string) (*stdio.StateEntry, error) { + kv, err := b.kv() + if err != nil { + return nil, err + } + e, err := kv.Get(b.key(key)) + if errors.Is(err, nats.ErrKeyNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + return &stdio.StateEntry{Key: key, Value: e.Value(), Revision: e.Revision()}, nil +} + +func (b BusState) Create(key string, value any) (uint64, error) { + kv, err := b.kv() + if err != nil { + return 0, err + } + raw, err := json.Marshal(value) + if err != nil { + return 0, err + } + rev, err := kv.Create(b.key(key), raw) + return rev, changed(err) +} + +func (b BusState) Update(key string, value any, revision uint64) (uint64, error) { + kv, err := b.kv() + if err != nil { + return 0, err + } + raw, err := json.Marshal(value) + if err != nil { + return 0, err + } + rev, err := kv.Update(b.key(key), raw, revision) + return rev, changed(err) +} + +func (b BusState) Delete(key string) error { + kv, err := b.kv() + if err != nil { + return err + } + return kv.Delete(b.key(key)) +} + +// Keys are this node's keys under prefix, without the node: a watch of `.>` alone, the one filter the node's +// grant names, its values left out. +func (b BusState) Keys(prefix string) ([]string, error) { + kv, err := b.kv() + if err != nil { + return nil, err + } + w, err := kv.Watch(b.Node+".>", nats.MetaOnly(), nats.IgnoreDeletes()) + if err != nil { + return nil, err + } + defer func() { _ = w.Stop() }() + var out []string + for { + select { + case e := <-w.Updates(): + if e == nil { + sort.Strings(out) + return out, nil + } + if k := strings.TrimPrefix(e.Key(), b.Node+"."); strings.HasPrefix(k, prefix) { + out = append(out, k) + } + case <-time.After(10 * time.Second): + return nil, errors.New("the node-engine's acts could not be listed within 10s") + } + } +} diff --git a/internal/link/handacts.go b/internal/link/handacts.go index 5a53f69..e4f7bf7 100644 --- a/internal/link/handacts.go +++ b/internal/link/handacts.go @@ -2,12 +2,12 @@ package link import ( "crypto/rand" + "crypto/sha256" "encoding/hex" "encoding/json" "errors" "fmt" - "os" - "path/filepath" + "sort" "strconv" "strings" "sync" @@ -30,7 +30,7 @@ import ( // // The read-only verbs need no warrant (§2): judge-directories changes nothing; setuid-search is bounded here, one at a // time and at most one started in an hour. Every act, with who asked as the runtime proved it, is kept in the engine's own -// state beside its node's state (HandActsFile), read by the verb acts. +// declared state acts (ActsBucket, created by the grants part), under its own node's keys alone, read by the verb acts. // The acts of the seat, as its verbs name them. const ( @@ -51,14 +51,13 @@ const ( // SearchEvery is how often a setuid search may be started on a node (ADR 0563 §2): one in any hour. const SearchEvery = time.Hour -// keptHandActs is how many acts the engine keeps, newest last. -const keptHandActs = 200 - // askMargin is how long past an ask's expiry the router is given to say it ended, before the engine closes it itself. const askMargin = 5 * time.Minute -// HandActsFile is where the engine keeps its acts and its asks, beside its state. -const HandActsFile = "hand-acts.json" +// ActsBucket is the node-engine's declared state acts (its manifest's state, ADR 0563 §2, §3), the bucket the grants +// part creates (mesh-sdk buscompose EngineActsBucket): keys `.act.`, `.open..` and +// `.ask.`, each written by this node's engine alone. +const ActsBucket = "mesh-host_acts" // Judged is what the engine judged of one act before it asks: what it is about, what it will set (the owner, group and // mode, in words), and what the disk holds now (Disk), compared again when the warrant arrives. @@ -112,8 +111,8 @@ type RootVerdict struct { // HandActs is the engine's side of the seat. Every func is the engine's own judgement or act; nil refuses. type HandActs struct { Node string - // State is where the acts and the asks are kept (FileState beside the node's state). - State *FileState + // State is where the acts and the asks are kept: this node's keys of ActsBucket (BusState), each key one act. + State ActStore // Asker asks the operator as asks.EngineAsker(Node), with State as its store. Asker *asker.Client // JudgeHandOver judges a hand-over's path, refusing in words; HandOver records it on the warrant. @@ -178,6 +177,9 @@ func (h *HandActs) Call(c EngineCall) (any, error) { } return map[string]any{"machines": []RootVerdict{h.RootFree()}}, nil case "acts": + if id := arg("id"); id != "" { + return h.Act(id) + } limit, _ := strconv.Atoi(arg("limit")) if f, ok := c.Args["limit"].(float64); ok { limit = int(f) @@ -189,70 +191,98 @@ func (h *HandActs) Call(c EngineCall) (any, error) { // ---- the record -------------------------------------------------------------------------------------------------- -const actsKey = "acts" +// ActStore is the engine's own state, this node's keys alone (BusState prefixes them with the node): each act one key, +// written by compare-and-set on its revision, an open ask one key made by a create. +type ActStore interface { + Get(key string) (*stdio.StateEntry, error) + Create(key string, value any) (uint64, error) + Update(key string, value any, revision uint64) (uint64, error) + Delete(key string) error + // Keys are the keys under prefix. + Keys(prefix string) ([]string, error) +} -func (h *HandActs) read() ([]HandAct, uint64, error) { - e, err := h.State.Get(actsKey) +func actKey(id string) string { return "act." + id } + +// openKey is the one key an open ask of act on subject holds: made by a create, so a second ask of it is refused. +func openKey(act, subject string) string { + sum := sha256.Sum256([]byte(subject)) + return "open." + act + "." + hex.EncodeToString(sum[:12]) +} + +// read is every act kept, oldest first. +func (h *HandActs) read() ([]HandAct, error) { + keys, err := h.State.Keys("act.") + if err != nil { + return nil, fmt.Errorf("the engine's acts cannot be listed: %w", err) + } + var out []HandAct + for _, k := range keys { + a, _, err := h.readOne(k) + if err != nil { + return nil, err + } + if a != nil { + out = append(out, *a) + } + } + sort.SliceStable(out, func(i, j int) bool { return out[i].At.Before(out[j].At) }) + return out, nil +} + +func (h *HandActs) readOne(key string) (*HandAct, uint64, error) { + e, err := h.State.Get(key) if err != nil || e == nil { return nil, 0, err } - var out []HandAct - if err := json.Unmarshal(e.Value, &out); err != nil { - return nil, 0, fmt.Errorf("the engine's record of its acts cannot be read: %w", err) + var a HandAct + if err := json.Unmarshal(e.Value, &a); err != nil { + return nil, 0, fmt.Errorf("the engine's act %s cannot be read: %w", key, err) } - return out, e.Revision, nil -} - -// change reads the record, lets edit change it, and writes it back by compare-and-set, trying again on a race. -func (h *HandActs) change(edit func([]HandAct) ([]HandAct, error)) error { - for range 5 { - acts, rev, err := h.read() - if err != nil { - return err - } - acts, err = edit(acts) - if err != nil { - return err - } - if len(acts) > keptHandActs { - acts = acts[len(acts)-keptHandActs:] - } - if rev == 0 { - _, err = h.State.Create(actsKey, acts) - } else { - _, err = h.State.Update(actsKey, acts, rev) - } - if !errors.Is(err, stdio.ErrChanged) { - return err - } - } - return errors.New("the engine's record of its acts kept changing under its write") + return &a, e.Revision, nil } func (h *HandActs) record(a HandAct) error { - return h.change(func(acts []HandAct) ([]HandAct, error) { return append(acts, a), nil }) + _, err := h.State.Create(actKey(a.ID), a) + return err } -// end marks the open act of an ask ended, once: false when it was not open. +// end marks the open act of an ask ended, once, by compare-and-set on its revision, and frees its open key: false when +// it was not open. func (h *HandActs) end(ask string, set func(*HandAct)) (HandAct, bool, error) { - var ended HandAct - found := false - err := h.change(func(acts []HandAct) ([]HandAct, error) { - for i := range acts { - if acts[i].Ask == ask && acts[i].State == HandAsked { - set(&acts[i]) - acts[i].Ended = h.now().UTC() - ended, found = acts[i], true - } + for range 5 { + a, rev, err := h.readOne(actKey(ask)) + if err != nil || a == nil || a.State != HandAsked { + return HandAct{}, false, err } - return acts, nil - }) - return ended, found, err + set(a) + a.Ended = h.now().UTC() + if _, err := h.State.Update(actKey(ask), *a, rev); errors.Is(err, stdio.ErrChanged) { + continue + } else if err != nil { + return HandAct{}, false, err + } + _ = h.State.Delete(openKey(a.Act, a.Subject)) + return *a, true, nil + } + return HandAct{}, false, errors.New("the engine's act " + ask + " kept changing under its write") +} + +// Act is the verb acts with an id: that one act. +func (h *HandActs) Act(id string) (any, error) { + a, _, err := h.readOne(actKey(id)) + if err != nil { + return nil, err + } + if a == nil { + return nil, fmt.Errorf("this engine keeps no act %q", id) + } + return map[string]any{"node": h.Node, "acts": []HandAct{*a}}, nil } // Acts is the verb acts: every act kept, newest first, at most limit when given. func (h *HandActs) Acts(limit int) (any, error) { - acts, _, err := h.read() + acts, err := h.read() if err != nil { return nil, err } @@ -299,7 +329,7 @@ func (h *HandActs) Search(caller, why string) (any, error) { if h.Searching != nil && h.Searching() { return nil, errors.New("a search for setuid programs runs on this machine now; no second is started while it does") } - acts, _, err := h.read() + acts, err := h.read() if err != nil { return nil, err } @@ -369,7 +399,7 @@ func (h *HandActs) AskAccess(caller, act, access, why string, auto bool) (any, e // AskedFor says whether an act on subject was ever asked by the engine itself: a library's take is asked by itself once // (ADR 0563 §4), and again only by hand. func (h *HandActs) AskedFor(act, subject string) bool { - acts, _, err := h.read() + acts, err := h.read() if err != nil { return true // unread is never a reason to ask again } @@ -391,16 +421,23 @@ func (h *HandActs) ask(act, caller, why string, auto bool, j Judged) (any, error a := HandAct{ID: h.id(act), Act: act, Subject: j.Subject, Path: j.Path, Caller: caller, Why: why, At: now, State: HandAsked, Sets: j.Sets, Disk: j.Disk, Expires: now.Add(asks.ApproveLasts - time.Minute), Auto: auto} a.Ask = a.ID - err := h.change(func(acts []HandAct) ([]HandAct, error) { - for _, o := range acts { - if o.Act == act && o.Subject == j.Subject && o.State == HandAsked { - return nil, fmt.Errorf("%s of %s is asked already: the ask %s waits for the operator's answer", act, - j.Subject, o.Ask) + // One ask open per path or access: its open key made by a create, which a second ask cannot make. + if _, err := h.State.Create(openKey(act, j.Subject), map[string]string{"ask": a.Ask}); errors.Is(err, stdio.ErrChanged) { + held := "another" + if e, _ := h.State.Get(openKey(act, j.Subject)); e != nil { + var o struct { + Ask string `json:"ask"` + } + if json.Unmarshal(e.Value, &o) == nil && o.Ask != "" { + held = o.Ask } } - return append(acts, a), nil - }) - if err != nil { + return nil, fmt.Errorf("%s of %s is asked already: the ask %s waits for the operator's answer", act, j.Subject, held) + } else if err != nil { + return nil, err + } + if err := h.record(a); err != nil { + _ = h.State.Delete(openKey(act, j.Subject)) return nil, err } bind, err := asks.ActDigest(h.bound(a)) @@ -444,16 +481,10 @@ func orWhy(why string) string { // ask: a no is recorded refused; a yes is held to the act the option bound and the disk as judged, judged again now, // and only then performed and recorded done with the warrant. func (h *HandActs) Warranted(o asks.Option, w asks.Warrant) error { - acts, _, err := h.read() + a, _, err := h.readOne(actKey(w.Ask)) if err != nil { return err } - var a *HandAct - for i := range acts { - if acts[i].Ask == w.Ask { - a = &acts[i] - } - } if a == nil || a.State != HandAsked { return fmt.Errorf("the warrant answers the ask %s, which is not open here: nothing is done", w.Ask) } @@ -548,7 +579,7 @@ func (h *HandActs) Reconcile() { if h.Asker == nil { return } - acts, _, err := h.read() + acts, err := h.read() if err != nil { return } @@ -568,98 +599,63 @@ func (h *HandActs) Reconcile() { } } -// ---- the engine's own state --------------------------------------------------------------------------------------- - -// FileState is a store of keys with revisions in one file beside the node's state, written whole and renamed into -// place: the engine's own state, which asker.Client keeps its asks in and HandActs its acts (compare-and-set on a -// revision, as a bucket's). -type FileState struct { - Path string +// MemState is an ActStore in memory: an engine's acts while no bus is reached in a test. +type MemState struct { mu sync.Mutex + keys map[string]stdio.StateEntry + top uint64 } -type fileEntry struct { - Value json.RawMessage `json:"value"` - Revision uint64 `json:"revision"` +func (m *MemState) Get(key string) (*stdio.StateEntry, error) { + m.mu.Lock() + defer m.mu.Unlock() + if e, ok := m.keys[key]; ok { + return &e, nil + } + return nil, nil } -func (f *FileState) load() (map[string]fileEntry, error) { - raw, err := os.ReadFile(f.Path) - if errors.Is(err, os.ErrNotExist) { - return map[string]fileEntry{}, nil - } - if err != nil { - return nil, fmt.Errorf("the engine's own state %s cannot be read: %w", f.Path, err) - } - out := map[string]fileEntry{} - if err := json.Unmarshal(raw, &out); err != nil { - return nil, fmt.Errorf("the engine's own state %s is not a state: %w", f.Path, err) - } - return out, nil +func (m *MemState) Create(key string, value any) (uint64, error) { return m.put(key, value, 0, true) } + +func (m *MemState) Update(key string, value any, revision uint64) (uint64, error) { + return m.put(key, value, revision, false) } -func (f *FileState) save(all map[string]fileEntry) error { - raw, err := json.Marshal(all) - if err != nil { - return err - } - if err := os.MkdirAll(filepath.Dir(f.Path), 0o700); err != nil { - return err - } - tmp := f.Path + ".tmp" - if err := os.WriteFile(tmp, raw, 0o600); err != nil { - return err - } - return os.Rename(tmp, f.Path) -} - -// Get is a key and its revision; nil when it is not kept. -func (f *FileState) Get(key string) (*stdio.StateEntry, error) { - f.mu.Lock() - defer f.mu.Unlock() - all, err := f.load() - if err != nil { - return nil, err - } - e, ok := all[key] - if !ok { - return nil, nil - } - return &stdio.StateEntry{Key: key, Value: e.Value, Revision: e.Revision}, nil -} - -// Create writes a key that is not kept, or says stdio.ErrChanged. -func (f *FileState) Create(key string, value any) (uint64, error) { return f.put(key, value, 0, true) } - -// Update writes a key at the revision read, or says stdio.ErrChanged. -func (f *FileState) Update(key string, value any, revision uint64) (uint64, error) { - return f.put(key, value, revision, false) -} - -func (f *FileState) put(key string, value any, revision uint64, create bool) (uint64, error) { +func (m *MemState) put(key string, value any, revision uint64, create bool) (uint64, error) { raw, err := json.Marshal(value) if err != nil { return 0, err } - f.mu.Lock() - defer f.mu.Unlock() - all, err := f.load() - if err != nil { - return 0, err - } - was, kept := all[key] + m.mu.Lock() + defer m.mu.Unlock() + was, kept := m.keys[key] if (create && kept) || (!create && (!kept || was.Revision != revision)) { return 0, stdio.ErrChanged } - var top uint64 - for _, e := range all { - if e.Revision > top { - top = e.Revision + if m.keys == nil { + m.keys = map[string]stdio.StateEntry{} + } + m.top++ + m.keys[key] = stdio.StateEntry{Key: key, Value: raw, Revision: m.top} + return m.top, nil +} + +func (m *MemState) Delete(key string) error { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.keys, key) + return nil +} + +func (m *MemState) Keys(prefix string) ([]string, error) { + m.mu.Lock() + defer m.mu.Unlock() + var out []string + for k := range m.keys { + if strings.HasPrefix(k, prefix) { + out = append(out, k) } } - all[key] = fileEntry{Value: raw, Revision: top + 1} - if err := f.save(all); err != nil { - return 0, err - } - return top + 1, nil + sort.Strings(out) + return out, nil } diff --git a/internal/link/handover_test.go b/internal/link/handover_test.go index 3734e48..10920db 100644 --- a/internal/link/handover_test.go +++ b/internal/link/handover_test.go @@ -9,6 +9,9 @@ import ( "strings" "testing" "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "github.com/nats-io/nats.go" ) // The hand-over's ask and answer hold these field names; the controller's side holds the same list @@ -202,7 +205,7 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) { // 774 (ADR 0563 §1): a call of the node seat reaches its verb with its caller as the subject proves it, unproven on // the subject that names none; a verb the seat does not serve is refused, and the answer is a tool's envelope. t.Run("a call of the node seat reaches its verb with its proven caller", func(t *testing.T) { - h := &HandActs{Node: "n1", State: &FileState{Path: t.TempDir() + "/" + HandActsFile}, + h := &HandActs{Node: "n1", State: &MemState{}, RootFree: func() RootVerdict { return RootVerdict{Machine: "n1", Free: true, Why: "judged"} }, JudgeDirectories: func(by string) (string, error) { return "judged for " + by, nil }} for _, row := range []struct{ verb, subject, want string }{ @@ -261,6 +264,45 @@ func TestNatsAHandOverIsAskedOfTheNode(t *testing.T) { if len(handled) != 1 { t.Fatalf("handled %d asks; want only the one with a reply", len(handled)) } + // 774 (ADR 0563 §2, §3): the engine's acts are its node's keys of its declared state on the bus, each written by + // compare-and-set: a second create and a stale revision lose, and another node's keys are not this node's. + t.Run("the acts are kept in the bus's own storage under this node's keys", func(t *testing.T) { + _ = js.DeleteKeyValue(ActsBucket) + if _, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: ActsBucket}); err != nil { + t.Fatal(err) + } + q := &Session{Membership: m, Timeout: time.Second} + q.bus = l + mine, theirs := BusState{Session: q, Node: m.Node}, BusState{Session: q, Node: "another"} + rev, err := mine.Create("act.a-1", map[string]string{"state": "asked"}) + if err != nil { + t.Fatal(err) + } + if _, err := mine.Create("act.a-1", map[string]string{}); !errors.Is(err, stdio.ErrChanged) { + t.Errorf("a second create: %v", err) + } + if _, err := mine.Update("act.a-1", map[string]string{"state": "done"}, rev+7); !errors.Is(err, stdio.ErrChanged) { + t.Errorf("an update at a stale revision: %v", err) + } + if _, err := mine.Update("act.a-1", map[string]string{"state": "done"}, rev); err != nil { + t.Errorf("an update at its revision: %v", err) + } + _, _ = theirs.Create("act.b-1", map[string]string{}) + if keys, err := mine.Keys("act."); err != nil || len(keys) != 1 || keys[0] != "act.a-1" { + t.Errorf("this node's acts are %v, %v", keys, err) + } + if e, _ := js.KeyValue(ActsBucket); e != nil { + if got, err := e.Get(m.Node + ".act.a-1"); err != nil || !strings.Contains(string(got.Value()), "done") { + t.Errorf("the bucket holds %v, %v", got, err) + } + } + if err := mine.Delete("act.a-1"); err != nil { + t.Fatal(err) + } + if e, err := mine.Get("act.a-1"); err != nil || e != nil { + t.Errorf("a deleted act reads %v, %v", e, err) + } + }) // 774 (ADR 0563 §1): the node seat is served on this node's subjects alone, naming its caller, and said to discovery. t.Run("the node seat is served on this node and discovered", func(t *testing.T) { l.ServeEngineSeat(func(c EngineCall) (any, error) { return c.Verb + " for " + c.Caller, nil }) diff --git a/internal/link/setuid_search_test.go b/internal/link/setuid_search_test.go index b9c7c72..4c42053 100644 --- a/internal/link/setuid_search_test.go +++ b/internal/link/setuid_search_test.go @@ -61,7 +61,7 @@ func TestASetuidSearchAskIsBelievedOnlySignedForThisNodeAndOnce(t *testing.T) { // next may start, and records who asked. t.Run("the node seat's search is bounded", func(t *testing.T) { at, running, started := now, false, 0 - h := &HandActs{Node: "n1", State: &FileState{Path: t.TempDir() + "/" + HandActsFile}, Now: func() time.Time { return at }, + h := &HandActs{Node: "n1", State: &MemState{}, Now: func() time.Time { return at }, Searching: func() bool { return running }, SetuidSearch: func(by string) (string, error) { started++; return "a new search starts", nil }} for _, row := range []struct { diff --git a/module.json b/module.json index 8fd4fa6..d10f063 100644 --- a/module.json +++ b/module.json @@ -17,6 +17,12 @@ "consumes": [ "restic.backed-up" ], + "state": [ + { + "name": "acts", + "ttl-seconds": 7776000 + } + ], "config": { "operator-signers": { "type": "string", diff --git a/packaging/operator_keys_test.go b/packaging/operator_keys_test.go index 5935047..6794b9e 100644 --- a/packaging/operator_keys_test.go +++ b/packaging/operator_keys_test.go @@ -68,4 +68,13 @@ func TestTheManifestConsumesTheBackupEvent(t *testing.T) { if len(m.Consumes) != 1 || m.Consumes[0] != "restic.backed-up" { t.Fatalf("consumes %v", m.Consumes) } + // 774 (ADR 0563 §2, §3): the acts of the node seat and the engine's own asks are its declared state acts, the bucket + // mesh-host_acts the grants part creates and each node's engine writes under its own keys. + parsed, err := catalogue.ParseManifest(raw) + if err != nil { + t.Fatal(err) + } + if len(parsed.State) != 1 || parsed.State[0].Name != "acts" || parsed.State[0].TTLSeconds != 90*24*60*60 { + t.Fatalf("the manifest declares the state %+v", parsed.State) + } }