fix DontAnnounce policy drops documents synced by clients

Problem: when we receive a sync message for a document which we don't
have in storage and for whom the AnnouncePolicy returns DontAnnounce
then we erroneously decide that the document is unavailable even if the
incoming sync message contains data about the document. This is because
we fail to process pending sync messages once the load has completed if
we don't have the document available or any connected peers who we could
request from.

Solution: process pending sync messages before deciding that the
document is unavailable.

Whilst I'm here I also cleaned up the logic around the phase transition
during load to make it more consistent with the rest of the phase
transitions and easier to read. One important improvement is that if the
document state changes mutliple times in one turn of a document actor,
we only notify of the last status. This is important because it means
that if a document transitions from loading through not found and into
requesting in the same turn (which can happen if a load completes after
receiving a sync message) then we don't notify the hub of the not found
state. This is in turn important because notifying the hub of a not
found state causes any outstanding find commands to complete with
`None`.

Fixes: #85

Co-authored-by: Carlos Scheidegger <285675+cscheid@users.noreply.github.com>
Co-authored-by: shikokuchuo <53399081+shikokuchuo@users.noreply.github.com>

Author: 3 people

CHANGELOG.md

@@ -2,14 +2,19 @@
 
 ## Unreleased
 
+### Fixed
+
+* A bug where locally unavailable documents sent by peers with an announce
+  policy set to false would be marked as unavailable
+
 ### Added
 
-- `TcpDialer::new` which takes a `Url` parameter, rather than a host and a port
+* `TcpDialer::new` which takes a `Url` parameter, rather than a host and a port
   or a socket address.
-- `Repo::dial_tcp()` to simplify construction of `TcpDialer`. 
-- Allow documents syncing over the TCP transport to be up to 8gb size instead
+* `Repo::dial_tcp()` to simplify construction of `TcpDialer`. 
+* Allow documents syncing over the TCP transport to be up to 8gb size instead
   of Tokio's default 8mb frame size
-- Exposed receiving `ConnectionHandle`s via `accept()`. Users can now subscribe
+* Exposed receiving `ConnectionHandle`s via `accept()`. Users can now subscribe
   to an `events()` stream directly on the handle, or `await` for
   `handshake_completed()`.
 

samod-core/src/actors/document.rs

@@ -40,8 +40,7 @@ mod load;
 mod on_disk_state;
 pub use on_disk_state::CompactionHash;
 mod peer_doc_connection;
-mod ready;
-mod request;
+mod phase;
 mod spawn_args;
 mod with_doc_result;
 pub use with_doc_result::WithDocResult;

samod-core/src/actors/document/doc_actor_result.rs

@@ -3,11 +3,11 @@ use std::collections::HashMap;
 use automerge::ChangeHash;
 
 use crate::{
-    ConnectionId, DocumentChanged, PeerId, StorageKey,
+    ConnectionId, DocumentChanged, DocumentId, PeerId, StorageKey,
     actors::{
         DocToHubMsg,
-        document::{SyncMessageStat, io::DocumentIoTask},
-        messages::DocToHubMsgPayload,
+        document::{DocumentStatus, SyncMessageStat, io::DocumentIoTask},
+        messages::{Broadcast, DocToHubMsgPayload, SyncMessage},
     },
     io::{IoTask, IoTaskId, StorageTask},
     network::PeerDocState,
@@ -58,8 +58,60 @@ impl DocActorResult {
     }
 
     /// Send a message back to the hub
-    pub(crate) fn send_message(&mut self, message: DocToHubMsgPayload) {
-        self.outgoing_messages.push(DocToHubMsg(message));
+    pub(crate) fn send_sync_message(
+        &mut self,
+        conn_id: ConnectionId,
+        doc_id: DocumentId,
+        message: SyncMessage,
+    ) {
+        self.outgoing_messages
+            .push(DocToHubMsg(DocToHubMsgPayload::SendSyncMessage {
+                connection_id: conn_id,
+                document_id: doc_id,
+                message,
+            }));
+    }
+
+    pub(crate) fn send_broadcast(&mut self, connections: Vec<ConnectionId>, msg: Broadcast) {
+        self.outgoing_messages
+            .push(DocToHubMsg(DocToHubMsgPayload::Broadcast {
+                connections,
+                msg,
+            }));
+    }
+
+    pub(crate) fn send_terminated(&mut self) {
+        self.outgoing_messages
+            .push(DocToHubMsg(DocToHubMsgPayload::Terminated));
+    }
+
+    pub(crate) fn send_peer_states_changes(
+        &mut self,
+        new_states: HashMap<ConnectionId, PeerDocState>,
+    ) {
+        // Remove previous peer state change messages as they are redundant
+        self.outgoing_messages
+            .retain(|m| !matches!(m.0, DocToHubMsgPayload::PeerStatesChanged { .. }));
+        self.outgoing_messages
+            .push(DocToHubMsg(DocToHubMsgPayload::PeerStatesChanged {
+                new_states,
+            }));
+    }
+
+    pub(crate) fn send_doc_status_update(&mut self, new_status: DocumentStatus) {
+        // remove any existing doc status update so that if the document status changes
+        // multiple times during a turn, only the latest status is sent to the hub.
+        // This is especially important to avoid bouncing through a NotFound state
+        // when loading a document as that will cause any outstanding find commands
+        // to fail even if the document loads successfully in this turn (as it might
+        // if we finish loading after receiving a sync message with the document
+        // content).
+        self.outgoing_messages
+            .retain(|m| !matches!(m.0, DocToHubMsgPayload::DocumentStatusChanged { .. }));
+        self.outgoing_messages
+            .push(DocToHubMsg(DocToHubMsgPayload::DocumentStatusChanged {
+                new_status,
+            }));
     }
 
     pub(crate) fn put(&mut self, key: StorageKey, value: Vec<u8>) -> IoTaskId {

samod-core/src/actors/document/doc_state.rs

@@ -6,16 +6,21 @@ use automerge::Automerge;
 use crate::{
     ConnectionId, DocumentId, StorageKey, UnixTimestamp,
     actors::{
-        document::{DocActorResult, SyncDirection, SyncMessageStat},
-        messages::{Broadcast, DocMessage, DocToHubMsgPayload, SyncMessage},
+        document::{
+            DocActorResult, SyncDirection, SyncMessageStat,
+            phase::{
+                loading::Loading,
+                ready::Ready,
+                request::{Request, RequestState},
+            },
+        },
+        messages::{Broadcast, DocMessage, SyncMessage},
     },
 };
 
 use super::{
     DocumentStatus,
     peer_doc_connection::{AnnouncePolicy, PeerDocConnection},
-    ready::Ready,
-    request::{Request, RequestState},
 };
 
 #[derive(Debug)]
@@ -32,9 +37,7 @@ pub(super) struct DocState {
 
 #[derive(Debug)]
 pub enum Phase {
-    Loading {
-        pending_sync_messages: HashMap<ConnectionId, Vec<SyncMessage>>,
-    },
+    Loading(Loading),
     Requesting(Request),
     Ready(Ready),
     NotFound,
@@ -56,9 +59,7 @@ impl DocState {
         any_dialer_connecting: bool,
     ) -> Self {
         Self {
-            phase: Phase::Loading {
-                pending_sync_messages: HashMap::new(),
-            },
+            phase: Phase::Loading(Loading::new()),
             document_id,
             doc,
             any_dialer_connecting,
@@ -79,43 +80,33 @@ impl DocState {
             PhaseTransition::None => {}
             PhaseTransition::ToReady => {
                 tracing::trace!("transitioning to ready");
-                out.send_message(DocToHubMsgPayload::DocumentStatusChanged {
-                    new_status: DocumentStatus::Ready,
-                });
+                out.send_doc_status_update(DocumentStatus::Ready);
                 out.emit_doc_changed(self.doc.get_heads());
                 self.phase = Phase::Ready(Ready::new());
             }
             PhaseTransition::ToNotFound => {
                 tracing::trace!("transitioning to NotFound");
-                out.send_message(DocToHubMsgPayload::DocumentStatusChanged {
-                    new_status: DocumentStatus::NotFound,
-                });
+                out.send_doc_status_update(DocumentStatus::NotFound);
                 if let Phase::Requesting(request) = &self.phase {
                     for peer in request.peers_waiting_for_us_to_respond() {
-                        out.send_message(DocToHubMsgPayload::SendSyncMessage {
-                            connection_id: peer,
-                            document_id: self.document_id.clone(),
-                            message: SyncMessage::DocUnavailable,
-                        });
+                        out.send_sync_message(
+                            peer,
+                            self.document_id.clone(),
+                            SyncMessage::DocUnavailable,
+                        );
                     }
                 }
                 self.phase = Phase::NotFound;
             }
             PhaseTransition::ToRequesting(request) => {
                 tracing::trace!("transitioning to requesting");
-                out.send_message(DocToHubMsgPayload::DocumentStatusChanged {
-                    new_status: DocumentStatus::Requesting,
-                });
+                out.send_doc_status_update(DocumentStatus::Requesting);
                 self.phase = Phase::Requesting(request);
             }
             PhaseTransition::ToLoading => {
                 tracing::trace!("transitioning to loading");
-                out.send_message(DocToHubMsgPayload::DocumentStatusChanged {
-                    new_status: DocumentStatus::Loading,
-                });
-                self.phase = Phase::Loading {
-                    pending_sync_messages: HashMap::new(),
-                };
+                out.send_doc_status_update(DocumentStatus::Loading);
+                self.phase = Phase::Loading(Loading::new());
             }
         }
     }
@@ -159,8 +150,9 @@ impl DocState {
         }
         // self.save_state
         //     .add_on_disk(snapshots.into_keys().chain(incrementals.into_keys()));
-        if matches!(self.phase, Phase::Loading { .. }) {
-            if self.doc.get_heads().is_empty() {
+        if let Phase::Loading(loading) = &mut self.phase {
+            let pending_sync_messages = loading.take_pending_sync_messages();
+            let phase_transition = if self.doc.get_heads().is_empty() {
                 let eligible_conns = peer_connections
                     .values()
                     .any(|p| p.announce_policy() != AnnouncePolicy::DontAnnounce);
@@ -170,47 +162,23 @@ impl DocState {
                         self.any_dialer_connecting,
                         "no data found on disk, requesting document"
                     );
-                    let mut next_phase = Phase::Requesting(Request::new(
+                    PhaseTransition::ToRequesting(Request::new(
                         self.document_id.clone(),
                         peer_connections.values(),
-                    ));
-                    std::mem::swap(&mut self.phase, &mut next_phase);
-                    let Phase::Loading {
-                        pending_sync_messages,
-                    } = next_phase
-                    else {
-                        unreachable!("we already checked");
-                    };
-                    for (conn_id, msgs) in pending_sync_messages {
-                        for msg in msgs {
-                            self.handle_sync_message(now, out, conn_id, peer_connections, msg, now);
-                        }
-                    }
-                    out.send_message(DocToHubMsgPayload::DocumentStatusChanged {
-                        new_status: DocumentStatus::Requesting,
-                    });
+                    ))
                 } else {
                     tracing::debug!(
                         "no data found on disk and no connections available, transitioning to NotFound"
                     );
-                    self.handle_phase_transition(out, PhaseTransition::ToNotFound);
+                    PhaseTransition::ToNotFound
                 }
-                return;
-            }
+            } else {
+                tracing::trace!("load complete, transitioning to ready");
+                PhaseTransition::ToReady
+            };
 
-            tracing::trace!("load complete, transitioning to ready");
+            self.handle_phase_transition(out, phase_transition);
 
-            let mut next_phase = Phase::Ready(Ready::new());
-            std::mem::swap(&mut self.phase, &mut next_phase);
-            let Phase::Loading {
-                pending_sync_messages,
-            } = next_phase
-            else {
-                unreachable!("we already checked");
-            };
-            out.send_message(DocToHubMsgPayload::DocumentStatusChanged {
-                new_status: DocumentStatus::Ready,
-            });
             for (conn_id, msgs) in pending_sync_messages {
                 for msg in msgs {
                     self.handle_sync_message(now, out, conn_id, peer_connections, msg, now);
@@ -254,10 +222,7 @@ impl DocState {
                         }
                     })
                     .collect();
-                out.send_message(DocToHubMsgPayload::Broadcast {
-                    connections: targets,
-                    msg: Broadcast::Gossip { msg },
-                });
+                out.send_broadcast(targets, Broadcast::Gossip { msg });
             }
             DocMessage::Sync(msg) => self.handle_sync_message(
                 now,
@@ -298,13 +263,8 @@ impl DocState {
         };
 
         let (transition, duration) = match &mut self.phase {
-            Phase::Loading {
-                pending_sync_messages,
-            } => {
-                pending_sync_messages
-                    .entry(connection_id)
-                    .or_default()
-                    .push(msg);
+            Phase::Loading(loading) => {
+                loading.receive_sync_message(connection_id, msg);
                 (PhaseTransition::None, None)
             }
             Phase::Requesting(request) => {
@@ -380,11 +340,8 @@ impl DocState {
     ) -> HashMap<ConnectionId, Vec<SyncMessage>> {
         let mut result: HashMap<ConnectionId, Vec<SyncMessage>> = HashMap::new();
         for (conn_id, peer_conn) in peer_connections {
-            if let Phase::Loading {
-                pending_sync_messages,
-            } = &self.phase
-            {
-                out.pending_sync_messages = pending_sync_messages.values().map(|v| v.len()).sum();
+            if let Phase::Loading(loading) = &self.phase {
+                out.pending_sync_messages = loading.pending_msg_count();
                 continue;
             }
 

samod-core/src/actors/document/document_actor.rs

@@ -70,9 +70,7 @@ impl DocumentActor {
 
         let state = if let Some(doc) = initial_content {
             // Let the hub know this document is ready immediately if we already have content
-            out.send_message(DocToHubMsgPayload::DocumentStatusChanged {
-                new_status: DocumentStatus::Ready,
-            });
+            out.send_doc_status_update(DocumentStatus::Ready);
             DocState::new_ready(document_id.clone(), doc, any_dialer_pending)
         } else {
             DocState::new_loading(document_id.clone(), Automerge::new(), any_dialer_pending)
@@ -400,7 +398,7 @@ impl DocumentActor {
         if self.run_state == RunState::Stopping {
             if self.on_disk_state.is_flushed() {
                 self.run_state = RunState::Stopped;
-                out.send_message(DocToHubMsgPayload::Terminated);
+                out.send_terminated();
                 out.stopped = true;
             }
             return;
@@ -458,11 +456,7 @@ impl DocumentActor {
                 .generate_sync_messages(now, out, &mut self.peer_connections)
         {
             for msg in msgs {
-                out.send_message(DocToHubMsgPayload::SendSyncMessage {
-                    connection_id: conn_id,
-                    document_id: doc_id.clone(),
-                    message: msg,
-                });
+                out.send_sync_message(conn_id, doc_id.clone(), msg);
             }
         }
     }
@@ -475,7 +469,7 @@ impl DocumentActor {
             .collect::<HashMap<_, _>>();
         if !states.is_empty() {
             out.peer_state_changes = states.clone();
-            out.send_message(DocToHubMsgPayload::PeerStatesChanged { new_states: states })
+            out.send_peer_states_changes(states)
         }
     }
 

samod-core/src/actors/document/phase.rs

@@ -0,0 +1,3 @@
+pub(crate) mod loading;
+pub(crate) mod ready;
+pub(crate) mod request;