rely on lock files instead of etag

Author: parmesant

src/metastore/metastore_traits.rs

@@ -23,6 +23,7 @@ use bytes::Bytes;
 use chrono::{DateTime, Utc};
 use dashmap::DashMap;
 use erased_serde::Serialize as ErasedSerialize;
+use relative_path::RelativePathBuf;
 use tonic::async_trait;
 use ulid::Ulid;
 
@@ -292,7 +293,17 @@ pub trait Metastore: std::fmt::Debug + Send + Sync {
         stream_name: &str,
         tenant_id: &Option<String>,
     ) -> Result<(), MetastoreError>;
-
+    async fn get_global_schema_locks(
+        &self,
+        stream_name: &str,
+        tenant_id: &Option<String>,
+    ) -> Result<Vec<String>, MetastoreError>;
+    async fn put_global_schema_lock(
+        &self,
+        lock_file_path: &RelativePathBuf,
+        tenant_id: &Option<String>,
+    ) -> Result<(), MetastoreError>;
+    async fn delete_global_schema_lock(&self, path: &RelativePathBuf, tenant_id: &Option<String>);
     /// Global schema operations used to detect and resolve schema conflicts.
     /// A missing expected version creates the record; a version updates it with
     /// an atomic compare-and-swap. `false` indicates only a version conflict.

src/metastore/metastores/object_store_metastore.rs

@@ -27,7 +27,6 @@ use bytes::Bytes;
 use chrono::{DateTime, Utc};
 use dashmap::DashMap;
 use futures::{StreamExt, TryStreamExt};
-use object_store::UpdateVersion;
 use relative_path::RelativePathBuf;
 use tonic::async_trait;
 use tracing::{info, warn};
@@ -1206,7 +1205,48 @@ impl Metastore for ObjectStoreMetastore {
             .put_object(&path, to_bytes(&obj), tenant_id)
             .await?)
     }
-
+    async fn get_global_schema_locks(
+        &self,
+        stream_name: &str,
+        tenant_id: &Option<String>,
+    ) -> Result<Vec<String>, MetastoreError> {
+        let path = if let Some(tenant) = tenant_id {
+            object_store::path::Path::from(format!(
+                "{}/{}/{}",
+                tenant, stream_name, STREAM_ROOT_DIRECTORY
+            ))
+        } else {
+            object_store::path::Path::from(format!("{}/{}", stream_name, STREAM_ROOT_DIRECTORY))
+        };
+        let mut result_file_list = Vec::new();
+        let resp = self.storage.list_with_delimiter(Some(path)).await?;
+        for obj in resp.objects.iter() {
+            let name = obj.location.filename();
+            if let Some(name) = name
+                && name.ends_with(".lock")
+            {
+                result_file_list.push(name.to_owned());
+            }
+        }
+        Ok(result_file_list)
+    }
+    async fn put_global_schema_lock(
+        &self,
+        lock_file_path: &RelativePathBuf,
+        tenant_id: &Option<String>,
+    ) -> Result<(), MetastoreError> {
+        let data = to_bytes(&String::new());
+        self.storage
+            .put_object(&lock_file_path, data, tenant_id)
+            .await?;
+        Ok(())
+    }
+    async fn delete_global_schema_lock(&self, path: &RelativePathBuf, tenant_id: &Option<String>) {
+        // lock file deletion is unconditional
+        if let Err(e) = self.storage.delete_object(&path, tenant_id).await {
+            tracing::error!("Error while deleting lock file- {e}");
+        }
+    }
     /// Retrieve a global schema and the validator from the same object-store GET.
     async fn get_global_schema(
         &self,
@@ -1216,11 +1256,15 @@ impl Metastore for ObjectStoreMetastore {
         // LocalFS does not participate in the shared registry. Only bypass
         // explicitly unsupported backends; propagate errors from cloud stores.
         let path = global_schema_path(stream_name, tenant_id);
-        let global_schema =
-            serde_json::from_slice(&self.storage.get_object(&path, tenant_id).await?)?;
-        let Some((schema, version)) = self.storage.get_object_versioned(&path, tenant_id).await?
-        else {
-            return Ok(None);
+        let global_schema = match self.storage.get_object(&path, tenant_id).await {
+            Ok(obj) => serde_json::from_slice(&obj)?,
+            Err(ObjectStorageError::NoSuchKey(_) | ObjectStorageError::PathError(_)) => {
+                return Ok(None);
+            }
+            Err(e) => {
+                tracing::error!("{e}");
+                return Err(MetastoreError::ObjectStorageError(e));
+            }
         };
         Ok(Some(global_schema))
     }

src/metastore/mod.rs

@@ -49,6 +49,12 @@ pub enum MetastoreError {
     #[error("JSON schema validation error: {message}")]
     JsonSchemaError { message: String },
 
+    #[error("CAS convergence failure: {message}")]
+    CASConvergenceFailure { message: String },
+
+    #[error("LockFile failure: {message}")]
+    LockFileFailure { message: String },
+
     #[error("Invalid JSON structure: expected {expected}, found {found}")]
     InvalidJsonStructure { expected: String, found: String },
 
@@ -100,6 +106,24 @@ impl MetastoreError {
                 metadata: std::collections::HashMap::new(),
                 status_code: 400,
             },
+            MetastoreError::CASConvergenceFailure { message } => MetastoreErrorDetail {
+                operation: "CASConvergenceFailure".to_string(),
+                message: message.to_string(),
+                stream_name: None,
+                file_path: None,
+                timestamp: Some(chrono::Utc::now()),
+                metadata: std::collections::HashMap::new(),
+                status_code: 500,
+            },
+            MetastoreError::LockFileFailure { message } => MetastoreErrorDetail {
+                operation: "LockFileFailure".to_string(),
+                message: message.to_string(),
+                stream_name: None,
+                file_path: None,
+                timestamp: Some(chrono::Utc::now()),
+                metadata: std::collections::HashMap::new(),
+                status_code: 500,
+            },
             MetastoreError::ManifestCodecError(e) => MetastoreErrorDetail {
                 operation: "ManifestCodecError".to_string(),
                 message: e.to_string(),
@@ -164,9 +188,11 @@ impl MetastoreError {
             MetastoreError::JsonParseError(..) => StatusCode::INTERNAL_SERVER_ERROR,
             MetastoreError::ManifestCodecError(..) => StatusCode::INTERNAL_SERVER_ERROR,
             MetastoreError::JsonSchemaError { .. } => StatusCode::INTERNAL_SERVER_ERROR,
-            MetastoreError::InvalidJsonStructure { .. } => StatusCode::INTERNAL_SERVER_ERROR,
+            MetastoreError::InvalidJsonStructure { .. } => StatusCode::BAD_REQUEST,
             MetastoreError::MissingJsonField { .. } => StatusCode::INTERNAL_SERVER_ERROR,
             MetastoreError::InvalidJsonValue { .. } => StatusCode::INTERNAL_SERVER_ERROR,
+            MetastoreError::CASConvergenceFailure { .. }
+            | MetastoreError::LockFileFailure { .. } => StatusCode::INTERNAL_SERVER_ERROR,
             MetastoreError::Error { status_code, .. } => *status_code,
         }
     }

src/schema_registry.rs

@@ -30,13 +30,17 @@ use std::{
 
 use arrow_array::{RecordBatch, RecordBatchOptions};
 use arrow_schema::{ArrowError, DataType, Field, FieldRef, Schema};
+use chrono::{DateTime, Utc};
+use itertools::Itertools;
 use tonic::async_trait;
 
 use crate::{
     event::format::get_datatype_suffix,
+    handlers::http::modal::ingest_server::INGESTOR_META,
     metastore::{MetastoreError, metastore_traits::Metastore},
     option::Mode,
     parseable::{GlobalSchema, PARSEABLE},
+    storage::object_storage::global_schema_lock_file_path,
 };
 
 const MAX_CAS_ATTEMPTS: usize = 16;
@@ -250,7 +254,88 @@ pub async fn resolve(
         stream_name,
         tenant_id,
     };
-    resolve_backend(&backend, stream_name, schemas, pinned).await
+    // before beginning, acquire lock
+    // lock file will be- `tenant/stream/.stream/<nodename>-<timestamp>.lock`
+    let locks = backend
+        .metastore
+        .get_global_schema_locks(stream_name, tenant_id)
+        .await?;
+    if locks.len() == 0 {
+        // proceed with writing lock file
+    } else if locks.len() == 1 {
+        // only one lock file
+        // check the time of the existing lock file
+        // and decide whether to backoff or not
+        let timestamp = DateTime::from_timestamp_micros(
+            *&locks[0].rsplit("-").collect_vec()[0]
+                .parse::<i64>()
+                .unwrap(),
+        )
+        .unwrap();
+
+        let mut slept_already = false;
+        loop {
+            // if delta > 3 minutes then delete and proceed
+            if (Utc::now() - timestamp).num_seconds() > 180 {
+                tracing::warn!(
+                    "delta between lock file- {} and now is greater than 3 minutes, deleting and proceeding",
+                    locks[0]
+                );
+                let lock_file =
+                    global_schema_lock_file_path(stream_name, tenant_id).join(&locks[0]);
+                backend
+                    .metastore
+                    .delete_global_schema_lock(&lock_file, tenant_id)
+                    .await;
+            } else if !slept_already {
+                // sleep for 5 seconds and try again
+                // if lock file still present, skip cycle
+                tokio::time::sleep(Duration::from_secs(5)).await;
+                slept_already = true;
+            } else {
+                // delta is < 180s and slept already
+                // check if lock file exists
+                let locks = backend
+                    .metastore
+                    .get_global_schema_locks(stream_name, tenant_id)
+                    .await?;
+                if locks.is_empty() {
+                    // proceed normally
+                } else {
+                    // skip current cycle
+                    return Err(MetastoreError::LockFileFailure {
+                        message: "Unable to obtain a clean lock file state".into(),
+                    });
+                }
+            }
+        }
+    } else {
+        // multiple lock files present, self healing required here
+        // multiple lock files should only be present in case of multiple failures
+        // safe to assume that conflict resolution can fail now
+        // delete all of them maybe?
+    }
+    let lock_file_path = global_schema_lock_file_path(stream_name, tenant_id).join(format!(
+        "{}-{}.lock",
+        INGESTOR_META
+            .get()
+            .expect("Ingestor node metadata should exist")
+            .node_id,
+        Utc::now().timestamp_micros()
+    ));
+    backend
+        .metastore
+        .put_global_schema_lock(&lock_file_path, tenant_id)
+        .await?;
+
+    // proceed with resolution
+    let res = resolve_backend(&backend, stream_name, schemas, pinned).await;
+
+    backend
+        .metastore
+        .delete_global_schema_lock(&lock_file_path, tenant_id)
+        .await;
+    res
 }
 
 async fn resolve_backend(
@@ -309,7 +394,7 @@ async fn resolve_backend(
 }
 
 fn schema_error(message: impl Into<String>) -> MetastoreError {
-    MetastoreError::JsonSchemaError {
+    MetastoreError::CASConvergenceFailure {
         message: message.into(),
     }
 }

src/storage/mod.rs

@@ -29,7 +29,6 @@ use crate::{
     hottier::StreamHotTier,
     metadata::SchemaVersion,
     metastore::{MetastoreErrorDetail, metastore_traits::MetastoreObject},
-    migration::to_bytes,
     option::StandaloneWithDistributed,
     parseable::{GlobalSchema, StreamNotFound},
     stats::FullStats,

src/storage/object_storage.rs

@@ -1527,6 +1527,15 @@ pub fn global_schema_path(stream_name: &str, tenant_id: &Option<String>) -> Rela
     ])
 }
 
+pub fn global_schema_lock_file_path(stream_name: &str, tenant_id: &Option<String>) -> RelativePathBuf {
+    let tenant = tenant_id.as_deref().unwrap_or("");
+    RelativePathBuf::from_iter([
+        tenant,
+        stream_name,
+        STREAM_ROOT_DIRECTORY,
+    ])
+}
+
 #[inline(always)]
 pub fn stream_json_path(stream_name: &str, tenant_id: &Option<String>) -> RelativePathBuf {
     let tenant = tenant_id.as_deref().unwrap_or("");