Add in-memory storage

Cargo.toml

@@ -7,7 +7,6 @@ members = [
     "postgres_ffi",
     "zenith_utils",
     "workspace_hack",
-    "proxy"
 ]
 
 [profile.release]

Dockerfile

@@ -82,7 +82,6 @@ RUN apk add --update openssl build-base
 RUN apk --no-cache --update --repository https://dl-cdn.alpinelinux.org/alpine/edge/testing add rocksdb
 COPY --from=build /zenith/target/release/pageserver /usr/local/bin
 COPY --from=build /zenith/target/release/wal_acceptor /usr/local/bin
-COPY --from=build /zenith/target/release/proxy /usr/local/bin
 COPY --from=pg-build /zenith/tmp_install /usr/local
 COPY docker-entrypoint.sh /docker-entrypoint.sh
 

Makefile

@@ -2,6 +2,7 @@
 UNAME_S := $(shell uname -s)
 ifeq ($(UNAME_S),Linux)
 	SECCOMP = --with-libseccomp
+	SECCOMP =
 else
 	SECCOMP =
 endif

Pipfile

@@ -1 +0,0 @@
-./test_runner/Pipfile

Pipfile.lock

@@ -1 +0,0 @@
-./test_runner/Pipfile.lock

README.md

@@ -117,7 +117,7 @@ To view your documentation in a browser, try running `cargo doc --no-deps --open
 `/control_plane`:
 
 Local control plane.
-Functions to start, configure and stop pageserver and postgres instances running as a local processes.
+Functions to start, cofigure and stop pageserver and postgres instances running as a local processes.
 Intended to be used in integration tests and in CLI tools for local installations.
 
 `/zenith`

control_plane/src/compute.rs

@@ -297,7 +297,9 @@ impl PostgresNode {
                 "shared_preload_libraries = zenith \n\
                  zenith.page_server_connstring = 'host={} port={}'\n\
                  zenith.zenith_timeline='{}'\n",
-                host, port, self.timelineid
+                host,
+                port,
+                self.timelineid
             ),
         )?;
 

control_plane/src/lib.rs

@@ -1,7 +1,7 @@
 //
 // Local control plane.
 //
-// Can start, configure and stop postgres instances running as a local processes.
+// Can start, cofigure and stop postgres instances running as a local processes.
 //
 // Intended to be used in integration tests and in CLI tools for
 // local installations.

control_plane/src/storage.rs

@@ -1,5 +1,5 @@
 use std::collections::HashMap;
-use std::net::TcpStream;
+use std::net::{TcpStream};
 use std::path::PathBuf;
 use std::process::Command;
 use std::thread;
@@ -12,8 +12,8 @@ use postgres::{Config, NoTls};
 
 use crate::local_env::LocalEnv;
 use crate::read_pidfile;
-use pageserver::branches::BranchInfo;
 use zenith_utils::connstring::connection_address;
+use pageserver::branches::BranchInfo;
 
 //
 // Control routines for pageserver.
@@ -36,9 +36,7 @@ impl PageServerNode {
     }
 
     fn default_config() -> Config {
-        "postgresql://no_user@localhost:64000/no_db"
-            .parse()
-            .unwrap()
+        "postgresql://no_user@localhost:64000/no_db".parse().unwrap()
     }
 
     pub fn connection_config(&self) -> Config {

pageserver/Cargo.toml

@@ -38,7 +38,7 @@ walkdir = "2"
 thiserror = "1.0"
 hex = "0.4.3"
 tar = "0.4.33"
-humantime = "2.1.0"
+parse_duration = "2.1.1"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1"
 fs_extra = "1.2.0"

pageserver/src/basebackup.rs

@@ -31,7 +31,6 @@ pub struct Basebackup<'a> {
     ar: Builder<&'a mut dyn Write>,
     timeline: &'a Arc<dyn Timeline>,
     lsn: Lsn,
-    prev_record_lsn: Lsn,
     snappath: String,
     slru_buf: [u8; pg_constants::SLRU_SEG_SIZE],
     slru_segno: u32,
@@ -44,14 +43,12 @@ impl<'a> Basebackup<'a> {
         timelineid: ZTimelineId,
         timeline: &'a Arc<dyn Timeline>,
         lsn: Lsn,
-        prev_record_lsn: Lsn,
         snapshot_lsn: Lsn,
     ) -> Basebackup<'a> {
         Basebackup {
             ar: Builder::new(write),
             timeline,
             lsn,
-            prev_record_lsn,
             snappath: format!("timelines/{}/snapshots/{:016X}", timelineid, snapshot_lsn.0),
             slru_path: "",
             slru_segno: u32::MAX,
@@ -123,8 +120,8 @@ impl<'a> Basebackup<'a> {
     }
 
     //
-    // Generate SLRU segment files from repository. Path identifies SLRU kind (pg_xact, pg_multixact/members, ...).
-    // Intially pass an empty string.
+    // Generate SRLU segment files from repository. Path identifiers SLRU kind (pg_xact, pg_multixact/members, ...).
+    // Intiallly pass is empty string.
     //
     fn add_slru_segment(
         &mut self,
@@ -132,9 +129,7 @@ impl<'a> Basebackup<'a> {
         tag: &ObjectTag,
         page: u32,
     ) -> anyhow::Result<()> {
-        let img = self
-            .timeline
-            .get_page_at_lsn_nowait(*tag, self.lsn, false)?;
+        let img = self.timeline.get_page_at_lsn_nowait(*tag, self.lsn)?;
         // Zero length image indicates truncated segment: just skip it
         if !img.is_empty() {
             assert!(img.len() == pg_constants::BLCKSZ as usize);
@@ -174,9 +169,7 @@ impl<'a> Basebackup<'a> {
     // Extract pg_filenode.map files from repository
     //
     fn add_relmap_file(&mut self, tag: &ObjectTag, db: &DatabaseTag) -> anyhow::Result<()> {
-        let img = self
-            .timeline
-            .get_page_at_lsn_nowait(*tag, self.lsn, false)?;
+        let img = self.timeline.get_page_at_lsn_nowait(*tag, self.lsn)?;
         info!("add_relmap_file {:?}", db);
         let path = if db.spcnode == pg_constants::GLOBALTABLESPACE_OID {
             String::from("global/pg_filenode.map") // filenode map for global tablespace
@@ -202,9 +195,7 @@ impl<'a> Basebackup<'a> {
         if self.timeline.get_tx_status(xid, self.lsn)?
             == pg_constants::TRANSACTION_STATUS_IN_PROGRESS
         {
-            let img = self
-                .timeline
-                .get_page_at_lsn_nowait(*tag, self.lsn, false)?;
+            let img = self.timeline.get_page_at_lsn_nowait(*tag, self.lsn)?;
             let mut buf = BytesMut::new();
             buf.extend_from_slice(&img[..]);
             let crc = crc32c::crc32c(&img[..]);
@@ -220,12 +211,12 @@ impl<'a> Basebackup<'a> {
     // Add generated pg_control file
     //
     fn add_pgcontrol_file(&mut self) -> anyhow::Result<()> {
-        let checkpoint_bytes =
-            self.timeline
-                .get_page_at_lsn_nowait(ObjectTag::Checkpoint, self.lsn, false)?;
-        let pg_control_bytes =
-            self.timeline
-                .get_page_at_lsn_nowait(ObjectTag::ControlFile, self.lsn, false)?;
+        let checkpoint_bytes = self
+            .timeline
+            .get_page_at_lsn_nowait(ObjectTag::Checkpoint, self.lsn)?;
+        let pg_control_bytes = self
+            .timeline
+            .get_page_at_lsn_nowait(ObjectTag::ControlFile, self.lsn)?;
         let mut pg_control = ControlFileData::decode(&pg_control_bytes)?;
         let mut checkpoint = CheckPoint::decode(&checkpoint_bytes)?;
 
@@ -248,10 +239,8 @@ impl<'a> Basebackup<'a> {
         pg_control.state = pg_constants::DB_SHUTDOWNED;
 
         // add zenith.signal file
-        self.ar.append(
-            &new_tar_header("zenith.signal", 8)?,
-            &self.prev_record_lsn.0.to_le_bytes()[..],
-        )?;
+        self.ar
+            .append(&new_tar_header("zenith.signal", 0)?, &b""[..])?;
 
         //send pg_control
         let pg_control_bytes = pg_control.encode();

pageserver/src/bin/pageserver.rs

@@ -26,9 +26,7 @@ use pageserver::{branches, page_cache, page_service, tui, PageServerConf};
 const DEFAULT_LISTEN_ADDR: &str = "127.0.0.1:64000";
 
 const DEFAULT_GC_HORIZON: u64 = 64 * 1024 * 1024;
-const DEFAULT_GC_PERIOD: Duration = Duration::from_secs(100);
-
-const DEFAULT_WAL_REDOERS: usize = 1;
+const DEFAULT_GC_PERIOD: Duration = Duration::from_secs(10);
 
 /// String arguments that can be declared via CLI or config file
 #[derive(Serialize, Deserialize)]
@@ -36,7 +34,6 @@ struct CfgFileParams {
     listen_addr: Option<String>,
     gc_horizon: Option<String>,
     gc_period: Option<String>,
-    wal_redoers: Option<String>,
     pg_distrib_dir: Option<String>,
 }
 
@@ -51,7 +48,6 @@ impl CfgFileParams {
             listen_addr: get_arg("listen"),
             gc_horizon: get_arg("gc_horizon"),
             gc_period: get_arg("gc_period"),
-            wal_redoers: get_arg("wal_redoers"),
             pg_distrib_dir: get_arg("postgres-distrib"),
         }
     }
@@ -63,7 +59,6 @@ impl CfgFileParams {
             listen_addr: self.listen_addr.or(other.listen_addr),
             gc_horizon: self.gc_horizon.or(other.gc_horizon),
             gc_period: self.gc_period.or(other.gc_period),
-            wal_redoers: self.wal_redoers.or(other.wal_redoers),
             pg_distrib_dir: self.pg_distrib_dir.or(other.pg_distrib_dir),
         }
     }
@@ -79,14 +74,10 @@ impl CfgFileParams {
             Some(horizon_str) => horizon_str.parse()?,
             None => DEFAULT_GC_HORIZON,
         };
-        let gc_period = match self.gc_period.as_ref() {
-            Some(period_str) => humantime::parse_duration(period_str)?,
-            None => DEFAULT_GC_PERIOD,
-        };
 
-        let wal_redoers = match self.wal_redoers.as_ref() {
-            Some(wal_redoers_str) => wal_redoers_str.parse::<usize>()?,
-            None => DEFAULT_WAL_REDOERS,
+        let gc_period: Duration = match self.gc_period.as_ref() {
+            Some(period_str) => parse_duration::parse(period_str)?,
+            None => DEFAULT_GC_PERIOD,
         };
 
         let pg_distrib_dir = match self.pg_distrib_dir.as_ref() {
@@ -101,12 +92,11 @@ impl CfgFileParams {
         Ok(PageServerConf {
             daemonize: false,
             interactive: false,
-            materialize: false,
 
             listen_addr,
             gc_horizon,
             gc_period,
-            wal_redoers,
+
             workdir: PathBuf::from("."),
 
             pg_distrib_dir,
@@ -131,12 +121,6 @@ fn main() -> Result<()> {
                 .takes_value(false)
                 .help("Interactive mode"),
         )
-        .arg(
-            Arg::with_name("materialize")
-                .long("materialize")
-                .takes_value(false)
-                .help("Materialize pages constructed by get_page_at"),
-        )
         .arg(
             Arg::with_name("daemonize")
                 .short("d")
@@ -162,12 +146,6 @@ fn main() -> Result<()> {
                 .takes_value(true)
                 .help("Interval between garbage collector iterations"),
         )
-        .arg(
-            Arg::with_name("wal_redoers")
-                .long("wal_redoers")
-                .takes_value(true)
-                .help("Number of wal-redo postgres instances"),
-        )
         .arg(
             Arg::with_name("workdir")
                 .short("D")
@@ -204,7 +182,6 @@ fn main() -> Result<()> {
 
     conf.daemonize = arg_matches.is_present("daemonize");
     conf.interactive = arg_matches.is_present("interactive");
-    conf.materialize = arg_matches.is_present("materialize");
 
     if init && (conf.daemonize || conf.interactive) {
         eprintln!("--daemonize and --interactive may not be used with --init");
@@ -292,8 +269,7 @@ fn start_pageserver(conf: &'static PageServerConf) -> Result<()> {
         }
     }
 
-    // Check that we can bind to address before starting threads to simplify shutdown
-    // sequence if port is occupied.
+    // Check that we can bind to address before further initialization
     info!("Starting pageserver on {}", conf.listen_addr);
     let pageserver_listener = TcpListener::bind(conf.listen_addr.clone())?;
 

pageserver/src/branches.rs

@@ -100,8 +100,8 @@ pub fn init_repo(conf: &'static PageServerConf, repo_dir: &Path) -> Result<()> {
     // and we failed to run initdb again in the same directory. This has been solved for the
     // rapid init+start case now, but the general race condition remains if you restart the
     // server quickly.
-    let storage = crate::rocksdb_storage::RocksObjectStore::create(conf)?;
-    //let storage = crate::inmem_storage::InmemObjectStore::create(conf)?;
+    //let storage = crate::rocksdb_storage::RocksObjectStore::create(conf)?;
+    let storage = crate::inmem_storage::InmemObjectStore::create(conf)?;
 
     let repo = crate::object_repository::ObjectRepository::new(
         conf,

pageserver/src/inmem_storage.rs

@@ -7,14 +7,14 @@ use crate::repository::RelTag;
 use crate::PageServerConf;
 use crate::ZTimelineId;
 use anyhow::{bail, Result};
-use serde::{Deserialize, Serialize};
-use std::collections::{BTreeMap, HashSet};
-use std::fs::File;
-use std::io::prelude::*;
-use std::ops::Bound::*;
+use std::collections::{BTreeMap,HashSet};
 use std::sync::RwLock;
-use zenith_utils::bin_ser::BeSer;
 use zenith_utils::lsn::Lsn;
+use std::ops::Bound::*;
+use serde::{Deserialize, Serialize};
+use zenith_utils::bin_ser::BeSer;
+use std::io::prelude::*;
+use std::fs::File;
 
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Deserialize, Serialize)]
 pub struct StorageKey {
@@ -64,7 +64,7 @@ impl ObjectStore for InmemObjectStore {
             let key = pair.0;
             return Ok(Some(key.obj_key.clone()));
         }
-        Ok(None)
+		Ok(None)
     }
 
     fn put(&self, key: &ObjectKey, lsn: Lsn, value: &[u8]) -> Result<()> {
@@ -104,12 +104,9 @@ impl ObjectStore for InmemObjectStore {
             lsn,
         };
         let db = self.db.read().unwrap();
-        let versions: Vec<(Lsn, Vec<u8>)> = db
-            .range(from..=till)
-            .map(|pair| (pair.0.lsn, pair.1.clone()))
-            .collect();
+		let versions: Vec<(Lsn, Vec<u8>)> = db.range(from..=till).map(|pair|(pair.0.lsn, pair.1.clone())).collect();
         Ok(Box::new(InmemObjectVersionIter::new(versions)))
-    }
+	}
 
     /// Iterate through all timeline objects
     fn list_objects<'a>(
@@ -121,10 +118,10 @@ impl ObjectStore for InmemObjectStore {
         let curr_key = StorageKey::timeline_start(timeline);
 
         Ok(Box::new(InmemObjectIter {
-            store: &self,
+			store: &self,
             curr_key,
             timeline,
-            nonrel_only,
+			nonrel_only,
             lsn,
         }))
     }
@@ -152,38 +149,38 @@ impl ObjectStore for InmemObjectStore {
             relnode: 0,
             forknum: 0u8,
         };
-        let db = self.db.read().unwrap();
-        'outer: loop {
-            let search_key = StorageKey {
-                obj_key: ObjectKey {
-                    timeline: timelineid,
-                    tag: ObjectTag::RelationMetadata(search_rel_tag),
-                },
-                lsn: Lsn(0),
-            };
-            for pair in db.range(&search_key..) {
-                let key = pair.0;
+		let db = self.db.read().unwrap();
+		'outer: loop {
+			let search_key = StorageKey {
+				obj_key: ObjectKey {
+					timeline: timelineid,
+					tag: ObjectTag::RelationMetadata(search_rel_tag),
+				},
+				lsn: Lsn(0),
+			};
+			for pair in db.range(&search_key..) {
+				let key = pair.0;
 
-                if let ObjectTag::RelationMetadata(rel_tag) = key.obj_key.tag {
-                    if spcnode != 0 && rel_tag.spcnode != spcnode
-                        || dbnode != 0 && rel_tag.dbnode != dbnode
-                    {
-                        break 'outer;
-                    }
-                    if key.lsn <= lsn {
-                        // visible in this snapshot
-                        rels.insert(rel_tag);
-                    }
-                    search_rel_tag = rel_tag;
-                    // skip to next relation
-                    // FIXME: What if relnode is u32::MAX ?
-                    search_rel_tag.relnode += 1;
-                    continue 'outer;
-                } else {
-                    // no more relation metadata entries
-                    break 'outer;
-                }
-            }
+				if let ObjectTag::RelationMetadata(rel_tag) = key.obj_key.tag {
+					if spcnode != 0 && rel_tag.spcnode != spcnode
+						|| dbnode != 0 && rel_tag.dbnode != dbnode
+					{
+						break 'outer;
+					}
+					if key.lsn <= lsn {
+						// visible in this snapshot
+						rels.insert(rel_tag);
+					}
+					search_rel_tag = rel_tag;
+					// skip to next relation
+					// FIXME: What if relnode is u32::MAX ?
+					search_rel_tag.relnode += 1;
+					continue  'outer;
+				} else {
+					// no more relation metadata entries
+					break 'outer;
+				}
+			}
         }
 
         Ok(rels)
@@ -201,33 +198,37 @@ impl ObjectStore for InmemObjectStore {
         let curr_key = StorageKey::timeline_start(timeline);
 
         Ok(Box::new(InmemObjects {
-            store: &self,
+			store: &self,
             curr_key,
             timeline,
             lsn,
         }))
     }
 
-    fn compact(&self) {}
+    fn compact(&self) {
+    }
 }
 
 impl Drop for InmemObjectStore {
-    fn drop(&mut self) {
-        let path = self.conf.workdir.join("objstore.dmp");
-        let mut f = File::create(path).unwrap();
-        f.write(&self.db.ser().unwrap()).unwrap();
-    }
+	fn drop(&mut self) {
+		let path = self.conf.workdir.join("objstore.dmp");
+		let mut f = File::create(path).unwrap();
+		f.write(&self.db.ser().unwrap()).unwrap();
+	}
 }
 
 impl InmemObjectStore {
     pub fn open(conf: &'static PageServerConf) -> Result<InmemObjectStore> {
-        let path = conf.workdir.join("objstore.dmp");
-        let mut f = File::open(path)?;
-        let mut buffer = Vec::new();
-        // read the whole file
-        f.read_to_end(&mut buffer)?;
-        let db = RwLock::new(BTreeMap::des(&buffer)?);
-        Ok(InmemObjectStore { conf: conf, db })
+		let path = conf.workdir.join("objstore.dmp");
+		let mut f = File::open(path)?;
+		let mut buffer = Vec::new();
+		// read the whole file
+		f.read_to_end(&mut buffer)?;
+		let db = RwLock::new(BTreeMap::des(&buffer)?);
+		Ok(InmemObjectStore {
+            conf: conf,
+            db
+        })
     }
 
     pub fn create(conf: &'static PageServerConf) -> Result<InmemObjectStore> {
@@ -243,31 +244,34 @@ impl InmemObjectStore {
 /// reverse LSN order.
 ///
 struct InmemObjectVersionIter {
-    versions: Vec<(Lsn, Vec<u8>)>,
-    curr: usize,
+	versions: Vec<(Lsn, Vec<u8>)>,
+	curr: usize,
 }
 impl InmemObjectVersionIter {
     fn new(versions: Vec<(Lsn, Vec<u8>)>) -> InmemObjectVersionIter {
-        let curr = versions.len();
-        InmemObjectVersionIter { versions, curr }
+		let curr = versions.len();
+        InmemObjectVersionIter {
+            versions,
+			curr
+		}
     }
 }
 impl Iterator for InmemObjectVersionIter {
     type Item = (Lsn, Vec<u8>);
 
     fn next(&mut self) -> std::option::Option<Self::Item> {
-        if self.curr == 0 {
-            None
-        } else {
-            self.curr -= 1;
-            Some(self.versions[self.curr].clone())
-        }
+		if self.curr == 0 {
+			None
+		} else {
+			self.curr -= 1;
+			Some(self.versions[self.curr].clone())
+		}
     }
 }
 
 struct InmemObjects<'r> {
     store: &'r InmemObjectStore,
-    curr_key: StorageKey,
+	curr_key: StorageKey,
     timeline: ZTimelineId,
     lsn: Lsn,
 }
@@ -277,26 +281,26 @@ impl<'r> Iterator for InmemObjects<'r> {
     type Item = Result<(ObjectTag, Lsn, Vec<u8>)>;
 
     fn next(&mut self) -> Option<Self::Item> {
-        self.next_result().transpose()
+		self.next_result().transpose()
     }
 }
 
 impl<'r> InmemObjects<'r> {
     fn next_result(&mut self) -> Result<Option<(ObjectTag, Lsn, Vec<u8>)>> {
-        let db = self.store.db.read().unwrap();
-        for pair in db.range((Excluded(&self.curr_key), Unbounded)) {
-            let key = pair.0;
-            if key.obj_key.timeline != self.timeline {
+		let db = self.store.db.read().unwrap();
+        for pair in db.range((Excluded(&self.curr_key),Unbounded)) {
+			let key = pair.0;
+			if key.obj_key.timeline != self.timeline {
                 return Ok(None);
             }
-            if key.lsn > self.lsn {
+			if key.lsn > self.lsn {
                 // TODO can speed up by seeking iterator
                 continue;
             }
-            self.curr_key = key.clone();
-            let value = pair.1.clone();
+			self.curr_key = key.clone();
+			let value = pair.1.clone();
             return Ok(Some((key.obj_key.tag, key.lsn, value)));
-        }
+		}
         Ok(None)
     }
 }
@@ -306,9 +310,9 @@ impl<'r> InmemObjects<'r> {
 ///
 struct InmemObjectIter<'a> {
     store: &'a InmemObjectStore,
-    curr_key: StorageKey,
+	curr_key: StorageKey,
     timeline: ZTimelineId,
-    nonrel_only: bool,
+	nonrel_only: bool,
     lsn: Lsn,
 }
 
@@ -316,30 +320,30 @@ impl<'a> Iterator for InmemObjectIter<'a> {
     type Item = ObjectTag;
 
     fn next(&mut self) -> std::option::Option<Self::Item> {
-        let db = self.store.db.read().unwrap();
-        'outer: loop {
-            for pair in db.range((Excluded(&self.curr_key), Unbounded)) {
-                let key = pair.0;
-                if key.obj_key.timeline != self.timeline {
-                    return None;
-                }
-                self.curr_key = key.clone();
-                self.curr_key.lsn = Lsn(u64::MAX); // next seek should skip all versions
-                if key.lsn <= self.lsn {
-                    // visible in this snapshot
-                    if self.nonrel_only {
-                        match key.obj_key.tag {
-                            ObjectTag::RelationMetadata(_) => return None,
-                            ObjectTag::RelationBuffer(_) => return None,
-                            _ => return Some(key.obj_key.tag),
-                        }
-                    } else {
-                        return Some(key.obj_key.tag);
-                    }
-                }
-                continue 'outer;
-            }
-            return None;
-        }
+		let db = self.store.db.read().unwrap();
+		'outer: loop {
+			for pair in db.range((Excluded(&self.curr_key),Unbounded)) {
+				let key = pair.0;
+				if key.obj_key.timeline != self.timeline {
+					return None;
+				}
+				self.curr_key = key.clone();
+				self.curr_key.lsn = Lsn(u64::MAX); // next seek should skip all versions
+				if key.lsn <= self.lsn {
+					// visible in this snapshot
+					if self.nonrel_only {
+						match key.obj_key.tag {
+							ObjectTag::RelationMetadata(_) => return None,
+							ObjectTag::RelationBuffer(_) => return None,
+							_ => return Some(key.obj_key.tag),
+						}
+					} else {
+						return Some(key.obj_key.tag);
+					}
+				}
+				continue 'outer;
+			}
+			return None;
+		}
     }
 }

pageserver/src/lib.rs

@@ -7,7 +7,6 @@ use std::time::Duration;
 
 pub mod basebackup;
 pub mod branches;
-pub mod inmem_storage;
 pub mod object_key;
 pub mod object_repository;
 pub mod object_store;
@@ -16,6 +15,7 @@ pub mod page_service;
 pub mod repository;
 pub mod restore_local_repo;
 pub mod rocksdb_storage;
+pub mod inmem_storage;
 pub mod tui;
 pub mod tui_event;
 mod tui_logger;
@@ -27,11 +27,9 @@ pub mod walredo;
 pub struct PageServerConf {
     pub daemonize: bool,
     pub interactive: bool,
-    pub materialize: bool,
     pub listen_addr: String,
     pub gc_horizon: u64,
     pub gc_period: Duration,
-    pub wal_redoers: usize,
 
     // Repository directory, relative to current working directory.
     // Normally, the page server changes the current working directory

pageserver/src/object_repository.rs

@@ -71,7 +71,6 @@ impl Repository for ObjectRepository {
             Some(timeline) => Ok(timeline.clone()),
             None => {
                 let timeline = ObjectTimeline::open(
-                    self.conf,
                     Arc::clone(&self.obj_store),
                     timelineid,
                     self.walredo_mgr.clone(),
@@ -111,7 +110,6 @@ impl Repository for ObjectRepository {
         let metadata = MetadataEntry {
             last_valid_lsn: start_lsn,
             last_record_lsn: start_lsn,
-            prev_record_lsn: Lsn(0),
             ancestor_timeline: None,
             ancestor_lsn: start_lsn,
         };
@@ -125,7 +123,6 @@ impl Repository for ObjectRepository {
         info!("Created empty timeline {}", timelineid);
 
         let timeline = ObjectTimeline::open(
-            self.conf,
             Arc::clone(&self.obj_store),
             timelineid,
             self.walredo_mgr.clone(),
@@ -149,7 +146,6 @@ impl Repository for ObjectRepository {
         let metadata = MetadataEntry {
             last_valid_lsn: at_lsn,
             last_record_lsn: at_lsn,
-            prev_record_lsn: src_timeline.get_prev_record_lsn(),
             ancestor_timeline: Some(src),
             ancestor_lsn: at_lsn,
         };
@@ -165,7 +161,7 @@ impl Repository for ObjectRepository {
             match tag {
                 ObjectTag::TimelineMetadataTag => {} // skip it
                 _ => {
-                    let img = src_timeline.get_page_at_lsn_nowait(tag, at_lsn, false)?;
+                    let img = src_timeline.get_page_at_lsn_nowait(tag, at_lsn)?;
                     let val = ObjectValue::Page(PageEntry::Page(img));
                     let key = ObjectKey { timeline: dst, tag };
                     self.obj_store.put(&key, at_lsn, &ObjectValue::ser(&val)?)?;
@@ -199,7 +195,6 @@ pub struct ObjectTimeline {
 
     // Backing key-value store
     obj_store: Arc<dyn ObjectStore>,
-    conf: &'static PageServerConf,
 
     // WAL redo manager, for reconstructing page versions from WAL records.
     walredo_mgr: Arc<dyn WalRedoManager>,
@@ -221,7 +216,6 @@ pub struct ObjectTimeline {
     //
     last_valid_lsn: SeqWait<Lsn>,
     last_record_lsn: AtomicLsn,
-    prev_record_lsn: AtomicLsn,
 
     ancestor_timeline: Option<ZTimelineId>,
     ancestor_lsn: Lsn,
@@ -234,7 +228,6 @@ impl ObjectTimeline {
     ///
     /// Loads the metadata for the timeline into memory.
     fn open(
-        conf: &'static PageServerConf,
         obj_store: Arc<dyn ObjectStore>,
         timelineid: ZTimelineId,
         walredo_mgr: Arc<dyn WalRedoManager>,
@@ -248,11 +241,9 @@ impl ObjectTimeline {
         let timeline = ObjectTimeline {
             timelineid,
             obj_store,
-            conf,
             walredo_mgr,
             last_valid_lsn: SeqWait::new(metadata.last_valid_lsn),
             last_record_lsn: AtomicLsn::new(metadata.last_record_lsn.0),
-            prev_record_lsn: AtomicLsn::new(metadata.prev_record_lsn.0),
             ancestor_timeline: metadata.ancestor_timeline,
             ancestor_lsn: metadata.ancestor_lsn,
             rel_meta: RwLock::new(BTreeMap::new()),
@@ -270,15 +261,10 @@ impl Timeline for ObjectTimeline {
     fn get_page_at_lsn(&self, tag: ObjectTag, req_lsn: Lsn) -> Result<Bytes> {
         let lsn = self.wait_lsn(req_lsn)?;
 
-        self.get_page_at_lsn_nowait(tag, lsn, self.conf.materialize)
+        self.get_page_at_lsn_nowait(tag, lsn)
     }
 
-    fn get_page_at_lsn_nowait(
-        &self,
-        tag: ObjectTag,
-        req_lsn: Lsn,
-        materialize: bool,
-    ) -> Result<Bytes> {
+    fn get_page_at_lsn_nowait(&self, tag: ObjectTag, req_lsn: Lsn) -> Result<Bytes> {
         const ZERO_PAGE: [u8; 8192] = [0u8; 8192];
         // Look up the page entry. If it's a page image, return that. If it's a WAL record,
         // ask the WAL redo service to reconstruct the page image from the WAL records.
@@ -300,16 +286,14 @@ impl Timeline for ObjectTimeline {
                     let (base_img, records) = self.collect_records_for_apply(tag, lsn)?;
                     page_img = self.walredo_mgr.request_redo(tag, lsn, base_img, records)?;
 
-                    if materialize {
-                        // Garbage collection assumes that we remember the materialized page
-                        // version. Otherwise we could opt to not do it, with the downside that
-                        // the next GetPage@LSN call of the same page version would have to
-                        // redo the WAL again.
-                        self.put_page_image(tag, lsn, page_img.clone(), false)?;
-                    }
+                    // Garbage collection assumes that we remember the materialized page
+                    // version. Otherwise we could opt to not do it, with the downside that
+                    // the next GetPage@LSN call of the same page version would have to
+                    // redo the WAL again.
+                    self.put_page_image(tag, lsn, page_img.clone(), false)?;
                 }
                 ObjectValue::SLRUTruncate => page_img = Bytes::from_static(&ZERO_PAGE),
-                _ => bail!("Invalid object kind, expected a page entry or SLRU truncate"),
+                _ => bail!("Invalid object kind, expected a page entry or SRLU truncate"),
             }
             // FIXME: assumes little-endian. Only used for the debugging log though
             let page_lsn_hi = u32::from_le_bytes(page_img.get(0..4).unwrap().try_into().unwrap());
@@ -450,7 +434,7 @@ impl Timeline for ObjectTimeline {
         Ok(())
     }
 
-    /// Truncate SLRU segment
+    /// Truncate SRLU segment
     fn put_slru_truncate(&self, tag: ObjectTag, lsn: Lsn) -> Result<()> {
         let key = ObjectKey {
             timeline: self.timelineid,
@@ -485,20 +469,14 @@ impl Timeline for ObjectTimeline {
     ///
     /// Memorize a full image of a page version
     ///
-    fn put_page_image(
-        &self,
-        tag: ObjectTag,
-        lsn: Lsn,
-        img: Bytes,
-        update_meta: bool,
-    ) -> Result<()> {
+    fn put_page_image(&self, tag: ObjectTag, lsn: Lsn, img: Bytes, update_meta: bool) -> Result<()> {
         self.put_page_entry(&tag, lsn, PageEntry::Page(img))?;
 
+		if !update_meta {
+			return Ok(());
+		}
         debug!("put_page_image rel {:?} at {}", tag, lsn);
 
-        if !update_meta {
-            return Ok(());
-        }
         if let ObjectTag::RelationBuffer(tag) = tag {
             // Also check if this created or extended the file
             let old_nblocks = self.relsize_get_nowait(tag.rel, lsn)?.unwrap_or(0);
@@ -574,7 +552,6 @@ impl Timeline for ObjectTimeline {
         assert!(old == Lsn(0));
         let old = self.last_record_lsn.fetch_max(lsn);
         assert!(old == Lsn(0));
-        self.prev_record_lsn.store(Lsn(0));
     }
 
     /// Like `advance_last_valid_lsn`, but this always points to the end of
@@ -589,9 +566,6 @@ impl Timeline for ObjectTimeline {
         let old = self.last_record_lsn.fetch_max(lsn);
         assert!(old <= lsn);
 
-        // Use old value of last_record_lsn as prev_record_lsn
-        self.prev_record_lsn.fetch_max(old);
-
         // Also advance last_valid_lsn
         let old = self.last_valid_lsn.advance(lsn);
         // Can't move backwards.
@@ -606,10 +580,6 @@ impl Timeline for ObjectTimeline {
         self.last_record_lsn.load()
     }
 
-    fn get_prev_record_lsn(&self) -> Lsn {
-        self.prev_record_lsn.load()
-    }
-
     ///
     /// Flush to disk all data that was written with the put_* functions
     ///
@@ -623,7 +593,6 @@ impl Timeline for ObjectTimeline {
         let metadata = MetadataEntry {
             last_valid_lsn: self.last_valid_lsn.load(),
             last_record_lsn: self.last_record_lsn.load(),
-            prev_record_lsn: self.prev_record_lsn.load(),
             ancestor_timeline: self.ancestor_timeline,
             ancestor_lsn: self.ancestor_lsn,
         };
@@ -644,7 +613,7 @@ impl Timeline for ObjectTimeline {
         }))
     }
 
-    fn gc_iteration(&self, horizon: u64, compact: bool) -> Result<GcResult> {
+    fn gc_iteration(&self, horizon: u64) -> Result<GcResult> {
         let last_lsn = self.get_last_valid_lsn();
         let mut result: GcResult = Default::default();
 
@@ -652,7 +621,7 @@ impl Timeline for ObjectTimeline {
         if let Some(horizon) = last_lsn.checked_sub(horizon) {
             // WAL is large enough to perform GC
             let now = Instant::now();
-            let mut prepared_horizon = Lsn(u64::MAX);
+
             // Iterate through all objects in timeline
             for obj in self
                 .obj_store
@@ -667,11 +636,10 @@ impl Timeline for ObjectTimeline {
                             tag: obj,
                         };
                         for vers in self.obj_store.object_versions(&key, horizon)? {
-                            let lsn = vers.0;
-                            prepared_horizon = Lsn::min(lsn, prepared_horizon);
                             if self.get_tx_status(prepare.xid, horizon)?
                                 != pg_constants::TRANSACTION_STATUS_IN_PROGRESS
                             {
+                                let lsn = vers.0;
                                 self.obj_store.unlink(&key, lsn)?;
                                 result.prep_deleted += 1;
                             }
@@ -719,12 +687,10 @@ impl Timeline for ObjectTimeline {
                             if last_version {
                                 result.truncated += 1;
                                 last_version = false;
-                                if let Some(rel_size) =
-                                    self.relsize_get_nowait(tag.rel, last_lsn)?
-                                {
+                                if let Some(rel_size) = self.relsize_get_nowait(tag.rel, last_lsn)? {
                                     if rel_size > tag.blknum {
                                         // preserve and materialize last version before deleting all preceeding
-                                        self.get_page_at_lsn_nowait(obj, lsn, true)?;
+                                        self.get_page_at_lsn_nowait(obj, lsn)?;
                                         continue;
                                     }
                                     debug!("Drop last block {} of relation {:?} at {} because it is beyond relation size {}", tag.blknum, tag.rel, lsn, rel_size);
@@ -753,10 +719,7 @@ impl Timeline for ObjectTimeline {
                             timeline: self.timelineid,
                             tag: obj,
                         };
-                        for vers in self
-                            .obj_store
-                            .object_versions(&key, Lsn::min(prepared_horizon, horizon))?
-                        {
+                        for vers in self.obj_store.object_versions(&key, horizon)? {
                             let lsn = vers.0;
                             if last_version {
                                 let content = vers.1;
@@ -767,7 +730,7 @@ impl Timeline for ObjectTimeline {
                                     }
                                     ObjectValue::Page(PageEntry::WALRecord(_)) => {
                                         // preserve and materialize last version before deleting all preceeding
-                                        self.get_page_at_lsn_nowait(obj, lsn, true)?;
+                                        self.get_page_at_lsn_nowait(obj, lsn)?;
                                     }
                                     _ => {} // do nothing if already materialized
                                 }
@@ -805,9 +768,7 @@ impl Timeline for ObjectTimeline {
             result.elapsed = now.elapsed();
             info!("Garbage collection completed in {:?}: {} relations inspected, {} object inspected, {} version histories truncated, {} versions deleted, {} relations dropped",
                   result.elapsed, result.n_relations, result.inspected, result.truncated, result.deleted, result.dropped);
-            if compact {
-                self.obj_store.compact();
-            }
+            self.obj_store.compact();
         }
         Ok(result)
     }
@@ -914,7 +875,7 @@ impl ObjectTimeline {
     fn gc_loop(&self, conf: &'static PageServerConf) -> Result<()> {
         loop {
             thread::sleep(conf.gc_period);
-            self.gc_iteration(conf.gc_horizon, false)?;
+            self.gc_iteration(conf.gc_horizon)?;
         }
     }
 
@@ -1153,7 +1114,6 @@ const fn relation_size_key(timelineid: ZTimelineId, rel: RelTag) -> ObjectKey {
 pub struct MetadataEntry {
     last_valid_lsn: Lsn,
     last_record_lsn: Lsn,
-    prev_record_lsn: Lsn,
     ancestor_timeline: Option<ZTimelineId>,
     ancestor_lsn: Lsn,
 }

pageserver/src/page_cache.rs

@@ -5,8 +5,8 @@
 
 use crate::object_repository::ObjectRepository;
 use crate::repository::Repository;
-use crate::rocksdb_storage::RocksObjectStore;
-//use crate::inmem_storage::InmemObjectStore;
+//use crate::rocksdb_storage::RocksObjectStore;
+use crate::inmem_storage::InmemObjectStore;
 use crate::walredo::PostgresRedoManager;
 use crate::PageServerConf;
 use lazy_static::lazy_static;
@@ -19,8 +19,8 @@ lazy_static! {
 pub fn init(conf: &'static PageServerConf) {
     let mut m = REPOSITORY.lock().unwrap();
 
-    let obj_store = RocksObjectStore::open(conf).unwrap();
-    //let obj_store = InmemObjectStore::open(conf).unwrap();
+    //let obj_store = RocksObjectStore::open(conf).unwrap();
+    let obj_store = InmemObjectStore::open(conf).unwrap();
 
     // Set up a WAL redo manager, for applying WAL records.
     let walredo_mgr = PostgresRedoManager::new(conf);

pageserver/src/page_service.rs

@@ -19,8 +19,8 @@ use std::net::TcpListener;
 use std::str::FromStr;
 use std::thread;
 use std::{io, net::TcpStream};
+use zenith_utils::postgres_backend;
 use zenith_utils::postgres_backend::PostgresBackend;
-use zenith_utils::postgres_backend::{self, AuthType};
 use zenith_utils::pq_proto::{
     BeMessage, FeMessage, RowDescriptor, HELLO_WORLD_ROW, SINGLE_COL_ROWDESC,
 };
@@ -154,7 +154,7 @@ pub fn thread_main(conf: &'static PageServerConf, listener: TcpListener) -> anyh
 
 fn page_service_conn_main(conf: &'static PageServerConf, socket: TcpStream) -> anyhow::Result<()> {
     let mut conn_handler = PageServerHandler::new(conf);
-    let mut pgbackend = PostgresBackend::new(socket, AuthType::Trust)?;
+    let mut pgbackend = PostgresBackend::new(socket)?;
     pgbackend.run(&mut conn_handler)
 }
 
@@ -303,7 +303,6 @@ impl PageServerHandler {
                 timelineid,
                 &timeline,
                 req_lsn,
-                timeline.get_prev_record_lsn(),
                 snapshot_lsn,
             );
             basebackup.send_tarball()?;
@@ -532,7 +531,7 @@ impl postgres_backend::Handler for PageServerHandler {
                 .unwrap_or(&self.conf.gc_horizon.to_string())
                 .parse()?;
 
-            let result = timeline.gc_iteration(horizon, true)?;
+            let result = timeline.gc_iteration(horizon)?;
 
             pgb.write_message_noflush(&BeMessage::RowDescription(&[
                 RowDescriptor {

pageserver/src/repository.rs

@@ -59,7 +59,7 @@ pub trait Timeline: Send + Sync {
     fn get_page_at_lsn(&self, tag: ObjectTag, lsn: Lsn) -> Result<Bytes>;
 
     /// Look up given page in the cache.
-    fn get_page_at_lsn_nowait(&self, tag: ObjectTag, lsn: Lsn, materialize: bool) -> Result<Bytes>;
+    fn get_page_at_lsn_nowait(&self, tag: ObjectTag, lsn: Lsn) -> Result<Bytes>;
 
     /// Get size of relation
     fn get_rel_size(&self, tag: RelTag, lsn: Lsn) -> Result<u32>;
@@ -89,8 +89,7 @@ pub trait Timeline: Send + Sync {
     fn put_raw_data(&self, tag: ObjectTag, lsn: Lsn, data: &[u8]) -> Result<()>;
 
     /// Like put_wal_record, but with ready-made image of the page.
-    fn put_page_image(&self, tag: ObjectTag, lsn: Lsn, img: Bytes, update_meta: bool)
-        -> Result<()>;
+    fn put_page_image(&self, tag: ObjectTag, lsn: Lsn, img: Bytes, update_meta: bool) -> Result<()>;
 
     /// Truncate relation
     fn put_truncation(&self, rel: RelTag, lsn: Lsn, nblocks: u32) -> Result<()>;
@@ -98,7 +97,7 @@ pub trait Timeline: Send + Sync {
     /// Unlink relation. This method is used for marking dropped relations.
     fn put_unlink(&self, tag: RelTag, lsn: Lsn) -> Result<()>;
 
-    /// Truncate SLRU segment
+    /// Truncate SRLU segment
     fn put_slru_truncate(&self, tag: ObjectTag, lsn: Lsn) -> Result<()>;
 
     // Get object tag greater or equal than specified
@@ -121,9 +120,6 @@ pub trait Timeline: Send + Sync {
     fn advance_last_record_lsn(&self, lsn: Lsn);
     fn get_last_record_lsn(&self) -> Lsn;
 
-    // Like `advance_last_record_lsn`, but points to the start position of last record
-    fn get_prev_record_lsn(&self) -> Lsn;
-
     ///
     /// Flush to disk all data that was written with the put_* functions
     ///
@@ -144,14 +140,7 @@ pub trait Timeline: Send + Sync {
     /// but it can be explicitly requested through page server API.
     ///
     /// `horizon` specifies delta from last LSN to preserve all object versions (PITR interval).
-    /// `compact` parameter is used to force compaction of storage.
-    /// Some storage implementation are based on LSM tree and require periodic merge (compaction).
-    /// Usually storage implementation determines itself when compaction should be performed.
-    /// But for GC tests it way be useful to force compaction just after completion of GC iteration
-    /// to make sure that all detected garbage is removed.
-    /// So right now `compact` is set to true when GC explicitly requested through page srver API,
-    /// and is st to false in GC threads which infinitely repeats GC iterations in loop.
-    fn gc_iteration(&self, horizon: u64, compact: bool) -> Result<GcResult>;
+    fn gc_iteration(&self, horizon: u64) -> Result<GcResult>;
 
     // Check transaction status
     fn get_tx_status(&self, xid: TransactionId, lsn: Lsn) -> anyhow::Result<u8> {
@@ -318,12 +307,6 @@ mod tests {
         relnode: 1000,
         forknum: 0,
     };
-    const TESTREL_B: RelTag = RelTag {
-        spcnode: 0,
-        dbnode: 111,
-        relnode: 1001,
-        forknum: 0,
-    };
 
     /// Convenience function to create a BufferTag for testing.
     /// Helps to keeps the tests shorter.
@@ -353,10 +336,8 @@ mod tests {
         let conf = PageServerConf {
             daemonize: false,
             interactive: false,
-            materialize: false,
             gc_horizon: 64 * 1024 * 1024,
             gc_period: Duration::from_secs(10),
-            wal_redoers: 1,
             listen_addr: "127.0.0.1:5430".to_string(),
             workdir: repo_dir,
             pg_distrib_dir: "".into(),
@@ -507,62 +488,10 @@ mod tests {
         Ok(())
     }
 
-    ///
-    /// Test branch creation
-    ///
-    #[test]
-    fn test_branch() -> Result<()> {
-        let repo = get_test_repo("test_branch")?;
-        let timelineid = ZTimelineId::from_str("11223344556677881122334455667788").unwrap();
-        let tline = repo.create_empty_timeline(timelineid, Lsn(0))?;
-
-        // Create a relation on the timeline
-        tline.init_valid_lsn(Lsn(1));
-        tline.put_page_image(TEST_BUF(0), Lsn(2), TEST_IMG("foo blk 0 at 2"), true)?;
-        tline.put_page_image(TEST_BUF(0), Lsn(3), TEST_IMG("foo blk 0 at 3"), true)?;
-        tline.put_page_image(TEST_BUF(0), Lsn(4), TEST_IMG("foo blk 0 at 4"), true)?;
-
-        // Create another relation
-        let buftag2 = ObjectTag::RelationBuffer(BufferTag {
-            rel: TESTREL_B,
-            blknum: 0,
-        });
-        tline.put_page_image(buftag2, Lsn(2), TEST_IMG("foobar blk 0 at 2"), true)?;
-
-        tline.advance_last_valid_lsn(Lsn(4));
-
-        // Branch the history, modify relation differently on the new timeline
-        let newtimelineid = ZTimelineId::from_str("AA223344556677881122334455667788").unwrap();
-        repo.branch_timeline(timelineid, newtimelineid, Lsn(3))?;
-        let newtline = repo.get_timeline(newtimelineid)?;
-
-        newtline.put_page_image(TEST_BUF(0), Lsn(4), TEST_IMG("bar blk 0 at 4"), true)?;
-        newtline.advance_last_valid_lsn(Lsn(4));
-
-        // Check page contents on both branches
-        assert_eq!(
-            tline.get_page_at_lsn(TEST_BUF(0), Lsn(4))?,
-            TEST_IMG("foo blk 0 at 4")
-        );
-
-        assert_eq!(
-            newtline.get_page_at_lsn(TEST_BUF(0), Lsn(4))?,
-            TEST_IMG("bar blk 0 at 4")
-        );
-
-        assert_eq!(
-            newtline.get_page_at_lsn(buftag2, Lsn(4))?,
-            TEST_IMG("foobar blk 0 at 2")
-        );
-
-        assert_eq!(newtline.get_rel_size(TESTREL_B, Lsn(4))?, 1);
-
-        Ok(())
-    }
-
     #[test]
     fn test_history() -> Result<()> {
         let repo = get_test_repo("test_snapshot")?;
+
         let timelineid = ZTimelineId::from_str("11223344556677881122334455667788").unwrap();
         let tline = repo.create_empty_timeline(timelineid, Lsn(0))?;
 

pageserver/src/restore_local_repo.rs

@@ -67,13 +67,7 @@ pub fn import_timeline_from_postgres_datadir(
 
             // These special files appear in the snapshot, but are not needed by the page server
             Some("pg_control") => {
-                import_nonrel_file(timeline, lsn, ObjectTag::ControlFile, &direntry.path())?;
-                // Extract checkpoint record from pg_control and store is as separate object
-                let pg_control_bytes =
-                    timeline.get_page_at_lsn_nowait(ObjectTag::ControlFile, lsn, false)?;
-                let pg_control = ControlFileData::decode(&pg_control_bytes)?;
-                let checkpoint_bytes = pg_control.checkPointCopy.encode();
-                timeline.put_page_image(ObjectTag::Checkpoint, lsn, checkpoint_bytes, false)?;
+                import_nonrel_file(timeline, lsn, ObjectTag::ControlFile, &direntry.path())?
             }
             Some("pg_filenode.map") => import_nonrel_file(
                 timeline,
@@ -262,12 +256,7 @@ fn import_slru_file(
         let r = file.read_exact(&mut buf);
         match r {
             Ok(_) => {
-                timeline.put_page_image(
-                    gen_tag(blknum),
-                    lsn,
-                    Bytes::copy_from_slice(&buf),
-                    false,
-                )?;
+                timeline.put_page_image(gen_tag(blknum), lsn, Bytes::copy_from_slice(&buf), false)?;
             }
 
             // TODO: UnexpectedEof is expected
@@ -298,9 +287,16 @@ pub fn import_timeline_wal(walpath: &Path, timeline: &dyn Timeline, startpoint:
     let mut offset = startpoint.segment_offset(pg_constants::WAL_SEGMENT_SIZE);
     let mut last_lsn = startpoint;
 
-    let checkpoint_bytes =
-        timeline.get_page_at_lsn_nowait(ObjectTag::Checkpoint, startpoint, false)?;
+    let checkpoint_bytes = timeline.get_page_at_lsn_nowait(ObjectTag::Checkpoint, startpoint)?;
     let mut checkpoint = CheckPoint::decode(&checkpoint_bytes)?;
+    // get_page_at_lsn_nowait returns pages with zeros when object is not found in the storage.
+    // nextXid can not be zero, so this check is used to detect situation when checkpoint record needs to be initialized.
+    if checkpoint.nextXid.value == 0 {
+        let pg_control_bytes =
+            timeline.get_page_at_lsn_nowait(ObjectTag::ControlFile, startpoint)?;
+        let pg_control = ControlFileData::decode(&pg_control_bytes)?;
+        checkpoint = pg_control.checkPointCopy;
+    }
 
     loop {
         // FIXME: assume postgresql tli 1 for now
@@ -579,7 +575,7 @@ fn save_xlog_dbase_create(timeline: &dyn Timeline, lsn: Lsn, rec: &XlCreateDatab
                 blknum,
             });
 
-            let content = timeline.get_page_at_lsn_nowait(src_key, req_lsn, false)?;
+            let content = timeline.get_page_at_lsn_nowait(src_key, req_lsn)?;
 
             debug!("copying block {:?} to {:?}", src_key, dst_key);
 
@@ -599,7 +595,7 @@ fn save_xlog_dbase_create(timeline: &dyn Timeline, lsn: Lsn, rec: &XlCreateDatab
         match tag {
             ObjectTag::FileNodeMap(db) => {
                 if db.spcnode == src_tablespace_id && db.dbnode == src_db_id {
-                    let img = timeline.get_page_at_lsn_nowait(tag, req_lsn, false)?;
+                    let img = timeline.get_page_at_lsn_nowait(tag, req_lsn)?;
                     let new_tag = ObjectTag::FileNodeMap(DatabaseTag {
                         spcnode: tablespace_id,
                         dbnode: db_id,

pageserver/src/waldecoder.rs

@@ -428,7 +428,7 @@ impl XlXactParsedRecord {
     pub fn decode(buf: &mut Bytes, mut xid: TransactionId, xl_info: u8) -> XlXactParsedRecord {
         let info = xl_info & pg_constants::XLOG_XACT_OPMASK;
         // The record starts with time of commit/abort
-        let xact_time = buf.get_i64_le();
+        let xact_time = buf.get_u64_le();
         let xinfo;
         if xl_info & pg_constants::XLOG_XACT_HAS_INFO != 0 {
             xinfo = buf.get_u32_le();

pageserver/src/walreceiver.rs

@@ -169,8 +169,7 @@ fn walreceiver_main(
 
     let mut waldecoder = WalStreamDecoder::new(startpoint);
 
-    let checkpoint_bytes =
-        timeline.get_page_at_lsn_nowait(ObjectTag::Checkpoint, startpoint, false)?;
+    let checkpoint_bytes = timeline.get_page_at_lsn_nowait(ObjectTag::Checkpoint, startpoint)?;
     let mut checkpoint = CheckPoint::decode(&checkpoint_bytes)?;
     trace!("CheckPoint.nextXid = {}", checkpoint.nextXid.value);
 
@@ -226,11 +225,10 @@ fn walreceiver_main(
                 // Somewhat arbitrarily, if we have at least 10 complete wal segments (16 MB each),
                 // "checkpoint" the repository to flush all the changes from WAL we've processed
                 // so far to disk. After this, we don't need the original WAL anymore, and it
-                // can be removed. This is probably too aggressive for production, but it's useful
-                // to expose bugs now.
+                // can be removed.
                 //
                 // TODO: We don't actually dare to remove the WAL. It's useful for debugging,
-                // and we might it for logical decoding other things in the future. Although
+                // and we might it for logical decoiding other things in the future. Although
                 // we should also be able to fetch it back from the WAL safekeepers or S3 if
                 // needed.
                 if prev_last_rec_lsn.segment_number(pg_constants::WAL_SEGMENT_SIZE)

pageserver/src/walredo.rs

@@ -26,7 +26,6 @@ use std::io::Error;
 use std::path::PathBuf;
 use std::process::Stdio;
 use std::sync::mpsc;
-use std::sync::Arc;
 use std::sync::Mutex;
 use std::time::Duration;
 use std::time::Instant;
@@ -104,16 +103,13 @@ struct PostgresRedoManagerInternal {
 }
 
 #[derive(Debug)]
-struct WalRedoRequestData {
+struct WalRedoRequest {
     tag: ObjectTag,
     lsn: Lsn,
+
     base_img: Option<Bytes>,
     records: Vec<WALRecord>,
-}
 
-#[derive(Debug)]
-struct WalRedoRequest {
-    data: WalRedoRequestData,
     response_channel: mpsc::Sender<Result<Bytes, WalRedoError>>,
 }
 
@@ -179,12 +175,10 @@ impl WalRedoManager for PostgresRedoManager {
         let (tx, rx) = mpsc::channel::<Result<Bytes, WalRedoError>>();
 
         let request = WalRedoRequest {
-            data: WalRedoRequestData {
-                tag,
-                lsn,
-                base_img,
-                records,
-            },
+            tag,
+            lsn,
+            base_img,
+            records,
             response_channel: tx,
         };
 
@@ -235,64 +229,40 @@ impl PostgresRedoManagerInternal {
             .build()
             .unwrap();
 
-        let processes: Vec<PostgresRedoProcess>;
+        let process: PostgresRedoProcess;
 
         info!("launching WAL redo postgres process");
 
-        let wal_redoers = self.conf.wal_redoers;
-        processes = (0..wal_redoers)
-            .map(|i| {
-                runtime
-                    .block_on(PostgresRedoProcess::launch(self.conf, i))
-                    .unwrap()
-            })
-            .collect();
+        process = runtime
+            .block_on(PostgresRedoProcess::launch(self.conf))
+            .unwrap();
 
         // Loop forever, handling requests as they come.
         loop {
-            let mut requests: Vec<WalRedoRequest> = Vec::new();
-            requests.push(
-                self.request_rx
-                    .recv()
-                    .expect("WAL redo request channel was closed"),
-            );
-            loop {
-                let req = self.request_rx.try_recv();
-                match req {
-                    Ok(req) => requests.push(req),
-                    Err(_) => break,
-                }
-            }
-            let request_data = requests.iter().map(|req| &req.data);
-            let mut rr = 0; // round robin
-            let results = runtime.block_on(async {
-                let futures = request_data.map(|req| {
-                    rr += 1;
-                    self.handle_apply_request(&processes[rr % wal_redoers], &req)
-                });
-                let mut results: Vec<Result<Bytes, WalRedoError>> = Vec::new();
-                for future in futures {
-                    results.push(future.await);
-                }
-                results
-            });
-            for (result, request) in results.into_iter().zip(requests.iter()) {
-                let result_ok = result.is_ok();
+            let request = self
+                .request_rx
+                .recv()
+                .expect("WAL redo request channel was closed");
 
-                // Send the result to the requester
-                let _ = request.response_channel.send(result);
+            let result = runtime.block_on(self.handle_apply_request(&process, &request));
+            let result_ok = result.is_ok();
 
-                if !result_ok {
-                    error!("wal-redo-postgres failed to apply request {:?}", request);
-                }
+            // Send the result to the requester
+            let _ = request.response_channel.send(result);
+
+            if !result_ok {
+                error!("wal-redo-postgres failed to apply request {:?}", request);
             }
         }
     }
 
+    ///
+    /// Process one request for WAL redo.
+    ///
     async fn handle_apply_request(
         &self,
         process: &PostgresRedoProcess,
-        request: &WalRedoRequestData,
+        request: &WalRedoRequest,
     ) -> Result<Bytes, WalRedoError> {
         let tag = request.tag;
         let lsn = request.lsn;
@@ -308,17 +278,17 @@ impl PostgresRedoManagerInternal {
             // Relational WAL records are applied using wal-redo-postgres
             apply_result = process.apply_wal_records(buf_tag, base_img, records).await;
         } else {
-            // Non-relational WAL records we apply ourselves.
+            // Non-relational WAL records we will aply ourselves.
             const ZERO_PAGE: [u8; 8192] = [0u8; 8192];
             let mut page = BytesMut::new();
             if let Some(fpi) = base_img {
                 // If full-page image is provided, then use it...
                 page.extend_from_slice(&fpi[..]);
             } else {
-                // otherwise initialize page with zeros
+                //  otherwise initialize page with zeros
                 page.extend_from_slice(&ZERO_PAGE);
             }
-            // Apply all collected WAL records
+            // Apply all callected WAL records
             for record in records {
                 let mut buf = record.rec.clone();
 
@@ -476,19 +446,19 @@ impl PostgresRedoManagerInternal {
 }
 
 struct PostgresRedoProcess {
-    stdin: Arc<RefCell<ChildStdin>>,
-    stdout: Arc<RefCell<ChildStdout>>,
+    stdin: RefCell<ChildStdin>,
+    stdout: RefCell<ChildStdout>,
 }
 
 impl PostgresRedoProcess {
     //
     // Start postgres binary in special WAL redo mode.
     //
-    async fn launch(conf: &PageServerConf, id: usize) -> Result<PostgresRedoProcess, Error> {
+    async fn launch(conf: &PageServerConf) -> Result<PostgresRedoProcess, Error> {
         // FIXME: We need a dummy Postgres cluster to run the process in. Currently, we
         // just create one with constant name. That fails if you try to launch more than
         // one WAL redo manager concurrently.
-        let datadir = conf.workdir.join(format!("wal-redo-datadir-{}", id));
+        let datadir = conf.workdir.join("wal-redo-datadir");
 
         // Create empty data directory for wal-redo postgres, deleting old one first.
         if datadir.exists() {
@@ -568,8 +538,8 @@ impl PostgresRedoProcess {
         tokio::spawn(f_stderr);
 
         Ok(PostgresRedoProcess {
-            stdin: Arc::new(RefCell::new(stdin)),
-            stdout: Arc::new(RefCell::new(stdout)),
+            stdin: RefCell::new(stdin),
+            stdout: RefCell::new(stdout),
         })
     }
 

postgres_ffi/src/xlog_utils.rs

@@ -41,7 +41,7 @@ pub const SIZE_OF_XLOG_RECORD_DATA_HEADER_SHORT: usize = 1 * 2;
 
 pub type XLogRecPtr = u64;
 pub type TimeLineID = u32;
-pub type TimestampTz = i64;
+pub type TimestampTz = u64;
 pub type XLogSegNo = u64;
 
 const XID_CHECKPOINT_INTERVAL: u32 = 1024;
@@ -95,9 +95,9 @@ pub fn get_current_timestamp() -> TimestampTz {
     const USECS_PER_SEC: u64 = 1000000;
     match SystemTime::now().duration_since(SystemTime::UNIX_EPOCH) {
         Ok(n) => {
-            ((n.as_secs() - ((POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * SECS_PER_DAY))
+            (n.as_secs() - ((POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * SECS_PER_DAY))
                 * USECS_PER_SEC
-                + n.subsec_micros() as u64) as i64
+                + n.subsec_micros() as u64
         }
         Err(_) => panic!("SystemTime before UNIX EPOCH!"),
     }
@@ -447,7 +447,7 @@ pub fn generate_wal_segment(pg_control: &ControlFileData) -> Bytes {
     seg_buf.extend_from_slice(&rec_shord_hdr_bytes);
     seg_buf.extend_from_slice(&checkpoint_bytes);
 
-    //zero out the rest of the file
+    //zero out remainig file
     seg_buf.resize(pg_constants::WAL_SEGMENT_SIZE, 0);
     seg_buf.freeze()
 }

proxy/Cargo.toml

@@ -1,21 +0,0 @@
-[package]
-name = "proxy"
-version = "0.1.0"
-authors = ["Stas Kelvich <stas.kelvich@gmail.com>"]
-edition = "2018"
-
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
-[dependencies]
-anyhow = "1.0"
-bytes = { version = "1.0.1", features = ['serde'] }
-md5 = "0.7.0"
-rand = "0.8.3"
-hex = "0.4.3"
-serde = "1"
-serde_json = "1"
-tokio = { version = "1.7.1", features = ["full"] }
-tokio-postgres = "0.7.2"
-clap = "2.33.0"
-
-zenith_utils = { path = "../zenith_utils" }

proxy/src/cplane_api.rs

@@ -1,92 +0,0 @@
-use anyhow::{bail, Result};
-use serde::{Deserialize, Serialize};
-use std::{
-    collections::HashMap,
-    net::{IpAddr, SocketAddr},
-};
-
-pub struct CPlaneApi {
-    // address: SocketAddr,
-}
-
-#[derive(Serialize, Deserialize)]
-pub struct DatabaseInfo {
-    pub host: IpAddr, // TODO: allow host name here too
-    pub port: u16,
-    pub dbname: String,
-    pub user: String,
-    pub password: String,
-}
-
-impl DatabaseInfo {
-    pub fn socket_addr(&self) -> SocketAddr {
-        SocketAddr::new(self.host, self.port)
-    }
-
-    pub fn conn_string(&self) -> String {
-        format!(
-            "dbname={} user={} password={}",
-            self.dbname, self.user, self.password
-        )
-    }
-}
-
-// mock cplane api
-impl CPlaneApi {
-    pub fn new(_address: &SocketAddr) -> CPlaneApi {
-        CPlaneApi {
-            // address: address.clone(),
-        }
-    }
-
-    pub fn check_auth(&self, user: &str, md5_response: &[u8], salt: &[u8; 4]) -> Result<()> {
-        // passwords for both is "mypass"
-        let auth_map: HashMap<_, &str> = vec![
-            ("stas@zenith", "716ee6e1c4a9364d66285452c47402b1"),
-            ("stas2@zenith", "3996f75df64c16a8bfaf01301b61d582"),
-        ]
-        .into_iter()
-        .collect();
-
-        let stored_hash = auth_map
-            .get(&user)
-            .ok_or_else(|| anyhow::Error::msg("user not found"))?;
-        let salted_stored_hash = format!(
-            "md5{:x}",
-            md5::compute([stored_hash.as_bytes(), salt].concat())
-        );
-
-        let received_hash = std::str::from_utf8(&md5_response)?;
-
-        println!(
-            "auth: {} rh={} sh={} ssh={} {:?}",
-            user, received_hash, stored_hash, salted_stored_hash, salt
-        );
-
-        if received_hash == salted_stored_hash {
-            Ok(())
-        } else {
-            bail!("Auth failed")
-        }
-    }
-
-    pub fn get_database_uri(&self, _user: &str, _database: &str) -> Result<DatabaseInfo> {
-        Ok(DatabaseInfo {
-            host: "127.0.0.1".parse()?,
-            port: 5432,
-            dbname: "stas".to_string(),
-            user: "stas".to_string(),
-            password: "mypass".to_string(),
-        })
-    }
-
-    // pub fn create_database(&self, _user: &String, _database: &String) -> Result<DatabaseInfo> {
-    //     Ok(DatabaseInfo {
-    //         host: "127.0.0.1".parse()?,
-    //         port: 5432,
-    //         dbname: "stas".to_string(),
-    //         user: "stas".to_string(),
-    //         password: "mypass".to_string(),
-    //     })
-    // }
-}

proxy/src/main.rs

@@ -1,106 +0,0 @@
-///
-/// Postgres protocol proxy/router.
-///
-/// This service listens psql port and can check auth via external service
-/// (control plane API in our case) and can create new databases and accounts
-/// in somewhat transparent manner (again via communication with control plane API).
-///
-use std::{
-    collections::HashMap,
-    net::{SocketAddr, TcpListener},
-    sync::{mpsc, Mutex},
-    thread,
-};
-
-use clap::{App, Arg};
-
-use cplane_api::DatabaseInfo;
-
-mod cplane_api;
-mod mgmt;
-mod proxy;
-
-pub struct ProxyConf {
-    /// main entrypoint for users to connect to
-    pub proxy_address: SocketAddr,
-
-    /// http management endpoint. Upon user account creation control plane
-    /// will notify us here, so that we can 'unfreeze' user session.
-    pub mgmt_address: SocketAddr,
-
-    /// send unauthenticated users to this URI
-    pub redirect_uri: String,
-
-    /// control plane address where we would check auth.
-    pub cplane_address: SocketAddr,
-}
-
-pub struct ProxyState {
-    pub conf: ProxyConf,
-    pub waiters: Mutex<HashMap<String, mpsc::Sender<anyhow::Result<DatabaseInfo>>>>,
-}
-
-fn main() -> anyhow::Result<()> {
-    let arg_matches = App::new("Zenith proxy/router")
-        .arg(
-            Arg::with_name("proxy")
-                .short("p")
-                .long("proxy")
-                .takes_value(true)
-                .help("listen for incoming client connections on ip:port")
-                .default_value("127.0.0.1:4432"),
-        )
-        .arg(
-            Arg::with_name("mgmt")
-                .short("m")
-                .long("mgmt")
-                .takes_value(true)
-                .help("listen for management callback connection on ip:port")
-                .default_value("127.0.0.1:7000"),
-        )
-        .arg(
-            Arg::with_name("uri")
-                .short("u")
-                .long("uri")
-                .takes_value(true)
-                .help("redirect unauthenticated users to given uri")
-                .default_value("http://localhost:3000/psql_session/"),
-        )
-        .get_matches();
-
-    let conf = ProxyConf {
-        proxy_address: arg_matches.value_of("proxy").unwrap().parse()?,
-        mgmt_address: arg_matches.value_of("mgmt").unwrap().parse()?,
-        redirect_uri: arg_matches.value_of("uri").unwrap().parse()?,
-        cplane_address: "127.0.0.1:3000".parse()?,
-    };
-    let state = ProxyState {
-        conf,
-        waiters: Mutex::new(HashMap::new()),
-    };
-    let state: &'static ProxyState = Box::leak(Box::new(state));
-
-    // Check that we can bind to address before further initialization
-    println!("Starting proxy on {}", state.conf.proxy_address);
-    let pageserver_listener = TcpListener::bind(state.conf.proxy_address)?;
-
-    println!("Starting mgmt on {}", state.conf.mgmt_address);
-    let mgmt_listener = TcpListener::bind(state.conf.mgmt_address)?;
-
-    let threads = vec![
-        // Spawn a thread to listen for connections. It will spawn further threads
-        // for each connection.
-        thread::Builder::new()
-            .name("Proxy thread".into())
-            .spawn(move || proxy::thread_main(&state, pageserver_listener))?,
-        thread::Builder::new()
-            .name("Mgmt thread".into())
-            .spawn(move || mgmt::thread_main(&state, mgmt_listener))?,
-    ];
-
-    for t in threads.into_iter() {
-        t.join().unwrap()?;
-    }
-
-    Ok(())
-}

proxy/src/mgmt.rs

@@ -1,111 +0,0 @@
-use std::{
-    net::{TcpListener, TcpStream},
-    thread,
-};
-
-use anyhow::bail;
-use bytes::Bytes;
-use serde::{Deserialize, Serialize};
-use zenith_utils::{
-    postgres_backend::{self, query_from_cstring, PostgresBackend},
-    pq_proto::{BeMessage, SINGLE_COL_ROWDESC},
-};
-
-use crate::{cplane_api::DatabaseInfo, ProxyState};
-
-///
-/// Main proxy listener loop.
-///
-/// Listens for connections, and launches a new handler thread for each.
-///
-pub fn thread_main(state: &'static ProxyState, listener: TcpListener) -> anyhow::Result<()> {
-    loop {
-        let (socket, peer_addr) = listener.accept()?;
-        println!("accepted connection from {}", peer_addr);
-        socket.set_nodelay(true).unwrap();
-
-        thread::spawn(move || {
-            if let Err(err) = mgmt_conn_main(state, socket) {
-                println!("error: {}", err);
-            }
-        });
-    }
-}
-
-pub fn mgmt_conn_main(state: &'static ProxyState, socket: TcpStream) -> anyhow::Result<()> {
-    let mut conn_handler = MgmtHandler { state };
-    let mut pgbackend = PostgresBackend::new(socket, postgres_backend::AuthType::Trust)?;
-    pgbackend.run(&mut conn_handler)
-}
-
-struct MgmtHandler {
-    state: &'static ProxyState,
-}
-/// Serialized examples:
-// {
-//     "session_id": "71d6d03e6d93d99a",
-//     "result": {
-//         "Success": {
-//             "host": "127.0.0.1",
-//             "port": 5432,
-//             "dbname": "stas",
-//             "user": "stas"
-//             "password": "mypass"
-//         }
-//     }
-// }
-// {
-//     "session_id": "71d6d03e6d93d99a",
-//     "result": {
-//         "Failure": "oops"
-//     }
-// }
-#[derive(Serialize, Deserialize)]
-pub struct PsqlSessionResponse {
-    session_id: String,
-    result: PsqlSessionResult,
-}
-
-#[derive(Serialize, Deserialize)]
-pub enum PsqlSessionResult {
-    Success(DatabaseInfo),
-    Failure(String),
-}
-
-impl postgres_backend::Handler for MgmtHandler {
-    fn process_query(
-        &mut self,
-        pgb: &mut PostgresBackend,
-        query_string: Bytes,
-    ) -> anyhow::Result<()> {
-        let query_string = query_from_cstring(query_string);
-
-        println!("Got mgmt query: '{}'", std::str::from_utf8(&query_string)?);
-
-        let resp: PsqlSessionResponse = serde_json::from_slice(&query_string)?;
-
-        let waiters = self.state.waiters.lock().unwrap();
-
-        let sender = waiters
-            .get(&resp.session_id)
-            .ok_or_else(|| anyhow::Error::msg("psql_session_id is not found"))?;
-
-        match resp.result {
-            PsqlSessionResult::Success(db_info) => {
-                sender.send(Ok(db_info))?;
-
-                pgb.write_message_noflush(&SINGLE_COL_ROWDESC)?
-                    .write_message_noflush(&BeMessage::DataRow(&[Some(b"ok")]))?
-                    .write_message_noflush(&BeMessage::CommandComplete(b"SELECT 1"))?;
-                pgb.flush()?;
-                Ok(())
-            }
-
-            PsqlSessionResult::Failure(message) => {
-                sender.send(Err(anyhow::Error::msg(message.clone())))?;
-
-                bail!("psql session request failed: {}", message)
-            }
-        }
-    }
-}

proxy/src/proxy.rs

@@ -1,256 +0,0 @@
-use crate::cplane_api::CPlaneApi;
-use crate::cplane_api::DatabaseInfo;
-use crate::ProxyState;
-
-use anyhow::bail;
-use tokio_postgres::NoTls;
-
-use rand::Rng;
-use std::sync::mpsc::channel;
-use std::thread;
-use tokio::io::AsyncWriteExt;
-use zenith_utils::postgres_backend::{PostgresBackend, ProtoState};
-use zenith_utils::pq_proto::*;
-use zenith_utils::{postgres_backend, pq_proto::BeMessage};
-
-///
-/// Main proxy listener loop.
-///
-/// Listens for connections, and launches a new handler thread for each.
-///
-pub fn thread_main(
-    state: &'static ProxyState,
-    listener: std::net::TcpListener,
-) -> anyhow::Result<()> {
-    loop {
-        let (socket, peer_addr) = listener.accept()?;
-        println!("accepted connection from {}", peer_addr);
-        socket.set_nodelay(true).unwrap();
-
-        thread::spawn(move || {
-            if let Err(err) = proxy_conn_main(state, socket) {
-                println!("error: {}", err);
-            }
-        });
-    }
-}
-
-// XXX: clean up fields
-struct ProxyConnection {
-    state: &'static ProxyState,
-
-    cplane: CPlaneApi,
-
-    user: String,
-    database: String,
-
-    pgb: PostgresBackend,
-    md5_salt: [u8; 4],
-
-    psql_session_id: String,
-}
-
-pub fn proxy_conn_main(
-    state: &'static ProxyState,
-    socket: std::net::TcpStream,
-) -> anyhow::Result<()> {
-    let mut conn = ProxyConnection {
-        state,
-        cplane: CPlaneApi::new(&state.conf.cplane_address),
-        user: "".into(),
-        database: "".into(),
-        pgb: PostgresBackend::new(socket, postgres_backend::AuthType::MD5)?,
-        md5_salt: [0u8; 4],
-        psql_session_id: "".into(),
-    };
-
-    // Check StartupMessage
-    // This will set conn.existing_user and we can decide on next actions
-    conn.handle_startup()?;
-
-    // both scenarious here should end up producing database connection string
-    let db_info = if conn.is_existing_user() {
-        conn.handle_existing_user()?
-    } else {
-        conn.handle_new_user()?
-    };
-
-    // ok, proxy pass user connection to database_uri
-    let runtime = tokio::runtime::Builder::new_current_thread()
-        .enable_all()
-        .build()
-        .unwrap();
-
-    let _ = runtime.block_on(proxy_pass(conn.pgb, db_info))?;
-
-    println!("proxy_conn_main done;");
-
-    Ok(())
-}
-
-impl ProxyConnection {
-    fn is_existing_user(&self) -> bool {
-        self.user.ends_with("@zenith")
-    }
-
-    fn handle_startup(&mut self) -> anyhow::Result<()> {
-        loop {
-            let msg = self.pgb.read_message()?;
-            println!("got message {:?}", msg);
-            match msg {
-                Some(FeMessage::StartupMessage(m)) => {
-                    println!("got startup message {:?}", m);
-
-                    match m.kind {
-                        StartupRequestCode::NegotiateGss | StartupRequestCode::NegotiateSsl => {
-                            println!("SSL requested");
-                            self.pgb.write_message(&BeMessage::Negotiate)?;
-                        }
-                        StartupRequestCode::Normal => {
-                            self.user = m
-                                .params
-                                .get("user")
-                                .ok_or_else(|| {
-                                    anyhow::Error::msg("user is required in startup packet")
-                                })?
-                                .into();
-                            self.database = m
-                                .params
-                                .get("database")
-                                .ok_or_else(|| {
-                                    anyhow::Error::msg("database is required in startup packet")
-                                })?
-                                .into();
-
-                            break;
-                        }
-                        StartupRequestCode::Cancel => break,
-                    }
-                }
-                None => {
-                    bail!("connection closed")
-                }
-                unexpected => {
-                    bail!("unexpected message type : {:?}", unexpected)
-                }
-            }
-        }
-        Ok(())
-    }
-
-    fn handle_existing_user(&mut self) -> anyhow::Result<DatabaseInfo> {
-        // ask password
-        rand::thread_rng().fill(&mut self.md5_salt);
-        self.pgb
-            .write_message(&BeMessage::AuthenticationMD5Password(&self.md5_salt))?;
-        self.pgb.state = ProtoState::Authentication; // XXX
-
-        // check password
-        println!("handle_existing_user");
-        let msg = self.pgb.read_message()?;
-        println!("got message {:?}", msg);
-        if let Some(FeMessage::PasswordMessage(m)) = msg {
-            println!("got password message '{:?}'", m);
-
-            assert!(self.is_existing_user());
-
-            let (_trailing_null, md5_response) = m
-                .split_last()
-                .ok_or_else(|| anyhow::Error::msg("unexpected password message"))?;
-
-            if let Err(e) = self.check_auth_md5(md5_response) {
-                self.pgb
-                    .write_message(&BeMessage::ErrorResponse(format!("{}", e)))?;
-                bail!("auth failed: {}", e);
-            } else {
-                self.pgb
-                    .write_message_noflush(&BeMessage::AuthenticationOk)?;
-                self.pgb
-                    .write_message_noflush(&BeMessage::ParameterStatus)?;
-                self.pgb.write_message(&BeMessage::ReadyForQuery)?;
-            }
-        }
-
-        // ok, we are authorized
-        self.cplane.get_database_uri(&self.user, &self.database)
-    }
-
-    fn handle_new_user(&mut self) -> anyhow::Result<DatabaseInfo> {
-        let mut psql_session_id_buf = [0u8; 8];
-        rand::thread_rng().fill(&mut psql_session_id_buf);
-        self.psql_session_id = hex::encode(psql_session_id_buf);
-
-        let hello_message = format!("☀️  Welcome to Zenith!
-
-To proceed with database creation open following link:
-
-    {}{}
-
-It needed to be done once and we will send you '.pgpass' file which will allow you to access or create
-databases without opening the browser.
-
-", self.state.conf.redirect_uri,self.psql_session_id);
-
-        self.pgb
-            .write_message_noflush(&BeMessage::AuthenticationOk)?;
-        self.pgb
-            .write_message_noflush(&BeMessage::ParameterStatus)?;
-        self.pgb
-            .write_message(&BeMessage::NoticeResponse(hello_message))?;
-
-        // await for database creation
-        let (tx, rx) = channel::<anyhow::Result<DatabaseInfo>>();
-        let _ = self
-            .state
-            .waiters
-            .lock()
-            .unwrap()
-            .insert(self.psql_session_id.clone(), tx);
-
-        // Wait for web console response
-        // XXX: respond with error to client
-        let dbinfo = rx.recv()??;
-
-        self.pgb.write_message_noflush(&BeMessage::NoticeResponse(
-            "Connecting to database.".to_string(),
-        ))?;
-        self.pgb.write_message(&BeMessage::ReadyForQuery)?;
-
-        Ok(dbinfo)
-    }
-
-    fn check_auth_md5(&self, md5_response: &[u8]) -> anyhow::Result<()> {
-        assert!(self.is_existing_user());
-        self.cplane
-            .check_auth(self.user.as_str(), md5_response, &self.md5_salt)
-    }
-}
-
-async fn proxy_pass(pgb: PostgresBackend, db_info: DatabaseInfo) -> anyhow::Result<()> {
-    let mut socket = tokio::net::TcpStream::connect(db_info.socket_addr()).await?;
-    let config = db_info.conn_string().parse::<tokio_postgres::Config>()?;
-    let _ = config.connect_raw(&mut socket, NoTls).await?;
-
-    println!("Connected to pg, proxying");
-
-    let incoming_std = pgb.into_stream();
-    incoming_std.set_nonblocking(true)?;
-    let mut incoming_conn = tokio::net::TcpStream::from_std(incoming_std)?;
-
-    let (mut ri, mut wi) = incoming_conn.split();
-    let (mut ro, mut wo) = socket.split();
-
-    let client_to_server = async {
-        tokio::io::copy(&mut ri, &mut wo).await?;
-        wo.shutdown().await
-    };
-
-    let server_to_client = async {
-        tokio::io::copy(&mut ro, &mut wi).await?;
-        wi.shutdown().await
-    };
-
-    tokio::try_join!(client_to_server, server_to_client)?;
-
-    Ok(())
-}

test_runner/batch_others/test_bulk_insert.py

@@ -1,25 +0,0 @@
-from contextlib import closing
-import psycopg2.extras
-
-pytest_plugins = ("fixtures.zenith_fixtures")
-
-#
-# Test insertion of larg number of records
-#
-# This test is pretty tightly coupled with the current implementation of page version storage
-# and garbage collection in object_repository.rs.
-#
-def test_bulk_insert(zenith_cli, pageserver, postgres, pg_bin):
-    zenith_cli.run(["branch", "test_bulk_insert", "empty"])
-    pg = postgres.create_start('test_bulk_insert')
-
-    with closing(pg.connect()) as conn:
-        with conn.cursor() as cur:
-          cur.execute("create table t(c1 bigint, c2 bigint, c3 bigint, c4 bigint, c5 bigint)")
-          cur.execute("create index on t(c1)")
-          cur.execute("create index on t(c2)")
-          cur.execute("create index on t(c3)")
-          cur.execute("create index on t(c4)")
-          cur.execute("create index on t(c5)")
-          cur.execute("insert into t values (generate_series(1,1000000),generate_series(1,1000000),generate_series(1,1000000),generate_series(1,1000000),generate_series(1,1000000))")
-          cur.execute("insert into t values (generate_series(1,1000000),random()*1000000,random()*1000000,random()*1000000,random()*1000000)")

test_runner/batch_others/test_seq_scan.py

@@ -1,26 +0,0 @@
-from contextlib import closing
-import psycopg2.extras
-import time
-
-pytest_plugins = ("fixtures.zenith_fixtures")
-
-#
-# Test insertion of larg number of records
-#
-# This test is pretty tightly coupled with the current implementation of page version storage
-# and garbage collection in object_repository.rs.
-#
-def test_seq_scan(zenith_cli, pageserver, postgres, pg_bin):
-    zenith_cli.run(["branch", "test_seq_scan", "empty"])
-    pg = postgres.create_start('test_seq_scan')
-
-    with closing(pg.connect()) as conn:
-        with conn.cursor() as cur:
-          cur.execute("create table t(c1 bigint, c2 bigint, c3 bigint, c4 bigint, c5 bigint)")
-          cur.execute("insert into t values (generate_series(1,1000000),generate_series(1,1000000),generate_series(1,1000000),generate_series(1,1000000),generate_series(1,1000000))")
-          cur.execute("set max_parallel_workers_per_gather=0");
-          for i in range(100):
-              start = time.time()
-              cur.execute("select count(*) from t");
-              stop = time.time()
-              print(f'Elapsed time for iterating through 1000000 records is {stop - start}')

walkeeper/Cargo.toml

@@ -27,7 +27,7 @@ postgres-protocol = { git = "https://github.com/zenithdb/rust-postgres.git", rev
 postgres = { git = "https://github.com/zenithdb/rust-postgres.git", rev="9eb0dbfbeb6a6c1b79099b9f7ae4a8c021877858" }
 anyhow = "1.0"
 crc32c = "0.6.0"
-humantime = "2.1.0"
+parse_duration = "2.1.1"
 walkdir = "2"
 serde = { version = "1.0", features = ["derive"] }
 hex = "0.4.3"

walkeeper/src/bin/wal_acceptor.rs

@@ -5,10 +5,12 @@ use anyhow::{Context, Result};
 use clap::{App, Arg};
 use daemonize::Daemonize;
 use log::*;
+use parse_duration::parse;
 use slog::Drain;
 use std::io;
 use std::path::{Path, PathBuf};
 use std::thread;
+use std::time::Duration;
 use std::{fs::File, fs::OpenOptions};
 
 use walkeeper::s3_offload;
@@ -100,11 +102,11 @@ fn main() -> Result<()> {
     }
 
     if let Some(ttl) = arg_matches.value_of("ttl") {
-        conf.ttl = Some(humantime::parse_duration(ttl)?);
+        conf.ttl = Some::<Duration>(parse(ttl)?);
     }
 
     if let Some(recall) = arg_matches.value_of("recall") {
-        conf.recall_period = Some(humantime::parse_duration(recall)?);
+        conf.recall_period = Some::<Duration>(parse(recall)?);
     }
 
     start_wal_acceptor(conf)

walkeeper/src/receive_wal.rs

@@ -6,6 +6,7 @@ use anyhow::{bail, Result};
 use log::*;
 use postgres::{Client, Config, NoTls};
 use serde::{Deserialize, Serialize};
+use zenith_utils::connstring::connection_host_port;
 use std::cmp::{max, min};
 use std::fs::{self, File, OpenOptions};
 use std::io::{BufReader, Read, Seek, SeekFrom, Write};
@@ -15,7 +16,6 @@ use std::sync::Arc;
 use std::thread;
 use std::thread::sleep;
 use zenith_utils::bin_ser::LeSer;
-use zenith_utils::connstring::connection_host_port;
 use zenith_utils::lsn::Lsn;
 
 use crate::replication::HotStandbyFeedback;
@@ -159,7 +159,10 @@ fn request_callback(conf: WalAcceptorConf, timelineid: ZTimelineId) {
     let (host, port) = connection_host_port(&me_conf);
     let callme = format!(
         "callmemaybe {} host={} port={} options='-c ztimelineid={}'",
-        timelineid, host, port, timelineid
+        timelineid,
+        host,
+        port,
+        timelineid
     );
     loop {
         info!(

walkeeper/src/wal_service.rs

@@ -11,7 +11,7 @@ use std::thread;
 use crate::receive_wal::ReceiveWalConn;
 use crate::send_wal::SendWalHandler;
 use crate::WalAcceptorConf;
-use zenith_utils::postgres_backend::{AuthType, PostgresBackend};
+use zenith_utils::postgres_backend::PostgresBackend;
 
 /// Accept incoming TCP connections and spawn them into a background thread.
 pub fn thread_main(conf: WalAcceptorConf) -> Result<()> {
@@ -50,7 +50,7 @@ fn handle_socket(mut socket: TcpStream, conf: WalAcceptorConf) -> Result<()> {
         ReceiveWalConn::new(socket, conf)?.run()?; // internal protocol between wal_proposer and wal_acceptor
     } else {
         let mut conn_handler = SendWalHandler::new(conf);
-        let mut pgbackend = PostgresBackend::new(socket, AuthType::Trust)?;
+        let mut pgbackend = PostgresBackend::new(socket)?;
         // libpq replication protocol between wal_acceptor and replicas/pagers
         pgbackend.run(&mut conn_handler)?;
     }

zenith/src/main.rs

@@ -102,7 +102,7 @@ fn main() -> Result<()> {
     // Create config file
     if let ("init", Some(sub_args)) = matches.subcommand() {
         let pageserver_uri = sub_args.value_of("pageserver-url");
-        local_env::init(pageserver_uri).with_context(|| "Failed to create config file")?;
+        local_env::init(pageserver_uri).with_context(|| "Failed to create cofig file")?;
     }
 
     // all other commands would need config

zenith_utils/Cargo.toml

@@ -14,7 +14,6 @@ bincode = "1.3"
 thiserror = "1.0"
 postgres = { git = "https://github.com/zenithdb/rust-postgres.git", rev="9eb0dbfbeb6a6c1b79099b9f7ae4a8c021877858" }
 workspace_hack = { path = "../workspace_hack" }
-rand = "0.8.3"
 
 [dev-dependencies]
 hex-literal = "0.3"

zenith_utils/src/connstring.rs

@@ -1,16 +1,8 @@
 use postgres::Config;
 
 pub fn connection_host_port(config: &Config) -> (String, u16) {
-    assert_eq!(
-        config.get_hosts().len(),
-        1,
-        "only one pair of host and port is supported in connection string"
-    );
-    assert_eq!(
-        config.get_ports().len(),
-        1,
-        "only one pair of host and port is supported in connection string"
-    );
+    assert_eq!(config.get_hosts().len(), 1, "only one pair of host and port is supported in connection string");
+    assert_eq!(config.get_ports().len(), 1, "only one pair of host and port is supported in connection string");
     let host = match &config.get_hosts()[0] {
         postgres::config::Host::Tcp(host) => host.as_ref(),
         postgres::config::Host::Unix(host) => host.to_str().unwrap(),
@@ -29,24 +21,14 @@ mod tests {
 
     #[test]
     fn test_connection_host_port() {
-        let config: Config = "postgresql://no_user@localhost:64000/no_db"
-            .parse()
-            .unwrap();
-        assert_eq!(
-            connection_host_port(&config),
-            ("localhost".to_owned(), 64000)
-        );
+        let config: Config = "postgresql://no_user@localhost:64000/no_db".parse().unwrap();
+        assert_eq!(connection_host_port(&config), ("localhost".to_owned(), 64000));
     }
 
     #[test]
     #[should_panic(expected = "only one pair of host and port is supported in connection string")]
     fn test_connection_host_port_multiple_ports() {
-        let config: Config = "postgresql://no_user@localhost:64000,localhost:64001/no_db"
-            .parse()
-            .unwrap();
-        assert_eq!(
-            connection_host_port(&config),
-            ("localhost".to_owned(), 64000)
-        );
+        let config: Config = "postgresql://no_user@localhost:64000,localhost:64001/no_db".parse().unwrap();
+        assert_eq!(connection_host_port(&config), ("localhost".to_owned(), 64000));
     }
 }

zenith_utils/src/postgres_backend.rs

@@ -8,11 +8,9 @@ use anyhow::bail;
 use anyhow::Result;
 use bytes::{Bytes, BytesMut};
 use log::*;
-use rand::Rng;
 use std::io;
 use std::io::{BufReader, Write};
-use std::net::Shutdown;
-use std::net::TcpStream;
+use std::net::{Shutdown, TcpStream};
 
 pub trait Handler {
     /// Handle single query.
@@ -20,33 +18,10 @@ pub trait Handler {
     /// might be not what we want after CopyData streaming, but currently we don't
     /// care).
     fn process_query(&mut self, pgb: &mut PostgresBackend, query_string: Bytes) -> Result<()>;
-
     /// Called on startup packet receival, allows to process params.
-    ///
-    /// If Ok(false) is returned postgres_backend will skip auth -- that is needed for new users
-    /// creation is the proxy code. That is quite hacky and ad-hoc solution, may be we could allow
-    /// to override whole init logic in implementations.
     fn startup(&mut self, _pgb: &mut PostgresBackend, _sm: &FeStartupMessage) -> Result<()> {
         Ok(())
     }
-
-    /// Check auth
-    fn check_auth_md5(&mut self, _pgb: &mut PostgresBackend, _md5_response: &[u8]) -> Result<()> {
-        bail!("Auth failed")
-    }
-}
-
-#[derive(PartialEq)]
-pub enum ProtoState {
-    Initialization,
-    Authentication,
-    Established,
-}
-
-#[derive(PartialEq)]
-pub enum AuthType {
-    Trust,
-    MD5,
 }
 
 pub struct PostgresBackend {
@@ -57,32 +32,25 @@ pub struct PostgresBackend {
     stream_out: TcpStream,
     // Output buffer. c.f. BeMessage::write why we are using BytesMut here.
     buf_out: BytesMut,
-
-    pub state: ProtoState,
-
-    md5_salt: [u8; 4],
-    auth_type: AuthType,
+    init_done: bool,
 }
 
-pub fn query_from_cstring(query_string: Bytes) -> Vec<u8> {
-    let mut query_string = query_string.to_vec();
-    if let Some(ch) = query_string.last() {
-        if *ch == 0 {
-            query_string.pop();
-        }
+// In replication.rs a separate thread is reading keepalives from the
+// socket. When main one finishes, tell it to get down by shutdowning the
+// socket.
+impl Drop for PostgresBackend {
+    fn drop(&mut self) {
+        let _res = self.stream_out.shutdown(Shutdown::Both);
     }
-    query_string
 }
 
 impl PostgresBackend {
-    pub fn new(socket: TcpStream, auth_type: AuthType) -> Result<Self, std::io::Error> {
+    pub fn new(socket: TcpStream) -> Result<Self, std::io::Error> {
         let mut pb = PostgresBackend {
             stream_in: None,
             stream_out: socket,
             buf_out: BytesMut::with_capacity(10 * 1024),
-            state: ProtoState::Initialization,
-            md5_salt: [0u8; 4],
-            auth_type,
+            init_done: false,
         };
         // if socket cloning fails, report the error and bail out
         pb.stream_in = match pb.stream_out.try_clone() {
@@ -96,10 +64,6 @@ impl PostgresBackend {
         Ok(pb)
     }
 
-    pub fn into_stream(self) -> TcpStream {
-        self.stream_out
-    }
-
     /// Get direct reference (into the Option) to the read stream.
     fn get_stream_in(&mut self) -> Result<&mut BufReader<TcpStream>> {
         match self.stream_in {
@@ -114,11 +78,10 @@ impl PostgresBackend {
 
     /// Read full message or return None if connection is closed.
     pub fn read_message(&mut self) -> Result<Option<FeMessage>> {
-        match self.state {
-            ProtoState::Initialization => FeStartupMessage::read(self.get_stream_in()?),
-            ProtoState::Authentication | ProtoState::Established => {
-                FeMessage::read(self.get_stream_in()?)
-            }
+        if !self.init_done {
+            FeStartupMessage::read(self.get_stream_in()?)
+        } else {
+            FeMessage::read(self.get_stream_in()?)
         }
     }
 
@@ -141,14 +104,7 @@ impl PostgresBackend {
         self.flush()
     }
 
-    // wrapper for run_internal() that shuts down socket when we are done
     pub fn run(&mut self, handler: &mut impl Handler) -> Result<()> {
-        let ret = self.run_internal(handler);
-        let _res = self.stream_out.shutdown(Shutdown::Both);
-        ret
-    }
-
-    fn run_internal(&mut self, handler: &mut impl Handler) -> Result<()> {
         let peer_addr = self.stream_out.peer_addr()?;
         info!("postgres backend to {:?} started", peer_addr);
         let mut unnamed_query_string = Bytes::new();
@@ -156,80 +112,28 @@ impl PostgresBackend {
         loop {
             let msg = self.read_message()?;
             trace!("got message {:?}", msg);
-
-            // Allow only startup and password messages during auth. Otherwise client would be able to bypass auth
-            // TODO: change that to proper top-level match of protocol state with separate message handling for each state
-            if self.state == ProtoState::Authentication || self.state == ProtoState::Initialization
-            {
-                match msg {
-                    Some(FeMessage::PasswordMessage(ref _m)) => {}
-                    Some(FeMessage::StartupMessage(ref _m)) => {}
-                    Some(_) => {
-                        bail!("protocol violation");
-                    }
-                    None => {}
-                };
-            }
-
             match msg {
                 Some(FeMessage::StartupMessage(m)) => {
                     trace!("got startup message {:?}", m);
 
+                    handler.startup(self, &m)?;
+
                     match m.kind {
                         StartupRequestCode::NegotiateGss | StartupRequestCode::NegotiateSsl => {
                             info!("SSL requested");
                             self.write_message(&BeMessage::Negotiate)?;
                         }
                         StartupRequestCode::Normal => {
-                            // NB: startup() may change self.auth_type -- we are using that in proxy code
-                            // to bypass auth for new users.
-                            handler.startup(self, &m)?;
-
-                            match self.auth_type {
-                                AuthType::Trust => {
-                                    self.write_message_noflush(&BeMessage::AuthenticationOk)?;
-                                    // psycopg2 will not connect if client_encoding is not
-                                    // specified by the server
-                                    self.write_message_noflush(&BeMessage::ParameterStatus)?;
-                                    self.write_message(&BeMessage::ReadyForQuery)?;
-                                    self.state = ProtoState::Established;
-                                }
-                                AuthType::MD5 => {
-                                    rand::thread_rng().fill(&mut self.md5_salt);
-                                    let md5_salt = self.md5_salt;
-                                    self.write_message(&BeMessage::AuthenticationMD5Password(
-                                        &md5_salt,
-                                    ))?;
-                                    self.state = ProtoState::Authentication;
-                                }
-                            }
+                            self.write_message_noflush(&BeMessage::AuthenticationOk)?;
+                            // psycopg2 will not connect if client_encoding is not
+                            // specified by the server
+                            self.write_message_noflush(&BeMessage::ParameterStatus)?;
+                            self.write_message(&BeMessage::ReadyForQuery)?;
+                            self.init_done = true;
                         }
                         StartupRequestCode::Cancel => break,
                     }
                 }
-
-                Some(FeMessage::PasswordMessage(m)) => {
-                    trace!("got password message '{:?}'", m);
-
-                    assert!(self.state == ProtoState::Authentication);
-
-                    let (_, md5_response) = m
-                        .split_last()
-                        .ok_or_else(|| anyhow::Error::msg("protocol violation"))?;
-
-                    if let Err(e) = handler.check_auth_md5(self, md5_response) {
-                        self.write_message(&BeMessage::ErrorResponse(format!("{}", e)))?;
-                        bail!("auth failed: {}", e);
-                    } else {
-                        self.write_message_noflush(&BeMessage::AuthenticationOk)?;
-                        // psycopg2 will not connect if client_encoding is not
-                        // specified by the server
-                        self.write_message_noflush(&BeMessage::ParameterStatus)?;
-                        self.write_message(&BeMessage::ReadyForQuery)?;
-                        self.state = ProtoState::Established;
-                    }
-                }
-
                 Some(FeMessage::Query(m)) => {
                     trace!("got query {:?}", m.body);
                     // xxx distinguish fatal and recoverable errors?

zenith_utils/src/pq_proto.rs

@@ -28,7 +28,6 @@ pub enum FeMessage {
     Terminate,
     CopyData(Bytes),
     CopyDone,
-    PasswordMessage(Bytes),
 }
 
 #[derive(Debug)]
@@ -116,7 +115,6 @@ impl FeMessage {
             b'X' => Ok(Some(FeMessage::Terminate)),
             b'd' => Ok(Some(FeMessage::CopyData(body))),
             b'c' => Ok(Some(FeMessage::CopyDone)),
-            b'p' => Ok(Some(FeMessage::PasswordMessage(body))),
             tag => Err(anyhow!("unknown message tag: {},'{:?}'", tag, body)),
         }
     }
@@ -309,7 +307,6 @@ fn read_null_terminated(buf: &mut Bytes) -> anyhow::Result<Bytes> {
 #[derive(Debug)]
 pub enum BeMessage<'a> {
     AuthenticationOk,
-    AuthenticationMD5Password(&'a [u8; 4]),
     BindComplete,
     CommandComplete(&'a [u8]),
     ControlFile,
@@ -331,7 +328,6 @@ pub enum BeMessage<'a> {
     ReadyForQuery,
     RowDescription(&'a [RowDescriptor<'a>]),
     XLogData(XLogDataBody<'a>),
-    NoticeResponse(String),
 }
 
 // One row desciption in RowDescription packet.
@@ -364,7 +360,7 @@ impl Default for RowDescriptor<'_> {
 pub struct XLogDataBody<'a> {
     pub wal_start: u64,
     pub wal_end: u64,
-    pub timestamp: i64,
+    pub timestamp: u64,
     pub data: &'a [u8],
 }
 
@@ -446,17 +442,7 @@ impl<'a> BeMessage<'a> {
             BeMessage::AuthenticationOk => {
                 buf.put_u8(b'R');
                 write_body(buf, |buf| {
-                    buf.put_i32(0); // Specifies that the authentication was successful.
-                    Ok::<_, io::Error>(())
-                })
-                .unwrap(); // write into BytesMut can't fail
-            }
-
-            BeMessage::AuthenticationMD5Password(salt) => {
-                buf.put_u8(b'R');
-                write_body(buf, |buf| {
-                    buf.put_i32(5); // Specifies that an MD5-encrypted password is required.
-                    buf.put_slice(&salt[..]);
+                    buf.put_i32(0);
                     Ok::<_, io::Error>(())
                 })
                 .unwrap(); // write into BytesMut can't fail
@@ -573,30 +559,6 @@ impl<'a> BeMessage<'a> {
                 .unwrap();
             }
 
-            // NoticeResponse has the same format as ErrorResponse. From doc: "The frontend should display the
-            // message but continue listening for ReadyForQuery or ErrorResponse"
-            BeMessage::NoticeResponse(error_msg) => {
-                // For all the errors set Severity to Error and error code to
-                // 'internal error'.
-
-                // 'N' signalizes NoticeResponse messages
-                buf.put_u8(b'N');
-                write_body(buf, |buf| {
-                    buf.put_u8(b'S'); // severity
-                    write_cstr(&Bytes::from("NOTICE"), buf)?;
-
-                    buf.put_u8(b'C'); // SQLSTATE error code
-                    write_cstr(&Bytes::from("CXX000"), buf)?;
-
-                    buf.put_u8(b'M'); // the message
-                    write_cstr(error_msg.as_bytes(), buf)?;
-
-                    buf.put_u8(0); // terminator
-                    Ok::<_, io::Error>(())
-                })
-                .unwrap();
-            }
-
             BeMessage::NoData => {
                 buf.put_u8(b'n');
                 write_body(buf, |_| Ok::<(), io::Error>(())).unwrap();
@@ -664,7 +626,7 @@ impl<'a> BeMessage<'a> {
                     buf.put_u8(b'w');
                     buf.put_u64(body.wal_start);
                     buf.put_u64(body.wal_end);
-                    buf.put_i64(body.timestamp);
+                    buf.put_u64(body.timestamp);
                     buf.put_slice(body.data);
                     Ok::<_, io::Error>(())
                 })