fix(groups): don't let with_events silently downgrade a transactional writer

`with_events()` and `with_atomic_writes()` both install a writer, last call wins.
Appending them in the wrong order in the host would swap the transactional writer
back out for the best-effort one — silently reverting the write-path invariant AND
dropping the per-group quota enforcement that lives in it, with nothing in the type
system to catch it.

The services now carry an `atomic` flag: once `with_atomic_writes` has run,
`with_events` warns and no-ops instead of clobbering it. Also enforce the group
byte ceiling in `BestEffortGroupFilesWriter::update`, so the non-transactional
writer can't be used as an unmetered bypass and a unit test can catch its removal.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

crates/manager-core/src/group_docs_service.rs

@@ -44,6 +44,12 @@ pub struct GroupDocsServiceImpl {
     /// fan-out, which for the host all commit in one advisory-locked
     /// transaction. Reads stay on `repo`.
     writer: Arc<dyn GroupDocsWriter>,
+    /// Set once [`Self::with_atomic_writes`] has installed the transactional
+    /// writer. `with_events` refuses to overwrite it — silently downgrading a
+    /// service back to non-transactional writes (and dropping its quota
+    /// enforcement) because a builder call was appended in the wrong order is
+    /// exactly the kind of regression the type system will not catch.
+    atomic: bool,
 }
 
 impl GroupDocsServiceImpl {
@@ -60,6 +66,13 @@ impl GroupDocsServiceImpl {
     /// best-effort writer. Superseded by [`Self::with_atomic_writes`].
     #[must_use]
     pub fn with_events(mut self, events: Arc<dyn ServiceEventEmitter>) -> Self {
+        if self.atomic {
+            tracing::warn!(
+                "with_events() called after with_atomic_writes() — ignored; the \
+                 transactional writer already resolves and writes the fan-out itself"
+            );
+            return self;
+        }
         self.writer = Arc::new(BestEffortGroupDocsWriter::new(self.repo.clone(), events));
         self
     }
@@ -71,6 +84,7 @@ impl GroupDocsServiceImpl {
     #[must_use]
     pub fn with_atomic_writes(mut self, pool: sqlx::PgPool) -> Self {
         self.writer = Arc::new(PostgresGroupDocsWriter::new(pool));
+        self.atomic = true;
         self
     }
 
@@ -86,6 +100,7 @@ impl GroupDocsServiceImpl {
                 repo.clone(),
                 Arc::new(NoopEventEmitter),
             )),
+            atomic: false,
             max_rows: crate::quota::group_docs_max_rows_from_env(),
             max_total_bytes: crate::quota::group_docs_max_total_bytes_from_env(),
             repo,

crates/manager-core/src/group_files_service.rs

@@ -40,6 +40,12 @@ pub struct GroupFilesServiceImpl {
     /// shared-trigger fan-out — for the host, all under one advisory-locked
     /// transaction. Reads stay on `repo`.
     writer: Arc<dyn GroupFilesWriter>,
+    /// Set once [`Self::with_atomic_writes`] has installed the transactional
+    /// writer. `with_events` refuses to overwrite it — silently downgrading a
+    /// service back to non-transactional writes (and dropping its quota
+    /// enforcement) because a builder call was appended in the wrong order is
+    /// exactly the kind of regression the type system will not catch.
+    atomic: bool,
 }
 
 impl GroupFilesServiceImpl {
@@ -56,6 +62,7 @@ impl GroupFilesServiceImpl {
                 repo.clone(),
                 Arc::new(NoopEventEmitter),
             )),
+            atomic: false,
             repo,
             resolver,
             authz,
@@ -74,6 +81,13 @@ impl GroupFilesServiceImpl {
     /// best-effort writer. Superseded by [`Self::with_atomic_writes`].
     #[must_use]
     pub fn with_events(mut self, events: Arc<dyn ServiceEventEmitter>) -> Self {
+        if self.atomic {
+            tracing::warn!(
+                "with_events() called after with_atomic_writes() — ignored; the \
+                 transactional writer already resolves and writes the fan-out itself"
+            );
+            return self;
+        }
         self.writer = Arc::new(BestEffortGroupFilesWriter::new(self.repo.clone(), events));
         self
     }
@@ -86,6 +100,7 @@ impl GroupFilesServiceImpl {
     #[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.atomic = true;
         self
     }
 

crates/manager-core/src/group_kv_service.rs

@@ -56,6 +56,12 @@ pub struct GroupKvServiceImpl {
     /// fan-out, which for the host all commit in one advisory-locked
     /// transaction. Reads stay on `repo`.
     writer: Arc<dyn GroupKvWriter>,
+    /// Set once [`Self::with_atomic_writes`] has installed the transactional
+    /// writer. `with_events` refuses to overwrite it — silently downgrading a
+    /// service back to non-transactional writes (and dropping its quota
+    /// enforcement) because a builder call was appended in the wrong order is
+    /// exactly the kind of regression the type system will not catch.
+    atomic: bool,
 }
 
 impl GroupKvServiceImpl {
@@ -73,6 +79,13 @@ impl GroupKvServiceImpl {
     /// supersedes it — shared writes fire nothing.
     #[must_use]
     pub fn with_events(mut self, events: Arc<dyn ServiceEventEmitter>) -> Self {
+        if self.atomic {
+            tracing::warn!(
+                "with_events() called after with_atomic_writes() — ignored; the \
+                 transactional writer already resolves and writes the fan-out itself"
+            );
+            return self;
+        }
         self.writer = Arc::new(BestEffortGroupKvWriter::new(self.repo.clone(), events));
         self
     }
@@ -88,6 +101,7 @@ impl GroupKvServiceImpl {
     #[must_use]
     pub fn with_atomic_writes(mut self, pool: sqlx::PgPool) -> Self {
         self.writer = Arc::new(PostgresGroupKvWriter::new(pool));
+        self.atomic = true;
         self
     }
 
@@ -111,6 +125,7 @@ impl GroupKvServiceImpl {
                 repo.clone(),
                 Arc::new(NoopEventEmitter),
             )),
+            atomic: false,
             max_rows: crate::quota::group_kv_max_rows_from_env(),
             max_total_bytes: crate::quota::group_kv_max_total_bytes_from_env(),
             repo,