Merge pull request '登録番組の定期確認を追加（ステップ6b）' (#10) from subscription-check into main

Reviewed-on: #10

PROJECT.md

@@ -3,7 +3,7 @@ title: tver の現状
 tags:
   - homelab
   - tver
-status: worker のダウンロード本体を実装中
+status: worker の登録番組確認を実装中
 ---
 
 # tver の現状
@@ -14,7 +14,7 @@ status: worker のダウンロード本体を実装中
 ## 今の状態
 
 - Go の HTTP サーバー、TVer API クライアント、ドラマ・バラエティの新着表示、テスト、web・worker コンテナの定義がある
-- **worker のダウンロード本体を実装中。** 待ち行列から yt-dlp を起動し、進捗・heartbeat・取消・出力検査・保存・失敗分類を扱う。登録番組の定期確認はまだ行わない
+- **worker のダウンロード本体と登録番組の定期確認を実装中。** 待ち行列からの取得に加え、有効な登録番組の新しい回を確認して自動登録する
 - リポジトリ: `https://gitea.kazu634.com/kazu634/tver.git`
 
 ## 目的
@@ -41,6 +41,9 @@ TVer（`https://tver.jp/`）で公開中の動画情報を取得し、選んだ
 | 保持期間 | **削除は手動のみ**。自動削除はしない |
 | 待ち行列 | **Postgres のテーブルだけで実装する**。専用のキュー基盤は持たない |
 | 定期確認の間隔 | 登録番組の新しい回を **24 時間ごと**に確認する |
+| 定期確認の走査 | worker のスケジューラが起動直後と1時間ごとに、最終確認から24時間以上経過した有効な番組を探す。番組は1件ずつ確認し、間に数秒空ける |
+| 定期確認の予告 | **180秒以下**の回は自動登録せず、既知としてだけ記録する。新着画面の予告判定と同じしきい値を使う |
+| 定期確認の失敗 | `last_checked_at` を更新せず、失敗日時と秘匿値を伏せた短いメッセージを保存する。次の1時間走査で再試行し、成功時にエラーを消す |
 | 登録時の既存回 | **登録の都度**「公開中の回もダウンロードする」かを選ぶ。選ばない場合も公開中の回を既知として記録する |
 | 番組の一時停止 | 登録は残し、自動ダウンロードだけ止める。再開後は停止中に公開された未確認の回も対象にする |
 | 動画のマウント先 | web と worker は `VIDEO_DIR`（既定 `/videos`）に同じ NAS 共有をマウントする。`download_jobs.file_path` は `VIDEO_DIR` からの相対パスを保存する |
@@ -86,7 +89,9 @@ TVer（`https://tver.jp/`）で公開中の動画情報を取得し、選んだ
 - 現在の `web` は `/healthz`、新着（`/`）、待ち行列（`/queue`）、登録番組（`/subscriptions`）、完了済みライブラリ（`/library`）、再生ページ（`/library/{id}`）、Range 対応の MP4 配信（`/videos/{id}`）を提供する
 - 動画配信と削除はクライアントからジョブ ID だけを受け取り、DB の相対 `file_path` を使う。絶対パス・`..`・`VIDEO_DIR` 外を指すシンボリックリンクを拒否する。削除後は新着から手動で再登録できるが、登録番組の既知記録は残るため自動再登録しない
 - 新着の「番組を登録」から、公開中の回をダウンロードするか選ぶ。web はサーバー側の新着キャッシュから番組 ID と名前を引き、TVer の season ごとの公開中エピソードを取得する。取得失敗時は登録しない
-- 登録と公開中の回の既知記録・任意の待ち行列投入は1トランザクションで行い、登録時刻を最終確認日時に設定する。`known_episodes` は番組解除時に連動して削除する。worker 用の未確認回の抽出と、未確認回だけの既知記録・投入も `internal/store` に用意した。一時停止中は投入を拒否する
+- 登録と公開中の回の既知記録・任意の待ち行列投入は1トランザクションで行い、登録時刻を最終確認日時に設定する。`known_episodes` は番組解除時に連動して削除する。一時停止中は投入を拒否する
+- worker のスケジューラは起動直後と1時間ごとに、`enabled` かつ最終確認から24時間以上の番組を抽出する。TVer から公開中の回を順次取得し、未確認の180秒超の回だけを `source=subscription` で投入する。180秒以下も既知にし、成功時の最終確認更新・エラー消去まで1トランザクションで行う
+- TVer の番組取得に失敗した場合は最終確認日時を変えず、失敗日時と伏字済みの短いメッセージを `subscriptions` に記録して画面に表示する。番組間には3秒空け、SIGTERM では取得中または待機中の確認を中断する
 - 新着の登録ボタンはエピソード ID とジャンルだけを送る。web が有効な新着キャッシュから番組情報を引き、`episode_id` の一意制約で重複登録を防ぐ。新着一覧はジョブ状態を一括取得し、失敗・取り消しには再試行を表示する
 - 待ち行列は実行中・待機中・失敗・取り消し・完了を状態別に直近50件ずつ表示する。段階別の進捗と失敗理由を表示し、待機中の取り消しと失敗・取り消しの再試行ができる。実行中または待機中のジョブがある間だけ5秒ごとに一覧を更新する
 - 実行中のジョブにも取り消しボタンを表示する。worker は DB の `canceled` を定期確認し、yt-dlp をプロセスグループごと停止する
@@ -97,7 +102,7 @@ TVer（`https://tver.jp/`）で公開中の動画情報を取得し、選んだ
 - 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 を外部コマンドとして使い、映像・音声・結合・移動の段階別進捗を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 で実行する
+- worker は起動時にマイグレーションと停止中ジョブの `RequeueStale` を行い、登録番組の確認も直ちに1回行う。SIGTERM では yt-dlp と確認処理を止め、担当中ジョブを `queued` に戻してから終了する。`Dockerfile.worker` は yt-dlp 2026.08.19 を ARG で固定し、ffmpeg・ffprobe とともに非 root で実行する
 - 保存先の NAS 共有フォルダは dockage VM に NFS でマウントし、`worker`（書き込み）と `web`（読み取り・削除）に同じ `VIDEO_DIR` でバインドマウントする
 
 ## 外部とのつながり
@@ -128,7 +133,6 @@ TVer（`https://tver.jp/`）で公開中の動画情報を取得し、選んだ
 - 新着 API が 100 件で打ち切りか、続きを取得できるか [要確認]
 - 新着のうち DRM 付きの作品の割合 [要確認]
 - 非公開 API と yt-dlp の TVer 対応は、TVer 側の変更で予告なく壊れうる。壊れたことを検知する手段が要る
-- 登録番組の24時間ごとの確認は未実装
 - NAS の準備（NFS 共有フォルダ、tver 専用 Postgres 17 コンテナ、DSM ファイアウォール）は DSM での手作業になる。DB の公開ポート（候補 15434）は本番前に実機で確認する [要確認]
 - DRM・地域制限の yt-dlp 失敗時の文言と終了コードは未実測 [要確認]
 
@@ -142,5 +146,6 @@ TVer（`https://tver.jp/`）で公開中の動画情報を取得し、選んだ
 - 新着画面のテストは偽物の TVer クライアント・store と差し替え可能な時計を使い、表示項目、絞り込みの境界、10 分キャッシュ、取得失敗、JST 表示、登録・状態表示・待ち行列の操作を確認する
 - ライブラリのテストは一時ディレクトリの小さな動画を使い、Range の 206・範囲外の 416、相対パスとシンボリックリンクの検査、404、ファイル削除後の記録削除を確認する。共有のファイル名規則は `internal/media` でテストする
 - worker のテストは `internal/worker/testdata` の偽 yt-dlp・ffprobe を使い、本物の TVer へ接続せずに正常完了、失敗分類、取消、終了時の待機復帰、進捗の間引き、heartbeat、出力検査、移動先、作業場所の片付けを確認する
+- 登録番組の定期確認は偽の TVer クライアントと差し替え可能な時計・タイマーを使い、24時間の境界、順次確認、1時間走査、失敗記録、予告の既知化、終了時の中断を確認する。store の抽出・既知化・投入・日時更新・エラー消去は embedded Postgres 17 で確認する
 - `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` で web を起動できる。worker は同じ `DATABASE_URL` と `VIDEO_DIR` を設定して `go run ./cmd/worker` で起動する。worker の実行は TVer への実アクセスを伴うため手動確認は少数回に留める

cmd/worker/main.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"gitea.kazu634.com/kazu634/tver/internal/store"
+	"gitea.kazu634.com/kazu634/tver/internal/tver"
 	workerpkg "gitea.kazu634.com/kazu634/tver/internal/worker"
 	"github.com/jackc/pgx/v5/pgxpool"
 )
@@ -34,14 +35,30 @@ func main() {
 		log.Fatal(err)
 	}
 	defer pool.Close()
-	runner, err := workerpkg.New(store.New(pool), config, log.Default())
+	jobStore := store.New(pool)
+	runner, err := workerpkg.New(jobStore, config, log.Default())
 	if err != nil {
 		log.Fatal(err)
 	}
+	tverClient, err := tver.NewClient("", nil)
+	if err != nil {
+		log.Fatal(err)
+	}
+	scheduler, err := workerpkg.NewSubscriptionScheduler(jobStore, tverClient,
+		workerpkg.SubscriptionSchedulerConfig{PreviewMaxSeconds: tver.PreviewMaxSeconds}, log.Default())
+	if err != nil {
+		log.Fatal(err)
+	}
+	schedulerDone := make(chan struct{})
+	go func() {
+		defer close(schedulerDone)
+		scheduler.Run(ctx)
+	}()
 	log.Printf("worker started (concurrency=%d)", config.Concurrency)
 	if err := runner.Run(ctx); err != nil {
 		log.Fatal(err)
 	}
+	<-schedulerDone
 	log.Print("worker stopped")
 }
 

internal/store/migrations/003_subscription_check_errors.sql

@@ -0,0 +1,6 @@
+ALTER TABLE subscriptions
+    ADD COLUMN last_check_error_at TIMESTAMPTZ,
+    ADD COLUMN last_check_error TEXT;
+
+CREATE INDEX subscriptions_due_idx
+    ON subscriptions (last_checked_at) WHERE enabled;

internal/store/store.go

@@ -59,6 +59,8 @@ type Subscription struct {
 	Enabled               bool
 	CreatedAt             time.Time
 	LastCheckedAt         *time.Time
+	LastCheckErrorAt      *time.Time
+	LastCheckError        *string
 	JobCount              int64
 }
 
@@ -301,6 +303,7 @@ func (s *Store) DeleteSubscription(ctx context.Context, seriesID string) (bool,
 
 func (s *Store) ListSubscriptions(ctx context.Context) ([]Subscription, error) {
 	rows, err := s.pool.Query(ctx, `SELECT s.series_id, s.series_title, s.enabled, s.created_at, s.last_checked_at,
+		s.last_check_error_at, s.last_check_error,
 		(SELECT count(*) FROM download_jobs j WHERE j.series_id=s.series_id)
 		FROM subscriptions s ORDER BY s.created_at DESC, s.series_id`)
 	if err != nil {
@@ -310,7 +313,8 @@ func (s *Store) ListSubscriptions(ctx context.Context) ([]Subscription, error) {
 	var subs []Subscription
 	for rows.Next() {
 		var sub Subscription
-		if err := rows.Scan(&sub.SeriesID, &sub.SeriesTitle, &sub.Enabled, &sub.CreatedAt, &sub.LastCheckedAt, &sub.JobCount); err != nil {
+		if err := rows.Scan(&sub.SeriesID, &sub.SeriesTitle, &sub.Enabled, &sub.CreatedAt, &sub.LastCheckedAt,
+			&sub.LastCheckErrorAt, &sub.LastCheckError, &sub.JobCount); err != nil {
 			return nil, err
 		}
 		subs = append(subs, sub)

internal/store/store_test.go

@@ -113,7 +113,7 @@ func TestMigrate(t *testing.T) {
 	if err := pool.QueryRow(ctx, `SELECT count(*), min(applied_at) FROM schema_migrations`).Scan(&count, &second); err != nil {
 		t.Fatal(err)
 	}
-	if count != 2 || !first.Equal(second) {
+	if count != 3 || !first.Equal(second) {
 		t.Fatalf("再適用で履歴が変わった: %d %v %v", count, first, second)
 	}
 }
@@ -138,7 +138,7 @@ func TestMigrateConcurrent(t *testing.T) {
 	if err := pool.QueryRow(context.Background(), `SELECT count(*) FROM schema_migrations`).Scan(&count); err != nil {
 		t.Fatal(err)
 	}
-	if count != 2 {
+	if count != 3 {
 		t.Fatalf("適用件数: %d", count)
 	}
 }
@@ -166,7 +166,7 @@ func TestMigrateAddsKnownEpisodesToExistingDatabase(t *testing.T) {
 		t.Fatal(err)
 	}
 	var count int
-	if err := pool.QueryRow(ctx, `SELECT count(*) FROM schema_migrations`).Scan(&count); err != nil || count != 2 {
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM schema_migrations`).Scan(&count); err != nil || count != 3 {
 		t.Fatalf("移行履歴: %d %v", count, err)
 	}
 	if _, err := pool.Exec(ctx, `INSERT INTO known_episodes (series_id, episode_id) VALUES ('existing', 'ep1')`); err != nil {
@@ -180,6 +180,47 @@ func TestMigrateAddsKnownEpisodesToExistingDatabase(t *testing.T) {
 	}
 }
 
+func TestMigrateAddsSubscriptionCheckErrorsToVersionTwoDatabase(t *testing.T) {
+	pool := testPool(t, false)
+	ctx := context.Background()
+	for _, name := range []string{"migrations/001_initial.sql", "migrations/002_known_episodes.sql"} {
+		sql, err := migrations.ReadFile(name)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, err := pool.Exec(ctx, string(sql), pgx.QueryExecModeSimpleProtocol); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := pool.Exec(ctx, `CREATE TABLE schema_migrations (version TEXT PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT now())`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := pool.Exec(ctx, `INSERT INTO schema_migrations (version) VALUES ('migrations/001_initial.sql'), ('migrations/002_known_episodes.sql')`); err != nil {
+		t.Fatal(err)
+	}
+	checked := time.Date(2026, 10, 1, 2, 3, 4, 0, time.UTC)
+	if _, err := pool.Exec(ctx, `INSERT INTO subscriptions (series_id, series_title, last_checked_at) VALUES ('existing-v2', '既存番組', $1)`, checked); err != nil {
+		t.Fatal(err)
+	}
+	if err := Migrate(ctx, pool); err != nil {
+		t.Fatal(err)
+	}
+	var title string
+	var last time.Time
+	var errorAt, errorMessage any
+	if err := pool.QueryRow(ctx, `SELECT series_title, last_checked_at, last_check_error_at, last_check_error FROM subscriptions WHERE series_id='existing-v2'`).Scan(&title, &last, &errorAt, &errorMessage); err != nil || title != "既存番組" || !last.Equal(checked) || errorAt != nil || errorMessage != nil {
+		t.Fatalf("v2データ保持と追加列: title=%q last=%v errorAt=%v error=%v err=%v", title, last, errorAt, errorMessage, err)
+	}
+	var indexDef string
+	if err := pool.QueryRow(ctx, `SELECT indexdef FROM pg_indexes WHERE schemaname=current_schema() AND indexname='subscriptions_due_idx'`).Scan(&indexDef); err != nil || !strings.Contains(indexDef, "last_checked_at") || !strings.Contains(indexDef, "WHERE enabled") {
+		t.Fatalf("期限抽出索引: %q %v", indexDef, err)
+	}
+	var versions int
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM schema_migrations`).Scan(&versions); err != nil || versions != 3 {
+		t.Fatalf("移行履歴: %d %v", versions, err)
+	}
+}
+
 func TestDownloadJobIndexes(t *testing.T) {
 	pool := testPool(t, true)
 	rows, err := pool.Query(context.Background(), `SELECT indexname, indexdef FROM pg_indexes WHERE schemaname=current_schema() AND tablename='download_jobs'`)

internal/store/subscriptions.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/jackc/pgx/v5"
 )
@@ -66,6 +67,95 @@ func (s *Store) QueueUnseenEpisodes(ctx context.Context, seriesID string, episod
 	return count, nil
 }
 
+// ListDueSubscriptions は有効で、cutoff 以前に確認された（または未確認の）番組を返す。
+func (s *Store) ListDueSubscriptions(ctx context.Context, cutoff time.Time) ([]Subscription, error) {
+	rows, err := s.pool.Query(ctx, `SELECT series_id, series_title, enabled, created_at, last_checked_at,
+		last_check_error_at, last_check_error, 0::bigint
+		FROM subscriptions
+		WHERE enabled AND (last_checked_at IS NULL OR last_checked_at <= $1)
+		ORDER BY last_checked_at ASC NULLS FIRST, created_at, series_id`, cutoff)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var result []Subscription
+	for rows.Next() {
+		var sub Subscription
+		if err := rows.Scan(&sub.SeriesID, &sub.SeriesTitle, &sub.Enabled, &sub.CreatedAt,
+			&sub.LastCheckedAt, &sub.LastCheckErrorAt, &sub.LastCheckError, &sub.JobCount); err != nil {
+			return nil, err
+		}
+		result = append(result, sub)
+	}
+	return result, rows.Err()
+}
+
+// CompleteSubscriptionCheck は全公開回を既知にし、しきい値を超える未確認回だけを投入する。
+// 最終確認日時の更新と前回エラーの消去も同じ取引で行う。
+func (s *Store) CompleteSubscriptionCheck(ctx context.Context, seriesID string, episodes []Episode, previewMaxSeconds int, checkedAt time.Time) (int, error) {
+	if previewMaxSeconds < 0 {
+		return 0, errors.New("preview threshold must not be negative")
+	}
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return 0, err
+	}
+	defer tx.Rollback(ctx)
+	var enabled bool
+	err = tx.QueryRow(ctx, `SELECT enabled FROM subscriptions WHERE series_id=$1 FOR UPDATE`, seriesID).Scan(&enabled)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return 0, ErrInvalidTransition
+	}
+	if err != nil {
+		return 0, err
+	}
+	if !enabled {
+		return 0, ErrSubscriptionPaused
+	}
+	queued := 0
+	for _, ep := range episodes {
+		if ep.ID == "" || ep.SeriesID != seriesID || ep.DurationSeconds < 0 {
+			return 0, fmt.Errorf("invalid episode identity for series %s", seriesID)
+		}
+		var inserted string
+		err := tx.QueryRow(ctx, `INSERT INTO known_episodes (series_id, episode_id)
+			VALUES ($1,$2) ON CONFLICT DO NOTHING RETURNING episode_id`, seriesID, ep.ID).Scan(&inserted)
+		if errors.Is(err, pgx.ErrNoRows) {
+			continue
+		}
+		if err != nil {
+			return 0, err
+		}
+		if ep.DurationSeconds <= previewMaxSeconds {
+			continue
+		}
+		tag, err := tx.Exec(ctx, `INSERT INTO download_jobs
+			(episode_id, series_id, series_title, title, broadcaster_name, duration_seconds, end_at, thumbnail_path, source)
+			VALUES ($1,$2,$3,$4,$5,$6,$7,$8,'subscription')
+			ON CONFLICT (episode_id) DO NOTHING`, ep.ID, seriesID, ep.SeriesTitle, ep.Title,
+			ep.BroadcasterName, ep.DurationSeconds, ep.EndAt, ep.ThumbnailPath)
+		if err != nil {
+			return 0, err
+		}
+		queued += int(tag.RowsAffected())
+	}
+	if _, err := tx.Exec(ctx, `UPDATE subscriptions SET last_checked_at=$2,
+		last_check_error_at=NULL, last_check_error=NULL WHERE series_id=$1`, seriesID, checkedAt); err != nil {
+		return 0, err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return 0, err
+	}
+	return queued, nil
+}
+
+// RecordSubscriptionCheckFailure は取得失敗を記録し、最終確認日時は変更しない。
+func (s *Store) RecordSubscriptionCheckFailure(ctx context.Context, seriesID string, failedAt time.Time, message string) error {
+	tag, err := s.pool.Exec(ctx, `UPDATE subscriptions SET last_check_error_at=$2, last_check_error=$3
+		WHERE series_id=$1 AND enabled`, seriesID, failedAt, message)
+	return transitionResult(tag.RowsAffected(), err)
+}
+
 func addKnownAndMaybeQueue(ctx context.Context, tx pgx.Tx, seriesID string, episodes []Episode, queue bool) (int, error) {
 	count := 0
 	for _, ep := range episodes {

internal/store/subscriptions_test.go

@@ -5,8 +5,144 @@ import (
 	"errors"
 	"sync"
 	"testing"
+	"time"
 )
 
+func TestListDueSubscriptionsBoundaryAndPaused(t *testing.T) {
+	pool := testPool(t, true)
+	s := New(pool)
+	ctx := context.Background()
+	now := time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC)
+	for _, id := range []string{"null", "old", "boundary", "fresh", "paused"} {
+		if created, err := s.RegisterSubscription(ctx, id, "番組 "+id, nil, false); err != nil || !created {
+			t.Fatalf("登録 %s: %t %v", id, created, err)
+		}
+	}
+	updates := map[string]*time.Time{
+		"null": nil, "old": timePointer(now.Add(-25 * time.Hour)),
+		"boundary": timePointer(now.Add(-24 * time.Hour)), "fresh": timePointer(now.Add(-24*time.Hour + time.Microsecond)),
+		"paused": timePointer(now.Add(-48 * time.Hour)),
+	}
+	for id, checked := range updates {
+		if _, err := pool.Exec(ctx, `UPDATE subscriptions SET last_checked_at=$2 WHERE series_id=$1`, id, checked); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := s.SetSubscriptionEnabled(ctx, "paused", false); err != nil {
+		t.Fatal(err)
+	}
+	due, err := s.ListDueSubscriptions(ctx, now.Add(-24*time.Hour))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(due) != 3 || due[0].SeriesID != "null" || due[1].SeriesID != "old" || due[2].SeriesID != "boundary" {
+		t.Fatalf("24時間境界と一時停止: %+v", due)
+	}
+}
+
+func TestCompleteSubscriptionCheckKnownPreviewErrorAndAtomicity(t *testing.T) {
+	pool := testPool(t, true)
+	s := New(pool)
+	ctx := context.Background()
+	if created, err := s.RegisterSubscription(ctx, "series-check", "番組", []Episode{seriesEpisode("known", "series-check")}, false); err != nil || !created {
+		t.Fatalf("登録: %t %v", created, err)
+	}
+	before := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)
+	failedAt := time.Date(2026, 10, 3, 1, 2, 3, 0, time.UTC)
+	if _, err := pool.Exec(ctx, `UPDATE subscriptions SET last_checked_at=$2 WHERE series_id=$1`, "series-check", before); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.RecordSubscriptionCheckFailure(ctx, "series-check", failedAt, "network token=[REDACTED]"); err != nil {
+		t.Fatal(err)
+	}
+	var unchanged time.Time
+	if err := pool.QueryRow(ctx, `SELECT last_checked_at FROM subscriptions WHERE series_id='series-check'`).Scan(&unchanged); err != nil || !unchanged.Equal(before) {
+		t.Fatalf("失敗時に最終確認を更新: %v %v", unchanged, err)
+	}
+	checkedAt := time.Date(2026, 10, 3, 4, 5, 6, 0, time.UTC)
+	preview := seriesEpisode("preview", "series-check")
+	preview.DurationSeconds = 180
+	normal := seriesEpisode("normal", "series-check")
+	normal.DurationSeconds = 181
+	count, err := s.CompleteSubscriptionCheck(ctx, "series-check", []Episode{
+		seriesEpisode("known", "series-check"), preview, normal,
+	}, 180, checkedAt)
+	if err != nil || count != 1 {
+		t.Fatalf("確認完了: count=%d err=%v", count, err)
+	}
+	unknown, err := s.UnknownEpisodeIDs(ctx, "series-check", []string{"known", "preview", "normal"})
+	if err != nil || len(unknown) != 0 {
+		t.Fatalf("予告を含む既知化: %v %v", unknown, err)
+	}
+	var jobs int
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM download_jobs WHERE series_id='series-check'`).Scan(&jobs); err != nil || jobs != 1 {
+		t.Fatalf("予告の投入除外: %d %v", jobs, err)
+	}
+	var last time.Time
+	var errorAt, errorMessage any
+	if err := pool.QueryRow(ctx, `SELECT last_checked_at, last_check_error_at, last_check_error FROM subscriptions WHERE series_id='series-check'`).Scan(&last, &errorAt, &errorMessage); err != nil || !last.Equal(checkedAt) || errorAt != nil || errorMessage != nil {
+		t.Fatalf("成功時の日時・エラー消去: %v %v %v %v", last, errorAt, errorMessage, err)
+	}
+
+	bad := seriesEpisode("bad", "series-check")
+	bad.DurationSeconds = -1
+	if _, err := s.CompleteSubscriptionCheck(ctx, "series-check", []Episode{seriesEpisode("rollback", "series-check"), bad}, 180, checkedAt.Add(time.Hour)); err == nil {
+		t.Fatal("不正な回を含む確認が成功した")
+	}
+	unknown, err = s.UnknownEpisodeIDs(ctx, "series-check", []string{"rollback"})
+	if err != nil || len(unknown) != 1 {
+		t.Fatalf("失敗時の既知記録を巻き戻さない: %v %v", unknown, err)
+	}
+	if err := pool.QueryRow(ctx, `SELECT last_checked_at FROM subscriptions WHERE series_id='series-check'`).Scan(&last); err != nil || !last.Equal(checkedAt) {
+		t.Fatalf("失敗時の最終確認を巻き戻さない: %v %v", last, err)
+	}
+	if err := s.SetSubscriptionEnabled(ctx, "series-check", false); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.CompleteSubscriptionCheck(ctx, "series-check", nil, 180, checkedAt.Add(time.Hour)); !errors.Is(err, ErrSubscriptionPaused) {
+		t.Fatalf("一時停止中の確認完了: %v", err)
+	}
+	if err := s.RecordSubscriptionCheckFailure(ctx, "series-check", failedAt, "ignored"); !errors.Is(err, ErrInvalidTransition) {
+		t.Fatalf("一時停止中のエラー記録: %v", err)
+	}
+}
+
+func TestCompleteSubscriptionCheckKeepsExistingManualJob(t *testing.T) {
+	pool := testPool(t, true)
+	s := New(pool)
+	ctx := context.Background()
+	if created, err := s.RegisterSubscription(ctx, "series-manual", "番組", nil, false); err != nil || !created {
+		t.Fatalf("登録: %t %v", created, err)
+	}
+	episode := seriesEpisode("manual-existing", "series-manual")
+	manual, _, err := s.Enqueue(ctx, episode, "manual")
+	if err != nil {
+		t.Fatal(err)
+	}
+	checkedAt := time.Date(2026, 10, 3, 4, 5, 6, 0, time.UTC)
+	queued, err := s.CompleteSubscriptionCheck(ctx, "series-manual", []Episode{episode, episode}, 180, checkedAt)
+	if err != nil || queued != 0 {
+		t.Fatalf("既存手動ジョブの確認: queued=%d err=%v", queued, err)
+	}
+	got, err := s.GetJob(ctx, manual.ID)
+	if err != nil || got.Source != "manual" {
+		t.Fatalf("既存手動ジョブを変更した: %+v %v", got, err)
+	}
+	var jobs, known int
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM download_jobs WHERE episode_id=$1`, episode.ID).Scan(&jobs); err != nil || jobs != 1 {
+		t.Fatalf("ジョブが二重化した: %d %v", jobs, err)
+	}
+	if err := pool.QueryRow(ctx, `SELECT count(*) FROM known_episodes WHERE series_id=$1 AND episode_id=$2`, episode.SeriesID, episode.ID).Scan(&known); err != nil || known != 1 {
+		t.Fatalf("既知記録: %d %v", known, err)
+	}
+	var last time.Time
+	if err := pool.QueryRow(ctx, `SELECT last_checked_at FROM subscriptions WHERE series_id=$1`, episode.SeriesID).Scan(&last); err != nil || !last.Equal(checkedAt) {
+		t.Fatalf("最終確認日時: %v %v", last, err)
+	}
+}
+
+func timePointer(value time.Time) *time.Time { return &value }
+
 func seriesEpisode(id, seriesID string) Episode {
 	ep := sampleEpisode(id)
 	ep.SeriesID = seriesID

internal/tver/client.go

@@ -12,7 +12,10 @@ import (
 	"time"
 )
 
-const DefaultBaseURL = "https://platform-api.tver.jp"
+const (
+	DefaultBaseURL    = "https://platform-api.tver.jp"
+	PreviewMaxSeconds = 180
+)
 
 type Session struct {
 	PlatformUID   string `json:"platform_uid"`

internal/web/newer.go

@@ -14,8 +14,7 @@ import (
 )
 
 const (
-	previewMaxSeconds = 180
-	newerCacheTTL     = 10 * time.Minute
+	newerCacheTTL = 10 * time.Minute
 )
 
 var platformTokenPattern = regexp.MustCompile(`(?i)(platform_token=)[^&\s]+`)
@@ -137,7 +136,7 @@ func (s *server) newerView(ctx context.Context, options newerOptions) newerView
 	}
 	visible := make([]tver.Episode, 0, len(episodes))
 	for _, episode := range episodes {
-		if options.hidePreviews && episode.Duration <= previewMaxSeconds {
+		if options.hidePreviews && episode.Duration <= tver.PreviewMaxSeconds {
 			continue
 		}
 		visible = append(visible, episode)

internal/web/subscriptions.go

@@ -15,6 +15,7 @@ type subscriptionActionView struct {
 
 type subscriptionView struct {
 	ID, Title, CreatedAtText, LastCheckedText string
+	CheckErrorAtText, CheckErrorText          string
 	Enabled                                   bool
 	Jobs                                      int64
 }
@@ -136,6 +137,10 @@ func (s *server) loadSubscriptions(ctx context.Context) (subscriptionsView, erro
 		if sub.LastCheckedAt != nil {
 			item.LastCheckedText = sub.LastCheckedAt.In(s.zone).Format("2006-01-02 15:04 MST")
 		}
+		if sub.LastCheckErrorAt != nil && sub.LastCheckError != nil {
+			item.CheckErrorAtText = sub.LastCheckErrorAt.In(s.zone).Format("2006-01-02 15:04 MST")
+			item.CheckErrorText = *sub.LastCheckError
+		}
 		view.Items = append(view.Items, item)
 	}
 	return view, nil

internal/web/subscriptions_test.go

@@ -159,9 +159,12 @@ func TestSubscriptionMutationsWithoutHtmxRedirect(t *testing.T) {
 
 func TestSubscriptionsPagePauseResumeDeleteAndEscaping(t *testing.T) {
 	checked := time.Date(2026, 10, 4, 2, 30, 0, 0, time.UTC)
+	failedAt := time.Date(2026, 10, 4, 3, 45, 0, 0, time.UTC)
+	checkError := `<script>alert("error")</script> token=[REDACTED]`
 	jobs := &fakeJobStore{subscriptions: map[string]store.Subscription{
 		"sr1": {SeriesID: "sr1", SeriesTitle: `<script>alert(1)</script>`, Enabled: true,
-			CreatedAt: time.Date(2026, 10, 3, 0, 0, 0, 0, time.UTC), LastCheckedAt: &checked, JobCount: 3},
+			CreatedAt: time.Date(2026, 10, 3, 0, 0, 0, 0, time.UTC), LastCheckedAt: &checked,
+			LastCheckErrorAt: &failedAt, LastCheckError: &checkError, JobCount: 3},
 		"sr2": {SeriesID: "sr2", SeriesTitle: "別番組", Enabled: false, JobCount: 0},
 	}}
 	handler := newTestHandlerWithStore(&fakeFetcher{}, jobs, nil)
@@ -170,13 +173,13 @@ func TestSubscriptionsPagePauseResumeDeleteAndEscaping(t *testing.T) {
 		t.Fatalf("登録番組ページ: %d", page.Code)
 	}
 	body := page.Body.String()
-	for _, want := range []string{"&lt;script&gt;", "登録日時: 2026-10-03 09:00 JST", "最終確認: 2026-10-04 11:30 JST", "最終確認: 未確認", "関連ジョブ: 3件", "有効", "一時停止", `hx-confirm="番組の登録を解除しますか？"`} {
+	for _, want := range []string{"&lt;script&gt;", "登録日時: 2026-10-03 09:00 JST", "最終確認: 2026-10-04 11:30 JST", "最終確認: 未確認", "最後の確認エラー（2026-10-04 12:45 JST）", `&lt;script&gt;alert(&#34;error&#34;)&lt;/script&gt; token=[REDACTED]`, "関連ジョブ: 3件", "有効", "一時停止", `hx-confirm="番組の登録を解除しますか？"`} {
 		if !strings.Contains(body, want) {
 			t.Errorf("登録番組の情報 %q がない", want)
 		}
 	}
-	if strings.Contains(body, `<script>alert(1)</script>`) {
-		t.Fatal("番組名がエスケープされていない")
+	if strings.Contains(body, `<script>`) {
+		t.Fatal("番組名または確認エラーがエスケープされていない")
 	}
 	paused := postForm(handler, "/subscriptions/sr1/pause", nil)
 	if paused.Code != http.StatusOK || jobs.subscriptions["sr1"].Enabled || !strings.Contains(paused.Body.String(), `action="/subscriptions/sr1/resume"`) || strings.Contains(paused.Body.String(), "<html") {

internal/web/templates/subscriptions.html

@@ -27,6 +27,9 @@
     <article class="subscription-item">
       <h2>{{.Title}}</h2>
       <p>登録日時: {{.CreatedAtText}}／最終確認: {{.LastCheckedText}}／状態: {{if .Enabled}}有効{{else}}一時停止{{end}}／関連ジョブ: {{.Jobs}}件</p>
+      {{if .CheckErrorText}}
+        <details class="error"><summary>最後の確認エラー（{{.CheckErrorAtText}}）</summary><p>{{.CheckErrorText}}</p></details>
+      {{end}}
       {{if .Enabled}}
         <form action="/subscriptions/{{.ID}}/pause" method="post" hx-post="/subscriptions/{{.ID}}/pause" hx-target="#subscriptions-list" hx-swap="outerHTML"><button type="submit">一時停止</button></form>
       {{else}}

internal/worker/subscriptions.go

@@ -0,0 +1,156 @@
+package worker
+
+import (
+	"context"
+	"errors"
+	"log"
+	"strings"
+	"time"
+	"unicode/utf8"
+
+	"gitea.kazu634.com/kazu634/tver/internal/store"
+	"gitea.kazu634.com/kazu634/tver/internal/tver"
+)
+
+type SubscriptionStore interface {
+	ListDueSubscriptions(context.Context, time.Time) ([]store.Subscription, error)
+	CompleteSubscriptionCheck(context.Context, string, []store.Episode, int, time.Time) (int, error)
+	RecordSubscriptionCheckFailure(context.Context, string, time.Time, string) error
+}
+
+type SeriesFetcher interface {
+	FetchSeriesEpisodes(context.Context, string) ([]tver.Episode, error)
+}
+
+type SubscriptionSchedulerConfig struct {
+	ScanInterval      time.Duration
+	DueAfter          time.Duration
+	SeriesInterval    time.Duration
+	PreviewMaxSeconds int
+	Now               func() time.Time
+	After             func(time.Duration) <-chan time.Time
+}
+
+type SubscriptionScheduler struct {
+	store   SubscriptionStore
+	fetcher SeriesFetcher
+	config  SubscriptionSchedulerConfig
+	logger  *log.Logger
+}
+
+func NewSubscriptionScheduler(subscriptions SubscriptionStore, fetcher SeriesFetcher, config SubscriptionSchedulerConfig, logger *log.Logger) (*SubscriptionScheduler, error) {
+	if subscriptions == nil || fetcher == nil {
+		return nil, errors.New("subscription store and fetcher are required")
+	}
+	if config.ScanInterval <= 0 {
+		config.ScanInterval = time.Hour
+	}
+	if config.DueAfter <= 0 {
+		config.DueAfter = 24 * time.Hour
+	}
+	if config.SeriesInterval <= 0 {
+		config.SeriesInterval = 3 * time.Second
+	}
+	if config.PreviewMaxSeconds == 0 {
+		config.PreviewMaxSeconds = tver.PreviewMaxSeconds
+	}
+	if config.PreviewMaxSeconds < 0 {
+		return nil, errors.New("preview threshold must not be negative")
+	}
+	if config.Now == nil {
+		config.Now = time.Now
+	}
+	if config.After == nil {
+		config.After = time.After
+	}
+	if logger == nil {
+		logger = log.Default()
+	}
+	return &SubscriptionScheduler{store: subscriptions, fetcher: fetcher, config: config, logger: logger}, nil
+}
+
+// Run は起動直後に確認し、その後は ScanInterval ごとに繰り返す。
+func (s *SubscriptionScheduler) Run(ctx context.Context) {
+	for {
+		s.checkDue(ctx)
+		if !s.wait(ctx, s.config.ScanInterval) {
+			return
+		}
+	}
+}
+
+func (s *SubscriptionScheduler) checkDue(ctx context.Context) {
+	now := s.config.Now()
+	subscriptions, err := s.store.ListDueSubscriptions(ctx, now.Add(-s.config.DueAfter))
+	if err != nil {
+		if ctx.Err() == nil {
+			s.logger.Printf("確認対象の登録番組を取得できません: %v", err)
+		}
+		return
+	}
+	for index, subscription := range subscriptions {
+		if index > 0 && !s.wait(ctx, s.config.SeriesInterval) {
+			return
+		}
+		episodes, err := s.fetcher.FetchSeriesEpisodes(ctx, subscription.SeriesID)
+		if err != nil {
+			if ctx.Err() != nil {
+				return
+			}
+			message := shortError(err)
+			if recordErr := s.store.RecordSubscriptionCheckFailure(ctx, subscription.SeriesID, s.config.Now(), message); recordErr != nil &&
+				!errors.Is(recordErr, store.ErrInvalidTransition) {
+				s.logger.Printf("登録番組 %s の確認エラーを記録できません: %v", subscription.SeriesID, recordErr)
+			}
+			s.logger.Printf("登録番組 %s の公開中エピソードを取得できません: %s", subscription.SeriesID, message)
+			continue
+		}
+		items := make([]store.Episode, 0, len(episodes))
+		for _, episode := range episodes {
+			items = append(items, subscriptionEpisode(episode))
+		}
+		queued, err := s.store.CompleteSubscriptionCheck(ctx, subscription.SeriesID, items,
+			s.config.PreviewMaxSeconds, s.config.Now())
+		if err != nil {
+			if ctx.Err() != nil {
+				return
+			}
+			if !errors.Is(err, store.ErrSubscriptionPaused) && !errors.Is(err, store.ErrInvalidTransition) {
+				s.logger.Printf("登録番組 %s の確認結果を保存できません: %v", subscription.SeriesID, err)
+			}
+			continue
+		}
+		s.logger.Printf("登録番組 %s を確認しました（新規ジョブ=%d）", subscription.SeriesID, queued)
+	}
+}
+
+func (s *SubscriptionScheduler) wait(ctx context.Context, duration time.Duration) bool {
+	select {
+	case <-ctx.Done():
+		return false
+	case <-s.config.After(duration):
+		return true
+	}
+}
+
+func subscriptionEpisode(episode tver.Episode) store.Episode {
+	item := store.Episode{ID: episode.ID, SeriesID: episode.SeriesID, SeriesTitle: episode.SeriesTitle,
+		Title: episode.Title, BroadcasterName: episode.BroadcasterName,
+		DurationSeconds: int(episode.Duration), ThumbnailPath: episode.ThumbnailPath}
+	if episode.EndAt > 0 {
+		end := time.Unix(episode.EndAt, 0)
+		item.EndAt = &end
+	}
+	return item
+}
+
+func shortError(err error) string {
+	message := redact(err.Error())
+	message = strings.Join(strings.Fields(message), " ")
+	const maxRunes = 512
+	if utf8.RuneCountInString(message) <= maxRunes {
+		return message
+	}
+	runes := []rune(message)
+	return string(runes[:maxRunes])
+}

internal/worker/subscriptions_test.go

@@ -0,0 +1,253 @@
+package worker
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"log"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"gitea.kazu634.com/kazu634/tver/internal/store"
+	"gitea.kazu634.com/kazu634/tver/internal/tver"
+)
+
+type schedulerCompletion struct {
+	seriesID string
+	episodes []store.Episode
+	limit    int
+	at       time.Time
+}
+
+type schedulerFailure struct {
+	seriesID, message string
+	at                time.Time
+}
+
+type fakeSubscriptionStore struct {
+	mu          sync.Mutex
+	due         []store.Subscription
+	cutoffs     []time.Time
+	completions []schedulerCompletion
+	failures    []schedulerFailure
+	listErr     error
+	completeErr error
+}
+
+func (f *fakeSubscriptionStore) ListDueSubscriptions(_ context.Context, cutoff time.Time) ([]store.Subscription, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.cutoffs = append(f.cutoffs, cutoff)
+	return append([]store.Subscription(nil), f.due...), f.listErr
+}
+
+func (f *fakeSubscriptionStore) CompleteSubscriptionCheck(_ context.Context, seriesID string, episodes []store.Episode, limit int, at time.Time) (int, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.completions = append(f.completions, schedulerCompletion{seriesID, append([]store.Episode(nil), episodes...), limit, at})
+	return len(episodes), f.completeErr
+}
+
+func (f *fakeSubscriptionStore) RecordSubscriptionCheckFailure(_ context.Context, seriesID string, at time.Time, message string) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.failures = append(f.failures, schedulerFailure{seriesID, message, at})
+	return nil
+}
+
+type fakeSeriesFetcher struct {
+	mu       sync.Mutex
+	calls    []string
+	episodes map[string][]tver.Episode
+	errors   map[string]error
+	block    bool
+	started  chan string
+}
+
+func (f *fakeSeriesFetcher) FetchSeriesEpisodes(ctx context.Context, seriesID string) ([]tver.Episode, error) {
+	f.mu.Lock()
+	f.calls = append(f.calls, seriesID)
+	block := f.block
+	started := f.started
+	episodes := append([]tver.Episode(nil), f.episodes[seriesID]...)
+	err := f.errors[seriesID]
+	f.mu.Unlock()
+	if started != nil {
+		started <- seriesID
+	}
+	if block {
+		<-ctx.Done()
+		return nil, ctx.Err()
+	}
+	return episodes, err
+}
+
+func TestSubscriptionSchedulerChecksImmediatelyAndMapsEpisodes(t *testing.T) {
+	now := time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC)
+	subscriptions := &fakeSubscriptionStore{due: []store.Subscription{{SeriesID: "one"}, {SeriesID: "two"}}}
+	fetcher := &fakeSeriesFetcher{episodes: map[string][]tver.Episode{
+		"one": {{ID: "ep1", SeriesID: "one", SeriesTitle: "番組", Title: "回", BroadcasterName: "局", Duration: 180, EndAt: 1_800_000_000, ThumbnailPath: "/image"}},
+		"two": {{ID: "ep2", SeriesID: "two", Duration: 181}},
+	}, errors: map[string]error{}}
+	var waits []time.Duration
+	after := func(duration time.Duration) <-chan time.Time {
+		waits = append(waits, duration)
+		ch := make(chan time.Time, 1)
+		ch <- now
+		return ch
+	}
+	scheduler, err := NewSubscriptionScheduler(subscriptions, fetcher, SubscriptionSchedulerConfig{
+		DueAfter: 24 * time.Hour, SeriesInterval: 4 * time.Second,
+		Now: func() time.Time { return now }, After: after,
+	}, log.New(&bytes.Buffer{}, "", 0))
+	if err != nil {
+		t.Fatal(err)
+	}
+	scheduler.checkDue(context.Background())
+	if len(subscriptions.cutoffs) != 1 || !subscriptions.cutoffs[0].Equal(now.Add(-24*time.Hour)) {
+		t.Fatalf("確認対象日時: %v", subscriptions.cutoffs)
+	}
+	if len(fetcher.calls) != 2 || fetcher.calls[0] != "one" || fetcher.calls[1] != "two" || len(waits) != 1 || waits[0] != 4*time.Second {
+		t.Fatalf("順次確認と間隔: calls=%v waits=%v", fetcher.calls, waits)
+	}
+	if len(subscriptions.completions) != 2 || subscriptions.completions[0].limit != tver.PreviewMaxSeconds || !subscriptions.completions[0].at.Equal(now) {
+		t.Fatalf("確認完了: %+v", subscriptions.completions)
+	}
+	got := subscriptions.completions[0].episodes[0]
+	if got.ID != "ep1" || got.DurationSeconds != 180 || got.EndAt == nil || got.EndAt.Unix() != 1_800_000_000 {
+		t.Fatalf("エピソード変換: %+v", got)
+	}
+}
+
+func TestShortErrorLimitsUnicodeByCharacters(t *testing.T) {
+	message := shortError(errors.New(strings.Repeat("界", 260) + "\n\t" + strings.Repeat("語", 300)))
+	if got := len([]rune(message)); got != 512 {
+		t.Fatalf("エラー文字数=%d want=512", got)
+	}
+	if strings.ContainsAny(message, "\r\n\t") || strings.Contains(message, "  ") {
+		t.Fatalf("空白を1行へ正規化していない: %q", message)
+	}
+	if !strings.HasSuffix(message, strings.Repeat("語", 251)) {
+		t.Fatalf("UTF-8文字の途中で切断または境界が不正: %q", message)
+	}
+}
+
+func TestSubscriptionSchedulerRecordsRedactedFailureWithoutCompleting(t *testing.T) {
+	now := time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC)
+	subscriptions := &fakeSubscriptionStore{due: []store.Subscription{{SeriesID: "failed"}}}
+	fetcher := &fakeSeriesFetcher{errors: map[string]error{
+		"failed": errors.New("request failed\nplatform_token=secret-token Cookie: session-secret"),
+	}}
+	var logs bytes.Buffer
+	scheduler, err := NewSubscriptionScheduler(subscriptions, fetcher, SubscriptionSchedulerConfig{Now: func() time.Time { return now }}, log.New(&logs, "", 0))
+	if err != nil {
+		t.Fatal(err)
+	}
+	scheduler.checkDue(context.Background())
+	if len(subscriptions.completions) != 0 || len(subscriptions.failures) != 1 || !subscriptions.failures[0].at.Equal(now) {
+		t.Fatalf("失敗の記録: completions=%v failures=%v", subscriptions.completions, subscriptions.failures)
+	}
+	message := subscriptions.failures[0].message
+	if strings.Contains(message, "secret-token") || strings.Contains(message, "session-secret") || strings.Contains(logs.String(), "secret") || strings.Contains(message, "\n") {
+		t.Fatalf("秘密値または改行が残った: message=%q logs=%q", message, logs.String())
+	}
+}
+
+func TestSubscriptionSchedulerRunStartsImmediatelyThenUsesScanInterval(t *testing.T) {
+	now := time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC)
+	subscriptions := &fakeSubscriptionStore{}
+	fetcher := &fakeSeriesFetcher{}
+	requested := make(chan time.Duration, 2)
+	tick := make(chan time.Time, 1)
+	after := func(duration time.Duration) <-chan time.Time {
+		requested <- duration
+		return tick
+	}
+	scheduler, err := NewSubscriptionScheduler(subscriptions, fetcher, SubscriptionSchedulerConfig{
+		ScanInterval: time.Hour, Now: func() time.Time { return now }, After: after,
+	}, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() { scheduler.Run(ctx); close(done) }()
+	select {
+	case duration := <-requested:
+		if duration != time.Hour {
+			t.Fatalf("走査間隔: %v", duration)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("起動直後の確認が行われなかった")
+	}
+	subscriptions.mu.Lock()
+	initialCalls := len(subscriptions.cutoffs)
+	subscriptions.mu.Unlock()
+	if initialCalls != 1 {
+		t.Fatalf("起動直後の確認回数: %d", initialCalls)
+	}
+	tick <- now
+	select {
+	case <-requested:
+	case <-time.After(time.Second):
+		t.Fatal("1時間後の確認が行われなかった")
+	}
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("終了できなかった")
+	}
+	subscriptions.mu.Lock()
+	defer subscriptions.mu.Unlock()
+	if len(subscriptions.cutoffs) != 2 {
+		t.Fatalf("確認回数: %d", len(subscriptions.cutoffs))
+	}
+}
+
+func TestSubscriptionSchedulerCancellationInterruptsFetchAndSeriesWait(t *testing.T) {
+	now := time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC)
+	t.Run("fetch", func(t *testing.T) {
+		subscriptions := &fakeSubscriptionStore{due: []store.Subscription{{SeriesID: "one"}}}
+		fetcher := &fakeSeriesFetcher{block: true, started: make(chan string, 1)}
+		scheduler, _ := NewSubscriptionScheduler(subscriptions, fetcher, SubscriptionSchedulerConfig{Now: func() time.Time { return now }}, nil)
+		ctx, cancel := context.WithCancel(context.Background())
+		done := make(chan struct{})
+		go func() { scheduler.checkDue(ctx); close(done) }()
+		select {
+		case <-fetcher.started:
+		case <-time.After(time.Second):
+			t.Fatal("取得が開始されなかった")
+		}
+		cancel()
+		select {
+		case <-done:
+		case <-time.After(time.Second):
+			t.Fatal("取得中断で終了しなかった")
+		}
+		if len(subscriptions.failures) != 0 || len(subscriptions.completions) != 0 {
+			t.Fatalf("中断した番組を記録した: failures=%v completions=%v", subscriptions.failures, subscriptions.completions)
+		}
+	})
+
+	t.Run("between series", func(t *testing.T) {
+		subscriptions := &fakeSubscriptionStore{due: []store.Subscription{{SeriesID: "one"}, {SeriesID: "two"}}}
+		fetcher := &fakeSeriesFetcher{episodes: map[string][]tver.Episode{}, errors: map[string]error{}}
+		waiting := make(chan struct{})
+		never := make(chan time.Time)
+		scheduler, _ := NewSubscriptionScheduler(subscriptions, fetcher, SubscriptionSchedulerConfig{
+			Now: func() time.Time { return now }, After: func(time.Duration) <-chan time.Time { close(waiting); return never },
+		}, nil)
+		ctx, cancel := context.WithCancel(context.Background())
+		done := make(chan struct{})
+		go func() { scheduler.checkDue(ctx); close(done) }()
+		<-waiting
+		cancel()
+		<-done
+		if len(fetcher.calls) != 1 || fetcher.calls[0] != "one" {
+			t.Fatalf("番組間の中断: %v", fetcher.calls)
+		}
+	})
+}