Fix bug in WALRecord serializer

integration_tests/tests/control_plane/mod.rs

@@ -338,6 +338,7 @@ impl ComputeControlPlane<'_> {
             shared_buffers = 1MB\n\
             max_connections = 100\n\
             wal_level = replica\n\
+			wal_sender_timeout = 0\n\
             listen_addresses = '{address}'\n\
             port = {port}\n\
         ",

pageserver/src/bin/pageserver.rs

@@ -7,7 +7,7 @@ use std::fs;
 use std::io;
 use std::path::PathBuf;
 use std::thread;
-use std::{fs::File, fs::OpenOptions, str::FromStr};
+use std::{fs::OpenOptions, str::FromStr};
 
 use clap::{App, Arg};
 use daemonize::Daemonize;
@@ -132,12 +132,12 @@ fn start_pageserver(conf: PageServerConf) -> Result<(), io::Error> {
         let stdout = OpenOptions::new()
             .create(true)
             .append(true)
-            .open(conf.data_dir.join("pageserver.log"))
+            .open(conf.data_dir.join("pageserver-stdout.log"))
             .unwrap();
         let stderr = OpenOptions::new()
             .create(true)
             .append(true)
-            .open(conf.data_dir.join("pageserver.log"))
+            .open(conf.data_dir.join("pageserver-stderr.log"))
             .unwrap();
 
         let daemonize = Daemonize::new()
@@ -222,7 +222,10 @@ fn init_logging(conf: &PageServerConf) -> slog_scope::GlobalLoggerGuard {
         tui::init_logging()
     } else if conf.daemonize {
         let log = conf.data_dir.join("pageserver.log");
-        let log_file = File::create(log).unwrap_or_else(|_| panic!("Could not create log file"));
+        let log_file =  OpenOptions::new()
+            .create(true)
+            .append(true)
+            .open(log).unwrap_or_else(|_| panic!("Could not create log file"));
         let decorator = slog_term::PlainSyncDecorator::new(log_file);
         let drain = slog_term::CompactFormat::new(decorator).build();
         let drain = slog::Filter::new(drain, |record: &slog::Record| {

pageserver/src/page_cache.rs

@@ -317,13 +317,13 @@ impl WALRecord {
     pub fn pack(&self, buf: &mut BytesMut) {
         buf.put_u64(self.lsn);
         buf.put_u8(self.will_init as u8);
-        buf.put_u16(self.rec.len() as u16);
+        buf.put_u32(self.rec.len() as u32);
         buf.put_slice(&self.rec[..]);
     }
     pub fn unpack(buf: &mut BytesMut) -> WALRecord {
         let lsn = buf.get_u64();
         let will_init = buf.get_u8() != 0;
-        let mut dst = vec![0u8; buf.get_u16() as usize];
+        let mut dst = vec![0u8; buf.get_u32() as usize];
         buf.copy_to_slice(&mut dst);
         WALRecord {
             lsn,
@@ -439,7 +439,7 @@ impl PageCache {
             self.put_page_image(tag, lsn, page_img.clone());
         } else {
             // No base image, and no WAL record. Huh?
-            return Err(format!("no page image or WAL record for requested page"))?;
+			panic!("no page image or WAL record for requested page");
         }
 
         // FIXME: assumes little-endian. Only used for the debugging log though

pageserver/src/page_service.rs

@@ -234,6 +234,7 @@ pub fn thread_main(conf: PageServerConf) {
         loop {
             let (socket, peer_addr) = listener.accept().await.unwrap();
             debug!("accepted connection from {}", peer_addr);
+			socket.set_nodelay(true).unwrap();
             let mut conn_handler = Connection::new(conf.clone(), socket);
 
             task::spawn(async move {

pageserver/src/walredo.rs

@@ -197,7 +197,7 @@ impl WalRedoProcess {
                 if res.unwrap() == 0 {
                     break;
                 }
-                debug!("wal-redo-postgres: {}", line.trim());
+                error!("wal-redo-postgres: {}", line.trim());
                 line.clear();
             }
             Ok::<(), Error>(())