From d9d9ff970880949ba00f0c02837b07bba8cc879d Mon Sep 17 00:00:00 2001 From: Gumayusy <2732381414@qq.com> Date: Sun, 4 Oct 2026 10:43:05 +0800 Subject: [PATCH] fix: fence stale notification delivery claims (#6211) * fix: fence stale notification delivery claims * test: cover notification claim migration upgrades --------- Co-authored-by: Willem Jiang --- .../app/scheduler/notification_delivery.py | 13 +- .../deerflow/persistence/migrations/AGENTS.md | 3 +- .../0030_notification_claim_tokens.py | 28 ++++ .../notification_deliveries/model.py | 3 + .../notification_deliveries/sql.py | 137 ++++++++++++++---- ...migration_0027_0028_notification_outbox.py | 57 +++++++- ...st_migration_0029_scheduler_agent_tasks.py | 6 +- ...test_notification_deliveries_repository.py | 90 ++++++++++-- .../test_notification_delivery_worker.py | 4 +- 9 files changed, 284 insertions(+), 57 deletions(-) create mode 100644 backend/packages/harness/deerflow/persistence/migrations/versions/0030_notification_claim_tokens.py diff --git a/backend/app/scheduler/notification_delivery.py b/backend/app/scheduler/notification_delivery.py index cd77550ebf..3de4d187b5 100644 --- a/backend/app/scheduler/notification_delivery.py +++ b/backend/app/scheduler/notification_delivery.py @@ -191,7 +191,7 @@ class NotificationDeliveryWorker: # reclaims the row on a later poll. logger.exception("Notification delivery %s crashed; isolating from the rest of the batch", row.get("id")) try: - await self._delivery_repo.mark_failed(row["id"], error="delivery crashed before completion") + await self._delivery_repo.mark_failed(row["id"], claim_token=row.get("claim_token"), error="delivery crashed before completion") except Exception: logger.warning("Could not mark crashed delivery %s as failed; stale reset will recover it", row.get("id"), exc_info=True) @@ -293,12 +293,13 @@ class NotificationDeliveryWorker: if still_connected is False: await self._delivery_repo.mark_failed( delivery_id, + claim_token=delivery.get("claim_token"), error=f"target is no longer a connected {provider} identity of its owner", terminal=True, ) return if still_connected is None: - await self._delivery_repo.mark_failed(delivery_id, error="could not verify the target's channel connection", count_attempt=False) + await self._delivery_repo.mark_failed(delivery_id, claim_token=delivery.get("claim_token"), error="could not verify the target's channel connection", count_attempt=False) return # Re-check channel liveness at delivery time, not enqueue time: the # channel may have been disabled or disconnected after the outbox row @@ -310,7 +311,7 @@ class NotificationDeliveryWorker: # Channel outage is not the delivery's fault: park the row # without consuming its retry budget so it survives an # hours-long outage and delivers once the channel returns. - await self._delivery_repo.mark_failed(delivery_id, error=f"channel '{provider}' is not running", count_attempt=False) + await self._delivery_repo.mark_failed(delivery_id, claim_token=delivery.get("claim_token"), error=f"channel '{provider}' is not running", count_attempt=False) return enriched = delivery if delivery.get("event") == "run_completed": @@ -326,12 +327,12 @@ class NotificationDeliveryWorker: text_markdown=render_notification_text(enriched), ) except ChannelUnavailable as exc: - await self._delivery_repo.mark_failed(delivery_id, error=str(exc), count_attempt=False) + await self._delivery_repo.mark_failed(delivery_id, claim_token=delivery.get("claim_token"), error=str(exc), count_attempt=False) return except Exception as exc: - await self._delivery_repo.mark_failed(delivery_id, error=str(exc)) + await self._delivery_repo.mark_failed(delivery_id, claim_token=delivery.get("claim_token"), error=str(exc)) return - await self._delivery_repo.mark_sent(delivery_id) + await self._delivery_repo.mark_sent(delivery_id, claim_token=delivery.get("claim_token")) async def _run_loop(self) -> None: while not self._stop.is_set(): diff --git a/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md b/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md index ac01e4aec5..60e1535186 100644 --- a/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md +++ b/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md @@ -26,7 +26,8 @@ The empty-DB path keeps using `create_all` because `Base.metadata` is the only a `0023_run_change_seq` → `0023_user_preferences` → `0024_project_documents` → `0025_repair_run_change_seq` → `0026_mcp_task_lease_tokens` → `0027_notification_deliveries` → -`0028_parked_attempts` (current head). The preference +`0028_parked_attempts` → `0029_scheduler_agent_tasks` → +`0030_notification_claim_tokens` (current head). The preference revision adds a separate owner/key table with a cascading users foreign key and does not alter users; the project-documents revision adds a new owner-scoped shelf table, and the MCP lease-token revision adds two nullable token columns to diff --git a/backend/packages/harness/deerflow/persistence/migrations/versions/0030_notification_claim_tokens.py b/backend/packages/harness/deerflow/persistence/migrations/versions/0030_notification_claim_tokens.py new file mode 100644 index 0000000000..136f01049e --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/migrations/versions/0030_notification_claim_tokens.py @@ -0,0 +1,28 @@ +"""Add fencing tokens to notification delivery claims. + +Revision ID: 0030_notification_claim_tokens +Revises: 0029_scheduler_agent_tasks +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa + +revision: str = "0030_notification_claim_tokens" +down_revision: str | Sequence[str] | None = "0029_scheduler_agent_tasks" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + from deerflow.persistence.migrations._helpers import safe_add_column + + safe_add_column("notification_deliveries", sa.Column("claim_token", sa.String(length=64), nullable=True)) + + +def downgrade() -> None: + from deerflow.persistence.migrations._helpers import safe_drop_column + + safe_drop_column("notification_deliveries", "claim_token") diff --git a/backend/packages/harness/deerflow/persistence/notification_deliveries/model.py b/backend/packages/harness/deerflow/persistence/notification_deliveries/model.py index e3032c2b2a..44f78d78ef 100644 --- a/backend/packages/harness/deerflow/persistence/notification_deliveries/model.py +++ b/backend/packages/harness/deerflow/persistence/notification_deliveries/model.py @@ -37,6 +37,9 @@ class NotificationDeliveryRow(Base): # pending -> sending -> sent | failed. A failure with retries remaining # returns the row to pending with a backoff ``available_at``. status: Mapped[str] = mapped_column(String(32), nullable=False, default="pending") + # Unique fencing token for the current sending claim. Completion writes + # must present the token they received when claiming the row. + claim_token: Mapped[str | None] = mapped_column(String(64), nullable=True) attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0) # Incremented on channel-down / transport parking (``count_attempt=False``) # so permanently absent channels eventually reach a terminal ``failed`` diff --git a/backend/packages/harness/deerflow/persistence/notification_deliveries/sql.py b/backend/packages/harness/deerflow/persistence/notification_deliveries/sql.py index 4f9f57e4af..86ae622277 100644 --- a/backend/packages/harness/deerflow/persistence/notification_deliveries/sql.py +++ b/backend/packages/harness/deerflow/persistence/notification_deliveries/sql.py @@ -3,8 +3,10 @@ The completion hook calls :meth:`NotificationDeliveryRepository.enqueue` (idempotent), and the delivery worker uses ``claim_due_deliveries`` / ``mark_sent`` / ``mark_failed``. Claim flips rows ``pending -> sending`` -inside one transaction guarded by ``status = 'pending'``, so concurrent -workers (multi-pod deployments) cannot double-send the same row. +inside one transaction guarded by ``status = 'pending'`` and stamps a fresh +``claim_token``. Completion writes that carry the token are conditionally +updated only while that claim owns the row, fencing stale workers after +reconciliation. """ from __future__ import annotations @@ -191,6 +193,8 @@ class NotificationDeliveryRepository: re-SELECT would also match rows a concurrent worker flipped in the same window and hand them out twice -- the multi-pod double-send the docstring above rules out. + + Each claim stamps a token that completion methods use for fencing. """ if limit <= 0: return [] @@ -220,7 +224,7 @@ class NotificationDeliveryRepository: NotificationDeliveryRow.id.in_(due_ids), NotificationDeliveryRow.status == "pending", ) - .values(status="sending", updated_at=now) + .values(status="sending", claim_token=uuid.uuid4().hex, updated_at=now) .returning(NotificationDeliveryRow) ) ) @@ -249,12 +253,12 @@ class NotificationDeliveryRepository: NotificationDeliveryRow.status == "sending", NotificationDeliveryRow.updated_at <= cutoff, ) - .values(status="pending", updated_at=now) + .values(status="pending", claim_token=None, updated_at=now) ) await session.commit() return result.rowcount - async def mark_sent(self, delivery_id: str) -> dict[str, Any]: + async def mark_sent(self, delivery_id: str, *, claim_token: str | None = None) -> dict[str, Any]: async with self.session_factory() as session: row = await session.get(NotificationDeliveryRow, delivery_id) if row is None: @@ -265,12 +269,100 @@ class NotificationDeliveryRepository: # must not reopen or relabel it. logger.info("notification delivery %s is already %s; ignoring late mark_sent", delivery_id, row.status) return self._to_dict(row) + if claim_token is not None: + finalized = await self._fenced_completion( + session, + delivery_id, + claim_token, + values={"status": "sent", "sent_at": datetime.now(UTC), "claim_token": None}, + ) + if finalized is None: + return await self._fenced_out_row(session, delivery_id, claim_token, "mark_sent") + await session.commit() + return self._to_dict(finalized) row.status = "sent" + row.claim_token = None row.sent_at = datetime.now(UTC) await session.commit() await session.refresh(row) return self._to_dict(row) + @staticmethod + async def _fenced_completion( + session: AsyncSession, + delivery_id: str, + claim_token: str, + *, + values: dict[str, Any], + ) -> NotificationDeliveryRow | None: + """Apply a completion only while the token still owns ``sending``.""" + values = {**values, "updated_at": datetime.now(UTC)} + result = await session.execute( + update(NotificationDeliveryRow) + .where( + NotificationDeliveryRow.id == delivery_id, + NotificationDeliveryRow.status == "sending", + NotificationDeliveryRow.claim_token == claim_token, + ) + .values(**values) + .returning(NotificationDeliveryRow) + ) + return result.scalars().first() + + async def _fenced_out_row( + self, + session: AsyncSession, + delivery_id: str, + claim_token: str, + operation: str, + ) -> dict[str, Any]: + logger.info( + "notification delivery %s is no longer owned by claim %s; ignoring late %s", + delivery_id, + claim_token[:8], + operation, + ) + await session.rollback() + current = await session.get(NotificationDeliveryRow, delivery_id) + if current is None: + raise LookupError(f"notification delivery {delivery_id} not found") + return self._to_dict(current) + + @staticmethod + def _failure_outcome( + row: NotificationDeliveryRow, + *, + error: str | None, + count_attempt: bool, + terminal: bool, + now: datetime, + ) -> dict[str, Any]: + values: dict[str, Any] = {"last_error": error, "claim_token": None, "updated_at": now} + if terminal: + values["status"] = "failed" + elif not count_attempt: + created_at = row.created_at + if created_at.tzinfo is None: + created_at = created_at.replace(tzinfo=UTC) + park_expired = (now - created_at) >= _CHANNEL_PARK_MAX_AGE or row.parked_attempts >= _CHANNEL_PARK_MAX_ATTEMPTS + if park_expired: + values["status"] = "failed" + values["last_error"] = error or "channel did not return before parking limit" + else: + values["parked_attempts"] = row.parked_attempts + 1 + values["status"] = "pending" + values["available_at"] = now + timedelta(seconds=_CHANNEL_DOWN_RETRY_SECONDS) + else: + attempts = row.attempts + 1 + values["attempts"] = attempts + if attempts >= row.max_attempts: + values["status"] = "failed" + else: + delay = min(_RETRY_BASE_SECONDS * 2 ** (attempts - 1), _RETRY_MAX_SECONDS) + values["status"] = "pending" + values["available_at"] = now + timedelta(seconds=delay) + return values + async def mark_failed( self, delivery_id: str, @@ -278,6 +370,7 @@ class NotificationDeliveryRepository: error: str | None = None, count_attempt: bool = True, terminal: bool = False, + claim_token: str | None = None, ) -> dict[str, Any]: """Record a failed attempt; reschedule with backoff while retries remain, otherwise finalize the row as ``failed``. @@ -305,30 +398,16 @@ class NotificationDeliveryRepository: # pending and send it later after all. logger.info("notification delivery %s is already %s; ignoring late mark_failed", delivery_id, row.status) return self._to_dict(row) - row.last_error = error - if terminal: - row.status = "failed" - elif not count_attempt: - now = datetime.now(UTC) - created_at = row.created_at - if created_at.tzinfo is None: - created_at = created_at.replace(tzinfo=UTC) - park_expired = (now - created_at) >= _CHANNEL_PARK_MAX_AGE or row.parked_attempts >= _CHANNEL_PARK_MAX_ATTEMPTS - if park_expired: - row.status = "failed" - row.last_error = error or "channel did not return before parking limit" - else: - row.parked_attempts += 1 - row.status = "pending" - row.available_at = now + timedelta(seconds=_CHANNEL_DOWN_RETRY_SECONDS) - else: - row.attempts += 1 - if row.attempts >= row.max_attempts: - row.status = "failed" - else: - delay = min(_RETRY_BASE_SECONDS * 2 ** (row.attempts - 1), _RETRY_MAX_SECONDS) - row.status = "pending" - row.available_at = datetime.now(UTC) + timedelta(seconds=delay) + now = datetime.now(UTC) + values = self._failure_outcome(row, error=error, count_attempt=count_attempt, terminal=terminal, now=now) + if claim_token is not None: + updated = await self._fenced_completion(session, delivery_id, claim_token, values=values) + if updated is None: + return await self._fenced_out_row(session, delivery_id, claim_token, "mark_failed") + await session.commit() + return self._to_dict(updated) + for name, value in values.items(): + setattr(row, name, value) await session.commit() await session.refresh(row) return self._to_dict(row) diff --git a/backend/tests/test_migration_0027_0028_notification_outbox.py b/backend/tests/test_migration_0027_0028_notification_outbox.py index 3fd7739803..0fe4be6931 100644 --- a/backend/tests/test_migration_0027_0028_notification_outbox.py +++ b/backend/tests/test_migration_0027_0028_notification_outbox.py @@ -1,8 +1,8 @@ """Migration tests for the scheduled-task notification outbox (issue #4254). ``0027_notification_deliveries`` creates the outbox table and -``0028_parked_attempts`` adds its parking counter. The latest scheduler-agent -migration test now owns the chain-head pin. +``0028_parked_attempts`` adds its parking counter. The scheduler-agent and +notification claim-token migrations extend that chain in order. """ from __future__ import annotations @@ -21,6 +21,8 @@ pytestmark = pytest.mark.asyncio OUTBOX = "0027_notification_deliveries" PARKED = "0028_parked_attempts" +SCHEDULER = "0029_scheduler_agent_tasks" +CLAIM_TOKENS = "0030_notification_claim_tokens" PREVIOUS = "0026_mcp_task_lease_tokens" TABLE = "notification_deliveries" @@ -30,8 +32,10 @@ async def test_outbox_revisions_chain_after_0026(): assert len(script.get_heads()) == 1 assert script.get_revision(OUTBOX).down_revision == PREVIOUS assert script.get_revision(PARKED).down_revision == OUTBOX + assert script.get_revision(SCHEDULER).down_revision == PARKED + assert script.get_revision(CLAIM_TOKENS).down_revision == SCHEDULER # alembic_version.version_num is VARCHAR(32); a longer id fails on Postgres. - assert max(len(OUTBOX), len(PARKED)) <= 32 + assert max(len(OUTBOX), len(PARKED), len(SCHEDULER), len(CLAIM_TOKENS)) <= 32 async def test_outbox_revisions_upgrade_and_downgrade(tmp_path): @@ -56,14 +60,61 @@ async def test_outbox_revisions_upgrade_and_downgrade(tmp_path): await asyncio.to_thread(bootstrap._upgrade, cfg, OUTBOX) columns = await outbox_columns() assert columns is not None and "parked_attempts" not in columns + assert "claim_token" not in columns await asyncio.to_thread(bootstrap._upgrade, cfg, "head") columns = await outbox_columns() assert columns["parked_attempts"]["nullable"] is False # The backfill default is dropped so the schema matches create_all. assert columns["parked_attempts"]["default"] is None + assert columns["claim_token"]["nullable"] is True await asyncio.to_thread(command.downgrade, cfg, PREVIOUS) assert await outbox_columns() is None finally: await engine.dispose() + + +async def test_bootstrap_claim_tokens_preserves_existing_scheduler_and_outbox(tmp_path): + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'scheduler.db'}") + cfg = bootstrap._get_alembic_config(engine) + try: + await asyncio.to_thread(bootstrap._upgrade, cfg, SCHEDULER) + async with engine.begin() as conn: + await conn.execute( + sa.text( + "INSERT INTO scheduled_tasks " + "(id,user_id,context_mode,title,prompt,schedule_type,schedule_spec,timezone,status,overlap_policy,run_count,goal_objective,created_at,updated_at) " + "VALUES ('task','owner','fresh_thread_per_run','Goal','Prompt','cron','{}','UTC','enabled','enqueue',7,'deliver',CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)" + ) + ) + await conn.execute( + sa.text( + "INSERT INTO notification_deliveries " + "(id,task_id,task_run_id,event,provider,target,owner_user_id,status,attempts,parked_attempts,max_attempts,payload_json,available_at,created_at,updated_at) " + "VALUES ('delivery','task','occurrence','success','telegram','target','owner','pending',2,1,5,'{}',CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)" + ) + ) + + async def assert_preserved(): + async with engine.connect() as conn: + assert (await conn.execute(sa.text("SELECT run_count,goal_objective FROM scheduled_tasks WHERE id='task'"))).one() == (7, "deliver") + assert (await conn.execute(sa.text("SELECT status,attempts,parked_attempts FROM notification_deliveries WHERE id='delivery'"))).one() == ("pending", 2, 1) + + for _ in range(2): + await bootstrap.bootstrap_schema(engine, backend="sqlite") + await assert_preserved() + async with engine.connect() as conn: + assert (await conn.execute(sa.text("SELECT version_num FROM alembic_version"))).scalar_one() == CLAIM_TOKENS + assert (await conn.execute(sa.text("SELECT claim_token FROM notification_deliveries WHERE id='delivery'"))).scalar_one() is None + + await asyncio.to_thread(command.downgrade, cfg, SCHEDULER) + await assert_preserved() + async with engine.connect() as conn: + columns = await conn.run_sync(lambda sync: {col["name"] for col in sa.inspect(sync).get_columns(TABLE)}) + assert "claim_token" not in columns + assert (await conn.execute(sa.text("SELECT version_num FROM alembic_version"))).scalar_one() == SCHEDULER + await bootstrap.bootstrap_schema(engine, backend="sqlite") + await assert_preserved() + finally: + await engine.dispose() diff --git a/backend/tests/test_migration_0029_scheduler_agent_tasks.py b/backend/tests/test_migration_0029_scheduler_agent_tasks.py index 2b600e08d6..62016b0226 100644 --- a/backend/tests/test_migration_0029_scheduler_agent_tasks.py +++ b/backend/tests/test_migration_0029_scheduler_agent_tasks.py @@ -23,16 +23,18 @@ from deerflow.persistence.postgres_schema import build_asyncpg_connect_args from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository REVISION = "0029_scheduler_agent_tasks" +CURRENT_HEAD = "0030_notification_claim_tokens" PREVIOUS = "0028_parked_attempts" TASK_FIELDS = {"origin_thread_id", "goal_objective", "max_runs", "end_at", "standing_notes"} OCCURRENCE_FIELDS = {"goal_objective", "goal_verdict", "stop_requested_run_id"} pytestmark = pytest.mark.asyncio -async def test_0029_owns_single_chain_head(): +async def test_0029_remains_in_the_single_migration_chain(): script = ScriptDirectory(str(bootstrap._MIGRATIONS_DIR)) - assert script.get_heads() == [REVISION] + assert script.get_heads() == [CURRENT_HEAD] assert script.get_revision(REVISION).down_revision == PREVIOUS + assert script.get_revision(CURRENT_HEAD).down_revision == REVISION assert len(REVISION) <= 32 diff --git a/backend/tests/test_notification_deliveries_repository.py b/backend/tests/test_notification_deliveries_repository.py index d9ed0669c7..b563e5097e 100644 --- a/backend/tests/test_notification_deliveries_repository.py +++ b/backend/tests/test_notification_deliveries_repository.py @@ -188,6 +188,11 @@ class TestNotificationDeliveryRepository: assert await repo.reset_stale_sending_rows(now=datetime.now(UTC) + timedelta(minutes=11), timeout=timedelta(minutes=10)) == 1 + async with repo.session_factory() as session: + recovered = await session.get(NotificationDeliveryRow, stuck["id"]) + assert recovered is not None + assert recovered.claim_token is None + reclaimed = await repo.claim_due_deliveries(now=datetime.now(UTC) + timedelta(minutes=11), limit=1) assert [row["id"] for row in reclaimed] == [stuck["id"]] @@ -196,7 +201,7 @@ class TestNotificationDeliveryRepository: await repo.enqueue(**_enqueue_kwargs(task_run_id="run-sent")) await repo.enqueue(**_enqueue_kwargs(task_run_id="run-pending")) claimed = await repo.claim_due_deliveries(now=datetime.now(UTC), limit=1) - await repo.mark_sent(claimed[0]["id"]) + await repo.mark_sent(claimed[0]["id"], claim_token=claimed[0]["claim_token"]) reset = await repo.reset_stale_sending_rows(now=datetime.now(UTC) + timedelta(hours=1), timeout=timedelta(minutes=10)) @@ -206,7 +211,8 @@ class TestNotificationDeliveryRepository: async def test_mark_sent_finalises_delivery(self, repo): delivery = await repo.enqueue(**_enqueue_kwargs()) - updated = await repo.mark_sent(delivery["id"]) + claimed = (await repo.claim_due_deliveries(now=datetime.now(UTC), limit=1))[0] + updated = await repo.mark_sent(delivery["id"], claim_token=claimed["claim_token"]) assert updated["status"] == "sent" assert updated["sent_at"] is not None @@ -216,7 +222,8 @@ class TestNotificationDeliveryRepository: delivery = await repo.enqueue(**_enqueue_kwargs()) before = datetime.now(UTC) - updated = await repo.mark_failed(delivery["id"], error="wecom: rate limited") + claimed = (await repo.claim_due_deliveries(now=datetime.now(UTC), limit=1))[0] + updated = await repo.mark_failed(delivery["id"], claim_token=claimed["claim_token"], error="wecom: rate limited") assert updated["status"] == "pending" assert updated["attempts"] == 1 @@ -232,7 +239,8 @@ class TestNotificationDeliveryRepository: # disconnected the target): no backoff, no parking, budget untouched. delivery = await repo.enqueue(**_enqueue_kwargs()) - updated = await repo.mark_failed(delivery["id"], error="target is no longer connected", terminal=True) + claimed = (await repo.claim_due_deliveries(now=datetime.now(UTC), limit=1))[0] + updated = await repo.mark_failed(delivery["id"], claim_token=claimed["claim_token"], error="target is no longer connected", terminal=True) assert updated["status"] == "failed" assert updated["attempts"] == 0 @@ -245,19 +253,69 @@ class TestNotificationDeliveryRepository: # A late write from a crash handler or a stale claimant must neither # reopen a finished row nor relabel it. dropped = await repo.enqueue(**_enqueue_kwargs()) - await repo.mark_failed(dropped["id"], error="target is no longer connected", terminal=True) - after_late_failure = await repo.mark_failed(dropped["id"], error="delivery crashed before completion") - after_late_sent = await repo.mark_sent(dropped["id"]) + dropped_claim = (await repo.claim_due_deliveries(now=datetime.now(UTC), limit=1))[0] + await repo.mark_failed(dropped["id"], claim_token=dropped_claim["claim_token"], error="target is no longer connected", terminal=True) + after_late_failure = await repo.mark_failed(dropped["id"], claim_token=dropped_claim["claim_token"], error="delivery crashed before completion") + after_late_sent = await repo.mark_sent(dropped["id"], claim_token=dropped_claim["claim_token"]) assert after_late_failure["status"] == "failed" assert after_late_sent["status"] == "failed" assert after_late_sent["last_error"] == "target is no longer connected" assert after_late_sent["attempts"] == 0 sent = await repo.enqueue(**_enqueue_kwargs(target="other-target")) - await repo.mark_sent(sent["id"]) - after_late_failure = await repo.mark_failed(sent["id"], error="late failure", count_attempt=False) + sent_claim = (await repo.claim_due_deliveries(now=datetime.now(UTC), limit=1))[0] + await repo.mark_sent(sent["id"], claim_token=sent_claim["claim_token"]) + after_late_failure = await repo.mark_failed(sent["id"], claim_token=sent_claim["claim_token"], error="late failure", count_attempt=False) assert after_late_failure["status"] == "sent" assert after_late_failure["last_error"] is None + + @pytest.mark.anyio + async def test_stale_worker_cannot_finalize_delivery_after_reclaim(self, repo): + """A recovered delivery must reject writes from its previous claim. + + Worker A claims the row, then its sending lease is recovered after a + timeout. Worker B claims the same row. A late failure from Worker A + must not rewrite B's active claim or consume another retry attempt. + """ + claimed_a = await repo.enqueue(**_enqueue_kwargs()) + first_claimed = await repo.claim_due_deliveries(now=datetime.now(UTC), limit=1) + assert first_claimed[0]["id"] == claimed_a["id"] + + recovery_time = datetime.now(UTC) + timedelta(minutes=11) + assert ( + await repo.reset_stale_sending_rows( + now=recovery_time, + timeout=timedelta(minutes=10), + ) + == 1 + ) + + second_claimed = await repo.claim_due_deliveries(now=recovery_time, limit=1) + assert second_claimed[0]["id"] == claimed_a["id"] + + late_result = await repo.mark_failed( + first_claimed[0]["id"], + claim_token=first_claimed[0]["claim_token"], + error="late result from stale worker", + ) + + assert late_result["status"] == "sending" + assert late_result["attempts"] == 0 + assert late_result["last_error"] is None + late_terminal = await repo.mark_failed( + first_claimed[0]["id"], + claim_token=first_claimed[0]["claim_token"], + error="late terminal result from stale worker", + terminal=True, + ) + assert late_terminal["status"] == "sending" + assert late_terminal["last_error"] is None + late_sent = await repo.mark_sent( + first_claimed[0]["id"], + claim_token=first_claimed[0]["claim_token"], + ) + assert late_sent["status"] == "sending" + assert late_sent["sent_at"] is None assert await repo.claim_due_deliveries(now=datetime.now(UTC) + timedelta(days=2), limit=10) == [] @pytest.mark.anyio @@ -268,7 +326,7 @@ class TestNotificationDeliveryRepository: for _ in range(delivery["max_attempts"]): claimed = await repo.claim_due_deliveries(now=datetime.now(UTC) + timedelta(days=1), limit=1) assert len(claimed) == 1 - updated = await repo.mark_failed(claimed[0]["id"], error="boom") + updated = await repo.mark_failed(claimed[0]["id"], claim_token=claimed[0]["claim_token"], error="boom") assert updated["status"] == "failed" # A failed (exhausted) row is never claimed again. @@ -281,7 +339,8 @@ class TestNotificationDeliveryRepository: delivery = await repo.enqueue(**_enqueue_kwargs()) before = datetime.now(UTC) - updated = await repo.mark_failed(delivery["id"], error="channel 'wecom' is not running", count_attempt=False) + claimed = (await repo.claim_due_deliveries(now=datetime.now(UTC), limit=1))[0] + updated = await repo.mark_failed(delivery["id"], claim_token=claimed["claim_token"], error="channel 'wecom' is not running", count_attempt=False) assert updated["status"] == "pending" assert updated["attempts"] == 0 @@ -292,11 +351,13 @@ class TestNotificationDeliveryRepository: assert retry_at >= before + timedelta(minutes=10) for _ in range(_CHANNEL_PARK_MAX_ATTEMPTS - 1): - updated = await repo.mark_failed(updated["id"], error="channel still down", count_attempt=False) + claimed = (await repo.claim_due_deliveries(now=datetime.now(UTC) + timedelta(days=1), limit=1))[0] + updated = await repo.mark_failed(claimed["id"], claim_token=claimed["claim_token"], error="channel still down", count_attempt=False) assert updated["status"] == "pending" assert updated["attempts"] == 0 - updated = await repo.mark_failed(updated["id"], error="channel still down", count_attempt=False) + claimed = (await repo.claim_due_deliveries(now=datetime.now(UTC) + timedelta(days=1), limit=1))[0] + updated = await repo.mark_failed(claimed["id"], claim_token=claimed["claim_token"], error="channel still down", count_attempt=False) assert updated["status"] == "failed" assert updated["attempts"] == 0 @@ -308,7 +369,8 @@ class TestNotificationDeliveryRepository: row.created_at = datetime.now(UTC) - _CHANNEL_PARK_MAX_AGE - timedelta(minutes=1) await session.commit() - updated = await repo.mark_failed(delivery["id"], error="channel 'wecom' is not running", count_attempt=False) + claimed = (await repo.claim_due_deliveries(now=datetime.now(UTC), limit=1))[0] + updated = await repo.mark_failed(delivery["id"], claim_token=claimed["claim_token"], error="channel 'wecom' is not running", count_attempt=False) assert updated["status"] == "failed" assert updated["attempts"] == 0 diff --git a/backend/tests/test_notification_delivery_worker.py b/backend/tests/test_notification_delivery_worker.py index 465da93cc7..552e50735f 100644 --- a/backend/tests/test_notification_delivery_worker.py +++ b/backend/tests/test_notification_delivery_worker.py @@ -39,10 +39,10 @@ class FakeDeliveryRepo: self.events.append("claim") return claimed - async def mark_sent(self, delivery_id): + async def mark_sent(self, delivery_id, *, claim_token=None): self.sent.append(delivery_id) - async def mark_failed(self, delivery_id, *, error=None, count_attempt=True, terminal=False): + async def mark_failed(self, delivery_id, *, claim_token=None, error=None, count_attempt=True, terminal=False): self.failed.append((delivery_id, error, count_attempt)) if terminal: self.terminal_failures.append((delivery_id, error))