From 6f150495d5833b626cde9da7dde6877b0b884d2d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=AD=A6=E8=94=B5=E4=B8=80=E5=BC=98?= Date: Sat, 3 Oct 2026 22:56:14 +0900 Subject: [PATCH] =?UTF-8?q?worker=20=E3=82=92=E8=BF=BD=E5=8A=A0=EF=BC=88yt?= =?UTF-8?q?-dlp=20=E3=81=AB=E3=82=88=E3=82=8B=E3=83=80=E3=82=A6=E3=83=B3?= =?UTF-8?q?=E3=83=AD=E3=83=BC=E3=83=89=E3=80=81=E6=AE=B5=E9=9A=8E=E5=88=A5?= =?UTF-8?q?=E9=80=B2=E6=8D=97=E3=80=81ffprobe=20=E6=A4=9C=E6=9F=BB?= =?UTF-8?q?=E3=80=81=E5=AE=9F=E8=A1=8C=E4=B8=AD=E3=81=AE=E5=8F=96=E3=82=8A?= =?UTF-8?q?=E6=B6=88=E3=81=97=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 1 + Dockerfile.worker | 21 + PROJECT.md | 22 +- cmd/worker/main.go | 82 +++ cmd/worker/main_test.go | 97 ++++ compose.yaml | 12 + internal/store/store.go | 17 + internal/store/store_test.go | 41 ++ internal/web/handler.go | 2 +- internal/web/queue.go | 43 +- internal/web/queue_test.go | 12 +- internal/web/store_test.go | 2 +- internal/web/templates/queue.html | 3 + internal/worker/progress.go | 81 +++ internal/worker/progress_test.go | 27 + internal/worker/runner.go | 500 ++++++++++++++++++ internal/worker/runner_test.go | 643 +++++++++++++++++++++++ internal/worker/testdata/fake-ffprobe.sh | 17 + internal/worker/testdata/fake-yt-dlp.sh | 67 +++ 19 files changed, 1657 insertions(+), 33 deletions(-) create mode 100644 Dockerfile.worker create mode 100644 cmd/worker/main.go create mode 100644 cmd/worker/main_test.go create mode 100644 internal/worker/progress.go create mode 100644 internal/worker/progress_test.go create mode 100644 internal/worker/runner.go create mode 100644 internal/worker/runner_test.go create mode 100755 internal/worker/testdata/fake-ffprobe.sh create mode 100755 internal/worker/testdata/fake-yt-dlp.sh diff --git a/.env.example b/.env.example index c34ef25..136c0c6 100644 --- a/.env.example +++ b/.env.example @@ -1,3 +1,4 @@ DATABASE_URL= VIDEO_HOST_DIR= VIDEO_DIR=/videos +WORKER_CONCURRENCY=1 diff --git a/Dockerfile.worker b/Dockerfile.worker new file mode 100644 index 0000000..b0da7fc --- /dev/null +++ b/Dockerfile.worker @@ -0,0 +1,21 @@ +FROM golang:1.27.1-alpine AS build + +WORKDIR /src +COPY go.mod go.sum ./ +RUN go mod download +COPY cmd ./cmd +COPY internal ./internal +RUN CGO_ENABLED=0 go build -trimpath -ldflags="-s -w" -o /worker ./cmd/worker + +FROM python:3.13-slim +ARG YT_DLP_VERSION=2026.08.19 +RUN apt-get update \ + && apt-get install -y --no-install-recommends ca-certificates ffmpeg tzdata \ + && pip install --no-cache-dir "yt-dlp[default]==${YT_DLP_VERSION}" \ + && rm -rf /var/lib/apt/lists/* +WORKDIR /app +COPY --from=build /worker /app/worker +ENV HOME=/tmp \ + XDG_CACHE_HOME=/tmp/yt-dlp-cache +USER 65532:65532 +ENTRYPOINT ["/app/worker"] diff --git a/PROJECT.md b/PROJECT.md index 9bb3a4a..91cb7d4 100644 --- a/PROJECT.md +++ b/PROJECT.md @@ -3,7 +3,7 @@ title: tver の現状 tags: - homelab - tver -status: ライブラリ・再生・削除を実装中 +status: worker のダウンロード本体を実装中 --- # tver の現状 @@ -13,8 +13,8 @@ status: ライブラリ・再生・削除を実装中 ## 今の状態 -- Go の HTTP サーバー、TVer API クライアント、ドラマ・バラエティの新着表示、テスト、web コンテナの定義がある -- **ライブラリ・再生・削除を実装中。** 完了したジョブを一覧し、ブラウザで Range 対応の再生とファイル・ジョブ記録の削除ができる。worker がまだ無いため、実際のダウンロード実行は行わない +- Go の HTTP サーバー、TVer API クライアント、ドラマ・バラエティの新着表示、テスト、web・worker コンテナの定義がある +- **worker のダウンロード本体を実装中。** 待ち行列から yt-dlp を起動し、進捗・heartbeat・取消・出力検査・保存・失敗分類を扱う。登録番組の定期確認はまだ行わない - リポジトリ: `https://gitea.kazu634.com/kazu634/tver.git` ## 目的 @@ -45,7 +45,12 @@ TVer（`https://tver.jp/`）で公開中の動画情報を取得し、選んだ | 番組の一時停止 | 登録は残し、自動ダウンロードだけ止める。再開後は停止中に公開された未確認の回も対象にする | | 動画のマウント先 | web と worker は `VIDEO_DIR`（既定 `/videos`）に同じ NAS 共有をマウントする。`download_jobs.file_path` は `VIDEO_DIR` からの相対パスを保存する | | ファイル名 | `<番組名>/<番組名> <回タイトル> [<エピソードID>].mp4`。禁止文字と制御文字を `_` に置換し、空の要素は `名称未設定`、各入力要素は UTF-8 で60バイトまでに切り詰める。エピソード ID で同名衝突を避ける | +| 同名ファイル | 移動先に同名ファイルがあれば上書きする。ファイル名にエピソード ID が入り `episode_id` は一意なので、同名は同じ回の前回処理の残りに限られる。再実行で同じ結果になることを優先する | | 動画の削除 | ファイルを先に削除してからジョブ記録を削除する。ファイルが既に無い場合も記録は削除する。登録番組の `known_episodes` は残す | +| 同時ダウンロード数 | 既定は **1**。`WORKER_CONCURRENCY` で **2まで**増やせる | +| ダウンロードの再試行 | 自動では再試行しない。失敗を `failed` で残し、画面から手動で再試行する | +| 実行中の取り消し | web で `running` を `canceled` にし、worker が検知して yt-dlp のプロセスグループを停止する | +| yt-dlp の作業場所 | `VIDEO_DIR/.incoming//` に `%(id)s.%(ext)s` で保存し、検査後に Go が `media.RelativeVideoPath` の場所へ同一ファイルシステム内で移動する。終了状態にかかわらず作業場所を片付ける | ## 要件 @@ -84,13 +89,15 @@ TVer（`https://tver.jp/`）で公開中の動画情報を取得し、選んだ - 登録と公開中の回の既知記録・任意の待ち行列投入は1トランザクションで行い、登録時刻を最終確認日時に設定する。`known_episodes` は番組解除時に連動して削除する。worker 用の未確認回の抽出と、未確認回だけの既知記録・投入も `internal/store` に用意した。一時停止中は投入を拒否する - 新着の登録ボタンはエピソード ID とジャンルだけを送る。web が有効な新着キャッシュから番組情報を引き、`episode_id` の一意制約で重複登録を防ぐ。新着一覧はジョブ状態を一括取得し、失敗・取り消しには再試行を表示する - 待ち行列は実行中・待機中・失敗・取り消し・完了を状態別に直近50件ずつ表示する。段階別の進捗と失敗理由を表示し、待機中の取り消しと失敗・取り消しの再試行ができる。実行中または待機中のジョブがある間だけ5秒ごとに一覧を更新する +- 実行中のジョブにも取り消しボタンを表示する。worker は DB の `canceled` を定期確認し、yt-dlp をプロセスグループごと停止する - 新着はジャンルごとにメモリへ 10 分間キャッシュする。キャッシュ中は TVer セッションの再発行と一覧の再取得を行わない。TVer の取得に失敗した場合、一覧の場所にエラーを表示して詳細をログへ記録する - 「予告を隠す」は既定でオンとし、180 秒以下の件を非表示にする。長さは秒から分へ切り上げて表示し、配信終了日時は Asia/Tokyo で表示する。ジャンルと予告表示の状態は URL に残す - 画面は `html/template` と htmx 2.0.11 を使い、htmx は Go の `embed` から `/static/` でローカル配信する -- `compose.yaml` は web のみを定義し、dockage VM の LAN アドレス `192.168.10.155:8080` で公開する。待ち受けは `WEB_ADDR` で変更でき、既定は `:8080`。DB 接続先は `DATABASE_URL`、NAS のホスト側マウント先は `VIDEO_HOST_DIR`、コンテナ内の保存先は `VIDEO_DIR`（既定 `/videos`）として `.env` から渡す（値の無い雛形は `.env.example`） +- `compose.yaml` は web と worker を定義する。web は dockage VM の LAN アドレス `192.168.10.155:8080` で公開し、待ち受けは `WEB_ADDR` で変更できる。両方へ `DATABASE_URL` と同じ動画ボリュームを渡し、worker の並行数は `WORKER_CONCURRENCY`（既定1、最大2）で指定する - DB は NAS 上に tver 専用 Postgres 17 コンテナを置く。`internal/store` の埋め込み SQL を `schema_migrations` で管理し、`pg_advisory_lock` で複数プロセスからの同時適用を直列化する。web は起動時に `pgxpool` で接続してマイグレーションを適用する - `download_jobs` は `episode_id` を一意にして二重登録を防ぎ、ジョブの状態・段階別進捗・結果を持つ。`subscriptions` はシリーズ単位の登録と最終確認日時を持つ。worker からの操作と画面用の一覧取得を `internal/store` にまとめる -- worker は `yt-dlp[default]` と ffmpeg を外部コマンドとして使い、映像・音声・結合・移動の段階別進捗を扱う。正確な全体パーセントは保存しない。`after_move:filepath` で実パスを取り、ffprobe で H.264 と AAC の MP4 を確認して成功とする。失敗は DRM、地域制限、ネットワーク、配信不可、抽出失敗、出力不正、その他の順に分類する +- worker は `yt-dlp[default]` と ffmpeg を外部コマンドとして使い、映像・音声・結合・移動の段階別進捗を1秒間隔に間引いて保存する。独立した heartbeat を5秒ごとに更新し、正確な全体パーセントは保存しない。`after_move:filepath` で実パスを取り、ffprobe で H.264 と AAC の MP4 を確認して成功とする。失敗は取り消し、DRM、地域制限、ネットワーク、配信不可、抽出失敗、出力不正、その他の順に扱い、自動再試行しない +- worker は起動時にマイグレーションと停止中ジョブの `RequeueStale` を行う。SIGTERM では yt-dlp を止め、担当中ジョブを `queued` に戻してから終了する。`Dockerfile.worker` は yt-dlp 2026.08.19 を ARG で固定し、ffmpeg・ffprobe とともに非 root で実行する - 保存先の NAS 共有フォルダは dockage VM に NFS でマウントし、`worker`（書き込み）と `web`（読み取り・削除）に同じ `VIDEO_DIR` でバインドマウントする ## 外部とのつながり @@ -121,7 +128,7 @@ TVer（`https://tver.jp/`）で公開中の動画情報を取得し、選んだ - 新着 API が 100 件で打ち切りか、続きを取得できるか [要確認] - 新着のうち DRM 付きの作品の割合 [要確認] - 非公開 API と yt-dlp の TVer 対応は、TVer 側の変更で予告なく壊れうる。壊れたことを検知する手段が要る -- 登録番組の24時間ごとの確認と、待ち行列のジョブを実行する worker は未実装 +- 登録番組の24時間ごとの確認は未実装 - NAS の準備（NFS 共有フォルダ、tver 専用 Postgres 17 コンテナ、DSM ファイアウォール）は DSM での手作業になる。DB の公開ポート（候補 15434）は本番前に実機で確認する [要確認] - DRM・地域制限の yt-dlp 失敗時の文言と終了コードは未実測 [要確認] @@ -134,5 +141,6 @@ TVer（`https://tver.jp/`）で公開中の動画情報を取得し、選んだ - 番組一覧の応答は `internal/tver/testdata/series_seasons.json` と `season_episodes.json`（2026-10-03 の実応答を各1件に縮小）に記録した。テストは `httptest` を使い、実 API にアクセスしない - 新着画面のテストは偽物の TVer クライアント・store と差し替え可能な時計を使い、表示項目、絞り込みの境界、10 分キャッシュ、取得失敗、JST 表示、登録・状態表示・待ち行列の操作を確認する - ライブラリのテストは一時ディレクトリの小さな動画を使い、Range の 206・範囲外の 416、相対パスとシンボリックリンクの検査、404、ファイル削除後の記録削除を確認する。共有のファイル名規則は `internal/media` でテストする +- worker のテストは `internal/worker/testdata` の偽 yt-dlp・ffprobe を使い、本物の TVer へ接続せずに正常完了、失敗分類、取消、終了時の待機復帰、進捗の間引き、heartbeat、出力検査、移動先、作業場所の片付けを確認する - `env GOPATH=/private/tmp/tver-gopath GOCACHE=/private/tmp/tver-go-cache GOMODCACHE=/private/tmp/tver-go-modcache GOPROXY=off go vet ./...` で静的解析する -- `DATABASE_URL`、`VIDEO_DIR`、`WEB_ADDR=127.0.0.1:18080` を設定して `go run ./cmd/web` で起動し、`/healthz`、`/`、`/queue`、`/subscriptions`、`/library` を確認できる。TVer への実アクセスを伴うため手動確認は少数回に留める +- `DATABASE_URL`、`VIDEO_DIR`、`WEB_ADDR=127.0.0.1:18080` を設定して `go run ./cmd/web` で web を起動できる。worker は同じ `DATABASE_URL` と `VIDEO_DIR` を設定して `go run ./cmd/worker` で起動する。worker の実行は TVer への実アクセスを伴うため手動確認は少数回に留める diff --git a/cmd/worker/main.go b/cmd/worker/main.go new file mode 100644 index 0000000..9b6dbb3 --- /dev/null +++ b/cmd/worker/main.go @@ -0,0 +1,82 @@ +package main + +import ( + "context" + "errors" + "fmt" + "log" + "os" + "os/signal" + "strconv" + "strings" + "syscall" + "time" + + "gitea.kazu634.com/kazu634/tver/internal/store" + workerpkg "gitea.kazu634.com/kazu634/tver/internal/worker" + "github.com/jackc/pgx/v5/pgxpool" +) + +const staleAfter = 30 * time.Second + +func main() { + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + config, err := loadConfig(os.Getenv) + if err != nil { + log.Fatal(err) + } + pool, err := initializeDatabase(ctx, os.Getenv("DATABASE_URL"), time.Now(), pgxpool.New, + store.Migrate, func(ctx context.Context, pool *pgxpool.Pool, cutoff time.Time) (int64, error) { + return store.New(pool).RequeueStale(ctx, cutoff) + }) + if err != nil { + log.Fatal(err) + } + defer pool.Close() + runner, err := workerpkg.New(store.New(pool), config, log.Default()) + if err != nil { + log.Fatal(err) + } + log.Printf("worker started (concurrency=%d)", config.Concurrency) + if err := runner.Run(ctx); err != nil { + log.Fatal(err) + } + log.Print("worker stopped") +} + +func loadConfig(getenv func(string) string) (workerpkg.Config, error) { + concurrency := 1 + if value := strings.TrimSpace(getenv("WORKER_CONCURRENCY")); value != "" { + parsed, err := strconv.Atoi(value) + if err != nil || parsed < 1 || parsed > 2 { + return workerpkg.Config{}, errors.New("WORKER_CONCURRENCY は 1 または 2 を指定してください") + } + concurrency = parsed + } + return workerpkg.Config{VideoDir: getenv("VIDEO_DIR"), YTDLPPath: getenv("YT_DLP_PATH"), + FFprobePath: getenv("FFPROBE_PATH"), Concurrency: concurrency}, nil +} + +func initializeDatabase(ctx context.Context, databaseURL string, now time.Time, + connect func(context.Context, string) (*pgxpool.Pool, error), + migrate func(context.Context, *pgxpool.Pool) error, + requeue func(context.Context, *pgxpool.Pool, time.Time) (int64, error), +) (*pgxpool.Pool, error) { + if strings.TrimSpace(databaseURL) == "" { + return nil, errors.New("DATABASE_URL が設定されていません") + } + pool, err := connect(ctx, databaseURL) + if err != nil { + return nil, errors.New("DATABASE_URL から接続プールを作成できません") + } + if err := migrate(ctx, pool); err != nil { + pool.Close() + return nil, fmt.Errorf("DB マイグレーションに失敗: %w", err) + } + if _, err := requeue(ctx, pool, now.Add(-staleAfter)); err != nil { + pool.Close() + return nil, fmt.Errorf("停止中ジョブの復帰に失敗: %w", err) + } + return pool, nil +} diff --git a/cmd/worker/main_test.go b/cmd/worker/main_test.go new file mode 100644 index 0000000..ba3dee7 --- /dev/null +++ b/cmd/worker/main_test.go @@ -0,0 +1,97 @@ +package main + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +func TestLoadConfigConcurrency(t *testing.T) { + values := map[string]string{"VIDEO_DIR": "/media", "YT_DLP_PATH": "/bin/downloader", "FFPROBE_PATH": "/bin/probe"} + getenv := func(key string) string { return values[key] } + config, err := loadConfig(getenv) + if err != nil || config.Concurrency != 1 || config.VideoDir != "/media" || config.YTDLPPath != "/bin/downloader" || config.FFprobePath != "/bin/probe" { + t.Fatalf("既定設定: %+v %v", config, err) + } + values["WORKER_CONCURRENCY"] = "2" + config, err = loadConfig(getenv) + if err != nil || config.Concurrency != 2 { + t.Fatalf("並行数2: %+v %v", config, err) + } + for _, invalid := range []string{"0", "3", "bad"} { + values["WORKER_CONCURRENCY"] = invalid + if _, err := loadConfig(getenv); err == nil { + t.Errorf("不正な並行数 %q を許した", invalid) + } + } +} + +func TestInitializeDatabaseRequiresURLMigratesAndRequeues(t *testing.T) { + ctx := context.Background() + now := time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC) + connectCalls, migrateCalls, requeueCalls := 0, 0, 0 + connect := func(_ context.Context, url string) (*pgxpool.Pool, error) { + connectCalls++ + if url != "postgres://test:secret@localhost/test" { + t.Errorf("URL=%q", url) + } + return pgxpool.New(ctx, url) + } + migrate := func(context.Context, *pgxpool.Pool) error { migrateCalls++; return nil } + requeue := func(_ context.Context, _ *pgxpool.Pool, cutoff time.Time) (int64, error) { + requeueCalls++ + if want := now.Add(-staleAfter); !cutoff.Equal(want) { + t.Errorf("cutoff=%v want=%v", cutoff, want) + } + return 2, nil + } + if _, err := initializeDatabase(ctx, " ", now, connect, migrate, requeue); err == nil || !strings.Contains(err.Error(), "DATABASE_URL") { + t.Fatalf("未設定 URL: %v", err) + } + if connectCalls != 0 || migrateCalls != 0 || requeueCalls != 0 { + t.Fatalf("未設定時の呼び出し: %d %d %d", connectCalls, migrateCalls, requeueCalls) + } + pool, err := initializeDatabase(ctx, "postgres://test:secret@localhost/test", now, connect, migrate, requeue) + if err != nil || pool == nil { + t.Fatalf("初期化: %v", err) + } + pool.Close() + if connectCalls != 1 || migrateCalls != 1 || requeueCalls != 1 { + t.Fatalf("呼び出し回数: %d %d %d", connectCalls, migrateCalls, requeueCalls) + } +} + +func TestInitializeDatabaseStopsOnMigrationOrRequeueFailure(t *testing.T) { + ctx := context.Background() + connect := func(context.Context, string) (*pgxpool.Pool, error) { + return pgxpool.New(ctx, "postgres://localhost/test") + } + requeueCalled := false + pool, err := initializeDatabase(ctx, "postgres://localhost/test", time.Now(), connect, + func(context.Context, *pgxpool.Pool) error { return errors.New("migration failed") }, + func(context.Context, *pgxpool.Pool, time.Time) (int64, error) { requeueCalled = true; return 0, nil }) + if pool != nil || err == nil || !strings.Contains(err.Error(), "migration failed") || requeueCalled { + t.Fatalf("migration failure: pool=%v err=%v requeue=%v", pool, err, requeueCalled) + } + pool, err = initializeDatabase(ctx, "postgres://localhost/test", time.Now(), connect, + func(context.Context, *pgxpool.Pool) error { return nil }, + func(context.Context, *pgxpool.Pool, time.Time) (int64, error) { return 0, errors.New("requeue failed") }) + if pool != nil || err == nil || !strings.Contains(err.Error(), "requeue failed") { + t.Fatalf("requeue failure: pool=%v err=%v", pool, err) + } +} + +func TestInitializeDatabaseDoesNotLeakURL(t *testing.T) { + _, err := initializeDatabase(context.Background(), "postgres://user:secret@localhost/test", time.Now(), + func(context.Context, string) (*pgxpool.Pool, error) { + return nil, errors.New("postgres://user:secret@localhost/test refused") + }, func(context.Context, *pgxpool.Pool) error { return nil }, + func(context.Context, *pgxpool.Pool, time.Time) (int64, error) { return 0, nil }) + if err == nil || strings.Contains(err.Error(), "secret") { + t.Fatalf("URL 漏えい: %v", err) + } +} diff --git a/compose.yaml b/compose.yaml index 5fa778b..7f80a49 100644 --- a/compose.yaml +++ b/compose.yaml @@ -11,3 +11,15 @@ services: ports: - "192.168.10.155:8080:8080" restart: unless-stopped + worker: + build: + context: . + dockerfile: Dockerfile.worker + env_file: .env + environment: + TZ: Asia/Tokyo + VIDEO_DIR: "${VIDEO_DIR:-/videos}" + WORKER_CONCURRENCY: "${WORKER_CONCURRENCY:-1}" + volumes: + - "${VIDEO_HOST_DIR:?VIDEO_HOST_DIR must be set}:${VIDEO_DIR:-/videos}" + restart: unless-stopped diff --git a/internal/store/store.go b/internal/store/store.go index cc1d9a8..3c8667c 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -183,6 +183,13 @@ func (s *Store) UpdateProgress(ctx context.Context, id int64, p Progress) error return transitionResult(tag.RowsAffected(), err) } +// Heartbeat は進捗出力がない間も実行中ジョブの生存を記録する。 +func (s *Store) Heartbeat(ctx context.Context, id int64) error { + tag, err := s.pool.Exec(ctx, `UPDATE download_jobs SET heartbeat_at=now(), updated_at=now() + WHERE id=$1 AND status='running'`, id) + return transitionResult(tag.RowsAffected(), err) +} + func (s *Store) Complete(ctx context.Context, id int64, filePath string, fileSizeBytes int64) error { tag, err := s.pool.Exec(ctx, `UPDATE download_jobs SET status='succeeded', phase=NULL, file_path=$2, file_size_bytes=$3, heartbeat_at=NULL, finished_at=now(), updated_at=now() @@ -222,6 +229,16 @@ func (s *Store) RequeueStale(ctx context.Context, cutoff time.Time) (int64, erro return tag.RowsAffected(), err } +// RequeueRunning は worker の正常終了時に担当中ジョブを待機中へ戻す。 +func (s *Store) RequeueRunning(ctx context.Context, id int64) error { + tag, err := s.pool.Exec(ctx, `UPDATE download_jobs SET status='queued', phase=NULL, + downloaded_bytes=NULL, total_bytes=NULL, total_is_estimate=false, speed_bps=NULL, + eta_seconds=NULL, fragment_index=NULL, fragment_count=NULL, format_id=NULL, + heartbeat_at=NULL, started_at=NULL, updated_at=now() + WHERE id=$1 AND status='running'`, id) + return transitionResult(tag.RowsAffected(), err) +} + func transitionResult(count int64, err error) error { if err != nil { return err diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 2e41f5f..401cfc6 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -405,6 +405,47 @@ func TestCancelAndStaleBoundary(t *testing.T) { } } +func TestHeartbeatAndGracefulRequeue(t *testing.T) { + s := New(testPool(t, true)) + ctx := context.Background() + job, _, err := s.Enqueue(ctx, sampleEpisode("heartbeat"), "manual") + if err != nil { + t.Fatal(err) + } + if err := s.Heartbeat(ctx, job.ID); !errors.Is(err, ErrInvalidTransition) { + t.Fatalf("待機中の heartbeat: %v", err) + } + claimed, err := s.ClaimNext(ctx) + if err != nil { + t.Fatal(err) + } + old := time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC) + if _, err := s.pool.Exec(ctx, `UPDATE download_jobs SET heartbeat_at=$2 WHERE id=$1`, claimed.ID, old); err != nil { + t.Fatal(err) + } + if err := s.Heartbeat(ctx, claimed.ID); err != nil { + t.Fatal(err) + } + heartbeated, err := s.GetJob(ctx, claimed.ID) + if err != nil || heartbeated.HeartbeatAt == nil || !heartbeated.HeartbeatAt.After(old) { + t.Fatalf("heartbeat: %+v %v", heartbeated, err) + } + bytes := int64(10) + if err := s.UpdateProgress(ctx, claimed.ID, Progress{Phase: "video", DownloadedBytes: &bytes}); err != nil { + t.Fatal(err) + } + if err := s.RequeueRunning(ctx, claimed.ID); err != nil { + t.Fatal(err) + } + requeued, err := s.GetJob(ctx, claimed.ID) + if err != nil || requeued.Status != "queued" || requeued.StartedAt != nil || requeued.HeartbeatAt != nil || requeued.Phase != nil || requeued.DownloadedBytes != nil || requeued.Attempts != 1 { + t.Fatalf("graceful requeue: %+v %v", requeued, err) + } + if err := s.RequeueRunning(ctx, claimed.ID); !errors.Is(err, ErrInvalidTransition) { + t.Fatalf("待機中の再復帰: %v", err) + } +} + func TestRejectedTransitions(t *testing.T) { pool := testPool(t, true) s := New(pool) diff --git a/internal/web/handler.go b/internal/web/handler.go index c59d0ef..10661a3 100644 --- a/internal/web/handler.go +++ b/internal/web/handler.go @@ -32,7 +32,7 @@ type JobStore interface { GetJob(context.Context, int64) (store.Job, error) Retry(context.Context, int64) error RetryEpisode(context.Context, string) error - CancelQueued(context.Context, int64) error + Cancel(context.Context, int64) error SubscribedSeries(context.Context, []string) (map[string]bool, error) RegisterSubscription(context.Context, string, string, []store.Episode, bool) (bool, error) ListSubscriptions(context.Context) ([]store.Subscription, error) diff --git a/internal/web/queue.go b/internal/web/queue.go index e81b270..22361be 100644 --- a/internal/web/queue.go +++ b/internal/web/queue.go @@ -145,34 +145,37 @@ func errorKindText(kind string) string { } func (s *server) retryJob(w http.ResponseWriter, r *http.Request) { - s.queueMutation(w, r, s.jobs.Retry, false) + s.queueMutation(w, r, s.jobs.Retry) } func (s *server) cancelJob(w http.ResponseWriter, r *http.Request) { - s.queueMutation(w, r, s.jobs.CancelQueued, true) -} - -func (s *server) queueMutation(w http.ResponseWriter, r *http.Request, change func(context.Context, int64) error, queuedOnly bool) { id, err := strconv.ParseInt(r.PathValue("id"), 10, 64) if err != nil || id <= 0 { http.Error(w, "ジョブ ID が不正です", http.StatusBadRequest) return } - if queuedOnly { - job, err := s.jobs.GetJob(r.Context(), id) - if errors.Is(err, pgx.ErrNoRows) { - http.Error(w, "ジョブが見つかりません", http.StatusNotFound) - return - } - if err != nil { - s.logger.Printf("ジョブ状態の取得に失敗: %v", err) - http.Error(w, "操作できませんでした", http.StatusInternalServerError) - return - } - if job.Status != "queued" { - http.Error(w, "待機中のジョブだけ取り消せます", http.StatusConflict) - return - } + job, err := s.jobs.GetJob(r.Context(), id) + if errors.Is(err, pgx.ErrNoRows) { + http.Error(w, "ジョブが見つかりません", http.StatusNotFound) + return + } + if err != nil { + s.logger.Printf("ジョブ状態の取得に失敗: %v", err) + http.Error(w, "操作できませんでした", http.StatusInternalServerError) + return + } + if job.Status != "queued" && job.Status != "running" { + http.Error(w, "待機中またはダウンロード中のジョブだけ取り消せます", http.StatusConflict) + return + } + s.queueMutation(w, r, s.jobs.Cancel) +} + +func (s *server) queueMutation(w http.ResponseWriter, r *http.Request, change func(context.Context, int64) error) { + id, err := strconv.ParseInt(r.PathValue("id"), 10, 64) + if err != nil || id <= 0 { + http.Error(w, "ジョブ ID が不正です", http.StatusBadRequest) + return } if err := change(r.Context(), id); err != nil { if errors.Is(err, store.ErrInvalidTransition) { diff --git a/internal/web/queue_test.go b/internal/web/queue_test.go index 584bc5d..3859761 100644 --- a/internal/web/queue_test.go +++ b/internal/web/queue_test.go @@ -155,7 +155,7 @@ func TestQueueGroupsProgressAndPolling(t *testing.T) { t.Errorf("待ち行列に %q がない", want) } } - if strings.Contains(body, errorMessage) || strings.Contains(body, `action="/jobs/1/cancel"`) || strings.Contains(body, "全体") { + if strings.Contains(body, errorMessage) || !strings.Contains(body, `action="/jobs/1/cancel"`) || strings.Contains(body, "全体") { t.Errorf("不正な詳細・操作・全体進捗: %s", body) } fragment := getPage(handler, "/queue/list") @@ -195,7 +195,7 @@ func TestQueueJobsAppearInCorrectSectionsWithLimit(t *testing.T) { cancelAction := `action="/jobs/` + string(rune('1'+i)) + `/cancel"` retryAction := `action="/jobs/` + string(rune('1'+i)) + `/retry"` switch statuses[i] { - case "queued": + case "running", "queued": if !strings.Contains(section[1], cancelAction) || strings.Contains(section[1], retryAction) { t.Errorf("%s の操作が不正: %s", statuses[i], section[1]) } @@ -246,9 +246,13 @@ func TestQueueCancelRetryAndStopPolling(t *testing.T) { if got := postForm(handler, "/jobs/bad/cancel", nil).Code; got != http.StatusBadRequest { t.Fatalf("不正な ID: %d", got) } + if got := postForm(handler, "/jobs/99/cancel", nil).Code; got != http.StatusNotFound { + t.Fatalf("存在しないジョブ: %d", got) + } jobs.jobs["ep"] = store.Job{ID: 1, Status: "running"} - if got := postForm(handler, "/jobs/1/cancel", nil).Code; got != http.StatusConflict { - t.Fatalf("実行中の取り消し: %d", got) + runningCanceled := postForm(handler, "/jobs/1/cancel", nil) + if runningCanceled.Code != http.StatusOK || jobs.jobs["ep"].Status != "canceled" || !strings.Contains(runningCanceled.Body.String(), `action="/jobs/1/retry"`) { + t.Fatalf("実行中の取り消し: %d %+v", runningCanceled.Code, jobs.jobs["ep"]) } } diff --git a/internal/web/store_test.go b/internal/web/store_test.go index b69646a..a554667 100644 --- a/internal/web/store_test.go +++ b/internal/web/store_test.go @@ -124,7 +124,7 @@ func (f *fakeJobStore) Cancel(_ context.Context, id int64) error { return f.err } for ep, job := range f.jobs { - if job.ID == id && job.Status == "queued" { + if job.ID == id && (job.Status == "queued" || job.Status == "running") { job.Status = "canceled" f.jobs[ep] = job return nil diff --git a/internal/web/templates/queue.html b/internal/web/templates/queue.html index 8055a1f..9a64f38 100644 --- a/internal/web/templates/queue.html +++ b/internal/web/templates/queue.html @@ -38,6 +38,9 @@ {{if .SpeedText}}／{{.SpeedText}}{{end}} {{if .ETAText}}／{{.ETAText}}{{end}} {{if .FragmentText}}／{{.FragmentText}}{{end}}

+
+ +
{{end}} {{if eq .Status "failed"}}

失敗理由: {{.ErrorKindText}}

diff --git a/internal/worker/progress.go b/internal/worker/progress.go new file mode 100644 index 0000000..5d33993 --- /dev/null +++ b/internal/worker/progress.go @@ -0,0 +1,81 @@ +package worker + +import ( + "encoding/json" + "strings" + + "gitea.kazu634.com/kazu634/tver/internal/store" +) + +type outputEvent struct { + Event string `json:"event"` + FormatID string `json:"format_id"` + Filepath string `json:"filepath"` + Progress struct { + DownloadedBytes *float64 `json:"downloaded_bytes"` + TotalBytes *float64 `json:"total_bytes"` + TotalBytesEstimate *float64 `json:"total_bytes_estimate"` + Speed *float64 `json:"speed"` + ETA *float64 `json:"eta"` + FragmentIndex *float64 `json:"fragment_index"` + FragmentCount *float64 `json:"fragment_count"` + } `json:"progress"` +} + +func parseProgress(line string) (store.Progress, bool) { + var event outputEvent + if err := json.Unmarshal([]byte(line), &event); err != nil { + return store.Progress{}, false + } + phase := "" + switch event.Event { + case "download": + if strings.Contains(strings.ToLower(event.FormatID), "audio") { + phase = "audio" + } else { + phase = "video" + } + case "postprocess": + phase = "merging" + default: + return store.Progress{}, false + } + total := int64Pointer(event.Progress.TotalBytes) + estimate := false + if total == nil && event.Progress.TotalBytesEstimate != nil { + total = int64Pointer(event.Progress.TotalBytesEstimate) + estimate = true + } + var formatID *string + if event.FormatID != "" { + formatID = &event.FormatID + } + return store.Progress{Phase: phase, DownloadedBytes: int64Pointer(event.Progress.DownloadedBytes), + TotalBytes: total, TotalIsEstimate: estimate, SpeedBPS: event.Progress.Speed, + ETASeconds: event.Progress.ETA, FragmentIndex: int32Pointer(event.Progress.FragmentIndex), + FragmentCount: int32Pointer(event.Progress.FragmentCount), FormatID: formatID}, true +} + +func int64Pointer(value *float64) *int64 { + if value == nil { + return nil + } + result := int64(*value) + return &result +} + +func int32Pointer(value *float64) *int32 { + if value == nil { + return nil + } + result := int32(*value) + return &result +} + +func parseAfterMove(line string) (string, bool) { + var event outputEvent + if err := json.Unmarshal([]byte(line), &event); err != nil || event.Event != "after_move" || event.Filepath == "" { + return "", false + } + return event.Filepath, true +} diff --git a/internal/worker/progress_test.go b/internal/worker/progress_test.go new file mode 100644 index 0000000..7fda9fe --- /dev/null +++ b/internal/worker/progress_test.go @@ -0,0 +1,27 @@ +package worker + +import "testing" + +func TestParseProgress(t *testing.T) { + video, ok := parseProgress(`{"event":"download","format_id":"hls-698-1","progress":{"downloaded_bytes":25,"total_bytes_estimate":100.0,"speed":10.5,"eta":3,"fragment_index":1,"fragment_count":4}}`) + if !ok || video.Phase != "video" || video.DownloadedBytes == nil || *video.DownloadedBytes != 25 || video.TotalBytes == nil || *video.TotalBytes != 100 || !video.TotalIsEstimate || video.SpeedBPS == nil || *video.SpeedBPS != 10.5 || video.ETASeconds == nil || *video.ETASeconds != 3 || video.FragmentIndex == nil || *video.FragmentIndex != 1 || video.FragmentCount == nil || *video.FragmentCount != 4 || video.FormatID == nil || *video.FormatID != "hls-698-1" { + t.Fatalf("映像進捗: %+v ok=%v", video, ok) + } + audio, ok := parseProgress(`{"event":"download","format_id":"hls-ts_AUDIO-0_2","progress":{"downloaded_bytes":50,"total_bytes":50}}`) + if !ok || audio.Phase != "audio" || audio.TotalIsEstimate || audio.TotalBytes == nil || *audio.TotalBytes != 50 { + t.Fatalf("音声進捗: %+v ok=%v", audio, ok) + } + merging, ok := parseProgress(`{"event":"postprocess","progress":{"status":"started","postprocessor":"Merger"}}`) + if !ok || merging.Phase != "merging" { + t.Fatalf("結合進捗: %+v ok=%v", merging, ok) + } + path, ok := parseAfterMove(`{"event":"after_move","filepath":"/videos/incoming/ep.mp4"}`) + if !ok || path != "/videos/incoming/ep.mp4" { + t.Fatalf("移動後パス: %q ok=%v", path, ok) + } + for _, invalid := range []string{"noise", `{"event":"other"}`} { + if _, ok := parseProgress(invalid); ok { + t.Errorf("不正な進捗を許した: %q", invalid) + } + } +} diff --git a/internal/worker/runner.go b/internal/worker/runner.go new file mode 100644 index 0000000..81f54b8 --- /dev/null +++ b/internal/worker/runner.go @@ -0,0 +1,500 @@ +package worker + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log" + "os" + "os/exec" + "path/filepath" + "regexp" + "strconv" + "strings" + "sync" + "syscall" + "time" + + "gitea.kazu634.com/kazu634/tver/internal/media" + "gitea.kazu634.com/kazu634/tver/internal/store" + "github.com/jackc/pgx/v5" +) + +var ( + errCanceled = errors.New("job canceled") + errShutdown = errors.New("worker shutting down") + errInvalidOutput = errors.New("invalid output") + headerSecretRE = regexp.MustCompile(`(?im)(authorization|cookie)([=:][ \t]*)([^\r\n]+)`) + paramSecretRE = regexp.MustCompile(`(?i)(token|platform_uid|platform_token)([=:][[:space:]]*)([^&[:space:]]+)`) +) + +type JobStore interface { + ClaimNext(context.Context) (store.Job, error) + UpdateProgress(context.Context, int64, store.Progress) error + Heartbeat(context.Context, int64) error + GetJob(context.Context, int64) (store.Job, error) + Complete(context.Context, int64, string, int64) error + Fail(context.Context, int64, string, string) error + RequeueRunning(context.Context, int64) error +} + +type Config struct { + VideoDir string + YTDLPPath string + FFprobePath string + Concurrency int + PollInterval time.Duration + ProgressInterval time.Duration + HeartbeatInterval time.Duration + CancelInterval time.Duration +} + +type Runner struct { + store JobStore + config Config + logger *log.Logger +} + +type cappedBuffer struct { + buffer bytes.Buffer + limit int +} + +func (b *cappedBuffer) Write(p []byte) (int, error) { + written := len(p) + remaining := b.limit - b.buffer.Len() + if remaining > 0 { + if remaining < len(p) { + p = p[:remaining] + } + _, _ = b.buffer.Write(p) + } + return written, nil +} + +func (b *cappedBuffer) String() string { return b.buffer.String() } + +func scanLines(reader io.Reader, capture *cappedBuffer) <-chan string { + lines := make(chan string) + go func() { + defer close(lines) + scanner := bufio.NewScanner(reader) + scanner.Buffer(make([]byte, 4096), 1<<20) + for scanner.Scan() { + line := scanner.Text() + if capture != nil { + _, _ = capture.Write([]byte(line + "\n")) + } + lines <- line + } + }() + return lines +} + +func New(jobs JobStore, config Config, logger *log.Logger) (*Runner, error) { + if config.VideoDir == "" { + config.VideoDir = "/videos" + } + if config.YTDLPPath == "" { + config.YTDLPPath = "yt-dlp" + } + if config.FFprobePath == "" { + config.FFprobePath = "ffprobe" + } + if config.Concurrency == 0 { + config.Concurrency = 1 + } + if config.Concurrency < 1 || config.Concurrency > 2 { + return nil, errors.New("worker concurrency must be 1 or 2") + } + if config.PollInterval <= 0 { + config.PollInterval = time.Second + } + if config.ProgressInterval <= 0 { + config.ProgressInterval = time.Second + } + if config.HeartbeatInterval <= 0 { + config.HeartbeatInterval = 5 * time.Second + } + if config.CancelInterval <= 0 { + config.CancelInterval = 500 * time.Millisecond + } + if logger == nil { + logger = log.Default() + } + return &Runner{store: jobs, config: config, logger: logger}, nil +} + +func (r *Runner) Run(ctx context.Context) error { + sem := make(chan struct{}, r.config.Concurrency) + var wg sync.WaitGroup + ticker := time.NewTicker(r.config.PollInterval) + defer ticker.Stop() + for { + for len(sem) < cap(sem) && ctx.Err() == nil { + job, err := r.store.ClaimNext(ctx) + if errors.Is(err, pgx.ErrNoRows) { + break + } + if err != nil { + r.logger.Printf("ジョブ取得に失敗: %v", err) + break + } + sem <- struct{}{} + wg.Add(1) + go func() { + defer wg.Done() + defer func() { <-sem }() + r.ProcessJob(ctx, job) + }() + } + select { + case <-ctx.Done(): + wg.Wait() + return nil + case <-ticker.C: + } + } +} + +func (r *Runner) ProcessJob(ctx context.Context, job store.Job) { + workDir := filepath.Join(r.config.VideoDir, ".incoming", strconv.FormatInt(job.ID, 10)) + if err := os.RemoveAll(workDir); err != nil { + r.fail(job.ID, "unknown", fmt.Sprintf("作業ディレクトリを初期化できません: %v", err)) + return + } + if err := os.MkdirAll(workDir, 0o755); err != nil { + r.fail(job.ID, "unknown", fmt.Sprintf("作業ディレクトリを作成できません: %v", err)) + return + } + defer os.RemoveAll(workDir) + + downloaded, stderr, err := r.download(ctx, job, workDir) + if errors.Is(err, errShutdown) { + r.requeue(job.ID) + return + } + if errors.Is(err, errCanceled) { + return + } + if err != nil { + message := redact(stderr) + if message == "" { + message = redact(err.Error()) + } + kind := classifyFailure(message) + if errors.Is(err, errInvalidOutput) { + kind = "invalid_output" + } + r.failIfRunning(job.ID, kind, message) + return + } + if r.canceled(job.ID) { + return + } + if err := r.probe(ctx, downloaded); err != nil { + if ctx.Err() != nil { + r.requeue(job.ID) + return + } + r.failIfRunning(job.ID, "invalid_output", redact(err.Error())) + return + } + if r.canceled(job.ID) { + return + } + if err := r.store.UpdateProgress(ctx, job.ID, store.Progress{Phase: "moving"}); err != nil { + if r.canceled(job.ID) { + return + } + r.failIfRunning(job.ID, "unknown", "移動状態を記録できません: "+err.Error()) + return + } + relative := media.RelativeVideoPath(job.SeriesTitle, job.Title, job.EpisodeID) + destination := filepath.Join(r.config.VideoDir, relative) + if err := os.MkdirAll(filepath.Dir(destination), 0o755); err != nil { + r.failIfRunning(job.ID, "unknown", "保存先を作成できません: "+err.Error()) + return + } + if err := os.Rename(downloaded, destination); err != nil { + r.failIfRunning(job.ID, "unknown", "動画を保存先へ移動できません: "+err.Error()) + return + } + info, err := os.Stat(destination) + if err != nil { + r.failIfRunning(job.ID, "unknown", "保存後の動画を確認できません: "+err.Error()) + return + } + if err := r.store.Complete(ctx, job.ID, filepath.ToSlash(relative), info.Size()); err != nil { + if r.canceled(job.ID) { + if removeErr := os.Remove(destination); removeErr != nil && !errors.Is(removeErr, os.ErrNotExist) { + r.logger.Printf("取り消されたジョブ %d の保存済みファイル削除に失敗: %v", job.ID, removeErr) + } + return + } + r.logger.Printf("ジョブ %d の完了記録に失敗: %v", job.ID, err) + } +} + +func (r *Runner) download(ctx context.Context, job store.Job, workDir string) (string, string, error) { + output := filepath.Join(workDir, "%(id)s.%(ext)s") + args := []string{"--no-playlist", "--no-simulate", "--quiet", "--progress", "--newline", + "--continue", "--retries", "0", "--fragment-retries", "0", "--extractor-retries", "0", + "-f", "bv*[vcodec^=avc1]+ba/b", "--merge-output-format", "mp4", + "--progress-template", `download:{"event":"download","format_id":%(info.format_id)j,"progress":%(progress)j}`, + "--progress-template", `postprocess:{"event":"postprocess","progress":%(progress)j}`, + "--print", `after_move:{"event":"after_move","filepath":%(filepath)j}`, "-o", output, + "https://tver.jp/episodes/" + job.EpisodeID} + cmd := exec.Command(r.config.YTDLPPath, args...) + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + stdout, err := cmd.StdoutPipe() + if err != nil { + return "", "", err + } + stderrPipe, err := cmd.StderrPipe() + if err != nil { + return "", "", err + } + if err := cmd.Start(); err != nil { + return "", "", err + } + stderr := cappedBuffer{limit: 64 << 10} + stdoutLines := scanLines(stdout, nil) + stderrLines := scanLines(stderrPipe, &stderr) + waited := make(chan error, 1) + go func() { waited <- cmd.Wait() }() + heartbeat := time.NewTicker(r.config.HeartbeatInterval) + cancelCheck := time.NewTicker(r.config.CancelInterval) + defer heartbeat.Stop() + defer cancelCheck.Stop() + _ = r.store.Heartbeat(ctx, job.ID) + var afterMove string + var pending *store.Progress + nextProgress := time.Time{} + lastPhase := "" + recordProgress := func(progress store.Progress) { + pending = &progress + now := time.Now() + if progress.Phase != lastPhase || now.After(nextProgress) { + _ = r.store.UpdateProgress(ctx, job.ID, progress) + nextProgress = now.Add(r.config.ProgressInterval) + lastPhase = progress.Phase + pending = nil + } + } + stopping := "" + var forceKill <-chan time.Time + ctxDone := ctx.Done() + waitComplete := false + var waitErr error + for { + if waitComplete && stdoutLines == nil && stderrLines == nil { + if pending != nil && stopping == "" { + updateCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + _ = r.store.UpdateProgress(updateCtx, job.ID, *pending) + cancel() + } + if stopping == "canceled" { + return "", stderr.String(), errCanceled + } + if stopping == "shutdown" { + return "", stderr.String(), errShutdown + } + if waitErr != nil { + r.logger.Printf("yt-dlp ジョブ %d 失敗: %s", job.ID, redact(stderr.String())) + return "", stderr.String(), waitErr + } + path, err := pathInside(workDir, afterMove) + if err != nil { + return "", stderr.String(), err + } + return path, stderr.String(), nil + } + select { + case line, ok := <-stdoutLines: + if !ok { + stdoutLines = nil + continue + } + if path, ok := parseAfterMove(line); ok { + afterMove = path + continue + } + if progress, ok := parseProgress(line); ok { + recordProgress(progress) + } + case line, ok := <-stderrLines: + if !ok { + stderrLines = nil + continue + } + if progress, ok := parseProgress(line); ok { + recordProgress(progress) + } + case <-heartbeat.C: + _ = r.store.Heartbeat(ctx, job.ID) + case <-cancelCheck.C: + if r.canceled(job.ID) && stopping == "" { + stopping = "canceled" + if !waitComplete { + stopProcessGroup(cmd) + forceKill = time.After(2 * time.Second) + } + } + case <-ctxDone: + if stopping == "" { + stopping = "shutdown" + if !waitComplete { + stopProcessGroup(cmd) + forceKill = time.After(2 * time.Second) + } + } + ctxDone = nil + case <-forceKill: + _ = syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL) + forceKill = nil + case waitErr = <-waited: + waitComplete = true + waited = nil + forceKill = nil + } + } +} + +func stopProcessGroup(cmd *exec.Cmd) { + if cmd.Process == nil { + return + } + _ = syscall.Kill(-cmd.Process.Pid, syscall.SIGTERM) +} + +func pathInside(root, candidate string) (string, error) { + if candidate == "" { + return "", errors.New("yt-dlp did not report after_move filepath") + } + rootAbs, err := filepath.Abs(root) + if err != nil { + return "", err + } + candidateAbs, err := filepath.Abs(candidate) + if err != nil { + return "", err + } + rel, err := filepath.Rel(rootAbs, candidateAbs) + if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) { + return "", fmt.Errorf("%w: yt-dlp filepath leaves working directory", errInvalidOutput) + } + info, err := os.Lstat(candidateAbs) + if err != nil { + return "", fmt.Errorf("%w: yt-dlp output file is missing: %v", errInvalidOutput, err) + } + if !info.Mode().IsRegular() { + return "", fmt.Errorf("%w: yt-dlp output is not a regular file", errInvalidOutput) + } + rootReal, err := filepath.EvalSymlinks(rootAbs) + if err != nil { + return "", fmt.Errorf("%w: resolve working directory: %v", errInvalidOutput, err) + } + candidateReal, err := filepath.EvalSymlinks(candidateAbs) + if err != nil { + return "", fmt.Errorf("%w: resolve yt-dlp output: %v", errInvalidOutput, err) + } + realRelative, err := filepath.Rel(rootReal, candidateReal) + if err != nil || realRelative == ".." || strings.HasPrefix(realRelative, ".."+string(filepath.Separator)) { + return "", fmt.Errorf("%w: yt-dlp output resolves outside working directory", errInvalidOutput) + } + return candidateAbs, nil +} + +func (r *Runner) probe(ctx context.Context, path string) error { + cmd := exec.CommandContext(ctx, r.config.FFprobePath, "-v", "error", "-show_entries", + "format=format_name", "-show_entries", "stream=codec_type,codec_name", "-of", "json", path) + output, err := cmd.Output() + if err != nil { + return fmt.Errorf("ffprobe failed: %w", err) + } + var result struct { + Format struct { + FormatName string `json:"format_name"` + } `json:"format"` + Streams []struct { + CodecType string `json:"codec_type"` + CodecName string `json:"codec_name"` + } `json:"streams"` + } + if err := json.Unmarshal(output, &result); err != nil { + return fmt.Errorf("ffprobe output is not JSON: %w", err) + } + video, audio := false, false + for _, stream := range result.Streams { + video = video || stream.CodecType == "video" && stream.CodecName == "h264" + audio = audio || stream.CodecType == "audio" && stream.CodecName == "aac" + } + if !strings.Contains(result.Format.FormatName, "mp4") || !video || !audio { + return fmt.Errorf("expected MP4 with H.264 and AAC, got format=%s video=%t audio=%t", result.Format.FormatName, video, audio) + } + return nil +} + +func (r *Runner) canceled(id int64) bool { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + job, err := r.store.GetJob(ctx, id) + return err == nil && job.Status == "canceled" +} + +func (r *Runner) failIfRunning(id int64, kind, message string) { + if r.canceled(id) { + return + } + r.fail(id, kind, message) +} + +func (r *Runner) fail(id int64, kind, message string) { + if len(message) > 8192 { + message = message[:8192] + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := r.store.Fail(ctx, id, kind, message); err != nil { + r.logger.Printf("ジョブ %d の失敗記録に失敗: %v", id, err) + } +} + +func (r *Runner) requeue(id int64) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := r.store.RequeueRunning(ctx, id); err != nil && !errors.Is(err, store.ErrInvalidTransition) { + r.logger.Printf("ジョブ %d の待機中への復帰に失敗: %v", id, err) + } +} + +func classifyFailure(message string) string { + lower := strings.ToLower(message) + for _, test := range []struct { + kind string + patterns []string + }{ + {"drm", []string{"drm", "digital rights management", "protected by"}}, + {"geo", []string{"geo-restricted", "not available in your country", "not available from your location", "geo restriction", "region restriction", "geographic restriction"}}, + {"network", []string{"network", "connection", "timed out", "temporary failure", "unable to download", "proxy error", "http error 5"}}, + {"unavailable", []string{"media_not_found", "video unavailable", "not available", "no longer available", "http error 404", "episode does not exist"}}, + {"extract", []string{"extractor error", "unable to extract", "unsupported url", "no video formats", "after_move", "output file is missing"}}, + } { + for _, pattern := range test.patterns { + if strings.Contains(lower, pattern) { + return test.kind + } + } + } + return "unknown" +} + +func redact(message string) string { + message = headerSecretRE.ReplaceAllString(message, "$1$2[REDACTED]") + return strings.TrimSpace(paramSecretRE.ReplaceAllString(message, "$1$2[REDACTED]")) +} diff --git a/internal/worker/runner_test.go b/internal/worker/runner_test.go new file mode 100644 index 0000000..4075cc6 --- /dev/null +++ b/internal/worker/runner_test.go @@ -0,0 +1,643 @@ +package worker + +import ( + "context" + "errors" + "io" + "log" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "testing" + "time" + + "gitea.kazu634.com/kazu634/tver/internal/media" + "gitea.kazu634.com/kazu634/tver/internal/store" + "github.com/jackc/pgx/v5" +) + +type fakeStore struct { + mu sync.Mutex + job store.Job + progress []store.Progress + heartbeats int + completed bool + completePath string + completeSize int64 + failedKind, failedMessage string + requeued bool + claimed bool +} + +type queueStore struct { + mu sync.Mutex + jobs []store.Job + claimCounts map[int64]int + active int + maxActive int +} + +func (f *queueStore) ClaimNext(context.Context) (store.Job, error) { + f.mu.Lock() + defer f.mu.Unlock() + for i := range f.jobs { + if f.jobs[i].Status != "queued" { + continue + } + f.jobs[i].Status = "running" + f.claimCounts[f.jobs[i].ID]++ + f.active++ + if f.active > f.maxActive { + f.maxActive = f.active + } + return f.jobs[i], nil + } + return store.Job{}, pgx.ErrNoRows +} + +func (f *queueStore) UpdateProgress(_ context.Context, id int64, _ store.Progress) error { + return f.requireRunning(id) +} + +func (f *queueStore) Heartbeat(_ context.Context, id int64) error { + return f.requireRunning(id) +} + +func (f *queueStore) requireRunning(id int64) error { + f.mu.Lock() + defer f.mu.Unlock() + for i := range f.jobs { + if f.jobs[i].ID == id && f.jobs[i].Status == "running" { + return nil + } + } + return store.ErrInvalidTransition +} + +func (f *queueStore) GetJob(_ context.Context, id int64) (store.Job, error) { + f.mu.Lock() + defer f.mu.Unlock() + for _, job := range f.jobs { + if job.ID == id { + return job, nil + } + } + return store.Job{}, pgx.ErrNoRows +} + +func (f *queueStore) Complete(_ context.Context, id int64, _ string, _ int64) error { + return f.finish(id, "succeeded") +} + +func (f *queueStore) Fail(_ context.Context, id int64, _, _ string) error { + return f.finish(id, "failed") +} + +func (f *queueStore) RequeueRunning(_ context.Context, id int64) error { + return f.finish(id, "queued") +} + +func (f *queueStore) finish(id int64, status string) error { + f.mu.Lock() + defer f.mu.Unlock() + for i := range f.jobs { + if f.jobs[i].ID == id && f.jobs[i].Status == "running" { + f.jobs[i].Status = status + f.active-- + return nil + } + } + return store.ErrInvalidTransition +} + +func (f *queueStore) state() ([]store.Job, map[int64]int, int) { + f.mu.Lock() + defer f.mu.Unlock() + jobs := append([]store.Job(nil), f.jobs...) + counts := make(map[int64]int, len(f.claimCounts)) + for id, count := range f.claimCounts { + counts[id] = count + } + return jobs, counts, f.maxActive +} + +func (f *fakeStore) ClaimNext(context.Context) (store.Job, error) { + f.mu.Lock() + defer f.mu.Unlock() + if f.claimed || f.job.Status != "queued" { + return store.Job{}, pgx.ErrNoRows + } + f.claimed = true + f.job.Status = "running" + return f.job, nil +} + +func (f *fakeStore) UpdateProgress(_ context.Context, _ int64, p store.Progress) error { + f.mu.Lock() + defer f.mu.Unlock() + if f.job.Status != "running" { + return store.ErrInvalidTransition + } + f.progress = append(f.progress, p) + return nil +} + +func (f *fakeStore) Heartbeat(context.Context, int64) error { + f.mu.Lock() + defer f.mu.Unlock() + if f.job.Status != "running" { + return store.ErrInvalidTransition + } + f.heartbeats++ + return nil +} + +func (f *fakeStore) GetJob(context.Context, int64) (store.Job, error) { + f.mu.Lock() + defer f.mu.Unlock() + return f.job, nil +} + +func (f *fakeStore) Complete(_ context.Context, _ int64, path string, size int64) error { + f.mu.Lock() + defer f.mu.Unlock() + if f.job.Status != "running" { + return store.ErrInvalidTransition + } + f.job.Status = "succeeded" + f.completed, f.completePath, f.completeSize = true, path, size + return nil +} + +func (f *fakeStore) Fail(_ context.Context, _ int64, kind, message string) error { + f.mu.Lock() + defer f.mu.Unlock() + if f.job.Status != "running" { + return store.ErrInvalidTransition + } + f.job.Status = "failed" + f.failedKind, f.failedMessage = kind, message + return nil +} + +func (f *fakeStore) RequeueRunning(context.Context, int64) error { + f.mu.Lock() + defer f.mu.Unlock() + if f.job.Status != "running" { + return store.ErrInvalidTransition + } + f.job.Status = "queued" + f.requeued = true + return nil +} + +func (f *fakeStore) setStatus(status string) { + f.mu.Lock() + defer f.mu.Unlock() + f.job.Status = status +} + +func (f *fakeStore) heartbeatCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.heartbeats +} + +func testRunner(t *testing.T, jobs *fakeStore, videoDir string) *Runner { + t.Helper() + ytdlp := filepath.Join("testdata", "fake-yt-dlp.sh") + ffprobe := filepath.Join("testdata", "fake-ffprobe.sh") + for _, path := range []string{ytdlp, ffprobe} { + info, err := os.Stat(path) + if err != nil || info.Mode().Perm()&0o111 == 0 { + t.Fatalf("偽コマンドが実行可能ではありません: %s: %v", path, err) + } + } + runner, err := New(jobs, Config{VideoDir: videoDir, YTDLPPath: ytdlp, FFprobePath: ffprobe, + Concurrency: 1, PollInterval: 5 * time.Millisecond, ProgressInterval: time.Millisecond, + HeartbeatInterval: 10 * time.Millisecond, CancelInterval: 5 * time.Millisecond}, log.New(io.Discard, "", 0)) + if err != nil { + t.Fatal(err) + } + return runner +} + +func runningJob() store.Job { + return store.Job{ID: 7, EpisodeID: "ep123", SeriesTitle: "番組名", Title: "第1話", Status: "running"} +} + +func TestProcessJobDownloadsProbesMovesAndCleans(t *testing.T) { + t.Setenv("FAKE_YTDLP_MODE", "success") + t.Setenv("FAKE_FFPROBE_MODE", "valid") + dir := t.TempDir() + argsFile := filepath.Join(dir, "args") + t.Setenv("FAKE_ARGS_FILE", argsFile) + jobs := &fakeStore{job: runningJob()} + testRunner(t, jobs, dir).ProcessJob(context.Background(), jobs.job) + want := filepath.ToSlash(media.RelativeVideoPath("番組名", "第1話", "ep123")) + if !jobs.completed || jobs.completePath != want || jobs.completeSize == 0 { + t.Fatalf("complete: done=%v path=%q size=%d", jobs.completed, jobs.completePath, jobs.completeSize) + } + if _, err := os.Stat(filepath.Join(dir, filepath.FromSlash(want))); err != nil { + t.Fatalf("保存先: %v", err) + } + if _, err := os.Stat(filepath.Join(dir, ".incoming", "7")); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("作業ディレクトリが残った: %v", err) + } + args, err := os.ReadFile(argsFile) + if err != nil { + t.Fatal(err) + } + argumentText := string(args) + for _, want := range []string{"--newline", "--no-playlist", "--no-simulate", "--quiet", "--progress", "--continue", "--merge-output-format", "bv*[vcodec^=avc1]+ba/b", `download:{"event":"download","format_id":%(info.format_id)j,"progress":%(progress)j}`, `postprocess:{"event":"postprocess","progress":%(progress)j}`, `after_move:{"event":"after_move","filepath":%(filepath)j}`, filepath.Join(dir, ".incoming", "7", "%(id)s.%(ext)s")} { + if !strings.Contains(argumentText, want) { + t.Errorf("yt-dlp 引数に %q がない: %s", want, argumentText) + } + } + if strings.Contains(argumentText, "番組名") || strings.Contains(argumentText, "第1話") { + t.Errorf("出力テンプレートに番組情報が入った: %s", argumentText) + } + phases := make(map[string]bool) + for _, progress := range jobs.progress { + phases[progress.Phase] = true + } + for _, phase := range []string{"video", "audio", "merging", "moving"} { + if !phases[phase] { + t.Errorf("進捗段階 %s がない: %+v", phase, jobs.progress) + } + } + if jobs.heartbeats < 2 { + t.Errorf("heartbeat が少ない: %d", jobs.heartbeats) + } +} + +func TestProcessJobKeepsPercentNamesOutOfYTDLPTemplate(t *testing.T) { + t.Setenv("FAKE_YTDLP_MODE", "success") + t.Setenv("FAKE_FFPROBE_MODE", "valid") + dir := t.TempDir() + argsFile := filepath.Join(dir, "args") + t.Setenv("FAKE_ARGS_FILE", argsFile) + job := runningJob() + job.SeriesTitle = "100%番組" + job.Title = "第%1話" + jobs := &fakeStore{job: job} + testRunner(t, jobs, dir).ProcessJob(context.Background(), job) + if !jobs.completed { + t.Fatalf("%% を含む番組を完了できなかった: %+v", jobs) + } + args, err := os.ReadFile(argsFile) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(args), job.SeriesTitle) || strings.Contains(string(args), job.Title) { + t.Fatalf("番組情報が yt-dlp 引数に入った: %s", args) + } + want := filepath.ToSlash(media.RelativeVideoPath(job.SeriesTitle, job.Title, job.EpisodeID)) + if jobs.completePath != want { + t.Fatalf("保存先=%q want=%q", jobs.completePath, want) + } +} + +func TestProcessJobReplacesExistingDestination(t *testing.T) { + t.Setenv("FAKE_YTDLP_MODE", "success") + t.Setenv("FAKE_FFPROBE_MODE", "valid") + dir := t.TempDir() + job := runningJob() + relative := media.RelativeVideoPath(job.SeriesTitle, job.Title, job.EpisodeID) + destination := filepath.Join(dir, relative) + if err := os.MkdirAll(filepath.Dir(destination), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(destination, []byte("stale file from previous attempt"), 0o644); err != nil { + t.Fatal(err) + } + jobs := &fakeStore{job: job} + testRunner(t, jobs, dir).ProcessJob(context.Background(), job) + got, err := os.ReadFile(destination) + if err != nil { + t.Fatal(err) + } + if !jobs.completed || string(got) != "fake mp4" || jobs.completeSize != int64(len(got)) { + t.Fatalf("既存ファイルの置換: completed=%v content=%q size=%d", jobs.completed, got, jobs.completeSize) + } +} + +func TestProcessJobRejectsAfterMoveSymlinkOutsideWorkDirectory(t *testing.T) { + t.Setenv("FAKE_YTDLP_MODE", "symlink") + t.Setenv("FAKE_FFPROBE_MODE", "valid") + dir := t.TempDir() + outside := filepath.Join(t.TempDir(), "outside.mp4") + if err := os.WriteFile(outside, []byte("outside video"), 0o644); err != nil { + t.Fatal(err) + } + t.Setenv("FAKE_OUTSIDE_FILE", outside) + jobs := &fakeStore{job: runningJob()} + testRunner(t, jobs, dir).ProcessJob(context.Background(), jobs.job) + if jobs.completed || jobs.failedKind != "invalid_output" { + t.Fatalf("作業ディレクトリ外を指すシンボリックリンクを受理した: %+v", jobs) + } + if got, err := os.ReadFile(outside); err != nil || string(got) != "outside video" { + t.Fatalf("作業ディレクトリ外のファイルを変更した: %q %v", got, err) + } +} + +func TestPathInsideSymlinkDerivedCases(t *testing.T) { + root := t.TempDir() + insideDir := filepath.Join(root, "real") + if err := os.Mkdir(insideDir, 0o755); err != nil { + t.Fatal(err) + } + insideFile := filepath.Join(insideDir, "video.mp4") + if err := os.WriteFile(insideFile, []byte("video"), 0o644); err != nil { + t.Fatal(err) + } + outsideDir := t.TempDir() + outsideFile := filepath.Join(outsideDir, "outside.mp4") + if err := os.WriteFile(outsideFile, []byte("outside"), 0o644); err != nil { + t.Fatal(err) + } + + internalFileLink := filepath.Join(root, "internal-file-link.mp4") + internalDirLink := filepath.Join(root, "internal-dir-link") + externalDirLink := filepath.Join(root, "external-dir-link") + rootLink := filepath.Join(t.TempDir(), "root-link") + for link, target := range map[string]string{ + internalFileLink: insideFile, + internalDirLink: insideDir, + externalDirLink: outsideDir, + rootLink: root, + } { + if err := os.Symlink(target, link); err != nil { + t.Fatal(err) + } + } + + for _, tt := range []struct { + name, testRoot, candidate string + wantOK bool + }{ + {"通常ファイル", root, insideFile, true}, + {"作業ディレクトリ内を指すファイルリンク", root, internalFileLink, false}, + {"after_moveがディレクトリ", root, insideDir, false}, + {"作業ディレクトリ内を指すディレクトリリンク配下", root, filepath.Join(internalDirLink, "video.mp4"), true}, + {"作業ディレクトリ外を指すディレクトリリンク配下", root, filepath.Join(externalDirLink, "outside.mp4"), false}, + {"作業ディレクトリ自体がリンク", rootLink, filepath.Join(rootLink, "real", "video.mp4"), true}, + } { + t.Run(tt.name, func(t *testing.T) { + got, err := pathInside(tt.testRoot, tt.candidate) + if tt.wantOK { + want, absErr := filepath.Abs(tt.candidate) + if absErr != nil { + t.Fatal(absErr) + } + if err != nil || got != want { + t.Fatalf("pathInside()=(%q, %v) want=(%q, nil)", got, err, want) + } + return + } + if !errors.Is(err, errInvalidOutput) || got != "" { + t.Fatalf("pathInside()=(%q, %v), invalid_output を期待", got, err) + } + }) + } +} + +func TestProcessJobDestinationDirectoryFailureCleansIncoming(t *testing.T) { + t.Setenv("FAKE_YTDLP_MODE", "success") + t.Setenv("FAKE_FFPROBE_MODE", "valid") + dir := t.TempDir() + job := runningJob() + seriesDir := filepath.Join(dir, filepath.Dir(media.RelativeVideoPath(job.SeriesTitle, job.Title, job.EpisodeID))) + if err := os.WriteFile(seriesDir, []byte("blocks directory"), 0o644); err != nil { + t.Fatal(err) + } + jobs := &fakeStore{job: job} + testRunner(t, jobs, dir).ProcessJob(context.Background(), job) + if jobs.completed || jobs.failedKind != "unknown" { + t.Fatalf("保存先作成失敗: %+v", jobs) + } + if _, err := os.Stat(filepath.Join(dir, ".incoming", "7")); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("作業ディレクトリが残った: %v", err) + } +} + +func TestFailureClassificationAndRedaction(t *testing.T) { + for _, tt := range []struct{ mode, kind string }{ + {"drm", "drm"}, {"geo", "geo"}, {"network", "network"}, + {"unavailable", "unavailable"}, {"extract", "extract"}, {"unknown", "unknown"}, + } { + t.Run(tt.mode, func(t *testing.T) { + t.Setenv("FAKE_YTDLP_MODE", tt.mode) + dir := t.TempDir() + jobs := &fakeStore{job: runningJob()} + testRunner(t, jobs, dir).ProcessJob(context.Background(), jobs.job) + if jobs.failedKind != tt.kind || jobs.completed || strings.Contains(jobs.failedMessage, "secret-value") { + t.Fatalf("failure: kind=%q message=%q completed=%v", jobs.failedKind, jobs.failedMessage, jobs.completed) + } + if _, err := os.Stat(filepath.Join(dir, ".incoming", "7")); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("作業ディレクトリが残った: %v", err) + } + }) + } +} + +func TestInvalidProbeFailsAndCleans(t *testing.T) { + t.Setenv("FAKE_YTDLP_MODE", "success") + for _, mode := range []string{"bad-container", "bad-video", "bad-audio", "non-json", "failure"} { + t.Run(mode, func(t *testing.T) { + t.Setenv("FAKE_FFPROBE_MODE", mode) + dir := t.TempDir() + jobs := &fakeStore{job: runningJob()} + testRunner(t, jobs, dir).ProcessJob(context.Background(), jobs.job) + if jobs.failedKind != "invalid_output" || jobs.completed { + t.Fatalf("ffprobe 不合格: kind=%q complete=%v", jobs.failedKind, jobs.completed) + } + if _, err := os.Stat(filepath.Join(dir, ".incoming", "7")); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("作業ディレクトリが残った: %v", err) + } + }) + } +} + +func TestProgressIsThrottledButHeartbeatContinues(t *testing.T) { + t.Setenv("FAKE_YTDLP_MODE", "burst") + t.Setenv("FAKE_FFPROBE_MODE", "valid") + dir := t.TempDir() + jobs := &fakeStore{job: runningJob()} + runner := testRunner(t, jobs, dir) + runner.config.ProgressInterval = 50 * time.Millisecond + runner.ProcessJob(context.Background(), jobs.job) + downloadUpdates := 0 + for _, progress := range jobs.progress { + if progress.Phase == "video" { + downloadUpdates++ + } + } + if downloadUpdates > 2 || downloadUpdates == 0 { + t.Fatalf("進捗の間引き: %d updates (%+v)", downloadUpdates, jobs.progress) + } + if jobs.heartbeats < 5 { + t.Fatalf("進捗が止まった間の heartbeat: %d", jobs.heartbeats) + } +} + +func TestRunConcurrencyTwoDoesNotDuplicateAndIsolatesFailure(t *testing.T) { + t.Setenv("FAKE_YTDLP_MODE", "mixed") + t.Setenv("FAKE_FFPROBE_MODE", "valid") + dir := t.TempDir() + jobs := &queueStore{ + jobs: []store.Job{ + {ID: 7, EpisodeID: "ep-ok", SeriesTitle: "成功番組", Title: "成功回", Status: "queued"}, + {ID: 8, EpisodeID: "ep-fail", SeriesTitle: "失敗番組", Title: "失敗回", Status: "queued"}, + }, + claimCounts: make(map[int64]int), + } + runner, err := New(jobs, Config{VideoDir: dir, + YTDLPPath: filepath.Join("testdata", "fake-yt-dlp.sh"), FFprobePath: filepath.Join("testdata", "fake-ffprobe.sh"), + Concurrency: 2, PollInterval: 5 * time.Millisecond, ProgressInterval: time.Millisecond, + HeartbeatInterval: 10 * time.Millisecond, CancelInterval: 5 * time.Millisecond}, log.New(io.Discard, "", 0)) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- runner.Run(ctx) }() + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + state, _, _ := jobs.state() + if state[0].Status == "succeeded" && state[1].Status == "failed" { + break + } + time.Sleep(5 * time.Millisecond) + } + cancel() + if err := <-done; err != nil { + t.Fatal(err) + } + state, counts, maxActive := jobs.state() + if state[0].Status != "succeeded" || state[1].Status != "failed" { + t.Fatalf("片方の失敗がもう片方へ波及した: %+v", state) + } + if counts[7] != 1 || counts[8] != 1 { + t.Fatalf("ジョブを二重取得した: %+v", counts) + } + if maxActive != 2 { + t.Fatalf("並行実行数の最大値=%d want=2", maxActive) + } + for _, id := range []int64{7, 8} { + if _, err := os.Stat(filepath.Join(dir, ".incoming", strconv.FormatInt(id, 10))); !errors.Is(err, os.ErrNotExist) { + t.Errorf("ジョブ %d の作業ディレクトリが残った: %v", id, err) + } + } +} + +func TestCancelStopsProcessGroupAndLeavesCanceled(t *testing.T) { + t.Setenv("FAKE_YTDLP_MODE", "hang-drm") + dir := t.TempDir() + ready := filepath.Join(dir, "ready") + terminated := filepath.Join(dir, "terminated") + t.Setenv("FAKE_READY_FILE", ready) + t.Setenv("FAKE_TERM_FILE", terminated) + jobs := &fakeStore{job: runningJob()} + runner := testRunner(t, jobs, dir) + job := jobs.job + done := make(chan struct{}) + go func() { runner.ProcessJob(context.Background(), job); close(done) }() + waitForFile(t, ready) + waitForHeartbeatCount(t, jobs, 2) + jobs.setStatus("canceled") + waitDone(t, done) + if jobs.failedKind != "" || jobs.requeued || jobs.completed || jobs.job.Status != "canceled" { + t.Fatalf("取り消し後: %+v", jobs) + } + waitForFile(t, terminated) + if _, err := os.Stat(filepath.Join(dir, ".incoming", "7")); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("作業ディレクトリが残った: %v", err) + } +} + +func TestShutdownStopsProcessAndRequeues(t *testing.T) { + t.Setenv("FAKE_YTDLP_MODE", "hang") + dir := t.TempDir() + ready := filepath.Join(dir, "ready") + t.Setenv("FAKE_READY_FILE", ready) + jobs := &fakeStore{job: runningJob()} + ctx, cancel := context.WithCancel(context.Background()) + runner := testRunner(t, jobs, dir) + job := jobs.job + done := make(chan struct{}) + go func() { runner.ProcessJob(ctx, job); close(done) }() + waitForFile(t, ready) + cancel() + waitDone(t, done) + if !jobs.requeued || jobs.job.Status != "queued" || jobs.failedKind != "" { + t.Fatalf("終了時の復帰: %+v", jobs) + } + if _, err := os.Stat(filepath.Join(dir, ".incoming", "7")); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("作業ディレクトリが残った: %v", err) + } +} + +func TestRunnerRejectsConcurrencyOutsideRange(t *testing.T) { + for _, concurrency := range []int{-1, 3} { + if _, err := New(&fakeStore{}, Config{Concurrency: concurrency}, nil); err == nil { + t.Errorf("concurrency=%d を許した", concurrency) + } + } +} + +func TestFailureClassificationOrderAndRedaction(t *testing.T) { + for _, tt := range []struct{ message, want string }{ + {"DRM geo-restricted connection video unavailable unable to extract", "drm"}, + {"geo-restricted connection video unavailable unable to extract", "geo"}, + {"connection video unavailable unable to extract", "network"}, + {"video unavailable unable to extract", "unavailable"}, + {"unable to extract", "extract"}, + {"different error", "unknown"}, + } { + if got := classifyFailure(tt.message); got != tt.want { + t.Errorf("classifyFailure(%q)=%q want=%q", tt.message, got, tt.want) + } + } + redacted := redact("Authorization: Bearer bearer-secret\nCookie: session=session-secret\ntoken=token-secret platform_uid=uid-secret") + for _, secret := range []string{"bearer-secret", "session-secret", "token-secret", "uid-secret"} { + if strings.Contains(redacted, secret) { + t.Errorf("秘密が残った: %q", redacted) + } + } +} + +func waitForFile(t *testing.T, path string) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if _, err := os.Stat(path); err == nil { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("ファイル待ちがタイムアウト: %s", path) +} + +func waitDone(t *testing.T, done <-chan struct{}) { + t.Helper() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("worker の終了待ちがタイムアウト") + } +} + +func waitForHeartbeatCount(t *testing.T, jobs *fakeStore, want int) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if jobs.heartbeatCount() >= want { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("heartbeat 待ちがタイムアウト: got=%d want=%d", jobs.heartbeatCount(), want) +} diff --git a/internal/worker/testdata/fake-ffprobe.sh b/internal/worker/testdata/fake-ffprobe.sh new file mode 100755 index 0000000..1c0d276 --- /dev/null +++ b/internal/worker/testdata/fake-ffprobe.sh @@ -0,0 +1,17 @@ +#!/bin/sh +set -eu +case "${FAKE_FFPROBE_MODE:-valid}" in + invalid|bad-container) + echo '{"format":{"format_name":"matroska"},"streams":[{"codec_type":"video","codec_name":"h264"},{"codec_type":"audio","codec_name":"aac"}]}' ;; + bad-video) + echo '{"format":{"format_name":"mp4"},"streams":[{"codec_type":"video","codec_name":"vp9"},{"codec_type":"audio","codec_name":"aac"}]}' ;; + bad-audio) + echo '{"format":{"format_name":"mp4"},"streams":[{"codec_type":"video","codec_name":"h264"},{"codec_type":"audio","codec_name":"opus"}]}' ;; + non-json) + echo 'not JSON' ;; + failure) + echo 'probe failed' >&2 + exit 1 ;; + *) + echo '{"format":{"format_name":"mov,mp4,m4a,3gp,3g2,mj2"},"streams":[{"codec_type":"video","codec_name":"h264"},{"codec_type":"audio","codec_name":"aac"}]}' ;; +esac diff --git a/internal/worker/testdata/fake-yt-dlp.sh b/internal/worker/testdata/fake-yt-dlp.sh new file mode 100755 index 0000000..569c9c1 --- /dev/null +++ b/internal/worker/testdata/fake-yt-dlp.sh @@ -0,0 +1,67 @@ +#!/bin/sh +set -eu + +[ -z "${FAKE_ARGS_FILE:-}" ] || printf '%s\n' "$@" > "$FAKE_ARGS_FILE" + +output_template= +episode_url= +while [ "$#" -gt 0 ]; do + if [ "$1" = "-o" ]; then + output_template=$2 + shift 2 + continue + fi + case "$1" in + https://tver.jp/episodes/*) episode_url=$1 ;; + esac + shift +done + +mode=${FAKE_YTDLP_MODE:-success} +if [ "$mode" = "mixed" ]; then + case "$episode_url" in + */ep-fail) mode=unknown ;; + *) mode=success ;; + esac +fi +case "$mode" in + drm) echo 'ERROR: This video is DRM protected token=secret-value' >&2; exit 1 ;; + geo) echo 'ERROR: This video is not available from your location due to geo restriction' >&2; exit 1 ;; + network) echo 'ERROR: Unable to download JSON metadata: [Errno 61] Connection refused' >&2; exit 1 ;; + unavailable) echo 'ERROR: MEDIA_NOT_FOUND: episode does not exist' >&2; exit 1 ;; + extract) echo 'ERROR: Unable to extract player data' >&2; exit 1 ;; + unknown) echo 'ERROR: surprising failure' >&2; exit 1 ;; + hang|hang-drm) + [ -z "${FAKE_READY_FILE:-}" ] || : > "$FAKE_READY_FILE" + if [ "$mode" = "hang-drm" ]; then + trap 'echo "ERROR: DRM protected" >&2; [ -z "${FAKE_TERM_FILE:-}" ] || : > "$FAKE_TERM_FILE"; exit 143' TERM INT + else + trap '[ -z "${FAKE_TERM_FILE:-}" ] || : > "$FAKE_TERM_FILE"; exit 143' TERM INT + fi + while :; do sleep 1; done + ;; +esac + +output=$(printf '%s' "$output_template" | sed 's/%(id)s/fake/g; s/%(ext)s/mp4/g') +mkdir -p "$(dirname "$output")" +printf 'fake mp4' > "$output" +if [ "$mode" = "symlink" ]; then + rm "$output" + ln -s "$FAKE_OUTSIDE_FILE" "$output" +fi + +if [ "$mode" = "burst" ]; then + i=1 + while [ "$i" -le 20 ]; do + printf '{"event":"download","format_id":"hls-698-1","progress":{"status":"downloading","downloaded_bytes":%s,"total_bytes":100,"speed":10,"eta":2,"fragment_index":%s,"fragment_count":20}}\n' "$i" "$i" + i=$((i + 1)) + done + sleep 0.12 +else + echo '{"event":"download","format_id":"hls-698-1","progress":{"status":"downloading","downloaded_bytes":1024,"fragment_index":0,"fragment_count":50,"filename":"/private/tmp/tver-spike/epq432nee4.fhls-698-1.mp4","tmpfilename":"/private/tmp/tver-spike/epq432nee4.fhls-698-1.mp4.part","max_progress":null,"progress_idx":null,"elapsed":0,"total_bytes_estimate":29798400.0,"speed":4834.271158821972,"eta":null,"_eta_str":"Unknown","_speed_str":" 4.72KiB/s","_percent":0.003436426116838488,"_percent_str":" 0.0%","_total_bytes_str":" N/A","_total_bytes_estimate_str":" 28.42MiB","_downloaded_bytes_str":" 1.00KiB","_elapsed_str":"00:00:00","_default_template":" 0.0% of ~ 28.42MiB at 4.72KiB/s ETA Unknown (frag 0/50)"}}' + sleep 0.02 + echo '{"event":"download","format_id":"hls-ts_AUDIO-0_2","progress":{"status":"finished","downloaded_bytes":50,"total_bytes":50,"speed":20,"eta":0}}' + sleep 0.02 + echo '{"event":"postprocess","progress":{"status":"started","postprocessor":"Merger"}}' >&2 +fi +printf '{"event":"after_move","filepath":"%s"}\n' "$output" -- 2.54.0