diff --git a/src/poll.rs b/src/poll.rs index c4fc76f..67cc958 100644 --- a/src/poll.rs +++ b/src/poll.rs @@ -139,27 +139,42 @@ impl PageBoundaryState { /// /// Extracts the final op so it can be used to fetch the following page pub async fn get_page(url: Url) -> Result<(ExportPage, Option), GetPageError> { + use futures::TryStreamExt; + use tokio::io::{AsyncBufReadExt, BufReader}; + use tokio_util::compat::FuturesAsyncReadCompatExt; + log::trace!("Getting page: {url}"); - let ops: Vec = CLIENT - .get(url) - .send() - .await? - .error_for_status()? - .text() - .await? - .trim() - .split('\n') - .filter_map(|s| { - serde_json::from_str::(s) - .inspect_err(|e| { - if !s.is_empty() { - log::warn!("failed to parse op: {e} ({s})") - } - }) - .ok() - }) - .collect(); + let res = CLIENT.get(url).send().await?.error_for_status()?; + let stream = Box::pin( + res.bytes_stream() + .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e)) + .into_async_read() + .compat(), + ); + + let mut lines = BufReader::new(stream).lines(); + let mut ops = Vec::new(); + + loop { + match lines.next_line().await { + Ok(Some(line)) => { + let line = line.trim(); + if line.is_empty() { + continue; + } + match serde_json::from_str::(line) { + Ok(op) => ops.push(op), + Err(e) => log::warn!("failed to parse op: {e} ({line})"), + } + } + Ok(None) => break, + Err(e) => { + log::warn!("transport error mid-page: {}; returning partial page", e); + break; + } + } + } let last_op = ops.last().map(Into::into); @@ -214,7 +229,14 @@ pub async fn poll_upstream( .append_pair("after", &pl.created_at.to_rfc3339()); }; - let (mut page, next_last) = get_page(url).await?; + let (mut page, next_last) = match get_page(url).await { + Ok(res) => res, + Err(e) => { + log::warn!("error polling upstream: {e}"); + continue; + } + }; + if let Some(ref mut state) = boundary_state { state.apply_to_next(&mut page); } else { diff --git a/src/weekly.rs b/src/weekly.rs index a72e316..a7b5b3b 100644 --- a/src/weekly.rs +++ b/src/weekly.rs @@ -197,30 +197,62 @@ pub async fn week_to_pages( dest: mpsc::Sender, ) -> anyhow::Result<()> { use futures::TryStreamExt; - let reader = source - .reader_for(week) - .await - .inspect_err(|e| log::error!("week_to_pages reader failed: {e}"))?; - let decoder = GzipDecoder::new(BufReader::new(reader)); - let mut chunks = pin!(LinesStream::new(BufReader::new(decoder).lines()).try_chunks(10000)); - - while let Some(chunk) = chunks - .try_next() - .await - .inspect_err(|e| log::error!("failed to get next chunk: {e}"))? - { - let ops: Vec = chunk - .into_iter() - .filter_map(|s| { - serde_json::from_str::(&s) - .inspect_err(|e| log::warn!("failed to parse op: {e} ({s})")) - .ok() - }) - .collect(); - let page = ExportPage { ops }; - dest.send(page) - .await - .inspect_err(|e| log::error!("failed to send page: {e}"))?; + let mut retry_backoff = std::time::Duration::from_secs(2); + + loop { + let reader = match source.reader_for(week).await { + Ok(r) => r, + Err(e) => { + log::warn!( + "week_to_pages reader_for failed {e}, retrying in {}s", + retry_backoff.as_secs() + ); + tokio::time::sleep(retry_backoff).await; + retry_backoff = (retry_backoff * 2).min(std::time::Duration::from_secs(300)); + continue; + } + }; + + let decoder = GzipDecoder::new(BufReader::new(reader)); + let mut chunks = pin!(LinesStream::new(BufReader::new(decoder).lines()).try_chunks(10000)); + let mut success = true; + + while let Some(chunk) = match chunks.as_mut().try_next().await { + Ok(Some(c)) => Some(c), + Ok(None) => None, + Err(e) => { + log::warn!( + "failed to get next chunk: {e}, retrying week in {}s", + retry_backoff.as_secs() + ); + tokio::time::sleep(retry_backoff).await; + retry_backoff = (retry_backoff * 2).min(std::time::Duration::from_secs(300)); + success = false; + None + } + } { + let ops: Vec = chunk + .into_iter() + .filter_map(|s| { + serde_json::from_str::(&s) + .inspect_err(|e| log::warn!("failed to parse op: {e} ({s})")) + .ok() + }) + .collect(); + + if ops.is_empty() { + continue; + } + + let page = ExportPage { ops }; + if let Err(e) = dest.send(page).await { + log::error!("failed to send page (receiver closed): {e}"); + return Err(e.into()); + } + } + + if success { + return Ok(()); + } } - Ok(()) }