Repository: p2panda/aquadoggo (language: rust)

Issue:
De- / Serialize replication messages



Discussion / hints:

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 24998e759..26d78d405 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -20,6 +20,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Add custom validation to GraphQL scalar types [#318](https://github.com/p2panda/aquadoggo/pull/318)
 - Introduce property tests for GraphQL query API with `proptest` [#338](https://github.com/p2panda/aquadoggo/pull/338)
 - Replication protocol session manager [#363](https://github.com/p2panda/aquadoggo/pull/363)
+- Replication message de- / serialization [#375](https://github.com/p2panda/aquadoggo/pull/375)
 
 ### Changed
 
diff --git a/aquadoggo/src/replication/errors.rs b/aquadoggo/src/replication/errors.rs
index ba7110b0b..b11028e00 100644
--- a/aquadoggo/src/replication/errors.rs
+++ b/aquadoggo/src/replication/errors.rs
@@ -13,3 +13,12 @@ pub enum ReplicationError {
     #[error("Tried to initialise duplicate outbound replication session with id {0}")]
     DuplicateOutboundRequest(u64),
 }
+
+#[derive(Error, Debug)]
+pub enum TargetSetError {
+    #[error("Target set does not contain any schema ids")]
+    ZeroSchemaIds,
+
+    #[error("Target set contains unsorted or duplicate schema ids")]
+    UnsortedSchemaIds,
+}
diff --git a/aquadoggo/src/replication/manager.rs b/aquadoggo/src/replication/manager.rs
index b289f92bd..7c1e360de 100644
--- a/aquadoggo/src/replication/manager.rs
+++ b/aquadoggo/src/replication/manager.rs
@@ -5,7 +5,7 @@ use std::collections::HashMap;
 use anyhow::Result;
 
 use crate::replication::errors::ReplicationError;
-use crate::replication::{Mode, Session, SessionId, SessionState, SyncMessage, TargetSet};
+use crate::replication::{Message, Mode, Session, SessionId, SessionState, SyncMessage, TargetSet};
 
 pub const INITIAL_SESSION_ID: SessionId = 0;
 
@@ -181,39 +181,31 @@ where
     pub fn handle_message(
         &mut self,
         remote_peer: &P,
-        message: &SyncMessage,
+        sync_message: &SyncMessage,
     ) -> Result<(), ReplicationError> {
-        match message {
-            SyncMessage::SyncRequest(mode, session_id, target_set) => {
-                self.handle_sync_request(remote_peer, mode, session_id, target_set)
+        match sync_message.message() {
+            Message::SyncRequest(mode, target_set) => {
+                self.handle_sync_request(remote_peer, mode, &sync_message.session_id(), target_set)
             }
-            SyncMessage::Other => todo!(),
+            _ => todo!(),
         }
     }
 }
 
 #[cfg(test)]
 mod tests {
-    use p2panda_rs::schema::{SchemaId, SchemaName};
-    use p2panda_rs::test_utils::fixtures::random_document_view_id;
     use rstest::rstest;
 
     use crate::replication::errors::ReplicationError;
+    use crate::replication::message::Message;
     use crate::replication::{Mode, SyncMessage, TargetSet};
+    use crate::test_utils::helpers::random_target_set;
 
     use super::{SyncManager, INITIAL_SESSION_ID};
 
     const PEER_ID_LOCAL: &'static str = "local";
     const PEER_ID_REMOTE: &'static str = "remote";
 
-    #[rstest::fixture]
-    fn random_target_set() -> TargetSet {
-        let document_view_id = random_document_view_id();
-        let schema_id =
-            SchemaId::new_application(&SchemaName::new("messages").unwrap(), &document_view_id);
-        TargetSet::new(&[schema_id])
-    }
-
     #[rstest]
     fn initiate_outbound_session(
         #[from(random_target_set)] target_set_1: TargetSet,
@@ -243,16 +235,16 @@ mod tests {
     ) {
         let mut manager = SyncManager::new(PEER_ID_LOCAL);
 
-        let message = SyncMessage::SyncRequest(Mode::Naive, 0, target_set_1.clone());
+        let message = SyncMessage::new(0, Message::SyncRequest(Mode::Naive, target_set_1.clone()));
         let result = manager.handle_message(&PEER_ID_REMOTE, &message);
         assert!(result.is_ok());
 
-        let message = SyncMessage::SyncRequest(Mode::Naive, 1, target_set_2.clone());
+        let message = SyncMessage::new(1, Message::SyncRequest(Mode::Naive, target_set_2.clone()));
         let result = manager.handle_message(&PEER_ID_REMOTE, &message);
         assert!(result.is_ok());
 
         // Reject attempt to create session again
-        let message = SyncMessage::SyncRequest(Mode::Naive, 0, target_set_1.clone());
+        let message = SyncMessage::new(0, Message::SyncRequest(Mode::Naive, target_set_1.clone()));
         let result = manager.handle_message(&PEER_ID_REMOTE, &message);
         assert!(matches!(
             result,
@@ -260,7 +252,7 @@ mod tests {
         ));
 
         // Reject different session concerning same target set
-        let message = SyncMessage::SyncRequest(Mode::Naive, 2, target_set_2.clone());
+        let message = SyncMessage::new(2, Message::SyncRequest(Mode::Naive, target_set_2.clone()));
         let result = manager.handle_message(&PEER_ID_REMOTE, &message);
         assert!(matches!(
             result,
@@ -272,14 +264,19 @@ mod tests {
     fn inbound_checks_supported_mode(#[from(random_target_set)] target_set: TargetSet) {
         // Should not fail when requesting supported replication mode
         let mut manager = SyncManager::new(PEER_ID_LOCAL);
-        let message = SyncMessage::SyncRequest(Mode::Naive, INITIAL_SESSION_ID, target_set.clone());
+        let message = SyncMessage::new(
+            INITIAL_SESSION_ID,
+            Message::SyncRequest(Mode::Naive, target_set.clone()),
+        );
         let result = manager.handle_message(&PEER_ID_REMOTE, &message);
         assert!(result.is_ok());
 
         // Should fail when requesting unsupported replication mode
         let mut manager = SyncManager::new(PEER_ID_LOCAL);
-        let message =
-            SyncMessage::SyncRequest(Mode::SetReconciliation, INITIAL_SESSION_ID, target_set);
+        let message = SyncMessage::new(
+            INITIAL_SESSION_ID,
+            Message::SyncRequest(Mode::SetReconciliation, target_set.clone()),
+        );
         let result = manager.handle_message(&PEER_ID_REMOTE, &message);
         assert!(result.is_err());
     }
diff --git a/aquadoggo/src/replication/message.rs b/aquadoggo/src/replication/message.rs
index 958e2c205..3d25b4404 100644
--- a/aquadoggo/src/replication/message.rs
+++ b/aquadoggo/src/replication/message.rs
@@ -1,15 +1,220 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+use std::fmt;
+
+use serde::de::Visitor;
+use serde::ser::SerializeSeq;
+use serde::{Deserialize, Serialize};
+
 use crate::replication::{Mode, SessionId, TargetSet};
 
-#[derive(Debug)]
-pub enum SyncMessage {
-    SyncRequest(Mode, SessionId, TargetSet),
-    Other,
-}
+pub const SYNC_REQUEST_TYPE: MessageType = 0;
+pub const ENTRY_TYPE: MessageType = 8;
+pub const SYNC_DONE_TYPE: MessageType = 9;
 
+pub type MessageType = u64;
+
+// @TODO
 #[derive(Clone, Debug)]
 pub enum StrategyMessage {
     Have,
     Entry,
 }
+
+#[derive(Debug, Clone, Eq, PartialEq)]
+pub enum Message {
+    SyncRequest(Mode, TargetSet),
+    Entry(Vec<u8>, Vec<u8>),
+    SyncDone(bool),
+}
+
+impl Message {
+    pub fn message_type(&self) -> MessageType {
+        match self {
+            Message::SyncRequest { .. } => SYNC_REQUEST_TYPE,
+            Message::Entry(_, _) => ENTRY_TYPE,
+            Message::SyncDone(_) => SYNC_DONE_TYPE,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Eq, PartialEq)]
+pub struct SyncMessage(SessionId, Message);
+
+impl SyncMessage {
+    pub fn new(session_id: SessionId, message: Message) -> Self {
+        Self(session_id, message)
+    }
+
+    pub fn message_type(&self) -> MessageType {
+        self.1.message_type()
+    }
+
+    pub fn session_id(&self) -> SessionId {
+        self.0
+    }
+
+    pub fn message(&self) -> &Message {
+        &self.1
+    }
+}
+
+impl Serialize for SyncMessage {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        // Always encode message type and session id first
+        let serialize_header = |mut seq: <S as serde::Serializer>::SerializeSeq| -> Result<<S as serde::Serializer>::SerializeSeq, S::Error> {
+            seq.serialize_element(&self.message_type())?;
+            seq.serialize_element(&self.session_id())?;
+            Ok(seq)
+        };
+
+        match self.message() {
+            Message::SyncRequest(mode, target_set) => {
+                let mut seq = serialize_header(serializer.serialize_seq(Some(4))?)?;
+                seq.serialize_element(mode)?;
+                seq.serialize_element(target_set)?;
+                seq.end()
+            }
+            Message::Entry(entry_bytes, operation_bytes) => {
+                let mut seq = serialize_header(serializer.serialize_seq(Some(4))?)?;
+                seq.serialize_element(entry_bytes)?;
+                seq.serialize_element(operation_bytes)?;
+                seq.end()
+            }
+            Message::SyncDone(live_mode) => {
+                let mut seq = serialize_header(serializer.serialize_seq(Some(3))?)?;
+                seq.serialize_element(live_mode)?;
+                seq.end()
+            }
+        }
+    }
+}
+
+impl<'de> Deserialize<'de> for SyncMessage {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        struct SyncMessageVisitor;
+
+        impl<'de> Visitor<'de> for SyncMessageVisitor {
+            type Value = SyncMessage;
+
+            fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
+                formatter.write_str("p2panda replication message")
+            }
+
+            fn visit_seq<A>(self, mut seq: A) -> Result<Self::Value, A::Error>
+            where
+                A: serde::de::SeqAccess<'de>,
+            {
+                let message_type: MessageType = seq.next_element()?.ok_or_else(|| {
+                    serde::de::Error::custom("missing message type in replication message")
+                })?;
+
+                let session_id: SessionId = seq.next_element()?.ok_or_else(|| {
+                    serde::de::Error::custom("missing session id in replication message")
+                })?;
+
+                let message = if message_type == SYNC_REQUEST_TYPE {
+                    let mode: Mode = seq.next_element()?.ok_or_else(|| {
+                        serde::de::Error::custom("missing mode in sync request message")
+                    })?;
+
+                    let target_set: TargetSet = seq.next_element()?.ok_or_else(|| {
+                        serde::de::Error::custom("missing target set in sync request message")
+                    })?;
+
+                    Ok(Message::SyncRequest(mode, target_set))
+                } else if message_type == ENTRY_TYPE {
+                    let entry_bytes: Vec<u8> = seq.next_element()?.ok_or_else(|| {
+                        serde::de::Error::custom("missing entry bytes in entry message")
+                    })?;
+
+                    let operation_bytes: Vec<u8> = seq.next_element()?.ok_or_else(|| {
+                        serde::de::Error::custom("missing operation bytes in entry message")
+                    })?;
+
+                    Ok(Message::Entry(entry_bytes, operation_bytes))
+                } else if message_type == SYNC_DONE_TYPE {
+                    let live_mode: bool = seq.next_element()?.ok_or_else(|| {
+                        serde::de::Error::custom("missing live mode flag in sync done message")
+                    })?;
+
+                    Ok(Message::SyncDone(live_mode))
+                } else {
+                    Err(serde::de::Error::custom(format!(
+                        "unknown message type {} in replication message",
+                        message_type
+                    )))
+                }?;
+
+                if let Some(items_left) = seq.size_hint() {
+                    if items_left > 0 {
+                        return Err(serde::de::Error::custom(
+                            "too many fields for replication message",
+                        ));
+                    }
+                };
+
+                Ok(SyncMessage::new(session_id, message))
+            }
+        }
+
+        deserializer.deserialize_seq(SyncMessageVisitor)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use ciborium::cbor;
+    use ciborium::value::{Error, Value};
+    use p2panda_rs::serde::{deserialize_into, serialize_from, serialize_value};
+    use rstest::rstest;
+
+    use crate::replication::{Mode, TargetSet};
+    use crate::test_utils::helpers::random_target_set;
+
+    use super::{Message, SyncMessage};
+
+    #[rstest]
+    fn serialize(#[from(random_target_set)] target_set: TargetSet) {
+        assert_eq!(
+            serialize_from(SyncMessage::new(
+                51,
+                Message::SyncRequest(Mode::SetReconciliation, target_set.clone())
+            )),
+            serialize_value(cbor!([0, 51, 1, target_set]))
+        );
+    }
+
+    #[rstest]
+    fn deserialize(#[from(random_target_set)] target_set: TargetSet) {
+        assert_eq!(
+            deserialize_into::<SyncMessage>(&serialize_value(cbor!([0, 12, 0, target_set])))
+                .unwrap(),
+            SyncMessage::new(12, Message::SyncRequest(Mode::Naive, target_set.clone()))
+        );
+    }
+
+    #[rstest]
+    #[should_panic(expected = "missing message type in replication message")]
+    #[case::no_fields(cbor!([]))]
+    #[should_panic(expected = "unknown message type 122 in replication message")]
+    #[case::unknown_message_type(cbor!([122, 0]))]
+    #[should_panic(expected = "missing session id in replication message")]
+    #[case::only_message_type(cbor!([0]))]
+    #[should_panic(expected = "too many fields for replication message")]
+    #[case::too_many_fields(cbor!([0, 0, 0, ["schema_field_definition_v1"], "too much"]))]
+    fn deserialize_invalid_messages(#[case] cbor: Result<Value, Error>) {
+        // Check the cbor is valid
+        assert!(cbor.is_ok());
+
+        // Deserialize into sync message, we unwrap here to cause a panic and then test for
+        // expected error stings
+        deserialize_into::<SyncMessage>(&serialize_value(cbor)).unwrap();
+    }
+}
diff --git a/aquadoggo/src/replication/mod.rs b/aquadoggo/src/replication/mod.rs
index c43557e14..57d0d7ca1 100644
--- a/aquadoggo/src/replication/mod.rs
+++ b/aquadoggo/src/replication/mod.rs
@@ -1,83 +1,17 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
-use p2panda_rs::schema::SchemaId;
-
 pub mod errors;
 mod manager;
 mod message;
+mod mode;
 mod session;
 mod strategies;
+mod target_set;
 pub mod traits;
 
 pub use manager::SyncManager;
-pub use message::{StrategyMessage, SyncMessage};
+pub use message::{Message, StrategyMessage, SyncMessage};
+pub use mode::Mode;
 pub use session::{Session, SessionId, SessionState};
 pub use strategies::{NaiveStrategy, SetReconciliationStrategy, StrategyResult};
-
-/// De-duplicated and sorted set of schema ids which define the target data for the replication
-/// session.
-#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd)]
-pub struct TargetSet(Vec<SchemaId>);
-
-impl TargetSet {
-    pub fn new(schema_ids: &[SchemaId]) -> Self {
-        // Remove duplicates
-        let mut deduplicated_set: Vec<SchemaId> = Vec::new();
-        for schema_id in schema_ids {
-            if !deduplicated_set.contains(schema_id) {
-                deduplicated_set.push(schema_id.clone());
-            }
-        }
-
-        // Sort schema ids to compare target sets easily
-        deduplicated_set.sort();
-
-        Self(deduplicated_set)
-    }
-}
-
-#[derive(Clone, Debug, Eq, PartialEq)]
-pub enum Mode {
-    Naive,
-    SetReconciliation,
-    Unknown,
-}
-
-#[cfg(test)]
-mod tests {
-    use p2panda_rs::document::DocumentViewId;
-    use p2panda_rs::schema::{SchemaId, SchemaName};
-    use p2panda_rs::test_utils::fixtures::random_document_view_id;
-    use rstest::rstest;
-
-    use super::TargetSet;
-
-    #[rstest]
-    fn compare_target_sets(
-        #[from(random_document_view_id)] document_view_id_1: DocumentViewId,
-        #[from(random_document_view_id)] document_view_id_2: DocumentViewId,
-    ) {
-        let schema_id_1 =
-            SchemaId::new_application(&SchemaName::new("messages").unwrap(), &document_view_id_1);
-        let schema_id_2 =
-            SchemaId::new_application(&SchemaName::new("profiles").unwrap(), &document_view_id_2);
-
-        // Sort schema ids
-        assert_eq!(
-            TargetSet::new(&[schema_id_1.clone(), schema_id_2.clone()]),
-            TargetSet::new(&[schema_id_2.clone(), schema_id_1.clone()]),
-        );
-
-        // De-duplicate schema ids
-        assert_eq!(
-            TargetSet::new(&[schema_id_1.clone(), schema_id_1.clone()]),
-            TargetSet::new(&[schema_id_1.clone()]),
-        );
-
-        // Distinct different target sets from each other
-        assert_ne!(
-            TargetSet::new(&[schema_id_1.clone()]),
-            TargetSet::new(&[schema_id_2.clone()]),
-        );
-    }
-}
+pub use target_set::TargetSet;
diff --git a/aquadoggo/src/replication/mode.rs b/aquadoggo/src/replication/mode.rs
new file mode 100644
index 000000000..c592c83ff
--- /dev/null
+++ b/aquadoggo/src/replication/mode.rs
@@ -0,0 +1,103 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use std::fmt::{self, Display};
+
+use p2panda_rs::Human;
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum Mode {
+    Naive,
+    SetReconciliation,
+    Unknown,
+}
+
+impl Mode {
+    pub fn as_str(&self) -> &str {
+        match self {
+            Mode::Naive => "naive",
+            Mode::SetReconciliation => "set-reconciliation",
+            Mode::Unknown => "unknown",
+        }
+    }
+
+    pub fn as_u64(&self) -> u64 {
+        match self {
+            Mode::Naive => 0,
+            Mode::SetReconciliation => 1,
+            Mode::Unknown => unreachable!("Can't create an unknown replication mode"),
+        }
+    }
+}
+
+impl From<u64> for Mode {
+    fn from(value: u64) -> Self {
+        match value {
+            0 => Mode::Naive,
+            1 => Mode::SetReconciliation,
+            _ => Mode::Unknown,
+        }
+    }
+}
+
+impl Serialize for Mode {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.serialize_u64(self.as_u64())
+    }
+}
+
+impl<'de> Deserialize<'de> for Mode {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let mode = u64::deserialize(deserializer)?;
+        Ok(mode.into())
+    }
+}
+
+impl Display for Mode {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.as_u64())
+    }
+}
+
+impl Human for Mode {
+    fn display(&self) -> String {
+        self.as_str().to_owned()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use ciborium::cbor;
+
+    use p2panda_rs::serde::{deserialize_into, serialize_from, serialize_value};
+
+    use super::Mode;
+
+    #[test]
+    fn u64_representation() {
+        assert_eq!(Mode::Naive.as_u64(), 0);
+        assert_eq!(Mode::SetReconciliation.as_u64(), 1);
+    }
+
+    #[test]
+    fn serialize() {
+        let bytes = serialize_from(Mode::Naive);
+        assert_eq!(bytes, vec![0]);
+    }
+
+    #[test]
+    fn deserialize() {
+        let version: Mode = deserialize_into(&serialize_value(cbor!(1))).unwrap();
+        assert_eq!(version, Mode::SetReconciliation);
+
+        // Can not be a string
+        let invalid_type = deserialize_into::<Mode>(&serialize_value(cbor!("0")));
+        assert!(invalid_type.is_err());
+    }
+}
diff --git a/aquadoggo/src/replication/target_set.rs b/aquadoggo/src/replication/target_set.rs
new file mode 100644
index 000000000..03ac3e64e
--- /dev/null
+++ b/aquadoggo/src/replication/target_set.rs
@@ -0,0 +1,133 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use p2panda_rs::schema::SchemaId;
+use p2panda_rs::Validate;
+use serde::{Deserialize, Deserializer, Serialize};
+
+use crate::replication::errors::TargetSetError;
+
+/// De-duplicated and sorted set of schema ids which define the target data for the replication
+/// session.
+#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize)]
+pub struct TargetSet(Vec<SchemaId>);
+
+impl TargetSet {
+    pub fn new(schema_ids: &[SchemaId]) -> Self {
+        // Remove duplicates
+        let mut deduplicated_set: Vec<SchemaId> = Vec::new();
+        for schema_id in schema_ids {
+            if !deduplicated_set.contains(schema_id) {
+                deduplicated_set.push(schema_id.clone());
+            }
+        }
+
+        // Sort schema ids to compare target sets easily
+        deduplicated_set.sort();
+
+        Self(deduplicated_set)
+    }
+
+    fn from_untrusted(schema_ids: Vec<SchemaId>) -> Result<Self, TargetSetError> {
+        // Create target set with potentially invalid data
+        let target_set = Self(schema_ids);
+
+        // Make sure its sorted and does not contain any duplicates
+        target_set.validate()?;
+
+        Ok(target_set)
+    }
+}
+
+impl Validate for TargetSet {
+    type Error = TargetSetError;
+
+    fn validate(&self) -> Result<(), Self::Error> {
+        // Check if at least one schema id is given
+        if self.0.is_empty() {
+            return Err(TargetSetError::ZeroSchemaIds);
+        };
+
+        let mut prev_schema_id: Option<&SchemaId> = None;
+
+        for schema_id in &self.0 {
+            // Check if it is sorted, this indirectly also checks against duplicates
+            if let Some(prev) = prev_schema_id {
+                if prev >= schema_id {
+                    return Err(TargetSetError::UnsortedSchemaIds);
+                }
+            }
+
+            prev_schema_id = Some(schema_id);
+        }
+
+        Ok(())
+    }
+}
+
+impl<'de> Deserialize<'de> for TargetSet {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        // Deserializer of `SchemaId` checks internally for the correct format of each id
+        let schema_ids: Vec<SchemaId> = Deserialize::deserialize(deserializer)?;
+        Self::from_untrusted(schema_ids).map_err(serde::de::Error::custom)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use ciborium::cbor;
+    use p2panda_rs::document::DocumentViewId;
+    use p2panda_rs::schema::{SchemaId, SchemaName};
+    use p2panda_rs::serde::{deserialize_into, serialize_value};
+    use p2panda_rs::test_utils::fixtures::random_document_view_id;
+    use rstest::rstest;
+
+    use super::TargetSet;
+
+    #[rstest]
+    fn compare_target_sets(
+        #[from(random_document_view_id)] document_view_id_1: DocumentViewId,
+        #[from(random_document_view_id)] document_view_id_2: DocumentViewId,
+    ) {
+        let schema_id_1 =
+            SchemaId::new_application(&SchemaName::new("messages").unwrap(), &document_view_id_1);
+        let schema_id_2 =
+            SchemaId::new_application(&SchemaName::new("profiles").unwrap(), &document_view_id_2);
+
+        // Sort schema ids
+        assert_eq!(
+            TargetSet::new(&[schema_id_1.clone(), schema_id_2.clone()]),
+            TargetSet::new(&[schema_id_2.clone(), schema_id_1.clone()]),
+        );
+
+        // De-duplicate schema ids
+        assert_eq!(
+            TargetSet::new(&[schema_id_1.clone(), schema_id_1.clone()]),
+            TargetSet::new(&[schema_id_1.clone()]),
+        );
+
+        // Distinct different target sets from each other
+        assert_ne!(
+            TargetSet::new(&[schema_id_1.clone()]),
+            TargetSet::new(&[schema_id_2.clone()]),
+        );
+    }
+
+    #[rstest]
+    fn deserialize_unsorted_target_set() {
+        let unsorted_schema_ids = [
+            "venues_0020c13cdc58dfc6f4ebd32992ff089db79980363144bdb2743693a019636fa72ec8",
+            "alpacas_00202dce4b32cd35d61cf54634b93a526df333c5ed3d93230c2f026f8d1ecabc0cd7",
+        ];
+        let result = deserialize_into::<TargetSet>(&serialize_value(cbor!(unsorted_schema_ids)));
+
+        let expected_result = ciborium::de::Error::<std::io::Error>::Semantic(
+            None,
+            "Target set contains unsorted or duplicate schema ids".to_string(),
+        );
+
+        assert_eq!(result.unwrap_err().to_string(), expected_result.to_string());
+    }
+}


Test patch:
diff --git a/aquadoggo/src/test_utils/helpers.rs b/aquadoggo/src/test_utils/helpers.rs
index f83999d64..725d88551 100644
--- a/aquadoggo/src/test_utils/helpers.rs
+++ b/aquadoggo/src/test_utils/helpers.rs
@@ -10,7 +10,10 @@ use p2panda_rs::operation::{
 use p2panda_rs::schema::{Schema, SchemaId, SchemaName};
 use p2panda_rs::storage_provider::traits::OperationStore;
 use p2panda_rs::test_utils::constants;
-use p2panda_rs::test_utils::fixtures::{schema, schema_fields};
+use p2panda_rs::test_utils::fixtures::{random_document_view_id, schema, schema_fields};
+use rstest::fixture;
+
+use crate::replication::TargetSet;
 
 /// Schema id used in aquadoggo tests.
 fn doggo_schema_id() -> SchemaId {
@@ -112,3 +115,11 @@ pub fn schema_from_fields(fields: Vec<(&str, OperationValue)>) -> Schema {
         "A doggo schema for testing",
     )
 }
+
+#[fixture]
+pub fn random_target_set() -> TargetSet {
+    let document_view_id = random_document_view_id();
+    let schema_id =
+        SchemaId::new_application(&SchemaName::new("messages").unwrap(), &document_view_id);
+    TargetSet::new(&[schema_id])
+}
diff --git a/aquadoggo/src/test_utils/mod.rs b/aquadoggo/src/test_utils/mod.rs
index 287d2bc70..dcda6881e 100644
--- a/aquadoggo/src/test_utils/mod.rs
+++ b/aquadoggo/src/test_utils/mod.rs
@@ -3,7 +3,7 @@
 mod client;
 mod config;
 mod db;
-mod helpers;
+pub mod helpers;
 mod node;
 mod runner;