newton-task-submission 0.7.2

//! Persistence boundary owned by the task planner.

use crate::{SubmissionId, SubmissionState, TaskExecutionIntent, TaskOperation, TaskSubmissionPayload};
use newton_submission_protocol::{ExecutionId, ExecutionProgress, ExecutionRequest};
use std::{error::Error, future::Future};

/// One durable task request eligible for a future batch.
#[derive(Debug, Clone)]
pub struct PendingTaskRecord {
    /// Stable task-submission identity.
    pub submission_id: SubmissionId,
    /// Contract operation requested for the task.
    pub operation: TaskOperation,
    /// Immutable contract payload.
    pub payload: TaskSubmissionPayload,
    /// Admission time in Unix milliseconds.
    pub accepted_at_ms: i64,
    /// Task deadline in Unix milliseconds.
    pub deadline_at_ms: i64,
    /// Number of prior effect-level retries.
    pub effect_retry_count: u32,
    /// Maximum permitted effect-level retries.
    pub max_effect_retries: u32,
}

/// Durable lifecycle of one immutable task plan around the execution channel.
///
/// The monotonic path is `Planned -> Submitted -> Projected -> Acknowledged`.
/// Plan insertion atomically claims every member. Execution-channel submission
/// occurs between the first two states and is idempotent by `plan_id`.
/// `Submitted -> Projected` atomically updates every task member from one
/// terminal common outcome. The final transition happens only after common
/// acknowledgement; retention requires both sides, so either acknowledgement
/// write may be safely retried after a crash.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TaskPlanState {
    /// Active and producer-owned; all members and intent bytes are durable but
    /// the common channel has not yet been recorded as owner.
    Planned,
    /// Active; the common channel owns signer selection, transaction retry,
    /// recovery, and terminal effect classification.
    Submitted,
    /// Active acknowledgement phase; the terminal outcome was atomically
    /// projected to every member, including any respond-only requeues.
    Projected,
    /// Terminal; common and task-domain acknowledgement are durable and the
    /// plan no longer participates in planning.
    Acknowledged,
}

impl TaskPlanState {
    /// Stable persistence representation.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Planned => "planned",
            Self::Submitted => "submitted",
            Self::Projected => "projected",
            Self::Acknowledged => "acknowledged",
        }
    }
}

/// Per-request retry metadata frozen into one task plan snapshot.
#[derive(Debug, Clone)]
pub struct TaskPlanMemberRecord {
    /// Task submission represented by this member.
    pub submission_id: SubmissionId,
    /// Effect retry count at planning time.
    pub effect_retry_count: u32,
    /// Effect retry limit at planning time.
    pub max_effect_retries: u32,
}

/// One task-domain aggregate producing one execution request.
#[derive(Debug, Clone)]
pub struct TaskPlanRecord {
    /// Stable identity reused as the execution-channel idempotency key.
    pub plan_id: ExecutionId,
    /// Target EVM chain.
    pub chain_id: u64,
    /// Common operation shared by all plan members.
    pub operation: TaskOperation,
    /// Fully materialized task-domain execution intent.
    pub intent: TaskExecutionIntent,
    /// Earliest task deadline in Unix milliseconds.
    pub deadline_at_ms: Option<i64>,
    /// Durable plan lifecycle state.
    pub state: TaskPlanState,
    /// Ordered task submissions represented by the intent.
    pub members: Vec<TaskPlanMemberRecord>,
}

impl TaskPlanRecord {
    /// Materializes the domain-owned request sent through the execution channel.
    pub fn execution(&self) -> ExecutionRequest<TaskExecutionIntent> {
        ExecutionRequest {
            execution_id: self.plan_id,
            chain_id: self.chain_id,
            intent: self.intent.clone(),
            deadline_at_ms: self.deadline_at_ms,
        }
    }
}

/// Consistent task-planning view loaded from persistence.
#[derive(Debug, Clone, Default)]
pub struct TaskPlanningSnapshot {
    /// Unplanned task submissions eligible for selection.
    pub pending: Vec<PendingTaskRecord>,
    /// Active durable plans requiring reconciliation.
    pub plans: Vec<TaskPlanRecord>,
}

/// Planner-computed terminal transition for one task request.
#[derive(Debug, Clone)]
pub struct TaskProjection {
    /// Task submission receiving the projected transition.
    pub submission_id: SubmissionId,
    /// New task-domain state.
    pub state: SubmissionState,
    /// Operation to retain when the task is requeued.
    pub operation: Option<TaskOperation>,
    /// Whether to increment the task's effect retry count.
    pub increment_effect_retry: bool,
    /// Stable terminal failure reason, when applicable.
    pub terminal_error: Option<String>,
}

/// Mechanical writes accepted by the task planner's persistence adapter.
#[derive(Debug, Clone)]
pub enum TaskPlanningWrite {
    /// Persist a newly selected immutable plan.
    InsertPlan(TaskPlanRecord),
    /// Record successful execution-channel emission.
    MarkSubmitted(ExecutionId),
    /// Project non-terminal execution progress.
    ProjectProgress {
        /// Plan receiving the progress update.
        plan_id: ExecutionId,
        /// Task-domain state corresponding to the execution phase.
        state: SubmissionState,
        /// Latest execution metadata.
        progress: ExecutionProgress,
    },
    /// Project a terminal execution outcome to every plan member.
    ProjectOutcome {
        /// Plan receiving the terminal outcome.
        plan_id: ExecutionId,
        /// Final execution metadata.
        progress: ExecutionProgress,
        /// Planner-computed member transitions.
        projections: Vec<TaskProjection>,
    },
    /// Record successful execution-channel acknowledgement.
    MarkAcknowledged(ExecutionId),
}

/// Result of an optimistic persistence write.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TaskPlanningCommit {
    /// Write was applied atomically.
    Applied,
    /// Expected state changed before the write could be applied.
    Stale,
}

/// Thin persistence slice required by task planning.
pub trait TaskPlanningStore: std::fmt::Debug + Send + Sync {
    /// Persistence failure.
    type Error: Error + Send + Sync + 'static;

    /// Loads a consistent bounded planning snapshot for one chain.
    fn load(
        &self,
        chain_id: u64,
        pending_limit: usize,
        plan_limit: usize,
    ) -> impl Future<Output = Result<TaskPlanningSnapshot, Self::Error>> + Send;

    /// Atomically applies one mechanical planner write.
    fn commit(&self, write: &TaskPlanningWrite)
        -> impl Future<Output = Result<TaskPlanningCommit, Self::Error>> + Send;
}