fix(worker): 记住轮询间隔内的容量需求

异步视频任务在 River 轮询空档会释放本地 lease,采样瞬间可能观察不到已经触顶的领取需求,导致自适应容量长期停在初始值。本提交记录采样窗口内出现过的容量饱和,同时保持空闲 Worker 不盲目升容。\n\n移植时保留主线现有的单节点容量阶梯、请求下限和严格失败门禁,仅合入失败路径先记录并排空任务与回调、再删除模拟器资源的清理顺序,并补充脚本守卫。\n\n验证:go test ./... -count=1;go vet ./...;bash -n;ShellCheck;production-acceptance-script-test.sh。
This commit is contained in:
2026-08-05 00:53:06 +08:00
parent ebdb96e7d7
commit 9ce9053d7b
4 changed files with 85 additions and 15 deletions
+23 -9
View File
@@ -70,13 +70,14 @@ type Snapshot struct {
type Controller struct { type Controller struct {
mu sync.Mutex mu sync.Mutex
mode string mode string
hardLimit int hardLimit int
activeLimit int activeLimit int
heavyLimit int heavyLimit int
claimLimit int claimLimit int
healthySamples int healthySamples int
healthyCount int healthyCount int
demandedSinceSample bool
preparing int preparing int
waiting int waiting int
@@ -179,10 +180,20 @@ func (c *Controller) TryStart() (*Lease, bool) {
c.mu.Lock() c.mu.Lock()
defer c.mu.Unlock() defer c.mu.Unlock()
activeLimit := min(c.activeLimit, c.claimLimit) activeLimit := min(c.activeLimit, c.claimLimit)
if activeLimit <= 0 || c.activeLocked() >= activeLimit || c.preparing+c.finalizing >= c.heavyLimit { if activeLimit <= 0 {
return nil, false
}
if c.activeLocked() >= activeLimit {
c.demandedSinceSample = true
return nil, false
}
if c.preparing+c.finalizing >= c.heavyLimit {
return nil, false return nil, false
} }
c.preparing++ c.preparing++
if c.activeLocked() >= activeLimit {
c.demandedSinceSample = true
}
c.last = c.snapshotLocked(ResourceSample{SampledAt: c.last.SampledAt}) c.last = c.snapshotLocked(ResourceSample{SampledAt: c.last.SampledAt})
return &Lease{controller: c, phase: PhasePreparing}, true return &Lease{controller: c, phase: PhasePreparing}, true
} }
@@ -276,6 +287,9 @@ func (l *Lease) Phase() Phase {
} }
func (c *Controller) adjustLocked(state PressureState, memory, cpu, database float64) { func (c *Controller) adjustLocked(state PressureState, memory, cpu, database float64) {
activeLimit := min(c.activeLimit, c.claimLimit)
demanded := c.demandedSinceSample || (activeLimit > 0 && c.activeLocked() >= activeLimit)
c.demandedSinceSample = false
switch state { switch state {
case PressureCritical: case PressureCritical:
c.healthyCount = 0 c.healthyCount = 0
@@ -287,7 +301,7 @@ func (c *Controller) adjustLocked(state PressureState, memory, cpu, database flo
c.activeLimit = max(1, c.activeLimit-step) c.activeLimit = max(1, c.activeLimit-step)
c.heavyLimit = min(c.heavyLimit, max(1, (c.activeLimit+3)/4)) c.heavyLimit = min(c.heavyLimit, max(1, (c.activeLimit+3)/4))
default: default:
if memory > .60 || cpu > .65 || database > .65 || c.activeLocked() < min(c.activeLimit, c.claimLimit) { if memory > .60 || cpu > .65 || database > .65 || !demanded {
c.healthyCount = 0 c.healthyCount = 0
return return
} }
@@ -29,6 +29,43 @@ func TestControllerStartsConservativelyAndGrowsUnderSustainedDemand(t *testing.T
} }
} }
func TestControllerRemembersDemandAcrossShortPollingRuns(t *testing.T) {
controller := New(Config{Mode: ModeAdaptive, HardLimit: 8, InitialActive: 4, InitialHeavy: 4, HealthySamples: 2})
for range 2 {
leases := make([]*Lease, 0, 4)
for range 4 {
lease, ok := controller.TryStart()
if !ok {
t.Fatal("short polling task was not admitted")
}
leases = append(leases, lease)
}
for _, lease := range leases {
lease.Release()
}
controller.Observe(ResourceSample{
MemoryCurrentBytes: 30,
MemoryLimitBytes: 100,
CPUUtilization: .3,
DBConnections: 3,
DBMaxConnections: 20,
})
}
if snapshot := controller.Snapshot(); snapshot.ActiveLimit != 5 || snapshot.SafeCapacity != 5 {
t.Fatalf("grown snapshot=%+v, want active=5 safe=5", snapshot)
}
}
func TestControllerDoesNotGrowWhileIdleWithoutDemand(t *testing.T) {
controller := New(Config{Mode: ModeAdaptive, HardLimit: 8, InitialActive: 4, InitialHeavy: 1, HealthySamples: 2})
for range 4 {
controller.Observe(ResourceSample{MemoryCurrentBytes: 20, MemoryLimitBytes: 100, CPUUtilization: .2})
}
if snapshot := controller.Snapshot(); snapshot.ActiveLimit != 4 {
t.Fatalf("idle snapshot=%+v, want active=4", snapshot)
}
}
func TestControllerBusyAndCriticalPressureReduceNewClaims(t *testing.T) { func TestControllerBusyAndCriticalPressureReduceNewClaims(t *testing.T) {
controller := New(Config{Mode: ModeAdaptive, HardLimit: 16, InitialActive: 8, InitialHeavy: 2}) controller := New(Config{Mode: ModeAdaptive, HardLimit: 16, InitialActive: 8, InitialHeavy: 2})
busy := controller.Observe(ResourceSample{MemoryCurrentBytes: 80, MemoryLimitBytes: 100}) busy := controller.Observe(ResourceSample{MemoryCurrentBytes: 80, MemoryLimitBytes: 100})
+6 -6
View File
@@ -381,12 +381,6 @@ cleanup() {
if [[ $remote_load_drivers_installed == true ]]; then if [[ $remote_load_drivers_installed == true ]]; then
cleanup_remote_load_drivers >/dev/null 2>&1 || true cleanup_remote_load_drivers >/dev/null 2>&1 || true
fi fi
if [[ $single_node_emulator_installed == true ]]; then
cleanup_single_node_emulator >/dev/null 2>&1 || true
fi
if [[ $single_node_acceptance == true ]]; then
cleanup_single_node_kubernetes_resources >/dev/null 2>&1 || true
fi
if (( status != 0 )) && [[ -n $run_id && $failure_recorded != true ]]; then if (( status != 0 )) && [[ -n $run_id && $failure_recorded != true ]]; then
set +e set +e
if [[ $single_node_acceptance == true ]]; then if [[ $single_node_acceptance == true ]]; then
@@ -398,6 +392,12 @@ cleanup() {
mark_run_failed "${failure_reason:-acceptance workflow exited unexpectedly}" mark_run_failed "${failure_reason:-acceptance workflow exited unexpectedly}"
set -e set -e
fi fi
if [[ $single_node_emulator_installed == true ]]; then
cleanup_single_node_emulator >/dev/null 2>&1 || true
fi
if [[ $single_node_acceptance == true ]]; then
cleanup_single_node_kubernetes_resources >/dev/null 2>&1 || true
fi
rm -rf -- "$temporary_root" rm -rf -- "$temporary_root"
exit "$status" exit "$status"
} }
@@ -247,6 +247,25 @@ for invalid_capacity_state in \
fi fi
done done
# shellcheck disable=SC2016 # Match the literal arithmetic expression in the target script.
grep -Fq 'requests=$((slots * 6))' "$script"
if grep -Fq '(( requests < 64 ))' "$script"; then
echo 'single-node video ladder still overdrives low-capacity steps with a fixed request floor' >&2
exit 1
fi
awk '
/^cleanup\(\)/ { capture = 1 }
capture { print }
capture && /^}/ { exit }
' "$script" >"$tmp/cleanup-function.sh"
mark_failed_line=$(grep -n 'mark_run_failed' "$tmp/cleanup-function.sh" | cut -d: -f1)
cleanup_resources_line=$(grep -n 'cleanup_single_node_kubernetes_resources' "$tmp/cleanup-function.sh" | cut -d: -f1)
[[ -n $mark_failed_line && -n $cleanup_resources_line && $mark_failed_line -lt $cleanup_resources_line ]] || {
echo 'single-node cleanup removes the callback collector before failed outbox work drains' >&2
exit 1
}
passing_pressure_row='2026-08-01T00:00:00Z,0,0,899,149,200,84,1535,0,32,32,0,0,0,0,2,48,0,8,8,0,0,0,16,16,0,0,1535,100' passing_pressure_row='2026-08-01T00:00:00Z,0,0,899,149,200,84,1535,0,32,32,0,0,0,0,2,48,0,8,8,0,0,0,16,16,0,0,1535,100'
pressure_row_passes_live_gates "$passing_pressure_row" pressure_row_passes_live_gates "$passing_pressure_row"
[[ $(pressure_row_live_gate_reason '2026-08-01T00:00:00Z,0,0,900,149,200,79,1535,0,32,32,0,0,0,0,2,48,0,8,8,0,0,0,16,16,0,0,1535,100') == \ [[ $(pressure_row_live_gate_reason '2026-08-01T00:00:00Z,0,0,900,149,200,79,1535,0,32,32,0,0,0,0,2,48,0,8,8,0,0,0,16,16,0,0,1535,100') == \