Pull as not paused when the pause cannot be read, and say so once, instead of ending the puller silently
mesh/delivery held for a person: merged without a passing check: only a person decides what follows
mesh/delivery-group group fix/steps-bucket-ages held: a member is held: novox/mesh-sdk@518dae5ec9c8 (merged without a passing check: only a person decides what…
mesh/check checking 518dae5ec9c8 on 7f89cb51de15 | novox/mesh-sdk#81 in group fix/steps-bucket-ages

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.
This commit is contained in:
2026-10-25 18:56:17 +01:00
parent e8f8ef63fa
commit 518dae5ec9
2 changed files with 26 additions and 1 deletions
+3 -1
View File
@@ -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
+23
View File
@@ -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