Remove the deadliner trait

Author: therustmonk

crates/core/src/deadline/calculator.rs

@@ -86,6 +86,17 @@ pub trait DeadlineCalculator: Send + Sync + 'static {
     fn deadline(&self, duty: &Duty) -> Result<Option<DateTime<Utc>>>;
 }
 
+/// Calculator that reports every duty as never expiring. Useful for
+/// scenarios that need to plug into the deadliner API but don't actually want
+/// any eviction (e.g. DKG, which is one-shot and outside the slot timeline).
+pub struct NeverExpiringCalculator;
+
+impl DeadlineCalculator for NeverExpiringCalculator {
+    fn deadline(&self, _duty: &Duty) -> Result<Option<DateTime<Utc>>> {
+        Ok(None)
+    }
+}
+
 impl DeadlineCalculator for DutyDeadlineCalculator {
     fn deadline(&self, duty: &Duty) -> Result<Option<DateTime<Utc>>> {
         if duty.duty_type.never_expires() {

crates/core/src/deadline/mod.rs

@@ -8,7 +8,7 @@
 //!
 //! ```no_run
 //! use pluto_core::{
-//!     deadline::{AddOutcome, Deadliner, DeadlinerTask, DutyDeadlineCalculator},
+//!     deadline::{AddOutcome, DeadlinerTask, DutyDeadlineCalculator},
 //!     types::{Duty, SlotNumber},
 //! };
 //! use pluto_eth2api::EthBeaconNodeApiClient;
@@ -37,10 +37,9 @@
 mod calculator;
 mod msecs;
 
-pub use calculator::{DeadlineCalculator, DutyDeadlineCalculator};
+pub use calculator::{DeadlineCalculator, DutyDeadlineCalculator, NeverExpiringCalculator};
 
 use crate::types::{Duty, DutyType, SlotNumber};
-use async_trait::async_trait;
 use chrono::{DateTime, Utc};
 use pluto_eth2api::EthBeaconNodeApiClientError;
 use std::{collections::HashSet, time::Duration};
@@ -109,21 +108,6 @@ impl AddOutcome {
     }
 }
 
-/// Deadliner provides duty deadline functionality.
-///
-/// Producers submit duties via [`add`](Self::add). Expired duties are
-/// delivered on the receiver paired with the handle at
-/// [`DeadlinerTask::start`].
-#[async_trait]
-pub trait Deadliner: Send + Sync {
-    /// Adds a duty for deadline scheduling.
-    ///
-    /// Idempotent: re-adding a duty already tracked returns
-    /// [`AddOutcome::Scheduled`] again. See [`AddOutcome`] for the meaning of
-    /// each variant.
-    async fn add(&self, duty: Duty) -> AddOutcome;
-}
-
 /// Internal message type for adding duties to the deadliner.
 struct DeadlineInput {
     duty: Duty,
@@ -139,24 +123,25 @@ pub struct DeadlinerHandle {
     input_tx: mpsc::Sender<DeadlineInput>,
 }
 
-#[async_trait]
-impl Deadliner for DeadlinerHandle {
-    async fn add(&self, duty: Duty) -> AddOutcome {
-        // Check if shut down
+impl DeadlinerHandle {
+    /// Adds a duty for deadline scheduling.
+    ///
+    /// Idempotent: re-adding a duty already tracked returns
+    /// [`AddOutcome::Scheduled`] again. See [`AddOutcome`] for the meaning of
+    /// each variant.
+    pub async fn add(&self, duty: Duty) -> AddOutcome {
         if self.cancel_token.is_cancelled() {
             return AddOutcome::FailedToCompute;
         }
 
         let (response_tx, response_rx) = oneshot::channel();
         let input = DeadlineInput { duty, response_tx };
 
-        // Send the duty to the background task
         if self.input_tx.send(input).await.is_err() {
             return AddOutcome::FailedToCompute;
         }
 
-        // Wait for response — `FailedToCompute` if the task dropped the
-        // sender (shutdown race).
+        // `FailedToCompute` if the task dropped the sender (shutdown race).
         response_rx.await.unwrap_or(AddOutcome::FailedToCompute)
     }
 }

crates/core/src/dutydb/memory.rs

@@ -2,7 +2,7 @@
 //!
 //! Equivalent to charon/core/dutydb/memory.go.
 
-use std::{collections::HashMap, sync::Arc};
+use std::collections::HashMap;
 
 use pluto_eth2api::{
     spec::{altair, phase0},
@@ -14,7 +14,7 @@ use tracing::{info, warn};
 use tree_hash::TreeHash;
 
 use crate::{
-    deadline::Deadliner,
+    deadline::DeadlinerHandle,
     signeddata::{
         AttestationData, SyncContribution, VersionedAggregatedAttestation, VersionedProposal,
     },
@@ -199,14 +199,14 @@ pub struct MemDB {
     aggregation_notify: Notify,
     contrib_notify: Notify,
     cancel: CancellationToken,
-    deadliner: Arc<dyn Deadliner>,
+    deadliner: DeadlinerHandle,
 }
 
 impl MemDB {
     /// Creates a new in-memory DutyDB. `deadliner_rx` is the receiver paired
     /// with `deadliner` (typically from `DeadlinerTask::start`).
     pub fn new(
-        deadliner: Arc<dyn Deadliner>,
+        deadliner: DeadlinerHandle,
         deadliner_rx: mpsc::Receiver<Duty>,
         cancel: &CancellationToken,
     ) -> Self {
@@ -601,27 +601,29 @@ impl State {
 
 #[cfg(test)]
 mod tests {
-    use std::sync::{Arc, Mutex};
+    use std::sync::Arc;
 
-    use async_trait::async_trait;
-    use tokio::sync::mpsc::{Receiver, Sender, channel};
+    use chrono::{DateTime, Utc};
+    use tokio::sync::mpsc::{Receiver, channel};
     use tokio_util::sync::CancellationToken;
 
     use super::*;
     use crate::{
-        deadline::AddOutcome,
+        deadline::{self, DeadlineCalculator, DeadlinerTask},
         signeddata::{AttesterDuty, ProposalBlock},
         testutils::random_core_pub_key,
         types::{DutyType, SlotNumber},
     };
 
-    /// Deadliner that always accepts duties and never expires them.
-    pub(crate) struct NoopDeadliner;
+    /// Test calculator whose every duty is `Scheduled` (deadline is `MAX_UTC`).
+    /// The deadliner never actually fires for this calculator, so the paired
+    /// output receiver stays silent — eviction in tests is driven manually
+    /// through a separate channel (see `duty_expiry`).
+    struct FarFutureCalculator;
 
-    #[async_trait]
-    impl Deadliner for NoopDeadliner {
-        async fn add(&self, _duty: Duty) -> AddOutcome {
-            AddOutcome::Scheduled
+    impl DeadlineCalculator for FarFutureCalculator {
+        fn deadline(&self, _: &Duty) -> deadline::Result<Option<DateTime<Utc>>> {
+            Ok(Some(DateTime::<Utc>::MAX_UTC))
         }
     }
 
@@ -631,57 +633,26 @@ mod tests {
         rx
     }
 
-    /// Deadliner that collects duties and can flush them to a channel on
-    /// demand.
-    pub(crate) struct TestDeadliner {
-        added: Mutex<Vec<Duty>>,
-        tx: Sender<Duty>,
-    }
-
-    impl TestDeadliner {
-        /// Returns the test deadliner alongside the matching expiry receiver
-        /// — call `expire()` on the deadliner to drive `rx`.
-        pub(crate) fn new() -> (Arc<Self>, Receiver<Duty>) {
-            let (tx, rx) = channel(64);
-            let deadliner = Arc::new(Self {
-                added: Mutex::new(Vec::new()),
-                tx,
-            });
-            (deadliner, rx)
-        }
-
-        /// Send all collected duties to the expiry channel.
-        pub(crate) async fn expire(&self) {
-            let duties: Vec<Duty> = {
-                let mut added = self.added.lock().unwrap();
-                std::mem::take(&mut *added)
-            };
-            for duty in duties {
-                let _ = self.tx.send(duty).await;
-            }
-        }
-    }
-
-    #[async_trait]
-    impl Deadliner for TestDeadliner {
-        async fn add(&self, duty: Duty) -> AddOutcome {
-            self.added.lock().unwrap().push(duty);
-            AddOutcome::Scheduled
-        }
+    /// Creates a real deadliner handle backed by [`FarFutureCalculator`] —
+    /// `add()` always reports `Scheduled` but nothing naturally expires.
+    fn far_future_handle() -> DeadlinerHandle {
+        let (handle, _drop_rx) = DeadlinerTask::start(
+            CancellationToken::new(),
+            "dutydb-tests",
+            FarFutureCalculator,
+        );
+        handle
     }
 
     fn make_db() -> MemDB {
         MemDB::new(
-            Arc::new(NoopDeadliner),
+            far_future_handle(),
             noop_deadliner_rx(),
             &CancellationToken::new(),
         )
     }
 
-    fn make_db_with_deadliner(
-        deadliner: Arc<dyn Deadliner>,
-        deadliner_rx: Receiver<Duty>,
-    ) -> MemDB {
+    fn make_db_with_deadliner(deadliner: DeadlinerHandle, deadliner_rx: Receiver<Duty>) -> MemDB {
         MemDB::new(deadliner, deadliner_rx, &CancellationToken::new())
     }
 
@@ -1112,8 +1083,12 @@ mod tests {
 
     #[tokio::test]
     async fn duty_expiry() {
-        let (deadliner, deadliner_rx) = TestDeadliner::new();
-        let db = make_db_with_deadliner(deadliner.clone(), deadliner_rx);
+        // Real handle so `store()`'s `add(...).is_scheduled()` check passes.
+        // Eviction is driven manually via `trim_tx` so the test stays
+        // deterministic instead of racing the deadliner's timer.
+        let deadliner = far_future_handle();
+        let (trim_tx, trim_rx) = channel::<Duty>(64);
+        let db = make_db_with_deadliner(deadliner, trim_rx);
 
         const SLOT: u64 = 123;
 
@@ -1130,8 +1105,12 @@ mod tests {
         // Should be findable now.
         db.pub_key_by_attestation(SLOT, 0, 0).await.unwrap();
 
-        // Expire the duty.
-        deadliner.expire().await;
+        // Expire the duty: simulate the deadliner emitting it.
+        let expired_duty = Duty::new(SlotNumber::new(SLOT), DutyType::Attester);
+        trim_tx
+            .send(expired_duty)
+            .await
+            .expect("trim_tx should be open");
 
         // Trigger expiry processing by storing another duty.
         let proposal = phase0_proposal(SLOT.saturating_add(1), 0);

crates/core/src/parsigdb/memory.rs

@@ -4,7 +4,7 @@ use tokio_util::sync::CancellationToken;
 use tracing::debug;
 
 use crate::{
-    deadline::Deadliner,
+    deadline::DeadlinerHandle,
     parsigdb::metrics::PARSIG_DB_METRICS,
     signeddata::SignedDataError,
     types::{Duty, DutyType, ParSignedData, ParSignedDataSet, PubKey},
@@ -178,7 +178,7 @@ pub struct MemDBInner {
 pub struct MemDB {
     ct: CancellationToken,
     inner: Arc<Mutex<MemDBInner>>,
-    deadliner: Arc<dyn Deadliner>,
+    deadliner: DeadlinerHandle,
     threshold: u64,
 }
 
@@ -189,7 +189,7 @@ impl MemDB {
     /// * `ct` - Cancellation token for graceful shutdown
     /// * `threshold` - Number of matching partial signatures required
     /// * `deadliner` - Deadliner for managing duty expiration
-    pub fn new(ct: CancellationToken, threshold: u64, deadliner: Arc<dyn Deadliner>) -> Self {
+    pub fn new(ct: CancellationToken, threshold: u64, deadliner: DeadlinerHandle) -> Self {
         Self {
             ct,
             inner: Arc::new(Mutex::new(MemDBInner {