Evict WAL files from disk (#8022)

Fixes #6337

Add safekeeper support to switch between `Present` and
`Offloaded(flush_lsn)` states. The offloading is disabled by default,
but can be controlled using new cmdline arguments:

```
      --enable-offload
          Enable automatic switching to offloaded state
      --delete-offloaded-wal
          Delete local WAL files after offloading. When disabled, they will be left on disk
      --control-file-save-interval <CONTROL_FILE_SAVE_INTERVAL>
          Pending updates to control file will be automatically saved after this interval [default: 300s]
```

Manager watches state updates and detects when there are no actvity on
the timeline and actual partial backup upload in remote storage. When
all conditions are met, the state can be switched to offloaded.

In `timeline.rs` there is `StateSK` enum to support switching between
states. When offloaded, code can access only control file structure and
cannot use `SafeKeeper` to accept new WAL.

`FullAccessTimeline` is now renamed to `WalResidentTimeline`. This
struct contains guard to notify manager about active tasks requiring
on-disk WAL access. All guards are issued by the manager, all requests
are sent via channel using `ManagerCtl`. When manager receives request
to issue a guard, it unevicts timeline if it's currently evicted.

Fixed a bug in partial WAL backup, it used `term` instead of
`last_log_term` previously.

After this commit is merged, next step is to roll this change out, as in
issue #6338.

safekeeper/src/bin/safekeeper.rs

@@ -28,8 +28,8 @@ use utils::pid_file;
 
 use metrics::set_build_info_metric;
 use safekeeper::defaults::{
-    DEFAULT_HEARTBEAT_TIMEOUT, DEFAULT_HTTP_LISTEN_ADDR, DEFAULT_MAX_OFFLOADER_LAG_BYTES,
-    DEFAULT_PARTIAL_BACKUP_TIMEOUT, DEFAULT_PG_LISTEN_ADDR,
+    DEFAULT_CONTROL_FILE_SAVE_INTERVAL, DEFAULT_HEARTBEAT_TIMEOUT, DEFAULT_HTTP_LISTEN_ADDR,
+    DEFAULT_MAX_OFFLOADER_LAG_BYTES, DEFAULT_PARTIAL_BACKUP_TIMEOUT, DEFAULT_PG_LISTEN_ADDR,
 };
 use safekeeper::http;
 use safekeeper::wal_service;
@@ -172,6 +172,7 @@ struct Args {
     walsenders_keep_horizon: bool,
     /// Enable partial backup. If disabled, safekeeper will not upload partial
     /// segments to remote storage.
+    /// TODO: now partial backup is always enabled, remove this flag.
     #[arg(long)]
     partial_backup_enabled: bool,
     /// Controls how long backup will wait until uploading the partial segment.
@@ -181,6 +182,15 @@ struct Args {
     /// be used in tests.
     #[arg(long)]
     disable_periodic_broker_push: bool,
+    /// Enable automatic switching to offloaded state.
+    #[arg(long)]
+    enable_offload: bool,
+    /// Delete local WAL files after offloading. When disabled, they will be left on disk.
+    #[arg(long)]
+    delete_offloaded_wal: bool,
+    /// Pending updates to control file will be automatically saved after this interval.
+    #[arg(long, value_parser = humantime::parse_duration, default_value = DEFAULT_CONTROL_FILE_SAVE_INTERVAL)]
+    control_file_save_interval: Duration,
 }
 
 // Like PathBufValueParser, but allows empty string.
@@ -328,9 +338,12 @@ async fn main() -> anyhow::Result<()> {
         sk_auth_token,
         current_thread_runtime: args.current_thread_runtime,
         walsenders_keep_horizon: args.walsenders_keep_horizon,
-        partial_backup_enabled: args.partial_backup_enabled,
+        partial_backup_enabled: true,
         partial_backup_timeout: args.partial_backup_timeout,
         disable_periodic_broker_push: args.disable_periodic_broker_push,
+        enable_offload: args.enable_offload,
+        delete_offloaded_wal: args.delete_offloaded_wal,
+        control_file_save_interval: args.control_file_save_interval,
     };
 
     // initialize sentry if SENTRY_DSN is provided

safekeeper/src/control_file.rs

@@ -72,6 +72,9 @@ impl FileStorage {
         conf: &SafeKeeperConf,
         state: TimelinePersistentState,
     ) -> Result<FileStorage> {
+        // we don't support creating new timelines in offloaded state
+        assert!(matches!(state.eviction_state, EvictionState::Present));
+
         let store = FileStorage {
             timeline_dir,
             no_sync: conf.no_sync,
@@ -103,7 +106,7 @@ impl FileStorage {
     }
 
     /// Load control file from given directory.
-    pub fn load_control_file_from_dir(timeline_dir: &Utf8Path) -> Result<TimelinePersistentState> {
+    fn load_control_file_from_dir(timeline_dir: &Utf8Path) -> Result<TimelinePersistentState> {
         let path = timeline_dir.join(CONTROL_FILE_NAME);
         Self::load_control_file(path)
     }

safekeeper/src/copy_timeline.rs

@@ -15,7 +15,7 @@ use crate::{
     control_file::{FileStorage, Storage},
     pull_timeline::{create_temp_timeline_dir, load_temp_timeline, validate_temp_timeline},
     state::TimelinePersistentState,
-    timeline::{FullAccessTimeline, Timeline, TimelineError},
+    timeline::{Timeline, TimelineError, WalResidentTimeline},
     wal_backup::copy_s3_segments,
     wal_storage::{wal_file_paths, WalReader},
     GlobalTimelines,
@@ -46,7 +46,7 @@ pub async fn handle_request(request: Request) -> Result<()> {
         }
     }
 
-    let source_tli = request.source.full_access_guard().await?;
+    let source_tli = request.source.wal_residence_guard().await?;
 
     let conf = &GlobalTimelines::get_global_config();
     let ttid = request.destination_ttid;
@@ -159,7 +159,7 @@ pub async fn handle_request(request: Request) -> Result<()> {
 }
 
 async fn copy_disk_segments(
-    tli: &FullAccessTimeline,
+    tli: &WalResidentTimeline,
     wal_seg_size: usize,
     start_lsn: Lsn,
     end_lsn: Lsn,
@@ -183,7 +183,7 @@ async fn copy_disk_segments(
         let copy_end = copy_end - segment_start;
 
         let wal_file_path = {
-            let (normal, partial) = wal_file_paths(tli_dir_path, segment, wal_seg_size)?;
+            let (normal, partial) = wal_file_paths(tli_dir_path, segment, wal_seg_size);
 
             if segment == last_segment {
                 partial

safekeeper/src/debug_dump.rs

@@ -28,7 +28,8 @@ use crate::send_wal::WalSenderState;
 use crate::state::TimelineMemState;
 use crate::state::TimelinePersistentState;
 use crate::timeline::get_timeline_dir;
-use crate::timeline::FullAccessTimeline;
+use crate::timeline::WalResidentTimeline;
+use crate::timeline_manager;
 use crate::GlobalTimelines;
 use crate::SafeKeeperConf;
 
@@ -168,6 +169,7 @@ pub struct Memory {
     pub last_removed_segno: XLogSegNo,
     pub epoch_start_lsn: Lsn,
     pub mem_state: TimelineMemState,
+    pub mgr_status: timeline_manager::Status,
 
     // PhysicalStorage state.
     pub write_lsn: Lsn,
@@ -326,7 +328,7 @@ pub struct TimelineDigest {
 }
 
 pub async fn calculate_digest(
-    tli: &FullAccessTimeline,
+    tli: &WalResidentTimeline,
     request: TimelineDigestRequest,
 ) -> Result<TimelineDigest> {
     if request.from_lsn > request.until_lsn {

safekeeper/src/http/routes.rs

@@ -214,10 +214,10 @@ async fn timeline_snapshot_handler(request: Request<Body>) -> Result<Response<Bo
     let tli = GlobalTimelines::get(ttid).map_err(ApiError::from)?;
     // Note: with evicted timelines it should work better then de-evict them and
     // stream; probably start_snapshot would copy partial s3 file to dest path
-    // and stream control file, or return FullAccessTimeline if timeline is not
+    // and stream control file, or return WalResidentTimeline if timeline is not
     // evicted.
     let tli = tli
-        .full_access_guard()
+        .wal_residence_guard()
         .await
         .map_err(ApiError::InternalServerError)?;
 
@@ -283,7 +283,7 @@ async fn timeline_digest_handler(request: Request<Body>) -> Result<Response<Body
 
     let tli = GlobalTimelines::get(ttid).map_err(ApiError::from)?;
     let tli = tli
-        .full_access_guard()
+        .wal_residence_guard()
         .await
         .map_err(ApiError::InternalServerError)?;
 
@@ -306,7 +306,7 @@ async fn timeline_checkpoint_handler(request: Request<Body>) -> Result<Response<
     tli.write_shared_state()
         .await
         .sk
-        .state
+        .state_mut()
         .flush()
         .await
         .map_err(ApiError::InternalServerError)?;

safekeeper/src/json_ctrl.rs

@@ -21,7 +21,7 @@ use crate::safekeeper::{
 };
 use crate::safekeeper::{Term, TermHistory, TermLsn};
 use crate::state::TimelinePersistentState;
-use crate::timeline::FullAccessTimeline;
+use crate::timeline::WalResidentTimeline;
 use crate::GlobalTimelines;
 use postgres_backend::PostgresBackend;
 use postgres_ffi::encode_logical_message;
@@ -102,7 +102,7 @@ pub async fn handle_json_ctrl<IO: AsyncRead + AsyncWrite + Unpin>(
 async fn prepare_safekeeper(
     ttid: TenantTimelineId,
     pg_version: u32,
-) -> anyhow::Result<FullAccessTimeline> {
+) -> anyhow::Result<WalResidentTimeline> {
     let tli = GlobalTimelines::create(
         ttid,
         ServerInfo {
@@ -115,11 +115,11 @@ async fn prepare_safekeeper(
     )
     .await?;
 
-    tli.full_access_guard().await
+    tli.wal_residence_guard().await
 }
 
 async fn send_proposer_elected(
-    tli: &FullAccessTimeline,
+    tli: &WalResidentTimeline,
     term: Term,
     lsn: Lsn,
 ) -> anyhow::Result<()> {
@@ -151,7 +151,7 @@ pub struct InsertedWAL {
 /// Extend local WAL with new LogicalMessage record. To do that,
 /// create AppendRequest with new WAL and pass it to safekeeper.
 pub async fn append_logical_message(
-    tli: &FullAccessTimeline,
+    tli: &WalResidentTimeline,
     msg: &AppendLogicalMessage,
 ) -> anyhow::Result<InsertedWAL> {
     let wal_data = encode_logical_message(&msg.lm_prefix, &msg.lm_message);

safekeeper/src/lib.rs

@@ -28,6 +28,8 @@ pub mod safekeeper;
 pub mod send_wal;
 pub mod state;
 pub mod timeline;
+pub mod timeline_eviction;
+pub mod timeline_guard;
 pub mod timeline_manager;
 pub mod timelines_set;
 pub mod wal_backup;
@@ -49,6 +51,7 @@ pub mod defaults {
     pub const DEFAULT_HEARTBEAT_TIMEOUT: &str = "5000ms";
     pub const DEFAULT_MAX_OFFLOADER_LAG_BYTES: u64 = 128 * (1 << 20);
     pub const DEFAULT_PARTIAL_BACKUP_TIMEOUT: &str = "15m";
+    pub const DEFAULT_CONTROL_FILE_SAVE_INTERVAL: &str = "300s";
 }
 
 #[derive(Debug, Clone)]
@@ -85,6 +88,9 @@ pub struct SafeKeeperConf {
     pub partial_backup_enabled: bool,
     pub partial_backup_timeout: Duration,
     pub disable_periodic_broker_push: bool,
+    pub enable_offload: bool,
+    pub delete_offloaded_wal: bool,
+    pub control_file_save_interval: Duration,
 }
 
 impl SafeKeeperConf {
@@ -124,6 +130,9 @@ impl SafeKeeperConf {
             partial_backup_enabled: false,
             partial_backup_timeout: Duration::from_secs(0),
             disable_periodic_broker_push: false,
+            enable_offload: false,
+            delete_offloaded_wal: false,
+            control_file_save_interval: Duration::from_secs(1),
         }
     }
 }

safekeeper/src/pull_timeline.rs

@@ -32,7 +32,7 @@ use crate::{
         routes::TimelineStatus,
     },
     safekeeper::Term,
-    timeline::{get_tenant_dir, get_timeline_dir, FullAccessTimeline, Timeline, TimelineError},
+    timeline::{get_tenant_dir, get_timeline_dir, Timeline, TimelineError, WalResidentTimeline},
     wal_storage::{self, open_wal_file, Storage},
     GlobalTimelines, SafeKeeperConf,
 };
@@ -46,7 +46,7 @@ use utils::{
 
 /// Stream tar archive of timeline to tx.
 #[instrument(name = "snapshot", skip_all, fields(ttid = %tli.ttid))]
-pub async fn stream_snapshot(tli: FullAccessTimeline, tx: mpsc::Sender<Result<Bytes>>) {
+pub async fn stream_snapshot(tli: WalResidentTimeline, tx: mpsc::Sender<Result<Bytes>>) {
     if let Err(e) = stream_snapshot_guts(tli, tx.clone()).await {
         // Error type/contents don't matter as they won't can't reach the client
         // (hyper likely doesn't do anything with it), but http stream will be
@@ -66,7 +66,7 @@ pub struct SnapshotContext {
     pub flush_lsn: Lsn,
     pub wal_seg_size: usize,
     // used to remove WAL hold off in Drop.
-    pub tli: FullAccessTimeline,
+    pub tli: WalResidentTimeline,
 }
 
 impl Drop for SnapshotContext {
@@ -80,7 +80,7 @@ impl Drop for SnapshotContext {
 }
 
 pub async fn stream_snapshot_guts(
-    tli: FullAccessTimeline,
+    tli: WalResidentTimeline,
     tx: mpsc::Sender<Result<Bytes>>,
 ) -> Result<()> {
     // tokio-tar wants Write implementor, but we have mpsc tx <Result<Bytes>>;
@@ -135,7 +135,7 @@ pub async fn stream_snapshot_guts(
     Ok(())
 }
 
-impl FullAccessTimeline {
+impl WalResidentTimeline {
     /// Start streaming tar archive with timeline:
     /// 1) stream control file under lock;
     /// 2) hold off WAL removal;
@@ -160,6 +160,7 @@ impl FullAccessTimeline {
         ar: &mut tokio_tar::Builder<W>,
     ) -> Result<SnapshotContext> {
         let mut shared_state = self.write_shared_state().await;
+        let wal_seg_size = shared_state.get_wal_seg_size();
 
         let cf_path = self.get_timeline_dir().join(CONTROL_FILE_NAME);
         let mut cf = File::open(cf_path).await?;
@@ -173,19 +174,19 @@ impl FullAccessTimeline {
         // lock and setting `wal_removal_on_hold` later, it guarantees that WAL
         // won't be removed until we're done.
         let from_lsn = min(
-            shared_state.sk.state.remote_consistent_lsn,
-            shared_state.sk.state.backup_lsn,
+            shared_state.sk.state().remote_consistent_lsn,
+            shared_state.sk.state().backup_lsn,
         );
         if from_lsn == Lsn::INVALID {
             // this is possible if snapshot is called before handling first
             // elected message
             bail!("snapshot is called on uninitialized timeline");
         }
-        let from_segno = from_lsn.segment_number(shared_state.get_wal_seg_size());
-        let term = shared_state.sk.get_term();
-        let last_log_term = shared_state.sk.get_last_log_term();
+        let from_segno = from_lsn.segment_number(wal_seg_size);
+        let term = shared_state.sk.state().acceptor_state.term;
+        let last_log_term = shared_state.sk.last_log_term();
         let flush_lsn = shared_state.sk.flush_lsn();
-        let upto_segno = flush_lsn.segment_number(shared_state.get_wal_seg_size());
+        let upto_segno = flush_lsn.segment_number(wal_seg_size);
         // have some limit on max number of segments as a sanity check
         const MAX_ALLOWED_SEGS: u64 = 1000;
         let num_segs = upto_segno - from_segno + 1;
@@ -206,14 +207,18 @@ impl FullAccessTimeline {
         }
         shared_state.wal_removal_on_hold = true;
 
+        // Drop shared_state to release the lock, before calling wal_residence_guard().
+        drop(shared_state);
+
+        let tli_copy = self.wal_residence_guard().await?;
         let bctx = SnapshotContext {
             from_segno,
             upto_segno,
             term,
             last_log_term,
             flush_lsn,
-            wal_seg_size: shared_state.get_wal_seg_size(),
-            tli: self.clone(),
+            wal_seg_size,
+            tli: tli_copy,
         };
 
         Ok(bctx)
@@ -225,8 +230,8 @@ impl FullAccessTimeline {
     /// forget this if snapshotting fails mid the way.
     pub async fn finish_snapshot(&self, bctx: &SnapshotContext) -> Result<()> {
         let shared_state = self.read_shared_state().await;
-        let term = shared_state.sk.get_term();
-        let last_log_term = shared_state.sk.get_last_log_term();
+        let term = shared_state.sk.state().acceptor_state.term;
+        let last_log_term = shared_state.sk.last_log_term();
         // There are some cases to relax this check (e.g. last_log_term might
         // change, but as long as older history is strictly part of new that's
         // fine), but there is no need to do it.