Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions crates/tinybox-drivers/src/box_store.rs
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,28 @@ impl DocumentBoxStore {
migrations::prepare(&storage)
.await
.map_err(|error| store_error("open", &error))?;
Self::on_prepared(&storage)
}

/// Open a store on ports a host manages, reached over the storage bus.
///
/// The host owns the schema and the data migration there: it declared the
/// collections when it bound the module's scope, and its grant does not
/// reach the v1 source collections, so this only confirms the current
/// declarations and migrates nothing.
///
/// # Errors
///
/// Returns [`Error::Store`] when the host refuses a declaration or the
/// bridge cannot be opened.
pub async fn open_host_managed(storage: Storage) -> Result<Self> {
migrations::declare(&storage)
.await
.map_err(|error| store_error("open", &error))?;
Self::on_prepared(&storage)
}

fn on_prepared(storage: &Storage) -> Result<Self> {
let documents: Arc<dyn DocumentStore> = Arc::clone(storage.documents());
let bridge = Blocking::new().map_err(|error| store_error("open", &error))?;
let boxes = Repo::new(&documents, BOX_COLLECTION)
Expand Down
30 changes: 30 additions & 0 deletions crates/tinybox-drivers/src/faulty_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,9 @@ pub(crate) struct Faulty {
source_replacement: Mutex<Option<(String, Value)>>,
target_replacement: Mutex<Option<(String, Value)>>,
receipt_race: std::sync::atomic::AtomicBool,
/// Like a host that owns the schema: declaring any collection outside this
/// list is refused.
grant: Option<Vec<String>>,
}

impl Faulty {
Expand Down Expand Up @@ -69,6 +72,7 @@ impl Faulty {
source_replacement: Mutex::new(None),
target_replacement: Mutex::new(None),
receipt_race: std::sync::atomic::AtomicBool::new(false),
grant: None,
})
}

Expand All @@ -82,6 +86,7 @@ impl Faulty {
source_replacement: Mutex::new(None),
target_replacement: Mutex::new(None),
receipt_race: std::sync::atomic::AtomicBool::new(false),
grant: None,
})
}

Expand All @@ -95,6 +100,7 @@ impl Faulty {
source_replacement: Mutex::new(None),
target_replacement: Mutex::new(None),
receipt_race: std::sync::atomic::AtomicBool::new(false),
grant: None,
})
}

Expand All @@ -112,6 +118,7 @@ impl Faulty {
source_replacement: Mutex::new(Some((source_id, replacement))),
target_replacement: Mutex::new(None),
receipt_race: std::sync::atomic::AtomicBool::new(false),
grant: None,
})
}

Expand All @@ -129,6 +136,23 @@ impl Faulty {
source_replacement: Mutex::new(None),
target_replacement: Mutex::new(Some((target_id, replacement))),
receipt_race: std::sync::atomic::AtomicBool::new(false),
grant: None,
})
}

/// Refuses to declare any collection but `granted`, as a host that owns
/// the module's schema does (`schema-is-host-only`).
pub(crate) fn granting(inner: Arc<dyn DocumentStore>, granted: &[&str]) -> Arc<Self> {
Arc::new(Self {
failing: Failing::new(Arc::clone(&inner)),
inner,
conflicts: AtomicUsize::new(0),
fail_target_readback: false,
target_written: std::sync::atomic::AtomicBool::new(false),
source_replacement: Mutex::new(None),
target_replacement: Mutex::new(None),
receipt_race: std::sync::atomic::AtomicBool::new(false),
grant: Some(granted.iter().map(|name| (*name).to_owned()).collect()),
})
}

Expand All @@ -142,6 +166,7 @@ impl Faulty {
source_replacement: Mutex::new(None),
target_replacement: Mutex::new(None),
receipt_race: std::sync::atomic::AtomicBool::new(true),
grant: None,
})
}
}
Expand All @@ -153,6 +178,11 @@ impl DocumentStore for Faulty {
}

async fn ensure_collection(&self, spec: &CollectionSpec) -> Result<()> {
if let Some(grant) = &self.grant
&& !grant.contains(&spec.name)
{
return Err(StorageError::invalid_input("schema-is-host-only"));
}
self.failing
.call("ensure", self.inner.ensure_collection(spec))
.await
Expand Down
24 changes: 24 additions & 0 deletions crates/tinybox-drivers/src/index_store.rs
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,30 @@ impl DocumentIndexStore {
migrations::prepare(&storage)
.await
.map_err(|error| to_io(&error))?;
Self::on_prepared(&storage, registry)
}

/// Open an index store on ports a host manages, reached over the storage
/// bus: only the current declarations are confirmed, and nothing is
/// migrated (see [`DocumentBoxStore::open_host_managed`]).
///
/// # Errors
///
/// Returns an error when the host refuses a declaration, the bridge cannot
/// start, or `registry` cannot be a document id.
///
/// [`DocumentBoxStore::open_host_managed`]: crate::DocumentBoxStore::open_host_managed
pub async fn open_host_managed(
storage: Storage,
registry: impl Into<String>,
) -> io::Result<Self> {
migrations::declare(&storage)
.await
.map_err(|error| to_io(&error))?;
Self::on_prepared(&storage, registry)
}

fn on_prepared(storage: &Storage, registry: impl Into<String>) -> io::Result<Self> {
let documents: Arc<dyn DocumentStore> = Arc::clone(storage.documents());
let registry = registry.into();
let bridge = Blocking::new().map_err(|e| to_io(&e))?;
Expand Down
21 changes: 18 additions & 3 deletions crates/tinybox-drivers/src/migrations.rs
Original file line number Diff line number Diff line change
Expand Up @@ -25,16 +25,31 @@ struct MigrationReceipt {
complete: bool,
}

/// Ensure `TinyBox`'s current declarations, and nothing else.
///
/// This is all a host-managed port gets ([`DocumentBoxStore::open_host_managed`]
/// and its siblings): the host declared these collections when it bound the
/// module's scope, so the call only confirms host and module agree on them. The
/// host's grant does not include the v1 source collections, and the host's own
/// import owns their migration.
///
/// [`DocumentBoxStore::open_host_managed`]: crate::DocumentBoxStore::open_host_managed
pub(crate) async fn declare(storage: &Storage) -> Result<()> {
let documents = storage.documents();
for spec in collections() {
documents.ensure_collection(&spec).await?;
}
Ok(())
}

/// Ensure `TinyBox`'s declarations and migrate every known legacy row.
///
/// Migration receipts record successful scans. Legacy rows are still scanned
/// on every open so a delayed old writer cannot leave an unreadable record
/// behind a stale completion marker.
pub(crate) async fn prepare(storage: &Storage) -> Result<()> {
declare(storage).await?;
let documents = Arc::clone(storage.documents());
for spec in collections() {
documents.ensure_collection(&spec).await?;
}
documents
.ensure_collection(&CollectionSpec::new(V1_BOX_COLLECTION))
.await?;
Expand Down
56 changes: 56 additions & 0 deletions crates/tinybox-drivers/src/migrations_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -712,3 +712,59 @@ async fn malformed_receipts_and_failed_target_writes_preserve_migration_sources(
);
Ok(())
}

/// The current collection names, as a host that owns the module's schema
/// grants them.
fn granted_collections() -> Vec<String> {
collections().into_iter().map(|spec| spec.name).collect()
}

fn with_host_grant(storage: &Storage) -> Storage {
let granted = granted_collections();
let granted: Vec<&str> = granted.iter().map(String::as_str).collect();
Storage::new(
Faulty::granting(Arc::clone(storage.documents()), &granted),
Arc::clone(storage.streams()),
Arc::clone(storage.kv()),
Arc::clone(storage.blobs()),
Arc::clone(storage.queues()),
Arc::clone(storage.secrets()),
)
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_host_owned_schema_refuses_what_open_declares_but_not_a_host_managed_open() -> TestResult
{
let granted = with_host_grant(&storage("host-grant")?);

assert!(
DocumentBoxStore::open(granted.clone()).await.is_err(),
"open declares the v1 sources, which the host's grant refuses"
);
DocumentBoxStore::open_host_managed(granted.clone()).await?;
DocumentIndexStore::open_host_managed(granted, "registry").await?;
Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_host_managed_open_migrates_nothing() -> TestResult {
let storage = storage("host-managed")?;
seed_legacy(&storage, "legacy", old_box("legacy"), None).await?;

DocumentBoxStore::open_host_managed(storage.clone()).await?;

let documents = storage.documents();
assert!(
documents
.get(MIGRATIONS_COLLECTION, BOXES_STEP)
.await?
.is_none(),
"no receipt is written: the host's import owns the migration"
);
assert!(documents.get(BOX_COLLECTION, "legacy").await?.is_none());
assert!(
documents.get(V1_BOX_COLLECTION, "legacy").await?.is_some(),
"the v1 row is left for the host"
);
Ok(())
}
10 changes: 8 additions & 2 deletions crates/tinybox-module/src/tinybus_module/storage.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,12 @@
//! the host binds the scope from the authenticated caller, so nothing here
//! names one. If the host does not serve storage, durable box operations fail
//! closed; the module never substitutes private in-memory records.
//!
//! These ports are host-managed: the host declared the collections when it
//! bound the scope and its grant is the allowlist, so the stores open with
//! `open_host_managed`, which confirms the current declarations and runs no
//! in-module migration (the v1 collections it reads lie outside the grant; the
//! host's import owns that migration).

use std::sync::Arc;

Expand Down Expand Up @@ -95,7 +101,7 @@ impl ModuleStorage {
if let Some(store) = slot.as_ref() {
return Ok(Arc::clone(store));
}
let opened = tinybox_drivers::DocumentBoxStore::open(storage)
let opened = tinybox_drivers::DocumentBoxStore::open_host_managed(storage)
.await
.inspect_err(|error| {
log::warn!("[tinybox_module] box store on host storage unavailable: {error}");
Expand All @@ -114,7 +120,7 @@ impl ModuleStorage {
name: String,
) -> Option<std::io::Result<tinybox_jail::JailRegistry>> {
let storage = self.scoped().await?;
let index = tinybox_drivers::DocumentIndexStore::open(storage, name).await;
let index = tinybox_drivers::DocumentIndexStore::open_host_managed(storage, name).await;
let opened = tokio::task::spawn_blocking(move || {
let index = index?;
tinybox_jail::JailRegistry::open_with_store(
Expand Down
14 changes: 14 additions & 0 deletions crates/tinybox-module/src/tinybus_module/storage_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -255,3 +255,17 @@ async fn a_recording_view_finds_only_its_own_record() -> TestResult {
assert!(first.first_inserted().is_none());
Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_bus_backed_store_leaves_the_schema_and_migrations_to_the_host() -> TestResult {
let rig = rig(true).await?;
let storage = ModuleStorage::connected(rig.module.clone());
storage.box_store().await?;

let receipts = host_docs(&rig, tinybox_drivers::MIGRATIONS_COLLECTION).await?;
assert!(
receipts.is_empty(),
"the module ran its own migrations on the host's ports: {receipts:?}"
);
Ok(())
}
Loading