Move socket to beginning of worker startup

- Moved socket connection to beginning of worker startup.
- Seemed to be causing landlock violations that didn't happen before.
- Removed the conversion to a tokio socket, to rule this out as a cause (we have
  to eventually remove the tokio dependency anyway).
- Realized we were still removing the socket on the host-side, which wasn't
  needed (on failed rendezvous the host will just wipe the whole worker dir).
  Removed this removal, to remove a potential cause.
- Thought we needed a landlock exception for the socket, so I expanded
  `try_restrict` to allow for multiple exceptions (wasn't needed in the end, but
  left it in because why not)
- Saw that Landlock was ignoring exceptions for files that didn't exist. I added
  a check for this case. It clued me in to the actual problem.
- Added a bunch of logging and some tests to try to narrow down this befuddling
  issue.
- The socket was fine. Tthe issue turned out to be that landlock exceptions are
  based on fd's and not paths. We were creating new files for each new job, so
  except for the first excepted files, no exceptions were accepted.
- Applied an exception to the whole worker dir and called it a day.

Author: mrcnski

polkadot/node/core/pvf/common/src/execute.rs

@@ -29,7 +29,7 @@ pub struct Handshake {
 }
 
 /// The response from an execution job on the worker.
-#[derive(Encode, Decode)]
+#[derive(Debug, Encode, Decode)]
 pub enum Response {
 	/// The job completed successfully.
 	Ok {

polkadot/node/core/pvf/common/src/lib.rs

@@ -31,8 +31,11 @@ pub use sp_tracing;
 
 const LOG_TARGET: &str = "parachain::pvf-common";
 
-use std::mem;
-use tokio::io::{self, AsyncRead, AsyncReadExt as _, AsyncWrite, AsyncWriteExt as _};
+use std::{
+	io::{Read, Write},
+	mem,
+};
+use tokio::io;
 
 #[cfg(feature = "test-utils")]
 pub mod tests {
@@ -51,20 +54,22 @@ pub struct SecurityStatus {
 	pub can_unshare_user_namespace_and_change_root: bool,
 }
 
-/// Write some data prefixed by its length into `w`.
-pub async fn framed_send(w: &mut (impl AsyncWrite + Unpin), buf: &[u8]) -> io::Result<()> {
+/// Write some data prefixed by its length into `w`. Sync version of `framed_send` to avoid
+/// dependency on tokio.
+pub fn framed_send_blocking(w: &mut (impl Write + Unpin), buf: &[u8]) -> io::Result<()> {
 	let len_buf = buf.len().to_le_bytes();
-	w.write_all(&len_buf).await?;
-	w.write_all(buf).await?;
+	w.write_all(&len_buf)?;
+	w.write_all(buf)?;
 	Ok(())
 }
 
-/// Read some data prefixed by its length from `r`.
-pub async fn framed_recv(r: &mut (impl AsyncRead + Unpin)) -> io::Result<Vec<u8>> {
+/// Read some data prefixed by its length from `r`. Sync version of `framed_recv` to avoid
+/// dependency on tokio.
+pub fn framed_recv_blocking(r: &mut (impl Read + Unpin)) -> io::Result<Vec<u8>> {
 	let mut len_buf = [0u8; mem::size_of::<usize>()];
-	r.read_exact(&mut len_buf).await?;
+	r.read_exact(&mut len_buf)?;
 	let len = usize::from_le_bytes(len_buf);
 	let mut buf = vec![0; len];
-	r.read_exact(&mut buf).await?;
+	r.read_exact(&mut buf)?;
 	Ok(buf)
 }

polkadot/node/core/pvf/common/src/worker/mod.rs

@@ -24,12 +24,12 @@ use futures::never::Never;
 use std::{
 	any::Any,
 	fmt,
-	os::unix::net::UnixStream as StdUnixStream,
+	os::unix::net::UnixStream,
 	path::PathBuf,
 	sync::mpsc::{Receiver, RecvTimeoutError},
 	time::Duration,
 };
-use tokio::{io, net::UnixStream, runtime::Runtime};
+use tokio::{io, runtime::Runtime};
 
 /// Use this macro to declare a `fn main() {}` that will create an executable that can be used for
 /// spawning the desired worker.
@@ -50,6 +50,8 @@ macro_rules! decl_worker_main {
 			// See <https://github.com/paritytech/polkadot/issues/7117>.
 			$crate::sp_tracing::try_init_simple();
 
+			let worker_pid = std::process::id();
+
 			let args = std::env::args().collect::<Vec<_>>();
 			if args.len() == 1 {
 				print_help($expected_command);
@@ -75,20 +77,25 @@ macro_rules! decl_worker_main {
 				},
 				"--check-can-unshare-user-namespace-and-change-root" => {
 					#[cfg(target_os = "linux")]
-					let status = if security::unshare_user_namespace_and_change_root(
-						$crate::worker::WorkerKind::Execute,
+					let status = if let Err(err) = security::unshare_user_namespace_and_change_root(
+						$crate::worker::WorkerKind::CheckPivotRoot,
+						worker_pid,
 						// We're not accessing any files, so we can try to pivot_root in the temp
 						// dir without conflicts with other processes.
 						&std::env::temp_dir(),
-					)
-					.is_ok()
-					{
-						0
-					} else {
+					) {
+						// Write the error to stderr, log it on the host-side.
+						eprintln!("{}", err);
 						-1
+					} else {
+						0
 					};
 					#[cfg(not(target_os = "linux"))]
-					let status = -1;
+					let status = {
+						// Write the error to stderr, log it on the host-side.
+						eprintln!("not available on macos");
+						-1
+					};
 					std::process::exit(status)
 				},
 
@@ -153,13 +160,15 @@ pub const JOB_TIMEOUT_OVERHEAD: Duration = Duration::from_millis(50);
 pub enum WorkerKind {
 	Prepare,
 	Execute,
+	CheckPivotRoot,
 }
 
 impl fmt::Display for WorkerKind {
 	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
 		match self {
 			Self::Prepare => write!(f, "prepare"),
 			Self::Execute => write!(f, "execute"),
+			Self::CheckPivotRoot => write!(f, "check pivot root"),
 		}
 	}
 }
@@ -178,13 +187,61 @@ pub fn worker_event_loop<F, Fut>(
 	Fut: futures::Future<Output = io::Result<Never>>,
 {
 	let worker_pid = std::process::id();
-	gum::debug!(target: LOG_TARGET, %worker_pid, ?worker_dir_path, "starting pvf worker ({})", worker_kind);
+	gum::debug!(
+		target: LOG_TARGET,
+		%worker_pid,
+		?worker_dir_path,
+		?security_status,
+		"starting pvf worker ({})",
+		worker_kind
+	);
+
+	// Check for a mismatch between the node and worker versions.
+	if let (Some(node_version), Some(worker_version)) = (node_version, worker_version) {
+		if node_version != worker_version {
+			gum::error!(
+				target: LOG_TARGET,
+				%worker_kind,
+				%worker_pid,
+				%node_version,
+				%worker_version,
+				"Node and worker version mismatch, node needs restarting, forcing shutdown",
+			);
+			kill_parent_node_in_emergency();
+			worker_shutdown_message(worker_kind, worker_pid, "Version mismatch");
+			return
+		}
+	}
+
+	// Make sure that we can read the worker dir path, and log its contents.
+	let entries = || -> Result<Vec<_>, io::Error> {
+		std::fs::read_dir(&worker_dir_path)?
+			.map(|res| res.map(|e| e.file_name()))
+			.collect()
+	}();
+	match entries {
+		Ok(entries) =>
+			gum::trace!(target: LOG_TARGET, %worker_pid, ?worker_dir_path, "content of worker dir: {:?}", entries),
+		Err(err) => {
+			gum::error!(
+				target: LOG_TARGET,
+				%worker_kind,
+				%worker_pid,
+				?worker_dir_path,
+				"Could not read worker dir: {}",
+				err.to_string()
+			);
+			worker_shutdown_message(worker_kind, worker_pid, &err.to_string());
+			return
+		},
+	}
 
 	// Connect to the socket.
-	let stream = || -> std::io::Result<StdUnixStream> {
-		let socket_path = worker_dir::socket(&worker_dir_path);
-		let stream = StdUnixStream::connect(&socket_path)?;
-		stream.set_nonblocking(true)?; // See note for `from_std`.
+	let socket_path = worker_dir::socket(&worker_dir_path);
+	let stream = || -> std::io::Result<UnixStream> {
+		let stream = UnixStream::connect(&socket_path)?;
+		// Remove the socket here. We don't also need to do this on the host-side; on failed
+		// rendezvous, the host will delete the whole worker dir.
 		std::fs::remove_file(&socket_path)?;
 		Ok(stream)
 	}();
@@ -203,23 +260,6 @@ pub fn worker_event_loop<F, Fut>(
 		},
 	};
 
-	// Check for a mismatch between the node and worker versions.
-	if let (Some(node_version), Some(worker_version)) = (node_version, worker_version) {
-		if node_version != worker_version {
-			gum::error!(
-				target: LOG_TARGET,
-				%worker_kind,
-				%worker_pid,
-				%node_version,
-				%worker_version,
-				"Node and worker version mismatch, node needs restarting, forcing shutdown",
-			);
-			kill_parent_node_in_emergency();
-			worker_shutdown_message(worker_kind, worker_pid, "Version mismatch");
-			return
-		}
-	}
-
 	// Enable some security features.
 	{
 		// Call based on whether we can change root. Error out if it should work but fails.
@@ -230,9 +270,11 @@ pub fn worker_event_loop<F, Fut>(
 		//       > CLONE_NEWUSER requires that the calling process is not threaded.
 		#[cfg(target_os = "linux")]
 		if security_status.can_unshare_user_namespace_and_change_root {
-			if let Err(err) =
-				security::unshare_user_namespace_and_change_root(worker_kind, &worker_dir_path)
-			{
+			if let Err(err) = security::unshare_user_namespace_and_change_root(
+				worker_kind,
+				worker_pid,
+				&worker_dir_path,
+			) {
 				// The filesystem may be in an inconsistent state, bail out.
 				gum::error!(
 					target: LOG_TARGET,
@@ -251,7 +293,7 @@ pub fn worker_event_loop<F, Fut>(
 		#[cfg(target_os = "linux")]
 		if security_status.can_enable_landlock {
 			let landlock_status =
-				security::landlock::enable_for_worker(worker_kind, &worker_dir_path);
+				security::landlock::enable_for_worker(worker_kind, worker_pid, &worker_dir_path);
 			if !matches!(landlock_status, Ok(landlock::RulesetStatus::FullyEnforced)) {
 				// We previously were able to enable, so this should never happen.
 				//
@@ -284,10 +326,7 @@ pub fn worker_event_loop<F, Fut>(
 	// Run the main worker loop.
 	let rt = Runtime::new().expect("Creates tokio runtime. If this panics the worker will die and the host will detect that and deal with it.");
 	let err = rt
-		.block_on(async move {
-			let stream = UnixStream::from_std(stream)?;
-			event_loop(stream, worker_dir_path).await
-		})
+		.block_on(event_loop(stream, worker_dir_path))
 		// It's never `Ok` because it's `Ok(Never)`.
 		.unwrap_err();