diff --git a/crates/data/src/ogc.rs b/crates/data/src/ogc.rs index 83b8ae1ba..d982b6071 100644 --- a/crates/data/src/ogc.rs +++ b/crates/data/src/ogc.rs @@ -1252,12 +1252,14 @@ fn arcgis(url: &str, v: &serde_json::Value) -> Result { } else { return Err("an ArcGIS address, but not a FeatureServer, MapServer, ImageServer or VectorTileServer".into()); } - // A feature layer is one request: the server's record limit bounds it (a cut-off answer is - // pointed out when it lands). + // A feature layer is read past the server's record limit, page by page, where the layer + // supports it ([`read_pages`]). A cut-off answer is pointed out when it lands. if let Some(max) = v.get("maxRecordCount").and_then(|m| m.as_i64()) && service.layers.iter().any(|l| matches!(l.offer, Offer::Features { .. })) { - service.notes.push(format!("a feature layer is read up to the server's {max} features per request")); + service.notes.push(format!( + "a feature layer is read up to {MAX_PAGED_FEATURES} features, {max} per request, where the server pages its layers" + )); } // A secured service's token goes with every request. if let Some(token) = query_value(url, "token").filter(|t| !t.is_empty()) { @@ -1479,11 +1481,7 @@ fn ogc_api(url: &str, v: &serde_json::Value) -> Result { } service.incomplete = unreadable > 0 || collections.len() > MAX_LAYERS; if service.layers.iter().any(|l| matches!(l.offer, Offer::Features { .. })) { - service.notes.push(if cfg!(target_arch = "wasm32") { - format!("a collection is read up to {OGC_API_LIMIT} features (one page of the server's)") - } else { - format!("a collection is read up to {MAX_PAGED_FEATURES} features, page by page") - }); + service.notes.push(format!("a collection is read up to {MAX_PAGED_FEATURES} features, page by page")); } if tiled > 0 { service.kind = if service.layers.iter().any(|l| matches!(l.offer, Offer::Features { .. })) { @@ -1691,60 +1689,300 @@ fn next_page(url: &str, page: &serde_json::Value) -> Option { resolve(url, link.get("href")?.as_str()?) } -/// A feature collection read page by page: `first` (from `url`), then each `next` page `get` -/// returns, up to `max` features. The pages' features in one collection, with the first page's -/// other members and without its paging ones, and a note when some features were left unread. A -/// first page without a `next` link comes back unchanged. +/// Whether an ArcGIS answer says it stopped at the server's record limit (`exceededTransferLimit`, +/// a member of the collection or, in GeoJSON, of its `properties`). +fn exceeded(page: &serde_json::Value) -> bool { + [page.get("exceededTransferLimit"), page.pointer("/properties/exceededTransferLimit")] + .iter() + .any(|v| v.and_then(|v| v.as_bool()) == Some(true)) +} + +/// Whether an address is an ArcGIS layer's query (`…/FeatureServer/0/query`, `…/MapServer/3/query`). +fn is_arcgis_query(url: &str) -> bool { + let b = base(url).trim_end_matches('/').to_ascii_lowercase(); + b.ends_with("/query") && (b.contains("/featureserver/") || b.contains("/mapserver/")) +} + +/// How an ArcGIS layer is read past its record limit: pages of `size` features, in the order of +/// its object ids (`order`) so that no feature is skipped or read twice. +#[derive(Debug, PartialEq)] +struct ArcgisPaging { + size: Option, + order: Option, +} + +/// How a layer's description (`…/FeatureServer/0?f=json`) lets it be read in pages: `None` unless it +/// says it supports them (`advancedQueryCapabilities.supportsPagination`, from ArcGIS Server 10.3). +/// Its `maxRecordCount` is the page size when it is a count RustGIS would ask for; its object id +/// field, the order, when it is a plain field name (it goes into the address). +fn arcgis_paging(layer: &serde_json::Value) -> Option { + if layer.pointer("/advancedQueryCapabilities/supportsPagination").and_then(|p| p.as_bool()) != Some(true) { + return None; + } + let size = layer + .get("maxRecordCount") + .and_then(|m| m.as_u64()) + .and_then(|m| usize::try_from(m).ok()) + .filter(|m| (1..=MAX_PAGED_FEATURES).contains(m)); + let oid_field = || { + layer + .get("fields")? + .as_array()? + .iter() + .find(|f| f.get("type").and_then(|t| t.as_str()) == Some("esriFieldTypeOID"))? + .get("name")? + .as_str() + }; + let order = layer + .get("objectIdField") + .and_then(|o| o.as_str()) + .or_else(oid_field) + .filter(|n| !n.is_empty() && n.len() <= 128 && n.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_')) + .map(str::to_string); + Some(ArcgisPaging { size, order }) +} + +/// The description of the layer an ArcGIS query reads (`…/FeatureServer/0?f=json`), with the +/// query's token. +fn arcgis_layer_url(query: &str) -> String { + let b = base(query).trim_end_matches('/'); + let layer = b.get(..b.len().saturating_sub("/query".len())).unwrap_or(b); + match query_value(query, "token").filter(|t| !t.is_empty()) { + Some(token) => format!("{layer}?f=json&token={token}"), + None => format!("{layer}?f=json"), + } +} + +/// The page of an ArcGIS query that starts after `offset` features (`resultOffset`), `size` long. +fn arcgis_page_url(query: &str, offset: usize, size: usize, order: Option<&str>) -> String { + let url = with_query(query, &format!("resultOffset={offset}&resultRecordCount={size}")); + match order { + Some(o) if query_value(&url, "orderByFields").is_none() => with_query(&url, &format!("orderByFields={o}")), + _ => url, + } +} + +/// A feature collection read page by page: `first` (from `url`), then each page `get` returns, up +/// to `max` features, as [`read_pages`] leads (the blocking way to drive it). The pages' features +/// in one collection, and a note when some features were left unread; a first page that leads +/// nowhere comes back unchanged. pub fn join_pages( url: &str, first: Vec, max: usize, mut get: impl FnMut(&str) -> Result, String>, ) -> (Vec, Option) { - let (head, tail) = edges(&first); - if !(head.contains("\"next\"") || tail.contains("\"next\"")) { - return (first, None); + let mut step = read_pages(url, first, max); + loop { + match step { + Step::Done(bytes, note) => return (bytes, note), + Step::Fetch(pages) => { + let answer = get(pages.next_url()); + step = (*pages).land(answer); + } + } } - let Ok(mut doc) = serde_json::from_slice::(&first) else { return (first, None) }; - let Some(mut next) = next_page(url, &doc) else { return (first, None) }; - let Some(mut features) = doc.get_mut("features").and_then(|f| f.as_array_mut()).map(std::mem::take) else { - return (first, None); +} + +/// Where reading a collection page by page has got to ([`read_pages`]). +#[derive(Debug)] +pub enum Step { + /// The next answer to fetch ([`Pages::next_url`]), to hand to [`Pages::land`]. + Fetch(Box), + /// The collection as read (the pages joined, or the first page as it came) and a note when + /// some features were left unread. + Done(Vec, Option), +} + +/// A feature collection being read page by page, sans I/O: it says which address to fetch next +/// and takes the answer, so a blocking command ([`join_pages`]) and the app, which can't wait, +/// read the same way. +#[derive(Debug)] +pub struct Pages { + /// The first page's address. + url: String, + /// The first page, without its features. + doc: serde_json::Value, + features: Vec, + max: usize, + seen: std::collections::HashSet, + /// The address to fetch now, and the one it was found from. + next: String, + from: String, + phase: Phase, +} + +/// What the answer being fetched is. +#[derive(Debug)] +enum Phase { + /// An ArcGIS layer's description, which says whether it can be read in pages; `first` is the + /// first page as it came, kept for when it can't. + Describe { first: Vec }, + /// An OGC API or WFS page, which links to the next. + Linked, + /// An ArcGIS page, `size` long from an offset, in `order`. `start` is the first such page's + /// first feature (a server that ignores the offset sends it again), once it is read; until + /// then `first` keeps the first page as it came, to fall back on. + Offsets { size: usize, order: Option, start: Option, first: Option> }, +} + +/// Starts reading a feature collection page by page from its first page (`first`, from `url`), up +/// to `max` features. An OGC API or WFS page leads to the next by its `next` link. An ArcGIS +/// query that stopped at the server's record limit (`exceededTransferLimit`) is read again in +/// pages (`resultOffset`, `resultRecordCount`, in the order of its object ids) when the layer's +/// description says it can be (fetched first: `advancedQueryCapabilities.supportsPagination`): +/// from the start, since the first page, asked for in no order, may not be the paged order's. The pages' features end up in one collection, +/// with the first page's other members and without its paging ones. A first page that leads +/// nowhere is done at once, unchanged (and [`partial_note`] says what it holds). +pub fn read_pages(url: &str, first: Vec, max: usize) -> Step { + let (head, tail) = edges(&first); + let linked = head.contains("\"next\"") || tail.contains("\"next\""); + // A query that asks for a window of its own (`resultOffset`, `resultRecordCount`) is read as asked. + let limited = is_arcgis_query(url) + && ["resultOffset", "resultRecordCount"].iter().all(|k| query_value(url, k).is_none()) + && (is_true(&head, "exceededTransferLimit") || is_true(&tail, "exceededTransferLimit")); + if !(linked || limited) { + return Step::Done(first, None); + } + let Ok(mut doc) = serde_json::from_slice::(&first) else { return Step::Done(first, None) }; + let Some(features) = doc.get_mut("features").and_then(|f| f.as_array_mut()).map(std::mem::take) else { + return Step::Done(first, None); }; - let mut seen = std::collections::HashSet::from([url.to_string()]); - let note = loop { - if features.len() >= max { - features.truncate(max); - break Some(format!("the first {max} features are read; the rest aren't")); - } - if !seen.insert(next.clone()) { - break Some("the service's pages lead back to an earlier page; the rest aren't read".into()); + let (next, phase) = if linked && let Some(next) = next_page(url, &doc) { + (next, Phase::Linked) + } else if limited && exceeded(&doc) { + (arcgis_layer_url(url), Phase::Describe { first }) + } else { + return Step::Done(first, None); + }; + let pages = Pages { + url: url.to_string(), + doc, + features, + max, + seen: std::collections::HashSet::from([url.to_string()]), + next, + from: url.to_string(), + phase, + }; + match pages.phase { + Phase::Linked => pages.advance(), + _ => Step::Fetch(Box::new(pages)), + } +} + +impl Pages { + /// The address to fetch now. + pub fn next_url(&self) -> &str { + &self.next + } + + /// The address [`next_url`](Self::next_url) was found from (the page or query that leads to + /// it), which it may only point where that one may. + pub fn referrer(&self) -> &str { + &self.from + } + + /// How many features have been read so far. + pub fn read(&self) -> usize { + self.features.len() + } + + /// Takes the answer to [`next_url`](Self::next_url) (or why it couldn't be fetched): the next + /// address to fetch, or the collection as read. + pub fn land(mut self, answer: Result, String>) -> Step { + if let Phase::Describe { first } = &mut self.phase { + let first = std::mem::take(first); + let paging = answer.ok().and_then(|b| serde_json::from_slice(&b).ok()).and_then(|l| arcgis_paging(&l)); + let Some(paging) = paging else { return Step::Done(first, None) }; + let size = paging.size.unwrap_or(self.features.len()); + if size == 0 { + return Step::Done(first, None); + } + // Offsets count in one order: the first page, asked for in none, may hold other + // features than the paged order's first, so the layer is read again from the start. + self.features.clear(); + self.next = arcgis_page_url(&self.url, 0, size, paging.order.as_deref()); + self.from = self.url.clone(); + self.phase = Phase::Offsets { size, order: paging.order, start: None, first: Some(first) }; + return self.advance(); } - let mut page = match get(&next).and_then(|b| serde_json::from_slice::(&b).map_err(|e| e.to_string())) { + let read = self.features.len(); + let mut page = match answer.and_then(|b| serde_json::from_slice::(&b).map_err(|e| e.to_string())) { Ok(page) => page, - Err(e) => break Some(format!("the page after feature {} couldn't be read ({e}); the rest aren't read", features.len())), + Err(e) => return self.stop(format!("the page after feature {read} couldn't be read ({e}); the rest aren't read")), }; let Some(more) = page.get_mut("features").and_then(|f| f.as_array_mut()).map(std::mem::take) else { - break Some(format!("the page after feature {} isn't a feature collection; the rest aren't read", features.len())); + return self.stop(format!("the page after feature {read} isn't a feature collection; the rest aren't read")); + }; + let following = match &mut self.phase { + Phase::Offsets { size, order, start, first } => { + // A server that doesn't honour `resultOffset` sends its first page again. + if start.is_some() && more.first() == start.as_ref() { + return self.stop("the server sent its first features again instead of the next ones; the rest aren't read".into()); + } + if start.is_none() && !more.is_empty() { + *start = more.first().cloned(); + *first = None; + } + exceeded(&page).then(|| arcgis_page_url(&self.url, read + more.len(), *size, order.as_deref())) + } + _ => next_page(&self.next, &page), }; - features.extend(more); - if features.len() > max { - features.truncate(max); - break Some(format!("the first {max} features are read; the rest aren't")); + self.features.extend(more); + if self.features.len() > self.max { + return self.finish_at_max(); } - match next_page(&next, &page) { - Some(n) => next = n, - None => break None, + match following { + Some(n) => { + self.from = std::mem::replace(&mut self.next, n); + self.advance() + } + None => self.finish(None), } - }; - if let Some(members) = doc.as_object_mut() { - for paging in ["links", "numberMatched", "numberReturned"] { - members.remove(paging); + } + + /// Done early, with a note; or, when an ArcGIS layer's paged reading stops before any of its + /// features are read, with the first page as it came (which says it was cut off). + fn stop(mut self, note: String) -> Step { + if let Phase::Offsets { first: Some(first), .. } = &mut self.phase { + return Step::Done(std::mem::take(first), None); } - members.insert("features".into(), serde_json::Value::Array(features)); + self.finish(Some(note)) } - match serde_json::to_vec(&doc) { - Ok(joined) => (joined, note), - Err(_) => (first, None), + + /// The next page fetched, unless the features are all read that may be or the pages lead + /// back to one already read. + fn advance(mut self) -> Step { + if self.features.len() >= self.max { + return self.finish_at_max(); + } + if !self.seen.insert(self.next.clone()) { + return self.stop("the service's pages lead back to an earlier page; the rest aren't read".into()); + } + Step::Fetch(Box::new(self)) + } + + /// Done with the first `max` features. + fn finish_at_max(mut self) -> Step { + let max = self.max; + self.features.truncate(max); + self.finish(Some(format!("the first {max} features are read; the rest aren't"))) + } + + /// The pages' features in the first page's collection, without its paging members. + fn finish(mut self, note: Option) -> Step { + if let Some(members) = self.doc.as_object_mut() { + for paging in ["links", "numberMatched", "numberReturned", "exceededTransferLimit"] { + members.remove(paging); + } + if let Some(p) = members.get_mut("properties").and_then(|p| p.as_object_mut()) { + p.remove("exceededTransferLimit"); + } + members.insert("features".into(), serde_json::Value::Array(self.features)); + } + // A JSON value always writes; an empty answer would be said as unreadable. + Step::Done(serde_json::to_vec(&self.doc).unwrap_or_default(), note) } } @@ -2102,6 +2340,148 @@ mod tests { assert_eq!(join_pages(url, html_only.to_vec(), 100, unasked), (html_only.to_vec(), None)); } + /// An ArcGIS layer past its record limit: read again from the start in pages of the layer's + /// `maxRecordCount` ordered by its object ids (its first page, asked for in no order, may hold + /// other features than the ordered first: Hawaii's zoning did), each from where the last + /// stopped, until a page no longer says it was cut off. A layer that can't be paged, or whose + /// paging fails before a page is read, comes back as it was (with its note). + #[test] + fn arcgis_pages_joined_from_offsets() { + let query = "https://x.org/arcgis/rest/services/P/FeatureServer/0/query?where=1%3D1&outFields=*&f=geojson&token=t"; + // A page of the features `from..to` (of 5), cut off unless it reaches the last. + let page = |from: u64, to: u64| { + let features: Vec<_> = (from..to).map(|i| serde_json::json!({"type": "Feature", "id": i, "properties": {}})).collect(); + let mut p = serde_json::json!({"type": "FeatureCollection", "features": features}); + if to < 5 { + p["properties"] = serde_json::json!({"exceededTransferLimit": true}); + } + serde_json::to_vec(&p).unwrap() + }; + let layer = |extra: &str| { + format!(r#"{{"currentVersion": 11.1, "objectIdField": "OBJECTID", "advancedQueryCapabilities": {{"supportsPagination": true}}{extra}}}"#) + .into_bytes() + }; + let ids = |b: &[u8]| -> Vec { + let v: serde_json::Value = serde_json::from_slice(b).unwrap(); + v["features"].as_array().unwrap().iter().map(|f| f["id"].as_u64().unwrap()).collect() + }; + // Answers by address, keeping the addresses asked for. + let server = |info: Vec| { + move |asked: &mut Vec, u: &str| -> Result, String> { + asked.push(u.to_string()); + if u == "https://x.org/arcgis/rest/services/P/FeatureServer/0?f=json&token=t" { + return Ok(info.clone()); + } + let at = |k: &str| query_value(u, k).and_then(|v| v.parse::().ok()); + match (at("resultOffset"), at("resultRecordCount")) { + (Some(o), Some(n)) => Ok(page(o, (o + n).min(5))), + _ => Err("http status: 404".into()), + } + } + }; + // The first page in the server's own order: two of the five, not the ordered first two. + let unordered = br#"{"type":"FeatureCollection","features":[{"id":3},{"id":0}],"properties":{"exceededTransferLimit":true}}"#; + let mut asked = Vec::new(); + let answer = server(layer(r#", "maxRecordCount": 2"#)); + let (joined, note) = join_pages(query, unordered.to_vec(), 100, |u| answer(&mut asked, u)); + assert_eq!((ids(&joined), note), (vec![0, 1, 2, 3, 4], None)); + assert_eq!(asked[1], format!("{query}&resultOffset=0&resultRecordCount=2&orderByFields=OBJECTID")); + assert!(asked[2].contains("resultOffset=2&resultRecordCount=2") && asked[3].contains("resultOffset=4&resultRecordCount=2")); + assert_eq!(asked.len(), 4); + let v: serde_json::Value = serde_json::from_slice(&joined).unwrap(); + assert!(v.pointer("/properties/exceededTransferLimit").is_none()); + assert!(partial_note(&joined).is_none()); + // The limit cuts it short, with a note. + let answer = server(layer(r#", "maxRecordCount": 2"#)); + let (joined, note) = join_pages(query, page(0, 2), 3, |u| answer(&mut Vec::new(), u)); + assert_eq!(ids(&joined), [0, 1, 2]); + assert!(note.is_some_and(|n| n.contains("first 3 features"))); + // A hostile `maxRecordCount` isn't asked for: the pages are as long as the first. + for hostile in + [r#", "maxRecordCount": 0"#, r#", "maxRecordCount": -5"#, r#", "maxRecordCount": 1e300"#, r#", "maxRecordCount": "x""#, ""] + { + let mut asked = Vec::new(); + let answer = server(layer(hostile)); + let (joined, note) = join_pages(query, page(0, 2), 100, |u| answer(&mut asked, u)); + assert_eq!((ids(&joined), note), (vec![0, 1, 2, 3, 4], None), "{hostile}"); + assert!(asked[1].contains("resultOffset=0&resultRecordCount=2&"), "{hostile}: {asked:?}"); + } + let answer = server(layer(r#", "maxRecordCount": 18446744073709551615"#)); + assert_eq!(ids(&join_pages(query, page(0, 2), 100, |u| answer(&mut Vec::new(), u)).0), [0, 1, 2, 3, 4]); + // A layer that doesn't page (or says nothing of it, before 10.3), or whose description + // can't be read: the first page as it came, which says it was cut off. + for info in [ + br#"{"currentVersion": 11, "advancedQueryCapabilities": {"supportsPagination": false}}"#.to_vec(), + br#"{"currentVersion": 10.22, "maxRecordCount": 2}"#.to_vec(), + b"".to_vec(), + ] { + let mut asked = Vec::new(); + let answer = server(info); + let (joined, note) = join_pages(query, page(0, 2), 100, |u| answer(&mut asked, u)); + assert_eq!((joined, note), (page(0, 2), None)); + assert_eq!(asked.len(), 1); + assert!(partial_note(&page(0, 2)).is_some_and(|n| n.contains("record limit"))); + } + // A server that ignores `resultOffset` sends its first page again: not read twice. + let (joined, note) = + join_pages(query, page(0, 2), 100, |u| if u.contains("resultOffset") { Ok(page(0, 2)) } else { Ok(layer("")) }); + assert_eq!(ids(&joined), [0, 1]); + assert!(note.is_some_and(|n| n.contains("first features again"))); + // Paging that fails before a page is read (a page that adds nothing would ask for itself + // again; a page that can't be fetched): the first page as it came. + let empty = br#"{"type":"FeatureCollection","features":[],"properties":{"exceededTransferLimit":true}}"#; + let (joined, note) = + join_pages(query, page(0, 2), 100, |u| if u.contains("resultOffset") { Ok(empty.to_vec()) } else { Ok(layer("")) }); + assert_eq!((joined, note), (page(0, 2), None)); + let (joined, note) = + join_pages(query, page(0, 2), 100, |u| if u.contains("resultOffset") { Err("http status: 500".into()) } else { Ok(layer("")) }); + assert_eq!((joined, note), (page(0, 2), None)); + // A page that fails later ends the reading with what was read, and says so. + let answer = server(layer(r#", "maxRecordCount": 2"#)); + let (joined, note) = join_pages(query, page(0, 2), 100, |u| { + if u.contains("resultOffset=2&") { Err("http status: 500".into()) } else { answer(&mut Vec::new(), u) } + }); + assert_eq!(ids(&joined), [0, 1]); + assert!(note.is_some_and(|n| n.contains("after feature 2 couldn't be read (http status: 500)"))); + // Only an ArcGIS query is asked again, only one that was cut off, and not one that asks + // for a window of its own. + let unasked = |_: &str| -> Result, String> { panic!("no page is asked for") }; + let elsewhere = "https://x.org/collections/a/items?f=json"; + assert_eq!(join_pages(elsewhere, page(0, 2), 100, unasked), (page(0, 2), None)); + assert_eq!(join_pages(query, page(0, 5), 100, unasked), (page(0, 5), None)); + let window = format!("{query}&resultRecordCount=2"); + assert_eq!(join_pages(&window, page(0, 2), 100, unasked), (page(0, 2), None)); + } + + /// The addresses an ArcGIS layer is paged with: its description's (with the token), each page's + /// (keeping the query's own order), and the order from the layer's object id field. + #[test] + fn arcgis_page_addresses() { + let q = "https://x.org/a/MapServer/3/query?where=1%3D1&f=geojson#top"; + assert!(is_arcgis_query(q) && !is_arcgis_query("https://x.org/a/MapServer/3") && !is_arcgis_query("https://x.org/query")); + assert_eq!(arcgis_layer_url(q), "https://x.org/a/MapServer/3?f=json"); + assert_eq!( + arcgis_page_url(q, 4, 2, Some("FID")), + "https://x.org/a/MapServer/3/query?where=1%3D1&f=geojson&resultOffset=4&resultRecordCount=2&orderByFields=FID#top" + ); + // An order the query already has is kept. + let ordered = "https://x.org/a/FeatureServer/0/query?orderByFields=NAME"; + assert_eq!(arcgis_page_url(ordered, 1, 1, Some("FID")), format!("{ordered}&resultOffset=1&resultRecordCount=1")); + // The object id field from the fields when `objectIdField` is missing; never a name that + // isn't plain. + let paging = |v: serde_json::Value| arcgis_paging(&v); + let pages = serde_json::json!({"supportsPagination": true}); + assert_eq!( + paging(serde_json::json!({"advancedQueryCapabilities": pages, "fields": [{"name": "FID", "type": "esriFieldTypeOID"}]})), + Some(ArcgisPaging { size: None, order: Some("FID".into()) }) + ); + assert_eq!( + paging(serde_json::json!({"advancedQueryCapabilities": pages, "maxRecordCount": 1000, "objectIdField": "id&f=html"})), + Some(ArcgisPaging { size: Some(1000), order: None }) + ); + assert_eq!(paging(serde_json::json!({"maxRecordCount": 1000})), None); + } + #[test] fn wms_layers_in_web_mercator() { let xml = r#" diff --git a/crates/engine/src/cmd/service.rs b/crates/engine/src/cmd/service.rs index 77254bf3d..976487667 100644 --- a/crates/engine/src/cmd/service.rs +++ b/crates/engine/src/cmd/service.rs @@ -196,30 +196,36 @@ pub fn features_answer(s: &mut Session, name: &str, bytes: Vec) -> Vec { rustgis_data::ogc::fix_axes(&bytes).unwrap_or(bytes) } -/// A service's features at `url`, with the pages its `next` links lead to: at most `max` bytes -/// and [`BLOCKING_TIMEOUT`](crate::render::rustgis_tiles::cache::BLOCKING_TIMEOUT) in all. A -/// `next` link is fetched only where its page may point. +/// A service's features at `url`, with the pages that follow (its `next` links, or an ArcGIS +/// layer's pages past its record limit, see [`read_pages`](rustgis_data::ogc::read_pages)): at +/// most `max` bytes and [`BLOCKING_TIMEOUT`](crate::render::rustgis_tiles::cache::BLOCKING_TIMEOUT) +/// in all. A page is fetched only where the page before it may point. #[cfg(not(target_arch = "wasm32"))] pub(crate) fn download_features(s: &mut Session, name: &str, url: &str, max: usize) -> Result, String> { use crate::render::rustgis_tiles::{address, cache}; let until = std::time::Instant::now() + cache::BLOCKING_TIMEOUT; let first = cache::download_blocking_limited(url, max).map_err(|e| format!("{url}: {e}"))?; let mut left = max.saturating_sub(first.len()); - let mut from = url.to_string(); - let (bytes, note) = rustgis_data::ogc::join_pages(url, first, rustgis_data::ogc::MAX_PAGED_FEATURES, |next| { - address::check(next, Some(&from))?; - let wait = until.saturating_duration_since(std::time::Instant::now()); - if wait.is_zero() { - return Err("the pages took too long".into()); - } - let page = cache::download_blocking_within(next, left, wait)?; - if page.len() > left { - return Err("the pages are larger than RustGIS downloads".into()); - } - left -= page.len(); - from = next.to_string(); - Ok(page) - }); + let mut step = rustgis_data::ogc::read_pages(url, first, rustgis_data::ogc::MAX_PAGED_FEATURES); + let (bytes, note) = loop { + let pages = match step { + rustgis_data::ogc::Step::Done(bytes, note) => break (bytes, note), + rustgis_data::ogc::Step::Fetch(pages) => pages, + }; + let answer = address::check(pages.next_url(), Some(pages.referrer())).and_then(|()| { + let wait = until.saturating_duration_since(std::time::Instant::now()); + if wait.is_zero() { + return Err("the pages took too long".to_string()); + } + let page = cache::download_blocking_within(pages.next_url(), left, wait)?; + if page.len() > left { + return Err("the pages are larger than RustGIS downloads".into()); + } + left -= page.len(); + Ok(page) + }); + step = (*pages).land(answer); + }; if let Some(note) = note { s.notify(format!("{name}: {note}"), false); } diff --git a/crates/engine/src/tests.rs b/crates/engine/src/tests.rs index c7a0fc8cc..c0a2bee4a 100644 --- a/crates/engine/src/tests.rs +++ b/crates/engine/src/tests.rs @@ -7613,6 +7613,62 @@ fn ogc_api_features_follow_next_pages() { assert!(s.notices.iter().any(|n| n.text.starts_with("Inward: ") && n.text.contains("never fetched"))); } +/// An ArcGIS layer past its record limit is read again in pages ordered by its object ids, from +/// the start (its first page, in the server's own order, may hold other features than the ordered +/// first: Hawaii's zoning did), when its description says it can be; a layer that can't keeps its +/// first page, and says so. +#[test] +fn arcgis_features_read_past_the_record_limit() { + use ogc_server::{Route, Server, arcgis_layer, arcgis_page}; + let q = "where=1%3D1&outFields=*&returnGeometry=true&outSR=4326&f=geojson"; + let parcels = "/arcgis/rest/services/P/MapServer/0"; + let page = |offset: u32| format!("{parcels}/query?{q}&resultOffset={offset}&resultRecordCount=2&orderByFields=OBJECTID"); + let server = Server::new(vec![ + // The first page in the server's own order: two of the five, not the ordered first two. + Route::new(&format!("{parcels}/query?{q}"), &arcgis_page(&[3, 0], true)), + Route::new(parcels, &arcgis_layer(2, true)), + Route::new(&page(0), &arcgis_page(&[0, 1], true)), + Route::new(&page(2), &arcgis_page(&[2, 3], true)), + Route::new(&page(4), &arcgis_page(&[4], false)), + Route::new("/arcgis/rest/services/P/MapServer/1/query", &arcgis_page(&[0, 1], true)), + Route::new("/arcgis/rest/services/P/MapServer/1", &arcgis_layer(2, false)), + ]); + let service = r#"{"currentVersion": 11.5, "maxRecordCount": 2, "layers": [{"id": 0, "name": "Parcels"}, {"id": 1, "name": "Zoning"}]}"#; + let url = format!("{}/arcgis/rest/services/P/MapServer", server.base); + let mut s = Session::new(Project::default()); + let r = s.run("service.add", &json!({"url": url, "layer": "Parcels", "text": service})).unwrap(); + assert_eq!(r["features"], 5); + let ids: Vec<_> = s.map().layer_by_name("Parcels").unwrap().dataset.features.iter().map(|f| f.attrs[0].as_f64()).collect(); + assert_eq!(ids, [Some(0.0), Some(1.0), Some(2.0), Some(3.0), Some(4.0)]); + assert!(!s.notices.iter().any(|n| n.text.contains("aren't read")), "{:?}", s.notices.iter().map(|n| &n.text).collect::>()); + let r = s.run("service.add", &json!({"url": url, "layer": "Zoning", "text": service})).unwrap(); + assert_eq!(r["features"], 2); + assert!(s.notices.iter().any(|n| n.text.starts_with("Zoning: ") && n.text.contains("record limit"))); + // The description once, then the pages from the start; a layer that can't page, its + // description alone. + let asked: Vec<_> = server + .requests() + .iter() + .map(|r| { + let offset = r.query.iter().find(|(k, _)| k == "resultOffset").map(|(_, v)| v.clone()); + (r.path.trim_start_matches("/arcgis/rest/services/P/MapServer").to_string(), offset) + }) + .collect(); + let at = |path: &str, offset: Option<&str>| (path.to_string(), offset.map(str::to_string)); + assert_eq!( + asked, + [ + at("/0/query", None), + at("/0", None), + at("/0/query", Some("0")), + at("/0/query", Some("2")), + at("/0/query", Some("4")), + at("/1/query", None), + at("/1", None) + ] + ); +} + /// A WCS coverage (from a local server) becomes a raster of the view, asked for in EPSG:4326 within /// its bounds; an error answer is refused. A catalogue's records come with the links RustGIS can /// follow, those pointing inward left out. diff --git a/crates/ui-egui/src/lib.rs b/crates/ui-egui/src/lib.rs index becdadbd5..ef36a824e 100644 --- a/crates/ui-egui/src/lib.rs +++ b/crates/ui-egui/src/lib.rs @@ -538,6 +538,8 @@ pub struct UiState { /// that can be read in place are added so, others downloaded (the `ui.addUrl` parameters, /// and the answer). pub cog_probes: Vec<(Value, Slot>)>, + /// Service layers whose features are being read page by page, after their first page landed. + pub service_pages: Vec, /// A link's data (`?data=`) has zoomed the map: the first of its files to land with something /// to show does, whichever it is. pub link_zoomed: bool, @@ -685,6 +687,7 @@ impl Default for UiState { project_download: None, sql: Default::default(), cog_probes: Vec::new(), + service_pages: Vec::new(), link_zoomed: false, service: None, service_browse: ServiceBrowse::default(), @@ -3107,32 +3110,11 @@ impl RustGisApp { for (id, url, mut p, r) in landed { let origin = connectiontree::take_origin_key(&mut p); match r { - Ok(bytes) => { - if !self.session.download_landed(id, bytes) { - self.session.notify( - i18n::fmt("{url}: downloaded after its project was closed, so it wasn't added", &[("url", &url)]), - false, - ); - continue; - } - p["id"] = json!(id); - p["url"] = json!(url); - let zoom = p.get("zoom").and_then(Value::as_bool) == Some(true); - if zoom && self.ui.link_zoomed { - p["zoom"] = json!(false); - } - let before = self.session.sources.generation; - match self.run("layer.addDownload", p) { - Ok(r) => { - connectiontree::name_recent(self, before, origin.as_deref()); - if r.get("zoomed").and_then(Value::as_bool) == Some(true) { - self.ui.link_zoomed = true; - } - self.map.invalidate(); - } - Err(e) => self.session.notify(if e.starts_with(&url) { e } else { format!("{url}: {e}") }, true), - } + // A service's features: the pages after the first are read, off the frame, first. + Ok(bytes) if p.get("service").and_then(Value::as_bool) == Some(true) => { + serviceui::read_pages(self, serviceui::Landed { id, url, params: p, origin }, bytes); } + Ok(bytes) => self.add_landed(serviceui::Landed { id, url, params: p, origin }, bytes), Err(e) => { self.session.download_failed(id); self.session.notify(if e.starts_with(&url) { e } else { format!("{url}: {e}") }, true); @@ -3141,6 +3123,33 @@ impl RustGisApp { } } + /// Adds a download that landed (`layer.addDownload`), named in Recent after the connection it + /// came from; one whose project was closed meanwhile is said and dropped. + pub(crate) fn add_landed(&mut self, landed: serviceui::Landed, bytes: Vec) { + let serviceui::Landed { id, url, params: mut p, origin } = landed; + if !self.session.download_landed(id, bytes) { + self.session.notify(i18n::fmt("{url}: downloaded after its project was closed, so it wasn't added", &[("url", &url)]), false); + return; + } + p["id"] = json!(id); + p["url"] = json!(url); + let zoom = p.get("zoom").and_then(Value::as_bool) == Some(true); + if zoom && self.ui.link_zoomed { + p["zoom"] = json!(false); + } + let before = self.session.sources.generation; + match self.run("layer.addDownload", p) { + Ok(r) => { + connectiontree::name_recent(self, before, origin.as_deref()); + if r.get("zoomed").and_then(Value::as_bool) == Some(true) { + self.ui.link_zoomed = true; + } + self.map.invalidate(); + } + Err(e) => self.session.notify(if e.starts_with(&url) { e } else { format!("{url}: {e}") }, true), + } + } + /// Tells how a service layer added outside the add-service dialog (or in it) went: in the /// dialog's own line, or as a notice for the Connections tree. fn report_step(&mut self, dialog: bool, r: Result) { @@ -3321,6 +3330,7 @@ impl RustGisApp { sqlpane::poll(self, ctx); self.poll_service(); serviceui::poll(self, ctx); + serviceui::poll_pages(self, ctx); // Sources the AI Assistant or a background add brought in (not through `run`) are kept // too: a generation compared each frame. So is a default raster style it set. sourcespane::save(self); diff --git a/crates/ui-egui/src/serviceui.rs b/crates/ui-egui/src/serviceui.rs index 5ed443f5a..e63b138a5 100644 --- a/crates/ui-egui/src/serviceui.rs +++ b/crates/ui-egui/src/serviceui.rs @@ -1,7 +1,8 @@ //! Background group import/refresh for both desktop and browser. Discovery and updates live in //! engine commands; this state machine only supplies their downloaded documents. -use crate::{RustGisApp, Slot, fetch_into, fetch_into_limited}; +use crate::{RustGisApp, Slot, fetch_into, fetch_into_limited, i18n}; +use rustgis_engine::data::ogc::{self, Pages, Step}; use serde_json::{Value, json}; use std::collections::VecDeque; @@ -249,3 +250,82 @@ pub fn poll(app: &mut RustGisApp, ctx: &egui::Context) { app.ui.service_import = Some(task); } } + +/// A download that landed, with what it is added with: its id, address and `layer.addDownload` +/// parameters, and the connection it came from (named in Recent). +pub(crate) struct Landed { + pub id: u64, + pub url: String, + pub params: Value, + pub origin: Option, +} + +/// A service layer's features read page by page off the frame, after its first page landed (#764): +/// each further answer [`Pages`] asks for is fetched in the background and handed to it, until +/// the collection is read; then it is added as the download it began as. +pub struct PageRead { + landed: Landed, + pages: Box, + /// The bytes the answers still to come may take (the download's budget, less those read). + left: usize, + slot: Slot, String>>, +} + +/// Starts reading a service's features past their first page (`first`, landed): added now when it +/// leads nowhere, else read on in the background ([`poll_pages`]). +pub(crate) fn read_pages(app: &mut RustGisApp, landed: Landed, first: Vec) { + let left = rustgis_engine::render::rustgis_tiles::cache::MAX_DOWNLOAD.saturating_sub(first.len()); + let step = ogc::read_pages(&landed.url, first, ogc::MAX_PAGED_FEATURES); + if matches!(step, Step::Fetch(_)) { + app.session.notify(i18n::fmt("Reading {url} page by page…", &[("url", &landed.url)]), false); + } + go_on(app, landed, left, step); +} + +/// Hands each page that landed to its read, and adds the layers whose features are all read. A +/// read whose project was closed meanwhile is dropped. +pub fn poll_pages(app: &mut RustGisApp, ctx: &egui::Context) { + if app.ui.service_pages.is_empty() { + return; + } + for read in std::mem::take(&mut app.ui.service_pages) { + if !app.session.download_is_pending(read.landed.id) { + let url = &read.landed.url; + app.session.notify(i18n::fmt("{url}: downloaded after its project was closed, so it wasn't added", &[("url", url)]), false); + continue; + } + let answer = read.slot.lock().unwrap_or_else(|e| e.into_inner()).take(); + match answer { + Some(answer) => { + let left = read.left.saturating_sub(answer.as_ref().map_or(0, Vec::len)); + let step = (*read.pages).land(answer); + go_on(app, read.landed, left, step); + } + None => app.ui.service_pages.push(read), + } + } + if !app.ui.service_pages.is_empty() { + ctx.request_repaint_after(std::time::Duration::from_millis(30)); + } +} + +/// The next answer fetched, or the collection added (with a note when some features were left +/// unread). +fn go_on(app: &mut RustGisApp, landed: Landed, left: usize, step: Step) { + match step { + Step::Fetch(pages) => { + let slot = match rustgis_engine::render::rustgis_tiles::address::check(pages.next_url(), Some(pages.referrer())) { + Ok(()) => fetch_into_limited(pages.next_url(), left), + Err(e) => Slot::new(std::sync::Mutex::new(Some(Err(e)))), + }; + app.ui.service_pages.push(PageRead { landed, pages, left, slot }); + } + Step::Done(bytes, note) => { + if let Some(note) = note { + let name = landed.params.get("name").and_then(Value::as_str).filter(|n| !n.is_empty()).unwrap_or(&landed.url); + app.session.notify(format!("{name}: {note}"), false); + } + app.add_landed(landed, bytes); + } + } +} diff --git a/crates/ui-egui/tests/smoke.rs b/crates/ui-egui/tests/smoke.rs index aed864063..022f12757 100644 --- a/crates/ui-egui/tests/smoke.rs +++ b/crates/ui-egui/tests/smoke.rs @@ -1402,6 +1402,55 @@ fn data_from_a_web_address_lands_later() { assert!(h.state().ui.downloads.is_empty()); } +/// A service layer added from the app (Add From Service, the Connections tree) is read past its +/// first page off the frame (#764): an ArcGIS layer again from the start in pages ordered by its +/// object ids, an OGC API collection by its `next` links; the layer is added once the pages are +/// in, with no note. +#[test] +fn service_features_are_read_page_by_page_off_the_frame() { + use ogc_server::{Route, Server, arcgis_layer, arcgis_page}; + let q = "where=1%3D1&outFields=*&f=geojson"; + let layer = "/arcgis/rest/services/P/MapServer/0"; + let page = |offset: u32| format!("{layer}/query?{q}&resultOffset={offset}&resultRecordCount=2&orderByFields=OBJECTID"); + let collection = |ids: &[u32], next: Option<&str>| { + let mut c: serde_json::Value = serde_json::from_str(&arcgis_page(ids, false)).unwrap(); + if let Some(href) = next { + c["links"] = json!([{"rel": "next", "href": href}]); + } + c.to_string() + }; + let server = Server::new(vec![ + Route::new(&format!("{layer}/query?{q}"), &arcgis_page(&[3, 0], true)), + Route::new(layer, &arcgis_layer(2, true)), + Route::new(&page(0), &arcgis_page(&[0, 1], true)), + Route::new(&page(2), &arcgis_page(&[2, 3], true)), + Route::new(&page(4), &arcgis_page(&[4], false)), + Route::new("/collections/wells/items?page=2", &collection(&[2], None)), + Route::new("/collections/wells/items", &collection(&[0, 1], Some("items?page=2"))), + ]); + let mut h = harness(); + h.step(); + let base = server.base.clone(); + h.state_mut().run("ui.addUrl", json!({"url": format!("{base}{layer}/query?{q}"), "name": "Parcels", "service": true})).unwrap(); + h.state_mut().run("ui.addUrl", json!({"url": format!("{base}/collections/wells/items"), "name": "Wells", "service": true})).unwrap(); + let ids = |app: &RustGisApp, name: &str| { + app.session.map().layer_by_name(name).map(|l| l.dataset.features.iter().map(|f| f.attrs[0].as_f64()).collect::>()) + }; + wait_for_ogc(&mut h, |app| ids(app, "Parcels").is_some() && ids(app, "Wells").is_some()); + let notes: Vec<_> = h.state().session.notices.iter().map(|n| n.text.clone()).collect(); + let n = |i: &[u32]| Some(i.iter().map(|&i| Some(f64::from(i))).collect::>()); + assert_eq!((ids(h.state(), "Parcels"), ids(h.state(), "Wells")), (n(&[0, 1, 2, 3, 4]), n(&[0, 1, 2])), "{notes:?}"); + assert!(!notes.iter().any(|n| n.contains("aren't read")), "{notes:?}"); + assert!(h.state().ui.service_pages.is_empty() && h.state().ui.downloads.is_empty()); + let offsets: Vec<_> = server + .requests() + .into_iter() + .filter(|r| r.path.ends_with("/0/query")) + .map(|r| r.query.into_iter().find(|(k, _)| k == "resultOffset").map(|(_, v)| v)) + .collect(); + assert_eq!(offsets, [None, Some("0".into()), Some("2".into()), Some("4".into())]); +} + /// A project by its web address (a `?project=` link) downloads off the frame and replaces the /// open project when it lands, its layers and view with it; a failed download is said and keeps /// the open project. diff --git a/docs/user-guide/data-formats.md b/docs/user-guide/data-formats.md index 5d18e1b54..ef1ed3168 100644 --- a/docs/user-guide/data-formats.md +++ b/docs/user-guide/data-formats.md @@ -174,10 +174,10 @@ Scripts use `service.import {"url":"…","name":"My OGC Layer"}` and | WMS 1.1.1 / 1.3.0 | Named layers offered in Web Mercator | A tile layer of GetMap images (PNG when the service offers it, else JPEG) | | WMTS | Layers whose tile matrix set is Web Mercator's pyramid | A tile layer from the tile template (or KVP GetTile) | | WFS 1.1 / 2.0 | Feature types | Their features downloaded as GeoJSON, in the output format the service names (up to 50,000) | -| OGC API Features | Collections | Their items downloaded as GeoJSON, following the service's `next` pages up to 100,000 features or a minute of downloading (the web app reads the first page) | +| OGC API Features | Collections | Their items downloaded as GeoJSON, following the service's `next` pages up to 100,000 features (a command such as `service.add` waits a minute at most) | | OGC API Tiles | Collections' map and vector tiles; a landing page's dataset map tiles | A tile layer in WebMercatorQuad, from the tile template the tileset advertises (its `item` link; `…/tiles/WebMercatorQuad/{z}/{y}/{x}`, the usual path, when that can't be read), keeping the tiles link's query (a token), map tiles in the server's image format, vector tiles drawn layer by layer in colours of their own | | WCS 1.0 / 2.0 (a 1.1 service is asked in 1.0) | Coverages, with their bounds | A raster: GetCoverage of the map's view within the coverage's bounds (or its whole extent: `"bbox": "full"`), as a GeoTIFF in EPSG:4326, at most 2048 pixels on a side | -| ArcGIS REST | FeatureServer and MapServer layers; a MapServer's map; an ImageServer; a VectorTileServer | Layers queried as GeoJSON (up to the server's record limit); a map as its tile cache when it is Web Mercator's pyramid, else exported images; an ImageServer as exported images; a VectorTileServer as its MapLibre style | +| ArcGIS REST | FeatureServer and MapServer layers; a MapServer's map; an ImageServer; a VectorTileServer | Layers queried as GeoJSON, page by page past the server's record limit (`resultOffset`) up to 100,000 features (a command such as `service.add` waits a minute at most), where the layer supports paging (ArcGIS Server 10.3 and later; otherwise up to the record limit); a map as its tile cache when it is Web Mercator's pyramid, else exported images; an ImageServer as exported images; a VectorTileServer as its MapLibre style | The type is guessed from the address (`?SERVICE=`, `/wmts`, `/wfs`, `/wcs`, `/collections`, `/rest/services`…) or chosen (select **OGC API Features and Tiles** when a landing page's address doesn't identify diff --git a/tests/support/ogc.rs b/tests/support/ogc.rs index 17027b8bb..b6d5cf627 100644 --- a/tests/support/ogc.rs +++ b/tests/support/ogc.rs @@ -58,3 +58,26 @@ pub fn fixture_lines() -> &'static str { LazyLock::new(|| feature("route-1", "Test Route", json!({"type":"LineString", "coordinates":[[10.0,50.0],[11.0,51.0]]}))); &TEXT } + +/// An ArcGIS layer query's GeoJSON answer: points whose `OBJECTID` are `ids`, cut off at the +/// server's record limit (`exceededTransferLimit`, among its `properties` as ArcGIS puts it) when +/// `more`. +pub fn arcgis_page(ids: &[u32], more: bool) -> String { + let features: Vec = ids + .iter() + .map(|&i| json!({"type":"Feature", "id":i, "properties":{"OBJECTID":i}, "geometry":{"type":"Point", "coordinates":[i, 1]}})) + .collect(); + let mut page = json!({"type":"FeatureCollection", "features":features}); + if more { + page["properties"] = json!({"exceededTransferLimit": true}); + } + page.to_string() +} + +/// An ArcGIS layer's description (`…/MapServer/0?f=json`): its record limit and whether it can be +/// read in pages. The object id field is named only among its fields, as a MapServer layer's is. +pub fn arcgis_layer(max: u32, pages: bool) -> String { + json!({"currentVersion":11.5, "maxRecordCount":max, "advancedQueryCapabilities":{"supportsPagination":pages}, + "fields":[{"name":"OBJECTID", "type":"esriFieldTypeOID"}]}) + .to_string() +} diff --git a/tests/support/ogc_server.rs b/tests/support/ogc_server.rs index 6da043872..ef7c06a9c 100644 --- a/tests/support/ogc_server.rs +++ b/tests/support/ogc_server.rs @@ -19,6 +19,8 @@ pub const LINE_ITEMS: &str = "/ogc/user/test-token/collections/routes/items"; #[derive(Clone)] pub struct Route { + /// The path answered, or a path and query (`/query?resultOffset=2`) to answer that request + /// alone (a page of a layer that comes in pages). pub path: String, pub body: String, pub status: u16, @@ -121,7 +123,10 @@ impl Server { }) .collect(), }); - let route = serving.lock().unwrap().iter().find(|r| r.path == path).cloned(); + let route = { + let routes = serving.lock().unwrap(); + routes.iter().find(|r| r.path == target).or_else(|| routes.iter().find(|r| r.path == path)).cloned() + }; let (status, body, extra) = match route { Some(r) => { let until = Instant::now() + r.delay;