[Architecture Review][P0] Unified state control-plane stores

[qqeasonchen]

Phase 2 — Unified state control plane

Execution meta — see plan: #5296

• Phase: 2 — unified state control plane
• Severity: P0
• Depends on: [Bug] Make RocksDB and Meta offset writes atomically monotonic #5289, [Bug] Do not retire deliveries when durable offset persistence fails #5290, [Bug] Prevent stale ACKs from matching reused delivery IDs #5291, [Bug] Make DLQ transition durable before retiring source delivery #5292 (Phase 1 — DONE: PR fix(reliability): atomic offset writes, durable DLQ, epoch delivery IDs (#5289 #5290 #5291 #5292) #5307 merged b43df620d 2026-08-19)
• Blocks: [Bug] Preserve unacknowledged broadcast deliveries during cursor recovery #5294, [Bug] Gate RocketMQ 5 POP broker ACK on distribution completion #5295, [Architecture Review][P1] A2A must not form a parallel Runtime #5302

Problem

State persistence is inconsistent across subsystems. Offsets have a RocksDB path, but subscription registration, push buffers, in-flight deliveries, and A2A tasks remain largely in-memory. A2A documentation also states that InMemoryA2AMessageTransport and the in-memory TaskRegistry are part of the current implementation.

Risks

• A Runtime crash, restart, migration, or network partition produces uncontrollable redelivery or state loss.
• A2A can bypass the Runtime's reliability, DLQ, audit, and authorization paths.
• "At-least-once" cannot be formed into an end-to-end, verifiable contract.

Current state (post #5307 + #5308)

Concrete on-develop assessment as of 2026-08-20 (b43df620d + 7260581):

Store	Current implementation	Persistent?	Cluster-shared?	Adequate?
OffsetStore	interface + InMemoryOffsetStore + RocksDBOffsetStore + MetaBackedOffsetStore (local+remote dirty-flush)	✓ RocksDB	✓ via Meta async flush	Yes — pattern is the template for the rest
SubscriptionStore	ClusterSubscriptionStore (Meta prefix-watch + local cache)	partial (Meta)	✓	API shape wrong — exposes targetsFor(...) cluster view, not a general KV API
DeliveryStateStore	ReliableDispatcher.pending = ConcurrentHashMap<String, Delivery>	none	n/a	No — crash = orphan all in-flight deliveries
SessionStore	SessionRegistry (Meta prefix-watch + local cache; agents/bindings/sessions)	partial (Meta)	✓	Acceptable — already unified, Meta round-trip is rare (lifecycle only, not per-message)
DeadLetterStore	DeadLetterSink interface (CompletableFuture)	downstream-only	n/a	No — only tracks "已 dead-letter" by retiring the delivery; no durable "DLQ ledger"
TaskStore	absent — a2a/TaskRegistry.java deleted by #5274 revert; only a2a/EventMeshA2ATransport.java (58 lines) remains	n/a	n/a	No — A2A tasks have no state model at all

The #5301 acceptance criterion "Each store has an interface + at least one durable implementation" is met only for OffsetStore. The other five either lack a clean interface (DeliveryStateStore is a ConcurrentHashMap field), have no persistent back-end, or do not exist.

The second criterion "Delivery state is recoverable after a hard restart" is not met — the most concrete gap.

The third "ACK validation binds to delivery ownership (id + epoch + idempotency key)" — the deliveryId already includes bootEpoch + instanceSalt + seq (#5291), but there is no persistent ledger to bind it against on restart.

The fourth "A2A tasks persist through TaskStore and route through the Runtime dispatcher" requires a fresh TaskStore plus integration with EventMeshA2ATransport.

Proposed direction (revised — staged delivery)

Three-tier persistence model

The 6 stores fall into three tiers by access pattern. Forcing them into a single backend is the wrong abstraction; each gets the one that matches its access pattern.

Tier	Stores	Backend	Rationale
local-only (per-instance, hot path)	OffsetStore, DeliveryStateStore	RocksDB + in-process LRU cache	High write rate (per-ACK), per-key independence — no need for cluster sharing, just crash recovery
cluster-shared (cross-instance, lifecycle)	SubscriptionStore, SessionStore, TaskStore	Meta (Nacos/etcd) + local cache	Low write rate, must be visible to peers — Meta is the natural fit
durable-egress (terminal state)	DeadLetterStore	Downstream DLQ topic (existing) + Meta "DLQ ledger" for "which deliveryIds are confirmed dead"	The message body lives in the MQ DLQ topic; we only need to durably mark the transition so the dispatcher can retire the delivery

This is not a single SPI — it is a contract per tier, with the existing MetaStore interface (#5308) acting as the cluster-shared tier's backend. The OffsetStore + MetaBackedOffsetStore pattern (local-first, async dirty-flush to Meta) is the reference design.

Acceptance contract per store

Store	Interface methods (min)	Backend	Atomicity	Restart-recovery
OffsetStore	readOffset, writeOffset (returns boolean), readAllOffsets, readAllTopics, flush, close	RocksDB (local) + Meta async flush	per-key monotonic via writeLock (#5289 ✓)	✓
DeliveryStateStore	put(Delivery), remove(deliveryId), get(deliveryId), iterate(consumer), count(), flush, close	RocksDB (local) + bounded LRU write-cache	per-delivery put is last-writer-wins; remove wins on retry-vs-ack race (#5290 pattern)	on restart: re-ACK all in-flight deliveries with stored offset as the cursor
SubscriptionStore	put(topic,clientId,sub), remove(topic,clientId), targetsFor(topic,event), instanceOf(clientId), topics()	Meta prefix /em/subs/... (existing)	Meta tryAcquire CAS on register race	partial — subscriptions re-register on instance restart, no separate recovery
SessionStore	(already on SessionRegistry)	Meta prefix /em/agents/, /em/bindings/, /em/sessions/ (existing)	per-agent heartbeat re-write	heartbeat TTL eviction; on restart we must mark self READY again
DeadLetterStore	recordDeadLetter(deliveryId, dlqOffset) (idempotent), isDeadLettered(deliveryId), flush, close	Meta key /em/dlq/<deliveryId> = <dlqTopic>:<offset>	CAS on first record	on restart, query ledger before retiring any pending delivery
TaskStore	createTask, getTask, updateStatus(taskId, status), listByAgent(agentId, status), expireStale(olderThanMs)	Meta prefix /em/tasks/<taskId> (new)	per-task last-writer-wins with task epoch	tasks survive Runtime restart; A2A SSE streams are rebuilt from getTask

Delivery model (revised)

The single highest-leverage change is making ReliableDispatcher state survive a hard restart. Concretely:

1. Persist pending to RocksDB on every put/remove/reschedule — bounded LRU cache (e.g. 10K entries) + async batch flush. Throughput target: ≤ 10% regression vs. the in-memory baseline (InMemoryOffsetStore-style).
2. On restart, UniRuntime.start() calls ReliableDispatcher.recover() — read all persisted pending entries, re-ACK each one with its stored offset and clientId (no re-deliver, the MQ cursor was already advanced past these). The MqAckCallback is not re-run on recovery (broker already considered the message gone).
3. tick() resumes from the persisted nextAttemptAt clock — so a delivery that was 3 seconds from retry when the JVM died retries 3 seconds after the new instance is up, not on the next 30-second ACK window.
4. Add DeadLetterStore.recordDeadLetter as a separate gate — when tick() calls dlqSink.deadLetter(...), we only retire after both the downstream DLQ topic write succeeds AND DeadLetterStore.recordDeadLetter succeeds (Meta CAS). The Meta record is the "is this deliveryId confirmed dead" ledger for restart.

Subscription and Session (revised scope)

Both already work via Meta + local cache. The #5301 work here is API alignment, not rewrite:

• SubscriptionStore gains a thin interface wrapping ClusterSubscriptionStore (so OffsetStore-style swapping is possible), and the local cache is moved from ClusterSubscriptionStore itself to a generic CachingMetaStore<K,V> helper. No semantic change.
• SessionStore already has the right shape; the work is documentation + a SessionStore interface extracted from SessionRegistry (same rationale).

TaskStore (new)

Reintroduce the A2A task state model lost in #5274. Foundations still in tree:

• a2a/EventMeshA2ATransport.java (58 lines) — transport contract
• a2a/TaskRegistry.java is gone; we rebuild a fresh, Meta-backed implementation

Design constraints (carried over from the original #5259 design, refined):

1. TaskStore is the only writer to /em/tasks/<taskId>; A2A handlers read through it.
2. TaskRecord = { taskId, agentId, clientId, status, createdAt, updatedAt, input, output? }; status ∈ {PENDING, RUNNING, COMPLETED, FAILED, CANCELED}.
3. A2A inbound calls (/a2a/tasks/send, /a2a/tasks/sendSubscribe) write TaskRecord via TaskStore.createTask; SSE streams (/a2a/tasks/{id}/stream) read status transitions through TaskStore.getTask.
4. The Runtime dispatcher is the sole transport — EventMeshA2ATransport is invoked from SubscriptionManager (mode = a new A2A_DISPATCH), not from a parallel A2AGateway ([ISSUE #5259] Add A2A Gateway: REST API, SSE streaming, Task lifecycle, Java SDK and tests #5260 was the parallel path we want to avoid). This is the property the original [Enhancement][A2A] Agent-to-Agent Gateway: REST API, SSE streaming, Task lifecycle, Java SDK #5259 design missed and the cause of Revert: drop A2A Gateway (#5260) and Agent Card Registry (#5246) — migrate to develop #5274.
5. A task-expiry reaper (mirrors SessionRegistry.expireStaleSessions) handles PENDING tasks idle for > N minutes.

Staged delivery (sub-PRs)

The work is too large for a single PR. Proposing four sub-PRs, each independently mergeable:

Sub-PR	Title	Scope	Touches	Approx. PR size	Unblocks
A	feat(state): introduce StatefulStore SPI + 6 store interfaces	Define 6 store interfaces; extract SubscriptionStore / SessionStore from their concrete classes; add StatefulStore factory for InMemory / RocksDB / Meta-backed backends; baseline unit tests for the SPI	~ 12 files (+400 / -200)	small	locks the API surface for B/C/D
B	fix(state): persist ReliableDispatcher.pending to RocksDB (fixes #5294 #5295)	Implement RocksDBDeliveryStateStore + bounded LRU; add ReliableDispatcher.recover() called from UniRuntime.start; in-process fault-injection tests (crash mid-delivery, restart, re-ACK); per-#5308 ClusterDeliveryFaultTest style	~ 6 files (+800 / -100)	medium	#5294 #5295 (concrete bugs)
C	feat(state): DeadLetterStore + TaskStore (fixes #5292 fully)	Implement DeadLetterStore (Meta-backed ledger, gates retirement); reintroduce TaskStore + TaskRecord; EventMeshA2ATransport wires to Runtime dispatcher; a2a mode in SubscriptionManager	~ 10 files (+700 / -100)	medium	#5302
D	test(state): cross-store crash-recovery + split-brain fault injection	6-store integration scenarios: 1) crash mid-ACK → re-ACK on restart; 2) Meta partition during DLQ transition; 3) A2A task cancelled mid-stream; 4) subscription re-register after split; 5) offset-store race vs. delivery-store recovery	~ 4 files (+600)	small	closes #5301 acceptance criteria

Sub-PR ordering: A → B → C → D (A is a pure refactor, B/C can proceed in parallel after A, D after B+C).

Sub-PR A and B are the minimum needed to consider #5301 materially advanced: A locks the contract, B delivers the headline acceptance criterion ("Delivery state is recoverable after a hard restart") and incidentally closes #5294 / #5295.

Acceptance criteria (revised)

• All 6 stores expose a stable interface (Sub-PR A)
• StatefulStore SPI with at least one persistent backend for each store (Sub-PR A)
• DeliveryStateStore recovers across a hard restart — no orphaned in-flight deliveries (Sub-PR B)
• DeadLetterStore.recordDeadLetter is the durable gate before a delivery is retired (Sub-PR C)
• TaskStore reintroduced; A2A tasks persist and route through the Runtime dispatcher, not a parallel gateway (Sub-PR C)
• ACK validation binds to delivery ownership — deliveryId already encodes bootEpoch + instanceSalt + seq ([Bug] Prevent stale ACKs from matching reused delivery IDs #5291) and is now persisted via DeliveryStateStore, so a stale ACK cannot match a fresh delivery after restart (Sub-PR B)
• Cross-store fault-injection tests pass (Sub-PR D)

Open questions for reviewer

1. Sub-PR A interface names — happy with OffsetStore / SubscriptionStore / DeliveryStateStore / SessionStore / DeadLetterStore / TaskStore? The current code already has OffsetStore and SessionRegistry-shape; only the latter four would gain new interface files.
2. A2A reintroduction scope — Sub-PR C recreates TaskStore and an A2A dispatch mode. Should it also bring back the A2A Gateway REST surface ([ISSUE #5259] Add A2A Gateway: REST API, SSE streaming, Task lifecycle, Java SDK and tests #5260, reverted by Revert: drop A2A Gateway (#5260) and Agent Card Registry (#5246) — migrate to develop #5274), or land TaskStore first and defer the HTTP surface to a follow-up issue?
3. DeliveryStateStore write throughput — proposing 10K-entry LRU + 100ms batch flush as the default. If the existing ReliableDispatcher baseline is 50K ACKs/s, this should keep ≥ 90% of that; if measured < 90%, fall back to RocksDB write-through (no LRU) and accept the latency.
4. Backward compatibility — the existing ClusterSubscriptionStore and SessionRegistry are public API. Sub-PR A must keep them as concrete classes; the interfaces are additions. If the project requires breaking changes, follow [Bug] Unify cluster delivery topology and fence stale partition owners #5293's enum-removal pattern and document the migration in a CHANGELOG entry.

Related concrete bugs

The following at-least-once reliability defects in the current develop code are the concrete manifestations of this missing unified state control plane. Landing this issue should fix them together:

• [Bug] Make RocksDB and Meta offset writes atomically monotonic #5289 — RocksDB/Meta offset writes are not atomically monotonic — fixed by fix(reliability): atomic offset writes, durable DLQ, epoch delivery IDs (#5289 #5290 #5291 #5292) #5307 (Phase 1)
• [Bug] Do not retire deliveries when durable offset persistence fails #5290 — deliveries are retired when durable offset persistence fails — fixed by fix(reliability): atomic offset writes, durable DLQ, epoch delivery IDs (#5289 #5290 #5291 #5292) #5307 (Phase 1)
• [Bug] Prevent stale ACKs from matching reused delivery IDs #5291 — stale ACKs match reused delivery IDs — fixed by fix(reliability): atomic offset writes, durable DLQ, epoch delivery IDs (#5289 #5290 #5291 #5292) #5307 (Phase 1)
• [Bug] Make DLQ transition durable before retiring source delivery #5292 — DLQ transition is not durable — partially fixed by fix(reliability): atomic offset writes, durable DLQ, epoch delivery IDs (#5289 #5290 #5291 #5292) #5307 (DLQ before retire); ledger fully closes in Sub-PR C
• [Bug] Preserve unacknowledged broadcast deliveries during cursor recovery #5294 — unacknowledged broadcast deliveries are lost during cursor recovery — fixed by Sub-PR B
• [Bug] Gate RocketMQ 5 POP broker ACK on distribution completion #5295 — RocketMQ 5 POP broker ACK is not gated on distribution completion — fixed by Sub-PR B + C

Related

• Plan: #5296
• Predecessors: [Bug] Make RocksDB and Meta offset writes atomically monotonic #5289, [Bug] Do not retire deliveries when durable offset persistence fails #5290, [Bug] Prevent stale ACKs from matching reused delivery IDs #5291, [Bug] Make DLQ transition durable before retiring source delivery #5292 (PR fix(reliability): atomic offset writes, durable DLQ, epoch delivery IDs (#5289 #5290 #5291 #5292) #5307 ✓), [Bug] Unify cluster delivery topology and fence stale partition owners #5293, [Bug] Start and manage membership heartbeats in sticky cluster mode #5288 (PR fix(cluster): unify delivery topology with Meta CAS + fencing (#5293 #5288) #5308 ✓)
• Successors: [Bug] Preserve unacknowledged broadcast deliveries during cursor recovery #5294, [Bug] Gate RocketMQ 5 POP broker ACK on distribution completion #5295, [Architecture Review][P1] A2A must not form a parallel Runtime #5302
• Architecture doc: docs/eventmesh-uni-architecture-redesign.md §13.2 (will be updated per sub-PR)

[qqeasonchen]

I've rewritten the issue body with a structured implementation plan, based on a code review of the current develop (b43df620d + 7260581). The key updates:

Problem characterization refined with current-state table. The previous body said "subscriptions/push buffers/in-flight deliveries/A2A tasks remain largely in-memory" without quantifying which. The new body tabulates all 6 stores and shows that only OffsetStore meets the #5301 acceptance criteria today; the other five either have no interface (DeliveryStateStore is a ConcurrentHashMap field), no persistent back-end, or do not exist at all.

Three-tier persistence model. Rather than forcing a single backend for all 6 stores, the new proposal groups them by access pattern:

• local-only (OffsetStore, DeliveryStateStore) → RocksDB + LRU
• cluster-shared (SubscriptionStore, SessionStore, TaskStore) → Meta + local cache
• durable-egress (DeadLetterStore) → downstream DLQ topic + Meta ledger

The existing OffsetStore + MetaBackedOffsetStore pattern (local-first, async dirty-flush to Meta) is the reference design.

Acceptance contract per store. A table listing the minimum interface methods, the backend, the atomicity model, and the restart-recovery story for each of the 6 stores.

Staged delivery (4 sub-PRs). The previous body had no delivery plan. The new body proposes:

• Sub-PR A — interfaces + SPI (small, ~400 lines, locks the API surface)
• Sub-PR B — DeliveryStateStore persistence + ReliableDispatcher.recover() (medium, ~800 lines, fixes [Bug] Preserve unacknowledged broadcast deliveries during cursor recovery #5294 and [Bug] Gate RocketMQ 5 POP broker ACK on distribution completion #5295)
• Sub-PR C — DeadLetterStore + TaskStore reintroduction (medium, ~700 lines, fully closes [Bug] Make DLQ transition durable before retiring source delivery #5292 and unblocks [Architecture Review][P1] A2A must not form a parallel Runtime #5302)
• Sub-PR D — cross-store fault-injection tests (small, ~600 lines, closes the [Architecture Review][P0] Unified state control-plane stores #5301 acceptance criteria)

Sub-PRs A → B → C → D in order; B and C can run in parallel after A. A + B alone would be a meaningful #5301 milestone (delivery-state recovery is the headline acceptance criterion).

Concrete design constraints carried over. EventMeshA2ATransport (58 lines, still in tree) becomes the Runtime-dispatcher transport, not a parallel A2AGateway — the property the original #5259 design missed and the reason #5274 reverted it.

Open questions for reviewer. Four: interface naming, A2A reintroduction scope, write-throughput budget, and backward-compatibility policy.

Happy to break this into the 4 sub-issues/PRs (or merge them, your call) once the maintainers weigh in on the open questions. Architecture doc (docs/eventmesh-uni-architecture-redesign.md §13.2) will be updated per sub-PR.