fix(scheduling): harden timing invariants

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>

Author: yordis

rsworkspace/crates/trogon-cron/src/domain.rs

@@ -177,7 +177,7 @@ impl JobSchedule {
     fn next_fire_after(&self, from: DateTime<Utc>) -> Option<DateTime<Utc>> {
         match self {
             Self::Interval(interval) => {
-                Some(from + chrono::Duration::seconds(interval.as_chrono_seconds()))
+                from.checked_add_signed(chrono::Duration::seconds(interval.as_chrono_seconds()))
             }
             Self::Cron(expr) => expr.next_after(from),
         }
@@ -499,4 +499,11 @@ mod tests {
             JobSchedule::Interval(_) => panic!("expected cron schedule"),
         }
     }
+
+    #[test]
+    fn interval_next_fire_overflow_returns_none() {
+        let schedule = JobSchedule::Interval(IntervalSeconds::new(1).unwrap());
+
+        assert_eq!(schedule.next_fire_after(DateTime::<Utc>::MAX_UTC), None);
+    }
 }

rsworkspace/crates/trogon-cron/src/scheduler.rs

@@ -72,7 +72,7 @@ impl Scheduler<NatsConfigStore, jetstream::Context, NatsKvLease> {
             tick_publisher: js,
             leader_lock,
             node_id: Uuid::new_v4().to_string(),
-            leader_renew_interval: leader_config.renew_interval().get(),
+            leader_renew_interval: leader_config.renew_interval(),
         })
     }
 }

rsworkspace/crates/trogon-nats/src/lease.rs

@@ -46,44 +46,47 @@ impl LeaseKey {
     }
 }
 
-#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd)]
-pub struct LeaseTtl(Duration);
+#[derive(Debug, Clone, Copy, Eq, PartialEq)]
+pub struct LeaseTiming {
+    ttl: Duration,
+    renew_interval: Duration,
+}
 
-impl LeaseTtl {
-    pub fn new(ttl: Duration) -> Result<Self, LeaseConfigError> {
+impl LeaseTiming {
+    pub fn new(ttl: Duration, renew_interval: Duration) -> Result<Self, LeaseConfigError> {
         if ttl.is_zero() {
             return Err(LeaseConfigError::ZeroTtl);
         }
-        Ok(Self(ttl))
-    }
-
-    pub fn get(self) -> Duration {
-        self.0
-    }
-}
-
-#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd)]
-pub struct LeaseRenewInterval(Duration);
-
-impl LeaseRenewInterval {
-    pub fn new(renew_interval: Duration) -> Result<Self, LeaseConfigError> {
         if renew_interval.is_zero() {
             return Err(LeaseConfigError::ZeroRenewInterval);
         }
-        Ok(Self(renew_interval))
+        if renew_interval >= ttl {
+            return Err(LeaseConfigError::RenewIntervalNotLessThanTtl {
+                renew_interval,
+                ttl,
+            });
+        }
+
+        Ok(Self {
+            ttl,
+            renew_interval,
+        })
+    }
+
+    pub fn ttl(self) -> Duration {
+        self.ttl
     }
 
-    pub fn get(self) -> Duration {
-        self.0
+    pub fn renew_interval(self) -> Duration {
+        self.renew_interval
     }
 }
 
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub struct NatsKvLeaseConfig {
     bucket: LeaseBucket,
     key: LeaseKey,
-    ttl: LeaseTtl,
-    renew_interval: LeaseRenewInterval,
+    timing: LeaseTiming,
 }
 
 impl NatsKvLeaseConfig {
@@ -95,21 +98,12 @@ impl NatsKvLeaseConfig {
     ) -> Result<Self, LeaseConfigError> {
         let bucket = LeaseBucket::new(bucket)?;
         let key = LeaseKey::new(key)?;
-        let ttl = LeaseTtl::new(ttl)?;
-        let renew_interval = LeaseRenewInterval::new(renew_interval)?;
-
-        if renew_interval.get() >= ttl.get() {
-            return Err(LeaseConfigError::RenewIntervalNotLessThanTtl {
-                renew_interval: renew_interval.get(),
-                ttl: ttl.get(),
-            });
-        }
+        let timing = LeaseTiming::new(ttl, renew_interval)?;
 
         Ok(Self {
             bucket,
             key,
-            ttl,
-            renew_interval,
+            timing,
         })
     }
 
@@ -121,12 +115,16 @@ impl NatsKvLeaseConfig {
         &self.key
     }
 
-    pub fn ttl(&self) -> LeaseTtl {
-        self.ttl
+    pub fn timing(&self) -> LeaseTiming {
+        self.timing
     }
 
-    pub fn renew_interval(&self) -> LeaseRenewInterval {
-        self.renew_interval
+    pub fn ttl(&self) -> Duration {
+        self.timing.ttl()
+    }
+
+    pub fn renew_interval(&self) -> Duration {
+        self.timing.renew_interval()
     }
 }
 
@@ -295,7 +293,7 @@ impl NatsKvLease {
             .create_key_value(kv::Config {
                 bucket: config.bucket().as_str().to_owned(),
                 history: 1,
-                max_age: config.ttl().get(),
+                max_age: config.ttl(),
                 ..Default::default()
             })
             .await

rsworkspace/crates/trogon-nats/src/lib.rs

@@ -56,8 +56,8 @@ pub use client::{FlushClient, PublishClient, RequestClient, SubscribeClient};
 pub use connect::{ConnectError, connect};
 pub use constants::REQ_ID_HEADER;
 pub use lease::{
-    LeaderElection, Lease, LeaseBucket, LeaseConfigError, LeaseError, LeaseKey, LeaseRenewInterval,
-    LeaseTtl, NatsKvLease, NatsKvLeaseConfig, TryAcquireOutcome,
+    LeaderElection, Lease, LeaseBucket, LeaseConfigError, LeaseError, LeaseKey, LeaseTiming,
+    NatsKvLease, NatsKvLeaseConfig, TryAcquireOutcome,
 };
 pub use messaging::{
     FlushPolicy, NatsError, PublishOperationError, PublishOptions, PublishOptionsBuilder,