Validating business data · Consumer

Separate deserialization from business rules and classify permanent and temporary failures explicitly.

Json<T> converts a message's JSON structure into a Rust type. Rules such as a positive order amount, a nonempty ID, or an account being open for processing must be written separately. Choosing the right error class prevents unfixable messages from entering unnecessary retry cycles.

Decoding versus business validation#

Condition	Responsible layer	Typical outcome
Invalid envelope/version/content	Framework contract selection	Permanent rejection before the handler
Malformed JSON or incorrect field type	`Json<T>` and Serde	Permanent decoding error
Empty ID or invalid amount	Application validation	Explicit permanent error
Temporary database or service failure	Application error conversion	Retryable error when appropriate
Actor lacks permission to process	Application guard or policy	Explicit policy rejection

The framework does not infer business retry decisions from error text. Deserialize validates the declared data representation. Do not transplant an HTTP validation extractor or an undefined #[validate] marker into a Consumer handler; call shared validation through an ordinary Rust function or service.

Write a pure validation function#

use lilyrs::queue::QueueHandlerError;
use serde::Deserialize;

#[derive(Deserialize)]
struct OrderInput {
    order_id: String,
    amount_minor: i64,
}

fn validate_order(input: &OrderInput) -> Result<(), QueueHandlerError> {
    if input.order_id.trim().is_empty() {
        return Err(QueueHandlerError::permanent("ORDER_ID_REQUIRED"));
    }
    if input.amount_minor <= 0 {
        return Err(QueueHandlerError::permanent("ORDER_AMOUNT_INVALID"));
    }
    Ok(())
}

The function performs no I/O, so it can be unit-tested without a broker. Error codes are bounded and stable; avoid embedding raw payloads, credentials, or sensitive business data in them. Another transport can call the same validator.

Use it in a handler#

Replace the first worker's method with the following and add the preceding OrderInput and validate_order definitions to the same module. Keep the first application's imports for queue, Json, and QueueHandlerError:

#[queue("orders.created", version = 1, content = "json")]
async fn created(&self, Json(event): Json<OrderInput>) -> Result<(), QueueHandlerError> {
    validate_order(&event)?;
    println!("{}:{}", event.order_id, event.amount_minor);
    Ok(())
}

This example adds amount_minor to the payload. Send the body below through the publisher's serde_json::json! call; the original body containing only order_id is insufficient for this Rust type:

{"order_id":"order-42","amount_minor":1250}

Permanent rejection does not mean the application manually ACKs the message. Lily carries the error class into retry/dead-letter and settlement policy. Completing required validation before the first business effect reduces the risk of partial side effects.

Handle external checks explicitly#

Checks requiring I/O, such as reading account status from a database, can live in a Scoped service. retryable_with_source can represent a temporary connection failure; business policy determines the class of a verified result such as a closed account. Treating every database error as permanent can lose work, while treating every error as retryable can repeatedly process permanent failures.

Payload validation does not provide authentication. A user_id field or header is not by itself a verified actor. See guards and delivery-local state for passing verified identity as typed local state and checking permission before the handler.