feat(media): add phased S3 layout migration

.env.example

@@ -95,6 +95,13 @@ BUZZ_S3_BUCKET=buzz-media
 BUZZ_S3_REGION=us-east-1
 BUZZ_S3_ADDRESSING_STYLE=path
 
+# Media payload write migration. Unset defaults to dual-read-legacy-write, so upgrades preserve
+# existing flat-only writes and do not duplicate objects automatically.
+# Stages: dual-read-legacy-write -> dual-read-dual-write -> sharded-only.
+# Deploy compatibility readers everywhere and backfill/reconcile objects before
+# advancing; do not roll sharded writers back to a version without sharded reads.
+# BUZZ_MEDIA_MIGRATION_PHASE=dual-read-legacy-write
+
 # -----------------------------------------------------------------------------
 # Media Upload Admission
 # -----------------------------------------------------------------------------

Dockerfile

@@ -69,14 +69,18 @@ RUN cargo chef cook --release --recipe-path recipe.json
 COPY . .
 RUN cargo build --release --locked -p buzz-relay --bin buzz-relay \
                                    -p buzz-admin --bin buzz-admin \
-                                   -p buzz-pair-relay --bin buzz-pair-relay
+                                   -p buzz-pair-relay --bin buzz-pair-relay \
+                                   -p buzz-media --bin buzz-media-layout-backfill \
+                                                 --bin buzz-media-layout-delete-legacy
 
 # Derive the normal release binaries from the same optimized ELF files as the
 # debug image so the two variants cannot drift at code-generation time.
 FROM builder AS stripped-binaries
 RUN strip target/release/buzz-relay \
     && strip target/release/buzz-admin \
-    && strip target/release/buzz-pair-relay
+    && strip target/release/buzz-pair-relay \
+    && strip target/release/buzz-media-layout-backfill \
+    && strip target/release/buzz-media-layout-delete-legacy
 
 # ─── Stage 4: web bundle (pnpm + vite) ──────────────────────────────────────
 # Independent of the Rust layers so a CSS change doesn't bust Rust cache and
@@ -169,10 +173,14 @@ FROM runtime-base AS runtime-debug
 COPY --from=builder /build/target/release/buzz-relay /usr/local/bin/buzz-relay
 COPY --from=builder /build/target/release/buzz-admin /usr/local/bin/buzz-admin
 COPY --from=builder /build/target/release/buzz-pair-relay /usr/local/bin/buzz-pair-relay
+COPY --from=builder /build/target/release/buzz-media-layout-backfill /usr/local/bin/buzz-media-layout-backfill
+COPY --from=builder /build/target/release/buzz-media-layout-delete-legacy /usr/local/bin/buzz-media-layout-delete-legacy
 
 # Keep the stripped runtime as the final/default Dockerfile target so existing
 # `docker build .` callers and release tags retain their current behavior.
 FROM runtime-base AS runtime
 COPY --from=stripped-binaries /build/target/release/buzz-relay /usr/local/bin/buzz-relay
 COPY --from=stripped-binaries /build/target/release/buzz-admin /usr/local/bin/buzz-admin
 COPY --from=stripped-binaries /build/target/release/buzz-pair-relay /usr/local/bin/buzz-pair-relay
+COPY --from=stripped-binaries /build/target/release/buzz-media-layout-backfill /usr/local/bin/buzz-media-layout-backfill
+COPY --from=stripped-binaries /build/target/release/buzz-media-layout-delete-legacy /usr/local/bin/buzz-media-layout-delete-legacy

crates/buzz-media/Cargo.toml

@@ -32,6 +32,10 @@ tempfile = "3"
 tokio-util = { version = "0.7", features = ["io"] }
 futures-util = "0.3"
 futures-core = "0.3"
+metrics = { workspace = true }
+clap = { version = "4", features = ["derive", "env"] }
+anyhow = { workspace = true }
+tracing-subscriber = { workspace = true }
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["test-util"] }

crates/buzz-media/src/bin/buzz-media-layout-backfill.rs

@@ -0,0 +1,94 @@
+//! Idempotently copy legacy media payloads into the sharded layout.
+
+use anyhow::{Context, Result};
+use buzz_media::migration::{objects_for_sidecar, parse_sidecar_key, RequestPacer};
+use clap::Parser;
+
+mod media_layout_common;
+use media_layout_common::CommonArgs;
+
+#[derive(Debug, Parser)]
+#[command(name = "buzz-media-layout-backfill")]
+struct Args {
+    #[command(flatten)]
+    common: CommonArgs,
+    /// Report actions without copying objects.
+    #[arg(long, env = "BUZZ_MEDIA_MIGRATION_DRY_RUN", default_value_t = false)]
+    dry_run: bool,
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    tracing_subscriber::fmt().with_target(false).init();
+    let args = Args::parse();
+    let storage = args.common.storage()?;
+    let mut pacer = RequestPacer::new(args.common.requests_per_second);
+    let mut continuation = None;
+    let mut start_after = args.common.start_after.clone();
+    let mut processed = 0_u64;
+    let mut copied = 0_u64;
+    let mut skipped = 0_u64;
+    let mut checkpoint = start_after.clone();
+
+    loop {
+        pacer.wait().await;
+        let page = storage
+            .list_prefix_page(
+                "_meta/",
+                continuation.take(),
+                start_after.take(),
+                args.common.page_size,
+            )
+            .await
+            .context("list media sidecars")?;
+        for (sidecar_key, _) in page.objects {
+            let Some((community, sha)) = parse_sidecar_key(&sidecar_key) else {
+                tracing::warn!(key = %sidecar_key, "skipping malformed sidecar key");
+                continue;
+            };
+            pacer.wait().await;
+            let bytes = storage
+                .get(&sidecar_key)
+                .await
+                .context("read media sidecar")?;
+            let meta = serde_json::from_slice(&bytes).context("parse media sidecar")?;
+            for object in objects_for_sidecar(community, sha, &meta)? {
+                processed += 1;
+                pacer.wait().await;
+                if storage.head(&object.sharded).await? {
+                    skipped += 1;
+                    continue;
+                }
+                pacer.wait().await;
+                let source_exists = storage.head(&object.legacy).await?;
+                if !source_exists {
+                    anyhow::bail!(
+                        "legacy source missing: {} (checkpoint: {sidecar_key})",
+                        object.legacy
+                    );
+                }
+                if args.dry_run {
+                    tracing::info!(source = %object.legacy, destination = %object.sharded, "would copy");
+                } else {
+                    pacer.wait().await;
+                    storage.copy(&object.legacy, &object.sharded).await?;
+                    pacer.wait().await;
+                    if !storage.head(&object.sharded).await? {
+                        anyhow::bail!("destination verification failed: {}", object.sharded);
+                    }
+                    copied += 1;
+                }
+            }
+            checkpoint = Some(sidecar_key);
+        }
+        if !page.is_truncated {
+            break;
+        }
+        continuation = page.next_continuation_token;
+        if continuation.is_none() {
+            anyhow::bail!("truncated S3 listing returned no continuation token");
+        }
+    }
+    tracing::info!(processed, copied, skipped, dry_run = args.dry_run, checkpoint = ?checkpoint, "backfill complete");
+    Ok(())
+}

crates/buzz-media/src/bin/buzz-media-layout-delete-legacy.rs

@@ -0,0 +1,116 @@
+//! Delete verified legacy media payloads after migration reconciliation.
+
+use anyhow::{Context, Result};
+use buzz_media::migration::{objects_for_sidecar, parse_sidecar_key, RequestPacer};
+use buzz_media::MediaStorage;
+use clap::Parser;
+
+mod media_layout_common;
+use media_layout_common::CommonArgs;
+
+const CONFIRMATION: &str = "delete-verified-legacy-media";
+
+#[derive(Debug, Parser)]
+#[command(name = "buzz-media-layout-delete-legacy")]
+struct Args {
+    #[command(flatten)]
+    common: CommonArgs,
+    /// Preview is the safe default. Set false only after reconciliation.
+    #[arg(long, env = "BUZZ_MEDIA_MIGRATION_DRY_RUN", default_value_t = true, action = clap::ArgAction::Set)]
+    dry_run: bool,
+    /// Required with --dry-run=false to guard accidental destructive Jobs.
+    #[arg(long, env = "BUZZ_MEDIA_DELETE_CONFIRM")]
+    confirm: Option<String>,
+}
+
+/// Scan all selected sidecars before deleting anything. This two-pass design is
+/// important because one flat legacy CAS key can serve multiple communities:
+/// every community destination must exist before the shared source is removed.
+async fn scan(
+    storage: &MediaStorage,
+    common: &CommonArgs,
+    pacer: &mut RequestPacer,
+    delete: bool,
+    dry_run: bool,
+) -> Result<(u64, u64, Option<String>)> {
+    let mut continuation = None;
+    let mut start_after = common.start_after.clone();
+    let mut changed = 0_u64;
+    let mut skipped = 0_u64;
+    let mut checkpoint = start_after.clone();
+    loop {
+        pacer.wait().await;
+        let page = storage
+            .list_prefix_page(
+                "_meta/",
+                continuation.take(),
+                start_after.take(),
+                common.page_size,
+            )
+            .await
+            .context("list media sidecars")?;
+        for (sidecar_key, _) in page.objects {
+            let Some((community, sha)) = parse_sidecar_key(&sidecar_key) else {
+                tracing::warn!(key = %sidecar_key, "skipping malformed sidecar key");
+                continue;
+            };
+            pacer.wait().await;
+            let bytes = storage
+                .get(&sidecar_key)
+                .await
+                .context("read media sidecar")?;
+            let meta = serde_json::from_slice(&bytes).context("parse media sidecar")?;
+            for object in objects_for_sidecar(community, sha, &meta)? {
+                pacer.wait().await;
+                if !storage.head(&object.sharded).await? {
+                    anyhow::bail!(
+                        "refusing deletion: sharded destination missing: {} (checkpoint: {sidecar_key})",
+                        object.sharded
+                    );
+                }
+                if !delete {
+                    continue;
+                }
+                pacer.wait().await;
+                if !storage.head(&object.legacy).await? {
+                    skipped += 1;
+                    continue;
+                }
+                if dry_run {
+                    tracing::info!(key = %object.legacy, "would delete verified legacy object");
+                } else {
+                    pacer.wait().await;
+                    storage.delete(&object.legacy).await?;
+                    changed += 1;
+                }
+            }
+            checkpoint = Some(sidecar_key);
+        }
+        if !page.is_truncated {
+            break;
+        }
+        continuation = page.next_continuation_token;
+        if continuation.is_none() {
+            anyhow::bail!("truncated S3 listing returned no continuation token");
+        }
+    }
+    Ok((changed, skipped, checkpoint))
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    tracing_subscriber::fmt().with_target(false).init();
+    let args = Args::parse();
+    if !args.dry_run && args.confirm.as_deref() != Some(CONFIRMATION) {
+        anyhow::bail!("destructive mode requires --confirm={CONFIRMATION}");
+    }
+    let storage = args.common.storage()?;
+    let mut pacer = RequestPacer::new(args.common.requests_per_second);
+    let (_, _, verified_checkpoint) =
+        scan(&storage, &args.common, &mut pacer, false, args.dry_run).await?;
+    tracing::info!(checkpoint = ?verified_checkpoint, "all selected sharded destinations verified; beginning deletion pass");
+    let (deleted, skipped, checkpoint) =
+        scan(&storage, &args.common, &mut pacer, true, args.dry_run).await?;
+    tracing::info!(deleted, skipped, dry_run = args.dry_run, checkpoint = ?checkpoint, "legacy deletion complete");
+    Ok(())
+}

crates/buzz-media/src/bin/media_layout_common/mod.rs

@@ -0,0 +1,60 @@
+use anyhow::{bail, Context, Result};
+use buzz_media::{MediaConfig, MediaMigrationPhase, MediaStorage, S3AddressingStyle};
+use clap::Args;
+
+#[derive(Debug, Args)]
+pub struct CommonArgs {
+    #[arg(long, env = "BUZZ_S3_ENDPOINT")]
+    pub s3_endpoint: String,
+    #[arg(long, env = "BUZZ_S3_BUCKET")]
+    pub s3_bucket: String,
+    #[arg(long, env = "BUZZ_S3_REGION", default_value = "us-east-1")]
+    pub s3_region: String,
+    #[arg(long, env = "BUZZ_S3_ACCESS_KEY", default_value = "")]
+    pub s3_access_key: String,
+    #[arg(long, env = "BUZZ_S3_SECRET_KEY", default_value = "")]
+    pub s3_secret_key: String,
+    #[arg(long, env = "BUZZ_S3_ADDRESSING_STYLE", default_value = "path")]
+    pub s3_addressing_style: S3AddressingStyle,
+    /// Maximum S3 requests per second, including listing and verification.
+    #[arg(
+        long,
+        env = "BUZZ_MEDIA_MIGRATION_REQUESTS_PER_SECOND",
+        default_value_t = 25
+    )]
+    pub requests_per_second: u32,
+    /// Resume after this sidecar key. The final log line prints the next value.
+    #[arg(long, env = "BUZZ_MEDIA_MIGRATION_START_AFTER")]
+    pub start_after: Option<String>,
+    #[arg(long, env = "BUZZ_MEDIA_MIGRATION_PAGE_SIZE", default_value_t = 100)]
+    pub page_size: usize,
+}
+
+impl CommonArgs {
+    pub fn storage(&self) -> Result<MediaStorage> {
+        if self.requests_per_second == 0 {
+            bail!("requests-per-second must be greater than zero");
+        }
+        if !(1..=1000).contains(&self.page_size) {
+            bail!("page-size must be between 1 and 1000");
+        }
+        MediaStorage::new(&MediaConfig {
+            s3_endpoint: self.s3_endpoint.clone(),
+            s3_access_key: self.s3_access_key.clone(),
+            s3_secret_key: self.s3_secret_key.clone(),
+            s3_bucket: self.s3_bucket.clone(),
+            s3_region: self.s3_region.clone(),
+            s3_addressing_style: self.s3_addressing_style,
+            migration_phase: MediaMigrationPhase::DualReadLegacyWrite,
+            max_image_bytes: 1,
+            max_gif_bytes: 1,
+            max_video_bytes: 1,
+            max_file_bytes: 1,
+            public_base_url: "http://localhost/media".into(),
+            upload_records_enabled: false,
+            upload_ip_header: None,
+            upload_port_header: None,
+        })
+        .context("create media S3 client")
+    }
+}