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 <node prefix>/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 <fengjiachun@gmail.com>

* test(log-store): drop a comment that restates the keys helper

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

* 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 <fengjiachun@gmail.com>

---------

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

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

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<str>) -> Result<Self> {
-        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<String> {
             reason: format!("object prefix {prefix:?} has an empty or relative component"),
         }
     );
-    Ok(format!("{prefix}/objects/"))
+    Ok(prefix.to_string())
 }
 
 #[cfg(test)]

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<Arc<Self>> {
         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<u8>,
+    }
+
+    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<mock::Metadata> {
+            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<String> {
+        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::<Vec<_>>();
+        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::<Vec<_>>();
+        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<Recovered> {
         let mut objects = Vec::new();