diff --git a/modules/mesh-build-go/buildstep/pull.go b/modules/mesh-build-go/buildstep/pull.go index 7f55b158c..b811bb897 100644 --- a/modules/mesh-build-go/buildstep/pull.go +++ b/modules/mesh-build-go/buildstep/pull.go @@ -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 diff --git a/modules/mesh-build-go/buildstep/pull_drop.go b/modules/mesh-build-go/buildstep/pull_drop.go index 0e6061a0b..b664802e6 100644 --- a/modules/mesh-build-go/buildstep/pull_drop.go +++ b/modules/mesh-build-go/buildstep/pull_drop.go @@ -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. (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 { diff --git a/modules/mesh-build-go/buildstep/pull_seat_test.go b/modules/mesh-build-go/buildstep/pull_seat_test.go index 4500d22a7..6a59c264b 100644 --- a/modules/mesh-build-go/buildstep/pull_seat_test.go +++ b/modules/mesh-build-go/buildstep/pull_seat_test.go @@ -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) } } diff --git a/modules/mesh-build-go/cmd/mesh-build-go/asks.go b/modules/mesh-build-go/cmd/mesh-build-go/asks.go index 2f2654ef2..415f217bf 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/asks.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/asks.go @@ -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 { diff --git a/modules/mesh-build-go/cmd/mesh-build-go/asks_test.go b/modules/mesh-build-go/cmd/mesh-build-go/asks_test.go index 7479b8015..c866703c9 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/asks_test.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/asks_test.go @@ -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) { diff --git a/modules/mesh-build-go/cmd/mesh-build-go/module.go b/modules/mesh-build-go/cmd/mesh-build-go/module.go index 9b2eae2aa..5fc8f74a2 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/module.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/module.go @@ -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 diff --git a/modules/mesh-build-go/module.json b/modules/mesh-build-go/module.json index d630360ce..b337cd644 100644 --- a/modules/mesh-build-go/module.json +++ b/modules/mesh-build-go/module.json @@ -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": [ diff --git a/modules/mesh-build-image/cmd/mesh-build-image/asks.go b/modules/mesh-build-image/cmd/mesh-build-image/asks.go index 37c0e7fa2..015c25ab1 100644 --- a/modules/mesh-build-image/cmd/mesh-build-image/asks.go +++ b/modules/mesh-build-image/cmd/mesh-build-image/asks.go @@ -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 { diff --git a/modules/mesh-build-image/cmd/mesh-build-image/main.go b/modules/mesh-build-image/cmd/mesh-build-image/main.go index efe965d29..82be8871b 100644 --- a/modules/mesh-build-image/cmd/mesh-build-image/main.go +++ b/modules/mesh-build-image/cmd/mesh-build-image/main.go @@ -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, diff --git a/modules/mesh-build-image/cmd/mesh-build-image/manifest_test.go b/modules/mesh-build-image/cmd/mesh-build-image/manifest_test.go index 548339b6e..f3c2cd6e2 100644 --- a/modules/mesh-build-image/cmd/mesh-build-image/manifest_test.go +++ b/modules/mesh-build-image/cmd/mesh-build-image/manifest_test.go @@ -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")) diff --git a/modules/mesh-build-image/cmd/mesh-build-image/module.go b/modules/mesh-build-image/cmd/mesh-build-image/module.go index 5f77f5578..73558e4e1 100644 --- a/modules/mesh-build-image/cmd/mesh-build-image/module.go +++ b/modules/mesh-build-image/cmd/mesh-build-image/module.go @@ -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) diff --git a/modules/mesh-build-image/module.json b/modules/mesh-build-image/module.json index fa94ba713..ac256f530 100644 --- a/modules/mesh-build-image/module.json +++ b/modules/mesh-build-image/module.json @@ -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": { diff --git a/modules/mesh-build-node/cmd/mesh-build-node/asks.go b/modules/mesh-build-node/cmd/mesh-build-node/asks.go index 4c02dbcc5..93147db9a 100644 --- a/modules/mesh-build-node/cmd/mesh-build-node/asks.go +++ b/modules/mesh-build-node/cmd/mesh-build-node/asks.go @@ -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 { diff --git a/modules/mesh-build-node/cmd/mesh-build-node/manifest_test.go b/modules/mesh-build-node/cmd/mesh-build-node/manifest_test.go index 71d61a1b1..06b8939d9 100644 --- a/modules/mesh-build-node/cmd/mesh-build-node/manifest_test.go +++ b/modules/mesh-build-node/cmd/mesh-build-node/manifest_test.go @@ -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) { diff --git a/modules/mesh-build-node/cmd/mesh-build-node/module.go b/modules/mesh-build-node/cmd/mesh-build-node/module.go index c3828902c..a98ca9209 100644 --- a/modules/mesh-build-node/cmd/mesh-build-node/module.go +++ b/modules/mesh-build-node/cmd/mesh-build-node/module.go @@ -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 diff --git a/modules/mesh-build-node/module.json b/modules/mesh-build-node/module.json index 9d48c9104..0e84060bf 100644 --- a/modules/mesh-build-node/module.json +++ b/modules/mesh-build-node/module.json @@ -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": {