pg + pageserver integration testing

src/control_plane.rs

@@ -36,10 +36,25 @@ impl StorageControlPlane {
         };
 
         let workdir = Path::new(env!("CARGO_MANIFEST_DIR")).join("tmp_install/pageserver1");
+        let pg_install_dir = Path::new(env!("CARGO_MANIFEST_DIR")).join("../tmp_install/bin");
 
         fs::create_dir(workdir.clone()).unwrap();
 
-        let pserver = PageServerNode{
+        // initialize data directory
+        // TODO: make wal-redo-postgres workable without data directory?
+        // XXX: common initdb method?
+        // XXX: shared paths
+        let initdb_path = pg_install_dir.join("initdb");
+        let initdb = Command::new(initdb_path)
+            .args(&["-D", workdir.join("wal_redo_pgdata").to_str().unwrap()])
+            .env_clear()
+            .status()
+            .expect("failed to execute initdb");
+        if !initdb.success() {
+            panic!("initdb failed");
+        }
+
+        let pserver = PageServerNode {
             page_service_addr: "127.0.0.1:65200".parse().unwrap(),
             wal_producer_addr: pg_addr,
             data_dir: workdir
@@ -74,7 +89,13 @@ impl PageServerNode {
         Path::new(env!("CARGO_MANIFEST_DIR")).join("./target/debug/pageserver")
     }
 
+    fn pg_install_path(&self) -> PathBuf {
+        Path::new(env!("CARGO_MANIFEST_DIR")).join("../tmp_install/bin")
+    }
+
     pub fn start(&self) {
+        let wal_redo_pgdata = Path::new(env!("CARGO_MANIFEST_DIR")).join("tmp_install/pageserver1/wal_redo_pgdata");
+
         let status = Command::new(self.binary_path())
             .args(&["-D", self.data_dir.to_str().unwrap()])
             .args(&["-w", self.wal_producer_addr.to_string().as_str()])
@@ -82,6 +103,8 @@ impl PageServerNode {
             .arg("-d")
             .arg("--skip-recovery")
             .env_clear()
+            .env("PATH", self.pg_install_path()) // path to postres-wal-redo binary
+            .env("PGDATA", wal_redo_pgdata)      // postres-wal-redo pgdata
             .status()
             .expect("failed to execute initdb");
 
@@ -280,6 +303,8 @@ impl PostgresNode {
             self.whoami()
         );
         let mut client = Client::connect(connstring.as_str(), NoTls).unwrap();
+
+        println!("Running {}", sql);
         client.query(sql, &[]).unwrap()
     }
 

src/page_service.rs

@@ -447,7 +447,11 @@ impl Connection {
         loop {
             let message = self.read_message().await?;
 
-            // println!("query: {:?}", message);
+            // XXX: none seems to appear a lot in log.
+            // Do we have conditions for busy-loop here?
+            if let Some(m) = &message {
+                info!("query: {:?}", m);
+            };
 
             match message {
                 Some(FeMessage::ZenithExistsRequest(req)) => {

src/walreceiver.rs

@@ -36,8 +36,8 @@ pub fn thread_main(conf: PageServerConf) {
             let _res = walreceiver_main(&conf).await;
 
             // TODO: print/log the error
-            info!("WAL streaming connection failed, retrying in 5 seconds...");
-            sleep(Duration::from_secs(5)).await;
+            info!("WAL streaming connection failed, retrying in 1 second...: {:?}", _res);
+            sleep(Duration::from_secs(1)).await;
         }
     });
 }
@@ -68,6 +68,9 @@ async fn walreceiver_main(conf: &PageServerConf) -> Result<(), Error> {
     // Start streaming the WAL, from where we left off previously.
     //
     let last_valid_lsn = page_cache::get_last_valid_lsn();
+    if last_valid_lsn == 0 {
+        page_cache::init_valid_lsn(u64::from(_identify_system.xlogpos()));
+    }
     let startpoint = tokio_postgres::types::Lsn::from(last_valid_lsn);
 
     let mut physical_stream = rclient

tests/pageserver.rs

@@ -11,7 +11,7 @@ use pageserver::control_plane::StorageControlPlane;
 // Handcrafted cases with wal records that are (were) problematic for redo.
 #[test]
 fn test_redo_cases() {
-    // Allocate postgres instance
+    // Allocate postgres instance, but don't start
     let mut compute_cplane = ComputeControlPlane::local();
     let node = compute_cplane.new_vanilla_node();
 
@@ -24,10 +24,12 @@ fn test_redo_cases() {
         page_server_connstring = 'host={} port={}'\n\
     ", pageserver_addr.ip(), pageserver_addr.port()).as_str());
 
-    println!("pew!");
-    sleep(Duration::from_secs(5));
-
+    // start postgres
     node.start();
+
+    println!("await pageserver connection...");
+    sleep(Duration::from_secs(3));
+
     node.safe_psql("postgres", "CREATE TABLE t(key int primary key, value text)");
     node.safe_psql("postgres", "INSERT INTO t SELECT generate_series(1,100000), 'payload'");