From 9b9e27194a78b261eecc27cb7074f9110920e67e Mon Sep 17 00:00:00 2001 From: JoyboyBrian Date: Wed, 29 Jul 2026 12:27:42 -0700 Subject: [PATCH 1/2] fix(api): return the JSON error envelope for unmatched routes --- src/api/proxy.rs | 31 ++++++++++++++++++++++++++++++- 1 file changed, 30 insertions(+), 1 deletion(-) diff --git a/src/api/proxy.rs b/src/api/proxy.rs index 207872b5..76eaefe6 100644 --- a/src/api/proxy.rs +++ b/src/api/proxy.rs @@ -224,7 +224,18 @@ where I: AsRef + Send + Sync, { if !has_routing_header(request.headers()) { - return StatusCode::NOT_FOUND.into_response(); + // Unmatched control-plane route: return the API error envelope so + // JSON clients surface "route not found" instead of failing to parse + // an empty 404 body. + let error = agentenv_http_server::models::Error::new( + 404, + format!( + "route not found: {} {}", + request.method(), + request.uri().path() + ), + ); + return (StatusCode::NOT_FOUND, axum::Json(error)).into_response(); } let forward_path = request.uri().path().to_owned(); with_route_source( @@ -2120,6 +2131,24 @@ mod tests { .unwrap(); assert_eq!(response.status(), StatusCode::NOT_FOUND); + let content_type = response + .headers() + .get(axum::http::header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .unwrap_or_default() + .to_owned(); + assert!( + content_type.starts_with("application/json"), + "unexpected content-type: {content_type}" + ); + let body = response.into_body().collect().await.unwrap().to_bytes(); + let payload: Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(payload["code"], 404); + let message = payload["message"].as_str().unwrap(); + assert!( + message.contains("route not found: GET /nonexistent/path"), + "unexpected message: {message}" + ); } #[tokio::test] From f9e9c3eb63a8865753bfbbfa0b35f7626bbb8623 Mon Sep 17 00:00:00 2001 From: JoyboyBrian Date: Wed, 29 Jul 2026 12:29:03 -0700 Subject: [PATCH 2/2] feat(snapshot): E2B alias rebuild semantics on template publish --- .../repository/backends/oss/repository.rs | 201 +++++++++++++++--- .../repository/backends/posixfs/backend.rs | 145 ++++++++++++- .../repository/backends/posixfs/catalog.rs | 179 +++++++++++----- 3 files changed, 426 insertions(+), 99 deletions(-) diff --git a/src/snapshot/repository/backends/oss/repository.rs b/src/snapshot/repository/backends/oss/repository.rs index 06f747ef..8af954a8 100644 --- a/src/snapshot/repository/backends/oss/repository.rs +++ b/src/snapshot/repository/backends/oss/repository.rs @@ -164,25 +164,28 @@ impl SnapshotRepository for OssSnapshotRepository { reason: format!("snapshot '{}' already exists", record.id), }); } + // When the alias already points at a live snapshot, leave the binding + // untouched so the existing template keeps resolving while the new + // build runs; a successful publish moves the alias to the new snapshot + // (E2B rebuild semantics). + let mut bind_on_create = true; if let Some(alias) = record.alias.as_ref() { if let Some(existing) = self.load_alias_target(alias.as_ref()).await? { if existing != record.id && self.snapshot_exists(&existing).await? { - return Err(RepositoryError::AliasConflict { - alias: alias.to_string(), - existing, - new_id: record.id.clone(), - }); + bind_on_create = false; } } } self.write_record(&record).await?; - if let Some(alias) = record.alias.as_ref() { - if let Err(error) = self.bind_alias(alias.as_ref(), &record.id).await { - let _ = self - .client - .delete(&OssSnapshotArtifactLayout::record_key(&record.id)) - .await; - return Err(error); + if bind_on_create { + if let Some(alias) = record.alias.as_ref() { + if let Err(error) = self.bind_alias(alias.as_ref(), &record.id, false).await { + let _ = self + .client + .delete(&OssSnapshotArtifactLayout::record_key(&record.id)) + .await; + return Err(error); + } } } Ok(record) @@ -287,26 +290,63 @@ impl SnapshotRepository for OssSnapshotRepository { disk_publications: disk_publications.clone(), }; - // 5. Bind alias (if present) with conflict detection. + // 5. Commit the record before moving the alias. This prevents an + // alias from ever resolving to a snapshot whose catalog record + // has not been published yet. The tradeoff is a crash window: + // dying after the record write but before `bind_alias` leaves a + // committed record whose `alias` field names an alias that still + // resolves to the previous snapshot, so readers of `record.alias` + // (listings) may observe the stale claim until the next rebind. + // Nothing reconciles that state automatically. + let previous_record = self.read_record(id).await?; + let previous_alias_target = if let Some(alias) = metadata.alias.as_ref() { + match self.load_alias_target(alias.as_ref()).await? { + Some(existing) if self.snapshot_exists(&existing).await? => Some(existing), + _ => None, + } + } else { + None + }; + let record = self + .write_committed_record( + metadata.id.clone(), + metadata.alias.clone(), + metadata.resources, + committed, + metadata.source.clone(), + ) + .await?; + + // 6. Move the alias only after the new record is readable. If the + // bind fails, restore the pending record and old alias. if let Some(ref alias) = metadata.alias { - if let Err(e) = self.bind_alias(alias.as_ref(), id).await { - // Best-effort rollback. Content-addressed managed layers are intentionally left - // in place; they are shared across snapshots and require separate GC. - if let Err(error) = self.client.delete_prefix(&layout.artifact_prefix()).await { - warn!(snapshot_id = %id, error = %error, "failed to roll back snapshot artifacts after alias bind failure"); + if let Err(error) = self.bind_alias(alias.as_ref(), id, true).await { + self.restore_alias_after_failed_bind( + alias.as_ref(), + id, + previous_alias_target.as_ref(), + ) + .await; + self.restore_record_after_failed_publish(id, previous_record.as_ref()) + .await; + return Err(error); + } + if let Some(previous_id) = previous_alias_target + .as_ref() + .filter(|previous_id| *previous_id != id) + { + if let Err(error) = self.clear_record_alias(previous_id, alias.as_ref()).await { + warn!( + alias = %alias, + previous_snapshot_id = %previous_id, + error = %error, + "failed to clear previous snapshot alias metadata" + ); } - return Err(e); } } - self.write_committed_record( - metadata.id.clone(), - metadata.alias.clone(), - metadata.resources, - committed, - metadata.source.clone(), - ) - .await + Ok(record) } .await; @@ -584,7 +624,8 @@ impl OssSnapshotRepository { /// Instead the algorithm is: /// 1. Read the current alias target. /// 2. If it already points to `id`, return success (idempotent). - /// 3. If it points to a live snapshot, return `AliasConflict`. + /// 3. If it points to a live snapshot: with `rebind` move the alias to + /// `id` (E2B rebuild semantics), otherwise return `AliasConflict`. /// 4. If it points to a deleted snapshot, remove the stale alias. /// 5. Write our binding unconditionally. /// 6. Read back and verify we won the race. If someone else wrote a @@ -595,7 +636,7 @@ impl OssSnapshotRepository { /// interval between our write and the subsequent read. This is weaker /// than a true CAS but sufficient for the current deployment model /// where concurrent publishes for the *same alias* are rare. - async fn bind_alias(&self, alias: &str, id: &SnapshotId) -> RepositoryResult<()> { + async fn bind_alias(&self, alias: &str, id: &SnapshotId, rebind: bool) -> RepositoryResult<()> { let key = validated_alias_key(alias)?; let payload = serde_json::to_vec(id) .map_err(|e| RepositoryError::backend("serialize alias binding", e))?; @@ -608,18 +649,19 @@ impl OssSnapshotRepository { } let still_exists = self.snapshot_exists(&existing_id).await?; - if still_exists { + if still_exists && !rebind { return Err(RepositoryError::AliasConflict { alias: alias.to_string(), existing: existing_id, new_id: id.clone(), }); } - - self.client - .delete(&key) - .await - .map_err(|e| RepositoryError::backend("delete stale alias", e))?; + if !still_exists { + self.client + .delete(&key) + .await + .map_err(|e| RepositoryError::backend("delete stale alias", e))?; + } } // Step 5: write our binding (unconditional — OSS does not @@ -669,6 +711,95 @@ impl OssSnapshotRepository { }) } + async fn restore_record_after_failed_publish( + &self, + id: &SnapshotId, + previous_record: Option<&SnapshotRecord>, + ) { + let result = match previous_record { + Some(record) => self.write_record(record).await, + None => self + .client + .delete(&OssSnapshotArtifactLayout::record_key(id)) + .await + .map_err(|error| RepositoryError::backend("remove failed snapshot record", error)), + }; + if let Err(error) = result { + warn!(snapshot_id = %id, error = %error, "failed to restore snapshot record after publish failure"); + } + } + + async fn restore_alias_after_failed_bind( + &self, + alias: &str, + id: &SnapshotId, + previous_id: Option<&SnapshotId>, + ) { + let current = match self.load_alias_target(alias).await { + Ok(current) => current, + Err(error) => { + warn!(alias, snapshot_id = %id, error = %error, "failed to inspect alias during publish rollback"); + return; + } + }; + // Skip when a concurrent publisher already moved the alias elsewhere. + // This only narrows the lost-update window: like `bind_alias`, the + // rollback cannot be atomic on a store without conditional writes, so a + // publisher that rebinds between this read and the write below is still + // clobbered. + if current.as_ref() != Some(id) { + return; + } + + let key = match validated_alias_key(alias) { + Ok(key) => key, + Err(error) => { + warn!(alias, snapshot_id = %id, error = %error, "failed to validate alias during publish rollback"); + return; + } + }; + let result = + match previous_id { + Some(previous_id) => match serde_json::to_vec(previous_id) { + Ok(payload) => { + self.client.put_bytes(&key, payload).await.map_err(|error| { + RepositoryError::backend("restore alias binding", error) + }) + } + Err(error) => Err(RepositoryError::backend( + "serialize restored alias binding", + error, + )), + }, + None => self.client.delete(&key).await.map_err(|error| { + RepositoryError::backend("remove failed alias binding", error) + }), + }; + if let Err(error) = result { + warn!(alias, snapshot_id = %id, error = %error, "failed to restore alias after publish failure"); + } + } + + /// Clears the alias field on the record that previously owned a rebound + /// alias so template listings do not report the moved name twice. + /// + /// Only `moved_alias` is cleared; a previous owner that already claims a + /// different name keeps it. + async fn clear_record_alias(&self, id: &SnapshotId, moved_alias: &str) -> RepositoryResult<()> { + if let Some(mut previous) = self.read_record(id).await? { + let claims_moved_alias = previous + .alias + .as_ref() + .is_some_and(|alias| alias.as_ref() == moved_alias); + if claims_moved_alias { + previous.alias = None; + previous.updated_at_unix_ms = now_unix_ms(); + self.write_record(&previous).await?; + } + } + Ok(()) + } + async fn export_managed_disk_image( &self, image_config_path: &Path, diff --git a/src/snapshot/repository/backends/posixfs/backend.rs b/src/snapshot/repository/backends/posixfs/backend.rs index 4dc7f7ba..459c1547 100644 --- a/src/snapshot/repository/backends/posixfs/backend.rs +++ b/src/snapshot/repository/backends/posixfs/backend.rs @@ -474,37 +474,162 @@ mod tests { } #[tokio::test] - async fn failed_commit_cleans_uncommitted_snapshot_directory() { + async fn publish_rebinds_existing_alias_to_new_snapshot() { let tempdir = TempDir::new().expect("tempdir should exist"); - let repository_root = tempdir.path().to_path_buf(); let repository = test_backend(tempdir.path()).repository(); let first_id = SnapshotId::generate(); let local_artifacts = seed_built_snapshot(tempdir.path()); - let first_metadata = sample_metadata(first_id.clone(), Some("conflict")); repository - .publish(first_metadata, local_artifacts) + .publish( + sample_metadata(first_id.clone(), Some("rebind")), + local_artifacts, + ) .await .expect("first publish should work"); let second_id = SnapshotId::generate(); let local_artifacts = seed_built_snapshot(tempdir.path()); - let err = repository + repository + .publish( + sample_metadata(second_id.clone(), Some("rebind")), + local_artifacts, + ) + .await + .expect("second publish should rebind the alias"); + + let resolved = repository + .resolve_alias("rebind") + .await + .expect("resolve should work") + .expect("alias should resolve"); + assert_eq!(resolved, second_id, "alias should move to the new snapshot"); + + let previous = repository + .get(first_id.to_string().as_str()) + .await + .expect("get should work") + .expect("previous snapshot should stay addressable by id"); + assert_eq!( + previous.alias, None, + "previous snapshot should lose the rebound alias" + ); + } + + #[tokio::test] + async fn failed_publish_keeps_previous_alias_and_removes_snapshot_dir() { + let tempdir = TempDir::new().expect("tempdir should exist"); + let repository = test_backend(tempdir.path()).repository(); + + let first_id = SnapshotId::generate(); + let local_artifacts = seed_built_snapshot(tempdir.path()); + repository .publish( - sample_metadata(second_id.clone(), Some("conflict")), + sample_metadata(first_id.clone(), Some("rebind")), local_artifacts, ) .await - .expect_err("second publish should fail"); + .expect("first publish should work"); + + let second_id = SnapshotId::generate(); + let broken_artifacts = seed_built_snapshot(tempdir.path()); + // `import_built_artifacts` copies `vm_state.bin` first, so removing it + // fails the publish before any catalog state is committed. + fs::remove_file(&broken_artifacts.vm_state.path).expect("remove seeded vm state"); + repository + .publish( + sample_metadata(second_id.clone(), Some("rebind")), + broken_artifacts, + ) + .await + .expect_err("publish should fail when the vm state artifact is missing"); - assert!(matches!(err, RepositoryError::AliasConflict { .. })); assert!( - !repository_root + !tempdir + .path() .join("snapshots") .join(second_id.to_string()) .exists(), - "failed publish should not leave a committed revision directory" + "failed publish should not leave a snapshot directory behind" + ); + + let resolved = repository + .resolve_alias("rebind") + .await + .expect("resolve should work") + .expect("alias should still resolve"); + assert_eq!( + resolved, first_id, + "alias should stay bound to the previously committed snapshot" ); + + let previous = repository + .get(first_id.to_string().as_str()) + .await + .expect("get should work") + .expect("previous snapshot should stay addressable by id"); + assert_eq!( + previous.alias.as_ref().map(ToString::to_string), + Some("rebind".to_string()), + "previous snapshot should keep the alias after a failed rebind" + ); + } + + #[tokio::test] + async fn create_keeps_existing_alias_until_new_build_commits() { + let tempdir = TempDir::new().expect("tempdir should exist"); + let repository = test_backend(tempdir.path()).repository(); + + let committed_id = SnapshotId::generate(); + let local_artifacts = seed_built_snapshot(tempdir.path()); + repository + .publish( + sample_metadata(committed_id.clone(), Some("stable")), + local_artifacts, + ) + .await + .expect("publish should work"); + + let waiting = SnapshotRecord::template_waiting( + SnapshotId::generate(), + Some(SnapshotAlias::parse("stable").expect("alias should parse")), + crate::types::SandboxResources { + cpu_count: 1, + memory_mib: 256, + disk_size_mib: 0, + }, + ); + let waiting_id = waiting.id.clone(); + repository + .create(waiting) + .await + .expect("create with an existing alias should be allowed"); + + let resolved = repository + .resolve_alias("stable") + .await + .expect("resolve should work") + .expect("alias should resolve"); + assert_eq!( + resolved, committed_id, + "alias should keep pointing at the committed snapshot while the rebuild is pending" + ); + + let local_artifacts = seed_built_snapshot(tempdir.path()); + repository + .publish( + sample_metadata(waiting_id.clone(), Some("stable")), + local_artifacts, + ) + .await + .expect("publishing the rebuild should rebind the alias"); + + let resolved = repository + .resolve_alias("stable") + .await + .expect("resolve should work") + .expect("alias should resolve"); + assert_eq!(resolved, waiting_id, "alias should move after commit"); } #[tokio::test] diff --git a/src/snapshot/repository/backends/posixfs/catalog.rs b/src/snapshot/repository/backends/posixfs/catalog.rs index 6c36d6bd..76b7486c 100644 --- a/src/snapshot/repository/backends/posixfs/catalog.rs +++ b/src/snapshot/repository/backends/posixfs/catalog.rs @@ -6,6 +6,7 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use serde::de::DeserializeOwned; use serde::Serialize; +use tracing::warn; use super::layout::PosixFsSnapshotArtifactLayout; use crate::snapshot::repository::SnapshotListFilter; @@ -79,9 +80,9 @@ impl PosixFsCatalogStore { /// /// Flow: /// 1. acquire the alias lock when an alias is present - /// 2. bind the alias - /// 3. write the commit marker - /// 4. write the committed snapshot record + /// 2. write the commit marker + /// 3. write the committed snapshot record + /// 4. atomically bind the alias as the final visible operation pub(crate) fn commit_publish( &self, session: &PublishSession, @@ -90,25 +91,30 @@ impl PosixFsCatalogStore { ) -> RepositoryResult { let now = now_unix_ms(); let snapshot_id = metadata.id.clone(); + let previous_record = self.load_record_by_id_unlocked(&snapshot_id)?; let write_result = if let Some(alias) = metadata.alias.as_ref() { self.with_alias_lock(alias, |store| { let record = store.committed_record_unlocked(&metadata, committed.clone(), now)?; let alias_path = PosixFsSnapshotArtifactLayout::alias_path(&store.root, alias); - if let Some(existing) = store.load_alias_target(alias)? { - if existing != snapshot_id { - if store.load_record_by_id_unlocked(&existing)?.is_some() { - return Err(RepositoryError::AliasConflict { - alias: alias.to_string(), - existing, - new_id: snapshot_id.clone(), - }); - } - store.remove_file_if_exists(&alias_path)?; - } - } - store.write_json(&alias_path, &snapshot_id)?; + let existing = store.load_alias_target(alias)?; store.write_commit_marker(&session.snapshot_id)?; store.write_committed_record_unlocked(&record)?; + // `write_json` uses an atomic rename. Keeping this as the final + // fallible operation means a failed rebuild leaves the old + // alias binding untouched. The tradeoff is a crash window: dying + // after the record write but before the alias write leaves a + // committed record whose `alias` field names an alias that still + // resolves to the previous snapshot, so readers of `record.alias` + // (listings) may observe the stale claim until the next rebind. + store.write_json(&alias_path, &snapshot_id)?; + + if let Some(existing) = existing.filter(|existing| existing != &snapshot_id) { + // The previous snapshot stays addressable by id, so running + // sandboxes and explicit id references keep working. Alias + // metadata cleanup is best effort because the binding has + // already moved successfully. + store.clear_moved_alias_on_previous_record(&existing, alias.as_ref(), now); + } Ok(record) }) } else { @@ -123,17 +129,7 @@ impl PosixFsCatalogStore { match write_result { Ok(record) => Ok(record), Err(error) => { - if let Some(alias) = metadata.alias.as_ref() { - let _ = self.with_alias_lock(alias, |store| { - let alias_path = - PosixFsSnapshotArtifactLayout::alias_path(&store.root, alias); - if store.load_alias_target(alias)?.as_ref() == Some(&snapshot_id) { - store.remove_file_if_exists(&alias_path)?; - } - Ok(()) - }); - } - let _ = self.cleanup_uncommitted_snapshot_dir(&session.snapshot_id); + self.rollback_failed_publish(&session.snapshot_id, previous_record.as_ref()); Err(error) } } @@ -164,12 +160,35 @@ impl PosixFsCatalogStore { if let Some(alias) = record.alias.as_ref() { self.with_alias_lock(alias, |store| { - store.ensure_alias_available(alias, &record.id)?; store.write_record_unlocked(&record)?; - store.write_json( - &PosixFsSnapshotArtifactLayout::alias_path(&store.root, alias), - &record.id, - ) + let alias_path = PosixFsSnapshotArtifactLayout::alias_path(&store.root, alias); + let bind = (|| -> RepositoryResult<()> { + match store.load_alias_target(alias)? { + // The alias currently points at a live snapshot. Leave the + // binding untouched so the existing template keeps resolving + // while the new build runs; a successful commit moves the + // alias to the new snapshot (E2B rebuild semantics). + Some(existing) + if existing != record.id + && store.load_record_by_id_unlocked(&existing)?.is_some() => + { + Ok(()) + } + Some(existing) if existing != record.id => { + store.remove_file_if_exists(&alias_path)?; + store.write_json(&alias_path, &record.id) + } + _ => store.write_json(&alias_path, &record.id), + } + })(); + if let Err(error) = bind { + // Keep creation all-or-nothing under the alias lock: a record + // that survives a failed binding claims the alias in listings + // with nothing left to reconcile it. + let _ = store.remove_file_if_exists(&store.record_path(&record.id)); + return Err(error); + } + Ok(()) })?; } else { self.write_record_unlocked(&record)?; @@ -376,6 +395,67 @@ impl PosixFsCatalogStore { self.write_record_unlocked(&record) } + /// Clears `moved_alias` from the record that owned it before a rebind. + /// + /// Best effort: the alias binding has already moved, so a failure here only + /// leaves stale alias metadata on the previous owner's record. + fn clear_moved_alias_on_previous_record( + &self, + previous_id: &SnapshotId, + moved_alias: &str, + now: i64, + ) { + // Lock order is alias lock first, then record lock; nothing takes them + // in the reverse order today. + let _guard = match self.acquire_record_lock(previous_id) { + Ok(guard) => guard, + Err(error) => { + warn!( + alias = %moved_alias, + previous_snapshot_id = %previous_id, + error = %error, + "failed to lock previous snapshot record for alias cleanup" + ); + return; + } + }; + + let mut previous = match self.load_record_by_id_unlocked(previous_id) { + Ok(Some(previous)) => previous, + Ok(None) => return, + Err(error) => { + warn!( + alias = %moved_alias, + previous_snapshot_id = %previous_id, + error = %error, + "failed to load previous snapshot alias metadata" + ); + return; + } + }; + + // Only clear the alias this publish actually moved; the previous owner + // may already claim a different name. + let claims_moved_alias = previous + .alias + .as_ref() + .is_some_and(|alias| alias.as_ref() == moved_alias); + if !claims_moved_alias { + return; + } + + previous.alias = None; + previous.updated_at_unix_ms = now; + if let Err(error) = self.write_record_unlocked(&previous) { + warn!( + alias = %moved_alias, + previous_snapshot_id = %previous_id, + error = %error, + "failed to clear previous snapshot alias metadata" + ); + } + } + fn read_json(&self, path: &Path) -> RepositoryResult where T: DeserializeOwned, @@ -498,6 +578,19 @@ impl PosixFsCatalogStore { self.remove_dir_if_exists(&snapshot_layout.snapshot_dir()) } + fn rollback_failed_publish(&self, id: &SnapshotId, previous_record: Option<&SnapshotRecord>) { + if let Err(error) = self.remove_dir_if_exists(&self.layout(id).snapshot_dir()) { + warn!(snapshot_id = %id, error = %error, "failed to remove snapshot artifacts after publish failure"); + } + let restore_result = match previous_record { + Some(record) => self.write_record_unlocked(record), + None => self.remove_file_if_exists(&self.record_path(id)), + }; + if let Err(error) = restore_result { + warn!(snapshot_id = %id, error = %error, "failed to restore snapshot record after publish failure"); + } + } + fn load_record_by_id_unlocked( &self, id: &SnapshotId, @@ -626,28 +719,6 @@ impl PosixFsCatalogStore { action(self) } - fn ensure_alias_available( - &self, - alias: &SnapshotAlias, - new_id: &SnapshotId, - ) -> RepositoryResult<()> { - let alias_path = PosixFsSnapshotArtifactLayout::alias_path(&self.root, alias); - if let Some(existing) = self.load_alias_target(alias)? { - if &existing == new_id { - return Ok(()); - } - if self.load_record_by_id_unlocked(&existing)?.is_some() { - return Err(RepositoryError::AliasConflict { - alias: alias.to_string(), - existing, - new_id: new_id.clone(), - }); - } - self.remove_file_if_exists(&alias_path)?; - } - Ok(()) - } - fn write_record_unlocked(&self, record: &SnapshotRecord) -> RepositoryResult<()> { self.write_json(&self.record_path(&record.id), record) }