From 9f0e572a5ec19ff15388b26a7c9dae05dd338a29 Mon Sep 17 00:00:00 2001 From: David W Bitner Date: Fri, 10 Jul 2026 12:30:26 -0500 Subject: [PATCH 1/3] bring streaming search traits and get ready for pgstac 0.10 --- Cargo.toml | 8 +- crates/cli/Cargo.toml | 4 +- crates/cli/src/lib.rs | 60 ++++++- crates/io/src/lib.rs | 2 + crates/io/src/stream.rs | 240 ++++++++++++++++++++++++++++ crates/server/Cargo.toml | 16 +- crates/server/src/backend/pgstac.rs | 176 ++++++-------------- crates/server/src/error.rs | 10 -- 8 files changed, 359 insertions(+), 157 deletions(-) create mode 100644 crates/io/src/stream.rs diff --git a/Cargo.toml b/Cargo.toml index 1c07cbc89..db7936d39 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -74,7 +74,10 @@ mime = "0.3.17" mockito = "1.5" object_store = "0.13.0" parquet = { version = "58.0.0" } -pgstac = "0.4.9" +# pgstac 0.10, from the main pgstac tree (v010io). One crate serves both roles: streaming +# `search --pgstac` (the `search-writer` feature) and the pgstac `serve` backend (the `pool` feature) — +# consumers pick. Shares this workspace's stac + stac-io (see [patch.crates-io]). +pgstac = { version = "0.9.11-dev", path = "/home/bitner/data/pgstac/src/pgstac-rs", default-features = false } quote = "1.0" referencing = { version = "0.46.0", features = ["retrieve-async"] } reqwest = { version = "0.13.1", features = ["query"] } @@ -105,3 +108,6 @@ wkb = "0.9.0" [patch.crates-io] stac = { path = "crates/core" } +# So pgstac 0.10 shares this workspace's stac-io — the one with the streaming ItemCollection writer — +# instead of resolving a separate published copy. +stac-io = { path = "crates/io" } diff --git a/crates/cli/Cargo.toml b/crates/cli/Cargo.toml index cd0136f25..f690f8c7c 100644 --- a/crates/cli/Cargo.toml +++ b/crates/cli/Cargo.toml @@ -13,6 +13,8 @@ rust-version.workspace = true [features] default = [] +# pgstac support: streaming `search --pgstac` (the search-writer StreamSearch backend) + pgstac-backed +# `serve` (via stac-server). Both resolve to the one pgstac 0.10 crate. pgstac = ["dep:pgstac", "stac-server/pgstac"] duckdb-bundled = ["stac-duckdb/bundled"] @@ -24,7 +26,7 @@ clap = { workspace = true, features = ["derive"] } clap_complete.workspace = true futures-core.workspace = true futures-util.workspace = true -pgstac = { workspace = true, optional = true } +pgstac = { workspace = true, optional = true, features = ["search-writer"] } serde_json.workspace = true stac = { version = "0.17.2", path = "../core" } stac-duckdb = { version = "0.3.9", path = "../duckdb" } diff --git a/crates/cli/src/lib.rs b/crates/cli/src/lib.rs index 3e526a67b..5eb807ccd 100644 --- a/crates/cli/src/lib.rs +++ b/crates/cli/src/lib.rs @@ -247,6 +247,18 @@ pub enum Command { value_parser = |s: &str| KeyValue::from_str(s).map(|kv| (kv.0, kv.1)) )] headers: Option, + + /// Request the total match count (the `context` extension's `numberMatched`). + /// + /// Only the `postgresql` implementation honors this; it costs a second query. + #[arg(long = "context")] + context: bool, + + /// Base URL that pagination (`next`/`prev`) links extend, i.e. the search endpoint's own + /// (`self`) URL. Without it, links are relative (`?token=…`). Only the `postgresql` + /// implementation emits pagination links. + #[arg(long = "self-href")] + self_href: Option, }, /// Serves a STAC API. @@ -426,6 +438,8 @@ impl Rustac { ref filter, ref limit, ref headers, + context, + ref self_href, } => { // Infer the search implementation from the href if not explicitly provided let search_impl = search_with.unwrap_or_else(|| { @@ -459,11 +473,53 @@ impl Rustac { SearchImplementation::Postgresql => { #[cfg(feature = "pgstac")] { - pgstac::search(href, search, *max_items).await? + use stac_io::StreamSearch as _; + use std::io::Write as _; + let backend = pgstac::PgstacPool::connect(pgstac::ConnectConfig { + dsn: Some(href.to_string()), + ..Default::default() + }) + .await?; + let dest = outfile + .as_deref() + .and_then(|path| if path == "-" { None } else { Some(path) }); + let pretty = matches!(self.output_format(dest), Format::Json(true)); + if let Some(path) = dest { + let file = std::fs::File::create(path)?; + backend + .write_search( + search, + *max_items, + context, + self_href.clone(), + file, + pretty, + ) + .await + .map_err(Error::from_boxed)?; + } else { + let stdout = std::io::stdout(); + let mut handle = stdout.lock(); + backend + .write_search( + search, + *max_items, + context, + self_href.clone(), + &mut handle, + pretty, + ) + .await + .map_err(Error::from_boxed)?; + handle.flush()?; + } + return Ok(()); } #[cfg(not(feature = "pgstac"))] { - return Err(anyhow!("rustac is not compiled with pgstac support")); + return Err(anyhow!( + "rustac is not compiled with pgstac search support (enable the `pgstac` feature)" + )); } } SearchImplementation::Duckdb => stac_duckdb::search(href, search, *max_items)?, diff --git a/crates/io/src/lib.rs b/crates/io/src/lib.rs index 10c0d278d..ea9ad9e7f 100644 --- a/crates/io/src/lib.rs +++ b/crates/io/src/lib.rs @@ -9,6 +9,7 @@ mod read; mod realized_href; #[cfg(feature = "store")] pub mod store; +pub mod stream; mod write; #[cfg(feature = "geoparquet")] @@ -22,6 +23,7 @@ pub use { ndjson::{FromNdjsonPath, ToNdjsonPath, ndjson_item_reader}, read::read, realized_href::RealizedHref, + stream::{Finalize, ItemStream, StreamSearch, StreamedSearch, write_item_collection}, write::write, }; diff --git a/crates/io/src/stream.rs b/crates/io/src/stream.rs new file mode 100644 index 000000000..4a1de440d --- /dev/null +++ b/crates/io/src/stream.rs @@ -0,0 +1,240 @@ +//! Streaming JSON writer for a search-response [`ItemCollection`](stac::api::ItemCollection): the +//! `features` array is written one item at a time, and the rest of the collection (links, context, +//! counts) is supplied by a `finalize` callback after the items drain (the `next` link needs the last +//! item; a `numberMatched` count may run concurrently). + +use futures::{Stream, StreamExt}; +use serde_json::Value; +use stac::api::{ItemCollection, Search}; +use std::{future::Future, io::Write, pin::Pin}; + +/// A boxed error so backend (item stream, finalize) and writer (IO/serialization) errors compose. +pub type BoxError = Box; + +/// A boxed, pinned stream of serialized STAC items. +pub type ItemStream = Pin> + Send>>; + +/// Produces the finished [`ItemCollection`] (with empty `items`; the writer fills `numberReturned`) from +/// the first item, the last item, and the number written. Called once, after the stream drains. +pub type Finalize = Box< + dyn FnOnce( + Option, + Option, + u64, + ) -> Pin> + Send>> + + Send, +>; + +/// A backend's streamed search: the item stream plus the finalizer for the collection footer. +pub struct StreamedSearch { + /// The response items, streamed one at a time. + pub items: ItemStream, + /// Produces the finished collection once the items drain. + pub finalize: Finalize, +} + +/// A backend that streams a search response as items plus a finished [`ItemCollection`]. How it produces +/// and paginates them is the implementation's concern. +/// +/// `context` requests `numberMatched` (the STAC context extension). `self_href` is the URL this response +/// is served at; pagination links are absolute against it, or relative when `None`. +pub trait StreamSearch: Send + Sync { + /// Begins a streamed search, capped at `max_items` total items. + fn stream_search( + &self, + search: Search, + max_items: Option, + context: bool, + self_href: Option, + ) -> impl Future> + Send; + + /// Drives this backend's streamed search into `writer` as one flat-memory JSON `ItemCollection`, + /// returning the number of items written. + fn write_search( + &self, + search: Search, + max_items: Option, + context: bool, + self_href: Option, + writer: W, + pretty: bool, + ) -> impl Future> { + async move { + let StreamedSearch { items, finalize } = self + .stream_search(search, max_items, context, self_href) + .await?; + write_item_collection(writer, items, pretty, finalize).await + } + } +} + +/// Writes a search response as a streamed `FeatureCollection`: the `features` array is written one item +/// at a time, then `finalize` supplies the footer (links + optional count). The bytes equal +/// `serde_json`-serializing the equivalent [`ItemCollection`], pretty or compact. Returns the item count. +pub async fn write_item_collection( + mut writer: W, + items: S, + pretty: bool, + finalize: F, +) -> Result +where + W: Write, + S: Stream>, + F: FnOnce(Option, Option, u64) -> Fut, + Fut: Future>, +{ + writer.write_all(if pretty { + b"{\n \"type\": \"FeatureCollection\",\n \"features\": [" + } else { + b"{\"type\":\"FeatureCollection\",\"features\":[" + })?; + + futures::pin_mut!(items); + let mut first: Option = None; + let mut pending: Option = None; + let mut count: u64 = 0; + while let Some(item) = items.next().await { + let item = item?; + if let Some(previous) = pending.take() { + write_element(&mut writer, &previous, count, pretty)?; + count += 1; + } else { + first = Some(item.clone()); + } + pending = Some(item); + } + if let Some(last) = &pending { + write_element(&mut writer, last, count, pretty)?; + count += 1; + } + writer.write_all(if pretty && count > 0 { b"\n ]" } else { b"]" })?; + + // The footer is the rest of a real ItemCollection (`links`, `numberMatched`, `numberReturned`, …), + // serialized by serde and spliced in after the streamed features — `type`/`features` dropped since + // they're already written. + let mut collection = finalize(first, pending, count).await?; + collection.number_returned = Some(count); + let value = serde_json::to_value(&collection)?; + let members: serde_json::Map = value + .as_object() + .expect("an ItemCollection serializes to a JSON object") + .iter() + .filter(|(key, _)| key.as_str() != "type" && key.as_str() != "features") + .map(|(key, value)| (key.clone(), value.clone())) + .collect(); + if !members.is_empty() { + let object = if pretty { + serde_json::to_string_pretty(&Value::Object(members))? + } else { + serde_json::to_string(&Value::Object(members))? + }; + let inner = object + .strip_prefix('{') + .and_then(|rest| rest.strip_suffix('}')) + .expect("serde_json serializes an object with braces"); + writer.write_all(b",")?; + writer.write_all(inner.trim_end().as_bytes())?; + } + + writer.write_all(if pretty { b"\n}" } else { b"}" })?; + Ok(count) +} + +/// Writes one item as an element of the `features` array. `index` is the +/// element's position (0-based); a non-zero index gets a leading separator. +fn write_element( + writer: &mut W, + item: &Value, + index: u64, + pretty: bool, +) -> Result<(), BoxError> { + if pretty { + writer.write_all(if index == 0 { b"\n" } else { b",\n" })?; + let element = serde_json::to_string_pretty(item)?; + for (line_index, line) in element.lines().enumerate() { + if line_index > 0 { + writer.write_all(b"\n")?; + } + // Elements sit two levels deep (indent 4) inside the root object. + writer.write_all(b" ")?; + writer.write_all(line.as_bytes())?; + } + } else { + if index > 0 { + writer.write_all(b",")?; + } + serde_json::to_writer(&mut *writer, item)?; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::write_item_collection; + use futures::stream; + use serde_json::Value; + use stac::{Item, Link, api::ItemCollection}; + + /// `n` serialized STAC items and the same items as api items, so the stream + /// input and the expected collection agree byte-for-byte. + fn items(n: usize) -> (Vec, Vec) { + let api: Vec = (0..n) + .map(|i| Item::new(format!("item-{i}")).try_into().unwrap()) + .collect(); + let values = api + .iter() + .map(|i| serde_json::to_value(i).unwrap()) + .collect(); + (values, api) + } + + async fn run( + values: Vec, + links: Vec, + matched: Option, + pretty: bool, + ) -> Vec { + let footer_links = links; + let mut buf = Vec::new(); + write_item_collection( + &mut buf, + stream::iter(values.into_iter().map(Ok)), + pretty, + |_first, _last, _count| async move { + let mut collection = ItemCollection::new(Vec::::new()).unwrap(); + collection.links = footer_links; + collection.number_matched = matched; + Ok(collection) + }, + ) + .await + .unwrap(); + buf + } + + #[tokio::test] + async fn byte_identical_to_buffered() { + let links = vec![Link::new("http://example.com/next?token=abc", "next")]; + for n in [0usize, 1, 2, 5] { + for pretty in [false, true] { + let matched = Some(n as u64 + 100); + let (values, api) = items(n); + let got = run(values, links.clone(), matched, pretty).await; + + let mut want_ic = ItemCollection::new(api).unwrap(); + want_ic.links = links.clone(); + want_ic.number_matched = matched; + let want = if pretty { + serde_json::to_vec_pretty(&want_ic).unwrap() + } else { + serde_json::to_vec(&want_ic).unwrap() + }; + assert_eq!( + String::from_utf8(got).unwrap(), + String::from_utf8(want).unwrap(), + "n={n} pretty={pretty}" + ); + } + } + } +} diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index cf2ea7367..8c7a51a5f 100644 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -14,33 +14,23 @@ rust-version.workspace = true [features] axum = ["dep:axum", "dep:bytes", "dep:mime", "dep:tower-http"] duckdb = ["dep:stac-duckdb", "dep:bb8"] -pgstac = [ - "dep:bb8", - "dep:bb8-postgres", - "dep:pgstac", - "dep:rustls", - "dep:tokio-postgres", - "dep:tokio-postgres-rustls", -] +pgstac = ["dep:pgstac", "dep:futures-util"] [dependencies] axum = { workspace = true, optional = true } bb8 = { workspace = true, optional = true } -bb8-postgres = { workspace = true, optional = true } bytes = { workspace = true, optional = true } futures-core.workspace = true +futures-util = { workspace = true, optional = true } http.workspace = true mime = { workspace = true, optional = true } -pgstac = { workspace = true, optional = true } -rustls = { workspace = true, optional = true } +pgstac = { workspace = true, optional = true, features = ["pool"] } serde.workspace = true serde_json.workspace = true serde_urlencoded.workspace = true stac = { version = "0.17.2", path = "../core", features = ["async"] } stac-duckdb = { version = "0.3.9", path = "../duckdb", features = ["async"], optional = true } thiserror.workspace = true -tokio-postgres = { workspace = true, optional = true } -tokio-postgres-rustls = { workspace = true, optional = true } tower-http = { workspace = true, features = ["cors", "trace"], optional = true } tracing.workspace = true url.workspace = true diff --git a/crates/server/src/backend/pgstac.rs b/crates/server/src/backend/pgstac.rs index b9753e807..c99caa484 100644 --- a/crates/server/src/backend/pgstac.rs +++ b/crates/server/src/backend/pgstac.rs @@ -1,38 +1,24 @@ use crate::{Backend, Error, Result}; -use bb8::Pool; -use bb8_postgres::PostgresConnectionManager; use futures_core::Stream; -use pgstac::Pgstac; -use rustls::{ClientConfig, RootCertStore}; -use serde_json::Map; +use futures_util::StreamExt; +use pgstac::{ConnectConfig, PgstacPool, ingest::ConflictPolicy}; +use serde_json::Value; use stac::api::{ CollectionsClient, ItemCollection, ItemsClient, Search, StreamItemsClient, TransactionClient, - stream_pages, }; use stac::{Collection, Item}; -use tokio_postgres::{ - Socket, - tls::{MakeTlsConnect, TlsConnect}, -}; -use tokio_postgres_rustls::MakeRustlsConnect; /// A backend for a [pgstac](https://github.com/stac-utils/pgstac) database. +/// +/// Wraps pgstac's own [`PgstacPool`] (a `deadpool` pool with rustls TLS). Every request routes through +/// the pool's `stac::api` client impls; this backend only adapts [`pgstac::Error`] to [`Error`]. #[derive(Clone, Debug)] -pub struct PgstacBackend -where - Tls: MakeTlsConnect + Clone + Send + Sync + 'static, - >::Stream: Send + Sync, - >::TlsConnect: Send, - <>::TlsConnect as TlsConnect>::Future: Send, -{ - pool: Pool>, +pub struct PgstacBackend { + pool: PgstacPool, } -impl PgstacBackend { - /// Creates a new PgstacBackend from a string-like configuration. - /// - /// This will use an unverified tls. To provide your own tls, use - /// [PgstacBackend::new_from_stringlike_and_tls]. +impl PgstacBackend { + /// Creates a new `PgstacBackend` from a connection string. /// /// # Examples /// @@ -42,157 +28,87 @@ impl PgstacBackend { /// let backend = PgstacBackend::new_from_stringlike("postgresql://username:password@localhost:5432/postgis").await.unwrap(); /// # }) /// ``` - pub async fn new_from_stringlike( - params: impl ToString, - ) -> Result> { - let _ = rustls::crypto::aws_lc_rs::default_provider() - .install_default() - .expect("The default provider should install without problems"); - let config = ClientConfig::builder() - .with_root_certificates(RootCertStore::empty()) - .with_no_client_auth(); - let tls = MakeRustlsConnect::new(config); - PgstacBackend::new_from_stringlike_and_tls(params, tls).await - } -} - -impl PgstacBackend -where - Tls: MakeTlsConnect + Clone + Send + Sync + 'static, - >::Stream: Send + Sync, - >::TlsConnect: Send, - <>::TlsConnect as TlsConnect>::Future: Send, -{ - /// Creates a new PgstacBackend from a string-like configuration and a tls. - pub async fn new_from_stringlike_and_tls( - params: impl ToString, - tls: Tls, - ) -> Result> { - let params = params.to_string(); - let connection_manager = PostgresConnectionManager::new_from_stringlike(params, tls)?; - let pool = Pool::builder().build(connection_manager).await?; + pub async fn new_from_stringlike(params: impl ToString) -> Result { + let config = ConnectConfig { + dsn: Some(params.to_string()), + ..Default::default() + }; + let pool = PgstacPool::connect(config).await?; Ok(PgstacBackend { pool }) } } -impl ItemsClient for PgstacBackend -where - Tls: MakeTlsConnect + Clone + Send + Sync + 'static, - >::Stream: Send + Sync, - >::TlsConnect: Send, - <>::TlsConnect as TlsConnect>::Future: Send, -{ +impl ItemsClient for PgstacBackend { type Error = Error; async fn search(&self, search: Search) -> Result { - let client = self.pool.get().await?; - let page = client.search(search).await?; - let next_token = page.next_token(); - let prev_token = page.prev_token(); - let mut item_collection = ItemCollection::new(page.features)?; - if let Some(next_token) = next_token { - let mut next = Map::new(); - let _ = next.insert("token".into(), next_token.into()); - item_collection.next = Some(next); - } - if let Some(prev_token) = prev_token { - let mut prev = Map::new(); - let _ = prev.insert("token".into(), prev_token.into()); - item_collection.prev = Some(prev); - } - item_collection.context = page.context; - Ok(item_collection) + self.pool.search(search).await.map_err(Error::from) } async fn item(&self, collection_id: &str, item_id: &str) -> Result> { - let client = self.pool.get().await?; - let value = client.item(item_id, Some(collection_id)).await?; - value - .map(serde_json::from_value) - .transpose() + self.pool + .item(collection_id, item_id) + .await .map_err(Error::from) } } -impl CollectionsClient for PgstacBackend -where - Tls: MakeTlsConnect + Clone + Send + Sync + 'static, - >::Stream: Send + Sync, - >::TlsConnect: Send, - <>::TlsConnect as TlsConnect>::Future: Send, -{ +impl CollectionsClient for PgstacBackend { type Error = Error; async fn collections(&self) -> Result> { - let client = self.pool.get().await?; - let values = client.collections().await?; - values - .into_iter() - .map(|v| serde_json::from_value(v).map_err(Error::from)) - .collect() + self.pool.collections().await.map_err(Error::from) } async fn collection(&self, id: &str) -> Result> { - let client = self.pool.get().await?; - let value = client.collection(id).await?; - value - .map(serde_json::from_value) - .transpose() - .map_err(Error::from) + self.pool.collection(id).await.map_err(Error::from) } } -impl TransactionClient for PgstacBackend -where - Tls: MakeTlsConnect + Clone + Send + Sync + 'static, - >::Stream: Send + Sync, - >::TlsConnect: Send, - <>::TlsConnect as TlsConnect>::Future: Send, -{ +impl TransactionClient for PgstacBackend { type Error = Error; async fn add_collection(&mut self, collection: Collection) -> Result<()> { - let client = self.pool.get().await?; - client.add_collection(collection).await.map_err(Error::from) + self.pool + .add_collection(collection) + .await + .map_err(Error::from) } async fn add_item(&mut self, item: Item) -> Result<()> { - let client = self.pool.get().await?; - client.add_item(item).await.map_err(Error::from) + self.pool.add_item(item).await.map_err(Error::from) } async fn add_items(&mut self, items: Vec) -> Result<()> { tracing::debug!("adding {} items using pgstac loading", items.len()); - let client = self.pool.get().await?; - client.add_items(&items).await.map_err(Error::from) + let values = items + .into_iter() + .map(serde_json::to_value) + .collect::, _>>()?; + self.pool + .create_items(values, ConflictPolicy::Error) + .await + .map(|_| ()) + .map_err(Error::from) } } -impl StreamItemsClient for PgstacBackend -where - Tls: MakeTlsConnect + Clone + Send + Sync + 'static, - >::Stream: Send + Sync, - >::TlsConnect: Send, - <>::TlsConnect as TlsConnect>::Future: Send, -{ +impl StreamItemsClient for PgstacBackend { type Error = Error; async fn search_stream( &self, search: Search, ) -> Result> + Send> { - let page = ItemsClient::search(self, search.clone()).await?; - Ok(stream_pages(self.clone(), search, page)) + // UFCS: PgstacPool has an inherent writer-based `search_stream`; this selects the trait method. + let stream = StreamItemsClient::search_stream(&self.pool, search) + .await + .map_err(Error::from)?; + Ok(stream.map(|result| result.map_err(Error::from))) } } -impl Backend for PgstacBackend -where - Tls: MakeTlsConnect + Clone + Send + Sync + 'static, - >::Stream: Send + Sync, - >::TlsConnect: Send, - <>::TlsConnect as TlsConnect>::Future: Send, -{ +impl Backend for PgstacBackend { fn has_item_search(&self) -> bool { true } diff --git a/crates/server/src/error.rs b/crates/server/src/error.rs index e406c49c0..eff87eb5d 100644 --- a/crates/server/src/error.rs +++ b/crates/server/src/error.rs @@ -4,11 +4,6 @@ use thiserror::Error; #[derive(Debug, Error)] #[non_exhaustive] pub enum Error { - /// [bb8::RunError] - #[cfg(feature = "pgstac")] - #[error(transparent)] - Bb8TokioPostgresRun(#[from] bb8::RunError), - /// [bb8::RunError] #[cfg(feature = "duckdb")] #[error(transparent)] @@ -44,11 +39,6 @@ pub enum Error { #[error("this backend is read-only")] ReadOnly, - /// [tokio_postgres::Error] - #[cfg(feature = "pgstac")] - #[error(transparent)] - TokioPostgres(#[from] tokio_postgres::Error), - /// [std::num::TryFromIntError] #[error(transparent)] TryFromInt(#[from] std::num::TryFromIntError), From 5c068da6a1d9adfa238b410435774b8ce4ff7836 Mon Sep 17 00:00:00 2001 From: David W Bitner Date: Tue, 21 Jul 2026 10:36:02 -0500 Subject: [PATCH 2/3] refactor(io): return stac-io errors from the streaming writer Addresses review feedback on #1088: the `BoxError` alias was public API that had nothing to do with streaming, and the writer returned it rather than the crate's own error. The stream types and `write_item_collection` now use `stac_io::Error`, and backends convert their own errors through a new `Error::Backend` boxed variant. Co-Authored-By: Claude Opus 4.8 --- crates/io/src/error.rs | 4 ++++ crates/io/src/stream.rs | 25 +++++++++---------------- 2 files changed, 13 insertions(+), 16 deletions(-) diff --git a/crates/io/src/error.rs b/crates/io/src/error.rs index 4d1c5b80a..5167f5947 100644 --- a/crates/io/src/error.rs +++ b/crates/io/src/error.rs @@ -4,6 +4,10 @@ use thiserror::Error; #[derive(Error, Debug)] #[non_exhaustive] pub enum Error { + /// An error from a [streaming search](crate::stream) backend, which produces its own error type. + #[error(transparent)] + Backend(#[from] Box), + /// Returned when unable to read a STAC value from a path. #[error("{io}: {path}")] FromPath { diff --git a/crates/io/src/stream.rs b/crates/io/src/stream.rs index 4a1de440d..0e54c3b47 100644 --- a/crates/io/src/stream.rs +++ b/crates/io/src/stream.rs @@ -3,16 +3,14 @@ //! counts) is supplied by a `finalize` callback after the items drain (the `next` link needs the last //! item; a `numberMatched` count may run concurrently). +use crate::Result; use futures::{Stream, StreamExt}; use serde_json::Value; use stac::api::{ItemCollection, Search}; use std::{future::Future, io::Write, pin::Pin}; -/// A boxed error so backend (item stream, finalize) and writer (IO/serialization) errors compose. -pub type BoxError = Box; - /// A boxed, pinned stream of serialized STAC items. -pub type ItemStream = Pin> + Send>>; +pub type ItemStream = Pin> + Send>>; /// Produces the finished [`ItemCollection`] (with empty `items`; the writer fills `numberReturned`) from /// the first item, the last item, and the number written. Called once, after the stream drains. @@ -21,7 +19,7 @@ pub type Finalize = Box< Option, Option, u64, - ) -> Pin> + Send>> + ) -> Pin> + Send>> + Send, >; @@ -46,7 +44,7 @@ pub trait StreamSearch: Send + Sync { max_items: Option, context: bool, self_href: Option, - ) -> impl Future> + Send; + ) -> impl Future> + Send; /// Drives this backend's streamed search into `writer` as one flat-memory JSON `ItemCollection`, /// returning the number of items written. @@ -58,7 +56,7 @@ pub trait StreamSearch: Send + Sync { self_href: Option, writer: W, pretty: bool, - ) -> impl Future> { + ) -> impl Future> { async move { let StreamedSearch { items, finalize } = self .stream_search(search, max_items, context, self_href) @@ -76,12 +74,12 @@ pub async fn write_item_collection( items: S, pretty: bool, finalize: F, -) -> Result +) -> Result where W: Write, - S: Stream>, + S: Stream>, F: FnOnce(Option, Option, u64) -> Fut, - Fut: Future>, + Fut: Future>, { writer.write_all(if pretty { b"{\n \"type\": \"FeatureCollection\",\n \"features\": [" @@ -142,12 +140,7 @@ where /// Writes one item as an element of the `features` array. `index` is the /// element's position (0-based); a non-zero index gets a leading separator. -fn write_element( - writer: &mut W, - item: &Value, - index: u64, - pretty: bool, -) -> Result<(), BoxError> { +fn write_element(writer: &mut W, item: &Value, index: u64, pretty: bool) -> Result<()> { if pretty { writer.write_all(if index == 0 { b"\n" } else { b",\n" })?; let element = serde_json::to_string_pretty(item)?; From 710dffb483640d99595e6a56026544c21f42de63 Mon Sep 17 00:00:00 2001 From: David W Bitner Date: Tue, 21 Jul 2026 10:38:02 -0500 Subject: [PATCH 3/3] refactor(cli): propagate stac-io errors from the streaming search writer Follows the #1088 review change: `write_search` now returns `stac_io::Error`, so the CLI no longer needs `anyhow::Error::from_boxed`. Co-Authored-By: Claude Opus 4.8 --- crates/cli/src/lib.rs | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/crates/cli/src/lib.rs b/crates/cli/src/lib.rs index 5eb807ccd..4567f3793 100644 --- a/crates/cli/src/lib.rs +++ b/crates/cli/src/lib.rs @@ -495,8 +495,7 @@ impl Rustac { file, pretty, ) - .await - .map_err(Error::from_boxed)?; + .await?; } else { let stdout = std::io::stdout(); let mut handle = stdout.lock(); @@ -509,8 +508,7 @@ impl Rustac { &mut handle, pretty, ) - .await - .map_err(Error::from_boxed)?; + .await?; handle.flush()?; } return Ok(());