Repository: p2panda/aquadoggo (language: rust)

Issue:
Announcement messages



Discussion / hints:

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index bb0983f26..a248669a8 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -31,6 +31,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Only replicate and materialize configured "supported schema" [#569](https://github.com/p2panda/aquadoggo/pull/469)
 - Parse supported schema ids from `config.toml` [#473](https://github.com/p2panda/aquadoggo/pull/473)
 - Fix relayed connections, add DCUtR Holepunching and reduce CLI args [#502](https://github.com/p2panda/aquadoggo/pull/502)
+- Announce supported schema ids in network before replication [#515](https://github.com/p2panda/aquadoggo/pull/515)
 
 ### Changed
 
diff --git a/aquadoggo/src/bus.rs b/aquadoggo/src/bus.rs
index addc9402c..c7bdbc90c 100644
--- a/aquadoggo/src/bus.rs
+++ b/aquadoggo/src/bus.rs
@@ -3,8 +3,7 @@
 use p2panda_rs::operation::OperationId;
 
 use crate::manager::Sender;
-use crate::network::Peer;
-use crate::replication::SyncMessage;
+use crate::network::{Peer, PeerMessage};
 
 /// Sender for cross-service communication bus.
 pub type ServiceSender = Sender<ServiceMessage>;
@@ -21,11 +20,11 @@ pub enum ServiceMessage {
     /// Node closed a connection to another node.
     PeerDisconnected(Peer),
 
-    /// Node sent a message to remote node for replication.
-    SentReplicationMessage(Peer, SyncMessage),
+    /// Node sent a message to remote node.
+    SentMessage(Peer, PeerMessage),
 
-    /// Node received a message from remote node for replication.
-    ReceivedReplicationMessage(Peer, SyncMessage),
+    /// Node received a message from remote node.
+    ReceivedMessage(Peer, PeerMessage),
 
     /// Replication protocol failed with an critical error.
     ReplicationFailed(Peer),
diff --git a/aquadoggo/src/network/mod.rs b/aquadoggo/src/network/mod.rs
index a5440ba00..514c5e9c7 100644
--- a/aquadoggo/src/network/mod.rs
+++ b/aquadoggo/src/network/mod.rs
@@ -10,6 +10,6 @@ mod swarm;
 mod transport;
 
 pub use config::NetworkConfiguration;
-pub use peers::Peer;
+pub use peers::{Peer, PeerMessage};
 pub use service::network_service;
 pub use shutdown::ShutdownHandler;
diff --git a/aquadoggo/src/network/peers/behaviour.rs b/aquadoggo/src/network/peers/behaviour.rs
index 898915f25..d5559715a 100644
--- a/aquadoggo/src/network/peers/behaviour.rs
+++ b/aquadoggo/src/network/peers/behaviour.rs
@@ -12,13 +12,12 @@ use libp2p::swarm::{
 use libp2p::{Multiaddr, PeerId};
 
 use crate::network::peers::handler::{Handler, HandlerFromBehaviour, HandlerToBehaviour};
-use crate::network::peers::Peer;
-use crate::replication::SyncMessage;
+use crate::network::peers::{Peer, PeerMessage};
 
 #[derive(Debug)]
 pub enum Event {
     /// Message received on the inbound stream.
-    MessageReceived(Peer, SyncMessage),
+    MessageReceived(Peer, PeerMessage),
 
     /// We established an inbound or outbound connection to a peer for the first time.
     PeerConnected(Peer),
@@ -105,7 +104,7 @@ impl Behaviour {
         &mut self,
         peer_id: PeerId,
         connection_id: ConnectionId,
-        message: SyncMessage,
+        message: PeerMessage,
     ) {
         let peer = Peer::new(peer_id, connection_id);
         self.push_event(ToSwarm::GenerateEvent(Event::MessageReceived(
@@ -130,7 +129,7 @@ impl Behaviour {
         self.enabled = true
     }
 
-    pub fn send_message(&mut self, peer: Peer, message: SyncMessage) {
+    pub fn send_message(&mut self, peer: Peer, message: PeerMessage) {
         self.push_event(ToSwarm::NotifyHandler {
             peer_id: peer.id(),
             event: HandlerFromBehaviour::Message(message),
@@ -236,7 +235,7 @@ mod tests {
     use p2panda_rs::schema::SchemaId;
     use rstest::rstest;
 
-    use crate::network::Peer;
+    use crate::network::{Peer, PeerMessage};
     use crate::replication::{Message, SyncMessage, TargetSet};
     use crate::test_utils::helpers::random_target_set;
 
@@ -306,7 +305,10 @@ mod tests {
         // Send a message from to swarm_1 local peer from swarm_2 local peer.
         swarm_1.behaviour_mut().send_message(
             Peer::new(swarm_2_peer_id, ConnectionId::new_unchecked(1)),
-            SyncMessage::new(0, Message::SyncRequest(0.into(), TargetSet::new(&vec![]))),
+            PeerMessage::SyncMessage(SyncMessage::new(
+                0,
+                Message::SyncRequest(0.into(), TargetSet::new(&vec![])),
+            )),
         );
 
         // Await a swarm event on swarm_2.
@@ -370,13 +372,19 @@ mod tests {
         // Send a message from swarm_1 to swarm_2
         swarm_1.behaviour_mut().send_message(
             peer_2,
-            SyncMessage::new(0, Message::SyncRequest(0.into(), target_set_1.clone())),
+            PeerMessage::SyncMessage(SyncMessage::new(
+                0,
+                Message::SyncRequest(0.into(), target_set_1.clone()),
+            )),
         );
 
         // Send a message from swarm_2 to swarm_1
         swarm_2.behaviour_mut().send_message(
             peer_1,
-            SyncMessage::new(1, Message::SyncRequest(0.into(), target_set_2.clone())),
+            PeerMessage::SyncMessage(SyncMessage::new(
+                1,
+                Message::SyncRequest(0.into(), target_set_2.clone()),
+            )),
         );
 
         // And again add the next behaviour events which occur in either swarms
@@ -395,7 +403,10 @@ mod tests {
         assert_eq!(peer.id(), swarm_2_peer_id);
         assert_eq!(
             message.unwrap(),
-            SyncMessage::new(1, Message::SyncRequest(0.into(), target_set_2.clone()))
+            PeerMessage::SyncMessage(SyncMessage::new(
+                1,
+                Message::SyncRequest(0.into(), target_set_2.clone())
+            ))
         );
 
         // swarm_2 should have received the message from swarm_1 peer
@@ -403,7 +414,10 @@ mod tests {
         assert_eq!(peer.id(), swarm_1_peer_id);
         assert_eq!(
             message.unwrap(),
-            SyncMessage::new(0, Message::SyncRequest(0.into(), target_set_1))
+            PeerMessage::SyncMessage(SyncMessage::new(
+                0,
+                Message::SyncRequest(0.into(), target_set_1)
+            ))
         );
     }
 }
diff --git a/aquadoggo/src/network/peers/handler.rs b/aquadoggo/src/network/peers/handler.rs
index b45ba9d24..43cf8de2f 100644
--- a/aquadoggo/src/network/peers/handler.rs
+++ b/aquadoggo/src/network/peers/handler.rs
@@ -15,8 +15,7 @@ use libp2p::swarm::{
 use log::warn;
 use thiserror::Error;
 
-use crate::network::peers::{Codec, CodecError, Protocol};
-use crate::replication::SyncMessage;
+use crate::network::peers::{Codec, CodecError, PeerMessage, Protocol};
 
 /// The time a connection is maintained to a peer without being in live mode and without
 /// send/receiving a message from. Connections that idle beyond this timeout are disconnected.
@@ -89,7 +88,7 @@ pub struct Handler {
     outbound_substream_establishing: bool,
 
     /// Queue of messages that we want to send to the remote.
-    send_queue: VecDeque<SyncMessage>,
+    send_queue: VecDeque<PeerMessage>,
 
     /// Last time we've observed inbound or outbound messaging activity.
     last_io_activity: Instant,
@@ -140,7 +139,7 @@ impl Handler {
 #[derive(Debug)]
 pub enum HandlerFromBehaviour {
     /// Message to send on outbound stream.
-    Message(SyncMessage),
+    Message(PeerMessage),
 
     /// Protocol failed with a critical error.
     CriticalError,
@@ -152,7 +151,7 @@ pub enum HandlerFromBehaviour {
 #[derive(Debug)]
 pub enum HandlerToBehaviour {
     /// Message received on the inbound stream.
-    Message(SyncMessage),
+    Message(PeerMessage),
 }
 
 #[derive(Debug, Error)]
@@ -181,7 +180,7 @@ enum OutboundSubstreamState {
     WaitingOutput(Stream),
 
     /// Waiting to send a message to the remote.
-    PendingSend(Stream, SyncMessage),
+    PendingSend(Stream, PeerMessage),
 
     /// Waiting to flush the substream so that the data arrives to the remote.
     PendingFlush(Stream),
diff --git a/aquadoggo/src/network/peers/message.rs b/aquadoggo/src/network/peers/message.rs
new file mode 100644
index 000000000..6f1c98ea6
--- /dev/null
+++ b/aquadoggo/src/network/peers/message.rs
@@ -0,0 +1,16 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use serde::{Deserialize, Serialize};
+
+use crate::replication::{AnnouncementMessage, SyncMessage};
+
+/// p2panda protocol messages which can be sent over the wire.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum PeerMessage {
+    /// Announcement of peers about the schema ids they are interest in.
+    Announce(AnnouncementMessage),
+
+    /// Replication status and data exchange.
+    SyncMessage(SyncMessage),
+}
diff --git a/aquadoggo/src/network/peers/mod.rs b/aquadoggo/src/network/peers/mod.rs
index 579c75f3c..a2fdbfa27 100644
--- a/aquadoggo/src/network/peers/mod.rs
+++ b/aquadoggo/src/network/peers/mod.rs
@@ -2,10 +2,12 @@
 
 mod behaviour;
 mod handler;
+mod message;
 mod peer;
 mod protocol;
 
 pub use behaviour::{Behaviour, Event};
 pub use handler::Handler;
+pub use message::PeerMessage;
 pub use peer::Peer;
 pub use protocol::{Codec, CodecError, Protocol, PROTOCOL_NAME};
diff --git a/aquadoggo/src/network/peers/protocol.rs b/aquadoggo/src/network/peers/protocol.rs
index ba06e0e55..17e670f50 100644
--- a/aquadoggo/src/network/peers/protocol.rs
+++ b/aquadoggo/src/network/peers/protocol.rs
@@ -7,13 +7,13 @@ use futures::{future, AsyncRead, AsyncWrite, Future};
 use libp2p::core::UpgradeInfo;
 use libp2p::{InboundUpgrade, OutboundUpgrade};
 
-use crate::replication::SyncMessage;
+use crate::network::peers::PeerMessage;
 
 pub const PROTOCOL_NAME: &str = "/p2p/p2panda/1.0.0";
 
 pub type CodecError = CborCodecError;
 
-pub type Codec = CborCodec<SyncMessage, SyncMessage>;
+pub type Codec = CborCodec<PeerMessage, PeerMessage>;
 
 #[derive(Clone, Debug)]
 pub struct Protocol;
diff --git a/aquadoggo/src/network/service.rs b/aquadoggo/src/network/service.rs
index f62e69a19..3e7b76f6a 100644
--- a/aquadoggo/src/network/service.rs
+++ b/aquadoggo/src/network/service.rs
@@ -344,11 +344,11 @@ impl EventLoop {
     /// Handle an incoming message via the communication bus from other services.
     async fn handle_service_message(&mut self, message: ServiceMessage) {
         match message {
-            ServiceMessage::SentReplicationMessage(peer, sync_message) => self
+            ServiceMessage::SentMessage(peer, peer_message) => self
                 .swarm
                 .behaviour_mut()
                 .peers
-                .send_message(peer, sync_message),
+                .send_message(peer, peer_message),
             ServiceMessage::ReplicationFailed(peer) => {
                 self.swarm.behaviour_mut().peers.handle_critical_error(peer);
             }
@@ -368,10 +368,7 @@ impl EventLoop {
             }
             peers::Event::MessageReceived(peer, message) => {
                 // Inform other services about received messages from peer
-                self.send_service_message(ServiceMessage::ReceivedReplicationMessage(
-                    *peer,
-                    message.clone(),
-                ))
+                self.send_service_message(ServiceMessage::ReceivedMessage(*peer, message.clone()))
             }
         }
     }
diff --git a/aquadoggo/src/replication/announcement.rs b/aquadoggo/src/replication/announcement.rs
new file mode 100644
index 000000000..cc78f2ec9
--- /dev/null
+++ b/aquadoggo/src/replication/announcement.rs
@@ -0,0 +1,194 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use p2panda_rs::Validate;
+use serde::de::Visitor;
+use serde::ser::SerializeSeq;
+use serde::{Deserialize, Serialize};
+
+use crate::replication::{MessageType, TargetSet, ANNOUNCE_TYPE, REPLICATION_PROTOCOL_VERSION};
+
+/// u64 timestamp from UNIX epoch until now.
+pub fn now() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .expect("System time invalid, operation system time configured before UNIX epoch")
+        .as_secs()
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Announcement {
+    /// This contains a list of schema ids this peer allowed to support.
+    // @TODO: `TargetSet` is not a good name anymore in this context. See related issue:
+    // https://github.com/p2panda/aquadoggo/issues/527
+    pub supported_schema_ids: TargetSet,
+
+    /// Timestamp of this announcement. Helps to understand if we can override the previous
+    /// announcement with a newer one.
+    pub timestamp: u64,
+}
+
+impl Announcement {
+    pub fn new(supported_schema_ids: TargetSet) -> Self {
+        Self {
+            timestamp: now(),
+            supported_schema_ids,
+        }
+    }
+}
+
+pub type ProtocolVersion = u64;
+
+/// Message which can be used to send announcements over the wire.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct AnnouncementMessage(ProtocolVersion, Announcement);
+
+impl AnnouncementMessage {
+    pub fn new(announcement: Announcement) -> Self {
+        Self(REPLICATION_PROTOCOL_VERSION, announcement)
+    }
+
+    pub fn announcement(&self) -> Announcement {
+        self.1.clone()
+    }
+
+    pub fn is_version_supported(&self) -> bool {
+        self.0 == REPLICATION_PROTOCOL_VERSION
+    }
+}
+
+impl Serialize for AnnouncementMessage {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        let mut seq = serializer.serialize_seq(Some(4))?;
+        seq.serialize_element(&ANNOUNCE_TYPE)?;
+        seq.serialize_element(&self.0)?;
+        seq.serialize_element(&self.1.timestamp)?;
+        seq.serialize_element(&self.1.supported_schema_ids)?;
+        seq.end()
+    }
+}
+
+impl<'de> Deserialize<'de> for AnnouncementMessage {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        struct MessageVisitor;
+
+        impl<'de> Visitor<'de> for MessageVisitor {
+            type Value = AnnouncementMessage;
+
+            fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
+                formatter.write_str("p2panda announce 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 announce message")
+                })?;
+
+                if message_type != ANNOUNCE_TYPE {
+                    return Err(serde::de::Error::custom(
+                        "invalid message type for announce message",
+                    ));
+                }
+
+                let protocol_version: ProtocolVersion = seq.next_element()?.ok_or_else(|| {
+                    serde::de::Error::custom("missing protocol version in announce message")
+                })?;
+
+                let timestamp: u64 = seq.next_element()?.ok_or_else(|| {
+                    serde::de::Error::custom("missing timestamp in announce message")
+                })?;
+
+                let supported_schema_ids: TargetSet = seq.next_element()?.ok_or_else(|| {
+                    serde::de::Error::custom("missing target set in announce message")
+                })?;
+                supported_schema_ids.validate().map_err(|_| {
+                    serde::de::Error::custom("invalid target set in announce message")
+                })?;
+
+                if let Some(items_left) = seq.size_hint() {
+                    if items_left > 0 {
+                        return Err(serde::de::Error::custom(
+                            "too many fields for announce message",
+                        ));
+                    }
+                };
+
+                Ok(AnnouncementMessage(
+                    protocol_version,
+                    Announcement {
+                        supported_schema_ids,
+                        timestamp,
+                    },
+                ))
+            }
+        }
+
+        deserializer.deserialize_seq(MessageVisitor)
+    }
+}
+
+#[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::TargetSet;
+    use crate::test_utils::helpers::random_target_set;
+
+    use super::{Announcement, AnnouncementMessage};
+
+    #[rstest]
+    fn serialize(#[from(random_target_set)] supported_schema_ids: TargetSet) {
+        let announcement = Announcement::new(supported_schema_ids.clone());
+        assert_eq!(
+            serialize_from(AnnouncementMessage::new(announcement.clone())),
+            serialize_value(cbor!([0, 1, announcement.timestamp, supported_schema_ids]))
+        );
+    }
+
+    #[rstest]
+    fn deserialize(#[from(random_target_set)] supported_schema_ids: TargetSet) {
+        assert_eq!(
+            deserialize_into::<AnnouncementMessage>(&serialize_value(cbor!([
+                0,
+                1,
+                12345678,
+                supported_schema_ids
+            ])))
+            .unwrap(),
+            AnnouncementMessage::new(Announcement {
+                timestamp: 12345678,
+                supported_schema_ids,
+            })
+        );
+    }
+
+    #[rstest]
+    #[should_panic(expected = "missing message type in announce message")]
+    #[case::missing_version(cbor!([]))]
+    #[should_panic(expected = "missing protocol version in announce message")]
+    #[case::missing_version(cbor!([0]))]
+    #[should_panic(expected = "missing timestamp in announce message")]
+    #[case::missing_timestamp(cbor!([0, 122]))]
+    #[should_panic(expected = "too many fields for announce message")]
+    #[case::too_many_fields(cbor!([0, 1, 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());
+
+        // We unwrap here to cause a panic and then test for expected error stings
+        deserialize_into::<AnnouncementMessage>(&serialize_value(cbor)).unwrap();
+    }
+}
diff --git a/aquadoggo/src/replication/errors.rs b/aquadoggo/src/replication/errors.rs
index b1a3204e3..cb2b737a8 100644
--- a/aquadoggo/src/replication/errors.rs
+++ b/aquadoggo/src/replication/errors.rs
@@ -56,9 +56,6 @@ pub enum IngestError {
 
 #[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/ingest.rs b/aquadoggo/src/replication/ingest.rs
index bfb52ba32..0aee97444 100644
--- a/aquadoggo/src/replication/ingest.rs
+++ b/aquadoggo/src/replication/ingest.rs
@@ -49,12 +49,16 @@ impl SyncIngest {
 
         let plain_operation = decode_operation(encoded_operation)?;
 
-        // If the node has been configured with supported_schema_ids, check that the sent
-        // operation follows one of our supported schema.
-        if let Some(supported_schema_ids) = self.schema_provider.supported_schema_ids() {
-            if supported_schema_ids.contains(plain_operation.schema_id()) {
-                return Err(IngestError::UnsupportedSchema);
-            }
+        // If the node has been configured with a whitelist of supported schema ids, check that the
+        // sent operation follows one of our supported schema
+        if self.schema_provider.is_whitelist_active()
+            && self
+                .schema_provider
+                .supported_schema_ids()
+                .await
+                .contains(plain_operation.schema_id())
+        {
+            return Err(IngestError::UnsupportedSchema);
         }
 
         // Retrieve the schema if it has been materialized on the node.
diff --git a/aquadoggo/src/replication/manager.rs b/aquadoggo/src/replication/manager.rs
index 675b41801..73586c899 100644
--- a/aquadoggo/src/replication/manager.rs
+++ b/aquadoggo/src/replication/manager.rs
@@ -520,8 +520,8 @@ mod tests {
     use tokio::sync::broadcast;
 
     use crate::replication::errors::{DuplicateSessionRequestError, ReplicationError};
-    use crate::replication::message::{Message, HAVE_TYPE, SYNC_DONE_TYPE};
-    use crate::replication::{Mode, SyncIngest, SyncMessage, TargetSet};
+    use crate::replication::message::Message;
+    use crate::replication::{Mode, SyncIngest, SyncMessage, TargetSet, HAVE_TYPE, SYNC_DONE_TYPE};
     use crate::schema::SchemaProvider;
     use crate::test_utils::helpers::random_target_set;
     use crate::test_utils::{
diff --git a/aquadoggo/src/replication/message.rs b/aquadoggo/src/replication/message.rs
index 1360463f5..e4d7dbfcf 100644
--- a/aquadoggo/src/replication/message.rs
+++ b/aquadoggo/src/replication/message.rs
@@ -6,19 +6,15 @@ use p2panda_rs::entry::EncodedEntry;
 use p2panda_rs::entry::{LogId, SeqNum};
 use p2panda_rs::identity::PublicKey;
 use p2panda_rs::operation::EncodedOperation;
-use p2panda_rs::Human;
+use p2panda_rs::{Human, Validate};
 use serde::de::Visitor;
 use serde::ser::SerializeSeq;
 use serde::{Deserialize, Serialize};
 
-use crate::replication::{Mode, SessionId, TargetSet};
-
-pub const SYNC_REQUEST_TYPE: MessageType = 0;
-pub const ENTRY_TYPE: MessageType = 8;
-pub const SYNC_DONE_TYPE: MessageType = 9;
-pub const HAVE_TYPE: MessageType = 10;
-
-pub type MessageType = u64;
+use crate::replication::{
+    MessageType, Mode, SessionId, TargetSet, ENTRY_TYPE, HAVE_TYPE, SYNC_DONE_TYPE,
+    SYNC_REQUEST_TYPE,
+};
 
 pub type LiveMode = bool;
 
@@ -159,6 +155,16 @@ impl<'de> Deserialize<'de> for SyncMessage {
                         serde::de::Error::custom("missing target set in sync request message")
                     })?;
 
+                    target_set.validate().map_err(|_| {
+                        serde::de::Error::custom("invalid target set in sync request message")
+                    })?;
+
+                    if target_set.is_empty() {
+                        return Err(serde::de::Error::custom(
+                            "empty target set in sync request message",
+                        ));
+                    }
+
                     Ok(Message::SyncRequest(mode, target_set))
                 } else if message_type == ENTRY_TYPE {
                     let entry_bytes: EncodedEntry = seq.next_element()?.ok_or_else(|| {
@@ -226,7 +232,7 @@ mod tests {
                 51,
                 Message::SyncRequest(Mode::SetReconciliation, target_set.clone())
             )),
-            serialize_value(cbor!([0, 51, 1, target_set]))
+            serialize_value(cbor!([1, 51, 1, target_set]))
         );
 
         assert_eq!(
@@ -253,7 +259,7 @@ mod tests {
     #[rstest]
     fn deserialize(#[from(random_target_set)] target_set: TargetSet, public_key: PublicKey) {
         assert_eq!(
-            deserialize_into::<SyncMessage>(&serialize_value(cbor!([0, 12, 0, target_set])))
+            deserialize_into::<SyncMessage>(&serialize_value(cbor!([1, 12, 0, target_set])))
                 .unwrap(),
             SyncMessage::new(
                 12,
@@ -296,9 +302,11 @@ mod tests {
     #[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]))]
+    #[case::only_message_type(cbor!([1]))]
+    #[should_panic(expected = "empty target set in sync request")]
+    #[case::only_message_type(cbor!([1, 0, 0, []]))]
     #[should_panic(expected = "too many fields for replication message")]
-    #[case::too_many_fields(cbor!([0, 0, 0, ["schema_field_definition_v1"], "too much"]))]
+    #[case::too_many_fields(cbor!([1, 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());
diff --git a/aquadoggo/src/replication/mod.rs b/aquadoggo/src/replication/mod.rs
index 599a64fb3..ba79445e7 100644
--- a/aquadoggo/src/replication/mod.rs
+++ b/aquadoggo/src/replication/mod.rs
@@ -1,5 +1,6 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+mod announcement;
 pub mod errors;
 mod ingest;
 mod manager;
@@ -11,6 +12,7 @@ mod strategies;
 mod target_set;
 pub mod traits;
 
+pub use announcement::{now, Announcement, AnnouncementMessage};
 pub use ingest::SyncIngest;
 pub use manager::SyncManager;
 pub use message::{LiveMode, LogHeights, Message, SyncMessage};
@@ -19,3 +21,15 @@ pub use service::replication_service;
 pub use session::{Session, SessionId, SessionState};
 pub use strategies::{LogHeightStrategy, SetReconciliationStrategy, StrategyResult};
 pub use target_set::TargetSet;
+
+pub type MessageType = u64;
+
+// Integers indicating message type for wire message format.
+pub const ANNOUNCE_TYPE: MessageType = 0;
+pub const SYNC_REQUEST_TYPE: MessageType = 1;
+pub const ENTRY_TYPE: MessageType = 2;
+pub const SYNC_DONE_TYPE: MessageType = 3;
+pub const HAVE_TYPE: MessageType = 10;
+
+/// Currently supported p2panda replication protocol version.
+pub const REPLICATION_PROTOCOL_VERSION: u64 = 1;
diff --git a/aquadoggo/src/replication/service.rs b/aquadoggo/src/replication/service.rs
index 66b5fade9..83268e07a 100644
--- a/aquadoggo/src/replication/service.rs
+++ b/aquadoggo/src/replication/service.rs
@@ -6,7 +6,7 @@ use std::time::Duration;
 use anyhow::Result;
 use libp2p::PeerId;
 use log::{debug, info, trace, warn};
-use p2panda_rs::{Human, Validate};
+use p2panda_rs::Human;
 use rand::seq::SliceRandom;
 use rand::thread_rng;
 use tokio::task;
@@ -19,10 +19,11 @@ use crate::context::Context;
 use crate::db::SqlStore;
 use crate::manager::{ServiceReadySender, Shutdown};
 use crate::network::identity::to_libp2p_peer_id;
-use crate::network::Peer;
+use crate::network::{Peer, PeerMessage};
 use crate::replication::errors::ReplicationError;
 use crate::replication::{
-    Message, Mode, Session, SessionId, SyncIngest, SyncManager, SyncMessage, TargetSet,
+    now, Announcement, AnnouncementMessage, Message, Mode, Session, SessionId, SyncIngest,
+    SyncManager, SyncMessage, TargetSet,
 };
 use crate::schema::SchemaProvider;
 
@@ -66,8 +67,20 @@ pub async fn replication_service(
 /// Statistics about successful and failed replication sessions for each connected peer.
 #[derive(Debug, Clone, PartialEq, Eq)]
 struct PeerStatus {
+    /// Connectivity information like libp2p peer id and connection id.
     peer: Peer,
+
+    /// Last known announcement of this peer.
+    announcement: Option<Announcement>,
+
+    /// Last time we've announced our local target set with this peer. Helps to check if we need to
+    /// inform them about any updates from our side.
+    sent_our_announcement_timestamp: u64,
+
+    /// Number of successful replication sessions.
     successful_count: usize,
+
+    /// Number of failed replication sessions.
     failed_count: usize,
 }
 
@@ -75,6 +88,8 @@ impl PeerStatus {
     pub fn new(peer: Peer) -> Self {
         Self {
             peer,
+            announcement: None,
+            sent_our_announcement_timestamp: 0,
             successful_count: 0,
             failed_count: 0,
         }
@@ -85,12 +100,13 @@ impl PeerStatus {
 ///
 /// This entails:
 ///
-/// 1. Handles incoming replication- and peer connection messages from other services
-/// 2. Maintains a list of currently connected p2panda peers
-/// 3. Routes messages to the right replication session with help of the `SyncManager` and returns
+/// 1. Manages announcements of us and other peers about which schema ids are supported
+/// 2. Handles incoming replication- and peer connection messages from other services
+/// 3. Maintains a list of currently connected p2panda peers.
+/// 4. Routes messages to the right replication session with help of the `SyncManager` and returns
 ///    responses to other services
-/// 4. Schedules new replication sessions
-/// 5. Handles replication errors and informs other services about them
+/// 5. Schedules new replication sessions
+/// 6. Handles replication errors and informs other services about them
 struct ConnectionManager {
     /// List of peers the connection mananger knows about and are available for replication.
     peers: HashMap<Peer, PeerStatus>,
@@ -110,6 +126,10 @@ struct ConnectionManager {
 
     /// Provider to retrieve our currently supported schema ids.
     schema_provider: SchemaProvider,
+
+    /// Our latest announcement state we want to propagate to all current and future peers. It
+    /// contains a list of schema ids we're supporting as a node.
+    announcement: Option<Announcement>,
 }
 
 impl ConnectionManager {
@@ -132,24 +152,14 @@ impl ConnectionManager {
             tx: tx.clone(),
             rx: BroadcastStream::new(tx.subscribe()),
             schema_provider: schema_provider.clone(),
+            announcement: None,
         }
     }
 
     /// Returns set of schema ids we are interested in and support on this node.
     async fn target_set(&self) -> TargetSet {
-        let supported_schema = match self.schema_provider.supported_schema_ids() {
-            // If supported_schema_ids is set return this list.
-            Some(supported_schema_ids) => supported_schema_ids.to_owned(),
-            // Otherwise return ids for all schema we know about on this node.
-            None => self
-                .schema_provider
-                .all()
-                .await
-                .iter()
-                .map(|schema| schema.id().to_owned())
-                .collect(),
-        };
-        TargetSet::new(&supported_schema)
+        let supported_schema_ids = self.schema_provider.supported_schema_ids().await;
+        TargetSet::new(&supported_schema_ids)
     }
 
     /// Register a new peer connection on the manager.
@@ -162,11 +172,20 @@ impl ConnectionManager {
             }
             None => {
                 self.peers.insert(peer, PeerStatus::new(peer));
-                self.update_sessions().await;
+                self.on_update().await;
             }
         }
     }
 
+    /// Routines which get executed on every scheduler beat and newly established connection.
+    async fn on_update(&mut self) {
+        // Inform new peers about our supported protocol version and schema ids
+        self.announce().await;
+
+        // Check if we can establish replication sessions with peers
+        self.update_sessions().await;
+    }
+
     /// Handle a peer connection closing.
     async fn on_connection_closed(&mut self, peer: Peer) {
         info!("Closed connection with peer: {}", peer.display());
@@ -183,18 +202,59 @@ impl ConnectionManager {
         }
     }
 
+    /// Update announcement state of a remote peer.
+    async fn on_announcement_message(&mut self, peer: Peer, message: AnnouncementMessage) {
+        // Check if this node supports our replication protocol version
+        if !message.is_version_supported() {
+            return;
+        }
+
+        let incoming_announcement = message.announcement();
+
+        match self.peers.get_mut(&peer) {
+            Some(status) => match &status.announcement {
+                Some(current) => {
+                    // Only update peer status when incoming announcement has a newer timestamp
+                    if current.timestamp < incoming_announcement.timestamp {
+                        trace!(
+                            "Received updated announcement state from peer {}",
+                            peer.display()
+                        );
+                        status.announcement = Some(incoming_announcement);
+                    }
+                }
+                None => {
+                    trace!(
+                        "Received first announcement state from peer {}",
+                        peer.display()
+                    );
+                    status.announcement = Some(incoming_announcement);
+                }
+            },
+            None => {
+                trace!("Tried to update announcement state of unknown peer");
+            }
+        }
+    }
+
     /// Route incoming replication messages to the right session.
     async fn on_replication_message(&mut self, peer: Peer, message: SyncMessage) {
         let session_id = message.session_id();
 
-        // If this is a SyncRequest message first we check if the contained TargetSet matches our
-        // own locally configured TargetSet.
-        if let Message::SyncRequest(_, target_set) = message.message() {
-            // If this node has been configured with supported_schema_ids then we check the target
-            // set of the requests matches our own, otherwise we skip this step and accept any
-            // target set.
-            if self.schema_provider.supported_schema_ids().is_some()
-                && target_set != &self.target_set().await
+        // If this is a SyncRequest message first we check if the contained target set matches our
+        // own locally configured one.
+        if let Message::SyncRequest(_, remote_supported_schema_ids) = message.message() {
+            let local_supported_schema_ids = &self
+                .announcement
+                .as_ref()
+                .expect("Announcement state needs to be set with 'update_announcement'")
+                .supported_schema_ids;
+
+            // If this node has been configured with a whitelist of schema ids then we check the
+            // target set of the requests matches our own, otherwise we skip this step and accept
+            // any target set.
+            if self.schema_provider.is_whitelist_active()
+                && !local_supported_schema_ids.is_valid_set(remote_supported_schema_ids)
             {
                 // If it doesn't match we signal that an error occurred and return at this point.
                 self.on_replication_error(peer, session_id, ReplicationError::UnsupportedTargetSet)
@@ -207,8 +267,9 @@ impl ConnectionManager {
         match self.sync_manager.handle_message(&peer, &message).await {
             Ok(result) => {
                 for message in result.messages {
-                    self.send_service_message(ServiceMessage::SentReplicationMessage(
-                        peer, message,
+                    self.send_service_message(ServiceMessage::SentMessage(
+                        peer,
+                        PeerMessage::SyncMessage(message),
                     ));
                 }
 
@@ -264,40 +325,63 @@ impl ConnectionManager {
         self.send_service_message(ServiceMessage::ReplicationFailed(peer));
     }
 
+    /// Generates our new announcement state we can then propagate to all known and future peers.
+    async fn update_announcement(&mut self) {
+        let target_set = self.target_set().await;
+        self.announcement = Some(Announcement::new(target_set));
+    }
+
     /// Determine if we can attempt new replication sessions with the peers we currently know
     /// about.
     async fn update_sessions(&mut self) {
-        // Determine the target set our node is interested in
-        let target_set = self.target_set().await;
-
-        if let Err(err) = target_set.validate() {
-            warn!("Not initiating replication: {err}");
-            return;
-        }
+        let local_supported_schema_ids = &self
+            .announcement
+            .as_ref()
+            .expect("Announcement state needs to be set with 'update_announcement'")
+            .supported_schema_ids;
 
         // Iterate through all currently connected peers
-        let mut attempt_peers: Vec<Peer> = self
+        let mut attempt_peers: Vec<(Peer, TargetSet)> = self
             .peers
             .clone()
             .into_iter()
-            .filter_map(|(peer, _)| {
+            .filter_map(|(peer, status)| {
                 let sessions = self.sync_manager.get_sessions(&peer);
 
-                // 1. Check if we're running too many sessions with that peer on this connection
+                // 1. Did we already receive this peers announcement state? If not we can't do
+                //    anything yet and need to wait.
+                let remote_supported_schema_ids: TargetSet =
+                    if let Some(announcement) = status.announcement {
+                        announcement.supported_schema_ids
+                    } else {
+                        return None;
+                    };
+
+                // 2. Calculate intersection of local and remote schema id sets. Do we have any
+                //    supported schema id's in common?
+                let target_set = TargetSet::from_intersection(
+                    local_supported_schema_ids,
+                    &remote_supported_schema_ids,
+                );
+                if target_set.is_empty() {
+                    return None;
+                }
+
+                // 3. Check if we're running too many sessions with that peer on this connection
                 //    already. This limit is configurable.
                 let active_sessions: Vec<&Session> = sessions
                     .iter()
                     .filter(|session| !session.is_done())
                     .collect();
 
-                // 2. Check if we're already having at least one session concerning the same target
+                // 4. Check if we're already having at least one session concerning the same target
                 //    set. If we would start that session again it would be considered an error.
                 let has_active_target_set_session = active_sessions
                     .iter()
                     .any(|session| session.target_set() == target_set);
 
                 if active_sessions.len() < MAX_SESSIONS_PER_PEER && !has_active_target_set_session {
-                    Some(peer)
+                    Some((peer, target_set))
                 } else {
                     None
                 }
@@ -312,8 +396,31 @@ impl ConnectionManager {
         attempt_peers.shuffle(&mut thread_rng());
         attempt_peers.truncate(MAX_PEER_SAMPLE);
 
-        for peer in attempt_peers {
-            self.initiate_replication(&peer, &target_set).await;
+        for (peer, target_set) in &attempt_peers {
+            self.initiate_replication(peer, target_set).await;
+        }
+    }
+
+    /// Send our local announcement state to all peers which are not informed yet.
+    async fn announce(&mut self) {
+        let local_announcement = self
+            .announcement
+            .as_ref()
+            .expect("Announcement state needs to be set with 'update_announcement'");
+
+        for (peer, status) in &self.peers {
+            if status.sent_our_announcement_timestamp < local_announcement.timestamp {
+                self.send_service_message(ServiceMessage::SentMessage(
+                    *peer,
+                    PeerMessage::Announce(AnnouncementMessage::new(local_announcement.clone())),
+                ));
+            }
+        }
+
+        for (_, status) in self.peers.iter_mut() {
+            if status.sent_our_announcement_timestamp < local_announcement.timestamp {
+                status.sent_our_announcement_timestamp = now();
+            }
         }
     }
 
@@ -326,8 +433,9 @@ impl ConnectionManager {
         {
             Ok(messages) => {
                 for message in messages {
-                    self.send_service_message(ServiceMessage::SentReplicationMessage(
-                        *peer, message,
+                    self.send_service_message(ServiceMessage::SentMessage(
+                        *peer,
+                        PeerMessage::SyncMessage(message),
                     ));
                 }
             }
@@ -346,9 +454,14 @@ impl ConnectionManager {
             ServiceMessage::PeerDisconnected(peer) => {
                 self.on_connection_closed(peer).await;
             }
-            ServiceMessage::ReceivedReplicationMessage(peer, message) => {
-                self.on_replication_message(peer, message).await;
-            }
+            ServiceMessage::ReceivedMessage(peer, message) => match message {
+                PeerMessage::SyncMessage(message) => {
+                    self.on_replication_message(peer, message).await;
+                }
+                PeerMessage::Announce(message) => {
+                    self.on_announcement_message(peer, message).await;
+                }
+            },
             _ => (), // Ignore all other messages
         }
     }
@@ -363,8 +476,15 @@ impl ConnectionManager {
 
     /// Main event loop running the async streams.
     pub async fn run(mut self) {
+        // Generate our own first announcement
+        self.update_announcement().await;
+
+        // Subscribe to updates when our target set got changed
+        let mut schema_provider_rx = self.schema_provider.on_schema_added();
+
         loop {
             tokio::select! {
+                // Service message arrived
                 event = self.rx.next() => match event {
                     Some(Ok(message)) => self.handle_service_message(message).await,
                     Some(Err(err)) => {
@@ -375,8 +495,15 @@ impl ConnectionManager {
                         return
                     },
                 },
+
+                // Target set got updated
+                Ok(_) = schema_provider_rx.recv() => {
+                    self.update_announcement().await;
+                },
+
+                // Announcement & replication schedule is due
                 Some(_) = self.scheduler.next() => {
-                    self.update_sessions().await
+                    self.on_update().await;
                 }
             }
         }
@@ -396,9 +523,11 @@ mod tests {
     use tokio::sync::broadcast;
 
     use crate::bus::ServiceMessage;
-    use crate::network::Peer;
+    use crate::network::{Peer, PeerMessage};
     use crate::replication::service::PeerStatus;
-    use crate::replication::{Message, Mode, SyncMessage, TargetSet};
+    use crate::replication::{
+        Announcement, AnnouncementMessage, Message, Mode, SyncMessage, TargetSet,
+    };
     use crate::schema::SchemaProvider;
     use crate::test_utils::{test_runner, TestNode};
 
@@ -422,6 +551,7 @@ mod tests {
             );
 
             let target_set = manager.target_set().await;
+            manager.update_announcement().await;
 
             // Inform connection manager about new peer
             let remote_peer = Peer::new(remote_peer_id, ConnectionId::new_unchecked(1));
@@ -433,20 +563,39 @@ mod tests {
             let status = manager
                 .peers
                 .get(&remote_peer)
-                .expect("Peer to be registered in connection manager");
+                .expect("Peer to be registered in connection manager")
+                .clone();
             assert_eq!(manager.peers.len(), 1);
             assert_eq!(status.peer, remote_peer);
+            assert!(status.sent_our_announcement_timestamp > 0);
 
-            // Manager attempts a replication session with that peer
+            // Manager announces target set with peer
             assert_eq!(rx.len(), 1);
             assert_eq!(
                 rx.recv().await,
-                Ok(ServiceMessage::SentReplicationMessage(
+                Ok(ServiceMessage::SentMessage(
                     remote_peer,
-                    SyncMessage::new(0, Message::SyncRequest(Mode::LogHeight, target_set))
+                    PeerMessage::Announce(AnnouncementMessage::new(Announcement::new(
+                        target_set.clone()
+                    )))
                 ))
             );
-            assert_eq!(manager.sync_manager.get_sessions(&remote_peer).len(), 1);
+
+            // Peer informs us about its target set
+            assert_eq!(status.announcement, None);
+            let announcement = Announcement::new(target_set.clone());
+            manager
+                .handle_service_message(ServiceMessage::ReceivedMessage(
+                    remote_peer.clone(),
+                    PeerMessage::Announce(AnnouncementMessage::new(announcement.clone())),
+                ))
+                .await;
+            let status = manager
+                .peers
+                .get(&remote_peer)
+                .expect("Peer to be registered in connection manager")
+                .clone();
+            assert_eq!(status.announcement, Some(announcement.clone()));
 
             // Inform manager about peer disconnected
             manager
@@ -473,6 +622,7 @@ mod tests {
             let schema_provider = SchemaProvider::new(vec![], Some(vec![]));
             let mut manager =
                 ConnectionManager::new(&schema_provider, &node.context.store, &tx, local_peer_id);
+            manager.update_announcement().await;
 
             let remote_peer = Peer::new(remote_peer_id, ConnectionId::new_unchecked(1));
 
@@ -486,12 +636,12 @@ mod tests {
             );
             let unsupported_target_set = TargetSet::new(&[unsupported_schema_id]);
             manager
-                .handle_service_message(ServiceMessage::ReceivedReplicationMessage(
+                .handle_service_message(ServiceMessage::ReceivedMessage(
                     remote_peer,
-                    SyncMessage::new(
+                    PeerMessage::SyncMessage(SyncMessage::new(
                         0,
                         Message::SyncRequest(Mode::LogHeight, unsupported_target_set),
-                    ),
+                    )),
                 ))
                 .await;
 
diff --git a/aquadoggo/src/replication/target_set.rs b/aquadoggo/src/replication/target_set.rs
index 95ca9afcd..f58e0b675 100644
--- a/aquadoggo/src/replication/target_set.rs
+++ b/aquadoggo/src/replication/target_set.rs
@@ -37,8 +37,10 @@ impl TargetSet {
         Self(deduplicated_set)
     }
 
-    pub fn contains(&self, schema_id: &SchemaId) -> bool {
-        self.0.contains(schema_id)
+    pub fn from_intersection(local_target_set: &TargetSet, remote_target_set: &TargetSet) -> Self {
+        let mut target_set = local_target_set.clone();
+        target_set.0.retain(|id| remote_target_set.contains(id));
+        target_set
     }
 
     fn from_untrusted(schema_ids: Vec<SchemaId>) -> Result<Self, TargetSetError> {
@@ -51,6 +53,19 @@ impl TargetSet {
         Ok(target_set)
     }
 
+    pub fn contains(&self, schema_id: &SchemaId) -> bool {
+        self.0.contains(schema_id)
+    }
+
+    /// Returns true if there are no unknown elements in external target set.
+    pub fn is_valid_set(&self, target_set: &TargetSet) -> bool {
+        !target_set.iter().any(|schema_id| !self.contains(schema_id))
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.0.is_empty()
+    }
+
     pub fn iter(&self) -> Iter<SchemaId> {
         self.0.iter()
     }
@@ -60,11 +75,6 @@ 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;
         let mut initial_system_schema = true;
 
@@ -167,6 +177,37 @@ mod tests {
         );
     }
 
+    #[rstest]
+    fn calculate_intersection(
+        #[from(random_document_view_id)] document_view_id_1: DocumentViewId,
+        #[from(random_document_view_id)] document_view_id_2: DocumentViewId,
+        #[from(random_document_view_id)] document_view_id_3: DocumentViewId,
+    ) {
+        // Correctly calculates intersections
+        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);
+        let schema_id_3 =
+            SchemaId::new_application(&SchemaName::new("events").unwrap(), &document_view_id_3);
+
+        let set_1 = TargetSet::new(&[schema_id_1.clone(), schema_id_2.clone()]);
+        let set_2 = TargetSet::new(&[schema_id_3.clone(), schema_id_2.clone()]);
+
+        assert_eq!(
+            TargetSet::from_intersection(&set_1, &set_2),
+            TargetSet::new(&[schema_id_2.clone()])
+        );
+
+        // Correctly verifies if both target sets know about all given elements
+        assert!(TargetSet::new(&[schema_id_2.clone()])
+            .is_valid_set(&TargetSet::new(&[schema_id_2.clone()])));
+        assert!(TargetSet::new(&[schema_id_3.clone(), schema_id_2.clone()])
+            .is_valid_set(&TargetSet::new(&[schema_id_2.clone()])));
+        assert!(!TargetSet::new(&[schema_id_1.clone()])
+            .is_valid_set(&TargetSet::new(&[schema_id_2.clone(), schema_id_1.clone()])));
+    }
+
     #[rstest]
     #[case(vec![
         "venues_0020c13cdc58dfc6f4ebd32992ff089db79980363144bdb2743693a019636fa72ec8".to_string(),
diff --git a/aquadoggo/src/schema/schema_provider.rs b/aquadoggo/src/schema/schema_provider.rs
index d548e68b5..7f67bd807 100644
--- a/aquadoggo/src/schema/schema_provider.rs
+++ b/aquadoggo/src/schema/schema_provider.rs
@@ -3,7 +3,7 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 
-use anyhow::{anyhow, Result};
+use anyhow::{bail, Result};
 use log::{debug, info};
 use p2panda_rs::schema::{Schema, SchemaId, SYSTEM_SCHEMAS};
 use p2panda_rs::Human;
@@ -18,9 +18,9 @@ pub struct SchemaProvider {
     /// In-memory store of registered schemas.
     schemas: Arc<Mutex<HashMap<SchemaId, Schema>>>,
 
-    /// Optional list of schema this provider supports. If set only these schema will be added to the schema
-    /// registry once materialized.
-    supported_schema_ids: Option<Vec<SchemaId>>,
+    /// Optional list of whitelisted schema ids. When set, only these schema ids will be accepted
+    /// on this node, if not set _all_ schema ids are accepted.
+    whitelisted_schema_ids: Option<Vec<SchemaId>>,
 
     /// Sender for broadcast channel informing subscribers about updated schemas.
     tx: Sender<SchemaId>,
@@ -30,7 +30,7 @@ impl SchemaProvider {
     /// Returns a `SchemaProvider` containing the given application schemas and all system schemas.
     pub fn new(
         application_schemas: Vec<Schema>,
-        supported_schema_ids: Option<Vec<SchemaId>>,
+        whitelisted_schema_ids: Option<Vec<SchemaId>>,
     ) -> Self {
         // Collect all system and application schemas.
         let mut schemas = SYSTEM_SCHEMAS.clone();
@@ -42,8 +42,8 @@ impl SchemaProvider {
             index.insert(schema.id().to_owned(), schema.to_owned());
         }
 
-        if let Some(supported_schema_ids) = &supported_schema_ids {
-            index.retain(|schema_id, _| supported_schema_ids.contains(schema_id));
+        if let Some(schema_ids) = &whitelisted_schema_ids {
+            index.retain(|id, _| schema_ids.contains(id));
         };
 
         let (tx, _) = channel(64);
@@ -59,7 +59,7 @@ impl SchemaProvider {
 
         Self {
             schemas: Arc::new(Mutex::new(index)),
-            supported_schema_ids,
+            whitelisted_schema_ids,
             tx,
         }
     }
@@ -84,11 +84,9 @@ impl SchemaProvider {
     /// Returns `true` if a schema was updated or it already existed in it's current state, and
     /// `false` if it was inserted.
     pub async fn update(&self, schema: Schema) -> Result<bool> {
-        if let Some(supported_schema) = self.supported_schema_ids.as_ref() {
-            if !supported_schema.contains(schema.id()) {
-                return Err(anyhow!(
-                    "Attempted to add unsupported schema to schema provider"
-                ));
+        if let Some(whitelisted_ids) = self.whitelisted_schema_ids.as_ref() {
+            if !whitelisted_ids.contains(schema.id()) {
+                bail!("Attempted to add unsupported schema to schema provider");
             }
         };
 
@@ -96,8 +94,8 @@ impl SchemaProvider {
         let schema_exists = schemas.get(schema.id()).is_some();
 
         if schema_exists {
-            // Return true here as the schema already exists in it's current state so we don't
-            // need to mutate the schema store or announce any change.
+            // Return true here as the schema already exists in it's current state so we don't need
+            // to mutate the schema store or announce any change.
             return Ok(true);
         }
 
@@ -114,9 +112,26 @@ impl SchemaProvider {
         Ok(is_update)
     }
 
-    // Return the configured supported schema.
-    pub fn supported_schema_ids(&self) -> Option<&Vec<SchemaId>> {
-        self.supported_schema_ids.as_ref()
+    /// Returns a list of all supported schema ids.
+    ///
+    /// If no whitelist was set it returns the list of all currently known schema ids. If a
+    /// whitelist was set it directly returns the list itself.
+    pub async fn supported_schema_ids(&self) -> Vec<SchemaId> {
+        match &self.whitelisted_schema_ids {
+            Some(schema_ids) => schema_ids.clone(),
+            None => self
+                .all()
+                .await
+                .iter()
+                .map(|schema| schema.id().to_owned())
+                .collect(),
+        }
+    }
+
+    /// Returns true if a whitelist of supported schema ids was provided through user
+    /// configuration.
+    pub fn is_whitelist_active(&self) -> bool {
+        self.whitelisted_schema_ids.is_some()
     }
 }
 
diff --git a/aquadoggo_cli/src/schemas.rs b/aquadoggo_cli/src/schemas.rs
index d29170c7b..bdddc548d 100644
--- a/aquadoggo_cli/src/schemas.rs
+++ b/aquadoggo_cli/src/schemas.rs
@@ -1,9 +1,10 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
-use anyhow::{anyhow, Result};
-use p2panda_rs::schema::SchemaId;
 use std::fs::File;
 use std::io::Read;
+
+use anyhow::{anyhow, Result};
+use p2panda_rs::schema::SchemaId;
 use toml::Table;
 
 pub fn read_schema_ids_from_file(file: &mut File) -> Result<Vec<SchemaId>> {
@@ -15,28 +16,3 @@ pub fn read_schema_ids_from_file(file: &mut File) -> Result<Vec<SchemaId>> {
     ))?;
     Ok(value.clone().try_into::<Vec<SchemaId>>()?)
 }
-
-#[cfg(test)]
-mod tests {
-    use std::fs::{create_dir_all, File};
-    use std::io::Write;
-
-    use tempfile::TempDir;
-
-    use super::read_schema_ids_from_file;
-
-    #[test]
-    fn reads_schema_ids_from_file() {
-        let tmp_dir = TempDir::new().unwrap();
-        let mut tmp_path = tmp_dir.path().to_owned();
-        tmp_path.push("schema.toml");
-
-        create_dir_all(tmp_path.parent().unwrap()).unwrap();
-        let mut file = File::create(&tmp_path).unwrap();
-        file.write_all("schemas = [\"plant_0020c11ab63099193a8c8516cc00f88bfc8cdd657b94a99fef2fce86aaaede84f87d\"]".as_bytes()).unwrap();
-
-        let result = read_schema_ids_from_file(&mut file);
-        println!("{result:?}");
-        assert!(result.is_ok());
-    }
-}


Test patch: