Interval 1: worker reliability fixes and replay delivery reporting

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
 

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

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
 

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 |

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
 

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),
     )
 
 

src/hymical_forms/delivery.py

@@ -9,6 +9,8 @@
 
 from __future__ import annotations
 
+import asyncio
+
 import httpx2
 
 from hymical_forms import __version__
@@ -53,15 +55,36 @@ 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
     :param client: the shared outbound client
     :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}",
     )
 
 

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(