diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 97505bfd9..e0906a02f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -752,6 +752,7 @@ jobs: system_remote::remote_read_preserves_projection_order_in_json_and_csv \ system_remote::remote_branch_create_list_merge_flow \ system_remote::remote_merge_delivery_loss_never_replays_committed_effect \ + system_remote::remote_blob_write_delivery_loss_never_replays_committed_effect \ system_remote::remote_merge_statement_delivery_loss_never_replays_committed_effect \ system_remote::remote_deployment_delivery_loss_preserves_owned_completion \ system_remote::remote_branch_delete_removes_branch \ diff --git a/Cargo.lock b/Cargo.lock index f40cc5497..e35f5af81 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5645,6 +5645,7 @@ dependencies = [ "assert_cmd", "async-trait", "base64 0.22.1", + "bytes", "clap", "color-eyre", "futures", diff --git a/changelog.d/blob-cell-put-clear.added.md b/changelog.d/blob-cell-put-clear.added.md index 5360f61f0..0b1df1ed8 100644 --- a/changelog.d/blob-cell-put-clear.added.md +++ b/changelog.d/blob-cell-put-clear.added.md @@ -9,5 +9,5 @@ is the one a read at that commit reports. One put holds at most the session's `write_max_bytes` (32 MiB by default), inclusive; a larger one fails before any table is opened with resource `Blob write payload bytes`. - The HTTP server exposes them as `PUT` and `DELETE /blob`; the CLI does not - expose them yet. + The HTTP server exposes them as `PUT` and `DELETE /blob`, and the CLI as + `blob put` and `blob clear`. diff --git a/changelog.d/blob-cli-put-clear.added.md b/changelog.d/blob-cli-put-clear.added.md new file mode 100644 index 000000000..ae198b4cd --- /dev/null +++ b/changelog.d/blob-cli-put-clear.added.md @@ -0,0 +1,11 @@ +- `omnigraph blob put ENTITY TYPE ID PROPERTY` stores the raw bytes of + `--file PATH` or stdin in one Blob cell, and `omnigraph blob clear` sets a + nullable cell to null, on a `--store` graph or a served one. Both take + `--branch`, `--if-match TAG` (`*` or entity tags) and `--json`, and print the + receipt the server returns: the selector, branch, state, size and ETag of a + managed value, the actor and the exact commit, which is `null` for a clear of + a cell that was already null. Input over 32 MiB and a malformed `--if-match` + are refused before the graph is addressed, and an interactive stdin is + refused rather than waited on. `--as` names the actor of a `--store` write. + A served Blob precondition failure exits 4, as a graph-commit one does; an + embedded one exits 1. A write is sent once and never retried. diff --git a/crates/omnigraph-cli/Cargo.toml b/crates/omnigraph-cli/Cargo.toml index 9e43c28a8..ddf7f93ba 100644 --- a/crates/omnigraph-cli/Cargo.toml +++ b/crates/omnigraph-cli/Cargo.toml @@ -13,6 +13,7 @@ name = "omnigraph" path = "src/main.rs" [dependencies] +bytes = { workspace = true } omnigraph = { package = "omnigraph-engine", path = "../omnigraph", version = "0.13.0" } omnigraph-compiler = { path = "../omnigraph-compiler", version = "0.13.0" } omnigraph-api-types = { path = "../omnigraph-api-types", version = "0.13.0" } diff --git a/crates/omnigraph-cli/src/blob_cli.rs b/crates/omnigraph-cli/src/blob_cli.rs index f290a9e57..b5c793bf2 100644 --- a/crates/omnigraph-cli/src/blob_cli.rs +++ b/crates/omnigraph-cli/src/blob_cli.rs @@ -1,20 +1,20 @@ -//! CLI-only shaping for graph-level Blob reads. +//! CLI-only shaping for graph-level Blob reads and writes. //! //! The engine and server own descriptor interpretation, snapshot fencing, and //! bounded payload reads. This module keeps the command line's selector, //! range, and error vocabulary identical across the embedded and served arms. use std::fs::File; -use std::io::{self, Write}; +use std::io::{self, IsTerminal, Read, Write}; use std::ops::Range; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use color_eyre::Report; use color_eyre::eyre::{Result, eyre}; use omnigraph::db::{ReadTarget, SnapshotId}; use omnigraph::error::{ManifestErrorKind, OmniError}; -use omnigraph::{BlobCell, EntityKind, ExternalBlobRef}; -use omnigraph_api_types::{BlobEntityKind, BlobReadQuery}; +use omnigraph::{BlobCell, BlobPrecondition, EntityKind, ExternalBlobRef}; +use omnigraph_api_types::{BlobEntityKind, BlobReadQuery, BlobWriteQuery}; use reqwest::StatusCode; use reqwest::header::{CONTENT_LENGTH, ETAG, HeaderMap, HeaderName}; @@ -164,6 +164,104 @@ pub(crate) fn blob_url(base_url: &str, query: &BlobReadQuery) -> Result crate::helpers::remote_url(base_url, &["blob"], ¶ms) } +pub(crate) fn blob_write_query( + entity: BlobEntityKind, + type_name: String, + id: String, + property: String, + branch: Option, +) -> BlobWriteQuery { + BlobWriteQuery { + entity, + r#type: type_name, + id, + property, + branch, + } +} + +pub(crate) fn blob_write_cell(query: &BlobWriteQuery) -> BlobCell { + BlobCell { + entity: match query.entity { + BlobEntityKind::Node => EntityKind::Node, + BlobEntityKind::Edge => EntityKind::Edge, + }, + type_name: query.r#type.clone(), + id: query.id.clone(), + property: query.property.clone(), + } +} + +pub(crate) fn blob_write_url(base_url: &str, query: &BlobWriteQuery) -> Result { + let entity = match query.entity { + BlobEntityKind::Node => "node", + BlobEntityKind::Edge => "edge", + }; + let mut params = vec![ + ("entity", entity), + ("type", query.r#type.as_str()), + ("id", query.id.as_str()), + ("property", query.property.as_str()), + ]; + if let Some(branch) = query.branch.as_deref() { + params.push(("branch", branch)); + } + crate::helpers::remote_url(base_url, &["blob"], ¶ms) +} + +/// Parse `--if-match` with the parser the server applies to the `If-Match` +/// header, so a malformed value is refused before any graph is addressed. +pub(crate) fn parse_if_match(raw: Option<&str>) -> Result> { + omnigraph_api_types::parse_blob_if_match(raw.map(str::as_bytes)).map_err(|error| eyre!(error)) +} + +/// Read `blob put`'s bytes from `--file` or stdin, never both, before any +/// graph is addressed. Input over the put limit is refused with the resource +/// the engine and the server report, without reading past one more byte; +/// an interactive stdin is refused rather than waited on. +pub(crate) fn read_put_input(file: Option<&Path>) -> Result { + let limit = omnigraph::BLOB_WRITE_MAX_BYTES; + let too_large = |actual: u64| -> Report { + OmniError::resource_limit(omnigraph::BLOB_WRITE_PAYLOAD_RESOURCE, limit, actual).into() + }; + let mut buffer = Vec::new(); + match file { + Some(path) => { + let file = File::open(path) + .map_err(|error| eyre!("cannot read --file {}: {error}", path.display()))?; + let length = file + .metadata() + .map_err(|error| eyre!("cannot read --file {}: {error}", path.display()))? + .len(); + if length > limit { + return Err(too_large(length)); + } + buffer.reserve(usize::try_from(length).unwrap_or(0)); + file.take(limit + 1) + .read_to_end(&mut buffer) + .map_err(|error| eyre!("cannot read --file {}: {error}", path.display()))?; + } + None => { + let stdin = io::stdin(); + if stdin.is_terminal() { + return Err(eyre!( + "blob put reads its bytes from stdin; pipe them in or pass --file PATH" + )); + } + stdin + .lock() + .take(limit + 1) + .read_to_end(&mut buffer) + .map_err(|error| eyre!("cannot read stdin: {error}"))?; + } + } + let read = buffer.len() as u64; + if read > limit { + return Err(too_large(read)); + } + Ok(bytes::Bytes::from(buffer)) +} + pub(crate) fn whole_external_uri(reference: &ExternalBlobRef) -> Result<&str> { reference .whole_object_uri() diff --git a/crates/omnigraph-cli/src/cli.rs b/crates/omnigraph-cli/src/cli.rs index 40802a054..328bf67f2 100644 --- a/crates/omnigraph-cli/src/cli.rs +++ b/crates/omnigraph-cli/src/cli.rs @@ -247,7 +247,7 @@ pub(crate) enum Command { #[arg(long = "type")] type_names: Vec, }, - /// Read one logical node or edge Blob cell. + /// Read, replace or clear one logical node or edge Blob cell. Blob { #[command(subcommand)] command: BlobCommand, @@ -516,6 +516,61 @@ pub(crate) enum BlobCommand { #[arg(long)] json: bool, }, + /// Replace one Blob value of an existing node or edge with raw bytes from + /// a file or stdin, as one graph commit. + Put { + /// Logical entity namespace. + #[arg(value_name = "ENTITY")] + entity: BlobEntityArg, + /// Accepted-schema node or edge type. + #[arg(value_name = "TYPE")] + type_name: String, + /// Logical entity id; the entity must exist. + #[arg(value_name = "ID")] + id: String, + /// Blob property name. + #[arg(value_name = "PROPERTY")] + property: String, + /// Write a named branch (defaults to main). + #[arg(long)] + branch: Option, + /// Read the bytes from this file instead of stdin. + #[arg(long, value_name = "PATH")] + file: Option, + /// Write only while the cell's ETag is one of these entity tags, or, + /// as `*`, while the cell holds any value. + #[arg(long, value_name = "TAG")] + if_match: Option, + /// Emit the receipt as JSON. + #[arg(long)] + json: bool, + }, + /// Set one nullable Blob value of an existing node or edge to null, as one + /// graph commit; a cell that is already null publishes nothing. + Clear { + /// Logical entity namespace. + #[arg(value_name = "ENTITY")] + entity: BlobEntityArg, + /// Accepted-schema node or edge type. + #[arg(value_name = "TYPE")] + type_name: String, + /// Logical entity id; the entity must exist. + #[arg(value_name = "ID")] + id: String, + /// Blob property name. + #[arg(value_name = "PROPERTY")] + property: String, + /// Write a named branch (defaults to main). + #[arg(long)] + branch: Option, + /// Write only while the cell's ETag is one of these entity tags, or, + /// as `*`, while the cell holds any value. + #[arg(long, value_name = "TAG")] + if_match: Option, + /// Emit the receipt as JSON. + #[arg(long)] + json: bool, + }, } #[derive(Debug, Subcommand)] diff --git a/crates/omnigraph-cli/src/client.rs b/crates/omnigraph-cli/src/client.rs index 16e5db37b..0b2dcb471 100644 --- a/crates/omnigraph-cli/src/client.rs +++ b/crates/omnigraph-cli/src/client.rs @@ -30,15 +30,16 @@ use omnigraph::db::{Omnigraph, ReadTarget}; use omnigraph::settings::{SessionSettings, SettingId, SettingValue, Source}; use omnigraph::{BLOB_READ_RANGE_MAX_BYTES, BlobContent, Session}; use omnigraph_api_types::{ - BlobReadQuery, BlobStatOutput, BranchCreateOutput, BranchCreateRequest, BranchDeleteOutput, - BranchListOutput, BranchMergeOutcome, BranchMergeOutput, BranchMergeRequest, - BranchOutcomeOutput, ChangeBaselineOutput, ChangeBaselineRecord, ChangeBaselineRequest, - ChangeFeedOutput, ChangeOpOutput, ChangeOutput, ChangeRequest, CommitChangesOutput, - CommitListOutput, CommitOutput, EntityKindOutput, ExportRequest, GraphBatchLoadOutput, - GraphDiscoveryResponse, GraphListResponse, InvokeStoredQueryRequest, QueryRequest, ReadOutput, - SchemaApplyOutput, SchemaOutput, SettingsRequest, SnapshotOutput, branch_list_read_output, - change_baseline_output, change_feed_output, change_scope, commit_changes_output, commit_output, - read_output, schema_apply_output, show_read_output, snapshot_payload, + BlobReadQuery, BlobStatOutput, BlobWriteOutput, BlobWriteQuery, BranchCreateOutput, + BranchCreateRequest, BranchDeleteOutput, BranchListOutput, BranchMergeOutcome, + BranchMergeOutput, BranchMergeRequest, BranchOutcomeOutput, ChangeBaselineOutput, + ChangeBaselineRecord, ChangeBaselineRequest, ChangeFeedOutput, ChangeOpOutput, ChangeOutput, + ChangeRequest, CommitChangesOutput, CommitListOutput, CommitOutput, EntityKindOutput, + ExportRequest, GraphBatchLoadOutput, GraphDiscoveryResponse, GraphListResponse, + InvokeStoredQueryRequest, QueryRequest, ReadOutput, SchemaApplyOutput, SchemaOutput, + SettingsRequest, SnapshotOutput, branch_list_read_output, change_baseline_output, + change_feed_output, change_scope, commit_changes_output, commit_output, read_output, + schema_apply_output, show_read_output, snapshot_payload, }; use omnigraph_compiler::catalog::Catalog; use omnigraph_compiler::query::ast::{BranchWrite, CommitWrite}; @@ -48,13 +49,14 @@ use reqwest::{Method, StatusCode}; use serde_json::Value; use crate::blob_cli::{ - BlobRangeRequest, blob_cell, blob_read_target, blob_url, external_response_headers, - managed_response_headers, map_embedded_blob_error, remote_blob_error, whole_external_uri, + BlobRangeRequest, blob_cell, blob_read_target, blob_url, blob_write_cell, blob_write_url, + external_response_headers, managed_response_headers, map_embedded_blob_error, + remote_blob_error, whole_external_uri, }; use crate::cli::CliLoadMode; use crate::graph_http::{ApiContractError, GraphHttpClient}; use crate::helpers::{ - apply_bearer_token, apply_server_flag, branch_statement_change_request, + PreconditionFailedCli, apply_bearer_token, apply_server_flag, branch_statement_change_request, branch_statement_query_request, is_remote_uri, precondition_failed_cli, query_params_from_json, remote_json, remote_json_bounded, remote_no_content, remote_response_json_bounded, remote_url, resolve_cli_actor, resolve_cli_graph, resolve_remote_bearer_token, resolve_server_flag, @@ -1988,6 +1990,80 @@ impl GraphClient { } } + /// `blob put` (with `bytes`) or `blob clear` (without): one Blob cell + /// write, sent once. The embedded arm calls the engine with the parsed + /// precondition; the served arm sends `if_match` as the `If-Match` header + /// the server parses the same way. Both answer the same receipt. + pub(crate) async fn blob_write( + &self, + query: &BlobWriteQuery, + bytes: Option, + if_match: Option<&str>, + precondition: Option, + ) -> Result { + let branch = query.branch.as_deref().unwrap_or("main"); + match self { + GraphClient::Embedded { uri, actor } => { + let session = Self::open_write_session(uri, &[]).await?; + let cell = blob_write_cell(query); + let outcome = match bytes { + Some(bytes) => { + session + .put_blob_at_as(branch, cell, bytes, precondition, actor.as_deref()) + .await + } + None => { + session + .clear_blob_at_as(branch, cell, precondition, actor.as_deref()) + .await + } + } + .map_err(|error| match error { + // The body a server answers, so `--json` is transport-uniform. + error @ omnigraph::error::OmniError::BlobWritePreconditionFailed { .. } => { + PreconditionFailedCli { + output: omnigraph_server::engine_error_output(error), + http_status: None, + retry_after: None, + } + .into() + } + other => color_eyre::eyre::Report::from(other), + })?; + Ok(omnigraph_api_types::blob_write_output( + query.into(), + branch, + &outcome, + actor.clone(), + )) + } + GraphClient::Remote { + http, + base_url, + token, + response_limit, + } => { + let url = blob_write_url(base_url, query)?; + let request = match bytes { + Some(bytes) => http + .request(Method::PUT, url) + .header(reqwest::header::CONTENT_TYPE, "application/octet-stream") + .body(bytes), + None => http.request(Method::DELETE, url), + }; + let request = apply_bearer_token(request, token.as_deref()); + let request = match if_match { + Some(tags) => request.header(reqwest::header::IF_MATCH, tags), + None => request, + }; + let response = http.send(request).await?; + remote_response_json_bounded(response, token.as_deref(), *response_limit) + .await + .map_err(verified_blob_precondition) + } + } + } + /// `graphs list` — enumerate the graphs a multi-graph server serves /// (`GET /graphs`). Reached only through registry-addressed clients /// (`resolve_registry` / the D7 probe's `registry_client`), which always @@ -2086,6 +2162,34 @@ fn validate_content_range( Ok(()) } +/// A served Blob write's 412 whose only detail is `blob_precondition_failure` +/// is a verified request precondition, treated as a graph-commit one is: the +/// whole command exits 4 when it was the command's only effectful request. +fn verified_blob_precondition(error: color_eyre::eyre::Report) -> color_eyre::eyre::Report { + match error.downcast::() { + Ok(remote) + if remote.status == StatusCode::PRECONDITION_FAILED + && remote.output.blob_precondition_failure.is_some() + && remote.output.code == Some(omnigraph_api_types::ErrorCode::Conflict) + && crate::command_outcome::single_request() + && { + let mut rest = remote.output.clone(); + rest.blob_precondition_failure = None; + crate::command_outcome::is_plain_refusal(&rest) + } => + { + PreconditionFailedCli { + output: remote.output, + http_status: Some(StatusCode::PRECONDITION_FAILED.as_u16()), + retry_after: remote.retry_after, + } + .into() + } + Ok(remote) => remote.into(), + Err(error) => error, + } +} + /// Shared spelling of the change-surface filters across the three verbs; one /// translation (`change_scope`) is used by the embedded arm and the server. pub(crate) struct ChangeFilterArgs<'a> { diff --git a/crates/omnigraph-cli/src/command_outcome.rs b/crates/omnigraph-cli/src/command_outcome.rs index 9ff9754ee..ea7e00164 100644 --- a/crates/omnigraph-cli/src/command_outcome.rs +++ b/crates/omnigraph-cli/src/command_outcome.rs @@ -8,7 +8,7 @@ use omnigraph_api_types::{ErrorCode, ErrorOutput}; use reqwest::StatusCode; use serde::Serialize; -use crate::cli::{BranchCommand, Cli, Command, SchemaCommand}; +use crate::cli::{BlobCommand, BranchCommand, Cli, Command, SchemaCommand}; use crate::graph_http::ApiContractError; use crate::helpers::{PreconditionFailedCli, RemoteErrorCli}; @@ -38,6 +38,9 @@ pub(crate) fn applies(cli: &Cli) -> bool { | Command::Schema { command: SchemaCommand::Apply { .. } } + | Command::Blob { + command: BlobCommand::Put { .. } | BlobCommand::Clear { .. } + } ) } @@ -212,6 +215,7 @@ impl Failure { Action::Recover } else if not_started || output.precondition_failure.is_some() + || output.blob_precondition_failure.is_some() || output.read_set_conflict.is_some() || output.published_dataset_version_conflict.is_some() || output.key_conflict.is_some() @@ -263,6 +267,7 @@ pub(crate) fn is_plain_refusal(output: &ErrorOutput) -> bool { && output.external_blob_source.is_none() && output.recovery_required.is_none() && output.precondition_failure.is_none() + && output.blob_precondition_failure.is_none() && output.change_feed_gap.is_none() && output.change_diff_refusal.is_none() && output.full_text_index_rebuild_required.is_none() diff --git a/crates/omnigraph-cli/src/main.rs b/crates/omnigraph-cli/src/main.rs index f3f356073..391cee102 100644 --- a/crates/omnigraph-cli/src/main.rs +++ b/crates/omnigraph-cli/src/main.rs @@ -1458,6 +1458,72 @@ async fn run(cli: Cli) -> Result<()> { print_blob_stat_human(&output); } } + BlobCommand::Put { + entity, + type_name, + id, + property, + branch, + file, + if_match, + json, + } => { + // A malformed precondition and input over the limit are + // refused before scope resolution, so neither arm starts. + let precondition = blob_cli::parse_if_match(if_match.as_deref())?; + let bytes = blob_cli::read_put_input(file.as_deref())?; + let query = + blob_cli::blob_write_query(entity.into(), type_name, id, property, branch); + let client = client::GraphClient::resolve_with_policy( + capability, + cli.server.as_deref(), + cli.graph.as_deref(), + None, + cli.as_actor.as_deref(), + cli.profile.as_deref(), + cli.store.as_deref(), + ) + .await?; + let output = client + .blob_write(&query, Some(bytes), if_match.as_deref(), precondition) + .await?; + if json { + print_json(&output)?; + } else { + print_blob_write_human(&output); + } + } + BlobCommand::Clear { + entity, + type_name, + id, + property, + branch, + if_match, + json, + } => { + let precondition = blob_cli::parse_if_match(if_match.as_deref())?; + let query = + blob_cli::blob_write_query(entity.into(), type_name, id, property, branch); + let client = client::GraphClient::resolve_with_policy( + capability, + cli.server.as_deref(), + cli.graph.as_deref(), + None, + cli.as_actor.as_deref(), + cli.profile.as_deref(), + cli.store.as_deref(), + ) + .await?; + let output = client + .blob_write(&query, None, if_match.as_deref(), precondition) + .await?; + if json { + print_json(&output)?; + } else { + print_blob_write_human(&output); + } + } }, Command::Query { name, diff --git a/crates/omnigraph-cli/src/output.rs b/crates/omnigraph-cli/src/output.rs index af89c3a48..ab111a941 100644 --- a/crates/omnigraph-cli/src/output.rs +++ b/crates/omnigraph-cli/src/output.rs @@ -711,6 +711,39 @@ pub(crate) fn print_snapshot_human( } } +/// A Blob write receipt: the cell, the state it was left in and the commit +/// that published it, or that nothing was published. +pub(crate) fn print_blob_write_human(output: &omnigraph_api_types::BlobWriteOutput) { + let entity = match output.selector.entity { + omnigraph_api_types::BlobEntityKind::Node => "node", + omnigraph_api_types::BlobEntityKind::Edge => "edge", + }; + println!("entity: {entity}"); + println!("type: {}", output.selector.r#type); + println!("id: {}", output.selector.id); + println!("property: {}", output.selector.property); + println!("branch: {}", output.branch); + match output.kind { + omnigraph_api_types::BlobWriteStateOutput::Managed => { + println!("kind: managed"); + if let Some(size) = output.size { + println!("size: {size}"); + } + if let Some(etag) = output.etag.as_deref() { + println!("etag: {etag}"); + } + } + omnigraph_api_types::BlobWriteStateOutput::Null => println!("kind: null"), + } + if let Some(actor) = output.actor_id.as_deref() { + println!("actor: {actor}"); + } + match &output.commit { + Some(commit) => println!("commit: {}", commit.graph_commit_id), + None => println!("commit: none (the cell was already null)"), + } +} + pub(crate) fn print_blob_stat_human(output: &BlobStatOutput) { let entity = match output.selector.entity { omnigraph_api_types::BlobEntityKind::Node => "node", diff --git a/crates/omnigraph-cli/src/planes.rs b/crates/omnigraph-cli/src/planes.rs index 44c4bf583..4fc1c6fc4 100644 --- a/crates/omnigraph-cli/src/planes.rs +++ b/crates/omnigraph-cli/src/planes.rs @@ -182,7 +182,8 @@ fn flag_applies(flag: ScopeFlag, capability: Capability, cmd: &Command) -> bool Served | Control | Local => false, }, // The actor rides direct-engine writes (`any` via --store) and cluster - // writes; Blob get/stat are reads and never consume it. Served writes + // writes; Blob get/stat are reads and never consume it, while Blob + // put/clear are writes and do. Served writes // resolve the actor from the bearer token // (rejected downstream with its own message). On `direct`, full-text // rebuild attributes its graph publication; other maintenance verbs @@ -196,8 +197,9 @@ fn flag_applies(flag: ScopeFlag, capability: Capability, cmd: &Command) -> bool // so accepting `--as` there is a silent no-op — reject it instead. Any => !matches!( cmd, - Command::Blob { .. } - | Command::Changes { .. } + Command::Blob { + command: BlobCommand::Get { .. } | BlobCommand::Stat { .. } + } | Command::Changes { .. } | Command::Commit { .. } | Command::Query { .. } ), @@ -332,6 +334,8 @@ pub(crate) fn command_label(cmd: &Command) -> &'static str { Command::Blob { command } => match command { BlobCommand::Get { .. } => "blob get", BlobCommand::Stat { .. } => "blob stat", + BlobCommand::Put { .. } => "blob put", + BlobCommand::Clear { .. } => "blob clear", }, Command::Commit { .. } => "commit", Command::Changes { command } => match command { diff --git a/crates/omnigraph-cli/tests/integration/cli_data.rs b/crates/omnigraph-cli/tests/integration/cli_data.rs index 6d914c53d..32f25bdeb 100644 --- a/crates/omnigraph-cli/tests/integration/cli_data.rs +++ b/crates/omnigraph-cli/tests/integration/cli_data.rs @@ -4,7 +4,7 @@ use std::fs; use assert_cmd::Command; -use serde_json::Value; +use serde_json::{Value, json}; use sha2::Digest; use tempfile::tempdir; @@ -571,6 +571,161 @@ fn blob_commands_reject_positional_and_cluster_scope_addressing() { ); } +/// `blob put` stores a file's or stdin's raw bytes and `blob clear` nulls the +/// cell, each answering the exact receipt; `--as` attributes an embedded +/// write. A stale `--if-match` fails with the Blob precondition detail, and +/// a malformed one or input over 32 MiB is refused before the graph is +/// addressed, so an unreachable store is never touched. +#[test] +fn blob_put_and_clear_write_receipts_and_refuse_bad_input_before_scope() { + let temp = tempdir().unwrap(); + let graph = graph_path(temp.path()); + init_blob_graph(&graph); + let blob = |verb: &str, selector: [&str; 4]| { + let mut command = cli(); + command + .arg("blob") + .arg(verb) + .args(selector) + .arg("--store") + .arg(&graph); + command + }; + let readme = ["node", "Document", "readme", "content"]; + let get = |selector: [&str; 4]| output_success(&mut blob("get", selector)).stdout; + let stat_etag = |selector: [&str; 4]| { + let output = output_success(blob("stat", selector).arg("--json")); + parse_stdout_json(&output)["etag"] + .as_str() + .unwrap() + .to_string() + }; + + let file = temp.path().join("payload.bin"); + fs::write(&file, b"from a file\0\xff").unwrap(); + let original = stat_etag(readme); + let put = output_success( + blob("put", readme) + .arg("--file") + .arg(&file) + .arg("--if-match") + .arg(&original) + .arg("--as") + .arg("act-writer") + .arg("--json"), + ); + let receipt = parse_stdout_json(&put); + assert_eq!( + receipt["selector"], + json!({"entity": "node", "type": "Document", "id": "readme", "property": "content"}) + ); + assert_eq!(receipt["branch"], "main"); + assert_eq!(receipt["kind"], "managed"); + assert_eq!(receipt["size"], 13); + assert_eq!(receipt["actor_id"], "act-writer"); + assert_eq!(receipt["commit"]["actor_id"], "act-writer"); + assert_eq!(receipt["etag"], stat_etag(readme).as_str()); + assert_eq!(get(readme), b"from a file\0\xff"); + + let edge = ["edge", "Attachment", "attachment-1", "payload"]; + let human = output_success(blob("put", edge).write_stdin(b"from stdin".to_vec())); + let human = stdout_string(&human); + assert!( + human.contains("kind: managed") && human.contains("size: 10"), + "{human}" + ); + assert_eq!(get(edge), b"from stdin"); + + let stale = output_failure( + blob("put", readme) + .arg("--file") + .arg(&file) + .arg("--if-match") + .arg(&original) + .arg("--json"), + ); + assert_eq!(stale.status.code(), Some(1), "an embedded mismatch exits 1"); + let refused = parse_stdout_json(&stale); + assert_eq!( + refused["blob_precondition_failure"]["current_etag"], + stat_etag(readme).as_str() + ); + assert!(refused.get("precondition_failure").is_none(), "{refused}"); + assert_eq!(refused["command_outcome"]["action"], "refresh"); + + let cleared = parse_stdout_json(&output_success( + blob("clear", readme) + .arg("--if-match") + .arg("*") + .arg("--json"), + )); + assert_eq!(cleared["kind"], "null"); + assert!( + cleared["commit"]["graph_commit_id"].is_string(), + "{cleared}" + ); + assert!(cleared.get("etag").is_none() && cleared.get("size").is_none()); + let again = parse_stdout_json(&output_success(blob("clear", readme).arg("--json"))); + assert!( + again["commit"].is_null(), + "a no-op clear publishes nothing: {again}" + ); + output_failure(&mut blob("get", readme)); + + // The process's `write_max_bytes` bounds the value a `--store` put writes. + let lowered = output_failure( + blob("put", edge) + .env("OMNIGRAPH_WRITE_MAX_BYTES", "4093") + .write_stdin(vec![7_u8; 4094]) + .arg("--json"), + ); + assert_eq!( + parse_stdout_json(&lowered)["resource_limit"], + json!({"resource": "Blob write payload bytes", "limit": 4093, "actual": 4094}) + ); + assert_eq!(get(edge), b"from stdin"); + + // Refused before the graph is addressed: the store does not exist. + let absent = temp.path().join("absent.omni"); + let unreachable = |verb: &str| { + let mut command = cli(); + command + .arg("blob") + .arg(verb) + .args(readme) + .arg("--store") + .arg(&absent); + command + }; + let malformed = output_failure(unreachable("clear").arg("--if-match").arg("unquoted")); + assert!( + String::from_utf8_lossy(&malformed.stderr).contains("If-Match"), + "{malformed:?}" + ); + let oversized = temp.path().join("oversized.bin"); + fs::File::create(&oversized) + .unwrap() + .set_len(32 * 1024 * 1024 + 1) + .unwrap(); + let too_large = output_failure( + unreachable("put") + .arg("--file") + .arg(&oversized) + .arg("--json"), + ); + let refused = parse_stdout_json(&too_large); + assert_eq!( + refused["resource_limit"], + json!({ + "resource": "Blob write payload bytes", + "limit": 32 * 1024 * 1024, + "actual": 32 * 1024 * 1024 + 1, + }) + ); + assert_eq!(refused["command_outcome"]["execution"], "not_started"); + assert!(!absent.exists(), "neither arm opened the store"); +} + #[test] fn embed_seed_fills_missing_and_preserves_existing_vectors_by_default() { let temp = tempdir().unwrap(); diff --git a/crates/omnigraph-cli/tests/integration/parity_matrix.rs b/crates/omnigraph-cli/tests/integration/parity_matrix.rs index 1f7e287c2..2b8666261 100644 --- a/crates/omnigraph-cli/tests/integration/parity_matrix.rs +++ b/crates/omnigraph-cli/tests/integration/parity_matrix.rs @@ -76,6 +76,21 @@ fn blob_parity() -> Parity { } } +/// The Blob fixture with an independent byte copy as the local arm, so each +/// arm's writes land in its own files and never in the hard-linked read twin. +fn blob_write_parity() -> Parity { + let p = blob_parity(); + let served = p + ._temp + .path() + .join("blob-parity-cluster") + .join("graphs") + .join(format!("{PARITY_GRAPH_ID}.omni")); + std::fs::remove_dir_all(&p.local).unwrap(); + copy_dir(&served, &p.local); + p +} + impl Parity { fn run(&self, args: &[&str]) -> (std::process::Output, std::process::Output) { run_both(&self.local, &self.server.base_url, args) @@ -1109,6 +1124,121 @@ fn parity_blob_shared_failures_keep_exit_codes_aligned() { } } +/// The receipts of a Blob write normalized for comparison: each arm's own +/// commit identities and the ETag, which hashes its commit's transaction +/// file, are replaced once they are shown to be present. +fn normalized_blob_receipt( + verb: &str, + arm: &str, + output: &std::process::Output, +) -> serde_json::Value { + assert!(output.status.success(), "{verb}: {arm} failed: {output:?}"); + let mut receipt = parse_stdout_json(output); + if let Some(etag) = receipt.get_mut("etag") { + assert!( + etag.as_str().is_some_and(|etag| etag.starts_with('"')), + "{verb}: {arm}" + ); + *etag = serde_json::Value::String("".into()); + } + if let Some(commit) = receipt + .get_mut("commit") + .and_then(|commit| commit.as_object_mut()) + { + assert!(commit["graph_commit_id"].is_string(), "{verb}: {arm}"); + for key in [ + "graph_commit_id", + "parent_commit_id", + "merged_parent_commit_id", + ] { + if let Some(value) = commit.get_mut(key) + && !value.is_null() + { + *value = serde_json::Value::String(format!("")); + } + } + } + scrub_volatile(&mut receipt); + receipt +} + +/// `blob ` on the fixture's readme content cell, with `--json`. +fn blob_write_args<'a>(verb: &'a str, extra: &[&'a str]) -> Vec<&'a str> { + let mut args = vec!["blob", verb, "node", "Document", "readme", "content"]; + args.extend_from_slice(extra); + args.push("--json"); + args +} + +#[test] +fn parity_blob_put_and_clear_share_receipts_and_refusals() { + let p = blob_write_parity(); + let payload = p._temp.path().join("parity-put.bin"); + std::fs::write(&payload, b"parity bytes\0\xff").unwrap(); + let payload = payload.to_str().unwrap(); + let with = blob_write_args; + + for (verb, args) in [ + ("blob put", with("put", &["--file", payload])), + ("blob clear", with("clear", &[])), + ("blob clear of a null cell", with("clear", &[])), + ] { + let (local, remote) = p.run(&args); + assert_eq!( + normalized_blob_receipt(verb, "local", &local), + normalized_blob_receipt(verb, "remote", &remote), + "{verb}: normalized receipts diverge (left=local, right=remote)" + ); + if verb == "blob put" { + assert_eq!(parse_stdout_json(&local)["actor_id"], PARITY_ACTOR); + } + if verb.ends_with("null cell") { + assert!(parse_stdout_json(&local)["commit"].is_null()); + } + } + + // A stale validator: the same Blob precondition detail on both arms. As + // for a graph-commit precondition, the served arm's verified 412 exits 4 + // and the embedded arm, which opened storage for writing, exits 1. + let (local, remote) = p.run(&with( + "put", + &["--file", payload, "--if-match", "\"stale\""], + )); + assert_eq!( + (local.status.code(), remote.status.code()), + (Some(1), Some(4)) + ); + let (local, remote) = (parse_stdout_json(&local), parse_stdout_json(&remote)); + for (arm, refused) in [("local", &local), ("remote", &remote)] { + assert_eq!(refused["code"], "conflict", "{arm}: {refused}"); + assert!( + refused.get("precondition_failure").is_none(), + "{arm}: {refused}" + ); + assert_eq!( + refused["blob_precondition_failure"], + serde_json::json!({}), + "{arm}: a null cell has no validator" + ); + assert_eq!(refused["command_outcome"]["action"], "refresh", "{arm}"); + } + assert_eq!(local["error"], remote["error"]); + + // Input over the limit is refused identically before either arm starts. + let oversized = p._temp.path().join("parity-oversized.bin"); + std::fs::File::create(&oversized) + .unwrap() + .set_len(32 * 1024 * 1024 + 1) + .unwrap(); + let (local, remote) = p.run(&with("put", &["--file", oversized.to_str().unwrap()])); + assert_eq!(local.status.code(), remote.status.code()); + assert_eq!(parse_stdout_json(&local), parse_stdout_json(&remote)); + assert_eq!( + parse_stdout_json(&local)["command_outcome"]["execution"], + "not_started" + ); +} + // ---- error parity: exit codes must match for shared failure cases ---- #[test] diff --git a/crates/omnigraph-cli/tests/integration/system_remote.rs b/crates/omnigraph-cli/tests/integration/system_remote.rs index 8a08221e8..780d68d58 100644 --- a/crates/omnigraph-cli/tests/integration/system_remote.rs +++ b/crates/omnigraph-cli/tests/integration/system_remote.rs @@ -765,6 +765,173 @@ fn assert_merge_delivery_loss_never_replays_committed_effect(statement: bool) { } } +/// A Blob put or clear whose successful receipt is lost in delivery is +/// published once and never resent: the CLI reports the failure, the server +/// holds exactly the receipt the proxy saw, and the proxy saw one submission. +/// The put carries 2 MiB of raw bytes, past the JSON request bound. +#[test] +fn remote_blob_write_delivery_loss_never_replays_committed_effect() { + use crate::support::managed_http::{IntentApiFixture, MergeDeliveryFault}; + use std::process::Stdio; + use std::time::{Duration, Instant}; + + let temp = tempfile::tempdir().unwrap(); + let (cluster_dir, _) = blob_parity_config(temp.path(), &temp.path().join("local.omni")); + let server = spawn_server_with_cluster_env( + &cluster_dir, + &[( + "OMNIGRAPH_SERVER_BEARER_TOKENS_JSON", + r#"{"act-parity":"parity-tok"}"#, + )], + ); + let client = graph_http_client(); + let graph_url = format!("{}/graphs/{PARITY_GRAPH_ID}", server.base_url); + let blob_url = format!("{graph_url}/blob?entity=node&type=Document&id=readme&property=content"); + let authorized = |request: reqwest::blocking::RequestBuilder| { + request.header("authorization", format!("Bearer {PARITY_TOKEN}")) + }; + let commits = || { + authorized(client.get(format!("{graph_url}/commits?branch=main"))) + .send() + .unwrap() + .error_for_status() + .unwrap() + .json::() + .unwrap() + }; + let payload = temp.path().join("payload.bin"); + let bytes: Vec = (0..2 * 1024 * 1024) + .map(|index| (index % 251) as u8) + .collect(); + std::fs::write(&payload, &bytes).unwrap(); + + for put in [true, false] { + for fault in [ + MergeDeliveryFault::Disconnect, + MergeDeliveryFault::Truncate, + MergeDeliveryFault::GatewayTimeout, + MergeDeliveryFault::CallerWait, + ] { + if !put { + authorized(client.put(&blob_url)) + .header("content-type", "application/octet-stream") + .body(b"to clear".to_vec()) + .send() + .unwrap() + .error_for_status() + .unwrap(); + } + let before = commits(); + let proxy = IntentApiFixture::graph_blob_write_proxy(&server.base_url, fault); + let mut args: Vec = ["blob", if put { "put" } else { "clear" }] + .into_iter() + .chain(["node", "Document", "readme", "content"]) + .map(Into::into) + .collect(); + if put { + args.extend(["--file".into(), payload.clone().into_os_string()]); + } + args.extend( + [ + "--server", + proxy.origin.as_str(), + "--graph", + PARITY_GRAPH_ID, + "--json", + ] + .map(Into::into), + ); + // A withheld reply is observed, not waited out: once the proxy + // holds the consumed upstream receipt, the caller is still waiting + // and is abandoned. + let output = if matches!(fault, MergeDeliveryFault::CallerWait) { + let mut caller = cli_process() + .env("OMNIGRAPH_BEARER_TOKEN", PARITY_TOKEN) + .args(&args) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let deadline = Instant::now() + Duration::from_secs(15); + while proxy.forwarded_responses().is_empty() { + assert!( + caller.try_wait().unwrap().is_none(), + "{put}/{fault:?}: caller must still wait while the proxy owns the consumed reply" + ); + assert!( + Instant::now() < deadline, + "{put}/{fault:?}: upstream write must complete before the caller is abandoned" + ); + std::thread::sleep(Duration::from_millis(5)); + } + assert!( + caller.try_wait().unwrap().is_none(), + "{put}/{fault:?}: caller must still wait when the consumed reply is withheld" + ); + caller.kill().unwrap(); + caller.wait_with_output().unwrap() + } else { + output_failure( + cli() + .env("OMNIGRAPH_BEARER_TOKEN", PARITY_TOKEN) + .args(&args) + .timeout(Duration::from_secs(15)), + ) + }; + assert!( + !output.status.success(), + "{put}/{fault:?}: lost delivery must not report success" + ); + if !matches!(fault, MergeDeliveryFault::CallerWait) { + assert_eq!(output.status.code(), Some(1), "{put}/{fault:?}"); + } + if !output.stdout.is_empty() { + let error = parse_stdout_json(&output); + assert!(error.get("error").is_some(), "{put}/{fault:?}: {error}"); + assert!( + error.get("commit").is_none(), + "lost delivery is not success" + ); + } + + let captured = proxy.forwarded_responses(); + assert_eq!(captured.len(), 1, "{put}/{fault:?}: one upstream write"); + let upstream: serde_json::Value = serde_json::from_slice(&captured[0].body).unwrap(); + assert_eq!(upstream["kind"], if put { "managed" } else { "null" }); + let receipt = &upstream["commit"]; + let after = commits(); + assert_eq!(after["commits"][0], *receipt, "{put}/{fault:?}"); + assert_eq!( + after["commits"].as_array().unwrap().len(), + before["commits"].as_array().unwrap().len() + 1, + "{put}/{fault:?}: published once" + ); + let read = authorized(client.get(&blob_url)).send().unwrap(); + if put { + assert_eq!(read.status(), reqwest::StatusCode::OK); + assert_eq!( + read.headers()["etag"].to_str().unwrap(), + upstream["etag"].as_str().unwrap() + ); + assert_eq!(read.bytes().unwrap().as_ref(), bytes.as_slice()); + } else { + assert_eq!(read.status(), reqwest::StatusCode::NOT_FOUND); + } + proxy.assert_complete(); + let requests = proxy.requests(); + assert_eq!(requests.len(), 2, "discovery and exactly one submission"); + assert_eq!(requests[0].method, "HEAD"); + assert_eq!(requests[1].method, if put { "PUT" } else { "DELETE" }); + if put { + assert_eq!( + requests[1].raw_body, bytes, + "the raw body crossed unchanged" + ); + } + } + } +} + #[test] fn remote_deployment_delivery_loss_preserves_owned_completion() { use crate::support::managed_http::{DeploymentDeliveryFault, IntentApiFixture}; diff --git a/crates/omnigraph-cli/tests/support/managed_http.rs b/crates/omnigraph-cli/tests/support/managed_http.rs index e4800db90..9d6ae7211 100644 --- a/crates/omnigraph-cli/tests/support/managed_http.rs +++ b/crates/omnigraph-cli/tests/support/managed_http.rs @@ -34,8 +34,9 @@ pub struct IntentReply { pub body: Vec, } -/// Delivery faults applied only after a real server's successful merge reply -/// has been fully read. They cannot stand in for cancelling a server request. +/// Delivery faults applied only after a real server's successful merge or +/// Blob write reply has been fully read. They cannot stand in for cancelling a +/// server request. #[derive(Debug, Clone, Copy)] pub enum MergeDeliveryFault { Disconnect, @@ -115,6 +116,12 @@ impl IntentApiFixture { ) } + /// Forward discovery and one Blob put or clear to an actual server, then + /// break only delivery of its successful receipt. + pub fn graph_blob_write_proxy(upstream: &str, fault: MergeDeliveryFault) -> Self { + Self::graph_merge_proxy(upstream, fault) + } + pub fn graph_deployment_proxy(upstream: &str, fault: DeploymentDeliveryFault) -> Self { Self::start_with_origin( |_| Vec::new(), @@ -214,17 +221,22 @@ impl IntentApiFixture { let length = headers .get("content-length") .map_or(0, |n| n.parse::().unwrap()); - assert!(length <= 1024 * 1024, "fixture request body bound"); + // A Blob put carries up to 32 MiB of raw bytes; every other + // request stays within the JSON bound. + assert!(length <= 33 * 1024 * 1024, "fixture request body bound"); let mut body = vec![0; length]; reader.read_exact(&mut body).unwrap(); - let ndjson = headers - .get("content-type") - .is_some_and(|value| value == "application/x-ndjson"); + let raw = headers.get("content-type").is_some_and(|value| { + value == "application/x-ndjson" || value == "application/octet-stream" + }); + if !raw { + assert!(length <= 1024 * 1024, "fixture JSON request body bound"); + } let request = IntentRequest { method: request_line[0].clone(), path: request_line[1].clone(), headers, - body: if body.is_empty() || ndjson { + body: if body.is_empty() || raw { Value::Null } else { serde_json::from_slice(&body).unwrap() @@ -346,12 +358,19 @@ impl IntentApiFixture { headers, body, }; + let blob_write = matches!(request.method.as_str(), "PUT" | "DELETE") + && request + .path + .split('?') + .next() + .is_some_and(|path| path.ends_with("/blob")); if let Forwarding::Merge(_, fault) = forwarding - && request.method == "POST" - && (request.path.ends_with("/branches/merge") - || request.path.ends_with("/mutate")) + && ((request.method == "POST" + && (request.path.ends_with("/branches/merge") + || request.path.ends_with("/mutate"))) + || blob_write) { - assert_eq!(reply.status, 200, "upstream merge must succeed"); + assert_eq!(reply.status, 200, "upstream write must succeed"); captured_responses.lock().unwrap().push(reply.clone()); match fault { MergeDeliveryFault::Disconnect => continue, diff --git a/crates/omnigraph-cli/tests/support/mod.rs b/crates/omnigraph-cli/tests/support/mod.rs index cd53b2484..1e642df55 100644 --- a/crates/omnigraph-cli/tests/support/mod.rs +++ b/crates/omnigraph-cli/tests/support/mod.rs @@ -1420,14 +1420,17 @@ pub fn run_both( if args.first() == Some(&"blob") { local.env("NO_COLOR", "1"); } - // The read commands (blob/query/changes/commit) reject `--as`: a served read - // resolves the actor from its bearer token and a direct read attributes - // none, so passing `--as` there is an error. Only the write verbs carry the - // parity actor on the direct arm. - if !matches!( - args.first().copied(), - Some("blob") | Some("query") | Some("changes") | Some("commit") - ) { + // The read commands (blob get/stat, query, changes, commit) reject `--as`: + // a served read resolves the actor from its bearer token and a direct read + // attributes none, so passing `--as` there is an error. Only the write + // verbs, `blob put` and `blob clear` among them, carry the parity actor on + // the direct arm. + let read = match args { + ["blob", verb, ..] => matches!(*verb, "get" | "stat"), + [verb, ..] => matches!(*verb, "query" | "changes" | "commit"), + [] => false, + }; + if !read { local.arg("--as").arg(PARITY_ACTOR); } diff --git a/docs/dev/blob.md b/docs/dev/blob.md index e9d0bf524..66306a628 100644 --- a/docs/dev/blob.md +++ b/docs/dev/blob.md @@ -2,7 +2,7 @@ OmniGraph treats a Blob as one property cell on an existing node or edge. Lance owns the Blob-v2 physical placement; OmniGraph owns logical identity, snapshot selection, authorization, external-source admission, bounded delivery, and graph-level publication. -This page describes the implemented read, ingestion and cell-write behavior. The CLI put and clear commands proposed by [RFC 0033](../rfcs/0033-blob-management.md) are not implemented yet. +This page describes the implemented read, ingestion and cell-write behavior proposed by [RFC 0033](../rfcs/0033-blob-management.md). ## Logical states @@ -86,7 +86,9 @@ An external value produces a `302` redirect with the stored URI; the server neve The CLI exposes `blob get` and `blob stat` for embedded and remote graphs. `get` streams managed bytes to stdout or `--out`; `stat` returns kind, resolved-view metadata, size/ETag for managed data, or the descriptor for external data. The CLI refuses to follow external references. -`PUT` and `DELETE /graphs/{graph_id}/blob` call the engine cell writes. `PUT` is a raw ingress route like `/load/ndjson`: the ingress middleware reserves the put limit (`AppState::blob_put_limit`: the process `write_max_bytes`, at most the engine's `BLOB_WRITE_MAX_BYTES`, known before the body because a put takes no request settings) without collecting the body, and the handler authorizes `change` on the branch, checks `Content-Type` (415) and a declared `Content-Length` against that limit (413) before it polls a byte, then collects the body under the shared body deadline (408) into one buffer sized from that length, which the engine's value column adopts. The ingress reservation shrinks to the received size. Both verbs admit the actor's workload and submit the engine call through `owned_write`, so a disconnect after admission loses only the response. `omnigraph_api_types::parse_blob_if_match` turns `If-Match` field lines into a `BlobPrecondition` (`*` alone, or the strong tags of a list; weak tags dropped; malformed fields refused), and `blob_write_output` maps the engine outcome to the receipt; the CLI's embedded writes will share both. `BlobWritePreconditionFailed` maps to 412 with `ErrorCode::Conflict`, the `blob_precondition_failure` detail and an `ETag` header, never the graph-commit `precondition_failure`. The CLI has no put or clear yet. +`PUT` and `DELETE /graphs/{graph_id}/blob` call the engine cell writes. `PUT` is a raw ingress route like `/load/ndjson`: the ingress middleware reserves the put limit (`AppState::blob_put_limit`: the process `write_max_bytes`, at most the engine's `BLOB_WRITE_MAX_BYTES`, known before the body because a put takes no request settings) without collecting the body, and the handler authorizes `change` on the branch, checks `Content-Type` (415) and a declared `Content-Length` against that limit (413) before it polls a byte, then collects the body under the shared body deadline (408) into one buffer sized from that length, which the engine's value column adopts. The ingress reservation shrinks to the received size. Both verbs admit the actor's workload and submit the engine call through `owned_write`, so a disconnect after admission loses only the response. `omnigraph_api_types::parse_blob_if_match` turns `If-Match` field lines into a `BlobPrecondition` (`*` alone, or the strong tags of a list; weak tags dropped; malformed fields refused), and `blob_write_output` maps the engine outcome to the receipt; the CLI shares both. `BlobWritePreconditionFailed` maps to 412 with `ErrorCode::Conflict`, the `blob_precondition_failure` detail and an `ETag` header, never the graph-commit `precondition_failure`. + +`blob put` and `blob clear` (`crates/omnigraph-cli`) parse `--if-match` and read `put`'s file or stdin, refusing more than `BLOB_WRITE_MAX_BYTES` without reading past one more byte, before scope resolution, so neither arm starts on bad input. That bound is the setting's maximum, because a served caller cannot see the server's `write_max_bytes`; a `--store` put is then bounded by the process `write_max_bytes` in the engine. The embedded arm calls the engine writes and renders a precondition failure through `engine_error_output`, the body a server answers; the served arm sends the raw `If-Match` value and the bytes once, with no retry. A served 412 whose only detail is `blob_precondition_failure` is a verified request precondition, so a single-request command exits 4, as for a graph-commit precondition; the embedded arm, which opened storage for writing, exits 1. ## Maintenance and export @@ -105,6 +107,6 @@ Export emits managed values as base64 and whole-object external values as URI de - Lance compatibility: `crates/omnigraph/tests/lance_surface_guards.rs`. - Cluster policy persistence and serving projection: `omnigraph-cluster` in-source tests. - HTTP transport: `crates/omnigraph-server/tests/integration/data_routes.rs`, `integration/auth_policy.rs`, and `integration/openapi.rs`. The writes: `data_routes.rs::blob_put_and_delete_return_exact_receipts_and_blob_preconditions`, `blob_put_raw_body_bounds_media_type_length_and_deadline`, the Blob doors of `disconnected_writes_keep_admission_until_the_original_operation_finishes`, the `change` cases of `auth_policy.rs::policy_blocks_change_on_protected_main_but_allows_unprotected_branch`, and the `If-Match` parser's tests in `omnigraph-api-types`. -- CLI and embedded/remote parity: `crates/omnigraph-cli/tests/integration/cli_data.rs` and `integration/parity_matrix.rs`. +- CLI and embedded/remote parity: `crates/omnigraph-cli/tests/integration/cli_data.rs` and `integration/parity_matrix.rs` (`parity_blob_put_and_clear_share_receipts_and_refusals` for the writes); `integration/system_remote.rs::remote_blob_write_delivery_loss_never_replays_committed_effect` owns lost delivery of a put or clear through the fault proxy. The user contract and examples live in [Blob values](../user/blobs.md). diff --git a/docs/dev/testing.md b/docs/dev/testing.md index 1e0a285ce..b09d6edcc 100644 --- a/docs/dev/testing.md +++ b/docs/dev/testing.md @@ -118,7 +118,7 @@ Blob coverage is deliberately split: - engine `integration/maintenance.rs` owns Blob compaction (the batch derived from a row's summed Blob columns, fragments with deleted rows, per-task sizing in `integration/maintenance.rs::optimize_sizes_each_compaction_task_from_its_own_fragments`, external references counting nothing in `integration/maintenance.rs::optimize_does_not_size_a_blob_batch_by_external_references`); - cluster tests own persisted external-source policy and serving projections; - server `integration/data_routes.rs`, `integration/auth_policy.rs`, and `integration/openapi.rs` own GET/HEAD, auth, conditions, ranges, redirects, backpressure, and schema drift; -- CLI `integration/cli_data.rs` owns `blob get/stat`; `integration/parity_matrix.rs` compares embedded and remote results. +- CLI `integration/cli_data.rs` owns `blob get/stat/put/clear`; `integration/parity_matrix.rs` compares embedded and remote results; `integration/system_remote.rs` owns a put's or clear's lost delivery through the fault proxy. Do not exercise a server promise solely through the engine facade. The complete contract is summarized in [blob.md](blob.md). diff --git a/docs/user/blobs.md b/docs/user/blobs.md index 8c2caa0db..626cfc6b0 100644 --- a/docs/user/blobs.md +++ b/docs/user/blobs.md @@ -126,8 +126,23 @@ Over HTTP: disconnects, so a lost response means the write may have landed: read the cell before writing again. -The CLI does not offer these writes yet; through it, change a Blob value with a -load or a mutation. +From the CLI, `blob put` stores the bytes of `--file` or stdin and `blob clear` +nulls the cell. Both take `--branch`, `--if-match` and `--json`, and print the +receipt the server returns: + +```bash +omnigraph blob put node Document manual content --file manual.pdf \ + --if-match '""' --store graph.omni +omnigraph blob clear node Document manual content --server prod --graph knowledge +``` + +The CLI refuses input over 32 MiB and a malformed `--if-match` before it +addresses the graph. The value must also fit `write_max_bytes`: the process's +`OMNIGRAPH_WRITE_MAX_BYTES` for a `--store` write, the server's for a served +one. `--as` names the actor of a `--store` write; a served +write takes its actor from the bearer token. A served Blob `412` exits 4, as a +graph-commit precondition does; an embedded one exits 1. Each write is sent +once: after an unknown outcome, read the cell before writing again. ## Query behavior @@ -136,8 +151,7 @@ filtered, ordered, or aggregated. Write them through load or mutation assignment, then read an individual Blob value through the dedicated CLI or HTTP surface. -There are no `blob put` or `blob clear` CLI commands yet. Use the normal graph -write path, or the HTTP and embedded writes in +To replace or clear one value, use the CLI, HTTP or embedded writes in [Replacing one Blob value](#replacing-one-blob-value); each is an atomic graph commit. diff --git a/docs/user/cli/index.md b/docs/user/cli/index.md index 10a50fdea..f6a66e0a4 100644 --- a/docs/user/cli/index.md +++ b/docs/user/cli/index.md @@ -125,9 +125,9 @@ from a request`. A stored query takes no `--set`: `omnigraph query ` without `-e` or `--query` runs under the process defaults. -## Read Blob values +## Read and write Blob values -Read a managed Blob cell to a file or inspect its metadata: +Read a managed Blob cell to a file, inspect its metadata, or replace it: ```bash omnigraph blob get node Document doc-42 body \ @@ -135,11 +135,16 @@ omnigraph blob get node Document doc-42 body \ omnigraph blob stat node Document doc-42 body \ --store ./graph.omni --json + +omnigraph blob put node Document doc-42 body \ + --store ./graph.omni --file body.bin ``` `blob get` writes bytes to stdout when `--out` is omitted. Add `--offset` and `--length` for a range. The CLI reports external references but does not follow -them. See [Blob values](../blobs.md) for the complete contract. +them. `blob put` reads stdin when `--file` is omitted, and `blob clear` sets a +nullable cell to null; both accept `--if-match`. See [Blob values](../blobs.md) +for the complete contract. ## Inspect and maintain a graph diff --git a/docs/user/cli/reference.md b/docs/user/cli/reference.md index ffef9c9fd..ca49a8008 100644 --- a/docs/user/cli/reference.md +++ b/docs/user/cli/reference.md @@ -43,7 +43,7 @@ server resolves the actor from the bearer token. Drop it, or use `--store ` | `query` | Run a read query, or the `branch list`, `show`, or `explain` statement | direct or served | | `mutate` | Run an insert/update/delete query (`as commit` leaves it unpublished), or a `branch create`, `branch delete`, `branch merge`, `commit publish`, or `commit drop` statement | direct or served | | `load` | Load graph JSONL in `overwrite`, `append`, or `merge` mode | direct or served | -| `blob get`, `blob stat` | Read or inspect one Blob cell | direct or served | +| `blob get/stat/put/clear` | Read, inspect, replace or clear one Blob cell | direct or served | | `branch create/list/delete/merge` | Manage graph branches; `merge --as-commit` leaves the merge unpublished | direct or served | | `snapshot` | Show a branch snapshot: `internal_schema_version`, `graph_manifest_version`, and the datasets of one captured graph version | direct or served | | `commit list/show/changes/publish/drop` | Inspect history (`list --unpublished` for a branch's unpublished commits) or one commit's entity changes; publish an unpublished commit onto its recorded target branch, or drop it | direct or served | @@ -175,13 +175,13 @@ in [Changes and Change Feeds](../branching/changes.md). ## Blob commands ```text -omnigraph blob get [scope] [options] -omnigraph blob stat [scope] [options] +omnigraph blob get|stat|put|clear [scope] [options] ``` `get` accepts `--branch` or `--snapshot`, `--offset`, `--length`, and -`--out `. `stat` accepts `--branch` or `--snapshot` and `--json`. -See [Blob values](../blobs.md). +`--out `. `stat` accepts `--branch` or `--snapshot` and `--json`. `put` +reads `--file ` or stdin; `put` and `clear` accept `--branch`, +`--if-match `, `--json` and, on `--store`, `--as`. See [Blob values](../blobs.md). ## Operator configuration