ozzy_replication/driver/
validation.rs

1use std::time::Duration;
2
3use ozzy_proto::NodeId;
4
5use super::{DriverError, ReplicaDriver};
6use crate::{
7    Admission, JournalGeneration, Prefix, PreparedOperation, ReplicationError, Scope, Status,
8};
9
10/// Image/authority fence captured before asynchronous application validation.
11///
12/// This is not proof of valid operation bodies, a write reservation, or a vote.
13/// The adapter must validate the exact submitted operations against the matching
14/// application image and retain the resulting plans until admission. Changed
15/// accepted/applied images or authority require fresh validation. A higher commit
16/// boundary alone changes neither image until application advances.
17#[derive(Debug, Clone, Copy, PartialEq, Eq)]
18pub struct ValidationTicket {
19    pub(crate) local: NodeId,
20    pub(crate) scope: Scope,
21    pub(crate) generation: JournalGeneration,
22    pub(crate) accepted: Prefix,
23    pub(crate) committed: Prefix,
24    pub(crate) applied: Prefix,
25}
26
27impl ValidationTicket {
28    /// Group, configuration, and view against which validation started.
29    pub const fn scope(self) -> Scope {
30        self.scope
31    }
32
33    /// Exact local writer incarnation. Reopen or installation invalidates it.
34    pub const fn generation(self) -> JournalGeneration {
35        self.generation
36    }
37
38    /// Speculative journal/application prefix used for validation.
39    pub const fn accepted(self) -> Prefix {
40        self.accepted
41    }
42
43    /// Established commit boundary when validation started.
44    pub const fn committed(self) -> Prefix {
45        self.committed
46    }
47
48    /// Completed application boundary when validation started.
49    pub const fn applied(self) -> Prefix {
50        self.applied
51    }
52
53    pub(crate) fn same_images_as(self, current: Self) -> bool {
54        self.local == current.local
55            && self.scope == current.scope
56            && self.generation == current.generation
57            && self.accepted == current.accepted
58            && self.applied == current.applied
59            // Both immutable tickets came from this exact accepted hash chain.
60            // Commit progress cannot alter speculative or applied state, but a
61            // regression or a conflicting digest at the same position is stale.
62            && (self.committed == current.committed || self.committed.op < current.committed.op)
63    }
64}
65
66impl ReplicaDriver {
67    /// Capture the exact live image and authority before worker-side validation.
68    ///
69    /// Validation itself may perform blocking identity-index reads on a worker.
70    /// Capturing this ticket neither reserves pipeline capacity nor delays timers.
71    /// Adapters bind speculative and applied images to their exact prefixes.
72    /// The captured commit boundary separately bounds read/application work.
73    pub fn begin_validation(&self) -> Result<ValidationTicket, DriverError> {
74        let normal = self.normal().ok_or(ReplicationError::NotNormal)?;
75        let snapshot = normal.snapshot();
76        if snapshot.status != Status::Normal {
77            return Err(ReplicationError::NotNormal.into());
78        }
79        if !snapshot.ready_for_appends {
80            return Err(ReplicationError::ActivationPending.into());
81        }
82        Ok(ValidationTicket {
83            local: self.configuration.voter(self.local),
84            scope: snapshot.scope,
85            generation: snapshot.journal.generation,
86            accepted: snapshot.accepted,
87            committed: snapshot.committed,
88            applied: snapshot.applied,
89        })
90    }
91
92    /// Capture a ticket for validation that the worker runs right after an
93    /// application through `applying.committed()`, where `applying` was
94    /// captured on the current image. The ticket names that prefix as applied,
95    /// the image the worker will hold. The core must reach it through
96    /// `apply_through` before admission, or admission reports stale validation.
97    pub fn begin_validation_after_apply(
98        &self,
99        applying: ValidationTicket,
100    ) -> Result<ValidationTicket, DriverError> {
101        let current = self.begin_validation()?;
102        let through = applying.committed;
103        if applying.local != current.local
104            || applying.scope != current.scope
105            || applying.generation != current.generation
106            || applying.accepted != current.accepted
107            || applying.applied != current.applied
108            || through.op < current.applied.op
109            || (through.op == current.applied.op && through != current.applied)
110            || through.op > current.committed.op
111            || (through.op == current.committed.op && through != current.committed)
112        {
113            return Err(DriverError::StaleValidation);
114        }
115        Ok(ValidationTicket {
116            applied: through,
117            ..current
118        })
119    }
120
121    /// Admit worker-validated operations only if their image and authority remain live.
122    ///
123    /// Call before installing speculative plans, writing, or sending PREPARE.
124    /// Stale validation is rejected without admitting operations or touching disk
125    /// evidence. The adapter discards its plans and retries against a fresh image.
126    /// A matching ticket still needs normal membership, lineage, and capacity checks.
127    /// Commit may advance on the unchanged accepted chain while validation runs;
128    /// this does not permit any change to the speculative or applied images.
129    pub fn prepare_validated(
130        &mut self,
131        from: NodeId,
132        ticket: ValidationTicket,
133        operations: &[PreparedOperation],
134        now: Duration,
135    ) -> Result<Admission, DriverError> {
136        self.observe_time(now)?;
137        if !self
138            .begin_validation()
139            .is_ok_and(|current| ticket.same_images_as(current))
140        {
141            return Err(DriverError::StaleValidation);
142        }
143        self.prepare(from, ticket.scope, operations, now)
144    }
145}
ozzy_replication/driver/validation.rs

ozzy_replication/driver/
validation.rs
