[Architecture Review][P1] #5301 Sub-PR D: cross-store fault-injection E2E

[qqeasonchen]

Phase 2 follow-up — Cross-store fault-injection E2E (#5301 Sub-PR D)

Source: Issue #5301 — Sub-PR D, deferred at issue close.
Severity: P1 (reliability correctness; not on the critical path for A2A / Runtime merge)
Blocks: — none directly; motivates future changes to DeadLetterStore / TaskStore / DeliveryStateStore

Why this is split out

The unified state control-plane stores (OffsetStore / SubscriptionStore / DeliveryStateStore / SessionStore / DeadLetterStore / TaskStore) were landed as separate SPI + concrete backends across PRs #5310 (Sub-PR A), #5311 (B), #5312 (C). Each sub-PR carried its own unit tests, but the six stores have never been exercised together in a fault-injection harness. Sub-PR D was the original plan for that.

Scenarios to cover

All six scenarios target the develop branch as of apache/eventmesh#5313 merged.

1. Crash mid-ACK → re-ACK on restart

   • Setup: submit a message; pause the dispatch loop just after DeliveryStateStore.put but before the connector ACK returns.
   • Inject: hard kill the JVM.
   • Expectation on restart: ReliableDispatcher.recover() re-enqueues the same deliveryId; consumer sees exactly one delivery.

2. Meta partition during DLQ transition

   • Setup: force a DLQ write; pause the connector ack callback.
   • Inject: kill the Meta client (close socket / 30s pause).
   • Expectation: DeadLetterStore.recordDeadLetter is the durable gate; the source delivery is NOT retired until the Meta write is acknowledged. After Meta recovers, the delivery is retired.

3. A2A task cancelled mid-stream

   • Setup: start a long-running A2A TaskRecord (e.g. via A2AGatewayService.submitTask against a slow agent).
   • Inject: cancelTask(taskId) while the agent is mid-SSE.
   • Expectation: TaskStore.updateStatus(..., Status.CANCELED, ...) succeeds; pending SSE subscribers get a terminal CANCELED event; the agent receives a cancel signal.

4. Subscription re-register after split

   • Setup: a consumer group is registered via SubscriptionStore.
   • Inject: Meta cluster split for 30s; two Runtime instances see different Meta views.
   • Expectation: when the split heals, both instances converge to the same subscription set; no duplicate redelivery for already-ACKed messages.

5. Offset-store race vs. delivery-store recovery

   • Setup: offset store and delivery state store are co-located but updated independently.
   • Inject: after a crash, recover delivery state first, then offset state.
   • Expectation: ACKs that landed between the two writes are honored exactly once.

6. A2A dispatch race vs. TaskStore state

   • Setup: an A2A agent is registered; a task is submitted through the Runtime dispatcher.
   • Inject: invoke cancelTask exactly at the moment the dispatcher dispatches the task to the agent.
   • Expectation: outcome is one of (delivered then canceled) or (canceled before dispatch) — both observable; final TaskRecord.status is CANCELED; DeadLetterStore does NOT receive a record for the canceled task.

Test harness

• Use Testcontainers + JUnit 5 (Java 21) — pattern established by the Runtime module's existing fault-injection tests in eventmesh-runtime/src/test/java/.../cluster/ClusterDeliveryFaultTest.
• For Meta: spin up the in-process Nacos-compatible Meta (InProcessMetaService) already used by the unit tests, plus an optional Meta kill-switch (close + reopen the socket) for partition scenarios.
• For the JVM crash scenario: use a child-process harness (ProcessBuilder + Runtime.exec) so the test can SIGKILL the runtime under test and re-launch it.

Acceptance

• All six scenarios have a JUnit test that reliably reproduces the failure mode (5/5 runs in CI).
• A short doc note (docs/architecture/state-stores-fault-injection.md) lists the scenarios and the property each one asserts.

Out of scope

• Performance / load (separate issue if needed).
• Production fault-injection tooling.
• Changes to the store APIs themselves — this is purely verification.

References

• Closed issue: [Architecture Review][P0] Unified state control-plane stores #5301
• Landed: feat(state): control-plane store SPI + deprecate zombie code (#5301 Sub-PR A) #5310 (A), fix(state): persist ReliableDispatcher.pending to DeliveryStateStore (issue #5301 Sub-PR B, fixes #5294 #5295) #5311 (B), feat(state): DeadLetterStore + TaskStore Meta backends (issue #5301 Sub-PR C, fixes #5292 fully, unblocks #5302) #5312 (C), feat(a2a): wire A2A Gateway onto Runtime via TaskStore (issue #5302 Sub-PR D1) #5313 (A2A via TaskStore)
• Related: [Architecture Review][P1] A2A must not form a parallel Runtime #5302 (A2A must not form a parallel Runtime, closed by feat(a2a): wire A2A Gateway onto Runtime via TaskStore (issue #5302 Sub-PR D1) #5313)

[qqeasonchen]

Plan: one-shot PR for all 6 scenarios

Plan to land #5314 as a single PR covering all 6 scenarios + harness + doc. Comment before coding so PMC can object early.

High-level structure

One new file eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/state/CrossStoreFaultInjectionTest.java with 6 @Nested classes (one per scenario), sharing a small harness package:

eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/state/fault/
├── MetaPartitionSwitch.java          # open/close Meta, simulate network split
├── JvmCrashHarness.java              # ProcessBuilder + SIGKILL + relaunch (scenario 1)
├── CrossStoreRaceProbe.java          # deterministic clock + ordered event log (scenarios 5/6)
└── PartitionedMetaStore.java         # already exists, reuse (scenarios 2/4)

One new doc docs/architecture/state-stores-fault-injection.md listing the 6 scenarios and the property each one asserts (per the issue's Acceptance section).

Scenarios mapping (file ↔ scenario)

@nested class	Scenario	Test technique
CrashMidAckReAck	1. Crash mid-ACK → re-ACK on restart	JvmCrashHarness (child ProcessBuilder JVM, SIGKILL at gate, relaunch, assert recover() re-enqueues)
MetaPartitionDuringDlq	2. Meta partition during DLQ	MetaPartitionSwitch + existing PartitionedMetaStore, gate the Meta write between DeadLetterStore.recordDeadLetter and source delivery retire
A2aCancelMidStream	3. A2A cancel mid-stream	A2AGatewayService + TaskStore (already exists in #5313), in-process SSE subscriber, cancelTask(taskId) race
SubscriptionReRegisterAfterSplit	4. Subscription re-register after Meta split	Two SubscriptionStore instances over PartitionedMetaStore, simulate split by closing the underlying InMemoryMetaStore write path
OffsetStoreRaceVsDeliveryStore	5. Offset vs delivery-store race	CrossStoreRaceProbe with deterministic clock + ordered writes; assert "ACKs that landed between the two writes are honored exactly once"
A2aDispatchRaceVsTaskStore	6. A2A dispatch race vs TaskStore state	Mock dispatcher with deferred deliver hook; cancelTask injected exactly at the deliver moment; assert TaskRecord.status == CANCELED and DeadLetterStore has no record for the canceled task

Testcontainers deviation

The issue recommends Testcontainers. I'm deviating for these reasons (called out explicitly in the PR description):

1. Local dev environment has no Docker available (verified docker: command not found).
2. The repo has no existing Testcontainers dependency in any module (verified git grep testcontainers returns empty).
3. All 6 scenarios are reproducible in-process:
   • Meta partition: PartitionedMetaStore wrapper (already exists in ClusterDeliveryFaultTest)
   • JVM crash: child-process ProcessBuilder (no Docker required, runs on plain JDK)
   • A2A: A2AGatewayService + TaskStore are already JVM-internal (no external broker required)
4. The Testcontainers addition would be a separate concern (it can be added later if/when a CI runner with Docker is wired up).

If PMC wants Testcontainers specifically, I can split the harness into a TestcontainersHarness profile and land the Docker version in a follow-up PR.

Sub-task split rationale

I'm choosing one-shot rather than two PRs because:

• The harness classes (MetaPartitionSwitch / JvmCrashHarness / CrossStoreRaceProbe) are needed by multiple scenarios
• Splitting would force init repo #1 to drag in the harness with only 1 user, then ignore some tests #2/How to achieve high concurrency and low latency？ #3 to add the remaining harness with 2-3 more users — twice the review surface
• Issue text already says "sub-PR D was the original plan for that" implying a single landing

Risk + rollback

• 6 scenarios in one file = ~600-800 LoC test code. Review burden is real; happy to split @nested classes into separate files if preferred.
• Scenario 1 (JVM crash) is the only one requiring real OS-level process control. If CI can't run it, mark it @EnabledIfEnvironmentVariable("ENABLE_JVM_CRASH_HARNESS") and keep the rest unconditional.
• No production code changes — pure verification (matches the issue's "Out of scope" section).

Estimated timeline

• Day 1-2: harness classes + scenario 2 (Meta partition DLQ) — simplest, validates harness
• Day 3-4: scenarios 3/4/6 (A2A + subscription) — share A2AGatewayService wiring
• Day 5-6: scenarios 1/5 (JVM crash + cross-store race) — most complex
• Day 7: doc + checkstyle + push PR

Pinging @qqeasonchen (me) for re-review after the PR is up.

— qqeasonchen (apache/eventmesh PMC)