forked from Deuxfleurs/garage
Compare commits
15 commits
main
...
storage-op
| Author | SHA1 | Date | |
|---|---|---|---|
| e44bec738c | |||
| f6f0c6f515 | |||
| 7ba1364c08 | |||
| cb02ce3eb2 | |||
| 5a52c4c7f3 | |||
| 64818d3720 | |||
| 852388134b | |||
| 96c7040a8a | |||
| 05dd441e9c | |||
| cffa9821c1 | |||
| d08778dea4 | |||
| 24d605fb05 | |||
| ae4982c309 | |||
| 9c9471c64f | |||
| 71a13f366e |
8 changed files with 530 additions and 263 deletions
130
Cargo.lock
generated
130
Cargo.lock
generated
|
|
@ -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_core",
|
||||
"rand_hc",
|
||||
"rand_chacha 0.3.0",
|
||||
"rand_core 0.6.2",
|
||||
"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",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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> {
|
||||
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,
|
||||
})
|
||||
}
|
||||
// ---- Parsing input to multipart upload calls ----
|
||||
|
||||
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(())
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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"] }
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
/// Hash of the block
|
||||
pub hash: Hash,
|
||||
pub enum BlockData {
|
||||
Plain(#[serde(with = "serde_bytes")] Vec<u8>),
|
||||
Compressed(#[serde(with = "serde_bytes")] Vec<u8>),
|
||||
}
|
||||
|
||||
/// Content of the block
|
||||
#[serde(with = "serde_bytes")]
|
||||
pub data: Vec<u8>,
|
||||
impl BlockData {
|
||||
pub fn is_compressed(&self) -> bool {
|
||||
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);
|
||||
fs::create_dir_all(&path).await?;
|
||||
let clean_plain = match self.is_block_compressed(hash).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?;
|
||||
f.write_all(data).await?;
|
||||
let buffer = data.buffer();
|
||||
|
||||
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> {
|
||||
let path = self.block_path(hash);
|
||||
pub async fn read_block(&self, hash: &Hash) -> Result<Message, Error> {
|
||||
let mut path = self.block_path(hash);
|
||||
|
||||
let mut f = match fs::File::open(&path).await {
|
||||
Ok(f) => f,
|
||||
Err(e) => {
|
||||
let mut data = vec![];
|
||||
let block = match self.is_block_compressed(hash).await {
|
||||
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 = fs::metadata(&path).await.is_ok();
|
||||
let exists = self.is_block_compressed(hash).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?;
|
||||
self.write_block(hash, &block_data[..]).await?;
|
||||
let block = self.rpc_get_raw_block(&hash).await?;
|
||||
self.write_block(hash, &block).await?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Ask nodes that might have a block for it
|
||||
pub async fn rpc_get_block(&self, hash: &Hash) -> Result<Vec<u8>, Error> {
|
||||
async fn rpc_get_raw_block(&self, hash: &Hash) -> Result<BlockData, 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 {
|
||||
return Ok(msg.data);
|
||||
match resp {
|
||||
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)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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>,
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue