CH-J Toolsserver management over SSH
Menu
Published source

CH-J Tools

Browse directories and files for a specific application release.

Download source ZIP
CH-J / Server Manager 1704 files
0.1.2-beta Added: · Built:
availability.rs Added: Date not recorded 27.1 KB · 644 lines
Code Download file
1//! Central availability monitor for saved servers: one Core task, native TCP connects, no login.
2//! `Scheduler` holds the state and the timing (no I/O, tested with explicit instants);
3//! `AvailabilityMonitor` runs it. The state stays in memory and is wiped when the vault locks.
4//! A result says "the SSH port answers from this network"; it never proves a server is off.
6use futures::future::BoxFuture;
7use parking_lot::Mutex;
8use serde_json::{json, Value};
9use std::io::ErrorKind;
10use std::sync::atomic::{AtomicBool, Ordering};
11use std::sync::Arc;
12use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
13use tokio::net::TcpStream;
14use tokio::sync::Notify;
15use tokio::task::JoinSet;
16use tokio_util::sync::CancellationToken;
18pub const DEFAULT_INTERVAL_SECS: u64 = 12;
19pub const INTERVALS_SECS: [u64; 6] = [5, 10, 12, 15, 30, 60];
20pub const TCP_TIMEOUT: Duration = Duration::from_millis(1500);
21pub const JUMP_TIMEOUT: Duration = Duration::from_secs(3);
22pub const RETRY_DELAY: Duration = Duration::from_secs(2);
23pub const FAILURES_FOR_OFFLINE: u32 = 2;
24pub const MAX_CONCURRENT: usize = 6;
25const DNS_TIMEOUT: Duration = Duration::from_secs(3);
26const MAX_ADDRESSES: usize = 3;
27/// Spacing of an immediate round; later rounds spread over the whole interval.
28const BURST_SPACING: Duration = Duration::from_millis(120);
29const HEARTBEAT: Duration = Duration::from_secs(1);
30/// Backstop for profile changes that do not call `changed()`.
31const RESYNC_EVERY: Duration = Duration::from_secs(30);
32/// A jump between wall clock and monotonic clock this large means the computer slept.
33const SLEEP_GAP: Duration = Duration::from_secs(5);
35#[derive(Clone, Debug, PartialEq, Eq)]
36pub struct Target {
37 pub id: String,
38 pub host: String,
39 pub port: u16,
40 pub jump_id: Option<String>,
43impl Target {
44 /// Only the routing fields of a profile; the monitor never touches credentials.
45 pub fn from_profile(profile: &Value) -> Option<Self> {
46 Some(Self {
47 id: profile["id"].as_str().filter(|id| !id.is_empty())?.to_string(),
48 host: profile["host"].as_str().filter(|host| !host.is_empty())?.to_string(),
49 port: u16::try_from(profile["port"].as_u64()?).ok()?,
50 jump_id: profile["jumpProfileId"].as_str().filter(|id| !id.is_empty()).map(str::to_string),
51 })
52 }
55#[derive(Clone, Debug, PartialEq)]
56pub enum Outcome {
57 Online { ms: f64 },
58 Failed { reason: &'static str },
59 /// The route could not be tested; says nothing about the server.
60 Unverified { reason: &'static str },
63struct Entry {
64 target: Target,
65 generation: u64,
66 status: &'static str,
67 latency_ms: Option<f64>,
68 last_success: Option<u64>,
69 last_attempt: Option<u64>,
70 failures: u32,
71 reason: Option<&'static str>,
72 next_due: Option<Instant>,
73 in_flight: bool,
74 /// Set by an immediate round: the next regular check takes this entry's phase in the interval.
75 spread: bool,
78impl Entry {
79 fn value(&self) -> Value {
80 json!({
81 "id": self.target.id,
82 "status": self.status,
83 "latencyMs": self.latency_ms,
84 "lastSuccessAt": self.last_success,
85 "lastAttemptAt": self.last_attempt,
86 "failures": self.failures,
87 "reason": self.reason,
88 "via": if self.target.jump_id.is_some() { "jump" } else { "direct" }
89 })
90 }
93pub struct Scheduler {
94 entries: Vec<Entry>,
95 enabled: bool,
96 interval: Duration,
97 next_generation: u64,
100impl Default for Scheduler {
101 fn default() -> Self {
102 Self { entries: Vec::new(), enabled: true, interval: Duration::from_secs(DEFAULT_INTERVAL_SECS), next_generation: 1 }
103 }
106impl Scheduler {
107 fn generation(&mut self) -> u64 {
108 self.next_generation += 1;
109 self.next_generation
110 }
112 /// Applies the saved profiles. Returns what the UI must hear about: changed entries and removed IDs.
113 pub fn sync(&mut self, targets: Vec<Target>, now: Instant) -> (Vec<Value>, Vec<String>) {
114 let removed: Vec<String> = self.entries.iter().filter(|e| !targets.iter().any(|t| t.id == e.target.id)).map(|e| e.target.id.clone()).collect();
115 self.entries.retain(|e| targets.iter().any(|t| t.id == e.target.id));
116 let mut changed = Vec::new();
117 let mut delay = Duration::ZERO;
118 for target in targets {
119 let position = self.entries.iter().position(|e| e.target.id == target.id);
120 if position.is_some_and(|i| self.entries[i].target == target) {
121 continue;
122 }
123 let generation = self.generation();
124 let enabled = self.enabled;
125 let fresh = Entry {
126 target,
127 generation,
128 status: if enabled { "checking" } else { "unknown" },
129 latency_ms: None,
130 last_success: None,
131 last_attempt: None,
132 failures: 0,
133 reason: None,
134 next_due: enabled.then_some(now + delay),
135 in_flight: false,
136 spread: true,
137 };
138 delay += BURST_SPACING;
139 changed.push(fresh.value());
140 match position {
141 Some(i) => self.entries[i] = fresh,
142 None => self.entries.push(fresh),
143 }
144 }
145 (changed, removed)
146 }
148 /// Returns true when the state visible to the UI changed.
149 pub fn set_policy(&mut self, enabled: bool, interval: Duration, now: Instant) -> bool {
150 let was_enabled = self.enabled;
151 self.enabled = enabled;
152 self.interval = interval;
153 if enabled && !was_enabled {
154 self.refresh_all(now);
155 for entry in &mut self.entries {
156 entry.status = "checking";
157 }
158 return !self.entries.is_empty();
159 }
160 if !enabled && was_enabled {
161 for i in 0..self.entries.len() {
162 let generation = self.generation();
163 let entry = &mut self.entries[i];
164 entry.generation = generation;
165 entry.status = "unknown";
166 entry.latency_ms = None;
167 entry.failures = 0;
168 entry.reason = None;
169 entry.next_due = None;
170 entry.in_flight = false;
171 }
172 return !self.entries.is_empty();
173 }
174 // A shorter interval must not wait for the old, longer one.
175 for entry in &mut self.entries {
176 if let Some(due) = entry.next_due {
177 entry.next_due = Some(due.min(now + interval));
178 }
179 }
180 false
181 }
183 /// Immediate round for every server, spaced so that they do not all start at one moment.
184 pub fn refresh_all(&mut self, now: Instant) {
185 if !self.enabled {
186 return;
187 }
188 for (i, entry) in self.entries.iter_mut().enumerate() {
189 entry.next_due = Some(now + BURST_SPACING * i as u32);
190 entry.spread = true;
191 }
192 }
194 /// Takes up to `slots` due checks and marks them as running.
195 pub fn take_due(&mut self, now: Instant, slots: usize) -> Vec<(Target, u64)> {
196 let mut due = Vec::new();
197 for entry in &mut self.entries {
198 if due.len() == slots {
199 break;
200 }
201 if !entry.in_flight && entry.next_due.is_some_and(|at| at <= now) {
202 entry.in_flight = true;
203 entry.next_due = None;
204 due.push((entry.target.clone(), entry.generation));
205 }
206 }
207 due
208 }
210 /// Records a finished check. A result for an older address or a removed server is dropped.
211 pub fn apply(&mut self, id: &str, generation: u64, outcome: Outcome, now: Instant, wall_ms: u64) -> Option<Value> {
212 let enabled = self.enabled;
213 let (interval, count) = (self.interval, self.entries.len().max(1) as u32);
214 let index = self.entries.iter().position(|e| e.target.id == id && e.generation == generation)?;
215 let entry = &mut self.entries[index];
216 entry.in_flight = false;
217 entry.last_attempt = Some(wall_ms);
218 let mut retry = false;
219 match outcome {
220 Outcome::Online { ms } => {
221 entry.status = "online";
222 entry.latency_ms = Some(ms);
223 entry.last_success = Some(wall_ms);
224 entry.failures = 0;
225 entry.reason = None;
226 }
227 Outcome::Failed { reason } => {
228 entry.failures += 1;
229 entry.reason = Some(reason);
230 if entry.failures >= FAILURES_FOR_OFFLINE {
231 entry.status = "offline";
232 entry.latency_ms = None;
233 } else {
234 retry = true;
235 }
236 }
237 Outcome::Unverified { reason } => {
238 entry.status = "unverified";
239 entry.latency_ms = None;
240 entry.failures = 0;
241 entry.reason = Some(reason);
242 }
243 }
244 entry.next_due = enabled.then(|| {
245 if retry {
246 return now + RETRY_DELAY;
247 }
248 let phase = if entry.spread { interval * index as u32 / count } else { Duration::ZERO };
249 entry.spread = false;
250 now + interval + phase
251 });
252 Some(entry.value())
253 }
255 pub fn next_due(&self) -> Option<Instant> {
256 self.entries.iter().filter(|e| !e.in_flight).filter_map(|e| e.next_due).min()
257 }
259 pub fn snapshot(&self) -> Vec<Value> {
260 self.entries.iter().map(Entry::value).collect()
261 }
263 pub fn clear(&mut self) {
264 self.entries.clear();
265 }
268fn failure_reason(error: &std::io::Error) -> &'static str {
269 match error.kind() {
270 ErrorKind::ConnectionRefused => "refused",
271 ErrorKind::TimedOut => "timeout",
272 ErrorKind::HostUnreachable | ErrorKind::NetworkUnreachable | ErrorKind::AddrNotAvailable => "unreachable",
273 ErrorKind::PermissionDenied => "denied",
274 _ => "error",
275 }
278/// TCP connect only: the socket closes right after the handshake, before any SSH traffic.
279pub async fn probe_tcp(host: &str, port: u16, timeout: Duration) -> Outcome {
280 let addresses: Vec<_> = match tokio::time::timeout(DNS_TIMEOUT, tokio::net::lookup_host((host, port))).await {
281 Ok(Ok(found)) => found.take(MAX_ADDRESSES).collect(),
282 _ => return Outcome::Failed { reason: "dns" },
283 };
284 let mut reason = "dns";
285 for address in addresses {
286 let start = Instant::now();
287 match tokio::time::timeout(timeout, TcpStream::connect(address)).await {
288 Ok(Ok(_stream)) => return Outcome::Online { ms: (start.elapsed().as_secs_f64() * 10_000.0).round() / 10.0 },
289 Ok(Err(error)) => reason = failure_reason(&error),
290 Err(_) => reason = "timeout",
291 }
292 }
293 Outcome::Failed { reason }
296/// What the monitor needs from the rest of the Core.
297pub trait MonitorHost: Send + Sync + 'static {
298 /// False while the vault is locked or no main window exists.
299 fn available(&self) -> bool;
300 fn settings(&self) -> (bool, Duration);
301 fn targets(&self) -> Option<Vec<Target>>;
302 fn publish(&self, event: Value);
303 /// Checks a target through an SSH session that is already open to its jump host.
304 fn probe_jump(&self, jump_id: &str, target: &Target) -> BoxFuture<'static, Outcome>;
307#[derive(Default)]
308pub struct AvailabilityMonitor {
309 state: Mutex<Scheduler>,
310 wake: Notify,
311 resync: AtomicBool,
312 refresh: AtomicBool,
313 stop: CancellationToken,
316fn wall_ms() -> u64 {
317 SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0)
320impl AvailabilityMonitor {
321 pub fn snapshot(&self) -> Vec<Value> {
322 self.state.lock().snapshot()
323 }
325 /// Profiles, settings or the vault state changed.
326 pub fn changed(&self) {
327 self.resync.store(true, Ordering::SeqCst);
328 self.wake.notify_one();
329 }
331 pub fn refresh(&self) {
332 self.refresh.store(true, Ordering::SeqCst);
333 self.wake.notify_one();
334 }
336 /// Ends the task; dropping its JoinSet aborts running checks and closes their sockets.
337 pub fn stop(&self) {
338 self.stop.cancel();
339 }
341 pub async fn run(self: Arc<Self>, host: Arc<dyn MonitorHost>) {
342 let mut checks: JoinSet<(String, u64, Outcome)> = JoinSet::new();
343 let mut active = false;
344 let mut last_sync = Instant::now();
345 let mut clocks = (Instant::now(), SystemTime::now());
346 loop {
347 let now = Instant::now();
348 let slept = {
349 let (mono, wall) = (now.duration_since(clocks.0), clocks.1.elapsed().unwrap_or_default());
350 clocks = (now, SystemTime::now());
351 wall > mono + SLEEP_GAP || mono > HEARTBEAT + SLEEP_GAP
352 };
353 if !host.available() {
354 if active {
355 active = false;
356 checks.abort_all();
357 self.state.lock().clear();
358 host.publish(json!({ "reset": true, "servers": [] }));
359 }
360 } else {
361 let first = !active;
362 active = true;
363 let (enabled, interval) = host.settings();
364 if self.state.lock().set_policy(enabled, interval, now) {
365 host.publish(json!({ "reset": true, "servers": self.snapshot() }));
366 }
367 if first || self.resync.swap(false, Ordering::SeqCst) || now.duration_since(last_sync) >= RESYNC_EVERY {
368 last_sync = now;
369 if let Some(targets) = host.targets() {
370 let (changed, removed) = self.state.lock().sync(targets, now);
371 if !changed.is_empty() || !removed.is_empty() {
372 host.publish(json!({ "servers": changed, "removed": removed }));
373 }
374 }
375 }
376 if self.refresh.swap(false, Ordering::SeqCst) || slept {
377 self.state.lock().refresh_all(now);
378 }
379 let slots = MAX_CONCURRENT.saturating_sub(checks.len());
380 for (target, generation) in self.state.lock().take_due(now, slots) {
381 let id = target.id.clone();
382 let via_jump = target.jump_id.clone().map(|jump_id| host.probe_jump(&jump_id, &target));
383 checks.spawn(async move {
384 let outcome = match via_jump {
385 Some(probe) => probe.await,
386 None => probe_tcp(&target.host, target.port, TCP_TIMEOUT).await,
387 };
388 (id, generation, outcome)
389 });
390 }
391 }
392 let wait = self.state.lock().next_due().map_or(HEARTBEAT, |at| at.saturating_duration_since(Instant::now()).clamp(Duration::from_millis(5), HEARTBEAT));
393 tokio::select! {
394 _ = self.stop.cancelled() => break,
395 _ = self.wake.notified() => {}
396 _ = tokio::time::sleep(wait) => {}
397 Some(done) = checks.join_next(), if !checks.is_empty() => {
398 if let Ok((id, generation, outcome)) = done {
399 let updated = self.state.lock().apply(&id, generation, outcome, Instant::now(), wall_ms());
400 if let Some(updated) = updated {
401 host.publish(json!({ "servers": [updated] }));
402 }
403 }
404 }
405 }
406 }
407 }
410#[cfg(test)]
411mod tests {
412 use super::*;
414 fn target(id: &str, port: u16) -> Target {
415 Target { id: id.into(), host: "server.test".into(), port, jump_id: None }
416 }
417 const FAIL: Outcome = Outcome::Failed { reason: "timeout" };
419 fn one(now: Instant) -> Scheduler {
420 let mut scheduler = Scheduler::default();
421 scheduler.sync(vec![target("a", 22)], now);
422 scheduler
423 }
424 fn run(scheduler: &mut Scheduler, now: Instant, outcome: Outcome) -> Value {
425 let (due, generation) = scheduler.take_due(now, 6).remove(0);
426 scheduler.apply(&due.id, generation, outcome, now, 1000).unwrap()
427 }
429 #[test]
430 fn new_server_is_checked_at_once() {
431 let now = Instant::now();
432 let mut scheduler = one(now);
433 assert_eq!(scheduler.snapshot()[0]["status"], "checking");
434 assert_eq!(scheduler.take_due(now, 6).len(), 1);
435 assert!(scheduler.take_due(now, 6).is_empty(), "a running check is not started twice");
436 }
438 #[test]
439 fn online_offline_online_needs_two_failures_two_seconds_apart() {
440 let now = Instant::now();
441 let mut scheduler = one(now);
442 assert_eq!(run(&mut scheduler, now, Outcome::Online { ms: 2.5 })["status"], "online");
443 let at = now + Duration::from_secs(12);
444 let first = run(&mut scheduler, at, FAIL);
445 assert_eq!((first["status"].as_str(), first["failures"].as_u64()), (Some("online"), Some(1)));
446 assert_eq!(scheduler.next_due(), Some(at + RETRY_DELAY));
447 let second = run(&mut scheduler, at + RETRY_DELAY, FAIL);
448 assert_eq!(second["status"], "offline");
449 assert_eq!(second["latencyMs"], Value::Null);
450 assert_eq!(second["reason"], "timeout");
451 assert_eq!(second["lastSuccessAt"], 1000);
452 assert_eq!(scheduler.next_due(), Some(at + RETRY_DELAY + Duration::from_secs(12)), "offline waits for the normal interval");
453 let back = run(&mut scheduler, at + Duration::from_secs(40), Outcome::Online { ms: 3.0 });
454 assert_eq!((back["status"].as_str(), back["failures"].as_u64(), back["reason"].is_null()), (Some("online"), Some(0), true));
455 }
457 #[test]
458 fn first_ever_failure_stays_checking_until_the_retry() {
459 let now = Instant::now();
460 let mut scheduler = one(now);
461 assert_eq!(run(&mut scheduler, now, FAIL)["status"], "checking");
462 assert_eq!(run(&mut scheduler, now + RETRY_DELAY, FAIL)["status"], "offline");
463 }
465 #[test]
466 fn concurrency_is_limited_and_rounds_spread_over_the_interval() {
467 let now = Instant::now();
468 let mut scheduler = Scheduler::default();
469 scheduler.sync((0..12).map(|i| target(&format!("s{i}"), 22)).collect(), now);
470 let late = now + Duration::from_secs(5);
471 let first = scheduler.take_due(late, MAX_CONCURRENT);
472 assert_eq!(first.len(), MAX_CONCURRENT);
473 for (t, generation) in first {
474 scheduler.apply(&t.id, generation, Outcome::Online { ms: 1.0 }, late, 1).unwrap();
475 }
476 let second = scheduler.take_due(late, MAX_CONCURRENT);
477 assert_eq!(second.len(), MAX_CONCURRENT);
478 for (t, generation) in second {
479 scheduler.apply(&t.id, generation, Outcome::Online { ms: 1.0 }, late, 1).unwrap();
480 }
481 let mut times: Vec<Instant> = scheduler.entries.iter().map(|e| e.next_due.unwrap()).collect();
482 times.dedup();
483 assert_eq!(times.len(), 12, "no two servers share a start");
484 assert_eq!(times[1] - times[0], Duration::from_secs(1), "12 servers in 12 s: one per second");
485 }
487 #[test]
488 fn immediate_round_is_spaced() {
489 let now = Instant::now();
490 let mut scheduler = Scheduler::default();
491 scheduler.sync(vec![target("a", 22), target("b", 22)], now);
492 assert_eq!(scheduler.take_due(now, 6).len(), 1);
493 assert_eq!(scheduler.take_due(now + BURST_SPACING, 6).len(), 1);
494 }
496 #[test]
497 fn changed_address_restarts_and_drops_the_old_result() {
498 let now = Instant::now();
499 let mut scheduler = one(now);
500 let (_, old) = scheduler.take_due(now, 6).remove(0);
501 let (changed, removed) = scheduler.sync(vec![target("a", 2222)], now);
502 assert_eq!((changed.len(), removed.len()), (1, 0));
503 assert_eq!(changed[0]["status"], "checking");
504 assert!(scheduler.apply("a", old, Outcome::Online { ms: 1.0 }, now, 1).is_none(), "a result from the old port must not count");
505 let (due, _) = scheduler.take_due(now, 6).remove(0);
506 assert_eq!(due.port, 2222);
507 assert_eq!(scheduler.sync(vec![target("a", 2222)], now).0.len(), 0, "unchanged data does not restart");
508 }
510 #[test]
511 fn removed_server_leaves_no_state() {
512 let now = Instant::now();
513 let mut scheduler = one(now);
514 let (_, generation) = scheduler.take_due(now, 6).remove(0);
515 let (changed, removed) = scheduler.sync(vec![], now);
516 assert!(changed.is_empty());
517 assert_eq!(removed, vec!["a".to_string()]);
518 assert!(scheduler.snapshot().is_empty());
519 assert!(scheduler.apply("a", generation, FAIL, now, 1).is_none());
520 assert_eq!(scheduler.next_due(), None);
521 }
523 #[test]
524 fn disabling_stops_checks_and_marks_unknown() {
525 let now = Instant::now();
526 let mut scheduler = one(now);
527 let (_, generation) = scheduler.take_due(now, 6).remove(0);
528 assert!(scheduler.set_policy(false, Duration::from_secs(12), now));
529 assert_eq!(scheduler.snapshot()[0]["status"], "unknown");
530 assert!(scheduler.apply("a", generation, Outcome::Online { ms: 1.0 }, now, 1).is_none());
531 assert_eq!(scheduler.next_due(), None);
532 scheduler.refresh_all(now);
533 assert!(scheduler.take_due(now, 6).is_empty(), "no checks while disabled");
534 assert!(scheduler.set_policy(true, Duration::from_secs(12), now));
535 assert_eq!(scheduler.snapshot()[0]["status"], "checking");
536 assert_eq!(scheduler.take_due(now, 6).len(), 1);
537 }
539 #[test]
540 fn shorter_interval_takes_effect_at_once() {
541 let now = Instant::now();
542 let mut scheduler = one(now);
543 scheduler.set_policy(true, Duration::from_secs(60), now);
544 run(&mut scheduler, now, Outcome::Online { ms: 1.0 });
545 scheduler.set_policy(true, Duration::from_secs(5), now);
546 assert!(scheduler.next_due().unwrap() <= now + Duration::from_secs(5));
547 }
549 #[test]
550 fn unverified_is_never_online_and_hides_old_latency() {
551 let now = Instant::now();
552 let mut scheduler = one(now);
553 run(&mut scheduler, now, Outcome::Online { ms: 4.0 });
554 let value = run(&mut scheduler, now + Duration::from_secs(12), Outcome::Unverified { reason: "noJumpSession" });
555 assert_eq!(value["status"], "unverified");
556 assert_eq!(value["latencyMs"], Value::Null);
557 assert_eq!(value["reason"], "noJumpSession");
558 }
560 #[test]
561 fn profile_fields_become_targets_without_secrets() {
562 let parsed = Target::from_profile(&json!({ "id": "x", "host": "2001:db8::1", "port": 2222, "jumpProfileId": "j", "password": "not-copied" })).unwrap();
563 assert_eq!(parsed, Target { id: "x".into(), host: "2001:db8::1".into(), port: 2222, jump_id: Some("j".into()) });
564 assert!(Target::from_profile(&json!({ "id": "x", "port": 22 })).is_none());
565 }
567 #[tokio::test]
568 async fn tcp_probe_measures_a_listener_and_reports_refusal() {
569 for bind in ["127.0.0.1", "::1"] {
570 let Ok(listener) = tokio::net::TcpListener::bind((bind, 0)).await else { continue };
571 let port = listener.local_addr().unwrap().port();
572 assert!(matches!(probe_tcp(bind, port, TCP_TIMEOUT).await, Outcome::Online { .. }), "{bind}");
573 // Port 1: a freed ephemeral port could be taken by a test running in parallel.
574 // Windows retries a refused connection for about 2 s, so there it ends as a timeout.
575 let closed = probe_tcp(bind, 1, TCP_TIMEOUT).await;
576 if cfg!(windows) { assert!(matches!(closed, Outcome::Failed { .. }), "{bind}: {closed:?}"); } else { assert_eq!(closed, Outcome::Failed { reason: "refused" }, "{bind}"); }
577 }
578 let listener = tokio::net::TcpListener::bind(("localhost", 0)).await.unwrap();
579 let port = listener.local_addr().unwrap().port();
580 assert!(matches!(probe_tcp("localhost", port, TCP_TIMEOUT).await, Outcome::Online { .. }), "DNS name, non-standard port");
581 assert_eq!(probe_tcp("no-such-host.invalid", 22, TCP_TIMEOUT).await, Outcome::Failed { reason: "dns" });
582 }
584 struct TestHost {
585 unlocked: AtomicBool,
586 targets: Mutex<Vec<Target>>,
587 events: Mutex<Vec<Value>>,
588 probes: Mutex<u32>,
589 }
590 impl MonitorHost for TestHost {
591 fn available(&self) -> bool { self.unlocked.load(Ordering::SeqCst) }
592 fn settings(&self) -> (bool, Duration) { (true, Duration::from_millis(300)) }
593 fn targets(&self) -> Option<Vec<Target>> { Some(self.targets.lock().clone()) }
594 fn publish(&self, event: Value) { self.events.lock().push(event); }
595 fn probe_jump(&self, _jump_id: &str, _target: &Target) -> BoxFuture<'static, Outcome> {
596 *self.probes.lock() += 1;
597 Box::pin(async { Outcome::Unverified { reason: "noJumpSession" } })
598 }
599 }
601 #[tokio::test]
602 async fn monitor_runs_all_servers_locks_clean_and_stops() {
603 let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)).await.unwrap();
604 let open = listener.local_addr().unwrap().port();
605 let host = Arc::new(TestHost {
606 unlocked: AtomicBool::new(true),
607 targets: Mutex::new(vec![
608 Target { id: "up".into(), host: "127.0.0.1".into(), port: open, jump_id: None },
609 Target { id: "down".into(), host: "127.0.0.1".into(), port: 1, jump_id: None },
610 Target { id: "behind".into(), host: "10.255.255.1".into(), port: 22, jump_id: Some("up".into()) },
611 ]),
612 events: Mutex::new(Vec::new()),
613 probes: Mutex::new(0),
614 });
615 let monitor = Arc::new(AvailabilityMonitor::default());
616 let task = tokio::spawn(monitor.clone().run(host.clone()));
617 let status = |id: &str| monitor.snapshot().iter().find(|e| e["id"] == id).map(|e| e["status"].as_str().unwrap().to_string());
618 tokio::time::sleep(Duration::from_millis(600)).await;
619 assert_eq!(status("up").as_deref(), Some("online"));
620 assert_eq!(status("behind").as_deref(), Some("unverified"), "a jump-only server gets no direct result");
621 tokio::time::timeout(Duration::from_secs(5), async { while status("down").as_deref() != Some("offline") { tokio::time::sleep(Duration::from_millis(50)).await; } }).await.unwrap();
623 // Removal and a new port are picked up at once.
624 host.targets.lock().retain(|t| t.id != "behind");
625 host.targets.lock()[1].port = open;
626 monitor.changed();
627 tokio::time::timeout(Duration::from_secs(2), async { while status("down").as_deref() != Some("online") { tokio::time::sleep(Duration::from_millis(20)).await; } }).await.unwrap();
628 assert_eq!(status("behind"), None);
630 // Lock wipes everything; unlock checks at once.
631 host.unlocked.store(false, Ordering::SeqCst);
632 monitor.changed();
633 tokio::time::sleep(Duration::from_millis(100)).await;
634 assert!(monitor.snapshot().is_empty());
635 assert_eq!(host.events.lock().last().unwrap()["reset"], true);
636 host.unlocked.store(true, Ordering::SeqCst);
637 monitor.changed();
638 tokio::time::timeout(Duration::from_secs(2), async { while status("up").as_deref() != Some("online") { tokio::time::sleep(Duration::from_millis(20)).await; } }).await.unwrap();
640 monitor.stop();
641 tokio::time::timeout(Duration::from_secs(1), task).await.expect("task ends on stop").unwrap();
642 assert!(*host.probes.lock() >= 1);
643 }
SHA-25610c7ddefd6d9d56b7d0125ba6bc430e290ed7199bf1b0da77d5c646d73038076

CH-J Proprietary Software License 1.14 · Source is provided under the CH-J Proprietary Software License 1.14. Its availability does not change the license terms or grant additional rights.

Archive SHA-256ac8fe790f4ba60fc098f8c985fc2656cfcc6cdc21fb34b30e308a2b54e62258f
Files