From 139f6d9e8ff8c5b71dffd5cb4eef65e62fd47ccf Mon Sep 17 00:00:00 2001 From: Sean Keever <33592180+swkeever@users.noreply.github.com> Date: Thu, 24 Sep 2026 06:32:41 -0400 Subject: [PATCH 01/11] fix(python): run full mutation audit through Poe shell task --- maintainers/quality-policy.lock.json | 5 ++++- pyproject.toml | 2 +- scripts/check_quality_policy.py | 2 +- 3 files changed, 6 insertions(+), 3 deletions(-) diff --git a/maintainers/quality-policy.lock.json b/maintainers/quality-policy.lock.json index 54d94fc..8c96562 100644 --- a/maintainers/quality-policy.lock.json +++ b/maintainers/quality-policy.lock.json @@ -400,7 +400,10 @@ "generated": "python -m scripts.check_openapi", "lint": "ruff check --config pyproject.toml .", "mutation": "bash scripts/mutation.sh", - "mutation-full": "MUTATION_FULL=1 bash scripts/mutation.sh", + "mutation-full": { + "interpreter": "bash", + "shell": "MUTATION_FULL=1 bash scripts/mutation.sh" + }, "mypy": "mypy --config-file pyproject.toml", "package-check": { "interpreter": "bash", diff --git a/pyproject.toml b/pyproject.toml index d012d25..286c0e9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -498,7 +498,7 @@ types = ["mypy", "basedpyright"] mypy = "mypy --config-file pyproject.toml" basedpyright = "basedpyright --project pyproject.toml" mutation = "bash scripts/mutation.sh" -mutation-full = "MUTATION_FULL=1 bash scripts/mutation.sh" +mutation-full = { shell = "MUTATION_FULL=1 bash scripts/mutation.sh", interpreter = "bash" } [tool.poe.tasks.test] interpreter = "bash" diff --git a/scripts/check_quality_policy.py b/scripts/check_quality_policy.py index 784e2d5..6ad29db 100644 --- a/scripts/check_quality_policy.py +++ b/scripts/check_quality_policy.py @@ -18,7 +18,7 @@ from collections.abc import Iterable GENERATED = "src/volcano_sdk/_generated" -LOCK_SHA256 = "489b1a41632f67d528dbb10267bb8921c38c78726c430d990b4ee99ebdd9236a" +LOCK_SHA256 = "557def9c362e95b993e315925f31ddab2cb4666acd6f97411aec64ec472b3f3d" TYPE_FIXTURES = { "tests/typing/contract_steps.py", "tests/typing/durable_callbacks.py", From ddf27cf6d9c06d0e8407572d981f56561ee7467c Mon Sep 17 00:00:00 2001 From: Sean Keever <33592180+swkeever@users.noreply.github.com> Date: Thu, 24 Sep 2026 06:42:29 -0400 Subject: [PATCH 02/11] feat(python): require full sharded mutation on pull requests --- .github/workflows/ci.yml | 31 +++++-- maintainers/quality-policy.lock.json | 2 +- pyproject.toml | 2 +- scripts/check_mutation_shards.py | 118 +++++++++++++++++++++++++++ scripts/check_quality_policy.py | 2 +- scripts/mutation.sh | 21 ++++- tests/unit/test_mutation_shards.py | 81 ++++++++++++++++++ 7 files changed, 247 insertions(+), 10 deletions(-) create mode 100644 scripts/check_mutation_shards.py create mode 100644 tests/unit/test_mutation_shards.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 041345b..fd6b0f4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -59,11 +59,15 @@ jobs: if-no-files-found: ignore mutation: - name: Mutation Gate + name: Mutation Gate (${{ matrix.shard }}) runs-on: ubuntu-latest - timeout-minutes: 40 + timeout-minutes: 90 permissions: contents: read + strategy: + fail-fast: false + matrix: + shard: [0, 1, 2, 3, 4, 5, 6, 7] steps: - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 with: @@ -76,15 +80,16 @@ jobs: with: version: "0.12.17" - run: uv sync --locked - - name: Mutate changed and critical runtime modules + - name: Mutate all handwritten runtime modules in this shard env: - MUTATION_BASE_SHA: ${{ github.event.pull_request.base.sha || github.event.merge_group.base_sha || github.event.before }} - run: uv run --locked poe mutation + MUTATION_SHARD_INDEX: ${{ matrix.shard }} + MUTATION_SHARD_COUNT: 8 + run: uv run --locked poe mutation-full - name: Preserve mutation outcomes if: ${{ always() }} uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 with: - name: mutation-outcomes + name: mutation-outcomes-${{ matrix.shard }} path: reports/mutation.json if-no-files-found: error @@ -95,6 +100,20 @@ jobs: runs-on: ubuntu-latest timeout-minutes: 5 steps: + - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + persist-credentials: false + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5 + with: + python-version: "3.12" + - uses: actions/download-artifact@37930b1c2abaa49bbe596cd826c3c89aef350131 # v7.0.0 + with: + pattern: mutation-outcomes-* + path: reports/shards + - name: Require every runtime module in one mutation shard + run: | + git ls-files -z 'src/volcano_sdk/*.py' > reports/shards/source.bin + python -m scripts.check_mutation_shards reports/shards reports/shards/source.bin 8 - name: Require every mandatory job to succeed env: RESULTS: ${{ toJSON(needs.*.result) }} diff --git a/maintainers/quality-policy.lock.json b/maintainers/quality-policy.lock.json index 8c96562..1c84ea9 100644 --- a/maintainers/quality-policy.lock.json +++ b/maintainers/quality-policy.lock.json @@ -415,7 +415,7 @@ "policy": "bash scripts/check_quality_policy.sh", "quality": [ "checks", - "mutation" + "mutation-full" ], "test": { "interpreter": "bash", diff --git a/pyproject.toml b/pyproject.toml index 286c0e9..947daae 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -486,7 +486,7 @@ also_copy = [ ] [tool.poe.tasks] -quality = ["checks", "mutation"] +quality = ["checks", "mutation-full"] checks = ["policy", "audit", "generated", "lint", "format-check", "types", "test", "coverage", "contract-check", "package-check", "package-extras"] policy = "bash scripts/check_quality_policy.sh" build = "uv build --no-sources --require-hashes" diff --git a/scripts/check_mutation_shards.py b/scripts/check_mutation_shards.py new file mode 100644 index 0000000..4a74c95 --- /dev/null +++ b/scripts/check_mutation_shards.py @@ -0,0 +1,118 @@ +"""Require every runtime module in exactly one successful mutation shard.""" + +from __future__ import annotations + +import json +import os +import sys +from collections import Counter +from pathlib import Path +from typing import cast + + +class MutationShardError(Exception): + """A required mutation shard or runtime module is missing or failed.""" + + +def source_modules(path: Path) -> set[str]: + """Read the tracked handwritten runtime inventory. + + Returns: + Runtime module paths. + + """ + return { + os.fsdecode(raw) + for raw in path.read_bytes().split(b"\0") + if raw and not raw.startswith(b"src/volcano_sdk/_generated/") + } + + +def valid_outcomes(raw: object) -> bool: + """Accept only nonempty native killed or statically invalid outcomes. + + Returns: + Whether the result counts are complete and successful. + + """ + if not isinstance(raw, dict): + return False + counts = cast("dict[object, object]", raw) + if not counts or any( + name not in {"killed", "type_checked"} + or not isinstance(count, int) + or isinstance(count, bool) + or count < 0 + for name, count in counts.items() + ): + return False + validated_counts = cast("dict[str, int]", counts) + return sum(validated_counts.values()) > 0 + + +def shard_modules(path: Path) -> list[str]: + """Validate one native Mutmut result report. + + Returns: + Paths audited by this shard. + + Raises: + MutationShardError: The report is incomplete or contains failed mutants. + + """ + raw = cast("object", json.loads(path.read_text(encoding="utf-8"))) + if not isinstance(raw, dict): + msg = f"Invalid mutation report: {path}" + raise MutationShardError(msg) + report = cast("dict[str, object]", raw) + modules = report.get("modules") + outcomes = report.get("outcomes") + failures = report.get("failures") + if ( + not isinstance(modules, list) + or not modules + or not isinstance(failures, list) + or failures + ): + msg = f"Incomplete mutation report: {path}" + raise MutationShardError(msg) + typed_modules = cast("list[object]", modules) + if not all(isinstance(module, str) for module in typed_modules): + msg = f"Invalid mutation modules: {path}" + raise MutationShardError(msg) + if not valid_outcomes(outcomes): + msg = f"Failed mutation outcomes: {path}" + raise MutationShardError(msg) + return cast("list[str]", modules) + + +def main(root: Path, source_path: Path, count: int) -> int: + """Check shard presence, successful results, and complete source coverage. + + Returns: + Zero only when every runtime module appears exactly once. + + Raises: + MutationShardError: A shard or source module is missing or invalid. + + """ + expected = source_modules(source_path) + if not expected or count < 1: + msg = "Empty source inventory or invalid shard count" + raise MutationShardError(msg) + actual: Counter[str] = Counter() + for index in range(count): + report = root / f"mutation-outcomes-{index}" / "mutation.json" + actual.update(shard_modules(report)) + extra = sorted(set(actual) - expected) + missing = sorted(expected - set(actual)) + duplicate = sorted(name for name, occurrences in actual.items() if occurrences != 1) + if extra or missing or duplicate: + msg = f"Mutation shard inventory differs: {extra=}, {missing=}, {duplicate=}" + raise MutationShardError(msg) + print(f"Mutation shards covered {len(expected)} handwritten runtime modules") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main(Path(sys.argv[1]), Path(sys.argv[2]), int(sys.argv[3]))) diff --git a/scripts/check_quality_policy.py b/scripts/check_quality_policy.py index 6ad29db..ff04159 100644 --- a/scripts/check_quality_policy.py +++ b/scripts/check_quality_policy.py @@ -18,7 +18,7 @@ from collections.abc import Iterable GENERATED = "src/volcano_sdk/_generated" -LOCK_SHA256 = "557def9c362e95b993e315925f31ddab2cb4666acd6f97411aec64ec472b3f3d" +LOCK_SHA256 = "53609176e29f4e07237ff264b096232048243adfe181e6cb5011cf09d59f1a13" TYPE_FIXTURES = { "tests/typing/contract_steps.py", "tests/typing/durable_callbacks.py", diff --git a/scripts/mutation.sh b/scripts/mutation.sh index d05e3ae..c311236 100644 --- a/scripts/mutation.sh +++ b/scripts/mutation.sh @@ -51,11 +51,30 @@ else done < reports/mutation-changed.bin fi +if [[ -n ${MUTATION_SHARD_INDEX:-} ]]; then + if [[ ${MUTATION_FULL:-0} != 1 || ! ${MUTATION_SHARD_INDEX} =~ ^[0-9]+$ || ! ${MUTATION_SHARD_COUNT:-} =~ ^[0-9]+$ || ${MUTATION_SHARD_COUNT:-0} -eq 0 || ${MUTATION_SHARD_INDEX} -ge ${MUTATION_SHARD_COUNT} ]]; then + echo 'Invalid full-mutation shard configuration' >&2 + exit 1 + fi + all_modules=("${modules[@]}") + modules=() + for index in "${!all_modules[@]}"; do + if (( index % MUTATION_SHARD_COUNT == MUTATION_SHARD_INDEX )); then + modules+=("${all_modules[index]}") + fi + done +fi + +if (( ${#modules[@]} == 0 )); then + echo 'No runtime modules selected for mutation' >&2 + exit 1 +fi + for path in "${modules[@]}"; do printf '%s\0' "$path" >> "$targets" done -if [[ ${MUTATION_FULL:-0} == 1 ]]; then +if [[ ${MUTATION_FULL:-0} == 1 && -z ${MUTATION_SHARD_INDEX:-} ]]; then if ! mutmut run --max-children 1; then printf '%s\0' "full mutation run" >> "$failed" fi diff --git a/tests/unit/test_mutation_shards.py b/tests/unit/test_mutation_shards.py new file mode 100644 index 0000000..21a2d0b --- /dev/null +++ b/tests/unit/test_mutation_shards.py @@ -0,0 +1,81 @@ +"""A full mutation gate must account for every tracked runtime module.""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING + +import pytest + +from scripts.check_mutation_shards import MutationShardError, main + +if TYPE_CHECKING: + from pathlib import Path + + +def write_report( + root: Path, index: int, module: str, *, outcomes: dict[str, int] | None = None +) -> None: + """Create one native-result-shaped shard fixture.""" + report = root / f"mutation-outcomes-{index}" / "mutation.json" + report.parent.mkdir(parents=True) + _ = report.write_text( + json.dumps( + { + "modules": [module], + "outcomes": outcomes or {"killed": 1}, + "unmutatable_modules": [], + "failures": [], + } + ), + encoding="utf-8", + ) + + +def test_full_mutation_shards_cover_each_runtime_module_once(tmp_path: Path) -> None: + source = tmp_path / "source.bin" + _ = source.write_bytes( + b"src/volcano_sdk/auth.py\0src/volcano_sdk/realtime.py\0" + b"src/volcano_sdk/_generated/client.py\0" + ) + write_report(tmp_path, 0, "src/volcano_sdk/auth.py") + write_report(tmp_path, 1, "src/volcano_sdk/realtime.py") + + assert main(tmp_path, source, 2) == 0 + + +@pytest.mark.parametrize( + ("failure", "message"), + [ + ("missing", "mutation.json"), + ("uncovered", "Failed mutation outcomes"), + ("survived", "Failed mutation outcomes"), + ("timed_out", "Failed mutation outcomes"), + ("zero", "Failed mutation outcomes"), + ("duplicate", "inventory differs"), + ("omitted", "inventory differs"), + ], +) +def test_incomplete_or_failed_mutation_shards_fail( + tmp_path: Path, failure: str, message: str +) -> None: + source = tmp_path / "source.bin" + _ = source.write_bytes(b"src/volcano_sdk/auth.py\0src/volcano_sdk/realtime.py\0") + write_report(tmp_path, 0, "src/volcano_sdk/auth.py") + if failure not in {"missing", "omitted"}: + module = ( + "src/volcano_sdk/auth.py" + if failure == "duplicate" + else "src/volcano_sdk/realtime.py" + ) + outcomes = None + if failure == "zero": + outcomes = {"killed": 0} + elif failure in {"uncovered", "survived", "timed_out"}: + outcomes = {failure: 1} + write_report(tmp_path, 1, module, outcomes=outcomes) + elif failure == "omitted": + write_report(tmp_path, 1, "src/volcano_sdk/other.py") + + with pytest.raises((MutationShardError, FileNotFoundError), match=message): + _ = main(tmp_path, source, 2) From ae89a7a374c567e9c80faa7f95b5ec951e50a765 Mon Sep 17 00:00:00 2001 From: Sean Keever <33592180+swkeever@users.noreply.github.com> Date: Thu, 24 Sep 2026 06:48:53 -0400 Subject: [PATCH 03/11] fix(python): document and authorize mutation gate aggregation --- .github/workflows/ci.yml | 2 ++ maintainers/mutation-testing.md | 20 ++++++++++++-------- 2 files changed, 14 insertions(+), 8 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fd6b0f4..cf62e9c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -99,6 +99,8 @@ jobs: needs: [test, mutation] runs-on: ubuntu-latest timeout-minutes: 5 + permissions: + contents: read steps: - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 with: diff --git a/maintainers/mutation-testing.md b/maintainers/mutation-testing.md index db16609..90fbc42 100644 --- a/maintainers/mutation-testing.md +++ b/maintainers/mutation-testing.md @@ -1,16 +1,20 @@ # Mutation testing -`uv run --locked poe quality` runs all native checks and mutates every changed -handwritten runtime module plus the lock acquisition, guard, renewal, and worker -modules. CI uses the same `checks` and `mutation` tasks in separate jobs, then -requires both through `Quality Gate`. The weekly `poe mutation-full` task audits -every handwritten runtime module without a debt baseline. +`uv run --locked poe quality` runs all native checks and mutates every handwritten +runtime module. CI runs `checks` and divides `mutation-full` across eight jobs; +the required `Quality Gate` checks every job and the complete runtime inventory. +The weekly full audit also runs `mutation-full` without sharding. `poe mutation` +remains a faster local diagnostic for changed modules and the lock runtime. Mutmut's [native configuration](https://mutmut.readthedocs.io/en/latest/) lives in `pyproject.toml`; it excludes only the generated OpenAPI client. Mutmut can -select modules by name but has no Git-changed-module option and returns success -when mutants survive. `scripts/mutation.sh` selects module names from Git and -`scripts/mutation_results.py` reads only those modules' native metadata. The +select modules by name but cannot divide a run across CI jobs or verify that +their combined results cover every tracked source module. GitHub's matrix +reports job success without checking that source inventory. +`scripts/mutation.sh` assigns Git-tracked runtime modules to shards, +`scripts/mutation_results.py` reads each shard's native metadata, and +`scripts/check_mutation_shards.py` checks that all eight reports cover each +handwritten module exactly once. The report at `reports/mutation.json` distinguishes killed, statically invalid, surviving, uncovered, timed-out, crashed, interrupted, and missing results. A pytest internal error is a harness crash, not a killed mutant. From 0fa660e3b6730de45d4e7249cadb72bd73facbca Mon Sep 17 00:00:00 2001 From: Sean Keever <33592180+swkeever@users.noreply.github.com> Date: Thu, 24 Sep 2026 07:18:41 -0400 Subject: [PATCH 04/11] test(python): close URL and realtime mutation gaps --- src/volcano_sdk/_realtime_fetch_worker.py | 15 ++--- src/volcano_sdk/connection_string.py | 33 +++++----- tests/unit/test_realtime_fetch_lifecycle.py | 68 +++++++++++++++++++++ 3 files changed, 90 insertions(+), 26 deletions(-) diff --git a/src/volcano_sdk/_realtime_fetch_worker.py b/src/volcano_sdk/_realtime_fetch_worker.py index 421e73f..c3fe1e1 100644 --- a/src/volcano_sdk/_realtime_fetch_worker.py +++ b/src/volcano_sdk/_realtime_fetch_worker.py @@ -202,21 +202,20 @@ def _discard_pending(self) -> None: self._queue.task_done() async def _run(self) -> None: - pending: PostgresFetchJob[FallbackT] | _StopWorker | None = None + item = await self._queue.get() while True: - item = pending if pending is not None else await self._queue.get() - pending = None batch: list[PostgresFetchJob[FallbackT]] = [] try: if isinstance(item, _StopWorker): return batch.append(item) - pending = await self._collect_batch(batch) + next_item = await self._collect_batch(batch) await self._fetch_and_deliver(batch) finally: self._queue.task_done() for _job in batch[1:]: self._queue.task_done() + item = next_item if next_item is not None else await self._queue.get() async def _collect_batch( self, @@ -244,11 +243,8 @@ async def _next_before( self, deadline: float, ) -> PostgresFetchJob[FallbackT] | _StopWorker | None: - remaining = deadline - asyncio.get_running_loop().time() - if remaining <= 0: - return None try: - async with asyncio.timeout(remaining): + async with asyncio.timeout_at(deadline): return await self._queue.get() except TimeoutError: return None @@ -295,7 +291,8 @@ async def _fetch_and_deliver( return if len(result) != len(jobs): raise RuntimeError(_INVALID_RESULT_COUNT) - for job, record in zip(jobs, result, strict=True): + for index, job in enumerate(jobs): + record = result[index] await self._deliver(PostgresFetchOutcome(job=job, record=record)) async def _deliver_failure( diff --git a/src/volcano_sdk/connection_string.py b/src/volcano_sdk/connection_string.py index b69ad54..260745c 100644 --- a/src/volcano_sdk/connection_string.py +++ b/src/volcano_sdk/connection_string.py @@ -1,7 +1,7 @@ """Postgres connection helpers for Volcano functions.""" import re -from urllib.parse import quote, unquote +from urllib.parse import quote, unquote, urlencode _FULL_ACCESS_APP_NAME = "volcano_full_access" _USER_ACCESS_APP_NAME = "volcano_user_access" @@ -40,8 +40,11 @@ def database_connection_string( target, query = _connection_parts(base_connection_string) parameters = _query_parameters(query) - application_name = quote(_database_application_name(user_id), safe="") - parameters.append(f"application_name={application_name}") + parameters.append( + urlencode( + {"application_name": _database_application_name(user_id)}, quote_via=quote + ) + ) return f"{target}?{'&'.join(parameters)}" @@ -50,18 +53,15 @@ def _connection_parts(value: str) -> tuple[str, str]: if prefix is None or _INVALID_PERCENT_ENCODING.search(value): raise ValueError(_INVALID_ERROR) - authority_end = value.find("/", prefix.end()) - possible_userinfo_end = value.find("@", prefix.end()) - userinfo_end = ( - possible_userinfo_end - if possible_userinfo_end != -1 - and (authority_end == -1 or possible_userinfo_end < authority_end) - else -1 + authority, _, _ = value[prefix.end() :].partition("/") + userinfo, separator, _ = authority.partition("@") + search_from = ( + prefix.end() + len(userinfo) + len(separator) if separator else prefix.end() ) - query_start = value.find("?", max(prefix.end(), userinfo_end + 1)) - if query_start == -1: + before_query, marker, query = value[search_from:].partition("?") + if not marker: return value, "" - return value[:query_start], value[query_start + 1 :] + return value[: search_from + len(before_query)], query def _query_parameters(query: str) -> list[str]: @@ -72,10 +72,9 @@ def _query_parameters(query: str) -> list[str]: for parameter in query.split("&") if unquote(parameter.partition("=")[0]) != "application_name" ] - for index in range(len(parameters) - 1, -1, -1): - if parameters[index]: - return parameters[: index + 1] - return [] + while parameters and not parameters[-1]: + _ = parameters.pop() + return parameters def _database_application_name(user_id: str | None) -> str: diff --git a/tests/unit/test_realtime_fetch_lifecycle.py b/tests/unit/test_realtime_fetch_lifecycle.py index 9f06716..3d9d3d1 100644 --- a/tests/unit/test_realtime_fetch_lifecycle.py +++ b/tests/unit/test_realtime_fetch_lifecycle.py @@ -9,6 +9,7 @@ OutcomeRecorder, RecordingBatchFetch, fetch_job, + passthrough_job, ) from volcano_sdk._realtime_fetch_worker import PostgresFetchOutcome, PostgresFetchWorker @@ -211,6 +212,73 @@ async def deliver(outcome: PostgresFetchOutcome[str]) -> None: await worker.abort() +async def test_default_batch_window_flushes_before_another_row_arrives() -> None: + fetch = RecordingBatchFetch() + delivered = asyncio.Event() + + async def deliver(_outcome: PostgresFetchOutcome[str]) -> None: + delivered.set() + + worker = PostgresFetchWorker(fetch, deliver, queue_limit=2, max_batch_size=2) + try: + await worker.enqueue(fetch_job(1)) + _ = await asyncio.wait_for(delivered.wait(), timeout=0.5) + await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.close(), timeout=1) + + assert fetch.calls == [(1,), (2,)] + finally: + await worker.abort() + + +async def test_full_batch_flushes_before_close() -> None: + fetch = RecordingBatchFetch() + delivered = asyncio.Event() + outcomes: list[PostgresFetchOutcome[str]] = [] + + async def deliver(outcome: PostgresFetchOutcome[str]) -> None: + outcomes.append(outcome) + if len(outcomes) == 2: + delivered.set() + + worker = PostgresFetchWorker( + fetch, deliver, queue_limit=2, max_batch_size=2, batch_window_seconds=60 + ) + try: + await worker.enqueue(fetch_job(1)) + await worker.enqueue(fetch_job(2)) + _ = await asyncio.wait_for(delivered.wait(), timeout=1) + await asyncio.wait_for(worker.close(), timeout=1) + + assert fetch.calls == [(1, 2)] + assert outcomes == [ + PostgresFetchOutcome(job=fetch_job(1), record={"id": 1}), + PostgresFetchOutcome(job=fetch_job(2), record={"id": 2}), + ] + finally: + await worker.abort() + + +async def test_passthrough_does_not_enter_a_fetch_batch() -> None: + fetch = RecordingBatchFetch() + deliver = OutcomeRecorder() + worker = PostgresFetchWorker( + fetch, deliver, queue_limit=2, max_batch_size=2, batch_window_seconds=0.01 + ) + try: + await worker.enqueue(passthrough_job("full-payload")) + await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.close(), timeout=1) + + assert fetch.calls == [(1,)] + assert deliver.items == [ + PostgresFetchOutcome(job=passthrough_job("full-payload")), + PostgresFetchOutcome(job=fetch_job(1), record={"id": 1}), + ] + finally: + await worker.abort() + + @pytest.mark.parametrize(("window", "expected"), [(0, [(1,), (2,)]), (60, [(1, 2)])]) async def test_batch_capacity_and_zero_window_preserve_delivery_order( window: float, expected: list[tuple[int, ...]] From 40abe6699d0b46772c299c502cce881e55b374f3 Mon Sep 17 00:00:00 2001 From: Sean Keever <33592180+swkeever@users.noreply.github.com> Date: Thu, 24 Sep 2026 07:20:56 -0400 Subject: [PATCH 05/11] feat(python): type-check every handwritten mutation --- maintainers/mutation-testing.md | 3 ++- maintainers/quality-policy.lock.json | 13 +++---------- pyproject.toml | 12 ++---------- scripts/check_quality_policy.py | 2 +- 4 files changed, 8 insertions(+), 22 deletions(-) diff --git a/maintainers/mutation-testing.md b/maintainers/mutation-testing.md index 90fbc42..b1e2487 100644 --- a/maintainers/mutation-testing.md +++ b/maintainers/mutation-testing.md @@ -18,7 +18,8 @@ handwritten module exactly once. The report at `reports/mutation.json` distinguishes killed, statically invalid, surviving, uncovered, timed-out, crashed, interrupted, and missing results. A pytest internal error is a harness crash, not a killed mutant. -The pinned Pyrefly check rejects type-invalid realtime and auth mutants before pytest; +The pinned Pyrefly check rejects type-invalid mutants in every handwritten +runtime module before pytest; the report counts these as `type_checked`, separately from test-killed mutants. Surviving, uncovered, timed-out, crashed, and incomplete mutants still fail. Mutmut passes pytest `-x` so a selected test's first assertion failure kills the diff --git a/maintainers/quality-policy.lock.json b/maintainers/quality-policy.lock.json index 1c84ea9..22f43e6 100644 --- a/maintainers/quality-policy.lock.json +++ b/maintainers/quality-policy.lock.json @@ -299,16 +299,9 @@ "pyrefly", "check", "--output-format=json", - "src/volcano_sdk/realtime.py", - "src/volcano_sdk/auth.py", - "src/volcano_sdk/storage.py", - "src/volcano_sdk/functions.py", - "src/volcano_sdk/durable_authoring.py", - "src/volcano_sdk/database.py", - "src/volcano_sdk/_function_resolution.py", - "src/volcano_sdk/_transport.py", - "src/volcano_sdk/_session.py", - "src/volcano_sdk/_session_operations.py" + "--project-excludes", + "src/volcano_sdk/_generated/**", + "src/volcano_sdk/*.py" ] }, "mypy": { diff --git a/pyproject.toml b/pyproject.toml index 947daae..bdcb43c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -458,16 +458,8 @@ partial_branches = [] source_paths = ["src/volcano_sdk"] type_check_command = [ "pyrefly", "check", "--output-format=json", - "src/volcano_sdk/realtime.py", - "src/volcano_sdk/auth.py", - "src/volcano_sdk/storage.py", - "src/volcano_sdk/functions.py", - "src/volcano_sdk/durable_authoring.py", - "src/volcano_sdk/database.py", - "src/volcano_sdk/_function_resolution.py", - "src/volcano_sdk/_transport.py", - "src/volcano_sdk/_session.py", - "src/volcano_sdk/_session_operations.py", + "--project-excludes", "src/volcano_sdk/_generated/**", + "src/volcano_sdk/*.py", ] cache_invalidation_files = ["tests/**/*.py"] pytest_add_cli_args_test_selection = ["tests/unit"] diff --git a/scripts/check_quality_policy.py b/scripts/check_quality_policy.py index ff04159..9623b62 100644 --- a/scripts/check_quality_policy.py +++ b/scripts/check_quality_policy.py @@ -18,7 +18,7 @@ from collections.abc import Iterable GENERATED = "src/volcano_sdk/_generated" -LOCK_SHA256 = "53609176e29f4e07237ff264b096232048243adfe181e6cb5011cf09d59f1a13" +LOCK_SHA256 = "64c4fcc180777984cca637b2cbe41684b984ab8e33884da9aa6d0def1ddd048e" TYPE_FIXTURES = { "tests/typing/contract_steps.py", "tests/typing/durable_callbacks.py", From 9e1da0ed201f2804c459ab9601f95a9bd49b1a14 Mon Sep 17 00:00:00 2001 From: Sean Keever <33592180+swkeever@users.noreply.github.com> Date: Thu, 24 Sep 2026 07:48:56 -0400 Subject: [PATCH 06/11] ci(python): isolate full mutation modules across shards --- .github/workflows/ci.yml | 6 +++--- maintainers/mutation-testing.md | 7 ++++--- 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cf62e9c..3a95026 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -67,7 +67,7 @@ jobs: strategy: fail-fast: false matrix: - shard: [0, 1, 2, 3, 4, 5, 6, 7] + shard: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22] steps: - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 with: @@ -83,7 +83,7 @@ jobs: - name: Mutate all handwritten runtime modules in this shard env: MUTATION_SHARD_INDEX: ${{ matrix.shard }} - MUTATION_SHARD_COUNT: 8 + MUTATION_SHARD_COUNT: 23 run: uv run --locked poe mutation-full - name: Preserve mutation outcomes if: ${{ always() }} @@ -115,7 +115,7 @@ jobs: - name: Require every runtime module in one mutation shard run: | git ls-files -z 'src/volcano_sdk/*.py' > reports/shards/source.bin - python -m scripts.check_mutation_shards reports/shards reports/shards/source.bin 8 + python -m scripts.check_mutation_shards reports/shards reports/shards/source.bin 23 - name: Require every mandatory job to succeed env: RESULTS: ${{ toJSON(needs.*.result) }} diff --git a/maintainers/mutation-testing.md b/maintainers/mutation-testing.md index b1e2487..af2396c 100644 --- a/maintainers/mutation-testing.md +++ b/maintainers/mutation-testing.md @@ -13,9 +13,10 @@ their combined results cover every tracked source module. GitHub's matrix reports job success without checking that source inventory. `scripts/mutation.sh` assigns Git-tracked runtime modules to shards, `scripts/mutation_results.py` reads each shard's native metadata, and -`scripts/check_mutation_shards.py` checks that all eight reports cover each -handwritten module exactly once. The -report at `reports/mutation.json` distinguishes killed, statically invalid, +`scripts/check_mutation_shards.py` checks that all 23 reports cover each +handwritten module exactly once. The export-only `__init__.py` shares a shard +with a mutatable module. `reports/mutation.json` distinguishes killed, +statically invalid, surviving, uncovered, timed-out, crashed, interrupted, and missing results. A pytest internal error is a harness crash, not a killed mutant. The pinned Pyrefly check rejects type-invalid mutants in every handwritten From 56f55c4e3fe202008d8f828248ef36e522f5a987 Mon Sep 17 00:00:00 2001 From: Sean Keever <33592180+swkeever@users.noreply.github.com> Date: Thu, 24 Sep 2026 08:05:39 -0400 Subject: [PATCH 07/11] fix(python): validate mutation baseline and isolate selector test --- maintainers/mutation-testing.md | 2 ++ scripts/mutation.sh | 3 +++ tests/unit/test_mutation_results.py | 3 +++ 3 files changed, 8 insertions(+) diff --git a/maintainers/mutation-testing.md b/maintainers/mutation-testing.md index af2396c..42c1355 100644 --- a/maintainers/mutation-testing.md +++ b/maintainers/mutation-testing.md @@ -22,6 +22,8 @@ A pytest internal error is a harness crash, not a killed mutant. The pinned Pyrefly check rejects type-invalid mutants in every handwritten runtime module before pytest; the report counts these as `type_checked`, separately from test-killed mutants. +The runner checks the unmutated source first, so an existing type error cannot +make every mutant appear invalid. Surviving, uncovered, timed-out, crashed, and incomplete mutants still fail. Mutmut passes pytest `-x` so a selected test's first assertion failure kills the mutant before a later selected test can hang on the same defect. Mutants that diff --git a/scripts/mutation.sh b/scripts/mutation.sh index c311236..532cba7 100644 --- a/scripts/mutation.sh +++ b/scripts/mutation.sh @@ -4,6 +4,9 @@ set -euo pipefail # Forked macOS workers must not query SystemConfiguration through urllib/httpx. export NO_PROXY='*' no_proxy='*' +# An invalid baseline must not turn every mutant into a type-checked result. +pyrefly check --output-format=json --project-excludes 'src/volcano_sdk/_generated/**' 'src/volcano_sdk/*.py' + # Refresh mutmut's test-to-mutant map so newly added tests are selected. rm -rf -- mutants diff --git a/tests/unit/test_mutation_results.py b/tests/unit/test_mutation_results.py index 626a5c0..4236be5 100644 --- a/tests/unit/test_mutation_results.py +++ b/tests/unit/test_mutation_results.py @@ -38,6 +38,7 @@ def test_scoped_mutation_excludes_prefix_sibling_module( esac """, "mutmut": "#!/bin/sh\nprintf '%s\\n' \"$@\" > mutation-args.txt\n", + "pyrefly": "#!/bin/sh\nexit 0\n", "python": "#!/bin/sh\nexit 0\n", } for name, content in stubs.items(): @@ -45,6 +46,8 @@ def test_scoped_mutation_excludes_prefix_sibling_module( _ = stub.write_text(content, encoding="utf-8") stub.chmod(0o755) monkeypatch.setenv("PATH", f"{bin_dir}:{os.environ['PATH']}") + for name in ("MUTATION_FULL", "MUTATION_SHARD_INDEX", "MUTATION_SHARD_COUNT"): + monkeypatch.delenv(name, raising=False) result = subprocess.run( ["/bin/bash", "scripts/mutation.sh"], From bae3f2bde62995fa18c72045ced14c46f92348e8 Mon Sep 17 00:00:00 2001 From: Sean Keever <33592180+swkeever@users.noreply.github.com> Date: Thu, 24 Sep 2026 08:37:48 -0400 Subject: [PATCH 08/11] fix(python): close realtime queue races and preserve URL queries --- src/volcano_sdk/_realtime_fetch_worker.py | 41 ++-- src/volcano_sdk/connection_string.py | 10 +- tests/unit/test_connection_string.py | 21 ++ tests/unit/test_realtime_fetch_lifecycle.py | 229 +++++++++++++++++++- 4 files changed, 278 insertions(+), 23 deletions(-) diff --git a/src/volcano_sdk/_realtime_fetch_worker.py b/src/volcano_sdk/_realtime_fetch_worker.py index c3fe1e1..3df39a6 100644 --- a/src/volcano_sdk/_realtime_fetch_worker.py +++ b/src/volcano_sdk/_realtime_fetch_worker.py @@ -145,8 +145,7 @@ async def close(self) -> None: task = self._task if self._closed and (task is None or task.cancelled()): return - if not self._closed: - self._closed = True + self._closed = True self._raise_worker_failure() if task is not None and self._stop_task is None: self._stop_task = asyncio.create_task(self._queue.put(_STOP_WORKER)) @@ -157,15 +156,13 @@ async def close(self) -> None: async def abort(self) -> None: """Discard obsolete jobs and stop without waiting for row fetches.""" self._closed = True - task = self._task - stop_task = self._stop_task - if stop_task is not None and not stop_task.done(): - _ = stop_task.cancel() - if task is not None and not task.done(): - _ = task.cancel() pending = tuple( - candidate for candidate in (task, stop_task) if candidate is not None + candidate + for candidate in (self._stop_task, self._task) + if candidate is not None ) + for candidate in pending: + _ = candidate.cancel() if pending: _ = await asyncio.gather(*pending, return_exceptions=True) self._discard_pending() @@ -182,20 +179,36 @@ async def _put_while_running( ) -> None: put_task = asyncio.create_task(self._queue.put(job)) try: - completed, _pending = await asyncio.wait( - (task, put_task), - return_when=asyncio.FIRST_COMPLETED, - ) + await self._await_put_or_worker(task, put_task) finally: if not put_task.done(): _ = put_task.cancel() _ = await asyncio.gather(put_task, return_exceptions=True) - if task in completed: + if task.done(): if task.cancelled(): raise RuntimeError(_WORKER_CLOSED) task.result() await put_task + @staticmethod + async def _await_put_or_worker( + task: asyncio.Task[None], + put_task: asyncio.Task[None], + ) -> None: + ready: asyncio.Future[None] = asyncio.get_running_loop().create_future() + + def notify(_completed: asyncio.Task[None]) -> None: + if not ready.done(): + ready.set_result(None) + + task.add_done_callback(notify) + put_task.add_done_callback(notify) + try: + await ready + finally: + _ = task.remove_done_callback(notify) + _ = put_task.remove_done_callback(notify) + def _discard_pending(self) -> None: while not self._queue.empty(): _ = self._queue.get_nowait() diff --git a/src/volcano_sdk/connection_string.py b/src/volcano_sdk/connection_string.py index 260745c..ab332d0 100644 --- a/src/volcano_sdk/connection_string.py +++ b/src/volcano_sdk/connection_string.py @@ -65,16 +65,12 @@ def _connection_parts(value: str) -> tuple[str, str]: def _query_parameters(query: str) -> list[str]: - if not query: - return [] - parameters = [ + kept = "&".join( parameter for parameter in query.split("&") if unquote(parameter.partition("=")[0]) != "application_name" - ] - while parameters and not parameters[-1]: - _ = parameters.pop() - return parameters + ).rstrip("&") + return kept.split("&") if kept else [] def _database_application_name(user_id: str | None) -> str: diff --git a/tests/unit/test_connection_string.py b/tests/unit/test_connection_string.py index 3ad56ef..b5c4192 100644 --- a/tests/unit/test_connection_string.py +++ b/tests/unit/test_connection_string.py @@ -77,6 +77,27 @@ def test_database_connection_string_drops_trailing_query_separator() -> None: ) +def test_database_connection_string_preserves_trailing_x_in_query_value() -> None: + assert database_connection_string("postgresql://host/db?label=X") == ( + "postgresql://host/db?label=X&application_name=volcano_full_access" + ) + + +def test_database_connection_string_drops_empty_query_after_replaced_name() -> None: + assert ( + database_connection_string( + "postgresql://host/db?label=one&&application_name=old&" + ) + == "postgresql://host/db?label=one&application_name=volcano_full_access" + ) + + +def test_database_connection_string_replaces_name_without_a_path() -> None: + assert database_connection_string("postgres://host?application_name=old") == ( + "postgres://host?application_name=volcano_full_access" + ) + + def test_database_connection_string_drops_repeated_trailing_separators() -> None: assert database_connection_string("postgresql://host/db?sslmode=require&&&") == ( "postgresql://host/db?sslmode=require&application_name=volcano_full_access" diff --git a/tests/unit/test_realtime_fetch_lifecycle.py b/tests/unit/test_realtime_fetch_lifecycle.py index 3d9d3d1..35769be 100644 --- a/tests/unit/test_realtime_fetch_lifecycle.py +++ b/tests/unit/test_realtime_fetch_lifecycle.py @@ -1,6 +1,8 @@ from __future__ import annotations import asyncio +import gc +import weakref from typing import TYPE_CHECKING import pytest @@ -11,8 +13,15 @@ fetch_job, passthrough_job, ) - -from volcano_sdk._realtime_fetch_worker import PostgresFetchOutcome, PostgresFetchWorker +from typing_extensions import override + +from volcano_sdk._realtime_fetch_worker import ( + PostgresFetchJob, + PostgresFetchOutcome, + PostgresFetchWorker, + _StopWorker, + _wait_for_close, +) if TYPE_CHECKING: from volcano_sdk.realtime import _PostgresFetchRequest @@ -24,6 +33,57 @@ async def cancel_operation(task: asyncio.Task[None] | None) -> None: _ = await asyncio.gather(task, return_exceptions=True) +class CancellationAwareQueue(asyncio.Queue[PostgresFetchJob[str] | _StopWorker]): + def __init__(self) -> None: + super().__init__(maxsize=1) + self.cancelled = asyncio.Event() + self.release = asyncio.Event() + self.pending_put: asyncio.Task[None] | None = None + + @override + async def put(self, item: PostgresFetchJob[str] | _StopWorker) -> None: + self.pending_put = asyncio.current_task() + try: + await super().put(item) + except asyncio.CancelledError: + self.cancelled.set() + _ = await self.release.wait() + raise + + +async def test_failed_close_waits_for_stop_task_cleanup() -> None: + failure = RuntimeError("worker failed") + cancellation_seen = asyncio.Event() + cleanup_release = asyncio.Event() + + async def fail() -> None: + raise failure + + async def stop() -> None: + try: + _ = await asyncio.Event().wait() + except asyncio.CancelledError: + cancellation_seen.set() + _ = await cleanup_release.wait() + raise + + worker_task = asyncio.create_task(fail()) + stop_task = asyncio.create_task(stop()) + waiter = asyncio.create_task(_wait_for_close(worker_task, stop_task)) + try: + _ = await asyncio.wait_for(cancellation_seen.wait(), timeout=1) + assert not waiter.done() + cleanup_release.set() + with pytest.raises(RuntimeError, match="worker failed") as caught: + await asyncio.wait_for(waiter, timeout=1) + assert caught.value is failure + assert stop_task.done() + finally: + cleanup_release.set() + await cancel_operation(waiter) + await cancel_operation(stop_task) + + @pytest.mark.parametrize( ("window", "size", "message"), [ @@ -51,6 +111,8 @@ async def test_unused_worker_can_close_and_abort_repeatedly() -> None: worker = PostgresFetchWorker(fetch, deliver, queue_limit=1) await worker.close() + assert worker._stop_task is None + assert worker._queue.empty() await worker.close() await worker.abort() await worker.abort() @@ -61,6 +123,168 @@ async def test_unused_worker_can_close_and_abort_repeatedly() -> None: assert deliver.items == [] +async def test_closed_worker_rejects_new_jobs_before_abort() -> None: + worker = PostgresFetchWorker( + RecordingBatchFetch(), OutcomeRecorder(), queue_limit=1 + ) + await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.close(), timeout=1) + + with pytest.raises(RuntimeError, match="fetch worker is closed"): + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) + + +async def test_abort_cancels_an_active_fetch() -> None: + fetch = BlockingRowFetch() + worker = PostgresFetchWorker(fetch, OutcomeRecorder(), queue_limit=1) + aborting: asyncio.Task[None] | None = None + try: + await worker.enqueue(fetch_job(1)) + _ = await asyncio.wait_for(fetch.started.wait(), timeout=1) + aborting = asyncio.create_task(worker.abort()) + _ = await asyncio.wait_for(fetch.cancelled.wait(), timeout=0.5) + await asyncio.wait_for(aborting, timeout=1) + finally: + fetch.release.set() + await cancel_operation(aborting) + await worker.abort() + + +async def test_cancelled_enqueue_waits_for_its_queue_put_to_finish() -> None: + fetch = BlockingRowFetch() + worker = PostgresFetchWorker(fetch, OutcomeRecorder(), queue_limit=1) + queue = CancellationAwareQueue() + worker._queue = queue + enqueueing: asyncio.Task[None] | None = None + try: + await worker.enqueue(fetch_job(1)) + _ = await asyncio.wait_for(fetch.started.wait(), timeout=1) + await worker.enqueue(fetch_job(2)) + enqueueing = asyncio.create_task(worker.enqueue(fetch_job(3))) + await asyncio.sleep(0) + assert not enqueueing.done() + + _ = enqueueing.cancel() + _ = await asyncio.wait_for(queue.cancelled.wait(), timeout=0.5) + assert not enqueueing.done() + queue.release.set() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(enqueueing, timeout=1) + finally: + queue.release.set() + await cancel_operation(queue.pending_put) + await cancel_operation(enqueueing) + await worker.abort() + + +async def test_enqueue_rejects_cancellation_after_queue_put_completes( + monkeypatch: pytest.MonkeyPatch, +) -> None: + worker = PostgresFetchWorker( + RecordingBatchFetch(), OutcomeRecorder(), queue_limit=1 + ) + + async def wait_forever() -> None: + _ = await asyncio.Event().wait() + + task = asyncio.create_task(wait_forever()) + + async def cancel_after_put( + worker_task: asyncio.Task[None], put_task: asyncio.Task[None] + ) -> None: + await put_task + _ = worker_task.cancel() + _ = await asyncio.gather(worker_task, return_exceptions=True) + + monkeypatch.setattr(worker, "_await_put_or_worker", cancel_after_put) + try: + with pytest.raises(RuntimeError, match="fetch worker is closed"): + await worker._put_while_running(fetch_job(1), task) + finally: + await cancel_operation(task) + + +async def test_enqueues_release_completion_callbacks( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fetch = BlockingRowFetch() + worker = PostgresFetchWorker(fetch, OutcomeRecorder(), queue_limit=1) + loop = asyncio.get_running_loop() + create_future = loop.create_future + created: list[weakref.ReferenceType[asyncio.Future[None]]] = [] + + def track_future() -> asyncio.Future[None]: + future: asyncio.Future[None] = create_future() + created.append(weakref.ref(future)) + return future + + try: + await worker.enqueue(fetch_job(1)) + _ = await asyncio.wait_for(fetch.started.wait(), timeout=1) + with monkeypatch.context() as patch: + patch.setattr(loop, "create_future", track_future) + await worker.enqueue(fetch_job(2)) + + await asyncio.sleep(0) + _ = gc.collect() + assert len(created) == 1 + assert created[0]() is None + finally: + await worker.abort() + + +async def test_race_helper_handles_two_already_completed_tasks() -> None: + async def complete() -> None: + return + + task = asyncio.create_task(complete()) + put_task = asyncio.create_task(complete()) + _ = await asyncio.gather(task, put_task) + + await asyncio.wait_for( + PostgresFetchWorker._await_put_or_worker(task, put_task), timeout=1 + ) + + +async def test_first_enqueue_finishes_before_worker_shutdown() -> None: + worker = PostgresFetchWorker( + RecordingBatchFetch(), OutcomeRecorder(), queue_limit=1 + ) + try: + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=0.5) + assert worker._task is not None + assert not worker._task.done() + finally: + await worker.abort() + + +async def test_known_worker_failure_does_not_schedule_a_stop_request() -> None: + failure = RuntimeError("delivery failed") + delivery_started = asyncio.Event() + release_delivery = asyncio.Event() + + async def fail_delivery(_outcome: PostgresFetchOutcome[str]) -> None: + delivery_started.set() + _ = await release_delivery.wait() + raise failure + + worker = PostgresFetchWorker(RecordingBatchFetch(), fail_delivery, queue_limit=1) + try: + await worker.enqueue(fetch_job(1)) + _ = await asyncio.wait_for(delivery_started.wait(), timeout=1) + release_delivery.set() + assert worker._task is not None + _ = await asyncio.gather(worker._task, return_exceptions=True) + + with pytest.raises(RuntimeError, match="delivery failed") as caught: + await asyncio.wait_for(worker.close(), timeout=1) + assert caught.value is failure + assert worker._stop_task is None + finally: + release_delivery.set() + await worker.abort() + + async def test_close_failure_cancels_a_blocked_stop_request() -> None: fetch = BlockingRowFetch() failure = RuntimeError("delivery failed") @@ -249,6 +473,7 @@ async def deliver(outcome: PostgresFetchOutcome[str]) -> None: await worker.enqueue(fetch_job(2)) _ = await asyncio.wait_for(delivered.wait(), timeout=1) await asyncio.wait_for(worker.close(), timeout=1) + await asyncio.wait_for(worker._queue.join(), timeout=1) assert fetch.calls == [(1, 2)] assert outcomes == [ From 06a6b1b346a6e22d168c6741e6884a80ed35ebe7 Mon Sep 17 00:00:00 2001 From: Sean Keever <33592180+swkeever@users.noreply.github.com> Date: Thu, 24 Sep 2026 08:58:27 -0400 Subject: [PATCH 09/11] test(python): fail mutation defects before harness timeouts --- src/volcano_sdk/_realtime_fetch_worker.py | 2 +- src/volcano_sdk/connection_string.py | 14 ++- tests/unit/test_client_session_boundaries.py | 1 + tests/unit/test_connection_string.py | 12 +++ tests/unit/test_database_refresh.py | 5 ++ .../unit/test_realtime_delivery_boundaries.py | 2 +- tests/unit/test_realtime_fetch_lifecycle.py | 89 ++++++++++--------- tests/unit/test_realtime_fetch_worker.py | 45 +++++----- 8 files changed, 96 insertions(+), 74 deletions(-) diff --git a/src/volcano_sdk/_realtime_fetch_worker.py b/src/volcano_sdk/_realtime_fetch_worker.py index 3df39a6..6ebdb9b 100644 --- a/src/volcano_sdk/_realtime_fetch_worker.py +++ b/src/volcano_sdk/_realtime_fetch_worker.py @@ -206,8 +206,8 @@ def notify(_completed: asyncio.Task[None]) -> None: try: await ready finally: + # The worker may outlive this enqueue; the put task cannot. _ = task.remove_done_callback(notify) - _ = put_task.remove_done_callback(notify) def _discard_pending(self) -> None: while not self._queue.empty(): diff --git a/src/volcano_sdk/connection_string.py b/src/volcano_sdk/connection_string.py index ab332d0..e276476 100644 --- a/src/volcano_sdk/connection_string.py +++ b/src/volcano_sdk/connection_string.py @@ -40,12 +40,11 @@ def database_connection_string( target, query = _connection_parts(base_connection_string) parameters = _query_parameters(query) - parameters.append( - urlencode( - {"application_name": _database_application_name(user_id)}, quote_via=quote - ) + application = urlencode( + {"application_name": _database_application_name(user_id)}, quote_via=quote ) - return f"{target}?{'&'.join(parameters)}" + separator = "&" if parameters else "" + return f"{target}?{parameters}{separator}{application}" def _connection_parts(value: str) -> tuple[str, str]: @@ -64,13 +63,12 @@ def _connection_parts(value: str) -> tuple[str, str]: return value[: search_from + len(before_query)], query -def _query_parameters(query: str) -> list[str]: - kept = "&".join( +def _query_parameters(query: str) -> str: + return "&".join( parameter for parameter in query.split("&") if unquote(parameter.partition("=")[0]) != "application_name" ).rstrip("&") - return kept.split("&") if kept else [] def _database_application_name(user_id: str | None) -> str: diff --git a/tests/unit/test_client_session_boundaries.py b/tests/unit/test_client_session_boundaries.py index 41bf080..77dd440 100644 --- a/tests/unit/test_client_session_boundaries.py +++ b/tests/unit/test_client_session_boundaries.py @@ -96,6 +96,7 @@ def test_callback_dispatch_state_has_boolean_ownership_and_empty_failure() -> No assert client._dispatching_auth_notifications is False +@pytest.mark.order(0) def test_unsubscribe_releases_callback_ownership() -> None: client = VolcanoClient(anon_key="anon") diff --git a/tests/unit/test_connection_string.py b/tests/unit/test_connection_string.py index b5c4192..adc7887 100644 --- a/tests/unit/test_connection_string.py +++ b/tests/unit/test_connection_string.py @@ -27,6 +27,18 @@ def test_database_connection_string_preserves_unrelated_query_encoding() -> None ) +def test_database_connection_string_keeps_multiple_unrelated_parameters() -> None: + base = ( + "postgresql://db.example.com/app?" + "sslmode=require&connect_timeout=5&application_name=old" + ) + + assert database_connection_string(base) == ( + "postgresql://db.example.com/app?" + "sslmode=require&connect_timeout=5&application_name=volcano_full_access" + ) + + def test_database_connection_string_selects_encoded_user_access() -> None: assert database_connection_string( "postgres://user:password@db.example.com/app", diff --git a/tests/unit/test_database_refresh.py b/tests/unit/test_database_refresh.py index d409bad..544cbb0 100644 --- a/tests/unit/test_database_refresh.py +++ b/tests/unit/test_database_refresh.py @@ -356,12 +356,16 @@ def on_auth_change(event: str, _session: Session | None) -> None: def test_refresh_listener_can_wait_for_another_refresh_thread() -> None: completed: list[bool] = [] workers: list[Thread] = [] + notifications: list[str] = [] client = make_client(handle_expired_read) def on_refresh(event: str, _session: Session | None) -> None: if event != "TOKEN_REFRESHED": return + notifications.append(event) + if len(notifications) > 1: + return subscription.unsubscribe() worker = Thread(target=client.auth.refresh_session) workers.append(worker) @@ -374,6 +378,7 @@ def on_refresh(event: str, _session: Session | None) -> None: for worker in workers: worker.join(timeout=5) assert completed == [True] + assert notifications == ["TOKEN_REFRESHED"] @pytest.mark.parametrize("replace_at", ["replay", "failed-refresh-listener"]) diff --git a/tests/unit/test_realtime_delivery_boundaries.py b/tests/unit/test_realtime_delivery_boundaries.py index 098de98..612e5e1 100644 --- a/tests/unit/test_realtime_delivery_boundaries.py +++ b/tests/unit/test_realtime_delivery_boundaries.py @@ -391,7 +391,7 @@ async def test_closed_postgres_worker_reports_only_current_delivery_failures( enqueue = worker.enqueue async def stop_before_enqueue(job: PostgresFetchJob[_PostgresDelivery]) -> None: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) if unsubscribe: await channel.unsubscribe() await enqueue(job) diff --git a/tests/unit/test_realtime_fetch_lifecycle.py b/tests/unit/test_realtime_fetch_lifecycle.py index 35769be..1c1db66 100644 --- a/tests/unit/test_realtime_fetch_lifecycle.py +++ b/tests/unit/test_realtime_fetch_lifecycle.py @@ -114,11 +114,11 @@ async def test_unused_worker_can_close_and_abort_repeatedly() -> None: assert worker._stop_task is None assert worker._queue.empty() await worker.close() - await worker.abort() - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) + await asyncio.wait_for(worker.abort(), timeout=1) with pytest.raises(RuntimeError, match="fetch worker is closed"): - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) assert fetch.calls == [] assert deliver.items == [] @@ -127,7 +127,7 @@ async def test_closed_worker_rejects_new_jobs_before_abort() -> None: worker = PostgresFetchWorker( RecordingBatchFetch(), OutcomeRecorder(), queue_limit=1 ) - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) await asyncio.wait_for(worker.close(), timeout=1) with pytest.raises(RuntimeError, match="fetch worker is closed"): @@ -139,7 +139,7 @@ async def test_abort_cancels_an_active_fetch() -> None: worker = PostgresFetchWorker(fetch, OutcomeRecorder(), queue_limit=1) aborting: asyncio.Task[None] | None = None try: - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(fetch.started.wait(), timeout=1) aborting = asyncio.create_task(worker.abort()) _ = await asyncio.wait_for(fetch.cancelled.wait(), timeout=0.5) @@ -147,7 +147,7 @@ async def test_abort_cancels_an_active_fetch() -> None: finally: fetch.release.set() await cancel_operation(aborting) - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) async def test_cancelled_enqueue_waits_for_its_queue_put_to_finish() -> None: @@ -157,9 +157,9 @@ async def test_cancelled_enqueue_waits_for_its_queue_put_to_finish() -> None: worker._queue = queue enqueueing: asyncio.Task[None] | None = None try: - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(fetch.started.wait(), timeout=1) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) enqueueing = asyncio.create_task(worker.enqueue(fetch_job(3))) await asyncio.sleep(0) assert not enqueueing.done() @@ -174,7 +174,7 @@ async def test_cancelled_enqueue_waits_for_its_queue_put_to_finish() -> None: queue.release.set() await cancel_operation(queue.pending_put) await cancel_operation(enqueueing) - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) async def test_enqueue_rejects_cancellation_after_queue_put_completes( @@ -219,20 +219,21 @@ def track_future() -> asyncio.Future[None]: return future try: - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(fetch.started.wait(), timeout=1) with monkeypatch.context() as patch: patch.setattr(loop, "create_future", track_future) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) await asyncio.sleep(0) _ = gc.collect() assert len(created) == 1 assert created[0]() is None finally: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) +@pytest.mark.order(0) async def test_race_helper_handles_two_already_completed_tasks() -> None: async def complete() -> None: return @@ -255,7 +256,7 @@ async def test_first_enqueue_finishes_before_worker_shutdown() -> None: assert worker._task is not None assert not worker._task.done() finally: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) async def test_known_worker_failure_does_not_schedule_a_stop_request() -> None: @@ -270,7 +271,7 @@ async def fail_delivery(_outcome: PostgresFetchOutcome[str]) -> None: worker = PostgresFetchWorker(RecordingBatchFetch(), fail_delivery, queue_limit=1) try: - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(delivery_started.wait(), timeout=1) release_delivery.set() assert worker._task is not None @@ -282,7 +283,7 @@ async def fail_delivery(_outcome: PostgresFetchOutcome[str]) -> None: assert worker._stop_task is None finally: release_delivery.set() - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) async def test_close_failure_cancels_a_blocked_stop_request() -> None: @@ -295,9 +296,9 @@ async def fail_delivery(_outcome: PostgresFetchOutcome[str]) -> None: worker = PostgresFetchWorker(fetch, fail_delivery, queue_limit=1) closing = None try: - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(fetch.started.wait(), timeout=1) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) closing = asyncio.create_task(worker.close()) await asyncio.sleep(0) stop_task = worker._stop_task @@ -312,7 +313,7 @@ async def fail_delivery(_outcome: PostgresFetchOutcome[str]) -> None: assert stop_task.cancelled() assert fetch.row_ids == [1] finally: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) await cancel_operation(closing) @@ -322,9 +323,9 @@ async def test_abort_unblocks_close_with_a_full_queue() -> None: worker = PostgresFetchWorker(fetch, deliver, queue_limit=1) closing = None try: - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(fetch.started.wait(), timeout=1) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) closing = asyncio.create_task(worker.close()) await asyncio.sleep(0) stop_task = worker._stop_task @@ -340,7 +341,7 @@ async def test_abort_unblocks_close_with_a_full_queue() -> None: assert fetch.cancelled.is_set() assert deliver.items == [] finally: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) await cancel_operation(closing) @@ -350,9 +351,9 @@ async def test_abort_rejects_an_enqueue_waiting_for_capacity() -> None: worker = PostgresFetchWorker(fetch, deliver, queue_limit=1) enqueueing = None try: - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(fetch.started.wait(), timeout=1) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) enqueueing = asyncio.create_task(worker.enqueue(fetch_job(3))) await asyncio.sleep(0) assert not enqueueing.done() @@ -364,7 +365,7 @@ async def test_abort_rejects_an_enqueue_waiting_for_capacity() -> None: assert fetch.cancelled.is_set() assert deliver.items == [] finally: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) await cancel_operation(enqueueing) @@ -380,13 +381,13 @@ async def no_results( deliver = OutcomeRecorder() worker = PostgresFetchWorker(no_results, deliver, queue_limit=1) try: - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) release.set() with pytest.raises(RuntimeError, match="unexpected result count"): await asyncio.wait_for(worker.close(), timeout=1) assert deliver.items == [] finally: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) async def test_fetch_cancellation_propagates_without_delivering_a_fallback() -> None: @@ -403,7 +404,7 @@ async def cancelled_fetch( deliver = OutcomeRecorder() worker = PostgresFetchWorker(cancelled_fetch, deliver, queue_limit=1) try: - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(started.wait(), timeout=1) release.set() @@ -412,7 +413,7 @@ async def cancelled_fetch( assert deliver.items == [] finally: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) async def test_batch_window_flushes_without_waiting_for_another_row_or_close() -> None: @@ -427,13 +428,13 @@ async def deliver(outcome: PostgresFetchOutcome[str]) -> None: fetch, deliver, queue_limit=2, max_batch_size=2, batch_window_seconds=0.1 ) try: - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(delivered.wait(), timeout=1) await asyncio.wait_for(worker.close(), timeout=1) assert fetch.calls == [(1,)] finally: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) async def test_default_batch_window_flushes_before_another_row_arrives() -> None: @@ -445,14 +446,14 @@ async def deliver(_outcome: PostgresFetchOutcome[str]) -> None: worker = PostgresFetchWorker(fetch, deliver, queue_limit=2, max_batch_size=2) try: - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(delivered.wait(), timeout=0.5) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) await asyncio.wait_for(worker.close(), timeout=1) assert fetch.calls == [(1,), (2,)] finally: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) async def test_full_batch_flushes_before_close() -> None: @@ -469,8 +470,8 @@ async def deliver(outcome: PostgresFetchOutcome[str]) -> None: fetch, deliver, queue_limit=2, max_batch_size=2, batch_window_seconds=60 ) try: - await worker.enqueue(fetch_job(1)) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) _ = await asyncio.wait_for(delivered.wait(), timeout=1) await asyncio.wait_for(worker.close(), timeout=1) await asyncio.wait_for(worker._queue.join(), timeout=1) @@ -481,7 +482,7 @@ async def deliver(outcome: PostgresFetchOutcome[str]) -> None: PostgresFetchOutcome(job=fetch_job(2), record={"id": 2}), ] finally: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) async def test_passthrough_does_not_enter_a_fetch_batch() -> None: @@ -491,8 +492,10 @@ async def test_passthrough_does_not_enter_a_fetch_batch() -> None: fetch, deliver, queue_limit=2, max_batch_size=2, batch_window_seconds=0.01 ) try: - await worker.enqueue(passthrough_job("full-payload")) - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for( + worker.enqueue(passthrough_job("full-payload")), timeout=1 + ) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) await asyncio.wait_for(worker.close(), timeout=1) assert fetch.calls == [(1,)] @@ -501,7 +504,7 @@ async def test_passthrough_does_not_enter_a_fetch_batch() -> None: PostgresFetchOutcome(job=fetch_job(1), record={"id": 1}), ] finally: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) @pytest.mark.parametrize(("window", "expected"), [(0, [(1,), (2,)]), (60, [(1, 2)])]) @@ -514,12 +517,12 @@ async def test_batch_capacity_and_zero_window_preserve_delivery_order( fetch, deliver, queue_limit=2, max_batch_size=2, batch_window_seconds=window ) try: - await worker.enqueue(fetch_job(1)) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) await asyncio.wait_for(worker.close(), timeout=1) - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) assert fetch.calls == expected assert [item.record for item in deliver.items] == [{"id": 1}, {"id": 2}] finally: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) diff --git a/tests/unit/test_realtime_fetch_worker.py b/tests/unit/test_realtime_fetch_worker.py index a7effb9..4e3f66d 100644 --- a/tests/unit/test_realtime_fetch_worker.py +++ b/tests/unit/test_realtime_fetch_worker.py @@ -103,9 +103,9 @@ async def scenario() -> None: outcomes = deliver.items worker = PostgresFetchWorker(fetch, deliver, queue_limit=1) - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(fetch.started.wait(), timeout=0.2) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) third_enqueue = asyncio.create_task(worker.enqueue(fetch_job(3))) await asyncio.sleep(0) @@ -143,9 +143,9 @@ async def scenario() -> None: batch_window_seconds=0.05, max_batch_size=3, ) - await worker.enqueue(fetch_job(1)) - await worker.enqueue(fetch_job(2)) - await worker.enqueue(fetch_job(3)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) + await asyncio.wait_for(worker.enqueue(fetch_job(3)), timeout=1) await asyncio.wait_for(worker.close(), timeout=0.2) assert fetch.calls == [(1, 2, 3)] @@ -176,8 +176,8 @@ async def scenario() -> None: batch_window_seconds=0.01, max_batch_size=2, ) - await worker.enqueue(fetch_job(1)) - await worker.enqueue(fetch_job(2, table="archive")) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) + await asyncio.wait_for(worker.enqueue(fetch_job(2, table="archive")), timeout=1) await asyncio.wait_for(worker.close(), timeout=0.2) assert fetch.calls == [(1,), (2,)] @@ -202,8 +202,8 @@ async def scenario() -> None: batch_window_seconds=0.01, max_batch_size=2, ) - await worker.enqueue(fetch_job(1)) - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) await asyncio.wait_for(worker.close(), timeout=0.2) assert fetch.calls == [(1,), (1,)] @@ -219,9 +219,11 @@ async def scenario() -> None: outcomes = deliver.items worker = PostgresFetchWorker(fetch, deliver, queue_limit=1) - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(fetch.started.wait(), timeout=0.2) - await worker.enqueue(passthrough_job("full-payload")) + await asyncio.wait_for( + worker.enqueue(passthrough_job("full-payload")), timeout=1 + ) assert outcomes == [] @@ -237,6 +239,7 @@ async def scenario() -> None: asyncio.run(scenario()) +@pytest.mark.order(0) def test_postgres_fetch_worker_aborts_obsolete_jobs_without_waiting() -> None: async def scenario() -> None: fetch = BlockingRowFetch() @@ -244,9 +247,9 @@ async def scenario() -> None: outcomes = deliver.items worker = PostgresFetchWorker(fetch, deliver, queue_limit=1) - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(fetch.started.wait(), timeout=0.2) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) await asyncio.wait_for(worker.abort(), timeout=0.2) @@ -274,8 +277,8 @@ async def scenario() -> None: deliver, queue_limit=2, ) - await worker.enqueue(fetch_job(1)) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) await asyncio.wait_for(worker.close(), timeout=0.2) assert outcomes == [ @@ -302,13 +305,13 @@ async def fail_delivery(_outcome: PostgresFetchOutcome[str]) -> None: fail_delivery, queue_limit=1, ) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) _ = await delivery_started.wait() release_delivery.set() await asyncio.sleep(0) with pytest.raises(RuntimeError) as raised: - await worker.enqueue(fetch_job(3)) + await asyncio.wait_for(worker.enqueue(fetch_job(3)), timeout=1) assert raised.value is failure @@ -322,9 +325,9 @@ async def scenario() -> None: outcomes = deliver.items worker = PostgresFetchWorker(fetch, deliver, queue_limit=1) - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(fetch.started.wait(), timeout=0.2) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) interrupted_close = asyncio.create_task(worker.close()) await asyncio.sleep(0) @@ -362,9 +365,9 @@ async def fail_delivery(_outcome: PostgresFetchOutcome[str]) -> None: fail_delivery, queue_limit=1, ) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) _ = await delivery_started.wait() - await worker.enqueue(fetch_job(3)) + await asyncio.wait_for(worker.enqueue(fetch_job(3)), timeout=1) blocked_enqueue = asyncio.create_task(worker.enqueue(fetch_job(4))) await asyncio.sleep(0) assert not blocked_enqueue.done() From c40e717323789d92beb7debead377a142c9c180e Mon Sep 17 00:00:00 2001 From: Sean Keever <33592180+swkeever@users.noreply.github.com> Date: Thu, 24 Sep 2026 09:05:01 -0400 Subject: [PATCH 10/11] test(python): isolate worker future count from timeout machinery --- tests/unit/test_realtime_fetch_lifecycle.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/unit/test_realtime_fetch_lifecycle.py b/tests/unit/test_realtime_fetch_lifecycle.py index 1c1db66..d6c532f 100644 --- a/tests/unit/test_realtime_fetch_lifecycle.py +++ b/tests/unit/test_realtime_fetch_lifecycle.py @@ -223,7 +223,8 @@ def track_future() -> asyncio.Future[None]: _ = await asyncio.wait_for(fetch.started.wait(), timeout=1) with monkeypatch.context() as patch: patch.setattr(loop, "create_future", track_future) - await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) + async with asyncio.timeout(1): + await worker.enqueue(fetch_job(2)) await asyncio.sleep(0) _ = gc.collect() From ed1a6d0dc28a527e0b78496876aca9722da5d4d7 Mon Sep 17 00:00:00 2001 From: Sean Keever <33592180+swkeever@users.noreply.github.com> Date: Thu, 24 Sep 2026 09:48:12 -0400 Subject: [PATCH 11/11] test(python): kill valid-batch mutation before timeout --- tests/unit/test_realtime_fetch_lifecycle.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/tests/unit/test_realtime_fetch_lifecycle.py b/tests/unit/test_realtime_fetch_lifecycle.py index d6c532f..a27ea58 100644 --- a/tests/unit/test_realtime_fetch_lifecycle.py +++ b/tests/unit/test_realtime_fetch_lifecycle.py @@ -51,6 +51,22 @@ async def put(self, item: PostgresFetchJob[str] | _StopWorker) -> None: raise +@pytest.mark.order(0) +async def test_successful_batch_delivers_each_result() -> None: + fetch = RecordingBatchFetch() + deliver = OutcomeRecorder() + worker = PostgresFetchWorker(fetch, deliver, queue_limit=2) + jobs = [fetch_job(1), fetch_job(2)] + + await asyncio.wait_for(worker._fetch_and_deliver(jobs), timeout=1) + + assert fetch.calls == [(1, 2)] + assert deliver.items == [ + PostgresFetchOutcome(job=jobs[0], record={"id": 1}), + PostgresFetchOutcome(job=jobs[1], record={"id": 2}), + ] + + async def test_failed_close_waits_for_stop_task_cleanup() -> None: failure = RuntimeError("worker failed") cancellation_seen = asyncio.Event()