runifold_workflow/
task_retention.rs

1//! Dynamically sharded supervision for fenced terminal Task cleanup.
2
3use std::{
4    fmt,
5    num::{NonZeroU32, NonZeroUsize},
6    sync::{
7        Arc,
8        atomic::{AtomicBool, AtomicU64, Ordering},
9    },
10    time::Duration,
11};
12
13use futures_util::{
14    StreamExt,
15    future::{Either, select},
16    stream::FuturesUnordered,
17};
18use runifold_core::CancellationToken;
19
20use crate::{
21    LeaseDuration, SystemWorkflowWorkerSleeper, WorkerId, WorkflowStoreError,
22    WorkflowStoreErrorKind, WorkflowTaskCleanupLease, WorkflowTaskCleanupLimit,
23    WorkflowTaskRetention, WorkflowTaskRetentionStore, WorkflowTenantId, WorkflowTenantListLimit,
24    WorkflowWorkerSleeper,
25};
26
27/// Stable assignment of tenants across cleanup processes.
28#[derive(Clone, Copy, Debug, Eq, PartialEq)]
29pub struct WorkflowTaskCleanupShard {
30    index: u32,
31    count: NonZeroU32,
32}
33
34impl WorkflowTaskCleanupShard {
35    /// Creates one zero-based shard.
36    ///
37    /// # Errors
38    ///
39    /// Rejects an index outside the shard count.
40    pub fn new(index: u32, count: NonZeroU32) -> Result<Self, WorkflowStoreError> {
41        if index >= count.get() {
42            return Err(invalid_config(
43                "Task cleanup shard index must be smaller than shard count",
44            ));
45        }
46        Ok(Self { index, count })
47    }
48
49    /// Zero-based shard index.
50    pub const fn index(self) -> u32 {
51        self.index
52    }
53
54    /// Total configured shards.
55    pub const fn count(self) -> u32 {
56        self.count.get()
57    }
58
59    /// Returns whether this shard owns a tenant under the stable hash.
60    pub fn owns(self, tenant_id: &WorkflowTenantId) -> bool {
61        stable_tenant_hash(tenant_id.as_str()) % u64::from(self.count.get())
62            == u64::from(self.index)
63    }
64}
65
66/// Validated bounds and timing for automatic terminal Task cleanup.
67#[derive(Clone, Copy, Debug, Eq, PartialEq)]
68pub struct WorkflowTaskCleanupSupervisorConfig {
69    shard: WorkflowTaskCleanupShard,
70    retention: WorkflowTaskRetention,
71    lease_duration: LeaseDuration,
72    heartbeat_interval: Duration,
73    cleanup_limit: WorkflowTaskCleanupLimit,
74    max_batches_per_tenant: NonZeroU32,
75    discovery_limit: WorkflowTenantListLimit,
76    max_concurrency: NonZeroUsize,
77    scan_interval: Duration,
78    error_backoff: Duration,
79}
80
81impl WorkflowTaskCleanupSupervisorConfig {
82    /// Creates conservative cleanup timing and work bounds.
83    ///
84    /// # Errors
85    ///
86    /// Rejects a zero heartbeat or one not shorter than the lease.
87    pub fn new(
88        shard: WorkflowTaskCleanupShard,
89        retention: WorkflowTaskRetention,
90        lease_duration: LeaseDuration,
91        heartbeat_interval: Duration,
92    ) -> Result<Self, WorkflowStoreError> {
93        let heartbeat_ms = u64::try_from(heartbeat_interval.as_millis())
94            .map_err(|_| invalid_config("Task cleanup heartbeat exceeds supported milliseconds"))?;
95        if heartbeat_ms == 0 || heartbeat_ms >= lease_duration.as_millis() {
96            return Err(invalid_config(
97                "Task cleanup heartbeat must be positive and shorter than the lease",
98            ));
99        }
100        Ok(Self {
101            shard,
102            retention,
103            lease_duration,
104            heartbeat_interval,
105            cleanup_limit: WorkflowTaskCleanupLimit::new(100)?,
106            max_batches_per_tenant: NonZeroU32::new(10).unwrap_or(NonZeroU32::MIN),
107            discovery_limit: WorkflowTenantListLimit::default(),
108            max_concurrency: NonZeroUsize::new(8).unwrap_or(NonZeroUsize::MIN),
109            scan_interval: Duration::from_secs(30),
110            error_backoff: Duration::from_secs(1),
111        })
112    }
113
114    /// Sets bounded discovery, concurrency, batch size, and work per claim.
115    #[must_use]
116    pub const fn with_work_limits(
117        mut self,
118        discovery_limit: WorkflowTenantListLimit,
119        max_concurrency: NonZeroUsize,
120        cleanup_limit: WorkflowTaskCleanupLimit,
121        max_batches_per_tenant: NonZeroU32,
122    ) -> Self {
123        self.discovery_limit = discovery_limit;
124        self.max_concurrency = max_concurrency;
125        self.cleanup_limit = cleanup_limit;
126        self.max_batches_per_tenant = max_batches_per_tenant;
127        self
128    }
129
130    /// Sets successful-scan and error delays.
131    ///
132    /// # Errors
133    ///
134    /// Rejects zero durations to prevent hot loops.
135    pub fn with_intervals(
136        mut self,
137        scan_interval: Duration,
138        error_backoff: Duration,
139    ) -> Result<Self, WorkflowStoreError> {
140        if scan_interval.is_zero() || error_backoff.is_zero() {
141            return Err(invalid_config(
142                "Task cleanup scan interval and error backoff must be positive",
143            ));
144        }
145        self.scan_interval = scan_interval;
146        self.error_backoff = error_backoff;
147        Ok(self)
148    }
149}
150
151/// Cumulative work performed by cleanup scans.
152#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
153pub struct WorkflowTaskCleanupSupervisorReport {
154    /// Completed full tenant discovery scans.
155    pub scans: u64,
156    /// Terminal-Task tenants returned by the store.
157    pub tenants_discovered: u64,
158    /// Discovered tenants assigned to this shard.
159    pub tenants_assigned: u64,
160    /// Successfully acquired tenant cleanup leases.
161    pub claims: u64,
162    /// Claims blocked by another active owner.
163    pub contended: u64,
164    /// Non-empty cleanup batches committed.
165    pub batches_cleaned: u64,
166    /// Tasks atomically tombstoned and deleted.
167    pub tasks_deleted: u64,
168    /// Lease-loss and stale-fencing failures.
169    pub leases_lost: u64,
170    /// Store failures isolated from other tenants.
171    pub infrastructure_errors: u64,
172    /// Discovery failures that aborted a scan.
173    pub discovery_errors: u64,
174}
175
176impl WorkflowTaskCleanupSupervisorReport {
177    fn merge(&mut self, other: Self) {
178        self.scans = self.scans.saturating_add(other.scans);
179        self.tenants_discovered = self
180            .tenants_discovered
181            .saturating_add(other.tenants_discovered);
182        self.tenants_assigned = self.tenants_assigned.saturating_add(other.tenants_assigned);
183        self.claims = self.claims.saturating_add(other.claims);
184        self.contended = self.contended.saturating_add(other.contended);
185        self.batches_cleaned = self.batches_cleaned.saturating_add(other.batches_cleaned);
186        self.tasks_deleted = self.tasks_deleted.saturating_add(other.tasks_deleted);
187        self.leases_lost = self.leases_lost.saturating_add(other.leases_lost);
188        self.infrastructure_errors = self
189            .infrastructure_errors
190            .saturating_add(other.infrastructure_errors);
191        self.discovery_errors = self.discovery_errors.saturating_add(other.discovery_errors);
192    }
193}
194
195/// Lock-free, low-cardinality health state for readiness and telemetry export.
196#[derive(Clone, Debug, Default)]
197pub struct WorkflowTaskCleanupSupervisorMetrics {
198    state: Arc<WorkflowTaskCleanupSupervisorMetricState>,
199}
200
201#[derive(Debug, Default)]
202struct WorkflowTaskCleanupSupervisorMetricState {
203    scan_active: AtomicBool,
204    scans: AtomicU64,
205    claims: AtomicU64,
206    contended: AtomicU64,
207    tasks_deleted: AtomicU64,
208    leases_lost: AtomicU64,
209    infrastructure_errors: AtomicU64,
210}
211
212/// Point-in-time Task cleanup supervisor health snapshot.
213#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
214pub struct WorkflowTaskCleanupSupervisorMetricSnapshot {
215    /// Whether a discovery scan is currently running.
216    pub scan_active: bool,
217    /// Completed scans.
218    pub scans: u64,
219    /// Successful tenant claims.
220    pub claims: u64,
221    /// Contended tenant claims.
222    pub contended: u64,
223    /// Tasks atomically deleted.
224    pub tasks_deleted: u64,
225    /// Lease-loss failures.
226    pub leases_lost: u64,
227    /// Other store failures.
228    pub infrastructure_errors: u64,
229}
230
231/// Optional low-cardinality observer for cleanup control-plane outcomes.
232///
233/// Implementations must not retain tenant identities or block cleanup.
234pub trait WorkflowTaskCleanupObserver: Send + Sync {
235    /// Observes one completed full scan.
236    fn observe_scan(&self, report: WorkflowTaskCleanupSupervisorReport);
237
238    /// Observes a discovery failure that aborted a scan.
239    fn observe_discovery_error(&self);
240}
241
242#[derive(Debug, Default)]
243struct NoopWorkflowTaskCleanupObserver;
244
245impl WorkflowTaskCleanupObserver for NoopWorkflowTaskCleanupObserver {
246    fn observe_scan(&self, _report: WorkflowTaskCleanupSupervisorReport) {}
247
248    fn observe_discovery_error(&self) {}
249}
250
251impl WorkflowTaskCleanupSupervisorMetrics {
252    /// Reads one lock-free operational snapshot.
253    pub fn snapshot(&self) -> WorkflowTaskCleanupSupervisorMetricSnapshot {
254        WorkflowTaskCleanupSupervisorMetricSnapshot {
255            scan_active: self.state.scan_active.load(Ordering::Relaxed),
256            scans: self.state.scans.load(Ordering::Relaxed),
257            claims: self.state.claims.load(Ordering::Relaxed),
258            contended: self.state.contended.load(Ordering::Relaxed),
259            tasks_deleted: self.state.tasks_deleted.load(Ordering::Relaxed),
260            leases_lost: self.state.leases_lost.load(Ordering::Relaxed),
261            infrastructure_errors: self.state.infrastructure_errors.load(Ordering::Relaxed),
262        }
263    }
264
265    fn begin_scan(&self) {
266        self.state.scan_active.store(true, Ordering::Relaxed);
267    }
268
269    fn finish_scan(&self, report: WorkflowTaskCleanupSupervisorReport) {
270        self.state.scan_active.store(false, Ordering::Relaxed);
271        self.state.scans.fetch_add(report.scans, Ordering::Relaxed);
272        self.state
273            .claims
274            .fetch_add(report.claims, Ordering::Relaxed);
275        self.state
276            .contended
277            .fetch_add(report.contended, Ordering::Relaxed);
278        self.state
279            .tasks_deleted
280            .fetch_add(report.tasks_deleted, Ordering::Relaxed);
281        self.state
282            .leases_lost
283            .fetch_add(report.leases_lost, Ordering::Relaxed);
284        self.state
285            .infrastructure_errors
286            .fetch_add(report.infrastructure_errors, Ordering::Relaxed);
287    }
288
289    fn discovery_error(&self) {
290        self.state.scan_active.store(false, Ordering::Relaxed);
291        self.state
292            .infrastructure_errors
293            .fetch_add(1, Ordering::Relaxed);
294    }
295}
296
297/// Discovers, shards, and cleans terminal Tasks with bounded concurrency.
298pub struct WorkflowTaskCleanupSupervisor<S> {
299    store: Arc<S>,
300    owner: WorkerId,
301    config: WorkflowTaskCleanupSupervisorConfig,
302    metrics: WorkflowTaskCleanupSupervisorMetrics,
303    observer: Arc<dyn WorkflowTaskCleanupObserver>,
304    sleeper: Arc<dyn WorkflowWorkerSleeper>,
305}
306
307impl<S> WorkflowTaskCleanupSupervisor<S>
308where
309    S: WorkflowTaskRetentionStore + 'static,
310{
311    /// Creates a cleanup supervisor with system timers.
312    pub fn new(
313        store: Arc<S>,
314        owner: WorkerId,
315        config: WorkflowTaskCleanupSupervisorConfig,
316    ) -> Self {
317        Self {
318            store,
319            owner,
320            config,
321            metrics: WorkflowTaskCleanupSupervisorMetrics::default(),
322            observer: Arc::new(NoopWorkflowTaskCleanupObserver),
323            sleeper: Arc::new(SystemWorkflowWorkerSleeper),
324        }
325    }
326
327    /// Overrides sleeping for deterministic runtimes and tests.
328    #[must_use]
329    pub fn with_sleeper(mut self, sleeper: Arc<dyn WorkflowWorkerSleeper>) -> Self {
330        self.sleeper = sleeper;
331        self
332    }
333
334    /// Uses shared lock-free health metrics.
335    #[must_use]
336    pub fn with_metrics(mut self, metrics: WorkflowTaskCleanupSupervisorMetrics) -> Self {
337        self.metrics = metrics;
338        self
339    }
340
341    /// Attaches a non-blocking low-cardinality outcome observer.
342    #[must_use]
343    pub fn with_observer(mut self, observer: Arc<dyn WorkflowTaskCleanupObserver>) -> Self {
344        self.observer = observer;
345        self
346    }
347
348    /// Returns this supervisor's health metrics.
349    pub const fn metrics(&self) -> &WorkflowTaskCleanupSupervisorMetrics {
350        &self.metrics
351    }
352
353    /// Performs one stable paginated discovery and bounded cleanup scan.
354    ///
355    /// Per-tenant failures are isolated. Discovery failure aborts the scan.
356    ///
357    /// # Errors
358    ///
359    /// Returns a typed store error when tenant discovery fails.
360    pub async fn scan_once(
361        &self,
362    ) -> Result<WorkflowTaskCleanupSupervisorReport, WorkflowStoreError> {
363        self.metrics.begin_scan();
364        let result = self.scan_pages().await;
365        match result {
366            Ok(report) => {
367                self.metrics.finish_scan(report);
368                self.observer.observe_scan(report);
369                Ok(report)
370            }
371            Err(error) => {
372                self.metrics.discovery_error();
373                self.observer.observe_discovery_error();
374                Err(error)
375            }
376        }
377    }
378
379    /// Continuously rescans until cancellation.
380    ///
381    /// A scan already in progress is drained before shutdown.
382    pub async fn run(&self, shutdown: &CancellationToken) -> WorkflowTaskCleanupSupervisorReport {
383        let mut report = WorkflowTaskCleanupSupervisorReport::default();
384        while !shutdown.is_cancelled() {
385            let delay = if let Ok(scan) = self.scan_once().await {
386                report.merge(scan);
387                self.config.scan_interval
388            } else {
389                report.discovery_errors = report.discovery_errors.saturating_add(1);
390                self.config.error_backoff
391            };
392            if wait_or_shutdown(Arc::clone(&self.sleeper), delay, shutdown).await {
393                break;
394            }
395        }
396        report
397    }
398
399    async fn scan_pages(&self) -> Result<WorkflowTaskCleanupSupervisorReport, WorkflowStoreError> {
400        let mut report = WorkflowTaskCleanupSupervisorReport::default();
401        let mut after = None;
402        loop {
403            let tenants = self
404                .store
405                .list_task_cleanup_tenants(after, self.config.discovery_limit)
406                .await?;
407            let page_len = tenants.len();
408            report.tenants_discovered = report
409                .tenants_discovered
410                .saturating_add(u64::try_from(page_len).unwrap_or(u64::MAX));
411            let last = tenants.last().cloned();
412            let assigned = tenants
413                .into_iter()
414                .filter(|tenant| self.config.shard.owns(tenant))
415                .collect::<Vec<_>>();
416            report.tenants_assigned = report
417                .tenants_assigned
418                .saturating_add(u64::try_from(assigned.len()).unwrap_or(u64::MAX));
419            self.cleanup_assigned(assigned, &mut report).await;
420            if page_len < usize::try_from(self.config.discovery_limit.get()).unwrap_or(usize::MAX) {
421                break;
422            }
423            after = last;
424        }
425        report.scans = 1;
426        Ok(report)
427    }
428
429    async fn cleanup_assigned(
430        &self,
431        tenants: Vec<WorkflowTenantId>,
432        report: &mut WorkflowTaskCleanupSupervisorReport,
433    ) {
434        let mut tenants = tenants.into_iter();
435        let mut active = FuturesUnordered::new();
436        for _ in 0..self.config.max_concurrency.get() {
437            let Some(tenant) = tenants.next() else {
438                break;
439            };
440            active.push(self.cleanup_tenant(tenant));
441        }
442        while let Some(outcome) = active.next().await {
443            match outcome {
444                CleanupOutcome::Contended => {
445                    report.contended = report.contended.saturating_add(1);
446                }
447                CleanupOutcome::Cleaned {
448                    batches,
449                    tasks_deleted,
450                } => {
451                    record_cleaned(report, batches, tasks_deleted);
452                }
453                CleanupOutcome::CleanedLeaseLost {
454                    batches,
455                    tasks_deleted,
456                } => {
457                    record_cleaned(report, batches, tasks_deleted);
458                    report.leases_lost = report.leases_lost.saturating_add(1);
459                }
460                CleanupOutcome::CleanedInfrastructureError {
461                    batches,
462                    tasks_deleted,
463                } => {
464                    record_cleaned(report, batches, tasks_deleted);
465                    report.infrastructure_errors = report.infrastructure_errors.saturating_add(1);
466                }
467                CleanupOutcome::LeaseLost => {
468                    report.claims = report.claims.saturating_add(1);
469                    report.leases_lost = report.leases_lost.saturating_add(1);
470                }
471                CleanupOutcome::InfrastructureError { claimed } => {
472                    report.claims = report.claims.saturating_add(u64::from(claimed));
473                    report.infrastructure_errors = report.infrastructure_errors.saturating_add(1);
474                }
475            }
476            if let Some(tenant) = tenants.next() {
477                active.push(self.cleanup_tenant(tenant));
478            }
479        }
480    }
481
482    async fn cleanup_tenant(&self, tenant: WorkflowTenantId) -> CleanupOutcome {
483        let lease = match self
484            .store
485            .claim_task_cleanup(tenant, self.owner.clone(), self.config.lease_duration)
486            .await
487        {
488            Ok(Some(lease)) => lease,
489            Ok(None) => return CleanupOutcome::Contended,
490            Err(_) => return CleanupOutcome::InfrastructureError { claimed: false },
491        };
492        let cleanup = cleanup_claimed(
493            Arc::clone(&self.store),
494            lease.clone(),
495            self.config.retention,
496            self.config.cleanup_limit,
497            self.config.max_batches_per_tenant,
498        );
499        let heartbeat = heartbeat_until_failure(
500            Arc::clone(&self.store),
501            lease.clone(),
502            self.config.lease_duration,
503            self.config.heartbeat_interval,
504            Arc::clone(&self.sleeper),
505        );
506        let outcome = match select(Box::pin(cleanup), Box::pin(heartbeat)).await {
507            Either::Left((result, _)) => match result {
508                Ok((batches, tasks_deleted)) => CleanupOutcome::Cleaned {
509                    batches,
510                    tasks_deleted,
511                },
512                Err(error) => classify_claimed_error(&error),
513            },
514            Either::Right((error, _)) => classify_claimed_error(&error),
515        };
516        let release = self.store.release_task_cleanup(lease).await;
517        if let (
518            CleanupOutcome::Cleaned {
519                batches,
520                tasks_deleted,
521            },
522            Err(error),
523        ) = (outcome, release)
524        {
525            return if error.kind == WorkflowStoreErrorKind::LeaseLost {
526                CleanupOutcome::CleanedLeaseLost {
527                    batches,
528                    tasks_deleted,
529                }
530            } else {
531                CleanupOutcome::CleanedInfrastructureError {
532                    batches,
533                    tasks_deleted,
534                }
535            };
536        }
537        outcome
538    }
539}
540
541impl<S> fmt::Debug for WorkflowTaskCleanupSupervisor<S> {
542    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
543        formatter
544            .debug_struct("WorkflowTaskCleanupSupervisor")
545            .field("owner", &self.owner)
546            .field("config", &self.config)
547            .field("metrics", &self.metrics.snapshot())
548            .field("observer", &"<task-cleanup-observer>")
549            .finish_non_exhaustive()
550    }
551}
552
553#[derive(Clone, Copy, Debug, Eq, PartialEq)]
554enum CleanupOutcome {
555    Contended,
556    Cleaned { batches: u32, tasks_deleted: u64 },
557    CleanedLeaseLost { batches: u32, tasks_deleted: u64 },
558    CleanedInfrastructureError { batches: u32, tasks_deleted: u64 },
559    LeaseLost,
560    InfrastructureError { claimed: bool },
561}
562
563fn record_cleaned(
564    report: &mut WorkflowTaskCleanupSupervisorReport,
565    batches: u32,
566    tasks_deleted: u64,
567) {
568    report.claims = report.claims.saturating_add(1);
569    report.batches_cleaned = report.batches_cleaned.saturating_add(u64::from(batches));
570    report.tasks_deleted = report.tasks_deleted.saturating_add(tasks_deleted);
571}
572
573async fn cleanup_claimed<S>(
574    store: Arc<S>,
575    lease: WorkflowTaskCleanupLease,
576    retention: WorkflowTaskRetention,
577    limit: WorkflowTaskCleanupLimit,
578    max_batches: NonZeroU32,
579) -> Result<(u32, u64), WorkflowStoreError>
580where
581    S: WorkflowTaskRetentionStore,
582{
583    let mut batches = 0_u32;
584    let mut tasks_deleted = 0_u64;
585    for _ in 0..max_batches.get() {
586        let tombstones = store
587            .compact_terminal_tasks(lease.clone(), retention, limit)
588            .await?;
589        let count = tombstones.len();
590        if count == 0 {
591            break;
592        }
593        batches = batches.saturating_add(1);
594        tasks_deleted = tasks_deleted.saturating_add(u64::try_from(count).unwrap_or(u64::MAX));
595        if count < usize::try_from(limit.get()).unwrap_or(usize::MAX) {
596            break;
597        }
598    }
599    Ok((batches, tasks_deleted))
600}
601
602async fn heartbeat_until_failure<S>(
603    store: Arc<S>,
604    mut lease: WorkflowTaskCleanupLease,
605    extension: LeaseDuration,
606    interval: Duration,
607    sleeper: Arc<dyn WorkflowWorkerSleeper>,
608) -> WorkflowStoreError
609where
610    S: WorkflowTaskRetentionStore,
611{
612    loop {
613        sleeper.sleep(interval).await;
614        match store.heartbeat_task_cleanup(lease, extension).await {
615            Ok(renewed) => lease = renewed,
616            Err(error) => return error,
617        }
618    }
619}
620
621fn classify_claimed_error(error: &WorkflowStoreError) -> CleanupOutcome {
622    if error.kind == WorkflowStoreErrorKind::LeaseLost {
623        CleanupOutcome::LeaseLost
624    } else {
625        CleanupOutcome::InfrastructureError { claimed: true }
626    }
627}
628
629fn invalid_config(message: &'static str) -> WorkflowStoreError {
630    WorkflowStoreError::new(WorkflowStoreErrorKind::InvalidInput, message)
631}
632
633async fn wait_or_shutdown(
634    sleeper: Arc<dyn WorkflowWorkerSleeper>,
635    duration: Duration,
636    shutdown: &CancellationToken,
637) -> bool {
638    matches!(
639        select(
640            Box::pin(shutdown.cancelled()),
641            Box::pin(sleeper.sleep(duration))
642        )
643        .await,
644        Either::Left(_)
645    )
646}
647
648fn stable_tenant_hash(value: &str) -> u64 {
649    value
650        .as_bytes()
651        .iter()
652        .fold(0xcbf2_9ce4_8422_2325, |hash, byte| {
653            (hash ^ u64::from(*byte)).wrapping_mul(0x0000_0100_0000_01b3)
654        })
655}
656
657#[cfg(test)]
658mod tests {
659    use std::num::NonZeroU32;
660
661    use super::*;
662
663    #[test]
664    fn shards_form_a_stable_complete_partition() {
665        let tenant = WorkflowTenantId::parse("tenant-a").unwrap();
666        let owners = (0..4)
667            .filter(|index| {
668                WorkflowTaskCleanupShard::new(*index, NonZeroU32::new(4).unwrap())
669                    .unwrap()
670                    .owns(&tenant)
671            })
672            .count();
673        assert_eq!(owners, 1);
674    }
675
676    #[test]
677    fn supervisor_rejects_unsafe_heartbeat_timing() {
678        let shard = WorkflowTaskCleanupShard::new(0, NonZeroU32::new(1).unwrap()).unwrap();
679        let retention = WorkflowTaskRetention::new(Duration::from_secs(1)).unwrap();
680        let lease = LeaseDuration::new(Duration::from_secs(1)).unwrap();
681        assert!(
682            WorkflowTaskCleanupSupervisorConfig::new(
683                shard,
684                retention,
685                lease,
686                Duration::from_secs(1),
687            )
688            .is_err()
689        );
690    }
691}
runifold_workflow/task_retention.rs

runifold_workflow/
task_retention.rs
