Fix a bug where clients don't recover from conflicting proposals. (#3892)

## Motivation

If two clients make conflicting proposals, it can happen that both of
them have their own proposal in their local node, so when retrying, they
will retry in the same round, because their own pending proposal isn't
treated as a "conflict".

## Proposal

Use the next round, even if the proposal is our own.

## Test Plan

The added test reproduces the issue.

## Release Plan

- These changes should be backported to the latest `devnet` branch, then
    - be released in a new SDK.

## Links

- [reviewer
checklist](https://github.com/linera-io/linera-protocol/blob/main/CONTRIBUTING.md#reviewer-checklist)

Author: afck

linera-core/src/client/mod.rs

@@ -2361,7 +2361,7 @@ impl<Env: Environment> ChainClient<Env> {
         // Using the round number during execution counts as an oracle.
         // Accessing the round number in single-leader rounds where we are not the leader
         // is not currently supported.
-        let round = match Self::round_for_new_proposal(&info, &identity, &proposed_block, true)? {
+        let round = match Self::round_for_new_proposal(&info, &identity, true)? {
             Either::Left(round) => round.multi_leader(),
             Either::Right(_) => None,
         };
@@ -2716,7 +2716,7 @@ impl<Env: Environment> ChainClient<Env> {
             // Use the round number assuming there are oracle responses.
             // Using the round number during execution counts as an oracle.
             let proposed_block = pending_proposal.block;
-            let round = match Self::round_for_new_proposal(&info, &owner, &proposed_block, true)? {
+            let round = match Self::round_for_new_proposal(&info, &owner, true)? {
                 Either::Left(round) => round.multi_leader(),
                 Either::Right(_) => None,
             };
@@ -2731,12 +2731,7 @@ impl<Env: Environment> ChainClient<Env> {
 
         let has_oracle_responses = block.has_oracle_responses();
         let (proposed_block, outcome) = block.into_proposal();
-        let round = match Self::round_for_new_proposal(
-            &info,
-            &owner,
-            &proposed_block,
-            has_oracle_responses,
-        )? {
+        let round = match Self::round_for_new_proposal(&info, &owner, has_oracle_responses)? {
             Either::Left(round) => round,
             Either::Right(timeout) => return Ok(ClientOutcome::WaitForTimeout(timeout)),
         };
@@ -2855,16 +2850,17 @@ impl<Env: Environment> ChainClient<Env> {
     fn round_for_new_proposal(
         info: &ChainInfo,
         identity: &AccountOwner,
-        block: &ProposedBlock,
         has_oracle_responses: bool,
     ) -> Result<Either<Round, RoundTimeout>, ChainClientError> {
         let manager = &info.manager;
         // If there is a conflicting proposal in the current round, we can only propose if the
         // next round can be started without a timeout, i.e. if we are in a multi-leader round.
         // Similarly, we cannot propose a block that uses oracles in the fast round.
-        let conflict = manager.requested_proposed.as_ref().is_some_and(|proposal| {
-            proposal.content.round == manager.current_round && proposal.content.block != *block
-        }) || (manager.current_round.is_fast() && has_oracle_responses);
+        let conflict = manager
+            .requested_proposed
+            .as_ref()
+            .is_some_and(|proposal| proposal.content.round == manager.current_round)
+            || (manager.current_round.is_fast() && has_oracle_responses);
         let round = if !conflict {
             manager.current_round
         } else if let Some(round) = manager

linera-core/src/unit_tests/client_tests.rs

@@ -1532,6 +1532,85 @@ where
     Ok(())
 }
 
+#[test_case(MemoryStorageBuilder::default(); "memory")]
+#[cfg_attr(feature = "storage-service", test_case(ServiceStorageBuilder::new().await; "storage_service"))]
+#[cfg_attr(feature = "rocksdb", test_case(RocksDbStorageBuilder::new().await; "rocks_db"))]
+#[cfg_attr(feature = "dynamodb", test_case(DynamoDbStorageBuilder::default(); "dynamo_db"))]
+#[cfg_attr(feature = "scylladb", test_case(ScyllaDbStorageBuilder::default(); "scylla_db"))]
+#[test_log::test(tokio::test)]
+async fn test_conflicting_proposals<B>(storage_builder: B) -> anyhow::Result<()>
+where
+    B: StorageBuilder,
+{
+    let mut signer = InMemorySigner::new(None);
+    let owner2 = signer.generate_new().into();
+    let mut builder = TestBuilder::new(storage_builder, 4, 0, &mut signer).await?;
+    let client1 = builder.add_root_chain(1, Amount::ONE).await?;
+    let chain_id = client1.chain_id();
+    let owner1 = client1.public_key().await?.into();
+    let owner_change_op = Operation::system(SystemOperation::ChangeOwnership {
+        super_owners: Vec::new(),
+        owners: vec![(owner1, 50), (owner2, 50)],
+        multi_leader_rounds: 10,
+        open_multi_leader_rounds: false,
+        timeout_config: TimeoutConfig::default(),
+    });
+    client1
+        .execute_operation(owner_change_op.clone())
+        .await
+        .unwrap();
+    let mut client2 = builder
+        .make_client(chain_id, client1.block_hash(), BlockHeight::from(1))
+        .await?;
+    client2.set_preferred_owner(owner2);
+    client2.synchronize_from_validators().await.unwrap();
+
+    // Client 1 makes a proposal to only validators 0 and 1.
+    builder
+        .set_fault_type([2, 3], FaultType::OfflineWithInfo)
+        .await;
+    assert!(client1
+        .burn(AccountOwner::CHAIN, Amount::from_millis(1))
+        .await
+        .is_err());
+
+    // Client 2's proposal reaches only 2 and 3.
+    builder
+        .set_fault_type([0, 1], FaultType::OfflineWithInfo)
+        .await;
+    builder.set_fault_type([2, 3], FaultType::Honest).await;
+    assert!(client2
+        .burn(AccountOwner::CHAIN, Amount::from_millis(2))
+        .await
+        .is_err());
+
+    // TODO(#3894): Make test_utils deterministic.
+    // The following condition is currently satisfied most of the time, but in some cases
+    // the faulty validators return an error before the honest ones process the proposal.
+
+    // for i in 0..4 {
+    //     let info = builder
+    //         .node(i)
+    //         .chain_info_with_manager_values(chain_id)
+    //         .await?;
+    //     assert_eq!(
+    //         AccountOwner::from(info.manager.requested_proposed.unwrap().public_key),
+    //         if i < 2 { owner1 } else { owner2 },
+    //     );
+    // }
+
+    // Once all validators are functional again, a new proposal should succeed.
+    builder
+        .set_fault_type([0, 1, 2, 3], FaultType::Honest)
+        .await;
+
+    client1.synchronize_from_validators().await.unwrap();
+    client1.publish_data_blob(b"foo".to_vec()).await?;
+
+    assert_eq!(client1.next_block_height(), BlockHeight::from(3));
+    Ok(())
+}
+
 #[test_case(MemoryStorageBuilder::default(); "memory")]
 #[cfg_attr(feature = "storage-service", test_case(ServiceStorageBuilder::new().await; "storage_service"))]
 #[cfg_attr(feature = "rocksdb", test_case(RocksDbStorageBuilder::new().await; "rocks_db"))]