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)
mesh/delivery merged: source.main moved to novox/mesh-catalog@85bbeb37e3e0
mesh/delivery-group group fix/1126-vault-own merged: every member is merged
mesh/check pass a9d96b30be8d on 69056948bfa8 | unit 1639 run, 4201 kept, 46 skipped | integration 1674 run, 4215 kept, 47 skipped | over bound: unit, …
mesh/delivery merged: source.main moved to novox/mesh-catalog@85bbeb37e3e0
mesh/delivery-group group fix/1126-vault-own merged: every member is merged
mesh/check pass a9d96b30be8d on 69056948bfa8 | unit 1639 run, 4201 kept, 46 skipped | integration 1674 run, 4215 kept, 47 skipped | over bound: unit, …
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.
This commit is contained in:
@@ -111,6 +111,9 @@ func fieldSet(f json.RawMessage) bool {
|
|||||||
// that the module's plan there names.
|
// that the module's plan there names.
|
||||||
type Declared struct {
|
type Declared struct {
|
||||||
Read AssignmentReader
|
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 is the vault's store, whose record of who holds a provision's one credential (pair.go) the rule reads.
|
||||||
Store *Store
|
Store *Store
|
||||||
}
|
}
|
||||||
@@ -220,15 +223,34 @@ type WatchedAssignments struct {
|
|||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
held map[string][]byte
|
held map[string][]byte
|
||||||
caught bool
|
caught bool
|
||||||
|
attempt uint64
|
||||||
Fallback AssignmentReader
|
Fallback AssignmentReader
|
||||||
}
|
}
|
||||||
|
|
||||||
// Take keeps one change the watch handed over: a value put, or a key deleted.
|
// WatchAttempt is one attempt at the watch: what it hands over counts only while it is the latest attempt, so a watch
|
||||||
func (w *WatchedAssignments) Take(c stdio.StateChange) error {
|
// 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()
|
w.mu.Lock()
|
||||||
defer w.mu.Unlock()
|
defer w.mu.Unlock()
|
||||||
if w.held == nil {
|
w.attempt++
|
||||||
w.held = map[string][]byte{}
|
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 {
|
if c.Op == "delete" || c.Op == "purge" || len(c.Value) == 0 {
|
||||||
delete(w.held, c.Key)
|
delete(w.held, c.Key)
|
||||||
@@ -238,14 +260,14 @@ func (w *WatchedAssignments) Take(c stdio.StateChange) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Caught says the watch handed every current value: reads are answered from memory from now on.
|
// Caught says the attempt handed every current value: reads are answered from memory from now on, if it is the latest.
|
||||||
func (w *WatchedAssignments) Caught() {
|
func (a WatchAttempt) Caught() {
|
||||||
|
w := a.w
|
||||||
w.mu.Lock()
|
w.mu.Lock()
|
||||||
defer w.mu.Unlock()
|
defer w.mu.Unlock()
|
||||||
if w.held == nil {
|
if a.n == w.attempt {
|
||||||
w.held = map[string][]byte{}
|
w.caught = true
|
||||||
}
|
}
|
||||||
w.caught = true
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Read is one value of the mirror, `<node>.<module>`: nil when there is none.
|
// Read is one value of the mirror, `<node>.<module>`: nil when there is none.
|
||||||
|
|||||||
@@ -51,7 +51,7 @@ func build() *Env {
|
|||||||
// Where it sees a consumer bind a provider (pair.go): the same mirror, every key.
|
// 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.
|
// The assignments as the vault's watch keeps them (watchAssignments), read over the bus until it has caught up.
|
||||||
assigned = &WatchedAssignments{Fallback: readAssignment}
|
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,
|
Assignments: Mirror{Keys: assignmentKeys, Read: readAssignment}, Emit: emit, Ask: stdio.Ask,
|
||||||
Now: time.Now, Log: say, Accounts: busAccounts{}, Haves: readHave,
|
Now: time.Now, Log: say, Accounts: busAccounts{}, Haves: readHave,
|
||||||
HaveKeys: func() ([]string, error) { return stdio.State(HaveRead).Keys() }}
|
HaveKeys: func() ([]string, error) { return stdio.State(HaveRead).Keys() }}
|
||||||
@@ -68,11 +68,14 @@ func watchAssignments() bool {
|
|||||||
if assigned == nil {
|
if assigned == nil {
|
||||||
return true
|
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)
|
say("the assignments are not watched, so each is read over the bus; begun again at the next pass: %v", err)
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
assigned.Caught()
|
attempt.Caught()
|
||||||
say("the assignments are watched: the vault's rules read them from memory")
|
say("the assignments are watched: the vault's rules read them from memory")
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -92,6 +92,12 @@ func (v *Vault) credentialRefusal(node string, c CredentialAsked, plan *planRead
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Sprintf(notYetPrefix+"the assignment %s.%s could not be read: %v", node, c.ConsumerModule, err)
|
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 {
|
if raw == nil {
|
||||||
return fmt.Sprintf("%s is not on %s: neither assigned there nor required by a module there", c.ConsumerModule, node)
|
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,
|
// 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).
|
// 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 {
|
type planReader struct {
|
||||||
node string
|
node string
|
||||||
reader AssignmentReader
|
reader AssignmentReader
|
||||||
why string
|
confirm AssignmentReader
|
||||||
got map[string]planRead
|
why string
|
||||||
|
got map[string]planRead
|
||||||
}
|
}
|
||||||
|
|
||||||
type planRead struct {
|
type planRead struct {
|
||||||
@@ -121,7 +128,11 @@ type planRead struct {
|
|||||||
|
|
||||||
func (v *Vault) plansOf(node string) *planReader {
|
func (v *Vault) plansOf(node string) *planReader {
|
||||||
read, why := v.assignmentReader(node)
|
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.
|
// 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
|
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,
|
// 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).
|
// 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) {
|
func (v *Vault) assignmentReader(node string) (AssignmentReader, string) {
|
||||||
|
|||||||
@@ -167,6 +167,14 @@ func TestOwnReadsEachAssignmentOnceAndSyncsItsRecordOnce(t *testing.T) {
|
|||||||
if n := len(mustHanded(t, v)) - handed; n != 6 {
|
if n := len(mustHanded(t, v)) - handed; n != 6 {
|
||||||
t.Errorf("one answer recorded %d hand-outs, want 6", n)
|
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.
|
// 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 {
|
if err := os.Remove(filepath.Join(v.dir, "store", "handed.jsonl")); err != nil {
|
||||||
t.Fatal(err)
|
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 {
|
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)
|
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{
|
for _, c := range []stdio.StateChange{
|
||||||
{Key: "anchor.web", Op: "put", Value: []byte(`{"node":"anchor","module":"web"}`), Current: true},
|
{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},
|
{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)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
w.Caught()
|
// Review round 1: the watch begun again holds anchor.db no more (removed meanwhile); what the first took is gone.
|
||||||
_ = w.Take(stdio.StateChange{Key: "anchor.db", Op: "delete"})
|
second := w.Begin()
|
||||||
_ = w.Take(stdio.StateChange{Key: "anchor.web", Op: "put", Value: []byte(`{"node":"anchor","module":"web","v":2}`)})
|
_ = 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
|
asked = 0
|
||||||
for key, want := range map[string]string{"anchor.web": `{"node":"anchor","module":"web","v":2}`, "anchor.db": "", "laptop.x": ""} {
|
for key, want := range map[string]string{"anchor.web": `{"node":"anchor","module":"web","v":2}`, "anchor.db": "", "laptop.x": ""} {
|
||||||
raw, err := w.Read(key)
|
raw, err := w.Read(key)
|
||||||
|
|||||||
Reference in New Issue
Block a user