[fastpay_core] Add support atomic-swaps on the server side

fastpay/Cargo.toml

@@ -8,6 +8,7 @@ edition = "2018"
 [dependencies]
 bytes = "0.5.6"
 clap = "2.33.3"
+either = "1.6.1"
 env_logger = "0.7.1"
 failure = "0.1.8"
 futures = "0.3.5"

fastpay/src/network.rs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::transport::*;
+use either::Either;
 use fastpay_core::{authority::*, base_types::*, client::*, error::*, messages::*, serialize::*};
 
 use bytes::Bytes;
@@ -164,6 +165,23 @@ impl MessageHandler for RunningServerState {
                                 Err(error) => Err(error),
                             }
                         }
+                        SerializedMessage::ConsensusOrder(message) => {
+                            match self.server.state.handle_consensus_order(*message) {
+                                Ok(Either::Left(vote)) => {
+                                    // Response
+                                    Ok(Some(serialize_vote(&vote)))
+                                }
+                                Ok(Either::Right(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::InfoQuery(message) => self
                             .server
                             .state

fastpay_core/Cargo.toml

@@ -8,6 +8,7 @@ edition = "2018"
 [dependencies]
 bcs = "0.1.3"
 bincode = "1.3.1"
+either = "1.6.1"
 failure = "0.1.8"
 futures = "0.3.5"
 generic-array = { version = "0.14.4", features = ["serde"] }

fastpay_core/src/account.rs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::{base_types::*, error::FastPayError, messages::*};
+use std::collections::BTreeMap;
 
 /// State of a FastPay account.
 #[derive(Debug, Default)]
@@ -108,10 +109,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)
     }
@@ -127,7 +152,9 @@ impl AccountState {
             operation
         );
         match operation {
-            Operation::OpenAccount { .. } => (),
+            Operation::OpenAccount { .. }
+            | Operation::StartConsensusInstance { .. }
+            | Operation::Skip => (),
             Operation::ChangeOwner { new_owner } => {
                 self.owner = Some(*new_owner);
             }
@@ -137,9 +164,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);

fastpay_core/src/authority.rs

@@ -2,8 +2,10 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::{
-    account::AccountState, base_types::*, committee::Committee, error::FastPayError, messages::*,
+    account::AccountState, base_types::*, committee::Committee, consensus::ConsensusState,
+    error::FastPayError, messages::*,
 };
+use either::Either;
 use std::collections::{BTreeMap, BTreeSet, HashSet};
 
 #[cfg(test)]
@@ -20,6 +22,8 @@ pub struct AuthorityState {
     pub key_pair: KeyPair,
     /// States of FastPay accounts.
     pub accounts: BTreeMap<AccountId, AccountState>,
+    /// States of consensus instances.
+    pub instances: BTreeMap<AccountId, ConsensusState>,
     /// The latest transaction index of the blockchain that the authority has seen.
     pub last_transaction_index: SequenceNumber,
     /// The sharding ID of this authority shard. 0 if one shard.
@@ -60,6 +64,12 @@ pub trait Authority {
         order: CoinCreationOrder,
     ) -> Result<(Vec<Vote>, Vec<CrossShardContinuation>), FastPayError>;
 
+    /// Process a message meant for a consensus instance.
+    fn handle_consensus_order(
+        &mut self,
+        order: ConsensusOrder,
+    ) -> Result<Either<Vote, Vec<CrossShardContinuation>>, FastPayError>;
+
     /// Force synchronization to finalize requests from Primary to FastPay.
     fn handle_primary_synchronization_order(
         &mut self,
@@ -148,14 +158,25 @@ impl AuthorityState {
         operation: Operation,
         certificate: Certificate,
     ) -> Result<(), FastPayError> {
-        let recipient = operation
-            .recipient()
-            .ok_or(FastPayError::InvalidCrossShardRequest)?;
-        // Verify sharding.
-        fp_ensure!(self.in_shard(recipient), FastPayError::WrongShard);
-        // Execute the recipient's side of the operation.
-        let account = self.accounts.entry(recipient.clone()).or_default();
-        account.apply_operation_as_recipient(&operation, certificate)?;
+        if let Some(recipient) = operation.recipient() {
+            fp_ensure!(self.in_shard(recipient), FastPayError::WrongShard);
+            // Execute the recipient's side of the operation.
+            let account = self.accounts.entry(recipient.clone()).or_default();
+            account.apply_operation_as_recipient(&operation, certificate)?;
+        }
+        match &operation {
+            Operation::StartConsensusInstance {
+                new_id,
+                accounts,
+                functionality,
+            } => {
+                fp_ensure!(self.in_shard(new_id), FastPayError::WrongShard);
+                assert!(!self.instances.contains_key(new_id)); // guaranteed under BFT assumptions.
+                let instance = ConsensusState::new(*functionality, accounts.clone());
+                self.instances.insert(new_id.clone(), instance);
+            }
+            _ => (),
+        }
         // This concludes the confirmation of `certificate`.
         Ok(())
     }
@@ -345,6 +366,167 @@ impl Authority for AuthorityState {
         Ok((votes, continuations))
     }
 
+    /// Process a message meant for a consensus instance.
+    fn handle_consensus_order(
+        &mut self,
+        order: ConsensusOrder,
+    ) -> Result<Either<Vote, Vec<CrossShardContinuation>>, FastPayError> {
+        match order {
+            ConsensusOrder::Propose {
+                proposal,
+                owner,
+                signature,
+                locks,
+            } => {
+                let instance = self
+                    .instances
+                    .get_mut(&proposal.instance_id)
+                    .ok_or_else(|| {
+                        FastPayError::UnknownConsensusInstance(proposal.instance_id.clone())
+                    })?;
+                // Process lock certificates.
+                for lock in locks {
+                    lock.check(&self.committee)?;
+                    match lock.value {
+                        Value::Lock(Request {
+                            account_id,
+                            operation: Operation::LockInto { instance_id, owner },
+                            sequence_number,
+                        }) if instance_id == proposal.instance_id
+                            && Some(&sequence_number)
+                                == instance.sequence_numbers.get(&account_id) =>
+                        {
+                            // Update locking status for `account_id`.
+                            instance.locked_accounts.insert(account_id, owner);
+                            instance.participants.insert(owner);
+                        }
+                        _ => fp_bail!(FastPayError::InvalidConsensusOrder),
+                    }
+                }
+                // Verify the signature and that the author of the signature is authorized.
+                fp_ensure!(
+                    instance.participants.contains(&owner),
+                    FastPayError::InvalidConsensusOrder
+                );
+                signature.check(&proposal, owner)?;
+                // Check validity of the proposal and obtain the corresponding requests.
+                let requests = instance.make_requests(proposal.decision)?;
+                // TODO: verify that `proposal.round` is "available".
+                // Verify safety.
+                if let Some(proposed) = &instance.proposed {
+                    fp_ensure!(
+                        (proposed.round == proposal.round
+                            && proposed.decision == proposal.decision)
+                            || proposed.round < proposal.round,
+                        FastPayError::UnsafeConsensusProposal
+                    );
+                }
+                if let Some(locked) = &instance.locked {
+                    fp_ensure!(
+                        locked.round < proposal.round && locked.decision == proposal.decision,
+                        FastPayError::UnsafeConsensusProposal
+                    );
+                }
+                // Update proposed decision.
+                instance.proposed = Some(proposal.clone());
+                // Vote in favor of pre-commit (aka lock).
+                let value = Value::PreCommit { proposal, requests };
+                let vote = Vote::new(value, &self.key_pair);
+                Ok(Either::Left(vote))
+            }
+            ConsensusOrder::HandlePreCommit { certificate } => {
+                certificate.check(&self.committee)?;
+                let (proposal, requests) = match certificate.value {
+                    Value::PreCommit { proposal, requests } => (proposal, requests),
+                    _ => fp_bail!(FastPayError::InvalidConsensusOrder),
+                };
+                let instance = self
+                    .instances
+                    .get_mut(&proposal.instance_id)
+                    .ok_or_else(|| {
+                        FastPayError::UnknownConsensusInstance(proposal.instance_id.clone())
+                    })?;
+                // Verify safety.
+                if let Some(proposed) = &instance.proposed {
+                    fp_ensure!(
+                        proposed.round <= proposal.round,
+                        FastPayError::UnsafeConsensusPreCommit
+                    );
+                }
+                if let Some(locked) = &instance.locked {
+                    fp_ensure!(
+                        locked.round <= proposal.round,
+                        FastPayError::UnsafeConsensusPreCommit
+                    );
+                }
+                // Update locked decision.
+                instance.locked = Some(proposal.clone());
+                // Vote in favor of commit.
+                let value = Value::Commit { proposal, requests };
+                let vote = Vote::new(value, &self.key_pair);
+                Ok(Either::Left(vote))
+            }
+            ConsensusOrder::HandleCommit { certificate, locks } => {
+                certificate.check(&self.committee)?;
+                let (proposal, requests) = match &certificate.value {
+                    Value::Commit { proposal, requests } => (proposal, requests),
+                    _ => fp_bail!(FastPayError::InvalidConsensusOrder),
+                };
+                // Success.
+                // Only execute the requests in the commit once.
+                let mut requests = {
+                    if self.instances.contains_key(&proposal.instance_id) {
+                        requests.clone()
+                    } else {
+                        Vec::new()
+                    }
+                };
+                // Process lock certificates to add skip requests if needed.
+                if let ConsensusDecision::Abort = &proposal.decision {
+                    for lock in locks {
+                        lock.check(&self.committee)?;
+                        match lock.value {
+                            Value::Lock(Request {
+                                account_id,
+                                operation:
+                                    Operation::LockInto {
+                                        instance_id,
+                                        owner: _,
+                                    },
+                                sequence_number,
+                            }) if instance_id == proposal.instance_id => {
+                                requests.push(Request {
+                                    account_id: account_id.clone(),
+                                    operation: Operation::Skip,
+                                    sequence_number,
+                                });
+                            }
+                            _ => fp_bail!(FastPayError::InvalidConsensusOrder),
+                        }
+                    }
+                }
+                // Remove the consensus instance if needed.
+                self.instances.remove(&proposal.instance_id);
+                // Prepate cross-requests.
+                let continuations = requests
+                    .iter()
+                    .map(|request| {
+                        let shard_id = self.which_shard(&request.account_id);
+                        CrossShardContinuation::Request {
+                            shard_id,
+                            request: Box::new(CrossShardRequest::ProcessConfirmedRequest {
+                                request: request.clone(),
+                                certificate: certificate.clone(),
+                            }),
+                        }
+                    })
+                    .collect();
+                Ok(Either::Right(continuations))
+            }
+        }
+    }
+
+    /// Finalize a request from Primary.
     fn handle_primary_synchronization_order(
         &mut self,
         order: PrimarySynchronizationOrder,
@@ -387,6 +569,13 @@ impl Authority for AuthorityState {
                 self.accounts.remove(&account_id);
                 Ok(())
             }
+            CrossShardRequest::ProcessConfirmedRequest {
+                request,
+                certificate,
+            } => {
+                self.process_confirmed_request(request, certificate)?; // TODO: process continuations
+                Ok(())
+            }
         }
     }
 
@@ -418,6 +607,7 @@ impl AuthorityState {
             name,
             key_pair,
             accounts: BTreeMap::new(),
+            instances: BTreeMap::new(),
             last_transaction_index: SequenceNumber::new(),
             shard_id: 0,
             number_of_shards: 1,
@@ -435,6 +625,7 @@ impl AuthorityState {
             name: key_pair.public(),
             key_pair,
             accounts: BTreeMap::new(),
+            instances: BTreeMap::new(),
             last_transaction_index: SequenceNumber::new(),
             shard_id,
             number_of_shards,

fastpay_core/src/client.rs

@@ -718,7 +718,10 @@ where
             | Operation::SpendAndTransfer { .. } => {
                 self.key_pair = None;
             }
-            Operation::OpenAccount { .. } => (),
+            Operation::OpenAccount { .. }
+            | Operation::Skip
+            | Operation::LockInto { .. }
+            | Operation::StartConsensusInstance { .. } => (),
         }
         // Record certificate.
         self.sent_certificates.push(certificate);