diff --git a/Cargo.lock b/Cargo.lock index 99402ae0..e1fdb404 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4994,7 +4994,6 @@ dependencies = [ "omnigraph-engine", "serde", "serde_json", - "sha2 0.10.9", "utoipa", ] diff --git a/crates/omnigraph-api-types/Cargo.toml b/crates/omnigraph-api-types/Cargo.toml index 6aed5f7c..114c2401 100644 --- a/crates/omnigraph-api-types/Cargo.toml +++ b/crates/omnigraph-api-types/Cargo.toml @@ -13,5 +13,4 @@ omnigraph = { package = "omnigraph-engine", path = "../omnigraph", version = "0. omnigraph-compiler = { path = "../omnigraph-compiler", version = "0.8.1" } serde = { workspace = true } serde_json = { workspace = true } -sha2 = { workspace = true } utoipa = { workspace = true } diff --git a/crates/omnigraph-api-types/src/lib.rs b/crates/omnigraph-api-types/src/lib.rs index 59a1c5d9..d827cef2 100644 --- a/crates/omnigraph-api-types/src/lib.rs +++ b/crates/omnigraph-api-types/src/lib.rs @@ -569,6 +569,23 @@ pub struct BlobQuery { pub snapshot: Option, } +/// Response body for `PUT /blob` (raw-bytes blob upload). Also printed +/// verbatim by the CLI's `blob put` so both arms emit the identical object. +#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] +pub struct BlobPutOutput { + #[serde(rename = "type")] + pub type_name: String, + pub id: String, + pub prop: String, + pub branch: String, + /// Stored payload size in bytes. + pub size: u64, + /// Strong quoted validator of the committed cell — identical to what a + /// follow-up `GET /blob` serves. + pub etag: String, + pub actor_id: Option, +} + #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct HealthOutput { pub status: String, @@ -802,19 +819,30 @@ pub struct GraphListResponse { /// the server's `/blob` responses and the CLI's embedded `blob stat` so /// both arms derive the identical validator from one definition. pub fn blob_etag(tag: &omnigraph::db::BlobVersionTag, property: &str) -> String { - use sha2::{Digest, Sha256}; - let mut hasher = Sha256::new(); - hasher.update(tag.stable_table_id.to_le_bytes()); - hasher.update(tag.table_incarnation_id.to_le_bytes()); - hasher.update(tag.table_version.to_le_bytes()); - hasher.update(tag.row_id.to_le_bytes()); - hasher.update(property.as_bytes()); - let digest = hasher.finalize(); - let mut out = String::with_capacity(34); - out.push('"'); - for byte in &digest[..16] { - out.push_str(&format!("{byte:02x}")); + omnigraph::db::blob_etag(tag, property) +} + +/// RFC 9110 `If-Match` header -> the engine's transport-neutral +/// [`omnigraph::db::BlobPrecondition`]. Shared by the server's `PUT /blob` +/// handler and the CLI's embedded `blob put` arm so both boundaries parse +/// the wire form identically (the engine itself compares opaque tokens +/// only). `*` is `AnyExisting`; otherwise a comma-separated entity-tag +/// list. If-Match requires STRONG comparison (RFC 9110 §13.1.1), so a weak +/// `W/"..."` tag never matches and is dropped here — a header of only weak +/// tags yields `Tags(vec![])`, which matches nothing (412). Splitting on +/// commas is exact for the hex tags this server mints; an exotic quoted +/// comma only costs a spurious 412, never a wrong success. +pub fn parse_if_match(header: &str) -> omnigraph::db::BlobPrecondition { + let header = header.trim(); + if header == "*" { + return omnigraph::db::BlobPrecondition::AnyExisting; } - out.push('"'); - out + omnigraph::db::BlobPrecondition::Tags( + header + .split(',') + .map(str::trim) + .filter(|tag| !tag.is_empty() && !tag.starts_with("W/")) + .map(str::to_string) + .collect(), + ) } diff --git a/crates/omnigraph-cli/src/cli.rs b/crates/omnigraph-cli/src/cli.rs index 4cd25714..f6b6f45c 100644 --- a/crates/omnigraph-cli/src/cli.rs +++ b/crates/omnigraph-cli/src/cli.rs @@ -15,7 +15,7 @@ pub(crate) const DEFAULT_BEARER_TOKEN_ENV: &str = "OMNIGRAPH_BEARER_TOKEN"; #[command(after_help = "\ COMMANDS BY CAPABILITY:\n \ any — run against a graph, served (--server / --profile) or embedded (--store / a \ -URI): query, mutate, load, branch, blob get/stat, snapshot, export, commit, schema show/apply.\n \ +URI): query, mutate, load, branch, blob get/put/stat, snapshot, export, commit, schema show/apply.\n \ served — require a server: graphs.\n \ direct — direct storage access; reject --server (init, optimize, repair, cleanup, \ schema plan, lint).\n \ @@ -514,6 +514,31 @@ pub(crate) enum BlobCommand { #[arg(long)] length: Option, }, + /// Upload raw bytes into one existing row's Blob property + Put { + /// Node type name + #[arg(value_name = "TYPE")] + type_name: String, + /// Logical row id (the @key value) + id: String, + /// Blob property name + prop: String, + /// Graph URI + #[arg(long)] + uri: Option, + /// Branch to write to (defaults to main) + #[arg(long)] + branch: Option, + /// Read the payload from PATH instead of stdin + #[arg(long, value_name = "PATH")] + file: Option, + /// Conditional write: entity-tag list or `*` (RFC 9110 If-Match, + /// strong comparison); a stale validator fails with the current one + #[arg(long, value_name = "ETAG")] + if_match: Option, + #[arg(long)] + json: bool, + }, /// Show one blob cell's metadata (kind, size, etag, external uri) Stat { /// Node type name diff --git a/crates/omnigraph-cli/src/client.rs b/crates/omnigraph-cli/src/client.rs index e94d688d..ae09dab9 100644 --- a/crates/omnigraph-cli/src/client.rs +++ b/crates/omnigraph-cli/src/client.rs @@ -27,7 +27,7 @@ use color_eyre::Result; use color_eyre::eyre::bail; use omnigraph::db::{BlobContent, Omnigraph, ReadTarget}; use omnigraph_api_types::{ - BranchCreateOutput, BranchCreateRequest, BranchDeleteOutput, BranchListOutput, + BlobPutOutput, BranchCreateOutput, BranchCreateRequest, BranchDeleteOutput, BranchListOutput, BranchMergeOutput, BranchMergeRequest, ChangeOutput, CommitListOutput, CommitOutput, ErrorOutput, ExportRequest, GraphListResponse, IngestOutput, IngestRequest, InvokeStoredQueryRequest, ReadOutput, @@ -124,6 +124,15 @@ fn blob_target_param(target: &ReadTarget) -> (&'static str, String) { } } +/// The one typed stale-precondition error BOTH blob-put arms return, so the +/// remote 412 and the embedded typed outcome are indistinguishable at the +/// CLI surface (pinned by the parity matrix). +fn blob_precondition_error(current_etag: &str) -> color_eyre::Report { + color_eyre::eyre::eyre!( + "blob precondition failed: If-Match did not match the cell's current validator {current_etag}" + ) +} + /// The `Location` of a `302` blob answer. A redirect without a location is /// a server contract violation, surfaced loudly rather than fabricated. fn blob_redirect_location(response: &reqwest::Response) -> Result { @@ -1049,6 +1058,93 @@ impl GraphClient { } } + /// `blob put` — upload raw payload bytes into one existing row's Blob + /// cell. Update-only; an optional `--if-match` validator (raw RFC 9110 + /// header form, parsed HERE at the boundary) makes the write conditional. + /// Both arms surface a stale precondition through the SAME typed error + /// text carrying the current validator, so exit codes and messages hold + /// parity (remote derives it from the 412's ETag header, embedded from + /// the engine's typed outcome). + #[allow(clippy::too_many_arguments)] + pub(crate) async fn blob_put( + &self, + branch: &str, + type_name: &str, + id: &str, + prop: &str, + bytes: Vec, + if_match: Option<&str>, + ) -> Result { + match self { + GraphClient::Remote { + http, + base_url, + token, + } => { + let url = remote_url( + base_url, + &["blob"], + &[ + ("type", type_name), + ("id", id), + ("prop", prop), + ("branch", branch), + ], + )?; + let mut request = apply_bearer_token(http.request(Method::PUT, url), token.as_deref()) + .body(bytes); + if let Some(validator) = if_match { + request = request.header(reqwest::header::IF_MATCH, validator); + } + let response = request.send().await?; + let status = response.status(); + if status == reqwest::StatusCode::PRECONDITION_FAILED { + let current = response + .headers() + .get(reqwest::header::ETAG) + .and_then(|value| value.to_str().ok()) + .unwrap_or("") + .to_string(); + return Err(blob_precondition_error(¤t)); + } + if !status.is_success() { + let text = response.text().await?; + if let Ok(error) = serde_json::from_str::(&text) { + bail!(error.error); + } + bail!("server returned {}: {}", status, text); + } + Ok(response.json::().await?) + } + GraphClient::Embedded { uri, actor } => { + let db = Self::open_embedded(uri).await?; + let actor = actor.as_deref(); + let precondition = if_match.map(omnigraph_api_types::parse_if_match); + let size_hint = bytes.len() as u64; + let outcome = db + .write_blob_at_as(branch, type_name, id, prop, bytes, precondition, actor) + .await?; + match outcome { + omnigraph::db::BlobWriteOutcome::Written { etag, size } => { + debug_assert_eq!(size, size_hint); + Ok(BlobPutOutput { + type_name: type_name.to_string(), + id: id.to_string(), + prop: prop.to_string(), + branch: branch.to_string(), + size, + etag, + actor_id: actor.map(String::from), + }) + } + omnigraph::db::BlobWriteOutcome::PreconditionFailed { current_etag } => { + Err(blob_precondition_error(¤t_etag)) + } + } + } + } + } + /// `graphs list` — enumerate the graphs a remote multi-graph server /// serves (`GET /graphs`). Remote-only by design: there is no local /// enumeration endpoint, so the Embedded arm fails loudly. Routing it diff --git a/crates/omnigraph-cli/src/main.rs b/crates/omnigraph-cli/src/main.rs index 3d7b7785..11b90adc 100644 --- a/crates/omnigraph-cli/src/main.rs +++ b/crates/omnigraph-cli/src/main.rs @@ -714,6 +714,47 @@ async fn main() -> Result<()> { } } } + BlobCommand::Put { + type_name, + id, + prop, + uri, + branch, + file, + if_match, + json, + } => { + let client = client::GraphClient::resolve_with_policy( + cli.server.as_deref(), + cli.graph.as_deref(), + uri, + cli.as_actor.as_deref(), + cli.profile.as_deref(), + cli.store.as_deref(), + ) + .await?; + let branch = resolve_branch(branch, None, "main"); + let bytes = match &file { + Some(path) => std::fs::read(path)?, + None => { + let mut buffer = Vec::new(); + std::io::Read::read_to_end(&mut std::io::stdin().lock(), &mut buffer)?; + buffer + } + }; + let output = client + .blob_put(&branch, &type_name, &id, &prop, bytes, if_match.as_deref()) + .await?; + if json { + print_json(&output)?; + } else { + println!( + "uploaded {} bytes to {} {} {} on {}", + output.size, output.type_name, output.id, output.prop, output.branch + ); + println!("etag: {}", output.etag); + } + } BlobCommand::Stat { type_name, id, diff --git a/crates/omnigraph-cli/src/planes.rs b/crates/omnigraph-cli/src/planes.rs index 713755cf..8b5dd10b 100644 --- a/crates/omnigraph-cli/src/planes.rs +++ b/crates/omnigraph-cli/src/planes.rs @@ -152,6 +152,7 @@ pub(crate) fn command_label(cmd: &Command) -> &'static str { Command::Branch { .. } => "branch", Command::Blob { command } => match command { BlobCommand::Get { .. } => "blob get", + BlobCommand::Put { .. } => "blob put", BlobCommand::Stat { .. } => "blob stat", }, Command::Schema { command } => match command { diff --git a/crates/omnigraph-cli/tests/cli_data.rs b/crates/omnigraph-cli/tests/cli_data.rs index 18235d50..e5802c67 100644 --- a/crates/omnigraph-cli/tests/cli_data.rs +++ b/crates/omnigraph-cli/tests/cli_data.rs @@ -2561,3 +2561,89 @@ fn blob_branch_and_snapshot_conflict() { "clap must reject the exclusive pair: {stderr}" ); } + +#[test] +fn blob_put_uploads_file_and_round_trips() { + let (temp, graph, _) = blob_graph(); + let payload_path = temp.path().join("payload.bin"); + fs::write(&payload_path, b"uploaded via file").unwrap(); + + let output = output_success( + blob_args(&mut cli(), "put", "readme", &graph) + .arg("--file") + .arg(&payload_path) + .arg("--json"), + ); + let payload = parse_stdout_json(&output); + assert_eq!(payload["type"], "Document"); + assert_eq!(payload["id"], "readme"); + assert_eq!(payload["prop"], "content"); + assert_eq!(payload["branch"], "main"); + assert_eq!(payload["size"], b"uploaded via file".len() as u64); + let put_etag = payload["etag"].as_str().unwrap().to_string(); + + let get = output_success(blob_args(&mut cli(), "get", "readme", &graph)); + assert_eq!(&get.stdout[..], b"uploaded via file"); + // The stat validator equals the one the put reported. + let stat = output_success(blob_args(&mut cli(), "stat", "readme", &graph).arg("--json")); + assert_eq!(parse_stdout_json(&stat)["etag"].as_str().unwrap(), put_etag); +} + +#[test] +fn blob_put_uploads_from_stdin() { + let (_temp, graph, _) = blob_graph(); + let output = output_success( + blob_args(&mut cli(), "put", "readme", &graph).write_stdin("streamed from stdin"), + ); + assert!(String::from_utf8_lossy(&output.stdout).contains("uploaded 19 bytes")); + let get = output_success(blob_args(&mut cli(), "get", "readme", &graph)); + assert_eq!(&get.stdout[..], b"streamed from stdin"); +} + +#[test] +fn blob_put_is_update_only() { + let (_temp, graph, _) = blob_graph(); + let output = output_failure(blob_args(&mut cli(), "put", "absent", &graph).write_stdin("x")); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!( + stderr.contains("no Document with id 'absent'"), + "update-only put must fail a missing row: {stderr}" + ); +} + +#[test] +fn blob_put_if_match_mismatch_fails_with_current_validator() { + let (_temp, graph, _) = blob_graph(); + let stat = output_success(blob_args(&mut cli(), "stat", "readme", &graph).arg("--json")); + let fresh = parse_stdout_json(&stat)["etag"].as_str().unwrap().to_string(); + + // Fresh validator lands. + output_success( + blob_args(&mut cli(), "put", "readme", &graph) + .arg("--if-match") + .arg(&fresh) + .write_stdin("v2"), + ); + + // The now-stale validator fails, naming the current one. + let current = parse_stdout_json(&output_success( + blob_args(&mut cli(), "stat", "readme", &graph).arg("--json"), + ))["etag"] + .as_str() + .unwrap() + .to_string(); + let output = output_failure( + blob_args(&mut cli(), "put", "readme", &graph) + .arg("--if-match") + .arg(&fresh) + .write_stdin("must not land"), + ); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("blob precondition failed"), "{stderr}"); + assert!( + stderr.contains(¤t), + "the error must carry the current validator: {stderr}" + ); + let get = output_success(blob_args(&mut cli(), "get", "readme", &graph)); + assert_eq!(&get.stdout[..], b"v2", "a failed precondition must write nothing"); +} diff --git a/crates/omnigraph-cli/tests/parity_matrix.rs b/crates/omnigraph-cli/tests/parity_matrix.rs index 2babaf87..9f337857 100644 --- a/crates/omnigraph-cli/tests/parity_matrix.rs +++ b/crates/omnigraph-cli/tests/parity_matrix.rs @@ -476,3 +476,118 @@ fn parity_blob_get_offset_beyond_size() { ); } } + +// ---- blob put rows ---- + +#[test] +fn parity_blob_put() { + let (p, _) = blob_parity(); + let payload = p._temp.path().join("upload.bin"); + std::fs::write(&payload, b"parity upload payload").unwrap(); + let (l, r) = p.run(&[ + "blob", + "put", + "Document", + "readme", + "content", + "--file", + payload.to_str().unwrap(), + "--json", + ]); + // Scrubbed-JSON equality; etag and size are NOT in the volatile + // allowlist, so the identity-derived validator must match exactly across + // arms (both start from the byte-mirrored graph and apply the identical + // operation). + assert_parity("blob put", &l, &r); + let (l_get, r_get) = p.run(&["blob", "get", "Document", "readme", "content"]); + assert_blob_get_parity( + "blob get after put", + b"parity upload payload", + &l_get, + &r_get, + ); +} + +#[test] +fn parity_blob_put_if_match_stale() { + let (p, _) = blob_parity(); + let payload = p._temp.path().join("upload.bin"); + std::fs::write(&payload, b"must not land").unwrap(); + // A well-formed strong validator that matches nothing on either arm. + let (l, r) = p.run(&[ + "blob", + "put", + "Document", + "readme", + "content", + "--file", + payload.to_str().unwrap(), + "--if-match", + "\"00000000000000000000000000000000\"", + ]); + assert_eq!( + l.status.code(), + r.status.code(), + "blob put stale if-match: exit codes diverge\nlocal {l:?}\nremote {r:?}" + ); + assert!(!l.status.success(), "a stale validator must fail: {l:?}"); + for (arm, output) in [("local", &l), ("remote", &r)] { + let stderr = String::from_utf8_lossy(&output.stderr); + assert!( + stderr.contains("blob precondition failed"), + "{arm} arm must surface the shared typed precondition error: {stderr}" + ); + } + // Nothing landed on either arm. + let (l_get, r_get) = p.run(&["blob", "get", "Document", "readme", "content"]); + assert_blob_get_parity( + "blob get after refused put", + b"Hello World", + &l_get, + &r_get, + ); +} + +#[test] +fn parity_blob_put_missing_row() { + let (p, _) = blob_parity(); + let payload = p._temp.path().join("upload.bin"); + std::fs::write(&payload, b"x").unwrap(); + let (l, r) = p.run(&[ + "blob", + "put", + "Document", + "no-such-row", + "content", + "--file", + payload.to_str().unwrap(), + ]); + assert_eq!( + l.status.code(), + r.status.code(), + "blob put missing row: exit codes diverge\nlocal {l:?}\nremote {r:?}" + ); + assert!(!l.status.success(), "update-only put must fail: {l:?}"); +} + +#[test] +fn parity_blob_put_oversize() { + let (p, _) = blob_parity(); + let payload = p._temp.path().join("oversize.bin"); + std::fs::write(&payload, vec![0_u8; 32 * 1024 * 1024 + 1]).unwrap(); + let (l, r) = p.run(&[ + "blob", + "put", + "Document", + "readme", + "content", + "--file", + payload.to_str().unwrap(), + ]); + assert_eq!( + l.status.code(), + r.status.code(), + "blob put oversize: exit codes diverge\nlocal {l:?}\nremote {r:?}" + ); + assert!(!l.status.success(), "an oversize payload must refuse: {l:?}"); +} diff --git a/crates/omnigraph-server/src/handlers.rs b/crates/omnigraph-server/src/handlers.rs index f3ac111d..0b1a3a6f 100644 --- a/crates/omnigraph-server/src/handlers.rs +++ b/crates/omnigraph-server/src/handlers.rs @@ -925,6 +925,124 @@ pub(crate) async fn server_blob_get( Ok((status, response_headers, body).into_response()) } +#[utoipa::path( + put, + path = "/blob", + tag = "data", + operation_id = "putBlob", + params(BlobQuery), + request_body( + content = Vec, + description = "Raw blob payload bytes (no base64, no JSON envelope). \ + Maximum 32 MiB minus the row's other materialized \ + content.", + content_type = "application/octet-stream" + ), + responses( + (status = 200, description = "Payload stored; `ETag` carries the committed cell's validator", body = BlobPutOutput), + (status = 400, description = "Property is not a Blob, snapshot addressing on a write, or payload over the batch budget", body = ErrorOutput), + (status = 401, description = "Unauthorized", body = ErrorOutput), + (status = 403, description = "Forbidden", body = ErrorOutput), + (status = 404, description = "Unknown type or id — the upload is update-only and never creates a row", body = ErrorOutput), + (status = 412, description = "`If-Match` did not match the cell's current validator; `ETag` carries it"), + (status = 413, description = "Request body exceeds the 32 MiB ingest limit"), + (status = 429, description = "Per-actor workload admission rejected the request; `Retry-After` is set", body = ErrorOutput), + ), + security(("bearer_token" = [])), +)] +/// Replace one row's Blob property with the raw request body. +/// +/// Update-only: the row must exist (404 otherwise) — create it with +/// `/load` or `/mutate` first, then attach the payload. `branch` defaults +/// to `main`; snapshot addressing is read-only and rejected here. An +/// optional `If-Match` (strong comparison, entity-tag list or `*`) makes +/// the write conditional on the cell's current validator; without it the +/// PUT is last-writer-wins. The ETag is table-version-granular, so ANY +/// write to the same table invalidates held validators — a stale +/// `If-Match` after an unrelated same-table write is intended +/// strong-validator behavior. A row whose OTHER blob columns hold +/// external references has them inlined by this write (the documented +/// update behavior) and charged against the 32 MiB batch budget. +pub(crate) async fn server_blob_put( + State(state): State, + Extension(handle): Extension>, + actor: Option>, + headers: HeaderMap, + Query(query): Query, + body: Bytes, +) -> std::result::Result { + if query.snapshot.is_some() { + return Err(ApiError::bad_request( + "blob upload targets a branch; snapshot addressing is read-only", + )); + } + let branch = query.branch.clone().unwrap_or_else(|| "main".to_string()); + let actor = actor.as_ref().map(|Extension(actor)| actor); + authorize_request( + actor, + handle.policy.as_deref(), + PolicyRequest { + action: PolicyAction::Change, + branch: Some(branch.clone()), + target_branch: None, + }, + )?; + let actor_arc = actor + .map(|actor| Arc::clone(&actor.actor_id)) + .unwrap_or_else(|| Arc::::from("anonymous")); + let _admission = state + .workload + .try_admit(&actor_arc, body.len() as u64) + .map_err(ApiError::from_workload_reject)?; + // RFC 9110 If-Match parsing happens HERE, at the transport boundary; the + // engine receives only the typed, transport-neutral precondition. + let precondition = headers + .get(IF_MATCH) + .and_then(|value| value.to_str().ok()) + .map(parse_if_match); + let actor_id = actor.map(|actor| actor.actor_id.as_ref()); + let outcome = handle + .engine + .write_blob_at_as( + &branch, + &query.type_name, + &query.id, + &query.prop, + body.to_vec(), + precondition, + actor_id, + ) + .await + .map_err(ApiError::from_omni)?; + match outcome { + omnigraph::db::BlobWriteOutcome::Written { etag, size } => { + let payload = BlobPutOutput { + type_name: query.type_name, + id: query.id, + prop: query.prop, + branch, + size, + etag: etag.clone(), + actor_id: actor_id.map(str::to_string), + }; + Ok(( + StatusCode::OK, + [(ETAG, HeaderValue::from_str(&etag).expect("hex etag"))], + Json(payload), + ) + .into_response()) + } + omnigraph::db::BlobWriteOutcome::PreconditionFailed { current_etag } => Ok(( + StatusCode::PRECONDITION_FAILED, + [( + ETAG, + HeaderValue::from_str(¤t_etag).expect("hex etag"), + )], + ) + .into_response()), + } +} + /// Shared implementation behind `POST /mutate` (canonical) and /// `POST /change` (deprecated alias). Returns the bare `ChangeOutput`; /// each route handler wraps it (the alias also attaches Deprecation diff --git a/crates/omnigraph-server/src/lib.rs b/crates/omnigraph-server/src/lib.rs index 9f3a4c44..cde91eb9 100644 --- a/crates/omnigraph-server/src/lib.rs +++ b/crates/omnigraph-server/src/lib.rs @@ -26,7 +26,8 @@ use std::path::PathBuf; use std::sync::Arc; use api::{ - BlobQuery, BranchCreateOutput, BranchCreateRequest, BranchDeleteOutput, BranchListOutput, blob_etag, + BlobPutOutput, BlobQuery, BranchCreateOutput, BranchCreateRequest, BranchDeleteOutput, + BranchListOutput, blob_etag, parse_if_match, BranchMergeOutput, BranchMergeRequest, ChangeOutput, ChangeRequest, CommitListOutput, CommitListQuery, ErrorCode, ErrorOutput, ExportRequest, GraphInfo, GraphListResponse, HealthOutput, IngestOutput, IngestRequest, InvokeStoredQueryRequest, InvokeStoredQueryResponse, @@ -40,7 +41,7 @@ use axum::extract::DefaultBodyLimit; use axum::extract::{Extension, OriginalUri, Path, Query, Request, State}; use axum::http::header::{ ACCEPT_RANGES, AUTHORIZATION, CONTENT_LENGTH, CONTENT_RANGE, CONTENT_TYPE, ETAG, HeaderName, - HeaderValue, IF_NONE_MATCH, IF_RANGE, LOCATION, RANGE, + HeaderValue, IF_MATCH, IF_NONE_MATCH, IF_RANGE, LOCATION, RANGE, }; use axum::http::{HeaderMap, Method, StatusCode}; use axum::middleware::{self, Next}; @@ -99,6 +100,7 @@ fn hash_bearer_token(token: &str) -> BearerTokenHash { handlers::server_query, handlers::server_export, handlers::server_blob_get, + handlers::server_blob_put, #[allow(deprecated)] handlers::server_change, handlers::server_mutate, handlers::server_list_queries, @@ -1115,7 +1117,15 @@ pub fn build_app(state: AppState) -> Router { // `{graph_id}` in the URI path). let per_graph_protected = Router::new() .route("/snapshot", get(server_snapshot)) - .route("/blob", get(server_blob_get)) + .route( + "/blob", + get(server_blob_get) + .put(server_blob_put) + // Raw payload bytes on PUT: the same 32 MiB ceiling the JSON + // load routes carry, but with zero base64/envelope overhead. + // GET/HEAD carry no body, so the raised limit is inert there. + .layer(DefaultBodyLimit::max(INGEST_REQUEST_BODY_LIMIT_BYTES)), + ) .route("/export", post(server_export)) // /read and /change are kept indefinitely for back-compat; // their handlers carry #[deprecated] so the OpenAPI operation is diff --git a/crates/omnigraph-server/tests/data_routes.rs b/crates/omnigraph-server/tests/data_routes.rs index 22b936c5..14832ed2 100644 --- a/crates/omnigraph-server/tests/data_routes.rs +++ b/crates/omnigraph-server/tests/data_routes.rs @@ -2091,3 +2091,158 @@ async fn blob_route_rejects_branch_and_snapshot() { error.error ); } + +async fn blob_put( + app: &axum::Router, + uri: &str, + body: Vec, + extra: &[(&str, &str)], +) -> axum::response::Response { + let mut builder = Request::builder().uri(g(uri)).method(Method::PUT); + for (name, value) in extra { + builder = builder.header(*name, *value); + } + app.clone() + .oneshot(builder.body(Body::from(body)).unwrap()) + .await + .unwrap() +} + +#[tokio::test(flavor = "multi_thread")] +async fn blob_route_put_round_trips_raw_bytes() { + let (_temp, app) = blob_app().await; + let response = blob_put( + &app, + "/blob?type=Document&id=readme&prop=content", + b"uploaded raw".to_vec(), + &[], + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + let header_etag = response + .headers() + .get("etag") + .unwrap() + .to_str() + .unwrap() + .to_string(); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let payload: Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(payload["type"], "Document"); + assert_eq!(payload["id"], "readme"); + assert_eq!(payload["prop"], "content"); + assert_eq!(payload["branch"], "main"); + assert_eq!(payload["size"], b"uploaded raw".len() as u64); + assert_eq!(payload["etag"].as_str().unwrap(), header_etag); + + // A follow-up GET serves the uploaded bytes under the SAME validator. + let get = blob_get(&app, "/blob?type=Document&id=readme&prop=content", &[]).await; + assert_eq!(get.status(), StatusCode::OK); + assert_eq!(get.headers().get("etag").unwrap().to_str().unwrap(), header_etag); + let bytes = to_bytes(get.into_body(), usize::MAX).await.unwrap(); + assert_eq!(&bytes[..], b"uploaded raw"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn blob_route_put_is_update_only() { + let (_temp, app) = blob_app().await; + let response = blob_put( + &app, + "/blob?type=Document&id=absent&prop=content", + b"x".to_vec(), + &[], + ) + .await; + assert_eq!(response.status(), StatusCode::NOT_FOUND); +} + +#[tokio::test(flavor = "multi_thread")] +async fn blob_route_put_if_match_flow() { + let (_temp, app) = blob_app().await; + let head = blob_get(&app, "/blob?type=Document&id=readme&prop=content", &[]).await; + let fresh = head + .headers() + .get("etag") + .unwrap() + .to_str() + .unwrap() + .to_string(); + + // Fresh validator (inside a list, per RFC 9110) → the write lands. + let listed = format!("\"stale\", {fresh}"); + let response = blob_put( + &app, + "/blob?type=Document&id=readme&prop=content", + b"v2".to_vec(), + &[("if-match", listed.as_str())], + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + let v2_etag = response + .headers() + .get("etag") + .unwrap() + .to_str() + .unwrap() + .to_string(); + assert_ne!(v2_etag, fresh); + + // The now-stale validator → 412 carrying the current one; cell untouched. + let response = blob_put( + &app, + "/blob?type=Document&id=readme&prop=content", + b"must not land".to_vec(), + &[("if-match", fresh.as_str())], + ) + .await; + assert_eq!(response.status(), StatusCode::PRECONDITION_FAILED); + assert_eq!( + response.headers().get("etag").unwrap().to_str().unwrap(), + v2_etag, + "412 must report the current validator" + ); + let get = blob_get(&app, "/blob?type=Document&id=readme&prop=content", &[]).await; + let bytes = to_bytes(get.into_body(), usize::MAX).await.unwrap(); + assert_eq!(&bytes[..], b"v2"); + + // `*` succeeds against an existing row. + let response = blob_put( + &app, + "/blob?type=Document&id=readme&prop=content", + b"v3".to_vec(), + &[("if-match", "*")], + ) + .await; + assert_eq!(response.status(), StatusCode::OK); +} + +#[tokio::test(flavor = "multi_thread")] +async fn blob_route_put_rejects_snapshot_addressing() { + let (_temp, app) = blob_app().await; + let response = blob_put( + &app, + "/blob?type=Document&id=readme&prop=content&snapshot=someid", + b"x".to_vec(), + &[], + ) + .await; + assert_eq!(response.status(), StatusCode::BAD_REQUEST); +} + +#[tokio::test(flavor = "multi_thread")] +async fn blob_route_put_oversize_body_is_rejected() { + let (_temp, app) = blob_app().await; + let oversize = vec![0_u8; 32 * 1024 * 1024 + 1]; + let response = blob_put( + &app, + "/blob?type=Document&id=readme&prop=content", + oversize, + &[], + ) + .await; + assert_eq!( + response.status(), + StatusCode::PAYLOAD_TOO_LARGE, + "the ingest body limit must reject before the engine sees the payload" + ); +} diff --git a/crates/omnigraph/src/db/mod.rs b/crates/omnigraph/src/db/mod.rs index 1120ff17..6bd6fbde 100644 --- a/crates/omnigraph/src/db/mod.rs +++ b/crates/omnigraph/src/db/mod.rs @@ -12,7 +12,8 @@ pub use manifest::{Snapshot, SnapshotScanner, SnapshotTable, SubTableEntry, SubT pub(crate) use omnigraph::ensure_public_branch_ref; pub(crate) use omnigraph::{DeferredTableFork, WriteAuthorityToken, WriteTxn}; pub use omnigraph::{ - BlobContent, BlobRead, BlobReader, BlobVersionTag, CleanupPolicyOptions, InitOptions, + BlobContent, BlobPrecondition, BlobRead, BlobReader, BlobVersionTag, BlobWriteOutcome, + CleanupPolicyOptions, InitOptions, blob_etag, MergeOutcome, Omnigraph, OpenMode, PendingIndex, RepairAction, RepairClassification, RepairOptions, RepairStats, SchemaApplyOptions, SchemaApplyResult, SkipReason, TableCleanupStats, TableOptimizeStats, TableRepairStats, diff --git a/crates/omnigraph/src/db/omnigraph.rs b/crates/omnigraph/src/db/omnigraph.rs index d7484070..91ebe519 100644 --- a/crates/omnigraph/src/db/omnigraph.rs +++ b/crates/omnigraph/src/db/omnigraph.rs @@ -41,7 +41,10 @@ mod schema_apply; mod stream_enrollment; mod table_ops; -pub use blob_read::{BlobContent, BlobRead, BlobReader, BlobVersionTag}; +pub use blob_read::{ + BlobContent, BlobPrecondition, BlobRead, BlobReader, BlobVersionTag, BlobWriteOutcome, + blob_etag, +}; pub use optimize::{CleanupPolicyOptions, SkipReason, TableCleanupStats, TableOptimizeStats}; pub use repair::{ RepairAction, RepairClassification, RepairOptions, RepairStats, TableRepairStats, diff --git a/crates/omnigraph/src/db/omnigraph/blob_read.rs b/crates/omnigraph/src/db/omnigraph/blob_read.rs index 5d1456ac..7a796e54 100644 --- a/crates/omnigraph/src/db/omnigraph/blob_read.rs +++ b/crates/omnigraph/src/db/omnigraph/blob_read.rs @@ -34,6 +34,61 @@ pub struct BlobVersionTag { pub row_id: u64, } +/// Strong validator string for one blob cell: an opaque digest of the +/// identity tuple `(stable_table_id, incarnation_id, table_version, row_id, +/// property)`. The stable-identity pair (not the mutable `table_key`) makes +/// the tag rename-stable and drop/re-add ABA-safe (invariant 8). Lives in +/// the engine so [`Omnigraph::write_blob_at`] can evaluate preconditions +/// against its pinned write base; `omnigraph-api-types` re-exports it for +/// the HTTP/CLI boundary. +pub fn blob_etag(tag: &BlobVersionTag, property: &str) -> String { + use sha2::{Digest, Sha256}; + let mut hasher = Sha256::new(); + hasher.update(tag.stable_table_id.to_le_bytes()); + hasher.update(tag.table_incarnation_id.to_le_bytes()); + hasher.update(tag.table_version.to_le_bytes()); + hasher.update(tag.row_id.to_le_bytes()); + hasher.update(property.as_bytes()); + let digest = hasher.finalize(); + let mut out = String::with_capacity(34); + out.push('"'); + for byte in &digest[..16] { + out.push_str(&format!("{byte:02x}")); + } + out.push('"'); + out +} + +/// Transport-neutral precondition for a blob write. The engine compares +/// opaque strong validator tokens only — RFC 9110 `If-Match` header parsing +/// (lists, `*`, weak-tag rejection) belongs to the HTTP/CLI boundary +/// (invariant 11). +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum BlobPrecondition { + /// `If-Match: *` — succeed as long as the cell currently exists. + AnyExisting, + /// Succeed iff the cell's current validator equals one of these opaque + /// strong tokens (each in the exact quoted form the read side serves). + Tags(Vec), +} + +/// Outcome of [`Omnigraph::write_blob_at`]. +/// +/// `PreconditionFailed` is an Ok VARIANT, not an error: a stale validator is +/// an expected, well-formed answer (HTTP 412), not a fault — and the closed +/// `OmniError` taxonomy stays untouched. Ignoring the variant and treating +/// any `Ok` as success is a correctness bug; hence `#[must_use]`. +#[must_use] +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum BlobWriteOutcome { + /// The cell was replaced and the graph commit published. `etag` is the + /// validator a follow-up read of the committed head serves. + Written { etag: String, size: u64 }, + /// The supplied precondition did not match the cell's current validator; + /// nothing was staged or published. + PreconditionFailed { current_etag: String }, +} + /// How one blob cell's content is reachable. pub enum BlobContent { /// Internal storage (inline / packed / dedicated): bytes stream through diff --git a/crates/omnigraph/src/exec/mutation.rs b/crates/omnigraph/src/exec/mutation.rs index e3706c5a..aa8b8e34 100644 --- a/crates/omnigraph/src/exec/mutation.rs +++ b/crates/omnigraph/src/exec/mutation.rs @@ -1,7 +1,7 @@ use super::*; use super::query::literal_to_sql; -use crate::storage_layer::PendingScanBudget; +use crate::storage_layer::{KEYED_WRITE_MAX_BYTES, PendingScanBudget}; // ─── Mutation helpers ──────────────────────────────────────────────────────── @@ -813,89 +813,361 @@ impl Omnigraph { Err(e) => Err(e), Ok(total) if staging.is_empty() => Ok(total), Ok(total) => { - self.validate_staged_mutation(&staging, &txn).await?; - let staged = staging.stage_all(self, requested.as_deref()).await?; - crate::failpoints::maybe_fail( - crate::failpoints::names::MUTATION_POST_STAGE_PRE_EFFECT_GATE, - )?; - let lineage_intent = self - .new_lineage_intent_for_branch(requested.as_deref(), actor_id) + self.publish_staged_mutation(staging, &txn, requested.as_deref(), actor_id) .await?; - // `_queue_guards` holds the root-shared schema gate, branch - // effect gate, and sorted table gates acquired by `commit_all`. - // They remain held through manifest publication, covering the - // complete same-process sidecar/effect lifetime. They are a - // local serialization aid; the exact publisher precondition and - // durable v3 recovery plan remain the correctness authorities. - let super::staging::CommittedMutation { - updates, - expected_versions, - sidecar_handle, - guards: _queue_guards, - } = staged - .commit_all( - self, - requested.as_deref(), - crate::db::manifest::SidecarKind::Mutation, - actor_id, - &txn, - &lineage_intent, - ) - .await?; - // Failpoint for the confirmed-effects → publisher boundary: - // table HEADs have advanced but graph visibility has not. The - // v3 sidecar already contains exact transaction identities, - // immutable manifest delta, and fixed lineage/rollback outcomes. - // Any failure from here is `RecoveryRequired`; synchronous heal - // or a read-write open converges the recorded outcome. See - // `tests/failpoints.rs::recovery_rolls_forward_after_finalize_publisher_failure`. - crate::failpoints::maybe_fail( - crate::failpoints::names::MUTATION_POST_FINALIZE_PRE_PUBLISHER, - )?; - let publish_result = self - .commit_updates_on_branch_with_expected( - requested.as_deref(), - &updates, - &expected_versions, - actor_id, - &txn, - lineage_intent, - ) - .await; - if let Err(err) = publish_result { - // A sidecar exists iff at least one table effect was - // committed. Lineage-only / zero-row mutations have no - // physical residual to recover, so preserve their original - // publish error (notably ReadSetChanged) and let the normal - // retry/409 path handle it. - return match sidecar_handle.as_ref() { - Some(handle) => Err(OmniError::recovery_required( - handle.operation_id.clone(), - err.to_string(), - )), - None => Err(err), - }; + Ok(total) + } + } + } + + /// Shared Mutation-v9 publish tail: end-of-query validation, exact + /// per-table staging, lineage intent, recovery-armed table commits, the + /// single manifest publication, `RecoveryRequired` wrapping, and + /// best-effort sidecar cleanup. One body serves both `mutate_one_attempt` + /// and the blob-upload writer so the orchestration cannot drift and every + /// durable call keeps its single registered site in this file + /// (`forbidden_apis` exact-count contract). + async fn publish_staged_mutation( + &self, + staging: MutationStaging, + txn: &crate::db::WriteTxn, + requested: Option<&str>, + actor_id: Option<&str>, + ) -> Result<()> { + self.validate_staged_mutation(&staging, txn).await?; + let staged = staging.stage_all(self, requested).await?; + crate::failpoints::maybe_fail( + crate::failpoints::names::MUTATION_POST_STAGE_PRE_EFFECT_GATE, + )?; + let lineage_intent = self + .new_lineage_intent_for_branch(requested, actor_id) + .await?; + // `_queue_guards` holds the root-shared schema gate, branch + // effect gate, and sorted table gates acquired by `commit_all`. + // They remain held through manifest publication, covering the + // complete same-process sidecar/effect lifetime. They are a + // local serialization aid; the exact publisher precondition and + // durable v3 recovery plan remain the correctness authorities. + let super::staging::CommittedMutation { + updates, + expected_versions, + sidecar_handle, + guards: _queue_guards, + } = staged + .commit_all( + self, + requested, + crate::db::manifest::SidecarKind::Mutation, + actor_id, + txn, + &lineage_intent, + ) + .await?; + // Failpoint for the confirmed-effects → publisher boundary: + // table HEADs have advanced but graph visibility has not. The + // v3 sidecar already contains exact transaction identities, + // immutable manifest delta, and fixed lineage/rollback outcomes. + // Any failure from here is `RecoveryRequired`; synchronous heal + // or a read-write open converges the recorded outcome. See + // `tests/failpoints.rs::recovery_rolls_forward_after_finalize_publisher_failure`. + crate::failpoints::maybe_fail( + crate::failpoints::names::MUTATION_POST_FINALIZE_PRE_PUBLISHER, + )?; + let publish_result = self + .commit_updates_on_branch_with_expected( + requested, + &updates, + &expected_versions, + actor_id, + txn, + lineage_intent, + ) + .await; + if let Err(err) = publish_result { + // A sidecar exists iff at least one table effect was + // committed. Lineage-only / zero-row mutations have no + // physical residual to recover, so preserve their original + // publish error (notably ReadSetChanged) and let the normal + // retry/409 path handle it. + return match sidecar_handle.as_ref() { + Some(handle) => Err(OmniError::recovery_required( + handle.operation_id.clone(), + err.to_string(), + )), + None => Err(err), + }; + } + if let Some(handle) = sidecar_handle { + // Best-effort cleanup: the manifest publish already + // succeeded, so the user's mutation is durable. A failed + // delete leaves a fixed, idempotent v3 outcome for the next + // synchronous heal or read-write open to audit and remove. + // Failing the user here would report an error for a write + // that already landed. + if let Err(err) = + crate::db::manifest::delete_sidecar(&handle, self.storage_adapter()).await + { + tracing::warn!( + error = %err, + operation_id = handle.operation_id.as_str(), + "recovery sidecar cleanup failed; the next open's recovery sweep will resolve it" + ); + } + } + Ok(()) + } + + /// Replace one existing node row's Blob cell with raw bytes. + /// + /// Update-only: a missing row is a typed not-found, never an implicit + /// insert. See [`Self::write_blob_at_as`] for the actor-carrying variant + /// and the full contract. + pub async fn write_blob_at( + &self, + branch: &str, + type_name: &str, + id: &str, + property: &str, + bytes: Vec, + precondition: Option, + ) -> Result { + self.write_blob_at_as(branch, type_name, id, property, bytes, precondition, None) + .await + } + + /// [`Self::write_blob_at`] with an explicit actor for policy enforcement + /// and lineage attribution. + /// + /// The write composes the Mutation-v9 protocol exactly — one pinned + /// write authority, one staged keyed upsert of the full row, one + /// recovery-armed commit, one manifest publication — so it is + /// indistinguishable from a one-row mutation to recovery and to the + /// commit graph. The payload replaces only `property`; every other cell + /// of the row is carried through by the same committed-cell + /// materialization an update uses (an external reference on ANOTHER blob + /// column is inlined, the documented update behavior, and charged + /// against the 32 MiB batch budget — the effective max payload is + /// therefore 32 MiB minus the row's other materialized content). + /// + /// `precondition` compares opaque strong validator tokens against the + /// cell's current state under the SAME pinned base the write prepares + /// from; a reprepare re-derives it, so the check is atomic by + /// construction. A mismatch returns + /// [`crate::db::BlobWriteOutcome::PreconditionFailed`] with nothing + /// staged. [`crate::db::BlobPrecondition::AnyExisting`] requires the + /// target ROW to exist (a null cell still accepts — attaching first + /// content to an existing row is the primary use). + /// + /// Node types only, mirroring `read_blob_at` (edge blob cells are an + /// explicit known gap on both surfaces). + #[allow(clippy::too_many_arguments)] + pub async fn write_blob_at_as( + &self, + branch: &str, + type_name: &str, + id: &str, + property: &str, + bytes: Vec, + precondition: Option, + actor_id: Option<&str>, + ) -> Result { + self.enforce( + omnigraph_policy::PolicyAction::Change, + &omnigraph_policy::ResourceScope::Branch(branch.to_string()), + actor_id, + )?; + const MAX_PRE_EFFECT_REPREPARES: usize = 32; + for attempt in 0..=MAX_PRE_EFFECT_REPREPARES { + match self + .write_blob_one_attempt(branch, type_name, id, property, &bytes, &precondition, actor_id) + .await + { + // Unlike a .gq update (whose replay would be a semantic + // rebase of a stale read-modify-write plan), a blob PUT is a + // declarative full-cell replacement: its entire read set — + // the row scan and the precondition evaluation — is + // re-derived per attempt, so replay after a pre-effect + // authority change is a fresh execution. This is what makes + // last-writer-wins absorb unrelated-write conflicts AND + // keeps a supplied precondition evaluated against the fresh + // base on every attempt. + Err(err) + if err.is_read_set_changed() && attempt < MAX_PRE_EFFECT_REPREPARES => + { + tracing::debug!( + attempt = attempt + 1, + branch, + "prepared blob write authority changed before effects; repreparing" + ); + self.refresh().await?; } - if let Some(handle) = sidecar_handle { - // Best-effort cleanup: the manifest publish already - // succeeded, so the user's mutation is durable. A failed - // delete leaves a fixed, idempotent v3 outcome for the next - // synchronous heal or read-write open to audit and remove. - // Failing the user here would report an error for a write - // that already landed. - if let Err(err) = - crate::db::manifest::delete_sidecar(&handle, self.storage_adapter()).await - { - tracing::warn!( - error = %err, - operation_id = handle.operation_id.as_str(), - "recovery sidecar cleanup failed; the next open's recovery sweep will resolve it" - ); - } + result => return result, + } + } + unreachable!("bounded blob write retry loop always returns") + } + + #[allow(clippy::too_many_arguments)] + async fn write_blob_one_attempt( + &self, + branch: &str, + type_name: &str, + id: &str, + property: &str, + bytes: &[u8], + precondition: &Option, + actor_id: Option<&str>, + ) -> Result { + let requested = Self::normalize_branch_name(branch)?; + if let Some(name) = requested.as_deref() { + crate::db::ensure_public_branch_ref(name, "write_blob")?; + } + if bytes.len() as u64 > KEYED_WRITE_MAX_BYTES { + return Err(OmniError::resource_limit( + format!("blob payload bytes for {type_name}.{property}"), + KEYED_WRITE_MAX_BYTES, + bytes.len() as u64, + )); + } + self.heal_pending_recovery_sidecars_for_write(&[requested.as_deref()]) + .await?; + let txn = self.open_write_txn(requested.as_deref()).await?; + let node_type = txn.catalog.node_types.get(type_name).ok_or_else(|| { + OmniError::manifest_not_found(format!("unknown node type '{}'", type_name)) + })?; + if !node_type.blob_properties.contains(property) { + return Err(OmniError::manifest(format!( + "property '{}' on type '{}' is not a Blob", + property, type_name + ))); + } + let schema = node_type.arrow_schema.clone(); + let table_key = format!("node:{}", type_name); + + // Evaluate a supplied precondition against the SAME pinned base this + // attempt prepares from (txn.base), so a reprepare re-evaluates it. + if let Some(precondition) = precondition { + let entry = txn.base.entry(&table_key).ok_or_else(|| { + OmniError::manifest_not_found(format!( + "no table for type '{}' in this snapshot", + type_name + )) + })?; + let handle = self + .storage() + .open_snapshot_at_table(&txn.base, &table_key) + .await?; + let row_id = self + .storage() + .first_row_id_for_id(&handle, id) + .await? + .ok_or_else(|| { + OmniError::manifest_not_found(format!( + "no {} with id '{}' found", + type_name, id + )) + })?; + let current_tag = crate::db::BlobVersionTag { + stable_table_id: entry.identity.stable_table_id, + table_incarnation_id: entry.identity.table_incarnation_id, + table_version: entry.table_version, + row_id, + }; + let current_etag = crate::db::blob_etag(¤t_tag, property); + let matches = match precondition { + crate::db::BlobPrecondition::AnyExisting => true, + crate::db::BlobPrecondition::Tags(tags) => { + tags.iter().any(|tag| tag == ¤t_etag) } - Ok(total) + }; + if !matches { + return Ok(crate::db::BlobWriteOutcome::PreconditionFailed { current_etag }); } } + + let mut staging = MutationStaging::default(); + let (handle, _full_path, _table_branch) = open_table_for_mutation( + self, + &mut staging, + requested.as_deref(), + &table_key, + crate::db::MutationOpKind::Update, + Some(&txn), + ) + .await?; + let ds = handle.expect("strict Update op always opens its dataset"); + + // Single-row committed scan with every OTHER blob column + // materialized to the logical schema; the target cell is null-filled + // (its old payload would only be discarded) and swapped below. + let pred_sql = format!("id = '{}'", id.replace('\'', "''")); + let (pending_rows, pending_bytes) = staging.pending_resource_usage(&table_key)?; + let scan_budget = PendingScanBudget::new(&table_key, pending_rows, pending_bytes); + let batches = self + .storage() + .scan_with_pending_materialized_blobs_skipping( + &ds, + &[], + None, + Some(&pred_sql), + Some("id"), + scan_budget, + Some(property), + ) + .await?; + if batches.is_empty() || batches.iter().all(|b| b.num_rows() == 0) { + return Err(OmniError::manifest_not_found(format!( + "no {} with id '{}' found", + type_name, id + ))); + } + let matched = concat_match_batches_to_schema(&schema, batches)?; + if matched.num_rows() != 1 { + return Err(OmniError::manifest_internal(format!( + "exact-id scan for {}.'{}' matched {} rows; ids are unique by contract", + type_name, + id, + matched.num_rows() + ))); + } + + // Swap ONLY the target column for the uploaded payload. + let mut builder = BlobArrayBuilder::new(1); + builder + .push_bytes(bytes.to_vec()) + .map_err(|e| OmniError::Lance(e.to_string()))?; + let uploaded = builder.finish().map_err(|e| OmniError::Lance(e.to_string()))?; + let column_index = matched.schema().index_of(property).map_err(|e| { + OmniError::manifest_internal(format!( + "blob column '{}' missing from matched batch: {}", + property, e + )) + })?; + let mut columns = matched.columns().to_vec(); + columns[column_index] = uploaded; + let updated = arrow_array::RecordBatch::try_new(matched.schema(), columns) + .map_err(|e| OmniError::Lance(e.to_string()))?; + + let size = bytes.len() as u64; + staging.append_batch(&table_key, updated.schema(), PendingMode::Upsert, updated)?; + self.publish_staged_mutation(staging, &txn, requested.as_deref(), actor_id) + .await?; + + // The committed validator: RewriteRows moved the row into new + // fragments with a fresh stable row id at a new table version, so + // re-derive the tag exactly the way a follow-up read does. + let committed = self + .read_blob_at( + crate::db::ReadTarget::branch(requested.as_deref().unwrap_or("main")), + type_name, + id, + property, + ) + .await?; + Ok(crate::db::BlobWriteOutcome::Written { + etag: crate::db::blob_etag(&committed.version_tag, property), + size, + }) } /// Lower + validate a named mutation query into its IR. diff --git a/crates/omnigraph/src/storage_layer.rs b/crates/omnigraph/src/storage_layer.rs index a14f393d..b6ac7064 100644 --- a/crates/omnigraph/src/storage_layer.rs +++ b/crates/omnigraph/src/storage_layer.rs @@ -521,6 +521,21 @@ pub trait TableStorage: sealed::Sealed + Send + Sync + Debug { budget: PendingScanBudget, ) -> Result>; + /// [`Self::scan_with_pending_materialized_blobs`] with one blob column + /// null-filled instead of materialized — the blob-upload writer replaces + /// that exact cell, so its old payload is neither read nor charged. + #[allow(clippy::too_many_arguments)] + async fn scan_with_pending_materialized_blobs_skipping( + &self, + snapshot: &SnapshotHandle, + pending: &[RecordBatch], + pending_schema: Option, + filter: Option<&str>, + key_column: Option<&str>, + budget: PendingScanBudget, + skip_blob_column: Option<&str>, + ) -> Result>; + async fn first_row_id_for_filter( &self, snapshot: &SnapshotHandle, @@ -949,7 +964,30 @@ impl TableStorage for TableStore { key_column: Option<&str>, budget: PendingScanBudget, ) -> Result> { - TableStore::scan_with_pending_materialized_blobs( + TableStorage::scan_with_pending_materialized_blobs_skipping( + self, + snapshot, + pending, + pending_schema, + filter, + key_column, + budget, + None, + ) + .await + } + + async fn scan_with_pending_materialized_blobs_skipping( + &self, + snapshot: &SnapshotHandle, + pending: &[RecordBatch], + pending_schema: Option, + filter: Option<&str>, + key_column: Option<&str>, + budget: PendingScanBudget, + skip_blob_column: Option<&str>, + ) -> Result> { + TableStore::scan_with_pending_materialized_blobs_skipping( self, snapshot.dataset(), pending, @@ -957,6 +995,7 @@ impl TableStorage for TableStore { filter, key_column, budget, + skip_blob_column, ) .await } diff --git a/crates/omnigraph/src/table_store.rs b/crates/omnigraph/src/table_store.rs index 2722f194..63b765e4 100644 --- a/crates/omnigraph/src/table_store.rs +++ b/crates/omnigraph/src/table_store.rs @@ -683,7 +683,7 @@ impl TableStore { .copied() .collect::>(); - Self::materialize_blob_batch_with_row_ids(ds, batch, &row_ids, max_blob_bytes).await + Self::materialize_blob_batch_with_row_ids(ds, batch, &row_ids, max_blob_bytes, None).await } /// Rebuild the blob columns in `batch` using explicit stable row ids. @@ -699,6 +699,7 @@ impl TableStore { batch: RecordBatch, row_ids: &[u64], max_blob_bytes: Option, + skip_blob_column: Option<&str>, ) -> Result { if batch.num_rows() != row_ids.len() { return Err(OmniError::Lance(format!( @@ -718,6 +719,20 @@ impl TableStore { OmniError::Lance(format!("batch missing column '{}'", field.name())) })?; if lance_field.is_blob() { + // A caller replacing this exact cell provides its own payload: + // reading the old content would only be discarded, and charging + // it against the byte budget would make near-cap cells + // un-overwritable. Null-fill and let the caller swap it. + if skip_blob_column == Some(field.name().as_str()) { + let mut builder = BlobArrayBuilder::new(batch.num_rows()); + for _ in 0..batch.num_rows() { + builder + .push_null() + .map_err(|e| OmniError::Lance(e.to_string()))?; + } + columns.push(builder.finish().map_err(|e| OmniError::Lance(e.to_string()))?); + continue; + } let descriptions = column .as_any() @@ -1973,6 +1988,7 @@ impl TableStore { descriptors, &[row_id], Some(KEYED_WRITE_MAX_BYTES), + None, ) .await .map_err(|error| { @@ -2716,6 +2732,33 @@ impl TableStore { filter: Option<&str>, key_column: Option<&str>, budget: PendingScanBudget, + ) -> Result> { + self.scan_with_pending_materialized_blobs_skipping( + committed_ds, + pending_batches, + pending_schema, + filter, + key_column, + budget, + None, + ) + .await + } + + /// [`Self::scan_with_pending_materialized_blobs`] with one blob column + /// left null-filled instead of materialized — the blob-upload write path + /// replaces that exact cell, so its old payload is neither read nor + /// charged against the scan's byte budget. + #[allow(clippy::too_many_arguments)] + pub async fn scan_with_pending_materialized_blobs_skipping( + &self, + committed_ds: &Dataset, + pending_batches: &[RecordBatch], + pending_schema: Option, + filter: Option<&str>, + key_column: Option<&str>, + budget: PendingScanBudget, + skip_blob_column: Option<&str>, ) -> Result> { let blob_columns = committed_ds .schema() @@ -2831,6 +2874,7 @@ impl TableStore { descriptors, &row_ids, Some(payload_budget), + skip_blob_column, ) .await { diff --git a/crates/omnigraph/tests/end_to_end.rs b/crates/omnigraph/tests/end_to_end.rs index 321802a3..a8e879a0 100644 --- a/crates/omnigraph/tests/end_to_end.rs +++ b/crates/omnigraph/tests/end_to_end.rs @@ -2235,3 +2235,216 @@ async fn full_flow_optimize_then_query_update_and_reopen() { "Alice's post-optimize age update must persist across reopen" ); } + +// ─── Blob upload (`write_blob_at`) ─────────────────────────────────────────── + +async fn blob_upload_graph() -> (tempfile::TempDir, Omnigraph) { + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + let mut db = Omnigraph::init(uri, BLOB_SCHEMA).await.unwrap(); + load_jsonl( + &mut db, + r#"{"type": "Document", "data": {"title": "readme", "content": "base64:SGVsbG8gV29ybGQ="}}"#, + LoadMode::Overwrite, + ) + .await + .unwrap(); + (dir, db) +} + +#[tokio::test] +async fn write_blob_at_replaces_cell_and_round_trips() { + use omnigraph::db::BlobWriteOutcome; + let (_dir, db) = blob_upload_graph().await; + + let outcome = db + .write_blob_at("main", "Document", "readme", "content", b"NewBytes!".to_vec(), None) + .await + .unwrap(); + let BlobWriteOutcome::Written { etag, size } = outcome else { + panic!("unconditional write must land: {outcome:?}"); + }; + assert_eq!(size, b"NewBytes!".len() as u64); + assert_eq!( + read_blob_bytes(&db, "Document", "readme", "content").await, + b"NewBytes!" + ); + // The returned validator is exactly what a follow-up read serves. + let read = db + .read_blob_at("main", "Document", "readme", "content") + .await + .unwrap(); + assert_eq!(etag, omnigraph::db::blob_etag(&read.version_tag, "content")); +} + +#[tokio::test] +async fn write_blob_at_missing_row_is_not_found() { + let (_dir, db) = blob_upload_graph().await; + let err = db + .write_blob_at("main", "Document", "absent", "content", b"x".to_vec(), None) + .await + .map(|_| ()) + .unwrap_err(); + assert!( + err.to_string().contains("no Document with id 'absent'"), + "update-only write must 404 a missing row: {err}" + ); +} + +#[tokio::test] +async fn write_blob_at_non_blob_property_is_typed() { + let (_dir, db) = blob_upload_graph().await; + let err = db + .write_blob_at("main", "Document", "readme", "title", b"x".to_vec(), None) + .await + .map(|_| ()) + .unwrap_err(); + assert!(err.to_string().contains("is not a Blob"), "{err}"); +} + +#[tokio::test] +async fn write_blob_at_oversize_is_refused_pre_effect() { + let (_dir, db) = blob_upload_graph().await; + let before = db + .snapshot_of(omnigraph::db::ReadTarget::branch("main")) + .await + .unwrap() + .version(); + let oversize = vec![0_u8; (32 * 1024 * 1024) + 1]; + let err = db + .write_blob_at("main", "Document", "readme", "content", oversize, None) + .await + .map(|_| ()) + .unwrap_err(); + assert!( + matches!(err, omnigraph::error::OmniError::ResourceLimitExceeded { .. }), + "oversize payload must be a typed resource refusal: {err:?}" + ); + let after = db + .snapshot_of(omnigraph::db::ReadTarget::branch("main")) + .await + .unwrap() + .version(); + assert_eq!(before, after, "refusal must precede any durable effect"); +} + +#[tokio::test] +async fn write_blob_at_if_match_flow() { + use omnigraph::db::{BlobPrecondition, BlobWriteOutcome, blob_etag}; + let (_dir, db) = blob_upload_graph().await; + + let current = db + .read_blob_at("main", "Document", "readme", "content") + .await + .unwrap(); + let fresh_tag = blob_etag(¤t.version_tag, "content"); + + // Matching strong tag → the write lands. + let outcome = db + .write_blob_at( + "main", + "Document", + "readme", + "content", + b"v2".to_vec(), + Some(BlobPrecondition::Tags(vec![fresh_tag.clone()])), + ) + .await + .unwrap(); + let BlobWriteOutcome::Written { etag: v2_etag, .. } = outcome else { + panic!("fresh If-Match must succeed: {outcome:?}"); + }; + assert_ne!(v2_etag, fresh_tag, "a landed write mints a new validator"); + + // The now-stale tag → PreconditionFailed carrying the current validator, + // and the cell is untouched. + let outcome = db + .write_blob_at( + "main", + "Document", + "readme", + "content", + b"v3-should-not-land".to_vec(), + Some(BlobPrecondition::Tags(vec![fresh_tag])), + ) + .await + .unwrap(); + let BlobWriteOutcome::PreconditionFailed { current_etag } = outcome else { + panic!("stale If-Match must fail the precondition: {outcome:?}"); + }; + assert_eq!(current_etag, v2_etag); + assert_eq!(read_blob_bytes(&db, "Document", "readme", "content").await, b"v2"); + + // `*` requires only that the row exists. + let outcome = db + .write_blob_at( + "main", + "Document", + "readme", + "content", + b"v3".to_vec(), + Some(BlobPrecondition::AnyExisting), + ) + .await + .unwrap(); + assert!(matches!(outcome, BlobWriteOutcome::Written { .. })); + assert_eq!(read_blob_bytes(&db, "Document", "readme", "content").await, b"v3"); +} + +#[tokio::test] +async fn write_blob_at_replaces_external_reference_cell() { + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + let blob_dir = tempfile::tempdir().unwrap(); + let blob_path = blob_dir.path().join("external.bin"); + std::fs::write(&blob_path, b"external-bytes").unwrap(); + let mut db = Omnigraph::init(uri, BLOB_SCHEMA).await.unwrap(); + let seed = format!( + r#"{{"type": "Document", "data": {{"title": "ext", "content": "file://{}"}}}}"#, + blob_path.display() + ); + load_jsonl(&mut db, &seed, LoadMode::Overwrite).await.unwrap(); + // The upload provides its own payload — the old external object is + // irrelevant, so delete it to prove nothing resolves it. + std::fs::remove_file(&blob_path).unwrap(); + + let outcome = db + .write_blob_at("main", "Document", "ext", "content", b"now inline".to_vec(), None) + .await + .unwrap(); + assert!(matches!(outcome, omnigraph::db::BlobWriteOutcome::Written { .. })); + assert_eq!(read_blob_bytes(&db, "Document", "ext", "content").await, b"now inline"); +} + +#[tokio::test] +async fn write_blob_at_carries_other_blob_columns() { + const TWO_BLOB_SCHEMA: &str = r#" +node Asset { + name: String @key + content: Blob? + thumb: Blob? +} +"#; + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + let mut db = Omnigraph::init(uri, TWO_BLOB_SCHEMA).await.unwrap(); + load_jsonl( + &mut db, + r#"{"type": "Asset", "data": {"name": "a1", "content": "base64:Ym9keQ==", "thumb": "base64:dGh1bWI="}}"#, + LoadMode::Overwrite, + ) + .await + .unwrap(); + + let outcome = db + .write_blob_at("main", "Asset", "a1", "content", b"replaced".to_vec(), None) + .await + .unwrap(); + assert!(matches!(outcome, omnigraph::db::BlobWriteOutcome::Written { .. })); + assert_eq!(read_blob_bytes(&db, "Asset", "a1", "content").await, b"replaced"); + assert_eq!( + read_blob_bytes(&db, "Asset", "a1", "thumb").await, + b"thumb", + "the untouched blob column must be carried through unchanged" + ); +} diff --git a/crates/omnigraph/tests/forbidden_apis.rs b/crates/omnigraph/tests/forbidden_apis.rs index 24c40ecf..79a31567 100644 --- a/crates/omnigraph/tests/forbidden_apis.rs +++ b/crates/omnigraph/tests/forbidden_apis.rs @@ -222,7 +222,7 @@ macro_rules! write_surfaces { write_surfaces! { "db/omnigraph.rs" => WriteProtocol::Bootstrap => ["init", "init_with_options"], "db/omnigraph.rs" => WriteProtocol::RecoveryExecutor => ["open", "open_with_storage", "refresh"], - "exec/mutation.rs" => MUTATION_V9 => ["mutate", "mutate_as"], + "exec/mutation.rs" => MUTATION_V9 => ["mutate", "mutate_as", "write_blob_at", "write_blob_at_as"], "loader/mod.rs" => LOAD_V9 => ["load_jsonl", "load_jsonl_file", "load", "load_file"], "loader/mod.rs" => WriteProtocol::Composed("optional branch create, then Load v9") => ["load_as", "load_file_as"], "loader/mod.rs" => WriteProtocol::Composed("branch create when absent, then Load v9 alias") => ["ingest", "ingest_as", "ingest_file", "ingest_file_as"], @@ -507,6 +507,7 @@ gateway_surfaces! { "ensure_expected_version", "scan", "scan_with_row_id", "scan_batches", "scan_batches_for_rewrite", "count_rows", "count_rows_with_staged", "scan_with_staged", "scan_with_pending", "scan_with_pending_materialized_blobs", + "scan_with_pending_materialized_blobs_skipping", "first_row_id_for_filter", "first_row_id_for_id", "table_state", "has_btree_index", "has_fts_index", "has_vector_index", "root_uri", "dataset_uri", "scan_stream", "scan_stream_bounded", "scan_stream_for_rewrite_bounded", @@ -542,7 +543,8 @@ gateway_surfaces! { "scan_stream_with", "scan", "scan_with", "scan_edges_by_endpoint", "key_column_index_coverage", "has_unindexed_fragments", "count_rows", "dataset_version", "table_state", "scan_with_staged", "scan_with_pending", - "scan_with_pending_materialized_blobs", "count_rows_with_staged", + "scan_with_pending_materialized_blobs", + "scan_with_pending_materialized_blobs_skipping", "count_rows_with_staged", "has_btree_index", "has_btree_index_on", "has_fts_index", "has_fts_index_on", "has_vector_index", "has_vector_index_on", "first_row_id_for_filter", "first_row_id_for_id", diff --git a/docs/user/cli/reference.md b/docs/user/cli/reference.md index 16c878ab..03e3bcb9 100644 --- a/docs/user/cli/reference.md +++ b/docs/user/cli/reference.md @@ -17,7 +17,7 @@ Top-level command families and subcommands. Graph-targeting commands accept a po | `snapshot` | print current snapshot (per-table version + row count) | | `export` | dump to JSONL on stdout (`--type T`, `--table K` filters) | | `branch create \| list \| delete \| merge` | branching ops. `merge --delete-branch` deletes the source branch after a successful merge (its own `branch_delete` policy check; a refusal is a stderr warning, not a failure — see [merge](../branching/merge.md)) | -| `blob get \| stat ` | one row's `Blob` property. `get` streams the payload bytes to stdout (or `--out `); `--offset`/`--length` request a bounded range; `--branch`/`--snapshot` (mutually exclusive) pick the read target. `stat` prints kind/size/etag metadata with zero payload I/O (`--json` for the object). External references are never fetched: `stat` reports the stored URI, `get` fails with a typed error carrying it. Graph addressed via `--uri `, `--store`, `--server` + `--graph`, or `--profile` | +| `blob get \| put \| stat ` | one row's `Blob` property. `get` streams the payload bytes to stdout (or `--out `); `--offset`/`--length` request a bounded range; `--branch`/`--snapshot` (mutually exclusive) pick the read target. `put` uploads raw bytes from `--file ` or stdin into an **existing** row's cell (update-only — create the row first; ≤32 MiB minus row overhead; `--if-match ` makes it conditional, failing with the current validator on mismatch; `--as` records the actor). `stat` prints kind/size/etag metadata with zero payload I/O (`--json` for the object). External references are never fetched: `stat` reports the stored URI, `get` fails with a typed error carrying it; `put` over one replaces it with the inline payload. Graph addressed via `--uri `, `--store`, `--server` + `--graph`, or `--profile` | | `commit list \| show` | inspect commit graph | | `schema plan \| apply \| show (alias: get)` | migrations. `apply` refuses a cluster-managed graph (one whose storage is inside a cluster) and points at `cluster apply` — those graphs evolve through the cluster ledger, not a direct apply | | `lint` (alias: `check`) | offline / graph-backed query validation. Replaces `query lint` / `query check`, which are kept as deprecated argv-level shims that print a one-line warning and rewrite to `omnigraph lint` | @@ -35,7 +35,7 @@ Top-level command families and subcommands. Graph-targeting commands accept a po Every command declares the **capability** it needs — what it requires to reach a graph — which determines the addressing flags that apply: -- **`any`** — `query`, `mutate`, `load`, `ingest`, `branch *`, `blob get`/`blob stat`, `snapshot`, `export`, `commit *`, `schema show`, `schema apply`. Run against a graph **served (via a server) or embedded (direct against a store)**: accept a positional `file://`/`s3://` URI, `--server ` (+ `--graph ` for multi-graph servers), `--store `, or `--profile `. A remote server is addressed with `--server` — a positional `http(s)://` URI does **not** dispatch to one. +- **`any`** — `query`, `mutate`, `load`, `ingest`, `branch *`, `blob *`, `snapshot`, `export`, `commit *`, `schema show`, `schema apply`. Run against a graph **served (via a server) or embedded (direct against a store)**: accept a positional `file://`/`s3://` URI, `--server ` (+ `--graph ` for multi-graph servers), `--store `, or `--profile `. A remote server is addressed with `--server` — a positional `http(s)://` URI does **not** dispatch to one. - **`served`** — `graphs list`. Requires a server (accepts `--server` / `--profile`). - **`direct`** — `init`, `optimize`, `repair`, `cleanup`, `schema plan`, `lint`. Need **direct storage access** (`file://` / `s3://`), never through a server. They accept a positional `URI`, but **not** `--server`, and a remote (`http(s)://`) URI is rejected. `optimize` / `repair` / `cleanup` additionally accept **`--cluster --graph `** (`--cluster` is a cluster directory or storage-root URI, named via `clusters:` in `~/.omnigraph/config.yaml` or a literal root), which resolves the graph's storage URI from the served cluster state (so you needn't know the `/graphs/.omni` layout). `--graph` is the one graph selector across all scopes — on these three verbs it picks the cluster graph; on the other `direct` verbs it does not apply. - **`control`** — `cluster *` via `--config `; `policy *` and `queries *` via `--cluster ` or a cluster profile. diff --git a/docs/user/operations/server.md b/docs/user/operations/server.md index 247badfa..1785fd3c 100644 --- a/docs/user/operations/server.md +++ b/docs/user/operations/server.md @@ -53,6 +53,7 @@ graph id from the cluster's applied revision: | GET | `/openapi.json` | none | — (strips security if auth disabled; emits the nested cluster paths with `cluster_` operation-id prefix) | | GET | `/graphs/{id}/snapshot?branch=` | bearer + `read` | snapshot of branch | | GET/HEAD | `/graphs/{id}/blob?type=&id=&prop=[&branch=\|&snapshot=]` | bearer + `read` | stream one row's Blob property (see "Blob content" below) | +| PUT | `/graphs/{id}/blob?type=&id=&prop=[&branch=]` | bearer + `change` | upload raw bytes into one existing row's Blob cell (see "Blob upload" below) | | POST | `/graphs/{id}/query` | bearer + `read` | inline read query (canonical; clean field names `query`/`name`; mutations → 400) | | POST | `/graphs/{id}/read` | bearer + `read` | **deprecated** alias of `/query` (legacy field names `query_source`/`query_name`, byte-stable response; carries `Deprecation: true` + `Link: ; rel="successor-version"`) | | POST | `/graphs/{id}/export` | bearer + `export` | NDJSON stream | @@ -183,8 +184,9 @@ omitting both reads `main` — the same target semantics as `/read`. Contract: `302 Found` with the stored absolute URI in `Location` for both `GET` and `HEAD`. The server never resolves, proxies, or validates the external location at read time — the reference itself is the answer. External blob - *sizes* are recorded once at load time (the load fails loudly if the - referenced object is unreadable at ingest). + sizes are unknown by design (the descriptor records none; recording them + would perturb the table's blob schema), but the load still fails loudly if + the referenced object is unreadable at ingest. - **`Content-Type`** is sniffed from at most the first 512 bytes (`HEAD` and `GET` sniff identically, so their headers agree) and falls back to `application/octet-stream`. Lance stores no media type; treat the sniff as @@ -193,6 +195,37 @@ omitting both reads `main` — the same target semantics as `/read`. Contract: non-Blob property is `400`. Read-only and not admission-gated, like every read endpoint. +### Blob upload (`PUT /graphs/{id}/blob`) + +`PUT /graphs/{id}/blob?type=&id=&prop=[&branch=]` +with the **raw payload bytes as the request body** (no base64, no JSON +envelope) replaces that one cell as a single atomic graph commit through the +same write path as `/mutate`. Requires bearer + `change`; admission-gated +like every write endpoint (`429` + `Retry-After`). Contract: + +- **Update-only.** The row must exist — a missing type or id is `404`, never + an implicit insert. Create the row (metadata) with `/load` or `/mutate` + first, then attach the payload. `branch` defaults to `main`; `snapshot` + addressing is read-only and rejected with `400`. +- **Response**: `200` with an `ETag` header carrying the committed cell's + validator (identical to what a follow-up `GET` serves) and a JSON body + `{type, id, prop, branch, size, etag, actor_id}`. +- **Conditional writes**: an optional `If-Match` header (entity-tag list or + `*`, strong comparison — a weak `W/` tag never matches) makes the write + conditional on the cell's current validator; a mismatch answers `412` with + the current validator in `ETag` and nothing written. Without `If-Match` + the PUT is last-writer-wins. The ETag is **table-version-granular**: any + write to the same table invalidates every held validator, so a stale + `If-Match` after an unrelated same-table write is intended + strong-validator behavior, not a bug. `If-Match: *` requires only that + the target row exists (a currently-null cell accepts — attaching first + content is the primary use). +- **Limits**: the request body is capped at 32 MiB; the effective maximum + payload is 32 MiB minus the row's other materialized content, refused with + a typed error. A row whose OTHER blob columns hold external references has + them inlined by this write (the same documented behavior as any update) + and charged against that budget. + ## Error model Uniform diff --git a/openapi.json b/openapi.json index 1faa7e06..c0b54376 100644 --- a/openapi.json +++ b/openapi.json @@ -205,6 +205,165 @@ "bearer_token": [] } ] + }, + "put": { + "tags": [ + "data" + ], + "summary": "Replace one row's Blob property with the raw request body.", + "description": "Update-only: the row must exist (404 otherwise) — create it with\n`/load` or `/mutate` first, then attach the payload. `branch` defaults\nto `main`; snapshot addressing is read-only and rejected here. An\noptional `If-Match` (strong comparison, entity-tag list or `*`) makes\nthe write conditional on the cell's current validator; without it the\nPUT is last-writer-wins. The ETag is table-version-granular, so ANY\nwrite to the same table invalidates held validators — a stale\n`If-Match` after an unrelated same-table write is intended\nstrong-validator behavior. A row whose OTHER blob columns hold\nexternal references has them inlined by this write (the documented\nupdate behavior) and charged against the 32 MiB batch budget.", + "operationId": "cluster_putBlob", + "parameters": [ + { + "name": "graph_id", + "in": "path", + "description": "Graph id to route the request to.", + "required": true, + "schema": { + "type": "string" + } + }, + { + "name": "type", + "in": "query", + "description": "Node type name.", + "required": true, + "schema": { + "type": "string" + } + }, + { + "name": "id", + "in": "query", + "description": "Logical row id (the `@key` value).", + "required": true, + "schema": { + "type": "string" + } + }, + { + "name": "prop", + "in": "query", + "description": "Blob property name.", + "required": true, + "schema": { + "type": "string" + } + }, + { + "name": "branch", + "in": "query", + "description": "Branch to read from. Mutually exclusive with `snapshot`. Defaults to `main`.", + "required": false, + "schema": { + "type": [ + "string", + "null" + ] + } + }, + { + "name": "snapshot", + "in": "query", + "description": "Snapshot id to read from. Mutually exclusive with `branch`.", + "required": false, + "schema": { + "type": [ + "string", + "null" + ] + } + } + ], + "requestBody": { + "description": "Raw blob payload bytes (no base64, no JSON envelope). Maximum 32 MiB minus the row's other materialized content.", + "content": { + "application/octet-stream": { + "schema": { + "type": "array", + "items": { + "type": "integer", + "format": "int32", + "minimum": 0 + } + } + } + }, + "required": true + }, + "responses": { + "200": { + "description": "Payload stored; `ETag` carries the committed cell's validator", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/BlobPutOutput" + } + } + } + }, + "400": { + "description": "Property is not a Blob, snapshot addressing on a write, or payload over the batch budget", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorOutput" + } + } + } + }, + "401": { + "description": "Unauthorized", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorOutput" + } + } + } + }, + "403": { + "description": "Forbidden", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorOutput" + } + } + } + }, + "404": { + "description": "Unknown type or id — the upload is update-only and never creates a row", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorOutput" + } + } + } + }, + "412": { + "description": "`If-Match` did not match the cell's current validator; `ETag` carries it" + }, + "413": { + "description": "Request body exceeds the 32 MiB ingest limit" + }, + "429": { + "description": "Per-actor workload admission rejected the request; `Retry-After` is set", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorOutput" + } + } + } + } + }, + "security": [ + { + "bearer_token": [] + } + ] } }, "/graphs/{graph_id}/branches": { @@ -1898,6 +2057,48 @@ }, "components": { "schemas": { + "BlobPutOutput": { + "type": "object", + "description": "Response body for `PUT /blob` (raw-bytes blob upload). Also printed\nverbatim by the CLI's `blob put` so both arms emit the identical object.", + "required": [ + "type", + "id", + "prop", + "branch", + "size", + "etag" + ], + "properties": { + "actor_id": { + "type": [ + "string", + "null" + ] + }, + "branch": { + "type": "string" + }, + "etag": { + "type": "string", + "description": "Strong quoted validator of the committed cell — identical to what a\nfollow-up `GET /blob` serves." + }, + "id": { + "type": "string" + }, + "prop": { + "type": "string" + }, + "size": { + "type": "integer", + "format": "int64", + "description": "Stored payload size in bytes.", + "minimum": 0 + }, + "type": { + "type": "string" + } + } + }, "BranchCreateOutput": { "type": "object", "required": [