The assigner reads no assignment and no lease: the placement alone names where modules run (ADR 0567 §8, ADR 0549 §5)
Every handed-away branch is the only one: world.assigned is the placement, the pair comparison and its conditions go, the lease and assignments buckets are no longer opened, watched or declared, and a proposal of any kind is taken. Tests: the pair-comparison test and the controller-composition record-ask test are replaced by one placement test; the rest are edited to place their modules instead of assigning them. (issue 774, 774/S26) Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
This commit is contained in:
@@ -30,7 +30,6 @@ import (
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"git.novox.be/novox/mesh-sdk/go/provider"
|
||||
"git.novox.be/novox/mesh-sdk/go/provisions"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
@@ -58,9 +57,7 @@ func main() {
|
||||
// holds the two lists equal), so the controller's broker grants exactly these.
|
||||
var readBuckets = []string{
|
||||
buscompose.ResolvedBucket,
|
||||
provider.AssignmentsBucket,
|
||||
delivery.PartGrantsBucket,
|
||||
buscompose.LeaseBucket,
|
||||
delivery.ControllerComposedBucket,
|
||||
delivery.AnswersBucket,
|
||||
delivery.BuilderBuildsBucket,
|
||||
@@ -127,40 +124,14 @@ type buckets map[string]jetstream.KeyValue
|
||||
// without it and takes no put further until it starts again (ADR 0546).
|
||||
func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any)) (buckets, error) {
|
||||
out := buckets{}
|
||||
// The controller's lease first (ADR 0549 §5, ADR 0561 §3): once it is handed away the seat's placement alone names
|
||||
// where modules run, and the controller's assignments are no bucket the assigner needs. Not granted or not there, the
|
||||
// controller holds it, as before.
|
||||
away, why := false, ""
|
||||
if kv, err := js.KeyValue(ctx, buscompose.LeaseBucket); err != nil {
|
||||
logf("the bucket %s cannot be opened (%v): the controller is taken to hold its lease, so the assignments are read beside the placement", buscompose.LeaseBucket, err)
|
||||
} else {
|
||||
out[buscompose.LeaseBucket] = kv
|
||||
e, gerr := kv.Get(ctx, buscompose.ControllerLeaseKey)
|
||||
var raw []byte
|
||||
if gerr == nil {
|
||||
raw = e.Value()
|
||||
}
|
||||
if errors.Is(gerr, jetstream.ErrKeyNotFound) {
|
||||
gerr = nil
|
||||
}
|
||||
away, why = buscompose.LeaseHandedAway(raw, raw != nil, gerr)
|
||||
}
|
||||
for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) {
|
||||
if name == buscompose.LeaseBucket {
|
||||
continue
|
||||
}
|
||||
kv, err := js.KeyValue(ctx, name)
|
||||
if name == provider.AssignmentsBucket && err != nil && away {
|
||||
logf("the bucket %s cannot be opened (%v), and none is needed: %s, so the seat's placement alone is read", name, err, why)
|
||||
continue
|
||||
}
|
||||
if name == delivery.AssignerPutsBucket && err != nil {
|
||||
logf("the bucket %s cannot be opened (%v): no put is taken further until the assigner starts again", name, err)
|
||||
continue
|
||||
}
|
||||
if slices.Contains(seatBuckets, name) && err != nil {
|
||||
// The seat's states come in the same delivery as this assigner (ADR 0567 §1): without placement the controller's
|
||||
// assignments alone name where modules run, as before any placement was written (ADR 0549 §5).
|
||||
// The seat's states come in the same delivery as this assigner (ADR 0567 §1): without placement no module is placed.
|
||||
logf("the bucket %s cannot be opened (%v): it is bound once the bus says it was made", name, err)
|
||||
continue
|
||||
}
|
||||
@@ -193,10 +164,9 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any
|
||||
logf("the bucket %s cannot be opened (%v): the assigner's conditions are said straight to the router until it starts again", name, err)
|
||||
continue
|
||||
}
|
||||
if sourceOfGrants(name) && err != nil {
|
||||
// The grants part's grants and the controller's lease choose where grants are read from (issue 774): without
|
||||
// them the controller's grants are read, as before the lease is handed away.
|
||||
logf("the bucket %s cannot be opened (%v): the controller's grants are read until the assigner starts again", name, err)
|
||||
if name == delivery.PartGrantsBucket && err != nil {
|
||||
// The grants part's grants (ADR 0571 clause 9): without them no module counts as granted until it is bound.
|
||||
logf("the bucket %s cannot be opened (%v): no grant is read until it is bound once the bus says it was made", name, err)
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
@@ -215,13 +185,7 @@ func startsWithout(name string) bool {
|
||||
delivery.AssignerPutsBucket:
|
||||
return true
|
||||
}
|
||||
return slices.Contains(seatBuckets, name) || sourceOfGrants(name)
|
||||
}
|
||||
|
||||
// sourceOfGrants says the bucket is the grants part's grants or the controller's lease: neither stops the assigner when it
|
||||
// is not there.
|
||||
func sourceOfGrants(name string) bool {
|
||||
return name == delivery.PartGrantsBucket || name == buscompose.LeaseBucket
|
||||
return slices.Contains(seatBuckets, name) || name == delivery.PartGrantsBucket
|
||||
}
|
||||
|
||||
// run is the program: it returns when ctx ends or the bus cannot be used.
|
||||
@@ -331,9 +295,7 @@ func (e emitting) Emit(eventType string, body any) error { return e.Conn.Emit(ev
|
||||
// watched only for keys, so a manifest that arrives wakes a composition that waits for it.
|
||||
var watched = []string{
|
||||
buscompose.ResolvedBucket,
|
||||
provider.AssignmentsBucket,
|
||||
delivery.PartGrantsBucket,
|
||||
buscompose.LeaseBucket,
|
||||
delivery.ControllerComposedBucket,
|
||||
delivery.AnswersBucket,
|
||||
delivery.BuilderBuildsBucket,
|
||||
@@ -417,12 +379,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
if e.Operation() == jetstream.KeyValuePut {
|
||||
value = e.Value()
|
||||
}
|
||||
before := w.handedAway()
|
||||
w.putEntry(name, e.Key(), value, e.Revision(), e.Created())
|
||||
if name == buscompose.LeaseBucket && w.handedAway() == before {
|
||||
// The controller renews its lease every few seconds: only the lease's being handed away makes a pass.
|
||||
continue
|
||||
}
|
||||
wake()
|
||||
}
|
||||
}
|
||||
@@ -458,13 +415,13 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
return
|
||||
}
|
||||
if bus != nil {
|
||||
p.acts = newPutActs(bus, wake, logger.Printf, w.handedAway)
|
||||
p.acts = newPutActs(bus, wake, logger.Printf)
|
||||
}
|
||||
logger.Printf("the bucket %s is open: puts are taken further", delivery.AssignerPutsBucket)
|
||||
}
|
||||
started := 0
|
||||
for _, name := range watched {
|
||||
if kvs[name] == nil && (startsWithout(name) || name == provider.AssignmentsBucket) {
|
||||
if kvs[name] == nil && startsWithout(name) {
|
||||
continue // not on the bus at this start (open said so)
|
||||
}
|
||||
started++
|
||||
@@ -480,8 +437,8 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
return err
|
||||
}
|
||||
// The seat's states arrive with the same delivery as this assigner (ADR 0567 §1): one not on the bus at this start is
|
||||
// bound once the bus says it was created, and watched from then on, with no restart (issue 1790's kind). Until then the
|
||||
// controller's assignments alone name where modules run (ADR 0549 §5).
|
||||
// bound once the bus says it was created, and watched from then on, with no restart (issue 1790's kind). Until then no
|
||||
// module is placed.
|
||||
type appeared struct {
|
||||
name string
|
||||
kv jetstream.KeyValue
|
||||
@@ -494,7 +451,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
}
|
||||
}
|
||||
// And the two that choose where grants are read from, with the same delivery as the grants part that writes one.
|
||||
for _, name := range []string{delivery.PartGrantsBucket, buscompose.LeaseBucket, assign.PoliciesBucket, provisions.ProvisionsBucket, keptBucket, delivery.SeatHoldersBucket} {
|
||||
for _, name := range []string{delivery.PartGrantsBucket, assign.PoliciesBucket, provisions.ProvisionsBucket, keptBucket, delivery.SeatHoldersBucket} {
|
||||
if kvs[name] == nil && js != nil {
|
||||
name := name
|
||||
process.AwaitBucket(ctx, js, nc, name, "until then the bucket is not read",
|
||||
@@ -521,7 +478,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
p := &passer{kvs: kvs, w: w, logf: logger.Printf, say: say, findings: findings, said: map[string]bool{}, answeredAt: map[string]uint64{}}
|
||||
p.asks = newComposeAsks(ctx, buildrun.Asker{Publisher: buildrun.JetStreamAsks{JS: js}, Waiter: buildrun.JetStreamAsks{JS: js}}, wake, logger.Printf)
|
||||
if bus != nil && kvs[delivery.AssignerPutsBucket] != nil {
|
||||
p.acts = newPutActs(bus, wake, logger.Printf, w.handedAway)
|
||||
p.acts = newPutActs(bus, wake, logger.Printf)
|
||||
}
|
||||
if bus != nil {
|
||||
p.seat = newSeatActs(bus, wake, logger.Printf)
|
||||
|
||||
@@ -19,10 +19,8 @@ import (
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/provider"
|
||||
)
|
||||
|
||||
// 1297/B2a: the buckets the program opens are exactly the manifest's reads and states, as the broker names them
|
||||
@@ -74,28 +72,21 @@ func TestOpensWhatItsManifestNames(t *testing.T) {
|
||||
}
|
||||
// ADR 0546: whatever keeps the puts' bucket from opening (not there, not granted, no answer in time), the assigner opens
|
||||
// the rest and starts, writing wants; any other bucket it cannot open stops the start, named.
|
||||
// Issue 774 (ADR 0549 §5): the controller's lease not granted, the assigner starts taking it as the controller's; the
|
||||
// assignments refused while the controller holds it stop the start, and once it is handed away they are not needed.
|
||||
handed := []byte(`{"handed-to":"the operator's hand-over","handed-by":"person.jochen"}`)
|
||||
for _, c := range []struct {
|
||||
missing string
|
||||
err error
|
||||
starts bool
|
||||
lease []byte
|
||||
said string
|
||||
}{
|
||||
{delivery.AssignerPutsBucket, jetstream.ErrBucketNotFound, true, nil, "no put is taken further"},
|
||||
{delivery.AssignerPutsBucket, errors.New("nats: permissions violation for subscription to \"$JS.API.STREAM.INFO.KV_mesh-delivery-assigner_puts\""), true, nil, "no put is taken further"},
|
||||
{delivery.AssignerPutsBucket, context.DeadlineExceeded, true, nil, "no put is taken further"},
|
||||
{delivery.AssignerWantsBucket, errors.New("nats: permissions violation"), false, nil, ""},
|
||||
// Issue 774: the grants part's grants and the controller's lease choose the grants' source; neither stops the start.
|
||||
{delivery.PartGrantsBucket, jetstream.ErrBucketNotFound, true, nil, "the controller's grants are read"},
|
||||
{buscompose.LeaseBucket, errors.New("nats: permissions violation"), true, nil, "taken to hold its lease"},
|
||||
{provider.AssignmentsBucket, errors.New("nats: permissions violation"), false, []byte(`{"instance":"novox/1"}`), ""},
|
||||
{provider.AssignmentsBucket, errors.New("nats: permissions violation"), true, handed, "the seat's placement alone is read"},
|
||||
{delivery.AssignerPutsBucket, jetstream.ErrBucketNotFound, true, "no put is taken further"},
|
||||
{delivery.AssignerPutsBucket, errors.New("nats: permissions violation for subscription to \"$JS.API.STREAM.INFO.KV_mesh-delivery-assigner_puts\""), true, "no put is taken further"},
|
||||
{delivery.AssignerPutsBucket, context.DeadlineExceeded, true, "no put is taken further"},
|
||||
{delivery.AssignerWantsBucket, errors.New("nats: permissions violation"), false, ""},
|
||||
// Issue 774: the grants part's grants not there do not stop the start.
|
||||
{delivery.PartGrantsBucket, jetstream.ErrBucketNotFound, true, "no grant is read"},
|
||||
} {
|
||||
var logged string
|
||||
kvs, err := open(context.Background(), refusingJS{missing: c.missing, err: c.err, lease: c.lease}, func(f string, a ...any) { logged += fmt.Sprintf(f, a...) })
|
||||
kvs, err := open(context.Background(), refusingJS{missing: c.missing, err: c.err}, func(f string, a ...any) { logged += fmt.Sprintf(f, a...) })
|
||||
switch {
|
||||
case c.starts && (err != nil || kvs[c.missing] != nil || len(kvs) != len(readBuckets)+len(ownBuckets)-1 || !strings.Contains(logged, c.said)):
|
||||
t.Errorf("%s refused with %v: the start answered %v, %d buckets, logged %q", c.missing, c.err, err, len(kvs), logged)
|
||||
@@ -112,12 +103,14 @@ func TestOpensWhatItsManifestNames(t *testing.T) {
|
||||
for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) {
|
||||
kvs[name] = &watchKV{memKV: &memKV{values: map[string][]byte{}}}
|
||||
}
|
||||
ckey, _ := delivery.ComposedNodeKey("ace")
|
||||
cd, err := delivery.EncodeComposedDigest(delivery.ComposedDigest{Digest: "sha256:" + strings.Repeat("d", 64), Judgement: delivery.JudgedUnchanged})
|
||||
placed, err := delivery.PlacementValues(delivery.Placement{Node: "ace"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
kvs[delivery.ControllerComposedBucket].(*watchKV).entries = []kvEntry{{key: ckey, value: cd}}
|
||||
for key, raw := range placed {
|
||||
kv := kvs[delivery.PlacementBucket].(*watchKV)
|
||||
kv.entries = append(kv.entries, kvEntry{key: key, value: raw})
|
||||
}
|
||||
kvs[delivery.AssignerPutsBucket].(*watchKV).closed = true
|
||||
kvs[delivery.AssignerWantsBucket].(*watchKV).wrote = wrote
|
||||
go follow(ctx, nil, nil, kvs, func(asks.Condition) error { return nil }, assign.Findings{}, nil, nil, log.New(io.Discard, "", 0))
|
||||
@@ -136,18 +129,13 @@ type refusingJS struct {
|
||||
jetstream.JetStream
|
||||
missing string
|
||||
err error
|
||||
lease []byte // the controller's lease, when it holds one
|
||||
}
|
||||
|
||||
func (r refusingJS) KeyValue(_ context.Context, name string) (jetstream.KeyValue, error) {
|
||||
if name == r.missing {
|
||||
return nil, r.err
|
||||
}
|
||||
kv := &memKV{values: map[string][]byte{}}
|
||||
if name == buscompose.LeaseBucket && r.lease != nil {
|
||||
kv.values[buscompose.ControllerLeaseKey] = r.lease
|
||||
}
|
||||
return kv, nil
|
||||
return &memKV{values: map[string][]byte{}}, nil
|
||||
}
|
||||
|
||||
// watchKV is a memKV whose watch hands its entries and then the marker that all stands was handed, or ends at once when
|
||||
|
||||
@@ -90,8 +90,7 @@ func (p *passer) pass(ctx context.Context) {
|
||||
p.backups(ctx)
|
||||
}
|
||||
|
||||
// nodeWants writes each node's own want, `want.<node>.node`, naming the digest of the node's own composition, once the
|
||||
// controller has composed it and again whenever that digest changes (ADR 0539 §6).
|
||||
// nodeWants writes each placed node's own want, `want.<node>.node`: the node composes its own (ADR 0547 §2, ADR 0571 §6).
|
||||
func (p *passer) nodeWants(ctx context.Context) {
|
||||
written := p.w.read(delivery.AssignerWantsBucket)
|
||||
placed := p.w.placedNodes()
|
||||
@@ -107,20 +106,6 @@ func (p *passer) nodeWants(ctx context.Context) {
|
||||
want := delivery.NodeWant{Self: true, Words: "the node composes its own composition (ADR 0547 §2)"}
|
||||
p.put(ctx, delivery.AssignerWantsBucket, key, func() ([]byte, error) { return delivery.EncodeNodeWant(want) })
|
||||
}
|
||||
for node, d := range p.w.nodeComposed() {
|
||||
if placed[node] {
|
||||
continue
|
||||
}
|
||||
key, err := delivery.WantNodeKey(node)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if before, err := delivery.DecodeNodeWant(written[key]); err == nil && before.Composed == d.Digest {
|
||||
continue
|
||||
}
|
||||
want := delivery.NodeWant{Composed: d.Digest, Words: "the node's own composition"}
|
||||
p.put(ctx, delivery.AssignerWantsBucket, key, func() ([]byte, error) { return delivery.EncodeNodeWant(want) })
|
||||
}
|
||||
}
|
||||
|
||||
// wants writes, for every module the trunk holds a fingerprint of, each node's want by the six rules (ADR 0429 §6), and
|
||||
@@ -128,7 +113,7 @@ func (p *passer) nodeWants(ctx context.Context) {
|
||||
// sends: the differences bucket is kept read-only with its history (ADR 0539 §8).
|
||||
func (p *passer) wants(ctx context.Context) {
|
||||
get := kvGetter(p.kvs[delivery.ManifestsBucket])
|
||||
assigned, _ := p.w.assigned()
|
||||
assigned := p.w.assigned()
|
||||
trunk := p.w.trunk()
|
||||
manifests := map[string]catalogue.Manifest{}
|
||||
for name, mod := range trunk {
|
||||
@@ -155,8 +140,6 @@ func (p *passer) wants(ctx context.Context) {
|
||||
defer p.seatHolders(ctx)
|
||||
defer p.policySplits(manifests, chosen)
|
||||
p.clearStaleCalls()
|
||||
differ := p.w.placementDiffers()
|
||||
frozen := p.sayPlacement(differ)
|
||||
placed := p.w.placedNodes()
|
||||
puts := p.putsByModule()
|
||||
p.staleAsks(ctx, trunk)
|
||||
@@ -184,15 +167,7 @@ func (p *passer) wants(ctx context.Context) {
|
||||
m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Chosen: chosenOf(chosen, name), Provisions: provisionsOf(name), Answering: answering(manifests, nodes), Nodes: nodesOf[name], Now: time.Now(), Puts: puts[name],
|
||||
Self: func(node string) bool { return placed[node] },
|
||||
Composed: func(node, fp string) (delivery.ComposedDigest, bool) {
|
||||
if placed[node] && p.w.handedAway() {
|
||||
return p.w.ownComposed(node) // from the lease write: the digest its own have names (ADR 0571 §6)
|
||||
}
|
||||
if placed[node] {
|
||||
// The node composes its own (ADR 0547 §2, §4): no composition of the controller's is waited for, the want
|
||||
// names the build alone, and the node judges the move itself.
|
||||
return delivery.ComposedDigest{Judgement: delivery.JudgedNeither, Words: "the node composes its own"}, true
|
||||
}
|
||||
return p.w.composed(node, name, fp)
|
||||
return p.w.composed(node, name, fp) // a placed node's own: the digest its own have names (ADR 0571 §6)
|
||||
},
|
||||
GrantedAt: func(fp string) bool {
|
||||
key, err := delivery.ControllerGrantKey(name, fp)
|
||||
@@ -265,9 +240,6 @@ 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)
|
||||
@@ -901,42 +873,6 @@ 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
|
||||
}
|
||||
|
||||
// yeses writes a placed node's wants naming the build alone, and sees to the moves a node that composes its own judged a
|
||||
// person must allow (ADR 0547 §3): for a have of phase
|
||||
// waiting-yes it raises the record ask once, naming the composition waiting and the node's judgement, and on a yes writes
|
||||
@@ -1001,11 +937,8 @@ func (p *passer) policySplits(manifests map[string]catalogue.Manifest, chosen ma
|
||||
now = p.now
|
||||
}
|
||||
split := map[string]bool{}
|
||||
theirs := map[string][]string{}
|
||||
if !p.w.handedAway() { // the controller writes no policy from the lease write (ADR 0571 §6)
|
||||
theirs = assign.ControllerPolicies(p.w.read(delivery.ControllerComposedBucket))
|
||||
}
|
||||
for _, s := range assign.Splits(manifests, chosen, theirs) {
|
||||
// The controller writes no policy any more (ADR 0571 §6): nothing is compared with.
|
||||
for _, s := range assign.Splits(manifests, chosen, nil) {
|
||||
split[s.Module] = true
|
||||
if p.splits[s.Module] {
|
||||
continue
|
||||
|
||||
@@ -14,10 +14,8 @@ import (
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/provider"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
)
|
||||
@@ -51,6 +49,34 @@ func newPasser(kvs buckets, at time.Time) *passer {
|
||||
now: func() time.Time { return at }}
|
||||
}
|
||||
|
||||
// placeOn puts the seat's placement of node, naming the modules, into the world: one node key and a key per module
|
||||
// (research 072 row 37).
|
||||
func placeOn(t *testing.T, w *world, node string, modules ...string) {
|
||||
t.Helper()
|
||||
pl := delivery.Placement{Node: node}
|
||||
for _, m := range modules {
|
||||
pl.Modules = append(pl.Modules, delivery.PlacedModule{Module: m})
|
||||
}
|
||||
values, err := delivery.PlacementValues(pl)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for key, raw := range values {
|
||||
w.put(delivery.PlacementBucket, key, raw)
|
||||
}
|
||||
}
|
||||
|
||||
// composes puts the digest the node's own have names as its composition (`<node>.node` of the haves).
|
||||
func composes(t *testing.T, w *world, node, digest string) {
|
||||
t.Helper()
|
||||
raw, err := delivery.EncodeHaveNode(delivery.HaveNode{Composed: digest})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
key, _ := delivery.HaveNodeKey(node)
|
||||
w.put(delivery.HaveBucket, key, raw)
|
||||
}
|
||||
|
||||
var fpTest = delivery.SourcePrefix2 + strings.Repeat("b", 64)
|
||||
|
||||
// 1906/H2: given a hold lifted by itself, when the assigner is started again and the same fingerprint is held again, then the
|
||||
@@ -195,7 +221,7 @@ func putOfTheBus(t *testing.T) {
|
||||
p := newPasser(kvs, now)
|
||||
bus := &fakeBus{calls: make(chan string, 8)}
|
||||
woke := make(chan struct{}, 8) // each act's goroutine wakes a pass once it is done
|
||||
p.acts = newPutActs(bus, func() { woke <- struct{}{} }, func(string, ...any) {}, nil)
|
||||
p.acts = newPutActs(bus, func() { woke <- struct{}{} }, func(string, ...any) {})
|
||||
p.acts.snapshot = func(context.Context, string, string) (string, error) {
|
||||
return "novox's restore point of nats (streams) taken 22:01", nil
|
||||
}
|
||||
@@ -210,7 +236,7 @@ func putOfTheBus(t *testing.T) {
|
||||
}
|
||||
p.w.put(bucket, key, data)
|
||||
}
|
||||
world(provider.AssignmentsBucket, "novox.nats", map[string]string{"node": "novox", "module": "nats"})
|
||||
placeOn(t, p.w, "novox", "nats")
|
||||
mdigest := strings.Repeat("e", 64)
|
||||
mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: mdigest, Repository: "novox/mesh-catalog", Commit: strings.Repeat("c", 40)})
|
||||
world(delivery.ModulesBucket, "nats", mod)
|
||||
@@ -266,9 +292,7 @@ func putOfTheBus(t *testing.T) {
|
||||
t.Errorf("a put asked %q: it reads the compositions and asks nobody for one", call)
|
||||
default:
|
||||
}
|
||||
ckey, _ := delivery.ComposedKey("novox", "nats", fp)
|
||||
cd, _ := delivery.EncodeComposedDigest(delivery.ComposedDigest{Digest: digest, Judgement: delivery.JudgedCannot, Words: "the have names no composition"})
|
||||
world(delivery.ControllerComposedBucket, ckey, cd)
|
||||
composes(t, p.w, "novox", digest)
|
||||
gkey, _ := delivery.ControllerGrantKey("nats", fp)
|
||||
g, _ := delivery.EncodeControllerGrant(delivery.ControllerGrant{})
|
||||
world(delivery.PartGrantsBucket, gkey, g)
|
||||
@@ -355,7 +379,7 @@ func putOfTheBus(t *testing.T) {
|
||||
// the journal, so it is loud with no operator channel.
|
||||
func TestAModuleOnNeitherListIsRaisedOnceAndClearedOnce(t *testing.T) {
|
||||
// A pair only one source names is a condition of the same kind (issue 1126): folded here to hold the count.
|
||||
aPairOneSourceNamesAloneGetsNoWantAndOneCondition(t)
|
||||
aPlacementAloneNamesWhereModulesRun(t)
|
||||
t.Run("gathers", testThePassGathersTheModulesOnNeitherList)
|
||||
p := newPasser(buckets{}, time.Date(2026, 10, 25, 15, 0, 0, 0, time.UTC))
|
||||
var said []asks.Condition
|
||||
@@ -529,7 +553,7 @@ func TestTheFirstPassAfterAStartClearsWhatAPreviousProcessRaised(t *testing.T) {
|
||||
p.say = func(c asks.Condition) error { said = append(said, c); return nil }
|
||||
mod, _ := delivery.EncodeModule(delivery.Module{FP2: fpTest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: delivery.RootPath})
|
||||
p.w.put(delivery.ModulesBucket, "web", mod)
|
||||
p.w.put(provider.AssignmentsBucket, "g14.web", []byte(`{"node":"g14","module":"web","build":"b-1"}`))
|
||||
placeOn(t, p.w, "g14", "web")
|
||||
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
|
||||
p.w.put(delivery.PresenceBucket, "g14", on)
|
||||
have, _ := delivery.EncodeHave(delivery.Have{FP: fpTest, Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
|
||||
@@ -567,7 +591,7 @@ func testAHeldModuleRaisesNoNeitherAndClearsAStandingOne(t *testing.T) {
|
||||
p.say = func(c asks.Condition) error { *said = append(*said, c); return nil }
|
||||
mod, _ := delivery.EncodeModule(delivery.Module{FP2: fpTest, Repository: "novox/resolved", Commit: strings.Repeat("1", 40), Path: delivery.RootPath})
|
||||
p.w.put(delivery.ModulesBucket, "systemd-resolved", mod)
|
||||
p.w.put(provider.AssignmentsBucket, "shanks.systemd-resolved", []byte(`{"node":"shanks","module":"systemd-resolved","build":"b-1"}`))
|
||||
placeOn(t, p.w, "shanks", "systemd-resolved")
|
||||
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
|
||||
p.w.put(delivery.PresenceBucket, "shanks", on)
|
||||
have, _ := delivery.EncodeHave(delivery.Have{FP: fpTest, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed})
|
||||
@@ -631,15 +655,13 @@ func TestAsksOfAFingerprintTheTrunkLeftBehindAreTakenAway(t *testing.T) {
|
||||
name string
|
||||
trunk map[string]delivery.Module
|
||||
gone []string // the keys taken away
|
||||
noDigit bool // the controller's composition is not read
|
||||
heldAt string // the node's want is at this fingerprint, whose composition is digOld
|
||||
noDigit bool // the node's own composition is not read
|
||||
}{
|
||||
{"the trunk moved on", trunkOf(fpNew), []string{holdOld, holdGone, restOld, recOld}, false, ""},
|
||||
{"the trunk is at the old fingerprint", trunkOf(fpOld), []string{holdNew, holdGone, restNew, recOld}, false, ""},
|
||||
{"the trunk's fingerprint is not read yet", trunkOf(""), []string{holdGone}, false, ""},
|
||||
{"the composition is not read yet", trunkOf(fpNew), []string{holdOld, holdGone, restOld}, true, ""},
|
||||
{"a node held back at the old fingerprint is still asked about its composition", trunkOf(fpNew), []string{holdOld, holdGone, restOld}, false, fpOld},
|
||||
{"the trunk is not read at all", nil, nil, false, ""},
|
||||
{"the trunk moved on", trunkOf(fpNew), []string{holdOld, holdGone, restOld, recOld}, false},
|
||||
{"the trunk is at the old fingerprint", trunkOf(fpOld), []string{holdNew, holdGone, restNew, recOld}, false},
|
||||
{"the trunk's fingerprint is not read yet", trunkOf(""), []string{holdGone}, false},
|
||||
{"the composition is not read yet", trunkOf(fpNew), []string{holdOld, holdGone, restOld}, true},
|
||||
{"the trunk is not read at all", nil, nil, false},
|
||||
} {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
asksKV, wantsKV := &memKV{values: map[string][]byte{}}, &memKV{values: map[string][]byte{}}
|
||||
@@ -666,21 +688,9 @@ func TestAsksOfAFingerprintTheTrunkLeftBehindAreTakenAway(t *testing.T) {
|
||||
asksKV.values[key] = v
|
||||
p.w.put(delivery.AssignerAsksBucket, key, v)
|
||||
}
|
||||
placeOn(t, p.w, "novox", "web") // a placed node composes its own: the digest its have names
|
||||
if !c.noDigit {
|
||||
ckey, _ := delivery.ComposedKey("novox", "web", fpNew)
|
||||
v, _ := delivery.EncodeComposedDigest(delivery.ComposedDigest{Digest: digNew, Judgement: delivery.JudgedUnchanged})
|
||||
p.w.put(delivery.ControllerComposedBucket, ckey, v)
|
||||
okey, _ := delivery.ComposedKey("novox", "web", fpOld)
|
||||
old, _ := delivery.EncodeComposedDigest(delivery.ComposedDigest{Digest: digNew, Judgement: delivery.JudgedUnchanged})
|
||||
if c.heldAt != "" {
|
||||
old, _ = delivery.EncodeComposedDigest(delivery.ComposedDigest{Digest: digOld, Judgement: delivery.JudgedUnchanged})
|
||||
}
|
||||
p.w.put(delivery.ControllerComposedBucket, okey, old)
|
||||
}
|
||||
if c.heldAt != "" {
|
||||
wkey, _ := delivery.WantKey("novox", "web")
|
||||
w, _ := delivery.EncodeWant(delivery.Want{FP: c.heldAt, Composed: digOld})
|
||||
p.w.put(delivery.AssignerWantsBucket, wkey, w)
|
||||
composes(t, p.w, "novox", digNew)
|
||||
}
|
||||
heldOld, _ := delivery.HeldKey("web", fpOld)
|
||||
wantsKV.values[heldOld] = []byte("x")
|
||||
@@ -778,134 +788,38 @@ func mustKey(key string, err error) string {
|
||||
func TestARecordAskOfAHeldBackNodeSurvivesPassesAndItsYesMovesTheWant(t *testing.T) {
|
||||
// A placed node's yes comes from its have (issue 1126): folded here to hold the count.
|
||||
aPlacedNodesWantsNameTheBuildAloneAndItsYesComesFromItsHave(t)
|
||||
fpOld := delivery.SourcePrefix2 + strings.Repeat("a", 64)
|
||||
revised, was := "sha256:"+strings.Repeat("e", 64), "sha256:"+strings.Repeat("d", 64)
|
||||
asksKV, wantsKV := &memKV{values: map[string][]byte{}}, &memKV{values: map[string][]byte{}}
|
||||
p := newPasser(buckets{delivery.AssignerAsksBucket: asksKV, delivery.AssignerWantsBucket: wantsKV}, time.Now())
|
||||
p.logf = t.Logf
|
||||
put := func(bucket, key string, v []byte) { p.w.put(bucket, key, v) }
|
||||
mod, _ := delivery.EncodeModule(delivery.Module{FP2: fpTest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: delivery.RootPath})
|
||||
put(delivery.ModulesBucket, "web", mod) // the trunk's build is not done: rule 1 holds g14 back at fpOld
|
||||
put(provider.AssignmentsBucket, "g14.web", []byte(`{"node":"g14","module":"web","build":"b-1"}`))
|
||||
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
|
||||
put(delivery.PresenceBucket, "g14", on)
|
||||
have, _ := delivery.EncodeHave(delivery.Have{FP: fpOld, Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
|
||||
put(delivery.HaveBucket, "g14.web", have)
|
||||
wkey, _ := delivery.WantKey("g14", "web")
|
||||
want, _ := delivery.EncodeWant(delivery.Want{FP: fpOld, Composed: was})
|
||||
wantsKV.values[wkey] = want
|
||||
put(delivery.AssignerWantsBucket, wkey, want)
|
||||
ckey, _ := delivery.ComposedKey("g14", "web", fpOld)
|
||||
cd, cerr := delivery.EncodeComposedDigest(delivery.ComposedDigest{Digest: revised, Judgement: delivery.JudgedRecreates, Policy: delivery.PolicyRecord, PolicyFrom: "default", Against: was, Words: "recreates"})
|
||||
if cerr != nil {
|
||||
t.Fatal(cerr)
|
||||
}
|
||||
put(delivery.ControllerComposedBucket, ckey, cd)
|
||||
tkey, _ := delivery.ComposedKey("g14", "web", fpTest) // the trunk's own composition, another digest than the revised one
|
||||
tcd, _ := delivery.EncodeComposedDigest(delivery.ComposedDigest{Digest: "sha256:" + strings.Repeat("f", 64), Judgement: delivery.JudgedUnchanged})
|
||||
put(delivery.ControllerComposedBucket, tkey, tcd)
|
||||
akey, _ := delivery.RecordAskKey("g14", "web", revised)
|
||||
for i := 0; i < 2; i++ {
|
||||
p.wants(context.Background())
|
||||
if _, there := asksKV.values[akey]; !there {
|
||||
t.Fatalf("pass %d: the record ask of the held-back node is not there", i+1)
|
||||
}
|
||||
}
|
||||
a, err := delivery.DecodeAssignerAsk(akey, asksKV.values[akey])
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
yes, _ := delivery.EncodeAnswer(delivery.Answer{Option: assign.OptionYes, Warrant: a.Ask.ID + "@2026-10-26T01:00:00Z", Words: "the operator"})
|
||||
put(delivery.AnswersBucket, akey, yes)
|
||||
p.wants(context.Background())
|
||||
got, err := delivery.DecodeWant(wantsKV.values[wkey])
|
||||
if err != nil || got.FP != fpOld || got.Composed != revised {
|
||||
t.Errorf("after the yes the want is %+v (%v), want it at %s naming %s", got, err, fpOld, revised)
|
||||
}
|
||||
if _, there := asksKV.values[akey]; !there {
|
||||
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 aPairOneSourceNamesAloneGetsNoWantAndOneCondition(t *testing.T) {
|
||||
// 774/S26/AC1 (ADR 0549 §5, ADR 0567 §8): the seat's placement alone names where modules run: no controller state is read, so a
|
||||
// module in an assignment only is no module of its node; before any placement is read nothing is placed and the check composes
|
||||
// no input; placed, the check composes over the input the nodes compose from (frombus, 1126, ADR 0547 §2).
|
||||
func aPlacementAloneNamesWhereModulesRun(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})
|
||||
}
|
||||
// One node key and a key per module (research 072 row 37); a module taken off leaves its key, which the node
|
||||
// key no longer names.
|
||||
values, err := delivery.PlacementValues(pl)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for key, raw := range values {
|
||||
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)
|
||||
p.w.put("mesh-controller_assignments", "g14.b", []byte(`{"node":"g14","module":"b","build":"b-1"}`)) // read no more
|
||||
if got := p.w.assigned(); len(got) != 0 {
|
||||
t.Fatalf("before any placement: assigned %v", got)
|
||||
}
|
||||
none := func(context.Context, string) ([]byte, error) { return nil, errNotThere }
|
||||
if m := p.w.mesh(context.Background(), none); m.Input != nil || m.InputUnread != "" {
|
||||
t.Errorf("before any placement the check composes over %+v (%q)", m.Input, m.InputUnread)
|
||||
}
|
||||
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 seat mesh-placement's placement names c on g14") {
|
||||
t.Fatalf("one source alone: assigned %v, frozen %v, said %+v", got, frozen, said)
|
||||
placeOn(t, p.w, "g14", "c", "a")
|
||||
placeOn(t, p.w, "n2", "d")
|
||||
if got := p.w.assigned(); !slices.Equal(got["g14"], []string{"a", "c"}) || !slices.Equal(got["n2"], []string{"d"}) || len(got) != 2 {
|
||||
t.Errorf("the placement alone: assigned %v", got)
|
||||
}
|
||||
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)
|
||||
}
|
||||
|
||||
// 1126 (ADR 0547 §2): placed, the check composes over the input the nodes compose from (frombus), once each placed node
|
||||
// wrote its record; until then it says why and composes the assignments alone.
|
||||
if m := p.w.mesh(context.Background(), none); m.Input != nil || !strings.Contains(m.InputUnread, "g14") {
|
||||
t.Errorf("a placed node with no record: input %+v, unread %q", m.Input, m.InputUnread)
|
||||
}
|
||||
placeOn(t, p.w, "g14", "a", "b", "c")
|
||||
rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "g14", TunnelKey: "k"}})
|
||||
p.w.put(delivery.HaveBucket, "g14.node", rec)
|
||||
rec2, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n2", TunnelKey: "k2"}})
|
||||
p.w.put(delivery.HaveBucket, "n2.node", rec2)
|
||||
m := p.w.mesh(context.Background(), none)
|
||||
if m.Input == nil || len(m.Input.Nodes) != 1 || !slices.Equal(m.Input.Nodes[0].Assigned, []string{"a", "b", "c"}) || m.Input.Nodes[0].Place.Key != "k" {
|
||||
if m.Input == nil || len(m.Input.Nodes) != 2 || !slices.Equal(m.Input.Nodes[0].Assigned, []string{"a", "b", "c"}) || m.Input.Nodes[0].Place.Key != "k" {
|
||||
t.Errorf("the check's input from the bus: %+v (%q)", m.Input, m.InputUnread)
|
||||
}
|
||||
|
||||
// Issue 774 (ADR 0549 §5, §6): once the controller's lease is handed away the placement alone names where modules
|
||||
// run: a pair in the placement only gets its want and no condition, and one in the assignments only is no module of
|
||||
// the node. A lease that goes back to a controller brings the comparison back.
|
||||
place("n2", "d")
|
||||
assigned("g14", "e")
|
||||
p.w.put(buscompose.LeaseBucket, buscompose.ControllerLeaseKey, []byte(`{"handed-to":"the operator's hand-over","handed-by":"person.jochen"}`))
|
||||
got, all := p.w.assigned()
|
||||
if !slices.Equal(got["n2"], []string{"d"}) || !slices.Equal(got["g14"], []string{"a", "b", "c"}) || len(all) != 0 {
|
||||
t.Errorf("handed away: assigned %v, assignments %v; want the placement alone", got, all)
|
||||
}
|
||||
if frozen := p.sayPlacement(p.w.placementDiffers()); len(frozen) != 0 || len(said) != 4 {
|
||||
t.Errorf("handed away: frozen %v, said %+v; want no condition", frozen, said[4:])
|
||||
}
|
||||
p.w.put(buscompose.LeaseBucket, buscompose.ControllerLeaseKey, []byte(`{"instance":"novox/1","epoch":58}`))
|
||||
if frozen := p.sayPlacement(p.w.placementDiffers()); !frozen["n2.d"] || !frozen["g14.e"] {
|
||||
t.Errorf("a controller's lease again: frozen %v; want both pairs held", frozen)
|
||||
}
|
||||
}
|
||||
|
||||
// 1126 (ADR 0547 §2 to §4): a node the catalogue places composes its own: its wants name the build alone and its own want
|
||||
@@ -920,7 +834,7 @@ func aPlacedNodesWantsNameTheBuildAloneAndItsYesComesFromItsHave(t *testing.T) {
|
||||
put := func(bucket, key string, v []byte) { p.w.put(bucket, key, v) }
|
||||
mod, _ := delivery.EncodeModule(delivery.Module{FP2: fpTest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: delivery.RootPath})
|
||||
put(delivery.ModulesBucket, "web", mod) // the trunk's build is not done: rule 1 holds g14 at fpOld
|
||||
put(provider.AssignmentsBucket, "g14.web", []byte(`{"node":"g14","module":"web","build":"b-1"}`))
|
||||
placeOn(t, p.w, "g14", "web")
|
||||
pl, _ := delivery.EncodePlacement(delivery.Placement{Node: "g14", Modules: []delivery.PlacedModule{{Module: "web"}}})
|
||||
put(delivery.PlacementBucket, "placement.g14", pl)
|
||||
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
|
||||
@@ -957,15 +871,10 @@ func aPlacedNodesWantsNameTheBuildAloneAndItsYesComesFromItsHave(t *testing.T) {
|
||||
if _, there := asksKV.values[akey]; !there {
|
||||
t.Fatal("a have waiting for a yes raised no record ask")
|
||||
}
|
||||
// Issue 774, 2026-10-31: while the controller still composes the placed node's module (its lease not handed away), its
|
||||
// digests are not the one the node's have waits on; the ask stays, and a pass that changes nothing puts nothing: on novox
|
||||
// the asks of postgres, keycloak and mongodb were taken away and made again every 8 s, and a want put unchanged each
|
||||
// pass made the node compose again each time.
|
||||
for _, fp := range []string{fpTest, fpOld} {
|
||||
ck, _ := delivery.ComposedKey("g14", "web", fp)
|
||||
cv, _ := delivery.EncodeComposedDigest(delivery.ComposedDigest{Digest: "sha256:" + strings.Repeat("c", 64), Judgement: delivery.JudgedUnchanged})
|
||||
put(delivery.ControllerComposedBucket, ck, cv)
|
||||
}
|
||||
// Issue 774, 2026-10-31: the node's own composition is not the digest its have waits on; the ask stays, and a pass that
|
||||
// changes nothing puts nothing: on novox the asks of postgres, keycloak and mongodb were taken away and made again every
|
||||
// 8 s, and a want put unchanged each pass made the node compose again each time.
|
||||
composes(t, p.w, "g14", "sha256:"+strings.Repeat("c", 64))
|
||||
asked := asksKV.values[akey]
|
||||
p.wants(context.Background())
|
||||
if !bytes.Equal(asksKV.values[akey], asked) {
|
||||
|
||||
@@ -55,8 +55,8 @@ type putActs struct {
|
||||
snapshot func(ctx context.Context, node, module string) (string, error)
|
||||
}
|
||||
|
||||
func newPutActs(bus caller, wake func(), logf func(string, ...any), leaseAway func() bool) *putActs {
|
||||
a := &putActs{rec: handacts.Recorder{Ask: bus.Call, LeaseAway: leaseAway}, bus: bus, wake: wake, logf: logf, snapping: map[string]bool{}, snapped: map[string]string{}, snapNo: map[string]string{}, warranted: map[string]bool{}, warrantNo: map[string]string{},
|
||||
func newPutActs(bus caller, wake func(), logf func(string, ...any)) *putActs {
|
||||
a := &putActs{rec: handacts.Recorder{Ask: bus.Call}, bus: bus, wake: wake, logf: logf, snapping: map[string]bool{}, snapped: map[string]string{}, snapNo: map[string]string{}, warranted: map[string]bool{}, warrantNo: map[string]string{},
|
||||
said: map[string]string{}}
|
||||
a.snapshot = a.snapshotTheBus
|
||||
return a
|
||||
@@ -97,7 +97,7 @@ func (p *passer) puts(ctx context.Context) {
|
||||
if len(all) == 0 {
|
||||
return
|
||||
}
|
||||
assigned, _ := p.w.assigned()
|
||||
assigned := p.w.assigned()
|
||||
trunk := p.w.trunk()
|
||||
get := kvGetter(p.kvs[delivery.ManifestsBucket])
|
||||
wanted := p.w.wantsOf()
|
||||
|
||||
@@ -524,7 +524,7 @@ func (p *passer) layAgain(ctx context.Context, prop delivery.PlacementProposal,
|
||||
held, recorded := p.lays(ctx)
|
||||
again, laid, findings := assign.Propose(assign.ProposeIn{ID: prop.ID, By: prop.By, Why: prop.Why, At: prop.At, Changes: prop.Changes,
|
||||
Current: current, Revisions: revs, Held: held, Recorded: recorded,
|
||||
Seed: prop.Seed, ExportDigest: prop.ExportDigest, Leased: p.w.handedAway()})
|
||||
Seed: prop.Seed, ExportDigest: prop.ExportDigest})
|
||||
if len(findings) > 0 {
|
||||
p.endProposal(ctx, prop, why+", and laid over the keys now it is refused: "+strings.Join(findings, "; "))
|
||||
return
|
||||
@@ -715,7 +715,7 @@ func (p *passer) proposeSeed(ctx context.Context, was delivery.PlacementProposal
|
||||
now := p.clock()
|
||||
seed, laid, findings := assign.Propose(assign.ProposeIn{ID: delivery.NewProposalID(now), By: was.By, Why: was.Why, At: now,
|
||||
Changes: assign.SeedChanges(export), Current: map[string]delivery.Placement{}, Revisions: map[string]uint64{},
|
||||
Held: held, Recorded: recorded, Seed: true, ExportDigest: delivery.ExportDigest(export), Leased: p.w.handedAway()})
|
||||
Held: held, Recorded: recorded, Seed: true, ExportDigest: delivery.ExportDigest(export)})
|
||||
if len(findings) > 0 {
|
||||
p.logf("the new export is not proposed as the seed: %s", strings.Join(findings, "; "))
|
||||
return
|
||||
|
||||
@@ -10,7 +10,6 @@ import (
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
|
||||
@@ -220,7 +219,7 @@ func (s *seatTools) doPropose(args map[string]any, by string) (any, error) {
|
||||
}
|
||||
now := s.now()
|
||||
p, laid, findings := assign.Propose(assign.ProposeIn{ID: delivery.NewProposalID(now), By: by, Why: arg(args, "why"), At: now,
|
||||
Changes: changes, Current: current, Revisions: revs, Held: held, Recorded: recorded, Leased: s.leased()})
|
||||
Changes: changes, Current: current, Revisions: revs, Held: held, Recorded: recorded})
|
||||
if len(findings) > 0 {
|
||||
return nil, fmt.Errorf("the proposal is refused: %s", strings.Join(findings, "; "))
|
||||
}
|
||||
@@ -304,7 +303,7 @@ func (s *seatTools) doAdopt(args map[string]any, by string) (any, error) {
|
||||
now := s.now()
|
||||
p, laid, findings := assign.Propose(assign.ProposeIn{ID: delivery.NewProposalID(now), By: by, Why: arg(args, "why"), At: now,
|
||||
Changes: assign.SeedChanges(export), Current: map[string]delivery.Placement{}, Revisions: map[string]uint64{},
|
||||
Held: held, Recorded: recorded, Seed: true, ExportDigest: delivery.ExportDigest(export), Leased: s.leased()})
|
||||
Held: held, Recorded: recorded, Seed: true, ExportDigest: delivery.ExportDigest(export)})
|
||||
if len(findings) > 0 {
|
||||
return nil, fmt.Errorf("the seed is refused: %s", strings.Join(findings, "; "))
|
||||
}
|
||||
@@ -314,48 +313,10 @@ func (s *seatTools) doAdopt(args map[string]any, by string) (any, error) {
|
||||
return proposed{Proposal: p, Follow: followProposal(p.ID)}, nil
|
||||
}
|
||||
|
||||
// leased says the controller's lease is handed away (mesh-sdk buscompose.LeaseHandedAway, ADR 0561 §3; issue 774): the lease
|
||||
// write of ADR 0547 that ends keeping the controller in step (ADR 0567 §8). A lease not read, a controller's, expired or
|
||||
// unreadable, is not handed away.
|
||||
func (s *seatTools) leased() bool {
|
||||
raw, _, found, err := s.st.get(buscompose.ControllerLease, buscompose.ControllerLeaseKey)
|
||||
away, _ := buscompose.LeaseHandedAway(raw, found && raw != nil, err)
|
||||
return away
|
||||
}
|
||||
|
||||
// doHandBack is the verb hand-back (§7): the operator's alone, so it is asked as a proposal is. It writes a hand-back
|
||||
// proposal naming every key at its revision; the assigner raises its authorising ask and, on its warrant, deletes the keys,
|
||||
// records the hand act and says placement-changed; every node returns to the controller's compositions.
|
||||
// doHandBack is the verb hand-back (§7): since the lease write placement is the mesh's alone, and there is no controller to
|
||||
// hand it back to (ADR 0547 §11).
|
||||
func (s *seatTools) doHandBack(args map[string]any, by string) (any, error) {
|
||||
if s.leased() {
|
||||
return nil, errors.New("placement is the mesh's alone since the lease write: there is no controller to hand it back to")
|
||||
}
|
||||
why := arg(args, "why")
|
||||
if why == "" {
|
||||
return nil, errors.New("hand-back says why: it is recorded with the hand act")
|
||||
}
|
||||
if by == "" {
|
||||
return nil, errors.New("the runtime names no caller: a hand-back is made by someone")
|
||||
}
|
||||
_, revs, err := s.placements()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(revs) == 0 {
|
||||
return nil, errors.New("there is no placement key to hand back")
|
||||
}
|
||||
now := s.now()
|
||||
p := delivery.PlacementProposal{ID: delivery.NewProposalID(now), By: by, Why: why, At: now.UTC(), Written: now.UTC(),
|
||||
HandBack: true, Base: revs, State: delivery.ProposalAnswering}
|
||||
raw, err := delivery.EncodePlacementProposal(p)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := s.st.put(delivery.PlacementProposalsState, p.ID, raw); err != nil {
|
||||
return nil, fmt.Errorf("the hand-back is not written: %w", err)
|
||||
}
|
||||
return proposed{Proposal: p, Follow: "the operator is asked to approve it; on the warrant every key is deleted and each " +
|
||||
"node returns to the controller's compositions"}, nil
|
||||
return nil, errors.New("placement is the mesh's alone since the lease write: there is no controller to hand it back to")
|
||||
}
|
||||
|
||||
// placementAnswer is what the verb placement answers.
|
||||
|
||||
@@ -13,7 +13,6 @@ import (
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
@@ -98,7 +97,7 @@ func TestTheSeatsVerbs(t *testing.T) {
|
||||
raw, _ := delivery.EncodePlacement(anchor)
|
||||
st[delivery.PlacementState].values["placement.anchor"] = raw
|
||||
if _, err := s.doPropose(map[string]any{"why": "x", "changes": `[{"op":"place","node":"anchor","module":"ghost"},{"op":"hold","node":"anchor","module":"web","claim":"x"}]`}, "op"); err == nil ||
|
||||
!strings.Contains(err.Error(), "ghost") || !strings.Contains(err.Error(), "lease write") || len(st[delivery.PlacementProposalsState].values) != 0 {
|
||||
!strings.Contains(err.Error(), "ghost") || len(st[delivery.PlacementProposalsState].values) != 0 {
|
||||
t.Errorf("a refused proposal: %v", err)
|
||||
}
|
||||
if _, err := s.doPropose(map[string]any{"why": "x", "changes": `[{"op":"swap","node":"anchor"}]`}, "op"); err == nil {
|
||||
@@ -146,32 +145,20 @@ func TestTheSeatsVerbs(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// hand-back: the operator's alone, written as a hand-back proposal naming every key at its revision, for the assigner
|
||||
// to ask; nothing is deleted by the verb.
|
||||
got, err = s.doHandBack(map[string]any{"why": "back"}, "op")
|
||||
if hb := got.(proposed).Proposal; err != nil || !hb.HandBack || hb.Base["anchor"] != uint64(len(raw)) || len(st[delivery.PlacementState].values) != 1 {
|
||||
t.Errorf("hand-back = %+v, %v", got, err)
|
||||
}
|
||||
// Issue 774 (ADR 0567 §8): once the controller's lease is handed away a change its verbs cannot take is proposed, and
|
||||
// there is no controller to hand placement back to.
|
||||
st[buscompose.ControllerLease] = &memKV{values: map[string][]byte{buscompose.ControllerLeaseKey: []byte(`{"handed-to":"the operator's hand-over","handed-by":"person.jochen"}`)}}
|
||||
// Issue 774 (ADR 0567 §8): every kind of change is proposed, and there is no controller to hand placement back to.
|
||||
if _, err := s.doPropose(map[string]any{"why": "hold", "changes": `[{"op":"hold","node":"anchor","module":"web","claim":"node-shell","scope":"node"}]`}, "op"); err != nil {
|
||||
t.Errorf("a hold after the lease write: %v", err)
|
||||
t.Errorf("a hold: %v", err)
|
||||
}
|
||||
if _, err := s.doHandBack(map[string]any{"why": "back"}, "op"); err == nil || !strings.Contains(err.Error(), "since the lease write") {
|
||||
t.Errorf("a hand-back after the lease write: %v", err)
|
||||
t.Errorf("a hand-back: %v", err)
|
||||
}
|
||||
if tools := s.tools(); len(tools) != 7 || tools[2].Name != "mesh-placement.propose" || tools[2].Run == nil || tools[2].Input == nil {
|
||||
t.Errorf("the seat's verbs as served: %+v", tools)
|
||||
}
|
||||
t.Run("1953/D4-D6: the seat mesh-wants' upgrade and bus", wantsVerbs)
|
||||
t.Run("1955/L7 (ADR 0571 §6): rule 7 reads the controller's composition until the lease write, the node's own after; an unplaced node is said", func(t *testing.T) {
|
||||
t.Run("1955/L7 (ADR 0571 §6): rule 7 reads the node's own composition; an unplaced node is said", func(t *testing.T) {
|
||||
w := newWorld()
|
||||
fp := delivery.SourcePrefix2 + strings.Repeat("2", 64)
|
||||
ckey, _ := delivery.ComposedKey("ace", "web", fp)
|
||||
cv, _ := delivery.EncodeComposedDigest(delivery.ComposedDigest{Digest: "sha256:" + strings.Repeat("c", 64), Judgement: delivery.JudgedNeither,
|
||||
Policy: catalogue.PolicyRoll, PolicyFrom: catalogue.FromDefault})
|
||||
w.putEntry(delivery.ControllerComposedBucket, ckey, cv, 1, time.Now())
|
||||
pl, _ := delivery.EncodePlacement(delivery.Placement{V: delivery.Version, Node: "ace", Modules: []delivery.PlacedModule{{Module: "web"}}})
|
||||
pkey, _ := delivery.LegacyPlacementKey("ace")
|
||||
w.putEntry(delivery.PlacementBucket, pkey, pl, 1, time.Now())
|
||||
@@ -180,15 +167,11 @@ func TestTheSeatsVerbs(t *testing.T) {
|
||||
hk, _ := delivery.HaveNodeKey(node)
|
||||
w.putEntry(delivery.HaveBucket, hk, hn, 1, time.Now())
|
||||
}
|
||||
if d, ok := w.composed("ace", "web", fp); !ok || d.Digest != "sha256:"+strings.Repeat("c", 64) || len(w.unplacedNodes()) != 0 {
|
||||
t.Fatalf("before the lease write: %+v %v, unplaced %v", d, ok, w.unplacedNodes())
|
||||
}
|
||||
w.putEntry(buscompose.LeaseBucket, buscompose.ControllerLeaseKey, []byte(`{"handed-to":"mesh-vault","handed-by":"operator"}`), 2, time.Now())
|
||||
if d, ok := w.composed("ace", "web", fp); !ok || d.Digest != "sha256:"+strings.Repeat("a", 64) {
|
||||
t.Errorf("after the lease write, ace: %+v %v", d, ok)
|
||||
t.Errorf("ace: %+v %v", d, ok)
|
||||
}
|
||||
if _, ok := w.composed("g14", "web", fp); ok || !slices.Equal(w.unplacedNodes(), []string{"g14"}) {
|
||||
t.Errorf("after the lease write, g14 unplaced: composed %v, unplaced %v", ok, w.unplacedNodes())
|
||||
t.Errorf("g14 unplaced: composed %v, unplaced %v", ok, w.unplacedNodes())
|
||||
}
|
||||
findings := map[string][]byte{}
|
||||
p := &passer{w: w, logf: func(string, ...any) {}, findings: assign.Findings{
|
||||
@@ -461,9 +444,10 @@ func wantsVerbs(t *testing.T) {
|
||||
man[name], _ = catalogue.Decode([]byte(kept.JSON))
|
||||
}
|
||||
findings["module.gone.policy-split"] = []byte("{}")
|
||||
// The controller writes no policy now (ADR 0571 §6): the process compares with nothing, and deletes a finding left.
|
||||
p.policySplits(man, nil)
|
||||
if _, ok := findings[assign.PolicySplitKey("web")]; !ok || len(findings) != 1 {
|
||||
t.Errorf("findings = %v, want web's split alone (a stale one deleted)", findings)
|
||||
if len(findings) != 0 {
|
||||
t.Errorf("findings = %v, want a stale one deleted", findings)
|
||||
}
|
||||
// a person's record for web: written, recorded, said; the split is gone.
|
||||
if _, err := w.doUpgrade(map[string]any{"module": "web", "policy": "record", "why": "the operator checks it"}, "person.jochen"); err != nil {
|
||||
|
||||
@@ -9,7 +9,6 @@ import (
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/handacts"
|
||||
@@ -56,13 +55,6 @@ func newWantsTools(st seatStore, ask func(string, any) (json.RawMessage, error),
|
||||
return &wantsTools{st: st, ask: ask, emit: emit, now: time.Now}
|
||||
}
|
||||
|
||||
// leased says the controller's lease is handed away.
|
||||
func (s *wantsTools) leased() bool {
|
||||
raw, _, found, err := s.st.get(buscompose.ControllerLease, buscompose.ControllerLeaseKey)
|
||||
away, _ := buscompose.LeaseHandedAway(raw, found && raw != nil, err)
|
||||
return away
|
||||
}
|
||||
|
||||
// values reads every key of a state: from the watched copy when there is one, else key by key.
|
||||
func (s *wantsTools) values(state string) (map[string][]byte, error) {
|
||||
if s.watched != nil {
|
||||
@@ -229,7 +221,7 @@ func (s *wantsTools) setPolicy(module, policy, why, by string, manifests map[str
|
||||
return nil, fmt.Errorf("the choice for %s cannot be written: %w", module, err)
|
||||
}
|
||||
}
|
||||
rec := handacts.Recorder{Ask: s.ask, LeaseAway: s.leased}
|
||||
rec := handacts.Recorder{Ask: s.ask}
|
||||
taken, rerr := rec.Record(handacts.Request{What: what, Why: why, Cause: "upgrade-policy", Verb: "mesh-wants.upgrade"})
|
||||
recorded := taken.Log
|
||||
if rerr != nil {
|
||||
|
||||
@@ -2,7 +2,6 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
@@ -19,7 +18,6 @@ import (
|
||||
"git.novox.be/novox/mesh-sdk/go/compose"
|
||||
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/provider"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
)
|
||||
@@ -28,18 +26,6 @@ import (
|
||||
// for (mesh-sdk delivery.CandidateNoneNoModule, which the indexer writes): the change removes the module.
|
||||
const noModuleWords = delivery.CandidateNoneNoModule
|
||||
|
||||
// assignment holds only the fields of the controller's Assignment the assigner reads (the view's own, mesh-delivery-view
|
||||
// internal/view/view.go): an assignment stands while it has not ended and has no state.
|
||||
type assignment struct {
|
||||
Node string `json:"node"`
|
||||
Module string `json:"module"`
|
||||
Build string `json:"build"`
|
||||
Ended time.Time `json:"ended"`
|
||||
State string `json:"state"`
|
||||
}
|
||||
|
||||
func (a assignment) open() bool { return a.Ended.IsZero() && a.State == "" }
|
||||
|
||||
// world is what the assigner watched, as it stands now: every value by bucket and key, and the manifests read by
|
||||
// digest. The watches write it; a pass reads it.
|
||||
type world struct {
|
||||
@@ -210,38 +196,15 @@ func (w *world) manifest(ctx context.Context, get getter, digest string) parsed
|
||||
return p
|
||||
}
|
||||
|
||||
// assigned is each node's modules both the controller's assignments and the seat mesh-placement'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.
|
||||
// Once the controller's lease is handed away it is the seat's placement alone, with no assignment (ADR 0549 §5, §6).
|
||||
func (w *world) assigned() (map[string][]string, map[string]assignment) {
|
||||
// A lease handed away before any placement was read keeps the assignments as they last stood, never an empty mesh:
|
||||
// the lease write's own check refuses that order (ADR 0547 §11).
|
||||
placed, seeded := w.placed()
|
||||
byAssignment, all := w.assignedByController()
|
||||
if away := w.handedAway(); away && seeded {
|
||||
both, _ := assign.Placed(byAssignment, placed, seeded, away)
|
||||
return both, map[string]assignment{}
|
||||
// assigned is each node's modules the seat mesh-placement's placement names, in name order (ADR 0549 §5, ADR 0567 §8): the
|
||||
// placement is the one source of where modules run. The controller's assignments are read no more (issue 774).
|
||||
func (w *world) assigned() map[string][]string {
|
||||
placed, _ := w.placed()
|
||||
out := make(map[string][]string, len(placed))
|
||||
for node, modules := range placed {
|
||||
out[node] = slices.Sorted(slices.Values(modules))
|
||||
}
|
||||
both, _ := assign.Placed(byAssignment, placed, seeded, false)
|
||||
return both, all
|
||||
}
|
||||
|
||||
// placementDiffers is every pair the controller's assignments and the seat's placement do not both name: none once
|
||||
// the controller's lease is handed away, since the placement is then the one source.
|
||||
func (w *world) placementDiffers() []assign.Pair {
|
||||
placed, seeded := w.placed()
|
||||
byAssignment, _ := w.assignedByController()
|
||||
_, differ := assign.Placed(byAssignment, placed, seeded, w.handedAway())
|
||||
return differ
|
||||
}
|
||||
|
||||
// handedAway says the controller's lease is handed to a holder that is no controller (mesh-sdk buscompose.LeaseHandedAway,
|
||||
// ADR 0561 §3): from then the seat mesh-placement's placement alone names where modules run (ADR 0549 §5). A lease not read (its
|
||||
// bucket not granted or not there), a controller's, expired or unreadable, is not handed away.
|
||||
func (w *world) handedAway() bool {
|
||||
raw, found := w.read(buscompose.LeaseBucket)[buscompose.ControllerLeaseKey]
|
||||
away, _ := buscompose.LeaseHandedAway(raw, found && raw != nil, nil)
|
||||
return away
|
||||
return out
|
||||
}
|
||||
|
||||
// placed is each node's modules in the seat's placement (ADR 0567 §1), and whether any placement key was read: a key
|
||||
@@ -271,24 +234,6 @@ func (w *world) placedNodes() map[string]bool {
|
||||
return out
|
||||
}
|
||||
|
||||
// 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
|
||||
if json.Unmarshal(value, &a) != nil || a.Node == "" || a.Module == "" || !a.open() {
|
||||
continue
|
||||
}
|
||||
out[a.Node] = append(out[a.Node], a.Module)
|
||||
all[a.Node+"."+a.Module] = a
|
||||
}
|
||||
for _, mods := range out {
|
||||
sort.Strings(mods)
|
||||
}
|
||||
return out, all
|
||||
}
|
||||
|
||||
// trunk is every module the indexer holds at the trunk, with its value.
|
||||
func (w *world) trunk() map[string]delivery.Module {
|
||||
out := map[string]delivery.Module{}
|
||||
@@ -307,7 +252,7 @@ func (w *world) trunk() map[string]delivery.Module {
|
||||
// mesh is the mesh as the trunk holds it, for the composition check.
|
||||
func (w *world) mesh(ctx context.Context, manifests getter) assign.Mesh {
|
||||
m := assign.Mesh{Manifests: map[string]catalogue.Manifest{}, Unreadable: map[string]string{}, Digests: map[string]string{}}
|
||||
m.Assigned, _ = w.assigned()
|
||||
m.Assigned = w.assigned()
|
||||
for name, mod := range w.trunk() {
|
||||
if mod.Manifest == "" {
|
||||
continue
|
||||
@@ -612,49 +557,14 @@ func (w *world) built(module, fp string) bool {
|
||||
return err == nil && b.State == delivery.BuildDone
|
||||
}
|
||||
|
||||
// composed is the controller's composition of module for node at fp: its digest and the judgement of the move to it
|
||||
// (mesh-controller_composed `<node>.<module>.<hex>`), and whether it exists and can be read.
|
||||
//
|
||||
// From the lease write the controller's copy is no longer read (ADR 0571 §6, a named exception of clause 10 until then): a
|
||||
// placed node composes its own, and its composition is the digest its own have names; an unplaced node has none, so it gets
|
||||
// no want (and the pass says so, unplacedFindings).
|
||||
// composed is the composition of module for node: a placed node composes its own, and its composition is the digest its own
|
||||
// have names (ADR 0571 §6); an unplaced node has none, so it gets no want (and the pass says so, unplacedFindings). The
|
||||
// controller's copy (mesh-controller_composed) is read no more (issue 774).
|
||||
func (w *world) composed(node, module, fp string) (delivery.ComposedDigest, bool) {
|
||||
if w.handedAway() {
|
||||
if !w.placedNodes()[node] {
|
||||
return delivery.ComposedDigest{}, false
|
||||
}
|
||||
return w.ownComposed(node)
|
||||
}
|
||||
key, err := delivery.ComposedKey(node, module, fp)
|
||||
if err != nil {
|
||||
if !w.placedNodes()[node] {
|
||||
return delivery.ComposedDigest{}, false
|
||||
}
|
||||
w.mu.Lock()
|
||||
value, ok := w.values[delivery.ControllerComposedBucket][key]
|
||||
w.mu.Unlock()
|
||||
if !ok {
|
||||
return delivery.ComposedDigest{}, false
|
||||
}
|
||||
d, err := delivery.DecodeComposedDigest(value)
|
||||
return d, err == nil
|
||||
}
|
||||
|
||||
// nodeComposed is, by node, the digest of the node's own composition (mesh-controller_composed `<node>.node`).
|
||||
func (w *world) nodeComposed() map[string]delivery.ComposedDigest {
|
||||
out := map[string]delivery.ComposedDigest{}
|
||||
if w.handedAway() {
|
||||
return out // the controller composes no node from the lease write (ADR 0571 §6)
|
||||
}
|
||||
for key, value := range w.read(delivery.ControllerComposedBucket) {
|
||||
node, _, _, isNode, err := delivery.ParseComposedKey(key)
|
||||
if err != nil || !isNode {
|
||||
continue
|
||||
}
|
||||
if d, err := delivery.DecodeComposedDigest(value); err == nil {
|
||||
out[node] = d
|
||||
}
|
||||
}
|
||||
return out
|
||||
return w.ownComposed(node)
|
||||
}
|
||||
|
||||
// answer is the teller's answer to the ask at key (mesh-delivery-teller_answers, the same key), and whether there is one.
|
||||
@@ -719,7 +629,7 @@ func (w *world) runningCommits(module string) []string {
|
||||
var keptBucket = "mesh-config_" + delivery.MeshConfigKeptState
|
||||
|
||||
// ownComposed is a placed node's own composition, as the digest its own have names (`<node>.node` of the haves, Composed):
|
||||
// what rule 7 reads from the lease write. A node whose have names none is not composed yet.
|
||||
// what rule 7 reads. A node whose have names none is not composed yet.
|
||||
func (w *world) ownComposed(node string) (delivery.ComposedDigest, bool) {
|
||||
key, err := delivery.HaveNodeKey(node)
|
||||
if err != nil {
|
||||
@@ -737,12 +647,9 @@ func (w *world) ownComposed(node string) (delivery.ComposedDigest, bool) {
|
||||
Words: "the node composes its own"}, true
|
||||
}
|
||||
|
||||
// unplacedNodes are, from the lease write, the nodes whose engine has a have and that no placement names: they get no
|
||||
// unplacedNodes are the nodes whose engine has a have and that no placement names: they get no
|
||||
// want, since nothing composes them.
|
||||
func (w *world) unplacedNodes() []string {
|
||||
if !w.handedAway() {
|
||||
return nil
|
||||
}
|
||||
placed := w.placedNodes()
|
||||
var out []string
|
||||
for key := range w.read(delivery.HaveBucket) {
|
||||
|
||||
@@ -117,7 +117,6 @@ func saidConditionsAreFindings(t *testing.T) {
|
||||
{HeldCondition("mail", fNew, "ace", "refused", true, true), LiftedCondition("mail", fNew), "module.mail." + strings.TrimPrefix(fNew, delivery.SourcePrefix2) + ".held"},
|
||||
{NodeFailedCondition("ace", "mail", fNew, NodeFailure{Words: "x", Others: []string{"g14"}}, true), NodeFailedCondition("ace", "mail", fNew, NodeFailure{}, false), "module.ace.mail.failed-alone"},
|
||||
{NeitherCondition("ace", "mail", Holding{Rule: 3, Words: "w"}, true), NeitherCondition("ace", "mail", Holding{}, false), "module.ace.mail.neither"},
|
||||
{PlacementCondition(Pair{Node: "ace", Module: "mail", Only: OnlyAssignments}, true), PlacementCondition(Pair{Node: "ace", Module: "mail"}, false), "module.ace.mail.placement"},
|
||||
{SeatWaitsCondition([]string{"placement"}), SeatWaitsCondition(nil), "module.mesh-delivery-assigner.seat-states"},
|
||||
{BackupCondition("ace", "mail", fNew, "disk full", true), BackupCondition("ace", "mail", fNew, "", false), "module.ace.mail." + strings.TrimPrefix(fNew, delivery.SourcePrefix2) + ".backup"},
|
||||
} {
|
||||
|
||||
@@ -1,117 +1,12 @@
|
||||
package assign
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"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 seat mesh-placement's placement (ADR 0549 §5, ADR 0567 §8). 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. After an approved proposal
|
||||
// the assigner keeps the controller in step itself, calling its assign, unassign, pin and unpin; a call refused or not
|
||||
// yet answered leaves the pair differing, and the condition names the call (Pair.Call). Before the first placement key
|
||||
// is written the controller's assignments are the one source, as before. From the lease write (the controller's lease
|
||||
// handed away) the placement is the one source.
|
||||
|
||||
// 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 seat mesh-placement'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. Once the controller's
|
||||
// lease is handed away (handedAway, ADR 0549 §5, §6) it is the placement alone, in name order, and no pair differs: the
|
||||
// assignments are then a source of nothing.
|
||||
func Placed(assigned, placed map[string][]string, seeded, handedAway bool) (map[string][]string, []Pair) {
|
||||
if handedAway && seeded {
|
||||
out := make(map[string][]string, len(placed))
|
||||
for node, modules := range placed {
|
||||
out[node] = slices.Sorted(slices.Values(modules))
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
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 seat'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("propose placing %s on %s (mesh-placement.propose), or unassign it at the controller", p.Module, p.Node)
|
||||
} else {
|
||||
needs = fmt.Sprintf("assign %s to %s at the controller, as the approved placement says, or propose taking it off", 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)}
|
||||
}
|
||||
|
||||
// SeatWaitsCondition is the assigner's word on the seat mesh-placement while a state of the seat is not bound (ADR 0567
|
||||
// §1): raised naming each, cleared when none is left.
|
||||
func SeatWaitsCondition(unbound []string) asks.Condition {
|
||||
|
||||
@@ -68,7 +68,7 @@ func StepPut(w PutWorld) PutNext {
|
||||
}
|
||||
fp := short(p.FP)
|
||||
if !w.Assigned {
|
||||
return say(delivery.PutRefused, "%s is not assigned to %s: a put moves a module where it stands assigned (mesh-controller.assign first)", w.Module, w.Node)
|
||||
return say(delivery.PutRefused, "%s is not assigned to %s: a put moves a module where it stands assigned (propose its placement first)", w.Module, w.Node)
|
||||
}
|
||||
switch w.Build {
|
||||
case delivery.BuildDone:
|
||||
|
||||
@@ -27,15 +27,10 @@ type ProposeIn struct {
|
||||
Recorded func(node string) bool
|
||||
Seed bool
|
||||
ExportDigest string
|
||||
// Leased says the lease write of ADR 0547 has happened (the controller's lease handed away, mesh-sdk
|
||||
// buscompose.LeaseHandedAway; issue 774), after which the controller's assignments are read no more (§8): until then
|
||||
// the holder keeps the controller in step and refuses what its verbs cannot take.
|
||||
Leased bool
|
||||
}
|
||||
|
||||
// Propose lays a proposal over the current keys (§2): the proposal (state answering, each changed node's base revision),
|
||||
// the laid placement of every node whose key it changes, and every finding, which refuses it whole: a change the
|
||||
// controller's verbs cannot take before the lease write (§8), a change that does not fit its node, and what
|
||||
// the laid placement of every node whose key it changes, and every finding, which refuses it whole: a change that does not fit its node, and what
|
||||
// CheckPlacements refuses of the result laid over every placement (two modules on one host port, a provider not placed,
|
||||
// a seat held once held twice, two nodes with one address, two hubs, a module the indexer does not hold, a node with no
|
||||
// record). A key whose value the proposal leaves as it is changes nothing and is not laid.
|
||||
@@ -57,11 +52,6 @@ func Propose(in ProposeIn) (delivery.PlacementProposal, map[string]*delivery.Pla
|
||||
findings = append(findings, fmt.Sprintf("change %d: the seed is adopt's alone, and never mixed with other changes", i+1))
|
||||
continue
|
||||
}
|
||||
if !in.Leased {
|
||||
if why := c.BeforeLease(); why != "" {
|
||||
findings = append(findings, fmt.Sprintf("change %d: %s", i+1, why))
|
||||
}
|
||||
}
|
||||
}
|
||||
laid, lay := delivery.LayChanges(in.Current, in.Changes)
|
||||
findings = append(findings, lay...)
|
||||
|
||||
@@ -37,39 +37,30 @@ func TestThePlacementSeatDecides(t *testing.T) {
|
||||
in ProposeIn
|
||||
want string
|
||||
}{
|
||||
"a port before the lease": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "laptop", Module: "web", Ports: map[string]int{"80": 8080}}), "lease write"},
|
||||
"a holding": {in(delivery.PlacementProposalChange{Op: delivery.OpHold, Node: "anchor", Module: "store", Claim: "mesh-store", Scope: "mesh"}), "lease write"},
|
||||
"a binding": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "laptop", Module: "web", Bindings: map[string]delivery.Provider{"db": {Node: "anchor", Module: "store"}}}), "lease write"},
|
||||
"a config place": {in(delivery.PlacementProposalChange{Op: delivery.OpConfig, Node: "anchor", Module: "web", Layer: map[string]any{"places": []any{}}}), "lease write"},
|
||||
"a config access": {in(delivery.PlacementProposalChange{Op: delivery.OpConfig, Node: "anchor", Module: "web", MeshWide: map[string]any{"accesses": []any{}}}), "lease write"},
|
||||
"a module not held": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "laptop", Module: "ghost"}), "the indexer holds no module ghost"},
|
||||
"a node with no record": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "stranger", Module: "web"}), "stranger is a node with no record"},
|
||||
"nothing changes": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "anchor", Module: "web"}), "leave every placement as it is"},
|
||||
"no why": {func() ProposeIn { i := in(move...); i.Why = " "; return i }(), "says why"},
|
||||
"no caller": {func() ProposeIn { i := in(move...); i.By = ""; return i }(), "names no caller"},
|
||||
"a seed by hand": {in(delivery.PlacementProposalChange{Op: delivery.OpSeed, Node: "laptop", Whole: &laptop}), "adopt's alone"},
|
||||
"a pin of nobody placed": {in(delivery.PlacementProposalChange{Op: delivery.OpPin, Node: "laptop", Provision: "db", Provider: &delivery.Provider{Node: "laptop", Module: "store"}}), "not placed there"},
|
||||
"a module not held": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "laptop", Module: "ghost"}), "the indexer holds no module ghost"},
|
||||
"a node with no record": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "stranger", Module: "web"}), "stranger is a node with no record"},
|
||||
"nothing changes": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "anchor", Module: "web"}), "leave every placement as it is"},
|
||||
"no why": {func() ProposeIn { i := in(move...); i.Why = " "; return i }(), "says why"},
|
||||
"no caller": {func() ProposeIn { i := in(move...); i.By = ""; return i }(), "names no caller"},
|
||||
"a seed by hand": {in(delivery.PlacementProposalChange{Op: delivery.OpSeed, Node: "laptop", Whole: &laptop}), "adopt's alone"},
|
||||
"a pin of nobody placed": {in(delivery.PlacementProposalChange{Op: delivery.OpPin, Node: "laptop", Provision: "db", Provider: &delivery.Provider{Node: "laptop", Module: "store"}}), "not placed there"},
|
||||
} {
|
||||
if _, _, f := Propose(c.in); !slices.ContainsFunc(f, func(s string) bool { return strings.Contains(s, c.want) }) {
|
||||
t.Errorf("%s: findings %v, want one naming %q", name, f, c.want)
|
||||
}
|
||||
}
|
||||
|
||||
// A module's config values may be proposed before the lease write (§13); once the lease is written every change is
|
||||
// the state's alone, and none waits.
|
||||
if _, _, f := Propose(in(delivery.PlacementProposalChange{Op: delivery.OpConfig, Node: "anchor", Module: "web", Values: map[string]any{"puid": 1}})); len(f) != 0 {
|
||||
t.Errorf("config values before the lease write: %v", f)
|
||||
}
|
||||
// Every kind of change is taken: the placement is the one source, and the controller's verbs limit nothing (ADR 0567 §8).
|
||||
for _, c := range []delivery.PlacementProposalChange{
|
||||
{Op: delivery.OpConfig, Node: "anchor", Module: "web", Values: map[string]any{"puid": 1}},
|
||||
{Op: delivery.OpPlace, Node: "laptop", Module: "web", Ports: map[string]int{"80": 8080}, Bindings: map[string]delivery.Provider{"db": {Node: "anchor", Module: "store"}}},
|
||||
{Op: delivery.OpConfig, Node: "anchor", Module: "web", Layer: map[string]any{"places": []any{}}},
|
||||
{Op: delivery.OpConfig, Node: "anchor", Module: "web", MeshWide: map[string]any{"accesses": []any{}}},
|
||||
{Op: delivery.OpDecide, Node: "laptop", Field: "site", Value: json.RawMessage(`"attic"`)},
|
||||
{Op: delivery.OpHold, Node: "anchor", Module: "store", Claim: "node-shell", Scope: "node"},
|
||||
} {
|
||||
leased := in(c)
|
||||
leased.Leased = true
|
||||
if _, _, f := Propose(leased); len(f) != 0 {
|
||||
t.Errorf("after the lease write %s is refused: %v", c.Op, f)
|
||||
if _, _, f := Propose(in(c)); len(f) != 0 {
|
||||
t.Errorf("%s is refused: %v", c.Op, f)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -33,7 +33,7 @@ func TestManifest(t *testing.T) {
|
||||
}
|
||||
reads := slices.Clone(m.Reads)
|
||||
slices.Sort(reads)
|
||||
want := []string{"mesh-build-go.steps", "mesh-conditions.takeover", "mesh-config.kept", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.composed", "mesh-controller.lease",
|
||||
want := []string{"mesh-build-go.steps", "mesh-conditions.takeover", "mesh-config.kept", "mesh-config.resolved", "mesh-controller.composed",
|
||||
"mesh-delivery-builder.build-log", "mesh-delivery-builder.builds", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-grants.grants", "mesh-provisioner.provisions", "node-engine.have"}
|
||||
if !reflect.DeepEqual(reads, want) {
|
||||
|
||||
@@ -30,9 +30,7 @@
|
||||
"mesh-conditions.takeover",
|
||||
"mesh-config.kept",
|
||||
"mesh-config.resolved",
|
||||
"mesh-controller.assignments",
|
||||
"mesh-controller.composed",
|
||||
"mesh-controller.lease",
|
||||
"mesh-delivery-builder.build-log",
|
||||
"mesh-delivery-builder.builds",
|
||||
"mesh-delivery-indexer.candidates",
|
||||
|
||||
Reference in New Issue
Block a user