feat: queue a rearm when extending a timer whose handler is running

AGENTS.md

@@ -34,7 +34,10 @@ will **not** tell you:
   runs in the database and one in the process. They cannot share an implementation, so
   `tests/test_client_contract.py` couples them by behaviour. A change to either adapter's insert, fetch,
   terminal, or DLQ semantics belongs in that suite.
-- `schema.py`: the three partial indexes and the `<table>_lease_ck` CHECK are load-bearing.
+- `schema.py`: the three partial indexes and the `<table>_lease_ck` and `<table>_rearm_ck` CHECKs are
+  load-bearing.
+- Every terminal write (delete, batched delete, DLQ, retry) must apply a queued rearm instead of
+  dropping it; `tests/test_client_contract.py` holds the rearm scenarios for both adapters.
 - `message.py`: the `DLQFailureReason` `Literal` is a **public contract**; operator queries key off
   those strings, so changing one is API-breaking.
 - `metrics/` (the recorder seam) and `prometheus/` + `opentelemetry/` (native middleware) are two

CONTEXT.md

@@ -69,9 +69,14 @@ _Avoid_: idempotency key, message key
 
 **Extend**:
 Re-publishing a `timer_id` with `timer_conflict="extend"`: an unleased timer's deadline moves to
-the later of the stored and requested one, in place. A leased timer is left to its worker.
+the later of the stored and requested one, in place. On a leased timer it queues a rearm.
 _Avoid_: refresh, reschedule, upsert
 
+**Rearm**:
+An extend queued on a leased timer, held in the `rearm_*` columns. The run's terminal write applies
+it, turning the row back into a pending timer instead of deleting it.
+_Avoid_: successor, follow-up timer
+
 **Relay**:
 Forwarding a delivered row onward through a native FastStream publisher chain to a foreign broker.
 _Avoid_: bridge, forwarder

docs/operations/alembic.md

@@ -200,6 +200,49 @@ table in a non-default schema, drop an `INVALID` index left by a failed
 concurrent build before retrying, and don't apply what
 `alembic revision --autogenerate` produces for this change.
 
+## Upgrading to 0.19: add the rearm columns { #upgrading-to-019 }
+
+0.19 lets `publish(..., timer_conflict="extend")` reach a timer whose
+handler is running: the extend is queued in three new nullable columns,
+`rearm_at`, `rearm_payload` and `rearm_headers`, and applied when the run
+ends. A CHECK, `outbox_rearm_ck`, keeps a queued rearm off unleased rows.
+`validate_schema()` reports a table without them as drift.
+
+Run the migration **before** deploying 0.19, which reads and writes the new
+columns on every completion. Adding a nullable column without a default
+doesn't rewrite the table, and adding the CHECK as `NOT VALID` and
+validating it separately avoids holding an exclusive lock while the table is
+scanned.
+
+```python
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+
+def upgrade() -> None:
+    op.add_column("outbox", sa.Column("rearm_at", sa.DateTime(timezone=True), nullable=True))
+    op.add_column("outbox", sa.Column("rearm_payload", sa.LargeBinary(), nullable=True))
+    op.add_column("outbox", sa.Column("rearm_headers", postgresql.JSONB(), nullable=True))
+    op.execute(
+        "ALTER TABLE outbox ADD CONSTRAINT outbox_rearm_ck "
+        "CHECK (rearm_at IS NULL OR acquired_token IS NOT NULL) NOT VALID"
+    )
+    op.execute("ALTER TABLE outbox VALIDATE CONSTRAINT outbox_rearm_ck")
+
+
+def downgrade() -> None:
+    op.drop_constraint("outbox_rearm_ck", "outbox", type_="check")
+    op.drop_column("outbox", "rearm_headers")
+    op.drop_column("outbox", "rearm_payload")
+    op.drop_column("outbox", "rearm_at")
+```
+
+Earlier versions ignore the new columns, and `validate_schema()` doesn't
+report extra columns, so deploying after the migration is safe. Don't
+downgrade while a rearm is queued: dropping the columns drops it, and the
+timer is deleted when its run ends.
+
 ## Adding the DLQ after the fact
 
 To opt into [Dead-letter queue](../usage/dlq.md) audit, add

docs/usage/timers.md

@@ -129,23 +129,46 @@ of order cannot pull a timer in. When the requested deadline is later, the
 row also takes the new payload and headers, and its retry history
 (`attempts_count`, `first_attempt_at`, `last_attempt_at`) is cleared. When
 it is not later, the row keeps everything it had and the call still returns
-its id. `deliveries_count` is never reset.
-
-An extend never touches a row a worker has leased, and returns `None` for
-it.
-
-- An id means that once your transaction commits, the timer will not fire
-  before the deadline you asked for. Until the commit, workers skip the
-  locked row; after it, they see the new deadline.
-- `None` means a worker claimed the timer first. Its handler is running, or
-  will run when an expired lease is reclaimed, for the deadline it claimed.
-  Publishing again does not help while that row exists.
-
-If a late heartbeat must win over an expiry that is already running, the
-library cannot decide that for you. Have the handler re-check, in the same
-transaction as its side effects, whether the timeout still applies (for
-example against a last-activity column the heartbeat writes), and publish a
-fresh timer from the heartbeat once the old row is gone.
+its id. `deliveries_count` is not reset.
+
+Once your transaction commits, the timer will not fire before the deadline
+you asked for. Until the commit, workers skip the locked row; after it, they
+see the new deadline.
+
+### Extending a timer that is already running
+
+When a worker has claimed the timer, its handler runs for the deadline it
+claimed, and the extend leaves that run alone. It queues a rearm on the row
+instead and still returns the row id. When the run ends, the rearm turns the
+row back into a pending timer with the queued deadline, payload and headers,
+so every accepted heartbeat leads to a firing at or after its deadline:
+
+- If the handler succeeds, the timer starts over at the rearm deadline, with
+  its retry history and `deliveries_count` cleared.
+- If the handler fails and would be retried, the rearm replaces the retry.
+  `deliveries_count` is kept, so `max_deliveries` still stops a handler that
+  fails on every run.
+- If the handler fails terminally, the failed run goes to the DLQ (when one
+  is configured) and the timer starts over at the rearm deadline.
+- If the worker dies, the expired lease is reclaimed and the run is
+  delivered again as usual; the rearm applies when that run ends.
+
+Queued rearms follow the same rule as pending timers: a rearm only moves
+later, and an extend whose deadline is not later than the running deadline
+queues nothing.
+
+A handler can rearm its own timer the same way, by publishing its own
+`timer_id` with `timer_conflict="extend"`, directly or through an
+`OutboxResponse`.
+
+The rearm doesn't undo what the running handler does. If a heartbeat must
+win over an expiry that is already running, have the handler re-check, in
+the same transaction as its side effects, whether the timeout still applies
+(for example against a last-activity column the heartbeat writes). The
+rearm then fires the handler again for the newer deadline.
+
+A rearm that becomes due when the run ends is picked up by polling, within
+the subscriber's `max_fetch_interval`. No NOTIFY is sent for it.
 
 Every extend writes a new row version, even when the deadline does not
 move. `next_attempt_at` is indexed, so these updates are not HOT updates:
@@ -155,25 +178,24 @@ Size autovacuum for your heartbeat rate (see
 
 ## Cancellation
 
-`broker.cancel_timer(*, queue, timer_id, session)` issues a `DELETE` on
-the caller's session, but only if the row is not yet leased:
+`broker.cancel_timer(*, queue, timer_id, session)` cancels a timer on the
+caller's session:
 
 ```python
-deleted = await broker.cancel_timer(
+cancelled = await broker.cancel_timer(
     queue="orders",
     timer_id="order-confirm-42",
     session=session,
 )
-# True if a row was deleted; False if it didn't exist or was already in flight.
+# True if a pending timer or a queued rearm was dropped.
 ```
 
-The underlying SQL is `DELETE WHERE queue=? AND timer_id=? AND
-acquired_token IS NULL`. The `acquired_token IS NULL` guard is
-load-bearing: it preserves the lease-token invariant by refusing to
-clobber a row whose handler is already running. If the timer fired in the
-race window between your application logic deciding to cancel and the
-`DELETE` landing, the delivery completes normally and `cancel_timer`
-returns `False`.
+It locks the `(queue, timer_id)` row first. An unleased row is deleted. A
+leased row is never deleted, because its handler is already running: the
+cancel only drops a queued rearm, so the timer does not come back after the
+run, and returns `False` when there was none. If the timer fired in the race
+window between your application logic deciding to cancel and the cancel
+landing, the delivery completes normally.
 
 ## Latency floor
 
@@ -198,3 +220,7 @@ on it, bind the `TestOutboxBroker` to a name and read
 `tb.fake_client.rows[0].next_attempt_at`.
 Pass `run_loops=True` if you need scheduled delivery to actually
 wait. See [Testing](./testing.md).
+
+A rearm applied when a run ends is not dispatched again in sync mode, so a
+handler that rearms its own timer runs once per publish. The rearmed row
+stays in `tb.fake_client.rows`.

faststream_outbox/broker.py

@@ -439,9 +439,10 @@ async def publish(  # ty: ignore[invalid-method-override]
         returns ``None``. ``"extend"`` updates a row no worker has leased, in place:
         its deadline becomes the later of the stored and the requested one, and only
         when the requested one is later does the row take the new payload and
-        headers and drop its retry history. Either way it returns the row id. It
-        returns ``None`` only when a worker holds the lease: that worker's handler
-        runs for the deadline it claimed, and the caller has to handle that case.
+        headers and drop its retry history. On a leased row it queues the same
+        update as a rearm, which applies when the running handler finishes; see
+        the timers guide for how it interacts with retries and the DLQ. An extend
+        always returns the row id.
 
         Returns the row's ``id`` (BigInt PK), or ``None`` as described above.
         """
@@ -510,19 +511,18 @@ async def cancel_timer(
         timer_id: str,
         session: SessionOrConnection,
     ) -> bool:
-        """Delete a not-yet-leased timer row. Idempotent at the ``(queue, timer_id)`` level.
+        """Delete a not-yet-leased timer row, or drop the rearm queued on a leased one.
 
         Same transactional contract as :meth:`publish` — runs on the caller's session and
-        commits with their transaction.
+        commits with their transaction. Idempotent at the ``(queue, timer_id)`` level.
 
-        Returns ``True`` if *this call* deleted the row. ``False`` means the row is no
-        longer cancelable by you, which covers three cases:
+        Returns ``True`` if *this call* deleted the row or dropped its queued rearm.
+        ``False`` means there was nothing left for you to cancel, which covers three cases:
 
-        1. A worker has claimed the row (``acquired_token IS NOT NULL``). The
-           ``acquired_token IS NULL`` guard prevents this call from clobbering the
-           lease. Normally the delivery then completes. If that worker died, the lease
-           stays set until ``lease_ttl_seconds`` passes and the row is delivered again;
-           ``cancel_timer`` can't stop that redelivery.
+        1. A worker has claimed the row and no rearm is queued on it. The call never
+           deletes a leased row, so the delivery completes normally. If that worker
+           died, the lease stays set until ``lease_ttl_seconds`` passes and the row is
+           delivered again; ``cancel_timer`` can't stop that redelivery.
         2. Another caller already canceled the same ``(queue, timer_id)`` in a concurrent
            transaction.
         3. No row exists with that ``(queue, timer_id)`` — either the timer was never