Line data Source code
1 : // Utils for running sync_safekeepers
2 : use anyhow::Result;
3 : use tracing::info;
4 : use utils::lsn::Lsn;
5 :
6 0 : #[derive(Copy, Clone, Debug)]
7 : pub enum TimelineStatusResponse {
8 : NotFound,
9 : Ok(TimelineStatusOkResponse),
10 : }
11 :
12 162 : #[derive(Copy, Clone, Debug)]
13 : pub struct TimelineStatusOkResponse {
14 : flush_lsn: Lsn,
15 : commit_lsn: Lsn,
16 : }
17 :
18 : /// Get a safekeeper's metadata for our timeline. The id is only used for logging
19 669 : pub async fn ping_safekeeper(
20 669 : id: String,
21 669 : config: tokio_postgres::Config,
22 669 : ) -> Result<TimelineStatusResponse> {
23 : // TODO add retries
24 :
25 : // Connect
26 669 : info!("connecting to {}", id);
27 1350 : let (client, conn) = config.connect(tokio_postgres::NoTls).await?;
28 656 : tokio::spawn(async move {
29 1250 : if let Err(e) = conn.await {
30 0 : eprintln!("connection error: {}", e);
31 625 : }
32 656 : });
33 :
34 : // Query
35 656 : info!("querying {}", id);
36 656 : let result = client.simple_query("TIMELINE_STATUS").await?;
37 :
38 : // Parse result
39 625 : info!("done with {}", id);
40 625 : if let postgres::SimpleQueryMessage::Row(row) = &result[0] {
41 : use std::str::FromStr;
42 173 : let response = TimelineStatusResponse::Ok(TimelineStatusOkResponse {
43 173 : flush_lsn: Lsn::from_str(row.get("flush_lsn").unwrap())?,
44 173 : commit_lsn: Lsn::from_str(row.get("commit_lsn").unwrap())?,
45 : });
46 173 : Ok(response)
47 : } else {
48 : // Timeline doesn't exist
49 452 : Ok(TimelineStatusResponse::NotFound)
50 : }
51 636 : }
52 :
53 : /// Given a quorum of responses, check if safekeepers are synced at some Lsn
54 526 : pub fn check_if_synced(responses: Vec<TimelineStatusResponse>) -> Option<Lsn> {
55 526 : // Check if all responses are ok
56 526 : let ok_responses: Vec<TimelineStatusOkResponse> = responses
57 526 : .iter()
58 599 : .filter_map(|r| match r {
59 162 : TimelineStatusResponse::Ok(ok_response) => Some(ok_response),
60 437 : _ => None,
61 599 : })
62 526 : .cloned()
63 526 : .collect();
64 526 : if ok_responses.len() < responses.len() {
65 401 : info!(
66 401 : "not synced. Only {} out of {} know about this timeline",
67 401 : ok_responses.len(),
68 401 : responses.len()
69 401 : );
70 401 : return None;
71 125 : }
72 125 :
73 125 : // Get the min and the max of everything
74 160 : let commit: Vec<Lsn> = ok_responses.iter().map(|r| r.commit_lsn).collect();
75 160 : let flush: Vec<Lsn> = ok_responses.iter().map(|r| r.flush_lsn).collect();
76 125 : let commit_max = commit.iter().max().unwrap();
77 125 : let commit_min = commit.iter().min().unwrap();
78 125 : let flush_max = flush.iter().max().unwrap();
79 125 : let flush_min = flush.iter().min().unwrap();
80 125 :
81 125 : // Check that all values are equal
82 125 : if commit_min != commit_max {
83 3 : info!("not synced. {:?} {:?}", commit_min, commit_max);
84 3 : return None;
85 122 : }
86 122 : if flush_min != flush_max {
87 0 : info!("not synced. {:?} {:?}", flush_min, flush_max);
88 0 : return None;
89 122 : }
90 122 :
91 122 : // Check that commit == flush
92 122 : if commit_max != flush_max {
93 0 : info!("not synced. {:?} {:?}", commit_max, flush_max);
94 0 : return None;
95 122 : }
96 122 :
97 122 : Some(*commit_max)
98 526 : }
|