From 8fe6442c4754dac230e1bc69ddbf2a0f9236cbc8 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 21:27:35 +0100 Subject: [PATCH] Clear a refused pin's condition once the controller's pins agree, and take over a previous process's conditions only once the findings are read (issue 774) A pin made by hand, as the condition's Needs asks, changed nothing the assigner looked at, so its urgent condition never cleared, and a restart left every one raised. The assigner now reads the controller's pins (placement-export, asked again only after the controller's compositions change, never on a timer) and clears a pin's condition once they agree with the placement's after being seen differing, as a module's place already did. The first pass takes over the call conditions a previous process raised, pins and places alike, and retries at the next pass when the findings cannot be read. --- .../cmd/mesh-delivery-assigner/pass.go | 6 +- .../cmd/mesh-delivery-assigner/pass_test.go | 108 ++++++++++ .../cmd/mesh-delivery-assigner/seat.go | 191 ++++++++++++++---- .../cmd/mesh-delivery-assigner/world.go | 9 + 4 files changed, 268 insertions(+), 46 deletions(-) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index 76d1583..a9f304b 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -58,9 +58,11 @@ type passer struct { // nodeFailed is, by node and module, the words of the condition last raised for a failure that is the node's own (ADR 0557 // §1, issue 1496): raised once per words, cleared once when it ends. nodeFailed map[string]string - // callSaid is, by the key of a refused controller call's condition, what this process raised (issue 774); nil until the - // first pass after a start, which clears what a previous process raised for a pair the two sources agree on. + // callSaid is, by the key of a refused controller call's condition, what is raised (issue 774): by this process, or by + // a previous one and taken over from the findings. callSaid map[string]*callSaid + // callsSwept says the first pass took over the refused calls' conditions a previous process raised (sweepCalls). + callsSwept bool // The seat mesh-placement's holder (seat.go, ADR 0567): its calls of the controller (nil: none is made), how it says // the seat's events (nil: none is said), the proposals it said, and what Intended computed by proposal and base. seat *seatActs diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index bed2ef1..0cba96b 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -7,6 +7,7 @@ import ( "fmt" "slices" "strings" + "sync" "testing" "time" @@ -976,9 +977,28 @@ func aPlacedNodesWantsNameTheBuildAloneAndItsYesComesFromItsHave(t *testing.T) { type refusingBus struct { fakeBus refuse map[string]string + // pins are the controller's pins its placement-export answers, by node. + mu sync.Mutex + pins map[string]map[string]delivery.Provider +} + +func (b *refusingBus) setPins(node string, pins map[string]delivery.Provider) { + b.mu.Lock() + defer b.mu.Unlock() + if b.pins == nil { + b.pins = map[string]map[string]delivery.Provider{} + } + b.pins[node] = pins } func (b *refusingBus) Call(key string, body any) (json.RawMessage, error) { + if key == exportCall { + b.mu.Lock() + defer b.mu.Unlock() + node, _ := body.(map[string]any)["node"].(string) + answer, _ := json.Marshal(map[string]delivery.Placement{node: {V: delivery.Version, Node: node, Pins: b.pins[node]}}) + return json.Marshal(map[string]any{"ok": true, "answer": json.RawMessage(answer)}) + } for verb, words := range b.refuse { if strings.HasSuffix(key, "."+verb) { return nil, errors.New(words) @@ -1040,3 +1060,91 @@ func TestARefusedControllerCallIsACondition(t *testing.T) { t.Fatalf("said again after the clear: %+v", conds) } } + +// Issue 774: a refused pin's condition, and one a previous process raised, clear once the pair agrees after it was seen differing: +// the controller's pins (its placement-export, read again when its compositions change) against the placement's, the controller's +// assignments against the placement for a module's place. A start whose findings cannot be read takes them over at a later pass. +func TestARefusedCallClearsWhenTheSourcesAgree(t *testing.T) { + store := delivery.Provider{Node: "anchor", Module: "store"} + pinDB := assign.ControllerCall{Verb: "pin", Node: "laptop", Module: "db", + Args: map[string]any{"node": "laptop", "provision": "db", "from": "anchor", "module": "store"}} + placeWeb := assign.ControllerCall{Verb: "assign", Node: "anchor", Module: "web", Args: map[string]any{"node": "anchor", "module": "web"}} + webDiffers := []assign.Pair{{Node: "anchor", Module: "web", Only: assign.OnlyPlacement}} + for name, c := range map[string]struct { + call *assign.ControllerCall // refused in this process; nil: raised by a previous one + findings []string // the findings a previous process left + keysFail int // how many first reads of the findings fail + differ []assign.Pair // the pairs the two sources name apart, while they differ + agree func(*refusingBus) // the person's hand call + key string + }{ + "a pin refused, pinned by hand": {call: &pinDB, agree: func(b *refusingBus) { b.setPins("laptop", map[string]delivery.Provider{"db": store}) }, + key: "mesh-delivery-assigner.controller-pin.laptop.db"}, + "a place refused, assigned by hand": {call: &placeWeb, differ: webDiffers, agree: func(*refusingBus) {}, + key: "mesh-delivery-assigner.controller-call.anchor.web"}, + "a pin raised before a start": {findings: []string{"module.laptop.db.controller-pin"}, keysFail: 1, + agree: func(b *refusingBus) { b.setPins("laptop", map[string]delivery.Provider{"db": store}) }, + key: "mesh-delivery-assigner.controller-pin.laptop.db"}, + "a place raised before a start": {findings: []string{"module.anchor.web.controller-call"}, differ: webDiffers, agree: func(*refusingBus) {}, + key: "mesh-delivery-assigner.controller-call.anchor.web"}, + } { + t.Run(name, func(t *testing.T) { + p := newPasser(buckets{}, time.Date(2026, 10, 30, 21, 4, 0, 0, time.UTC)) + var conds []asks.Condition + p.say = func(c asks.Condition) error { conds = append(conds, c); return nil } + fails := c.keysFail + p.findings.Keys = func() ([]string, error) { + if fails > 0 { + fails-- + return nil, errors.New("the bucket is not there yet") + } + return c.findings, nil + } + bus := &refusingBus{refuse: map[string]string{"pin": "refused", "assign": "refused"}} + woke := make(chan struct{}, 16) + p.seat = newSeatActs(bus, func() { woke <- struct{}{} }, func(string, ...any) {}, func() bool { return false }) + // The placement pins db on laptop; the controller does not. + raw, _ := delivery.EncodePlacement(delivery.Placement{V: delivery.Version, Node: "laptop", Pins: map[string]delivery.Provider{"db": store}}) + key, _ := delivery.PlacementKey("laptop") + p.w.put(delivery.PlacementBucket, key, raw) + // pass is a pass, then one more for each background read it started. + pass := func(differ []assign.Pair) { + p.sayCalls(differ) + for { + select { + case <-woke: + p.sayCalls(differ) + case <-time.After(20 * time.Millisecond): + return + } + } + } + if c.call != nil { + p.seat.callAll([]assign.ControllerCall{*c.call}) + <-woke + } + pass(c.differ) + pass(c.differ) + for _, cond := range conds { + if cond.State == asks.StateCleared { + t.Fatalf("cleared while the sources differ: %+v", conds) + } + } + c.agree(bus) + p.w.put(delivery.ControllerComposedBucket, "laptop.x", []byte("moved")) // a pin changes a composition + pass(nil) + if len(conds) == 0 { + t.Fatal("nothing was raised") + } + last := conds[len(conds)-1] + if last.Key != c.key || last.State != asks.StateCleared || !strings.Contains(last.Resolved, "agree") { + t.Fatalf("the hand call does not clear it: %+v", conds) + } + n := len(conds) + pass(nil) + if len(conds) != n { + t.Errorf("said again after the clear: %+v", conds[n:]) + } + }) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go index 7de5f98..02c4adf 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go @@ -71,6 +71,8 @@ type seatActs struct { // outcomes is, by the key of a call's condition, the last call made for its pair once answered or refused: refused // "" when the controller answered it (issue 774: a refused call is a condition, never only a journal line). outcomes map[string]callOutcome + // pins is, by node, the controller's pins as its placement-export last answered them (pinsOf). + pins map[string]pinRead // rec records the seat's acts in the hand-act log (ADR 0565 §11). rec handacts.Recorder } @@ -90,7 +92,7 @@ type exportCheck struct { func newSeatActs(bus caller, wake func(), logf func(string, ...any), leaseAway func() bool) *seatActs { return &seatActs{bus: bus, wake: wake, logf: logf, export: map[string]exportCheck{}, calls: map[string]string{}, - outcomes: map[string]callOutcome{}, rec: handacts.Recorder{Ask: bus.Call, LeaseAway: leaseAway}} + outcomes: map[string]callOutcome{}, pins: map[string]pinRead{}, rec: handacts.Recorder{Ask: bus.Call, LeaseAway: leaseAway}} } // exportOf is the export read again for a proposal, asked once in the background. @@ -193,66 +195,86 @@ func (a *seatActs) outcomesNow() map[string]callOutcome { return maps.Clone(a.outcomes) } +// pinRead is the controller's pins of one node as its placement-export answered them, for the generation of the +// controller's compositions it was asked at: a pin made at the controller changes a composition, which the assigner +// watches, so a read is asked again only after such a change, never on a timer. +type pinRead struct { + gen uint64 + running bool + known bool + pins map[string]delivery.Provider +} + +// pinsOf is the controller's pins of node as last read, and whether a read answered; a read is asked in the background +// when none was made at gen, and wakes a pass when it answers. +func (a *seatActs) pinsOf(node string, gen uint64) (map[string]delivery.Provider, bool) { + if a == nil { + return nil, false + } + a.mu.Lock() + defer a.mu.Unlock() + r, ok := a.pins[node] + if !ok || (r.gen != gen && !r.running) { + r.gen, r.running = gen, true + a.pins[node] = r + go func() { + defer a.wake() + raw, err := a.bus.Call(exportCall, map[string]any{"node": node}) + var pins map[string]delivery.Provider + var known bool + if err == nil { + var export map[string]delivery.Placement + if export, err = assign.ExportOf(raw); err == nil { + pins, known = export[node].Pins, true + } + } + if err != nil { + a.logf("the controller's pins of %s were not read, so a refused pin's condition there waits: %v", node, err) + } + a.mu.Lock() + now := a.pins[node] + now.running = false + if known { + now.pins, now.known = pins, true + } + a.pins[node] = now + a.mu.Unlock() + }() + } + return r.pins, r.known +} + // sayCalls makes a refused call of the controller loud (issue 774): the condition of each pair whose last call was refused -// is raised once per refusal, naming the verb, the node, the module and the controller's words; it is cleared when a later -// call for the pair is answered, or, for a module's place, once the pair differed after the refusal and the two sources now -// agree (a person called the verb by hand). The first pass after a start clears what a previous process raised for a pair -// the two sources agree on; a pin's stays, since nothing here can tell it was made since. A say that fails is made again at -// the next pass. +// is raised once per refusal, naming the verb, the node, the module and the controller's words. It is cleared when a later +// call for the pair is answered, or once the pair, seen differing since the refusal, agrees between the two sources (a +// person made the call by hand): for a module's place the controller's assignments and the placement, for a pin the +// controller's pins (its placement-export, read again after its compositions change) and the placement's. The first pass +// after a start takes over what a previous process raised (its findings) and clears each whose pair agrees; until the +// findings are read it is tried again at every pass. A say that fails is made again at the next pass. func (p *passer) sayCalls(differ []assign.Pair) { if p.say == nil { return } + if p.callSaid == nil { + p.callSaid = map[string]*callSaid{} + } differs := map[string]bool{} for _, d := range differ { differs[d.Node+"."+d.Module] = true } outcomes := p.seat.outcomesNow() - if p.callSaid == nil { - p.callSaid = map[string]*callSaid{} - if p.findings.Keys != nil { - if keys, err := p.findings.Keys(); err == nil { - for _, k := range keys { - pair, ok := strings.CutSuffix(strings.TrimPrefix(k, "module."), "."+assign.KindControllerCall) - node, module, two := strings.Cut(pair, ".") - if !ok || !two || !strings.HasPrefix(k, "module.") || differs[pair] { - continue - } - c := assign.ControllerCall{Verb: "assign", Node: node, Module: module} - if _, made := outcomes[assign.CallConditionKey(c)]; made { - continue - } - if err := p.say(assign.CallRefusedCondition(c, "", false, "the controller's assignments and the placement agree on "+module+" on "+node+".")); err != nil { - p.logf("the refused call's condition of %s on %s could not be cleared: %v", module, node, err) - } - } - } - } + if !p.callsSwept { + p.callsSwept = p.sweepCalls(outcomes) } for key, o := range outcomes { said := p.callSaid[key] - pair := o.call.Node + "." + o.call.Module switch { - case o.refused != "" && (said == nil || said.refused != o.refused): + case o.refused != "" && (said == nil || said.adopted || said.refused != o.refused): if err := p.say(assign.CallRefusedCondition(o.call, o.refused, true, "")); err != nil { p.logf("that %s was refused could not be said, and is said at the next pass: %v", o.call.Key(), err) continue } - p.callSaid[key] = &callSaid{refused: o.refused, differed: differs[pair]} - case o.refused != "" && o.call.CallKind() == assign.KindControllerCall: - if differs[pair] { - said.differed = true - continue - } - if !said.differed { - continue // the pair was not seen differing since the refusal: the world may not hold the write yet - } - if err := p.say(assign.CallRefusedCondition(o.call, "", false, "the controller's assignments and the placement agree on "+o.call.Module+" on "+o.call.Node+".")); err != nil { - p.logf("the condition of %s could not be cleared, and is cleared at the next pass: %v", o.call.Key(), err) - continue - } - p.seat.forgetOutcome(key, o) - delete(p.callSaid, key) + p.callSaid[key] = &callSaid{call: o.call, refused: o.refused} case o.refused == "" && said != nil: if err := p.say(assign.CallRefusedCondition(o.call, "", false, o.call.Key()+" was answered.")); err != nil { p.logf("the condition of %s could not be cleared, and is cleared at the next pass: %v", o.call.Key(), err) @@ -261,12 +283,93 @@ func (p *passer) sayCalls(differ []assign.Pair) { delete(p.callSaid, key) } } + var placements map[string]delivery.Placement + for key, said := range p.callSaid { + c := said.call + var apart bool + if c.CallKind() == assign.KindControllerCall { + apart = differs[c.Node+"."+c.Module] + } else { + theirs, known := p.seat.pinsOf(c.Node, p.w.bucketGen(delivery.ControllerComposedBucket)) + if !known { + continue + } + if placements == nil { + placements, _ = p.seatPlacements() + } + ours, has := placements[c.Node].Pins[c.Module] + to, had := theirs[c.Module] + apart = has != had || ours != to + } + if apart { + said.differed = true + continue + } + if !said.differed { + continue // not seen differing since the refusal: the world may not hold the write yet + } + if err := p.say(assign.CallRefusedCondition(c, "", false, agreeWords(c))); err != nil { + p.logf("the condition of %s could not be cleared, and is cleared at the next pass: %v", c.Key(), err) + continue + } + if o, ok := outcomes[key]; ok { + p.seat.forgetOutcome(key, o) + } + delete(p.callSaid, key) + } } -// callSaid is the refusal a raised call's condition said, and whether its pair was seen differing since. +// sweepCalls takes over the refused calls' conditions a previous process raised, from the findings: each is followed as +// one seen differing, so it clears at the first pass its pair agrees. It answers whether the findings were read. +func (p *passer) sweepCalls(outcomes map[string]callOutcome) bool { + if p.findings.Keys == nil { + return true + } + keys, err := p.findings.Keys() + if err != nil { + p.logf("the findings were not read, so the refused calls a previous process raised are taken over at the next pass: %v", err) + return false + } + for _, k := range keys { + rest, ok := strings.CutPrefix(k, "module.") + if !ok { + continue + } + for _, kind := range []string{assign.KindControllerCall, assign.KindControllerPin} { + pair, ok := strings.CutSuffix(rest, "."+kind) + node, module, two := strings.Cut(pair, ".") + if !ok || !two { + continue + } + verb := "assign" + if kind == assign.KindControllerPin { + verb = "pin" + } + c := assign.ControllerCall{Verb: verb, Node: node, Module: module} + key := assign.CallConditionKey(c) + if _, made := outcomes[key]; made || p.callSaid[key] != nil { + continue + } + p.callSaid[key] = &callSaid{call: c, differed: true, adopted: true} + } + } + return true +} + +func agreeWords(c assign.ControllerCall) string { + if c.CallKind() == assign.KindControllerPin { + return "the controller's pins and the placement agree on " + c.Module + " for " + c.Node + "." + } + return "the controller's assignments and the placement agree on " + c.Module + " on " + c.Node + "." +} + +// callSaid is a raised call's condition: its call, the refusal it said, whether its pair was seen differing since, and +// whether it was taken over from a previous process. type callSaid struct { + call assign.ControllerCall refused string differed bool + adopted bool } // forgetOutcome drops a refused call's outcome once its condition is cleared, unless a later call replaced it. diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index 1acdc2b..c848c28 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -100,6 +100,7 @@ func (w *world) put(bucket, key string, value []byte) { if w.values[bucket] == nil { w.values[bucket] = map[string][]byte{} } + w.changes[bucket]++ // the bucket's generation (bucketGen), a delete included if value == nil { delete(w.values[bucket], key) return @@ -662,6 +663,14 @@ func (w *world) answer(key string) (delivery.Answer, bool) { return a, err == nil } +// bucketGen is how many times any key of the bucket was written or deleted in this process: a read of what it shows +// elsewhere is asked again only after it moves. +func (w *world) bucketGen(bucket string) uint64 { + w.mu.Lock() + defer w.mu.Unlock() + return w.changes[bucket] +} + // changed is how many values the key has taken in this process (see world.changes). func (w *world) changed(bucket, key string) uint64 { w.mu.Lock()