systemprompt_database/repository/service/liveness.rs
1//! Replica liveness for [`InstanceRegistry`]: a replica is alive while its
2//! boot-time instance claim is held or its `services` heartbeat is fresher
3//! than [`INSTANCE_DEAD_AFTER_SECS`].
4//!
5//! The claim is the authoritative signal — it vanishes with the claiming
6//! session within [`super::CLAIM_DEAD_PEER_SECS`] of a crash — and the
7//! heartbeat covers a replica that never claimed (a CLI one-shot) or whose
8//! claim was taken on another server. Both are read on the primary: a read
9//! replica sees neither the primary's locks nor its latest heartbeat.
10//!
11//! Copyright (c) systemprompt.io — Business Source License 1.1.
12//! See <https://systemprompt.io> for licensing details.
13
14use async_trait::async_trait;
15use systemprompt_identifiers::InstanceId;
16use systemprompt_traits::{CoordinationResult, InstanceRegistry, RepositoryError};
17
18use super::claim::INSTANCE_CLAIM_CLASS;
19use super::repo::ServiceRepository;
20
21// Why: six missed 15 s heartbeats — long enough that a replica stalled by a
22// slow deploy or a GC pause is not judged dead while it is still serving.
23pub const INSTANCE_DEAD_AFTER_SECS: i64 = 90;
24
25#[async_trait]
26impl InstanceRegistry for ServiceRepository {
27 async fn is_alive(&self, instance: &InstanceId) -> CoordinationResult<bool> {
28 let alive = sqlx::query_scalar!(
29 r#"
30 SELECT (
31 EXISTS (
32 SELECT 1 FROM pg_locks l
33 WHERE l.locktype = 'advisory'
34 AND l.granted
35 AND l.classid = ($1::int)::oid
36 AND l.objid = (hashtext($2))::oid
37 AND l.objsubid = 2
38 )
39 OR EXISTS (
40 SELECT 1 FROM services s
41 WHERE s.instance_id = $2
42 AND s.heartbeat_at
43 >= CURRENT_TIMESTAMP - make_interval(secs => $3::double precision)
44 )
45 ) AS "alive!"
46 "#,
47 INSTANCE_CLAIM_CLASS,
48 instance.as_str(),
49 INSTANCE_DEAD_AFTER_SECS as f64
50 )
51 .fetch_one(&*self.write_pool)
52 .await
53 .map_err(RepositoryError::from)?;
54 Ok(alive)
55 }
56}