WIP

libs/walproposer/build.rs

@@ -115,6 +115,7 @@ fn main() -> anyhow::Result<()> {
         .allowlist_var("am_wal_proposer")
         .allowlist_var("neon_timeline_walproposer")
         .allowlist_var("neon_tenant_walproposer")
+        .allowlist_var("syncSafekeepers")
         .clang_arg(format!("-I{inc_server_path}"))
         .clang_arg(format!("-I{inc_pgxn_path}"))
         // Finish the builder and generate the bindings.

libs/walproposer/src/sim.rs

@@ -107,6 +107,18 @@ pub extern "C" fn sim_epoll_rcv(timeout: i64) -> Event {
                 any_message: anymessage_tag(&message),
             }
         }
+        NodeEvent::Internal(message) => {
+            // store message in thread local storage, C code should use
+            // sim_msg_* functions to access it.
+            MESSAGE_BUF.with(|cell| {
+                *cell.borrow_mut() = message.clone();
+            });
+            Event {
+                tag: EventTag::Internal,
+                tcp: 0,
+                any_message: anymessage_tag(&message),
+            }
+        }
         NodeEvent::WakeTimeout(_) => {
             // can't happen
             unreachable!()

libs/walproposer/src/sim_proto.rs

@@ -87,6 +87,7 @@ pub enum EventTag {
     Accept,
     Closed,
     Message,
+    Internal,
 }
 
 #[repr(u8)]

libs/walproposer/src/simtest/wp_sk.rs

@@ -10,7 +10,7 @@ use utils::{id::TenantTimelineId, logging, lsn::Lsn};
 use crate::{
     bindings::{
         neon_tenant_walproposer, neon_timeline_walproposer, wal_acceptor_connection_timeout,
-        wal_acceptor_reconnect_timeout, wal_acceptors_list, WalProposerRust, WalProposerCleanup,
+        wal_acceptor_reconnect_timeout, wal_acceptors_list, WalProposerRust, WalProposerCleanup, syncSafekeepers,
     },
     c_context,
     simtest::safekeeper::run_server,
@@ -96,6 +96,7 @@ impl Test {
             unsafe {
                 WalProposerCleanup();
 
+                syncSafekeepers = true;
                 wal_acceptors_list = list.into_raw();
                 wal_acceptor_reconnect_timeout = 1000;
                 wal_acceptor_connection_timeout = 5000;
@@ -125,6 +126,52 @@ impl Test {
         let lsn = Lsn::from_str(&res.1)?;
         Ok(lsn)
     }
+
+    fn launch_walproposer(&self, lsn: Lsn) -> WalProposer {
+        let client_node = self.world.new_node();
+
+        // start the client thread
+        let guc = self.safekeepers_guc.clone();
+        let ttid = self.ttid.clone();
+        client_node.launch(move |_| {
+            let list = CString::new(guc).unwrap();
+
+            unsafe {
+                WalProposerCleanup();
+                
+                // TODO: set LSN to a variable
+
+                syncSafekeepers = false;
+                wal_acceptors_list = list.into_raw();
+                wal_acceptor_reconnect_timeout = 1000;
+                wal_acceptor_connection_timeout = 5000;
+                neon_tenant_walproposer =
+                    CString::new(ttid.tenant_id.to_string()).unwrap().into_raw();
+                neon_timeline_walproposer = CString::new(ttid.timeline_id.to_string())
+                    .unwrap()
+                    .into_raw();
+                WalProposerRust();
+            }
+        });
+
+        self.world.await_all();
+
+        WalProposer { 
+            node: client_node,
+        }
+    }
+}
+
+struct WalProposer {
+    node: Arc<Node>,
+}
+
+impl WalProposer {
+    fn gen_wal_record(&self) -> Lsn {
+        // TODO:
+
+        panic!("implement me")
+    }
 }
 
 #[test]
@@ -142,3 +189,20 @@ fn sync_empty_safekeepers() {
     assert_eq!(lsn, Lsn(0));
     println!("Sucessfully synced empty safekeepers at 0/0");
 }
+
+#[test]
+fn run_walproposer_generate_wal() {
+    logging::init(logging::LogFormat::Plain).unwrap();
+
+    let config = TestConfig::new();
+    let test = config.start(1337);
+
+    let lsn = test.sync_safekeepers().unwrap();
+    assert_eq!(lsn, Lsn(0));
+    println!("Sucessfully synced empty safekeepers at 0/0");
+
+    let wp = test.launch_walproposer(lsn);
+    let rec1 = wp.gen_wal_record();
+
+    // TODO:
+}

pgxn/neon/walproposer.c

@@ -71,7 +71,7 @@
 #include "walproposer.h"
 #include "walproposer_utils.h"
 
-static bool syncSafekeepers = false;
+bool syncSafekeepers = false;
 
 char	   *wal_acceptors_list;
 int			wal_acceptor_reconnect_timeout;
@@ -350,8 +350,48 @@ void WalProposerCleanup()
 
 void WalProposerRust()
 {
+	struct stat stat_buf;
+
 	walprop_log(LOG, "WalProposerRust");
-	WalProposerSync(0, NULL);
+#if PG_VERSION_NUM < 150000
+	ThisTimeLineID = 1;
+#endif
+
+	/*
+	 * Initialize postmaster_alive_fds as WaitEventSet checks them.
+	 *
+	 * Copied from InitPostmasterDeathWatchHandle()
+	 */
+	if (pipe(postmaster_alive_fds) < 0)
+		ereport(FATAL,
+				(errcode_for_file_access(),
+					errmsg_internal("could not create pipe to monitor postmaster death: %m")));
+	if (fcntl(postmaster_alive_fds[POSTMASTER_FD_WATCH], F_SETFL, O_NONBLOCK) == -1)
+		ereport(FATAL,
+				(errcode_for_socket_access(),
+					errmsg_internal("could not set postmaster death monitoring pipe to nonblocking mode: %m")));
+
+	ChangeToDataDir();
+
+	/* Create pg_wal directory, if it doesn't exist */
+	if (stat(XLOGDIR, &stat_buf) != 0)
+	{
+		ereport(LOG, (errmsg("creating missing WAL directory \"%s\"", XLOGDIR)));
+		if (MakePGDirectory(XLOGDIR) < 0)
+		{
+			ereport(ERROR,
+					(errcode_for_file_access(),
+						errmsg("could not create directory \"%s\": %m",
+							   XLOGDIR)));
+			exit(1);
+		}
+	}
+
+	WalProposerInit(0, 0);
+
+	BackgroundWorkerUnblockSignals();
+
+	WalProposerStart();
 }
 
 /*
@@ -442,6 +482,7 @@ SimWaitEventSetWait(Safekeeper **sk, long timeout, WaitEvent *occurred_events)
 	} else {
 		Assert(false);
 	}
+	// TODO: handle notification about new LSN available
 }
 #endif
 

pgxn/neon/walproposer.h

@@ -46,6 +46,8 @@
  */
 #define WL_NO_EVENTS 0
 
+extern bool syncSafekeepers;
+
 extern char *wal_acceptors_list;
 extern int	wal_acceptor_reconnect_timeout;
 extern int	wal_acceptor_connection_timeout;

safekeeper/src/simlib/world.rs

@@ -380,6 +380,7 @@ pub enum NodeEvent {
     Accept(TCP),
     Closed(TCP),
     Message((AnyMessage, TCP)),
+    Internal(AnyMessage),
     WakeTimeout(u64),
     // TODO: close?
 }