The assigner reads the catalogue's placement alone once the controller's lease is handed away, and reads the lease to know it (issue 774, ADR 0549 §5, §6)

This commit is contained in:
2026-10-28 21:54:47 +01:00
parent 9a70d697af
commit 9e3d1d5e68
7 changed files with 106 additions and 19 deletions
@@ -21,6 +21,7 @@ import (
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/announce"
"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/compose/judge"
"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -8,6 +8,7 @@ package main
import (
"context"
"errors"
"fmt"
"io"
"log"
@@ -107,8 +108,33 @@ 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 catalogue'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 catalogue'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
@@ -297,7 +323,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
}
started := 0
for _, name := range watched {
if kvs[name] == nil && (slices.Contains(seatBuckets, name) || sourceOfGrants(name)) {
if kvs[name] == nil && (slices.Contains(seatBuckets, name) || sourceOfGrants(name) || name == provider.AssignmentsBucket) {
continue // not on the bus at this start (open said so)
}
started++
@@ -20,6 +20,7 @@ import (
"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
@@ -63,23 +64,30 @@ 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},
{delivery.AssignerPutsBucket, errors.New("nats: permissions violation for subscription to \"$JS.API.STREAM.INFO.KV_mesh-delivery-assigner_puts\""), true},
{delivery.AssignerPutsBucket, context.DeadlineExceeded, true},
{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},
{buscompose.LeaseBucket, errors.New("nats: permissions violation"), true},
{delivery.AssignerWantsBucket, errors.New("nats: permissions violation"), false},
{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 catalogue's placement alone is read"},
} {
var logged string
kvs, err := open(context.Background(), refusingJS{missing: c.missing, err: c.err}, func(f string, a ...any) { logged = fmt.Sprintf(f, a...) })
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...) })
switch {
case c.starts && (err != nil || kvs[c.missing] != nil || len(kvs) != len(readBuckets)+len(ownBuckets)-1 || !strings.Contains(logged, "until the assigner starts again")):
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)
case !c.starts && (err == nil || !strings.Contains(err.Error(), "grant of "+c.missing)): // 1945: named with its grant
t.Errorf("%s refused with %v: the start answered %v", c.missing, c.err, err)
@@ -118,13 +126,18 @@ 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
}
return &memKV{values: map[string][]byte{}}, nil
kv := &memKV{values: map[string][]byte{}}
if name == buscompose.LeaseBucket && r.lease != nil {
kv.values[buscompose.ControllerLeaseKey] = r.lease
}
return kv, nil
}
// watchKV is a memKV whose watch hands its entries and then the marker that all stands was handed, or ends at once when
@@ -13,6 +13,7 @@ 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"
@@ -873,6 +874,24 @@ func aPairOneSourceNamesAloneGetsNoWantAndOneCondition(t *testing.T) {
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" {
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
@@ -195,21 +195,38 @@ func (w *world) manifest(ctx context.Context, get getter, digest string) parsed
// 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) {
byAssignment, all := w.assignedByController()
// 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()
both, _ := assign.Placed(byAssignment, placed, seeded)
byAssignment, all := w.assignedByController()
if away := w.handedAway(); away && seeded {
both, _ := assign.Placed(byAssignment, placed, seeded, away)
return both, map[string]assignment{}
}
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.
// 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 {
byAssignment, _ := w.assignedByController()
placed, seeded := w.placed()
_, differ := assign.Placed(byAssignment, placed, seeded)
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
}
// placed is each node's modules in the seat's placement (ADR 0567 §1), 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) {
@@ -2,6 +2,7 @@ package assign
import (
"fmt"
"slices"
"sort"
"strings"
@@ -15,7 +16,8 @@ import (
// 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.
// 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.
@@ -32,8 +34,17 @@ const (
)
// 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) {
// 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
}
@@ -33,7 +33,7 @@ func TestManifest(t *testing.T) {
}
reads := slices.Clone(m.Reads)
slices.Sort(reads)
want := []string{"mesh-build-go.steps", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-controller.grants",
want := []string{"mesh-build-go.steps", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-controller.grants", "mesh-controller.lease",
"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"}
if !reflect.DeepEqual(reads, want) {