fix blocksync not resyncing compressed blocs properly

set_extension don't behave verywell with extensions containing multiple dots

Cargo.lock

@@ -95,6 +95,9 @@ name = "cc"
 version = "1.0.67"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e3c69b077ad434294d3ce9f1f6143a2a4b89a8a2d54ef813d85003a4fd1137fd"
+dependencies = [
+ "jobserver",
+]
 
 [[package]]
 name = "cfg-if"
@@ -221,6 +224,12 @@ dependencies = [
  "synstructure",
 ]
 
+[[package]]
+name = "fastcdc"
+version = "1.0.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5afa29be46b12c8c380b997def8d1ac77c2665da93eb0a768fab0bf4db79333f"
+
 [[package]]
 name = "fnv"
 version = "1.0.7"
@@ -365,7 +374,7 @@ dependencies = [
  "hex",
  "log",
  "pretty_env_logger",
- "rand",
+ "rand 0.8.3",
  "rmp-serde",
  "serde",
  "sled",
@@ -383,6 +392,7 @@ dependencies = [
  "chrono",
  "crypto-mac 0.10.0",
  "err-derive",
+ "fastcdc",
  "futures",
  "futures-util",
  "garage_model",
@@ -397,6 +407,7 @@ dependencies = [
  "log",
  "md-5",
  "percent-encoding",
+ "rand 0.7.3",
  "roxmltree",
  "sha2",
  "tokio",
@@ -415,12 +426,13 @@ dependencies = [
  "garage_util",
  "hex",
  "log",
- "rand",
+ "rand 0.8.3",
  "rmp-serde",
  "serde",
  "serde_bytes",
  "sled",
  "tokio",
+ "zstd",
 ]
 
 [[package]]
@@ -459,7 +471,7 @@ dependencies = [
  "garage_util",
  "hexdump",
  "log",
- "rand",
+ "rand 0.8.3",
  "rmp-serde",
  "serde",
  "serde_bytes",
@@ -479,7 +491,7 @@ dependencies = [
  "http",
  "hyper",
  "log",
- "rand",
+ "rand 0.8.3",
  "rmp-serde",
  "rustls",
  "serde",
@@ -529,6 +541,17 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "getrandom"
+version = "0.1.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8fc3cb4d91f53b50155bdcfd23f6a4c39ae1969c2ae85982b135750cccaf5fce"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "wasi 0.9.0+wasi-snapshot-preview1",
+]
+
 [[package]]
 name = "getrandom"
 version = "0.2.2"
@@ -537,7 +560,7 @@ checksum = "c9495705279e7140bf035dde1f6e750c162df8b625267cd52cc44e0b156732c8"
 dependencies = [
  "cfg-if",
  "libc",
- "wasi",
+ "wasi 0.10.2+wasi-snapshot-preview1",
 ]
 
 [[package]]
@@ -760,6 +783,15 @@ version = "0.4.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "dd25036021b0de88a0aff6b850051563c6516d0bf53f8638938edbb9de732736"
 
+[[package]]
+name = "jobserver"
+version = "0.1.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5c71313ebb9439f74b00d9d2dcec36440beaf57a6aa0623068441dd7cd81a7f2"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "js-sys"
 version = "0.3.49"
@@ -1043,6 +1075,19 @@ dependencies = [
  "proc-macro2",
 ]
 
+[[package]]
+name = "rand"
+version = "0.7.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6a6b1679d49b24bbfe0c803429aa1874472f50d9b363131f0e89fc356b544d03"
+dependencies = [
+ "getrandom 0.1.16",
+ "libc",
+ "rand_chacha 0.2.2",
+ "rand_core 0.5.1",
+ "rand_hc 0.2.0",
+]
+
 [[package]]
 name = "rand"
 version = "0.8.3"
@@ -1050,9 +1095,19 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0ef9e7e66b4468674bfcb0c81af8b7fa0bb154fa9f28eb840da5c447baeb8d7e"
 dependencies = [
  "libc",
- "rand_chacha",
+ "rand_chacha 0.3.0",
- "rand_core",
+ "rand_core 0.6.2",
- "rand_hc",
+ "rand_hc 0.3.0",
+]
+
+[[package]]
+name = "rand_chacha"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f4c8ed856279c9737206bf725bf36935d8666ead7aa69b52be55af369d193402"
+dependencies = [
+ "ppv-lite86",
+ "rand_core 0.5.1",
 ]
 
 [[package]]
@@ -1062,7 +1117,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e12735cf05c9e10bf21534da50a147b924d555dc7a547c42e6bb2d5b6017ae0d"
 dependencies = [
  "ppv-lite86",
- "rand_core",
+ "rand_core 0.6.2",
+]
+
+[[package]]
+name = "rand_core"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "90bde5296fc891b0cef12a6d03ddccc162ce7b2aff54160af9338f8d40df6d19"
+dependencies = [
+ "getrandom 0.1.16",
 ]
 
 [[package]]
@@ -1071,7 +1135,16 @@ version = "0.6.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "34cf66eb183df1c5876e2dcf6b13d57340741e8dc255b48e40a26de954d06ae7"
 dependencies = [
- "getrandom",
+ "getrandom 0.2.2",
+]
+
+[[package]]
+name = "rand_hc"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ca3129af7b92a17112d59ad498c6f81eaf463253766b90396d39ea7a39d6613c"
+dependencies = [
+ "rand_core 0.5.1",
 ]
 
 [[package]]
@@ -1080,7 +1153,7 @@ version = "0.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3190ef7066a446f2e7f42e239d161e905420ccab01eb967c9eb27d21b2322a73"
 dependencies = [
- "rand_core",
+ "rand_core 0.6.2",
 ]
 
 [[package]]
@@ -1581,6 +1654,12 @@ dependencies = [
  "try-lock",
 ]
 
+[[package]]
+name = "wasi"
+version = "0.9.0+wasi-snapshot-preview1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cccddf32554fecc6acb585f82a32a72e28b48f8c4c1883ddfeeeaa96f7d8e519"
+
 [[package]]
 name = "wasi"
 version = "0.10.2+wasi-snapshot-preview1"
@@ -1703,3 +1782,32 @@ name = "xxhash-rust"
 version = "0.8.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e575e15bedf6e57b5c2d763ffc6c3c760143466cbd09d762d539680ab5992ded"
+
+[[package]]
+name = "zstd"
+version = "0.6.1+zstd.1.4.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5de55e77f798f205d8561b8fe2ef57abfb6e0ff2abe7fd3c089e119cdb5631a3"
+dependencies = [
+ "zstd-safe",
+]
+
+[[package]]
+name = "zstd-safe"
+version = "3.0.1+zstd.1.4.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1387cabcd938127b30ce78c4bf00b30387dddf704e3f0881dbc4ff62b5566f8c"
+dependencies = [
+ "libc",
+ "zstd-sys",
+]
+
+[[package]]
+name = "zstd-sys"
+version = "1.4.20+zstd.1.4.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ebd5b733d7cf2d9447e2c3e76a5589b4f5e5ae065c22a2bc0b023cbc331b6c8e"
+dependencies = [
+ "cc",
+ "libc",
+]

script/dev-cluster.sh

@@ -35,6 +35,9 @@ data_replication_factor = 3
 meta_replication_factor = 3
 meta_epidemic_fanout = 3
 
+enable_compression = true
+compressin_level = 10
+
 [s3_api]
 api_bind_addr = "0.0.0.0:$((3910+$count))"	# the S3 API port, HTTP without TLS. Add a reverse proxy for the TLS part.
 s3_region = "garage"				# set this to anything. S3 API calls will fail if they are not made against the region set here.

script/test-smoke.sh

@@ -23,39 +23,43 @@ dd if=/dev/urandom of=/tmp/garage.1.rnd bs=1k count=2 # < INLINE_THRESHOLD = 307
 dd if=/dev/urandom of=/tmp/garage.2.rnd bs=1M count=5
 dd if=/dev/urandom of=/tmp/garage.3.rnd bs=1M count=10
 
+dd if=/dev/urandom bs=1k count=2  | base64 -w0 > /tmp/garage.1.b64
+dd if=/dev/urandom bs=1M count=5  | base64 -w0 > /tmp/garage.2.b64
+dd if=/dev/urandom bs=1M count=10 | base64 -w0 > /tmp/garage.3.b64
+
 echo "s3 api testing..."
-for idx in $(seq 1 3); do
+for idx in {1,2,3}.{rnd,b64}; do
   # AWS sends	
-  awsgrg cp /tmp/garage.$idx.rnd s3://eprouvette/garage.$idx.aws
+  awsgrg cp /tmp/garage.$idx s3://eprouvette/garage.$idx.aws
   
   awsgrg ls s3://eprouvette
   
   awsgrg cp s3://eprouvette/garage.$idx.aws /tmp/garage.$idx.dl
-  diff /tmp/garage.$idx.rnd /tmp/garage.$idx.dl
+  diff /tmp/garage.$idx /tmp/garage.$idx.dl
   rm /tmp/garage.$idx.dl
   
   s3grg get s3://eprouvette/garage.$idx.aws /tmp/garage.$idx.dl
-  diff /tmp/garage.$idx.rnd /tmp/garage.$idx.dl
+  diff /tmp/garage.$idx /tmp/garage.$idx.dl
   rm /tmp/garage.$idx.dl
   
   awsgrg rm s3://eprouvette/garage.$idx.aws
 
   # S3CMD sends
-  s3grg put /tmp/garage.$idx.rnd s3://eprouvette/garage.$idx.s3cmd
+  s3grg put /tmp/garage.$idx s3://eprouvette/garage.$idx.s3cmd
 
   s3grg ls s3://eprouvette
 	
   s3grg get s3://eprouvette/garage.$idx.s3cmd /tmp/garage.$idx.dl
-  diff /tmp/garage.$idx.rnd /tmp/garage.$idx.dl
+  diff /tmp/garage.$idx /tmp/garage.$idx.dl
   rm /tmp/garage.$idx.dl
   
   awsgrg cp s3://eprouvette/garage.$idx.s3cmd /tmp/garage.$idx.dl
-  diff /tmp/garage.$idx.rnd /tmp/garage.$idx.dl
+  diff /tmp/garage.$idx /tmp/garage.$idx.dl
   rm /tmp/garage.$idx.dl
 	
   s3grg rm s3://eprouvette/garage.$idx.s3cmd
 done
-rm /tmp/garage.{1,2,3}.rnd
+rm /tmp/garage.{1,2,3}.{rnd,b64}
 
 echo "website testing"
 echo "<h1>hello world</h1>" > /tmp/garage-index.html

src/api/Cargo.toml

@@ -22,10 +22,12 @@ bytes = "1.0"
 chrono = "0.4"
 crypto-mac = "0.10"
 err-derive = "0.3"
+fastcdc = "1.0.5"
 hex = "0.4"
 hmac = "0.10"
 log = "0.4"
 md-5 = "0.9"
+rand = "0.7"
 sha2 = "0.9"
 
 futures = "0.3"

src/api/s3_put.rs

@@ -2,6 +2,7 @@ use std::collections::{BTreeMap, VecDeque};
 use std::fmt::Write;
 use std::sync::Arc;
 
+use fastcdc::{Chunk, FastCDC};
 use futures::stream::*;
 use hyper::{Body, Request, Response};
 use md5::{digest::generic_array::*, Digest as Md5Digest, Md5};
@@ -22,6 +23,8 @@ use crate::encoding::*;
 use crate::error::*;
 use crate::signature::verify_signed_content;
 
+// ---- PutObject call ----
+
 pub async fn handle_put(
 	garage: Arc<Garage>,
 	req: Request<Body>,
@@ -150,159 +153,6 @@ pub async fn handle_put(
 	Ok(put_response(version_uuid, md5sum_hex))
 }
 
-/// Validate MD5 sum against content-md5 header
-/// and sha256sum against signed content-sha256
-fn ensure_checksum_matches(
-	data_md5sum: &[u8],
-	data_sha256sum: garage_util::data::FixedBytes32,
-	content_md5: Option<&str>,
-	content_sha256: Option<garage_util::data::FixedBytes32>,
-) -> Result<(), Error> {
-	if let Some(expected_sha256) = content_sha256 {
-		if expected_sha256 != data_sha256sum {
-			return Err(Error::BadRequest(format!(
-				"Unable to validate x-amz-content-sha256"
-			)));
-		} else {
-			trace!("Successfully validated x-amz-content-sha256");
-		}
-	}
-	if let Some(expected_md5) = content_md5 {
-		if expected_md5.trim_matches('"') != base64::encode(data_md5sum) {
-			return Err(Error::BadRequest(format!("Unable to validate content-md5")));
-		} else {
-			trace!("Successfully validated content-md5");
-		}
-	}
-	Ok(())
-}
-
-async fn read_and_put_blocks(
-	garage: &Garage,
-	version: &Version,
-	part_number: u64,
-	first_block: Vec<u8>,
-	first_block_hash: Hash,
-	chunker: &mut BodyChunker,
-) -> Result<(u64, GenericArray<u8, typenum::U16>, Hash), Error> {
-	let mut md5hasher = Md5::new();
-	let mut sha256hasher = Sha256::new();
-	md5hasher.update(&first_block[..]);
-	sha256hasher.update(&first_block[..]);
-
-	let mut next_offset = first_block.len();
-	let mut put_curr_version_block = put_block_meta(
-		&garage,
-		&version,
-		part_number,
-		0,
-		first_block_hash,
-		first_block.len() as u64,
-	);
-	let mut put_curr_block = garage
-		.block_manager
-		.rpc_put_block(first_block_hash, first_block);
-
-	loop {
-		let (_, _, next_block) =
-			futures::try_join!(put_curr_block, put_curr_version_block, chunker.next())?;
-		if let Some(block) = next_block {
-			md5hasher.update(&block[..]);
-			sha256hasher.update(&block[..]);
-			let block_hash = blake2sum(&block[..]);
-			let block_len = block.len();
-			put_curr_version_block = put_block_meta(
-				&garage,
-				&version,
-				part_number,
-				next_offset as u64,
-				block_hash,
-				block_len as u64,
-			);
-			put_curr_block = garage.block_manager.rpc_put_block(block_hash, block);
-			next_offset += block_len;
-		} else {
-			break;
-		}
-	}
-
-	let total_size = next_offset as u64;
-	let data_md5sum = md5hasher.finalize();
-
-	let data_sha256sum = sha256hasher.finalize();
-	let data_sha256sum = Hash::try_from(&data_sha256sum[..]).unwrap();
-
-	Ok((total_size, data_md5sum, data_sha256sum))
-}
-
-async fn put_block_meta(
-	garage: &Garage,
-	version: &Version,
-	part_number: u64,
-	offset: u64,
-	hash: Hash,
-	size: u64,
-) -> Result<(), GarageError> {
-	let mut version = version.clone();
-	version.blocks.put(
-		VersionBlockKey {
-			part_number,
-			offset,
-		},
-		VersionBlock { hash, size },
-	);
-
-	let block_ref = BlockRef {
-		block: hash,
-		version: version.uuid,
-		deleted: false.into(),
-	};
-
-	futures::try_join!(
-		garage.version_table.insert(&version),
-		garage.block_ref_table.insert(&block_ref),
-	)?;
-	Ok(())
-}
-
-struct BodyChunker {
-	body: Body,
-	read_all: bool,
-	block_size: usize,
-	buf: VecDeque<u8>,
-}
-
-impl BodyChunker {
-	fn new(body: Body, block_size: usize) -> Self {
-		Self {
-			body,
-			read_all: false,
-			block_size,
-			buf: VecDeque::with_capacity(2 * block_size),
-		}
-	}
-	async fn next(&mut self) -> Result<Option<Vec<u8>>, GarageError> {
-		while !self.read_all && self.buf.len() < self.block_size {
-			if let Some(block) = self.body.next().await {
-				let bytes = block?;
-				trace!("Body next: {} bytes", bytes.len());
-				self.buf.extend(&bytes[..]);
-			} else {
-				self.read_all = true;
-			}
-		}
-		if self.buf.len() == 0 {
-			Ok(None)
-		} else if self.buf.len() <= self.block_size {
-			let block = self.buf.drain(..).collect::<Vec<u8>>();
-			Ok(Some(block))
-		} else {
-			let block = self.buf.drain(..self.block_size).collect::<Vec<u8>>();
-			Ok(Some(block))
-		}
-	}
-}
-
 pub fn put_response(version_uuid: UUID, md5sum_hex: String) -> Response<Body> {
 	Response::builder()
 		.header("x-amz-version-id", hex::encode(version_uuid))
@@ -311,6 +161,8 @@ pub fn put_response(version_uuid: UUID, md5sum_hex: String) -> Response<Body> {
 		.unwrap()
 }
 
+// ---- Mutlipart upload calls ----
+
 pub async fn handle_create_multipart_upload(
 	garage: Arc<Garage>,
 	req: &Request<Body>,
@@ -575,59 +427,7 @@ pub async fn handle_abort_multipart_upload(
 	Ok(Response::new(Body::from(vec![])))
 }
 
-fn get_mime_type(req: &Request<Body>) -> Result<String, Error> {
+// ---- Parsing input to multipart upload calls ----
-	Ok(req
-		.headers()
-		.get(hyper::header::CONTENT_TYPE)
-		.map(|x| x.to_str())
-		.unwrap_or(Ok("blob"))?
-		.to_string())
-}
-
-pub(crate) fn get_headers(req: &Request<Body>) -> Result<ObjectVersionHeaders, Error> {
-	let content_type = get_mime_type(req)?;
-	let mut other = BTreeMap::new();
-
-	// Preserve standard headers
-	let standard_header = vec![
-		hyper::header::CACHE_CONTROL,
-		hyper::header::CONTENT_DISPOSITION,
-		hyper::header::CONTENT_ENCODING,
-		hyper::header::CONTENT_LANGUAGE,
-		hyper::header::EXPIRES,
-	];
-	for h in standard_header.iter() {
-		if let Some(v) = req.headers().get(h) {
-			match v.to_str() {
-				Ok(v_str) => {
-					other.insert(h.to_string(), v_str.to_string());
-				}
-				Err(e) => {
-					warn!("Discarding header {}, error in .to_str(): {}", h, e);
-				}
-			}
-		}
-	}
-
-	// Preserve x-amz-meta- headers
-	for (k, v) in req.headers().iter() {
-		if k.as_str().starts_with("x-amz-meta-") {
-			match v.to_str() {
-				Ok(v_str) => {
-					other.insert(k.to_string(), v_str.to_string());
-				}
-				Err(e) => {
-					warn!("Discarding header {}, error in .to_str(): {}", k, e);
-				}
-			}
-		}
-	}
-
-	Ok(ObjectVersionHeaders {
-		content_type,
-		other,
-	})
-}
 
 fn decode_upload_id(id: &str) -> Result<UUID, Error> {
 	let id_bin = hex::decode(id).ok_or_bad_request("Invalid upload ID")?;
@@ -674,3 +474,224 @@ fn parse_complete_multpart_upload_body(
 
 	Some(parts)
 }
+
+// ---- Common code ----
+
+pub(crate) fn get_headers(req: &Request<Body>) -> Result<ObjectVersionHeaders, Error> {
+	let content_type = req
+		.headers()
+		.get(hyper::header::CONTENT_TYPE)
+		.map(|x| x.to_str())
+		.unwrap_or(Ok("blob"))?
+		.to_string();
+
+	let mut other = BTreeMap::new();
+
+	// Preserve standard headers
+	let standard_header = vec![
+		hyper::header::CACHE_CONTROL,
+		hyper::header::CONTENT_DISPOSITION,
+		hyper::header::CONTENT_ENCODING,
+		hyper::header::CONTENT_LANGUAGE,
+		hyper::header::EXPIRES,
+	];
+	for h in standard_header.iter() {
+		if let Some(v) = req.headers().get(h) {
+			match v.to_str() {
+				Ok(v_str) => {
+					other.insert(h.to_string(), v_str.to_string());
+				}
+				Err(e) => {
+					warn!("Discarding header {}, error in .to_str(): {}", h, e);
+				}
+			}
+		}
+	}
+
+	// Preserve x-amz-meta- headers
+	for (k, v) in req.headers().iter() {
+		if k.as_str().starts_with("x-amz-meta-") {
+			match v.to_str() {
+				Ok(v_str) => {
+					other.insert(k.to_string(), v_str.to_string());
+				}
+				Err(e) => {
+					warn!("Discarding header {}, error in .to_str(): {}", k, e);
+				}
+			}
+		}
+	}
+
+	Ok(ObjectVersionHeaders {
+		content_type,
+		other,
+	})
+}
+
+struct BodyChunker {
+	body: Body,
+	read_all: bool,
+	min_block_size: usize,
+	avg_block_size: usize,
+	max_block_size: usize,
+	buf: VecDeque<u8>,
+}
+
+impl BodyChunker {
+	fn new(body: Body, block_size: usize) -> Self {
+		let min_block_size = block_size / 4 * 3;
+		let avg_block_size = block_size;
+		let max_block_size = block_size * 2;
+		Self {
+			body,
+			read_all: false,
+			min_block_size,
+			avg_block_size,
+			max_block_size,
+			buf: VecDeque::with_capacity(2 * max_block_size),
+		}
+	}
+	async fn next(&mut self) -> Result<Option<Vec<u8>>, GarageError> {
+		while !self.read_all && self.buf.len() < self.max_block_size {
+			if let Some(block) = self.body.next().await {
+				let bytes = block?;
+				trace!("Body next: {} bytes", bytes.len());
+				self.buf.extend(&bytes[..]);
+			} else {
+				self.read_all = true;
+			}
+		}
+		if self.buf.len() == 0 {
+			Ok(None)
+		} else {
+			let mut iter = FastCDC::with_eof(
+				self.buf.make_contiguous(),
+				self.min_block_size,
+				self.avg_block_size,
+				self.max_block_size,
+				self.read_all,
+			);
+			if let Some(Chunk { length, .. }) = iter.next() {
+				let block = self.buf.drain(..length).collect::<Vec<u8>>();
+				Ok(Some(block))
+			} else {
+				unreachable!("FastCDC returned not chunk")
+			}
+		}
+	}
+}
+
+async fn read_and_put_blocks(
+	garage: &Garage,
+	version: &Version,
+	part_number: u64,
+	first_block: Vec<u8>,
+	first_block_hash: Hash,
+	chunker: &mut BodyChunker,
+) -> Result<(u64, GenericArray<u8, typenum::U16>, Hash), Error> {
+	let mut md5hasher = Md5::new();
+	let mut sha256hasher = Sha256::new();
+	md5hasher.update(&first_block[..]);
+	sha256hasher.update(&first_block[..]);
+
+	let mut next_offset = first_block.len();
+	let mut put_curr_version_block = put_block_meta(
+		&garage,
+		&version,
+		part_number,
+		0,
+		first_block_hash,
+		first_block.len() as u64,
+	);
+	let mut put_curr_block = garage
+		.block_manager
+		.rpc_put_block(first_block_hash, first_block);
+
+	loop {
+		let (_, _, next_block) =
+			futures::try_join!(put_curr_block, put_curr_version_block, chunker.next())?;
+		if let Some(block) = next_block {
+			md5hasher.update(&block[..]);
+			sha256hasher.update(&block[..]);
+			let block_hash = blake2sum(&block[..]);
+			let block_len = block.len();
+			put_curr_version_block = put_block_meta(
+				&garage,
+				&version,
+				part_number,
+				next_offset as u64,
+				block_hash,
+				block_len as u64,
+			);
+			put_curr_block = garage.block_manager.rpc_put_block(block_hash, block);
+			next_offset += block_len;
+		} else {
+			break;
+		}
+	}
+
+	let total_size = next_offset as u64;
+	let data_md5sum = md5hasher.finalize();
+
+	let data_sha256sum = sha256hasher.finalize();
+	let data_sha256sum = Hash::try_from(&data_sha256sum[..]).unwrap();
+
+	Ok((total_size, data_md5sum, data_sha256sum))
+}
+
+async fn put_block_meta(
+	garage: &Garage,
+	version: &Version,
+	part_number: u64,
+	offset: u64,
+	hash: Hash,
+	size: u64,
+) -> Result<(), GarageError> {
+	let mut version = version.clone();
+	version.blocks.put(
+		VersionBlockKey {
+			part_number,
+			offset,
+		},
+		VersionBlock { hash, size },
+	);
+
+	let block_ref = BlockRef {
+		block: hash,
+		version: version.uuid,
+		deleted: false.into(),
+	};
+
+	futures::try_join!(
+		garage.version_table.insert(&version),
+		garage.block_ref_table.insert(&block_ref),
+	)?;
+	Ok(())
+}
+
+/// Validate MD5 sum against content-md5 header
+/// and sha256sum against signed content-sha256
+fn ensure_checksum_matches(
+	data_md5sum: &[u8],
+	data_sha256sum: garage_util::data::FixedBytes32,
+	content_md5: Option<&str>,
+	content_sha256: Option<garage_util::data::FixedBytes32>,
+) -> Result<(), Error> {
+	if let Some(expected_sha256) = content_sha256 {
+		if expected_sha256 != data_sha256sum {
+			return Err(Error::BadRequest(format!(
+				"Unable to validate x-amz-content-sha256"
+			)));
+		} else {
+			trace!("Successfully validated x-amz-content-sha256");
+		}
+	}
+	if let Some(expected_md5) = content_md5 {
+		if expected_md5.trim_matches('"') != base64::encode(data_md5sum) {
+			return Err(Error::BadRequest(format!("Unable to validate content-md5")));
+		} else {
+			trace!("Successfully validated content-md5");
+		}
+	}
+	Ok(())
+}

src/model/Cargo.toml

@@ -21,8 +21,8 @@ arc-swap = "1.0"
 hex = "0.4"
 log = "0.4"
 rand = "0.8"
-
 sled = "0.34"
+zstd = "0.6.1"
 
 rmp-serde = "0.15"
 serde = { version = "1.0", default-features = false, features = ["derive", "rc"] }

src/model/block.rs

@@ -9,6 +9,7 @@ use serde::{Deserialize, Serialize};
 use tokio::fs;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::sync::{watch, Mutex, Notify};
+use zstd::stream::{decode_all as zstd_decode, Encoder};
 
 use garage_util::data::*;
 use garage_util::error::Error;
@@ -42,22 +43,37 @@ pub enum Message {
 	GetBlock(Hash),
 	/// Message to send a block of data, either because requested, of for first delivery of new
 	/// block
-	PutBlock(PutBlockMessage),
+	PutBlock {
+		hash: Hash,
+		data: BlockData,
+	},
 	/// Ask other node if they should have this block, but don't actually have it
 	NeedBlockQuery(Hash),
 	/// Response : whether the node do require that block
 	NeedBlockReply(bool),
 }
 
-/// Structure used to send a block
+/// A possibly compressed block of data
 #[derive(Debug, Serialize, Deserialize)]
-pub struct PutBlockMessage {
+pub enum BlockData {
-	/// Hash of the block
+	Plain(#[serde(with = "serde_bytes")] Vec<u8>),
-	pub hash: Hash,
+	Compressed(#[serde(with = "serde_bytes")] Vec<u8>),
+}
 
-	/// Content of the block
+impl BlockData {
-	#[serde(with = "serde_bytes")]
+	pub fn is_compressed(&self) -> bool {
-	pub data: Vec<u8>,
+		match self {
+			BlockData::Plain(_) => false,
+			BlockData::Compressed(_) => true,
+		}
+	}
+
+	pub fn buffer(&self) -> &Vec<u8> {
+		match self {
+			BlockData::Plain(b) => b,
+			BlockData::Compressed(b) => b,
+		}
+	}
 }
 
 impl RpcMessage for Message {}
@@ -134,7 +150,7 @@ impl BlockManager {
 
 	async fn handle(self: Arc<Self>, msg: &Message) -> Result<Message, Error> {
 		match msg {
-			Message::PutBlock(m) => self.write_block(&m.hash, &m.data).await,
+			Message::PutBlock { hash, data } => self.write_block(&hash, &data).await,
 			Message::GetBlock(h) => self.read_block(h).await,
 			Message::NeedBlockQuery(h) => self.need_block(h).await.map(Message::NeedBlockReply),
 			_ => Err(Error::BadRPC(format!("Unexpected RPC message"))),
@@ -157,54 +173,96 @@ impl BlockManager {
 	}
 
 	/// Write a block to disk
-	async fn write_block(&self, hash: &Hash, data: &[u8]) -> Result<Message, Error> {
+	pub async fn write_block(&self, hash: &Hash, data: &BlockData) -> Result<Message, Error> {
+		let mut path = self.block_dir(hash);
+
 		let _lock = self.data_dir_lock.lock().await;
 
-		let mut path = self.block_dir(hash);
+		let clean_plain = match self.is_block_compressed(hash).await {
-		fs::create_dir_all(&path).await?;
+			Ok(true) => return Ok(Message::Ok),
+			Ok(false) if !data.is_compressed() => return Ok(Message::Ok), // we have a plain block, and the provided block is not compressed either
+			Ok(false) => true,
+			Err(_) => false,
+		};
 
+		fs::create_dir_all(&path).await?;
 		path.push(hex::encode(hash));
-		if fs::metadata(&path).await.is_ok() {
+
-			return Ok(Message::Ok);
+		if data.is_compressed() {
+			path.set_extension("zst");
 		}
 
-		let mut f = fs::File::create(path).await?;
+		let buffer = data.buffer();
-		f.write_all(data).await?;
+
+		let mut f = fs::File::create(path.clone()).await?;
+		f.write_all(&buffer).await?;
 		drop(f);
 
+		if clean_plain {
+			path.set_extension("");
+			fs::remove_file(path).await?;
+		}
+
 		Ok(Message::Ok)
 	}
 
 	/// Read block from disk, verifying it's integrity
-	async fn read_block(&self, hash: &Hash) -> Result<Message, Error> {
+	pub async fn read_block(&self, hash: &Hash) -> Result<Message, Error> {
-		let path = self.block_path(hash);
+		let mut path = self.block_path(hash);
 
-		let mut f = match fs::File::open(&path).await {
+		let mut data = vec![];
-			Ok(f) => f,
+		let block = match self.is_block_compressed(hash).await {
-			Err(e) => {
+			Ok(false) => {
+				let f = fs::File::open(&path).await;
+				f.map(|f| (f, false)).map_err(Into::into)
+			}
+			Ok(true) => {
+				path.set_extension("zst");
+				let f = fs::File::open(&path).await;
+				f.map(|f| (f, true)).map_err(Into::into)
+			}
+			Err(e) => Err(e),
+		};
+		let (mut f, compressed) = match block {
+			Ok(ok) => ok,
+			e => {
 				// Not found but maybe we should have had it ??
 				self.put_to_resync(hash, Duration::from_millis(0))?;
-				return Err(Into::into(e));
+				e?
 			}
 		};
-		let mut data = vec![];
 		f.read_to_end(&mut data).await?;
 		drop(f);
 
-		if blake2sum(&data[..]) != *hash {
+		let sum_ok = if compressed {
+			zstd_check_checksum(&data[..])
+		} else {
+			blake2sum(&data[..]) == *hash
+		};
+		if !sum_ok {
 			let _lock = self.data_dir_lock.lock().await;
 			warn!(
 				"Block {:?} is corrupted. Renaming to .corrupted and resyncing.",
 				hash
 			);
 			let mut path2 = path.clone();
-			path2.set_extension(".corrupted");
+			path2.set_extension("corrupted");
 			fs::rename(path, path2).await?;
 			self.put_to_resync(&hash, Duration::from_millis(0))?;
 			return Err(Error::CorruptData(*hash));
 		}
 
-		Ok(Message::PutBlock(PutBlockMessage { hash: *hash, data }))
+		if compressed {
+			Ok(Message::PutBlock {
+				hash: *hash,
+				data: BlockData::Compressed(data),
+			})
+		} else {
+			Ok(Message::PutBlock {
+				hash: *hash,
+				data: BlockData::Plain(data),
+			})
+		}
 	}
 
 	/// Check if this node should have a block, but don't actually have it
@@ -215,14 +273,23 @@ impl BlockManager {
 			.map(|x| u64_from_be_bytes(x) > 0)
 			.unwrap_or(false);
 		if needed {
-			let path = self.block_path(hash);
+			let exists = self.is_block_compressed(hash).await.is_ok();
-			let exists = fs::metadata(&path).await.is_ok();
 			Ok(!exists)
 		} else {
 			Ok(false)
 		}
 	}
 
+	async fn is_block_compressed(&self, hash: &Hash) -> Result<bool, Error> {
+		let mut path = self.block_path(hash);
+		path.set_extension("zst");
+		if fs::metadata(&path).await.is_ok() {
+			return Ok(true);
+		}
+		path.set_extension("");
+		fs::metadata(&path).await.map(|_| false).map_err(Into::into)
+	}
+
 	fn block_dir(&self, hash: &Hash) -> PathBuf {
 		let mut path = self.data_dir.clone();
 		path.push(hex::encode(&hash.as_slice()[0..1]));
@@ -323,7 +390,7 @@ impl BlockManager {
 
 		let path = self.block_path(hash);
 
-		let exists = fs::metadata(&path).await.is_ok();
+		let exists = self.is_block_compressed(hash).await.is_ok();
 		let needed = self
 			.rc
 			.get(hash.as_ref())?
@@ -402,15 +469,14 @@ impl BlockManager {
 			// TODO find a way to not do this if they are sending it to us
 			// Let's suppose this isn't an issue for now with the BLOCK_RW_TIMEOUT delay
 			// between the RC being incremented and this part being called.
-			let block_data = self.rpc_get_block(&hash).await?;
+			let block = self.rpc_get_raw_block(&hash).await?;
-			self.write_block(hash, &block_data[..]).await?;
+			self.write_block(hash, &block).await?;
 		}
 
 		Ok(())
 	}
 
-	/// Ask nodes that might have a block for it
+	async fn rpc_get_raw_block(&self, hash: &Hash) -> Result<BlockData, Error> {
-	pub async fn rpc_get_block(&self, hash: &Hash) -> Result<Vec<u8>, Error> {
 		let who = self.replication.read_nodes(&hash);
 		let resps = self
 			.rpc_client
@@ -424,8 +490,9 @@ impl BlockManager {
 			.await?;
 
 		for resp in resps {
-			if let Message::PutBlock(msg) = resp {
+			match resp {
-				return Ok(msg.data);
+				Message::PutBlock { data, .. } => return Ok(data),
+				_ => {}
 			}
 		}
 		Err(Error::Message(format!(
@@ -434,13 +501,47 @@ impl BlockManager {
 		)))
 	}
 
+	/// Ask nodes that might have a block for it
+	pub async fn rpc_get_block(&self, hash: &Hash) -> Result<Vec<u8>, Error> {
+		self.rpc_get_raw_block(hash)
+			.await
+			.and_then(|data| match data {
+				BlockData::Plain(data) => Ok(data),
+				BlockData::Compressed(data) => {
+					zstd_decode(&data[..]).map_err(|_| Error::CorruptData(*hash))
+				}
+			})
+	}
+
 	/// Send block to nodes that should have it
 	pub async fn rpc_put_block(&self, hash: Hash, data: Vec<u8>) -> Result<(), Error> {
+		let garage = self.garage.load_full().unwrap();
+
+		let compressed = if garage.config.enable_compression {
+			zstd_encode(&data[..], garage.config.compression_level).ok()
+		} else {
+			None
+		};
+
+		// If compressed data is not less than 7/8 of the size of the original data, i.e. if we
+		// don't gain a significant margin by compressing, then we store the plain data instead
+		// so that we don't lose time decompressing it on reads.
+		let block_data =
+			if compressed.is_some() && compressed.as_ref().unwrap().len() < (data.len() * 7) / 8 {
+				BlockData::Compressed(compressed.unwrap())
+			} else {
+				BlockData::Plain(data)
+			};
+
+		let message = Message::PutBlock {
+			hash,
+			data: block_data,
+		};
 		let who = self.replication.write_nodes(&hash);
 		self.rpc_client
 			.try_call_many(
 				&who[..],
-				Message::PutBlock(PutBlockMessage { hash, data }),
+				message,
 				RequestStrategy::with_quorum(self.replication.write_quorum())
 					.with_timeout(BLOCK_RW_TIMEOUT),
 			)
@@ -498,11 +599,19 @@ impl BlockManager {
 					Err(_) => continue,
 				};
 				let ent_type = data_dir_ent.file_type().await?;
+				let mut iter_name = name.split('.');
+				let name = iter_name
+					.next()
+					.expect("split always contain at least one item");
+				let corrupted = iter_name
+					.last()
+					.map(|ext| ext == "corrupted")
+					.unwrap_or(false);
 
 				if name.len() == 2 && hex::decode(&name).is_ok() && ent_type.is_dir() {
 					self.repair_aux_read_dir_rec(&data_dir_ent.path(), must_exit)
 						.await?;
-				} else if name.len() == 64 {
+				} else if name.len() == 64 && !corrupted {
 					let hash_bytes = match hex::decode(&name) {
 						Ok(h) => h,
 						Err(_) => continue,
@@ -537,3 +646,16 @@ fn u64_from_be_bytes<T: AsRef<[u8]>>(bytes: T) -> u64 {
 	x8.copy_from_slice(bytes.as_ref());
 	u64::from_be_bytes(x8)
 }
+
+fn zstd_check_checksum<R: std::io::Read>(source: R) -> bool {
+	zstd::stream::copy_decode(source, std::io::sink()).is_ok()
+}
+
+fn zstd_encode<R: std::io::Read>(mut source: R, level: i32) -> std::io::Result<Vec<u8>> {
+	let mut result = Vec::<u8>::new();
+	let mut encoder = Encoder::new(&mut result, level)?;
+	encoder.include_checksum(true)?;
+	std::io::copy(&mut source, &mut encoder)?;
+	encoder.finish()?;
+	Ok(result)
+}

src/util/config.rs

@@ -45,6 +45,13 @@ pub struct Config {
 	#[serde(default = "default_replication_factor")]
 	pub data_replication_factor: usize,
 
+	/// Enable Zstd compression of block data
+	pub enable_compression: bool,
+
+	/// Zstd compression level used on data blocks
+	#[serde(default)]
+	pub compression_level: i32,
+
 	/// Configuration for RPC TLS
 	pub rpc_tls: Option<TlsConfig>,