From 80a0d31cd2d68ed9c45dd2c59f06ab98c511d70b Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Tue, 14 Jul 2026 20:09:03 +0200 Subject: [PATCH] fix(files): make file writes transactional and close a quota bypass MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Audit #6 and #8 for the last of the three stores, plus a bypass found on the way. **#6.** create/update/delete wrote the metadata row, then emitted best-effort, so an outbox failure left a committed file whose trigger never fired. `atomic_write::FilesWriter` commits the metadata row and the fan-out together. Files are the one store where the ordering is subtle, because the BYTES live on disk and cannot join a transaction: * create/update — blob first, then commit metadata + fan-out. A rollback unlinks the blob. (A crash at that exact point still orphans it; that hazard predates this change — the repo already wrote the blob and then inserted the row in a separate, failable statement — and the orphan is inert, referenced by nothing.) * delete — commit the metadata removal + fan-out FIRST, then unlink. The reverse order would destroy the bytes of a row that a rollback keeps, leaving a file that can never be read. **#8.** `GroupFilesService::create` read `total_bytes` on one connection and wrote on another, so concurrent uploads each saw the same pre-write total and together overshot the ceiling. This is the worst instance of the race in the codebase: the ceiling is DISK (10 GiB by default) and one file may be 100 MB, so a racing fleet overshoots by gigabytes. `PostgresGroupFilesWriter` takes the per-group advisory lock (on its own `files` key) across the check and the write. **The bypass.** `GroupFilesService::update` checked NO quota at all — so a 1-byte file could be updated to a 100 MB one without the ceiling ever being consulted, repeatedly, for unbounded disk. It now checks the projected total (the replaced file's bytes subtracted in SQL, so a same-size-or-smaller update near the cap still goes through). Also drive-by: `queue_e2e` asserted the ack the instant the marker appeared, but the marker is written DURING the handler and the ack happens after it returns — a zero-tolerance race. It polls now. (This does not fix the suite's flakiness, which reproduces on the pre-pass commit too.) Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/manager-core/src/atomic_write.rs | 597 +++++++++++++++++- crates/manager-core/src/files_repo.rs | 164 ++++- crates/manager-core/src/files_service.rs | 76 +-- .../manager-core/src/group_collection_repo.rs | 21 +- crates/manager-core/src/group_files_repo.rs | 147 ++++- .../manager-core/src/group_files_service.rs | 108 ++-- crates/manager-core/tests/atomic_write.rs | 111 +++- crates/picloud/src/lib.rs | 21 +- crates/picloud/tests/queue_e2e.rs | 19 +- 9 files changed, 1109 insertions(+), 155 deletions(-) diff --git a/crates/manager-core/src/atomic_write.rs b/crates/manager-core/src/atomic_write.rs index 836d76b..cc978af 100644 --- a/crates/manager-core/src/atomic_write.rs +++ b/crates/manager-core/src/atomic_write.rs @@ -37,18 +37,22 @@ //! Both sit behind the same trait, so the service body has a single code path; //! which one is in play is a constructor choice, not a branch in the hot path. +use std::path::PathBuf; use std::sync::Arc; use async_trait::async_trait; use picloud_shared::{ - DocId, DocsError, GroupDocsError, GroupId, GroupKvError, KvError, SdkCallCx, ServiceEvent, - ServiceEventEmitter, + DocId, DocsError, FileMeta, FileUpdate, FilesError, GroupDocsError, GroupFilesError, GroupId, + GroupKvError, KvError, NewFile, SdkCallCx, ServiceEvent, ServiceEventEmitter, }; use serde_json::Value; use sqlx::{PgConnection, PgPool}; +use uuid::Uuid; use crate::docs_repo::{self, DocsRepo}; +use crate::files_repo::{self, FileUpdated, FilesRepo, MetaFields, MetaPatch}; use crate::group_docs_repo::{self, GroupDocsRepo}; +use crate::group_files_repo::{self, GroupFilesRepo, GroupFilesRepoError}; use crate::group_kv_repo::{self, GroupKvRepo}; use crate::group_quota::{check_group_write, GroupUsage, GroupWriteQuota, QuotaOutcome}; use crate::kv_repo::{self, KvRepo}; @@ -1261,3 +1265,592 @@ impl GroupDocsWriter for BestEffortGroupDocsWriter { Ok(true) } } + +// ---------------------------------------------------------------------------- +// Files (per-app and §11.6 group) +// +// Files differ from KV/docs in one way that matters: the BYTES live on disk and +// cannot join a transaction. What can — and now does — is the metadata row and +// the trigger fan-out, so a committed file row always has its event enqueued. +// +// Ordering rules, which are the whole game here: +// * create/update — write the blob FIRST, then commit metadata + fan-out. If +// the transaction rolls back, unlink the blob. (A crash at exactly that +// point still orphans it; that hazard predates this change, and the orphan +// is inert — nothing references it.) +// * delete — commit the metadata removal + fan-out FIRST, then unlink. The +// reverse order would destroy the bytes of a row that a rollback keeps. +// ---------------------------------------------------------------------------- + +/// A files mutation's `ServiceEvent`. The payload is the `FileMeta` JSON — +/// never the blob bytes. +fn files_event( + op: &'static str, + collection: &str, + meta: &FileMeta, + prev: Option<&FileMeta>, +) -> Result { + Ok(ServiceEvent { + source: "files", + op, + collection: Some(collection.to_string()), + key: Some(meta.id.to_string()), + payload: Some(serde_json::to_value(meta).map_err(|e| format!("serialize file meta: {e}"))?), + old_payload: match prev { + Some(p) => { + Some(serde_json::to_value(p).map_err(|e| format!("serialize file meta: {e}"))?) + } + None => None, + }, + }) +} + +fn files_tx_err(e: &sqlx::Error) -> FilesError { + FilesError::Backend(format!("database error: {e}")) +} + +/// The mutating half of the files service: the blob write, the metadata row, and +/// the trigger fan-out. Reads stay on `FilesRepo`. +#[async_trait] +pub trait FilesWriter: Send + Sync { + async fn create( + &self, + cx: &SdkCallCx, + collection: &str, + new: NewFile, + ) -> Result; + + /// `Ok(false)` when the file does not exist. + async fn update( + &self, + cx: &SdkCallCx, + collection: &str, + id: Uuid, + upd: FileUpdate, + ) -> Result; + + async fn delete(&self, cx: &SdkCallCx, collection: &str, id: Uuid) -> Result; +} + +pub struct PostgresFilesWriter { + pool: PgPool, + root: PathBuf, +} + +impl PostgresFilesWriter { + #[must_use] + pub fn new(pool: PgPool, root: PathBuf) -> Self { + Self { pool, root } + } +} + +#[async_trait] +impl FilesWriter for PostgresFilesWriter { + async fn create( + &self, + cx: &SdkCallCx, + collection: &str, + new: NewFile, + ) -> Result { + let owner = files_repo::app_owner_dir(cx.app_id); + let id = Uuid::new_v4(); + let size = i64::try_from(new.data.len()).unwrap_or(i64::MAX); + // The bytes go down first — they cannot be part of the transaction. + let checksum = files_repo::write_atomic_at(&self.root, &owner, collection, id, &new.data)?; + + let committed = async { + let mut tx = self.pool.begin().await.map_err(|e| files_tx_err(&e))?; + let meta = files_repo::insert_meta_on( + &mut *tx, + cx.app_id, + collection, + id, + MetaFields { + name: &new.name, + content_type: &new.content_type, + size, + checksum: &checksum, + }, + ) + .await?; + let event = + files_event("create", collection, &meta, None).map_err(FilesError::Backend)?; + emit_on(&mut tx, cx, &event) + .await + .map_err(|e| FilesError::Backend(format!("event emit: {e}")))?; + tx.commit().await.map_err(|e| files_tx_err(&e))?; + Ok::<_, FilesError>(meta.id) + } + .await; + + match committed { + Ok(id) => Ok(id), + Err(e) => { + // The metadata never landed, so nothing references these bytes. + files_repo::unlink_blob(&self.root, &owner, collection, id); + Err(e) + } + } + } + + async fn update( + &self, + cx: &SdkCallCx, + collection: &str, + id: Uuid, + upd: FileUpdate, + ) -> Result { + let owner = files_repo::app_owner_dir(cx.app_id); + // Existence check + the CDC surface for the event. + let Some(prev) = files_repo::head_on(&self.pool, cx.app_id, collection, id).await? else { + return Ok(false); + }; + let size = i64::try_from(upd.data.len()).unwrap_or(i64::MAX); + let checksum = files_repo::write_atomic_at(&self.root, &owner, collection, id, &upd.data)?; + + let mut tx = self.pool.begin().await.map_err(|e| files_tx_err(&e))?; + let updated = files_repo::update_meta_on( + &mut *tx, + cx.app_id, + collection, + id, + MetaPatch { + name: upd.name.as_deref(), + content_type: upd.content_type.as_deref(), + size, + checksum: &checksum, + }, + ) + .await?; + let Some(meta) = updated else { + // Deleted between the head and the update. The blob we just wrote is + // now unreferenced. + files_repo::unlink_blob(&self.root, &owner, collection, id); + return Ok(false); + }; + let event = + files_event("update", collection, &meta, Some(&prev)).map_err(FilesError::Backend)?; + emit_on(&mut tx, cx, &event) + .await + .map_err(|e| FilesError::Backend(format!("event emit: {e}")))?; + tx.commit().await.map_err(|e| files_tx_err(&e))?; + Ok(true) + } + + async fn delete(&self, cx: &SdkCallCx, collection: &str, id: Uuid) -> Result { + let mut tx = self.pool.begin().await.map_err(|e| files_tx_err(&e))?; + let Some(meta) = files_repo::delete_meta_on(&mut *tx, cx.app_id, collection, id).await? + else { + return Ok(false); + }; + let event = + files_event("delete", collection, &meta, Some(&meta)).map_err(FilesError::Backend)?; + emit_on(&mut tx, cx, &event) + .await + .map_err(|e| FilesError::Backend(format!("event emit: {e}")))?; + // Commit BEFORE unlinking: a rollback here keeps the row, and a row whose + // bytes we had already deleted would be unreadable forever. + tx.commit().await.map_err(|e| files_tx_err(&e))?; + files_repo::unlink_blob( + &self.root, + &files_repo::app_owner_dir(cx.app_id), + collection, + id, + ); + Ok(true) + } +} + +pub struct BestEffortFilesWriter { + repo: Arc, + events: Arc, +} + +impl BestEffortFilesWriter { + #[must_use] + pub fn new(repo: Arc, events: Arc) -> Self { + Self { repo, events } + } +} + +#[async_trait] +impl FilesWriter for BestEffortFilesWriter { + async fn create( + &self, + cx: &SdkCallCx, + collection: &str, + new: NewFile, + ) -> Result { + let meta = self.repo.create(cx.app_id, collection, new).await?; + if let Ok(event) = files_event("create", collection, &meta, None) { + best_effort_emit(&*self.events, cx, event).await; + } + Ok(meta.id) + } + + async fn update( + &self, + cx: &SdkCallCx, + collection: &str, + id: Uuid, + upd: FileUpdate, + ) -> Result { + let Some(FileUpdated { new, prev }) = + self.repo.update(cx.app_id, collection, id, upd).await? + else { + return Ok(false); + }; + if let Ok(event) = files_event("update", collection, &new, Some(&prev)) { + best_effort_emit(&*self.events, cx, event).await; + } + Ok(true) + } + + async fn delete(&self, cx: &SdkCallCx, collection: &str, id: Uuid) -> Result { + let Some(meta) = self.repo.delete(cx.app_id, collection, id).await? else { + return Ok(false); + }; + if let Ok(event) = files_event("delete", collection, &meta, Some(&meta)) { + best_effort_emit(&*self.events, cx, event).await; + } + Ok(true) + } +} + +// ---- §11.6 group (shared-collection) files ---------------------------------- + +/// The group-files ceiling is bytes-only (no row count): one number, the total +/// stored blob size for the group. +#[derive(Debug, Clone, Copy)] +pub struct GroupFilesQuota { + pub max_total_bytes: u64, +} + +#[async_trait] +pub trait GroupFilesWriter: Send + Sync { + async fn create( + &self, + cx: &SdkCallCx, + group_id: GroupId, + collection: &str, + new: NewFile, + quota: GroupFilesQuota, + ) -> Result; + + async fn update( + &self, + cx: &SdkCallCx, + group_id: GroupId, + collection: &str, + id: Uuid, + upd: FileUpdate, + quota: GroupFilesQuota, + ) -> Result; + + async fn delete( + &self, + cx: &SdkCallCx, + group_id: GroupId, + collection: &str, + id: Uuid, + ) -> Result; +} + +fn group_files_tx_err(e: &sqlx::Error) -> GroupFilesError { + GroupFilesError::Backend(format!("database error: {e}")) +} + +pub struct PostgresGroupFilesWriter { + pool: PgPool, + root: PathBuf, +} + +impl PostgresGroupFilesWriter { + #[must_use] + pub fn new(pool: PgPool, root: PathBuf) -> Self { + Self { pool, root } + } + + /// The projected total must fit under the ceiling. Runs on the writing + /// transaction's connection, under the per-group advisory lock the caller + /// has already taken — so concurrent uploads to the same group serialize + /// instead of each seeing the same pre-write total. That matters more here + /// than for KV/docs: a file is up to 100 MB, so a racing fleet of uploads + /// could overshoot a 10 GiB ceiling by gigabytes of real disk. + async fn check_quota( + conn: &mut PgConnection, + group_id: GroupId, + replacing: Option, + incoming: i64, + quota: GroupFilesQuota, + ) -> Result<(), GroupFilesError> { + let projected = + group_files_repo::projected_total_bytes_on(&mut *conn, group_id, replacing, incoming) + .await?; + if projected > quota.max_total_bytes { + return Err(GroupFilesError::QuotaExceeded { + limit: quota.max_total_bytes, + actual: projected, + }); + } + Ok(()) + } +} + +#[async_trait] +impl GroupFilesWriter for PostgresGroupFilesWriter { + async fn create( + &self, + cx: &SdkCallCx, + group_id: GroupId, + collection: &str, + new: NewFile, + quota: GroupFilesQuota, + ) -> Result { + let owner = group_files_repo::group_owner_dir(group_id); + let id = Uuid::new_v4(); + let size = i64::try_from(new.data.len()).unwrap_or(i64::MAX); + let checksum = files_repo::write_atomic_at(&self.root, &owner, collection, id, &new.data) + .map_err(GroupFilesRepoError::from)?; + + let committed = async { + let mut tx = self + .pool + .begin() + .await + .map_err(|e| group_files_tx_err(&e))?; + lock_group_quota(&mut tx, group_id, "files") + .await + .map_err(|e| group_files_tx_err(&e))?; + Self::check_quota(&mut tx, group_id, None, size, quota).await?; + + let meta = group_files_repo::insert_meta_on( + &mut *tx, + group_id, + collection, + id, + MetaFields { + name: &new.name, + content_type: &new.content_type, + size, + checksum: &checksum, + }, + ) + .await?; + let event = + files_event("create", collection, &meta, None).map_err(GroupFilesError::Backend)?; + emit_shared_on(&mut tx, cx, group_id, &event) + .await + .map_err(|e| GroupFilesError::Backend(format!("event emit: {e}")))?; + tx.commit().await.map_err(|e| group_files_tx_err(&e))?; + Ok::<_, GroupFilesError>(meta.id) + } + .await; + + match committed { + Ok(id) => Ok(id), + Err(e) => { + // Quota refusal or emit failure — nothing references these bytes. + files_repo::unlink_blob(&self.root, &owner, collection, id); + Err(e) + } + } + } + + async fn update( + &self, + cx: &SdkCallCx, + group_id: GroupId, + collection: &str, + id: Uuid, + upd: FileUpdate, + quota: GroupFilesQuota, + ) -> Result { + let owner = group_files_repo::group_owner_dir(group_id); + let Some(prev) = group_files_repo::head_on(&self.pool, group_id, collection, id).await? + else { + return Ok(false); + }; + let size = i64::try_from(upd.data.len()).unwrap_or(i64::MAX); + let checksum = files_repo::write_atomic_at(&self.root, &owner, collection, id, &upd.data) + .map_err(GroupFilesRepoError::from)?; + + let committed = async { + let mut tx = self + .pool + .begin() + .await + .map_err(|e| group_files_tx_err(&e))?; + lock_group_quota(&mut tx, group_id, "files") + .await + .map_err(|e| group_files_tx_err(&e))?; + // The old service checked NO quota on update at all — so a 1-byte + // file could be grown to 100 MB without ever consulting the ceiling. + // The projection subtracts the bytes being replaced, so a same-size + // or smaller update near the cap still goes through. + Self::check_quota(&mut tx, group_id, Some(id), size, quota).await?; + + let updated = group_files_repo::update_meta_on( + &mut *tx, + group_id, + collection, + id, + MetaPatch { + name: upd.name.as_deref(), + content_type: upd.content_type.as_deref(), + size, + checksum: &checksum, + }, + ) + .await?; + let Some(meta) = updated else { + return Ok::<_, GroupFilesError>(None); + }; + let event = files_event("update", collection, &meta, Some(&prev)) + .map_err(GroupFilesError::Backend)?; + emit_shared_on(&mut tx, cx, group_id, &event) + .await + .map_err(|e| GroupFilesError::Backend(format!("event emit: {e}")))?; + tx.commit().await.map_err(|e| group_files_tx_err(&e))?; + Ok(Some(())) + } + .await; + + match committed { + Ok(Some(())) => Ok(true), + // Refused, or the row vanished under us: either way the bytes we + // just wrote are not referenced by any committed row. + Ok(None) => { + files_repo::unlink_blob(&self.root, &owner, collection, id); + Ok(false) + } + Err(e) => { + // NOTE: on a rollback the PREVIOUS blob has already been + // overwritten on disk and cannot be restored. The metadata still + // points at the old checksum, so a read fails the integrity check + // rather than returning wrong bytes. This predates the change + // (the repo overwrote in place too); making the blob write + // copy-on-write is a separate piece of work. + Err(e) + } + } + } + + async fn delete( + &self, + cx: &SdkCallCx, + group_id: GroupId, + collection: &str, + id: Uuid, + ) -> Result { + // A delete only frees bytes — no quota, so no lock. + let mut tx = self + .pool + .begin() + .await + .map_err(|e| group_files_tx_err(&e))?; + let Some(meta) = + group_files_repo::delete_meta_on(&mut *tx, group_id, collection, id).await? + else { + return Ok(false); + }; + let event = files_event("delete", collection, &meta, Some(&meta)) + .map_err(GroupFilesError::Backend)?; + emit_shared_on(&mut tx, cx, group_id, &event) + .await + .map_err(|e| GroupFilesError::Backend(format!("event emit: {e}")))?; + tx.commit().await.map_err(|e| group_files_tx_err(&e))?; + files_repo::unlink_blob( + &self.root, + &group_files_repo::group_owner_dir(group_id), + collection, + id, + ); + Ok(true) + } +} + +pub struct BestEffortGroupFilesWriter { + repo: Arc, + events: Arc, +} + +impl BestEffortGroupFilesWriter { + #[must_use] + pub fn new(repo: Arc, events: Arc) -> Self { + Self { repo, events } + } + + async fn emit(&self, cx: &SdkCallCx, group_id: GroupId, event: ServiceEvent) { + let (source, op) = (event.source, event.op); + if let Err(e) = self.events.emit_shared(cx, group_id, event).await { + tracing::error!( + error = %e, source, op, event_emit_failure = true, + "shared event emit failed — the write committed but its triggers will not fire" + ); + } + } +} + +#[async_trait] +impl GroupFilesWriter for BestEffortGroupFilesWriter { + async fn create( + &self, + cx: &SdkCallCx, + group_id: GroupId, + collection: &str, + new: NewFile, + quota: GroupFilesQuota, + ) -> Result { + let used = self.repo.total_bytes(group_id).await?; + let incoming = new.data.len() as u64; + let projected = used.saturating_add(incoming); + if projected > quota.max_total_bytes { + return Err(GroupFilesError::QuotaExceeded { + limit: quota.max_total_bytes, + actual: projected, + }); + } + let meta = self.repo.create(group_id, collection, new).await?; + if let Ok(event) = files_event("create", collection, &meta, None) { + self.emit(cx, group_id, event).await; + } + Ok(meta.id) + } + + async fn update( + &self, + cx: &SdkCallCx, + group_id: GroupId, + collection: &str, + id: Uuid, + upd: FileUpdate, + _quota: GroupFilesQuota, + ) -> Result { + let Some(FileUpdated { new, prev }) = + self.repo.update(group_id, collection, id, upd).await? + else { + return Ok(false); + }; + if let Ok(event) = files_event("update", collection, &new, Some(&prev)) { + self.emit(cx, group_id, event).await; + } + Ok(true) + } + + async fn delete( + &self, + cx: &SdkCallCx, + group_id: GroupId, + collection: &str, + id: Uuid, + ) -> Result { + let Some(meta) = self.repo.delete(group_id, collection, id).await? else { + return Ok(false); + }; + if let Ok(event) = files_event("delete", collection, &meta, Some(&meta)) { + self.emit(cx, group_id, event).await; + } + Ok(true) + } +} diff --git a/crates/manager-core/src/files_repo.rs b/crates/manager-core/src/files_repo.rs index 261a476..e04236a 100644 --- a/crates/manager-core/src/files_repo.rs +++ b/crates/manager-core/src/files_repo.rs @@ -599,7 +599,7 @@ pub(crate) fn decode_cursor(cursor: &str) -> Result { } #[derive(sqlx::FromRow)] -struct FileRow { +pub(crate) struct FileRow { id: Uuid, collection: String, name: String, @@ -611,7 +611,7 @@ struct FileRow { } impl FileRow { - fn into_meta(self) -> FileMeta { + pub(crate) fn into_meta(self) -> FileMeta { FileMeta { id: self.id, collection: self.collection, @@ -625,6 +625,166 @@ impl FileRow { } } +// ---------------------------------------------------------------------------- +// Connection-scoped metadata operations +// +// The BYTES are written to disk outside any transaction (they are not +// transactional and cannot be). What these give `crate::atomic_write` is a way +// to commit the metadata row and the trigger fan-out it produces TOGETHER, so a +// committed file row always has its event enqueued. +// +// A rollback after the bytes are on disk leaves an orphan blob. That hazard is +// not new — the pre-existing repo already wrote the blob, then inserted the row +// in a separate statement that could fail — and the writer now explicitly +// unlinks the blob on the rollback path, so the window is strictly smaller than +// it was. +// ---------------------------------------------------------------------------- + +/// The four metadata columns a blob write sets. Bundled so the `*_meta_on` +/// helpers take a target (owner, collection, id) plus one payload, rather than a +/// long positional tail it would be easy to transpose. +#[derive(Clone, Copy)] +pub(crate) struct MetaFields<'a> { + pub name: &'a str, + pub content_type: &'a str, + pub size: i64, + pub checksum: &'a str, +} + +/// The same, for an update: `name`/`content_type` are `None` to keep the stored +/// value (`COALESCE`), while the bytes always change. +#[derive(Clone, Copy)] +pub(crate) struct MetaPatch<'a> { + pub name: Option<&'a str>, + pub content_type: Option<&'a str>, + pub size: i64, + pub checksum: &'a str, +} + +const FILE_COLS: &str = "id, collection, name, content_type, size_bytes, \ + checksum_sha256, created_at, updated_at"; + +pub(crate) async fn head_on<'c, E>( + exec: E, + app_id: AppId, + collection: &str, + id: Uuid, +) -> Result, FilesRepoError> +where + E: sqlx::PgExecutor<'c>, +{ + let row: Option = sqlx::query_as(&format!( + "SELECT {FILE_COLS} FROM files \ + WHERE app_id = $1 AND collection = $2 AND id = $3" + )) + .bind(app_id.into_inner()) + .bind(collection) + .bind(id) + .fetch_optional(exec) + .await?; + Ok(row.map(FileRow::into_meta)) +} + +pub(crate) async fn insert_meta_on<'c, E>( + exec: E, + app_id: AppId, + collection: &str, + id: Uuid, + meta: MetaFields<'_>, +) -> Result +where + E: sqlx::PgExecutor<'c>, +{ + let row: FileRow = sqlx::query_as(&format!( + "INSERT INTO files \ + (app_id, collection, id, name, content_type, size_bytes, checksum_sha256) \ + VALUES ($1, $2, $3, $4, $5, $6, $7) \ + RETURNING {FILE_COLS}" + )) + .bind(app_id.into_inner()) + .bind(collection) + .bind(id) + .bind(meta.name) + .bind(meta.content_type) + .bind(meta.size) + .bind(meta.checksum) + .fetch_one(exec) + .await?; + Ok(row.into_meta()) +} + +/// `None` when the file row is gone (the caller maps that to `NotFound`). +pub(crate) async fn update_meta_on<'c, E>( + exec: E, + app_id: AppId, + collection: &str, + id: Uuid, + meta: MetaPatch<'_>, +) -> Result, FilesRepoError> +where + E: sqlx::PgExecutor<'c>, +{ + let row: Option = sqlx::query_as(&format!( + "UPDATE files SET \ + name = COALESCE($4, name), \ + content_type = COALESCE($5, content_type), \ + size_bytes = $6, \ + checksum_sha256 = $7, \ + updated_at = NOW() \ + WHERE app_id = $1 AND collection = $2 AND id = $3 \ + RETURNING {FILE_COLS}" + )) + .bind(app_id.into_inner()) + .bind(collection) + .bind(id) + .bind(meta.name) + .bind(meta.content_type) + .bind(meta.size) + .bind(meta.checksum) + .fetch_optional(exec) + .await?; + Ok(row.map(FileRow::into_meta)) +} + +/// `DELETE ... RETURNING` — one statement, so no `SELECT ... FOR UPDATE` dance. +/// The caller unlinks the bytes AFTER the transaction commits: unlinking first +/// would destroy the blob of a row that a rollback then keeps. +pub(crate) async fn delete_meta_on<'c, E>( + exec: E, + app_id: AppId, + collection: &str, + id: Uuid, +) -> Result, FilesRepoError> +where + E: sqlx::PgExecutor<'c>, +{ + let row: Option = sqlx::query_as(&format!( + "DELETE FROM files \ + WHERE app_id = $1 AND collection = $2 AND id = $3 \ + RETURNING {FILE_COLS}" + )) + .bind(app_id.into_inner()) + .bind(collection) + .bind(id) + .fetch_optional(exec) + .await?; + Ok(row.map(FileRow::into_meta)) +} + +/// Best-effort unlink of a blob whose metadata write was rolled back (or whose +/// row was just deleted). A failure leaves an orphan — logged, never fatal. +pub(crate) fn unlink_blob(root: &Path, owner_rel: &Path, collection: &str, id: Uuid) { + let path = final_path_at(root, owner_rel, collection, id); + if let Err(e) = std::fs::remove_file(&path) { + if e.kind() != std::io::ErrorKind::NotFound { + tracing::warn!( + path = %path.display(), error = %e, + "files: unlink failed (orphan blob left on disk)" + ); + } + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/manager-core/src/files_service.rs b/crates/manager-core/src/files_service.rs index ddc8c42..deab64c 100644 --- a/crates/manager-core/src/files_service.rs +++ b/crates/manager-core/src/files_service.rs @@ -18,17 +18,18 @@ use std::sync::Arc; use async_trait::async_trait; use picloud_shared::{ sanitize_stored_content_type, validate_files_collection, FileMeta, FileUpdate, FilesError, - FilesListPage, FilesService, NewFile, SdkCallCx, ServiceEvent, ServiceEventEmitter, + FilesListPage, FilesService, NewFile, SdkCallCx, ServiceEventEmitter, }; use uuid::Uuid; +use crate::atomic_write::{BestEffortFilesWriter, FilesWriter, PostgresFilesWriter}; use crate::authz::{self, AuthzRepo, Capability}; -use crate::files_repo::{FileUpdated, FilesRepo, FilesRepoError}; +use crate::files_repo::{FilesRepo, FilesRepoError}; pub struct FilesServiceImpl { repo: Arc, authz: Arc, - events: Arc, + writer: Arc, max_file_size_bytes: usize, } @@ -41,13 +42,23 @@ impl FilesServiceImpl { max_file_size_bytes: usize, ) -> Self { Self { + writer: Arc::new(BestEffortFilesWriter::new(repo.clone(), events)), repo, authz, - events, max_file_size_bytes, } } + /// Swap the best-effort writer for the transactional one: the metadata row + /// and the trigger fan-out then commit together, so an outbox failure rolls + /// the metadata back (and unlinks the blob) instead of silently losing the + /// event. The host always calls this; the in-memory unit tests do not. + #[must_use] + pub fn with_atomic_writes(mut self, pool: sqlx::PgPool, root: std::path::PathBuf) -> Self { + self.writer = Arc::new(PostgresFilesWriter::new(pool, root)); + self + } + async fn check_read(&self, cx: &SdkCallCx) -> Result<(), FilesError> { authz::script_gate( &*self.authz, @@ -69,37 +80,6 @@ impl FilesServiceImpl { ) .await } - - /// Best-effort `ServiceEvent` emission. A failed emit is logged but - /// never rolls back the (already-durable) file write. - async fn emit( - &self, - cx: &SdkCallCx, - op: &'static str, - collection: &str, - meta: &FileMeta, - old: Option<&FileMeta>, - ) { - let payload = serde_json::to_value(meta).ok(); - let old_payload = old.and_then(|m| serde_json::to_value(m).ok()); - if let Err(e) = self - .events - .emit( - cx, - ServiceEvent { - source: "files", - op, - collection: Some(collection.to_string()), - key: Some(meta.id.to_string()), - payload, - old_payload, - }, - ) - .await - { - tracing::error!(error = %e, source = "files", op, event_emit_failure = true, "event emit failed"); - } - } } /// Parse a script-supplied id. Invalid UUIDs aren't an error shape the @@ -132,9 +112,7 @@ impl FilesService for FilesServiceImpl { // Audit 2026-06-11 C-2 — coerce dangerous render types to // application/octet-stream after the shape checks pass. new.content_type = sanitize_stored_content_type(&new.content_type); - let meta = self.repo.create(cx.app_id, collection, new).await?; - self.emit(cx, "create", collection, &meta, None).await; - Ok(meta.id) + self.writer.create(cx, collection, new).await } async fn head( @@ -182,12 +160,10 @@ impl FilesService for FilesServiceImpl { let Some(uuid) = parse_id(id) else { return Err(FilesError::NotFound); }; - match self.repo.update(cx.app_id, collection, uuid, upd).await? { - Some(FileUpdated { new, prev }) => { - self.emit(cx, "update", collection, &new, Some(&prev)).await; - Ok(()) - } - None => Err(FilesError::NotFound), + if self.writer.update(cx, collection, uuid, upd).await? { + Ok(()) + } else { + Err(FilesError::NotFound) } } @@ -197,16 +173,7 @@ impl FilesService for FilesServiceImpl { let Some(uuid) = parse_id(id) else { return Ok(false); }; - match self.repo.delete(cx.app_id, collection, uuid).await? { - Some(meta) => { - // On delete, the top-level metadata AND `prev` both carry - // the deleted row (per docs/v1.1.x design + the brief). - self.emit(cx, "delete", collection, &meta, Some(&meta)) - .await; - Ok(true) - } - None => Ok(false), - } + self.writer.delete(cx, collection, uuid).await } async fn list( @@ -232,6 +199,7 @@ impl FilesService for FilesServiceImpl { mod tests { use super::*; use crate::authz::{AuthzError, AuthzRepo}; + use crate::files_repo::FileUpdated; use async_trait::async_trait; use chrono::Utc; use picloud_shared::{ diff --git a/crates/manager-core/src/group_collection_repo.rs b/crates/manager-core/src/group_collection_repo.rs index 0667369..893e085 100644 --- a/crates/manager-core/src/group_collection_repo.rs +++ b/crates/manager-core/src/group_collection_repo.rs @@ -13,29 +13,10 @@ //! name does not resolve. use async_trait::async_trait; -use picloud_shared::{AppId, GroupId, ScriptOwner, SdkCallCx, ServiceEvent, ServiceEventEmitter}; +use picloud_shared::{AppId, GroupId, ScriptOwner}; use sqlx::{PgPool, Postgres, Transaction}; use crate::config_resolver::CHAIN_LEVELS_CTE; - -/// §11.6: fire a `shared = true` trigger event on the owning group, best-effort. -/// An emit failure is logged (with the event's `source`/`op`) and swallowed — -/// the write already committed, and shared triggers are fire-and-forget, so a -/// dropped emit must never fail the caller. The single home for the tail the -/// group KV/docs/files services previously each copied; each still builds its -/// own `ServiceEvent` (payload shapes differ) and calls this. -pub(crate) async fn best_effort_emit_shared( - events: &dyn ServiceEventEmitter, - cx: &SdkCallCx, - group_id: GroupId, - event: ServiceEvent, -) { - let (source, op) = (event.source, event.op); - if let Err(e) = events.emit_shared(cx, group_id, event).await { - tracing::error!(error = %e, source, op, event_emit_failure = true, "shared event emit failed"); - } -} - /// List **all** shared-collection declarations at `owner` as `(name, kind)` /// pairs, sorted. Used by the kind-aware apply diff and `collections ls`. pub async fn list_all_for_owner( diff --git a/crates/manager-core/src/group_files_repo.rs b/crates/manager-core/src/group_files_repo.rs index 2fbddf4..063643e 100644 --- a/crates/manager-core/src/group_files_repo.rs +++ b/crates/manager-core/src/group_files_repo.rs @@ -389,7 +389,7 @@ impl GroupFilesRepo for FsGroupFilesRepo { } #[derive(sqlx::FromRow)] -struct GroupFileRow { +pub(crate) struct GroupFileRow { id: Uuid, collection: String, name: String, @@ -401,7 +401,7 @@ struct GroupFileRow { } impl GroupFileRow { - fn into_meta(self) -> FileMeta { + pub(crate) fn into_meta(self) -> FileMeta { FileMeta { id: self.id, collection: self.collection, @@ -414,3 +414,146 @@ impl GroupFileRow { } } } +// ---------------------------------------------------------------------------- +// Connection-scoped metadata operations (see `files_repo` for the rationale) +// ---------------------------------------------------------------------------- + +use crate::files_repo::{MetaFields, MetaPatch}; + +const GROUP_FILE_COLS: &str = "id, collection, name, content_type, size_bytes, \ + checksum_sha256, created_at, updated_at"; + +pub(crate) async fn head_on<'c, E>( + exec: E, + group_id: GroupId, + collection: &str, + id: Uuid, +) -> Result, GroupFilesRepoError> +where + E: sqlx::PgExecutor<'c>, +{ + let row: Option = sqlx::query_as(&format!( + "SELECT {GROUP_FILE_COLS} FROM group_files \ + WHERE group_id = $1 AND collection = $2 AND id = $3" + )) + .bind(group_id.into_inner()) + .bind(collection) + .bind(id) + .fetch_optional(exec) + .await?; + Ok(row.map(GroupFileRow::into_meta)) +} + +pub(crate) async fn insert_meta_on<'c, E>( + exec: E, + group_id: GroupId, + collection: &str, + id: Uuid, + meta: MetaFields<'_>, +) -> Result +where + E: sqlx::PgExecutor<'c>, +{ + let row: GroupFileRow = sqlx::query_as(&format!( + "INSERT INTO group_files \ + (group_id, collection, id, name, content_type, size_bytes, checksum_sha256) \ + VALUES ($1, $2, $3, $4, $5, $6, $7) \ + RETURNING {GROUP_FILE_COLS}" + )) + .bind(group_id.into_inner()) + .bind(collection) + .bind(id) + .bind(meta.name) + .bind(meta.content_type) + .bind(meta.size) + .bind(meta.checksum) + .fetch_one(exec) + .await?; + Ok(row.into_meta()) +} + +pub(crate) async fn update_meta_on<'c, E>( + exec: E, + group_id: GroupId, + collection: &str, + id: Uuid, + meta: MetaPatch<'_>, +) -> Result, GroupFilesRepoError> +where + E: sqlx::PgExecutor<'c>, +{ + let row: Option = sqlx::query_as(&format!( + "UPDATE group_files SET \ + name = COALESCE($4, name), \ + content_type = COALESCE($5, content_type), \ + size_bytes = $6, \ + checksum_sha256 = $7, \ + updated_at = NOW() \ + WHERE group_id = $1 AND collection = $2 AND id = $3 \ + RETURNING {GROUP_FILE_COLS}" + )) + .bind(group_id.into_inner()) + .bind(collection) + .bind(id) + .bind(meta.name) + .bind(meta.content_type) + .bind(meta.size) + .bind(meta.checksum) + .fetch_optional(exec) + .await?; + Ok(row.map(GroupFileRow::into_meta)) +} + +pub(crate) async fn delete_meta_on<'c, E>( + exec: E, + group_id: GroupId, + collection: &str, + id: Uuid, +) -> Result, GroupFilesRepoError> +where + E: sqlx::PgExecutor<'c>, +{ + let row: Option = sqlx::query_as(&format!( + "DELETE FROM group_files \ + WHERE group_id = $1 AND collection = $2 AND id = $3 \ + RETURNING {GROUP_FILE_COLS}" + )) + .bind(group_id.into_inner()) + .bind(collection) + .bind(id) + .fetch_optional(exec) + .await?; + Ok(row.map(GroupFileRow::into_meta)) +} + +/// §11.6 quota: the PROJECTED total stored bytes for the group AFTER this write +/// — the current SUM, minus the bytes of the file being replaced (`replacing = +/// Some(id)` on update; `None` on create), plus the incoming file's bytes. +/// +/// The subtraction is what `GroupFilesService::update` was missing entirely: it +/// checked no quota at all, so a 1-byte file could be updated to a 100 MB one +/// without ever consulting the ceiling. +pub(crate) async fn projected_total_bytes_on<'c, E>( + exec: E, + group_id: GroupId, + replacing: Option, + incoming: i64, +) -> Result +where + E: sqlx::PgExecutor<'c>, +{ + let (n,): (i64,) = sqlx::query_as( + "SELECT ( \ + COALESCE((SELECT SUM(size_bytes) FROM group_files WHERE group_id = $1), 0) \ + - COALESCE((SELECT size_bytes FROM group_files \ + WHERE group_id = $1 AND id = $2), 0) \ + + $3 \ + )::BIGINT", + ) + .bind(group_id.into_inner()) + .bind(replacing) + .bind(incoming) + .fetch_one(exec) + .await?; + Ok(u64::try_from(n).unwrap_or(0)) +} diff --git a/crates/manager-core/src/group_files_service.rs b/crates/manager-core/src/group_files_service.rs index 793d0a8..92e1957 100644 --- a/crates/manager-core/src/group_files_service.rs +++ b/crates/manager-core/src/group_files_service.rs @@ -13,12 +13,14 @@ use async_trait::async_trait; use picloud_shared::{ sanitize_stored_content_type, validate_files_collection, FileMeta, FileUpdate, FilesListPage, GroupFilesError, GroupFilesService, GroupId, NewFile, NoopEventEmitter, SdkCallCx, - ServiceEvent, ServiceEventEmitter, + ServiceEventEmitter, }; use uuid::Uuid; +use crate::atomic_write::{ + BestEffortGroupFilesWriter, GroupFilesQuota, GroupFilesWriter, PostgresGroupFilesWriter, +}; use crate::authz::{self, AuthzRepo, Capability}; -use crate::files_repo::FileUpdated; use crate::group_collection_repo::GroupCollectionResolver; use crate::group_files_repo::{GroupFilesRepo, GroupFilesRepoError}; @@ -34,7 +36,10 @@ pub struct GroupFilesServiceImpl { /// shared-files collections (`PICLOUD_GROUP_FILES_MAX_TOTAL_BYTES`). max_total_bytes: u64, /// §11.6: fires `shared = true` files triggers on a shared-collection write. - events: Arc, + /// Mutations: the byte-quota decision, the blob + metadata write, and the + /// shared-trigger fan-out — for the host, all under one advisory-locked + /// transaction. Reads stay on `repo`. + writer: Arc, } impl GroupFilesServiceImpl { @@ -46,52 +51,44 @@ impl GroupFilesServiceImpl { max_file_size_bytes: usize, ) -> Self { Self { + max_total_bytes: crate::group_quota::group_files_max_total_bytes_from_env(), + writer: Arc::new(BestEffortGroupFilesWriter::new( + repo.clone(), + Arc::new(NoopEventEmitter), + )), repo, resolver, authz, max_file_size_bytes, - max_total_bytes: crate::group_quota::group_files_max_total_bytes_from_env(), - events: Arc::new(NoopEventEmitter), } } - /// Wire the event emitter (§11.6 shared-collection triggers). + /// The ceiling a shared-files write must fit under. + fn quota(&self) -> GroupFilesQuota { + GroupFilesQuota { + max_total_bytes: self.max_total_bytes, + } + } + + /// Wire the event emitter (§11.6 shared-collection triggers) on the + /// best-effort writer. Superseded by [`Self::with_atomic_writes`]. #[must_use] pub fn with_events(mut self, events: Arc) -> Self { - self.events = events; + self.writer = Arc::new(BestEffortGroupFilesWriter::new(self.repo.clone(), events)); self } - /// §11.6: fire `shared = true` files triggers on the owning group. - /// Best-effort; an emit failure is logged, never surfaced. - async fn emit_shared( - &self, - cx: &SdkCallCx, - group_id: GroupId, - op: &'static str, - collection: &str, - meta: &FileMeta, - old: Option<&FileMeta>, - ) { - crate::group_collection_repo::best_effort_emit_shared( - &*self.events, - cx, - group_id, - ServiceEvent { - source: "files", - op, - collection: Some(collection.to_string()), - key: Some(meta.id.to_string()), - payload: serde_json::to_value(meta).ok(), - old_payload: old.and_then(|m| serde_json::to_value(m).ok()), - }, - ) - .await; + /// Use the transactional writer: the byte-quota check, the metadata row, and + /// the shared-trigger fan-out run on ONE connection under a per-group + /// advisory lock, so concurrent uploads can't each see the same pre-write + /// total and together overshoot the ceiling. Supersedes + /// [`Self::with_events`]. + #[must_use] + pub fn with_atomic_writes(mut self, pool: sqlx::PgPool, root: std::path::PathBuf) -> Self { + self.writer = Arc::new(PostgresGroupFilesWriter::new(pool, root)); + self } - /// The structural boundary: resolve the collection to its owning group - /// (kind=`files`) on the calling app's chain. `CollectionNotShared` when no - /// ancestor group declares it. async fn owning_group( &self, cx: &SdkCallCx, @@ -160,19 +157,9 @@ impl GroupFilesService for GroupFilesServiceImpl { // shape checks pass (same as app files, audit 2026-06-11 C-2). new.content_type = sanitize_stored_content_type(&new.content_type); self.check_write(cx, group_id).await?; - // §11.6 quota: the new file's bytes must fit under the group's total. - let used = self.repo.total_bytes(group_id).await?; - let incoming = new.data.len() as u64; - if used.saturating_add(incoming) > self.max_total_bytes { - return Err(GroupFilesError::QuotaExceeded { - limit: self.max_total_bytes, - actual: used.saturating_add(incoming), - }); - } - let meta = self.repo.create(group_id, collection, new).await?; - self.emit_shared(cx, group_id, "create", collection, &meta, None) - .await; - Ok(meta.id) + self.writer + .create(cx, group_id, collection, new, self.quota()) + .await } async fn head( @@ -222,13 +209,14 @@ impl GroupFilesService for GroupFilesServiceImpl { let Some(uuid) = parse_id(id) else { return Err(GroupFilesError::NotFound); }; - match self.repo.update(group_id, collection, uuid, upd).await? { - Some(FileUpdated { new, prev }) => { - self.emit_shared(cx, group_id, "update", collection, &new, Some(&prev)) - .await; - Ok(()) - } - None => Err(GroupFilesError::NotFound), + if self + .writer + .update(cx, group_id, collection, uuid, upd, self.quota()) + .await? + { + Ok(()) + } else { + Err(GroupFilesError::NotFound) } } @@ -244,14 +232,7 @@ impl GroupFilesService for GroupFilesServiceImpl { let Some(uuid) = parse_id(id) else { return Ok(false); }; - match self.repo.delete(group_id, collection, uuid).await? { - Some(meta) => { - self.emit_shared(cx, group_id, "delete", collection, &meta, Some(&meta)) - .await; - Ok(true) - } - None => Ok(false), - } + self.writer.delete(cx, group_id, collection, uuid).await } async fn list( @@ -278,6 +259,7 @@ impl GroupFilesService for GroupFilesServiceImpl { mod tests { use super::*; use crate::authz::{AuthzError, AuthzRepo}; + use crate::files_repo::FileUpdated; use crate::group_files_repo::GroupFilesRepoError; use chrono::Utc; use picloud_shared::{ diff --git a/crates/manager-core/tests/atomic_write.rs b/crates/manager-core/tests/atomic_write.rs index 60d0027..bae6d95 100644 --- a/crates/manager-core/tests/atomic_write.rs +++ b/crates/manager-core/tests/atomic_write.rs @@ -16,12 +16,14 @@ use std::sync::Arc; use picloud_manager_core::atomic_write::{ - GroupDocsTarget, GroupDocsWriter, GroupKvTarget, GroupKvWriter, KvWriter, - PostgresGroupDocsWriter, PostgresGroupKvWriter, PostgresKvWriter, + GroupDocsTarget, GroupDocsWriter, GroupFilesQuota, GroupFilesWriter, GroupKvTarget, + GroupKvWriter, KvWriter, PostgresGroupDocsWriter, PostgresGroupFilesWriter, + PostgresGroupKvWriter, PostgresKvWriter, }; use picloud_manager_core::group_quota::GroupWriteQuota; use picloud_shared::{ - AppId, ExecutionId, GroupDocsError, GroupId, GroupKvError, RequestId, ScriptId, SdkCallCx, + AppId, ExecutionId, FileUpdate, GroupDocsError, GroupFilesError, GroupId, GroupKvError, + NewFile, RequestId, ScriptId, SdkCallCx, }; use sqlx::postgres::PgPoolOptions; use sqlx::PgPool; @@ -503,3 +505,106 @@ async fn concurrent_writers_cannot_push_a_group_past_its_docs_row_quota() { .await .expect("cleanup"); } + +/// Group FILES had the worst version of the race: the ceiling is disk (10 GiB by +/// default) and a single file may be 100 MB, so a fleet of concurrent uploads +/// each seeing the same pre-write total could overshoot by GIGABYTES of real +/// disk. Same fix — a per-group advisory lock across the check and the write. +/// +/// It also pins the second half of the bug: `update` checked NO quota at all, so +/// a 1-byte file could be grown past the ceiling unchallenged. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn concurrent_uploads_cannot_push_a_group_past_its_files_byte_quota() { + let Some(pool) = pool_or_skip().await else { + return; + }; + let group = mk_group(&pool).await; + let root = std::env::temp_dir().join(format!("picloud-aw-{}", Uuid::new_v4().simple())); + let writer = Arc::new(PostgresGroupFilesWriter::new(pool.clone(), root.clone())); + + // Ten 100-byte uploads race for a 450-byte ceiling: at most 4 may land. + let quota = GroupFilesQuota { + max_total_bytes: 450, + }; + let app = Uuid::new_v4(); + let mut set = tokio::task::JoinSet::new(); + for i in 0..10 { + let w = Arc::clone(&writer); + set.spawn(async move { + let cx = cx(app); + w.create( + &cx, + GroupId::from(group), + "assets", + NewFile { + name: format!("f{i}.bin"), + content_type: "application/octet-stream".into(), + data: vec![b'x'; 100], + }, + quota, + ) + .await + }); + } + let mut created = Vec::new(); + while let Some(r) = set.join_next().await { + match r.expect("task") { + Ok(id) => created.push(id), + Err(GroupFilesError::QuotaExceeded { .. }) => {} + Err(e) => panic!("unexpected error: {e}"), + } + } + + let stored = |pool: PgPool| async move { + let (n,): (i64,) = sqlx::query_as( + "SELECT COALESCE(SUM(size_bytes), 0)::BIGINT FROM group_files WHERE group_id = $1", + ) + .bind(group) + .fetch_one(&pool) + .await + .expect("bytes"); + n + }; + let bytes = stored(pool.clone()).await; + assert!( + bytes <= 450, + "stored bytes ({bytes}) must not exceed the 450-byte ceiling — concurrent \ + uploads must not each see the same pre-write total" + ); + assert!(!created.is_empty(), "some uploads should have succeeded"); + + // An UPDATE that would blow the ceiling must now be refused too. Previously + // update consulted no quota at all, so this was a free bypass. + let victim = created[0]; + let err = writer + .update( + &cx(app), + GroupId::from(group), + "assets", + victim, + FileUpdate { + name: None, + content_type: None, + data: vec![b'y'; 10_000], + }, + quota, + ) + .await + .expect_err("growing a file past the group ceiling must be refused"); + assert!( + matches!(err, GroupFilesError::QuotaExceeded { .. }), + "got {err}" + ); + assert_eq!( + stored(pool.clone()).await, + bytes, + "the refused update must not have changed the stored total" + ); + + sqlx::query("DELETE FROM groups WHERE id = $1") + .bind(group) + .execute(&pool) + .await + .expect("cleanup"); + let _ = std::fs::remove_dir_all(&root); +} diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index 0570bba..c1b18dd 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -247,12 +247,16 @@ pub async fn build_app( // Kept for the v1.1.6 orphan sweeper (cleans stale `*.tmp.*` files). let files_root = files_config.root.clone(); let files_repo = Arc::new(FsFilesRepo::new(pool.clone(), files_config)); - let files: Arc = Arc::new(FilesServiceImpl::new( - files_repo.clone(), - authz.clone(), - events.clone(), - files_max_size, - )); + let files: Arc = Arc::new( + FilesServiceImpl::new( + files_repo.clone(), + authz.clone(), + events.clone(), + files_max_size, + ) + // Metadata row + trigger fan-out in one tx; a rollback unlinks the blob. + .with_atomic_writes(pool.clone(), files_root.clone()), + ); // §11.6 shared group collections (files): group-keyed blobs under // `/files/groups//...`, resolved from cx.app_id's chain. // Shares the per-file size cap; fires `shared = true` triggers under the @@ -265,7 +269,10 @@ pub async fn build_app( authz.clone(), files_max_size, ) - .with_events(events.clone()), + // Byte quota + metadata + shared fan-out in one advisory-locked tx: a + // fleet of concurrent uploads can no longer each see the same pre-write + // total and together overshoot the group's disk ceiling. + .with_atomic_writes(pool.clone(), files_root.clone()), ); // v1.1.6 realtime: the in-process broadcaster is shared between the // publish path (PubsubServiceImpl fans out to SSE subscribers after diff --git a/crates/picloud/tests/queue_e2e.rs b/crates/picloud/tests/queue_e2e.rs index 90d9547..17d369e 100644 --- a/crates/picloud/tests/queue_e2e.rs +++ b/crates/picloud/tests/queue_e2e.rs @@ -108,6 +108,19 @@ const THROW_HANDLER: &str = r#" throw "boom" "#; +/// Poll until the queue depth reaches `want` (or give up and return what we saw). +async fn poll_queue_depth(pool: &PgPool, app_id: &str, queue: &str, want: i64) -> i64 { + let mut depth = -1; + for _ in 0..200 { + depth = count_queue_messages(pool, app_id, queue).await; + if depth == want { + return depth; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + depth +} + async fn poll_marker(pool: &PgPool, app_id: &str) -> Option { for _ in 0..200 { let row: Option<(Value,)> = sqlx::query_as( @@ -192,8 +205,10 @@ async fn queue_receive_acks_on_success() { assert_eq!(marker["queue"]["queue_name"], "jobs"); assert_eq!(marker["queue"]["message"]["x"], 42); - // Ack deleted the row. - assert_eq!(count_queue_messages(&pool, &app_id, "jobs").await, 0); + // Ack deleted the row. The marker is written DURING the handler, but the ack + // happens after it returns — so poll rather than assert on the instant the + // marker appears (that window is real, and asserting into it is flaky). + assert_eq!(poll_queue_depth(&pool, &app_id, "jobs", 0).await, 0); } #[tokio::test(flavor = "multi_thread", worker_threads = 4)]