forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
5.7 KiB
1//! Change signals for CI jobs and runs. [`CiLive`] keeps one generation counter per job and one per
2//! run. A write that changes a job or a run bumps its counter, and every subscriber wakes on the
3//! change. The counter carries no data: a subscriber reads the database or the log file again when
4//! it wakes. Nothing polls. A signal with no receivers left is removed, so the maps hold only the
5//! jobs and runs somebody is watching. The hub is in-process only, like [`NotifyHub`].
6//!
7//! [`NotifyHub`]: crate::notify_hub::NotifyHub
8
9use std::collections::HashMap;
10use std::sync::{Mutex, MutexGuard, PoisonError};
11
12use tokio::sync::watch;
13
14/// The change counters of CI jobs and runs, each keyed by its id.
15#[derive(Default)]
16pub struct CiLive {
17 jobs: Signals,
18 runs: Signals,
19}
20
21impl CiLive {
22 /// Subscribes to changes of job `job_id`. The receiver starts as seen, so a bump after this
23 /// call wakes it. Subscribe before the state you read, so a change in between is not lost.
24 pub fn subscribe_job(&self, job_id: &str) -> watch::Receiver<u64> {
25 self.jobs.subscribe(job_id)
26 }
27
28 /// Subscribes to changes of run `run_id`. Same rules as [`CiLive::subscribe_job`].
29 // No caller yet: the run status stream that waits on this is the next change.
30 #[allow(dead_code)]
31 pub fn subscribe_run(&self, run_id: &str) -> watch::Receiver<u64> {
32 self.runs.subscribe(run_id)
33 }
34
35 /// Wakes the subscribers of job `job_id`: its log grew or its state changed. Does nothing when
36 /// nobody is subscribed.
37 pub fn bump_job(&self, job_id: &str) {
38 self.jobs.bump(job_id);
39 }
40
41 /// Wakes the subscribers of run `run_id`: one of its jobs or its own status changed.
42 pub fn bump_run(&self, run_id: &str) {
43 self.runs.bump(run_id);
44 }
45
46 /// Wakes the subscribers of job `job_id` once more and drops its signal. Call it when the job
47 /// has reached a terminal state: no change follows, so the signal is not kept. A subscriber
48 /// that waits on a closed signal wakes once and then sees it closed.
49 pub fn close_job(&self, job_id: &str) {
50 self.jobs.close(job_id);
51 }
52}
53
54/// One keyed set of counters. A key has an entry only while it has a receiver or a bump is due.
55#[derive(Default)]
56struct Signals {
57 counters: Mutex<HashMap<String, watch::Sender<u64>>>,
58}
59
60impl Signals {
61 fn counters(&self) -> MutexGuard<'_, HashMap<String, watch::Sender<u64>>> {
62 self.counters.lock().unwrap_or_else(PoisonError::into_inner)
63 }
64
65 fn subscribe(&self, key: &str) -> watch::Receiver<u64> {
66 self.counters()
67 .entry(key.to_owned())
68 .or_insert_with(|| watch::channel(0).0)
69 .subscribe()
70 }
71
72 /// Counts one change. Without a subscriber there is no entry, so the bump is a no-op. An entry
73 /// whose last receiver is gone is removed here.
74 fn bump(&self, key: &str) {
75 let mut counters = self.counters();
76 let Some(tx) = counters.get(key) else {
77 return;
78 };
79 tx.send_modify(|generation| *generation = generation.wrapping_add(1));
80 if tx.receiver_count() == 0 {
81 counters.remove(key);
82 }
83 }
84
85 /// Counts one last change and removes the entry. The sender is dropped afterwards, so each
86 /// subscriber wakes once more and then finds the signal closed.
87 fn close(&self, key: &str) {
88 let removed = self.counters().remove(key);
89 if let Some(tx) = removed {
90 tx.send_modify(|generation| *generation = generation.wrapping_add(1));
91 }
92 }
93}
94
95#[cfg(test)]
96mod tests {
97 use std::time::Duration;
98
99 use super::*;
100
101 #[tokio::test]
102 async fn a_bump_wakes_only_the_subscribers_of_that_job() {
103 let live = CiLive::default();
104 let mut job = live.subscribe_job("job-1");
105 let other = live.subscribe_job("job-2");
106 let run = live.subscribe_run("run-1");
107
108 live.bump_job("job-1");
109
110 tokio::time::timeout(Duration::from_secs(1), job.changed())
111 .await
112 .expect("woken")
113 .expect("open");
114 assert_eq!(*job.borrow_and_update(), 1);
115 assert!(!other.has_changed().expect("open"));
116 assert!(!run.has_changed().expect("open"));
117 }
118
119 #[tokio::test]
120 async fn a_bump_before_subscribing_is_not_seen_and_a_bump_after_is() {
121 let live = CiLive::default();
122 let early = live.subscribe_run("run-1");
123 live.bump_run("run-1");
124 let mut late = live.subscribe_run("run-1");
125 assert!(early.has_changed().expect("open"));
126 assert!(!late.has_changed().expect("open"));
127 live.bump_run("run-1");
128 assert!(late.has_changed().expect("open"));
129 late.changed().await.expect("open");
130 assert_eq!(*late.borrow_and_update(), 2);
131 }
132
133 #[test]
134 fn a_bump_without_subscribers_keeps_no_entry() {
135 let live = CiLive::default();
136 live.bump_job("job-1");
137 live.bump_run("run-1");
138 assert!(live.jobs.counters().is_empty());
139 assert!(live.runs.counters().is_empty());
140 }
141
142 #[test]
143 fn a_bump_drops_the_entry_once_its_last_receiver_is_gone() {
144 let live = CiLive::default();
145 let rx = live.subscribe_job("job-1");
146 assert_eq!(live.jobs.counters().len(), 1);
147 drop(rx);
148 live.bump_job("job-1");
149 assert!(live.jobs.counters().is_empty());
150 }
151
152 #[tokio::test]
153 async fn close_wakes_the_subscriber_once_and_then_ends_the_signal() {
154 let live = CiLive::default();
155 let mut rx = live.subscribe_job("job-1");
156 live.close_job("job-1");
157 assert!(live.jobs.counters().is_empty());
158
159 rx.changed().await.expect("the last change is delivered");
160 assert_eq!(*rx.borrow_and_update(), 1);
161 assert!(rx.changed().await.is_err());
162 }
163}