diff --git a/Cargo.lock b/Cargo.lock index a58410f1..e0250f72 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2271,7 +2271,7 @@ dependencies = [ [[package]] name = "roas" -version = "0.20.0" +version = "0.20.1" dependencies = [ "clap", "enumset", @@ -2302,7 +2302,9 @@ dependencies = [ "enumset", "regex", "reqwest", + "roas", "roas-arazzo", + "roas-asyncapi", "serde_json", "serde_json_path", "serde_yaml_ng", @@ -2364,7 +2366,7 @@ dependencies = [ [[package]] name = "roas-http-fetcher" -version = "0.2.4" +version = "0.2.5" dependencies = [ "reqwest", "roas", diff --git a/Cargo.toml b/Cargo.toml index 914fa800..d28094e7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,6 +10,8 @@ homepage = "https://github.com/sv-tools/roas" repository = "https://github.com/sv-tools/roas" [workspace.dependencies] +roas = { version = "0.20.1", path = "crates/roas", default-features = false } +roas-asyncapi = { version = "0.4", path = "crates/roas-asyncapi", default-features = false } actix-web = { version = "4.15", default-features = false } anyhow = "1.0.99" axum = "0.8" diff --git a/crates/roas-arazzo-executor/Cargo.toml b/crates/roas-arazzo-executor/Cargo.toml index 0c9f95cd..b4adcc19 100644 --- a/crates/roas-arazzo-executor/Cargo.toml +++ b/crates/roas-arazzo-executor/Cargo.toml @@ -24,9 +24,13 @@ v1_0 = ["roas-arazzo/v1_0"] # Ready-made HTTP clients built on `reqwest` — blocking and async. # Without this the caller supplies its own `HttpClient`. reqwest = ["dep:reqwest", "dep:tokio"] +# Complete source registries and graph loading through caller-configured roas loaders. +source-graph = ["dep:roas", "dep:roas-asyncapi", "roas-arazzo/v1_0"] [dependencies] enumset.workspace = true +roas = { workspace = true, optional = true, features = ["v2", "v3_0", "v3_1", "v3_2"] } +roas-asyncapi = { workspace = true, optional = true, features = ["v2_6", "v3_0", "v3_1"] } roas-arazzo = { version = "0.3", path = "../roas-arazzo", features = ["v1_1"] } regex.workspace = true reqwest = { workspace = true, optional = true } diff --git a/crates/roas-arazzo-executor/README.md b/crates/roas-arazzo-executor/README.md index 5ec24ff8..064c2512 100644 --- a/crates/roas-arazzo-executor/README.md +++ b/crates/roas-arazzo-executor/README.md @@ -49,6 +49,78 @@ The engine decides *what* to send and asks a client to send it. That is what let Source descriptions are the same story: fetching them is IO, so the caller passes the parsed documents to `Options::source`. [`roas-file-fetcher`](https://crates.io/crates/roas-file-fetcher) and [`roas-http-fetcher`](https://crates.io/crates/roas-http-fetcher) do that job for the loader and do it here just as well. +### Document identities and source graphs + +The optional `source-graph` feature adds `SourceRegistry` and bounded sync/async +traversal through a caller-configured `roas::loader::Loader`. It registers no file +or network fetchers itself. The existing `Options::source` API remains available +without this feature. + +```rust,no_run +use roas::loader::Loader; +use roas_arazzo_executor::{Options, SourceLoadOptions, SourceRegistry, prepare}; +use serde_json::Value; + +# fn example(root_json: Value, loader: &mut Loader) -> Result<(), Box> { +let mut registry = SourceRegistry::new(); +let root = registry.insert("https://example.test/workflows/root.json", root_json)?; +// Insert every other supplied document here, before loading any links. +let loading = registry.load_sources(root, loader, &SourceLoadOptions::default())?; +for diagnostic in &loading.diagnostics { + eprintln!("{}: {diagnostic}", registry.document(diagnostic.owner)?.identity()); +} +let options = Options::new().source_registry(®istry, root)?; +let description = registry.document(root)?.arazzo().expect("an Arazzo root"); +let plan = prepare(description, &options)?; +// plan.execute(&mut client), or plan.execute_async(&mut client).await +# Ok(()) } +``` + +Documents retain their original value, retrieval URI, canonical identity, effective +reference base, and written version. Arazzo is deserialized in full before its +references are resolved. A relative `$self` resolves against the retrieval URI +(the final redirect location when exposed by the fetcher); document references +then resolve against that identity. `$self` fragments and conflicting documents +claiming the same identity/location are rejected. URI normalization removes +fragments for document lookup and handles dot segments/default ports; queries +remain distinct. It does not canonicalize filesystem symlinks or all percent escapes. + +Canonical Arazzo identities follow +[identity-based referencing](https://spec.openapis.org/arazzo/v1.1.0.html#identity-based-referencing). +An Arazzo retrieval URL different from its `$self` is accepted only with +`SourceLoadOptions::retrieval_aliases = true`, a compatibility extension. Explicit +`override_source(owner, name, target)` is also available. Aliases and +`override_base_url` are scoped to the owning document; identical names in different +documents never overwrite each other. Explicit `Options::source` / `base_url` +entries win over the registry adapter. `Options::source_document` exposes the +metadata of registry-backed sources, and returns `None` for legacy sources. +Registry-backed options (including cloned options and source aliases) share the +loader's immutable raw value. They do not keep another full JSON copy. Arazzo also +has its parsed typed model; API documents remain raw values with checked versions. + +Cycles are retained as back edges, not recursively expanded documents. Shared +dependencies reuse handles and loaded resources. The default limits are 256 +existing documents plus distinct loader attempts, and depth 32 (root depth zero). +Failed attempts and different retrieval aliases also consume the document budget; +known cycles/diamonds do not consume additional depth. `root_sources` selects root +aliases; linked Arazzo documents are traversed in full. These limits are independent +of workflow step/retry/call-depth limits. Supplied documents must be inserted before +loading; the caller is responsible for bounding those inputs and response sizes. + +Loading failures carry owner/alias/field locations and leave readable documents +available. Preparation decides whether that partial graph is sufficient: an +unrelated missing source need not block a qualified operation, but a missing +candidate source still prevents proving a bare `operationId` unique. Loading does +not silently certify a partial graph as complete. + +Recognized versions are Arazzo 1.0/1.1, OpenAPI 2.0/3.0/3.1/3.2, and AsyncAPI +2.6/3.0/3.1. Arazzo 1.0 retains its wire version and is upconverted for execution. +API documents retain complete raw values and model-checked versions; loading is +**not** API structural or schema validation. AsyncAPI loading does not enable +broker execution. Cross-document workflow execution, external OpenAPI Path Item +resolution, and relative API-server computation are not added by this feature. +Document bases are distinct from API endpoint overrides. + ## Testing a workflow `testing::Fake` answers from a script and keeps what it was asked, so a workflow can be tested without a server: diff --git a/crates/roas-arazzo-executor/src/lib.rs b/crates/roas-arazzo-executor/src/lib.rs index 125ed52d..b090ded5 100644 --- a/crates/roas-arazzo-executor/src/lib.rs +++ b/crates/roas-arazzo-executor/src/lib.rs @@ -28,6 +28,10 @@ //! Source descriptions are loaded the same way: fetching them is IO, so //! the caller supplies the parsed documents through //! [`Options::source`]. +//! The optional `source-graph` feature adds a document registry and bounded +//! sync/async loading through explicitly configured `roas` fetchers. It retains +//! canonical Arazzo identities, retrieval metadata, local aliases and loading +//! diagnostics without adding IO to the execution state machine. //! //! ## Checked preparation //! @@ -53,6 +57,17 @@ mod report; mod run; mod runtime_syntax; mod select; +#[cfg(feature = "source-graph")] +mod source_graph; +#[cfg(feature = "source-graph")] +mod source_registry; +#[cfg(feature = "source-graph")] +pub use source_graph::{SourceCycle, SourceLoadOptions, SourceLoadReport}; +#[cfg(feature = "source-graph")] +pub use source_registry::{ + DocumentId, SourceDiagnostic, SourceDocument, SourceError, SourceLink, SourceRegistry, + SourceVersion, +}; pub mod testing; diff --git a/crates/roas-arazzo-executor/src/operation.rs b/crates/roas-arazzo-executor/src/operation.rs index 1780332e..88603173 100644 --- a/crates/roas-arazzo-executor/src/operation.rs +++ b/crates/roas-arazzo-executor/src/operation.rs @@ -21,8 +21,26 @@ const METHODS: [&str; 8] = [ pub(crate) struct Source { /// The URL the description was declared with. pub url: String, - /// The parsed document. - pub document: Value, + pub data: SourceData, +} + +/// Legacy callers own their value; registry-backed sources share the same +/// immutable document with the registry and loader, including cloned Options. +#[derive(Clone, Debug)] +pub(crate) enum SourceData { + Owned(Value), + #[cfg(feature = "source-graph")] + Registry(std::sync::Arc), +} + +impl Source { + pub(crate) fn document(&self) -> &Value { + match &self.data { + SourceData::Owned(value) => value, + #[cfg(feature = "source-graph")] + SourceData::Registry(document) => document.value(), + } + } } /// Where a step's request is going. @@ -135,7 +153,7 @@ fn by_id<'s>( let source = sources .get(name) .ok_or_else(|| OperationError::MissingSource(name.to_owned()))?; - let found = search(&source.document, id).ok_or_else(|| OperationError::Unknown { + let found = search(source.document(), id).ok_or_else(|| OperationError::Unknown { operation: operation.to_owned(), })?; return Ok(( @@ -151,7 +169,7 @@ fn by_id<'s>( // so, and if it is not, guessing would send the request somewhere // the author did not name. let mut hits = sources.iter().filter_map(|(name, source)| { - search(&source.document, operation).map(|found| { + search(source.document(), operation).map(|found| { ( source, Found { @@ -222,7 +240,7 @@ fn by_path<'s>( .get(&name) .ok_or_else(|| OperationError::MissingSource(name.clone()))?; - if source.document.pointer(pointer).is_none() { + if source.document().pointer(pointer).is_none() { return Err(bad("the document has nothing at that pointer")); } // `/paths/~1pets~1{petId}/get` — the pointer itself says which path @@ -278,7 +296,7 @@ fn endpoint( let base = base_urls .get(&found.name) .cloned() - .or_else(|| server(&source.document, &found.path, &found.method)) + .or_else(|| server(source.document(), &found.path, &found.method)) .ok_or_else(|| OperationError::NoServer(named.to_owned()))?; Ok(Endpoint { method: found.method.to_uppercase(), @@ -364,7 +382,7 @@ pub(crate) mod tests { "petStore".to_owned(), Source { url: "https://api.example.com/openapi.json".to_owned(), - document: petstore(), + data: SourceData::Owned(petstore()), }, )]) } @@ -430,7 +448,7 @@ pub(crate) mod tests { "mirror".to_owned(), Source { url: "https://mirror.example.com/openapi.json".to_owned(), - document: petstore(), + data: SourceData::Owned(petstore()), }, ); let error = resolve( @@ -540,7 +558,7 @@ pub(crate) mod tests { "petStore".to_owned(), Source { url: "https://api.example.com/openapi.json".to_owned(), - document, + data: SourceData::Owned(document), }, )]); assert_eq!( @@ -568,7 +586,7 @@ pub(crate) mod tests { "petStore".to_owned(), Source { url: "https://api.example.com/swagger.json".to_owned(), - document, + data: SourceData::Owned(document), }, )]); assert_eq!( @@ -589,7 +607,9 @@ pub(crate) mod tests { "petStore".to_owned(), Source { url: "u".to_owned(), - document: json!({ "paths": { "/pets": { "get": { "operationId": "listPets" } } } }), + data: SourceData::Owned( + json!({ "paths": { "/pets": { "get": { "operationId": "listPets" } } } }), + ), }, )]); assert_eq!( diff --git a/crates/roas-arazzo-executor/src/run.rs b/crates/roas-arazzo-executor/src/run.rs index 8749aa6d..7daa1757 100644 --- a/crates/roas-arazzo-executor/src/run.rs +++ b/crates/roas-arazzo-executor/src/run.rs @@ -123,7 +123,7 @@ impl Options { name.into(), Source { url: url.into(), - document, + data: crate::operation::SourceData::Owned(document), }, ); self diff --git a/crates/roas-arazzo-executor/src/source_graph.rs b/crates/roas-arazzo-executor/src/source_graph.rs new file mode 100644 index 00000000..4a171bc1 --- /dev/null +++ b/crates/roas-arazzo-executor/src/source_graph.rs @@ -0,0 +1,369 @@ +//! Bounded source traversal; the caller owns all fetch policy and IO. + +use crate::source_registry::{join, resource_uri}; +use crate::{DocumentId, SourceDiagnostic, SourceError, SourceRegistry}; +use roas::loader::{LoadedDocument, Loader, LoaderError}; +use std::collections::{BTreeMap, BTreeSet, VecDeque}; +use std::sync::Arc; +use url::Url; + +/// Loading policy, independent of workflow step/retry/call-depth limits. +#[derive(Clone, Debug)] +#[non_exhaustive] +pub struct SourceLoadOptions { + /// Existing registry documents plus distinct fetch attempts allowed per load. + /// Failed attempts and different retrieval aliases consume a slot too. + pub max_documents: usize, + /// Root depth is zero; a direct source has depth one. Known cycle/diamond + /// targets do not expand again or consume an additional depth allowance. + pub max_depth: usize, + /// Only these root aliases are traversed, or all root aliases when absent. + /// Sources of linked Arazzo documents are traversed in full. + pub root_sources: Option>, + /// Compatibility extension: allow an Arazzo retrieval URI instead of `$self`. + /// False by default, following identity-based referencing. + pub retrieval_aliases: bool, +} + +impl Default for SourceLoadOptions { + fn default() -> Self { + Self { + max_documents: 256, + max_depth: 32, + root_sources: None, + retrieval_aliases: false, + } + } +} + +/// An edge back into the active ancestry. Source cycles are representable data, +/// not execution permission or a reason to discard the readable documents. +#[derive(Clone, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub struct SourceCycle { + /// Document containing the back edge. + pub owner: DocumentId, + /// Local alias of the back edge. + pub source_name: String, + /// Ancestor document reached by this edge. + pub target: DocumentId, +} + +/// Loading results. Preparation, not this report, decides whether execution can +/// proceed with the available sources. A nonempty diagnostic list is not success. +#[derive(Debug)] +#[non_exhaustive] +pub struct SourceLoadReport { + /// Located unresolved/invalid/budget-limited source edges. + pub diagnostics: Vec, + /// Back edges, distinct from shared diamond dependencies. + pub cycles: Vec, + /// Distinct calls to the loader; its own cache may satisfy a call without IO. + pub fetch_attempts: usize, +} + +impl SourceRegistry { + /// Load a selected source graph through an explicitly configured loader. + /// Insert all supplied documents first. Failures retain other readable + /// documents and are reported per edge after identity discovery completes. + /// # Errors + /// Invalid root/selection or an already-exceeded initial document budget. + /// Source failures otherwise belong to the returned report. + pub fn load_sources( + &mut self, + root: DocumentId, + loader: &mut Loader, + options: &SourceLoadOptions, + ) -> Result { + let mut traversal = Traversal::new(self, root, options)?; + while let Some(uri) = traversal.next(self) { + let document = loader.load_document_shared(uri.as_str()); + traversal.accept(self, uri, document); + } + Ok(traversal.finish(self)) + } + + /// Async loading with exactly the same graph policy and diagnostics. + /// Uses the loader's registered async fetchers on cache misses. + pub async fn load_sources_async( + &mut self, + root: DocumentId, + loader: &mut Loader, + options: &SourceLoadOptions, + ) -> Result { + let mut traversal = Traversal::new(self, root, options)?; + while let Some(uri) = traversal.next(self) { + let document = loader.load_document_shared_async(uri.as_str()).await; + traversal.accept(self, uri, document); + } + Ok(traversal.finish(self)) + } +} + +type Edge = (DocumentId, String); + +struct Traversal<'a> { + root: DocumentId, + options: &'a SourceLoadOptions, + initial_documents: usize, + queue: VecDeque<(Edge, usize)>, + seen: BTreeMap, + edges: BTreeSet, + pending: BTreeMap, + errors: BTreeMap>, + attempted: BTreeSet, + failed: BTreeMap>, + generation: usize, + retried: usize, +} + +impl<'a> Traversal<'a> { + fn new( + registry: &SourceRegistry, + root: DocumentId, + options: &'a SourceLoadOptions, + ) -> Result { + registry.document(root)?; + if registry.len() > options.max_documents { + return Err(SourceError::Limit { + kind: "document", + limit: options.max_documents, + }); + } + if let Some(names) = &options.root_sources { + for name in names { + if registry.source(root, name).is_none() { + return Err(SourceError::Alias { + owner: root, + name: name.clone(), + }); + } + } + } + let mut traversal = Self { + root, + options, + initial_documents: registry.len(), + queue: VecDeque::new(), + seen: BTreeMap::new(), + edges: BTreeSet::new(), + pending: BTreeMap::new(), + errors: BTreeMap::new(), + attempted: BTreeSet::new(), + failed: BTreeMap::new(), + generation: 0, + retried: 0, + }; + traversal.enqueue(registry, root, 0); + Ok(traversal) + } + + fn enqueue(&mut self, registry: &SourceRegistry, owner: DocumentId, depth: usize) { + if self + .seen + .get(&owner) + .is_some_and(|previous| *previous <= depth) + { + return; + } + self.seen.insert(owner, depth); + for source in registry + .sources(owner) + .expect("registered traversal document") + { + if owner == self.root + && self + .options + .root_sources + .as_ref() + .is_some_and(|names| !names.contains(&source.name)) + { + continue; + } + let edge = (owner, source.name.clone()); + self.edges.insert(edge.clone()); + self.queue.push_back((edge, depth.saturating_add(1))); + } + } + + fn reject(&mut self, edge: Edge, depth: usize, error: Arc) { + self.pending.insert(edge.clone(), depth); + self.errors.insert(edge, error); + } + + fn next(&mut self, registry: &mut SourceRegistry) -> Option { + loop { + let Some((edge, _)) = self.queue.pop_front() else { + // A later document may supply an earlier reference's identity. + // Reconcile after complete discovery, without refetching failures. + if self.retried != self.generation { + self.retried = self.generation; + self.queue.extend(std::mem::take(&mut self.pending)); + continue; + } + return None; + }; + let depth = self.seen[&edge.0].saturating_add(1); + let link = registry.links.get(&edge).expect("declared edge").clone(); + registry.links.get_mut(&edge).expect("declared edge").target = None; + let target = registry.overrides.get(&edge).copied().map_or_else( + || registry.resolve(edge.0, &link.declared_uri, self.options.retrieval_aliases), + |id| Ok(Some(id)), + ); + let target = match target { + Ok(target) => target, + Err(error) => { + self.reject(edge, depth, Arc::new(error)); + continue; + } + }; + if depth > self.options.max_depth + && target.is_none_or(|id| !self.seen.contains_key(&id)) + { + self.reject( + edge, + depth, + Arc::new(SourceError::Limit { + kind: "depth", + limit: self.options.max_depth, + }), + ); + continue; + } + if let Some(target) = target { + if let Err(error) = registry.check_kind(&link, target) { + self.reject(edge, depth, Arc::new(error)); + continue; + } + registry.links.get_mut(&edge).expect("declared edge").target = Some(target); + self.pending.remove(&edge); + self.errors.remove(&edge); + self.enqueue(registry, target, depth); + continue; + } + let uri = join( + registry.document(edge.0).expect("owner").base_uri(), + &link.declared_uri, + ) + .and_then(|uri| resource_uri(uri.as_str())); + let uri = match uri { + Ok(uri) => uri, + Err(error) => { + self.reject(edge, depth, Arc::new(error)); + continue; + } + }; + if let Some(error) = self.failed.get(&uri) { + self.reject(edge, depth, Arc::clone(error)); + continue; + } + if self.attempted.contains(&uri) { + self.reject( + edge, + depth, + Arc::new(SourceError::Unresolved(uri.to_string())), + ); + continue; + } + if self.attempted.len() >= self.options.max_documents - self.initial_documents { + self.reject( + edge, + depth, + Arc::new(SourceError::Limit { + kind: "document", + limit: self.options.max_documents, + }), + ); + continue; + } + self.attempted.insert(uri.clone()); + self.queue.push_front((edge, depth)); + return Some(uri); + } + } + + fn accept( + &mut self, + registry: &mut SourceRegistry, + uri: Url, + result: Result, LoaderError>, + ) { + let result = result.map_err(SourceError::Load).and_then(|loaded| { + let id = registry.insert_document(loaded)?; + registry.add_retrieval_alias(id, uri.as_str())?; + Ok(id) + }); + match result { + Ok(_) => self.generation += 1, + Err(error) => { + self.failed.insert(uri, Arc::new(error)); + } + } + } + + fn finish(self, registry: &SourceRegistry) -> SourceLoadReport { + let mut diagnostics = self + .errors + .into_iter() + .map(|((owner, source_name), error)| { + let link = registry + .source(owner, &source_name) + .expect("declared source"); + SourceDiagnostic { + owner, + source_name, + path: format!("#.sourceDescriptions[{}].url", link.index), + declared_uri: link.declared_uri.clone(), + error, + } + }) + .collect::>(); + diagnostics.sort_by_key(|diagnostic| { + ( + diagnostic.owner, + registry + .source(diagnostic.owner, &diagnostic.source_name) + .expect("source") + .index, + ) + }); + SourceLoadReport { + diagnostics, + cycles: cycles(registry, self.root, &self.edges), + fetch_attempts: self.attempted.len(), + } + } +} + +fn cycles(registry: &SourceRegistry, root: DocumentId, edges: &BTreeSet) -> Vec { + let mut cycles = Vec::new(); + let mut active = BTreeSet::from([root]); + let mut done = BTreeSet::new(); + let mut stack = vec![(root, 0)]; + while let Some((owner, next)) = stack.last_mut() { + let sources = registry.sources(*owner).expect("registered graph document"); + let Some(link) = sources.get(*next) else { + done.insert(*owner); + active.remove(owner); + stack.pop(); + continue; + }; + *next += 1; + if !edges.contains(&(*owner, link.name.clone())) { + continue; + } + if let Some(target) = link.target { + if active.contains(&target) { + cycles.push(SourceCycle { + owner: *owner, + source_name: link.name.clone(), + target, + }); + } else if !done.contains(&target) { + active.insert(target); + stack.push((target, 0)); + } + } + } + cycles +} diff --git a/crates/roas-arazzo-executor/src/source_registry.rs b/crates/roas-arazzo-executor/src/source_registry.rs new file mode 100644 index 00000000..f2cdd351 --- /dev/null +++ b/crates/roas-arazzo-executor/src/source_registry.rs @@ -0,0 +1,565 @@ +//! Complete documents, canonical identities and document-local source aliases. + +use roas::{LoadedDocument, LoaderError}; +use roas_arazzo::v1_1::{Description, SourceType}; +use serde_json::Value; +use std::collections::{BTreeMap, BTreeSet}; +use std::sync::Arc; +use url::Url; + +/// Stable, opaque document handle, local to the registry that created it. +/// Handles survive insertion; registries never remove or replace documents. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct DocumentId(usize); + +/// Model family whose version grammar the document declares. +/// Recognition/loading does not imply validation or execution support. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub enum SourceVersion { + /// Arazzo 1.0, upconverted to the 1.1 execution model. + Arazzo1_0, + /// Arazzo 1.1. + Arazzo1_1, + /// OpenAPI (Swagger) 2.0. + OpenApi2, + /// OpenAPI 3.0. + OpenApi3_0, + /// OpenAPI 3.1. + OpenApi3_1, + /// OpenAPI 3.2. + OpenApi3_2, + /// AsyncAPI 2.6; loading only, not broker execution. + AsyncApi2_6, + /// AsyncAPI 3.0; loading only, not broker execution. + AsyncApi3_0, + /// AsyncAPI 3.1; loading only, not broker execution. + AsyncApi3_1, +} + +impl SourceVersion { + pub(crate) fn kind(self) -> SourceType { + match self { + Self::Arazzo1_0 | Self::Arazzo1_1 => SourceType::Arazzo, + Self::OpenApi2 | Self::OpenApi3_0 | Self::OpenApi3_1 | Self::OpenApi3_2 => { + SourceType::Openapi + } + Self::AsyncApi2_6 | Self::AsyncApi3_0 | Self::AsyncApi3_1 => SourceType::Asyncapi, + } + } +} + +/// An immutable complete document. Arazzo is also deserialized in full; API +/// documents retain raw JSON and a checked version without structural validation. +#[derive(Debug)] +pub struct SourceDocument { + loaded: Arc, + retrieval: Url, + identity: Url, + model: SourceVersion, + version: String, + arazzo: Option, +} + +impl SourceDocument { + /// Complete original JSON-compatible value; references are not rewritten. + pub fn value(&self) -> &Value { + &self.loaded.document + } + /// Actual retrieval location (the final redirect location when available). + pub fn retrieval_uri(&self) -> &Url { + &self.retrieval + } + /// Resolved Arazzo `$self`, or the retrieval URI if no identity is declared. + pub fn identity(&self) -> &Url { + &self.identity + } + /// Base for this document's references, never an API endpoint override. + pub fn base_uri(&self) -> &Url { + &self.identity + } + /// Recognized model family. This is not a structural-validation result. + pub fn model(&self) -> SourceVersion { + self.model + } + /// Version exactly as written in the source document. + pub fn version(&self) -> &str { + &self.version + } + /// Fully parsed Arazzo, upconverted from v1.0 where necessary. + pub fn arazzo(&self) -> Option<&Description> { + self.arazzo.as_ref() + } +} + +/// One owner's declared source edge. A missing target is not a successful load. +#[derive(Clone, Debug)] +#[non_exhaustive] +pub struct SourceLink { + /// Alias local to the owning document. + pub name: String, + /// URI-reference exactly as declared in `sourceDescriptions`. + pub declared_uri: String, + /// URI after resolution against the owner's effective base, if valid. + pub resolved_uri: Option, + /// Linked document; cycles retain handles rather than expanding objects. + pub target: Option, + pub(crate) kind: Option, + pub(crate) index: usize, +} + +/// Source-specific failures, separate from workflow execution errors/limits. +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum SourceError { + /// Invalid retrieval URI, reference resolution or `$self` fragment. + #[error("invalid document URI `{uri}`: {reason}")] + InvalidUri { + /// URI or URI-reference being processed. + uri: String, + /// Parse/base-resolution explanation. + reason: String, + }, + /// Arazzo deserialization or a recognized model's version grammar failed. + #[error("cannot parse complete source document `{uri}`: {source}")] + Parse { + /// Document retrieval location. + uri: String, + /// Original model deserialization error. + #[source] + source: serde_json::Error, + }, + /// No supported, unambiguous version discriminator at this URI. + #[error("unsupported or missing document version in `{0}`")] + Version(String), + /// Nonidentical documents claim the same identity or retrieval location. + #[error("different documents claim identity or retrieval URI `{0}`")] + Conflict(String), + /// A handle is outside this registry's document arena. + #[error("document handle {0:?} is not in this registry")] + UnknownDocument(DocumentId), + /// An alias is undeclared or duplicated within its owner. + #[error("document {owner:?} has no unique source alias `{name}`")] + Alias { + /// Owner (prospective handle when insertion rejects duplicate aliases). + owner: DocumentId, + /// Invalid local alias. + name: String, + }, + /// A linked document does not match the source's declared type. + #[error("source `{name}` declares {expected:?}, but its document is {actual:?}")] + Kind { + /// Owner-local source name. + name: String, + /// Declared source type. + expected: SourceType, + /// Type detected from the document's version discriminator. + actual: SourceType, + }, + /// A noncanonical Arazzo retrieval alias needs explicit compatibility opt-in. + #[error( + "source reference `{reference}` names a retrieval alias, not the Arazzo identity `{identity}`" + )] + Identity { + /// Resolved reference that named a retrieval location. + reference: String, + /// Document's resolved `$self` identity. + identity: String, + }, + /// A fetched reference still has no registered target. + #[error("source reference `{0}` could not be resolved")] + Unresolved(String), + /// Graph loading exceeded its document-attempt or expansion-depth budget. + #[error("source graph {kind} limit ({limit}) exceeded")] + Limit { + /// Budget name (`document` or `depth`). + kind: &'static str, + /// Configured maximum. + limit: usize, + }, + /// Failure from the caller-configured loader, retaining its source chain. + #[error(transparent)] + Load(#[from] LoaderError), +} + +/// A loading diagnostic retains the owner and exact source field, while other +/// readable documents remain available for preparation. +#[derive(Clone, Debug)] +#[non_exhaustive] +pub struct SourceDiagnostic { + /// Owning Arazzo document, not the failed target. + pub owner: DocumentId, + /// Owner-local source name. + pub source_name: String, + /// Field location within that owner. + pub path: String, + /// Declared URI-reference, before base resolution. + pub declared_uri: String, + /// Typed failure shared by edges that attempted the same unavailable resource. + pub error: Arc, +} + +impl std::fmt::Display for SourceDiagnostic { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "document {:?} {} (`{}`): {}", + self.owner, self.path, self.source_name, self.error + ) + } +} + +/// Registry of immutable documents and owner-scoped source/base-URL overrides. +/// Insert all supplied documents before loading/resolving links, so `$self` +/// identities can be found even when their retrieval locations differ. +#[derive(Debug, Default)] +pub struct SourceRegistry { + pub(crate) documents: Vec>, + identities: BTreeMap, + retrievals: BTreeMap, + pub(crate) links: BTreeMap<(DocumentId, String), SourceLink>, + pub(crate) overrides: BTreeMap<(DocumentId, String), DocumentId>, + pub(crate) base_urls: BTreeMap<(DocumentId, String), String>, +} + +impl SourceRegistry { + /// Empty registry; no fetching or ambient file/network policy. + pub fn new() -> Self { + Self::default() + } + + /// Parse a complete supplied document and register its identity atomically. + /// Identical documents reuse a handle; conflicting identities are rejected. + /// # Errors + /// Invalid URI/version, malformed Arazzo, duplicate aliases or identity collision. + pub fn insert(&mut self, retrieval: &str, value: Value) -> Result { + let retrieval = resource_uri(retrieval)?; + self.insert_document(Arc::new(LoadedDocument::new(value, retrieval))) + } + + pub(crate) fn insert_document( + &mut self, + loaded: Arc, + ) -> Result { + let retrieval = resource_uri(loaded.retrieval_uri.as_str())?; + let value = &loaded.document; + if let Some(id) = self.retrievals.get(&retrieval).copied() { + return if self.documents[id.0].value() == value { + Ok(id) + } else { + Err(SourceError::Conflict(retrieval.to_string())) + }; + } + let (model, version, arazzo) = parse_document(value, &retrieval)?; + let identity = match arazzo + .as_ref() + .and_then(|document| document.self_.as_deref()) + { + Some(self_) => { + let identity = join(&retrieval, self_)?; + if identity.fragment().is_some() { + return Err(SourceError::InvalidUri { + uri: self_.into(), + reason: "`$self` must not contain a fragment".into(), + }); + } + identity + } + None => retrieval.clone(), + }; + if let Some(id) = self.identities.get(&identity).copied() { + if self.documents[id.0].value() != value { + return Err(SourceError::Conflict(identity.to_string())); + } + self.add_retrieval_alias(id, retrieval.as_str())?; + return Ok(id); + } + // Never let a new canonical identity replace another document's alias. + if self.retrievals.contains_key(&identity) || self.identities.contains_key(&retrieval) { + return Err(SourceError::Conflict(identity.to_string())); + } + let id = DocumentId(self.documents.len()); + let mut names = BTreeSet::new(); + if let Some(description) = &arazzo { + for source in &description.source_descriptions { + if !names.insert(source.name.as_str()) { + return Err(SourceError::Alias { + owner: id, + name: source.name.clone(), + }); + } + } + for (index, source) in description.source_descriptions.iter().enumerate() { + self.links.insert( + (id, source.name.clone()), + SourceLink { + name: source.name.clone(), + declared_uri: source.url.clone(), + resolved_uri: join(&identity, &source.url).ok(), + target: None, + kind: source.type_, + index, + }, + ); + } + } + self.identities.insert(identity.clone(), id); + self.retrievals.insert(retrieval.clone(), id); + self.documents.push(Arc::new(SourceDocument { + loaded, + retrieval, + identity, + model, + version, + arazzo, + })); + Ok(id) + } + + /// Retain a requested URI in addition to a fetcher's final retrieval URI. + /// This records a location alias; it does not change canonical identity. + pub fn add_retrieval_alias(&mut self, id: DocumentId, uri: &str) -> Result<(), SourceError> { + self.document(id)?; + let uri = resource_uri(uri)?; + if self + .retrievals + .get(&uri) + .or_else(|| self.identities.get(&uri)) + .is_some_and(|existing| *existing != id) + { + return Err(SourceError::Conflict(uri.to_string())); + } + self.retrievals.insert(uri, id); + Ok(()) + } + + /// Read a document by its stable registry-local handle. + pub fn document(&self, id: DocumentId) -> Result<&SourceDocument, SourceError> { + self.documents + .get(id.0) + .map(Arc::as_ref) + .ok_or(SourceError::UnknownDocument(id)) + } + /// Number of unique documents registered, not the number of aliases. + pub fn len(&self) -> usize { + self.documents.len() + } + /// Whether this registry has no documents. + pub fn is_empty(&self) -> bool { + self.documents.is_empty() + } + /// Sources declared by one owner, in document order. + pub fn sources(&self, owner: DocumentId) -> Result, SourceError> { + self.document(owner)?; + let mut sources = self + .links + .iter() + .filter_map(|((id, _), link)| (*id == owner).then_some(link)) + .collect::>(); + sources.sort_by_key(|link| link.index); + Ok(sources) + } + /// Lookup a local alias; names from other documents cannot collide. + pub fn source(&self, owner: DocumentId, name: &str) -> Option<&SourceLink> { + self.links.get(&(owner, name.into())) + } + /// Resolve a document reference without IO. Canonical identities win. + /// Retrieval aliases for Arazzo documents with `$self` require explicit opt-in. + pub fn resolve( + &self, + owner: DocumentId, + reference: &str, + retrieval_aliases: bool, + ) -> Result, SourceError> { + let mut uri = join(self.document(owner)?.base_uri(), reference)?; + uri.set_fragment(None); + if let Some(id) = self.identities.get(&uri) { + return Ok(Some(*id)); + } + if let Some(id) = self.retrievals.get(&uri) { + let document = self.document(*id)?; + if !retrieval_aliases && document.arazzo.as_ref().is_some_and(|d| d.self_.is_some()) { + return Err(SourceError::Identity { + reference: uri.to_string(), + identity: document.identity.to_string(), + }); + } + return Ok(Some(*id)); + } + Ok(None) + } + /// Explicit caller override of one owner's source. This is a deliberate + /// override, not implicit identity-based resolution. Type must still match. + pub fn override_source( + &mut self, + owner: DocumentId, + name: &str, + target: DocumentId, + ) -> Result<(), SourceError> { + let link = self.source(owner, name).ok_or_else(|| SourceError::Alias { + owner, + name: name.into(), + })?; + self.check_kind(link, target)?; + self.overrides.insert((owner, name.into()), target); + self.links + .get_mut(&(owner, name.into())) + .expect("checked alias") + .target = Some(target); + Ok(()) + } + /// Scope an API endpoint override to one document's alias. Explicit Options + /// overrides take precedence when adapting the registry for execution. + pub fn override_base_url( + &mut self, + owner: DocumentId, + name: &str, + url: impl Into, + ) -> Result<(), SourceError> { + if self.source(owner, name).is_none() { + return Err(SourceError::Alias { + owner, + name: name.into(), + }); + } + self.base_urls.insert((owner, name.into()), url.into()); + Ok(()) + } + pub(crate) fn check_kind( + &self, + link: &SourceLink, + target: DocumentId, + ) -> Result<(), SourceError> { + let actual = self.document(target)?.model.kind(); + if let Some(expected) = link.kind + && expected != actual + { + return Err(SourceError::Kind { + name: link.name.clone(), + expected, + actual, + }); + } + Ok(()) + } + pub(crate) fn shared(&self, id: DocumentId) -> Result, SourceError> { + self.documents + .get(id.0) + .cloned() + .ok_or(SourceError::UnknownDocument(id)) + } +} + +pub(crate) fn resource_uri(uri: &str) -> Result { + let mut url = Url::parse(uri).map_err(|error| SourceError::InvalidUri { + uri: uri.into(), + reason: error.to_string(), + })?; + url.set_fragment(None); + Ok(url) +} + +pub(crate) fn join(base: &Url, reference: &str) -> Result { + base.join(reference) + .map_err(|error| SourceError::InvalidUri { + uri: reference.into(), + reason: if base.cannot_be_a_base() && matches!(Url::parse(reference), Err(url::ParseError::RelativeUrlWithoutBase)) { + format!("base `{base}` is not hierarchical — give the source an absolute URL, or the Arazzo description a hierarchical `$self`") + } else { + format!("resolving against `{base}`: {error}") + }, + }) +} + +fn parse_document( + value: &Value, + uri: &Url, +) -> Result<(SourceVersion, String, Option), SourceError> { + let parse_error = |source| SourceError::Parse { + uri: uri.to_string(), + source, + }; + // Reject ambiguous discriminators rather than choosing a document kind silently. + if ["arazzo", "swagger", "openapi", "asyncapi"] + .iter() + .filter(|key| value.get(**key).is_some()) + .count() + != 1 + { + return Err(SourceError::Version(uri.to_string())); + } + if let Some(version) = value.get("arazzo").and_then(Value::as_str) { + if version.starts_with("1.0.") { + let document = serde_json::from_value::(value.clone()) + .map_err(parse_error)?; + return Ok(( + SourceVersion::Arazzo1_0, + version.into(), + Some(document.into()), + )); + } + if version.starts_with("1.1.") { + let document = serde_json::from_value(value.clone()).map_err(parse_error)?; + return Ok((SourceVersion::Arazzo1_1, version.into(), Some(document))); + } + } + macro_rules! version { + ($key:literal, $prefix:literal, $type:ty, $model:ident) => { + if let Some(written) = value.get($key).and_then(Value::as_str) + && (written == $prefix || written.starts_with(concat!($prefix, "."))) + { + serde_json::from_value::<$type>(value[$key].clone()).map_err(parse_error)?; + return Ok((SourceVersion::$model, written.into(), None)); + } + }; + } + version!("swagger", "2", roas::v2::spec::Version, OpenApi2); + version!("openapi", "3.0", roas::v3_0::spec::Version, OpenApi3_0); + version!("openapi", "3.1", roas::v3_1::spec::Version, OpenApi3_1); + version!("openapi", "3.2", roas::v3_2::spec::Version, OpenApi3_2); + version!("asyncapi", "2.6", roas_asyncapi::v2_6::Version, AsyncApi2_6); + version!("asyncapi", "3.0", roas_asyncapi::v3_0::Version, AsyncApi3_0); + version!("asyncapi", "3.1", roas_asyncapi::v3_1::Version, AsyncApi3_1); + Err(SourceError::Version(uri.to_string())) +} + +impl crate::Options { + /// Adapt readable sources from one registry owner into executor options. + /// Existing `source`/`base_url` entries are explicit overrides and win. + /// Unresolved graph links remain absent, so checked preparation still rejects + /// a missing required source (or an unprovable bare operation ID). + /// # Errors + /// An invalid registry handle. Loading diagnostics remain in SourceLoadReport. + pub fn source_registry( + mut self, + registry: &SourceRegistry, + owner: DocumentId, + ) -> Result { + for link in registry.sources(owner)? { + if let Some(target) = link.target { + let document = registry.shared(target)?; + self.sources + .entry(link.name.clone()) + .or_insert_with(|| crate::operation::Source { + url: link.declared_uri.clone(), + data: crate::operation::SourceData::Registry(document), + }); + } + if let Some(url) = registry.base_urls.get(&(owner, link.name.clone())) { + self.base_urls + .entry(link.name.clone()) + .or_insert_with(|| url.clone()); + } + } + Ok(self) + } + + /// Original document metadata for a registry-backed source. Legacy + /// `Options::source` values have no invented retrieval URI or identity. + pub fn source_document(&self, name: &str) -> Option<&SourceDocument> { + match &self.sources.get(name)?.data { + crate::operation::SourceData::Registry(document) => Some(document), + crate::operation::SourceData::Owned(_) => None, + } + } +} diff --git a/crates/roas-arazzo-executor/tests/source_graph_test.rs b/crates/roas-arazzo-executor/tests/source_graph_test.rs new file mode 100644 index 00000000..f10379fa --- /dev/null +++ b/crates/roas-arazzo-executor/tests/source_graph_test.rs @@ -0,0 +1,780 @@ +#![cfg(feature = "source-graph")] + +use roas::loader::{AsyncResourceFetcher, FetchFuture, Loader, LoaderError, ResourceFetcher}; +use roas_arazzo_executor::{ + Options, SourceError, SourceLoadOptions, SourceRegistry, SourceVersion, prepare, testing::Fake, +}; +use serde_json::{Value, json}; +use std::cell::RefCell; +use std::collections::BTreeMap; +use std::rc::Rc; +use url::Url; + +fn workflow(sources: Value) -> Value { + json!({ "arazzo": "1.1.0", "info": { "title": "Graph", "version": "1" }, + "sourceDescriptions": sources, + "workflows": [{ "workflowId": "w", "steps": [{ "stepId": "get", "operationId": "$sourceDescriptions.api.check" }] }] }) +} + +fn api() -> Value { + json!({ "openapi": "3.1.0", "info": { "title": "API", "version": "1" }, + "servers": [{ "url": "https://api.example.test" }], + "paths": { "/check": { "get": { "operationId": "check" } } } }) +} + +#[derive(Clone)] +struct Memory { + documents: BTreeMap, + reads: Rc>>, +} + +impl Memory { + fn new(documents: impl IntoIterator) -> Self { + Self { + documents: documents + .into_iter() + .map(|(uri, value)| (uri.into(), value)) + .collect(), + reads: Rc::default(), + } + } + fn loader(&self) -> Loader { + let mut loader = Loader::new(); + loader.register_fetcher("https://", self.clone()); + loader + } +} + +impl ResourceFetcher for Memory { + fn fetch(&mut self, uri: &Url) -> Result { + self.reads.borrow_mut().push(uri.to_string()); + self.documents + .get(uri.as_str()) + .cloned() + .ok_or_else(|| LoaderError::NoFetcherRegistered { + uri: uri.to_string(), + }) + } +} +impl AsyncResourceFetcher for Memory { + fn fetch<'a>(&'a mut self, uri: &'a Url) -> FetchFuture<'a> { + Box::pin(async move { ResourceFetcher::fetch(self, uri) }) + } +} + +#[test] +fn relative_self_and_equivalent_references_reuse_one_document() { + let mut value = workflow(json!([ + { "name": "api", "url": "./nested/../api.json" }, + { "name": "same", "url": "https://EXAMPLE.test:443/ids/./api.json#/paths" } + ])); + value["$self"] = json!("../ids/root.json"); + value["components"] = + json!({ "inputs": { "schema": { "$id": "nested/", "$ref": "../other.json" } } }); + let mut registry = SourceRegistry::new(); + assert!(registry.is_empty()); + let root = registry + .insert("https://example.test/cache/root.json", value.clone()) + .unwrap(); + let memory = Memory::new([("https://example.test/ids/api.json", api())]); + let report = registry + .load_sources(root, &mut memory.loader(), &SourceLoadOptions::default()) + .unwrap(); + assert!(report.diagnostics.is_empty()); + assert!(report.cycles.is_empty()); + assert_eq!(report.fetch_attempts, 1); + assert_eq!(registry.len(), 2); + assert_eq!(memory.reads.borrow().len(), 1); + assert_eq!(registry.document(root).unwrap().value(), &value); + assert_eq!( + registry.document(root).unwrap().identity().as_str(), + "https://example.test/ids/root.json" + ); + assert_eq!( + registry.document(root).unwrap().retrieval_uri().as_str(), + "https://example.test/cache/root.json" + ); + assert_eq!( + registry.source(root, "api").unwrap().target, + registry.source(root, "same").unwrap().target + ); + let options = Options::new().source_registry(®istry, root).unwrap(); + assert_eq!(options.source_document("api").unwrap().version(), "3.1.0"); + assert_eq!( + options.source_document("api").unwrap().model(), + SourceVersion::OpenApi3_1 + ); + assert!(options.source_document("absent").is_none()); +} + +#[tokio::test] +async fn graph_and_cloned_options_share_the_loaders_document_value() { + let memory = Memory::new([("https://graph.test/api.json", api())]); + let mut loader = Loader::new(); + loader.register_async_fetcher("https://", memory.clone()); + let mut registry = SourceRegistry::new(); + let root = registry + .insert( + "https://graph.test/root.json", + workflow(json!([ + {"name":"api", "url":"api.json"}, {"name":"same", "url":"./api.json"} + ])), + ) + .unwrap(); + registry + .load_sources_async(root, &mut loader, &SourceLoadOptions::default()) + .await + .unwrap(); + let loaded = loader + .load_document_shared("https://graph.test/api.json") + .unwrap(); + let options = Options::new().source_registry(®istry, root).unwrap(); + let cloned = options.clone(); + let api_id = registry.source(root, "api").unwrap().target.unwrap(); + assert!(std::ptr::eq( + registry.document(api_id).unwrap().value(), + &loaded.document + )); + for entry in [&options, &cloned] { + for alias in ["api", "same"] { + assert!(std::ptr::eq( + entry.source_document(alias).unwrap().value(), + &loaded.document + )); + } + } + let description = registry.document(root).unwrap().arazzo().unwrap().clone(); + drop(registry); + drop(loader); + let mut client = Fake::new().reply(200, &json!({})); + assert!( + prepare(&description, &cloned) + .unwrap() + .execute(&mut client) + .unwrap() + .is_success() + ); + assert_eq!(memory.reads.borrow().len(), 1); +} + +#[test] +fn opaque_self_errors_explain_both_remedies_without_rejecting_absolute_sources() { + let mut document = workflow(json!([ + {"name":"relative", "url":"api.json"}, + {"name":"absolute", "url":"https://graph.test/api.json"}, + {"name":"malformed", "url":"http://["} + ])); + document["$self"] = json!("urn:example:root"); + let mut registry = SourceRegistry::new(); + let root = registry.insert("file:///root.json", document).unwrap(); + let api = registry + .insert("https://graph.test/api.json", api()) + .unwrap(); + let report = registry + .load_sources(root, &mut Loader::new(), &SourceLoadOptions::default()) + .unwrap(); + assert_eq!(report.diagnostics.len(), 2); + let message = report.diagnostics[0].error.to_string(); + assert!(message.contains("absolute URL"), "{message}"); + assert!(message.contains("hierarchical `$self`"), "{message}"); + assert!(message.contains("urn:example:root"), "{message}"); + assert!( + !report.diagnostics[1] + .error + .to_string() + .contains("hierarchical `$self`") + ); + assert_eq!(registry.source(root, "absolute").unwrap().target, Some(api)); +} + +#[test] +fn failed_aliases_share_an_attempt_but_queries_remain_distinct() { + let mut registry = SourceRegistry::new(); + let root = registry + .insert( + "https://graph.test/root.json", + workflow(json!([ + {"name": "first", "url": "missing.json?revision=1"}, + {"name": "same", "url": "./missing.json?revision=1#/ignored"}, + {"name": "other", "url": "missing.json?revision=2"} + ])), + ) + .unwrap(); + let memory = Memory::new([]); + let mut options = SourceLoadOptions::default(); + options.max_documents = 2; // root plus one failed attempt + let report = registry + .load_sources(root, &mut memory.loader(), &options) + .unwrap(); + assert_eq!(report.fetch_attempts, 1); + assert_eq!(memory.reads.borrow().len(), 1); + assert_eq!(report.diagnostics.len(), 3); + assert!(std::sync::Arc::ptr_eq( + &report.diagnostics[0].error, + &report.diagnostics[1].error + )); + assert!(matches!( + *report.diagnostics[2].error, + SourceError::Limit { + kind: "document", + limit: 2 + } + )); + assert!(report.diagnostics[0].path.ends_with("[0].url")); + assert!(report.diagnostics[1].path.ends_with("[1].url")); + assert!(report.diagnostics[2].path.ends_with("[2].url")); +} + +#[test] +fn wrong_kinds_invalid_bases_and_alias_collisions_are_typed_and_preserve_documents() { + let mut registry = SourceRegistry::new(); + let root = registry + .insert( + "https://graph.test/root.json", + workflow(json!([ + {"name": "wrong", "url": "api.json", "type": "arazzo"}, + {"name": "invalid", "url": "http://["}, + {"name": "valid", "url": "api.json", "type": "openapi"} + ])), + ) + .unwrap(); + let api_id = registry + .insert("https://graph.test/api.json", api()) + .unwrap(); + assert!(matches!( + registry.override_source(root, "wrong", api_id), + Err(SourceError::Kind { .. }) + )); + assert!(matches!( + registry.add_retrieval_alias(api_id, "https://graph.test/root.json"), + Err(SourceError::Conflict(_)) + )); + let report = registry + .load_sources(root, &mut Loader::new(), &SourceLoadOptions::default()) + .unwrap(); + assert_eq!(report.diagnostics.len(), 2); + assert!(matches!( + *report.diagnostics[0].error, + SourceError::Kind { .. } + )); + assert!(matches!( + *report.diagnostics[1].error, + SourceError::InvalidUri { .. } + )); + assert_eq!(registry.source(root, "valid").unwrap().target, Some(api_id)); + assert_eq!(registry.len(), 2); + + let mut value = workflow(json!([{"name": "relative", "url": "api.json"}])); + value["$self"] = json!("urn:example:opaque"); + let opaque = registry.insert("file:///opaque.json", value).unwrap(); + let report = registry + .load_sources(opaque, &mut Loader::new(), &SourceLoadOptions::default()) + .unwrap(); + assert!(matches!( + *report.diagnostics[0].error, + SourceError::InvalidUri { .. } + )); + assert_eq!(report.fetch_attempts, 0); + assert!(matches!( + registry.resolve(opaque, "api.json", false), + Err(SourceError::InvalidUri { .. }) + )); + + let mut collision = workflow(json!([])); + collision["$self"] = json!("https://graph.test/api.json"); + assert!(matches!( + registry.insert("file:///collision.json", collision), + Err(SourceError::Conflict(_)) + )); + assert_eq!(registry.len(), 3); + let foreign = SourceRegistry::new(); + assert!(matches!( + Options::new().source_registry(&foreign, root), + Err(SourceError::UnknownDocument(_)) + )); + assert!(matches!( + registry.override_source(root, "valid", opaque), + Err(SourceError::Kind { .. }) + )); +} + +#[test] +fn redirect_metadata_controls_relative_self_and_child_references() { + struct Redirect; + impl ResourceFetcher for Redirect { + fn fetch(&mut self, _: &Url) -> Result { + panic!("metadata-aware path required") + } + fn fetch_document(&mut self, _: &Url) -> Result { + let mut value = workflow(json!([{"name":"api", "url":"api.json"}])); + value["$self"] = json!("../identity/child.json"); + Ok(roas::LoadedDocument::new( + value, + Url::parse("https://graph.test/redirected/child.json").unwrap(), + )) + } + } + let mut registry = SourceRegistry::new(); + let root = registry + .insert( + "https://graph.test/root.json", + workflow(json!([ + {"name":"child", "url":"https://graph.test/identity/child.json", "type":"arazzo"} + ])), + ) + .unwrap(); + let api_id = registry + .insert("https://graph.test/identity/api.json", api()) + .unwrap(); + let mut loader = Loader::new(); + loader.register_fetcher("https://", Redirect); + let report = registry + .load_sources(root, &mut loader, &SourceLoadOptions::default()) + .unwrap(); + assert!(report.diagnostics.is_empty(), "{report:?}"); + let child = registry.source(root, "child").unwrap().target.unwrap(); + assert_eq!( + registry.document(child).unwrap().retrieval_uri().as_str(), + "https://graph.test/redirected/child.json" + ); + assert_eq!( + registry.document(child).unwrap().identity().as_str(), + "https://graph.test/identity/child.json" + ); + assert_eq!(registry.source(child, "api").unwrap().target, Some(api_id)); + assert_eq!(report.fetch_attempts, 1); +} + +#[test] +fn deferred_shorter_identity_path_revisits_depth_limited_descendants() { + let mut registry = SourceRegistry::new(); + let root = registry + .insert( + "https://graph.test/root.json", + workflow(json!([ + {"name":"short", "url":"identity.json", "type":"arazzo"}, + {"name":"long", "url":"a.json", "type":"arazzo"} + ])), + ) + .unwrap(); + let mut identified = workflow(json!([{"name":"api", "url":"api.json"}])); + identified["$self"] = json!("https://graph.test/identity.json"); + let memory = Memory::new([ + ( + "https://graph.test/a.json", + workflow(json!([{"name":"b", "url":"b.json", "type":"arazzo"}])), + ), + ( + "https://graph.test/b.json", + workflow(json!([{"name":"mirror", "url":"mirror.json", "type":"arazzo"}])), + ), + ("https://graph.test/mirror.json", identified), + ("https://graph.test/api.json", api()), + ]); + let mut options = SourceLoadOptions::default(); + options.retrieval_aliases = true; + options.max_depth = 3; + let report = registry + .load_sources(root, &mut memory.loader(), &options) + .unwrap(); + assert!(report.diagnostics.is_empty(), "{report:?}"); + assert_eq!(report.fetch_attempts, 5); // one failed identity lookup, four readable documents + let child = registry.source(root, "short").unwrap().target.unwrap(); + assert!(registry.source(child, "api").unwrap().target.is_some()); + assert_eq!(registry.len(), 5); +} + +#[test] +fn all_supplied_identities_are_indexed_before_loading_and_aliases_are_scoped() { + let mut registry = SourceRegistry::new(); + let root = registry + .insert( + "file:///supplied/root.json", + workflow(json!([ + { "name": "child", "url": "https://identity.test/child.json", "type": "arazzo" }, + { "name": "api", "url": "https://api.test/root.json" } + ])), + ) + .unwrap(); + let mut child = workflow(json!([{ "name": "api", "url": "https://api.test/child.json" }])); + child["$self"] = json!("https://identity.test/child.json"); + let child = registry + .insert("file:///offline/mirror.json", child) + .unwrap(); + let root_api = registry + .insert("https://api.test/root.json", api()) + .unwrap(); + let child_api = registry + .insert("https://api.test/child.json", api()) + .unwrap(); + let report = registry + .load_sources(root, &mut Loader::new(), &SourceLoadOptions::default()) + .unwrap(); + assert!(report.diagnostics.is_empty()); + assert_eq!(report.fetch_attempts, 0); + assert_eq!(registry.source(root, "child").unwrap().target, Some(child)); + assert_eq!(registry.source(root, "api").unwrap().target, Some(root_api)); + assert_eq!( + registry.source(child, "api").unwrap().target, + Some(child_api) + ); + assert!(matches!( + registry.resolve(root, "file:///offline/mirror.json", false), + Err(SourceError::Identity { .. }) + )); + assert_eq!( + registry + .resolve(root, "file:///offline/mirror.json", true) + .unwrap(), + Some(child) + ); + assert_eq!( + registry + .resolve(root, "https://identity.test/child.json#/workflows/0", false) + .unwrap(), + Some(child) + ); +} + +#[test] +fn diamonds_are_shared_and_back_edges_are_cycles_not_expanded_documents() { + let mut registry = SourceRegistry::new(); + let root = registry + .insert( + "https://graph.test/root.json", + workflow(json!([ + { "name": "a", "url": "a.json", "type": "arazzo" }, + { "name": "b", "url": "b.json", "type": "arazzo" } + ])), + ) + .unwrap(); + let branches = workflow(json!([{ "name": "shared", "url": "shared.json", "type": "arazzo" }])); + let memory = Memory::new([ + ("https://graph.test/a.json", branches.clone()), + ("https://graph.test/b.json", branches), + ( + "https://graph.test/shared.json", + workflow(json!([{ "name": "back", "url": "root.json", "type": "arazzo" }])), + ), + ]); + let report = registry + .load_sources(root, &mut memory.loader(), &SourceLoadOptions::default()) + .unwrap(); + assert!(report.diagnostics.is_empty()); + assert_eq!(report.fetch_attempts, 3); + assert_eq!(registry.len(), 4); + assert_eq!(report.cycles.len(), 1); + assert_eq!(report.cycles[0].target, root); + let a = registry.source(root, "a").unwrap().target.unwrap(); + let b = registry.source(root, "b").unwrap().target.unwrap(); + assert_eq!( + registry.source(a, "shared").unwrap().target, + registry.source(b, "shared").unwrap().target + ); +} + +#[test] +fn later_discovery_repairs_an_earlier_unresolved_identity() { + let mut registry = SourceRegistry::new(); + let root = registry + .insert( + "https://graph.test/root.json", + workflow(json!([ + { "name": "identity", "url": "https://identity.test/child", "type": "arazzo" }, + { "name": "location", "url": "mirror.json", "type": "arazzo" } + ])), + ) + .unwrap(); + let mut child = workflow(json!([{ "name": "api", "url": "https://api.test/openapi.json" }])); + child["$self"] = json!("https://identity.test/child"); + let memory = Memory::new([ + ("https://graph.test/mirror.json", child), + ("https://api.test/openapi.json", api()), + ]); + let report = registry + .load_sources(root, &mut memory.loader(), &SourceLoadOptions::default()) + .unwrap(); + assert_eq!(report.diagnostics.len(), 1); + assert_eq!(report.diagnostics[0].source_name, "location"); + assert!(matches!( + &*report.diagnostics[0].error, + SourceError::Identity { .. } + )); + assert!(registry.source(root, "identity").unwrap().target.is_some()); + assert!(registry.source(root, "location").unwrap().target.is_none()); + let mut options = SourceLoadOptions::default(); + options.retrieval_aliases = true; + let report = registry + .load_sources(root, &mut Loader::new(), &options) + .unwrap(); + assert!(report.diagnostics.is_empty()); + assert_eq!(report.fetch_attempts, 0); +} + +#[test] +fn unavailable_unrelated_source_does_not_make_a_qualified_run_fail_or_a_bare_run_pass() { + let mut value = workflow(json!([ + { "name": "api", "url": "api.json" }, { "name": "unavailable", "url": "absent.json" } + ])); + let mut registry = SourceRegistry::new(); + let root = registry + .insert("https://graph.test/root.json", value.clone()) + .unwrap(); + let memory = Memory::new([("https://graph.test/api.json", api())]); + let report = registry + .load_sources(root, &mut memory.loader(), &SourceLoadOptions::default()) + .unwrap(); + assert_eq!(report.diagnostics.len(), 1); + assert_eq!(report.diagnostics[0].owner, root); + assert_eq!(report.diagnostics[0].path, "#.sourceDescriptions[1].url"); + assert!(report.diagnostics[0].to_string().contains("unavailable")); + let options = Options::new().source_registry(®istry, root).unwrap(); + let description = registry.document(root).unwrap().arazzo().unwrap(); + assert!( + prepare(description, &options) + .unwrap() + .execute(&mut Fake::new().reply(200, &json!({}))) + .unwrap() + .is_success() + ); + value["workflows"][0]["steps"][0]["operationId"] = json!("check"); + let error = prepare(&serde_json::from_value(value).unwrap(), &options).unwrap_err(); + assert!(error.to_string().contains("was not supplied")); +} + +#[test] +fn graph_budgets_are_located_and_independent_of_execution_limits() { + for (max_documents, max_depth, kind) in [(1, 32, "document"), (256, 0, "depth")] { + let mut registry = SourceRegistry::new(); + let root = registry + .insert( + "https://graph.test/root.json", + workflow(json!([{ "name": "api", "url": "api.json" }])), + ) + .unwrap(); + let memory = Memory::new([("https://graph.test/api.json", api())]); + let mut options = SourceLoadOptions::default(); + options.max_documents = max_documents; + options.max_depth = max_depth; + let report = registry + .load_sources(root, &mut memory.loader(), &options) + .unwrap(); + assert_eq!(report.diagnostics.len(), 1); + assert!( + matches!(&*report.diagnostics[0].error, SourceError::Limit { kind: actual, .. } if *actual == kind) + ); + assert_eq!(report.fetch_attempts, 0); + assert!(memory.reads.borrow().is_empty()); + options.max_documents = 0; + assert!(matches!( + registry.load_sources(root, &mut memory.loader(), &options), + Err(SourceError::Limit { .. }) + )); + assert_eq!(registry.len(), 1); + } +} + +#[test] +fn selection_and_scoped_overrides_preserve_legacy_precedence() { + let value = workflow( + json!([{ "name": "api", "url": "api.json" }, { "name": "unused", "url": "unused.json" }]), + ); + let mut registry = SourceRegistry::new(); + let root = registry + .insert("https://graph.test/root.json", value.clone()) + .unwrap(); + let other = registry + .insert("https://graph.test/other.json", value) + .unwrap(); + let a = registry.insert("file:///supplied/a.json", api()).unwrap(); + let b = registry.insert("file:///supplied/b.json", api()).unwrap(); + registry.override_source(root, "api", a).unwrap(); + registry.override_source(other, "api", b).unwrap(); + registry + .override_base_url(root, "api", "https://root.test") + .unwrap(); + registry + .override_base_url(other, "api", "https://other.test") + .unwrap(); + let mut load = SourceLoadOptions::default(); + load.root_sources = Some(["api".into()].into()); + let report = registry + .load_sources(root, &mut Loader::new(), &load) + .unwrap(); + assert!(report.diagnostics.is_empty()); + assert_eq!(report.fetch_attempts, 0); + assert!(registry.source(root, "unused").unwrap().target.is_none()); + for (owner, expected) in [ + (root, "https://root.test/check"), + (other, "https://other.test/check"), + ] { + let options = Options::new().source_registry(®istry, owner).unwrap(); + let mut client = Fake::new().reply(200, &json!({})); + prepare( + registry.document(owner).unwrap().arazzo().unwrap(), + &options, + ) + .unwrap() + .execute(&mut client) + .unwrap(); + assert_eq!(client.sent()[0].url, expected); + } + let options = Options::new() + .source("api", "api.json", api()) + .base_url("api", "https://explicit.test") + .source_registry(®istry, root) + .unwrap(); + assert!(options.source_document("api").is_none()); + let mut client = Fake::new().reply(200, &json!({})); + prepare(registry.document(root).unwrap().arazzo().unwrap(), &options) + .unwrap() + .execute(&mut client) + .unwrap(); + assert_eq!(client.sent()[0].url, "https://explicit.test/check"); + load.root_sources = Some(["typo".into()].into()); + assert!(matches!( + registry.load_sources(root, &mut Loader::new(), &load), + Err(SourceError::Alias { .. }) + )); + assert!(registry.override_source(root, "typo", a).is_err()); + assert!( + registry + .override_base_url(root, "typo", "https://example.test") + .is_err() + ); +} + +#[test] +fn complete_arazzo_parse_and_identity_conflicts_are_not_silent_overwrites() { + let mut registry = SourceRegistry::new(); + let mut value = workflow(json!([{ "name": "api", "url": "https://api.test" }])); + value["$self"] = json!("https://identity.test/w"); + let id = registry + .insert("file:///first.json", value.clone()) + .unwrap(); + assert_eq!( + registry + .insert("file:///second.json", value.clone()) + .unwrap(), + id + ); + assert_eq!( + registry + .insert("file:///first.json", value.clone()) + .unwrap(), + id + ); + value["info"]["title"] = json!("conflict"); + assert!(matches!( + registry.insert("file:///first.json", value.clone()), + Err(SourceError::Conflict(_)) + )); + assert!(matches!( + registry.insert("file:///third.json", value.clone()), + Err(SourceError::Conflict(_)) + )); + value["workflows"] = json!([{ "workflowId": "broken", "steps": "not a list" }]); + assert!(matches!( + registry.insert("file:///broken.json", value), + Err(SourceError::Parse { .. }) + )); + assert_eq!(registry.len(), 1); + let mut value = workflow(json!([{ "name": "api", "url": "a" }, { "name": "api", "url": "b" }])); + assert!(matches!( + registry.insert("file:///duplicates.json", value.clone()), + Err(SourceError::Alias { .. }) + )); + value["sourceDescriptions"] = json!([]); + value["$self"] = json!("https://identity.test/w#fragment"); + assert!(matches!( + registry.insert("file:///fragment.json", value), + Err(SourceError::InvalidUri { .. }) + )); + assert!(registry.insert("relative.json", api()).is_err()); + assert!(SourceRegistry::new().document(id).is_err()); +} + +#[test] +fn source_versions_are_checked_without_implying_broker_support() { + for (field, version, model) in [ + ("swagger", "2.0", SourceVersion::OpenApi2), + ("openapi", "3.0.4", SourceVersion::OpenApi3_0), + ("openapi", "3.1.1", SourceVersion::OpenApi3_1), + ("openapi", "3.2.0", SourceVersion::OpenApi3_2), + ("asyncapi", "2.6.0", SourceVersion::AsyncApi2_6), + ("asyncapi", "3.0.0", SourceVersion::AsyncApi3_0), + ("asyncapi", "3.1.0", SourceVersion::AsyncApi3_1), + ] { + let mut registry = SourceRegistry::new(); + let id = registry + .insert("file:///document.json", json!({ field: version })) + .unwrap(); + assert_eq!(registry.document(id).unwrap().model(), model); + assert_eq!(registry.document(id).unwrap().version(), version); + assert!(registry.document(id).unwrap().arazzo().is_none()); + } + for version in ["1.0.1", "1.1.0"] { + let mut value = workflow(json!([{ "name": "api", "url": "api.json" }])); + value["arazzo"] = json!(version); + let mut registry = SourceRegistry::new(); + let id = registry.insert("file:///document.json", value).unwrap(); + assert_eq!(registry.document(id).unwrap().version(), version); + assert_eq!( + registry + .document(id) + .unwrap() + .arazzo() + .unwrap() + .arazzo + .as_str(), + "1.1.0" + ); + } + for value in [ + json!({}), + json!({ "openapi": "3.9.0" }), + json!({ "asyncapi": "3.1.1" }), + json!({ "arazzo": "2.0.0" }), + json!({ "arazzo": "1.1.x" }), + json!({ "openapi": "3.1.0", "arazzo": "1.1.0" }), + ] { + assert!( + SourceRegistry::new() + .insert("file:///bad.json", value) + .is_err() + ); + } +} + +#[tokio::test] +async fn async_graph_loading_uses_async_policy_and_shares_cache_with_sync() { + let value = workflow(json!([{ "name": "api", "url": "api.json" }])); + let mut registry = SourceRegistry::new(); + let root = registry + .insert("https://graph.test/root.json", value) + .unwrap(); + let memory = Memory::new([("https://graph.test/api.json", api())]); + let mut loader = Loader::new(); + loader.register_async_fetcher("https://", memory.clone()); + let report = registry + .load_sources(root, &mut loader, &SourceLoadOptions::default()) + .unwrap(); + assert_eq!(report.diagnostics.len(), 1); + assert!(memory.reads.borrow().is_empty()); + let report = registry + .load_sources_async(root, &mut loader, &SourceLoadOptions::default()) + .await + .unwrap(); + assert!(report.diagnostics.is_empty()); + assert_eq!(memory.reads.borrow().len(), 1); + assert!( + registry + .load_sources(root, &mut Loader::new(), &SourceLoadOptions::default()) + .unwrap() + .diagnostics + .is_empty() + ); + fn send_sync() {} + send_sync::(); + send_sync::(); +} diff --git a/crates/roas-cli/Cargo.toml b/crates/roas-cli/Cargo.toml index 9d1bcdb7..62037c63 100644 --- a/crates/roas-cli/Cargo.toml +++ b/crates/roas-cli/Cargo.toml @@ -18,14 +18,14 @@ name = "roas" path = "src/main.rs" [dependencies] -roas = { version = "0.20", path = "../roas", features = ["clap", "v2", "v3_0", "v3_1", "v3_2"] } +roas = { workspace = true, features = ["clap", "v2", "v3_0", "v3_1", "v3_2"] } roas-overlay = { version = "0.3", path = "../roas-overlay", features = ["clap", "v1_0", "v1_1"] } roas-arazzo = { version = "0.3", path = "../roas-arazzo", features = ["clap", "v1_0", "v1_1"] } # 0.2 adds the parsed condition profile and non-exhaustive expression diagnostics. -roas-arazzo-executor = { version = "0.2", path = "../roas-arazzo-executor", features = ["reqwest"] } -roas-asyncapi = { version = "0.4", path = "../roas-asyncapi", features = ["clap", "v2_6", "v3_0", "v3_1"] } +roas-arazzo-executor = { version = "0.2", path = "../roas-arazzo-executor", features = ["reqwest", "source-graph"] } +roas-asyncapi = { workspace = true, features = ["clap", "v2_6", "v3_0", "v3_1"] } roas-file-fetcher = { version = "0.1.4", path = "../roas-file-fetcher", features = ["yaml"] } -roas-http-fetcher = { version = "0.2.4", path = "../roas-http-fetcher", features = ["yaml"] } +roas-http-fetcher = { version = "0.2.5", path = "../roas-http-fetcher", features = ["yaml"] } anyhow.workspace = true axum.workspace = true clap.workspace = true diff --git a/crates/roas-cli/README.md b/crates/roas-cli/README.md index 63f6198e..45986337 100644 --- a/crates/roas-cli/README.md +++ b/crates/roas-cli/README.md @@ -197,6 +197,29 @@ It needs the source descriptions the steps point at: name them with `--source ` lets one pass, as `arazzo validate --ignore` does. +Source loading uses document-local aliases and Arazzo `$self` identities. Relative +references use `$self` when present, otherwise the retrieval location; a relative +`$self` first resolves against that location. HTTP redirects retain the final URL. +`--source =` is an explicit root-alias override. Repeat +`--source-document ` to preload additional possible documents by identity +before any links are resolved, without assigning a root alias. + +By default only sources needed by the selected workflow (plus explicit `--source` +entries) are traversed. `--load-all-sources` traverses all root sources and linked +Arazzo documents; it does **not** grant file/network access without `--load`. +`--source-max-documents` (default 256) bounds existing supplied documents plus +distinct loader attempts, including failures. `--source-max-depth` (default 32) +bounds graph expansion, independently of `--max-steps`. Cycles and shared sources +reuse document handles. These are not byte-size limits or a network sandbox; +only enable `--load http` for documents whose referenced destinations you trust. + +An unavailable unrelated source is a located warning, not necessarily a failed run. +Checked preparation still rejects missing required sources and an unprovable bare +operation ID before sending API requests. Canonical `$self` identities are used by +default; `--allow-source-retrieval-aliases` explicitly permits noncanonical Arazzo +retrieval URLs as a compatibility extension. Loading linked Arazzo/AsyncAPI documents +does not add external workflow calls or broker execution. + The report goes to **stderr** and the workflow's outputs to **stdout**, so the outputs pipe onward; `--quiet` silences the report. The exit status follows the workflow: non-zero when it failed. ### `asyncapi` diff --git a/crates/roas-cli/src/arazzo.rs b/crates/roas-cli/src/arazzo.rs index 67fe0a88..0fe59085 100644 --- a/crates/roas-cli/src/arazzo.rs +++ b/crates/roas-cli/src/arazzo.rs @@ -11,9 +11,10 @@ use clap::{Subcommand, ValueEnum}; use enumset::EnumSet; use roas_arazzo::validation::{Error as ArazzoError, Validate, ValidationOptions}; use roas_arazzo::{v1_0, v1_1}; -use roas_arazzo_executor::{Client, Options, prepare, required_sources}; +use roas_arazzo_executor::{ + Client, Options, SourceLoadOptions, SourceRegistry, prepare, required_sources, +}; use serde_json::Value; -use std::collections::BTreeMap; use std::path::PathBuf; use url::Url; @@ -141,7 +142,7 @@ pub(crate) enum ArazzoCommand { Convert(ArazzoConvertArgs), /// Run a workflow: perform every step's request and report what /// happened. - Run(ArazzoRunArgs), + Run(Box), /// List the workflows a description offers, and what each one takes. List(ArazzoListArgs), } @@ -227,6 +228,27 @@ pub(crate) struct ArazzoRunArgs { #[arg(long, value_name = "NAME=PATH")] source: Vec, + /// Preload a possible source document by identity, without assigning a root alias. + #[arg(long, value_name = "FILE")] + source_document: Vec, + + /// Traverse every declared source, including linked Arazzo documents. + /// Fetching still requires --load; unrelated failures are reported as diagnostics. + #[arg(long)] + load_all_sources: bool, + + /// Source-graph document/fetch-attempt budget, separate from workflow limits. + #[arg(long, default_value_t = 256, value_name = "N")] + source_max_documents: usize, + + /// Source-graph depth budget (entry document is depth zero). + #[arg(long, default_value_t = 32, value_name = "N")] + source_max_depth: usize, + + /// Allow Arazzo retrieval aliases instead of canonical $self identities. + #[arg(long)] + allow_source_retrieval_aliases: bool, + /// Send a source description's requests somewhere else, e.g. /// `--base-url petStore=http://127.0.0.1:8080` (repeatable) — /// whatever its document says. @@ -274,7 +296,7 @@ pub(crate) fn run_arazzo(cmd: ArazzoCommand) -> Result<()> { match cmd { ArazzoCommand::Validate(args) => run_arazzo_validate(args), ArazzoCommand::Convert(args) => run_arazzo_convert(args), - ArazzoCommand::Run(args) => run_arazzo_run(args), + ArazzoCommand::Run(args) => run_arazzo_run(*args), ArazzoCommand::List(args) => run_arazzo_list(args), } } @@ -398,7 +420,7 @@ fn inputs_of(workflow: &v1_1::Workflow) -> Vec { fn run_arazzo_run(args: ArazzoRunArgs) -> Result<()> { let source = resolve_input_source(args.file.as_deref())?; let (value, input_format) = read_input(&source, args.format)?; - let detected = detect_or_use_arazzo(None, value)?; + let detected = detect_or_use_arazzo(None, value.clone())?; // One interpreter: a v1.0 description is upconverted first. let description = match detected { DetectedArazzo::V1_1(description) => description, @@ -482,9 +504,14 @@ fn run_arazzo_run(args: ArazzoRunArgs) -> Result<()> { if let Some(max_steps) = args.max_steps { options = options.max_steps(max_steps); } - let (options, any) = sources(options, &description, &source, &args)?; + let (options, any, source_diagnostics) = sources(options, &description, value, &source, &args)?; let explain = |error: anyhow::Error| { + let error = if source_diagnostics.is_empty() { + error + } else { + anyhow!("{error}\n{}", source_diagnostics.join("\n")) + }; if any || description.source_descriptions.is_empty() { anyhow!(error) } else { @@ -497,13 +524,20 @@ fn run_arazzo_run(args: ArazzoRunArgs) -> Result<()> { } }; let plan = prepare(&description, &options).map_err(|error| explain(anyhow!(error)))?; + // Preparation failures carry these diagnostics in the returned error. + // Once preparation succeeds, emit optional warnings here, exactly once. + if !args.quiet { + for diagnostic in &source_diagnostics { + eprintln!("- {diagnostic}"); + } + } let report = plan.execute(&mut Client::blocking()).map_err(|failure| { if !args.quiet && let Some(report) = &failure.report { eprint!("{report}"); } - explain(anyhow!(failure.error)) + anyhow!(failure.error) })?; if !args.quiet { @@ -528,67 +562,71 @@ fn run_arazzo_run(args: ArazzoRunArgs) -> Result<()> { /// line, then — where `--load` allows it — those the description points /// at itself. fn sources( - mut options: Options, + options: Options, description: &v1_1::Description, + value: Value, from: &InputSource, args: &ArazzoRunArgs, -) -> Result<(Options, bool)> { - let mut supplied = BTreeMap::new(); +) -> Result<(Options, bool, Vec)> { + let needed = required_sources(description, &options)?; + let mut registry = SourceRegistry::new(); + let root = registry.insert(base_uri(from, None)?.as_str(), value)?; + // Index every explicitly supplied document before linking any source. + let mut supplied = Vec::new(); for source in &args.source { let (name, path) = split_pair(source, "--source")?; - let (document, _) = read_input(&InputSource::File(PathBuf::from(path)), None) + if registry.source(root, name).is_none() { + bail!("`--source {name}=…` names no source description of this document"); + } + let from = InputSource::File(PathBuf::from(path)); + let (document, _) = read_input(&from, None) .with_context(|| format!("reading source description {path}"))?; - supplied.insert(name.to_owned(), document); + let id = registry.insert(base_uri(&from, None)?.as_str(), document)?; + supplied.push((name.to_owned(), id)); } - - let needed = required_sources(description, &options)?; - let mut loader = build_loader(&args.load); - let base = base_uri(from, description.self_.as_deref())?; - let mut any = false; - for declared in &description.source_descriptions { - let url = declared.url.clone(); - let document = match supplied.remove(&declared.name) { - Some(document) => document, - None => { - if !needed.contains(&declared.name) { - continue; - } - let Some(loader) = loader.as_mut() else { - // Nothing to load it with. The executor says so if a - // step turns out to need it, naming the source. - continue; - }; - let uri = base - .join(&url) - .map_err(|error| { - if base.cannot_be_a_base() { - anyhow!( - "`{url}` is relative, and `$self` (`{base}`) is not something a \ - relative reference can be resolved against — give the source an \ - absolute URL, or the description a hierarchical `$self`" - ) - } else { - anyhow!("resolving `{url}` against `{base}`: {error}") - } - })? - .to_string(); - loader - .load_resource(&uri) - .with_context(|| { - format!("loading source description `{}` from {uri}", declared.name) - })? - .clone() - } - }; - options = options.source(declared.name.clone(), url, document); - any = true; + for path in &args.source_document { + let from = InputSource::File(path.clone()); + let (document, _) = read_input(&from, None) + .with_context(|| format!("reading source document {}", path.display()))?; + registry.insert(base_uri(&from, None)?.as_str(), document)?; } - // A `--source` for something the description does not declare is - // more likely a typo than a spare. - if let Some((name, _)) = supplied.into_iter().next() { - bail!("`--source {name}=…` names no source description of this document"); + for (name, id) in supplied { + registry.override_source(root, &name, id)?; } - Ok((options, any)) + let mut selection = needed; + selection.extend( + args.source + .iter() + .filter_map(|source| source.split_once('=').map(|(name, _)| name.to_owned())), + ); + let mut load_options = SourceLoadOptions::default(); + load_options.root_sources = (!args.load_all_sources).then_some(selection); + load_options.max_documents = args.source_max_documents; + load_options.max_depth = args.source_max_depth; + load_options.retrieval_aliases = args.allow_source_retrieval_aliases; + let mut loader = build_loader(&args.load).unwrap_or_default(); + let report = registry.load_sources(root, &mut loader, &load_options)?; + let diagnostics = report + .diagnostics + .iter() + .map(|diagnostic| { + format!( + "{}: {} (`{}`): {}", + registry + .document(diagnostic.owner) + .expect("diagnostic owner") + .identity(), + diagnostic.path, + diagnostic.source_name, + diagnostic.error, + ) + }) + .collect::>(); + let any = registry + .sources(root)? + .iter() + .any(|source| source.target.is_some()); + Ok((options.source_registry(®istry, root)?, any, diagnostics)) } /// What a relative source description URL is resolved against. @@ -839,6 +877,11 @@ mod tests { input: vec!["petId=7".to_owned()], inputs: None, source: vec![format!("petStore={}", openapi.0.display())], + source_document: Vec::new(), + load_all_sources: false, + source_max_documents: 256, + source_max_depth: 32, + allow_source_retrieval_aliases: false, base_url: vec![format!("petStore={base}")], header: vec!["Authorization: Bearer abc".to_owned()], load: Vec::new(), @@ -859,7 +902,7 @@ mod tests { args.base_url = vec!["petStroe=http://127.0.0.1:9".to_owned()]; args.quiet = true; - let error = run_arazzo(ArazzoCommand::Run(args)).unwrap_err(); + let error = run_arazzo(ArazzoCommand::Run(args.into())).unwrap_err(); assert_eq!( error.to_string(), @@ -890,7 +933,7 @@ mod tests { args.workflow = Some("w".to_owned()); args.quiet = true; - let error = run_arazzo(ArazzoCommand::Run(args)).unwrap_err(); + let error = run_arazzo(ArazzoCommand::Run(args.into())).unwrap_err(); assert!( error.to_string().contains("does not validate"), @@ -953,6 +996,101 @@ mod tests { assert_eq!(join.join().unwrap().len(), 1); } + #[test] + fn full_graph_failures_are_located_but_only_required_sources_block_preparation() { + let (description, openapi) = runnable(); + let (mut value, _) = read_input(&InputSource::File(description.0.clone()), None).unwrap(); + value["sourceDescriptions"] + .as_array_mut() + .unwrap() + .push(json!({ + "name": "unavailable", "url": "missing-for-source-graph.json", "type": "openapi" + })); + value["workflows"][0]["steps"][0]["operationId"] = + json!("$sourceDescriptions.petStore.getPetById"); + let parsed = serde_json::from_value(value.clone()).unwrap(); + let mut args = run_args(&description, &openapi, "http://127.0.0.1:1"); + args.load_all_sources = true; + args.load.push(LoaderKind::File); + let (options, any, diagnostics) = sources( + Options::new(), + &parsed, + value.clone(), + &InputSource::File(description.0.clone()), + &args, + ) + .unwrap(); + assert!(any); + assert_eq!(diagnostics.len(), 1); + assert!(diagnostics[0].contains("sourceDescriptions[1].url")); + assert!(diagnostics[0].contains("unavailable")); + prepare(&parsed, &options).unwrap(); + value["workflows"][0]["steps"][0]["operationId"] = json!("getPetById"); + let bare = serde_json::from_value(value).unwrap(); + assert!( + prepare(&bare, &options) + .unwrap_err() + .to_string() + .contains("was not supplied") + ); + + args.source_max_documents = 1; // root + explicitly supplied API exceed the initial budget + let error = run_arazzo_run(args).unwrap_err().to_string(); + assert!(error.contains("source graph document limit (1)"), "{error}"); + } + + #[test] + fn supplied_documents_are_indexed_offline_before_linking_and_follow_identity_policy() { + let (description, openapi) = runnable(); + let (mut value, _) = read_input(&InputSource::File(description.0.clone()), None).unwrap(); + value["workflows"][0]["steps"][0]["operationId"] = + json!("$sourceDescriptions.petStore.getPetById"); + let mut child = value.clone(); + child["$self"] = json!("https://identity.test/child.json"); + child["sourceDescriptions"] = json!([]); + let supplied = TempFile::write("child.json", &child); + value["sourceDescriptions"] + .as_array_mut() + .unwrap() + .push(json!({ + "name": "child", "url": "https://identity.test/child.json", "type": "arazzo" + })); + let mut args = run_args(&description, &openapi, "http://127.0.0.1:1"); + args.load_all_sources = true; + args.source_document.push(supplied.0.clone()); + let from = InputSource::File(description.0.clone()); + let parsed = serde_json::from_value(value.clone()).unwrap(); + let (options, _, diagnostics) = + sources(Options::new(), &parsed, value.clone(), &from, &args).unwrap(); + assert!(diagnostics.is_empty(), "{diagnostics:?}"); + assert_eq!( + options + .source_document("child") + .unwrap() + .identity() + .as_str(), + "https://identity.test/child.json" + ); + + value["sourceDescriptions"][1]["url"] = json!( + base_uri(&InputSource::File(supplied.0.clone()), None) + .unwrap() + .as_str() + ); + let parsed = serde_json::from_value(value.clone()).unwrap(); + let (_, _, diagnostics) = + sources(Options::new(), &parsed, value.clone(), &from, &args).unwrap(); + assert_eq!(diagnostics.len(), 1); + assert!(diagnostics[0].contains("not the Arazzo identity")); + args.allow_source_retrieval_aliases = true; + let (_, _, diagnostics) = sources(Options::new(), &parsed, value, &from, &args).unwrap(); + assert!(diagnostics.is_empty()); + + args.source_document = vec![description.0.with_extension("not-found")]; + let error = run_arazzo_run(args).unwrap_err(); + assert!(format!("{error:#}").contains("reading source document")); + } + #[test] fn run_retains_a_runtime_output_failure_with_or_without_quiet() { let (_, openapi) = runnable(); @@ -1005,15 +1143,18 @@ mod tests { let mut args = run_args(&lenient, &openapi, &base); args.quiet = true; - let strict = run_arazzo(ArazzoCommand::Run(ArazzoRunArgs { - ignore: Vec::new(), - ..clone_args(&args) - })) + let strict = run_arazzo(ArazzoCommand::Run( + ArazzoRunArgs { + ignore: Vec::new(), + ..clone_args(&args) + } + .into(), + )) .unwrap_err(); assert!(strict.to_string().contains("does not validate"), "{strict}"); args.ignore = vec![ValidationOptions::IgnoreEmptyInfoTitle]; - run_arazzo(ArazzoCommand::Run(args)).expect("the check was let pass"); + run_arazzo(ArazzoCommand::Run(args.into())).expect("the check was let pass"); assert_eq!(join.join().expect("the server thread").len(), 1); } @@ -1026,6 +1167,11 @@ mod tests { input: args.input.clone(), inputs: args.inputs.clone(), source: args.source.clone(), + source_document: args.source_document.clone(), + load_all_sources: args.load_all_sources, + source_max_documents: args.source_max_documents, + source_max_depth: args.source_max_depth, + allow_source_retrieval_aliases: args.allow_source_retrieval_aliases, base_url: args.base_url.clone(), header: args.header.clone(), load: args.load.clone(), @@ -1152,7 +1298,7 @@ mod tests { args.workflow = None; args.quiet = true; - let error = run_arazzo(ArazzoCommand::Run(args)).unwrap_err(); + let error = run_arazzo(ArazzoCommand::Run(args.into())).unwrap_err(); assert_eq!( error.to_string(), @@ -1169,7 +1315,7 @@ mod tests { args.workflow = None; args.quiet = true; - run_arazzo(ArazzoCommand::Run(args)).expect("one workflow is no choice at all"); + run_arazzo(ArazzoCommand::Run(args.into())).expect("one workflow is no choice at all"); assert_eq!(join.join().expect("the server thread").len(), 1); } @@ -1234,8 +1380,10 @@ mod tests { let (description, openapi) = runnable(); let (base, join) = server(1, 200, r#"{"id":7,"name":"fluffy"}"#); - run_arazzo(ArazzoCommand::Run(run_args(&description, &openapi, &base))) - .expect("the workflow runs"); + run_arazzo(ArazzoCommand::Run( + run_args(&description, &openapi, &base).into(), + )) + .expect("the workflow runs"); let asked = join.join().expect("the server thread"); assert_eq!(asked.len(), 1); @@ -1247,8 +1395,10 @@ mod tests { let (description, openapi) = runnable(); let (base, join) = server(1, 500, r#"{"error":"gone"}"#); - let error = - run_arazzo(ArazzoCommand::Run(run_args(&description, &openapi, &base))).unwrap_err(); + let error = run_arazzo(ArazzoCommand::Run( + run_args(&description, &openapi, &base).into(), + )) + .unwrap_err(); assert!( error.to_string().contains("workflow `buyPet` failed"), @@ -1268,7 +1418,7 @@ mod tests { args.input = vec!["petId=9".to_owned()]; args.quiet = true; - run_arazzo(ArazzoCommand::Run(args)).expect("the workflow runs"); + run_arazzo(ArazzoCommand::Run(args.into())).expect("the workflow runs"); let asked = join.join().expect("the server thread"); assert!(asked[0].starts_with("GET /pets/9 "), "{}", asked[0]); @@ -1303,7 +1453,7 @@ mod tests { let mut args = run_args(&description, &openapi, "http://127.0.0.1:1"); args.quiet = true; change(&mut args); - let error = run_arazzo(ArazzoCommand::Run(args)).unwrap_err(); + let error = run_arazzo(ArazzoCommand::Run(args.into())).unwrap_err(); assert!( format!("{error:#}").contains(expected), "expected {expected:?}, got: {error:#}" @@ -1322,6 +1472,14 @@ mod tests { "petId=7", "--source", "petStore=./openapi.yaml", + "--source-document", + "./child.yaml", + "--load-all-sources", + "--source-max-documents", + "10", + "--source-max-depth", + "3", + "--allow-source-retrieval-aliases", "--base-url", "petStore=http://127.0.0.1:8080", "--header", @@ -1339,6 +1497,11 @@ mod tests { assert_eq!(a.workflow.as_deref(), Some("buyPet")); assert_eq!(a.input, ["petId=7"]); assert_eq!(a.source, ["petStore=./openapi.yaml"]); + assert_eq!(a.source_document, [PathBuf::from("./child.yaml")]); + assert!(a.load_all_sources); + assert_eq!(a.source_max_documents, 10); + assert_eq!(a.source_max_depth, 3); + assert!(a.allow_source_retrieval_aliases); assert_eq!(a.base_url, ["petStore=http://127.0.0.1:8080"]); assert_eq!(a.header, ["Authorization: Bearer abc"]); assert_eq!(a.max_steps, Some(50)); @@ -1379,6 +1542,11 @@ mod tests { input: vec!["petId=7".to_owned()], inputs: None, source: Vec::new(), + source_document: Vec::new(), + load_all_sources: false, + source_max_documents: 256, + source_max_depth: 32, + allow_source_retrieval_aliases: false, base_url: Vec::new(), header: Vec::new(), load: Vec::new(), @@ -1388,7 +1556,7 @@ mod tests { format: None, output_format: None, }; - let error = run_arazzo(ArazzoCommand::Run(args)).unwrap_err(); + let error = run_arazzo(ArazzoCommand::Run(args.into())).unwrap_err(); let error = format!("{error:#}"); assert!( error.contains("no source description was supplied"), diff --git a/crates/roas-cli/tests/fixtures/source-graph/api.json b/crates/roas-cli/tests/fixtures/source-graph/api.json new file mode 100644 index 00000000..cfcc4f7e --- /dev/null +++ b/crates/roas-cli/tests/fixtures/source-graph/api.json @@ -0,0 +1,5 @@ +{ + "openapi": "3.1.0", + "info": { "title": "API", "version": "1" }, + "paths": { "/check": { "get": { "operationId": "check", "responses": { "200": { "description": "OK" } } } } } +} diff --git a/crates/roas-cli/tests/fixtures/source-graph/left.yaml b/crates/roas-cli/tests/fixtures/source-graph/left.yaml new file mode 100644 index 00000000..571de751 --- /dev/null +++ b/crates/roas-cli/tests/fixtures/source-graph/left.yaml @@ -0,0 +1,8 @@ +arazzo: 1.1.0 +info: {title: Left, version: '1'} +sourceDescriptions: + - {name: shared, url: shared.yaml, type: arazzo} + - {name: api, url: api.json, type: openapi} +workflows: + - workflowId: left + steps: [{stepId: s, operationId: check}] diff --git a/crates/roas-cli/tests/fixtures/source-graph/right.json b/crates/roas-cli/tests/fixtures/source-graph/right.json new file mode 100644 index 00000000..4ddc03bf --- /dev/null +++ b/crates/roas-cli/tests/fixtures/source-graph/right.json @@ -0,0 +1,9 @@ +{ + "arazzo": "1.1.0", + "info": { "title": "Right", "version": "1" }, + "sourceDescriptions": [ + { "name": "shared", "url": "./shared.yaml", "type": "arazzo" }, + { "name": "api", "url": "./api.json", "type": "openapi" } + ], + "workflows": [{ "workflowId": "right", "steps": [{ "stepId": "s", "operationId": "check" }] }] +} diff --git a/crates/roas-cli/tests/fixtures/source-graph/root.json b/crates/roas-cli/tests/fixtures/source-graph/root.json new file mode 100644 index 00000000..53c6655f --- /dev/null +++ b/crates/roas-cli/tests/fixtures/source-graph/root.json @@ -0,0 +1,9 @@ +{ + "arazzo": "1.1.0", + "info": { "title": "Mixed-format graph", "version": "1" }, + "sourceDescriptions": [ + { "name": "left", "url": "left.yaml", "type": "arazzo" }, + { "name": "right", "url": "right.json", "type": "arazzo" } + ], + "workflows": [{ "workflowId": "root", "steps": [{ "stepId": "s", "operationId": "check" }] }] +} diff --git a/crates/roas-cli/tests/fixtures/source-graph/shared.yaml b/crates/roas-cli/tests/fixtures/source-graph/shared.yaml new file mode 100644 index 00000000..9cd171bd --- /dev/null +++ b/crates/roas-cli/tests/fixtures/source-graph/shared.yaml @@ -0,0 +1,7 @@ +arazzo: 1.1.0 +info: {title: Shared, version: '1'} +sourceDescriptions: + - {name: root, url: ./root.json, type: arazzo} +workflows: + - workflowId: shared + steps: [{stepId: s, operationId: check}] diff --git a/crates/roas-cli/tests/source_diagnostics_test.rs b/crates/roas-cli/tests/source_diagnostics_test.rs new file mode 100644 index 00000000..acc0e936 --- /dev/null +++ b/crates/roas-cli/tests/source_diagnostics_test.rs @@ -0,0 +1,148 @@ +use serde_json::json; +use std::io::Write; +use std::process::{Command, Stdio}; + +#[test] +fn prepared_runs_emit_source_warnings_once_except_when_quiet() { + use std::io::{BufRead, BufReader}; + use std::net::TcpListener; + use std::time::{Duration, Instant}; + for outcome in ["success", "failure", "runtime-error"] { + for quiet in [false, true] { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + let base = format!("api=http://{}", listener.local_addr().unwrap()); + let server = std::thread::spawn(move || { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + match listener.accept() { + Ok((mut stream, _)) => { + stream.set_nonblocking(false).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut reader = BufReader::new(&stream); + let mut line = String::new(); + loop { + line.clear(); + assert_ne!(reader.read_line(&mut line).unwrap(), 0); + if line == "\r\n" { + break; + } + } + stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 2\r\nConnection: close\r\n\r\n{}").unwrap(); + break; + } + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + assert!( + Instant::now() < deadline, + "no API request reached the test server" + ); + std::thread::sleep(Duration::from_millis(5)); + } + Err(error) => panic!("{error}"), + } + } + }); + let mut value = json!({ + "arazzo":"1.1.0", "info":{"title":"Diagnostic", "version":"1"}, + "sourceDescriptions":[ + {"name":"api", "url":"https://example.test/api.json", "type":"openapi"}, + {"name":"unused", "url":"https://unavailable.test/unused.json", "type":"openapi"} + ], + "workflows":[{"workflowId":"w", "steps":[{"stepId":"s", "operationId":"$sourceDescriptions.api.check"}]}] + }); + match outcome { + "failure" => { + value["workflows"][0]["steps"][0]["successCriteria"] = + json!([{"condition":"$statusCode == 201"}]) + } + "runtime-error" => { + value["workflows"][0]["steps"][0]["outputs"] = + json!({"missing":"$response.body#/absent"}) + } + _ => {} + } + let fixture = format!( + "api={}/tests/fixtures/source-graph/api.json", + env!("CARGO_MANIFEST_DIR") + ); + let mut command = Command::new(env!("CARGO_BIN_EXE_roas")); + command.args([ + "arazzo", + "run", + "--format", + "json", + "--load-all-sources", + "--source", + &fixture, + "--base-url", + &base, + ]); + if quiet { + command.arg("--quiet"); + } + let mut child = command + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + child + .stdin + .take() + .unwrap() + .write_all(value.to_string().as_bytes()) + .unwrap(); + let output = child.wait_with_output().unwrap(); + server.join().unwrap(); + let stderr = String::from_utf8(output.stderr).unwrap(); + assert_eq!(output.status.success(), outcome == "success", "{stderr}"); + assert_eq!( + stderr.matches("#.sourceDescriptions[1].url").count(), + usize::from(!quiet), + "{outcome}: {stderr}" + ); + assert!(!stderr.contains("DocumentId("), "{stderr}"); + } + } +} + +#[test] +fn failing_run_prints_each_located_source_diagnostic_once_with_or_without_quiet() { + for quiet in [false, true] { + let mut command = Command::new(env!("CARGO_BIN_EXE_roas")); + command.args(["arazzo", "run", "--format", "json"]); + if quiet { + command.arg("--quiet"); + } + let mut child = command + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let value = json!({ + "arazzo":"1.1.0", "info":{"title":"Diagnostic", "version":"1"}, + "sourceDescriptions":[{"name":"api", "url":"https://unavailable.test/openapi.json", "type":"openapi"}], + "workflows":[{"workflowId":"w", "steps":[{"stepId":"s", "operationId":"$sourceDescriptions.api.check"}]}] + }); + child + .stdin + .take() + .unwrap() + .write_all(value.to_string().as_bytes()) + .unwrap(); + let output = child.wait_with_output().unwrap(); + assert!(!output.status.success()); + let stderr = String::from_utf8(output.stderr).unwrap(); + assert_eq!( + stderr.matches("#.sourceDescriptions[0].url").count(), + 1, + "{stderr}" + ); + assert!(!stderr.contains("DocumentId("), "{stderr}"); + assert!(stderr.contains("no fetcher registered"), "{stderr}"); + assert!(stderr.contains("--source ="), "{stderr}"); + } +} diff --git a/crates/roas-cli/tests/source_graph_test.rs b/crates/roas-cli/tests/source_graph_test.rs new file mode 100644 index 00000000..67469187 --- /dev/null +++ b/crates/roas-cli/tests/source_graph_test.rs @@ -0,0 +1,49 @@ +use roas::loader::{Loader, LoaderError, ResourceFetcher}; +use roas_arazzo_executor::{SourceLoadOptions, SourceRegistry}; +use roas_file_fetcher::FileFetcher; +use serde_json::Value; +use std::{cell::RefCell, path::Path, rc::Rc}; +use url::Url; + +struct CountingFiles(Rc>>); + +impl ResourceFetcher for CountingFiles { + fn fetch(&mut self, uri: &Url) -> Result { + self.0.borrow_mut().push(uri.clone()); + FileFetcher::new().fetch(uri) + } +} + +#[test] +fn mixed_json_yaml_file_graph_preserves_cycles_and_fetches_diamonds_once() { + let path = Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/source-graph/root.json"); + let uri = Url::from_file_path(path).unwrap(); + let root_value = serde_json::from_str(include_str!("fixtures/source-graph/root.json")).unwrap(); + let mut registry = SourceRegistry::new(); + let root = registry.insert(uri.as_str(), root_value).unwrap(); + let reads = Rc::default(); + let mut loader = Loader::new(); + loader.register_fetcher("file://", CountingFiles(Rc::clone(&reads))); + let report = registry + .load_sources(root, &mut loader, &SourceLoadOptions::default()) + .unwrap(); + assert!(report.diagnostics.is_empty(), "{report:?}"); + assert_eq!(report.fetch_attempts, 4); + assert_eq!(reads.borrow().len(), 4); + assert_eq!(registry.len(), 5); + assert_eq!(report.cycles.len(), 1); + assert_eq!(report.cycles[0].target, root); + let left = registry.source(root, "left").unwrap().target.unwrap(); + let right = registry.source(root, "right").unwrap().target.unwrap(); + for name in ["shared", "api"] { + assert_eq!( + registry.source(left, name).unwrap().target, + registry.source(right, name).unwrap().target + ); + } + let again = registry + .load_sources(root, &mut loader, &SourceLoadOptions::default()) + .unwrap(); + assert_eq!(again.fetch_attempts, 0); + assert_eq!(again.cycles, report.cycles); +} diff --git a/crates/roas-file-fetcher/Cargo.toml b/crates/roas-file-fetcher/Cargo.toml index e4e8e456..8e930fe7 100644 --- a/crates/roas-file-fetcher/Cargo.toml +++ b/crates/roas-file-fetcher/Cargo.toml @@ -24,7 +24,7 @@ async = ["dep:tokio"] yaml = ["dep:serde_yaml_ng"] [dependencies] -roas = { version = "0.20", path = "../roas" } +roas = { workspace = true, features = ["v3_2"] } serde.workspace = true serde_json.workspace = true serde_yaml_ng = { workspace = true, optional = true } diff --git a/crates/roas-http-fetcher/Cargo.toml b/crates/roas-http-fetcher/Cargo.toml index 314b0bc2..f8079745 100644 --- a/crates/roas-http-fetcher/Cargo.toml +++ b/crates/roas-http-fetcher/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "roas-http-fetcher" -version = "0.2.4" +version = "0.2.5" edition.workspace = true authors.workspace = true license.workspace = true @@ -16,12 +16,12 @@ publish = true [features] default = [] # Parse YAML response bodies in addition to JSON. The fetcher sniffs the -# response Content-Type and falls back to the URL path extension to decide -# which parser to use. +# response Content-Type and falls back to either the requested or final URL +# path extension to decide which parser to use. yaml = ["dep:serde_yaml_ng"] [dependencies] -roas = { version = "0.20", path = "../roas" } +roas = { workspace = true, features = ["v3_2"] } reqwest.workspace = true serde.workspace = true serde_json.workspace = true diff --git a/crates/roas-http-fetcher/README.md b/crates/roas-http-fetcher/README.md index 5c627db9..3990eff2 100644 --- a/crates/roas-http-fetcher/README.md +++ b/crates/roas-http-fetcher/README.md @@ -23,7 +23,7 @@ Both forms are `Clone` so a single fetcher can be registered for both `http://` ```toml [dependencies] -roas-http-fetcher = { version = "0.1", features = ["yaml"] } +roas-http-fetcher = { version = "0.2.5", features = ["yaml"] } ``` ## Usage @@ -58,6 +58,16 @@ A non-2xx HTTP response, transport failure, or unreadable body is surfaced throu [`LoaderError::Fetch`](https://docs.rs/roas/latest/roas/loader/enum.LoaderError.html) with a [`HttpFetchError`](https://docs.rs/roas-http-fetcher/latest/roas_http_fetcher/enum.HttpFetchError.html) source. +The fetchers also implement `fetch_document`, used by `Loader::load_document` and +`load_document_async`, to preserve the final response URL after redirects alongside +the unchanged value. With `yaml`, a missing, empty or octet-stream Content-Type +uses a YAML extension on either the requested or final URL. Thus a `.yaml` URL +redirecting to an extensionless blob retains its original format hint, and an +extensionless URL redirecting to `.yaml` works too. Explicit Content-Type still +takes precedence over both URLs; disabling `yaml` still means JSON only. `with_client` +continues to honor the supplied client's redirect, timeout, proxy and TLS policies; +metadata collection does not install a different client or enable more URI schemes. + ## License Licensed under either of [Apache License, Version 2.0](../../LICENSE-APACHE) or [MIT license](../../LICENSE-MIT) at your option. diff --git a/crates/roas-http-fetcher/src/lib.rs b/crates/roas-http-fetcher/src/lib.rs index cb57404b..1ae2ab2a 100644 --- a/crates/roas-http-fetcher/src/lib.rs +++ b/crates/roas-http-fetcher/src/lib.rs @@ -24,7 +24,10 @@ use reqwest::Client as AsyncClient; use reqwest::StatusCode; use reqwest::blocking::Client; use reqwest::header::CONTENT_TYPE; -use roas::loader::{AsyncResourceFetcher, FetchFuture, LoaderError, ResourceFetcher}; +use roas::loader::{ + AsyncResourceFetcher, DocumentFetchFuture, FetchFuture, LoadedDocument, LoaderError, + ResourceFetcher, +}; #[cfg(feature = "yaml")] use serde::de::Error as _; use serde_json::Value; @@ -122,6 +125,10 @@ impl Default for Fetcher { impl ResourceFetcher for Fetcher { fn fetch(&mut self, uri: &Url) -> Result { + Ok(self.fetch_document(uri)?.document) + } + + fn fetch_document(&mut self, uri: &Url) -> Result { check_scheme(uri)?; let response = self.client.get(uri.as_str()).send().map_err(|source| { fetch_error(uri.as_str().to_string(), HttpFetchError::Request { source }) @@ -141,16 +148,22 @@ impl ResourceFetcher for Fetcher { .and_then(|v| v.to_str().ok()) .map(|s| s.to_string()); + let retrieval = response.url().clone(); let bytes = response.bytes().map_err(|source| { fetch_error(uri.as_str().to_string(), HttpFetchError::Body { source }) })?; - parse_body(uri, content_type.as_deref(), &bytes) + parse_body(uri, &retrieval, content_type.as_deref(), &bytes) + .map(|document| LoadedDocument::new(document, retrieval)) } } impl AsyncResourceFetcher for Fetcher { fn fetch<'a>(&'a mut self, uri: &'a Url) -> FetchFuture<'a> { + Box::pin(async move { Ok(self.fetch_document(uri).await?.document) }) + } + + fn fetch_document<'a>(&'a mut self, uri: &'a Url) -> DocumentFetchFuture<'a> { let client = self.client.clone(); Box::pin(async move { check_scheme(uri)?; @@ -172,11 +185,13 @@ impl AsyncResourceFetcher for Fetcher { .and_then(|v| v.to_str().ok()) .map(|s| s.to_string()); + let retrieval = response.url().clone(); let bytes = response.bytes().await.map_err(|source| { fetch_error(uri.as_str().to_string(), HttpFetchError::Body { source }) })?; - parse_body(uri, content_type.as_deref(), &bytes) + parse_body(uri, &retrieval, content_type.as_deref(), &bytes) + .map(|document| LoadedDocument::new(document, retrieval)) }) } } @@ -188,12 +203,19 @@ fn check_scheme(uri: &Url) -> Result<(), LoaderError> { } } -fn parse_body(uri: &Url, content_type: Option<&str>, bytes: &[u8]) -> Result { - if is_yaml(content_type, uri) { - parse_yaml(uri, bytes) +fn parse_body( + requested: &Url, + retrieval: &Url, + content_type: Option<&str>, + bytes: &[u8], +) -> Result { + // Preserve the original request's extension hint across redirects. Either + // URL may add a YAML hint, but an explicit non-YAML media type wins over both. + if is_yaml(content_type, requested) || is_yaml(content_type, retrieval) { + parse_yaml(retrieval, bytes) } else { serde_json::from_slice(bytes).map_err(|source| LoaderError::Parse { - uri: uri.as_str().to_string(), + uri: retrieval.as_str().to_string(), source, }) } diff --git a/crates/roas-http-fetcher/tests/http_test.rs b/crates/roas-http-fetcher/tests/http_test.rs index 9ad352c9..42f40ef9 100644 --- a/crates/roas-http-fetcher/tests/http_test.rs +++ b/crates/roas-http-fetcher/tests/http_test.rs @@ -1,6 +1,6 @@ use roas::loader::{AsyncResourceFetcher, LoaderError, ResourceFetcher}; use roas_http_fetcher::{AsyncHttpFetcher, HttpFetchError, HttpFetcher}; -use std::io::{Read, Write}; +use std::io::{BufRead, BufReader, Write}; use std::net::{TcpListener, TcpStream}; use std::sync::mpsc::{Sender, channel}; use std::thread::{self, JoinHandle}; @@ -16,6 +16,7 @@ struct TestResponse { status: u16, reason: &'static str, content_type: Option<&'static str>, + location: Option<&'static str>, body: Vec, } @@ -25,6 +26,7 @@ impl TestResponse { status: 200, reason: "OK", content_type: None, + location: None, body, } } @@ -56,6 +58,11 @@ impl TestServer { } match listener.accept() { Ok((stream, _)) => { + // Accepted sockets inherit non-blocking mode on some platforms. + stream.set_nonblocking(false).expect("blocking request"); + stream + .set_read_timeout(Some(std::time::Duration::from_secs(5))) + .expect("request timeout"); let request_line = read_request_line(&stream); let resp = handler(&request_line); write_response(stream, resp); @@ -89,11 +96,20 @@ impl Drop for TestServer { } } -fn read_request_line(mut stream: &TcpStream) -> String { - let mut buf = [0u8; 1024]; - let n = stream.read(&mut buf).expect("read request"); - let raw = String::from_utf8_lossy(&buf[..n]).to_string(); - raw.lines().next().unwrap_or("").to_string() +fn read_request_line(stream: &TcpStream) -> String { + let mut reader = BufReader::new(stream); + let mut request = String::new(); + assert_ne!(reader.read_line(&mut request).expect("read request"), 0); + // Consume complete headers before closing the connection with a response. + let mut header = String::new(); + loop { + header.clear(); + assert_ne!(reader.read_line(&mut header).expect("read header"), 0); + if header == "\r\n" { + break; + } + } + request.trim_end().to_owned() } fn write_response(mut stream: TcpStream, resp: TestResponse) { @@ -106,6 +122,9 @@ fn write_response(mut stream: TcpStream, resp: TestResponse) { if let Some(ct) = resp.content_type { header.push_str(&format!("Content-Type: {ct}\r\n")); } + if let Some(location) = resp.location { + header.push_str(&format!("Location: {location}\r\n")); + } header.push_str("\r\n"); stream.write_all(header.as_bytes()).expect("write header"); stream.write_all(&resp.body).expect("write body"); @@ -119,12 +138,154 @@ fn http_fetcher_returns_parsed_json_on_success() { assert_eq!(value, serde_json::json!({ "hello": "world" })); } +fn redirect_response(request: &str) -> TestResponse { + if request.starts_with("GET /start ") { + TestResponse { + status: 302, + reason: "Found", + content_type: None, + location: Some("/nested/document.json"), + body: Vec::new(), + } + } else { + assert!(request.starts_with("GET /nested/document.json ")); + TestResponse::ok_body(br#"{"$self":"../identity.json","$ref":"./other.json"}"#.to_vec()) + } +} + +#[test] +fn document_metadata_retains_final_redirect_uri_and_caller_redirect_policy() { + let server = TestServer::start(redirect_response); + let mut fetcher = HttpFetcher::new(); + let loaded = fetcher.fetch_document(&server.url("start")).unwrap(); + assert_eq!(loaded.retrieval_uri, server.url("nested/document.json")); + assert_eq!(loaded.document["$self"], "../identity.json"); + assert_eq!(loaded.document["$ref"], "./other.json"); + let client = reqwest::blocking::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .build() + .unwrap(); + let error = HttpFetcher::with_client(client) + .fetch_document(&server.url("start")) + .unwrap_err(); + assert!(matches!(error, LoaderError::Fetch { .. })); +} + +#[tokio::test] +async fn async_document_metadata_retains_final_redirect_uri() { + let server = TestServer::start(redirect_response); + let loaded = AsyncHttpFetcher::new() + .fetch_document(&server.url("start")) + .await + .unwrap(); + assert_eq!(loaded.retrieval_uri, server.url("nested/document.json")); + assert_eq!(loaded.document["$ref"], "./other.json"); + let client = reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .build() + .unwrap(); + assert!(matches!( + AsyncHttpFetcher::with_client(client) + .fetch_document(&server.url("start")) + .await, + Err(LoaderError::Fetch { .. }) + )); +} + +#[cfg(feature = "yaml")] +#[test] +fn redirected_yaml_uses_the_final_extension_when_content_type_is_absent() { + let server = TestServer::start(|request| { + if request.starts_with("GET /start ") { + TestResponse { + location: Some("/document.yaml"), + ..redirect_response(request) + } + } else { + TestResponse::ok_body(b"name: final\n".to_vec()) + } + }); + let loaded = HttpFetcher::new() + .fetch_document(&server.url("start")) + .unwrap(); + assert_eq!(loaded.document, serde_json::json!({"name": "final"})); + assert_eq!(loaded.retrieval_uri, server.url("document.yaml")); +} + +fn yaml_blob_server(content_type: Option<&'static str>) -> TestServer { + TestServer::start(move |request| { + if request.contains(" /source.yaml ") { + TestResponse { + status: 302, + reason: "Found", + content_type: None, + location: Some("/blob"), + body: Vec::new(), + } + } else { + TestResponse { + content_type, + ..TestResponse::ok_body(b"openapi: 3.1.0\n".to_vec()) + } + } + }) +} + +#[test] +fn requested_yaml_hint_survives_redirects_for_both_sync_apis() { + for content_type in [ + None, + Some("application/octet-stream"), + Some("application/json"), + ] { + let server = yaml_blob_server(content_type); + let uri = server.url("source.yaml"); + let mut fetcher = HttpFetcher::new(); + let plain = fetcher.fetch(&uri); + let metadata = fetcher.fetch_document(&uri); + if cfg!(feature = "yaml") && content_type != Some("application/json") { + assert_eq!(plain.unwrap(), serde_json::json!({"openapi": "3.1.0"})); + let loaded = metadata.unwrap(); + assert_eq!(loaded.retrieval_uri, server.url("blob")); + assert_eq!(loaded.document["openapi"], "3.1.0"); + } else { + assert!(matches!(plain, Err(LoaderError::Parse { .. }))); + assert!(matches!(metadata, Err(LoaderError::Parse { .. }))); + } + } +} + +#[tokio::test] +async fn requested_yaml_hint_survives_redirects_for_both_async_apis() { + for content_type in [ + None, + Some("application/octet-stream"), + Some("application/json"), + ] { + let server = yaml_blob_server(content_type); + let uri = server.url("source.yaml"); + let mut fetcher = AsyncHttpFetcher::new(); + let plain = fetcher.fetch(&uri).await; + let metadata = fetcher.fetch_document(&uri).await; + if cfg!(feature = "yaml") && content_type != Some("application/json") { + assert_eq!(plain.unwrap(), serde_json::json!({"openapi": "3.1.0"})); + let loaded = metadata.unwrap(); + assert_eq!(loaded.retrieval_uri, server.url("blob")); + assert_eq!(loaded.document["openapi"], "3.1.0"); + } else { + assert!(matches!(plain, Err(LoaderError::Parse { .. }))); + assert!(matches!(metadata, Err(LoaderError::Parse { .. }))); + } + } +} + #[test] fn http_fetcher_surfaces_non_2xx_as_loader_error_fetch_with_status() { let server = TestServer::start(|_req| TestResponse { status: 404, reason: "Not Found", content_type: None, + location: None, body: b"missing".to_vec(), }); let mut fetcher = HttpFetcher::new(); @@ -264,6 +425,7 @@ async fn async_http_fetcher_surfaces_non_2xx_as_loader_error_fetch_with_status() status: 404, reason: "Not Found", content_type: None, + location: None, body: b"missing".to_vec(), }); let mut fetcher = AsyncHttpFetcher::new(); diff --git a/crates/roas-http-validator/Cargo.toml b/crates/roas-http-validator/Cargo.toml index fbeddf9f..8836e0a4 100644 --- a/crates/roas-http-validator/Cargo.toml +++ b/crates/roas-http-validator/Cargo.toml @@ -42,7 +42,7 @@ rocket = ["dep:rocket"] # were written as; the `serde_json` feature below is the same toggle # for the request bodies this crate parses itself. Both are named # because they are two different reasons to want it. -roas = { version = "0.20", path = "../roas", features = ["v3_2", "exact-numbers"] } +roas = { workspace = true, features = ["v3_2", "exact-numbers"] } percent-encoding.workspace = true regex.workspace = true # The whole point of `decimal.rs`: this keeps a parsed number's diff --git a/crates/roas/Cargo.toml b/crates/roas/Cargo.toml index 68c0fccf..8883cb72 100644 --- a/crates/roas/Cargo.toml +++ b/crates/roas/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "roas" -version = "0.20.0" +version = "0.20.1" edition.workspace = true authors.workspace = true license.workspace = true diff --git a/crates/roas/README.md b/crates/roas/README.md index f54c2a57..277141e7 100644 --- a/crates/roas/README.md +++ b/crates/roas/README.md @@ -79,6 +79,31 @@ For `http(s)://` refs, register a `roas_http_fetcher::HttpFetcher` on the `http://` and `https://` prefixes the same way (it's `Clone`, so one client can serve both). +### Unchanged documents and retrieval metadata + +`Loader::load_document` and `load_document_async` return a `LoadedDocument`: the +complete parsed value without `$ref` rewriting, plus its actual retrieval URI. +This is useful for consumers that implement their own document identity/base rules, +such as Arazzo `$self`. Both APIs use only explicitly registered fetchers. + +Existing fetchers need no changes: the new `fetch_document` trait methods default +to `fetch` and report the requested URI. Redirect-aware fetchers can override them. +`roas-http-fetcher` 0.2.5 supplies the final response URL. + +Raw and existing reference-loading APIs share fetched resources across sync/async +cache hits. The legacy `load_resource` / `resolve_reference` APIs still return references +rewritten against the requested resource URI. Legacy-only callers retain one value +plus a compact journal of changed `$ref` strings, not another full document. A raw +view is reconstructed from that journal only when requested, without another fetch. +Using both views retains both values. `preload_resource` replaces the cache entry, +invalidates its raw view and clears the typed cache as before. + +`load_document_shared` / `load_document_shared_async` return `Arc` +handles to the same raw cache entry, so a source registry and its execution options +can share the value without cloning it. Existing handles remain valid after cache +replacement or loader drop. Loader types/traits are exported at the crate root; +their existing `roas::loader` paths also remain valid. + ## License Licensed under either of [Apache License, Version 2.0](../../LICENSE-APACHE) or [MIT license](../../LICENSE-MIT) at your diff --git a/crates/roas/src/lib.rs b/crates/roas/src/lib.rs index 8a352e85..87b9eb29 100644 --- a/crates/roas/src/lib.rs +++ b/crates/roas/src/lib.rs @@ -12,6 +12,10 @@ pub mod common; pub mod loader; +pub use loader::{ + AsyncResourceFetcher, DocumentFetchFuture, FetchFuture, JsonFileFetcher, LoadedDocument, + Loader, LoaderError, ResourceFetcher, +}; pub mod merge; pub mod validation; diff --git a/crates/roas/src/loader.rs b/crates/roas/src/loader.rs index 824ff95b..20c73161 100644 --- a/crates/roas/src/loader.rs +++ b/crates/roas/src/loader.rs @@ -19,12 +19,46 @@ use url::Url; /// Boxed future returned by resource fetchers. pub type FetchFuture<'a> = Pin> + 'a>>; +/// A complete parsed document and the location it was actually retrieved from. +/// The value is unchanged: in particular, `$ref` and `$self` are not rewritten. +#[derive(Clone, Debug)] +#[non_exhaustive] +pub struct LoadedDocument { + /// Complete JSON-compatible document, before reference rewriting. + pub document: Value, + /// Final retrieval URI, after redirects when the fetcher exposes them. + pub retrieval_uri: Url, +} + +impl LoadedDocument { + /// Associate an unchanged document with its actual retrieval location. + pub fn new(document: Value, mut retrieval_uri: Url) -> Self { + retrieval_uri.set_fragment(None); + Self { + document, + retrieval_uri, + } + } +} + +/// Future returned by metadata-aware asynchronous fetchers. +pub type DocumentFetchFuture<'a> = + Pin> + 'a>>; + /// Fetches and parses resources for the loader. /// /// Fetchers receive the resource URL without its fragment and return a parsed document. /// They do not manage the loader cache. pub trait ResourceFetcher { fn fetch(&mut self, uri: &Url) -> Result; + + /// Fetch without rewriting references, retaining retrieval metadata. + /// Existing fetchers default to the requested location. Redirect-aware + /// fetchers can override this without changing their `fetch` API. + fn fetch_document(&mut self, uri: &Url) -> Result { + self.fetch(uri) + .map(|document| LoadedDocument::new(document, uri.clone())) + } } /// Asynchronously fetches and parses resources for the loader. @@ -33,6 +67,15 @@ pub trait ResourceFetcher { /// document. They do not manage the loader cache. pub trait AsyncResourceFetcher { fn fetch<'a>(&'a mut self, uri: &'a Url) -> FetchFuture<'a>; + + /// Async counterpart of [`ResourceFetcher::fetch_document`]. + fn fetch_document<'a>(&'a mut self, uri: &'a Url) -> DocumentFetchFuture<'a> { + Box::pin(async move { + self.fetch(uri) + .await + .map(|document| LoadedDocument::new(document, uri.clone())) + }) + } } /// JSON file-system fetcher. @@ -123,18 +166,26 @@ pub enum LoaderError { /// External resource loader with a fetcher registry and document cache. /// -/// Two layers of caching are maintained: the raw `Value` cache keyed by -/// resource URI (so the same file/URL is fetched once), and a typed -/// cache keyed by `(reference, TypeId)` (so a `$ref` deserialized into -/// some concrete `T` is parsed only once across the run, regardless of -/// how many places point to it). +/// A resource is fetched once. Legacy-only readers retain a rewritten value and +/// a compact journal of changed reference strings, not a second full document. +/// Unchanged documents are materialized only on demand and can be shared with +/// callers. A typed cache avoids repeatedly deserializing the same `$ref`. pub struct Loader { fetchers: BTreeMap>, async_fetchers: BTreeMap>, cache: BTreeMap, + documents: BTreeMap>, + origins: BTreeMap, typed_cache: BTreeMap<(String, TypeId), Box>, } +/// Enough information to restore the raw view of an immutable rewritten cache +/// entry. Ordinals follow `visit_refs` order; the cached tree is never reordered. +struct ResourceOrigin { + retrieval_uri: Url, + original_refs: Vec<(usize, String)>, +} + impl Loader { /// Create a loader with no registered fetchers. pub fn new() -> Self { @@ -142,6 +193,8 @@ impl Loader { fetchers: BTreeMap::new(), async_fetchers: BTreeMap::new(), cache: BTreeMap::new(), + documents: BTreeMap::new(), + origins: BTreeMap::new(), typed_cache: BTreeMap::new(), } } @@ -193,9 +246,8 @@ impl Loader { document: Value, ) -> Result, LoaderError> { let (key, _) = parse_reference(uri.as_ref())?; - let mut document = document; - rewrite_refs_against(&mut document, &key); - let previous = self.cache.insert(key, document); + self.documents.remove(&key); + let previous = self.cache_resource(key.clone(), LoadedDocument::new(document, key)); if previous.is_some() { self.typed_cache.clear(); } @@ -212,19 +264,11 @@ impl Loader { fn load_resource_by_key(&mut self, key: Url) -> Result<&Value, LoaderError> { if !self.cache.contains_key(&key) { - let fetcher_key = best_fetcher_key(&self.fetchers, key.as_str()).ok_or_else(|| { - LoaderError::NoFetcherRegistered { - uri: key.as_str().to_string(), - } - })?; - let mut parsed = self - .fetchers - .get_mut(&fetcher_key) - .expect("fetcher key came from the registry") - .fetch(&key)?; - rewrite_refs_against(&mut parsed, &key); - - self.cache.insert(key.clone(), parsed); + let loaded = match self.documents.get(&key) { + Some(raw) => raw.as_ref().clone(), // both views explicitly requested + None => self.fetch_document(&key)?, + }; + self.cache_resource(key.clone(), loaded); } Ok(self @@ -244,21 +288,11 @@ impl Loader { async fn load_resource_by_key_async(&mut self, key: Url) -> Result<&Value, LoaderError> { if !self.cache.contains_key(&key) { - let fetcher_key = - best_fetcher_key(&self.async_fetchers, key.as_str()).ok_or_else(|| { - LoaderError::NoFetcherRegistered { - uri: key.as_str().to_string(), - } - })?; - let mut parsed = self - .async_fetchers - .get_mut(&fetcher_key) - .expect("async fetcher key came from the registry") - .fetch(&key) - .await?; - rewrite_refs_against(&mut parsed, &key); - - self.cache.insert(key.clone(), parsed); + let loaded = match self.documents.get(&key) { + Some(raw) => raw.as_ref().clone(), + None => self.fetch_document_async(&key).await?, + }; + self.cache_resource(key.clone(), loaded); } Ok(self @@ -267,6 +301,140 @@ impl Loader { .expect("resource was inserted into the cache")) } + /// Load a complete document without rewriting `$ref` or `$self`. + /// + /// Shares the fetch cache with legacy resource loading. The latter keeps a + /// separate rewritten projection only when requested. Fetch policies and + /// longest-prefix selection are unchanged; no fetcher is enabled implicitly. + pub fn load_document(&mut self, uri: &str) -> Result<&LoadedDocument, LoaderError> { + let (key, _) = parse_reference(uri)?; + self.restore_document(&key); + if !self.documents.contains_key(&key) { + let document = self.fetch_document(&key)?; + self.documents.insert(key.clone(), Arc::new(document)); + } + Ok(self.documents.get(&key).expect("document was cached")) + } + + /// Async counterpart of [`Self::load_document`], using only async fetchers + /// on cache misses. Both loading modes share complete-document cache hits. + pub async fn load_document_async(&mut self, uri: &str) -> Result<&LoadedDocument, LoaderError> { + let (key, _) = parse_reference(uri)?; + self.restore_document(&key); + if !self.documents.contains_key(&key) { + let document = self.fetch_document_async(&key).await?; + self.documents.insert(key.clone(), Arc::new(document)); + } + Ok(self.documents.get(&key).expect("document was cached")) + } + + /// Share the unchanged cached document without cloning its value. The handle + /// stays valid if the loader is dropped or the resource is later preloaded. + /// + /// ``` + /// use roas::Loader; + /// use serde_json::json; + /// use std::sync::Arc; + /// # fn main() -> Result<(), roas::LoaderError> { + /// let mut loader = Loader::new(); + /// loader.preload_resource("https://example.test/api.json", json!({"openapi":"3.1.0"}))?; + /// let first = loader.load_document_shared("https://example.test/api.json")?; + /// let second = loader.load_document_shared("https://example.test/api.json")?; + /// assert!(Arc::ptr_eq(&first, &second)); + /// drop(loader); + /// assert_eq!(first.document["openapi"], "3.1.0"); + /// # Ok(()) } + /// ``` + pub fn load_document_shared(&mut self, uri: &str) -> Result, LoaderError> { + self.load_document(uri)?; + let (key, _) = parse_reference(uri)?; + Ok(Arc::clone( + self.documents.get(&key).expect("document was cached"), + )) + } + + /// Async counterpart of [`Self::load_document_shared`], with the same cache + /// and registered-fetcher policy as [`Self::load_document_async`]. + pub async fn load_document_shared_async( + &mut self, + uri: &str, + ) -> Result, LoaderError> { + self.load_document_async(uri).await?; + let (key, _) = parse_reference(uri)?; + Ok(Arc::clone( + self.documents.get(&key).expect("document was cached"), + )) + } + + fn fetch_document(&mut self, key: &Url) -> Result { + let prefix = best_fetcher_key(&self.fetchers, key.as_str()).ok_or_else(|| { + LoaderError::NoFetcherRegistered { + uri: key.to_string(), + } + })?; + self.fetchers + .get_mut(&prefix) + .expect("registered fetcher") + .fetch_document(key) + } + + async fn fetch_document_async(&mut self, key: &Url) -> Result { + let prefix = best_fetcher_key(&self.async_fetchers, key.as_str()).ok_or_else(|| { + LoaderError::NoFetcherRegistered { + uri: key.to_string(), + } + })?; + self.async_fetchers + .get_mut(&prefix) + .expect("registered async fetcher") + .fetch_document(key) + .await + } + + fn cache_resource(&mut self, key: Url, mut loaded: LoadedDocument) -> Option { + let original_refs = rewrite_refs_against(&mut loaded.document, &key); + self.origins.insert( + key.clone(), + ResourceOrigin { + retrieval_uri: loaded.retrieval_uri, + original_refs, + }, + ); + self.cache.insert(key, loaded.document) + } + + fn restore_document(&mut self, key: &Url) { + if self.documents.contains_key(key) { + return; + } + if let Some(value) = self.cache.get(key) { + let origin = self + .origins + .get(key) + .expect("cached resources retain their origin"); + let mut document = value.clone(); // raw view is now explicitly requested + let mut originals = origin.original_refs.iter().peekable(); + let mut ordinal = 0; + visit_refs(&mut document, &mut |reference| { + if let Some((index, value)) = originals.peek() + && *index == ordinal + { + reference.clone_from(value); + originals.next(); + } + ordinal += 1; + }); + debug_assert!( + originals.next().is_none(), + "immutable cached reference order" + ); + self.documents.insert( + key.clone(), + Arc::new(LoadedDocument::new(document, origin.retrieval_uri.clone())), + ); + } + } + /// Resolve a reference and return the referenced JSON value. /// /// `reference` must include a resource (`common.json#/Pet`, @@ -438,21 +606,33 @@ fn best_fetcher_key(fetchers: &BTreeMap>, uri: &str) - /// Strings that don't successfully `Url::join` (e.g. exotic malformed /// inputs) are left as-is rather than silently corrupted — validation /// downstream will catch them. -fn rewrite_refs_against(value: &mut Value, base: &Url) { +fn rewrite_refs_against(value: &mut Value, base: &Url) -> Vec<(usize, String)> { + let mut originals = Vec::new(); + let mut ordinal = 0; + visit_refs(value, &mut |reference| { + if let Ok(joined) = base.join(reference) + && joined.as_str() != reference + { + originals.push((ordinal, std::mem::replace(reference, joined.to_string()))); + } + ordinal += 1; + }); + originals +} + +fn visit_refs(value: &mut Value, visitor: &mut impl FnMut(&mut String)) { match value { Value::Object(map) => { - if let Some(Value::String(s)) = map.get_mut("$ref") - && let Ok(joined) = base.join(s) - { - *s = joined.to_string(); + if let Some(Value::String(s)) = map.get_mut("$ref") { + visitor(s); } for v in map.values_mut() { - rewrite_refs_against(v, base); + visit_refs(v, visitor); } } Value::Array(items) => { for v in items.iter_mut() { - rewrite_refs_against(v, base); + visit_refs(v, visitor); } } _ => {} @@ -595,6 +775,42 @@ mod tests { assert!(matches!(err, LoaderError::NoFetcherRegistered { .. })); } + #[test] + fn legacy_only_fetches_do_not_retain_a_second_document_tree() { + let uri = "https://example.test/doc.json"; + let mut sync = Loader::new(); + sync.register_fetcher("https://", StaticFetcher::default()); + sync.load_resource(uri).unwrap(); + assert_eq!(sync.cache.len(), 1); + assert!(sync.documents.is_empty()); + + let mut asynchronous = Loader::new(); + asynchronous.register_async_fetcher("https://", AsyncStaticFetcher::default()); + block_on(asynchronous.load_resource_async(uri)).unwrap(); + assert_eq!(asynchronous.cache.len(), 1); + assert!(asynchronous.documents.is_empty()); + + let mut preloaded = Loader::new(); + let value = serde_json::json!({ + "payload": "x".repeat(1024 * 1024), + "a": {"$ref":"./other.json"}, + "b": {"$ref":"https://example.test/unchanged.json"} + }); + preloaded.preload_resource(uri, value).unwrap(); + assert!(preloaded.documents.is_empty()); + let origin = preloaded.origins.get(&Url::parse(uri).unwrap()).unwrap(); + assert_eq!(origin.original_refs, [(0, "./other.json".into())]); + assert_eq!( + preloaded.load_resource(uri).unwrap()["payload"] + .as_str() + .unwrap() + .len(), + 1024 * 1024 + ); + preloaded.load_document(uri).unwrap(); // only an explicit raw read materializes the other view + assert_eq!(preloaded.documents.len(), 1); + } + #[test] fn file_resource_is_fetched_once_and_cached() { let dir = std::env::temp_dir(); diff --git a/crates/roas/tests/loader_document_test.rs b/crates/roas/tests/loader_document_test.rs new file mode 100644 index 00000000..9cd28d9d --- /dev/null +++ b/crates/roas/tests/loader_document_test.rs @@ -0,0 +1,191 @@ +use roas::{ + AsyncResourceFetcher, FetchFuture, LoadedDocument, Loader, LoaderError, ResourceFetcher, +}; +use serde_json::{Value, json}; +use std::cell::Cell; +use std::rc::Rc; +use url::Url; + +struct Fetcher(Rc>); +impl ResourceFetcher for Fetcher { + fn fetch(&mut self, _: &Url) -> Result { + self.0.set(self.0.get() + 1); + Ok(json!({ "$self": "canonical.json", "nested": { "$ref": "other.json#/thing" } })) + } +} +impl AsyncResourceFetcher for Fetcher { + fn fetch<'a>(&'a mut self, uri: &'a Url) -> FetchFuture<'a> { + Box::pin(async move { ResourceFetcher::fetch(self, uri) }) + } +} + +#[test] +fn raw_documents_and_legacy_rewritten_resources_share_one_fetch() { + for raw_first in [true, false] { + let count = Rc::new(Cell::new(0)); + let mut loader = Loader::new(); + loader.register_fetcher("https://", Fetcher(count.clone())); + let uri = "https://example.test/folder/root.json"; + if raw_first { + loader.load_document(uri).unwrap(); + } else { + loader.load_resource(uri).unwrap(); + } + assert_eq!( + loader.load_resource(uri).unwrap()["nested"]["$ref"], + "https://example.test/folder/other.json#/thing" + ); + let raw = loader.load_document(uri).unwrap(); + assert_eq!(raw.document["nested"]["$ref"], "other.json#/thing"); + assert_eq!(raw.document["$self"], "canonical.json"); + assert_eq!(raw.retrieval_uri.as_str(), uri); + assert_eq!(count.get(), 1); + loader + .preload_resource(uri, json!({ "$ref": "replacement.json" })) + .unwrap(); + assert_eq!( + loader.load_document(uri).unwrap().document["$ref"], + "replacement.json" + ); + assert_eq!( + loader.load_resource(uri).unwrap()["$ref"], + "https://example.test/folder/replacement.json" + ); + assert_eq!(count.get(), 1); + } +} + +struct Redirect; +impl ResourceFetcher for Redirect { + fn fetch(&mut self, _: &Url) -> Result { + panic!("metadata method is used") + } + fn fetch_document(&mut self, _: &Url) -> Result { + Ok(LoadedDocument::new( + json!({ "$ref": "relative.json" }), + Url::parse("https://final.test/path/doc.json#discard").unwrap(), + )) + } +} + +#[test] +fn raw_metadata_retains_redirects_without_changing_legacy_rewrite_behavior() { + let uri = "https://original.test/doc.json"; + for raw_first in [true, false] { + let mut loader = Loader::new(); + loader.register_fetcher("https://", Redirect); + if raw_first { + loader.load_document(uri).unwrap(); + } else { + loader.load_resource(uri).unwrap(); + } + let shared = ready(loader.load_document_shared_async(uri)).unwrap(); + assert_eq!( + shared.retrieval_uri.as_str(), + "https://final.test/path/doc.json" + ); + assert_eq!(shared.document["$ref"], "relative.json"); + assert_eq!( + loader.load_resource(uri).unwrap()["$ref"], + "https://original.test/relative.json" + ); + assert!(loader.load_document("http://[invalid").is_err()); + } + assert!(Loader::new().load_document(uri).is_err()); +} + +// The crate has no async runtime dependency: these futures complete immediately. +fn ready(future: impl std::future::Future) -> T { + use std::task::{Context, Poll, Waker}; + match std::pin::pin!(future) + .as_mut() + .poll(&mut Context::from_waker(Waker::noop())) + { + Poll::Ready(value) => value, + Poll::Pending => panic!("memory fetch must complete immediately"), + } +} + +#[test] +fn asynchronous_raw_fetching_and_legacy_loading_share_cache_and_policy() { + let count = Rc::new(Cell::new(0)); + let mut loader = Loader::new(); + loader.register_async_fetcher("https://", Fetcher(count.clone())); + let uri = "https://example.test/doc.json"; + assert!(loader.load_document(uri).is_err()); + assert_eq!( + ready(loader.load_document_async(uri)).unwrap().document["nested"]["$ref"], + "other.json#/thing" + ); + assert_eq!( + ready(loader.load_resource_async(uri)).unwrap()["nested"]["$ref"], + "https://example.test/other.json#/thing" + ); + assert!(loader.load_document(uri).is_ok()); + assert_eq!(count.get(), 1); + assert!(ready(Loader::new().load_document_async(uri)).is_err()); +} + +#[test] +fn shared_raw_views_restore_every_reference_and_survive_cache_replacement() { + use std::sync::Arc; + let uri = "https://example.test/folder/root.json"; + let original = json!({ + "$ref": 42, + "a": [{"$ref":"#/x"}, {"$ref":"https://already.test/x"}, {"$ref":"http://["}, {"$ref":null}], + "b": {"$ref":{"$ref":"../nested.json"}, "": {"$ref":"child.json#/%E2%82%AC"}, "a/b~": {"$ref":""}}, + "c": {"reference":"raw.json", "$self":"identity.json"}, + "x": null + }); + let mut loader = Loader::new(); + loader.preload_resource(uri, original.clone()).unwrap(); + assert_eq!( + loader.load_resource(uri).unwrap()["a"][0]["$ref"], + format!("{uri}#/x") + ); + let shared = loader.load_document_shared(uri).unwrap(); + assert_eq!(shared.document, original); + let asynchronous = ready(loader.load_document_shared_async(uri)).unwrap(); + assert!(Arc::ptr_eq(&shared, &asynchronous)); + assert!(std::ptr::eq( + &shared.document, + &loader.load_document(uri).unwrap().document + )); + let typed = loader + .resolve_reference_as_arc::(&format!("{uri}#/x")) + .unwrap(); + assert_eq!(*typed, Value::Null); + loader.preload_resource(uri, json!({"x":200})).unwrap(); + let replacement = ready(loader.load_document_shared_async(uri)).unwrap(); + assert!(!Arc::ptr_eq(&shared, &replacement)); + assert_eq!( + *loader + .resolve_reference_as_arc::(&format!("{uri}#/x")) + .unwrap(), + json!(200) + ); + drop(loader); + assert_eq!(shared.document, original); + assert_eq!(replacement.document, json!({"x":200})); +} + +#[test] +fn shared_async_first_reads_and_legacy_sync_reads_fetch_once() { + let count = Rc::new(Cell::new(0)); + let mut loader = Loader::new(); + loader.register_async_fetcher("https://", Fetcher(count.clone())); + let uri = "https://example.test/doc.json"; + let shared = ready(loader.load_document_shared_async(uri)).unwrap(); + assert_eq!( + loader.load_resource(uri).unwrap()["nested"]["$ref"], + "https://example.test/other.json#/thing" + ); + assert_eq!(shared.document["nested"]["$ref"], "other.json#/thing"); + assert_eq!(count.get(), 1); + let second = loader.load_document_shared(uri).unwrap(); + assert!(std::sync::Arc::ptr_eq(&shared, &second)); + assert!(loader.load_document_shared("http://[").is_err()); + assert!(ready(loader.load_document_shared_async("http://[")).is_err()); + let _: Option> = None; + let _: roas::loader::JsonFileFetcher = roas::JsonFileFetcher; +}