Trait Storage 

pub trait Storage: Send + Sync {
    // Required methods
    fn create_file<'life0, 'async_trait>(
        &'life0 self,
        name: String,
        description: Option<String>,
        templates: Vec<RequestTemplateInput>,
    ) -> Pin<Box<dyn Future<Output = Result<FileId>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn create_file_stream<'life0, 'async_trait, S>(
        &'life0 self,
        stream: S,
    ) -> Pin<Box<dyn Future<Output = Result<FileStreamResult>> + Send + 'async_trait>>
       where S: 'async_trait + Stream<Item = FileStreamItem> + Send + Unpin,
             Self: 'async_trait,
             'life0: 'async_trait;
    fn get_file<'life0, 'async_trait>(
        &'life0 self,
        file_id: FileId,
    ) -> Pin<Box<dyn Future<Output = Result<File>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn get_file_from_primary_pool<'life0, 'async_trait>(
        &'life0 self,
        file_id: FileId,
    ) -> Pin<Box<dyn Future<Output = Result<File>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn list_files<'life0, 'async_trait>(
        &'life0 self,
        filter: FileFilter,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<File>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn get_file_content<'life0, 'async_trait>(
        &'life0 self,
        file_id: FileId,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<FileContentItem>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn get_file_content_stream(
        &self,
        file_id: FileId,
        offset: usize,
        search: Option<String>,
    ) -> Pin<Box<dyn Stream<Item = Result<FileContentItem>> + Send>>;
    fn get_file_template_stats<'life0, 'async_trait>(
        &'life0 self,
        file_id: FileId,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<ModelTemplateStats>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn delete_file<'life0, 'async_trait>(
        &'life0 self,
        file_id: FileId,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn create_batch<'life0, 'async_trait>(
        &'life0 self,
        input: BatchInput,
    ) -> Pin<Box<dyn Future<Output = Result<Batch>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn create_batch_record<'life0, 'async_trait>(
        &'life0 self,
        input: BatchInput,
    ) -> Pin<Box<dyn Future<Output = Result<Batch>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn populate_batch<'life0, 'async_trait>(
        &'life0 self,
        batch_id: BatchId,
        file_id: FileId,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn get_batch<'life0, 'async_trait>(
        &'life0 self,
        batch_id: BatchId,
    ) -> Pin<Box<dyn Future<Output = Result<Batch>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn get_batch_status<'life0, 'async_trait>(
        &'life0 self,
        batch_id: BatchId,
    ) -> Pin<Box<dyn Future<Output = Result<BatchStatus>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn list_file_batches<'life0, 'async_trait>(
        &'life0 self,
        file_id: FileId,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<BatchStatus>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn list_batches<'life0, 'async_trait>(
        &'life0 self,
        filter: ListBatchesFilter,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<Batch>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn get_batch_by_output_file_id<'life0, 'async_trait>(
        &'life0 self,
        file_id: FileId,
        file_type: OutputFileType,
    ) -> Pin<Box<dyn Future<Output = Result<Option<Batch>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn get_batch_requests<'life0, 'async_trait>(
        &'life0 self,
        batch_id: BatchId,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<AnyRequest>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn get_batch_results_stream(
        &self,
        batch_id: BatchId,
        offset: usize,
        search: Option<String>,
        status: Option<String>,
    ) -> Pin<Box<dyn Stream<Item = Result<BatchResultItem>> + Send>>;
    fn get_cancelled_batch_ids<'life0, 'life1, 'async_trait>(
        &'life0 self,
        batch_ids: &'life1 [BatchId],
    ) -> Pin<Box<dyn Future<Output = Result<Vec<BatchId>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn cancel_batch<'life0, 'async_trait>(
        &'life0 self,
        batch_id: BatchId,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn cascade_batch_state_to_requests<'life0, 'async_trait>(
        &'life0 self,
        batch_id: BatchId,
        target_state: CascadeTargetState,
    ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn delete_batch<'life0, 'async_trait>(
        &'life0 self,
        batch_id: BatchId,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn delete_request<'life0, 'async_trait>(
        &'life0 self,
        request_id: RequestId,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn bulk_delete_data<'life0, 'life1, 'async_trait>(
        &'life0 self,
        creator_id: &'life1 str,
        batch_size: i64,
    ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn retry_failed_requests<'life0, 'async_trait>(
        &'life0 self,
        ids: Vec<RequestId>,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<Result<()>>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn retry_failed_requests_for_batch<'life0, 'async_trait>(
        &'life0 self,
        batch_id: BatchId,
    ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn get_pending_request_counts_by_model_and_window<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>(
        &'life0 self,
        windows: &'life1 [(String, Option<i64>, i64)],
        states: &'life2 [String],
        model_filter: &'life3 [String],
        service_tier_filter: &'life4 ServiceTierFilter,
        priority_decay_window: Option<i64>,
        strict: bool,
    ) -> Pin<Box<dyn Future<Output = Result<HashMap<String, HashMap<String, i64>>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait,
             'life3: 'async_trait,
             'life4: 'async_trait;
    fn sum_owner_batch_requests_in_window<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        owner: &'life1 str,
        completion_window: &'life2 str,
        cutoff: DateTime<Utc>,
        strict: bool,
    ) -> Pin<Box<dyn Future<Output = Result<i64>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait;
    fn count_owner_flex_requests_since<'life0, 'life1, 'async_trait>(
        &'life0 self,
        owner: &'life1 str,
        cutoff: DateTime<Utc>,
        strict: bool,
    ) -> Pin<Box<dyn Future<Output = Result<i64>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn get_requests<'life0, 'async_trait>(
        &'life0 self,
        ids: Vec<RequestId>,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<Result<AnyRequest>>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn claim_requests<'life0, 'life1, 'life2, 'life3, 'async_trait>(
        &'life0 self,
        limit: usize,
        daemon_id: DaemonId,
        available_capacity: &'life1 HashMap<String, usize>,
        user_active_counts: &'life2 HashMap<String, usize>,
        leak_cooldown: &'life3 HashSet<(String, String, String)>,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<Request<Claimed>>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait,
             'life3: 'async_trait;
    fn append_model_filter_event<'life0, 'life1, 'async_trait>(
        &'life0 self,
        entry: &'life1 ModelFilter,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn append_model_filter_events<'life0, 'life1, 'async_trait>(
        &'life0 self,
        entries: &'life1 [ModelFilter],
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn list_model_filters<'life0, 'async_trait>(
        &'life0 self,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<ModelFilter>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn current_filter_states<'life0, 'async_trait>(
        &'life0 self,
    ) -> Pin<Box<dyn Future<Output = Result<HashMap<String, (ModelFilterState, DateTime<Utc>)>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn persist<'life0, 'life1, 'async_trait, T>(
        &'life0 self,
        request: &'life1 Request<T>,
    ) -> Pin<Box<dyn Future<Output = Result<Option<RequestId>>> + Send + 'async_trait>>
       where AnyRequest: From<Request<T>>,
             T: 'async_trait + RequestState + Clone,
             Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn reschedule_for_retry<'life0, 'async_trait>(
        &'life0 self,
        request_id: RequestId,
        owner: DaemonId,
        retry_attempt: u32,
        not_before: Option<DateTime<Utc>>,
    ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn list_requests<'life0, 'async_trait>(
        &'life0 self,
        filter: ListRequestsFilter,
    ) -> Pin<Box<dyn Future<Output = Result<RequestListResult>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn get_request_detail<'life0, 'async_trait>(
        &'life0 self,
        request_id: RequestId,
    ) -> Pin<Box<dyn Future<Output = Result<RequestDetail>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn create_realtime<'life0, 'async_trait>(
        &'life0 self,
        input: CreateRealtimeInput,
    ) -> Pin<Box<dyn Future<Output = Result<RequestId>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn create_flex<'life0, 'async_trait>(
        &'life0 self,
        input: CreateFlexInput,
    ) -> Pin<Box<dyn Future<Output = Result<RequestId>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn complete_request<'life0, 'life1, 'async_trait>(
        &'life0 self,
        request_id: RequestId,
        response_body: &'life1 str,
        status_code: u16,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn fail_request<'life0, 'life1, 'async_trait>(
        &'life0 self,
        request_id: RequestId,
        error: &'life1 str,
        status_code: u16,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn persist_completed_realtime_batch<'life0, 'life1, 'async_trait>(
        &'life0 self,
        records: &'life1 [PersistCompletedRealtimeInput],
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;

    // Provided methods
    fn cancel_requests<'life0, 'async_trait>(
        &'life0 self,
        ids: Vec<RequestId>,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<Result<()>>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait { ... }
    fn claim_batchless_requests<'life0, 'life1, 'life2, 'life3, 'async_trait>(
        &'life0 self,
        limit: usize,
        daemon_id: DaemonId,
        available_capacity: &'life1 HashMap<String, usize>,
        user_active_counts: &'life2 HashMap<String, usize>,
        leak_cooldown: &'life3 HashSet<(String, String, String)>,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<Request<Claimed>>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait,
             'life3: 'async_trait { ... }
    fn claim_batch_requests<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        limit: usize,
        batch_limit: usize,
        daemon_id: DaemonId,
        available_capacity: &'life1 HashMap<String, usize>,
        user_active_counts: &'life2 HashMap<String, usize>,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<Request<Claimed>>>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait { ... }
    fn supports_batch_claims(&self) -> bool { ... }
}

Storage trait for persisting and querying requests.

This trait provides atomic operations for request lifecycle management. The type system ensures valid state transitions, so implementations don’t need to validate them.

Required Methods

fn create_file<'life0, 'async_trait>( &'life0 self, name: String, description: Option<String>, templates: Vec<RequestTemplateInput>, ) -> Pin<Box<dyn Future<Output = Result<FileId>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Create a new file with templates.

fn create_file_stream<'life0, 'async_trait, S>( &'life0 self, stream: S, ) -> Pin<Box<dyn Future<Output = Result<FileStreamResult>> + Send + 'async_trait>>

where S: 'async_trait + Stream<Item = FileStreamItem> + Send + Unpin, Self: 'async_trait, 'life0: 'async_trait,

Create a new file with templates from a stream.

The stream yields FileStreamItem which can be either:

• Metadata: File metadata (can appear anywhere, will be accumulated)
• Template: Request templates (processed as they arrive)
• Abort: Producer initiated rollback without treating it as a fusillade error

fn get_file<'life0, 'async_trait>( &'life0 self, file_id: FileId, ) -> Pin<Box<dyn Future<Output = Result<File>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Get a file by ID.

fn get_file_from_primary_pool<'life0, 'async_trait>( &'life0 self, file_id: FileId, ) -> Pin<Box<dyn Future<Output = Result<File>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Get a file by ID from the primary pool for read-after-write consistency.

Use this immediately after creating or modifying a file to ensure you read the latest committed data. For normal reads, use get_file() which may use read replicas for better performance.

fn list_files<'life0, 'async_trait>( &'life0 self, filter: FileFilter, ) -> Pin<Box<dyn Future<Output = Result<Vec<File>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

List files with optional filtering.

fn get_file_content<'life0, 'async_trait>( &'life0 self, file_id: FileId, ) -> Pin<Box<dyn Future<Output = Result<Vec<FileContentItem>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Get all content for a file.

fn get_file_content_stream( &self, file_id: FileId, offset: usize, search: Option<String>, ) -> Pin<Box<dyn Stream<Item = Result<FileContentItem>> + Send>>

Stream file content. Returns different content types based on the file’s purpose:

• Regular files (purpose=‘batch’): RequestTemplateInput
• Batch output files (purpose=‘batch_output’): BatchOutputItem
• Batch error files (purpose=‘batch_error’): BatchErrorItem

Arguments

• file_id - The file ID to stream content from
• offset - Number of lines to skip (0-indexed)
• search - Optional filter by custom_id (case-insensitive substring match)

fn get_file_template_stats<'life0, 'async_trait>( &'life0 self, file_id: FileId, ) -> Pin<Box<dyn Future<Output = Result<Vec<ModelTemplateStats>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Get aggregated statistics for request templates grouped by model. This is optimized for cost estimation - it only fetches model names and body sizes, avoiding the overhead of streaming full template data.

Returns a vector of per-model statistics including request count and total body bytes.

fn delete_file<'life0, 'async_trait>( &'life0 self, file_id: FileId, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Delete a file (cascades to batches and executions).

fn create_batch<'life0, 'async_trait>( &'life0 self, input: BatchInput, ) -> Pin<Box<dyn Future<Output = Result<Batch>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Create a batch from a file’s current templates.

Convenience method that calls [create_batch_record] to insert the batch row, then [populate_batch] to copy templates into requests. Returns the fully-populated batch.

fn create_batch_record<'life0, 'async_trait>( &'life0 self, input: BatchInput, ) -> Pin<Box<dyn Future<Output = Result<Batch>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Create a batch record with virtual output/error files, without populating requests.

Inserts the batch row and creates virtual output/error files so their IDs are available in the API response immediately. Returns a batch in "validating" status (requests_started_at is NULL). total_requests will be set from input.total_requests if provided, or 0 otherwise. Use [populate_batch] to copy templates into requests afterward.

fn populate_batch<'life0, 'async_trait>( &'life0 self, batch_id: BatchId, file_id: FileId, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Populate an existing batch with requests from its file’s templates.

Copies templates into the requests table and updates the batch with total_requests and requests_started_at. If the file has no templates, returns a ValidationError and the caller is responsible for marking the batch as failed.

fn get_batch<'life0, 'async_trait>( &'life0 self, batch_id: BatchId, ) -> Pin<Box<dyn Future<Output = Result<Batch>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Get a batch by ID.

Arguments

• batch_id - The batch ID to retrieve

fn get_batch_status<'life0, 'async_trait>( &'life0 self, batch_id: BatchId, ) -> Pin<Box<dyn Future<Output = Result<BatchStatus>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Get batch status.

Arguments

• batch_id - The batch ID to retrieve status for

fn list_file_batches<'life0, 'async_trait>( &'life0 self, file_id: FileId, ) -> Pin<Box<dyn Future<Output = Result<Vec<BatchStatus>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

List all batches for a file.

Arguments

• file_id - The file ID to list batches for

fn list_batches<'life0, 'async_trait>( &'life0 self, filter: ListBatchesFilter, ) -> Pin<Box<dyn Future<Output = Result<Vec<Batch>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

List batches with optional filtering and cursor-based pagination. Returns batches sorted by created_at DESC (or active-first when active_first is set).

See ListBatchesFilter for available filter options including:

• created_by - Filter by batch creator user ID
• search - Case-insensitive substring match against metadata JSON text, input filename, or batch ID
• after / limit - Cursor-based pagination (limit defaults to 100 if not set)
• api_key_ids - Filter by API key UUID(s) that created the batch (for per-member attribution)
• status - Filter by batch status. Supported values: "in_progress", "completed", "failed", "cancelled", "expired". "in_progress" covers all non-terminal batches (including validating and finalizing sub-states). "cancelled" includes batches that are still cancelling. "expired" matches batches with SLA issues: in-progress past their deadline, or terminal batches that finished after their deadline. Unrecognized values return an error.
• created_after / created_before - Time range filter on batch creation timestamp
• active_first - When true, sorts active batches before terminal ones (completed, failed, cancelled, or cancelling), with each group sorted by created_at DESC. Cancelling batches are terminal because cancel_batch sets both timestamps atomically. Cursor pagination respects this ordering.

fn get_batch_by_output_file_id<'life0, 'async_trait>( &'life0 self, file_id: FileId, file_type: OutputFileType, ) -> Pin<Box<dyn Future<Output = Result<Option<Batch>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Get a batch by its output or error file ID.

fn get_batch_requests<'life0, 'async_trait>( &'life0 self, batch_id: BatchId, ) -> Pin<Box<dyn Future<Output = Result<Vec<AnyRequest>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Get all requests for a batch.

fn get_batch_results_stream( &self, batch_id: BatchId, offset: usize, search: Option<String>, status: Option<String>, ) -> Pin<Box<dyn Stream<Item = Result<BatchResultItem>> + Send>>

Stream batch results with merged input/output data.

Returns a stream of BatchResultItem, each containing:

• The original input body from the request template
• The response body (for completed requests)
• The error message (for failed requests)
• The current status

Arguments

• batch_id - The batch to get results for
• offset - Number of results to skip (for pagination)
• search - Optional custom_id filter (case-insensitive substring match)
• status - Optional status filter (completed, failed, pending, in_progress)

fn get_cancelled_batch_ids<'life0, 'life1, 'async_trait>( &'life0 self, batch_ids: &'life1 [BatchId], ) -> Pin<Box<dyn Future<Output = Result<Vec<BatchId>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Given a list of batch IDs, return those that have been cancelled (cancelling_at IS NOT NULL).

fn cancel_batch<'life0, 'async_trait>( &'life0 self, batch_id: BatchId, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Cancel all pending/in-progress requests for a batch.

fn cascade_batch_state_to_requests<'life0, 'async_trait>( &'life0 self, batch_id: BatchId, target_state: CascadeTargetState, ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Transition in-flight child requests (pending, claimed, processing) to a terminal state after a batch has been cancelled, failed, or expired.

Intended to be called asynchronously by the caller after the batch has already reached a terminal state. Requests already in a terminal state (completed, failed, canceled) are left untouched.

Returns the number of rows updated.

fn delete_batch<'life0, 'async_trait>( &'life0 self, batch_id: BatchId, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Soft-delete a batch by setting deleted_at.

The batch row is marked deleted and (if not already terminal) cancelled in the same UPDATE. Child requests and their templates are not touched inline — they are hidden from active views via the deleted_at filter and hard-deleted asynchronously by the orphan-purge daemon (see purge_orphaned_rows) for right-to-erasure compliance.

fn delete_request<'life0, 'async_trait>( &'life0 self, request_id: RequestId, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Hard-delete a single request row for right-to-erasure compliance.

Removes the requests row and, if its template is batchless (file_id IS NULL, dedicated 1:1 to this request), the request_templates row as well — batchless templates carry the prompt body, so leaving them defeats the erasure. File-backed templates (shared across siblings in a batch) are not touched here; the orphan-purge daemon cleans those up after the parent file is soft-deleted.

FK behavior on the deleted requests row:

• response_steps.request_id → ON DELETE CASCADE: removes only the step row(s) whose request_id matches this request. After migration 20260430000000 (response_steps re-anchoring), each step points at its own per-step sub-request fusillade row, so deleting one request only cascade-removes that step. Callers wanting to erase a whole multi-step response chain must walk the chain and call this method for each backing request.
• Self-references escalated_from_request_id / superseded_by_request_id → ON DELETE SET NULL, so sibling rows lose their pointer cleanly.

In-flight handling: this is an unconditional hard delete. A daemon mid- update on the row sees 0 rows affected on its next write; a streaming proxy mid-INSERT of response_steps FK-violates (logged, not corrupted). Both are acceptable for explicit user-initiated erasure.

Unlike Self::delete_batch (soft-delete + async purge), this is immediate because the caller has resolved a specific request to erase.

Returns RequestNotFound if the request does not exist (or was already deleted).

fn bulk_delete_data<'life0, 'life1, 'async_trait>( &'life0 self, creator_id: &'life1 str, batch_size: i64, ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Erase all of a creator’s fusillade data, for right-to-erasure compliance.

Processes up to batch_size rows per category per call, using FOR UPDATE SKIP LOCKED so it is safe to run concurrently and under load. Returns the count of top-level rows processed this call — batchless requests deleted plus batches and files soft-deleted. It does NOT include the batchless templates removed alongside those requests, nor the batch/file child rows the purge daemon reaps later. It is purely a loop-termination signal: callers should loop until it returns 0 to drain everything. A batch_size < 1 returns 0 (nothing to do). Idempotent.

Three categories, keyed on created_by / uploaded_by = creator_id:

• Batchless requests (batch_id IS NULL — realtime/flex) are hard-deleted along with their batchless request_templates (which carry the prompt body). The orphan-purge daemon never reaches these because they have no soft-deleted parent batch, so they must be removed here or the erasure is incomplete.
• Batches are soft-deleted (cancelled if active) with metadata nullified (it can contain the user’s email). Their child requests are hard-deleted afterwards by the orphan-purge daemon (purge_orphaned_rows).
• Files are soft-deleted; their request_templates are likewise reaped by the orphan-purge daemon once files.deleted_at is set.

Note: completion is therefore eventually-consistent — when this returns 0, all batches/files are soft-deleted and batchless rows are gone, but batch/file child rows are erased on the next purge-daemon pass.

fn retry_failed_requests<'life0, 'async_trait>( &'life0 self, ids: Vec<RequestId>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Result<()>>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Retry failed requests by resetting them to pending state.

This resets the specified failed requests to pending state with retry_attempt = 0, allowing them to be picked up by the daemon for reprocessing.

Arguments

• ids - Request IDs to retry

Returns

A vector of results, one for each request ID. Each result indicates whether the retry succeeded or failed.

Errors

Individual retry results may fail if:

• Request ID doesn’t exist
• Request is not in failed state

fn retry_failed_requests_for_batch<'life0, 'async_trait>( &'life0 self, batch_id: BatchId, ) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Retry all failed requests for a batch in a single database operation.

This is much more efficient than retry_failed_requests when retrying all failed requests for a batch, as it performs a single UPDATE query instead of loading all requests into memory.

Returns

The number of requests that were retried.

fn get_pending_request_counts_by_model_and_window<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, windows: &'life1 [(String, Option<i64>, i64)], states: &'life2 [String], model_filter: &'life3 [String], service_tier_filter: &'life4 ServiceTierFilter, priority_decay_window: Option<i64>, strict: bool, ) -> Pin<Box<dyn Future<Output = Result<HashMap<String, HashMap<String, i64>>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait,

Get request counts grouped by model and deadline window.

Each window is the half-open interval [now + start_secs, now + end_secs) applied to each request’s deadline. A request is counted in a window if its deadline falls inside that range. Because the end is exclusive, adjacent windows (e.g. (_, Some(0), 3600) and (_, Some(3600), 86400)) never double-count a request sitting on the boundary.

A request’s deadline is its batch’s expires_at. Batchless rows (flex/async responses, batch_id IS NULL) have no batch expiry, so their deadline is synthesized as created_at + W, where W is mapped from the row’s service_tier via DaemonConfig.service_tier_completion_windows_ms ('flex' → 1h by default, NULL/unmapped → default_completion_window_ms, 24h) — the same window the claim path uses, so reported queue depth matches what the daemon will claim.

start_secs is optional. When None, the lower bound is unbounded (the query matches every request with a deadline strictly before now + end_secs, including overdue ones). Callers that want the legacy “due within N, including overdue” semantics pass (label, None, N). Callers that specifically want the “future N seconds” starting at now pass (label, Some(0), N).

• windows: Vec of (label, start_secs, end_secs). When start_secs is Some(s), s must be <= end_secs.
• states: request states to include (e.g. ["pending"], or ["pending","claimed","processing"]).
• model_filter: optional model whitelist (empty = all).
• service_tier_filter: filter on service_tier. Any (default) applies no filter; Include/Exclude use Option<String> where None represents the batch tier (service_tier IS NULL).
• priority_decay_window: optional lookback in seconds. When set, recently completed service_tier = 'flex' requests are added to the "1h" bucket so realtime traffic can decay out of scheduling pressure after successful completion. No effect if the requested windows do not include a "1h" label.
• strict: bool. For critical/sensitive operations, set true to use the write pool and avoid read lags.

Excludes:

• Requests without a template_id
• Requests in batches being cancelled

fn sum_owner_batch_requests_in_window<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, owner: &'life1 str, completion_window: &'life2 str, cutoff: DateTime<Utc>, strict: bool, ) -> Pin<Box<dyn Future<Output = Result<i64>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Sum the total_requests of a creditor’s batches for a given completion window created on or after cutoff.

Used by the control layer to enforce the unverified upload-volume cap at batch creation: an unverified creditor may submit at most unverified_requests_per_completion_hour * window_hours requests within a rolling window equal to the completion window. Served by idx_batches_completion_window (completion_window, created_by).

• owner: the batch created_by — the creditor (organization id for org members, user id otherwise).
• cutoff: only batches with created_at >= cutoff are counted.
• strict: set true to read from the write pool and avoid read lag, so a just-created batch is reflected immediately (required for enforcement).

fn count_owner_flex_requests_since<'life0, 'life1, 'async_trait>( &'life0 self, owner: &'life1 str, cutoff: DateTime<Utc>, strict: bool, ) -> Pin<Box<dyn Future<Output = Result<i64>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Count a creditor’s batchless flex requests created on or after cutoff.

The flex counterpart of Storage::sum_owner_batch_requests_in_window: flex requests are batchless (batch_id IS NULL, attribution via requests.created_by) and always map to the 1h completion window. Served by idx_requests_user_created_sort (created_by, created_at DESC, id DESC, service_tier) WHERE created_by IS NOT NULL.

• owner: the request created_by — the creditor id.
• cutoff: only requests with created_at >= cutoff are counted.
• strict: set true to read from the write pool and avoid read lag.

fn get_requests<'life0, 'async_trait>( &'life0 self, ids: Vec<RequestId>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Result<AnyRequest>>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Get in progress requests by IDs.

fn claim_requests<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, limit: usize, daemon_id: DaemonId, available_capacity: &'life1 HashMap<String, usize>, user_active_counts: &'life2 HashMap<String, usize>, leak_cooldown: &'life3 HashSet<(String, String, String)>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Request<Claimed>>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Compatibility method for callers and storage implementations that have not yet moved to the explicit request daemon API.

New daemon code should call Storage::claim_batchless_requests or Storage::claim_batch_requests directly. This method is kept as a batchless-only alias so the request and batch policies cannot be accidentally recombined.

fn append_model_filter_event<'life0, 'life1, 'async_trait>( &'life0 self, entry: &'life1 ModelFilter, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Append a single event to the model_filters log. Used by the controller when a model’s internal liveness CHANGES (live / coming / absent).

The gate reads only state (live ⇒ claim full; coming/absent ⇒ not-live). expected_ready_at is retained on the type/column for the controller’s own use but is not read by the claim gate — callers may leave it None.

This is append-only: there is no delete and no upsert. Retraction is appending an Absent event. Appending only on change (so the log stays a transition log rather than a poll log) is the caller’s responsibility — this function always inserts a row.

fn append_model_filter_events<'life0, 'life1, 'async_trait>( &'life0 self, entries: &'life1 [ModelFilter], ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Append a batch of events to the model_filters log (one row each, in order). Convenience for the controller publishing several transitions in one sync. Same append-only / append-on-change semantics as Storage::append_model_filter_event.

fn list_model_filters<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<ModelFilter>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

List the CURRENT state of every model (the latest event per model), excluding models whose latest event is an Absent tombstone (observability / tests).

fn current_filter_states<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<HashMap<String, (ModelFilterState, DateTime<Utc>)>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

The CURRENT state of every model and when that state began: the latest event per model as (state, since), where since is that event’s created_at. Includes Absent (a model’s latest event may be a tombstone); the caller decides what to do with each state.

This is the controller’s read for time-based decisions off the log — a Live model’s since is when it went live (minimum-lifetime / anti-thrash), a Coming model’s since is when it started launching (a stuck-coming watchdog). The timestamp comes from the persisted log, not in-memory state, so it survives controller restarts.

fn persist<'life0, 'life1, 'async_trait, T>( &'life0 self, request: &'life1 Request<T>, ) -> Pin<Box<dyn Future<Output = Result<Option<RequestId>>> + Send + 'async_trait>>

where AnyRequest: From<Request<T>>, T: 'async_trait + RequestState + Clone, Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Update an existing request’s state in storage.

Returns Some(request_id) if a racing pair was superseded (for cancellation purposes).

fn reschedule_for_retry<'life0, 'async_trait>( &'life0 self, request_id: RequestId, owner: DaemonId, retry_attempt: u32, not_before: Option<DateTime<Utc>>, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Reschedule an in-flight request back to pending for an automatic retry, fenced on the worker that currently owns it.

This is the daemon’s per-attempt retry path. Unlike Storage::persist (which matches on id only, so the manual retry path can intentionally resurrect a failed row), this transition is guarded by state = 'processing' AND daemon_id = <owner>: it applies ONLY if the row is still the in-flight claim held by daemon_id.

The guard prevents a finalize-then-resurrect race: if another writer (a zombie/duplicate worker, or a stale-claim reclaim) has already moved the row to a terminal state — and a finalizer has sealed the parent batch as a result — a late retry from this worker must NOT flip it back to pending, orphaning it under a completed batch.

Returns true if the row was rescheduled, false if the worker no longer owns it (lost the race). A false result is normal under contention and should be logged, not treated as an error.

fn list_requests<'life0, 'async_trait>( &'life0 self, filter: ListRequestsFilter, ) -> Pin<Box<dyn Future<Output = Result<RequestListResult>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

List individual requests across batches with filtering and pagination.

Supports filtering by creator, completion window, status, model(s), date range, and active-first sorting. Uses offset-based pagination.

Note: Token and cost metrics are NOT included — callers should join against their own analytics tables for that data.

fn get_request_detail<'life0, 'async_trait>( &'life0 self, request_id: RequestId, ) -> Pin<Box<dyn Future<Output = Result<RequestDetail>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Get a single request by ID with full detail (body, response, error).

fn create_realtime<'life0, 'async_trait>( &'life0 self, input: CreateRealtimeInput, ) -> Pin<Box<dyn Future<Output = Result<RequestId>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Create a realtime response that the proxy is already handling.

Inserts a request template (no parent file) and a request row with batch_id = NULL in processing state. The proxy completes/fails the row directly via complete_request / fail_request; the daemon never claims it.

fn create_flex<'life0, 'async_trait>( &'life0 self, input: CreateFlexInput, ) -> Pin<Box<dyn Future<Output = Result<RequestId>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Create a flex (async) response that the daemon will process.

Inserts a request template (no parent file) and a request row with batch_id = NULL in pending state. The daemon claims and processes it via the standard flex pipeline.

fn complete_request<'life0, 'life1, 'async_trait>( &'life0 self, request_id: RequestId, response_body: &'life1 str, status_code: u16, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Complete a processing request with the response body.

Transitions the request from “processing” to “completed” and stores the response body and HTTP status code.

fn fail_request<'life0, 'life1, 'async_trait>( &'life0 self, request_id: RequestId, error: &'life1 str, status_code: u16, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Fail a processing request with an error message and HTTP status code.

Transitions the request from “processing” to “failed” and stores the error as a NonRetriableHttpStatus JSON object with the given status code.

fn persist_completed_realtime_batch<'life0, 'life1, 'async_trait>( &'life0 self, records: &'life1 [PersistCompletedRealtimeInput], ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Persist a batch of already-completed realtime responses in one transaction.

Designed for the dwctl responses writer: dwctl proxies a realtime request, captures the upstream response, and flushes a buffer of completed records here. Two cases are handled together:

• Background realtime: a processing row exists (created inline by create_realtime before the 202 response). UPDATEd to completed.
• Non-background realtime: no row exists. INSERTed (template + request) directly in completed state.

Rows already in a terminal state (rare: duplicate enqueues, late completions for flex slip-through) are left alone via ON CONFLICT.

All work runs in a single transaction so commit overhead amortises across the batch. An empty input is a no-op.

Provided Methods

fn cancel_requests<'life0, 'async_trait>( &'life0 self, ids: Vec<RequestId>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Result<()>>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Cancel one or more individual pending or in-progress requests.

Requests that have already completed or failed cannot be canceled. This is a best-effort operation - some requests may have already been processed.

Returns a result for each request ID indicating whether cancellation succeeded.

Errors

Individual cancellation results may fail if:

• Request ID doesn’t exist
• Request is already in a terminal state (completed/failed)

fn claim_batchless_requests<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, limit: usize, daemon_id: DaemonId, available_capacity: &'life1 HashMap<String, usize>, user_active_counts: &'life2 HashMap<String, usize>, leak_cooldown: &'life3 HashSet<(String, String, String)>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Request<Claimed>>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Atomically claim pending batchless requests for processing.

available_capacity maps model names to the number of permits the daemon is currently holding for that model. Only models present in this map will be claimed — this is the authoritative set of models to process.

user_active_counts maps user identifiers to their current number of in-flight requests across all models. Used to prioritise users with fewer active requests for per-user fair scheduling. Pass an empty map to disable user-level prioritisation (falls back to deadline-only ordering).

Implementations may blend user-fairness with SLA urgency (batch deadline proximity) via DaemonConfig::urgency_weight. See the PostgreSQL implementation for the composite scoring formula.

The claim gate consults the latest model_filters event per model: state = 'live' or no events at all ⇒ claim at full capacity (a model with no events is unmanaged by the controller, so there is no internal capacity to wait for — it flows straight through to OpenRouter). An EXPLICIT not-live event (coming/absent) ⇒ the request is either claimed at full capacity (→ OpenRouter) when within ramp(W) of its completion-window deadline, or otherwise released only via the per-(user, window-class, model) leaky bucket. So with an empty model_filters table the gate is a no-op (everything claims at full capacity) — it only engages once the controller starts writing not-live events.

leak_cooldown is the set of (user, window-class, model) triples whose leaky bucket has no token this cycle (the daemon stamped next_token_at in the future after a recent leak). Source B skips these triples, claiming ≤ 1 per (user, window-class, model) not in cooldown. Pass an empty set to allow every bucket its first token. Claimed rows carry a leaked flag (via the returned request) so the daemon knows which buckets to stamp.

fn claim_batch_requests<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, limit: usize, batch_limit: usize, daemon_id: DaemonId, available_capacity: &'life1 HashMap<String, usize>, user_active_counts: &'life2 HashMap<String, usize>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Request<Claimed>>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Atomically claim pending requests that belong to live-model batches.

The batch daemon owns this policy. Implementations should select candidate batches before probing request rows, limit selected batches by batch_limit, and gate on model liveness: models whose latest model_filters event is live are always eligible; models with no filter event (external / always-on providers that scouter does not manage) are eligible unless DaemonConfig::batch_claim_require_live is set; models whose latest event is coming/absent are eligible only once the batch is within the deadline ramp (claim_ramp_exponent) — the SLA escape hatch to fallback providers. No leaky-bucket trickle applies to batched rows.

fn supports_batch_claims(&self) -> bool

Whether this backend implements Storage::claim_batch_requests.

The daemon only spawns its batch claim loop when this returns true. Defaults to true so a backend that forgets to override BOTH methods fails loudly (the default claim_batch_requests errors) instead of silently never claiming batched rows. A deliberately request-only backend should override this to return false.

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors