fix(files): make file writes transactional and close a quota bypass

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) <noreply@anthropic.com>

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<Option<FileMeta>, GroupFilesRepoError>
+where
+    E: sqlx::PgExecutor<'c>,
+{
+    let row: Option<GroupFileRow> = 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<FileMeta, GroupFilesRepoError>
+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<Option<FileMeta>, GroupFilesRepoError>
+where
+    E: sqlx::PgExecutor<'c>,
+{
+    let row: Option<GroupFileRow> = 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<Option<FileMeta>, GroupFilesRepoError>
+where
+    E: sqlx::PgExecutor<'c>,
+{
+    let row: Option<GroupFileRow> = 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<Uuid>,
+    incoming: i64,
+) -> Result<u64, GroupFilesRepoError>
+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))
+}