cookie_agent_engine/
grant_journal.rs

1//! Durable exact tree-grant invalidations committed before mutations execute.
2
3use std::{
4    collections::HashSet,
5    path::PathBuf,
6    sync::{Arc, Mutex},
7};
8#[cfg(test)]
9use std::{fs::OpenOptions, io::Write};
10
11use cookie_agent_protocol::{SessionId, Sha256Digest, TreeApprovalGrantId};
12use jiff::Timestamp;
13use serde::{Deserialize, Serialize};
14use thiserror::Error;
15
16use crate::events::{EventLogError, append_jsonl, load_jsonl_shared};
17
18#[derive(Clone, Debug, Deserialize, Serialize)]
19#[serde(deny_unknown_fields)]
20struct GrantInvalidationRecord {
21    seq: u64,
22    timestamp: Timestamp,
23    root_session_id: SessionId,
24    grant_ids: Vec<TreeApprovalGrantId>,
25    resource_digests: Vec<Sha256Digest>,
26}
27
28#[derive(Debug, Error)]
29pub enum GrantJournalError {
30    #[error(transparent)]
31    Event(#[from] EventLogError),
32    #[error("invalid grant invalidation journal at {path}: {message}")]
33    Corrupt { path: PathBuf, message: String },
34    #[error("grant invalidation journal at {path} is poisoned and must be reopened")]
35    Poisoned { path: PathBuf },
36}
37
38#[derive(Debug)]
39pub struct GrantInvalidationJournal {
40    path: PathBuf,
41    state: Mutex<GrantJournalState>,
42    #[cfg(test)]
43    failure: Mutex<Option<TestFailure>>,
44}
45
46#[derive(Debug)]
47struct GrantJournalState {
48    records: Vec<GrantInvalidationRecord>,
49    poisoned: bool,
50}
51
52#[cfg(test)]
53#[derive(Clone, Copy, Debug)]
54pub(crate) enum TestFailure {
55    PartialWrite,
56    DurableButError,
57}
58
59impl GrantInvalidationJournal {
60    pub fn open(path: PathBuf) -> Result<Arc<Self>, GrantJournalError> {
61        let records = load_jsonl_shared::<GrantInvalidationRecord>(&path)?;
62        for (index, record) in records.iter().enumerate() {
63            let expected = index as u64 + 1;
64            if record.seq != expected {
65                return Err(GrantJournalError::Corrupt {
66                    path,
67                    message: format!(
68                        "grant invalidation sequence {} is not contiguous; expected {expected}",
69                        record.seq
70                    ),
71                });
72            }
73            if record.grant_ids.is_empty() || record.resource_digests.is_empty() {
74                return Err(GrantJournalError::Corrupt {
75                    path,
76                    message: "grant invalidation must contain grants and normalized resources"
77                        .into(),
78                });
79            }
80            let unique = record.grant_ids.iter().copied().collect::<HashSet<_>>();
81            if unique.len() != record.grant_ids.len() {
82                return Err(GrantJournalError::Corrupt {
83                    path,
84                    message: "grant invalidation contains duplicate grant IDs".into(),
85                });
86            }
87        }
88        Ok(Arc::new(Self {
89            path,
90            state: Mutex::new(GrantJournalState {
91                records,
92                poisoned: false,
93            }),
94            #[cfg(test)]
95            failure: Mutex::new(None),
96        }))
97    }
98
99    pub fn invalidate(
100        &self,
101        root_session_id: SessionId,
102        mut grant_ids: Vec<TreeApprovalGrantId>,
103        mut resource_digests: Vec<Sha256Digest>,
104    ) -> Result<(), GrantJournalError> {
105        grant_ids.sort_by_key(ToString::to_string);
106        grant_ids.dedup();
107        resource_digests.sort_by(|left, right| left.as_str().cmp(right.as_str()));
108        resource_digests.dedup();
109        let mut state = self
110            .state
111            .lock()
112            .unwrap_or_else(|poisoned| poisoned.into_inner());
113        if state.poisoned {
114            return Err(GrantJournalError::Poisoned {
115                path: self.path.clone(),
116            });
117        }
118        if grant_ids.is_empty() {
119            return Ok(());
120        }
121        let seq = state
122            .records
123            .last()
124            .map_or(Some(1), |record| record.seq.checked_add(1))
125            .ok_or_else(|| {
126                state.poisoned = true;
127                GrantJournalError::Poisoned {
128                    path: self.path.clone(),
129                }
130            })?;
131        let record = GrantInvalidationRecord {
132            seq,
133            timestamp: Timestamp::now(),
134            root_session_id,
135            grant_ids,
136            resource_digests,
137        };
138        if let Err(error) = self.append_record(&record) {
139            state.poisoned = true;
140            return Err(error);
141        }
142        state.records.push(record);
143        Ok(())
144    }
145
146    fn append_record(&self, record: &GrantInvalidationRecord) -> Result<(), GrantJournalError> {
147        #[cfg(test)]
148        if let Some(failure) = self
149            .failure
150            .lock()
151            .unwrap_or_else(|poisoned| poisoned.into_inner())
152            .take()
153        {
154            match failure {
155                TestFailure::PartialWrite => {
156                    let bytes = serde_json::to_vec(record).expect("record serializes");
157                    let mut file = OpenOptions::new()
158                        .create(true)
159                        .append(true)
160                        .open(&self.path)
161                        .map_err(|source| EventLogError::Io {
162                            path: self.path.clone(),
163                            source,
164                        })?;
165                    file.write_all(&bytes[..bytes.len() / 2])
166                        .and_then(|()| file.sync_data())
167                        .map_err(|source| EventLogError::Io {
168                            path: self.path.clone(),
169                            source,
170                        })?;
171                    return Err(GrantJournalError::Event(EventLogError::Io {
172                        path: self.path.clone(),
173                        source: std::io::Error::other("injected partial write failure"),
174                    }));
175                }
176                TestFailure::DurableButError => {
177                    append_jsonl(&self.path, record)?;
178                    return Err(GrantJournalError::Event(EventLogError::Io {
179                        path: self.path.clone(),
180                        source: std::io::Error::other("injected post-fsync failure"),
181                    }));
182                }
183            }
184        }
185        append_jsonl(&self.path, record).map_err(Into::into)
186    }
187
188    #[must_use]
189    pub fn invalidated_ids(&self) -> HashSet<TreeApprovalGrantId> {
190        self.state
191            .lock()
192            .unwrap_or_else(|poisoned| poisoned.into_inner())
193            .records
194            .iter()
195            .flat_map(|record| record.grant_ids.iter().copied())
196            .collect()
197    }
198
199    #[cfg(test)]
200    pub(crate) fn inject_failure(&self, failure: TestFailure) {
201        *self
202            .failure
203            .lock()
204            .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(failure);
205    }
206}
207
208#[cfg(test)]
209mod tests {
210    use std::fs;
211
212    use cookie_agent_protocol::{SessionId, Sha256Digest, TreeApprovalGrantId};
213    use uuid::Uuid;
214
215    use super::{GrantInvalidationJournal, TestFailure};
216
217    fn invalidate(journal: &GrantInvalidationJournal, grant: TreeApprovalGrantId) {
218        journal
219            .invalidate(
220                SessionId(Uuid::from_u128(2)),
221                vec![grant],
222                vec![Sha256Digest::new("33".repeat(32)).expect("digest")],
223            )
224            .expect("commit invalidation");
225    }
226
227    #[test]
228    fn invalidation_is_absent_before_commit_and_durable_after_reopen() {
229        let directory = tempfile::tempdir().expect("tempdir");
230        let path = directory.path().join("grant-invalidations.jsonl");
231        let journal = GrantInvalidationJournal::open(path.clone()).expect("open journal");
232        let grant = TreeApprovalGrantId(Uuid::from_u128(1));
233        assert!(!journal.invalidated_ids().contains(&grant));
234
235        journal
236            .invalidate(
237                SessionId(Uuid::from_u128(2)),
238                vec![grant],
239                vec![Sha256Digest::new("11".repeat(32)).expect("digest")],
240            )
241            .expect("commit invalidation");
242        drop(journal);
243
244        let reopened = GrantInvalidationJournal::open(path).expect("reopen journal");
245        assert!(reopened.invalidated_ids().contains(&grant));
246    }
247
248    #[test]
249    fn journal_rejects_noncontiguous_records() {
250        let directory = tempfile::tempdir().expect("tempdir");
251        let path = directory.path().join("grant-invalidations.jsonl");
252        let grant = TreeApprovalGrantId(Uuid::from_u128(1));
253        let journal = GrantInvalidationJournal::open(path.clone()).expect("open journal");
254        journal
255            .invalidate(
256                SessionId(Uuid::from_u128(2)),
257                vec![grant],
258                vec![Sha256Digest::new("22".repeat(32)).expect("digest")],
259            )
260            .expect("commit invalidation");
261        drop(journal);
262
263        let source =
264            fs::read_to_string(&path)
265                .expect("read journal")
266                .replacen("\"seq\":1", "\"seq\":2", 1);
267        fs::write(&path, source).expect("tamper journal");
268        assert!(GrantInvalidationJournal::open(path).is_err());
269    }
270
271    #[test]
272    fn ambiguous_partial_write_poisons_until_reopen() {
273        let directory = tempfile::tempdir().expect("tempdir");
274        let path = directory.path().join("grant-invalidations.jsonl");
275        let journal = GrantInvalidationJournal::open(path.clone()).expect("open journal");
276        journal.inject_failure(TestFailure::PartialWrite);
277        let first = TreeApprovalGrantId(Uuid::from_u128(10));
278        assert!(
279            journal
280                .invalidate(
281                    SessionId(Uuid::from_u128(2)),
282                    vec![first],
283                    vec![Sha256Digest::new("44".repeat(32)).expect("digest")],
284                )
285                .is_err()
286        );
287        let second = TreeApprovalGrantId(Uuid::from_u128(11));
288        assert!(
289            journal
290                .invalidate(
291                    SessionId(Uuid::from_u128(2)),
292                    vec![second],
293                    vec![Sha256Digest::new("55".repeat(32)).expect("digest")],
294                )
295                .is_err()
296        );
297        drop(journal);
298
299        let before = fs::read(&path).expect("read torn journal");
300        let reopened = GrantInvalidationJournal::open(path.clone()).expect("read torn tail");
301        assert!(!reopened.invalidated_ids().contains(&first));
302        assert!(!reopened.invalidated_ids().contains(&second));
303        assert_eq!(fs::read(path).expect("reread torn journal"), before);
304    }
305
306    #[test]
307    fn durable_but_error_poisons_and_reopen_recovers_committed_record() {
308        let directory = tempfile::tempdir().expect("tempdir");
309        let path = directory.path().join("grant-invalidations.jsonl");
310        let journal = GrantInvalidationJournal::open(path.clone()).expect("open journal");
311        journal.inject_failure(TestFailure::DurableButError);
312        let first = TreeApprovalGrantId(Uuid::from_u128(20));
313        assert!(
314            journal
315                .invalidate(
316                    SessionId(Uuid::from_u128(2)),
317                    vec![first],
318                    vec![Sha256Digest::new("66".repeat(32)).expect("digest")],
319                )
320                .is_err()
321        );
322        let second = TreeApprovalGrantId(Uuid::from_u128(21));
323        assert!(
324            journal
325                .invalidate(
326                    SessionId(Uuid::from_u128(2)),
327                    vec![second],
328                    vec![Sha256Digest::new("77".repeat(32)).expect("digest")],
329                )
330                .is_err()
331        );
332        drop(journal);
333
334        let reopened = GrantInvalidationJournal::open(path).expect("reopen durable record");
335        assert!(reopened.invalidated_ids().contains(&first));
336        assert!(!reopened.invalidated_ids().contains(&second));
337        invalidate(&reopened, second);
338    }
339}
cookie_agent_engine/grant_journal.rs

cookie_agent_engine/
grant_journal.rs
