Make message fetching concurrent with rate limiting and deadlock-free SQLite writes

[zbuc]

Closes #5

What changed

• SQLite fixed first, deadlock class eliminated. WAL journal mode and Synchronous::Normal are now actually enabled (they were commented-out stubs), plus a 5s busy_timeout. All writes go through a single writer task that owns the only write connection, fed by an mpsc channel — fetch tasks send (message_id, sender) and the writer commits one transaction per message. With exactly one writer, SQLite writer-vs-writer deadlocks cannot occur. The pool dropped from max_connections(100) to a small read-only pool sized to the fetch concurrency (8), used solely for the seen_mails checks, which WAL allows concurrently with the writer.
• Bounded concurrent fetching. parse_messages() now fans out via stream::iter(...).try_for_each_concurrent(8, ...) — bounded, no unbounded task-per-message spawns. The dead commented-out task::spawn scaffolding from 14f9d60/0be8d53 and the stale connect-option comments are deleted.
• Client-side rate limiting + per-message retry. Every messages_get waits on a shared 25ms-interval tick (~40 req/sec, under Gmail's ~250 quota units/sec at 5 units per messages.get). A 429/403 response backs off exponentially (1s, 2s, 4s, ... up to 5 retries) and retries just that message rather than crashing or restarting the mailbox scan.
• Sender-count logic (seen_mail / mark_seen / increment_sender_mails) is unchanged except for taking the pre-extracted message_id/sender strings so results can cross the channel; each message is still counted exactly once, so totals match a sequential run.
• Cargo.toml: added tokio sync + time features (mpsc, interval, sleep).

Verification

• cargo build succeeds.
• cargo clippy reports no new warnings — the sole warning (unused_mut on retries) exists identically on main.
• Not runtime-tested: this environment has no Gmail OAuth credentials, so I could not run the app end-to-end or empirically compare sender counts against a sequential run over a real mailbox. The equivalence argument is structural (same per-message read/insert/increment logic, single serialized writer); a real-mailbox comparison run is worth doing before relying on the numbers.

[zbuc]

Rebased onto main (picking up the CI workflow and clippy fixes) and pushed two fixes for review findings:

1. Rate-limit backoff never triggered. The retry arm in fetch_message matched only Error::Failure(response), but in google-gmail1 7.0 / google-apis-common 8.0 doit() returns Error::BadRequest(json) whenever the error body parses as JSON — and Gmail's 429/403 rate-limit responses always carry Google's standard JSON error envelope. So real rate limits fell into the generic Err(e) arm, failing the message, cancelling the page, and restarting the whole mailbox scan — exactly what issue #5's acceptance criterion 4 forbids. Added an is_rate_limit_error() helper that recognizes both variants: BadRequest via error.code == 429, error.status == "RESOURCE_EXHAUSTED", or rateLimitExceeded/userRateLimitExceeded reasons; Failure (non-JSON bodies) via HTTP status 429/403.

2. One writer-task DB error caused a permanent, silent livelock. db_writer exits on its first error (e.g. a transient database is locked past the 5s busy timeout), dropping the receiver — after which every write_tx.send() fails forever. The outer loop's retries += 1 was still commented out, so the retries > 3 cap never fired, and writer_handle.await?? was only reachable after a successful break, so the root-cause DB error was never printed: the program would re-list the mailbox and hammer the API indefinitely while printing only "DB writer task closed unexpectedly". Now the loop actually counts retries (bailing with the last error after 3), and when the writer task has died it awaits it, prints the underlying DB error, and spawns a fresh writer + channel so a transient SQLite error can succeed on the next retry.

Verified locally: cargo build, cargo test, and cargo clippy --all-targets -- -D warnings all pass.

[zbuc]

Reviewed the concurrency implementation against issue #5's acceptance criteria and found two problems, both fixed in 6b24519.

1. Retry after a transient error could double-count messages (violated the "results match a sequential run" criterion)

The seen-check (read pool) and the seen-mark (writer task commit) are separated by the mpsc queue. When work() failed mid-page (e.g. one messages_get hits a transient 500), main's retry loop restarted work() immediately without draining the writer's backlog. The retry's re-list then raced the writer: any message still queued-but-uncommitted read as unseen, was fetched and sent again, and — since seen_mails had no UNIQUE constraint — both writer transactions succeeded, permanently inflating that sender's count. The same window existed between consecutive pages if a message id appeared on two pages.

Fixed on two levels:

• Writer quiescing: the retry loop now closes the channel and awaits the writer task before every retry, so the full backlog is committed before any seen-check runs again. This also subsumes the old writer-died restart path.
• Idempotent replays: seen_mails.mail_id now has a UNIQUE index (created on startup, after collapsing any pre-existing duplicate rows), mark_seen uses INSERT OR IGNORE, and the sender counter is only incremented when the row was actually inserted. Even if a message slips through the read-side check twice, it is counted exactly once — this closes the cross-page window too.

As a side effect, the schema is now created automatically on startup (resolving the "add DB schema" TODO), so the manual sqlite3 setup step was removed from the README.

2. Concurrency limit was bounded but not configurable (first acceptance criterion asks for "a bounded, configurable concurrency limit")

FETCH_CONCURRENCY and RATE_LIMIT_INTERVAL were compile-time consts. They are now runtime-configurable via GMAIL_STATS_FETCH_CONCURRENCY (default 8, must be >= 1) and GMAIL_STATS_RATE_LIMIT_MS (default 25), documented in the README, so users with lower per-user Gmail quotas can tune the request rate without rebuilding.

Verified: cargo build, cargo test, and cargo clippy --all-targets -- -D warnings all pass, and cargo fmt --check is clean.

[zbuc]

Fixed two startup issues found in review (commit a703df2):

1. Fresh install was broken — database file was never created. The README change in this PR removed the manual sqlite3 stats.db setup step, claiming tables are created automatically on startup. But the connect options lacked .create_if_missing(true) (sqlx defaults to false), so on a clean checkout the pool connect failed with (code: 14) unable to open database file before init_schema ever ran. Fix: added .create_if_missing(true) to the SqliteConnectOptions. Verified on a clean checkout that stats.db is now created with both tables (seen_mails, senders) and the seen_mails_mail_id_unique index on first run.

2. GMAIL_STATS_RATE_LIMIT_MS=0 panicked instead of failing with a validation error. The value was passed unvalidated to tokio::time::interval, which asserts a non-zero period, so setting the var to 0 crashed with `period` must be non-zero — an unhandled panic rather than the clean anyhow error this PR intends for env handling. Fix: added anyhow::ensure!(rate_limit_ms >= 1, ...) mirroring the existing GMAIL_STATS_FETCH_CONCURRENCY check, and documented the minimum in the README. Verified it now exits with Error: GMAIL_STATS_RATE_LIMIT_MS must be at least 1.

cargo build, cargo test, and cargo clippy --all-targets -- -D warnings all pass.

[zbuc]

Automated fix pass for the reported merge conflict:

Findings

• The PR branch (issue-5, head a703df2) is already based on the current origin/main tip (f31f147) — the merge-base equals main, and GitHub reports the PR as MERGEABLE/CLEAN. The previously reported conflict has already been resolved by an earlier rebase, so no further rebase or conflict resolution was needed.

Verification (on a703df2)

• cargo build — passes
• cargo test — passes (0 tests defined)
• cargo clippy --all-targets -- -D warnings — passes with no warnings

No code changes were required, so nothing new was committed or pushed.

[zbuc]

Addressed a remaining gap against issue #5 acceptance criterion 4 (commit 8a7f88f):

Found: The rate-limiter tick and exponential-backoff retry were applied only to messages_get (fetch_message). The two messages_list calls in work() bypassed the rate limiter and used plain doit().await?, so a 429/rateLimitExceeded on any list call propagated out of work() into main's outer retry loop, restarting the entire mailbox scan from page one — the exact behavior criterion 4 forbids. That outer loop also retried immediately with no backoff, so under sustained quota exhaustion (precisely when list 429s occur) the program fired 4 back-to-back unthrottled restarts within seconds and exited with "too many retries". On a large mailbox, retries being cumulative meant the 4th sporadic 429 anywhere in a multi-hour run aborted the process.

Fixed:

• Added list_messages(), which waits for the shared rate-limiter tick before each attempt and retries rate-limit responses (via the existing is_rate_limit_error) with exponential backoff, mirroring fetch_message. work() now pages through it in a single loop instead of two raw doit() call sites.
• Added exponential backoff (2s/4s/8s) between outer-loop retries in main, so errors that outlast the per-call retries (including a single message rate-limited past MAX_FETCH_RETRIES) no longer cause immediate back-to-back full-mailbox restarts.

Data integrity was already safe across restarts (writer quiesce + UNIQUE index); this was purely a robustness gap. cargo build, cargo test, and cargo clippy --all-targets -- -D warnings all pass.

[zbuc]

Follow-up fix pushed (cbd0b2c): the final-failure exit path now quiesces the DB writer instead of returning immediately.

Problem found: in main()'s retry loop, the terminal bail (retries > 3) returned before the writer-quiesce block that runs on retry iterations 1-3. Two consequences on the 4th failure:

1. #[tokio::main] runtime shutdown cancelled the db_writer task mid-drain, silently discarding up to a channel's worth (100) of queued-but-uncommitted SeenMail results — already-fetched messages had to be re-fetched (paid Gmail quota) on the next run. No corruption thanks to the UNIQUE index, just wasted work.
2. If the 4th failure was the writer itself dying, work() only saw the opaque "DB writer task closed unexpectedly" send error; the underlying SQLite error inside the writer's JoinHandle was never awaited, so e.g. a disk-full error exited with no root cause printed.

Fix: the bail path now drops write_tx, awaits writer_handle so every queued result is drained and committed, and attaches the writer's own error (or panic) to the returned anyhow chain alongside "too many retries" — matching what the retry path already did.

Verified: cargo build, cargo test, and cargo clippy --all-targets -- -D warnings all pass.

[zbuc]

Fixed a remaining reliability gap in the retry handling (commit 02c834c).

Found: Per-call retry with backoff existed only for rate-limit responses — the retry arm in both list_messages and fetch_message was gated on is_rate_limit_error, which matched only 429/RESOURCE_EXHAUSTED/rateLimitExceeded JSON errors or non-JSON 429/403s. Every other transient error — Google's sporadic 503/UNAVAILABLE JSON responses (which arrive as Error::BadRequest), 500s, and Error::HttpError/Error::Io connection resets or TLS hiccups — fell into the Err(e) => return Err(e.into()) arm, propagated out of work(), and restarted the whole mailbox scan from page one. Worse, the outer loop's retries counter was never reset on forward progress, so the cap was 4 failures cumulative over the entire process lifetime: a large scan making 10^5+ requests would statistically hit its 4th sporadic 503 or connection reset and abort with "too many retries" instead of completing. This was also a behavioral regression versus main, where the commented-out retries += 1 meant this error class was retried indefinitely.

Fixed:

1. Broadened the per-call retry predicate (renamed is_rate_limit_error → is_retryable_error) to also treat as retryable: 5xx codes and UNAVAILABLE/INTERNAL statuses in JSON error envelopes, backendError reasons, non-JSON 5xx responses (Error::Failure with status.is_server_error()), and network-level transients (Error::HttpError, Error::Io). These now back off exponentially and retry in place, same as rate limits.
2. Reset the outer retries counter whenever a work() attempt makes forward progress (completes at least one page, tracked via a pages_done counter). The abort cap now applies only to consecutive no-progress failures, so sporadic transients spread across a long run can never accumulate into an abort, while a genuinely stuck run (e.g. revoked credentials) still exits after 4 fruitless attempts.

Verified: cargo build, cargo test, and cargo clippy --all-targets -- -D warnings all pass.