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::evaluation::{digest_json, validate_digest};
11use anyhow::{bail, Context, Result};
12use serde::{Deserialize, Serialize};
13use std::collections::BTreeMap;
14use std::path::{Path, PathBuf};
15use tokio::fs::{self, File, OpenOptions};
16use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
17
18pub const SESSION_STORE_WAL_ENTRY_SCHEMA_V1: &str = "a3s.code.session-store-wal-entry.v1";
19const SESSION_STORE_WAL_DIGEST_DOMAIN: &str = "a3s.code.session-store-wal-entry.identity.v1";
20const SESSION_STORE_SNAPSHOT_DIGEST_DOMAIN: &str = "a3s.code.session-store-snapshot.identity.v1";
21
22/// Lifecycle phase of one WAL record for a session snapshot commit.
23#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
24#[serde(rename_all = "camelCase")]
25pub enum SessionStoreWalPhaseV1 {
26    Intent,
27    Committed,
28}
29
30/// One append-only WAL record for a FileSessionStore snapshot commit.
31#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
32#[serde(rename_all = "camelCase", deny_unknown_fields)]
33pub struct SessionStoreWalEntryV1 {
34    pub schema: String,
35    pub sequence: u64,
36    pub session_id: String,
37    pub snapshot_digest: String,
38    pub phase: SessionStoreWalPhaseV1,
39    pub recorded_at_ms: u64,
40    pub entry_digest: String,
41}
42
43impl SessionStoreWalEntryV1 {
44    pub fn new(
45        sequence: u64,
46        session_id: impl Into<String>,
47        snapshot_digest: impl Into<String>,
48        phase: SessionStoreWalPhaseV1,
49        recorded_at_ms: u64,
50    ) -> Result<Self> {
51        let mut entry = Self {
52            schema: SESSION_STORE_WAL_ENTRY_SCHEMA_V1.to_owned(),
53            sequence,
54            session_id: session_id.into(),
55            snapshot_digest: snapshot_digest.into(),
56            phase,
57            recorded_at_ms,
58            entry_digest: String::new(),
59        };
60        entry.validate_without_digest()?;
61        entry.entry_digest = entry.expected_digest()?;
62        Ok(entry)
63    }
64
65    pub fn validate(&self) -> Result<()> {
66        self.validate_without_digest()?;
67        validate_digest(&self.entry_digest)
68            .map_err(|_| anyhow::anyhow!("session store WAL entryDigest is invalid"))?;
69        if self.entry_digest != self.expected_digest()? {
70            bail!("session store WAL entryDigest does not match contents");
71        }
72        Ok(())
73    }
74
75    fn validate_without_digest(&self) -> Result<()> {
76        if self.schema != SESSION_STORE_WAL_ENTRY_SCHEMA_V1 {
77            bail!("session store WAL schema is unsupported");
78        }
79        if self.sequence == 0 {
80            bail!("session store WAL sequence must be one-based");
81        }
82        if self.session_id.is_empty()
83            || self.session_id.contains('\0')
84            || self.session_id.contains(['\r', '\n'])
85        {
86            bail!("session store WAL sessionId is invalid");
87        }
88        validate_digest(&self.snapshot_digest)
89            .map_err(|_| anyhow::anyhow!("session store WAL snapshotDigest is invalid"))?;
90        Ok(())
91    }
92
93    fn expected_digest(&self) -> Result<String> {
94        #[derive(Serialize)]
95        struct Identity<'a> {
96            schema: &'a str,
97            sequence: u64,
98            session_id: &'a str,
99            snapshot_digest: &'a str,
100            phase: SessionStoreWalPhaseV1,
101            recorded_at_ms: u64,
102        }
103        digest_json(
104            SESSION_STORE_WAL_DIGEST_DOMAIN,
105            &Identity {
106                schema: &self.schema,
107                sequence: self.sequence,
108                session_id: &self.session_id,
109                snapshot_digest: &self.snapshot_digest,
110                phase: self.phase,
111                recorded_at_ms: self.recorded_at_ms,
112            },
113        )
114        .context("failed to digest session store WAL entry")
115    }
116}
117
118/// Digest a complete session snapshot for WAL identity fencing.
119pub fn snapshot_content_digest<T: Serialize>(snapshot: &T) -> Result<String> {
120    digest_json(SESSION_STORE_SNAPSHOT_DIGEST_DOMAIN, snapshot)
121        .context("failed to digest session snapshot for WAL")
122}
123
124/// Durable append-only JSONL WAL under a FileSessionStore root.
125pub struct FileSessionStoreWal {
126    path: PathBuf,
127}
128
129impl FileSessionStoreWal {
130    pub fn new(store_root: impl AsRef<Path>) -> Self {
131        Self {
132            path: store_root
133                .as_ref()
134                .join("v1")
135                .join("wal")
136                .join("session-store.ndjson"),
137        }
138    }
139
140    pub fn path(&self) -> &Path {
141        &self.path
142    }
143
144    /// Load every validated entry and return the next one-based sequence.
145    pub async fn load_entries(&self) -> Result<(Vec<SessionStoreWalEntryV1>, u64)> {
146        if !self.path.exists() {
147            return Ok((Vec::new(), 1));
148        }
149        let file = File::open(&self.path).await.with_context(|| {
150            format!("Failed to open session store WAL: {}", self.path.display())
151        })?;
152        let mut lines = BufReader::new(file).lines();
153        let mut entries = Vec::new();
154        let mut seen_sequences = BTreeMap::new();
155        while let Some(line) = lines.next_line().await? {
156            if line.trim().is_empty() {
157                continue;
158            }
159            let entry: SessionStoreWalEntryV1 = serde_json::from_str(&line).with_context(|| {
160                format!(
161                    "Failed to decode session store WAL line in {}",
162                    self.path.display()
163                )
164            })?;
165            entry.validate()?;
166            if let Some(previous) = seen_sequences.insert(entry.sequence, entry.phase) {
167                // Intent then Committed for the same sequence is the only
168                // allowed reuse; any other collision fails closed.
169                let allowed = matches!(previous, SessionStoreWalPhaseV1::Intent)
170                    && matches!(entry.phase, SessionStoreWalPhaseV1::Committed);
171                if !allowed {
172                    bail!(
173                        "session store WAL sequence {} conflicts with a retained entry",
174                        entry.sequence
175                    );
176                }
177            }
178            entries.push(entry);
179        }
180        let next = entries
181            .iter()
182            .map(|entry| entry.sequence)
183            .max()
184            .map(|max| max + 1)
185            .unwrap_or(1);
186        Ok((entries, next))
187    }
188
189    /// Append one validated entry with a data sync so reopen can observe it.
190    pub async fn append(&self, entry: &SessionStoreWalEntryV1) -> Result<()> {
191        entry.validate()?;
192        if let Some(parent) = self.path.parent() {
193            fs::create_dir_all(parent)
194                .await
195                .with_context(|| format!("Failed to create WAL directory: {}", parent.display()))?;
196        }
197        let mut encoded = serde_json::to_vec(entry).context("Failed to encode WAL entry")?;
198        encoded.push(b'\n');
199        let mut file = OpenOptions::new()
200            .create(true)
201            .append(true)
202            .open(&self.path)
203            .await
204            .with_context(|| format!("Failed to open WAL for append: {}", self.path.display()))?;
205        file.write_all(&encoded)
206            .await
207            .with_context(|| format!("Failed to append WAL entry to {}", self.path.display()))?;
208        file.sync_all()
209            .await
210            .with_context(|| format!("Failed to sync WAL entry to {}", self.path.display()))?;
211        Ok(())
212    }
213
214    /// Seal open intents whose durable snapshot already matches.
215    ///
216    /// `snapshot_digest_for` returns the digest of the currently durable
217    /// snapshot for a session, if any. Matching intents are acknowledged with
218    /// a Committed record so crash recovery does not leave a generation
219    /// half-published.
220    pub async fn recover<F, Fut>(&self, mut snapshot_digest_for: F) -> Result<u64>
221    where
222        F: FnMut(&str) -> Fut,
223        Fut: std::future::Future<Output = Result<Option<String>>>,
224    {
225        let (entries, mut next_sequence) = self.load_entries().await?;
226        let mut open_intents: BTreeMap<u64, SessionStoreWalEntryV1> = BTreeMap::new();
227        for entry in entries {
228            match entry.phase {
229                SessionStoreWalPhaseV1::Intent => {
230                    open_intents.insert(entry.sequence, entry);
231                }
232                SessionStoreWalPhaseV1::Committed => {
233                    open_intents.remove(&entry.sequence);
234                }
235            }
236        }
237        for (_, intent) in open_intents {
238            let Some(current) = snapshot_digest_for(&intent.session_id).await? else {
239                continue;
240            };
241            if current != intent.snapshot_digest {
242                continue;
243            }
244            let committed = SessionStoreWalEntryV1::new(
245                intent.sequence,
246                intent.session_id,
247                intent.snapshot_digest,
248                SessionStoreWalPhaseV1::Committed,
249                intent.recorded_at_ms.saturating_add(1),
250            )?;
251            self.append(&committed).await?;
252            next_sequence = next_sequence.max(committed.sequence + 1);
253        }
254        Ok(next_sequence)
255    }
256}
257
258#[cfg(test)]
259mod tests {
260    use super::*;
261
262    fn digest(ch: char) -> String {
263        format!("sha256:{}", ch.to_string().repeat(64))
264    }
265
266    #[test]
267    fn wal_entry_is_digest_bound_and_rejects_zero_sequence() {
268        let entry = SessionStoreWalEntryV1::new(
269            1,
270            "session-1",
271            digest('a'),
272            SessionStoreWalPhaseV1::Intent,
273            10,
274        )
275        .unwrap();
276        entry.validate().unwrap();
277        assert!(SessionStoreWalEntryV1::new(
278            0,
279            "session-1",
280            digest('a'),
281            SessionStoreWalPhaseV1::Intent,
282            10,
283        )
284        .is_err());
285    }
286}
a3s_code_core/store/wal.rs

a3s_code_core/store/
wal.rs
