TSI-3582: fix(worker): 进度序列单次跳变（71.1%→0.5%）解析器丢弃回退帧

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: <key>`、编码器改写/回退）不属于进度输出，`--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

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

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
 				}

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)
+			}
 		}
 	}
 

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)
+	}
+}

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")