novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

mesh-grants: stop the controller on its placed node and read asks from the router's bucket

Browse Source 
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
This commit is contained in:
jschoubben
2026-11-02 09:42:19 +01:00
parent f1c3343929
commit 7cee3842b8
4 changed files with 91 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-grants/cmd/mesh-grants/asking.go
+6 -1
View File
@@ -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) {
modules/mesh-grants/cmd/mesh-grants/integration_test.go
+14 -2
View File
@@ -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)
}
modules/mesh-grants/internal/grants/leasewrite.go
+25 -8
View File
@@ -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
modules/mesh-grants/leasewrite_test.go
+46 -2
View File
@@ -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) {