feat(data): read ArcGIS layers past their record limit, in the app and the browser

crates/engine/src/cmd/service.rs

@@ -196,30 +196,36 @@ pub fn features_answer(s: &mut Session, name: &str, bytes: Vec<u8>) -> Vec<u8> {
     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<Vec<u8>, 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);
     }

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::<Vec<_>>());
+    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.

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<Result<bool, String>>)>,
+    /// Service layers whose features are being read page by page, after their first page landed.
+    pub service_pages: Vec<serviceui::PageRead>,
     /// 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<u8>) {
+        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<String, String>) {
@@ -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);

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<String>,
+}
+
+/// 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<Pages>,
+    /// The bytes the answers still to come may take (the download's budget, less those read).
+    left: usize,
+    slot: Slot<Result<Vec<u8>, 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<u8>) {
+    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);
+        }
+    }
+}

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::<Vec<_>>())
+    };
+    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::<Vec<_>>());
+    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.