Grants part: hear root-not-free from the seat the take-over key names, never both, so the switch to the condition keeper moves it with the router (issue 1952 G1)

Watchman: subscribe exactly what its grant lets it hear (buscompose.WatchmanHears). Both pin mesh-sdk 69fcb6c.
This commit is contained in:
2026-10-29 21:46:29 +01:00
parent da787ae360
commit b8e8981730
11 changed files with 162 additions and 29 deletions
@@ -24,6 +24,20 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6
t.Errorf("an essential bucket refused: %v, want its grant named", err)
}
})
t.Run("1952/WM2: every subject its grant lets it hear goes to a row", func(t *testing.T) {
if len(HeardSubjects) < 7 {
t.Fatalf("hears %v", HeardSubjects)
}
for _, s := range HeardSubjects {
concrete := strings.ReplaceAll(strings.ReplaceAll(s, "*", "x"), ">", "x.y")
if routeOf(concrete) == "" {
t.Errorf("%s (%s) goes to no row", s, concrete)
}
}
if routeOf("mesh.mod.web.event.started") != "" {
t.Error("another module's event goes to a row")
}
})
err := run(context.Background(), func(string) string { return "" }, io.Discard)
if err == nil || !strings.Contains(err.Error(), "MESH_BROKER_FILE") {
t.Fatalf("run = %v, want a refusal naming MESH_BROKER_FILE", err)
@@ -13,6 +13,7 @@ import (
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/health"
"mesh-delivery-watchman/internal/deadletters"
"mesh-delivery-watchman/internal/socket"
@@ -21,40 +22,62 @@ import (
// The subjects the watchman hears for the signals table's rows (novox/hq ADR 0564 §4, Limits 3 and 11), each on the bus,
// none in the controller's store. Its grant names them (mesh-sdk buscompose, the watchman's principal).
const CheckedSubject = "mesh.seat." + health.SelfCheckModule + ".event." + health.EventChecked
// HeardSubjects is every subject the watchman subscribes for its rows, beside the bus's connect and disconnect: the ones
// its grant lets it hear (buscompose.WatchmanHears), so the two cannot drift.
var HeardSubjects = buscompose.WatchmanHears()
// The rows a heard subject goes to.
const (
PowerSubject = "mesh.mod.power.event.*"
FailingSubject = "mesh.mod.*.event.provisioner.failing"
RecoveredSubject = "mesh.mod.*.event.provisioner.recovered"
ToolsAliveSubject = "mesh.control.*.tools-alive"
CheckedSubject = "mesh.seat.mesh-self-check.event.checked"
DeadKeptSubject = "mesh.events.dead.>"
routePower = "power"
routeStanding = "standing"
routeTools = "tools-alive"
routeChecked = "checked"
routeAdvisory = "advisory"
routeDead = "dead-letter"
)
// HeardSubjects is every subject the watchman subscribes for its rows, beside the bus's connect and disconnect.
var HeardSubjects = []string{PowerSubject, FailingSubject, RecoveredSubject, ToolsAliveSubject, CheckedSubject,
watch.ConsumerDeletedAdvisories, DeadKeptSubject}
// routeOf is the row a heard subject goes to; empty for one the watchman does not read.
func routeOf(subject string) string {
switch {
case strings.HasPrefix(subject, "mesh.mod.power.event."):
return routePower
case strings.HasSuffix(subject, ".event.provisioner.failing") || strings.HasSuffix(subject, ".event.provisioner.recovered"):
return routeStanding
case strings.HasSuffix(subject, ".tools-alive"):
return routeTools
case subject == CheckedSubject:
return routeChecked
case strings.HasPrefix(subject, "$JS.EVENT.ADVISORY."):
return routeAdvisory
case strings.HasPrefix(subject, "mesh.events.dead."):
return routeDead
}
return ""
}
// heard hands one message of a row's subject to the watchman.
func heard(w *watch.Watchman, dead *deadKeeper, m *nats.Msg, logf func(string, ...any)) {
switch {
case strings.HasPrefix(m.Subject, "mesh.mod.power.event."):
switch routeOf(m.Subject) {
case routePower:
w.Power(m.Header.Get("x-node"), strings.TrimPrefix(m.Subject, "mesh.mod.power.event."))
case strings.HasSuffix(m.Subject, ".event.provisioner.failing") || strings.HasSuffix(m.Subject, ".event.provisioner.recovered"):
case routeStanding:
st, err := watch.ReadStanding(m.Subject, m.Data)
if err != nil {
logf("%v; left", err)
return
}
w.Stood(st)
case strings.HasSuffix(m.Subject, ".tools-alive"):
case routeTools:
var body struct {
Every float64 `json:"every"`
}
_ = json.Unmarshal(m.Data, &body)
w.ToolsAlive(watch.ToolsNode(m.Subject), time.Duration(body.Every*float64(time.Second)))
case m.Subject == CheckedSubject:
case routeChecked:
w.Checked(m.Header.Get("x-node"))
case strings.HasPrefix(m.Subject, "$JS.EVENT.ADVISORY."):
case routeAdvisory:
a, ok := watch.ReadAdvisory(m.Subject, m.Data, deadletters.ConsumerInWords)
if !ok {
return
@@ -64,7 +87,7 @@ func heard(w *watch.Watchman, dead *deadKeeper, m *nats.Msg, logf func(string, .
return
}
w.Advised(a)
case strings.HasPrefix(m.Subject, "mesh.events.dead."):
case routeDead:
dead.count()
}
}
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-watchman
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204315-2a04bf0ac5bf
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
)
+2 -2
View File
@@ -1,5 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204315-2a04bf0ac5bf h1:Yy51rUWqyk+cZPAcSX02dnK4JVSTUJIfLa9j+741+1Y=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204315-2a04bf0ac5bf/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba h1:uXwkcQWgbu/UthizGHFaKPiA4NBVJgqL9q2jJnnGR1o=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op h1:1BOWQJweNyvZMlpAHXGLiZQn9S+QXGcz3xh94lC0w6E=
github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-grants
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029200316-22c9ddd32b6a
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba
github.com/nats-io/nats.go v1.54.0
)
+2 -2
View File
@@ -1,5 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029200316-22c9ddd32b6a h1:nhRFmtdcI4VkghGUdJhpt1BnPuTQcekioOIRzW8fcG0=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029200316-22c9ddd32b6a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba h1:uXwkcQWgbu/UthizGHFaKPiA4NBVJgqL9q2jJnnGR1o=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
+33
View File
@@ -22,6 +22,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/health"
"git.novox.be/novox/mesh-sdk/go/outward"
"git.novox.be/novox/mesh-sdk/go/seal"
@@ -656,8 +657,40 @@ func TestGrants(t *testing.T) {
if passes() != before {
t.Errorf("a condition of another kind woke a pass")
}
// 1952/G1 (ADR 0564 §9): the keeper's events are not heard while the take-over key follows the controller
handle("mesh.seat.mesh-conditions.event.condition-raised", []byte(`{"kind":"root-not-free","key":"machine.a.root-not-free"}`))
time.Sleep(100 * time.Millisecond)
if passes() != before || wf.Follows() != "mesh-controller" {
t.Errorf("the keeper's event woke a pass while the part follows %s", wf.Follows())
}
handle(grants.EventConditionCleared, []byte(`{"kind":"root-not-free","key":"machine.a.root-not-free"}`))
waitFor(t, func() bool { return passes() > before })
// and from the take-over only the keeper's are heard, never both
watcher := wf.Watch.(*fakeWatcher)
waitFor(t, func() bool {
watcher.mu.Lock()
defer watcher.mu.Unlock()
return watcher.changes[grants.ReadTakeover] != nil
})
watcher.mu.Lock()
toKeeper := watcher.changes[grants.ReadTakeover]
watcher.mu.Unlock()
toKeeper(health.TakeoverKey, []byte(`{"v":1,"follow":"keeper","by":"operator","at":"2026-10-29T20:00:00Z"}`))
if wf.Follows() != health.KeeperModule {
t.Fatalf("the part follows %s after the take-over", wf.Follows())
}
before = passes()
handle("mesh.seat.mesh-controller.event.condition-raised", []byte(`{"kind":"root-not-free","key":"machine.a.root-not-free"}`))
time.Sleep(100 * time.Millisecond)
if passes() != before {
t.Errorf("the controller's event woke a pass after the take-over")
}
handle(grants.EventKeeperRaised, []byte(`{"kind":"root-not-free","key":"machine.a.root-not-free"}`))
waitFor(t, func() bool { return passes() > before })
toKeeper(health.TakeoverKey, nil) // handed back: the key deleted follows the controller
if wf.Follows() != "mesh-controller" {
t.Errorf("the part follows %s after the hand-back", wf.Follows())
}
wcancel()
<-wdone
if !grants.IsRootNotFree([]byte(`{"kind":"root-not-free"}`)) || !grants.IsRootNotFree([]byte(`{"key":"machine.novox.root-not-free"}`)) ||
+43 -2
View File
@@ -8,6 +8,7 @@ import (
"git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/health"
)
// Watcher watches another part's state: changed hears each change after what stands, ended hears the watch end. A watch
@@ -19,7 +20,7 @@ type Watcher interface {
// Followed are the states whose change composes again: placement, the indexer's manifests by module, the resolved
// config values, the node agents' haves (a machine's first have, and the bus node's loaded list) and the assigner's
// wants (the manifest each machine is wanted to run), and the vault's hashes of the bus accounts.
var Followed = []string{ReadAssignments, ReadModules, ReadResolved, ReadHave, ReadWants, ReadAccounts, ReadPlacement, ReadLease}
var Followed = []string{ReadAssignments, ReadModules, ReadResolved, ReadHave, ReadWants, ReadAccounts, ReadPlacement, ReadLease, ReadTakeover}
// Follower composes again on every change it hears, never on a timer. It never stops for a refusal: a watch it cannot
// open is said, composed without, and opened again at the next wake (any change it does hear, or a reconnect).
@@ -49,6 +50,19 @@ type Follower struct {
refusedMans string
// leaseOn is whether the lease was handed away at its last change heard, once leaseSeen.
leaseOn, leaseSeen bool
// follows is the seat whose condition events the part hears, as the take-over key names it: the controller's until
// the operator's take-over (an absent key, or one not readable, follows the controller).
follows string
}
// Follows is the seat whose condition events the part hears now.
func (f *Follower) Follows() string {
f.mu.Lock()
defer f.mu.Unlock()
if f.follows == "" {
return "mesh-controller"
}
return f.follows
}
func (f *Follower) init() {
@@ -88,6 +102,25 @@ func (f *Follower) logf(format string, args ...any) {
// heard says whether a change composes again: every change but a have's, and of the haves a machine's first and the
// bus module's loaded list moving.
func (f *Follower) heard(state, key string, value []byte) bool {
if state == ReadTakeover {
// Which seat's condition events are heard: no pass, since nothing composed reads it.
if key != health.TakeoverKey {
return false
}
t, err := health.DecodeTakeover(value, len(value) > 0)
if err != nil {
f.logf("the take-over key cannot be read (%v): the controller's condition events are heard", err)
t = health.Takeover{Follow: health.FollowController}
}
f.mu.Lock()
was := f.follows
f.follows = t.Seat()
f.mu.Unlock()
if was != t.Seat() {
f.logf("root-free words are heard from %s's condition events (the take-over key follows %s)", t.Seat(), t.Follow)
}
return false
}
if state == ReadLease {
// Renewed every few seconds by its holder: only the write that hands it away, or back, composes again (ADR 0561 §6).
on, _ := buscompose.LeaseHandedAway(value, len(value) > 0, nil)
@@ -136,7 +169,15 @@ func (f *Follower) open() {
}
if f.Hear != nil && !f.heard_ {
err := f.Hear(func(event string, body []byte) {
if IsAccountIssued(event) || IsRootNotFree(body) {
if IsAccountIssued(event) {
f.Wake()
return
}
// A condition event of the seat the take-over key does not name is not heard (ADR 0564 §9).
if from := HeardFrom(event); from != "" && from != f.Follows() {
return
}
if IsRootNotFree(body) {
f.Wake()
}
})
@@ -42,7 +42,7 @@ const (
// controller holds the lease, when the assignments are unioned with the placement (ReadInputs); after the lease write
// they place nothing, and the first change after it takes them out (../assignments-readers names the part until then).
var Reads = []string{ReadAssignments, ReadControllerGrant, ReadModules, ReadManifests, ReadResolved, ReadHave, ReadWants, ReadBuilds,
ReadAccounts, ReadPlacement, ReadLease}
ReadAccounts, ReadPlacement, ReadLease, ReadTakeover}
// The bus module's tools the part calls, as its manifest's invokes name them: read-only, each.
const (
@@ -8,27 +8,46 @@ import (
"strings"
"git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/health"
)
// CallRootFree is the controller seat's verb that judges machines root-free (novox/hq ADR 0259 §8): the one judgement
// the composer needs and the part cannot make from the bus (issue 1945). Only the controller answers it, on purpose.
const CallRootFree = "seat:mesh-controller.root-free"
// The controller seat's condition events the part hears (never a timer): a machine's root-free word changes with the
// raising and the clearing of its `root-not-free` condition.
// The condition events the part hears (never a timer): a machine's root-free word changes with the raising and the
// clearing of its `root-not-free` condition. They are the controller seat's until the operator's take-over, and the
// condition keeper's (mesh-conditions) from it: the part hears only the seat the take-over key names, never both (novox/hq
// ADR 0564 §9, issue 1952 G1).
const (
EventConditionRaised = "mesh-controller.condition-raised"
EventConditionCleared = "mesh-controller.condition-cleared"
EventKeeperRaised = health.KeeperModule + ".condition-raised"
EventKeeperCleared = health.KeeperModule + ".condition-cleared"
// KindRootNotFree is the condition kind whose events wake a pass.
KindRootNotFree = "root-not-free"
)
// ReadTakeover is the condition keeper's take-over key's state (`switch`): which seat's condition events the part hears.
const ReadTakeover = health.KeeperModule + "." + health.TakeoverState
// HeardFrom is the seat a condition event heard comes from, by its name (`<seat>.<event>`) or its bus subject
// (`mesh.seat.<seat>.event.<event>`); empty for any other.
func HeardFrom(event string) string {
for _, seat := range []string{health.KeeperModule, "mesh-controller"} {
if strings.HasPrefix(event, seat+".") || strings.HasPrefix(event, "mesh.seat."+seat+".event.") {
return seat
}
}
return ""
}
// EventAccountIssued is the vault's word that it minted an account (novox/hq ADR 0561 §1): its hash is in the vault's
// bus-accounts, so a pass composes the list with it.
const EventAccountIssued = "mesh-vault." + buscompose.AccountIssuedEvent
// Consumes is every event the part hears, as its manifest's consumes name them.
var Consumes = []string{EventConditionRaised, EventConditionCleared, EventAccountIssued}
var Consumes = []string{EventConditionRaised, EventConditionCleared, EventKeeperRaised, EventKeeperCleared, EventAccountIssued}
// IsAccountIssued says an event heard is the vault's account-issued, by its name or its bus subject.
func IsAccountIssued(event string) bool {
+4 -1
View File
@@ -13,11 +13,14 @@
"mesh-delivery-assigner.placement",
"node-engine.have",
"mesh-vault.bus-accounts",
"mesh-controller.lease"
"mesh-controller.lease",
"mesh-conditions.takeover"
],
"consumes": [
"mesh-controller.condition-raised",
"mesh-controller.condition-cleared",
"mesh-conditions.condition-raised",
"mesh-conditions.condition-cleared",
"mesh-vault.account-issued"
],
"uses": [