From a9d96b30be8d89820eed67cb07092756f530de99 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 14:55:36 +0100 Subject: [PATCH] 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)