From 466497ebb0c2ca05767841d597d4005fa985ec79 Mon Sep 17 00:00:00 2001 From: Rowan-Paul Date: Tue, 6 Oct 2026 15:31:37 +0200 Subject: [PATCH] Enable PostgreSQL Spaces with deployment safeguards The experimental flag left Spaces unavailable by default, while private blob storage, resource bounds and upgrade compatibility were incomplete. Enable routing and notifications automatically and preserve public clients. Add authenticated private blobs, atomic publication ownership checks, account quotas, retained sync history, bounded streamed exports and upload publication. Preserve legacy public blobs and shared content on deletion; queue physical cleanup durably and support incremental blob enumeration. Document coordinated upgrades, backup and rollback requirements, and the remaining account-migration and policy boundaries. Verify 15 Spaces/public-blob regressions, reference-PDS interoperability, Clippy with warnings denied, formatting, disposable PostgreSQL upgrade and restore, and the ARM64 container build/startup/health without a Spaces flag. Both Fable 5.1 review axes found no confirmed deployment blockers; apply their remaining operational and compatibility recommendations. Supports Rowan-Paul/opnshelf#252. Co-Authored-By: GPT-6-Astra --- CONTEXT.md | 8 +- Cargo.lock | 2 + Dockerfile | 6 +- crates/tranquil-api/src/repo/blob.rs | 159 ++++- crates/tranquil-api/src/spaces.rs | 87 ++- crates/tranquil-api/src/spaces/auth.rs | 12 + crates/tranquil-api/src/spaces/blobs.rs | 81 +++ .../tranquil-api/src/spaces/notifications.rs | 15 +- crates/tranquil-config/src/lib.rs | 95 ++- crates/tranquil-db-traits/src/repo.rs | 1 + crates/tranquil-db/Cargo.toml | 3 + crates/tranquil-db/src/postgres/blob.rs | 20 +- crates/tranquil-db/src/postgres/blob_refs.rs | 94 +++ crates/tranquil-db/src/postgres/mod.rs | 1 + crates/tranquil-db/src/postgres/repo.rs | 2 + crates/tranquil-db/src/postgres/user.rs | 24 +- crates/tranquil-infra/src/lib.rs | 7 + crates/tranquil-pds/Cargo.toml | 3 +- .../tranquil-pds/src/rate_limit/extractor.rs | 13 + crates/tranquil-pds/src/rate_limit/mod.rs | 6 + crates/tranquil-pds/src/repo_ops.rs | 13 + crates/tranquil-pds/src/scheduled.rs | 49 +- crates/tranquil-pds/src/spaces/mod.rs | 3 + .../tranquil-pds/src/spaces/notifications.rs | 5 +- crates/tranquil-pds/src/spaces/store.rs | 139 +++- crates/tranquil-pds/src/spaces/store/blobs.rs | 98 +++ .../src/spaces/store/notifications.rs | 22 + crates/tranquil-pds/src/spaces/store/read.rs | 100 +-- crates/tranquil-pds/src/state.rs | 32 + .../tranquil-pds/src/storage/blob_access.rs | 71 +++ crates/tranquil-pds/src/storage/mod.rs | 214 +++++++ crates/tranquil-pds/tests/interop/README.md | 2 +- crates/tranquil-pds/tests/spaces.rs | 603 +++++++++++++++++- crates/tranquil-pds/tests/sync_blob.rs | 15 + crates/tranquil-server/src/main.rs | 1 + crates/tranquil-storage/src/lib.rs | 41 +- crates/tranquil-sync/src/blob.rs | 29 + docs/7_ATPROTO_SPACES.md | 62 +- docs/8_SPACES_OPERATIONS.md | 106 +++ example.toml | 11 +- .../20261006_space_resource_accounting.sql | 36 ++ migrations/20261007_space_blobs.sql | 100 +++ scripts/verify-spaces-upgrade.sh | 67 ++ 43 files changed, 2268 insertions(+), 190 deletions(-) create mode 100644 crates/tranquil-api/src/spaces/blobs.rs create mode 100644 crates/tranquil-db/src/postgres/blob_refs.rs create mode 100644 crates/tranquil-pds/src/spaces/store/blobs.rs create mode 100644 crates/tranquil-pds/src/storage/blob_access.rs create mode 100644 docs/8_SPACES_OPERATIONS.md create mode 100644 migrations/20261006_space_resource_accounting.sql create mode 100644 migrations/20261007_space_blobs.sql create mode 100644 scripts/verify-spaces-upgrade.sh 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.' -- knotmirror 0.0.1