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

ozzy_replication/driver/
validation.rs
