From 9ce9053d7bc78b5f82d669844c02ee12f5344852 Mon Sep 17 00:00:00 2001 From: wangbo Date: Mon, 3 Aug 2026 13:59:36 +0800 Subject: [PATCH] =?UTF-8?q?fix(worker):=20=E8=AE=B0=E4=BD=8F=E8=BD=AE?= =?UTF-8?q?=E8=AF=A2=E9=97=B4=E9=9A=94=E5=86=85=E7=9A=84=E5=AE=B9=E9=87=8F?= =?UTF-8?q?=E9=9C=80=E6=B1=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 异步视频任务在 River 轮询空档会释放本地 lease，采样瞬间可能观察不到已经触顶的领取需求，导致自适应容量长期停在初始值。本提交记录采样窗口内出现过的容量饱和，同时保持空闲 Worker 不盲目升容。\n\n移植时保留主线现有的单节点容量阶梯、请求下限和严格失败门禁，仅合入失败路径先记录并排空任务与回调、再删除模拟器资源的清理顺序，并补充脚本守卫。\n\n验证：go test ./... -count=1；go vet ./...；bash -n；ShellCheck；production-acceptance-script-test.sh。 --- apps/api/internal/workerload/controller.go | 32 +++++++++++----- .../internal/workerload/controller_test.go | 37 +++++++++++++++++++ scripts/cluster/run-production-acceptance.sh | 12 +++--- .../production-acceptance-script-test.sh | 19 ++++++++++ 4 files changed, 85 insertions(+), 15 deletions(-) diff --git a/apps/api/internal/workerload/controller.go b/apps/api/internal/workerload/controller.go index 98ffe3e..f880aa4 100644 --- a/apps/api/internal/workerload/controller.go +++ b/apps/api/internal/workerload/controller.go @@ -70,13 +70,14 @@ type Snapshot struct { type Controller struct { mu sync.Mutex - mode string - hardLimit int - activeLimit int - heavyLimit int - claimLimit int - healthySamples int - healthyCount int + mode string + hardLimit int + activeLimit int + heavyLimit int + claimLimit int + healthySamples int + healthyCount int + demandedSinceSample bool preparing int waiting int @@ -179,10 +180,20 @@ func (c *Controller) TryStart() (*Lease, bool) { c.mu.Lock() defer c.mu.Unlock() 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 } c.preparing++ + if c.activeLocked() >= activeLimit { + c.demandedSinceSample = true + } c.last = c.snapshotLocked(ResourceSample{SampledAt: c.last.SampledAt}) 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) { + activeLimit := min(c.activeLimit, c.claimLimit) + demanded := c.demandedSinceSample || (activeLimit > 0 && c.activeLocked() >= activeLimit) + c.demandedSinceSample = false switch state { case PressureCritical: 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.heavyLimit = min(c.heavyLimit, max(1, (c.activeLimit+3)/4)) 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 return } diff --git a/apps/api/internal/workerload/controller_test.go b/apps/api/internal/workerload/controller_test.go index 4d714d1..e9aa2f3 100644 --- a/apps/api/internal/workerload/controller_test.go +++ b/apps/api/internal/workerload/controller_test.go @@ -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) { controller := New(Config{Mode: ModeAdaptive, HardLimit: 16, InitialActive: 8, InitialHeavy: 2}) busy := controller.Observe(ResourceSample{MemoryCurrentBytes: 80, MemoryLimitBytes: 100}) diff --git a/scripts/cluster/run-production-acceptance.sh b/scripts/cluster/run-production-acceptance.sh index bd511ca..1628bdd 100755 --- a/scripts/cluster/run-production-acceptance.sh +++ b/scripts/cluster/run-production-acceptance.sh @@ -381,12 +381,6 @@ cleanup() { if [[ $remote_load_drivers_installed == true ]]; then cleanup_remote_load_drivers >/dev/null 2>&1 || true 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 set +e if [[ $single_node_acceptance == true ]]; then @@ -398,6 +392,12 @@ cleanup() { mark_run_failed "${failure_reason:-acceptance workflow exited unexpectedly}" set -e 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" exit "$status" } diff --git a/tests/release/production-acceptance-script-test.sh b/tests/release/production-acceptance-script-test.sh index 093b97f..253d8d8 100755 --- a/tests/release/production-acceptance-script-test.sh +++ b/tests/release/production-acceptance-script-test.sh @@ -247,6 +247,25 @@ for invalid_capacity_state in \ fi 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' 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') == \