feat: US-093 - Add end-to-end dual-purpose shard cache tests

Author: NathanFlurry

engine/packages/depot/CLAUDE.md

@@ -178,6 +178,7 @@ We explicitly do **not** import:
 - Tests run against real UDB via `test_db()` (RocksDB-backed temp instance). No mocks for storage paths.
 - Shared depot integration-test helpers live in `tests/common/mod.rs`; use them for temp UDB creation, memory UPS construction, `Db` construction, and raw key reads instead of redefining helpers per test file.
 - Cold-tier tests use `ColdTier::Filesystem` (local filesystem stand-in for S3). UPS dispatch tests use the UPS memory driver. No real S3 required.
+- Workflow cold-publish e2e tests may need a short-lived restore point to pin commit metadata until cold publish; delete it before asserting shard-cache eviction.
 - Workflow compaction tests using real `Db::commit` should assert versionstamp-independent invariants; real UDB versionstamps do not encode txid bytes.
 - Workflow force-compaction tests should wait for manager companion workflow ids before signaling so the manager is in its durable loop.
 - Workflow compaction race tests can use debug-only workflow `test_hooks`; use stored `Notify` permits with `notify_one()` when a notification may arrive before the waiter arms.

engine/packages/depot/src/workflows/compaction/cold.rs

@@ -13,10 +13,11 @@ pub async fn upload_cold_job(
 	input: &UploadColdJobInput,
 ) -> Result<UploadColdJobOutput> {
 	let input = input.clone();
+	let input_for_tx = input.clone();
 	let upload = ctx
 		.udb()?
 		.run(move |tx| {
-			let input = input.clone();
+			let input = input_for_tx.clone();
 			async move { prepare_cold_upload_tx(&tx, &input).await }
 		})
 		.await?;
@@ -42,10 +43,21 @@ pub async fn upload_cold_job(
 		});
 	};
 	for object in objects {
-		cold_tier
-			.put_object(&object.object_key, &object.bytes)
-			.await
-			.with_context(|| format!("put sqlite workflow cold shard {}", object.object_key))?;
+		if let Err(err) = cold_tier.put_object(&object.object_key, &object.bytes).await {
+			tracing::warn!(
+				?input.database_branch_id,
+				?input.job_id,
+				object_key = object.object_key.as_str(),
+				error = ?err,
+				"sqlite workflow cold shard upload failed"
+			);
+			return Ok(UploadColdJobOutput {
+				status: CompactionJobStatus::Rejected {
+					reason: "cold shard upload failed".to_string(),
+				},
+				output_refs: Vec::new(),
+			});
+		}
 	}
 
 	Ok(UploadColdJobOutput {