diff --git a/docs/meeting-publication-migration.md b/docs/meeting-publication-migration.md new file mode 100644 index 00000000..7bd7baef --- /dev/null +++ b/docs/meeting-publication-migration.md @@ -0,0 +1,57 @@ +# Migrating legacy meeting artifacts + +Native meeting publication v3 publishes immutable, digest-verified snapshots through the existing authenticated SQL service. Activation adds the connector publication schema and rejects generic writes to protected catalog tables. It does not verify or convert legacy bodies automatically. + +A migration should inventory and validate the existing catalog and raw KV bodies, preserve a private copy and resumable operation plan, pause legacy artifact writes, activate the SQL fence, and revalidate the plan before publication. Verify every published copy before releasing the legacy KV pause. Keep the original capture extent and provenance unknown unless independently established. + +## Temporary write barrier + +These controls use the existing `tinycloud.meetingPublication.v3` statement at the exact SQL path `xyz.tinycloud.tinychat/connectors`. They require `tinycloud.sql/write` and the existing unconstrained ancestor-chain authority. Their SQL result has a `receipt` column containing one JSON string. + +```json +{"contractVersion":3,"operation":"legacy_freeze_status"} +``` + +A fresh space reports: + +```json +{"contractVersion":3,"legacyWritesFrozen":false,"legacyFreezeGeneration":0} +``` + +Persist the expected generation before issuing a control request: + +```json +{"contractVersion":3,"operation":"freeze_legacy","expectedGeneration":0} +``` + +The successful receipt reports `legacyWritesFrozen:true` and `legacyFreezeGeneration:1`. After publication and verification, release that generation: + +```json +{"contractVersion":3,"operation":"unfreeze_legacy","expectedGeneration":1} +``` + +Release reports `legacyWritesFrozen:false` and `legacyFreezeGeneration:2`. Each actual transition increments the generation. An immediate identical retry returns the same result; a stale request from an older cycle fails with `legacy_freeze_generation_conflict` (HTTP 400). Invalid generation inputs also return 400. Generations distinguish migration cycles, not independent operators issuing identical simultaneous requests; coordinate one operator per migration. + +The native route advertises `legacyWriteFreeze:true` in its publication capabilities. Status, freeze and release remain available when content quota is exhausted; content-growing publication operations remain quota checked. + +## Protected scope and guarantees + +The pause applies only to these paths under `xyz.tinycloud.tinychat/connectors/{fireflies,google-meet,tinycloud-transcriber}/`: + +- `transcript/…` +- `meeting/…` +- `archive-copy/transcript/…` + +Reads, chat keys, cursors, credentials, other spaces and native snapshot publication remain available. Frozen ordinary KV put/delete returns HTTP 409. Frozen native delete/purge returns HTTP 403 before known catalog mutation. The core guard also rejects internal legacy cleanup. + +The durable guard is locked by the protected KV mutation transaction before its first database read. PostgreSQL row locking and SQLite writer serialization hold that lock through storage persistence and commit. A freeze acknowledgement therefore drains earlier protected KV commits. Failed guard reads fail closed, and restarting the node does not release a pause. + +The early native delete/purge check does not make the separate SQL publication and KV cleanup transactions globally atomic across multiple instances. Deploy compatible code to every traffic-serving node and coordinate migration writers. Freeze alone does not fence generic SQL; activation supplies that separate catalog fence. + +Release allows later legacy artifact mutations. It does not deactivate the SQL fence or weaken immutable snapshots, which retain the verified original body independently. Older writers cannot publish to an activated catalog and should be replaced by compatible clients. + +## Rollout and recovery + +The central `meeting_legacy_write_guard` migration creates the table without automatically freezing a space. Old binaries that do not recognize its migration name may reject startup against the upgraded database. Its down operation refuses to silently discard the guard generation. A rollback build must recognize the migration; after catalog activation it must also retain the publication protocol and writer fences. + +Keep a compatible node running for inspection, repair and resume. Retain the original private plan and operation IDs after interruption or a lost acknowledgement. Do not invent new operations to recover uncertain publications, edit migration history, or restore a shared database wholesale to recover one space. diff --git a/tinycloud-core/src/database_artifacts.rs b/tinycloud-core/src/database_artifacts.rs index 1270f864..a321dc1b 100644 --- a/tinycloud-core/src/database_artifacts.rs +++ b/tinycloud-core/src/database_artifacts.rs @@ -144,6 +144,7 @@ pub trait DatabaseArtifactRepository: Send + Sync { #[derive(Clone)] pub struct SeaOrmDatabaseArtifactRepository { conn: DatabaseConnection, + sqlite_writer_lock: Option>>, /// Test-only rendezvous seam (see [`wait_at_race_barrier`]) that lets two /// writers read the same base revision before either commits, so the /// full-checkpoint CAS conflict path can be exercised deterministically. @@ -155,11 +156,22 @@ impl SeaOrmDatabaseArtifactRepository { pub fn new(conn: DatabaseConnection) -> Self { Self { conn, + sqlite_writer_lock: None, #[cfg(test)] race_barrier: None, } } + /// Share the node's SQLite gate: artifact writes use the capability DB and + /// must not advance its WAL beneath a delegation transaction's snapshot. + pub fn with_sqlite_writer_lock( + mut self, + lock: Option>>, + ) -> Self { + self.sqlite_writer_lock = lock; + self + } + #[cfg(test)] fn with_race_barrier_for_test(mut self, barrier: std::sync::Arc) -> Self { self.race_barrier = Some(barrier); @@ -222,6 +234,10 @@ impl DatabaseArtifactRepository for SeaOrmDatabaseArtifactRepository { payload: Vec, expected: ArtifactExpectation, ) -> Result { + let _writer = match &self.sqlite_writer_lock { + Some(lock) => Some(lock.lock().await), + None => None, + }; let size_bytes = i64::try_from(payload.len()) .map_err(|_| DatabaseArtifactError::PayloadTooLarge(payload.len() as u64))?; let content_hash = hash(&payload).to_cid(0x55).to_string(); @@ -391,6 +407,10 @@ impl DatabaseArtifactRepository for SeaOrmDatabaseArtifactRepository { payload: Vec, expected: ArtifactExpectation, ) -> Result { + let _writer = match &self.sqlite_writer_lock { + Some(lock) => Some(lock.lock().await), + None => None, + }; let existing = database_artifact::Entity::find_by_id(( service.to_string(), space.to_string(), diff --git a/tinycloud-core/src/db.rs b/tinycloud-core/src/db.rs index 562bc8dc..232c3735 100644 --- a/tinycloud-core/src/db.rs +++ b/tinycloud-core/src/db.rs @@ -95,12 +95,43 @@ enum InvokeMode { /// Pre-authorized by a trusted application protocol seam /// (`invoke_internal_kv_put`). Skips the full UCAN validator entirely. Internal, + /// Fixed native meeting publication; the only writer of immutable snapshot keys. + MeetingPublication, /// Envelope already verified once at admission. Authorization, /// revocation, caveat containment, and signed-time validity are still /// re-checked at execution; only the signature check is skipped. Admitted, } +fn meeting_snapshot_write_allowed(mode: InvokeMode, path: &str) -> bool { + mode == InvokeMode::MeetingPublication + || !crate::sql::publication::protected_snapshot_path(path) +} + +#[cfg(test)] +mod meeting_snapshot_fence_tests { + use super::*; + #[test] + fn publication_snapshot_prefix_rejects_legacy_and_share_writes() { + let path = "xyz.tinycloud.tinychat/connectors/fireflies/snapshot/source/digest"; + for mode in [ + InvokeMode::Public, + InvokeMode::Admitted, + InvokeMode::Internal, + ] { + assert!(!meeting_snapshot_write_allowed(mode, path)); + } + assert!(meeting_snapshot_write_allowed( + InvokeMode::MeetingPublication, + path + )); + assert!(meeting_snapshot_write_allowed( + InvokeMode::Public, + "xyz.tinycloud.tinychat/connectors/fireflies/transcript/source" + )); + } +} + #[derive(Debug, Clone)] pub struct SpaceDatabase { conn: C, @@ -217,6 +248,10 @@ where Io(#[from] std::io::Error), #[error("Missing Input for requested action")] MissingInput, + #[error("meeting snapshot keys require the native publication boundary")] + MeetingSnapshotProtected, + #[error("legacy meeting artifacts are frozen")] + LegacyMeetingFrozen, #[error("KV precondition failed")] KvPreconditionFailed, #[error("conditional KV transaction conflicted; retry the request")] @@ -307,6 +342,41 @@ impl SpaceDatabase where C: TransactionTrait, { + /// Change the legacy artifact barrier. Freeze commit is the drain acknowledgement. + pub async fn set_legacy_meeting_write_freeze( + &self, + space: &SpaceId, + frozen: bool, + expected_generation: i64, + ) -> Result + where + C: ConnectionTrait, + { + let _writer = match &self.writer_lock { + Some(lock) => Some(lock.lock().await), + None => None, + }; + let tx = self + .conn + .begin_with_config(chain_isolation_level(&self.conn), None) + .await?; + let status = + crate::meeting_legacy_guard::transition(&tx, space, frozen, expected_generation) + .await?; + tx.commit().await?; + Ok(status) + } + + pub async fn legacy_meeting_freeze_status( + &self, + space: &SpaceId, + ) -> Result + where + C: ConnectionTrait, + { + crate::meeting_legacy_guard::status(&self.conn, space).await + } + // to allow users to make custom read queries pub async fn readable(&self) -> Result { self.conn @@ -1213,6 +1283,88 @@ where stage: HashBuffer, precondition: Option, ) -> Result> + where + B: ImmutableWriteStore + ImmutableReadStore, + S: ImmutableStaging, + S::Writable: 'static + Unpin, + { + self.invoke_internal_kv_change::( + space, + path, + metadata, + Some(stage), + precondition, + InvokeMode::Internal, + ) + .await + .and_then(|hash| hash.ok_or(TxStoreError::MissingInput)) + } + pub async fn invoke_internal_meeting_snapshot_put( + &self, + space: SpaceId, + path: Path, + metadata: Metadata, + stage: HashBuffer, + precondition: Option, + ) -> Result> + where + B: ImmutableWriteStore + ImmutableReadStore, + S: ImmutableStaging, + S::Writable: 'static + Unpin, + { + if !crate::sql::publication::protected_snapshot_path(path.as_str()) { + return Err(TxStoreError::MeetingSnapshotProtected); + } + self.invoke_internal_kv_change::( + space, + path, + metadata, + Some(stage), + precondition, + InvokeMode::MeetingPublication, + ) + .await + .and_then(|hash| hash.ok_or(TxStoreError::MissingInput)) + } + pub async fn invoke_internal_meeting_snapshot_delete( + &self, + space: SpaceId, + path: Path, + ) -> Result<(), TxStoreError> + where + B: ImmutableWriteStore + ImmutableReadStore, + S: ImmutableStaging, + S::Writable: 'static + Unpin, + { + if !crate::sql::publication::connector_owned_path(path.as_str()) { + return Err(TxStoreError::MeetingSnapshotProtected); + } + self.invoke_internal_kv_change::( + space, + path, + Metadata(Default::default()), + None, + None, + InvokeMode::MeetingPublication, + ) + .await + .map(|_| ()) + .or_else(|error| match error { + TxStoreError::Tx(TxError::InvalidInvocation( + crate::models::invocation::InvocationError::MissingKvWrite(_), + )) => Ok(()), + error => Err(error), + }) + } + async fn invoke_internal_kv_change( + &self, + space: SpaceId, + path: Path, + metadata: Metadata, + stage: Option>, + precondition: Option, + mode: InvokeMode, + ) -> Result, TxStoreError> where B: ImmutableWriteStore + ImmutableReadStore, S: ImmutableStaging, @@ -1244,9 +1396,13 @@ where let invocation = make_invocation( vec![( resource, - vec!["tinycloud.kv/put" - .parse::() - .map_err(|_| TxStoreError::MissingInput)?], + vec![(if stage.is_some() { + "tinycloud.kv/put" + } else { + "tinycloud.kv/del" + }) + .parse::() + .map_err(|_| TxStoreError::MissingInput)?], )], &delegation, &jwk, @@ -1266,7 +1422,9 @@ where .into_bytes(); let invocation = crate::events::SerializedEvent(info, serialized); let mut inputs = HashMap::new(); - inputs.insert((space, path), (metadata, stage)); + if let Some(stage) = stage { + inputs.insert((space.clone(), path.clone()), (metadata, stage)); + } let mut options = KvInvokeOptions::default(); if let Some(precondition) = precondition { let key = inputs @@ -1277,12 +1435,13 @@ where options.preconditions.insert(key, precondition); } let (_, mut outcomes) = self - .invoke_with_options_internal(invocation, inputs, options) + .invoke_with_options_mode(invocation, inputs, options, mode) .await?; let result = outcomes .drain(..) .find_map(|outcome| match outcome { - InvocationOutcome::KvWrite(hash) => Some(hash), + InvocationOutcome::KvWrite(hash) => Some(Some(hash)), + InvocationOutcome::KvDelete(hash) => Some(hash), _ => None, }) .ok_or(TxStoreError::MissingInput); @@ -1554,21 +1713,6 @@ where .await } - async fn invoke_with_options_internal( - &self, - invocation: Invocation, - inputs: InvocationInputs, - options: KvInvokeOptions, - ) -> Result<(TransactResult, Vec>), TxStoreError> - where - B: ImmutableWriteStore + ImmutableReadStore, - S: ImmutableStaging, - S::Writable: 'static + Unpin, - { - self.invoke_with_options_mode(invocation, inputs, options, InvokeMode::Internal) - .await - } - async fn invoke_with_options_mode( &self, invocation: Invocation, @@ -1581,6 +1725,21 @@ where S: ImmutableStaging, S::Writable: 'static + Unpin, { + for cap in &invocation.0.capabilities { + if let Some(resource) = cap.resource.tinycloud_resource() { + let ability = + crate::policy_capability::resolve_alias(cap.ability.as_ref().as_ref()); + if resource.service().as_str() == "kv" + && matches!(ability, "tinycloud.kv/put" | "tinycloud.kv/del") + { + if let Some(path) = resource.path() { + if !meeting_snapshot_write_allowed(mode, path.as_str()) { + return Err(TxStoreError::MeetingSnapshotProtected); + } + } + } + } + } let roots: Vec = invocation .0 .parents @@ -1707,6 +1866,21 @@ where begin_start.elapsed(), ); let tx = tx_result?; + // This must be the transaction's FIRST database operation: a preceding + // SQLite read would permit a read-to-write upgrade race. Distinct spaces + // are locked in a stable order to prevent cross-space deadlocks on PG. + let mut legacy_spaces: Vec<_> = mutation_keys + .iter() + .filter(|(_, path)| crate::meeting_legacy_guard::protects(path.as_str())) + .map(|(space, _)| space) + .collect(); + legacy_spaces.sort_by_key(|space| space.to_string()); + legacy_spaces.dedup(); + for space in legacy_spaces { + if crate::meeting_legacy_guard::lock_writer(&tx, space).await? { + return Err(TxStoreError::LegacyMeetingFrozen); + } + } // DbTxBody spans post-begin to pre-commit. The guard defaults to an // `error` outcome so any `?`/early return inside the transaction is // recorded as a failure; it is disarmed to `ok` right before commit. @@ -1768,7 +1942,9 @@ where }); // verify and commit invocation and kv operations let event = match mode { - InvokeMode::Internal => Event::InternalInvocation(Box::new(invocation), ops), + InvokeMode::Internal | InvokeMode::MeetingPublication => { + Event::InternalInvocation(Box::new(invocation), ops) + } InvokeMode::Admitted => Event::AdmittedInvocation(Box::new(invocation), ops), InvokeMode::Public => Event::Invocation(Box::new(invocation), ops), }; @@ -1940,13 +2116,15 @@ where .await .map_err(TxError::::from)? } - InvokeMode::Public | InvokeMode::Internal => invocation::verify_and_authorize( - &self.conn, - &invocation.0, - OffsetDateTime::now_utc(), - ) - .await - .map_err(TxError::::from)?, + InvokeMode::Public | InvokeMode::Internal | InvokeMode::MeetingPublication => { + invocation::verify_and_authorize( + &self.conn, + &invocation.0, + OffsetDateTime::now_utc(), + ) + .await + .map_err(TxError::::from)? + } }; let requested_spaces = invocation.0.spaces().cloned().collect::>(); @@ -3894,6 +4072,475 @@ mod test { let _db = get_db().await.unwrap(); } + async fn frozen_legacy_fixture() -> ( + SpaceDatabase, + SpaceId, + ) { + let db = get_db().await.unwrap(); + let space = test_space_id("legacy-freeze"); + space::ActiveModel { + id: Set(SpaceIdWrap(space.clone())), + } + .insert(&db.conn) + .await + .unwrap(); + db.conn.execute_unprepared("CREATE TABLE IF NOT EXISTS meeting_legacy_write_guard(space TEXT PRIMARY KEY, frozen BOOLEAN NOT NULL)").await.unwrap(); + db.conn + .execute(Statement::from_sql_and_values( + DbBackend::Sqlite, + "INSERT INTO meeting_legacy_write_guard(space,frozen) VALUES(?,true)", + [space.to_string().into()], + )) + .await + .unwrap(); + (db, space) + } + + #[tokio::test] + async fn legacy_freeze_rejects_internal_put_from_durable_guard() { + use futures::io::AsyncWriteExt; + let (db, space) = frozen_legacy_fixture().await; + let mut stage = HashBuffer::new(Vec::new()); + stage + .write_all(b"original must stay unchanged") + .await + .unwrap(); + let result = db + .invoke_internal_kv_put::( + space, + "xyz.tinycloud.tinychat/connectors/fireflies/transcript/old" + .parse() + .unwrap(), + Metadata(Default::default()), + stage, + None, + ) + .await; + assert!( + result.is_err(), + "persisted freeze must reject an internal legacy put" + ); + assert!(result + .unwrap_err() + .to_string() + .contains("legacy meeting artifacts are frozen")); + } + + #[tokio::test] + async fn legacy_freeze_rejects_native_cleanup_even_when_key_is_absent() { + let (db, space) = frozen_legacy_fixture().await; + let result = db + .invoke_internal_meeting_snapshot_delete::( + space, + "xyz.tinycloud.tinychat/connectors/google-meet/meeting/old" + .parse() + .unwrap(), + ) + .await; + assert!( + result.is_err(), + "native cleanup must not bypass a persisted legacy freeze" + ); + assert!(result + .unwrap_err() + .to_string() + .contains("legacy meeting artifacts are frozen")); + } + + fn legacy_test_invocation(space: &SpaceId, path: &Path, ability: &str) -> Invocation { + let jwk = JWK::generate_ed25519().unwrap(); + let did = DID_METHODS.generate(&jwk, "key").unwrap().to_string(); + let verification_method = format!("{did}#{}", did.rsplit(':').next().unwrap()); + let delegation = tinycloud_auth::ipld_core::cid::Cid::new_v1( + 0x55, + tinycloud_auth::multihash_codetable::Code::Blake3_256.digest(b"legacy-freeze-test"), + ); + let signed = make_invocation( + vec![( + space + .clone() + .to_resource("kv".parse().unwrap(), Some(path.clone()), None, None), + vec![ability.parse().unwrap()], + )], + &delegation, + &jwk, + &verification_method, + (OffsetDateTime::now_utc() + time::Duration::minutes(5)).unix_timestamp() as f64, + InvocationOptions { + proof: Some(vec![]), + ..Default::default() + }, + ) + .unwrap(); + let encoded = signed.encode().unwrap().into_bytes(); + crate::events::SerializedEvent( + crate::util::InvocationInfo::try_from(signed).unwrap(), + encoded, + ) + } + + #[tokio::test] + async fn legacy_freeze_covers_public_admitted_put_delete_and_deprecated_delete() { + use crate::storage::memory::MemoryStaging; + use futures::io::AsyncWriteExt; + let (db, space) = frozen_legacy_fixture().await; + let path: Path = "xyz.tinycloud.tinychat/connectors/fireflies/archive-copy/transcript/old" + .parse() + .unwrap(); + for ability in [ + "tinycloud.kv/put", + "tinycloud.kv/del", + "tinycloud.kv/delete", + ] { + for admitted in [false, true] { + let invocation = legacy_test_invocation(&space, &path, ability); + let mut inputs = HashMap::new(); + if ability == "tinycloud.kv/put" { + let mut stage = HashBuffer::new(Vec::new()); + stage.write_all(b"replacement").await.unwrap(); + inputs.insert( + (space.clone(), path.clone()), + (Metadata(Default::default()), stage), + ); + } + let result = if admitted { + db.invoke_admitted::( + AdmittedInvocation::admit(invocation, 600).await.unwrap(), + inputs, + ) + .await + } else { + db.invoke::(invocation, inputs).await + }; + assert!( + matches!(result, Err(TxStoreError::LegacyMeetingFrozen)), + "{ability} admitted={admitted}" + ); + } + } + } + + #[tokio::test] + async fn legacy_freeze_preserves_artifacts_but_allows_cursor_chat_other_space_and_native_snapshot( + ) { + use crate::storage::memory::MemoryStaging; + use futures::io::AsyncWriteExt; + let db = get_db().await.unwrap(); + let space = test_space_id("legacy-scope"); + let other = test_space_id("legacy-other"); + for id in [&space, &other] { + space::ActiveModel { + id: Set(SpaceIdWrap(id.clone())), + } + .insert(&db.conn) + .await + .unwrap(); + } + let mut keys = vec![]; + for source in ["fireflies", "google-meet", "tinycloud-transcriber"] { + for family in ["transcript", "meeting", "archive-copy/transcript"] { + let path: Path = format!("xyz.tinycloud.tinychat/connectors/{source}/{family}/old") + .parse() + .unwrap(); + let mut stage = HashBuffer::new(Vec::new()); + stage.write_all(b"original exact bytes\r\n").await.unwrap(); + let hash = db + .invoke_internal_kv_put::( + space.clone(), + path.clone(), + Metadata(Default::default()), + stage, + None, + ) + .await + .map_err(|e| e.to_string()) + .unwrap(); + keys.push((path, hash)); + } + } + db.set_legacy_meeting_write_freeze(&space, true, 0) + .await + .unwrap(); + for (path, hash) in keys { + let mut stage = HashBuffer::new(Vec::new()); + stage.write_all(b"changed").await.unwrap(); + let put = db + .invoke_internal_kv_put::( + space.clone(), + path.clone(), + Metadata(Default::default()), + stage, + None, + ) + .await; + assert!(matches!(put, Err(TxStoreError::LegacyMeetingFrozen))); + let delete = db + .invoke_internal_meeting_snapshot_delete::( + space.clone(), + path.clone(), + ) + .await; + assert!(matches!(delete, Err(TxStoreError::LegacyMeetingFrozen))); + assert_eq!(db.kv_get(&space, &path).await.unwrap().unwrap().1, hash); + } + for (id, key, native) in [ + ( + &space, + "xyz.tinycloud.tinychat/connectors/google-meet/drive-page-token", + false, + ), + (&space, "xyz.tinycloud.tinychat/chat/thread", false), + ( + &other, + "xyz.tinycloud.tinychat/connectors/fireflies/transcript/other", + false, + ), + ( + &space, + "xyz.tinycloud.tinychat/connectors/fireflies/snapshot/new/revision", + true, + ), + ] { + let path: Path = key.parse().unwrap(); + let mut stage = HashBuffer::new(Vec::new()); + stage.write_all(b"allowed").await.unwrap(); + if native { + db.invoke_internal_meeting_snapshot_put::( + id.clone(), + path.clone(), + Metadata(Default::default()), + stage, + None, + ) + .await + .map_err(|e| e.to_string()) + .unwrap(); + db.invoke_internal_meeting_snapshot_delete::(id.clone(), path) + .await + .map_err(|e| e.to_string()) + .unwrap(); + } else { + db.invoke_internal_kv_put::( + id.clone(), + path, + Metadata(Default::default()), + stage, + None, + ) + .await + .map_err(|e| e.to_string()) + .unwrap(); + } + } + } + + #[tokio::test] + async fn legacy_freeze_guard_database_failure_rejects_internal_write() { + use futures::io::AsyncWriteExt; + let (db, space) = frozen_legacy_fixture().await; + db.conn + .execute_unprepared("DROP TABLE meeting_legacy_write_guard") + .await + .unwrap(); + assert!(db.legacy_meeting_freeze_status(&space).await.is_err()); + let mut stage = HashBuffer::new(Vec::new()); + stage.write_all(b"must not save").await.unwrap(); + let result = db + .invoke_internal_kv_put::( + space, + "xyz.tinycloud.tinychat/connectors/fireflies/transcript/old" + .parse() + .unwrap(), + Metadata(Default::default()), + stage, + None, + ) + .await; + assert!(matches!(result, Err(TxStoreError::Tx(TxError::Db(_))))); + } + + #[derive(Clone, Default)] + struct LegacyPausedStore { + inner: MemoryStore, + entered: Arc, + release: Arc, + } + #[async_trait::async_trait] + impl StorageSetup for LegacyPausedStore { + type Error = std::io::Error; + async fn create(&self, space: &SpaceId) -> Result<(), Self::Error> { + self.inner.create(space).await + } + } + #[async_trait::async_trait] + impl ImmutableReadStore for LegacyPausedStore { + type Error = std::io::Error; + type Readable = futures::io::Cursor>; + async fn contains(&self, space: &SpaceId, hash: &Hash) -> Result { + self.inner.contains(space, hash).await + } + async fn read( + &self, + space: &SpaceId, + hash: &Hash, + ) -> Result>, Self::Error> { + self.inner.read(space, hash).await + } + async fn read_range( + &self, + space: &SpaceId, + hash: &Hash, + range: crate::storage::ByteRangeSpec, + ) -> Result>, Self::Error> { + self.inner.read_range(space, hash, range).await + } + } + #[async_trait::async_trait] + impl ImmutableWriteStore for LegacyPausedStore { + type Error = std::io::Error; + async fn persist( + &self, + space: &SpaceId, + stage: HashBuffer>, + ) -> Result { + self.entered.notify_one(); + self.release.notified().await; + self.inner.persist(space, stage).await + } + } + #[tokio::test] + async fn legacy_freeze_ack_waits_for_actual_in_flight_kv_storage_and_commit() { + use crate::storage::memory::MemoryStaging; + use futures::io::AsyncWriteExt; + let directory = tempfile::tempdir().unwrap(); + let url = format!( + "sqlite://{}?mode=rwc", + directory.path().join("native.sqlite").display() + ); + let storage = LegacyPausedStore::default(); + let first = SpaceDatabase::new( + Database::connect(url.clone()).await.unwrap(), + storage.clone(), + StaticSecret::new(vec![0; 32]).unwrap(), + ) + .await + .unwrap(); + // Independent DB pool and process-local mutex: only the durable lock can order these. + let second = SpaceDatabase::new( + Database::connect(url).await.unwrap(), + storage.clone(), + StaticSecret::new(vec![0; 32]).unwrap(), + ) + .await + .unwrap(); + let space = test_space_id("legacy-actual-in-flight"); + space::ActiveModel { + id: Set(SpaceIdWrap(space.clone())), + } + .insert(&first.conn) + .await + .unwrap(); + let path: Path = "xyz.tinycloud.tinychat/connectors/fireflies/transcript/inflight" + .parse() + .unwrap(); + let writer_space = space.clone(); + let writer_path = path.clone(); + let writer = tokio::spawn(async move { + let mut stage = HashBuffer::new(Vec::new()); + stage.write_all(b"in-flight original").await.unwrap(); + first + .invoke_internal_kv_put::( + writer_space, + writer_path, + Metadata(Default::default()), + stage, + None, + ) + .await + .map_err(|e| e.to_string()) + }); + tokio::time::timeout( + std::time::Duration::from_secs(10), + storage.entered.notified(), + ) + .await + .unwrap(); + let freeze_space = space.clone(); + let freeze_db = second.clone(); + let mut freeze = tokio::spawn(async move { + freeze_db + .set_legacy_meeting_write_freeze(&freeze_space, true, 0) + .await + }); + assert!( + tokio::time::timeout(std::time::Duration::from_millis(100), &mut freeze) + .await + .is_err() + ); + storage.release.notify_one(); + let original_hash = tokio::time::timeout(std::time::Duration::from_secs(10), writer) + .await + .unwrap() + .unwrap() + .unwrap(); + assert!( + tokio::time::timeout(std::time::Duration::from_secs(10), freeze) + .await + .unwrap() + .unwrap() + .unwrap() + .frozen + ); + assert_eq!( + second.kv_get(&space, &path).await.unwrap().unwrap().1, + original_hash + ); + let mut stage = HashBuffer::new(Vec::new()); + stage.write_all(b"too late").await.unwrap(); + assert!(matches!( + second + .invoke_internal_kv_put::( + space, + path, + Metadata(Default::default()), + stage, + None + ) + .await, + Err(TxStoreError::LegacyMeetingFrozen) + )); + } + + #[tokio::test] + async fn publication_cleanup_accepts_already_absent_legacy_keys() { + use sea_orm::ActiveValue::Set; + let db = get_db().await.map_err(|error| error.to_string()).unwrap(); + let space = test_space_id("publication-delete-absent"); + space::ActiveModel { + id: Set(SpaceIdWrap(space.clone())), + } + .insert(&db.conn) + .await + .map_err(|error| error.to_string()) + .unwrap(); + let key: Path = "xyz.tinycloud.tinychat/connectors/fireflies/transcript/absent" + .parse() + .unwrap(); + db.invoke_internal_meeting_snapshot_delete::( + space.clone(), + key.clone(), + ) + .await + .map_err(|error| error.to_string()) + .unwrap(); + db.invoke_internal_meeting_snapshot_delete::( + space, key, + ) + .await + .map_err(|error| error.to_string()) + .unwrap(); + } + #[test] fn kv_preconditions_require_the_expected_object_state() { let current = crate::hash::hash(b"current"); diff --git a/tinycloud-core/src/lib.rs b/tinycloud-core/src/lib.rs index 2410768c..63435752 100644 --- a/tinycloud-core/src/lib.rs +++ b/tinycloud-core/src/lib.rs @@ -10,6 +10,7 @@ pub mod events; pub mod hash; pub mod keys; pub mod manifest; +pub mod meeting_legacy_guard; pub mod migrations; pub mod models; pub mod policy_authority; diff --git a/tinycloud-core/src/meeting_legacy_guard.rs b/tinycloud-core/src/meeting_legacy_guard.rs new file mode 100644 index 00000000..51856d4d --- /dev/null +++ b/tinycloud-core/src/meeting_legacy_guard.rs @@ -0,0 +1,326 @@ +//! Generation-checked per-space preservation barrier for legacy meeting artifacts. +//! +//! The write-first UPSERT takes a PostgreSQL row lock or SQLite writer lock. +//! Callers must hold the transaction through the protected KV mutation's commit. +use crate::models::meeting_legacy_write_guard::{Column, Entity}; +use sea_orm::{ + sea_query::{Expr, OnConflict, Query}, + ColumnTrait, ConnectionTrait, DatabaseTransaction, DbErr, EntityTrait, QueryFilter, +}; +use tinycloud_auth::resource::SpaceId; + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct FreezeStatus { + pub frozen: bool, + pub generation: i64, +} + +#[derive(Debug, thiserror::Error)] +pub enum FreezeError { + #[error(transparent)] + Db(#[from] DbErr), + #[error("legacy_freeze_generation_conflict")] + GenerationConflict, + #[error("legacy_freeze_invalid_expected_generation")] + InvalidExpectedGeneration, +} + +pub fn protects(path: &str) -> bool { + let Some(tail) = path.strip_prefix("xyz.tinycloud.tinychat/connectors/") else { + return false; + }; + let Some((source, artifact)) = tail.split_once('/') else { + return false; + }; + matches!( + source, + "fireflies" | "google-meet" | "tinycloud-transcriber" + ) && ["transcript/", "meeting/", "archive-copy/transcript/"] + .iter() + .any(|prefix| artifact.starts_with(prefix)) +} + +/// A missing row is writable; a failed database read is never treated as writable. +pub async fn status(conn: &C, space: &SpaceId) -> Result { + Ok(Entity::find_by_id(space.to_string()) + .one(conn) + .await? + .map(|row| FreezeStatus { + frozen: row.frozen, + generation: row.generation, + }) + .unwrap_or_default()) +} + +async fn lock(tx: &DatabaseTransaction, space: &SpaceId) -> Result { + let statement = Query::insert() + .into_table(Entity) + .columns([Column::Space, Column::Frozen, Column::Generation]) + .values_panic([space.to_string().into(), false.into(), 0_i64.into()]) + .on_conflict( + OnConflict::column(Column::Space) + // Writers must never reset an existing flag or generation. + .update_column(Column::Space) + .to_owned(), + ) + .to_owned(); + tx.execute(tx.get_database_backend().build(&statement)) + .await?; + Entity::find_by_id(space.to_string()) + .one(tx) + .await? + .map(|row| FreezeStatus { + frozen: row.frozen, + generation: row.generation, + }) + .ok_or_else(|| DbErr::Custom("legacy meeting write guard disappeared while locked".into())) +} + +pub(crate) async fn lock_writer(tx: &DatabaseTransaction, space: &SpaceId) -> Result { + Ok(lock(tx, space).await?.frozen) +} + +pub(crate) async fn transition( + tx: &DatabaseTransaction, + space: &SpaceId, + frozen: bool, + expected: i64, +) -> Result { + let next = expected + .checked_add(1) + .filter(|_| expected >= 0) + .ok_or(FreezeError::InvalidExpectedGeneration)?; + let current = lock(tx, space).await?; + // Only the immediately following matching transition is an idempotent retry. + if current.frozen == frozen && current.generation == next { + return Ok(current); + } + if current.frozen == frozen || current.generation != expected { + return Err(FreezeError::GenerationConflict); + } + let updated = Entity::update_many() + .col_expr(Column::Frozen, Expr::value(frozen)) + .col_expr(Column::Generation, Expr::value(next)) + .filter(Column::Space.eq(space.to_string())) + .exec(tx) + .await?; + if updated.rows_affected != 1 { + return Err( + DbErr::Custom("legacy meeting write guard disappeared while locked".into()).into(), + ); + } + Ok(FreezeStatus { + frozen, + generation: next, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use sea_orm::{ConnectOptions, Database, DatabaseConnection, TransactionTrait}; + use sea_orm_migration::{MigrationTrait, SchemaManager}; + use std::time::Duration; + use tinycloud_auth::{resolver::DID_METHODS, ssi::jwk::JWK}; + + fn space(name: &str) -> SpaceId { + SpaceId::new( + DID_METHODS + .generate(&JWK::generate_ed25519().unwrap(), "key") + .unwrap(), + name.parse().unwrap(), + ) + } + + #[test] + fn legacy_freeze_scope_excludes_credentials_cursors_snapshots_and_neighbours() { + for source in ["fireflies", "google-meet", "tinycloud-transcriber"] { + let prefix = format!("xyz.tinycloud.tinychat/connectors/{source}"); + for artifact in ["transcript/id", "meeting/id", "archive-copy/transcript/id"] { + assert!(protects(&format!("{prefix}/{artifact}"))); + } + for neighbour in [ + "drive-page-token", + "credentials", + "snapshot/id/revision", + "transcript-old/id", + "archive-copy/other/id", + "meeting-state", + ] { + assert!(!protects(&format!("{prefix}/{neighbour}")), "{neighbour}"); + } + } + assert!(!protects( + "xyz.tinycloud.tinychat/connectors/other/transcript/id" + )); + assert!(!protects("xyz.tinycloud.tinychat/chat/thread")); + } + + async fn migrate(conn: &DatabaseConnection) { + crate::migrations::m20260915_000000_meeting_legacy_write_guard::Migration + .up(&SchemaManager::new(conn)) + .await + .unwrap(); + } + + async fn ordering_and_persistence(first: DatabaseConnection, second: DatabaseConnection) { + migrate(&first).await; + let target = space("legacy-freeze-order"); + let other = space("legacy-freeze-other"); + assert!(!status(&first, &target).await.unwrap().frozen); + let writer = first.begin().await.unwrap(); + assert!(!lock_writer(&writer, &target).await.unwrap()); + // This models work done after the guard, in the same KV transaction. + writer + .execute_unprepared("CREATE TABLE guard_ordering_probe(value INTEGER)") + .await + .unwrap(); + writer + .execute_unprepared("INSERT INTO guard_ordering_probe VALUES (1)") + .await + .unwrap(); + let freeze_conn = second.clone(); + let freeze_space = target.clone(); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let mut freezing = tokio::spawn(async move { + let tx = freeze_conn.begin().await.unwrap(); + started_tx.send(()).unwrap(); + transition(&tx, &freeze_space, true, 0).await.unwrap(); + // A completed freeze must observe the preceding writer's commit. + let probe = tx + .query_one(sea_orm::Statement::from_string( + tx.get_database_backend(), + "SELECT value FROM guard_ordering_probe".to_string(), + )) + .await + .unwrap() + .unwrap(); + assert_eq!(probe.try_get::("", "value").unwrap(), 1); + tx.commit().await.unwrap(); + }); + started_rx.await.unwrap(); + assert!( + tokio::time::timeout(Duration::from_millis(100), &mut freezing) + .await + .is_err(), + "freeze acknowledged before earlier write committed" + ); + writer.commit().await.unwrap(); + tokio::time::timeout(Duration::from_secs(10), freezing) + .await + .unwrap() + .unwrap(); + assert!(status(&first, &target).await.unwrap().frozen); + assert!(!status(&first, &other).await.unwrap().frozen); + let after = first.begin().await.unwrap(); + assert!(lock_writer(&after, &target).await.unwrap()); + after.commit().await.unwrap(); + assert!( + status(&second, &target).await.unwrap().frozen, + "writer UPSERT reset frozen flag" + ); + let repeat = second.begin().await.unwrap(); + transition(&repeat, &target, true, 0).await.unwrap(); + repeat.commit().await.unwrap(); + assert!(status(&first, &target).await.unwrap().frozen); + for (frozen, expected, generation) in + [(false, 1, 2), (false, 1, 2), (true, 2, 3), (true, 2, 3)] + { + let tx = first.begin().await.unwrap(); + assert_eq!( + transition(&tx, &target, frozen, expected).await.unwrap(), + FreezeStatus { frozen, generation } + ); + tx.commit().await.unwrap(); + } + // Neither an old release nor an old freeze can affect a new cycle. + for (frozen, expected) in [(false, 1), (true, 0), (true, 3), (false, 0), (true, 4)] { + let tx = second.begin().await.unwrap(); + assert!(matches!( + transition(&tx, &target, frozen, expected).await, + Err(FreezeError::GenerationConflict) + )); + tx.rollback().await.unwrap(); + } + for expected in [-1, i64::MAX] { + let tx = second.begin().await.unwrap(); + assert!(matches!( + transition(&tx, &target, false, expected).await, + Err(FreezeError::InvalidExpectedGeneration) + )); + tx.rollback().await.unwrap(); + } + assert_eq!( + status(&first, &target).await.unwrap(), + FreezeStatus { + frozen: true, + generation: 3 + } + ); + } + + #[tokio::test] + async fn legacy_freeze_sqlite_serializes_independent_connections_and_persists() { + let directory = tempfile::tempdir().unwrap(); + let url = format!( + "sqlite://{}?mode=rwc", + directory.path().join("guard.sqlite").display() + ); + let connect = || Database::connect(ConnectOptions::new(url.clone())); + let first = connect().await.unwrap(); + let second = connect().await.unwrap(); + ordering_and_persistence(first.clone(), second.clone()).await; + first.close().await.unwrap(); + second.close().await.unwrap(); + let reopened = connect().await.unwrap(); + assert!(Entity::find().one(&reopened).await.unwrap().unwrap().frozen); + } + + #[tokio::test] + async fn postgres_legacy_freeze_serializes_independent_connections_and_persists() { + let Some(url) = crate::test_support::postgres_test_url( + "postgres_legacy_freeze_serializes_independent_connections_and_persists", + ) else { + return; + }; + let admin = Database::connect(url.clone()).await.unwrap(); + let schema = format!( + "legacy_freeze_{}_{}", + std::process::id(), + time::OffsetDateTime::now_utc().unix_timestamp_nanos() + ); + admin + .execute_unprepared(&format!("CREATE SCHEMA {schema}")) + .await + .unwrap(); + let connect = || { + let mut options = ConnectOptions::new(url.clone()); + options.set_schema_search_path(schema.clone()); + Database::connect(options) + }; + let first = connect().await.unwrap(); + let second = connect().await.unwrap(); + ordering_and_persistence(first.clone(), second.clone()).await; + first.close().await.unwrap(); + second.close().await.unwrap(); + let reopened = connect().await.unwrap(); + assert!(Entity::find().one(&reopened).await.unwrap().unwrap().frozen); + reopened.close().await.unwrap(); + admin + .execute_unprepared(&format!("DROP SCHEMA {schema} CASCADE")) + .await + .unwrap(); + } + + #[tokio::test] + async fn legacy_freeze_database_failures_are_not_writable_status() { + let conn = Database::connect("sqlite::memory:").await.unwrap(); + let target = space("legacy-freeze-failed-db"); + assert!(status(&conn, &target).await.is_err()); + let tx = conn.begin().await.unwrap(); + assert!(lock_writer(&tx, &target).await.is_err()); + tx.rollback().await.unwrap(); + let tx = conn.begin().await.unwrap(); + assert!(transition(&tx, &target, true, 0).await.is_err()); + } +} diff --git a/tinycloud-core/src/migrations/m20260725_000000_request_path_indexes.rs b/tinycloud-core/src/migrations/m20260725_000000_request_path_indexes.rs index 03f75c9f..8d9ea32f 100644 --- a/tinycloud-core/src/migrations/m20260725_000000_request_path_indexes.rs +++ b/tinycloud-core/src/migrations/m20260725_000000_request_path_indexes.rs @@ -782,7 +782,7 @@ mod tests { // Apply TC-282 itself and re-collect the ANALYZE'd table stats so // the planner sees realistic cardinalities, not empty-table // defaults. - Migrator::up(&db, None).await.unwrap(); + Migrator::up(&db, Some(1)).await.unwrap(); db.execute(Statement::from_string( DbBackend::Sqlite, "ANALYZE".to_string(), @@ -819,11 +819,9 @@ mod tests { // down() drops all 11 and leaves the schema exactly as it was // before TC-282 ran. // - // TC-381: `Some(1)` assumed TC-282 was the last applied migration and - // silently began rolling back an unrelated later migration instead. - // Roll back everything from the end down to and including TC-282. - let rollback = migrations.len() as u32 - before_this; - Migrator::down(&db, Some(rollback)).await.unwrap(); + // TC-381: locate the prefix by name above and apply only TC-282, so + // this single rollback cannot accidentally target a later migration. + Migrator::down(&db, Some(1)).await.unwrap(); for (_, table, index_name) in EXPECTED_INDEXES .iter() .map(|(name, table, _)| (*name, *table, *name)) @@ -837,7 +835,7 @@ mod tests { } // up() is re-runnable thanks to `.if_not_exists()`. - Migrator::up(&db, None).await.unwrap(); + Migrator::up(&db, Some(1)).await.unwrap(); for (index_name, table, _) in EXPECTED_INDEXES { assert!( index_names(&db, table) diff --git a/tinycloud-core/src/migrations/m20260915_000000_meeting_legacy_write_guard.rs b/tinycloud-core/src/migrations/m20260915_000000_meeting_legacy_write_guard.rs new file mode 100644 index 00000000..f3645d74 --- /dev/null +++ b/tinycloud-core/src/migrations/m20260915_000000_meeting_legacy_write_guard.rs @@ -0,0 +1,86 @@ +use crate::models::meeting_legacy_write_guard; +use sea_orm_migration::prelude::*; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + manager + .create_table( + Table::create() + .table(meeting_legacy_write_guard::Entity) + .if_not_exists() + .col( + ColumnDef::new(meeting_legacy_write_guard::Column::Space) + .string() + .not_null() + .primary_key(), + ) + .col( + ColumnDef::new(meeting_legacy_write_guard::Column::Frozen) + .boolean() + .not_null(), + ) + .col( + ColumnDef::new(meeting_legacy_write_guard::Column::Generation) + .big_integer() + .not_null() + .default(0), + ) + .to_owned(), + ) + .await + } + + async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> { + // Dropping the generation would permit stale releases after a rollback. + Err(DbErr::Custom( + "legacy meeting guard rollback requires an explicit recovery plan".into(), + )) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::meeting_legacy_guard::{status, transition, FreezeStatus}; + use sea_orm::{Database, TransactionTrait}; + use tinycloud_auth::{resolver::DID_METHODS, resource::SpaceId, ssi::jwk::JWK}; + + #[tokio::test] + async fn legacy_freeze_migration_rollback_preserves_frozen_and_released_generation() { + let db = Database::connect("sqlite::memory:").await.unwrap(); + let schema = SchemaManager::new(&db); + Migration.up(&schema).await.unwrap(); + let space = SpaceId::new( + DID_METHODS + .generate(&JWK::generate_ed25519().unwrap(), "key") + .unwrap(), + "legacy-freeze-rollback".parse().unwrap(), + ); + + for (frozen, generation) in [(true, 1), (false, 2)] { + let tx = db.begin().await.unwrap(); + transition(&tx, &space, frozen, generation - 1) + .await + .unwrap(); + tx.commit().await.unwrap(); + let expected = FreezeStatus { frozen, generation }; + assert_eq!(status(&db, &space).await.unwrap(), expected); + + let error = Migration.down(&schema).await.unwrap_err(); + assert!(matches!(error, DbErr::Custom(message) if message == + "legacy meeting guard rollback requires an explicit recovery plan")); + assert!(schema + .has_table("meeting_legacy_write_guard") + .await + .unwrap()); + assert_eq!(status(&db, &space).await.unwrap(), expected); + + Migration.up(&schema).await.unwrap(); + assert_eq!(status(&db, &space).await.unwrap(), expected); + } + } +} diff --git a/tinycloud-core/src/migrations/mod.rs b/tinycloud-core/src/migrations/mod.rs index 7866415d..5719e851 100644 --- a/tinycloud-core/src/migrations/mod.rs +++ b/tinycloud-core/src/migrations/mod.rs @@ -19,6 +19,7 @@ pub mod m20260726_000000_owner_share_policy; pub mod m20260726_000001_owner_share_policy_proof; pub mod m20260726_000002_owner_share_enforcement_bytes; pub mod m20260731_000000_policy_v3; +pub mod m20260915_000000_meeting_legacy_write_guard; pub struct Migrator; @@ -46,6 +47,7 @@ impl MigratorTrait for Migrator { Box::new(m20260726_000001_owner_share_policy_proof::Migration), Box::new(m20260726_000002_owner_share_enforcement_bytes::Migration), Box::new(m20260731_000000_policy_v3::Migration), + Box::new(m20260915_000000_meeting_legacy_write_guard::Migration), ] } } diff --git a/tinycloud-core/src/models/delegation.rs b/tinycloud-core/src/models/delegation.rs index 85820e52..87382ce8 100644 --- a/tinycloud-core/src/models/delegation.rs +++ b/tinycloud-core/src/models/delegation.rs @@ -142,7 +142,7 @@ pub enum DelegationError { #[error("Unauthorized Delegator: {0}")] UnauthorizedDelegator(String), #[error("Unauthorized Capability: {0}, {1}")] - UnauthorizedCapability(Resource, Ability), + UnauthorizedCapability(Box, Ability), #[error("Cannot find parent delegation")] MissingParents, #[error("Child delegation expiry exceeds parent expiry")] @@ -395,7 +395,7 @@ async fn validate( ) }) { return Err(DelegationError::UnauthorizedCapability( - c.resource.clone(), + Box::new(c.resource.clone()), c.ability.clone(), ) .into()); @@ -432,7 +432,7 @@ async fn validate( if candidates.peek().is_none() { return Err(DelegationError::UnauthorizedCapability( - c.resource.clone(), + Box::new(c.resource.clone()), c.ability.clone(), ) .into()); diff --git a/tinycloud-core/src/models/invocation.rs b/tinycloud-core/src/models/invocation.rs index 9218b48e..22aa1de8 100644 --- a/tinycloud-core/src/models/invocation.rs +++ b/tinycloud-core/src/models/invocation.rs @@ -80,7 +80,7 @@ pub enum InvocationError { #[error("Unauthorized Invoker")] UnauthorizedInvoker(String), #[error("Unauthorized Action: {0} / {1}")] - UnauthorizedAction(Resource, Ability), + UnauthorizedAction(Box, Ability), #[error("Cannot find parent delegation")] MissingParents, #[error("No Such Key: {0}")] @@ -339,7 +339,7 @@ async fn validate( { return match dependant_caps.first() { Some(capability) => Err(InvocationError::UnauthorizedAction( - capability.resource.clone(), + Box::new(capability.resource.clone()), capability.ability.clone(), ) .into()), @@ -389,7 +389,7 @@ async fn validate( if candidates.peek().is_none() { return Err(InvocationError::UnauthorizedAction( - c.resource.clone(), + Box::new(c.resource.clone()), c.ability.clone(), ) .into()); @@ -884,6 +884,29 @@ mod tests { SpaceId::new(did, name.parse().unwrap()) } + #[test] + fn authorization_errors_fit_the_inline_result_budget() { + // Clippy's default large-error threshold is 128 bytes. Keep the shared + // errors below it at their source, including the concrete model wrappers. + for (name, size) in [ + ("InvocationError", std::mem::size_of::()), + ("invocation::Error", std::mem::size_of::()), + ( + "DelegationError", + std::mem::size_of::(), + ), + ( + "delegation::Error", + std::mem::size_of::(), + ), + ] { + assert!( + size <= 128, + "{name} is {size} bytes, exceeding the 128-byte result budget" + ); + } + } + fn test_write(space: &SpaceId, key: &str, label: &str, seq: i64) -> kv_write::Model { kv_write::Model { space: SpaceIdWrap(space.clone()), diff --git a/tinycloud-core/src/models/meeting_legacy_write_guard.rs b/tinycloud-core/src/models/meeting_legacy_write_guard.rs new file mode 100644 index 00000000..a18c29ca --- /dev/null +++ b/tinycloud-core/src/models/meeting_legacy_write_guard.rs @@ -0,0 +1,14 @@ +use sea_orm::entity::prelude::*; + +#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)] +#[sea_orm(table_name = "meeting_legacy_write_guard")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = false)] + pub space: String, + pub frozen: bool, + pub generation: i64, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} +impl ActiveModelBehavior for ActiveModel {} diff --git a/tinycloud-core/src/models/mod.rs b/tinycloud-core/src/models/mod.rs index 6ea652bc..82fa68ff 100644 --- a/tinycloud-core/src/models/mod.rs +++ b/tinycloud-core/src/models/mod.rs @@ -16,6 +16,7 @@ pub mod invocation; pub mod invocation_replay; pub mod kv_delete; pub mod kv_write; +pub mod meeting_legacy_write_guard; pub mod owner_share_policy; pub mod policy_challenge; pub mod policy_delegation; diff --git a/tinycloud-core/src/policy_authority/mod.rs b/tinycloud-core/src/policy_authority/mod.rs index 899cb22a..38542e08 100644 --- a/tinycloud-core/src/policy_authority/mod.rs +++ b/tinycloud-core/src/policy_authority/mod.rs @@ -2768,7 +2768,16 @@ mod tests { use sea_orm_migration::{MigratorTrait, SchemaManager}; let db = Database::connect("sqlite::memory:").await.unwrap(); - crate::migrations::Migrator::up(&db, None).await.unwrap(); + // Apply the prefix through this migration, not future migrations whose + // rollback contracts are independent of these policy tables. + let migrations = crate::migrations::Migrator::migrations(); + let policy_index = migrations + .iter() + .position(|migration| migration.name() == "m20260715_000000_policy_authority") + .unwrap(); + crate::migrations::Migrator::up(&db, Some(policy_index as u32 + 1)) + .await + .unwrap(); let schema = SchemaManager::new(&db); for table in [ "policy_delegation", @@ -2779,12 +2788,7 @@ mod tests { assert!(schema.has_table(table).await.unwrap(), "missing {table}"); } - let migrations = crate::migrations::Migrator::migrations(); - let policy_index = migrations - .iter() - .position(|migration| migration.name() == "m20260715_000000_policy_authority") - .unwrap(); - crate::migrations::Migrator::down(&db, Some((migrations.len() - policy_index) as u32)) + crate::migrations::Migrator::down(&db, Some(1)) .await .unwrap(); for table in [ @@ -2795,6 +2799,19 @@ mod tests { ] { assert!(!schema.has_table(table).await.unwrap(), "retained {table}"); } + + crate::migrations::Migrator::up(&db, Some(1)).await.unwrap(); + for table in [ + "policy_delegation", + "policy_challenge", + "policy_issuance_audit", + "policy_edge", + ] { + assert!( + schema.has_table(table).await.unwrap(), + "not restored {table}" + ); + } } #[test] diff --git a/tinycloud-core/src/sql/database.rs b/tinycloud-core/src/sql/database.rs index 32858991..ef33ac6e 100644 --- a/tinycloud-core/src/sql/database.rs +++ b/tinycloud-core/src/sql/database.rs @@ -7,7 +7,6 @@ use rusqlite::hooks::{AuthContext, Authorization}; use tokio::sync::{mpsc, oneshot}; use super::{ - authorizer, caveats::SqlCaveats, parser, storage::{self, StorageMode}, @@ -20,6 +19,10 @@ const MAX_BOUNDED_QUERY_BYTES: usize = 4 * 1024 * 1024; const IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(300); // 5 min enum DbMessage { + Publication { + command: serde_json::Value, + response_tx: oneshot::Sender>, + }, Execute { request: SqlRequest, caveats: Option, @@ -48,6 +51,22 @@ pub struct DatabaseHandle { } impl DatabaseHandle { + pub(crate) async fn publication( + &self, + command: serde_json::Value, + ) -> Result { + let (response_tx, response_rx) = oneshot::channel(); + self.tx + .send(DbMessage::Publication { + command, + response_tx, + }) + .await + .map_err(|_| SqlError::Internal("Database actor not available".into()))?; + response_rx + .await + .map_err(|_| SqlError::Internal("Database actor dropped response".into()))? + } pub async fn execute( &self, request: SqlRequest, @@ -137,6 +156,34 @@ pub fn spawn_actor( }; match msg { + DbMessage::Publication { + command, + response_tx, + } => { + conn.authorizer(None::) -> Authorization>); + let readonly = matches!( + command["operation"].as_str(), + Some("capabilities" | "inspect") + ); + let result = + super::publication::execute(&conn, &space_id, &command).map(|receipt| { + SqlExecutionResult { + response: SqlResponse::Query(QueryResponse { + columns: vec!["receipt".into()], + rows: vec![vec![SqlValue::Text(receipt.to_string())]], + row_count: 1, + }), + write_targets: if readonly { + vec![] + } else { + vec![crate::write_hooks::TouchedTables::supported(vec![ + "connector_meeting".into(), + ])] + }, + } + }); + let _ = response_tx.send(result); + } DbMessage::Execute { request, caveats, @@ -144,6 +191,7 @@ pub fn spawn_actor( response_tx, } => { let result = handle_message(&conn, &request, &caveats, &ability); + conn.authorizer(None::) -> Authorization>); // Post-write promotion check if result.is_ok() && matches!(mode, StorageMode::InMemory) { @@ -272,8 +320,12 @@ fn handle_message( } => { let parsed = parser::validate_sql(sql, caveats, ability)?; - let auth = - authorizer::create_authorizer(caveats.clone(), ability.to_string(), is_admin); + let auth = super::publication::authorizer( + conn, + caveats.clone(), + ability.to_string(), + is_admin, + ); conn.authorizer(Some(auth)); let result = execute_query(conn, sql, params, *max_rows, *max_bytes); @@ -296,7 +348,8 @@ fn handle_message( for stmt_sql in schema_stmts { let parsed = parser::validate_sql(stmt_sql, caveats, ability)?; write_targets.extend(parsed.write_targets); - let auth = authorizer::create_authorizer( + let auth = super::publication::authorizer( + conn, caveats.clone(), ability.to_string(), is_admin, @@ -309,8 +362,12 @@ fn handle_message( } let parsed = parser::validate_sql(sql, caveats, ability)?; - let auth = - authorizer::create_authorizer(caveats.clone(), ability.to_string(), is_admin); + let auth = super::publication::authorizer( + conn, + caveats.clone(), + ability.to_string(), + is_admin, + ); conn.authorizer(Some(auth)); let result = execute_statement(conn, sql, params, is_insert_statement(&parsed)); @@ -338,8 +395,12 @@ fn handle_message( let mut results = Vec::new(); for (stmt, is_insert) in statements.iter().zip(insert_statements) { - let auth = - authorizer::create_authorizer(caveats.clone(), ability.to_string(), is_admin); + let auth = super::publication::authorizer( + conn, + caveats.clone(), + ability.to_string(), + is_admin, + ); conn.authorizer(Some(auth)); let result = execute_statement(conn, &stmt.sql, &stmt.params, is_insert); conn.authorizer(None::) -> Authorization>); @@ -361,8 +422,12 @@ fn handle_message( let parsed = parser::validate_sql(&prepared.sql, caveats, ability)?; - let auth = - authorizer::create_authorizer(caveats.clone(), ability.to_string(), is_admin); + let auth = super::publication::authorizer( + conn, + caveats.clone(), + ability.to_string(), + is_admin, + ); conn.authorizer(Some(auth)); let result = if is_query_statement(&parsed) { @@ -734,3 +799,17 @@ mod tests { )); } } + +#[cfg(test)] +mod publication_fencing_tests { + use super::*; + #[test] + fn publication_rejects_old_unfenced_sql_writers_after_activation() { + for sql in ["PRAGMA writable_schema=ON", "INSERT INTO connector_meeting(id,source,source_id,created_at,updated_at) VALUES('x','fireflies','x','now','now')", "DELETE FROM connector_meeting", "UPDATE connector_meeting SET title='old'", "DROP TABLE connector_meeting", "UPDATE connector_publication_control SET active=0", "DROP TABLE connector_publication_snapshot", "INSERT INTO connector_meeting_alias VALUES('x','y')"] { + let conn=rusqlite::Connection::open_in_memory().unwrap(); + super::super::publication::execute(&conn,"space",&serde_json::json!({"contractVersion":3,"operation":"activate"})).unwrap(); + let result=handle_message(&conn,&SqlRequest::Execute{sql:sql.into(),params:vec![],schema:None},&None,"tinycloud.sql/admin"); + assert!(result.is_err(),"legacy write accepted: {sql}"); + } + } +} diff --git a/tinycloud-core/src/sql/mod.rs b/tinycloud-core/src/sql/mod.rs index db99d18e..522473fb 100644 --- a/tinycloud-core/src/sql/mod.rs +++ b/tinycloud-core/src/sql/mod.rs @@ -12,3 +12,5 @@ pub use types::{ BatchResponse, ExecuteResponse, QueryResponse, SqlError, SqlExecutionResult, SqlRequest, SqlResponse, SqlValue, }; + +pub mod publication; diff --git a/tinycloud-core/src/sql/publication.rs b/tinycloud-core/src/sql/publication.rs new file mode 100644 index 00000000..09669b3b --- /dev/null +++ b/tinycloud-core/src/sql/publication.rs @@ -0,0 +1,827 @@ +//! Fixed conditional TinyChat publication protocol over its existing SQL catalog. +use super::types::SqlError; +use rusqlite::{params, Connection, OptionalExtension}; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; + +pub const STATEMENT: &str = "tinycloud.meetingPublication.v3"; +pub const DATABASE: &str = "connectors"; +pub const SQL_PATH: &str = "xyz.tinycloud.tinychat/connectors"; +pub const ENVELOPE_LIMIT: usize = 2_097_152; +fn err(code: &str) -> SqlError { + SqlError::InvalidStatement(code.into()) +} +fn sql(err: rusqlite::Error) -> SqlError { + SqlError::Sqlite(err.to_string()) +} +fn text<'a>(v: &'a Value, key: &str) -> Result<&'a str, SqlError> { + v.get(key) + .and_then(Value::as_str) + .filter(|s| !s.is_empty()) + .ok_or_else(|| err("publication_invalid_command")) +} +pub fn digest(raw: &str) -> String { + hex::encode(Sha256::digest(raw.as_bytes())) +} +fn source(command: &Value) -> Result<&str, SqlError> { + let value = text(command, "source")?; + if !["fireflies", "google-meet", "tinycloud-transcriber"].contains(&value) { + return Err(err("publication_invalid_source")); + } + Ok(value) +} +fn source_id(command: &Value) -> Result<&str, SqlError> { + let value = text(command, "sourceId")?; + if value.len() > 512 || value.contains('/') || value.contains('\\') || value.contains("..") { + return Err(err("publication_invalid_identity")); + } + Ok(value) +} +fn encoded(value: &str) -> String { + value + .bytes() + .map(|b| { + if b.is_ascii_alphanumeric() || b"-_.!~*'()".contains(&b) { + (b as char).to_string() + } else { + format!("%{b:02X}") + } + }) + .collect() +} +pub fn connector_owned_path(path: &str) -> bool { + path.strip_prefix(&format!("{SQL_PATH}/")) + .is_some_and(|tail| { + matches!( + tail.split('/').next(), + Some("fireflies" | "google-meet" | "tinycloud-transcriber") + ) + }) +} +pub fn protected_snapshot_path(path: &str) -> bool { + path.strip_prefix(&format!("{SQL_PATH}/")) + .is_some_and(|tail| tail.split('/').nth(1) == Some("snapshot")) +} +pub fn active(conn: &Connection) -> bool { + conn.query_row( + "SELECT active FROM connector_publication_control WHERE id=1", + [], + |row| row.get::<_, i64>(0), + ) + .unwrap_or(0) + == 1 +} +pub fn protected_table(table: &str) -> bool { + table.eq_ignore_ascii_case("connector_meeting") + || table.eq_ignore_ascii_case("connector_meeting_alias") + || table + .to_ascii_lowercase() + .starts_with("connector_publication_") +} +pub fn authorizer( + conn: &Connection, + caveats: Option, + ability: String, + is_admin: bool, +) -> impl FnMut(rusqlite::hooks::AuthContext<'_>) -> rusqlite::hooks::Authorization { + let fenced = active(conn); + let mut ordinary = super::authorizer::create_authorizer(caveats, ability, is_admin); + move |ctx| { + use rusqlite::hooks::{AuthAction::*, Authorization}; + if fenced + && matches!(ctx.action,Pragma{pragma_name,..} if pragma_name.eq_ignore_ascii_case("writable_schema")) + { + return Authorization::Deny; + } + let table = match ctx.action { + CreateView { view_name } + | CreateTempView { view_name } + | DropView { view_name } + | DropTempView { view_name } => Some(view_name), + Insert { table_name } + | Delete { table_name } + | Update { table_name, .. } + | DropTable { table_name } + | AlterTable { table_name, .. } + | CreateIndex { table_name, .. } + | DropIndex { table_name, .. } + | CreateTable { table_name } + | CreateTempTable { table_name } + | DropTempTable { table_name } + | CreateTrigger { table_name, .. } + | DropTrigger { table_name, .. } + | CreateTempTrigger { table_name, .. } + | DropTempTrigger { table_name, .. } + | CreateTempIndex { table_name, .. } + | DropTempIndex { table_name, .. } + | CreateVtable { table_name, .. } + | DropVtable { table_name, .. } => Some(table_name), + _ => None, + }; + if table.is_some_and(|table| { + (fenced && protected_table(table)) + || table + .to_ascii_lowercase() + .starts_with("connector_publication_") + || table.eq_ignore_ascii_case("connector_meeting_alias") + }) { + Authorization::Deny + } else { + ordinary(ctx) + } + } +} +fn schema(conn: &Connection) -> Result<(), SqlError> { + conn.execute_batch("PRAGMA writable_schema=OFF;") + .map_err(sql)?; + let triggers:bool=conn.query_row("SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='trigger' AND lower(sql) LIKE '%connector_%')",[],|r|r.get(0)).map_err(sql)?; + if triggers { + return Err(err("publication_legacy_trigger_requires_review")); + } + let shadow: bool = conn + .query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_temp_master WHERE lower(name) LIKE 'connector_%')", + [], + |r| r.get(0), + ) + .map_err(sql)?; + if shadow { + return Err(err("publication_temp_shadow")); + } + conn.execute_batch("CREATE TABLE IF NOT EXISTS connector_meeting(id TEXT PRIMARY KEY,source TEXT NOT NULL,source_id TEXT NOT NULL,title TEXT,started_at TEXT,duration_secs INTEGER,organizer_email TEXT,participants TEXT,summary_overview TEXT,summary_action_items TEXT,keywords TEXT,meeting_type TEXT,metadata TEXT,created_at TEXT NOT NULL,updated_at TEXT NOT NULL); + CREATE TABLE IF NOT EXISTS connector_publication_deletion(operation_id TEXT PRIMARY KEY,command TEXT NOT NULL,receipt TEXT NOT NULL); + CREATE TABLE IF NOT EXISTS connector_publication_control(id INTEGER PRIMARY KEY,active INTEGER NOT NULL); + CREATE TABLE IF NOT EXISTS connector_meeting_alias(alias TEXT PRIMARY KEY,meeting_id TEXT NOT NULL); + CREATE TABLE IF NOT EXISTS connector_publication_operation(operation_id TEXT PRIMARY KEY,source TEXT NOT NULL,source_id TEXT NOT NULL,meeting_id TEXT NOT NULL,generation INTEGER NOT NULL,expected_head TEXT,created_at TEXT NOT NULL,inserted INTEGER NOT NULL); + CREATE TABLE IF NOT EXISTS connector_publication_snapshot(revision TEXT PRIMARY KEY,snapshot_key TEXT NOT NULL UNIQUE,meeting_id TEXT NOT NULL,operation_id TEXT NOT NULL,generation INTEGER NOT NULL,snapshot_metadata TEXT NOT NULL,staged INTEGER NOT NULL DEFAULT 0,published INTEGER NOT NULL DEFAULT 0);").map_err(sql)?; + let columns = conn + .prepare("PRAGMA table_info(connector_meeting)") + .map_err(sql)? + .query_map([], |row| row.get::<_, String>(1)) + .map_err(sql)? + .collect::, _>>() + .map_err(sql)?; + for (column, kind) in [ + ("head_revision", "TEXT"), + ("head_snapshot_key", "TEXT"), + ("publication_state", "TEXT"), + ("publication_operation", "TEXT"), + ("publication_head_operation", "TEXT"), + ("publication_generation", "INTEGER NOT NULL DEFAULT 0"), + ("publication_unavailable_reason", "TEXT"), + ] { + if !columns.iter().any(|existing| existing == column) { + conn.execute( + &format!("ALTER TABLE connector_meeting ADD COLUMN {column} {kind}"), + [], + ) + .map_err(sql)?; + } + } + let collisions:i64=conn.query_row("SELECT COUNT(*) FROM (SELECT source,source_id FROM connector_meeting GROUP BY source,source_id HAVING COUNT(*)>1)",[],|row|row.get(0)).map_err(sql)?; + if collisions == 0 { + conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS connector_publication_identity ON connector_meeting(source,source_id)",[]).map_err(sql)?; + } + conn.execute("UPDATE connector_meeting SET publication_state='unavailable',publication_unavailable_reason='original_not_verified' WHERE head_revision IS NULL AND (publication_state IS NULL OR publication_state='unverified')",[]).map_err(sql)?; + conn.execute("UPDATE connector_meeting SET publication_state='unavailable',publication_unavailable_reason='identity_collision' WHERE (publication_state IS NULL OR publication_state != 'deleted') AND (source,source_id) IN (SELECT source,source_id FROM connector_meeting GROUP BY source,source_id HAVING COUNT(*)>1)",[]).map_err(sql)?; + conn.execute("INSERT INTO connector_publication_control VALUES(1,1) ON CONFLICT(id) DO UPDATE SET active=1",[]).map_err(sql)?; + Ok(()) +} +#[derive(Debug)] +struct Head { + id: String, + revision: Option, + key: Option, + operation: Option, + head_operation: Option, + generation: i64, + state: Option, + created_at: String, +} +fn head(conn: &Connection, source: &str, id: &str) -> Result, SqlError> { + let count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM connector_meeting WHERE source=? AND source_id=?", + params![source, id], + |r| r.get(0), + ) + .map_err(sql)?; + if count > 1 { + return Err(err("publication_identity_collision")); + } + conn.query_row("SELECT id,head_revision,head_snapshot_key,publication_operation,publication_head_operation,publication_generation,publication_state,created_at FROM connector_meeting WHERE source=? AND source_id=?",params![source,id],|r|Ok(Head{id:r.get(0)?,revision:r.get(1)?,key:r.get(2)?,operation:r.get(3)?,head_operation:r.get(4)?,generation:r.get(5)?,state:r.get(6)?,created_at:r.get(7)?})).optional().map_err(sql) +} +fn current(conn: &Connection, command: &Value) -> Result { + let row = head(conn, source(command)?, source_id(command)?)? + .ok_or_else(|| err("publication_reservation_missing"))?; + if row.operation.as_deref() != Some(text(command, "operationId")?) + || Some(row.generation) != command["generation"].as_i64() + || row.revision.as_deref() != command["expectedHead"].as_str() + || row.state.as_deref() != Some("reserved") + || Some(row.id.as_str()) != command["meetingRef"].as_str() + { + return Err(err("publication_superseded")); + } + Ok(row) +} +fn previous(conn: &Connection, head: &Head) -> Result { + let Some(revision) = &head.revision else { + return conn.query_row("SELECT source,source_id,title,started_at,duration_secs,organizer_email,participants,summary_overview,summary_action_items,keywords,meeting_type,metadata FROM connector_meeting WHERE id=?",[&head.id],|r|{ + let json_cell=|i|{let raw:Option=r.get(i)?;Ok::(raw.and_then(|raw|serde_json::from_str(&raw).ok()).unwrap_or(Value::Null))}; + Ok(json!({"id":head.id,"source":r.get::<_,String>(0)?,"sourceId":r.get::<_,String>(1)?,"title":r.get::<_,Option>(2)?,"startedAt":r.get::<_,Option>(3)?,"durationSecs":r.get::<_,Option>(4)?,"organizerEmail":r.get::<_,Option>(5)?,"participants":json_cell(6)?.as_array().cloned().unwrap_or_default(),"summaryOverview":r.get::<_,Option>(7)?,"summaryActionItems":r.get::<_,Option>(8)?,"keywords":json_cell(9)?,"meetingType":r.get::<_,Option>(10)?,"metadata":json_cell(11)?.as_object().cloned().unwrap_or_default()})) + }).map_err(sql); + }; + let raw:Option=conn.query_row("SELECT snapshot_metadata FROM connector_publication_snapshot WHERE revision=? AND staged=1 AND published=1",[revision],|r|r.get(0)).optional().map_err(sql)?; + let Some(raw) = raw else { + return Ok(Value::Null); + }; + let snapshot: Value = + serde_json::from_str(&raw).map_err(|_| err("publication_snapshot_invalid"))?; + let m = &snapshot["metadata"]; + let fields = &m["metadata"]["connector_fields"]; + let participants = m["participants"] + .as_array() + .cloned() + .unwrap_or_default() + .into_iter() + .map(|p| json!({"name":p["name"],"email":p.get("email").cloned().unwrap_or(Value::Null)})) + .collect::>(); + Ok( + json!({"id":head.id,"source":snapshot["source"],"sourceId":snapshot["sourceId"],"title":m["title"],"startedAt":m["startedAt"],"organizerEmail":m["organizerEmail"],"participants":participants,"metadata":m["metadata"],"durationSecs":fields["durationSecs"],"summaryOverview":snapshot["overview"]["text"],"summaryActionItems":fields["summaryActionItems"],"keywords":fields["keywords"],"meetingType":fields["meetingType"]}), + ) +} +fn reservation( + conn: &Connection, + row: &Head, + operation: &str, + inserted: bool, +) -> Result { + Ok( + json!({"contractVersion":3,"status":"reserved","operationId":operation,"generation":row.generation,"expectedHead":row.revision,"meetingRef":row.id,"inserted":inserted,"createdAt":row.created_at,"previousMeeting":previous(conn,row)?}), + ) +} +pub fn validate_snapshot(command: &Value) -> Result { + let raw = text(command, "snapshotRaw")?; + if raw.len() > ENVELOPE_LIMIT { + return Err(err("publication_capacity")); + } + let revision = text(command, "revision")?; + if revision.len() != 64 || digest(raw) != revision { + return Err(err("publication_digest_mismatch")); + } + let snapshot: Value = + serde_json::from_str(raw).map_err(|_| err("publication_snapshot_invalid"))?; + for field in ["meetingRef", "source", "sourceId", "operationId"] { + if snapshot.get(field) != command.get(field) { + return Err(err("publication_identity_mismatch")); + } + } + if snapshot["contractVersion"] != 3 + || !snapshot["metadata"].is_object() + || !snapshot["metadata"]["participants"].is_array() + || !snapshot["metadata"]["metadata"].is_object() + || !snapshot["aliases"].is_array() + { + return Err(err("publication_snapshot_invalid")); + } + if !snapshot["body"].is_null() { + let body = &snapshot["body"]; + let raw = body["raw"] + .as_str() + .ok_or_else(|| err("publication_snapshot_invalid"))?; + if raw.len() > 1_048_576 + || body["encoding"] != "utf-8" + || body["original"]["byteLength"].as_u64() != Some(raw.len() as u64) + || body["original"]["digest"].as_str() != Some(digest(raw).as_str()) + { + return Err(err("publication_original_mismatch")); + } + } + let expected = format!( + "{SQL_PATH}/{}/snapshot/{}/{revision}", + source(command)?, + encoded(source_id(command)?) + ); + if command["snapshotKey"].as_str() != Some(expected.as_str()) { + return Err(err("publication_snapshot_key_mismatch")); + } + Ok(snapshot) +} +pub fn execute(conn: &Connection, space: &str, command: &Value) -> Result { + if command["contractVersion"] != 3 { + return Err(err("publication_upgrade_required")); + } + let operation = text(command, "operation")?; + if operation == "capabilities" { + return Ok( + json!({"contractVersion":3,"writerFencing":true,"snapshotImmutability":true,"digestVerification":true}), + ); + } + let tx = conn.unchecked_transaction().map_err(sql)?; + if operation == "activate" { + schema(&tx)?; + tx.commit().map_err(sql)?; + return Ok(json!({"contractVersion":3,"status":"ready"})); + } + if !active(&tx) { + return Err(err("publication_activation_required")); + } + let now = time::OffsetDateTime::now_utc() + .format(&time::format_description::well_known::Rfc3339) + .map_err(|_| err("publication_time_invalid"))?; + let result = match operation { + "reserve" => { + let source = source(command)?; + let source_id = source_id(command)?; + let op = text(command, "operationId")?; + if op.len() > 128 { + return Err(err("publication_invalid_operation")); + } + let old = head(&tx, source, source_id)?; + let deleted:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM connector_publication_deletion WHERE operation_id=?)",[op],|r|r.get(0)).map_err(sql)?; + if deleted { + return Err(err("publication_operation_reused")); + } + let seen:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM connector_publication_operation WHERE operation_id=?)",[op],|r|r.get(0)).map_err(sql)?; + if seen { + let row = old.ok_or_else(|| err("publication_superseded"))?; + if row.operation.as_deref() != Some(op) || row.state.as_deref() != Some("reserved") + { + return Err(err("publication_superseded")); + } + reservation(&tx, &row, op, false)? + } else { + let inserted = old.is_none(); + let id = old + .as_ref() + .map(|h| h.id.clone()) + .unwrap_or_else(|| digest(&json!([space, source, source_id]).to_string())); + if inserted { + tx.execute("INSERT INTO connector_meeting(id,source,source_id,participants,metadata,created_at,updated_at,publication_generation,publication_state) VALUES(?,?,?,'[]','{}',?,?,0,'unverified')",params![id,source,source_id,now,now]).map_err(sql)?; + } + tx.execute("UPDATE connector_meeting SET publication_generation=publication_generation+1,publication_operation=?,publication_state='reserved' WHERE id=?",params![op,id]).map_err(sql)?; + let row = head(&tx, source, source_id)?.unwrap(); + tx.execute( + "INSERT INTO connector_publication_operation VALUES(?,?,?,?,?,?,?,?)", + params![ + op, + source, + source_id, + id, + row.generation, + row.revision, + now, + inserted + ], + ) + .map_err(sql)?; + reservation(&tx, &row, op, inserted)? + } + } + "prepare_stage" => { + let row = current(&tx, command)?; + let mut snapshot = validate_snapshot(command)?; + snapshot["body"] + .as_object_mut() + .map(|body| body.remove("raw")); + tx.execute("INSERT INTO connector_publication_snapshot(revision,snapshot_key,meeting_id,operation_id,generation,snapshot_metadata) VALUES(?,?,?,?,?,?) ON CONFLICT(revision) DO NOTHING",params![text(command,"revision")?,text(command,"snapshotKey")?,row.id,text(command,"operationId")?,row.generation,snapshot.to_string()]).map_err(sql)?; + json!({"contractVersion":3,"status":"prepared"}) + } + "stage" => { + let row = current(&tx, command)?; + let changed=tx.execute("UPDATE connector_publication_snapshot SET staged=1 WHERE revision=? AND snapshot_key=? AND meeting_id=? AND operation_id=? AND generation=?",params![text(command,"revision")?,text(command,"snapshotKey")?,row.id,text(command,"operationId")?,row.generation]).map_err(sql)?; + if changed != 1 { + return Err(err("publication_stage_missing")); + } + json!({"contractVersion":3,"status":"staged","revision":command["revision"],"snapshotKey":command["snapshotKey"]}) + } + "publish" => { + let source = source(command)?; + let source_id = source_id(command)?; + let existing = head(&tx, source, source_id)? + .ok_or_else(|| err("publication_reservation_missing"))?; + if existing.state.as_deref() == Some("published") + && existing.head_operation.as_deref() == command["operationId"].as_str() + && existing.revision.as_deref() == command["revision"].as_str() + { + json!({"contractVersion":3,"status":"published","meetingRef":existing.id,"operationId":command["operationId"],"revision":existing.revision,"snapshotKey":existing.key}) + } else { + let row = current(&tx, command)?; + let raw:Option=tx.query_row("SELECT snapshot_metadata FROM connector_publication_snapshot WHERE revision=? AND operation_id=? AND generation=? AND snapshot_key=? AND staged=1",params![text(command,"revision")?,text(command,"operationId")?,row.generation,text(command,"snapshotKey")?],|r|r.get(0)).optional().map_err(sql)?; + let snapshot: Value = + serde_json::from_str(&raw.ok_or_else(|| err("publication_stage_missing"))?) + .map_err(|_| err("publication_snapshot_invalid"))?; + let m = &snapshot["metadata"]; + let fields = &m["metadata"]["connector_fields"]; + for alias in snapshot["aliases"].as_array().unwrap() { + let alias = alias + .as_str() + .filter(|a| !a.is_empty() && a.len() <= 128) + .ok_or_else(|| err("publication_alias_invalid"))?; + let collision:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM connector_meeting WHERE id=? AND id!=? UNION ALL SELECT 1 FROM connector_meeting_alias WHERE alias=? AND meeting_id!=?)",params![alias,row.id,alias,row.id],|r|r.get(0)).map_err(sql)?; + if collision { + return Err(err("publication_alias_collision")); + } + tx.execute("INSERT INTO connector_meeting_alias VALUES(?,?) ON CONFLICT(alias) DO NOTHING",params![alias,row.id]).map_err(sql)?; + } + tx.execute("UPDATE connector_meeting SET title=?,started_at=?,duration_secs=?,organizer_email=?,participants=?,summary_overview=?,summary_action_items=?,keywords=?,meeting_type=?,metadata=?,updated_at=?,head_revision=?,head_snapshot_key=?,publication_head_operation=?,publication_state='published',publication_unavailable_reason=NULL WHERE id=?",params![m["title"].as_str(),m["startedAt"].as_str(),fields["durationSecs"].as_f64(),m["organizerEmail"].as_str(),m["participants"].to_string(),snapshot["overview"]["text"].as_str(),fields["summaryActionItems"].as_str(),if fields["keywords"].is_null(){None}else{Some(fields["keywords"].to_string())},fields["meetingType"].as_str(),m["metadata"].to_string(),now,text(command,"revision")?,text(command,"snapshotKey")?,text(command,"operationId")?,row.id]).map_err(sql)?; + tx.execute( + "UPDATE connector_publication_snapshot SET published=1 WHERE revision=?", + [text(command, "revision")?], + ) + .map_err(sql)?; + json!({"contractVersion":3,"status":"published","meetingRef":row.id,"operationId":command["operationId"],"revision":command["revision"],"snapshotKey":command["snapshotKey"]}) + } + } + "inspect" => { + let row = head(&tx, source(command)?, source_id(command)?)? + .ok_or_else(|| err("publication_reservation_missing"))?; + json!({"contractVersion":3,"status":if row.head_operation.as_deref()==command["operationId"].as_str()&&row.revision.is_some(){"published"}else{"superseded"},"meetingRef":row.id,"operationId":row.head_operation,"revision":row.revision,"snapshotKey":row.key}) + } + "delete" | "purge" => { + let source = source(command)?; + let op = text(command, "operationId")?; + let mut public_command = command.clone(); + public_command + .as_object_mut() + .unwrap() + .remove("cleanupKeys"); + let public_command_raw = public_command.to_string(); + let seen:Option<(String,String)>=tx.query_row("SELECT command,receipt FROM connector_publication_deletion WHERE operation_id=?",[op],|r|Ok((r.get(0)?,r.get(1)?))).optional().map_err(sql)?; + if let Some((prior, receipt)) = seen { + if prior != public_command_raw { + return Err(err("publication_operation_reused")); + } + return serde_json::from_str(&receipt) + .map_err(|_| err("publication_receipt_invalid")); + } + let reserved:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM connector_publication_operation WHERE operation_id=?)",[op],|r|r.get(0)).map_err(sql)?; + if reserved { + return Err(err("publication_operation_reused")); + } + let condition = if operation == "delete" { + "source=?1 AND source_id=?2" + } else { + "source=?1" + }; + let source_id = if operation == "delete" { + source_id(command)? + } else { + "" + }; + let query=format!("SELECT snapshot_key FROM connector_publication_snapshot WHERE meeting_id IN (SELECT id FROM connector_meeting WHERE {condition})"); + let mut keys = if operation == "delete" { + tx.prepare(&query) + .map_err(sql)? + .query_map(params![source, source_id], |r| r.get::<_, String>(0)) + .map_err(sql)? + .collect::, _>>() + .map_err(sql)? + } else { + tx.prepare(&query) + .map_err(sql)? + .query_map([source], |r| r.get::<_, String>(0)) + .map_err(sql)? + .collect::, _>>() + .map_err(sql)? + }; + if operation == "delete" { + keys.push(format!("{SQL_PATH}/{source}/transcript/{source_id}")); + } + if let Some(cleanup) = command["cleanupKeys"].as_array() { + for key in cleanup { + let key = key + .as_str() + .filter(|key| key.starts_with(&format!("{SQL_PATH}/{source}/"))) + .ok_or_else(|| err("publication_cleanup_invalid"))?; + keys.push(key.to_owned()); + } + } + keys.sort(); + keys.dedup(); + let count_query=format!("SELECT COUNT(*) FROM connector_meeting WHERE {condition} AND (publication_state IS NULL OR publication_state!='deleted')"); + let deleted_count: i64 = if operation == "delete" { + tx.query_row(&count_query, params![source, source_id], |r| r.get(0)) + .map_err(sql)? + } else { + tx.query_row(&count_query, [source], |r| r.get(0)) + .map_err(sql)? + }; + let revoke=format!("UPDATE connector_publication_snapshot SET published=0,staged=0 WHERE meeting_id IN (SELECT id FROM connector_meeting WHERE {condition})"); + if operation == "delete" { + tx.execute(&revoke, params![source, source_id]) + .map_err(sql)?; + } else { + tx.execute(&revoke, [source]).map_err(sql)?; + } + let update=format!("UPDATE connector_meeting SET title=NULL,started_at=NULL,duration_secs=NULL,organizer_email=NULL,participants='[]',summary_overview=NULL,summary_action_items=NULL,keywords=NULL,meeting_type=NULL,metadata='{{}}',publication_generation=publication_generation+1,publication_operation=?3,publication_head_operation=NULL,head_revision=NULL,head_snapshot_key=NULL,publication_state='deleted' WHERE {condition}"); + // Numbered binds keep the source-wide command's unused second bind explicit. + tx.execute(&update, params![source, source_id, op]) + .map_err(sql)?; + let receipt = json!({"contractVersion":3,"status":if operation=="delete"{"deleted"}else{"purged"},"operationId":op,"snapshotKeys":keys,"deletedCount":deleted_count}); + tx.execute( + "INSERT INTO connector_publication_deletion VALUES(?,?,?)", + params![op, public_command_raw, receipt.to_string()], + ) + .map_err(sql)?; + receipt + } + _ => return Err(err("publication_invalid_operation")), + }; + tx.commit().map_err(sql)?; + Ok(result) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + use sha2::{Digest, Sha256}; + fn call(conn: &Connection, mut command: Value) -> Result { + command["contractVersion"] = json!(3); + execute(conn, "synthetic-space", &command) + } + fn ready() -> Connection { + let conn = Connection::open_in_memory().unwrap(); + call(&conn, json!({"operation":"activate"})).unwrap(); + conn + } + fn reserve(conn: &Connection, op: &str) -> Value { + call(conn,json!({"operation":"reserve","source":"fireflies","sourceId":"source","operationId":op})).unwrap() + } + fn stage(conn: &Connection, reservation: &Value, text: &str) -> Value { + stage_with_metadata(conn, reservation, text, json!({})) + } + fn stage_with_metadata( + conn: &Connection, + reservation: &Value, + text: &str, + metadata: Value, + ) -> Value { + let raw=json!({"contractVersion":3,"source":"fireflies","sourceId":"source","meetingRef":reservation["meetingRef"],"operationId":reservation["operationId"],"createdAt":"2026-09-14T00:00:00Z","metadata":{"title":text,"startedAt":null,"organizerEmail":null,"participants":[],"metadata":metadata},"body":{"basis":"transcript","encoding":"utf-8","schema":"text","raw":text,"original":{"digest":hex::encode(Sha256::digest(text.as_bytes())),"byteLength":text.len(),"recordCount":1,"extent":"unknown","captureComplete":null},"omissions":[]},"overview":null,"aliases":[]}).to_string(); + let revision = hex::encode(Sha256::digest(raw.as_bytes())); + let mut command = reservation.clone(); + command["source"] = json!("fireflies"); + command["sourceId"] = json!("source"); + command["revision"] = json!(revision); + command["snapshotKey"] = json!(format!("{SQL_PATH}/fireflies/snapshot/source/{revision}")); + command["snapshotRaw"] = json!(raw); + command["operation"] = json!("prepare_stage"); + call(conn, command.clone()).unwrap(); + command["operation"] = json!("stage"); + call(conn, command.clone()).unwrap(); + command.as_object_mut().unwrap().remove("snapshotRaw"); + command + } + fn legacy_real_duration(duration: Option) -> Connection { + let conn = Connection::open_in_memory().unwrap(); + conn.execute_batch("CREATE TABLE connector_meeting(id TEXT PRIMARY KEY,source TEXT NOT NULL,source_id TEXT NOT NULL,title TEXT,started_at TEXT,duration_secs REAL,organizer_email TEXT,participants TEXT,summary_overview TEXT,summary_action_items TEXT,keywords TEXT,meeting_type TEXT,metadata TEXT,created_at TEXT NOT NULL,updated_at TEXT NOT NULL);").unwrap(); + conn.execute("INSERT INTO connector_meeting(id,source,source_id,duration_secs,created_at,updated_at) VALUES('legacy','fireflies','source',?,'now','now')",[duration]).unwrap(); + let storage_type: String = conn + .query_row( + "SELECT typeof(duration_secs) FROM connector_meeting", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!( + storage_type, + if duration.is_some() { "real" } else { "null" } + ); + call(&conn, json!({"operation":"activate"})).unwrap(); + conn + } + #[test] + fn publication_reserves_legacy_integral_real_duration() { + let conn = legacy_real_duration(Some(600.0)); + let reserved = reserve(&conn, "integral"); + assert_eq!( + reserved["previousMeeting"]["durationSecs"].as_f64(), + Some(600.0) + ); + } + #[test] + fn publication_reserves_legacy_fractional_real_duration() { + let conn = legacy_real_duration(Some(1.25)); + let reserved = reserve(&conn, "fractional"); + assert_eq!( + reserved["previousMeeting"]["durationSecs"].as_f64(), + Some(1.25) + ); + } + #[test] + fn publication_reserves_legacy_null_duration() { + let conn = legacy_real_duration(None); + assert!(reserve(&conn, "null")["previousMeeting"]["durationSecs"].is_null()); + } + #[test] + fn publication_preserves_numeric_and_null_durations_in_catalog_and_next_reservation() { + for duration in [json!(600), json!(600.0), json!(1.25), Value::Null] { + let conn = ready(); + let reserved = reserve(&conn, "duration"); + let mut command = stage_with_metadata( + &conn, + &reserved, + "synthetic original", + json!({"connector_fields":{"durationSecs":duration}}), + ); + command["operation"] = json!("publish"); + call(&conn, command).unwrap(); + let stored: Option = conn + .query_row("SELECT duration_secs FROM connector_meeting", [], |r| { + r.get(0) + }) + .unwrap(); + assert_eq!(stored, duration.as_f64(), "published duration {duration}"); + assert_eq!( + reserve(&conn, "next")["previousMeeting"]["durationSecs"], + duration + ); + } + } + #[test] + fn publication_capabilities_and_first_insert_race() { + let conn = ready(); + assert_eq!( + call(&conn, json!({"operation":"capabilities"})).unwrap()["writerFencing"], + true + ); + let a = reserve(&conn, "a"); + let b = reserve(&conn, "b"); + assert_eq!(a["meetingRef"], b["meetingRef"]); + assert_eq!(a["inserted"], true); + assert_eq!(b["inserted"], false); + assert!(b["generation"].as_i64() > a["generation"].as_i64()); + assert!(call(&conn,json!({"operation":"reserve","source":"fireflies","sourceId":"source","operationId":"a"})).is_err()); + } + #[test] + fn publication_preserves_head_and_fences_stale_workers() { + let conn = ready(); + let a = reserve(&conn, "a"); + let mut a = stage(&conn, &a, "first"); + a["operation"] = json!("publish"); + call(&conn, a.clone()).unwrap(); + let b = reserve(&conn, "b"); + assert_eq!(b["expectedHead"], a["revision"]); + assert_eq!(b["previousMeeting"]["title"], "first"); + let mut b = stage(&conn, &b, "new"); + let _c = reserve(&conn, "c"); + b["operation"] = json!("publish"); + assert!(call(&conn, b).is_err()); + assert_eq!(call(&conn,json!({"operation":"inspect","source":"fireflies","sourceId":"source","operationId":"a"})).unwrap()["revision"],a["revision"]); + } + #[test] + fn publication_lost_ack_after_newer_head_is_superseded() { + let conn = ready(); + let a = reserve(&conn, "a"); + let mut a = stage(&conn, &a, "first"); + a["operation"] = json!("publish"); + call(&conn, a.clone()).unwrap(); + let b = reserve(&conn, "b"); + let mut b = stage(&conn, &b, "second"); + b["operation"] = json!("publish"); + call(&conn, b).unwrap(); + assert_eq!(call(&conn,json!({"operation":"inspect","source":"fireflies","sourceId":"source","operationId":"a"})).unwrap()["status"],"superseded"); + assert!(call(&conn, a).is_err()); + } + #[test] + fn publication_delete_recreate_fences_old_stage() { + let conn = ready(); + let a = reserve(&conn, "a"); + let mut a = stage(&conn, &a, "first"); + call(&conn,json!({"operation":"delete","source":"fireflies","sourceId":"source","operationId":"del"})).unwrap(); + let b = reserve(&conn, "b"); + assert!(b["generation"].as_i64() > a["generation"].as_i64()); + a["operation"] = json!("publish"); + assert!(call(&conn, a).is_err()); + assert!(b["expectedHead"].is_null()); + } + #[test] + fn publication_rejects_bad_digests_and_cannot_publish_unstaged_body() { + let conn = ready(); + let a = reserve(&conn, "a"); + let mut bad = a.clone(); + bad["operation"] = json!("publish"); + bad["revision"] = json!("a".repeat(64)); + bad["source"] = json!("fireflies"); + bad["sourceId"] = json!("source"); + assert!(call(&conn, bad).is_err()); + } + #[test] + fn publication_delete_retry_does_not_delete_recreated_meeting() { + let conn = ready(); + let a = reserve(&conn, "a"); + let _a = stage(&conn, &a, "first"); + let command = json!({"operation":"delete","source":"fireflies","sourceId":"source","operationId":"del"}); + let first = call(&conn, command.clone()).unwrap(); + let b = reserve(&conn, "b"); + let mut b = stage(&conn, &b, "second"); + b["operation"] = json!("publish"); + call(&conn, b.clone()).unwrap(); + assert_eq!(call(&conn, command).unwrap(), first); + assert_eq!(call(&conn,json!({"operation":"inspect","source":"fireflies","sourceId":"source","operationId":"b"})).unwrap()["revision"],b["revision"]); + } + #[test] + fn publication_rejects_changed_key_between_stage_and_publish() { + let conn = ready(); + let a = reserve(&conn, "a"); + let mut a = stage(&conn, &a, "first"); + a["operation"] = json!("publish"); + a["snapshotKey"] = json!("arbitrary-key"); + assert!(call(&conn, a).is_err()); + } + #[test] + fn publication_activation_preserves_legacy_collisions_and_metadata() { + let conn = Connection::open_in_memory().unwrap(); + schema(&conn).unwrap(); + conn.execute("DROP INDEX connector_publication_identity", []) + .unwrap(); + conn.execute("INSERT INTO connector_meeting(id,source,source_id,title,started_at,summary_overview,participants,metadata,created_at,updated_at) VALUES('old','fireflies','source','Older title','2025-01-01T10:00:00Z','Old summary','[]','{}','2025-01-01','2025-01-01'),('dup1','fireflies','collision','one',NULL,NULL,'[]','{}','2025-01-01','2025-01-01'),('dup2','fireflies','collision','two',NULL,NULL,'[]','{}','2025-01-01','2025-01-01')",[]).unwrap(); + call(&conn, json!({"operation":"activate"})).unwrap(); + let old = reserve(&conn, "migrate"); + assert_eq!(old["meetingRef"], "old"); + assert_eq!(old["previousMeeting"]["startedAt"], "2025-01-01T10:00:00Z"); + assert_eq!(old["previousMeeting"]["summaryOverview"], "Old summary"); + assert!(call(&conn,json!({"operation":"reserve","source":"fireflies","sourceId":"collision","operationId":"dup"})).is_err()); + assert_eq!(conn.query_row("SELECT COUNT(*) FROM connector_meeting WHERE publication_unavailable_reason='identity_collision'",[],|r|r.get::<_,i64>(0)).unwrap(),2); + } + #[test] + fn publication_legacy_null_metadata_has_safe_previous_shape() { + let conn = ready(); + conn.execute("INSERT INTO connector_meeting(id,source,source_id,created_at,updated_at) VALUES('legacy','fireflies','source','2025-01-01','2025-01-01')",[]).unwrap(); + let row = reserve(&conn, "migrate"); + assert_eq!(row["previousMeeting"]["participants"], json!([])); + assert_eq!(row["previousMeeting"]["metadata"], json!({})); + } + + #[test] + fn publication_membership_excludes_staged_and_revoked_revisions() { + let conn = ready(); + let a = reserve(&conn, "a"); + let mut a = stage(&conn, &a, "deleted content"); + assert_eq!( + conn.query_row( + "SELECT published FROM connector_publication_snapshot", + [], + |r| r.get::<_, i64>(0) + ) + .unwrap(), + 0 + ); + a["operation"] = json!("publish"); + call(&conn, a).unwrap(); + assert_eq!( + conn.query_row( + "SELECT published FROM connector_publication_snapshot", + [], + |r| r.get::<_, i64>(0) + ) + .unwrap(), + 1 + ); + let deleted=call(&conn,json!({"operation":"delete","source":"fireflies","sourceId":"source","operationId":"delete"})).unwrap(); + assert_eq!(deleted["deletedCount"], 1); + let b = reserve(&conn, "b"); + assert!(b["previousMeeting"]["title"].is_null()); + assert_eq!( + conn.query_row( + "SELECT COUNT(*) FROM connector_publication_snapshot WHERE staged=1 OR published=1", + [], + |r| r.get::<_, i64>(0) + ) + .unwrap(), + 0 + ); + } + #[test] + fn publication_purge_retry_retains_original_cleanup_inventory() { + let conn = ready(); + let _a = reserve(&conn, "a"); + let old = format!("{SQL_PATH}/fireflies/archive-copy/old"); + let new = format!("{SQL_PATH}/fireflies/archive-copy/new"); + let first=call(&conn,json!({"operation":"purge","source":"fireflies","operationId":"purge","cleanupKeys":[old]})).unwrap(); + let replay=call(&conn,json!({"operation":"purge","source":"fireflies","operationId":"purge","cleanupKeys":[new]})).unwrap(); + assert_eq!(first, replay); + assert!(!replay["snapshotKeys"] + .as_array() + .unwrap() + .contains(&json!(new))); + } + #[test] + fn publication_reactivation_keeps_legacy_collision_tombstones_deleted() { + let conn = ready(); + conn.execute("DROP INDEX connector_publication_identity", []) + .unwrap(); + conn.execute("INSERT INTO connector_meeting(id,source,source_id,created_at,updated_at,publication_state) VALUES('a','fireflies','collision','now','now','deleted'),('b','fireflies','collision','now','now','deleted')",[]).unwrap(); + call(&conn, json!({"operation":"activate"})).unwrap(); + assert_eq!( + conn.query_row( + "SELECT COUNT(*) FROM connector_meeting WHERE publication_state='deleted'", + [], + |r| r.get::<_, i64>(0) + ) + .unwrap(), + 2 + ); + } +} diff --git a/tinycloud-core/src/sql/service.rs b/tinycloud-core/src/sql/service.rs index 481fa0ab..8a46fb7e 100644 --- a/tinycloud-core/src/sql/service.rs +++ b/tinycloud-core/src/sql/service.rs @@ -37,6 +37,7 @@ type HydrationLockRegistry = pub struct SqlService { databases: Arc>, hydration_locks: HydrationLockRegistry, + publication_locks: HydrationLockRegistry, /// What each live actor's local database derives from, carried into every /// durable save so a stale actor is rejected instead of clobbering. Written /// on hydration (the only path that creates an actor) and after each @@ -56,6 +57,7 @@ impl SqlService { Self { databases: Arc::new(DashMap::new()), hydration_locks: Arc::new(tokio::sync::Mutex::new(HashMap::new())), + publication_locks: Arc::new(tokio::sync::Mutex::new(HashMap::new())), lineage: Arc::new(DashMap::new()), base_path, memory_threshold, @@ -71,6 +73,11 @@ impl SqlService { caveats: Option, ability: String, ) -> Result { + let _publication_guard = if db_name == super::publication::DATABASE { + Some(self.publication_lock(space).await.lock_owned().await) + } else { + None + }; let key = (space.to_string(), db_name.to_string()); let mut handle = self.handle(space, db_name).await?; @@ -98,6 +105,37 @@ impl SqlService { Ok(result) } + async fn publication_lock(&self, space: &SpaceId) -> Arc { + let key = (space.to_string(), super::publication::DATABASE.into()); + let mut registry = self.publication_locks.lock().await; + registry.retain(|_, lock| lock.strong_count() > 0); + if let Some(lock) = registry.get(&key).and_then(Weak::upgrade) { + return lock; + } + let lock = Arc::new(HydrationLock::new(())); + registry.insert(key, Arc::downgrade(&lock)); + lock + } + /// Called only after the native route has checked publication authority and verified KV bytes. + pub async fn meeting_publication( + &self, + space: &SpaceId, + command: serde_json::Value, + ) -> Result { + let _guard = self.publication_lock(space).await.lock_owned().await; + let db_name = super::publication::DATABASE; + let key = (space.to_string(), db_name.to_string()); + let handle = self.handle(space, db_name).await?; + let result = handle.publication(command).await?; + if !result.write_targets.is_empty() { + if let Err(error) = self.persist_write(space, db_name, &handle).await { + let _ = self.discard_local_state(&key).await; + return Err(error); + } + } + Ok(result) + } + pub async fn export(&self, space: &SpaceId, db_name: &str) -> Result, SqlError> { let key = (space.to_string(), db_name.to_string()); @@ -1049,4 +1087,49 @@ mod tests { Err(SqlError::DatabaseNotFound) )); } + #[tokio::test] + async fn publication_durable_cas_fences_another_node_and_recovers() { + let repo = artifact_repository().await; + let space = test_space_id("publication-cas"); + let one = TempDir::new().unwrap(); + let two = TempDir::new().unwrap(); + let first = SqlService::new(one.path().to_string_lossy().into(), u64::MAX, repo.clone()); + let second = SqlService::new(two.path().to_string_lossy().into(), u64::MAX, repo); + first + .meeting_publication( + &space, + serde_json::json!({"contractVersion":3,"operation":"activate"}), + ) + .await + .unwrap(); + second + .meeting_publication( + &space, + serde_json::json!({"contractVersion":3,"operation":"capabilities"}), + ) + .await + .unwrap(); + let command = |op| serde_json::json!({"contractVersion":3,"operation":"reserve","source":"fireflies","sourceId":"source","operationId":op}); + first + .meeting_publication(&space, command("first")) + .await + .unwrap(); + assert!(second + .meeting_publication(&space, command("stale")) + .await + .is_err()); + let recovered = second + .meeting_publication(&space, command("recovered")) + .await + .unwrap(); + let SqlResponse::Query(receipt) = recovered.response else { + panic!("query receipt required") + }; + let SqlValue::Text(raw) = &receipt.rows[0][0] else { + panic!("raw JSON receipt required") + }; + let value: serde_json::Value = serde_json::from_str(raw).unwrap(); + assert_eq!(value["generation"], 2); + assert_eq!(value["inserted"], false); + } } diff --git a/tinycloud-core/src/sql/types.rs b/tinycloud-core/src/sql/types.rs index ed42eb7d..3024885a 100644 --- a/tinycloud-core/src/sql/types.rs +++ b/tinycloud-core/src/sql/types.rs @@ -25,7 +25,7 @@ pub enum SqlRequest { }, #[serde(rename = "batch")] Batch { statements: Vec }, - #[serde(rename = "executeStatement")] + #[serde(rename = "executeStatement", alias = "execute_statement")] ExecuteStatement { name: String, #[serde(default)] @@ -167,6 +167,12 @@ impl From<&SqlValue> for rusqlite::types::Value { mod request_tests { use super::*; + #[test] + fn publication_sdk_execute_statement_spelling_is_supported() { + let request=serde_json::from_str::(r#"{"action":"execute_statement","name":"tinycloud.meetingPublication.v3","params":["{}"]}"#).unwrap(); + assert!(matches!(request, SqlRequest::ExecuteStatement { .. })); + } + #[test] fn query_deserializes_optional_camel_case_bounds() { let request: SqlRequest = serde_json::from_str( diff --git a/tinycloud-node-server/src/invocation_replay.rs b/tinycloud-node-server/src/invocation_replay.rs index 51168e6f..44a5599b 100644 --- a/tinycloud-node-server/src/invocation_replay.rs +++ b/tinycloud-node-server/src/invocation_replay.rs @@ -1,4 +1,4 @@ -use std::time::Instant; +use std::{sync::Arc, time::Instant}; use rocket::http::Status; use time::{Duration, OffsetDateTime}; @@ -40,11 +40,22 @@ pub enum InvocationReplayError { #[derive(Clone)] pub struct InvocationReplayCache { conn: DatabaseConnection, + sqlite_writer_lock: Option>>, } impl InvocationReplayCache { pub fn new(conn: DatabaseConnection) -> Self { - Self { conn } + Self { + conn, + sqlite_writer_lock: None, + } + } + + /// Replay writes share the node's SQLite gate with delegation transactions + /// so they cannot invalidate a transaction's read-before-write snapshot. + pub fn with_sqlite_writer_lock(mut self, lock: Option>>) -> Self { + self.sqlite_writer_lock = lock; + self } /// Record an admitted invocation in the durable replay table. @@ -105,6 +116,10 @@ impl InvocationReplayCache { key: Hash, expires_at: OffsetDateTime, ) -> Result<(), InvocationReplayError> { + let _writer = match &self.sqlite_writer_lock { + Some(lock) => Some(lock.lock().await), + None => None, + }; let inserted = invocation_replay::Entity::insert(invocation_replay::ActiveModel { content_hash: Set(key), expires_at: Set(expires_at), @@ -141,6 +156,10 @@ impl InvocationReplayCache { now: OffsetDateTime, max_lifetime_secs: u64, ) -> Result { + let _writer = match &self.sqlite_writer_lock { + Some(lock) => Some(lock.lock().await), + None => None, + }; invocation_replay::Entity::delete_many() .filter( Condition::any() @@ -274,6 +293,105 @@ mod tests { db } + #[tokio::test] + async fn replay_insert_cannot_invalidate_a_delegation_write_snapshot() { + use tinycloud_core::models::{actor, delegation}; + use tinycloud_core::sea_orm::ActiveModelTrait; + use tinycloud_core::storage::{either::Either, StorageConfig}; + + let directory = tempfile::tempdir().unwrap(); + let url = format!( + "sqlite:{}?mode=rwc", + directory.path().join("caps.db").display() + ); + let connection = Database::connect(crate::sqlite_connect_options(&url)) + .await + .unwrap(); + let storage = + crate::storage::file_system::FileSystemConfig::new(directory.path().join("blocks")) + .open() + .await + .unwrap(); + let node = crate::TinyCloud::new( + connection, + Either::B(storage), + tinycloud_core::keys::StaticSecret::new(vec![7u8; 64]).unwrap(), + ) + .await + .unwrap(); + // Reuse the production pool configuration and cache constructor without + // starting the process-global log collector in a parallel unit suite. + let replay = crate::node_replay_cache(&node); + let actor_id = "did:key:sqlite-race-fixture"; + actor::ActiveModel { + id: Set(actor_id.to_owned()), + } + .insert(node.connection()) + .await + .unwrap(); + replay + .check_and_insert_key( + hash(b"expired-replay"), + OffsetDateTime::now_utc() - Duration::seconds(1), + ) + .await + .unwrap(); + + // Delegation transactions hold this gate while reading authority and + // then writing graph rows. A replay write between those steps advances + // SQLite's WAL and makes the older snapshot fail with code 517. + let gate = node.sqlite_writer_lock().unwrap(); + let writer = gate.lock().await; + let transaction = node.readable().await.unwrap(); + delegation::Entity::find() + .count(&transaction) + .await + .unwrap(); + let cleanup_replay = replay.clone(); + let mut insertion = tokio::spawn(async move { + replay + .check_and_insert_key( + hash(b"concurrent-replay"), + OffsetDateTime::now_utc() + Duration::seconds(60), + ) + .await + }); + let mut cleanup = + tokio::spawn( + async move { cleanup_replay.cleanup(OffsetDateTime::now_utc(), 300).await }, + ); + let early = + tokio::time::timeout(std::time::Duration::from_millis(100), &mut insertion).await; + let early_cleanup = + tokio::time::timeout(std::time::Duration::from_millis(100), &mut cleanup).await; + + delegation::ActiveModel { + id: Set(hash(b"delegation-write-snapshot")), + delegator: Set(actor_id.to_owned()), + delegatee: Set(actor_id.to_owned()), + expiry: Set(None), + issued_at: Set(None), + not_before: Set(None), + facts: Set(None), + serialization: Set(b"synthetic database race fixture".to_vec()), + } + .insert(&transaction) + .await + .expect("concurrent replay must not invalidate the delegation snapshot"); + transaction.commit().await.unwrap(); + assert!( + early.is_err(), + "replay insertion must wait for the node's writer gate" + ); + assert!( + early_cleanup.is_err(), + "replay cleanup must wait for the node's writer gate" + ); + drop(writer); + insertion.await.unwrap().unwrap(); + assert_eq!(cleanup.await.unwrap().unwrap(), 1); + } + #[tokio::test] async fn duplicate_survives_new_cache_instance() { let directory = tempfile::tempdir().unwrap(); diff --git a/tinycloud-node-server/src/lib.rs b/tinycloud-node-server/src/lib.rs index b873c755..a088fde7 100644 --- a/tinycloud-node-server/src/lib.rs +++ b/tinycloud-node-server/src/lib.rs @@ -215,6 +215,16 @@ fn sqlite_connect_options(database: &str) -> ConnectOptions { connect_opts } +fn node_replay_cache(tinycloud: &TinyCloud) -> InvocationReplayCache { + InvocationReplayCache::new(tinycloud.connection().clone()) + .with_sqlite_writer_lock(tinycloud.sqlite_writer_lock()) +} + +fn node_artifact_repository(tinycloud: &TinyCloud) -> SeaOrmDatabaseArtifactRepository { + SeaOrmDatabaseArtifactRepository::new(tinycloud.connection().clone()) + .with_sqlite_writer_lock(tinycloud.sqlite_writer_lock()) +} + pub async fn app(config: &Figment) -> Result> { let tinycloud_config = config.extract::()?; app_with_control(config, &tinycloud_config, None).await @@ -393,16 +403,10 @@ pub async fn app_with_control( let database_connection = Database::connect(connect_opts).await?; // SQL/DuckDB artifact-size mirror folded into `store_size`. Empty here; - // wired into the decorator + SpaceDatabase BEFORE migrations, then seeded + // wired into SpaceDatabase, then seeded // from DB truth AFTER `TinyCloud::new` runs migrations (see below). let sql_sizes = SqlSizes::new(); let seed_conn = database_connection.clone(); - let raw_artifact_repository = Arc::new(SeaOrmDatabaseArtifactRepository::new( - database_connection.clone(), - )); - let database_artifact_repository: Arc = Arc::new( - SizeTrackingArtifactRepository::new(raw_artifact_repository, sql_sizes.clone()), - ); // Encryption module: seal network private keys with the same kind of derived // key used for DB column encryption. In DStack mode the seal is rooted in @@ -428,6 +432,11 @@ pub async fn app_with_control( .with_sql_sizes(sql_sizes.clone()); let encryption_service = encryption_service.with_sqlite_writer_lock(tinycloud.sqlite_writer_lock()); + let database_artifact_repository: Arc = + Arc::new(SizeTrackingArtifactRepository::new( + Arc::new(node_artifact_repository(&tinycloud)), + sql_sizes.clone(), + )); // Seed the SQL-size mirror AFTER `TinyCloud::new` ran migrations — the // `database_artifact` table now exists (seeding before migrations would @@ -492,7 +501,7 @@ pub async fn app_with_control( tinycloud_config.storage.limit, std::env::var("TINYCLOUD_QUOTA_URL").ok(), ); - let invocation_replay_cache = InvocationReplayCache::new(seed_conn.clone()); + let invocation_replay_cache = node_replay_cache(&tinycloud); let replay_cleanup = invocation_replay_cache.clone(); // TC-341: the periodic sweep also reclaims rows beyond the lifetime cap. let replay_max_lifetime_secs = tinycloud_config.invocation.max_lifetime_secs; @@ -1000,6 +1009,120 @@ mod sqlite_tuning_tests { assert_eq!(pragma(&db, "temp_store").await, "2"); // 2 = MEMORY assert_eq!(pragma(&db, "mmap_size").await, "268435456"); } + + async fn artifact_save_preserves_delegation_snapshot(delta: bool) { + use tinycloud_core::{ + database_artifacts::ArtifactExpectation, + hash::hash, + models::{actor, delegation}, + sea_orm::{ActiveModelTrait, ActiveValue::Set, EntityTrait, PaginatorTrait}, + storage::{either::Either, StorageConfig}, + }; + + let directory = tempfile::tempdir().unwrap(); + let url = format!( + "sqlite:{}?mode=rwc", + directory.path().join("caps.db").display() + ); + let connection = Database::connect(sqlite_connect_options(&url)) + .await + .unwrap(); + let blocks = storage::file_system::FileSystemConfig::new(directory.path().join("blocks")) + .open() + .await + .unwrap(); + let node = TinyCloud::new( + connection, + Either::B(blocks), + tinycloud_core::keys::StaticSecret::new(vec![7u8; 64]).unwrap(), + ) + .await + .unwrap(); + // Exercise the production repository wiring as well as its write path. + let repository = node_artifact_repository(&node); + let actor_id = "did:key:artifact-snapshot-fixture"; + actor::ActiveModel { + id: Set(actor_id.to_owned()), + } + .insert(node.connection()) + .await + .unwrap(); + repository + .save( + "sql", + "space", + "main", + vec![1; 100], + ArtifactExpectation::Absent, + ) + .await + .unwrap(); + + let gate = node.sqlite_writer_lock().unwrap(); + let writer = gate.lock().await; + let transaction = node.readable().await.unwrap(); + delegation::Entity::find() + .count(&transaction) + .await + .unwrap(); + let mut save = tokio::spawn(async move { + if delta { + repository + .save_delta( + "sql", + "space", + "main", + vec![2; 12], + ArtifactExpectation::Any, + ) + .await + .map(|saved| saved.revision) + } else { + repository + .save( + "sql", + "space", + "main", + vec![3; 120], + ArtifactExpectation::Any, + ) + .await + .map(|saved| saved.revision) + } + }); + let early = tokio::time::timeout(std::time::Duration::from_millis(100), &mut save).await; + + delegation::ActiveModel { + id: Set(hash(b"artifact-delegation-write-snapshot")), + delegator: Set(actor_id.to_owned()), + delegatee: Set(actor_id.to_owned()), + expiry: Set(None), + issued_at: Set(None), + not_before: Set(None), + facts: Set(None), + serialization: Set(b"synthetic database race fixture".to_vec()), + } + .insert(&transaction) + .await + .expect("artifact persistence must not invalidate the delegation snapshot"); + transaction.commit().await.unwrap(); + assert!( + early.is_err(), + "artifact persistence must wait for the node's writer gate" + ); + drop(writer); + assert_eq!(save.await.unwrap().unwrap(), 2); + } + + #[tokio::test] + async fn artifact_checkpoint_cannot_invalidate_a_delegation_write_snapshot() { + artifact_save_preserves_delegation_snapshot(false).await; + } + + #[tokio::test] + async fn artifact_delta_cannot_invalidate_a_delegation_write_snapshot() { + artifact_save_preserves_delegation_snapshot(true).await; + } } #[cfg(test)] diff --git a/tinycloud-node-server/src/policy_v3.rs b/tinycloud-node-server/src/policy_v3.rs index dcaf5666..f799d3f7 100644 --- a/tinycloud-node-server/src/policy_v3.rs +++ b/tinycloud-node-server/src/policy_v3.rs @@ -2389,10 +2389,8 @@ fn validate_stored_root_status( if previous_digest.is_some() { return Err("root-status-chain-invalid"); } - } else { - if previous_digest != root.previous_checkpoint_digest_hex.as_deref() { - return Err("root-status-chain-invalid"); - } + } else if previous_digest != root.previous_checkpoint_digest_hex.as_deref() { + return Err("root-status-chain-invalid"); } let signature = object .get("signature") diff --git a/tinycloud-node-server/src/routes/meeting_publication.rs b/tinycloud-node-server/src/routes/meeting_publication.rs new file mode 100644 index 00000000..3d48940e --- /dev/null +++ b/tinycloud-node-server/src/routes/meeting_publication.rs @@ -0,0 +1,728 @@ +//! Native publication commands share /invoke admission and the existing SQL catalog. +use super::*; +use tinycloud_core::sql::{publication, SqlExecutionResult, SqlResponse, SqlValue}; + +pub(super) fn command( + request: &SqlRequest, + path: Option<&str>, + ability: &str, + caveats: &Option, +) -> Result, (Status, String)> { + let params = match request { + SqlRequest::ExecuteStatement { name, params } if name == publication::STATEMENT => params, + SqlRequest::Execute { + sql, + params, + schema, + } if sql == publication::STATEMENT => { + if schema.is_some() { + return Err((Status::BadRequest, "publication_schema_forbidden".into())); + } + params + } + _ => return Ok(None), + }; + if path != Some(publication::SQL_PATH) + || !tinycloud_core::policy_capability::ability_matches(ability, "tinycloud.sql/write") + || caveats.is_some() + { + return Err((Status::Forbidden, "publication_authority_required".into())); + } + let [SqlValue::Text(raw)] = params.as_slice() else { + return Err((Status::BadRequest, "publication_invalid_command".into())); + }; + let command: serde_json::Value = serde_json::from_str(raw) + .map_err(|_| (Status::BadRequest, "publication_invalid_command".into()))?; + if command["contractVersion"] != 3 + || !matches!( + command["operation"].as_str(), + Some( + "capabilities" + | "activate" + | "reserve" + | "stage" + | "publish" + | "inspect" + | "delete" + | "purge" + | "freeze_legacy" + | "unfreeze_legacy" + | "legacy_freeze_status" + ) + ) + { + return Err((Status::BadRequest, "publication_invalid_command".into())); + } + if matches!( + command["operation"].as_str(), + Some("freeze_legacy" | "unfreeze_legacy") + ) && expected_generation(&command).is_none() + { + return Err(( + Status::BadRequest, + "legacy_freeze_invalid_expected_generation".into(), + )); + } + Ok(Some(command)) +} +fn expected_generation(command: &serde_json::Value) -> Option { + command["expectedGeneration"] + .as_i64() + .filter(|generation| *generation >= 0 && *generation < i64::MAX) +} +/// Publication does not reinterpret a delegated table/column/statement caveat as unrestricted SQL. +pub(super) async fn require_unconstrained_chain( + tinycloud: &TinyCloud, + parents: &[tinycloud_auth::authorization::Cid], +) -> Result<(), (Status, String)> { + use tinycloud_core::{hash::Hash, models::abilities, relationships::parent_delegations}; + let conn = tinycloud + .readable() + .await + .map_err(|e| (Status::InternalServerError, e.to_string()))?; + let mut frontier: Vec = parents.iter().copied().map(Hash::from).collect(); + let mut visited = HashSet::new(); + while !frontier.is_empty() { + let batch: Vec<_> = frontier.drain(..).filter(|h| visited.insert(*h)).collect(); + if batch.is_empty() { + break; + } + let rows = abilities::Entity::find() + .filter(abilities::Column::Delegation.is_in(batch.clone())) + .all(&conn) + .await + .map_err(|e| (Status::InternalServerError, e.to_string()))?; + for row in rows { + if row + .resource + .tinycloud_resource() + .is_some_and(|resource| resource.service().as_str() == "sql") + && !row.caveats.0.is_empty() + { + return Err(( + Status::Forbidden, + "publication_unconstrained_authority_required".into(), + )); + } + } + let links = parent_delegations::Entity::find() + .filter(parent_delegations::Column::Child.is_in(batch)) + .all(&conn) + .await + .map_err(|e| (Status::InternalServerError, e.to_string()))?; + for link in links { + if !visited.contains(&link.parent) { + frontier.push(link.parent); + } + } + } + Ok(()) +} + +fn bad(code: &str) -> SqlError { + SqlError::InvalidStatement(code.into()) +} +fn internal(error: impl std::fmt::Display) -> SqlError { + SqlError::Internal(error.to_string()) +} +fn receipt(value: serde_json::Value) -> SqlExecutionResult { + SqlExecutionResult { + response: SqlResponse::Query(tinycloud_core::sql::QueryResponse { + columns: vec!["receipt".into()], + rows: vec![vec![SqlValue::Text(value.to_string())]], + row_count: 1, + }), + write_targets: vec![], + } +} +async fn verify( + tinycloud: &TinyCloud, + space: &SpaceId, + key: &str, + revision: &str, +) -> Result { + let path: Path = key.parse().map_err(internal)?; + let Some((_, _, content)) = tinycloud.kv_get(space, &path).await.map_err(internal)? else { + return Ok(false); + }; + let mut reader = Box::pin(content).take((publication::ENVELOPE_LIMIT + 1) as u64); + let mut bytes = Vec::new(); + FuturesAsyncReadExt::read_to_end(&mut reader, &mut bytes) + .await + .map_err(internal)?; + if bytes.len() > publication::ENVELOPE_LIMIT { + return Err(bad("publication_capacity")); + } + let raw = std::str::from_utf8(&bytes).map_err(|_| bad("publication_snapshot_invalid"))?; + if publication::digest(raw) != revision { + return Err(bad("publication_digest_mismatch")); + } + Ok(true) +} +// Serialize the SQL/KV publication sequence as one native operation on this node. +lazy_static::lazy_static! {static ref PUBLICATION_LOCKS:std::sync::Mutex>>>=std::sync::Mutex::new(HashMap::new());} +fn protocol_lock(space: &SpaceId) -> std::sync::Arc> { + let mut locks = PUBLICATION_LOCKS.lock().unwrap(); + locks.retain(|_, lock| lock.strong_count() > 0); + let key = space.to_string(); + if let Some(lock) = locks.get(&key).and_then(std::sync::Weak::upgrade) { + return lock; + } + let lock = std::sync::Arc::new(tokio::sync::Mutex::new(())); + locks.insert(key, std::sync::Arc::downgrade(&lock)); + lock +} +pub(super) async fn execute( + tinycloud: &TinyCloud, + sql: &SqlService, + staging: &BlockStage, + space: &SpaceId, + mut command: serde_json::Value, +) -> Result { + let _guard = protocol_lock(space).lock_owned().await; + let operation = command["operation"] + .as_str() + .ok_or_else(|| bad("publication_invalid_command"))? + .to_owned(); + if matches!(operation.as_str(), "freeze_legacy" | "unfreeze_legacy") { + let expected = expected_generation(&command) + .ok_or_else(|| bad("legacy_freeze_invalid_expected_generation"))?; + let status = tinycloud + .set_legacy_meeting_write_freeze(space, operation == "freeze_legacy", expected) + .await + .map_err(|error| match error { + tinycloud_core::meeting_legacy_guard::FreezeError::Db(error) => internal(error), + error => bad(&error.to_string()), + })?; + return Ok(receipt( + serde_json::json!({"contractVersion":3,"legacyWritesFrozen":status.frozen,"legacyFreezeGeneration":status.generation}), + )); + } + if operation == "legacy_freeze_status" { + let status = tinycloud + .legacy_meeting_freeze_status(space) + .await + .map_err(internal)?; + return Ok(receipt( + serde_json::json!({"contractVersion":3,"legacyWritesFrozen":status.frozen,"legacyFreezeGeneration":status.generation}), + )); + } + // Reject known frozen cleanup before SQL tombstones/removals. The core KV + // transaction guard independently closes races and internal cleanup bypasses. + if matches!(operation.as_str(), "delete" | "purge") + && tinycloud + .legacy_meeting_freeze_status(space) + .await + .map_err(internal)? + .frozen + { + return Err(SqlError::PermissionDenied( + "legacy meeting artifacts are frozen".into(), + )); + } + if operation == "stage" { + publication::validate_snapshot(&command)?; + let raw = command["snapshotRaw"] + .as_str() + .ok_or_else(|| bad("publication_invalid_command"))? + .to_owned(); + let key = command["snapshotKey"] + .as_str() + .ok_or_else(|| bad("publication_invalid_command"))? + .to_owned(); + let revision = command["revision"] + .as_str() + .ok_or_else(|| bad("publication_invalid_command"))? + .to_owned(); + command["operation"] = serde_json::json!("prepare_stage"); + sql.meeting_publication(space, command.clone()).await?; + if !verify(tinycloud, space, &key, &revision).await? { + let mut buffer = staging.stage(space).await.map_err(internal)?; + buffer.write_all(raw.as_bytes()).await.map_err(internal)?; + buffer.flush().await.map_err(internal)?; + match tinycloud + .invoke_internal_meeting_snapshot_put::( + space.clone(), + key.parse().map_err(internal)?, + Metadata(BTreeMap::from([( + "content-type".into(), + "application/json".into(), + )])), + buffer, + Some(KvPrecondition::DoesNotExist), + ) + .await + { + Ok(_) | Err(TxStoreError::KvPreconditionFailed) => {} + Err(error) => return Err(internal(error)), + } + } + if !verify(tinycloud, space, &key, &revision).await? { + return Err(bad("publication_snapshot_missing")); + } + command["operation"] = serde_json::json!("stage"); + command.as_object_mut().unwrap().remove("snapshotRaw"); + // A superseding reservation/delete cannot publish this verified staged object. + // Failed finalization remains indexed for purge. Do not delete here: another + // node may already have published an idempotent attempt with the same digest. + return sql.meeting_publication(space, command).await; + } + if operation == "publish" { + let key = command["snapshotKey"] + .as_str() + .ok_or_else(|| bad("publication_invalid_command"))?; + let revision = command["revision"] + .as_str() + .ok_or_else(|| bad("publication_invalid_command"))?; + if !publication::protected_snapshot_path(key) + || !verify(tinycloud, space, key, revision).await? + { + return Err(bad("publication_snapshot_missing")); + } + } + command.as_object_mut().unwrap().remove("cleanupKeys"); + if operation == "purge" { + let source = command["source"] + .as_str() + .filter(|source| { + matches!( + *source, + "fireflies" | "google-meet" | "tinycloud-transcriber" + ) + }) + .ok_or_else(|| bad("publication_invalid_source"))?; + let prefix: Path = format!("{}/{source}/", publication::SQL_PATH) + .parse() + .map_err(internal)?; + let keys = tinycloud + .public_kv_list(space, &prefix) + .await + .map_err(internal)?; + command["cleanupKeys"] = serde_json::json!(keys + .into_iter() + .map(|key| key.to_string()) + .collect::>()); + } + let mut result = sql.meeting_publication(space, command).await?; + if operation == "capabilities" { + let SqlResponse::Query(query) = &mut result.response else { + return Err(bad("publication_receipt_invalid")); + }; + let Some(SqlValue::Text(raw)) = query.rows.first_mut().and_then(|row| row.first_mut()) + else { + return Err(bad("publication_receipt_invalid")); + }; + let mut value: serde_json::Value = serde_json::from_str(raw).map_err(internal)?; + value["legacyWriteFreeze"] = serde_json::json!(true); + *raw = value.to_string(); + } + if matches!(operation.as_str(), "delete" | "purge") { + let SqlResponse::Query(query) = &result.response else { + return Err(bad("publication_receipt_invalid")); + }; + let Some(SqlValue::Text(raw)) = query.rows.first().and_then(|r| r.first()) else { + return Err(bad("publication_receipt_invalid")); + }; + let receipt: serde_json::Value = serde_json::from_str(raw).map_err(internal)?; + for key in receipt["snapshotKeys"] + .as_array() + .ok_or_else(|| bad("publication_receipt_invalid"))? + { + let key = key + .as_str() + .ok_or_else(|| bad("publication_receipt_invalid"))?; + if !publication::connector_owned_path(key) { + return Err(bad("publication_receipt_invalid")); + } + tinycloud + .invoke_internal_meeting_snapshot_delete::( + space.clone(), + key.parse().map_err(internal)?, + ) + .await + .map_err(internal)?; + } + } + Ok(result) +} + +#[cfg(test)] +mod tests { + use super::*; + use tinycloud_core::sea_orm::{ActiveModelTrait, ActiveValue::Set, ConnectOptions, Database}; + use tinycloud_core::{ + database_artifacts::SeaOrmDatabaseArtifactRepository, + keys::StaticSecret, + storage::{either::Either, StorageConfig}, + types::SpaceIdWrap, + }; + + async fn legacy_route_fixture( + activate: bool, + ) -> ( + TinyCloud, + SqlService, + BlockStage, + SpaceId, + tempfile::TempDir, + ) { + let directory = tempfile::tempdir().unwrap(); + let conn = Database::connect(ConnectOptions::new("sqlite::memory:".to_string())) + .await + .unwrap(); + let storage = + crate::storage::file_system::FileSystemConfig::new(directory.path().join("blocks")) + .open() + .await + .unwrap(); + let tinycloud = TinyCloud::new( + conn.clone(), + Either::B(storage), + StaticSecret::new(vec![0; 32]).unwrap(), + ) + .await + .unwrap(); + let key = tinycloud_auth::ssi::jwk::JWK::generate_ed25519().unwrap(); + let space = SpaceId::new( + tinycloud_auth::resolver::DID_METHODS + .generate(&key, "key") + .unwrap(), + "freeze-catalog".parse().unwrap(), + ); + tinycloud_core::models::space::ActiveModel { + id: Set(SpaceIdWrap(space.clone())), + } + .insert(&conn) + .await + .unwrap(); + let sql = SqlService::new( + directory.path().join("sql").display().to_string(), + u64::MAX, + std::sync::Arc::new(SeaOrmDatabaseArtifactRepository::new(conn)), + ); + if activate { + sql.meeting_publication( + &space, + serde_json::json!({"contractVersion":3,"operation":"activate"}), + ) + .await + .unwrap(); + sql.meeting_publication(&space, serde_json::json!({"contractVersion":3,"operation":"reserve","source":"fireflies","sourceId":"old","operationId":"initial"})).await.unwrap(); + } + let staging = BlockStage::from(crate::config::StagingStorage::Memory); + (tinycloud, sql, staging, space, directory) + } + + async fn catalog(sql: &SqlService, space: &SpaceId) -> serde_json::Value { + let result = sql + .execute( + space, + publication::DATABASE, + SqlRequest::Query { + sql: "SELECT * FROM connector_meeting ORDER BY id".into(), + params: vec![], + max_rows: None, + max_bytes: None, + }, + None, + "tinycloud.sql/read".into(), + ) + .await + .unwrap(); + serde_json::to_value(result.response).unwrap() + } + + #[tokio::test] + async fn legacy_freeze_delete_and_purge_leave_catalog_unchanged() { + for operation in ["delete", "purge"] { + let (tinycloud, sql, staging, space, _directory) = legacy_route_fixture(true).await; + let before = catalog(&sql, &space).await; + let frozen = execute(&tinycloud,&sql,&staging,&space,serde_json::json!({"contractVersion":3,"operation":"freeze_legacy","expectedGeneration":0})).await.unwrap(); + let SqlResponse::Query(query) = frozen.response else { + panic!("freeze receipt") + }; + assert_eq!(query.columns, vec!["receipt"]); + let SqlValue::Text(raw) = &query.rows[0][0] else { + panic!("freeze receipt") + }; + assert_eq!( + serde_json::from_str::(raw).unwrap(), + serde_json::json!({"contractVersion":3,"legacyWritesFrozen":true,"legacyFreezeGeneration":1}) + ); + let result = execute(&tinycloud,&sql,&staging,&space,serde_json::json!({"contractVersion":3,"operation":operation,"source":"fireflies","sourceId":"old","operationId":operation})).await; + assert!(result.is_err(), "frozen {operation} must reject"); + assert_eq!( + catalog(&sql, &space).await, + before, + "frozen {operation} changed catalog before cleanup rejection" + ); + assert!(result + .unwrap_err() + .to_string() + .contains("legacy meeting artifacts are frozen")); + execute(&tinycloud,&sql,&staging,&space,serde_json::json!({"contractVersion":3,"operation":"unfreeze_legacy","expectedGeneration":1})).await.unwrap(); + execute(&tinycloud,&sql,&staging,&space,serde_json::json!({"contractVersion":3,"operation":operation,"source":"fireflies","sourceId":"old","operationId":operation})).await.unwrap(); + assert_ne!( + catalog(&sql, &space).await, + before, + "released {operation} did not resume" + ); + } + } + fn parsed_receipt(result: SqlExecutionResult) -> serde_json::Value { + let SqlResponse::Query(query) = result.response else { + panic!("query receipt") + }; + assert_eq!(query.columns, vec!["receipt"]); + let SqlValue::Text(raw) = &query.rows[0][0] else { + panic!("JSON receipt") + }; + serde_json::from_str(raw).unwrap() + } + #[tokio::test] + async fn legacy_freeze_controls_are_generation_checked_without_catalog_activation() { + let (tinycloud, sql, staging, space, _directory) = legacy_route_fixture(false).await; + let status = parsed_receipt( + execute( + &tinycloud, + &sql, + &staging, + &space, + serde_json::json!({"contractVersion":3,"operation":"legacy_freeze_status"}), + ) + .await + .unwrap(), + ); + assert_eq!( + status, + serde_json::json!({"contractVersion":3,"legacyWritesFrozen":false,"legacyFreezeGeneration":0}) + ); + let capabilities = parsed_receipt( + execute( + &tinycloud, + &sql, + &staging, + &space, + serde_json::json!({"contractVersion":3,"operation":"capabilities"}), + ) + .await + .unwrap(), + ); + assert_eq!(capabilities["legacyWriteFreeze"], true); + for (operation, expected, frozen, generation) in [ + ("freeze_legacy", 0, true, 1), + ("freeze_legacy", 0, true, 1), + ("unfreeze_legacy", 1, false, 2), + ("unfreeze_legacy", 1, false, 2), + ("freeze_legacy", 2, true, 3), + ] { + let receipt=parsed_receipt(execute(&tinycloud,&sql,&staging,&space,serde_json::json!({"contractVersion":3,"operation":operation,"expectedGeneration":expected})).await.unwrap()); + assert_eq!( + receipt, + serde_json::json!({"contractVersion":3,"legacyWritesFrozen":frozen,"legacyFreezeGeneration":generation}) + ); + } + let stale=execute(&tinycloud,&sql,&staging,&space,serde_json::json!({"contractVersion":3,"operation":"unfreeze_legacy","expectedGeneration":1})).await.unwrap_err(); + assert!(stale + .to_string() + .contains("legacy_freeze_generation_conflict")); + let reserve=sql.meeting_publication(&space,serde_json::json!({"contractVersion":3,"operation":"reserve","source":"fireflies","sourceId":"old","operationId":"probe"})).await.unwrap_err(); + assert!( + reserve + .to_string() + .contains("publication_activation_required"), + "freeze unexpectedly activated catalog: {reserve}" + ); + } + fn request(operation: &str) -> SqlRequest { + SqlRequest::ExecuteStatement { + name: publication::STATEMENT.into(), + params: vec![SqlValue::Text( + serde_json::json!({"contractVersion":3,"operation":operation,"expectedGeneration":0}).to_string(), + )], + } + } + #[test] + fn legacy_freeze_rejects_invalid_expected_generation() { + for expected in [ + serde_json::Value::Null, + serde_json::json!(-1), + serde_json::json!(1.5), + serde_json::json!("0"), + serde_json::json!(i64::MAX), + serde_json::json!(u64::MAX), + ] { + let request = SqlRequest::ExecuteStatement { name:publication::STATEMENT.into(), params:vec![SqlValue::Text(serde_json::json!({"contractVersion":3,"operation":"freeze_legacy","expectedGeneration":expected}).to_string())] }; + assert!( + command( + &request, + Some(publication::SQL_PATH), + "tinycloud.sql/write", + &None + ) + .is_err(), + "accepted invalid generation {expected}" + ); + } + } + #[test] + fn publication_route_rejects_read_only_and_constrained_authority() { + assert!(command( + &request("reserve"), + Some(publication::SQL_PATH), + "tinycloud.sql/read", + &None + ) + .is_err()); + assert!(command( + &request("reserve"), + Some(publication::SQL_PATH), + "tinycloud.sql/write", + &Some(SqlCaveats::default()) + ) + .is_err()); + } + #[test] + fn publication_route_rejects_wrong_database_and_private_operations() { + assert!(command( + &request("reserve"), + Some("other/connectors"), + "tinycloud.sql/write", + &None + ) + .is_err()); + assert!(command( + &request("prepare_stage"), + Some(publication::SQL_PATH), + "tinycloud.sql/write", + &None + ) + .is_err()); + } + #[test] + fn publication_route_recognizes_fixed_command() { + assert!(command( + &request("activate"), + Some(publication::SQL_PATH), + "tinycloud.sql/*", + &None + ) + .unwrap() + .is_some()); + assert_eq!( + command( + &request("capabilities"), + Some(publication::SQL_PATH), + "tinycloud.sql/write", + &None + ) + .unwrap() + .unwrap()["operation"], + "capabilities" + ); + } + + #[test] + fn legacy_freeze_commands_require_existing_unconstrained_publication_authority() { + for operation in ["freeze_legacy", "unfreeze_legacy", "legacy_freeze_status"] { + assert!(command( + &request(operation), + Some(publication::SQL_PATH), + "tinycloud.sql/write", + &None + ) + .unwrap() + .is_some()); + assert!(command( + &request(operation), + Some(publication::SQL_PATH), + "tinycloud.sql/read", + &None + ) + .is_err()); + assert!(command( + &request(operation), + Some(publication::SQL_PATH), + "tinycloud.sql/write", + &Some(SqlCaveats::default()) + ) + .is_err()); + assert!(command( + &request(operation), + Some("other/connectors"), + "tinycloud.sql/write", + &None + ) + .is_err()); + } + } + #[test] + fn publication_route_accepts_fixed_execute_without_schema() { + let request = SqlRequest::Execute { + sql: publication::STATEMENT.into(), + params: vec![SqlValue::Text( + serde_json::json!({"contractVersion":3,"operation":"capabilities"}).to_string(), + )], + schema: None, + }; + assert!(command( + &request, + Some(publication::SQL_PATH), + "tinycloud.sql/write", + &None + ) + .unwrap() + .is_some()); + let SqlRequest::Execute { sql, params, .. } = request else { + unreachable!() + }; + assert!(command( + &SqlRequest::Execute { + sql, + params, + schema: Some(vec!["DROP TABLE connector_meeting".into()]) + }, + Some(publication::SQL_PATH), + "tinycloud.sql/write", + &None + ) + .is_err()); + } + #[rocket::post("/", data = "")] + async fn framing(data: rocket::Data<'_>) -> Result { + super::super::read_json_body_limited(DataIn::One(data), publication::ENVELOPE_LIMIT) + .await + .map(|body| body.len().to_string()) + } + #[rocket::async_test] + async fn publication_route_checks_complete_utf8_framing() { + let client = rocket::local::asynchronous::Client::tracked( + rocket::build().mount("/", rocket::routes![framing]), + ) + .await + .unwrap(); + let body = "é".repeat(publication::ENVELOPE_LIMIT / 2); + assert_eq!( + client + .post("/") + .body(body.clone()) + .dispatch() + .await + .status(), + Status::Ok + ); + assert_eq!( + client + .post("/") + .body(format!("{body} ")) + .dispatch() + .await + .status(), + Status::PayloadTooLarge + ); + assert_eq!( + client.post("/").body(vec![255]).dispatch().await.status(), + Status::BadRequest + ); + } +} diff --git a/tinycloud-node-server/src/routes/mod.rs b/tinycloud-node-server/src/routes/mod.rs index 444ad8a9..b3a38b3a 100644 --- a/tinycloud-node-server/src/routes/mod.rs +++ b/tinycloud-node-server/src/routes/mod.rs @@ -64,6 +64,7 @@ pub mod admin; pub mod attestation; pub mod encryption; pub mod hooks; +mod meeting_publication; pub mod node_keys; pub mod public; #[cfg(feature = "tc-bench-v1")] @@ -1451,6 +1452,7 @@ async fn invoke_impl( admitted, data, tinycloud, + staging, sql_service, hook_runtime, quota_cache, @@ -1839,6 +1841,8 @@ async fn invoke_impl( match &e { TxStoreError::Tx(TxError::SpaceNotFound) => Status::NotFound, TxStoreError::KvPreconditionFailed => Status::PreconditionFailed, + TxStoreError::MeetingSnapshotProtected => Status::Forbidden, + TxStoreError::LegacyMeetingFrozen => Status::Conflict, TxStoreError::KvSerializationConflict => Status::ServiceUnavailable, TxStoreError::KvResponseTooLarge { .. } => Status::PayloadTooLarge, TxStoreError::Tx(TxError::InvalidInvocation( @@ -2014,15 +2018,27 @@ async fn emit_kv_hook_events( /// Read the request body as a JSON string. async fn read_json_body(data: DataIn<'_>) -> Result { + read_json_body_limited(data, 1_048_576).await +} +async fn read_json_body_limited( + data: DataIn<'_>, + limit: usize, +) -> Result { let start = Instant::now(); match data { DataIn::One(d) => { let mut buf = Vec::new(); - let mut reader = d.open(1u8.megabytes()); + let mut reader = d.open(((limit + 1) as u64).bytes()); reader .read_to_end(&mut buf) .await .map_err(|e| (Status::BadRequest, e.to_string()))?; + if buf.len() > limit { + return Err(( + Status::PayloadTooLarge, + "JSON body exceeds complete envelope limit".into(), + )); + } let result = String::from_utf8(buf).map_err(|e| (Status::BadRequest, e.to_string())); crate::prometheus::observe_stage( crate::prometheus::InvocationStage::RequestDecode, @@ -2080,6 +2096,7 @@ async fn handle_sql_invoke( admitted: AdmittedInvocation, data: DataIn<'_>, tinycloud: &State, + staging: &State, sql_service: &State, hook_runtime: &State, quota_cache: &State, @@ -2116,7 +2133,8 @@ async fn handle_sql_invoke( // verification a second time. let auth_result = verify_auth_admitted("server.sql.auth", admitted, tinycloud).await?; let body_start = Instant::now(); - let body_result = read_json_body(data).await; + let body_result = + read_json_body_limited(data, tinycloud_core::sql::publication::ENVELOPE_LIMIT).await; crate::prometheus::observe_span( "server.sql.read_body", if body_result.is_ok() { "ok" } else { "error" }, @@ -2174,20 +2192,43 @@ async fn handle_sql_invoke( // post-execute, so a write crossing the limit is admitted and the next // write 402s. No shrink — DELETE does not reduce artifact size without // VACUUM, so an over-quota space cannot self-serve shrink. - if sql_request_is_write(&sql_request, &exec_caveats, ability) { + let publication_command = + meeting_publication::command(&sql_request, path, ability, &exec_caveats)?; + if publication_command.is_some() { + meeting_publication::require_unconstrained_chain(tinycloud, &parent_cids).await?; + } + // Barrier controls change graph state only and must remain available when + // content storage is full, especially release after a completed migration. + let grows_content = match publication_command.as_ref() { + Some(command) => !matches!( + command["operation"].as_str(), + Some( + "capabilities" + | "inspect" + | "freeze_legacy" + | "unfreeze_legacy" + | "legacy_freeze_status" + ) + ), + None => sql_request_is_write(&sql_request, &exec_caveats, ability), + }; + if grows_content { staged_batch_remaining(space, tinycloud, config, quota_cache).await?; } - let execute_start = Instant::now(); - let execute_result = sql_service - .execute( - space, - &db_name, - sql_request, - exec_caveats, - ability.to_string(), - ) - .await; + let execute_result = if let Some(command) = publication_command { + meeting_publication::execute(tinycloud, sql_service, staging, space, command).await + } else { + sql_service + .execute( + space, + &db_name, + sql_request, + exec_caveats, + ability.to_string(), + ) + .await + }; crate::prometheus::observe_span( "server.sql.execute", if execute_result.is_ok() { @@ -5999,6 +6040,71 @@ mod tests { manage_tc405_test_state(rocket, conn) } + #[tokio::test] + async fn legacy_freeze_controls_work_over_quota_while_publication_growth_rejects() -> Result<()> + { + use rocket::{ + data::ByteUnit, + http::{ContentType, Header}, + local::asynchronous::Client, + }; + use tinycloud_core::{ + models::abilities, + sea_orm::{ActiveModelTrait, ActiveValue::Set}, + types::Caveats, + }; + let mut setup = metered_sql_http_setup("legacy-freeze-quota").await?; + setup.resource = setup.space.clone().to_resource( + "sql".parse()?, + Some(tinycloud_core::sql::publication::SQL_PATH.parse()?), + None, + None, + ); + abilities::ActiveModel { + delegation: Set(setup.parent_cid.into()), + resource: Set(Resource::TinyCloud(setup.resource.clone())), + ability: Set(Ability::try_from("tinycloud.sql/write".to_string()).unwrap()), + caveats: Set(Caveats(Default::default())), + } + .insert(&setup.replay_db) + .await?; + let cases = [ + ("legacy_freeze_status", 0, 200), + ("freeze_legacy", 0, 200), + ("unfreeze_legacy", 1, 200), + ("reserve", 0, 402), + ("stage", 0, 402), + ("publish", 0, 402), + ]; + let mut requests = vec![]; + for (index, (operation, expected, status)) in cases.into_iter().enumerate() { + requests.push((operation,status,sql_invocation_header(&setup,"tinycloud.sql/write",&format!("urn:uuid:legacy-quota-{index}"))?, + serde_json::to_string(&SqlRequest::ExecuteStatement{name:tinycloud_core::sql::publication::STATEMENT.into(),params:vec![SqlValue::Text(serde_json::json!({"contractVersion":3,"operation":operation,"expectedGeneration":expected}).to_string())]})?)); + } + let client = Client::tracked(metered_sql_rocket(setup, ByteUnit::Byte(1))).await?; + for (operation, expected, auth, body) in requests { + let response = client + .post("/invoke") + .header(Header::new("Authorization", auth)) + .header(ContentType::JSON) + .body(body) + .dispatch() + .await; + let status = response.status().code; + let body = response.into_string().await.unwrap_or_default(); + assert_eq!(status, expected, "{operation} at exhausted quota: {body}"); + if operation == "unfreeze_legacy" { + let response: serde_json::Value = serde_json::from_str(&body)?; + let receipt: serde_json::Value = serde_json::from_str( + response["rows"][0][0].as_str().expect("receipt JSON cell"), + )?; + assert_eq!(receipt["legacyWritesFrozen"], false); + assert_eq!(receipt["legacyFreezeGeneration"], 2); + } + } + Ok(()) + } + #[tokio::test] async fn sql_write_over_limit_returns_402_with_kv_message() -> Result<()> { use rocket::data::ByteUnit;