From 5c0ee0f0c264e8edc70ffa62c33aab8ae02bf632 Mon Sep 17 00:00:00 2001
From: Sam Gwilym <sam@gwil.garden>
Date: Mon, 11 Nov 2024 16:11:13 +0100
Subject: [PATCH 1/6] Reorganise, rename, send prelude

---
 wgps/src/commitment_scheme/mod.rs             |   2 +
 wgps/src/commitment_scheme/receive_prelude.rs | 171 ++++++++++++++++
 wgps/src/commitment_scheme/send_prelude.rs    |  18 ++
 wgps/src/lib.rs                               |   4 +-
 wgps/src/ready_transport.rs                   | 188 ------------------
 5 files changed, 193 insertions(+), 190 deletions(-)
 create mode 100644 wgps/src/commitment_scheme/mod.rs
 create mode 100644 wgps/src/commitment_scheme/receive_prelude.rs
 create mode 100644 wgps/src/commitment_scheme/send_prelude.rs
 delete mode 100644 wgps/src/ready_transport.rs

diff --git a/wgps/src/commitment_scheme/mod.rs b/wgps/src/commitment_scheme/mod.rs
new file mode 100644
index 0000000..9e8e6d7
--- /dev/null
+++ b/wgps/src/commitment_scheme/mod.rs
@@ -0,0 +1,2 @@
+pub mod receive_prelude;
+pub mod send_prelude;
diff --git a/wgps/src/commitment_scheme/receive_prelude.rs b/wgps/src/commitment_scheme/receive_prelude.rs
new file mode 100644
index 0000000..2a7e93c
--- /dev/null
+++ b/wgps/src/commitment_scheme/receive_prelude.rs
@@ -0,0 +1,171 @@
+use either::Either;
+use ufotofu::local_nb::BulkProducer;
+
+/// When things go wrong while trying to make a WGPS transport ready.
+#[derive(Debug)]
+pub enum ReceivePreludeError<E> {
+    /// The transport returned an error of its own.
+    Transport(E),
+    /// The received max payload power was invalid, i.e. greater than 64.
+    MaxPayloadInvalid,
+    /// The transport stopped producing bytes before it could be deemed ready.
+    FinishedTooSoon,
+}
+
+impl<E: core::fmt::Display> core::fmt::Display for ReceivePreludeError<E> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            ReceivePreludeError::Transport(e) => write!(f, "{}", e),
+            ReceivePreludeError::MaxPayloadInvalid => write!(
+                f,
+                "The received max payload power was invalid, i.e. greater than 64."
+            ),
+            ReceivePreludeError::FinishedTooSoon => write!(
+                f,
+                "The transport stopped producing bytes before it could be deemed ready."
+            ),
+        }
+    }
+}
+
+/// The result of intercepting the first few bytes of a WGPS transport.
+#[derive(Debug)]
+#[allow(dead_code)] // TODO: Remove when this is used.
+pub(crate) struct ReceivedPrelude<const CHALLENGE_HASH_LENGTH: usize> {
+    /// The maximum payload size which may be sent without being explicitly requested.
+    pub maximum_payload_size: usize,
+    /// The challenge hash of a nonce.
+    pub received_commitment: [u8; CHALLENGE_HASH_LENGTH],
+}
+
+/// Given a producer of bytes which is to immediately produce the bytes corresponding to the WGPS' [maximum payload size](https://willowprotocol.org/specs/sync/index.html#peer_max_payload_size) and [received commitment](https://willowprotocol.org/specs/sync/index.html#received_commitment), returns the computed maximum payload size, received commitment, and a 'ready' transport set to produce encoded WGPS messages.
+
+#[allow(dead_code)] // TODO: Remove when this is used.
+pub(crate) async fn receive_prelude<
+    const CHALLENGE_HASH_LENGTH: usize,
+    P: BulkProducer<Item = u8>,
+>(
+    transport: &mut P,
+) -> Result<ReceivedPrelude<CHALLENGE_HASH_LENGTH>, ReceivePreludeError<P::Error>>
+where
+    P::Error: core::fmt::Display,
+{
+    let maximum_payload_power = match transport.produce().await? {
+        Either::Left(byte) => byte,
+        Either::Right(_) => return Err(ReceivePreludeError::FinishedTooSoon),
+    };
+
+    if maximum_payload_power > 64 {
+        return Err(ReceivePreludeError::MaxPayloadInvalid);
+    }
+
+    let maximum_payload_size = 2_usize.pow(maximum_payload_power as u32);
+
+    let mut received_commitment = [0_u8; CHALLENGE_HASH_LENGTH];
+
+    if let Err(e) = transport
+        .bulk_overwrite_full_slice(&mut received_commitment)
+        .await
+    {
+        match e.reason {
+            Either::Left(_) => return Err(ReceivePreludeError::FinishedTooSoon),
+            Either::Right(e) => return Err(ReceivePreludeError::Transport(e)),
+        }
+    };
+
+    Ok(ReceivedPrelude {
+        maximum_payload_size,
+        received_commitment,
+    })
+}
+
+impl<E: core::fmt::Display> From<E> for ReceivePreludeError<E> {
+    fn from(value: E) -> Self {
+        ReceivePreludeError::Transport(value)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+
+    use super::*;
+
+    use ufotofu::local_nb::producer::FromSlice;
+
+    #[test]
+    fn empty_producer() {
+        let mut empty_transport = FromSlice::new(&[]);
+
+        smol::block_on(async {
+            let result = receive_prelude::<4, _>(&mut empty_transport).await;
+
+            assert!(matches!(result, Err(ReceivePreludeError::FinishedTooSoon)))
+        });
+    }
+
+    #[test]
+    fn only_power_producer() {
+        let mut only_power_transport = FromSlice::new(&[0_u8]);
+
+        smol::block_on(async {
+            let result = receive_prelude::<4, _>(&mut only_power_transport).await;
+
+            assert!(matches!(result, Err(ReceivePreludeError::FinishedTooSoon)))
+        });
+    }
+
+    #[test]
+    fn invalid_power_producer() {
+        let mut only_power_transport = FromSlice::new(&[65_u8]);
+
+        smol::block_on(async {
+            let result = receive_prelude::<4, _>(&mut only_power_transport).await;
+
+            assert!(matches!(
+                result,
+                Err(ReceivePreludeError::MaxPayloadInvalid)
+            ))
+        });
+    }
+
+    #[test]
+    fn invalid_power_producer_correct_length() {
+        let mut only_power_transport = FromSlice::new(&[65_u8, 0, 0, 0, 0]);
+
+        smol::block_on(async {
+            let result = receive_prelude::<4, _>(&mut only_power_transport).await;
+
+            assert!(matches!(
+                result,
+                Err(ReceivePreludeError::MaxPayloadInvalid)
+            ))
+        });
+    }
+
+    #[test]
+    fn commitment_too_short() {
+        let mut only_power_transport = FromSlice::new(&[0_u8, 0]);
+
+        smol::block_on(async {
+            let result = receive_prelude::<4, _>(&mut only_power_transport).await;
+
+            assert!(matches!(result, Err(ReceivePreludeError::FinishedTooSoon)))
+        });
+    }
+
+    #[test]
+    fn success() {
+        let mut only_power_transport = FromSlice::new(&[1_u8, 1, 2, 3, 4, 5]);
+
+        smol::block_on(async {
+            let result = receive_prelude::<4, _>(&mut only_power_transport).await;
+
+            if let Ok(ready) = result {
+                assert!(ready.maximum_payload_size == 2);
+                assert!(ready.received_commitment == [1, 2, 3, 4]);
+            } else {
+                panic!()
+            }
+        });
+    }
+}
diff --git a/wgps/src/commitment_scheme/send_prelude.rs b/wgps/src/commitment_scheme/send_prelude.rs
new file mode 100644
index 0000000..9ba0f9f
--- /dev/null
+++ b/wgps/src/commitment_scheme/send_prelude.rs
@@ -0,0 +1,18 @@
+use ufotofu::{local_nb::BulkConsumer, nb::ConsumeFullSliceError};
+
+/// Sends the prelude of maximum payload power and what is to be the recipient's `received_commitment` via a transport.
+pub(crate) async fn send_prelude<const CHALLENGE_HASH_LENGTH: usize, C: BulkConsumer<Item = u8>>(
+    max_payload_power: u8,
+    commitment: [u8; CHALLENGE_HASH_LENGTH],
+    transport: &mut C,
+) -> Result<(), C::Error> {
+    transport.consume(max_payload_power).await?;
+
+    if let Err(ConsumeFullSliceError { reason, .. }) =
+        transport.bulk_consume_full_slice(&commitment).await
+    {
+        return Err(reason);
+    }
+
+    Ok(())
+}
diff --git a/wgps/src/lib.rs b/wgps/src/lib.rs
index c66bbff..a33e8e6 100644
--- a/wgps/src/lib.rs
+++ b/wgps/src/lib.rs
@@ -7,8 +7,8 @@ use willow_data_model::{
     ResumptionFailedError, Store, StoreEvent, SubspaceId,
 };
 
-mod ready_transport;
-pub use ready_transport::*;
+mod commitment_scheme;
+pub use commitment_scheme::*;
 
 /// Options to specify how ranges should be partitioned.
 #[derive(Debug, Clone, Copy)]
diff --git a/wgps/src/ready_transport.rs b/wgps/src/ready_transport.rs
deleted file mode 100644
index 1170092..0000000
--- a/wgps/src/ready_transport.rs
+++ /dev/null
@@ -1,188 +0,0 @@
-use either::Either;
-use ufotofu::local_nb::BulkProducer;
-
-/** When things go wrong while trying to make a WGPS transport ready. */
-#[derive(Debug)]
-pub enum ReadyTransportError<E: core::fmt::Display> {
-    /** The transport returned an error of its own. */
-    Transport(E),
-    /** The received max payload power was invalid, i.e. greater than 64. */
-    MaxPayloadInvalid,
-    /** The transport stopped producing bytes before it could be deemed ready. */
-    FinishedTooSoon,
-}
-
-impl<E: core::fmt::Display> core::fmt::Display for ReadyTransportError<E> {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        match self {
-            ReadyTransportError::Transport(e) => write!(f, "{}", e),
-            ReadyTransportError::MaxPayloadInvalid => write!(
-                f,
-                "The received max payload power was invalid, i.e. greater than 64."
-            ),
-            ReadyTransportError::FinishedTooSoon => write!(
-                f,
-                "The transport stopped producing bytes before it could be deemed ready."
-            ),
-        }
-    }
-}
-
-/** The result of intercepting the first few bytes of a WGPS transport. */
-#[derive(Debug)]
-#[allow(dead_code)] // TODO: Remove when this is used.
-pub(crate) struct ReadyTransport<
-    const CHALLENGE_HASH_LENGTH: usize,
-    E: core::fmt::Display,
-    P: BulkProducer<Item = u8, Final = (), Error = E>,
-> {
-    /** The maximum payload size which may be sent without being explicitly requested.*/
-    pub maximum_payload_size: usize,
-    /** The challenge hash of a nonce. */
-    pub received_commitment: [u8; CHALLENGE_HASH_LENGTH],
-    /** A 'ready' transport set to immediately produce encoded WGPS messages. */
-    pub transport: P,
-}
-
-impl<
-        const CHALLENGE_HASH_LENGTH: usize,
-        E: core::fmt::Display,
-        P: BulkProducer<Item = u8, Final = (), Error = E>,
-    > ReadyTransport<CHALLENGE_HASH_LENGTH, E, P>
-{
-    #[allow(dead_code)] // TODO: Remove when this is used.
-    pub(crate) fn transport(&self) -> &P {
-        &self.transport
-    }
-}
-
-/** Given a producer of bytes which is to immediately produce the bytes corresponding to the WGPS' [maximum payload size](https://willowprotocol.org/specs/sync/index.html#peer_max_payload_size) and [received commitment](https://willowprotocol.org/specs/sync/index.html#received_commitment), returns the computed maximum payload size, received commitment, and a 'ready' transport set to produce encoded WGPS messages.
-*/
-#[allow(dead_code)] // TODO: Remove when this is used.
-pub(crate) async fn ready_transport<
-    const CHALLENGE_HASH_LENGTH: usize,
-    E: core::fmt::Display,
-    P: BulkProducer<Item = u8, Final = (), Error = E>,
->(
-    mut transport: P,
-) -> Result<ReadyTransport<CHALLENGE_HASH_LENGTH, E, P>, ReadyTransportError<E>> {
-    let maximum_payload_power = match transport.produce().await? {
-        Either::Left(byte) => byte,
-        Either::Right(_) => return Err(ReadyTransportError::FinishedTooSoon),
-    };
-
-    if maximum_payload_power > 64 {
-        return Err(ReadyTransportError::MaxPayloadInvalid);
-    }
-
-    let maximum_payload_size = 2_usize.pow(maximum_payload_power as u32);
-
-    let mut received_commitment = [0_u8; CHALLENGE_HASH_LENGTH];
-
-    if let Err(e) = transport
-        .bulk_overwrite_full_slice(&mut received_commitment)
-        .await
-    {
-        match e.reason {
-            Either::Left(_) => return Err(ReadyTransportError::FinishedTooSoon),
-            Either::Right(e) => return Err(ReadyTransportError::Transport(e)),
-        }
-    };
-
-    Ok(ReadyTransport {
-        maximum_payload_size,
-        received_commitment,
-        transport,
-    })
-}
-
-impl<E: core::fmt::Display> From<E> for ReadyTransportError<E> {
-    fn from(value: E) -> Self {
-        ReadyTransportError::Transport(value)
-    }
-}
-
-#[cfg(test)]
-mod tests {
-
-    use super::*;
-
-    use ufotofu::local_nb::producer::FromSlice;
-
-    #[test]
-    fn empty_producer() {
-        let empty_transport = FromSlice::new(&[]);
-
-        smol::block_on(async {
-            let result = ready_transport::<4, _, _>(empty_transport).await;
-
-            assert!(matches!(result, Err(ReadyTransportError::FinishedTooSoon)))
-        });
-    }
-
-    #[test]
-    fn only_power_producer() {
-        let only_power_transport = FromSlice::new(&[0_u8]);
-
-        smol::block_on(async {
-            let result = ready_transport::<4, _, _>(only_power_transport).await;
-
-            assert!(matches!(result, Err(ReadyTransportError::FinishedTooSoon)))
-        });
-    }
-
-    #[test]
-    fn invalid_power_producer() {
-        let only_power_transport = FromSlice::new(&[65_u8]);
-
-        smol::block_on(async {
-            let result = ready_transport::<4, _, _>(only_power_transport).await;
-
-            assert!(matches!(
-                result,
-                Err(ReadyTransportError::MaxPayloadInvalid)
-            ))
-        });
-    }
-
-    #[test]
-    fn invalid_power_producer_correct_length() {
-        let only_power_transport = FromSlice::new(&[65_u8, 0, 0, 0, 0]);
-
-        smol::block_on(async {
-            let result = ready_transport::<4, _, _>(only_power_transport).await;
-
-            assert!(matches!(
-                result,
-                Err(ReadyTransportError::MaxPayloadInvalid)
-            ))
-        });
-    }
-
-    #[test]
-    fn commitment_too_short() {
-        let only_power_transport = FromSlice::new(&[0_u8, 0]);
-
-        smol::block_on(async {
-            let result = ready_transport::<4, _, _>(only_power_transport).await;
-
-            assert!(matches!(result, Err(ReadyTransportError::FinishedTooSoon)))
-        });
-    }
-
-    #[test]
-    fn success() {
-        let only_power_transport = FromSlice::new(&[1_u8, 1, 2, 3, 4, 5]);
-
-        smol::block_on(async {
-            let result = ready_transport::<4, _, _>(only_power_transport).await;
-
-            if let Ok(ready) = result {
-                assert!(ready.maximum_payload_size == 2);
-                assert!(ready.received_commitment == [1, 2, 3, 4]);
-            } else {
-                panic!()
-            }
-        });
-    }
-}

From 859cae9b3933d7d730bc739467126631a678edab Mon Sep 17 00:00:00 2001
From: Sam Gwilym <sam@gwil.garden>
Date: Mon, 11 Nov 2024 17:03:12 +0100
Subject: [PATCH 2/6] smelly

---
 Cargo.lock                                    | 98 ++++++++++++++++++-
 wgps/Cargo.toml                               |  1 +
 wgps/src/commitment_scheme/execute_prelude.rs | 32 ++++++
 wgps/src/commitment_scheme/mod.rs             |  1 +
 wgps/src/commitment_scheme/receive_prelude.rs |  8 +-
 5 files changed, 130 insertions(+), 10 deletions(-)
 create mode 100644 wgps/src/commitment_scheme/execute_prelude.rs

diff --git a/Cargo.lock b/Cargo.lock
index 004cac3..237a0c8 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -267,17 +267,53 @@ version = "2.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9fc0510504f03c51ada170672ac806f1f105a88aa97a5281117e1ddc3368e51a"
 
+[[package]]
+name = "futures"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "65bc07b1a8bc7c85c5f2e110c476c7389b4554ba72af57d8445ea63a576b0876"
+dependencies = [
+ "futures-channel",
+ "futures-core",
+ "futures-executor",
+ "futures-io",
+ "futures-sink",
+ "futures-task",
+ "futures-util",
+]
+
+[[package]]
+name = "futures-channel"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2dff15bf788c671c1934e366d07e30c1814a8ef514e1af724a602e8a2fbe1b10"
+dependencies = [
+ "futures-core",
+ "futures-sink",
+]
+
 [[package]]
 name = "futures-core"
-version = "0.3.30"
+version = "0.3.31"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dfc6580bb841c5a68e9ef15c77ccc837b40a7504914d52e47b8b0e9bbda25a1d"
+checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e"
+
+[[package]]
+name = "futures-executor"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1e28d1d997f585e54aebc3f97d39e72338912123a67330d723fdbb564d646c9f"
+dependencies = [
+ "futures-core",
+ "futures-task",
+ "futures-util",
+]
 
 [[package]]
 name = "futures-io"
-version = "0.3.30"
+version = "0.3.31"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a44623e20b9681a318efdd71c299b6b222ed6f231972bfe2f224ebad6311f0c1"
+checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6"
 
 [[package]]
 name = "futures-lite"
@@ -292,6 +328,47 @@ dependencies = [
  "pin-project-lite",
 ]
 
+[[package]]
+name = "futures-macro"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "futures-sink"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e575fab7d1e0dcb8d0c7bcf9a63ee213816ab51902e6d244a95819acacf1d4f7"
+
+[[package]]
+name = "futures-task"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f90f7dce0722e95104fcb095585910c0977252f286e354b5e3bd38902cd99988"
+
+[[package]]
+name = "futures-util"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81"
+dependencies = [
+ "futures-channel",
+ "futures-core",
+ "futures-io",
+ "futures-macro",
+ "futures-sink",
+ "futures-task",
+ "memchr",
+ "pin-project-lite",
+ "pin-utils",
+ "slab",
+]
+
 [[package]]
 name = "hermit-abi"
 version = "0.4.0"
@@ -349,6 +426,12 @@ dependencies = [
  "willow-encoding",
 ]
 
+[[package]]
+name = "memchr"
+version = "2.7.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "78ca9ab1a0babb1e7d5695e3530886289c18cf2f87ec19a575a0abdce112e3a3"
+
 [[package]]
 name = "once_cell"
 version = "1.19.0"
@@ -367,6 +450,12 @@ version = "0.2.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bda66fc9667c18cb2758a2ac84d1167245054bcf85d5d1aaa6923f45801bdd02"
 
+[[package]]
+name = "pin-utils"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
+
 [[package]]
 name = "piper"
 version = "0.2.3"
@@ -532,6 +621,7 @@ name = "wgps"
 version = "0.1.0"
 dependencies = [
  "either",
+ "futures",
  "smol",
  "ufotofu",
  "willow-data-model",
diff --git a/wgps/Cargo.toml b/wgps/Cargo.toml
index 068365c..67ba1ae 100644
--- a/wgps/Cargo.toml
+++ b/wgps/Cargo.toml
@@ -6,6 +6,7 @@ edition = "2021"
 [dependencies]
 willow-data-model = { path = "../data-model", version = "0.1.0" }
 ufotofu = { version = "0.4.2", features = ["std"] }
+futures = { version = "0.3.31" }
 either = "1.10.0"
 
 [dev-dependencies]
diff --git a/wgps/src/commitment_scheme/execute_prelude.rs b/wgps/src/commitment_scheme/execute_prelude.rs
new file mode 100644
index 0000000..1d56f48
--- /dev/null
+++ b/wgps/src/commitment_scheme/execute_prelude.rs
@@ -0,0 +1,32 @@
+use futures::future::{select, Either};
+
+// send our own commitment reveal message only after we have received the prelude.
+
+// everything that needs the challenge has to wait until we have received their commitment reveal message.
+
+use ufotofu::local_nb::{BulkConsumer, BulkProducer};
+
+use super::{receive_prelude::receive_prelude, send_prelude::send_prelude};
+
+async fn execute_prelude<
+    const CHALLENGE_HASH_LENGTH: usize,
+    C: BulkConsumer<Item = u8>,
+    P: BulkProducer<Item = u8>,
+>(
+    max_payload_size: u8,
+    commitment: [u8; CHALLENGE_HASH_LENGTH],
+    consumer: &mut C,
+    producer: &mut P,
+) {
+    // This should be an async block doing more!
+    // Should also send a commitment reveal message after having received the prelude.
+    let receive_fut = Box::pin(receive_prelude::<CHALLENGE_HASH_LENGTH, _>(producer));
+    let send_fut = Box::pin(send_prelude(max_payload_size, commitment, consumer));
+
+    match select(receive_fut, send_fut).await {
+        Either::Left((Ok(received), mut send_fut)) => todo!(),
+        Either::Left((Err(error), mut send_fut)) => todo!(),
+        Either::Right((Err(error), mut receive_fut)) => todo!(),
+        Either::Right((Ok(()), mut receive_fut)) => todo!(),
+    }
+}
diff --git a/wgps/src/commitment_scheme/mod.rs b/wgps/src/commitment_scheme/mod.rs
index 9e8e6d7..935ffda 100644
--- a/wgps/src/commitment_scheme/mod.rs
+++ b/wgps/src/commitment_scheme/mod.rs
@@ -1,2 +1,3 @@
+pub mod execute_prelude;
 pub mod receive_prelude;
 pub mod send_prelude;
diff --git a/wgps/src/commitment_scheme/receive_prelude.rs b/wgps/src/commitment_scheme/receive_prelude.rs
index 2a7e93c..13686c6 100644
--- a/wgps/src/commitment_scheme/receive_prelude.rs
+++ b/wgps/src/commitment_scheme/receive_prelude.rs
@@ -39,17 +39,13 @@ pub(crate) struct ReceivedPrelude<const CHALLENGE_HASH_LENGTH: usize> {
 }
 
 /// Given a producer of bytes which is to immediately produce the bytes corresponding to the WGPS' [maximum payload size](https://willowprotocol.org/specs/sync/index.html#peer_max_payload_size) and [received commitment](https://willowprotocol.org/specs/sync/index.html#received_commitment), returns the computed maximum payload size, received commitment, and a 'ready' transport set to produce encoded WGPS messages.
-
 #[allow(dead_code)] // TODO: Remove when this is used.
 pub(crate) async fn receive_prelude<
     const CHALLENGE_HASH_LENGTH: usize,
     P: BulkProducer<Item = u8>,
 >(
     transport: &mut P,
-) -> Result<ReceivedPrelude<CHALLENGE_HASH_LENGTH>, ReceivePreludeError<P::Error>>
-where
-    P::Error: core::fmt::Display,
-{
+) -> Result<ReceivedPrelude<CHALLENGE_HASH_LENGTH>, ReceivePreludeError<P::Error>> {
     let maximum_payload_power = match transport.produce().await? {
         Either::Left(byte) => byte,
         Either::Right(_) => return Err(ReceivePreludeError::FinishedTooSoon),
@@ -79,7 +75,7 @@ where
     })
 }
 
-impl<E: core::fmt::Display> From<E> for ReceivePreludeError<E> {
+impl<E> From<E> for ReceivePreludeError<E> {
     fn from(value: E) -> Self {
         ReceivePreludeError::Transport(value)
     }

From c9e05fc2d8ba9280ac8a8b789c4904355f68eff0 Mon Sep 17 00:00:00 2001
From: Sam Gwilym <sam@gwil.garden>
Date: Tue, 19 Nov 2024 14:30:34 +0100
Subject: [PATCH 3/6] Add execute_prelude function

---
 Cargo.lock                                    |  1 +
 wgps/Cargo.toml                               |  1 +
 wgps/src/commitment_scheme/execute_prelude.rs | 64 ++++++++++++++-----
 wgps/src/lib.rs                               |  6 ++
 wgps/src/messages/commitment_reveal.rs        | 21 ++++++
 wgps/src/messages/mod.rs                      |  2 +
 wgps/src/parameters.rs                        |  3 +
 7 files changed, 83 insertions(+), 15 deletions(-)
 create mode 100644 wgps/src/messages/commitment_reveal.rs
 create mode 100644 wgps/src/messages/mod.rs
 create mode 100644 wgps/src/parameters.rs

diff --git a/Cargo.lock b/Cargo.lock
index 237a0c8..4ca7838 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -625,6 +625,7 @@ dependencies = [
  "smol",
  "ufotofu",
  "willow-data-model",
+ "willow-encoding",
 ]
 
 [[package]]
diff --git a/wgps/Cargo.toml b/wgps/Cargo.toml
index 67ba1ae..6e89fa4 100644
--- a/wgps/Cargo.toml
+++ b/wgps/Cargo.toml
@@ -5,6 +5,7 @@ edition = "2021"
 
 [dependencies]
 willow-data-model = { path = "../data-model", version = "0.1.0" }
+willow-encoding = { path = "../encoding", version = "0.1.0" }
 ufotofu = { version = "0.4.2", features = ["std"] }
 futures = { version = "0.3.31" }
 either = "1.10.0"
diff --git a/wgps/src/commitment_scheme/execute_prelude.rs b/wgps/src/commitment_scheme/execute_prelude.rs
index 1d56f48..681dc42 100644
--- a/wgps/src/commitment_scheme/execute_prelude.rs
+++ b/wgps/src/commitment_scheme/execute_prelude.rs
@@ -1,32 +1,66 @@
-use futures::future::{select, Either};
+use futures::future::{select, try_select, Either};
 
 // send our own commitment reveal message only after we have received the prelude.
 
 // everything that needs the challenge has to wait until we have received their commitment reveal message.
 
 use ufotofu::local_nb::{BulkConsumer, BulkProducer};
+use willow_encoding::Encodable;
 
-use super::{receive_prelude::receive_prelude, send_prelude::send_prelude};
+use crate::{ChallengeHash, CommitmentReveal};
 
-async fn execute_prelude<
+use super::{
+    receive_prelude::{receive_prelude, ReceivePreludeError, ReceivedPrelude},
+    send_prelude::send_prelude,
+};
+
+pub(crate) enum ExecutePreludeError<E> {
+    ReceiveError(ReceivePreludeError<E>),
+    SendError(E),
+}
+
+impl<E> From<ReceivePreludeError<E>> for ExecutePreludeError<E> {
+    fn from(value: ReceivePreludeError<E>) -> Self {
+        ExecutePreludeError::ReceiveError(value)
+    }
+}
+
+impl<E> From<E> for ExecutePreludeError<E> {
+    fn from(value: E) -> Self {
+        ExecutePreludeError::SendError(value)
+    }
+}
+
+/// Given a consumer and producer, send a max payload size and commitment, and wait for the other side's corresponding `ReceivedPrelude`. Then send a `CommitmentReveal` message, before finally returning the received prelude.
+///
+/// Attention: waiting for the other peer's prelude means that we delay sending our first messages, even though technically we would be allowed to do that. PRs welcome.
+pub(crate) async fn execute_prelude<
+    const CHALLENGE_LENGTH: usize,
     const CHALLENGE_HASH_LENGTH: usize,
-    C: BulkConsumer<Item = u8>,
-    P: BulkProducer<Item = u8>,
+    CH: ChallengeHash<CHALLENGE_LENGTH, CHALLENGE_HASH_LENGTH>,
+    E,
+    C: BulkConsumer<Item = u8, Error = E>,
+    P: BulkProducer<Item = u8, Error = E>,
 >(
     max_payload_size: u8,
-    commitment: [u8; CHALLENGE_HASH_LENGTH],
+    our_nonce: [u8; CHALLENGE_LENGTH],
     consumer: &mut C,
     producer: &mut P,
-) {
-    // This should be an async block doing more!
-    // Should also send a commitment reveal message after having received the prelude.
+) -> Result<ReceivedPrelude<CHALLENGE_HASH_LENGTH>, ExecutePreludeError<E>> {
+    let commitment = CH::hash(our_nonce);
+
     let receive_fut = Box::pin(receive_prelude::<CHALLENGE_HASH_LENGTH, _>(producer));
     let send_fut = Box::pin(send_prelude(max_payload_size, commitment, consumer));
 
-    match select(receive_fut, send_fut).await {
-        Either::Left((Ok(received), mut send_fut)) => todo!(),
-        Either::Left((Err(error), mut send_fut)) => todo!(),
-        Either::Right((Err(error), mut receive_fut)) => todo!(),
-        Either::Right((Ok(()), mut receive_fut)) => todo!(),
-    }
+    let (received_prelude, ()) = match try_select(receive_fut, send_fut).await {
+        Ok(Either::Left((received, send_fut))) => (received, send_fut.await?),
+        Ok(Either::Right(((), receive_fut))) => (receive_fut.await?, ()),
+        Err(Either::Left((error, _))) => return Err(error.into()),
+        Err(Either::Right((error, _))) => return Err(error.into()),
+    };
+
+    let msg = CommitmentReveal { nonce: our_nonce };
+    msg.encode(consumer).await?;
+
+    Ok(received_prelude)
 }
diff --git a/wgps/src/lib.rs b/wgps/src/lib.rs
index a33e8e6..c320a14 100644
--- a/wgps/src/lib.rs
+++ b/wgps/src/lib.rs
@@ -7,6 +7,12 @@ use willow_data_model::{
     ResumptionFailedError, Store, StoreEvent, SubspaceId,
 };
 
+mod parameters;
+pub use parameters::*;
+
+mod messages;
+pub use messages::*;
+
 mod commitment_scheme;
 pub use commitment_scheme::*;
 
diff --git a/wgps/src/messages/commitment_reveal.rs b/wgps/src/messages/commitment_reveal.rs
new file mode 100644
index 0000000..33ec069
--- /dev/null
+++ b/wgps/src/messages/commitment_reveal.rs
@@ -0,0 +1,21 @@
+use ufotofu::local_nb::BulkConsumer;
+use willow_encoding::Encodable;
+
+pub struct CommitmentReveal<const CHALLENGE_LENGTH: usize> {
+    pub nonce: [u8; CHALLENGE_LENGTH],
+}
+
+impl<const CHALLENGE_LENGTH: usize> Encodable for CommitmentReveal<CHALLENGE_LENGTH> {
+    async fn encode<Consumer>(&self, consumer: &mut Consumer) -> Result<(), Consumer::Error>
+    where
+        Consumer: BulkConsumer<Item = u8>,
+    {
+        consumer.consume(0x0).await?;
+        consumer
+            .bulk_consume_full_slice(&self.nonce[..])
+            .await
+            .map_err(|err| err.reason)?;
+
+        Ok(())
+    }
+}
diff --git a/wgps/src/messages/mod.rs b/wgps/src/messages/mod.rs
new file mode 100644
index 0000000..54e87d5
--- /dev/null
+++ b/wgps/src/messages/mod.rs
@@ -0,0 +1,2 @@
+mod commitment_reveal;
+pub use commitment_reveal::*;
diff --git a/wgps/src/parameters.rs b/wgps/src/parameters.rs
new file mode 100644
index 0000000..cdea2db
--- /dev/null
+++ b/wgps/src/parameters.rs
@@ -0,0 +1,3 @@
+pub trait ChallengeHash<const CHALLENGE_LENGTH: usize, const CHALLENGE_HASH_LENGTH: usize> {
+    fn hash(nonce: [u8; CHALLENGE_LENGTH]) -> [u8; CHALLENGE_HASH_LENGTH];
+}

From e80edde148fe397df5f80a9242773a39245acaaf Mon Sep 17 00:00:00 2001
From: Sam Gwilym <sam@gwil.garden>
Date: Tue, 19 Nov 2024 14:55:37 +0100
Subject: [PATCH 4/6] Add_sync_with_peer

---
 wgps/src/commitment_scheme/execute_prelude.rs |  6 +--
 wgps/src/lib.rs                               | 44 ++++++++++++++++++-
 2 files changed, 46 insertions(+), 4 deletions(-)

diff --git a/wgps/src/commitment_scheme/execute_prelude.rs b/wgps/src/commitment_scheme/execute_prelude.rs
index 681dc42..37e066b 100644
--- a/wgps/src/commitment_scheme/execute_prelude.rs
+++ b/wgps/src/commitment_scheme/execute_prelude.rs
@@ -14,7 +14,7 @@ use super::{
     send_prelude::send_prelude,
 };
 
-pub(crate) enum ExecutePreludeError<E> {
+pub enum ExecutePreludeError<E> {
     ReceiveError(ReceivePreludeError<E>),
     SendError(E),
 }
@@ -42,7 +42,7 @@ pub(crate) async fn execute_prelude<
     C: BulkConsumer<Item = u8, Error = E>,
     P: BulkProducer<Item = u8, Error = E>,
 >(
-    max_payload_size: u8,
+    max_payload_power: u8,
     our_nonce: [u8; CHALLENGE_LENGTH],
     consumer: &mut C,
     producer: &mut P,
@@ -50,7 +50,7 @@ pub(crate) async fn execute_prelude<
     let commitment = CH::hash(our_nonce);
 
     let receive_fut = Box::pin(receive_prelude::<CHALLENGE_HASH_LENGTH, _>(producer));
-    let send_fut = Box::pin(send_prelude(max_payload_size, commitment, consumer));
+    let send_fut = Box::pin(send_prelude(max_payload_power, commitment, consumer));
 
     let (received_prelude, ()) = match try_select(receive_fut, send_fut).await {
         Ok(Either::Left((received, send_fut))) => (received, send_fut.await?),
diff --git a/wgps/src/lib.rs b/wgps/src/lib.rs
index c320a14..1f06e48 100644
--- a/wgps/src/lib.rs
+++ b/wgps/src/lib.rs
@@ -1,6 +1,7 @@
 use std::future::Future;
 
-use ufotofu::local_nb::Producer;
+use execute_prelude::ExecutePreludeError;
+use ufotofu::local_nb::{BulkConsumer, BulkProducer, Producer};
 use willow_data_model::{
     grouping::{AreaOfInterest, Range3d},
     AuthorisationToken, LengthyAuthorisedEntry, NamespaceId, PayloadDigest, QueryIgnoreParams,
@@ -14,8 +15,49 @@ mod messages;
 pub use messages::*;
 
 mod commitment_scheme;
+use commitment_scheme::execute_prelude::execute_prelude;
 pub use commitment_scheme::*;
 
+pub enum WgpsError<E> {
+    Prelude(ExecutePreludeError<E>),
+}
+
+impl<E> From<ExecutePreludeError<E>> for WgpsError<E> {
+    fn from(value: ExecutePreludeError<E>) -> Self {
+        Self::Prelude(value)
+    }
+}
+
+pub struct SyncOptions<const CHALLENGE_LENGTH: usize> {
+    max_payload_power: u8,
+    challenge_nonce: [u8; CHALLENGE_LENGTH],
+}
+
+pub async fn sync_with_peer<
+    const CHALLENGE_LENGTH: usize,
+    const CHALLENGE_HASH_LENGTH: usize,
+    CH: ChallengeHash<CHALLENGE_LENGTH, CHALLENGE_HASH_LENGTH>,
+    E,
+    C: BulkConsumer<Item = u8, Error = E>,
+    P: BulkProducer<Item = u8, Error = E>,
+>(
+    options: &SyncOptions<CHALLENGE_LENGTH>,
+    mut consumer: C,
+    mut producer: P,
+) -> Result<(), WgpsError<E>> {
+    execute_prelude::<CHALLENGE_LENGTH, CHALLENGE_HASH_LENGTH, CH, _, _, _>(
+        options.max_payload_power,
+        options.challenge_nonce,
+        &mut consumer,
+        &mut producer,
+    )
+    .await?;
+
+    // TODO: The rest of the WGPS. No probs!
+
+    Ok(())
+}
+
 /// Options to specify how ranges should be partitioned.
 #[derive(Debug, Clone, Copy)]
 pub struct PartitionOpts {

From 3db8f38ee248cfbe0f0550c4a3df0bd95709a90e Mon Sep 17 00:00:00 2001
From: Sam Gwilym <sam@gwil.garden>
Date: Tue, 19 Nov 2024 15:37:24 +0100
Subject: [PATCH 5/6] Add some more error stuff.

---
 wgps/src/commitment_scheme/execute_prelude.rs | 14 ++++++++++++++
 wgps/src/lib.rs                               |  9 +++++++++
 2 files changed, 23 insertions(+)

diff --git a/wgps/src/commitment_scheme/execute_prelude.rs b/wgps/src/commitment_scheme/execute_prelude.rs
index 37e066b..04d6c3d 100644
--- a/wgps/src/commitment_scheme/execute_prelude.rs
+++ b/wgps/src/commitment_scheme/execute_prelude.rs
@@ -14,8 +14,11 @@ use super::{
     send_prelude::send_prelude,
 };
 
+/// An error which only occurs during the initial phase of a WGPS session.
 pub enum ExecutePreludeError<E> {
+    /// There was a problem receiving their prelude.
     ReceiveError(ReceivePreludeError<E>),
+    /// There was a problem sending our prelude.
     SendError(E),
 }
 
@@ -31,6 +34,17 @@ impl<E> From<E> for ExecutePreludeError<E> {
     }
 }
 
+impl<E: core::fmt::Display> core::fmt::Display for ExecutePreludeError<E> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            ExecutePreludeError::ReceiveError(receive_prelude_error) => {
+                write!(f, "{}", receive_prelude_error)
+            }
+            ExecutePreludeError::SendError(error) => write!(f, "{}", error),
+        }
+    }
+}
+
 /// Given a consumer and producer, send a max payload size and commitment, and wait for the other side's corresponding `ReceivedPrelude`. Then send a `CommitmentReveal` message, before finally returning the received prelude.
 ///
 /// Attention: waiting for the other peer's prelude means that we delay sending our first messages, even though technically we would be allowed to do that. PRs welcome.
diff --git a/wgps/src/lib.rs b/wgps/src/lib.rs
index 1f06e48..7a7a98f 100644
--- a/wgps/src/lib.rs
+++ b/wgps/src/lib.rs
@@ -18,6 +18,7 @@ mod commitment_scheme;
 use commitment_scheme::execute_prelude::execute_prelude;
 pub use commitment_scheme::*;
 
+/// An error which can occur during a WGPS synchronisation session.
 pub enum WgpsError<E> {
     Prelude(ExecutePreludeError<E>),
 }
@@ -28,6 +29,14 @@ impl<E> From<ExecutePreludeError<E>> for WgpsError<E> {
     }
 }
 
+impl<E: core::fmt::Display> core::fmt::Display for WgpsError<E> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            WgpsError::Prelude(execute_prelude_error) => write!(f, "{}", execute_prelude_error),
+        }
+    }
+}
+
 pub struct SyncOptions<const CHALLENGE_LENGTH: usize> {
     max_payload_power: u8,
     challenge_nonce: [u8; CHALLENGE_LENGTH],

From c9ea44feaa2c0da190fad333584b04d04422160c Mon Sep 17 00:00:00 2001
From: Sam Gwilym <sam@gwil.garden>
Date: Wed, 20 Nov 2024 11:05:00 +0100
Subject: [PATCH 6/6] Fix lint

---
 wgps/src/commitment_scheme/execute_prelude.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/wgps/src/commitment_scheme/execute_prelude.rs b/wgps/src/commitment_scheme/execute_prelude.rs
index 04d6c3d..b5531a7 100644
--- a/wgps/src/commitment_scheme/execute_prelude.rs
+++ b/wgps/src/commitment_scheme/execute_prelude.rs
@@ -1,4 +1,4 @@
-use futures::future::{select, try_select, Either};
+use futures::future::{try_select, Either};
 
 // send our own commitment reveal message only after we have received the prelude.