mod checkpoint;
mod configuration;
mod index_api;
mod maintenance;
mod mutation;
mod query_api;
mod query_contract;
mod query_engine;
mod resource;
mod validation;
#[cfg(feature = "async")]
mod async_api;
use crate::config::{ConfigBuilder, IoBackend};
use crate::doc::{Doc, DocumentMap};
use crate::error::{Error, Result};
use crate::index::IndexRegistry;
use crate::schema::{AddColumnOption, AlterColumnOption, CollectionSchema, FieldSchema};
use crate::stats::{assess_collection_health, CollectionHealthInput, StatsRegistry, StatsSnapshot};
pub use crate::stats::{CollectionHealth, CollectionHealthStatus, IndexStat};
use crate::storage::StorageHandle;
use crate::storage_ceilings::StorageCeilings;
use checkpoint::{
append_prepared_schema_change, persist_index_cache, publish_prepared_schema_change,
};
pub use configuration::CollectionOptions;
use configuration::{options_config, resolved_storage_ceilings};
pub use maintenance::{
CollectionMaintenanceHealth, CollectionMaintenanceOptions, CollectionMaintenancePhase,
CollectionMaintenanceRuntime,
};
pub use mutation::{DocWriteResult, WriteResult};
use rayon::prelude::*;
pub use resource::CollectionResourceLimits;
use resource::ResourceUsage;
use serde::{Deserialize, Serialize};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
use std::sync::{Arc, Mutex, RwLock};
use validation::{normalize_doc, parse_default_expression, validate_doc};
const MAX_SCHEMA_WORKERS: usize = 256;
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct CollectionStats {
pub doc_count: u64,
pub indexes: Vec<IndexStat>,
pub revision: u64,
#[serde(default)]
pub index_cache_hit: bool,
#[serde(default)]
pub io_backend: IoBackend,
pub read_only: bool,
pub wal_active_seq: u64,
pub wal_checkpoint_seq: u64,
pub wal_ops_since_checkpoint: u64,
pub wal_bytes_since_checkpoint: u64,
#[serde(default)]
pub accounted_document_bytes: u64,
#[serde(default)]
pub estimated_index_bytes: u64,
#[serde(default)]
pub accounted_bytes: u64,
#[serde(default)]
pub resource_limits: CollectionResourceLimits,
#[serde(default)]
pub storage_ceilings: StorageCeilings,
#[serde(default)]
pub resource_limit_rejections: u64,
}
#[derive(Debug, Clone)]
struct CollectionState {
path: PathBuf,
schema: CollectionSchema,
docs: Arc<DocumentMap>,
revision: u64,
options: CollectionOptions,
config: ConfigBuilder,
stats: Arc<StatsRegistry>,
indexes: Arc<IndexRegistry>,
index_cache_hit: bool,
resource_usage: ResourceUsage,
}
#[derive(Debug, Clone)]
struct CollectionSnapshot {
schema: CollectionSchema,
docs: Arc<DocumentMap>,
revision: u64,
stats: Arc<StatsRegistry>,
indexes: Arc<IndexRegistry>,
resource_limits: CollectionResourceLimits,
}
#[derive(Debug)]
struct CollectionInner {
state: RwLock<CollectionState>,
storage: Mutex<StorageHandle>,
writer: Mutex<()>,
closed: AtomicBool,
maintenance_claimed: AtomicBool,
}
#[derive(Clone, Debug)]
pub struct Collection {
inner: Arc<CollectionInner>,
}
impl Collection {
pub fn create_and_open(
path: &str,
schema: &CollectionSchema,
options: Option<&CollectionOptions>,
) -> Result<Self> {
let options = options.cloned().unwrap_or_default();
let config = options_config(&options);
let ceilings = resolved_storage_ceilings(&options);
let root = Path::new(path);
schema.validate()?;
let docs = DocumentMap::new();
let indexes = IndexRegistry::build(schema, &docs, 0)?;
let resource_usage = options
.resource_limits
.enforce_state(schema, &docs, &indexes)?;
let storage = StorageHandle::create(root, schema, options.read_only, ceilings)?;
let state = CollectionState {
path: root.to_path_buf(),
schema: schema.clone(),
docs: Arc::new(docs),
revision: 0,
options,
config,
stats: Arc::new(StatsRegistry::default()),
indexes: Arc::new(indexes),
index_cache_hit: false,
resource_usage,
};
Ok(Self {
inner: Arc::new(CollectionInner {
state: RwLock::new(state),
storage: Mutex::new(storage),
writer: Mutex::new(()),
closed: AtomicBool::new(false),
maintenance_claimed: AtomicBool::new(false),
}),
})
}
pub fn create(
path: &str,
schema: &CollectionSchema,
options: Option<&CollectionOptions>,
) -> Result<Self> {
Self::create_and_open(path, schema, options)
}
pub fn open(path: &str, options: Option<&CollectionOptions>) -> Result<Self> {
let options = options.cloned().unwrap_or_default();
let config = options_config(&options);
let ceilings = resolved_storage_ceilings(&options);
let (storage, schema, docs) =
StorageHandle::open(Path::new(path), options.read_only, ceilings)?;
if schema.name.trim().is_empty() {
return Err(Error::internal("persisted collection has an empty name"));
}
let revision = storage.manifest.revision;
let mut recovered_docs = DocumentMap::new();
for doc in docs {
let doc = normalize_doc(&schema, &doc).map_err(|error| {
Error::internal(format!(
"persisted document cannot be normalized: {}",
error.message
))
})?;
validate_doc(&schema, &doc, true).map_err(|error| {
Error::internal(format!("persisted document is invalid: {}", error.message))
})?;
let id = doc
.get_pk()
.ok_or_else(|| Error::internal("persisted document has no primary key"))?
.to_string();
if recovered_docs.insert(id.clone(), Arc::new(doc)).is_some() {
return Err(Error::internal(format!(
"persisted collection contains duplicate primary key '{id}'"
)));
}
}
let docs = recovered_docs;
let cached_indexes = storage.read_index_cache().ok().flatten().and_then(|bytes| {
let diskann_file = storage.open_diskann_file().ok().flatten();
IndexRegistry::restore_cache(
&bytes,
diskann_file,
config.io_backend,
&schema,
&docs,
revision,
&storage.index_cache_identity(),
storage.ceilings,
)
});
let index_cache_hit = cached_indexes.is_some();
let indexes = cached_indexes.map_or_else(
|| {
IndexRegistry::build(&schema, &docs, revision).map_err(|error| {
Error::internal(format!(
"rebuild persisted indexes at revision {revision}: {}",
error.message
))
})
},
Ok,
)?;
let resource_usage = options
.resource_limits
.enforce_state(&schema, &docs, &indexes)?;
if !index_cache_hit && !options.read_only {
persist_index_cache(&storage, &schema, &indexes, revision, false);
}
let state = CollectionState {
path: PathBuf::from(path),
schema: schema.clone(),
docs: Arc::new(docs),
revision,
options,
config,
stats: Arc::new(StatsRegistry::default()),
indexes: Arc::new(indexes),
index_cache_hit,
resource_usage,
};
Ok(Self {
inner: Arc::new(CollectionInner {
state: RwLock::new(state),
storage: Mutex::new(storage),
writer: Mutex::new(()),
closed: AtomicBool::new(false),
maintenance_claimed: AtomicBool::new(false),
}),
})
}
pub fn path(&self) -> PathBuf {
self.inner
.state
.read()
.map(|state| state.path.clone())
.unwrap_or_default()
}
pub fn is_open(&self) -> bool {
!self.inner.closed.load(AtomicOrdering::Acquire)
}
pub fn flush(&self) -> Result<()> {
self.ensure_open()?;
let _writer = self
.inner
.writer
.lock()
.map_err(|_| Error::internal("writer lock poisoned"))?;
let (schema, docs, indexes, revision) = {
let state = self
.inner
.state
.read()
.map_err(|_| Error::internal("collection state lock poisoned"))?;
(
state.schema.clone(),
Arc::clone(&state.docs),
Arc::clone(&state.indexes),
state.revision,
)
};
let mut storage = self
.inner
.storage
.lock()
.map_err(|_| Error::internal("storage lock poisoned"))?;
storage.checkpoint(&schema, docs.as_ref(), revision, true)?;
persist_index_cache(&storage, &schema, &indexes, revision, true);
Ok(())
}
pub fn close(self) -> Result<()> {
if self.is_open() {
let read_only = self
.inner
.state
.read()
.map_err(|_| Error::internal("collection state lock poisoned"))?
.options
.read_only;
if !read_only {
self.flush()?;
}
self.inner.closed.store(true, AtomicOrdering::Release);
}
Ok(())
}
pub fn destroy(self) -> Result<()> {
let path = self.path();
self.close()?;
if path.exists() {
std::fs::remove_dir_all(&path)
.map_err(|e| Error::internal(format!("destroy collection: {e}")))?;
}
Ok(())
}
pub fn schema(&self) -> Result<CollectionSchema> {
self.ensure_open()?;
self.inner
.state
.read()
.map(|state| state.schema.clone())
.map_err(|_| Error::internal("collection state lock poisoned"))
}
pub fn stats(&self) -> Result<CollectionStats> {
self.ensure_open()?;
self.collect_stats().map(|(stats, _)| stats)
}
pub fn health(&self) -> Result<CollectionHealth> {
let (stats, storage_revision) = self.collect_stats()?;
Ok(assess_collection_health(CollectionHealthInput {
is_open: self.is_open(),
revision: stats.revision,
storage_revision,
doc_count: stats.doc_count,
indexes: &stats.indexes,
read_only: stats.read_only,
wal_ops_since_checkpoint: stats.wal_ops_since_checkpoint,
wal_bytes_since_checkpoint: stats.wal_bytes_since_checkpoint,
maintenance_active: self.inner.maintenance_claimed.load(AtomicOrdering::Acquire),
}))
}
fn collect_stats(&self) -> Result<(CollectionStats, u64)> {
let state = self
.inner
.state
.read()
.map_err(|_| Error::internal("collection state lock poisoned"))?;
let storage = self
.inner
.storage
.lock()
.map_err(|_| Error::internal("storage lock poisoned"))?;
let mut indexes = state
.indexes
.stats(&state.schema, &state.docs, state.revision);
indexes.sort_by(|left, right| left.name.cmp(&right.name));
let usage = state.resource_usage;
Ok((
CollectionStats {
doc_count: state.docs.len() as u64,
indexes,
revision: state.revision,
index_cache_hit: state.index_cache_hit,
io_backend: state.config.io_backend,
read_only: state.options.read_only,
wal_active_seq: storage.manifest.wal_active_seq,
wal_checkpoint_seq: storage.manifest.wal_checkpoint_seq,
wal_ops_since_checkpoint: storage.manifest.wal_ops_since_checkpoint,
wal_bytes_since_checkpoint: storage.manifest.wal_bytes_since_checkpoint,
accounted_document_bytes: usage.documents,
estimated_index_bytes: usage.indexes,
accounted_bytes: usage.total,
resource_limits: state.options.resource_limits,
storage_ceilings: storage.ceilings,
resource_limit_rejections: state
.stats
.resource_limit_rejections
.load(AtomicOrdering::Relaxed),
},
storage.manifest.revision,
))
}
pub fn stats_snapshot(&self) -> Result<StatsSnapshot> {
let basic = self.stats()?;
let state = self
.inner
.state
.read()
.map_err(|_| Error::internal("collection state lock poisoned"))?;
let registry = Arc::clone(&state.stats);
Ok(StatsSnapshot {
collection_name: state.schema.name.clone(),
revision: basic.revision,
doc_count: basic.doc_count,
query_count: registry.query_count.load(AtomicOrdering::Relaxed),
fts_query_count: registry.fts_query_count.load(AtomicOrdering::Relaxed),
fts_index_query_count: registry.fts_index_query_count.load(AtomicOrdering::Relaxed),
ann_query_count: registry.ann_query_count.load(AtomicOrdering::Relaxed),
diskann_query_count: registry.diskann_query_count.load(AtomicOrdering::Relaxed),
diskann_mmap_query_count: registry
.diskann_mmap_query_count
.load(AtomicOrdering::Relaxed),
diskann_sector_read_count: registry
.diskann_sector_read_count
.load(AtomicOrdering::Relaxed),
exact_query_count: registry.exact_query_count.load(AtomicOrdering::Relaxed),
filtered_query_count: registry.filtered_query_count.load(AtomicOrdering::Relaxed),
scalar_index_query_count: registry
.scalar_index_query_count
.load(AtomicOrdering::Relaxed),
radius_query_count: registry.radius_query_count.load(AtomicOrdering::Relaxed),
candidates_scanned: registry.candidates_scanned.load(AtomicOrdering::Relaxed),
indexed_field_count: basic.indexes.len(),
indexes: basic.indexes,
index_cache_hit: basic.index_cache_hit,
io_backend: basic.io_backend,
read_only: basic.read_only,
wal_active_seq: basic.wal_active_seq,
wal_checkpoint_seq: basic.wal_checkpoint_seq,
wal_ops_since_checkpoint: basic.wal_ops_since_checkpoint,
wal_bytes_since_checkpoint: basic.wal_bytes_since_checkpoint,
accounted_document_bytes: basic.accounted_document_bytes,
estimated_index_bytes: basic.estimated_index_bytes,
accounted_bytes: basic.accounted_bytes,
resource_limits: basic.resource_limits,
resource_limit_rejections: basic.resource_limit_rejections,
})
}
pub fn count(&self) -> Result<usize> {
self.ensure_open()?;
self.inner
.state
.read()
.map(|state| state.docs.len())
.map_err(|_| Error::internal("collection state lock poisoned"))
}
pub fn add_column(&self, field_schema: &FieldSchema, default_expr: Option<&str>) -> Result<()> {
self.add_column_with_options(field_schema, default_expr, AddColumnOption::default())
}
pub fn add_column_with_options(
&self,
field_schema: &FieldSchema,
default_expr: Option<&str>,
option: AddColumnOption,
) -> Result<()> {
self.ensure_open()?;
let _writer = self
.inner
.writer
.lock()
.map_err(|_| Error::internal("writer lock poisoned"))?;
let state = self
.inner
.state
.write()
.map_err(|_| Error::internal("collection state lock poisoned"))?;
ensure_writable(&state.options)?;
let mut next = state.clone();
next.schema.add_field(field_schema)?;
let default = default_expr
.map(|expression| parse_default_expression(expression, field_schema.data_type))
.transpose()?;
if let Some(value) = default {
next.docs = Arc::new(transform_documents_with_concurrency(
&next.docs,
option.concurrency,
|doc| doc.set_field_value(&field_schema.name, value.clone()),
)?);
}
validate_documents_with_concurrency(&next.schema, &next.docs, option.concurrency)?;
let config = state.config.clone();
let previous_docs = Arc::clone(&state.docs);
let previous_revision = state.revision;
let previous_schema = state.schema.clone();
drop(state);
finish_schema_commit(
self,
&previous_docs,
previous_revision,
&previous_schema,
next,
&config,
)
}
pub fn drop_column(&self, name: &str) -> Result<()> {
self.ensure_open()?;
let _writer = self
.inner
.writer
.lock()
.map_err(|_| Error::internal("writer lock poisoned"))?;
let state = self
.inner
.state
.write()
.map_err(|_| Error::internal("collection state lock poisoned"))?;
ensure_writable(&state.options)?;
let mut next = state.clone();
next.schema.drop_field(name)?;
next.docs = Arc::new(transform_documents(&next.docs, |doc| {
doc.remove_field(name)
})?);
let config = state.config.clone();
let previous_docs = Arc::clone(&state.docs);
let previous_revision = state.revision;
let previous_schema = state.schema.clone();
drop(state);
finish_schema_commit(
self,
&previous_docs,
previous_revision,
&previous_schema,
next,
&config,
)
}
pub fn rename_column(&self, old_name: &str, new_name: &str) -> Result<()> {
if old_name.trim().is_empty() || old_name.contains('\0') {
return Err(Error::invalid_argument("old field name is invalid"));
}
if new_name.trim().is_empty() || new_name.contains('\0') {
return Err(Error::invalid_argument("new field name is invalid"));
}
if old_name == new_name {
return Ok(());
}
self.ensure_open()?;
let _writer = self
.inner
.writer
.lock()
.map_err(|_| Error::internal("writer lock poisoned"))?;
let state = self
.inner
.state
.write()
.map_err(|_| Error::internal("collection state lock poisoned"))?;
ensure_writable(&state.options)?;
let mut next = state.clone();
if next.schema.has_field(new_name) {
return Err(Error::already_exists(format!(
"field '{new_name}' already exists"
)));
}
if let Some(field) = next
.schema
.fields
.iter_mut()
.find(|field| field.name == old_name)
{
field.name = new_name.to_string();
} else if let Some(field) = next
.schema
.vectors
.iter_mut()
.find(|field| field.name == old_name)
{
field.name = new_name.to_string();
} else {
return Err(Error::not_found(format!("field '{old_name}' not found")));
}
next.docs = Arc::new(transform_documents(&next.docs, |doc| {
if let Some(value) = doc.field(old_name).cloned() {
doc.remove_field(old_name)?;
doc.set_field_value(new_name, value)?;
} else if let Some(value) = doc.vector(old_name).cloned() {
doc.remove_field(old_name)?;
doc.set_vector_value(new_name, value)?;
}
Ok(())
})?);
let config = state.config.clone();
let previous_docs = Arc::clone(&state.docs);
let previous_revision = state.revision;
let previous_schema = state.schema.clone();
drop(state);
finish_schema_commit(
self,
&previous_docs,
previous_revision,
&previous_schema,
next,
&config,
)
}
pub fn alter_column(
&self,
field_schema: &FieldSchema,
option: AlterColumnOption,
) -> Result<()> {
self.ensure_open()?;
let _writer = self
.inner
.writer
.lock()
.map_err(|_| Error::internal("writer lock poisoned"))?;
let state = self
.inner
.state
.write()
.map_err(|_| Error::internal("collection state lock poisoned"))?;
ensure_writable(&state.options)?;
let mut next = state.clone();
let target = next
.schema
.fields
.iter_mut()
.find(|field| field.name == field_schema.name)
.ok_or_else(|| Error::not_found(format!("field '{}' not found", field_schema.name)))?;
if target.data_type != field_schema.data_type || target.dimension != field_schema.dimension
{
return Err(Error::invalid_argument(
"altering a field's data type or dimension would invalidate existing data",
));
}
*target = field_schema.clone();
next.schema.validate()?;
validate_documents_with_concurrency(&next.schema, &next.docs, option.concurrency)?;
let config = state.config.clone();
let previous_docs = Arc::clone(&state.docs);
let previous_revision = state.revision;
let previous_schema = state.schema.clone();
drop(state);
finish_schema_commit(
self,
&previous_docs,
previous_revision,
&previous_schema,
next,
&config,
)
}
fn snapshot_state(&self) -> Result<CollectionSnapshot> {
let state = self
.inner
.state
.read()
.map_err(|_| Error::internal("collection state lock poisoned"))?;
Ok(CollectionSnapshot {
schema: state.schema.clone(),
docs: state.docs.clone(),
revision: state.revision,
stats: Arc::clone(&state.stats),
indexes: state.indexes.clone(),
resource_limits: state.options.resource_limits,
})
}
fn ensure_open(&self) -> Result<()> {
if self.inner.closed.load(AtomicOrdering::Acquire) {
Err(Error::failed_precondition("collection is closed"))
} else {
Ok(())
}
}
#[cfg(test)]
pub(crate) fn test_arm_wal_sync_stall(&self) -> crate::storage::StallGate {
let storage = self.inner.storage.lock().expect("storage lock poisoned");
storage.arm_wal_sync_stall()
}
#[cfg(test)]
pub(crate) fn test_arm_diskann_write_fault(&self) {
let storage = self.inner.storage.lock().expect("storage lock poisoned");
storage.arm_diskann_write_fault();
}
#[cfg(test)]
pub(crate) fn test_diskann_write_fault_fired(&self) -> bool {
let storage = self.inner.storage.lock().expect("storage lock poisoned");
storage.diskann_write_fault_fired()
}
}
fn ensure_writable(options: &CollectionOptions) -> Result<()> {
if options.read_only {
Err(Error::permission_denied("collection is read-only"))
} else {
Ok(())
}
}
fn ensure_same_generation(current: &CollectionState, expected: &CollectionState) -> Result<()> {
if current.revision == expected.revision && current.schema == expected.schema {
Ok(())
} else {
Err(Error::failed_precondition(
"collection generation changed during index construction",
))
}
}
fn transform_documents(
docs: &DocumentMap,
transform: impl Fn(&mut Doc) -> Result<()> + Send + Sync,
) -> Result<DocumentMap> {
transform_documents_with_concurrency(docs, 0, transform)
}
fn transform_documents_with_concurrency(
docs: &DocumentMap,
concurrency: u32,
transform: impl Fn(&mut Doc) -> Result<()> + Send + Sync,
) -> Result<DocumentMap> {
let entries: Vec<(String, Arc<Doc>)> = docs
.iter()
.map(|(id, doc)| (id.clone(), Arc::clone(doc)))
.collect();
let transform_one = |(id, doc): &(String, Arc<Doc>)| {
let mut next = doc.as_ref().clone();
let result = transform(&mut next).map(|()| next);
(id.clone(), result)
};
let transformed: Vec<(String, Result<Doc>)> =
if let Some(threads) = schema_worker_count(concurrency, entries.len())? {
let pool = rayon::ThreadPoolBuilder::new()
.num_threads(threads)
.build()
.map_err(|error| {
Error::resource_exhausted(format!("build schema worker pool: {error}"))
})?;
pool.install(|| entries.par_iter().map(transform_one).collect())
} else {
entries.iter().map(transform_one).collect()
};
let mut output = DocumentMap::new();
for (id, result) in transformed {
output.insert(id, Arc::new(result?));
}
Ok(output)
}
fn validate_documents_with_concurrency(
schema: &CollectionSchema,
docs: &DocumentMap,
concurrency: u32,
) -> Result<()> {
let entries: Vec<Arc<Doc>> = docs.values().cloned().collect();
let validate_one = |doc: &Arc<Doc>| validate_doc(schema, doc, true);
let Some(threads) = schema_worker_count(concurrency, entries.len())? else {
for doc in &entries {
validate_one(doc)?;
}
return Ok(());
};
let pool = rayon::ThreadPoolBuilder::new()
.num_threads(threads)
.build()
.map_err(|error| Error::resource_exhausted(format!("build schema worker pool: {error}")))?;
let results: Vec<Result<()>> = pool.install(|| entries.par_iter().map(validate_one).collect());
for result in results {
result?;
}
Ok(())
}
fn schema_worker_count(concurrency: u32, work_items: usize) -> Result<Option<usize>> {
if concurrency == 0 || work_items < 2 {
return Ok(None);
}
let requested = usize::try_from(concurrency)
.map_err(|_| Error::resource_exhausted("schema concurrency exceeds this platform"))?;
let available = std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get);
let threads = requested
.min(work_items)
.min(available)
.min(MAX_SCHEMA_WORKERS);
Ok((threads > 1).then_some(threads))
}
fn finish_schema_commit(
collection: &Collection,
previous_docs: &Arc<DocumentMap>,
previous_revision: u64,
previous_schema: &CollectionSchema,
next: CollectionState,
config: &ConfigBuilder,
) -> Result<()> {
let next = prepare_schema_change(next)?;
{
let mut storage = collection
.inner
.storage
.lock()
.map_err(|_| Error::internal("storage lock poisoned"))?;
append_prepared_schema_change(
&mut storage,
previous_docs,
previous_revision,
&next,
config,
)?;
}
let mut state = collection
.inner
.state
.write()
.map_err(|_| Error::internal("collection state lock poisoned"))?;
if state.revision != previous_revision || state.schema != *previous_schema {
return Err(Error::failed_precondition(
"collection generation changed during index construction",
));
}
let mut storage = collection
.inner
.storage
.lock()
.map_err(|_| Error::internal("storage lock poisoned"))?;
publish_prepared_schema_change(&mut storage, &mut state, next, config)
}
fn prepare_schema_change(mut next: CollectionState) -> Result<CollectionState> {
let revision = next_revision(next.revision)?;
next.revision = revision;
next.indexes = Arc::new(IndexRegistry::build(&next.schema, &next.docs, revision)?);
next.resource_usage =
match next
.options
.resource_limits
.enforce_state(&next.schema, &next.docs, &next.indexes)
{
Ok(usage) => usage,
Err(error) => {
next.stats.record_resource_limit_rejection();
return Err(error);
}
};
Ok(next)
}
fn next_revision(current: u64) -> Result<u64> {
current
.checked_add(1)
.ok_or_else(|| Error::resource_exhausted("collection revision overflow"))
}
#[cfg(test)]
mod ga_contract;
#[cfg(test)]
mod tests;