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.
This commit is contained in:
+11
-4
@@ -65,10 +65,13 @@ type Puller struct {
|
|||||||
PullWait time.Duration
|
PullWait time.Duration
|
||||||
Heartbeat time.Duration
|
Heartbeat time.Duration
|
||||||
Logf func(format string, args ...any)
|
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
|
// 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
|
||||||
// node's pause key (delivery.PauseKey) before it pulls, and Pause and Resume write it with Put, so a pause outlives
|
// reads the node's pause key (delivery.PauseKey) before it pulls, and Pause and Resume write it with PausePut, so a
|
||||||
// a restart.
|
// pause outlives a restart.
|
||||||
Load func(key string) ([]byte, error)
|
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
|
// 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
|
// 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.
|
// 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 {
|
if err != nil {
|
||||||
return err
|
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)
|
return fmt.Errorf("buildstep: writing the pause of %s: %w", p.Node, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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
|
// 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.
|
// stays paused until resumed.
|
||||||
func TestPullPauseOutlivesARestart(t *testing.T) {
|
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 }
|
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,
|
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: steps.put, PausePut: st.put, Load: st.load,
|
||||||
Capacity: func() int { return 2 }, Run: run}
|
PullWait: 10 * time.Millisecond, Capacity: func() int { return 2 }, Run: run}
|
||||||
if err := p.Pause("the operator"); err != nil {
|
if err := p.Pause("the operator"); err != nil {
|
||||||
t.Fatal(err)
|
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" {
|
if v, err := delivery.DecodePause(raw); err != nil || !v.Paused || v.By != "the operator" {
|
||||||
t.Fatalf("the pause key holds %s (%v)", raw, err)
|
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.
|
// 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,
|
again := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: steps.put, PausePut: st.put, Load: st.load,
|
||||||
Capacity: func() int { return 2 }, Run: run}
|
PullWait: 10 * time.Millisecond, Capacity: func() int { return 2 }, Run: run}
|
||||||
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
|
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|||||||
@@ -535,9 +535,13 @@ func ParseAskStepKey(key string) (node, askID string, err error) {
|
|||||||
return node, askID, nil
|
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
|
// `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) {
|
func PauseKey(node string) (string, error) {
|
||||||
if err := checkName("node", node); err != nil {
|
if err := checkName("node", node); err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
|
|||||||
Reference in New Issue
Block a user