mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-17 10:17:55 +00:00
test(table-catalog): move the store-side stateful object backend into test_support
First half of the issue's PR2: the store tests' TestCatalogObjectBackend cluster (state/record/locks/pause types, the seed/fail/pause instrumented inherent impl, the TableCatalogObjectBackend trait impl, and the BlockingObjectPublication/UnserializedTestPublication commit-publication fakes — 544 lines) moves verbatim from table_catalog/tests.rs into test_support.rs, with pub(crate) visibility on the items and fields the tests reach directly. Pure move, no behavior change; the admin handler tests' TestTableCatalogObjectBackend union (its put barrier / fail-path / lock-attempt instrumentation folding into this fake) is the second half. Verification: cargo test -p rustfs --lib table_catalog 481 passed; clippy --lib --tests -D warnings clean; make pre-commit green. Ref rustfs/backlog#1837 (PR2, part 1).
This commit is contained in:
overtrue
@@ -22,6 +22,15 @@
|
||||
//! fixed values (sequence 7 / snapshot 20), which keeps every produced byte
|
||||
//! identical to the pre-extraction fixtures.
|
||||
|
||||
use std::collections::{BTreeMap, BTreeSet};
|
||||
use std::sync::Arc;
|
||||
use time::OffsetDateTime;
|
||||
|
||||
use super::{
|
||||
StrongTableCatalogRuntime, TableCatalogObject, TableCatalogObjectBackend, TableCatalogObjectMetadata,
|
||||
TableCatalogPutPrecondition, TableCatalogStoreError, TableCatalogStoreResult, TableCommitPublication,
|
||||
};
|
||||
|
||||
pub(crate) fn table_metadata_json(table_uuid: &str, location: &str) -> serde_json::Value {
|
||||
serde_json::json!({
|
||||
"format-version": 2,
|
||||
@@ -226,3 +235,561 @@ pub(crate) fn manifest_avro_bytes_with_nullable_sequences(files: &[(&str, i32, i
|
||||
}
|
||||
writer.into_inner().expect("manifest avro bytes should flush")
|
||||
}
|
||||
|
||||
// --- Stateful object backend shared by the store and admin handler tests
|
||||
// (backlog#1837 PR2). Superset instrumentation lands here incrementally;
|
||||
// this is the store-side fake moved verbatim.
|
||||
|
||||
#[derive(Clone, Default)]
|
||||
pub(crate) struct TestCatalogObjectBackend {
|
||||
pub(crate) state: Arc<tokio::sync::Mutex<TestCatalogObjectState>>,
|
||||
pub(crate) locks: TestCatalogObjectLocks,
|
||||
pub(crate) strong_runtime: Option<StrongTableCatalogRuntime>,
|
||||
}
|
||||
|
||||
pub(crate) type TestCatalogObjectLockKey = (String, String);
|
||||
pub(crate) type TestCatalogObjectLock = Arc<tokio::sync::RwLock<()>>;
|
||||
pub(crate) type TestCatalogObjectLocks = Arc<tokio::sync::Mutex<BTreeMap<TestCatalogObjectLockKey, TestCatalogObjectLock>>>;
|
||||
|
||||
#[derive(Clone, Default)]
|
||||
pub(crate) struct TestCatalogObjectPause {
|
||||
started: Arc<tokio::sync::Notify>,
|
||||
release: Arc<tokio::sync::Notify>,
|
||||
}
|
||||
|
||||
impl TestCatalogObjectPause {
|
||||
pub(crate) async fn wait_started(&self) {
|
||||
self.started.notified().await;
|
||||
}
|
||||
|
||||
pub(crate) fn release(&self) {
|
||||
self.release.notify_one();
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub(crate) struct BlockingObjectPublication {
|
||||
backend: TestCatalogObjectBackend,
|
||||
object: String,
|
||||
started: Arc<tokio::sync::Notify>,
|
||||
guard: Arc<parking_lot::Mutex<Option<Box<dyn Send>>>>,
|
||||
}
|
||||
|
||||
impl BlockingObjectPublication {
|
||||
pub(crate) fn new(backend: TestCatalogObjectBackend, object: impl Into<String>) -> Self {
|
||||
Self {
|
||||
backend,
|
||||
object: object.into(),
|
||||
started: Arc::new(tokio::sync::Notify::new()),
|
||||
guard: Arc::new(parking_lot::Mutex::new(None)),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) async fn wait_started(&self) {
|
||||
self.started.notified().await;
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
pub(crate) struct UnserializedTestPublication;
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl TableCommitPublication for UnserializedTestPublication {
|
||||
async fn begin_table_bucket(&self, _table_bucket: &str) -> TableCatalogStoreResult<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn prepare(&self, _table_bucket: &str, _namespace: &str, _table: &str) -> TableCatalogStoreResult<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn holds_table_bucket(&self, _table_bucket: &str) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn holds_table(&self, _table_bucket: &str, _namespace: &str, _table: &str) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn complete(&self) {}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl TableCommitPublication for BlockingObjectPublication {
|
||||
async fn begin_table_bucket(&self, _table_bucket: &str) -> TableCatalogStoreResult<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn prepare(&self, table_bucket: &str, _namespace: &str, _table: &str) -> TableCatalogStoreResult<()> {
|
||||
self.started.notify_one();
|
||||
let guard = self.backend.acquire_read_lock(table_bucket, &self.object).await?;
|
||||
*self.guard.lock() = Some(guard);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn holds_table_bucket(&self, _table_bucket: &str) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn holds_table(&self, _table_bucket: &str, _namespace: &str, _table: &str) -> bool {
|
||||
self.guard.lock().is_some()
|
||||
}
|
||||
|
||||
fn complete(&self) {
|
||||
drop(self.guard.lock().take());
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
pub(crate) struct TestCatalogObjectState {
|
||||
pub(crate) objects: BTreeMap<(String, String), TestCatalogObjectRecord>,
|
||||
pub(crate) etagless_objects: BTreeSet<(String, String)>,
|
||||
pub(crate) fail_read_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
|
||||
pub(crate) pause_before_read_attempts: BTreeMap<(String, String), BTreeMap<usize, TestCatalogObjectPause>>,
|
||||
pub(crate) pause_read_attempts: BTreeMap<(String, String), BTreeMap<usize, TestCatalogObjectPause>>,
|
||||
pub(crate) read_attempts: BTreeMap<(String, String), usize>,
|
||||
pub(crate) read_limits: Vec<((String, String), usize)>,
|
||||
pub(crate) fail_put_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
|
||||
pub(crate) fail_after_put_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
|
||||
pub(crate) pause_put_attempts: BTreeMap<(String, String), BTreeMap<usize, TestCatalogObjectPause>>,
|
||||
pub(crate) fail_delete_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
|
||||
pub(crate) fail_after_delete_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
|
||||
pub(crate) put_attempts: BTreeMap<(String, String), usize>,
|
||||
pub(crate) delete_attempts: BTreeMap<(String, String), usize>,
|
||||
pub(crate) write_lock_acquisitions: BTreeMap<(String, String), usize>,
|
||||
pub(crate) read_lock_acquisitions: BTreeMap<(String, String), usize>,
|
||||
pub(crate) read_calls: usize,
|
||||
pub(crate) metadata_calls: usize,
|
||||
pub(crate) list_calls: usize,
|
||||
pub(crate) next_etag: u64,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub(crate) struct TestCatalogObjectRecord {
|
||||
pub(crate) data: Vec<u8>,
|
||||
pub(crate) etag: String,
|
||||
pub(crate) mod_time: Option<OffsetDateTime>,
|
||||
}
|
||||
|
||||
impl TestCatalogObjectBackend {
|
||||
pub(crate) async fn seed_object(&self, bucket: &str, object: &str, data: Vec<u8>) {
|
||||
self.seed_object_with_mod_time(bucket, object, data, Some(OffsetDateTime::UNIX_EPOCH))
|
||||
.await;
|
||||
}
|
||||
|
||||
pub(crate) async fn seed_object_with_mod_time(
|
||||
&self,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
data: Vec<u8>,
|
||||
mod_time: Option<OffsetDateTime>,
|
||||
) {
|
||||
let mut state = self.state.lock().await;
|
||||
let etag = state.next_etag();
|
||||
state
|
||||
.objects
|
||||
.insert((bucket.to_string(), object.to_string()), TestCatalogObjectRecord { data, etag, mod_time });
|
||||
}
|
||||
|
||||
pub(crate) async fn fail_put_attempt(&self, bucket: &str, object: &str, attempt: usize) {
|
||||
let mut state = self.state.lock().await;
|
||||
state
|
||||
.fail_put_attempts
|
||||
.entry((bucket.to_string(), object.to_string()))
|
||||
.or_default()
|
||||
.insert(attempt);
|
||||
}
|
||||
|
||||
pub(crate) async fn fail_delete_attempt(&self, bucket: &str, object: &str, attempt: usize) {
|
||||
let mut state = self.state.lock().await;
|
||||
state
|
||||
.fail_delete_attempts
|
||||
.entry((bucket.to_string(), object.to_string()))
|
||||
.or_default()
|
||||
.insert(attempt);
|
||||
}
|
||||
|
||||
pub(crate) async fn list_call_count(&self) -> usize {
|
||||
self.state.lock().await.list_calls
|
||||
}
|
||||
|
||||
pub(crate) async fn read_call_count(&self) -> usize {
|
||||
self.state.lock().await.read_calls
|
||||
}
|
||||
|
||||
pub(crate) async fn metadata_call_count(&self) -> usize {
|
||||
self.state.lock().await.metadata_calls
|
||||
}
|
||||
|
||||
pub(crate) async fn reset_call_counts(&self) {
|
||||
let mut state = self.state.lock().await;
|
||||
state.read_calls = 0;
|
||||
state.metadata_calls = 0;
|
||||
state.list_calls = 0;
|
||||
}
|
||||
|
||||
pub(crate) async fn write_lock_acquisition_count(&self, bucket: &str, object: &str) -> usize {
|
||||
self.state
|
||||
.lock()
|
||||
.await
|
||||
.write_lock_acquisitions
|
||||
.get(&(bucket.to_string(), object.to_string()))
|
||||
.copied()
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
pub(crate) async fn read_lock_acquisition_count(&self, bucket: &str, object: &str) -> usize {
|
||||
self.state
|
||||
.lock()
|
||||
.await
|
||||
.read_lock_acquisitions
|
||||
.get(&(bucket.to_string(), object.to_string()))
|
||||
.copied()
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
pub(crate) async fn fail_next_read(&self, bucket: &str, object: &str) {
|
||||
let mut state = self.state.lock().await;
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let next_attempt = state.read_attempts.get(&key).copied().unwrap_or_default() + 1;
|
||||
state.fail_read_attempts.entry(key).or_default().insert(next_attempt);
|
||||
}
|
||||
|
||||
pub(crate) async fn pause_next_read(&self, bucket: &str, object: &str) -> TestCatalogObjectPause {
|
||||
let mut state = self.state.lock().await;
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let next_attempt = state.read_attempts.get(&key).copied().unwrap_or_default() + 1;
|
||||
let pause = TestCatalogObjectPause::default();
|
||||
state
|
||||
.pause_read_attempts
|
||||
.entry(key)
|
||||
.or_default()
|
||||
.insert(next_attempt, pause.clone());
|
||||
pause
|
||||
}
|
||||
|
||||
pub(crate) async fn pause_before_next_read(&self, bucket: &str, object: &str) -> TestCatalogObjectPause {
|
||||
let mut state = self.state.lock().await;
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let next_attempt = state.read_attempts.get(&key).copied().unwrap_or_default() + 1;
|
||||
let pause = TestCatalogObjectPause::default();
|
||||
state
|
||||
.pause_before_read_attempts
|
||||
.entry(key)
|
||||
.or_default()
|
||||
.insert(next_attempt, pause.clone());
|
||||
pause
|
||||
}
|
||||
|
||||
pub(crate) async fn omit_etag_for_object(&self, bucket: &str, object: &str) {
|
||||
self.state
|
||||
.lock()
|
||||
.await
|
||||
.etagless_objects
|
||||
.insert((bucket.to_string(), object.to_string()));
|
||||
}
|
||||
|
||||
pub(crate) async fn last_read_limit(&self, bucket: &str, object: &str) -> Option<usize> {
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
self.state
|
||||
.lock()
|
||||
.await
|
||||
.read_limits
|
||||
.iter()
|
||||
.rev()
|
||||
.find_map(|(read_key, limit)| (read_key == &key).then_some(*limit))
|
||||
}
|
||||
|
||||
pub(crate) async fn fail_next_put(&self, bucket: &str, object: &str) {
|
||||
let mut state = self.state.lock().await;
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let next_attempt = state.put_attempts.get(&key).copied().unwrap_or_default() + 1;
|
||||
state.fail_put_attempts.entry(key).or_default().insert(next_attempt);
|
||||
}
|
||||
|
||||
pub(crate) async fn fail_after_next_put(&self, bucket: &str, object: &str) {
|
||||
let mut state = self.state.lock().await;
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let next_attempt = state.put_attempts.get(&key).copied().unwrap_or_default() + 1;
|
||||
state.fail_after_put_attempts.entry(key).or_default().insert(next_attempt);
|
||||
}
|
||||
|
||||
pub(crate) async fn fail_after_next_delete(&self, bucket: &str, object: &str) {
|
||||
let mut state = self.state.lock().await;
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let next_attempt = state.delete_attempts.get(&key).copied().unwrap_or_default() + 1;
|
||||
state.fail_after_delete_attempts.entry(key).or_default().insert(next_attempt);
|
||||
}
|
||||
|
||||
pub(crate) async fn pause_next_put(&self, bucket: &str, object: &str) -> TestCatalogObjectPause {
|
||||
let mut state = self.state.lock().await;
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let next_attempt = state.put_attempts.get(&key).copied().unwrap_or_default() + 1;
|
||||
let pause = TestCatalogObjectPause::default();
|
||||
state
|
||||
.pause_put_attempts
|
||||
.entry(key)
|
||||
.or_default()
|
||||
.insert(next_attempt, pause.clone());
|
||||
pause
|
||||
}
|
||||
|
||||
pub(crate) async fn put_attempt_count(&self, bucket: &str, object: &str) -> usize {
|
||||
self.state
|
||||
.lock()
|
||||
.await
|
||||
.put_attempts
|
||||
.get(&(bucket.to_string(), object.to_string()))
|
||||
.copied()
|
||||
.unwrap_or_default()
|
||||
}
|
||||
}
|
||||
|
||||
impl TestCatalogObjectState {
|
||||
pub(crate) fn next_etag(&mut self) -> String {
|
||||
self.next_etag += 1;
|
||||
format!("etag-{}", self.next_etag)
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl TableCatalogObjectBackend for TestCatalogObjectBackend {
|
||||
fn strong_catalog_runtime(&self) -> Option<StrongTableCatalogRuntime> {
|
||||
self.strong_runtime.clone()
|
||||
}
|
||||
|
||||
async fn read_object(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Option<TableCatalogObject>> {
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let (attempt, pause_before) = {
|
||||
let mut state = self.state.lock().await;
|
||||
state.read_calls += 1;
|
||||
let attempt = {
|
||||
let attempts = state.read_attempts.entry(key.clone()).or_default();
|
||||
*attempts += 1;
|
||||
*attempts
|
||||
};
|
||||
if state
|
||||
.fail_read_attempts
|
||||
.get(&key)
|
||||
.is_some_and(|attempts| attempts.contains(&attempt))
|
||||
{
|
||||
return Err(TableCatalogStoreError::Internal(format!(
|
||||
"injected read failure for {object} attempt {attempt}"
|
||||
)));
|
||||
}
|
||||
let pause = state
|
||||
.pause_before_read_attempts
|
||||
.get_mut(&key)
|
||||
.and_then(|attempts| attempts.remove(&attempt));
|
||||
(attempt, pause)
|
||||
};
|
||||
if let Some(pause) = pause_before {
|
||||
pause.started.notify_one();
|
||||
pause.release.notified().await;
|
||||
}
|
||||
let (result, pause) = {
|
||||
let mut state = self.state.lock().await;
|
||||
let etagless = state.etagless_objects.contains(&key);
|
||||
let result = state.objects.get(&key).map(|record| TableCatalogObject {
|
||||
data: record.data.clone(),
|
||||
etag: (!etagless).then(|| record.etag.clone()),
|
||||
mod_time: record.mod_time,
|
||||
});
|
||||
let pause = state
|
||||
.pause_read_attempts
|
||||
.get_mut(&key)
|
||||
.and_then(|attempts| attempts.remove(&attempt));
|
||||
(result, pause)
|
||||
};
|
||||
if let Some(pause) = pause {
|
||||
pause.started.notify_one();
|
||||
pause.release.notified().await;
|
||||
}
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
async fn read_object_limited(
|
||||
&self,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
max_size: usize,
|
||||
) -> TableCatalogStoreResult<Option<TableCatalogObject>> {
|
||||
self.state
|
||||
.lock()
|
||||
.await
|
||||
.read_limits
|
||||
.push(((bucket.to_string(), object.to_string()), max_size));
|
||||
let result = self.read_object(bucket, object).await?;
|
||||
if result.as_ref().is_some_and(|object| object.data.len() > max_size) {
|
||||
return Err(TableCatalogStoreError::Invalid(format!(
|
||||
"catalog object {bucket}/{object} exceeds the maximum size of {max_size} bytes"
|
||||
)));
|
||||
}
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
async fn object_metadata(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Option<TableCatalogObjectMetadata>> {
|
||||
let mut state = self.state.lock().await;
|
||||
state.metadata_calls += 1;
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let etagless = state.etagless_objects.contains(&key);
|
||||
Ok(state.objects.get(&key).map(|record| TableCatalogObjectMetadata {
|
||||
etag: (!etagless).then(|| record.etag.clone()),
|
||||
mod_time: record.mod_time,
|
||||
}))
|
||||
}
|
||||
|
||||
async fn object_exists(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<bool> {
|
||||
let state = self.state.lock().await;
|
||||
Ok(state.objects.contains_key(&(bucket.to_string(), object.to_string())))
|
||||
}
|
||||
|
||||
async fn put_object(
|
||||
&self,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
data: Vec<u8>,
|
||||
precondition: TableCatalogPutPrecondition,
|
||||
) -> TableCatalogStoreResult<()> {
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let (attempt, pause) = {
|
||||
let mut state = self.state.lock().await;
|
||||
let attempt = {
|
||||
let attempts = state.put_attempts.entry(key.clone()).or_default();
|
||||
*attempts += 1;
|
||||
*attempts
|
||||
};
|
||||
if state
|
||||
.fail_put_attempts
|
||||
.get(&key)
|
||||
.is_some_and(|attempts| attempts.contains(&attempt))
|
||||
{
|
||||
return Err(TableCatalogStoreError::Internal(format!(
|
||||
"injected put failure for {object} attempt {attempt}"
|
||||
)));
|
||||
}
|
||||
let pause = state
|
||||
.pause_put_attempts
|
||||
.get_mut(&key)
|
||||
.and_then(|attempts| attempts.remove(&attempt));
|
||||
(attempt, pause)
|
||||
};
|
||||
if let Some(pause) = pause {
|
||||
pause.started.notify_one();
|
||||
pause.release.notified().await;
|
||||
}
|
||||
|
||||
let mut state = self.state.lock().await;
|
||||
match precondition {
|
||||
TableCatalogPutPrecondition::IfAbsent if state.objects.contains_key(&key) => {
|
||||
return Err(TableCatalogStoreError::Conflict(format!("object already exists: {object}")));
|
||||
}
|
||||
TableCatalogPutPrecondition::IfMatch(expected) => {
|
||||
let Some(current) = state.objects.get(&key) else {
|
||||
return Err(TableCatalogStoreError::Conflict(format!("object is missing: {object}")));
|
||||
};
|
||||
if current.etag != expected {
|
||||
return Err(TableCatalogStoreError::Conflict(format!("object changed: {object}")));
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
|
||||
let etag = state.next_etag();
|
||||
state.objects.insert(
|
||||
key.clone(),
|
||||
TestCatalogObjectRecord {
|
||||
data,
|
||||
etag,
|
||||
mod_time: Some(OffsetDateTime::now_utc()),
|
||||
},
|
||||
);
|
||||
if state
|
||||
.fail_after_put_attempts
|
||||
.get(&key)
|
||||
.is_some_and(|attempts| attempts.contains(&attempt))
|
||||
{
|
||||
return Err(TableCatalogStoreError::Internal(format!(
|
||||
"injected post-commit put failure for {object} attempt {attempt}"
|
||||
)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn delete_object(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<()> {
|
||||
let mut state = self.state.lock().await;
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let attempt = {
|
||||
let attempts = state.delete_attempts.entry(key.clone()).or_default();
|
||||
*attempts += 1;
|
||||
*attempts
|
||||
};
|
||||
if state
|
||||
.fail_delete_attempts
|
||||
.get(&key)
|
||||
.is_some_and(|attempts| attempts.contains(&attempt))
|
||||
{
|
||||
return Err(TableCatalogStoreError::Internal(format!(
|
||||
"injected delete failure for {object} attempt {attempt}"
|
||||
)));
|
||||
}
|
||||
state.objects.remove(&key);
|
||||
if state
|
||||
.fail_after_delete_attempts
|
||||
.get(&key)
|
||||
.is_some_and(|attempts| attempts.contains(&attempt))
|
||||
{
|
||||
return Err(TableCatalogStoreError::Internal(format!(
|
||||
"injected post-commit delete failure for {object} attempt {attempt}"
|
||||
)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn list_objects(&self, bucket: &str, prefix: &str) -> TableCatalogStoreResult<Vec<String>> {
|
||||
let mut state = self.state.lock().await;
|
||||
state.list_calls += 1;
|
||||
Ok(state
|
||||
.objects
|
||||
.keys()
|
||||
.filter(|(entry_bucket, object)| entry_bucket == bucket && object.starts_with(prefix))
|
||||
.map(|(_, object)| object.clone())
|
||||
.collect())
|
||||
}
|
||||
|
||||
async fn acquire_write_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Box<dyn Send>> {
|
||||
{
|
||||
let mut state = self.state.lock().await;
|
||||
*state
|
||||
.write_lock_acquisitions
|
||||
.entry((bucket.to_string(), object.to_string()))
|
||||
.or_default() += 1;
|
||||
}
|
||||
let lock = {
|
||||
let mut locks = self.locks.lock().await;
|
||||
locks
|
||||
.entry((bucket.to_string(), object.to_string()))
|
||||
.or_insert_with(|| std::sync::Arc::new(tokio::sync::RwLock::new(())))
|
||||
.clone()
|
||||
};
|
||||
Ok(Box::new(lock.write_owned().await))
|
||||
}
|
||||
|
||||
async fn acquire_read_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Box<dyn Send>> {
|
||||
{
|
||||
let mut state = self.state.lock().await;
|
||||
*state
|
||||
.read_lock_acquisitions
|
||||
.entry((bucket.to_string(), object.to_string()))
|
||||
.or_default() += 1;
|
||||
}
|
||||
let lock = {
|
||||
let mut locks = self.locks.lock().await;
|
||||
locks
|
||||
.entry((bucket.to_string(), object.to_string()))
|
||||
.or_insert_with(|| std::sync::Arc::new(tokio::sync::RwLock::new(())))
|
||||
.clone()
|
||||
};
|
||||
Ok(Box::new(lock.read_owned().await))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@ use super::identifier::{
|
||||
default_table_lifecycle_path, default_table_marker_path, default_table_root_prefix, is_valid_table_metadata_file_name,
|
||||
namespace_name_from_marker_path, table_name_from_marker_path, validate_object_mutation,
|
||||
};
|
||||
use super::test_support::{BlockingObjectPublication, TestCatalogObjectBackend, UnserializedTestPublication};
|
||||
use super::*;
|
||||
use datafusion::{
|
||||
arrow::{
|
||||
@@ -1077,312 +1078,6 @@ fn catalog_object_entry_paths_use_internal_root_and_hashed_untrusted_ids() {
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Default)]
|
||||
struct TestCatalogObjectBackend {
|
||||
state: Arc<tokio::sync::Mutex<TestCatalogObjectState>>,
|
||||
locks: TestCatalogObjectLocks,
|
||||
strong_runtime: Option<StrongTableCatalogRuntime>,
|
||||
}
|
||||
|
||||
type TestCatalogObjectLockKey = (String, String);
|
||||
type TestCatalogObjectLock = Arc<tokio::sync::RwLock<()>>;
|
||||
type TestCatalogObjectLocks = Arc<tokio::sync::Mutex<BTreeMap<TestCatalogObjectLockKey, TestCatalogObjectLock>>>;
|
||||
|
||||
#[derive(Clone, Default)]
|
||||
struct TestCatalogObjectPause {
|
||||
started: Arc<tokio::sync::Notify>,
|
||||
release: Arc<tokio::sync::Notify>,
|
||||
}
|
||||
|
||||
impl TestCatalogObjectPause {
|
||||
async fn wait_started(&self) {
|
||||
self.started.notified().await;
|
||||
}
|
||||
|
||||
fn release(&self) {
|
||||
self.release.notify_one();
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct BlockingObjectPublication {
|
||||
backend: TestCatalogObjectBackend,
|
||||
object: String,
|
||||
started: Arc<tokio::sync::Notify>,
|
||||
guard: Arc<parking_lot::Mutex<Option<Box<dyn Send>>>>,
|
||||
}
|
||||
|
||||
impl BlockingObjectPublication {
|
||||
fn new(backend: TestCatalogObjectBackend, object: impl Into<String>) -> Self {
|
||||
Self {
|
||||
backend,
|
||||
object: object.into(),
|
||||
started: Arc::new(tokio::sync::Notify::new()),
|
||||
guard: Arc::new(parking_lot::Mutex::new(None)),
|
||||
}
|
||||
}
|
||||
|
||||
async fn wait_started(&self) {
|
||||
self.started.notified().await;
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct UnserializedTestPublication;
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl TableCommitPublication for UnserializedTestPublication {
|
||||
async fn begin_table_bucket(&self, _table_bucket: &str) -> TableCatalogStoreResult<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn prepare(&self, _table_bucket: &str, _namespace: &str, _table: &str) -> TableCatalogStoreResult<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn holds_table_bucket(&self, _table_bucket: &str) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn holds_table(&self, _table_bucket: &str, _namespace: &str, _table: &str) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn complete(&self) {}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl TableCommitPublication for BlockingObjectPublication {
|
||||
async fn begin_table_bucket(&self, _table_bucket: &str) -> TableCatalogStoreResult<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn prepare(&self, table_bucket: &str, _namespace: &str, _table: &str) -> TableCatalogStoreResult<()> {
|
||||
self.started.notify_one();
|
||||
let guard = self.backend.acquire_read_lock(table_bucket, &self.object).await?;
|
||||
*self.guard.lock() = Some(guard);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn holds_table_bucket(&self, _table_bucket: &str) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn holds_table(&self, _table_bucket: &str, _namespace: &str, _table: &str) -> bool {
|
||||
self.guard.lock().is_some()
|
||||
}
|
||||
|
||||
fn complete(&self) {
|
||||
drop(self.guard.lock().take());
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct TestCatalogObjectState {
|
||||
objects: BTreeMap<(String, String), TestCatalogObjectRecord>,
|
||||
etagless_objects: BTreeSet<(String, String)>,
|
||||
fail_read_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
|
||||
pause_before_read_attempts: BTreeMap<(String, String), BTreeMap<usize, TestCatalogObjectPause>>,
|
||||
pause_read_attempts: BTreeMap<(String, String), BTreeMap<usize, TestCatalogObjectPause>>,
|
||||
read_attempts: BTreeMap<(String, String), usize>,
|
||||
read_limits: Vec<((String, String), usize)>,
|
||||
fail_put_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
|
||||
fail_after_put_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
|
||||
pause_put_attempts: BTreeMap<(String, String), BTreeMap<usize, TestCatalogObjectPause>>,
|
||||
fail_delete_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
|
||||
fail_after_delete_attempts: BTreeMap<(String, String), BTreeSet<usize>>,
|
||||
put_attempts: BTreeMap<(String, String), usize>,
|
||||
delete_attempts: BTreeMap<(String, String), usize>,
|
||||
write_lock_acquisitions: BTreeMap<(String, String), usize>,
|
||||
read_lock_acquisitions: BTreeMap<(String, String), usize>,
|
||||
read_calls: usize,
|
||||
metadata_calls: usize,
|
||||
list_calls: usize,
|
||||
next_etag: u64,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct TestCatalogObjectRecord {
|
||||
data: Vec<u8>,
|
||||
etag: String,
|
||||
mod_time: Option<OffsetDateTime>,
|
||||
}
|
||||
|
||||
impl TestCatalogObjectBackend {
|
||||
async fn seed_object(&self, bucket: &str, object: &str, data: Vec<u8>) {
|
||||
self.seed_object_with_mod_time(bucket, object, data, Some(OffsetDateTime::UNIX_EPOCH))
|
||||
.await;
|
||||
}
|
||||
|
||||
async fn seed_object_with_mod_time(&self, bucket: &str, object: &str, data: Vec<u8>, mod_time: Option<OffsetDateTime>) {
|
||||
let mut state = self.state.lock().await;
|
||||
let etag = state.next_etag();
|
||||
state
|
||||
.objects
|
||||
.insert((bucket.to_string(), object.to_string()), TestCatalogObjectRecord { data, etag, mod_time });
|
||||
}
|
||||
|
||||
async fn fail_put_attempt(&self, bucket: &str, object: &str, attempt: usize) {
|
||||
let mut state = self.state.lock().await;
|
||||
state
|
||||
.fail_put_attempts
|
||||
.entry((bucket.to_string(), object.to_string()))
|
||||
.or_default()
|
||||
.insert(attempt);
|
||||
}
|
||||
|
||||
async fn fail_delete_attempt(&self, bucket: &str, object: &str, attempt: usize) {
|
||||
let mut state = self.state.lock().await;
|
||||
state
|
||||
.fail_delete_attempts
|
||||
.entry((bucket.to_string(), object.to_string()))
|
||||
.or_default()
|
||||
.insert(attempt);
|
||||
}
|
||||
|
||||
async fn list_call_count(&self) -> usize {
|
||||
self.state.lock().await.list_calls
|
||||
}
|
||||
|
||||
async fn read_call_count(&self) -> usize {
|
||||
self.state.lock().await.read_calls
|
||||
}
|
||||
|
||||
async fn metadata_call_count(&self) -> usize {
|
||||
self.state.lock().await.metadata_calls
|
||||
}
|
||||
|
||||
async fn reset_call_counts(&self) {
|
||||
let mut state = self.state.lock().await;
|
||||
state.read_calls = 0;
|
||||
state.metadata_calls = 0;
|
||||
state.list_calls = 0;
|
||||
}
|
||||
|
||||
async fn write_lock_acquisition_count(&self, bucket: &str, object: &str) -> usize {
|
||||
self.state
|
||||
.lock()
|
||||
.await
|
||||
.write_lock_acquisitions
|
||||
.get(&(bucket.to_string(), object.to_string()))
|
||||
.copied()
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
async fn read_lock_acquisition_count(&self, bucket: &str, object: &str) -> usize {
|
||||
self.state
|
||||
.lock()
|
||||
.await
|
||||
.read_lock_acquisitions
|
||||
.get(&(bucket.to_string(), object.to_string()))
|
||||
.copied()
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
async fn fail_next_read(&self, bucket: &str, object: &str) {
|
||||
let mut state = self.state.lock().await;
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let next_attempt = state.read_attempts.get(&key).copied().unwrap_or_default() + 1;
|
||||
state.fail_read_attempts.entry(key).or_default().insert(next_attempt);
|
||||
}
|
||||
|
||||
async fn pause_next_read(&self, bucket: &str, object: &str) -> TestCatalogObjectPause {
|
||||
let mut state = self.state.lock().await;
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let next_attempt = state.read_attempts.get(&key).copied().unwrap_or_default() + 1;
|
||||
let pause = TestCatalogObjectPause::default();
|
||||
state
|
||||
.pause_read_attempts
|
||||
.entry(key)
|
||||
.or_default()
|
||||
.insert(next_attempt, pause.clone());
|
||||
pause
|
||||
}
|
||||
|
||||
async fn pause_before_next_read(&self, bucket: &str, object: &str) -> TestCatalogObjectPause {
|
||||
let mut state = self.state.lock().await;
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let next_attempt = state.read_attempts.get(&key).copied().unwrap_or_default() + 1;
|
||||
let pause = TestCatalogObjectPause::default();
|
||||
state
|
||||
.pause_before_read_attempts
|
||||
.entry(key)
|
||||
.or_default()
|
||||
.insert(next_attempt, pause.clone());
|
||||
pause
|
||||
}
|
||||
|
||||
async fn omit_etag_for_object(&self, bucket: &str, object: &str) {
|
||||
self.state
|
||||
.lock()
|
||||
.await
|
||||
.etagless_objects
|
||||
.insert((bucket.to_string(), object.to_string()));
|
||||
}
|
||||
|
||||
async fn last_read_limit(&self, bucket: &str, object: &str) -> Option<usize> {
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
self.state
|
||||
.lock()
|
||||
.await
|
||||
.read_limits
|
||||
.iter()
|
||||
.rev()
|
||||
.find_map(|(read_key, limit)| (read_key == &key).then_some(*limit))
|
||||
}
|
||||
|
||||
async fn fail_next_put(&self, bucket: &str, object: &str) {
|
||||
let mut state = self.state.lock().await;
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let next_attempt = state.put_attempts.get(&key).copied().unwrap_or_default() + 1;
|
||||
state.fail_put_attempts.entry(key).or_default().insert(next_attempt);
|
||||
}
|
||||
|
||||
async fn fail_after_next_put(&self, bucket: &str, object: &str) {
|
||||
let mut state = self.state.lock().await;
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let next_attempt = state.put_attempts.get(&key).copied().unwrap_or_default() + 1;
|
||||
state.fail_after_put_attempts.entry(key).or_default().insert(next_attempt);
|
||||
}
|
||||
|
||||
async fn fail_after_next_delete(&self, bucket: &str, object: &str) {
|
||||
let mut state = self.state.lock().await;
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let next_attempt = state.delete_attempts.get(&key).copied().unwrap_or_default() + 1;
|
||||
state.fail_after_delete_attempts.entry(key).or_default().insert(next_attempt);
|
||||
}
|
||||
|
||||
async fn pause_next_put(&self, bucket: &str, object: &str) -> TestCatalogObjectPause {
|
||||
let mut state = self.state.lock().await;
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let next_attempt = state.put_attempts.get(&key).copied().unwrap_or_default() + 1;
|
||||
let pause = TestCatalogObjectPause::default();
|
||||
state
|
||||
.pause_put_attempts
|
||||
.entry(key)
|
||||
.or_default()
|
||||
.insert(next_attempt, pause.clone());
|
||||
pause
|
||||
}
|
||||
|
||||
async fn put_attempt_count(&self, bucket: &str, object: &str) -> usize {
|
||||
self.state
|
||||
.lock()
|
||||
.await
|
||||
.put_attempts
|
||||
.get(&(bucket.to_string(), object.to_string()))
|
||||
.copied()
|
||||
.unwrap_or_default()
|
||||
}
|
||||
}
|
||||
|
||||
impl TestCatalogObjectState {
|
||||
fn next_etag(&mut self) -> String {
|
||||
self.next_etag += 1;
|
||||
format!("etag-{}", self.next_etag)
|
||||
}
|
||||
}
|
||||
|
||||
fn maintenance_object_report<'a>(
|
||||
report: &'a TableMetadataMaintenanceReport,
|
||||
metadata_location: &str,
|
||||
@@ -2466,248 +2161,6 @@ fn parquet_i32_values(data: Vec<u8>) -> Vec<i32> {
|
||||
values
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl TableCatalogObjectBackend for TestCatalogObjectBackend {
|
||||
fn strong_catalog_runtime(&self) -> Option<StrongTableCatalogRuntime> {
|
||||
self.strong_runtime.clone()
|
||||
}
|
||||
|
||||
async fn read_object(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Option<TableCatalogObject>> {
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let (attempt, pause_before) = {
|
||||
let mut state = self.state.lock().await;
|
||||
state.read_calls += 1;
|
||||
let attempt = {
|
||||
let attempts = state.read_attempts.entry(key.clone()).or_default();
|
||||
*attempts += 1;
|
||||
*attempts
|
||||
};
|
||||
if state
|
||||
.fail_read_attempts
|
||||
.get(&key)
|
||||
.is_some_and(|attempts| attempts.contains(&attempt))
|
||||
{
|
||||
return Err(TableCatalogStoreError::Internal(format!(
|
||||
"injected read failure for {object} attempt {attempt}"
|
||||
)));
|
||||
}
|
||||
let pause = state
|
||||
.pause_before_read_attempts
|
||||
.get_mut(&key)
|
||||
.and_then(|attempts| attempts.remove(&attempt));
|
||||
(attempt, pause)
|
||||
};
|
||||
if let Some(pause) = pause_before {
|
||||
pause.started.notify_one();
|
||||
pause.release.notified().await;
|
||||
}
|
||||
let (result, pause) = {
|
||||
let mut state = self.state.lock().await;
|
||||
let etagless = state.etagless_objects.contains(&key);
|
||||
let result = state.objects.get(&key).map(|record| TableCatalogObject {
|
||||
data: record.data.clone(),
|
||||
etag: (!etagless).then(|| record.etag.clone()),
|
||||
mod_time: record.mod_time,
|
||||
});
|
||||
let pause = state
|
||||
.pause_read_attempts
|
||||
.get_mut(&key)
|
||||
.and_then(|attempts| attempts.remove(&attempt));
|
||||
(result, pause)
|
||||
};
|
||||
if let Some(pause) = pause {
|
||||
pause.started.notify_one();
|
||||
pause.release.notified().await;
|
||||
}
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
async fn read_object_limited(
|
||||
&self,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
max_size: usize,
|
||||
) -> TableCatalogStoreResult<Option<TableCatalogObject>> {
|
||||
self.state
|
||||
.lock()
|
||||
.await
|
||||
.read_limits
|
||||
.push(((bucket.to_string(), object.to_string()), max_size));
|
||||
let result = self.read_object(bucket, object).await?;
|
||||
if result.as_ref().is_some_and(|object| object.data.len() > max_size) {
|
||||
return Err(TableCatalogStoreError::Invalid(format!(
|
||||
"catalog object {bucket}/{object} exceeds the maximum size of {max_size} bytes"
|
||||
)));
|
||||
}
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
async fn object_metadata(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Option<TableCatalogObjectMetadata>> {
|
||||
let mut state = self.state.lock().await;
|
||||
state.metadata_calls += 1;
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let etagless = state.etagless_objects.contains(&key);
|
||||
Ok(state.objects.get(&key).map(|record| TableCatalogObjectMetadata {
|
||||
etag: (!etagless).then(|| record.etag.clone()),
|
||||
mod_time: record.mod_time,
|
||||
}))
|
||||
}
|
||||
|
||||
async fn object_exists(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<bool> {
|
||||
let state = self.state.lock().await;
|
||||
Ok(state.objects.contains_key(&(bucket.to_string(), object.to_string())))
|
||||
}
|
||||
|
||||
async fn put_object(
|
||||
&self,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
data: Vec<u8>,
|
||||
precondition: TableCatalogPutPrecondition,
|
||||
) -> TableCatalogStoreResult<()> {
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let (attempt, pause) = {
|
||||
let mut state = self.state.lock().await;
|
||||
let attempt = {
|
||||
let attempts = state.put_attempts.entry(key.clone()).or_default();
|
||||
*attempts += 1;
|
||||
*attempts
|
||||
};
|
||||
if state
|
||||
.fail_put_attempts
|
||||
.get(&key)
|
||||
.is_some_and(|attempts| attempts.contains(&attempt))
|
||||
{
|
||||
return Err(TableCatalogStoreError::Internal(format!(
|
||||
"injected put failure for {object} attempt {attempt}"
|
||||
)));
|
||||
}
|
||||
let pause = state
|
||||
.pause_put_attempts
|
||||
.get_mut(&key)
|
||||
.and_then(|attempts| attempts.remove(&attempt));
|
||||
(attempt, pause)
|
||||
};
|
||||
if let Some(pause) = pause {
|
||||
pause.started.notify_one();
|
||||
pause.release.notified().await;
|
||||
}
|
||||
|
||||
let mut state = self.state.lock().await;
|
||||
match precondition {
|
||||
TableCatalogPutPrecondition::IfAbsent if state.objects.contains_key(&key) => {
|
||||
return Err(TableCatalogStoreError::Conflict(format!("object already exists: {object}")));
|
||||
}
|
||||
TableCatalogPutPrecondition::IfMatch(expected) => {
|
||||
let Some(current) = state.objects.get(&key) else {
|
||||
return Err(TableCatalogStoreError::Conflict(format!("object is missing: {object}")));
|
||||
};
|
||||
if current.etag != expected {
|
||||
return Err(TableCatalogStoreError::Conflict(format!("object changed: {object}")));
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
|
||||
let etag = state.next_etag();
|
||||
state.objects.insert(
|
||||
key.clone(),
|
||||
TestCatalogObjectRecord {
|
||||
data,
|
||||
etag,
|
||||
mod_time: Some(OffsetDateTime::now_utc()),
|
||||
},
|
||||
);
|
||||
if state
|
||||
.fail_after_put_attempts
|
||||
.get(&key)
|
||||
.is_some_and(|attempts| attempts.contains(&attempt))
|
||||
{
|
||||
return Err(TableCatalogStoreError::Internal(format!(
|
||||
"injected post-commit put failure for {object} attempt {attempt}"
|
||||
)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn delete_object(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<()> {
|
||||
let mut state = self.state.lock().await;
|
||||
let key = (bucket.to_string(), object.to_string());
|
||||
let attempt = {
|
||||
let attempts = state.delete_attempts.entry(key.clone()).or_default();
|
||||
*attempts += 1;
|
||||
*attempts
|
||||
};
|
||||
if state
|
||||
.fail_delete_attempts
|
||||
.get(&key)
|
||||
.is_some_and(|attempts| attempts.contains(&attempt))
|
||||
{
|
||||
return Err(TableCatalogStoreError::Internal(format!(
|
||||
"injected delete failure for {object} attempt {attempt}"
|
||||
)));
|
||||
}
|
||||
state.objects.remove(&key);
|
||||
if state
|
||||
.fail_after_delete_attempts
|
||||
.get(&key)
|
||||
.is_some_and(|attempts| attempts.contains(&attempt))
|
||||
{
|
||||
return Err(TableCatalogStoreError::Internal(format!(
|
||||
"injected post-commit delete failure for {object} attempt {attempt}"
|
||||
)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn list_objects(&self, bucket: &str, prefix: &str) -> TableCatalogStoreResult<Vec<String>> {
|
||||
let mut state = self.state.lock().await;
|
||||
state.list_calls += 1;
|
||||
Ok(state
|
||||
.objects
|
||||
.keys()
|
||||
.filter(|(entry_bucket, object)| entry_bucket == bucket && object.starts_with(prefix))
|
||||
.map(|(_, object)| object.clone())
|
||||
.collect())
|
||||
}
|
||||
|
||||
async fn acquire_write_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Box<dyn Send>> {
|
||||
{
|
||||
let mut state = self.state.lock().await;
|
||||
*state
|
||||
.write_lock_acquisitions
|
||||
.entry((bucket.to_string(), object.to_string()))
|
||||
.or_default() += 1;
|
||||
}
|
||||
let lock = {
|
||||
let mut locks = self.locks.lock().await;
|
||||
locks
|
||||
.entry((bucket.to_string(), object.to_string()))
|
||||
.or_insert_with(|| std::sync::Arc::new(tokio::sync::RwLock::new(())))
|
||||
.clone()
|
||||
};
|
||||
Ok(Box::new(lock.write_owned().await))
|
||||
}
|
||||
|
||||
async fn acquire_read_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Box<dyn Send>> {
|
||||
{
|
||||
let mut state = self.state.lock().await;
|
||||
*state
|
||||
.read_lock_acquisitions
|
||||
.entry((bucket.to_string(), object.to_string()))
|
||||
.or_default() += 1;
|
||||
}
|
||||
let lock = {
|
||||
let mut locks = self.locks.lock().await;
|
||||
locks
|
||||
.entry((bucket.to_string(), object.to_string()))
|
||||
.or_insert_with(|| std::sync::Arc::new(tokio::sync::RwLock::new(())))
|
||||
.clone()
|
||||
};
|
||||
Ok(Box::new(lock.read_owned().await))
|
||||
}
|
||||
}
|
||||
|
||||
fn test_bucket_entry(bucket: &str) -> TableBucketEntry {
|
||||
TableBucketEntry {
|
||||
version: TABLE_CATALOG_ENTRY_VERSION,
|
||||
|
||||
Reference in New Issue
Block a user