obeli-sk-wasm-workers 0.32.0

Internal package of obeli-sk
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use concepts::{storage::TimeoutOutcome, time::ClockFn};
use std::{cmp::min, pin::Pin, time::Duration};
use tracing::{trace, warn};

#[async_trait]
pub trait DeadlineTracker: Send + Sync {
    /// Return a future that resolves on deadline. If `max_duration` is specified, the future resolves
    /// at deadline or after this duration, whatever expires first.
    fn track(
        &self,
        max_duration: Option<Duration>,
    ) -> Option<Pin<Box<dyn Future<Output = TimeoutOutcome> + Send>>>;

    fn close_to_expired(&self) -> bool;

    /// Return new lock expiry date (now + duration). Internally track that time minus leeway.
    fn extend_by(&mut self, lock_extension: Duration) -> DateTime<Utc>;
}

pub trait DeadlineTrackerFactory: Send + Sync {
    fn create(
        &self,
        lock_expires_at: DateTime<Utc>,
    ) -> Result<Box<dyn DeadlineTracker>, LockAlreadyExpired>;
}

#[derive(Debug, thiserror::Error)]
#[error("lock already expired before {started_at}")]
pub struct LockAlreadyExpired {
    pub started_at: DateTime<Utc>,
}

pub(crate) struct DeadlineTrackerTokio<C: ClockFn> {
    pub(crate) deadline: tokio::time::Instant, // Tracked as instant because calling track happens later after creation.
    pub(crate) clock_fn: C,
    pub(crate) leeway: Duration, // Fire this much sooner than requested.
}

#[async_trait]
impl<C: ClockFn> DeadlineTracker for DeadlineTrackerTokio<C> {
    fn track(
        &self,
        max_duration: Option<Duration>,
    ) -> Option<Pin<Box<dyn Future<Output = TimeoutOutcome> + Send>>> {
        if self.close_to_expired() {
            None
        } else {
            let expiry = if let Some(max_duration) = max_duration {
                let max_instant = tokio::time::Instant::now() + max_duration;
                min(max_instant, self.deadline)
            } else {
                self.deadline
            };

            Some(Box::pin(async move {
                tokio::time::sleep_until(expiry).await;
                TimeoutOutcome::Timeout
            }))
        }
    }

    fn close_to_expired(&self) -> bool {
        self.deadline <= tokio::time::Instant::now()
    }

    fn extend_by(&mut self, lock_extension: Duration) -> DateTime<Utc> {
        let now = self.clock_fn.now();
        let lock_duration = if lock_extension > self.leeway {
            lock_extension.checked_sub(self.leeway).unwrap()
        } else {
            warn!("Not setting the leeway as deadline duration is too short");
            lock_extension
        };
        self.deadline = tokio::time::Instant::now() + lock_duration;

        now + lock_extension
    }
}

#[derive(Clone)]
pub struct DeadlineTrackerFactoryTokio<C: ClockFn> {
    pub leeway: Duration, // Fire this much sooner than requested.
    pub clock_fn: C,
}

impl<C: ClockFn> DeadlineTrackerFactory for DeadlineTrackerFactoryTokio<C> {
    fn create(
        &self,
        lock_expires_at: DateTime<Utc>,
    ) -> Result<Box<dyn DeadlineTracker>, LockAlreadyExpired> {
        let started_at = self.clock_fn.now();
        let Ok(deadline_duration) = (lock_expires_at - started_at).to_std() else {
            return Err(LockAlreadyExpired { started_at });
        };
        let deadline_duration = if deadline_duration > self.leeway {
            deadline_duration.checked_sub(self.leeway).unwrap()
        } else {
            warn!("Not setting the leeway as deadline duration is too short");
            deadline_duration
        };
        trace!("Setting deadline to now + {deadline_duration:?}");
        let deadline = tokio::time::Instant::now() + deadline_duration;
        let tracker = DeadlineTrackerTokio {
            deadline,
            clock_fn: self.clock_fn.clone(),
            leeway: self.leeway,
        };
        Ok(Box::new(tracker))
    }
}

#[cfg(test)]
#[must_use]
pub fn deadline_tracker_factory_test(
    sim_clock: test_utils::sim_clock::SimClock,
) -> std::sync::Arc<impl DeadlineTrackerFactory> {
    std::sync::Arc::new(DeadlineTrackerFactoryTokio {
        leeway: Duration::ZERO,
        clock_fn: sim_clock,
    })
}