push changes experiment

10 files changed, 240 insertions, 20 deletions

diff --git a/src/document/sync.rs b/src/document/sync.rs
index e3ff157..e7a2665 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -28,6 +28,28 @@ impl Received {
         !self.0.changes.is_empty()
     }
 
+    /// Changes pushed along a writer's tree (`SyncBody::Changes`), as the Automerge sync
+    /// message that would have brought them: the sender's `heads`, nothing needed, and
+    /// as what it has, everything we had (`ours`, our heads before: it takes the
+    /// message as delivered, so it had all of them).
+    pub(crate) fn pushed(
+        heads: Vec<crate::document::ChangeHash>,
+        changes: Vec<Vec<u8>>,
+        ours: Vec<crate::document::ChangeHash>,
+    ) -> Self {
+        Self(sync::Message {
+            heads,
+            need: Vec::new(),
+            have: vec![sync::Have {
+                last_sync: ours,
+                bloom: sync::BloomFilter::default(),
+            }],
+            changes: changes.into(),
+            flags: None,
+            version: sync::MessageVersion::V2,
+        })
+    }
+
     /// The sender's heads, as it says.
     pub(crate) fn heads(&self) -> &[crate::document::ChangeHash] {
         &self.0.heads
@@ -287,6 +309,58 @@ impl SyncSession {
         }
     }
 
+    /// Instead of `generate`, where the message would be taken as delivered (the peer
+    /// has nothing we lack): the changes it would carry, as they are, with our heads
+    /// (`SyncBody::Changes`). Those since the peer's heads that don't count as sent
+    /// (as Automerge picks them; `known_to_peer`, `with_predecessors` alike), counted as
+    /// sent now, the message marked as handed over as `generate` does; the caller takes
+    /// it as delivered. None if there is nothing to push, the peer's heads aren't all
+    /// known, or the changes come to more than `limit` bytes (then `generate`).
+    pub(crate) fn push(
+        &mut self,
+        doc: &Document,
+        limit: usize,
+    ) -> Option<(Vec<crate::document::ChangeHash>, Vec<Vec<u8>>)> {
+        let theirs = self.state.their_heads.as_ref()?;
+        if theirs
+            .iter()
+            .any(|hash| doc.get_change_by_hash(hash).is_none())
+        {
+            return None;
+        }
+        let changes: Vec<automerge::Change> = doc
+            .get_changes(theirs)
+            .into_iter()
+            .filter(|change| !self.state.sent_hashes.contains(&change.hash()))
+            .collect();
+        let bytes: usize = changes.iter().map(|change| change.raw_bytes().len()).sum();
+        if changes.is_empty() || bytes > limit {
+            return None;
+        }
+        let sent: Vec<crate::document::ChangeHash> =
+            changes.iter().map(automerge::Change::hash).collect();
+        self.before_generate = (
+            self.state.last_sent_heads.clone(),
+            self.state.have_responded,
+        );
+        self.state.sent_hashes.extend(sent.iter().copied());
+        let heads = doc.get_heads();
+        self.state.last_sent_heads = heads.clone();
+        self.state.have_responded = true;
+        self.sending = Some(Sending {
+            sent,
+            regenerate: false,
+            assumed: None,
+        });
+        Some((
+            heads,
+            changes
+                .iter()
+                .map(|change| change.raw_bytes().to_vec())
+                .collect(),
+        ))
+    }
+
     /// The message just generated is too large to send in one piece: take back what it
     /// carried, as if it had never been generated, and offer a delta instead. The
     /// message being handed over becomes the offer (it carries nothing); the answer is
@@ -1365,6 +1439,55 @@ mod tests {
         assert_eq!(theirs.get_heads(), ours.get_heads(), "both in one message");
     }
 
+    /// Pushed changes (`push`) do what the sync message they stand for would: the peer
+    /// applies them, is level with us, and has nothing to answer; nothing goes twice.
+    #[test]
+    fn pushed_changes_stand_for_a_message() {
+        use crate::document::testing::change;
+        use automerge::{ROOT, transaction::Transactable};
+        let (mut ours, mut theirs) = (Document::new(), Document::new());
+        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
+        let (mut a, mut b) = in_step(&mut ours, &mut theirs);
+        let (now, bytes) = (Instant::now(), SyncBytes::default());
+
+        change(&mut ours, |doc| doc.put(ROOT, "news", 2).unwrap());
+        let (heads, changes) = a.push(&ours, 1 << 20).expect("news to push");
+        assert_eq!(changes.len(), 1);
+        a.assume_delivered(&ours.get_heads());
+        a.send_succeeded(a.id());
+        assert!(a.push(&ours, 1 << 20).is_none(), "sent already");
+
+        let before = theirs.get_heads();
+        b.receive(
+            &mut theirs,
+            Received::pushed(heads, changes, before.clone()),
+        )
+        .unwrap();
+        b.came_from_peer(&theirs, &before);
+        assert_eq!(theirs.get_heads(), ours.get_heads());
+        assert!(b.answered_as_delivered(&theirs, &theirs.get_heads()));
+        assert!(
+            b.generate(&theirs, now, bytes).is_none(),
+            "nothing to answer"
+        );
+    }
+
+    /// Changes larger than the limit aren't pushed: they go as `generate` sends them.
+    #[test]
+    fn large_changes_are_not_pushed() {
+        use crate::document::testing::change;
+        use automerge::{ROOT, transaction::Transactable};
+        let (mut ours, mut theirs) = (Document::new(), Document::new());
+        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
+        let (mut a, _b) = in_step(&mut ours, &mut theirs);
+        change(&mut ours, |doc| {
+            doc.put(ROOT, "news", "x".repeat(100)).unwrap()
+        });
+        assert!(a.push(&ours, 10).is_none());
+        assert!(a.sending.is_none(), "nothing handed over");
+        assert!(a.push(&ours, 1 << 20).is_some());
+    }
+
     /// Our transport refused a message taken as delivered: the peer's heads are as
     /// before, so the next message carries the change again.
     #[test]
diff --git a/src/main.rs b/src/main.rs
index 128e5a0..8c52879 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -219,6 +219,7 @@ async fn main() -> Result<()> {
         timing: state::Timing::new(args.keepalive, sync_reply_timeout, delivery_limit)
             .with_probe_max(args.dormant_probe_max),
         neighbours_k: args.neighbours,
+        push_changes: false,
         alias: args.alias.clone(),
         dashboard_url: args.dashboard_url.clone(),
         electorate_window: args.electorate_window,
diff --git a/src/node.rs b/src/node.rs
index 5bfce31..155ed85 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -44,6 +44,9 @@ pub(crate) struct NodeConfig {
     /// limit)`.
     pub(crate) timing: Timing,
     pub(crate) neighbours_k: usize,
+    /// Changes along writers' trees as they are, not in sync messages
+    /// (`AppState::push_changes`).
+    pub(crate) push_changes: bool,
     pub(crate) alias: Option<String>,
     /// `--dashboard-url`.
     pub(crate) dashboard_url: Option<String>,
@@ -119,6 +122,7 @@ impl Node {
             run_id: uuid::Uuid::new_v4().into_bytes(),
             handshakes: Default::default(),
             neighbours_k: config.neighbours_k,
+            push_changes: config.push_changes,
             alias: config.alias,
             dashboard_url: config.dashboard_url,
             electorate_window: config.electorate_window,
diff --git a/src/simtest/bench.rs b/src/simtest/bench.rs
index 19603d2..91b0aa7 100644
--- a/src/simtest/bench.rs
+++ b/src/simtest/bench.rs
@@ -40,6 +40,13 @@ const ONLY: &str = "BENCH_SCENARIO";
 /// cores, but they share them, so delays (and what depends on timing) may come out
 /// higher than alone; message counts barely.
 const PARALLEL: &str = "BENCH_PARALLEL";
+/// If `1`, changes along writers' trees go as they are, not in Automerge sync messages
+/// (`AppState::push_changes`): `BENCH_PUSH=1`, to measure against.
+const PUSH: &str = "BENCH_PUSH";
+
+fn pushing() -> bool {
+    std::env::var(PUSH).is_ok_and(|value| value == "1")
+}
 const SETTLE: Duration = Duration::from_secs(180);
 
 /// How the swarm is made.
@@ -432,10 +439,12 @@ async fn run(seed: u64, scenario: &'static Scenario) -> String {
     let ids: Vec<String> = (0..scenario.nodes).map(|i| format!("n{i:03}")).collect();
     (scenario.net)(&sim, &ids);
     let reply_timeout = scenario.reply_timeout.unwrap_or(SYNC_REPLY_TIMEOUT);
+    let push = pushing();
     let timing = move |config: &mut crate::node::NodeConfig| {
         config.timing = Timing::new(KEEPALIVE, reply_timeout, DELIVERY_LIMIT);
         // As in production (`--neighbours`).
         config.neighbours_k = 12;
+        config.push_changes = push;
     };
     let mut nodes = vec![sim.start_on_with(&ids[0], (scenario.on)(0), &[], Start::Genesis, timing)];
     // Those there from the start; the others join during the workloads.
@@ -486,10 +495,15 @@ async fn run(seed: u64, scenario: &'static Scenario) -> String {
 
     let _ = writeln!(
         out,
-        "\n{} ({} writes at {} a second each)",
+        "\n{} ({} writes at {} a second each{})",
         scenario.name,
         WRITES,
-        1000 / WRITE_EVERY.as_millis()
+        1000 / WRITE_EVERY.as_millis(),
+        if pushing() {
+            "; changes pushed along trees"
+        } else {
+            ""
+        }
     );
     let mut joined = initial;
     for (w, (workload, writer)) in [
diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs
index e9d8c15..26e85c7 100644
--- a/src/simtest/mod.rs
+++ b/src/simtest/mod.rs
@@ -187,6 +187,7 @@ impl Sim {
             earlier_actors: Vec::new(),
             timing: Timing::new(KEEPALIVE, SYNC_REPLY_TIMEOUT, DELIVERY_LIMIT),
             neighbours_k: 8,
+            push_changes: false,
             alias: Some(id.to_owned()),
             dashboard_url: None,
             electorate_window: Duration::from_secs(30 * 24 * 3600),
diff --git a/src/state/mod.rs b/src/state/mod.rs
index e127877..f9479a1 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -105,6 +105,10 @@ pub(crate) struct AppState {
     /// Target number of neighbours (`--neighbours`); up to this many + 1 members the
     /// swarm is a full mesh.
     pub(crate) neighbours_k: usize,
+    /// Changes passed on along writers' trees go as they are (`SyncBody::Changes`), not
+    /// in Automerge sync messages: an experiment, measured against those
+    /// (`simtest::bench`, `BENCH_PUSH`), off otherwise.
+    pub(crate) push_changes: bool,
     /// Our alias (`--alias`), published in the member list (`swarm::members`).
     pub(crate) alias: Option<String>,
     /// Where people reach our dashboard (`--dashboard-url`), published in the member list
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 8190e5c..d28df33 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -138,6 +138,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
             .map(|estimate| estimate.bytes_per_sec);
         let failed = peer_state_mutex.lock().await.failed_whole(Instant::now());
         let delta_threshold = state.blobs.delta_threshold_for(rate, failed);
+        let push_changes = state.push_changes;
 
         let stats = Arc::clone(&state.stats);
         let clock = Arc::clone(&state.clock);
@@ -167,7 +168,20 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                         delta_threshold,
                         Instant::now(),
                     );
-                    let body = if let Some(from) = again {
+                    // Along a tree, taken as delivered: the changes as they are, if pushing
+                    // (`AppState::push_changes`).
+                    let pushed = (push_changes && level && writer.is_some() && again.is_none())
+                        .then(|| peer_state.sync.push(&replica.doc, delta_threshold))
+                        .flatten();
+                    let body = if let Some((heads, changes)) = pushed {
+                        Some(transport::SyncBody::Changes {
+                            heads,
+                            changes: changes
+                                .into_iter()
+                                .map(serde_bytes::ByteBuf::from)
+                                .collect(),
+                        })
+                    } else if let Some(from) = again {
                         let spec = transport::DeltaSpec::new(from, replica.heads().to_vec());
                         debug!(
                             from = spec.from.len(),
@@ -219,11 +233,24 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                             Some(transport::SyncBody::Delta(spec))
                         })
                     };
-                    let delivered = level && matches!(body, Some(transport::SyncBody::Message(_)));
+                    let delivered = level
+                        && matches!(
+                            body,
+                            Some(
+                                transport::SyncBody::Message(_)
+                                    | transport::SyncBody::Changes { .. }
+                            )
+                        );
                     if delivered {
                         peer_state.sync.assume_delivered(replica.heads());
-                        if let Some(transport::SyncBody::Message(payload)) = &body {
-                            stats.delivered_sent(payload.0.len());
+                        match &body {
+                            Some(transport::SyncBody::Message(payload)) => {
+                                stats.delivered_sent(payload.0.len());
+                            }
+                            Some(transport::SyncBody::Changes { changes, .. }) => {
+                                stats.delivered_sent(changes.iter().map(|c| c.len()).sum());
+                            }
+                            _ => {}
                         }
                     }
                     (
@@ -259,6 +286,9 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                 let inner = match &body {
                     transport::SyncBody::Message(payload)
                     | transport::SyncBody::Deflated(payload) => Some(payload.0.len()),
+                    transport::SyncBody::Changes { changes, .. } => {
+                        Some(changes.iter().map(|change| change.len()).sum())
+                    }
                     transport::SyncBody::Delta(_) => None,
                 };
                 let ids = rmp_serde::encode::to_vec(&transport::EpochTag::of(epoch))
@@ -434,17 +464,33 @@ pub(crate) async fn handle_sync(
     let from_ready = envelope.ready;
     let writer = envelope.writer;
     let delivered = envelope.delivered;
-    let (payload, deflated) = match envelope.body {
-        transport::SyncBody::Message(payload) => (payload, false),
-        transport::SyncBody::Deflated(payload) => (payload, true),
+    // Changes pushed along a tree (`SyncBody::Changes`): their heads and bytes, made
+    // into the Automerge message that would have brought them once the document's lock
+    // is held (`Received::pushed`).
+    let (payload, deflated, pushed) = match envelope.body {
+        transport::SyncBody::Message(payload) => (payload, false, None),
+        transport::SyncBody::Deflated(payload) => (payload, true, None),
+        transport::SyncBody::Changes { heads, changes } => {
+            let changes: Vec<Vec<u8>> = changes
+                .into_iter()
+                .map(serde_bytes::ByteBuf::into_vec)
+                .collect();
+            let len = changes.iter().map(Vec::len).sum::<usize>();
+            (
+                transport::SyncPayload(bytes::Bytes::new()),
+                false,
+                Some((heads, changes, len)),
+            )
+        }
         transport::SyncBody::Delta(spec) => {
             fetch_delta(state, peer, epoch, spec);
             return Ok(());
         }
     };
-    if payload.0.len() > crate::swarm::blob::MAX_SYNC_MESSAGE {
+    let wire_len = pushed.as_ref().map_or(payload.0.len(), |(_, _, len)| *len);
+    if wire_len > crate::swarm::blob::MAX_SYNC_MESSAGE {
         // Not decoded: a node of this version sends a delta instead.
-        warn!(peer = %peer, bytes = payload.0.len(), "Dropping a sync message larger than any sender should send");
+        warn!(peer = %peer, bytes = wire_len, "Dropping a sync message larger than any sender should send");
         return Ok(());
     }
 
@@ -485,14 +531,14 @@ pub(crate) async fn handle_sync(
             } else {
                 message
             };
-            let payload_len = message.0.len();
+            let (payload_len, decoded) = match &pushed {
+                Some((_, _, len)) => (*len, None),
+                None => (
+                    message.0.len(),
+                    Some(crate::document::sync::decode(&message)?),
+                ),
+            };
             stats.sync_received(payload_len);
-            let message = crate::document::sync::decode(&message)?;
-            let carried = message.carries_changes();
-            let sizes = if carried { message.sizes() } else { (0, 0) };
-            if carried {
-                stats.carrying(sizes);
-            }
             let mut replica = doc.blocking_write();
 
             // Authoritative check, under the same lock as the apply: the document may
@@ -500,6 +546,22 @@ pub(crate) async fn handle_sync(
             if replica.epoch != epoch {
                 return Ok(None);
             }
+            let message = match decoded {
+                Some(message) => message,
+                None => {
+                    let (heads, changes, _) = pushed.expect("pushed or decoded");
+                    crate::document::sync::Received::pushed(
+                        heads,
+                        changes,
+                        replica.heads().to_vec(),
+                    )
+                }
+            };
+            let carried = message.carries_changes();
+            let sizes = if carried { message.sizes() } else { (0, 0) };
+            if carried {
+                stats.carrying(sizes);
+            }
 
             let mut peer_state = peer_state_mutex.blocking_lock();
             peer_state.reports_ready = Some(from_ready);
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 474e386..7e5d41b 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -743,6 +743,15 @@ pub(crate) enum SyncBody {
     Message(#[serde(with = "crate::transport::wire::codec::sync_message")] SyncPayload),
     /// An Automerge sync message, deflated (`compress`): a large one, which shrank.
     Deflated(#[serde(with = "crate::transport::wire::codec::sync_message")] SyncPayload),
+    /// Changes passed on along a writer's tree as they are, with the sender's heads,
+    /// instead of an Automerge sync message (`AppState::push_changes`): where the sender
+    /// takes the message as delivered, it has everything the receiver has, and needs
+    /// none of the rest of one (`SyncSession::push`).
+    Changes {
+        #[serde(with = "crate::transport::wire::codec::change_hashes")]
+        heads: Vec<ChangeHash>,
+        changes: Vec<serde_bytes::ByteBuf>,
+    },
     /// Instead of a message too large to send in one piece: fetch these changes as a
     /// blob (`BlobId::Delta`), then sync on.
     Delta(DeltaSpec),
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index df95bce..11aa70d 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -198,7 +198,9 @@ fn related(peer: &str, message: &ParsedMessage) -> Option<String> {
             SyncBody::Delta(spec) => {
                 format!("blob {}", crate::transport::BlobId::Delta(spec.id()))
             }
-            SyncBody::Message(_) | SyncBody::Deflated(_) => return None,
+            SyncBody::Message(_) | SyncBody::Deflated(_) | SyncBody::Changes { .. } => {
+                return None;
+            }
         },
         _ => return None,
     })
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index 2794e69..70ac3ce 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -14,7 +14,7 @@ use super::{EncodedMessage, MessageKind, ParsedMessage};
 
 /// The wire format's version, the first byte of every payload. Bump it with every change
 /// to a message type or a codec (`golden_bytes` fails then).
-pub(crate) const VERSION: u8 = 32;
+pub(crate) const VERSION: u8 = 33;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {