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

unifier/
tick.rs
