From 7cee3842b8fb4c191fc4527fec217771de479968 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 2 Nov 2026 09:42:19 +0100 Subject: [PATCH] mesh-grants: stop the controller on its placed node and read asks from the router's bucket The lease write addressed node-service-manager at the lease holder's host, a hostname no node carries ("novox-be" on novox), so the operator's yes timed out at its first step and failed without a log line. The node now comes from placement (exactly one node placing mesh-controller, else refused by name), a failed act is logged beside its condition, and the asks record is read from messenger_asks, the bucket the composed grant names, instead of asks, which no grant allows. The fixtures give the lease a host unlike the node, so the tests no longer agree with the mistake. (issue 774) Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- modules/mesh-grants/cmd/mesh-grants/asking.go | 7 ++- .../cmd/mesh-grants/integration_test.go | 16 ++++++- .../mesh-grants/internal/grants/leasewrite.go | 33 +++++++++---- modules/mesh-grants/leasewrite_test.go | 48 ++++++++++++++++++- 4 files changed, 91 insertions(+), 13 deletions(-) diff --git a/modules/mesh-grants/cmd/mesh-grants/asking.go b/modules/mesh-grants/cmd/mesh-grants/asking.go index 73e428a..00e4446 100644 --- a/modules/mesh-grants/cmd/mesh-grants/asking.go +++ b/modules/mesh-grants/cmd/mesh-grants/asking.go @@ -22,6 +22,11 @@ import ( const recordWithin = 30 * time.Second +// messengerAsks is the router's records bucket as the seat operator-channel's manifest declares it (its `records`), the +// one the controller composes this account's direct get on (mesh-sdk buscompose SeatTrafficOf): never `asks`, a bucket +// no grant names (issue 774). The SDK names no constant for it; mesh-mail and pacman keep their own the same way. +const messengerAsks = "messenger_asks" + func publishVia(ctx context.Context, js jetstream.JetStream) func(subject string, body any, id string) error { return func(subject string, body any, id string) error { data, err := json.Marshal(body) @@ -41,7 +46,7 @@ func recordVia(ctx context.Context, conn *nats.Conn, module string) func(id stri return func(id string) (*stdio.StateEntry, error) { ctx, cancel := context.WithTimeout(ctx, recordWithin) defer cancel() - msg, err := conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET.KV_asks.$KV.asks."+module+"."+id, nil) + msg, err := conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET.KV_"+messengerAsks+".$KV."+messengerAsks+"."+module+"."+id, nil) // No responders is the bucket out of reach for a moment, never "no record": read as none, a live ask would be // closed by Reconcile a minute later and the operator's yes lost. if errors.Is(err, nats.ErrNoResponders) { diff --git a/modules/mesh-grants/cmd/mesh-grants/integration_test.go b/modules/mesh-grants/cmd/mesh-grants/integration_test.go index 6864000..6c79d19 100644 --- a/modules/mesh-grants/cmd/mesh-grants/integration_test.go +++ b/modules/mesh-grants/cmd/mesh-grants/integration_test.go @@ -9,6 +9,7 @@ import ( "fmt" "os" "path/filepath" + "slices" "strings" "sync" "testing" @@ -189,7 +190,7 @@ func TestTheLeaseWriteOnTheBus(t *testing.T) { if err != nil { t.Fatal(err) } - controller, _ := json.Marshal(buscompose.LeaseHolder{Instance: "controller@a pid 1", Host: "a", Epoch: 60}) + controller, _ := json.Marshal(buscompose.LeaseHolder{Instance: "controller@box-a pid 1", Host: "box-a", Epoch: 60}) if _, err := lease.Create(ctx, buscompose.ControllerLeaseKey, controller); err != nil { t.Fatal(err) } @@ -215,7 +216,7 @@ func TestTheLeaseWriteOnTheBus(t *testing.T) { put(bucket(grants.ReadManifests), digest, delivery.Manifest{V: delivery.Version, Module: "nats", JSON: nats}) put(bucket(grants.ReadModules), "nats", delivery.Module{V: delivery.Version, FP2: "src2:" + strings.Repeat("ab", 32), Manifest: digest, Repository: "novox/mesh-catalog", Commit: "c"}) - pl, _ := delivery.EncodePlacement(delivery.Placement{Node: "a", Modules: []delivery.PlacedModule{{Module: "nats"}}}) + pl, _ := delivery.EncodePlacement(delivery.Placement{Node: "a", Modules: []delivery.PlacedModule{{Module: "nats"}, {Module: grants.Controller}}}) if _, err := bucket(grants.ReadPlacement).Put(ctx, "placement.a", pl); err != nil { t.Fatal(err) } @@ -264,6 +265,13 @@ func TestTheLeaseWriteOnTheBus(t *testing.T) { return st.TTL() } + // Issue 774: the operator's answer is read from the router's own bucket, the one the controller grants (the seat's + // records, messenger_asks), as the process reads it. + put(bucket(messengerAsks), "mesh-grants.lease-write-0001", map[string]any{"status": "decided"}) + if rec, err := recordVia(ctx, nc, "mesh-grants")("lease-write-0001"); err != nil || rec == nil { + t.Fatalf("the router's record of the ask: %v, %v; want it read", rec, err) + } + // 774/S23/AC1. act(p.LeaseWrite) e, err := lease.Get(ctx, buscompose.ControllerLeaseKey) @@ -276,6 +284,10 @@ func TestTheLeaseWriteOnTheBus(t *testing.T) { if expiry() != 0 || seats.active != "inactive" { t.Fatalf("the lease's expiry is %s, the controller %s", expiry(), seats.active) } + // Issue 774: the stop goes to the node placement places the controller on, not the lease's host. + if !slices.Contains(seats.calls, grants.CallServiceStop+"@a") || slices.Contains(seats.calls, grants.CallServiceStop+"@box-a") { + t.Fatalf("the stops asked: %v, want one on a", seats.calls) + } if _, err := lease.Create(ctx, buscompose.ControllerLeaseKey, controller); !errors.Is(err, jetstream.ErrKeyExists) { t.Fatalf("the controller's retake: %v, want refused", err) } diff --git a/modules/mesh-grants/internal/grants/leasewrite.go b/modules/mesh-grants/internal/grants/leasewrite.go index b6915b9..f89ab6e 100644 --- a/modules/mesh-grants/internal/grants/leasewrite.go +++ b/modules/mesh-grants/internal/grants/leasewrite.go @@ -580,6 +580,9 @@ func (p *Part) advanceLease() error { } a := r.Open fail := func(why string) error { + if p.Logf != nil { + p.Logf("%s failed at its step %s: %s", a.Act, a.Step, why) + } r.end(StepFailed, why, p.now()) p.sayLeaseRefused(a.Act, why) return p.writeLease(r) @@ -708,17 +711,31 @@ func (p *Part) advanceHandBack(r *LeaseRecord, fail func(string) error) error { return nil } -// controllerNode is the node the controller holds its lease on, from the lease's holder. +// controllerNode is the node placement places mesh-controller on (issue 774). Never the lease holder's host: that is +// the machine's hostname, and no node need be named so, so a seat verb addressed to it reaches nobody. None, or more +// than one, is refused by name and nothing is stopped. func (p *Part) controllerNode() (string, string) { - raw, found, err := p.Read.Get(ReadLease, LeaseKey) - if err != nil || !found { - return "", "the lease cannot be read, so the controller's node is not known: nothing is stopped" + if raw, found, err := p.Read.Get(ReadLease, LeaseKey); err == nil && found { + var h buscompose.LeaseHolder + if json.Unmarshal(raw, &h) == nil && h.HandedTo != "" { + return "", "the lease is handed to " + h.HandedTo + ", not the controller's: nothing is stopped" + } } - var h buscompose.LeaseHolder - if json.Unmarshal(raw, &h) != nil || h.Host == "" || h.HandedTo != "" { - return "", "the lease names no controller's host, so the controller's node is not known: nothing is stopped" + in := ReadInputs(p.Read, &p.cache) + var nodes []string + for node, modules := range in.Assigned { + if slices.Contains(modules, Controller) { + nodes = append(nodes, node) + } } - return h.Host, "" + slices.Sort(nodes) + switch len(nodes) { + case 0: + return "", "no node places " + Controller + ", so the controller's node is not known: nothing is stopped" + case 1: + return nodes[0], "" + } + return "", "two or more nodes place " + Controller + " (" + strings.Join(nodes, ", ") + "), so the controller's node is not one: nothing is stopped" } // service asks the node seat's verb of the controller's unit and reads it back with status: stopped for a stop, started diff --git a/modules/mesh-grants/leasewrite_test.go b/modules/mesh-grants/leasewrite_test.go index 44f56ef..59c263b 100644 --- a/modules/mesh-grants/leasewrite_test.go +++ b/modules/mesh-grants/leasewrite_test.go @@ -32,6 +32,7 @@ type leaseFake struct { expiry time.Duration // the lease bucket's ttls map[string]time.Duration log []string // the write's acts on the bus, in order + logged []string // the process's log lines } func newLeaseFake(t *testing.T) (*leaseFake, *grants.Part) { @@ -39,8 +40,11 @@ func newLeaseFake(t *testing.T) (*leaseFake, *grants.Part) { b := fixture() b.manifestOf("nats", `{"module":"nats","version":"1","claims":[{"name":"mesh-broker"}],"own-secrets":{"broker":"/var/lib/nats/broker"},`+ `"tools":["nats_bucket_expiry","nats_consumer_remove","nats_stream_remove"]}`) - b.put(grants.ReadLease, grants.LeaseKey, map[string]any{"instance": "controller@a pid 1", "host": "a", "epoch": 60, + // The lease's host is the machine's hostname, never a node's name (issue 774: "novox-be" on the node novox); the + // controller's node is the one placement places it on. + b.put(grants.ReadLease, grants.LeaseKey, map[string]any{"instance": "controller@box-a pid 1", "host": "box-a", "epoch": 60, "taken": "2026-10-30T14:52:10Z"}) + b.assign("a", grants.Controller) b.put(grants.ReadTakeover, health.TakeoverKey, health.Takeover{V: 1, Follow: health.FollowKeeper, By: "operator", Why: "the keeper took over"}) busKey, err := seal.GenerateKeyPair() if err != nil { @@ -88,6 +92,7 @@ func newLeaseFake(t *testing.T) (*leaseFake, *grants.Part) { p.Call, p.Store, p.Lease, p.Asker, p.Objects = f, ttlStore{ownStore{b}, f}, f, &fakeAsker{wheres: map[string]*asks.Warrant{}}, &fakeObjects{} p.SaySeat = func(event string, _ any) error { f.log = append(f.log, "event "+event); return nil } p.Nonce = func() string { return "0001" } + p.Logf = func(format string, a ...any) { f.logged = append(f.logged, fmt.Sprintf(format, a...)) } return f, p } @@ -340,7 +345,7 @@ func TestTheLeaseWrite(t *testing.T) { // 774/S18/AC1: the key or the expiry changed by another hand is said once, and nothing is written. f.log = nil - f.put(grants.ReadLease, grants.LeaseKey, map[string]any{"instance": "controller@a pid 9", "host": "a", "epoch": 61}) + f.put(grants.ReadLease, grants.LeaseKey, map[string]any{"instance": "controller@a pid 9", "host": "box-a", "epoch": 61}) p.Compose() p.Compose() f.expiry = 15 * time.Second @@ -350,6 +355,45 @@ func TestTheLeaseWrite(t *testing.T) { } }) + // Issue 774 (the lease write met at novox): the controller's service is stopped on the node placement places + // mesh-controller on, never on the lease's host (a hostname); none or two such nodes refuse by name, and a failed act + // is logged beside its condition. + t.Run("the controller's node", func(t *testing.T) { + for _, row := range []struct { + name, stop, said string + place func(f *leaseFake) + }{ + {"placed on a", "a", "", func(*leaseFake) {}}, + {"placed nowhere", "", "no node places mesh-controller", func(f *leaseFake) { + delete(f.states[grants.ReadPlacement], "placement.a") + f.assign("a", "nats") + f.assign("a", "mod-x") + }}, + {"placed twice", "", "two or more nodes place mesh-controller (a, b)", func(f *leaseFake) { f.assign("b", grants.Controller) }}, + } { + f, p := newLeaseFake(t) + if _, err := p.LeaseWrite("operator@shanks", "the controller goes"); err != nil { + t.Fatal(err) + } + p.Compose() + row.place(f) + if err := warrant(t, p, "yes"); err != nil { + t.Fatal(err) + } + stops := slices.DeleteFunc(slices.Clone(f.calls), func(c string) bool { return !strings.HasPrefix(c, grants.CallServiceStop+"@") }) + r := record(t, p) + switch { + case row.stop != "" && (!slices.Equal(stops, []string{grants.CallServiceStop + "@" + row.stop}) || r.Node != row.stop): + t.Errorf("%s: stopped %v, node %q; want the stop on %s alone", row.name, stops, r.Node, row.stop) + case row.stop == "" && (len(stops) != 0 || r.Open != nil || len(r.Acts) != 1 || r.Acts[0].Step != grants.StepFailed || + !strings.Contains(r.Acts[0].Said, row.said)): + t.Errorf("%s: stopped %v, %+v; want the act failed saying %q, nothing stopped", row.name, stops, r, row.said) + case row.stop == "" && !slices.ContainsFunc(f.logged, func(l string) bool { return strings.Contains(l, row.said) }): + t.Errorf("%s: logged %q, want a line saying %q", row.name, f.logged, row.said) + } + } + }) + // 774/S19/AC1, AC2: hand-back restores the expiry, deletes holder and removal-due, and starts the controller only once // the list with its rights is loaded; once the removal ran it is refused. t.Run("hand-back", func(t *testing.T) {