a3s_code_core/store/
wal.rs

1//! Append-only session-store write-ahead log (KRN-6 / STORE-WAL1).
2//!
3//! The file adapter records an intent before replacing a session snapshot and
4//! a commit after the atomic replace succeeds. Reopen recovers by sealing any
5//! intent whose durable snapshot already matches, so a crash between rename
6//! and commit acknowledgement cannot lose the generation or invent a second
7//! one. The log never stores session plaintext payloads — only digests and
8//! identities.
9
10use crate::content_digest::{digest_json, validate_digest};
11use anyhow::{bail, Context, Result};
12use serde::{Deserialize, Serialize};
13use std::collections::BTreeMap;
14use std::path::{Component, Path, PathBuf};
15use std::time::{SystemTime, UNIX_EPOCH};
16use tokio::fs::{self, File, OpenOptions};
17use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
18
19pub const SESSION_STORE_WAL_ENTRY_SCHEMA_V1: &str = "a3s.code.session-store-wal-entry.v1";
20const SESSION_STORE_WAL_DIGEST_DOMAIN: &str = "a3s.code.session-store-wal-entry.identity.v1";
21const SESSION_STORE_SNAPSHOT_DIGEST_DOMAIN: &str = "a3s.code.session-store-snapshot.identity.v1";
22
23/// Lifecycle phase of one WAL record for a session snapshot commit.
24#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
25#[serde(rename_all = "camelCase")]
26pub enum SessionStoreWalPhaseV1 {
27    Intent,
28    Committed,
29}
30
31/// One append-only WAL record for a FileSessionStore snapshot commit.
32#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
33#[serde(rename_all = "camelCase", deny_unknown_fields)]
34pub struct SessionStoreWalEntryV1 {
35    pub schema: String,
36    pub sequence: u64,
37    pub session_id: String,
38    pub snapshot_digest: String,
39    pub phase: SessionStoreWalPhaseV1,
40    pub recorded_at_ms: u64,
41    pub entry_digest: String,
42}
43
44impl SessionStoreWalEntryV1 {
45    pub fn new(
46        sequence: u64,
47        session_id: impl Into<String>,
48        snapshot_digest: impl Into<String>,
49        phase: SessionStoreWalPhaseV1,
50        recorded_at_ms: u64,
51    ) -> Result<Self> {
52        let mut entry = Self {
53            schema: SESSION_STORE_WAL_ENTRY_SCHEMA_V1.to_owned(),
54            sequence,
55            session_id: session_id.into(),
56            snapshot_digest: snapshot_digest.into(),
57            phase,
58            recorded_at_ms,
59            entry_digest: String::new(),
60        };
61        entry.validate_without_digest()?;
62        entry.entry_digest = entry.expected_digest()?;
63        Ok(entry)
64    }
65
66    pub fn validate(&self) -> Result<()> {
67        self.validate_without_digest()?;
68        validate_digest(&self.entry_digest)
69            .map_err(|_| anyhow::anyhow!("session store WAL entryDigest is invalid"))?;
70        if self.entry_digest != self.expected_digest()? {
71            bail!("session store WAL entryDigest does not match contents");
72        }
73        Ok(())
74    }
75
76    fn validate_without_digest(&self) -> Result<()> {
77        if self.schema != SESSION_STORE_WAL_ENTRY_SCHEMA_V1 {
78            bail!("session store WAL schema is unsupported");
79        }
80        if self.sequence == 0 {
81            bail!("session store WAL sequence must be one-based");
82        }
83        if self.session_id.is_empty()
84            || self.session_id.contains('\0')
85            || self.session_id.contains(['\r', '\n'])
86        {
87            bail!("session store WAL sessionId is invalid");
88        }
89        validate_digest(&self.snapshot_digest)
90            .map_err(|_| anyhow::anyhow!("session store WAL snapshotDigest is invalid"))?;
91        Ok(())
92    }
93
94    fn expected_digest(&self) -> Result<String> {
95        #[derive(Serialize)]
96        struct Identity<'a> {
97            schema: &'a str,
98            sequence: u64,
99            session_id: &'a str,
100            snapshot_digest: &'a str,
101            phase: SessionStoreWalPhaseV1,
102            recorded_at_ms: u64,
103        }
104        digest_json(
105            SESSION_STORE_WAL_DIGEST_DOMAIN,
106            &Identity {
107                schema: &self.schema,
108                sequence: self.sequence,
109                session_id: &self.session_id,
110                snapshot_digest: &self.snapshot_digest,
111                phase: self.phase,
112                recorded_at_ms: self.recorded_at_ms,
113            },
114        )
115        .context("failed to digest session store WAL entry")
116    }
117}
118
119/// Digest a complete session snapshot for WAL identity fencing.
120pub fn snapshot_content_digest<T: Serialize>(snapshot: &T) -> Result<String> {
121    digest_json(SESSION_STORE_SNAPSHOT_DIGEST_DOMAIN, snapshot)
122        .context("failed to digest session snapshot for WAL")
123}
124
125/// Durable append-only JSONL WAL under a FileSessionStore root.
126pub struct FileSessionStoreWal {
127    root: PathBuf,
128    path: PathBuf,
129}
130
131impl FileSessionStoreWal {
132    pub fn new(store_root: impl AsRef<Path>) -> Self {
133        let root = store_root.as_ref().to_path_buf();
134        Self {
135            path: root.join("v1").join("wal").join("session-store.ndjson"),
136            root,
137        }
138    }
139
140    /// A symlink whose resolved path leaves this store is not the WAL.
141    fn refuse_leave(&self) -> Result<()> {
142        let root = std::fs::canonicalize(&self.root)
143            .with_context(|| format!("Failed to resolve session store: {}", self.root.display()))?;
144        let relative = self
145            .path
146            .strip_prefix(&self.root)
147            .unwrap_or(self.path.as_path());
148        let mut current = root.clone();
149        for component in relative.components() {
150            let Component::Normal(name) = component else {
151                bail!("session store WAL path must stay inside the store");
152            };
153            current.push(name);
154            match std::fs::symlink_metadata(&current) {
155                Ok(metadata) if metadata.file_type().is_symlink() => {
156                    let canonical = std::fs::canonicalize(&current).with_context(|| {
157                        format!(
158                            "refusing to follow a symbolic link in {}",
159                            current.display()
160                        )
161                    })?;
162                    if !canonical.starts_with(&root) {
163                        bail!(
164                            "session store WAL is a symbolic link outside the store: {}",
165                            current.display()
166                        );
167                    }
168                }
169                Ok(_) => {}
170                Err(error) if error.kind() == std::io::ErrorKind::NotFound => break,
171                Err(error) => bail!("failed to inspect session store WAL path: {error}"),
172            }
173        }
174        Ok(())
175    }
176
177    pub fn path(&self) -> &Path {
178        &self.path
179    }
180
181    /// Load every validated entry and return the next one-based sequence.
182    pub async fn load_entries(&self) -> Result<(Vec<SessionStoreWalEntryV1>, u64)> {
183        self.refuse_leave()?;
184        if !self.path.exists() {
185            return Ok((Vec::new(), 1));
186        }
187        let file = File::open(&self.path).await.with_context(|| {
188            format!("Failed to open session store WAL: {}", self.path.display())
189        })?;
190        let mut lines = BufReader::new(file).lines();
191        let mut entries = Vec::new();
192        // Sequence may appear twice only as Intent → Committed for the *same*
193        // session. Cross-session reuse (concurrent writers) fails closed.
194        let mut seen_sequences: BTreeMap<u64, (SessionStoreWalPhaseV1, String)> = BTreeMap::new();
195        while let Some(line) = lines.next_line().await? {
196            if line.trim().is_empty() {
197                continue;
198            }
199            let entry: SessionStoreWalEntryV1 = serde_json::from_str(&line).with_context(|| {
200                format!(
201                    "Failed to decode session store WAL line in {}",
202                    self.path.display()
203                )
204            })?;
205            entry.validate()?;
206            if let Some((previous_phase, previous_session)) =
207                seen_sequences.insert(entry.sequence, (entry.phase, entry.session_id.clone()))
208            {
209                let allowed = matches!(previous_phase, SessionStoreWalPhaseV1::Intent)
210                    && matches!(entry.phase, SessionStoreWalPhaseV1::Committed)
211                    && previous_session == entry.session_id;
212                if !allowed {
213                    bail!(
214                        "session store WAL sequence {} conflicts with a retained entry",
215                        entry.sequence
216                    );
217                }
218            }
219            entries.push(entry);
220        }
221        let next = entries
222            .iter()
223            .map(|entry| entry.sequence)
224            .max()
225            .map(|max| max + 1)
226            .unwrap_or(1);
227        Ok((entries, next))
228    }
229
230    /// Whether an open/load error is the fail-closed duplicate-sequence case.
231    pub fn is_sequence_conflict(error: &anyhow::Error) -> bool {
232        error.chain().any(|cause| {
233            cause
234                .to_string()
235                .contains("conflicts with a retained entry")
236        })
237    }
238
239    /// Rename a corrupt WAL aside so a fresh log can be opened. Session
240    /// snapshots under `v1/sessions/` are left untouched.
241    pub async fn quarantine_corrupt(&self) -> Result<PathBuf> {
242        self.refuse_leave()?;
243        if !self.path.exists() {
244            bail!(
245                "session store WAL does not exist at {} so nothing to quarantine",
246                self.path.display()
247            );
248        }
249        let stamp = SystemTime::now()
250            .duration_since(UNIX_EPOCH)
251            .map(|duration| duration.as_secs())
252            .unwrap_or(0);
253        let dest = self
254            .path
255            .with_file_name(format!("session-store.ndjson.corrupt.{stamp}"));
256        fs::rename(&self.path, &dest).await.with_context(|| {
257            format!(
258                "Failed to quarantine session store WAL {} -> {}",
259                self.path.display(),
260                dest.display()
261            )
262        })?;
263        Ok(dest)
264    }
265
266    /// Append one validated entry with a data sync so reopen can observe it.
267    pub async fn append(&self, entry: &SessionStoreWalEntryV1) -> Result<()> {
268        entry.validate()?;
269        self.refuse_leave()?;
270        if let Some(parent) = self.path.parent() {
271            fs::create_dir_all(parent)
272                .await
273                .with_context(|| format!("Failed to create WAL directory: {}", parent.display()))?;
274        }
275        let mut encoded = serde_json::to_vec(entry).context("Failed to encode WAL entry")?;
276        encoded.push(b'\n');
277        let mut file = OpenOptions::new()
278            .create(true)
279            .append(true)
280            .open(&self.path)
281            .await
282            .with_context(|| format!("Failed to open WAL for append: {}", self.path.display()))?;
283        file.write_all(&encoded)
284            .await
285            .with_context(|| format!("Failed to append WAL entry to {}", self.path.display()))?;
286        file.sync_all()
287            .await
288            .with_context(|| format!("Failed to sync WAL entry to {}", self.path.display()))?;
289        Ok(())
290    }
291
292    /// Seal open intents whose durable snapshot already matches.
293    ///
294    /// `snapshot_digest_for` returns the digest of the currently durable
295    /// snapshot for a session, if any. Matching intents are acknowledged with
296    /// a Committed record so crash recovery does not leave a generation
297    /// half-published.
298    pub async fn recover<F, Fut>(&self, mut snapshot_digest_for: F) -> Result<u64>
299    where
300        F: FnMut(&str) -> Fut,
301        Fut: std::future::Future<Output = Result<Option<String>>>,
302    {
303        let (entries, mut next_sequence) = self.load_entries().await?;
304        let mut open_intents: BTreeMap<u64, SessionStoreWalEntryV1> = BTreeMap::new();
305        for entry in entries {
306            match entry.phase {
307                SessionStoreWalPhaseV1::Intent => {
308                    open_intents.insert(entry.sequence, entry);
309                }
310                SessionStoreWalPhaseV1::Committed => {
311                    open_intents.remove(&entry.sequence);
312                }
313            }
314        }
315        for (_, intent) in open_intents {
316            let Some(current) = snapshot_digest_for(&intent.session_id).await? else {
317                continue;
318            };
319            if current != intent.snapshot_digest {
320                continue;
321            }
322            let committed = SessionStoreWalEntryV1::new(
323                intent.sequence,
324                intent.session_id,
325                intent.snapshot_digest,
326                SessionStoreWalPhaseV1::Committed,
327                intent.recorded_at_ms.saturating_add(1),
328            )?;
329            self.append(&committed).await?;
330            next_sequence = next_sequence.max(committed.sequence + 1);
331        }
332        Ok(next_sequence)
333    }
334}
335
336#[cfg(test)]
337mod tests {
338    use super::*;
339
340    fn digest(ch: char) -> String {
341        format!("sha256:{}", ch.to_string().repeat(64))
342    }
343
344    #[test]
345    fn wal_entry_is_digest_bound_and_rejects_zero_sequence() {
346        let entry = SessionStoreWalEntryV1::new(
347            1,
348            "session-1",
349            digest('a'),
350            SessionStoreWalPhaseV1::Intent,
351            10,
352        )
353        .unwrap();
354        entry.validate().unwrap();
355        assert!(SessionStoreWalEntryV1::new(
356            0,
357            "session-1",
358            digest('a'),
359            SessionStoreWalPhaseV1::Intent,
360            10,
361        )
362        .is_err());
363    }
364}
a3s_code_core/store/wal.rs

a3s_code_core/store/
wal.rs
