rich paging on both fal and seedance webhook handlers (#1289)

This commit is contained in:
Brandon Thomas
2026-04-02 03:51:46 -04:00
committed by GitHub
parent 63d3d63709
commit 004a5c157a
11 changed files with 263 additions and 153 deletions
@@ -1,4 +1,3 @@
use std::fmt::{Debug, Display};
use std::sync::Arc;
use chrono::{DateTime, Utc};
@@ -7,7 +6,7 @@ use crate::notification::notification_details::NotificationDetails;
use crate::notification::notification_urgency::NotificationUrgency;
pub struct NotificationDetailsBuilder {
title: String,
pub(crate) title: Option<String>,
pub(crate) description: Option<String>,
pub(crate) urgency: Option<NotificationUrgency>,
pub(crate) event_time: DateTime<Utc>,
@@ -26,10 +25,12 @@ pub struct NotificationDetailsBuilder {
}
impl NotificationDetailsBuilder {
/// Create a builder from a summary string.
// --- Constructors ---
/// Create a builder with an explicit title.
pub fn from_title(title: String) -> Self {
Self {
title,
title: Some(title),
description: None,
urgency: None,
event_time: Utc::now(),
@@ -45,15 +46,16 @@ impl NotificationDetailsBuilder {
}
}
/// Create a builder from a boxed error. Converts to `Arc` internally for shareability.
/// Create a builder from a boxed error. Converts to `Arc` internally.
pub fn from_boxed_error(error: Box<dyn std::error::Error + Send + Sync + 'static>) -> Self {
Self::from_error(Arc::from(error))
}
/// Create a builder from an `Arc`'d error.
/// Title will be auto-generated from the error if not explicitly set.
pub fn from_error(error: Arc<dyn std::error::Error + Send + Sync + 'static>) -> Self {
Self {
title: "Notification from Error".to_string(),
title: None,
description: None,
urgency: None,
event_time: Utc::now(),
@@ -69,8 +71,10 @@ impl NotificationDetailsBuilder {
}
}
// --- Setters ---
pub fn set_title(mut self, title: String) -> Self {
self.title = title;
self.title = Some(title);
self
}
@@ -124,9 +128,16 @@ impl NotificationDetailsBuilder {
self
}
// --- Build ---
pub fn build(self) -> NotificationDetails {
let title = match self.title {
Some(t) => t,
None => Self::generate_title(&self.maybe_error, &self.http_method, &self.http_path),
};
NotificationDetails {
title: self.title,
title,
description: self.description,
urgency: self.urgency,
event_time: self.event_time,
@@ -141,4 +152,41 @@ impl NotificationDetailsBuilder {
third_party_id: self.third_party_id,
}
}
// --- Private helpers ---
/// Generate a reasonable title from available context.
///
/// Priority:
/// 1. Error + HTTP info: "POST /v1/foo - SomeError: details"
/// 2. Error alone: "SomeError: details"
/// 3. HTTP info alone: "POST /v1/foo - Unknown Error"
/// 4. Nothing: "Unknown Error"
fn generate_title(
maybe_error: &Option<Arc<dyn std::error::Error + Send + Sync + 'static>>,
http_method: &Option<String>,
http_path: &Option<String>,
) -> String {
let http_prefix = match (http_method, http_path) {
(Some(method), Some(path)) => Some(format!("{} {}", method, path)),
(None, Some(path)) => Some(path.clone()),
_ => None,
};
let error_summary = maybe_error.as_ref().map(|err| {
let msg = format!("{}", err);
if msg.len() > 150 {
format!("{}...", &msg[..147])
} else {
msg
}
});
match (http_prefix, error_summary) {
(Some(prefix), Some(summary)) => format!("{} - {}", prefix, summary),
(None, Some(summary)) => summary,
(Some(prefix), None) => format!("{} - Unknown Error", prefix),
(None, None) => "Unknown Error".to_string(),
}
}
}
@@ -19,6 +19,18 @@ pub enum DatabaseQueryError {
AnyhowError(anyhow::Error),
}
impl std::error::Error for DatabaseQueryError {}
impl std::fmt::Display for DatabaseQueryError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
DatabaseQueryError::IdempotencyDuplicateKeyError => write!(f, "Idempotency duplicate key error"),
DatabaseQueryError::SqlxError(err) => write!(f, "SQLX error: {}", err),
DatabaseQueryError::AnyhowError(err) => write!(f, "Anyhow error: {}", err),
}
}
}
impl From<anyhow::Error> for DatabaseQueryError {
fn from(err: anyhow::Error) -> Self {
DatabaseQueryError::AnyhowError(err)
@@ -58,7 +58,12 @@ pub fn run_http_server(args: CreateServerArgs) -> AnyhowResult<Server> {
.route(web::head().to(|| HttpResponse::Ok())),
)
})
.bind(bind_address)?
.bind(&bind_address)
.unwrap_or_else(|err| {
eprintln!("FATAL: Failed to bind to address '{}': {}", bind_address, err);
eprintln!("The address is likely already in use by another process.");
std::process::exit(1);
})
.workers(num_workers)
.run();
@@ -58,7 +58,12 @@ pub fn run_http_server(args: CreateServerArgs) -> AnyhowResult<Server> {
.route(web::head().to(|| HttpResponse::Ok())),
)
})
.bind(bind_address)?
.bind(&bind_address)
.unwrap_or_else(|err| {
eprintln!("FATAL: Failed to bind to address '{}': {}", bind_address, err);
eprintln!("The address is likely already in use by another process.");
std::process::exit(1);
})
.workers(num_workers)
.run();
@@ -48,6 +48,11 @@ async fn main() -> Result<(), impl Error> {
.url("/api-docs/openapi.json", ApiDoc::openapi()),
)
})
.bind(("localhost", 8989)).unwrap()
.bind(("localhost", 8989))
.unwrap_or_else(|err| {
eprintln!("FATAL: Failed to bind docs server to localhost:8989: {}", err);
eprintln!("The address is likely already in use by another process.");
std::process::exit(1);
})
.run().await
}
@@ -3,7 +3,7 @@ use std::sync::Arc;
use actix_http::StatusCode;
use actix_web::{HttpResponse, HttpResponseBuilder, ResponseError};
use anyhow::anyhow;
use crate::http_server::web_utils::user_session::require_user_session::RequireUserSessionError;
/// An error type for actix-web handlers that wraps causal errors for debugging
@@ -28,6 +28,7 @@ use crate::http_server::web_utils::user_session::require_user_session::RequireUs
/// Errors converted via `From<T>` become `UncaughtServerError` (always 500).
/// The wrapped cause is never shown to users but is available to the error
/// alerting middleware for paging and logging.
#[derive(Clone)]
pub enum AdvancedCommonWebError {
/// 400 Bad Request with a user-facing message.
BadInputWithSimpleMessage(String),
@@ -55,6 +56,17 @@ impl AdvancedCommonWebError {
Self::UncaughtServerError(Arc::new(error))
}
/// Wrap an `anyhow::Error` as an `UncaughtServerError`.
/// (`anyhow::Error` doesn't implement `std::error::Error`, so `from_error` can't accept it.)
pub fn from_anyhow_error(error: anyhow::Error) -> Self {
let boxed: Box<dyn std::error::Error + Send + Sync> = error.into();
Self::UncaughtServerError(Arc::from(boxed))
}
pub fn server_error_with_message(msg: &str) -> Self {
Self::from_anyhow_error(anyhow!("ServerErrorWithMessage: {:?}", msg))
}
/// Extract the wrapped causal error (only present for `UncaughtServerError`).
pub fn cause(&self) -> Option<&(dyn std::error::Error + Send + Sync + 'static)> {
match self {
@@ -182,6 +194,19 @@ impl From<RequireUserSessionError> for AdvancedCommonWebError {
}
}
impl From<crate::http_server::common_responses::common_web_error::CommonWebError> for AdvancedCommonWebError {
fn from(value: crate::http_server::common_responses::common_web_error::CommonWebError) -> Self {
use crate::http_server::common_responses::common_web_error::CommonWebError;
match value {
CommonWebError::BadInputWithSimpleMessage(msg) => Self::BadInputWithSimpleMessage(msg),
CommonWebError::NotAuthorized => Self::NotAuthorized,
CommonWebError::NotFound => Self::NotFound,
CommonWebError::PaymentRequired => Self::PaymentRequired,
CommonWebError::ServerError => Self::from_error(value),
}
}
}
// =============== Serialization helpers (private) ===============
#[derive(Debug, Serialize)]
@@ -199,6 +224,47 @@ struct JsonErrorWithMessage<'a> {
message: &'a str,
}
// =============== OpenAPI schema ===============
impl utoipa::PartialSchema for AdvancedCommonWebError {
fn schema() -> utoipa::openapi::RefOr<utoipa::openapi::Schema> {
utoipa::openapi::ObjectBuilder::new()
.property(
"success",
utoipa::openapi::ObjectBuilder::new()
.schema_type(utoipa::openapi::schema::Type::Boolean),
)
.required("success")
.property(
"error_code",
utoipa::openapi::ObjectBuilder::new()
.schema_type(utoipa::openapi::schema::Type::Integer)
.format(Some(utoipa::openapi::SchemaFormat::KnownFormat(
utoipa::openapi::KnownFormat::Int32,
))),
)
.required("error_code")
.property(
"error_code_str",
utoipa::openapi::ObjectBuilder::new()
.schema_type(utoipa::openapi::schema::Type::String),
)
.property(
"message",
utoipa::openapi::ObjectBuilder::new()
.schema_type(utoipa::openapi::schema::Type::String)
.description(Some("User-facing error message (present only for bad input errors)")),
)
.into()
}
}
impl utoipa::ToSchema for AdvancedCommonWebError {
fn name() -> std::borrow::Cow<'static, str> {
std::borrow::Cow::Borrowed("AdvancedCommonWebError")
}
}
// =============== Tests ===============
#[cfg(test)]
@@ -2,9 +2,10 @@ use std::collections::HashMap;
use std::sync::Arc;
use crate::billing::wallets::attempt_wallet_deduction::attempt_wallet_deduction_else_common_web_error;
use crate::http_server::common_responses::common_web_error::CommonWebError;
use crate::http_server::common_responses::advanced_common_web_error::AdvancedCommonWebError;
use crate::http_server::endpoint_helpers::refund_wallet_after_api_failure::refund_wallet_after_api_failure;
use crate::http_server::endpoints::generate::common::payments_error_test::payments_error_test;
use crate::http_server::session::lookup::user_session_feature_flags::UserSessionFeatureFlags;
use crate::http_server::validations::validate_idempotency_token_format::validate_idempotency_token_format;
use crate::state::server_state::ServerState;
use crate::util::http_download_url_to_bytes::http_download_url_to_bytes;
@@ -17,11 +18,11 @@ use artcraft_api_defs::generate::video::multi_function::seedance_2p0_multi_funct
};
use enums::by_table::prompt_context_items::prompt_context_semantic_type::PromptContextSemanticType;
use enums::by_table::prompts::prompt_type::PromptType;
use enums::common::generation_provider::GenerationProvider;
use enums::common::generation::common_model_type::CommonModelType;
use enums::common::visibility::Visibility;
use enums::common::generation::common_aspect_ratio::CommonAspectRatio;
use enums::common::generation::common_generation_mode::CommonGenerationMode;
use enums::common::generation::common_model_type::CommonModelType;
use enums::common::generation_provider::GenerationProvider;
use enums::common::visibility::Visibility;
use http_server_common::request::get_request_ip::get_request_ip;
use log::{error, info, warn};
use mysql_queries::queries::generic_inference::seedance2pro::insert_generic_inference_job_for_seedance2pro_queue_with_apriori_job_token::{
@@ -33,6 +34,8 @@ use mysql_queries::queries::prompt_context_items::insert_batch_prompt_context_it
insert_batch_prompt_context_items, InsertBatchArgs, PromptContextItem,
};
use mysql_queries::queries::prompts::insert_prompt::{insert_prompt, InsertPromptArgs};
use pager::notification::notification_details_builder::NotificationDetailsBuilder;
use pager::notification::notification_urgency::NotificationUrgency;
use seedance2pro_client::creds::seedance2pro_session::Seedance2ProSession;
use seedance2pro_client::requests::generate_video::generate_video::{
generate_video, BatchCount, GenerateVideoArgs, GenerateVideoResponse, ModelType, Resolution,
@@ -46,9 +49,6 @@ use tokens::tokens::generic_inference_jobs::InferenceJobToken;
use tokens::tokens::media_files::MediaFileToken;
use url::Url;
use url_utils::extension::extract_extension_from_url::{extract_extension_from_url, ExtractExtensions};
use pager::notification::notification_details_builder::NotificationDetailsBuilder;
use pager::notification::notification_urgency::NotificationUrgency;
use crate::http_server::session::lookup::user_session_feature_flags::UserSessionFeatureFlags;
// ======================== Result of a successful generation ========================
@@ -76,7 +76,7 @@ pub async fn seedance_2p0_multi_function_video_gen_handler(
http_request: HttpRequest,
request: Json<Seedance2p0MultiFunctionVideoGenRequest>,
server_state: web::Data<Arc<ServerState>>,
) -> Result<Json<Seedance2p0MultiFunctionVideoGenResponse>, CommonWebError> {
) -> Result<Json<Seedance2p0MultiFunctionVideoGenResponse>, AdvancedCommonWebError> {
payments_error_test(&request.prompt.as_deref().unwrap_or(""))?;
@@ -94,20 +94,20 @@ pub async fn seedance_2p0_multi_function_video_gen_handler(
.await
.map_err(|e| {
warn!("Session checker error: {:?}", e);
CommonWebError::ServerError
AdvancedCommonWebError::from_anyhow_error(e)
})?;
let user_session = match maybe_user_session {
Some(session) => session,
None => {
return Err(CommonWebError::NotAuthorized);
return Err(AdvancedCommonWebError::NotAuthorized);
}
};
let user_token = &user_session.user_token;
if let Err(reason) = validate_idempotency_token_format(&request.uuid_idempotency_token) {
return Err(CommonWebError::BadInputWithSimpleMessage(reason));
return Err(AdvancedCommonWebError::BadInputWithSimpleMessage(reason));
}
// --- Collect all media tokens to look up ---
@@ -148,7 +148,7 @@ pub async fn seedance_2p0_multi_function_video_gen_handler(
.await
.map_err(|err| {
error!("Error inserting idempotency token: {:?}", err);
CommonWebError::BadInputWithSimpleMessage("repeated idempotency token".to_string())
AdvancedCommonWebError::BadInputWithSimpleMessage("repeated idempotency token".to_string())
})?;
// --- Calculate cost and charge wallet upfront (before uploads) ---
@@ -221,16 +221,15 @@ pub async fn seedance_2p0_multi_function_video_gen_handler(
}
Err(err) => {
warn!("Regular session fallback also failed for user {:?}: {:?}", user_token, err);
if let Err(page_err) = server_state.pager.enqueue_page(
NotificationDetailsBuilder::from_boxed_error(err.into())
.set_title("Seedance 2.0 generation failed (whitelist + fallback)".to_string())
.set_urgency(Some(NotificationUrgency::High))
.build()
) {
let notification = NotificationDetailsBuilder::from_boxed_error(err.clone().into())
.set_title("Seedance 2.0 generation failed (whitelist + fallback)".to_string())
.set_urgency(Some(NotificationUrgency::High))
.build();
if let Err(page_err) = server_state.pager.enqueue_page(notification) {
warn!("Failed to enqueue pager alert: {:?}", page_err);
}
refund_wallet_after_api_failure(&deduction_result.ledger_entry_token, &mut mysql_connection).await?;
return Err(CommonWebError::ServerError);
return Err(err);
}
}
}
@@ -255,16 +254,15 @@ pub async fn seedance_2p0_multi_function_video_gen_handler(
Ok(result) => result,
Err(err) => {
warn!("Error calling seedance2pro generate_video: {:?}", err);
if let Err(page_err) = server_state.pager.enqueue_page(
NotificationDetailsBuilder::from_boxed_error(err.into())
.set_title("Seedance 2.0 generation failed".to_string())
.set_urgency(Some(NotificationUrgency::High))
.build()
) {
let notification = NotificationDetailsBuilder::from_boxed_error(err.clone().into())
.set_title("Seedance 2.0 generation failed".to_string())
.set_urgency(Some(NotificationUrgency::High))
.build();
if let Err(page_err) = server_state.pager.enqueue_page(notification) {
warn!("Failed to enqueue pager alert: {:?}", page_err);
}
refund_wallet_after_api_failure(&deduction_result.ledger_entry_token, &mut mysql_connection).await?;
return Err(CommonWebError::ServerError);
return Err(err);
}
}
};
@@ -286,7 +284,7 @@ pub async fn seedance_2p0_multi_function_video_gen_handler(
.await
.map_err(|err| {
error!("Error starting MySQL transaction: {:?}", err);
CommonWebError::ServerError
AdvancedCommonWebError::from_error(err)
})?;
// NB: Don't fail the job if the prompt insert fails.
@@ -419,7 +417,7 @@ pub async fn seedance_2p0_multi_function_video_gen_handler(
Err(err) => {
warn!("Error inserting seedance2pro inference job (order_id={}): {:?}", order_id, err);
if i == 0 {
return Err(CommonWebError::ServerError);
return Err(AdvancedCommonWebError::from_error(err));
}
}
}
@@ -427,7 +425,7 @@ pub async fn seedance_2p0_multi_function_video_gen_handler(
let first_job_token = all_job_tokens.first().cloned().ok_or_else(|| {
error!("No inference job token was created");
CommonWebError::ServerError
AdvancedCommonWebError::server_error_with_message("No inference job token was created")
})?;
transaction
@@ -435,7 +433,7 @@ pub async fn seedance_2p0_multi_function_video_gen_handler(
.await
.map_err(|err| {
error!("Error committing MySQL transaction: {:?}", err);
CommonWebError::ServerError
AdvancedCommonWebError::from_error(err)
})?;
Ok(Json(Seedance2p0MultiFunctionVideoGenResponse {
@@ -497,14 +495,14 @@ async fn upload_and_generate(
resolution: Resolution,
batch_count: BatchCount,
duration_seconds: u8,
) -> Result<SeedanceGenerationResult, CommonWebError> {
) -> Result<SeedanceGenerationResult, AdvancedCommonWebError> {
// --- Upload files to seedance2pro CDN ---
let start_frame_url = match request.start_frame_media_token.as_ref() {
None => None,
Some(token) => match file_urls_by_token.get(token) {
None => return Err(CommonWebError::BadInputWithSimpleMessage("Start frame media not found.".to_string())),
None => return Err(AdvancedCommonWebError::BadInputWithSimpleMessage("Start frame media not found.".to_string())),
Some(url) => Some(upload_to_seedance2pro(session, url).await?),
}
};
@@ -512,7 +510,7 @@ async fn upload_and_generate(
let end_frame_url = match request.end_frame_media_token.as_ref() {
None => None,
Some(token) => match file_urls_by_token.get(token) {
None => return Err(CommonWebError::BadInputWithSimpleMessage("End frame media not found.".to_string())),
None => return Err(AdvancedCommonWebError::BadInputWithSimpleMessage("End frame media not found.".to_string())),
Some(url) => Some(upload_to_seedance2pro(session, url).await?),
}
};
@@ -578,7 +576,7 @@ async fn upload_and_generate(
let gen_response = generate_video(video_gen_args).await
.map_err(|err| {
warn!("Error calling seedance2pro generate_video: {:?}", err);
CommonWebError::ServerError
AdvancedCommonWebError::from_error(err)
})?;
Ok(SeedanceGenerationResult {
@@ -594,7 +592,7 @@ async fn upload_reference_tokens_to_seedance2pro(
file_urls_by_token: &HashMap<MediaFileToken, Url>,
maybe_tokens: Option<&[MediaFileToken]>,
label: &str,
) -> Result<Option<Vec<String>>, CommonWebError> {
) -> Result<Option<Vec<String>>, AdvancedCommonWebError> {
let tokens = match maybe_tokens {
None => return Ok(None),
Some(tokens) if tokens.is_empty() => return Ok(None),
@@ -605,7 +603,7 @@ async fn upload_reference_tokens_to_seedance2pro(
for token in tokens {
match file_urls_by_token.get(token) {
None => return Err(CommonWebError::BadInputWithSimpleMessage(
None => return Err(AdvancedCommonWebError::BadInputWithSimpleMessage(
format!("{} media not found: {:?}", label, token),
)),
Some(url) => {
@@ -620,7 +618,7 @@ async fn upload_reference_tokens_to_seedance2pro(
async fn upload_to_seedance2pro(
session: &Seedance2ProSession,
our_cdn_url: &Url,
) -> Result<String, CommonWebError> {
) -> Result<String, AdvancedCommonWebError> {
let extension = extract_extension_from_url(our_cdn_url, &ExtractExtensions::All)
.map(|ext| ext.without_period().to_string())
.unwrap_or_else(|| "png".to_string());
@@ -631,7 +629,7 @@ async fn upload_to_seedance2pro(
.await
.map_err(|err| {
warn!("Error downloading media file from CDN: {:?}", err);
CommonWebError::ServerError
AdvancedCommonWebError::from_anyhow_error(err)
})?
.to_vec();
@@ -643,7 +641,7 @@ async fn upload_to_seedance2pro(
.await
.map_err(|err| {
warn!("Error preparing seedance2pro file upload: {:?}", err);
CommonWebError::ServerError
AdvancedCommonWebError::from_error(err)
})?;
let upload_result = upload_file(UploadFileArgs {
@@ -654,7 +652,7 @@ async fn upload_to_seedance2pro(
.await
.map_err(|err| {
warn!("Error uploading file to seedance2pro: {:?}", err);
CommonWebError::ServerError
AdvancedCommonWebError::from_error(err)
})?;
Ok(upload_result.public_url)
@@ -1,65 +1,19 @@
use std::fmt;
use std::sync::Arc;
use crate::http_server::common_responses::advanced_common_web_error::AdvancedCommonWebError;
use crate::http_server::endpoints::webhooks::process_failure::handle_failed_fal_webhook::handle_failed_fal_webhook;
use crate::http_server::endpoints::webhooks::process_success::handle_successful_fal_webhook::handle_successful_fal_webhook;
use crate::state::server_state::ServerState;
use actix_web::error::ResponseError;
use actix_web::http::StatusCode;
use actix_web::web::Bytes;
use actix_web::web::Json;
use actix_web::{web, HttpRequest, HttpResponse};
use anyhow::Error;
use actix_web::{web, HttpRequest};
use fal_client::webhook_api::parse_webhook_inner_payload::parse_webhook_inner_payload;
use fal_client::webhook_api::payload::webhook_inner_payload::WebhookInnerPayload;
use fal_client::webhook_api::parse_webhook_payload::parse_webhook_payload;
use fal_client::webhook_api::payload::webhook_inner_payload::WebhookInnerPayload;
use http_server_common::response::response_success_helpers::SimpleGenericJsonSuccess;
use http_server_common::response::serialize_as_json_error::serialize_as_json_error;
use log::{error, info, warn};
use pager::notification::notification_details_builder::NotificationDetailsBuilder;
use pager::notification::notification_urgency::NotificationUrgency;
use utoipa::ToSchema;
// =============== Error Response ===============
#[derive(Debug, Serialize, ToSchema)]
pub enum FalWebhookError {
BadInput(String),
NotFound,
NotAuthorized,
ServerError,
}
impl ResponseError for FalWebhookError {
fn status_code(&self) -> StatusCode {
match *self {
FalWebhookError::BadInput(_) => StatusCode::BAD_REQUEST,
FalWebhookError::NotFound => StatusCode::NOT_FOUND,
FalWebhookError::NotAuthorized => StatusCode::UNAUTHORIZED,
FalWebhookError::ServerError => StatusCode::INTERNAL_SERVER_ERROR,
}
}
fn error_response(&self) -> HttpResponse {
serialize_as_json_error(self)
}
}
impl std::error::Error for FalWebhookError {}
// NB: Not using derive_more::Display since Clion doesn't understand it.
impl fmt::Display for FalWebhookError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{:?}", self)
}
}
impl From<anyhow::Error> for FalWebhookError {
fn from(value: Error) -> Self {
info!("Converting anyhow::Error to FalWebhookError: {:?}", value);
FalWebhookError::ServerError
}
}
// =============== Handler ===============
@@ -68,14 +22,16 @@ pub async fn fal_webhook_handler(
http_request: HttpRequest,
request_body_bytes: Bytes,
server_state: web::Data<Arc<ServerState>>,
) -> Result<Json<SimpleGenericJsonSuccess>, FalWebhookError> {
) -> Result<Json<SimpleGenericJsonSuccess>, AdvancedCommonWebError> {
// Step 1: Parse bytes into a UTF-8 string and log it.
let raw_body = String::from_utf8(request_body_bytes.to_vec())
.map_err(|err| {
error!("FAL webhook: could not decode request body to UTF-8: {:?}", err);
enqueue_parse_error_alert(&server_state, &http_request, "UTF-8 decode failed", &err, None);
FalWebhookError::BadInput("Could not decode request body to UTF-8".to_string())
// Ordinarily this is a user input error, but I want to know when this happens and see the error trace:
// AdvancedCommonWebError::BadInputWithSimpleMessage("Could not decode request body to UTF-8".to_string())
AdvancedCommonWebError::from_error(err)
})?;
info!("Received FAL webhook body: {}", raw_body);
@@ -85,7 +41,9 @@ pub async fn fal_webhook_handler(
.map_err(|err| {
error!("FAL webhook: could not parse webhook payload: {:?}", err);
enqueue_parse_error_alert(&server_state, &http_request, "JSON parse failed", &err, Some(&raw_body));
FalWebhookError::BadInput("Could not parse webhook payload".to_string())
// Ordinarily this is a user input error, but I want to know when this happens and see the error trace:
// AdvancedCommonWebError::BadInputWithSimpleMessage("Could not parse webhook payload".to_string())
AdvancedCommonWebError::from_error(err)
})?;
let request_id = webhook_payload.request_id.as_str();
@@ -113,15 +71,14 @@ pub async fn fal_webhook_handler(
"FAL webhook payload_error for request_id {}: {}",
request_id, payload_error_data.payload_error,
);
// Treat payload encoding errors as server errors — the request was OK but
// FAL couldn't encode the payload, so we can't process it.
Err(FalWebhookError::ServerError)
Err(AdvancedCommonWebError::from_anyhow_error(
anyhow::anyhow!("FAL payload_error: {}", payload_error_data.payload_error)
))
}
};
match result {
Ok(success) => Ok(success),
Err(err) => {
if let Err(ref err) = result {
if err.is_server_error() {
error!("FAL webhook error for request_id {}: {:?}", request_id, err);
let description = format!(
@@ -129,24 +86,33 @@ pub async fn fal_webhook_handler(
request_id, err, raw_body,
);
let notification = NotificationDetailsBuilder::from_boxed_error(err.into())
.set_title("FAL webhook processing failed".to_string())
let mut builder = if let Some(cause_arc) = err.clone_cause_arc() {
NotificationDetailsBuilder::from_error(cause_arc)
.set_title(format!("FAL webhook processing failed (request_id: {})", request_id))
} else {
NotificationDetailsBuilder::from_title(
format!("FAL webhook processing failed (request_id: {})", request_id)
)
};
builder = builder
.set_description(Some(description))
.set_third_party_id(Some(request_id.to_string()))
.set_urgency(Some(NotificationUrgency::High))
.set_http_method(Some(http_request.method().to_string()))
.set_http_path(Some(http_request.path().to_string()))
.build();
.set_http_path(Some(http_request.path().to_string()));
if let Err(pager_err) = server_state.pager.enqueue_page(notification) {
if let Err(pager_err) = server_state.pager.enqueue_page(builder.build()) {
error!("Failed to enqueue FAL webhook pager alert: {:?}", pager_err);
}
Err(FalWebhookError::ServerError)
}
}
result
}
// =============== Private helpers ===============
/// Send a pager alert for early parse failures (before we have a request_id).
fn enqueue_parse_error_alert<E: std::fmt::Debug>(
server_state: &ServerState,
@@ -1,32 +1,13 @@
use crate::http_server::common_responses::advanced_common_web_error::AdvancedCommonWebError;
use crate::state::server_state::ServerState;
use actix_web::web::Json;
use log::{error, info, warn};
use mysql_queries::queries::generic_inference::fal::get_inference_job_by_fal_id::get_inference_job_by_fal_id;
use mysql_queries::queries::generic_inference::job::mark_job_failed_by_token::{mark_job_failed_by_token, MarkJobFailedByTokenArgs};
use http_server_common::response::response_success_helpers::SimpleGenericJsonSuccess;
use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory;
use fal_client::webhook_api::payload::webhook_error_type::WebhookErrorType;
use fal_client::webhook_api::payload::webhook_inner_payload::ErrorData;
use crate::http_server::endpoints::webhooks::fal_webhook_handler::FalWebhookError;
use crate::state::server_state::ServerState;
/// Map a FAL WebhookErrorType to a frontend failure category.
///
/// Returns `GenerationFailed` as the default if the error type is None or unrecognized.
fn guess_failure_category(error_type: Option<&WebhookErrorType>) -> FrontendFailureCategory {
match error_type {
Some(WebhookErrorType::ContentPolicyViolation) => FrontendFailureCategory::RuleBansUserContent,
Some(WebhookErrorType::FaceDetectionError) => FrontendFailureCategory::FaceNotDetected,
// These all map to GenerationFailed for now, but are listed explicitly,
// so we can refine them later.
Some(WebhookErrorType::NoMediaGenerated)
| Some(WebhookErrorType::ImageTooSmall)
| Some(WebhookErrorType::ImageTooLarge)
| Some(WebhookErrorType::ImageLoadError)
| Some(WebhookErrorType::FileDownloadError)
| Some(WebhookErrorType::FileTooLarge) => FrontendFailureCategory::GenerationFailed,
_ => FrontendFailureCategory::GenerationFailed,
}
}
use http_server_common::response::response_success_helpers::SimpleGenericJsonSuccess;
use log::{error, info, warn};
use mysql_queries::queries::generic_inference::fal::get_inference_job_by_fal_id::get_inference_job_by_fal_id;
use mysql_queries::queries::generic_inference::job::mark_job_failed_by_token::{mark_job_failed_by_token, MarkJobFailedByTokenArgs};
/// Handle a FAL webhook with status ERROR.
///
@@ -36,7 +17,7 @@ pub async fn handle_failed_fal_webhook(
request_id: &str,
error_data: &ErrorData,
maybe_top_level_error: Option<&str>,
) -> Result<Json<SimpleGenericJsonSuccess>, FalWebhookError> {
) -> Result<Json<SimpleGenericJsonSuccess>, AdvancedCommonWebError> {
info!(
"FAL webhook ERROR for request_id {}: top_level_error={:?}, error_type={:?}, message={:?}",
@@ -51,11 +32,11 @@ pub async fn handle_failed_fal_webhook(
Ok(Some(record)) => record,
Ok(None) => {
warn!("Could not find job record by fal request_id: {}", request_id);
return Err(FalWebhookError::NotFound);
return Err(AdvancedCommonWebError::NotFound);
}
Err(err) => {
error!("Error querying job record for request_id {}: {:?}", request_id, err);
return Err(FalWebhookError::ServerError);
return Err(AdvancedCommonWebError::from_anyhow_error(err));
}
};
@@ -99,7 +80,7 @@ pub async fn handle_failed_fal_webhook(
request_id,
err,
);
return Err(FalWebhookError::ServerError);
return Err(AdvancedCommonWebError::from_anyhow_error(err));
}
info!(
@@ -110,3 +91,22 @@ pub async fn handle_failed_fal_webhook(
Ok(SimpleGenericJsonSuccess::wrapped(true))
}
// =============== Private helpers ===============
/// Map a FAL WebhookErrorType to a frontend failure category.
///
/// Returns `GenerationFailed` as the default if the error type is None or unrecognized.
fn guess_failure_category(error_type: Option<&WebhookErrorType>) -> FrontendFailureCategory {
match error_type {
Some(WebhookErrorType::ContentPolicyViolation) => FrontendFailureCategory::RuleBansUserContent,
Some(WebhookErrorType::FaceDetectionError) => FrontendFailureCategory::FaceNotDetected,
Some(WebhookErrorType::NoMediaGenerated)
| Some(WebhookErrorType::ImageTooSmall)
| Some(WebhookErrorType::ImageTooLarge)
| Some(WebhookErrorType::ImageLoadError)
| Some(WebhookErrorType::FileDownloadError)
| Some(WebhookErrorType::FileTooLarge) => FrontendFailureCategory::GenerationFailed,
_ => FrontendFailureCategory::GenerationFailed,
}
}
@@ -1,11 +1,11 @@
use actix_web::web::Json;
use http_server_common::response::response_success_helpers::SimpleGenericJsonSuccess;
use log::{info, warn};
use mysql_queries::queries::generic_inference::fal::get_inference_job_by_fal_id::get_inference_job_by_fal_id;
use mysql_queries::queries::generic_inference::fal::mark_fal_generic_inference_job_successfully_done::{mark_fal_generic_inference_job_successfully_done, MarkJobArgs};
use http_server_common::response::response_success_helpers::SimpleGenericJsonSuccess;
use serde_json::Value;
use crate::http_server::endpoints::webhooks::fal_webhook_handler::FalWebhookError;
use crate::http_server::common_responses::advanced_common_web_error::AdvancedCommonWebError;
use crate::state::server_state::ServerState;
use super::process_image_payload::process_image_payload;
@@ -18,7 +18,7 @@ pub async fn handle_successful_fal_webhook(
server_state: &ServerState,
request_id: &str,
payload: &Value,
) -> Result<Json<SimpleGenericJsonSuccess>, FalWebhookError> {
) -> Result<Json<SimpleGenericJsonSuccess>, AdvancedCommonWebError> {
let db_result = get_inference_job_by_fal_id(
request_id,
@@ -29,11 +29,11 @@ pub async fn handle_successful_fal_webhook(
Ok(Some(record)) => record,
Ok(None) => {
warn!("Could not find job record by fal request_id: {}", request_id);
return Err(FalWebhookError::NotFound)
return Err(AdvancedCommonWebError::NotFound)
},
Err(err) => {
warn!("Error querying job record for request_id {}: {:?}", request_id, err);
return Err(FalWebhookError::ServerError);
return Err(AdvancedCommonWebError::from_anyhow_error(err));
}
};
@@ -74,7 +74,7 @@ pub async fn handle_successful_fal_webhook(
phantom: Default::default(),
}).await.map_err(|err| {
warn!("Error marking job as successfully done for request_id {}: {:?}", request_id, err);
FalWebhookError::ServerError
AdvancedCommonWebError::from_anyhow_error(err)
})?;
} else {
warn!("No media token found in payload for request_id {} / job {:?}", request_id, job.job_token);
@@ -713,7 +713,12 @@ pub async fn serve(server_state: ServerState) -> AnyhowResult<()>
add_routes(app, old_server_environment)
})
.bind(bind_address)?
.bind(&bind_address)
.unwrap_or_else(|err| {
eprintln!("FATAL: Failed to bind to address '{}': {}", bind_address, err);
eprintln!("The address is likely already in use by another process.");
std::process::exit(1);
})
.workers(num_workers)
.run()
.await?;