perf(imap): Improve performance of processing messages from IMAP ingestion - #425
Open
CafeLungo wants to merge 9 commits into
Open
perf(imap): Improve performance of processing messages from IMAP ingestion#425CafeLungo wants to merge 9 commits into
CafeLungo wants to merge 9 commits into
Conversation
Add a Compose-based devcontainer that layers a Node 22 + pnpm `dev` service on top of the existing backing services (postgres, valkey, meilisearch, tika) from the root docker-compose.yml, sharing the open-archiver-net network so the app runs from source and reaches the services by name. The prebuilt production `open-archiver` container is not started (runServices). post-create.sh pins pnpm 10.13.1, seeds .env if missing, installs deps, and builds @open-archiver/types. Also: - docker-compose.yml: give the open-archiver service both `image` and `build` (same tag) so it can be pulled or built from local source. - AGENTS.md: how to develop in the devcontainer. - .gitignore: repo-local .pnpm-store (pnpm's cross-device hardlink fallback under the bind mount).
The repo currently ships no test tooling. Add vitest (dev dependency +
`test`/`test:watch` scripts), a config that runs pure Node unit tests from
`src/**/*.{test,spec}.ts` without requiring the live Postgres/Valkey/Meili/
Tika services, and a setup file. `tsconfig.json` excludes spec/test files
from the production `tsc` build. Includes one example smoke test so `pnpm
--filter @open-archiver/backend test` is green out of the box.
IMAP ingestion was the bottleneck: after batch-fetching envelopes, the connector downloaded each non-duplicate message body with an individual fetchOne — one IMAP FETCH round-trip per message. Replace that with a single ranged FETCH over the non-duplicate UID set (pipelined by the server), streaming + yielding so heap stays bounded to one message. A mid-stream disconnect reconnects and re-fetches only the un-yielded UIDs, preserving the previous per-message retry robustness; a parse failure still aborts the mailbox. Also hoist findGroupSourceIds out of the per-email path: the process-mailbox loop resolves the merge-group IDs once and passes them into doesEmailExist and processEmail, avoiding a DB query + credential decrypt + children query on every message. Testing: add vitest (first tests in the repo) and cover the batching behavior via an injected fake ImapFlow client — asserting bodies are pulled in one ranged FETCH (not per-message), duplicates are skipped before their body is fetched, and sync state advances. A small client-factory seam on ImapConnector enables the injection; test files are excluded from the production tsc build.
After the batched IMAP fetch removed the per-message network round-trip, the remaining cost per email is sequential DB + object-storage latency: the consume loop awaited each processEmail before pulling the next, so those waits ran strictly one-at-a-time and never overlapped. Dispatch each yielded email to a pool bounded by a small counting semaphore (INGESTION_EMAIL_CONCURRENCY, default 8) so DB/storage latency overlaps across emails. The IMAP generator is still pulled sequentially (async generators are not safe to iterate concurrently); only the downstream processing is parallelized. Shared counters and the indexing batch are mutated only from settled callbacks — single-threaded, no locking — and the batch flush captures+resets synchronously so a concurrent completion can't push into a batch mid-flush. The loop drains all in-flight work (by acquiring every permit) before recording the mailbox result. Trade-off: two distinct new emails sharing a Message-ID within the same concurrency window could both pass the dedup gates (message_id_header has an index, not a unique constraint). Rare — Pass 1 already filters already-archived dupes — and INGESTION_EMAIL_CONCURRENCY=1 restores the old strictly-serial behavior. Adds a Semaphore helper (zero runtime deps) with unit tests.
Collect every ingestion/indexing concurrency and batch-size knob into a single config/ingestion.ts, all overridable via environment variables with defaults identical to the previous behavior: - INGESTION_EMAIL_CONCURRENCY (8) per-mailbox processing concurrency - INGESTION_WORKER_CONCURRENCY (5) moved from inline env read - IMAP_FETCH_BATCH_SIZE (250) was hard-coded in ImapConnector - INDEXING_WORKER_CONCURRENCY (1) was BullMQ's implicit default - REINDEX_RESET_BATCH_SIZE (5000) was hard-coded in resetIndexedFlag Wire each call site to read from config.ingestion, and document all five in .env.example. intFromEnv() falls back to the default on a non-numeric or <= 0 value so a bad override can't zero a pool or batch.
The per-batch document build fan-out in IndexingService.indexEmailBatch (read .eml from storage + parse, before the batch is written to Meilisearch in a single addDocuments request) was hard-coded at 10. Expose it as INDEXING_DOCUMENT_BUILD_CONCURRENCY (default 10) via config/ingestion.ts, completing the set of tunable ingestion/indexing knobs. Default unchanged.
The Pass 1 pre-check ran one dedup query per envelope (doesEmailExist -> findFirst), so re-scanning an already-archived mailbox issued one DB round-trip per message — the dominant cost of the long silent "catch-up" before new mail is reached. Add IngestionService.filterExistingMessageIds(), a set-based form of doesEmailExist that returns which of many message-ids already exist in a single query. Thread an optional checkDuplicatesBatch callback through IEmailConnector.fetchEmails and have ImapConnector's Pass 1 collect the batch's envelopes first, then dedup them all in one call — turning N per-message queries into one per ~250-UID batch. Falls back to the per-message checkDuplicate when no batch checker is supplied, so the other connectors are unchanged. Adds a batch-path unit test; the existing per-message tests still cover the fallback.
…ptions sync_state (maxUid) was only persisted at mailbox completion, so a large mailbox interrupted before finishing (rate-limit throttle, connection hang, restart) discarded all progress and re-scanned from UID 1 next cycle — which for a huge mailbox under a daily bandwidth cap never completes, so it never persists anything. Persist maxUid incrementally: - SyncSessionService.checkpointSyncState(): the same jsonb || merge as recordMailboxResult, without touching session counters. - process-mailbox.processor: on the existing ~5-min heartbeat tick, drain the in-flight processEmail pool FIRST (so the checkpointed maxUid only covers durably-archived messages), then persist. - ImapConnector: advance the checkpointable maxUid at BATCH end (after the batch's bodies are yielded) instead of per-envelope during the Pass 1 scan — otherwise a checkpoint could report a UID whose body hasn't been fetched yet and the next resume would skip it. Final maxUid is unchanged, so existing tests still hold. Now an interrupted sync resumes from the last checkpoint, making steady forward progress across daily throttle windows instead of looping.
With incremental checkpointing, a normal sync resumes from the stored maxUid. Make the user-initiated "Force sync" a deliberate full re-import by clearing sync_state (for the source and any cascaded children) before dispatching the continuous-sync job, so it re-scans from the beginning rather than resuming.
Author
|
For what it is worth, I have been running this locally against a personal gmail imap. I keep hitting the daily 2.5GB cap in traffic, but so far this has been working out pretty well. Initially I got ~24k messages/day ingested. Now I get more like 140k/day before hitting the daily cap. Then it enters a retry loop where the IMAP server just stops responding for the rest of the 24hr period. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Depends on #423 (for added tests, per CLA)
Motivation
IMAP ingestion was a throughput bottleneck. For each mailbox the connector fetched envelopes in a batch but then downloaded each message body with an individual
fetchOne(one FETCH round-trip per message), ran one dedup DB query per envelope, and processed emails strictly one-at-a-time (eachprocessEmailawaited before the next). On top of that, sync progress (maxUid) was only persisted at mailbox completion, so a large mailbox interrupted mid-scan (throttle, hang, restart) threw away all progress and re-scanned from UID 1 — under a daily bandwidth cap, a huge mailbox could loop forever without ever checkpointing.What's in here
Performance
fetchOnewith a single ranged FETCH over the non-duplicate UID set (server-pipelined), streamed + yielded so heap stays bounded to one message. Mid-stream disconnect reconnects and re-fetches only the un-yielded UIDs (retains the old per-message retry robustness).IngestionService.filterExistingMessageIds()turns N per-envelope existence queries into one query per ~250-UID batch, threaded throughfetchEmailsvia an optionalcheckDuplicatesBatchcallback (falls back to per-message for other connectors).INGESTION_EMAIL_CONCURRENCY, default 8) so DB + object-storage latency overlaps instead of running serially. The IMAP generator is still pulled sequentially; only downstream processing is parallelized. Zero-depSemaphorehelper added.findGroupSourceIdsis resolved once per mailbox instead of per message (saves a DB query + credential decrypt + children query on every email).Resumable sync
maxUidon the existing ~5-min heartbeat (draining the in-flight pool first so the checkpoint only covers durably-archived messages, and advancing at batch end so we never checkpoint a UID whose body isn't fetched yet). Interrupted syncs now make steady forward progress across throttle windows.sync_state(source + cascaded children) so it deliberately re-imports from the beginning rather than resuming.Config — every ingestion/indexing concurrency & batch knob is centralized in
config/ingestion.tsand overridable via env, with defaults identical to prior behavior:INGESTION_EMAIL_CONCURRENCYINGESTION_WORKER_CONCURRENCYIMAP_FETCH_BATCH_SIZEINDEXING_WORKER_CONCURRENCYINDEXING_DOCUMENT_BUILD_CONCURRENCYREINDEX_RESET_BATCH_SIZEintFromEnv()falls back to the default on non-numeric/≤0 values so a bad override can't zero a pool.Behavior change / trade-off
With concurrency > 1, two distinct new emails sharing a
Message-IDwithin the same window could both pass the dedup gates (message_id_headeris indexed, not unique). Rare (Pass 1 already filters already-archived dupes), andINGESTION_EMAIL_CONCURRENCY=1restores strictly-serial behavior. Defaults preserve existing behavior everywhere else.Testing
Adds unit tests for the batching path (injected fake
ImapFlow: bodies pulled in one ranged FETCH, dupes skipped before body fetch, sync-state advances), the batch dedup path, and theSemaphorehelper.