flowctl: handle suspended journals when reading collections

Updates `collections read` to handle journals that are fully suspended.
Previously this would result in a rather confusing error message, and the
process would exit with an error even if the `--follow` flag was provided. The
new behavior is to just squelch the error in the common case that
`--follow=false`, so it gets treated like any other journal that doesn't have
any content. In case `--follow` is provided, we use a sizable deley and just
keep retrying. This also applies to `flowctl logs` and `flowctl raw stats`,
which use the same code for reading collections.

crates/flowctl/Cargo.toml

@@ -51,6 +51,7 @@ pbjson-types = { workspace = true }
 portpicker = { workspace = true }
 postgrest = { workspace = true }
 prost = { workspace = true }
+rand = { workspace = true }
 reqwest = { workspace = true }
 rusqlite = { workspace = true }
 rustls = { workspace = true }

crates/flowctl/src/collection/read/mod.rs

@@ -3,6 +3,7 @@ use anyhow::Context;
 use futures::StreamExt;
 use gazette::journal::ReadJsonLine;
 use proto_gazette::broker;
+use rand::Rng;
 use std::io::Write;
 use time::OffsetDateTime;
 
@@ -135,7 +136,19 @@ pub async fn read_collection_journal(
                 v.push(b'\n');
                 () = stdout.write_all(&v)?;
             }
-            Err(gazette::Error::BrokerStatus(broker::Status::OffsetNotYetAvailable)) => {
+            Err(gazette::Error::BrokerStatus(broker::Status::Suspended)) if bounds.follow => {
+                // The journal is fully suspended, so we use a pretty long delay
+                // here because it's unlikely to be resumed quickly and also
+                // unlikely that anyone will care about the little bit of extra
+                // latency in this case.
+                let delay_secs = rand::thread_rng().gen_range(30..=60);
+                tracing::info!(delay_secs, "journal suspended, will retry");
+                tokio::time::sleep(std::time::Duration::from_secs(delay_secs)).await;
+            }
+            Err(gazette::Error::BrokerStatus(
+                status @ broker::Status::OffsetNotYetAvailable | status @ broker::Status::Suspended,
+            )) => {
+                tracing::debug!(?status, "stopping read at end of journal content");
                 break; // Graceful EOF of non-blocking read.
             }
             Err(err) if err.is_transient() => {