From bbb9744010573fc0d7dbe10a6158f45d849b2119 Mon Sep 17 00:00:00 2001 From: Nikolaj Goodger Date: Mon, 19 Oct 2026 13:29:32 +1100 Subject: [PATCH 1/2] feat(relay): record who uploads each media blob Add a community-scoped `media_uploaders` table (migration 0067, write-fenced and purged with the community) and record the uploader on every successful upload, including deduplicated re-uploads. Uploading the exact bytes proves possession, which channel-scoped media reads (block/buzz#8444) will use to admit an uploader even when the same bytes are attached somewhere they cannot read. Nothing reads the table yet. Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Nikolaj Goodger --- crates/buzz-db/src/runtime/migration.rs | 16 +++- crates/buzz-db/src/store/deletion.rs | 3 + crates/buzz-db/src/store/media_uploader.rs | 97 ++++++++++++++++++++++ crates/buzz-db/src/store/mod.rs | 2 + crates/buzz-relay/src/api/media.rs | 22 +++-- migrations/0067_media_uploaders.sql | 14 ++++ schema/schema.sql | 2 + schema/tables/public/media_uploaders.sql | 11 +++ schema/triggers/media_uploaders.sql | 1 + 9 files changed, 162 insertions(+), 6 deletions(-) create mode 100644 crates/buzz-db/src/store/media_uploader.rs create mode 100644 migrations/0067_media_uploaders.sql create mode 100644 schema/tables/public/media_uploaders.sql create mode 100644 schema/triggers/media_uploaders.sql diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 7cb0879ff87..6f359ecfd5c 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -709,7 +709,7 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 66); + assert_eq!(migrations.len(), 67); assert_eq!(migrations[64].version, 65); let retire_ttl = strip_sql_comments(migrations[64].sql.as_str()); for statement in [ @@ -746,6 +746,11 @@ mod postgres_tests { .sql .as_str() .contains("CREATE TABLE community_nip_fi_issuers")); + assert_eq!(migrations[66].version, 67); + assert!(migrations[66] + .sql + .as_str() + .contains("CREATE TABLE media_uploaders")); assert_eq!(migrations[55].version, 56); assert!(migrations[55] .sql @@ -2317,6 +2322,15 @@ mod postgres_tests { .as_str() .contains("attach_community_write_fence('community_nip_fi_issuers')")); expected_fences.insert("community_nip_fi_issuers".to_owned()); + // 0067 fences the media uploaders table. + assert!(MIGRATOR + .iter() + .find(|m| m.version == 67) + .expect("media uploaders migration") + .sql + .as_str() + .contains("attach_community_write_fence('media_uploaders')")); + expected_fences.insert("media_uploaders".to_owned()); assert_eq!( expected_fences, schema.fence_attachments, "write-fence attachment targets differ after recovery policy" diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 1e1e68cb054..1d910493a84 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -90,6 +90,7 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[ "events", "git_repo_names", "join_policy_acceptances", + "media_uploaders", "moderation_actions", "moderation_reports", "parameterized_event_watermarks", @@ -156,6 +157,7 @@ pub const PURGE_SCOPED_TABLES: &[&str] = &[ "audit_log", "community_bans", "community_nip_fi_issuers", + "media_uploaders", "pubkey_allowlist", "relay_members", "users", @@ -4423,6 +4425,7 @@ DROP TRIGGER IF EXISTS community_write_fence_event_mentions ON event_mentions; DROP TRIGGER IF EXISTS community_write_fence_events ON events; DROP TRIGGER IF EXISTS community_write_fence_git_repo_names ON git_repo_names; DROP TRIGGER IF EXISTS community_write_fence_join_policy_acceptances ON join_policy_acceptances; +DROP TRIGGER IF EXISTS community_write_fence_media_uploaders ON media_uploaders; DROP TRIGGER IF EXISTS community_write_fence_moderation_actions ON moderation_actions; DROP TRIGGER IF EXISTS community_write_fence_moderation_reports ON moderation_reports; DROP TRIGGER IF EXISTS community_write_fence_parameterized_event_watermarks ON parameterized_event_watermarks; diff --git a/crates/buzz-db/src/store/media_uploader.rs b/crates/buzz-db/src/store/media_uploader.rs new file mode 100644 index 00000000000..f4567a43682 --- /dev/null +++ b/crates/buzz-db/src/store/media_uploader.rs @@ -0,0 +1,97 @@ +//! Who has uploaded each blob, as proof of possession for media reads. + +use buzz_datastore_tracing::datastore_span; +use sqlx::PgPool; + +use crate::{error::Result, Db}; +use buzz_core::CommunityId; + +/// Record that `pubkey` uploaded the blob `sha256`; repeat uploads are no-ops. +pub async fn record_media_uploader( + pool: &PgPool, + community_id: CommunityId, + sha256: &str, + pubkey: &[u8], +) -> Result<()> { + sqlx::query( + "INSERT INTO media_uploaders (community_id, sha256, pubkey) VALUES ($1, $2, $3) \ + ON CONFLICT DO NOTHING", + ) + .bind(community_id.as_uuid()) + .bind(sha256) + .bind(pubkey) + .execute(pool) + .await?; + Ok(()) +} + +impl Db { + /// See [`record_media_uploader`]. + #[datastore_span(name = "record_media_uploader", system = "postgresql")] + pub async fn record_media_uploader( + &self, + community_id: CommunityId, + sha256: &str, + pubkey: &[u8], + ) -> Result<()> { + record_media_uploader(&self.pool, community_id, sha256, pubkey).await + } +} + +#[cfg(test)] +mod postgres_tests { + use super::*; + use uuid::Uuid; + + const SHA: &str = "abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789"; + + async fn community(db: &Db) -> CommunityId { + let host = format!("uploader-{}.example", Uuid::new_v4().simple()); + db.ensure_configured_community(&host) + .await + .expect("seed community") + .id + } + + async fn uploader_rows(db: &Db, community_id: CommunityId) -> i64 { + sqlx::query_scalar("SELECT count(*) FROM media_uploaders WHERE community_id = $1") + .bind(community_id.as_uuid()) + .fetch_one(&db.pool) + .await + .expect("count uploaders") + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn repeat_upload_records_one_row() { + let db = Db::from_pool( + PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect to test DB"), + ); + let community_id = community(&db).await; + for _ in 0..2 { + db.record_media_uploader(community_id, SHA, &[7; 32]) + .await + .expect("record uploader"); + } + assert_eq!(uploader_rows(&db, community_id).await, 1); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn deleting_community_refuses_new_uploaders() { + let db = Db::from_pool( + PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect to test DB"), + ); + let community_id = community(&db).await; + crate::test_support::quiesce_community_for_tests(&db.pool, community_id).await; + assert!(db + .record_media_uploader(community_id, SHA, &[7; 32]) + .await + .is_err()); + assert_eq!(uploader_rows(&db, community_id).await, 0); + } +} diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index f759a770a25..ccafad2f810 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -32,6 +32,8 @@ mod event_follow_up_postgres_tests; pub mod feed; /// Git repository name registry (NIP-34 kind:30617). pub mod git_repo; +/// Who has uploaded each blob, as proof of possession for media reads. +pub mod media_uploader; /// Community moderation: reports, bans/timeouts, audit actions. pub mod moderation; /// Deployment-global operator-listener mention registrations and delivery queues. diff --git a/crates/buzz-relay/src/api/media.rs b/crates/buzz-relay/src/api/media.rs index 6864cb15ae2..3182a22512d 100644 --- a/crates/buzz-relay/src/api/media.rs +++ b/crates/buzz-relay/src/api/media.rs @@ -576,6 +576,16 @@ async fn upload_blob_result( auth.tenant.host(), ); + state + .db + .record_media_uploader( + auth.tenant.community(), + &descriptor.sha256, + auth.auth_event.pubkey.as_bytes(), + ) + .await + .map_err(|_| MediaError::Internal)?; + // Normalize MIME to a known set to bound label cardinality. let mime_label = match descriptor.mime_type.as_str() { "image/jpeg" | "image/png" | "image/gif" | "image/webp" | "video/mp4" => { @@ -2135,11 +2145,13 @@ mod tests { const TEST_ISSUER: &str = "https://issuer.example"; const TEST_AUDIENCE: &str = "https://relay.example"; const TEST_KID: &str = "test-key-1"; - const TEST_EC_PKCS8_PEM: &str = "-----BEGIN PRIVATE KEY-----\n\ - MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgcnxDM4EiirH9dHUE\n\ - WZc759TX4s5PAn8kO5ovXSnGxCWhRANCAARFb6ZnsfkqOOXyEhj3KBQphGKF4vTa\n\ - zhebbavbZ1ZoklqkF1cGg+jTO7rONAVEzXvXUWtV6CdDV+rybiVmFP2w\n\ - -----END PRIVATE KEY-----\n"; + const TEST_EC_PKCS8_PEM: &str = concat!( + "-----BEGIN PRIVATE KEY-----\n", // sadscan:disable kingfisher.privkey.2 -- static test-only key + "MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgcnxDM4EiirH9dHUE\n", + "WZc759TX4s5PAn8kO5ovXSnGxCWhRANCAARFb6ZnsfkqOOXyEhj3KBQphGKF4vTa\n", + "zhebbavbZ1ZoklqkF1cGg+jTO7rONAVEzXvXUWtV6CdDV+rybiVmFP2w\n", + "-----END PRIVATE KEY-----\n", + ); /// Build an AppState with NIP-FI Enforce and a real injected P-256 verifier. async fn media_enforce_test_state() -> Option> { diff --git a/migrations/0067_media_uploaders.sql b/migrations/0067_media_uploaders.sql new file mode 100644 index 00000000000..458381b7da3 --- /dev/null +++ b/migrations/0067_media_uploaders.sql @@ -0,0 +1,14 @@ +-- Who has uploaded each blob in a community. Uploading the exact bytes proves +-- possession, so channel-scoped media reads can always admit an uploader even +-- when the same bytes are also attached somewhere they cannot read. +-- +-- Additive. Nothing reads this table yet. +CREATE TABLE media_uploaders ( + community_id UUID NOT NULL REFERENCES communities(id), + sha256 TEXT NOT NULL CHECK (sha256 ~ '^[0-9a-f]{64}$'), + pubkey BYTEA NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (community_id, sha256, pubkey) +); + +SELECT attach_community_write_fence('media_uploaders'); diff --git a/schema/schema.sql b/schema/schema.sql index d252cbdb7ea..47a06c8f178 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -107,6 +107,7 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; \i tables/public/moderation_reports.sql \i tables/public/community_bans.sql \i tables/public/community_nip_fi_issuers.sql +\i tables/public/media_uploaders.sql \i tables/public/_operator_global_tables.sql \i tables/public/git_repo_names.sql \i tables/public/parameterized_event_watermarks.sql @@ -159,6 +160,7 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; \i triggers/event_mentions.sql \i triggers/git_repo_names.sql \i triggers/join_policy_acceptances.sql +\i triggers/media_uploaders.sql \i triggers/moderation_actions.sql \i triggers/moderation_reports.sql \i triggers/parameterized_event_watermarks.sql diff --git a/schema/tables/public/media_uploaders.sql b/schema/tables/public/media_uploaders.sql new file mode 100644 index 00000000000..7a1255ccdd3 --- /dev/null +++ b/schema/tables/public/media_uploaders.sql @@ -0,0 +1,11 @@ +-- ── Media uploaders ─────────────────────────────────────────────────────────── +-- Who has uploaded each blob in a community. Uploading the exact bytes proves +-- possession, so channel-scoped media reads can always admit an uploader. + +CREATE TABLE media_uploaders ( + community_id UUID NOT NULL REFERENCES communities(id), + sha256 TEXT NOT NULL CHECK (sha256 ~ '^[0-9a-f]{64}$'), + pubkey BYTEA NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (community_id, sha256, pubkey) +); diff --git a/schema/triggers/media_uploaders.sql b/schema/triggers/media_uploaders.sql new file mode 100644 index 00000000000..23367202e24 --- /dev/null +++ b/schema/triggers/media_uploaders.sql @@ -0,0 +1 @@ +SELECT attach_community_write_fence('media_uploaders'); From 4474d69f42ea98e774871fa3ec9b631bb8c14cff Mon Sep 17 00:00:00 2001 From: Nikolaj Goodger Date: Mon, 19 Oct 2026 13:36:27 +1100 Subject: [PATCH 2/2] fix(relay): bind uploader records to the upload's serving lease Record the uploader in a transaction admitted with the upload's serving-write lease, like other final writes of an external side effect, so an admitted upload can finish while its community quiesces and nothing lands after the lease is gone. Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Nikolaj Goodger --- crates/buzz-db/src/store/media_uploader.rs | 101 ++++++++++++--------- crates/buzz-relay/src/api/media.rs | 4 +- 2 files changed, 59 insertions(+), 46 deletions(-) diff --git a/crates/buzz-db/src/store/media_uploader.rs b/crates/buzz-db/src/store/media_uploader.rs index f4567a43682..9a720aa6366 100644 --- a/crates/buzz-db/src/store/media_uploader.rs +++ b/crates/buzz-db/src/store/media_uploader.rs @@ -1,56 +1,76 @@ //! Who has uploaded each blob, as proof of possession for media reads. use buzz_datastore_tracing::datastore_span; -use sqlx::PgPool; -use crate::{error::Result, Db}; -use buzz_core::CommunityId; - -/// Record that `pubkey` uploaded the blob `sha256`; repeat uploads are no-ops. -pub async fn record_media_uploader( - pool: &PgPool, - community_id: CommunityId, - sha256: &str, - pubkey: &[u8], -) -> Result<()> { - sqlx::query( - "INSERT INTO media_uploaders (community_id, sha256, pubkey) VALUES ($1, $2, $3) \ - ON CONFLICT DO NOTHING", - ) - .bind(community_id.as_uuid()) - .bind(sha256) - .bind(pubkey) - .execute(pool) - .await?; - Ok(()) -} +use crate::deletion::ServingWriteLease; +use crate::{error::Result, observability, Db}; impl Db { - /// See [`record_media_uploader`]. + /// Record that `pubkey` uploaded the blob `sha256`; repeat uploads are + /// no-ops. The write presents the upload's serving lease, so it may finish + /// while the community quiesces but never after the lease is gone. #[datastore_span(name = "record_media_uploader", system = "postgresql")] pub async fn record_media_uploader( &self, - community_id: CommunityId, + lease: &ServingWriteLease, sha256: &str, pubkey: &[u8], ) -> Result<()> { - record_media_uploader(&self.pool, community_id, sha256, pubkey).await + let connection = + observability::acquire_writer(&self.pool, observability::WriterOperation::EventWrite) + .await?; + let mut tx = sqlx::Transaction::begin(connection, None).await?; + self.deletion_store() + .guard_transaction_with_serving_lease(&mut tx, lease) + .await?; + sqlx::query( + "INSERT INTO media_uploaders (community_id, sha256, pubkey) VALUES ($1, $2, $3) \ + ON CONFLICT DO NOTHING", + ) + .bind(lease.community_id.as_uuid()) + .bind(sha256) + .bind(pubkey) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(()) } } #[cfg(test)] mod postgres_tests { + use std::time::Duration; + use super::*; + use buzz_core::CommunityId; + use sqlx::PgPool; use uuid::Uuid; const SHA: &str = "abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789"; - async fn community(db: &Db) -> CommunityId { + async fn db_with_lease() -> (Db, CommunityId, ServingWriteLease) { + let db = Db::from_pool( + PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect to test DB"), + ); let host = format!("uploader-{}.example", Uuid::new_v4().simple()); - db.ensure_configured_community(&host) + let community_id = db + .ensure_configured_community(&host) .await .expect("seed community") - .id + .id; + let lease = db + .deletion_store() + .acquire_serving_write_lease( + community_id, + "media_upload", + "media-uploader-test", + Duration::from_secs(60), + ) + .await + .expect("acquire lease"); + (db, community_id, lease) } async fn uploader_rows(db: &Db, community_id: CommunityId) -> i64 { @@ -64,14 +84,9 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] async fn repeat_upload_records_one_row() { - let db = Db::from_pool( - PgPool::connect(&crate::test_support::database_url()) - .await - .expect("connect to test DB"), - ); - let community_id = community(&db).await; + let (db, community_id, lease) = db_with_lease().await; for _ in 0..2 { - db.record_media_uploader(community_id, SHA, &[7; 32]) + db.record_media_uploader(&lease, SHA, &[7; 32]) .await .expect("record uploader"); } @@ -80,16 +95,14 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] - async fn deleting_community_refuses_new_uploaders() { - let db = Db::from_pool( - PgPool::connect(&crate::test_support::database_url()) - .await - .expect("connect to test DB"), - ); - let community_id = community(&db).await; - crate::test_support::quiesce_community_for_tests(&db.pool, community_id).await; + async fn released_lease_refuses_the_write() { + let (db, community_id, lease) = db_with_lease().await; + db.deletion_store() + .release_serving_write_lease(&lease) + .await + .expect("release lease"); assert!(db - .record_media_uploader(community_id, SHA, &[7; 32]) + .record_media_uploader(&lease, SHA, &[7; 32]) .await .is_err()); assert_eq!(uploader_rows(&db, community_id).await, 0); diff --git a/crates/buzz-relay/src/api/media.rs b/crates/buzz-relay/src/api/media.rs index 3182a22512d..000ffaf6feb 100644 --- a/crates/buzz-relay/src/api/media.rs +++ b/crates/buzz-relay/src/api/media.rs @@ -579,12 +579,12 @@ async fn upload_blob_result( state .db .record_media_uploader( - auth.tenant.community(), + serving_write.lease(), &descriptor.sha256, auth.auth_event.pubkey.as_bytes(), ) .await - .map_err(|_| MediaError::Internal)?; + .map_err(|error| serving_lease_lost(error.into()))?; // Normalize MIME to a known set to bound label cardinality. let mime_label = match descriptor.mime_type.as_str() {