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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169

use tokio::time::Duration;

use crate::UpstreamError;
use smallvec::SmallVec;

pub(crate) enum InnerCommitPolicy {
    Noop,
    Immediate(ImmediateCommitWaiter),
    Within(Duration),
}

#[derive(Clone, Copy)]
pub enum DefaultCommitPolicy {
    /// The data must be persisted successfully before the execution returns.
    Immediate,
    /// The data should try to be persisted with at most a `Duration` delay.
    Within(Duration),
}

pub(super) enum CommitPolicy {
    /// The data should persist with the default persistence policy of the service.
    Default,
    /// The mutation was a no-op, and no additional persistence should be enqueued.
    Noop,
    /// The data must be persisted successfully before the execution returns.
    Immediate,
    /// The data should try to be persisted with at most a `Duration` delay.
    Within(Duration),
}

pub(super) enum CommitPolicyNoDefault {
    Noop,
    Immediate,
    Within(Duration),
}

impl CommitPolicy {
    pub(super) fn apply_default(
        self,
        default_commit_policy: DefaultCommitPolicy,
    ) -> CommitPolicyNoDefault {
        match self {
            CommitPolicy::Default => default_commit_policy.into(),
            CommitPolicy::Noop => CommitPolicyNoDefault::Noop,
            CommitPolicy::Within(duration) => CommitPolicyNoDefault::Within(duration),
            CommitPolicy::Immediate => CommitPolicyNoDefault::Immediate,
        }
    }
}

/// Specifies how the `execute_mut` should be committed to the upstream.
///
/// This allows you to control the durability of your data mutation.
///
/// [`execute_mut`]: ServiceHandle::execute_mut
pub struct Commit<Data>(CommitPolicy, Data);

impl<Data> Commit<Data> {
    /// Persist the item immediately. The execution future will not resolve until your write has been been
    /// acknowledged by the upstream layer.
    pub fn immediately(data: Data) -> Self {
        Self(CommitPolicy::Immediate, data)
    }

    /// Persist the data, using the default persistence policy for the service.
    pub fn default(data: Data) -> Self {
        Self(CommitPolicy::Default, data)
    }

    /// Ensure that the data is persisted within a given duration.
    pub fn within(data: Data, duration: Duration) -> Self {
        Self(CommitPolicy::Within(duration), data)
    }

    /// The mutation resulted in a no-op. Do not try to persist the data in any way.
    ///
    /// # Safety
    ///
    /// This function is unsafe, as it's up to you to make sure to only call this *if and only if*
    /// the data truly did not change. Since you're getting a `&mut` to the data, this contract
    /// cannot be statically enforced. So, it's up to you to use correctly.
    pub unsafe fn noop(data: Data) -> Self {
        Self(CommitPolicy::Noop, data)
    }

    pub(super) fn into_inner(self) -> (CommitPolicy, Data) {
        (self.0, self.1)
    }
}

impl Into<CommitPolicyNoDefault> for DefaultCommitPolicy {
    fn into(self) -> CommitPolicyNoDefault {
        match self {
            DefaultCommitPolicy::Within(duration) => CommitPolicyNoDefault::Within(duration),
            DefaultCommitPolicy::Immediate => CommitPolicyNoDefault::Immediate,
        }
    }
}

pub(super) type ImmediateCommitWaiter = Box<dyn FnOnce(Result<(), UpstreamError>) + Send + 'static>;

pub(super) type ImmediateCommitWaiters = SmallVec<[ImmediateCommitWaiter; 1]>;

pub(super) enum AccumulatedCommitPolicy {
    Noop,
    Within(Duration),
    Immediate(ImmediateCommitWaiters),
}

impl From<InnerCommitPolicy> for AccumulatedCommitPolicy {
    fn from(commit_policy: InnerCommitPolicy) -> Self {
        match commit_policy {
            InnerCommitPolicy::Within(duration) => AccumulatedCommitPolicy::Within(duration),
            InnerCommitPolicy::Immediate(callback) => callback.into(),
            InnerCommitPolicy::Noop => AccumulatedCommitPolicy::Noop,
        }
    }
}

impl From<ImmediateCommitWaiter> for AccumulatedCommitPolicy {
    fn from(waiter: ImmediateCommitWaiter) -> Self {
        let mut vec = SmallVec::new();
        vec.push(waiter);
        AccumulatedCommitPolicy::Immediate(vec)
    }
}

impl AccumulatedCommitPolicy {
    pub(super) fn new() -> Self {
        AccumulatedCommitPolicy::Noop
    }

    pub(super) fn did_mutate(&self) -> bool {
        match self {
            AccumulatedCommitPolicy::Noop => false,
            AccumulatedCommitPolicy::Within(_) | AccumulatedCommitPolicy::Immediate(_) => true,
        }
    }

    pub(super) fn accumulate(self, commit_policy: InnerCommitPolicy) -> Self {
        match commit_policy {
            InnerCommitPolicy::Within(duration) => self.accumulate_commit_within(duration),
            InnerCommitPolicy::Immediate(waiter) => self.accumulate_commit_immediate(waiter),
            InnerCommitPolicy::Noop => self,
        }
    }

    fn accumulate_commit_within(self, duration: Duration) -> Self {
        match self {
            AccumulatedCommitPolicy::Noop => AccumulatedCommitPolicy::Within(duration),
            AccumulatedCommitPolicy::Within(existing_duration) => {
                // We want to keep the minimum of the two durations.
                AccumulatedCommitPolicy::Within(std::cmp::min(existing_duration, duration))
            }
            // There's nothing faster than immediate.
            AccumulatedCommitPolicy::Immediate(_) => self,
        }
    }

    fn accumulate_commit_immediate(self, waiter: ImmediateCommitWaiter) -> Self {
        match self {
            AccumulatedCommitPolicy::Noop | AccumulatedCommitPolicy::Within(_) => waiter.into(),
            AccumulatedCommitPolicy::Immediate(mut waiters) => {
                waiters.push(waiter);
                AccumulatedCommitPolicy::Immediate(waiters)
            }
        }
    }
}