newton-task-submission 0.7.2

//! Transport-neutral task-submission handlers and their narrow ports.

use crate::{
    derive_idempotency_key, EventCursor, SubmissionEventPage, SubmissionResource, TaskOperation, TaskSubmissionRequest,
};
use std::{collections::HashMap, error::Error, future::Future, sync::Arc, time::Duration};

/// Task-domain policy for one configured chain.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TaskChainPolicy {
    /// Target EVM chain.
    pub chain_id: u64,
    /// Maximum task effects in one execution intent.
    pub max_batch_size: usize,
    /// Maximum admitted task age before planning fails it.
    pub max_task_age_secs: u64,
    /// Maximum number of effect-level retries.
    pub max_effect_retries: u32,
}

/// Durable planner timing policy.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct TaskPlannerConfig {
    /// Maximum batching delay in milliseconds.
    pub batch_interval_ms: u64,
    /// Reconciliation polling interval in milliseconds.
    pub poll_interval_ms: u64,
}

impl TaskPlannerConfig {
    /// Returns the bounded reconciliation interval.
    pub fn poll_interval(&self) -> Duration {
        Duration::from_millis(self.poll_interval_ms.max(10))
    }
}

/// Result of idempotent task admission.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AdmissionOutcome {
    /// A new task submission was committed.
    Accepted(SubmissionResource),
    /// The same immutable request was already committed.
    Existing(SubmissionResource),
    /// The idempotency key is bound to different immutable bytes.
    Conflict(SubmissionResource),
}

/// Persistence operations required by synchronous task handlers.
pub trait TaskSubmissionStore: std::fmt::Debug + Send + Sync {
    /// Persistence failure.
    type Error: Error + Send + Sync + 'static;

    /// Idempotently admits a validated task submission.
    fn admit(
        &self,
        request: &TaskSubmissionRequest,
        policy: &TaskChainPolicy,
    ) -> impl Future<Output = Result<AdmissionOutcome, Self::Error>> + Send;

    /// Loads one task-submission resource.
    fn submission(
        &self,
        id: crate::SubmissionId,
    ) -> impl Future<Output = Result<Option<SubmissionResource>, Self::Error>> + Send;

    /// Reads or waits for a bounded page of replayable task events.
    fn events(
        &self,
        after: EventCursor,
        limit: u32,
        wait: Duration,
    ) -> impl Future<Output = Result<SubmissionEventPage, Self::Error>> + Send;
}

/// Planner progress and bounded telemetry owned by the composition root.
pub trait TaskPlannerObserver: std::fmt::Debug + Send + Sync {
    /// Records planner liveness.
    fn heartbeat(&self);
    /// Records the latest planner failure.
    fn failed(&self, error: String);
    /// Records one newly persisted task plan.
    fn planned(&self, chain_id: u64, operation: TaskOperation);
}

/// Application-level task handlers independent of HTTP and persistence technology.
#[derive(Debug)]
pub struct TaskSubmissionService<S> {
    store: Arc<S>,
    chains: HashMap<u64, TaskChainPolicy>,
}

impl<S> TaskSubmissionService<S>
where
    S: TaskSubmissionStore,
{
    /// Creates transport-neutral handlers for configured task chains.
    pub fn new(store: Arc<S>, chains: HashMap<u64, TaskChainPolicy>) -> Self {
        Self { store, chains }
    }

    /// Returns configured task-chain policies.
    pub fn chains(&self) -> &HashMap<u64, TaskChainPolicy> {
        &self.chains
    }

    /// Validates and idempotently admits a task request.
    pub async fn admit(
        &self,
        request: &TaskSubmissionRequest,
    ) -> Result<AdmissionOutcome, TaskSubmissionError<S::Error>> {
        request
            .validate()
            .map_err(|error| TaskSubmissionError::Invalid(error.to_string()))?;
        let policy = self
            .chains
            .get(&request.payload.chain_id)
            .ok_or(TaskSubmissionError::UnsupportedChain)?;
        let expected_key = derive_idempotency_key(
            request.producer_id.as_bytes(),
            policy.chain_id,
            request.payload.operation,
            request.payload.task_id,
            request.payload.task_response_digest,
        );
        if request.idempotency_key != expected_key {
            return Err(TaskSubmissionError::Invalid(
                "idempotency key does not match the producer and payload".to_string(),
            ));
        }
        self.store
            .admit(request, policy)
            .await
            .map_err(TaskSubmissionError::Store)
    }

    /// Returns the current task-submission resource when present.
    pub async fn status(
        &self,
        id: crate::SubmissionId,
    ) -> Result<Option<SubmissionResource>, TaskSubmissionError<S::Error>> {
        self.store.submission(id).await.map_err(TaskSubmissionError::Store)
    }

    /// Returns a bounded replay page after the supplied cursor.
    pub async fn events(
        &self,
        after: EventCursor,
        limit: u32,
        wait: Duration,
    ) -> Result<SubmissionEventPage, TaskSubmissionError<S::Error>> {
        self.store
            .events(after, limit, wait)
            .await
            .map_err(TaskSubmissionError::Store)
    }
}

/// Failure returned by transport-neutral task handlers.
#[derive(Debug, thiserror::Error)]
pub enum TaskSubmissionError<E: Error + 'static> {
    /// Request validation failed.
    #[error("invalid request: {0}")]
    Invalid(String),
    /// No task policy exists for the requested chain.
    #[error("unsupported chain")]
    UnsupportedChain,
    /// Task persistence failed.
    #[error(transparent)]
    Store(E),
}