Make provider coordinators event-driven off a single chain connection#291
Open
ilchu wants to merge 9 commits into
Open
Make provider coordinators event-driven off a single chain connection#291ilchu wants to merge 9 commits into
ilchu wants to merge 9 commits into
Conversation
Mechanical migration to the redesigned subxt 0.50 API, no behavior change intended: - blocks().subscribe_finalized() -> stream_blocks(); block-level events via block.at().events().fetch() - storage().at_latest().fetch/iter -> at_current_block() + storage().try_fetch/iter; addresses are now <KeyParts, Value> generic and keys are passed at call time - constants().at(..) -> at_current_block().constants().entry(..) - tx() -> at_current_block().transactions(); dynamic tx payload construction is unchanged - event decoding: variant_name() -> event_name(), field_values() -> decode_fields_unchecked_as; decoded values drop the u32 type-id context (Value<u32> -> Value) - fetch_raw missing values surface as StorageError::NoValueFound instead of None; call sites map it back to preserve error strings Verified: 424 tests pass, clippy -D warnings clean, and the live zombienet flows (just demo, fs-demo-ci, s3-demo-ci) all pass against a running relay + parachain + provider. Known semantic change inherited from subxt 0.50: transaction nonces are read from state at the anchored finalized block rather than the pool-aware system_accountNextIndex, so a transaction submitted within one finality lag of a previous same-account transaction (across processes) can see a stale nonce. In-process flows are unaffected because every submission waits for finalization.
One finalized-block follower (the chain-state coordinator) now owns the chain connection and fans out decoded events; the polling coordinators react to events instead of scanning storage maps on fixed intervals. - New chain_connection module: the single construction site for the chain client, published to all consumers through a watch channel. The provider's three independent sockets collapse into one connection. - New chain_events module: per-block BlockEvent fan-out (broadcast channel) decoded once by the follower - the block clock, challenge creation, replica agreements, checkpoint updates, and a Resubscribed marker after every (re)connect. - chain-state coordinator: adds a 60s finality-stall watchdog (a stream that yields no finalized block gets torn down and rebuilt - a stalled backend otherwise hangs forever with no error) and republishes the connection on every rebuild. - challenge responder: reacts to ChallengeCreated events with a targeted point read of the challenge (the event carries the id, not the proof parameters). The full Challenges scan now runs only at bootstrap/resubscribe and on a slow safety net (default 300s, was every 6s; 0 disables) - a missed challenge means getting slashed, so the event path is backstopped rather than trusted blindly. - replica-sync coordinator: duty passes run on relevant agreement and checkpoint events plus the same bootstrap/safety-net pattern (default 600s, was every 12s). - checkpoint coordinator: clocked by finalized blocks instead of wall time (checkpoint windows are a function of block height); the duty source itself remains the pre-existing stub. - SubxtChainClient and the auth membership resolver borrow the shared watch connection; extrinsic submission gains a bounded retry that treats pallet duplicate-rejections (ChallengeNotFound, CheckpointAlreadySubmitted, SyncTooFrequent) on the retry as proof the first attempt landed. Verified: 282 provider-node tests pass including new tests/event_fanout_integration.rs; clippy clean; live zombienet check confirmed a challenge autonomously defended through the event path alone (safety-net scan disabled, client never responding).
The patch-coverage gate flagged the subxt 0.50 migration's rewritten chain reads (RealChainStateClient, the finalized-block follow loop) as untested - previously they were only exercised against a live chain. Back a real OnlineClient (legacy backend) with subxt-rpcs' MockRpcClient and the repo's tracked runtime metadata snapshot, so the tests run the actual dynamic storage/constants/events machinery: - RequestTimeout constant lookup, asserted self-consistent against the raw constant bytes in the same metadata - Providers reads: absent -> None, and a full round-trip where the storage value is scale_value-encoded against the real runtime ProviderInfo type and decoded back (this immediately caught that the minimal decoder fixture lacks the runtime's public_key field) - the whole follow() pipeline over one finalized block: header subscription, System.Events decode, ProviderRegistered recognition, and the resulting provider-state refresh and nonce bootstrap connect_and_follow is split into connect + follow(api) so the tests can drive the full loop over the mock connection; behavior unchanged.
Reconciles the follower with PR A's follow-up work: connect_and_follow keeps the connect + follow(handle) split so the mock-RPC tests drive the full pipeline, now including the connection publish and the NewBlock/Resubscribed broadcast assertions.
Replace the substring matching in submit_and_finalize with structural matching on subxt 0.50's split error types, following the same style polkadot-sdk adopted in its 0.50 migration (paritytech/polkadot-sdk#12096): - a dispatch failure is TransactionFinalizedSuccessError::SuccessError( TransactionEventsError::ExtrinsicFailed(_)) - anything else means the watch or transport died before a verdict, which is the retryable class - a duplicate rejection is a DispatchError::Module whose metadata-resolved pallet is StorageProvider and whose error variant is one of ALREADY_DONE_ERRORS, instead of a substring of the formatted message
Member
|
Great work! I have the same idea about using an event queue/fan-out to wire all the provider components with the chain state coordinator. Based on our discussion about #178, I think it's better to merge your previous PR #268 first. After that, we'll do a big refactoring. The project will be much cleaner, and it will be easier to resolve all remaining technical debts. |
Collaborator
Author
This was referenced Jul 16, 2026
Open
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
Brings in the storage-subxt static bindings crate (#282) and converts this branch's new chain access to it as part of the resolution: - chain_events decodes ChallengeCreated / ReplicaAgreementEstablished / ProviderCheckpointSubmitted / BucketCheckpointed through the static event types instead of dynamic name + scale-value field lookups; a shape drift now surfaces as a logged decode failure (backstopped by the safety-net scans) instead of silent None lookups - fetch_challenge point-reads Challenges through the typed static address (unvalidated: bindings are generated from the paseo runtime and the local runtime shares the pallet), replacing the raw-offset decode on this path and picking up the ChunkLocation target shape - pre-existing dynamic call sites (subxt_client scans and tx payloads, chain-state reads, auth membership) are untouched: converting them is a repo-wide follow-up, not part of this branch Also folds the event fan-out tests into the coordinators test binary, reusing its shared helpers (test_state, wait_for, and the committed- chunk fixture now hosted in main.rs) instead of duplicating them in a separate integration target.
The patch-coverage gate flagged the static-bindings merge additions: - the three untested BlockEvent From impls get direct construction tests, and the mock-RPC follower fixture now emits a ChallengeCreated alongside ProviderRegistered, so the static decode path runs against real runtime metadata and the fan-out broadcast is asserted end to end - chain_connection: unreachable-chain connect and pre-connect current_api error paths - auth: the membership resolver's pre-connect lookup error path - checkpoint coordinator: block-tick arms (irrelevant event skipped, NewBlock/Resubscribed duty check, paused skip) via a live channel - replica-sync: BucketCheckpointUpdated relevance both ways (bucket held locally vs not) and a full duty pass through sync_and_confirm to PrimaryUnavailable (new root, no reachable primaries, no network); the checkpoint mock is reused crate-wide instead of duplicated
danielbui12
reviewed
Jul 20, 2026
Comment on lines
+339
to
+344
| BlockEvent::BucketCheckpointUpdated { bucket_id } => self | ||
| .state | ||
| .storage | ||
| .list_buckets() | ||
| .iter() | ||
| .any(|b| b.bucket_id == *bucket_id), |
Member
There was a problem hiding this comment.
O(1) lookup
Suggested change
| BlockEvent::BucketCheckpointUpdated { bucket_id } => self | |
| .state | |
| .storage | |
| .list_buckets() | |
| .iter() | |
| .any(|b| b.bucket_id == *bucket_id), | |
| BlockEvent::BucketCheckpointUpdated { bucket_id } => self | |
| .state | |
| .storage | |
| .get_bucket(bucket_id) | |
| .is_some(), |
Comment on lines
+22
to
26
| /// Retained for compatibility; duty checks are clocked by finalized | ||
| /// blocks from the chain-state coordinator, not by wall time (checkpoint | ||
| /// windows are a function of block height, so block arrival is the only | ||
| /// meaningful tick). | ||
| pub poll_interval: Duration, |
Member
There was a problem hiding this comment.
I don't get the "Retained for compatibility", could you elaborate more
If it should not be removed, mark it as #[deprecated]
danielbui12
reviewed
Jul 20, 2026
Comment on lines
+131
to
+153
| let checkpoint_events = events_tx.subscribe(); | ||
| let replica_events = events_tx.subscribe(); | ||
| let challenge_events = events_tx.subscribe(); | ||
|
|
||
| // Start optional background services (failures are non-fatal) | ||
| let _chain_state_handle = start_chain_state_coordinator(&cli, state.clone()); | ||
| let checkpoint_handle = | ||
| start_checkpoint_coordinator(&cli, chain_client.as_ref(), state.clone()).await; | ||
| let _chain_state_handle = | ||
| start_chain_state_coordinator(transport, chain_tx, events_tx, state.clone()); | ||
| let checkpoint_handle = start_checkpoint_coordinator( | ||
| &cli, | ||
| chain_client.as_ref(), | ||
| checkpoint_events, | ||
| state.clone(), | ||
| ) | ||
| .await; | ||
| if let Some(ref handle) = checkpoint_handle { | ||
| state.set_checkpoint_handle(handle); | ||
| } | ||
| let _replica_sync_handle = | ||
| start_replica_sync_coordinator(&cli, chain_client.as_ref(), state.clone()).await; | ||
| start_replica_sync_coordinator(&cli, chain_client.as_ref(), replica_events, state.clone()) | ||
| .await; | ||
| let _challenge_responder_handle = | ||
| start_challenge_responder(&cli, chain_client.as_ref(), state.clone()).await; | ||
| start_challenge_responder(&cli, chain_client.as_ref(), challenge_events, state.clone()) | ||
| .await; |
Member
There was a problem hiding this comment.
It is potentially to have shared signer race here, concurrent transactions may be submitted by signer from replica_sync_coordinator, challenge_responder.
We should add module to queue transaction (first thing come out in my mind, may be something else better to avoid race condition).
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Implements the event-driven coordinator redesign (#81) — the polling loops become event consumers, and the provider node's chain access consolidates onto a single owned connection. This is the second PR of the #222 sequence, stacked on #288; the smoldot transport follows and stays small because this PR builds the machinery it needs (single construction seam, watchdog, restart-driven re-bootstrap, implicit sync gating).
Architecture
One finalized-block follower — the existing chain-state coordinator — now owns the chain connection and fans out per-block events; everything else consumes.
chain_connectionmodule: the single place a subxt client is built, published to every consumer through atokio::sync::watchchannel. The provider's three independent WebSocket connections (shared signing client, chain-state coordinator, auth membership resolver) collapse into one. The transport enum has one variant today (Rpc); the smoldot variant lands in the follow-up as a construction-site-only change.chain_eventsmodule: the follower decodes each finalized block's events once and broadcasts the coordinator-relevant subset — the block clock (NewBlock),ChallengeCreated,ReplicaAgreementEstablished, checkpoint updates, and aResubscribedmarker after every (re)connect.Resubscribed, so every consumer re-bootstraps; no per-consumer reconnect logic remains anywhere.Coordinator changes
ChallengeCreatedevents with a targeted point read of the challenge (the event carries the id, not the proof parameters). The fullChallengesmap scan — previously every 6s — now runs only at startup/resubscribe/lag and on a slow safety net (--challenge-poll-interval, default 300s,0disables). The safety net is deliberate: a missed challenge means getting slashed, so the event path is backstopped rather than trusted blindly.--replica-poll-interval, default 600s, was every 12s with a fullStorageAgreementsscan per tick).Extrinsic submission
submit_and_finalizegains a bounded retry (one resubmit) on transport-level failures — a dropped socket or a backend resubscription killing the transaction watch leaves the outcome unknown, so the retry resolves it. Failure classification is structural on subxt 0.50's split error types (same style as polkadot-sdk's own 0.50 migration, paritytech/polkadot-sdk#12096): a dispatch failure isSuccessError(ExtrinsicFailed(_)), and a duplicate rejection is aDispatchError::Modulewhose metadata-resolved pallet/variant is one ofChallengeNotFound/CheckpointAlreadySubmitted/SyncTooFrequent— those on a retry prove the first attempt landed (the pallet rejects duplicates harmlessly by construction, verified against the extrinsic guards).Verification
283 provider-node tests pass, including a new
tests/event_fanout_integration.rs(event-triggered response with the safety net disabled, foreign-event filtering, resubscribe-triggered bootstrap scans) and the mock-RPC-backed follower tests from #288 extended to assert connection publishing and event fan-out. Live zombienet check: with--challenge-poll-interval 0(event path only) and the client deliberately never responding, the provider autonomously defended an on-chain challenge —ChallengeCreatedbroadcast → point read → proof → finalizedrespond_to_challenge.Relates to #222. Closes #81.