add SwapStatus to keep track of progress of each swap

src/onion.rs

@@ -129,7 +129,9 @@ impl Readable for Onion {
 		let mut enc_payloads: Vec<RawBytes> = Vec::new();
 		let len = reader.read_u64()?;
 		for _ in 0..len {
-			enc_payloads.push(RawBytes::read(reader)?);
+			let size = reader.read_u64()?;
+			let bytes = reader.read_fixed_bytes(size as usize)?;
+			enc_payloads.push(bytes);
 		}
 		Ok(Onion {
 			ephemeral_pubkey,
@@ -265,11 +267,13 @@ impl From<ser::Error> for OnionError {
 #[cfg(test)]
 pub mod test_util {
 	use super::{Onion, OnionError, RawBytes};
-	use crate::secp::{Commitment, PublicKey, Secp256k1, SecretKey, SharedSecret};
+	use crate::secp::test_util::{rand_commit, rand_proof, rand_pubkey};
+	use crate::secp::{self, Commitment, PublicKey, Secp256k1, SecretKey, SharedSecret};
 	use crate::types::Payload;
 
-	use crate::secp;
 	use chacha20::cipher::StreamCipher;
+	use grin_core::core::FeeFields;
+	use rand::RngCore;
 
 	#[derive(Clone)]
 	pub struct Hop {
@@ -315,6 +319,29 @@ pub mod test_util {
 		Ok(onion)
 	}
 
+	pub fn rand_onion() -> Onion {
+		let commit = rand_commit();
+		let mut hops = Vec::new();
+		let k = (rand::thread_rng().next_u64() % 5) + 1;
+		for i in 0..k {
+			let hop = Hop {
+				pubkey: rand_pubkey(),
+				payload: Payload {
+					excess: secp::random_secret(),
+					fee: FeeFields::from(rand::thread_rng().next_u32()),
+					rangeproof: if i == (k - 1) {
+						Some(rand_proof())
+					} else {
+						None
+					},
+				},
+			};
+			hops.push(hop);
+		}
+
+		create_onion(&commit, &hops).unwrap()
+	}
+
 	/// Calculates the expected next ephemeral pubkey after peeling a layer off of the Onion.
 	pub fn next_ephemeral_pubkey(
 		onion: &Onion,

src/secp.rs

@@ -216,6 +216,39 @@ pub fn sign(sk: &SecretKey, msg: &Message) -> Result<Signature, secp256k1zkp::Er
 	Ok(sig)
 }
 
+#[cfg(test)]
+pub mod test_util {
+	use crate::secp::{self, Commitment, PublicKey, RangeProof, Secp256k1};
+	use grin_core::core::hash::Hash;
+	use grin_util::ToHex;
+	use rand::RngCore;
+
+	pub fn rand_commit() -> Commitment {
+		secp::commit(rand::thread_rng().next_u64(), &secp::random_secret()).unwrap()
+	}
+
+	pub fn rand_hash() -> Hash {
+		Hash::from_hex(secp::random_secret().to_hex().as_str()).unwrap()
+	}
+
+	pub fn rand_proof() -> RangeProof {
+		let secp = Secp256k1::new();
+		secp.bullet_proof(
+			rand::thread_rng().next_u64(),
+			secp::random_secret(),
+			secp::random_secret(),
+			secp::random_secret(),
+			None,
+			None,
+		)
+	}
+
+	pub fn rand_pubkey() -> PublicKey {
+		let secp = Secp256k1::new();
+		PublicKey::from_secret_key(&secp, &secp::random_secret()).unwrap()
+	}
+}
+
 #[cfg(test)]
 mod tests {
 	use super::{ComSigError, ComSignature, ContextFlag, Secp256k1, SecretKey};

src/server.rs

@@ -2,10 +2,11 @@ use crate::config::ServerConfig;
 use crate::node::{self, GrinNode};
 use crate::onion::{Onion, OnionError};
 use crate::secp::{ComSignature, Commitment, Secp256k1, SecretKey};
-use crate::store::{StoreError, SwapData, SwapStore};
+use crate::store::{StoreError, SwapData, SwapStatus, SwapStore};
 use crate::wallet::{self, Wallet};
 
-use grin_core::core::{Input, Output, OutputFeatures, TransactionBody};
+use grin_core::core::hash::Hashed;
+use grin_core::core::{Input, Output, OutputFeatures, Transaction, TransactionBody};
 use grin_core::global::DEFAULT_ACCEPT_FEE_BASE;
 use itertools::Itertools;
 use std::result::Result;
@@ -46,7 +47,7 @@ pub trait Server: Send + Sync {
 	/// and assemble the coinswap transaction, posting the transaction to the configured node.
 	///
 	/// Currently only a single mix node is used. Milestone 3 will include support for multiple mix nodes.
-	fn execute_round(&self) -> Result<(), Box<dyn std::error::Error>>;
+	fn execute_round(&self) -> Result<Option<Transaction>, Box<dyn std::error::Error>>;
 }
 
 /// The standard MWixnet server implementation
@@ -137,14 +138,18 @@ impl Server for ServerImpl {
 		let locked = self.store.lock().unwrap();
 
 		locked
-			.save_swap(&SwapData {
-				excess: peeled.0.excess,
-				output_commit: peeled.1.commit,
-				rangeproof: peeled.0.rangeproof,
-				input,
-				fee,
-				onion: peeled.1,
-			})
+			.save_swap(
+				&SwapData {
+					excess: peeled.0.excess,
+					output_commit: peeled.1.commit,
+					rangeproof: peeled.0.rangeproof,
+					input,
+					fee,
+					onion: peeled.1,
+					status: SwapStatus::Unprocessed,
+				},
+				false,
+			)
 			.map_err(|e| match e {
 				StoreError::AlreadyExists(_) => SwapError::AlreadySwapped {
 					commit: onion.commit.clone(),
@@ -154,13 +159,17 @@ impl Server for ServerImpl {
 		Ok(())
 	}
 
-	fn execute_round(&self) -> Result<(), Box<dyn std::error::Error>> {
+	fn execute_round(&self) -> Result<Option<Transaction>, Box<dyn std::error::Error>> {
 		let locked_store = self.store.lock().unwrap();
 		let next_block_height = self.node.get_chain_height()? + 1;
 
 		let spendable: Vec<SwapData> = locked_store
 			.swaps_iter()?
 			.unique_by(|s| s.output_commit)
+			.filter(|s| match s.status {
+				SwapStatus::Unprocessed => true,
+				_ => false,
+			})
 			.filter(|s| {
 				node::is_spendable(&self.node, &s.input.commit, next_block_height).unwrap_or(false)
 			})
@@ -168,7 +177,7 @@ impl Server for ServerImpl {
 			.collect();
 
 		if spendable.len() == 0 {
-			return Ok(());
+			return Ok(None);
 		}
 
 		let total_fee: u64 = spendable.iter().enumerate().map(|(_, s)| s.fee).sum();
@@ -203,9 +212,15 @@ impl Server for ServerImpl {
 		)?;
 
 		self.node.post_tx(&tx)?;
-		// todo: Update swap statuses
 
-		Ok(())
+		// Update status to in process
+		let kernel_hash = tx.kernels().first().unwrap().hash();
+		for mut swap in spendable {
+			swap.status = SwapStatus::InProcess { kernel_hash };
+			locked_store.save_swap(&swap, true)?;
+		}
+
+		Ok(Some(tx))
 	}
 }
 
@@ -215,6 +230,7 @@ pub mod mock {
 	use crate::onion::Onion;
 	use crate::secp::ComSignature;
 
+	use grin_core::core::Transaction;
 	use std::collections::HashMap;
 
 	pub struct MockServer {
@@ -242,8 +258,8 @@ pub mod mock {
 			Ok(())
 		}
 
-		fn execute_round(&self) -> Result<(), Box<dyn std::error::Error>> {
-			Ok(())
+		fn execute_round(&self) -> Result<Option<Transaction>, Box<dyn std::error::Error>> {
+			Ok(None)
 		}
 	}
 }
@@ -258,10 +274,11 @@ mod tests {
 		self, ComSignature, Commitment, PublicKey, RangeProof, Secp256k1, SecretKey,
 	};
 	use crate::server::{Server, ServerImpl, SwapError};
-	use crate::store::{SwapData, SwapStore};
+	use crate::store::{SwapData, SwapStatus, SwapStore};
 	use crate::types::Payload;
 	use crate::wallet::mock::MockWallet;
 
+	use grin_core::core::hash::Hashed;
 	use grin_core::core::{Committed, FeeFields, Input, OutputFeatures, Transaction, Weighting};
 	use grin_core::global::{self, ChainTypes};
 	use std::net::TcpListener;
@@ -379,6 +396,7 @@ mod tests {
 				commit: output_commit.clone(),
 				enc_payloads: vec![],
 			},
+			status: SwapStatus::Unprocessed,
 		};
 
 		{
@@ -388,13 +406,18 @@ mod tests {
 			assert_eq!(expected, store.get_swap(&input_commit).unwrap());
 		}
 
-		server.execute_round()?;
+		let tx = server.execute_round()?;
+		assert!(tx.is_some());
 
-		// todo: Make sure entry is removed from server.submissions
-		// assert_eq!(
-		// 	0,
-		// 	server.store.lock().unwrap().swaps_iter().unwrap().count()
-		// );
+		{
+			// check that status was updated
+			let store = server.store.lock().unwrap();
+			assert!(match store.get_swap(&input_commit)?.status {
+				SwapStatus::InProcess { kernel_hash } =>
+					kernel_hash == tx.unwrap().kernels().first().unwrap().hash(),
+				_ => false,
+			});
+		}
 
 		// check that the transaction was posted
 		let posted_txns = node.get_posted_txns();

src/store.rs

@@ -1,6 +1,7 @@
 use crate::onion::Onion;
 use crate::secp::{self, Commitment, RangeProof, SecretKey};
 use crate::types::{read_optional, write_optional};
+use grin_core::core::hash::Hash;
 
 use grin_core::core::Input;
 use grin_core::ser::{
@@ -16,6 +17,62 @@ const STORE_SUBPATH: &str = "swaps";
 const CURRENT_VERSION: u8 = 0;
 const SWAP_PREFIX: u8 = b'S';
 
+/// Swap statuses
+#[derive(Clone, Debug, PartialEq)]
+pub enum SwapStatus {
+	Unprocessed,
+	InProcess { kernel_hash: Hash },
+	Completed { kernel_hash: Hash, block_hash: Hash },
+}
+
+impl Writeable for SwapStatus {
+	fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
+		match self {
+			SwapStatus::Unprocessed => {
+				writer.write_u8(0)?;
+			}
+			SwapStatus::InProcess { kernel_hash } => {
+				writer.write_u8(1)?;
+				kernel_hash.write(writer)?;
+			}
+			SwapStatus::Completed {
+				kernel_hash,
+				block_hash,
+			} => {
+				writer.write_u8(2)?;
+				kernel_hash.write(writer)?;
+				block_hash.write(writer)?;
+			}
+		};
+
+		Ok(())
+	}
+}
+
+impl Readable for SwapStatus {
+	fn read<R: Reader>(reader: &mut R) -> Result<SwapStatus, ser::Error> {
+		let status = match reader.read_u8()? {
+			0 => SwapStatus::Unprocessed,
+			1 => {
+				let kernel_hash = Hash::read(reader)?;
+				SwapStatus::InProcess { kernel_hash }
+			}
+			2 => {
+				let kernel_hash = Hash::read(reader)?;
+				let block_hash = Hash::read(reader)?;
+				SwapStatus::Completed {
+					kernel_hash,
+					block_hash,
+				}
+			}
+			_ => {
+				return Err(ser::Error::CorruptedData);
+			}
+		};
+		Ok(status)
+	}
+}
+
 /// Data needed to swap a single output.
 #[derive(Clone, Debug, PartialEq)]
 pub struct SwapData {
@@ -31,7 +88,8 @@ pub struct SwapData {
 	pub fee: u64,
 	/// The remaining onion after peeling off our layer
 	pub onion: Onion,
-	// todo: include a SwapStatus enum value
+	/// The status of the swap
+	pub status: SwapStatus,
 }
 
 impl Writeable for SwapData {
@@ -43,6 +101,7 @@ impl Writeable for SwapData {
 		self.input.write(writer)?;
 		writer.write_u64(self.fee.into())?;
 		self.onion.write(writer)?;
+		self.status.write(writer)?;
 
 		Ok(())
 	}
@@ -61,6 +120,7 @@ impl Readable for SwapData {
 		let input = Input::read(reader)?;
 		let fee = reader.read_u64()?;
 		let onion = Onion::read(reader)?;
+		let status = SwapStatus::read(reader)?;
 		Ok(SwapData {
 			excess,
 			output_commit,
@@ -68,6 +128,7 @@ impl Readable for SwapData {
 			input,
 			fee,
 			onion,
+			status,
 		})
 	}
 }
@@ -112,10 +173,11 @@ impl SwapStore {
 		prefix: u8,
 		k: K,
 		value: &Vec<u8>,
+		overwrite: bool,
 	) -> Result<bool, store::lmdb::Error> {
 		let batch = self.db.batch()?;
 		let key = store::to_key(prefix, k);
-		if batch.exists(&key[..])? {
+		if !overwrite && batch.exists(&key[..])? {
 			Ok(false)
 		} else {
 			batch.put(&key[..], &value[..])?;
@@ -133,10 +195,10 @@ impl SwapStore {
 	}
 
 	/// Saves a swap to the database
-	pub fn save_swap(&self, s: &SwapData) -> Result<(), StoreError> {
+	pub fn save_swap(&self, s: &SwapData, overwrite: bool) -> Result<(), StoreError> {
 		let data = ser::ser_vec(&s, ProtocolVersion::local())?;
 		let saved = self
-			.write(SWAP_PREFIX, &s.input.commit, &data)
+			.write(SWAP_PREFIX, &s.input.commit, &data, overwrite)
 			.map_err(StoreError::WriteError)?;
 		if !saved {
 			Err(StoreError::AlreadyExists(s.input.commit.clone()))
@@ -174,3 +236,106 @@ impl SwapStore {
 		self.read(SWAP_PREFIX, input_commit)
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	use crate::onion::test_util::rand_onion;
+	use crate::secp::test_util::{rand_commit, rand_hash, rand_proof};
+	use crate::store::{SwapData, SwapStatus, SwapStore};
+	use crate::{secp, StoreError};
+	use grin_core::core::{Input, OutputFeatures};
+	use grin_core::global::{self, ChainTypes};
+	use rand::RngCore;
+	use std::cmp::Ordering;
+
+	fn new_store(test_name: &str) -> SwapStore {
+		global::set_local_chain_type(ChainTypes::AutomatedTesting);
+		let db_root = format!("./target/tmp/.{}", test_name);
+		let _ = std::fs::remove_dir_all(db_root.as_str());
+		SwapStore::new(db_root.as_str()).unwrap()
+	}
+
+	fn rand_swap_with_status(status: SwapStatus) -> SwapData {
+		SwapData {
+			excess: secp::random_secret(),
+			output_commit: rand_commit(),
+			rangeproof: Some(rand_proof()),
+			input: Input::new(OutputFeatures::Plain, rand_commit()),
+			fee: rand::thread_rng().next_u64(),
+			onion: rand_onion(),
+			status,
+		}
+	}
+
+	fn rand_swap() -> SwapData {
+		let s = rand::thread_rng().next_u64() % 3;
+		let status = if s == 0 {
+			SwapStatus::Unprocessed
+		} else if s == 1 {
+			SwapStatus::InProcess {
+				kernel_hash: rand_hash(),
+			}
+		} else {
+			SwapStatus::Completed {
+				kernel_hash: rand_hash(),
+				block_hash: rand_hash(),
+			}
+		};
+		rand_swap_with_status(status)
+	}
+
+	#[test]
+	fn swap_iter() -> Result<(), Box<dyn std::error::Error>> {
+		let store = new_store("swap_iter");
+		let mut swaps: Vec<SwapData> = Vec::new();
+		for _ in 0..5 {
+			let swap = rand_swap();
+			store.save_swap(&swap, false)?;
+			swaps.push(swap);
+		}
+
+		swaps.sort_by(|a, b| {
+			if a.input.commit < b.input.commit {
+				Ordering::Less
+			} else if a.input.commit == b.input.commit {
+				Ordering::Equal
+			} else {
+				Ordering::Greater
+			}
+		});
+
+		let mut i: usize = 0;
+		for swap in store.swaps_iter()? {
+			assert_eq!(swap, *swaps.get(i).unwrap());
+			i += 1;
+		}
+
+		Ok(())
+	}
+
+	#[test]
+	fn save_swap() -> Result<(), Box<dyn std::error::Error>> {
+		let store = new_store("save_swap");
+
+		let mut swap = rand_swap_with_status(SwapStatus::Unprocessed);
+		assert!(!store.swap_exists(&swap.input.commit)?);
+
+		store.save_swap(&swap, false)?;
+		assert_eq!(swap, store.get_swap(&swap.input.commit)?);
+		assert!(store.swap_exists(&swap.input.commit)?);
+
+		swap.status = SwapStatus::InProcess {
+			kernel_hash: rand_hash(),
+		};
+		let result = store.save_swap(&swap, false);
+		assert_eq!(
+			Err(StoreError::AlreadyExists(swap.input.commit.clone())),
+			result
+		);
+
+		store.save_swap(&swap, true)?;
+		assert_eq!(swap, store.get_swap(&swap.input.commit)?);
+
+		Ok(())
+	}
+}