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::{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    path: PathBuf,
128}
129
130impl FileSessionStoreWal {
131    pub fn new(store_root: impl AsRef<Path>) -> Self {
132        Self {
133            path: store_root
134                .as_ref()
135                .join("v1")
136                .join("wal")
137                .join("session-store.ndjson"),
138        }
139    }
140
141    pub fn path(&self) -> &Path {
142        &self.path
143    }
144
145    /// Load every validated entry and return the next one-based sequence.
146    pub async fn load_entries(&self) -> Result<(Vec<SessionStoreWalEntryV1>, u64)> {
147        if !self.path.exists() {
148            return Ok((Vec::new(), 1));
149        }
150        let file = File::open(&self.path).await.with_context(|| {
151            format!("Failed to open session store WAL: {}", self.path.display())
152        })?;
153        let mut lines = BufReader::new(file).lines();
154        let mut entries = Vec::new();
155        // Sequence may appear twice only as Intent → Committed for the *same*
156        // session. Cross-session reuse (concurrent writers) fails closed.
157        let mut seen_sequences: BTreeMap<u64, (SessionStoreWalPhaseV1, String)> = BTreeMap::new();
158        while let Some(line) = lines.next_line().await? {
159            if line.trim().is_empty() {
160                continue;
161            }
162            let entry: SessionStoreWalEntryV1 = serde_json::from_str(&line).with_context(|| {
163                format!(
164                    "Failed to decode session store WAL line in {}",
165                    self.path.display()
166                )
167            })?;
168            entry.validate()?;
169            if let Some((previous_phase, previous_session)) =
170                seen_sequences.insert(entry.sequence, (entry.phase, entry.session_id.clone()))
171            {
172                let allowed = matches!(previous_phase, SessionStoreWalPhaseV1::Intent)
173                    && matches!(entry.phase, SessionStoreWalPhaseV1::Committed)
174                    && previous_session == entry.session_id;
175                if !allowed {
176                    bail!(
177                        "session store WAL sequence {} conflicts with a retained entry",
178                        entry.sequence
179                    );
180                }
181            }
182            entries.push(entry);
183        }
184        let next = entries
185            .iter()
186            .map(|entry| entry.sequence)
187            .max()
188            .map(|max| max + 1)
189            .unwrap_or(1);
190        Ok((entries, next))
191    }
192
193    /// Whether an open/load error is the fail-closed duplicate-sequence case.
194    pub fn is_sequence_conflict(error: &anyhow::Error) -> bool {
195        error.chain().any(|cause| {
196            cause
197                .to_string()
198                .contains("conflicts with a retained entry")
199        })
200    }
201
202    /// Rename a corrupt WAL aside so a fresh log can be opened. Session
203    /// snapshots under `v1/sessions/` are left untouched.
204    pub async fn quarantine_corrupt(&self) -> Result<PathBuf> {
205        if !self.path.exists() {
206            bail!(
207                "session store WAL does not exist at {} so nothing to quarantine",
208                self.path.display()
209            );
210        }
211        let stamp = SystemTime::now()
212            .duration_since(UNIX_EPOCH)
213            .map(|duration| duration.as_secs())
214            .unwrap_or(0);
215        let dest = self
216            .path
217            .with_file_name(format!("session-store.ndjson.corrupt.{stamp}"));
218        fs::rename(&self.path, &dest).await.with_context(|| {
219            format!(
220                "Failed to quarantine session store WAL {} -> {}",
221                self.path.display(),
222                dest.display()
223            )
224        })?;
225        Ok(dest)
226    }
227
228    /// Append one validated entry with a data sync so reopen can observe it.
229    pub async fn append(&self, entry: &SessionStoreWalEntryV1) -> Result<()> {
230        entry.validate()?;
231        if let Some(parent) = self.path.parent() {
232            fs::create_dir_all(parent)
233                .await
234                .with_context(|| format!("Failed to create WAL directory: {}", parent.display()))?;
235        }
236        let mut encoded = serde_json::to_vec(entry).context("Failed to encode WAL entry")?;
237        encoded.push(b'\n');
238        let mut file = OpenOptions::new()
239            .create(true)
240            .append(true)
241            .open(&self.path)
242            .await
243            .with_context(|| format!("Failed to open WAL for append: {}", self.path.display()))?;
244        file.write_all(&encoded)
245            .await
246            .with_context(|| format!("Failed to append WAL entry to {}", self.path.display()))?;
247        file.sync_all()
248            .await
249            .with_context(|| format!("Failed to sync WAL entry to {}", self.path.display()))?;
250        Ok(())
251    }
252
253    /// Seal open intents whose durable snapshot already matches.
254    ///
255    /// `snapshot_digest_for` returns the digest of the currently durable
256    /// snapshot for a session, if any. Matching intents are acknowledged with
257    /// a Committed record so crash recovery does not leave a generation
258    /// half-published.
259    pub async fn recover<F, Fut>(&self, mut snapshot_digest_for: F) -> Result<u64>
260    where
261        F: FnMut(&str) -> Fut,
262        Fut: std::future::Future<Output = Result<Option<String>>>,
263    {
264        let (entries, mut next_sequence) = self.load_entries().await?;
265        let mut open_intents: BTreeMap<u64, SessionStoreWalEntryV1> = BTreeMap::new();
266        for entry in entries {
267            match entry.phase {
268                SessionStoreWalPhaseV1::Intent => {
269                    open_intents.insert(entry.sequence, entry);
270                }
271                SessionStoreWalPhaseV1::Committed => {
272                    open_intents.remove(&entry.sequence);
273                }
274            }
275        }
276        for (_, intent) in open_intents {
277            let Some(current) = snapshot_digest_for(&intent.session_id).await? else {
278                continue;
279            };
280            if current != intent.snapshot_digest {
281                continue;
282            }
283            let committed = SessionStoreWalEntryV1::new(
284                intent.sequence,
285                intent.session_id,
286                intent.snapshot_digest,
287                SessionStoreWalPhaseV1::Committed,
288                intent.recorded_at_ms.saturating_add(1),
289            )?;
290            self.append(&committed).await?;
291            next_sequence = next_sequence.max(committed.sequence + 1);
292        }
293        Ok(next_sequence)
294    }
295}
296
297#[cfg(test)]
298mod tests {
299    use super::*;
300
301    fn digest(ch: char) -> String {
302        format!("sha256:{}", ch.to_string().repeat(64))
303    }
304
305    #[test]
306    fn wal_entry_is_digest_bound_and_rejects_zero_sequence() {
307        let entry = SessionStoreWalEntryV1::new(
308            1,
309            "session-1",
310            digest('a'),
311            SessionStoreWalPhaseV1::Intent,
312            10,
313        )
314        .unwrap();
315        entry.validate().unwrap();
316        assert!(SessionStoreWalEntryV1::new(
317            0,
318            "session-1",
319            digest('a'),
320            SessionStoreWalPhaseV1::Intent,
321            10,
322        )
323        .is_err());
324    }
325}
a3s_code_core/store/wal.rs

a3s_code_core/store/
wal.rs
