diff --git a/src/lib.rs b/src/lib.rs index 01e4564..0412e54 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -32,8 +32,10 @@ pub mod ops; #[cfg(feature = "ops")] pub use ops::{ - FanoutChunk, FanoutPublisher, FanoutReceiver, FanoutRegistry, OpsConfig, OpsContext, - PinConverter, PinToken, RenewalOutcome, SlowSubscriberPolicy, TokenRegistry, + delete_pair, have_pair, stat_pair, FanoutChunk, FanoutPublisher, FanoutReceiver, + FanoutRegistry, OpsConfig, OpsContext, PinConverter, PinToken, RenewalOutcome, + SlowSubscriberPolicy, TokenRegistry, BLOB_MISSING, GC_REFUSED, INVALID_DIGEST, + INVALID_RENEWAL_FORM, }; #[cfg(any(feature = "postgres", feature = "pg-lo"))] diff --git a/src/ops.rs b/src/ops.rs index a5c58cb..75c23d5 100644 --- a/src/ops.rs +++ b/src/ops.rs @@ -1,9 +1,11 @@ -//! The ops surface's state substrate (ADR-001 §3 placement; the -//! wave-7 foundation): the feature-gated module's context type, the -//! pin-token layer, the namespace→liveness-source binding, the +//! The ops surface's state substrate + the JSON control plane's +//! spec/handler pairs (ADR-001 §3 placement; the wave-7 foundation and +//! the ops-json-ops handler family): the feature-gated module's context +//! type, the pin-token layer, the namespace→liveness-source binding, the //! crate-internal store-core renewal access, and the fetch fanout -//! registry — everything the op handlers ride, none of the handlers -//! themselves (spec/handler pairs land in the follow-on tasks). +//! registry — everything the op handlers ride — plus the JSON control +//! ops' exported pairs; the binary (`blobs/fetch`/`blobs/put`) data +//! plane lands in the follow-on tasks. //! //! **The ops layer is a store-facade consumer** (ADR-001 §4): its //! store access is the public facade plus exactly one crate-internal @@ -138,8 +140,8 @@ //! (ADR-012 §2), and no engine call runs. //! //! What never happens: a *consumer* reaching the seam, or a public -//! API widening for this. The have handler (later task) rides -//! [`OpsContext::renew`] and sees none of the engine. +//! API widening for this. The have handler (the renewal form below) +//! rides [`OpsContext::renew`] and sees none of the engine. //! //! ## The fetch fanout registry (ops-surface.md; POC #3 finding A3) //! @@ -171,7 +173,8 @@ //! ## The renewal-cadence boundary (ADR-013 §1) //! //! This module ships the renewal *mechanism* (the have form's -//! handler, later task, composing [`OpsContext::renew`]). The TTL/3 +//! handler, the JSON control plane's renewal arm, composing +//! [`OpsContext::renew`]). The TTL/3 //! *cadence* — the putter's renewal worker — is **embedder //! composition above the facade** (scheduler/queues; ADR-008's //! "embedder owns cadence"), never in-crate machinery: no renewal @@ -184,14 +187,62 @@ //! dependencies: the default build's dependency tree is unchanged, //! and `ops` pulls alkcall plus these, default-off. The base crate //! stays lean. +//! +//! ## The JSON control plane (ops-surface.md §Op family; alkcall +//! ADR-016's spec/schema/handler shape): +//! [`stat_pair`]/[`have_pair`]/[`delete_pair`] — the three JSON ops +//! exported as spec/handler pairs the embedder assembles (the +//! `OperationRegistryBuilder` registration; the module's test tree +//! realizes the assembly pattern). The binary data plane +//! (`blobs/fetch`, `blobs/put`) lands with the follow-on tasks. +//! +//! **The machinery ridden (never re-invented)**: JSON Schema validation +//! at the registry (`INVALID_INPUT` on violation, before the handler), +//! the typed error schemas (`ErrorDefinition`s), the ACL at op entry — +//! alkcall's `AccessControl` with the **namespace-as-resource mapping**: +//! `resource_type: "blob-namespace"` with the `resource_id_path` +//! selecting the payload's `namespace` field; stat/have carry the +//! **read** action; **unlisted namespaces deny by default** (OQ-09's +//! closed-by-default — no ACL entry ⇒ no read, no write; a grant per +//! namespace is the deferred cost, deliberately paid over accidental +//! exposure). This module never re-implements an authorization check +//! (convention 6). +//! +//! **The digest wire form is hex** — the canonical digest hex (the same +//! form `blobs/put`'s response returns), parsed via the key layer's hex +//! ingress ([`Digest::from_git_oid_hex`]): lowercase, 40 or 64 chars +//! (the algorithm lengths' disambiguation); malformed input is rejected +//! at the schema (the protocol's `INVALID_INPUT`) and at the handlers' +//! ingress ([`INVALID_DIGEST`], declared) — rejects, never guesses. +//! +//! **The handlers are store-facade consumers** (ADR-001 §4): `stat` +//! rides `Store::stat` (kv first, then the large tier — the fall-through +//! general rule, ADR-012 §6.1), the non-token `have` rides `Store::has` +//! (ADR-008's has discipline — a pure existence probe), `delete` rides +//! `Store::delete` (the arbitration-then-delete path), and the renewal +//! form rides exactly the crate-internal seam this module already owns +//! ([`OpsContext::renew`]) — no backend side-door, no engine types. +//! `delete` is **`Visibility::Internal`** — never callable from the +//! wire (an external discovery answers the protocol's `NOT_FOUND`), not +//! namespace-gated (the `namespace` field exists for audit/triage +//! scoping), gated by the internal authority context the embedder +//! declares at registration (alkcall ADR-017: internal calls switch +//! authority context, never skip ACL). use std::collections::BTreeMap; use std::sync::Arc; +use alkcall::protocol::wire::{CallError, ResponseEnvelope}; +use alkcall::registry::registration::{make_handler, Handler}; +use alkcall::registry::spec::{ + AccessControl, ErrorDefinition, OperationSpec, OperationType, Visibility, +}; use parking_lot::Mutex; +use serde_json::{json, Value}; use tokio::sync::mpsc; use crate::facade::Pin; +use crate::hashing::Digest; use crate::keys::Key; use crate::store::Store; use crate::Error; @@ -878,6 +929,446 @@ impl OpsContext { } } +/// The domain error codes the JSON ops declare (alkcall ADR-016's +/// operation-level codes — distinct from the protocol-level codes; the +/// dispatch carries the declared code and its schema-conforming +/// `details` to the caller). +pub const BLOB_MISSING: &str = "BLOB_MISSING"; +/// `blobs/stat`/`blobs/have`/`blobs/delete`'s ingress rejection: the +/// digest field is not the canonical digest hex (lowercase, 40 or 64 +/// chars — the key layer's rejects-never-guesses posture). +pub const INVALID_DIGEST: &str = "INVALID_DIGEST"; +/// `blobs/have`'s renewal-form arity rejection: a renewing probe +/// carries the token alongside exactly one digest (ADR-012 §2). +pub const INVALID_RENEWAL_FORM: &str = "INVALID_RENEWAL_FORM"; +/// `blobs/delete`'s protection refusal (`Error::GcRefuse`): the digest +/// is pinned, liveness-live, or protect-callback-named — nothing +/// deleted. +pub const GC_REFUSED: &str = "GC_REFUSED"; + +/// The digest wire form's ingress (see the section doc): the key +/// layer's hex parse — lowercase 40/64-char digest hex — wrapped into +/// the typed [`INVALID_DIGEST`] error carrying the offending string. +fn digest_ingress(digest_hex: &str) -> Result { + match Digest::from_git_oid_hex(digest_hex) { + Ok(digest) => Ok(Key::from(digest)), + Err(_) => Err(CallError::new( + INVALID_DIGEST, + "the digest is not canonical digest hex (lowercase, 40 or 64 chars)", + false, + ) + .with_details(json!({ "digest": digest_hex }))), + } +} + +fn envelope_error(request_id: &str, error: CallError) -> ResponseEnvelope { + ResponseEnvelope::error(request_id.to_owned(), error) +} + +fn schema_validation_error(request_id: &str) -> ResponseEnvelope { + envelope_error( + request_id, + CallError::invalid_input("input failed input_schema validation"), + ) +} + +fn store_internal_error(request_id: &str, stage: &str, error: Error) -> ResponseEnvelope { + envelope_error( + request_id, + CallError::internal(format!("the store core failed at {stage}: {error}")), + ) +} + +fn declared_error(code: &str, description: &str, schema: Value) -> ErrorDefinition { + ErrorDefinition { + code: code.to_owned(), + description: description.to_owned(), + schema, + http_status: None, + } +} + +/// The digest-scoped error's `details` schema (the declared error +/// payload: `{digest}`). +fn digest_details_schema() -> Value { + json!({ + "type": "object", + "properties": { + "digest": { "type": "string" } + }, + "required": ["digest"], + "additionalProperties": false + }) +} + +const DIGEST_WIRE_PATTERN: &str = "^[0-9a-f]{40}$|^[0-9a-f]{64}$"; + +fn namespace_schema() -> Value { + json!({ "type": "string", "minLength": 1 }) +} + +fn digest_wire_schema() -> Value { + json!({ "type": "string", "pattern": DIGEST_WIRE_PATTERN }) +} + +fn digests_wire_schema() -> Value { + json!({ + "type": "array", + "items": digest_wire_schema() + }) +} + +fn present_wire_schema() -> Value { + json!({ + "type": "object", + "properties": { "present": { "type": "array", "items": { "type": "boolean" } } }, + "required": ["present"], + "additionalProperties": false + }) +} + +fn json_string_field<'a>(input: &'a Value, field: &str) -> Option<&'a str> { + input.get(field).and_then(Value::as_str) +} + +fn json_string_array<'a>(input: &'a Value, field: &str) -> Option> { + input.get(field).and_then(Value::as_array).map(|values| { + values + .iter() + .map(Value::as_str) + .collect::>>() + })? +} + +/// **`blobs/stat`** `{namespace, digest}` → `{len}` — the probe before +/// offering (ops-surface.md §Op family): a **Query**, the **read** +/// action on the namespace resource; the handler rides `Store::stat` +/// (kv first, then the large tier — the fall-through general rule, +/// ADR-012 §6.1). A miss is the typed missing-error schema +/// ([`BLOB_MISSING`] with the digest as details) — never a +/// `{len: null}`. +pub fn stat_pair(ctx: Arc) -> (OperationSpec, Handler) { + let spec = OperationSpec::new( + "blobs/stat", + OperationType::Query, + Visibility::External, + json!({ + "type": "object", + "properties": { + "namespace": namespace_schema(), + "digest": digest_wire_schema() + }, + "required": ["namespace", "digest"], + "additionalProperties": false + }), + json!({ + "type": "object", + "properties": { "len": { "type": "integer", "minimum": 0 } }, + "required": ["len"], + "additionalProperties": false + }), + vec![ + declared_error( + BLOB_MISSING, + "the digest is absent from the namespace's pool", + digest_details_schema(), + ), + declared_error( + INVALID_DIGEST, + "the digest is not canonical digest hex (lowercase, 40 or 64 chars)", + digest_details_schema(), + ), + ], + AccessControl { + resource_type: Some("blob-namespace".to_owned()), + resource_action: Some("read".to_owned()), + ..AccessControl::default() + }, + Some("$.namespace".to_owned()), + ) + .with_description( + "the entry's byte length for a digest, no content load (the probe before offering)", + ); + let handler = make_handler(move |input, context| { + let ctx = Arc::clone(&ctx); + async move { + let request_id = context.request_id.clone(); + let Some(digest_hex) = json_string_field(&input, "digest") else { + return schema_validation_error(&request_id); + }; + let key = match digest_ingress(digest_hex) { + Ok(key) => key, + Err(error) => return envelope_error(&request_id, error), + }; + match ctx.store().stat(&key.as_bytes()).await { + Ok(Some(meta)) => ResponseEnvelope::ok(request_id, json!({ "len": meta.len })), + Ok(None) => envelope_error( + &request_id, + CallError::new( + BLOB_MISSING, + "the digest is absent from the namespace's pool", + false, + ) + .with_details(json!({ "digest": digest_hex })), + ), + Err(error) => store_internal_error(&request_id, "stat", error), + } + } + }); + (spec, handler) +} + +/// **`blobs/have`** `{namespace, digests[], token?}` → `{present[]}` — +/// the have half of have/need set diffing (ops-surface.md §Op family; +/// ADR-012 §2): a **Query**, **read-gated** exactly like fetch (an +/// existence probe over arbitrary digests is a discovery surface) — +/// hashes only, content never traverses this op; `present[]` +/// corresponds to `digests[]` by index. +/// +/// **Without `token`** — the pure existence probe (ADR-008's has +/// discipline) over the facade's `has`; an engine failure folds to +/// `present: false` (the probes stay probes; the caller's next act — +/// fetch against the miss — is always safe, and the putter's backstop +/// re-put is CAS-idempotent, never harmful). +/// +/// **With `token`** — the **renewal form** (ADR-012 §2; ADR-010 §4): +/// exactly one digest accompanies the token (a different count is the +/// typed [`INVALID_RENEWAL_FORM`] error); the handler rides +/// [`OpsContext::renew`] — the foundation's verify-the-mapping-then- +/// renew probe (the validity domain, the one-token-one-row digest +/// check, the store core's classification then the renewal) — and +/// answers `present[]` as usual: a live row renews (`present: true`), +/// and a malformed, unknown (invalid or foreign — the validity-domain +/// refusal), mismatched, or expired answer is `present: false, no +/// renewal occurred` (never an error — the putter's backstop is re-put; +/// the store-free malformed-token arm is [`RenewalOutcome::UnknownToken`]'s +/// mapping). +pub fn have_pair(ctx: Arc) -> (OperationSpec, Handler) { + let spec = OperationSpec::new( + "blobs/have", + OperationType::Query, + Visibility::External, + json!({ + "type": "object", + "properties": { + "namespace": namespace_schema(), + "digests": digests_wire_schema(), + "token": { "type": "string" } + }, + "required": ["namespace", "digests"], + "additionalProperties": false, + "if": { "required": ["token"] }, + "then": { "properties": { "digests": { "minItems": 1, "maxItems": 1 } } } + }), + present_wire_schema(), + vec![ + declared_error( + INVALID_RENEWAL_FORM, + "a renewal probe carries exactly one digest alongside its token", + json!({ + "type": "object", + "properties": { "digests_len": { "type": "integer", "minimum": 0 } }, + "required": ["digests_len"], + "additionalProperties": false + }), + ), + declared_error( + INVALID_DIGEST, + "the digest is not canonical digest hex (lowercase, 40 or 64 chars)", + digest_details_schema(), + ), + ], + AccessControl { + resource_type: Some("blob-namespace".to_owned()), + resource_action: Some("read".to_owned()), + ..AccessControl::default() + }, + Some("$.namespace".to_owned()), + ) + .with_description("the existence probe over digests (index-corresponded present[]); the pin-token renewal form is its single-digest shape"); + let handler = make_handler(move |input, context| { + let ctx = Arc::clone(&ctx); + async move { + let request_id = context.request_id.clone(); + let Some(digests_wire) = json_string_array(&input, "digests") else { + return schema_validation_error(&request_id); + }; + match json_string_field(&input, "token") { + Some(token_hex) => have_renewal(ctx, &request_id, &digests_wire, token_hex).await, + None => have_probe(ctx, &request_id, &digests_wire).await, + } + } + }); + (spec, handler) +} + +/// The renewal form's handler arm: the one-digest arity check, the +/// digest's ingress, the token's ingress (malformed tokens answer the +/// unknown-token form, `present: false`, not an error), then the +/// foundation's renewal probe ([`OpsContext::renew`]) — its outcome is +/// the wire answer. +async fn have_renewal( + ctx: Arc, + request_id: &str, + digests_wire: &[&str], + token_hex: &str, +) -> ResponseEnvelope { + if digests_wire.len() != 1 { + return envelope_error( + request_id, + CallError::new( + INVALID_RENEWAL_FORM, + "a renewal probe carries exactly one digest alongside its token", + false, + ) + .with_details(json!({ "digests_len": digests_wire.len() })), + ); + } + let key = match digest_ingress(digests_wire[0]) { + Ok(key) => key, + Err(error) => return envelope_error(request_id, error), + }; + let present = match PinToken::from_hex(token_hex) { + Some(token) => ctx.renew(&token, &key).await.present(), + None => false, + }; + ResponseEnvelope::ok(request_id, json!({ "present": [present] })) +} + +/// The pure existence probe's handler arm: the digests' ingress, then +/// `Store::has` per digest, index-corresponded; engine failures fold to +/// `present: false` (the probes stay probes — ADR-008's discipline). +async fn have_probe( + ctx: Arc, + request_id: &str, + digests_wire: &[&str], +) -> ResponseEnvelope { + let mut keys = Vec::with_capacity(digests_wire.len()); + for digest_hex in digests_wire { + match digest_ingress(digest_hex) { + Ok(key) => keys.push(key), + Err(error) => return envelope_error(request_id, error), + } + } + let mut present = Vec::with_capacity(keys.len()); + for key in &keys { + present.push(ctx.store().has(&key.as_bytes()).await.unwrap_or(false)); + } + ResponseEnvelope::ok(request_id, json!({ "present": present })) +} + +/// **`blobs/delete`** `{namespace, digests[]}` — the operator's +/// correction flow (ops-surface.md §Op family; ADR-017's privilege +/// model): a **Mutation** with **`Visibility::Internal`** — not callable +/// from the wire (an external discovery answers the protocol's +/// `NOT_FOUND`, the visibility discipline's non-leakage form; only +/// internal composition reaches it) and **not namespace-gated** — the +/// `namespace` field exists for audit/triage scoping (which slice of the +/// pool the deletion targets), not for gating; the gate is the +/// **internal authority context**: the ACL check runs against the +/// composition authority the embedder declares at registration (alkcall +/// ADR-017 — internal calls switch authority context, never skip ACL). +/// +/// The handler rides `Store::delete` (the kv-large fall-through consult +/// then the full-protection-set arbitration; ADR-012 §6.1's general +/// rule): a deleted digest answers `deleted[]` in input order; a +/// protected key is the typed [`GC_REFUSED`] error (nothing deleted — +/// also the preceding digests in this call are the only ones ever +/// deleted: the path is fail-fast); an absent key is +/// [`BLOB_MISSING`]; the embedder's correction flow for a refusal is +/// deregisters/re-registers around the delete (ADR-012 §6.4). +pub fn delete_pair(ctx: Arc) -> (OperationSpec, Handler) { + let spec = OperationSpec::new( + "blobs/delete", + OperationType::Mutation, + Visibility::Internal, + json!({ + "type": "object", + "properties": { + "namespace": namespace_schema(), + "digests": digests_wire_schema() + }, + "required": ["namespace", "digests"], + "additionalProperties": false + }), + json!({ + "type": "object", + "properties": { "deleted": digests_wire_schema() }, + "required": ["deleted"], + "additionalProperties": false + }), + vec![ + declared_error( + BLOB_MISSING, + "the digest is absent: nothing deleted", + digest_details_schema(), + ), + declared_error( + GC_REFUSED, + "the digest is protected (pinned, liveness-live, or protect-callback-named): \ + nothing deleted", + digest_details_schema(), + ), + declared_error( + INVALID_DIGEST, + "the digest is not canonical digest hex (lowercase, 40 or 64 chars)", + digest_details_schema(), + ), + ], + AccessControl::default(), + None, + ) + .with_description("the operator's correction-flow deletion (internal authority context; failures are fail-fast)"); + let handler = make_handler(move |input, context| { + let ctx = Arc::clone(&ctx); + async move { + let request_id = context.request_id.clone(); + let Some(digests_wire) = json_string_array(&input, "digests") else { + return schema_validation_error(&request_id); + }; + let mut keys = Vec::with_capacity(digests_wire.len()); + for digest_hex in &digests_wire { + match digest_ingress(digest_hex) { + Ok(key) => keys.push(key), + Err(error) => return envelope_error(&request_id, error), + } + } + let mut deleted = Vec::with_capacity(digests_wire.len()); + for (key, digest_hex) in keys.iter().zip(&digests_wire) { + match ctx.store().delete(&key.as_bytes()).await { + Ok(()) => deleted.push(digest_hex.to_owned()), + Err(Error::Missing) => { + return envelope_error( + &request_id, + CallError::new( + BLOB_MISSING, + "the digest is absent: nothing deleted", + false, + ) + .with_details(json!({ "digest": digest_hex })), + ); + } + Err(Error::GcRefuse) => { + return envelope_error( + &request_id, + CallError::new( + GC_REFUSED, + "the digest is protected (pinned, liveness-live, or \ + protect-callback-named): nothing deleted", + false, + ) + .with_details(json!({ "digest": digest_hex })), + ); + } + Err(error) => return store_internal_error(&request_id, "delete", error), + } + } + ResponseEnvelope::ok(request_id, json!({ "deleted": deleted })) + } + }); + (spec, handler) +} + #[cfg(test)] mod tests { use super::*; @@ -935,7 +1426,7 @@ mod tests { } #[cfg(any(feature = "mem", feature = "sqlite"))] - async fn test_store() -> Arc { + pub(crate) async fn test_store() -> Arc { #[cfg(feature = "mem")] let store = Store::mem_only(); #[cfg(not(feature = "mem"))] @@ -1334,13 +1825,20 @@ mod harness { use alkcall::protocol::connection::{split_single_stream, CallConnection}; use alkcall::protocol::dispatch::Dispatcher; use alkcall::protocol::wire::ResponseEnvelope; + use alkcall::registry::context::{ + AbortPolicy, CompositionAuthority, OperationContext, ScopedPeerEnv, + }; + use alkcall::registry::env::OperationEnv; use alkcall::registry::registration::{make_handler, Handler, OperationRegistry}; use alkcall::registry::spec::{ AccessControl, ChannelOpenSpec, OperationSpec, OperationType, Visibility, }; use alkcall::OperationRegistryBuilder; + use serde_json::{json, Value}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use super::tests::test_store; + const DUPLEX_BUFFER: usize = 64 * 1024; const CALL_STREAMS: usize = 8; const TEST_TIMEOUT: Duration = Duration::from_secs(15); @@ -1518,6 +2016,46 @@ mod harness { open_handler: OpenHandler, } + /// A registration bundle the harness's assembly carries: the + /// spec/handler pair plus the registration-side authority/seam + /// declarations the embedder's assembly layer makes (ADR-017/ADR-019 + /// — the composition authority the internal ACL checks run under + /// and the scoped composition env the reachability gate reads). + /// The plain `assemble` passes `None`/`None` (the common external + /// pair). + struct OpsHarnessRegistration { + spec: OperationSpec, + handler: Handler, + composition_authority: Option, + scoped_env: Option, + } + + impl OpsHarnessRegistration { + fn pair(pair: (OperationSpec, Handler)) -> Self { + let (spec, handler) = pair; + OpsHarnessRegistration { + spec, + handler, + composition_authority: None, + scoped_env: None, + } + } + + fn with_composition_authority(self, authority: CompositionAuthority) -> Self { + OpsHarnessRegistration { + composition_authority: Some(authority), + ..self + } + } + + fn with_scoped_env(self, scoped_env: ScopedPeerEnv) -> Self { + OpsHarnessRegistration { + scoped_env: Some(scoped_env), + ..self + } + } + } + struct OpsHarness { registry: Arc, identity_provider: Arc, @@ -1530,11 +2068,28 @@ mod harness { /// assembly runs; the harness serves whatever is registered and /// nothing else. fn assemble(specs: Vec<(OperationSpec, Handler)>) -> OpsHarness { + Self::assemble_registrations( + specs + .into_iter() + .map(OpsHarnessRegistration::pair) + .collect(), + ) + } + + /// The assembly with the authority/seam declarations carried + /// (the delete tests' internal-authority and composition shapes). + fn assemble_registrations(entries: Vec) -> OpsHarness { let mut builder = OperationRegistryBuilder::new(); - for (spec, handler) in specs { + for entry in entries { builder = builder - .with_local(spec, handler, None, None, Capabilities::new()) - .expect("stub op registration compiles"); + .with_local( + entry.spec, + entry.handler, + entry.composition_authority, + entry.scoped_env, + Capabilities::new(), + ) + .expect("op registration compiles"); } OpsHarness { registry: Arc::new(builder.build()), @@ -1699,6 +2254,72 @@ mod harness { ) } + /// The composition parent for the delete tests: an External stub + /// op whose handler composes `blobs/delete` through the operation + /// env — the internal authority context's entry (the composed call + /// runs `internal: true`, ACL-checked against `blobs/delete`'s + /// declared composition authority — the switch, never the skip; + /// ADR-017). + fn composer_pair() -> (OperationSpec, Handler) { + ( + OperationSpec::new( + "blobs/compose-delete", + OperationType::Query, + Visibility::External, + serde_json::json!({"type": "object"}), + serde_json::json!({"type": "object"}), + Vec::new(), + AccessControl::default(), + None, + ), + make_handler(|input, context| async move { + let composed = context + .env + .invoke_with_policy("blobs", "delete", input, &context, context.abort_policy) + .await; + match composed.result { + Ok(output) => ResponseEnvelope::ok(context.request_id, output), + Err(error) => ResponseEnvelope::error(context.request_id, error), + } + }), + ) + } + + /// A bare `OperationContext` for direct handler invocation (the + /// handler-level ingress/arity tests — the shapes the wire's + /// schema gate would reject first): no identity, no env dispatch. + fn bare_context() -> OperationContext { + struct NoCompositionEnv; + #[async_trait::async_trait] + impl OperationEnv for NoCompositionEnv { + async fn invoke_with_policy( + &self, + namespace: &str, + operation: &str, + _input: Value, + _parent: &OperationContext, + _policy: AbortPolicy, + ) -> ResponseEnvelope { + ResponseEnvelope::not_found("test", &format!("{namespace}/{operation}")) + } + } + OperationContext { + request_id: "test-request".to_owned(), + parent_request_id: None, + identity: None, + handler_identity: None, + forwarded_for: None, + capabilities: Capabilities::new(), + metadata: std::collections::HashMap::new(), + scoped_env: ScopedPeerEnv::empty(), + env: Arc::new(NoCompositionEnv), + abort_policy: alkcall::registry::context::AbortPolicy::default(), + deadline: None, + internal: false, + ownership: None, + } + } + /// The stub channel-open op (the fetch shape: `Sub`, ADR-021/047): /// `channels/stub/sub` over the `alk/stub` data-plane ALPN. The /// establisher is the awaited Establishment phase — it records its @@ -2010,4 +2631,642 @@ mod harness { "no channel opened; the pump never saw bytes" ); } + + // ---------------------------------------------------------------- + // The JSON control plane's spec/handler pairs (ops-json-ops): the + // stat/have/delete families through the assembled registry. + // ---------------------------------------------------------------- + + #[cfg(any(feature = "mem", feature = "sqlite"))] + async fn json_plane(store: Arc) -> (OpsHarness, Arc) { + let ctx = Arc::new(OpsContext::new(store, OpsConfig::default())); + let harness = OpsHarness::assemble(vec![ + stat_pair(Arc::clone(&ctx)), + have_pair(Arc::clone(&ctx)), + ]); + (harness, ctx) + } + + fn call_payload(op: &str, input: Value, auth_token: Option<&str>) -> Value { + let mut payload = json!({ + "operationId": op, + "input": input, + }); + if let Some(token) = auth_token { + payload["auth_token"] = json!(token); + } + payload + } + + fn field_str<'a>(envelope: &'a ResponseEnvelope, code_field: &str) -> Option<&'a str> { + match &envelope.result { + Err(error) => error + .details + .as_ref() + .and_then(|details| details.get(code_field)) + .and_then(Value::as_str), + Ok(_) => None, + } + } + + fn error_code(envelope: &ResponseEnvelope) -> Option<&str> { + match &envelope.result { + Err(error) => Some(error.code.as_str()), + Ok(_) => None, + } + } + + #[cfg(any(feature = "mem", feature = "sqlite"))] + async fn seeded_entry(store: &Arc, seed: &[u8]) -> (String, Key) { + let len = seed.len() as u64; + let (key, _pin) = store + .put(Some(len), std::io::Cursor::new(seed)) + .await + .expect("seed put lands"); + let digest_hex = hex::encode(key.digest().bytes()); + (digest_hex, key) + } + + /// `blobs/stat` (criterion: the stat round-trip + the miss shape): + /// a granted identity receives `{len}` for a seeded entry without a + /// content load (the read action's probe-before-offering shape); a + /// miss is the typed `BLOB_MISSING` error (not a `{len: null}`); + /// a malformed digest is rejected at the schema (`INVALID_INPUT`). + #[cfg(any(feature = "mem", feature = "sqlite"))] + #[tokio::test] + async fn stat_answers_len_the_miss_is_typed_and_the_digest_ingress_rejects() { + let store = test_store().await; + let (harness, _ctx) = json_plane(Arc::clone(&store)).await; + harness.with_token("token:alice", identity_named("alice")); + harness.grant("alice", "ns-1", &["read"]); + let client = harness.spawn_call_halves(); + + let (digest_hex, _) = seeded_entry(&store, b"stat round trip").await; + + let answer = await_call(client.call_with_payload(call_payload( + "blobs/stat", + json!({ "namespace": "ns-1", "digest": digest_hex }), + Some("token:alice"), + ))) + .await; + assert_eq!( + answer.result.unwrap(), + json!({ "len": 15 }), + "the seeded entry's length answers" + ); + + let missed = await_call(client.call_with_payload(call_payload( + "blobs/stat", + json!({ "namespace": "ns-1", "digest": "a".repeat(64) }), + Some("token:alice"), + ))) + .await; + assert_eq!(error_code(&missed), Some(BLOB_MISSING)); + assert_eq!( + field_str(&missed, "digest").map(str::to_owned), + Some("a".repeat(64)), + "the missing-error's details carry the digest" + ); + + let malformed = await_call(client.call_with_payload(call_payload( + "blobs/stat", + json!({ "namespace": "ns-1", "digest": "ZZZ-not-hex" }), + Some("token:alice"), + ))) + .await; + assert_eq!( + error_code(&malformed), + Some("INVALID_INPUT"), + "the schema gate rejects a malformed digest before the handler" + ); + } + + /// `blobs/have` without the token (criterion: present/absent mixed + /// by index correspondence) and the read gate (closed-by-default): + /// an unlisted namespace's probe denies, the granted identity reads. + #[cfg(any(feature = "mem", feature = "sqlite"))] + #[tokio::test] + async fn have_answers_by_index_and_the_unlisted_namespace_denies() { + let store = test_store().await; + let (harness, _ctx) = json_plane(Arc::clone(&store)).await; + harness.with_token("token:alice", identity_named("alice")); + harness.grant("alice", "ns-1", &["read"]); + let client = harness.spawn_call_halves(); + + let (digest_hex, _) = seeded_entry(&store, b"have probe").await; + let absent = hex::encode(Digest::derive(b"never put", 9).bytes()); + + let mixed = await_call(client.call_with_payload(call_payload( + "blobs/have", + json!({ "namespace": "ns-1", "digests": [digest_hex.clone(), absent.clone(), digest_hex.clone()] }), + Some("token:alice"), + ))) + .await; + assert_eq!( + mixed.result.unwrap(), + json!({ "present": [true, false, true] }), + "present[] corresponds to digests[] by index" + ); + + let denied = await_call(client.call_with_payload(call_payload( + "blobs/have", + json!({ "namespace": "ns-2-the-unlisted", "digests": [digest_hex.clone()] }), + Some("token:alice"), + ))) + .await; + assert_eq!( + error_code(&denied), + Some("FORBIDDEN"), + "closed-by-default: a grant per namespace is the deferred cost" + ); + } + + /// `blobs/have`'s renewal form (criterion: the three answers, the + /// non-fleet arms): a live custodial row renews (`present: true`), + /// an unknown token answers `present: false` (never an error), a + /// mismatched digest answers `present: false`, and a malformed + /// token's ingress answers the unknown-token form too. The arity + /// violation is the typed `INVALID_RENEWAL_FORM` — exercised at the + /// handler's ingress (the wire's schema gate would reject first). + #[cfg(any(feature = "mem", feature = "sqlite"))] + #[tokio::test] + async fn the_renewal_form_answers_and_the_ingress_rejects_the_arity_violation() { + let store = test_store().await; + let (harness, ctx) = json_plane(Arc::clone(&store)).await; + harness.with_token("token:alice", identity_named("alice")); + harness.grant("alice", "ns-1", &["read"]); + let client = harness.spawn_call_halves(); + + let (digest_hex, key) = seeded_entry(&store, b"renewal form").await; + let other = Key::from(Digest::derive(b"another entry", 13)); + let pin = store.gc().acquire(&key).await; + let token = ctx.mint_custodial("ns-1", key.clone(), pin).unwrap(); + let unknown_token = PinToken::generate().unwrap(); + + let renewed = await_call(client.call_with_payload(call_payload( + "blobs/have", + json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": token.to_hex() }), + Some("token:alice"), + ))) + .await; + assert_eq!( + renewed.result.unwrap(), + json!({ "present": [true] }), + "a live custodial row renews from the held guard (nothing expires)" + ); + + let refused = await_call(client.call_with_payload(call_payload( + "blobs/have", + json!({ "namespace": "ns-1", "digests": [hex::encode(other.digest().bytes())], "token": token.to_hex() }), + Some("token:alice"), + ))) + .await; + assert_eq!( + refused.result.unwrap(), + json!({ "present": [false] }), + "a digest the token does not bind: the one-token-one-row check, present: false" + ); + let foreign = await_call(client.call_with_payload(call_payload( + "blobs/have", + json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": unknown_token.to_hex() }), + Some("token:alice"), + ))) + .await; + assert_eq!( + foreign.result.unwrap(), + json!({ "present": [false] }), + "a token this node did not mint: the validity-domain refusal, present: false, not an error" + ); + + let malformed_token = hex::encode(b"not-a-64-char-canonical-hex-token"); + let malformed = await_call(client.call_with_payload(call_payload( + "blobs/have", + json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": malformed_token }), + Some("token:alice"), + ))) + .await; + assert_eq!( + malformed.result.unwrap(), + json!({ "present": [false] }), + "a malformed token is an unknown token (present: false, never an error)" + ); + + let arity_wire = await_call(client.call_with_payload(call_payload( + "blobs/have", + json!({ "namespace": "ns-1", "digests": [digest_hex.clone(), "0".repeat(64)], "token": token.to_hex() }), + Some("token:alice"), + ))) + .await; + assert_eq!( + error_code(&arity_wire), + Some("INVALID_INPUT"), + "the schema gate enforces the renewal form's one-digest arity (if/then) at the wire" + ); + + let (_, handler) = have_pair(Arc::clone(&ctx)); + let arity = (handler)( + json!({ "namespace": "ns-1", "digests": [digest_hex.clone(), "0".repeat(64)], "token": token.to_hex() }), + bare_context(), + ) + .await; + assert_eq!( + error_code(&arity), + Some(INVALID_RENEWAL_FORM), + "the renewal form's one-digest arity is the handler's typed error" + ); + + let invalid = (handler)( + json!({ "namespace": "ns-1", "digests": ["NOT-DIGEST-HEX"], "token": token.to_hex() }), + bare_context(), + ) + .await; + assert_eq!( + error_code(&invalid), + Some(INVALID_DIGEST), + "the digest ingress rejects, never guesses" + ); + } + + /// `blobs/delete` (criterion: external discovery NOT_FOUND; the + /// internal operator-context composition deletes; the namespace + /// field is audit scoping, not gating — a grantless caller's + /// composed internal call runs under the operator authority): + /// deleted, fail-fast on the absent key, and `GcRefuse` as the + /// typed `GC_REFUSED`. + #[cfg(any(feature = "mem", feature = "sqlite"))] + #[tokio::test] + async fn delete_is_external_not_found_and_the_internal_authority_composition_deletes() { + let store = test_store().await; + let ctx = Arc::new(OpsContext::new(store, OpsConfig::default())); + let harness = OpsHarness::assemble_registrations(vec![ + OpsHarnessRegistration::pair(delete_pair(Arc::clone(&ctx))) + .with_composition_authority(CompositionAuthority::new("blob-operator", [])), + OpsHarnessRegistration::pair(composer_pair()) + .with_scoped_env(ScopedPeerEnv::new(["blobs/delete"])), + ]); + harness.with_token("token:stranger", identity_named("stranger")); + let client = harness.spawn_call_halves(); + + let external = await_call(client.call_with_payload(call_payload( + "blobs/delete", + json!({ "namespace": "ns-1", "digests": ["0".repeat(64)] }), + Some("token:stranger"), + ))) + .await; + assert_eq!( + error_code(&external), + Some("NOT_FOUND"), + "an Internal op is not discovered from the wire (the visibility discipline)" + ); + + let (digest_hex, key) = seeded_entry(ctx.store(), b"deleted by the operator").await; + + let composed = await_call(client.call_with_payload(call_payload( + "blobs/compose-delete", + json!({ "namespace": "ns-1", "digests": [digest_hex.clone()] }), + Some("token:stranger"), + ))) + .await; + assert_eq!( + composed.result.unwrap(), + json!({ "deleted": [digest_hex.clone()] }), + "the operator-authority composition deletes; the namespace grant is not the gate" + ); + assert!( + !ctx.store().has(&key.as_bytes()).await.unwrap_or(true), + "the entry is gone from the facade after the internal delete" + ); + + let (present_hex, present_key) = seeded_entry(ctx.store(), b"fail fast").await; + let absent = hex::encode(Digest::derive(b"absent entry", 12).bytes()); + let fail_fast = await_call(client.call_with_payload(call_payload( + "blobs/compose-delete", + json!({ "namespace": "ns-1", "digests": [present_hex, absent.clone()] }), + Some("token:stranger"), + ))) + .await; + assert_eq!( + error_code(&fail_fast), + Some(BLOB_MISSING), + "the absent key is the typed missing error" + ); + assert_eq!( + field_str(&fail_fast, "digest").map(str::to_owned), + Some(absent), + "the missing-error's details carry the failing digest" + ); + assert!( + !ctx.store() + .has(&present_key.as_bytes()) + .await + .unwrap_or(true), + "the fail-fast path's preceding digests deleted (nothing atomic was specced)" + ); + + let (pinned_hex, pinned_key) = seeded_entry(ctx.store(), b"protected entry").await; + let _guard = ctx.store().gc().acquire(&pinned_key).await; + let refused = await_call(client.call_with_payload(call_payload( + "blobs/compose-delete", + json!({ "namespace": "ns-1", "digests": [pinned_hex] }), + Some("token:stranger"), + ))) + .await; + assert_eq!( + error_code(&refused), + Some(GC_REFUSED), + "a protected key's refusal is the typed error" + ); + assert_eq!( + field_str(&refused, "digest").map(str::to_owned), + Some(pinned_hex), + "the refusal's details carry the protected digest" + ); + } + + // ---------------------------------------------------------------- + // The renewal form's fleet arms (the foundation's noted fleet + // forms) — against the live harness server (the fleet column's + // SKIP posture adapted in-crate): env-driven DSN, skipping cleanly + // server-less, acceptance green against the server. + // ---------------------------------------------------------------- + + /// The harness-server DSN (`ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB` + /// — the alkstore convention; host gated, the rest defaulting to + /// the harness server's values). `None` → the caller skips. + #[cfg(feature = "postgres")] + async fn fleet_dsn() -> Option { + let host = std::env::var("ALKBLOBS_PG_HOST") + .ok() + .filter(|v| !v.is_empty())?; + let port = std::env::var("ALKBLOBS_PG_PORT") + .ok() + .filter(|v| !v.is_empty()) + .unwrap_or_else(|| "15432".to_owned()); + let user = std::env::var("ALKBLOBS_PG_USER") + .ok() + .filter(|v| !v.is_empty()) + .unwrap_or_else(|| "postgres".to_owned()); + let password = std::env::var("ALKBLOBS_PG_PASSWORD") + .ok() + .filter(|v| !v.is_empty()) + .unwrap_or_else(|| "poc".to_owned()); + let db = std::env::var("ALKBLOBS_PG_DB") + .ok() + .filter(|v| !v.is_empty()) + .unwrap_or_else(|| "blobs".to_owned()); + let dsn = format!("host={host} port={port} user={user} password={password} dbname={db}"); + match tokio::time::timeout(std::time::Duration::from_secs(3), async { + let (client, connection) = tokio_postgres::connect(&dsn, tokio_postgres::NoTls) + .await + .map_err(|e| Error::Io(format!("postgres harness: {e}")))?; + tokio::spawn(async move { + let _ = connection.await; + }); + Ok::(client) + }) + .await + { + Ok(Ok(_)) => Some(dsn), + Ok(Err(e)) => { + eprintln!("skipping (harness server unreachable): {e}"); + None + } + Err(_) => { + eprintln!("skipping (harness server unreachable: connect timeout)"); + None + } + } + } + + /// The pin-row instrument (the fleet column's medium-oracle + /// pattern, the have handler's renewal arm's observation): one + /// independent connection reads the schema's + /// `gc_pins(key, owner, expiry)` — the durable pin state — for one + /// key (`(owner, expiry)` per row; nanos are absolute wall time). + #[cfg(feature = "postgres")] + async fn fleet_pin_row( + dsn: &str, + schema: &str, + key: &[u8], + ) -> Result, i64)>, Error> { + let (client, connection) = tokio_postgres::connect(dsn, tokio_postgres::NoTls) + .await + .map_err(|e| Error::Io(format!("postgres harness: {e}")))?; + tokio::spawn(async move { + let _ = connection.await; + }); + let sql = format!( + "SELECT owner, expiry FROM \"{}\".gc_pins WHERE key = $1", + schema.replace('"', "\"\"") + ); + let row = client + .query_opt(&sql, &[&key]) + .await + .map_err(|e| Error::Io(format!("postgres harness: {e}")))?; + Ok(row.map(|row| { + let owner: Vec = row.get(0); + let expiry: i64 = row.get(1); + (owner, expiry) + })) + } + + #[cfg(feature = "postgres")] + async fn fleet_schema_drop(dsn: &str, schema: &str) { + if let Ok((admin, connection)) = tokio_postgres::connect(dsn, tokio_postgres::NoTls).await { + tokio::spawn(async move { + let _ = connection.await; + }); + let sql = format!( + "DROP SCHEMA IF EXISTS \"{}\" CASCADE", + schema.replace('"', "\"\"") + ); + let _ = admin.batch_execute(&sql).await; + } + } + + /// Two fleet-declared dual-tier stores over one freshly minted + /// schema (the fleet column's factory pattern, the have handler's + /// renewal arm's shape): the serving instance A mints the token, + /// the peer B plays the foreign mint — each with its own + /// store-instance id (the pin rows compose by owner) and the short + /// test pin TTL (the expiry horizon's observable cadence). + /// Returns the two stores; the caller owns the temp root and the + /// schema's teardown. + #[cfg(feature = "postgres")] + async fn open_fleet_pair( + dsn: &str, + schema: &str, + root_a: &std::path::Path, + root_b: &std::path::Path, + ) -> Result<(Store, Store), Error> { + let ttl = std::time::Duration::from_millis(900); + let opts = |schema: &str| crate::PgOpts { + schema: schema.to_owned(), + ..crate::PgOpts::default() + }; + let store_a = crate::Store::dual_tier_postgres_local_fleet( + dsn, + opts(schema), + root_a.to_path_buf(), + 1024, + std::time::Duration::ZERO, + ttl, + crate::FleetSharedMedia::asserted(), + ) + .await?; + let store_b = crate::Store::dual_tier_postgres_local_fleet( + dsn, + opts(schema), + root_b.to_path_buf(), + 1024, + std::time::Duration::ZERO, + ttl, + crate::FleetSharedMedia::asserted(), + ) + .await?; + Ok((store_a, store_b)) + } + + /// The have handler's renewal form against a **live fleet store** + /// (the foundation's noted fleet forms; ADR-012 §2's three-step + /// duty on the wire): the fleet token renews the live row + /// (`present: true` — the row's expiry moves, observed by the + /// independent connection's `gc_pins` read); a putter pausing past + /// the TTL answers `present: false` and **no renewal occurs** (the + /// expiry frozen at the last renewal — the never-resurrect + /// discipline, ADR-008 §2); a peer's foreign mint answers + /// `present: false`, not an error — the validity domain is the + /// minting node (ADR-012 §2). The custodial-mint-is-refused-under- + /// fleet arm rides the same contexts. + #[cfg(feature = "postgres")] + #[tokio::test] + async fn the_fleet_renewal_form_renews_the_live_row_and_never_resurrects_the_expired_one() { + let Some(dsn) = fleet_dsn().await else { + eprintln!( + "skipping (no postgres harness server): set ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB" + ); + return; + }; + let schema = format!( + "alkblobs_ops_have_renewal_{}_{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + ); + let root = std::env::temp_dir().join(format!( + "alkblobs-ops-fleet-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + )); + std::fs::create_dir_all(&root).expect("the temp root created"); + + let (store_a, store_b) = open_fleet_pair(&dsn, &schema, &root.join("a"), &root.join("b")) + .await + .expect("the fleet pair constructs against the harness server"); + let store_a = Arc::new(store_a); + let store_b = Arc::new(store_b); + let ctx = Arc::new(OpsContext::new(Arc::clone(&store_a), OpsConfig::default())); + let peer = Arc::new(OpsContext::new(Arc::clone(&store_b), OpsConfig::default())); + + let harness = OpsHarness::assemble(vec![have_pair(Arc::clone(&ctx))]); + harness.with_token("token:alice", identity_named("alice")); + harness.grant("alice", "ns-1", &["read"]); + let client = harness.spawn_call_halves(); + + let seed: &[u8] = b"fleet renewal"; + let (key, pin_guard) = store_a + .put(Some(seed.len() as u64), std::io::Cursor::new(seed)) + .await + .expect("the fleet put commits the pin row"); + // The guard's drop at response completion is the put handler's + // own act (ADR-010 §4) — from here the pin row's expiry is the + // protection, and the TTL/3 renewal duty has stopped. + drop(pin_guard); + let digest_hex = hex::encode(key.digest().bytes()); + let owner = ctx + .fleet_owner() + .expect("the fleet-declared store's instance id"); + let token = ctx + .mint_fleet("ns-1", key.clone(), &owner) + .expect("the fleet mint over the serving instance"); + let minted_expiry = fleet_pin_row(&dsn, &schema, &key.as_bytes()) + .await + .expect("the row instrument") + .expect("the put's pin row"); + + let custodial_refused: Pin = store_b.gc().acquire(&key).await; + assert!( + peer.mint_custodial("ns-1", key.clone(), custodial_refused) + .is_err(), + "the custodial mint is refused on a fleet-declared store (the row is the protection)" + ); + + let renewed = await_call(client.call_with_payload(call_payload( + "blobs/have", + json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": token.to_hex() }), + Some("token:alice"), + ))) + .await; + assert_eq!( + renewed.result.expect("the renewal probe answers"), + json!({ "present": [true] }), + "the live row renews on the wire" + ); + let renewed_expiry = fleet_pin_row(&dsn, &schema, &key.as_bytes()) + .await + .expect("the row instrument") + .expect("the row"); + assert!( + renewed_expiry.1 > minted_expiry.1, + "the renewal moved the row's expiry forward (minted {}, renewed {})", + minted_expiry.1, + renewed_expiry.1 + ); + assert_eq!( + renewed_expiry.0, owner, + "the renewed row is the minting instance's" + ); + + let foreign = await_call(client.call_with_payload(call_payload( + "blobs/have", + json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": PinToken::generate().expect("the CSPRNG minted").to_hex() }), + Some("token:alice"), + ))) + .await; + assert_eq!( + foreign.result.expect("the renewal probe answers"), + json!({ "present": [false] }), + "a token this node did not mint: the validity-domain refusal, present: false, not an error" + ); + + tokio::time::sleep(std::time::Duration::from_millis(1400)).await; + let expired = await_call(client.call_with_payload(call_payload( + "blobs/have", + json!({ "namespace": "ns-1", "digests": [digest_hex.clone()], "token": token.to_hex() }), + Some("token:alice"), + ))) + .await; + assert_eq!( + expired.result.expect("the renewal probe answers"), + json!({ "present": [false] }), + "a putter paused past the TTL: the row expired, present: false, no renewal" + ); + let expired_expiry = fleet_pin_row(&dsn, &schema, &key.as_bytes()) + .await + .expect("the row instrument"); + assert!( + expired_expiry.is_none() || expired_expiry.map(|row| row.1) == Some(renewed_expiry.1), + "the expired probe renewed nothing: the row's expiry froze at the last renewal, \ + and the row is gone only when a sweep's maintenance step reaped it" + ); + + fleet_schema_drop(&dsn, &schema).await; + let _ = std::fs::remove_dir_all(&root); + let _ = (store_a, store_b, ctx, peer); + } } diff --git a/tasks/ops-json-ops.md b/tasks/ops-json-ops.md index d2ccf30..87b2526 100644 --- a/tasks/ops-json-ops.md +++ b/tasks/ops-json-ops.md @@ -1,7 +1,7 @@ --- id: ops-json-ops name: the JSON control plane — blobs/stat, blobs/have (+ the token-renewal form), blobs/delete -status: pending +status: completed depends_on: [ops-foundation, ops-call-harness] scope: moderate risk: medium @@ -73,20 +73,20 @@ an authorization check): ## Acceptance Criteria -- [ ] Three spec/handler pairs exported (stat, have, delete); input/ +- [x] Three spec/handler pairs exported (stat, have, delete); input/ output schemas + typed error schemas per alkcall ADR-016 -- [ ] The ACL mapping: namespace-as-resource with the +- [x] The ACL mapping: namespace-as-resource with the resource_id_path; the read action on stat/have; closed-by-default verified by test (unlisted namespace denies) -- [ ] delete: `Visibility::Internal`, the internal-authority-context +- [x] delete: `Visibility::Internal`, the internal-authority-context requirement (not namespace-gated), external discovery NOT_FOUND -- [ ] have's renewal form: one digest + token; verify-mapping → +- [x] have's renewal form: one digest + token; verify-mapping → `pin_renew`; expired/foreign → `present: false` with no renewal; the non-fleet form answers from the held custody -- [ ] The handlers touch only the store facade + the foundation's +- [x] The handlers touch only the store facade + the foundation's crate-internal renewal access (no backend side-door — the ADR-001 §4 rule) -- [ ] Harness tests green; `cargo test` (default) green; +- [x] Harness tests green; `cargo test` (default) green; `cargo test --features ops` green; clippy/fmt green on both shapes; `cargo doc --no-deps` clean @@ -112,8 +112,98 @@ an authorization check): ## Notes -> To be filled by implementation agent +> Landed in `src/ops.rs` (the ops module's "JSON control plane" +> section, appended to the foundation's substrate) + `src/lib.rs` +> re-exports (`stat_pair`/`have_pair`/`delete_pair` and the four +> domain codes `BLOB_MISSING`/`INVALID_DIGEST`/`INVALID_RENEWAL_FORM`/ +> `GC_REFUSED`). Each export is `pub fn X_pair(ctx: Arc) -> +> (OperationSpec, Handler)` — the embedder assembles; the harness's +> assembly (extended: `assemble_registrations` now carries the +> registration-side composition authority and scoped composition env, +> ADR-017/019 — the pair-level `assemble` shape is unchanged). +> +> **Schemas (alkcall ADR-016):** input/output JSON Schema per op — the +> namespace field's syntax enforced at the schema (`minLength: 1`, the +> non-empty-UTF-8 rule); the digest wire form hex with the canonical +> pattern (`^[0-9a-f]{40}$|^[0-9a-f]{64}$`, the key layer's +> lowercase-only ingress also enforced at the handlers — +> rejects-never-guesses). The have schema carries the renewal form's +> arity as `if`/`then` (token ⇒ digests minItems/maxItems 1) — verified +> live at the wire (`INVALID_INPUT` on the two-digest+token shape) — +> and re-enforced at the handler (`INVALID_RENEWAL_FORM`, the +> typed error for non-registry dispatch paths, exercised by direct +> handler invocation). Typed error schemas: `BLOB_MISSING` +> (stat's miss — details `{digest}`; a `{len: null}` was never an +> option), `INVALID_DIGEST` (the ingress), `INVALID_RENEWAL_FORM` +> (have's arity), `GC_REFUSED` (delete's `Error::GcRefuse`). +> +> **The ACL mapping (OQ-09's resolution, verified by test):** +> namespace-as-resource — `resource_type: "blob-namespace"`, +> `resource_id_path: "$.namespace"`, the read action on stat/have; +> closed-by-default proven on the wire (an unlisted namespace denies +> the granted identity's probe; a granted namespace reads; the +> anonymous call answers the authentication required gate). delete: +> `Visibility::Internal`, `AccessControl` left un-gated (not +> namespace-gated — the field is audit/triage scoping), the +> operator-admin authority context realized as the **composition +> authority the embedder declares at registration** (the pair exports +> it authority-agnostic); external discovery is the registry's +> NOT_FOUND (tested); the internal path proven through a composant op +> (`blobs/compose-delete` test stub, scoped env → delete) whose +> grantless caller composes — the check ran against the declared +> operator authority, the namespace grant not the gate. GcRefuse and +> Missing surface as the typed errors (fail-fast; no atomic multi- +> delete was specced — the preceding digests' deletions stand, and the +> doc says so). +> +> **The renewal form** (ADR-012 §2's three-step duty, riding the +> foundation's `OpsContext::renew` — no new store access, no backend +> side-door): the non-fleet arms answer from custody (renew → +> present: true; unknown/foreign/mismatched → present: false, never an +> error; the malformed-token ingress maps to the unknown-token answer); +> the fleet arms ran against the harness server (env-driven +> `ALKBLOBS_PG_HOST/...`, skipping cleanly server-less — the fleet +> column's SKIP posture adapted in-crate): the live row renews with +> the row's expiry observed moving (the independent `gc_pins` +> instrument), the expired row answers `present: false` with the +> expiry frozen at the last renewal (no renewal occurred — the +> never-resurrect discipline), and the peer mint's foreign token +> answers `present: false` the same way (the validity domain = the +> minting node). The foundation's noted `mint_custodial`-refused-under- +> fleet arm is asserted in the same test. The put-side arms (mint the +> token from a wire-carried put) ride the put's handler (ops-put). ## Summary -> To be filled on completion \ No newline at end of file +> Completed 2026-10-13 (wave 7). The JSON control plane's three +> spec/handler pairs landed in `src/ops.rs` under feature `ops`: +> `blobs/stat` (Query/read — `Store::stat`, the miss as the typed +> `BLOB_MISSING`), `blobs/have` (Query/read — the pure existence probe +> over `Store::has` with index-corresponded `present[]`, engine +> failures folding to present: false, and the token-renewal form +> riding `OpsContext::renew` with the one-digest arity enforced at the +> schema (`if`/`then`) and the handler), and `blobs/delete` +> (Mutation/`Visibility::Internal` — external discovery NOT_FOUND; the +> operator authority context via the registration-time composition +> authority; not namespace-gated; `Store::delete` with `GC_REFUSED`/ +> `BLOB_MISSING` as the typed errors, fail-fast). ACL: namespace-as- +> resource, the read action, closed-by-default (verified by test); +> digests wire-form hex parsed via the key layer's ingress +> (rejects-never-guesses); domain error codes `BLOB_MISSING`/ +> `INVALID_DIGEST`/`INVALID_RENEWAL_FORM`/`GC_REFUSED` declared per op +> with their details schemas. Five harness tests added (stat +> round-trip/miss/ingress, have's index correspondence + closed-by- +> default matrix, the renewal form's arms incl. the arity gate at both +> the wire and the handler, delete's NOT_FOUND + internal composition +> + refusal arms) plus one fleet harness-server test (the three +> renewal answers with the row-expiry instrument; the foreign token; +> the custodial-mint refusal under fleet) — env-driven, skipping +> cleanly server-less (SKIP posture). Gates green: `cargo test` +> (default 73/19/3), `cargo test --features ops` (92/19/3), +> `cargo test --all-features` against the harness server (246/125/4 — +> one pre-existing fleet-timing row flaked once under full-suite load +> and passed on the isolated + full reruns, untouched by this task's +> diff: the fleet column reads no ops code), clippy +> `--all-targets -- -D warnings` on the default/ops/all-features +> shapes plus the pg-pure checks, `cargo fmt --check`, +> `cargo doc --no-deps` clean on the default and ops builds. \ No newline at end of file