feat: US-035 - Add Depot inspect API with big metadata JSON and paginated row scans

Author: NathanFlurry

.agent/specs/depot-inspect-api.md

@@ -9,6 +9,7 @@ license.workspace = true
 anyhow.workspace = true
 axum.workspace = true
 base64.workspace = true
+depot.workspace = true
 epoxy.workspace = true
 epoxy-protocol.workspace = true
 futures-util.workspace = true
@@ -32,3 +33,7 @@ universalpubsub.workspace = true
 utoipa.workspace = true
 uuid.workspace = true
 universaldb.workspace = true
+
+[dev-dependencies]
+axum-test.workspace = true
+tempfile.workspace = true

Cargo.lock

@@ -0,0 +1,166 @@
+use anyhow::{Context, Result};
+use depot::{
+	inspect::{self, CatalogQuery, RawScanQuery, RowsQuery, SampleQuery},
+	types::{BucketId, DatabaseBranchId},
+};
+use rivet_api_builder::ApiCtx;
+use serde::Deserialize;
+use uuid::Uuid;
+
+#[derive(Debug, Deserialize)]
+pub struct BucketPath {
+	pub bucket_id: String,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct DatabasePath {
+	pub bucket_id: String,
+	pub database_id: String,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct BranchPath {
+	pub branch_id: String,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct BranchRowsPath {
+	pub branch_id: String,
+	pub family: String,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct PageTracePath {
+	pub branch_id: String,
+	pub pgno: u32,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct RawKeyPath {
+	pub key: String,
+}
+
+pub async fn summary(ctx: ApiCtx, _path: (), _query: ()) -> Result<inspect::InspectResponse> {
+	let udb = ctx.pools().udb()?;
+	inspect::summary(&udb, ctx.pools().node_id()).await
+}
+
+pub async fn catalog(
+	ctx: ApiCtx,
+	_path: (),
+	query: CatalogQuery,
+) -> Result<inspect::CatalogResponse> {
+	let udb = ctx.pools().udb()?;
+	inspect::catalog(&udb, ctx.pools().node_id(), query).await
+}
+
+pub async fn bucket(
+	ctx: ApiCtx,
+	path: BucketPath,
+	query: SampleQuery,
+) -> Result<inspect::InspectResponse> {
+	let bucket_id = parse_bucket_id(&path.bucket_id)?;
+	let udb = ctx.pools().udb()?;
+	inspect::bucket(&udb, ctx.pools().node_id(), bucket_id, query).await
+}
+
+pub async fn database(
+	ctx: ApiCtx,
+	path: DatabasePath,
+	query: SampleQuery,
+) -> Result<inspect::InspectResponse> {
+	let bucket_id = parse_bucket_id(&path.bucket_id)?;
+	let udb = ctx.pools().udb()?;
+	inspect::database(
+		&udb,
+		ctx.pools().node_id(),
+		bucket_id,
+		path.database_id,
+		query,
+	)
+	.await
+}
+
+pub async fn branch(
+	ctx: ApiCtx,
+	path: BranchPath,
+	query: SampleQuery,
+) -> Result<inspect::InspectResponse> {
+	let branch_id = parse_database_branch_id(&path.branch_id)?;
+	let udb = ctx.pools().udb()?;
+	inspect::branch(&udb, ctx.pools().node_id(), branch_id, query).await
+}
+
+pub async fn page_trace(
+	ctx: ApiCtx,
+	path: PageTracePath,
+	_query: (),
+) -> Result<inspect::InspectResponse> {
+	let branch_id = parse_database_branch_id(&path.branch_id)?;
+	let udb = ctx.pools().udb()?;
+	inspect::page_trace(
+		&udb,
+		ctx.pools().node_id(),
+		branch_id,
+		path.pgno,
+	)
+	.await
+}
+
+pub async fn branch_rows(
+	ctx: ApiCtx,
+	path: BranchRowsPath,
+	query: RowsQuery,
+) -> Result<inspect::PaginatedRowsResponse> {
+	let branch_id = parse_database_branch_id(&path.branch_id)?;
+	let family = inspect::RowFamily::parse(&path.family)?;
+	let udb = ctx.pools().udb()?;
+	inspect::branch_rows(
+		&udb,
+		ctx.pools().node_id(),
+		branch_id,
+		family,
+		query,
+	)
+	.await
+}
+
+pub async fn raw_key(
+	ctx: ApiCtx,
+	path: RawKeyPath,
+	_query: (),
+) -> Result<inspect::InspectResponse> {
+	let key = inspect::decode_path_key(&path.key)?;
+	let udb = ctx.pools().udb()?;
+	inspect::raw_key(&udb, ctx.pools().node_id(), key).await
+}
+
+pub async fn raw_scan(
+	ctx: ApiCtx,
+	_path: (),
+	query: RawScanQuery,
+) -> Result<inspect::PaginatedRowsResponse> {
+	let udb = ctx.pools().udb()?;
+	inspect::raw_scan(&udb, ctx.pools().node_id(), query).await
+}
+
+pub async fn decode_key(
+	ctx: ApiCtx,
+	path: RawKeyPath,
+	_query: (),
+) -> Result<inspect::InspectResponse> {
+	let key = inspect::decode_path_key(&path.key)?;
+	inspect::decode_key_response(ctx.pools().node_id(), key)
+}
+
+fn parse_bucket_id(value: &str) -> Result<BucketId> {
+	Ok(BucketId::from_uuid(
+		Uuid::parse_str(value).context("parse Depot bucket id")?,
+	))
+}
+
+fn parse_database_branch_id(value: &str) -> Result<DatabaseBranchId> {
+	Ok(DatabaseBranchId::from_uuid(
+		Uuid::parse_str(value).context("parse Depot database branch id")?,
+	))
+}

engine/packages/api-peer/Cargo.toml

@@ -3,6 +3,7 @@ use std::net::SocketAddr;
 use anyhow::*;
 
 pub mod actors;
+pub mod depot_inspect;
 pub mod envoys;
 pub mod internal;
 pub mod namespaces;

engine/packages/api-peer/src/depot_inspect.rs

@@ -1,6 +1,6 @@
 use rivet_api_builder::{create_router, prelude::*};
 
-use crate::{actors, envoys, internal, namespaces, runner_configs, runners};
+use crate::{actors, depot_inspect, envoys, internal, namespaces, runner_configs, runners};
 
 #[tracing::instrument(skip_all)]
 pub async fn router(
@@ -40,6 +40,38 @@ pub async fn router(
 			.route("/runners/names", get(runners::list_names))
 			// MARK: Envoys
 			.route("/envoys", get(envoys::list))
+			// MARK: Depot inspect
+			.route("/depot/inspect/summary", get(depot_inspect::summary))
+			.route("/depot/inspect/catalog", get(depot_inspect::catalog))
+			.route(
+				"/depot/inspect/buckets/{bucket_id}",
+				get(depot_inspect::bucket),
+			)
+			.route(
+				"/depot/inspect/buckets/{bucket_id}/databases/{database_id}",
+				get(depot_inspect::database),
+			)
+			.route(
+				"/depot/inspect/branches/{branch_id}",
+				get(depot_inspect::branch),
+			)
+			.route(
+				"/depot/inspect/branches/{branch_id}/pages/{pgno}/trace",
+				get(depot_inspect::page_trace),
+			)
+			.route(
+				"/depot/inspect/branches/{branch_id}/rows/{family}",
+				get(depot_inspect::branch_rows),
+			)
+			.route(
+				"/depot/inspect/raw/key/{key}",
+				get(depot_inspect::raw_key),
+			)
+			.route("/depot/inspect/raw/scan", get(depot_inspect::raw_scan))
+			.route(
+				"/depot/inspect/raw/decode-key/{key}",
+				get(depot_inspect::decode_key),
+			)
 			// MARK: Internal
 			.route("/cache/purge", post(internal::cache_purge))
 			.route(

engine/packages/api-peer/src/lib.rs

@@ -0,0 +1,35 @@
+use axum_test::TestServer;
+use anyhow::Result;
+use base64::{Engine, prelude::BASE64_URL_SAFE_NO_PAD};
+use rivet_config::config::{Database, Root, db::FileSystem};
+
+#[tokio::test]
+async fn depot_inspect_routes_are_registered_on_api_peer() -> Result<()> {
+	let tempdir = tempfile::tempdir()?;
+	let mut root = Root::default();
+	root.database = Some(Database::FileSystem(FileSystem {
+		path: tempdir.path().join("udb"),
+	}));
+	let config = rivet_config::Config::from_root(root);
+	let pools = rivet_pools::Pools::test(config.clone()).await?;
+	let app = rivet_api_peer::create_router(config, pools).await?;
+	let server = TestServer::new(app)?;
+
+	let res = server.get("/depot/inspect/summary").await;
+	res.assert_status_ok();
+
+	let key = BASE64_URL_SAFE_NO_PAD.encode([depot::keys::SQLITE_SUBSPACE_PREFIX]);
+	let res = server
+		.get(&format!("/depot/inspect/raw/decode-key/{key}"))
+		.await;
+	res.assert_status_ok();
+
+	let branch_id = uuid::Uuid::nil();
+	let res = server
+		.get(&format!("/depot/inspect/branches/{branch_id}/rows/commits"))
+		.add_query_param("limit", "1")
+		.await;
+	res.assert_status_ok();
+
+	Ok(())
+}

engine/packages/api-peer/src/router.rs

@@ -98,6 +98,7 @@ These come from `r2-prior-art/.agent/research/sqlite/requirements.md` and supers
 - **`Db` constructors do not take UPS.** Keep workflow wakeups behind `CompactionSignaler`; do not reintroduce legacy compactor PubSub wiring.
 - **Truncate cleanup fences observed PIDX and SHARD values before writing.** Snapshot-selected rows must be re-read with Serializable isolation before cleanup clears or rewrites them so concurrent workflow output cannot be clobbered and quota accounting retries cleanly.
 - **UDB helper scans return logical conveyer keys.** Apply the supplied `Subspace` to the physical range scan, then strip it before returning rows to callers.
+- **Depot inspect decode and pagination logic lives in `depot::inspect`.** `api-peer` should only mount thin internal `/depot/inspect/...` handlers and must not expose this surface through public SDKs.
 - **Conveyer persisted payload structs use `serde::{Serialize, Deserialize}` as the serde_bare/vbare-compatible derive pattern.** Add `OwnedVersionedData` wrappers when introducing encode/decode helpers.
 - **Conveyer type domains live behind the `conveyer/types.rs` facade.** Add branch, restore_point, compaction, history-pin, cold-manifest, storage, page, and id payloads under `conveyer/types/*.rs` and re-export public names from the facade.
 - **META splits into single-writer sub-keys:** `/META/head` (commit-owned), `/META/quota` (atomic-add counter, raw i64 LE, not vbare), and workflow `CMP/root` (manager-owned compaction manifest).

engine/packages/api-peer/tests/depot_inspect.rs

@@ -11,6 +11,7 @@ async-channel.workspace = true
 async-trait.workspace = true
 aws-config = "1"
 aws-sdk-s3 = "1"
+base64.workspace = true
 futures-util.workspace = true
 gas.workspace = true
 lazy_static.workspace = true

engine/packages/depot/CLAUDE.md

@@ -1,6 +1,6 @@
 use std::{collections::BTreeSet, sync::Arc};
 
-use anyhow::{Context, Result};
+use anyhow::{Context, Result, ensure};
 use universaldb::{
 	options::MutationType,
 	utils::IsolationLevel::{Serializable, Snapshot},
@@ -43,6 +43,8 @@ impl Db {
 		db_size_pages: u32,
 		now_ms: i64,
 	) -> Result<()> {
+		validate_dirty_pages(&dirty_pages)?;
+
 		let node_id = self.node_id.to_string();
 		let labels = &[node_id.as_str()];
 		let _timer = metrics::SQLITE_PUMP_COMMIT_DURATION
@@ -401,6 +403,27 @@ impl Db {
 	}
 }
 
+fn validate_dirty_pages(dirty_pages: &[DirtyPage]) -> Result<()> {
+	let mut seen = BTreeSet::new();
+	for page in dirty_pages {
+		ensure!(page.pgno > 0, "sqlite commit does not accept page 0");
+		ensure!(
+			page.bytes.len() == keys::PAGE_SIZE as usize,
+			"sqlite commit page {} had {} bytes, expected {}",
+			page.pgno,
+			page.bytes.len(),
+			keys::PAGE_SIZE
+		);
+		ensure!(
+			seen.insert(page.pgno),
+			"sqlite commit duplicated page {} in a single request",
+			page.pgno
+		);
+	}
+
+	Ok(())
+}
+
 struct CommitTxResult {
 	branch_id: DatabaseBranchId,
 	branch_ancestry: BranchAncestry,