1use crate::layout::{layout_graph, GraphLayout};
4use crate::protocol::{apply_patch, PageKind, PatchOp};
5use crate::source_loader::{LoadRequest, SourceLoader};
6use crate::state::reader::{
7 LoadedRun, ProjectionCursors, ProjectionPage, ProjectionReader, RunIndexRow, ViewerDeltaRead,
8 ViewerRevisionDelta,
9};
10use crate::state::types::{
11 DefinitionSnapshot, Manifest, RunState, SessionBinding, SessionCapture, SessionEntryRecord,
12 SessionEventRecord,
13};
14use anyhow::{bail, Result};
15use chrono::Utc;
16use serde_json::{json, Value};
17use std::collections::{BTreeMap, BTreeSet};
18use std::path::{Path, PathBuf};
19
20pub type WindowCursor = ProjectionCursors;
21
22pub struct RunEntry {
23 pub dir: PathBuf,
24 pub manifest: Manifest,
25 pub manifest_raw: Value,
26 pub workflow: Value,
27 pub graph_layout: Option<GraphLayout>,
28 pub state_raw: Value,
29 pub graph_steps: Vec<crate::state::types::StepRecord>,
30 pub taken_transitions: Vec<String>,
31 pub graph_cursor: u64,
32 pub step_start: u64,
33 pub step_total: u64,
34 pub events: Vec<Value>,
35 pub trace_start: u64,
36 pub trace_total: u64,
37 pub session_binding: Option<Value>,
38 pub session_entries: Vec<Value>,
39 pub session_entry_start: u64,
40 pub session_entry_total: u64,
41 pub session_events: Vec<Value>,
42 pub session_event_start: u64,
43 pub session_event_total: u64,
44 pub session_events_malformed: bool,
45 pub session_events_torn_tail: bool,
46 pub session_capture: Option<Value>,
47 pub session_replay_checkpoint: Option<Value>,
48 pub settings_scopes: Vec<Value>,
49 pub settings_start: u64,
50 pub settings_total: u64,
51 pub follow_up_queue: Option<Value>,
52 pub follow_up_start: u64,
53 pub follow_up_total: u64,
54 pub update_start: u64,
55 pub update_total: u64,
56 pub state: RunState,
57 pub snapshot: Option<DefinitionSnapshot>,
58 pub live: bool,
59 pub possibly_interrupted: bool,
60 pub revision: u64,
61 pub graph_revision: u64,
62}
63
64impl RunEntry {
65 fn from_loaded(database_path: &Path, loaded: LoadedRun) -> Result<Self> {
66 let manifest_raw = serde_json::to_value(&loaded.manifest)?;
67 let graph_layout = loaded.snapshot.as_ref().map(layout_graph);
68 let workflow = loaded
69 .snapshot
70 .as_ref()
71 .map(serde_json::to_value)
72 .transpose()?
73 .unwrap_or(Value::Null);
74 let state_raw = serde_json::to_value(&loaded.state)?;
75 let events = loaded
76 .trace
77 .iter()
78 .map(serde_json::to_value)
79 .collect::<Result<Vec<_>, _>>()?;
80 let session_binding = loaded
81 .session_binding
82 .as_ref()
83 .map(serde_json::to_value)
84 .transpose()?;
85 let session_entries = loaded
86 .session_entries
87 .iter()
88 .map(serde_json::to_value)
89 .collect::<Result<Vec<_>, _>>()?;
90 let session_events = loaded
91 .session_events
92 .iter()
93 .map(serde_json::to_value)
94 .collect::<Result<Vec<_>, _>>()?;
95 let session_capture = loaded
96 .session_capture
97 .as_ref()
98 .map(serde_json::to_value)
99 .transpose()?;
100 let live = !loaded.state.status.is_terminal();
101 Ok(Self {
102 dir: database_path.to_path_buf(),
103 manifest: loaded.manifest,
104 manifest_raw,
105 workflow,
106 graph_layout,
107 state_raw,
108 graph_steps: loaded.graph_steps,
109 taken_transitions: loaded.taken_transitions,
110 graph_cursor: loaded.graph_cursor,
111 step_start: loaded.step_start,
112 step_total: loaded.step_total,
113 events,
114 trace_start: loaded.trace_start,
115 trace_total: loaded.trace_total,
116 session_binding,
117 session_entries,
118 session_entry_start: loaded.session_entry_start,
119 session_entry_total: loaded.session_entry_total,
120 session_events,
121 session_event_start: loaded.session_event_start,
122 session_event_total: loaded.session_event_total,
123 session_events_malformed: false,
124 session_events_torn_tail: false,
125 session_capture,
126 session_replay_checkpoint: loaded.session_replay_checkpoint,
127 settings_scopes: loaded.settings_scopes,
128 settings_start: loaded.settings_start,
129 settings_total: loaded.settings_total,
130 follow_up_queue: loaded.follow_up_queue,
131 follow_up_start: loaded.follow_up_start,
132 follow_up_total: loaded.follow_up_total,
133 update_start: loaded.update_start,
134 update_total: loaded.update_total,
135 state: loaded.state,
136 snapshot: loaded.snapshot,
137 live,
138 possibly_interrupted: loaded.possibly_interrupted,
139 revision: loaded.presentation_revision,
140 graph_revision: loaded.presentation_revision,
141 })
142 }
143
144 fn session_value(&self) -> Value {
145 if self.session_binding.is_none()
146 && self.session_entries.is_empty()
147 && self.session_events.is_empty()
148 && self.session_capture.is_none()
149 {
150 Value::Null
151 } else {
152 json!({
153 "binding": self.session_binding,
154 "presentationRevision": self.revision,
155 "entryPage": {
156 "presentationRevision": self.revision,
157 "start": self.session_entry_start,
158 "total": self.session_entry_total,
159 "items": self.session_entries,
160 },
161 "eventPage": {
162 "presentationRevision": self.revision,
163 "start": self.session_event_start,
164 "total": self.session_event_total,
165 "items": self.session_events,
166 },
167 "eventsMalformed": self.session_events_malformed,
168 "eventsTornTail": self.session_events_torn_tail,
169 "capture": self.session_capture,
170 "replayCheckpoint": self.session_replay_checkpoint,
171 })
172 }
173 }
174
175 fn apply_root_patch(&mut self, patch: &[PatchOp]) -> bool {
176 let mut document = self.view();
177 if apply_patch(&mut document, patch).is_err() {
178 return false;
179 }
180 let Some(manifest_raw) = document.get("manifest").cloned() else {
181 return false;
182 };
183 let Some(state_raw) = document.get("state").cloned() else {
184 return false;
185 };
186 let Ok(manifest) = serde_json::from_value(manifest_raw.clone()) else {
187 return false;
188 };
189 let Ok(state) = serde_json::from_value(state_raw.clone()) else {
190 return false;
191 };
192 let Ok(graph_steps) = serde_json::from_value(
193 document
194 .get("graphSteps")
195 .cloned()
196 .unwrap_or_else(|| json!([])),
197 ) else {
198 return false;
199 };
200 let Ok(taken_transitions) = serde_json::from_value(
201 document
202 .get("takenTransitions")
203 .cloned()
204 .unwrap_or_else(|| json!([])),
205 ) else {
206 return false;
207 };
208 let Some(graph_revision) = document.get("graphRevision").and_then(Value::as_u64) else {
209 return false;
210 };
211 let Some(graph_cursor) = document.get("graphCursor").and_then(Value::as_u64) else {
212 return false;
213 };
214 let Some(step_start) = document.get("stepStart").and_then(Value::as_u64) else {
215 return false;
216 };
217 let Some(step_total) = document.get("stepTotal").and_then(Value::as_u64) else {
218 return false;
219 };
220 let Some(update_start) = document.get("updateStart").and_then(Value::as_u64) else {
221 return false;
222 };
223 let Some(update_total) = document.get("updateTotal").and_then(Value::as_u64) else {
224 return false;
225 };
226 let Some(live) = document.get("live").and_then(Value::as_bool) else {
227 return false;
228 };
229
230 self.manifest_raw = manifest_raw;
231 self.manifest = manifest;
232 self.state_raw = state_raw;
233 self.state = state;
234 self.graph_steps = graph_steps;
235 self.taken_transitions = taken_transitions;
236 self.graph_revision = graph_revision;
237 self.graph_cursor = graph_cursor;
238 self.step_start = step_start;
239 self.step_total = step_total;
240 self.update_start = update_start;
241 self.update_total = update_total;
242 self.settings_scopes = document
243 .get("settingsScopes")
244 .and_then(Value::as_array)
245 .cloned()
246 .unwrap_or_default();
247 self.settings_start = document
248 .get("settingsStart")
249 .and_then(Value::as_u64)
250 .unwrap_or(0);
251 self.settings_total = document
252 .get("settingsTotal")
253 .and_then(Value::as_u64)
254 .unwrap_or(self.settings_scopes.len() as u64);
255 self.follow_up_queue = document
256 .get("followUpQueue")
257 .cloned()
258 .filter(|value| !value.is_null());
259 self.follow_up_start = document
260 .get("followUpStart")
261 .and_then(Value::as_u64)
262 .unwrap_or(0);
263 self.follow_up_total = document
264 .get("followUpTotal")
265 .and_then(Value::as_u64)
266 .unwrap_or_else(|| {
267 self.follow_up_queue
268 .as_ref()
269 .and_then(|queue| queue.get("items"))
270 .and_then(Value::as_array)
271 .map_or(0, |items| items.len() as u64)
272 });
273 self.live = live;
274 if let Some(possibly_interrupted) =
275 document.get("possiblyInterrupted").and_then(Value::as_bool)
276 {
277 self.possibly_interrupted = possibly_interrupted;
278 }
279 if let Some(session) = document.get("session") {
280 if session.is_null() {
281 self.session_binding = None;
282 self.session_entries.clear();
283 self.session_entry_start = 0;
284 self.session_entry_total = 0;
285 self.session_events.clear();
286 self.session_event_start = 0;
287 self.session_event_total = 0;
288 self.session_capture = None;
289 self.session_replay_checkpoint = None;
290 } else {
291 self.session_binding = session
292 .get("binding")
293 .cloned()
294 .filter(|value| !value.is_null());
295 self.session_entries = session
296 .pointer("/entryPage/items")
297 .and_then(Value::as_array)
298 .cloned()
299 .unwrap_or_default();
300 self.session_entry_start = session
301 .pointer("/entryPage/start")
302 .and_then(Value::as_u64)
303 .unwrap_or(0);
304 self.session_entry_total = session
305 .pointer("/entryPage/total")
306 .and_then(Value::as_u64)
307 .unwrap_or(self.session_entries.len() as u64);
308 self.session_events = session
309 .pointer("/eventPage/items")
310 .and_then(Value::as_array)
311 .cloned()
312 .unwrap_or_default();
313 self.session_event_start = session
314 .pointer("/eventPage/start")
315 .and_then(Value::as_u64)
316 .unwrap_or(0);
317 self.session_event_total = session
318 .pointer("/eventPage/total")
319 .and_then(Value::as_u64)
320 .unwrap_or(self.session_events.len() as u64);
321 self.session_capture = session
322 .get("capture")
323 .cloned()
324 .filter(|value| !value.is_null());
325 self.session_replay_checkpoint = session
326 .get("replayCheckpoint")
327 .cloned()
328 .filter(|value| !value.is_null());
329 self.session_events_malformed = session
330 .get("eventsMalformed")
331 .and_then(Value::as_bool)
332 .unwrap_or(false);
333 self.session_events_torn_tail = session
334 .get("eventsTornTail")
335 .and_then(Value::as_bool)
336 .unwrap_or(false);
337 }
338 }
339 true
340 }
341
342 fn apply_delta(&mut self, delta: &ViewerRevisionDelta) -> bool {
343 if delta.revision != self.revision + 1
344 || delta.targets.iter().any(|target| {
345 !target.patch.iter().any(|operation| {
346 !matches!(
347 operation,
348 PatchOp::Replace { path, .. }
349 if path == "/presentationRevision" || path == "/graphRevision"
350 )
351 })
352 })
353 {
354 return false;
355 }
356 for target in &delta.targets {
357 let applied = match (target.target_type.as_str(), target.target_key.as_str()) {
358 ("replay", "steps:reload") | ("timeline", "session:reload") => false,
359 ("summary" | "graph" | "replay" | "inspector", _) => {
360 self.apply_root_patch(&target.patch)
361 }
362 ("conversation", key) if key.starts_with("entries:") => apply_page_patch(
363 &mut self.session_entry_start,
364 &mut self.session_entry_total,
365 &mut self.session_entries,
366 self.revision,
367 &target.patch,
368 ),
369 ("timeline", key) if key.starts_with("trace:") => apply_page_patch(
370 &mut self.trace_start,
371 &mut self.trace_total,
372 &mut self.events,
373 self.revision,
374 &target.patch,
375 ),
376 ("timeline", key) if key.starts_with("session:") => apply_page_patch(
377 &mut self.session_event_start,
378 &mut self.session_event_total,
379 &mut self.session_events,
380 self.revision,
381 &target.patch,
382 ),
383 ("conversation", "capture") => {
384 let mut document = json!({
385 "presentationRevision": self.revision,
386 "capture": self.session_capture,
387 });
388 if apply_patch(&mut document, &target.patch).is_err() {
389 false
390 } else {
391 self.session_capture = document
392 .get("capture")
393 .filter(|value| !value.is_null())
394 .cloned();
395 true
396 }
397 }
398 _ => false,
399 };
400 if !applied {
401 return false;
402 }
403 }
404 self.revision = delta.revision;
405 true
406 }
407
408 pub fn view(&self) -> Value {
409 json!({
410 "presentationRevision": self.revision,
411 "graphRevision": self.graph_revision,
412 "manifest": self.manifest_raw,
413 "workflow": self.workflow,
414 "graphScene": self.graph_layout,
415 "state": self.state_raw,
416 "graphSteps": self.graph_steps,
417 "takenTransitions": self.taken_transitions,
418 "graphCursor": self.graph_cursor,
419 "stepStart": self.step_start,
420 "stepTotal": self.step_total,
421 "tracePage": {
422 "presentationRevision": self.revision,
423 "start": self.trace_start,
424 "total": self.trace_total,
425 "items": self.events,
426 },
427 "session": self.session_value(),
428 "settingsScopes": self.settings_scopes,
429 "settingsStart": self.settings_start,
430 "settingsTotal": self.settings_total,
431 "followUpQueue": self.follow_up_queue,
432 "followUpStart": self.follow_up_start,
433 "followUpTotal": self.follow_up_total,
434 "updateStart": self.update_start,
435 "updateTotal": self.update_total,
436 "live": self.live,
437 "possiblyInterrupted": self.possibly_interrupted,
438 })
439 }
440}
441
442pub struct ProjectionUpdate {
443 pub run_id: String,
444 pub delta: ViewerRevisionDelta,
445}
446
447pub struct RefreshOutcome {
448 pub updates: Vec<ProjectionUpdate>,
449 pub snapshots_required: Vec<String>,
450 pub listing_changed: bool,
451}
452
453fn apply_page_patch(
454 start: &mut u64,
455 total: &mut u64,
456 items: &mut Vec<Value>,
457 revision: u64,
458 patch: &[PatchOp],
459) -> bool {
460 if start.saturating_add(items.len() as u64) != *total {
461 return true;
462 }
463 let mut page = json!({
464 "presentationRevision": revision,
465 "start": *start,
466 "total": *total,
467 "items": items,
468 });
469 if apply_patch(&mut page, patch).is_err() {
470 return false;
471 }
472 let (Some(next_start), Some(next_total), Some(next_items)) = (
473 page.get("start").and_then(Value::as_u64),
474 page.get("total").and_then(Value::as_u64),
475 page.get("items").and_then(Value::as_array),
476 ) else {
477 return false;
478 };
479 *start = next_start;
480 *total = next_total;
481 *items = next_items.clone();
482 true
483}
484
485#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
486pub struct SourceStats {
487 pub data_version_checks: u64,
488 pub index_reads: u64,
489 pub window_reads: u64,
490 pub page_reads: u64,
491 pub payload_rows_read: u64,
492}
493
494pub struct RunSource {
495 database_path: PathBuf,
496 reader: ProjectionReader,
497 loader: SourceLoader,
498 next_generation: u64,
499 pending: BTreeMap<String, u64>,
500 data_version: u64,
501 index: BTreeMap<String, RunIndexRow>,
502 runs: BTreeMap<String, RunEntry>,
503 cursors: BTreeMap<String, WindowCursor>,
504 watched: BTreeMap<String, usize>,
505 local_selected: Option<String>,
506 single_run_id: Option<String>,
507 load_errors: BTreeMap<String, String>,
508 stats: SourceStats,
509}
510
511impl RunSource {
512 pub fn new(database_path: &Path) -> Result<Self> {
513 let reader = ProjectionReader::open(database_path)?;
514 let loader = SourceLoader::new(database_path)?;
515 let data_version = reader.data_version()?;
516 let index = reader
517 .list_run_index()?
518 .into_iter()
519 .map(|row| (row.manifest.run_id.clone(), row))
520 .collect();
521 Ok(Self {
522 database_path: database_path.to_path_buf(),
523 reader,
524 loader,
525 next_generation: 0,
526 pending: BTreeMap::new(),
527 data_version,
528 index,
529 runs: BTreeMap::new(),
530 cursors: BTreeMap::new(),
531 watched: BTreeMap::new(),
532 local_selected: None,
533 single_run_id: None,
534 load_errors: BTreeMap::new(),
535 stats: SourceStats {
536 data_version_checks: 1,
537 index_reads: 1,
538 ..SourceStats::default()
539 },
540 })
541 }
542
543 pub fn single(database_path: &Path, run_id: &str) -> Result<Self> {
544 let mut source = Self::new(database_path)?;
545 if !source.index.contains_key(run_id) {
546 bail!("workflow run not found: {run_id}");
547 }
548 source.single_run_id = Some(run_id.to_string());
549 source.select(run_id)?;
550 Ok(source)
551 }
552
553 pub fn database_path(&self) -> &Path {
554 &self.database_path
555 }
556
557 pub fn get(&self, run_id: &str) -> Option<&RunEntry> {
558 self.runs.get(run_id)
559 }
560
561 pub fn ordered_run_ids(&self) -> Vec<String> {
562 let mut ids: Vec<String> = self.index.keys().cloned().collect();
563 ids.sort_by(|left, right| {
564 let left_row = &self.index[left];
565 let right_row = &self.index[right];
566 right_row
567 .manifest
568 .started_at
569 .cmp(&left_row.manifest.started_at)
570 .then_with(|| right.cmp(left))
571 });
572 ids
573 }
574
575 pub fn summaries(&self) -> Vec<Value> {
576 self.ordered_run_ids()
577 .iter()
578 .filter_map(|id| self.index.get(id))
579 .map(|row| {
580 json!({
581 "presentationRevision": row.presentation_revision,
582 "manifest": row.manifest,
583 "live": row.live,
584 "possiblyInterrupted": row.possibly_interrupted,
585 })
586 })
587 .collect()
588 }
589
590 pub fn select(&mut self, run_id: &str) -> Result<()> {
591 if self.local_selected.as_deref() == Some(run_id) && self.runs.contains_key(run_id) {
592 return Ok(());
593 }
594 if !self.index.contains_key(run_id) {
595 bail!("workflow run not found: {run_id}");
596 }
597 let previous = self.local_selected.replace(run_id.to_string());
598 self.submit_load(run_id);
599 if let Some(previous) = previous {
600 if previous != run_id {
601 self.pending.remove(&previous);
602 if !self.watched.contains_key(&previous) {
603 self.runs.remove(&previous);
604 self.cursors.remove(&previous);
605 self.load_errors.remove(&previous);
606 }
607 }
608 }
609 Ok(())
610 }
611
612 pub fn watch(&mut self, run_id: &str) -> Result<()> {
613 if !self.index.contains_key(run_id) {
614 bail!("workflow run not found: {run_id}");
615 }
616 let count = self.watched.entry(run_id.to_string()).or_default();
617 *count += 1;
618 if *count == 1 && !self.runs.contains_key(run_id) {
619 self.load(run_id)?;
620 }
621 Ok(())
622 }
623
624 pub fn unwatch(&mut self, run_id: &str) {
625 let Some(count) = self.watched.get_mut(run_id) else {
626 return;
627 };
628 *count = count.saturating_sub(1);
629 if *count == 0 {
630 self.watched.remove(run_id);
631 if self.local_selected.as_deref() != Some(run_id) {
632 self.runs.remove(run_id);
633 self.cursors.remove(run_id);
634 self.load_errors.remove(run_id);
635 }
636 }
637 }
638
639 pub fn watcher_count(&self, run_id: &str) -> usize {
640 self.watched.get(run_id).copied().unwrap_or(0)
641 }
642
643 pub fn deltas_after(&self, run_id: &str, revision: u64) -> Result<ViewerDeltaRead> {
644 self.reader.read_deltas(run_id, revision)
645 }
646
647 pub fn page(
648 &mut self,
649 run_id: &str,
650 kind: PageKind,
651 cursor: u64,
652 ) -> Result<(u64, ProjectionPage)> {
653 let (revision, page) = self.reader.read_page(run_id, kind, cursor)?;
654 self.stats.page_reads += 1;
655 self.stats.payload_rows_read += page.items.len() as u64;
656 Ok((revision, page))
657 }
658
659 pub fn stats(&self) -> SourceStats {
660 self.stats
661 }
662
663 pub fn load_error(&self, run_id: &str) -> Option<&str> {
664 self.load_errors.get(run_id).map(String::as_str)
665 }
666
667 pub fn is_stale(&self, run_id: &str) -> bool {
668 self.runs.contains_key(run_id) && self.load_errors.contains_key(run_id)
669 }
670
671 pub fn cursor(&self, run_id: &str) -> WindowCursor {
672 self.cursors.get(run_id).copied().unwrap_or_default()
673 }
674
675 pub fn request_window(&mut self, run_id: &str, cursor: WindowCursor) -> Result<()> {
676 self.cursors.insert(run_id.to_string(), cursor);
677 if self.watched.contains_key(run_id) {
678 self.load(run_id)?;
679 } else if self.local_selected.as_deref() == Some(run_id) {
680 self.submit_load(run_id);
681 }
682 Ok(())
683 }
684
685 pub fn drain(&mut self) {
686 for result in self.loader.drain() {
687 if self.pending.get(&result.run_id).copied() != Some(result.generation) {
688 continue;
689 }
690 self.pending.remove(&result.run_id);
691 match result.loaded {
692 Ok(loaded) => {
693 self.stats.window_reads += 1;
694 self.stats.payload_rows_read += loaded_payload_rows(&loaded);
695 if self.index.get(&result.run_id).is_some_and(|row| {
696 row.presentation_revision == loaded.presentation_revision
697 }) {
698 match RunEntry::from_loaded(&self.database_path, loaded) {
699 Ok(mut entry) => {
700 if let Some(layout) = self
701 .runs
702 .get(&result.run_id)
703 .and_then(|current| current.graph_layout.clone())
704 {
705 entry.graph_layout = Some(layout);
706 }
707 self.load_errors.remove(&result.run_id);
708 self.runs.insert(result.run_id, entry);
709 }
710 Err(_) => {
711 self.load_errors
712 .insert(result.run_id, "run data is unavailable".to_string());
713 }
714 }
715 }
716 }
717 Err(_) => {
718 self.load_errors
719 .insert(result.run_id, "run data is unavailable".to_string());
720 }
721 }
722 }
723 }
724
725 pub fn refresh_all(&mut self) -> RefreshOutcome {
726 self.drain();
727 self.stats.data_version_checks += 1;
728 let Ok(next_data_version) = self.reader.data_version() else {
729 return RefreshOutcome {
730 updates: Vec::new(),
731 snapshots_required: Vec::new(),
732 listing_changed: false,
733 };
734 };
735 let mut listing_changed = self.refresh_interruption_clock();
736 if next_data_version == self.data_version {
737 return RefreshOutcome {
738 updates: Vec::new(),
739 snapshots_required: Vec::new(),
740 listing_changed,
741 };
742 }
743 self.data_version = next_data_version;
744 let Ok(rows) = self.reader.list_run_index() else {
745 return RefreshOutcome {
746 updates: Vec::new(),
747 snapshots_required: Vec::new(),
748 listing_changed,
749 };
750 };
751 self.stats.index_reads += 1;
752 let next_index: BTreeMap<String, RunIndexRow> = rows
753 .into_iter()
754 .map(|row| (row.manifest.run_id.clone(), row))
755 .collect();
756 listing_changed |= index_changed(&self.index, &next_index);
757 self.index = next_index;
758
759 let demanded: BTreeSet<String> = self
760 .watched
761 .keys()
762 .cloned()
763 .chain(self.local_selected.iter().cloned())
764 .collect();
765 let mut updates = Vec::new();
766 let mut snapshots_required = Vec::new();
767 for run_id in demanded {
768 let Some(index) = self.index.get(&run_id) else {
769 self.runs.remove(&run_id);
770 self.load_errors.remove(&run_id);
771 continue;
772 };
773 let target_revision = index.presentation_revision;
774 let previous_revision = self.runs.get(&run_id).map_or(0, |entry| entry.revision);
775 if previous_revision == target_revision {
776 continue;
777 }
778 let mut needs_load = previous_revision == 0;
779 match self.reader.read_deltas(&run_id, previous_revision) {
780 Ok(ViewerDeltaRead::Deltas { deltas, .. }) => {
781 for delta in deltas {
782 if let Some(entry) = self.runs.get_mut(&run_id) {
783 needs_load |= !entry.apply_delta(&delta);
784 }
785 updates.push(ProjectionUpdate {
786 run_id: run_id.clone(),
787 delta,
788 });
789 }
790 }
791 Ok(ViewerDeltaRead::SnapshotRequired { .. }) | Err(_) => {
792 needs_load = true;
793 snapshots_required.push(run_id.clone());
794 }
795 }
796 needs_load |= self
797 .runs
798 .get(&run_id)
799 .is_none_or(|entry| entry.revision != target_revision);
800 if !needs_load {
801 continue;
802 }
803 if self.watched.contains_key(&run_id) {
804 let _ = self.load(&run_id);
805 } else {
806 self.submit_load(&run_id);
807 }
808 }
809 RefreshOutcome {
810 updates,
811 snapshots_required,
812 listing_changed,
813 }
814 }
815
816 fn submit_load(&mut self, run_id: &str) {
817 self.next_generation = self.next_generation.wrapping_add(1);
818 let generation = self.next_generation;
819 self.pending.insert(run_id.to_string(), generation);
820 self.loader.submit(LoadRequest {
821 run_id: run_id.to_string(),
822 cursor: self.cursors.get(run_id).copied().unwrap_or_default(),
823 generation,
824 });
825 }
826
827 fn load(&mut self, run_id: &str) -> Result<()> {
828 let cursor = self.cursors.get(run_id).copied().unwrap_or_default();
829 let loaded = match self.reader.read_window(run_id, cursor) {
830 Ok(loaded) => loaded,
831 Err(error) => {
832 self.load_errors
833 .insert(run_id.to_string(), "run data is unavailable".to_string());
834 return Err(error);
835 }
836 };
837 self.stats.window_reads += 1;
838 self.stats.payload_rows_read += loaded_payload_rows(&loaded);
839 match RunEntry::from_loaded(&self.database_path, loaded) {
840 Ok(mut entry) => {
841 if let Some(layout) = self
842 .runs
843 .get(run_id)
844 .and_then(|current| current.graph_layout.clone())
845 {
846 entry.graph_layout = Some(layout);
847 }
848 self.load_errors.remove(run_id);
849 self.runs.insert(run_id.to_string(), entry);
850 Ok(())
851 }
852 Err(error) => {
853 self.load_errors
854 .insert(run_id.to_string(), "run data is unavailable".to_string());
855 Err(error)
856 }
857 }
858 }
859
860 fn refresh_interruption_clock(&mut self) -> bool {
861 let now = Utc::now().timestamp_millis();
862 let mut changed = false;
863 for row in self.index.values_mut() {
864 let next = row.live
865 && (row.lease_owner_id.is_none()
866 || row
867 .lease_expires_at
868 .is_none_or(|expires_at| expires_at <= now));
869 if next != row.possibly_interrupted {
870 row.possibly_interrupted = next;
871 if let Some(entry) = self.runs.get_mut(&row.manifest.run_id) {
872 entry.possibly_interrupted = next;
873 }
874 changed = true;
875 }
876 }
877 changed
878 }
879}
880
881fn loaded_payload_rows(loaded: &LoadedRun) -> u64 {
882 (loaded.state.steps.len()
883 + loaded.graph_steps.len()
884 + loaded.trace.len()
885 + loaded.session_entries.len()
886 + loaded.session_events.len()
887 + loaded.settings_scopes.len()
888 + loaded.state.updates.as_ref().map_or(0, Vec::len)
889 + usize::from(loaded.follow_up_queue.is_some())) as u64
890}
891
892fn index_changed(
893 before: &BTreeMap<String, RunIndexRow>,
894 after: &BTreeMap<String, RunIndexRow>,
895) -> bool {
896 if before.len() != after.len() || before.keys().ne(after.keys()) {
897 return true;
898 }
899 before.iter().any(|(run_id, prior)| {
900 after.get(run_id).is_none_or(|next| {
901 prior.manifest != next.manifest
902 || prior.live != next.live
903 || prior.possibly_interrupted != next.possibly_interrupted
904 })
905 })
906}
907
908#[allow(dead_code)]
909fn _retain_public_types(
910 _binding: Option<SessionBinding>,
911 _entries: Vec<SessionEntryRecord>,
912 _events: Vec<SessionEventRecord>,
913 _capture: Option<SessionCapture>,
914) {
915}