aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/swarm/blob.rs31
-rw-r--r--src/swarm/peer/inbound.rs3
-rw-r--r--src/transport/mod.rs12
-rw-r--r--src/transport/wire.rs2
4 files changed, 42 insertions, 6 deletions
diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index 11bf3dc..0bd2d37 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -9,7 +9,8 @@
//! one sync message would be too large; `swarm::sync`).
//!
//! * Serving: a request without chunks gets the manifest (size, chunk size, a BLAKE3
-//! hash per chunk); one with chunks gets each chunk as a message of its own. A node
+//! hash per chunk); one with chunks gets each chunk as a message of its own, deflated
+//! where that shrinks it (the hash is of the chunk, so any source's matches). A node
//! that doesn't hold the blob answers `Missing`. The bytes of the blob served last
//! are kept, so serving a download doesn't extract them for every chunk. Serving runs
//! in tasks of its own, at most [`MAX_ANSWERING`] requests and [`MAX_SERVING`] chunk
@@ -335,8 +336,19 @@ async fn serve(
let Ok(_permit) = state.blobs.serving.acquire().await else {
return;
};
- let data = bytes::Bytes::copy_from_slice(&prepared.bytes[start..end]);
- reply(state, peer, id, BlobBody::Chunk { index, data }).await;
+ let chunk = &prepared.bytes[start..end];
+ // Deflated where it shrinks (`compress`): changes are stored uncompressed.
+ let body = match crate::compress::deflate(chunk) {
+ Some(deflated) => BlobBody::DeflatedChunk {
+ index,
+ data: deflated.into(),
+ },
+ None => BlobBody::Chunk {
+ index,
+ data: bytes::Bytes::copy_from_slice(chunk),
+ },
+ };
+ reply(state, peer, id, body).await;
}
}
@@ -527,7 +539,18 @@ pub(crate) async fn fetch(
request(state, &source.peer, id, batch, delta).await;
}
- match tokio::time::timeout(Duration::from_millis(500), answers.recv()).await {
+ let answer = tokio::time::timeout(Duration::from_millis(500), answers.recv()).await;
+ // A deflated chunk inflated, to no more than a chunk; one that doesn't is bad.
+ let answer = match answer {
+ Ok(Some((from, BlobBody::DeflatedChunk { index, data }))) => {
+ let data = crate::compress::inflate(&data, manifest.chunk_size as usize)
+ .map(bytes::Bytes::from)
+ .unwrap_or_default();
+ Ok(Some((from, BlobBody::Chunk { index, data })))
+ }
+ other => other,
+ };
+ match answer {
Ok(Some((from, BlobBody::Chunk { index, data }))) => {
let Some(source) = sources.iter_mut().find(|source| source.peer == from) else {
continue;
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index c75f66b..1542d4f 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -174,7 +174,8 @@ pub(crate) async fn handle_message(
// for: the request's round trip is no exchange of sync bytes to time).
if let Some(delay_ms) = crate::swarm::hop::note(state, &peer, blob.sent).await
&& delay_ms > 0
- && let transport::BlobBody::Chunk { data, .. } = &blob.body
+ && let transport::BlobBody::Chunk { data, .. }
+ | transport::BlobBody::DeflatedChunk { data, .. } = &blob.body
{
let rtt =
state
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 1e9e5a9..eeb9d34 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -322,6 +322,13 @@ pub(crate) enum BlobBody {
},
/// The sender doesn't have the blob.
Missing,
+ /// A chunk, deflated (`compress`): the receiver inflates it to the chunk, which the
+ /// manifest's hash is of, whoever sent it.
+ DeflatedChunk {
+ index: u32,
+ #[serde(with = "crate::transport::wire::codec::shared_bytes")]
+ data: bytes::Bytes,
+ },
}
/// Sizes instead of contents: logged messages stay short.
@@ -349,6 +356,11 @@ impl std::fmt::Debug for BlobBody {
.field("bytes", &data.len())
.finish(),
BlobBody::Missing => f.write_str("Missing"),
+ BlobBody::DeflatedChunk { index, data } => f
+ .debug_struct("DeflatedChunk")
+ .field("index", index)
+ .field("bytes", &data.len())
+ .finish(),
}
}
}
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index a722065..2794e69 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 = 31;
+pub(crate) const VERSION: u8 = 32;
/// What the second byte of a payload says.
fn tag(kind: MessageKind) -> u8 {