C1c, C6: any holder serves drop and every node follows it by a watch of the shared pauses; the build modules serve drop and ask the tool runner's seat-settle-delay (ADR 0664 §1, §4, §5, issue 1748)

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
2026-11-09 23:19:28 +01:00
parent 3c3fdd3446
commit 6b8e7fc621
16 changed files with 74 additions and 13 deletions
+5
View File
@@ -84,6 +84,10 @@ type Puller struct {
// deleted when the ask is refused at pull, and the start sweeps the keys of asks the stream no longer holds (Drop).
ForgetPause func(key string) error
PauseKeys func() ([]string, error)
// WatchPauses follows the keys of the pauses state that key names (`**` for the rest): stdio.State(...).Watch in a module. The
// state is the module's, shared by every node that holds the seat, so a drop another node's holder wrote reaches the node
// that runs the ask (Drop).
WatchPauses func(key string, handle func(stdio.StateChange) error) error
// WatchConfig, when set, follows the module's config value `capacity` (CapacitySetting) as mesh-config resolves it
// for this node, and wakes the puller on each change (watchCapacity): WatchConfigServed in a module, a test's own
// double. Until it hands a value the puller judges by Capacity.
@@ -438,6 +442,7 @@ func (p *Puller) Pull(ctx context.Context) error {
p.logf("the pause of %s cannot be read: %v; pulling as not paused", p.Node, err)
}
p.sweepDropped()
p.watchDrops()
p.follow() // WatchConfigServed waits for Serve, which a module starts after its pull
go p.watchSettings(ctx)
wait := p.PullWait
+34 -1
View File
@@ -10,12 +10,16 @@ import (
buildrun "git.novox.be/novox/mesh-delivery/sdk/buildrun"
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// **The seat's verb drop ends an ask its caller no longer waits for** (novox/hq ADR 0664 §5, issue 1748; research 086
// choice 28's cancel). It writes the ask in this node's own pauses state under dropped.<ask id> (the part build-step), so a
// holder that restarts still refuses the ask when the bus gives it again, and ends the step if it runs here: the step's
// context is cancelled, its process ended, and its key is written refused naming who dropped it and why.
// context is cancelled, its process ended, and its key is written refused naming who dropped it and why. The state is the
// module's, shared by every node: any holder answers the verb, every node refuses the ask when it pulls it (an ask that
// waits leaves the stream at its next pull, with no work done), and the node that runs it hears the key by its watch
// (watchDrops) and ends the step.
// DroppedKeep is how long a dropped ask's key stays: the start deletes older ones. An ask the seat's stream still holds is
// younger than its stream's maximum age, which no contract this module reads states, so the bound is a generous week.
@@ -79,6 +83,35 @@ func (p *Puller) Drop(id, by, why string) error {
return nil
}
// watchDrops follows the drops written in the pauses state, by any holder of the seat: one for an ask that runs here ends its
// step. An ask that waits is refused when it is pulled (take), on whichever node pulls it. A watch that cannot start is said;
// the pull then refuses a dropped ask but cannot end a running one.
func (p *Puller) watchDrops() {
if p.WatchPauses == nil {
return
}
err := p.WatchPauses(sdkstep.DroppedPrefix+"**", func(ch stdio.StateChange) error {
id, ok := strings.CutPrefix(ch.Key, sdkstep.DroppedPrefix)
if !ok || ch.Op != "put" {
return nil
}
var d sdkstep.Drop
if json.Unmarshal(ch.Value, &d) != nil {
return nil
}
p.mu.Lock()
cancel := p.cancels[id]
p.mu.Unlock()
if cancel != nil {
cancel(&droppedError{by: d.By, why: d.Why})
}
return nil
})
if err != nil {
p.logf("the drops in the pauses of %s cannot be followed: %v; a dropped ask that runs here is not ended", p.Node, err)
}
}
// droppedAsk is the drop kept for the ask, nil for none (or no state to read it from).
func (p *Puller) droppedAsk(id string) *sdkstep.Drop {
if p.Load == nil {
@@ -89,8 +89,10 @@ func TestADroppedAskIsRefusedAtPullAndARunningOneIsEnded(t *testing.T) {
a := testAsk(sdkstep.AskDelivery)
started := make(chan struct{})
ran := false
var heard func(stdio.StateChange) error
p := &Puller{Source: src, Language: sdkstep.AskGo, Node: "shanks", Put: st.put, PausePut: kv.put, Load: kv.load, ForgetPause: kv.forget, PauseKeys: kv.keys,
Capacity: func() int { return 1 }, PullWait: 10 * time.Millisecond,
WatchPauses: func(_ string, h func(stdio.StateChange) error) error { heard = h; return nil },
Capacity: func() int { return 1 }, PullWait: 10 * time.Millisecond,
Run: func(ctx context.Context, _ sdkstep.StepAsk) (any, error) {
ran = true
close(started)
@@ -107,7 +109,10 @@ func TestADroppedAskIsRefusedAtPullAndARunningOneIsEnded(t *testing.T) {
go p.Pull(ctx)
if running {
<-started
if err := p.Drop(a.ID, "mesh-delivery-builder", "superseded"); err != nil {
// Another node's holder served the verb: its write reaches this node by the watch of the shared pauses state.
raw, _ := json.Marshal(sdkstep.Drop{V: sdkstep.Version, By: "mesh-delivery-builder", Why: "superseded", At: time.Now()})
kv.put(sdkstep.DropKey(a.ID), json.RawMessage(raw))
if err := heard(stdio.StateChange{Key: sdkstep.DropKey(a.ID), Op: "put", Value: raw}); err != nil {
t.Fatal(err)
}
}
@@ -113,7 +113,7 @@ func (m *Module) Puller(steps []StepTool, source buildstep.AskSource) *buildstep
return raw, err
}
}
return &buildstep.Puller{Source: source, Language: sdkstep.AskGo, Node: m.Node, Put: m.Steps, PausePut: m.Pauses, Load: load, ForgetPause: m.ForgetPause, PauseKeys: m.ReadPauses.Keys,
return &buildstep.Puller{Source: source, Language: sdkstep.AskGo, Node: m.Node, Put: m.Steps, PausePut: m.Pauses, Load: load, ForgetPause: m.ForgetPause, WatchPauses: m.WatchPauses, PauseKeys: m.ReadPauses.Keys,
Capacity: capacity, SettingsFile: settings, WatchConfig: watch, CoresPerStep: cores, BuildHours: hours, Logf: m.Say, Has: m.Has,
Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) {
if err := m.Check(ctx); err != nil {
@@ -118,7 +118,7 @@ func mustJSON(v any) []byte { b, _ := json.Marshal(v); return b }
// the node's build capacity as a setting, with the builder's default; the seat's verbs are served as its tools.
func TestManifestClaimsTheGoBuildSeatAndReadsItsCapacity(t *testing.T) {
m := readManifest(t)
if len(m.Claims) != 1 || m.Claims[0].Name != Seat || !slices.Equal(m.Claims[0].Serves, []string{"current", "pause", "resume"}) {
if len(m.Claims) != 1 || m.Claims[0].Name != Seat || !slices.Equal(m.Claims[0].Serves, []string{"current", "drop", "pause", "resume"}) {
t.Fatalf("claims = %+v", m.Claims)
}
if !slices.Contains(m.Reads, KeptState) {
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"fmt"
stdio "git.novox.be/novox/mesh-sdk/go"
"net/http"
"sync"
"time"
@@ -37,6 +38,8 @@ type Module struct {
ReadPauses StateRead
// ForgetPause deletes one key of the pauses state: a dropped ask's, once it is refused (the seat's verb drop).
ForgetPause func(key string) error
// WatchPauses follows the drops other holders write in the shared pauses state (stdio.State(...).Watch).
WatchPauses func(key string, handle func(stdio.StateChange) error) error
// Emit tells the bus an event (stdio.Emit); nil emits nothing. The guard emits toolchain.reported after each write of
// the toolchains key, so the indexer, which holds no timer, learns that the bucket exists (issue 1444).
Emit func(eventType string, body any) error
+2 -1
View File
@@ -9,6 +9,7 @@
"runs-as": "mesh-build-go",
"capabilities": [
"package-manager",
"seat-settle-delay",
"user-namespaces"
],
"state": [
@@ -20,7 +21,7 @@
{
"name": "node-build-go",
"scope": "node",
"serves": ["current", "pause", "resume"]
"serves": ["current", "drop", "pause", "resume"]
}
],
"consumes": [
@@ -62,7 +62,7 @@ func (m *Module) Puller(steps []StepTool, source buildstep.AskSource) *buildstep
if m.ReadPauses.Get != nil {
load = func(key string) ([]byte, error) { return m.ReadPauses.Get(key) }
}
return &buildstep.Puller{Source: source, Language: sdkstep.AskImage, Node: m.Node, Put: m.Steps, PausePut: m.Pauses, Load: load, ForgetPause: m.ForgetPause, PauseKeys: m.ReadPauses.Keys,
return &buildstep.Puller{Source: source, Language: sdkstep.AskImage, Node: m.Node, Put: m.Steps, PausePut: m.Pauses, Load: load, ForgetPause: m.ForgetPause, WatchPauses: m.WatchPauses, PauseKeys: m.ReadPauses.Keys,
Capacity: capacity, SettingsFile: settings, WatchConfig: watch, CoresPerStep: cores, BuildHours: hours, Eligible: m.Eligible, Logf: m.Say,
Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) {
if err := m.Check(ctx); err != nil {
@@ -38,6 +38,7 @@ func main() {
Pauses: stdio.State(sdkstep.PausesState).Put,
ReadPauses: readOf(stdio.State(sdkstep.PausesState)),
ForgetPause: stdio.State(sdkstep.PausesState).Delete,
WatchPauses: stdio.State(sdkstep.PausesState).Watch,
Ctx: ctx,
Say: logf,
Work: WorkDir,
@@ -72,7 +72,7 @@ func TestManifestDeclaresTheAccountTheCapabilityAndThePackages(t *testing.T) {
if home["type"] != "directory" || home["mode"] != "0700" || home["owner"] != "mesh-build-image" || home["place"] != "." {
t.Errorf("home = %v, want its directory at mode 0700 owned by the account", home)
}
for _, c := range []string{"user-namespaces", "package-manager"} {
for _, c := range []string{"user-namespaces", "package-manager", "seat-settle-delay"} {
if !slices.Contains(m.Capabilities, c) {
t.Errorf("capabilities = %v, want %s", m.Capabilities, c)
}
@@ -125,7 +125,7 @@ func TestManifestKeepsItsStatesClaimsItsBuildSeatAndDeclaresItsHealthTool(t *tes
if want := []string{"keys", MirrorWaitsState, "pauses", "steps"}; !slices.Equal(states, want) {
t.Errorf("state = %v, want %v", states, want)
}
if len(m.Claims) != 1 || m.Claims[0].Name != Seat || !slices.Equal(m.Claims[0].Serves, []string{"current", "pause", "resume"}) {
if len(m.Claims) != 1 || m.Claims[0].Name != Seat || !slices.Equal(m.Claims[0].Serves, []string{"current", "drop", "pause", "resume"}) {
t.Errorf("claims = %v, want the seat %s alone", m.Claims, Seat)
}
raw, _ := os.ReadFile(filepath.Join(moduleDir, "module.json"))
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
stdio "git.novox.be/novox/mesh-sdk/go"
"sync"
"time"
@@ -32,6 +33,8 @@ type Module struct {
ReadPauses StateRead
// ForgetPause deletes one key of the pauses state: a dropped ask's, once it is refused (the seat's verb drop).
ForgetPause func(key string) error
// WatchPauses follows the drops other holders write in the shared pauses state (stdio.State(...).Watch).
WatchPauses func(key string, handle func(stdio.StateChange) error) error
// Ctx is the process's context, which a step runs under (the tool call's is cut at 30 s, the step's is not).
Ctx context.Context
Say func(format string, a ...any)
+2 -1
View File
@@ -9,6 +9,7 @@
"runs-as": "mesh-build-image",
"capabilities": [
"package-manager",
"seat-settle-delay",
"user-namespaces"
],
"state": [
@@ -20,7 +21,7 @@
{
"name": "node-build-image",
"scope": "node",
"serves": ["current", "pause", "resume"]
"serves": ["current", "drop", "pause", "resume"]
}
],
"config": {
@@ -72,7 +72,7 @@ func (m *Module) Puller(steps []StepTool, source buildstep.AskSource) *buildstep
if m.ReadPauses.Get != nil {
load = func(key string) ([]byte, error) { return m.ReadPauses.Get(key) }
}
return &buildstep.Puller{Source: source, Language: sdkstep.AskNode, Node: m.Node, Put: m.Steps, PausePut: m.Pauses, Load: load, ForgetPause: m.ForgetPause, PauseKeys: m.ReadPauses.Keys,
return &buildstep.Puller{Source: source, Language: sdkstep.AskNode, Node: m.Node, Put: m.Steps, PausePut: m.Pauses, Load: load, ForgetPause: m.ForgetPause, WatchPauses: m.WatchPauses, PauseKeys: m.ReadPauses.Keys,
Capacity: capacity, SettingsFile: settings, WatchConfig: watch, CoresPerStep: cores, BuildHours: hours, Logf: m.Say,
Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) {
if err := m.Prepare(ctx); err != nil {
@@ -65,6 +65,11 @@ func TestManifestDeclaresTheAccountTheCapabilityAndThePackages(t *testing.T) {
if home["type"] != "directory" || home["mode"] != "0700" || home["owner"] != "mesh-build-node" || home["place"] != "." {
t.Errorf("home = %v, want its directory at mode 0700 owned by the account", home)
}
// ADR 0664 §1 (issue 1748): the delay a step given again waits is the tool runner's capability; the want waits on a node whose
// runner does not state it.
if !slices.Contains(m.Capabilities, "seat-settle-delay") {
t.Errorf("capabilities = %v, want seat-settle-delay", m.Capabilities)
}
if !slices.Contains(m.Capabilities, "user-namespaces") {
t.Errorf("capabilities = %v, want user-namespaces", m.Capabilities)
}
@@ -112,7 +117,7 @@ func TestManifestKeepsItsStatesClaimsItsBuildSeatAndDeclaresItsHealthTool(t *tes
if want := []string{"keys", "pauses", "steps", "toolchains"}; !slices.Equal(states, want) {
t.Errorf("state = %v, want %v", states, want)
}
if len(m.Claims) != 1 || m.Claims[0].Name != Seat || !slices.Equal(m.Claims[0].Serves, []string{"current", "pause", "resume"}) {
if len(m.Claims) != 1 || m.Claims[0].Name != Seat || !slices.Equal(m.Claims[0].Serves, []string{"current", "drop", "pause", "resume"}) {
t.Errorf("claims = %v, want the seat %s alone", m.Claims, Seat)
}
if !slices.Contains(m.Tools, HealthTool) {
@@ -3,6 +3,7 @@ package main
import (
"context"
"fmt"
stdio "git.novox.be/novox/mesh-sdk/go"
"net/http"
"os"
"sync"
@@ -51,6 +52,8 @@ type Module struct {
ReadPauses StateRead
// ForgetPause deletes one key of the pauses state: a dropped ask's, once it is refused (the seat's verb drop).
ForgetPause func(key string) error
// WatchPauses follows the drops other holders write in the shared pauses state (stdio.State(...).Watch).
WatchPauses func(key string, handle func(stdio.StateChange) error) error
// Emit tells the bus an event (stdio.Emit); nil emits nothing. The guard emits toolchain.reported after each write of
// the toolchains key, so the indexer, which holds no timer, learns that the bucket exists (issue 1444).
Emit func(eventType string, body any) error
+2 -1
View File
@@ -9,6 +9,7 @@
"runs-as": "mesh-build-node",
"capabilities": [
"package-manager",
"seat-settle-delay",
"user-namespaces"
],
"state": [
@@ -20,7 +21,7 @@
{
"name": "node-build-node",
"scope": "node",
"serves": ["current", "pause", "resume"]
"serves": ["current", "drop", "pause", "resume"]
}
],
"config": {