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

Merge pull request 'The lease write stops the controller on its placed node and reads the operator's answer from the right bucket (issue 774)' (#135) from fix/774-lease-write-node into main

Browse Source
This commit was merged in pull request #135.
This commit is contained in:
mesh-tools
2026-11-02 08:52:47 +00:00
parent f1c3343929 8f87dbe94d
commit cc9c681fa3
6 changed files with 120 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go
+10 -4
View File
@@ -37,14 +37,20 @@ func publishVia(ctx context.Context, js jetstream.JetStream) func(subject string
}
}
// recordVia reads the router's record of one of this module's asks: the key `<module>.<id>` of the router's bucket asks,
// by a direct get on the subject the controller grants for it, `$JS.API.DIRECT.GET.KV_asks.$KV.asks.<module>.>`. The
// record is {state, warrant}. An ask the router holds no record of is nil.
// 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.
const messengerAsks = "messenger_asks"
// recordVia reads the router's record of one of this module's asks: the key `<module>.<id>` of the router's bucket
// messenger_asks, by a direct get on the subject the controller grants for it,
// `$JS.API.DIRECT.GET.KV_messenger_asks.$KV.messenger_asks.<module>.>`. The record is {state, warrant}. An ask the
// router holds no record of is nil.
func recordVia(ctx context.Context, conn *nats.Conn, module string) func(id string) (*stdio.StateEntry, error) {
return func(id string) (*stdio.StateEntry, error) {
ctx, cancel := context.WithTimeout(ctx, callTimeout)
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)
if errors.Is(err, nats.ErrNoResponders) {
return nil, nil
}
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go
+19 -4
View File
@@ -130,9 +130,9 @@ func TestTellerWaitsForTheGate(t *testing.T) { // 1240/AC2
// them (internal/broker): a module's own state is the read grants of its bucket and `$KV.<bucket>.>` to write it
// (state.go stateGrants, readGrants), a read of another module's bucket is the read grants alone, and a module that uses
// the seat operator-channel submits its own `accept.ask.<module>`, `accept.cancel.<module>` and `accept.condition.<module>`,
// hears its own `event.decided.<module>` and reads `$KV.asks.<module>.>` of the router's record (seattraffic.go
// hears its own `event.decided.<module>` and reads `$KV.messenger_asks.<module>.>` of the router's record (seattraffic.go
// SeatTrafficOf, with the messenger's seat declaring accepts ask, cancel, condition, emits decided, by-caller and records
// asks). The gate is a bundle on the tool runner's account (ADR 0437 §3), so the gate's writes are the account node-tools'.
// messenger_asks). The gate is a bundle on the tool runner's account (ADR 0437 §3), so the gate's writes are the account node-tools'.
// The permissions are these lists written out, not the controller's function: the broker tests of issue 1243 prove that.
func readOf(bucket string) []string {
stream := "KV_" + bucket
@@ -176,7 +176,7 @@ const (
func fullConf(dir string) string {
teller := append(append(append(keepsOf(tellerAnswers), keepsOf(tellerAsked)...), readOf(gateAsksKV)...),
seatAsk, seatCancel, seatCondition, "$JS.API.DIRECT.GET.KV_asks.$KV.asks.mesh-delivery-teller.>")
seatAsk, seatCancel, seatCondition, "$JS.API.DIRECT.GET.KV_"+messengerAsks+".$KV."+messengerAsks+".mesh-delivery-teller.>")
teller = append(append(teller, readOf(assignerAsks)...), readOf(viewAnswers)...) // issue 1297 B10: the manifest's reads
view := append(append(append(readOf(tellerAnswers), readOf(gateAsksKV)...), readOf(assignerAsks)...), keepsOf(viewAnswers)...)
assigner := keepsOf(assignerAsks)
@@ -244,7 +244,7 @@ func startFull(t *testing.T) *fullBed {
}
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
t.Cleanup(cancel)
for name, history := range map[string]uint8{gateAsksKV: delivery.AsksHistory, tellerAnswers: delivery.AnswersHistory, tellerAsked: 1, assignerAsks: delivery.AssignerAsksHistory, viewAnswers: delivery.ViewAnswersHistory} {
for name, history := range map[string]uint8{gateAsksKV: delivery.AsksHistory, tellerAnswers: delivery.AnswersHistory, tellerAsked: 1, messengerAsks: 1, assignerAsks: delivery.AssignerAsksHistory, viewAnswers: delivery.ViewAnswersHistory} {
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history}); err != nil {
t.Fatal(err)
}
@@ -331,6 +331,21 @@ func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2
rn := startRunner(b.ctx, func(c context.Context) error { return run(c, func(k string) string { return env[k] }, logs) })
t.Cleanup(func() { rn.finish(t, 10*time.Second) })
// Issue 774: the router's record is read on the teller's own credential from the bucket the controller grants it.
if mkv, err := b.js.KeyValue(b.ctx, messengerAsks); err != nil {
t.Fatal(err)
} else if _, err := mkv.Put(b.ctx, "mesh-delivery-teller.r1", []byte(`{"state":"open"}`)); err != nil {
t.Fatal(err)
}
tc, err := nats.Connect(b.srv.ClientURL(), nats.UserInfo("n1.mesh-delivery-teller", "pw"), nats.CustomInboxPrefix("_INBOX.n1.mesh-delivery-teller"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(tc.Close)
if rec, err := recordVia(b.ctx, tc, "mesh-delivery-teller")("r1"); err != nil || rec == nil {
t.Fatalf("the router's record read as the teller: %v, %v; want it", rec, err)
}
key, value, ga := gateAsk(t, twoHeads(), twoHeads())
kv, err := tools.KeyValue(b.ctx, gateAsksKV)
if err != nil {
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) {