The assigner writes wants only for a pair both the controller's assignments and the catalogue's placement name, and says each pair one names alone (issue 1126, ADR 0549 §5)
This commit is contained in:
@@ -59,6 +59,7 @@ var readBuckets = []string{
|
||||
delivery.CandidatesBucket,
|
||||
delivery.ManifestsBucket,
|
||||
delivery.ModulesBucket,
|
||||
delivery.PlacementBucket,
|
||||
delivery.PresenceBucket,
|
||||
delivery.HaveBucket,
|
||||
delivery.StepsBucket(goBuildModule),
|
||||
@@ -169,6 +170,7 @@ var watched = []string{
|
||||
delivery.BuilderBuildsBucket,
|
||||
delivery.CandidatesBucket,
|
||||
delivery.ModulesBucket,
|
||||
delivery.PlacementBucket,
|
||||
delivery.PresenceBucket,
|
||||
delivery.HaveBucket,
|
||||
delivery.AssignerAsksBucket,
|
||||
|
||||
@@ -97,6 +97,7 @@ func (p *passer) wants(ctx context.Context) {
|
||||
}
|
||||
}
|
||||
wanted := p.w.wantsOf()
|
||||
frozen := p.sayPlacement(p.w.placementDiffers())
|
||||
puts := p.putsByModule()
|
||||
p.staleAsks(ctx, trunk)
|
||||
names := map[string]bool{}
|
||||
@@ -183,6 +184,9 @@ func (p *passer) wants(ctx context.Context) {
|
||||
p.recordAsk(ctx, name, fp, need)
|
||||
}
|
||||
for node, want := range out.Wants {
|
||||
if frozen[node+"."+name] {
|
||||
continue // one source alone names it (ADR 0549 §5): no want, and the one written before stays
|
||||
}
|
||||
key, err := delivery.WantKey(node, name)
|
||||
if err != nil {
|
||||
p.logf("no want of %s on %s is written: %v", name, node, err)
|
||||
@@ -728,3 +732,39 @@ func (p *passer) backupAsk(ctx context.Context, key, node, module string, want d
|
||||
}
|
||||
p.said[c.Key] = true
|
||||
}
|
||||
|
||||
// sayPlacement raises, once, the condition of every pair the controller's assignments and the catalogue's placement do not
|
||||
// both name, clears it once they agree, and answers those pairs by node and module (ADR 0549 §5): the pass writes no want
|
||||
// for them and keeps the one written before. A say that fails is made again at the next pass.
|
||||
func (p *passer) sayPlacement(differ []assign.Pair) map[string]bool {
|
||||
frozen := map[string]bool{}
|
||||
now := map[string]bool{}
|
||||
for _, d := range differ {
|
||||
frozen[d.Node+"."+d.Module] = true
|
||||
c := assign.PlacementCondition(d, true)
|
||||
now[c.Key] = true
|
||||
if p.said[c.Key+".raised"] {
|
||||
continue
|
||||
}
|
||||
p.logf("%s on %s is named by %s alone: no want is written for it there until both sources name it (ADR 0549 §5)", d.Module, d.Node, d.Only)
|
||||
if err := p.say(c); err != nil {
|
||||
p.logf("the placement condition of %s on %s could not be said, and is said at the next pass: %v", d.Module, d.Node, err)
|
||||
continue
|
||||
}
|
||||
p.said[c.Key+".raised"] = true
|
||||
}
|
||||
prefix := delivery.AssignerModule + ".placement."
|
||||
for key := range p.said {
|
||||
c, raised := strings.CutSuffix(key, ".raised")
|
||||
if !raised || !strings.HasPrefix(c, prefix) || now[c] {
|
||||
continue
|
||||
}
|
||||
node, module, _ := strings.Cut(strings.TrimPrefix(c, prefix), ".")
|
||||
if err := p.say(assign.PlacementCondition(assign.Pair{Node: node, Module: module}, false)); err != nil {
|
||||
p.logf("the placement condition of %s on %s could not be cleared, and is cleared at the next pass: %v", module, node, err)
|
||||
continue
|
||||
}
|
||||
delete(p.said, key)
|
||||
}
|
||||
return frozen
|
||||
}
|
||||
|
||||
@@ -750,3 +750,47 @@ func TestARecordAskOfAHeldBackNodeSurvivesPassesAndItsYesMovesTheWant(t *testing
|
||||
t.Error("the answered record ask was taken away")
|
||||
}
|
||||
}
|
||||
|
||||
// 1126 (ADR 0549 §5): until the third milestone a pair the controller's assignments and the catalogue's placement do not
|
||||
// both name gets no want and one condition, raised once and cleared once both name it; before any placement was read the
|
||||
// assignments are the one source and nothing is raised.
|
||||
func TestAPairOneSourceNamesAloneGetsNoWantAndOneCondition(t *testing.T) {
|
||||
p := newPasser(buckets{}, time.Date(2026, 10, 26, 18, 0, 0, 0, time.UTC))
|
||||
var said []asks.Condition
|
||||
p.say = func(c asks.Condition) error { said = append(said, c); return nil }
|
||||
assigned := func(node, module string) {
|
||||
raw, _ := json.Marshal(map[string]string{"node": node, "module": module})
|
||||
p.w.put(provider.AssignmentsBucket, node+"."+module, raw)
|
||||
}
|
||||
place := func(node string, modules ...string) {
|
||||
pl := delivery.Placement{Node: node}
|
||||
for _, m := range modules {
|
||||
pl.Modules = append(pl.Modules, delivery.PlacedModule{Module: m})
|
||||
}
|
||||
raw, err := delivery.EncodePlacement(pl)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
key, _ := delivery.PlacementKey(node)
|
||||
p.w.put(delivery.PlacementBucket, key, raw)
|
||||
}
|
||||
assigned("g14", "a")
|
||||
assigned("g14", "b")
|
||||
if got, _ := p.w.assigned(); !slices.Equal(got["g14"], []string{"a", "b"}) || len(p.sayPlacement(p.w.placementDiffers())) != 0 || len(said) != 0 {
|
||||
t.Fatalf("before any placement: assigned %v, said %v", got, said)
|
||||
}
|
||||
place("g14", "a", "c")
|
||||
got, _ := p.w.assigned()
|
||||
frozen := p.sayPlacement(p.w.placementDiffers())
|
||||
p.sayPlacement(p.w.placementDiffers())
|
||||
if !slices.Equal(got["g14"], []string{"a"}) || !frozen["g14.b"] || !frozen["g14.c"] || frozen["g14.a"] || len(said) != 2 ||
|
||||
said[0].State != asks.StateRaised || !strings.Contains(said[0].Explanation, "the controller's assignments names b on g14") ||
|
||||
!strings.Contains(said[1].Explanation, "the catalogue's placement names c on g14") {
|
||||
t.Fatalf("one source alone: assigned %v, frozen %v, said %+v", got, frozen, said)
|
||||
}
|
||||
place("g14", "a", "b", "c")
|
||||
assigned("g14", "c")
|
||||
if frozen := p.sayPlacement(p.w.placementDiffers()); len(frozen) != 0 || len(said) != 4 || said[2].State != asks.StateCleared || said[3].State != asks.StateCleared {
|
||||
t.Errorf("both agree again: frozen %v, said %+v", frozen, said)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -162,8 +162,45 @@ func (w *world) manifest(ctx context.Context, get getter, digest string) parsed
|
||||
return p
|
||||
}
|
||||
|
||||
// assigned is each node's modules with an open assignment, read from the controller's assignments, in name order.
|
||||
// assigned is each node's modules both the controller's assignments and the indexer's placement name, in name order
|
||||
// (ADR 0549 §5; the assignments alone before any placement was read), with every open assignment by node and module.
|
||||
func (w *world) assigned() (map[string][]string, map[string]assignment) {
|
||||
byAssignment, all := w.assignedByController()
|
||||
placed, seeded := w.placed()
|
||||
both, _ := assign.Placed(byAssignment, placed, seeded)
|
||||
return both, all
|
||||
}
|
||||
|
||||
// placementDiffers is every pair the controller's assignments and the indexer's placement do not both name.
|
||||
func (w *world) placementDiffers() []assign.Pair {
|
||||
byAssignment, _ := w.assignedByController()
|
||||
placed, seeded := w.placed()
|
||||
_, differ := assign.Placed(byAssignment, placed, seeded)
|
||||
return differ
|
||||
}
|
||||
|
||||
// placed is each node's modules in the indexer's placement (ADR 0549 §2), and whether any placement key was read: a key
|
||||
// that does not read places nothing on its node and is said by the indexer's own read tool.
|
||||
func (w *world) placed() (map[string][]string, bool) {
|
||||
out := map[string][]string{}
|
||||
values := w.read(delivery.PlacementBucket)
|
||||
for key, value := range values {
|
||||
node, err := delivery.ParsePlacementKey(key)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
p, err := delivery.DecodePlacement(value)
|
||||
if err != nil || p.Node != node {
|
||||
continue
|
||||
}
|
||||
out[node] = p.Assigned()
|
||||
}
|
||||
return out, len(values) > 0
|
||||
}
|
||||
|
||||
// assignedByController is each node's modules with an open assignment, read from the controller's assignments, in
|
||||
// name order.
|
||||
func (w *world) assignedByController() (map[string][]string, map[string]assignment) {
|
||||
out, all := map[string][]string{}, map[string]assignment{}
|
||||
for _, value := range w.read(provider.AssignmentsBucket) {
|
||||
var a assignment
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-assigner
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026171759-a431746c0e8c
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -8,6 +8,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025203528-db26f7e8b6f5 h1:gE80bR0t
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025203528-db26f7e8b6f5/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026171759-a431746c0e8c h1:zaSfYTJd03b0pF29QTZ6G5ajfjOXRXjHIpOiGu/niHA=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026171759-a431746c0e8c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -0,0 +1,99 @@
|
||||
package assign
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sort"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// Until ADR 0547's third milestone two sources name which module runs on which node: the controller's assignments and
|
||||
// the indexer's placement, read from the catalogue (ADR 0549 §5). The assigner writes wants only for a pair both name:
|
||||
// for a pair one source names alone it writes no want and touches none it wrote before, so nothing moves and nothing is
|
||||
// removed on one source's word, and it raises a condition naming the pair until both agree. Before the first placement
|
||||
// file reaches the indexer (no placement key at all) the controller's assignments are the one source, as before.
|
||||
|
||||
// Pair is a module on a node that one source names and the other does not; Only says which: OnlyAssignments or
|
||||
// OnlyPlacement.
|
||||
type Pair struct {
|
||||
Node, Module, Only string
|
||||
}
|
||||
|
||||
// The source that names a pair alone.
|
||||
const (
|
||||
OnlyAssignments = "the controller's assignments"
|
||||
OnlyPlacement = "the catalogue's placement"
|
||||
)
|
||||
|
||||
// Placed is, by node, the modules both sources name, in name order, and every pair one names alone, by node then
|
||||
// module. With no placement read yet (seeded false) it is the assignments, and no pair differs.
|
||||
func Placed(assigned, placed map[string][]string, seeded bool) (map[string][]string, []Pair) {
|
||||
if !seeded {
|
||||
return assigned, nil
|
||||
}
|
||||
set := func(m map[string][]string) map[string]bool {
|
||||
out := map[string]bool{}
|
||||
for node, modules := range m {
|
||||
for _, module := range modules {
|
||||
out[node+"\x00"+module] = true
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
inAssigned, inPlaced := set(assigned), set(placed)
|
||||
both := map[string][]string{}
|
||||
var differ []Pair
|
||||
for node, modules := range assigned {
|
||||
for _, module := range modules {
|
||||
if inPlaced[node+"\x00"+module] {
|
||||
both[node] = append(both[node], module)
|
||||
} else {
|
||||
differ = append(differ, Pair{Node: node, Module: module, Only: OnlyAssignments})
|
||||
}
|
||||
}
|
||||
}
|
||||
for node, modules := range placed {
|
||||
for _, module := range modules {
|
||||
if !inAssigned[node+"\x00"+module] {
|
||||
differ = append(differ, Pair{Node: node, Module: module, Only: OnlyPlacement})
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, modules := range both {
|
||||
sort.Strings(modules)
|
||||
}
|
||||
sort.Slice(differ, func(i, j int) bool {
|
||||
if differ[i].Node != differ[j].Node {
|
||||
return differ[i].Node < differ[j].Node
|
||||
}
|
||||
return differ[i].Module < differ[j].Module
|
||||
})
|
||||
return both, differ
|
||||
}
|
||||
|
||||
// PlacementCondition is the condition of a pair the two sources do not both name: raised while they differ, cleared once
|
||||
// both name it or neither does.
|
||||
func PlacementCondition(p Pair, raised bool) asks.Condition {
|
||||
key := delivery.AssignerModule + ".placement." + p.Node + "." + p.Module
|
||||
subject := asks.Scope{Scope: "module", ID: p.Module, Machine: p.Node}
|
||||
if !raised {
|
||||
return asks.Condition{Key: key, Subject: subject, State: asks.StateCleared,
|
||||
Resolved: fmt.Sprintf("the controller's assignments and the catalogue's placement agree on %s on %s.", p.Module, p.Node)}
|
||||
}
|
||||
headline := fmt.Sprintf("%s on %s is placed by one source only", p.Module, p.Node)
|
||||
if n := len([]rune(headline)); n > asks.HeadlineLength {
|
||||
headline = string([]rune(headline)[:asks.HeadlineLength])
|
||||
}
|
||||
var needs string
|
||||
if p.Only == OnlyAssignments {
|
||||
needs = fmt.Sprintf("add %s to the catalogue's placement file of %s, or unassign it at the controller", p.Module, p.Node)
|
||||
} else {
|
||||
needs = fmt.Sprintf("assign %s to %s at the controller, as the merged placement file says", p.Module, p.Node)
|
||||
}
|
||||
return asks.Condition{Key: key, Severity: asks.SeverityUrgent, Subject: subject, State: asks.StateRaised,
|
||||
Summary: headline, Headline: headline, Needs: needs,
|
||||
Explanation: fmt.Sprintf("%s%s. %s names %s on %s and the other source does not. Until ADR 0547's third milestone both must "+
|
||||
"name a module on a node (ADR 0549 §5): the assigner writes no want for it there and leaves the one it wrote before, so "+
|
||||
"nothing moves or is removed. This clears once both name it or neither does.", asks.NeedsYou, needs, p.Only, p.Module, p.Node)}
|
||||
}
|
||||
@@ -34,7 +34,7 @@ func TestManifest(t *testing.T) {
|
||||
slices.Sort(reads)
|
||||
want := []string{"mesh-build-go.steps", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-controller.grants",
|
||||
"mesh-delivery-builder.builds", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "node-engine.have"}
|
||||
"mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "node-engine.have"}
|
||||
if !reflect.DeepEqual(reads, want) {
|
||||
t.Errorf("reads = %v, want exactly %v", reads, want)
|
||||
}
|
||||
|
||||
@@ -12,6 +12,7 @@
|
||||
"mesh-delivery-indexer.candidates",
|
||||
"mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules",
|
||||
"mesh-delivery-indexer.placement",
|
||||
"mesh-delivery-teller.answers",
|
||||
"mesh-delivery-watchman.presence",
|
||||
"node-engine.have"
|
||||
|
||||
Reference in New Issue
Block a user