diff --git a/CHANGELOG.md b/CHANGELOG.md index eefbe124..7b21f162 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ All notable changes to this project will be documented in this file. This projec If you are upgrading from v0.16.x, replace the binary (or run `docker pull`). If you are upgrading from v0.15.x and below, please read the [upgrading documentation](https://github.com/stalwartlabs/stalwart/blob/main/UPGRADING/v0_16.md) for more information on how to upgrade from previous versions. ## Added +- S3: `verifyAfterWrite` option to verify that objects have persisted after writing. ## Changed - Allow HTTP to be used for configuring the server. diff --git a/crates/registry/src/pickle.rs b/crates/registry/src/pickle.rs index 4ac749c0..dcfcc9f3 100644 --- a/crates/registry/src/pickle.rs +++ b/crates/registry/src/pickle.rs @@ -22,6 +22,7 @@ pub trait Pickle: Sized { pub struct PickledStream<'x> { data: Cow<'x, [u8]>, pos: usize, + version: u8, } pub(crate) fn maybe_compress_pickle(input: Vec) -> Vec { @@ -54,17 +55,20 @@ pub(crate) fn maybe_compress_pickle(input: Vec) -> Vec { impl<'x> PickledStream<'x> { pub fn new(data: &'x [u8]) -> Option { let (marker, data) = data.split_first()?; + let version = marker & !COMPRESS_MARKER; if marker & COMPRESS_MARKER != 0 { lz4_flex::block::decompress_size_prepended(data) .ok() .map(|data| PickledStream { data: Cow::Owned(data), pos: 0, + version, }) } else { PickledStream { data: Cow::Borrowed(data), pos: 0, + version, } .into() } @@ -84,22 +88,26 @@ impl<'x> PickledStream<'x> { }) } + #[inline(always)] pub fn read_bytes(&mut self, len: usize) -> Option<&'_ [u8]> { self.data.get(self.pos..self.pos + len).inspect(|_| { self.pos += len; }) } + #[inline(always)] pub fn eof(&self) -> bool { self.pos >= self.data.len() } + #[inline(always)] pub fn bytes(&self) -> &'_ [u8] { self.data.as_ref() } - pub fn assert_version(&mut self, expected: u8) -> Option { - self.read().filter(|&version| version == expected) + #[inline(always)] + pub fn version(&self) -> u8 { + self.version } } diff --git a/crates/registry/src/schema/properties.rs b/crates/registry/src/schema/properties.rs index 864549d8..9f2ac709 100644 --- a/crates/registry/src/schema/properties.rs +++ b/crates/registry/src/schema/properties.rs @@ -1119,6 +1119,7 @@ pub enum Property { ValidateDomain = 413, Value = 492, VariableName = 675, + VerifyAfterWrite = 874, Version = 80, Vrfy = 526, WaitOnFail = 548, diff --git a/crates/registry/src/schema/properties_impl.rs b/crates/registry/src/schema/properties_impl.rs index b9b3d500..2c456ae1 100644 --- a/crates/registry/src/schema/properties_impl.rs +++ b/crates/registry/src/schema/properties_impl.rs @@ -1272,6 +1272,7 @@ impl EnumImpl for Property { b"validateDomain" => Property::ValidateDomain, b"value" => Property::Value, b"variableName" => Property::VariableName, + b"verifyAfterWrite" => Property::VerifyAfterWrite, b"version" => Property::Version, b"vrfy" => Property::Vrfy, b"waitOnFail" => Property::WaitOnFail, @@ -2151,6 +2152,7 @@ impl EnumImpl for Property { Property::ValidateDomain => "validateDomain", Property::Value => "value", Property::VariableName => "variableName", + Property::VerifyAfterWrite => "verifyAfterWrite", Property::Version => "version", Property::Vrfy => "vrfy", Property::WaitOnFail => "waitOnFail", @@ -3034,6 +3036,7 @@ impl EnumImpl for Property { 413 => Some(Property::ValidateDomain), 492 => Some(Property::Value), 675 => Some(Property::VariableName), + 874 => Some(Property::VerifyAfterWrite), 80 => Some(Property::Version), 526 => Some(Property::Vrfy), 548 => Some(Property::WaitOnFail), diff --git a/crates/registry/src/schema/structs.rs b/crates/registry/src/schema/structs.rs index f974651e..3b867bfb 100644 --- a/crates/registry/src/schema/structs.rs +++ b/crates/registry/src/schema/structs.rs @@ -3370,6 +3370,8 @@ pub struct S3Store { pub key_prefix: Option, #[serde(rename = "allowInvalidCerts")] pub allow_invalid_certs: bool, + #[serde(rename = "verifyAfterWrite")] + pub verify_after_write: bool, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] diff --git a/crates/registry/src/schema/structs_impl.rs b/crates/registry/src/schema/structs_impl.rs index 346cde22..254b0418 100644 --- a/crates/registry/src/schema/structs_impl.rs +++ b/crates/registry/src/schema/structs_impl.rs @@ -3596,7 +3596,7 @@ impl RegistryJsonPropertyPatch for AzureStore { impl ObjectImpl for BlobStore { const FLAGS: u64 = OBJ_SINGLETON; - const VERSION: u8 = 0; + const VERSION: u8 = 1; const OBJECT: ObjectType = ObjectType::BlobStore; fn validate(&self, errors: &mut Vec) -> bool { @@ -25121,6 +25121,7 @@ impl Pickle for S3Store { self.max_retries.pickle(out); self.key_prefix.pickle(out); self.allow_invalid_certs.pickle(out); + self.verify_after_write.pickle(out); } fn unpickle(stream: &mut crate::pickle::PickledStream<'_>) -> Option { @@ -25136,6 +25137,9 @@ impl Pickle for S3Store { this.max_retries = Pickle::unpickle(stream)?; this.key_prefix = Pickle::unpickle(stream)?; this.allow_invalid_certs = Pickle::unpickle(stream)?; + if stream.version() >= 1 { + this.verify_after_write = Pickle::unpickle(stream)?; + } Some(this) } } @@ -25154,13 +25158,14 @@ impl Default for S3Store { max_retries: 3u64, key_prefix: Default::default(), allow_invalid_certs: false, + verify_after_write: true, } } } impl IntoValue for S3Store { fn into_value(self) -> JmapValue<'static> { - let mut map = jmap_tools::Map::with_capacity(13); + let mut map = jmap_tools::Map::with_capacity(14); map.insert_unchecked(Property::Region, self.region.into_value()); map.insert_unchecked(Property::Bucket, self.bucket.into_value()); map.insert_unchecked(Property::AccessKey, self.access_key.into_value()); @@ -25175,6 +25180,10 @@ impl IntoValue for S3Store { Property::AllowInvalidCerts, self.allow_invalid_certs.into_value(), ); + map.insert_unchecked( + Property::VerifyAfterWrite, + self.verify_after_write.into_value(), + ); JmapValue::Object(map) } } @@ -25205,6 +25214,7 @@ impl RegistryJsonPropertyPatch for S3Store { .key_prefix .patch(pointer.with_validators(&[StringValidator::Trim]), value), Some(Property::AllowInvalidCerts) => self.allow_invalid_certs.patch(pointer, value), + Some(Property::VerifyAfterWrite) => self.verify_after_write.patch(pointer, value), Some(Property::Type) => Ok(MaybeUnpatched::Unpatched { property: Property::Type, value, diff --git a/crates/store/src/backend/s3/mod.rs b/crates/store/src/backend/s3/mod.rs index 9c1f7ad4..7e004cfc 100644 --- a/crates/store/src/backend/s3/mod.rs +++ b/crates/store/src/backend/s3/mod.rs @@ -14,6 +14,7 @@ pub struct S3Store { bucket: Box, prefix: Option, max_retries: u32, + verify_after_write: bool, } impl S3Store { @@ -89,6 +90,7 @@ impl S3Store { .map_err(|err| format!("Failed to create bucket: {err:?}"))?, max_retries: config.max_retries as u32, prefix: config.key_prefix, + verify_after_write: config.verify_after_write, }))) } @@ -136,17 +138,53 @@ impl S3Store { } pub(crate) async fn put_blob(&self, key: &[u8], data: &[u8]) -> trc::Result<()> { + let path = self.build_key(key); let mut retries_left = self.max_retries; loop { let response = self .bucket - .put_object(self.build_key(key), data) + .put_object(&path, data) .await .map_err(into_error)?; match response.status_code() { - 200..=299 => return Ok(()), + 200..=299 => { + if !self.verify_after_write { + return Ok(()); + } + + // Some S3-compatible backends acknowledge a PUT before the + // write is durable. HEAD the object to confirm it is visible + // to the read path before reporting success. + let (_, head_status) = + self.bucket.head_object(&path).await.map_err(into_error)?; + + match head_status { + 200..=299 => return Ok(()), + 404 | 500..=599 if retries_left > 0 => { + tokio::time::sleep(Duration::from_secs( + 1 << (self.max_retries - retries_left).min(6), + )) + .await; + + retries_left -= 1; + } + 404 => { + return Err(trc::StoreEvent::S3Error + .reason(concat!( + "PUT acknowledged with 2xx but object not visible", + "to read path; backend may be silently losing writes" + )) + .ctx(trc::Key::Code, head_status)); + } + code => { + return Err(trc::StoreEvent::S3Error + .reason("HEAD verification failed after PUT") + .ctx(trc::Key::Code, code)); + } + } + } 500..=599 if retries_left > 0 => { // wait backoff tokio::time::sleep(Duration::from_secs( diff --git a/resources/schema/schema.json.gz b/resources/schema/schema.json.gz index 38e993b6..70b685bb 100644 Binary files a/resources/schema/schema.json.gz and b/resources/schema/schema.json.gz differ diff --git a/resources/schema/schema.json.sha256 b/resources/schema/schema.json.sha256 index 79b6bb88..2475ff30 100644 --- a/resources/schema/schema.json.sha256 +++ b/resources/schema/schema.json.sha256 @@ -1 +1 @@ -_pIIRU33XTKDaP9kLFoPsYXayXbaRi79QFBUrdgixCY \ No newline at end of file +QRJ4vibPf1dYpzaA4YziIqMnAMR1UOHVUErrjOyjDeE \ No newline at end of file diff --git a/tests/Cargo.toml b/tests/Cargo.toml index efc9cbe0..176a3795 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -6,7 +6,7 @@ edition = "2024" [features] #default = ["sqlite", "postgres", "mysql", "rocks", "s3", "redis", "nats", "azure", "foundationdb"] #default = ["sqlite", "postgres", "mysql", "rocks", "s3", "redis", "foundationdb"] -default = ["rocks"] +default = ["rocks", "s3"] sqlite = ["store/sqlite", "directory/sqlite"] foundationdb = ["store/foundation", "common/foundation"] postgres = ["store/postgres", "directory/postgres"]