storage/hot.rs
1// This Source Code Form is subject to the terms of the Mozilla Public
2// License, v. 2.0. If a copy of the MPL was not distributed with this
3// file, You can obtain one at https://mozilla.org/MPL/2.0/.
4// Copyright (c) 2026 2youg1 and the sprawling contributors
5
6//! The in-memory hot view: what the interface asks
7//! for constantly — which Runs exist, which phase each is in, how far
8//! each has got — answered without touching the disk.
9//!
10//! It is a fold over the ledger, nothing more. Feeding the same record
11//! twice changes nothing, because seq only moves forward; that is what
12//! lets a caller replay from any point without first proving where it
13//! left off.
14
15use std::collections::btree_map::Entry;
16use std::collections::{BTreeMap, BTreeSet};
17
18use kernel::{Address, EventKind, EventRecord, RunId, Seq, TimeMs};
19
20use crate::error::StorageError;
21
22/// A Run's phase as the hot view sees it. Freezing is terminal here:
23/// the ledger may keep appending to a frozen Run's history, but the
24/// phase never travels backwards.
25#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
26pub enum RunPhase {
27 Active,
28 Frozen,
29}
30
31#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
32pub struct RunHot {
33 pub phase: RunPhase,
34 pub last_seq: Seq,
35 pub last_kind: EventKind,
36 pub who: String,
37 /// The room this run works in, as its `run_started` record named
38 /// it. `None` until that record is seen: a checkpoint can land before
39 /// the opening, and a window may hold only a tail.
40 pub addr: Option<Address>,
41 /// When the run began, from the same record.
42 pub started: Option<TimeMs>,
43 /// How the run ended, from its `run_frozen` record's `completion`.
44 pub completion: Option<String>,
45 /// The branch of the last pull request the run opened: the city's
46 /// pull requests are named by their branch and carry no number.
47 pub pr: Option<String>,
48 /// What the run waits for the person to allow. Present exactly while
49 /// the run's last record is an `approval_requested`, so it cannot say
50 /// a run waits when `last_kind` says it moved on.
51 pub ask: Option<String>,
52 /// What the person asked for, from the `run_started` record's
53 /// `task`; `None` when that record is unseen or the task is empty.
54 pub task: Option<String>,
55 /// What finishing looks like, from the same record's `goal`, read
56 /// the same way.
57 pub goal: Option<String>,
58 /// The room this run stopped on a synchronous `send` to hear from,
59 /// and until when: present from a `signal_wait_started` until a
60 /// `signal_wait_ended` or the freeze, so a frozen run never waits
61 /// (`crates/storage/spec/Hot.lean` D29, `frozen_run_never_waits`).
62 /// Defaulted, because a hot view stored before the field existed
63 /// had no waits to remember.
64 #[serde(default)]
65 pub waiting: Option<RunWaiting>,
66}
67
68/// What a run waits for: the room it spoke to and the instant on the
69/// city's clock at which it stops waiting.
70#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
71pub struct RunWaiting {
72 pub on: Address,
73 pub until: TimeMs,
74}
75
76impl RunWaiting {
77 /// The wait a `signal_wait_started` record states, read by key like
78 /// the rest of the fold; `None` when either field is missing or does
79 /// not read, because the view does not guess what it cannot see.
80 fn stated_by(record: &EventRecord) -> Option<RunWaiting> {
81 let data = record.data().as_map();
82 let on = data
83 .get("on")
84 .and_then(serde_json::Value::as_str)
85 .and_then(|on| Address::parse(on).ok())?;
86 let until = data
87 .get("deadline_ms")
88 .and_then(serde_json::Value::as_u64)
89 .map(TimeMs::new)?;
90 Some(RunWaiting { on, until })
91 }
92}
93
94impl RunHot {
95 fn first_seen(record: &EventRecord) -> RunHot {
96 RunHot {
97 phase: RunPhase::Active,
98 last_seq: record.seq(),
99 last_kind: record.kind(),
100 who: record.who().to_owned(),
101 addr: None,
102 started: None,
103 completion: None,
104 pr: None,
105 ask: None,
106 task: None,
107 goal: None,
108 waiting: None,
109 }
110 }
111
112 /// Folds one record that is newer than everything this run has seen.
113 fn absorb(&mut self, record: &EventRecord) {
114 let kind = record.kind();
115 self.last_seq = record.seq();
116 self.last_kind = kind;
117 if kind == EventKind::RunFrozen {
118 self.phase = RunPhase::Frozen;
119 }
120 if kind == EventKind::RunStarted {
121 self.addr = record.addr().cloned();
122 self.started = Some(record.t());
123 }
124 let stated = |field: &str| {
125 record
126 .data()
127 .as_map()
128 .get(field)
129 .and_then(serde_json::Value::as_str)
130 .map(str::to_owned)
131 };
132 if kind == EventKind::RunStarted {
133 let named = |field: &str| stated(field).filter(|said| !said.is_empty());
134 self.task = named("task");
135 self.goal = named("goal");
136 }
137 if kind == EventKind::RunFrozen {
138 self.completion = stated("completion");
139 }
140 if kind == EventKind::PrOpened {
141 self.pr = stated("branch");
142 }
143 self.ask = (kind == EventKind::ApprovalRequested)
144 .then(|| stated("action_desc"))
145 .flatten();
146 if kind == EventKind::SignalWaitStarted && self.phase == RunPhase::Active {
147 self.waiting = RunWaiting::stated_by(record);
148 }
149 if kind == EventKind::SignalWaitEnded || kind == EventKind::RunFrozen {
150 self.waiting = None;
151 }
152 }
153}
154
155/// How many frozen runs the hot view holds besides every active one,
156/// and so how many a city view carries. A bound on the size of an
157/// answer on the wire, not a machine reading, so it is a constant
158/// (`crates/storage/spec/Hot.lean` §8-5).
159pub const RECENT_FROZEN: usize = 32;
160
161#[derive(Clone, Default, serde::Serialize, serde::Deserialize)]
162pub struct HotView {
163 runs: BTreeMap<RunId, RunHot>,
164 /// Frozen runs pushed out of `runs`, by id alone. Freezing is
165 /// terminal, so a later record on one of these is the tail of an old
166 /// run, never the checkpoint of a new one.
167 evicted: BTreeSet<RunId>,
168}
169
170impl HotView {
171 pub fn new() -> HotView {
172 HotView::default()
173 }
174
175 /// Folds one record in. Records at or below a Run's last seq are
176 /// ignored, so re-feeding a segment is free of consequence.
177 pub fn apply(&mut self, record: &EventRecord) -> Result<(), StorageError> {
178 let run = record.run();
179 // A city-level record belongs to the city, not to a run: raising a
180 // building and the genesis record both carry the nil id, and the
181 // Ledger says so (`kernel::event`: "RunId::CITY (nil) marks
182 // city-level records"). Admitting one here invented a run nobody
183 // started, and the count it fed said a city was working the
184 // moment it existed.
185 if run == RunId::CITY || self.evicted.contains(&run) {
186 return Ok(());
187 }
188 let kind = record.kind();
189 match self.runs.entry(run) {
190 Entry::Occupied(held) => {
191 let hot = held.into_mut();
192 if record.seq() > hot.last_seq {
193 hot.absorb(record);
194 }
195 }
196 Entry::Vacant(slot) => slot.insert(RunHot::first_seen(record)).absorb(record),
197 }
198 if kind == EventKind::RunFrozen {
199 self.evict_the_oldest_frozen_beyond_recent();
200 }
201 Ok(())
202 }
203
204 /// Keeps at most [`RECENT_FROZEN`] frozen runs, tombstoning the one
205 /// with the lowest `last_seq`. One freeze adds one frozen run, so one
206 /// eviction restores the bound; the scan is O(active + RECENT_FROZEN).
207 fn evict_the_oldest_frozen_beyond_recent(&mut self) {
208 let frozen = || {
209 self.runs
210 .iter()
211 .filter(|(_, hot)| hot.phase == RunPhase::Frozen)
212 };
213 if frozen().count() <= RECENT_FROZEN {
214 return;
215 }
216 let Some(oldest) = frozen()
217 .min_by_key(|(_, hot)| hot.last_seq)
218 .map(|(run, _)| *run)
219 else {
220 return;
221 };
222 self.runs.remove(&oldest);
223 self.evicted.insert(oldest);
224 }
225
226 /// Every active run and the [`RECENT_FROZEN`] frozen runs with the
227 /// latest `last_seq`: what a city view carries. Iteration is in RunId
228 /// order — the same order on every process, so a rendered list never
229 /// reshuffles between restarts.
230 pub fn runs(&self) -> impl Iterator<Item = (&RunId, &RunHot)> {
231 self.runs.iter()
232 }
233
234 pub fn get(&self, run: &RunId) -> Option<&RunHot> {
235 self.runs.get(run)
236 }
237
238 /// Whether `run` froze and was pushed out of the view. Its records
239 /// are still in the Ledger; a reader that needs them goes there.
240 pub fn was_evicted(&self, run: &RunId) -> bool {
241 self.evicted.contains(run)
242 }
243
244 pub fn active_count(&self) -> u64 {
245 self.count_phase(RunPhase::Active)
246 }
247
248 pub fn frozen_count(&self) -> u64 {
249 let evicted = u64::try_from(self.evicted.len()).unwrap_or(u64::MAX);
250 self.count_phase(RunPhase::Frozen).saturating_add(evicted)
251 }
252
253 fn count_phase(&self, phase: RunPhase) -> u64 {
254 u64::try_from(self.runs.values().filter(|h| h.phase == phase).count()).unwrap_or(u64::MAX)
255 }
256}
257
258#[cfg(test)]
259#[allow(
260 clippy::unwrap_used,
261 clippy::expect_used,
262 clippy::panic,
263 clippy::indexing_slicing,
264 reason = "test code"
265)]
266mod tests;