feat(xmtp_mls): event-driven pending-self-remove via TaskRunner

[insipx]

Folds the poll-based PendingSelfRemoveWorker into the durable TaskRunner. PR 3 of 3 — stacked on #3745 (proto regen), which is stacked on #3743 (disappearing).

Problem

PendingSelfRemoveWorker polled get_groups_have_pending_leave_request() every 2s (~43,200 empty queries/day per device). The work — building an MLS RemoveProposal/Commit to evict a member who sent a LeaveRequest — is discrete, immediate, and must retry (only a super-admin can commit it). That is exactly what the existing durable TaskRunner provides.

Changes

• Enqueue: on LeaveRequest, a ProcessPendingSelfRemove task is created in the same DB transaction as the pending_remove insert (atomic, restart-safe), deduped per group via pending_self_remove_task_exists() — a pre-insert check (no tasks migration) that excludes dead rows (expired / max-attempts-exhausted) so a fresh backstop isn't suppressed by a not-yet-GC'd task.
• Fast-path kept: the inline process_admin_pending_remove_actions call stays; the task is the retry/backstop. Removal logic lives in one shared helper MlsGroup::process_pending_self_removals() (remove + cleanup) called from both.
• Dispatch: TaskRunner::run_task gains a ProcessPendingSelfRemove arm that loads the group and runs the shared helper; idempotent, no-op when not super-admin, so retries are safe.
• Removed: PendingSelfRemoveWorker, its WorkerKind, builder registration; the disconnect-propagation test is repurposed to TaskWorkerError::LoadGroup. The wasm/mobile/node bindings that mirrored WorkerKind are updated.

The ProcessPendingSelfRemove proto variant this consumes is regenerated in #3745 (split out per review).

Testing

pending_self_remove_task_exists + task_worker_load_group_forwards_disconnect pass; native + wasm + mobile + node compile; just lint-rust green. (Integration tests requiring the XMTP Docker node not run here.)

🤖 Generated with Claude Code

Note

Replace PendingSelfRemoveWorker with event-driven TaskRunner-based self-removal

• Removes the dedicated PendingSelfRemoveWorker polling loop; self-removal is now triggered by process_leave_request_message, which durably enqueues a ProcessPendingSelfRemove task and wakes the TaskRunner post-commit.
• Adds upsert_pending_self_remove_task to QueryTasks in tasks.rs to deduplicate live tasks per group while replacing dead/exhausted ones.
• On client startup, backfill_pending_self_remove_tasks in builder.rs seeds tasks for any pre-existing pending leave requests so in-flight removals survive restarts.
• Replaces the DeferredEvents disappearing-rearm boolean with a HashSet<WorkerKind> so multiple workers can be woken idempotently after a commit.
• Removes PendingSelfRemove from all public FFI/WASM/Node WorkerKind enums.
• Risk: tasks silently skip rescheduling when the row is deleted concurrently (NotFound), and startup backfill errors are swallowed with a warning rather than failing client setup.

^{Macroscope summarized 1419643.}

[claude]

Claude finished @insipx's task —— View job

---

PR Review Complete

Review Date: June 9, 2026, 21:35 UTC

Todo List

• Read and understand the implementation changes
• Review existing comments to avoid duplication
• Analyze code quality and potential issues
• Provide focused feedback

---

Review Feedback

The PR successfully consolidates self-removal into the event-driven TaskRunner, and the author has addressed most issues raised during review. Here are the remaining items not covered in existing comments:

1. Performance: Full Table Scan in upsert_pending_self_remove_task (High Priority)

Location: crates/xmtp_db/src/encrypted_store/tasks.rs:196-214

The upsert_pending_self_remove_task implementation performs a full table scan of all tasks on every LeaveRequest:

let rows: Vec<(i32, i32, i32, i64, Vec<u8>)> = tasks::table
    .select((tasks::id, tasks::attempts, tasks::max_attempts, tasks::expires_at_ns, tasks::data))
    .load(conn)?;
for (id, attempts, max_attempts, expires_at_ns, data) in rows {
    let is_self_remove = matches!(
        TaskProto::decode(data.as_slice()).ok().and_then(|t| t.task),
        Some(TaskKind::ProcessPendingSelfRemove(p)) if p.group_id == group_id.as_slice()
    );
    // ...
}

Problem: This loads and decodes every task row (welcome processing, disappearing messages, sync archives, etc.) just to find one group's self-remove task. With a busy tasks table, this creates O(n) overhead per LeaveRequest.

Recommendation: Add a discriminator column or use the data_hash index more efficiently. For example:

• Pre-filter by a task-type column if one exists, or
• Use a targeted query with the known data_hash for this group's task (since the proto is deterministic)
• Or, store group_id as a searchable column for ProcessPendingSelfRemove tasks

This is a critical path — every LeaveRequest message (including external ones) hits this query inside the message-processing transaction.

---

2. Test Coverage Gap: Concurrent Upsert During Task Execution

Location: Test suite

The review comments identified a subtle race between upsert_pending_self_remove_task (which deletes dead rows) and the TaskRunner's update_task. The fix added a NotFound guard in run_and_reschedule_task:

Err(StorageError::DieselResult(diesel::result::Error::NotFound)) => {
    tracing::debug!("Task {} vanished before reschedule; skipping", task.id);
}

Missing: No test exercises this race. Consider adding a test that:

1. Starts a task execution (mock or real)
2. Concurrently calls upsert_pending_self_remove_task while the task is mid-flight
3. Verifies the worker doesn't crash when update_task returns NotFound

This would ensure the defense-in-depth fix actually works as intended.

---

3. Minor: Own-Leave Path Never Wakes TaskRunner (Low Priority)

Location: crates/xmtp_mls/src/groups/mls_sync.rs:1078 and 1603

When a user sends their own LeaveRequest (via leave_group()), the code path is:

1. process_own_message → process_own_leave_request_message (line 1078)
2. Which calls process_leave_request_message(..., deferred_events = None) (line 1603)
3. The wake at line 1696-1698 is skipped because deferred_events.is_none()

Observation: The comment at line 1694 explains this is intentional ("The own-leave path has no DeferredEvents (its task is a no-op)"), but there's a subtle issue: the task isn't a no-op if the user is later promoted to super-admin before the task expires. The task would need to run eventually to clean up.

Current behavior: The task sits until the TaskRunner's next natural wakeup (could be up to 24h if the table was empty).

Recommendation: Either:

• Accept this as a known limitation and document it (the promotion-then-cleanup flow is covered by the fast-path anyway), or
• Pass Some(&mut deferred_events) here too (with a dummy DeferredEvents if needed) so the TaskRunner wakes even for own-leaves

This is low-severity because the fast-path handles the common case, and the task is still durable.

---

Summary

The implementation is sound, and the extensive review process has caught and fixed most issues. The main outstanding concern is the full table scan performance issue, which should be addressed before merge given the frequency of LeaveRequest processing. The test coverage gap is lower priority but would strengthen confidence in the fix. The own-leave wake issue is a minor edge case that likely doesn't need changes.

Overall, this is a well-architected migration from polling to event-driven processing. Nice work consolidating the workers!

---

[macroscopeapp]

🟢 Low groups/mod.rs:1720

process_pending_self_removals() documents itself as a no-op for non-super-admin clients, but line 1722 unconditionally calls cleanup_pending_removal_list() even when remove_members_pending_removal() has already early-returned for a non-super-admin. cleanup_pending_removal_list() then fetches members().await, which triggers network work and can return GroupError failures—so follower devices still do expensive work and potentially fail/exit with error instead of harmlessly skipping the entire operation.

    pub(crate) async fn process_pending_self_removals(&self) -> Result<(), GroupError> {
-        self.remove_members_pending_removal().await?;
-        self.cleanup_pending_removal_list().await?;
-        Ok(())
+        // Only super-admins have work to do; this is a documented no-op otherwise
+        if self.is_super_admin(self.context.inbox_id().to_string())? {
+            self.remove_members_pending_removal().await?;
+            self.cleanup_pending_removal_list().await?;
+        }
+        Ok(())
     }

Also found in 1 other location(s)

crates/xmtp_mls/src/worker/tasks.rs:280

ProcessPendingSelfRemove retries permanently missing groups instead of dropping the task. MlsStore::group() returns MlsStoreError::NotFound when the referenced group_id is absent, but run_and_reschedule_task() reschedules every Err. If the group was deleted before this durable task runs, line 280 keeps the stale task retrying until max_attempts/expiry, generating repeated warnings and delaying cleanup instead of treating the task as completed.

🚀 Reply "fix it for me" or copy this AI Prompt for your agent:

In file @crates/xmtp_mls/src/groups/mod.rs around lines 1720-1724:

`process_pending_self_removals()` documents itself as a no-op for non-super-admin clients, but line 1722 unconditionally calls `cleanup_pending_removal_list()` even when `remove_members_pending_removal()` has already early-returned for a non-super-admin. `cleanup_pending_removal_list()` then fetches `members().await`, which triggers network work and can return `GroupError` failures—so follower devices still do expensive work and potentially fail/exit with error instead of harmlessly skipping the entire operation.

Also found in 1 other location(s):
- crates/xmtp_mls/src/worker/tasks.rs:280 -- `ProcessPendingSelfRemove` retries permanently missing groups instead of dropping the task. `MlsStore::group()` returns `MlsStoreError::NotFound` when the referenced `group_id` is absent, but `run_and_reschedule_task()` reschedules every `Err`. If the group was deleted before this durable task runs, line `280` keeps the stale task retrying until `max_attempts`/expiry, generating repeated warnings and delaying cleanup instead of treating the task as completed.

[insipx]

cleanup_pending_removal_list() is intentionally admin-agnostic and idempotent: it removes pending-remove rows for inbox-ids no longer in the group (true for any observer, not just admins) and clears has_pending_leave_request once the list drains. Running it on a non-super-admin client is harmless and correct — that client still observes departures and keeps its local pending list consistent. The 'no-op for non-admin' doc refers to remove_members_pending_removal (which can't commit a removal without admin rights), not to cleanup. So calling both from the shared helper is intended; no change here.

[macroscopeapp]

Approvability

Verdict: Needs human review

1 blocking correctness issue found. This PR replaces a 2-second polling worker with event-driven task processing for self-removals, a significant architectural change. An unresolved high-severity comment identifies potential day-long delays when the task table is empty, which requires human evaluation before merging.

^{You can customize Macroscope's approvability policy. Learn more.}

[macroscopeapp]

🟠 High

libxmtp/crates/xmtp_mls/src/builder.rs

Line 372 in b118d32

if enabled(WorkerKind::TaskRunner) {

Removing PendingSelfRemoveWorker causes pending self-removals to be processed by TaskWorker, but process_leave_request_message now inserts tasks via storage.db().create_task(...) instead of context.task_channels().send(...). TaskWorker::run() only re-queries the database after receiving on task_receiver or when its current sleep(next_wakeup) finishes. When the task table is empty, this sleep lasts up to one day, so a newly inserted ProcessPendingSelfRemove task can be delayed until another unrelated task or event arrives — leaving self-removals stuck far longer than the previous 2s polling interval.

🚀 Reply "fix it for me" or copy this AI Prompt for your agent:

In file @crates/xmtp_mls/src/builder.rs around line 372:

Removing `PendingSelfRemoveWorker` causes pending self-removals to be processed by `TaskWorker`, but `process_leave_request_message` now inserts tasks via `storage.db().create_task(...)` instead of `context.task_channels().send(...)`. `TaskWorker::run()` only re-queries the database after receiving on `task_receiver` or when its current `sleep(next_wakeup)` finishes. When the task table is empty, this sleep lasts up to one day, so a newly inserted `ProcessPendingSelfRemove` task can be delayed until another unrelated task or event arrives — leaving self-removals stuck far longer than the previous 2s polling interval.

[insipx]

Verified this is not the case here. process_leave_request_message runs inside provider.key_store().transaction(|conn| { let storage = conn.key_store(); ... }) (mls_sync.rs ~1297). Within that closure, storage.db() is SqlKeyStore::db() → DbConnection::new(&self.conn), pointing at the same transaction-scoped connection that immediate_transaction opened — not a fresh autocommitting one. So PendingRemove.store_or_ignore(&storage.db()) and storage.db().create_task(...) execute on the same connection inside the same transaction and commit/roll back together (this is also why the existing rollback-on-error path for the pending-remove insert works). The atomicity comment is accurate. See the sibling thread — storage.db() inside the key_store().transaction() closure reuses the transactional connection, so the task insert is atomic with the pending-remove insert.

[tylerhawkes]

That's not the concern here. The issue is that the task worker is maximally lazy and will only reload when it knows it needs to. Sending it a message forces it to store and re-evaluate. If you want an atomic commit, then we'll need a new message that just wakes the task runner up and has it re-query for tasks (so that the new one can potentially be run before whatever it was planning on)

[macroscopeapp]

You're correct. I confirmed that when process_own_leave_request_message is called (from within the process_message transaction at mls_sync.rs:1078), it passes deferred_events = None. The conditional wake at line 1696 (if let Some(deferred_events) = deferred_events { deferred_events.wake_worker(...) }) doesn't fire, and there's no post-commit wake call after the transaction completes. The task sits until the worker's next scheduled iteration, which could be up to 24 hours if the table was previously empty.

The disappearing-messages path (lines 2098-2111) shows the pattern needed: set a flag during processing and wake after the transaction. The own-leave path lacks this mechanism.

[insipx]

Done in 1419643 — the Notify is gone. The task channel now carries a TaskMessage enum (New(NewTask) | Wake), and wake() just sends TaskMessage::Wake through the existing mpsc. The run loop handles Wake as a no-op and falls through to re-query get_next_task, so a task row committed directly in a DB transaction gets re-evaluated immediately (and can run ahead of whatever the worker was previously sleeping toward). One select arm fewer, no Notify permit semantics.

[macroscopeapp]

The TaskMessage::Wake infrastructure looks good — cleaner than the separate Notify. However, I still see process_own_leave_request_message (line 1603) passing None for deferred_events, so the conditional wake at line 1696 is skipped. The task is inserted but the worker isn't woken, meaning it could sit until the next scheduled poll (up to 24h if the table was empty). Should the own-leave path set a flag in ProcessedMessageOutcome (like disappearing_message_stored) and wake post-commit, similar to the disappearing-messages pattern at lines 2098-2111?

[insipx]

CI fix: d14n test_super_admin_promotion_marks_pending_leave_requests

The one hard failure in the prior test-workspace run (d14n shard, 4/4 tries) was test_super_admin_promotion_marks_pending_leave_requests — the rest were network flakes that retried green.

Root cause: the test asserts the transient has_pending_leave_request = Some(true) flag right after promoting Bola to super-admin. With event-driven self-remove, a freshly-promoted super-admin now processes the pending removal immediately via the TaskRunner (builds the removal commit, runs cleanup_pending_removal_list, which clears the flag) — so the transient true window is no longer observable; the assertion saw Some(false). It passed under v3 only because there the removal happened later.

Fix: run this test (and its sibling test_super_admin_demotion_clears_pending_leave_requests) with disable_workers, so the flag-marking-on-admin-change path is exercised in isolation without the TaskRunner racing the assertion. The eventual removal behavior is already covered by the test_self_removal* suite.

Verified locally under d14n (promotion 4/4 deterministic pass) and v3. The 9 test_commit_log_* failures seen when running the touched modules locally are not in scope — the ci-d14n nextest profile excludes all commit_log tests by default (not (test(/\w*commit_log*/))); they only run under v3.

[insipx]

CI fix: node-sdk + browser-sdk should filter messages with options

Same root cause as the Rust d14n fix, surfacing in the JS SDK test suites. The test does group2.requestRemoval() near the top purely to inject a LeaveRequest message for the content-type filter — but with event-driven self-remove, client1 (the super-admin) now processes that request during the test via the worker, adding a removal GroupUpdated commit. That bumped messages.length 13 → 14 and shifted several filtered counts.

The removal previously landed opportunistically during the ~13 sequential send* round-trips, so a hard count would be timing-fragile. Fix makes it deterministic: wait for the removal up front (syncAll() + sleep(4000), mirroring the sibling should remove a member after processing their removal request test) so the removal commit always lands before the sends, then update the affected expectations:

• total 13 → 14, Published 13 → 14
• GroupUpdated|LeaveRequest|Custom filter 3 → 4 (two GroupUpdated now)
• MembershipChange kind 1 → 2
• excludeSender(client2) 12 → 13, excludeContentTypes(Text/MD/Reply) 10 → 11
• sentBefore(reply) 5 → 6 (removal commit precedes the reply); sentAfter(reply) unchanged at 7

Applied identically to both sdks/js/node-sdk and sdks/js/browser-sdk.

[tylerhawkes]

This is a fair amount of work to avoid just adding a new Task enum that only wakes up the task runner to get it to re-evaluate.

[insipx]

Agreed — reworked in 1419643. Dropped the Notify entirely; wake() now sends a TaskMessage::Wake variant through the existing task mpsc channel, which the loop treats as "re-query the tasks table". The only remaining indirection is that the send still goes through DeferredEvents post-commit — sending mid-transaction would let the worker consume the wake and re-query before the row is visible (the race we hit with the Notify originally).

[macroscopeapp]

🟢 Low encrypted_store/tasks.rs:102

upsert_pending_self_remove_task takes a group_id and a separate NewTask, but nothing enforces that task encodes ProcessPendingSelfRemove for that group_id. A caller can pass a mismatched task — the function will delete dead rows for group_id, then insert the unrelated task unchanged, leaving stale self-remove rows or creating spurious tasks. Consider taking only group_id and constructing the NewTask internally, or validating that task matches group_id before proceeding.

🚀 Reply "fix it for me" or copy this AI Prompt for your agent:

In file @crates/xmtp_db/src/encrypted_store/tasks.rs around line 102:

`upsert_pending_self_remove_task` takes a `group_id` and a separate `NewTask`, but nothing enforces that `task` encodes `ProcessPendingSelfRemove` for that `group_id`. A caller can pass a mismatched `task` — the function will delete dead rows for `group_id`, then insert the unrelated `task` unchanged, leaving stale self-remove rows or creating spurious tasks. Consider taking only `group_id` and constructing the `NewTask` internally, or validating that `task` matches `group_id` before proceeding.

[macroscopeapp]

🟢 Low worker/tasks.rs:179

When TaskMessage::Wake arrives before next_wakeup, the tokio::select! drops the in-flight sleep future and restarts a fresh one from "now". If wake messages arrive frequently enough, the timer branch never becomes ready and due tasks are postponed indefinitely. Consider ignoring wake messages when a sleep is already pending, or using a deadline-based approach that doesn't reset the wait each iteration.

🚀 Reply "fix it for me" or copy this AI Prompt for your agent:

In file @crates/xmtp_mls/src/worker/tasks.rs around line 179:

When `TaskMessage::Wake` arrives before `next_wakeup`, the `tokio::select!` drops the in-flight sleep future and restarts a fresh one from "now". If wake messages arrive frequently enough, the timer branch never becomes ready and due tasks are postponed indefinitely. Consider ignoring wake messages when a sleep is already pending, or using a deadline-based approach that doesn't reset the wait each iteration.