WIP keep only mem state

pageserver/src/bin/pageserver.rs

@@ -185,9 +185,6 @@ fn start_pageserver(conf: &'static PageServerConf, daemonize: bool) -> Result<()
     // Initialize logger
     let log_file = logging::init(LOG_FILE_NAME, daemonize)?;
 
-    // TODO init only if configured
-    pageserver::wal_metadata::init(conf).expect("wal_metadata init failed");
-
     info!("version: {}", GIT_VERSION);
 
     // TODO: Check that it looks like a valid repository before going further

pageserver/src/bin/psbench.rs

@@ -1,158 +0,0 @@
-//! Pageserver benchmark tool
-//!
-//! Usually it's easier to write python perf tests, but here the performance
-//! of the tester matters, and the API is easier to work with from rust.
-use std::{collections::HashSet, io::{BufRead, BufReader, Cursor}, time::Duration};
-use pageserver::wal_metadata::{Page, WalEntryMetadata};
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
-use bytes::{BufMut, BytesMut};
-use clap::{App, Arg};
-use std::fs::File;
-use zenith_utils::{GIT_VERSION, lsn::Lsn, pq_proto::{BeMessage, FeMessage}};
-use std::time::Instant;
-
-use anyhow::Result;
-
-const BYTES_IN_PAGE: usize = 8 * 1024;
-
-pub fn read_lines_buffered(file_name: &str) -> impl Iterator<Item = String> {
-    BufReader::new(File::open(file_name).unwrap())
-        .lines()
-        .map(|result| result.unwrap())
-}
-
-pub async fn get_page(
-    pagestream: &mut tokio::net::TcpStream,
-    lsn: &Lsn,
-    page: &Page,
-    latest: bool,
-) -> anyhow::Result<Vec<u8>> {
-    let latest: u8 = if latest {1} else {0};
-    let msg = {
-        let query = {
-            let mut query = BytesMut::new();
-            query.put_u8(2);  // Specifies get_page query
-            query.put_u8(latest);
-            query.put_u64(lsn.0);
-            page.write(&mut query).await?;
-            query.freeze()
-        };
-
-        let mut buf = BytesMut::new();
-        let copy_msg = BeMessage::CopyData(&query);
-        BeMessage::write(&mut buf, &copy_msg)?;
-        buf.freeze()
-    };
-
-    pagestream.write(&msg).await?;
-
-    let response = match FeMessage::read_fut(pagestream).await? {
-        Some(FeMessage::CopyData(page)) => page,
-        r => panic!("Expected CopyData message, got: {:?}", r),
-    };
-
-    let page = {
-        let mut cursor = Cursor::new(response);
-        let tag = AsyncReadExt::read_u8(&mut cursor).await?;
-
-        match tag {
-            102 => {
-                let mut page = Vec::<u8>::new();
-                cursor.read_to_end(&mut page).await?;
-                if page.len() != BYTES_IN_PAGE {
-                    panic!("Expected 8kb page, got: {:?}", page.len());
-                }
-                page
-            },
-            103 => {
-                let mut bytes = Vec::<u8>::new();
-                cursor.read_to_end(&mut bytes).await?;
-                let message = String::from_utf8(bytes)?;
-                panic!("Got error message: {}", message);
-            },
-            _ => panic!("Unhandled tag {:?}", tag)
-        }
-    };
-
-    Ok(page)
-}
-
-#[tokio::main]
-async fn main() -> Result<()> {
-    let arg_matches = App::new("LALALA")
-        .about("lalala")
-        .version(GIT_VERSION)
-        .arg(
-            Arg::new("wal_metadata_file")
-                .help("Path to wal metadata file")
-                .required(true)
-                .index(1),
-        )
-        .arg(
-            Arg::new("tenant_hex")
-                .help("TODO")
-                .required(true)
-                .index(2),
-        )
-        .arg(
-            Arg::new("timeline")
-                .help("TODO")
-                .required(true)
-                .index(3),
-        )
-        .get_matches();
-
-    let metadata_file = arg_matches.value_of("wal_metadata_file").unwrap();
-    let tenant_hex = arg_matches.value_of("tenant_hex").unwrap();
-    let timeline = arg_matches.value_of("timeline").unwrap();
-
-    // Parse log lines
-    let wal_metadata: Vec<WalEntryMetadata> = read_lines_buffered(metadata_file)
-        .map(|line| serde_json::from_str(&line).expect("corrupt metadata file"))
-        .collect();
-
-    // Get raw TCP connection to the pageserver postgres protocol port
-    let mut socket = tokio::net::TcpStream::connect("localhost:15000").await?;
-    let (client, conn) = tokio_postgres::Config::new()
-        .host("127.0.0.1")
-        .port(15000)
-        .dbname("postgres")
-        .user("zenith_admin")
-        .connect_raw(&mut socket, tokio_postgres::NoTls)
-        .await?;
-
-    // Enter pagestream protocol
-    let init_query = format!("pagestream {} {}", tenant_hex, timeline);
-    tokio::select! {
-        _ = conn => panic!("AAAA"),
-        _ = client.query(init_query.as_str(), &[]) => (),
-    };
-
-    // Derive some variables
-    let total_wal_size: usize = wal_metadata.iter().map(|m| m.size).sum();
-    let affected_pages: HashSet<_> = wal_metadata.iter().map(|m| m.affected_pages.clone())
-        .flatten().collect();
-    let latest_lsn = wal_metadata.iter().map(|m| m.lsn).max().unwrap();
-
-    // Get all latest pages
-    let mut durations: Vec<Duration> = vec![];
-    for page in &affected_pages {
-        let start = Instant::now();
-        let _page_bytes = get_page(&mut socket, &latest_lsn, &page, true).await?;
-        let duration = start.elapsed();
-
-        durations.push(duration);
-    }
-
-    durations.sort();
-    // Results are a space separated table of "metric_name value unit", for ease of parsing
-    println!("test_param num_pages {}", affected_pages.len());
-    println!("test_param num_wal_entries {}", wal_metadata.len());
-    println!("test_param total_wal_size {} bytes", total_wal_size);
-    println!("lower_is_better fastest {:?} microseconds", durations.first().unwrap().as_micros());
-    println!("lower_is_better median {:?} microseconds", durations[durations.len() / 2].as_micros());
-    println!("lower_is_better p99 {:?} microseconds", durations[durations.len() - 1 - durations.len() / 100].as_micros());
-    println!("lower_is_better slowest {:?} microseconds", durations.last().unwrap().as_micros());
-
-    Ok(())
-}

pageserver/src/lib.rs

@@ -19,7 +19,6 @@ pub mod walingest;
 pub mod walreceiver;
 pub mod walrecord;
 pub mod walredo;
-pub mod wal_metadata;
 
 use lazy_static::lazy_static;
 use tracing::info;

pageserver/src/page_cache.rs

@@ -52,7 +52,6 @@ use zenith_utils::{
     zid::{ZTenantId, ZTimelineId},
 };
 
-use crate::config::PageServerConf;
 use crate::layered_repository::writeback_ephemeral_file;
 use crate::repository::Key;
 

pageserver/src/wal_metadata.rs

@@ -1,78 +0,0 @@
-use anyhow::Result;
-use once_cell::sync::OnceCell;
-use zenith_utils::lsn::Lsn;
-use std::fs::File;
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
-use bytes::{BufMut, Bytes, BytesMut};
-use serde::{Deserialize, Serialize};
-
-use crate::{config::PageServerConf, repository::Key, walrecord::DecodedBkpBlock};
-
-pub static WAL_METADATA_FILE: OnceCell<File> = OnceCell::new();
-
-#[derive(Copy, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Debug, Serialize, Deserialize)]
-pub struct Page {
-    spcnode: u32,
-    dbnode: u32,
-    relnode: u32,
-    forknum: u8,
-    blkno: u32,
-}
-
-impl Page {
-    pub async fn read<Reader>(buf: &mut Reader) -> Result<Page>
-    where
-        Reader: tokio::io::AsyncRead + Unpin,
-    {
-        let spcnode = buf.read_u32().await?;
-        let dbnode = buf.read_u32().await?;
-        let relnode = buf.read_u32().await?;
-        let forknum = buf.read_u8().await?;
-        let blkno = buf.read_u32().await?;
-        Ok(Page { spcnode, dbnode, relnode, forknum, blkno })
-    }
-
-    pub async fn write(&self, buf: &mut BytesMut) -> Result<()> {
-        buf.put_u32(self.spcnode);
-        buf.put_u32(self.dbnode);
-        buf.put_u32(self.relnode);
-        buf.put_u8(self.forknum);
-        buf.put_u32(self.blkno);
-        Ok(())
-    }
-}
-
-impl From<&DecodedBkpBlock> for Page {
-    fn from(blk: &DecodedBkpBlock) -> Self {
-        Page {
-            spcnode: blk.rnode_spcnode,
-            dbnode: blk.rnode_dbnode,
-            relnode: blk.rnode_relnode,
-            forknum: blk.forknum,
-            blkno: blk.blkno,
-        }
-    }
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub struct WalEntryMetadata {
-    pub lsn: Lsn,
-    pub size: usize,
-    pub affected_pages: Vec<Page>,
-}
-
-pub fn init(conf: &'static PageServerConf) -> Result<()> {
-    let wal_metadata_file_dir = conf.workdir.join("wal_metadata.log");
-    WAL_METADATA_FILE.set(File::create(wal_metadata_file_dir)?)
-        .expect("wal_metadata file is already created");
-    Ok(())
-}
-
-pub fn write(wal_meta: WalEntryMetadata) -> Result<()> {
-    if let Some(mut file) = WAL_METADATA_FILE.get() {
-        let mut line = serde_json::to_string(&wal_meta)?;
-        line.push('\n');
-        std::io::prelude::Write::write_all(&mut file, line.as_bytes())?;
-    }
-    Ok(())
-}

pageserver/src/walingest.rs

@@ -82,7 +82,6 @@ impl<'a, R: Repository> WalIngest<'a, R> {
     ) -> Result<()> {
         let mut modification = timeline.begin_modification(lsn);
 
-        let recdata_len = recdata.len();
         let mut decoded = decode_wal_record(recdata);
         let mut buf = decoded.record.clone();
         buf.advance(decoded.main_data_offset);
@@ -250,13 +249,6 @@ impl<'a, R: Repository> WalIngest<'a, R> {
             self.ingest_decoded_block(&mut modification, lsn, &decoded, blk)?;
         }
 
-        // Emit wal entry metadata, if configured to do so
-        crate::wal_metadata::write(crate::wal_metadata::WalEntryMetadata {
-            lsn,
-            size: recdata_len,
-            affected_pages: decoded.blocks.iter().map(|blk| blk.into()).collect()
-        });
-
         // If checkpoint data was updated, store the new version in the repository
         if self.checkpoint_modified {
             let new_checkpoint_bytes = self.checkpoint.encode();

test_runner/fixtures/zenith_fixtures.py

@@ -638,13 +638,6 @@ class ZenithEnv:
         """ Get list of safekeeper endpoints suitable for wal_acceptors GUC  """
         return ','.join([f'localhost:{wa.port.pg}' for wa in self.safekeepers])
 
-    def run_psbench(self, timeline):
-        wal_metadata_filename = os.path.join(self.repo_dir, "wal_metadata.log")
-        psbench_binpath = os.path.join(str(zenith_binpath), 'psbench')
-        tenant_hex = self.initial_tenant.hex
-        args = [psbench_binpath, wal_metadata_filename, tenant_hex, timeline]
-        return subprocess.run(args, capture_output=True).stdout.decode("UTF-8").strip()
-
     @cached_property
     def auth_keys(self) -> AuthKeys:
         pub = (Path(self.repo_dir) / 'auth_public_key.pem').read_bytes()

test_runner/performance/test_pageserver.py

@@ -1,48 +0,0 @@
-from contextlib import closing
-from fixtures.zenith_fixtures import ZenithEnv, PgBin
-from fixtures.benchmark_fixture import MetricReport, ZenithBenchmarker
-
-
-def test_get_page(zenith_simple_env: ZenithEnv,
-                  zenbenchmark: ZenithBenchmarker,
-                  pg_bin: PgBin):
-    env = zenith_simple_env
-    env.zenith_cli.create_branch("test_pageserver", "empty")
-    pg = env.postgres.create_start('test_pageserver')
-    tenant_hex = env.initial_tenant.hex
-    timeline = pg.safe_psql("SHOW zenith.zenith_timeline")[0][0]
-
-    # Long-lived cursor, useful for flushing
-    psconn = env.pageserver.connect()
-    pscur = psconn.cursor()
-
-    with closing(pg.connect()) as conn:
-        with conn.cursor() as cur:
-            workload = "pgbench"
-
-            print(f"Running workload {workload}")
-            if workload == "hot page":
-                cur.execute('create table t (i integer);')
-                cur.execute('insert into t values (0);')
-                for i in range(100000):
-                    cur.execute(f'update t set i = {i};')
-            elif workload == "pgbench":
-                pg_bin.run_capture(['pgbench', '-s5', '-i', pg.connstr()])
-                pg_bin.run_capture(['pgbench', '-c1', '-t5000', pg.connstr()])
-            elif workload == "pgbench big":
-                pg_bin.run_capture(['pgbench', '-s100', '-i', pg.connstr()])
-                pg_bin.run_capture(['pgbench', '-c1', '-t100000', pg.connstr()])
-            elif workload == "pgbench long":
-                pg_bin.run_capture(['pgbench', '-s100', '-i', pg.connstr()])
-                pg_bin.run_capture(['pgbench', '-c1', '-t1000000', pg.connstr()])
-
-            pscur.execute(f"checkpoint {env.initial_tenant.hex} {timeline} 0")
-
-    output = env.run_psbench(timeline)
-    for line in output.split("\n"):
-        tokens = line.split(" ")
-        report = tokens[0]
-        name = tokens[1]
-        value = tokens[2]
-        unit = tokens[3] if len(tokens) > 3 else ""
-        zenbenchmark.record(name, value, unit, report=report)

walkeeper/src/control_file.rs

@@ -6,6 +6,7 @@ use lazy_static::lazy_static;
 
 use std::fs::{self, File, OpenOptions};
 use std::io::{Read, Write};
+use std::ops::Deref;
 use std::path::{Path, PathBuf};
 
 use tracing::*;
@@ -37,8 +38,10 @@ lazy_static! {
     .expect("Failed to register safekeeper_persist_control_file_seconds histogram vec");
 }
 
-pub trait Storage {
-    /// Persist safekeeper state on disk.
+/// Storage should keep actual state inside of it. It should implement Deref
+/// trait to access state fields and have persist method for updating that state.
+pub trait Storage: Deref<Target = SafeKeeperState> {
+    /// Persist safekeeper state on disk and update internal state.
     fn persist(&mut self, s: &SafeKeeperState) -> Result<()>;
 }
 
@@ -48,19 +51,47 @@ pub struct FileStorage {
     timeline_dir: PathBuf,
     conf: SafeKeeperConf,
     persist_control_file_seconds: Histogram,
+
+    /// Last state persisted to disk.
+    state: SafeKeeperState,
 }
 
 impl FileStorage {
-    pub fn new(zttid: &ZTenantTimelineId, conf: &SafeKeeperConf) -> FileStorage {
+    pub fn restore_new(zttid: &ZTenantTimelineId, conf: &SafeKeeperConf) -> Result<FileStorage> {
         let timeline_dir = conf.timeline_dir(zttid);
         let tenant_id = zttid.tenant_id.to_string();
         let timeline_id = zttid.timeline_id.to_string();
-        FileStorage {
+
+        let state = Self::load_control_file_conf(conf, zttid)?;
+
+        Ok(FileStorage {
             timeline_dir,
             conf: conf.clone(),
             persist_control_file_seconds: PERSIST_CONTROL_FILE_SECONDS
                 .with_label_values(&[&tenant_id, &timeline_id]),
-        }
+            state,
+        })
+    }
+
+    pub fn create_new(
+        zttid: &ZTenantTimelineId,
+        conf: &SafeKeeperConf,
+        state: SafeKeeperState,
+    ) -> Result<FileStorage> {
+        let timeline_dir = conf.timeline_dir(zttid);
+        let tenant_id = zttid.tenant_id.to_string();
+        let timeline_id = zttid.timeline_id.to_string();
+
+        let mut store = FileStorage {
+            timeline_dir,
+            conf: conf.clone(),
+            persist_control_file_seconds: PERSIST_CONTROL_FILE_SECONDS
+                .with_label_values(&[&tenant_id, &timeline_id]),
+            state: state.clone(),
+        };
+
+        store.persist(&state)?;
+        Ok(store)
     }
 
     // Check the magic/version in the on-disk data and deserialize it, if possible.
@@ -141,6 +172,14 @@ impl FileStorage {
     }
 }
 
+impl Deref for FileStorage {
+    type Target = SafeKeeperState;
+
+    fn deref(&self) -> &Self::Target {
+        &self.state
+    }
+}
+
 impl Storage for FileStorage {
     // persists state durably to underlying storage
     // for description see https://lwn.net/Articles/457667/
@@ -201,6 +240,9 @@ impl Storage for FileStorage {
                 .and_then(|f| f.sync_all())
                 .context("failed to sync control file directory")?;
         }
+
+        // update internal state
+        self.state = s.clone();
         Ok(())
     }
 }
@@ -228,7 +270,7 @@ mod test {
     ) -> Result<(FileStorage, SafeKeeperState)> {
         fs::create_dir_all(&conf.timeline_dir(zttid)).expect("failed to create timeline dir");
         Ok((
-            FileStorage::new(zttid, conf),
+            FileStorage::restore_new(zttid, conf)?,
             FileStorage::load_control_file_conf(conf, zttid)?,
         ))
     }
@@ -239,8 +281,7 @@ mod test {
     ) -> Result<(FileStorage, SafeKeeperState)> {
         fs::create_dir_all(&conf.timeline_dir(zttid)).expect("failed to create timeline dir");
         let state = SafeKeeperState::empty();
-        let mut storage = FileStorage::new(zttid, conf);
-        storage.persist(&state)?;
+        let storage = FileStorage::create_new(zttid, conf, state.clone())?;
         Ok((storage, state))
     }
 

walkeeper/src/safekeeper.rs

@@ -195,21 +195,31 @@ pub struct SafeKeeperState {
     /// pushed to s3. We don't remove WAL beyond it. Persisted only for
     /// informational purposes, we receive it from pageserver (or broker).
     pub remote_consistent_lsn: Lsn,
-    // Peers and their state as we remember it. Knowing peers themselves is
-    // fundamental; but state is saved here only for informational purposes and
-    // obviously can be stale. (Currently not saved at all, but let's provision
-    // place to have less file version upgrades).
+    /// Peers and their state as we remember it. Knowing peers themselves is
+    /// fundamental; but state is saved here only for informational purposes and
+    /// obviously can be stale. (Currently not saved at all, but let's provision
+    /// place to have less file version upgrades).
     pub peers: Peers,
 }
 
 #[derive(Debug, Clone)]
-// In memory safekeeper state. Fields mirror ones in `SafeKeeperState`; values
-// are not flushed yet.
+/// In memory safekeeper state. Some fields mirror ones in `SafeKeeperState`;
+/// values are not flushed yet.
 pub struct SafekeeperMemState {
+    // LSNs waiting to be persisted to disk;
     pub commit_lsn: Lsn,
     pub s3_wal_lsn: Lsn, // TODO: keep only persistent version
     pub peer_horizon_lsn: Lsn,
     pub remote_consistent_lsn: Lsn,
+
+    /// Maximum commit_lsn between all nodes, can be ahead of local flush_lsn.
+    pub global_commit_lsn: Lsn,
+
+    /// UUID of a elected proposer.
+    proposer_uuid: PgUuid,
+    /// LSN since the proposer safekeeper currently talking to appends WAL;
+    /// determines epoch switch point.
+    epoch_start_lsn: Lsn,
 }
 
 impl SafeKeeperState {
@@ -492,19 +502,15 @@ impl SafeKeeperMetrics {
 /// SafeKeeper which consumes events (messages from compute) and provides
 /// replies.
 pub struct SafeKeeper<CTRL: control_file::Storage, WAL: wal_storage::Storage> {
-    // Cached metrics so we don't have to recompute labels on each update.
+    /// Cached metrics so we don't have to recompute labels on each update.
     metrics: SafeKeeperMetrics,
-
-    /// Maximum commit_lsn between all nodes, can be ahead of local flush_lsn.
-    pub global_commit_lsn: Lsn,
-    /// LSN since the proposer safekeeper currently talking to appends WAL;
-    /// determines epoch switch point.
-    epoch_start_lsn: Lsn,
-
-    pub inmem: SafekeeperMemState, // in memory part
-    pub s: SafeKeeperState,        // persistent part
-
-    pub control_store: CTRL,
+    /// In-memory safekeeper state. Some fields repeat ones from
+    /// persistent SafeKeeperState and waiting to be persisted
+    /// to disk.
+    pub inmem: SafekeeperMemState,
+    /// SafeKeeperState persisted to disk and methods for update.
+    pub state: CTRL,
+    /// WAL storage with write functions.
     pub wal_store: WAL,
 }
 
@@ -516,42 +522,43 @@ where
     // constructor
     pub fn new(
         ztli: ZTimelineId,
-        control_store: CTRL,
+        state: CTRL,
         mut wal_store: WAL,
-        state: SafeKeeperState,
     ) -> Result<SafeKeeper<CTRL, WAL>> {
         if state.timeline_id != ZTimelineId::from([0u8; 16]) && ztli != state.timeline_id {
             bail!("Calling SafeKeeper::new with inconsistent ztli ({}) and SafeKeeperState.server.timeline_id ({})", ztli, state.timeline_id);
         }
 
+        // initialize wal_store, if state is already initialized
         wal_store.init_storage(&state)?;
 
         Ok(SafeKeeper {
             metrics: SafeKeeperMetrics::new(state.tenant_id, ztli),
             global_commit_lsn: state.commit_lsn,
-            epoch_start_lsn: Lsn(0),
+            elected: ElectedProposerInfo::empty(),
             inmem: SafekeeperMemState {
                 commit_lsn: state.commit_lsn,
                 s3_wal_lsn: state.s3_wal_lsn,
                 peer_horizon_lsn: state.peer_horizon_lsn,
                 remote_consistent_lsn: state.remote_consistent_lsn,
             },
-            s: state,
-            control_store,
+            state,
             wal_store,
         })
     }
 
     /// Get history of term switches for the available WAL
     fn get_term_history(&self) -> TermHistory {
-        self.s
+        self.state
             .acceptor_state
             .term_history
             .up_to(self.wal_store.flush_lsn())
     }
 
     pub fn get_epoch(&self) -> Term {
-        self.s.acceptor_state.get_epoch(self.wal_store.flush_lsn())
+        self.state
+            .acceptor_state
+            .get_epoch(self.wal_store.flush_lsn())
     }
 
     /// Process message from proposer and possibly form reply. Concurrent
@@ -587,46 +594,47 @@ where
             );
         }
         /* Postgres upgrade is not treated as fatal error */
-        if msg.pg_version != self.s.server.pg_version
-            && self.s.server.pg_version != UNKNOWN_SERVER_VERSION
+        if msg.pg_version != self.state.server.pg_version
+            && self.state.server.pg_version != UNKNOWN_SERVER_VERSION
         {
             info!(
                 "incompatible server version {}, expected {}",
-                msg.pg_version, self.s.server.pg_version
+                msg.pg_version, self.state.server.pg_version
             );
         }
-        if msg.tenant_id != self.s.tenant_id {
+        if msg.tenant_id != self.state.tenant_id {
             bail!(
                 "invalid tenant ID, got {}, expected {}",
                 msg.tenant_id,
-                self.s.tenant_id
+                self.state.tenant_id
             );
         }
-        if msg.ztli != self.s.timeline_id {
+        if msg.ztli != self.state.timeline_id {
             bail!(
                 "invalid timeline ID, got {}, expected {}",
                 msg.ztli,
-                self.s.timeline_id
+                self.state.timeline_id
             );
         }
 
         // set basic info about server, if not yet
         // TODO: verify that is doesn't change after
-        self.s.server.system_id = msg.system_id;
-        self.s.server.wal_seg_size = msg.wal_seg_size;
-        self.control_store
-            .persist(&self.s)
-            .context("failed to persist shared state")?;
+        {
+            let mut state = self.state.clone();
+            state.server.system_id = msg.system_id;
+            state.server.wal_seg_size = msg.wal_seg_size;
+            self.state.persist(&state)?;
+        }
 
         // pass wal_seg_size to read WAL and find flush_lsn
-        self.wal_store.init_storage(&self.s)?;
+        self.wal_store.init_storage(&self.state)?;
 
         info!(
             "processed greeting from proposer {:?}, sending term {:?}",
-            msg.proposer_id, self.s.acceptor_state.term
+            msg.proposer_id, self.state.acceptor_state.term
         );
         Ok(Some(AcceptorProposerMessage::Greeting(AcceptorGreeting {
-            term: self.s.acceptor_state.term,
+            term: self.state.acceptor_state.term,
         })))
     }
 
@@ -637,17 +645,19 @@ where
     ) -> Result<Option<AcceptorProposerMessage>> {
         // initialize with refusal
         let mut resp = VoteResponse {
-            term: self.s.acceptor_state.term,
+            term: self.state.acceptor_state.term,
             vote_given: false as u64,
             flush_lsn: self.wal_store.flush_lsn(),
-            truncate_lsn: self.s.peer_horizon_lsn,
+            truncate_lsn: self.state.peer_horizon_lsn,
             term_history: self.get_term_history(),
         };
-        if self.s.acceptor_state.term < msg.term {
-            self.s.acceptor_state.term = msg.term;
+        if self.state.acceptor_state.term < msg.term {
+            let mut state = self.state.clone();
+            state.acceptor_state.term = msg.term;
             // persist vote before sending it out
-            self.control_store.persist(&self.s)?;
-            resp.term = self.s.acceptor_state.term;
+            self.state.persist(&state)?;
+
+            resp.term = self.state.acceptor_state.term;
             resp.vote_given = true as u64;
         }
         info!("processed VoteRequest for term {}: {:?}", msg.term, &resp);
@@ -656,9 +666,10 @@ where
 
     /// Bump our term if received a note from elected proposer with higher one
     fn bump_if_higher(&mut self, term: Term) -> Result<()> {
-        if self.s.acceptor_state.term < term {
-            self.s.acceptor_state.term = term;
-            self.control_store.persist(&self.s)?;
+        if self.state.acceptor_state.term < term {
+            let mut state = self.state.clone();
+            state.acceptor_state.term = term;
+            self.state.persist(&state)?;
         }
         Ok(())
     }
@@ -666,9 +677,9 @@ where
     /// Form AppendResponse from current state.
     fn append_response(&self) -> AppendResponse {
         let ar = AppendResponse {
-            term: self.s.acceptor_state.term,
+            term: self.state.acceptor_state.term,
             flush_lsn: self.wal_store.flush_lsn(),
-            commit_lsn: self.s.commit_lsn,
+            commit_lsn: self.state.commit_lsn,
             // will be filled by the upper code to avoid bothering safekeeper
             hs_feedback: HotStandbyFeedback::empty(),
             zenith_feedback: ZenithFeedback::empty(),
@@ -681,7 +692,7 @@ where
         info!("received ProposerElected {:?}", msg);
         self.bump_if_higher(msg.term)?;
         // If our term is higher, ignore the message (next feedback will inform the compute)
-        if self.s.acceptor_state.term > msg.term {
+        if self.state.acceptor_state.term > msg.term {
             return Ok(None);
         }
 
@@ -692,8 +703,11 @@ where
         self.wal_store.truncate_wal(msg.start_streaming_at)?;
 
         // and now adopt term history from proposer
-        self.s.acceptor_state.term_history = msg.term_history.clone();
-        self.control_store.persist(&self.s)?;
+        {
+            let mut state = self.state.clone();
+            state.acceptor_state.term_history = msg.term_history.clone();
+            self.state.persist(&state)?;
+        }
 
         info!("start receiving WAL since {:?}", msg.start_streaming_at);
 
@@ -715,13 +729,13 @@ where
         // Also note that commit_lsn can reach epoch_start_lsn earlier
         // that we receive new epoch_start_lsn, and we still need to sync
         // control file in this case.
-        if commit_lsn == self.epoch_start_lsn && self.s.commit_lsn != commit_lsn {
+        if commit_lsn == self.elected.epoch_start_lsn && self.state.commit_lsn != commit_lsn {
             self.persist_control_file()?;
         }
 
         // We got our first commit_lsn, which means we should sync
         // everything to disk, to initialize the state.
-        if self.s.commit_lsn == Lsn(0) && commit_lsn > Lsn(0) {
+        if self.state.commit_lsn == Lsn(0) && commit_lsn > Lsn(0) {
             self.wal_store.flush_wal()?;
             self.persist_control_file()?;
         }
@@ -731,10 +745,12 @@ where
 
     /// Persist in-memory state to the disk.
     fn persist_control_file(&mut self) -> Result<()> {
-        self.s.commit_lsn = self.inmem.commit_lsn;
-        self.s.peer_horizon_lsn = self.inmem.peer_horizon_lsn;
+        let mut state = self.state.clone();
 
-        self.control_store.persist(&self.s)
+        state.commit_lsn = self.inmem.commit_lsn;
+        state.peer_horizon_lsn = self.inmem.peer_horizon_lsn;
+        state.proposer_uuid = self.elected.proposer_uuid;
+        self.state.persist(&state)
     }
 
     /// Handle request to append WAL.
@@ -744,22 +760,21 @@ where
         msg: &AppendRequest,
         require_flush: bool,
     ) -> Result<Option<AcceptorProposerMessage>> {
-        if self.s.acceptor_state.term < msg.h.term {
+        if self.state.acceptor_state.term < msg.h.term {
             bail!("got AppendRequest before ProposerElected");
         }
 
         // If our term is higher, immediately refuse the message.
-        if self.s.acceptor_state.term > msg.h.term {
-            let resp = AppendResponse::term_only(self.s.acceptor_state.term);
+        if self.state.acceptor_state.term > msg.h.term {
+            let resp = AppendResponse::term_only(self.state.acceptor_state.term);
             return Ok(Some(AcceptorProposerMessage::AppendResponse(resp)));
         }
 
         // Now we know that we are in the same term as the proposer,
         // processing the message.
 
-        self.epoch_start_lsn = msg.h.epoch_start_lsn;
-        // TODO: don't update state without persisting to disk
-        self.s.proposer_uuid = msg.h.proposer_uuid;
+        self.elected.epoch_start_lsn = msg.h.epoch_start_lsn;
+        self.elected.proposer_uuid = msg.h.proposer_uuid;
 
         // do the job
         if !msg.wal_data.is_empty() {
@@ -790,7 +805,7 @@ where
         // Update truncate and commit LSN in control file.
         // To avoid negative impact on performance of extra fsync, do it only
         // when truncate_lsn delta exceeds WAL segment size.
-        if self.s.peer_horizon_lsn + (self.s.server.wal_seg_size as u64)
+        if self.state.peer_horizon_lsn + (self.state.server.wal_seg_size as u64)
             < self.inmem.peer_horizon_lsn
         {
             self.persist_control_file()?;
@@ -829,6 +844,8 @@ where
 
 #[cfg(test)]
 mod tests {
+    use std::ops::Deref;
+
     use super::*;
     use crate::wal_storage::Storage;
 
@@ -844,6 +861,14 @@ mod tests {
         }
     }
 
+    impl Deref for InMemoryState {
+        type Target = SafeKeeperState;
+
+        fn deref(&self) -> &Self::Target {
+            &self.persisted_state
+        }
+    }
+
     struct DummyWalStore {
         lsn: Lsn,
     }
@@ -879,7 +904,7 @@ mod tests {
         };
         let wal_store = DummyWalStore { lsn: Lsn(0) };
         let ztli = ZTimelineId::from([0u8; 16]);
-        let mut sk = SafeKeeper::new(ztli, storage, wal_store, SafeKeeperState::empty()).unwrap();
+        let mut sk = SafeKeeper::new(ztli, storage, wal_store).unwrap();
 
         // check voting for 1 is ok
         let vote_request = ProposerAcceptorMessage::VoteRequest(VoteRequest { term: 1 });
@@ -890,11 +915,11 @@ mod tests {
         }
 
         // reboot...
-        let state = sk.control_store.persisted_state.clone();
+        let state = sk.state.persisted_state.clone();
         let storage = InMemoryState {
-            persisted_state: state.clone(),
+            persisted_state: state,
         };
-        sk = SafeKeeper::new(ztli, storage, sk.wal_store, state).unwrap();
+        sk = SafeKeeper::new(ztli, storage, sk.wal_store).unwrap();
 
         // and ensure voting second time for 1 is not ok
         vote_resp = sk.process_msg(&vote_request);
@@ -911,7 +936,7 @@ mod tests {
         };
         let wal_store = DummyWalStore { lsn: Lsn(0) };
         let ztli = ZTimelineId::from([0u8; 16]);
-        let mut sk = SafeKeeper::new(ztli, storage, wal_store, SafeKeeperState::empty()).unwrap();
+        let mut sk = SafeKeeper::new(ztli, storage, wal_store).unwrap();
 
         let mut ar_hdr = AppendRequestHeader {
             term: 1,

walkeeper/src/timeline.rs

@@ -21,7 +21,6 @@ use crate::broker::SafekeeperInfo;
 use crate::callmemaybe::{CallmeEvent, SubscriptionStateKey};
 
 use crate::control_file;
-use crate::control_file::Storage as cf_storage;
 use crate::safekeeper::{
     AcceptorProposerMessage, ProposerAcceptorMessage, SafeKeeper, SafeKeeperState,
     SafekeeperMemState,
@@ -98,10 +97,9 @@ impl SharedState {
         peer_ids: Vec<ZNodeId>,
     ) -> Result<Self> {
         let state = SafeKeeperState::new(zttid, peer_ids);
-        let control_store = control_file::FileStorage::new(zttid, conf);
+        let control_store = control_file::FileStorage::create_new(zttid, conf, state)?;
         let wal_store = wal_storage::PhysicalStorage::new(zttid, conf);
-        let mut sk = SafeKeeper::new(zttid.timeline_id, control_store, wal_store, state)?;
-        sk.control_store.persist(&sk.s)?;
+        let sk = SafeKeeper::new(zttid.timeline_id, control_store, wal_store)?;
 
         Ok(Self {
             notified_commit_lsn: Lsn(0),
@@ -116,18 +114,14 @@ impl SharedState {
     /// Restore SharedState from control file.
     /// If file doesn't exist, bails out.
     fn restore(conf: &SafeKeeperConf, zttid: &ZTenantTimelineId) -> Result<Self> {
-        let state = control_file::FileStorage::load_control_file_conf(conf, zttid)
-            .context("failed to load from control file")?;
-
-        let control_store = control_file::FileStorage::new(zttid, conf);
-
+        let control_store = control_file::FileStorage::restore_new(zttid, conf)?;
         let wal_store = wal_storage::PhysicalStorage::new(zttid, conf);
 
         info!("timeline {} restored", zttid.timeline_id);
 
         Ok(Self {
             notified_commit_lsn: Lsn(0),
-            sk: SafeKeeper::new(zttid.timeline_id, control_store, wal_store, state)?,
+            sk: SafeKeeper::new(zttid.timeline_id, control_store, wal_store)?,
             replicas: Vec::new(),
             active: false,
             num_computes: 0,
@@ -419,7 +413,7 @@ impl Timeline {
 
     pub fn get_state(&self) -> (SafekeeperMemState, SafeKeeperState) {
         let shared_state = self.mutex.lock().unwrap();
-        (shared_state.sk.inmem.clone(), shared_state.sk.s.clone())
+        (shared_state.sk.inmem.clone(), shared_state.sk.state.clone())
     }
 
     /// Prepare public safekeeper info for reporting.