feat: integrate mailing list service (CM-1318)#4346
Conversation
Signed-off-by: Uroš Marolt <uros@marolt.me>
Signed-off-by: Uroš Marolt <uros@marolt.me>
Signed-off-by: Uroš Marolt <uros@marolt.me>
Signed-off-by: Uroš Marolt <uros@marolt.me>
Signed-off-by: Uroš Marolt <uros@marolt.me>
Signed-off-by: Uroš Marolt <uros@marolt.me>
Signed-off-by: Uroš Marolt <uros@marolt.me>
Signed-off-by: Uroš Marolt <uros@marolt.me>
Signed-off-by: Uroš Marolt <uros@marolt.me>
Signed-off-by: Uroš Marolt <uros@marolt.me>
Signed-off-by: Uroš Marolt <uros@marolt.me>
PR SummaryMedium Risk Overview Backend & data: New Ops: Docker image, compose service, build env, CLI ignore list, and a CI ruff job for the new app. ADR-0009 documents skipping messages with implausible Ingestion hardening: Reviewed by Cursor Bugbot for commit cf4735c. Bugbot is set up for automated code reviews on this repo. Configure here. |
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
| git_blob_id, | ||
| git_dir, | ||
| ) | ||
| sys.exit(1) |
There was a problem hiding this comment.
Git errors terminate worker
High Severity
When a commit or blob is missing in a shard, get_email_from_git calls sys.exit(1). The list worker invokes that via read_email, and except Exception does not catch SystemExit, so one bad commit can exit the whole FastAPI/uvicorn process instead of marking the list failed and releasing the lock.
Reviewed by Cursor Bugbot for commit 212875a. Configure here.
| await queue_service.send_batch_activities(activities_kafka) | ||
|
|
||
| await update_processed_heads(mailing_list.id, heads) | ||
| state = ListState.COMPLETED |
There was a problem hiding this comment.
Kafka failure orphans DB rows
High Severity
Activities are inserted into integration.results before Kafka emit succeeds. If send_batch_activities fails afterward, the list is marked failed and shard heads are not advanced, but pending rows remain and the data-sink worker never receives process_integration_result messages for them.
Reviewed by Cursor Bugbot for commit 212875a. Configure here.
|
|
||
| if activities_db: | ||
| await batch_insert_activities(activities_db) | ||
| await queue_service.send_batch_activities(activities_kafka) |
There was a problem hiding this comment.
Onboarding buffers entire shard
High Severity
First-time processing walks every commit in each shard and appends all activities into in-memory lists before a single DB insert and Kafka batch. Large lore lists can exhaust memory or run far beyond practical limits, unlike chunked processing in git_integration.
Reviewed by Cursor Bugbot for commit 212875a. Configure here.
| logger.info(f"Saving {len(records)} activities into integration.results") | ||
| for i in range(0, len(records), batch_size): | ||
| batch = records[i : i + batch_size] | ||
| await executemany(sql_query, batch) |
There was a problem hiding this comment.
Activity batches not transactional
Medium Severity
batch_insert_activities issues multiple executemany calls without a shared transaction. If a later batch fails, earlier batches may already be committed, leaving partial data and no matching Kafka emit for the full run.
Reviewed by Cursor Bugbot for commit 212875a. Configure here.
There was a problem hiding this comment.
Pull request overview
Adds a containerized Python worker for ingesting public-inbox mailing lists into CDP through Kafka and integration.results.
Changes:
- Adds mirroring, email parsing, processing, and Kafka publishing.
- Adds mailing-list database schema and development seed.
- Adds container, Compose, CI, and local tooling.
Review note: This exceeds the recommended 1,000-line PR target.
Reviewed changes
Copilot reviewed 26 out of 35 changed files in this pull request and generated 15 comments.
Show a summary per file
| File | Description |
|---|---|
services/apps/mailing_list_integration/src/test/test_noteren.py |
Adds parser and CLI tests. |
services/apps/mailing_list_integration/src/runner.sh |
Starts Uvicorn. |
services/apps/mailing_list_integration/src/crowdmail/worker/list_worker.py |
Implements ingestion loop. |
services/apps/mailing_list_integration/src/crowdmail/worker/__init__.py |
Initializes worker package. |
services/apps/mailing_list_integration/src/crowdmail/settings.py |
Defines environment configuration. |
services/apps/mailing_list_integration/src/crowdmail/services/queue/queue_service.py |
Adds Kafka producer. |
services/apps/mailing_list_integration/src/crowdmail/services/queue/__init__.py |
Initializes queue package. |
services/apps/mailing_list_integration/src/crowdmail/services/parse/noteren.py |
Parses public-inbox emails. |
services/apps/mailing_list_integration/src/crowdmail/services/parse/__init__.py |
Initializes parser package. |
services/apps/mailing_list_integration/src/crowdmail/services/mirror/mirror_service.py |
Manages public-inbox mirrors. |
services/apps/mailing_list_integration/src/crowdmail/services/mirror/__init__.py |
Initializes mirror package. |
services/apps/mailing_list_integration/src/crowdmail/services/__init__.py |
Initializes services package. |
services/apps/mailing_list_integration/src/crowdmail/server.py |
Adds FastAPI lifecycle and health route. |
services/apps/mailing_list_integration/src/crowdmail/models/list.py |
Defines mailing-list model. |
services/apps/mailing_list_integration/src/crowdmail/models/__init__.py |
Initializes models package. |
services/apps/mailing_list_integration/src/crowdmail/logger.py |
Configures Loguru. |
services/apps/mailing_list_integration/src/crowdmail/errors.py |
Defines service errors. |
services/apps/mailing_list_integration/src/crowdmail/enums.py |
Defines service enums. |
services/apps/mailing_list_integration/src/crowdmail/database/registry.py |
Adds database helpers. |
services/apps/mailing_list_integration/src/crowdmail/database/crud.py |
Adds processing-state queries. |
services/apps/mailing_list_integration/src/crowdmail/database/connection.py |
Manages asyncpg pooling. |
services/apps/mailing_list_integration/src/crowdmail/database/__init__.py |
Initializes database package. |
services/apps/mailing_list_integration/src/crowdmail/__init__.py |
Initializes application package. |
services/apps/mailing_list_integration/README.md |
Documents operation and setup. |
services/apps/mailing_list_integration/pyproject.toml |
Defines package and tooling. |
services/apps/mailing_list_integration/Makefile |
Adds development commands. |
services/apps/mailing_list_integration/dev/seed.sql |
Seeds an example integration. |
scripts/services/mailing-list-integration.yaml |
Adds Compose services. |
scripts/services/docker/Dockerfile.mailing_list_integration |
Builds the worker image. |
scripts/cli |
Registers the service in clean-start handling. |
scripts/builders/mailing-list-integration.env |
Configures image publishing. |
backend/src/database/migrations/V1784048135__mailinglist-schema.sql |
Creates mailing-list tables. |
.github/workflows/backend-lint.yaml |
Adds Python linting CI. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| RUN apt-get update && apt-get install -y \ | ||
| ca-certificates \ | ||
| git \ | ||
| public-inbox \ |
| class ListWorker: | ||
| """Worker that mirrors, parses and emits activities for mailing lists""" |
| shard = shard_index(shard_path) | ||
| commit_ids = await new_commits(shard_path, heads.get(shard)) | ||
| for git_id in commit_ids: | ||
| message, blob_id = read_email(shard_path, git_id) |
| activities_db = [] | ||
| activities_kafka = [] |
| version = "0.0.1" | ||
| description = "Crowd.dev mailing list integration for the Linux Foundation" | ||
| readme = "README.md" | ||
| license = { file = "LICENSE" } |
| futures = [ | ||
| _producer.send( | ||
| topic=CROWD_KAFKA_TOPIC, | ||
| key=activity["message_id"].encode("utf-8", errors="replace"), | ||
| value=activity["payload"].encode("utf-8", errors="replace"), | ||
| ) | ||
| for activity in activities_kafka | ||
| ] | ||
| await asyncio.gather(*futures, return_exceptions=False) |
| else: | ||
| logging.error( | ||
| 'Unable to retrieve git blob %s from the git repo "%s"', | ||
| git_blob_id, | ||
| git_dir, | ||
| ) | ||
| sys.exit(1) |
| - name: Install dependencies | ||
| if: steps.changes.outputs.python_changed == 'true' | ||
| run: uv sync --group dev --frozen |
| ) | ||
| # add a fake set of microseconds just to make date parsers on the injest side happier | ||
| date = date_dt.strftime("%Y-%m-%dT%H:%M:%S") + ".000000Z" | ||
| except (ValueError, OverflowError): |
Signed-off-by: Uroš Marolt <uros@marolt.me>
…ion (CM-1318) Signed-off-by: Uroš Marolt <uros@marolt.me>
Signed-off-by: Uroš Marolt <uros@marolt.me>
…318) Signed-off-by: Uroš Marolt <uros@marolt.me>
| logger.info("Worker shutdown complete") | ||
| except TimeoutError: | ||
| logger.warning("Worker shutdown timeout, forcing cancellation") | ||
| worker_task.cancel() |
There was a problem hiding this comment.
Shutdown waits on null task
Low Severity
The lifespan finally block always calls asyncio.wait_for(worker_task, …) and may call worker_task.cancel(), but worker_task stays None if asyncio.create_task fails during startup. That raises TypeError during shutdown and can mask the original startup error.
Reviewed by Cursor Bugbot for commit 037a969. Configure here.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 40 out of 51 changed files in this pull request and generated 16 comments.
Comments suppressed due to low confidence (3)
services/apps/mailing_list_integration/src/crowdmail/worker/list_worker.py:90
- Initial onboarding materializes every commit ID and retains every parsed DB/Kafka record until the entire archive finishes. For LKML-scale archives this is unbounded memory usage and no progress is checkpointed before a likely OOM/restart. Process bounded commit batches and persist each batch/head checkpoint before continuing, similar to the 250-commit chunking in
git_integration/src/crowdgit/services/commit/commit_service.py:709-743.
.github/workflows/backend-lint.yaml:147 - This CI job installs pytest and the PR adds a substantial test suite, but the job runs only Ruff. As a result, parser regressions can merge even though the PR reports the tests as coverage. Run pytest in this job after lint/format checks.
- name: Check Python linting and formatting
if: steps.changes.outputs.python_changed == 'true'
run: |
uv run ruff check src/ --output-format=github
uv run ruff format --check src/
services/apps/mailing_list_integration/src/crowdmail/worker/list_worker.py:31
- This introduces a new class-based worker, contrary to the repository's functions-over-classes direction for new service code (
CLAUDE.md:42-43andservices-checklist.md:41-43). Keep shutdown state in a small functional runner/context and expose plain processing functions so the polling and per-list logic remain independently testable.
| state = ListState.FAILED | ||
|
|
||
| try: | ||
| list_dir = await ensure_mirror(mailing_list.name) |
| lists: z | ||
| .array( | ||
| z.object({ | ||
| name: z.string().trim().min(1), |
| for git_id in commit_ids: | ||
| message, blob_id = read_email(shard_path, git_id) | ||
| parsed = parse_email( |
| def read_email(shard_path: str, git_id: str) -> tuple[bytes, str]: | ||
| """Raw email bytes + blob id for a commit in a shard.""" | ||
| return get_email_from_git(shard_path, git_id) |
| # Retried on any failure: if this write never lands, the list stays stuck | ||
| # in its acquired state (e.g. PROCESSING) forever, since acquire_onboarding_list | ||
| # only picks PENDING and acquire_recurrent_list excludes PROCESSING. |
| psql "$CROWD_DB_WRITE_URI" -f dev/seed.sql | ||
| ``` | ||
|
|
||
| It creates a `public.integrations` row (`platform='groupsio'`), a |
| - emits one Kafka message per result (`process_integration_result`, | ||
| `platform=groupsio`), |
| 4. Confirm `data_sink_worker` consumes the message, resolves the integration's | ||
| `platform='groupsio'`, and creates the member (email-based verified |
| - Onboarding UI/backend to create `integrations` + `mailinglist.lists` rows | ||
| (this is what `dev/seed.sql` stands in for). |
| # of emitting a malformed timestamp. | ||
| if not 1970 <= date_dt.year <= 9999: | ||
| raise ValueError(f"implausible year {date_dt.year}") | ||
| # add a fake set of microseconds just to make date parsers on the injest side happier |
![github-license-compliance[bot]](https://avatars.githubusercontent.com/u/9919?s=60&v=4){kind=small}
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 2 potential issues.
There are 7 total unresolved issues (including 5 from previous reviews).
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit cf4735c. Configure here.
| await _run_shell_command(["public-inbox-fetch", "-q"], cwd=list_dir) | ||
| else: | ||
| logger.info(f"Fetching updates for lore list '{list_name}'") | ||
| await _run_shell_command(["public-inbox-fetch", "-q", "-C", list_dir]) |
There was a problem hiding this comment.
Mirror ignores stored source URL
Medium Severity
Onboarding persists a per-list sourceUrl, but mirroring always clones or fetches https://lore.kernel.org/{name}/ using only the list name. A list whose sourceUrl is not that lore path is mirrored from the wrong host/path while activities still record the stored sourceUrl, yielding incorrect or empty ingestion.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit cf4735c. Configure here.
| - name: Check Python linting and formatting | ||
| if: steps.changes.outputs.python_changed == 'true' | ||
| run: | | ||
| uv run ruff check src/ --output-format=github |
There was a problem hiding this comment.
Git integration CI drops format check
Low Severity
The lint-python-git-integration job’s lint step now runs only ruff check. The previous ruff format --check for that job was removed when the mailing-list lint job was added, so unformatted Python under git_integration can pass CI.
Reviewed by Cursor Bugbot for commit cf4735c. Configure here.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 40 out of 51 changed files in this pull request and generated 5 comments.
Comments suppressed due to low confidence (10)
services/apps/mailing_list_integration/src/crowdmail/worker/list_worker.py:84
- This ignores the stored
source_url;ensure_mirroralways cloneshttps://lore.kernel.org/{name}/. A valid public-inbox URL with a different host or path will silently ingest the wrong archive. Pass the source URL to the mirror layer and keep the local directory key separate.
services/apps/mailing_list_integration/src/crowdmail/worker/list_worker.py:91 - On initial onboarding,
new_commits(..., None)materializes the shard’s entire history, and this method then retains every DB and Kafka record until all shards finish. Public-inbox shards can contain hundreds of thousands of messages, so the worker can exhaust memory before reaching the existing insert chunks. Iterate in bounded batches and checkpoint each batch only after persistence and emission succeed.
services/apps/mailing_list_integration/src/crowdmail/worker/list_worker.py:93 read_emaillaunches synchronous Git subprocesses on the FastAPI event loop—twice per message. A large import will block health requests, shutdown handling, and other async work for long periods. Offload this blocking call to a thread.
services/apps/mailing_list_integration/src/crowdmail/worker/list_worker.py:94- Any parser exception escapes this per-commit loop, marks the whole list failed, and leaves the shard head unchanged. The next retry starts from the same malformed message, so one poison email can permanently block every later message in the shard. Catch failures per commit and skip/quarantine them with observable logging or metrics, as the Git integration does in
commit_service.py:652-655.
services/apps/mailing_list_integration/src/crowdmail/services/parse/noteren.py:72 - This helper is called by the long-running worker, so
sys.exit()raisesSystemExit, bypasses itsexcept Exceptionhandlers, and can terminate the service because of one malformed commit. Raise a domain exception and let the worker apply its per-message failure policy instead.
services/apps/mailing_list_integration/src/crowdmail/services/parse/noteren.py:87 - The blob-read failure also calls
sys.exit(). In worker mode this can terminate the entire service instead of failing one message/list. Raise a normal domain exception here, consistent with the commit-not-found branch.
services/apps/mailing_list_integration/src/crowdmail/services/parse/noteren.py:354 - Messages without
Message-IDare emitted with an emptysourceId. Activity deduplication keys includesourceId, so multiple such messages in the same segment/platform/type/channel can collapse into one activity. Use a stable fallback such as the Git commit/blob ID, or explicitly skip these messages.
services/apps/mailing_list_integration/src/crowdmail/services/mirror/mirror_service.py:58 - The five-minute default applies to initial
public-inbox-cloneand fetch operations. LKML-scale archives can legitimately take much longer, so onboarding will repeatedly kill the clone and never complete. Use a separately configurable mirror timeout sized for large archives while retaining shorter limits for lightweight commands.
services/apps/mailing_list_integration/src/crowdmail/database/crud.py:117 - A process crash after acquisition leaves
state='processing'andlockedAtset. This exclusion, combined with both selectors requiringlockedAt IS NULL, means the list is never acquired again; stale onboarding rows also consume the concurrency count forever. Treat locks as leases and reclaim/reset processing rows older than a configured threshold.
states_to_exclude = (ListState.PENDING, ListState.PROCESSING, ListState.PENDING_REONBOARD)
services/apps/mailing_list_integration/README.md:16
- This documentation still says the new worker emits
platform=groupsio, while the implementation, seed, activity type, and identities all useplatform=mailinglist. The same stale Groups.io references recur at lines 46, 62, and 66; update them so operators verify the actual platform.
parses new messages, writes activities to `integration.results`, and emits
Kafka messages to the `data-sink-worker` topic — same plumbing as
git_integration, with `platform=groupsio`.
| // find members to create | ||
| for (const payload of payloadsWithoutDbMembers) { | ||
| const key = `${payload.platform}:${payload.activity.username}` | ||
| const key = `${payload.platform}:${payload.activity.username?.toLowerCase()}` |
| // find object members to create | ||
| for (const payload of payloadsWithoutDbObjectMembers) { | ||
| const key = `${payload.platform}:${payload.activity.objectMemberUsername}` | ||
| const key = `${payload.platform}:${payload.activity.objectMemberUsername?.toLowerCase()}` |
| "channel": channel, | ||
| "title": subject, | ||
| "body": json_body, | ||
| "url": "https://lore.kernel.org/r/" + msgid, |
| def list_mirror_dir(list_name: str, mirror_dir: str = LORE_MIRROR_DIR) -> str: | ||
| """Local directory where a lore list mirror lives.""" | ||
| return os.path.join(mirror_dir, list_name) |
| INSERT INTO "activityTypes" ("activityType", platform, "isCodeContribution", "isCollaboration", description) VALUES | ||
| ('message', 'mailinglist', false, true, 'Sent a message to a mailing list'); |
Summary
Integrate mailing list (public-inbox) support into CDP as a containerized worker. This implements email ingestion from public-inbox repositories (like LKML), mirroring shards, parsing emails, and emitting activities to Kafka for the data sink.
mailing_list_integrationservice (Python, FastAPI)How it works
mailinglist.listProcessingwithFOR UPDATE SKIP LOCKEDpublic-inbox-clone(initial) orpublic-inbox-fetch(incremental)0.git,1.git, etc.) and walks new commits since last processedintegration.results(state=pending)data-sink-workerfor activity processinglastProcessedHeads(per-shard tracking) and marks list as COMPLETEDTest coverage
Out of scope
This PR targets hardcoded integrations for initial deployment; UI follows in a separate ticket.