From 1e97a0e7c3e9c2e6023c12bbfe4b04e283f0aaa5 Mon Sep 17 00:00:00 2001 From: jeremyhi Date: Mon, 26 Oct 2026 09:08:13 +0000 Subject: [PATCH] fix(log-store): refuse an object store that ignores conditional create (#9909) * fix(log-store): refuse an object store that ignores conditional create The object store WAL checks only the write_with_if_not_exists capability before it opens, and OpenDAL declares that capability per service. A store behind the S3 service can accept If-None-Match and overwrite the object anyway, so the WAL could overwrite its own objects and lose acknowledged writes without an error. Before recovery, the store now creates a probe key, unique to the open, twice under /probes/, outside the objects it lists, and deletes it. The second create must be refused with a condition failure, otherwise the open fails with an error naming the object store. The probe costs two writes and one delete per open. Signed-off-by: jeremyhi * test(log-store): drop a comment that restates the keys helper Signed-off-by: jeremyhi * fix(log-store): accept AlreadyExists from the second probe create A backend may report an existing conditional-create target as AlreadyExists. The first create stored the probe key, so that error can only refer to it. Signed-off-by: jeremyhi --------- Signed-off-by: jeremyhi --- src/log-store/Cargo.toml | 2 +- src/log-store/src/object_store_wal/io.rs | 59 +++++++++- src/log-store/src/object_store_wal/store.rs | 116 +++++++++++++++++++- 3 files changed, 171 insertions(+), 6 deletions(-) diff --git a/src/log-store/Cargo.toml b/src/log-store/Cargo.toml index 54e0b48c41d..ee147fbcd4b 100644 --- a/src/log-store/Cargo.toml +++ b/src/log-store/Cargo.toml @@ -46,6 +46,7 @@ snafu.workspace = true store-api.workspace = true tokio.workspace = true tokio-util = { workspace = true, features = ["rt"] } +uuid.workspace = true [dev-dependencies] common-meta = { workspace = true, features = ["testing"] } @@ -54,7 +55,6 @@ common-wal = { workspace = true, features = ["testing"] } object-store = { workspace = true, features = ["testing"] } rand.workspace = true rskafka = { workspace = true, features = ["unstable-fuzzing"] } -uuid.workspace = true [package.metadata.cargo-shear] # `protobuf` is required by the code generated by `build.rs`: the generated diff --git a/src/log-store/src/object_store_wal/io.rs b/src/log-store/src/object_store_wal/io.rs index 43f018a7ecb..0c087141c52 100644 --- a/src/log-store/src/object_store_wal/io.rs +++ b/src/log-store/src/object_store_wal/io.rs @@ -17,12 +17,14 @@ use async_stream::try_stream; use bytes::Bytes; +use common_telemetry::warn; use futures::TryStreamExt; use futures::stream::BoxStream; use object_store::{Entry, ErrorKind, ObjectStore}; use snafu::{OptionExt, ResultExt, ensure}; use tokio::sync::mpsc; use tokio_util::task::AbortOnDropHandle; +use uuid::Uuid; use crate::error::{ CorruptedWalObjectSnafu, InvalidWalObjectStoreSnafu, Result, WalObjectConflictSnafu, @@ -33,6 +35,7 @@ use crate::error::{ /// [`u64::MAX`] so that lexicographic and numeric order agree. const OBJECT_SEQ_WIDTH: usize = 20; const OBJECT_SUFFIX: &str = ".wal"; +const PROBE_CONTENT: &[u8] = b"conditional create probe"; /// Objects a listing may hold ahead of the reads that take them, four LIST /// pages of S3, so the listing stays pages ahead of the reads while what it /// holds is bounded. @@ -61,6 +64,9 @@ pub(crate) struct ListedObject { pub(crate) struct ObjectStoreIo { store: ObjectStore, object_prefix: String, + /// Holds the keys of [`Self::probe_conditional_create`], outside + /// `object_prefix` so that no listing of the objects returns one. + probe_prefix: String, } impl ObjectStoreIo { @@ -68,7 +74,7 @@ impl ObjectStoreIo { /// conditional creates, which is what keeps a retried write from /// overwriting a durable object. pub(crate) fn new(store: ObjectStore, prefix: impl AsRef) -> Result { - let object_prefix = normalize_prefix(prefix.as_ref())?; + let prefix = normalize_prefix(prefix.as_ref())?; ensure!( store.info().capability().write_with_if_not_exists, InvalidWalObjectStoreSnafu { @@ -77,10 +83,57 @@ impl ObjectStoreIo { ); Ok(Self { store, - object_prefix, + object_prefix: format!("{prefix}/objects/"), + probe_prefix: format!("{prefix}/probes/"), }) } + /// Checks that the store refuses a conditional create of an object that + /// exists. The capability checked by [`Self::new`] is declared per + /// service, and a store behind the S3 service may accept the condition and + /// overwrite the object anyway, which would let a retried write replace a + /// durable object. The probe creates a key of its own, unique to this + /// call, twice, and then deletes it. + pub(crate) async fn probe_conditional_create(&self) -> Result<()> { + let path = format!("{}{}", self.probe_prefix, Uuid::new_v4()); + let create = || { + self.store + .write_with(&path, PROBE_CONTENT) + .if_not_exists(true) + }; + create().await.context(WalObjectStoreSnafu { + operation: "write", + path: path.clone(), + })?; + let recreated = create().await; + if let Err(error) = self.store.delete(&path).await { + warn!(error; "Failed to delete the conditional create probe {path} of the object store WAL"); + } + match recreated { + // The first create stored this key, so `AlreadyExists` can only + // refer to it. + Err(error) + if matches!( + error.kind(), + ErrorKind::ConditionNotMatch | ErrorKind::AlreadyExists + ) => + { + Ok(()) + } + Err(error) => Err(error).context(WalObjectStoreSnafu { + operation: "write", + path, + }), + Ok(_) => InvalidWalObjectStoreSnafu { + reason: format!( + "the object store ({}) overwrote the existing object {path} on a conditional create, but the object store WAL requires conditional create support; a versioned bucket or the S3-compatible endpoint of GCS can cause this", + self.storage() + ), + } + .fail(), + } + } + /// Creates the object `object_seq` unless it already exists. A retry with /// the same content is accepted, different content is rejected. /// @@ -309,7 +362,7 @@ fn normalize_prefix(prefix: &str) -> Result { reason: format!("object prefix {prefix:?} has an empty or relative component"), } ); - Ok(format!("{prefix}/objects/")) + Ok(prefix.to_string()) } #[cfg(test)] diff --git a/src/log-store/src/object_store_wal/store.rs b/src/log-store/src/object_store_wal/store.rs index 40d6e75326c..a741c337d88 100644 --- a/src/log-store/src/object_store_wal/store.rs +++ b/src/log-store/src/object_store_wal/store.rs @@ -295,8 +295,10 @@ impl fmt::Debug for ObjectStoreLogStore { impl ObjectStoreLogStore { /// Builds the store under the node and generation prefix derived from `config`, /// recovering the catalog from the objects that already exist and writing - /// the object that starts the epoch of this instance. Recovery fails on the - /// first corrupted or conflicting object. + /// the object that starts the epoch of this instance. Before it recovers, + /// it fails unless the object store refuses to overwrite an object on a + /// conditional create. Recovery fails on the first corrupted or + /// conflicting object. pub async fn try_new( object_store: ObjectStore, config: &ObjectStoreWalConfig, @@ -305,6 +307,7 @@ impl ObjectStoreLogStore { ) -> Result> { let prefix = config.node_prefix(node_id, generation); let io = ObjectStoreIo::new(object_store, &prefix)?; + io.probe_conditional_create().await?; Self::open(Arc::new(io), config, prefix).await } @@ -4011,6 +4014,115 @@ mod tests { } } + /// A writer that stores the object whether it exists or not, as an object + /// store that ignores the condition of a create does. + struct OverwritingWriter { + store: ObjectStore, + path: String, + content: Vec, + } + + impl oio::Write for OverwritingWriter { + async fn write(&mut self, buffer: mock::Buffer) -> mock::Result<()> { + self.content.extend_from_slice(&buffer.to_bytes()); + Ok(()) + } + + async fn close(&mut self) -> mock::Result { + let content = std::mem::take(&mut self.content); + self.store.write(&self.path, content).await + } + + async fn abort(&mut self) -> mock::Result<()> { + Ok(()) + } + } + + async fn keys(object_store: &ObjectStore) -> Vec { + let mut keys = object_store + .list_with(&format!("{PREFIX}/")) + .recursive(true) + .await + .unwrap() + .into_iter() + .filter(|entry| entry.metadata().is_file()) + .map(|entry| entry.path().to_string()) + .collect::>(); + keys.sort_unstable(); + keys + } + + #[tokio::test] + async fn test_store_open_deletes_its_conditional_create_probe() { + let _serialized = SERIALIZED.lock().await; + let object_store = memory_store(); + let store = open(object_store.clone(), &eager()).await; + store.stop().await.unwrap(); + + assert_eq!( + vec![object_path(&object_store, 0)], + keys(&object_store).await + ); + } + + #[tokio::test] + async fn test_store_open_refuses_an_object_store_that_ignores_conditional_create() { + let _serialized = SERIALIZED.lock().await; + let object_store = memory_store(); + let target = object_store.clone(); + let ignoring = object_store.clone().layer( + MockLayerBuilder::default() + .writer_factory(Arc::new(move |path, _, _| { + Box::new(OverwritingWriter { + store: target.clone(), + path: path.to_string(), + content: Vec::new(), + }) + })) + .build() + .unwrap(), + ); + + let error = ObjectStoreLogStore::try_new(ignoring, &eager(), 1, 2) + .await + .unwrap_err(); + assert!( + matches!(&error, Error::InvalidWalObjectStore { reason, .. } + if reason.contains("overwrote the existing object")), + "unexpected error: {error:?}" + ); + // The open stops before it writes a start object, and deletes the + // probe. + assert!(keys(&object_store).await.is_empty()); + } + + #[tokio::test] + async fn test_store_recovery_ignores_a_probe_left_behind() { + let _serialized = SERIALIZED.lock().await; + let object_store = memory_store(); + // OpenDAL's unit deleter refuses every delete, so the probe stays. + let undeletable = object_store.clone().layer( + MockLayerBuilder::default() + .deleter_factory(Arc::new(|_| Box::new(()))) + .build() + .unwrap(), + ); + open(undeletable, &eager()).await.stop().await.unwrap(); + let probe_prefix = format!("{PREFIX}/probes/"); + let probes = keys(&object_store) + .await + .into_iter() + .filter(|key| key.starts_with(&probe_prefix)) + .collect::>(); + assert_eq!(1, probes.len(), "{probes:?}"); + + // The next open recovers the start object of the first one only. + let store = open(object_store.clone(), &eager()).await; + let io = ObjectStoreIo::new(object_store, PREFIX).unwrap(); + assert_eq!(vec![0, 1], object_seqs(&io).await); + store.stop().await.unwrap(); + } + /// Rebuilds the catalog by decoding whole objects as a recovery oracle. async fn recover_by_decoding(io: &dyn WalObjectIo) -> Result { let mut objects = Vec::new();