From 4ddeda1e4177c910be5938227aee711efaf559aa Mon Sep 17 00:00:00 2001 From: Alexander Samusev <41779041+alvicsam@users.noreply.github.com> Date: Thu, 18 Apr 2024 16:22:00 +0200 Subject: [PATCH 1/7] [ci] Use ci-unified reference (#4196) close https://github.com/paritytech/ci_cd/issues/974 --- .gitlab-ci.yml | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index 77d31d96ee107..5e57dd86f1416 100644 --- a/.gitlab-ci.yml +++ b/.gitlab-ci.yml @@ -21,8 +21,7 @@ workflow: - if: $CI_COMMIT_BRANCH variables: - # CI_IMAGE: !reference [.ci-unified, variables, CI_IMAGE] - CI_IMAGE: "docker.io/paritytech/ci-unified:bullseye-1.77.0-2024-04-10-v20240408" + CI_IMAGE: !reference [.ci-unified, variables, CI_IMAGE] # BUILDAH_IMAGE is defined in group variables BUILDAH_COMMAND: "buildah --storage-driver overlay2" RELENG_SCRIPTS_BRANCH: "master" From 91d4a207af43f8f81f56e4f24af74f7c6f590148 Mon Sep 17 00:00:00 2001 From: ordian Date: Thu, 18 Apr 2024 16:32:14 +0200 Subject: [PATCH 2/7] chain-selection: allow reverting current block (#4103) Block reversion of the current block is technically possible as can be seen from https://github.com/paritytech/polkadot-sdk/blob/39b1f50f1c251def87c1625d68567ed252dc6272/polkadot/runtime/parachains/src/disputes.rs#L1215-L1223 - [x] Fix the test --- polkadot/node/core/chain-selection/src/lib.rs | 6 +-- .../node/core/chain-selection/src/tests.rs | 43 +++++++++++++++++-- .../node/core/chain-selection/src/tree.rs | 42 ++++++++++-------- 3 files changed, 65 insertions(+), 26 deletions(-) diff --git a/polkadot/node/core/chain-selection/src/lib.rs b/polkadot/node/core/chain-selection/src/lib.rs index 6f864fefb6110..07c245e839bf1 100644 --- a/polkadot/node/core/chain-selection/src/lib.rs +++ b/polkadot/node/core/chain-selection/src/lib.rs @@ -619,7 +619,7 @@ async fn handle_active_leaf( // Extract all reversion logs from a header in ascending order. // -// Ignores logs with number >= the block header number. +// Ignores logs with number > the block header number. fn extract_reversion_logs(header: &Header) -> Vec { let number = header.number; let mut logs = header @@ -639,14 +639,14 @@ fn extract_reversion_logs(header: &Header) -> Vec { None }, - Ok(Some(ConsensusLog::Revert(b))) if b < number => Some(b), + Ok(Some(ConsensusLog::Revert(b))) if b <= number => Some(b), Ok(Some(ConsensusLog::Revert(b))) => { gum::warn!( target: LOG_TARGET, revert_target = b, block_number = number, block_hash = ?header.hash(), - "Block issued invalid revert digest targeting itself or future" + "Block issued invalid revert digest targeting future" ); None diff --git a/polkadot/node/core/chain-selection/src/tests.rs b/polkadot/node/core/chain-selection/src/tests.rs index bc998f268a0da..1fe87f04cd585 100644 --- a/polkadot/node/core/chain-selection/src/tests.rs +++ b/polkadot/node/core/chain-selection/src/tests.rs @@ -966,19 +966,54 @@ fn ancestor_of_unviable_is_not_leaf_if_has_children() { } #[test] -fn self_and_future_reversions_are_ignored() { +fn self_reversions_are_not_ignored() { test_harness(|backend, _, mut virtual_overseer| async move { let finalized_number = 0; let finalized_hash = Hash::repeat_byte(0); // F <- A1 <- A2 <- A3. // - // A3 reverts itself and future blocks. ignored. + // A3 reverts itself + + let (_, chain_a) = + construct_chain_on_base(vec![1, 2, 3], finalized_number, finalized_hash, |h| { + if h.number == 3 { + add_reversions(h, vec![3]) + } + }); + + let a2_hash = chain_a.iter().rev().nth(1).unwrap().0.hash(); + + import_blocks_into( + &mut virtual_overseer, + &backend, + Some((finalized_number, finalized_hash)), + chain_a.clone(), + ) + .await; + + assert_backend_contains(&backend, chain_a.iter().map(|(h, _)| h)); + assert_leaves(&backend, vec![a2_hash]); + assert_leaves_query(&mut virtual_overseer, vec![a2_hash]).await; + + virtual_overseer + }); +} + +#[test] +fn future_reversions_are_ignored() { + test_harness(|backend, _, mut virtual_overseer| async move { + let finalized_number = 0; + let finalized_hash = Hash::repeat_byte(0); + + // F <- A1 <- A2 <- A3. + // + // A3 reverts future blocks. ignored. let (a3_hash, chain_a) = construct_chain_on_base(vec![1, 2, 3], finalized_number, finalized_hash, |h| { if h.number == 3 { - add_reversions(h, vec![3, 4, 100]) + add_reversions(h, vec![4, 100]) } }); @@ -1006,7 +1041,7 @@ fn revert_finalized_is_ignored() { // F <- A1 <- A2 <- A3. // - // A3 reverts itself and future blocks. ignored. + // A3 reverts finalized F and its ancestors. ignored. let (a3_hash, chain_a) = construct_chain_on_base(vec![1, 2, 3], finalized_number, finalized_hash, |h| { diff --git a/polkadot/node/core/chain-selection/src/tree.rs b/polkadot/node/core/chain-selection/src/tree.rs index b4aba30368a62..1eb6c13a7f823 100644 --- a/polkadot/node/core/chain-selection/src/tree.rs +++ b/polkadot/node/core/chain-selection/src/tree.rs @@ -236,7 +236,7 @@ fn propagate_viability_update( Ok(()) } -/// Imports a new block and applies any reversions to ancestors. +/// Imports a new block and applies any reversions to ancestors or the block itself. pub(crate) fn import_block( backend: &mut OverlayedBackend, block_hash: Hash, @@ -246,25 +246,29 @@ pub(crate) fn import_block( weight: BlockWeight, stagnant_at: Timestamp, ) -> Result<(), Error> { - add_block(backend, block_hash, block_number, parent_hash, weight, stagnant_at)?; - apply_ancestor_reversions(backend, block_hash, block_number, reversion_logs)?; + let block_entry = + add_block(backend, block_hash, block_number, parent_hash, weight, stagnant_at)?; + apply_reversions(backend, block_entry, reversion_logs)?; Ok(()) } // Load the given ancestor's block entry, in descending order from the `block_hash`. -// The ancestor_number must be at least one block less than the `block_number`. +// The ancestor_number must be not higher than the `block_entry`'s. // // The returned entry will be `None` if the range is invalid or any block in the path had // no entry present. If any block entry was missing, it can safely be assumed to // be finalized. fn load_ancestor( backend: &mut OverlayedBackend, - block_hash: Hash, - block_number: BlockNumber, + block_entry: &BlockEntry, ancestor_number: BlockNumber, ) -> Result, Error> { - if block_number <= ancestor_number { + let block_hash = block_entry.block_hash; + let block_number = block_entry.block_number; + if block_number == ancestor_number { + return Ok(Some(block_entry.clone())) + } else if block_number < ancestor_number { return Ok(None) } @@ -300,7 +304,7 @@ fn add_block( parent_hash: Hash, weight: BlockWeight, stagnant_at: Timestamp, -) -> Result<(), Error> { +) -> Result { let mut leaves = backend.load_leaves()?; let parent_entry = backend.load_block_entry(&parent_hash)?; @@ -308,7 +312,7 @@ fn add_block( parent_entry.as_ref().and_then(|parent| parent.non_viable_ancestor_for_child()); // 1. Add the block to the DB assuming it's not reverted. - backend.write_block_entry(BlockEntry { + let block_entry = BlockEntry { block_hash, block_number, parent_hash, @@ -319,7 +323,8 @@ fn add_block( approval: Approval::Unapproved, }, weight, - }); + }; + backend.write_block_entry(block_entry.clone()); // 2. Update leaves if inherited viability is fine. if inherited_viability.is_none() { @@ -344,26 +349,25 @@ fn add_block( stagnant_at_list.push(block_hash); backend.write_stagnant_at(stagnant_at, stagnant_at_list); - Ok(()) + Ok(block_entry) } /// Assuming that a block is already imported, accepts the number of the block /// as well as a list of reversions triggered by the block in ascending order. -fn apply_ancestor_reversions( +fn apply_reversions( backend: &mut OverlayedBackend, - block_hash: Hash, - block_number: BlockNumber, + block_entry: BlockEntry, reversions: Vec, ) -> Result<(), Error> { // Note: since revert numbers are in ascending order, the expensive propagation // of unviability is only heavy on the first log. for revert_number in reversions { - let maybe_block_entry = load_ancestor(backend, block_hash, block_number, revert_number)?; - if let Some(block_entry) = &maybe_block_entry { + let maybe_block_entry = load_ancestor(backend, &block_entry, revert_number)?; + if let Some(entry) = &maybe_block_entry { gum::trace!( target: LOG_TARGET, ?revert_number, - revert_hash = ?block_entry.block_hash, + revert_hash = ?entry.block_hash, "Block marked as reverted via scraped on-chain reversions" ); } @@ -372,8 +376,8 @@ fn apply_ancestor_reversions( maybe_block_entry, None, revert_number, - Some(block_hash), - Some(block_number), + Some(block_entry.block_hash), + Some(block_entry.block_number), )?; } From 9f12d2196e156e8822e5373975644aacfc266d14 Mon Sep 17 00:00:00 2001 From: Alexander Samusev <41779041+alvicsam@users.noreply.github.com> Date: Thu, 18 Apr 2024 17:57:22 +0200 Subject: [PATCH 3/7] [ci] Use native git cli in cargo (#4200) More details can be found [here](https://github.com/paritytech/ci_cd/issues/939#issuecomment-2064061845) --- .github/workflows/test-github-actions.yml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.github/workflows/test-github-actions.yml b/.github/workflows/test-github-actions.yml index 09cb4a25b9a30..c8ce49cb462b0 100644 --- a/.github/workflows/test-github-actions.yml +++ b/.github/workflows/test-github-actions.yml @@ -8,6 +8,9 @@ concurrency: group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} cancel-in-progress: true +env: + CARGO_NET_GIT_FETCH_WITH_CLI: true + jobs: test-linux-stable-int: runs-on: arc-runners-polkadot-sdk From 0e552893d0f656f83d366ae9118aaeb0f898aabf Mon Sep 17 00:00:00 2001 From: Alexandru Gheorghe <49718502+alexggh@users.noreply.github.com> Date: Thu, 18 Apr 2024 18:57:23 +0300 Subject: [PATCH 4/7] Fix next_retry busy waiting on first retry (#4192) The `next_retry_time` gets populated when a request receives an error timeout or any other error, after thatn next_retry would check all requests in the queue returns the smallest one, which then gets used to move the main loop by creating a Delay ``` futures_timer::Delay::new(instant.saturating_duration_since(Instant::now())).await, ``` However when we retry a task for the first time we still keep it in the queue an mark it as in flight so its next_retry_time would be the oldest and it would be small than `now`, so the Delay will always triggers, so that would make the main loop essentially busy wait untill we received a response for the retry request. Fix this by excluding the tasks that are already in-flight. --------- Signed-off-by: Alexandru Gheorghe Co-authored-by: Andrei Sandu <54316454+sandreim@users.noreply.github.com> --- .../statement-distribution/src/v2/requests.rs | 2 +- .../src/v2/tests/requests.rs | 26 ++++++++++++++++++- 2 files changed, 26 insertions(+), 2 deletions(-) diff --git a/polkadot/node/network/statement-distribution/src/v2/requests.rs b/polkadot/node/network/statement-distribution/src/v2/requests.rs index fe270c8a58e81..1ed18ffd42a9f 100644 --- a/polkadot/node/network/statement-distribution/src/v2/requests.rs +++ b/polkadot/node/network/statement-distribution/src/v2/requests.rs @@ -288,7 +288,7 @@ impl RequestManager { /// Returns an instant at which the next request to be retried will be ready. pub fn next_retry_time(&mut self) -> Option { let mut next = None; - for (_id, request) in &self.requests { + for (_id, request) in self.requests.iter().filter(|(_id, request)| !request.in_flight) { if let Some(next_retry_time) = request.next_retry_time { if next.map_or(true, |next| next_retry_time < next) { next = Some(next_retry_time); diff --git a/polkadot/node/network/statement-distribution/src/v2/tests/requests.rs b/polkadot/node/network/statement-distribution/src/v2/tests/requests.rs index dc2c8f55290b4..8cf139802148c 100644 --- a/polkadot/node/network/statement-distribution/src/v2/tests/requests.rs +++ b/polkadot/node/network/statement-distribution/src/v2/tests/requests.rs @@ -2606,7 +2606,31 @@ fn should_delay_before_retrying_dropped_requests() { // Sleep for the given amount of time. This should reset the delay for the first candidate. futures_timer::Delay::new(REQUEST_RETRY_DELAY).await; - // We re-try the first request. + // We re-try the first request the second time drop it again. + assert_matches!( + overseer.recv().await, + AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendRequests(mut requests, IfDisconnected::ImmediateError)) => { + assert_eq!(requests.len(), 1); + assert_matches!( + requests.pop().unwrap(), + Requests::AttestedCandidateV2(outgoing) => { + assert_eq!(outgoing.peer, Recipient::Peer(peer_c)); + assert_eq!(outgoing.payload.candidate_hash, candidate_hash_1); + assert_eq!(outgoing.payload.mask, mask); + } + ); + } + ); + + assert_matches!( + overseer_recv_with_timeout(&mut overseer, Duration::from_millis(100)).await, + None + ); + + // Sleep for the given amount of time. This should reset the delay for the first candidate. + futures_timer::Delay::new(REQUEST_RETRY_DELAY).await; + + // We re-try the first request, for the third time, so let's answer to it. { let statements = vec![ state From 37e338f0469c2ca5b716c4423d8b683e237ead21 Mon Sep 17 00:00:00 2001 From: Alexandru Gheorghe <49718502+alexggh@users.noreply.github.com> Date: Thu, 18 Apr 2024 18:57:34 +0300 Subject: [PATCH 5/7] approval-voting: Make sure we always mark approved candidates approved in a different relay chain context (#4153) ... see for more detail why this is needed https://github.com/paritytech/polkadot-sdk/issues/4149#issuecomment-2058472444 ## TODO: - [x] Unittests - [x] Replicate scenario from https://github.com/paritytech/polkadot-sdk/issues/4149 and confirm this fixes it: https://github.com/paritytech/polkadot-sdk/issues/4149 [ Replicated on a zombienet with some hacked nodes, that we can end up in this state where no-wake is schedule and the nodes are pending new assignments] --------- Signed-off-by: Alexandru Gheorghe Co-authored-by: Andrei Sandu <54316454+sandreim@users.noreply.github.com> --- polkadot/node/core/approval-voting/src/lib.rs | 51 +++++ .../node/core/approval-voting/src/tests.rs | 182 +++++++++++++++++- 2 files changed, 232 insertions(+), 1 deletion(-) diff --git a/polkadot/node/core/approval-voting/src/lib.rs b/polkadot/node/core/approval-voting/src/lib.rs index 7ecc2b2595bce..b5ed92fa39c87 100644 --- a/polkadot/node/core/approval-voting/src/lib.rs +++ b/polkadot/node/core/approval-voting/src/lib.rs @@ -978,6 +978,7 @@ where woken_block, woken_candidate, &subsystem.metrics, + &wakeups, ).await? } next_msg = ctx.recv().fuse() => { @@ -1152,6 +1153,7 @@ async fn handle_actions( candidate_hash, delayed_approvals_timers, approval_request, + &wakeups, ) .await? .into_iter() @@ -1663,6 +1665,7 @@ async fn handle_from_overseer( |r| { let _ = res.send(r); }, + &wakeups, ) .await? .0, @@ -2477,6 +2480,7 @@ async fn check_and_import_approval( metrics: &Metrics, approval: IndirectSignedApprovalVoteV2, with_response: impl FnOnce(ApprovalCheckResult) -> T, + wakeups: &Wakeups, ) -> SubsystemResult<(Vec, T)> where Sender: SubsystemSender, @@ -2655,6 +2659,7 @@ where approved_candidate_hash, candidate_entry, ApprovalStateTransition::RemoteApproval(approval.validator), + wakeups, ) .await; actions.extend(new_actions); @@ -2689,6 +2694,10 @@ impl ApprovalStateTransition { ApprovalStateTransition::WakeupProcessed => false, } } + + fn is_remote_approval(&self) -> bool { + matches!(*self, ApprovalStateTransition::RemoteApproval(_)) + } } // Advance the approval state, either by importing an approval vote which is already checked to be @@ -2705,6 +2714,7 @@ async fn advance_approval_state( candidate_hash: CandidateHash, mut candidate_entry: CandidateEntry, transition: ApprovalStateTransition, + wakeups: &Wakeups, ) -> Vec where Sender: SubsystemSender, @@ -2835,6 +2845,43 @@ where status.required_tranches, )); + if is_approved && transition.is_remote_approval() { + // Make sure we wake other blocks in case they have + // a no-show that might be covered by this approval. + for (fork_block_hash, fork_approval_entry) in candidate_entry + .block_assignments + .iter() + .filter(|(hash, _)| **hash != block_hash) + { + let assigned_on_fork_block = validator_index + .as_ref() + .map(|validator_index| fork_approval_entry.is_assigned(*validator_index)) + .unwrap_or_default(); + if wakeups.wakeup_for(*fork_block_hash, candidate_hash).is_none() && + !fork_approval_entry.is_approved() && + assigned_on_fork_block + { + let fork_block_entry = db.load_block_entry(fork_block_hash); + if let Ok(Some(fork_block_entry)) = fork_block_entry { + actions.push(Action::ScheduleWakeup { + block_hash: *fork_block_hash, + block_number: fork_block_entry.block_number(), + candidate_hash, + // Schedule the wakeup next tick, since the assignment must be a + // no-show, because there is no-wakeup scheduled. + tick: tick_now + 1, + }) + } else { + gum::debug!( + target: LOG_TARGET, + ?fork_block_entry, + ?fork_block_hash, + "Failed to load block entry" + ) + } + } + } + } // We have no need to write the candidate entry if all of the following // is true: // @@ -2896,6 +2943,7 @@ async fn process_wakeup( relay_block: Hash, candidate_hash: CandidateHash, metrics: &Metrics, + wakeups: &Wakeups, ) -> SubsystemResult> { let mut span = state .spans @@ -3064,6 +3112,7 @@ async fn process_wakeup( candidate_hash, candidate_entry, ApprovalStateTransition::WakeupProcessed, + wakeups, ) .await, ); @@ -3294,6 +3343,7 @@ async fn issue_approval( candidate_hash: CandidateHash, delayed_approvals_timers: &mut DelayedApprovalTimer, ApprovalVoteRequest { validator_index, block_hash }: ApprovalVoteRequest, + wakeups: &Wakeups, ) -> SubsystemResult> { let mut issue_approval_span = state .spans @@ -3415,6 +3465,7 @@ async fn issue_approval( candidate_hash, candidate_entry, ApprovalStateTransition::LocalApproval(validator_index as _), + wakeups, ) .await; diff --git a/polkadot/node/core/approval-voting/src/tests.rs b/polkadot/node/core/approval-voting/src/tests.rs index f7bbbca4b8a1c..312d805bbefb7 100644 --- a/polkadot/node/core/approval-voting/src/tests.rs +++ b/polkadot/node/core/approval-voting/src/tests.rs @@ -834,7 +834,6 @@ impl ChainBuilder { cur_hash = cur_header.parent_hash; } ancestry.reverse(); - import_block( overseer, ancestry.as_ref(), @@ -1922,6 +1921,187 @@ fn subsystem_assignment_import_updates_candidate_entry_and_schedules_wakeup() { }); } +#[test] +fn subsystem_always_has_a_wakeup_when_pending() { + // Approvals sent after all assignments are no-show, the approval + // should be counted on the fork relay chain on the next tick. + test_approvals_on_fork_are_always_considered_after_no_show( + 30, + vec![(29, false), (30, false), (31, true)], + ); + // Approvals sent before fork no-shows, the approval + // should be counted on the fork relay chain when it no-shows. + test_approvals_on_fork_are_always_considered_after_no_show( + 8, // a tick smaller than the no-show tick which is 30. + vec![(7, false), (8, false), (29, false), (30, true), (31, true)], + ); +} + +fn test_approvals_on_fork_are_always_considered_after_no_show( + tick_to_send_approval: Tick, + expected_approval_status: Vec<(Tick, bool)>, +) { + let config = HarnessConfig::default(); + let store = config.backend(); + + test_harness(config, |test_harness| async move { + let TestHarness { + mut virtual_overseer, + clock, + sync_oracle_handle: _sync_oracle_handle, + .. + } = test_harness; + + assert_matches!( + overseer_recv(&mut virtual_overseer).await, + AllMessages::ChainApi(ChainApiMessage::FinalizedBlockNumber(rx)) => { + rx.send(Ok(0)).unwrap(); + } + ); + let candidate_hash = Hash::repeat_byte(0x04); + + let candidate_descriptor = make_candidate(ParaId::from(1_u32), &candidate_hash); + let candidate_hash = candidate_descriptor.hash(); + + let block_hash = Hash::repeat_byte(0x01); + let block_hash_fork = Hash::repeat_byte(0x02); + + let candidate_index = 0; + let validator = ValidatorIndex(0); + let validators = vec![ + Sr25519Keyring::Alice, + Sr25519Keyring::Bob, + Sr25519Keyring::Charlie, + Sr25519Keyring::Dave, + Sr25519Keyring::Eve, + ]; + // Add block hash 0x01 and for 0x02 + ChainBuilder::new() + .add_block( + block_hash, + ChainBuilder::GENESIS_HASH, + 1, + BlockConfig { + slot: Slot::from(1), + candidates: Some(vec![( + candidate_descriptor.clone(), + CoreIndex(0), + GroupIndex(0), + )]), + session_info: Some(SessionInfo { + validator_groups: IndexedVec::>::from( + vec![ + vec![ValidatorIndex(0), ValidatorIndex(1)], + vec![ValidatorIndex(2)], + vec![ValidatorIndex(3), ValidatorIndex(4)], + ], + ), + needed_approvals: 1, + ..session_info(&validators) + }), + end_syncing: false, + }, + ) + .add_block( + block_hash_fork, + ChainBuilder::GENESIS_HASH, + 1, + BlockConfig { + slot: Slot::from(1), + candidates: Some(vec![(candidate_descriptor, CoreIndex(0), GroupIndex(0))]), + session_info: Some(SessionInfo { + validator_groups: IndexedVec::>::from( + vec![ + vec![ValidatorIndex(0), ValidatorIndex(1)], + vec![ValidatorIndex(2)], + vec![ValidatorIndex(3), ValidatorIndex(4)], + ], + ), + needed_approvals: 1, + ..session_info(&validators) + }), + end_syncing: false, + }, + ) + .build(&mut virtual_overseer) + .await; + + // Send assignments for the same candidate on both forks + let rx = check_and_import_assignment( + &mut virtual_overseer, + block_hash, + candidate_index, + validator, + ) + .await; + assert_eq!(rx.await, Ok(AssignmentCheckResult::Accepted)); + + let rx = check_and_import_assignment( + &mut virtual_overseer, + block_hash_fork, + candidate_index, + validator, + ) + .await; + + assert_eq!(rx.await, Ok(AssignmentCheckResult::Accepted)); + // Wake on APPROVAL_DELAY first + assert!(clock.inner.lock().current_wakeup_is(2)); + clock.inner.lock().set_tick(2); + futures_timer::Delay::new(Duration::from_millis(100)).await; + + // Wake up on no-show + assert!(clock.inner.lock().current_wakeup_is(30)); + + for (tick, status) in expected_approval_status + .iter() + .filter(|(tick, _)| *tick < tick_to_send_approval) + { + // Wake up on no-show + clock.inner.lock().set_tick(*tick); + futures_timer::Delay::new(Duration::from_millis(100)).await; + let block_entry = store.load_block_entry(&block_hash).unwrap().unwrap(); + let block_entry_fork = store.load_block_entry(&block_hash_fork).unwrap().unwrap(); + assert!(!block_entry.is_fully_approved()); + assert_eq!(block_entry_fork.is_fully_approved(), *status); + } + + clock.inner.lock().set_tick(tick_to_send_approval); + futures_timer::Delay::new(Duration::from_millis(100)).await; + + // Send the approval for candidate just in the context of 0x01 block. + let rx = check_and_import_approval( + &mut virtual_overseer, + block_hash, + candidate_index, + validator, + candidate_hash, + 1, + false, + None, + ) + .await; + + assert_eq!(rx.await, Ok(ApprovalCheckResult::Accepted),); + + // Check approval status for the fork_block is correctly transitioned. + for (tick, status) in expected_approval_status + .iter() + .filter(|(tick, _)| *tick >= tick_to_send_approval) + { + // Wake up on no-show + clock.inner.lock().set_tick(*tick); + futures_timer::Delay::new(Duration::from_millis(100)).await; + let block_entry = store.load_block_entry(&block_hash).unwrap().unwrap(); + let block_entry_fork = store.load_block_entry(&block_hash_fork).unwrap().unwrap(); + assert!(block_entry.is_fully_approved()); + assert_eq!(block_entry_fork.is_fully_approved(), *status); + } + + virtual_overseer + }); +} + #[test] fn subsystem_process_wakeup_schedules_wakeup() { test_harness(HarnessConfig::default(), |test_harness| async move { From c891fdabf4d519b25829490723fb70b1a2ffc0e5 Mon Sep 17 00:00:00 2001 From: Alexandru Vasile <60601340+lexnv@users.noreply.github.com> Date: Thu, 18 Apr 2024 18:57:44 +0300 Subject: [PATCH 6/7] tx: Remove tx_broadcast transaction from the pool (#4050) This PR ensures that broadcast future cleans-up the submitted extrinsic from the pool, iff the `broadcast_stop` operation has been called. This effectively cleans-up transactions from the pool when the `broadcast_stop` is called. cc @paritytech/subxt-team --------- Signed-off-by: Alexandru Vasile --- .../tests/transaction_broadcast_tests.rs | 9 ++--- .../src/transaction/transaction_broadcast.rs | 40 +++++++++++++------ 2 files changed, 31 insertions(+), 18 deletions(-) diff --git a/substrate/client/rpc-spec-v2/src/transaction/tests/transaction_broadcast_tests.rs b/substrate/client/rpc-spec-v2/src/transaction/tests/transaction_broadcast_tests.rs index 77a28968aedf8..14e188b6a8730 100644 --- a/substrate/client/rpc-spec-v2/src/transaction/tests/transaction_broadcast_tests.rs +++ b/substrate/client/rpc-spec-v2/src/transaction/tests/transaction_broadcast_tests.rs @@ -94,7 +94,7 @@ async fn tx_broadcast_enters_pool() { #[tokio::test] async fn tx_broadcast_invalid_tx() { - let (_, pool, _, tx_api, mut exec_middleware, _) = setup_api(Default::default()); + let (_, pool, _, tx_api, exec_middleware, _) = setup_api(Default::default()); // Invalid parameters. let err = tx_api @@ -114,13 +114,10 @@ async fn tx_broadcast_invalid_tx() { assert_eq!(0, pool.status().ready); - // Await the broadcast future to exit. - // Without this we'd be subject to races, where we try to call the stop before the tx is - // dropped. - let _ = get_next_event!(&mut exec_middleware.recv); + // The broadcast future should never be spawned when the tx decoding fails. assert_eq!(0, exec_middleware.num_tasks()); - // The broadcast future was dropped, and the operation is no longer active. + // The operation ID is no longer active. // When the operation is not active, either from the tx being finalized or a // terminal error; the stop method should return an error. let err = tx_api diff --git a/substrate/client/rpc-spec-v2/src/transaction/transaction_broadcast.rs b/substrate/client/rpc-spec-v2/src/transaction/transaction_broadcast.rs index 6eaf50d6b2e28..ef1a426865d5d 100644 --- a/substrate/client/rpc-spec-v2/src/transaction/transaction_broadcast.rs +++ b/substrate/client/rpc-spec-v2/src/transaction/transaction_broadcast.rs @@ -37,7 +37,7 @@ use std::{collections::HashMap, sync::Arc}; use super::error::ErrorBroadcast; /// An API for transaction RPC calls. -pub struct TransactionBroadcast { +pub struct TransactionBroadcast { /// Substrate client. client: Arc, /// Transactions pool. @@ -45,16 +45,18 @@ pub struct TransactionBroadcast { /// Executor to spawn subscriptions. executor: SubscriptionTaskExecutor, /// The broadcast operation IDs. - broadcast_ids: Arc>>, + broadcast_ids: Arc>>>, } /// The state of a broadcast operation. -struct BroadcastState { +struct BroadcastState { /// Handle to abort the running future that broadcasts the transaction. handle: AbortHandle, + /// Associated tx hash. + tx_hash: ::Hash, } -impl TransactionBroadcast { +impl TransactionBroadcast { /// Creates a new [`TransactionBroadcast`]. pub fn new(client: Arc, pool: Arc, executor: SubscriptionTaskExecutor) -> Self { TransactionBroadcast { client, pool, executor, broadcast_ids: Default::default() } @@ -106,17 +108,22 @@ where // The unique ID of this operation. let id = self.generate_unique_id(); + // The JSON-RPC server might check whether the transaction is valid before broadcasting it. + // If it does so and if the transaction is invalid, the server should silently do nothing + // and the JSON-RPC client is not informed of the problem. Invalid transactions should still + // count towards the limit to the number of simultaneously broadcasted transactions. + let Ok(decoded_extrinsic) = TransactionFor::::decode(&mut &bytes[..]) else { + return Ok(Some(id)); + }; + // Save the tx hash to remove it later. + let tx_hash = pool.hash_of(&decoded_extrinsic); + let mut best_block_import_stream = Box::pin(self.client.import_notification_stream().filter_map( |notification| async move { notification.is_new_best.then_some(notification.hash) }, )); let broadcast_transaction_fut = async move { - // There is nothing we could do with an extrinsic of invalid format. - let Ok(decoded_extrinsic) = TransactionFor::::decode(&mut &bytes[..]) else { - return; - }; - // Flag to determine if the we should broadcast the transaction again. let mut is_done = false; @@ -169,17 +176,26 @@ where let (fut, handle) = futures::future::abortable(broadcast_transaction_fut); let broadcast_ids = self.broadcast_ids.clone(); let drop_id = id.clone(); + let pool = self.pool.clone(); // The future expected by the executor must be `Future` instead of // `Future>`. - let fut = fut.map(move |_| { + let fut = fut.map(move |result| { // Remove the entry from the broadcast IDs map. - broadcast_ids.write().remove(&drop_id); + let Some(broadcast_state) = broadcast_ids.write().remove(&drop_id) else { return }; + + // The broadcast was not stopped. + if result.is_ok() { + return + } + + // Best effort pool removal (tx can already be finalized). + pool.remove_invalid(&[broadcast_state.tx_hash]); }); // Keep track of this entry and the abortable handle. { let mut broadcast_ids = self.broadcast_ids.write(); - broadcast_ids.insert(id.clone(), BroadcastState { handle }); + broadcast_ids.insert(id.clone(), BroadcastState { handle, tx_hash }); } sc_rpc::utils::spawn_subscription_task(&self.executor, fut); From 88a2f360238787bf5256cfdd14b40c08f519b38e Mon Sep 17 00:00:00 2001 From: Alexandru Vasile <60601340+lexnv@users.noreply.github.com> Date: Thu, 18 Apr 2024 20:19:04 +0300 Subject: [PATCH 7/7] chainHead: Stabilize chainHead to version 1 (#4168) This PR stabilizes the chainHead API to version 1. Needs: - https://github.com/paritytech/polkadot-sdk/pull/3667 cc @paritytech/subxt-team --------- Signed-off-by: Alexandru Vasile --- prdoc/pr_4168.prdoc | 8 + .../client/rpc-spec-v2/src/chain_head/api.rs | 26 +- .../src/chain_head/subscription/inner.rs | 2 +- .../rpc-spec-v2/src/chain_head/tests.rs | 225 ++++++++---------- 4 files changed, 118 insertions(+), 143 deletions(-) create mode 100644 prdoc/pr_4168.prdoc diff --git a/prdoc/pr_4168.prdoc b/prdoc/pr_4168.prdoc new file mode 100644 index 0000000000000..9a498500f08b4 --- /dev/null +++ b/prdoc/pr_4168.prdoc @@ -0,0 +1,8 @@ +title: Stabilize chianHead RPC class to version 1 + +doc: + - audience: Node Dev + description: | + The chainHead RPC API is stabilized to version 1. + +crates: [ ] diff --git a/substrate/client/rpc-spec-v2/src/chain_head/api.rs b/substrate/client/rpc-spec-v2/src/chain_head/api.rs index 3851adac2644d..23cb0bbf54585 100644 --- a/substrate/client/rpc-spec-v2/src/chain_head/api.rs +++ b/substrate/client/rpc-spec-v2/src/chain_head/api.rs @@ -37,15 +37,15 @@ pub trait ChainHeadApi { /// /// This method is unstable and subject to change in the future. #[subscription( - name = "chainHead_unstable_follow" => "chainHead_unstable_followEvent", - unsubscribe = "chainHead_unstable_unfollow", + name = "chainHead_v1_follow" => "chainHead_v1_followEvent", + unsubscribe = "chainHead_v1_unfollow", item = FollowEvent, )] fn chain_head_unstable_follow(&self, with_runtime: bool); /// Retrieves the body (list of transactions) of a pinned block. /// - /// This method should be seen as a complement to `chainHead_unstable_follow`, + /// This method should be seen as a complement to `chainHead_v1_follow`, /// allowing the JSON-RPC client to retrieve more information about a block /// that has been reported. /// @@ -54,7 +54,7 @@ pub trait ChainHeadApi { /// # Unstable /// /// This method is unstable and subject to change in the future. - #[method(name = "chainHead_unstable_body", raw_method)] + #[method(name = "chainHead_v1_body", raw_method)] async fn chain_head_unstable_body( &self, follow_subscription: String, @@ -63,7 +63,7 @@ pub trait ChainHeadApi { /// Retrieves the header of a pinned block. /// - /// This method should be seen as a complement to `chainHead_unstable_follow`, + /// This method should be seen as a complement to `chainHead_v1_follow`, /// allowing the JSON-RPC client to retrieve more information about a block /// that has been reported. /// @@ -73,7 +73,7 @@ pub trait ChainHeadApi { /// # Unstable /// /// This method is unstable and subject to change in the future. - #[method(name = "chainHead_unstable_header", raw_method)] + #[method(name = "chainHead_v1_header", raw_method)] async fn chain_head_unstable_header( &self, follow_subscription: String, @@ -85,7 +85,7 @@ pub trait ChainHeadApi { /// # Unstable /// /// This method is unstable and subject to change in the future. - #[method(name = "chainHead_unstable_storage", raw_method)] + #[method(name = "chainHead_v1_storage", raw_method)] async fn chain_head_unstable_storage( &self, follow_subscription: String, @@ -99,7 +99,7 @@ pub trait ChainHeadApi { /// # Unstable /// /// This method is unstable and subject to change in the future. - #[method(name = "chainHead_unstable_call", raw_method)] + #[method(name = "chainHead_v1_call", raw_method)] async fn chain_head_unstable_call( &self, follow_subscription: String, @@ -118,7 +118,7 @@ pub trait ChainHeadApi { /// # Unstable /// /// This method is unstable and subject to change in the future. - #[method(name = "chainHead_unstable_unpin", raw_method)] + #[method(name = "chainHead_v1_unpin", raw_method)] async fn chain_head_unstable_unpin( &self, follow_subscription: String, @@ -131,21 +131,21 @@ pub trait ChainHeadApi { /// # Unstable /// /// This method is unstable and subject to change in the future. - #[method(name = "chainHead_unstable_continue", raw_method)] + #[method(name = "chainHead_v1_continue", raw_method)] async fn chain_head_unstable_continue( &self, follow_subscription: String, operation_id: String, ) -> Result<(), Error>; - /// Stops an operation started with chainHead_unstable_body, chainHead_unstable_call, or - /// chainHead_unstable_storage. If the operation was still in progress, this interrupts it. If + /// Stops an operation started with chainHead_v1_body, chainHead_v1_call, or + /// chainHead_v1_storage. If the operation was still in progress, this interrupts it. If /// the operation was already finished, this call has no effect. /// /// # Unstable /// /// This method is unstable and subject to change in the future. - #[method(name = "chainHead_unstable_stopOperation", raw_method)] + #[method(name = "chainHead_v1_stopOperation", raw_method)] async fn chain_head_unstable_stop_operation( &self, follow_subscription: String, diff --git a/substrate/client/rpc-spec-v2/src/chain_head/subscription/inner.rs b/substrate/client/rpc-spec-v2/src/chain_head/subscription/inner.rs index 0e5ccb91d39a6..3495d9e54490c 100644 --- a/substrate/client/rpc-spec-v2/src/chain_head/subscription/inner.rs +++ b/substrate/client/rpc-spec-v2/src/chain_head/subscription/inner.rs @@ -186,7 +186,7 @@ impl OperationState { /// Stops the operation if `waitingForContinue` event was emitted for the associated /// operation ID. /// - /// Returns nothing in accordance with `chainHead_unstable_stopOperation`. + /// Returns nothing in accordance with `chainHead_v1_stopOperation`. pub fn stop_operation(&self) { // `waitingForContinue` not generated. if !self.shared_state.requested_continue.load(std::sync::atomic::Ordering::Acquire) { diff --git a/substrate/client/rpc-spec-v2/src/chain_head/tests.rs b/substrate/client/rpc-spec-v2/src/chain_head/tests.rs index 14f664858a0d8..4bab2194e082a 100644 --- a/substrate/client/rpc-spec-v2/src/chain_head/tests.rs +++ b/substrate/client/rpc-spec-v2/src/chain_head/tests.rs @@ -156,7 +156,7 @@ async fn setup_api() -> ( ) .into_rpc(); - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [true]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [true]).await.unwrap(); let sub_id = sub.subscription_id(); let sub_id = serde_json::to_string(&sub_id).unwrap(); @@ -266,7 +266,7 @@ async fn follow_subscription_produces_blocks() { .into_rpc(); let finalized_hash = client.info().finalized_hash; - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [false]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [false]).await.unwrap(); // Initialized must always be reported first. let event: FollowEvent = get_next_event(&mut sub).await; @@ -337,7 +337,7 @@ async fn follow_with_runtime() { .into_rpc(); let finalized_hash = client.info().finalized_hash; - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [true]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [true]).await.unwrap(); // Initialized must always be reported first. let event: FollowEvent = get_next_event(&mut sub).await; @@ -448,14 +448,14 @@ async fn get_header() { // Invalid subscription ID must produce no results. let res: Option = api - .call("chainHead_unstable_header", ["invalid_sub_id", &invalid_hash]) + .call("chainHead_v1_header", ["invalid_sub_id", &invalid_hash]) .await .unwrap(); assert!(res.is_none()); // Valid subscription with invalid block hash will error. let err = api - .call::<_, serde_json::Value>("chainHead_unstable_header", [&sub_id, &invalid_hash]) + .call::<_, serde_json::Value>("chainHead_v1_header", [&sub_id, &invalid_hash]) .await .unwrap_err(); assert_matches!(err, @@ -463,7 +463,7 @@ async fn get_header() { ); // Obtain the valid header. - let res: String = api.call("chainHead_unstable_header", [&sub_id, &block_hash]).await.unwrap(); + let res: String = api.call("chainHead_v1_header", [&sub_id, &block_hash]).await.unwrap(); let bytes = array_bytes::hex2bytes(&res).unwrap(); let header: Header = Decode::decode(&mut &bytes[..]).unwrap(); assert_eq!(header, block.header); @@ -476,15 +476,13 @@ async fn get_body() { let invalid_hash = hex_string(&INVALID_HASH); // Subscription ID is invalid. - let response: MethodResponse = api - .call("chainHead_unstable_body", ["invalid_sub_id", &invalid_hash]) - .await - .unwrap(); + let response: MethodResponse = + api.call("chainHead_v1_body", ["invalid_sub_id", &invalid_hash]).await.unwrap(); assert_matches!(response, MethodResponse::LimitReached); // Block hash is invalid. let err = api - .call::<_, serde_json::Value>("chainHead_unstable_body", [&sub_id, &invalid_hash]) + .call::<_, serde_json::Value>("chainHead_v1_body", [&sub_id, &invalid_hash]) .await .unwrap_err(); assert_matches!(err, @@ -493,7 +491,7 @@ async fn get_body() { // Valid call. let response: MethodResponse = - api.call("chainHead_unstable_body", [&sub_id, &block_hash]).await.unwrap(); + api.call("chainHead_v1_body", [&sub_id, &block_hash]).await.unwrap(); let operation_id = match response { MethodResponse::Started(started) => started.operation_id, MethodResponse::LimitReached => panic!("Expected started response"), @@ -534,7 +532,7 @@ async fn get_body() { // Valid call to a block with extrinsics. let response: MethodResponse = - api.call("chainHead_unstable_body", [&sub_id, &block_hash]).await.unwrap(); + api.call("chainHead_v1_body", [&sub_id, &block_hash]).await.unwrap(); let operation_id = match response { MethodResponse::Started(started) => started.operation_id, MethodResponse::LimitReached => panic!("Expected started response"), @@ -556,10 +554,7 @@ async fn call_runtime() { // Subscription ID is invalid. let response: MethodResponse = api - .call( - "chainHead_unstable_call", - ["invalid_sub_id", &block_hash, "BabeApi_current_epoch", "0x00"], - ) + .call("chainHead_v1_call", ["invalid_sub_id", &block_hash, "BabeApi_current_epoch", "0x00"]) .await .unwrap(); assert_matches!(response, MethodResponse::LimitReached); @@ -567,7 +562,7 @@ async fn call_runtime() { // Block hash is invalid. let err = api .call::<_, serde_json::Value>( - "chainHead_unstable_call", + "chainHead_v1_call", [&sub_id, &invalid_hash, "BabeApi_current_epoch", "0x00"], ) .await @@ -579,7 +574,7 @@ async fn call_runtime() { // Pass an invalid parameters that cannot be decode. let err = api .call::<_, serde_json::Value>( - "chainHead_unstable_call", + "chainHead_v1_call", // 0x0 is invalid. [&sub_id, &block_hash, "BabeApi_current_epoch", "0x0"], ) @@ -595,7 +590,7 @@ async fn call_runtime() { let call_parameters = hex_string(&alice_id.encode()); let response: MethodResponse = api .call( - "chainHead_unstable_call", + "chainHead_v1_call", [&sub_id, &block_hash, "AccountNonceApi_account_nonce", &call_parameters], ) .await @@ -614,7 +609,7 @@ async fn call_runtime() { // The `current_epoch` takes no parameters and not draining the input buffer // will cause the execution to fail. let response: MethodResponse = api - .call("chainHead_unstable_call", [&sub_id, &block_hash, "BabeApi_current_epoch", "0x00"]) + .call("chainHead_v1_call", [&sub_id, &block_hash, "BabeApi_current_epoch", "0x00"]) .await .unwrap(); let operation_id = match response { @@ -651,7 +646,7 @@ async fn call_runtime_without_flag() { ) .into_rpc(); - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [false]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [false]).await.unwrap(); let sub_id = sub.subscription_id(); let sub_id = serde_json::to_string(&sub_id).unwrap(); @@ -685,7 +680,7 @@ async fn call_runtime_without_flag() { let call_parameters = hex_string(&alice_id.encode()); let err = api .call::<_, serde_json::Value>( - "chainHead_unstable_call", + "chainHead_v1_call", [&sub_id, &block_hash, "AccountNonceApi_account_nonce", &call_parameters], ) .await @@ -706,7 +701,7 @@ async fn get_storage_hash() { // Subscription ID is invalid. let response: MethodResponse = api .call( - "chainHead_unstable_storage", + "chainHead_v1_storage", rpc_params![ "invalid_sub_id", &invalid_hash, @@ -720,7 +715,7 @@ async fn get_storage_hash() { // Block hash is invalid. let err = api .call::<_, serde_json::Value>( - "chainHead_unstable_storage", + "chainHead_v1_storage", rpc_params![ &sub_id, &invalid_hash, @@ -736,7 +731,7 @@ async fn get_storage_hash() { // Valid call without storage at the key. let response: MethodResponse = api .call( - "chainHead_unstable_storage", + "chainHead_v1_storage", rpc_params![ &sub_id, &block_hash, @@ -779,7 +774,7 @@ async fn get_storage_hash() { // Valid call with storage at the key. let response: MethodResponse = api .call( - "chainHead_unstable_storage", + "chainHead_v1_storage", rpc_params![ &sub_id, &block_hash, @@ -812,7 +807,7 @@ async fn get_storage_hash() { // Valid call with storage at the key. let response: MethodResponse = api .call( - "chainHead_unstable_storage", + "chainHead_v1_storage", rpc_params![ &sub_id, &genesis_hash, @@ -869,7 +864,7 @@ async fn get_storage_multi_query_iter() { // Valid call with storage at the key. let response: MethodResponse = api .call( - "chainHead_unstable_storage", + "chainHead_v1_storage", rpc_params![ &sub_id, &block_hash, @@ -920,7 +915,7 @@ async fn get_storage_multi_query_iter() { let expected_value = hex_string(&CHILD_VALUE); let response: MethodResponse = api .call( - "chainHead_unstable_storage", + "chainHead_v1_storage", rpc_params![ &sub_id, &genesis_hash, @@ -974,7 +969,7 @@ async fn get_storage_value() { // Subscription ID is invalid. let response: MethodResponse = api .call( - "chainHead_unstable_storage", + "chainHead_v1_storage", rpc_params![ "invalid_sub_id", &invalid_hash, @@ -988,7 +983,7 @@ async fn get_storage_value() { // Block hash is invalid. let err = api .call::<_, serde_json::Value>( - "chainHead_unstable_storage", + "chainHead_v1_storage", rpc_params![ &sub_id, &invalid_hash, @@ -1004,7 +999,7 @@ async fn get_storage_value() { // Valid call without storage at the key. let response: MethodResponse = api .call( - "chainHead_unstable_storage", + "chainHead_v1_storage", rpc_params![ &sub_id, &block_hash, @@ -1047,7 +1042,7 @@ async fn get_storage_value() { // Valid call with storage at the key. let response: MethodResponse = api .call( - "chainHead_unstable_storage", + "chainHead_v1_storage", rpc_params![ &sub_id, &block_hash, @@ -1079,7 +1074,7 @@ async fn get_storage_value() { let response: MethodResponse = api .call( - "chainHead_unstable_storage", + "chainHead_v1_storage", rpc_params![ &sub_id, &genesis_hash, @@ -1121,7 +1116,7 @@ async fn get_storage_non_queryable_key() { let response: MethodResponse = api .call( - "chainHead_unstable_storage", + "chainHead_v1_storage", rpc_params![ &sub_id, &block_hash, @@ -1146,7 +1141,7 @@ async fn get_storage_non_queryable_key() { let prefixed_key = hex_string(&prefixed_key); let response: MethodResponse = api .call( - "chainHead_unstable_storage", + "chainHead_v1_storage", rpc_params![ &sub_id, &block_hash, @@ -1171,7 +1166,7 @@ async fn get_storage_non_queryable_key() { let prefixed_key = hex_string(&prefixed_key); let response: MethodResponse = api .call( - "chainHead_unstable_storage", + "chainHead_v1_storage", rpc_params![ &sub_id, &block_hash, @@ -1197,7 +1192,7 @@ async fn get_storage_non_queryable_key() { let prefixed_key = hex_string(&prefixed_key); let response: MethodResponse = api .call( - "chainHead_unstable_storage", + "chainHead_v1_storage", rpc_params![ &sub_id, &block_hash, @@ -1227,9 +1222,9 @@ async fn unique_operation_ids() { // Ensure that operation IDs are unique for multiple method calls. for _ in 0..5 { - // Valid `chainHead_unstable_body` call. + // Valid `chainHead_v1_body` call. let response: MethodResponse = - api.call("chainHead_unstable_body", [&sub_id, &block_hash]).await.unwrap(); + api.call("chainHead_v1_body", [&sub_id, &block_hash]).await.unwrap(); let operation_id = match response { MethodResponse::Started(started) => started.operation_id, MethodResponse::LimitReached => panic!("Expected started response"), @@ -1241,11 +1236,11 @@ async fn unique_operation_ids() { // Ensure uniqueness. assert!(op_ids.insert(operation_id)); - // Valid `chainHead_unstable_storage` call. + // Valid `chainHead_v1_storage` call. let key = hex_string(&KEY); let response: MethodResponse = api .call( - "chainHead_unstable_storage", + "chainHead_v1_storage", rpc_params![ &sub_id, &block_hash, @@ -1266,12 +1261,12 @@ async fn unique_operation_ids() { // Ensure uniqueness. assert!(op_ids.insert(operation_id)); - // Valid `chainHead_unstable_call` call. + // Valid `chainHead_v1_call` call. let alice_id = AccountKeyring::Alice.to_account_id(); let call_parameters = hex_string(&alice_id.encode()); let response: MethodResponse = api .call( - "chainHead_unstable_call", + "chainHead_v1_call", [&sub_id, &block_hash, "AccountNonceApi_account_nonce", &call_parameters], ) .await @@ -1313,12 +1308,11 @@ async fn separate_operation_ids_for_subscriptions() { .into_rpc(); // Create two separate subscriptions. - let mut sub_first = api.subscribe_unbounded("chainHead_unstable_follow", [true]).await.unwrap(); + let mut sub_first = api.subscribe_unbounded("chainHead_v1_follow", [true]).await.unwrap(); let sub_id_first = sub_first.subscription_id(); let sub_id_first = serde_json::to_string(&sub_id_first).unwrap(); - let mut sub_second = - api.subscribe_unbounded("chainHead_unstable_follow", [true]).await.unwrap(); + let mut sub_second = api.subscribe_unbounded("chainHead_v1_follow", [true]).await.unwrap(); let sub_id_second = sub_second.subscription_id(); let sub_id_second = serde_json::to_string(&sub_id_second).unwrap(); @@ -1362,17 +1356,15 @@ async fn separate_operation_ids_for_subscriptions() { // Each `chainHead_follow` subscription receives a separate operation ID. let response: MethodResponse = - api.call("chainHead_unstable_body", [&sub_id_first, &block_hash]).await.unwrap(); + api.call("chainHead_v1_body", [&sub_id_first, &block_hash]).await.unwrap(); let operation_id: String = match response { MethodResponse::Started(started) => started.operation_id, MethodResponse::LimitReached => panic!("Expected started response"), }; assert_eq!(operation_id, "0"); - let response: MethodResponse = api - .call("chainHead_unstable_body", [&sub_id_second, &block_hash]) - .await - .unwrap(); + let response: MethodResponse = + api.call("chainHead_v1_body", [&sub_id_second, &block_hash]).await.unwrap(); let operation_id_second: String = match response { MethodResponse::Started(started) => started.operation_id, MethodResponse::LimitReached => panic!("Expected started response"), @@ -1449,7 +1441,7 @@ async fn follow_generates_initial_blocks() { let block_2_f_hash = block_2_f.header.hash(); client.import(BlockOrigin::Own, block_2_f.clone()).await.unwrap(); - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [false]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [false]).await.unwrap(); // Initialized must always be reported first. let event: FollowEvent = get_next_event(&mut sub).await; @@ -1561,7 +1553,7 @@ async fn follow_exceeding_pinned_blocks() { ) .into_rpc(); - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [false]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [false]).await.unwrap(); let block = BlockBuilderBuilder::new(&*client) .on_parent_block(client.chain_info().genesis_hash) @@ -1640,7 +1632,7 @@ async fn follow_with_unpin() { ) .into_rpc(); - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [false]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [false]).await.unwrap(); let sub_id = sub.subscription_id(); let sub_id = serde_json::to_string(&sub_id).unwrap(); @@ -1672,17 +1664,14 @@ async fn follow_with_unpin() { // Unpin an invalid subscription ID must return Ok(()). let invalid_hash = hex_string(&INVALID_HASH); let _res: () = api - .call("chainHead_unstable_unpin", rpc_params!["invalid_sub_id", &invalid_hash]) + .call("chainHead_v1_unpin", rpc_params!["invalid_sub_id", &invalid_hash]) .await .unwrap(); // Valid subscription with invalid block hash. let invalid_hash = hex_string(&INVALID_HASH); let err = api - .call::<_, serde_json::Value>( - "chainHead_unstable_unpin", - rpc_params![&sub_id, &invalid_hash], - ) + .call::<_, serde_json::Value>("chainHead_v1_unpin", rpc_params![&sub_id, &invalid_hash]) .await .unwrap_err(); assert_matches!(err, @@ -1690,10 +1679,7 @@ async fn follow_with_unpin() { ); // To not exceed the number of pinned blocks, we need to unpin before the next import. - let _res: () = api - .call("chainHead_unstable_unpin", rpc_params![&sub_id, &block_hash]) - .await - .unwrap(); + let _res: () = api.call("chainHead_v1_unpin", rpc_params![&sub_id, &block_hash]).await.unwrap(); // Block tree: // finalized_block -> block -> block2 @@ -1754,7 +1740,7 @@ async fn unpin_duplicate_hashes() { ) .into_rpc(); - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [false]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [false]).await.unwrap(); let sub_id = sub.subscription_id(); let sub_id = serde_json::to_string(&sub_id).unwrap(); @@ -1786,7 +1772,7 @@ async fn unpin_duplicate_hashes() { // Try to unpin duplicate hashes. let err = api .call::<_, serde_json::Value>( - "chainHead_unstable_unpin", + "chainHead_v1_unpin", rpc_params![&sub_id, vec![&block_hash, &block_hash]], ) .await @@ -1821,7 +1807,7 @@ async fn unpin_duplicate_hashes() { // Try to unpin duplicate hashes. let err = api .call::<_, serde_json::Value>( - "chainHead_unstable_unpin", + "chainHead_v1_unpin", rpc_params![&sub_id, vec![&block_hash, &block_hash_2, &block_hash]], ) .await @@ -1832,7 +1818,7 @@ async fn unpin_duplicate_hashes() { // Can unpin blocks. let _res: () = api - .call("chainHead_unstable_unpin", rpc_params![&sub_id, vec![&block_hash, &block_hash_2]]) + .call("chainHead_v1_unpin", rpc_params![&sub_id, vec![&block_hash, &block_hash_2]]) .await .unwrap(); } @@ -1859,7 +1845,7 @@ async fn follow_with_multiple_unpin_hashes() { ) .into_rpc(); - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [false]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [false]).await.unwrap(); let sub_id = sub.subscription_id(); let sub_id = serde_json::to_string(&sub_id).unwrap(); @@ -1930,16 +1916,13 @@ async fn follow_with_multiple_unpin_hashes() { // Unpin an invalid subscription ID must return Ok(()). let invalid_hash = hex_string(&INVALID_HASH); let _res: () = api - .call("chainHead_unstable_unpin", rpc_params!["invalid_sub_id", &invalid_hash]) + .call("chainHead_v1_unpin", rpc_params!["invalid_sub_id", &invalid_hash]) .await .unwrap(); // Valid subscription with invalid block hash. let err = api - .call::<_, serde_json::Value>( - "chainHead_unstable_unpin", - rpc_params![&sub_id, &invalid_hash], - ) + .call::<_, serde_json::Value>("chainHead_v1_unpin", rpc_params![&sub_id, &invalid_hash]) .await .unwrap_err(); assert_matches!(err, @@ -1947,14 +1930,14 @@ async fn follow_with_multiple_unpin_hashes() { ); let _res: () = api - .call("chainHead_unstable_unpin", rpc_params![&sub_id, &block_1_hash]) + .call("chainHead_v1_unpin", rpc_params![&sub_id, &block_1_hash]) .await .unwrap(); // One block hash is invalid. Block 1 is already unpinned. let err = api .call::<_, serde_json::Value>( - "chainHead_unstable_unpin", + "chainHead_v1_unpin", rpc_params![&sub_id, vec![&block_1_hash, &block_2_hash, &block_3_hash]], ) .await @@ -1965,16 +1948,13 @@ async fn follow_with_multiple_unpin_hashes() { // Unpin multiple blocks. let _res: () = api - .call("chainHead_unstable_unpin", rpc_params![&sub_id, vec![&block_2_hash, &block_3_hash]]) + .call("chainHead_v1_unpin", rpc_params![&sub_id, vec![&block_2_hash, &block_3_hash]]) .await .unwrap(); // Check block 2 and 3 are unpinned. let err = api - .call::<_, serde_json::Value>( - "chainHead_unstable_unpin", - rpc_params![&sub_id, &block_2_hash], - ) + .call::<_, serde_json::Value>("chainHead_v1_unpin", rpc_params![&sub_id, &block_2_hash]) .await .unwrap_err(); assert_matches!(err, @@ -1982,10 +1962,7 @@ async fn follow_with_multiple_unpin_hashes() { ); let err = api - .call::<_, serde_json::Value>( - "chainHead_unstable_unpin", - rpc_params![&sub_id, &block_3_hash], - ) + .call::<_, serde_json::Value>("chainHead_v1_unpin", rpc_params![&sub_id, &block_3_hash]) .await .unwrap_err(); assert_matches!(err, @@ -2016,7 +1993,7 @@ async fn follow_prune_best_block() { .into_rpc(); let finalized_hash = client.info().finalized_hash; - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [false]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [false]).await.unwrap(); // Initialized must always be reported first. let event: FollowEvent = get_next_event(&mut sub).await; @@ -2178,7 +2155,7 @@ async fn follow_prune_best_block() { let sub_id = sub.subscription_id(); let sub_id = serde_json::to_string(&sub_id).unwrap(); let hash = format!("{:?}", block_2_hash); - let _res: () = api.call("chainHead_unstable_unpin", rpc_params![&sub_id, &hash]).await.unwrap(); + let _res: () = api.call("chainHead_v1_unpin", rpc_params![&sub_id, &hash]).await.unwrap(); } #[tokio::test] @@ -2282,7 +2259,7 @@ async fn follow_forks_pruned_block() { // Block 2_f and 3_f are not pruned, pruning happens at height (N - 1). client.finalize_block(block_3_hash, None).unwrap(); - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [false]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [false]).await.unwrap(); // Initialized must always be reported first. let event: FollowEvent = get_next_event(&mut sub).await; @@ -2444,7 +2421,7 @@ async fn follow_report_multiple_pruned_block() { let block_3_f = block_builder.build().unwrap().block; let block_3_f_hash = block_3_f.hash(); client.import(BlockOrigin::Own, block_3_f.clone()).await.unwrap(); - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [false]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [false]).await.unwrap(); // Initialized must always be reported first. let event: FollowEvent = get_next_event(&mut sub).await; @@ -2630,7 +2607,7 @@ async fn pin_block_references() { } } - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [false]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [false]).await.unwrap(); let sub_id = sub.subscription_id(); let sub_id = serde_json::to_string(&sub_id).unwrap(); @@ -2669,10 +2646,7 @@ async fn pin_block_references() { wait_pinned_references(&backend, &hash, 1).await; // To not exceed the number of pinned blocks, we need to unpin before the next import. - let _res: () = api - .call("chainHead_unstable_unpin", rpc_params![&sub_id, &block_hash]) - .await - .unwrap(); + let _res: () = api.call("chainHead_v1_unpin", rpc_params![&sub_id, &block_hash]).await.unwrap(); // Make sure unpin clears out the reference. let refs = backend.pin_refs(&hash).unwrap(); @@ -2765,7 +2739,7 @@ async fn follow_finalized_before_new_block() { let block_1_hash = block_1.header.hash(); client.import(BlockOrigin::Own, block_1.clone()).await.unwrap(); - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [false]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [false]).await.unwrap(); // Trigger the `FinalizedNotification` for block 1 before the `BlockImportNotification`, and // expect for the `chainHead` to generate `NewBlock`, `BestBlock` and `Finalized` events. @@ -2870,7 +2844,7 @@ async fn ensure_operation_limits_works() { ) .into_rpc(); - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [true]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [true]).await.unwrap(); let sub_id = sub.subscription_id(); let sub_id = serde_json::to_string(&sub_id).unwrap(); @@ -2909,7 +2883,7 @@ async fn ensure_operation_limits_works() { ]; let response: MethodResponse = api - .call("chainHead_unstable_storage", rpc_params![&sub_id, &block_hash, items]) + .call("chainHead_v1_storage", rpc_params![&sub_id, &block_hash, items]) .await .unwrap(); let operation_id = match response { @@ -2932,7 +2906,7 @@ async fn ensure_operation_limits_works() { let call_parameters = hex_string(&alice_id.encode()); let response: MethodResponse = api .call( - "chainHead_unstable_call", + "chainHead_v1_call", [&sub_id, &block_hash, "AccountNonceApi_account_nonce", &call_parameters], ) .await @@ -2977,7 +2951,7 @@ async fn check_continue_operation() { ) .into_rpc(); - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [true]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [true]).await.unwrap(); let sub_id = sub.subscription_id(); let sub_id = serde_json::to_string(&sub_id).unwrap(); @@ -3014,17 +2988,17 @@ async fn check_continue_operation() { // Invalid subscription ID must produce no results. let _res: () = api - .call("chainHead_unstable_continue", ["invalid_sub_id", &invalid_hash]) + .call("chainHead_v1_continue", ["invalid_sub_id", &invalid_hash]) .await .unwrap(); // Invalid operation ID must produce no results. - let _res: () = api.call("chainHead_unstable_continue", [&sub_id, &invalid_hash]).await.unwrap(); + let _res: () = api.call("chainHead_v1_continue", [&sub_id, &invalid_hash]).await.unwrap(); // Valid call with storage at the key. let response: MethodResponse = api .call( - "chainHead_unstable_storage", + "chainHead_v1_storage", rpc_params![ &sub_id, &block_hash, @@ -3060,7 +3034,7 @@ async fn check_continue_operation() { std::time::Duration::from_secs(DOES_NOT_PRODUCE_EVENTS_SECONDS), ) .await; - let _res: () = api.call("chainHead_unstable_continue", [&sub_id, &operation_id]).await.unwrap(); + let _res: () = api.call("chainHead_v1_continue", [&sub_id, &operation_id]).await.unwrap(); assert_matches!( get_next_event::>(&mut sub).await, FollowEvent::OperationStorageItems(res) if res.operation_id == operation_id && @@ -3079,7 +3053,7 @@ async fn check_continue_operation() { std::time::Duration::from_secs(DOES_NOT_PRODUCE_EVENTS_SECONDS), ) .await; - let _res: () = api.call("chainHead_unstable_continue", [&sub_id, &operation_id]).await.unwrap(); + let _res: () = api.call("chainHead_v1_continue", [&sub_id, &operation_id]).await.unwrap(); assert_matches!( get_next_event::>(&mut sub).await, FollowEvent::OperationStorageItems(res) if res.operation_id == operation_id && @@ -3099,7 +3073,7 @@ async fn check_continue_operation() { std::time::Duration::from_secs(DOES_NOT_PRODUCE_EVENTS_SECONDS), ) .await; - let _res: () = api.call("chainHead_unstable_continue", [&sub_id, &operation_id]).await.unwrap(); + let _res: () = api.call("chainHead_v1_continue", [&sub_id, &operation_id]).await.unwrap(); assert_matches!( get_next_event::>(&mut sub).await, FollowEvent::OperationStorageItems(res) if res.operation_id == operation_id && @@ -3118,7 +3092,7 @@ async fn check_continue_operation() { std::time::Duration::from_secs(DOES_NOT_PRODUCE_EVENTS_SECONDS), ) .await; - let _res: () = api.call("chainHead_unstable_continue", [&sub_id, &operation_id]).await.unwrap(); + let _res: () = api.call("chainHead_v1_continue", [&sub_id, &operation_id]).await.unwrap(); assert_matches!( get_next_event::>(&mut sub).await, FollowEvent::OperationStorageItems(res) if res.operation_id == operation_id && @@ -3162,7 +3136,7 @@ async fn stop_storage_operation() { ) .into_rpc(); - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [true]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [true]).await.unwrap(); let sub_id = sub.subscription_id(); let sub_id = serde_json::to_string(&sub_id).unwrap(); @@ -3196,20 +3170,17 @@ async fn stop_storage_operation() { // Invalid subscription ID must produce no results. let _res: () = api - .call("chainHead_unstable_stopOperation", ["invalid_sub_id", &invalid_hash]) + .call("chainHead_v1_stopOperation", ["invalid_sub_id", &invalid_hash]) .await .unwrap(); // Invalid operation ID must produce no results. - let _res: () = api - .call("chainHead_unstable_stopOperation", [&sub_id, &invalid_hash]) - .await - .unwrap(); + let _res: () = api.call("chainHead_v1_stopOperation", [&sub_id, &invalid_hash]).await.unwrap(); // Valid call with storage at the key. let response: MethodResponse = api .call( - "chainHead_unstable_storage", + "chainHead_v1_storage", rpc_params![ &sub_id, &block_hash, @@ -3241,10 +3212,7 @@ async fn stop_storage_operation() { ); // Stop the operation. - let _res: () = api - .call("chainHead_unstable_stopOperation", [&sub_id, &operation_id]) - .await - .unwrap(); + let _res: () = api.call("chainHead_v1_stopOperation", [&sub_id, &operation_id]).await.unwrap(); does_not_produce_event::>( &mut sub, @@ -3272,7 +3240,7 @@ async fn storage_closest_merkle_value() { // Valid call with storage at the keys. let response: MethodResponse = api .call( - "chainHead_unstable_storage", + "chainHead_v1_storage", rpc_params![ &sub_id, &block_hash, @@ -3466,7 +3434,7 @@ async fn chain_head_stop_all_subscriptions() { ) .into_rpc(); - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [true]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [true]).await.unwrap(); // Ensure the imported block is propagated and pinned for this subscription. assert_matches!( @@ -3500,8 +3468,7 @@ async fn chain_head_stop_all_subscriptions() { ); } - let mut second_sub = - api.subscribe_unbounded("chainHead_unstable_follow", [true]).await.unwrap(); + let mut second_sub = api.subscribe_unbounded("chainHead_v1_follow", [true]).await.unwrap(); // Lagging detected, the stop event is delivered immediately. assert_matches!( get_next_event::>(&mut second_sub).await, @@ -3512,14 +3479,14 @@ async fn chain_head_stop_all_subscriptions() { assert_matches!(get_next_event::>(&mut sub).await, FollowEvent::Stop); // Other subscriptions cannot be started until the suspension period is over. - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [true]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [true]).await.unwrap(); // Should receive the stop event immediately. assert_matches!(get_next_event::>(&mut sub).await, FollowEvent::Stop); // For the next subscription, lagging distance must be smaller. client.finalize_block(parent_hash, None).unwrap(); - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [true]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [true]).await.unwrap(); assert_matches!( get_next_event::>(&mut sub).await, FollowEvent::Initialized(_) @@ -3681,12 +3648,12 @@ async fn chain_head_limit_reached() { ) .into_rpc(); - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [true]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [true]).await.unwrap(); // Initialized must always be reported first. let _event: FollowEvent = get_next_event(&mut sub).await; - let error = api.subscribe_unbounded("chainHead_unstable_follow", [true]).await.unwrap_err(); + let error = api.subscribe_unbounded("chainHead_v1_follow", [true]).await.unwrap_err(); assert!(error .to_string() .contains("Maximum number of chainHead_follow has been reached")); @@ -3696,7 +3663,7 @@ async fn chain_head_limit_reached() { // Ensure the `chainHead_unfollow` is propagated to the server. tokio::time::sleep(std::time::Duration::from_secs(5)).await; - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [true]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [true]).await.unwrap(); // Initialized must always be reported first. let _event: FollowEvent = get_next_event(&mut sub).await; } @@ -3723,7 +3690,7 @@ async fn follow_unique_pruned_blocks() { .into_rpc(); let finalized_hash = client.info().finalized_hash; - let mut sub = api.subscribe_unbounded("chainHead_unstable_follow", [false]).await.unwrap(); + let mut sub = api.subscribe_unbounded("chainHead_v1_follow", [false]).await.unwrap(); // Initialized must always be reported first. let event: FollowEvent = get_next_event(&mut sub).await; @@ -3827,7 +3794,7 @@ async fn follow_unique_pruned_blocks() { let sub_id = sub.subscription_id(); let sub_id = serde_json::to_string(&sub_id).unwrap(); let hash = format!("{:?}", block_2_hash); - let _res: () = api.call("chainHead_unstable_unpin", rpc_params![&sub_id, &hash]).await.unwrap(); + let _res: () = api.call("chainHead_v1_unpin", rpc_params![&sub_id, &hash]).await.unwrap(); // Import block 7 and check it. let block_7_hash = import_block(client.clone(), block_6_hash, 3).await.hash();