gatehouse_runtime_protocol/lifecycle/journal/storage/
reservation.rs

1//! Durable pre-create reservation. No implicit resume, adoption or engine call.
2use super::*;
3#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
4#[serde(rename_all = "kebab-case")]
5enum Phase {
6    Pending,
7    Resolved,
8    Retained,
9    Abandoned,
10}
11#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
12#[serde(deny_unknown_fields)]
13struct ReservationRecord {
14    schema: String,
15    request: Request,
16    plan: CreatePlan,
17    phase: Phase,
18    response: Option<Response>,
19    abandonment_reason: Option<String>,
20}
21impl ReservationRecord {
22    fn validate(&self) -> Result<()> {
23        if self.schema != "gatehouse-lifecycle-reservation/v1" {
24            return Err(Invalid("invalid reservation schema"));
25        }
26        self.plan.validate(&self.request)?;
27        if let Some(response) = &self.response {
28            response.validate(&self.request)?;
29        }
30        let image = matches!(
31            self.response.as_ref().map(|r| &r.outcome),
32            Some(Outcome::Image { .. })
33        );
34        let unknown = matches!(
35            self.response.as_ref().map(|r| &r.outcome),
36            None | Some(Outcome::Unknown { .. } | Outcome::RejectedBeforeEffect { .. })
37        );
38        let valid = match self.phase {
39            Phase::Pending => self.response.is_none(),
40            Phase::Resolved => image,
41            Phase::Retained => unknown,
42            Phase::Abandoned => image || unknown,
43        };
44        if !valid
45            || (self.phase == Phase::Abandoned)
46                != (self.abandonment_reason.as_deref() == Some("operator-abandoned-before-create"))
47            || (self.phase != Phase::Abandoned && self.abandonment_reason.is_some())
48        {
49            return Err(Invalid("reservation evidence/phase mismatch"));
50        }
51        Ok(())
52    }
53    fn bytes(&self) -> Result<Vec<u8>> {
54        self.validate()?;
55        canonical::encode(
56            &serde_json::to_value(self).map_err(|_| Invalid("reservation encoding failed"))?,
57        )
58    }
59}
60pub struct Reservation {
61    root: PathBuf,
62    directory: LockedDirectory,
63    record: ReservationRecord,
64    bytes: Vec<u8>,
65    ready_resolve: bool,
66    ready_create: bool,
67    poisoned: bool,
68}
69/// Non-Clone, borrowed, one-use authorization for only the reserved ResolveImage.
70#[must_use]
71pub struct ResolutionDispatch<'a> {
72    reservation: &'a mut Reservation,
73}
74impl ResolutionDispatch<'_> {
75    pub fn request(&self) -> &Request {
76        &self.reservation.record.request
77    }
78    /// Call only after process/installation/cancellation checks, like Dispatch.
79    pub fn observe(self, bytes: Option<&[u8]>) -> Result<()> {
80        let mut next = self.reservation.record.clone();
81        next.response = match bytes {
82            Some(bytes) => match wire::parse_response(bytes, &next.request) {
83                Ok(response) => Some(response),
84                Err(error) => {
85                    self.reservation.poisoned = true;
86                    return Err(error);
87                }
88            },
89            None => None,
90        };
91        next.phase = if matches!(
92            next.response.as_ref().map(|r| &r.outcome),
93            Some(Outcome::Image { .. })
94        ) {
95            Phase::Resolved
96        } else {
97            Phase::Retained
98        };
99        self.reservation.commit(next)?;
100        self.reservation.ready_create = self.reservation.record.phase == Phase::Resolved;
101        Ok(())
102    }
103}
104impl Reservation {
105    pub fn create(root: &Path, request: &Request, plan: CreatePlan) -> Result<Self> {
106        let directory = Journal::directory(root)?;
107        let record = ReservationRecord {
108            schema: "gatehouse-lifecycle-reservation/v1".into(),
109            request: request.clone(),
110            plan,
111            phase: Phase::Pending,
112            response: None,
113            abandonment_reason: None,
114        };
115        let bytes = record.bytes()?;
116        for name in [
117            "active",
118            "reservation.json",
119            "reservation-tombstone.json",
120            "lease.json",
121            "tombstone.json",
122        ] {
123            if Journal::read(&directory, name)?.is_some() {
124                return Err(Invalid("resource already reserved/recorded"));
125            }
126        }
127        let mut marker = Journal::at(
128            &directory,
129            "active",
130            libc::O_WRONLY | libc::O_CREAT | libc::O_EXCL,
131        )?
132        .ok_or(Invalid("reservation marker missing"))?;
133        io(marker.write_all(request.binding.operation_key.as_bytes()))?;
134        io(marker.sync_all())?;
135        io(directory.sync_all())?;
136        Journal::atomic(&directory, "reservation.json", &bytes)?;
137        Ok(Self {
138            root: root.into(),
139            directory,
140            record,
141            bytes,
142            ready_resolve: true,
143            ready_create: false,
144            poisoned: false,
145        })
146    }
147    pub fn open(root: &Path, request: &Request, plan: &CreatePlan) -> Result<Self> {
148        request.validate()?;
149        plan.validate(request)?;
150        let directory = Journal::directory(root)?;
151        let bytes = Journal::read(&directory, "reservation.json")?
152            .ok_or(Invalid("reservation missing; no adoption"))?;
153        let record: ReservationRecord = serde_json::from_value(canonical::parse_canonical(&bytes)?)
154            .map_err(|_| Invalid("reservation schema rejected"))?;
155        if record.bytes()? != bytes || record.request != *request || record.plan != *plan {
156            return Err(Invalid("reservation binding changed"));
157        }
158        let mut value = Self {
159            root: root.into(),
160            directory,
161            record,
162            bytes,
163            ready_resolve: false,
164            ready_create: false,
165            poisoned: false,
166        };
167        value.check()?;
168        if value.record.phase == Phase::Pending {
169            let mut next = value.record.clone();
170            next.phase = Phase::Retained;
171            value.commit(next)?;
172        }
173        Ok(value)
174    }
175    fn check(&self) -> Result<()> {
176        if self.poisoned {
177            return Err(Invalid("reservation poisoned; reopen required"));
178        }
179        let now = io(fs::symlink_metadata(&self.root))?;
180        let held = io(self.directory.metadata())?;
181        private(&now, true)?;
182        if now.dev() != held.dev() || now.ino() != held.ino() {
183            return Err(Invalid("reservation directory replaced"));
184        }
185        if Journal::read(&self.directory, "reservation.json")?.as_deref()
186            != Some(self.bytes.as_slice())
187            || Journal::read(&self.directory, "lease.json")?.is_some()
188            || Journal::read(&self.directory, "tombstone.json")?.is_some()
189        {
190            return Err(Invalid(
191                "reservation changed/transferred; use journal recovery",
192            ));
193        }
194        let marker = Journal::read(&self.directory, "active")?;
195        let tombstone = Journal::read(&self.directory, "reservation-tombstone.json")?;
196        if let Some(t) = &tombstone {
197            if self.record.phase != Phase::Abandoned || t != &self.bytes {
198                return Err(Invalid("reservation tombstone mismatch"));
199            }
200        }
201        if let Some(marker) = marker {
202            if marker != self.record.request.binding.operation_key.as_bytes() {
203                return Err(Invalid("reservation marker changed"));
204            }
205        } else if self.record.phase != Phase::Abandoned || tombstone.as_ref() != Some(&self.bytes) {
206            return Err(Invalid("reservation marker missing"));
207        }
208        Ok(())
209    }
210    fn commit(&mut self, next: ReservationRecord) -> Result<()> {
211        self.check()?;
212        self.poisoned = true;
213        self.ready_resolve = false;
214        self.ready_create = false;
215        let bytes = next.bytes()?;
216        Journal::atomic(&self.directory, "reservation.json", &bytes)?;
217        self.bytes = bytes;
218        self.record = next;
219        self.poisoned = false;
220        Ok(())
221    }
222    pub fn dispatch(&mut self) -> Result<ResolutionDispatch<'_>> {
223        self.check()?;
224        if !self.ready_resolve || self.record.phase != Phase::Pending {
225            return Err(Invalid("no fresh resolution authority"));
226        }
227        self.ready_resolve = false;
228        Ok(ResolutionDispatch { reservation: self })
229    }
230    pub fn retain_unknown(&mut self) -> Result<()> {
231        self.check()?;
232        if self.record.phase != Phase::Pending {
233            return Err(Invalid("no pending resolution"));
234        }
235        let mut next = self.record.clone();
236        next.phase = Phase::Retained;
237        self.commit(next)
238    }
239    /// Explicit trusted operator policy, not human authentication. Normal recovery
240    /// must not call this. No lease file may exist and image evidence is retained.
241    pub fn authorize_operator_resume_create(&mut self) -> Result<()> {
242        self.check()?;
243        if self.record.phase != Phase::Resolved {
244            return Err(Invalid("no resolved reservation"));
245        }
246        self.ready_create = true;
247        Ok(())
248    }
249    pub fn into_journal(self, invocation: &str) -> Result<Journal> {
250        self.transfer_with(invocation, |_| Ok(()))
251    }
252    fn transfer_with(
253        self,
254        invocation: &str,
255        stage: impl FnMut(u8) -> Result<()>,
256    ) -> Result<Journal> {
257        self.check()?;
258        if !self.ready_create || self.record.phase != Phase::Resolved {
259            return Err(Invalid("create transfer not authorized"));
260        }
261        let proof = ResolutionProof {
262            request: self.record.request.clone(),
263            plan: self.record.plan.clone(),
264            response: self
265                .record
266                .response
267                .clone()
268                .ok_or(Invalid("resolution proof missing"))?,
269        };
270        let create = proof.create(invocation)?;
271        let record = Record::new_resolved(&create, Some(proof))?;
272        let bytes = Journal::encode(&record)?;
273        // Keep the original reservation marker. A durable lease is the transfer
274        // commit: any reopen then uses journal recovery and cannot resolve again.
275        Journal::atomic_with(&self.directory, "lease.json", &bytes, stage)?;
276        Ok(Journal {
277            root: self.root,
278            directory: self.directory,
279            record,
280            bytes,
281            poisoned: false,
282            ready: true,
283            fresh_status: false,
284        })
285    }
286    /// Explicit abandonment is allowed only before any lease/creation intent.
287    /// Tombstone is persisted before releasing the active marker; records remain.
288    pub fn abandon(&mut self) -> Result<()> {
289        self.check()?;
290        if self.record.phase == Phase::Pending {
291            return Err(Invalid("retain incomplete resolution before abandonment"));
292        }
293        if self.record.phase != Phase::Abandoned {
294            let mut next = self.record.clone();
295            next.phase = Phase::Abandoned;
296            next.abandonment_reason = Some("operator-abandoned-before-create".into());
297            self.commit(next)?;
298        }
299        Journal::atomic(&self.directory, "reservation-tombstone.json", &self.bytes)?;
300        let name = CString::new("active").unwrap();
301        if unsafe { libc::unlinkat(self.directory.as_raw_fd(), name.as_ptr(), 0) } != 0
302            && std::io::Error::last_os_error().kind() != std::io::ErrorKind::NotFound
303        {
304            return Err(Invalid("reservation marker release failed"));
305        }
306        io(self.directory.sync_all())?;
307        self.ready_create = false;
308        self.ready_resolve = false;
309        Ok(())
310    }
311}
312impl Journal {
313    /// Reopen a transferred lease only with the exact trusted reservation inputs.
314    /// Even pending Create is consumed as unknown; no automatic second dispatch.
315    pub fn open_reserved(root: &Path, request: &Request, plan: &CreatePlan) -> Result<Self> {
316        plan.validate(request)?;
317        let directory = Self::directory(root)?;
318        let bytes =
319            Self::read(&directory, "lease.json")?.ok_or(Invalid("transferred journal missing"))?;
320        let record: Record = serde_json::from_value(canonical::parse_canonical(&bytes)?)
321            .map_err(|_| Invalid("journal schema rejected"))?;
322        if Self::encode(&record)? != bytes {
323            return Err(Invalid("journal encoding mismatch"));
324        }
325        let proof = record
326            .resolution
327            .as_ref()
328            .ok_or(Invalid("journal has no resolution proof"))?;
329        if proof.request != *request || proof.plan != *plan {
330            return Err(Invalid("reservation/journal identity mismatch"));
331        }
332        Self::open_held(root, &record.create, directory)
333    }
334}
335
336#[cfg(test)]
337mod tests;
gatehouse_runtime_protocol/lifecycle/journal/storage/reservation.rs

gatehouse_runtime_protocol/lifecycle/journal/storage/
reservation.rs
