diff --git a/docs/architecture/concurrency.md b/docs/architecture/concurrency.md index a25b021..4a5ab62 100644 --- a/docs/architecture/concurrency.md +++ b/docs/architecture/concurrency.md @@ -1,7 +1,8 @@ # Concurrency -Four places in this service have a race, and all four are settled by the database -rather than by a check in Python. +Four places in this service have a race, and every one of them is settled by the +database rather than by a check in Python. The first has two halves: taking a +delivery, and recording what happened to it. The pattern is the same every time: **read, compare in application code, then write** is never used, because two requests can both read, both find room, and @@ -32,6 +33,35 @@ keeps the claim correct there. work and never share it; a locked row is skipped rather than waited on; an expired lease is reclaimed by exactly one of two racing workers. +### Recording a result under a claim that was superseded + +Claiming and recording are separated by an HTTP request to somebody else's +server, which can take longer than the lease. So the second half of the claim has +a race of its own: worker A's lease expires, worker B legitimately reclaims the +delivery, and *then* A's request comes back. + +Each claim mints a `claim_token`, and the transition that records a result is +conditional on the row still carrying it: + +```sql +UPDATE webhook_deliveries +SET state = ..., cycle_attempts = ..., claim_expires_at = NULL, claim_token = NULL +WHERE id = :id AND claim_token = :token +``` + +The lease expiry cannot do this job. It says when a claim ends, not which claim it +is, and A is still looking at a `processing` row, so neither the state nor the +lease tells A that anything changed. A's update matches no row, and B's state +stands. + +The request A made is still recorded, taking the next free lifetime attempt +number read under a row lock. See +[Delivery semantics](delivery-semantics.md#what-happens-to-a-late-attempt). + +**Tested against real PostgreSQL** with two connections holding two different +claims on one row, in both result orderings, so what the fence keys on is +ownership rather than whether the late worker happened to succeed. + ## 2. Two retries with the same idempotency key A unique constraint on `(endpoint_id, idempotency_key)` is the authority. diff --git a/docs/architecture/delivery-semantics.md b/docs/architecture/delivery-semantics.md index 024c2ba..b4d6eba 100644 --- a/docs/architecture/delivery-semantics.md +++ b/docs/architecture/delivery-semantics.md @@ -23,6 +23,49 @@ A delivery is claimable when it is `pending` and due, **or** when it is `processing` and its lease has expired. The second case is the whole point: a worker that dies holding a job does not strand it in `processing` forever. +## Who owns a claim + +A lease says *when* a claim ends. It does not say *which* claim the row is +holding, and that is a different question with a different answer. + +Every claim mints a fresh `claim_token` onto the delivery. A worker records its +result against the token it claimed under, and that update is conditional on the +row still carrying it: + +```sql +UPDATE webhook_deliveries +SET state = ..., cycle_attempts = ..., claim_expires_at = NULL, claim_token = NULL +WHERE id = :id AND claim_token = :the_token_this_request_was_sent_under +``` + +This matters because a worker whose lease ran out mid-request has no way of +noticing on its own. It is still looking at a row that says `processing`, so +neither the state nor the lease tells it anything. The token does: once another +worker has reclaimed the delivery, the token no longer matches, and the late +worker's transition matches no row. + +So a superseded worker **cannot**: + +- clear or shorten the current owner's lease +- move the delivery to `delivered`, `failed` or `pending` +- spend an attempt from the current owner's retry cycle +- move `next_attempt_at` + +What it still does is record the request it genuinely made. See +[what happens to a late attempt](#what-happens-to-a-late-attempt) below. + +**Tested against real PostgreSQL** with independent connections holding two +different claims on one row. See [Testing](../development/testing.md). + +!!! warning "Every worker has to be on the same build" + + This holds only while every running worker maintains the token. A worker from + a build older than the `claim_token` migration neither writes it when it + claims nor checks it when it completes, so one left running alongside a newer + worker defeats the fence. Stop the old workers, migrate, then start the new + ones, which is the [deploy order](../operations/worker.md#deploy-order) this + project documents anyway. + ## The crash window Here is the sequence that produces a duplicate: @@ -42,6 +85,11 @@ Here is the sequence that produces a duplicate: No queue closes this on its own. It needs the receiver's cooperation. +Note what the claim token does and does not change here. It stops a late worker +from overwriting a newer worker's state. It does **not** stop the request from +having been sent twice: by the time ownership is checked, both requests have +already left. Your receiver still has to deduplicate. + ## What you should do about it **Deduplicate on the submission `id` in the signed payload.** @@ -88,8 +136,25 @@ ever being reused, however often a delivery is replayed. resets it to zero. That is what lets a replay grant a whole fresh schedule rather than one last attempt against an allowance that is already spent. -Until something is replayed the two are the same number, which is why the -migration that introduced the second one backfilled it from the first. +The two numbers start out equal, which is why the migration that introduced the +second one backfilled it from the first. They diverge for one of two reasons: the +delivery was replayed, or a worker that had lost its claim recorded a request it +really made. Only the current owner may draw on the retry allowance, so a late +attempt raises `attempts` and leaves `cycle_attempts` alone. + +## What happens to a late attempt + +A worker whose claim was superseded has usually already made a real HTTP request. +Somebody's server received it. Discarding that would make the attempt history +claim fewer requests went out than actually did, so it is recorded: + +- it takes the **next free lifetime attempt number**, read from the stored row + rather than from whatever the worker held in memory, so no number is reused +- it raises `attempts`, because the delivery really did produce that request +- it changes **nothing else** + +The number is taken under a row lock, so a late worker and the current owner +recording at the same instant still get two different numbers. ## Terminal is terminal, until an operator says otherwise diff --git a/docs/development/testing.md b/docs/development/testing.md index 35ac4e0..fb28909 100644 --- a/docs/development/testing.md +++ b/docs/development/testing.md @@ -61,6 +61,12 @@ database of its own, so migrating from genuinely nothing is what is being tested never share it. A row another worker holds is skipped rather than waited on. An expired lease is reclaimed by exactly one of two racing workers. +**Completion ownership.** Two connections hold two different claims on one +delivery: a worker whose lease has lapsed, and the worker that legitimately +reclaimed it. The lapsed worker cannot clear the newer lease, move the state, +spend the retry cycle or reuse an attempt number, in either result ordering. The +request it really made is still recorded, under a number of its own. + **Manual replay.** Several real connections replay one failed delivery at the same instant, and exactly one of them wins. diff --git a/docs/operations/migrations.md b/docs/operations/migrations.md index 0ad1367..e837da0 100644 --- a/docs/operations/migrations.md +++ b/docs/operations/migrations.md @@ -71,6 +71,8 @@ the DDL before anything touches it. Migrations through `0004` replay against SQLite too, in batch mode, because SQLite cannot `ALTER` a column in place and has to rebuild the table instead. +`0006` is a plain `ADD COLUMN`, which SQLite performs happily, but it sits behind +`0005` and so is unreachable there for the reason below. !!! warning "A fresh SQLite database cannot reach `head`" diff --git a/docs/operations/worker.md b/docs/operations/worker.md index 1b70b4f..35c6093 100644 --- a/docs/operations/worker.md +++ b/docs/operations/worker.md @@ -58,6 +58,12 @@ There is no coordination service and no broker. PostgreSQL is the queue. See The defaults leave a wide margin (60 seconds against 15). Keep it that way if you change them. + A lease that runs out mid-request costs you a duplicate send, not a corrupted + delivery: the overtaken worker cannot overwrite the state of whoever reclaimed + the delivery. It logs a warning saying so, which is the signal that the lease + is too short for how long that destination takes to answer. See + [Delivery semantics](../architecture/delivery-semantics.md#who-owns-a-claim). + ## Startup checks The worker performs the same schema check the API does. It reaches the database, diff --git a/docs/reference/limitations.md b/docs/reference/limitations.md index e11361e..6d7f5df 100644 --- a/docs/reference/limitations.md +++ b/docs/reference/limitations.md @@ -18,7 +18,9 @@ deploy it, and it is kept complete rather than flattering. than the sum. But if `FORMS_WORKER_LEASE_SECONDS` were set below the connect and read timeouts combined, another worker could claim a delivery that is still in flight and send it twice. The defaults leave a wide margin; keep it that way if - you change them. + you change them. The overtaken worker cannot overwrite the newer worker's state, + and logs a warning when it finds it has lost the claim, but the duplicate + request has already gone out by then. - **A signing secret cannot be rotated in place.** Rotation happens only as a side effect of changing the destination, so re-keying a receiver that stays at the same URL means pointing the endpoint elsewhere and back, or standing up a diff --git a/src/hymical_forms/migrations/versions/0006_20260826_delivery_claim_tokens.py b/src/hymical_forms/migrations/versions/0006_20260826_delivery_claim_tokens.py new file mode 100644 index 0000000..6a9d505 --- /dev/null +++ b/src/hymical_forms/migrations/versions/0006_20260826_delivery_claim_tokens.py @@ -0,0 +1,77 @@ +""" +delivery claim tokens + +A worker claims a delivery, sends it, and then records what happened. Nothing +made those last two steps agree about who owned the delivery: a worker whose +lease expired mid-request was still holding a row that said ``processing``, and +its completion overwrote the state, the lease and the counters belonging to the +worker that had legitimately reclaimed it. + +This adds one nullable column, ``claim_token``, holding the identity of the +claim the row is currently under. A fresh value is written every time the +delivery is claimed and cleared when it stops being claimed, so a completion can +be fenced on the exact claim its request was made under. The lease expiry cannot +do that job: it says when a claim ends, not which claim it is. + +Nothing is backfilled. NULL means "no claim is held", which is true of every +delivery that is not ``processing`` and of every row this column is added to. + +**Every worker must be on the new build before any of them runs against this +schema.** Applying the migration does not fence a worker that predates it. A +pre-0006 worker neither writes the column when it claims nor consults it when it +completes, so running one alongside a new worker breaks the guarantee in both +directions: its own completion still overwrites a newer worker's state exactly +as before, and because its claim leaves whatever token was already on the row in +place, a superseded new worker can present that stale token, match, and overwrite +the old worker's live claim. That is the deploy order this project already +documents (stop the old processes, migrate, start the new ones), and this +revision is one of the migrations for which it is load-bearing rather than +merely tidy. + +The downgrade drops the column. A build without it fences on nothing, which is +the behaviour this revision replaced. + +Nothing in this revision imports application code, for the reason given in +``0001``: a migration is a frozen record of a change, not a view of the current +models. + +revision: 0006 +revises: 0005 +created: 2026-08-26 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "0006" +down_revision: str | None = "0005" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """ + add the column naming the claim a delivery is currently held under + """ + # Nullable, and stays nullable: the absence of a token is meaningful rather + # than a gap to be filled, so there is nothing to backfill and no ALTER to + # tighten afterwards. A plain ADD COLUMN replays under SQLite as well as + # PostgreSQL, though a fresh SQLite database still cannot reach this revision + # for the reason recorded in ``0005``. + op.add_column( + "webhook_deliveries", sa.Column("claim_token", sa.String(length=36), nullable=True) + ) + + +def downgrade() -> None: + """ + remove the claim token column + """ + # A delivery that is ``processing`` when this runs keeps its lease and is + # recovered the ordinary way once that lease expires. The older build simply + # has no fence, which is the weakness ``0006`` exists to close. + op.drop_column("webhook_deliveries", "claim_token") diff --git a/src/hymical_forms/models.py b/src/hymical_forms/models.py index 5a07947..41c44d9 100644 --- a/src/hymical_forms/models.py +++ b/src/hymical_forms/models.py @@ -37,6 +37,7 @@ from hymical_forms.ingestion import Submission as DomainSubmission from hymical_forms.ratelimit import LIMITER_MAX_LENGTH, SUBJECT_MAX_LENGTH from hymical_forms.webhooks import ( + CLAIM_TOKEN_MAX_LENGTH, DELIVERY_ATTEMPT_ID_MAX_LENGTH, DELIVERY_ERROR_MAX_LENGTH, DELIVERY_STATE_MAX_LENGTH, @@ -331,6 +332,18 @@ class WebhookDelivery(Base): # leaving a permanent ``processing`` tombstone. claim_expires_at: Mapped[datetime | None] = mapped_column(UtcDateTime, default=None) + # Which claim the row is currently holding. A fresh value is minted every time + # the delivery is claimed and cleared the moment it stops being claimed, so it + # says *which* claim rather than when one ends. + # + # That distinction is the whole reason it exists. A worker whose lease ran out + # while its request was still in flight is still looking at a ``processing`` + # row, so neither the state nor the lease tells it that the delivery has been + # taken over. The token does: it presents the one it claimed under, no longer + # matches, and its completion is refused rather than overwriting the newer + # owner's state. + claim_token: Mapped[str | None] = mapped_column(String(CLAIM_TOKEN_MAX_LENGTH), default=None) + created_at: Mapped[datetime] = mapped_column(UtcDateTime, default=utcnow) completed_at: Mapped[datetime | None] = mapped_column(UtcDateTime, default=None) diff --git a/src/hymical_forms/storage.py b/src/hymical_forms/storage.py index 4905349..363b093 100644 --- a/src/hymical_forms/storage.py +++ b/src/hymical_forms/storage.py @@ -54,6 +54,7 @@ RetryPolicy, WebhookTarget, is_retryable, + new_claim_token, new_delivery_attempt_id, new_webhook_delivery_id, ) @@ -626,7 +627,7 @@ def claim_due_deliveries( :param now: the instant to judge dueness against :param lease_seconds: how long the claim protects a delivery from other workers :param limit: the most deliveries to claim at once - :returns: the deliveries this worker now owns + :returns: the deliveries this worker now owns, each carrying the token it was claimed under """ delivery = models.WebhookDelivery due = due_condition(now) @@ -642,6 +643,10 @@ def claim_due_deliveries( claimed: list[models.WebhookDelivery] = [] expires_at = now + timedelta(seconds=lease_seconds) for candidate in candidates: + # A token per claim, not per batch, so reclaiming one delivery of a batch + # never lets a stale worker's completion pass for another delivery's. + token = new_claim_token() + # The conditional update is the guarantee on backends without row locking: # whoever gets there first flips the row out of the due condition, and the # loser's update matches nothing. Redundant under SKIP LOCKED, and cheap. @@ -651,7 +656,11 @@ def claim_due_deliveries( update(delivery) .where(delivery.id == candidate.id) .where(due) - .values(state=DeliveryState.PROCESSING, claim_expires_at=expires_at) + .values( + state=DeliveryState.PROCESSING, + claim_expires_at=expires_at, + claim_token=token, + ) .execution_options(synchronize_session="fetch") ), ) @@ -688,6 +697,22 @@ def load_submissions( return {job_id: rows[submission_id] for job_id, submission_id in carried.items()} +@dataclass(frozen=True, slots=True) +class CompletedAttempt: + """ + the attempt that was recorded, and what recording it was allowed to settle + """ + + # ``owned`` is False when the worker's lease ran out and another worker + # reclaimed the delivery while the request was still in flight. The request + # was really made and is really recorded, but the delivery's state belongs to + # whoever holds the claim now, so ``state`` is that owner's state rather than + # anything this worker decided. + attempt: models.DeliveryAttempt + state: str + owned: bool + + def complete_attempt( session: Session, delivery: models.WebhookDelivery, @@ -695,27 +720,52 @@ def complete_attempt( *, now: datetime, policy: RetryPolicy, -) -> models.DeliveryAttempt: + claim_token: str | None, +) -> CompletedAttempt: """ - record one outbound request and move the delivery to whatever it earned + record one outbound request and, if the claim still holds, move the delivery on :param session: the session to write through :param delivery: the delivery the attempt was made for :param result: what the attempt produced :param now: the instant the attempt finished :param policy: how many attempts are allowed and how long to wait between them - :returns: the committed attempt record + :param claim_token: the claim the request was made under, as it stood when it was made + :returns: the recorded attempt, the delivery's state, and whether the claim still held """ # The audit row and the state it justifies are written together, so the # history can never disagree with the job about how many attempts happened. # + # Everything is decided from the stored row rather than from the loaded + # object. The request in between took as long as somebody else's server did, + # and the session that claimed this delivery does not expire what it loaded, + # so the values read at claim time can be several attempts out of date by the + # time a response comes back. + row = models.WebhookDelivery + reading = select(row.attempts, row.cycle_attempts, row.state, row.claim_token).where( + row.id == delivery.id + ) + if session.get_bind().dialect.name == "postgresql": + # Held for the rest of this transaction, so the attempt number taken here + # and the update that records it cannot be split by another worker. SQLite + # has no row locking and serialises writers anyway; it is not a production + # target. + reading = reading.with_for_update() + current = session.execute(reading).one() + + # Ownership is the exact claim, and nothing weaker. A lease that ran out and + # was reclaimed leaves the row in ``processing`` with a fresh token, so a test + # on the state or on the lease would wave exactly that case through. + owned = claim_token is not None and current.claim_token == claim_token + # Two counters, because they answer two different questions. The attempt is # numbered from the lifetime total, so a number is never reused however often - # the delivery has been replayed. The retry allowance is measured against the - # current cycle, so a replayed delivery gets a whole schedule again rather - # than failing immediately on a total that is already spent. Until something - # is replayed the two are the same number. - attempt_number = delivery.attempts + 1 - cycle_attempt = delivery.cycle_attempts + 1 + # the delivery has been replayed, and a superseded worker's real request takes + # the next free number rather than one the current owner has already spent. + # The retry allowance is measured against the current cycle, so a replayed + # delivery gets a whole schedule again rather than failing immediately on a + # total that is already spent. Only the worker that still owns the claim may + # draw on that allowance. + attempt_number = current.attempts + 1 attempt = models.DeliveryAttempt( id=new_delivery_attempt_id(), delivery_id=delivery.id, @@ -729,24 +779,38 @@ def complete_attempt( ) session.add(attempt) - delivery.attempts = attempt_number - delivery.cycle_attempts = cycle_attempt - delivery.claim_expires_at = None - - if result.outcome is DeliveryOutcome.SUCCEEDED: - delivery.state = DeliveryState.DELIVERED - delivery.completed_at = now - elif is_retryable(result) and not policy.is_exhausted(cycle_attempt): - delivery.state = DeliveryState.PENDING - delivery.next_attempt_at = now + policy.delay_after(cycle_attempt) - else: - # Either the receiver said something repeating will not fix, or the - # allowance ran out. Either way this is the last word on the delivery. - delivery.state = DeliveryState.FAILED - delivery.completed_at = now + # A superseded worker writes the lifetime total and nothing else. Its request + # went out, so the count of requests this delivery has ever produced is the + # one thing it is still entitled to change; the state, the lease, the retry + # cycle and the schedule all belong to whoever holds the claim now. + values: dict[str, Any] = {"attempts": attempt_number} + state: str = current.state + if owned: + cycle_attempt = current.cycle_attempts + 1 + values |= {"cycle_attempts": cycle_attempt, "claim_expires_at": None, "claim_token": None} + if result.outcome is DeliveryOutcome.SUCCEEDED: + state = DeliveryState.DELIVERED + values |= {"completed_at": now} + elif is_retryable(result) and not policy.is_exhausted(cycle_attempt): + state = DeliveryState.PENDING + values |= {"next_attempt_at": now + policy.delay_after(cycle_attempt)} + else: + # Either the receiver said something repeating will not fix, or the + # allowance ran out. Either way this is the last word on the delivery. + state = DeliveryState.FAILED + values |= {"completed_at": now} + values |= {"state": state} + + transition = update(row).where(row.id == delivery.id) + if owned: + # Redundant while the row lock above holds, and the whole of the guarantee + # on SQLite, which has none. Same reasoning as the claim's conditional + # update, and the same statement proves the claim was still ours. + transition = transition.where(row.claim_token == claim_token) + session.execute(transition.values(**values).execution_options(synchronize_session=False)) session.commit() - return attempt + return CompletedAttempt(attempt, str(state), owned) def list_deliveries( @@ -845,7 +909,10 @@ def requeue_failed_delivery(session: Session, delivery_id: str, *, now: datetime # attempt are left exactly as they are: a replay resumes a delivery, it does # not start a new one. Clearing ``cycle_attempts`` is what grants the fresh # retry allowance, and clearing ``completed_at`` is required by the check - # constraint that says a delivery is finished exactly when it says it is. + # constraint that says a delivery is finished exactly when it says it is. The + # lease and its token are already empty on anything that failed; clearing them + # here states the rule that a delivery outside ``processing`` holds no claim + # rather than relying on how it got here. delivery = models.WebhookDelivery result = cast( "CursorResult[Any]", @@ -858,6 +925,7 @@ def requeue_failed_delivery(session: Session, delivery_id: str, *, now: datetime cycle_attempts=0, next_attempt_at=now, claim_expires_at=None, + claim_token=None, completed_at=None, ) .execution_options(synchronize_session=False) diff --git a/src/hymical_forms/webhooks.py b/src/hymical_forms/webhooks.py index bf09522..bef2b46 100644 --- a/src/hymical_forms/webhooks.py +++ b/src/hymical_forms/webhooks.py @@ -40,6 +40,11 @@ WEBHOOK_DELIVERY_ID_MAX_LENGTH = len(WEBHOOK_DELIVERY_ID_PREFIX) + 32 DELIVERY_STATE_MAX_LENGTH = 16 +# Names one worker's claim on a delivery rather than the delivery itself, so a +# claim and the claim that later supersedes it are always distinguishable. +CLAIM_TOKEN_PREFIX = "clm_" +CLAIM_TOKEN_MAX_LENGTH = len(CLAIM_TOKEN_PREFIX) + 32 + # Failure text is written by whatever the destination did, so it is attacker # influenced and has to be bounded before it reaches a column. DELIVERY_ERROR_MAX_LENGTH = 500 @@ -244,6 +249,17 @@ def new_webhook_delivery_id() -> str: return f"{WEBHOOK_DELIVERY_ID_PREFIX}{uuid.uuid4().hex}" +def new_claim_token() -> str: + """ + generate an identifier for one worker's claim on a delivery + :returns: a fresh claim token such as ``clm_1f0c9a...`` + """ + # Random rather than counted up from the row, so minting one needs no read of + # the delivery it will be written to and two workers reclaiming the same + # delivery can never derive the same value. + return f"{CLAIM_TOKEN_PREFIX}{uuid.uuid4().hex}" + + @dataclass(frozen=True, slots=True) class WebhookTarget: """ diff --git a/src/hymical_forms/worker.py b/src/hymical_forms/worker.py index 3229290..bea243d 100644 --- a/src/hymical_forms/worker.py +++ b/src/hymical_forms/worker.py @@ -57,6 +57,12 @@ async def process_batch( submissions = storage.load_submissions(session, claimed) + # 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 + # let this worker overwrite the state of whoever reclaimed the delivery. + claims = {job.id: job.claim_token for job in claimed} + # The network calls overlap so that one unresponsive destination does not # hold up the rest of the batch for its whole timeout. They are made with no # database transaction open: holding one across somebody else's server would @@ -77,14 +83,31 @@ 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): - storage.complete_attempt(session, job, result, now=now, policy=policy) - logger.info( - "delivery %s attempt %d %s (%s)", - job.id, - job.attempts, - result.outcome, - job.state, + recorded = storage.complete_attempt( + session, job, result, now=now, policy=policy, claim_token=claims[job.id] ) + if recorded.owned: + logger.info( + "delivery %s attempt %d %s (%s)", + job.id, + recorded.attempt.attempt_number, + result.outcome, + recorded.state, + ) + else: + # The lease ran out while this request was in flight and another + # worker took the delivery over. The request really was sent, so it + # stays in the history, but this worker no longer says what the + # delivery is. Worth a warning rather than a note: it means the lease + # is too short for how long this destination takes to answer. + logger.warning( + "delivery %s attempt %d %s recorded without ownership, " + "the lease expired and another worker now holds it (%s)", + job.id, + recorded.attempt.attempt_number, + result.outcome, + recorded.state, + ) return len(claimed) diff --git a/tests/integration/test_delivery_fencing_postgres.py b/tests/integration/test_delivery_fencing_postgres.py new file mode 100644 index 0000000..761cf82 --- /dev/null +++ b/tests/integration/test_delivery_fencing_postgres.py @@ -0,0 +1,334 @@ +""" +completion ownership against real PostgreSQL + +A worker claims a delivery, spends as long as somebody else's server takes over +the request, and only then records what happened. Between those two moments its +lease can run out and another worker can legitimately take the delivery over. +What must not happen is the first worker coming back and overwriting the second +one's state. + +That is a race between two connections holding two different claims on one row, +so SQLite cannot show it: it serialises writers and would make the sequence look +safe whether or not the fence exists. These tests hold the claims explicitly, in +independent sessions, and assert on what the database settled on. + +Duplicate delivery is not what is under test here. Both workers may genuinely +have sent the request, and Hymical Forms still promises only at-least-once. +""" + +from __future__ import annotations + +import threading +from concurrent.futures import ThreadPoolExecutor +from datetime import UTC, datetime, timedelta + +from sqlalchemy.orm import Session, sessionmaker + +from hymical_forms import models, storage +from hymical_forms.webhooks import DeliveryOutcome, DeliveryResult, DeliveryState, RetryPolicy +from integration.support import seed_due_deliveries, seed_endpoint + +NOW = datetime(2026, 8, 26, 12, 0, tzinfo=UTC) +LEASE = 60.0 + +# Long enough that a retryable failure schedules another attempt rather than +# exhausting the allowance, so the retry branch is the one being exercised. +POLICY = RetryPolicy(max_attempts=5, initial_seconds=10, max_seconds=3600) + +SUCCESS = DeliveryResult(DeliveryOutcome.SUCCEEDED, response_status=200) +PERMANENT_FAILURE = DeliveryResult( + DeliveryOutcome.HTTP_ERROR, response_status=404, error="destination responded with HTTP 404" +) +RETRYABLE_FAILURE = DeliveryResult( + DeliveryOutcome.HTTP_ERROR, response_status=503, error="destination responded with HTTP 503" +) + + +def claim(session: Session, *, now: datetime) -> models.WebhookDelivery: + """ + take the single due delivery as one worker, through the ordinary claim + :param session: the session standing in for one worker's connection + :param now: the instant that worker judges dueness against + :returns: the claimed delivery, carrying the token it was claimed under + """ + claimed = storage.claim_due_deliveries(session, now=now, lease_seconds=LEASE, limit=10) + assert len(claimed) == 1, "the fixture should offer exactly one due delivery" + return claimed[0] + + +def read(sessions: sessionmaker[Session], delivery_id: str) -> models.WebhookDelivery: + """ + read a delivery back on a connection of its own, so this is the committed state + :param sessions: factory handing out independent connections + :param delivery_id: the delivery to read + :returns: the delivery as the database now holds it + """ + with sessions() as observer: + delivery = observer.get(models.WebhookDelivery, delivery_id) + assert delivery is not None + return delivery + + +def attempt_numbers(sessions: sessionmaker[Session], delivery_id: str) -> list[int]: + """ + read the attempt numbers recorded for a delivery, lowest first + :param sessions: factory handing out independent connections + :param delivery_id: the delivery whose history to read + :returns: the recorded attempt numbers + """ + with sessions() as observer: + return [row.attempt_number for row in storage.list_delivery_attempts(observer, delivery_id)] + + +def seed_one(sessions: sessionmaker[Session]) -> str: + """ + put one due delivery in the queue + :param sessions: factory handing out independent connections + :returns: the delivery id + """ + with sessions() as setup: + seed_endpoint(setup) + return seed_due_deliveries(setup, 1, now=NOW)[0] + + +def test_a_superseded_worker_cannot_overwrite_the_current_owner( + sessions: sessionmaker[Session], +) -> None: + """ + the whole invariant in one sequence: A claims, A's lease lapses, B reclaims, A returns + :param sessions: factory handing out independent connections + """ + delivery_id = seed_one(sessions) + reclaimed_at = NOW + timedelta(seconds=LEASE + 1) + + with sessions() as worker_a, sessions() as worker_b: + job_a = claim(worker_a, now=NOW) + # A's lease has lapsed by this point, so B is entitled to the delivery. + job_b = claim(worker_b, now=reclaimed_at) + assert job_b.claim_token != job_a.claim_token + + # A's request finally comes back and A tries to record it. Under the + # unfenced implementation this set the row to delivered, cleared B's lease + # and rewound the counters while B was still sending. + stale = storage.complete_attempt( + worker_a, job_a, SUCCESS, now=NOW, policy=POLICY, claim_token=job_a.claim_token + ) + assert stale.owned is False + + held = read(sessions, delivery_id) + assert held.state == DeliveryState.PROCESSING, "a superseded worker ended B's delivery" + assert held.claim_token == job_b.claim_token, "a superseded worker took the claim back" + assert held.claim_expires_at == reclaimed_at + timedelta(seconds=LEASE), ( + "a superseded worker cleared the current owner's lease" + ) + assert held.completed_at is None + assert held.cycle_attempts == 0, "a superseded worker spent the current retry cycle" + + # B, which still owns the claim, has the last word. + owner = storage.complete_attempt( + worker_b, + job_b, + PERMANENT_FAILURE, + now=reclaimed_at, + policy=POLICY, + claim_token=job_b.claim_token, + ) + assert owner.owned is True + + settled = read(sessions, delivery_id) + assert settled.state == DeliveryState.FAILED + assert settled.completed_at == reclaimed_at + assert settled.claim_token is None + assert settled.claim_expires_at is None + + +def test_fencing_follows_ownership_rather_than_the_result( + sessions: sessionmaker[Session], +) -> None: + """ + the opposing case: the superseded worker failed and the current owner succeeded + :param sessions: factory handing out independent connections + """ + # The mirror of the test above. If the fence were accidentally keyed on the + # kind of result rather than on who holds the claim, one of these two + # orderings would pass and the other would not. + delivery_id = seed_one(sessions) + reclaimed_at = NOW + timedelta(seconds=LEASE + 1) + + with sessions() as worker_a, sessions() as worker_b: + job_a = claim(worker_a, now=NOW) + job_b = claim(worker_b, now=reclaimed_at) + + storage.complete_attempt( + worker_a, + job_a, + PERMANENT_FAILURE, + now=NOW, + policy=POLICY, + claim_token=job_a.claim_token, + ) + assert read(sessions, delivery_id).state == DeliveryState.PROCESSING + + storage.complete_attempt( + worker_b, + job_b, + SUCCESS, + now=reclaimed_at, + policy=POLICY, + claim_token=job_b.claim_token, + ) + + settled = read(sessions, delivery_id) + assert settled.state == DeliveryState.DELIVERED, "a superseded failure buried a real success" + assert settled.completed_at == reclaimed_at + + +def test_a_superseded_worker_does_not_move_the_retry_schedule( + sessions: sessionmaker[Session], +) -> None: + """ + the retry cycle and the next due time belong to whoever holds the claim + :param sessions: factory handing out independent connections + """ + delivery_id = seed_one(sessions) + reclaimed_at = NOW + timedelta(seconds=LEASE + 1) + + with sessions() as worker_a, sessions() as worker_b: + job_a = claim(worker_a, now=NOW) + job_b = claim(worker_b, now=reclaimed_at) + + storage.complete_attempt( + worker_a, + job_a, + RETRYABLE_FAILURE, + now=NOW, + policy=POLICY, + claim_token=job_a.claim_token, + ) + storage.complete_attempt( + worker_b, + job_b, + RETRYABLE_FAILURE, + now=reclaimed_at, + policy=POLICY, + claim_token=job_b.claim_token, + ) + + settled = read(sessions, delivery_id) + assert settled.state == DeliveryState.PENDING + # One cycle attempt, not two, so the first delay of the schedule is what the + # owner earned rather than the second. + assert settled.cycle_attempts == 1 + assert settled.next_attempt_at == reclaimed_at + timedelta(seconds=POLICY.initial_seconds) + # Both requests really went out, and the lifetime count says so. + assert settled.attempts == 2 + + +def test_a_superseded_workers_request_stays_in_the_history( + sessions: sessionmaker[Session], +) -> None: + """ + a request that was really sent is really recorded, under a number of its own + :param sessions: factory handing out independent connections + """ + delivery_id = seed_one(sessions) + reclaimed_at = NOW + timedelta(seconds=LEASE + 1) + + with sessions() as worker_a, sessions() as worker_b: + job_a = claim(worker_a, now=NOW) + job_b = claim(worker_b, now=reclaimed_at) + + stale = storage.complete_attempt( + worker_a, job_a, SUCCESS, now=NOW, policy=POLICY, claim_token=job_a.claim_token + ) + owner = storage.complete_attempt( + worker_b, + job_b, + SUCCESS, + now=reclaimed_at, + policy=POLICY, + claim_token=job_b.claim_token, + ) + + assert stale.attempt.attempt_number == 1 + assert owner.attempt.attempt_number == 2 + assert attempt_numbers(sessions, delivery_id) == [1, 2] + assert read(sessions, delivery_id).attempts == 2 + + +def test_two_workers_completing_at_once_never_reuse_an_attempt_number( + sessions: sessionmaker[Session], +) -> None: + """ + the superseded worker and the current owner record from two connections at the same instant + :param sessions: factory handing out independent connections + """ + delivery_id = seed_one(sessions) + reclaimed_at = NOW + timedelta(seconds=LEASE + 1) + + with sessions() as setup_a, sessions() as setup_b: + job_a = claim(setup_a, now=NOW) + job_b = claim(setup_b, now=reclaimed_at) + token_a, token_b = job_a.claim_token, job_b.claim_token + + barrier = threading.Barrier(2) + + def record(token: str | None, moment: datetime) -> storage.CompletedAttempt: + barrier.wait() + with sessions() as session: + job = session.get(models.WebhookDelivery, delivery_id) + assert job is not None + return storage.complete_attempt( + session, job, SUCCESS, now=moment, policy=POLICY, claim_token=token + ) + + with ThreadPoolExecutor(max_workers=2) as pool: + futures = [ + pool.submit(record, token_a, NOW), + pool.submit(record, token_b, reclaimed_at), + ] + stale, owner = (future.result() for future in futures) + + assert [stale.owned, owner.owned] == [False, True] + assert attempt_numbers(sessions, delivery_id) == [1, 2], "an attempt number was reused" + settled = read(sessions, delivery_id) + assert settled.attempts == 2 + assert settled.state == DeliveryState.DELIVERED + assert settled.cycle_attempts == 1 + + +def test_reclaiming_after_a_superseded_completion_still_works( + sessions: sessionmaker[Session], +) -> None: + """ + refusing a superseded completion must not strand the delivery in processing + :param sessions: factory handing out independent connections + """ + delivery_id = seed_one(sessions) + reclaimed_at = NOW + timedelta(seconds=LEASE + 1) + + with sessions() as worker_a, sessions() as worker_b: + job_a = claim(worker_a, now=NOW) + claim(worker_b, now=reclaimed_at) + storage.complete_attempt( + worker_a, job_a, SUCCESS, now=NOW, policy=POLICY, claim_token=job_a.claim_token + ) + + # B is now abandoned in its turn, and the ordinary expired-lease recovery has + # to keep working over a row a superseded worker has already written to. + with sessions() as worker_c: + recovered = claim(worker_c, now=reclaimed_at + timedelta(seconds=LEASE + 1)) + assert recovered.id == delivery_id + + outcome = storage.complete_attempt( + worker_c, + recovered, + SUCCESS, + now=reclaimed_at + timedelta(seconds=LEASE + 1), + policy=POLICY, + claim_token=recovered.claim_token, + ) + + assert outcome.owned is True + assert read(sessions, delivery_id).state == DeliveryState.DELIVERED + assert attempt_numbers(sessions, delivery_id) == [1, 2]