superfluid_daemon/
wal.rs

1use std::fs::{File, OpenOptions};
2use std::io::{BufReader, Read, Seek, SeekFrom, Write};
3use std::path::Path;
4
5use serde::{Deserialize, Serialize};
6
7use crate::DaemonError;
8
9#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize, Default)]
10pub struct GenParams {
11    pub temperature: f32,
12    pub top_p: f32,
13    pub min_p: f32,
14    pub top_k: u32,
15    pub seed: u64,
16}
17
18pub mod role {
19    pub const SYSTEM: u32 = 0;
20    pub const USER: u32 = 1;
21    pub const ASSISTANT: u32 = 2;
22    pub const TOOL: u32 = 3;
23}
24
25pub mod channel {
26    pub const TEXT: u32 = 0;
27    pub const REASONING: u32 = 1;
28    pub const TOOL_CALL: u32 = 2;
29}
30
31#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
32pub enum EventBody {
33    Created { parent: Option<u64>, params: GenParams },
34    Appended { text: Option<String>, span: Vec<u32> },
35    Message { role: u32, text: String, span: Vec<u32> },
36    GenerationPrompt { span: Vec<u32> },
37    Generated { span: Vec<u32>, text: String, channel: u32, finish: u32 },
38    ToolUse { name: String, arguments: String },
39    ToolResult { call_id: u64, content: String, span: Vec<u32> },
40    EpochBump,
41    GenerationFingerprint { digest: u64 },
42    ToolParseFailure { raw: String },
43    Forked { parent: u64, fork_at: u64, params: GenParams },
44    MetaUpdated { version: u64, title: Option<String>, archived: Option<bool> },
45    Rebased { parent: u64, fork_at: u64, params: GenParams, edits: Vec<RebaseEdit> },
46    ToolExpired { call_id: u64, reason: String },
47    Purged { generation: u64, descendants: PurgeMode },
48    Rerooted { purged_parent: u64, prefix: Vec<(u64, EventBody)> },
49    QosSet { class: u8, batch_invariant: bool },
50    Block { role: u32, kind: u32, payload: String, span: Vec<u32>, visibility_version: u32 },
51    PermissionRequest { call_id: u64, text: String },
52    PermissionResponse { request_id: u64, granted: bool },
53    ToolCancelRequested { call_id: u64 },
54    ToolOutcome { call_id: u64, outcome: u8, note: String },
55    ToolReconciliation { call_id: u64, note: String },
56    ToolLease { call_id: u64, deadline_unix_ms: u64 },
57    Pinned { deadline_unix_ms: u64 },
58}
59
60impl EventBody {
61    pub fn is_stream_neutral(&self) -> bool {
62        matches!(
63            self,
64            EventBody::EpochBump
65                | EventBody::GenerationFingerprint { .. }
66                | EventBody::QosSet { .. }
67                | EventBody::Pinned { .. }
68                | EventBody::MetaUpdated { .. }
69        )
70    }
71
72    pub fn span(&self) -> Option<&[u32]> {
73        match self {
74            EventBody::Appended { span, .. }
75            | EventBody::Message { span, .. }
76            | EventBody::GenerationPrompt { span }
77            | EventBody::Generated { span, .. }
78            | EventBody::ToolResult { span, .. }
79            | EventBody::Block { span, .. } => Some(span),
80            _ => None,
81        }
82    }
83}
84
85pub mod block_kind {
86    pub const FILE_DIFF: u32 = 1;
87    pub const ARTIFACT: u32 = 2;
88    pub const CITATION: u32 = 3;
89    pub const PROGRESS: u32 = 4;
90    pub const DIAGNOSTIC: u32 = 5;
91    pub const WORKSPACE_REF: u32 = 6;
92    pub const IMAGE: u32 = 7;
93}
94
95pub mod tool_outcome {
96    pub const FAILED: u8 = 1;
97    pub const CANCELLED: u8 = 2;
98}
99
100pub mod ledger_state {
101    pub const OPEN: u8 = 0;
102    pub const CANCEL_REQUESTED: u8 = 1;
103    pub const SUCCEEDED: u8 = 2;
104    pub const FAILED: u8 = 3;
105    pub const CANCELLED: u8 = 4;
106    pub const EXPIRED: u8 = 5;
107    pub const CANCELLED_WITH_LATE_EFFECT: u8 = 6;
108}
109
110#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
111pub enum RebaseEdit {
112    Drop { from: u64, to: u64 },
113    Replace { from: u64, to: u64, role: u32, text: String },
114}
115
116#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
117pub enum PurgeMode {
118    Cascade,
119    Reroot,
120}
121
122#[derive(Debug, Clone, PartialEq)]
123pub struct WalRecord {
124    pub session: u64,
125    pub event_id: u64,
126    pub epoch: u64,
127    pub ts_unix_ms: u64,
128    pub body: EventBody,
129}
130
131#[derive(Serialize, Deserialize)]
132struct WalRecordV1 {
133    session: u64,
134    event_id: u64,
135    epoch: u64,
136    body: EventBody,
137}
138
139#[derive(Serialize, Deserialize)]
140struct WalRecordV2 {
141    session: u64,
142    event_id: u64,
143    epoch: u64,
144    ts_unix_ms: u64,
145    body: EventBody,
146}
147
148pub const WAL_MAGIC_V2: &[u8; 8] = b"BRTWAL02";
149
150pub fn now_unix_ms() -> u64 {
151    std::time::SystemTime::now()
152        .duration_since(std::time::UNIX_EPOCH)
153        .map(|d| d.as_millis() as u64)
154        .unwrap_or(0)
155}
156
157pub struct Wal {
158    file: Option<File>,
159    version: u8,
160}
161
162impl Wal {
163    pub fn ephemeral() -> Wal {
164        Wal {
165            file: None,
166            version: 2,
167        }
168    }
169
170    pub fn open(path: &Path, mut apply: impl FnMut(WalRecord)) -> Result<Wal, DaemonError> {
171        let mut file = OpenOptions::new()
172            .read(true)
173            .write(true)
174            .create(true)
175            .truncate(false)
176            .open(path)?;
177        let len = file.metadata()?.len();
178        let mut reader = BufReader::new(&mut file);
179        let mut good_end: u64 = 0;
180        let version = if len == 0 {
181            2
182        } else {
183            let mut head = [0u8; 8];
184            let mut n = 0;
185            while n < 8 {
186                let k = reader.read(&mut head[n..])?;
187                if k == 0 {
188                    break;
189                }
190                n += k;
191            }
192            if n == 8 && &head == WAL_MAGIC_V2 {
193                good_end = 8;
194                2
195            } else if n < 8 && head[..n] == WAL_MAGIC_V2[..n] {
196                good_end = 0;
197                2
198            } else {
199                drop(reader);
200                file.seek(SeekFrom::Start(0))?;
201                reader = BufReader::new(&mut file);
202                1
203            }
204        };
205        loop {
206            if version == 2 && good_end == 0 {
207                break;
208            }
209            match read_record(&mut reader, version) {
210                Ok(Frame::Record(record, consumed)) => {
211                    good_end += consumed;
212                    apply(record);
213                }
214                Ok(Frame::End) => break,
215                // Only the last append can be torn (each is fsynced before the
216                // next), so a failed frame is that append only if nothing but
217                // zero fill follows it.
218                Ok(Frame::Failed) => {
219                    if !only_zeros(&mut reader)? {
220                        return Err(DaemonError::WalCorrupt);
221                    }
222                    break;
223                }
224                Err(DaemonError::Io(e))
225                    if e.kind() == std::io::ErrorKind::UnexpectedEof =>
226                {
227                    break
228                }
229                Err(e) => return Err(e),
230            }
231        }
232        drop(reader);
233        if good_end < len {
234            tracing::warn!(
235                wal = %path.display(),
236                dropped_bytes = len - good_end,
237                "the session log ended in a torn record; dropped it"
238            );
239            file.set_len(good_end)?;
240        }
241        if len == 0 {
242            file.write_all(WAL_MAGIC_V2)?;
243            file.sync_data()?;
244        } else if version == 2 && good_end < 8 {
245            file.set_len(0)?;
246            file.write_all(WAL_MAGIC_V2)?;
247            file.sync_data()?;
248        }
249        file.seek(SeekFrom::End(0))?;
250        Ok(Wal { file: Some(file), version })
251    }
252
253    pub fn version(&self) -> u8 {
254        self.version
255    }
256
257    pub fn append(&mut self, record: &WalRecord) -> Result<(), DaemonError> {
258        let Some(file) = self.file.as_mut() else {
259            return Ok(());
260        };
261        let payload = if self.version == 1 {
262            postcard::to_stdvec(&WalRecordV1 {
263                session: record.session,
264                event_id: record.event_id,
265                epoch: record.epoch,
266                body: record.body.clone(),
267            })?
268        } else {
269            postcard::to_stdvec(&WalRecordV2 {
270                session: record.session,
271                event_id: record.event_id,
272                epoch: record.epoch,
273                ts_unix_ms: record.ts_unix_ms,
274                body: record.body.clone(),
275            })?
276        };
277        let mut frame = Vec::with_capacity(12 + payload.len());
278        frame.extend_from_slice(&(payload.len() as u32).to_le_bytes());
279        frame.extend_from_slice(&xxhash_rust::xxh3::xxh3_64(&payload).to_le_bytes());
280        frame.extend_from_slice(&payload);
281        file.write_all(&frame)?;
282        durable_flush(file)?;
283        Ok(())
284    }
285}
286
287pub(crate) fn durable_flush(file: &std::fs::File) -> std::io::Result<()> {
288    if std::env::var_os("SUPERFLUID_FULLFSYNC").is_some() {
289        return file.sync_data();
290    }
291    #[cfg(unix)]
292    {
293        use std::os::unix::io::AsRawFd;
294        // SAFETY: `file` is a live, open WAL handle for the duration of the call.
295        let rc = unsafe { libc::fsync(file.as_raw_fd()) };
296        if rc != 0 {
297            return Err(std::io::Error::last_os_error());
298        }
299        Ok(())
300    }
301    #[cfg(not(unix))]
302    {
303        file.sync_data()
304    }
305}
306
307enum Frame {
308    Record(WalRecord, u64),
309    End,
310    Failed,
311}
312
313fn only_zeros(r: &mut impl Read) -> std::io::Result<bool> {
314    let mut buf = [0u8; 8192];
315    loop {
316        match r.read(&mut buf) {
317            Ok(0) => return Ok(true),
318            Ok(n) if buf[..n].iter().any(|&b| b != 0) => return Ok(false),
319            Ok(_) => {}
320            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
321            Err(e) => return Err(e),
322        }
323    }
324}
325
326fn read_record(r: &mut impl Read, version: u8) -> Result<Frame, DaemonError> {
327    let mut len_buf = [0u8; 4];
328    match r.read_exact(&mut len_buf) {
329        Ok(()) => {}
330        Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(Frame::End),
331        Err(e) => return Err(e.into()),
332    }
333    let len = u32::from_le_bytes(len_buf) as usize;
334    if len > MAX_RECORD {
335        return Err(DaemonError::WalCorrupt);
336    }
337    let mut sum_buf = [0u8; 8];
338    r.read_exact(&mut sum_buf)?;
339    let expect = u64::from_le_bytes(sum_buf);
340    let mut payload = vec![0u8; len];
341    r.read_exact(&mut payload)?;
342    if xxhash_rust::xxh3::xxh3_64(&payload) != expect {
343        return Ok(Frame::Failed);
344    }
345    let record = if version == 1 {
346        let r: WalRecordV1 = postcard::from_bytes(&payload)?;
347        WalRecord {
348            session: r.session,
349            event_id: r.event_id,
350            epoch: r.epoch,
351            ts_unix_ms: 0,
352            body: r.body,
353        }
354    } else {
355        let r: WalRecordV2 = postcard::from_bytes(&payload)?;
356        WalRecord {
357            session: r.session,
358            event_id: r.event_id,
359            epoch: r.epoch,
360            ts_unix_ms: r.ts_unix_ms,
361            body: r.body,
362        }
363    };
364    Ok(Frame::Record(record, 12 + len as u64))
365}
366
367const MAX_RECORD: usize = 1 << 30;
superfluid_daemon/wal.rs

superfluid_daemon/
wal.rs
