Ensure that bad HTTP status codes result in the producer retrying

Before, this would result in a non-ephemeral `InternalProducerClientError`
as the client would still attempt to decode JSON from the response body
even on a non-200 HTTP status.

Author: Qqwy

opsqueue/src/producer/client.rs

@@ -62,7 +62,8 @@ impl Client {
                 .http_client
                 .get(format!("{base_url}/submissions/count"))
                 .send()
-                .await?;
+                .await?
+                .error_for_status()?;
             let bytes = resp.bytes().await?;
             let body = serde_json::from_slice(&bytes)?;
 
@@ -92,7 +93,8 @@ impl Client {
                 .headers(otel_trace_carrier.try_into().unwrap_or_default())
                 .json(submission)
                 .send()
-                .await?;
+                .await?
+                .error_for_status()?;
             let bytes = resp.bytes().await?;
             let body = serde_json::from_slice(&bytes)?;
             Ok(body)
@@ -118,7 +120,8 @@ impl Client {
                 .http_client
                 .get(format!("{base_url}/submissions/{submission_id}"))
                 .send()
-                .await?;
+                .await?
+                .error_for_status()?;
             let bytes = resp.bytes().await?;
             let body = serde_json::from_slice(&bytes)?;
             Ok(body)
@@ -143,7 +146,8 @@ impl Client {
                     "{base_url}/submissions/lookup_id_by_prefix/{prefix}"
                 ))
                 .send()
-                .await?;
+                .await?
+                .error_for_status()?;
             let bytes = resp.bytes().await?;
             let body = serde_json::from_slice(&bytes)?;
             Ok(body)
@@ -183,14 +187,14 @@ pub enum InternalProducerClientError {
 impl InternalProducerClientError {
     pub fn is_ephemeral(&self) -> bool {
         match self {
-            // NOTE: In the case of an ungraceful restart, this case might theoretically trigger.
+            // In the case of an ungraceful restart, this case might theoretically trigger.
             // So even cleaner would be a tiny retry loop for this special case.
             // However, we certainly **do not** want to wait multiple minutes before returning.
             Self::ResponseDecodingError(_) => false,
-            // NOTE: reqwest doesn't make this very easy as it has a single error typed used for _everything_
+            // Reqwest doesn't make this very easy as it has a single error typed used for _everything_
             // Maybe a different HTTP client library is nicer in this regard?
             Self::HTTPClientError(inner) => {
-                inner.is_connect() || inner.is_timeout() || inner.is_decode()
+                inner.is_status() || inner.is_connect() || inner.is_timeout() || inner.is_decode()
             }
         }
     }