p2panda-sync 0.7.0

// SPDX-License-Identifier: MIT OR Apache-2.0

//! Manager for initiating and orchestrating topic log sync sessions.
//!
//! Concurrently running sessions perform message forwarding with de-duplication. Events from all
//! running sync sessions can be consumed via a single manager event stream.
mod event_stream;
mod session_map;

use std::collections::HashMap;
use std::fmt::Debug;
use std::hash::Hash as StdHash;
use std::marker::PhantomData;
use std::pin::Pin;

use futures_channel::mpsc;
use futures_util::future::ready;
use futures_util::sink::{Sink, SinkExt};
use futures_util::stream::{SelectAll, Stream, StreamExt};
use p2panda_core::{Extensions, Hash, LogId, Operation, SeqNum, VerifyingKey};
use p2panda_store::logs::LogStore;
use p2panda_store::topics::TopicStore;
use serde::{Deserialize, Serialize};
use thiserror::Error;
use tokio::sync::broadcast;
use tokio_stream::wrappers::BroadcastStream;
use tokio_stream::wrappers::errors::BroadcastStreamRecvError;
use tracing::debug;

use crate::dedup::DeduplicationBuffer;
use crate::manager::event_stream::{ManagerEventStreamState, StreamDebug};
use crate::protocols::{TopicLogSync, TopicLogSyncError, TopicLogSyncEvent};
use crate::traits::Manager;
use crate::{FromSync, SessionConfig, ToSync};

pub use event_stream::ManagerEventStream;
pub use session_map::SessionTopicMap;

static CHANNEL_BUFFER: usize = 1028;

pub type ToTopicSync<E> = ToSync<Operation<E>>;

/// Create and manage topic log sync sessions.
///
/// Sync sessions are created via the manager, which instantiates them with access to the shared
/// topic map and operation store as well as channels for receiving sync events and for sending
/// newly arriving operations in live mode.
///
/// A handle can be acquired to a sync session via the session_handle method for sending any live
/// mode operations to a specific session. It's expected that users map sessions (by their id) to
/// any topic subscriptions in order to understand the correct mappings.
///
/// There are two points where message deduplication occurs; 1) per-subscription deduplication of
/// received operations across all sessions for this manager occurs in the event stream 2)
/// per-session deduplication occurs on sending operations to a remote, this is especially
/// required during live-mode when we may receive the same new operation from several sources. The
/// former stops any local consumers from receiving the same operation more than once, the latter
/// reduces unnecessary noise on the network.
#[allow(clippy::type_complexity)]
#[derive(Debug)]
pub struct TopicSyncManager<T, S, L, E>
where
    T: Clone,
    E: Extensions,
{
    store: S,
    session_topic_map: SessionTopicMap<T, mpsc::Sender<ToTopicSync<E>>>,
    from_session_tx: HashMap<(u64, VerifyingKey), broadcast::Sender<TopicLogSyncEvent<E>>>,
    from_session_rx: HashMap<(u64, VerifyingKey), broadcast::Receiver<TopicLogSyncEvent<E>>>,
    manager_tx: Vec<mpsc::Sender<SessionStream<T, E>>>,
    _phantom: PhantomData<L>,
}

#[derive(Debug)]
pub(crate) struct SessionStream<T, E>
where
    T: Clone,
    E: Clone,
{
    pub session_id: u64,
    pub topic: T,
    pub remote: VerifyingKey,
    pub event_rx: broadcast::Receiver<TopicLogSyncEvent<E>>,
    pub live_tx: mpsc::Sender<ToTopicSync<E>>,
}

impl<T, S, L, E> TopicSyncManager<T, S, L, E>
where
    T: Clone,
    E: Extensions,
{
    pub fn new(store: S) -> Self {
        Self {
            store,
            manager_tx: Default::default(),
            session_topic_map: Default::default(),
            from_session_tx: Default::default(),
            from_session_rx: Default::default(),
            _phantom: PhantomData,
        }
    }
}

impl<T, S, L, E> Manager<T> for TopicSyncManager<T, S, L, E>
where
    T: Clone + Debug + Eq + StdHash + Serialize + for<'a> Deserialize<'a> + Send + 'static,
    S: LogStore<Operation<E>, VerifyingKey, L, SeqNum, Hash>
        + TopicStore<T, VerifyingKey, L>
        + Clone
        + Send
        + 'static,
    L: LogId + Debug + Send + 'static,
    E: Extensions + Send + 'static,
{
    type Protocol = TopicLogSync<T, S, L, E>;
    type Args = S;
    type Event = TopicLogSyncEvent<E>;
    type Message = Operation<E>;
    type Error = TopicSyncManagerError;

    /// Instantiate a manager from arguments.
    fn from_args(store: Self::Args) -> Self {
        Self::new(store)
    }

    /// Instantiate a new sync session.
    async fn session(&mut self, session_id: u64, config: &SessionConfig<T>) -> Self::Protocol {
        let (live_tx, live_rx) = mpsc::channel(CHANNEL_BUFFER);
        let (event_tx, event_rx) = broadcast::channel::<Self::Event>(CHANNEL_BUFFER);

        self.from_session_tx
            .insert((session_id, config.remote), event_tx.clone());

        self.from_session_rx
            .insert((session_id, config.remote), event_rx);

        self.session_topic_map
            .insert_with_topic(session_id, config.topic.clone(), live_tx.clone());

        for manager_tx in self.manager_tx.iter_mut() {
            if let Err(err) = manager_tx
                .send(SessionStream {
                    session_id,
                    topic: config.topic.clone(),
                    remote: config.remote,
                    event_rx: event_tx.subscribe(),
                    live_tx: live_tx.clone(),
                })
                .await
            {
                debug!("manager handle dropped: {err:?}");
            };
        }

        let live_rx = if config.live_mode {
            Some(live_rx)
        } else {
            None
        };

        TopicLogSync::new(config.topic.clone(), self.store.clone(), live_rx, event_tx)
    }

    /// Retrieve a handle to a running sync session.
    async fn session_handle(
        &self,
        session_id: u64,
    ) -> Option<Pin<Box<dyn Sink<ToTopicSync<E>, Error = Self::Error>>>> {
        let map_fn = |to_sync: ToSync<Self::Message>| {
            ready({
                match to_sync {
                    ToSync::Payload(operation) => Ok::<_, Self::Error>(ToSync::Payload(operation)),
                    ToSync::Close => Ok::<_, Self::Error>(ToSync::Close),
                }
            })
        };

        self.session_topic_map.sender(session_id).map(|tx| {
            Box::pin(tx.clone().with(map_fn))
                as Pin<Box<dyn Sink<ToTopicSync<E>, Error = Self::Error>>>
        })
    }

    /// Subscribe to the event stream for all running sync sessions.
    fn subscribe(&mut self) -> impl Stream<Item = FromSync<Self::Event>> + Send + Unpin + 'static {
        let (manager_tx, manager_rx) = mpsc::channel(CHANNEL_BUFFER);
        self.manager_tx.push(manager_tx);

        let mut session_rx_set = SelectAll::new();
        for ((id, remote), tx) in self.from_session_tx.iter() {
            let session_id = *id;
            let remote = *remote;
            let stream = BroadcastStream::new(tx.subscribe());

            #[allow(clippy::type_complexity)]
            let stream: Pin<
                Box<dyn StreamDebug<Option<FromSync<TopicLogSyncEvent<E>>>>>,
            > = Box::pin(stream.map(Box::new(
                move |event: Result<TopicLogSyncEvent<E>, BroadcastStreamRecvError>| {
                    event.ok().map(|event| FromSync {
                        session_id,
                        remote,
                        event,
                    })
                },
            )));
            session_rx_set.push(stream);
        }

        let state = ManagerEventStreamState {
            manager_rx,
            session_rx_set,
            session_topic_map: self.session_topic_map.clone(),
            dedup: DeduplicationBuffer::default(),
        };

        let stream = ManagerEventStream {
            state: Some(state),
            pending: Default::default(),
        };

        Box::pin(stream)
    }
}

/// Error types which can be returned from `TopicSyncManager`.
#[derive(Debug, Error)]
pub enum TopicSyncManagerError {
    #[error(transparent)]
    TopicLogSync(#[from] TopicLogSyncError),

    #[error("received operation before topic agreed")]
    OperationBeforeTopic,

    #[error(transparent)]
    Send(#[from] mpsc::SendError),
}
#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;
    use std::time::Duration;

    use futures_channel::mpsc;
    use futures_util::{SinkExt, StreamExt};
    use p2panda_core::test_utils::setup_logging;
    use p2panda_core::{Body, Operation, Topic};
    use p2panda_store::SqliteStore;

    use crate::protocols::TopicLogSyncEvent;
    use crate::test_utils::{Peer, TestLogId, TestTopicSyncManager, drain_stream, run_protocol};
    use crate::traits::{Manager, Protocol};
    use crate::{FromSync, SessionConfig, ToSync};

    #[tokio::test]
    async fn from_args() {
        let store = SqliteStore::temporary().await;
        let _: TestTopicSyncManager = Manager::from_args(store);
    }

    #[tokio::test]
    async fn manager_e2e() {
        setup_logging();

        const LOG_ID: TestLogId = 0;
        const SESSION_ID: u64 = 0;

        let topic = Topic::random();

        // Setup Peer A
        let mut peer_a = Peer::new(0).await;
        let body = Body::new("Hello from Peer A".as_bytes());
        let _ = peer_a.create_operation(&body, LOG_ID).await;
        let logs = BTreeMap::from([(peer_a.id(), vec![LOG_ID])]);
        peer_a.associate(&topic, &logs).await;
        let mut peer_a_manager = TestTopicSyncManager::new(peer_a.store.clone());

        // Setup Peer B
        let mut peer_b = Peer::new(1).await;
        let body = Body::new("Hello from Peer B".as_bytes());
        let _ = peer_b.create_operation(&body, LOG_ID).await;
        let logs = BTreeMap::from([(peer_b.id(), vec![LOG_ID])]);
        peer_b.associate(&topic, &logs).await;
        let mut peer_b_manager = TestTopicSyncManager::new(peer_b.store.clone());

        let config = SessionConfig {
            topic,
            remote: peer_b.id(),
            live_mode: true,
        };

        // Subscribe to both managers.
        let mut event_stream_a = peer_a_manager.subscribe();
        let mut event_stream_b = peer_b_manager.subscribe();

        // Instantiate sync sessions.
        let peer_a_session = peer_a_manager.session(SESSION_ID, &config).await;
        let peer_b_session = peer_b_manager.session(SESSION_ID, &config).await;

        // Get a handle to Peer A sync session.
        let mut peer_a_handle = peer_a_manager.session_handle(SESSION_ID).await.unwrap();

        // Create and send a new live-mode message.
        let (header_1, _) = peer_a.create_operation_no_insert(&body, LOG_ID).await;
        peer_a_handle
            .send(ToSync::Payload(Operation {
                hash: header_1.hash(),
                header: header_1.clone(),
                body: Some(body.clone()),
            }))
            .await
            .unwrap();
        peer_a_handle.send(ToSync::Close).await.unwrap();

        // Actually run the protocol.
        run_protocol(peer_a_session, peer_b_session).await.unwrap();

        // Assert Peer A's events.
        for index in 0..=4 {
            let event = event_stream_a.next().await.unwrap();
            assert_eq!(event.session_id(), 0);
            match index {
                0 => std::assert_matches!(
                    event,
                    FromSync {
                        event: TopicLogSyncEvent::SyncStarted { .. },
                        ..
                    }
                ),
                1 => std::assert_matches!(
                    event,
                    FromSync {
                        event: TopicLogSyncEvent::OperationReceived { .. },
                        ..
                    }
                ),
                2 => std::assert_matches!(
                    event,
                    FromSync {
                        event: TopicLogSyncEvent::SyncFinished { .. },
                        ..
                    }
                ),
                3 => std::assert_matches!(
                    event,
                    FromSync {
                        event: TopicLogSyncEvent::LiveModeStarted,
                        ..
                    }
                ),
                4 => std::assert_matches!(
                    event,
                    FromSync {
                        event: TopicLogSyncEvent::SessionFinished { .. },
                        ..
                    }
                ),
                _ => panic!(),
            }
        }

        // Assert Peer B's events.
        for index in 0..=5 {
            let event = event_stream_b.next().await.unwrap();
            match index {
                0 => std::assert_matches!(
                    event,
                    FromSync {
                        session_id: 0,
                        event: TopicLogSyncEvent::SyncStarted { .. },
                        ..
                    }
                ),
                1 => std::assert_matches!(
                    event,
                    FromSync {
                        session_id: 0,
                        event: TopicLogSyncEvent::OperationReceived { .. },
                        ..
                    }
                ),
                2 => std::assert_matches!(
                    event,
                    FromSync {
                        session_id: 0,
                        event: TopicLogSyncEvent::SyncFinished { .. },
                        ..
                    }
                ),
                3 => std::assert_matches!(
                    event,
                    FromSync {
                        event: TopicLogSyncEvent::LiveModeStarted,
                        ..
                    }
                ),
                4 => std::assert_matches!(
                    event,
                    FromSync {
                        session_id: 0,
                        event: TopicLogSyncEvent::OperationReceived { .. },
                        ..
                    }
                ),
                5 => std::assert_matches!(
                    event,
                    FromSync {
                        event: TopicLogSyncEvent::SessionFinished { .. },
                        ..
                    }
                ),
                _ => panic!(),
            }
        }
    }

    #[tokio::test]
    async fn live_mode_three_peer_forwarding() {
        setup_logging();

        const LOG_ID: TestLogId = 0;
        const SESSION_AB: u64 = 0;
        const SESSION_AC: u64 = 1;
        const SESSION_BA: u64 = 2;
        const SESSION_CA: u64 = 3;

        let topic = Topic::random();

        // Peer A
        let mut peer_a = Peer::new(0).await;
        let body_a = Body::new("Hello from A".as_bytes());
        let (peer_a_header_0, _) = peer_a.create_operation(&body_a, LOG_ID).await;
        let mut manager_a = TestTopicSyncManager::new(peer_a.store.clone());

        // Peer B
        let mut peer_b = Peer::new(1).await;
        let body_b = Body::new("Hello from B".as_bytes());
        let (peer_b_header_0, _) = peer_b.create_operation(&body_b, LOG_ID).await;
        let mut manager_b = TestTopicSyncManager::new(peer_b.store.clone());

        // Peer C
        let mut peer_c = Peer::new(2).await;
        let body_c = Body::new("Hello from C".as_bytes());
        let (peer_c_header_0, _) = peer_c.create_operation(&body_c, LOG_ID).await;
        let mut manager_c = TestTopicSyncManager::new(peer_c.store.clone());

        let logs = BTreeMap::from([
            (peer_a.id(), vec![LOG_ID]),
            (peer_b.id(), vec![LOG_ID]),
            (peer_c.id(), vec![LOG_ID]),
        ]);
        peer_a.associate(&topic, &logs).await;
        peer_b.associate(&topic, &logs).await;
        peer_c.associate(&topic, &logs).await;

        // Session A -> B
        let mut config = SessionConfig {
            topic: topic.clone(),
            remote: peer_b.id(),
            live_mode: true,
        };
        let session_ab = manager_a.session(SESSION_AB, &config).await;
        config.remote = peer_a.id();
        let session_ba = manager_b.session(SESSION_BA, &config).await;

        // Session A -> C
        let mut config = SessionConfig {
            topic: topic.clone(),
            remote: peer_c.id(),
            live_mode: true,
        };
        let session_ac = manager_a.session(SESSION_AC, &config).await;
        config.remote = peer_a.id();
        let session_ca = manager_c.session(SESSION_CA, &config).await;

        let mut event_stream_a = manager_a.subscribe();
        let mut event_stream_b = manager_b.subscribe();
        let mut event_stream_c = manager_c.subscribe();

        // Run both protocols concurrently.
        {
            let (mut local_message_tx, local_message_rx) = mpsc::channel(128);
            let (mut remote_message_tx, remote_message_rx) = mpsc::channel(128);
            let mut local_message_rx = local_message_rx.map(Ok::<_, ()>);
            let mut remote_message_rx = remote_message_rx.map(Ok::<_, ()>);
            tokio::spawn(async move {
                session_ab
                    .run(&mut local_message_tx, &mut remote_message_rx)
                    .await
                    .unwrap();
            });
            tokio::spawn(async move {
                session_ba
                    .run(&mut remote_message_tx, &mut local_message_rx)
                    .await
                    .unwrap();
            });
        }

        {
            let (mut local_message_tx, local_message_rx) = mpsc::channel(128);
            let (mut remote_message_tx, remote_message_rx) = mpsc::channel(128);
            let mut local_message_rx = local_message_rx.map(Ok::<_, ()>);
            let mut remote_message_rx = remote_message_rx.map(Ok::<_, ()>);
            tokio::spawn(async move {
                session_ac
                    .run(&mut local_message_tx, &mut remote_message_rx)
                    .await
                    .unwrap();
            });
            tokio::spawn(async move {
                session_ca
                    .run(&mut remote_message_tx, &mut local_message_rx)
                    .await
                    .unwrap();
            });
        }

        // Send live-mode messages from all peers.
        let mut handle_ab = manager_a.session_handle(SESSION_AB).await.unwrap();
        let mut handle_ac = manager_a.session_handle(SESSION_AC).await.unwrap();
        let mut handle_ba = manager_b.session_handle(SESSION_BA).await.unwrap();
        let mut handle_ca = manager_c.session_handle(SESSION_CA).await.unwrap();

        let body_a = Body::new("Hello again from A".as_bytes());
        let body_b = Body::new("Hello again from B".as_bytes());
        let body_c = Body::new("Hello again from C".as_bytes());
        let (peer_a_header_1, _) = peer_a.create_operation(&body_a, LOG_ID).await;
        let (peer_b_header_1, _) = peer_b.create_operation(&body_b, LOG_ID).await;
        let (peer_c_header_1, _) = peer_c.create_operation(&body_c, LOG_ID).await;

        let operation_a = Operation {
            hash: peer_a_header_1.hash(),
            header: peer_a_header_1.clone(),
            body: Some(body_a.clone()),
        };
        let operation_b = Operation {
            hash: peer_b_header_1.hash(),
            header: peer_b_header_1.clone(),
            body: Some(body_b.clone()),
        };
        let operation_c = Operation {
            hash: peer_c_header_1.hash(),
            header: peer_c_header_1.clone(),
            body: Some(body_c.clone()),
        };

        handle_ab
            .send(ToSync::Payload(operation_a.clone()))
            .await
            .unwrap();
        handle_ac.send(ToSync::Payload(operation_a)).await.unwrap();
        handle_ba.send(ToSync::Payload(operation_b)).await.unwrap();
        handle_ca.send(ToSync::Payload(operation_c)).await.unwrap();

        // Collect all operations each peer receives on the event stream.
        let mut operations_a = vec![];
        let mut operations_b = vec![];
        let mut operations_c = vec![];
        let _ = tokio::time::timeout(Duration::from_millis(500), async {
            loop {
                tokio::select! {
                    Some(event) = event_stream_a.next() => {
                        if let TopicLogSyncEvent::OperationReceived { operation, .. } = event.event() {
                            operations_a.push(operation.header().clone());
                        }
                    }
                    Some(event) = event_stream_b.next() => {
                        if let TopicLogSyncEvent::OperationReceived { operation, .. } = event.event() {
                            operations_b.push(operation.header().clone());
                        }
                    }
                    Some(event) = event_stream_c.next() => {
                        if let TopicLogSyncEvent::OperationReceived { operation, .. } = event.event() {
                            operations_c.push(operation.header().clone());
                        }
                    }
                    else => tokio::time::sleep(Duration::from_millis(20)).await
                }
            }
        })
        .await;

        // A receives 2 operations each from B & C
        assert_eq!(operations_a.len(), 4);
        // B receives 2 operations each from A & C (via A)
        assert_eq!(operations_b.len(), 4);
        // C receives 2 operations each from A & B (via A)
        assert_eq!(operations_c.len(), 4);

        // No peers received their own operations back again.
        assert!(!operations_a.contains(&peer_a_header_0));
        assert!(!operations_a.contains(&peer_a_header_1));
        assert!(!operations_b.contains(&peer_b_header_0));
        assert!(!operations_b.contains(&peer_b_header_1));
        assert!(!operations_c.contains(&peer_c_header_0));
        assert!(!operations_c.contains(&peer_c_header_1));
    }

    #[tokio::test]
    async fn non_blocking_manager_stream() {
        const LOG_ID: TestLogId = 0;
        const SESSION_ID: u64 = 0;

        let topic = Topic::random();

        // Setup Peer A
        let mut peer_a = Peer::new(0).await;
        let body = Body::new("Hello from Peer A".as_bytes());
        let _ = peer_a.create_operation(&body, LOG_ID).await;
        let logs = BTreeMap::from([(peer_a.id(), vec![LOG_ID])]);
        peer_a.associate(&topic, &logs).await;
        let mut peer_a_manager = TestTopicSyncManager::new(peer_a.store.clone());

        // Spawn a task polling peer a's manager stream.
        let mut peer_a_stream = peer_a_manager.subscribe();
        tokio::task::spawn(async move {
            loop {
                peer_a_stream.next().await;
            }
        });

        // Setup Peer B
        let mut peer_b = Peer::new(1).await;
        let body = Body::new("Hello from Peer B".as_bytes());
        let _ = peer_b.create_operation(&body, LOG_ID).await;
        let logs = BTreeMap::from([(peer_b.id(), vec![LOG_ID])]);
        peer_b.associate(&topic, &logs).await;
        let mut peer_b_manager = TestTopicSyncManager::new(peer_b.store.clone());

        let config = SessionConfig {
            topic,
            remote: peer_b.id(),
            live_mode: true,
        };

        let peer_a_session = peer_a_manager.session(SESSION_ID, &config).await;

        let event_stream = peer_b_manager.subscribe();
        let peer_b_session = peer_b_manager.session(SESSION_ID, &config).await;

        // Get a handle to Peer A sync session.
        let mut peer_a_handle = peer_a_manager.session_handle(SESSION_ID).await.unwrap();

        // Create and send a new live-mode message.
        let (header_1, _) = peer_a.create_operation_no_insert(&body, LOG_ID).await;
        peer_a_handle
            .send(ToSync::Payload(Operation {
                hash: header_1.hash(),
                header: header_1.clone(),
                body: Some(body.clone()),
            }))
            .await
            .unwrap();
        peer_a_handle.send(ToSync::Close).await.unwrap();

        // Actually run the protocol.
        run_protocol(peer_a_session, peer_b_session).await.unwrap();

        // Assert Peer B's events.
        let events = drain_stream(event_stream);
        assert_eq!(events.len(), 6);
        for (index, event) in events.into_iter().enumerate() {
            match index {
                0 => std::assert_matches!(
                    event,
                    FromSync {
                        session_id: 0,
                        event: TopicLogSyncEvent::SyncStarted { .. },
                        ..
                    }
                ),
                1 => std::assert_matches!(
                    event,
                    FromSync {
                        session_id: 0,
                        event: TopicLogSyncEvent::OperationReceived { .. },
                        ..
                    }
                ),
                2 => std::assert_matches!(
                    event,
                    FromSync {
                        session_id: 0,
                        event: TopicLogSyncEvent::SyncFinished { .. },
                        ..
                    }
                ),
                3 => std::assert_matches!(
                    event,
                    FromSync {
                        event: TopicLogSyncEvent::LiveModeStarted,
                        ..
                    }
                ),
                4 => std::assert_matches!(
                    event,
                    FromSync {
                        session_id: 0,
                        event: TopicLogSyncEvent::OperationReceived { .. },
                        ..
                    }
                ),
                5 => std::assert_matches!(
                    event,
                    FromSync {
                        event: TopicLogSyncEvent::SessionFinished { .. },
                        ..
                    }
                ),
                _ => panic!(),
            }
        }
    }
}