feat: bind FeatureStorage identity and restore to persistent owners

This commit is contained in:
2026-11-04 12:11:35 +09:00
parent d9295f4439
commit dfa04107ab
24 changed files with 5165 additions and 208 deletions
+51 -7
View File
@@ -1,11 +1,55 @@
# feature-storage # feature-storage
Manages the SQLite lifecycle for Features within a trusted Host. Owns physical SQLite identity and lifecycle for trusted Host repositories.
Does not own Feature domain schemas or authentication, or provide SQL access to Workers or models. Features own domain schema, authorization and transactions; storage identity is not a grant.
Workers and models do not receive SQL connections.
## Main entry points [`ScopedFeatureStorage::provision`] is fresh-only;
[`ScopedFeatureStorage::adopt_existing`] requires explicit Host authority and
persisted domain attribution. [`ScopedFeatureStorage::open_existing`] consumes
an expected Workspace/Feature/physical identity and never creates a missing DB.
The lazy [`ScopedFeatureStorage::open`] requests initial Host provisioning only
when no ownership record exists; an unbound existing file requires adoption,
and a missing previously bound file is an error, not a new incarnation.
- [`FeatureStorage`], [`ScopedFeatureStorage`]: registration, scope isolation, storage, and shutdown [`FeatureDatabase`] operations retain the verified connection and shared
- [`FeatureDatabase`]: connections and transactions cross-process lifecycle lease through consumer commit/rollback.
- [`FeatureMigration`]: Feature-owned schema changes [`DatabaseGuard`] extends the short lease over a writer-release decision without
- [`configure_connection`]: shared SQLite connection policy retaining a SQLite transaction. Long Job waits retain only [`DatabaseIdentity`];
resumption must reacquire storage and current domain authorization.
Lock order is scope lifecycle file, manager state, connection, SQLite transaction.
The Host must protect the root, owner catalog and stable lockfiles against native
code bypassing the owner. A successful same-root lock does not prove arbitrary
external writers, a previous Host, or a different filesystem participate.
Same-path replacement is checked by physical file identity as well as DB identity;
old receipt bytes alone cannot authorize a copied file. Restore records a new
physical lifecycle and retains original provenance. A durable erasure fence lives
outside the removed scope and cannot be cleared by restoring an older snapshot.
`restore` recovers missing scope files using the same owner's retained source
receipts, never by calling `delete` first. `restore_from` requires the Host to
select the source owner explicitly for a cross-root copy. Both check completed
snapshot provenance and the source erasure fence through publication; a fresh
root is not a way to forget that fence. Cross-root locks are acquired in canonical
root order, source shared and target exclusive; same-root recovery takes one
exclusive lock. A restore retires, but does not remove, the previous owner
receipt. Only the new physical identity is current, including after exact retry.
A snapshot without matching source-owner evidence requires explicit, authorized
legacy adoption with domain attribution/shape validation instead of blind restore.
These checks cover owner-managed operations, not deliberate destruction of the
owner catalog/fences or arbitrary Host code inventing authority evidence.
Adoption identity and domain migrations share a SQLite transaction. A separate
owner receipt binds the stable request, target, authority evidence and inode.
If the DB commits but completion or its response is lost, the same request
reconciles the same physical DB. An unfinished reservation is not an adoption
receipt; a different target or request cannot reuse it. Failure after a fresh file
is created but before its reservation commits leaves an unowned file requiring
explicit Host reconciliation, never discovery-based issuance.
[`FeatureRegistration`] uses the existing ordered migration owner; a Feature can
validate its finite published receipt/shape families without rewriting original
migration identities or treating physical acceptance as domain-policy adoption.
[`FeatureStorage::open_host_catalog`] connects the Host catalog to the same
identity/lifecycle owner; [`configure_connection`] owns shared SQLite durability policy.
@@ -0,0 +1,993 @@
//! Component tests of FeatureStorage + owner receipts + real SQLite. The only
//! injected failures are returned validator errors or abandoned real transactions;
//! these do not claim arbitrary crash/power-loss durability. Run with the normal
//! feature-storage lib target. Unix child tests use this same test executable,
//! isolated roots, explicit child-only environment, and observed lock contention.
use super::*;
use std::cell::Cell;
use std::process::{Child, Command, Stdio};
use std::time::Instant;
const SCOPE: &str = "identity-workspace";
const FEATURE: &str = "identity-feature";
const REQUEST: &str = "host-adoption-request";
const EVIDENCE: &str = "Host selected this legacy Workspace and Feature";
const WAIT: Duration = Duration::from_secs(15);
fn create_records(tx: &Transaction<'_>) -> Result<()> {
tx.execute_batch("CREATE TABLE IF NOT EXISTS records (value TEXT NOT NULL UNIQUE);")?;
Ok(())
}
const PLAN: &[FeatureMigration] = &[FeatureMigration::new(1, "records", create_records)];
fn registered(root: &Path) -> (FeatureStorage, RegisteredFeature) {
let manager = FeatureStorage::new(root);
let registration = manager
.register(FeatureRegistration::new(FEATURE, PLAN))
.unwrap();
(manager, registration)
}
fn database_path(root: &Path, scope: &str) -> PathBuf {
root.join(scope)
.join("features")
.join(format!("{FEATURE}.sqlite"))
}
fn legacy(root: &Path, scope: &str) -> PathBuf {
let path = database_path(root, scope);
fs::create_dir_all(path.parent().unwrap()).unwrap();
let connection = Connection::open(&path).unwrap();
connection
.execute_batch(
"CREATE TABLE records (value TEXT NOT NULL UNIQUE);
INSERT INTO records VALUES ('preserved');
CREATE TABLE attribution (scope TEXT NOT NULL, feature TEXT NOT NULL);",
)
.unwrap();
connection
.execute(
"INSERT INTO attribution VALUES (?1, ?2)",
params![scope, FEATURE],
)
.unwrap();
path
}
fn validate_legacy(tx: &Transaction<'_>) -> Result<()> {
let attribution: (String, String) =
tx.query_row("SELECT scope, feature FROM attribution", [], |r| {
Ok((r.get(0)?, r.get(1)?))
})?;
if attribution != (SCOPE.into(), FEATURE.into()) {
return Err(FeatureStorageError::operation("legacy attribution differs"));
}
// Exercise the actual supported shape, not just a validator invocation flag.
let _: String = tx.query_row("SELECT value FROM records", [], |r| r.get(0))?;
Ok(())
}
fn values(database: &FeatureDatabase) -> Vec<String> {
database
.with_connection(|connection| {
Ok(connection
.prepare("SELECT value FROM records ORDER BY value")?
.query_map([], |r| r.get(0))?
.collect::<std::result::Result<Vec<_>, _>>()?)
})
.unwrap()
}
fn insert(database: &FeatureDatabase, value: &str) {
database
.transaction(|tx| {
tx.execute("INSERT INTO records VALUES (?1)", [value])?;
Ok(())
})
.unwrap();
}
fn assert_identity_error<T>(result: Result<T>) {
match result {
Err(FeatureStorageError::Identity(_)) => {}
Err(error) => panic!("expected identity refusal, got {error:?}"),
Ok(_) => panic!("expected identity refusal, operation succeeded"),
}
}
fn has_identity(connection: &Connection) -> bool {
connection
.query_row(
"SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE name='__yoi_storage_identity')",
[],
|r| r.get(0),
)
.unwrap()
}
#[test]
fn provision_then_reopen_retains_physical_identity_and_committed_data() {
let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("storage");
let (manager, registration) = registered(&root);
assert!(!root.exists(), "registration must remain lazy");
let scoped = manager.scope(SCOPE).unwrap();
let database = scoped.provision(&registration, REQUEST, EVIDENCE).unwrap();
let identity = database.identity().clone();
insert(&database, "acknowledged");
assert_eq!(identity.scope, SCOPE);
assert_eq!(identity.feature, FEATURE);
assert!(!identity.physical.is_empty());
drop(database);
drop(scoped);
drop(manager);
let (manager, registration) = registered(&root);
let database = manager
.scope(SCOPE)
.unwrap()
.open_existing(&registration, &identity)
.unwrap();
assert_eq!(database.identity(), &identity);
assert_eq!(values(&database), ["acknowledged"]);
let scoped = manager.scope(SCOPE).unwrap();
let replay = scoped.provision(&registration, REQUEST, EVIDENCE).unwrap();
assert_eq!(replay.identity(), &identity);
assert_eq!(values(&replay), ["acknowledged"]);
assert_identity_error(scoped.provision(&registration, "different-request", EVIDENCE));
assert_identity_error(manager.scope("another-workspace").unwrap().provision(
&registration,
REQUEST,
EVIDENCE,
));
assert!(!database_path(&root, "another-workspace").exists());
let owner = Connection::open(root.join(".owner.sqlite")).unwrap();
let receipt: (String, i64) = owner
.query_row(
"SELECT physical, completed FROM storage_requests WHERE scope=?1 AND feature=?2",
params![SCOPE, FEATURE],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.unwrap();
assert_eq!(receipt, (identity.physical, 1));
}
#[test]
fn unowned_legacy_requires_explicit_evidence_and_real_domain_validation() {
let temp = tempfile::tempdir().unwrap();
let path = legacy(temp.path(), SCOPE);
let (manager, registration) = registered(temp.path());
let scoped = manager.scope(SCOPE).unwrap();
assert!(
scoped.open(&registration).is_err(),
"existence is not adoption permission"
);
assert!(
scoped
.provision(&registration, "new-provision-request", EVIDENCE)
.is_err()
);
assert!(!has_identity(&Connection::open(&path).unwrap()));
assert_identity_error(scoped.adopt_existing(&registration, REQUEST, " ", validate_legacy));
assert!(!has_identity(&Connection::open(&path).unwrap()));
let database = scoped
.adopt_existing(&registration, REQUEST, EVIDENCE, validate_legacy)
.unwrap();
let identity = database.identity().clone();
assert_eq!(values(&database), ["preserved"]);
drop(database);
drop(scoped);
drop(manager);
let (manager, registration) = registered(temp.path());
let reopened = manager
.scope(SCOPE)
.unwrap()
.open_existing(&registration, &identity)
.unwrap();
assert_eq!(values(&reopened), ["preserved"]);
}
#[test]
fn rejected_legacy_attribution_or_shape_rolls_back_without_completing_ownership() {
for corrupt in [
"UPDATE attribution SET scope='another-workspace'",
"ALTER TABLE records RENAME COLUMN value TO unsupported",
] {
let temp = tempfile::tempdir().unwrap();
let path = legacy(temp.path(), SCOPE);
let raw = Connection::open(&path).unwrap();
raw.execute_batch(corrupt).unwrap();
let (manager, registration) = registered(temp.path());
assert!(
manager
.scope(SCOPE)
.unwrap()
.adopt_existing(&registration, REQUEST, EVIDENCE, |tx| {
tx.execute_batch("CREATE TABLE must_rollback (value TEXT);")?;
validate_legacy(tx)
},)
.is_err()
);
assert!(!has_identity(&raw));
let leaked: bool = raw.query_row(
"SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE name IN ('must_rollback', '__yoi_feature_schema_migrations'))",
[], |r| r.get(0),
).unwrap();
assert!(
!leaked,
"failed validator must roll back its own writes and admission"
);
let catalog = Connection::open(temp.path().join(".owner.sqlite")).unwrap();
let completed: i64 = catalog
.query_row(
"SELECT completed FROM storage_requests WHERE request=?1",
[REQUEST],
|r| r.get(0),
)
.unwrap();
assert_eq!(completed, 0);
}
}
#[test]
fn adoption_response_loss_replays_same_identity_but_refuses_changed_request_or_target() {
let temp = tempfile::tempdir().unwrap();
legacy(temp.path(), SCOPE);
let (manager, registration) = registered(temp.path());
let database = manager
.scope(SCOPE)
.unwrap()
.adopt_existing(&registration, REQUEST, EVIDENCE, validate_legacy)
.unwrap();
let expected = database.identity().clone();
insert(&database, "after-admission");
// Lose the returned handle/manager, not the committed ownership receipt.
drop(database);
drop(manager);
let (manager, registration) = registered(temp.path());
let scoped = manager.scope(SCOPE).unwrap();
let replay = scoped
.adopt_existing(&registration, REQUEST, EVIDENCE, validate_legacy)
.unwrap();
assert_eq!(replay.identity(), &expected);
assert_eq!(values(&replay), ["after-admission", "preserved"]);
assert_identity_error(scoped.adopt_existing(
&registration,
"different-request",
EVIDENCE,
validate_legacy,
));
assert_identity_error(scoped.adopt_existing(
&registration,
REQUEST,
"changed-authority",
validate_legacy,
));
legacy(temp.path(), "another-workspace");
assert_identity_error(manager.scope("another-workspace").unwrap().adopt_existing(
&registration,
REQUEST,
EVIDENCE,
|_| panic!("changed target must be refused before domain validation"),
));
let owner = Connection::open(temp.path().join(".owner.sqlite")).unwrap();
let count: i64 = owner
.query_row("SELECT COUNT(*) FROM storage_requests", [], |r| r.get(0))
.unwrap();
assert_eq!(count, 1, "replay must not manufacture another identity");
}
#[test]
fn reserved_adoption_reconciles_at_each_real_identity_commit_boundary() {
// 0: before identity writes, 1: identity written but rolled back,
// 2: identity committed but owner completion missing, 3: completion response lost.
for boundary in 0..4 {
let temp = tempfile::tempdir().unwrap();
let path = legacy(temp.path(), SCOPE);
let lifecycle = owner::Lifecycle {
root: temp.path().into(),
scope: SCOPE.into(),
};
let expected = {
let _lease = lifecycle.lock(true).unwrap();
let binding = lifecycle
.reserve(SCOPE, FEATURE, &path, REQUEST, "adopt", EVIDENCE)
.unwrap();
let mut connection =
Connection::open_with_flags(&path, OpenFlags::SQLITE_OPEN_READ_WRITE).unwrap();
configure_connection(&connection).unwrap();
let tx = connection
.transaction_with_behavior(TransactionBehavior::Immediate)
.unwrap();
validate_legacy(&tx).unwrap();
migrate(&tx, FEATURE, FeatureRegistration::new(FEATURE, PLAN)).unwrap();
tx.execute("INSERT INTO records VALUES ('boundary-write')", [])
.unwrap();
if boundary >= 1 {
binding.persist(&tx).unwrap();
}
if boundary >= 2 {
tx.commit().unwrap();
} else {
tx.rollback().unwrap();
}
if boundary == 3 {
lifecycle.complete(&binding).unwrap();
}
binding.identity
};
let raw = Connection::open(&path).unwrap();
assert_eq!(has_identity(&raw), boundary >= 2);
let count: i64 = raw
.query_row("SELECT COUNT(*) FROM records", [], |r| r.get(0))
.unwrap();
assert_eq!(count, if boundary >= 2 { 2 } else { 1 });
drop(raw);
let (manager, registration) = registered(temp.path());
let scoped = manager.scope(SCOPE).unwrap();
if boundary < 3 {
assert_identity_error(scoped.open_existing(&registration, &expected));
}
let database = scoped
.adopt_existing(&registration, REQUEST, EVIDENCE, validate_legacy)
.unwrap();
assert_eq!(
database.identity(),
&expected,
"boundary {boundary} changed reserved identity"
);
assert_eq!(values(&database).len(), count as usize);
drop(database);
drop(scoped);
drop(manager);
let (manager, registration) = registered(temp.path());
let database = manager
.scope(SCOPE)
.unwrap()
.open_existing(&registration, &expected)
.unwrap();
assert_eq!(values(&database).len(), count as usize);
let catalog = Connection::open(temp.path().join(".owner.sqlite")).unwrap();
let receipt: (i64, i64) = catalog
.query_row(
"SELECT COUNT(*), SUM(completed) FROM storage_requests",
[],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.unwrap();
assert_eq!(receipt, (1, 1));
}
}
#[test]
fn owner_completion_sqlite_failure_reconciles_the_already_committed_identity() {
let temp = tempfile::tempdir().unwrap();
let path = legacy(temp.path(), SCOPE);
let (manager, registration) = registered(temp.path());
{
let lifecycle = manager.lifecycle(SCOPE);
let _lease = lifecycle.lock(true).unwrap();
assert!(lifecycle.lookup(SCOPE, FEATURE).unwrap().is_none());
lifecycle
.reserve(SCOPE, FEATURE, &path, REQUEST, "adopt", EVIDENCE)
.unwrap();
let catalog = Connection::open(temp.path().join(".owner.sqlite")).unwrap();
catalog
.execute_batch(
"CREATE TRIGGER reject_completion BEFORE UPDATE OF completed ON storage_requests
BEGIN SELECT RAISE(ABORT, 'injected owner completion failure'); END;",
)
.unwrap();
}
let scoped = manager.scope(SCOPE).unwrap();
assert!(matches!(
scoped.adopt_existing(&registration, REQUEST, EVIDENCE, validate_legacy),
Err(FeatureStorageError::OutcomeUnknown { .. })
));
let raw = Connection::open(&path).unwrap();
let physical: String = raw
.query_row("SELECT physical FROM __yoi_storage_identity", [], |r| {
r.get(0)
})
.unwrap();
let expected = DatabaseIdentity {
scope: SCOPE.into(),
feature: FEATURE.into(),
physical,
};
assert_identity_error(scoped.open_existing(&registration, &expected));
drop(raw);
drop(scoped);
drop(manager);
{
let catalog = Connection::open(temp.path().join(".owner.sqlite")).unwrap();
let completed: i64 = catalog
.query_row(
"SELECT completed FROM storage_requests WHERE request=?1",
[REQUEST],
|r| r.get(0),
)
.unwrap();
assert_eq!(completed, 0);
catalog
.execute_batch("DROP TRIGGER reject_completion")
.unwrap();
}
let (manager, registration) = registered(temp.path());
let scoped = manager.scope(SCOPE).unwrap();
let database = scoped
.adopt_existing(&registration, REQUEST, EVIDENCE, validate_legacy)
.unwrap();
assert_eq!(database.identity(), &expected);
assert_eq!(values(&database), ["preserved"]);
assert_eq!(
scoped
.open_existing(&registration, &expected)
.unwrap()
.identity(),
&expected
);
}
#[test]
fn open_existing_never_creates_an_unbound_or_missing_owned_database() {
let temp = tempfile::tempdir().unwrap();
let (manager, registration) = registered(temp.path());
let scoped = manager.scope(SCOPE).unwrap();
let fabricated = DatabaseIdentity {
scope: SCOPE.into(),
feature: FEATURE.into(),
physical: "never-owned".into(),
};
assert_identity_error(scoped.open_existing(&registration, &fabricated));
let path = database_path(temp.path(), SCOPE);
assert!(!path.exists());
let database = scoped.open(&registration).unwrap();
let expected = database.identity().clone();
drop(database);
drop(scoped);
drop(manager);
fs::remove_file(&path).unwrap();
let (manager, registration) = registered(temp.path());
let scoped = manager.scope(SCOPE).unwrap();
assert!(scoped.open_existing(&registration, &expected).is_err());
assert!(scoped.open(&registration).is_err());
assert!(
!path.exists(),
"missing owned DB must not be silently recreated"
);
}
#[test]
fn wrong_expected_workspace_feature_or_physical_identity_never_enters_release_closure() {
let temp = tempfile::tempdir().unwrap();
let (manager, registration) = registered(temp.path());
let scoped = manager.scope(SCOPE).unwrap();
let database = scoped.open(&registration).unwrap();
for field in 0..3 {
let mut wrong = database.identity().clone();
match field {
0 => wrong.scope = "another-workspace".into(),
1 => wrong.feature = "another-feature".into(),
_ => wrong.physical = Uuid::now_v7().to_string(),
}
assert_identity_error(scoped.open_existing(&registration, &wrong));
assert_identity_error(database.with_identity(&wrong, |_| -> Result<()> {
panic!("release closure must not run for another target")
}));
}
}
#[test]
fn replaced_path_is_refused_even_when_replacement_copies_the_persisted_identity() {
let temp = tempfile::tempdir().unwrap();
let (manager, registration) = registered(temp.path());
let scoped = manager.scope(SCOPE).unwrap();
let database = scoped.open(&registration).unwrap();
let expected = database.identity().clone();
insert(&database, "preserved");
let path = database_path(temp.path(), SCOPE);
let replacement = temp.path().join("replacement.sqlite");
backup_database(&path, &replacement).unwrap();
fs::rename(&path, temp.path().join("retired.sqlite")).unwrap();
fs::rename(&replacement, &path).unwrap();
assert_identity_error(database.with_identity(&expected, |_| -> Result<()> {
panic!("stale inode must not release a result")
}));
assert_identity_error(scoped.open_existing(&registration, &expected));
let (other, other_registration) = registered(temp.path());
assert_identity_error(
other
.scope(SCOPE)
.unwrap()
.open_existing(&other_registration, &expected),
);
}
#[test]
fn corrupt_persisted_identity_or_receipt_is_refused_without_repair() {
for corrupt in [
"UPDATE __yoi_storage_identity SET scope='wrong-workspace'",
"UPDATE __yoi_storage_identity SET feature='wrong-feature'",
"UPDATE __yoi_storage_identity SET physical='wrong-physical'",
"UPDATE __yoi_storage_identity SET request='wrong-request'",
"DELETE FROM __yoi_storage_identity",
] {
let temp = tempfile::tempdir().unwrap();
let (manager, registration) = registered(temp.path());
let scoped = manager.scope(SCOPE).unwrap();
let database = scoped.open(&registration).unwrap();
let expected = database.identity().clone();
let raw = Connection::open(database_path(temp.path(), SCOPE)).unwrap();
raw.execute_batch(corrupt).unwrap();
assert_identity_error(database.with_connection(|_| -> Result<()> {
panic!("corrupt identity must be checked before a domain query")
}));
assert_identity_error(scoped.open_existing(&registration, &expected));
drop(database);
drop(scoped);
drop(manager);
let (manager, registration) = registered(temp.path());
assert_identity_error(
manager
.scope(SCOPE)
.unwrap()
.open_existing(&registration, &expected),
);
let still_matches: i64 = raw.query_row(
"SELECT COUNT(*) FROM __yoi_storage_identity WHERE scope=?1 AND feature=?2 AND physical=?3",
params![expected.scope, expected.feature, expected.physical], |r| r.get(0),
).unwrap();
if !corrupt.contains("SET request=") {
assert_eq!(still_matches, 0, "refusal must not repair the row");
} else {
let request: String = raw
.query_row("SELECT request FROM __yoi_storage_identity", [], |r| {
r.get(0)
})
.unwrap();
assert_eq!(request, "wrong-request");
}
}
}
#[test]
fn changed_owner_receipt_invalidates_cached_handle_and_reopen() {
let temp = tempfile::tempdir().unwrap();
let (manager, registration) = registered(temp.path());
let scoped = manager.scope(SCOPE).unwrap();
let database = scoped.open(&registration).unwrap();
let identity = database.identity().clone();
let catalog = Connection::open(temp.path().join(".owner.sqlite")).unwrap();
catalog.execute(
"UPDATE storage_requests SET request='replaced-owner-receipt' WHERE scope=?1 AND feature=?2",
params![SCOPE, FEATURE],
).unwrap();
assert_identity_error(database.with_identity(&identity, |_| -> Result<()> {
panic!("cached handle must not release against a replaced owner receipt")
}));
assert_identity_error(scoped.open_existing(&registration, &identity));
let (fresh, registration) = registered(temp.path());
assert_identity_error(
fresh
.scope(SCOPE)
.unwrap()
.open_existing(&registration, &identity),
);
}
#[test]
fn restore_preserves_data_but_assigns_a_new_physical_identity() {
let temp = tempfile::tempdir().unwrap();
let source = temp.path().join("source");
let destination = temp.path().join("destination");
let snapshot = temp.path().join("snapshot");
let (manager, registration) = registered(&source);
let source_scope = manager.scope(SCOPE).unwrap();
let database = source_scope.open(&registration).unwrap();
insert(&database, "snapshot-data");
let original = database.identity().clone();
source_scope.backup(&snapshot).unwrap();
let (restored, registration) = registered(&destination);
let scoped = restored.scope(SCOPE).unwrap();
assert!(
scoped.restore(&snapshot).is_err(),
"a fresh root is not source provenance"
);
scoped.restore_from(&snapshot, &source_scope).unwrap();
let database = scoped.open(&registration).unwrap();
let identity = database.identity().clone();
assert_eq!(identity.scope, original.scope);
assert_eq!(identity.feature, original.feature);
assert_ne!(identity.physical, original.physical);
assert_eq!(values(&database), ["snapshot-data"]);
assert_identity_error(scoped.open_existing(&registration, &original));
drop(database);
drop(scoped);
drop(restored);
let (restored, registration) = registered(&destination);
let database = restored
.scope(SCOPE)
.unwrap()
.open_existing(&registration, &identity)
.unwrap();
assert_eq!(values(&database), ["snapshot-data"]);
}
#[test]
fn same_owner_loss_recovery_retires_old_receipts_and_erasure_blocks_every_restore_target() {
let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("storage");
let snapshot = temp.path().join("snapshot");
let (manager, registration) = registered(&root);
let scoped = manager.scope(SCOPE).unwrap();
let database = scoped.provision(&registration, REQUEST, EVIDENCE).unwrap();
insert(&database, "saved");
let old = database.identity().clone();
scoped.backup(&snapshot).unwrap();
drop(database);
manager.shutdown().unwrap();
// Accidental data-file loss is not an authorized erase: the owner is retained.
fs::remove_dir_all(root.join(SCOPE)).unwrap();
let (manager, registration) = registered(&root);
let scoped = manager.scope(SCOPE).unwrap();
scoped.restore(&snapshot).unwrap();
scoped.restore(&snapshot).unwrap();
let database = scoped.open(&registration).unwrap();
let current = database.identity().clone();
assert_ne!(old, current);
assert_eq!(values(&database), ["saved"]);
assert_identity_error(scoped.open_existing(&registration, &old));
assert_identity_error(scoped.provision(&registration, REQUEST, EVIDENCE));
let owner = Connection::open(root.join(".owner.sqlite")).unwrap();
let retained: (String, bool, bool) = owner
.query_row(
"SELECT physical,completed,current FROM storage_requests WHERE request=?1",
[REQUEST],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
)
.unwrap();
assert_eq!(retained, (old.physical, true, false));
drop(database);
scoped.delete().unwrap();
let (reopened, _) = registered(&root);
let source = reopened.scope(SCOPE).unwrap();
assert!(matches!(
source.restore(&snapshot),
Err(FeatureStorageError::ScopeMaintenance(_))
));
let fresh_root = temp.path().join("fresh-target");
let (target, _) = registered(&fresh_root);
assert!(matches!(
target
.scope(SCOPE)
.unwrap()
.restore_from(&snapshot, &source),
Err(FeatureStorageError::ScopeMaintenance(_))
));
assert!(!fresh_root.join(SCOPE).exists());
}
struct ChildGuard(Child);
impl ChildGuard {
fn start(root: &Path, signals: &Path) -> Self {
Self::start_action(root, signals, None)
}
fn start_action(root: &Path, signals: &Path, snapshot: Option<&Path>) -> Self {
Self(
Command::new(std::env::current_exe().unwrap())
.args([
"--exact",
"identity_tests::lifecycle_delete_child",
"--nocapture",
"--test-threads=1",
])
.current_dir(signals)
.env("YOI_IDENTITY_DELETE_CHILD_ROOT", root)
.env("YOI_IDENTITY_DELETE_CHILD_SIGNALS", signals)
.env(
"YOI_IDENTITY_RESTORE_CHILD_SNAPSHOT",
snapshot.unwrap_or_else(|| Path::new("")),
)
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::inherit())
.spawn()
.unwrap(),
)
}
fn wait_for(&mut self, signal: &Path) {
let deadline = Instant::now() + WAIT;
loop {
if signal.is_file() {
return;
}
assert!(
self.0.try_wait().unwrap().is_none(),
"child exited before {}",
signal.display()
);
assert!(
Instant::now() < deadline,
"timed out waiting for {}",
signal.display()
);
std::thread::yield_now();
}
}
fn finish(&mut self) {
let deadline = Instant::now() + WAIT;
loop {
if let Some(status) = self.0.try_wait().unwrap() {
assert!(status.success(), "delete child failed: {status}");
return;
}
assert!(
Instant::now() < deadline,
"delete child did not finish after lease release"
);
std::thread::yield_now();
}
}
}
impl Drop for ChildGuard {
fn drop(&mut self) {
let _ = self.0.kill();
let _ = self.0.wait();
}
}
#[test]
fn lifecycle_delete_child() {
let Some(root) = std::env::var_os("YOI_IDENTITY_DELETE_CHILD_ROOT") else {
return;
};
let signals = PathBuf::from(std::env::var_os("YOI_IDENTITY_DELETE_CHILD_SIGNALS").unwrap());
let manager = FeatureStorage::new(PathBuf::from(root));
let lock = fs::OpenOptions::new()
.read(true)
.write(true)
.open(
manager
.inner
.root
.join(".lifecycle")
.join(format!("{SCOPE}.lock")),
)
.unwrap();
assert!(
matches!(lock.try_lock(), Err(fs::TryLockError::WouldBlock)),
"exclusive lifecycle lock must contend with active operation"
);
fs::write(signals.join("contended"), b"shared lease observed").unwrap();
if let Some(snapshot) =
std::env::var_os("YOI_IDENTITY_RESTORE_CHILD_SNAPSHOT").filter(|s| !s.is_empty())
{
assert!(matches!(
manager
.scope(SCOPE)
.unwrap()
.restore(PathBuf::from(snapshot)),
Err(FeatureStorageError::DestinationExists(_))
));
fs::write(
signals.join("restore-rejected"),
b"existing physical lifecycle retained",
)
.unwrap();
} else {
manager.scope(SCOPE).unwrap().delete().unwrap();
fs::write(signals.join("deleted"), b"deletion completed").unwrap();
}
}
#[test]
fn cross_process_restore_waits_for_short_guard_and_cannot_replace_the_live_lifecycle() {
let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("storage");
let snapshot = temp.path().join("snapshot");
let signals = temp.path().join("signals");
fs::create_dir(&signals).unwrap();
let (manager, registration) = registered(&root);
let scoped = manager.scope(SCOPE).unwrap();
let database = scoped.open(&registration).unwrap();
insert(&database, "still-current");
scoped.backup(&snapshot).unwrap();
let identity = database.identity().clone();
let guard = database.guard(&identity).unwrap();
let mut child = ChildGuard::start_action(&root, &signals, Some(&snapshot));
child.wait_for(&signals.join("contended"));
assert!(!signals.join("restore-rejected").exists());
assert_eq!(guard.identity(), &identity);
drop(guard);
child.finish();
assert!(signals.join("restore-rejected").is_file());
assert_eq!(database.identity(), &identity);
assert_eq!(values(&database), ["still-current"]);
}
#[test]
fn cross_process_delete_waits_for_transaction_and_release_then_persistently_fences_handles() {
for operation in ["commit", "rollback", "release"] {
let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("storage");
let signals = temp.path().join("signals");
fs::create_dir(&signals).unwrap();
let (manager, registration) = registered(&root);
let scoped = manager.scope(SCOPE).unwrap();
let database = scoped.open(&registration).unwrap();
let identity = database.identity().clone();
insert(&database, "before-delete");
let mut child = None;
let mut contend = || {
let mut process = ChildGuard::start(&root, &signals);
process.wait_for(&signals.join("contended"));
assert!(!signals.join("deleted").exists());
assert!(database_path(&root, SCOPE).is_file());
assert!(
!manager.lifecycle(SCOPE).fenced(SCOPE).unwrap(),
"delete may not fence an operation before lease release"
);
child = Some(process);
};
if operation == "release" {
database
.with_identity(&identity, |connection| {
contend();
let value: String =
connection.query_row("SELECT value FROM records", [], |r| r.get(0))?;
assert_eq!(value, "before-delete");
Ok(())
})
.unwrap();
} else {
let result = database.transaction(|tx| {
tx.execute("INSERT INTO records VALUES ('transaction-write')", [])?;
contend();
if operation == "rollback" {
Err(FeatureStorageError::operation("rollback held transaction"))
} else {
Ok(())
}
});
if operation == "rollback" {
assert!(matches!(result, Err(FeatureStorageError::Operation(_))));
} else {
result.unwrap();
}
}
let mut child = child.unwrap();
child.finish();
assert!(signals.join("deleted").is_file());
assert!(!root.join(SCOPE).exists());
let entered = Cell::new(false);
assert!(
database
.with_identity(&identity, |_| {
entered.set(true);
Ok(())
})
.is_err()
);
assert!(
!entered.get(),
"stale handles must not enter release closure after delete"
);
assert!(
database
.transaction(|_| -> Result<()> { panic!("stale transaction entered") })
.is_err()
);
assert!(scoped.open_existing(&registration, &identity).is_err());
let (fresh, registration) = registered(&root);
assert!(fresh.scope(SCOPE).unwrap().open(&registration).is_err());
fresh.scope(SCOPE).unwrap().delete().unwrap();
assert!(
!root.join(SCOPE).exists(),
"fresh manager must observe persistent deletion fence"
);
}
}
#[test]
fn published_physical_identity_receipt_and_erasure_fences_survive_authorized_adoption() {
let temp = tempfile::tempdir().unwrap();
let path = legacy(temp.path(), SCOPE);
let physical = Uuid::now_v7().to_string();
let fingerprint = format!("sha256:{}", "a".repeat(64));
let connection = Connection::open(&path).unwrap();
// Exact published physical DDL from both 9edebfa and 7b39a44 identity.rs.
connection.execute_batch("CREATE TABLE __yoi_feature_database_identity (
singleton INTEGER PRIMARY KEY CHECK(singleton=1), format INTEGER NOT NULL CHECK(format=1),
scope_id TEXT NOT NULL,feature_id TEXT NOT NULL,incarnation TEXT NOT NULL);
CREATE TABLE __yoi_feature_provisioning_receipt (singleton INTEGER PRIMARY KEY CHECK(singleton=1),request_fingerprint TEXT NOT NULL);").unwrap();
connection
.execute(
"INSERT INTO __yoi_feature_database_identity VALUES(1,1,?1,?2,?3)",
params![SCOPE, FEATURE, physical],
)
.unwrap();
connection
.execute(
"INSERT INTO __yoi_feature_provisioning_receipt VALUES(1,?1)",
[&fingerprint],
)
.unwrap();
let (manager, registration) = registered(temp.path());
let scope = manager.scope(SCOPE).unwrap();
let database = scope
.adopt_existing(&registration, REQUEST, EVIDENCE, validate_legacy)
.unwrap();
assert_eq!(database.identity().physical, physical);
assert_eq!(values(&database), ["preserved"]);
let saved: (String,String) = connection.query_row(
"SELECT incarnation,request_fingerprint FROM __yoi_feature_database_identity CROSS JOIN __yoi_feature_provisioning_receipt",[],|r|Ok((r.get(0)?,r.get(1)?))).unwrap();
assert_eq!(saved, (physical, fingerprint));
scope.delete().unwrap();
let (other, registration) = registered(temp.path());
assert!(other.scope(SCOPE).unwrap().open(&registration).is_err());
assert!(
temp.path()
.join(".lifecycle")
.join(format!("{SCOPE}.deleted"))
.exists()
);
}
#[test]
fn published_scope_erasure_or_unfinished_lifecycle_is_not_lost_on_owner_upgrade() {
for phase in ["deleted", "deleting", "creating", "restoring"] {
let temp = tempfile::tempdir().unwrap();
let path = legacy(temp.path(), SCOPE);
fs::create_dir(temp.path().join(".scope-locks")).unwrap();
fs::write(
temp.path()
.join(".scope-locks")
.join(format!("{SCOPE}.lock")),
format!("v1 {phase} {}\n", Uuid::now_v7()),
)
.unwrap();
let (manager, registration) = registered(temp.path());
assert!(
manager
.scope(SCOPE)
.unwrap()
.adopt_existing(&registration, REQUEST, EVIDENCE, validate_legacy)
.is_err()
);
assert!(!has_identity(&Connection::open(path).unwrap()));
assert!(!temp.path().join(".owner.sqlite").exists());
}
}
#[test]
fn explicit_guard_pins_short_decision_without_holding_connection_or_sqlite_transaction() {
let temp = tempfile::tempdir().unwrap();
let (manager, registration) = registered(temp.path());
let scope = manager.scope(SCOPE).unwrap();
let database = scope.open(&registration).unwrap();
let guard = database.guard(database.identity()).unwrap();
assert_eq!(guard.identity(), database.identity());
// Unlike a connection closure, the short guard permits consumer queries and
// their own transaction while holding the same physical lifecycle lease.
let reopened = scope
.open_existing(&registration, guard.identity())
.unwrap();
insert(&reopened, "guarded");
assert_eq!(values(&database), ["guarded"]);
let lock = fs::OpenOptions::new()
.read(true)
.write(true)
.open(temp.path().join(".lifecycle").join(format!("{SCOPE}.lock")))
.unwrap();
assert!(matches!(lock.try_lock(), Err(fs::TryLockError::WouldBlock)));
drop(guard);
lock.try_lock().unwrap();
}
+136
View File
@@ -0,0 +1,136 @@
//! Retained physical metadata from published FeatureStorage writers. These are
//! provenance, not authority: adoption still needs a Host grant, attribution and
//! receipt/shape validation. Never rewrite their identity or provisioning receipt.
use super::*;
pub(super) fn identity(path: &Path, scope: &str, feature: &str) -> Result<Option<String>> {
let connection = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY)?;
let exists = |name: &str| -> Result<bool> {
Ok(connection.query_row(
"SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE type='table' AND name=?1)",
[name],
|r| r.get(0),
)?)
};
if !exists("__yoi_feature_database_identity")? {
if exists("__yoi_feature_provisioning_receipt")? {
return Err(FeatureStorageError::Identity(
"legacy provisioning receipt has no physical identity".into(),
));
}
return Ok(None);
}
verify_shape(
&connection,
"__yoi_feature_database_identity",
"CREATE TABLE __yoi_feature_database_identity (singleton INTEGER PRIMARY KEY CHECK(singleton=1), format INTEGER NOT NULL CHECK(format=1), scope_id TEXT NOT NULL,feature_id TEXT NOT NULL,incarnation TEXT NOT NULL)",
)?;
let count: i64 = connection.query_row(
"SELECT COUNT(*) FROM __yoi_feature_database_identity",
[],
|r| r.get(0),
)?;
let (format,saved_scope,saved_feature,physical): (i64,String,String,String) = connection.query_row(
"SELECT format,scope_id,feature_id,incarnation FROM __yoi_feature_database_identity WHERE singleton=1",[],
|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?)))?;
let id = Uuid::parse_str(&physical)
.map_err(|_| FeatureStorageError::Identity("corrupt legacy physical identity".into()))?;
if count != 1
|| format != 1
|| scope != saved_scope
|| feature != saved_feature
|| id.get_version_num() != 7
|| id.to_string() != physical
{
return Err(FeatureStorageError::Identity(
"legacy physical attribution differs or is corrupt".into(),
));
}
if exists("__yoi_feature_provisioning_receipt")? {
verify_shape(
&connection,
"__yoi_feature_provisioning_receipt",
"CREATE TABLE __yoi_feature_provisioning_receipt (singleton INTEGER PRIMARY KEY CHECK(singleton=1),request_fingerprint TEXT NOT NULL)",
)?;
let (count,fingerprint): (i64,String) = connection.query_row(
"SELECT (SELECT COUNT(*) FROM __yoi_feature_provisioning_receipt),request_fingerprint FROM __yoi_feature_provisioning_receipt WHERE singleton=1",[],|r|Ok((r.get(0)?,r.get(1)?)))?;
if count != 1
|| fingerprint.len() != 71
|| !fingerprint.starts_with("sha256:")
|| !fingerprint.as_bytes()[7..]
.iter()
.all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(b))
{
return Err(FeatureStorageError::Identity(
"corrupt legacy provisioning receipt".into(),
));
}
}
Ok(Some(physical))
}
pub(super) fn fenced(root: &Path, scope: &str) -> Result<bool> {
if root
.join(".lifecycle")
.join(format!("{scope}.deleted"))
.try_exists()?
{
return Ok(true);
}
let record_path = root.join(".scope-locks").join(format!("{scope}.lock"));
let record = match fs::read_to_string(record_path) {
Ok(record) if !record.is_empty() => record,
Ok(_) => return Ok(false),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false),
Err(error) => return Err(error.into()),
};
let fields: Vec<_> = record.trim_end_matches('\n').split(' ').collect();
if fields.len() != 3 || fields[0] != "v1" {
return Err(FeatureStorageError::Identity(
"corrupt legacy scope record".into(),
));
}
let id = Uuid::parse_str(fields[2])
.map_err(|_| FeatureStorageError::Identity("corrupt legacy scope identity".into()))?;
if record != format!("v1 {} {id}\n", fields[1]) {
return Err(FeatureStorageError::Identity(
"noncanonical legacy scope record".into(),
));
}
match fields[1] {
"deleted" | "deleting" => Ok(true),
"live" => {
if fs::read_to_string(root.join(scope).join(".incarnation"))? != format!("{id}\n") {
return Err(FeatureStorageError::Identity(
"legacy scope incarnation differs".into(),
));
}
Ok(false)
}
_ => Err(FeatureStorageError::Identity(
"unfinished legacy scope requires reconciliation by its owner".into(),
)),
}
}
fn verify_shape(connection: &Connection, table: &str, expected: &str) -> Result<()> {
let sql: String = connection.query_row(
"SELECT sql FROM sqlite_schema WHERE type='table' AND name=?1",
[table],
|r| r.get(0),
)?;
// These two frozen DDL definitions contain no SQL string literals. Whitespace
// and token case are immaterial; constraints and column programs are not.
let normalized = |s: &str| {
s.chars()
.filter(|c| !c.is_ascii_whitespace())
.collect::<String>()
.to_ascii_lowercase()
};
if normalized(&sql) != normalized(expected) {
return Err(FeatureStorageError::Identity(
"legacy physical metadata shape differs".into(),
));
}
Ok(())
}
File diff suppressed because it is too large Load Diff
+406
View File
@@ -0,0 +1,406 @@
//! Physical ownership is independent of domain authorization. The trusted Host
//! supplies adoption evidence and validates domain attribution inside the identity
//! transaction. Merely finding a file never grants adoption authority.
//!
//! Lock order is lifecycle file, manager state, connection, SQLite transaction.
//! The stable scope lock is shared through consumer commit/rollback and exclusive
//! during provisioning, adoption, deletion and restore, across managers/processes.
//! It is advisory: arbitrary native code bypassing this owner can rename files.
//! Such replacement is rejected on the next access using the file identity and
//! persisted physical identity. Hosts must protect the root and its lock/owner DB
//! against uncooperative writers; this is not a filesystem sandbox.
use super::*;
use rusqlite::OptionalExtension;
use std::fs::{File, OpenOptions};
use std::os::unix::fs::MetadataExt;
const IDENTITY_TABLE: &str = "__yoi_storage_identity";
/// Expected storage target, retained with authorization and rechecked on resume.
/// This is a physical identity, not an authorization credential or a Job lease.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct DatabaseIdentity {
pub scope: String,
pub feature: String,
pub physical: String,
}
#[derive(Clone, Debug)]
pub(super) struct Binding {
pub identity: DatabaseIdentity,
pub device: u64,
pub inode: u64,
pub request: String,
}
pub(super) struct Lifecycle {
pub root: PathBuf,
pub scope: String,
}
impl Lifecycle {
pub fn lock(&self, exclusive: bool) -> Result<File> {
durable_create_directory(&self.root.join(".lifecycle"), &mut sync_directory)?;
let lock = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(
self.root
.join(".lifecycle")
.join(format!("{}.lock", self.scope)),
)?;
if exclusive {
lock.lock()?;
} else {
lock.lock_shared()?;
}
Ok(lock)
}
fn catalog(&self, create: bool) -> Result<Connection> {
let flags = OpenFlags::SQLITE_OPEN_READ_WRITE
| if create {
OpenFlags::SQLITE_OPEN_CREATE
} else {
OpenFlags::empty()
};
let connection = Connection::open_with_flags(self.root.join(".owner.sqlite"), flags)?;
configure_connection(&connection)?;
if create {
connection.execute_batch(
"
CREATE TABLE IF NOT EXISTS storage_requests (
request TEXT PRIMARY KEY, scope TEXT NOT NULL, feature TEXT NOT NULL,
physical TEXT NOT NULL, device INTEGER NOT NULL, inode INTEGER NOT NULL,
kind TEXT NOT NULL, evidence TEXT NOT NULL, completed INTEGER NOT NULL DEFAULT 0,
current INTEGER NOT NULL DEFAULT 1);
CREATE UNIQUE INDEX IF NOT EXISTS storage_current_target ON storage_requests(scope,feature) WHERE current=1;
CREATE TABLE IF NOT EXISTS storage_deletions (scope TEXT PRIMARY KEY);
",
)?;
}
Ok(connection)
}
pub fn fenced(&self, scope: &str) -> Result<bool> {
if legacy_owner::fenced(&self.root, scope)? {
return Ok(true);
}
if !self.root.join(".owner.sqlite").try_exists()? {
return Ok(false);
}
Ok(self.catalog(false)?.query_row(
"SELECT EXISTS(SELECT 1 FROM storage_deletions WHERE scope=?1)",
[scope],
|r| r.get(0),
)?)
}
pub fn delete(&self, scope: &str) -> Result<()> {
let directory = self.root.join(".lifecycle");
durable_create_directory(&directory, &mut sync_directory)?;
let fence = OpenOptions::new()
.write(true)
.create(true)
.truncate(false)
.open(directory.join(format!("{scope}.deleted")))?;
fence.sync_all()?;
sync_directory(&directory)?;
self.catalog(true)?.execute(
"INSERT OR IGNORE INTO storage_deletions(scope) VALUES(?1)",
[scope],
)?;
Ok(())
}
pub fn lookup(&self, scope: &str, feature: &str) -> Result<Option<Binding>> {
if self.fenced(scope)? {
return Err(FeatureStorageError::ScopeMaintenance(scope.into()));
}
if !self.root.join(".owner.sqlite").try_exists()? {
return Ok(None);
}
let catalog = self.catalog(false)?;
Ok(catalog.query_row("SELECT physical,device,inode,request FROM storage_requests WHERE scope=?1 AND feature=?2 AND completed=1 AND current=1", params![scope,feature], |r| Ok(Binding {
identity: DatabaseIdentity { scope: scope.into(), feature: feature.into(), physical: r.get(0)? },
device: r.get(1)?, inode: r.get(2)?, request: r.get(3)?,
})).optional()?)
}
/// Called under exclusive lifecycle lock. The reservation precedes identity
/// persistence and survives response loss; exact retries reconcile the same
/// inode and UUID rather than manufacture past provisioning permission.
pub fn reserve(
&self,
scope: &str,
feature: &str,
path: &Path,
request: &str,
kind: &str,
evidence: &str,
) -> Result<Binding> {
if request.trim().is_empty() || evidence.trim().is_empty() {
return Err(FeatureStorageError::Identity(
"request and Host authority evidence are required".into(),
));
}
if self.fenced(scope)? {
return Err(FeatureStorageError::ScopeMaintenance(scope.into()));
}
if kind != "provision" && !path.is_file() {
return Err(FeatureStorageError::Identity(
"expected existing database is missing".into(),
));
}
let mut catalog = self.catalog(true)?;
let tx = catalog.transaction_with_behavior(TransactionBehavior::Immediate)?;
let saved: Option<(String,String,String,u64,u64,String,String,bool)> = tx.query_row(
"SELECT scope,feature,physical,device,inode,kind,evidence,current FROM storage_requests WHERE request=?1", [request],
|r| Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?,r.get(4)?,r.get(5)?,r.get(6)?,r.get(7)?))).optional()?;
if let Some((s, f, physical, device, inode, k, e, current)) = saved {
if s != scope || f != feature || k != kind || e != evidence || !current {
return Err(FeatureStorageError::Identity(
"request identity was reused for a different target or authority".into(),
));
}
let binding = Binding {
identity: DatabaseIdentity {
scope: s,
feature: f,
physical,
},
device,
inode,
request: request.into(),
};
binding.check_file(path)?;
return Ok(binding);
}
let busy: bool = tx.query_row(
"SELECT EXISTS(SELECT 1 FROM storage_requests WHERE scope=?1 AND feature=?2 AND current=1)",
params![scope, feature],
|r| r.get(0),
)?;
if busy && kind != "restore" {
return Err(FeatureStorageError::Identity(
"target already has an ownership request; reconcile that request".into(),
));
}
if kind == "provision" {
durable_create_directory(
path.parent()
.ok_or_else(|| FeatureStorageError::Identity("missing parent".into()))?,
&mut sync_directory,
)?;
let file = OpenOptions::new().write(true).create_new(true).open(path)?;
file.sync_all()?;
sync_directory(path.parent().unwrap())?;
}
// A verified restore supersedes the active physical lifecycle, but keeps
// every old request/receipt as provenance. It cannot replay as current authority.
if kind == "restore" {
tx.execute(
"UPDATE storage_requests SET current=0 WHERE scope=?1 AND feature=?2 AND current=1",
params![scope, feature],
)?;
}
let metadata = fs::metadata(path)?;
if !metadata.is_file() {
return Err(FeatureStorageError::Identity(
"database is not a regular file".into(),
));
}
let binding = Binding {
identity: DatabaseIdentity {
scope: scope.into(),
feature: feature.into(),
physical: if kind == "adopt" {
legacy_owner::identity(path, scope, feature)?
.unwrap_or_else(|| Uuid::now_v7().to_string())
} else {
Uuid::now_v7().to_string()
},
},
device: metadata.dev(),
inode: metadata.ino(),
request: request.into(),
};
tx.execute("INSERT INTO storage_requests(request,scope,feature,physical,device,inode,kind,evidence) VALUES(?1,?2,?3,?4,?5,?6,?7,?8)",params![request,scope,feature,binding.identity.physical,binding.device,binding.inode,kind,evidence])?;
tx.commit()?;
Ok(binding)
}
/// Snapshot bytes are not authority. Match their saved identity to this
/// explicitly supplied source owner's completed provenance, including an old
/// lifecycle superseded by a prior restore. The caller holds its lifecycle lock.
pub fn verify_snapshot(&self, scope: &str, feature: &str, path: &Path) -> Result<()> {
if self.fenced(scope)? {
return Err(FeatureStorageError::ScopeMaintenance(scope.into()));
}
if !self.root.join(".owner.sqlite").is_file() {
return Err(FeatureStorageError::Identity(
"snapshot source owner is missing".into(),
));
}
let connection = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY)?;
let (s, f, physical, request): (String, String, String, String) = connection.query_row(
"SELECT scope,feature,physical,request FROM __yoi_storage_identity WHERE singleton=1",
[],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)),
)?;
Uuid::parse_str(&physical).map_err(|_| {
FeatureStorageError::Identity("invalid snapshot physical identity".into())
})?;
let known: bool = self.catalog(false)?.query_row(
"SELECT EXISTS(SELECT 1 FROM storage_requests WHERE scope=?1 AND feature=?2 AND physical=?3 AND request=?4 AND completed=1)",
params![scope, feature, physical, request], |r| r.get(0))?;
if s != scope || f != feature || !known {
return Err(FeatureStorageError::Identity(
"snapshot does not belong to the selected source owner".into(),
));
}
Ok(())
}
pub fn request_kind(&self, request: &str) -> Result<Option<String>> {
if !self.root.join(".owner.sqlite").try_exists()? {
return Ok(None);
}
Ok(self
.catalog(false)?
.query_row(
"SELECT kind FROM storage_requests WHERE request=?1 AND current=1",
[request],
|r| r.get(0),
)
.optional()?)
}
pub fn complete(&self, binding: &Binding) -> Result<()> {
let changed = self.catalog(false)?.execute("UPDATE storage_requests SET completed=1 WHERE request=?1 AND scope=?2 AND feature=?3 AND physical=?4 AND device=?5 AND inode=?6 AND current=1",params![binding.request,binding.identity.scope,binding.identity.feature,binding.identity.physical,binding.device,binding.inode])?;
if changed != 1 {
return Err(FeatureStorageError::Identity(
"ownership reservation changed".into(),
));
}
Ok(())
}
pub fn verify(&self, binding: &Binding, path: &Path, connection: &Connection) -> Result<()> {
let current = self
.lookup(&binding.identity.scope, &binding.identity.feature)?
.ok_or_else(|| FeatureStorageError::Identity("no completed ownership record".into()))?;
if current.identity != binding.identity
|| current.device != binding.device
|| current.inode != binding.inode
|| current.request != binding.request
{
return Err(FeatureStorageError::Identity(
"physical ownership changed".into(),
));
}
binding.check_file(path)?;
binding.check_database(connection)
}
}
impl Binding {
pub fn unknown(&self, error: FeatureStorageError) -> FeatureStorageError {
FeatureStorageError::OutcomeUnknown {
identity: self.identity.clone(),
detail: error.to_string(),
}
}
pub fn check_file(&self, path: &Path) -> Result<()> {
let metadata = fs::metadata(path)?;
if metadata.dev() != self.device || metadata.ino() != self.inode {
return Err(FeatureStorageError::Identity(
"database path was replaced; explicit authorized adoption is required".into(),
));
}
Ok(())
}
pub fn check_database(&self, connection: &Connection) -> Result<()> {
Uuid::parse_str(&self.identity.physical)
.map_err(|_| FeatureStorageError::Identity("invalid physical identity".into()))?;
let found: Option<(String, String, String, String)> = connection
.query_row(
&format!(
"SELECT scope,feature,physical,request FROM {IDENTITY_TABLE} WHERE singleton=1"
),
[],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)),
)
.optional()?;
if found
!= Some((
self.identity.scope.clone(),
self.identity.feature.clone(),
self.identity.physical.clone(),
self.request.clone(),
))
{
return Err(FeatureStorageError::Identity(
"database identity/receipt is missing or mismatched".into(),
));
}
Ok(())
}
/// Restore is a new physical lifecycle. Retain old provenance but never
/// return its identity as authority over the restored inode.
pub fn restore_identity(&self, connection: &mut Connection) -> Result<()> {
let tx = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let exists: bool = tx.query_row(
"SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE name='__yoi_storage_identity')",
[],
|r| r.get(0),
)?;
if exists {
if self.check_database(&tx).is_ok() {
return Ok(());
}
let (scope, feature): (String, String) = tx.query_row(
"SELECT scope,feature FROM __yoi_storage_identity WHERE singleton=1",
[],
|r| Ok((r.get(0)?, r.get(1)?)),
)?;
if scope != self.identity.scope || feature != self.identity.feature {
return Err(FeatureStorageError::Identity(
"snapshot belongs to another scope".into(),
));
}
tx.execute_batch("CREATE TABLE IF NOT EXISTS __yoi_storage_origins (scope TEXT NOT NULL, feature TEXT NOT NULL, physical TEXT NOT NULL, request TEXT NOT NULL, PRIMARY KEY(physical,request)); INSERT OR IGNORE INTO __yoi_storage_origins SELECT scope,feature,physical,request FROM __yoi_storage_identity; DELETE FROM __yoi_storage_identity;")?;
}
self.persist(&tx)?;
tx.commit()?;
Ok(())
}
pub fn persist(&self, tx: &Transaction<'_>) -> Result<()> {
tx.execute_batch(&format!(
"CREATE TABLE IF NOT EXISTS {IDENTITY_TABLE} (
singleton INTEGER PRIMARY KEY CHECK(singleton=1), scope TEXT NOT NULL,
feature TEXT NOT NULL, physical TEXT NOT NULL, request TEXT NOT NULL);"
))?;
let count: i64 =
tx.query_row(&format!("SELECT COUNT(*) FROM {IDENTITY_TABLE}"), [], |r| {
r.get(0)
})?;
if count == 0 {
tx.execute(
&format!("INSERT INTO {IDENTITY_TABLE} VALUES(1,?1,?2,?3,?4)"),
params![
self.identity.scope,
self.identity.feature,
self.identity.physical,
self.request
],
)?;
}
self.check_database(tx)
}
}
+12 -8
View File
@@ -159,10 +159,21 @@ fn purge_failure_rolls_back_content_but_not_the_committed_fence_and_restart_reco
c.execute_batch("CREATE TRIGGER injected_failure BEFORE DELETE ON subjects BEGIN SELECT RAISE(ABORT, 'simulated purge failure'); END;")?; c.execute_batch("CREATE TRIGGER injected_failure BEFORE DELETE ON subjects BEGIN SELECT RAISE(ABORT, 'simulated purge failure'); END;")?;
Ok(()) Ok(())
}).unwrap(); }).unwrap();
assert!(store.finish_subject_deletion(&id, "delete").is_err()); let error = store.finish_subject_deletion(&id, "delete").unwrap_err();
assert!(
error.to_string().contains("simulated purge failure"),
"{error}"
);
assert_eq!(count(&store, "memory_records"), 1); assert_eq!(count(&store, "memory_records"), 1);
assert_eq!(count(&store, "memory_changes"), 1); assert_eq!(count(&store, "memory_changes"), 1);
assert_eq!(store.subject_deletion(&id).unwrap(), Some(pending.clone())); assert_eq!(store.subject_deletion(&id).unwrap(), Some(pending.clone()));
store
.database
.try_with_connection::<_, SubjektivError>(|c| {
c.execute_batch("DROP TRIGGER injected_failure")?;
Ok(())
})
.unwrap();
drop(store); drop(store);
drop(scope); drop(scope);
manager.shutdown().unwrap(); manager.shutdown().unwrap();
@@ -177,13 +188,6 @@ fn purge_failure_rolls_back_content_but_not_the_committed_fence_and_restart_reco
pending pending
); );
assert!(store.list_memories(&id).is_err()); assert!(store.list_memories(&id).is_err());
store
.database
.try_with_connection::<_, SubjektivError>(|c| {
c.execute_batch("DROP TRIGGER injected_failure")?;
Ok(())
})
.unwrap();
assert!( assert!(
store store
.finish_subject_deletion(&id, "delete") .finish_subject_deletion(&id, "delete")
+65 -30
View File
@@ -498,8 +498,8 @@ mod input_delivery {
} }
#[test] #[test]
fn input_delivery_migration_failure_rolls_back_table_creation_and_retries() { fn input_delivery_migration_collision_is_rejected_before_ddl_and_retries() {
// A colliding trigger is created before migration so failure occurs after its new table DDL. // A colliding trigger must be diagnosed before pending migration DDL.
let temp = tempfile::tempdir().unwrap(); let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("rollback"); let root = temp.path().join("rollback");
let manager = FeatureStorage::new(&root); let manager = FeatureStorage::new(&root);
@@ -529,12 +529,16 @@ mod input_delivery {
let manager = FeatureStorage::new(&root); let manager = FeatureStorage::new(&root);
let scope = manager.scope("input-delivery-rollback").unwrap(); let scope = manager.scope("input-delivery-rollback").unwrap();
let registration = SubjektivStore::register(&scope).unwrap(); let registration = SubjektivStore::register(&scope).unwrap();
assert!(matches!( match SubjektivStore::open(&scope, &registration) {
SubjektivStore::open(&scope, &registration),
Err(crate::SubjektivError::Storage( Err(crate::SubjektivError::Storage(
feature_storage::FeatureStorageError::MigrationFailed { .. } feature_storage::FeatureStorageError::MigrationHistory { detail, .. },
)) )) => assert!(
)); detail.contains("journal_generation_input_delivery_no_update"),
"{detail}"
),
Err(error) => panic!("expected pre-DDL shape refusal, got {error:?}"),
Ok(_) => panic!("colliding schema was accepted"),
}
drop(scope); drop(scope);
manager.shutdown().unwrap(); manager.shutdown().unwrap();
drop(manager); drop(manager);
@@ -551,6 +555,11 @@ mod input_delivery {
assert_eq!(store.subject(&subject.id).unwrap(), Some(subject.clone())); assert_eq!(store.subject(&subject.id).unwrap(), Some(subject.clone()));
store.database.try_transaction(|tx| { store.database.try_transaction(|tx| {
assert!(!tx.query_row("SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE name='journal_generation_input_deliveries')", [], |r| r.get::<_, bool>(0))?); assert!(!tx.query_row("SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE name='journal_generation_input_deliveries')", [], |r| r.get::<_, bool>(0))?);
let receipts = tx.prepare("SELECT version,name FROM __yoi_feature_schema_migrations ORDER BY version")?
.query_map([], |r| Ok((r.get::<_, u32>(0)?, r.get::<_, String>(1)?)))?
.collect::<std::result::Result<Vec<_>, _>>()?;
let expected: Vec<_> = (&crate::MIGRATIONS[..12]).iter().map(|m| (m.version(), m.name().to_string())).collect();
assert_eq!(receipts, expected);
tx.execute_batch("DROP TRIGGER journal_generation_input_delivery_no_update")?; tx.execute_batch("DROP TRIGGER journal_generation_input_delivery_no_update")?;
Ok::<_, JournalError>(()) Ok::<_, JournalError>(())
}).unwrap(); }).unwrap();
@@ -5083,10 +5092,12 @@ fn backup_restore_preserves_generation_records_corrections_sources_position_and_
vec![], vec![],
); );
fixture.close(&generation); fixture.close(&generation);
fixture.workspace.as_ref().unwrap().delete().unwrap();
fixture.store.take(); fixture.store.take();
fixture.workspace.take(); fixture.workspace.take();
fixture.manager.take().unwrap().shutdown().unwrap(); fixture.manager.take().unwrap().shutdown().unwrap();
// Data-file loss is recoverable while the durable source owner remains intact.
// Explicit delete is a separate irreversible fence, covered by storage-owner tests.
std::fs::remove_dir_all(fixture.temp.path().join("storage/journal-tests")).unwrap();
let manager = FeatureStorage::new(fixture.temp.path().join("storage")); let manager = FeatureStorage::new(fixture.temp.path().join("storage"));
let workspace = manager.workspace("journal-tests").unwrap(); let workspace = manager.workspace("journal-tests").unwrap();
workspace.restore(&snapshot).unwrap(); workspace.restore(&snapshot).unwrap();
@@ -5222,16 +5233,20 @@ fn failure_before_receipt_rolls_back_record_and_checkpoint_then_exact_retry_succ
); );
let before = fixture.state(); let before = fixture.state();
let sources = fixture.sources(); let sources = fixture.sources();
assert!(matches!( let error = fixture
fixture.store().execute_journal(&ctx, record.clone()), .store()
Err(JournalError::Storage(_)) .execute_journal(&ctx, record.clone())
)); .unwrap_err();
assert!(
error.to_string().contains("injected before receipt"),
"{error}"
);
fixture.sql("DROP TRIGGER fail_test_receipt;").unwrap();
fixture.reopen(); fixture.reopen();
assert_eq!(fixture.state(), before); assert_eq!(fixture.state(), before);
assert_eq!(fixture.sources(), sources); assert_eq!(fixture.sources(), sources);
assert!(fixture.entries().is_empty()); assert!(fixture.entries().is_empty());
assert!(fixture.receipt("fault-record").is_none()); assert!(fixture.receipt("fault-record").is_none());
fixture.sql("DROP TRIGGER fail_test_receipt;").unwrap();
let saved = fixture let saved = fixture
.store() .store()
.execute_journal(&ctx, record.clone()) .execute_journal(&ctx, record.clone())
@@ -5248,16 +5263,20 @@ fn failure_before_receipt_rolls_back_record_and_checkpoint_then_exact_retry_succ
); );
fixture.sql("CREATE TRIGGER fail_test_receipt BEFORE INSERT ON journal_operations WHEN NEW.request_id='fault-checkpoint' BEGIN SELECT RAISE(ABORT, 'injected before receipt'); END;").unwrap(); fixture.sql("CREATE TRIGGER fail_test_receipt BEFORE INSERT ON journal_operations WHEN NEW.request_id='fault-checkpoint' BEGIN SELECT RAISE(ABORT, 'injected before receipt'); END;").unwrap();
let before_checkpoint = fixture.state(); let before_checkpoint = fixture.state();
assert!(matches!( let error = fixture
fixture.store().execute_journal(&ctx, cp.clone()), .store()
Err(JournalError::Storage(_)) .execute_journal(&ctx, cp.clone())
)); .unwrap_err();
assert!(
error.to_string().contains("injected before receipt"),
"{error}"
);
fixture.sql("DROP TRIGGER fail_test_receipt;").unwrap();
fixture.reopen(); fixture.reopen();
assert_eq!(fixture.state(), before_checkpoint); assert_eq!(fixture.state(), before_checkpoint);
assert_eq!(fixture.sources(), sources); assert_eq!(fixture.sources(), sources);
assert_eq!(fixture.receipt("fault-record"), Some(saved)); assert_eq!(fixture.receipt("fault-record"), Some(saved));
assert!(fixture.receipt("fault-checkpoint").is_none()); assert!(fixture.receipt("fault-checkpoint").is_none());
fixture.sql("DROP TRIGGER fail_test_receipt;").unwrap();
let reflected = fixture.store().execute_journal(&ctx, cp.clone()).unwrap(); let reflected = fixture.store().execute_journal(&ctx, cp.clone()).unwrap();
assert_eq!((reflected.process_order, reflected.process_byte), (2, 0)); assert_eq!((reflected.process_order, reflected.process_byte), (2, 0));
fixture.reopen(); fixture.reopen();
@@ -5529,7 +5548,8 @@ fn serialized_metadata_and_text_budgets_preserve_continuation_and_availability_r
} }
#[test] #[test]
fn journal_migration_failure_rolls_back_ddl_and_retries_without_changing_existing_subject() { fn journal_migration_collision_is_rejected_before_ddl_and_retries_without_changing_existing_subject()
{
for (baseline, collision, rolled_back) in [ for (baseline, collision, rolled_back) in [
(&crate::MIGRATIONS[..7], "journal_entries", "journals"), (&crate::MIGRATIONS[..7], "journal_entries", "journals"),
( (
@@ -5564,12 +5584,13 @@ fn journal_migration_failure_rolls_back_ddl_and_retries_without_changing_existin
let manager = FeatureStorage::new(&root); let manager = FeatureStorage::new(&root);
let scope = manager.scope("migration-rollback").unwrap(); let scope = manager.scope("migration-rollback").unwrap();
let registration = SubjektivStore::register(&scope).unwrap(); let registration = SubjektivStore::register(&scope).unwrap();
assert!(matches!( match SubjektivStore::open(&scope, &registration) {
SubjektivStore::open(&scope, &registration),
Err(crate::SubjektivError::Storage( Err(crate::SubjektivError::Storage(
feature_storage::FeatureStorageError::MigrationFailed { .. } feature_storage::FeatureStorageError::MigrationHistory { detail, .. },
)) )) => assert!(detail.contains(collision), "{detail}"),
)); Err(error) => panic!("expected pre-DDL shape refusal, got {error:?}"),
Ok(_) => panic!("colliding schema was accepted"),
}
drop(scope); drop(scope);
manager.shutdown().unwrap(); manager.shutdown().unwrap();
drop(manager); drop(manager);
@@ -5591,6 +5612,17 @@ fn journal_migration_failure_rolls_back_ddl_and_retries_without_changing_existin
[rolled_back], [rolled_back],
|r| r.get::<_, bool>(0) |r| r.get::<_, bool>(0)
)?); )?);
let receipts = tx
.prepare(
"SELECT version,name FROM __yoi_feature_schema_migrations ORDER BY version",
)?
.query_map([], |r| Ok((r.get::<_, u32>(0)?, r.get::<_, String>(1)?)))?
.collect::<std::result::Result<Vec<_>, _>>()?;
let expected: Vec<_> = (baseline)
.iter()
.map(|m| (m.version(), m.name().to_string()))
.collect();
assert_eq!(receipts, expected);
tx.execute_batch(&format!("DROP TABLE {collision}"))?; tx.execute_batch(&format!("DROP TABLE {collision}"))?;
Ok(()) Ok(())
}) })
@@ -5994,16 +6026,19 @@ fn subject_deletion_fence_denies_all_journal_read_write_binding_and_recovery_pat
), ),
); );
// Failed purge and restart must not revive any cached context or exact receipt. // Failed purge and restart must not revive any cached context or exact receipt.
f.sql("CREATE TRIGGER injected_subject_purge_failure BEFORE DELETE ON subjects BEGIN SELECT RAISE(ABORT,'injected'); END;").unwrap(); f.sql("CREATE TRIGGER injected_subject_purge_failure BEFORE DELETE ON subjects BEGIN SELECT RAISE(ABORT,'injected subject purge'); END;").unwrap();
let error = f
.store()
.finish_subject_deletion(&f.host.subject_id, "delete")
.unwrap_err();
assert!( assert!(
f.store() error.to_string().contains("injected subject purge"),
.finish_subject_deletion(&f.host.subject_id, "delete") "{error}"
.is_err()
); );
f.reopen();
subject_deleted(f.store().execute_journal(&ctx, operation));
f.sql("DROP TRIGGER injected_subject_purge_failure;") f.sql("DROP TRIGGER injected_subject_purge_failure;")
.unwrap(); .unwrap();
f.reopen();
subject_deleted(f.store().execute_journal(&ctx, operation));
f.store() f.store()
.finish_subject_deletion(&f.host.subject_id, "delete") .finish_subject_deletion(&f.host.subject_id, "delete")
.unwrap(); .unwrap();
+6 -2
View File
@@ -673,8 +673,11 @@ static MIGRATIONS: &[FeatureMigration] = &[
), ),
]; ];
mod storage_formats;
pub const REGISTRATION: FeatureRegistration = pub const REGISTRATION: FeatureRegistration =
FeatureRegistration::new(SUBJEKTIV_FEATURE_ID, MIGRATIONS); FeatureRegistration::new(SUBJEKTIV_FEATURE_ID, MIGRATIONS)
.with_history_validation(storage_formats::validate);
/// Subject Memory repository restricted to a Host storage scope. /// Subject Memory repository restricted to a Host storage scope.
/// Owns schemas, validation, queries, and domain transactions; delegates physical layout /// Owns schemas, validation, queries, and domain transactions; delegates physical layout
@@ -5623,10 +5626,11 @@ mod tests {
) )
.unwrap(); .unwrap();
assert!(second.previous_change_id.is_some()); assert!(second.previous_change_id.is_some());
workspace.delete().unwrap();
drop(reopened); drop(reopened);
drop(workspace); drop(workspace);
drop(manager); drop(manager);
// Simulated physical loss retains owner receipts, unlike explicit erasure.
std::fs::remove_dir_all(root.join("workspace-a")).unwrap();
let manager = FeatureStorage::new(&root); let manager = FeatureStorage::new(&root);
let workspace = manager.workspace("workspace-a").unwrap(); let workspace = manager.workspace("workspace-a").unwrap();
@@ -0,0 +1,581 @@
//! Storage component fixtures: real SQLite and FeatureStorage registration/open/reopen.
//! Published extension DDL is frozen shape evidence, not a replacement domain implementation.
//! The suite proves physical retention and receipt/shape rejection, not permission adoption,
//! Job recovery, Journal release, or Host authorization. Identity/adoption races belong to the
//! FeatureStorage owner's suite. No production DB, Runtime, or external service is used.
use super::*;
use crate::{REGISTRATION, SubjectRole, SubjektivStore};
use feature_storage::{FeatureDatabase, FeatureRegistration, FeatureStorage, ScopedFeatureStorage};
use rusqlite::types::Value;
use std::path::Path;
const SCOPE: &str = "published-subjektiv-formats";
const PUBLISHED: [Format; 7] = [
Format::Permission18,
Format::Consolidation18,
Format::Fixed18,
Format::Cancellation19,
Format::Consolidation20,
Format::Fixed20,
Format::Cancellation20,
];
type Receipt = (u32, String, String);
type TableRows = Vec<(String, Vec<Vec<Value>>)>;
fn open(root: &Path, registration: FeatureRegistration) -> (ScopedFeatureStorage, FeatureDatabase) {
let scope = FeatureStorage::new(root).scope(SCOPE).unwrap();
let registered = scope.register(registration).unwrap();
let database = scope.open(&registered).unwrap();
(scope, database)
}
fn receipts(connection: &Connection) -> Vec<Receipt> {
connection
.prepare(&format!(
"SELECT version,name,applied_at FROM {HISTORY_TABLE} ORDER BY version"
))
.unwrap()
.query_map([], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))
.unwrap()
.collect::<std::result::Result<Vec<_>, _>>()
.unwrap()
}
fn mark_receipts(connection: &Connection) {
// Distinct saved bytes detect delete/reinsert, even in a single wall-clock second.
connection
.execute(
"UPDATE __yoi_feature_schema_migrations
SET applied_at=printf('original-published-%d',version)",
[],
)
.unwrap();
}
fn all_rows(connection: &Connection) -> TableRows {
let names: Vec<String> = connection
.prepare("SELECT name FROM sqlite_schema WHERE type='table' ORDER BY name")
.unwrap()
.query_map([], |row| row.get(0))
.unwrap()
.collect::<std::result::Result<_, _>>()
.unwrap();
names
.into_iter()
.map(|name| {
let quoted = name.replace('"', "\"\"");
let columns = connection
.prepare(&format!("SELECT * FROM \"{quoted}\""))
.unwrap()
.column_count();
let order = (1..=columns)
.map(|column| column.to_string())
.collect::<Vec<_>>()
.join(",");
let mut statement = connection
.prepare(&format!("SELECT * FROM \"{quoted}\" ORDER BY {order}"))
.unwrap();
let rows = statement
.query_map([], |row| {
(0..columns)
.map(|column| row.get(column))
.collect::<std::result::Result<Vec<Value>, _>>()
})
.unwrap()
.collect::<std::result::Result<Vec<_>, _>>()
.unwrap();
(name, rows)
})
.collect()
}
fn history(connection: &Connection) -> Vec<(u32, String)> {
receipts(connection)
.into_iter()
.map(|(version, name, _)| (version, name))
.collect()
}
fn assert_valid(connection: &mut Connection) {
let tx = connection.transaction().unwrap();
validate(&tx, &history(&tx)).unwrap();
tx.commit().unwrap();
}
fn assert_foreign_keys(connection: &Connection) {
assert!(
connection
.prepare("PRAGMA foreign_key_check")
.unwrap()
.query([])
.unwrap()
.next()
.unwrap()
.is_none()
);
}
// These fixture functions execute the historical SQL, including rebuilds and triggers. They
// never invent alternative tables with similar names or stamp a canonical receipt on old data.
fn permission_fixture(connection: &Connection) {
connection.execute_batch(PERMISSION_SEAL_SQL).unwrap();
for table in PERMISSION_FENCE_TABLES {
for action in ["INSERT", "UPDATE"] {
connection
.execute_batch(&format!(
"CREATE TRIGGER permission_sealed_{table}_{action} BEFORE {action} ON {table}
WHEN EXISTS(SELECT 1 FROM permission_authority_seal)
BEGIN SELECT RAISE(ABORT, 'legacy permission authority is sealed'); END;"
))
.unwrap();
}
connection
.execute_batch(&format!(
"CREATE TRIGGER permission_sealed_{table}_DELETE BEFORE DELETE ON {table}
WHEN EXISTS(SELECT 1 FROM permission_authority_seal)
AND NOT EXISTS(SELECT 1 FROM subject_deletions WHERE subject_id=OLD.subject_id)
BEGIN SELECT RAISE(ABORT, 'legacy permission authority is sealed'); END;"
))
.unwrap();
}
connection.execute_batch(
"INSERT INTO subject_creation_receipts VALUES('creation','historical-subject','fingerprint',' {\"execution\":[]} ');
INSERT INTO permission_authority_seal VALUES(1,'transfer',' {\"opaque_saved_export\":true} ');"
).unwrap();
}
fn scope_fixture(connection: &Connection) {
connection.execute_batch(FIXED_SCOPE_SQL).unwrap();
// Opaque saved payloads prove byte retention, not valid domain release authority.
connection.execute_batch(
"INSERT INTO journal_final_scopes VALUES('fixed','remove','binding','owner-json','members',1,1,'session',NULL);
INSERT INTO journal_final_scope_sessions VALUES('fixed','session','metadata-json');
INSERT INTO journal_final_scope_pages VALUES('fixed','','session','page-digest');
UPDATE journal_final_scopes SET scope_digest='frozen' WHERE request_id='fixed';
INSERT INTO journal_scope_release_receipts VALUES('fixed','release-json');"
).unwrap();
}
fn cancellation_fixture(connection: &Connection) {
let before: Vec<_> = all_rows(connection)
.into_iter()
.filter(|(name, _)| {
name.starts_with("journal_final_scope") || name == "journal_scope_release_receipts"
})
.collect();
connection.execute_batch(CANCELLATION_SQL).unwrap();
for (name, rows) in before {
let after = all_rows(connection);
assert_eq!(
after.iter().find(|(table, _)| table == &name).unwrap().1,
rows
);
}
connection
.execute_batch(
r#"INSERT INTO journal_scope_cancellations VALUES('subject-release:cancel',
'{"request_id":"subject-release:cancel","removal_operation_id":"cancel"}');"#,
)
.unwrap();
}
fn consolidation_fixture(connection: &Connection) {
connection.execute_batch(CONSOLIDATION_SQL).unwrap();
connection.execute_batch(
"INSERT INTO consolidation_write_bindings VALUES('published-subjektiv-formats','job','attempt','subject','runtime','worker','input','grant','deadline','revoked');"
).unwrap();
}
fn record(connection: &Connection, version: u32, name: &str) {
connection
.execute(
&format!("INSERT INTO {HISTORY_TABLE}(version,name) VALUES(?1,?2)"),
params![version, name],
)
.unwrap();
}
fn install_published_fixture(connection: &Connection, format: Format) {
match format {
Format::Permission18 => {
permission_fixture(connection);
record(connection, 18, PERMISSION_18);
}
Format::Consolidation18 | Format::Consolidation20 => {
consolidation_fixture(connection);
record(connection, 18, CONSOLIDATION_18);
if format == Format::Consolidation20 {
scope_fixture(connection);
cancellation_fixture(connection);
record(connection, 20, CONVERGENCE_20);
}
}
Format::Fixed18 | Format::Cancellation19 | Format::Fixed20 | Format::Cancellation20 => {
scope_fixture(connection);
record(connection, 18, FIXED_18);
if format != Format::Fixed18 {
cancellation_fixture(connection);
if matches!(format, Format::Cancellation19 | Format::Cancellation20) {
record(connection, 19, CANCELLATION_19);
}
}
if matches!(format, Format::Fixed20 | Format::Cancellation20) {
consolidation_fixture(connection);
record(connection, 20, CONVERGENCE_20);
}
}
Format::Common(_) => unreachable!("common fixtures use their original registration"),
}
}
fn published_fixture(root: &Path, format: Format) {
let (scope, database) = open(
root,
FeatureRegistration::new(SUBJEKTIV_FEATURE_ID, &MIGRATIONS[..COMMON_COUNT]),
);
let store = SubjektivStore::from_database(database.clone()).unwrap();
store
.create_subject_with_behavior(
SubjectRole::new("retained").unwrap(),
"原文: retained\n".into(),
)
.unwrap();
database.transaction(|tx| {
// Permanent erasure evidence is deliberately not tied to a live Subject FK.
tx.execute_batch(
"INSERT INTO journal_erased_sessions VALUES('erased-session','erased-subject','erasure','erased-at');
INSERT INTO subject_deletions VALUES('erased-subject','delete-request',1,'started-at');
INSERT INTO subject_deleted_workers VALUES('erased-subject','old-runtime','old-worker');
INSERT INTO subject_deleted_sessions VALUES('erased-subject','erased-session');"
)?;
install_published_fixture(tx, format);
mark_receipts(tx);
Ok(())
}).unwrap();
drop(store);
drop(database);
scope.shutdown().unwrap();
}
fn database_path(root: &Path) -> std::path::PathBuf {
root.join(SCOPE).join("features/subjektiv.sqlite")
}
#[test]
fn published_families_reopen_without_changing_shapes_rows_or_original_receipts() {
for format in PUBLISHED {
let temp = tempfile::tempdir().unwrap();
published_fixture(temp.path(), format);
let mut original = Connection::open(database_path(temp.path())).unwrap();
assert_valid(&mut original);
assert_eq!(recognize(&history(&original)).unwrap(), format);
let saved_shape = domain_shape(&original).unwrap();
let saved_rows = all_rows(&original);
let saved_receipts = receipts(&original);
drop(original);
// This goes through the real registered migration engine, not a test-only planner.
// There is intentionally no new storage/domain migration or convergence receipt.
for _ in 0..2 {
let (scope, database) = open(temp.path(), REGISTRATION);
database
.with_connection(|connection| {
assert_eq!(domain_shape(connection)?, saved_shape, "{format:?}");
assert_eq!(all_rows(connection), saved_rows, "{format:?}");
assert_eq!(receipts(connection), saved_receipts, "{format:?}");
assert_foreign_keys(connection);
Ok(())
})
.unwrap();
drop(database);
scope.shutdown().unwrap();
}
}
}
#[test]
fn common_prefixes_migrate_to_fresh_shape_and_keep_saved_receipts_and_oldest_data() {
let fresh = tempfile::tempdir().unwrap();
let (fresh_scope, fresh_db) = open(fresh.path(), REGISTRATION);
let fresh_shape = fresh_db.with_connection(domain_shape).unwrap();
drop(fresh_db);
fresh_scope.shutdown().unwrap();
for count in 0..=COMMON_COUNT {
let temp = tempfile::tempdir().unwrap();
let (old_scope, old_db) = open(
temp.path(),
FeatureRegistration::new(SUBJEKTIV_FEATURE_ID, &MIGRATIONS[..count]),
);
let original_json = serde_json::json!({
"schema_version": 1, "id": "oldest", "role": "preserved",
"state": "active", "store_revision": 0,
"created_at": "created", "updated_at": "updated"
})
.to_string();
old_db.with_connection(|connection| {
if count > 0 {
connection.execute("INSERT INTO store_scope VALUES(1,?1)", [SCOPE])?;
}
if count == 1 {
connection.execute(
"INSERT INTO subjects VALUES('oldest','preserved','active',0,?1,'created','updated')",
[&original_json],
)?;
}
mark_receipts(connection);
Ok(())
}).unwrap();
let saved = old_db.with_connection(|c| Ok(receipts(c))).unwrap();
drop(old_db);
old_scope.shutdown().unwrap();
let (scope, db) = open(temp.path(), REGISTRATION);
db.with_connection(|connection| {
assert_eq!(domain_shape(connection)?, fresh_shape, "prefix {count}");
assert_eq!(&receipts(connection)[..count], saved.as_slice());
assert_foreign_keys(connection);
if count > 0 {
assert_eq!(
connection
.query_row("SELECT workspace_id FROM store_scope", [], |row| row
.get::<_, String>(0))?,
SCOPE
);
}
if count == 1 {
let archive: String = connection.query_row(
"SELECT row_json FROM legacy_record_archive WHERE source_table='subjects'",
[],
|row| row.get(0),
)?;
let archived: serde_json::Value = serde_json::from_str(&archive).unwrap();
assert_eq!(archived["record_json"], original_json);
}
Ok(())
})
.unwrap();
if count == 1 {
let store = SubjektivStore::from_database(db.clone()).unwrap();
let subject = store.subject("oldest").unwrap().unwrap();
assert_eq!(subject.role.as_str(), "preserved");
}
drop(db);
scope.shutdown().unwrap();
}
}
#[test]
fn unknown_mixed_missing_receipts_and_shape_mismatches_refuse_without_mutation() {
let cases = [
(
Format::Permission18,
"UPDATE __yoi_feature_schema_migrations SET name='unknown' WHERE version=18",
),
(
Format::Permission18,
"DELETE FROM __yoi_feature_schema_migrations WHERE version=18",
),
(
Format::Permission18,
"DROP TRIGGER permission_sealed_subject_permissions_INSERT",
),
(
Format::Permission18,
"ALTER TABLE permission_authority_seal ADD COLUMN unexpected TEXT",
),
(
Format::Permission18,
"INSERT INTO __yoi_feature_schema_migrations(version,name) VALUES(20,'converge published Journal scope and consolidation write histories')",
),
(
Format::Consolidation18,
"UPDATE __yoi_feature_schema_migrations SET name='fixed Journal removal scopes' WHERE version=18",
),
(
Format::Consolidation18,
"INSERT INTO __yoi_feature_schema_migrations(version,name) VALUES(19,'retain exact Subject release scope cancellations')",
),
(
Format::Consolidation18,
"INSERT INTO __yoi_feature_schema_migrations(version,name) VALUES(20,'converge published Journal scope and consolidation write histories')",
),
(
Format::Consolidation18,
"DROP TRIGGER consolidation_write_binding_no_delete",
),
(Format::Fixed18, "DROP TABLE journal_scope_release_receipts"),
(
Format::Cancellation19,
"DELETE FROM __yoi_feature_schema_migrations WHERE version=19",
),
(
Format::Consolidation20,
"DELETE FROM __yoi_feature_schema_migrations WHERE version=18",
),
(
Format::Fixed20,
"UPDATE __yoi_feature_schema_migrations SET name='unknown provenance' WHERE version=20",
),
(
Format::Cancellation20,
"DELETE FROM __yoi_feature_schema_migrations WHERE version=3",
),
(
Format::Cancellation20,
"INSERT INTO __yoi_feature_schema_migrations(version,name) VALUES(21,'future')",
),
(
Format::Cancellation20,
"CREATE TABLE unknown_payload(secret BLOB)",
),
];
for (format, corrupt) in cases {
let temp = tempfile::tempdir().unwrap();
published_fixture(temp.path(), format);
let mut raw = Connection::open(database_path(temp.path())).unwrap();
raw.execute_batch(corrupt).unwrap();
// Snapshot AFTER deliberate corruption. Rejection must not repair or adopt it.
let saved_shape = domain_shape(&raw).unwrap();
let saved_rows = all_rows(&raw);
let tx = raw.transaction().unwrap();
let error = validate(&tx, &history(&tx)).unwrap_err();
assert!(
matches!(error, FeatureStorageError::MigrationHistory { .. }),
"{format:?}: {corrupt}: {error}"
);
drop(tx);
drop(raw);
let scope = FeatureStorage::new(temp.path()).scope(SCOPE).unwrap();
let registered = scope.register(REGISTRATION).unwrap();
let result = scope.open(&registered);
assert!(
matches!(result, Err(FeatureStorageError::MigrationHistory { .. })),
"{format:?}: {corrupt}"
);
scope.shutdown().unwrap();
let raw = Connection::open(database_path(temp.path())).unwrap();
assert_eq!(domain_shape(&raw).unwrap(), saved_shape);
assert_eq!(all_rows(&raw), saved_rows);
}
}
#[test]
fn mixed_known_families_are_not_inferred_from_column_presence() {
let temp = tempfile::tempdir().unwrap();
published_fixture(temp.path(), Format::Permission18);
let mut raw = Connection::open(database_path(temp.path())).unwrap();
consolidation_fixture(&raw);
let before = all_rows(&raw);
let shape = domain_shape(&raw).unwrap();
let tx = raw.transaction().unwrap();
let error = validate(&tx, &history(&tx)).unwrap_err();
assert!(error.to_string().contains("SQLite shape"));
drop(tx);
assert_eq!(all_rows(&raw), before);
assert_eq!(domain_shape(&raw).unwrap(), shape);
drop(raw);
let scope = FeatureStorage::new(temp.path()).scope(SCOPE).unwrap();
let registered = scope.register(REGISTRATION).unwrap();
assert!(matches!(
scope.open(&registered),
Err(FeatureStorageError::MigrationHistory { .. })
));
scope.shutdown().unwrap();
let raw = Connection::open(database_path(temp.path())).unwrap();
assert_eq!(all_rows(&raw), before);
assert_eq!(domain_shape(&raw).unwrap(), shape);
}
#[test]
fn nonempty_schema_without_receipt_table_is_not_treated_as_new() {
let temp = tempfile::tempdir().unwrap();
published_fixture(temp.path(), Format::Permission18);
let mut raw = Connection::open(database_path(temp.path())).unwrap();
raw.execute_batch("DROP TABLE __yoi_feature_schema_migrations;")
.unwrap();
let before = all_rows(&raw);
let shape = domain_shape(&raw).unwrap();
let tx = raw.transaction().unwrap();
assert!(matches!(
validate(&tx, &[]),
Err(FeatureStorageError::MigrationHistory { .. })
));
drop(tx);
drop(raw);
let scope = FeatureStorage::new(temp.path()).scope(SCOPE).unwrap();
let registered = scope.register(REGISTRATION).unwrap();
assert!(matches!(
scope.open(&registered),
Err(FeatureStorageError::MigrationHistory { .. })
));
scope.shutdown().unwrap();
let raw = Connection::open(database_path(temp.path())).unwrap();
// Even the engine's CREATE TABLE IF NOT EXISTS must roll back after rejection.
assert_eq!(all_rows(&raw), before);
assert_eq!(domain_shape(&raw).unwrap(), shape);
}
#[test]
fn common_receipt_write_failure_rolls_back_in_sqlite_then_engine_retry_succeeds() {
let temp = tempfile::tempdir().unwrap();
let (old_scope, old_db) = open(
temp.path(),
FeatureRegistration::new(SUBJEKTIV_FEATURE_ID, &MIGRATIONS[..16]),
);
old_db
.with_connection(|connection| {
connection.execute("INSERT INTO store_scope VALUES(1,?1)", [SCOPE])?;
mark_receipts(connection);
Ok(())
})
.unwrap();
drop(old_db);
old_scope.shutdown().unwrap();
let raw = Connection::open(database_path(temp.path())).unwrap();
let before_shape = domain_shape(&raw).unwrap();
let before_rows = all_rows(&raw);
let saved_receipts = receipts(&raw);
drop(raw);
// Direct SQLite transaction coverage, not engine fault injection: public opens cannot
// install a TEMP trigger between validation and receipt insertion. Exercise the frozen
// migration plus receipt-write failure here, then retry through the actual engine.
let mut raw = Connection::open(database_path(temp.path())).unwrap();
raw.execute_batch("PRAGMA foreign_keys=ON;").unwrap();
raw.execute_batch(
"CREATE TEMP TRIGGER fail_receipt BEFORE INSERT ON main.__yoi_feature_schema_migrations
WHEN NEW.version=17 BEGIN SELECT RAISE(ABORT,'injected receipt failure'); END;",
)
.unwrap();
{
let tx = raw.transaction().unwrap();
validate(&tx, &history(&tx)).unwrap();
MIGRATIONS[16].apply(&tx).unwrap();
assert!(
tx.execute(
&format!("INSERT INTO {HISTORY_TABLE}(version,name) VALUES(?1,?2)"),
params![MIGRATIONS[16].version(), MIGRATIONS[16].name()],
)
.is_err()
);
}
assert_eq!(domain_shape(&raw).unwrap(), before_shape);
assert_eq!(all_rows(&raw), before_rows);
drop(raw);
let (scope, db) = open(temp.path(), REGISTRATION);
db.with_connection(|connection| {
assert_eq!(
domain_shape(connection)?,
reference_shape(Format::Common(17))?
);
assert_eq!(&receipts(connection)[..16], saved_receipts.as_slice());
Ok(())
})
.unwrap();
drop(db);
scope.shutdown().unwrap();
}
+432
View File
@@ -0,0 +1,432 @@
//! Physical compatibility with published Subjektiv stores, not adoption of domain authority.
//!
//! FeatureStorage calls `validate` in its migration transaction, before applying pending common
//! migrations. Published receipts and complete SQLite shapes must agree. Recognized extensions
//! stay physically intact: this module never installs a permission policy, transforms Job states,
//! or advances Journal scopes on the target. Those transitions belong to their domain owners.
use crate::{MIGRATIONS, SUBJEKTIV_FEATURE_ID};
use feature_storage::FeatureStorageError;
use rusqlite::{Connection, Transaction, params};
use std::sync::OnceLock;
const HISTORY_TABLE: &str = "__yoi_feature_schema_migrations";
const COMMON_COUNT: usize = 17;
const PERMISSION_18: &str =
"seal legacy permission authority and retain creation enrollment defaults";
const CONSOLIDATION_18: &str = "retain irreversible consolidation write bindings";
const FIXED_18: &str = "fixed Journal removal scopes";
const CANCELLATION_19: &str = "retain exact Subject release scope cancellations";
const CONVERGENCE_20: &str = "converge published Journal scope and consolidation write histories";
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum Format {
Common(usize),
Permission18,
Consolidation18,
Fixed18,
Cancellation19,
// Equal final shapes do not make their receipt histories interchangeable.
Consolidation20,
Fixed20,
Cancellation20,
}
fn incompatible(detail: impl Into<String>) -> FeatureStorageError {
FeatureStorageError::MigrationHistory {
feature: SUBJEKTIV_FEATURE_ID.into(),
detail: detail.into(),
}
}
fn recognize(applied: &[(u32, String)]) -> feature_storage::Result<Format> {
let count = applied.len().min(COMMON_COUNT);
if !applied[..count]
.iter()
.zip(&MIGRATIONS[..count])
.all(|((version, name), expected)| {
*version == expected.version() && name == expected.name()
})
{
return Err(incompatible(
"receipt prefix differs from published Subjektiv 1..17",
));
}
if applied.len() <= COMMON_COUNT {
return Ok(Format::Common(count));
}
let tail: Vec<_> = applied[COMMON_COUNT..]
.iter()
.map(|(version, name)| (*version, name.as_str()))
.collect();
match tail.as_slice() {
[(18, PERMISSION_18)] => Ok(Format::Permission18),
[(18, CONSOLIDATION_18)] => Ok(Format::Consolidation18),
[(18, FIXED_18)] => Ok(Format::Fixed18),
[(18, FIXED_18), (19, CANCELLATION_19)] => Ok(Format::Cancellation19),
[(18, CONSOLIDATION_18), (20, CONVERGENCE_20)] => Ok(Format::Consolidation20),
[(18, FIXED_18), (20, CONVERGENCE_20)] => Ok(Format::Fixed20),
[(18, FIXED_18), (19, CANCELLATION_19), (20, CONVERGENCE_20)] => Ok(Format::Cancellation20),
_ => Err(incompatible(
"unknown, missing or mixed published Subjektiv receipt tail",
)),
}
}
/// Validates physical compatibility only. The caller owns identity/adoption and the transaction.
/// Never relabel/delete original receipts, fill historical gaps, or append a domain receipt here.
/// After success the engine may apply only the pending registered common migrations (1..17).
pub(super) fn validate(
tx: &Transaction<'_>,
applied: &[(u32, String)],
) -> feature_storage::Result<()> {
let format = recognize(applied)?;
let expected = reference_shape(format)?;
let actual = domain_shape(tx)?;
if actual != expected {
let difference = actual
.iter()
.find(|object| !expected.contains(object))
.or_else(|| expected.iter().find(|object| !actual.contains(object)));
return Err(incompatible(format!(
"published receipts disagree with SQLite shape at {}",
difference.map_or("unknown object", |object| object.1.as_str())
)));
}
Ok(())
}
// Compare tables, indexes, views, triggers and their full SQL (including constraints), not
// merely column presence. Exact physical-owner metadata is excluded, never a wildcard prefix.
type SchemaObject = (String, String, String, Option<String>);
fn domain_shape(connection: &Connection) -> feature_storage::Result<Vec<SchemaObject>> {
let mut statement = connection.prepare(
"SELECT type,name,tbl_name,sql FROM sqlite_schema
WHERE name NOT GLOB 'sqlite_*'
AND name NOT IN ('__yoi_feature_schema_migrations', '__yoi_storage_identity', '__yoi_storage_origins', '__yoi_feature_database_identity', '__yoi_feature_provisioning_receipt')
ORDER BY type,name",
)?;
let shape = statement
.query_map([], |row| {
Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
})?
.collect::<std::result::Result<Vec<_>, _>>()?;
Ok(shape)
}
fn reference_shape(format: Format) -> feature_storage::Result<Vec<SchemaObject>> {
// Cache reference shapes only; every target is reread under its caller's transaction.
static SHAPES: [OnceLock<Vec<SchemaObject>>; 25] = [const { OnceLock::new() }; 25];
let index = match format {
Format::Common(count) => count,
Format::Permission18 => 18,
Format::Consolidation18 => 19,
Format::Fixed18 => 20,
Format::Cancellation19 => 21,
Format::Consolidation20 => 22,
Format::Fixed20 => 23,
Format::Cancellation20 => 24,
};
if let Some(shape) = SHAPES[index].get() {
return Ok(shape.clone());
}
let mut reference = Connection::open_in_memory()?;
reference.execute_batch("PRAGMA foreign_keys=ON;")?;
let tx = reference.transaction()?;
tx.execute_batch(
"CREATE TABLE __yoi_feature_schema_migrations (
version INTEGER PRIMARY KEY CHECK(version>0), name TEXT NOT NULL,
applied_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP);",
)?;
let common = match format {
Format::Common(count) => count,
_ => COMMON_COUNT,
};
for migration in &MIGRATIONS[..common] {
migration.apply(&tx)?;
tx.execute(
&format!("INSERT INTO {HISTORY_TABLE}(version,name) VALUES(?1,?2)"),
params![migration.version(), migration.name()],
)?;
}
// These frozen DDL fragments run ONLY on the disposable reference, never on a target.
// Preserve ALTER/rebuild provenance: sqlite_schema text can differ for equal columns.
match format {
Format::Common(_) => {}
Format::Permission18 => {
tx.execute_batch(PERMISSION_SEAL_SQL)?;
for table in PERMISSION_FENCE_TABLES {
for action in ["INSERT", "UPDATE"] {
tx.execute_batch(&format!(
"CREATE TRIGGER permission_sealed_{table}_{action} BEFORE {action} ON {table}
WHEN EXISTS(SELECT 1 FROM permission_authority_seal)
BEGIN SELECT RAISE(ABORT, 'legacy permission authority is sealed'); END;"
))?;
}
tx.execute_batch(&format!(
"CREATE TRIGGER permission_sealed_{table}_DELETE BEFORE DELETE ON {table}
WHEN EXISTS(SELECT 1 FROM permission_authority_seal)
AND NOT EXISTS(SELECT 1 FROM subject_deletions WHERE subject_id=OLD.subject_id)
BEGIN SELECT RAISE(ABORT, 'legacy permission authority is sealed'); END;"
))?;
}
}
Format::Consolidation18 => tx.execute_batch(CONSOLIDATION_SQL)?,
Format::Fixed18 => tx.execute_batch(FIXED_SCOPE_SQL)?,
Format::Cancellation19 => {
tx.execute_batch(FIXED_SCOPE_SQL)?;
tx.execute_batch(CANCELLATION_SQL)?;
}
Format::Consolidation20 | Format::Fixed20 | Format::Cancellation20 => {
tx.execute_batch(FIXED_SCOPE_SQL)?;
tx.execute_batch(CANCELLATION_SQL)?;
tx.execute_batch(CONSOLIDATION_SQL)?;
}
}
let shape = domain_shape(&tx)?;
let _ = SHAPES[index].set(shape.clone());
Ok(shape)
}
const PERMISSION_FENCE_TABLES: [&str; 4] = [
"subject_permissions",
"subject_permission_changes",
"subject_permission_receipts",
"worker_execution_grants",
];
// Frozen physical shape evidence, not target migrations. Permission SQL is from
// 9edebfa5936f2411adb6affffc793d3e93cc21bd:crates/subjektiv/src/permissions/authority.rs.
// Journal/consolidation SQL is from 7b39a44ca698c474287e5511f674f2e8baae4c03:
// crates/subjektiv/src/{consolidation_write_bindings.sql,journal/final_scope.sql,
// journal/final_scope/cancellation.sql}. Receipt provenance follows that ref's recognizer;
// no permission-seal -> 20 history is published there.
const PERMISSION_SEAL_SQL: &str = r#"CREATE TABLE permission_authority_seal (
singleton INTEGER PRIMARY KEY CHECK(singleton=1),
transfer_id TEXT NOT NULL CHECK(length(transfer_id)>0),
export_json TEXT NOT NULL
);
CREATE TRIGGER permission_authority_seal_immutable BEFORE UPDATE ON permission_authority_seal
BEGIN SELECT RAISE(ABORT, 'permission authority seal is immutable'); END;
CREATE TRIGGER permission_authority_seal_retained BEFORE DELETE ON permission_authority_seal
BEGIN SELECT RAISE(ABORT, 'permission authority seal is permanent'); END;
ALTER TABLE subject_creation_receipts ADD COLUMN permission_enrollment_json TEXT;
CREATE TRIGGER subject_creation_receipts_immutable BEFORE UPDATE ON subject_creation_receipts
BEGIN SELECT RAISE(ABORT, 'Subject creation receipts are immutable'); END;
CREATE TRIGGER subject_creation_receipts_retained BEFORE DELETE ON subject_creation_receipts
BEGIN SELECT RAISE(ABORT, 'Subject creation receipts are retained'); END;"#;
const CONSOLIDATION_SQL: &str = r#"-- Domain write fence only: no Job runner state and no Subject content.
-- Deliberately no Subject foreign key: erasure must retain this identity fence.
CREATE TABLE consolidation_write_bindings (
scope_id TEXT NOT NULL,
job_id TEXT NOT NULL,
attempt_id TEXT NOT NULL,
subject_id TEXT NOT NULL,
runtime_id TEXT NOT NULL,
worker_id TEXT NOT NULL,
input_digest TEXT NOT NULL,
grant_digest TEXT NOT NULL,
deadline_at TEXT NOT NULL,
revoked_at TEXT,
PRIMARY KEY (scope_id, job_id, attempt_id),
CHECK (revoked_at IS NULL OR length(revoked_at) > 0)
);
-- BEFORE INSERT also catches INSERT OR REPLACE, even with recursive_triggers off.
CREATE TRIGGER consolidation_write_binding_no_replace
BEFORE INSERT ON consolidation_write_bindings
WHEN EXISTS (SELECT 1 FROM consolidation_write_bindings
WHERE scope_id = NEW.scope_id AND job_id = NEW.job_id AND attempt_id = NEW.attempt_id)
BEGIN
SELECT RAISE(ABORT, 'consolidation write binding already exists');
END;
CREATE TRIGGER consolidation_write_binding_immutable
BEFORE UPDATE ON consolidation_write_bindings
WHEN NEW.scope_id IS NOT OLD.scope_id
OR NEW.job_id IS NOT OLD.job_id OR NEW.attempt_id IS NOT OLD.attempt_id
OR NEW.subject_id IS NOT OLD.subject_id OR NEW.runtime_id IS NOT OLD.runtime_id
OR NEW.worker_id IS NOT OLD.worker_id OR NEW.input_digest IS NOT OLD.input_digest
OR NEW.grant_digest IS NOT OLD.grant_digest OR NEW.deadline_at IS NOT OLD.deadline_at
OR (OLD.revoked_at IS NOT NULL AND NEW.revoked_at IS NOT OLD.revoked_at)
BEGIN
SELECT RAISE(ABORT, 'consolidation write binding is immutable or revoked');
END;
CREATE TRIGGER consolidation_write_binding_no_delete
BEFORE DELETE ON consolidation_write_bindings
BEGIN
SELECT RAISE(ABORT, 'consolidation write binding must be retained');
END;
"#;
const FIXED_SCOPE_SQL: &str = r#"-- Fixed removal scope metadata only; no source bodies.
-- No FK to live Subject/writer: retained identity evidence cannot block explicit deletion.
-- Every API rechecks the live Subject, exact binding and permanent erase fences.
CREATE TABLE journal_final_scopes (
request_id TEXT PRIMARY KEY,
removal_operation_id TEXT NOT NULL UNIQUE,
binding_id TEXT NOT NULL UNIQUE,
owner_json TEXT NOT NULL,
member_digest TEXT NOT NULL,
member_count INTEGER NOT NULL CHECK(member_count >= 0),
collected_count INTEGER NOT NULL DEFAULT 0 CHECK(collected_count >= 0 AND collected_count <= member_count),
after_session_id TEXT,
scope_digest TEXT
);
CREATE TRIGGER journal_final_scopes_guard BEFORE UPDATE ON journal_final_scopes
WHEN OLD.scope_digest IS NOT NULL OR NEW.request_id != OLD.request_id
OR NEW.removal_operation_id != OLD.removal_operation_id OR NEW.binding_id != OLD.binding_id
OR NEW.owner_json != OLD.owner_json OR NEW.member_digest != OLD.member_digest
OR NEW.member_count != OLD.member_count OR NEW.collected_count < OLD.collected_count
OR (NEW.scope_digest IS NOT NULL AND NEW.collected_count != NEW.member_count)
BEGIN SELECT RAISE(ABORT, 'fixed Journal scope owner and frozen header are immutable'); END;
CREATE TRIGGER journal_final_scopes_no_delete BEFORE DELETE ON journal_final_scopes
BEGIN SELECT RAISE(ABORT, 'fixed Journal scopes are retained'); END;
-- One normalized permanent metadata row per collected Session. No parallel full-scope JSON.
CREATE TABLE journal_final_scope_sessions (
request_id TEXT NOT NULL REFERENCES journal_final_scopes(request_id) ON DELETE RESTRICT,
session_id TEXT NOT NULL,
metadata_json TEXT NOT NULL,
PRIMARY KEY(request_id,session_id)
);
CREATE TRIGGER journal_final_scope_sessions_frozen_insert BEFORE INSERT ON journal_final_scope_sessions
WHEN (SELECT scope_digest FROM journal_final_scopes WHERE request_id=NEW.request_id) IS NOT NULL
BEGIN SELECT RAISE(ABORT, 'frozen Session set is immutable'); END;
CREATE TRIGGER journal_final_scope_sessions_no_update BEFORE UPDATE ON journal_final_scope_sessions
BEGIN SELECT RAISE(ABORT, 'fixed Session metadata is immutable'); END;
CREATE TRIGGER journal_final_scope_sessions_no_delete BEFORE DELETE ON journal_final_scope_sessions
BEGIN SELECT RAISE(ABORT, 'fixed Session metadata is retained'); END;
-- Page receipts contain identities only; replay metadata is read from normalized rows.
CREATE TABLE journal_final_scope_pages (
request_id TEXT NOT NULL REFERENCES journal_final_scopes(request_id) ON DELETE RESTRICT,
after_session_id TEXT NOT NULL,
through_session_id TEXT NOT NULL,
page_digest TEXT NOT NULL,
PRIMARY KEY(request_id,after_session_id)
);
CREATE TRIGGER journal_final_scope_pages_no_update BEFORE UPDATE ON journal_final_scope_pages
BEGIN SELECT RAISE(ABORT, 'scope page receipts are immutable'); END;
CREATE TRIGGER journal_final_scope_pages_no_delete BEFORE DELETE ON journal_final_scope_pages
BEGIN SELECT RAISE(ABORT, 'scope page receipts are retained'); END;
CREATE TABLE journal_scope_release_receipts (
request_id TEXT PRIMARY KEY REFERENCES journal_final_scopes(request_id) ON DELETE RESTRICT,
receipt_json TEXT NOT NULL
);
CREATE TRIGGER journal_scope_release_no_update BEFORE UPDATE ON journal_scope_release_receipts
BEGIN SELECT RAISE(ABORT, 'Journal release receipts are immutable'); END;
CREATE TRIGGER journal_scope_release_no_delete BEFORE DELETE ON journal_scope_release_receipts
BEGIN SELECT RAISE(ABORT, 'Journal release receipts are retained'); END;
"#;
const CANCELLATION_SQL: &str = r#"-- Migration 19 replaces only migration-18's unconditional binding UNIQUE with a live-owner
-- constraint. Transaction-local copies preserve every column/row. Published 18 is unchanged.
-- No Journal history, body input or checkpoint is removed or rewritten.
CREATE TEMP TABLE scope_cancel_copy_journal_final_scopes AS SELECT * FROM journal_final_scopes;
CREATE TEMP TABLE scope_cancel_copy_journal_final_scope_sessions AS SELECT * FROM journal_final_scope_sessions;
CREATE TEMP TABLE scope_cancel_copy_journal_final_scope_pages AS SELECT * FROM journal_final_scope_pages;
CREATE TEMP TABLE scope_cancel_copy_journal_scope_release_receipts AS SELECT * FROM journal_scope_release_receipts;
DROP TABLE journal_scope_release_receipts;
DROP TABLE journal_final_scope_pages;
DROP TABLE journal_final_scope_sessions;
DROP TABLE journal_final_scopes;
CREATE TABLE journal_final_scopes (
request_id TEXT PRIMARY KEY,
removal_operation_id TEXT NOT NULL UNIQUE,
binding_id TEXT NOT NULL,
owner_json TEXT NOT NULL,
member_digest TEXT NOT NULL,
member_count INTEGER NOT NULL CHECK(member_count >= 0),
collected_count INTEGER NOT NULL DEFAULT 0 CHECK(collected_count >= 0 AND collected_count <= member_count),
after_session_id TEXT,
scope_digest TEXT
);
CREATE TABLE journal_final_scope_sessions (
request_id TEXT NOT NULL REFERENCES journal_final_scopes(request_id) ON DELETE RESTRICT,
session_id TEXT NOT NULL,
metadata_json TEXT NOT NULL,
PRIMARY KEY(request_id,session_id)
);
CREATE TABLE journal_final_scope_pages (
request_id TEXT NOT NULL REFERENCES journal_final_scopes(request_id) ON DELETE RESTRICT,
after_session_id TEXT NOT NULL,
through_session_id TEXT NOT NULL,
page_digest TEXT NOT NULL,
PRIMARY KEY(request_id,after_session_id)
);
CREATE TABLE journal_scope_release_receipts (
request_id TEXT PRIMARY KEY REFERENCES journal_final_scopes(request_id) ON DELETE RESTRICT,
receipt_json TEXT NOT NULL
);
INSERT INTO journal_final_scopes SELECT * FROM scope_cancel_copy_journal_final_scopes;
DROP TABLE scope_cancel_copy_journal_final_scopes;
INSERT INTO journal_final_scope_sessions SELECT * FROM scope_cancel_copy_journal_final_scope_sessions;
DROP TABLE scope_cancel_copy_journal_final_scope_sessions;
INSERT INTO journal_final_scope_pages SELECT * FROM scope_cancel_copy_journal_final_scope_pages;
DROP TABLE scope_cancel_copy_journal_final_scope_pages;
INSERT INTO journal_scope_release_receipts SELECT * FROM scope_cancel_copy_journal_scope_release_receipts;
DROP TABLE scope_cancel_copy_journal_scope_release_receipts;
CREATE TRIGGER journal_final_scopes_guard BEFORE UPDATE ON journal_final_scopes
WHEN OLD.scope_digest IS NOT NULL OR NEW.request_id != OLD.request_id
OR NEW.removal_operation_id != OLD.removal_operation_id OR NEW.binding_id != OLD.binding_id
OR NEW.owner_json != OLD.owner_json OR NEW.member_digest != OLD.member_digest
OR NEW.member_count != OLD.member_count OR NEW.collected_count < OLD.collected_count
OR (NEW.scope_digest IS NOT NULL AND NEW.collected_count != NEW.member_count)
BEGIN SELECT RAISE(ABORT, 'fixed Journal scope owner and frozen header are immutable'); END;
CREATE TRIGGER journal_final_scopes_no_delete BEFORE DELETE ON journal_final_scopes
BEGIN SELECT RAISE(ABORT, 'fixed Journal scopes are retained'); END;
CREATE TRIGGER journal_final_scope_sessions_frozen_insert BEFORE INSERT ON journal_final_scope_sessions
WHEN (SELECT scope_digest FROM journal_final_scopes WHERE request_id=NEW.request_id) IS NOT NULL
BEGIN SELECT RAISE(ABORT, 'frozen Session set is immutable'); END;
CREATE TRIGGER journal_final_scope_sessions_no_update BEFORE UPDATE ON journal_final_scope_sessions
BEGIN SELECT RAISE(ABORT, 'fixed Session metadata is immutable'); END;
CREATE TRIGGER journal_final_scope_sessions_no_delete BEFORE DELETE ON journal_final_scope_sessions
BEGIN SELECT RAISE(ABORT, 'fixed Session metadata is retained'); END;
CREATE TRIGGER journal_final_scope_pages_no_update BEFORE UPDATE ON journal_final_scope_pages
BEGIN SELECT RAISE(ABORT, 'scope page receipts are immutable'); END;
CREATE TRIGGER journal_final_scope_pages_no_delete BEFORE DELETE ON journal_final_scope_pages
BEGIN SELECT RAISE(ABORT, 'scope page receipts are retained'); END;
CREATE TRIGGER journal_scope_release_no_update BEFORE UPDATE ON journal_scope_release_receipts
BEGIN SELECT RAISE(ABORT, 'Journal release receipts are immutable'); END;
CREATE TRIGGER journal_scope_release_no_delete BEFORE DELETE ON journal_scope_release_receipts
BEGIN SELECT RAISE(ABORT, 'Journal release receipts are retained'); END;
-- One immutable cancellation fact with its parent identity, including before collection.
CREATE TABLE journal_scope_cancellations (
request_id TEXT PRIMARY KEY,
owner_json TEXT NOT NULL
);
CREATE TRIGGER journal_scope_cancellations_no_update BEFORE UPDATE ON journal_scope_cancellations
BEGIN SELECT RAISE(ABORT, 'scope cancellation is immutable'); END;
CREATE TRIGGER journal_scope_cancellations_no_delete BEFORE DELETE ON journal_scope_cancellations
BEGIN SELECT RAISE(ABORT, 'scope cancellation is retained'); END;
CREATE TRIGGER journal_scope_cancellation_guard BEFORE INSERT ON journal_scope_cancellations
WHEN NEW.request_id != 'subject-release:' || json_extract(NEW.owner_json,'$.removal_operation_id')
OR NEW.request_id != json_extract(NEW.owner_json,'$.request_id')
OR EXISTS(SELECT 1 FROM journal_scope_release_receipts WHERE request_id=NEW.request_id)
OR EXISTS(SELECT 1 FROM journal_final_scopes s WHERE s.request_id=NEW.request_id AND s.owner_json!=NEW.owner_json)
BEGIN SELECT RAISE(ABORT, 'only exact unreleased Subject release scope can be cancelled'); END;
CREATE TRIGGER journal_final_scopes_live_owner BEFORE INSERT ON journal_final_scopes
WHEN EXISTS(SELECT 1 FROM journal_scope_cancellations c WHERE c.request_id=NEW.request_id)
OR EXISTS(SELECT 1 FROM journal_final_scopes s WHERE s.binding_id=NEW.binding_id
AND NOT EXISTS(SELECT 1 FROM journal_scope_cancellations c WHERE c.request_id=s.request_id))
BEGIN SELECT RAISE(ABORT, 'fixed scope writer is occupied or request cancelled'); END;
CREATE TRIGGER journal_scope_release_not_cancelled BEFORE INSERT ON journal_scope_release_receipts
WHEN EXISTS(SELECT 1 FROM journal_scope_cancellations c WHERE c.request_id=NEW.request_id)
BEGIN SELECT RAISE(ABORT, 'cancelled scope cannot release'); END;
CREATE TRIGGER journal_final_scope_cancelled_update BEFORE UPDATE ON journal_final_scopes
WHEN EXISTS(SELECT 1 FROM journal_scope_cancellations c WHERE c.request_id=OLD.request_id)
BEGIN SELECT RAISE(ABORT, 'cancelled scope cannot continue'); END;
CREATE TRIGGER journal_final_scope_cancelled_session BEFORE INSERT ON journal_final_scope_sessions
WHEN EXISTS(SELECT 1 FROM journal_scope_cancellations c WHERE c.request_id=NEW.request_id)
BEGIN SELECT RAISE(ABORT, 'cancelled scope cannot continue'); END;
CREATE TRIGGER journal_final_scope_cancelled_page BEFORE INSERT ON journal_final_scope_pages
WHEN EXISTS(SELECT 1 FROM journal_scope_cancellations c WHERE c.request_id=NEW.request_id)
BEGIN SELECT RAISE(ABORT, 'cancelled scope cannot continue'); END;
"#;
#[cfg(test)]
#[path = "storage_format_tests.rs"]
mod tests;
+2 -1
View File
@@ -473,8 +473,9 @@ fn stopped_writer_backup_restores_ids_last_mutation_ids_receipts_and_sequence_wi
std::fs::copy(file.path(), restored_blobs.join(file.file_name())).unwrap(); std::fs::copy(file.path(), restored_blobs.join(file.file_name())).unwrap();
} }
let restored = FeatureStorage::new(t.path().join("restored/metadata")); let restored = FeatureStorage::new(t.path().join("restored/metadata"));
let source_scope = scope;
let scope = restored.workspace("ws").unwrap(); let scope = restored.workspace("ws").unwrap();
scope.restore(&snapshot).unwrap(); scope.restore_from(&snapshot, &source_scope).unwrap();
let registered = Drive::register(&scope).unwrap(); let registered = Drive::register(&scope).unwrap();
let d = Drive::open(&scope, &registered, &restored_blobs).unwrap(); let d = Drive::open(&scope, &registered, &restored_blobs).unwrap();
assert_eq!(d.metadata(latest.id).unwrap(), latest); assert_eq!(d.metadata(latest.id).unwrap(), latest);
@@ -470,7 +470,7 @@ fn migrated_workdir_request_replays_and_retries_without_accepting_changed_displa
rename_saved_secret_receipt_counters(&conn).unwrap(); rename_saved_secret_receipt_counters(&conn).unwrap();
let store = SqliteWorkspaceStore { let store = SqliteWorkspaceStore {
db_executor: crate::db_executor::DbExecutor::shared(), db_executor: crate::db_executor::DbExecutor::shared(),
conn: Arc::new(Mutex::new(conn)), conn: StoreConnection::Fixture(Arc::new(Mutex::new(conn))),
}; };
let migrated = store let migrated = store
.load_workdir_create_operation("space", "create-workdir") .load_workdir_create_operation("space", "create-workdir")
@@ -654,7 +654,7 @@ fn deleted_successful_workdir_archives_verbatim_history_without_authorizing_retr
rename_saved_secret_receipt_counters(&conn).unwrap(); rename_saved_secret_receipt_counters(&conn).unwrap();
let store = Arc::new(SqliteWorkspaceStore { let store = Arc::new(SqliteWorkspaceStore {
db_executor: crate::db_executor::DbExecutor::shared(), db_executor: crate::db_executor::DbExecutor::shared(),
conn: Arc::new(Mutex::new(conn)), conn: StoreConnection::Fixture(Arc::new(Mutex::new(conn))),
}); });
let archived = store let archived = store
.load_workdir_create_legacy_ssh_archive("space", "create-workdir") .load_workdir_create_legacy_ssh_archive("space", "create-workdir")
@@ -1030,7 +1030,7 @@ fn migrated_ssh_mutations_replay_complete_old_input_without_reexecution_or_recei
rename_saved_secret_receipt_counters(&conn).unwrap(); rename_saved_secret_receipt_counters(&conn).unwrap();
let store = Arc::new(SqliteWorkspaceStore { let store = Arc::new(SqliteWorkspaceStore {
db_executor: crate::db_executor::DbExecutor::shared(), db_executor: crate::db_executor::DbExecutor::shared(),
conn: Arc::new(Mutex::new(conn)), conn: StoreConnection::Fixture(Arc::new(Mutex::new(conn))),
}); });
let service = let service =
crate::repository_access::RepositorySecretService::open(store.clone(), &database).unwrap(); crate::repository_access::RepositorySecretService::open(store.clone(), &database).unwrap();
@@ -1183,7 +1183,7 @@ fn deleted_legacy_receipt_does_not_bind_to_later_reused_key_ordinal() {
rename_saved_secret_receipt_counters(&conn).unwrap(); rename_saved_secret_receipt_counters(&conn).unwrap();
let store = Arc::new(SqliteWorkspaceStore { let store = Arc::new(SqliteWorkspaceStore {
db_executor: crate::db_executor::DbExecutor::shared(), db_executor: crate::db_executor::DbExecutor::shared(),
conn: Arc::new(Mutex::new(conn)), conn: StoreConnection::Fixture(Arc::new(Mutex::new(conn))),
}); });
let service = let service =
crate::repository_access::RepositorySecretService::open(store.clone(), &database).unwrap(); crate::repository_access::RepositorySecretService::open(store.clone(), &database).unwrap();
@@ -534,7 +534,7 @@ mod existing_workspace_provisioning {
// unrelated full Server schema. Exercise the real service/store methods. // unrelated full Server schema. Exercise the real service/store methods.
Arc::new(SqliteWorkspaceStore { Arc::new(SqliteWorkspaceStore {
db_executor: crate::db_executor::DbExecutor::shared(), db_executor: crate::db_executor::DbExecutor::shared(),
conn: Arc::new(Mutex::new(conn)), conn: StoreConnection::Fixture(Arc::new(Mutex::new(conn))),
}) })
} }
@@ -646,7 +646,7 @@ mod existing_workspace_provisioning {
drop(store); drop(store);
let store = Arc::new(SqliteWorkspaceStore { let store = Arc::new(SqliteWorkspaceStore {
db_executor: crate::db_executor::DbExecutor::shared(), db_executor: crate::db_executor::DbExecutor::shared(),
conn: Arc::new(Mutex::new(Connection::open(&path).unwrap())), conn: StoreConnection::Fixture(Arc::new(Mutex::new(Connection::open(&path).unwrap()))),
}); });
let service = WorkspaceSigningIdentityService::new( let service = WorkspaceSigningIdentityService::new(
store.clone(), store.clone(),
@@ -530,7 +530,7 @@ fn implementation_legacy96_fixture() -> (tempfile::TempDir, SqliteWorkspaceStore
INSERT INTO worker_singleton_owners(workspace_id,singleton_key,runtime_id,worker_id,generation,created_at,updated_at) VALUES('w','workspace-orchestrator','r','worker',1,'1','1'); INSERT INTO worker_singleton_owners(workspace_id,singleton_key,runtime_id,worker_id,generation,created_at,updated_at) VALUES('w','workspace-orchestrator','r','worker',1,'1','1');
INSERT INTO repositories(workspace_id,repository_id,repository_key,kind,provider,uri,created_at,updated_at,source_kind,source_uri,source_fingerprint) VALUES('w','main','main','git','git','file:///main','1','1','file','file:///main','sha256:fixture');").unwrap(); INSERT INTO repositories(workspace_id,repository_id,repository_key,kind,provider,uri,created_at,updated_at,source_kind,source_uri,source_fingerprint) VALUES('w','main','main','git','git','file:///main','1','1','file','file:///main','sha256:fixture');").unwrap();
let store = SqliteWorkspaceStore { let store = SqliteWorkspaceStore {
conn: Arc::new(Mutex::new(conn)), conn: StoreConnection::Fixture(Arc::new(Mutex::new(conn))),
db_executor: crate::db_executor::DbExecutor::new( db_executor: crate::db_executor::DbExecutor::new(
crate::db_executor::DbExecutorLimits::default(), crate::db_executor::DbExecutorLimits::default(),
) )
+15
View File
@@ -4661,6 +4661,7 @@ impl WorkspaceApi {
} }
let database_path = config.database_path.clone(); let database_path = config.database_path.clone();
let workspace_id = config.workspace_id.clone(); let workspace_id = config.workspace_id.clone();
let storage_catalog = config_store.clone();
let (feature_storage, subjektiv_registration, authority) = store let (feature_storage, subjektiv_registration, authority) = store
.db_executor() .db_executor()
.run(move || { .run(move || {
@@ -4673,12 +4674,26 @@ impl WorkspaceApi {
.map_err(|error| { .map_err(|error| {
Error::Store(format!("failed to register subjektiv storage: {error}")) Error::Store(format!("failed to register subjektiv storage: {error}"))
})?; })?;
storage_catalog.adopt_workspace_feature_storage(
&feature_storage,
&registration,
"store_scope",
)?;
let authority = database_preparation.open_authority(database_path, workspace_id)?; let authority = database_preparation.open_authority(database_path, workspace_id)?;
Ok::<_, Error>((feature_storage, registration, authority)) Ok::<_, Error>((feature_storage, registration, authority))
}) })
.await?; .await?;
let drive_host = let drive_host =
Arc::new(crate::workspace_drive_host::WorkspaceDriveHost::new(&feature_storage).await?); Arc::new(crate::workspace_drive_host::WorkspaceDriveHost::new(&feature_storage).await?);
{
let catalog = config_store.clone();
let scope = feature_storage.clone();
let host = drive_host.clone();
store
.db_executor()
.run(move || host.adopt_legacy_storage(&scope, &catalog))
.await?;
}
let mut api = Self { let mut api = Self {
host_executors, host_executors,
drive_host, drive_host,
@@ -104,7 +104,7 @@ async fn scoped_subject_database_rereads_policy_changed_on_independent_connectio
} }
#[tokio::test] #[tokio::test]
async fn scoped_subject_database_does_not_survive_deletion_or_refresh_itself_after_restore() { async fn scoped_subject_database_observes_irreversible_erasure_after_check_finishes() {
let f = Fixture::new(None).await; let f = Fixture::new(None).await;
let snapshot_dir = tempfile::tempdir().unwrap(); let snapshot_dir = tempfile::tempdir().unwrap();
let snapshot = snapshot_dir.path().join("snapshot"); let snapshot = snapshot_dir.path().join("snapshot");
@@ -112,25 +112,19 @@ async fn scoped_subject_database_does_not_survive_deletion_or_refresh_itself_aft
let request = permission(&f); let request = permission(&f);
plugin_host::with_authority_database(&f.api, || { plugin_host::with_authority_database(&f.api, || {
subject_authorization::authorize_scope(&f.api, &f.source, &request).unwrap(); subject_authorization::authorize_scope(&f.api, &f.source, &request).unwrap();
// The guard retains no connection lock: deletion completes synchronously.
f.api.feature_storage.delete().unwrap();
assert!(subject_authorization::authorize_scope(&f.api, &f.source, &request).is_err());
assert!(f.api.feature_storage.restore(&snapshot).is_err());
let fresh =
crate::FeatureStorage::for_server_database(&f.api.config.database_path).unwrap();
let scope = fresh.workspace(TEST_WORKSPACE_ID).unwrap();
scope.restore(&snapshot).unwrap();
let registration = crate::subjektiv::SubjektivStore::register(&scope).unwrap();
let mut restored = f.api.clone();
restored.feature_storage = scope;
restored.subjektiv_registration = registration;
// A fresh Host binding can read the restored DB; the old binding stays fenced.
subject_authorization::authorize_scope(&restored, &f.source, &request).unwrap();
assert!(subject_authorization::authorize_scope(&f.api, &f.source, &request).is_err());
restored.feature_storage.shutdown().unwrap();
Ok(()) Ok(())
}) }).unwrap();
.unwrap(); // The short check's lifecycle guard is released before destructive maintenance.
f.api.feature_storage.delete().unwrap();
assert!(subject_authorization::authorize_scope(&f.api, &f.source, &request).is_err());
assert!(f.api.feature_storage.restore(&snapshot).is_err());
let fresh = crate::FeatureStorage::for_server_database(&f.api.config.database_path).unwrap();
let scope = fresh.workspace(TEST_WORKSPACE_ID).unwrap();
assert!(scope.restore(&snapshot).is_err());
let another = crate::FeatureStorage::new(snapshot_dir.path().join("another-owner"));
assert!(another.workspace(TEST_WORKSPACE_ID).unwrap()
.restore_from(&snapshot, &scope).is_err());
assert!(subject_authorization::authorize_scope(&f.api, &f.source, &request).is_err());
f.shutdown().await; f.shutdown().await;
} }
@@ -141,10 +141,13 @@ pub(super) fn with_authority_database<T>(
api: &WorkspaceApi, api: &WorkspaceApi,
check: impl FnOnce() -> crate::Result<T>, check: impl FnOnce() -> crate::Result<T>,
) -> crate::Result<T> { ) -> crate::Result<T> {
let _database = api let database = api
.feature_storage .feature_storage
.open(&api.subjektiv_registration) .open(&api.subjektiv_registration)
.map_err(|error| Error::Store(error.to_string()))?; .map_err(|error| Error::Store(error.to_string()))?;
let _guard = database
.guard(database.identity())
.map_err(|error| Error::Store(error.to_string()))?;
check() check()
} }
@@ -0,0 +1,362 @@
//! Storage component tests: real retained migration functions and published SQL
//! on isolated SQLite databases; no authority, Runtime or domain recovery fakes.
//! The normal workspace-server lib test target runs this module. These fixtures
//! attest physical acceptance/data retention, not policy transfer or Job recovery.
use super::*;
use rusqlite::{backup::Backup, params};
fn save_history(
conn: &Connection,
stage: Stage,
baseline: i64,
predecessor: Option<(Family, usize)>,
) {
conn.execute("DELETE FROM __yoi_schema_migrations", [])
.unwrap();
conn.execute(
"INSERT INTO __yoi_schema_migrations(version,name) VALUES(?1,?2)",
params![baseline, super::super::SCHEMA_BASELINE_NAME],
)
.unwrap();
let published = match stage {
Stage::Published(family, count) => Some((family, count)),
_ => predecessor,
};
if let Some((family, count)) = published {
for (index, name) in family.names().iter().take(count).enumerate() {
let version = 114 + index as i64;
if version > baseline {
conn.execute(
"INSERT INTO __yoi_schema_migrations(version,name) VALUES(?1,?2)",
params![version, name],
)
.unwrap();
}
}
}
if let Stage::Converged(current) = stage {
for (version, name) in [
(119, CONVERGED),
(120, PRIOR_RELEASE),
(121, CLEANUP_COVERAGE),
] {
if version > baseline && version <= current {
conn.execute(
"INSERT INTO __yoi_schema_migrations(version,name) VALUES(?1,?2)",
params![version, name],
)
.unwrap();
}
}
}
}
fn fixture(stage: Stage, baseline: i64, predecessor: Option<(Family, usize)>) -> Connection {
let conn = reference_base().unwrap();
install_reference(&conn, stage).unwrap();
save_history(&conn, stage, baseline, predecessor);
conn.execute_batch("INSERT INTO accounts(account_id,kind,handle,display_name,created_at,updated_at)
VALUES('owner','user','owner','Owner','1','1');
INSERT INTO workspaces(workspace_id,owner_account_id,display_name,state,created_at,updated_at)
VALUES('workspace-a','owner','Saved workspace','active','1','1');
INSERT INTO backend_jobs(workspace_id,job_id,purpose,input_digest,input_ref,request_json,intent_fingerprint,state,current_attempt,created_at,updated_at,input_disposition)
VALUES('workspace-a','saved-job','journal','digest','input','{}','fingerprint','pending',1,'1','1','erased');
INSERT INTO backend_job_attempts(workspace_id,job_id,attempt_id,attempt,input_digest,state,deadline_at,created_at,updated_at)
VALUES('workspace-a','saved-job','saved-attempt',1,'digest','reserved','2','1','1');
INSERT INTO worker_removal_operations(operation_id,plan_id,input_fingerprint,workspace_id,runtime_id,worker_id,worker_updated_at,policy_id,policy_digest,session_disposition,metadata_disposition,archive_retention_kind,diagnostics_disposition,blockers_json,state,reason,created_at,updated_at)
VALUES('saved-removal','saved-plan','saved-fingerprint','workspace-a','runtime','worker','1','policy','digest','erase','erase','none','erase','[]','succeeded','saved','1','1');").unwrap();
if super::super::table_exists(&conn, "permission_policy_receipts").unwrap() {
conn.execute_batch("INSERT INTO permission_policy_receipts(workspace_id,operation_id,fingerprint,result_json)
VALUES('workspace-a','saved-permission','saved-fingerprint','{\"saved\":true}');").unwrap();
}
if super::super::table_exists(&conn, "server_feature_root").unwrap() {
conn.execute_batch("INSERT INTO server_feature_root(root_slot,reservation_id,request_fingerprint,state,accepted_receipt)
VALUES(1,'saved-reservation','saved-fingerprint','accepted','saved-root-receipt');
INSERT INTO content_history_targets(workspace,target,head,erased)
VALUES('workspace-a','workspace-config',NULL,1);").unwrap();
}
conn
}
fn saved_rows(conn: &Connection) -> Vec<(String, Vec<String>)> {
let names = conn.prepare("SELECT name FROM sqlite_schema WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name").unwrap()
.query_map([], |row| row.get::<_, String>(0)).unwrap()
.collect::<rusqlite::Result<Vec<_>>>().unwrap();
names
.into_iter()
.map(|name| {
let mut statement = conn
.prepare(&format!("SELECT * FROM \"{}\"", name.replace('"', "\"\"")))
.unwrap();
let count = statement.column_count();
let mut rows = statement
.query_map([], |row| {
(0..count)
.map(|index| row.get::<_, rusqlite::types::Value>(index))
.collect::<rusqlite::Result<Vec<_>>>()
})
.unwrap()
.map(|row| format!("{:?}", row.unwrap()))
.collect::<Vec<_>>();
rows.sort();
(name, rows)
})
.collect()
}
fn assert_accepted_without_domain_changes(conn: &Connection, current: i64) {
let rows = saved_rows(conn);
let shape = snapshot(conn).unwrap();
verify_history(conn, current).unwrap();
if current >= 114 {
let plan = published_plan(conn, current).unwrap().unwrap();
assert!(plan.migrations.is_empty());
assert_eq!(plan.target_schema_version, current);
}
assert_eq!(rows, saved_rows(conn));
assert_eq!(shape, snapshot(conn).unwrap());
}
#[test]
fn each_published_receipt_prefix_and_fresh_branch_baseline_requires_its_real_shape() {
for family in FAMILIES {
for count in 1..=family.names().len() {
let stage = Stage::Published(family, count);
let current = 113 + count as i64;
// A baseline can start at any retained branch stage, not just 113.
for baseline in 113..=current {
let conn = fixture(stage, baseline, None);
assert_accepted_without_domain_changes(&conn, current);
}
}
}
}
#[test]
fn converged_shapes_keep_all_real_predecessor_receipts_and_erasure_fences() {
let origins = std::iter::once(None).chain(
FAMILIES
.into_iter()
.flat_map(|family| (1..=family.names().len()).map(move |count| Some((family, count)))),
);
for origin in origins {
for current in 119..=121 {
let conn = fixture(Stage::Converged(current), 113, origin);
assert_accepted_without_domain_changes(&conn, current);
let erased: i64 = conn
.query_row("SELECT erased FROM content_history_targets", [], |row| {
row.get(0)
})
.unwrap();
assert_eq!(erased, 1);
// A physical check must not act as the Job owner's new Failed migration.
let state: String = conn
.query_row("SELECT state FROM backend_jobs", [], |row| row.get(0))
.unwrap();
assert_eq!(state, "pending");
}
}
for current in 119..=121 {
for baseline in 119..=current {
assert_accepted_without_domain_changes(
&fixture(Stage::Converged(current), baseline, None),
current,
);
}
}
}
#[test]
fn published_database_runs_normal_store_preparation_and_reopens_without_downgrade() {
for stage in [
Stage::Published(Family::Permissions, 2),
Stage::Published(Family::Removal, 4),
Stage::Published(Family::Jobs, 5),
Stage::Converged(119),
Stage::Converged(120),
Stage::Converged(121),
] {
let current = match stage {
Stage::Published(_, count) => 113 + count as i64,
Stage::Converged(version) => version,
Stage::Base => 113,
};
let source = fixture(stage, 113, None);
let expected = saved_rows(&source);
let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("catalog.db");
let mut destination = Connection::open(&path).unwrap();
Backup::new(&source, &mut destination)
.unwrap()
.run_to_completion(128, std::time::Duration::ZERO, None)
.unwrap();
drop(destination);
for _ in 0..2 {
let conn = Connection::open(&path).unwrap();
// Production store owner, not a separate test migration implementation.
super::super::prepare_connection(&conn).unwrap();
assert_eq!(
super::super::current_schema_version(&conn).unwrap(),
current
);
assert_eq!(saved_rows(&conn), expected);
}
}
}
#[test]
fn unknown_mixed_missing_receipts_and_shapes_are_diagnosed_without_writes() {
for (sql, current) in [
(
"UPDATE __yoi_schema_migrations SET name='unknown receipt' WHERE version=115",
118,
),
("DELETE FROM __yoi_schema_migrations WHERE version=116", 118),
(
"UPDATE __yoi_schema_migrations SET name='Worker removal expected Feature database identity' WHERE version=115",
118,
),
("DROP TRIGGER backend_job_cleanup_owner_immutable", 118),
(
"CREATE INDEX unreceipted_job_index ON backend_jobs(job_id)",
118,
),
(
"INSERT INTO __yoi_schema_migrations(version,name) VALUES(122,'unknown future')",
122,
),
] {
let conn = fixture(Stage::Published(Family::Jobs, 5), 113, None);
conn.execute_batch(sql).unwrap();
let rows = saved_rows(&conn);
let shape = snapshot(&conn).unwrap();
assert!(verify_history(&conn, current).is_err(), "{sql}");
assert_eq!(saved_rows(&conn), rows);
assert_eq!(snapshot(&conn).unwrap(), shape);
}
let conn = fixture(Stage::Published(Family::Permissions, 2), 113, None);
conn.execute_batch(COLLECTION_SQL).unwrap();
assert!(
verify_history(&conn, 115)
.unwrap_err()
.to_string()
.contains("mismatch")
);
let conn = fixture(Stage::Converged(121), 113, None);
conn.execute_batch("DROP TRIGGER backend_job_cleanup_origin_immutable")
.unwrap();
assert!(verify_history(&conn, 121).is_err());
}
#[test]
fn missing_or_weakened_constraints_and_literal_changes_are_not_shape_equivalent() {
let conn = fixture(Stage::Published(Family::Permissions, 1), 113, None);
conn.execute_batch("DROP TABLE permission_policy_receipts;
CREATE TABLE permission_policy_receipts(workspace_id TEXT NOT NULL,operation_id TEXT NOT NULL,fingerprint TEXT NOT NULL,result_json TEXT NOT NULL,PRIMARY KEY(workspace_id,operation_id));").unwrap();
assert!(verify_history(&conn, 114).is_err());
let conn = fixture(Stage::Published(Family::Jobs, 5), 113, None);
conn.execute_batch("DROP TRIGGER backend_job_cleanup_owner_immutable;
CREATE TRIGGER backend_job_cleanup_owner_immutable BEFORE UPDATE ON backend_job_cleanup_invocation_owners
BEGIN SELECT RAISE(ABORT,'Cleanup invocation ownership is immutable'); END;").unwrap();
assert!(verify_history(&conn, 118).is_err());
}
fn retained_to(target: i64, alternate_103: bool) -> (tempfile::TempDir, Connection) {
let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("retained.db");
super::super::tests::prepare_schema_v50(&path, Some("workspace-a"));
let conn = Connection::open(&path).unwrap();
super::super::configure_sqlite(&conn).unwrap();
for migration in super::super::MIGRATIONS
.iter()
.filter(|migration| migration.version <= target)
{
let current = super::super::current_schema_version(&conn).unwrap();
verify_history(&conn, current).unwrap();
let migration = if migration.version == 103 && alternate_103 {
&super::super::JOB_INPUT_RETENTION_SCHEMA_103
} else {
migration
};
(migration.apply)(&conn).unwrap();
}
verify_history(&conn, target).unwrap();
(temp, conn)
}
#[test]
fn oldest_data_bearing_schema_and_both_real_103_histories_migrate_without_relabeling() {
for alternate in [false, true] {
let (_temp, conn) = retained_to(103, alternate);
let original = super::super::workspace_schema_migration_history(&conn).unwrap();
super::super::prepare_connection(&conn).unwrap();
verify_history(&conn, 113).unwrap();
let updated = super::super::workspace_schema_migration_history(&conn).unwrap();
assert_eq!(&updated[..original.len()], &original);
let name: String = conn
.query_row(
"SELECT display_name FROM workspaces WHERE workspace_id='workspace-a'",
[],
|row| row.get(0),
)
.unwrap();
assert_eq!(name, "Workspace A");
}
}
#[test]
fn fresh_103_baselines_use_shape_and_never_repair_missing_receipts_or_guards() {
for alternate in [false, true] {
let (_temp, conn) = retained_to(103, alternate);
conn.execute_batch("DELETE FROM __yoi_schema_migrations;
INSERT INTO __yoi_schema_migrations(version,name) VALUES(103,'workspace schema baseline');").unwrap();
assert_accepted_without_domain_changes(&conn, 103);
if alternate {
conn.execute_batch("DROP TRIGGER backend_job_input_retirement")
.unwrap();
} else {
conn.execute_batch("DROP TABLE passkey_addition_receipts")
.unwrap();
}
let rows = saved_rows(&conn);
assert!(verify_history(&conn, 103).is_err());
assert_eq!(saved_rows(&conn), rows);
}
let (_temp, conn) = retained_to(102, false);
conn.execute_batch(
"INSERT INTO __yoi_schema_migrations(version,name) VALUES(103,'workspace schema baseline')",
)
.unwrap();
assert!(verify_history(&conn, 103).is_err());
}
#[test]
fn later_family_objects_without_receipts_and_half_converged_103_are_rejected() {
let conn = fixture(Stage::Base, 113, None);
conn.execute_batch(CLAIMS_SQL).unwrap();
assert!(verify_history(&conn, 113).is_err());
let (_temp, conn) = retained_to(103, false);
conn.execute_batch("ALTER TABLE backend_jobs ADD COLUMN input_disposition TEXT NOT NULL DEFAULT 'available' CHECK(input_disposition IN ('available','reclaimed','erased','invalidated'))").unwrap();
assert!(verify_history(&conn, 103).is_err());
}
#[test]
fn schema_comparison_preserves_literals_programs_and_composite_key_order() {
let left = declarations(
tokens("CREATE TABLE item(a TEXT, b TEXT, PRIMARY KEY(a,b), CHECK(a='Sensitive'))")
.unwrap(),
)
.unwrap();
let reordered = declarations(
tokens("CREATE TABLE \"item\"(b TEXT, CHECK(a='Sensitive'), a TEXT, PRIMARY KEY(a,b))")
.unwrap(),
)
.unwrap();
assert_eq!(left, reordered);
for sql in [
"CREATE TABLE item(a TEXT, b TEXT, PRIMARY KEY(b,a), CHECK(a='Sensitive'))",
"CREATE TABLE item(a TEXT, b TEXT, PRIMARY KEY(a,b), CHECK(a='sensitive'))",
] {
assert_ne!(left, declarations(tokens(sql).unwrap()).unwrap());
}
}
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,151 @@
//! Real Server catalog -> FeatureStorage -> Subjektiv connection tests. The
//! fixture creates saved domain data through public APIs, then removes only the
//! new physical identity metadata from a SQLite backup to represent a pre-owner
//! database. No production paths, Runtime, or authorization provider is borrowed.
use super::*;
use crate::subjektiv::{SubjectRole, SubjektivStore};
use feature_storage::FeatureStorage;
fn catalog(path: &Path) -> SqliteWorkspaceStore {
let store = SqliteWorkspaceStore::open_for_test(path).unwrap();
store.with_conn(|conn| {
conn.execute_batch("INSERT INTO accounts(account_id,kind,handle,display_name,created_at,updated_at)
VALUES('storage-owner','user','storage-owner','Owner','1','1');
INSERT INTO workspaces(workspace_id,owner_account_id,display_name,state,created_at,updated_at)
VALUES('storage-workspace','storage-owner','Saved','active','1','1');")?;
Ok(())
}).unwrap();
store
}
fn legacy_subject(source: &Path, destination: &Path, attribution: &str) -> String {
let storage = FeatureStorage::new(source).scope(attribution).unwrap();
let registration = SubjektivStore::register(&storage).unwrap();
let database = storage.open(&registration).unwrap();
let subject_store = SubjektivStore::from_database(database.clone()).unwrap();
let subject = subject_store
.create_subject_with_behavior(
SubjectRole::new("retained").unwrap(),
"original behavior".into(),
)
.unwrap();
std::fs::create_dir_all(destination.parent().unwrap()).unwrap();
database
.with_connection(|conn| {
let mut copy = Connection::open(destination)?;
Backup::new(conn, &mut copy)?.run_to_completion(128, Duration::from_millis(1), None)?;
copy.execute_batch("DROP TABLE __yoi_storage_identity;")?;
Ok(())
})
.unwrap();
subject.id
}
#[test]
fn catalog_authorized_legacy_adoption_reopens_and_preserves_domain_data() {
let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("server.db");
let catalog = catalog(&path);
let owner = FeatureStorage::for_server_database(&path).unwrap();
let scope = owner.scope("storage-workspace").unwrap();
let registration = SubjektivStore::register(&scope).unwrap();
let feature_path = temp
.path()
.join("feature-storage/storage-workspace/features/subjektiv.sqlite");
let subject = legacy_subject(
&temp.path().join("source"),
&feature_path,
"storage-workspace",
);
assert!(
scope.open(&registration).is_err(),
"path existence is not adoption authority"
);
catalog
.adopt_workspace_feature_storage(&scope, &registration, "store_scope")
.unwrap();
let store = SubjektivStore::open(&scope, &registration).unwrap();
assert_eq!(store.subject(&subject).unwrap().unwrap().id, subject);
drop(store);
scope.shutdown().unwrap();
let owner = FeatureStorage::for_server_database(&path).unwrap();
let scope = owner.scope("storage-workspace").unwrap();
let registration = SubjektivStore::register(&scope).unwrap();
catalog
.adopt_workspace_feature_storage(&scope, &registration, "store_scope")
.unwrap();
assert_eq!(
SubjektivStore::open(&scope, &registration)
.unwrap()
.subject(&subject)
.unwrap()
.unwrap()
.id,
subject
);
}
#[test]
fn missing_catalog_or_different_feature_attribution_never_grants_adoption() {
for attribution in ["storage-workspace", "another-workspace"] {
let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("server.db");
let catalog = catalog(&path);
let scope = FeatureStorage::for_server_database(&path)
.unwrap()
.scope("storage-workspace")
.unwrap();
let registration = SubjektivStore::register(&scope).unwrap();
let feature_path = temp
.path()
.join("feature-storage/storage-workspace/features/subjektiv.sqlite");
legacy_subject(&temp.path().join("source"), &feature_path, attribution);
if attribution == "storage-workspace" {
catalog
.with_conn(|conn| {
conn.execute("DELETE FROM workspaces", [])?;
Ok(())
})
.unwrap();
}
assert!(
catalog
.adopt_workspace_feature_storage(&scope, &registration, "store_scope")
.is_err()
);
let raw = Connection::open(&feature_path).unwrap();
let count: i64 = raw
.query_row(
"SELECT COUNT(*) FROM sqlite_schema WHERE name='__yoi_storage_identity'",
[],
|r| r.get(0),
)
.unwrap();
assert_eq!(count, 0, "refusal must not mint identity");
}
}
#[test]
fn catalog_cached_connection_rejects_replaced_file_and_missing_reopen() {
let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("server.db");
let store = catalog(&path);
let replacement = temp.path().join("replacement.db");
store
.with_conn(|conn| {
let mut copy = Connection::open(&replacement)?;
Backup::new(conn, &mut copy)?.run_to_completion(128, Duration::from_millis(1), None)?;
Ok(())
})
.unwrap();
std::fs::rename(&path, temp.path().join("old.db")).unwrap();
std::fs::rename(&replacement, &path).unwrap();
assert!(store.with_conn(|_| Ok(())).is_err());
assert!(SqliteWorkspaceStore::open_for_test(&path).is_err());
std::fs::remove_file(&path).unwrap();
assert!(SqliteWorkspaceStore::open_for_test(&path).is_err());
assert!(
!path.exists(),
"bound catalog must not silently CREATE a successor"
);
}
+101 -17
View File
@@ -129,6 +129,9 @@ pub use ticket::{
}; };
const OLDEST_SCHEMA_VERSION: i64 = 50; const OLDEST_SCHEMA_VERSION: i64 = 50;
#[path = "storage_formats.rs"]
mod storage_formats;
const LATEST_SCHEMA_VERSION: i64 = 113; const LATEST_SCHEMA_VERSION: i64 = 113;
const WORKSPACE_DRIVE_GRANTS_MIGRATION_NAME: &str = "durable Workspace Drive grants"; const WORKSPACE_DRIVE_GRANTS_MIGRATION_NAME: &str = "durable Workspace Drive grants";
const WORKSPACE_CONFIG_GRANTS_MIGRATION_NAME: &str = "Workspace config grants and logical Workdirs"; const WORKSPACE_CONFIG_GRANTS_MIGRATION_NAME: &str = "Workspace config grants and logical Workdirs";
@@ -2785,9 +2788,21 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
) -> Result<Option<WorkerWorkdirLinkRecord>>; ) -> Result<Option<WorkerWorkdirLinkRecord>>;
} }
#[derive(Clone)]
enum StoreConnection {
Managed(feature_storage::FeatureDatabase),
Fixture(Arc<Mutex<Connection>>),
}
impl From<feature_storage::FeatureStorageError> for Error {
fn from(error: feature_storage::FeatureStorageError) -> Self {
Error::Store(error.to_string())
}
}
#[derive(Clone)] #[derive(Clone)]
pub struct SqliteWorkspaceStore { pub struct SqliteWorkspaceStore {
conn: Arc<Mutex<Connection>>, conn: StoreConnection,
db_executor: crate::db_executor::DbExecutor, db_executor: crate::db_executor::DbExecutor,
} }
@@ -2797,8 +2812,18 @@ pub mod fixture_support;
impl SqliteWorkspaceStore { impl SqliteWorkspaceStore {
pub fn open(path: impl AsRef<Path>) -> Result<Self> { pub fn open(path: impl AsRef<Path>) -> Result<Self> {
let conn = Connection::open(path)?; let path = path.as_ref();
Self::from_connection(conn) let owner = feature_storage::FeatureStorage::for_server_database(path)?;
let database = owner.open_host_catalog(
path,
"server-catalog",
"Host configured Server database",
|conn| prepare_connection(conn),
)?;
Ok(Self {
conn: StoreConnection::Managed(database),
db_executor: crate::db_executor::DbExecutor::shared(),
})
} }
pub fn in_memory() -> Result<Self> { pub fn in_memory() -> Result<Self> {
@@ -2842,7 +2867,7 @@ impl SqliteWorkspaceStore {
) -> Result<Self> { ) -> Result<Self> {
prepare_connection(&conn)?; prepare_connection(&conn)?;
Ok(Self { Ok(Self {
conn: Arc::new(Mutex::new(conn)), conn: StoreConnection::Fixture(Arc::new(Mutex::new(conn))),
db_executor, db_executor,
}) })
} }
@@ -2851,7 +2876,11 @@ impl SqliteWorkspaceStore {
// fixture's admission/SQL/drain lifetime or a permanent global worker. // fixture's admission/SQL/drain lifetime or a permanent global worker.
#[cfg(test)] #[cfg(test)]
pub(crate) fn open_for_test(path: impl AsRef<Path>) -> Result<Self> { pub(crate) fn open_for_test(path: impl AsRef<Path>) -> Result<Self> {
Self::from_connection_for_test(Connection::open(path)?) Ok(
Self::open(path)?.with_db_executor(crate::db_executor::DbExecutor::new(
crate::db_executor::DbExecutorLimits::default(),
)?),
)
} }
#[cfg(test)] #[cfg(test)]
@@ -2878,22 +2907,30 @@ impl SqliteWorkspaceStore {
} }
pub(crate) fn with_conn<T>(&self, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> { pub(crate) fn with_conn<T>(&self, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
let conn = self match &self.conn {
.conn StoreConnection::Managed(database) => database.try_with_connection(f),
.lock() StoreConnection::Fixture(connection) => {
.map_err(|_| Error::Store("sqlite connection lock poisoned".to_string()))?; let conn = connection
f(&conn) .lock()
.map_err(|_| Error::Store("sqlite connection lock poisoned".into()))?;
f(&conn)
}
}
} }
pub(crate) fn with_conn_mut<T>( pub(crate) fn with_conn_mut<T>(
&self, &self,
f: impl FnOnce(&mut Connection) -> Result<T>, f: impl FnOnce(&mut Connection) -> Result<T>,
) -> Result<T> { ) -> Result<T> {
let mut conn = self match &self.conn {
.conn StoreConnection::Managed(database) => database.try_with_connection_mut(f),
.lock() StoreConnection::Fixture(connection) => {
.map_err(|_| Error::Store("sqlite connection lock poisoned".to_string()))?; let mut conn = connection
f(&mut conn) .lock()
.map_err(|_| Error::Store("sqlite connection lock poisoned".into()))?;
f(&mut conn)
}
}
} }
/// Acquire SQLite write ownership before the first authority/journal read (Restore and /// Acquire SQLite write ownership before the first authority/journal read (Restore and
@@ -2915,6 +2952,43 @@ impl SqliteWorkspaceStore {
}) })
} }
/// The current catalog row authorizes Host storage setup, not model access
/// or past provisioning. Persisted Feature attribution must independently
/// match before adopting a data-bearing legacy file.
pub(crate) fn adopt_workspace_feature_storage(
&self,
storage: &feature_storage::ScopedFeatureStorage,
registration: &feature_storage::RegisteredFeature,
attribution_table: &str,
) -> Result<()> {
let table = match attribution_table {
"store_scope" => "store_scope",
"drive_state" => "drive_state",
_ => {
return Err(Error::InvalidInput(
"unsupported storage attribution owner".into(),
));
}
};
self.with_immediate_transaction(|tx| {
let row: Option<(String,String)> = tx.query_row(
"SELECT owner_account_id,created_at FROM workspaces WHERE workspace_id=?1 AND state='active'",[storage.scope_id()],
|row| Ok((row.get(0)?,row.get(1)?))).optional()?;
let (owner,created) = row.ok_or_else(|| Error::Store("Workspace storage requires a current catalog owner".into()))?;
let evidence = format!("catalog:{}:{owner}:{created}",storage.scope_id());
let request = format!("legacy-storage:{}:{}",storage.scope_id(),registration.feature_id());
storage.adopt_legacy_if_present(registration,&request,&evidence,|feature_tx| {
let stored: String = feature_tx.query_row(
&format!("SELECT workspace_id FROM {table} WHERE singleton=1"),[],|row|row.get(0))?;
if stored != storage.scope_id() {
return Err(feature_storage::FeatureStorageError::Identity("legacy Feature attribution differs from authorized Workspace".into()));
}
Ok(())
})?;
Ok(())
})
}
/// Host-only setup; never expose this operation through a public bootstrap route. /// Host-only setup; never expose this operation through a public bootstrap route.
/// Creating a user, granting hosting authority and issuing its short-lived initial /// Creating a user, granting hosting authority and issuing its short-lived initial
/// bearer are atomic. Selecting an existing user never issues credentials. A repeat /// bearer are atomic. Selecting an existing user never issues credentials. A repeat
@@ -16987,6 +17061,9 @@ fn workspace_schema_migration_history(conn: &Connection) -> Result<Vec<(i64, Str
} }
fn verify_schema_history(conn: &Connection, current: i64) -> Result<()> { fn verify_schema_history(conn: &Connection, current: i64) -> Result<()> {
if current > LATEST_SCHEMA_VERSION {
return storage_formats::verify_history(conn, current);
}
let rows = workspace_schema_migration_history(conn)?; let rows = workspace_schema_migration_history(conn)?;
let Some((baseline_version, baseline_name)) = rows.first() else { let Some((baseline_version, baseline_name)) = rows.first() else {
return Err(Error::Store(format!( return Err(Error::Store(format!(
@@ -17030,6 +17107,9 @@ fn verify_schema_history(conn: &Connection, current: i64) -> Result<()> {
fn workspace_schema_migration_plan(conn: &Connection) -> Result<WorkspaceSchemaMigrationPlan> { fn workspace_schema_migration_plan(conn: &Connection) -> Result<WorkspaceSchemaMigrationPlan> {
let current = current_schema_version(conn)?; let current = current_schema_version(conn)?;
if let Some(plan) = storage_formats::published_plan(conn, current)? {
return Ok(plan);
}
if current > LATEST_SCHEMA_VERSION { if current > LATEST_SCHEMA_VERSION {
return Err(Error::Store(format!( return Err(Error::Store(format!(
"database schema version {current} is newer than this server supports ({LATEST_SCHEMA_VERSION}); refusing to serve with an older binary" "database schema version {current} is newer than this server supports ({LATEST_SCHEMA_VERSION}); refusing to serve with an older binary"
@@ -17041,7 +17121,7 @@ fn workspace_schema_migration_plan(conn: &Connection) -> Result<WorkspaceSchemaM
))); )));
} }
if current != 0 { if current != 0 {
verify_schema_history(conn, current)?; storage_formats::verify_history(conn, current)?;
} }
let migrations = if current == 0 { let migrations = if current == 0 {
@@ -17150,7 +17230,7 @@ fn apply_migrations_with_secret_source(
} }
} }
verify_schema_history(conn, LATEST_SCHEMA_VERSION)?; storage_formats::verify_history(conn, current_schema_version(conn)?)?;
crate::plugin_catalog::verify_with_data(conn)?; crate::plugin_catalog::verify_with_data(conn)?;
verify_plugin_data_schema(conn)?; verify_plugin_data_schema(conn)?;
verify_plugin_service_schema(conn)?; verify_plugin_service_schema(conn)?;
@@ -17186,6 +17266,10 @@ fn table_columns(conn: &Connection, table_name: &str) -> Result<Vec<String>> {
.map_err(Error::from) .map_err(Error::from)
} }
#[cfg(test)]
#[path = "storage_owner_tests.rs"]
mod storage_owner_tests;
#[cfg(test)] #[cfg(test)]
#[path = "store_test_support.rs"] #[path = "store_test_support.rs"]
mod test_support; mod test_support;
@@ -25,7 +25,7 @@ impl SqliteWorkspaceStore {
let conn = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_WRITE)?; let conn = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_WRITE)?;
configure_sqlite(&conn)?; configure_sqlite(&conn)?;
Ok(Self { Ok(Self {
conn: Arc::new(Mutex::new(conn)), conn: StoreConnection::Fixture(Arc::new(Mutex::new(conn))),
db_executor: crate::db_executor::DbExecutor::new( db_executor: crate::db_executor::DbExecutor::new(
crate::db_executor::DbExecutorLimits::default(), crate::db_executor::DbExecutorLimits::default(),
)?, )?,
@@ -56,7 +56,7 @@ impl SqliteWorkspaceStore {
// Existing DBs above still pass through full production startup checks. // Existing DBs above still pass through full production startup checks.
configure_sqlite(&conn)?; configure_sqlite(&conn)?;
Ok(Self { Ok(Self {
conn: Arc::new(Mutex::new(conn)), conn: StoreConnection::Fixture(Arc::new(Mutex::new(conn))),
db_executor: crate::db_executor::DbExecutor::new( db_executor: crate::db_executor::DbExecutor::new(
crate::db_executor::DbExecutorLimits::default(), crate::db_executor::DbExecutorLimits::default(),
)?, )?,
@@ -76,6 +76,14 @@ impl WorkspaceDriveHost {
drive: Mutex::new(None), drive: Mutex::new(None),
}) })
} }
pub(crate) fn adopt_legacy_storage(
&self,
storage: &WorkspaceFeatureStorage,
catalog: &crate::SqliteWorkspaceStore,
) -> Result<()> {
catalog.adopt_workspace_feature_storage(storage, &self.registration, "drive_state")
}
pub(crate) async fn open( pub(crate) async fn open(
&self, &self,
storage: &WorkspaceFeatureStorage, storage: &WorkspaceFeatureStorage,
+11
View File
@@ -1,4 +1,15 @@
[ [
{
"path": "crates/subjektiv/src/storage_format_tests.rs",
"kind": "saved_data",
"reason": "最古対応Subjektiv保存形式のデータ入り移行fixture。旧Subject JSONの保存keyを実形式のまま検証し、移行前原文と移行後データの保持を確認するため残す。現行状態契約の新設ではない。",
"lines": [
{
"sha256": "5b22a15424eb27bcd68e737cf3549b7ce4231632cdda2bd2d826b4a2c766e270",
"count": 1
}
]
},
{ {
"path": "tests/e2e/fixtures/worker-codex-v1/manifest.json", "path": "tests/e2e/fixtures/worker-codex-v1/manifest.json",
"kind": "saved_data", "kind": "saved_data",