1pub trait SynchronizedViewRevision: Copy + Ord {
2 type ChangeId: Copy + Ord;
3
4 fn changed_through(change_id: Self::ChangeId) -> Self;
5
6 fn next_change_id(self) -> Option<Self::ChangeId>;
7}
8
9pub enum SynchronizedViewSnapshotRequiredReason<Revision, Rejection> {
10 NoSnapshotApplied,
11 ChangeGap { synchronized_through: Revision },
12 ChangeRejected(Rejection),
13 AlreadyRequired,
14}
15
16#[must_use]
17pub enum ApplySynchronizedViewSnapshotOutcome<Revision> {
18 Synchronized {
19 through: Revision,
20 },
21 Stale {
22 snapshot_through: Revision,
23 required_through: Revision,
24 },
25}
26
27#[must_use]
28pub enum ApplySynchronizedViewChangeOutcome<Revision, Rejection> {
29 Applied {
30 through: Revision,
31 },
32 Stale {
33 change_through: Revision,
34 synchronized_through: Revision,
35 },
36 SnapshotRequired {
37 observed_through: Revision,
38 reason: SynchronizedViewSnapshotRequiredReason<Revision, Rejection>,
39 },
40}
41
42pub enum ApplySynchronizedViewRevisionSpanOutcome<Revision, Rejection> {
43 Applied {
44 through: Revision,
45 },
46 Stale {
47 span_through: Revision,
48 synchronized_through: Revision,
49 },
50 SnapshotRequired {
51 observed_through: Revision,
52 reason: SynchronizedViewSnapshotRequiredReason<Revision, Rejection>,
53 },
54}
55
56pub enum SynchronizedViewRead<'a, Revision, Projection> {
57 AwaitingSnapshot {
58 observed_through: Revision,
59 },
60 Synchronized {
61 through: Revision,
62 projection: &'a Projection,
63 },
64 SynchronizationRequired {
65 synchronized_through: Revision,
66 observed_through: Revision,
67 projection: &'a Projection,
68 },
69}
70
71enum SynchronizedViewState<Revision, Projection> {
72 AwaitingSnapshot {
73 observed_through: Revision,
74 },
75 Observing {
76 synchronized_through: Revision,
77 observed_through: Revision,
78 synchronization: SynchronizedViewSynchronization,
79 projection: Projection,
80 },
81}
82
83enum SynchronizedViewSynchronization {
84 Synchronized,
85 Required,
86}
87
88pub struct SynchronizedView<Revision, Projection> {
89 state: SynchronizedViewState<Revision, Projection>,
90}
91
92impl<Revision, Projection> SynchronizedView<Revision, Projection>
93where
94 Revision: Copy + Ord,
95{
96 pub const fn new(initial_revision: Revision) -> Self {
97 Self {
98 state: SynchronizedViewState::AwaitingSnapshot {
99 observed_through: initial_revision,
100 },
101 }
102 }
103
104 pub fn apply_snapshot(
105 &mut self,
106 snapshot_through: Revision,
107 projection: Projection,
108 ) -> ApplySynchronizedViewSnapshotOutcome<Revision> {
109 let required_through = self.required_snapshot_through();
110 if snapshot_through < required_through {
111 return ApplySynchronizedViewSnapshotOutcome::Stale {
112 snapshot_through,
113 required_through,
114 };
115 }
116 self.state = SynchronizedViewState::Observing {
117 synchronized_through: snapshot_through,
118 observed_through: snapshot_through,
119 synchronization: SynchronizedViewSynchronization::Synchronized,
120 projection,
121 };
122 ApplySynchronizedViewSnapshotOutcome::Synchronized {
123 through: snapshot_through,
124 }
125 }
126
127 pub fn required_snapshot_through(&self) -> Revision {
128 match &self.state {
129 SynchronizedViewState::AwaitingSnapshot { observed_through }
130 | SynchronizedViewState::Observing {
131 observed_through, ..
132 } => *observed_through,
133 }
134 }
135
136 pub fn read(&self) -> SynchronizedViewRead<'_, Revision, Projection> {
137 match &self.state {
138 SynchronizedViewState::AwaitingSnapshot { observed_through } => {
139 SynchronizedViewRead::AwaitingSnapshot {
140 observed_through: *observed_through,
141 }
142 }
143 SynchronizedViewState::Observing {
144 synchronized_through,
145 observed_through: _,
146 synchronization: SynchronizedViewSynchronization::Synchronized,
147 projection,
148 } => SynchronizedViewRead::Synchronized {
149 through: *synchronized_through,
150 projection,
151 },
152 SynchronizedViewState::Observing {
153 synchronized_through,
154 observed_through,
155 synchronization: SynchronizedViewSynchronization::Required,
156 projection,
157 } => SynchronizedViewRead::SynchronizationRequired {
158 synchronized_through: *synchronized_through,
159 observed_through: *observed_through,
160 projection,
161 },
162 }
163 }
164
165 pub fn require_snapshot_through(&mut self, required_through: Revision) {
166 match &mut self.state {
167 SynchronizedViewState::AwaitingSnapshot { observed_through } => {
168 if required_through > *observed_through {
169 *observed_through = required_through;
170 }
171 }
172 SynchronizedViewState::Observing {
173 observed_through,
174 synchronization,
175 ..
176 } => {
177 if required_through > *observed_through {
178 *observed_through = required_through;
179 }
180 *synchronization = SynchronizedViewSynchronization::Required;
181 }
182 }
183 }
184
185 pub fn apply_revision_span<Prepared, Rejection, Prepare, Reserve, Commit>(
186 &mut self,
187 expected: Revision,
188 through: Revision,
189 prepare: Prepare,
190 reserve: Reserve,
191 commit: Commit,
192 ) -> ApplySynchronizedViewRevisionSpanOutcome<Revision, Rejection>
193 where
194 Prepare: FnOnce(&Projection) -> Result<Prepared, Rejection>,
195 Reserve: FnOnce(&mut Projection, &Prepared) -> Result<(), Rejection>,
196 Commit: FnOnce(&mut Projection, Prepared),
197 {
198 match &mut self.state {
199 SynchronizedViewState::AwaitingSnapshot { observed_through } => {
200 if through > *observed_through {
201 *observed_through = through;
202 }
203 ApplySynchronizedViewRevisionSpanOutcome::SnapshotRequired {
204 observed_through: *observed_through,
205 reason: SynchronizedViewSnapshotRequiredReason::NoSnapshotApplied,
206 }
207 }
208 SynchronizedViewState::Observing {
209 synchronized_through,
210 ..
211 } if through <= *synchronized_through => {
212 ApplySynchronizedViewRevisionSpanOutcome::Stale {
213 span_through: through,
214 synchronized_through: *synchronized_through,
215 }
216 }
217 SynchronizedViewState::Observing {
218 observed_through,
219 synchronization: SynchronizedViewSynchronization::Required,
220 ..
221 } => {
222 if through > *observed_through {
223 *observed_through = through;
224 }
225 ApplySynchronizedViewRevisionSpanOutcome::SnapshotRequired {
226 observed_through: *observed_through,
227 reason: SynchronizedViewSnapshotRequiredReason::AlreadyRequired,
228 }
229 }
230 SynchronizedViewState::Observing {
231 synchronized_through,
232 observed_through,
233 synchronization,
234 projection,
235 } => {
236 if expected != *synchronized_through || through <= expected {
237 if through > *observed_through {
238 *observed_through = through;
239 }
240 *synchronization = SynchronizedViewSynchronization::Required;
241 return ApplySynchronizedViewRevisionSpanOutcome::SnapshotRequired {
242 observed_through: *observed_through,
243 reason: SynchronizedViewSnapshotRequiredReason::ChangeGap {
244 synchronized_through: *synchronized_through,
245 },
246 };
247 }
248 match prepare(projection).and_then(|prepared| {
249 reserve(projection, &prepared)?;
250 Ok(prepared)
251 }) {
252 Ok(prepared) => {
253 commit(projection, prepared);
254 *synchronized_through = through;
255 *observed_through = through;
256 ApplySynchronizedViewRevisionSpanOutcome::Applied { through }
257 }
258 Err(rejection) => {
259 *observed_through = through;
260 *synchronization = SynchronizedViewSynchronization::Required;
261 ApplySynchronizedViewRevisionSpanOutcome::SnapshotRequired {
262 observed_through: through,
263 reason: SynchronizedViewSnapshotRequiredReason::ChangeRejected(
264 rejection,
265 ),
266 }
267 }
268 }
269 }
270 }
271 }
272}
273
274impl<Revision, Projection> SynchronizedView<Revision, Projection>
275where
276 Revision: SynchronizedViewRevision,
277{
278 pub fn apply_change<Prepared, Rejection, Prepare, Reserve, Commit>(
279 &mut self,
280 change_id: Revision::ChangeId,
281 prepare: Prepare,
282 reserve: Reserve,
283 commit: Commit,
284 ) -> ApplySynchronizedViewChangeOutcome<Revision, Rejection>
285 where
286 Prepare: FnOnce(&Projection) -> Result<Prepared, Rejection>,
287 Reserve: FnOnce(&mut Projection, &Prepared) -> Result<(), Rejection>,
288 Commit: FnOnce(&mut Projection, Prepared),
289 {
290 let change_through = Revision::changed_through(change_id);
291 match &mut self.state {
292 SynchronizedViewState::AwaitingSnapshot { observed_through } => {
293 if change_through > *observed_through {
294 *observed_through = change_through;
295 }
296 ApplySynchronizedViewChangeOutcome::SnapshotRequired {
297 observed_through: *observed_through,
298 reason: SynchronizedViewSnapshotRequiredReason::NoSnapshotApplied,
299 }
300 }
301 SynchronizedViewState::Observing {
302 synchronized_through,
303 ..
304 } if change_through <= *synchronized_through => {
305 ApplySynchronizedViewChangeOutcome::Stale {
306 change_through,
307 synchronized_through: *synchronized_through,
308 }
309 }
310 SynchronizedViewState::Observing {
311 synchronized_through: _,
312 observed_through,
313 synchronization: SynchronizedViewSynchronization::Required,
314 ..
315 } => {
316 if change_through > *observed_through {
317 *observed_through = change_through;
318 }
319 ApplySynchronizedViewChangeOutcome::SnapshotRequired {
320 observed_through: *observed_through,
321 reason: SynchronizedViewSnapshotRequiredReason::AlreadyRequired,
322 }
323 }
324 SynchronizedViewState::Observing {
325 synchronized_through,
326 observed_through,
327 synchronization,
328 projection,
329 } => {
330 if synchronized_through.next_change_id() != Some(change_id) {
331 *observed_through = change_through;
332 *synchronization = SynchronizedViewSynchronization::Required;
333 return ApplySynchronizedViewChangeOutcome::SnapshotRequired {
334 observed_through: change_through,
335 reason: SynchronizedViewSnapshotRequiredReason::ChangeGap {
336 synchronized_through: *synchronized_through,
337 },
338 };
339 }
340 match prepare(projection).and_then(|prepared| {
341 reserve(projection, &prepared)?;
342 Ok(prepared)
343 }) {
344 Ok(prepared) => {
345 commit(projection, prepared);
346 *synchronized_through = change_through;
347 *observed_through = change_through;
348 ApplySynchronizedViewChangeOutcome::Applied {
349 through: change_through,
350 }
351 }
352 Err(rejection) => {
353 *observed_through = change_through;
354 *synchronization = SynchronizedViewSynchronization::Required;
355 ApplySynchronizedViewChangeOutcome::SnapshotRequired {
356 observed_through: change_through,
357 reason: SynchronizedViewSnapshotRequiredReason::ChangeRejected(
358 rejection,
359 ),
360 }
361 }
362 }
363 }
364 }
365 }
366}
367
368#[cfg(test)]
369mod tests;
370
371#[cfg(kani)]
372mod proofs;