Don't save Kinovi CDN error pages as image results (#216)

The Kinovi worker's Midjourney image importer
(`process_successful_image_job`) and its character importer
(`process_successful_character`) read the response body without checking
the HTTP status. If a result URL answers 403, 404 or 5xx (an expired
signed URL, a CDN hiccup), the error body is uploaded as `image/png`,
inserted as the user's media file, and the job is marked
`complete_success`. Nothing retries it and nothing refunds the user. In
the character importer, the error page becomes the character's avatar or
full image.

The video and audio importers in the same crate already call
`error_for_status()` before reading the body.

## How to trigger it

A completed Midjourney order whose image URL returns a non-2xx status
when the worker downloads it.

## Fix

Both importers now download through a small `download_result_bytes`
helper that returns an error on non-2xx responses. Each caller then
handles the error the way it handles a network failure today:

- **Images:** a failed first image returns an error, and the order is
retried on the next poll. A failed later image in the grid is logged and
skipped. The video importer already behaves this way for HTTP errors.
- **Characters:** a failed image is logged and skipped, as before. If
every image fails, the character is still activated without images and
the job still completes. That's the existing behavior, which I haven't
changed. The only difference is that an error page is no longer saved as
the avatar.

The video and audio importers are unchanged. Unlike the video path, an
image download failure doesn't page. An image link that has expired for
good is retried on every poll while the job is pending, the same as a
video order with a dead link.

`grok_api_job` (`process_successful_job.rs`), `gmicloud_job`
(`process_successful_job.rs`, video and thumbnail) and `worldlabs_job`
(`process_successful_job.rs`, splat and thumbnail) also download with
`reqwest::get(...).bytes()` and don't check the status. Those crates
don't share a dependency with this one where the helper could live. I
left them out to keep this change small, and can follow up if you tell
me where a shared helper should go.

## Tests

`download_result_bytes::tests` serves one canned response from a local
`TcpListener`. A 404 response must be an error (this test fails without
the `error_for_status()` call), and a 200 response must return the body.

- `SQLX_OFFLINE=true cargo test -p seedance2-pro-job`: 25 passed, 0
failed (2 of them new).
- `cargo clippy -p seedance2-pro-job --no-deps` reports nothing on the
changed lines.


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

<!-- coderabbit-ai-generated:v1 -->

## Summary by CodeRabbit

* **Bug Fixes**
* Improved handling of generated image downloads. Failed downloads now
produce clearer error messages, and unsuccessful HTTP responses are
rejected.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

crates/service/job/seedance2_pro_job/src/download_result_bytes.rs

@@ -0,0 +1,63 @@
+use anyhow::anyhow;
+
+use errors::AnyhowResult;
+
+/// Download a finished result file (eg. a generated image) from the provider's CDN.
+///
+/// Non-2xx responses are errors: otherwise an expired link's error page would be
+/// stored as the user's result and the job marked successful.
+pub async fn download_result_bytes(client: &reqwest::Client, url: &str) -> AnyhowResult<Vec<u8>> {
+  let bytes = client.get(url)
+    .send()
+    .await
+    .map_err(|err| anyhow!("reqwest error downloading {}: {:?}", url, err))?
+    .error_for_status()
+    .map_err(|err| anyhow!("HTTP error downloading {}: {:?}", url, err))?
+    .bytes()
+    .await
+    .map_err(|err| anyhow!("error reading bytes from {}: {:?}", url, err))?;
+
+  Ok(bytes.to_vec())
+}
+
+#[cfg(test)]
+mod tests {
+  use std::io::{Read, Write};
+  use std::net::TcpListener;
+  use std::thread;
+
+  use super::*;
+
+  #[tokio::test]
+  async fn returns_body_on_success() {
+    let url = serve_one_response("200 OK", b"image-bytes");
+    let bytes = download_result_bytes(&test_client(), &url).await.unwrap();
+    assert_eq!(bytes, b"image-bytes");
+  }
+
+  #[tokio::test]
+  async fn rejects_http_error_status() {
+    // An expired or missing CDN object must not be saved as the result file.
+    let url = serve_one_response("404 Not Found", b"<Error>NoSuchKey</Error>");
+    assert!(download_result_bytes(&test_client(), &url).await.is_err());
+  }
+
+  fn test_client() -> reqwest::Client {
+    reqwest::Client::builder().no_proxy().build().unwrap()
+  }
+
+  /// Serve a single canned HTTP response on a random local port and return its URL.
+  fn serve_one_response(status: &'static str, body: &'static [u8]) -> String {
+    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    let url = format!("http://{}/result.png", listener.local_addr().unwrap());
+    thread::spawn(move || {
+      let (mut stream, _) = listener.accept().unwrap();
+      let mut request = [0u8; 4096];
+      let _ = stream.read(&mut request);
+      let head = format!("HTTP/1.1 {}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", status, body.len());
+      stream.write_all(head.as_bytes()).unwrap();
+      stream.write_all(body).unwrap();
+    });
+    url
+  }
+}

crates/service/job/seedance2_pro_job/src/jobs/character_polling_job/process_successful_character.rs

@@ -16,6 +16,7 @@ use mysql_queries::queries::generic_inference::web::mark_generic_inference_job_s
 use mysql_queries::queries::media_files::create::insert_builder::media_file_insert_builder::MediaFileInsertBuilder;
 use kinovi_web_client::requests::poll_characters::poll_characters::CharacterStatus;
 
+use crate::download_result_bytes::download_result_bytes;
 use crate::job_dependencies::JobDependencies;
 
 const PREFIX: &str = "artcraft_";
@@ -117,14 +118,7 @@ async fn download_and_create_media_file(
 ) -> AnyhowResult<tokens::tokens::media_files::MediaFileToken> {
   info!("Downloading character image: {}", image_url);
 
-  let image_bytes: Vec<u8> = deps.download_client.get(image_url)
-    .send()
-    .await
-    .map_err(|err| anyhow!("reqwest error downloading image: {:?}", err))?
-    .bytes()
-    .await
-    .map_err(|err| anyhow!("error reading image bytes: {:?}", err))?
-    .to_vec();
+  let image_bytes: Vec<u8> = download_result_bytes(&deps.download_client, image_url).await?;
 
   info!("Downloaded {} bytes from {}", image_bytes.len(), image_url);
 

crates/service/job/seedance2_pro_job/src/jobs/order_processing_job/process_successful_image_job.rs

@@ -19,6 +19,7 @@ use tokens::tokens::batch_generations::BatchGenerationToken;
 use tokens::tokens::media_files::MediaFileToken;
 
 use crate::alert_on_error::alert_pager_and_return_err;
+use crate::download_result_bytes::download_result_bytes;
 use crate::job_dependencies::JobDependencies;
 use crate::jobs::order_processing_job::is_job_status_terminal::is_job_status_terminal;
 
@@ -186,14 +187,9 @@ async fn download_and_store_one_image(
     index + 1, order.results.len(), order.order_id, image_url,
   );
 
-  let image_bytes: Vec<u8> = deps.download_client.get(image_url)
-    .send()
+  let image_bytes: Vec<u8> = download_result_bytes(&deps.download_client, image_url)
     .await
-    .map_err(|err| anyhow!("reqwest error downloading image {}: {:?}", index, err))?
-    .bytes()
-    .await
-    .map_err(|err| anyhow!("error reading image bytes for index {}: {:?}", index, err))?
-    .to_vec();
+    .map_err(|err| anyhow!("error downloading image {}: {:?}", index, err))?;
 
   info!(
     "Downloaded {} bytes for image {}/{} of order {}",

crates/service/job/seedance2_pro_job/src/main.rs

@@ -39,6 +39,7 @@ use crate::startup::build_pager::build_pager;
 use crate::startup::kinovi_setup::{get_kinovi_session, get_kinovi_version};
 
 pub mod alert_on_error;
+pub mod download_result_bytes;
 pub mod http_server;
 pub mod job_dependencies;
 pub mod jobs;