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

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"