kasl/db/
server_pulse.rs

1//! The last pulse this machine sent to kasl-server, and how it went.
2//!
3//! The watcher sends the pulse and `kasl server status` reports on it, and
4//! the two are different processes: this row is how one tells the other. It
5//! holds the latest attempt and nothing behind it - the server keeps only the
6//! latest claim, and a local history of when someone was at their desk would
7//! be the very record the pulse was designed not to leave (ADR 0014 in
8//! kasl-server).
9//!
10//! ```rust,no_run
11//! # fn main() -> anyhow::Result<()> {
12//! use chrono::{Duration, Utc};
13//! use kasl::api::kasl_server::AgentState;
14//! use kasl::db::server_pulse::ServerPulse;
15//!
16//! let mut pulse = ServerPulse::new()?;
17//! let now = Utc::now();
18//! pulse.record_accepted(now, now + Duration::seconds(60), AgentState::Working, 180, 0)?;
19//!
20//! if let Some(last) = pulse.last()? {
21//!     println!("last attempt at {}", last.attempted_at);
22//! }
23//! # Ok(())
24//! # }
25//! ```
26
27use crate::api::kasl_server::AgentState;
28use crate::db::db::Db;
29use anyhow::Result;
30use chrono::{DateTime, Utc};
31use rusqlite::{Connection, OptionalExtension};
32
33/// An accepted pulse replaces everything, the error included.
34const RECORD_ACCEPTED: &str = "INSERT INTO server_pulse
35        (id, attempted_at, next_at, sent_at, state, stale_after_seconds, clock_skew_seconds, error)
36     VALUES (1, ?1, ?2, ?1, ?3, ?4, ?5, NULL)
37     ON CONFLICT(id) DO UPDATE SET
38        attempted_at = excluded.attempted_at,
39        next_at = excluded.next_at,
40        sent_at = excluded.sent_at,
41        state = excluded.state,
42        stale_after_seconds = excluded.stale_after_seconds,
43        clock_skew_seconds = excluded.clock_skew_seconds,
44        error = NULL";
45
46/// A failed attempt leaves the last accepted pulse in place: "the last one
47/// that arrived" is what someone reading a failure needs next.
48const RECORD_FAILED: &str = "INSERT INTO server_pulse (id, attempted_at, next_at, error)
49     VALUES (1, ?1, ?2, ?3)
50     ON CONFLICT(id) DO UPDATE SET
51        attempted_at = excluded.attempted_at,
52        next_at = excluded.next_at,
53        error = excluded.error";
54
55const SELECT_LAST: &str = "SELECT attempted_at, next_at, sent_at, state, stale_after_seconds, clock_skew_seconds, error FROM server_pulse WHERE id = 1";
56
57const DELETE_ALL: &str = "DELETE FROM server_pulse";
58
59/// The latest attempt, as the watcher recorded it.
60#[derive(Debug, Clone, PartialEq)]
61pub struct PulseRecord {
62    /// When the watcher last tried, whatever came of it.
63    pub attempted_at: DateTime<Utc>,
64
65    /// When it means to try next. A watcher that is gone stops moving this,
66    /// which is how a reader tells "quiet because nothing changed" from
67    /// "quiet because nobody is sending".
68    pub next_at: DateTime<Utc>,
69
70    /// When a pulse last arrived; `None` until one has.
71    pub sent_at: Option<DateTime<Utc>>,
72
73    /// The state that pulse claimed.
74    pub state: Option<AgentState>,
75
76    /// How long the server believes a pulse, from its last answer.
77    pub stale_after_seconds: Option<i64>,
78
79    /// This machine's clock against the server's, from its last answer.
80    pub clock_skew_seconds: Option<i64>,
81
82    /// Why the last attempt did not arrive; `None` when it did.
83    pub error: Option<String>,
84}
85
86/// Access to the pulse row.
87pub struct ServerPulse {
88    pub conn: Connection,
89}
90
91impl ServerPulse {
92    /// Opens the database; the table comes from the migrations.
93    pub fn new() -> Result<Self> {
94        Ok(ServerPulse { conn: Db::new()?.conn })
95    }
96
97    /// Records a pulse the server accepted.
98    pub fn record_accepted(
99        &mut self,
100        at: DateTime<Utc>,
101        next_at: DateTime<Utc>,
102        state: AgentState,
103        stale_after_seconds: i64,
104        clock_skew_seconds: i64,
105    ) -> Result<()> {
106        self.conn.execute(
107            RECORD_ACCEPTED,
108            rusqlite::params![at, next_at, state.as_str(), stale_after_seconds, clock_skew_seconds],
109        )?;
110        Ok(())
111    }
112
113    /// Records an attempt that did not arrive, and why.
114    pub fn record_failed(&mut self, at: DateTime<Utc>, next_at: DateTime<Utc>, error: &str) -> Result<()> {
115        self.conn.execute(RECORD_FAILED, rusqlite::params![at, next_at, error])?;
116        Ok(())
117    }
118
119    /// The latest attempt, or `None` when there has been none since the pulse
120    /// was last turned on.
121    pub fn last(&self) -> Result<Option<PulseRecord>> {
122        let record = self
123            .conn
124            .query_row(SELECT_LAST, [], |row| {
125                let state: Option<String> = row.get(3)?;
126                Ok(PulseRecord {
127                    attempted_at: row.get(0)?,
128                    next_at: row.get(1)?,
129                    sent_at: row.get(2)?,
130                    state: state.as_deref().and_then(AgentState::parse),
131                    stale_after_seconds: row.get(4)?,
132                    clock_skew_seconds: row.get(5)?,
133                    error: row.get(6)?,
134                })
135            })
136            .optional()?;
137        Ok(record)
138    }
139
140    /// Forgets the record, so what is reported next belongs to a pulse sent
141    /// after this moment rather than to one from before it was turned off.
142    pub fn clear(&mut self) -> Result<()> {
143        self.conn.execute(DELETE_ALL, [])?;
144        Ok(())
145    }
146}
kasl/db/server_pulse.rs

kasl/db/
server_pulse.rs
