From 544c7b9ae68c55527ce2a3e04919a8a9b81de4d8 Mon Sep 17 00:00:00 2001 From: Fangliquan Date: Fri, 24 Jul 2026 18:00:00 +0800 Subject: [PATCH 1/3] fix(gateway): label redelivery after mid-recovery crash Claiming a pending obligation now stamps attempting and treats prior attempts as needing the recovered marker, so a crash mid-redelivery cannot silently duplicate a platform-ACK'd send. --- gateway/delivery_ledger.py | 24 +++++++---- gateway/run.py | 10 +++++ tests/gateway/test_delivery_ledger.py | 61 +++++++++++++++++++++++++++ 3 files changed, 86 insertions(+), 9 deletions(-) diff --git a/gateway/delivery_ledger.py b/gateway/delivery_ledger.py index e17f92b8ab40..ecd52a23a918 100644 --- a/gateway/delivery_ledger.py +++ b/gateway/delivery_ledger.py @@ -241,11 +241,13 @@ def sweep_recoverable( """Claim undelivered rows owned by dead processes; return them for redelivery. - Claiming atomically re-stamps the owner to THIS process and increments - ``attempts``, so a second gateway racing the same sweep cannot - double-claim (the UPDATE is guarded on the previous owner stamp). - Rows over the attempts cap or older than the stale cutoff transition to - 'abandoned' instead of being returned. + Claiming atomically re-stamps the owner to THIS process, sets + ``state='attempting'``, and increments ``attempts``, so a second + gateway racing the same sweep cannot double-claim (the UPDATE is + guarded on the previous owner stamp) and a crash mid-redelivery is + treated as ambiguous on the next boot. Rows over the attempts cap + or older than the stale cutoff transition to 'abandoned' instead of + being returned. ``deliverable_platforms`` (platform value strings) restricts claiming to platforms the caller can actually send on this boot. ``attempts`` is the @@ -285,7 +287,8 @@ def sweep_recoverable( continue cursor = conn.execute( """UPDATE delivery_obligations - SET owner_pid=?, owner_started_at=?, attempts=attempts+1, + SET state='attempting', + owner_pid=?, owner_started_at=?, attempts=attempts+1, updated_at=? WHERE obligation_id=? AND (owner_pid IS ? OR owner_pid=?)""", (pid, started, now, oid, owner_pid, owner_pid), @@ -298,9 +301,12 @@ def sweep_recoverable( "chat_id": chat_id, "thread_id": thread_id, "content": content, - # pending = send never started, redeliver plainly; - # attempting/failed = ambiguous or rejected, carry marker. - "needs_marker": state != "pending", + # pending + attempts==0 = send never started, redeliver + # plainly. attempting/failed, or a prior reclaim that + # already spent an attempt (stuck pending rows from + # older builds), carry the recovered-reply marker so a + # possible platform-side duplicate is labeled. + "needs_marker": state != "pending" or attempts >= 1, "attempts": attempts + 1, }) return claimed diff --git a/gateway/run.py b/gateway/run.py index 911021a36ee6..e1cf99a51827 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -9889,6 +9889,7 @@ async def _redeliver_pending_obligations(self) -> int: from gateway.delivery_ledger import ( RECOVERED_MARKER, ledger_enabled, + mark_attempting, mark_delivered, mark_failed, sweep_recoverable, @@ -9932,6 +9933,15 @@ async def _redeliver_pending_obligations(self) -> int: metadata = ( {"thread_id": row["thread_id"]} if row.get("thread_id") else None ) + # Mirror the producer path: claim already stamps attempting, but + # stamp again immediately before await so any crash mid-send + # leaves an unambiguous ledger state for the next boot. + try: + mark_attempting(row["obligation_id"]) + except Exception: + logger.debug( + "delivery ledger mark_attempting failed", exc_info=True + ) try: result = await adapter.send( chat_id=row["chat_id"], diff --git a/tests/gateway/test_delivery_ledger.py b/tests/gateway/test_delivery_ledger.py index cec3b8339ed0..f13e604cf99b 100644 --- a/tests/gateway/test_delivery_ledger.py +++ b/tests/gateway/test_delivery_ledger.py @@ -90,10 +90,47 @@ def test_dead_owner_pending_claimed_without_marker(self): assert len(claimed) == 1 assert claimed[0]["needs_marker"] is False assert claimed[0]["attempts"] == 1 + # Claim starts the recovery send: state must leave pending so a + # crash mid-redelivery is treated as ambiguous on the next boot. + assert _row("ob-1")["state"] == "attempting" # Claim re-stamps ownership: a second sweep in the same (live) # process must not double-claim. assert dl.sweep_recoverable() == [] + def test_pending_reclaim_after_prior_attempt_needs_marker(self): + """Crash mid-redelivery of a pending row must not silent-duplicate. + + First reclaim of orphaned pending redelivers plainly (send never + started). If the process dies after that claim/send starts but + before mark_delivered, the next reclaim must carry the recovered + marker — even if state was stuck at pending historically. + """ + _record() + _orphan("ob-1") + first = dl.sweep_recoverable() + assert first[0]["needs_marker"] is False + assert _row("ob-1")["state"] == "attempting" + assert _row("ob-1")["attempts"] == 1 + + _orphan("ob-1") + second = dl.sweep_recoverable() + assert len(second) == 1 + assert second[0]["needs_marker"] is True + assert second[0]["attempts"] == 2 + + def test_stuck_pending_with_prior_attempts_needs_marker(self): + """Pre-fix rows: pending + attempts>=1 still need the marker.""" + _record() + with dl._connect() as conn: + conn.execute( + "UPDATE delivery_obligations SET attempts=1, " + "owner_pid=999999999, owner_started_at=1 WHERE obligation_id=?", + ("ob-1",), + ) + claimed = dl.sweep_recoverable() + assert claimed[0]["needs_marker"] is True + assert claimed[0]["attempts"] == 2 + class TestPrune: def test_old_delivered_rows_pruned(self): @@ -171,6 +208,30 @@ async def test_attempting_redelivers_with_marker(self): assert sent["content"].startswith(dl.RECOVERED_MARKER) assert sent["content"].endswith("the final answer") + @pytest.mark.asyncio + async def test_second_boot_pending_recovery_carries_marker(self): + """Simulate crash after first pending reclaim started send.""" + _record() + _orphan("ob-1") + first = self._adapter() + n1 = await self._runner(first)._redeliver_pending_obligations() + assert n1 == 1 + assert not first.send.call_args.kwargs["content"].startswith( + dl.RECOVERED_MARKER + ) + # Pretend the first recovery send never reached mark_delivered. + with dl._connect() as conn: + conn.execute( + "UPDATE delivery_obligations SET state='attempting', " + "owner_pid=999999999, owner_started_at=1 WHERE obligation_id=?", + ("ob-1",), + ) + second = self._adapter() + await self._runner(second)._redeliver_pending_obligations() + sent = second.send.call_args.kwargs["content"] + assert sent.startswith(dl.RECOVERED_MARKER) + assert sent.endswith("the final answer") + class TestAttemptsOnlySpentOnRealSends: """``attempts`` is the redelivery budget — it must buy a send. From d679408c352eeaeb2899bc724d4b01d69b3cf58b Mon Sep 17 00:00:00 2001 From: fangliquanflq Date: Fri, 31 Jul 2026 14:59:49 +0800 Subject: [PATCH 2/3] chore(contributors): add contributor map fangliquan@qq.com --- contributors/emails/fangliquan@qq.com | 1 + 1 file changed, 1 insertion(+) create mode 100644 contributors/emails/fangliquan@qq.com diff --git a/contributors/emails/fangliquan@qq.com b/contributors/emails/fangliquan@qq.com new file mode 100644 index 000000000000..58b0678f6bd7 --- /dev/null +++ b/contributors/emails/fangliquan@qq.com @@ -0,0 +1 @@ +fangliquanflq From 39173309a03bbc040d5299bb3c0579128a1d8a5e Mon Sep 17 00:00:00 2001 From: fangliquanflq Date: Tue, 4 Aug 2026 11:39:11 +0800 Subject: [PATCH 3/3] chore(contributors): normalize email mapping line endings to LF --- contributors/emails/fangliquan@qq.com | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/contributors/emails/fangliquan@qq.com b/contributors/emails/fangliquan@qq.com index 2d2320ef90ba..b1e421acaef0 100644 --- a/contributors/emails/fangliquan@qq.com +++ b/contributors/emails/fangliquan@qq.com @@ -1,2 +1,2 @@ -fangliquanflq -# PR #73031 author email +fangliquanflq +# PR #73031 author email