From afcc1142024c942ab4ebe745fa22bde800255727 Mon Sep 17 00:00:00 2001 From: Arietids <69024461+ArietidsZ@users.noreply.github.com> Date: Wed, 28 Oct 2026 10:25:08 +0900 Subject: [PATCH 1/6] fix: preserve query-sensitive source identities across feed and API --- README.md | 15 +++++ tests/v2/test_api_app.py | 4 +- tests/v2/test_source_identity.py | 30 +++++++++ tests/v2/test_store-service_ingester.py | 84 +++++++++++++++++++++---- vra/api/app.py | 7 ++- vra/identity.py | 15 +++++ vra/service/ingester.py | 17 +---- 7 files changed, 142 insertions(+), 30 deletions(-) create mode 100644 tests/v2/test_source_identity.py create mode 100644 vra/identity.py diff --git a/README.md b/README.md index e36fe9d..6a3f1a0 100644 --- a/README.md +++ b/README.md @@ -142,3 +142,18 @@ Architecture and frozen contracts: `docs/v2/SPEC.md`. Per-stage evidence - An OpenAI-compatible engine serving `VRA_MODEL` โ€” see `vra engine` for the exact per-platform commands; ASR server optional (`VRA_ASR=off` to skip) + +### Source identity compatibility + +New feed and manual/API submissions share a versioned `source:v2:` SHA-256 +identity of the trimmed acquisition URL or path. Query parameters, their order, +encoding, fragments and path case are preserved; feed GUID is metadata, not a +substitute for the media source. This deliberately favors a possible duplicate +for differently spelled URLs over silently collapsing different videos. + +Existing legacy IDs, video UUIDs, jobs and saved artifacts are not migrated, +merged or deleted. Old query-stripped records may already contain mismatched +source metadata and summaries, so they are not automatically reused. Re-ingesting +an old source can create a new record and duplicate RSS item; historical repair +requires explicit review/reprocessing. No database schema-version bump or reset +is part of this change. Explicit processing by an existing video UUID is retained. diff --git a/tests/v2/test_api_app.py b/tests/v2/test_api_app.py index 2b2296b..863b756 100644 --- a/tests/v2/test_api_app.py +++ b/tests/v2/test_api_app.py @@ -303,10 +303,10 @@ def test_process_requires_source_or_video_id() -> None: assert resp.status_code == 400 -def test_canonical_for_source_is_stable_sha1_prefix() -> None: +def test_canonical_for_source_is_stable_versioned_source() -> None: assert canonical_for_source("https://x/y") == canonical_for_source("https://x/y") assert canonical_for_source("https://x/y") != canonical_for_source("https://x/z") - assert canonical_for_source("https://x/y").startswith("url:") + assert canonical_for_source("https://x/y").startswith("source:v2:") def test_jobs_list_and_filters() -> None: diff --git a/tests/v2/test_source_identity.py b/tests/v2/test_source_identity.py new file mode 100644 index 0000000..a84bd33 --- /dev/null +++ b/tests/v2/test_source_identity.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +import pytest + +from vra.api.app import canonical_for_source +from vra.identity import source_identity +from vra.service.ingester import _canonical_id + + +@pytest.mark.parametrize('left,right', [ + ('https://youtube.com/watch?v=A', 'https://youtube.com/watch?v=B'), + ('https://x/?a=1&b=2', 'https://x/?b=2&a=1'), + ('https://x/?a=1&a=2', 'https://x/?a=2'), + ('https://x/?v=a%2Fb', 'https://x/?v=a/b'), + ('/Media/A.mp4', '/Media/a.mp4'), +]) +def test_meaningful_source_distinctions_are_preserved(left, right): + assert source_identity(left) != source_identity(right) + + +def test_feed_and_api_use_the_same_identity_independent_of_guid(): + source = 'https://youtube.com/watch?v=A' + assert _canonical_id('https://unrelated-guid', source) == canonical_for_source(source) + assert _canonical_id('yt:video:A', source) == canonical_for_source(' ' + source + ' ') + assert source_identity(source).startswith('source:v2:') + + +def test_blank_source_rejected(): + with pytest.raises(ValueError): + source_identity(' ') diff --git a/tests/v2/test_store-service_ingester.py b/tests/v2/test_store-service_ingester.py index ffb594f..6e830e1 100644 --- a/tests/v2/test_store-service_ingester.py +++ b/tests/v2/test_store-service_ingester.py @@ -2,7 +2,6 @@ from __future__ import annotations -import hashlib from collections.abc import AsyncIterator, Callable from datetime import UTC, datetime from pathlib import Path @@ -13,11 +12,12 @@ import vra.service.ingester as ingester_module from vra.config import Config +from vra.identity import source_identity from vra.service.ingester import FeedIngester, HttpFeedFetcher from vra.store.db import Database from vra.store.queue import SqlJobQueue from vra.store.repos import SqlVideoStore -from vra.types import JobKind, JobState, VideoStatus +from vra.types import JobKind, JobState, Summary, VideoStatus _RSS = """ @@ -81,15 +81,15 @@ async def test_ingest_new_videos_enqueued( assert result.feed.last_checked is not None assert result.feed.error is None - # first entry: absolute http guid becomes the canonical id - first = await store.get_video_by_canonical("https://youtu.be/aaa") + # GUID is metadata; the acquisition source determines identity + first = await store.get_video_by_canonical(source_identity("https://youtu.be/aaa")) assert first is not None assert first.status == VideoStatus.PENDING assert first.guid == "https://youtu.be/aaa" assert first.published_at == datetime(2025, 6, 1, tzinfo=UTC) - # second entry: normalized url (host lowered, query dropped), sha1[:16] - expected = "url:" + hashlib.sha1(b"https://example.com/b").hexdigest()[:16] + # Query and case are preserved rather than guessed equivalent + expected = source_identity("https://Example.com/b?utm_source=x") second = await store.get_video_by_canonical(expected) assert second is not None assert second.source_url == "https://Example.com/b?utm_source=x" @@ -119,9 +119,9 @@ async def test_ingest_done_skipped_and_pending_deduped( # mark both done: third poll skips entirely for video in [ - await store.get_video_by_canonical("https://youtu.be/aaa"), + await store.get_video_by_canonical(source_identity("https://youtu.be/aaa")), await store.get_video_by_canonical( - "url:" + hashlib.sha1(b"https://example.com/b").hexdigest()[:16] + source_identity("https://Example.com/b?utm_source=x") ), ]: assert video is not None @@ -138,9 +138,9 @@ async def test_ingest_failed_video_reenqueued( ingester = FeedIngester(FakeFetcher(_entries()), store, queue, Config()) await ingester.ingest("https://example.com/rss") - first = await store.get_video_by_canonical("https://youtu.be/aaa") + first = await store.get_video_by_canonical(source_identity("https://youtu.be/aaa")) second = await store.get_video_by_canonical( - "url:" + hashlib.sha1(b"https://example.com/b").hexdigest()[:16] + source_identity("https://Example.com/b?utm_source=x") ) assert first is not None and second is not None @@ -222,7 +222,7 @@ async def test_ingest_skips_entries_without_source_url( result = await ingester.ingest("https://example.com/rss") assert (result.new_videos, result.enqueued) == (1, 1) assert await store.get_video_by_canonical( - "url:" + hashlib.sha1(b"https://x/1").hexdigest()[:16] + source_identity("https://x/1") ) is not None @@ -412,3 +412,65 @@ def handle(request): with pytest.raises(ValueError, match='limit'): await HttpFeedFetcher(client, allow_private=True).fetch('http://localhost/start') assert len(seen) == 6 + + +async def test_query_distinctions_and_guid_reuse_do_not_overwrite(stack): + _, store, queue = stack + urls = ['https://youtube.com/watch?v=A', 'https://youtube.com/watch?v=B'] + entries = [dict(source_url=url, title=url, guid='https://same-guid') for url in urls] + ingester = FeedIngester(FakeFetcher(entries), store, queue, Config()) + result = await ingester.ingest('https://feed') + assert result.new_videos == 2 + rows = [await store.get_video_by_canonical(source_identity(url)) for url in urls] + assert rows[0].id != rows[1].id + assert [row.source_url for row in rows] == urls + assert len(await queue.list(kind=JobKind.PROCESS)) == 2 + assert (await ingester.ingest('https://feed')).new_videos == 0 + + +async def test_ambiguous_legacy_row_is_not_adopted_or_mutated(stack): + db, store, queue = stack + source = 'https://youtube.com/watch?v=B' + old, _ = await store.upsert_video_ref( + canonical_id='url:old-lossy-hash', source_url=source, feed_id=None, + guid='yt:video:B', title='old possibly mismatched artifact', published_at=None) + await store.set_video_status(old.id, VideoStatus.DONE) + await store.save_summary(old.id, Summary(summary='possibly source A, not B')) + old_summary = dict(await db.fetchone('SELECT * FROM summaries WHERE video_id = ?', (old.id,))) + old_job = await queue.enqueue(JobKind.PROCESS, {'video_id': old.id}, video_id=old.id) + result = await FeedIngester(FakeFetcher([dict(source_url=source, guid='yt:video:B')]), + store, queue, Config()).ingest('https://feed') + assert result.new_videos == 1 + current = await store.get_video_by_canonical(source_identity(source)) + legacy = await store.get_video_by_canonical('url:old-lossy-hash') + assert current.id != legacy.id == old.id + assert legacy.status == VideoStatus.DONE + assert legacy.title == 'old possibly mismatched artifact' + assert dict(await db.fetchone('SELECT * FROM summaries WHERE video_id = ?', (old.id,))) == old_summary + assert any(job.id == old_job.id and job.video_id == old.id for job in await queue.list(kind=JobKind.PROCESS)) + + +@pytest.mark.parametrize('feed_first', [True, False]) +async def test_feed_and_api_share_record_and_default_job(stack, feed_first): + from types import SimpleNamespace + + from vra.api.app import create_app + _, store, queue = stack + source = 'http://localhost/video?v=A' + config = Config(allow_private_net=True) + ingester = FeedIngester(FakeFetcher([dict(source_url=source, guid='yt:video:A')]), + store, queue, config) + app = create_app(SimpleNamespace(config=config, store=store, queue=queue)) + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), + base_url='http://test') as client: + if feed_first: + await ingester.ingest('http://localhost/feed') + response = await client.post('/process', json={'source_url': source}) + assert response.status_code == 200 + if not feed_first: + await ingester.ingest('http://localhost/feed') + record = await store.get_video_by_canonical(source_identity(source)) + assert response.json()['video_id'] == record.id + jobs = await queue.list(kind=JobKind.PROCESS) + assert len(jobs) == 1 + assert jobs[0].video_id == record.id diff --git a/vra/api/app.py b/vra/api/app.py index b990811..45884c4 100644 --- a/vra/api/app.py +++ b/vra/api/app.py @@ -25,7 +25,6 @@ from __future__ import annotations import asyncio -import hashlib import importlib.resources import ipaddress import logging @@ -55,6 +54,7 @@ setup_payload, ) from vra.config import AsrMode, Config +from vra.identity import source_identity from vra.rss import RssWriter from vra.types import JobKind, JobState @@ -97,7 +97,7 @@ class ProcessBody(BaseModel): def canonical_for_source(source: str) -> str: """Cheap canonical id for a source URL/path before platform resolution.""" - return "url:" + hashlib.sha1(source.encode("utf-8")).hexdigest()[:16] + return source_identity(source) def _iso(value: Any) -> str | None: @@ -369,6 +369,7 @@ async def process_source(body: ProcessBody) -> dict[str, Any]: if body.video_id: video_id = body.video_id elif body.source_url: + body.source_url = body.source_url.strip() await validate_source_url(body.source_url, config) canonical = canonical_for_source(body.source_url) record = await runtime.store.get_video_by_canonical(canonical) @@ -387,7 +388,7 @@ async def process_source(body: ProcessBody) -> dict[str, Any]: job = await runtime.queue.enqueue( JobKind.PROCESS, - {"video_id": video_id, "force": body.force}, + {"video_id": video_id, **({"force": True} if body.force else {})}, video_id=video_id, ) return {"job_id": job.id, "video_id": video_id} diff --git a/vra/identity.py b/vra/identity.py new file mode 100644 index 0000000..740d543 --- /dev/null +++ b/vra/identity.py @@ -0,0 +1,15 @@ +"""Conservative source identity shared by feed ingestion and manual requests. + +Legacy lossy identities are never adopted automatically: their saved artifacts +may belong to an earlier colliding URL even when source_url now looks correct. +""" +from __future__ import annotations + +import hashlib + + +def source_identity(source: str) -> str: + value = source.strip() + if not value: + raise ValueError('source must not be empty') + return 'source:v2:' + hashlib.sha256(value.encode('utf-8')).hexdigest() diff --git a/vra/service/ingester.py b/vra/service/ingester.py index 56d5ad5..ff8a22f 100644 --- a/vra/service/ingester.py +++ b/vra/service/ingester.py @@ -17,7 +17,6 @@ from __future__ import annotations import asyncio -import hashlib import ipaddress import logging import socket @@ -31,6 +30,7 @@ from vra.config import Config from vra.contracts import FeedFetcher, JobQueue, VideoStore +from vra.identity import source_identity from vra.types import IngestResult, JobKind, VideoStatus log = logging.getLogger(__name__) @@ -219,20 +219,9 @@ async def _validate_url(self, url: str) -> None: # canonical identity + FeedIngester # -------------------------------------------------------------------------- -def _normalize_url(url: str) -> str: - """scheme://host/path form with query/fragment dropped (host lowered).""" - parts = urlparse(url) - if parts.scheme and parts.netloc: - return f"{parts.scheme}://{parts.netloc.lower()}{parts.path}" - return parts.path or url - - def _canonical_id(guid: str | None, source_url: str) -> str: - """Absolute-http guid wins; otherwise sha1 of the normalized URL.""" - if guid and guid.startswith("http"): - return guid - digest = hashlib.sha1(_normalize_url(source_url).encode("utf-8")).hexdigest() - return f"url:{digest[:16]}" + """GUID remains metadata; acquisition source determines new identity.""" + return source_identity(source_url) class FeedIngester: From cf961001624604019cf6cb01cbe45dd83dd35f3a Mon Sep 17 00:00:00 2001 From: Arietids <69024461+ArietidsZ@users.noreply.github.com> Date: Wed, 28 Oct 2026 10:28:37 +0900 Subject: [PATCH 2/6] fix: reject blank source before local-path validation --- tests/v2/test_api_app.py | 8 ++++++++ vra/api/app.py | 2 ++ 2 files changed, 10 insertions(+) diff --git a/tests/v2/test_api_app.py b/tests/v2/test_api_app.py index 863b756..a567f5f 100644 --- a/tests/v2/test_api_app.py +++ b/tests/v2/test_api_app.py @@ -488,3 +488,11 @@ def test_asr_off_satisfies_ready_logic() -> None: asr = report["dependencies"]["asr"] assert asr["enabled"] is False assert report["runtime_view"]["state"] == "ready" + + +def test_blank_source_is_bad_request_even_when_local_paths_allowed(): + runtime = FakeRuntime(_config(allow_local_path=True)) + with _client(runtime) as client: + response = client.post('/process', json={'source_url': ' '}) + assert response.status_code == 400 + assert response.json()['detail'] == 'source_url must not be blank' diff --git a/vra/api/app.py b/vra/api/app.py index 45884c4..a68351b 100644 --- a/vra/api/app.py +++ b/vra/api/app.py @@ -370,6 +370,8 @@ async def process_source(body: ProcessBody) -> dict[str, Any]: video_id = body.video_id elif body.source_url: body.source_url = body.source_url.strip() + if not body.source_url: + raise HTTPException(status_code=400, detail="source_url must not be blank") await validate_source_url(body.source_url, config) canonical = canonical_for_source(body.source_url) record = await runtime.store.get_video_by_canonical(canonical) From 0eea809a6e6714a2141302d5e3e830b88b17920e Mon Sep 17 00:00:00 2001 From: Arietids <69024461+ArietidsZ@users.noreply.github.com> Date: Thu, 29 Oct 2026 17:07:00 +0900 Subject: [PATCH 3/6] Normalize new queue deadlines to precise UTC instants Normalize datetime and ISO string input before dedup; reject invalid deadlines. 46 SQLite store tests passed; scoped Ruff passed. Existing non-UTC database rows are intentionally not migrated. --- README.md | 11 ++++++ tests/v2/test_store-service_queue.py | 56 ++++++++++++++++++++++++++++ vra/store/queue.py | 27 ++++++++------ 3 files changed, 83 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index 6a3f1a0..8c67365 100644 --- a/README.md +++ b/README.md @@ -42,6 +42,17 @@ vra status # counts + jobs curl -s localhost:8080/rss?limit=20 # consume the output ``` +## Queue deadlines + +New job deadlines accept datetimes or ISO-8601 strings and are stored in UTC +with six-digit microsecond precision. Naive values are interpreted as UTC; +invalid values are rejected even when the payload matches an existing job. +This prevents timezone offsets from changing when a job becomes due. + +Existing database rows are not rewritten by this change. Historical non-UTC +`run_at` values require a separately reviewed, transactional normalization +before replay; no automatic migration or database reset is performed. + ## The stack (one posture โ€” `vra engine` prints these exact commands) | role | model ยท quant | size | diff --git a/tests/v2/test_store-service_queue.py b/tests/v2/test_store-service_queue.py index c24eca6..989f541 100644 --- a/tests/v2/test_store-service_queue.py +++ b/tests/v2/test_store-service_queue.py @@ -238,3 +238,59 @@ async def test_queue_close_delegates_to_database(tmp_path: Path) -> None: queue = SqlJobQueue(db) await queue.close() assert db.path == str(tmp_path / "vra.db") # attributes intact after close + + +@pytest.mark.parametrize("as_string", [False, True]) +async def test_offset_deadlines_use_utc_instant(queue: SqlJobQueue, monkeypatch, as_string: bool) -> None: + from datetime import timezone + + from vra.store import queue as module + + now = datetime(2026, 10, 29, 12, tzinfo=UTC) + monkeypatch.setattr(module, "_now_iso", lambda: module._iso(now)) + past = datetime(2026, 10, 29, 19, tzinfo=timezone(timedelta(hours=8))) + future = datetime(2026, 10, 29, 5, tzinfo=timezone(timedelta(hours=-8))) + later = await queue.enqueue(JobKind.PROCESS, {"offset": "future"}, + run_at=future.isoformat() if as_string else future) + due = await queue.enqueue(JobKind.PROCESS, {"offset": "past"}, + run_at=past.isoformat() if as_string else past) + assert due.run_at == datetime(2026, 10, 29, 11, tzinfo=UTC) + assert later.run_at == datetime(2026, 10, 29, 13, tzinfo=UTC) + assert (await queue.claim((JobKind.PROCESS,))).id == due.id + assert await queue.claim((JobKind.PROCESS,)) is None + + +async def test_deadline_microsecond_boundary(queue: SqlJobQueue, monkeypatch) -> None: + from vra.store import queue as module + + now = datetime(2026, 10, 29, 12, tzinfo=UTC) + monkeypatch.setattr(module, "_now_iso", lambda: module._iso(now)) + await queue.enqueue(JobKind.PROCESS, {"future": 1}, run_at=now + timedelta(microseconds=1)) + due = await queue.enqueue(JobKind.PROCESS, {"due": 1}, run_at=now.replace(tzinfo=None)) + assert (await queue.claim((JobKind.PROCESS,))).id == due.id + assert await queue.claim((JobKind.PROCESS,)) is None + + +@pytest.mark.parametrize("value", ["not a date", "2026-13-01", 123, [], {}]) +async def test_invalid_deadline_never_inserts(queue: SqlJobQueue, value) -> None: + with pytest.raises((TypeError, ValueError)): + await queue.enqueue(JobKind.PROCESS, {"invalid": str(value)}, run_at=value) + assert await queue.list() == [] + + +def test_deadline_canonical_serialization() -> None: + from vra.store.queue import _iso + + assert _iso(None) is None + expected = "2026-10-29T12:00:00.000000+00:00" + for value in ["2026-10-29T12:00:00Z", "2026-10-29T20:00:00+08:00", + "2026-10-29T12:00:00", datetime(2026, 10, 29, 12)]: + assert _iso(value) == expected + + +async def test_invalid_duplicate_deadline_is_rejected(queue: SqlJobQueue) -> None: + original = await queue.enqueue(JobKind.PROCESS, {"same": 1}) + with pytest.raises(ValueError): + await queue.enqueue(JobKind.PROCESS, {"same": 1}, run_at="invalid") + jobs = await queue.list() + assert len(jobs) == 1 and jobs[0].id == original.id diff --git a/vra/store/queue.py b/vra/store/queue.py index 0a2a8fa..96613a2 100644 --- a/vra/store/queue.py +++ b/vra/store/queue.py @@ -18,7 +18,9 @@ claimable. Timestamps compare lexicographically โ€” all writers use -``datetime.now(UTC).isoformat()`` (fixed-width UTC ISO-8601). +UTC ISO-8601 with an explicit six-digit microsecond field. Legacy stored +non-UTC deadlines need normalization before replay; serialization only +normalizes new writes. """ from __future__ import annotations @@ -41,18 +43,20 @@ def _utc_now() -> datetime: def _now_iso() -> str: - return datetime.now(UTC).isoformat() + return datetime.now(UTC).isoformat(timespec="microseconds") def _iso(value: Any) -> str | None: - """Normalize a datetime (naive treated as UTC) to an ISO-8601 string.""" + """Canonical UTC deadline; naive datetimes/ISO strings are treated as UTC.""" if value is None: return None - if isinstance(value, datetime): - if value.tzinfo is None: - value = value.replace(tzinfo=UTC) - return value.isoformat() - return str(value) + if isinstance(value, str): + value = datetime.fromisoformat(value) + if not isinstance(value, datetime): + raise TypeError("run_at must be a datetime, ISO-8601 string, or None") + if value.tzinfo is None: + value = value.replace(tzinfo=UTC) + return value.astimezone(UTC).isoformat(timespec="microseconds") def _parse_dt(value: Any) -> datetime | None: @@ -115,6 +119,7 @@ async def enqueue( otherwise inserts a fresh queued job. ``dedup_key`` is accepted for caller readability and not persisted. """ + normalized_run_at = _iso(run_at) del dedup_key # advisory only; identity is kind + canonical payload payload_json = json.dumps(payload, sort_keys=True, separators=(",", ":")) now = _now_iso() @@ -139,7 +144,7 @@ async def enqueue( ) VALUES (?, ?, 'queued', ?, 0, 3, ?, ?, ?, ?) """, - (job_id, kind.value, payload_json, _iso(run_at), video_id, now, now), + (job_id, kind.value, payload_json, normalized_run_at, video_id, now, now), commit=False, ) row = await self._db.fetchone( @@ -219,8 +224,8 @@ async def fail(self, job_id: str, error: str, *, retry: bool) -> None: async def requeue_stale(self, older_than_s: float) -> int: """Requeue running jobs with stale updated_at; returns the count.""" now_dt = _utc_now() - cutoff = (now_dt - timedelta(seconds=older_than_s)).isoformat() - now = now_dt.isoformat() + cutoff = _iso(now_dt - timedelta(seconds=older_than_s)) + now = _iso(now_dt) async with self._db.lock: rows = await self._db.execute( """ From 67c186f1e17bb762b22c62fa23fc5015a8a325d8 Mon Sep 17 00:00:00 2001 From: Arietids <69024461+ArietidsZ@users.noreply.github.com> Date: Fri, 30 Oct 2026 09:16:33 +0900 Subject: [PATCH 4/6] fix(runtime): persist interrupted jobs before shutdown --- tests/v2/test_integration.py | 95 ++++++++++++++++++++++++++++++++++++ vra/runtime.py | 21 ++++++++ 2 files changed, 116 insertions(+) diff --git a/tests/v2/test_integration.py b/tests/v2/test_integration.py index a8d450d..d16a6e7 100644 --- a/tests/v2/test_integration.py +++ b/tests/v2/test_integration.py @@ -292,3 +292,98 @@ async def test_asr_off_still_completes_without_transcript(tmp_path): assert counts["videos_by_status"].get("done", 0) == 1 finally: await runtime.queue.close() + + +@pytest.mark.parametrize("kind", [JobKind.INGEST, JobKind.PROCESS]) +@pytest.mark.parametrize("last_attempt", [False, True]) +async def test_cancelled_job_persists_retry_or_terminal_state(tmp_path, kind, last_attempt): + import asyncio + + from vra.types import VideoStatus + + runtime = await _make_runtime( + tmp_path, FakeFetcher(entries=[]), FakeAcquirer(), FakeTranscriber(), + FakeExtractor(), FakeSummarizer(), + ) + entered = asyncio.Event() + video, _ = await runtime.store.upsert_video_ref( + canonical_id="cancel:test", source_url="https://example.com/video", + feed_id=None, guid=None, title=None, published_at=None, + ) + job = await runtime.queue.enqueue(kind, {"video_id": video.id}) + if last_attempt: + await runtime.queue._db.execute("UPDATE jobs SET attempts = 2 WHERE id = ?", (job.id,)) + claimed = await runtime.queue.claim((kind,)) + assert claimed is not None + + async def interrupted(): + if kind is JobKind.PROCESS: + await runtime.store.set_video_status(video.id, VideoStatus.ACQUIRING) + entered.set() + await asyncio.Event().wait() + + task = asyncio.create_task(_run_job(claimed, interrupted(), runtime.queue, runtime.store)) + try: + await asyncio.wait_for(entered.wait(), 2) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + saved = await runtime.queue.get(job.id) + assert saved.state is (JobState.FAILED if last_attempt else JobState.QUEUED) + assert saved.attempts == (3 if last_attempt else 1) + assert saved.error == "job cancelled during execution" + if kind is JobKind.PROCESS: + assert (await runtime.store.get_video(video.id)).status is VideoStatus.FAILED + if not last_attempt: + assert saved.run_at is not None + # Advance the durable deadline without sleeping or contacting a service. + await runtime.queue._db.execute("UPDATE jobs SET run_at = NULL WHERE id = ?", (job.id,)) + retry = await runtime.queue.claim((kind,)) + assert retry.id == job.id and retry.attempts == 2 + async def done(): + return "done" + await _run_job(retry, done(), runtime.queue, runtime.store) + assert (await runtime.queue.get(job.id)).state is JobState.DONE + finally: + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + await runtime.queue.close() + + +async def test_runtime_close_requeues_worker_and_preserves_done_video(tmp_path): + import asyncio + + from vra.types import VideoStatus + + entered = asyncio.Event() + + class CompletingProcessor: + async def process(self, video_id, *, force=False): + await runtime.store.set_video_status(video_id, VideoStatus.DONE) + entered.set() + await asyncio.Event().wait() + + runtime = await _make_runtime( + tmp_path, FakeFetcher(entries=[]), FakeAcquirer(), FakeTranscriber(), + FakeExtractor(), FakeSummarizer(), cpu_workers=1, + ) + runtime = dc.replace(runtime, processor=CompletingProcessor()) + video, _ = await runtime.store.upsert_video_ref( + canonical_id="cancel:done", source_url="https://example.com/done", + feed_id=None, guid=None, title=None, published_at=None, + ) + job = await runtime.queue.enqueue(JobKind.PROCESS, {"video_id": video.id}) + try: + await runtime.start() + await asyncio.wait_for(entered.wait(), 2) + finally: + await runtime.close() + # A fresh connection proves recovery was committed before close returned. + db = await Database.connect(runtime.config.database_path) + try: + saved = await SqlJobQueue(db).get(job.id) + assert saved.state is JobState.QUEUED + assert (await SqlVideoStore(db).get_video(video.id)).status is VideoStatus.DONE + finally: + await db.close() diff --git a/vra/runtime.py b/vra/runtime.py index 97a6be5..40e5d7f 100644 --- a/vra/runtime.py +++ b/vra/runtime.py @@ -16,6 +16,9 @@ - Inference concurrency is managed by the serving engine (continuous batching); the app does not gate it further. - On boot, running jobs stale for > 30 minutes are requeued. +- Normal worker cancellation records a retry (or terminal failure at the + existing attempt limit) before SQLite closes. This is not a heartbeat or + recovery guarantee for forced termination, repeated cancellation or DB errors. """ from __future__ import annotations @@ -41,6 +44,7 @@ from vra.store.db import Database from vra.store.queue import SqlJobQueue from vra.store.repos import SqlVideoStore +from vra.types import VideoStatus log = logging.getLogger(__name__) @@ -55,6 +59,23 @@ async def _run_job(job: Job, coro: Any, queue: SqlJobQueue, store: SqlVideoStore await queue.fail(job.id, detail or "processing failed", retry=True) else: await queue.complete(job.id) + except asyncio.CancelledError: + # Normal shutdown waits for workers before closing SQLite. Persist the + # interruption here so a quick restart cannot strand a fresh running + # job behind the startup-only 30-minute stale threshold. + detail = "job cancelled during execution" + if job.kind is JobKind.PROCESS: + video_id = str(job.payload.get("video_id", "")) + try: + rec = await store.get_video(video_id) + if rec is not None and rec.status in ( + VideoStatus.ACQUIRING, VideoStatus.SUMMARIZING, + ): + await store.set_video_status(video_id, VideoStatus.FAILED, error=detail) + except Exception: + log.exception("could not mark interrupted video %s as failed", video_id) + await queue.fail(job.id, detail, retry=True) + raise except Exception as exc: log.warning("job %s failed: %s", job.id, exc) await queue.fail(job.id, str(exc), retry=True) From f938dd09510ecbd16fd62ca8428adb2401785b69 Mon Sep 17 00:00:00 2001 From: Arietids <69024461+ArietidsZ@users.noreply.github.com> Date: Fri, 30 Oct 2026 09:21:21 +0900 Subject: [PATCH 5/6] fix(runtime): close owned feed and ASR clients --- tests/v2/test_integration.py | 31 +++++++++++++++++++++++++++++++ vra/runtime.py | 17 ++++++++++++++--- 2 files changed, 45 insertions(+), 3 deletions(-) diff --git a/tests/v2/test_integration.py b/tests/v2/test_integration.py index d16a6e7..b7f3f67 100644 --- a/tests/v2/test_integration.py +++ b/tests/v2/test_integration.py @@ -387,3 +387,34 @@ async def process(self, video_id, *, force=False): assert (await SqlVideoStore(db).get_video(video.id)).status is VideoStatus.DONE finally: await db.close() + + +@pytest.mark.parametrize("cleanup_fails", [False, True]) +async def test_built_runtime_closes_feed_and_transcription_clients(tmp_path, cleanup_fails): + from vra.runtime import build_runtime + + runtime = await build_runtime(_cfg(tmp_path)) + feed_http = runtime.ingester._fetcher._client + transcriber = runtime.processor._transcriber + asr_http = transcriber._client + if cleanup_fails: + class FailingSummarizer: + async def close(self): + await runtime.engine.close() + raise RuntimeError("cleanup failed") + runtime = dc.replace(runtime, summarizer=FailingSummarizer()) + try: + if cleanup_fails: + with pytest.raises(RuntimeError, match="cleanup failed"): + await runtime.close() + else: + await runtime.close() + assert feed_http.is_closed + assert asr_http.is_closed + assert runtime.queue._db._closed + finally: + # The red-phase regression also cleans up its deliberately exposed leak. + await feed_http.aclose() + await transcriber.close() + await runtime.engine.close() + await runtime.queue.close() diff --git a/vra/runtime.py b/vra/runtime.py index 40e5d7f..05e8a89 100644 --- a/vra/runtime.py +++ b/vra/runtime.py @@ -152,6 +152,8 @@ class AppRuntime: scheduler: PollScheduler summarizer: VlmSummarizer engine: EngineClient + _feed_http: httpx.AsyncClient | None = field(default=None, repr=False, compare=False) + _transcriber: OpenAiAsrTranscriber | None = field(default=None, repr=False, compare=False) _state: dict[str, Any] = field(default_factory=dict, repr=False, compare=False) async def start(self) -> None: @@ -193,9 +195,16 @@ async def close(self) -> None: self._state["tasks"] = [] self._state.pop("stop", None) self._state["started"] = False - await self.scheduler.stop() - await self.summarizer.close() - await self.queue.close() + # These clients are created by build_runtime, not owned by the + # summarizer. Unwind every owned resource even if another close fails. + async with contextlib.AsyncExitStack() as cleanup: + cleanup.push_async_callback(self.queue.close) + if self._feed_http is not None: + cleanup.push_async_callback(self._feed_http.aclose) + if self._transcriber is not None: + cleanup.push_async_callback(self._transcriber.close) + cleanup.push_async_callback(self.summarizer.close) + await self.scheduler.stop() def stats(self) -> dict[str, Any]: return { @@ -246,4 +255,6 @@ async def build_runtime(config: Config | None = None) -> AppRuntime: scheduler=scheduler, summarizer=summarizer, engine=engine, + _feed_http=http, + _transcriber=transcriber, ) From f80742104bded5dae5f5b4ba2dcc3c0f507b96b0 Mon Sep 17 00:00:00 2001 From: Arietids <69024461+ArietidsZ@users.noreply.github.com> Date: Fri, 30 Oct 2026 21:19:18 +0900 Subject: [PATCH 6/6] perf(queue): wake idle workers after due enqueue commits Preserve durable polling fallback and cancellation isolation; include bounded offline measurements and regressions. --- docs/v2/performance/queue-wakeup/README.md | 21 ++ .../v2/performance/queue-wakeup/baseline.json | 1 + .../performance/queue-wakeup/candidate.json | 1 + .../queue-wakeup/intermediate-broadcast.json | 1 + docs/v2/performance/queue-wakeup/probe.py | 121 ++++++++ tests/v2/test_queue_wakeup.py | 279 ++++++++++++++++++ vra/runtime.py | 10 +- vra/store/queue.py | 60 ++-- 8 files changed, 475 insertions(+), 19 deletions(-) create mode 100644 docs/v2/performance/queue-wakeup/README.md create mode 100644 docs/v2/performance/queue-wakeup/baseline.json create mode 100644 docs/v2/performance/queue-wakeup/candidate.json create mode 100644 docs/v2/performance/queue-wakeup/intermediate-broadcast.json create mode 100644 docs/v2/performance/queue-wakeup/probe.py create mode 100644 tests/v2/test_queue_wakeup.py diff --git a/docs/v2/performance/queue-wakeup/README.md b/docs/v2/performance/queue-wakeup/README.md new file mode 100644 index 0000000..7876f7d --- /dev/null +++ b/docs/v2/performance/queue-wakeup/README.md @@ -0,0 +1,21 @@ +# Local queue wakeup experiment (2026-10-30) + +Baseline: `f938dd09510ecbd16fd62ca8428adb2401785b69`. Candidate: the sibling commit containing this report. + +Scope: real temporary SQLite, fake no-op processing, no network/model/device calls. Run `PROBE_LABEL=