fix(reader): the Takeout import keeps a later upload, an un-mark and its memory bound

- An ad-hoc request made while its target's run is under way is the run after it: the scheduler
  dropped a second fire of a running id, so a second Takeout file was never imported and its key
  stayed queued until a restart.
- Un-marking or resetting a video is remembered (video_unmarks): an import of an older YouTube view
  no longer marks it watched again; un-marks take the user's Takeout lock.
- One user's failure in the Shorts job's Takeout pass is theirs: the job keeps its own result, the
  others are settled, and a failed lookup spends none of the daily share.
- The upload reads the user's id before giving the read connection back.
- A watched row waits on a discovery video (search or explore) until the catalog keeps it, so a
  mark never pins it against the cleanup jobs.
- The history is decoded a window at a time, a Watch later file over 2 MiB is refused unread, and
  two uploads at once across all users (takeout_server_busy).
- A Watch later list goes in as one presence read and one insert.

backend/tests/test_scheduler.py

@@ -6,6 +6,7 @@ through (at the top of _job): holding _activity_lock across the check-and-set me
 caller can hold a job's "running" claim at a time, and it's claimable again once released."""
 
 import logging
+import threading
 import time
 from datetime import datetime, timedelta, timezone
 
@@ -426,6 +427,40 @@ def test_a_second_ad_hoc_request_before_the_first_starts_is_that_same_run(
     live_scheduler.remove_all_jobs()  # never run: a run finishing after the test would land in the next
 
 
+def test_a_request_made_while_its_run_is_under_way_is_the_run_after_it(
+    db, make_user, live_scheduler
+):
+    """The scheduler drops a second fire of an id whose run has not returned: the second Takeout file
+    of a pair, uploaded while the first one's import ran, was never imported, and its key stayed
+    queued until a restart (the page spun on "Processing…" for good)."""
+    user = make_user("again@example.com")
+    first_runs, release = threading.Event(), threading.Event()
+    calls = []
+
+    def first():
+        first_runs.set()
+        assert release.wait(10)
+        calls.append("first")
+        return {"n": 1}
+
+    def second():
+        calls.append("second")
+        return {"n": 2}
+
+    scheduler.run_ad_hoc(scheduler.CHANNEL_BACKFILL, first, actor_id=user.id, target="UC_again")
+    assert first_runs.wait(10)
+    scheduler.run_ad_hoc(scheduler.CHANNEL_BACKFILL, second, actor_id=user.id, target="UC_again")
+    assert scheduler.ad_hoc_queued(scheduler.CHANNEL_BACKFILL, "UC_again")
+    release.set()
+
+    deadline = time.monotonic() + 10
+    while time.monotonic() < deadline and len(_finished(db, "channel_backfill")) < 2:
+        time.sleep(0.05)
+    assert calls == ["first", "second"]
+    assert sorted(r.result["n"] for r in _finished(db, "channel_backfill")) == [1, 2]
+    assert not scheduler.ad_hoc_queued(scheduler.CHANNEL_BACKFILL, "UC_again")
+
+
 def test_a_manual_run_leaves_no_actor_behind_on_its_thread(db, make_user):
     """The scheduler's pool threads are reused: an actor a manual run left set made the next
     scheduled run on that thread an admin's "manual" run, with a notice in their inbox."""