Shared event-batching slot has no ownership: concurrent writers can lose or misorder domain events #3217

Overview

NodeService routes every domain event through one process-wide slot, batch_state: Arc<Mutex<BatchState>>, shared by every clone of the service (UI writes, sync replay, import, play engine). The slot records what mode emission is in (Immediate, Batching, Transactional { buf, parked }) but not who owns it. Three concurrency defects follow from that single root cause. All three predate #3210 (which fixed a fourth instance, the ownerless-batch-after-guard-drop race, by parking an open batch inside the slot) and were found during that PR's review.

Problem Statement

1. A non-transactional write's event can be buffered into — and dropped by — another task's transaction

Non-_in_tx writes emit after releasing the store write guard: e.g. SqliteStore::create_node executes the INSERT under self.write(), drops the guard, then calls self.notify(...) (packages/core/src/db/sqlite_store/nodes.rs, create_node), and many NodeService methods call self.emit_event(...) after a non-transactional store call. If another task's with_transaction acquires the write guard in that window, it sets the slot to Transactional, and the first write's already-committed event is pushed onto that transaction's buffer (emit_event's Transactional { buf, .. } arm and the store-notifier closure's equivalent arm in NodeService::new). If that transaction rolls back, the buffer is discarded — the event for a genuinely committed write is lost (UI never updates, cloud push never sees it, reactive plays never fire). If it commits, the event is delivered but attributed to the wrong unit of work.

2. Two concurrent begin_batch_emit guards overwrite each other

begin_batch_emit only debug_assert!s the slot is free; in release it replaces an existing Batching map (or parked batch) with a fresh one, and the first guard's buffered events are silently discarded. Guards carry no identity, so when either guard drops it flushes/clears whatever batch is in the slot regardless of whose it is. Reachable in practice: e.g. bulk_delete (which opens its own guard) called while the sync replay holds its per-page guard.

3. A batch can keep an older NodeUpdated snapshot after two transactions commit

When a transaction commits inside an open batch, its buffered events are merged into the batch last-write-wins per node id after with_transaction returns (post-commit, outside the write guard). Two transactions on different tasks can therefore merge in the opposite order to their commits, leaving the batch holding the earlier transaction's NodeUpdated { node, .. } snapshot for a node. Consumers that re-read by id are unaffected; consumers that trust the event's embedded node payload see stale state.

Proposed Solution

Give the slot ownership instead of a bare mode:

• A batch guard gets an identity (e.g. a monotonically increasing token); BatchState holds batches keyed by owner, and a guard only ever flushes/clears its own batch. A second concurrent guard gets its own buffer rather than overwriting.
• Route an event to a transaction's buffer only when the emitting code is inside that transaction — e.g. carry the buffer on NodeServiceTx (ADR-069 §1b already names NodeServiceTx::buffer_event as the intended API) and have emit_event outside a transaction bypass Transactional entirely (broadcast, or join the emitting task's own batch).
• Perform the post-commit merge into a batch while still ordered with commit (e.g. merge before releasing the write guard, or stamp events with a commit sequence and keep the newest per node).

Architecture Impact

• ADR-069 §2 (nodespace-docs/decisions/069-unit-of-work-transaction-seam.md, "Events are collected during a transaction and emitted only after commit — BatchEmitGuard and the transaction buffer are the same mechanism") currently says a transactional buffer active while a batch guard is active should be rejected, and describes BatchState::Transactional(Vec<..>) with a reset to Immediate on commit/rollback. Dispatch ADR-060 invariant rules from every bulk write path #3210 already moved the code past that text (a transaction now parks an open batch in Transactional { buf, parked } and hands committed events to it). This issue extends §2 further (owned batches, tx-scoped buffering). A companion nodespace-docs edit to ADR-069 §2 is required so the ADR describes the ownership model actually implemented — including the Dispatch ADR-060 invariant rules from every bulk write path #3210 parking behaviour.

Acceptance Criteria

• An event emitted by a non-transactional write can never be buffered into, or discarded by, another task's transaction; regression test with a tokio::sync::Notify-paused transaction on one task and a plain create_node on another, rolling the transaction back, asserts the plain write's event is delivered
• Two concurrent batch guards each deliver exactly their own events on drop, in either drop order; no debug_assert is reachable in that scenario
• After two transactions on different tasks commit inside one open batch, the batch holds the event from the later commit for a shared node; regression test
• Existing Dispatch ADR-060 invariant rules from every bulk write path #3210 regressions still pass: batch_guard_dropped_during_a_transaction_leaves_the_slot_immediate, batch_opened_during_a_transaction_receives_its_committed_events, and packages/core/tests/bulk_invariant_dispatch_test.rs
• ADR-069 §2 updated in nodespace-docs to describe the ownership model
• Code passes bun run quality:fix; bun run test:changed passes

Technical Specifications

Reference Files

• packages/core/src/services/node_service/mod.rs: BatchState (+ end_transaction), BatchEmitGuard and its Drop, with_transaction (slot check-and-set, post-commit merge into an open batch), emit_event, begin_batch_emit, and the store-notifier closure registered in NodeService::new
• packages/core/src/db/sqlite_store/nodes.rs: create_node and other non-_in_tx writes that notify after dropping the write guard
• packages/core/src/services/node_service/bulk.rs: bulk_delete opens its own guard (candidate for the two-guards scenario)
• Downstream consumer: sync repo nodespaced-pro/src/cloud_sync/reconnect.rs / apply.rs hold a per-page begin_batch_emit guard around bulk_create/bulk_update with a per-row fallback — any ownership change must keep that contract (events for a page flush when the page guard drops)

Implementation Details

• The Dispatch ADR-060 invariant rules from every bulk write path #3210 regression tests in mod.rs show the pattern for deterministic cross-task interleaving: a with_transaction closure on a spawned task that signals started and waits on proceed (tokio::sync::Notify).
• Keep Vec ordering for a transaction's own events (ADR-069 §2: move_node relies on in-tx order); only batch merges are last-write-wins.

Housekeeping for the next core-pin bump (sync repo)

Comments in the sync repo's nodespaced-pro/src/cloud_sync/orchestrator.rs (around lines 780 and 991) still name bulk_create_hierarchy_root_notify, which #3210 removed (folded into bulk_create_hierarchy, identical behaviour: all nodes emit NodeCreated). Update them when sync next bumps its core pin; no code change needed there.

Related Issues

• ADR-060 synchronous invariant-rule dispatch is wired into every single-node create/update path but into none of the bulk write paths #2960 (bulk invariant dispatch; its PR introduced batch parking)
• Dispatch ADR-060 invariant rules from every bulk write path #3210 (PR; review surfaced these findings)