fix(tams): stop retrying non-retryable uploads, clear stale segments, guard overflow

Three fixes from the PR review:

- retry(): a gateway/storage HTTP 4xx (except 408/429) can never succeed on
  retry, yet the uploader slept through the full backoff and re-attempted 3x —
  producing repeated identical errors (e.g. the 413 Payload Too Large we hit).
  Errors now carry the status via a typed HttpStatusError and retry() gives up
  immediately on non-retryable ones.
- Clear leftover seg_* files from a previous run at setup. Permanently-failed
  uploads are kept on disk but nothing reads them back, and there is no teardown
  hook, so they accumulated across restarts. Cleaning at setup (before the new
  run writes anything) bounds disk use without touching keep-on-failure.
- max-size-time now uses saturating_mul, matching tail_segment_ns, so a large
  operator-supplied segment_duration_secs can't overflow u64.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: srperens

backend/src/blocks/builtin/tams_output.rs

@@ -458,7 +458,7 @@ fn build_flow_chain(
         .build()
         .map_err(|e| BlockBuildError::ElementCreation(format!("splitmuxsink: {}", e)))?;
     splitmuxsink.set_property("muxer", &mux);
-    splitmuxsink.set_property("max-size-time", segment_secs * 1_000_000_000);
+    splitmuxsink.set_property("max-size-time", segment_secs.saturating_mul(1_000_000_000));
     // Fallback location template; the format-location-full signal overrides it.
     let fallback = temp_dir.join(format!("seg_%05d.{}", container.file_ext()));
     splitmuxsink.set_property("location", fallback.to_string_lossy().as_ref());
@@ -641,6 +641,30 @@ fn build_flow_chain(
     let file_ext = container.file_ext().to_string();
     let tail_segment_ns = segment_secs.saturating_mul(1_000_000_000);
     ctx.register_element_setup(Box::new(move |flow_id, events| {
+        // Clear segment files left by a previous run of this instance. Failed
+        // uploads are kept on disk, but nothing ever reads them back, so without
+        // this they accumulate across restarts (the temp dir is per-instance and
+        // there is no teardown hook). Safe here: the new run has not written yet.
+        if let Ok(entries) = std::fs::read_dir(&temp_dir_for_setup) {
+            let mut removed = 0u32;
+            for entry in entries.flatten() {
+                let path = entry.path();
+                let is_segment = path
+                    .file_name()
+                    .and_then(|n| n.to_str())
+                    .is_some_and(|n| n.starts_with("seg_"));
+                if is_segment && std::fs::remove_file(&path).is_ok() {
+                    removed += 1;
+                }
+            }
+            if removed > 0 {
+                info!(
+                    "TAMS {}: cleared {} leftover segment(s) from a previous run",
+                    block_id, removed
+                );
+            }
+        }
+
         // OSC auth keys its PAT by the flow id (tenant isolation on a shared
         // instance), which is only known here — so finalize the gateway client now.
         let credential_key = flow_id.to_string();

backend/src/tams/client.rs

@@ -22,6 +22,35 @@ use std::time::Duration;
 // frontend can share them. Re-exported here for ergonomic use within the backend.
 pub use strom_types::tams::{format_timerange, FORMAT_AUDIO, FORMAT_VIDEO};
 
+/// A non-success HTTP response from the gateway or presigned storage, carrying
+/// the status code so the uploader can decide whether retrying could ever help.
+#[derive(Debug)]
+pub struct HttpStatusError {
+    pub status: reqwest::StatusCode,
+    /// Human context, e.g. `presigned PUT` or `POST http://.../flows/x/segments`.
+    pub context: String,
+    pub body: String,
+}
+
+impl std::fmt::Display for HttpStatusError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{} -> {}: {}", self.context, self.status, self.body)
+    }
+}
+
+impl std::error::Error for HttpStatusError {}
+
+impl HttpStatusError {
+    /// A 4xx (other than 408 Request Timeout and 429 Too Many Requests) means the
+    /// request itself must change before it can succeed — retrying the identical
+    /// bytes/headers will fail the same way (e.g. 413 Payload Too Large, 401
+    /// Unauthorized). Network errors and 5xx/408/429 are worth retrying.
+    pub fn is_retryable(&self) -> bool {
+        let code = self.status.as_u16();
+        !((400..500).contains(&code) && code != 408 && code != 429)
+    }
+}
+
 /// Metadata describing a flow to create on the gateway.
 #[derive(Debug, Clone)]
 pub struct FlowSpec {
@@ -129,8 +158,13 @@ impl TamsClient {
             .with_context(|| format!("PUT {}", url))?;
         if !resp.status().is_success() {
             let status = resp.status();
-            let text = resp.text().await.unwrap_or_default();
-            return Err(anyhow!("PUT {} -> {}: {}", url, status, text));
+            let body = resp.text().await.unwrap_or_default();
+            return Err(HttpStatusError {
+                status,
+                context: format!("PUT {}", url),
+                body,
+            }
+            .into());
         }
         Ok(())
     }
@@ -153,8 +187,13 @@ impl TamsClient {
             .with_context(|| format!("POST {}", url))?;
         if !resp.status().is_success() {
             let status = resp.status();
-            let text = resp.text().await.unwrap_or_default();
-            return Err(anyhow!("POST {} -> {}: {}", url, status, text));
+            let body = resp.text().await.unwrap_or_default();
+            return Err(HttpStatusError {
+                status,
+                context: format!("POST {}", url),
+                body,
+            }
+            .into());
         }
         let parsed: StorageResponse = resp.json().await.context("parsing storage response")?;
         let obj = parsed
@@ -187,8 +226,13 @@ impl TamsClient {
             .context("PUT presigned S3 url")?;
         if !resp.status().is_success() {
             let status = resp.status();
-            let text = resp.text().await.unwrap_or_default();
-            return Err(anyhow!("presigned PUT -> {}: {}", status, text));
+            let body = resp.text().await.unwrap_or_default();
+            return Err(HttpStatusError {
+                status,
+                context: "presigned PUT".to_string(),
+                body,
+            }
+            .into());
         }
         Ok(())
     }
@@ -217,8 +261,13 @@ impl TamsClient {
             return Err(anyhow!("segment registration partially failed: {}", text));
         }
         if !status.is_success() {
-            let text = resp.text().await.unwrap_or_default();
-            return Err(anyhow!("POST {} -> {}: {}", url, status, text));
+            let body = resp.text().await.unwrap_or_default();
+            return Err(HttpStatusError {
+                status,
+                context: format!("POST {}", url),
+                body,
+            }
+            .into());
         }
         Ok(())
     }

backend/src/tams/uploader.rs

@@ -262,6 +262,12 @@ where
         match op().await {
             Ok(v) => return Ok(v),
             Err(e) => {
+                // A non-retryable HTTP status (e.g. 413 Payload Too Large, 401
+                // Unauthorized) will fail identically on every attempt — give up
+                // now rather than sleeping through the whole backoff schedule.
+                if !is_retryable(&e) {
+                    return Err(e);
+                }
                 if attempt + 1 < attempts {
                     let backoff = Duration::from_secs(1u64 << attempt);
                     tokio::time::sleep(backoff).await;
@@ -273,6 +279,14 @@ where
     Err(last_err.unwrap_or_else(|| anyhow::anyhow!("operation failed")))
 }
 
+/// Whether an error is worth retrying. A gateway/storage HTTP 4xx (except
+/// 408/429) cannot succeed on retry; everything else (network errors, 5xx) can.
+fn is_retryable(e: &anyhow::Error) -> bool {
+    e.downcast_ref::<crate::tams::client::HttpStatusError>()
+        .map(|h| h.is_retryable())
+        .unwrap_or(true)
+}
+
 /// Allocate, upload and register a single fragment (bytes already read by the
 /// caller). Returns the registered `(object_id, timerange)` on success.
 async fn upload_one(