Repository: influxdata/influxdb
File: influxdb3_catalog/src/object_store.rs (Rust)

Code before:
        let catalog_path = CatalogFilePath::checkpoint(&self.prefix);

        let content =
            serialize_catalog_snapshot(snapshot).context("failed to serialize catalog snapshot")?;

        // NOTE: not sure if this should be done in a loop, i.e., what error variants from
        // the object store would warrant a retry.
        match self.store.put(&catalog_path, content.clone().into()).await {
            Ok(put_result) => {
                info!(sequence, "persisted catalog checkpoint file");
                debug!(put_result = ?put_result, "object store PUT result");
                Ok(())
            }
            Err(object_store::Error::NotModified { .. }) => Ok(()),
            Err(err) => {
                error!(error = ?err, "failed to persist catalog checkpoint file");
                Err(err.into())
            }
        }
    }

    /// Persist the `CatalogSnapshot` as a checkpoint in the background but don't check that the
    /// operation succeeds.
    pub(crate) fn background_persist_catalog_checkpoint(
        &self,
        snapshot: &CatalogSnapshot,
    ) -> Result<()> {
        let sequence = snapshot.sequence_number().get();
        debug!(sequence, "background persist of catalog checkpoint");
        let catalog_path = CatalogFilePath::checkpoint(&self.prefix);

        let content =
            serialize_catalog_snapshot(snapshot).context("failed to serialize catalog snapshot")?;

        let store = Arc::clone(&self.store);

        tokio::spawn(async move {
            // NOTE: not sure if this should be done in a loop, i.e., what error variants from
            // the object store would warrant a retry.
            match store.put(&catalog_path, content.clone().into()).await {
                Ok(put_result) => {
                    info!(sequence, "persisted catalog checkpoint file");
                    debug!(put_result = ?put_result, "object store PUT result");
                }
                Err(object_store::Error::NotModified { .. }) => {}
                Err(err) => {
                    error!(error = ?err, "failed to persist catalog checkpoint file");
                }
            }
        });


Review comment (bug):
Using `PutMode::Create` should be optional, and only used during catalog initialization to prevent the race condition.

Otherwise we want to use normal puts, since it always overwrites an existing file.

Code after:
        self.catalog_update_if_not_exists(catalog_path, content)
            .await
    }

    /// Persist the `CatalogSnapshot` as a checkpoint and ensure that the operation succeeds unless
    /// the file already exists. Object Storage should handle the concurrent requests in order and
    /// make sure that only one of them gets the win in terms of who writes first
    pub(crate) async fn persist_catalog_checkpoint(
        &self,
        snapshot: &CatalogSnapshot,
    ) -> Result<PersistCatalogResult> {
        let sequence = snapshot.sequence_number().get();
        let catalog_path = CatalogFilePath::checkpoint(&self.prefix);

        let content =
            serialize_catalog_snapshot(snapshot).context("failed to serialize catalog snapshot")?;

        // NOTE: not sure if this should be done in a loop, i.e., what error variants from
        // the object store would warrant a retry.
        match self
            .store
            .put_opts(
                &catalog_path,
                content.clone().into(),
                PutOptions {
                    mode: object_store::PutMode::Create,
                    ..Default::default()
                },
            )
            .await
        {
            Ok(put_result) => {
                info!(sequence, "persisted catalog checkpoint file");
                debug!(put_result = ?put_result, "object store PUT result");
                Ok(PersistCatalogResult::Success)
            }
            Err(object_store::Error::AlreadyExists { .. }) => {
                Ok(PersistCatalogResult::AlreadyExists)
            }
            Err(err) => {
                error!(error = ?err, "failed to persist catalog checkpoint file");
                Err(err.into())
            }
        }
    }

    /// Persist the `CatalogSnapshot` as a checkpoint in the background but don't check that the
    /// operation succeeds.
    pub(crate) fn background_persist_catalog_checkpoint(
        &self,
        snapshot: &CatalogSnapshot,