From 21c7bf802e68c8fecdae41360c983ba84c135238 Mon Sep 17 00:00:00 2001 From: jagadish-zentiti Date: Sat, 11 Jul 2026 08:53:36 +0530 Subject: [PATCH] fix(email): atomically claim scheduled emails before sending (#5110) _scheduled_poll_once selected rows WHERE status='pending' and only wrote status='sent'/'failed' after the SMTP send and IMAP append completed - no atomic claim in between. Two overlapping callers (the in-process 30s poller and an externally cron/systemd-driven 'odysseus-mail poll-scheduled', or the CLI run manually) can both SELECT the same pending row before either UPDATEs it, and both send it. _start_poller's own docstring already names this exact risk ('avoid two copies of _scheduled_poll_once racing on the same SQLite') but nothing in the code enforced it - it was advisory only. Add an atomic per-row claim: UPDATE ... SET status='sending' WHERE id=? AND status='pending', proceeding only when rowcount == 1. The loser of the race sees rowcount == 0 and skips the row instead of sending a duplicate. Adds a regression test that drives two real threads through the real _scheduled_poll_once against a shared SQLite file, synchronized with a barrier and a widened send-path window, and asserts exactly one send fires. Reverting the fix makes the test fail reliably (5/5 runs); with the fix it passes reliably (5/5 runs). Fixes #5109 --- routes/email_pollers.py | 22 ++++++ tests/test_scheduled_poll_race.py | 116 ++++++++++++++++++++++++++++++ 2 files changed, 138 insertions(+) create mode 100644 tests/test_scheduled_poll_race.py diff --git a/routes/email_pollers.py b/routes/email_pollers.py index 13f846a71..8408103da 100644 --- a/routes/email_pollers.py +++ b/routes/email_pollers.py @@ -1068,6 +1068,28 @@ def _scheduled_poll_once() -> dict: for r in rows: sid = r[0] try: + # Atomically claim this row before doing any work. Two + # pollers can race here (the in-process asyncio task and an + # externally cron-driven `odysseus-mail poll-scheduled`, or + # an admin running the CLI manually alongside the in-process + # one despite the ODYSSEUS_INPROCESS_POLLERS=0 guidance) - + # both can SELECT the same 'pending' row before either has + # updated its status. The UPDATE...WHERE status='pending' is + # the atomicity boundary: only the poller whose UPDATE + # actually changes a row (rowcount == 1) proceeds to send; + # a loser sees rowcount == 0 and skips it instead of sending + # a duplicate. + claim_conn = sqlite3.connect(SCHEDULED_DB) + claim_cur = claim_conn.execute( + "UPDATE scheduled_emails SET status='sending' WHERE id=? AND status='pending'", + (sid,), + ) + claim_conn.commit() + claimed = claim_cur.rowcount == 1 + claim_conn.close() + if not claimed: + continue + attachments = json.loads(r[8] or "[]") row_account_id = r[9] if len(r) > 9 else None odysseus_kind = r[10] if len(r) > 10 else "scheduled" diff --git a/tests/test_scheduled_poll_race.py b/tests/test_scheduled_poll_race.py new file mode 100644 index 000000000..92575526e --- /dev/null +++ b/tests/test_scheduled_poll_race.py @@ -0,0 +1,116 @@ +"""Regression: two concurrent callers of `_scheduled_poll_once` (the +in-process 30s poller and the `odysseus-mail poll-scheduled` CLI, which the +project's own docstrings warn can race on the same SQLite when +ODYSSEUS_INPROCESS_POLLERS is left enabled alongside an external cron/systemd +driver) must not both send the same scheduled email. + +The old code selected pending rows, then only updated their status to 'sent' +*after* the SMTP send completed - two overlapping calls can both SELECT the +same 'pending' row before either UPDATEs it, so both send it. The fix adds +an atomic claim step (`UPDATE ... SET status='sending' WHERE status='pending'`) +before any work happens; only the caller whose UPDATE actually changes a row +proceeds, the other sees rowcount == 0 and skips it. + +This test drives two real threads through the real `_scheduled_poll_once` +against a shared SQLite file, synchronized with a barrier so both reach the +SELECT at (as close to) the same moment as possible, and asserts the send +callback fired exactly once. +""" +import sqlite3 +import threading +import time + + +def test_concurrent_pollers_do_not_double_send(tmp_path, monkeypatch): + import routes.email_helpers as email_helpers + import routes.email_pollers as email_pollers + + db_path = tmp_path / "scheduled_emails.db" + monkeypatch.setattr(email_helpers, "SCHEDULED_DB", db_path) + monkeypatch.setattr(email_pollers, "SCHEDULED_DB", db_path) + email_helpers._init_scheduled_db() + + conn = sqlite3.connect(db_path) + conn.execute( + """ + INSERT INTO scheduled_emails + (id, to_addr, subject, body, attachments, send_at, created_at, status, account_id, owner) + VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', ?, ?) + """, + ( + "sched-race-1", + "recipient@example.com", + "Subject", + "Body", + "[]", + "2000-01-01T00:00:00", + "1999-12-31T00:00:00", + "acct-alice", + "alice", + ), + ) + conn.commit() + conn.close() + + send_calls = [] + send_lock = threading.Lock() + barrier = threading.Barrier(2) + + def fake_get_email_config(account_id=None, owner=""): + return { + "from_address": "alice@example.com", + "smtp_host": "smtp.example.com", + "smtp_user": "alice@example.com", + "smtp_password": "secret", + } + + def fake_send_smtp_message(*args, **kwargs): + # Widen the window between the claim and the actual send so a + # buggy (unclaimed) second poller has every opportunity to also + # get past its SELECT and attempt to send. + time.sleep(0.05) + with send_lock: + send_calls.append(threading.get_ident()) + + class FakeImap: + def __init__(self, account_id=None, owner=""): + pass + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def append(self, folder, flags, date_time, message): + pass + + monkeypatch.setattr(email_pollers, "_get_email_config", fake_get_email_config) + monkeypatch.setattr(email_pollers, "_send_smtp_message", fake_send_smtp_message) + monkeypatch.setattr(email_pollers, "_imap", FakeImap) + monkeypatch.setattr(email_pollers, "_detect_sent_folder", lambda imap: "Sent") + monkeypatch.setattr(email_pollers, "_cleanup_compose_uploads", lambda attachments: None) + + results = [] + + def _run(): + barrier.wait() + results.append(email_pollers._scheduled_poll_once()) + + threads = [threading.Thread(target=_run) for _ in range(2)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=5) + + assert len(send_calls) == 1, ( + f"expected exactly one send for the racing pollers, got {len(send_calls)}: " + "the second poller must lose the atomic claim and skip the row" + ) + + conn = sqlite3.connect(db_path) + status = conn.execute( + "SELECT status FROM scheduled_emails WHERE id=?", ("sched-race-1",) + ).fetchone()[0] + conn.close() + assert status == "sent"