From 5a32e746cf7aec3e4ca76616b6b66f023de53988 Mon Sep 17 00:00:00 2001 From: tv Date: Mon, 26 Oct 2026 22:08:32 +0100 Subject: add delta blobs --- src/simtest/deltas.rs | 98 +++++++++++++++++++++++++++++++++++++++++++++++++++ src/simtest/mod.rs | 1 + 2 files changed, 99 insertions(+) create mode 100644 src/simtest/deltas.rs (limited to 'src/simtest') diff --git a/src/simtest/deltas.rs b/src/simtest/deltas.rs new file mode 100644 index 0000000..ce74607 --- /dev/null +++ b/src/simtest/deltas.rs @@ -0,0 +1,98 @@ +//! Large transfers as deltas (`docs/peers.md`): a node joining a swarm with much +//! history, and one returning after much was written, get the bulk as a blob, and no +//! sync message they receive is larger than the threshold. +//! +//! The nodes use 1 KiB chunks, so the threshold is 16 KiB, and a few hundred KiB of +//! writes are well above it. + +use std::time::Duration; + +use serde_json::json; + +use super::{Sim, read, write}; +use crate::node::NodeConfig; +use crate::state::AppState; +use crate::swarm::bootstrap::Start; + +const SOON: Duration = Duration::from_secs(10); +/// Catching up a few hundred KiB in 1 KiB chunks. +const CATCH_UP: Duration = Duration::from_secs(60); +const CHUNK: u32 = 1024; +/// 16 chunks (`swarm::blob`). +const THRESHOLD: u64 = 16 * CHUNK as u64; + +fn small_chunks(config: &mut NodeConfig) { + config.blob_chunk_bytes = CHUNK; +} + +/// About 1 KB that doesn't compress (Automerge compresses large changes). +fn noise(seed: u64) -> String { + let mut rng = crate::testrand::Rng::new(seed); + (0..63) + .map(|_| format!("{:016x}", rng.next_u64())) + .collect() +} + +/// `count` writes of about 1 KB each, under `prefix`. +async fn write_much(node: &AppState, prefix: &str, count: usize) { + for i in 0..count { + write(node, &format!("{prefix}/{i}"), json!(noise(i as u64))).await; + } +} + +fn largest_sync_received(node: &AppState) -> u64 { + node.stats.snapshot().largest_sync_received +} + +fn blob_bytes_received(node: &AppState) -> u64 { + node.traffic.total().received.blob.bytes +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn joining_node_gets_the_history_as_a_delta() { + let sim = Sim::new(41); + let a = sim.start_with("a", &[], Start::Genesis, small_chunks); + sim.wait_ready(&[&a], SOON).await; + write_much(&a, "history", 200).await; + + let b = sim.start_with("b", &["a"], Start::Join, small_chunks); + sim.wait_ready(&[&b], CATCH_UP).await; + sim.wait_converged(&[&a, &b], CATCH_UP).await; + assert_eq!(read(&b, "history/199").await, Some(json!(noise(199)))); + + assert!( + blob_bytes_received(&b) > 100 * 1_000, + "{} blob bytes received", + blob_bytes_received(&b) + ); + assert!( + largest_sync_received(&b) <= THRESHOLD, + "a sync message of {} bytes", + largest_sync_received(&b) + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn returning_node_gets_what_it_missed_as_a_delta() { + let sim = Sim::new(42); + let a = sim.start_with("a", &[], Start::Genesis, small_chunks); + let b = sim.start_with("b", &["a"], Start::Join, small_chunks); + let (a, b) = (&a, &b); + sim.wait_ready(&[a, b], SOON).await; + write(a, "early", json!(1)).await; + sim.wait_converged(&[a, b], SOON).await; + + sim.net.take_down("b"); + write_much(a, "while/away", 200).await; + let blobs_before = blob_bytes_received(b); + sim.net.bring_up("b"); + + sim.wait_converged(&[a, b], CATCH_UP).await; + assert_eq!(read(b, "while/away/199").await, Some(json!(noise(199)))); + assert!(blob_bytes_received(b) - blobs_before > 100 * 1_000); + assert!( + largest_sync_received(b) <= THRESHOLD, + "a sync message of {} bytes", + largest_sync_received(b) + ); +} diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs index f7af61b..5e9cc2f 100644 --- a/src/simtest/mod.rs +++ b/src/simtest/mod.rs @@ -9,6 +9,7 @@ //! `RUST_LOG=continuum=debug cargo test simtest -- --nocapture` shows the nodes' logs //! (all nodes interleaved; messages name peers by ID). +mod deltas; mod remakes; mod soak; mod timing; -- cgit v1.3.1