feat(ethexe/rpc): impl multi-watcher fan-out for `injected_sendTransactionAndWatch'

[vobradovich]

Closes #5386 , #5402

Summary

Reworks injected_sendTransactionAndWatch to support multiple concurrent watchers on the same transaction hash, instead of one watcher per transaction.

• PromiseSubscriptionManager now keys subscribers by transaction hash only, using one tokio::sync::watch channel per hash that fans the receipt out to any number of receiver clones — no per-subscriber id bookkeeping.
• Late watchers (subscribing after the receipt is already stored) are served immediately from the database instead of racing the pending path; a stale Purged receipt no longer blocks a resubmission from getting its own relay.
• TransactionsRelayer dedups in-flight relays per transaction hash: concurrent watchers (and plain injected_sendTransaction callers) of the same transaction share a single relay and observe the same Accept/Reject outcome.
• PendingSubscriber now owns its cleanup: its Drop impl removes the subscribers-map entry once its receiver is the last one alive, so cleanup happens uniformly whether the watch finishes normally, is released before spawning, or is cancelled — no call site can forget it.

How to test

• cargo nextest run -p ethexe-rpc-server --no-fail-fast
• New coverage in ethexe/rpc/server/src/tests.rs and apis/injected/*: multiple/late watchers on one transaction, relay dedup and failure/rejection cleanup, concurrent clients, and the promise-manager race-window/Purged-receipt edge cases.

Notes

None.

Checklist

• PR title follows Conventional Commits (type(scope): description)
• Single logical change
• Tests added or updated (if logic changed)
• Docs updated (if needed)

[gemini-code-assist]

Summary of Changes

Hello, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request refactors the subscription layer for transaction watching to support multiple concurrent clients for a single transaction hash. By transitioning to a HashMap-based storage system and introducing a unique SubscriberId, the implementation now allows for efficient fan-out of receipts and immediate access to cached results for late callers, significantly improving system concurrency and reliability.

Highlights

• Multi-watcher support: Replaced single-subscriber storage with a HashMap keyed by SubscriberId, allowing multiple concurrent watchers for the same transaction hash.
• Cached delivery: Introduced a Ready path for late subscribers, enabling immediate delivery of cached receipts from the database without the need for re-relaying.
• Isolated cleanup: Implemented per-watcher cleanup using unique SubscriberIds, ensuring atomic removal and preventing race conditions during concurrent subscription management.
• Enhanced testing: Added comprehensive integration tests to verify concurrent watcher behavior, late cached delivery, and robust cleanup under various failure scenarios.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	/gemini review	Performs a code review for the current pull request in its current state.
Pull Request Summary	/gemini summary	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	/gemini help	Displays a list of available commands.

Customization

To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here.

Footnotes

1. Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

[gemini-code-assist]

Code Review

This pull request updates the ethexe RPC layer to support multiple concurrent watchers and late subscribers for the same transaction. The PromiseSubscriptionManager now keys subscribers by transaction hash and a unique SubscriberId, storing them in an inner HashMap within the DashMap to allow receipts to be fanned out to all registered watchers. Additionally, a new Ready path immediately serves late subscribers from the database if the receipt is already stored, bypassing the transaction relay. Comprehensive unit and integration tests have been added to verify these concurrent subscription, cancellation, and late-watcher behaviors. There are no review comments, so no further feedback is provided.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

Code Review

This pull request refactors the ethexe RPC server's transaction subscription and relaying mechanism to support multiple concurrent watchers per transaction hash. It replaces oneshot channels with watch channels, deduplicates in-flight relays so concurrent watchers share a single network broadcast, bounds the maximum number of concurrent watchers per transaction, and ensures automatic cleanup of subscribers via Drop implementations. A review comment identifies a potential race condition in relay.rs where a fast resubmission could observe a stale/cached outcome because the in-flight entry is removed after the outcome is published, suggesting to drop the scope guard before sending the outcome.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

There is a potential race condition where a subsequent call to relay for the same transaction hash can observe a stale/cached outcome from the previous relay. Because _guard is dropped after outcome_tx.send(Some(result)) is called, the in-flight entry might still exist in in_flight when a fast resubmission or retry occurs, causing it to return the cached result instead of initiating a fresh relay. Explicitly dropping the _guard before publishing the outcome ensures that any subsequent calls to relay will see a vacant entry and relay afresh.

Suggested change

	let result = relay_to_service(rpc_sender, transaction, tx_hash).await;
	// Publish first, or a concurrent caller could see a vacant entry and relay again.
	let _ = outcome_tx.send(Some(result));
	let result = relay_to_service(rpc_sender, transaction, tx_hash).await;
	drop(_guard);
	let _ = outcome_tx.send(Some(result));