diff --git a/apps/api/services/outreach/sending.py b/apps/api/services/outreach/sending.py index 1457ae4e..8566290c 100644 --- a/apps/api/services/outreach/sending.py +++ b/apps/api/services/outreach/sending.py @@ -307,6 +307,8 @@ class _CapTrigger: # records the failure + schedules a retry. _settle_reservation(store, ws_id, reservation, ok=False) _finalize_send(store, idem, status="failed", error=err) + if enrollment_id is not None: + store.advance_enrollment(enrollment_id, status="failed", error=err) raise RuntimeError(f"outreach send transient failure: {err[:160]}") diff --git a/apps/api/services/outreach/store.py b/apps/api/services/outreach/store.py index 4b4bf615..85a6774b 100644 --- a/apps/api/services/outreach/store.py +++ b/apps/api/services/outreach/store.py @@ -252,6 +252,12 @@ def due_enrollments(self, seq_id: str, limit: int = 200) -> List[dict]: OutreachEnrollment.sequence_id == seq_id, OutreachEnrollment.status.in_(("pending", "scheduled")), OutreachEnrollment.next_send_at <= now, + ~s.query(OutreachSend.id).filter( + OutreachSend.workspace_id == self.workspace_id, + OutreachSend.sequence_id == seq_id, + OutreachSend.enrollment_id == OutreachEnrollment.id, + OutreachSend.step_number == OutreachEnrollment.current_step, + ).exists(), ) .order_by(OutreachEnrollment.next_send_at.asc()) .limit(limit) diff --git a/tests/test_outreach_send.py b/tests/test_outreach_send.py index b1cfb9e8..29c404b7 100644 --- a/tests/test_outreach_send.py +++ b/tests/test_outreach_send.py @@ -367,3 +367,65 @@ async def _handle(ws, payload): with workspace_scope(ws): store = PgOutreachStore(ws) await _do_send(store, ws, payload["idempotency_key"], payload) + + +@pytest.mark.parametrize("historical_pending", [False, True]) +def test_failed_send_does_not_starve_later_due_enrollment(SL, env, monkeypatch, historical_pending): + from apps.api.services.outreach import sender, sending + from apps.api.services.outreach.store import PgOutreachStore + + calls = [] + + async def fail_send(**kwargs): + calls.append(kwargs["to_email"]) + return sender.SendResult(success=False, error="Connection refused") + + monkeypatch.setattr(sender, "send_email", fail_send) + monkeypatch.setattr(settings, "OUTREACH_SEND_COST_USD", 0) + monkeypatch.setattr(settings, "OUTREACH_TICK_MAX_ENQUEUE", 1) + store = PgOutreachStore(WS) + seq = _mk_active_seq(store) + first_id = store.enroll(seq["id"], 1, "first@example.com") + first_idem = sending.sequence_step_idem(WS, seq["id"], first_id, 0) + payload = {"workspace_id": WS, "sequence_id": seq["id"], "enrollment_id": first_id, + "lead_id": 1, "step_number": 0, "to_email": "first@example.com", + "idempotency_key": first_idem} + with pytest.raises(RuntimeError, match="transient failure"): + _run(sending.handle_send(1, payload)) + assert store.get_send_by_idem(first_idem).status == "failed" + _run(sending.handle_send(1, payload)) + assert calls == ["first@example.com"] + if historical_pending: + # Existing deployments persisted failed sends with pending enrollments. + store.advance_enrollment(first_id, status="pending") + second_id = store.enroll(seq["id"], 2, "second@example.com") + assert sending.tick_sequence(WS, seq["id"]) == 1 + with SL() as db: + sends = db.query(Job).filter(Job.type == "send", Job.status == "pending").all() + assert len(sends) == 1 + assert sends[0].payload["enrollment_id"] == second_id + if not historical_pending: + assert store.get_enrollment(first_id).status == "failed" + assert store.get_enrollment(first_id).error == "Connection refused" + + +def test_sent_previous_step_does_not_hide_next_due_step(SL, env, monkeypatch): + from apps.api.services.outreach import sending + from apps.api.services.outreach.store import PgOutreachStore + + monkeypatch.setattr(settings, "OUTREACH_SEND_COST_USD", 0) + store = PgOutreachStore(WS) + seq = _mk_active_seq(store) + seq["steps"][1]["delay_hours"] = 0 + store.update_sequence(seq["id"], {"steps": seq["steps"]}) + enrollment_id = store.enroll(seq["id"], 1, "reader@example.com") + idem = sending.sequence_step_idem(WS, seq["id"], enrollment_id, 0) + _run(sending.handle_send(1, { + "workspace_id": WS, "sequence_id": seq["id"], "enrollment_id": enrollment_id, + "lead_id": 1, "step_number": 0, "to_email": "reader@example.com", "idempotency_key": idem, + })) + assert store.get_send_by_idem(idem).status == "sent" + assert sending.tick_sequence(WS, seq["id"]) == 1 + with SL() as db: + job = db.query(Job).filter(Job.type == "send", Job.status == "pending").one() + assert job.payload["enrollment_id"] == enrollment_id and job.payload["step_number"] == 1