Merge pull request 'Keep a build node's pause in a state of its own, and pull as not paused when the pause cannot be read (steps bucket full, 2026-10-25)' (#81) from fix/steps-bucket-ages into main

Reviewed-on: #81

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)
 		}
 	}
@@ -227,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

go/buildstep/pull_test.go

@@ -4,8 +4,10 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -252,10 +254,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 +266,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()
@@ -279,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

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.<node>` (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