From cdfee769b30e504d19adbc6c8f465e77e4fd8cd7 Mon Sep 17 00:00:00 2001 From: Daniel Rouleau Date: Wed, 21 Oct 2026 13:31:32 -0700 Subject: [PATCH] fix(relay): bound snapshot reads and deletion Signed-off-by: Daniel Rouleau --- crates/buzz-db/src/runtime/migration.rs | 8 +- .../src/store/artifact_postgres_tests.rs | 14 +- crates/buzz-db/src/store/artifact_snapshot.rs | 64 ++++++- .../buzz-deletion/src/artifact_snapshots.rs | 173 ++++++++++++++++++ crates/buzz-deletion/src/lib.rs | 85 +-------- .../snapshot_postgres_tests.rs | 108 +++++++++-- .../buzz-relay/src/api/artifact_snapshot.rs | 145 ++++++++++++--- docs/nips/NIP-AR.md | 7 +- migrations/0067_artifact_revision_lookup.sql | 5 + schema/tables/public/artifact_revisions.sql | 3 + 10 files changed, 484 insertions(+), 128 deletions(-) create mode 100644 crates/buzz-deletion/src/artifact_snapshots.rs create mode 100644 migrations/0067_artifact_revision_lookup.sql diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 7cb0879ff8..eed8e8331d 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 [ @@ -2676,7 +2676,6 @@ mod postgres_tests { .await .expect("read index shapes") } - let base_url = std::env::var("BUZZ_TEST_DATABASE_URL") .or_else(|_| std::env::var("DATABASE_URL")) .unwrap_or_else(|_| TEST_DB_URL.to_owned()); @@ -2748,14 +2747,15 @@ mod postgres_tests { .await .expect("connect migrated probe database"); MIGRATOR - .run_to(55, &migrated) + .run_to(67, &migrated) .await - .expect("apply migrations 1-55"); + .expect("apply migrations 1-67"); for table in [ "relay_admin_actions", "relay_admin_outbox", "relay_operator_audit", + "artifact_revisions", ] { assert_eq!( columns(&desired, table).await, diff --git a/crates/buzz-db/src/store/artifact_postgres_tests.rs b/crates/buzz-db/src/store/artifact_postgres_tests.rs index 1aba26cacf..8792256d99 100644 --- a/crates/buzz-db/src/store/artifact_postgres_tests.rs +++ b/crates/buzz-db/src/store/artifact_postgres_tests.rs @@ -657,7 +657,19 @@ async fn artifact_snapshot_keeps_captured_revisions_and_invalidates_removed_data f.db.capture_artifact_manifest(f.community, reader.as_bytes(), &query) }; let manifest = capture().await.unwrap(); - let newer = f.revision(d, "update", f.a, Some(&first), &f.owner, vec![]); + let newer = f.revision( + d, + "update", + f.a, + Some(&first), + &f.owner, + vec![vec![ + "label".into(), + "op".into(), + "delete".into(), + "move".into(), + ]], + ); f.accept(&newer, None).await; let second = f.revision(Uuid::new_v4(), "create", f.a, None, &f.owner, vec![]); f.accept(&second, None).await; diff --git a/crates/buzz-db/src/store/artifact_snapshot.rs b/crates/buzz-db/src/store/artifact_snapshot.rs index ad452b1cde..a9644b0a9c 100644 --- a/crates/buzz-db/src/store/artifact_snapshot.rs +++ b/crates/buzz-db/src/store/artifact_snapshot.rs @@ -39,6 +39,58 @@ pub struct Manifest { active: Vec, } +/// One revision and its captured access scope for a bounded Redis body read. +#[derive(Deserialize, Serialize)] +pub struct BodyRecord { + entry: Entry, + revision: String, + policy: (Uuid, String, Option), +} + +impl Manifest { + /// Separate capture metadata from revision records for direct body lookup. + pub fn body_records(&self) -> Result<(Self, Vec)> { + let policies = self + .policy + .iter() + .map(|p| (p.0, p)) + .collect::>(); + let records = self + .entries + .iter() + .zip(&self.revisions) + .map(|(entry, revision)| { + Ok(BodyRecord { + entry: entry.clone(), + revision: revision.clone(), + policy: (*policies.get(&entry.channel_id).ok_or_else(|| { + DbError::InvalidData("missing snapshot authority".into()) + })?) + .clone(), + }) + }) + .collect::>>()?; + let mut metadata = self.clone(); + metadata.entries.clear(); + metadata.revisions.clear(); + metadata.policy.clear(); + metadata.predicates.clear(); + Ok((metadata, records)) + } + + /// Restore only requested records with the original capture boundary. + pub fn with_body_records(mut self, records: Vec) -> Self { + let mut policies = std::collections::BTreeMap::new(); + for record in records { + self.entries.push(record.entry); + self.revisions.push(record.revision); + policies.insert(record.policy.0, record.policy); + } + self.policy = policies.into_values().collect(); + self + } +} + async fn begin(db: &Db) -> Result> { let connection = crate::observability::acquire_writer( &db.pool, @@ -417,9 +469,15 @@ impl Db { WHERE r.community_id=$1 AND r.artifact_id=ANY($2) \ AND (age(r.xmin)<=age($3::text::xid) OR r.xmin::text=ANY($4)) \ AND (e.id IS NULL OR e.deleted_at IS NOT NULL \ - OR e.tags @> '[[\"op\",\"delete\"]]'::jsonb OR e.tags @> '[[\"op\",\"move\"]]'::jsonb))", - ).bind(community.as_uuid()).bind(&ids).bind(&manifest.xmax).bind(&manifest.active) - .fetch_one(&mut *tx).await?; + OR EXISTS(SELECT 1 FROM jsonb_array_elements(e.tags) tag \ + WHERE tag->>0='op' AND tag->>1 IN ('delete','move'))))", + ) + .bind(community.as_uuid()) + .bind(&ids) + .bind(&manifest.xmax) + .bind(&manifest.active) + .fetch_one(&mut *tx) + .await?; if changed { return Err(DbError::ArtifactSnapshotReset); } diff --git a/crates/buzz-deletion/src/artifact_snapshots.rs b/crates/buzz-deletion/src/artifact_snapshots.rs new file mode 100644 index 0000000000..c0e5f870d6 --- /dev/null +++ b/crates/buzz-deletion/src/artifact_snapshots.rs @@ -0,0 +1,173 @@ +//! Redis cleanup includes shared artifact snapshots. +use super::*; + +/// Shared eviction unit for bounded, short-lived artifact snapshots. +pub const ARTIFACT_SNAPSHOT_STORE: &str = "buzz:{artifact-snapshots-v1}:store"; + +/// Remove one community's snapshots and their quota charges atomically. +pub async fn purge_artifact_snapshots( + pool: &deadpool_redis::Pool, + community: buzz_core::CommunityId, +) -> Result { + let mut connection = pool.get().await?; + snapshot_namespace(&mut connection, community, true).await +} + +async fn snapshot_namespace( + connection: &mut deadpool_redis::Connection, + community: buzz_core::CommunityId, + remove: bool, +) -> Result { + Ok(redis::Script::new(r#" +local stored = redis.call('HGET', KEYS[1], 'ids') +if not stored then return 0 end +local ids = cjson.decode(stored) +local retained, count = {}, 0 +for _, id in ipairs(ids) do + if string.sub(id, 1, string.len(ARGV[1])) == ARGV[1] then + count = count + 1 + if ARGV[2] == '1' then + local rows = cjson.decode(redis.call('HGET', KEYS[1], 'rows:' .. id) or '[]') + for _, revision in ipairs(rows) do redis.call('HDEL', KEYS[1], 'body:' .. id .. ':' .. revision) end + local size = tonumber(redis.call('HGET', KEYS[1], 'size:' .. id)) + redis.call('HINCRBY', KEYS[1], 'total', -size) + redis.call('HDEL', KEYS[1], 'expiry:' .. id, 'size:' .. id, 'owner:' .. id, + 'data:' .. id, 'progress:' .. id, 'meta:' .. id, 'rows:' .. id) + end + else + table.insert(retained, id) + end +end +if ARGV[2] == '1' then redis.call('HSET', KEYS[1], 'ids', cjson.encode(retained)) end +return count +"#).key(ARTIFACT_SNAPSHOT_STORE) + .arg(format!("buzz:{{artifact-snapshots-v1}}:{}:", community.as_uuid())) + .arg(if remove { "1" } else { "0" }) + .invoke_async(&mut **connection).await?) +} + +pub(super) async fn purge_redis_namespace( + pool: &deadpool_redis::Pool, + community: buzz_core::CommunityId, +) -> Result { + let mut connection = pool.get().await?; + let pattern = format!("buzz:{community}:*"); + let mut cursor = 0u64; + let mut deleted = snapshot_namespace(&mut connection, community, true).await?; + loop { + let (next, keys): (u64, Vec) = redis::cmd("SCAN") + .arg(cursor) + .arg("MATCH") + .arg(&pattern) + .arg("COUNT") + .arg(1000) + .query_async(&mut *connection) + .await?; + if !keys.is_empty() { + let count: u64 = redis::cmd("UNLINK") + .arg(&keys) + .query_async(&mut *connection) + .await?; + deleted = deleted.saturating_add(count); + } + if next == 0 { + break; + } + cursor = next; + } + Ok(deleted) +} + +pub(super) fn scan_proves_absence(pages: &[(u64, Vec)]) -> bool { + pages.last().is_some_and(|(cursor, _)| *cursor == 0) + && pages.iter().all(|(_, keys)| keys.is_empty()) +} + +async fn scan_redis_namespace( + connection: &mut deadpool_redis::Connection, + pattern: &str, +) -> Result)>> { + let mut cursor = 0u64; + let mut pages = Vec::new(); + loop { + let page: (u64, Vec) = redis::cmd("SCAN") + .arg(cursor) + .arg("MATCH") + .arg(pattern) + .arg("COUNT") + .arg(1000) + .query_async(&mut **connection) + .await?; + cursor = page.0; + pages.push(page); + if cursor == 0 { + return Ok(pages); + } + } +} + +pub(super) async fn verify_redis_absence( + pool: &deadpool_redis::Pool, + community: buzz_core::CommunityId, +) -> Result<()> { + let mut connection = pool.get().await?; + let pattern = format!("buzz:{community}:*"); + // SCAN is weakly consistent. Two complete empty passes ensure a cursor + // rollover or concurrent expiry cannot make one sparse pass look absent. + let first = scan_redis_namespace(&mut connection, &pattern).await?; + let second = scan_redis_namespace(&mut connection, &pattern).await?; + let snapshots = snapshot_namespace(&mut connection, community, false).await?; + if scan_proves_absence(&first) && scan_proves_absence(&second) && snapshots == 0 { + Ok(()) + } else { + Err(transient( + "logical verification found a Redis namespace key", + )) + } +} + +#[cfg(test)] +mod postgres_tests { + use super::*; + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn artifact_snapshot_community_purge_checks_shared_storage() { + let pool = deadpool_redis::Config::from_url( + std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".into()), + ) + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .unwrap(); + let community = buzz_core::CommunityId::from_uuid(Uuid::new_v4()); + let id = format!( + "buzz:{{artifact-snapshots-v1}}:{}:fixture", + community.as_uuid() + ); + let mut conn = pool.get().await.unwrap(); + let _: i64 = redis::Script::new( + r#" +local ids = cjson.decode(redis.call('HGET', KEYS[1], 'ids') or '[]') +table.insert(ids, ARGV[1]) +redis.call('HSET', KEYS[1], 'ids', cjson.encode(ids), 'size:' .. ARGV[1], 10, + 'rows:' .. ARGV[1], '["revision"]', 'body:' .. ARGV[1] .. ':revision', 'body') +redis.call('HINCRBY', KEYS[1], 'total', 10) +return 1 +"#, + ) + .key(ARTIFACT_SNAPSHOT_STORE) + .arg(&id) + .invoke_async(&mut *conn) + .await + .unwrap(); + assert!(verify_redis_absence(&pool, community).await.is_err()); + assert_eq!(purge_redis_namespace(&pool, community).await.unwrap(), 1); + verify_redis_absence(&pool, community).await.unwrap(); + let body: Option = redis::cmd("HGET") + .arg(ARTIFACT_SNAPSHOT_STORE) + .arg(format!("body:{id}:revision")) + .query_async(&mut *conn) + .await + .unwrap(); + assert_eq!(body, None); + } +} diff --git a/crates/buzz-deletion/src/lib.rs b/crates/buzz-deletion/src/lib.rs index c3fb598523..6e1cbe4187 100644 --- a/crates/buzz-deletion/src/lib.rs +++ b/crates/buzz-deletion/src/lib.rs @@ -24,6 +24,12 @@ use serde::Serialize; use tokio_util::sync::CancellationToken; use uuid::Uuid; +mod artifact_snapshots; +#[cfg(test)] +use artifact_snapshots::scan_proves_absence; +pub use artifact_snapshots::{purge_artifact_snapshots, ARTIFACT_SNAPSHOT_STORE}; +use artifact_snapshots::{purge_redis_namespace, verify_redis_absence}; + /// Fleet-wide object cap for one observational taxonomy sweep. const DEFAULT_SWEEP_OBJECT_CAP: u64 = 10_000_000; /// Keys per frozen side-table chunk (one `DeleteObjects`-sized unit × 10). @@ -1688,85 +1694,6 @@ async fn publish_disconnect_community( Ok(()) } -async fn purge_redis_namespace( - pool: &deadpool_redis::Pool, - community: buzz_core::CommunityId, -) -> Result { - let mut connection = pool.get().await?; - let pattern = format!("buzz:{community}:*"); - let mut cursor = 0u64; - let mut deleted = 0u64; - loop { - let (next, keys): (u64, Vec) = redis::cmd("SCAN") - .arg(cursor) - .arg("MATCH") - .arg(&pattern) - .arg("COUNT") - .arg(1000) - .query_async(&mut *connection) - .await?; - if !keys.is_empty() { - let count: u64 = redis::cmd("UNLINK") - .arg(&keys) - .query_async(&mut *connection) - .await?; - deleted = deleted.saturating_add(count); - } - if next == 0 { - break; - } - cursor = next; - } - Ok(deleted) -} - -fn scan_proves_absence(pages: &[(u64, Vec)]) -> bool { - pages.last().is_some_and(|(cursor, _)| *cursor == 0) - && pages.iter().all(|(_, keys)| keys.is_empty()) -} - -async fn scan_redis_namespace( - connection: &mut deadpool_redis::Connection, - pattern: &str, -) -> Result)>> { - let mut cursor = 0u64; - let mut pages = Vec::new(); - loop { - let page: (u64, Vec) = redis::cmd("SCAN") - .arg(cursor) - .arg("MATCH") - .arg(pattern) - .arg("COUNT") - .arg(1000) - .query_async(&mut **connection) - .await?; - cursor = page.0; - pages.push(page); - if cursor == 0 { - return Ok(pages); - } - } -} - -async fn verify_redis_absence( - pool: &deadpool_redis::Pool, - community: buzz_core::CommunityId, -) -> Result<()> { - let mut connection = pool.get().await?; - let pattern = format!("buzz:{community}:*"); - // SCAN is weakly consistent. Two complete empty passes ensure a cursor - // rollover or concurrent expiry cannot make one sparse pass look absent. - let first = scan_redis_namespace(&mut connection, &pattern).await?; - let second = scan_redis_namespace(&mut connection, &pattern).await?; - if scan_proves_absence(&first) && scan_proves_absence(&second) { - Ok(()) - } else { - Err(transient( - "logical verification found a Redis namespace key", - )) - } -} - fn sweep_object_cap() -> u64 { std::env::var("BUZZ_DELETION_SWEEP_MAX_OBJECTS") .ok() diff --git a/crates/buzz-relay/src/api/artifact_postgres_tests/snapshot_postgres_tests.rs b/crates/buzz-relay/src/api/artifact_postgres_tests/snapshot_postgres_tests.rs index 12e3470bf6..d1775f99c9 100644 --- a/crates/buzz-relay/src/api/artifact_postgres_tests/snapshot_postgres_tests.rs +++ b/crates/buzz-relay/src/api/artifact_postgres_tests/snapshot_postgres_tests.rs @@ -559,21 +559,9 @@ async fn artifact_snapshot_tokens_expiry_and_allocation_are_bounded() { } async fn cleanup(f: &Fixture) { - let mut conn = f.state.redis_pool.get().await.unwrap(); - let prefix = format!("buzz:{{artifact-snapshots-v1}}:{}:", f.community.as_uuid()); - let _: i64 = redis::Script::new(r#" - for _, field in ipairs(redis.call('HKEYS', KEYS[1])) do - if string.sub(field,1,7) == 'expiry:' then - local id = string.sub(field,8) - if string.sub(id,1,string.len(ARGV[1])) == ARGV[1] then - local size = tonumber(redis.call('HGET',KEYS[1],'size:' .. id)) - redis.call('HINCRBY',KEYS[1],'total',-size) - redis.call('HDEL',KEYS[1],field,'size:' .. id,'owner:' .. id,'data:' .. id,'progress:' .. id) - end - end - end - return 1 - "#).key("buzz:{artifact-snapshots-v1}:store").arg(prefix).invoke_async(&mut *conn).await.unwrap(); + buzz_deletion::purge_artifact_snapshots(&f.state.redis_pool, f.community) + .await + .unwrap(); } #[tokio::test] @@ -614,3 +602,93 @@ async fn artifact_snapshot_new_matching_home_resets_then_a_fresh_capture_complet assert!(status.is_success(), "{result}"); cleanup(&f).await; } + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn artifact_snapshot_deletion_removes_only_its_community_and_fences_capture() { + let f = Fixture::new().await; + let other = Fixture::new().await; + let event = f.revision(Uuid::new_v4(), "create", None); + f.publish(&event).await; + let start = json!([{"artifact":"snapshot","action":"start","query":{"artifact":"current"}}]); + let (status, captured) = f.request("/query", start.clone()).await; + assert!(status.is_success(), "{captured}"); + let (status, retained) = other.request("/query", start.clone()).await; + assert!(status.is_success(), "{retained}"); + let token = captured["token"].as_str().unwrap(); + let snapshot_key = format!( + "buzz:{{artifact-snapshots-v1}}:{}:{}:{}", + f.community.as_uuid(), + f.owner.public_key().to_hex(), + token + ); + let mut conn = f.state.redis_pool.get().await.unwrap(); + let bytes_before: usize = redis::cmd("HGET") + .arg(buzz_deletion::ARTIFACT_SNAPSHOT_STORE) + .arg("total") + .query_async(&mut *conn) + .await + .unwrap(); + let charge: usize = redis::cmd("HGET") + .arg(buzz_deletion::ARTIFACT_SNAPSHOT_STORE) + .arg(format!("size:{snapshot_key}")) + .query_async(&mut *conn) + .await + .unwrap(); + let metadata: Vec = redis::cmd("HGET") + .arg(buzz_deletion::ARTIFACT_SNAPSHOT_STORE) + .arg(format!("meta:{snapshot_key}")) + .query_async(&mut *conn) + .await + .unwrap(); + let metadata: Value = serde_json::from_slice(&metadata).unwrap(); + assert_eq!(metadata["manifest"]["entries"], json!([])); + assert_eq!(metadata["manifest"]["policy"], json!([])); + let tenant = crate::tenant::bind_community(&f.state.db, &f.host) + .await + .unwrap(); + let mut tx = f.pool.begin().await.unwrap(); + sqlx::query("SELECT set_config('buzz.deletion_executor_community', $1, true), set_config('buzz.deletion_fence_generation', '0', true)") + .bind(f.community.as_uuid().to_string()).execute(&mut *tx).await.unwrap(); + sqlx::query("UPDATE communities SET deletion_state='quiescing' WHERE id=$1") + .bind(f.community.as_uuid()) + .execute(&mut *tx) + .await + .unwrap(); + tx.commit().await.unwrap(); + // An already-bound HTTP request must also fail at the external write seam. + let denied = + crate::api::artifact_snapshot::query(&f.state, &tenant, &f.owner.public_key(), &start[0]) + .await; + assert_eq!(denied.unwrap_err().0, StatusCode::SERVICE_UNAVAILABLE); + assert_eq!( + buzz_deletion::purge_artifact_snapshots(&f.state.redis_pool, f.community) + .await + .unwrap(), + 1 + ); + let bytes_after: usize = redis::cmd("HGET") + .arg(buzz_deletion::ARTIFACT_SNAPSHOT_STORE) + .arg("total") + .query_async(&mut *conn) + .await + .unwrap(); + assert_eq!(bytes_after, bytes_before - charge); + let fields: Vec = redis::cmd("HKEYS") + .arg(buzz_deletion::ARTIFACT_SNAPSHOT_STORE) + .query_async(&mut *conn) + .await + .unwrap(); + assert!(!fields.iter().any(|field| field.contains(&snapshot_key))); + let result = crate::api::artifact_snapshot::query(&f.state, &tenant, &f.owner.public_key(), + &json!({"artifact":"snapshot","action":"bodies","token":token,"revisionIds":[event.id.to_hex()]})).await; + assert_eq!(result.unwrap_err().0, StatusCode::CONFLICT); + let (status, completed) = other + .request( + "/query", + json!([{"artifact":"snapshot","action":"complete","token":retained["token"]}]), + ) + .await; + assert!(status.is_success(), "{completed}"); + cleanup(&other).await; +} diff --git a/crates/buzz-relay/src/api/artifact_snapshot.rs b/crates/buzz-relay/src/api/artifact_snapshot.rs index 8a3d9452c9..aad5a86b6e 100644 --- a/crates/buzz-relay/src/api/artifact_snapshot.rs +++ b/crates/buzz-relay/src/api/artifact_snapshot.rs @@ -2,7 +2,7 @@ use crate::state::AppState; use axum::{http::StatusCode, Json}; use buzz_core::{artifact::ArtifactView, TenantContext}; -use buzz_db::artifact_snapshot::{Manifest, MAX_BYTES, MAX_ROWS}; +use buzz_db::artifact_snapshot::{BodyRecord, Manifest, MAX_BYTES, MAX_ROWS}; use deadpool_redis::redis; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; @@ -16,7 +16,7 @@ const BODY_ROWS: usize = 100; const BODY_BYTES: usize = 8_388_608; const GLOBAL_BYTES: usize = 64 * 1024 * 1024; const GLOBAL_COUNT: usize = 32; -const STORE_KEY: &str = "buzz:{artifact-snapshots-v1}:store"; +const STORE_KEY: &str = buzz_deletion::ARTIFACT_SNAPSHOT_STORE; static WORK: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(4); type Response = Result, (StatusCode, Json)>; @@ -145,6 +145,13 @@ async fn execute( .await .map_err(|_| unavailable())?; if let Request::Start { artifact, query } = request { + let serving_write = buzz_deletion::acquire_serving_write( + &state.db, + tenant.community(), + "artifact_snapshot", + ) + .await + .map_err(|_| unavailable())?; if artifact != "snapshot" || query.get("limit").is_some() || query.get("offset").is_some() { return Err(invalid()); } @@ -176,18 +183,52 @@ async fn execute( let response = page(&snapshot, &token, 0)?; let snapshot_key = key(tenant, reader, &token); let principal_key = key(tenant, reader, "active"); - let stored: i64 = redis::Script::new(REGISTER) - .key(STORE_KEY) - .arg(expires_at) - .arg(data.len()) - .arg(GLOBAL_BYTES) - .arg(GLOBAL_COUNT) - .arg(data) - .arg(&snapshot_key) - .arg(principal_key) - .invoke_async(&mut *conn) + let (metadata, records) = snapshot + .manifest + .body_records() + .map_err(|e| db_error(e, tenant, reader))?; + let metadata = serde_json::to_vec(&Snapshot { + manifest: metadata, + query: Value::Null, + generation: token.clone(), + expires_at, + cursors: vec![], + }) + .map_err(|_| unavailable())?; + let rows = records + .iter() + .zip(&snapshot.manifest.entries) + .map(|(record, entry)| { + Ok(( + entry.revision_id.clone(), + serde_json::to_string(record).map_err(|_| unavailable())?, + )) + }) + .collect::, _>>()?; + let rows = serde_json::to_vec(&rows).map_err(|_| unavailable())?; + // Charge duplicate records, field names, and Redis object overhead. + let stored_bytes = data.len() + metadata.len() + rows.len() + records.len() * 512; + let registration = async { + redis::Script::new(REGISTER) + .key(STORE_KEY) + .arg(expires_at) + .arg(stored_bytes) + .arg(GLOBAL_BYTES) + .arg(GLOBAL_COUNT) + .arg(data) + .arg(&snapshot_key) + .arg(principal_key) + .arg(metadata) + .arg(rows) + .invoke_async::(&mut *conn) + .await + }; + let stored = serving_write + .protect(registration) .await + .map_err(|_| unavailable())? .map_err(|_| unavailable())?; + serving_write.finish().await.map_err(|_| unavailable())?; if stored == 0 { return Err(limit()); } @@ -207,20 +248,63 @@ async fn execute( return Err(invalid()); } let snapshot_key = key(tenant, reader, token); - let data: Option> = redis::cmd("HGET") - .arg(STORE_KEY) - .arg(format!("data:{snapshot_key}")) - .query_async(&mut *conn) - .await - .map_err(|_| unavailable())?; + let body_ids = match &request { + Request::Bodies { revision_ids, .. } => { + if revision_ids.is_empty() { + return Err(invalid()); + } + if revision_ids.len() > BODY_ROWS { + return Err(limit()); + } + let mut seen = std::collections::HashSet::new(); + if revision_ids + .iter() + .any(|id| buzz_core::artifact::event_id(id).is_err() || !seen.insert(id)) + { + return Err(invalid()); + } + Some(revision_ids) + } + _ => None, + }; + let (data, body_rows) = if let Some(ids) = body_ids { + let mut fields = vec![format!("meta:{snapshot_key}")]; + fields.extend(ids.iter().map(|id| format!("body:{snapshot_key}:{id}"))); + let mut rows: Vec>> = redis::cmd("HMGET") + .arg(STORE_KEY) + .arg(fields) + .query_async(&mut *conn) + .await + .map_err(|_| unavailable())?; + let data = rows.remove(0); + (data, rows) + } else { + let data: Option> = redis::cmd("HGET") + .arg(STORE_KEY) + .arg(format!("data:{snapshot_key}")) + .query_async(&mut *conn) + .await + .map_err(|_| unavailable())?; + (data, vec![]) + }; let data = data.ok_or_else(reset)?; if data.len() > MAX_BYTES { return Err(unavailable()); } - let snapshot: Snapshot = serde_json::from_slice(&data).map_err(|_| unavailable())?; + let mut snapshot: Snapshot = serde_json::from_slice(&data).map_err(|_| unavailable())?; if snapshot.generation != *token || now >= snapshot.expires_at { return Err(reset()); } + if body_ids.is_some() { + let records = body_rows + .into_iter() + .map(|row| { + serde_json::from_slice::(&row.ok_or_else(invalid)?) + .map_err(|_| unavailable()) + }) + .collect::, _>>()?; + snapshot.manifest = snapshot.manifest.with_body_records(records); + } let requested = match &request { Request::Bodies { revision_ids, .. } => { if revision_ids.is_empty() { @@ -320,25 +404,38 @@ async fn execute( const REGISTER: &str = r#" local now = tonumber(redis.call('TIME')[1]) local count, principal, latest = 0, 0, tonumber(ARGV[1]) -for _, field in ipairs(redis.call('HKEYS', KEYS[1])) do - if string.sub(field, 1, 7) == 'expiry:' then - local id = string.sub(field, 8) +local ids = cjson.decode(redis.call('HGET', KEYS[1], 'ids') or '[]') +local retained = {} +for _, id in ipairs(ids) do + local field = 'expiry:' .. id local expiry = tonumber(redis.call('HGET', KEYS[1], field)) if expiry <= now then local size = tonumber(redis.call('HGET', KEYS[1], 'size:' .. id)) redis.call('HINCRBY', KEYS[1], 'total', -size) - redis.call('HDEL', KEYS[1], field, 'size:' .. id, 'owner:' .. id, 'data:' .. id, 'progress:' .. id) + local rows = cjson.decode(redis.call('HGET', KEYS[1], 'rows:' .. id) or '[]') + for _, revision in ipairs(rows) do redis.call('HDEL', KEYS[1], 'body:' .. id .. ':' .. revision) end + redis.call('HDEL', KEYS[1], field, 'size:' .. id, 'owner:' .. id, 'data:' .. id, 'progress:' .. id, 'meta:' .. id, 'rows:' .. id) else + table.insert(retained, id) count = count + 1 latest = math.max(latest, expiry) if redis.call('HGET', KEYS[1], 'owner:' .. id) == ARGV[7] then principal = principal + 1 end end - end end +redis.call('HSET', KEYS[1], 'ids', cjson.encode(retained)) if tonumber(ARGV[1]) <= now or count >= tonumber(ARGV[4]) or principal >= 2 or tonumber(redis.call('HGET', KEYS[1], 'total') or '0') + tonumber(ARGV[2]) > tonumber(ARGV[3]) then return 0 end redis.call('HSET', KEYS[1], 'data:' .. ARGV[6], ARGV[5], 'progress:' .. ARGV[6], 1, 'expiry:' .. ARGV[6], ARGV[1], 'size:' .. ARGV[6], ARGV[2], 'owner:' .. ARGV[6], ARGV[7]) +local rows = cjson.decode(ARGV[9] or '[]') +local revisions = {} +for _, row in ipairs(rows) do + redis.call('HSET', KEYS[1], 'body:' .. ARGV[6] .. ':' .. row[1], row[2]) + table.insert(revisions, row[1]) +end +redis.call('HSET', KEYS[1], 'meta:' .. ARGV[6], ARGV[8] or '', 'rows:' .. ARGV[6], cjson.encode(revisions)) +table.insert(retained, ARGV[6]) +redis.call('HSET', KEYS[1], 'ids', cjson.encode(retained)) redis.call('HINCRBY', KEYS[1], 'total', ARGV[2]) redis.call('EXPIREAT', KEYS[1], latest) return 1 diff --git a/docs/nips/NIP-AR.md b/docs/nips/NIP-AR.md index d0bbd52f48..7d009888c4 100644 --- a/docs/nips/NIP-AR.md +++ b/docs/nips/NIP-AR.md @@ -208,7 +208,7 @@ A capture above the row or byte limit fails. It is never truncated. The relay in and current candidates per operation. The candidate limit applies after the immutable type predicate and before other tag predicates. Each internal manifest is at most 16 MiB. Shared state is limited to 32 active -captures and 64 MiB of stored manifest bytes. Each reader has at most two +captures and 64 MiB of charged snapshot storage. Each reader has at most two captures per community. Each process runs at most four snapshot operations at once. Each SQL statement has a two-second limit. Each operation has an eight-second elapsed limit. These limits include access checks. @@ -232,7 +232,10 @@ across relay processes without session affinity. The complete manifest is published atomically before its token is returned. The fixed expiry does not extend on reads. Redis expiry, eviction, or loss requires a reset. A Redis outage returns a service failure. The relay uses no memory fallback. -One Redis hash holds manifests and their accounting, so eviction removes both. +One Redis hash holds manifests, direct body records, and quota charges. +Eviction removes them together. Body reads fetch only the requested records. +A serving write lease protects capture registration. Community deletion +removes its records and quota charges. Each capture has a fixed logical expiry. A bounded allocation step removes expired captures. The hash expires when its last capture expires. A completed capture remains replayable until expiry and counts against the resource limits during that diff --git a/migrations/0067_artifact_revision_lookup.sql b/migrations/0067_artifact_revision_lookup.sql new file mode 100644 index 0000000000..256c7b1f7d --- /dev/null +++ b/migrations/0067_artifact_revision_lookup.sql @@ -0,0 +1,5 @@ +SET LOCAL lock_timeout = '5s'; +SET LOCAL statement_timeout = '30s'; + +CREATE INDEX idx_artifact_revisions_artifact + ON artifact_revisions (community_id, artifact_id); diff --git a/schema/tables/public/artifact_revisions.sql b/schema/tables/public/artifact_revisions.sql index 5708601814..5aa24c2572 100644 --- a/schema/tables/public/artifact_revisions.sql +++ b/schema/tables/public/artifact_revisions.sql @@ -6,3 +6,6 @@ CREATE TABLE artifact_revisions ( artifact_id UUID NOT NULL, PRIMARY KEY (community_id, event_id) ); + +CREATE INDEX idx_artifact_revisions_artifact + ON artifact_revisions (community_id, artifact_id);