diff --git a/CONTEXT.md b/CONTEXT.md index 0b683ae7..dc835566 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -6,15 +6,15 @@ behalf. ## Language -### Permissioned data (experimental) +### Permissioned data (Spaces proposal) **Space**: An access-control and synchronization boundary identified by an authority DID, a space type NSID, and a space key. Each author keeps a separate repository in the Space. It is distinct from the author's public repository and does not -provide end-to-end encryption. The fork supports an opt-in PostgreSQL flow for private records, shared reads, -and local synchronization; see `docs/7_ATPROTO_SPACES.md` for its limits and -remaining federation work. +provide end-to-end encryption. The fork supports PostgreSQL Spaces by default: private records and blobs, shared +reads and cross-host synchronization. See `docs/7_ATPROTO_SPACES.md` for supported +policies and `docs/8_SPACES_OPERATIONS.md` for deployment and migration boundaries. **Space authority**: The DID controlling access to a Space. It can differ from the DID of an author diff --git a/Cargo.lock b/Cargo.lock index 335eff39..70fc5400 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7844,8 +7844,10 @@ version = "0.6.6" dependencies = [ "async-trait", "chrono", + "ipld-core", "rand 0.8.5", "serde", + "serde_ipld_dagcbor", "serde_json", "sqlx", "tracing", diff --git a/Dockerfile b/Dockerfile index 81fa0b95..8327e57f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,4 +1,4 @@ -ARG DISTROLESS_IMAGE=gcr.io/distroless/cc-debian13:latest@sha256:1e3c6d9c255be500eb680cdea0ad07554f52ae92dfcbdf07043a2a435b4c1fe3 +ARG DISTROLESS_IMAGE=gcr.io/distroless/cc-debian13:latest@sha256:159783207c2cd44c2aa5715961d13c8612368ac9bd450f887e3f08fc8ea461e3 FROM node:24-trixie-slim AS frontend RUN corepack enable && corepack prepare pnpm@latest --activate @@ -62,9 +62,9 @@ RUN --mount=type=cache,id=cargo-registry,target=/usr/local/cargo/registry \ --mount=type=cache,id=cargo-git,target=/usr/local/cargo/git \ --mount=type=cache,id=tranquil-target,target=/app/target,sharing=locked \ if [ "$SLIM" = "true" ]; then \ - SQLX_OFFLINE=true cargo build --release -p tranquil-server --no-default-features; \ + SQLX_OFFLINE=true cargo build --locked --release -p tranquil-server --no-default-features; \ else \ - SQLX_OFFLINE=true cargo build --release -p tranquil-server; \ + SQLX_OFFLINE=true cargo build --locked --release -p tranquil-server; \ fi && \ cp target/release/tranquil-server /tmp/tranquil-pds && \ if [ "$COMPRESS" = "true" ] && command -v upx >/dev/null 2>&1; then upx --best --lzma /tmp/tranquil-pds; fi diff --git a/crates/tranquil-api/src/repo/blob.rs b/crates/tranquil-api/src/repo/blob.rs index aa0f7b17..bcddc713 100644 --- a/crates/tranquil-api/src/repo/blob.rs +++ b/crates/tranquil-api/src/repo/blob.rs @@ -94,21 +94,38 @@ pub async fn upload_blob( let max_size = tranquil_config::get().server.max_blob_size; let body_stream = body.into_data_stream(); - let mapped_stream = - body_stream.map(|result| result.map_err(|e| std::io::Error::other(e.to_string()))); + let too_large = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let stream_too_large = too_large.clone(); + let mut received = 0u64; + let mapped_stream = body_stream.map(move |result| { + let bytes = result.map_err(|e| std::io::Error::other(e.to_string()))?; + received = received.saturating_add(bytes.len() as u64); + if received > max_size { + stream_too_large.store(true, std::sync::atomic::Ordering::Relaxed); + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "Blob exceeds maximum upload size", + )); + } + Ok(bytes) + }); let pinned_stream: Pin> + Send>> = Box::pin(mapped_stream); info!("Starting streaming blob upload to temp key: {}", temp_key); - let upload_result = state - .blob_store - .put_stream(&temp_key, pinned_stream) - .await - .map_err(|e| { + let upload_result = match state.blob_store.put_stream(&temp_key, pinned_stream).await { + Ok(result) => result, + Err(e) => { + let _ = state.blob_store.delete(&temp_key).await; error!("Failed to stream blob to storage: {:?}", e); - ApiError::InternalError(Some("Failed to store blob".into())) - })?; + return Err(if too_large.load(std::sync::atomic::Ordering::Relaxed) { + ApiError::PayloadTooLarge("Blob exceeds maximum upload size".into()) + } else { + ApiError::InternalError(Some("Failed to store blob".into())) + }); + } + }; let size = upload_result.size; if size > max_size { @@ -148,39 +165,113 @@ pub async fn upload_blob( size, cid_str ); - match state - .repos - .blob - .insert_blob( - &cid_link, - &mime_type, - i64::try_from(size).unwrap_or(i64::MAX), - user_id, - &storage_key, - ) - .await + // Bound DB connections held during content-addressed publication. Queue + // short bursts outside the database pool, then return a retryable response. + let _publication_permit = match tokio::time::timeout( + std::time::Duration::from_secs(5), + state.blob_publication_semaphore.acquire(), + ) + .await { - Ok(_) => {} - Err(e) => { + Ok(Ok(permit)) => permit, + _ => { let _ = state.blob_store.delete(&temp_key).await; - error!("Failed to insert blob record: {:?}", e); - return Err(ApiError::InternalError(None)); + return Err(ApiError::ServiceUnavailable(Some( + "Blob publication is busy; retry later".into(), + ))); } }; - - if let Err(e) = state.blob_store.copy(&temp_key, &storage_key).await { + let publication_result = tokio::time::timeout(std::time::Duration::from_secs(30), async { + let mut publication = match &state.repos.pool { + Some(pool) => { + match tranquil_pds::storage::lock_blob_upload(pool, &storage_key, user_id).await { + Ok(tx) => Some(tx), + Err(error) => { + let _ = state.blob_store.delete(&temp_key).await; + return Err( + if error.as_database_error().and_then(|e| e.code()).as_deref() + == Some("55P03") + { + ApiError::ServiceUnavailable(Some( + "Blob storage is busy; retry later".into(), + )) + } else { + ApiError::InternalError(None) + }, + ); + } + } + } + None => None, + }; + let result = if let Some(tx) = publication.as_mut() { + tranquil_pds::storage::register_blob_upload( + tx, + tranquil_pds::storage::UploadedBlob { + cid: &cid_link, + mime_type: &mime_type, + size_bytes: size as i64, + user_id, + author: &did, + storage_key: &storage_key, + }, + ) + .await + } else { + Ok(()) + }; + if let Err(error) = result { + let _ = state.blob_store.delete(&temp_key).await; + return Err(error); + } + // Publish metadata only after the object exists. A failed deduplicated + // upload must never delete metadata belonging to an earlier upload. + if let Err(e) = state.blob_store.copy(&temp_key, &storage_key).await { + let _ = state.blob_store.delete(&temp_key).await; + error!("Failed to copy blob to final location: {:?}", e); + return Err(ApiError::InternalError(Some("Failed to store blob".into()))); + } + let result = if publication.is_none() { + state + .repos + .blob + .insert_blob(&cid_link, &mime_type, size as i64, user_id, &storage_key) + .await + .map(|_| ()) + .map_err(|_| ApiError::InternalError(None)) + } else { + Ok(()) + }; let _ = state.blob_store.delete(&temp_key).await; - if let Err(db_err) = state.repos.blob.delete_blob_by_cid(&cid_link).await { - error!( - "Failed to clean up orphaned blob record after copy failure: {:?}", - db_err - ); + result?; + if let Some(tx) = publication { + tx.commit() + .await + .map_err(|_| ApiError::InternalError(None))?; + } + Ok::<(), ApiError>(()) + }) + .await; + match publication_result { + Ok(Ok(())) => {} + Ok(Err(error)) => return Err(error), + Err(_) => { + let _ = state.blob_store.delete(&temp_key).await; + return Err(ApiError::ServiceUnavailable(Some( + "Blob publication timed out; retry later".into(), + ))); } - error!("Failed to copy blob to final location: {:?}", e); - return Err(ApiError::InternalError(Some("Failed to store blob".into()))); } - let _ = state.blob_store.delete(&temp_key).await; + let stored = state + .repos + .blob + .get_blob_metadata(&cid_link) + .await + .map_err(|_| ApiError::InternalError(None))? + .ok_or(ApiError::InternalError(None))?; + let mime_type = stored.mime_type; + let size = stored.size_bytes; if let Some(ref controller) = controller_did && let Err(e) = state diff --git a/crates/tranquil-api/src/spaces.rs b/crates/tranquil-api/src/spaces.rs index d0b0d2d8..c7723341 100644 --- a/crates/tranquil-api/src/spaces.rs +++ b/crates/tranquil-api/src/spaces.rs @@ -1,5 +1,6 @@ use tranquil_pds::spaces::store::{OperationPage, RecordPage}; mod auth; +mod blobs; mod notifications; use auth::ReadAuth; use axum::{ @@ -60,6 +61,8 @@ pub fn routes() -> Router { "/com.atproto.space.unregisterNotify", post(notifications::unregister), ) + .route("/com.atproto.space.getBlob", get(blobs::get_blob)) + .route("/com.atproto.space.listBlobs", get(blobs::list_blobs)) .route("/com.atproto.space.getRecord", get(get_record)) .route("/com.atproto.space.listRecords", get(list_records)) .route("/com.atproto.space.listRepos", get(list_repos)) @@ -71,9 +74,6 @@ pub fn routes() -> Router { } fn space_store(state: &AppState) -> Result { - if !tranquil_config::get().server.enable_atproto_spaces { - return Err(SpaceError::NotImplemented); - } Ok(SpaceStore::new( state.repos.pool.clone().ok_or(SpaceError::NotImplemented)?, )) @@ -114,7 +114,28 @@ fn require_authority(auth: &Auth, space: &SpaceRef) -> Result<()> { Err(SpaceError::Forbidden) } } +async fn check_principal_limit( + state: &AppState, + kind: tranquil_pds::state::RateLimitKind, + principal: &str, +) -> Result<()> { + if !state + .check_rate_limit(kind, &format!("spaces-principal:{principal}")) + .await + { + return Err(SpaceError::Api( + tranquil_pds::api::ApiError::RateLimitExceeded(None), + )); + } + Ok(()) +} async fn require_write_access(state: &AppState, auth: &Auth) -> Result<()> { + check_principal_limit( + state, + tranquil_pds::state::RateLimitKind::SpacesWrite, + auth.did.as_str(), + ) + .await?; require_not_migrated(state, &auth.did).await?; require_verified_or_delegated(state, &auth.0).await?; Ok(()) @@ -231,6 +252,7 @@ async fn delete_space( Ok(Json(json!({}))) } async fn get_space( + _limit: tranquil_pds::rate_limit::RateLimited, State(state): State, auth: ReadAuth, Query(input): Query, @@ -302,6 +324,7 @@ pub struct MembersQuery { limit: Option, } async fn list_members( + _limit: tranquil_pds::rate_limit::RateLimited, State(state): State, auth: Auth, Query(input): Query, @@ -358,16 +381,6 @@ async fn declarations(auth: &Auth, space: &SpaceRef) -> Vec { vec![] } } -fn contains_blob(value: &Value) -> bool { - match value { - Value::Object(map) => { - map.get("$type").and_then(Value::as_str) == Some("blob") - || map.values().any(contains_blob) - } - Value::Array(v) => v.iter().any(contains_blob), - _ => false, - } -} async fn prepare( mode: WriteMode, collection: Nsid, @@ -377,10 +390,8 @@ async fn prepare( ) -> Result { let mut status = None; let record = if let Some(record) = record { - if !record.is_object() || contains_blob(&record) { - return Err(SpaceError::Invalid( - "Space records must be objects; private blob storage is not yet supported".into(), - )); + if !record.is_object() { + return Err(SpaceError::Invalid("Space records must be objects".into())); } let record = crate::repo::record::write::ensure_record_type(&record, &collection).into_owned(); @@ -577,6 +588,7 @@ async fn apply_writes( Ok(Json(json!({"results":results}))) } async fn get_record( + _limit: tranquil_pds::rate_limit::RateLimited, State(state): State, auth: ReadAuth, Query(input): Query, @@ -604,6 +616,7 @@ pub struct RecordsQuery { exclude_values: bool, } async fn list_records( + _limit: tranquil_pds::rate_limit::RateLimited, State(state): State, auth: ReadAuth, Query(input): Query, @@ -636,6 +649,7 @@ pub struct SpacesQuery { cursor: Option, } async fn list_spaces( + _limit: tranquil_pds::rate_limit::RateLimited, State(state): State, auth: Auth, Query(input): Query, @@ -715,6 +729,7 @@ async fn sign_commit( )?)) } async fn latest_commit( + _limit: tranquil_pds::rate_limit::RateLimited, State(state): State, auth: ReadAuth, Query(input): Query, @@ -741,6 +756,7 @@ pub struct OpsQuery { exclude_values: bool, } async fn list_ops( + _limit: tranquil_pds::rate_limit::RateLimited, State(state): State, auth: ReadAuth, Query(input): Query, @@ -780,6 +796,7 @@ async fn list_ops( } async fn get_repo( + _limit: tranquil_pds::rate_limit::RateLimited, State(state): State, auth: ReadAuth, Query(input): Query, @@ -790,25 +807,20 @@ async fn get_repo( let store = space_store(&state)?; auth.require_read_access(&state, &input.space, &input.repo) .await?; - let _permit = state - .repo_export_semaphore - .acquire() + let permit = state + .spaces_export_semaphore + .clone() + .acquire_owned() .await .map_err(|_| SpaceError::NotImplemented)?; let snapshot = store.export_snapshot(&input.space, &input.repo).await?; let signed = sign_commit(&state, &input.space, &input.repo, snapshot.state) .await? .ok_or(SpaceError::RepoNotFound)?; - let mut index = std::collections::BTreeMap::new(); - let mut blocks = Vec::new(); - for record in snapshot.records { - index.insert(record.path, Ipld::Link(record.cid)); - blocks.push((record.cid, record.data)); - } let signed_bytes = serde_ipld_dagcbor::to_vec(&tranquil_pds::util::json_to_ipld(&signed)) .map_err(|_| SpaceError::internal())?; - let index_bytes = - serde_ipld_dagcbor::to_vec(&Ipld::Map(index)).map_err(|_| SpaceError::internal())?; + let index_bytes = serde_ipld_dagcbor::to_vec(&Ipld::Map(snapshot.index)) + .map_err(|_| SpaceError::internal())?; let commit_cid = jacquard_repo::mst::util::compute_cid(&signed_bytes).map_err(|_| SpaceError::internal())?; let index_cid = @@ -825,13 +837,24 @@ async fn get_repo( ld_write(&mut car, &header).map_err(|_| SpaceError::internal())?; car.extend(encode_car_block(&commit_cid, &signed_bytes)); car.extend(encode_car_block(&index_cid, &index_bytes)); - for (cid, data) in blocks { - car.extend(encode_car_block(&cid, &data)); - } - Ok(([("content-type", "application/vnd.ipld.car")], car).into_response()) + let stream = tranquil_pds::storage::stream_export( + snapshot.blocks, + car, + permit, + std::time::Duration::from_secs(tranquil_config::get().server.spaces_export_timeout_seconds), + ); + Ok(( + [ + ("content-type", "application/vnd.ipld.car"), + ("cache-control", "private, no-store"), + ], + axum::body::Body::from_stream(stream), + ) + .into_response()) } async fn list_repos( + _limit: tranquil_pds::rate_limit::RateLimited, State(state): State, auth: ReadAuth, Query(input): Query, diff --git a/crates/tranquil-api/src/spaces/auth.rs b/crates/tranquil-api/src/spaces/auth.rs index 7f9d9d8e..87e1a781 100644 --- a/crates/tranquil-api/src/spaces/auth.rs +++ b/crates/tranquil-api/src/spaces/auth.rs @@ -77,6 +77,12 @@ pub(super) async fn delegation( space_store(&state)?; require_scope(&auth, &input.space, SpaceOperation::DelegateRead)?; require_not_migrated(&state, &auth.did).await?; + check_principal_limit( + &state, + tranquil_pds::state::RateLimitKind::SpacesCredential, + auth.did.as_str(), + ) + .await?; let claims = Claims::delegation(auth.did.clone(), input.space); Ok(Json( json!({"token":credential::sign(&state,&claims).await?}), @@ -104,6 +110,12 @@ pub(super) async fn exchange( .await .map_err(|_| SpaceError::InvalidDelegationToken)?; let kid = credential::verify_http_signature(&headers, None, None)?; + check_principal_limit( + &state, + tranquil_pds::state::RateLimitKind::SpacesCredential, + delegation.iss.as_str(), + ) + .await?; assert_repo_availability( &*state.repos.repo, input.space.authority(), diff --git a/crates/tranquil-api/src/spaces/blobs.rs b/crates/tranquil-api/src/spaces/blobs.rs new file mode 100644 index 00000000..cb76e4c9 --- /dev/null +++ b/crates/tranquil-api/src/spaces/blobs.rs @@ -0,0 +1,81 @@ +use super::*; +use axum::response::IntoResponse; +use tranquil_types::CidLink; + +#[derive(Deserialize)] +pub(super) struct BlobQuery { + space: SpaceRef, + repo: Did, + cid: CidLink, +} +pub(super) async fn get_blob( + _limit: tranquil_pds::rate_limit::RateLimited, + State(state): State, + auth: ReadAuth, + Query(input): Query, +) -> Result { + auth.require_read_access(&state, &input.space, &input.repo) + .await?; + if !space_store(&state)? + .has_blob(&input.space, &input.repo, &input.cid) + .await? + { + return Err(SpaceError::Api( + tranquil_pds::api::error::ApiError::BlobNotFound(None), + )); + } + let meta = state + .repos + .blob + .get_blob_metadata(&input.cid) + .await + .map_err(|_| SpaceError::internal())? + .ok_or(SpaceError::Api( + tranquil_pds::api::error::ApiError::BlobNotFound(None), + ))?; + let data = state + .blob_store + .get_stream(&meta.storage_key) + .await + .map_err(|_| SpaceError::internal())?; + let disposition = format!("attachment; filename=\"{}\"", input.cid); + Ok(( + [ + ("content-type", meta.mime_type.as_str()), + ("cache-control", "private, no-store"), + ("content-disposition", disposition.as_str()), + ("x-content-type-options", "nosniff"), + ("content-security-policy", "default-src 'none'; sandbox"), + ], + axum::body::Body::from_stream(data), + ) + .into_response()) +} +#[derive(Deserialize)] +pub(super) struct ListBlobsQuery { + space: SpaceRef, + repo: Did, + cursor: Option, + since: Option, + limit: Option, +} +pub(super) async fn list_blobs( + _limit: tranquil_pds::rate_limit::RateLimited, + State(state): State, + auth: ReadAuth, + Query(input): Query, +) -> Result> { + auth.require_read_access(&state, &input.space, &input.repo) + .await?; + Ok(Json( + space_store(&state)? + .blobs( + &input.space, + &input.repo, + input.cursor.as_deref(), + page_limit(input.limit, 500, 1000)?, + input.since.as_ref().map(tranquil_types::Tid::as_str), + ) + .await?, + )) +} diff --git a/crates/tranquil-api/src/spaces/notifications.rs b/crates/tranquil-api/src/spaces/notifications.rs index 7cee2fba..bcc94107 100644 --- a/crates/tranquil-api/src/spaces/notifications.rs +++ b/crates/tranquil-api/src/spaces/notifications.rs @@ -13,7 +13,14 @@ async fn service_auth( let token = credential::header(headers, "authorization")? .strip_prefix("Bearer ") .ok_or(SpaceError::Forbidden)?; - service::verify(state, token, method).await + let claims = service::verify(state, token, method).await?; + check_principal_limit( + state, + tranquil_pds::state::RateLimitKind::SpacesWrite, + claims.iss.as_str(), + ) + .await?; + Ok(claims) } pub(super) async fn notify_write( @@ -78,6 +85,12 @@ async fn check_subscription( .await?; let store = space_store(state)?; store.config(&input.space).await?; + check_principal_limit( + state, + tranquil_pds::state::RateLimitKind::SpacesWrite, + &input.service, + ) + .await?; Ok(store) } pub(super) async fn register( diff --git a/crates/tranquil-config/src/lib.rs b/crates/tranquil-config/src/lib.rs index 798e7126..3a98471e 100644 --- a/crates/tranquil-config/src/lib.rs +++ b/crates/tranquil-config/src/lib.rs @@ -176,6 +176,56 @@ impl TranquilConfig { .map(|(var, guidance)| format!("{var} is no longer supported: {guidance}")), ); + for (name, value) in [ + ( + "spaces_export_timeout_seconds", + self.server.spaces_export_timeout_seconds, + ), + ( + "spaces_max_repos_per_account", + u64::from(self.server.spaces_max_repos_per_account), + ), + ( + "spaces_max_records_per_account", + u64::from(self.server.spaces_max_records_per_account), + ), + ( + "spaces_max_record_bytes_per_account", + self.server.spaces_max_record_bytes_per_account, + ), + ( + "spaces_max_ops_per_repo", + u64::from(self.server.spaces_max_ops_per_repo), + ), + ( + "spaces_max_members_per_space", + u64::from(self.server.spaces_max_members_per_space), + ), + ( + "spaces_max_syncers_per_space", + u64::from(self.server.spaces_max_syncers_per_space), + ), + ] { + if value == 0 || value > i64::MAX as u64 { + errors.push(format!("server.{name} must be between 1 and {}", i64::MAX)); + } + } + + if self.server.spaces_max_blob_bytes_per_account == 0 + || self.server.spaces_max_blob_bytes_per_account > i64::MAX as u64 + { + errors.push("server.spaces_max_blob_bytes_per_account must be positive and fit in a signed 64-bit integer".into()); + } + + if self.server.max_pending_blob_bytes_per_account == 0 + || self.server.max_pending_blob_bytes_per_account > i64::MAX as u64 + || self.server.max_pending_blobs_per_account == 0 + { + errors.push( + "Pending blob quotas must be positive and fit in a signed 64-bit integer".into(), + ); + } + // -- secrets ---------------------------------------------------------- if !ignore_secrets && !self.secrets.allow_insecure && !cfg!(test) { if let Some(ref s) = self.secrets.jwt_secret { @@ -429,9 +479,48 @@ impl TranquilConfig { #[derive(Debug, Config)] #[config(layer_attr(serde(deny_unknown_fields)))] pub struct ServerConfig { - /// Experimental permissioned Spaces APIs (Postgres backend only). - #[config(env = "ENABLE_ATPROTO_SPACES", default = false)] - pub enable_atproto_spaces: bool, + /// Deprecated compatibility setting; Spaces is always available on PostgreSQL. + /// Retained so existing configuration files continue to deserialize. + #[config(env = "ENABLE_ATPROTO_SPACES")] + pub enable_atproto_spaces: Option, + + /// Deadline for a streamed private repository export, including slow readers. + #[config(env = "SPACES_EXPORT_TIMEOUT_SECONDS", default = 900)] + pub spaces_export_timeout_seconds: u64, + + /// Maximum number of private repositories hosted per account. + #[config(env = "SPACES_MAX_REPOS_PER_ACCOUNT", default = 1000)] + pub spaces_max_repos_per_account: u32, + + /// Maximum live private records across an account's repositories. + #[config(env = "SPACES_MAX_RECORDS_PER_ACCOUNT", default = 100000)] + pub spaces_max_records_per_account: u32, + + /// Maximum DAG-CBOR record bytes across an account's private repositories. + #[config(env = "SPACES_MAX_RECORD_BYTES_PER_ACCOUNT", default = 1073741824)] + pub spaces_max_record_bytes_per_account: u64, + + /// Maximum distinct blob bytes referenced by an account's private repositories. + #[config(env = "SPACES_MAX_BLOB_BYTES_PER_ACCOUNT", default = 10737418240u64)] + pub spaces_max_blob_bytes_per_account: u64, + + /// Unattached uploads are bounded separately, and expire after 24 hours. + #[config(env = "MAX_PENDING_BLOB_BYTES_PER_ACCOUNT", default = 1073741824)] + pub max_pending_blob_bytes_per_account: u64, + #[config(env = "MAX_PENDING_BLOBS_PER_ACCOUNT", default = 10000)] + pub max_pending_blobs_per_account: u32, + + /// Retained operations per private repository; older positions require getRepo. + #[config(env = "SPACES_MAX_OPS_PER_REPO", default = 10000)] + pub spaces_max_ops_per_repo: u32, + + /// Maximum member entries and tracked writers per Space. + #[config(env = "SPACES_MAX_MEMBERS_PER_SPACE", default = 10000)] + pub spaces_max_members_per_space: u32, + + /// Maximum active notification services per Space. + #[config(env = "SPACES_MAX_SYNCERS_PER_SPACE", default = 100)] + pub spaces_max_syncers_per_space: u32, /// Public hostname of the PDS, such as `pds.example.com`. #[config(env = "PDS_HOSTNAME")] diff --git a/crates/tranquil-db-traits/src/repo.rs b/crates/tranquil-db-traits/src/repo.rs index 35207534..a2bd4ec3 100644 --- a/crates/tranquil-db-traits/src/repo.rs +++ b/crates/tranquil-db-traits/src/repo.rs @@ -317,6 +317,7 @@ pub struct ApplyCommitResult { #[derive(Debug, Clone, PartialEq, Eq)] pub enum ApplyCommitError { + BlobNotOwned, RepoNotFound, ConcurrentModification, Database(String), diff --git a/crates/tranquil-db/Cargo.toml b/crates/tranquil-db/Cargo.toml index e468eb39..a350872d 100644 --- a/crates/tranquil-db/Cargo.toml +++ b/crates/tranquil-db/Cargo.toml @@ -20,3 +20,6 @@ tracing = { workspace = true } uuid = { workspace = true } sqlx = { workspace = true } + +ipld-core = { workspace = true } +serde_ipld_dagcbor = { workspace = true } diff --git a/crates/tranquil-db/src/postgres/blob.rs b/crates/tranquil-db/src/postgres/blob.rs index 6e69a83b..2c5b8d8f 100644 --- a/crates/tranquil-db/src/postgres/blob.rs +++ b/crates/tranquil-db/src/postgres/blob.rs @@ -192,10 +192,12 @@ impl BlobRepository for PostgresBlobRepository { } async fn get_blob_storage_keys_by_user(&self, user_id: Uuid) -> Result, DbError> { - let results = sqlx::query_scalar!( - r#"SELECT storage_key as "storage_key!" FROM blobs WHERE created_by_user = $1"#, - user_id - ) + let results = sqlx::query_scalar( + r#"SELECT b.storage_key FROM blobs b WHERE b.created_by_user=$1 + AND NOT EXISTS(SELECT 1 FROM blob_upload_claims c JOIN users u ON u.did=c.author WHERE c.cid=b.cid AND u.id<>$1) + AND NOT EXISTS(SELECT 1 FROM record_blobs r WHERE r.blob_cid=b.cid AND r.repo_id<>$1 AND b.legacy_public) + AND NOT EXISTS(SELECT 1 FROM space_record_blobs r JOIN users u ON u.did=r.author WHERE r.cid=b.cid AND u.id<>$1)"#) + .bind(user_id) .fetch_all(&self.pool) .await .map_err(map_sqlx_error)?; @@ -212,15 +214,15 @@ impl BlobRepository for PostgresBlobRepository { let uris_str: Vec<&str> = record_uris.iter().map(|u| u.as_str()).collect(); let cids_str: Vec<&str> = blob_cids.iter().map(|c| c.as_str()).collect(); - sqlx::query!( + sqlx::query( r#"INSERT INTO record_blobs (repo_id, record_uri, blob_cid) SELECT $1, record_uri, blob_cid FROM UNNEST($2::text[], $3::text[]) AS t(record_uri, blob_cid) - ON CONFLICT (repo_id, record_uri, blob_cid) DO NOTHING"#, - repo_id, - &uris_str as &[&str], - &cids_str as &[&str] + ON CONFLICT (repo_id, record_uri, blob_cid) DO UPDATE SET record_cid=EXCLUDED.record_cid"# ) + .bind(repo_id) + .bind(&uris_str) + .bind(&cids_str) .execute(&self.pool) .await .map_err(map_sqlx_error)?; diff --git a/crates/tranquil-db/src/postgres/blob_refs.rs b/crates/tranquil-db/src/postgres/blob_refs.rs new file mode 100644 index 00000000..bd44fb02 --- /dev/null +++ b/crates/tranquil-db/src/postgres/blob_refs.rs @@ -0,0 +1,94 @@ +use ipld_core::ipld::Ipld; +use tranquil_db_traits::ApplyCommitInput; + +fn find(value: &Ipld, out: &mut Vec) { + match value { + Ipld::Map(map) if map.get("$type") == Some(&Ipld::String("blob".into())) => { + if let Some(Ipld::Link(cid)) = map.get("ref") { + out.push(cid.to_string()); + } + } + Ipld::Map(map) => map.values().for_each(|value| find(value, out)), + Ipld::List(values) => values.iter().for_each(|value| find(value, out)), + _ => {} + } +} + +/// Blob visibility and record publication change in the same transaction. +pub(super) async fn index_commit( + tx: &mut sqlx::PgConnection, + input: &ApplyCommitInput, +) -> Result<(), tranquil_db_traits::ApplyCommitError> { + for (collection, key) in input + .record_upserts + .iter() + .map(|r| (&r.collection, &r.rkey)) + .chain( + input + .record_deletes + .iter() + .map(|r| (&r.collection, &r.rkey)), + ) + { + let uri = format!("at://{}/{collection}/{key}", input.did); + sqlx::query("DELETE FROM record_blobs WHERE repo_id=$1 AND record_uri=$2") + .bind(input.user_id) + .bind(uri) + .execute(&mut *tx) + .await + .map_err(|e| tranquil_db_traits::ApplyCommitError::Database(e.to_string()))?; + } + let mut references = Vec::new(); + for record in &input.record_upserts { + let cid_bytes = record + .cid + .to_cid() + .ok_or_else(|| { + tranquil_db_traits::ApplyCommitError::Database("Invalid record CID".into()) + })? + .to_bytes(); + let block = input + .commit_event + .blocks + .as_deref() + .unwrap_or_default() + .iter() + .find(|block| block.cid_bytes == cid_bytes) + .ok_or_else(|| { + tranquil_db_traits::ApplyCommitError::Database( + "Missing committed record block".into(), + ) + })?; + let value: Ipld = serde_ipld_dagcbor::from_slice(&block.data).map_err(|_| { + tranquil_db_traits::ApplyCommitError::Database("Invalid committed record block".into()) + })?; + let mut blobs = Vec::new(); + find(&value, &mut blobs); + let uri = format!("at://{}/{}/{}", input.did, record.collection, record.rkey); + references.push((uri, record.cid.as_str(), blobs)); + } + let cids: std::collections::BTreeSet<_> = references + .iter() + .flat_map(|(_, _, blobs)| blobs.iter()) + .collect(); + for cid in cids { + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended('blob:' || $1,0))") + .bind(cid) + .execute(&mut *tx) + .await + .map_err(|e| tranquil_db_traits::ApplyCommitError::Database(e.to_string()))?; + } + for (uri, record_cid, blobs) in references { + for blob in blobs { + let allowed: bool = sqlx::query_scalar( + "SELECT NOT EXISTS(SELECT 1 FROM blobs WHERE cid=$1) OR EXISTS(SELECT 1 FROM blobs WHERE cid=$1 AND (legacy_public OR created_by_user=$2)) OR EXISTS(SELECT 1 FROM blob_upload_claims WHERE cid=$1 AND author=$3)") + .bind(&blob).bind(input.user_id).bind(input.did.as_str()).fetch_one(&mut *tx).await.map_err(|e| tranquil_db_traits::ApplyCommitError::Database(e.to_string()))?; + if !allowed { + return Err(tranquil_db_traits::ApplyCommitError::BlobNotOwned); + } + sqlx::query("INSERT INTO record_blobs(repo_id,record_uri,blob_cid,record_cid) VALUES($1,$2,$3,$4) ON CONFLICT DO NOTHING") + .bind(input.user_id).bind(&uri).bind(blob).bind(record_cid).execute(&mut *tx).await.map_err(|e| tranquil_db_traits::ApplyCommitError::Database(e.to_string()))?; + } + } + Ok(()) +} diff --git a/crates/tranquil-db/src/postgres/mod.rs b/crates/tranquil-db/src/postgres/mod.rs index 1f97998e..fce75e80 100644 --- a/crates/tranquil-db/src/postgres/mod.rs +++ b/crates/tranquil-db/src/postgres/mod.rs @@ -1,5 +1,6 @@ mod backlink; mod blob; +mod blob_refs; mod delegation; mod event_notifier; mod infra; diff --git a/crates/tranquil-db/src/postgres/repo.rs b/crates/tranquil-db/src/postgres/repo.rs index 6b10f6ec..c84eb75e 100644 --- a/crates/tranquil-db/src/postgres/repo.rs +++ b/crates/tranquil-db/src/postgres/repo.rs @@ -1513,6 +1513,8 @@ impl RepoRepository for PostgresRepoRepository { .map_err(|e| ApplyCommitError::Database(e.to_string()))?; } + super::blob_refs::index_commit(&mut tx, &input).await?; + if !input.backlinks_to_remove.is_empty() { let remove_uris: Vec<&str> = input .backlinks_to_remove diff --git a/crates/tranquil-db/src/postgres/user.rs b/crates/tranquil-db/src/postgres/user.rs index bedb465a..0a919d47 100644 --- a/crates/tranquil-db/src/postgres/user.rs +++ b/crates/tranquil-db/src/postgres/user.rs @@ -2089,6 +2089,11 @@ impl UserRepository for PostgresUserRepository { async fn delete_account_complete(&self, user_id: Uuid, did: &Did) -> Result<(), DbError> { let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?; + sqlx::query("SELECT id FROM users WHERE id=$1 FOR NO KEY UPDATE") + .bind(user_id) + .fetch_optional(&mut *tx) + .await + .map_err(map_sqlx_error)?; sqlx::query!("DELETE FROM session_tokens WHERE did = $1", did.as_str()) .execute(&mut *tx) .await @@ -2101,7 +2106,8 @@ impl UserRepository for PostgresUserRepository { .execute(&mut *tx) .await .map_err(map_sqlx_error)?; - sqlx::query!("DELETE FROM blobs WHERE created_by_user = $1", user_id) + sqlx::query("SELECT release_account_blobs($1)") + .bind(user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; @@ -2146,6 +2152,11 @@ impl UserRepository for PostgresUserRepository { async fn admin_delete_account_complete(&self, user_id: Uuid, did: &Did) -> Result<(), DbError> { let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?; + sqlx::query("SELECT id FROM users WHERE id=$1 FOR NO KEY UPDATE") + .bind(user_id) + .fetch_optional(&mut *tx) + .await + .map_err(map_sqlx_error)?; sqlx::query!("DELETE FROM session_tokens WHERE did = $1", did.as_str()) .execute(&mut *tx) .await @@ -2165,7 +2176,8 @@ impl UserRepository for PostgresUserRepository { .execute(&mut *tx) .await .map_err(map_sqlx_error)?; - sqlx::query!("DELETE FROM blobs WHERE created_by_user = $1", user_id) + sqlx::query("SELECT release_account_blobs($1)") + .bind(user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; @@ -2403,8 +2415,14 @@ impl UserRepository for PostgresUserRepository { async fn delete_account_with_firehose(&self, user_id: Uuid, did: &Did) -> Result<(), DbError> { let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?; + sqlx::query("SELECT id FROM users WHERE id=$1 FOR NO KEY UPDATE") + .bind(user_id) + .fetch_optional(&mut *tx) + .await + .map_err(map_sqlx_error)?; - sqlx::query!("DELETE FROM blobs WHERE created_by_user = $1", user_id) + sqlx::query("SELECT release_account_blobs($1)") + .bind(user_id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; diff --git a/crates/tranquil-infra/src/lib.rs b/crates/tranquil-infra/src/lib.rs index 76a07832..35d9a69d 100644 --- a/crates/tranquil-infra/src/lib.rs +++ b/crates/tranquil-infra/src/lib.rs @@ -30,12 +30,19 @@ pub struct StreamUploadResult { pub size: u64, } +pub type BlobDownloadStream = Pin> + Send>>; + #[async_trait] pub trait BlobStorage: Send + Sync { async fn put(&self, key: &str, data: &[u8]) -> Result<(), StorageError>; async fn put_bytes(&self, key: &str, data: Bytes) -> Result<(), StorageError>; async fn get(&self, key: &str) -> Result, StorageError>; async fn get_bytes(&self, key: &str) -> Result; + /// Real storage backends override this with bounded, backpressured reads. + async fn get_stream(&self, key: &str) -> Result { + let bytes = self.get_bytes(key).await?; + Ok(Box::pin(futures::stream::once(async move { Ok(bytes) }))) + } async fn get_head(&self, key: &str, size: usize) -> Result; async fn delete(&self, key: &str) -> Result<(), StorageError>; async fn put_stream( diff --git a/crates/tranquil-pds/Cargo.toml b/crates/tranquil-pds/Cargo.toml index 188f9a24..df4d81d0 100644 --- a/crates/tranquil-pds/Cargo.toml +++ b/crates/tranquil-pds/Cargo.toml @@ -77,6 +77,8 @@ webauthn-rs = { workspace = true } webauthn-rs-proto = { workspace = true } sfv = "0.15" +tempfile = "3" + [features] bsky = ["bsky-support"] bsky-support = [] @@ -89,7 +91,6 @@ native-tls-roots = ["tranquil-oauth/native-tls-roots"] [dev-dependencies] tranquil-infra = { workspace = true, features = ["testing"] } -tempfile = "3" ciborium = { workspace = true } ctor = { workspace = true } testcontainers = { workspace = true } diff --git a/crates/tranquil-pds/src/rate_limit/extractor.rs b/crates/tranquil-pds/src/rate_limit/extractor.rs index 68a5db75..82e113e5 100644 --- a/crates/tranquil-pds/src/rate_limit/extractor.rs +++ b/crates/tranquil-pds/src/rate_limit/extractor.rs @@ -275,3 +275,16 @@ pub async fn check_user_rate_limit_with_message( } Ok(UserRateLimitProof::new()) } + +pub struct SpacesReadLimit; +impl RateLimitPolicy for SpacesReadLimit { + const KIND: RateLimitKind = RateLimitKind::SpacesRead; +} +pub struct SpacesWriteLimit; +impl RateLimitPolicy for SpacesWriteLimit { + const KIND: RateLimitKind = RateLimitKind::SpacesWrite; +} +pub struct SpacesCredentialLimit; +impl RateLimitPolicy for SpacesCredentialLimit { + const KIND: RateLimitKind = RateLimitKind::SpacesCredential; +} diff --git a/crates/tranquil-pds/src/rate_limit/mod.rs b/crates/tranquil-pds/src/rate_limit/mod.rs index f14d1913..1818aeb7 100644 --- a/crates/tranquil-pds/src/rate_limit/mod.rs +++ b/crates/tranquil-pds/src/rate_limit/mod.rs @@ -19,6 +19,9 @@ fn keyed_limiter(kind: RateLimitKind) -> Arc { #[derive(Clone)] pub struct RateLimiters { + pub spaces_read: Arc, + pub spaces_write: Arc, + pub spaces_credential: Arc, pub login: Arc, pub oauth_token: Arc, pub oauth_authorize: Arc, @@ -50,6 +53,9 @@ impl Default for RateLimiters { impl RateLimiters { pub fn new() -> Self { Self { + spaces_read: keyed_limiter(RateLimitKind::SpacesRead), + spaces_write: keyed_limiter(RateLimitKind::SpacesWrite), + spaces_credential: keyed_limiter(RateLimitKind::SpacesCredential), login: keyed_limiter(RateLimitKind::Login), oauth_token: keyed_limiter(RateLimitKind::OAuthToken), oauth_authorize: keyed_limiter(RateLimitKind::OAuthAuthorize), diff --git a/crates/tranquil-pds/src/repo_ops.rs b/crates/tranquil-pds/src/repo_ops.rs index 56c3cfb0..6eff9936 100644 --- a/crates/tranquil-pds/src/repo_ops.rs +++ b/crates/tranquil-pds/src/repo_ops.rs @@ -24,6 +24,7 @@ use uuid::Uuid; #[derive(Debug)] pub enum CommitError { + BlobNotOwned, InvalidDid(String), SigningFailed(String), SerializationFailed(String), @@ -45,6 +46,7 @@ pub enum CommitError { impl std::fmt::Display for CommitError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { + Self::BlobNotOwned => write!(f, "Blob was not uploaded by this account"), Self::InvalidDid(e) => write!(f, "Invalid DID: {}", e), Self::SigningFailed(e) => write!(f, "Failed to sign commit: {}", e), Self::SerializationFailed(e) => write!(f, "Failed to serialize signed commit: {}", e), @@ -74,6 +76,7 @@ impl std::error::Error for CommitError {} impl From for ApiError { fn from(err: CommitError) -> Self { match err { + CommitError::BlobNotOwned => ApiError::Forbidden, CommitError::ConcurrentModification => { ApiError::InvalidSwap(Some("Repo has been modified".into())) } @@ -570,6 +573,15 @@ pub async fn finalize_repo_write( mst: Mst, params: FinalizeParams<'_>, ) -> Result { + if let Some(pool) = &state.repos.pool { + crate::storage::BlobAccess::new(pool.clone()) + .check_blob_ownership(params.did, params.blob_cids) + .await + .map_err(|error| match error { + crate::spaces::SpaceError::Forbidden => ApiError::Forbidden, + _ => ApiError::InternalError(None), + })?; + } let new_mst_root = mst .persist() .await @@ -1100,6 +1112,7 @@ pub async fn commit_and_log( .apply_commit(input) .await .map_err(|e| match e { + ApplyCommitError::BlobNotOwned => CommitError::BlobNotOwned, ApplyCommitError::RepoNotFound => CommitError::RepoNotFound, ApplyCommitError::ConcurrentModification => CommitError::ConcurrentModification, ApplyCommitError::Database(msg) => CommitError::DatabaseError(msg), diff --git a/crates/tranquil-pds/src/scheduled.rs b/crates/tranquil-pds/src/scheduled.rs index 438e3c6a..8ebfd25f 100644 --- a/crates/tranquil-pds/src/scheduled.rs +++ b/crates/tranquil-pds/src/scheduled.rs @@ -361,6 +361,7 @@ pub async fn start_scheduled_tasks( repo_repo: Arc, block_store: AnyBlockStore, eventlog_segments_dir: Option, + pool: Option, shutdown: CancellationToken, ) { let cfg = tranquil_config::get(); @@ -461,10 +462,16 @@ pub async fn start_scheduled_tasks( user_repo.as_ref(), blob_repo.as_ref(), blob_store.as_ref(), + pool.as_ref(), ).await { error!("Error processing scheduled deletions: {}", e); } + if let Some(pool) = pool.as_ref() + && let Err(error) = crate::storage::reclaim_expired_uploads(pool, blob_store.as_ref()).await { + error!(%error, "Expired upload cleanup failed"); + } + match sso_repo.cleanup_expired_sso_auth_states().await { Ok(count) if count > 0 => { info!(count = count, "Cleaned up expired SSO auth states"); @@ -691,6 +698,7 @@ async fn process_scheduled_deletions( user_repo: &dyn UserRepository, blob_repo: &dyn BlobRepository, blob_store: &dyn BlobStorage, + pool: Option<&sqlx::PgPool>, ) -> anyhow::Result<()> { let accounts_to_delete = user_repo .get_accounts_scheduled_for_deletion(100) @@ -708,8 +716,15 @@ async fn process_scheduled_deletions( ); futures::future::join_all(accounts_to_delete.into_iter().map(|account| async move { - let result = - delete_account_data(user_repo, blob_repo, blob_store, account.id, &account.did).await; + let result = delete_account_data( + user_repo, + blob_repo, + blob_store, + pool, + account.id, + &account.did, + ) + .await; (account.did, account.handle, result) })) .await @@ -728,6 +743,7 @@ async fn delete_account_data( user_repo: &dyn UserRepository, blob_repo: &dyn BlobRepository, blob_store: &dyn BlobStorage, + pool: Option<&sqlx::PgPool>, user_id: uuid::Uuid, did: &Did, ) -> anyhow::Result<()> { @@ -736,20 +752,29 @@ async fn delete_account_data( .await .context("DB error fetching blob keys")?; - futures::future::join_all(blob_storage_keys.iter().map(|storage_key| async move { - (storage_key, blob_store.delete(storage_key).await) - })) - .await - .into_iter() - .filter_map(|(key, result)| result.err().map(|e| (key, e))) - .for_each(|(key, e)| { - warn!(storage_key = %key, error = %e, "Failed to delete blob from storage (continuing anyway)"); - }); - user_repo .delete_account_with_firehose(user_id, did) .await .context("Failed to delete account")?; + for storage_key in &blob_storage_keys { + let mut guard = match pool { + Some(pool) => Some(crate::storage::lock_blob_object(pool, storage_key).await?), + None => None, + }; + if let Some(tx) = guard.as_mut() { + let referenced: bool = + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM blobs WHERE storage_key=$1)") + .bind(storage_key) + .fetch_one(&mut **tx) + .await?; + if referenced { + continue; + } + } + if let Err(e) = blob_store.delete(storage_key).await { + warn!(storage_key = %storage_key, error = %e, "Failed to delete account blob from storage"); + } + } info!( did = %did, diff --git a/crates/tranquil-pds/src/spaces/mod.rs b/crates/tranquil-pds/src/spaces/mod.rs index fa9f8a50..c92f5e04 100644 --- a/crates/tranquil-pds/src/spaces/mod.rs +++ b/crates/tranquil-pds/src/spaces/mod.rs @@ -15,6 +15,8 @@ use serde_json::json; #[derive(Debug, thiserror::Error)] pub enum SpaceError { + #[error("RepoHistoryUnavailable")] + RepoHistoryUnavailable, #[error("FutureRev")] FutureRev, #[error("ServiceNotResolvable")] @@ -66,6 +68,7 @@ impl IntoResponse for SpaceError { fn into_response(self) -> Response { match self { Self::Api(error) => error.into_response(), + Self::RepoHistoryUnavailable => (StatusCode::BAD_REQUEST, Json(json!({"error":"RepoHistoryUnavailable","message":"Sync position is no longer retained; use com.atproto.space.getRepo"}))).into_response(), Self::Invalid(message) => ( StatusCode::BAD_REQUEST, Json(json!({"error":"InvalidRequest","message":message})), diff --git a/crates/tranquil-pds/src/spaces/notifications.rs b/crates/tranquil-pds/src/spaces/notifications.rs index ddc2d173..4c7d0ae0 100644 --- a/crates/tranquil-pds/src/spaces/notifications.rs +++ b/crates/tranquil-pds/src/spaces/notifications.rs @@ -58,9 +58,6 @@ async fn deliver(state: &AppState, event: &Delivery) -> Result /// Bounded drain, also used by disposable integration fixtures. pub async fn drain(state: &AppState) -> Result<(), SpaceError> { - if !tranquil_config::get().server.enable_atproto_spaces { - return Ok(()); - } let Some(pool) = state.repos.pool.clone() else { return Ok(()); }; @@ -79,7 +76,7 @@ pub async fn drain(state: &AppState) -> Result<(), SpaceError> { } pub async fn run(state: AppState, shutdown: CancellationToken) { - if !tranquil_config::get().server.enable_atproto_spaces || state.repos.pool.is_none() { + if state.repos.pool.is_none() { return; } let mut ticks = tokio::time::interval(std::time::Duration::from_secs(1)); diff --git a/crates/tranquil-pds/src/spaces/store.rs b/crates/tranquil-pds/src/spaces/store.rs index 7e3f7943..3c0351b2 100644 --- a/crates/tranquil-pds/src/spaces/store.rs +++ b/crates/tranquil-pds/src/spaces/store.rs @@ -1,7 +1,8 @@ +mod blobs; pub mod notifications; mod read; use super::{SpaceError, commit}; -pub use read::{ExportRecord, OperationPage, RecordPage, RepoExport, RepoOps}; +pub use read::{OperationPage, RecordPage, RepoExport, RepoOps}; use serde_json::{Value, json}; use sha2::{Digest, Sha256}; use sqlx::{PgPool, Row}; @@ -92,6 +93,74 @@ fn next_revision(previous: Option<&str>) -> Result { } impl SpaceStore { + // A single account lock serializes quota checks across different spaces. + // Always acquire it before repository/configuration locks. + async fn lock_account(tx: &mut sqlx::PgConnection, author: &Did) -> Result<(), SpaceError> { + // Bound waiters behind blob publication; they must not occupy the pool + // for the whole object-store deadline. + sqlx::query("SET LOCAL lock_timeout='2s'") + .execute(&mut *tx) + .await?; + sqlx::query("SELECT did FROM users WHERE did=$1 FOR NO KEY UPDATE") + .bind(author.as_str()) + .fetch_optional(tx) + .await + .map_err(|error| { + if error.as_database_error().and_then(|e| e.code()).as_deref() == Some("55P03") { + SpaceError::Api(crate::api::ApiError::ServiceUnavailable(Some( + "Account storage is busy; retry later".into(), + ))) + } else { + SpaceError::from(error) + } + })? + .ok_or(SpaceError::RepoNotFound)?; + Ok(()) + } + + async fn check_account_usage( + tx: &mut sqlx::PgConnection, + author: &Did, + ) -> Result<(), SpaceError> { + let usage = sqlx::query("SELECT COUNT(*) AS repos, COALESCE(SUM(record_count),0)::bigint AS records, COALESCE(SUM(record_bytes),0)::bigint AS bytes FROM space_repos WHERE author=$1") + .bind(author.as_str()).fetch_one(&mut *tx).await?; + let blob_bytes: i64 = sqlx::query_scalar("SELECT COALESCE(SUM(b.size_bytes),0)::bigint FROM blobs b WHERE EXISTS(SELECT 1 FROM space_record_blobs r WHERE r.cid=b.cid AND r.author=$1)") + .bind(author.as_str()).fetch_one(&mut *tx).await?; + let limits = &tranquil_config::get().server; + if blob_bytes as u64 > limits.spaces_max_blob_bytes_per_account + || usage.get::("repos") as u64 > u64::from(limits.spaces_max_repos_per_account) + || usage.get::("records") as u64 + > u64::from(limits.spaces_max_records_per_account) + || usage.get::("bytes") as u64 > limits.spaces_max_record_bytes_per_account + { + return Err(SpaceError::Api( + crate::api::error::ApiError::InvalidRequest( + "Private repository storage quota exceeded".into(), + ), + )); + } + Ok(()) + } + + pub async fn compact_operations( + tx: &mut sqlx::PgConnection, + space: &SpaceRef, + author: &Did, + retain: i64, + ) -> Result<(), SpaceError> { + // Drop only entire revisions so a retained batch is never partial. + sqlx::query(r#"WITH boundary AS ( + SELECT rev FROM space_ops WHERE space=$1 AND author=$2 ORDER BY seq DESC OFFSET $3 LIMIT 1 + ), removed AS ( + DELETE FROM space_ops WHERE space=$1 AND author=$2 AND rev<(SELECT rev FROM boundary) RETURNING seq,rev + ) UPDATE space_repos SET ops_floor_rev=COALESCE((SELECT MAX(rev) FROM removed),ops_floor_rev), + ops_floor_seq=COALESCE((SELECT MAX(seq) FROM removed),ops_floor_seq), + ops_count=ops_count-(SELECT COUNT(*) FROM removed) + WHERE space=$1 AND author=$2"#) + .bind(space.to_string()).bind(author.as_str()).bind(retain.max(1)).execute(tx).await?; + Ok(()) + } + pub fn new(pool: PgPool) -> Self { Self { pool } } @@ -220,6 +289,7 @@ impl SpaceStore { } pub async fn create(&self, space: &SpaceRef, config: &Value) -> Result<(), SpaceError> { let mut tx = self.pool.begin().await?; + Self::lock_account(&mut tx, space.authority()).await?; let query = r#" INSERT INTO space_configs(uri,owner_did,config) VALUES($1,$2,$3) @@ -242,6 +312,17 @@ impl SpaceStore { .bind(space.authority().as_str()) .bind(vec![0u8;2048]) .execute(&mut *tx).await?; + Self::check_account_usage(&mut tx, space.authority()).await?; + let configs: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM space_configs WHERE owner_did=$1") + .bind(space.authority().as_str()) + .fetch_one(&mut *tx) + .await?; + if configs > i64::from(tranquil_config::get().server.spaces_max_repos_per_account) { + return Err(SpaceError::Invalid( + "Space configuration quota exceeded".into(), + )); + } tx.commit().await?; Ok(()) } @@ -343,6 +424,17 @@ impl SpaceStore { .execute(&mut *tx) .await?; } + let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM space_members WHERE space=$1") + .bind(space.to_string()) + .fetch_one(&mut *tx) + .await?; + if access.is_some() + && count > i64::from(tranquil_config::get().server.spaces_max_members_per_space) + { + return Err(SpaceError::Invalid( + "Space membership quota exceeded".into(), + )); + } if access.is_none_or(|access| !access.read) { Self::revoke_issued(&mut tx, space, Some(did)).await?; } @@ -393,13 +485,29 @@ impl SpaceStore { } let uri = space.to_string(); let mut tx = self.pool.begin().await?; + Self::lock_account(&mut tx, author).await?; + let blobs: std::collections::BTreeSet<_> = writes + .iter() + .filter_map(|write| write.record.as_ref()) + .flat_map(|record| crate::repo_ops::extract_blob_cids(&record.value)) + .map(|cid| cid.to_string()) + .collect(); + for cid in blobs { + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended('blob:' || $1,0))") + .bind(cid) + .execute(&mut *tx) + .await?; + } + let config = sqlx::query_scalar::<_, bool>( "SELECT deleted FROM space_configs WHERE uri=$1 FOR UPDATE", ) .bind(&uri) .fetch_optional(&mut *tx) .await?; - if config == Some(true) { + // Authors retain control of their own records after an authority deletes + // the Space. Permit scoped deletes so storage quota is not trapped. + if config == Some(true) && writes.iter().any(|write| write.record.is_some()) { return Err(SpaceError::SpaceDeleted); } sqlx::query("INSERT INTO space_repos(space,author,hash_state) VALUES($1,$2,$3) ON CONFLICT DO NOTHING") @@ -477,10 +585,10 @@ impl SpaceStore { false, )?; let query = r#" - INSERT INTO space_records(space,author,collection,rkey,cid,data,value) - VALUES($1,$2,$3,$4,$5,$6,$7) + INSERT INTO space_records(space,author,collection,rkey,cid,data,value,rev) + VALUES($1,$2,$3,$4,$5,$6,$7,$8) ON CONFLICT(space,author,collection,rkey) DO UPDATE - SET cid=$5,data=$6,value=$7 + SET cid=$5,data=$6,value=$7,rev=$8 "#; sqlx::query(query) .bind(&uri) @@ -490,6 +598,7 @@ impl SpaceStore { .bind(cid) .bind(data) .bind(value) + .bind(&rev) .execute(&mut *tx) .await?; Some(cid.clone()) @@ -502,6 +611,7 @@ impl SpaceStore { .execute(&mut *tx).await?; None }; + Self::attach_blobs(&mut tx, space, author, write).await?; sqlx::query("INSERT INTO space_ops(space,author,rev,collection,rkey,cid,prev) VALUES($1,$2,$3,$4,$5,$6,$7)") .bind(&uri) .bind(author.as_str()) @@ -528,10 +638,20 @@ impl SpaceStore { .bind(&hash) .execute(&mut *tx) .await?; + // Deletions must remain possible when an operator lowers a quota. + let count: i64 = sqlx::query_scalar("UPDATE space_repos SET ops_count=ops_count+$3 WHERE space=$1 AND author=$2 RETURNING ops_count") + .bind(space.to_string()).bind(author.as_str()).bind(writes.len() as i64).fetch_one(&mut *tx).await?; + let retain = i64::from(tranquil_config::get().server.spaces_max_ops_per_repo.max(1)); + if count > retain + 1000 { + Self::compact_operations(&mut tx, space, author, retain).await?; + } + if writes.iter().any(|write| write.record.is_some()) { + Self::check_account_usage(&mut tx, author).await?; + } let digest = Sha256::digest(&hash).to_vec(); if config == Some(false) { Self::track_writer(&mut tx, space, author, &rev, &digest).await?; - } else { + } else if config.is_none() { use base64::Engine; Self::enqueue(&mut tx, notifications::OutboundNotification { space, issuer: author, target: &super::service::space_host_audience(space.authority()), @@ -539,6 +659,13 @@ impl SpaceStore { body: json!({"space":space,"repo":author,"repoRev":rev,"hash":{"$bytes":base64::engine::general_purpose::STANDARD.encode(digest)}}), write_key: Some(format!("{space}/{author}")) }).await?; } + if config == Some(true) { + sqlx::query("DELETE FROM space_repos WHERE space=$1 AND author=$2 AND record_count=0") + .bind(&uri) + .bind(author.as_str()) + .execute(&mut *tx) + .await?; + } tx.commit().await?; Ok(results) } diff --git a/crates/tranquil-pds/src/spaces/store/blobs.rs b/crates/tranquil-pds/src/spaces/store/blobs.rs new file mode 100644 index 00000000..b3753d7b --- /dev/null +++ b/crates/tranquil-pds/src/spaces/store/blobs.rs @@ -0,0 +1,98 @@ +use super::*; +use tranquil_types::CidLink; + +impl SpaceStore { + pub(super) async fn attach_blobs( + tx: &mut sqlx::PgConnection, + space: &SpaceRef, + author: &Did, + write: &PreparedWrite, + ) -> Result<(), SpaceError> { + sqlx::query("DELETE FROM space_record_blobs WHERE space=$1 AND author=$2 AND collection=$3 AND rkey=$4") + .bind(space.to_string()).bind(author.as_str()).bind(write.collection.as_str()).bind(write.key.as_str()).execute(&mut *tx).await?; + let Some(record) = &write.record else { + return Ok(()); + }; + let mut blobs = Vec::new(); + collect_blob_objects(&record.value, &mut blobs); + for blob in blobs { + let cid = blob + .get("ref") + .and_then(|value| value.get("$link")) + .and_then(Value::as_str) + .and_then(|value| CidLink::new(value).ok()) + .ok_or_else(|| SpaceError::Invalid("Invalid blob reference".into()))?; + let found = sqlx::query( + r#"SELECT b.mime_type,b.size_bytes FROM blobs b JOIN users u ON u.did=$2 + WHERE b.cid=$1 AND b.takedown_ref IS NULL AND + (EXISTS(SELECT 1 FROM blob_upload_claims c WHERE c.cid=b.cid AND c.author=$2) + OR (b.created_by_user=u.id AND b.legacy_public))"#, + ) + .bind(cid.as_str()) + .bind(author.as_str()) + .fetch_optional(&mut *tx) + .await?; + let found = found.ok_or(SpaceError::Api(crate::api::error::ApiError::BlobNotFound( + None, + )))?; + if blob.get("mimeType").and_then(Value::as_str) + != Some(found.get::("mime_type").as_str()) + || blob.get("size").and_then(Value::as_i64) + != Some(found.get::("size_bytes")) + { + return Err(SpaceError::Invalid( + "Blob metadata does not match the uploaded bytes".into(), + )); + } + sqlx::query("INSERT INTO space_record_blobs(space,author,collection,rkey,cid) VALUES($1,$2,$3,$4,$5) ON CONFLICT DO NOTHING") + .bind(space.to_string()).bind(author.as_str()).bind(write.collection.as_str()).bind(write.key.as_str()).bind(cid.as_str()).execute(&mut *tx).await?; + } + Ok(()) + } + + pub async fn has_blob( + &self, + space: &SpaceRef, + author: &Did, + cid: &CidLink, + ) -> Result { + Ok(sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM space_record_blobs r JOIN blobs b ON b.cid=r.cid JOIN space_records record ON (record.space,record.author,record.collection,record.rkey)=(r.space,r.author,r.collection,r.rkey) WHERE r.space=$1 AND r.author=$2 AND r.cid=$3 AND b.takedown_ref IS NULL)") + .bind(space.to_string()).bind(author.as_str()).bind(cid.as_str()).fetch_one(&self.pool).await?) + } + + pub async fn blobs( + &self, + space: &SpaceRef, + author: &Did, + cursor: Option<&str>, + limit: i64, + since: Option<&str>, + ) -> Result { + self.state(space, author).await?; + let mut cids: Vec = sqlx::query_scalar("SELECT DISTINCT r.cid FROM space_record_blobs r JOIN blobs b ON b.cid=r.cid JOIN space_records record ON (record.space,record.author,record.collection,record.rkey)=(r.space,r.author,r.collection,r.rkey) WHERE r.space=$1 AND r.author=$2 AND b.takedown_ref IS NULL AND ($3::text IS NULL OR r.cid>$3) AND ($5::text IS NULL OR record.rev>$5) ORDER BY r.cid LIMIT $4") + .bind(space.to_string()).bind(author.as_str()).bind(cursor).bind(limit+1).bind(since).fetch_all(&self.pool).await?; + let more = cids.len() > limit as usize; + cids.truncate(limit as usize); + let cursor = cids.last().cloned(); + let mut out = json!({"cids":cids}); + if more { + out["cursor"] = json!(cursor); + } + Ok(out) + } +} + +fn collect_blob_objects<'a>(value: &'a Value, out: &mut Vec<&'a Value>) { + match value { + Value::Object(object) if object.get("$type").and_then(Value::as_str) == Some("blob") => { + out.push(value) + } + Value::Object(object) => object + .values() + .for_each(|value| collect_blob_objects(value, out)), + Value::Array(values) => values + .iter() + .for_each(|value| collect_blob_objects(value, out)), + _ => {} + } +} diff --git a/crates/tranquil-pds/src/spaces/store/notifications.rs b/crates/tranquil-pds/src/spaces/store/notifications.rs index 54f51dd9..06302331 100644 --- a/crates/tranquil-pds/src/spaces/store/notifications.rs +++ b/crates/tranquil-pds/src/spaces/store/notifications.rs @@ -162,6 +162,13 @@ impl SpaceStore { .bind(&space_rev) .execute(&mut *tx) .await?; + let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM space_writers WHERE space=$1") + .bind(space.to_string()) + .fetch_one(&mut *tx) + .await?; + if count > i64::from(tranquil_config::get().server.spaces_max_members_per_space) { + return Err(SpaceError::Invalid("Space writer quota exceeded".into())); + } let mut body = json!({"space":space,"repo":author,"repoRev":rev,"hash":{"$bytes":STANDARD.encode(hash)},"spaceRev":space_rev}); if let Some(previous) = previous { body["prevSpaceRev"] = json!(previous); @@ -235,6 +242,21 @@ impl SpaceStore { .bind(expires_at) .execute(&mut *tx) .await?; + sqlx::query("DELETE FROM space_notify_subscriptions WHERE space=$1 AND expires_at<=$2") + .bind(space.to_string()) + .bind(chrono::Utc::now().timestamp()) + .execute(&mut *tx) + .await?; + let count: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM space_notify_subscriptions WHERE space=$1") + .bind(space.to_string()) + .fetch_one(&mut *tx) + .await?; + if count > i64::from(tranquil_config::get().server.spaces_max_syncers_per_space) { + return Err(SpaceError::Invalid( + "Space notification subscriber quota exceeded".into(), + )); + } tx.commit().await?; Ok(()) } diff --git a/crates/tranquil-pds/src/spaces/store/read.rs b/crates/tranquil-pds/src/spaces/store/read.rs index d8f8acda..f55bb7d5 100644 --- a/crates/tranquil-pds/src/spaces/store/read.rs +++ b/crates/tranquil-pds/src/spaces/store/read.rs @@ -1,5 +1,4 @@ use super::*; -use crate::api::error::ApiError; use base64::{Engine, engine::general_purpose::STANDARD_NO_PAD}; pub struct RecordPage { pub collection: Option, @@ -19,14 +18,10 @@ pub struct RepoOps { pub value: Value, pub more: bool, } -pub struct ExportRecord { - pub path: String, - pub cid: cid::Cid, - pub data: Vec, -} pub struct RepoExport { pub state: RepoState, - pub records: Vec, + pub index: std::collections::BTreeMap, + pub blocks: tokio::fs::File, } impl SpaceStore { pub async fn records( @@ -116,12 +111,27 @@ impl SpaceStore { .execute(&mut *tx) .await?; let repo = - sqlx::query("SELECT rev,hash_state FROM space_repos WHERE space=$1 AND author=$2") + sqlx::query("SELECT rev,hash_state,ops_floor_rev,ops_floor_seq FROM space_repos WHERE space=$1 AND author=$2") .bind(space.to_string()) .bind(author.as_str()) .fetch_optional(&mut *tx) .await? .ok_or(SpaceError::RepoNotFound)?; + let expired = if let Some(cursor) = cursor { + repo.get::, _>("ops_floor_seq") + .is_some_and(|floor| cursor < floor) + } else if let Some(since) = &options.since { + repo.get::, _>("ops_floor_rev") + .is_some_and(|floor| since.as_str() < floor.as_str()) + } else { + false + }; + if expired { + // SPECAMB: proposal 0016 requires full recovery for a pruned position, + // but the pinned listRepoOps lexicon names no error. Use a stable + // machine-readable code instead of a silently incomplete delta. + return Err(SpaceError::RepoHistoryUnavailable); + } let query = r#" SELECT o.seq,o.rev,o.collection,o.rkey,o.cid,o.prev,r.value FROM space_ops o @@ -182,50 +192,44 @@ impl SpaceStore { .fetch_optional(&mut *tx) .await? .ok_or(SpaceError::RepoNotFound)?; - // Bound memory on this experimental export path; pagination remains available for larger repos. - let total:i64=sqlx::query_scalar("SELECT COALESCE(SUM(octet_length(data)),0)::bigint FROM space_records WHERE space=$1 AND author=$2") - .bind(space.to_string()) - .bind(author.as_str()) - .fetch_one(&mut *tx).await?; - if total > 32 * 1024 * 1024 { - return Err(SpaceError::Api(ApiError::PayloadTooLarge( - "Experimental Spaces export is limited to 32 MiB".into(), - ))); + use futures::TryStreamExt; + use tokio::io::{AsyncSeekExt, AsyncWriteExt}; + let file = tokio::task::spawn_blocking(tempfile::tempfile) + .await + .map_err(|_| SpaceError::internal())? + .map_err(|_| SpaceError::internal())?; + let mut blocks = tokio::fs::File::from_std(file); + let mut index = std::collections::BTreeMap::new(); + let mut rows = sqlx::query( + r#"SELECT collection,rkey,cid,data FROM space_records WHERE space=$1 AND author=$2 ORDER BY octet_length(collection || '/' || rkey),(collection || '/' || rkey) COLLATE "C""#, + ) + .bind(space.to_string()) + .bind(author.as_str()) + .fetch(&mut *tx); + while let Some(row) = rows.try_next().await? { + let cid: cid::Cid = row + .get::("cid") + .parse() + .map_err(|_| SpaceError::internal())?; + let path = format!( + "{}/{}", + row.get::("collection"), + row.get::("rkey") + ); + index.insert(path, ipld_core::ipld::Ipld::Link(cid)); + let data: Vec = row.get("data"); + blocks + .write_all(&crate::sync::car::encode_car_block(&cid, &data)) + .await + .map_err(|_| SpaceError::internal())?; } - let query = r#" - SELECT collection,rkey,cid,data - FROM space_records - WHERE space=$1 - AND author=$2 - ORDER BY octet_length(collection || '/' || rkey),(collection || '/' || rkey) COLLATE "C" - "#; - let records = sqlx::query(query) - .bind(space.to_string()) - .bind(author.as_str()) - .fetch_all(&mut *tx) - .await?; + drop(rows); tx.commit().await?; - let records = records - .into_iter() - .map(|row| { - let cid = row - .get::("cid") - .parse() - .map_err(|_| SpaceError::internal())?; - Ok(ExportRecord { - path: format!( - "{}/{}", - row.get::("collection"), - row.get::("rkey") - ), - cid, - data: row.get("data"), - }) - }) - .collect::, SpaceError>>()?; + blocks.rewind().await.map_err(|_| SpaceError::internal())?; Ok(RepoExport { state: RepoState::from_row(&repo), - records, + index, + blocks, }) } pub async fn writers( diff --git a/crates/tranquil-pds/src/state.rs b/crates/tranquil-pds/src/state.rs index 40f5b464..d9a61d02 100644 --- a/crates/tranquil-pds/src/state.rs +++ b/crates/tranquil-pds/src/state.rs @@ -58,6 +58,8 @@ pub struct AppState { pub signal_store_provider: Option>, pub eventlog_segments_dir: Option, pub repo_export_semaphore: Arc, + pub blob_publication_semaphore: Arc, + pub spaces_export_semaphore: Arc, } #[derive(Debug, Clone, Copy)] @@ -79,6 +81,9 @@ impl RateLimitParams { #[derive(Debug, Clone, Copy)] pub enum RateLimitKind { + SpacesRead, + SpacesWrite, + SpacesCredential, Login, AccountCreation, PasswordReset, @@ -104,6 +109,9 @@ pub enum RateLimitKind { impl RateLimitKind { const fn key_prefix(&self) -> &'static str { match self { + Self::SpacesRead => "spaces_read", + Self::SpacesWrite => "spaces_write", + Self::SpacesCredential => "spaces_credential", Self::Login => "login", Self::AccountCreation => "account_creation", Self::PasswordReset => "password_reset", @@ -129,6 +137,18 @@ impl RateLimitKind { pub const fn params(&self) -> RateLimitParams { match self { + Self::SpacesRead => RateLimitParams { + limit: 3000, + window_ms: 60_000, + }, + Self::SpacesWrite => RateLimitParams { + limit: 600, + window_ms: 60_000, + }, + Self::SpacesCredential => RateLimitParams { + limit: 120, + window_ms: 60_000, + }, Self::Login => RateLimitParams { limit: 10, window_ms: 60_000, @@ -403,7 +423,13 @@ impl AppState { .expect("Failed to create WebAuthn config at startup"), ); + let publication_slots = repos + .pool + .as_ref() + .map(|pool| (pool.options().get_max_connections() as usize / 4).clamp(1, 4)) + .unwrap_or(4); Self { + blob_publication_semaphore: Arc::new(tokio::sync::Semaphore::new(publication_slots)), repos, block_store, blob_store, @@ -423,6 +449,9 @@ impl AppState { signal_sender: None, signal_store_provider, eventlog_segments_dir, + spaces_export_semaphore: Arc::new(tokio::sync::Semaphore::new( + cfg.firehose.max_concurrent_repo_exports, + )), repo_export_semaphore: Arc::new(tokio::sync::Semaphore::new( cfg.firehose.max_concurrent_repo_exports, )), @@ -467,6 +496,9 @@ impl AppState { let limiter_name = kind.key_prefix(); let limiter = match kind { + RateLimitKind::SpacesRead => &self.rate_limiters.spaces_read, + RateLimitKind::SpacesWrite => &self.rate_limiters.spaces_write, + RateLimitKind::SpacesCredential => &self.rate_limiters.spaces_credential, RateLimitKind::Login => &self.rate_limiters.login, RateLimitKind::AccountCreation => &self.rate_limiters.account_creation, RateLimitKind::PasswordReset => &self.rate_limiters.password_reset, diff --git a/crates/tranquil-pds/src/storage/blob_access.rs b/crates/tranquil-pds/src/storage/blob_access.rs new file mode 100644 index 00000000..6d1128db --- /dev/null +++ b/crates/tranquil-pds/src/storage/blob_access.rs @@ -0,0 +1,71 @@ +use crate::spaces::SpaceError; +use tranquil_types::{CidLink, Did}; + +pub struct BlobAccess { + pool: sqlx::PgPool, +} +impl BlobAccess { + pub fn new(pool: sqlx::PgPool) -> Self { + Self { pool } + } + /// Existing uploads predate private storage. New uploads are visible only + /// through a live public record of the requested account. + pub async fn public_blob(&self, did: &Did, cid: &CidLink) -> Result { + // SPECAMB: Proposal 0016 requires private blob authorization but does not + // define upgrades from a globally readable CID store. Preserve visibility + // of pre-upgrade bytes; only new uploads use record-bound publication. + Ok(sqlx::query_scalar(r#"SELECT EXISTS( + SELECT 1 FROM blobs b JOIN users u ON u.did=$1 + WHERE b.cid=$2 AND b.takedown_ref IS NULL AND ( + b.legacy_public + OR ((b.legacy_public OR b.created_by_user=u.id OR EXISTS(SELECT 1 FROM blob_upload_claims c WHERE c.cid=b.cid AND c.author=u.did)) AND EXISTS(SELECT 1 FROM record_blobs rb JOIN records r + ON r.repo_id=rb.repo_id + AND rb.record_uri='at://' || u.did || '/' || r.collection || '/' || r.rkey + WHERE rb.repo_id=u.id AND rb.blob_cid=b.cid AND rb.record_cid=r.record_cid AND r.takedown_ref IS NULL)) + ))"#).bind(did.as_str()).bind(cid.as_str()).fetch_one(&self.pool).await?) + } + + pub async fn public_blobs( + &self, + did: &Did, + cursor: &str, + limit: i64, + since: Option<&str>, + ) -> Result, SpaceError> { + let rows: Vec = sqlx::query_scalar(r#"SELECT b.cid FROM blobs b JOIN users u ON u.did=$1 + WHERE b.cid>$2 AND b.takedown_ref IS NULL AND ( + (b.legacy_public AND ((b.created_by_user=u.id AND $4::text IS NULL) OR ($4::text IS NOT NULL AND EXISTS(SELECT 1 FROM repo_seq s WHERE s.did=u.did AND s.rev>$4 AND b.cid=ANY(s.blobs))))) + OR ((b.legacy_public OR b.created_by_user=u.id OR EXISTS(SELECT 1 FROM blob_upload_claims c WHERE c.cid=b.cid AND c.author=u.did)) AND EXISTS(SELECT 1 FROM record_blobs rb JOIN records r + ON r.repo_id=rb.repo_id AND rb.record_uri='at://' || u.did || '/' || r.collection || '/' || r.rkey + WHERE rb.repo_id=u.id AND rb.blob_cid=b.cid AND rb.record_cid=r.record_cid + AND r.takedown_ref IS NULL AND ($4::text IS NULL OR r.repo_rev>$4))) + ) ORDER BY b.cid LIMIT $3"#) + .bind(did.as_str()).bind(cursor).bind(limit).bind(since).fetch_all(&self.pool).await?; + rows.into_iter() + .map(|cid| CidLink::new(&cid).map_err(|_| SpaceError::internal())) + .collect() + } + + /// Knowing the hash of another account's private upload is not publication permission. + pub async fn check_blob_ownership( + &self, + did: &Did, + cids: &[CidLink], + ) -> Result<(), SpaceError> { + for cid in cids { + let permitted: bool = sqlx::query_scalar( + r#"SELECT NOT EXISTS(SELECT 1 FROM blobs WHERE cid=$1) + OR EXISTS(SELECT 1 FROM blobs b JOIN users u ON u.did=$2 WHERE b.cid=$1 AND (b.legacy_public OR b.created_by_user=u.id)) + OR EXISTS(SELECT 1 FROM blob_upload_claims WHERE cid=$1 AND author=$2)"#, + ) + .bind(cid.as_str()) + .bind(did.as_str()) + .fetch_one(&self.pool) + .await?; + if !permitted { + return Err(SpaceError::Forbidden); + } + } + Ok(()) + } +} diff --git a/crates/tranquil-pds/src/storage/mod.rs b/crates/tranquil-pds/src/storage/mod.rs index c1057610..46c12213 100644 --- a/crates/tranquil-pds/src/storage/mod.rs +++ b/crates/tranquil-pds/src/storage/mod.rs @@ -1,6 +1,220 @@ +mod blob_access; +pub use blob_access::BlobAccess; pub use tranquil_storage::{ BlobStorage, FilesystemBlobStorage, StorageError, StreamUploadResult, create_blob_storage, }; #[cfg(feature = "s3")] pub use tranquil_storage::S3BlobStorage; + +/// Serialize content-addressed object publication and physical reclamation +/// across PDS processes. Dropping the transaction also releases the lock. +pub async fn lock_blob_object( + pool: &sqlx::PgPool, + key: &str, +) -> Result, sqlx::Error> { + let mut tx = pool.begin().await?; + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended('blob:' || $1, 0))") + .bind(key) + .execute(&mut *tx) + .await?; + Ok(tx) +} + +pub async fn lock_blob_upload( + pool: &sqlx::PgPool, + key: &str, + user_id: uuid::Uuid, +) -> Result, sqlx::Error> { + let mut tx = pool.begin().await?; + sqlx::query("SET LOCAL lock_timeout='2s'") + .execute(&mut *tx) + .await?; + sqlx::query("SELECT id FROM users WHERE id=$1 FOR NO KEY UPDATE") + .bind(user_id) + .fetch_one(&mut *tx) + .await?; + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended('blob:' || $1, 0))") + .bind(key) + .execute(&mut *tx) + .await?; + Ok(tx) +} + +pub struct UploadedBlob<'a> { + pub cid: &'a tranquil_types::CidLink, + pub mime_type: &'a str, + pub size_bytes: i64, + pub user_id: uuid::Uuid, + pub author: &'a tranquil_types::Did, + pub storage_key: &'a str, +} + +pub async fn register_blob_upload( + tx: &mut sqlx::PgConnection, + blob: UploadedBlob<'_>, +) -> Result<(), crate::api::ApiError> { + use crate::api::ApiError; + use sqlx::Row; + sqlx::query("INSERT INTO blobs(cid,mime_type,size_bytes,created_by_user,storage_key) VALUES($1,$2,$3,$4,$5) ON CONFLICT DO NOTHING") + .bind(blob.cid.as_str()).bind(blob.mime_type).bind(blob.size_bytes).bind(blob.user_id).bind(blob.storage_key) + .execute(&mut *tx).await.map_err(|_| ApiError::InternalError(None))?; + sqlx::query("INSERT INTO blob_upload_claims(cid,author) VALUES($1,$2) ON CONFLICT(cid,author) DO UPDATE SET created_at=NOW()") + .bind(blob.cid.as_str()) + .bind(blob.author.as_str()) + .execute(&mut *tx) + .await.map_err(|_| ApiError::InternalError(None))?; + let usage = sqlx::query(r#"SELECT COUNT(*) AS count,COALESCE(SUM(b.size_bytes),0)::bigint AS bytes + FROM blob_upload_claims c JOIN blobs b ON b.cid=c.cid + WHERE c.author=$1 AND NOT EXISTS(SELECT 1 FROM blob_record_references r WHERE r.author=c.author AND r.cid=c.cid)"#) + .bind(blob.author.as_str()).fetch_one(&mut *tx).await.map_err(|_| ApiError::InternalError(None))?; + let config = &tranquil_config::get().server; + if usage.get::("count") > i64::from(config.max_pending_blobs_per_account) + || usage.get::("bytes") as u64 > config.max_pending_blob_bytes_per_account + { + return Err(ApiError::InvalidRequest( + "Unattached blob upload quota exceeded".into(), + )); + } + Ok(()) +} + +/// A bounded producer owns the temporary file and permit. Its deadline runs +/// even when the HTTP peer stops polling, so stalled downloads release both. +pub fn stream_export( + mut file: tokio::fs::File, + prefix: Vec, + permit: tokio::sync::OwnedSemaphorePermit, + timeout: std::time::Duration, +) -> tranquil_storage::BlobDownloadStream { + let deadline = tokio::time::Instant::now() + timeout; + let (sender, receiver) = tokio::sync::mpsc::channel(1); + tokio::spawn(async move { + let _permit = permit; + let _ = tokio::time::timeout_at(deadline, async move { + use tokio::io::AsyncReadExt; + for chunk in prefix.chunks(64 * 1024) { + if sender + .send(Ok(bytes::Bytes::copy_from_slice(chunk))) + .await + .is_err() + { + return; + } + } + drop(prefix); + loop { + let mut buffer = vec![0; 64 * 1024]; + match file.read(&mut buffer).await { + Ok(0) => break, + Ok(n) => { + buffer.truncate(n); + if sender.send(Ok(bytes::Bytes::from(buffer))).await.is_err() { + break; + } + } + Err(error) => { + let _ = sender.send(Err(error)).await; + break; + } + } + } + }) + .await; + }); + Box::pin(futures::stream::try_unfold( + receiver, + move |mut receiver| async move { + match tokio::time::timeout_at(deadline, receiver.recv()).await { + Ok(Some(Ok(bytes))) => Ok(Some((bytes, receiver))), + Ok(Some(Err(error))) => Err(error), + Ok(None) => Ok(None), + Err(_) => Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "Spaces export deadline exceeded", + )), + } + }, + )) +} + +/// Reclaim unattached uploads after 24 hours, serialized against publication. +/// Referenced claims survive, including public references bound to record CIDs. +pub async fn reclaim_expired_uploads( + pool: &sqlx::PgPool, + store: &dyn BlobStorage, +) -> Result<(), sqlx::Error> { + let candidates: Vec = sqlx::query_scalar(r#"SELECT b.cid FROM blobs b + WHERE b.created_at Result<(), sqlx::Error> { + // A durable queue covers every metadata deletion path, including admin + // deletion. Re-uploaded CIDs are checked under the same publication lock. + let deleted: Vec<(String, String)> = sqlx::query_as( + "SELECT cid,storage_key FROM blob_deletion_queue ORDER BY created_at LIMIT 100", + ) + .fetch_all(pool) + .await?; + for (cid, key) in deleted { + let mut tx = lock_blob_object(pool, &cid).await?; + let restored: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM blobs WHERE cid=$1)") + .bind(&cid) + .fetch_one(&mut *tx) + .await?; + if !restored && let Err(error) = store.delete(&key).await { + tracing::warn!(%error, "Queued blob deletion failed; retained for retry"); + continue; + } + sqlx::query("DELETE FROM blob_deletion_queue WHERE cid=$1") + .bind(&cid) + .execute(&mut *tx) + .await?; + tx.commit().await?; + } + Ok(()) +} + +#[cfg(test)] +mod export_tests { + use super::*; + #[tokio::test] + async fn stalled_export_releases_its_permit_without_being_polled() { + let file = tempfile::tempfile().unwrap(); + file.set_len(256 * 1024).unwrap(); + let semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(1)); + let permit = semaphore.clone().acquire_owned().await.unwrap(); + let _unpolled = stream_export( + tokio::fs::File::from_std(file), + vec![1, 2, 3], + permit, + std::time::Duration::from_millis(25), + ); + let recovered = + tokio::time::timeout(std::time::Duration::from_secs(1), semaphore.acquire()).await; + assert!(recovered.is_ok()); + } +} diff --git a/crates/tranquil-pds/tests/interop/README.md b/crates/tranquil-pds/tests/interop/README.md index f83cb8eb..eb292e51 100644 --- a/crates/tranquil-pds/tests/interop/README.md +++ b/crates/tranquil-pds/tests/interop/README.md @@ -18,7 +18,7 @@ From the fork root, choose a new temporary directory: reference_parent=$(mktemp -d /tmp/spaces252-reference-build.XXXXXX) sh crates/tranquil-pds/tests/interop/prepare-reference.sh "$reference_parent/reference" ATPROTO_REFERENCE_ROOT="$reference_parent/reference" \ - SQLX_OFFLINE=true ENABLE_ATPROTO_SPACES=true \ + SQLX_OFFLINE=true \ cargo test --locked -p tranquil-pds --test spaces \ interop::reference_pds_interoperability -- --ignored --exact --nocapture ``` diff --git a/crates/tranquil-pds/tests/spaces.rs b/crates/tranquil-pds/tests/spaces.rs index 31b837fd..79237b8b 100644 --- a/crates/tranquil-pds/tests/spaces.rs +++ b/crates/tranquil-pds/tests/spaces.rs @@ -168,7 +168,7 @@ async fn create_user_and_oauth_session( (session, mock_client) } -const FULL: &str = "atproto space:com.example.settings?collection=com.example.setting&manage=create&manage=update&manage=delete"; +const FULL: &str = "atproto blob:*/* repo:com.example.setting space:com.example.settings?collection=com.example.setting&manage=create&manage=update&manage=delete"; async fn post(session: &OAuthSession, method: &str, body: Value) -> Value { let response = client() @@ -350,6 +350,7 @@ async fn spaces_oauth_crud_is_private_durable_and_transactional() { .await .unwrap(); assert!(!result.status().is_success()); + assert_private_blob_publication(&owner).await; } #[tokio::test] @@ -366,7 +367,7 @@ async fn space_scopes_do_not_leak_across_authors_collections_or_operation_kinds( let (other, _m2) = create_user_and_oauth_session( "space-bob", "https://example.com/callback", - "atproto space:*?authority=*&collection=*", + "atproto blob:*/* repo:com.example.setting space:*?authority=*&collection=*", ) .await; let res = client() @@ -445,6 +446,7 @@ async fn space_scopes_do_not_leak_across_authors_collections_or_operation_kinds( .unwrap() .is_empty() ); + assert_shared_blob_survives_account_deletion(&owner, &other, &legacy, &uri).await; } fn signed_headers( @@ -797,6 +799,7 @@ async fn space_sync_exports_private_blocks_and_paginated_operations() { commit["hash"], ipld_core::ipld::Ipld::Bytes(Sha256::digest([0u8; 2048]).to_vec()) ); + assert_large_export_and_retention(&owner, &uri).await; } #[tokio::test] @@ -928,6 +931,34 @@ async fn space_writes_tracking_notifications_and_delivery_recovery() { own["value"]["text"], "private copy", "authority deletion cannot erase another author's repo" ); + for key in std::iter::once("self".to_string()) + .chain((0..8).map(|i| format!("key-{i}"))) + .chain(std::iter::once("public-writer".to_string())) + { + post( + &writer, + "com.atproto.space.deleteRecord", + json!({"space":uri,"repo":writer.did,"collection":"com.example.setting","rkey":key}), + ) + .await; + } + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(2) + .connect_with((*common::get_test_db_pool().await.connect_options()).clone()) + .await + .unwrap(); + let retained: bool = + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM space_repos WHERE space=$1 AND author=$2)") + .bind(&uri) + .bind(&writer.did) + .fetch_one(&pool) + .await + .unwrap(); + assert!( + !retained, + "deleting the final record releases the tombstoned repo quota" + ); + pool.close().await; assert_eq!( space(&owner).await, uri, @@ -1251,3 +1282,571 @@ mod interop; #[path = "interop/notifications.rs"] mod notifications; + +async fn assert_private_blob_publication(owner: &OAuthSession) { + let uri = space(&owner).await; + let bytes = format!("private bytes {}", uuid::Uuid::new_v4()); + let response = client() + .post(format!( + "{}/xrpc/com.atproto.repo.uploadBlob", + base_url().await + )) + .bearer_auth(&owner.access_token) + .header("Content-Type", "text/plain") + .body(bytes.clone()) + .send() + .await + .unwrap(); + let status = response.status(); + let response: Value = response.json().await.unwrap(); + assert_eq!(status, StatusCode::OK, "{response}"); + let blob = &response["blob"]; + let cid = blob["ref"]["$link"].as_str().unwrap(); + let public_url = format!("{}/xrpc/com.atproto.sync.getBlob", base_url().await); + let public_query = [("did", owner.did.as_str()), ("cid", cid)]; + assert!( + !client() + .get(&public_url) + .query(&public_query) + .send() + .await + .unwrap() + .status() + .is_success() + ); + let mut input = record(&uri, &owner.did, "attachment", "private"); + input["record"]["attachment"] = blob.clone(); + // Duplicate references in one record share one stored blob. + input["record"]["duplicate"] = blob.clone(); + post(&owner, "com.atproto.space.putRecord", input.clone()).await; + let private_url = format!("{}/xrpc/com.atproto.space.getBlob", base_url().await); + let private_query = [ + ("space", uri.as_str()), + ("repo", owner.did.as_str()), + ("cid", cid), + ]; + assert!( + !client() + .get(&private_url) + .query(&private_query) + .send() + .await + .unwrap() + .status() + .is_success() + ); + let downloaded = client() + .get(&private_url) + .query(&private_query) + .bearer_auth(&owner.access_token) + .send() + .await + .unwrap(); + assert_eq!(downloaded.status(), StatusCode::OK); + assert_eq!(downloaded.headers()["cache-control"], "private, no-store"); + assert_eq!( + downloaded.headers()["content-disposition"], + format!("attachment; filename=\"{cid}\"") + ); + assert_eq!(downloaded.text().await.unwrap(), bytes); + assert!( + !client() + .get(&public_url) + .query(&public_query) + .send() + .await + .unwrap() + .status() + .is_success() + ); + assert_pending_blob_limits_and_cleanup(owner, &uri, cid).await; + let listed = get(&owner, "com.atproto.space.listBlobs", &private_query[..2]).await; + assert_eq!(listed["cids"], json!([cid])); + let head = get( + owner, + "com.atproto.space.getLatestCommit", + &private_query[..2], + ) + .await; + let rev = head["commit"]["rev"].as_str().unwrap(); + let since_query = [ + ("space", uri.as_str()), + ("repo", owner.did.as_str()), + ("since", rev), + ]; + assert_eq!( + get(owner, "com.atproto.space.listBlobs", &since_query).await["cids"], + json!([]) + ); + input["record"]["text"] = json!("new revision, same attachment"); + post(owner, "com.atproto.space.putRecord", input.clone()).await; + assert_eq!( + get(owner, "com.atproto.space.listBlobs", &since_query).await["cids"], + json!([cid]) + ); + input.as_object_mut().unwrap().remove("space"); + post(&owner, "com.atproto.repo.putRecord", input.clone()).await; + // Identical bytes under a second key, then delete/re-create, must still + // include the committed block needed for atomic blob-reference indexing. + let mut duplicate = input.clone(); + duplicate["rkey"] = json!("duplicate-content"); + for _ in 0..2 { + post(owner, "com.atproto.repo.putRecord", duplicate.clone()).await; + post( + owner, + "com.atproto.repo.deleteRecord", + json!({"repo":owner.did,"collection":"com.example.setting","rkey":"duplicate-content"}), + ) + .await; + } + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(2) + .connect_with((*common::get_test_db_pool().await.connect_options()).clone()) + .await + .unwrap(); + // Re-import indexing must rebind a pre-existing row to the live record. + sqlx::query("UPDATE record_blobs SET record_cid='stale-import-version' WHERE blob_cid=$1") + .bind(cid) + .execute(&pool) + .await + .unwrap(); + let repos = tranquil_db::PostgresRepositories::new(pool.clone()); + let user_id = repos + .user + .get_id_by_did(&owner.did.parse().unwrap()) + .await + .unwrap() + .unwrap(); + repos + .blob + .insert_record_blobs( + user_id, + &[format!("at://{}/com.example.setting/attachment", owner.did) + .parse() + .unwrap()], + &[cid.parse().unwrap()], + ) + .await + .unwrap(); + pool.close().await; + let published = client() + .get(&public_url) + .query(&public_query) + .send() + .await + .unwrap(); + assert_eq!(published.status(), StatusCode::OK); + assert_eq!(published.text().await.unwrap(), bytes); + post( + &owner, + "com.atproto.repo.deleteRecord", + json!({"repo":owner.did,"collection":"com.example.setting","rkey":"attachment"}), + ) + .await; + assert!( + !client() + .get(&public_url) + .query(&public_query) + .send() + .await + .unwrap() + .status() + .is_success() + ); + assert_eq!( + client() + .get(&private_url) + .query(&private_query) + .bearer_auth(&owner.access_token) + .send() + .await + .unwrap() + .status(), + StatusCode::OK + ); +} + +async fn upload_test_blob(owner: &OAuthSession, bytes: &str) -> Value { + let response = client() + .post(format!( + "{}/xrpc/com.atproto.repo.uploadBlob", + base_url().await + )) + .bearer_auth(&owner.access_token) + .header("content-type", "text/plain") + .body(bytes.to_owned()) + .send() + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + response.json::().await.unwrap()["blob"].clone() +} + +async fn assert_shared_blob_survives_account_deletion( + owner: &OAuthSession, + other: &OAuthSession, + outsider: &OAuthSession, + uri: &str, +) { + let bytes = format!("shared {}", uuid::Uuid::new_v4()); + let blob = upload_test_blob(owner, &bytes).await; + let mut input = record(uri, &other.did, "shared", "attachment"); + input["record"]["attachment"] = blob.clone(); + let rejected = client() + .post(format!( + "{}/xrpc/com.atproto.space.putRecord", + base_url().await + )) + .bearer_auth(&other.access_token) + .json(&input) + .send() + .await + .unwrap(); + assert!(!rejected.status().is_success()); + let mut public = input.clone(); + public.as_object_mut().unwrap().remove("space"); + let rejected = client() + .post(format!( + "{}/xrpc/com.atproto.repo.putRecord", + base_url().await + )) + .bearer_auth(&other.access_token) + .json(&public) + .send() + .await + .unwrap(); + assert_eq!(rejected.status(), StatusCode::FORBIDDEN); + assert_eq!(upload_test_blob(other, &bytes).await, blob); + post(other, "com.atproto.space.putRecord", input).await; + let pool = &sqlx::postgres::PgPoolOptions::new() + .max_connections(2) + .connect_with((*common::get_test_db_pool().await.connect_options()).clone()) + .await + .unwrap(); + // An imported public reference alone is not proof of upload ownership. + // It must not acquire a private blob when the real uploader is deleted. + let secret = upload_test_blob(owner, &format!("unshared {}", uuid::Uuid::new_v4())).await; + let secret_cid = secret["ref"]["$link"].as_str().unwrap(); + let public = post(outsider, "com.atproto.repo.putRecord", json!({"repo":outsider.did,"collection":"com.example.setting","rkey":"fake","validate":false,"record":{"$type":"com.example.setting","text":"no upload claim"}})).await; + sqlx::query("INSERT INTO record_blobs(repo_id,record_uri,blob_cid,record_cid) SELECT id,$2,$3,$4 FROM users WHERE did=$1") + .bind(&outsider.did).bind(public["uri"].as_str().unwrap()).bind(secret_cid).bind(public["cid"].as_str().unwrap()).execute(pool).await.unwrap(); + let repos = tranquil_db::PostgresRepositories::new(pool.clone()); + let did = owner.did.parse().unwrap(); + let id = repos.user.get_id_by_did(&did).await.unwrap().unwrap(); + assert_eq!( + repos.blob.get_blob_storage_keys_by_user(id).await.unwrap(), + vec![secret_cid.to_string()] + ); + repos + .user + .delete_account_with_firehose(id, &did) + .await + .unwrap(); + let queued: bool = + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM blob_deletion_queue WHERE cid=$1)") + .bind(secret_cid) + .fetch_one(pool) + .await + .unwrap(); + assert!(queued, "account deletion must retain physical cleanup work"); + let state = common::get_test_app_state().await; + tranquil_pds::storage::reclaim_expired_uploads(pool, state.blob_store.as_ref()) + .await + .unwrap(); + assert!(state.blob_store.get_bytes(secret_cid).await.is_err()); + let downloaded = client() + .get(format!( + "{}/xrpc/com.atproto.space.getBlob", + base_url().await + )) + .bearer_auth(&other.access_token) + .query(&[ + ("space", uri), + ("repo", other.did.as_str()), + ("cid", blob["ref"]["$link"].as_str().unwrap()), + ]) + .send() + .await + .unwrap(); + assert_eq!(downloaded.status(), StatusCode::OK); + assert_eq!(downloaded.text().await.unwrap(), bytes); + let inaccessible = client() + .get(format!( + "{}/xrpc/com.atproto.sync.getBlob", + base_url().await + )) + .query(&[("did", outsider.did.as_str()), ("cid", secret_cid)]) + .send() + .await + .unwrap(); + assert_eq!(inaccessible.status(), StatusCode::NOT_FOUND); + // Model an old globally deduplicated upload referenced by another account, + // with no record_blobs indexing (the pre-upgrade commit path). + let old = upload_test_blob(other, &format!("legacy {}", uuid::Uuid::new_v4())).await; + let old_cid = old["ref"]["$link"].as_str().unwrap(); + sqlx::query("UPDATE blobs SET legacy_public=true WHERE cid=$1") + .bind(old_cid) + .execute(pool) + .await + .unwrap(); + post(outsider, "com.atproto.repo.putRecord", json!({"repo":outsider.did,"collection":"com.example.setting","rkey":"legacy","validate":false,"record":{"$type":"com.example.setting","attachment":old}})).await; + sqlx::query("DELETE FROM record_blobs WHERE blob_cid=$1") + .bind(old_cid) + .execute(pool) + .await + .unwrap(); + let legacy_read = client() + .get(format!( + "{}/xrpc/com.atproto.sync.getBlob", + base_url().await + )) + .query(&[("did", outsider.did.as_str()), ("cid", old_cid)]) + .send() + .await + .unwrap(); + assert_eq!(legacy_read.status(), StatusCode::OK); + let catchup = client() + .get(format!( + "{}/xrpc/com.atproto.sync.listBlobs", + base_url().await + )) + .query(&[("did", outsider.did.as_str()), ("since", "2222222222222")]) + .send() + .await + .unwrap(); + assert_eq!(catchup.status(), StatusCode::OK); + assert!( + catchup.json::().await.unwrap()["cids"] + .as_array() + .unwrap() + .contains(&json!(old_cid)) + ); + pool.close().await; +} + +async fn assert_large_export_and_retention(owner: &OAuthSession, uri: &str) { + let payload = "x".repeat(900_000); + for n in 0..38 { + post( + owner, + "com.atproto.space.putRecord", + record(uri, &owner.did, &format!("large-{n}"), &payload), + ) + .await; + } + let response = client() + .get(format!( + "{}/xrpc/com.atproto.space.getRepo", + base_url().await + )) + .bearer_auth(&owner.access_token) + .query(&[("space", uri), ("repo", owner.did.as_str())]) + .send() + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let bytes = response.bytes().await.unwrap(); + assert!(bytes.len() > 32 * 1024 * 1024); + let mut car = iroh_car::CarReader::new(std::io::Cursor::new(bytes)) + .await + .unwrap(); + let mut count = 0; + while let Some((cid, bytes)) = car.next_block().await.unwrap() { + assert_eq!(cid, jacquard_repo::mst::util::compute_cid(&bytes).unwrap()); + count += 1; + } + assert_eq!(count, 40); + let pool = &sqlx::postgres::PgPoolOptions::new() + .max_connections(2) + .connect_with((*common::get_test_db_pool().await.connect_options()).clone()) + .await + .unwrap(); + let (records, bytes, actual): (i64,i64,i64) = sqlx::query_as("SELECT record_count,record_bytes,(SELECT SUM(octet_length(data))::bigint FROM space_records WHERE space=$1 AND author=$2) FROM space_repos WHERE space=$1 AND author=$2") + .bind(uri).bind(&owner.did).fetch_one(pool).await.unwrap(); + assert_eq!(records, 38); + assert_eq!(bytes, actual); + sqlx::query("UPDATE space_repos SET record_bytes=$3 WHERE space=$1 AND author=$2") + .bind(uri) + .bind(&owner.did) + .bind( + tranquil_config::get() + .server + .spaces_max_record_bytes_per_account as i64, + ) + .execute(pool) + .await + .unwrap(); + let rejected = client() + .post(format!( + "{}/xrpc/com.atproto.space.putRecord", + base_url().await + )) + .bearer_auth(&owner.access_token) + .json(&record(uri, &owner.did, "over-quota", "blocked")) + .send() + .await + .unwrap(); + assert_eq!(rejected.status(), StatusCode::BAD_REQUEST); + sqlx::query("UPDATE space_repos SET record_bytes=$3 WHERE space=$1 AND author=$2") + .bind(uri) + .bind(&owner.did) + .bind(bytes) + .execute(pool) + .await + .unwrap(); + let mut tx = pool.begin().await.unwrap(); + tranquil_pds::spaces::store::SpaceStore::compact_operations( + &mut tx, + &uri.parse().unwrap(), + &owner.did.parse().unwrap(), + 2, + ) + .await + .unwrap(); + tx.commit().await.unwrap(); + let expired = client() + .get(format!( + "{}/xrpc/com.atproto.space.listRepoOps", + base_url().await + )) + .bearer_auth(&owner.access_token) + .query(&[ + ("space", uri), + ("repo", owner.did.as_str()), + ("cursor", "0"), + ]) + .send() + .await + .unwrap(); + assert_eq!(expired.status(), StatusCode::BAD_REQUEST); + assert!( + expired.json::().await.unwrap()["message"] + .as_str() + .unwrap() + .contains("getRepo") + ); + pool.close().await; +} + +async fn assert_pending_blob_limits_and_cleanup( + owner: &OAuthSession, + uri: &str, + attached_cid: &str, +) { + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(2) + .connect_with((*common::get_test_db_pool().await.connect_options()).clone()) + .await + .unwrap(); + let state = common::get_test_app_state().await; + let slots = state.blob_publication_semaphore.available_permits() as u32; + let busy = state + .blob_publication_semaphore + .acquire_many(slots) + .await + .unwrap(); + let rejected = client() + .post(format!( + "{}/xrpc/com.atproto.repo.uploadBlob", + base_url().await + )) + .bearer_auth(&owner.access_token) + .header("content-type", "text/plain") + .body("publication slots are occupied") + .send() + .await + .unwrap(); + assert_eq!(rejected.status(), StatusCode::SERVICE_UNAVAILABLE); + drop(busy); + let pending = upload_test_blob(owner, &format!("pending {}", uuid::Uuid::new_v4())).await; + let pending_cid = pending["ref"]["$link"].as_str().unwrap(); + let limit = tranquil_config::get() + .server + .max_pending_blob_bytes_per_account as i64; + sqlx::query("UPDATE blobs SET size_bytes=$2 WHERE cid=$1") + .bind(pending_cid) + .bind(limit) + .execute(&pool) + .await + .unwrap(); + let rejected = client() + .post(format!( + "{}/xrpc/com.atproto.repo.uploadBlob", + base_url().await + )) + .bearer_auth(&owner.access_token) + .header("content-type", "text/plain") + .body(format!("exceeds quota {}", uuid::Uuid::new_v4())) + .send() + .await + .unwrap(); + assert_eq!(rejected.status(), StatusCode::BAD_REQUEST); + assert!( + rejected.json::().await.unwrap()["message"] + .as_str() + .unwrap() + .contains("quota") + ); + let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM blob_upload_claims WHERE author=$1") + .bind(&owner.did) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(count, 2, "failed upload must roll back its claim"); + sqlx::query("UPDATE blobs SET size_bytes=$2 WHERE cid=$1") + .bind(pending_cid) + .bind(pending["size"].as_i64().unwrap()) + .execute(&pool) + .await + .unwrap(); + for cid in [pending_cid, attached_cid] { + sqlx::query("UPDATE blobs SET created_at=NOW()-INTERVAL '25 hours' WHERE cid=$1") + .bind(cid) + .execute(&pool) + .await + .unwrap(); + sqlx::query( + "UPDATE blob_upload_claims SET created_at=NOW()-INTERVAL '25 hours' WHERE cid=$1", + ) + .bind(cid) + .execute(&pool) + .await + .unwrap(); + } + let state = common::get_test_app_state().await; + tranquil_pds::storage::reclaim_expired_uploads(&pool, state.blob_store.as_ref()) + .await + .unwrap(); + let remaining: Vec = + sqlx::query_scalar("SELECT cid FROM blob_upload_claims WHERE author=$1") + .bind(&owner.did) + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!(remaining, vec![attached_cid.to_owned()]); + assert!(state.blob_store.get_bytes(pending_cid).await.is_err()); + assert!(state.blob_store.get_bytes(attached_cid).await.is_ok()); + let mut held = pool.begin().await.unwrap(); + sqlx::query("SELECT did FROM users WHERE did=$1 FOR NO KEY UPDATE") + .bind(&owner.did) + .execute(&mut *held) + .await + .unwrap(); + let blocked = tokio::time::timeout( + std::time::Duration::from_secs(5), + client() + .post(format!( + "{}/xrpc/com.atproto.space.putRecord", + base_url().await + )) + .bearer_auth(&owner.access_token) + .json(&record(uri, &owner.did, "lock-busy", "must retry")) + .send(), + ) + .await + .expect("account lock must not hold a pooled waiter indefinitely") + .unwrap(); + assert_eq!(blocked.status(), StatusCode::SERVICE_UNAVAILABLE); + held.rollback().await.unwrap(); + pool.close().await; +} diff --git a/crates/tranquil-pds/tests/sync_blob.rs b/crates/tranquil-pds/tests/sync_blob.rs index 5c1b9d58..1b5ced26 100644 --- a/crates/tranquil-pds/tests/sync_blob.rs +++ b/crates/tranquil-pds/tests/sync_blob.rs @@ -21,6 +21,8 @@ async fn test_list_blobs_success() { .await .expect("Failed to upload blob"); assert_eq!(blob_res.status(), StatusCode::OK); + let blob_body: Value = blob_res.json().await.unwrap(); + publish_blob(&client, &access_jwt, &did, blob_body["blob"].clone()).await; let params = [("did", did.as_str())]; let res = client .get(format!( @@ -74,6 +76,7 @@ async fn test_get_blob_success() { .expect("Failed to upload blob"); assert_eq!(blob_res.status(), StatusCode::OK); let blob_body: Value = blob_res.json().await.expect("Response was not valid JSON"); + publish_blob(&client, &access_jwt, &did, blob_body["blob"].clone()).await; let cid = blob_body["blob"]["ref"]["$link"].as_str().expect("No CID"); let params = [("did", did.as_str()), ("cid", cid)]; let res = client @@ -120,3 +123,15 @@ async fn test_get_blob_not_found() { let body: Value = res.json().await.expect("Response was not valid JSON"); assert_eq!(body["error"], "BlobNotFound"); } + +async fn publish_blob(client: &reqwest::Client, token: &str, did: &str, blob: Value) { + let response = client.post(format!("{}/xrpc/com.atproto.repo.createRecord", base_url().await)) + .bearer_auth(token).json(&serde_json::json!({"repo":did,"collection":"com.example.blob","validate":false,"record":{"$type":"com.example.blob","attachment":blob}})) + .send().await.unwrap(); + assert_eq!( + response.status(), + StatusCode::OK, + "{}", + response.text().await.unwrap() + ); +} diff --git a/crates/tranquil-server/src/main.rs b/crates/tranquil-server/src/main.rs index 9c19213f..4e2f3ec9 100644 --- a/crates/tranquil-server/src/main.rs +++ b/crates/tranquil-server/src/main.rs @@ -327,6 +327,7 @@ async fn run() -> Result<(), Box> { state.repos.repo.clone(), state.block_store.clone(), state.eventlog_segments_dir.clone(), + state.repos.pool.clone(), shutdown.clone(), )); diff --git a/crates/tranquil-storage/src/lib.rs b/crates/tranquil-storage/src/lib.rs index f5093541..249339da 100644 --- a/crates/tranquil-storage/src/lib.rs +++ b/crates/tranquil-storage/src/lib.rs @@ -1,4 +1,4 @@ -pub use tranquil_infra::{BlobStorage, StorageError, StreamUploadResult}; +pub use tranquil_infra::{BlobDownloadStream, BlobStorage, StorageError, StreamUploadResult}; use async_trait::async_trait; use bytes::Bytes; @@ -198,6 +198,26 @@ mod s3 { .map_err(|e| StorageError::Backend(e.to_string())) } + async fn get_stream(&self, key: &str) -> Result { + let response = self + .client + .get_object() + .bucket(&self.bucket) + .key(key) + .send() + .await + .map_err(|e| StorageError::Backend(e.to_string()))?; + Ok(Box::pin(futures::stream::try_unfold( + response.body, + |mut body| async move { + body.try_next() + .await + .map(|chunk| chunk.map(|bytes| (bytes, body))) + .map_err(|e| std::io::Error::other(e.to_string())) + }, + ))) + } + async fn get_head(&self, key: &str, size: usize) -> Result { let range = format!("bytes=0-{}", size.saturating_sub(1)); let resp = self @@ -485,6 +505,25 @@ impl BlobStorage for FilesystemBlobStorage { self.get(key).await.map(Bytes::from) } + async fn get_stream(&self, key: &str) -> Result { + let file = tokio::fs::File::open(self.resolve_path(key)?) + .await + .map_err(map_io_not_found(key))?; + Ok(Box::pin(futures::stream::try_unfold( + file, + |mut file| async move { + use tokio::io::AsyncReadExt; + let mut buffer = vec![0; 64 * 1024]; + let n = file.read(&mut buffer).await?; + if n == 0 { + return Ok::<_, std::io::Error>(None); + } + buffer.truncate(n); + Ok(Some((Bytes::from(buffer), file))) + }, + ))) + } + async fn get_head(&self, key: &str, size: usize) -> Result { use tokio::io::AsyncReadExt; let path = self.resolve_path(key)?; diff --git a/crates/tranquil-sync/src/blob.rs b/crates/tranquil-sync/src/blob.rs index 0e1f43a4..d86be421 100644 --- a/crates/tranquil-sync/src/blob.rs +++ b/crates/tranquil-sync/src/blob.rs @@ -35,6 +35,16 @@ pub async fn get_blob( Err(e) => return e.into_response(), }; + if let Some(pool) = &state.repos.pool { + match tranquil_pds::storage::BlobAccess::new(pool.clone()) + .public_blob(&did, &cid) + .await + { + Ok(true) => {} + Ok(false) => return ApiError::BlobNotFound(None).into_response(), + Err(_) => return ApiError::InternalError(None).into_response(), + } + } let blob_result = state.repos.blob.get_blob_metadata(&cid).await; match blob_result { Ok(Some(metadata)) => match state.blob_store.get(&metadata.storage_key).await { @@ -92,6 +102,25 @@ pub async fn list_blobs( let cursor_cid = params.cursor.as_deref().unwrap_or(""); let user_id = account.user_id; + if let Some(pool) = &state.repos.pool { + let result = tranquil_pds::storage::BlobAccess::new(pool.clone()) + .public_blobs( + &did, + cursor_cid, + limit + 1, + params.since.as_ref().map(Tid::as_str), + ) + .await; + return match result { + Ok(mut cids) => { + let more = cids.len() > limit as usize; + cids.truncate(limit as usize); + let cursor = if more { cids.last().cloned() } else { None }; + Json(ListBlobsOutput { cursor, cids }).into_response() + } + Err(_) => ApiError::InternalError(None).into_response(), + }; + } let cids_result: Result, _> = if let Some(since) = ¶ms.since { state .repos diff --git a/docs/7_ATPROTO_SPACES.md b/docs/7_ATPROTO_SPACES.md index c7e3c905..2eb23a4e 100644 --- a/docs/7_ATPROTO_SPACES.md +++ b/docs/7_ATPROTO_SPACES.md @@ -1,15 +1,15 @@ -# Experimental AT Protocol Spaces +# AT Protocol Spaces on PostgreSQL -This fork implements an opt-in Spaces flow for PostgreSQL. It supports -OAuth-authorized private records and shared reads using application-key-bound -credentials. It is not yet a complete federated Spaces host. +Spaces is available by default on PostgreSQL. No enable flag is required. +The former `ENABLE_ATPROTO_SPACES` / `server.enable_atproto_spaces` setting is +accepted for configuration compatibility and ignored, including `false`. +The embedded backend still returns `NotImplemented`; private data never falls +back to public repositories. This is a defined implementation of the pinned +proposal below, not a claim to implement every proposed policy or backend. -Enable it with `ENABLE_ATPROTO_SPACES=true` (or `server.enable_atproto_spaces = true` -in TOML). The default is false. The embedded storage backend returns -`NotImplemented`; no private data falls back to public storage. Normal startup -applies `migrations/20261003_spaces.sql` and -`migrations/20261005_space_notifications.sql` to PostgreSQL. Use disposable local -infrastructure for this experiment, not the shared Opnshelf PDS. +Normal startup applies the Spaces, notifications, resource-accounting and private +blob migrations. Read [Spaces operations](8_SPACES_OPERATIONS.md) before upgrading: +old and new server binaries must not serve the same blob store concurrently. ## Protocol baseline @@ -87,7 +87,7 @@ DID and exact requested scope. Handles are shown only when the DID document's `alsoKnownAs` and forward handle resolution agree; failed or timed-out resolution falls back to the DID. -## Remaining compatibility work +## Supported boundaries and remaining compatibility work - Immediate revocation reaches known writers only; an unannounced remote copy still relies on credential expiry. Deletion sends `notifySpaceDeleted` to @@ -95,11 +95,29 @@ falls back to the DID. retain OAuth access to their own copies. - App-attestation verification, `allowList`, and `managingAppPolicy`. These are explicitly rejected rather than partially enforced. -- Private blobs: records containing blob references are rejected. Public blob - uploads must not be used as a substitute for private storage. -- Embedded storage, account migration/export/import integration, and production - resource controls. CAR export currently caps record data at 32 MiB; paginated - reads remain available for larger repos. +- Embedded storage and automatic account migration/export/import integration. + Public repository exports do not contain Spaces. A full PostgreSQL backup plus + the blob store preserves private data; the public-account migration tools do + not transfer it. See the operational constraints before moving an account. +- Private blob uploads use `com.atproto.repo.uploadBlob`, with the usual blob + upload permission. Reference the returned blob in a Space record, then fetch + it with authenticated `com.atproto.space.getBlob` / `listBlobs`. Uploading alone + does not publish bytes through public sync. A current public record explicitly + publishes its referenced uploads. Pre-upgrade public uploads remain public. +- Full CAR exports spool record blocks to an anonymous temporary file and stream + the response using a separate private-export concurrency pool and a configurable + deadline. The former 32 MiB + cutoff is removed. Record metadata/index memory and disk use remain bounded + by configured account quotas; temporary storage must have capacity for exports. +- Resource controls bound account repositories, records, record bytes and private + blob bytes; Space members, writers and subscribers; retained operation logs; + and request rates. See the defaults in the operations guide. +- An expired `listRepoOps` revision or cursor returns HTTP 400 with + `RepoHistoryUnavailable`; clients must recover with `getRepo`. This is a known + divergence: the pinned reference silently returns its retained operation window, + and its lexicon does not name this error. The proposal requires full-repo + fallback but leaves the signal unspecified (`SPECAMB` in the implementation). + Syncers that only recognize the reference's named errors need this recovery case. - Application-level client integration and production DNS/TLS testing. Disposable two-server credential/read/sync and notification interoperability pass against the pinned reference PDS. @@ -112,7 +130,7 @@ PLC directory automatically. Docker must be running. Do not set a hosted `DATABASE_URL` or run with external infrastructure enabled. ```sh -SQLX_OFFLINE=true ENABLE_ATPROTO_SPACES=true cargo test --locked \ +SQLX_OFFLINE=true cargo test --locked \ -p tranquil-pds --test spaces --test oauth_lifecycle -- --test-threads=1 SQLX_OFFLINE=true cargo test --locked \ -p tranquil-pds -p tranquil-api -p tranquil-types -p tranquil-scopes \ @@ -176,8 +194,7 @@ an extra account also exposed the harness's shared creation rate limit; the new regressions now reuse the existing credential-test account, and the grouped suite passes without relaxing production limits. -Commands used for this review pass (with `SQLX_OFFLINE=true`; also set -`ENABLE_ATPROTO_SPACES=true` for integration tests): +Commands used for this review pass (with `SQLX_OFFLINE=true`; no Spaces enable flag is needed): ```sh cargo check --locked -p tranquil-api -p tranquil-oauth-server @@ -217,7 +234,7 @@ Spaces crypto tests, and the 13 API library tests. In addition to the formatting and diff checks above, the affected-package commands are: ```sh -SQLX_OFFLINE=true ENABLE_ATPROTO_SPACES=true cargo test --locked -p tranquil-pds --test spaces -- --test-threads=1 +SQLX_OFFLINE=true cargo test --locked -p tranquil-pds --test spaces -- --test-threads=1 SQLX_OFFLINE=true cargo test --locked -p tranquil-pds --lib spaces:: -- --test-threads=1 SQLX_OFFLINE=true cargo test --locked -p tranquil-api --lib -- --test-threads=1 SQLX_OFFLINE=true cargo clippy --locked -p tranquil-api -p tranquil-pds --lib -- -D warnings @@ -244,7 +261,7 @@ SQLX_OFFLINE=true cargo clippy --locked -p tranquil-pds --test spaces -- -D warn - `tranquil-pds/tests/spaces.rs` and `tests/interop/`: local end-to-end tests, notification regressions, and the opt-in reference-PDS harness. - `tranquil-config`, `example.toml`, `CONTEXT.md`, crate manifests and lockfile: - opt-in configuration, terminology, and cryptographic/parser dependencies. + configuration, terminology, and cryptographic/parser dependencies. ## Local reference-PDS interoperability (2026-10-05) @@ -306,8 +323,7 @@ enforcement by Tranquil. Storage regressions cover service-auth boundaries, duplicate/stale/future revisions, retention, durable retries, and completion of an older queued generation after a newer write. -Notification verification commands (all passed; `SQLX_OFFLINE=true`, and -`ENABLE_ATPROTO_SPACES=true` for integration runs): +Notification verification commands (all passed; `SQLX_OFFLINE=true`): ```sh cargo check --locked -p tranquil-api -p tranquil-server diff --git a/docs/8_SPACES_OPERATIONS.md b/docs/8_SPACES_OPERATIONS.md new file mode 100644 index 00000000..485211a2 --- /dev/null +++ b/docs/8_SPACES_OPERATIONS.md @@ -0,0 +1,106 @@ +# Operating PostgreSQL Spaces + +Spaces endpoints and notification delivery start automatically with PostgreSQL. +The old enable setting is ignored. Existing clients keep using their public +repositories; the server does not move their records or opt them into Spaces. +Private writes require explicit OAuth Space permissions. Space credentials grant +reads only. See [protocol support](7_ATPROTO_SPACES.md) for supported policies. + +## Upgrade and rollback + +1. Back up PostgreSQL, the blob store, the public block store and the existing + signing/encryption configuration through the operator's secure backup system. + Keep key material out of logs and support reports. Restore all components + from a consistent snapshot when recovering a whole PDS. +2. Stop old server replicas before starting the new binary against the database. + Startup applies additive SQL migrations and backfills private usage counters + and existing public blob visibility. Do not run a rolling mixture of old and + new binaries: the old public blob endpoint did not enforce private access. +3. Start the normal full server image with the existing PostgreSQL and blob + storage configuration. No Spaces flag is needed. Ensure its temporary + directory is writable and has space for concurrent full exports. Set `TMPDIR` + to disk-backed storage; a memory-backed `/tmp` would charge exports to RAM. +4. Check health, an existing public account login/read/write/blob flow, and a + synthetic OAuth Space create/write/read/delete flow. Confirm an unauthenticated + public blob request cannot retrieve a newly uploaded private attachment. + +After accepting private blob uploads, **do not roll back to a binary without the +blob visibility checks**. Use a forward fix or a compatible rollback build that +retains these checks. Disabling a former feature flag does not protect bytes from +an old public blob endpoint. Never restore an older database while retaining a +newer blob store and assume that private access metadata is intact. + +## Resource limits + +All values below are defaults. TOML names are the environment names lowercased. +Quota checks serialize per account and roll back the complete write batch when +exceeded. Deletes remain available when an operator lowers an account quota. + +| Environment setting | Default | Scope | +| --- | ---: | --- | +| `SPACES_EXPORT_TIMEOUT_SECONDS` | 900 | Export deadline; increase for slow clients or large repos | +| `SPACES_MAX_REPOS_PER_ACCOUNT` | 1,000 | Hosted private repos and owned Space configurations, including tombstones | +| `SPACES_MAX_RECORDS_PER_ACCOUNT` | 100,000 | Live records across private repos | +| `SPACES_MAX_RECORD_BYTES_PER_ACCOUNT` | 1 GiB | Encoded private records | +| `SPACES_MAX_BLOB_BYTES_PER_ACCOUNT` | 10 GiB | Distinct blobs referenced by private records | +| `SPACES_MAX_OPS_PER_REPO` | 10,000 | Retained sync operations, plus batching slack | +| `SPACES_MAX_MEMBERS_PER_SPACE` | 10,000 | Members and tracked writers, counted separately | +| `SPACES_MAX_SYNCERS_PER_SPACE` | 100 | Active notification services | +| `MAX_PENDING_BLOB_BYTES_PER_ACCOUNT` | 1 GiB | Uploaded blobs without a live record reference | +| `MAX_PENDING_BLOBS_PER_ACCOUNT` | 10,000 | Uploaded blobs without a live record reference | + +Operation pruning keeps whole revisions. A cursor or revision older than the +retained boundary returns an explicit error directing the caller to `getRepo`. +An omitted `since` returns the retained window, as the proposal specifies; it is +not a full-history export. Full exports use a separate concurrency pool from public exports and retain +their permit until completion, disconnect, or the configured deadline (15 minutes by default). The deadline +runs even when a client stops reading. Temporary export files disappear on close. +Filesystem and S3 private blob downloads stream with backpressure. Unattached +upload claims expire 24 hours after upload (deleting the last reference does not +restart that grace period); the scheduled sweep reclaims unreferenced +new blob objects. It preserves existing public blobs and live record references. +Blob publication admits at most four concurrent uploads per process, further +limited to one quarter of the DB pool (minimum one), with a 30-second deadline. +Short bursts wait up to five seconds for a publication slot outside the DB pool. +Busy or timed-out publication returns a retryable 503 response. Database lock +waits have a separate two-second timeout. + +Spaces request limits use the existing distributed limiter with local fallback: +600 writes/notifications and 120 credential operations per minute per verified +account or service identity, in separate buckets. Subscription mutations are +keyed by the service URL after credential verification. Reads retain a 3,000/min +client-IP limit governed by the existing trusted-proxy configuration. These +supplement storage quotas and body-size limits. + +## Backup, deletion and account migration + +A database backup must include every `space_*` table, `blob_upload_claims`, +`blobs`, `record_blobs` and `blob_deletion_queue`. Blob bytes remain in the configured blob store. +Copying only public repository CAR files is not a Spaces backup. Account deletion +removes its private records and claims; shared content survives when another +account uploaded or legitimately references the same bytes. Metadata deletions +enqueue physical cleanup durably; storage failures are retried by the sweep. + +Automatic account migration currently transfers the public repository only. +It does not transfer Space configurations, memberships, private repositories or +private blobs. Do not use that workflow for an account with Spaces until its +private state has been transferred and verified independently. A full-host +backup/restore preserves this state; public `importRepo` is not a substitute. +A migrated-away or deactivated account cannot read or delegate its private data; +that data remains stored until deletion. Migration interfaces should warn before +deactivating accounts with private repositories and verify a separate transfer. +This limitation is separate from Spaces being available without an enable flag. + +## Reproducible verification + +`sh scripts/verify-spaces-upgrade.sh` creates an isolated PostgreSQL container, +loads the preceding schema with synthetic public/private data, applies the new +migrations, and verifies a PostgreSQL dump/restore plus counters and cascades. +It exposes no ports and removes its own container afterward. It tests database +metadata; operators must separately verify their configured blob-store backups. + +The disposable `spaces`, `repo_blob` and `sync_blob` integration suites cover +OAuth boundaries, private/public blob publication, deduplicated account deletion, +quota rollback, operation retention and a CAR larger than 32 MiB. The separate +[reference interoperability harness](../crates/tranquil-pds/tests/interop/README.md) +checks cross-host credentials, reads, signed commits, CARs and notifications. diff --git a/example.toml b/example.toml index e5af2669..1ab0d0fe 100644 --- a/example.toml +++ b/example.toml @@ -1,9 +1,14 @@ [server] # Optional DNS server for lexicon/permission TXT lookups (system DNS by default). # protocol_dns_server = "1.1.1.1:53" -# Experimental Spaces support (PostgreSQL only); see docs/7_ATPROTO_SPACES.md. -# Also available as ENABLE_ATPROTO_SPACES. Disabled by default. -#enable_atproto_spaces = false +# Spaces is always available with PostgreSQL; there is no enable switch. +#spaces_export_timeout_seconds = 900 +#spaces_max_repos_per_account = 1000 +#spaces_max_records_per_account = 100000 +#spaces_max_record_bytes_per_account = 1073741824 +#spaces_max_blob_bytes_per_account = 10737418240 +#max_pending_blob_bytes_per_account = 1073741824 +#max_pending_blobs_per_account = 10000 # Public hostname of the PDS, such as `pds.example.com`. # diff --git a/migrations/20261006_space_resource_accounting.sql b/migrations/20261006_space_resource_accounting.sql new file mode 100644 index 00000000..d87ff61f --- /dev/null +++ b/migrations/20261006_space_resource_accounting.sql @@ -0,0 +1,36 @@ +-- Transactional usage accounting keeps quota checks independent of record size. +ALTER TABLE space_repos ADD COLUMN record_count BIGINT NOT NULL DEFAULT 0 CHECK(record_count >= 0); +ALTER TABLE space_repos ADD COLUMN record_bytes BIGINT NOT NULL DEFAULT 0 CHECK(record_bytes >= 0); +UPDATE space_repos r SET record_count=s.n, record_bytes=s.bytes +FROM (SELECT space,author,COUNT(*) n,SUM(octet_length(data)) bytes + FROM space_records GROUP BY space,author) s +WHERE r.space=s.space AND r.author=s.author; +CREATE INDEX space_repos_author ON space_repos(author); +CREATE INDEX space_configs_owner ON space_configs(owner_did); + +CREATE FUNCTION account_space_record_usage() RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN + IF TG_OP = 'INSERT' THEN + UPDATE space_repos SET record_count=record_count+1, + record_bytes=record_bytes+octet_length(NEW.data) + WHERE space=NEW.space AND author=NEW.author; + RETURN NEW; + ELSIF TG_OP = 'UPDATE' THEN + UPDATE space_repos SET record_bytes=record_bytes+octet_length(NEW.data)-octet_length(OLD.data) + WHERE space=NEW.space AND author=NEW.author; + RETURN NEW; + ELSE + UPDATE space_repos SET record_count=record_count-1, + record_bytes=record_bytes-octet_length(OLD.data) + WHERE space=OLD.space AND author=OLD.author; + RETURN OLD; + END IF; +END; +$$; +CREATE TRIGGER space_record_usage AFTER INSERT OR UPDATE OR DELETE ON space_records +FOR EACH ROW EXECUTE FUNCTION account_space_record_usage(); + +ALTER TABLE space_repos ADD COLUMN ops_floor_rev TEXT; +ALTER TABLE space_repos ADD COLUMN ops_floor_seq BIGINT; +ALTER TABLE space_repos ADD COLUMN ops_count BIGINT NOT NULL DEFAULT 0; +UPDATE space_repos r SET ops_count=(SELECT COUNT(*) FROM space_ops o WHERE o.space=r.space AND o.author=r.author); diff --git a/migrations/20261007_space_blobs.sql b/migrations/20261007_space_blobs.sql new file mode 100644 index 00000000..ebffee12 --- /dev/null +++ b/migrations/20261007_space_blobs.sql @@ -0,0 +1,100 @@ +ALTER TABLE blobs ADD COLUMN legacy_public BOOLEAN NOT NULL DEFAULT TRUE; +ALTER TABLE blobs ALTER COLUMN legacy_public SET DEFAULT FALSE; +-- New uploads are private until referenced by a public record. Claims record +-- which accounts uploaded the bytes, independent of global CID deduplication. +CREATE TABLE blob_upload_claims ( + cid TEXT NOT NULL REFERENCES blobs(cid) ON DELETE CASCADE, + author TEXT NOT NULL REFERENCES users(did) ON DELETE CASCADE, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY(cid, author) +); +CREATE TABLE space_record_blobs ( + space TEXT NOT NULL, + author TEXT NOT NULL, + collection TEXT NOT NULL, + rkey TEXT NOT NULL, + cid TEXT NOT NULL REFERENCES blobs(cid) ON DELETE CASCADE, + PRIMARY KEY(space,author,collection,rkey,cid), + FOREIGN KEY(space,author,collection,rkey) REFERENCES space_records(space,author,collection,rkey) ON DELETE CASCADE +); +CREATE INDEX space_record_blobs_cid ON space_record_blobs(cid); +CREATE INDEX space_record_blobs_author ON space_record_blobs(author,cid); +CREATE INDEX blob_upload_claims_author ON blob_upload_claims(author,cid); + +ALTER TABLE record_blobs ADD COLUMN record_cid TEXT; +UPDATE record_blobs rb SET record_cid=r.record_cid FROM records r JOIN users u ON u.id=r.repo_id +WHERE rb.repo_id=r.repo_id AND rb.record_uri='at://' || u.did || '/' || r.collection || '/' || r.rkey; + + +-- Shared definition of a live attachment, used by pending-upload accounting +-- and reclamation. Public entries are bound to the exact committed version. +CREATE VIEW blob_record_references AS +SELECT cid,author FROM space_record_blobs +UNION +SELECT rb.blob_cid,u.did FROM record_blobs rb +JOIN records r ON r.repo_id=rb.repo_id AND r.record_cid=rb.record_cid +JOIN users u ON u.id=r.repo_id +WHERE rb.record_uri='at://' || u.did || '/' || r.collection || '/' || r.rkey; + +-- Backfill/import callers index only records already committed. Bind their +-- entries to that version, so stale references cannot disclose later uploads. +CREATE FUNCTION bind_public_blob_record() RETURNS TRIGGER LANGUAGE plpgsql AS $$ +BEGIN + IF NEW.record_cid IS NULL THEN + SELECT r.record_cid INTO NEW.record_cid FROM records r JOIN users u ON u.id=r.repo_id + WHERE r.repo_id=NEW.repo_id AND NEW.record_uri='at://' || u.did || '/' || r.collection || '/' || r.rkey; + END IF; + RETURN NEW; +END; +$$; +CREATE TRIGGER bind_public_blob_record BEFORE INSERT ON record_blobs +FOR EACH ROW EXECUTE FUNCTION bind_public_blob_record(); + +-- Global deduplication must not let the first uploader's deletion remove +-- another account's upload or a referenced public/private blob. +CREATE FUNCTION release_account_blobs(account UUID) RETURNS VOID LANGUAGE plpgsql AS $$ +DECLARE + locked_cid TEXT; +BEGIN + FOR locked_cid IN SELECT cid FROM blobs WHERE created_by_user=account ORDER BY cid LOOP + PERFORM pg_advisory_xact_lock(hashtextextended('blob:' || locked_cid, 0)); + END LOOP; + DELETE FROM space_repos WHERE author=(SELECT did FROM users WHERE id=account); + DELETE FROM blob_upload_claims WHERE author=(SELECT did FROM users WHERE id=account); + DELETE FROM record_blobs WHERE repo_id=account; + UPDATE blobs b SET created_by_user=keepers.id FROM ( + SELECT cid, MIN(id::text)::uuid AS id FROM ( + SELECT c.cid,u.id FROM blob_upload_claims c JOIN users u ON u.did=c.author WHERE u.id<>account + UNION ALL SELECT r.blob_cid,r.repo_id FROM record_blobs r JOIN blobs b ON b.cid=r.blob_cid WHERE r.repo_id<>account AND b.legacy_public + UNION ALL SELECT r.cid,u.id FROM space_record_blobs r JOIN users u ON u.did=r.author WHERE u.id<>account + ) remaining GROUP BY cid + ) keepers WHERE b.cid=keepers.cid AND b.created_by_user=account; + DELETE FROM blobs WHERE created_by_user=account; +END; +$$; + +CREATE INDEX blob_upload_claims_age ON blob_upload_claims(created_at); + +-- Physical cleanup must survive account/admin deletion and object-store outages. +-- This has no FK: the queue deliberately outlives the metadata it removes. +CREATE TABLE blob_deletion_queue ( + cid TEXT PRIMARY KEY, + storage_key TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); +CREATE FUNCTION enqueue_blob_deletion() RETURNS TRIGGER LANGUAGE plpgsql AS $$ +BEGIN + INSERT INTO blob_deletion_queue(cid,storage_key) VALUES(OLD.cid,OLD.storage_key) + ON CONFLICT(cid) DO UPDATE SET storage_key=EXCLUDED.storage_key; + RETURN OLD; +END; +$$; +CREATE TRIGGER enqueue_blob_deletion AFTER DELETE ON blobs +FOR EACH ROW EXECUTE FUNCTION enqueue_blob_deletion(); + +CREATE INDEX blobs_created_at ON blobs(created_at); +ALTER TABLE space_records ADD COLUMN rev TEXT; +UPDATE space_records r SET rev=COALESCE( + (SELECT MAX(o.rev) FROM space_ops o WHERE o.space=r.space AND o.author=r.author AND o.collection=r.collection AND o.rkey=r.rkey), + (SELECT repo.rev FROM space_repos repo WHERE repo.space=r.space AND repo.author=r.author) +); diff --git a/scripts/verify-spaces-upgrade.sh b/scripts/verify-spaces-upgrade.sh new file mode 100644 index 00000000..93289c06 --- /dev/null +++ b/scripts/verify-spaces-upgrade.sh @@ -0,0 +1,67 @@ +#!/bin/sh +# Disposable, synthetic PostgreSQL upgrade and metadata backup/restore check. +# No ports, hosted database configuration, or real account data are used. +set -eu +root=$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd) +scratch=$(mktemp -d) +container="tranquil-spaces-upgrade-$$" +created=false +cleanup() { + if [ "$created" = true ]; then docker rm -f "$container" >/dev/null; fi + rm -rf "$scratch" +} +trap cleanup EXIT HUP INT TERM +docker run -d --name "$container" --label tranquil_spaces_upgrade=true \ + -e POSTGRES_HOST_AUTH_METHOD=trust postgres:17 > /dev/null +created=true +attempt=0 +until docker exec "$container" pg_isready -U postgres >/dev/null 2>&1; do + attempt=$((attempt + 1)) + if [ "$attempt" -gt 30 ]; then echo 'Disposable PostgreSQL did not start' >&2; exit 1; fi + sleep 1 +done +for migration in "$root"/migrations/*.sql; do + case "$migration" in + */20261006_space_resource_accounting.sql|*/20261007_space_blobs.sql) continue ;; + esac + cat "$migration" + printf '\n' +done > "$scratch/baseline.sql" +docker exec -i "$container" psql -X -U postgres -v ON_ERROR_STOP=1 -q < "$scratch/baseline.sql" +docker exec -i "$container" psql -X -U postgres -v ON_ERROR_STOP=1 -q <<'SQL' +INSERT INTO users(id,handle,did) VALUES('00000000-0000-0000-0000-000000000001','upgrade.test','did:plc:aaaaaaaaaaaaaaaaaaaaaaaa'); +INSERT INTO repos(user_id,repo_root_cid) VALUES('00000000-0000-0000-0000-000000000001','public-root'); +INSERT INTO records(repo_id,collection,rkey,record_cid) VALUES('00000000-0000-0000-0000-000000000001','com.example.record','one','public-record'); +INSERT INTO blobs(cid,mime_type,size_bytes,created_by_user,storage_key) VALUES('old-upload','text/plain',3,'00000000-0000-0000-0000-000000000001','old-upload'); +INSERT INTO record_blobs(repo_id,record_uri,blob_cid) VALUES('00000000-0000-0000-0000-000000000001','at://did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/com.example.record/one','old-upload'); +INSERT INTO space_repos(space,author,hash_state) VALUES('test-space','did:plc:aaaaaaaaaaaaaaaaaaaaaaaa',decode(repeat('00',2048),'hex')); +INSERT INTO space_records(space,author,collection,rkey,cid,data,value) VALUES('test-space','did:plc:aaaaaaaaaaaaaaaaaaaaaaaa','com.example.record','private','private-record',decode('010203','hex'),'{}'); +SQL +for migration in 20261006_space_resource_accounting.sql 20261007_space_blobs.sql; do + docker exec -i "$container" psql -X -U postgres -v ON_ERROR_STOP=1 -q < "$root/migrations/$migration" +done +docker exec -i "$container" psql -X -U postgres -v ON_ERROR_STOP=1 -q <<'SQL' +DO $$ BEGIN + IF NOT EXISTS(SELECT 1 FROM space_repos WHERE record_count=1 AND record_bytes=3) THEN RAISE EXCEPTION 'Private usage was not backfilled'; END IF; + IF NOT EXISTS(SELECT 1 FROM blobs WHERE cid='old-upload' AND legacy_public) THEN RAISE EXCEPTION 'Existing public upload lost its visibility'; END IF; + IF NOT EXISTS(SELECT 1 FROM record_blobs WHERE record_cid='public-record') THEN RAISE EXCEPTION 'Existing public reference was not bound'; END IF; +END $$; +INSERT INTO blobs(cid,mime_type,size_bytes,created_by_user,storage_key) VALUES('new-upload','text/plain',3,'00000000-0000-0000-0000-000000000001','new-upload'); +INSERT INTO blob_upload_claims(cid,author) VALUES('new-upload','did:plc:aaaaaaaaaaaaaaaaaaaaaaaa'); +INSERT INTO space_record_blobs(space,author,collection,rkey,cid) VALUES('test-space','did:plc:aaaaaaaaaaaaaaaaaaaaaaaa','com.example.record','private','new-upload'); +SQL +docker exec "$container" pg_dump -U postgres -Fc postgres > "$scratch/backup.dump" +docker exec "$container" createdb -U postgres restored +docker exec -i "$container" pg_restore -U postgres --exit-on-error -d restored < "$scratch/backup.dump" +docker exec -i "$container" psql -X -U postgres -d restored -v ON_ERROR_STOP=1 -q <<'SQL' +DO $$ BEGIN + IF NOT EXISTS(SELECT 1 FROM space_records WHERE encode(data,'hex')='010203') THEN RAISE EXCEPTION 'Private record was not restored'; END IF; + IF NOT EXISTS(SELECT 1 FROM space_record_blobs r JOIN blobs b ON b.cid=r.cid JOIN blob_upload_claims c ON c.cid=b.cid AND c.author=r.author WHERE NOT b.legacy_public) THEN RAISE EXCEPTION 'Private blob access metadata was not restored'; END IF; +END $$; +DELETE FROM space_records; +DO $$ BEGIN + IF EXISTS(SELECT 1 FROM space_repos WHERE record_count<>0 OR record_bytes<>0) THEN RAISE EXCEPTION 'Restored usage trigger failed'; END IF; + IF EXISTS(SELECT 1 FROM space_record_blobs) THEN RAISE EXCEPTION 'Restored blob-reference cascade failed'; END IF; +END $$; +SQL +echo 'Spaces upgrade and PostgreSQL metadata backup/restore passed.'