fix(reader): the Takeout import holds up under abuse, races and a run it cannot see

- Uploads: one at a time per user and ten an hour, each refused only after its body arrived;
  the history read an entry at a time (32 MiB cap), a Watch later list capped at YouTube's 5000,
  a user's kept videos at 200k; any unreadable shape is a 422, not a 500.
- Quota: Watch later lookups stop at the backfill reserve and at a per-user daily share; what is
  not asked about waits, known videos still go in, and the shorts job retries the rest.
- Staging and settling share one per-user advisory lock (no deadlock); a run settles only the
  rows it read, so a row staged meanwhile is never called gone.
- A failed run's notice names a translated failure, never the upstream's text.
- The status says when an import is queued; the page watches it to its end even when left, then
  reads the feed, its counts and the playlists again.
- Settings: one live region, the zone kept (unavailable) while a run goes, focus handed back,
  a missed drop kept from the browser, an oversize file refused before it is sent.
- Scheduler and run lists show Takeout runs by the importing user.

backend/app/bodylimit.py

@@ -72,9 +72,11 @@ DEFAULT_MAX_BODY_BYTES = 1024 * 1024
 # A route belongs here ONLY IF ITS OWN CAP IS LARGER THAN THE DEFAULT. That is why
 # `POST /api/diagnostics/events` is absent: its `capped_body` limit is 512 KiB, already below the
 # default, so listing it could only widen what it accepts.
-# A Google Takeout file (`reader.takeout`): a real watch history of 9500 views, the 18 months Google
-# keeps by default, is 3.7 MB; four times that is still well inside one request.
-TAKEOUT_MAX_BYTES = 16 * 1024 * 1024
+# A Google Takeout file (`reader.takeout`). A real watch history measured 3.7 MB for 9500 views, ~390
+# bytes a view, so 32 MiB holds ~86k views. YouTube keeps the history 36 months by default on an
+# account made since mid-2020 and for good on an older one, so a heavy viewer's can pass 16 MiB. The
+# file is read an entry at a time (`takeout._json_list`), so it costs its own size, not a tree of it.
+TAKEOUT_MAX_BYTES = 32 * 1024 * 1024
 
 ROUTE_MAX_BODY_BYTES: dict[tuple[str, str], int] = {
     # Ingesting a local file. The body IS the file now — no multipart envelope to allow for —

backend/app/reader/queries.py

@@ -7,6 +7,7 @@ length, whether a video exists); another unit asks the questions at the end of t
 than joining Reader's tables itself.
 """
 
+from collections.abc import Iterable
 from datetime import datetime
 
 from sqlalchemy import ColumnElement, and_, func, or_, select
@@ -407,11 +408,41 @@ def takeout_waiting(db: Session, user: User, kind: TakeoutKind) -> list[tuple[in
     )
 
 
+# Ids per IN list: a list is bound one parameter an id, and a statement takes at most 65,535.
+_IN_CHUNK = 5000
+
+
 def known_video_ids(db: Session, ids: list[str]) -> set[str]:
     """Which of `ids` the catalog holds."""
-    if not ids:
-        return set()
-    return set(db.execute(select(Video.id).where(Video.id.in_(ids))).scalars())
+    known: set[str] = set()
+    for start in range(0, len(ids), _IN_CHUNK):
+        chunk = ids[start : start + _IN_CHUNK]
+        known.update(db.execute(select(Video.id).where(Video.id.in_(chunk))).scalars())
+    return known
+
+
+def takeout_rows_besides(
+    db: Session, user: User, kind: TakeoutKind, video_ids: Iterable[str]
+) -> int:
+    """How many Takeout rows the user keeps other than `video_ids` of `kind`: what an upload of
+    those videos would be added to."""
+    total = (
+        db.scalar(select(func.count(TakeoutItem.id)).where(owned_by(TakeoutItem.user_id, user)))
+        or 0
+    )
+    ids = list(video_ids)
+    for start in range(0, len(ids), _IN_CHUNK):
+        total -= (
+            db.scalar(
+                select(func.count(TakeoutItem.id)).where(
+                    owned_by(TakeoutItem.user_id, user),
+                    TakeoutItem.kind == kind,
+                    TakeoutItem.video_id.in_(ids[start : start + _IN_CHUNK]),
+                )
+            )
+            or 0
+        )
+    return total
 
 
 def takeout_outcomes(db: Session, user: User) -> dict[tuple[str, str | None], int]:
@@ -449,3 +480,14 @@ def takeout_users_ready(db: Session) -> list[int]:
             .distinct()
         ).scalars()
     )
+
+
+def takeout_users_watch_later_waiting(db: Session) -> list[int]:
+    """Every user with a Watch later row a run could not put in yet."""
+    return list(
+        db.execute(
+            select(TakeoutItem.user_id)
+            .where(TakeoutItem.kind == TakeoutKind.WATCH_LATER, TakeoutItem.outcome.is_(None))
+            .distinct()
+        ).scalars()
+    )

backend/app/reader/service.py

@@ -510,6 +510,15 @@ def detach_user_tag(db: Session, user: User, channel_id: str, tag_id: int) -> No
 # --- Google Takeout --------------------------------------------------------------------------
 
 _TAKEOUT_BATCH = 1000
+# The advisory-lock namespace of one user's Takeout rows ("TO"), so a key cannot meet another lock's.
+_TAKEOUT_LOCK_NS = 0x544F
+
+
+def lock_takeout(db: Session, user: User) -> None:
+    """Hold the user's Takeout rows for the rest of the transaction. An upload's staging and an
+    import's settling each lock many of the same rows, one statement after another and in no common
+    order, so without one lock over both the two can deadlock."""
+    db.execute(select(func.pg_advisory_xact_lock(_TAKEOUT_LOCK_NS, user.id)))
 
 
 def stage_takeout(
@@ -518,6 +527,7 @@ def stage_takeout(
     """Keep the videos of one Takeout file, `{video id: (time, channel id)}`, for the import job.
     A second upload of the same export, or a newer one, adds what is new: a video already kept keeps
     the later time, and a row the import has settled keeps its outcome."""
+    lock_takeout(db, user)
     values = [
         {"user_id": user.id, "kind": kind, "video_id": vid, "at": at, "channel_id": channel}
         for vid, (at, channel) in rows.items()
@@ -551,6 +561,7 @@ def apply_takeout_watched(db: Session, user: User) -> dict[str, int]:
     past is not watching it), and a state the user set here wins: watched or hidden, or anything
     changed after the Takeout time. The upsert's own condition is that rule, so a state set while
     the import runs wins too."""
+    lock_takeout(db, user)
     short: list[int] = []
     todo: dict[str, tuple[int, datetime]] = {}
     for row_id, video_id, at, is_short in queries.takeout_watched_ready(db, user):
@@ -590,19 +601,22 @@ def apply_takeout_watched(db: Session, user: User) -> dict[str, int]:
     return {"watched": len(marked), "kept": len(kept), "shorts": len(short)}
 
 
-def apply_takeout_watch_later(db: Session, user: User) -> dict[str, int]:
-    """Put the videos of the user's YouTube Watch later list into theirs here, oldest first, after
-    what it already holds. Called once the catalog has fetched what it could: a video still unknown
-    is one YouTube no longer has."""
-    waiting = queries.takeout_waiting(db, user, TakeoutKind.WATCH_LATER)
-    if not waiting:
+def apply_takeout_watch_later(
+    db: Session, user: User, rows: list[tuple[int, str]], asked: set[str]
+) -> dict[str, int]:
+    """Put the videos of the user's waiting Watch later `rows`, (row id, video id) oldest first, into
+    their list here, after what it already holds. `asked` is the videos YouTube was just asked for: one
+    of those the catalog still lacks is gone from YouTube; any other unknown one keeps waiting."""
+    if not rows:
         return {}
-    known = queries.known_video_ids(db, [vid for _, vid in waiting])
+    lock_takeout(db, user)
+    known = queries.known_video_ids(db, [vid for _, vid in rows])
     pl = watch_later(db, user)
     done: dict[TakeoutOutcome, list[int]] = {o: [] for o in TakeoutOutcome}
-    for row_id, vid in waiting:
+    for row_id, vid in rows:
         if vid not in known:
-            done[TakeoutOutcome.MISSING].append(row_id)
+            if vid in asked:
+                done[TakeoutOutcome.MISSING].append(row_id)
         elif add_item(db, pl, vid, mark_dirty=False):
             done[TakeoutOutcome.ADDED].append(row_id)
         else:

backend/app/reader/takeout.py

@@ -24,6 +24,7 @@ from collections.abc import Iterator
 from contextlib import contextmanager
 from dataclasses import dataclass, field
 from datetime import datetime
+from itertools import chain
 
 from sqlalchemy.orm import Session
 
@@ -31,6 +32,7 @@ from app import quota
 from app.bodylimit import TAKEOUT_MAX_BYTES
 from app.db import SessionLocal
 from app.models import User
+from app.ratelimit import RateLimiter
 from app.reader import queries, service
 from app.reader.models import TakeoutKind
 from app.sync.runner import get_service_user
@@ -39,6 +41,12 @@ from app.youtube.client import YouTubeClient
 
 # The largest file the upload takes (its edge cap and the route's own are one number).
 MAX_BYTES = TAKEOUT_MAX_BYTES
+# The most videos a Watch later file may hold: YouTube's own cap on one playlist. Each unknown one is
+# looked up on the shared API budget, so a longer list is not one YouTube made.
+WATCH_LATER_MAX = 5000
+# The most Takeout videos one user keeps: twenty times the real export measured here (~9,000 videos
+# across both files), so it only stops uploads of made-up ones piling up for good.
+ROWS_MAX = 200_000
 
 _VIDEO_ID = re.compile(r"^[A-Za-z0-9_-]{11}$")
 _WATCH_URL = re.compile(
@@ -84,7 +92,7 @@ def _keep(parsed: Parsed, vid: str, at: datetime, channel: str | None) -> None:
         parsed.videos[vid] = (at, channel or (old[1] if old else None))
 
 
-def _history(data: list) -> Parsed:
+def _history(data: Iterator[object]) -> Parsed:
     parsed = Parsed(TakeoutKind.WATCHED)
     for entry in data:
         parsed.entries += 1
@@ -117,22 +125,54 @@ def _history(data: list) -> Parsed:
 
 def _watch_later(text: str) -> Parsed:
     parsed = Parsed(TakeoutKind.WATCH_LATER)
-    rows = [row for row in csv.reader(io.StringIO(text)) if any(cell.strip() for cell in row)]
-    if not rows or len(rows[0]) < 2:
+    rows = (row for row in csv.reader(io.StringIO(text)) if any(cell.strip() for cell in row))
+    first = next(rows, None)
+    if first is None or len(first) < 2:
         raise TakeoutError("not_takeout")
-    if not _VIDEO_ID.match(rows[0][0].strip()):
-        rows = rows[1:]  # the header, in the account's language
-    for row in rows:
+    # The first row is the header, in the account's language, unless it is already a video.
+    for row in chain([first], rows) if _VIDEO_ID.match(first[0].strip()) else rows:
         parsed.entries += 1
-        vid = row[0].strip() if row else ""
+        vid = row[0].strip()
         at = _time(row[1].strip()) if len(row) > 1 else None
         if not _VIDEO_ID.match(vid) or at is None:
             parsed.malformed += 1
             continue
         _keep(parsed, vid, at, None)
+        if len(parsed.videos) > WATCH_LATER_MAX:
+            raise TakeoutError("too_many")
     return parsed
 
 
+def _skip(text: str, at: int) -> int:
+    """The first index at or after `at` that is not JSON whitespace."""
+    while at < len(text) and text[at] in " \t\n\r":
+        at += 1
+    return at
+
+
+def _json_list(text: str) -> Iterator[object]:
+    """The entries of a JSON list, one at a time: the whole tree is never built, so a file of a
+    million tiny entries costs its text and one entry, not a million dicts at once."""
+    decoder = json.JSONDecoder()
+    at = _skip(text, _skip(text, 0) + 1)  # past the "[" the caller saw
+    if text.startswith("]", at):
+        at += 1
+    else:
+        while True:
+            entry, at = decoder.raw_decode(text, at)
+            yield entry
+            at = _skip(text, at)
+            if text.startswith(",", at):
+                at = _skip(text, at + 1)
+            elif text.startswith("]", at):
+                at += 1
+                break
+            else:
+                raise ValueError("not a JSON list")
+    if _skip(text, at) != len(text):
+        raise ValueError("text after the JSON list")
+
+
 def parse(body: bytes) -> Parsed:
     """Read an uploaded Takeout file: the watch history (JSON) or the Watch later list (csv), told
     apart by their content, not their name. Raises `TakeoutError` for anything else, and for a file
@@ -142,19 +182,19 @@ def parse(body: bytes) -> Parsed:
     except UnicodeDecodeError:
         raise TakeoutError("not_takeout") from None
     stripped = text.lstrip()
-    if stripped.startswith("["):
-        try:
-            data = json.loads(stripped)
-        except json.JSONDecodeError:
-            raise TakeoutError("not_takeout") from None
-        if not isinstance(data, list):
-            raise TakeoutError("not_takeout")
-        parsed = _history(data)
-    elif stripped.startswith(("{", "<")):
+    if stripped.startswith(("{", "<")):
         # Another JSON file of the export, or the history's HTML variant.
         raise TakeoutError("not_takeout")
-    else:
-        parsed = _watch_later(stripped)
+    try:
+        parsed = (
+            _history(_json_list(stripped)) if stripped.startswith("[") else _watch_later(stripped)
+        )
+    # A file the import cannot read, however it fails: bad JSON (a ValueError, as is a number too
+    # long to convert), nesting deeper than the decoder recurses, a csv field past csv's limit.
+    except (ValueError, RecursionError, csv.Error) as exc:
+        if isinstance(exc, TakeoutError):
+            raise
+        raise TakeoutError("not_takeout") from None
     if not parsed.videos:
         raise TakeoutError("no_videos")
     return parsed
@@ -181,6 +221,61 @@ def run_target(user_id: int) -> str:
     return f"user:{user_id}"
 
 
+def target_title(db: Session, target: str) -> str | None:
+    """Who an import run's target is, for the admin's run lists: the user's name, else their email."""
+    _, _, raw = target.partition("user:")
+    user = db.get(User, int(raw)) if raw.isdigit() else None
+    return (user.display_name or user.email) if user is not None else None
+
+
+class TooMany(Exception):
+    """The upload would take the user past `ROWS_MAX` kept videos."""
+
+
+def stage(db: Session, user: User, parsed: Parsed) -> None:
+    """Keep an uploaded file's videos for the import, unless that would take the user past
+    `ROWS_MAX` (`TooMany`). The caller commits."""
+    service.lock_takeout(db, user)  # the count and the insert are one user's, not two uploads' each
+    if (
+        queries.takeout_rows_besides(db, user, parsed.kind, parsed.videos) + len(parsed.videos)
+        > ROWS_MAX
+    ):
+        raise TooMany
+    service.stage_takeout(db, user, parsed.kind, parsed.videos)
+
+
+# How many unknown Watch later videos one user may have looked up on the shared API budget in a day:
+# two full Watch later lists (two units a hundred ids). The rest wait for the next day's runs.
+_lookups = RateLimiter(max_events=2 * WATCH_LATER_MAX, window_seconds=24 * 3600)
+
+
+def _put_in_watch_later(db: Session, user: User) -> dict:
+    """Put the user's waiting Watch later videos into theirs here: the ones the catalog knows at once,
+    the others once YouTube was asked for them — when a YouTube-connected account can ask, today's
+    budget is above the backfill reserve, and the user's own daily share of lookups allows it. A video
+    not asked about yet waits for a later run. The lookup commits what it fetched; the caller commits
+    the rest."""
+    waiting = queries.takeout_waiting(db, user, TakeoutKind.WATCH_LATER)
+    if not waiting:
+        return {}
+    ids = [vid for _, vid in waiting]
+    known = queries.known_video_ids(db, ids)
+    ask = [vid for vid in ids if vid not in known][:WATCH_LATER_MAX]
+    asked: set[str] = set()
+    fetcher = get_service_user(db) if ask else None
+    if (
+        fetcher is not None
+        and not quota.backfill_reserve_reached()
+        and _lookups.allow(str(user.id), cost=len(ask))
+    ):
+        with quota.attribute(user.id, quota.QuotaAction.VIDEOS_LOOKUP):
+            with YouTubeClient(db, fetcher) as yt:
+                ensure_videos(db, yt, ask)
+        asked = set(ask)
+    # Only the rows read above: one a second upload staged meanwhile was never asked about.
+    return service.apply_takeout_watch_later(db, user, waiting, asked)
+
+
 def run_import(user_id: int) -> dict:
     """The import job: apply what the user's kept Takeout rows can say now, as one recorded run.
     The watched part commits before the Watch later part asks YouTube for the videos the catalog
@@ -191,15 +286,8 @@ def run_import(user_id: int) -> dict:
             return {"skipped": "no user"}
         result = service.apply_takeout_watched(db, user)
         db.commit()
-        waiting = queries.takeout_waiting(db, user, TakeoutKind.WATCH_LATER)
-        if waiting:
-            fetcher = get_service_user(db)
-            if fetcher is not None:
-                with quota.attribute(user_id, quota.QuotaAction.VIDEOS_LOOKUP):
-                    with YouTubeClient(db, fetcher) as yt:
-                        ensure_videos(db, yt, [vid for _, vid in waiting])
-                result |= service.apply_takeout_watch_later(db, user)
-                db.commit()
+        result |= _put_in_watch_later(db, user)
+        db.commit()
         outcomes = queries.takeout_outcomes(db, user)
         result["waiting"] = outcomes.get((TakeoutKind.WATCHED, None), 0)
         if outcomes.get((TakeoutKind.WATCH_LATER, None)):
@@ -208,17 +296,28 @@ def run_import(user_id: int) -> dict:
 
 
 def apply_waiting(db: Session) -> dict:
-    """Mark watched the waiting Takeout rows whose video the catalog has come to know and the Shorts
-    probe has cleared since the import — every user's, one commit each. Run after each Shorts probe
-    (the scheduler's `shorts` job), the only place a video's Short-ness becomes known. Returns its
-    counts, or an empty dict when nothing was waiting on a probed video."""
-    users = queries.takeout_users_ready(db)
-    marked = 0
-    for user_id in users:
+    """Settle what every user's import left waiting and can be settled now, one commit each: a
+    watched video the catalog has come to know and the Shorts probe has cleared (by its job or a
+    search's), and the Watch later videos a run could not put in yet. Run after each run of the
+    `shorts` job, which is what clears the most of them. Returns its counts, or an empty dict when
+    nothing was waiting that could move."""
+    marked = added = 0
+    ready = set(queries.takeout_users_ready(db))
+    later = set(queries.takeout_users_watch_later_waiting(db))
+    for user_id in sorted(ready | later):
         user = db.get(User, user_id)
         if user is None:
             continue
         with _one_import(user_id):
-            marked += service.apply_takeout_watched(db, user)["watched"]
-            db.commit()
-    return {"takeout_watched": marked} if users else {}
+            if user_id in ready:
+                marked += service.apply_takeout_watched(db, user)["watched"]
+                db.commit()
+            if user_id in later:
+                added += _put_in_watch_later(db, user).get("watch_later_added", 0)
+                db.commit()
+    out = {}
+    if ready:
+        out["takeout_watched"] = marked
+    if added:
+        out["takeout_watch_later_added"] = added
+    return out

backend/app/routes/_common.py

@@ -124,6 +124,14 @@ async def capped_body(request: Request, limit: int) -> bytes:
     return bytes(body)
 
 
+async def drain_body(request: Request) -> None:
+    """Read the rest of the body to nowhere, before a refusal is raised: answered mid-upload, the
+    client reports a network failure, not the refusal (`capped_stream_to` says why). `bodylimit`
+    bounds how much there can be."""
+    async for _ in request.stream():
+        pass
+
+
 class ByteSink(Protocol):
     """Where `capped_stream_to` writes: a file, or anything that takes the bytes on their way to one."""
 

backend/app/routes/feed.py

@@ -1,3 +1,5 @@
+import threading
+from collections.abc import AsyncIterator
 from datetime import date
 from functools import partial
 
@@ -10,12 +12,14 @@ from app import jobruns, quota, scheduler
 from app.auth import current_user
 from app.db import get_db, release_read_connection
 from app.mailtext import lang_for_user
+from app.ratelimit import RateLimiter
 from app.feedquery import BadCursor, facet_counts, feed_count, feed_page, feed_summary, feed_years
 from app.models import User
 from app.reader import queries, service, takeout
 from app.routes._common import (
     ProgressIn,
     capped_body,
+    drain_body,
     kept_position,
     not_found,
     youtube_refusal,
@@ -305,17 +309,44 @@ def get_video_detail(
 # --- Google Takeout ------------------------------------------------------------------------------
 
 
-async def _takeout_body(request: Request, db: Session = Depends(get_db)) -> bytes:
+# One upload at a time per user, and ten an hour: a file is held whole while it is read, so these
+# bound what one account can make the process hold. The section uploads its files one after another.
+_takeout_limiter = RateLimiter(max_events=10, window_seconds=3600)
+_takeout_uploading: set[int] = set()
+_takeout_uploading_guard = threading.Lock()
+
+
+async def _takeout_body(
+    request: Request, user: User = Depends(current_user), db: Session = Depends(get_db)
+) -> AsyncIterator[bytes]:
     # The sign-in check has run, and it was a SELECT: hand the connection back before a file of
     # megabytes arrives, so a slow upload does not hold one `idle in transaction`.
     release_read_connection(db)
-    return await capped_body(request, takeout.MAX_BYTES)
+    with _takeout_uploading_guard:
+        busy = user.id in _takeout_uploading
+        if not busy:
+            _takeout_uploading.add(user.id)
+    if busy or not _takeout_limiter.allow(str(user.id)):
+        if not busy:
+            with _takeout_uploading_guard:
+                _takeout_uploading.discard(user.id)
+        # Read to the end before answering: refused mid-upload, the browser reports a network
+        # failure instead of the refusal (capped_stream_to says why).
+        await drain_body(request)
+        if busy:
+            raise refused(429, "takeout_busy")
+        raise refused(429, "takeout_cooldown")
+    try:
+        yield await capped_body(request, takeout.MAX_BYTES)
+    finally:
+        with _takeout_uploading_guard:
+            _takeout_uploading.discard(user.id)
 
 
 @router.post("/takeout")
 def upload_takeout(
-    # The sign-in check FIRST: dependencies resolve in order, so a stranger's file is refused before
-    # a byte of it is read.
+    # The sign-in check FIRST: the body's own dependency needs the signed-in user, so a stranger's
+    # file is refused before a byte of it is read.
     user: User = Depends(current_user),
     body: bytes = Depends(_takeout_body),
     db: Session = Depends(get_db),
@@ -327,8 +358,14 @@ def upload_takeout(
     except takeout.TakeoutError as exc:
         if exc.reason == "no_videos":
             raise refused(422, "takeout_no_videos") from None
+        if exc.reason == "too_many":
+            raise refused(422, "takeout_list_too_long", limit=takeout.WATCH_LATER_MAX) from None
         raise refused(422, "takeout_unreadable") from None
-    service.stage_takeout(db, user, parsed.kind, parsed.videos)
+    try:
+        takeout.stage(db, user, parsed)
+    except takeout.TooMany:
+        db.rollback()
+        raise refused(422, "takeout_too_many", limit=takeout.ROWS_MAX) from None
     db.commit()
     scheduler.run_ad_hoc(
         scheduler.TAKEOUT_IMPORT,
@@ -353,8 +390,13 @@ def takeout_status(user: User = Depends(current_user), db: Session = Depends(get
     """What the user's Takeout import has done: its last run (none before the first upload), and
     every video kept from the export, by kind and outcome — a video the catalog did not know yet
     counts as "waiting" until it does, and is marked then."""
-    run = jobruns.last_run(db, scheduler.TAKEOUT_IMPORT, takeout.run_target(user.id))
+    target = takeout.run_target(user.id)
+    run = jobruns.last_run(db, scheduler.TAKEOUT_IMPORT, target)
     return {
+        # Queued and not started: its run row is not written yet, and the last run is still the
+        # previous one, so the page waits for it rather than taking that one for this upload's.
+        "queued": scheduler.ad_hoc_queued(scheduler.TAKEOUT_IMPORT, target),
+        "max_bytes": takeout.MAX_BYTES,
         "run": None
         if run is None
         else {

backend/app/routes/runs.py

@@ -3,7 +3,7 @@
 Admin-only and read-only — rows leave the table through the retention job
 (`job_run_retention_days`). Filtered on the server and paged by `_keyset`, like the server log and
 the audit log beside it on the activity page. A row names what a reader needs without a second
-request: the channel a user's ad-hoc run worked on, and how many log lines the run left (only
+request: what a user's ad-hoc run worked on (the channel, the importing user), and how many log lines the run left (only
 WARNING and up are stored, so most runs have none — the count says so before the click).
 """
 
@@ -116,8 +116,12 @@ def list_runs(
         limit,
         lambda r: r.started_at,
     )
-    channels = {r.target for r in runs if r.job in scheduler.AD_HOC_JOBS and r.target}
+    channels = {r.target for r in runs if r.job == scheduler.CHANNEL_BACKFILL and r.target}
     titles = catalog_queries.channel_titles(db, channels)
+    for target in {r.target for r in runs if r.job == scheduler.TAKEOUT_IMPORT and r.target}:
+        title = scheduler.ad_hoc_target_title(db, scheduler.TAKEOUT_IMPORT, target)
+        if title is not None:
+            titles[target] = title
     lines: dict[int, int] = {}
     if runs:
         counted = db.execute(

backend/app/scheduler.py

@@ -274,7 +274,6 @@ def scheduler_snapshot(db) -> dict:
     open_runs = _open_run_ids()
     recent = jobruns.recent_runs(db, [*JOB_INTERVALS, *AD_HOC_JOBS], RECENT_RUNS)
     open_runs |= _open_run_ids()
-    going_id = max((r for r, j in open_runs.items() if j == CHANNEL_BACKFILL), default=None)
     jobs = []
     for job_id, default_interval in JOB_INTERVALS.items():
         a = acts.get(job_id, {})
@@ -290,30 +289,11 @@ def scheduler_snapshot(db) -> dict:
         )
     # Work a user starts on one target: no interval, no "run now" (it needs its target), but its
     # runs are shown beside the jobs' — it spends the same quota. Its one in-memory guard is the
-    # target's own (sync/runner), so "running" is the runs this process has open: two channels can
-    # backfill at once, so a run still going counts even when a newer one already ended. That run then
-    # leads the row, so its channel, its "no outcome yet" and its dot are one run's, never the
-    # newer run's outcome under the older one's channel.
-    backfills = recent.get(CHANNEL_BACKFILL, [])
-    going = db.get(JobRun, going_id) if going_id is not None else None
-    if going is not None and going.status != RunStatus.RUNNING:
-        going = None  # its outcome is written; only the bookkeeping has not caught up
-    if going is not None:
-        backfills = [going, *(r for r in backfills if r.id != going.id)][:RECENT_RUNS]
-    shown = backfills[0] if backfills else None
-    ad_hoc = [
-        {
-            "id": CHANNEL_BACKFILL,
-            "running": going is not None,
-            "target": shown.target if shown else None,
-            "target_title": (
-                catalog_queries.channel_title(db, shown.target)
-                if shown is not None and shown.target
-                else None
-            ),
-            **_last_runs(backfills, open_runs),
-        }
-    ]
+    # target's own (sync/runner, the import's per-user lock), so "running" is the runs this process
+    # has open: two targets can run at once, so a run still going counts even when a newer one already
+    # ended. That run then leads the row, so its target, its "no outcome yet" and its dot are one
+    # run's, never the newer run's outcome under the older one's target.
+    ad_hoc = [_ad_hoc_row(db, job, recent.get(job, []), open_runs) for job in AD_HOC_JOBS]
     return {
         "running_here": running_here,
         "enabled": settings.scheduler_enabled,
@@ -335,6 +315,27 @@ def _seen(run, open_runs: dict[int, str]) -> RunStatus:
     return RunStatus(run.status)
 
 
+def _ad_hoc_row(db, job: str, runs: list[JobRun], open_runs: dict[int, str]) -> dict:
+    going_id = max((r for r, j in open_runs.items() if j == job), default=None)
+    going = db.get(JobRun, going_id) if going_id is not None else None
+    if going is not None and going.status != RunStatus.RUNNING:
+        going = None  # its outcome is written; only the bookkeeping has not caught up
+    if going is not None:
+        runs = [going, *(r for r in runs if r.id != going.id)][:RECENT_RUNS]
+    shown = runs[0] if runs else None
+    return {
+        "id": job,
+        "running": going is not None,
+        "target": shown.target if shown else None,
+        "target_title": (
+            ad_hoc_target_title(db, job, shown.target)
+            if shown is not None and shown.target
+            else None
+        ),
+        **_last_runs(runs, open_runs),
+    }
+
+
 def _last_runs(runs: list, open_runs: dict[int, str]) -> dict:
     """A row's run fields from its newest runs (newest first): how the last one went, and the few
     before it for the history dots."""
@@ -619,6 +620,29 @@ CHANNEL_BACKFILL = "channel_backfill"
 TAKEOUT_IMPORT = "takeout_import"
 # Every ad-hoc job id: the page names each one (a frontend contract test reads this tuple).
 AD_HOC_JOBS = (CHANNEL_BACKFILL, TAKEOUT_IMPORT)
+# What each one's target is called on the admin's pages: a channel's title, an importing user's name.
+_TARGET_TITLES: dict[str, Callable[..., str | None]] = {
+    CHANNEL_BACKFILL: catalog_queries.channel_title,
+    TAKEOUT_IMPORT: takeout.target_title,
+}
+
+
+def ad_hoc_target_title(db, job: str, target: str) -> str | None:
+    """The name of an ad-hoc run's target (None for a job with none, or a target since gone)."""
+    title = _TARGET_TITLES.get(job)
+    return title(db, target) if title is not None else None
+
+
+# The ad-hoc runs added and not yet recorded, `job@target`: between the two the target's newest
+# run row is still the previous run's, so a page watching it asks `ad_hoc_queued` too.
+_queued: set[str] = set()
+_queued_guard = threading.Lock()
+
+
+def ad_hoc_queued(job: str, target: str) -> bool:
+    """Whether a run of `job` on `target` is queued and its run row not written yet."""
+    with _queued_guard:
+        return f"{job}@{target}" in _queued
 
 
 def run_ad_hoc(
@@ -634,17 +658,24 @@ def run_ad_hoc(
     if sched is None:
         # Only between shutdown and exit: start_scheduler runs at every API startup.
         raise RuntimeError("the scheduler is not running")
+    key = f"{job}@{target}"
+    with _queued_guard:
+        _queued.add(key)
     try:
         sched.add_job(
             _ad_hoc,
             "date",
             args=(job, fn, actor_id, target, notify),
-            id=f"{job}@{target}",
+            id=key,
             misfire_grace_time=None,
         )
     # silent-ok: not a failure — the run for this target is already queued, and it does this work.
     except ConflictingIdError:
         log.info("ad-hoc %s for %s already queued", job, target)
+    except Exception:
+        with _queued_guard:
+            _queued.discard(key)
+        raise
 
 
 def _ad_hoc(
@@ -656,6 +687,10 @@ def _ad_hoc(
         # No row to tell it in: the database is what failed. The log is the record.
         log.exception("ad-hoc %s for %s not started: its run could not be recorded", job, target)
         return
+    finally:
+        # Recorded (or never will be): the run row now speaks for it.
+        with _queued_guard:
+            _queued.discard(f"{job}@{target}")
     _opened(run_id, job)
     try:
         _ad_hoc_work(job, fn, run_id, actor_id if notify else None)
@@ -672,7 +707,11 @@ def _ad_hoc_work(job: str, fn: Callable[[], dict], run_id: int, notify_id: int |
             log.exception("job %s failed", job, extra=logs.event("scheduler.job.failed", job=job))
             err, failure = _failure(exc)
             _close(run_id, RunStatus.ERROR, error=err, failure=failure)
-            _notify_ad_hoc(notify_id, job, RunStatus.ERROR, err, failure)
+            # The user's notice names the failure, never the upstream's own text: that is an English
+            # diagnostic, for the admin's run row and the log, not a user's inbox.
+            _notify_ad_hoc(
+                notify_id, job, RunStatus.ERROR, None, failure or {"failure": "unexpected"}
+            )
             return
         summary, data = _finish_ok(job, run_id, result)
         _notify_ad_hoc(notify_id, job, RunStatus.OK, summary, data)
@@ -725,3 +764,5 @@ def shutdown_scheduler() -> None:
     if _scheduler is not None:
         _scheduler.shutdown(wait=False)
         _scheduler = None
+    with _queued_guard:
+        _queued.clear()  # what it held will not run