From 277a4d00605f245848eb6e4503159d028ab541b8 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 14:47:20 +0100 Subject: [PATCH 1/2] mesh-vault answers own in one pass: assignments from its watch, each module's once, one synced record per answer (issue 1126) Seed-9's probe (13:31-13:36Z): novox's own of 57 secrets and 146 credentials took 26 s, past the node-engine's 15 s, and ace and novox refused their whole shadow answer. own read the controller's assignment over the bus once per secret and per credential (203 reads), read every record of every node, and synced the hand-out record once per value, under the one lock every node's answers share. Now the vault watches the assignments mirror (a watch, never a poll) and reads it from memory once caught up, over the bus until then; one answer reads each module's assignment once; only own// records are read; and every hand-out of one answer is recorded in one synced write, none handed when it cannot be. Quick fix weighed: a longer bound in the node-engine (VaultWithin). Thorough fix chosen: the cost grows with every module and every node, and the lock is shared by every shadow answer and every apply. --- modules/mesh-vault/cmd/mesh-vault/askers.go | 58 +++++++++++ modules/mesh-vault/cmd/mesh-vault/main.go | 28 +++++- modules/mesh-vault/cmd/mesh-vault/ownnode.go | 65 +++++++++--- .../mesh-vault/cmd/mesh-vault/ownnode_test.go | 99 +++++++++++++++++++ modules/mesh-vault/cmd/mesh-vault/store.go | 34 +++++-- modules/mesh-vault/cmd/mesh-vault/vault.go | 21 +++- .../mesh-vault/cmd/mesh-vault/vault_test.go | 4 +- 7 files changed, 280 insertions(+), 29 deletions(-) diff --git a/modules/mesh-vault/cmd/mesh-vault/askers.go b/modules/mesh-vault/cmd/mesh-vault/askers.go index e8eab5ab..5f522d3e 100644 --- a/modules/mesh-vault/cmd/mesh-vault/askers.go +++ b/modules/mesh-vault/cmd/mesh-vault/askers.go @@ -4,6 +4,9 @@ import ( "encoding/json" "fmt" "strings" + "sync" + + stdio "git.novox.be/novox/mesh-sdk/go" ) // Who may ask `get` for which secret (novox/hq ADR 0349 ยง11, design 57 V4, issue 647). @@ -207,3 +210,58 @@ func (d Declared) May(caller, name string) (bool, string) { // bus error rather than an answer. The vault says "holds no secret named ... yet", which the client waits on and asks // again, and records no refusal. const notYetPrefix = "not yet: " + +// WatchedAssignments is the assignments mirror as the vault's watch of it hands it over (issue 1126, 2026-10-28: novox's +// `own` read one assignment over the bus per own secret and per credential, 203 reads, and answered after 26 s, past +// the node-engine's 15 s). The state is the bus's; the vault watches it, never polls it, and reads it from memory once the +// watch has handed every current value. Until then, and when the watch could not be begun, a read goes to the bus +// (Fallback), as before. +type WatchedAssignments struct { + mu sync.Mutex + held map[string][]byte + caught bool + Fallback AssignmentReader +} + +// Take keeps one change the watch handed over: a value put, or a key deleted. +func (w *WatchedAssignments) Take(c stdio.StateChange) error { + w.mu.Lock() + defer w.mu.Unlock() + if w.held == nil { + w.held = map[string][]byte{} + } + if c.Op == "delete" || c.Op == "purge" || len(c.Value) == 0 { + delete(w.held, c.Key) + return nil + } + w.held[c.Key] = append([]byte(nil), c.Value...) + return nil +} + +// Caught says the watch handed every current value: reads are answered from memory from now on. +func (w *WatchedAssignments) Caught() { + w.mu.Lock() + defer w.mu.Unlock() + if w.held == nil { + w.held = map[string][]byte{} + } + w.caught = true +} + +// Read is one value of the mirror, `.`: nil when there is none. +func (w *WatchedAssignments) Read(key string) ([]byte, error) { + w.mu.Lock() + if w.caught { + raw, ok := w.held[key] + w.mu.Unlock() + if !ok { + return nil, nil + } + return append([]byte(nil), raw...), nil + } + w.mu.Unlock() + if w.Fallback == nil { + return nil, fmt.Errorf("the assignments mirror is not watched yet, and nothing reads it from the bus") + } + return w.Fallback(key) +} diff --git a/modules/mesh-vault/cmd/mesh-vault/main.go b/modules/mesh-vault/cmd/mesh-vault/main.go index e78a7ff7..cce1473c 100644 --- a/modules/mesh-vault/cmd/mesh-vault/main.go +++ b/modules/mesh-vault/cmd/mesh-vault/main.go @@ -49,13 +49,34 @@ func build() *Env { } // Who may ask (askers.go): a node's provisioner client, for what its module's plan there names. // Where it sees a consumer bind a provider (pair.go): the same mirror, every key. - env.Vault = &Vault{Store: store, Own: own, Askers: Declared{Read: readAssignment, Store: store}, + // The assignments as the vault's watch keeps them (watchAssignments), read over the bus until it has caught up. + assigned = &WatchedAssignments{Fallback: readAssignment} + env.Vault = &Vault{Store: store, Own: own, Askers: Declared{Read: assigned.Read, Store: store}, Assignments: Mirror{Keys: assignmentKeys, Read: readAssignment}, Emit: emit, Ask: stdio.Ask, Now: time.Now, Log: say, Accounts: busAccounts{}, Haves: readHave, HaveKeys: func() ([]string, error) { return stdio.State(HaveRead).Keys() }} return env } +// assigned is the assignments mirror as the vault's watch keeps it. +var assigned *WatchedAssignments + +// watchAssignments begins the watch of the assignments mirror the vault's rules read (issue 1126, 2026-10-28): a +// watch, never a poll, so `own` and `get` read no assignment over the bus once it has caught up. It answers whether the +// watch began; one that did not is begun again at the next minute pass, and reads go to the bus meanwhile. +func watchAssignments() bool { + if assigned == nil { + return true + } + if err := stdio.State(AssignmentsRead).Watch("", assigned.Take); err != nil { + say("the assignments are not watched, so each is read over the bus; begun again at the next pass: %v", err) + return false + } + assigned.Caught() + say("the assignments are watched: the vault's rules read them from memory") + return true +} + // readHave is one key of the node-engines' have bucket, read through the runtime under the vault's own account (its // manifest `reads` it), or nil when there is none. func readHave(key string) ([]byte, error) { @@ -149,11 +170,14 @@ func main() { written, failed := env.Vault.HashHeld() say("hashed %d bus accounts at start; %d could not be, tried again every minute: %v", len(written), len(failed), failed) } - watched := watchHandings(env) + watched, watching := watchHandings(env), watchAssignments() for { if !watched { watched = watchHandings(env) } + if !watching { + watching = watchAssignments() + } env.pass() if env.Vault != nil { env.Vault.HashOwed() diff --git a/modules/mesh-vault/cmd/mesh-vault/ownnode.go b/modules/mesh-vault/cmd/mesh-vault/ownnode.go index b873dc44..c93fd003 100644 --- a/modules/mesh-vault/cmd/mesh-vault/ownnode.go +++ b/modules/mesh-vault/cmd/mesh-vault/ownnode.go @@ -70,7 +70,7 @@ func credentialNames(c CredentialAsked) []string { // credentialRefusal says why node may not have c, or "": a consumer's credential to the machine its consuming module // stands on, a provider's to the providing machine, and nothing to a machine that is neither. -func (v *Vault) credentialRefusal(node string, c CredentialAsked) string { +func (v *Vault) credentialRefusal(node string, c CredentialAsked, plan *planReader) string { for _, part := range []string{c.Provision, c.Consumer, c.ConsumerModule, c.Provider} { if !namePart.MatchString(part) { return fmt.Sprintf("%q is not a name the mesh writes", part) @@ -85,11 +85,10 @@ func (v *Vault) credentialRefusal(node string, c CredentialAsked) string { return fmt.Sprintf("%s asks as a consumer on %s, and is not that machine", node, c.Consumer) } // The consuming module stands on the machine: read, never assumed (an assignment that cannot be read refuses). - read, why := v.assignmentReader(node) + raw, why, err := plan.read(c.ConsumerModule) if why != "" { return why } - raw, err := read(node + "." + c.ConsumerModule) if err != nil { return fmt.Sprintf(notYetPrefix+"the assignment %s.%s could not be read: %v", node, c.ConsumerModule, err) } @@ -106,6 +105,38 @@ func (v *Vault) credentialRefusal(node string, c CredentialAsked) string { return fmt.Sprintf("the role %q is not consumer or provider", c.Role) } +// planReader reads the assignments of one machine's modules for one answer of `own`, each module's once (issue 1126, +// 2026-10-28: one read per own secret and per credential, 203 on novox, held the answer past the node-engine's bound). +type planReader struct { + node string + reader AssignmentReader + why string + got map[string]planRead +} + +type planRead struct { + raw []byte + err error +} + +func (v *Vault) plansOf(node string) *planReader { + read, why := v.assignmentReader(node) + return &planReader{node: node, reader: read, why: why, got: map[string]planRead{}} +} + +// read is module's assignment on the machine, nil when there is none, or why the vault reads none. +func (p *planReader) read(module string) ([]byte, string, error) { + if p.why != "" { + return nil, p.why, nil + } + r, ok := p.got[module] + if !ok { + r.raw, r.err = p.reader(p.node + "." + module) + p.got[module] = r + } + return r.raw, "", r.err +} + // assignmentReader is the vault's reader of the assignments, or why it has none: the rule `own` keeps reads them, // and without them it refuses rather than hands (a rule that cannot be read is no rule). func (v *Vault) assignmentReader(node string) (AssignmentReader, string) { @@ -177,17 +208,21 @@ func (v *Vault) OwnOfNode(caller string, asked ...CredentialAsked) (OwnAnswer, e if err != nil { return OwnAnswer{}, fmt.Errorf("%s's sealing key as recorded is not usable: %w", node, err) } - all, err := v.Store.List() + // Only this machine's own records are read, each module's assignment once, and every hand-out recorded in one synced + // write after all are sealed: none is handed when that record cannot be written (issue 1126, 2026-10-28). + all, err := v.Store.ListPrefix("own/" + node + "/") if err != nil { return OwnAnswer{}, err } + plan := v.plansOf(node) + var handings []Handing out := OwnAnswer{Node: node, Key: key} for _, r := range all { parts := strings.Split(r.Name, "/") if len(parts) != 4 || parts[0] != "own" || parts[1] != node { continue } - handed, err := v.handed(node, parts[2]) + handed, err := plan.handed(parts[2]) if err != nil { return OwnAnswer{}, err } @@ -202,9 +237,7 @@ func (v *Vault) OwnOfNode(caller string, asked ...CredentialAsked) (OwnAnswer, e if err != nil { return OwnAnswer{}, err } - if err := v.recordHanding(Handing{To: caller, Name: r.Name, Generation: r.Generation, Outcome: "handed"}); err != nil { - return OwnAnswer{}, fmt.Errorf("%s is not handed out: who was handed it could not be recorded: %w", r.Name, err) - } + handings = append(handings, Handing{To: caller, Name: r.Name, Generation: r.Generation, Outcome: "handed"}) out.Own = append(out.Own, NodeOwnSecret{Module: parts[2], Name: parts[3], Origin: composeOrigin(r.Origin), Generation: r.Generation, Sealed: sealed}) } @@ -215,7 +248,7 @@ func (v *Vault) OwnOfNode(caller string, asked ...CredentialAsked) (OwnAnswer, e return out.Own[i].Name < out.Own[j].Name }) for _, c := range asked { - if why := v.credentialRefusal(node, c); why != "" { + if why := v.credentialRefusal(node, c, plan); why != "" { return OwnAnswer{}, fmt.Errorf("%s may not ask for the credential of %s from %s to %s on %s: %s", caller, c.Provision, c.Provider, c.ConsumerModule, c.Consumer, why) } @@ -248,9 +281,7 @@ func (v *Vault) OwnOfNode(caller string, asked ...CredentialAsked) (OwnAnswer, e if err != nil { return OwnAnswer{}, err } - if err := v.recordHanding(Handing{To: caller, Name: name, Generation: r.Generation, Outcome: "handed"}); err != nil { - return OwnAnswer{}, fmt.Errorf("%s is not handed out: who was handed it could not be recorded: %w", name, err) - } + handings = append(handings, Handing{To: caller, Name: name, Generation: r.Generation, Outcome: "handed"}) out.Credentials = append(out.Credentials, NodeCredential{CredentialAsked: c, Name: name, Generation: r.Generation, Sealed: sealed}) found = true break @@ -259,6 +290,10 @@ func (v *Vault) OwnOfNode(caller string, asked ...CredentialAsked) (OwnAnswer, e out.Missing = append(out.Missing, c) } } + if err := v.recordHandings(handings); err != nil { + return OwnAnswer{}, fmt.Errorf("nothing is handed out to %s: who was handed its %d values could not be recorded: %w", + caller, len(handings), err) + } v.log("handed %s's node-engine the %d own secrets of its modules and %d credentials (%d not held), sealed to its key", node, len(out.Own), len(out.Credentials), len(out.Missing)) return out, nil @@ -267,12 +302,12 @@ func (v *Vault) OwnOfNode(caller string, asked ...CredentialAsked) (OwnAnswer, e // handed says the plan of module on node hands it to the provisioner, so its secrets are filled from `get`, not composed. // A plan that cannot be read is an error, never "not handed": `own` refuses rather than hands a secret the provisioner // holds. -func (v *Vault) handed(node, module string) (bool, error) { - read, why := v.assignmentReader(node) +func (p *planReader) handed(module string) (bool, error) { + node := p.node + raw, why, err := p.read(module) if why != "" { return false, errors.New(why) } - raw, err := read(node + "." + module) if err != nil { return false, fmt.Errorf(notYetPrefix+"the assignment %s.%s could not be read, so whether it is handed to the "+ "provisioner is not known: %w", node, module, err) diff --git a/modules/mesh-vault/cmd/mesh-vault/ownnode_test.go b/modules/mesh-vault/cmd/mesh-vault/ownnode_test.go index 131e4455..294f61c8 100644 --- a/modules/mesh-vault/cmd/mesh-vault/ownnode_test.go +++ b/modules/mesh-vault/cmd/mesh-vault/ownnode_test.go @@ -2,9 +2,12 @@ package main import ( "errors" + "os" + "path/filepath" "strings" "testing" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/seal" ) @@ -119,3 +122,99 @@ func TestOwnAnswersAMachineItsModulesSecretsSealedToItsKey(t *testing.T) { t.Errorf("own with no reader of the assignments: %v", err) } } + +// Issue 1126, 2026-10-28 (seed-9's probe: novox's 57 own secrets and 146 credentials took 26 s, past the node-engine's +// 15 s): one answer of `own` reads each module's assignment once, and records every hand-out in one synced write. +func TestOwnReadsEachAssignmentOnceAndSyncsItsRecordOnce(t *testing.T) { + plans := assignments{"anchor.web": `{"node":"anchor","module":"web"}`, "anchor.db": `{"node":"anchor","module":"db"}`} + reads := map[string]int{} + v := newVault(t, nil) + v.Askers = Declared{Read: func(key string) ([]byte, error) { reads[key]++; return plans.read(key) }, Store: v.Store} + pub, _ := aSealingKey(t) + for _, m := range []MoveOf{ + {Node: "anchor", Module: "web", Name: "password", Origin: "made", Generation: "g-1", SealingKey: pub}, + {Node: "anchor", Module: "web", Name: "api-key", Origin: "made", Generation: "g-1", SealingKey: pub}, + {Node: "anchor", Module: "web", Name: "cookie", Origin: "made", Generation: "g-1", SealingKey: pub}, + {Node: "anchor", Module: "db", Name: "root", Origin: "made", Generation: "g-1", SealingKey: pub}, + {Provider: "anchor", Provision: "postgres", Consumer: "anchor", ConsumerModule: "web", Origin: "made", Generation: "g-1", SealingKey: pub}, + {Provider: "anchor", Provision: "redis", Consumer: "anchor", ConsumerModule: "web", Origin: "made", Generation: "g-1", SealingKey: pub}, + } { + if _, err := moveIn(t, v, "node.anchor", m, "v-"+m.Name+m.Provision); err != nil { + t.Fatal(err) + } + } + for k := range reads { + delete(reads, k) + } + syncs, handed := v.Store.syncs, len(mustHanded(t, v)) + got, err := v.OwnOfNode("node.anchor", + CredentialAsked{Role: "consumer", Provision: "postgres", Consumer: "anchor", ConsumerModule: "web", Provider: "anchor"}, + CredentialAsked{Role: "consumer", Provision: "redis", Consumer: "anchor", ConsumerModule: "web", Provider: "anchor"}) + if err != nil || len(got.Own) != 4 || len(got.Credentials) != 2 { + t.Fatalf("anchor was answered %+v, %v", got, err) + } + for key, n := range reads { + if n != 1 { + t.Errorf("the assignment %s was read %d times in one answer", key, n) + } + } + if len(reads) != 2 { + t.Errorf("the answer read %v, want each of anchor's two modules once", reads) + } + if n := v.Store.syncs - syncs; n != 1 { + t.Errorf("one answer synced the record %d times", n) + } + if n := len(mustHanded(t, v)) - handed; n != 6 { + t.Errorf("one answer recorded %d hand-outs, want 6", n) + } + // A record that cannot be written hands nothing, the batch as one. + if err := os.Remove(filepath.Join(v.dir, "store", "handed.jsonl")); err != nil { + t.Fatal(err) + } + if err := os.Mkdir(filepath.Join(v.dir, "store", "handed.jsonl"), 0o700); err != nil { + t.Fatal(err) + } + if a, err := v.OwnOfNode("node.anchor"); err == nil { + t.Errorf("handed with no record of it: %+v", a) + } +} + +func mustHanded(t *testing.T, v testVault) []Handing { + t.Helper() + h, err := v.Store.Handed(1000) + if err != nil { + t.Fatal(err) + } + return h +} + +// Issue 1126, 2026-10-28: the vault keeps the assignments mirror as its watch hands it over, so `own` and `get` read +// no assignment over the bus once the watch has caught up; before that a read goes to the bus. +func TestTheWatchedAssignmentsAnswerWithoutTheBusOnceCaughtUp(t *testing.T) { + asked := 0 + w := &WatchedAssignments{Fallback: func(key string) ([]byte, error) { asked++; return []byte(`{"from":"bus"}`), nil }} + if raw, err := w.Read("anchor.web"); err != nil || string(raw) != `{"from":"bus"}` || asked != 1 { + t.Errorf("before the watch caught up: %s, %v, %d asks", raw, err, asked) + } + for _, c := range []stdio.StateChange{ + {Key: "anchor.web", Op: "put", Value: []byte(`{"node":"anchor","module":"web"}`), Current: true}, + {Key: "anchor.db", Op: "put", Value: []byte(`{"node":"anchor","module":"db"}`), Current: true}, + } { + if err := w.Take(c); err != nil { + t.Fatal(err) + } + } + w.Caught() + _ = w.Take(stdio.StateChange{Key: "anchor.db", Op: "delete"}) + _ = w.Take(stdio.StateChange{Key: "anchor.web", Op: "put", Value: []byte(`{"node":"anchor","module":"web","v":2}`)}) + asked = 0 + for key, want := range map[string]string{"anchor.web": `{"node":"anchor","module":"web","v":2}`, "anchor.db": "", "laptop.x": ""} { + raw, err := w.Read(key) + if err != nil || string(raw) != want || (want == "") != (raw == nil) { + t.Errorf("%s read %q, %v; want %q", key, raw, err, want) + } + } + if asked != 0 { + t.Errorf("caught up, the mirror asked the bus %d times", asked) + } +} diff --git a/modules/mesh-vault/cmd/mesh-vault/store.go b/modules/mesh-vault/cmd/mesh-vault/store.go index 426132c4..430f2ea2 100644 --- a/modules/mesh-vault/cmd/mesh-vault/store.go +++ b/modules/mesh-vault/cmd/mesh-vault/store.go @@ -81,6 +81,8 @@ type Store struct { handedMu sync.Mutex rolled int + // syncs counts the record's synced appends since the vault started: how a test proves one sync per answer. + syncs int // asksMu serialises the record of asks, so two asks for one secret at once do not both pass its bounds. asksMu sync.Mutex } @@ -173,22 +175,37 @@ type Handing struct { // RecordHanding appends one handing to the store's record, synced before it returns. It says whether the // record was rolled to make room. func (s *Store) RecordHanding(h Handing) (rolled bool, err error) { - raw, err := json.Marshal(h) - if err != nil { - return false, err + return s.RecordHandings([]Handing{h}) +} + +// RecordHandings appends every handing of one answer to the store's record in one write, synced once before it returns +// (issue 1126, 2026-10-28: one sync per secret, under the one lock every node's answers share, held `own` past the +// node-engine's bound). It says whether the record was rolled to make room; nothing is written when one cannot be. +func (s *Store) RecordHandings(hs []Handing) (rolled bool, err error) { + if len(hs) == 0 { + return false, nil + } + var raw []byte + for _, h := range hs { + line, err := json.Marshal(h) + if err != nil { + return false, err + } + raw = append(append(raw, line...), '\n') } s.handedMu.Lock() defer s.handedMu.Unlock() path := filepath.Join(s.dir, "handed.jsonl") max := s.HandedBound() - if info, err := os.Stat(path); err == nil && info.Mode().IsRegular() && info.Size()+int64(len(raw))+1 > max { + if info, err := os.Stat(path); err == nil && info.Mode().IsRegular() && info.Size()+int64(len(raw)) > max { if err := s.roll(path); err != nil { return false, err } s.rolled++ rolled = true } - return rolled, appendSynced(path, append(raw, '\n')) + s.syncs++ + return rolled, appendSynced(path, raw) } // roll moves handed.jsonl to .1, .1 to .2 and so on, dropping what falls past HandedKept. @@ -266,7 +283,10 @@ func (s *Store) HandedBound() int64 { } // List is every record, by name. -func (s *Store) List() ([]Record, error) { +func (s *Store) List() ([]Record, error) { return s.ListPrefix("") } + +// ListPrefix is every record whose name begins with prefix, by name: only those files are read. +func (s *Store) ListPrefix(prefix string) ([]Record, error) { entries, err := os.ReadDir(filepath.Join(s.dir, "secrets")) if err != nil { return nil, err @@ -277,7 +297,7 @@ func (s *Store) List() ([]Record, error) { continue } name, err := url.PathUnescape(strings.TrimSuffix(e.Name(), ".json")) - if err != nil { + if err != nil || !strings.HasPrefix(name, prefix) { continue } r, ok, err := s.Read(name) diff --git a/modules/mesh-vault/cmd/mesh-vault/vault.go b/modules/mesh-vault/cmd/mesh-vault/vault.go index 05155395..1d93f4bb 100644 --- a/modules/mesh-vault/cmd/mesh-vault/vault.go +++ b/modules/mesh-vault/cmd/mesh-vault/vault.go @@ -235,11 +235,24 @@ func (v *Vault) announce(c Changed) { // recordHanding keeps who asked for what, and what they got, in the vault's own store, and says loudly // when the record had to be rolled to make room. -func (v *Vault) recordHanding(h Handing) error { - h.At = v.Now().UTC() - rolled, err := v.Store.RecordHanding(h) +func (v *Vault) recordHanding(h Handing) error { return v.recordHandings([]Handing{h}) } + +// recordHandings keeps every handing of one answer in one synced write, as recordHanding keeps one. +func (v *Vault) recordHandings(hs []Handing) error { + if len(hs) == 0 { + return nil + } + at := v.Now().UTC() + for i := range hs { + hs[i].At = at + } + rolled, err := v.Store.RecordHandings(hs) if err != nil { - v.log("the record of %s asking for %s could not be written: %v", h.To, h.Name, err) + names := make([]string, len(hs)) + for i, h := range hs { + names[i] = h.Name + } + v.log("the record of %s asking for %s could not be written: %v", hs[0].To, strings.Join(names, ", "), err) return err } if rolled { diff --git a/modules/mesh-vault/cmd/mesh-vault/vault_test.go b/modules/mesh-vault/cmd/mesh-vault/vault_test.go index 1ae1cb08..1a3b002c 100644 --- a/modules/mesh-vault/cmd/mesh-vault/vault_test.go +++ b/modules/mesh-vault/cmd/mesh-vault/vault_test.go @@ -167,8 +167,10 @@ func TestNoBucketHoldsASecret(t *testing.T) { code = []byte(strings.Replace(string(code), "stdio.State(BusAccountsState).Put(", "", 1)) code = []byte(strings.Replace(string(code), "stdio.State(BusAccountsState).Keys(", "", 1)) } - // The haves, in main.go alone: one read, one watch and their keys (own_drift), never a write. + // The haves, in main.go alone: one read, one watch and their keys (own_drift), never a write; and one watch of + // the assignments, which the vault's rules read from memory once it caught up (issue 1126). if f == "main.go" { + code = []byte(strings.Replace(string(code), "stdio.State(AssignmentsRead).Watch(", "", 1)) code = []byte(strings.Replace(string(code), "stdio.State(HaveRead).Get(", "", 1)) code = []byte(strings.Replace(string(code), "stdio.State(HaveRead).Watch(", "", 1)) code = []byte(strings.Replace(string(code), "stdio.State(HaveRead).Keys(", "", 1)) From a9d96b30be8d89820eed67cb07092756f530de99 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 14:55:36 +0100 Subject: [PATCH 2/2] mesh-vault: a watch of the assignments begun again keeps nothing an earlier attempt took, and a module the mirror lacks is read directly before a credential is refused (issue 1126, review round 1) A watch that failed partway kept what it had taken; begun again, the runtime does not hand a key deleted meanwhile, so a removed assignment read as standing and credentialRefusal could hand a consumer credential to a node no longer running its module. Each attempt now begins from nothing held, and a late delivery to an older attempt changes nothing. A 'not on the node' refusal is confirmed by a direct read first, since a watched mirror can be behind. --- modules/mesh-vault/cmd/mesh-vault/askers.go | 40 ++++++++++++++----- modules/mesh-vault/cmd/mesh-vault/main.go | 9 +++-- modules/mesh-vault/cmd/mesh-vault/ownnode.go | 34 +++++++++++++--- .../mesh-vault/cmd/mesh-vault/ownnode_test.go | 26 ++++++++++-- 4 files changed, 88 insertions(+), 21 deletions(-) diff --git a/modules/mesh-vault/cmd/mesh-vault/askers.go b/modules/mesh-vault/cmd/mesh-vault/askers.go index 5f522d3e..e5267120 100644 --- a/modules/mesh-vault/cmd/mesh-vault/askers.go +++ b/modules/mesh-vault/cmd/mesh-vault/askers.go @@ -111,6 +111,9 @@ func fieldSet(f json.RawMessage) bool { // that the module's plan there names. type Declared struct { Read AssignmentReader + // Confirm, when set, reads an assignment directly (over the bus): a module Read does not hold is confirmed through it + // before a credential is refused, since a watched mirror can be behind (issue 1126, review round 1). + Confirm AssignmentReader // Store is the vault's store, whose record of who holds a provision's one credential (pair.go) the rule reads. Store *Store } @@ -220,15 +223,34 @@ type WatchedAssignments struct { mu sync.Mutex held map[string][]byte caught bool + attempt uint64 Fallback AssignmentReader } -// Take keeps one change the watch handed over: a value put, or a key deleted. -func (w *WatchedAssignments) Take(c stdio.StateChange) error { +// WatchAttempt is one attempt at the watch: what it hands over counts only while it is the latest attempt, so a watch +// that failed partway and is begun again keeps nothing it took (a key deleted meanwhile is never handed again), and a +// late delivery to an older attempt changes nothing (issue 1126, review round 1). +type WatchAttempt struct { + w *WatchedAssignments + n uint64 +} + +// Begin starts a new attempt from nothing held: reads go to the bus until it has caught up. +func (w *WatchedAssignments) Begin() WatchAttempt { w.mu.Lock() defer w.mu.Unlock() - if w.held == nil { - w.held = map[string][]byte{} + w.attempt++ + w.held, w.caught = map[string][]byte{}, false + return WatchAttempt{w: w, n: w.attempt} +} + +// Take keeps one change the attempt handed over, a value put or a key deleted, while it is the latest attempt. +func (a WatchAttempt) Take(c stdio.StateChange) error { + w := a.w + w.mu.Lock() + defer w.mu.Unlock() + if a.n != w.attempt { + return nil } if c.Op == "delete" || c.Op == "purge" || len(c.Value) == 0 { delete(w.held, c.Key) @@ -238,14 +260,14 @@ func (w *WatchedAssignments) Take(c stdio.StateChange) error { return nil } -// Caught says the watch handed every current value: reads are answered from memory from now on. -func (w *WatchedAssignments) Caught() { +// Caught says the attempt handed every current value: reads are answered from memory from now on, if it is the latest. +func (a WatchAttempt) Caught() { + w := a.w w.mu.Lock() defer w.mu.Unlock() - if w.held == nil { - w.held = map[string][]byte{} + if a.n == w.attempt { + w.caught = true } - w.caught = true } // Read is one value of the mirror, `.`: nil when there is none. diff --git a/modules/mesh-vault/cmd/mesh-vault/main.go b/modules/mesh-vault/cmd/mesh-vault/main.go index cce1473c..7370f0af 100644 --- a/modules/mesh-vault/cmd/mesh-vault/main.go +++ b/modules/mesh-vault/cmd/mesh-vault/main.go @@ -51,7 +51,7 @@ func build() *Env { // Where it sees a consumer bind a provider (pair.go): the same mirror, every key. // The assignments as the vault's watch keeps them (watchAssignments), read over the bus until it has caught up. assigned = &WatchedAssignments{Fallback: readAssignment} - env.Vault = &Vault{Store: store, Own: own, Askers: Declared{Read: assigned.Read, Store: store}, + env.Vault = &Vault{Store: store, Own: own, Askers: Declared{Read: assigned.Read, Confirm: readAssignment, Store: store}, Assignments: Mirror{Keys: assignmentKeys, Read: readAssignment}, Emit: emit, Ask: stdio.Ask, Now: time.Now, Log: say, Accounts: busAccounts{}, Haves: readHave, HaveKeys: func() ([]string, error) { return stdio.State(HaveRead).Keys() }} @@ -68,11 +68,14 @@ func watchAssignments() bool { if assigned == nil { return true } - if err := stdio.State(AssignmentsRead).Watch("", assigned.Take); err != nil { + // Each attempt begins from nothing held: one that failed partway leaves nothing a later one would not hand again. + attempt := assigned.Begin() + if err := stdio.State(AssignmentsRead).Watch("", attempt.Take); err != nil { + assigned.Begin() // what the failed attempt took counts for nothing, and reads go to the bus say("the assignments are not watched, so each is read over the bus; begun again at the next pass: %v", err) return false } - assigned.Caught() + attempt.Caught() say("the assignments are watched: the vault's rules read them from memory") return true } diff --git a/modules/mesh-vault/cmd/mesh-vault/ownnode.go b/modules/mesh-vault/cmd/mesh-vault/ownnode.go index c93fd003..2112ed68 100644 --- a/modules/mesh-vault/cmd/mesh-vault/ownnode.go +++ b/modules/mesh-vault/cmd/mesh-vault/ownnode.go @@ -92,6 +92,12 @@ func (v *Vault) credentialRefusal(node string, c CredentialAsked, plan *planRead if err != nil { return fmt.Sprintf(notYetPrefix+"the assignment %s.%s could not be read: %v", node, c.ConsumerModule, err) } + if raw == nil { + // Confirmed directly before refusing: the mirror may not hold a module assigned a moment ago. + if raw, err = plan.confirmed(c.ConsumerModule); err != nil { + return fmt.Sprintf(notYetPrefix+"the assignment %s.%s could not be read: %v", node, c.ConsumerModule, err) + } + } if raw == nil { return fmt.Sprintf("%s is not on %s: neither assigned there nor required by a module there", c.ConsumerModule, node) } @@ -108,10 +114,11 @@ func (v *Vault) credentialRefusal(node string, c CredentialAsked, plan *planRead // planReader reads the assignments of one machine's modules for one answer of `own`, each module's once (issue 1126, // 2026-10-28: one read per own secret and per credential, 203 on novox, held the answer past the node-engine's bound). type planReader struct { - node string - reader AssignmentReader - why string - got map[string]planRead + node string + reader AssignmentReader + confirm AssignmentReader + why string + got map[string]planRead } type planRead struct { @@ -121,7 +128,11 @@ type planRead struct { func (v *Vault) plansOf(node string) *planReader { read, why := v.assignmentReader(node) - return &planReader{node: node, reader: read, why: why, got: map[string]planRead{}} + var confirm AssignmentReader + if d, ok := v.Askers.(Declared); ok && d.Confirm != nil { + confirm = standing(d.Confirm) + } + return &planReader{node: node, confirm: confirm, reader: read, why: why, got: map[string]planRead{}} } // read is module's assignment on the machine, nil when there is none, or why the vault reads none. @@ -137,6 +148,19 @@ func (p *planReader) read(module string) ([]byte, string, error) { return r.raw, "", r.err } +// confirmed is module's assignment read directly, for one the reader does not hold (a watched mirror can be behind), +// nil when there is none or nothing reads directly. +func (p *planReader) confirmed(module string) ([]byte, error) { + if p.confirm == nil { + return nil, nil + } + raw, err := p.confirm(p.node + "." + module) + if err == nil && raw != nil { + p.got[module] = planRead{raw: raw} + } + return raw, err +} + // assignmentReader is the vault's reader of the assignments, or why it has none: the rule `own` keeps reads them, // and without them it refuses rather than hands (a rule that cannot be read is no rule). func (v *Vault) assignmentReader(node string) (AssignmentReader, string) { diff --git a/modules/mesh-vault/cmd/mesh-vault/ownnode_test.go b/modules/mesh-vault/cmd/mesh-vault/ownnode_test.go index 294f61c8..5eb7a682 100644 --- a/modules/mesh-vault/cmd/mesh-vault/ownnode_test.go +++ b/modules/mesh-vault/cmd/mesh-vault/ownnode_test.go @@ -167,6 +167,14 @@ func TestOwnReadsEachAssignmentOnceAndSyncsItsRecordOnce(t *testing.T) { if n := len(mustHanded(t, v)) - handed; n != 6 { t.Errorf("one answer recorded %d hand-outs, want 6", n) } + // Review round 1, note (b): a consumer module the mirror does not hold yet (its watch behind) is read directly before + // the credential is refused. + v.Askers = Declared{Read: func(string) ([]byte, error) { return nil, nil }, Confirm: plans.read, Store: v.Store} + if got, err := v.OwnOfNode("node.anchor", CredentialAsked{Role: "consumer", Provision: "postgres", Consumer: "anchor", + ConsumerModule: "web", Provider: "anchor"}); err != nil || len(got.Credentials) != 1 { + t.Errorf("a module the mirror does not hold yet, read directly: %+v, %v", got, err) + } + v.Askers = Declared{Read: plans.read, Store: v.Store} // A record that cannot be written hands nothing, the batch as one. if err := os.Remove(filepath.Join(v.dir, "store", "handed.jsonl")); err != nil { t.Fatal(err) @@ -196,17 +204,27 @@ func TestTheWatchedAssignmentsAnswerWithoutTheBusOnceCaughtUp(t *testing.T) { if raw, err := w.Read("anchor.web"); err != nil || string(raw) != `{"from":"bus"}` || asked != 1 { t.Errorf("before the watch caught up: %s, %v, %d asks", raw, err, asked) } + // A first watch takes two values and fails before it caught up; a late value still arrives on it. + first := w.Begin() for _, c := range []stdio.StateChange{ {Key: "anchor.web", Op: "put", Value: []byte(`{"node":"anchor","module":"web"}`), Current: true}, {Key: "anchor.db", Op: "put", Value: []byte(`{"node":"anchor","module":"db"}`), Current: true}, } { - if err := w.Take(c); err != nil { + if err := first.Take(c); err != nil { t.Fatal(err) } } - w.Caught() - _ = w.Take(stdio.StateChange{Key: "anchor.db", Op: "delete"}) - _ = w.Take(stdio.StateChange{Key: "anchor.web", Op: "put", Value: []byte(`{"node":"anchor","module":"web","v":2}`)}) + // Review round 1: the watch begun again holds anchor.db no more (removed meanwhile); what the first took is gone. + second := w.Begin() + _ = second.Take(stdio.StateChange{Key: "anchor.web", Op: "put", Value: []byte(`{"node":"anchor","module":"web"}`), Current: true}) + _ = first.Take(stdio.StateChange{Key: "anchor.db", Op: "put", Value: []byte(`{"node":"anchor","module":"db"}`)}) + first.Caught() + if raw, _ := w.Read("anchor.db"); string(raw) != `{"from":"bus"}` { + t.Errorf("the failed watch's caught up counted: %s", raw) + } + second.Caught() + _ = first.Take(stdio.StateChange{Key: "anchor.db", Op: "put", Value: []byte(`{"node":"anchor","module":"db"}`)}) + _ = second.Take(stdio.StateChange{Key: "anchor.web", Op: "put", Value: []byte(`{"node":"anchor","module":"web","v":2}`)}) asked = 0 for key, want := range map[string]string{"anchor.web": `{"node":"anchor","module":"web","v":2}`, "anchor.db": "", "laptop.x": ""} { raw, err := w.Read(key)