saddle-runtime 0.2.0-rc.6

//! Linear bridge between a paired startup plan and DB normal-return work.
//!
//! This is an internal component capability. It deliberately contains no
//! duration and never constructs a timer. The caller must poll the request
//! task's existing timer and must synchronously discard the DB return owner
//! before acknowledging an expired budget.

use std::{
    future::Future,
    pin::Pin,
    task::{Context, Poll},
};

use saddle_admission::StartupPublishToken;
use tokio::time::Sleep;

#[doc(hidden)]
pub const DB_RETURN_BUDGET_ISSUER_IDENTITY: [u8; 32] = [0xd6; 32];

/// Database-owned, consuming adapter for its sealed normal-return work proof.
///
/// The release Gate must lock the only production implementation and assembly.
#[doc(hidden)]
pub trait DbReturnWorkProofAdapter: Sized {
    fn work_identity(&self) -> [u8; 32];
    fn adapter_provenance(&self) -> [u8; 32];
}

/// Startup-plan-bound issuer for one-at-a-time request DB return budgets.
#[doc(hidden)]
#[derive(Debug)]
pub struct DbReturnBudgetIssuer {
    plan_digest: [u8; 32],
    artifact_identity: [u8; 32],
    work_identity: [u8; 32],
    adapter_provenance: [u8; 32],
    next_generation: u64,
    active_generation: Option<u64>,
}

/// A single-use budget bound to one plan, work proof and request generation.
#[doc(hidden)]
#[derive(Debug)]
pub struct DbReturnBudget {
    binding: Option<BudgetBinding>,
}

/// Proof that the existing request timer won the return race.
#[doc(hidden)]
#[derive(Debug)]
pub struct DbReturnBudgetExpired {
    binding: BudgetBinding,
}

/// Proof that the DB owner synchronously discarded the pending return owner.
#[doc(hidden)]
#[derive(Debug)]
pub struct DbReturnDiscarded {
    binding: BudgetBinding,
}

/// Completion of the normal return before its budget expired.
#[doc(hidden)]
#[derive(Debug)]
pub struct DbReturnCompleted {
    binding: BudgetBinding,
}

#[derive(Debug)]
struct BudgetBinding {
    plan_digest: [u8; 32],
    artifact_identity: [u8; 32],
    work_identity: [u8; 32],
    adapter_provenance: [u8; 32],
    generation: u64,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[doc(hidden)]
pub enum DbReturnBudgetError {
    ActiveBudget,
    GenerationExhausted,
    ForeignOrStale,
    PlanIssuerMismatch,
}

#[derive(Debug)]
#[doc(hidden)]
pub enum DbReturnPoll {
    Returned(DbReturnCompleted),
    BudgetExpired(DbReturnBudgetExpired),
}

impl DbReturnBudgetIssuer {
    /// Linearly splits the opaque paired-plan token into its continuation and
    /// the only DB return budget issuer for that startup.
    #[doc(hidden)]
    pub fn bind<C, A>(
        publish: StartupPublishToken<C>,
        adapter: A,
    ) -> Result<(Self, C), DbReturnBudgetError>
    where
        A: DbReturnWorkProofAdapter,
    {
        if publish.db_return_issuer_identity() != DB_RETURN_BUDGET_ISSUER_IDENTITY
            || publish.db_return_work_identity() != adapter.work_identity()
        {
            return Err(DbReturnBudgetError::PlanIssuerMismatch);
        }
        let issuer = Self {
            plan_digest: publish.plan_digest(),
            artifact_identity: publish.artifact_identity(),
            work_identity: adapter.work_identity(),
            adapter_provenance: adapter.adapter_provenance(),
            next_generation: 1,
            active_generation: None,
        };
        Ok((issuer, publish.into_continuation()))
    }

    #[doc(hidden)]
    pub fn issue(&mut self) -> Result<DbReturnBudget, DbReturnBudgetError> {
        if self.active_generation.is_some() {
            return Err(DbReturnBudgetError::ActiveBudget);
        }
        let generation = self.next_generation;
        self.next_generation = self
            .next_generation
            .checked_add(1)
            .ok_or(DbReturnBudgetError::GenerationExhausted)?;
        self.active_generation = Some(generation);
        Ok(DbReturnBudget {
            binding: Some(self.binding(generation)),
        })
    }

    #[doc(hidden)]
    pub fn acknowledge_return(
        &mut self,
        completed: DbReturnCompleted,
    ) -> Result<(), DbReturnBudgetError> {
        self.finish(completed.binding)
    }

    /// Converts expiry into a completion receipt only after Database has
    /// synchronously dropped/poison-discarded the pending return owner.
    #[doc(hidden)]
    pub fn acknowledge_discard(
        &mut self,
        discarded: DbReturnDiscarded,
    ) -> Result<(), DbReturnBudgetError> {
        self.finish(discarded.binding)
    }

    fn binding(&self, generation: u64) -> BudgetBinding {
        BudgetBinding {
            plan_digest: self.plan_digest,
            artifact_identity: self.artifact_identity,
            work_identity: self.work_identity,
            adapter_provenance: self.adapter_provenance,
            generation,
        }
    }

    fn finish(&mut self, binding: BudgetBinding) -> Result<(), DbReturnBudgetError> {
        if binding != self.binding(binding.generation)
            || self.active_generation != Some(binding.generation)
        {
            return Err(DbReturnBudgetError::ForeignOrStale);
        }
        self.active_generation = None;
        Ok(())
    }
}

impl DbReturnBudgetExpired {
    /// Consumes and synchronously drops the DB return owner before producing
    /// the only receipt that the issuer accepts.
    #[doc(hidden)]
    pub fn synchronously_discard<R>(self, return_owner: R) -> DbReturnDiscarded {
        drop(return_owner);
        DbReturnDiscarded {
            binding: self.binding,
        }
    }
}

impl DbReturnBudget {
    /// Polls normal return first, then the request task's already-existing
    /// timer. This method allocates nothing and creates no timer or waiter.
    #[doc(hidden)]
    pub fn poll_return<R>(
        self: Pin<&mut Self>,
        return_to_pool: Pin<&mut R>,
        request_timer: Pin<&mut Sleep>,
        context: &mut Context<'_>,
    ) -> Poll<DbReturnPoll>
    where
        R: Future<Output = ()>,
    {
        if return_to_pool.poll(context).is_ready() {
            return Poll::Ready(DbReturnPoll::Returned(DbReturnCompleted {
                binding: self.take_binding(),
            }));
        }
        if request_timer.poll(context).is_ready() {
            return Poll::Ready(DbReturnPoll::BudgetExpired(DbReturnBudgetExpired {
                binding: self.take_binding(),
            }));
        }
        Poll::Pending
    }

    fn take_binding(self: Pin<&mut Self>) -> BudgetBinding {
        // `DbReturnBudget` is structurally Unpin. Repoll after completion is a
        // contract violation and must not manufacture a second receipt.
        let this = self.get_mut();
        this.binding.take().unwrap_or_else(|| std::process::abort())
    }
}

impl PartialEq for BudgetBinding {
    fn eq(&self, other: &Self) -> bool {
        self.plan_digest == other.plan_digest
            && self.artifact_identity == other.artifact_identity
            && self.work_identity == other.work_identity
            && self.adapter_provenance == other.adapter_provenance
            && self.generation == other.generation
    }
}

impl Eq for BudgetBinding {}

#[cfg(test)]
mod tests {
    use std::{
        future::{Future, ready},
        pin::pin,
        sync::{
            Arc,
            atomic::{AtomicBool, Ordering},
        },
        task::Poll,
        time::Duration,
    };

    use super::*;

    struct DropOwner(Arc<AtomicBool>);

    impl Drop for DropOwner {
        fn drop(&mut self) {
            self.0.store(true, Ordering::SeqCst);
        }
    }

    struct PendingReturn {
        _owner: DropOwner,
    }

    impl Future for PendingReturn {
        type Output = ();

        fn poll(self: Pin<&mut Self>, _context: &mut Context<'_>) -> Poll<Self::Output> {
            Poll::Pending
        }
    }

    fn issuer(seed: u8) -> DbReturnBudgetIssuer {
        DbReturnBudgetIssuer {
            plan_digest: [seed; 32],
            artifact_identity: [seed.wrapping_add(1); 32],
            work_identity: [seed.wrapping_add(2); 32],
            adapter_provenance: [seed.wrapping_add(3); 32],
            next_generation: 1,
            active_generation: None,
        }
    }

    #[tokio::test]
    async fn return_wins_before_existing_request_timer() {
        let mut issuer = issuer(1);
        let mut budget = pin!(issuer.issue().unwrap());
        let mut return_to_pool = pin!(ready(()));
        let timer = tokio::time::sleep(Duration::from_secs(60));
        tokio::pin!(timer);

        let outcome = std::future::poll_fn(|context| {
            budget
                .as_mut()
                .poll_return(return_to_pool.as_mut(), timer.as_mut(), context)
        })
        .await;
        let DbReturnPoll::Returned(completed) = outcome else {
            panic!("normal return must win");
        };
        issuer.acknowledge_return(completed).unwrap();
        assert!(issuer.active_generation.is_none());
    }

    #[tokio::test]
    async fn existing_timer_wins_and_owner_is_dropped_before_receipt() {
        let dropped = Arc::new(AtomicBool::new(false));
        let mut issuer = issuer(2);
        let mut budget = pin!(issuer.issue().unwrap());
        let mut return_to_pool = PendingReturn {
            _owner: DropOwner(Arc::clone(&dropped)),
        };
        let timer = tokio::time::sleep(Duration::ZERO);
        tokio::pin!(timer);

        let outcome = std::future::poll_fn(|context| {
            budget
                .as_mut()
                .poll_return(Pin::new(&mut return_to_pool), timer.as_mut(), context)
        })
        .await;
        let DbReturnPoll::BudgetExpired(expired) = outcome else {
            panic!("existing timer must expire");
        };
        assert!(!dropped.load(Ordering::SeqCst));
        let discarded = expired.synchronously_discard(return_to_pool);
        assert!(dropped.load(Ordering::SeqCst));
        issuer.acknowledge_discard(discarded).unwrap();
    }

    #[test]
    fn duplicate_and_foreign_receipts_are_rejected() {
        let mut first = issuer(3);
        let mut second = issuer(4);
        let first_binding = first.issue().unwrap().binding.unwrap();
        let _second_budget = second.issue().unwrap();

        let foreign = DbReturnCompleted {
            binding: second.binding(first_binding.generation),
        };
        assert_eq!(
            first.acknowledge_return(foreign),
            Err(DbReturnBudgetError::ForeignOrStale)
        );

        first
            .acknowledge_return(DbReturnCompleted {
                binding: first_binding,
            })
            .unwrap();
        assert!(first.issue().is_ok());
        assert_eq!(
            first.acknowledge_return(DbReturnCompleted {
                binding: second.binding(1),
            }),
            Err(DbReturnBudgetError::ForeignOrStale)
        );
        assert!(matches!(
            first.issue(),
            Err(DbReturnBudgetError::ActiveBudget)
        ));

        // Ensure the test does not leave a live logical role.
        let generation = first.active_generation.unwrap();
        first.finish(first.binding(generation)).unwrap();
        let second_receipt = second.binding(1);
        second.finish(second_receipt).unwrap();
    }

    #[test]
    fn budget_is_not_clone_and_has_no_duration_or_timer_storage() {
        assert_eq!(
            std::mem::size_of::<DbReturnBudget>(),
            std::mem::size_of::<Option<BudgetBinding>>()
        );
    }
}