Fix poll logic for schedule email send

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:

backend/apps/workers/horoscope_email.py

@@ -90,26 +90,57 @@ 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)
+    sub_id = subscription["id"]
+
+    if status is None:
+        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 None:
+        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:
-        # 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)

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
+        repository=repository,
+        email_sender=email_sender,
+        horoscope_service=horoscope_service,  # ty:ignore[invalid-argument-type]
     )
 
 

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()

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)