From 39d52f20d1541b45c4599fef5651461407ec9b87 Mon Sep 17 00:00:00 2001 From: Vulcan Date: Sun, 27 Sep 2026 05:55:55 +0800 Subject: [PATCH] =?UTF-8?q?fix(worker):=20=E8=BF=9B=E5=BA=A6=E5=BA=8F?= =?UTF-8?q?=E5=88=97=E5=8D=95=E6=AC=A1=E8=B7=B3=E5=8F=98=EF=BC=8871.1%?= =?UTF-8?q?=E2=86=920.5%=EF=BC=89=E8=A7=A3=E6=9E=90=E5=99=A8=E4=B8=A2?= =?UTF-8?q?=E5=BC=83=E5=9B=9E=E9=80=80=E5=B8=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ffmpeg 中途死亡（OOM/SIGABRT，归类 process_crash）会触发 Worker 多阶段重试； 重试从零重跑时 ffmpeg 的 time= 回到 0，进度解析器原样转发，已推送的作业进度 因此回退（复现实测 9.3% → 3.2%，与观察到的 71.1% → 0.5% 同因）。 ProgressParser 记录已发出的最高百分比，低于该值的帧不再推送，但速度 EWMA 与 wall-clock 状态照常更新；每次重跑由 retry executor 经 router 显式上报 attempt 边界（不靠 time= 回退猜测），据此重基 ETA/速度基线且不动水位。裁剪任务的输出 窗口不再被输入头的 Duration: 行改写。作业迁移到另一 Worker 时，Server 按 jobID 单调钳制进度（低于已存值既不存储也不广播），README 同步说明。 --- README.md | 2 + pkg/server/db/db.go | 24 +- pkg/server/db/integrity_test.go | 2 +- pkg/server/handlers/handlers.go | 18 +- .../handlers/progress_monotonic_test.go | 184 ++++++++++++++ pkg/worker/lifecycle2_test.go | 4 +- pkg/worker/progress_parser.go | 79 +++++- pkg/worker/progress_parser_test.go | 226 +++++++++++++++++- pkg/worker/progress_router.go | 9 + pkg/worker/retry_executor.go | 15 +- pkg/worker/retry_executor_test.go | 6 +- pkg/worker/worker.go | 2 +- 12 files changed, 541 insertions(+), 30 deletions(-) create mode 100644 pkg/server/handlers/progress_monotonic_test.go diff --git a/README.md b/README.md index 29f9df4..dd7d7e8 100644 --- a/README.md +++ b/README.md @@ -128,6 +128,8 @@ RFFMPEG_WORKER_NAME=worker-1 RFFMPEG_MAX_CONCURRENT=2 ./bin/rffmpeg-worker **静默模式（`-q` / `--quiet`）**：`--quiet` 抑制进度类输出——banner、上传/下载进度（`Uploading`/`Uploaded`/`Downloading`/`Output saved` 等）、`Job status:` 状态轮询、`Progress: …% | ETA: …` 转码进度行，以及 Worker 实时 stderr 中 ffmpeg 自身的输出（输入头/编码器信息、错误与警告行）。`[rffmpeg]` 前缀的通知行（如 `[rffmpeg] Cache hit: `、编码器改写/回退）不属于进度输出，`--quiet` 下仍写入 stderr；任务终态失败报告（`Job failed:` / `Job timed out:` 等）始终打印、不受 `--quiet` 影响。需要查看 ffmpeg 原始错误/警告时，去掉 `--quiet` 重新运行即可。 +**进度行单调不回退**：`Progress: …% | ETA: …` 只增不减，且作业级同样成立。Worker 在 ffmpeg 中途崩溃（OOM、SIGABRT 等）后走多阶段重试、从零重跑时，重试的早期帧被丢弃而不是把已显示的百分比拉回去——stderr 里先出现的 `[RETRY] Initial attempt failed (…)` 行就是这个重跑的信号，进度停在崩溃点直到重试追平。作业被迁移到另一 Worker 重新执行时，Server 按 jobID 拒绝回退的进度上报（既不存储也不广播），因此新 Worker 从 0 重跑的低百分比不会让已显示的进度回退：进度停在迁移前的高点，待新 Worker 追平后继续推进（此时 stderr 中会再次出现 ffmpeg 头与改写通知行）。 + **硬件编码升级（`--auto-hw`）与参数去向**：`--auto-hw` 让 Worker 把软件编码升级为本地硬件编码（如 `libx264 → h264_qsv`），改写结果通过 `[rffmpeg]` 行输出：升级/回退行之后紧跟一行「参数去向」，逐个列出用户传入的编码参数在目标编码器上的形态——改名的显示映射关系（`-crf 23 → -global_quality 23`），值被转换的显示新值（`-preset slow → -preset p7`），名字与值都不变的原样列出（`-profile high`）： ```bash diff --git a/pkg/server/db/db.go b/pkg/server/db/db.go index f89e6e8..719538f 100644 --- a/pkg/server/db/db.go +++ b/pkg/server/db/db.go @@ -994,20 +994,30 @@ func (d *Database) updateJobTerminalStatusWithOwner(jobID, workerID string, stat return nil } -// UpdateJobProgress updates the progress and ETA seconds for a job. -func (d *Database) UpdateJobProgress(id string, progress float64, etaSeconds int) error { +// UpdateJobProgress stores the progress and ETA seconds for a job, reporting +// whether the report was accepted. A job's progress is monotonic: a report +// that would move the percent backward is rejected without writing, so a +// re-dispatched job — its new worker re-encodes from zero and reports low +// percentages again — cannot rewind the value the API and the CLI already +// published. The comparison runs inside the UPDATE, so concurrent reporters +// (a late request racing the new owner's) serialize on the row. +func (d *Database) UpdateJobProgress(id string, progress float64, etaSeconds int) (bool, error) { now := time.Now() - _, err := d.db.Exec(` + res, err := d.db.Exec(` UPDATE jobs SET progress_percent = ?, eta_seconds = ?, updated_at = ? - WHERE id = ? - `, progress, etaSeconds, now, id) + WHERE id = ? AND progress_percent <= ? + `, progress, etaSeconds, now, id, progress) if err != nil { - return fmt.Errorf("failed to update job progress: %w", err) + return false, fmt.Errorf("failed to update job progress: %w", err) + } + written, err := res.RowsAffected() + if err != nil { + return false, fmt.Errorf("failed to update job progress: %w", err) } - return nil + return written > 0, nil } // AssignJobToWorker atomically assigns a job to a worker: one statement sets diff --git a/pkg/server/db/integrity_test.go b/pkg/server/db/integrity_test.go index 9bedff0..812142a 100644 --- a/pkg/server/db/integrity_test.go +++ b/pkg/server/db/integrity_test.go @@ -332,7 +332,7 @@ func TestConcurrentWritesNoDatabaseLocked(t *testing.T) { } continue } - if err := database.UpdateJobProgress(job.ID, float64(i), i); isLockedErr(err) { + if _, err := database.UpdateJobProgress(job.ID, float64(i), i); isLockedErr(err) { errCh <- err return } diff --git a/pkg/server/handlers/handlers.go b/pkg/server/handlers/handlers.go index 945e239..3aa552c 100644 --- a/pkg/server/handlers/handlers.go +++ b/pkg/server/handlers/handlers.go @@ -927,13 +927,21 @@ func (h *Handler) UpdateJob(w http.ResponseWriter, r *http.Request) { // Handle progress update from worker if req.Progress > 0 || req.EtaSeconds > 0 { - // Store progress in DB - if err := h.db.UpdateJobProgress(jobID, req.Progress, req.EtaSeconds); err != nil { + // Store progress in DB. A report that would move the job's percent + // backward is rejected (see Database.UpdateJobProgress): the job's + // progress is monotonic across re-dispatches, and neither the stored + // value nor the broadcast may rewind what the CLI already showed. + accepted, err := h.db.UpdateJobProgress(jobID, req.Progress, req.EtaSeconds) + if err != nil { log.Printf("Failed to update job progress for job %s: %v", jobID, err) } - // Broadcast progress via WebSocket - if err := h.wsHub.BroadcastProgress(jobID, req.Progress, req.TimeUs, req.DurationUs, req.Speed, req.EtaSeconds); err != nil { - log.Printf("Failed to broadcast progress for job %s: %v", jobID, err) + // Broadcast when the report was accepted, or when the store failed so + // a degraded DB does not also silence the live stream. A rejected + // (regressing) report is skipped on both sinks. + if accepted || err != nil { + if err := h.wsHub.BroadcastProgress(jobID, req.Progress, req.TimeUs, req.DurationUs, req.Speed, req.EtaSeconds); err != nil { + log.Printf("Failed to broadcast progress for job %s: %v", jobID, err) + } } } diff --git a/pkg/server/handlers/progress_monotonic_test.go b/pkg/server/handlers/progress_monotonic_test.go new file mode 100644 index 0000000..e93aba8 --- /dev/null +++ b/pkg/server/handlers/progress_monotonic_test.go @@ -0,0 +1,184 @@ +package handlers_test + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/go-chi/chi/v5" + gorillaws "github.com/gorilla/websocket" + + "github.com/tsic404/rffmpeg/pkg/protocol" + "github.com/tsic404/rffmpeg/pkg/server/websocket" +) + +// reportJobProgress PATCHes a progress/ETA report for jobID, as a worker does. +func reportJobProgress(t *testing.T, router *chi.Mux, jobID string, progress float64, eta int) { + t.Helper() + body, err := json.Marshal(protocol.JobUpdateRequest{Progress: progress, EtaSeconds: eta}) + if err != nil { + t.Fatalf("marshal update: %v", err) + } + req := httptest.NewRequest(http.MethodPatch, "/api/v1/jobs/"+jobID, bytes.NewReader(body)) + req.Header.Set("Authorization", "Bearer test-token") + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("PATCH progress %.1f = %d, want 200; body=%s", progress, rec.Code, rec.Body.String()) + } +} + +// getJobProgress reads back the stored progress/ETA of jobID via the public API. +func getJobProgress(t *testing.T, router *chi.Mux, jobID string) (float64, int) { + t.Helper() + req := httptest.NewRequest(http.MethodGet, "/api/v1/jobs/"+jobID, nil) + req.Header.Set("Authorization", "Bearer test-token") + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("GET job = %d, want 200", rec.Code) + } + var resp struct { + Job struct { + ProgressPercent float64 `json:"progress_percent"` + EtaSeconds int `json:"eta_seconds"` + } `json:"job"` + } + if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil { + t.Fatalf("decode job: %v", err) + } + return resp.Job.ProgressPercent, resp.Job.EtaSeconds +} + +// TestJobProgressNeverRegresses locks the job-level progress contract on the +// server. A job migrated after a worker failure re-encodes from zero on its new +// worker and reports low percentages again; those reports must not rewind the +// value the API and the CLI already published — the reported 71.1% → 0.5% +// sequence. A regressing report lands on neither sink (stored value and +// broadcast), while equal and higher reports still land so the ETA keeps +// refreshing at a held percent. +func TestJobProgressNeverRegresses(t *testing.T) { + h, router, cleanup := setupTest(t) + defer cleanup() + + job, err := h.GetDB().CreateJob(`["in.mp4"]`, `["-c:v","libx264"]`, "out.mp4", false) + if err != nil { + t.Fatalf("CreateJob: %v", err) + } + + reportJobProgress(t, router, job.ID, 71.1, 7) + if got, _ := getJobProgress(t, router, job.ID); got != 71.1 { + t.Fatalf("stored progress = %v, want 71.1", got) + } + + // The migrated worker's attempt restarted from zero: rejected outright, so + // the ETA that travelled with the regressing report is not stored either. + reportJobProgress(t, router, job.ID, 0.5, 900) + got, eta := getJobProgress(t, router, job.ID) + if got != 71.1 { + t.Errorf("progress after a regressing report = %v, want 71.1 (rewound)", got) + } + if eta != 7 { + t.Errorf("eta after a regressing report = %d, want 7 (the report must not land)", eta) + } + + // Equal and higher reports still land: the ETA refreshes while the percent + // is held at the mark, and the restarted attempt advances past it. + reportJobProgress(t, router, job.ID, 71.1, 3) + if _, eta := getJobProgress(t, router, job.ID); eta != 3 { + t.Errorf("eta at an equal percent = %d, want 3 (equal reports must still land)", eta) + } + reportJobProgress(t, router, job.ID, 80, 2) + if got, _ := getJobProgress(t, router, job.ID); got != 80 { + t.Errorf("progress after a higher report = %v, want 80", got) + } +} + +// TestJobProgressBroadcastGate covers the stream half of that contract: a +// rejected report must not reach CLI subscribers, while an equal report still +// must, since the CLI renders the stream rather than the stored row. Without +// this the gate could regress to an unconditional broadcast and the +// stored-value assertions above would stay green while every CLI showed the +// rewind again. +func TestJobProgressBroadcastGate(t *testing.T) { + h, router, cleanup := setupTest(t) + defer cleanup() + + job, err := h.GetDB().CreateJob(`["in.mp4"]`, `["-c:v","libx264"]`, "out.mp4", false) + if err != nil { + t.Fatalf("CreateJob: %v", err) + } + + // Subscribe a real WebSocket client to the job, the way the CLI does. + hub := h.GetWSHub() + h.StartWSHub() // the hub dispatches registrations from its Run loop + wsRouter := chi.NewRouter() + wsRouter.Get("/ws/jobs/{jobId}", func(w http.ResponseWriter, r *http.Request) { + websocket.HandleJobLogWithHub(hub, w, r) + }) + wsSrv := httptest.NewServer(wsRouter) + defer wsSrv.Close() + + conn, _, err := gorillaws.DefaultDialer.Dial( + "ws"+strings.TrimPrefix(wsSrv.URL, "http")+"/ws/jobs/"+job.ID, nil) + if err != nil { + t.Fatalf("dial job log websocket: %v", err) + } + defer conn.Close() + + // The hub registers asynchronously; a broadcast before that would be + // dispatched to no one and read as "not broadcast". + registered := time.Now().Add(2 * time.Second) + for hub.TotalClients() < 1 { + if time.Now().After(registered) { + t.Fatal("websocket client was never registered with the hub") + } + time.Sleep(5 * time.Millisecond) + } + + // readProgress returns the next progress frame broadcast within window. + // A timeout fails the test: every call here expects a frame (the negative + // assertion below is expressed as wire order, not as expected silence, since + // a read timeout corrupts a gorilla connection). + readProgress := func(window time.Duration) (percent float64, eta int) { + until := time.Now().Add(window) + for { + if err := conn.SetReadDeadline(until); err != nil { + t.Fatalf("set read deadline: %v", err) + } + _, data, err := conn.ReadMessage() + if err != nil { + t.Fatalf("no progress frame broadcast within %v: %v", window, err) + } + var msg protocol.WSMessage + if err := json.Unmarshal(data, &msg); err != nil || msg.Type != protocol.WSMsgProgress { + continue // heartbeats and other frames are not progress + } + payload, _ := msg.Data.(map[string]interface{}) + percent, _ = payload["percent"].(float64) + if secs, ok := payload["eta_seconds"].(float64); ok { + eta = int(secs) + } + return percent, eta + } + } + + reportJobProgress(t, router, job.ID, 71.1, 7) + if percent, eta := readProgress(2 * time.Second); percent != 71.1 || eta != 7 { + t.Fatalf("accepted report broadcast = (%.1f, eta=%d), want (71.1, 7) — a gate that rejects everything would hide the rewind too", percent, eta) + } + + // A rejected report and the next accepted one, back to back. The hub + // broadcasts in report order, so a regressing frame would surface here as + // 0.5 instead of the equal-percent frame that must follow it. + reportJobProgress(t, router, job.ID, 0.5, 900) + reportJobProgress(t, router, job.ID, 71.1, 3) + if percent, eta := readProgress(2 * time.Second); percent != 71.1 || eta != 3 { + t.Errorf("broadcast after a rejected report = (%.1f, eta=%d), want (71.1, 3): a regressing report must not reach subscribers, an equal one still must", percent, eta) + } +} diff --git a/pkg/worker/lifecycle2_test.go b/pkg/worker/lifecycle2_test.go index 4ffae20..b95dc5e 100644 --- a/pkg/worker/lifecycle2_test.go +++ b/pkg/worker/lifecycle2_test.go @@ -44,7 +44,7 @@ func TestExecuteWithRetry_StaleOutputFromFailedAttemptDoesNotFoolNextAttempt(t * // ffmpeg exits 0 but writes nothing to outputPath (output goes to null). args := []string{"-f", "lavfi", "-i", "testsrc=duration=0.1", "-f", "null", "-"} - result := re.ExecuteWithRetry(context.Background(), args, outputPath, false, nil, nil) + result := re.ExecuteWithRetry(context.Background(), args, outputPath, false, nil, nil, nil) if result.Success { t.Fatal("attempt with no real output was judged successful — stale-output cleanup is broken") @@ -86,7 +86,7 @@ func TestExecuteWithRetry_CancelledContextSkipsSoftwareFallback(t *testing.T) { cancel() start := time.Now() - result := re.ExecuteWithRetry(ctx, []string{"-f", "lavfi", "-i", "testsrc=duration=5", "out.mp4"}, "", false, nil, nil) + result := re.ExecuteWithRetry(ctx, []string{"-f", "lavfi", "-i", "testsrc=duration=5", "out.mp4"}, "", false, nil, nil, nil) if time.Since(start) > 5*time.Second { t.Fatal("ExecuteWithRetry ran a full transcode despite cancelled context") diff --git a/pkg/worker/progress_parser.go b/pkg/worker/progress_parser.go index 25bccb7..a4a7b5d 100644 --- a/pkg/worker/progress_parser.go +++ b/pkg/worker/progress_parser.go @@ -59,6 +59,20 @@ type ProgressParser struct { // etaMinWallSeconds of wall-clock time suppress ETA: not enough data has // accumulated for a stable wall-clock rate. speedSamples int + // maxPercent is the highest percent emitted for this job, or -1 when none + // has been. Frames that would report less are dropped by ParseLine: ffmpeg's + // media timeline restarts at zero when a second attempt re-runs it (the + // worker's multi-stage retry after a mid-encode crash — OOM, SIGABRT), and a + // job's user-visible progress must not rewind with it. The floor is measured + // in percent, so a changed denominator (setDurationUs) re-bases it, and it + // spans attempts — StartAttempt deliberately leaves it alone. + maxPercent float64 + // hasSeekWindow records that SetSeekWindow established the denominator for a + // trimmed job. That window stays authoritative: SetDuration is ignored + // afterwards so the input header's Duration: line (which the router feeds + // through SetDuration) cannot widen the denominator back to the full input + // and re-base every later percent onto the untrimmed scale. + hasSeekWindow bool } // etaSpeedAlpha is the EWMA smoothing factor for speed observations used in @@ -84,6 +98,7 @@ func NewProgressParser() *ProgressParser { return &ProgressParser{ minPushInterval: 1 * time.Second, // throttle to 1 push per second nowFunc: time.Now, + maxPercent: -1, } } @@ -96,7 +111,26 @@ func (p *ProgressParser) now() time.Time { } // SetDuration sets the known total duration from an external source (e.g., prior probe). +// It is ignored once SetSeekWindow has established the denominator: the output +// window, not the full input, is what percent and ETA are measured against for +// a trimmed job. func (p *ProgressParser) SetDuration(durationUs int64) { + if p.hasSeekWindow { + return + } + p.setDurationUs(durationUs) +} + +// setDurationUs sets the percent denominator. The monotonic floor is measured +// in percent, so it is cleared whenever the denominator actually changes: a +// re-based scale (the input header's Duration: line can overwrite the window +// the probe seeded, or exceed the probe's value) would otherwise leave the +// restarted scale reporting percentages below the old floor forever, stalling +// progress while the media time keeps advancing. +func (p *ProgressParser) setDurationUs(durationUs int64) { + if durationUs != p.durationUs { + p.maxPercent = -1 + } p.durationUs = durationUs } @@ -139,7 +173,8 @@ func (p *ProgressParser) SetSeekWindow(inputDurationUs, seekUs, tUs, toUs int64) // Always overwrite: a degenerate window (0) must clear the full-input // duration that SetDuration seeded, so percent falls back to -1 instead // of reporting a meaningless ratio against the full input. - p.durationUs = windowUs + p.hasSeekWindow = true + p.setDurationUs(windowUs) } // ParseDurationLine attempts to extract Duration from an ffmpeg header line. @@ -179,11 +214,11 @@ func (p *ProgressParser) ParseLine(line string) *ProgressFrame { // Use stored duration if available, otherwise try to parse from line durationUs := p.durationUs if durationUs == 0 { - durMatches := durationRegex.FindStringSubmatch(line) - if durMatches != nil { - durationUs = parseTimeToUs(durMatches[1], durMatches[2], durMatches[3], durMatches[4]) - p.durationUs = durationUs - } + // SetDuration itself refuses to re-base a seek window, so a trimmed job + // whose window is degenerate stays at "unknown" instead of silently + // reporting against the full input. + p.SetDuration(ParseDurationLine(line)) + durationUs = p.durationUs } frame.DurationUs = durationUs @@ -234,6 +269,20 @@ func (p *ProgressParser) ParseLine(line string) *ProgressFrame { } } + // A job's percent never goes backward, so a restarted attempt's frames are + // dropped instead of rewinding progress the CLI already displayed. The + // drop suppresses the push only: every frame still updated the attempt + // state above, so the displayed speed and the ETA stay live while the + // restarted attempt climbs back to the mark. Frames with an unknown + // duration (-1) carry no percent to compare against and pass through + // untouched. + if frame.Percent >= 0 { + if p.maxPercent >= 0 && frame.Percent < p.maxPercent { + return nil + } + p.maxPercent = frame.Percent + } + return frame } @@ -247,14 +296,30 @@ func (p *ProgressParser) ShouldPush() bool { return false } +// StartAttempt re-bases the state ParseLine derives from one ffmpeg process — +// the wall-clock baseline the ETA is measured against and the speed EWMA — for +// a new process re-running the same job (the worker's multi-stage retry). The +// producer reports the boundary instead of the parser inferring it from time= +// going backward: a legitimate timeline restart (segment sharding, sequential +// outputs) is not a re-run and must keep its baseline, while a real re-run must +// not leave the dead attempt's downtime in the ETA. The percent floor is not +// touched: it spans the whole job. +func (p *ProgressParser) StartAttempt() { + p.firstSampleTime = time.Time{} + p.ewmaSpeed = 0 + p.speedSamples = 0 +} + // Reset resets the parser state for a new job. func (p *ProgressParser) Reset() { p.startTime = p.now() p.lastPushTime = time.Time{} - p.durationUs = 0 + p.setDurationUs(0) p.firstSampleTime = time.Time{} p.ewmaSpeed = 0 p.speedSamples = 0 + p.maxPercent = -1 + p.hasSeekWindow = false } // parseTimeToUs converts hh, mm, ss, ms components to microseconds. diff --git a/pkg/worker/progress_parser_test.go b/pkg/worker/progress_parser_test.go index b0cbf42..74f3c9b 100644 --- a/pkg/worker/progress_parser_test.go +++ b/pkg/worker/progress_parser_test.go @@ -55,9 +55,6 @@ func TestParseDurationLine(t *testing.T) { } func TestProgressParser_ParseLine(t *testing.T) { - p := NewProgressParser() - p.SetDuration(100_000_000) // 100 seconds in microseconds - tests := []struct { line string wantNil bool @@ -72,7 +69,12 @@ func TestProgressParser_ParseLine(t *testing.T) { {line: "frame= 1000 fps= 60 size= 50000kB time=00:01:30.00 bitrate=4500.0kbits/s speed=3.00x", wantTimeUs: 90_000_000, wantDurationUs: 100_000_000, wantPercentLow: 89, wantPercentHigh: 91}, } + // Each case gets its own parser: ParseLine drops frames reporting less than + // the job's high-water mark, so cases must not share one parser's mark. for _, tt := range tests { + p := NewProgressParser() + p.SetDuration(100_000_000) // 100 seconds in microseconds + frame := p.ParseLine(tt.line) if tt.wantNil { if frame != nil { @@ -895,3 +897,221 @@ func TestProgressRouterHandlerFiltersStatsLines(t *testing.T) { } } } + +// TestProgressRouter_ProgressNeverRegresses pins the job-level progress +// contract: the percents pushed to the server never decrease. A retryable +// ffmpeg death mid-encode (OOM/SIGABRT, process_crash) makes the worker's +// multi-stage retry re-run ffmpeg from scratch on the same parser, so the +// media timeline restarts at ~0. The restarted attempt must not drag the +// progress already displayed back down — that rewind is the reported +// 71.1% → 0.5% sequence. +func TestProgressRouter_ProgressNeverRegresses(t *testing.T) { + var mu sync.Mutex + var percents []float64 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + var req protocol.JobUpdateRequest + if err := json.Unmarshal(body, &req); err == nil { + mu.Lock() + percents = append(percents, req.Progress) + mu.Unlock() + } + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + client := NewClient(srv.URL, "worker-1", "") + router := NewProgressRouter(client, "job-1", func(string) {}) + router.SetDuration(60_000_000) // 60s input, what probeInputDurationUs seeds + router.sendDelay = 0 // the 1s throttle is not what this test pins + + h := router.Handler() + // Attempt 1 encodes to 9.3% of the 60s input before ffmpeg dies. + h(buildProgressLine(2_000_000, 2.0) + "\n") // 3.3% + h(buildProgressLine(5_600_000, 2.0) + "\n") // 9.3% + // The retry re-runs ffmpeg from scratch: time= restarts at zero. + h(buildProgressLine(1_900_000, 2.0) + "\n") // 3.2% — must not be pushed + h(buildProgressLine(5_800_000, 2.0) + "\n") // 9.7% + h(buildProgressLine(12_000_000, 2.0) + "\n") // 20.0% + + mu.Lock() + defer mu.Unlock() + if len(percents) < 4 { + t.Fatalf("pushed %d progress updates, want >= 4: %v", len(percents), percents) + } + for i := 1; i < len(percents); i++ { + if percents[i] < percents[i-1] { + t.Fatalf("progress regressed at update %d: %v — a restarted attempt must not rewind the job's progress", i, percents) + } + } + last := percents[len(percents)-1] + if last < 19.99 || last > 20.01 { + t.Errorf("final pushed percent = %v, want 20 (the last frame's own value)", last) + } +} + +// TestProgressParser_ResetClearsProgressFloor: the monotonic floor is +// per-job state, so Reset must clear it — a parser reused for a new job must +// not stay pinned to the previous job's high-water mark. +func TestProgressParser_ResetClearsProgressFloor(t *testing.T) { + p := NewProgressParser() + p.SetDuration(60_000_000) + + mid := p.ParseLine(buildProgressLine(30_000_000, 2.0)) + if mid == nil || mid.Percent < 49.99 || mid.Percent > 50.01 { + t.Fatalf("first frame percent = %v, want ~50", mid) + } + if late := p.ParseLine(buildProgressLine(1_000_000, 2.0)); late != nil { + t.Fatalf("regressing frame accepted mid-job: percent=%.1f", late.Percent) + } + + p.Reset() + p.SetDuration(60_000_000) + restarted := p.ParseLine(buildProgressLine(1_000_000, 2.0)) + if restarted == nil || restarted.Percent < 1.66 || restarted.Percent > 1.67 { + t.Fatalf("after Reset the new job's first frame was dropped: %v", restarted) + } +} + +// TestProgressParser_DurationChangeRebasesFloor: the floor is measured in +// percent, so a denominator change re-bases it. Without that, a late header +// Duration: line that grows the denominator would leave every later frame below +// the old floor and the progress would stall for the rest of the job even +// though the media time keeps advancing. +func TestProgressParser_DurationChangeRebasesFloor(t *testing.T) { + p := NewProgressParser() + p.SetDuration(60_000_000) + + if f := p.ParseLine(buildProgressLine(18_000_000, 2.0)); f == nil || f.Percent < 29.99 || f.Percent > 30.01 { + t.Fatalf("first frame percent = %v, want ~30", f) + } + + // The denominator grows to three times the seeded value (the input + // header's Duration: overwrites what the probe seeded), so the same media + // position now reads as a lower percent. + p.SetDuration(180_000_000) + f := p.ParseLine(buildProgressLine(21_000_000, 2.0)) + if f == nil { + t.Fatal("frame after the denominator change was dropped: progress would stall below the old floor") + } + if f.Percent < 11.66 || f.Percent > 11.67 { + t.Fatalf("percent after the re-base = %v, want ~11.7", f.Percent) + } +} + +// TestProgressParser_AttemptBoundaryRebasesETABaseline: StartAttempt — what the +// retry executor reports before each ffmpeg re-run — re-bases the wall-clock +// baseline, so a restarted attempt's ETA is not derived from a clock that still +// includes the dead attempt's downtime. A timeline that merely goes backward +// without that boundary (a legitimate restart such as segment sharding or +// sequential outputs) keeps its baseline. The percent floor is unaffected +// either way: a restarted attempt stays off the wire until it catches up. +func TestProgressParser_AttemptBoundaryRebasesETABaseline(t *testing.T) { + run := func(t *testing.T, newAttemptBoundary bool) int { + t.Helper() + p := NewProgressParser() + p.SetDuration(60_000_000) + wallSec := 0.0 + p.nowFunc = func() time.Time { + return time.Unix(0, int64(wallSec*1e9)) + } + + // Attempt 1 reaches 70% with a live wall-clock rate. + p.ParseLine(buildProgressLine(30_000_000, 2.0)) // seeds the baseline at wall 0 + wallSec = 12 + if f := p.ParseLine(buildProgressLine(42_000_000, 2.0)); f == nil || f.EtaSeconds <= 0 { + t.Fatalf("attempt 1 ETA = %v, want > 0", f) + } + + // ffmpeg dies; the next attempt re-runs from zero a second later. + wallSec = 13 + if newAttemptBoundary { + p.StartAttempt() + } + if f := p.ParseLine(buildProgressLine(300_000, 2.0)); f != nil { + t.Fatalf("restarted attempt's frame was pushed: percent=%.1f", f.Percent) + } + + // Past the mark, the retry's ETA is only free of the dead attempt's + // clock when the boundary was reported. + wallSec = 14 + f := p.ParseLine(buildProgressLine(43_000_000, 2.0)) + if f == nil { + t.Fatal("retry frame above the mark was dropped") + } + return f.EtaSeconds + } + + if eta := run(t, true); eta != 0 { + t.Errorf("ETA right after a reported attempt boundary = %ds, want 0 (the dead attempt's clock must not be used)", eta) + } + if eta := run(t, false); eta <= 0 { + t.Errorf("ETA after a bare time= rewind = %ds, want > 0 (a legitimate timeline restart keeps its baseline)", eta) + } + + // Ten seconds into a restarted attempt the ETA returns, computed from the + // restarted attempt's own rate. + p := NewProgressParser() + p.SetDuration(60_000_000) + wallSec := 0.0 + p.nowFunc = func() time.Time { + return time.Unix(0, int64(wallSec*1e9)) + } + p.ParseLine(buildProgressLine(30_000_000, 2.0)) + wallSec = 13 + p.StartAttempt() + p.ParseLine(buildProgressLine(300_000, 2.0)) + wallSec = 14 + p.ParseLine(buildProgressLine(43_000_000, 2.0)) + wallSec = 24 + f := p.ParseLine(buildProgressLine(51_000_000, 2.0)) + if f == nil { + t.Fatal("retry frame above the mark was dropped") + } + if f.EtaSeconds <= 0 { + t.Errorf("ETA 10s into the restarted attempt = %ds, want > 0", f.EtaSeconds) + } +} + +// TestProgressRouter_HeaderDurationDoesNotWidenSeekWindow: for a trimmed job the +// output window is the denominator, so the input header's Duration: line — which +// the router feeds through SetDuration on every job — must not re-base the +// percent onto the full input. Doing so would read a finished trimmed segment as +// a few percent, and with the monotonic floor in place the job would then stall +// there for the rest of the run. +func TestProgressRouter_HeaderDurationDoesNotWidenSeekWindow(t *testing.T) { + var mu sync.Mutex + var percents []float64 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + var req protocol.JobUpdateRequest + if err := json.Unmarshal(body, &req); err == nil { + mu.Lock() + percents = append(percents, req.Progress) + mu.Unlock() + } + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + client := NewClient(srv.URL, "worker-1", "") + router := NewProgressRouter(client, "job-1", func(string) {}) + router.sendDelay = 0 // the throttle is not what this test pins + + // -ss 240 -t 60 over a 5-minute input: a 60s output window. + router.SetDuration(300_000_000) + router.SetSeekWindow(300_000_000, 240_000_000, 60_000_000, 0) + + h := router.Handler() + h(" Duration: 00:05:00.00, start: 0.000000, bitrate: 1000 kb/s\n") + h(buildProgressLine(30_000_000, 2.0) + "\n") // 30s into the 60s window + + mu.Lock() + defer mu.Unlock() + if len(percents) != 1 { + t.Fatalf("pushed %d progress updates, want 1: %v", len(percents), percents) + } + if percents[0] < 49.99 || percents[0] > 50.01 { + t.Errorf("percent = %v, want ~50 (the 60s output window, not the 5-minute input)", percents[0]) + } +} diff --git a/pkg/worker/progress_router.go b/pkg/worker/progress_router.go index f4b7fc2..fcbe5e7 100644 --- a/pkg/worker/progress_router.go +++ b/pkg/worker/progress_router.go @@ -46,6 +46,15 @@ func (r *ProgressRouter) SetSeekWindow(inputDurationUs, seekUs, tUs, toUs int64) r.parser.SetSeekWindow(inputDurationUs, seekUs, tUs, toUs) } +// StartAttempt tells the parser that a new ffmpeg process is starting for the +// same job (the retry executor's next attempt), so it re-bases the attempt-scoped +// state its ETA and displayed speed are derived from. The monotonic floor is +// intentionally not re-based: it spans attempts, which is what keeps a re-run +// from rewinding the progress the CLI already showed. +func (r *ProgressRouter) StartAttempt() { + r.parser.StartAttempt() +} + // SendFinal sends a terminal 100% progress update to the server so CLI // clients always see a completed progress line even when ffmpeg's last // -stats frame landed at ~95%. Bypasses the throttle: the job is done. diff --git a/pkg/worker/retry_executor.go b/pkg/worker/retry_executor.go index 89c60cc..8573bd4 100644 --- a/pkg/worker/retry_executor.go +++ b/pkg/worker/retry_executor.go @@ -220,7 +220,10 @@ func NewRetryExecutorWithFallback(executor *Executor, config *RetryConfig, fallb // stderrHandler, when non-nil, receive stdout chunks / stderr lines on every // retry attempt so streamed data reaches the client and the full ffmpeg log // streams to the CLI in real time on each attempt. -func (e *RetryExecutor) ExecuteWithRetry(ctx context.Context, args []string, outputPath string, networkOutput bool, stdoutHandler StdoutHandler, stderrHandler StderrHandler) *RetryResult { +// onAttemptStart, when non-nil, is called once before each attempt launches +// ffmpeg, so callers can re-base state that is scoped to one ffmpeg process +// (the progress parser's ETA baseline and speed EWMA). +func (e *RetryExecutor) ExecuteWithRetry(ctx context.Context, args []string, outputPath string, networkOutput bool, stdoutHandler StdoutHandler, stderrHandler StderrHandler, onAttemptStart func()) *RetryResult { result := &RetryResult{ AuditTrail: make([]RetryAuditEntry, 0), FinalStage: RetryStageInitial, @@ -257,6 +260,12 @@ func (e *RetryExecutor) ExecuteWithRetry(ctx context.Context, args []string, out attemptStart := time.Now() stage = e.determineStage(attempt) + // This attempt runs a fresh ffmpeg process: announce it before any of + // its stderr reaches the caller's handler. + if onAttemptStart != nil { + onAttemptStart() + } + // Idempotency: a failed attempt may have left a partial output file // behind. A later attempt that exits 0 without producing output would // otherwise be misjudged as successful by the os.Stat validation. @@ -409,6 +418,10 @@ func (e *RetryExecutor) ExecuteWithRetry(ctx context.Context, args []string, out if fallbackArgs != nil { log.Printf("Attempting final software encoder fallback") attemptStart := time.Now() + // Same boundary as a retry attempt: this is a fresh ffmpeg process. + if onAttemptStart != nil { + onAttemptStart() + } var execResult ExecResult if stderrHandler != nil { execResult = e.executor.ExecuteWithHandlers(ctx, fallbackArgs, stdoutHandler, stderrHandler) diff --git a/pkg/worker/retry_executor_test.go b/pkg/worker/retry_executor_test.go index 5320d41..972a117 100644 --- a/pkg/worker/retry_executor_test.go +++ b/pkg/worker/retry_executor_test.go @@ -561,7 +561,7 @@ func TestExecuteWithRetry_StdoutOutputSkipsFileValidation(t *testing.T) { // ffmpeg exits 0 writing only to stdout ("-"); no local file is created. args := []string{"-f", "lavfi", "-i", "testsrc=duration=0.1", "-f", "mpegts", "-"} - result := re.ExecuteWithRetry(context.Background(), args, "-", false, nil, nil) + result := re.ExecuteWithRetry(context.Background(), args, "-", false, nil, nil, nil) if !result.Success { t.Fatalf("expected success for stdout output, got stage=%s attempts=%d", @@ -615,7 +615,7 @@ func TestExecuteWithRetry_SigAbortRetriesAndSucceeds(t *testing.T) { } args := []string{"-i", "input.mp4", "-c:v", "libx264", "-f", "mpegts", "pipe:1"} - result := re.ExecuteWithRetry(context.Background(), args, "-", false, nil, nil) + result := re.ExecuteWithRetry(context.Background(), args, "-", false, nil, nil, nil) if !result.Success { t.Fatalf("expected retry success after SIGABRT, got stage=%s attempts=%d", @@ -701,7 +701,7 @@ func TestExecuteWithRetry_StreamsStderrEveryAttempt(t *testing.T) { var streamed []string handler := func(chunk string) { streamed = append(streamed, chunk) } - result := re.ExecuteWithRetry(context.Background(), []string{"-f", "lavfi", "-i", "testsrc", "-f", "mpegts", "-"}, "-", false, nil, handler) + result := re.ExecuteWithRetry(context.Background(), []string{"-f", "lavfi", "-i", "testsrc", "-f", "mpegts", "-"}, "-", false, nil, handler, nil) if !result.Success { t.Fatalf("expected retry success after SIGABRT, got stage=%s attempts=%d", diff --git a/pkg/worker/worker.go b/pkg/worker/worker.go index 19339aa..83ac0ad 100644 --- a/pkg/worker/worker.go +++ b/pkg/worker/worker.go @@ -1194,7 +1194,7 @@ func (w *Worker) processJob(ctx context.Context, job protocol.JobInfo, cancel co progressRouter.Handler()(notification + "\n") // Use RetryExecutor for multi-stage retry: - retryResult := w.retryExecutor.ExecuteWithRetry(execCtx, args, outputPath, networkOutput, stdoutHandler, progressRouter.Handler()) + retryResult := w.retryExecutor.ExecuteWithRetry(execCtx, args, outputPath, networkOutput, stdoutHandler, progressRouter.Handler(), progressRouter.StartAttempt) if retryResult.Success { // Retry succeeded — use the final result