- Offline CLI `upload-migrate OLD_CONFIG NEW_CONFIG --component HOST

[--execute]`: dry run по умолчанию, local ↔ S3 и S3 ↔ S3 без удаления
  source.
- Последовательное копирование с проверкой размера и полным SHA-256
  readback
  destination; конфликтующий объект не перезаписывается. Временный диск
  —
  максимум один переносимый файл дополнительно к постоянному
  destination.
- Durable migration journal позволяет продолжить прерванную операцию с
  теми
  же конфигурациями. Незавершённый или повреждённый journal блокирует
  startup;
  backend identity меняется только после проверки всех файлов.
- Native `uploads`/`upload_stats`, slots и JWT сохраняются;
  поддерживаются
  native локальные файлы без Rust completion sidecar. Миграция требует
  остановки RustXMPP и Prosody, использующих общий `data_path`.
- Multipart intent сохраняется до создания удалённого upload. Recovery
  находит потерянный UploadId через paginated listing и очищает только
  точный
  key принадлежащего slot. При сохранных metadata cleanup не зависит от
  lifecycle; lifecycle остаётся дополнительной защитой, а не гарантией.
- Перед completion сохраняется durable marker. Неопределённый результат
  completion при отсутствии подтверждённого объекта помещает только этот
  slot в quarantine: PUT 409, retention не удаляет его, maintenance
  повторяет
  сверку. Startup продолжается; сохраняется single-use семантика.
- Добавлены network regressions миграции и расширены проверки multipart
  recovery, потерянных ответов, отмены запросов и конфликтов
  destination.

Настройка, команды миграции, recovery и ограничения отката:
[docs/HTTP-UPLOAD.md](docs/HTTP-UPLOAD.md).

Команды, результаты и границы подтверждения, включая qualification
реального
object store и 10 GiB:
[docs/VALIDATION-0.7.9.md](docs/VALIDATION-0.7.9.md).

Итог: 265 тестов, нагрузка 500 TLS клиентов и native reference —
успешно.
Реальные 10 GiB через MinIO (описанный test adapter), 1280parts и полный
SHA-256
пройдены; подробности — [VALIDATION-0.7.9](docs/VALIDATION-0.7.9.md).

src/http_config.rs

@@ -28,7 +28,7 @@ fn string(options: &Options, key: &str) -> Result<Option<String>> {
         })
         .transpose()
 }
-fn valid_authority(authority: &axum::http::uri::Authority) -> bool {
+pub(crate) fn valid_authority(authority: &axum::http::uri::Authority) -> bool {
     let raw = authority.as_str();
     if raw.contains('@') {
         return false;

src/http_service.rs

@@ -1,6 +1,7 @@
 //! Streaming XEP-0363 upload endpoint. Slots are single-use and bounded before streaming.
 use crate::{
     logging::{Level, Logger},
+    object_store::{ObjectStore, PART_SIZE, UploadBackendConfig},
     storage,
     wire::escape,
 };
@@ -56,13 +57,29 @@ struct Record {
     token: String,
     complete: bool,
     #[serde(default)]
+    upload_id: Option<String>,
+    // Durable before CreateMultipartUpload: UploadId may be lost in a crash.
+    #[serde(default)]
+    multipart_intent: bool,
+    #[serde(default)]
+    completion_attempted: bool,
+    #[serde(default)]
     xml: Option<String>,
 }
+#[derive(Debug)]
+struct CompletionUncertain;
+impl std::fmt::Display for CompletionUncertain {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("S3 completion outcome uncertain; reservation quarantined for reconciliation")
+    }
+}
+impl std::error::Error for CompletionUncertain {}
 #[derive(Clone)]
 pub struct UploadService {
     config: Arc<UploadConfig>,
     records: Arc<Mutex<BTreeMap<String, Record>>>,
     active: Arc<Mutex<BTreeSet<String>>>,
+    backend: Option<Arc<ObjectStore>>,
 }
 pub struct Slot {
     pub url: String,
@@ -105,6 +122,13 @@ fn url_decode(value: &str) -> Option<Vec<u8>> {
 }
 impl UploadService {
     pub fn new(config: UploadConfig) -> Result<Self> {
+        Self::new_with_backend(config, UploadBackendConfig::Local)
+    }
+    pub fn new_with_backend(config: UploadConfig, backend: UploadBackendConfig) -> Result<Self> {
+        let backend = match backend {
+            UploadBackendConfig::Local => None,
+            UploadBackendConfig::S3(c) => Some(Arc::new(ObjectStore::new(c)?)),
+        };
         ensure!(
             config.base_url.starts_with("https://") || config.base_url.starts_with("http://"),
             "invalid upload base URL"
@@ -131,7 +155,24 @@ impl UploadService {
                 .attribute("size")
                 .ok_or_else(|| anyhow::anyhow!("upload size missing"))?
                 .parse()?;
-            let token = records.remove(&e.id).map(|r| r.token).unwrap_or_default();
+            let cached = records.remove(&e.id);
+            let complete = cached.as_ref().is_some_and(|r| {
+                r.complete && r.owner == e.peer && r.filename == filename && r.size == size
+            });
+            let upload_id = cached
+                .as_ref()
+                .filter(|r| r.owner == e.peer && r.filename == filename && r.size == size)
+                .and_then(|r| r.upload_id.clone());
+            let multipart_intent = cached.as_ref().is_some_and(|r| {
+                r.owner == e.peer && r.filename == filename && r.size == size && r.multipart_intent
+            });
+            let completion_attempted = cached.as_ref().is_some_and(|r| {
+                r.owner == e.peer
+                    && r.filename == filename
+                    && r.size == size
+                    && r.completion_attempted
+            });
+            let token = cached.map(|r| r.token).unwrap_or_default();
             authoritative.insert(
                 e.id,
                 Record {
@@ -144,20 +185,151 @@ impl UploadService {
                         .into(),
                     created: e.when as u64,
                     token,
-                    complete: false,
+                    complete,
+                    upload_id,
+                    multipart_intent,
+                    completion_attempted,
                     xml: Some(e.xml),
                 },
             );
         }
         records = authoritative;
+        let identity = backend
+            .as_ref()
+            .map(|b| b.identity())
+            .unwrap_or_else(|| serde_json::json!({"backend":"local"}));
+        let identity_path = config
+            .root
+            .join(storage::encode_name(&config.host))
+            .join("rust_upload_backend.json");
+        if identity_path.exists() {
+            let previous: serde_json::Value =
+                serde_json::from_slice(&std::fs::read(&identity_path)?)?;
+            ensure!(
+                records.is_empty() || previous == identity,
+                "upload backend changed while slots still exist"
+            );
+        } else {
+            ensure!(
+                backend.is_none() || records.is_empty(),
+                "existing upload slots require an explicit storage migration"
+            );
+        }
+        storage::atomic_write(&identity_path, &serde_json::to_vec(&identity)?)?;
         let s = Self {
             config: Arc::new(config),
             records: Arc::new(Mutex::new(records)),
             active: Arc::new(Mutex::new(BTreeSet::new())),
+            backend,
         };
-        s.prune()?;
+        // Remote pending records must be reconciled before the abandoned-slot
+        // timeout is applied. The caller awaits recover_backend before listening.
+        if s.backend.is_none() {
+            s.prune()?;
+        }
         Ok(s)
     }
+    /// Recover durable multipart state before enabling listeners or slot requests.
+    pub async fn recover_backend(&self) -> Result<()> {
+        if self.backend.is_none() {
+            return Ok(());
+        }
+        let pending: Vec<_> = self
+            .records
+            .lock()
+            .unwrap()
+            .iter()
+            .filter(|(_, r)| {
+                !r.complete || r.upload_id.is_some() || r.multipart_intent || r.completion_attempted
+            })
+            .map(|(id, r)| (id.clone(), r.clone()))
+            .collect();
+        for (id, record) in pending {
+            if let Err(error) = self.reconcile_record(id, record).await
+                && error.downcast_ref::<CompletionUncertain>().is_none()
+            {
+                return Err(error);
+            }
+        }
+        let service = self.clone();
+        tokio::task::spawn_blocking(move || service.prune()).await??;
+        Ok(())
+    }
+    async fn reconcile_record(&self, id: String, record: Record) -> Result<()> {
+        let Some(backend) = &self.backend else {
+            return Ok(());
+        };
+        if !record.complete || record.multipart_intent {
+            // Also cover old journals from before multipart_intent existed.
+            // This exact durable reservation is exclusively claimed by the
+            // startup barrier or the active-set lock. Never sweep a bucket.
+            backend.abort_pending(&id).await?;
+        }
+        if let Some(upload_id) = &record.upload_id {
+            backend.abort(&id, upload_id).await?;
+        }
+        // Completion may have committed while abort was in flight. Only the
+        // post-abort HEAD can establish the current single-use object state.
+        let head = backend.read(&id, true, None).await?;
+        let complete = match head.status() {
+            StatusCode::OK => {
+                ensure!(
+                    head.headers()
+                        .get("content-length")
+                        .and_then(|v| v.to_str().ok())
+                        .and_then(|v| v.parse::<u64>().ok())
+                        == Some(record.size),
+                    "S3 recovered object size mismatch"
+                );
+                true
+            }
+            StatusCode::NOT_FOUND => false,
+            _ => anyhow::bail!("S3 recovery HEAD failed"),
+        };
+        if !complete && record.completion_attempted {
+            return Err(CompletionUncertain.into());
+        }
+        let service = self.clone();
+        tokio::task::spawn_blocking(move || -> Result<()> {
+            let mut records = service.records.lock().unwrap();
+            if let Some(r) = records.get_mut(&id)
+                && r.upload_id == record.upload_id
+                && r.multipart_intent == record.multipart_intent
+            {
+                r.complete = complete;
+                r.upload_id = None;
+                r.multipart_intent = false;
+                r.completion_attempted = false;
+            }
+            service.persist(&records)
+        })
+        .await??;
+        Ok(())
+    }
+    /// Read-only identifiers for unresolved completion attempts. Credentials,
+    /// bearer tokens and object contents are never included.
+    pub fn quarantined_slots(&self) -> Vec<String> {
+        let records = self.records.lock().unwrap();
+        let active = self.active.lock().unwrap();
+        records
+            .iter()
+            .filter(|(id, r)| r.completion_attempted && !r.complete && !active.contains(*id))
+            .map(|(id, _)| id.clone())
+            .collect()
+    }
+    fn claim_upload(&self, id: &str) -> std::result::Result<Record, StatusCode> {
+        // Match prune's lock order, and read all mutable completion state under
+        // the same records lock used to establish the active claim.
+        let records = self.records.lock().unwrap();
+        let record = records.get(id).ok_or(StatusCode::GONE)?;
+        if record.complete || (self.backend.is_none() && self.file(id).exists()) {
+            return Err(StatusCode::CONFLICT);
+        }
+        if !self.active.lock().unwrap().insert(id.to_owned()) {
+            return Err(StatusCode::CONFLICT);
+        }
+        Ok(record.clone())
+    }
     fn persist(&self, r: &BTreeMap<String, Record>) -> Result<()> {
         let mut entries: Vec<_> = r
             .iter()
@@ -214,11 +386,41 @@ impl UploadService {
     }
     pub fn prune(&self) -> Result<()> {
         let mut r = self.records.lock().unwrap();
-        let active = self.active.lock().unwrap();
+        let mut active = self.active.lock().unwrap();
+        if self.backend.is_some()
+            && let Ok(handle) = tokio::runtime::Handle::try_current()
+        {
+            for (id, record) in r
+                .iter()
+                .filter(|(id, r)| {
+                    (r.upload_id.is_some() || r.multipart_intent || r.completion_attempted)
+                        && !active.contains(*id)
+                })
+                .map(|(id, r)| (id.clone(), r.clone()))
+                .collect::<Vec<_>>()
+            {
+                active.insert(id.clone());
+                let service = self.clone();
+                handle.spawn(async move {
+                    let recovered = service.reconcile_record(id.clone(), record).await.is_ok();
+                    tokio::task::spawn_blocking(move || {
+                        service.active.lock().unwrap().remove(&id);
+                        if recovered {
+                            let _ = service.prune();
+                        }
+                    })
+                    .await
+                    .ok();
+                });
+            }
+        }
         let obsolete: Vec<String> = r
             .iter()
             .filter(|(id, v)| {
                 !active.contains(*id)
+                    && v.upload_id.is_none()
+                    && !v.multipart_intent
+                    && !v.completion_attempted
                     && (self
                         .config
                         .expires_after
@@ -230,6 +432,29 @@ impl UploadService {
             .map(|(id, _)| id.clone())
             .collect();
         for id in obsolete {
+            if let Some(backend) = self.backend.clone() {
+                // Keep the reservation and metadata until remote deletion succeeds.
+                // The active claim prevents duplicate workers and concurrent PUTs.
+                if let Ok(handle) = tokio::runtime::Handle::try_current() {
+                    active.insert(id.clone());
+                    let service = self.clone();
+                    handle.spawn(async move {
+                        let deleted = backend.delete(&id).await.is_ok();
+                        tokio::task::spawn_blocking(move || {
+                            let mut records = service.records.lock().unwrap();
+                            let mut active = service.active.lock().unwrap();
+                            if deleted {
+                                records.remove(&id);
+                                let _ = service.persist(&records);
+                            }
+                            active.remove(&id);
+                        })
+                        .await
+                        .ok();
+                    });
+                }
+                continue;
+            }
             match std::fs::remove_file(self.file(&id)) {
                 Ok(()) => {}
                 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
@@ -314,6 +539,9 @@ impl UploadService {
                 created: now(),
                 token,
                 complete: false,
+                upload_id: None,
+                multipart_intent: false,
+                completion_attempted: false,
                 xml: None,
             },
         );
@@ -358,9 +586,10 @@ impl UploadService {
             let config = Arc::downgrade(&self.config);
             let records = Arc::downgrade(&self.records);
             let active = Arc::downgrade(&self.active);
+            let backend = self.backend.clone();
             handle.spawn(async move {
                 loop {
-                    tokio::time::sleep(std::time::Duration::from_secs(3600)).await;
+                    tokio::time::sleep(std::time::Duration::from_secs(60)).await;
                     let (Some(config), Some(records), Some(active)) =
                         (config.upgrade(), records.upgrade(), active.upgrade())
                     else {
@@ -370,6 +599,7 @@ impl UploadService {
                         config,
                         records,
                         active,
+                        backend: backend.clone(),
                     };
                     let _ = tokio::task::spawn_blocking(move || service.prune()).await;
                 }
@@ -380,6 +610,14 @@ impl UploadService {
             .with_state(self.clone())
             .layer(axum::Extension(UploadPath(prefix)))
     }
+    pub async fn handle_at(&self, req: Request, internal_path: &str) -> Response {
+        endpoint(
+            State(self.clone()),
+            axum::Extension(UploadPath(internal_path.trim_end_matches('/').to_owned())),
+            req,
+        )
+        .await
+    }
     pub async fn serve(&self, listener: TcpListener, tls: Option<TlsAcceptor>) -> Result<()> {
         serve_router(listener, tls, self.router()).await
     }
@@ -496,6 +734,7 @@ async fn serve_router_inner(
                             Some(crate::http_transport::HttpRoute::Websocket) => "websocket",
                             Some(crate::http_transport::HttpRoute::Bosh) => "bosh",
                             Some(crate::http_transport::HttpRoute::Altconnect) => "altconnect",
+                            Some(crate::http_transport::HttpRoute::Upload) => "upload",
                             None => default_route,
                         };
                         let status = response.status();
@@ -693,20 +932,15 @@ async fn handle_at(s: UploadService, req: Request, prefix: &str) -> Response {
         {
             return response(StatusCode::PAYLOAD_TOO_LARGE);
         }
-        if record.complete || s.file(id).exists() {
+        if record.complete || (s.backend.is_none() && s.file(id).exists()) {
             return response(StatusCode::CONFLICT);
         };
-        {
-            // Use the same lock order as prune: a reservation must still exist
-            // when it becomes active, so pruning cannot orphan an upload.
-            let records = s.records.lock().unwrap();
-            if !records.contains_key(id) {
-                return response(StatusCode::GONE);
-            }
-            if !s.active.lock().unwrap().insert(id.into()) {
-                return response(StatusCode::CONFLICT);
-            }
-        }
+        // Refresh the snapshot atomically with the exclusive claim. A request
+        // may have waited while an earlier PUT committed or was quarantined.
+        let record = match s.claim_upload(id) {
+            Ok(record) => record,
+            Err(status) => return response(status),
+        };
         let dest = s.file(id);
         let temp = dest.with_extension(format!("{}.tmp", uuid::Uuid::new_v4()));
         let _guard = UploadGuard {
@@ -714,6 +948,212 @@ async fn handle_at(s: UploadService, req: Request, prefix: &str) -> Response {
             id: id.into(),
             temp: temp.clone(),
         };
+        if let Some(backend) = &s.backend {
+            if (record.upload_id.is_some()
+                || record.multipart_intent
+                || record.completion_attempted)
+                && let Err(error) = s.reconcile_record(id.to_owned(), record.clone()).await
+            {
+                return response(if error.downcast_ref::<CompletionUncertain>().is_some() {
+                    StatusCode::CONFLICT
+                } else {
+                    StatusCode::BAD_GATEWAY
+                });
+            }
+            // Reconcile a remote commit whose response or local metadata write
+            // was interrupted, rather than overwriting a single-use object.
+            match backend.read(id, true, None).await {
+                Ok(reply) if reply.status() == StatusCode::NOT_FOUND => {}
+                Ok(reply) if reply.status() == StatusCode::OK => {
+                    if reply
+                        .headers()
+                        .get("content-length")
+                        .and_then(|v| v.to_str().ok())
+                        .and_then(|v| v.parse::<u64>().ok())
+                        != Some(record.size)
+                    {
+                        return response(StatusCode::BAD_GATEWAY);
+                    }
+                    let service = s.clone();
+                    let id = id.to_owned();
+                    return response(
+                        tokio::task::spawn_blocking(move || {
+                            let _guard = _guard;
+                            let mut records = service.records.lock().unwrap();
+                            if let Some(r) = records.get_mut(&id) {
+                                r.complete = true;
+                                r.upload_id = None;
+                                r.multipart_intent = false;
+                                r.completion_attempted = false;
+                            }
+                            if service.persist(&records).is_err() {
+                                StatusCode::INTERNAL_SERVER_ERROR
+                            } else {
+                                StatusCode::CONFLICT
+                            }
+                        })
+                        .await
+                        .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR),
+                    );
+                }
+                _ => return response(StatusCode::BAD_GATEWAY),
+            }
+            // Hold the single-use claim across initiation and journaling even
+            // if the HTTP request is cancelled. A persisted intent closes the
+            // crash window before S3 returns its UploadId.
+            let service = s.clone();
+            let backend = backend.clone();
+            let slot_id = id.to_owned();
+            let content_type = record.content_type.clone();
+            let initiation = tokio::spawn(async move {
+                let guard = _guard;
+                let journal = service.clone();
+                let journal_id = slot_id.clone();
+                tokio::task::spawn_blocking(move || -> Result<()> {
+                    let mut records = journal.records.lock().unwrap();
+                    let r = records
+                        .get_mut(&journal_id)
+                        .ok_or_else(|| anyhow::anyhow!("upload reservation missing"))?;
+                    r.multipart_intent = true;
+                    journal.persist(&records)
+                })
+                .await??;
+                let mut upload = backend.begin(&slot_id, &content_type).await?;
+                let upload_id = upload.upload_id().to_owned();
+                let cleanup = service.clone();
+                upload.after_abort(move || {
+                    // HEAD must reconcile a possible remote commit before
+                    // clearing the intent, even when abort returns 404.
+                    tokio::task::spawn_blocking(move || {
+                        let _ = cleanup.prune();
+                    });
+                });
+                tokio::task::spawn_blocking(move || -> Result<()> {
+                    let mut records = service.records.lock().unwrap();
+                    let r = records
+                        .get_mut(&slot_id)
+                        .ok_or_else(|| anyhow::anyhow!("upload reservation missing"))?;
+                    r.upload_id = Some(upload_id);
+                    service.persist(&records)
+                })
+                .await??;
+                Ok::<_, anyhow::Error>((upload, guard))
+            });
+            let (mut upload, _guard) = match initiation.await {
+                Ok(Ok(value)) => value,
+                _ => return response(StatusCode::BAD_GATEWAY),
+            };
+            let mut body = req.into_body();
+            let mut count = 0u64;
+            let mut part = Vec::with_capacity(PART_SIZE);
+            loop {
+                let frame =
+                    match tokio::time::timeout(std::time::Duration::from_secs(120), body.frame())
+                        .await
+                    {
+                        Ok(Some(Ok(frame))) => frame,
+                        Ok(Some(Err(_))) => return response(StatusCode::BAD_REQUEST),
+                        Ok(None) => break,
+                        Err(_) => return response(StatusCode::REQUEST_TIMEOUT),
+                    };
+                if let Ok(bytes) = frame.into_data() {
+                    count = count.saturating_add(bytes.len() as u64);
+                    if count > record.size {
+                        return response(StatusCode::PAYLOAD_TOO_LARGE);
+                    }
+                    let mut bytes = bytes.as_ref();
+                    while !bytes.is_empty() {
+                        let take = (PART_SIZE - part.len()).min(bytes.len());
+                        part.extend_from_slice(&bytes[..take]);
+                        bytes = &bytes[take..];
+                        if part.len() == PART_SIZE
+                            && upload
+                                .part(std::mem::replace(&mut part, Vec::with_capacity(PART_SIZE)))
+                                .await
+                                .is_err()
+                        {
+                            return response(StatusCode::BAD_GATEWAY);
+                        }
+                    }
+                }
+            }
+            if count != record.size {
+                return response(StatusCode::BAD_REQUEST);
+            }
+            if (!part.is_empty() || count == 0) && upload.part(part).await.is_err() {
+                return response(StatusCode::BAD_GATEWAY);
+            }
+            // Once completing begins, finish the durable commit even if the
+            // HTTP client disconnects. Keep the single-use claim until then.
+            let service = s.clone();
+            let id = id.to_owned();
+            let commit = tokio::spawn(async move {
+                let _guard = _guard;
+                let journal = service.clone();
+                let journal_id = id.clone();
+                if !matches!(
+                    tokio::task::spawn_blocking(move || -> Result<()> {
+                        let mut records = journal.records.lock().unwrap();
+                        let r = records
+                            .get_mut(&journal_id)
+                            .ok_or_else(|| anyhow::anyhow!("upload reservation missing"))?;
+                        r.completion_attempted = true;
+                        journal.persist(&records)
+                    })
+                    .await,
+                    Ok(Ok(()))
+                ) {
+                    return StatusCode::INTERNAL_SERVER_ERROR;
+                }
+                if let Err(error) = upload.complete().await {
+                    if error
+                        .downcast_ref::<crate::object_store::CompletionRejected>()
+                        .is_some()
+                    {
+                        // Explicit terminal rejection is retryable after HEAD
+                        // reconciliation. Network/XML uncertainty stays frozen.
+                        let journal = service.clone();
+                        let journal_id = id.clone();
+                        let _ = tokio::task::spawn_blocking(move || -> Result<()> {
+                            let mut records = journal.records.lock().unwrap();
+                            if let Some(r) = records.get_mut(&journal_id) {
+                                r.completion_attempted = false;
+                            }
+                            journal.persist(&records)
+                        })
+                        .await;
+                    }
+                    // Multipart::drop may finish its abort while this commit
+                    // still owns the active claim. Its callback's prune then
+                    // correctly skips the slot, so it cannot be our only
+                    // reconciliation trigger. Reconcile under this same claim
+                    // before returning the ambiguous completion response.
+                    let pending = service.records.lock().unwrap().get(&id).cloned();
+                    if let Some(record) = pending {
+                        let _ = service.reconcile_record(id.clone(), record).await;
+                    }
+                    return StatusCode::BAD_GATEWAY;
+                }
+                tokio::task::spawn_blocking(move || {
+                    let _guard = _guard;
+                    let mut records = service.records.lock().unwrap();
+                    if let Some(r) = records.get_mut(&id) {
+                        r.complete = true;
+                        r.upload_id = None;
+                        r.multipart_intent = false;
+                        r.completion_attempted = false;
+                    }
+                    if service.persist(&records).is_err() {
+                        StatusCode::INTERNAL_SERVER_ERROR
+                    } else {
+                        StatusCode::CREATED
+                    }
+                })
+                .await
+                .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR)
+            });
+            return response(commit.await.unwrap_or(StatusCode::INTERNAL_SERVER_ERROR));
+        }
         if tokio::fs::create_dir_all(dest.parent().unwrap())
             .await
             .is_err()
@@ -769,36 +1209,78 @@ async fn handle_at(s: UploadService, req: Request, prefix: &str) -> Response {
                 return response(StatusCode::INTERNAL_SERVER_ERROR);
             }
         }
-        let mut records = s.records.lock().unwrap();
-        if let Some(r) = records.get_mut(id) {
-            r.complete = true;
-        }
-        if s.persist(&records).is_err() {
-            return response(StatusCode::INTERNAL_SERVER_ERROR);
-        };
-        return response(StatusCode::CREATED);
+        let id = id.to_owned();
+        return response(
+            tokio::task::spawn_blocking(move || {
+                let _guard = _guard;
+                let mut records = s.records.lock().unwrap();
+                if let Some(r) = records.get_mut(&id) {
+                    r.complete = true;
+                }
+                if s.persist(&records).is_err() {
+                    StatusCode::INTERNAL_SERVER_ERROR
+                } else {
+                    StatusCode::CREATED
+                }
+            })
+            .await
+            .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR),
+        );
     }
     if req.method() != "GET" && req.method() != "HEAD" {
         return response(StatusCode::METHOD_NOT_ALLOWED);
     }
-    let Ok(mut file) = tokio::fs::File::open(s.file(id)).await else {
+    if s.backend.is_some() && !record.complete {
         return response(StatusCode::GONE);
+    }
+    let mut local_file = if s.backend.is_none() {
+        match tokio::fs::File::open(s.file(id)).await {
+            Ok(file) => Some(file),
+            Err(_) => return response(StatusCode::GONE),
+        }
+    } else {
+        None
     };
     let modified = chrono::DateTime::from_timestamp(record.created as i64, 0)
         .map(|d| d.format("%a, %d %b %Y %H:%M:%S GMT").to_string())
         .unwrap_or_default();
-    if req
-        .headers()
-        .get("if-modified-since")
-        .and_then(|v| v.to_str().ok())
-        == Some(modified.as_str())
-    {
-        return response(StatusCode::NOT_MODIFIED);
+    let etag = format!("\"{id}-{}\"", record.size);
+    let date_matches = |value: &str| {
+        chrono::DateTime::parse_from_rfc2822(value)
+            .ok()
+            .is_some_and(|d| d.timestamp() >= record.created as i64)
+    };
+    let not_modified = if let Some(value) = req.headers().get("if-none-match") {
+        value.to_str().ok().is_some_and(|v| {
+            v.split(',').any(|candidate| {
+                let candidate = candidate.trim();
+                candidate == "*" || candidate.strip_prefix("W/").unwrap_or(candidate) == etag
+            })
+        })
+    } else {
+        req.headers()
+            .get("if-modified-since")
+            .and_then(|v| v.to_str().ok())
+            .is_some_and(date_matches)
+    };
+    if not_modified {
+        return Response::builder()
+            .status(StatusCode::NOT_MODIFIED)
+            .header("etag", &etag)
+            .header("last-modified", &modified)
+            .body(Body::empty())
+            .unwrap();
     }
+    let range_allowed = req.headers().get("if-range").is_none_or(|value| {
+        value
+            .to_str()
+            .ok()
+            .is_some_and(|v| v == etag || date_matches(v))
+    });
     let mut status = StatusCode::OK;
     let mut start = 0;
     let mut end = record.size.saturating_sub(1);
-    if let Some(range) = req.headers().get("range") {
+    if let Some(range) = req.headers().get("range").filter(|_| range_allowed) {
         let parsed = range
             .to_str()
             .ok()
@@ -833,14 +1315,50 @@ async fn handle_at(s: UploadService, req: Request, prefix: &str) -> Response {
         status = StatusCode::PARTIAL_CONTENT;
     }
     let length = if record.size == 0 { 0 } else { end - start + 1 };
-    if file.seek(std::io::SeekFrom::Start(start)).await.is_err() {
-        return response(StatusCode::INTERNAL_SERVER_ERROR);
-    };
     use tokio::io::AsyncReadExt;
-    let body = if req.method() == "HEAD" {
-        Body::empty()
+    let body = if let Some(backend) = &s.backend {
+        let range = (status == StatusCode::PARTIAL_CONTENT).then(|| format!("bytes={start}-{end}"));
+        let Ok(reply) = backend
+            .read(id, req.method() == "HEAD", range.as_deref())
+            .await
+        else {
+            return response(StatusCode::BAD_GATEWAY);
+        };
+        if reply.status() == StatusCode::NOT_FOUND {
+            return response(StatusCode::GONE);
+        }
+        if reply.status() != status {
+            return response(StatusCode::BAD_GATEWAY);
+        }
+        if reply
+            .headers()
+            .get("content-length")
+            .and_then(|v| v.to_str().ok())
+            .and_then(|v| v.parse::<u64>().ok())
+            != Some(length)
+        {
+            return response(StatusCode::BAD_GATEWAY);
+        }
+        if status == StatusCode::PARTIAL_CONTENT
+            && reply
+                .headers()
+                .get("content-range")
+                .and_then(|v| v.to_str().ok())
+                != Some(format!("bytes {start}-{end}/{}", record.size).as_str())
+        {
+            return response(StatusCode::BAD_GATEWAY);
+        }
+        reply.into_body()
     } else {
-        Body::from_stream(ReaderStream::new(file.take(length)))
+        let mut file = local_file.take().unwrap();
+        if file.seek(std::io::SeekFrom::Start(start)).await.is_err() {
+            return response(StatusCode::INTERNAL_SERVER_ERROR);
+        }
+        if req.method() == "HEAD" {
+            Body::empty()
+        } else {
+            Body::from_stream(ReaderStream::new(file.take(length)))
+        }
     };
     let disposition = if record.content_type == "text/plain"
         || ["image/", "video/", "audio/"]
@@ -869,6 +1387,7 @@ async fn handle_at(s: UploadService, req: Request, prefix: &str) -> Response {
         )
         .header("access-control-allow-origin", "*")
         .header("accept-ranges", "bytes")
+        .header("etag", etag)
         .header("last-modified", modified)
         .header("cache-control", "max-age=31556952, immutable")
         .header("x-frame-options", "DENY")
@@ -991,6 +1510,60 @@ mod tests {
         );
     }
 
+    #[test]
+    fn upload_claim_refreshes_completion_state_after_an_earlier_snapshot() {
+        let temp = tempfile::tempdir().unwrap();
+        let service = UploadService::new(UploadConfig {
+            root: temp.path().into(),
+            host: "upload.example.test".into(),
+            base_url: "https://upload.example.test/file_share".into(),
+            size_limit: 1,
+            daily_quota: 3,
+            global_quota: 3,
+            expires_after: None,
+            allowed_types: vec![],
+            secret: None,
+        })
+        .unwrap();
+        service
+            .request_slot("a@example.test", "pending", 1, "text/plain")
+            .unwrap();
+        let (id, stale) = service
+            .records
+            .lock()
+            .unwrap()
+            .iter()
+            .next()
+            .map(|(id, r)| (id.clone(), r.clone()))
+            .unwrap();
+        assert!(!stale.completion_attempted);
+        // Another PUT finishes ambiguously after this request has read its
+        // initial snapshot, but before it acquires the active claim.
+        service
+            .records
+            .lock()
+            .unwrap()
+            .get_mut(&id)
+            .unwrap()
+            .completion_attempted = true;
+        let fresh = service.claim_upload(&id).unwrap();
+        assert!(fresh.completion_attempted);
+        assert!(service.active.lock().unwrap().contains(&id));
+        service.active.lock().unwrap().remove(&id);
+        service
+            .records
+            .lock()
+            .unwrap()
+            .get_mut(&id)
+            .unwrap()
+            .complete = true;
+        assert!(matches!(
+            service.claim_upload(&id),
+            Err(StatusCode::CONFLICT)
+        ));
+        assert!(!service.active.lock().unwrap().contains(&id));
+    }
+
     #[test]
     fn abandoned_slots_release_quota_without_pruning_existing_or_active_files() {
         let temp = tempfile::tempdir().unwrap();

src/http_transport.rs

@@ -30,6 +30,7 @@ pub enum HttpRoute {
     Websocket,
     Bosh,
     Altconnect,
+    Upload,
 }
 
 #[derive(Clone)]

src/lib.rs

@@ -47,3 +47,7 @@ pub mod muc_recovery;
 pub mod muc_history;
 
 pub mod http_altconnect;
+
+pub mod object_store;
+
+pub mod upload_migration;

src/main.rs

@@ -12,6 +12,19 @@ async fn main() {
 async fn run() -> Result<()> {
     let args: Vec<_> = env::args().skip(1).collect();
     match args.first().map(String::as_str) {
+        Some("upload-migrate") if args.len() == 5 || args.len() == 6 => {
+            anyhow::ensure!(
+                args[3] == "--component" && (args.len() == 5 || args[5] == "--execute"),
+                "usage: upload-migrate OLD_CONFIG NEW_CONFIG --component HOST [--execute]"
+            );
+            rust_xmpp::upload_migration::migrate(
+                Path::new(&args[1]),
+                Path::new(&args[2]),
+                &args[4],
+                args.len() == 6,
+            )
+            .await?;
+        }
         Some("check") => {
             let path = args
                 .get(1)
@@ -108,7 +121,7 @@ async fn run() -> Result<()> {
             println!("Account created: {}@{}", args[2], domain);
         }
         Some("--help") | None => println!(
-            "rust-xmpp check-config PATH\nrust-xmpp convert PATH json|yaml|toml OUTPUT\nrust-xmpp --version\n\nExperimental runtime: serve CONFIG --experimental\ncreate-user CONFIG USER --password-stdin\ncheck [CONFIG] (running server healthcheck)\nshell CONFIG COMMAND\n\nSTARTTLS, SCRAM/PLAIN, local messaging, roster, MAM, MUC, PEP/pubsub, stream management, push, federation and HTTP transports are implemented with limitations; see docs/COMPATIBILITY.md."
+            "rust-xmpp upload-migrate OLD_CONFIG NEW_CONFIG --component HOST [--execute]\nrust-xmpp check-config PATH\nrust-xmpp convert PATH json|yaml|toml OUTPUT\nrust-xmpp --version\n\nExperimental runtime: serve CONFIG --experimental\ncreate-user CONFIG USER --password-stdin\ncheck [CONFIG] (running server healthcheck)\nshell CONFIG COMMAND\n\nSTARTTLS, SCRAM/PLAIN, local messaging, roster, MAM, MUC, PEP/pubsub, stream management, push, federation and HTTP transports are implemented with limitations; see docs/COMPATIBILITY.md."
         ),
         _ => bail!("invalid command; use --help"),
     }

src/object_store.rs

@@ -0,0 +1,605 @@
+//! Path-style S3-compatible storage with authenticated, bounded multipart writes.
+use anyhow::{Result, ensure};
+use axum::{
+    body::{Body, Bytes},
+    http::{Request, Response, Uri},
+};
+use hmac::{Hmac, Mac};
+use http_body_util::{BodyExt, Full, Limited};
+use sha2::{Digest, Sha256};
+use std::{collections::BTreeMap, sync::Arc, time::Duration};
+use tokio::{
+    io::{AsyncRead, AsyncWrite},
+    net::TcpStream,
+};
+use tokio_rustls::{
+    TlsConnector,
+    rustls::{ClientConfig, pki_types::ServerName},
+};
+
+#[derive(Clone, Default)]
+pub enum UploadBackendConfig {
+    #[default]
+    Local,
+    S3(S3Config),
+}
+// Deliberately no Debug implementation: credentials must never enter diagnostics.
+#[derive(Clone)]
+pub struct S3Config {
+    pub endpoint: String,
+    pub bucket: String,
+    pub region: String,
+    pub access_key: String,
+    pub secret_key: String,
+    pub session_token: Option<String>,
+    pub key_prefix: String,
+}
+#[derive(Clone)]
+pub struct ObjectStore {
+    config: S3Config,
+    endpoint: Uri,
+    tls: Option<Arc<ClientConfig>>,
+    uploads: Arc<tokio::sync::Semaphore>,
+}
+trait Transport: AsyncRead + AsyncWrite + Unpin + Send {}
+impl<T: AsyncRead + AsyncWrite + Unpin + Send> Transport for T {}
+fn encode(s: &str) -> String {
+    s.bytes()
+        .map(|b| {
+            if b.is_ascii_alphanumeric() || b"-._~".contains(&b) {
+                (b as char).to_string()
+            } else {
+                format!("%{b:02X}")
+            }
+        })
+        .collect()
+}
+fn decode_key(value: &str) -> Result<String> {
+    let mut bytes = value.bytes();
+    let mut decoded = Vec::with_capacity(value.len());
+    while let Some(byte) = bytes.next() {
+        if byte == b'%' {
+            let high = bytes.next().and_then(|v| (v as char).to_digit(16));
+            let low = bytes.next().and_then(|v| (v as char).to_digit(16));
+            match (high, low) {
+                (Some(h), Some(l)) => decoded.push((h * 16 + l) as u8),
+                _ => anyhow::bail!("invalid S3 listing key encoding"),
+            }
+        } else {
+            decoded.push(byte);
+        }
+    }
+    Ok(String::from_utf8(decoded)?)
+}
+fn mac(key: &[u8], data: &str) -> Vec<u8> {
+    let mut h = Hmac::<Sha256>::new_from_slice(key).unwrap();
+    h.update(data.as_bytes());
+    h.finalize().into_bytes().to_vec()
+}
+impl ObjectStore {
+    pub fn new(config: S3Config) -> Result<Self> {
+        let endpoint: Uri = config
+            .endpoint
+            .parse()
+            .map_err(|_| anyhow::anyhow!("invalid S3 endpoint"))?;
+        let scheme = endpoint.scheme_str().unwrap_or("");
+        let host = endpoint.host().unwrap_or("").trim_matches(['[', ']']);
+        ensure!(
+            !host.is_empty()
+                && endpoint.query().is_none()
+                && crate::http_config::valid_authority(endpoint.authority().unwrap())
+                && !config.endpoint.contains('#')
+                && !endpoint.path().contains('%')
+                && !endpoint.path().split('/').any(|p| p == "." || p == ".."),
+            "invalid S3 endpoint"
+        );
+        ensure!(
+            scheme == "https"
+                || (scheme == "http"
+                    && (host == "localhost"
+                        || host
+                            .parse::<std::net::IpAddr>()
+                            .is_ok_and(|a| a.is_loopback()))),
+            "S3 endpoint requires HTTPS (HTTP is allowed only for loopback)"
+        );
+        ensure!(
+            !config.bucket.is_empty()
+                && config
+                    .bucket
+                    .bytes()
+                    .all(|b| b.is_ascii_alphanumeric() || b".-_".contains(&b)),
+            "invalid S3 bucket"
+        );
+        ensure!(
+            !config.region.is_empty()
+                && !config.access_key.is_empty()
+                && !config.secret_key.is_empty(),
+            "S3 credentials and region are required"
+        );
+        let tls = if scheme == "https" {
+            let roots = crate::federation::Federation::native_roots()?;
+            Some(Arc::new(
+                ClientConfig::builder()
+                    .with_root_certificates(roots)
+                    .with_no_client_auth(),
+            ))
+        } else {
+            None
+        };
+        Ok(Self {
+            config,
+            endpoint,
+            tls,
+            uploads: Arc::new(tokio::sync::Semaphore::new(8)),
+        })
+    }
+    pub fn identity(&self) -> serde_json::Value {
+        serde_json::json!({"backend":"s3", "endpoint":self.config.endpoint.trim_end_matches('/'), "bucket":self.config.bucket, "prefix":self.config.key_prefix.trim_matches('/')})
+    }
+    fn path(&self, id: &str) -> String {
+        let key = [self.config.key_prefix.trim_matches('/'), id]
+            .into_iter()
+            .filter(|s| !s.is_empty())
+            .collect::<Vec<_>>()
+            .join("/");
+        format!(
+            "{}/{}/{}",
+            self.endpoint.path().trim_end_matches('/'),
+            encode(&self.config.bucket),
+            key.split('/').map(encode).collect::<Vec<_>>().join("/")
+        )
+    }
+    async fn request(
+        &self,
+        method: &str,
+        id: &str,
+        query: &[(&str, &str)],
+        body: Bytes,
+        extra: &[(&str, &str)],
+    ) -> Result<Response<hyper::body::Incoming>> {
+        // A deadline includes DNS, TLS, request transmission and response headers.
+        tokio::time::timeout(
+            Duration::from_secs(120),
+            self.request_inner(method, self.path(id), query, body, extra),
+        )
+        .await
+        .map_err(|_| anyhow::anyhow!("S3 request timed out"))?
+    }
+    async fn request_inner(
+        &self,
+        method: &str,
+        path: String,
+        query: &[(&str, &str)],
+        body: Bytes,
+        extra: &[(&str, &str)],
+    ) -> Result<Response<hyper::body::Incoming>> {
+        let host = self.endpoint.host().unwrap().trim_matches(['[', ']']);
+        let authority = self.endpoint.authority().unwrap().as_str();
+        let port = self
+            .endpoint
+            .port_u16()
+            .unwrap_or(if self.tls.is_some() { 443 } else { 80 });
+        let mut query: Vec<_> = query.iter().map(|(k, v)| (encode(k), encode(v))).collect();
+        query.sort();
+        let query = query
+            .iter()
+            .map(|(k, v)| format!("{k}={v}"))
+            .collect::<Vec<_>>()
+            .join("&");
+        let uri = if query.is_empty() {
+            path.clone()
+        } else {
+            format!("{path}?{query}")
+        };
+        let date = chrono::Utc::now();
+        let stamp = date.format("%Y%m%dT%H%M%SZ").to_string();
+        let day = date.format("%Y%m%d").to_string();
+        let hash = hex::encode(Sha256::digest(&body));
+        let mut headers = BTreeMap::from([
+            ("host".to_string(), authority.to_string()),
+            ("x-amz-content-sha256".into(), hash.clone()),
+            ("x-amz-date".into(), stamp.clone()),
+        ]);
+        if let Some(token) = &self.config.session_token {
+            headers.insert("x-amz-security-token".into(), token.clone());
+        }
+        let signed = headers.keys().cloned().collect::<Vec<_>>().join(";");
+        let canonical_headers = headers
+            .iter()
+            .map(|(k, v)| {
+                format!(
+                    "{k}:{}\n",
+                    v.split_whitespace().collect::<Vec<_>>().join(" ")
+                )
+            })
+            .collect::<String>();
+        let canonical = format!("{method}\n{path}\n{query}\n{canonical_headers}\n{signed}\n{hash}");
+        let scope = format!("{day}/{}/s3/aws4_request", self.config.region);
+        let to_sign = format!(
+            "AWS4-HMAC-SHA256\n{stamp}\n{scope}\n{}",
+            hex::encode(Sha256::digest(canonical.as_bytes()))
+        );
+        let key = mac(format!("AWS4{}", self.config.secret_key).as_bytes(), &day);
+        let key = mac(&key, &self.config.region);
+        let key = mac(&key, "s3");
+        let key = mac(&key, "aws4_request");
+        let auth = format!(
+            "AWS4-HMAC-SHA256 Credential={}/{scope}, SignedHeaders={signed}, Signature={}",
+            self.config.access_key,
+            hex::encode(mac(&key, &to_sign))
+        );
+        let mut req = Request::builder()
+            .method(method)
+            .uri(uri)
+            .header("authorization", auth)
+            .header("content-length", body.len());
+        for (k, v) in &headers {
+            req = req.header(k, v);
+        }
+        for (k, v) in extra {
+            req = req.header(*k, *v);
+        }
+        let req = req.body(Full::new(body))?;
+        let tcp = TcpStream::connect((host, port)).await?;
+        let stream: Box<dyn Transport> = if let Some(tls) = &self.tls {
+            Box::new(
+                TlsConnector::from(tls.clone())
+                    .connect(ServerName::try_from(host.to_string())?, tcp)
+                    .await?,
+            )
+        } else {
+            Box::new(tcp)
+        };
+        let (mut sender, conn) =
+            hyper::client::conn::http1::handshake(hyper_util::rt::TokioIo::new(stream)).await?;
+        tokio::spawn(async move {
+            let _ = conn.await;
+        });
+        Ok(sender.send_request(req).await?)
+    }
+    async fn xml(response: Response<hyper::body::Incoming>) -> Result<String> {
+        ensure!(response.status().is_success(), "S3 operation failed");
+        Self::xml_unchecked(response).await
+    }
+    async fn xml_unchecked(response: Response<hyper::body::Incoming>) -> Result<String> {
+        let bytes = tokio::time::timeout(
+            Duration::from_secs(120),
+            Limited::new(response.into_body(), 1024 * 1024).collect(),
+        )
+        .await
+        .map_err(|_| anyhow::anyhow!("S3 response timed out"))?
+        .map_err(|_| anyhow::anyhow!("invalid or oversized S3 response"))?
+        .to_bytes();
+        Ok(String::from_utf8(bytes.to_vec())?)
+    }
+    pub async fn begin(&self, id: &str, content_type: &str) -> Result<Multipart> {
+        // Shield initiation: cancellation must still consume UploadId and abort it.
+        let store = self.clone();
+        let id = id.to_owned();
+        let content_type = content_type.to_owned();
+        tokio::spawn(async move { store.begin_inner(&id, &content_type).await }).await?
+    }
+    async fn begin_inner(&self, id: &str, content_type: &str) -> Result<Multipart> {
+        let permit = self
+            .uploads
+            .clone()
+            .try_acquire_owned()
+            .map_err(|_| anyhow::anyhow!("S3 uploads busy"))?;
+        let r = self
+            .request(
+                "POST",
+                id,
+                &[("uploads", "")],
+                Bytes::new(),
+                &[("content-type", content_type)],
+            )
+            .await?;
+        let xml = Self::xml(r).await?;
+        let doc = roxmltree::Document::parse(&xml)?;
+        let upload_id = doc
+            .descendants()
+            .find(|n| n.has_tag_name("UploadId"))
+            .and_then(|n| n.text())
+            .filter(|s| !s.is_empty())
+            .ok_or_else(|| anyhow::anyhow!("S3 upload identifier missing"))?
+            .to_string();
+        Ok(Multipart {
+            store: self.clone(),
+            id: id.into(),
+            upload_id,
+            etags: Vec::new(),
+            finished: false,
+            after_abort: None,
+            _permit: permit,
+        })
+    }
+    pub async fn read(&self, id: &str, head: bool, range: Option<&str>) -> Result<Response<Body>> {
+        let extra: Vec<_> = range.map(|r| vec![("range", r)]).unwrap_or_default();
+        let r = self
+            .request(
+                if head { "HEAD" } else { "GET" },
+                id,
+                &[],
+                Bytes::new(),
+                &extra,
+            )
+            .await?;
+        let (parts, body) = r.into_parts();
+        Ok(Response::from_parts(
+            parts,
+            if head { Body::empty() } else { Body::new(body) },
+        ))
+    }
+    /// Abort only multiparts for an exclusively owned, durable slot key. The
+    /// caller must exclude concurrent initiation for this key throughout recovery.
+    /// Listing a prefix alone does not establish ownership: compare the full key.
+    pub async fn abort_pending(&self, id: &str) -> Result<()> {
+        let key = [self.config.key_prefix.trim_matches('/'), id]
+            .into_iter()
+            .filter(|v| !v.is_empty())
+            .collect::<Vec<_>>()
+            .join("/");
+        let bucket_path = format!(
+            "{}/{}",
+            self.endpoint.path().trim_end_matches('/'),
+            encode(&self.config.bucket)
+        );
+        let mut markers: Option<(String, String)> = None;
+        let mut seen = std::collections::BTreeSet::new();
+        loop {
+            let mut query = vec![
+                ("uploads", ""),
+                ("prefix", key.as_str()),
+                ("max-uploads", "1000"),
+                ("encoding-type", "url"),
+            ];
+            if let Some((key_marker, upload_marker)) = &markers {
+                query.push(("key-marker", key_marker));
+                query.push(("upload-id-marker", upload_marker));
+            }
+            let response = tokio::time::timeout(
+                Duration::from_secs(120),
+                self.request_inner("GET", bucket_path.clone(), &query, Bytes::new(), &[]),
+            )
+            .await
+            .map_err(|_| anyhow::anyhow!("S3 listing timed out"))??;
+            let xml = Self::xml(response).await?;
+            let doc = roxmltree::Document::parse(&xml)?;
+            let root = doc.root_element();
+            ensure!(
+                root.has_tag_name("ListMultipartUploadsResult"),
+                "invalid S3 multipart listing"
+            );
+            let field = |node: roxmltree::Node<'_, '_>, name: &str| {
+                node.children()
+                    .find(|n| n.has_tag_name(name))
+                    .and_then(|n| n.text())
+                    .map(str::to_owned)
+            };
+            for upload in root.children().filter(|n| n.has_tag_name("Upload")) {
+                let listed_key = field(upload, "Key")
+                    .ok_or_else(|| anyhow::anyhow!("S3 listing key missing"))?;
+                let upload_id = field(upload, "UploadId")
+                    .filter(|s| !s.is_empty())
+                    .ok_or_else(|| anyhow::anyhow!("S3 listing upload identifier missing"))?;
+                if decode_key(&listed_key)? == key {
+                    self.abort(id, &upload_id).await?;
+                }
+            }
+            match field(root, "IsTruncated").as_deref() {
+                Some("false") => break,
+                Some("true") => {
+                    let next = (
+                        decode_key(
+                            &field(root, "NextKeyMarker")
+                                .ok_or_else(|| anyhow::anyhow!("S3 listing marker missing"))?,
+                        )?,
+                        field(root, "NextUploadIdMarker")
+                            .ok_or_else(|| anyhow::anyhow!("S3 listing upload marker missing"))?,
+                    );
+                    ensure!(
+                        !next.0.is_empty()
+                            && !next.1.is_empty()
+                            && markers.as_ref().is_none_or(|previous| &next > previous)
+                            && seen.insert(next.clone()),
+                        "S3 listing did not advance"
+                    );
+                    markers = Some(next);
+                }
+                _ => anyhow::bail!("invalid S3 listing truncation flag"),
+            }
+        }
+        Ok(())
+    }
+    /// HEAD with an explicit distinction between absence and service errors.
+    pub async fn stat(&self, id: &str) -> Result<Option<u64>> {
+        let reply = self.read(id, true, None).await?;
+        if reply.status() == 404 {
+            return Ok(None);
+        }
+        ensure!(reply.status() == 200, "S3 stat failed");
+        let size = reply
+            .headers()
+            .get("content-length")
+            .and_then(|v| v.to_str().ok())
+            .and_then(|v| v.parse().ok())
+            .ok_or_else(|| anyhow::anyhow!("S3 object size missing"))?;
+        Ok(Some(size))
+    }
+    /// Bounded-memory streaming upload. Callers journal ownership before calling
+    /// this method and retain it until commit, allowing abort_pending after crash.
+    pub async fn write_stream(
+        &self,
+        id: &str,
+        content_type: &str,
+        mut body: Body,
+        expected_size: u64,
+    ) -> Result<()> {
+        let mut upload = self.begin(id, content_type).await?;
+        let mut count = 0u64;
+        let mut part = Vec::with_capacity(PART_SIZE);
+        while let Some(frame) = tokio::time::timeout(Duration::from_secs(120), body.frame())
+            .await
+            .map_err(|_| anyhow::anyhow!("S3 source timed out"))?
+        {
+            let frame = frame.map_err(|_| anyhow::anyhow!("S3 source read failed"))?;
+            if let Ok(bytes) = frame.into_data() {
+                count = count
+                    .checked_add(bytes.len() as u64)
+                    .ok_or_else(|| anyhow::anyhow!("S3 source too large"))?;
+                ensure!(count <= expected_size, "S3 source too large");
+                let mut bytes = bytes.as_ref();
+                while !bytes.is_empty() {
+                    let take = (PART_SIZE - part.len()).min(bytes.len());
+                    part.extend_from_slice(&bytes[..take]);
+                    bytes = &bytes[take..];
+                    if part.len() == PART_SIZE {
+                        upload
+                            .part(std::mem::replace(&mut part, Vec::with_capacity(PART_SIZE)))
+                            .await?;
+                    }
+                }
+            }
+        }
+        ensure!(count == expected_size, "S3 source size mismatch");
+        if !part.is_empty() || count == 0 {
+            upload.part(part).await?;
+        }
+        upload.complete().await
+    }
+    pub async fn copy_from(
+        &self,
+        source: &ObjectStore,
+        id: &str,
+        content_type: &str,
+        expected_size: u64,
+    ) -> Result<()> {
+        let reply = source.read(id, false, None).await?;
+        ensure!(reply.status() == 200, "S3 copy source unavailable");
+        self.write_stream(id, content_type, reply.into_body(), expected_size)
+            .await
+    }
+    pub async fn abort(&self, id: &str, upload_id: &str) -> Result<()> {
+        let r = self
+            .request("DELETE", id, &[("uploadId", upload_id)], Bytes::new(), &[])
+            .await?;
+        ensure!(
+            r.status().is_success() || r.status() == 404,
+            "S3 multipart abort failed"
+        );
+        Ok(())
+    }
+    pub async fn delete(&self, id: &str) -> Result<()> {
+        let r = self.request("DELETE", id, &[], Bytes::new(), &[]).await?;
+        ensure!(
+            r.status().is_success() || r.status() == 404,
+            "S3 deletion failed"
+        );
+        Ok(())
+    }
+}
+/// A well-formed terminal rejection from CompleteMultipartUpload. Transport,
+/// timeout and malformed response errors deliberately do not carry this marker.
+#[derive(Debug)]
+pub struct CompletionRejected;
+impl std::fmt::Display for CompletionRejected {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("S3 completion rejected")
+    }
+}
+impl std::error::Error for CompletionRejected {}
+pub const PART_SIZE: usize = 8 * 1024 * 1024;
+pub struct Multipart {
+    store: ObjectStore,
+    id: String,
+    upload_id: String,
+    etags: Vec<String>,
+    finished: bool,
+    after_abort: Option<Box<dyn FnOnce() + Send>>,
+    _permit: tokio::sync::OwnedSemaphorePermit,
+}
+impl Multipart {
+    pub fn upload_id(&self) -> &str {
+        &self.upload_id
+    }
+    pub fn after_abort(&mut self, callback: impl FnOnce() + Send + 'static) {
+        self.after_abort = Some(Box::new(callback));
+    }
+    pub async fn part(&mut self, data: Vec<u8>) -> Result<()> {
+        ensure!(self.etags.len() < 10000, "S3 part limit exceeded");
+        let number = (self.etags.len() + 1).to_string();
+        let r = self
+            .store
+            .request(
+                "PUT",
+                &self.id,
+                &[("partNumber", &number), ("uploadId", &self.upload_id)],
+                Bytes::from(data),
+                &[],
+            )
+            .await?;
+        ensure!(r.status().is_success(), "S3 part failed");
+        let etag = r
+            .headers()
+            .get("etag")
+            .and_then(|v| v.to_str().ok())
+            .ok_or_else(|| anyhow::anyhow!("S3 part ETag missing"))?
+            .to_string();
+        self.etags.push(etag);
+        Ok(())
+    }
+    pub async fn complete(mut self) -> Result<()> {
+        let mut xml = "<CompleteMultipartUpload>".to_string();
+        for (i, etag) in self.etags.iter().enumerate() {
+            xml.push_str(&format!(
+                "<Part><PartNumber>{}</PartNumber><ETag>{}</ETag></Part>",
+                i + 1,
+                crate::wire::escape(etag)
+            ));
+        }
+        xml.push_str("</CompleteMultipartUpload>");
+        let r = self
+            .store
+            .request(
+                "POST",
+                &self.id,
+                &[("uploadId", &self.upload_id)],
+                Bytes::from(xml),
+                &[("content-type", "application/xml"), ("if-none-match", "*")],
+            )
+            .await?;
+        let successful = r.status().is_success();
+        let xml = ObjectStore::xml_unchecked(r).await?;
+        let doc = roxmltree::Document::parse(&xml)?;
+        if doc.root_element().has_tag_name("Error") {
+            return Err(CompletionRejected.into());
+        }
+        ensure!(successful, "S3 completion failed");
+        ensure!(
+            doc.root_element()
+                .has_tag_name("CompleteMultipartUploadResult"),
+            "S3 completion failed"
+        );
+        self.finished = true;
+        Ok(())
+    }
+}
+impl Drop for Multipart {
+    fn drop(&mut self) {
+        if !self.finished {
+            let store = self.store.clone();
+            let id = self.id.clone();
+            let upload_id = self.upload_id.clone();
+            let after_abort = self.after_abort.take();
+            if let Ok(handle) = tokio::runtime::Handle::try_current() {
+                handle.spawn(async move {
+                    if store.abort(&id, &upload_id).await.is_ok()
+                        && let Some(callback) = after_abort
+                    {
+                        callback();
+                    }
+                });
+            }
+        }
+    }
+}

src/server.rs

@@ -49,6 +49,136 @@ pub struct Runtime {
     federation: Option<crate::federation::Federation>,
     stopping: Arc<std::sync::atomic::AtomicBool>,
 }
+/// Keep upload services separate: their HTTP hosts, mounts and credentials belong
+/// to the component, while all listeners share one dispatcher.
+#[derive(Clone)]
+struct UploadRoute {
+    name: String,
+    config: crate::http_config::HttpConfig,
+    service: crate::http_service::UploadService,
+}
+pub(crate) fn upload_backend(
+    config: &Config,
+    name: &str,
+) -> Result<crate::object_store::UploadBackendConfig> {
+    use crate::object_store::{S3Config, UploadBackendConfig};
+    let backend = config
+        .option(name, "http_file_share_backend")
+        .map(|value| {
+            value
+                .as_str()
+                .context("http_file_share_backend must be a string")
+        })
+        .transpose()?
+        .unwrap_or("local");
+    match backend {
+        "local" => Ok(UploadBackendConfig::Local),
+        "s3" => {
+            let options = config
+                .option(name, "http_file_share_s3")
+                .and_then(|value| value.as_object())
+                .context("http_file_share_s3 must be a table for S3 uploads")?;
+            let field = |key: &str| -> Result<Option<String>> {
+                if let Some(value) = options.get(&format!("{key}_env")) {
+                    let variable = value.as_str().with_context(|| {
+                        format!("http_file_share_s3.{key}_env must be a string")
+                    })?;
+                    return Ok(Some(std::env::var(variable).with_context(|| {
+                        format!("environment variable for http_file_share_s3.{key} is unavailable")
+                    })?));
+                }
+                options
+                    .get(key)
+                    .map(|value| {
+                        value
+                            .as_str()
+                            .map(str::to_owned)
+                            .with_context(|| format!("http_file_share_s3.{key} must be a string"))
+                    })
+                    .transpose()
+            };
+            let required = |key: &str| {
+                field(key)?
+                    .filter(|value| !value.is_empty())
+                    .with_context(|| format!("http_file_share_s3.{key} is required"))
+            };
+            Ok(UploadBackendConfig::S3(S3Config {
+                endpoint: required("endpoint")?,
+                bucket: required("bucket")?,
+                region: required("region")?,
+                access_key: required("access_key")?,
+                secret_key: required("secret_key")?,
+                session_token: field("session_token")?,
+                key_prefix: field("key_prefix")?.unwrap_or_default(),
+            }))
+        }
+        other => bail!("unsupported http_file_share_backend: {other}"),
+    }
+}
+fn upload_mount_matches(mount: &str, path: &str) -> bool {
+    mount == "/"
+        || path == mount
+        || path
+            .strip_prefix(mount)
+            .is_some_and(|rest| rest.starts_with('/'))
+}
+fn upload_routes(
+    config: &Config,
+    uploads: &BTreeMap<String, crate::http_service::UploadService>,
+) -> Result<Arc<Vec<UploadRoute>>> {
+    let mut routes: Vec<UploadRoute> = Vec::new();
+    for (name, service) in uploads {
+        let http = crate::http_config::HttpConfig::for_host(config, name)?;
+        for existing in &routes {
+            ensure!(
+                existing.config.host != http.host
+                    || !(upload_mount_matches(
+                        &existing.config.file_share_path,
+                        &http.file_share_path
+                    ) || upload_mount_matches(
+                        &http.file_share_path,
+                        &existing.config.file_share_path
+                    )),
+                "ambiguous HTTP upload components '{}' and '{}': HTTP host '{}' has overlapping mounts '{}' and '{}'",
+                existing.name,
+                name,
+                http.host,
+                existing.config.file_share_path,
+                http.file_share_path
+            );
+        }
+        routes.push(UploadRoute {
+            name: name.clone(),
+            config: http,
+            service: service.clone(),
+        });
+    }
+    Ok(Arc::new(routes))
+}
+async fn dispatch_upload(
+    routes: Arc<Vec<UploadRoute>>,
+    request: axum::extract::Request,
+) -> axum::response::Response {
+    use axum::response::IntoResponse;
+    let host = request
+        .headers()
+        .get(axum::http::header::HOST)
+        .and_then(|value| value.to_str().ok());
+    if let Some(route) = routes.iter().find(|route| {
+        host.is_some_and(|host| route.config.accepts_host(host))
+            && upload_mount_matches(&route.config.file_share_path, request.uri().path())
+    }) {
+        let mut response = route
+            .service
+            .handle_at(request, &route.config.file_share_path)
+            .await;
+        response
+            .extensions_mut()
+            .insert(crate::http_transport::HttpRoute::Upload);
+        return response;
+    }
+    axum::http::StatusCode::NOT_FOUND.into_response()
+}
 struct ConnectionLog {
     logger: Logger,
     id: String,
@@ -557,6 +687,7 @@ pub async fn serve(path: &Path) -> Result<()> {
 
     let (data, _, hash) = account_config(path)?;
     let data_guard = Arc::new(crate::data_lock::DataDirectoryGuard::acquire(&data)?);
+    crate::upload_migration::ensure_no_pending(&data)?;
     let ssl = c.option(&domain, "ssl").context("ssl required")?;
     let cert = ssl
         .get("certificate")
@@ -630,8 +761,8 @@ pub async fn serve(path: &Path) -> Result<()> {
                     .and_then(|v| v.as_str())
                     .map(|s| s.trim_end_matches('/').to_owned())
                     .unwrap_or(base);
-                let service =
-                    crate::http_service::UploadService::new(crate::http_service::UploadConfig {
+                let service = crate::http_service::UploadService::new_with_backend(
+                    crate::http_service::UploadConfig {
                         root: data.clone(),
                         host: name.clone(),
                         base_url: base,
@@ -667,17 +798,30 @@ pub async fn serve(path: &Path) -> Result<()> {
                             .option(name, "http_file_share_secret")
                             .and_then(|v| v.as_str())
                             .map(str::to_owned),
-                    })?;
+                    },
+                    upload_backend(&c, name)?,
+                )?;
+                service.recover_backend().await?;
+                for slot in service.quarantined_slots() {
+                    logger.event(
+                        Level::Warn,
+                        "upload",
+                        None,
+                        "completion-quarantined",
+                        &[
+                            ("component", name),
+                            ("slot", &slot),
+                            ("reason", "remote-completion-outcome-uncertain"),
+                        ],
+                    );
+                }
                 uploads.insert(name.clone(), service);
             }
             Some("pubsub") | None => {}
             Some(other) => bail!("component not integrated: {other}"),
         }
     }
-    ensure!(
-        uploads.len() <= 1,
-        "multiple HTTP upload components not yet integrated"
-    );
+    let upload_routes = upload_routes(&c, &uploads)?;
     let runtime = Runtime {
         _data_guard: data_guard,
         domain: domain.clone(),
@@ -892,31 +1036,11 @@ pub async fn serve(path: &Path) -> Result<()> {
                 })
             };
             let http_config = crate::http_config::HttpConfig::for_host(&c, &domain)?;
-            let fallback = if let Some((name, service)) = runtime.uploads.iter().next() {
-                let config = crate::http_config::HttpConfig::for_host(&c, name)?;
-                service
-                    .router_at(&config.file_share_path)?
-                    .layer(axum::middleware::from_fn(
-                        move |req: axum::extract::Request, next: axum::middleware::Next| {
-                            let accepted = req
-                                .headers()
-                                .get(axum::http::header::HOST)
-                                .and_then(|v| v.to_str().ok())
-                                .is_some_and(|h| config.accepts_host(h));
-                            async move {
-                                if accepted {
-                                    next.run(req).await
-                                } else {
-                                    axum::response::IntoResponse::into_response(
-                                        axum::http::StatusCode::NOT_FOUND,
-                                    )
-                                }
-                            }
-                        },
-                    ))
-            } else {
-                axum::Router::new()
-            };
+            let routes = upload_routes.clone();
+            let fallback = axum::Router::new().fallback(move |request: axum::extract::Request| {
+                let routes = routes.clone();
+                async move { dispatch_upload(routes, request).await }
+            });
             let app = crate::http_transport::configured_router_with_fallback(
                 handler,
                 http_config,

src/upload_migration.rs

@@ -0,0 +1,565 @@
+//! Offline, source-preserving upload migration. The journal is the startup barrier.
+use crate::{
+    config::Config,
+    object_store::{ObjectStore, PART_SIZE, UploadBackendConfig},
+    storage,
+};
+use anyhow::{Context, Result, ensure};
+use http_body_util::BodyExt;
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use sha2::{Digest, Sha256};
+use std::{
+    collections::BTreeMap,
+    io::{Read, Write},
+    path::{Path, PathBuf},
+};
+
+const JOURNAL: &str = "rust_upload_migration.json";
+#[derive(Serialize, Deserialize)]
+struct Journal {
+    version: u32,
+    component: String,
+    source: Value,
+    destination: Value,
+    metadata: String,
+    committed: bool,
+    objects: BTreeMap<String, Progress>,
+}
+#[derive(Default, Serialize, Deserialize)]
+struct Progress {
+    checksum: Option<String>,
+    verified: bool,
+    upload_id: Option<String>,
+}
+struct Backend(Option<ObjectStore>);
+impl Backend {
+    fn new(c: UploadBackendConfig) -> Result<Self> {
+        Ok(Self(match c {
+            UploadBackendConfig::Local => None,
+            UploadBackendConfig::S3(c) => Some(ObjectStore::new(c)?),
+        }))
+    }
+    fn identity(&self) -> Value {
+        self.0
+            .as_ref()
+            .map(ObjectStore::identity)
+            .unwrap_or_else(|| serde_json::json!({"backend":"local"}))
+    }
+}
+fn save(path: &Path, journal: &Journal) -> Result<()> {
+    storage::atomic_write(path, &serde_json::to_vec_pretty(journal)?)
+}
+fn digest(bytes: &[u8]) -> String {
+    hex::encode(Sha256::digest(bytes))
+}
+fn local_path(dir: &Path, id: &str) -> PathBuf {
+    dir.join("http_file_share")
+        .join(format!("{}.bin", storage::encode_name(id)))
+}
+
+/// Check all component directories, including components removed from the new config.
+pub fn ensure_no_pending(root: &Path) -> Result<()> {
+    if !root.exists() {
+        return Ok(());
+    }
+    for entry in std::fs::read_dir(root)? {
+        let path = entry?.path().join(JOURNAL);
+        if path.exists() {
+            let journal: Journal = serde_json::from_slice(&std::fs::read(&path)?)
+                .context("invalid upload migration journal; offline recovery required")?;
+            ensure!(
+                journal.version == 1 && journal.committed,
+                "pending upload migration; resume upload-migrate before starting the server"
+            );
+        }
+    }
+    Ok(())
+}
+fn validate_component(config: &Config, component: &str) -> Result<()> {
+    ensure!(
+        config
+            .hosts
+            .get(component)
+            .and_then(|h| h.component.as_deref())
+            == Some("http_file_share"),
+        "component must be an HTTP file share in both configurations"
+    );
+    Ok(())
+}
+fn stable_settings(old: &Config, new: &Config, component: &str) -> Result<()> {
+    // Keep URL construction and JWT verification exactly as configured, including inherited options.
+    for key in [
+        "http_file_share_secret",
+        "http_file_share_base_url",
+        "http_host",
+        "http_default_host",
+        "http_external_url",
+        "http_paths",
+        "http_ports",
+        "https_ports",
+    ] {
+        ensure!(
+            old.option(component, key) == new.option(component, key),
+            "migration must preserve {key}; change public routing separately after migration"
+        );
+    }
+    Ok(())
+}
+
+/// Plan by default. Execution never deletes source objects or rewrites native metadata.
+pub async fn migrate(
+    old_path: &Path,
+    new_path: &Path,
+    component: &str,
+    execute: bool,
+) -> Result<()> {
+    let (old, old_domain) = crate::server::settings_config(old_path)?;
+    let (new, new_domain) = crate::server::settings_config(new_path)?;
+    ensure!(
+        old_domain == new_domain,
+        "virtual host must remain unchanged"
+    );
+    for key in ["http_ports", "https_ports", "data_path"] {
+        ensure!(
+            old.option(&old_domain, key) == new.option(&new_domain, key),
+            "migration must preserve runtime {key}"
+        );
+    }
+    stable_settings(&old, &new, component)?;
+    validate_component(&old, component)?;
+    validate_component(&new, component)?;
+    let root = PathBuf::from(
+        old.option(&old_domain, "data_path")
+            .and_then(Value::as_str)
+            .context("data_path required")?,
+    );
+    let new_root = PathBuf::from(
+        new.option(&new_domain, "data_path")
+            .and_then(Value::as_str)
+            .context("data_path required")?,
+    );
+    ensure!(
+        root.is_dir() && new_root.is_dir(),
+        "existing data directories required"
+    );
+    ensure!(
+        root.canonicalize()? == new_root.canonicalize()?,
+        "migration requires the same data directory; native metadata must stay in place"
+    );
+    let _guard = crate::data_lock::DataDirectoryGuard::acquire(&root)?;
+    let source = Backend::new(crate::server::upload_backend(&old, component)?)?;
+    let target = Backend::new(crate::server::upload_backend(&new, component)?)?;
+    ensure!(
+        source.identity() != target.identity(),
+        "source and destination backend identities are identical"
+    );
+    let dir = root.join(storage::encode_name(component));
+    let cache = match std::fs::read(dir.join("rust_uploads.json")) {
+        Ok(v) => v,
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => b"{}".to_vec(),
+        Err(e) => return Err(e.into()),
+    };
+    let records: BTreeMap<String, Value> = serde_json::from_slice(&cache)?;
+    ensure!(
+        !storage::path(&root, component, None, "uploads")
+            .with_extension("rappend")
+            .exists(),
+        "upload archive has pending append recovery; recover source server before migration"
+    );
+    let entries = crate::archive::load_store(&root, component, None, "uploads")?;
+    let mut seen = std::collections::BTreeSet::new();
+    let mut objects = BTreeMap::new();
+    let mut fingerprint = Sha256::new();
+    // Lua table traversal can reorder reconstructed XML attributes between
+    // processes. Bind exact authoritative archive bytes, not reconstructed XML.
+    fingerprint.update((cache.len() as u64).to_le_bytes());
+    fingerprint.update(&cache);
+    let archive_path = storage::path(&root, component, None, "uploads").with_extension("list");
+    let archive_bytes = match std::fs::read(archive_path) {
+        Ok(bytes) => bytes,
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Vec::new(),
+        Err(e) => return Err(e.into()),
+    };
+    fingerprint.update((archive_bytes.len() as u64).to_le_bytes());
+    fingerprint.update(&archive_bytes);
+    for e in entries {
+        ensure!(
+            seen.insert(e.id.clone()),
+            "duplicate authoritative upload id"
+        );
+        let doc = roxmltree::Document::parse(&e.xml)?;
+        let node = doc.root_element();
+        let size: u64 = node
+            .attribute("size")
+            .context("upload size missing")?
+            .parse()?;
+        let filename = node
+            .attribute("filename")
+            .context("upload filename missing")?;
+        let r = records.get(&e.id);
+        if let Some(r) = r {
+            ensure!(
+                r.get("owner").and_then(Value::as_str) == Some(&e.peer)
+                    && r.get("filename").and_then(Value::as_str) == Some(filename)
+                    && r.get("size").and_then(Value::as_u64) == Some(size),
+                "slot metadata/cache mismatch; recover source server first"
+            );
+            ensure!(
+                r.get("upload_id").is_none_or(Value::is_null)
+                    && r.get("multipart_intent").and_then(Value::as_bool) != Some(true)
+                    && r.get("completion_attempted").and_then(Value::as_bool) != Some(true),
+                "slot has unfinished multipart state; recover source server first"
+            );
+        }
+        // Native Prosody files need no Rust completion sidecar. A present final
+        // object of exact declared size is completed; absent reservations stay pending.
+        let complete = r.and_then(|r| r.get("complete")).and_then(Value::as_bool) == Some(true);
+        let present = if complete {
+            true
+        } else if let Some(store) = &source.0 {
+            match store.stat(&e.id).await? {
+                Some(n) => {
+                    ensure!(n == size, "source object size mismatch");
+                    true
+                }
+                None => false,
+            }
+        } else {
+            local_path(&dir, &e.id).exists()
+        };
+        if complete || present {
+            ensure!(
+                objects
+                    .insert(
+                        e.id.clone(),
+                        (
+                            size,
+                            node.attribute("content-type")
+                                .unwrap_or("application/octet-stream")
+                                .to_owned()
+                        )
+                    )
+                    .is_none(),
+                "duplicate upload id"
+            );
+        }
+    }
+    let metadata = hex::encode(fingerprint.finalize());
+    let journal_path = dir.join(JOURNAL);
+    let mut resumed = false;
+    let fresh = || Journal {
+        version: 1,
+        component: component.into(),
+        source: source.identity(),
+        destination: target.identity(),
+        metadata: metadata.clone(),
+        committed: false,
+        objects: objects
+            .keys()
+            .map(|id| (id.clone(), Progress::default()))
+            .collect(),
+    };
+    let mut journal = if journal_path.exists() {
+        let j: Journal = serde_json::from_slice(&std::fs::read(&journal_path)?)?;
+        if j.version == 1
+            && j.committed
+            && (j.source != source.identity()
+                || j.destination != target.identity()
+                || j.metadata != metadata)
+        {
+            fresh()
+        } else {
+            ensure!(
+                j.version == 1
+                    && j.component == component
+                    && j.source == source.identity()
+                    && j.destination == target.identity()
+                    && j.metadata == metadata
+                    && j.objects.keys().eq(objects.keys()),
+                "migration journal does not match configurations or metadata"
+            );
+            resumed = true;
+            j
+        }
+    } else {
+        fresh()
+    };
+    let identity_path = dir.join("rust_upload_backend.json");
+    let current: Value = if identity_path.exists() {
+        serde_json::from_slice(&std::fs::read(&identity_path)?)?
+    } else {
+        serde_json::json!({"backend":"local"})
+    };
+    ensure!(
+        current == source.identity() || (resumed && current == target.identity()),
+        "recorded backend is not the migration source"
+    );
+    println!(
+        "{} completed objects; source objects and all native metadata will be preserved",
+        objects.len()
+    );
+    if !execute {
+        println!(
+            "Dry run: no objects, journal or backend identity changed. Use --execute to copy and verify."
+        );
+        return Ok(());
+    }
+    if journal.committed {
+        ensure!(
+            current == target.identity(),
+            "committed migration identity mismatch"
+        );
+        println!("Migration already committed.");
+        return Ok(());
+    }
+    save(&journal_path, &journal)?;
+    for (id, (size, content_type)) in &objects {
+        // Deterministic journal-owned spools can be removed safely after a crash.
+        let temp = dir.join(format!(".migration-{}.tmp", digest(id.as_bytes())));
+        if temp.exists() {
+            std::fs::remove_file(&temp)?;
+        }
+        let result = async {
+            let checksum = spool(&source, &dir, id, *size, &temp)
+                .await?
+                .context("source object missing")?;
+            if let Some(previous) = &journal.objects[id].checksum {
+                ensure!(
+                    *previous == checksum,
+                    "source checksum changed during migration"
+                );
+            }
+            journal.objects.get_mut(id).unwrap().checksum = Some(checksum.clone());
+            save(&journal_path, &journal)?;
+            // The durable per-object checksum precedes initiation. Sweep this
+            // exact key, including an initiation whose response was lost.
+            if let Some(store) = &target.0 {
+                store.abort_pending(id).await?;
+            }
+            if let Some(upload_id) = journal.objects[id].upload_id.as_ref() {
+                target
+                    .0
+                    .as_ref()
+                    .context("invalid local multipart journal")?
+                    .abort(id, upload_id)
+                    .await?;
+                journal.objects.get_mut(id).unwrap().upload_id = None;
+                save(&journal_path, &journal)?;
+            }
+            let existing = object_checksum(&target, &dir, id, *size).await;
+            let existing = existing?;
+            if let Some(existing) = existing {
+                ensure!(
+                    existing == checksum,
+                    "destination object conflicts; refusing overwrite"
+                );
+            } else {
+                ensure!(
+                    !journal.objects[id].verified,
+                    "previously verified destination object disappeared"
+                );
+                if let Some(store) = &target.0 {
+                    let mut upload = store.begin(id, content_type).await?;
+                    journal.objects.get_mut(id).unwrap().upload_id =
+                        Some(upload.upload_id().into());
+                    save(&journal_path, &journal)?;
+                    let mut file = std::fs::File::open(&temp)?;
+                    loop {
+                        let mut part = vec![0; PART_SIZE];
+                        let mut n = 0;
+                        while n < part.len() {
+                            let read = file.read(&mut part[n..])?;
+                            if read == 0 {
+                                break;
+                            }
+                            n += read;
+                        }
+                        part.truncate(n);
+                        if n == 0 {
+                            if *size == 0 {
+                                upload.part(part).await?;
+                            }
+                            break;
+                        }
+                        upload.part(part).await?;
+                    }
+                    upload.complete().await?;
+                    journal.objects.get_mut(id).unwrap().upload_id = None;
+                    save(&journal_path, &journal)?;
+                } else {
+                    let path = local_path(&dir, id);
+                    std::fs::create_dir_all(path.parent().unwrap())?;
+                    // Hard-link publishes without clobbering an existing object; spool is on the same filesystem.
+                    std::fs::hard_link(&temp, &path)?;
+                    std::fs::File::open(path.parent().unwrap())?.sync_all()?;
+                }
+                let verified = object_checksum(&target, &dir, id, *size).await;
+                ensure!(
+                    verified?.as_deref() == Some(checksum.as_str()),
+                    "destination checksum mismatch"
+                );
+            }
+            journal.objects.get_mut(id).unwrap().verified = true;
+            save(&journal_path, &journal)?;
+            Ok::<_, anyhow::Error>(())
+        }
+        .await;
+        let _ = std::fs::remove_file(&temp);
+        result.with_context(|| {
+            format!(
+                "migration stopped at object {id}; source preserved, resume with the same command"
+            )
+        })?;
+    }
+    ensure!(
+        journal.objects.values().all(|p| p.verified),
+        "unverified migration objects"
+    );
+    storage::atomic_write(&identity_path, &serde_json::to_vec(&target.identity())?)?;
+    journal.committed = true;
+    save(&journal_path, &journal)?;
+    println!("Migration committed: all destination objects checksum verified; source preserved.");
+    Ok(())
+}
+
+/// Stream through a private, bounded spool; never trust ETag as a content checksum.
+async fn spool(
+    backend: &Backend,
+    dir: &Path,
+    id: &str,
+    size: u64,
+    temp: &Path,
+) -> Result<Option<String>> {
+    spool_to(backend, dir, id, size, Some(temp)).await
+}
+async fn object_checksum(
+    backend: &Backend,
+    dir: &Path,
+    id: &str,
+    size: u64,
+) -> Result<Option<String>> {
+    spool_to(backend, dir, id, size, None).await
+}
+async fn spool_to(
+    backend: &Backend,
+    dir: &Path,
+    id: &str,
+    size: u64,
+    temp: Option<&Path>,
+) -> Result<Option<String>> {
+    let mut options = std::fs::OpenOptions::new();
+    options.write(true).create_new(true);
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::OpenOptionsExt;
+        options.mode(0o600);
+    }
+    let mut file = temp.map(|temp| options.open(temp)).transpose()?;
+    let mut hash = Sha256::new();
+    let mut total = 0u64;
+    if let Some(store) = &backend.0 {
+        let response = store.read(id, false, None).await?;
+        if response.status() == 404 {
+            return Ok(None);
+        }
+        ensure!(
+            response.status() == 200,
+            "object download failed or returned partial content"
+        );
+        let mut body = response.into_body();
+        while let Some(frame) =
+            tokio::time::timeout(std::time::Duration::from_secs(120), body.frame())
+                .await
+                .context("migration download timed out")?
+        {
+            let frame = frame?;
+            if let Some(bytes) = frame.data_ref() {
+                total = total
+                    .checked_add(bytes.len() as u64)
+                    .context("size overflow")?;
+                ensure!(total <= size, "object exceeds metadata size");
+                hash.update(bytes);
+                if let Some(file) = &mut file {
+                    file.write_all(bytes)?;
+                }
+            }
+        }
+    } else {
+        let path = local_path(dir, id);
+        let metadata = match std::fs::symlink_metadata(&path) {
+            Ok(m) => m,
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+            Err(e) => return Err(e.into()),
+        };
+        ensure!(
+            metadata.file_type().is_file() && metadata.len() == size,
+            "local object is not a regular file of metadata size"
+        );
+        let mut source = std::fs::File::open(path)?;
+        let mut bytes = vec![0; PART_SIZE];
+        loop {
+            let n = source.read(&mut bytes)?;
+            if n == 0 {
+                break;
+            }
+            total += n as u64;
+            ensure!(total <= size, "object exceeds metadata size");
+            hash.update(&bytes[..n]);
+            if let Some(file) = &mut file {
+                file.write_all(&bytes[..n])?;
+            }
+        }
+    }
+    ensure!(total == size, "object size differs from metadata");
+    if let Some(file) = file {
+        file.sync_all()?;
+    }
+    Ok(Some(hex::encode(hash.finalize())))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[tokio::test]
+    async fn local_spool_checks_size_and_hash_and_preserves_source() {
+        let d = tempfile::tempdir().unwrap();
+        let path = local_path(d.path(), "a/b");
+        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
+        std::fs::write(&path, b"payload").unwrap();
+        let temp = d.path().join("spool");
+        assert_eq!(
+            spool(&Backend(None), d.path(), "a/b", 7, &temp)
+                .await
+                .unwrap(),
+            Some(digest(b"payload"))
+        );
+        assert_eq!(std::fs::read(&path).unwrap(), b"payload");
+        assert!(
+            spool(&Backend(None), d.path(), "a/b", 8, &d.path().join("bad"))
+                .await
+                .is_err()
+        );
+    }
+    #[test]
+    fn pending_and_corrupt_journals_block_startup() {
+        let d = tempfile::tempdir().unwrap();
+        let path = d.path().join("removed-component").join(JOURNAL);
+        let mut j = Journal {
+            version: 1,
+            component: "upload".into(),
+            source: Value::Null,
+            destination: Value::Null,
+            metadata: "a".into(),
+            committed: false,
+            objects: BTreeMap::new(),
+        };
+        save(&path, &j).unwrap();
+        assert!(ensure_no_pending(d.path()).is_err());
+        j.committed = true;
+        save(&path, &j).unwrap();
+        assert!(ensure_no_pending(d.path()).is_ok());
+        std::fs::write(path, b"broken").unwrap();
+        assert!(ensure_no_pending(d.path()).is_err());
+    }
+}