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)
mesh/delivery merged: source.main moved to novox/mesh-delivery@52a1570b3e3c
mesh/delivery-group group fix/held-module-and-assigner-contract merged: every member is merged
mesh/check checking 8fe6442c4754 on c104e7500d21 | novox/mesh-delivery#118 in group fix/held-module-and-assigner-contract
mesh/delivery merged: source.main moved to novox/mesh-delivery@52a1570b3e3c
mesh/delivery-group group fix/held-module-and-assigner-contract merged: every member is merged
mesh/check checking 8fe6442c4754 on c104e7500d21 | novox/mesh-delivery#118 in group fix/held-module-and-assigner-contract
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.
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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:])
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user