From e8f8ef63fa112acd6f213375c43a3b162ce8a8f5 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 18:51:04 +0100 Subject: [PATCH 1/2] Keep a build node's pause in a pauses state of its own, so the steps state can drop its old keys without ending a pause The steps state of the build modules filled to the bus's 64 MiB cap (27,370 keys of mesh-build-go, none ever removed) and refused a passed check's end. Its keys get an age in the catalogue; the pause key lived among them and would then have ended by itself. Puller.PausePut writes it to delivery.PausesState instead; Put stays the fallback for a module not yet moved. --- go/buildstep/pull.go | 15 +++++++++++---- go/buildstep/pull_test.go | 13 ++++++++----- go/delivery/stepasks.go | 8 ++++++-- 3 files changed, 25 insertions(+), 11 deletions(-) diff --git a/go/buildstep/pull.go b/go/buildstep/pull.go index 1bcc354..249e9f9 100644 --- a/go/buildstep/pull.go +++ b/go/buildstep/pull.go @@ -65,10 +65,13 @@ type Puller struct { PullWait time.Duration Heartbeat time.Duration Logf func(format string, args ...any) - // Load reads one key of the module's steps state (nil and no error when it is absent). When set, Pull reads the - // node's pause key (delivery.PauseKey) before it pulls, and Pause and Resume write it with Put, so a pause outlives - // a restart. + // Load reads one key of the state the node's pause is kept in (nil and no error when it is absent). When set, Pull + // reads the node's pause key (delivery.PauseKey) before it pulls, and Pause and Resume write it with PausePut, so a + // pause outlives a restart. Load func(key string) ([]byte, error) + // PausePut writes the node's pause key: the module's pauses state (delivery.PausesState), which has no age, so a + // pause never ends by itself while the steps state drops its old keys. Put when nil. + PausePut Put // 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. @@ -121,7 +124,11 @@ func (p *Puller) setPaused(paused bool, by string) error { if err != nil { return err } - if _, err := p.Put(key, json.RawMessage(raw)); err != nil { + put := p.PausePut + if put == nil { + put = p.Put + } + if _, err := put(key, json.RawMessage(raw)); err != nil { return fmt.Errorf("buildstep: writing the pause of %s: %w", p.Node, err) } } diff --git a/go/buildstep/pull_test.go b/go/buildstep/pull_test.go index 59a26fa..6cc49fc 100644 --- a/go/buildstep/pull_test.go +++ b/go/buildstep/pull_test.go @@ -252,10 +252,10 @@ func TestPullAtCapacityZeroTakesNothing(t *testing.T) { // 1887/A2b: a paused puller takes nothing, its pause is kept on the pause key, and a puller started again on that state // stays paused until resumed. func TestPullPauseOutlivesARestart(t *testing.T) { - src, st := NewFakeAsks(), &kv{} + src, st, steps := NewFakeAsks(), &kv{}, &kv{} run := func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil } - p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Load: st.load, PullWait: 10 * time.Millisecond, - Capacity: func() int { return 2 }, Run: run} + p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: steps.put, PausePut: st.put, Load: st.load, + PullWait: 10 * time.Millisecond, Capacity: func() int { return 2 }, Run: run} if err := p.Pause("the operator"); err != nil { t.Fatal(err) } @@ -264,9 +264,12 @@ func TestPullPauseOutlivesARestart(t *testing.T) { if v, err := delivery.DecodePause(raw); err != nil || !v.Paused || v.By != "the operator" { t.Fatalf("the pause key holds %s (%v)", raw, err) } + if raw, _ := steps.load(key); raw != nil { + t.Fatalf("the pause was written in the steps state, whose keys age out: %s", raw) + } // The restart: a new puller on the same state. - again := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Load: st.load, PullWait: 10 * time.Millisecond, - Capacity: func() int { return 2 }, Run: run} + again := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: steps.put, PausePut: st.put, Load: st.load, + PullWait: 10 * time.Millisecond, Capacity: func() int { return 2 }, Run: run} _ = src.PublishAsk(testAsk(delivery.AskDelivery)) ctx, cancel := context.WithCancel(context.Background()) defer cancel() diff --git a/go/delivery/stepasks.go b/go/delivery/stepasks.go index 134dae8..9008be8 100644 --- a/go/delivery/stepasks.go +++ b/go/delivery/stepasks.go @@ -535,9 +535,13 @@ func ParseAskStepKey(key string) (node, askID string, err error) { return node, askID, nil } -// PauseKey is the key, in a build module's StepsBucket, that keeps whether the module on node takes asks: +// PausesState is the state a build module keeps its nodes' pause keys in. It has no age, unlike StepsState, whose keys +// are per step and dropped after their age: a pause kept among the steps would end by itself. +const PausesState = "pauses" + +// PauseKey is the key, in a build module's PausesState, that keeps whether the module on node takes asks: // `paused.` (issue 1887's design limit 2: every holder pulls from one shared worker, so a node's pause is its holder -// not fetching, kept here so a restart keeps it). A pause key is never an ask's step key: its second part is a node. +// not fetching, kept so a restart keeps it). A pause key is never an ask's step key: its second part is a node. func PauseKey(node string) (string, error) { if err := checkName("node", node); err != nil { return "", err -- 2.54.0 From 518dae5ec9c88c96906961a3314c901361a394c5 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 18:56:17 +0100 Subject: [PATCH 2/2] Pull as not paused when the pause cannot be read, and say so once, instead of ending the puller silently A Load error ended Pull, and the modules run it in a goroutine whose error nobody reads: every build node would stop pulling when the new pauses state could not be read (or when Pull reached restore before Serve). A node kept from building is held by capacity 0, not by a pause. --- go/buildstep/pull.go | 4 +++- go/buildstep/pull_test.go | 23 +++++++++++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/go/buildstep/pull.go b/go/buildstep/pull.go index 249e9f9..cc95414 100644 --- a/go/buildstep/pull.go +++ b/go/buildstep/pull.go @@ -234,7 +234,9 @@ func (p *Puller) Pull(ctx context.Context) error { return errors.New("buildstep: a puller needs its source, its steps' put and its run") } if err := p.restore(); err != nil { - return err // pulling on a pause that cannot be read could build on a node paused on purpose + // An unreadable pause degrades to not paused, said once: a puller that stops silently is the worse failure, and a + // node kept from building is held by its capacity (0), not by a pause. + p.logf("the pause of %s cannot be read: %v; pulling as not paused", p.Node, err) } go p.watchCapacity() // WatchConfigServed waits for Serve, which a module starts after its pull wait := p.PullWait diff --git a/go/buildstep/pull_test.go b/go/buildstep/pull_test.go index 6cc49fc..9269c45 100644 --- a/go/buildstep/pull_test.go +++ b/go/buildstep/pull_test.go @@ -4,8 +4,10 @@ import ( "context" "encoding/json" "errors" + "fmt" "strings" "sync" + "sync/atomic" "testing" "time" @@ -282,6 +284,27 @@ func TestPullPauseOutlivesARestart(t *testing.T) { t.Fatal(err) } waitFor(t, "the ask taken after the resume", func() bool { a, _, _ := src.Fates(); return len(a) == 1 }) + + t.Run("a pause that cannot be read is said and pulled as not paused", func(t *testing.T) { + src := NewFakeAsks() + var said atomic.Value + p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: steps.put, PausePut: st.put, + Load: func(string) ([]byte, error) { return nil, errors.New("no runtime yet") }, + PullWait: 10 * time.Millisecond, Capacity: func() int { return 2 }, Run: run, + Logf: func(f string, a ...any) { + if m := fmt.Sprintf(f, a...); strings.Contains(m, "cannot be read") { + said.Store(m) + } + }} + _ = src.PublishAsk(testAsk(delivery.AskDelivery)) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go p.Pull(ctx) + waitFor(t, "the ask taken", func() bool { a, _, _ := src.Fates(); return len(a) == 1 }) + if m, _ := said.Load().(string); !strings.Contains(m, "pulling as not paused") { + t.Fatalf("the unreadable pause was not said: %q", m) + } + }) } // 1887/A2c: a long step tells the bus it is still being worked; a holder that stops mid-step neither ends nor -- 2.54.0