p2panda_sync/protocols/
topic_log_sync.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3//! Two-party sync protocol over a topic associated with a collection of append-only logs.
4use std::collections::BTreeMap;
5use std::fmt::Debug;
6use std::hash::Hash as StdHash;
7use std::marker::PhantomData;
8use std::pin::Pin;
9use std::task::{Context, Poll};
10
11use futures::channel::mpsc;
12use futures::{Sink, SinkExt, Stream, StreamExt};
13use p2panda_core::{Body, Extensions, Hash, Header, LogId, Operation, VerifyingKey};
14use p2panda_store::logs::LogStore;
15use p2panda_store::topics::TopicStore;
16use pin_project_lite::pin_project;
17use serde::{Deserialize, Serialize};
18use thiserror::Error;
19use tokio::sync::broadcast;
20use tracing::{Level, debug, enabled, trace, warn};
21
22use crate::ToSync;
23use crate::dedup::DEFAULT_BUFFER_CAPACITY;
24use crate::protocols::ShortFormat;
25use crate::protocols::log_sync::{
26    LogSync, LogSyncError, LogSyncEvent, LogSyncMessage, LogSyncMetrics,
27};
28use crate::traits::Protocol;
29
30/// Protocol for synchronizing logs which are associated with a generic T topic.
31///
32/// The mapping of T to a set of logs is handled on the application layer using an implementation
33/// of the `TopicStore` trait.
34///
35/// After sync is complete peers optionally enter "live-mode" where concurrently received and
36/// future messages will be sent directly to the application layer and forwarded to any
37/// concurrently running sync sessions. As we may receive messages from many sync sessions
38/// concurrently, messages forwarded to a sync session in live-mode are de-duplicated in order to
39/// avoid flooding the network with redundant data.
40///
41/// It is assumed that the T topic has been negotiated between parties prior to initiating this
42/// sync protocol.
43#[derive(Debug)]
44pub struct TopicLogSync<T, S, L, E> {
45    pub topic: T,
46    pub store: S,
47    pub event_tx: broadcast::Sender<TopicLogSyncEvent<E>>,
48    pub live_mode_rx: Option<mpsc::Receiver<ToSync<Operation<E>>>>,
49    pub buffer_capacity: usize,
50    pub _phantom: PhantomData<L>,
51}
52
53impl<T, S, L, E> TopicLogSync<T, S, L, E>
54where
55    T: Eq + StdHash + Serialize + for<'a> Deserialize<'a>,
56    S: LogStore<Operation<E>, VerifyingKey, L, u64, Hash>
57        + TopicStore<T, VerifyingKey, L>
58        + Clone
59        + Send
60        + 'static,
61    L: LogId,
62    E: Extensions,
63{
64    /// Returns a new sync protocol instance, configured with a store and `TopicStore` implementation
65    /// which associates the to-be-synced logs with a given topic.
66    pub fn new(
67        topic: T,
68        store: S,
69        live_mode_rx: Option<mpsc::Receiver<ToSync<Operation<E>>>>,
70        event_tx: broadcast::Sender<TopicLogSyncEvent<E>>,
71    ) -> Self {
72        Self::new_with_capacity(
73            topic,
74            store,
75            live_mode_rx,
76            event_tx,
77            DEFAULT_BUFFER_CAPACITY,
78        )
79    }
80
81    /// Instantiates a sync protocol with custom buffer capacity.
82    pub fn new_with_capacity(
83        topic: T,
84        store: S,
85        live_mode_rx: Option<mpsc::Receiver<ToSync<Operation<E>>>>,
86        event_tx: broadcast::Sender<TopicLogSyncEvent<E>>,
87        buffer_capacity: usize,
88    ) -> Self {
89        Self {
90            topic,
91            store,
92            event_tx,
93            live_mode_rx,
94            buffer_capacity,
95            _phantom: PhantomData,
96        }
97    }
98}
99
100impl<T, S, L, E> Protocol for TopicLogSync<T, S, L, E>
101where
102    T: Debug + Eq + StdHash + Serialize + for<'a> Deserialize<'a> + Send + 'static,
103    S: LogStore<Operation<E>, VerifyingKey, L, u64, Hash>
104        + TopicStore<T, VerifyingKey, L>
105        + Clone
106        + Send
107        + 'static,
108    L: LogId + Debug + Send + 'static,
109    E: Extensions + Send + 'static,
110{
111    type Error = TopicLogSyncError;
112    type Message = TopicLogSyncMessage<L, E>;
113    type Output = ();
114
115    async fn run(
116        self,
117        mut sink: &mut (impl Sink<Self::Message, Error = impl Debug> + Unpin),
118        mut stream: &mut (impl Stream<Item = Result<Self::Message, impl Debug>> + Unpin),
119    ) -> Result<Self::Output, Self::Error> {
120        // TODO: check there is overlap between the local and remote topic filters and end the
121        // session now if not.
122        debug!(
123            live_mode = self.live_mode_rx.is_some(),
124            "start sync session"
125        );
126
127        // Get the log ids which are associated with this topic query.
128        let logs = self
129            .store
130            .resolve(&self.topic)
131            .await
132            .map_err(|err| TopicLogSyncError::TopicStore(err.to_string()))?;
133
134        if enabled!(Level::DEBUG) {
135            let display_logs: BTreeMap<String, usize> =
136                logs.iter().map(|(k, v)| (k.fmt_short(), v.len())).collect();
137            debug!(logs = ?display_logs, "local topic logs retrieved");
138        }
139
140        if enabled!(Level::TRACE) {
141            let trace_logs: Vec<(String, &Vec<L>)> =
142                logs.iter().map(|(k, v)| (k.fmt_short(), v)).collect();
143            trace!(logs = ?trace_logs, "local topic logs retrieved");
144        }
145
146        // Run the log sync protocol passing in our local topic logs.
147        let (mut dedup, sync_metrics) = {
148            let (mut log_sync_sink, mut log_sync_stream) = sync_channels(&mut sink, &mut stream);
149            let protocol = LogSync::new_with_capacity(
150                self.store.clone(),
151                logs,
152                self.event_tx.clone(),
153                self.buffer_capacity,
154            );
155            let result = protocol.run(&mut log_sync_sink, &mut log_sync_stream).await;
156
157            // If the log sync session ended with an error, then send a "failed" event and return
158            // here with the error itself.
159            match result {
160                Ok((dedup, metrics)) => {
161                    self.event_tx
162                        .send(TopicLogSyncEvent::SyncFinished {
163                            metrics: metrics.clone().into(),
164                        })
165                        .map_err(|_| TopicLogSyncChannelError::EventSend)?;
166
167                    (dedup, metrics)
168                }
169                Err(err) => {
170                    self.event_tx
171                        .send(TopicLogSyncEvent::Failed {
172                            error: err.to_string(),
173                        })
174                        .map_err(|_| TopicLogSyncChannelError::EventSend)?;
175
176                    log_sync_sink
177                        .close()
178                        .await
179                        .map_err(|err| TopicLogSyncChannelError::MessageSink(format!("{err:?}")))?;
180
181                    return Err(err.into());
182                }
183            }
184        };
185
186        let mut metrics: Metrics = sync_metrics.into();
187
188        let result = match self.live_mode_rx {
189            None => Ok(()),
190            Some(mut live_mode_rx) => {
191                // Enter live-mode.
192                //
193                // In live-mode we process messages sent from the remote peer and received locally from a
194                // subscription or other concurrent sync sessions. In both cases we should deduplicate
195                // messages and also check they are part of our topic sub-set selection before forwarding
196                // them on the event stream, or to the remote peer.
197                let mut close_sent = false;
198                self.event_tx
199                    .send(TopicLogSyncEvent::LiveModeStarted)
200                    .map_err(|_| TopicLogSyncChannelError::EventSend)?;
201
202                loop {
203                    tokio::select! {
204                        biased;
205                        Some(message) = live_mode_rx.next() => {
206                            match message {
207                                ToSync::Payload(operation) => {
208                                    if !dedup.insert(operation.hash) {
209                                        trace!(id = ?operation.hash.fmt_short(), "ignore duplicate operation sent on live-mode channel");
210                                        continue;
211                                    }
212
213                                    metrics.sent_live_bytes +=
214                                        operation.header.to_bytes().len() as u64 + operation.header.payload_size;
215                                    metrics.sent_live_operations += 1;
216
217                                    trace!(
218                                        phase = "live",
219                                        id = ?operation.hash.fmt_short(),
220                                        sent_ops = ?metrics.sent_live_operations,
221                                        sent_bytes = ?metrics.sent_live_bytes,
222                                        "sent operation"
223                                    );
224
225                                    let result = sink
226                                        .send(TopicLogSyncMessage::Live(
227                                            operation.header,
228                                            operation.body,
229                                        ))
230                                        .await
231                                        .map_err(|err| TopicLogSyncChannelError::MessageSink(format!("{err:?}")).into());
232
233                                    if result.is_err() {
234                                        break result;
235                                    };
236                                }
237                                ToSync::Close => {
238                                    // We send the close and wait for the remote to close the
239                                    // connection.
240
241                                    debug!("closing sync session");
242
243                                    let result = sink
244                                        .send(TopicLogSyncMessage::Close)
245                                        .await
246                                        .map_err(|err| TopicLogSyncChannelError::MessageSink(format!("{err:?}")).into());
247                                    if result.is_err() {
248                                        break result;
249                                    };
250                                    close_sent = true;
251                                }
252                            };
253                        }
254                        message = stream.next() => {
255                            let Some(message) = message else {
256                                if close_sent {
257                                    break Ok(());
258                                }
259                                break Err(TopicLogSyncError::UnexpectedStreamClosure);
260                            };
261
262                            match message {
263                                Ok(message) => {
264                                    if let TopicLogSyncMessage::Close = message {
265                                        // We received the remotes close message and should close the
266                                        // connection ourselves.
267                                        debug!("received close message from remote");
268                                        break Ok(());
269                                    };
270
271                                    let TopicLogSyncMessage::Live(header, body) = message else {
272                                        break Err(TopicLogSyncError::UnexpectedProtocolMessage(
273                                            message.to_string(),
274                                        ));
275                                    };
276
277                                    // TODO: check that this message is a part of our topic T set.
278
279                                    // Insert operation hash into deduplication buffer and if it was
280                                    // previously present do not forward the operation to the application
281                                    // layer.
282                                    if !dedup.insert(header.hash()) {
283                                        trace!(phase = "live", operation_id = ?header.hash().fmt_short(), "ignore duplicate operation sent from remote");
284                                        continue;
285                                    }
286
287                                    metrics.received_live_bytes += header.to_bytes().len() as u64 + header.payload_size;
288                                    metrics.received_live_operations += 1;
289
290                                    trace!(
291                                        phase = "live",
292                                        operation_id = ?header.hash().fmt_short(),
293                                        received_ops = %metrics.received_live_operations,
294                                        received_bytes = %metrics.received_live_bytes,
295                                        "received operation"
296                                    );
297
298                                    self.event_tx
299                                        .send(TopicLogSyncEvent::OperationReceived{operation: Box::new(Operation {
300                                            hash: header.hash(),
301                                            header,
302                                            body,
303                                        }), metrics: metrics.clone()})
304                                        .map_err(|_| TopicLogSyncChannelError::EventSend)?;
305                                }
306                                Err(err) => {
307                                    if close_sent {
308                                        break Ok(());
309                                    }
310                                    break Err(TopicLogSyncError::DecodeMessage(format!("{err:?}")));
311                                }
312                            }
313                        }
314                    }
315                }
316            }
317        };
318
319        sink.close()
320            .await
321            .map_err(|err| TopicLogSyncChannelError::MessageSink(format!("{err:?}")))?;
322
323        let final_event = match result.as_ref() {
324            Ok(_) => {
325                debug!(
326                    sent_ops = ?metrics.sent_operations(),
327                    sent_bytes = ?metrics.sent_bytes(),
328                    received_ops = %metrics.received_operations(),
329                    received_bytes = %metrics.received_bytes(),
330                    "sync session closed"
331                );
332                TopicLogSyncEvent::SessionFinished { metrics }
333            }
334            Err(err) => {
335                warn!(error = ?err, "sync session closed with error");
336                TopicLogSyncEvent::Failed {
337                    error: err.to_string(),
338                }
339            }
340        };
341
342        self.event_tx
343            .send(final_event)
344            .map_err(|_| TopicLogSyncChannelError::EventSend)?;
345
346        result
347    }
348}
349
350/// Map raw message sink and stream into log sync protocol specific channels.
351#[allow(clippy::complexity)]
352fn sync_channels<'a, L, E>(
353    sink: &mut (impl Sink<TopicLogSyncMessage<L, E>, Error = impl Debug> + Unpin),
354    stream: &mut (impl Stream<Item = Result<TopicLogSyncMessage<L, E>, impl Debug>> + Unpin),
355) -> (
356    impl Sink<LogSyncMessage<L>, Error = TopicLogSyncChannelError> + Unpin,
357    impl Stream<Item = Result<LogSyncMessage<L>, TopicLogSyncChannelError>> + Unpin,
358)
359where
360    L: LogId,
361    E: Extensions,
362{
363    let log_sync_sink = LogSyncSink::new(sink);
364
365    let log_sync_stream = stream.by_ref().map(|message| match message {
366        Ok(TopicLogSyncMessage::Sync(message)) => Ok(message),
367        Ok(TopicLogSyncMessage::Live { .. }) | Ok(TopicLogSyncMessage::Close) => Err(
368            TopicLogSyncChannelError::MessageStream("non-protocol message received".to_string()),
369        ),
370        Err(err) => Err(TopicLogSyncChannelError::MessageStream(format!("{err:?}"))),
371    });
372
373    (log_sync_sink, log_sync_stream)
374}
375
376/// Error type occurring in topic log sync channels.
377#[derive(Debug, Error)]
378pub enum TopicLogSyncChannelError {
379    #[error("error sending on message sink: {0}")]
380    MessageSink(String),
381
382    #[error("error receiving from message stream: {0}")]
383    MessageStream(String),
384
385    #[error("no active receivers for broadcast")]
386    EventSend,
387}
388
389/// Error type occurring in topic log sync protocol.
390#[derive(Debug, Error)]
391pub enum TopicLogSyncError {
392    #[error(transparent)]
393    Sync(#[from] LogSyncError),
394
395    #[error("topic store error: {0}")]
396    TopicStore(String),
397
398    #[error("unexpected protocol message: {0}")]
399    UnexpectedProtocolMessage(String),
400
401    #[error(transparent)]
402    Channel(#[from] TopicLogSyncChannelError),
403
404    #[error("remote unexpectedly closed stream in live-mode")]
405    UnexpectedStreamClosure,
406
407    #[error("{0}")]
408    DecodeMessage(String),
409}
410
411#[derive(Clone, Debug, Default, PartialEq)]
412pub struct Metrics {
413    pub outbound_sync_bytes: u64,
414    pub outbound_sync_operations: u64,
415    pub inbound_sync_bytes: u64,
416    pub inbound_sync_operations: u64,
417    pub sent_sync_bytes: u64,
418    pub sent_sync_operations: u64,
419    pub received_sync_bytes: u64,
420    pub received_sync_operations: u64,
421    pub sent_live_bytes: u64,
422    pub sent_live_operations: u64,
423    pub received_live_bytes: u64,
424    pub received_live_operations: u64,
425}
426
427impl Metrics {
428    pub fn sent_bytes(&self) -> u64 {
429        self.sent_sync_bytes + self.sent_live_bytes
430    }
431
432    pub fn received_bytes(&self) -> u64 {
433        self.received_sync_bytes + self.received_live_bytes
434    }
435
436    pub fn sent_operations(&self) -> u64 {
437        self.sent_sync_operations + self.sent_live_operations
438    }
439
440    pub fn received_operations(&self) -> u64 {
441        self.received_sync_operations + self.received_live_operations
442    }
443}
444
445impl From<LogSyncMetrics> for Metrics {
446    fn from(value: LogSyncMetrics) -> Metrics {
447        Metrics {
448            inbound_sync_bytes: value.inbound_bytes,
449            outbound_sync_bytes: value.outbound_bytes,
450            sent_sync_bytes: value.sent_bytes,
451            received_sync_bytes: value.received_bytes,
452            inbound_sync_operations: value.inbound_operations,
453            outbound_sync_operations: value.outbound_operations,
454            sent_sync_operations: value.sent_operations,
455            received_sync_operations: value.received_operations,
456            ..Default::default()
457        }
458    }
459}
460
461/// Events emitted from topic log sync sessions.
462#[derive(Debug, Clone, PartialEq)]
463pub enum TopicLogSyncEvent<E = ()> {
464    /// A session has been initiated locally.
465    ///
466    /// This event is always sent and will be followed by `SyncStarted` or `Failed` events.
467    SessionStarted,
468
469    /// We have exchanged initial session metrics with the remote and the sync phase of this
470    /// session has started.
471    ///
472    /// This event will be followed by any number of `OperationReceived` events, or a `SyncFinished` or `Failed`.
473    SyncStarted { metrics: Metrics },
474
475    /// All past state has been replicated and we will now enter live mode `LiveModeStarted` (if configured) or the
476    /// session will end `SessionFinished`.
477    ///
478    /// This event will be followed by a `LiveModeStarted` event or a `SyncFinished` or `Failed` event.
479    SyncFinished { metrics: Metrics },
480
481    /// The session has entered live mode, we will send and receive operations in realtime.
482    ///
483    /// This event will be followed by any number of `OperationReceived` events or a `SyncFinished` or `Failed` event.
484    LiveModeStarted,
485
486    /// An operation has been received, this can be in the "sync" or "live mode" phase of a session.
487    OperationReceived {
488        operation: Box<Operation<E>>,
489        metrics: Metrics,
490    },
491
492    /// The session has finished.
493    ///
494    /// When no error occurs this event will always be sent at the end of a session.
495    SessionFinished { metrics: Metrics },
496
497    /// The session failed.
498    ///
499    /// This event will always be the final event sent when an error occurred in any phase of the
500    /// session.
501    Failed { error: String },
502}
503
504impl<E> From<LogSyncEvent<E>> for TopicLogSyncEvent<E> {
505    fn from(event: LogSyncEvent<E>) -> Self {
506        match event {
507            LogSyncEvent::MetricsExchanged { metrics } => TopicLogSyncEvent::SyncStarted {
508                metrics: metrics.into(),
509            },
510            LogSyncEvent::OperationReceived { operation, metrics } => {
511                TopicLogSyncEvent::OperationReceived {
512                    operation,
513                    metrics: metrics.into(),
514                }
515            }
516        }
517    }
518}
519
520/// Protocol message types.
521#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
522#[serde(bound(deserialize = "L: LogId"))]
523#[serde(tag = "type", content = "value")]
524#[allow(clippy::large_enum_variant)]
525pub enum TopicLogSyncMessage<L, E>
526where
527    L: LogId,
528    E: Extensions,
529{
530    Sync(LogSyncMessage<L>),
531    Live(Header<E>, Option<Body>),
532    Close,
533}
534
535impl<L, E> std::fmt::Display for TopicLogSyncMessage<L, E>
536where
537    L: LogId,
538    E: Extensions,
539{
540    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
541        let value = match self {
542            TopicLogSyncMessage::Sync(_) => "sync",
543            TopicLogSyncMessage::Live(_, _) => "live",
544            TopicLogSyncMessage::Close => "close",
545        };
546        write!(f, "{value}")
547    }
548}
549
550pin_project! {
551    /// Sink wrapper which converts messages and errors into the expected types.
552    pub struct LogSyncSink<S, L, E> {
553        #[pin]
554        inner: S,
555        _phantom: std::marker::PhantomData<(L, E)>,
556    }
557}
558
559impl<S, L, E> LogSyncSink<S, L, E> {
560    pub fn new(inner: S) -> Self {
561        Self {
562            inner,
563            _phantom: std::marker::PhantomData,
564        }
565    }
566}
567
568impl<S, L, E> Sink<LogSyncMessage<L>> for LogSyncSink<S, L, E>
569where
570    L: LogId,
571    S: Sink<TopicLogSyncMessage<L, E>>,
572    S::Error: Debug,
573    E: Extensions,
574{
575    type Error = TopicLogSyncChannelError;
576
577    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
578        let this = self.project();
579        this.inner
580            .poll_ready(cx)
581            .map_err(|err| TopicLogSyncChannelError::MessageSink(format!("{err:?}")))
582    }
583
584    fn start_send(self: Pin<&mut Self>, item: LogSyncMessage<L>) -> Result<(), Self::Error> {
585        let this = self.project();
586        let msg = TopicLogSyncMessage::Sync(item);
587        this.inner
588            .start_send(msg)
589            .map_err(|err| TopicLogSyncChannelError::MessageSink(format!("{err:?}")))
590    }
591
592    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
593        let this = self.project();
594        this.inner
595            .poll_flush(cx)
596            .map_err(|err| TopicLogSyncChannelError::MessageSink(format!("{err:?}")))
597    }
598
599    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
600        let this = self.project();
601        this.inner
602            .poll_close(cx)
603            .map_err(|err| TopicLogSyncChannelError::MessageSink(format!("{err:?}")))
604    }
605}
606#[cfg(test)]
607pub mod tests {
608    use std::collections::BTreeMap;
609
610    use assert_matches::assert_matches;
611    use futures::channel::mpsc;
612    use futures::{SinkExt, StreamExt};
613    use p2panda_core::test_utils::setup_logging;
614    use p2panda_core::{Body, Operation, Topic};
615
616    use crate::ToSync;
617    use crate::protocols::{LogSyncError, LogSyncMessage};
618    use crate::test_utils::{Peer, TestTopicSyncMessage, run_protocol, run_protocol_uni};
619    use crate::traits::Protocol;
620
621    use super::{TopicLogSyncError, TopicLogSyncEvent};
622
623    #[tokio::test]
624    async fn sync_session_no_operations() {
625        let topic = Topic::random();
626        let mut peer = Peer::new(0).await;
627        peer.associate(&topic, &BTreeMap::default()).await;
628
629        let (session, mut events_rx, _) = peer.topic_sync_protocol(topic.clone(), false);
630
631        let (_, remote_rx) = run_protocol_uni(
632            session,
633            &[
634                TestTopicSyncMessage::Sync(LogSyncMessage::Have(BTreeMap::default())),
635                TestTopicSyncMessage::Sync(LogSyncMessage::Done),
636            ],
637        )
638        .await
639        .unwrap();
640
641        assert_matches!(
642            events_rx.recv().await.unwrap(),
643            TopicLogSyncEvent::SyncStarted { .. }
644        );
645        assert_matches!(
646            events_rx.recv().await.unwrap(),
647            TopicLogSyncEvent::SyncFinished { .. }
648        );
649        assert_matches!(
650            events_rx.recv().await.unwrap(),
651            TopicLogSyncEvent::SessionFinished { .. }
652        );
653
654        let messages = remote_rx.collect::<Vec<_>>().await;
655        assert_eq!(messages.len(), 2);
656        for (index, message) in messages.into_iter().enumerate() {
657            match index {
658                0 => assert_eq!(
659                    message,
660                    TestTopicSyncMessage::Sync(LogSyncMessage::Have(BTreeMap::default()))
661                ),
662                1 => {
663                    assert_eq!(message, TestTopicSyncMessage::Sync(LogSyncMessage::Done));
664                    break;
665                }
666                _ => panic!(),
667            };
668        }
669    }
670
671    #[tokio::test]
672    async fn sync_operations_accept() {
673        setup_logging();
674
675        let log_id = 0;
676        let topic = Topic::random();
677        let mut peer = Peer::new(0).await;
678
679        let body = Body::new("Hello, Sloth!".as_bytes());
680        let (header_0, header_bytes_0) = peer.create_operation(&body, log_id).await;
681        let (header_1, header_bytes_1) = peer.create_operation(&body, log_id).await;
682        let (header_2, header_bytes_2) = peer.create_operation(&body, log_id).await;
683
684        let logs = BTreeMap::from([(peer.id(), vec![log_id])]);
685        peer.associate(&topic, &logs).await;
686
687        let (session, mut events_rx, _) = peer.topic_sync_protocol(topic.clone(), false);
688
689        let (_, remote_rx) = run_protocol_uni(
690            session,
691            &[
692                TestTopicSyncMessage::Sync(LogSyncMessage::Have(BTreeMap::default())),
693                TestTopicSyncMessage::Sync(LogSyncMessage::Done),
694            ],
695        )
696        .await
697        .unwrap();
698
699        assert_matches!(
700            events_rx.recv().await.unwrap(),
701            TopicLogSyncEvent::SyncStarted { .. }
702        );
703        assert_matches!(
704            events_rx.recv().await.unwrap(),
705            TopicLogSyncEvent::SyncFinished { .. }
706        );
707        assert_matches!(
708            events_rx.recv().await.unwrap(),
709            TopicLogSyncEvent::SessionFinished { .. }
710        );
711
712        let messages = remote_rx.collect::<Vec<_>>().await;
713        assert_eq!(messages.len(), 6);
714        for (index, message) in messages.into_iter().enumerate() {
715            match index {
716                0 => assert_eq!(
717                    message,
718                    TestTopicSyncMessage::Sync(LogSyncMessage::Have(BTreeMap::from([(
719                        peer.id(),
720                        BTreeMap::from([(0, 2)])
721                    )])))
722                ),
723                1 => {
724                    let expected_bytes = header_0.payload_size
725                        + header_bytes_0.len() as u64
726                        + header_1.payload_size
727                        + header_bytes_1.len() as u64
728                        + header_2.payload_size
729                        + header_bytes_2.len() as u64;
730
731                    assert_eq!(
732                        message,
733                        TestTopicSyncMessage::Sync(LogSyncMessage::PreSync {
734                            total_operations: 3,
735                            total_bytes: expected_bytes
736                        })
737                    )
738                }
739                2 => {
740                    let (header, body_inner) = assert_matches!(message, TestTopicSyncMessage::Sync(LogSyncMessage::Operation(
741                        header,
742                        Some(body),
743                    )) => (header, body));
744                    assert_eq!(header, header_bytes_0);
745                    assert_eq!(Body::new(&body_inner), body)
746                }
747                3 => {
748                    let (header, body_inner) = assert_matches!(message, TestTopicSyncMessage::Sync(LogSyncMessage::Operation(
749                        header,
750                        Some(body),
751                    )) => (header, body));
752                    assert_eq!(header, header_bytes_1);
753                    assert_eq!(Body::new(&body_inner), body)
754                }
755                4 => {
756                    let (header, body_inner) = assert_matches!(message, TestTopicSyncMessage::Sync(LogSyncMessage::Operation(
757                        header,
758                        Some(body),
759                    )) => (header, body));
760                    assert_eq!(header, header_bytes_2);
761                    assert_eq!(Body::new(&body_inner), body)
762                }
763                5 => {
764                    assert_eq!(message, TestTopicSyncMessage::Sync(LogSyncMessage::Done));
765                    break;
766                }
767                _ => panic!(),
768            };
769        }
770    }
771
772    #[tokio::test]
773    async fn topic_log_sync_full_duplex() {
774        setup_logging();
775        let topic = Topic::random();
776        let log_id = 0;
777
778        let mut peer_a = Peer::new(0).await;
779        let mut peer_b = Peer::new(1).await;
780
781        let body = Body::new("Hello, Sloth!".as_bytes());
782        let (header_0, _) = peer_a.create_operation(&body, 0).await;
783        let (header_1, _) = peer_a.create_operation(&body, 0).await;
784        let (header_2, _) = peer_a.create_operation(&body, 0).await;
785
786        let logs = BTreeMap::from([(peer_a.id(), vec![log_id])]);
787        peer_a.associate(&topic, &logs).await;
788
789        let (peer_a_session, mut peer_a_events_rx, _) =
790            peer_a.topic_sync_protocol(topic.clone(), false);
791
792        let (peer_b_session, mut peer_b_events_rx, _) =
793            peer_b.topic_sync_protocol(topic.clone(), false);
794
795        run_protocol(peer_a_session, peer_b_session).await.unwrap();
796
797        // Assert peer a events.
798        assert_matches!(
799            peer_a_events_rx.recv().await.unwrap(),
800            TopicLogSyncEvent::SyncStarted { .. }
801        );
802        assert_matches!(
803            peer_a_events_rx.recv().await.unwrap(),
804            TopicLogSyncEvent::SyncFinished { .. }
805        );
806        assert_matches!(
807            peer_a_events_rx.recv().await.unwrap(),
808            TopicLogSyncEvent::SessionFinished { .. }
809        );
810
811        // Assert peer b events.
812        assert_matches!(
813            peer_b_events_rx.recv().await.unwrap(),
814            TopicLogSyncEvent::SyncStarted { .. }
815        );
816        let (header, body_inner) = assert_matches!(
817            peer_b_events_rx.recv().await.unwrap(),
818            TopicLogSyncEvent::OperationReceived { operation, .. } => {
819                let Operation { header, body, .. } = *operation;
820                (header, body)
821            }
822        );
823        assert_eq!(header, header_0);
824        assert_eq!(body_inner.unwrap(), body);
825        let (header, body_inner) = assert_matches!(
826            peer_b_events_rx.recv().await.unwrap(),
827            TopicLogSyncEvent::OperationReceived { operation, .. } => {
828                let Operation { header, body, .. } = *operation;
829                (header, body)
830            }
831        );
832        assert_eq!(header, header_1);
833        assert_eq!(body_inner.unwrap(), body);
834        let (header, body_inner) = assert_matches!(
835            peer_b_events_rx.recv().await.unwrap(),
836            TopicLogSyncEvent::OperationReceived { operation, .. } => {
837                let Operation { header, body, .. } = *operation;
838                (header, body)
839            }
840        );
841        assert_eq!(header, header_2);
842        assert_eq!(body_inner.unwrap(), body);
843        assert_matches!(
844            peer_b_events_rx.recv().await.unwrap(),
845            TopicLogSyncEvent::SyncFinished { .. }
846        );
847        assert_matches!(
848            peer_b_events_rx.recv().await.unwrap(),
849            TopicLogSyncEvent::SessionFinished { .. }
850        );
851    }
852
853    #[tokio::test]
854    async fn live_mode() {
855        let log_id = 0;
856        let topic = Topic::random();
857        let mut peer_a = Peer::new(0).await;
858        let mut peer_b = Peer::new(1).await;
859
860        let body = Body::new("Hello, Sloth!".as_bytes());
861        let (header_0, header_bytes_0) = peer_b.create_operation(&body, log_id).await;
862
863        let logs = BTreeMap::from([(peer_a.id(), vec![log_id])]);
864        peer_a.associate(&topic, &logs).await;
865
866        let logs = BTreeMap::default();
867        peer_a.associate(&topic, &logs).await;
868
869        let (header_1, _) = peer_b.create_operation_no_insert(&body, log_id).await;
870        let expected_bytes_received = header_0.payload_size
871            + header_0.to_bytes().len() as u64
872            + header_1.payload_size
873            + header_1.to_bytes().len() as u64;
874        let (header_2, _) = peer_a.create_operation_no_insert(&body, log_id).await;
875        let expected_bytes_sent = header_2.payload_size + header_2.to_bytes().len() as u64;
876
877        let (protocol, mut events_rx, mut live_mode_tx) =
878            peer_a.topic_sync_protocol(topic.clone(), true);
879
880        live_mode_tx
881            .send(ToSync::Payload(Operation {
882                hash: header_2.hash(),
883                header: header_2.clone(),
884                body: Some(body.clone()),
885            }))
886            .await
887            .unwrap();
888        live_mode_tx.send(ToSync::Close).await.unwrap();
889
890        let total_bytes = header_bytes_0.len() + body.to_bytes().len();
891        let (_, remote_rx) = run_protocol_uni(
892            protocol,
893            &[
894                TestTopicSyncMessage::Sync(LogSyncMessage::Have(BTreeMap::default())),
895                TestTopicSyncMessage::Sync(LogSyncMessage::PreSync {
896                    total_operations: 1,
897                    total_bytes: total_bytes as u64,
898                }),
899                TestTopicSyncMessage::Sync(LogSyncMessage::Operation(
900                    header_bytes_0,
901                    Some(body.to_bytes()),
902                )),
903                TestTopicSyncMessage::Sync(LogSyncMessage::Done),
904                TestTopicSyncMessage::Live(header_1.clone(), Some(body.clone())),
905                TestTopicSyncMessage::Close,
906            ],
907        )
908        .await
909        .unwrap();
910
911        assert_matches!(
912            events_rx.recv().await.unwrap(),
913            TopicLogSyncEvent::SyncStarted { .. }
914        );
915        assert_matches!(
916            events_rx.recv().await.unwrap(),
917            TopicLogSyncEvent::OperationReceived { .. }
918        );
919        assert_matches!(
920            events_rx.recv().await.unwrap(),
921            TopicLogSyncEvent::SyncFinished { .. }
922        );
923        assert_matches!(
924            events_rx.recv().await.unwrap(),
925            TopicLogSyncEvent::LiveModeStarted
926        );
927        let metrics = assert_matches!(
928            events_rx.recv().await.unwrap(),
929            TopicLogSyncEvent::OperationReceived { metrics, .. } => metrics
930        );
931        assert_eq!(metrics.received_operations(), 2);
932        assert_eq!(metrics.sent_operations(), 1);
933        assert_eq!(metrics.received_bytes(), expected_bytes_received);
934        assert_eq!(metrics.sent_bytes(), expected_bytes_sent);
935        assert_matches!(
936            events_rx.recv().await.unwrap(),
937            TopicLogSyncEvent::SessionFinished { .. }
938        );
939
940        let messages = remote_rx.collect::<Vec<_>>().await;
941        assert_eq!(messages.len(), 4);
942        for (index, message) in messages.into_iter().enumerate() {
943            match index {
944                0 => assert_matches!(message, TestTopicSyncMessage::Sync(LogSyncMessage::Have(_))),
945                1 => {
946                    assert_matches!(message, TestTopicSyncMessage::Sync(LogSyncMessage::Done))
947                }
948                2 => {
949                    let (header, body_inner) = assert_matches!(message, TestTopicSyncMessage::Live(
950                        header,
951                        Some(body)
952                    ) => (header, body));
953                    assert_eq!(header, header_2);
954                    assert_eq!(body_inner, body);
955                }
956                3 => {
957                    assert_matches!(message, TestTopicSyncMessage::Close)
958                }
959                _ => panic!(),
960            };
961        }
962    }
963
964    #[tokio::test]
965    async fn dedup_live_mode_messages() {
966        let log_id = 0;
967        let topic = Topic::random();
968        let mut peer_a = Peer::new(0).await;
969        let mut peer_b = Peer::new(1).await;
970
971        let body = Body::new("Hello, Sloth!".as_bytes());
972        let (header_0, header_bytes_0) = peer_b.create_operation(&body, log_id).await;
973
974        let logs = BTreeMap::from([(peer_a.id(), vec![log_id])]);
975        peer_a.associate(&topic, &logs).await;
976
977        let logs = BTreeMap::default();
978        peer_a.associate(&topic, &logs).await;
979
980        let (header_1, _) = peer_b.create_operation_no_insert(&body, log_id).await;
981        let expected_bytes_received = header_0.payload_size
982            + header_0.to_bytes().len() as u64
983            + header_1.payload_size
984            + header_1.to_bytes().len() as u64;
985        let (header_2, _) = peer_a.create_operation_no_insert(&body, log_id).await;
986        let expected_bytes_sent = header_2.payload_size + header_2.to_bytes().len() as u64;
987
988        let (protocol, mut events_rx, mut live_mode_tx) =
989            peer_a.topic_sync_protocol(topic.clone(), true);
990
991        live_mode_tx
992            .send(ToSync::Payload(Operation {
993                hash: header_2.hash(),
994                header: header_2.clone(),
995                body: Some(body.clone()),
996            }))
997            .await
998            .unwrap();
999
1000        // Sending subscription message twice.
1001        live_mode_tx
1002            .send(ToSync::Payload(Operation {
1003                hash: header_2.hash(),
1004                header: header_2.clone(),
1005                body: Some(body.clone()),
1006            }))
1007            .await
1008            .unwrap();
1009
1010        live_mode_tx.send(ToSync::Close).await.unwrap();
1011
1012        let total_bytes = header_bytes_0.len() + body.to_bytes().len();
1013        let (_, remote_rx) = run_protocol_uni(
1014            protocol,
1015            &[
1016                TestTopicSyncMessage::Sync(LogSyncMessage::Have(BTreeMap::default())),
1017                TestTopicSyncMessage::Sync(LogSyncMessage::PreSync {
1018                    total_operations: 1,
1019                    total_bytes: total_bytes as u64,
1020                }),
1021                TestTopicSyncMessage::Sync(LogSyncMessage::Operation(
1022                    header_bytes_0,
1023                    Some(body.to_bytes()),
1024                )),
1025                TestTopicSyncMessage::Sync(LogSyncMessage::Done),
1026                TestTopicSyncMessage::Live(header_1.clone(), Some(body.clone())),
1027                // Duplicate of message sent during sync.
1028                TestTopicSyncMessage::Live(header_0.clone(), Some(body.clone())),
1029                // Duplicate of message sent earlier in live-mode.
1030                TestTopicSyncMessage::Live(header_1.clone(), Some(body.clone())),
1031                TestTopicSyncMessage::Close,
1032            ],
1033        )
1034        .await
1035        .unwrap();
1036
1037        assert_matches!(
1038            events_rx.recv().await.unwrap(),
1039            TopicLogSyncEvent::SyncStarted { .. }
1040        );
1041        assert_matches!(
1042            events_rx.recv().await.unwrap(),
1043            TopicLogSyncEvent::OperationReceived { .. }
1044        );
1045        assert_matches!(
1046            events_rx.recv().await.unwrap(),
1047            TopicLogSyncEvent::SyncFinished { .. }
1048        );
1049        assert_matches!(
1050            events_rx.recv().await.unwrap(),
1051            TopicLogSyncEvent::LiveModeStarted
1052        );
1053        let metrics = assert_matches!(
1054            events_rx.recv().await.unwrap(),
1055            TopicLogSyncEvent::OperationReceived { metrics, .. } => metrics
1056        );
1057        assert_eq!(metrics.received_operations(), 2);
1058        assert_eq!(metrics.sent_operations(), 1);
1059        assert_eq!(metrics.received_bytes(), expected_bytes_received);
1060        assert_eq!(metrics.sent_bytes(), expected_bytes_sent);
1061        assert_matches!(
1062            events_rx.recv().await.unwrap(),
1063            TopicLogSyncEvent::SessionFinished { .. }
1064        );
1065
1066        let messages = remote_rx.collect::<Vec<_>>().await;
1067        assert_eq!(messages.len(), 4);
1068        for (index, message) in messages.into_iter().enumerate() {
1069            match index {
1070                0 => assert_matches!(message, TestTopicSyncMessage::Sync(LogSyncMessage::Have(_))),
1071                1 => {
1072                    assert_matches!(message, TestTopicSyncMessage::Sync(LogSyncMessage::Done))
1073                }
1074                2 => {
1075                    let (header, body_inner) = assert_matches!(message, TestTopicSyncMessage::Live(header, Some(body)) => (header, body));
1076                    assert_eq!(header, header_2);
1077                    assert_eq!(body_inner, body);
1078                }
1079                3 => {
1080                    assert_matches!(message, TestTopicSyncMessage::Close)
1081                }
1082                _ => panic!(),
1083            };
1084        }
1085    }
1086
1087    #[tokio::test]
1088    async fn unexpected_stream_closure_sync() {
1089        let topic = Topic::random();
1090        let mut peer = Peer::new(0).await;
1091        peer.associate(&topic, &Default::default()).await;
1092
1093        let (session, mut events_rx, _live_tx) = peer.topic_sync_protocol(topic.clone(), true);
1094
1095        let messages = [TestTopicSyncMessage::Sync(LogSyncMessage::Have(
1096            BTreeMap::default(),
1097        ))];
1098
1099        let (mut local_message_tx, _remote_message_rx) = mpsc::channel(128);
1100        let (mut remote_message_tx, local_message_rx) = mpsc::channel(128);
1101        let mut local_message_rx = local_message_rx.map(|message| Ok::<_, ()>(message));
1102
1103        for message in messages {
1104            remote_message_tx.send(message.to_owned()).await.unwrap();
1105        }
1106
1107        let handle = tokio::spawn(async move {
1108            session
1109                .run(&mut local_message_tx, &mut local_message_rx)
1110                .await
1111                .expect_err("expected unexpected stream closure")
1112        });
1113
1114        drop(remote_message_tx);
1115
1116        let err = handle.await.unwrap();
1117        assert_matches!(
1118            err,
1119            TopicLogSyncError::Sync(LogSyncError::UnexpectedStreamClosure)
1120        );
1121
1122        while let Ok(event) = events_rx.recv().await {
1123            if let TopicLogSyncEvent::Failed { error } = event {
1124                assert_eq!(
1125                    error,
1126                    "remote unexpectedly closed stream during initial sync".to_string()
1127                );
1128                break;
1129            }
1130        }
1131    }
1132
1133    #[tokio::test]
1134    async fn unexpected_stream_closure_live_mode() {
1135        let topic = Topic::random();
1136        let mut peer = Peer::new(0).await;
1137        peer.associate(&topic, &Default::default()).await;
1138
1139        let (session, mut events_rx, _live_tx) = peer.topic_sync_protocol(topic.clone(), true);
1140
1141        let messages = [
1142            TestTopicSyncMessage::Sync(LogSyncMessage::Have(BTreeMap::default())),
1143            TestTopicSyncMessage::Sync(LogSyncMessage::Done),
1144        ];
1145
1146        let (mut local_message_tx, _remote_message_rx) = mpsc::channel(128);
1147        let (mut remote_message_tx, local_message_rx) = mpsc::channel(128);
1148        let mut local_message_rx = local_message_rx.map(|message| Ok::<_, ()>(message));
1149
1150        for message in messages {
1151            remote_message_tx.send(message.to_owned()).await.unwrap();
1152        }
1153
1154        let handle = tokio::spawn(async move {
1155            session
1156                .run(&mut local_message_tx, &mut local_message_rx)
1157                .await
1158                .expect_err("expected unexpected stream closure")
1159        });
1160
1161        drop(remote_message_tx);
1162
1163        let err = handle.await.unwrap();
1164        assert_matches!(err, TopicLogSyncError::UnexpectedStreamClosure);
1165
1166        while let Ok(event) = events_rx.recv().await {
1167            if let TopicLogSyncEvent::Failed { error } = event {
1168                assert_eq!(
1169                    error,
1170                    "remote unexpectedly closed stream in live-mode".to_string()
1171                );
1172                break;
1173            }
1174        }
1175    }
1176}
p2panda_sync/protocols/topic_log_sync.rs

p2panda_sync/protocols/
topic_log_sync.rs
