unifier/
tick.rs

1//! ACID tick staging: read previous tick at start, write current tick in memory, commit at end.
2
3use std::collections::{BTreeMap, BTreeSet};
4use std::fs;
5use std::time::{Duration, SystemTime, UNIX_EPOCH};
6
7use chrono::Utc;
8use serde::{Deserialize, Serialize};
9use serde_json::Value;
10use uuid::Uuid;
11
12use crate::constants::{DEFAULT_EVENT_TTL_SECS, EVENTS, KEYS, TICKS, TICK_CURRENT};
13use crate::error::{Error, Result};
14use crate::fs_text::{read_text, write_text, write_text_atomic};
15use crate::home::UnifierHome;
16use crate::store::{validate_key, ActiveTick, EventState, HotStore, KeyState, StagingValue};
17
18#[derive(Debug, Clone, PartialEq, Eq)]
19pub enum TickStartOutcome {
20    Started { tick: u64 },
21    Queued { position: usize, label: String },
22}
23
24#[derive(Debug, Clone, PartialEq, Eq)]
25pub struct TickStatus {
26    pub committed_tick: u64,
27    pub active_tick: Option<u64>,
28    pub queued: usize,
29    pub locked_keys: Vec<String>,
30}
31
32#[derive(Serialize)]
33struct TickMeta {
34    tick: u64,
35    committed_at: String,
36    keys_written: usize,
37}
38
39impl HotStore {
40    pub fn tick_start(&mut self, label: &str) -> Result<TickStartOutcome> {
41        if self.active_tick.is_some() {
42            self.tick_queue.push_back(label.to_string());
43            return Ok(TickStartOutcome::Queued {
44                position: self.tick_queue.len(),
45                label: label.to_string(),
46            });
47        }
48        let tick = self.begin_tick()?;
49        Ok(TickStartOutcome::Started { tick })
50    }
51
52    pub fn tick_end(&mut self, home: &UnifierHome) -> Result<u64> {
53        let Some(active) = self.active_tick.take() else {
54            return Err(Error::msg("no active tick"));
55        };
56
57        for (key, staged) in &active.staging {
58            match staged {
59                StagingValue::Present(value) => {
60                    self.keys.insert(
61                        key.clone(),
62                        KeyState::Present {
63                            value: value.clone(),
64                            dirty: true,
65                        },
66                    );
67                }
68                StagingValue::Deleted => {
69                    if self.keys.contains_key(key) {
70                        self.keys
71                            .insert(key.clone(), KeyState::Deleted { dirty: true });
72                    }
73                }
74            }
75        }
76
77        self.committed_tick = active.number;
78        commit_tick_version(home, active.number, &active.staging)?;
79        write_committed_tick(home, active.number)?;
80
81        if let Some(next) = self.tick_queue.pop_front() {
82            eprintln!(
83                "tick queue: starting queued tick {:?} ({} remaining)",
84                next,
85                self.tick_queue.len()
86            );
87            self.begin_tick()?;
88        }
89
90        Ok(active.number)
91    }
92
93    pub fn tick_status(&self) -> TickStatus {
94        TickStatus {
95            committed_tick: self.committed_tick,
96            active_tick: self.active_tick.as_ref().map(|t| t.number),
97            queued: self.tick_queue.len(),
98            locked_keys: self
99                .active_tick
100                .as_ref()
101                .map(|t| t.locks.iter().cloned().collect())
102                .unwrap_or_default(),
103        }
104    }
105
106    pub fn tick_lock(&mut self, key: &str) -> Result<()> {
107        let Some(tick) = &mut self.active_tick else {
108            return Err(Error::msg("no active tick"));
109        };
110        validate_key(key)?;
111        tick.locks.insert(key.to_string());
112        Ok(())
113    }
114
115    pub fn tick_unlock(&mut self, key: &str) -> Result<bool> {
116        let Some(tick) = &mut self.active_tick else {
117            return Err(Error::msg("no active tick"));
118        };
119        Ok(tick.locks.remove(key))
120    }
121
122    pub fn post_event(&mut self, payload: &str, ttl_secs: Option<u64>) -> Result<Uuid> {
123        let value: Value = serde_json::from_str(payload)
124            .map_err(|e| Error::msg(format!("event payload must be valid JSON: {e}")))?;
125        let created_at = Utc::now().to_rfc3339();
126        let expires_at = resolve_event_expiry(&value, ttl_secs)?;
127        let id = Uuid::new_v4();
128        self.events.insert(
129            id,
130            EventState {
131                body: payload.to_string(),
132                created_at,
133                expires_at,
134                dirty: true,
135            },
136        );
137        Ok(id)
138    }
139
140    /// Drop expired events from memory and delete their files. Returns how many were removed.
141    pub fn expire_events(&mut self, home: &UnifierHome) -> Result<usize> {
142        let mut gone = Vec::new();
143        for (id, event) in &self.events {
144            if event_is_expired(event.expires_at.as_deref()) {
145                gone.push(*id);
146            }
147        }
148        let n = gone.len();
149        if n == 0 {
150            return Ok(0);
151        }
152        let root = home.path().join(EVENTS);
153        for id in gone {
154            self.events.remove(&id);
155            let path = root.join(format!("{}.json", id.hyphenated()));
156            if path.is_file() {
157                fs::remove_file(path)?;
158            }
159        }
160        Ok(n)
161    }
162
163    pub fn send_agent_message(&mut self, from: &str, to: &str, payload: &str) -> Result<Uuid> {
164        serde_json::from_str::<serde_json::Value>(payload)
165            .map_err(|e| Error::msg(format!("message payload must be valid JSON: {e}")))?;
166        self.send_from(from, to, payload)
167    }
168
169    pub(crate) fn flush_events(&mut self, home: &UnifierHome) -> Result<()> {
170        self.expire_events(home)?;
171        let root = home.path().join(EVENTS);
172        for (id, event) in self.events.iter_mut() {
173            if event.dirty {
174                if let Some(p) = root.parent() {
175                    let _ = fs::create_dir_all(p);
176                }
177                fs::create_dir_all(&root)?;
178                let record = EventRecord {
179                    payload: serde_json::from_str(&event.body)
180                        .unwrap_or(Value::String(event.body.clone())),
181                    created_at: event.created_at.clone(),
182                    expires_at: event.expires_at.clone(),
183                };
184                write_text(
185                    &root.join(format!("{}.json", id.hyphenated())),
186                    &serde_json::to_string(&record)?,
187                )?;
188                event.dirty = false;
189            }
190        }
191        Ok(())
192    }
193
194    fn begin_tick(&mut self) -> Result<u64> {
195        let number = self.committed_tick.saturating_add(1);
196        let read_snapshot = self.committed_key_snapshot();
197        self.active_tick = Some(ActiveTick {
198            number,
199            read_snapshot,
200            staging: BTreeMap::new(),
201            locks: BTreeSet::new(),
202        });
203        Ok(number)
204    }
205
206    fn committed_key_snapshot(&self) -> BTreeMap<String, String> {
207        self.keys
208            .iter()
209            .filter_map(|(k, state)| match state {
210                KeyState::Present { value, .. } => Some((k.clone(), value.clone())),
211                KeyState::Deleted { .. } => None,
212            })
213            .collect()
214    }
215}
216
217pub(crate) fn load_committed_tick(home: &UnifierHome) -> Result<u64> {
218    let path = home.path().join(TICKS).join(TICK_CURRENT);
219    if !path.is_file() {
220        return Ok(0);
221    }
222    let text = read_text(&path)?;
223    text.parse::<u64>()
224        .map_err(|_| Error::msg("invalid ticks/current"))
225}
226
227pub(crate) fn load_events(home: &UnifierHome) -> Result<BTreeMap<Uuid, EventState>> {
228    let mut events = BTreeMap::new();
229    let root = home.path().join(EVENTS);
230    if !root.is_dir() {
231        return Ok(events);
232    }
233    for entry in fs::read_dir(&root)? {
234        let entry = entry?;
235        if !entry.file_type()?.is_file() {
236            continue;
237        }
238        let name = entry.file_name().to_string_lossy().into_owned();
239        let Some(stem) = name.strip_suffix(".json") else {
240            continue;
241        };
242        let Ok(id) = Uuid::parse_str(stem) else {
243            continue;
244        };
245        let path = entry.path();
246        let text = read_text(&path)?;
247        let Some(state) = parse_event_file(&text, &path)? else {
248            let _ = fs::remove_file(&path);
249            continue;
250        };
251        if event_is_expired(state.expires_at.as_deref()) {
252            let _ = fs::remove_file(&path);
253            continue;
254        }
255        events.insert(id, state);
256    }
257    Ok(events)
258}
259
260#[derive(Serialize, Deserialize)]
261struct EventRecord {
262    payload: Value,
263    created_at: String,
264    #[serde(default, skip_serializing_if = "Option::is_none")]
265    expires_at: Option<String>,
266}
267
268fn parse_event_file(text: &str, path: &std::path::Path) -> Result<Option<EventState>> {
269    let value: Value = match serde_json::from_str(text) {
270        Ok(v) => v,
271        Err(_) => {
272            return Ok(Some(EventState {
273                body: text.to_string(),
274                created_at: Utc::now().to_rfc3339(),
275                expires_at: expires_from_mtime(path),
276                dirty: false,
277            }));
278        }
279    };
280    if is_event_record(&value) {
281        let record: EventRecord = serde_json::from_value(value)?;
282        return Ok(Some(EventState {
283            body: serde_json::to_string(&record.payload)?,
284            created_at: record.created_at,
285            expires_at: record.expires_at,
286            dirty: false,
287        }));
288    }
289    Ok(Some(EventState {
290        body: text.trim().to_string(),
291        created_at: Utc::now().to_rfc3339(),
292        expires_at: expires_from_mtime(path),
293        dirty: false,
294    }))
295}
296
297fn is_event_record(value: &Value) -> bool {
298    value.get("payload").is_some() && value.get("created_at").is_some()
299}
300
301/// Seconds until events expire when `--ttl` and payload fields are omitted.
302/// Override with `$UNIFIER_EVENT_TTL_SECS` (`0` = never expire by default).
303pub fn default_event_ttl_secs() -> u64 {
304    std::env::var("UNIFIER_EVENT_TTL_SECS")
305        .ok()
306        .and_then(|s| s.parse().ok())
307        .unwrap_or(DEFAULT_EVENT_TTL_SECS)
308}
309
310fn resolve_event_expiry(payload: &Value, ttl_flag: Option<u64>) -> Result<Option<String>> {
311    if let Some(secs) = ttl_flag {
312        return Ok(ttl_to_expires_at(secs));
313    }
314    if let Some(exp) = payload.get("expires_at") {
315        if exp.is_null() {
316            return Ok(None);
317        }
318        if let Some(s) = exp.as_str() {
319            chrono::DateTime::parse_from_rfc3339(s)
320                .map_err(|e| Error::msg(format!("invalid expires_at: {e}")))?;
321            return Ok(Some(s.to_string()));
322        }
323        return Err(Error::msg("expires_at must be an RFC3339 string or null"));
324    }
325    if let Some(ttl) = payload.get("ttl") {
326        let secs = ttl
327            .as_u64()
328            .ok_or_else(|| Error::msg("ttl must be a non-negative integer"))?;
329        return Ok(ttl_to_expires_at(secs));
330    }
331    Ok(ttl_to_expires_at(default_event_ttl_secs()))
332}
333
334fn ttl_to_expires_at(secs: u64) -> Option<String> {
335    if secs == 0 {
336        return None;
337    }
338    Some(rfc3339_from_system(SystemTime::now() + Duration::from_secs(secs)))
339}
340
341fn expires_from_mtime(path: &std::path::Path) -> Option<String> {
342    let ttl = default_event_ttl_secs();
343    if ttl == 0 {
344        return None;
345    }
346    let modified = fs::metadata(path)
347        .ok()
348        .and_then(|m| m.modified().ok())
349        .unwrap_or_else(SystemTime::now);
350    Some(rfc3339_from_system(modified + Duration::from_secs(ttl)))
351}
352
353fn rfc3339_from_system(t: SystemTime) -> String {
354    let secs = t.duration_since(UNIX_EPOCH).unwrap_or_default().as_secs() as i64;
355    chrono::DateTime::from_timestamp(secs, 0)
356        .unwrap_or_else(Utc::now)
357        .to_rfc3339()
358}
359
360fn event_is_expired(expires_at: Option<&str>) -> bool {
361    let Some(exp) = expires_at else {
362        return false;
363    };
364    match chrono::DateTime::parse_from_rfc3339(exp) {
365        Ok(dt) => dt.with_timezone(&Utc) < Utc::now(),
366        Err(_) => false,
367    }
368}
369
370fn write_committed_tick(home: &UnifierHome, tick: u64) -> Result<()> {
371    let dir = home.path().join(TICKS);
372    fs::create_dir_all(&dir)?;
373    write_text_atomic(&dir.join(TICK_CURRENT), &tick.to_string())
374}
375
376fn commit_tick_version(
377    home: &UnifierHome,
378    tick: u64,
379    staging: &BTreeMap<String, StagingValue>,
380) -> Result<()> {
381    let tick_root = home.path().join(TICKS).join(tick.to_string());
382    let tick_keys = tick_root.join(KEYS);
383    fs::create_dir_all(&tick_keys)?;
384
385    for (key, staged) in staging {
386        match staged {
387            StagingValue::Present(value) => {
388                let dest = tick_keys.join(key);
389                if let Some(p) = dest.parent() {
390                    fs::create_dir_all(p)?;
391                }
392                write_text_atomic(&dest, value)?;
393            }
394            StagingValue::Deleted => {}
395        }
396    }
397
398    let meta = TickMeta {
399        tick,
400        committed_at: Utc::now().to_rfc3339(),
401        keys_written: staging.len(),
402    };
403    write_text(
404        &tick_root.join("meta.json"),
405        &serde_json::to_string_pretty(&meta)?,
406    )?;
407    Ok(())
408}
409
410#[cfg(test)]
411mod tests {
412    use super::*;
413    use crate::home::UnifierHome;
414    use tempfile::tempdir;
415
416    #[test]
417    fn tick_reads_previous_committed_state_only() {
418        let tmp = tempdir().unwrap();
419        let home = UnifierHome::resolve(Some(tmp.path().to_path_buf()), None).unwrap();
420        let mut store = HotStore::load(&home).unwrap();
421
422        store.put_key("counter", "1").unwrap();
423        store.tick_start("t1").unwrap();
424        assert_eq!(store.get_key("counter").unwrap(), Some("1".into()));
425        store.put_key("counter", "2").unwrap();
426        assert_eq!(store.get_key("counter").unwrap(), Some("2".into()));
427
428        store.tick_end(&home).unwrap();
429        assert_eq!(store.get_key("counter").unwrap(), Some("2".into()));
430        assert!(home.path().join("ticks/1/meta.json").is_file());
431    }
432
433    #[test]
434    fn locked_key_rejects_put() {
435        let tmp = tempdir().unwrap();
436        let home = UnifierHome::resolve(Some(tmp.path().to_path_buf()), None).unwrap();
437        let mut store = HotStore::load(&home).unwrap();
438        store.tick_start("t1").unwrap();
439        store.tick_lock("config/x").unwrap();
440        assert!(store.put_key("config/x", "y").is_err());
441    }
442
443    #[test]
444    fn post_event_defaults_to_24h_expiry() {
445        let tmp = tempdir().unwrap();
446        let home = UnifierHome::resolve(Some(tmp.path().to_path_buf()), None).unwrap();
447        let mut store = HotStore::load(&home).unwrap();
448        let id = store.post_event(r#"{"name":"ping"}"#, None).unwrap();
449        let event = store.events.get(&id).unwrap();
450        let exp = event.expires_at.as_ref().unwrap();
451        let dt = chrono::DateTime::parse_from_rfc3339(exp).unwrap();
452        let delta = dt.with_timezone(&Utc) - Utc::now();
453        assert!(delta.num_seconds() > 23 * 3600);
454        assert!(delta.num_seconds() <= 24 * 3600);
455    }
456
457    #[test]
458    fn post_event_ttl_zero_never_expires() {
459        let tmp = tempdir().unwrap();
460        let home = UnifierHome::resolve(Some(tmp.path().to_path_buf()), None).unwrap();
461        let mut store = HotStore::load(&home).unwrap();
462        let id = store.post_event(r#"{"name":"keep"}"#, Some(0)).unwrap();
463        assert_eq!(store.events.get(&id).unwrap().expires_at, None);
464    }
465
466    #[test]
467    fn payload_expires_at_is_honored() {
468        let v = serde_json::json!({"name":"x","expires_at":"2099-01-01T00:00:00Z"});
469        let exp = resolve_event_expiry(&v, None).unwrap();
470        assert_eq!(exp.as_deref(), Some("2099-01-01T00:00:00Z"));
471    }
472
473    #[test]
474    fn expired_events_are_deleted_on_load() {
475        let tmp = tempdir().unwrap();
476        let home = UnifierHome::resolve(Some(tmp.path().to_path_buf()), None).unwrap();
477        let dir = home.path().join("events");
478        std::fs::create_dir_all(&dir).unwrap();
479        let id = Uuid::new_v4();
480        let record = EventRecord {
481            payload: serde_json::json!({"name":"stale"}),
482            created_at: "2000-01-01T00:00:00Z".into(),
483            expires_at: Some("2000-01-02T00:00:00Z".into()),
484        };
485        std::fs::write(
486            dir.join(format!("{}.json", id.hyphenated())),
487            serde_json::to_string(&record).unwrap(),
488        )
489        .unwrap();
490        let store = HotStore::load(&home).unwrap();
491        assert!(store.events.is_empty());
492        assert!(!dir.join(format!("{}.json", id.hyphenated())).exists());
493    }
494
495    #[test]
496    fn flush_drops_expired_events() {
497        let tmp = tempdir().unwrap();
498        let home = UnifierHome::resolve(Some(tmp.path().to_path_buf()), None).unwrap();
499        let mut store = HotStore::load(&home).unwrap();
500        let id = store.post_event(r#"{"name":"soon"}"#, Some(3600)).unwrap();
501        store.events.get_mut(&id).unwrap().expires_at = Some("2000-01-01T00:00:00Z".into());
502        store.flush(&home).unwrap();
503        assert!(store.events.is_empty());
504    }
505
506    #[test]
507    fn expire_events_drops_past_due() {
508        let tmp = tempdir().unwrap();
509        let home = UnifierHome::resolve(Some(tmp.path().to_path_buf()), None).unwrap();
510        let mut store = HotStore::load(&home).unwrap();
511        let id = store.post_event(r#"{"name":"soon"}"#, Some(3600)).unwrap();
512        store.flush(&home).unwrap();
513        store.events.get_mut(&id).unwrap().expires_at = Some("2000-01-01T00:00:00Z".into());
514        assert_eq!(store.expire_events(&home).unwrap(), 1);
515        assert!(store.events.is_empty());
516        assert!(!home
517            .path()
518            .join("events")
519            .join(format!("{}.json", id.hyphenated()))
520            .exists());
521    }
522}
unifier/tick.rs

unifier/
tick.rs
