fix(relay): bound snapshot reads and deletion

Signed-off-by: Daniel Rouleau <drouleau@squareup.com>

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,

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;

crates/buzz-db/src/store/artifact_snapshot.rs

@@ -39,6 +39,58 @@ pub struct Manifest {
     active: Vec<String>,
 }
 
+/// 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<String>),
+}
+
+impl Manifest {
+    /// Separate capture metadata from revision records for direct body lookup.
+    pub fn body_records(&self) -> Result<(Self, Vec<BodyRecord>)> {
+        let policies = self
+            .policy
+            .iter()
+            .map(|p| (p.0, p))
+            .collect::<std::collections::HashMap<_, _>>();
+        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::<Result<Vec<_>>>()?;
+        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<BodyRecord>) -> 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<sqlx::Transaction<'static, sqlx::Postgres>> {
     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);
         }

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<u64> {
+    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<u64> {
+    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<u64> {
+    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<String>) = 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<String>)]) -> 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<Vec<(u64, Vec<String>)>> {
+    let mut cursor = 0u64;
+    let mut pages = Vec::new();
+    loop {
+        let page: (u64, Vec<String>) = 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<String> = redis::cmd("HGET")
+            .arg(ARTIFACT_SNAPSHOT_STORE)
+            .arg(format!("body:{id}:revision"))
+            .query_async(&mut *conn)
+            .await
+            .unwrap();
+        assert_eq!(body, None);
+    }
+}

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<u64> {
-    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<String>) = 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<String>)]) -> 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<Vec<(u64, Vec<String>)>> {
-    let mut cursor = 0u64;
-    let mut pages = Vec::new();
-    loop {
-        let page: (u64, Vec<String>) = 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()

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<u8> = 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<String> = 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;
+}

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<Json<Value>, (StatusCode, Json<Value>)>;
 
@@ -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::<Result<Vec<_>, _>>()?;
+        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::<i64>(&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<Vec<u8>> = 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<Option<Vec<u8>>> = 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<Vec<u8>> = 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::<BodyRecord>(&row.ok_or_else(invalid)?)
+                    .map_err(|_| unavailable())
+            })
+            .collect::<Result<Vec<_>, _>>()?;
+        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

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

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);

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);