From 8f35050499eb345662ac87b5b2c822d73a4862bd Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Tue, 14 Jul 2026 21:12:01 +0200 Subject: [PATCH] fix(groups): don't let with_events silently downgrade a transactional writer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `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) --- crates/manager-core/src/group_docs_service.rs | 15 +++++++++++++++ crates/manager-core/src/group_files_service.rs | 15 +++++++++++++++ crates/manager-core/src/group_kv_service.rs | 15 +++++++++++++++ 3 files changed, 45 insertions(+) diff --git a/crates/manager-core/src/group_docs_service.rs b/crates/manager-core/src/group_docs_service.rs index 2ab4d6e..872e488 100644 --- a/crates/manager-core/src/group_docs_service.rs +++ b/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, + /// 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) -> 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, diff --git a/crates/manager-core/src/group_files_service.rs b/crates/manager-core/src/group_files_service.rs index 34784ac..4dea0af 100644 --- a/crates/manager-core/src/group_files_service.rs +++ b/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, + /// 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) -> 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 } diff --git a/crates/manager-core/src/group_kv_service.rs b/crates/manager-core/src/group_kv_service.rs index ecac694..99c87a1 100644 --- a/crates/manager-core/src/group_kv_service.rs +++ b/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, + /// 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) -> 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,