fix(catalog): a 24/7 stream archive's years-long length is unknown, not summed or sorted by

YouTube reports a 24/7 stream's archive with the years it has run (prod: 122,962,914 s), and one such video made the Watch later total read 44,073 hours. A length past a day is now no length (`Video.length_seconds`, LENGTH_MAX_S): the API sends none, the playlist and channel totals leave it out and count it as `live_without_length` ("+N live" beside the sum), and the duration filter, sort and Home budget never match it. Migration 0117 moves the duration sort indexes onto the length expression. An AST test holds who may still read the raw column. The migration 0110 test now ends its read before the way down: 0117 builds an index CONCURRENTLY there, which waited for that open transaction forever.
2026-11-07 02:20:03 +01:00
parent 1bf3d6a592
commit 5b846136ad
26 changed files with 446 additions and 39 deletions
@@ -0,0 +1,76 @@
+"""videos: the duration sorts ride an index on the video's length, not on the raw column
+
+Revision ID: 0117_videos_length_sort_indexes
+Revises: 0116_followed_explored_channels
+Create Date: 2026-11-07
+
+A YouTube length past a day is not a length (`catalog.models.LENGTH_MAX_S`): a 24/7 stream's archive
+reports the years it has run. The feed's duration sorts now order by `Video.length_seconds`, which
+is NULL for such a video, so "longest first" no longer opens on a stream archive. Migration 0096's
+two duration indexes are on the raw column and no longer match that ORDER BY; they are replaced by
+the same indexes over the length expression. The expression is spelled out here as SQLAlchemy emits
+it, with the limit as a literal (an index matches a query's constant, never a bound parameter);
+tests/test_feed_sort_indexes.py fails if the two drift apart.
+
+CONCURRENTLY (autocommit_block), re-runnable, as in 0096.
+"""
+
+from alembic import op
+from sqlalchemy import text
+
+revision: str = "0117_videos_length_sort_indexes"
+down_revision: str | None = "0116_followed_explored_channels"
+branch_labels: str | None = None
+depends_on: str | None = None
+
+# Frozen with this migration: `catalog.models.LENGTH_MAX_S` is 24 h.
+LENGTH = "(CASE WHEN (duration_seconds > 86400) THEN NULL ELSE duration_seconds END)"
+
+NEW = {
+    "ix_videos_length_desc": f"{LENGTH} DESC NULLS LAST, id",
+    "ix_videos_length_asc": f"{LENGTH} ASC NULLS LAST, id",
+}
+OLD = {
+    "ix_videos_duration_desc": "duration_seconds DESC NULLS LAST, id",
+    "ix_videos_duration_asc": "duration_seconds ASC NULLS LAST, id",
+}
+
+
+def _valid(name: str) -> bool | None:
+    """The index's `indisvalid`, or None when there is no index of that name."""
+    row = (
+        op.get_bind()
+        .execute(
+            text(
+                "SELECT i.indisvalid FROM pg_index i JOIN pg_class c ON c.oid = i.indexrelid "
+                "WHERE c.relname = :name"
+            ),
+            {"name": name},
+        )
+        .first()
+    )
+    return None if row is None else bool(row[0])
+
+
+def _build(name: str, columns: str) -> None:
+    """CREATE INDEX CONCURRENTLY, re-runnable: an INVALID leftover of a failed build is dropped first
+    (see 0096 `_build`)."""
+    if _valid(name) is False:
+        op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {name}")
+    op.execute(f"CREATE INDEX CONCURRENTLY IF NOT EXISTS {name} ON videos ({columns})")
+
+
+def upgrade() -> None:
+    with op.get_context().autocommit_block():
+        for name, columns in NEW.items():
+            _build(name, columns)
+        for name in OLD:
+            op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {name}")
+
+
+def downgrade() -> None:
+    with op.get_context().autocommit_block():
+        for name, columns in OLD.items():
+            _build(name, columns)
+        for name in NEW:
+            op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {name}")