feat(http): POST with a JSON body, and paging and cursors in the body

crates/sqlload/src/http_source/_helpers/fetch.rs

@@ -1,8 +1,7 @@
-//! One GET with the shared rate limit, retries with backoff, `Retry-After`, and `on_status`.
+//! One request with the shared rate limit, retries with backoff, `Retry-After`, and `on_status`.
 
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
-use reqwest::Url;
 use reqwest::header::{LINK, RETRY_AFTER};
 
 use crate::http::constants::{FIRST_SERVER_ERROR, TOO_MANY_REQUESTS, UNAUTHORIZED};
@@ -12,11 +11,11 @@ use crate::http_source::constants::{
     ERROR_EXCERPT_CHARS, FIRST_RETRY_DELAY, MAX_REDIRECTS, MAX_RETRY_AFTER, MAX_RETRY_DELAY,
     SKIPPED_REFUSAL_RENEWAL_AGE,
 };
-use crate::http_source::models::{Attempt, Calls, Credential, Fetched, StatusAction};
+use crate::http_source::models::{Attempt, Calls, Credential, Fetched, PageCall, StatusAction};
 use crate::python_sources::errors::ReadError;
 
-/// The body at `url`, or `None` when `on_status` skips its status; transport errors, 429 and 5xx are retried up to `calls.retries` times, and are still worth retrying later once those run out.
-pub(crate) async fn fetch(calls: &Calls, url: &Url) -> Result<Option<Fetched>, ReadError> {
+/// The response to `call`, its JSON body resent on every retry, or `None` when `on_status` skips its status; transport errors, 429 and 5xx are retried up to `calls.retries` times, and are still worth retrying later once those run out.
+pub(crate) async fn fetch(calls: &Calls, call: &PageCall) -> Result<Option<Fetched>, ReadError> {
     let policy = RetryPolicy {
         first_delay: FIRST_RETRY_DELAY,
         max_delay: MAX_RETRY_DELAY,
@@ -37,7 +36,7 @@ pub(crate) async fn fetch(calls: &Calls, url: &Url) -> Result<Option<Fetched>, R
             ),
             _ => None,
         };
-        match once(calls, url, access.as_deref(), !renewed).await? {
+        match once(calls, call, access.as_deref(), !renewed).await? {
             Attempt::Done(fetched) => return Ok(fetched),
             Attempt::Refused => {
                 refused = access;
@@ -58,18 +57,27 @@ pub(crate) async fn fetch(calls: &Calls, url: &Url) -> Result<Option<Fetched>, R
     }
 }
 
-/// One try at `url`, with the OAuth2 `access` token when the connection uses one; a 401 from the connection's origin comes back as `Refused` while `may_renew`, unless `on_status` skips 401 and the token is younger than `SKIPPED_REFUSAL_RENEWAL_AGE`.
+/// One try at `call`, with the OAuth2 `access` token when the connection uses one; a 401 from the connection's origin comes back as `Refused` while `may_renew`, unless `on_status` skips 401 and the token is younger than `SKIPPED_REFUSAL_RENEWAL_AGE`.
 async fn once(
     calls: &Calls,
-    url: &Url,
+    call: &PageCall,
     access: Option<&str>,
     may_renew: bool,
 ) -> Result<Attempt, String> {
     let endpoint = &calls.endpoint;
     let secrets = endpoint.current_secrets();
-    let shown = redacted(&format!("GET {url}"), &secrets);
+    let url = &call.url;
+    let shown = redacted(&format!("{} {url}", call.method.as_str()), &secrets);
     let own = same_origin(&endpoint.base_url, url);
-    let mut request = calls.http.get(url.clone());
+    if !own && call.body.is_some() {
+        return Err(format!(
+            "{shown}: the request is on another origin than the connection's base_url, and the job's json_body never leaves the connection's origin"
+        ));
+    }
+    let mut request = match &call.body {
+        Some(_) => calls.http.post(url.clone()),
+        None => calls.http.get(url.clone()),
+    };
     for (name, value) in endpoint.headers.iter().filter(|_| own) {
         request = request.header(name, value);
     }
@@ -79,6 +87,9 @@ async fn once(
         (Credential::OAuth2(_), true) => request.bearer_auth(access.unwrap_or_default()),
         _ => request,
     };
+    if let Some(body) = &call.body {
+        request = request.json(body);
+    }
     let response = match request.send().await {
         Ok(response) => response,
         Err(error) if error.is_builder() => {
@@ -138,7 +149,7 @@ async fn once(
         _ if skipped => Attempt::Done(None),
         300..=399 => {
             return Err(format!(
-                "{shown}: HTTP {status}: the redirect was not followed: it leads to another origin, has no Location, or goes past {MAX_REDIRECTS} hops; credentials never leave the connection's origin, so use the final URL as base_url"
+                "{shown}: HTTP {status}: the redirect was not followed: it leads to another origin, has no Location, goes past {MAX_REDIRECTS} hops, or is a 301, 302 or 303 after a POST, which would drop its body; credentials never leave the connection's origin, so use the final URL as base_url"
             ));
         }
         TOO_MANY_REQUESTS | FIRST_SERVER_ERROR.. => Attempt::Retry {

crates/sqlload/src/http_source/_helpers/job.rs

@@ -10,7 +10,7 @@ use tokio::task::JoinSet;
 
 use crate::http_source::_helpers::endpoint::{endpoint, interval, shared_limiter};
 use crate::http_source::_helpers::oauth::token_url;
-use crate::http_source::_helpers::pages::paginate;
+use crate::http_source::_helpers::pages::{paginate, set_at};
 use crate::http_source::_helpers::selector::{selected_rows, selector};
 use crate::http_source::_helpers::urls::{filled_path, placeholders};
 use crate::python_sources::errors::ReadError;
@@ -20,10 +20,12 @@ use crate::row_batches::models::{Projection, ProjectionRules};
 use std::time::Duration;
 
 use crate::http::main::install_tls_provider::install_tls_provider;
-use crate::http_source::constants::{JOB_THREADS, MAX_REDIRECTS, PAGES_AHEAD};
+use crate::http_source::constants::{
+    BODY_DROPPING_REDIRECTS, JOB_THREADS, MAX_REDIRECTS, PAGES_AHEAD,
+};
 use crate::http_source::models::{
-    Auth, Calls, FanOut, FieldSpec, HttpJob, HttpPage, KeyFeed, Over, Paginator, Request, Selector,
-    StatusAction,
+    Auth, Calls, FanOut, FieldSpec, HttpJob, HttpPage, KeyFeed, Location, Method, Over, Paginator,
+    Request, Selector, StatusAction,
 };
 
 /// Every problem the job's definition has, found before any request, including projected rows that break the table's `rules`.
@@ -84,6 +86,7 @@ fn request_problems(request: &Request) -> Vec<String> {
     let mut problems: Vec<String> = Vec::new();
     problems.extend(selector(request.select.as_deref()).err());
     problems.extend(paginator_problem(&request.paginator));
+    problems.extend(body_problems(request));
     problems.extend(match &request.client.auth {
         Auth::Oauth2ClientCredentials(spec) | Auth::Oauth2RefreshToken(spec) => {
             token_url(spec).err()
@@ -135,6 +138,64 @@ fn paginator_problem(paginator: &Paginator) -> Option<String> {
     }
 }
 
+/// Why the job's body settings cannot work: a body or body paging on a `GET`, a body path that is not a plain path, or body paging into a body that is not an object.
+fn body_problems(request: &Request) -> Vec<String> {
+    let (paths, location) = match &request.paginator {
+        Paginator::Offset {
+            param,
+            limit_param,
+            location,
+            ..
+        } => (vec![param, limit_param], *location),
+        Paginator::PageNumber {
+            param, location, ..
+        }
+        | Paginator::Cursor {
+            param, location, ..
+        } => (vec![param], *location),
+        Paginator::Single | Paginator::NextLink { .. } | Paginator::LinkHeader => {
+            (Vec::new(), Location::Query)
+        }
+    };
+    let mut problems: Vec<String> = Vec::new();
+    if request.method == Method::Get && request.json_body.is_some() {
+        problems.push("json_body is sent only with method=\"POST\"".to_owned());
+    }
+    if location == Location::Query {
+        return problems;
+    }
+    if request.method == Method::Get {
+        problems.push("a paginator with location=\"body\" needs method=\"POST\"".to_owned());
+    }
+    if request
+        .json_body
+        .as_ref()
+        .is_some_and(|body| !body.is_object())
+    {
+        problems.push(
+            "a paginator with location=\"body\" sets fields in json_body, so json_body must be an object".to_owned(),
+        );
+    }
+    problems.extend(paths.iter().filter_map(|path| match selector(Some(path)) {
+        Ok(parsed) if parsed.field.is_none() && !parsed.rows.is_empty() => None,
+        Ok(_) | Err(_) => Some(format!(
+            "body path {path:?} is not a path such as \"$.page.cursor\""
+        )),
+    }));
+    if problems.is_empty() {
+        let mut body = request
+            .json_body
+            .clone()
+            .unwrap_or_else(|| Value::Object(Map::new()));
+        problems.extend(
+            paths
+                .iter()
+                .filter_map(|path| set_at(&mut body, path, Value::Null).err()),
+        );
+    }
+    problems
+}
+
 /// Why a header cannot be sent, if it cannot.
 fn header_problem(name: &str, value: &str) -> Option<String> {
     match (
@@ -248,7 +309,14 @@ async fn run_job(
     connections: Map<String, Value>,
     pages: Sender<HttpPage>,
 ) -> Result<(), ReadError> {
-    let http = job_client()?;
+    let posts = match &job {
+        HttpJob::Paginate(request) => request.method == Method::Post,
+        HttpJob::FanOut(fan_out) => {
+            fan_out.request.method == Method::Post
+                || matches!(&fan_out.over, Over::Paginate(over) if over.method == Method::Post)
+        }
+    };
+    let http = job_client(posts)?;
     match job {
         HttpJob::Paginate(request) => {
             let calls = request_calls(&request, &connections, &http)?;
@@ -258,11 +326,16 @@ async fn run_job(
     }
 }
 
-/// A pooled client that follows redirects only within the first request's origin, so connection headers and credentials never reach another host.
-fn job_client() -> Result<reqwest::Client, String> {
+/// A pooled client that follows redirects only within the first request's origin, so connection headers and credentials never reach another host; when the job `posts`, it also stops at 301, 302 and 303, which would resend the request as a `GET` without its body.
+fn job_client(posts: bool) -> Result<reqwest::Client, String> {
     install_tls_provider();
     reqwest::Client::builder()
-        .redirect(reqwest::redirect::Policy::custom(same_origin_redirect))
+        .redirect(reqwest::redirect::Policy::custom(move |attempt| {
+            match posts && BODY_DROPPING_REDIRECTS.contains(&attempt.status().as_u16()) {
+                true => attempt.stop(),
+                false => same_origin_redirect(attempt),
+            }
+        }))
         .pool_idle_timeout(Duration::from_secs(90))
         .connect_timeout(Duration::from_secs(15))
         .timeout(Duration::from_secs(300))