diff --git a/docs/architecture/delivery-semantics.md b/docs/architecture/delivery-semantics.md index b4d6eba..82d3f92 100644 --- a/docs/architecture/delivery-semantics.md +++ b/docs/architecture/delivery-semantics.md @@ -117,8 +117,9 @@ It does not remove the window, and set too low it **creates** duplicates: another worker can claim a delivery that is still in flight. The lease must comfortably outlast `FORMS_WEBHOOK_CONNECT_TIMEOUT_SECONDS` and -`FORMS_WEBHOOK_READ_TIMEOUT_SECONDS` combined. The defaults leave a wide margin, -60 seconds against 15. See [Worker](../operations/worker.md). +`FORMS_WEBHOOK_READ_TIMEOUT_SECONDS` combined, which is the overall deadline every +attempt is held to. The defaults leave a wide margin, 60 seconds against 15. See +[Worker](../operations/worker.md). ## Two attempt counters diff --git a/docs/guides/idempotency.md b/docs/guides/idempotency.md index 48e8a47..1fddf42 100644 --- a/docs/guides/idempotency.md +++ b/docs/guides/idempotency.md @@ -41,7 +41,9 @@ winner, not a check in application code. See [Concurrency](../architecture/concurrency.md). A replay reports the same `delivery.queued` as the original and does **not** -queue a second delivery. One submission owes at most one delivery, enforced by a +queue a second delivery. It describes the stored submission, so adding or +removing the endpoint's webhook after the original request does not change what a +replay reports. One submission owes at most one delivery, enforced by a unique constraint on the submission. ## Conflict semantics diff --git a/docs/guides/webhooks.md b/docs/guides/webhooks.md index b19efc4..5785227 100644 --- a/docs/guides/webhooks.md +++ b/docs/guides/webhooks.md @@ -181,8 +181,15 @@ Do not enable it in production. | `FORMS_WEBHOOK_CONNECT_TIMEOUT_SECONDS` | `5` | Wait for the destination to accept a connection | | `FORMS_WEBHOOK_READ_TIMEOUT_SECONDS` | `10` | Wait for the destination to respond | -Both must stay comfortably below `FORMS_WORKER_LEASE_SECONDS`, or another worker -could claim a delivery that is still in flight. See [Worker](../operations/worker.md). +Each of these bounds a single wait, so on top of them every attempt has an overall +deadline of the two combined (15 seconds by default). A destination that answers +one byte at a time cannot hold a request open past it; such an attempt is +recorded as a `timeout` and retried. Only the response status is used: the +response body is never read. + +The deadline must stay comfortably below `FORMS_WORKER_LEASE_SECONDS`, or another +worker could claim a delivery that is still in flight. The worker logs a warning +at startup if it does not. See [Worker](../operations/worker.md). ## Related diff --git a/docs/operations/configuration-reference.md b/docs/operations/configuration-reference.md index 367a98a..29cc5a0 100644 --- a/docs/operations/configuration-reference.md +++ b/docs/operations/configuration-reference.md @@ -56,7 +56,7 @@ value. It must be at least 16 characters. | Variable | Default | Meaning | | --- | --- | --- | | `FORMS_WEBHOOK_CONNECT_TIMEOUT_SECONDS` | `5` | Wait for a webhook to accept a connection | -| `FORMS_WEBHOOK_READ_TIMEOUT_SECONDS` | `10` | Wait for a webhook to respond | +| `FORMS_WEBHOOK_READ_TIMEOUT_SECONDS` | `10` | Wait for a webhook to respond. With the connect timeout, also the overall deadline for one attempt | | `FORMS_WEBHOOK_MAX_ATTEMPTS` | `5` | Attempts before a delivery is given up on | | `FORMS_WEBHOOK_RETRY_INITIAL_SECONDS` | `10` | Wait before the second attempt; later waits double | | `FORMS_WEBHOOK_RETRY_MAX_SECONDS` | `3600` | Cap on the wait between attempts | diff --git a/docs/operations/worker.md b/docs/operations/worker.md index 35c6093..b70f7b8 100644 --- a/docs/operations/worker.md +++ b/docs/operations/worker.md @@ -49,11 +49,13 @@ There is no coordination service and no broker. PostgreSQL is the queue. See !!! danger "The lease must outlast a delivery attempt" - A batch is delivered concurrently, so it takes about as long as its slowest - single delivery rather than the sum. But if `FORMS_WORKER_LEASE_SECONDS` were - set below `FORMS_WEBHOOK_CONNECT_TIMEOUT_SECONDS` and - `FORMS_WEBHOOK_READ_TIMEOUT_SECONDS` combined, another worker could claim a - delivery that is still in flight and send it twice. + Every attempt has an overall deadline of + `FORMS_WEBHOOK_CONNECT_TIMEOUT_SECONDS` plus + `FORMS_WEBHOOK_READ_TIMEOUT_SECONDS`, however slowly the destination answers. + A batch is delivered concurrently, so it takes about as long as that deadline + at most rather than the sum. But if `FORMS_WORKER_LEASE_SECONDS` were set at + or below the deadline, another worker could claim a delivery that is still in + flight and send it twice. The worker logs a warning at startup when it is. The defaults leave a wide margin (60 seconds against 15). Keep it that way if you change them. @@ -75,8 +77,22 @@ the database is at migration '0003' but this build expects '0004'. Run 'alembic upgrade head' before starting. ``` -Migrating is an operator action, run when the operator chooses. See -[Database migrations](migrations.md). +The worker exits with status 1 rather than polling. Migrating is an operator +action, run when the operator chooses. See [Database migrations](migrations.md). + +## Stopping + +`SIGTERM` or `SIGINT` asks the worker to finish the batch it is on and exit. A +batch in flight ends within one attempt deadline, after which its outcomes are +recorded. A worker killed outright (`SIGKILL`, a crash, a lost host) leaves its +claimed deliveries in `processing`; they become claimable again once their lease +expires and are sent again, which is the at-least-once case described in +[Delivery semantics](../architecture/delivery-semantics.md). + +If recording one attempt fails, for example because the database connection +dropped, the worker logs it and goes on to record the rest of the batch. The +delivery it could not record stays claimed and is retried after its lease, so the +destination may receive it twice. ## Deploy order diff --git a/src/hymical_forms/api/submissions.py b/src/hymical_forms/api/submissions.py index da64f43..d104dac 100644 --- a/src/hymical_forms/api/submissions.py +++ b/src/hymical_forms/api/submissions.py @@ -467,7 +467,7 @@ async def submit(endpoint_id: str, request: Request, session: SessionDep) -> Sub received_at=stored.submission.received_at, field_count=stored.submission.field_count, idempotent_replay=stored.replayed, - delivery=DeliveryStatus(queued=webhook is not None), + delivery=DeliveryStatus(queued=stored.delivery_queued), ) diff --git a/src/hymical_forms/delivery.py b/src/hymical_forms/delivery.py index fec5ad0..ca6f237 100644 --- a/src/hymical_forms/delivery.py +++ b/src/hymical_forms/delivery.py @@ -9,6 +9,8 @@ from __future__ import annotations +import asyncio + import httpx2 from hymical_forms import __version__ @@ -53,8 +55,28 @@ def create_webhook_client(settings: Settings) -> httpx2.AsyncClient: ) +def attempt_deadline_seconds(settings: Settings) -> float: + """ + work out the longest one delivery attempt is allowed to take, start to finish + :param settings: active configuration, read for its timeouts + :returns: the overall deadline for one attempt, in seconds + """ + # The client's timeouts each bound a single wait: one connect, one read of + # whatever bytes arrive next. None of them bounds the attempt, so a + # destination that answers one byte at a time could hold a request open for + # as long as it liked, past any lease. Connecting plus one full read timeout + # is the most an attempt is worth waiting for, and it is the figure the + # operations guide tells operators to size the lease against. + return settings.webhook_connect_timeout_seconds + settings.webhook_read_timeout_seconds + + async def deliver( - client: httpx2.AsyncClient, *, url: str, secret: str, body: bytes + client: httpx2.AsyncClient, + *, + url: str, + secret: str, + body: bytes, + deadline_seconds: float, ) -> DeliveryResult: """ make one attempt to deliver a signed payload @@ -62,6 +84,7 @@ async def deliver( :param url: the destination to post to :param secret: the destination's signing secret :param body: the exact bytes to sign and transmit + :param deadline_seconds: the longest the whole attempt may take, see attempt_deadline_seconds :returns: the outcome, never raising for a destination that misbehaves """ # ``content=body`` transmits these bytes verbatim. Passing the payload object @@ -74,22 +97,34 @@ async def deliver( "User-Agent": USER_AGENT, } + # The response is streamed and its body never read. The status is the whole + # answer, and reading a body nobody looks at would let the destination decide + # how much memory an attempt takes and how long it lasts. try: - response = await client.post(url, content=body, headers=headers) + async with ( + asyncio.timeout(deadline_seconds), + client.stream("POST", url, content=body, headers=headers) as response, + ): + status = response.status_code except httpx2.TimeoutException as exc: return DeliveryResult(DeliveryOutcome.TIMEOUT, error=_describe("timed out", exc)) except httpx2.RequestError as exc: return DeliveryResult( DeliveryOutcome.NETWORK_ERROR, error=_describe("could not connect", exc) ) + except TimeoutError: + return DeliveryResult( + DeliveryOutcome.TIMEOUT, + error=f"timed out: no response within the {deadline_seconds:g}s attempt deadline", + ) - if 200 <= response.status_code < 300: - return DeliveryResult(DeliveryOutcome.SUCCEEDED, response_status=response.status_code) + if 200 <= status < 300: + return DeliveryResult(DeliveryOutcome.SUCCEEDED, response_status=status) return DeliveryResult( DeliveryOutcome.HTTP_ERROR, - response_status=response.status_code, - error=f"destination responded with HTTP {response.status_code}", + response_status=status, + error=f"destination responded with HTTP {status}", ) diff --git a/src/hymical_forms/storage.py b/src/hymical_forms/storage.py index 363b093..87c7a35 100644 --- a/src/hymical_forms/storage.py +++ b/src/hymical_forms/storage.py @@ -24,11 +24,10 @@ from collections.abc import Iterator from dataclasses import dataclass from datetime import datetime, timedelta -from typing import Any, TypeVar, cast +from typing import Any, Protocol, Self, TypeVar, cast from sqlalchemy import ( ColumnElement, - Select, and_, delete, func, @@ -67,7 +66,25 @@ # Any submission select, whatever it happens to be selecting. The submission # filters are applied to a listing, an export and a bounded count alike, and # narrowing a select never changes what it returns. -_Statement = TypeVar("_Statement", bound=Select[Any]) +_Statement = TypeVar("_Statement", bound="_Narrowable") + + +class _Narrowable(Protocol): + """ + a select of any shape, which is all the submission filters need from one + """ + + # A protocol rather than a bound on ``Select`` itself, because how a select's + # row type is spelled changed between SQLAlchemy 2.0 and 2.1 and no single + # ``Select[...]`` bound type checks against both. + def where(self, *whereclause: ColumnElement[bool]) -> Self: + """ + narrow the select by further conditions + :param whereclause: the conditions to add + :returns: the narrowed select + """ + ... + # How many rows a streamed export pulls from the server at a time. Bounded so # that an export builds its response as it goes rather than materialising every @@ -242,8 +259,12 @@ class StoredSubmission: the outcome of storing a submission """ + # ``delivery_queued`` describes the stored submission, not the endpoint as it + # is configured now. A replay answers for the submission the key was spent + # on, and the endpoint's webhook may have been added or removed since. submission: Submission replayed: bool + delivery_queued: bool def find_by_idempotency_key( @@ -292,12 +313,12 @@ def store_submission( if idempotency_key is None: _add_submission(session, submission, now=now, webhook=webhook) session.commit() - return StoredSubmission(submission, replayed=False) + return StoredSubmission(submission, replayed=False, delivery_queued=webhook is not None) # Fast path for the ordinary retry, where the first attempt already landed. existing = find_by_idempotency_key(session, submission.endpoint_id, idempotency_key) if existing is not None: - return StoredSubmission(_settle(existing, payload_fingerprint), replayed=True) + return _replayed(session, existing, payload_fingerprint) _add_submission( session, @@ -325,9 +346,27 @@ def store_submission( # The violation was something other than the idempotency constraint, # so it is not ours to interpret and must not be reported as success. raise - return StoredSubmission(_settle(existing, payload_fingerprint), replayed=True) + return _replayed(session, existing, payload_fingerprint) + + return StoredSubmission(submission, replayed=False, delivery_queued=webhook is not None) - return StoredSubmission(submission, replayed=False) + +def _replayed( + session: Session, existing: models.Submission, payload_fingerprint: str | None +) -> StoredSubmission: + """ + describe the earlier submission a retried request resolved to + :param session: the session to query through + :param existing: the submission the key was already spent on + :param payload_fingerprint: digest of the content submitted this time + :returns: the earlier submission, flagged as a replay, with the delivery it actually owes + :raises IdempotencyKeyReused: if the content differs from the earlier attempt + """ + settled = _settle(existing, payload_fingerprint) + owed = session.scalar( + select(models.WebhookDelivery.id).where(models.WebhookDelivery.submission_id == existing.id) + ) + return StoredSubmission(settled, replayed=True, delivery_queued=owed is not None) def _add_submission( diff --git a/src/hymical_forms/worker.py b/src/hymical_forms/worker.py index bea243d..dbb675f 100644 --- a/src/hymical_forms/worker.py +++ b/src/hymical_forms/worker.py @@ -15,6 +15,7 @@ import asyncio import logging import signal +import sys import threading from datetime import datetime from types import FrameType @@ -25,8 +26,9 @@ from hymical_forms import storage, webhooks from hymical_forms.config import Settings from hymical_forms.db import create_engine_from_url, create_session_factory -from hymical_forms.delivery import create_webhook_client, deliver +from hymical_forms.delivery import attempt_deadline_seconds, create_webhook_client, deliver from hymical_forms.models import utcnow +from hymical_forms.schema import SchemaNotReady, verify_schema logger = logging.getLogger(__name__) @@ -57,6 +59,14 @@ async def process_batch( submissions = storage.load_submissions(session, claimed) + # Loading the submissions opened a read transaction, and nothing else will + # end it before the requests below go out. Ending it here is what keeps the + # promise that no transaction is held across somebody else's server. + # Committing rather than rolling back, because a rollback would expire the + # claimed rows and send the recording step back to the database to reload + # them. + session.commit() + # The claim each request is about to be made under, captured before any of # them go out. Completing is fenced on this rather than on whatever the row # says once a response comes back, so a lease that expires mid-request cannot @@ -71,9 +81,16 @@ async def process_batch( job.id: webhooks.serialize_payload(webhooks.build_payload(submissions[job.id])) for job in claimed } + deadline = attempt_deadline_seconds(settings) results = await asyncio.gather( *( - deliver(client, url=job.destination_url, secret=job.signing_secret, body=bodies[job.id]) + deliver( + client, + url=job.destination_url, + secret=job.signing_secret, + body=bodies[job.id], + deadline_seconds=deadline, + ) for job in claimed ) ) @@ -83,9 +100,24 @@ async def process_batch( # would gain nothing real and would make every retry schedule approximate. policy = settings.retry_policy() for job, result in zip(claimed, results, strict=True): - recorded = storage.complete_attempt( - session, job, result, now=now, policy=policy, claim_token=claims[job.id] - ) + try: + recorded = storage.complete_attempt( + session, job, result, now=now, policy=policy, claim_token=claims[job.id] + ) + except Exception: + # Every request in the batch has already gone out. Giving up on the + # rest because one could not be recorded would leave them all to be + # sent again once their leases ran out. This one is left claimed, so + # it is retried after its lease, which is the at-least-once outcome a + # worker crashing here would have produced anyway. + session.rollback() + logger.exception( + "delivery %s attempt was made (%s) but could not be recorded, " + "it will be retried once its lease expires", + job.id, + result.outcome, + ) + continue if recorded.owned: logger.info( "delivery %s attempt %d %s (%s)", @@ -119,6 +151,25 @@ async def run_worker(settings: Settings, *, stop: threading.Event) -> None: :param stop: event that ends the loop once set """ engine = create_engine_from_url(settings.database_url) + try: + # The same check the API and the CLI make. A worker built for another + # revision would claim and complete deliveries under rules the schema no + # longer matches, which is worse than not delivering at all. + verify_schema(engine) + except BaseException: + engine.dispose() + raise + + deadline = attempt_deadline_seconds(settings) + if settings.worker_lease_seconds <= deadline: + logger.warning( + "FORMS_WORKER_LEASE_SECONDS (%.1fs) does not outlast one delivery attempt " + "(%.1fs, the connect and read timeouts combined). Another worker can " + "reclaim a delivery that is still in flight and send it twice.", + settings.worker_lease_seconds, + deadline, + ) + session_factory: sessionmaker[Session] = create_session_factory(engine) client = create_webhook_client(settings) @@ -168,7 +219,11 @@ def request_stop(signum: int, frame: FrameType | None) -> None: signal.signal(signal.SIGINT, request_stop) signal.signal(signal.SIGTERM, request_stop) - asyncio.run(run_worker(settings, stop=stop)) + try: + asyncio.run(run_worker(settings, stop=stop)) + except SchemaNotReady as exc: + logger.error("the database schema is not ready: %s", exc) + sys.exit(1) if __name__ == "__main__": diff --git a/tests/test_idempotency.py b/tests/test_idempotency.py index 5532807..bf2e7b0 100644 --- a/tests/test_idempotency.py +++ b/tests/test_idempotency.py @@ -175,6 +175,44 @@ def test_a_key_is_scoped_to_one_endpoint(empty_client: TestClient) -> None: assert {row.endpoint_id for row in stored(empty_client)} == {"contact-form", "waitlist"} +# --- what a replay reports about delivery ------------------------------------ + + +def test_a_replay_reports_no_delivery_when_the_webhook_was_added_since( + make_client: ClientFactory, +) -> None: + # The replay describes the submission the key was spent on. That one owes no + # delivery, so a webhook configured afterwards must not make it claim one. + client = make_client(allow_private_webhook_targets=True) + first = submit(client) + changed = client.patch( + "/endpoints/contact-form", json={"webhook_url": "https://hooks.example.com/forms"} + ) + assert changed.status_code == 200, changed.text + + replay = submit(client) + + assert first.json()["delivery"]["queued"] is False + assert replay.json()["idempotent_replay"] is True + assert replay.json()["delivery"]["queued"] is False + + +def test_a_replay_still_reports_its_delivery_after_the_webhook_is_removed( + make_client: ClientFactory, +) -> None: + client = make_client(seed_endpoint=False) + create_endpoint(client, webhook_url="https://hooks.example.com/forms") + first = submit(client) + changed = client.patch("/endpoints/contact-form", json={"webhook_url": None}) + assert changed.status_code == 200, changed.text + + replay = submit(client) + + assert first.json()["delivery"]["queued"] is True + assert replay.json()["idempotent_replay"] is True + assert replay.json()["delivery"]["queued"] is True + + # --- what counts as the same payload ----------------------------------------- diff --git a/tests/test_worker.py b/tests/test_worker.py index 5976d4c..7e8717c 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -8,20 +8,36 @@ from __future__ import annotations +import asyncio import hashlib import hmac import json +import logging +import threading +import time from datetime import UTC, datetime, timedelta +from pathlib import Path from typing import Any import pytest from fastapi.testclient import TestClient from sqlalchemy import select from sqlalchemy.dialects import postgresql - -from conftest import ClientFactory, app_settings, create_endpoint, open_session, work_once -from hymical_forms import models, storage -from hymical_forms.webhooks import DeliveryState, RetryPolicy +from sqlalchemy.exc import OperationalError + +from conftest import ( + ClientFactory, + app_settings, + build_settings, + create_endpoint, + open_session, + work_once, +) +from hymical_forms import models, storage, worker +from hymical_forms.db import create_engine_from_url +from hymical_forms.delivery import create_webhook_client, deliver +from hymical_forms.schema import SchemaNotReady, create_all +from hymical_forms.webhooks import DeliveryResult, DeliveryState, RetryPolicy from webhook_server import WebhookRecorder, unused_local_url ENDPOINT = "/f/contact-form" @@ -627,3 +643,169 @@ def test_the_configured_policy_reaches_the_worker(make_client: ClientFactory) -> assert policy.max_attempts == 7 assert policy.initial_seconds == 3 + + +# --- bounding one attempt ---------------------------------------------------- + + +def test_a_trickling_response_cannot_outlast_the_attempt_deadline( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + # Each byte arrives well inside the read timeout, so only an overall deadline + # can stop this. Without one the attempt runs for as long as the destination + # cares to keep writing, which outlives any lease and stalls the whole batch. + webhook.trickle_seconds = 0.05 + client, _ = queued_client( + make_client, + webhook.url, + webhook_connect_timeout_seconds=0.25, + webhook_read_timeout_seconds=0.25, + ) + client.post(ENDPOINT, data={"email": "dev@example.com"}) + + started = time.monotonic() + work_once(client, now=due_at(client)) + elapsed = time.monotonic() - started + + assert elapsed < 3 + attempt = attempts_of(client)[0] + assert attempt.outcome == "timeout" + assert delivery_of(client).state == DeliveryState.PENDING + + +def test_a_trickling_response_body_does_not_hold_up_a_success( + make_client: ClientFactory, webhook: WebhookRecorder +) -> None: + # The status is the whole answer. Reading a body nobody looks at would let a + # destination decide how long, and how much memory, an attempt takes. + webhook.trickle_seconds = 0.05 + webhook.trickle_body = True + client, _ = queued_client( + make_client, + webhook.url, + webhook_connect_timeout_seconds=0.25, + webhook_read_timeout_seconds=0.25, + ) + client.post(ENDPOINT, data={"email": "dev@example.com"}) + + started = time.monotonic() + work_once(client, now=due_at(client)) + elapsed = time.monotonic() - started + + assert elapsed < 3 + assert attempts_of(client)[0].outcome == "succeeded" + assert delivery_of(client).state == DeliveryState.DELIVERED + + +# --- the batch and its transaction ------------------------------------------- + + +def test_no_transaction_is_open_while_requests_are_in_flight( + make_client: ClientFactory, webhook: WebhookRecorder, monkeypatch: pytest.MonkeyPatch +) -> None: + client, _ = queued_client(make_client, webhook.url) + client.post(ENDPOINT, data={"email": "dev@example.com"}) + settings = app_settings(client) + observed: list[bool] = [] + + async def run() -> None: + webhook_client = create_webhook_client(settings) + try: + with open_session(client) as session: + + async def watching(*args: Any, **kwargs: Any) -> DeliveryResult: + observed.append(session.in_transaction()) + return await deliver(*args, **kwargs) + + monkeypatch.setattr(worker, "deliver", watching) + await worker.process_batch(session, webhook_client, settings, now=due_at(client)) + finally: + await webhook_client.aclose() + + asyncio.run(run()) + + assert observed == [False] + assert delivery_of(client).state == DeliveryState.DELIVERED + + +def test_one_failed_recording_does_not_lose_the_rest_of_the_batch( + make_client: ClientFactory, webhook: WebhookRecorder, monkeypatch: pytest.MonkeyPatch +) -> None: + # Every request in the batch has already been sent by the time recording + # starts. Abandoning the rest on the first database error would leave them + # claimed with nothing recorded, to be sent a second time once the lease runs + # out, for no reason other than a neighbour's failure. + client, _ = queued_client(make_client, webhook.url) + client.post(ENDPOINT, data={"email": "first@example.com"}) + client.post(ENDPOINT, data={"email": "second@example.com"}) + real = storage.complete_attempt + calls = 0 + + def flaky(*args: Any, **kwargs: Any) -> storage.CompletedAttempt: + nonlocal calls + calls += 1 + if calls == 1: + raise OperationalError("UPDATE", {}, Exception("connection reset")) + return real(*args, **kwargs) + + monkeypatch.setattr(storage, "complete_attempt", flaky) + with open_session(client) as session: + latest = max(row.next_attempt_at for row in session.scalars(select(models.WebhookDelivery))) + + handled = work_once(client, now=latest) + + assert handled == 2 + assert len(webhook.received) == 2 + with open_session(client) as session: + states = sorted(row.state for row in session.scalars(select(models.WebhookDelivery))) + assert states == [DeliveryState.DELIVERED, DeliveryState.PROCESSING] + + +# --- starting up ------------------------------------------------------------- + + +def test_the_worker_refuses_to_start_against_an_unmigrated_database(tmp_path: Path) -> None: + settings = build_settings(database_url=f"sqlite:///{tmp_path / 'empty.db'}") + stop = threading.Event() + stop.set() + + with pytest.raises(SchemaNotReady): + asyncio.run(worker.run_worker(settings, stop=stop)) + + +def test_the_worker_warns_when_the_lease_cannot_outlast_an_attempt( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + url = f"sqlite:///{tmp_path / 'forms.db'}" + engine = create_engine_from_url(url) + create_all(engine) + engine.dispose() + settings = build_settings( + database_url=url, + worker_lease_seconds=10, + webhook_connect_timeout_seconds=5, + webhook_read_timeout_seconds=10, + ) + stop = threading.Event() + stop.set() + + with caplog.at_level(logging.WARNING, logger="hymical_forms.worker"): + asyncio.run(worker.run_worker(settings, stop=stop)) + + assert "FORMS_WORKER_LEASE_SECONDS" in caplog.text + + +def test_the_default_lease_draws_no_warning( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + url = f"sqlite:///{tmp_path / 'forms.db'}" + engine = create_engine_from_url(url) + create_all(engine) + engine.dispose() + stop = threading.Event() + stop.set() + + with caplog.at_level(logging.WARNING, logger="hymical_forms.worker"): + asyncio.run(worker.run_worker(build_settings(database_url=url), stop=stop)) + + assert "FORMS_WORKER_LEASE_SECONDS" not in caplog.text diff --git a/tests/webhook_server.py b/tests/webhook_server.py index 035fec4..06bdae1 100644 --- a/tests/webhook_server.py +++ b/tests/webhook_server.py @@ -35,6 +35,12 @@ class WebhookRecorder: status: int = 200 delay_seconds: float = 0.0 + # When set, the response is written one byte at a time with this gap between + # bytes, which no single read ever waits long enough on to time out. + # ``trickle_body`` chooses whether the status line and headers arrive + # promptly and only the body trickles, or whether the whole response does. + trickle_seconds: float = 0.0 + trickle_body: bool = False received: list[ReceivedWebhook] = field(default_factory=list) _server: ThreadingHTTPServer | None = None _thread: threading.Thread | None = None @@ -74,10 +80,35 @@ def do_POST(self) -> None: # timeout is meant to catch. if recorder.delay_seconds: time.sleep(recorder.delay_seconds) + if recorder.trickle_seconds: + self._trickle() + return self.send_response(recorder.status) self.send_header("Content-Length", "0") self.end_headers() + def _trickle(self) -> None: + """ + write a response slowly enough to defeat any per-read timeout + """ + body = b"x" * 400 + head = ( + f"HTTP/1.1 {recorder.status} OK\r\nContent-Length: {len(body)}\r\n\r\n" + ).encode("ascii") + if recorder.trickle_body: + self.wfile.write(head) + self.wfile.flush() + head = b"" + try: + for byte in head + body: + self.wfile.write(bytes([byte])) + self.wfile.flush() + time.sleep(recorder.trickle_seconds) + except OSError: + # The client gave up and closed the connection, which is the + # behaviour a trickling response is used to provoke. + pass + def log_message(self, format: str, *args: object) -> None: """ silence the per-request logging that would otherwise flood the test output