eredu_runtime/cache/
storage.rs

1//! Backend-neutral cache storage transition protocol.
2
3use serde::{Deserialize, Serialize};
4
5use eredu_core::cache::{CacheBlockId, CacheTier};
6
7/// Stable phase of one cache block's physical resources.
8#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
9#[serde(rename_all = "snake_case")]
10pub enum CacheStoragePhase {
11    /// The execution backend owns device resources.
12    Device,
13    /// Device resources are retained while a host copy is produced.
14    DemotingToHost,
15    /// Host resources exist without a durable backing.
16    HostUnbacked,
17    /// Host resources are retained by an exact disk-write operation.
18    HostWriting,
19    /// Host resources have a durable backing.
20    HostBacked,
21    /// Only the durable backing is resident.
22    DiskReady,
23    /// A disk-read operation owns the backing and an admitted host allocation.
24    DiskReading,
25}
26
27impl CacheStoragePhase {
28    /// Logical accounting tier for the phase.
29    pub const fn tier(self) -> CacheTier {
30        match self {
31            Self::Device | Self::DemotingToHost => CacheTier::Device,
32            Self::HostUnbacked | Self::HostWriting | Self::HostBacked => CacheTier::Host,
33            Self::DiskReady | Self::DiskReading => CacheTier::Disk,
34        }
35    }
36}
37
38/// Kind of asynchronous backing-store operation.
39#[derive(Debug, Clone, Copy, Eq, Hash, PartialEq, Serialize, Deserialize)]
40#[serde(rename_all = "snake_case")]
41pub enum CacheIoOperationKind {
42    /// Publish host resources to a durable backing.
43    Write,
44    /// Reconstruct host resources from a durable backing.
45    Read,
46}
47
48/// Exact identity of one backing-store operation.
49#[derive(Debug, Clone, Eq, Hash, PartialEq, Serialize, Deserialize)]
50pub struct CacheIoOperationKey {
51    /// Cache generation in which the operation was submitted.
52    pub generation: u64,
53    /// Block whose resources the operation owns.
54    pub id: CacheBlockId,
55    /// Direction of the operation.
56    pub kind: CacheIoOperationKind,
57}
58
59/// Opaque backend host-demotion completion with stable identity.
60pub trait CacheHostDemotionOperation {
61    /// Block whose device resources the operation owns.
62    fn block_id(&self) -> &CacheBlockId;
63
64    /// Monotonic operation identity within the backend cache session.
65    fn operation_id(&self) -> u64;
66}
67
68/// Opaque backend I/O completion with an exact neutral key.
69pub trait CacheIoOperation {
70    /// Returns the operation identity owned by this completion.
71    fn key(&self) -> &CacheIoOperationKey;
72}
73
74/// Exact rollback ownership returned by a host-to-device promotion.
75#[derive(Debug)]
76pub struct CacheHostPromotion<H> {
77    id: CacheBlockId,
78    source: CacheStoragePhase,
79    host: H,
80}
81
82/// Canonical physical-resource state machine for one cache block.
83///
84/// `D`, `H`, and `B` are backend-owned device, host, and backing resources.
85/// `HD` and `IO` are backend-owned exact completions. Private resource slots
86/// prevent a backend from constructing contradictory phase/resource states.
87#[derive(Debug, Clone)]
88pub struct CacheBlockStorage<D, H, B, HD, IO> {
89    id: CacheBlockId,
90    phase: CacheStoragePhase,
91    device: Option<D>,
92    host: Option<H>,
93    backing: Option<B>,
94    host_demotion: Option<HD>,
95    io: Option<IO>,
96}
97
98impl<D, H, B, HD, IO> CacheBlockStorage<D, H, B, HD, IO> {
99    /// Creates device residency, optionally retaining an existing backing.
100    pub fn device(id: CacheBlockId, device: D, backing: Option<B>) -> Self {
101        Self {
102            id,
103            phase: CacheStoragePhase::Device,
104            device: Some(device),
105            host: None,
106            backing,
107            host_demotion: None,
108            io: None,
109        }
110    }
111
112    /// Creates host residency, optionally retaining an existing backing.
113    pub fn host(id: CacheBlockId, host: H, backing: Option<B>) -> Self {
114        Self {
115            id,
116            phase: if backing.is_some() {
117                CacheStoragePhase::HostBacked
118            } else {
119                CacheStoragePhase::HostUnbacked
120            },
121            device: None,
122            host: Some(host),
123            backing,
124            host_demotion: None,
125            io: None,
126        }
127    }
128
129    /// Creates ready backing-only residency.
130    pub fn disk(id: CacheBlockId, backing: B) -> Self {
131        Self {
132            id,
133            phase: CacheStoragePhase::DiskReady,
134            device: None,
135            host: None,
136            backing: Some(backing),
137            host_demotion: None,
138            io: None,
139        }
140    }
141
142    /// Current canonical phase.
143    pub const fn phase(&self) -> CacheStoragePhase {
144        self.phase
145    }
146
147    /// Block whose physical resources are owned by this state machine.
148    pub const fn id(&self) -> &CacheBlockId {
149        &self.id
150    }
151
152    /// Current logical accounting tier.
153    pub const fn tier(&self) -> CacheTier {
154        self.phase.tier()
155    }
156
157    /// Backend device resources, including while a demotion is pending.
158    pub fn device_resource(&self) -> Option<&D> {
159        self.device.as_ref()
160    }
161
162    /// Backend host resources.
163    pub fn host_resource(&self) -> Option<&H> {
164        self.host.as_ref()
165    }
166
167    /// Durable backing retained by the current phase.
168    pub fn backing(&self) -> Option<&B> {
169        self.backing.as_ref()
170    }
171
172    /// Exact pending host-demotion completion.
173    pub fn host_demotion(&self) -> Option<&HD> {
174        self.host_demotion.as_ref()
175    }
176
177    /// Exact pending backing-store completion.
178    pub fn io(&self) -> Option<&IO> {
179        self.io.as_ref()
180    }
181}
182
183impl<D, H, B, HD: CacheHostDemotionOperation, IO: CacheIoOperation>
184    CacheBlockStorage<D, H, B, HD, IO>
185{
186    /// Returns whether the exact backing-store operation is pending.
187    pub fn io_matches(&self, key: &CacheIoOperationKey) -> bool {
188        self.io.as_ref().is_some_and(|io| io.key() == key)
189    }
190
191    /// Fails the operation only when the exact key is still pending.
192    pub fn fail_io_if_matches(&mut self, key: &CacheIoOperationKey) -> bool {
193        if self.io_matches(key) {
194            self.fail_io(key)
195                .expect("matching pending I/O has a valid source phase");
196            true
197        } else {
198            false
199        }
200    }
201
202    /// Begins an exact device-to-host transition while retaining device state.
203    pub fn begin_host_demotion(&mut self, operation: HD) -> Result<(), CacheStorageError> {
204        self.require_phase(CacheStoragePhase::Device)?;
205        self.require_block(operation.block_id())?;
206        if self.backing.is_some() {
207            return Err(CacheStorageError::BackingAlreadyExists);
208        }
209        self.host_demotion = Some(operation);
210        self.phase = CacheStoragePhase::DemotingToHost;
211        Ok(())
212    }
213
214    /// Commits the matching host demotion and returns released device resources.
215    pub fn finish_host_demotion(
216        &mut self,
217        operation_id: u64,
218        host: H,
219    ) -> Result<(D, HD), CacheStorageError> {
220        self.require_host_demotion(operation_id)?;
221        let device = self
222            .device
223            .take()
224            .expect("demoting phase retains device resources");
225        let operation = self
226            .host_demotion
227            .take()
228            .expect("demoting phase retains its exact operation");
229        self.host = Some(host);
230        self.phase = CacheStoragePhase::HostUnbacked;
231        Ok((device, operation))
232    }
233
234    /// Abandons the matching host demotion and restores device residency.
235    pub fn fail_host_demotion(&mut self, operation_id: u64) -> Result<HD, CacheStorageError> {
236        self.require_host_demotion(operation_id)?;
237        let operation = self
238            .host_demotion
239            .take()
240            .expect("demoting phase retains its exact operation");
241        self.phase = CacheStoragePhase::Device;
242        Ok(operation)
243    }
244
245    /// Releases backed device resources directly to backing-only residency.
246    pub fn release_device_to_disk(&mut self) -> Result<D, CacheStorageError> {
247        self.require_phase(CacheStoragePhase::Device)?;
248        if self.backing.is_none() {
249            return Err(CacheStorageError::BackingRequired);
250        }
251        let device = self
252            .device
253            .take()
254            .expect("device phase owns device resources");
255        self.phase = CacheStoragePhase::DiskReady;
256        Ok(device)
257    }
258
259    /// Releases backed host resources directly to backing-only residency.
260    pub fn release_host_to_disk(&mut self) -> Result<H, CacheStorageError> {
261        self.require_phase(CacheStoragePhase::HostBacked)?;
262        let host = self
263            .host
264            .take()
265            .expect("host-backed phase owns host resources");
266        self.phase = CacheStoragePhase::DiskReady;
267        Ok(host)
268    }
269
270    /// Starts the exact write that owns unbacked host resources.
271    pub fn begin_write(&mut self, operation: IO) -> Result<(), CacheStorageError> {
272        self.require_phase(CacheStoragePhase::HostUnbacked)?;
273        self.require_block(&operation.key().id)?;
274        self.require_kind(operation.key(), CacheIoOperationKind::Write)?;
275        self.io = Some(operation);
276        self.phase = CacheStoragePhase::HostWriting;
277        Ok(())
278    }
279
280    /// Commits the matching write and returns released host resources.
281    pub fn finish_write(
282        &mut self,
283        key: &CacheIoOperationKey,
284        backing: B,
285    ) -> Result<(H, IO), CacheStorageError> {
286        self.require_io(CacheStoragePhase::HostWriting, key)?;
287        let host = self
288            .host
289            .take()
290            .expect("host-writing phase retains host resources");
291        let operation = self.io.take().expect("host-writing phase retains I/O");
292        self.backing = Some(backing);
293        self.phase = CacheStoragePhase::DiskReady;
294        Ok((host, operation))
295    }
296
297    /// Starts the exact read that owns backing-only resources.
298    pub fn begin_read(&mut self, operation: IO) -> Result<(), CacheStorageError> {
299        self.require_phase(CacheStoragePhase::DiskReady)?;
300        self.require_block(&operation.key().id)?;
301        self.require_kind(operation.key(), CacheIoOperationKind::Read)?;
302        self.io = Some(operation);
303        self.phase = CacheStoragePhase::DiskReading;
304        Ok(())
305    }
306
307    /// Commits the matching read while retaining its durable backing.
308    pub fn finish_read(
309        &mut self,
310        key: &CacheIoOperationKey,
311        host: H,
312    ) -> Result<IO, CacheStorageError> {
313        self.require_io(CacheStoragePhase::DiskReading, key)?;
314        let operation = self.io.take().expect("disk-reading phase retains I/O");
315        self.host = Some(host);
316        self.phase = CacheStoragePhase::HostBacked;
317        Ok(operation)
318    }
319
320    /// Cancels or fails the matching I/O and restores its stable source phase.
321    pub fn fail_io(&mut self, key: &CacheIoOperationKey) -> Result<IO, CacheStorageError> {
322        let stable = match self.phase {
323            CacheStoragePhase::HostWriting => CacheStoragePhase::HostUnbacked,
324            CacheStoragePhase::DiskReading => CacheStoragePhase::DiskReady,
325            actual => return Err(CacheStorageError::IoNotPending { actual }),
326        };
327        self.require_io(self.phase, key)?;
328        let operation = self.io.take().expect("pending phase retains I/O");
329        self.phase = stable;
330        Ok(operation)
331    }
332
333    /// Promotes stable host resources to device residency.
334    pub fn promote_host(&mut self, device: D) -> Result<CacheHostPromotion<H>, CacheStorageError> {
335        if !matches!(
336            self.phase,
337            CacheStoragePhase::HostUnbacked | CacheStoragePhase::HostBacked
338        ) {
339            return Err(CacheStorageError::InvalidPhase {
340                expected: CacheStoragePhase::HostUnbacked,
341                actual: self.phase,
342            });
343        }
344        let source = self.phase;
345        let host = self
346            .host
347            .take()
348            .expect("stable host phase owns host resources");
349        self.device = Some(device);
350        self.phase = CacheStoragePhase::Device;
351        Ok(CacheHostPromotion {
352            id: self.id.clone(),
353            source,
354            host,
355        })
356    }
357
358    /// Restores host resources after a rejected promotion.
359    pub fn restore_host(
360        &mut self,
361        promotion: CacheHostPromotion<H>,
362    ) -> Result<D, CacheStorageError> {
363        self.require_phase(CacheStoragePhase::Device)?;
364        self.require_block(&promotion.id)?;
365        let expected_source = if self.backing.is_some() {
366            CacheStoragePhase::HostBacked
367        } else {
368            CacheStoragePhase::HostUnbacked
369        };
370        if promotion.source != expected_source {
371            return Err(CacheStorageError::InvalidPhase {
372                expected: expected_source,
373                actual: promotion.source,
374            });
375        }
376        let device = self
377            .device
378            .take()
379            .expect("device phase owns device resources");
380        self.host = Some(promotion.host);
381        self.phase = promotion.source;
382        Ok(device)
383    }
384
385    fn require_phase(&self, expected: CacheStoragePhase) -> Result<(), CacheStorageError> {
386        if self.phase == expected {
387            Ok(())
388        } else {
389            Err(CacheStorageError::InvalidPhase {
390                expected,
391                actual: self.phase,
392            })
393        }
394    }
395
396    fn require_host_demotion(&self, operation_id: u64) -> Result<(), CacheStorageError> {
397        self.require_phase(CacheStoragePhase::DemotingToHost)?;
398        let actual = self
399            .host_demotion
400            .as_ref()
401            .expect("demoting phase retains its exact operation")
402            .operation_id();
403        if actual == operation_id {
404            Ok(())
405        } else {
406            Err(CacheStorageError::HostDemotionMismatch {
407                expected: actual,
408                actual: operation_id,
409            })
410        }
411    }
412
413    fn require_io(
414        &self,
415        phase: CacheStoragePhase,
416        key: &CacheIoOperationKey,
417    ) -> Result<(), CacheStorageError> {
418        self.require_phase(phase)?;
419        let expected = self
420            .io
421            .as_ref()
422            .expect("pending I/O phase retains its exact operation")
423            .key();
424        if expected == key {
425            Ok(())
426        } else {
427            Err(CacheStorageError::IoMismatch {
428                expected: Box::new(expected.clone()),
429                actual: Box::new(key.clone()),
430            })
431        }
432    }
433
434    fn require_kind(
435        &self,
436        key: &CacheIoOperationKey,
437        expected: CacheIoOperationKind,
438    ) -> Result<(), CacheStorageError> {
439        if key.kind == expected {
440            Ok(())
441        } else {
442            Err(CacheStorageError::IoKindMismatch {
443                expected,
444                actual: key.kind,
445            })
446        }
447    }
448
449    fn require_block(&self, actual: &CacheBlockId) -> Result<(), CacheStorageError> {
450        if actual == &self.id {
451            Ok(())
452        } else {
453            Err(CacheStorageError::BlockMismatch {
454                expected: Box::new(self.id.clone()),
455                actual: Box::new(actual.clone()),
456            })
457        }
458    }
459}
460
461/// Illegal cache storage phase transition or completion observation.
462#[derive(Debug, Clone, Eq, PartialEq, thiserror::Error)]
463pub enum CacheStorageError {
464    /// An operation belongs to a different cache block.
465    #[error("cache storage block mismatch: expected {expected:?}, observed {actual:?}")]
466    BlockMismatch {
467        /// Block owned by the state machine.
468        expected: Box<CacheBlockId>,
469        /// Block owned by the presented operation.
470        actual: Box<CacheBlockId>,
471    },
472    /// An operation is not legal from the current phase.
473    #[error("cache storage phase is {actual:?}, expected {expected:?}")]
474    InvalidPhase {
475        /// Required source phase.
476        expected: CacheStoragePhase,
477        /// Observed source phase.
478        actual: CacheStoragePhase,
479    },
480    /// No backing-store operation is pending in the current phase.
481    #[error("cache storage phase {actual:?} has no pending I/O")]
482    IoNotPending {
483        /// Observed stable or unrelated phase.
484        actual: CacheStoragePhase,
485    },
486    /// A device block already had a durable backing.
487    #[error("cache storage already has a durable backing")]
488    BackingAlreadyExists,
489    /// A direct release required a durable backing.
490    #[error("cache storage requires a durable backing")]
491    BackingRequired,
492    /// A different host-demotion completion attempted to resolve the phase.
493    #[error("host demotion operation mismatch: expected {expected}, observed {actual}")]
494    HostDemotionMismatch {
495        /// Operation retained by the state machine.
496        expected: u64,
497        /// Operation presented by the backend.
498        actual: u64,
499    },
500    /// A different backing-store completion attempted to resolve the phase.
501    #[error("cache I/O operation mismatch: expected {expected:?}, observed {actual:?}")]
502    IoMismatch {
503        /// Operation retained by the state machine.
504        expected: Box<CacheIoOperationKey>,
505        /// Operation presented by the backend.
506        actual: Box<CacheIoOperationKey>,
507    },
508    /// A read/write completion was submitted to the opposite transition.
509    #[error("cache I/O kind is {actual:?}, expected {expected:?}")]
510    IoKindMismatch {
511        /// Required operation direction.
512        expected: CacheIoOperationKind,
513        /// Observed operation direction.
514        actual: CacheIoOperationKind,
515    },
516}
517
518#[cfg(test)]
519mod tests {
520    use super::*;
521    use eredu_core::cache::CacheRepresentation;
522
523    #[derive(Debug, Clone, Eq, PartialEq)]
524    struct HostOp {
525        id: CacheBlockId,
526        operation_id: u64,
527    }
528
529    impl CacheHostDemotionOperation for HostOp {
530        fn block_id(&self) -> &CacheBlockId {
531            &self.id
532        }
533
534        fn operation_id(&self) -> u64 {
535            self.operation_id
536        }
537    }
538
539    #[derive(Debug, Clone, Eq, PartialEq)]
540    struct IoOp(CacheIoOperationKey);
541
542    impl CacheIoOperation for IoOp {
543        fn key(&self) -> &CacheIoOperationKey {
544            &self.0
545        }
546    }
547
548    fn block() -> CacheBlockId {
549        CacheBlockId {
550            session_id: 1,
551            global_layer: 2,
552            representation: CacheRepresentation::KeyValue,
553            start: 0,
554            end: 4,
555            rank: None,
556        }
557    }
558
559    fn io(kind: CacheIoOperationKind, generation: u64) -> IoOp {
560        IoOp(CacheIoOperationKey {
561            generation,
562            id: block(),
563            kind,
564        })
565    }
566
567    #[test]
568    fn exact_host_demotion_commits_or_rolls_back_without_losing_device_state() {
569        let id = block();
570        let mut storage = CacheBlockStorage::<_, String, String, _, IoOp>::device(
571            id.clone(),
572            "device".to_owned(),
573            None,
574        );
575        storage
576            .begin_host_demotion(HostOp {
577                id,
578                operation_id: 7,
579            })
580            .unwrap();
581        assert_eq!(storage.phase(), CacheStoragePhase::DemotingToHost);
582        assert_eq!(
583            storage.finish_host_demotion(8, "host".to_owned()),
584            Err(CacheStorageError::HostDemotionMismatch {
585                expected: 7,
586                actual: 8,
587            })
588        );
589        assert_eq!(
590            storage.device_resource().map(String::as_str),
591            Some("device")
592        );
593        storage.fail_host_demotion(7).unwrap();
594        assert_eq!(storage.phase(), CacheStoragePhase::Device);
595    }
596
597    #[test]
598    fn write_read_and_promotion_follow_one_exact_transaction_chain() {
599        let mut storage = CacheBlockStorage::<String, String, String, HostOp, _>::host(
600            block(),
601            "host".to_owned(),
602            None,
603        );
604        let write = io(CacheIoOperationKind::Write, 3);
605        storage.begin_write(write.clone()).unwrap();
606        let stale = io(CacheIoOperationKind::Write, 2);
607        assert!(matches!(
608            storage.finish_write(stale.key(), "disk".to_owned()),
609            Err(CacheStorageError::IoMismatch { .. })
610        ));
611        let (host, observed) = storage
612            .finish_write(write.key(), "disk".to_owned())
613            .unwrap();
614        assert_eq!(host, "host");
615        assert_eq!(observed, write);
616        assert_eq!(storage.phase(), CacheStoragePhase::DiskReady);
617
618        let read = io(CacheIoOperationKind::Read, 4);
619        storage.begin_read(read.clone()).unwrap();
620        storage
621            .finish_read(read.key(), "host-2".to_owned())
622            .unwrap();
623        assert_eq!(storage.phase(), CacheStoragePhase::HostBacked);
624        let promotion = storage.promote_host("device-2".to_owned()).unwrap();
625        assert_eq!(promotion.host, "host-2");
626        assert_eq!(storage.backing().map(String::as_str), Some("disk"));
627        storage.release_device_to_disk().unwrap();
628        assert_eq!(storage.phase(), CacheStoragePhase::DiskReady);
629    }
630
631    #[test]
632    fn failed_io_restores_the_stable_source_phase() {
633        let mut write = CacheBlockStorage::<String, _, String, HostOp, _>::host(block(), 3u8, None);
634        let write_op = io(CacheIoOperationKind::Write, 1);
635        write.begin_write(write_op.clone()).unwrap();
636        assert_eq!(write.fail_io(write_op.key()).unwrap(), write_op);
637        assert_eq!(write.phase(), CacheStoragePhase::HostUnbacked);
638
639        let mut read =
640            CacheBlockStorage::<String, u8, _, HostOp, _>::disk(block(), "disk".to_owned());
641        let read_op = io(CacheIoOperationKind::Read, 1);
642        read.begin_read(read_op.clone()).unwrap();
643        assert_eq!(read.fail_io(read_op.key()).unwrap(), read_op);
644        assert_eq!(read.phase(), CacheStoragePhase::DiskReady);
645    }
646
647    #[test]
648    fn operation_for_another_block_fails_without_changing_phase() {
649        let owned = block();
650        let mut foreign = block();
651        foreign.global_layer += 1;
652        let mut storage =
653            CacheBlockStorage::<String, u8, String, HostOp, _>::host(owned.clone(), 3, None);
654        let operation = IoOp(CacheIoOperationKey {
655            generation: 1,
656            id: foreign.clone(),
657            kind: CacheIoOperationKind::Write,
658        });
659
660        assert_eq!(
661            storage.begin_write(operation),
662            Err(CacheStorageError::BlockMismatch {
663                expected: Box::new(owned),
664                actual: Box::new(foreign),
665            })
666        );
667        assert_eq!(storage.phase(), CacheStoragePhase::HostUnbacked);
668    }
669
670    #[test]
671    fn phase_and_operation_identity_round_trip_portably() {
672        let key = io(CacheIoOperationKind::Read, 9).0;
673        let encoded = serde_json::to_string(&(CacheStoragePhase::DiskReading, &key)).unwrap();
674        let decoded: (CacheStoragePhase, CacheIoOperationKey) =
675            serde_json::from_str(&encoded).unwrap();
676        assert_eq!(decoded, (CacheStoragePhase::DiskReading, key));
677    }
678}
eredu_runtime/cache/storage.rs

eredu_runtime/cache/
storage.rs
