systemprompt_cli/commands/cloud/doctor/
distributed.rs

1//! Multi-replica checks: everything a second node must agree with the first on.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6use std::collections::HashMap;
7use std::hash::BuildHasher;
8use std::time::Duration;
9
10use sha2::{Digest, Sha256};
11use systemprompt_database::{PostgresProvider, replica_status};
12use systemprompt_models::Profile;
13
14use super::CheckResult;
15
16const IDENTITY_SECRETS: [&str; 3] = [
17    "oauth_at_rest_pepper",
18    "manifest_signing_secret_seed",
19    "signing_key_pem",
20];
21const REPLICA_LAG_WARN_SECS: f64 = 5.0;
22const READYZ_TIMEOUT: Duration = Duration::from_secs(5);
23
24fn fingerprint(value: &str) -> String {
25    hex::encode(&Sha256::digest(value.as_bytes())[..8])
26}
27
28pub fn check_identity_fingerprints<S: BuildHasher>(
29    secrets: &HashMap<String, String, S>,
30) -> CheckResult {
31    let mut missing = Vec::new();
32    let mut prints = Vec::new();
33    for name in IDENTITY_SECRETS {
34        match secrets.get(name).filter(|v| !v.is_empty()) {
35            Some(value) => prints.push(format!("{name}={}", fingerprint(value))),
36            None => missing.push(name),
37        }
38    }
39    if missing.is_empty() {
40        CheckResult::pass(
41            "identity-fingerprints",
42            format!(
43                "compare across nodes, every value must match: {}",
44                prints.join(" ")
45            ),
46        )
47    } else {
48        CheckResult::fail(
49            "identity-fingerprints",
50            format!(
51                "missing {}; every replica must share one identity — run `systemprompt admin \
52                 identity generate --json` once and distribute the values",
53                missing.join(", ")
54            ),
55        )
56    }
57}
58
59pub fn check_instance_id(profile: &Profile) -> CheckResult {
60    profile
61        .server
62        .instance_id
63        .as_deref()
64        .map(str::trim)
65        .filter(|id| !id.is_empty())
66        .map_or_else(
67            || {
68                // Why: a cloud boot without server.instance_id relies on the
69                // platform exporting HOSTNAME to the process and to every spawned
70                // MCP server. Fly does not, and a wrong guess here is a refused
71                // boot, so off-host the only verifiable state is the explicit id.
72                if profile.target.is_cloud() {
73                    CheckResult::fail(
74                        "instance-id",
75                        "server.instance_id is not set on a cloud profile; the boot refuses \
76                         unless the platform exports HOSTNAME to the gateway and to every \
77                         MCP subprocess — set it",
78                    )
79                } else {
80                    CheckResult::warn(
81                        "instance-id",
82                        "server.instance_id is not set; the replica falls back to HOSTNAME, \
83                         which must be stable across restarts on this platform",
84                    )
85                }
86            },
87            |id| CheckResult::pass("instance-id", format!("server.instance_id = {id}")),
88        )
89}
90
91pub fn check_trusted_proxies(profile: &Profile) -> CheckResult {
92    if profile.server.trusted_proxies.is_empty() {
93        CheckResult::fail(
94            "trusted-proxies",
95            "server.trusted_proxies is empty; every caller behind the balancer would share one \
96             rate-limit bucket and one ban target",
97        )
98    } else {
99        CheckResult::pass(
100            "trusted-proxies",
101            format!(
102                "{} range(s) configured",
103                profile.server.trusted_proxies.len()
104            ),
105        )
106    }
107}
108
109pub async fn check_write_primary<S: BuildHasher + Sync>(
110    secrets: &HashMap<String, String, S>,
111) -> CheckResult {
112    let Some(url) = secrets.get("database_write_url").filter(|v| !v.is_empty()) else {
113        return CheckResult::fail(
114            "write-primary",
115            "database_write_url is not set; with regional replicas the write pool must be \
116             pinned to the primary explicitly",
117        );
118    };
119    match PostgresProvider::new(url).await {
120        Err(err) => CheckResult::warn(
121            "write-primary",
122            format!("could not connect to database_write_url ({err}); fine if run off-host"),
123        ),
124        Ok(provider) => match replica_status(&provider).await {
125            Ok(status) if status.in_recovery => CheckResult::fail(
126                "write-primary",
127                "database_write_url points at a standby; writes, migrations and LISTEN/NOTIFY \
128                 need the primary",
129            ),
130            Ok(_) => CheckResult::pass("write-primary", "database_write_url is a primary"),
131            Err(err) => CheckResult::warn("write-primary", format!("probe failed: {err}")),
132        },
133    }
134}
135
136pub async fn check_replica_lag<S: BuildHasher + Sync>(
137    secrets: &HashMap<String, String, S>,
138) -> CheckResult {
139    let read = secrets.get("database_url").filter(|v| !v.is_empty());
140    let write = secrets.get("database_write_url").filter(|v| !v.is_empty());
141    let Some(read_url) = read else {
142        return CheckResult::fail("replica-lag", "database_url is not set");
143    };
144    if write.is_none_or(|w| w == read_url) {
145        return CheckResult::pass(
146            "replica-lag",
147            "database_url is the primary; no replica reads configured",
148        );
149    }
150    match PostgresProvider::new(read_url).await {
151        Err(err) => CheckResult::warn(
152            "replica-lag",
153            format!("could not connect to database_url ({err}); fine if run off-host"),
154        ),
155        Ok(provider) => match replica_status(&provider).await {
156            Ok(status) if !status.in_recovery => CheckResult::warn(
157                "replica-lag",
158                "database_url differs from database_write_url but is not a standby",
159            ),
160            Ok(status) => match status.replay_lag_secs {
161                Some(lag) if lag > REPLICA_LAG_WARN_SECS => CheckResult::warn(
162                    "replica-lag",
163                    format!(
164                        "standby replay lag is {lag:.1}s (warn above {REPLICA_LAG_WARN_SECS}s)"
165                    ),
166                ),
167                Some(lag) => {
168                    CheckResult::pass("replica-lag", format!("standby replay lag {lag:.1}s"))
169                },
170                None => CheckResult::pass("replica-lag", "standby has replayed nothing yet"),
171            },
172            Err(err) => CheckResult::warn("replica-lag", format!("probe failed: {err}")),
173        },
174    }
175}
176
177pub async fn check_readyz(profile: &Profile) -> CheckResult {
178    let url = format!(
179        "{}/readyz",
180        profile.server.api_internal_url.trim_end_matches('/')
181    );
182    let client = match reqwest::Client::builder().timeout(READYZ_TIMEOUT).build() {
183        Ok(client) => client,
184        Err(err) => return CheckResult::warn("readyz", format!("http client: {err}")),
185    };
186    match client.get(&url).send().await {
187        Ok(response) if response.status().is_success() => {
188            CheckResult::pass("readyz", format!("{url} answered {}", response.status()))
189        },
190        Ok(response) => {
191            let status = response.status();
192            let body = response.text().await.unwrap_or_default();
193            CheckResult::warn("readyz", format!("{url} answered {status}: {body}"))
194        },
195        Err(err) => CheckResult::warn(
196            "readyz",
197            format!("{url} unreachable ({err}); fine if run off-host"),
198        ),
199    }
200}
201
202pub async fn run<S: BuildHasher + Sync>(
203    profile: &Profile,
204    secrets: &HashMap<String, String, S>,
205) -> Vec<CheckResult> {
206    vec![
207        check_identity_fingerprints(secrets),
208        check_instance_id(profile),
209        check_trusted_proxies(profile),
210        check_write_primary(secrets).await,
211        check_replica_lag(secrets).await,
212        check_readyz(profile).await,
213    ]
214}
systemprompt_cli/commands/cloud/doctor/distributed.rs

systemprompt_cli/commands/cloud/doctor/
distributed.rs
