[draft] Add support for atomic-swaps on the server side

benchmark_client/src/client.rs

@@ -159,7 +159,7 @@ impl BenchmarkClient {
                 tokio::select! {
                     Some(bytes) = rx_certificate.recv() => {
                         match deserialize_message(&*bytes).unwrap() {
-                            SerializedMessage::InfoResponse(response) => {
+                            SerializedMessage::AccountInfoResponse(response) => {
                                 let id = response
                                     .account_id
                                     .sequence_number()

benchmark_client/src/coco_client.rs

@@ -186,7 +186,7 @@ impl CocoBenchmarkClient {
                 tokio::select! {
                     Some(bytes) = rx_certificate.recv() => {
                         match deserialize_message(&*bytes).unwrap() {
-                            SerializedMessage::InfoResponse(response) => {
+                            SerializedMessage::AccountInfoResponse(response) => {
                                 let id = response
                                     .account_id
                                     .sequence_number()

benchmark_server/src/core.rs

@@ -9,8 +9,8 @@ use fastpay_core::{
     base_types::AuthorityName,
     error::FastPayError,
     serialize::{
-        serialize_coin_creation_response, serialize_cross_shard_request, serialize_error,
-        serialize_info_response, SerializedMessage,
+        serialize_account_info_response, serialize_coin_creation_response,
+        serialize_cross_shard_request, serialize_error, SerializedMessage,
     },
 };
 use log::{debug, warn};
@@ -58,14 +58,14 @@ impl MessageHandler for Core {
             SerializedMessage::RequestOrder(message) => self
                 .state
                 .handle_request_order(*message)
-                .map(|info| Some(serialize_info_response(&info))),
+                .map(|info| Some(serialize_account_info_response(&info))),
             SerializedMessage::ConfirmationOrder(message) => {
                 match self.state.handle_confirmation_order(*message) {
                     Ok((info, continuation)) => {
                         // Cross-shard request
                         self.handle_continuation(continuation).await;
                         // Response
-                        Ok(Some(serialize_info_response(&info)))
+                        Ok(Some(serialize_account_info_response(&info)))
                     }
                     Err(error) => Err(error),
                 }

fastpay/Cargo.toml

@@ -9,6 +9,7 @@ edition = "2018"
 bytes = "1.1.0"
 clap = "2.33.3"
 env_logger = "0.9.0"
+either = "1.6.1"
 failure = "0.1.8"
 futures = "0.3.17"
 log = "0.4.14"

fastpay/src/client.rs

@@ -341,7 +341,7 @@ impl ClientContext {
 
 fn deserialize_response(response: &[u8]) -> Option<AccountInfoResponse> {
     match deserialize_message(response) {
-        Ok(SerializedMessage::InfoResponse(info)) => Some(*info),
+        Ok(SerializedMessage::AccountInfoResponse(info)) => Some(*info),
         Ok(SerializedMessage::Error(error)) => {
             error!("Received error value: {}", error);
             None

fastpay/src/network.rs

@@ -191,14 +191,14 @@ impl MessageHandler for RunningServerState {
                             .server
                             .state
                             .handle_request_order(*message)
-                            .map(|info| Some(serialize_info_response(&info))),
+                            .map(|info| Some(serialize_account_info_response(&info))),
                         SerializedMessage::ConfirmationOrder(message) => {
                             match self.server.state.handle_confirmation_order(*message) {
                                 Ok((info, continuation)) => {
                                     // Cross-shard request
                                     self.handle_continuation(continuation).await;
                                     // Response
-                                    Ok(Some(serialize_info_response(&info)))
+                                    Ok(Some(serialize_account_info_response(&info)))
                                 }
                                 Err(error) => Err(error),
                             }
@@ -216,11 +216,32 @@ impl MessageHandler for RunningServerState {
                                 Err(error) => Err(error),
                             }
                         }
-                        SerializedMessage::InfoQuery(message) => self
+                        SerializedMessage::ConsensusOrder(message) => {
+                            match self.server.state.handle_consensus_order(*message) {
+                                Ok(ConsensusResponse::Info(info)) => {
+                                    // Response
+                                    Ok(Some(serialize_consensus_info_response(&info)))
+                                }
+                                Ok(ConsensusResponse::Vote(vote)) => {
+                                    // Response
+                                    Ok(Some(serialize_vote(&vote)))
+                                }
+                                Ok(ConsensusResponse::Continuations(continuations)) => {
+                                    // Cross-shard requests
+                                    for continuation in continuations {
+                                        self.handle_continuation(continuation).await;
+                                    }
+                                    // No response. (TODO: this is a bit rough)
+                                    Ok(None)
+                                }
+                                Err(error) => Err(error),
+                            }
+                        }
+                        SerializedMessage::AccountInfoQuery(message) => self
                             .server
                             .state
                             .handle_account_info_query(*message)
-                            .map(|info| Some(serialize_info_response(&info))),
+                            .map(|info| Some(serialize_account_info_response(&info))),
                         SerializedMessage::CrossShardRequest(request) => {
                             match self.server.state.handle_cross_shard_request(*request) {
                                 Ok(()) => (),
@@ -234,7 +255,8 @@ impl MessageHandler for RunningServerState {
                         SerializedMessage::Vote(_)
                         | SerializedMessage::CoinCreationResponse(_)
                         | SerializedMessage::Error(_)
-                        | SerializedMessage::InfoResponse(_) => {
+                        | SerializedMessage::AccountInfoResponse(_)
+                        | SerializedMessage::ConsensusInfoResponse(_) => {
                             Err(FastPayError::UnexpectedMessage)
                         }
                     }
@@ -349,7 +371,7 @@ impl Client {
             Ok(response) => {
                 // Parse reply
                 match deserialize_message(&response[..]) {
-                    Ok(SerializedMessage::InfoResponse(resp)) => Ok(*resp),
+                    Ok(SerializedMessage::AccountInfoResponse(resp)) => Ok(*resp),
                     Ok(SerializedMessage::Error(error)) => Err(*error),
                     Err(_) => Err(FastPayError::InvalidDecoding),
                     _ => Err(FastPayError::UnexpectedMessage),
@@ -432,7 +454,7 @@ impl AuthorityClient for Client {
     ) -> AsyncResult<AccountInfoResponse, FastPayError> {
         Box::pin(async move {
             let shard = AuthorityState::get_shard(self.num_shards, &request.account_id);
-            self.send_recv_info_bytes(shard, serialize_info_query(&request))
+            self.send_recv_info_bytes(shard, serialize_account_info_query(&request))
                 .await
         })
     }

fastpay_core/Cargo.toml

@@ -9,6 +9,7 @@ edition = "2018"
 bcs = "0.1.3"
 bincode = "1.3.3"
 coconut = { version = "0.1.0", path = "../coconut", features = ["with_serde"] }
+either = "1.6.1"
 failure = "0.1.8"
 ff = "0.11.0"
 futures = "0.3.17"

fastpay_core/src/account.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::{base_types::*, error::FastPayError, messages::*};
-use std::collections::BTreeSet;
+use std::collections::{BTreeMap, BTreeSet};
 
 /// State of a FastPay account.
 #[derive(Debug, Default)]
@@ -152,10 +152,34 @@ impl AccountState {
                 );
                 Value::Confirm(request)
             }
-            Operation::CloseAccount | Operation::ChangeOwner { .. } => {
+            Operation::StartConsensusInstance {
+                new_id, accounts, ..
+            } => {
+                // Verify the new UID.
+                let expected_id = request.account_id.make_child(request.sequence_number);
+                fp_ensure!(
+                    new_id == &expected_id,
+                    FastPayError::InvalidNewAccountId(new_id.clone())
+                );
+                // Make sure accounts are unique.
+                let numbers = accounts
+                    .clone()
+                    .into_iter()
+                    .collect::<BTreeMap<AccountId, _>>();
+                fp_ensure!(
+                    numbers.len() == accounts.len(),
+                    FastPayError::InvalidRequestOrder
+                );
+                Value::Confirm(request)
+            }
+            Operation::Skip | Operation::CloseAccount | Operation::ChangeOwner { .. } => {
                 // Nothing to check.
                 Value::Confirm(request)
             }
+            Operation::LockInto { .. } => {
+                // Nothing to check.
+                Value::Lock(request)
+            }
         };
         Ok(value)
     }
@@ -171,7 +195,9 @@ impl AccountState {
             operation
         );
         match operation {
-            Operation::OpenAccount { .. } => (),
+            Operation::OpenAccount { .. }
+            | Operation::StartConsensusInstance { .. }
+            | Operation::Skip => (),
             Operation::ChangeOwner { new_owner } => {
                 self.owner = Some(*new_owner);
             }
@@ -181,9 +207,9 @@ impl AccountState {
             Operation::Transfer { amount, .. } => {
                 self.balance.try_sub_assign((*amount).into())?;
             }
-            Operation::Spend { .. } => {
+            Operation::Spend { .. } | Operation::LockInto { .. } => {
                 // impossible under BFT assumptions.
-                unreachable!("Spend operation are never confirmed");
+                unreachable!("Spend and lock operation are never confirmed");
             }
         };
         self.confirmed_log.push(certificate);