Do self review

safekeeper/src/broker.rs

@@ -46,7 +46,7 @@ async fn push_loop(conf: SafeKeeperConf) -> anyhow::Result<()> {
         return Ok(());
     }
 
-    let timelines_set = GlobalTimelines::get_global_broker_active_set();
+    let active_timelines_set = GlobalTimelines::get_global_broker_active_set();
 
     let mut client =
         storage_broker::connect(conf.broker_endpoint.clone(), conf.broker_keepalive_interval)?;
@@ -59,7 +59,7 @@ async fn push_loop(conf: SafeKeeperConf) -> anyhow::Result<()> {
             // sensitive and there is no risk of deadlock as we don't await while
             // lock is held.
             let now = Instant::now();
-            let all_tlis = timelines_set.get_all();
+            let all_tlis = active_timelines_set.get_all();
             let mut n_pushed_tlis = 0;
             for tli in &all_tlis {
                 let sk_info = tli.get_safekeeper_info(&conf).await;

safekeeper/src/metrics.rs

@@ -745,9 +745,9 @@ impl Collector for TimelineCollector {
 
 async fn collect_timeline_metrics() -> Vec<FullTimelineInfo> {
     let mut res = vec![];
-    let timelines = GlobalTimelines::get_global_broker_active_set().get_all();
+    let active_timelines = GlobalTimelines::get_global_broker_active_set().get_all();
 
-    for tli in timelines {
+    for tli in active_timelines {
         if let Some(info) = tli.info_for_metrics().await {
             res.push(info);
         }

safekeeper/src/receive_wal.rs

@@ -113,7 +113,7 @@ impl WalReceivers {
         self.mutex.lock().get_num()
     }
 
-    /// Get number of walreceivers (compute connections).
+    /// Get channel for number of walreceivers.
     pub fn get_num_rx(self: &Arc<WalReceivers>) -> tokio::sync::watch::Receiver<usize> {
         self.num_computes_rx.clone()
     }

safekeeper/src/timelines_global_map.rs

@@ -12,6 +12,7 @@ use once_cell::sync::Lazy;
 use serde::Serialize;
 use std::collections::HashMap;
 use std::str::FromStr;
+use std::sync::atomic::Ordering;
 use std::sync::{Arc, Mutex};
 use tracing::*;
 use utils::id::{TenantId, TenantTimelineId, TimelineId};
@@ -327,6 +328,8 @@ impl GlobalTimelines {
         let tli_res = TIMELINES_STATE.lock().unwrap().get(ttid);
         match tli_res {
             Ok(timeline) => {
+                let was_active = timeline.broker_active.load(Ordering::Relaxed);
+
                 // Take a lock and finish the deletion holding this mutex.
                 let mut shared_state = timeline.write_shared_state().await;
 
@@ -340,7 +343,7 @@ impl GlobalTimelines {
 
                 Ok(TimelineDeleteForceResult {
                     dir_existed,
-                    was_active: true, // TODO: we probably should remove this field
+                    was_active, // TODO: we probably should remove this field
                 })
             }
             Err(_) => {

safekeeper/src/timelines_set.rs

@@ -4,7 +4,12 @@ use utils::id::TenantTimelineId;
 
 use crate::timeline::Timeline;
 
-/// Set of timelines, used to keep subset of timelines used for some tasks (like WAL removal).
+/// Set of timelines, supports operations:
+/// - add timeline
+/// - remove timeline
+/// - clone the set
+///
+/// Used for keeping subset of active (for which broker push is required) timelines.
 pub struct TimelinesSet {
     timelines: std::sync::Mutex<HashMap<TenantTimelineId, Arc<Timeline>>>,
 }
@@ -26,6 +31,7 @@ impl TimelinesSet {
         self.timelines.lock().unwrap().remove(ttid);
     }
 
+    /// If present is true, adds timeline to the set, otherwise removes it.
     pub fn set_present(&self, tli: Arc<Timeline>, present: bool) {
         if present {
             self.insert(tli);
@@ -38,10 +44,12 @@ impl TimelinesSet {
         self.timelines.lock().unwrap().contains_key(ttid)
     }
 
+    /// Returns all timelines in the set.
     pub fn get_all(&self) -> Vec<Arc<Timeline>> {
         self.timelines.lock().unwrap().values().cloned().collect()
     }
 
+    /// Returns a timeline guard for easy presence control.
     pub fn guard(self: &Arc<Self>, tli: Arc<Timeline>) -> TimelineSetGuard {
         let is_present = self.is_present(&tli.ttid);
         TimelineSetGuard {
@@ -52,6 +60,10 @@ impl TimelinesSet {
     }
 }
 
+/// Guard is used to add or remove timeline from the set.
+/// If the timeline present in set, it will be removed from it on drop.
+/// Note: do not use more than one guard for the same timeline, it caches the presence state.
+/// It is designed to be used in the manager task only.
 pub struct TimelineSetGuard {
     timelines_set: Arc<TimelinesSet>,
     tli: Arc<Timeline>,

safekeeper/src/wal_backup.rs

@@ -47,7 +47,7 @@ pub struct WalBackupTaskHandle {
     handle: JoinHandle<()>,
 }
 
-/// Do we have anything to upload to S3, i.e. should we run the backup task?
+/// Do we have anything to upload to S3, i.e. should safekeepers run backup activity?
 pub fn is_wal_backup_required(
     wal_seg_size: usize,
     num_computes: usize,
@@ -68,11 +68,16 @@ pub async fn update_task(
     state: &StateSnapshot,
     entry: &mut Option<WalBackupTaskHandle>,
 ) {
-    let (offloader, election_dbg_str) =
-        determine_offloader(&state.peers, state.backup_lsn, ttid, conf);
-    let elected_me = Some(conf.my_id) == offloader;
-    let should_task_run = need_backup && elected_me;
+    let (should_task_run, election_dbg_str) = if need_backup {
+        let (offloader, election_dbg_str) =
+            determine_offloader(&state.peers, state.backup_lsn, ttid, conf);
+        let elected_me = Some(conf.my_id) == offloader;
+        (elected_me, election_dbg_str)
+    } else {
+        (false, String::new())
+    };
 
+    // start or stop the task
     if should_task_run != (entry.is_some()) {
         if should_task_run {
             info!("elected for backup: {}", election_dbg_str);