From 7870d579cceb1d619a8f0c6968fb94401b654938 Mon Sep 17 00:00:00 2001 From: AndrewAct Date: Wed, 22 Jul 2026 00:12:48 -0500 Subject: [PATCH 1/5] Fix the logic of jump increasing of id in horoscope_deliveries --- .../horoscope_subscriptions/repository.py | 12 +++++ backend/apps/workers/horoscope_email.py | 24 +++++++--- ...test_horoscope_subscriptions_repository.py | 23 ++++++++++ .../services/test_worker_horoscope_email.py | 44 ++++++++++++------- 4 files changed, 79 insertions(+), 24 deletions(-) diff --git a/backend/apps/services/horoscope_subscriptions/repository.py b/backend/apps/services/horoscope_subscriptions/repository.py index 98e9bb7..4b15500 100644 --- a/backend/apps/services/horoscope_subscriptions/repository.py +++ b/backend/apps/services/horoscope_subscriptions/repository.py @@ -205,6 +205,18 @@ async def list_active(self) -> list[asyncpg.Record]: # --- deliveries --- + async def get_delivery_status(self, subscription_id: int, local_date: date) -> str | None: + """Read-only precheck the worker runs before attempting any claim/reclaim write. + Lets it skip subscriptions already resolved today without burning a bigserial id + on a doomed INSERT ... ON CONFLICT DO NOTHING. Returns None if no row exists yet.""" + async with self._get_pool().acquire() as conn: + row = await conn.fetchrow( + "SELECT status FROM horoscope_deliveries WHERE subscription_id = $1 AND local_date = $2", + subscription_id, + local_date, + ) + return row["status"] if row else None + async def claim_delivery( self, subscription_id: int, local_date: date, idempotency_key: str ) -> int | None: diff --git a/backend/apps/workers/horoscope_email.py b/backend/apps/workers/horoscope_email.py index 4ca35c3..c8ef20f 100644 --- a/backend/apps/workers/horoscope_email.py +++ b/backend/apps/workers/horoscope_email.py @@ -90,14 +90,20 @@ async def _claim_and_send( local_date = now_utc.astimezone(ZoneInfo(subscription["timezone"])).date() idempotency_key = f"horoscope:{subscription['id']}:{local_date.isoformat()}" - delivery_id = await deps.repository.claim_delivery( - subscription["id"], local_date, idempotency_key - ) - if delivery_id is None: + # Cheap read first, so a subscription already resolved today (sent/delivered) never + # reaches an INSERT — every prior tick for the rest of the day used to attempt one + # anyway and lose it to ON CONFLICT DO NOTHING, burning a bigserial id each time. + status = await deps.repository.get_delivery_status(subscription["id"], local_date) + + if status is None: + delivery_id = await deps.repository.claim_delivery( + subscription["id"], local_date, idempotency_key + ) + elif status == "failed": delivery_id = await deps.repository.reclaim_failed_delivery( subscription["id"], local_date, deps.max_attempts ) - if delivery_id is None: + elif status == "pending": # Covers a worker killed between claiming a delivery and recording its outcome # (e.g. SIGKILL during a deploy — no signal handler catches it, so the row is # simply abandoned at status='pending'). A row that old was never going to @@ -107,9 +113,13 @@ async def _claim_and_send( delivery_id = await deps.repository.reclaim_stale_pending_delivery( subscription["id"], local_date, deps.max_attempts, stale_before ) + else: + # 'sent', 'delivered', or any other terminal status — already resolved today. + delivery_id = None + if delivery_id is None: - # Already sent/delivered today, retries exhausted, or claimed by another worker - # between our read and write — all fine, nothing to do. + # Retries exhausted, nothing stale yet, or claimed by another worker between our + # read and write — all fine, nothing to do. return await deps.delivery_service.send_daily_horoscope(subscription, delivery_id, local_date) diff --git a/backend/tests/services/test_horoscope_subscriptions_repository.py b/backend/tests/services/test_horoscope_subscriptions_repository.py index 77d94aa..000b29d 100644 --- a/backend/tests/services/test_horoscope_subscriptions_repository.py +++ b/backend/tests/services/test_horoscope_subscriptions_repository.py @@ -176,6 +176,29 @@ async def test_list_active_filters_by_status(): assert "status = 'active'" in conn.fetch.await_args.args[0] +@pytest.mark.asyncio +async def test_get_delivery_status_returns_status_when_row_exists(): + pool, conn = make_pool() + conn.fetchrow.return_value = {"status": "sent"} + repo = SubscriptionRepository(pool=pool) + + result = await repo.get_delivery_status(1, date(2026, 7, 19)) + + assert result == "sent" + query, subscription_id, local_date = conn.fetchrow.await_args.args + assert "SELECT status FROM horoscope_deliveries" in query + assert (subscription_id, local_date) == (1, date(2026, 7, 19)) + + +@pytest.mark.asyncio +async def test_get_delivery_status_returns_none_when_no_row(): + pool, conn = make_pool() + conn.fetchrow.return_value = None + repo = SubscriptionRepository(pool=pool) + + assert await repo.get_delivery_status(1, date(2026, 7, 19)) is None + + @pytest.mark.asyncio async def test_claim_delivery_returns_id_on_success(): pool, conn = make_pool() diff --git a/backend/tests/services/test_worker_horoscope_email.py b/backend/tests/services/test_worker_horoscope_email.py index 423044b..b2263fa 100644 --- a/backend/tests/services/test_worker_horoscope_email.py +++ b/backend/tests/services/test_worker_horoscope_email.py @@ -22,6 +22,7 @@ def make_deps( ) -> WorkerDeps: repository = SimpleNamespace( list_active=AsyncMock(return_value=subscriptions), + get_delivery_status=AsyncMock(return_value=None), claim_delivery=AsyncMock(return_value=42), reclaim_failed_delivery=AsyncMock(return_value=None), reclaim_stale_pending_delivery=AsyncMock(return_value=None), @@ -56,6 +57,7 @@ async def test_tick_claims_and_sends_when_due(): result = await tick(deps) assert result == TickResult(due=1, sent=1, skipped=0) + deps.repository.get_delivery_status.assert_awaited_once_with(1, date(2026, 7, 19)) deps.repository.claim_delivery.assert_awaited_once_with( 1, date(2026, 7, 19), "horoscope:1:2026-07-19" ) @@ -65,26 +67,31 @@ async def test_tick_claims_and_sends_when_due(): @pytest.mark.asyncio -async def test_tick_skips_when_already_claimed_and_not_retryable(): +async def test_tick_skips_when_already_resolved_today(): + # status='sent' (or 'delivered', etc.) must short-circuit before any write is + # attempted — that's the whole point of the precheck: no wasted claim_delivery call. deps = make_deps([make_subscription()], now_utc=datetime(2026, 7, 19, 16, 0, tzinfo=UTC)) - deps.repository.claim_delivery.return_value = None - deps.repository.reclaim_failed_delivery.return_value = None + deps.repository.get_delivery_status.return_value = "sent" result = await tick(deps) assert result == TickResult(due=1, sent=0, skipped=0) deps.delivery_service.send_daily_horoscope.assert_not_awaited() + deps.repository.claim_delivery.assert_not_awaited() + deps.repository.reclaim_failed_delivery.assert_not_awaited() + deps.repository.reclaim_stale_pending_delivery.assert_not_awaited() @pytest.mark.asyncio async def test_tick_retries_via_reclaim_when_previous_attempt_failed(): deps = make_deps([make_subscription()], now_utc=datetime(2026, 7, 19, 16, 0, tzinfo=UTC)) - deps.repository.claim_delivery.return_value = None + deps.repository.get_delivery_status.return_value = "failed" deps.repository.reclaim_failed_delivery.return_value = 99 result = await tick(deps) assert result.sent == 1 + deps.repository.claim_delivery.assert_not_awaited() deps.repository.reclaim_failed_delivery.assert_awaited_once_with(1, date(2026, 7, 19), 5) deps.delivery_service.send_daily_horoscope.assert_awaited_once_with( make_subscription(), 99, date(2026, 7, 19) @@ -94,18 +101,18 @@ async def test_tick_retries_via_reclaim_when_previous_attempt_failed(): @pytest.mark.asyncio async def test_tick_reclaims_delivery_abandoned_mid_send_after_crash(): # Simulates a worker killed between claiming a delivery and recording its outcome - # (e.g. SIGKILL during a deploy): the row already exists (claim_delivery -> None) and - # isn't 'failed' (reclaim_failed_delivery -> None) — only the stale-pending path - # should resume it. + # (e.g. SIGKILL during a deploy): the row is still status='pending' — only the + # stale-pending path should resume it. now_utc = datetime(2026, 7, 19, 16, 0, tzinfo=UTC) deps = make_deps([make_subscription()], now_utc=now_utc, stale_pending_seconds=300) - deps.repository.claim_delivery.return_value = None - deps.repository.reclaim_failed_delivery.return_value = None + deps.repository.get_delivery_status.return_value = "pending" deps.repository.reclaim_stale_pending_delivery.return_value = 77 result = await tick(deps) assert result.sent == 1 + deps.repository.claim_delivery.assert_not_awaited() + deps.repository.reclaim_failed_delivery.assert_not_awaited() deps.repository.reclaim_stale_pending_delivery.assert_awaited_once_with( 1, date(2026, 7, 19), 5, now_utc - timedelta(seconds=300) ) @@ -116,11 +123,10 @@ async def test_tick_reclaims_delivery_abandoned_mid_send_after_crash(): @pytest.mark.asyncio async def test_tick_does_not_reclaim_when_nothing_is_stale(): - # claim_delivery/reclaim_failed_delivery/reclaim_stale_pending_delivery all miss — - # e.g. another (hypothetical) worker legitimately still has this in flight. + # status='pending' but not yet past the staleness window — e.g. another + # (hypothetical) worker legitimately still has this in flight. deps = make_deps([make_subscription()], now_utc=datetime(2026, 7, 19, 16, 0, tzinfo=UTC)) - deps.repository.claim_delivery.return_value = None - deps.repository.reclaim_failed_delivery.return_value = None + deps.repository.get_delivery_status.return_value = "pending" deps.repository.reclaim_stale_pending_delivery.return_value = None result = await tick(deps) @@ -181,7 +187,10 @@ async def test_dst_fall_back_repeated_hour_does_not_double_send(): repository = SimpleNamespace( list_active=AsyncMock(return_value=[subscription]), - claim_delivery=AsyncMock(side_effect=[42, None]), + # First pass: no row yet -> claim_delivery. Second pass: first pass's send + # already resolved the row to 'sent' -> precheck skips without a second claim. + get_delivery_status=AsyncMock(side_effect=[None, "sent"]), + claim_delivery=AsyncMock(return_value=42), reclaim_failed_delivery=AsyncMock(return_value=None), reclaim_stale_pending_delivery=AsyncMock(return_value=None), ) @@ -194,10 +203,11 @@ async def test_dst_fall_back_repeated_hour_does_not_double_send(): await tick(deps_second) assert delivery_service.send_daily_horoscope.await_count == 1 - # Both claim attempts must target the same local_date, despite the UTC offset + repository.claim_delivery.assert_awaited_once() + # Both precheck calls must target the same local_date, despite the UTC offset # differing between them (PDT vs PST) — otherwise this whole test would be vacuous. - first_call_date = repository.claim_delivery.await_args_list[0].args[1] - second_call_date = repository.claim_delivery.await_args_list[1].args[1] + first_call_date = repository.get_delivery_status.await_args_list[0].args[1] + second_call_date = repository.get_delivery_status.await_args_list[1].args[1] assert first_call_date == second_call_date == date(2026, 11, 1) From dc68c2b21888c0adfca1e378286d9204520ea3eb Mon Sep 17 00:00:00 2001 From: AndrewAct Date: Wed, 22 Jul 2026 13:04:40 -0500 Subject: [PATCH 2/5] Add logs --- backend/apps/workers/horoscope_email.py | 39 +++++++++++++++++++------ 1 file changed, 30 insertions(+), 9 deletions(-) diff --git a/backend/apps/workers/horoscope_email.py b/backend/apps/workers/horoscope_email.py index c8ef20f..391a830 100644 --- a/backend/apps/workers/horoscope_email.py +++ b/backend/apps/workers/horoscope_email.py @@ -94,28 +94,49 @@ async def _claim_and_send( # reaches an INSERT — every prior tick for the rest of the day used to attempt one # anyway and lose it to ON CONFLICT DO NOTHING, burning a bigserial id each time. status = await deps.repository.get_delivery_status(subscription["id"], local_date) + sub_id = subscription["id"] if status is None: - delivery_id = await deps.repository.claim_delivery( - subscription["id"], local_date, idempotency_key - ) + delivery_id = await deps.repository.claim_delivery(sub_id, local_date, idempotency_key) + if delivery_id is None: + # Precheck saw no row, yet the INSERT still lost to ON CONFLICT DO NOTHING — + # a bigserial id just got burned for nothing. Shouldn't happen with one + # worker; repeated hits mean two worker processes are live at once. + logger.warning( + "sub=%s date=%s claim lost a race after a clean precheck", sub_id, local_date + ) + else: + logger.info("sub=%s date=%s claimed delivery_id=%s", sub_id, local_date, delivery_id) elif status == "failed": delivery_id = await deps.repository.reclaim_failed_delivery( - subscription["id"], local_date, deps.max_attempts + sub_id, local_date, deps.max_attempts ) + if delivery_id is not None: + logger.info( + "sub=%s date=%s retried failed delivery_id=%s", sub_id, local_date, delivery_id + ) + else: + logger.warning( + "sub=%s date=%s gave up after max_attempts=%s", + sub_id, + local_date, + deps.max_attempts, + ) elif status == "pending": # Covers a worker killed between claiming a delivery and recording its outcome - # (e.g. SIGKILL during a deploy — no signal handler catches it, so the row is - # simply abandoned at status='pending'). A row that old was never going to - # resolve on its own; treat it the same as a failed attempt and give it another - # try, still bounded by max_attempts. + # (e.g. SIGKILL during a deployment). Bounded by max_attempts like the failed path. stale_before = now_utc - timedelta(seconds=deps.stale_pending_seconds) delivery_id = await deps.repository.reclaim_stale_pending_delivery( - subscription["id"], local_date, deps.max_attempts, stale_before + sub_id, local_date, deps.max_attempts, stale_before ) + if delivery_id is not None: + logger.info( + "sub=%s date=%s reclaimed stale delivery_id=%s", sub_id, local_date, delivery_id + ) else: # 'sent', 'delivered', or any other terminal status — already resolved today. delivery_id = None + logger.debug("sub=%s date=%s already resolved (status=%s)", sub_id, local_date, status) if delivery_id is None: # Retries exhausted, nothing stale yet, or claimed by another worker between our From 38e2a4dbf49d6c1aecec9c61b5bd90ef8094cab7 Mon Sep 17 00:00:00 2001 From: AndrewAct Date: Wed, 22 Jul 2026 13:15:18 -0500 Subject: [PATCH 3/5] Add ty hint --- backend/tests/services/test_horoscope_subscriptions_delivery.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/backend/tests/services/test_horoscope_subscriptions_delivery.py b/backend/tests/services/test_horoscope_subscriptions_delivery.py index 9bbb56a..ee8728a 100644 --- a/backend/tests/services/test_horoscope_subscriptions_delivery.py +++ b/backend/tests/services/test_horoscope_subscriptions_delivery.py @@ -36,7 +36,7 @@ def make_delivery_service() -> DeliveryService: get_daily_horoscope=AsyncMock(return_value=HOROSCOPE), ) return DeliveryService( - repository=repository, email_sender=email_sender, horoscope_service=horoscope_service + repository=repository, email_sender=email_sender, horoscope_service=horoscope_service # ty:ignore[invalid-argument-type] ) From 29842b3069797100c27401a3a6fa2db51bfda0d7 Mon Sep 17 00:00:00 2001 From: AndrewAct Date: Wed, 22 Jul 2026 13:26:44 -0500 Subject: [PATCH 4/5] Format code --- .../tests/services/test_horoscope_subscriptions_delivery.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/backend/tests/services/test_horoscope_subscriptions_delivery.py b/backend/tests/services/test_horoscope_subscriptions_delivery.py index ee8728a..fbc5acb 100644 --- a/backend/tests/services/test_horoscope_subscriptions_delivery.py +++ b/backend/tests/services/test_horoscope_subscriptions_delivery.py @@ -36,7 +36,9 @@ def make_delivery_service() -> DeliveryService: get_daily_horoscope=AsyncMock(return_value=HOROSCOPE), ) return DeliveryService( - repository=repository, email_sender=email_sender, horoscope_service=horoscope_service # ty:ignore[invalid-argument-type] + repository=repository, + email_sender=email_sender, + horoscope_service=horoscope_service, # ty:ignore[invalid-argument-type] ) From 713c6c5425733924b16508c28ecbc49bea307809 Mon Sep 17 00:00:00 2001 From: AndrewAct Date: Wed, 22 Jul 2026 22:55:03 -0500 Subject: [PATCH 5/5] some .idea changes --- .idea/modules.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.idea/modules.xml b/.idea/modules.xml index 4723288..09b0dc8 100644 --- a/.idea/modules.xml +++ b/.idea/modules.xml @@ -2,7 +2,7 @@ - +