feat(tvix/nix-compat/nix_daemon/version): add ProtocolVersion
This provides a nice wrapper struct to deal with versions. Change-Id: I6acc03bc9f8d84a0583196073b52776c45d3fe92 Reviewed-on: https://cl.tvl.fyi/c/depot/+/11454 Autosubmit: flokli <flokli@flokli.de> Tested-by: BuildkiteCI Reviewed-by: picnoir picnoir <picnoir@alternativebit.fr>
This commit is contained in:
parent
2c884b8bd2
commit
cf86a098cf
5 changed files with 160 additions and 27 deletions
|
@ -14,3 +14,5 @@ pub mod wire;
|
||||||
mod nix_daemon;
|
mod nix_daemon;
|
||||||
#[cfg(feature = "wire")]
|
#[cfg(feature = "wire")]
|
||||||
pub use nix_daemon::worker_protocol;
|
pub use nix_daemon::worker_protocol;
|
||||||
|
#[cfg(feature = "wire")]
|
||||||
|
pub use nix_daemon::ProtocolVersion;
|
||||||
|
|
|
@ -1 +1,4 @@
|
||||||
pub mod worker_protocol;
|
pub mod worker_protocol;
|
||||||
|
|
||||||
|
mod protocol_version;
|
||||||
|
pub use protocol_version::ProtocolVersion;
|
||||||
|
|
123
tvix/nix-compat/src/nix_daemon/protocol_version.rs
Normal file
123
tvix/nix-compat/src/nix_daemon/protocol_version.rs
Normal file
|
@ -0,0 +1,123 @@
|
||||||
|
/// Protocol versions are represented as a u16.
|
||||||
|
/// The upper 8 bits are the major version, the lower bits the minor.
|
||||||
|
/// This is not aware of any endianness, use [crate::wire::read_u64] to get an
|
||||||
|
/// u64 first, and the try_from() impl from here if you're receiving over the
|
||||||
|
/// Nix Worker protocol.
|
||||||
|
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
||||||
|
pub struct ProtocolVersion(u16);
|
||||||
|
|
||||||
|
impl ProtocolVersion {
|
||||||
|
pub const fn from_parts(major: u8, minor: u8) -> Self {
|
||||||
|
Self(((major as u16) << 8) | minor as u16)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn major(&self) -> u8 {
|
||||||
|
((self.0 & 0xff00) >> 8) as u8
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn minor(&self) -> u8 {
|
||||||
|
(self.0 & 0x00ff) as u8
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl PartialOrd for ProtocolVersion {
|
||||||
|
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
|
||||||
|
Some(self.cmp(other))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Ord for ProtocolVersion {
|
||||||
|
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
|
||||||
|
match self.major().cmp(&other.major()) {
|
||||||
|
std::cmp::Ordering::Less => std::cmp::Ordering::Less,
|
||||||
|
std::cmp::Ordering::Greater => std::cmp::Ordering::Greater,
|
||||||
|
std::cmp::Ordering::Equal => {
|
||||||
|
// same major, compare minor
|
||||||
|
self.minor().cmp(&other.minor())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl From<u16> for ProtocolVersion {
|
||||||
|
fn from(value: u16) -> Self {
|
||||||
|
Self::from_parts(((value & 0xff00) >> 8) as u8, (value & 0x00ff) as u8)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl TryFrom<u64> for ProtocolVersion {
|
||||||
|
type Error = &'static str;
|
||||||
|
|
||||||
|
fn try_from(value: u64) -> Result<Self, Self::Error> {
|
||||||
|
if value & !0xffff != 0 {
|
||||||
|
return Err("only two least significant bits might be populated");
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok((value as u16).into())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl From<ProtocolVersion> for u16 {
|
||||||
|
fn from(value: ProtocolVersion) -> Self {
|
||||||
|
value.0
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl From<ProtocolVersion> for u64 {
|
||||||
|
fn from(value: ProtocolVersion) -> Self {
|
||||||
|
value.0 as u64
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl std::fmt::Display for ProtocolVersion {
|
||||||
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||||
|
write!(f, "{}.{}", self.major(), self.minor())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::ProtocolVersion;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn from_parts() {
|
||||||
|
let version = ProtocolVersion::from_parts(1, 37);
|
||||||
|
assert_eq!(version.major(), 1, "correct major");
|
||||||
|
assert_eq!(version.minor(), 37, "correct minor");
|
||||||
|
assert_eq!("1.37", &version.to_string(), "to_string");
|
||||||
|
|
||||||
|
assert_eq!(0x0125, Into::<u16>::into(version));
|
||||||
|
assert_eq!(0x0125, Into::<u64>::into(version));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn from_u16() {
|
||||||
|
let version = ProtocolVersion::from(0x0125_u16);
|
||||||
|
assert_eq!("1.37", &version.to_string());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn from_u64() {
|
||||||
|
let version = ProtocolVersion::try_from(0x0125_u64).expect("must succeed");
|
||||||
|
assert_eq!("1.37", &version.to_string());
|
||||||
|
}
|
||||||
|
|
||||||
|
/// This contains data in higher bits, which should fail.
|
||||||
|
#[test]
|
||||||
|
fn from_u64_fail() {
|
||||||
|
ProtocolVersion::try_from(0xaa0125_u64).expect_err("must fail");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn ord() {
|
||||||
|
let v0_37 = ProtocolVersion::from_parts(0, 37);
|
||||||
|
let v1_37 = ProtocolVersion::from_parts(1, 37);
|
||||||
|
let v1_40 = ProtocolVersion::from_parts(1, 40);
|
||||||
|
|
||||||
|
assert!(v0_37 < v1_37);
|
||||||
|
assert!(v1_37 > v0_37);
|
||||||
|
assert!(v1_37 < v1_40);
|
||||||
|
assert!(v1_40 > v1_37);
|
||||||
|
assert!(v1_40 <= v1_40);
|
||||||
|
}
|
||||||
|
}
|
|
@ -9,11 +9,13 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||||
|
|
||||||
use crate::wire;
|
use crate::wire;
|
||||||
|
|
||||||
|
use super::ProtocolVersion;
|
||||||
|
|
||||||
static WORKER_MAGIC_1: u64 = 0x6e697863; // "nixc"
|
static WORKER_MAGIC_1: u64 = 0x6e697863; // "nixc"
|
||||||
static WORKER_MAGIC_2: u64 = 0x6478696f; // "dxio"
|
static WORKER_MAGIC_2: u64 = 0x6478696f; // "dxio"
|
||||||
pub static STDERR_LAST: u64 = 0x616c7473; // "alts"
|
pub static STDERR_LAST: u64 = 0x616c7473; // "alts"
|
||||||
/// Protocol version (1.37)
|
|
||||||
static PROTOCOL_VERSION: [u8; 8] = [37, 1, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00];
|
static PROTOCOL_VERSION: ProtocolVersion = ProtocolVersion::from_parts(1, 37);
|
||||||
|
|
||||||
/// Max length of a Nix setting name/value. In bytes.
|
/// Max length of a Nix setting name/value. In bytes.
|
||||||
///
|
///
|
||||||
|
@ -127,7 +129,7 @@ pub struct ClientSettings {
|
||||||
/// FUTUREWORK: write serialization.
|
/// FUTUREWORK: write serialization.
|
||||||
pub async fn read_client_settings<R: AsyncReadExt + Unpin>(
|
pub async fn read_client_settings<R: AsyncReadExt + Unpin>(
|
||||||
r: &mut R,
|
r: &mut R,
|
||||||
client_version: u64,
|
client_version: ProtocolVersion,
|
||||||
) -> std::io::Result<ClientSettings> {
|
) -> std::io::Result<ClientSettings> {
|
||||||
let keep_failed = wire::read_bool(r).await?;
|
let keep_failed = wire::read_bool(r).await?;
|
||||||
let keep_going = wire::read_bool(r).await?;
|
let keep_going = wire::read_bool(r).await?;
|
||||||
|
@ -148,7 +150,7 @@ pub async fn read_client_settings<R: AsyncReadExt + Unpin>(
|
||||||
let build_cores = wire::read_u64(r).await?;
|
let build_cores = wire::read_u64(r).await?;
|
||||||
let use_substitutes = wire::read_bool(r).await?;
|
let use_substitutes = wire::read_bool(r).await?;
|
||||||
let mut overrides = HashMap::new();
|
let mut overrides = HashMap::new();
|
||||||
if client_version >= 12 {
|
if client_version.minor() >= 12 {
|
||||||
let num_overrides = wire::read_u64(r).await?;
|
let num_overrides = wire::read_u64(r).await?;
|
||||||
for _ in 0..num_overrides {
|
for _ in 0..num_overrides {
|
||||||
let name = wire::read_string(r, 0..MAX_SETTING_SIZE).await?;
|
let name = wire::read_string(r, 0..MAX_SETTING_SIZE).await?;
|
||||||
|
@ -173,8 +175,8 @@ pub async fn read_client_settings<R: AsyncReadExt + Unpin>(
|
||||||
/// Performs the initial handshake the server is sending to a connecting client.
|
/// Performs the initial handshake the server is sending to a connecting client.
|
||||||
///
|
///
|
||||||
/// During the handshake, the client first send a magic u64, to which
|
/// During the handshake, the client first send a magic u64, to which
|
||||||
/// the daemon needs to respond with another magic u64. Then, the
|
/// the daemon needs to respond with another magic u64.
|
||||||
/// daemon retrieve the client version, and discard a bunch of now
|
/// Then, the daemon retrieves the client version, and discards a bunch of now
|
||||||
/// obsolete data.
|
/// obsolete data.
|
||||||
///
|
///
|
||||||
/// # Arguments
|
/// # Arguments
|
||||||
|
@ -186,12 +188,12 @@ pub async fn read_client_settings<R: AsyncReadExt + Unpin>(
|
||||||
///
|
///
|
||||||
/// # Return
|
/// # Return
|
||||||
///
|
///
|
||||||
/// The protocol version of a client encoded as a u64.
|
/// The protocol version of the client.
|
||||||
pub async fn server_handshake_client<'a, RW: 'a>(
|
pub async fn server_handshake_client<'a, RW: 'a>(
|
||||||
mut conn: &'a mut RW,
|
mut conn: &'a mut RW,
|
||||||
nix_version: &str,
|
nix_version: &str,
|
||||||
trusted: Trust,
|
trusted: Trust,
|
||||||
) -> std::io::Result<u64>
|
) -> std::io::Result<ProtocolVersion>
|
||||||
where
|
where
|
||||||
&'a mut RW: AsyncReadExt + AsyncWriteExt + Unpin,
|
&'a mut RW: AsyncReadExt + AsyncWriteExt + Unpin,
|
||||||
{
|
{
|
||||||
|
@ -203,39 +205,41 @@ where
|
||||||
))
|
))
|
||||||
} else {
|
} else {
|
||||||
wire::write_u64(&mut conn, WORKER_MAGIC_2).await?;
|
wire::write_u64(&mut conn, WORKER_MAGIC_2).await?;
|
||||||
conn.write_all(&PROTOCOL_VERSION).await?;
|
wire::write_u64(&mut conn, PROTOCOL_VERSION.into()).await?;
|
||||||
conn.flush().await?;
|
conn.flush().await?;
|
||||||
let client_version = wire::read_u64(&mut conn).await?;
|
let client_version = wire::read_u64(&mut conn).await?;
|
||||||
if client_version < 0x10a {
|
// Parse into ProtocolVersion.
|
||||||
|
let client_version: ProtocolVersion = client_version
|
||||||
|
.try_into()
|
||||||
|
.map_err(|e| Error::new(ErrorKind::Unsupported, e))?;
|
||||||
|
if client_version < ProtocolVersion::from_parts(1, 10) {
|
||||||
return Err(Error::new(
|
return Err(Error::new(
|
||||||
ErrorKind::Unsupported,
|
ErrorKind::Unsupported,
|
||||||
format!("The nix client version {} is too old", client_version),
|
format!("The nix client version {} is too old", client_version),
|
||||||
));
|
));
|
||||||
}
|
}
|
||||||
let protocol_minor = client_version & 0x00ff;
|
if client_version.minor() >= 14 {
|
||||||
let _protocol_major = client_version & 0xff00;
|
|
||||||
if protocol_minor >= 14 {
|
|
||||||
// Obsolete CPU affinity.
|
// Obsolete CPU affinity.
|
||||||
let read_affinity = wire::read_u64(&mut conn).await?;
|
let read_affinity = wire::read_u64(&mut conn).await?;
|
||||||
if read_affinity != 0 {
|
if read_affinity != 0 {
|
||||||
let _cpu_affinity = wire::read_u64(&mut conn).await?;
|
let _cpu_affinity = wire::read_u64(&mut conn).await?;
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
if protocol_minor >= 11 {
|
if client_version.minor() >= 11 {
|
||||||
// Obsolete reserveSpace
|
// Obsolete reserveSpace
|
||||||
let _reserve_space = wire::read_u64(&mut conn).await?;
|
let _reserve_space = wire::read_u64(&mut conn).await?;
|
||||||
}
|
}
|
||||||
if protocol_minor >= 33 {
|
if client_version.minor() >= 33 {
|
||||||
// Nix version. We're plain lying, we're not Nix, but eh…
|
// Nix version. We're plain lying, we're not Nix, but eh…
|
||||||
// Setting it to the 2.3 lineage. Not 100% sure this is a
|
// Setting it to the 2.3 lineage. Not 100% sure this is a
|
||||||
// good idea.
|
// good idea.
|
||||||
wire::write_bytes(&mut conn, nix_version).await?;
|
wire::write_bytes(&mut conn, nix_version).await?;
|
||||||
conn.flush().await?;
|
conn.flush().await?;
|
||||||
}
|
}
|
||||||
if protocol_minor >= 35 {
|
if client_version.minor() >= 35 {
|
||||||
write_worker_trust_level(&mut conn, trusted).await?;
|
write_worker_trust_level(&mut conn, trusted).await?;
|
||||||
}
|
}
|
||||||
Ok(protocol_minor)
|
Ok(client_version)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -290,10 +294,10 @@ mod tests {
|
||||||
let mut test_conn = tokio_test::io::Builder::new()
|
let mut test_conn = tokio_test::io::Builder::new()
|
||||||
.read(&WORKER_MAGIC_1.to_le_bytes())
|
.read(&WORKER_MAGIC_1.to_le_bytes())
|
||||||
.write(&WORKER_MAGIC_2.to_le_bytes())
|
.write(&WORKER_MAGIC_2.to_le_bytes())
|
||||||
.write(&PROTOCOL_VERSION)
|
.write(&[37, 1, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00])
|
||||||
// Let's say the client is in sync with the daemon
|
// Let's say the client is in sync with the daemon
|
||||||
// protocol-wise
|
// protocol-wise
|
||||||
.read(&PROTOCOL_VERSION)
|
.read(&[37, 1, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00])
|
||||||
// cpu affinity
|
// cpu affinity
|
||||||
.read(&[0; 8])
|
.read(&[0; 8])
|
||||||
// reservespace
|
// reservespace
|
||||||
|
@ -305,9 +309,11 @@ mod tests {
|
||||||
// Trusted (1 == client trusted
|
// Trusted (1 == client trusted
|
||||||
.write(&[1, 0, 0, 0, 0, 0, 0, 0])
|
.write(&[1, 0, 0, 0, 0, 0, 0, 0])
|
||||||
.build();
|
.build();
|
||||||
server_handshake_client(&mut test_conn, "2.18.2", Trust::Trusted)
|
let client_version = server_handshake_client(&mut test_conn, "2.18.2", Trust::Trusted)
|
||||||
.await
|
.await
|
||||||
.unwrap();
|
.unwrap();
|
||||||
|
|
||||||
|
assert_eq!(client_version, PROTOCOL_VERSION)
|
||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
|
@ -329,7 +335,7 @@ mod tests {
|
||||||
00 00 00 00 00 00 00 00"
|
00 00 00 00 00 00 00 00"
|
||||||
);
|
);
|
||||||
let mut mock = Builder::new().read(&wire_bits).build();
|
let mut mock = Builder::new().read(&wire_bits).build();
|
||||||
let settings = read_client_settings(&mut mock, 21)
|
let settings = read_client_settings(&mut mock, ProtocolVersion::from_parts(1, 21))
|
||||||
.await
|
.await
|
||||||
.expect("should parse");
|
.expect("should parse");
|
||||||
let expected = ClientSettings {
|
let expected = ClientSettings {
|
||||||
|
@ -380,7 +386,7 @@ mod tests {
|
||||||
72 72 65 00 00 00 00 00"
|
72 72 65 00 00 00 00 00"
|
||||||
);
|
);
|
||||||
let mut mock = Builder::new().read(&wire_bits).build();
|
let mut mock = Builder::new().read(&wire_bits).build();
|
||||||
let settings = read_client_settings(&mut mock, 21)
|
let settings = read_client_settings(&mut mock, ProtocolVersion::from_parts(1, 21))
|
||||||
.await
|
.await
|
||||||
.expect("should parse");
|
.expect("should parse");
|
||||||
let overrides = HashMap::from([
|
let overrides = HashMap::from([
|
||||||
|
|
|
@ -3,8 +3,8 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||||
use tokio_listener::{self, SystemOptions, UserOptions};
|
use tokio_listener::{self, SystemOptions, UserOptions};
|
||||||
use tracing::{debug, error, info, instrument, Level};
|
use tracing::{debug, error, info, instrument, Level};
|
||||||
|
|
||||||
use nix_compat::wire;
|
|
||||||
use nix_compat::worker_protocol::{self, server_handshake_client, ClientSettings, Trust};
|
use nix_compat::worker_protocol::{self, server_handshake_client, ClientSettings, Trust};
|
||||||
|
use nix_compat::{wire, ProtocolVersion};
|
||||||
|
|
||||||
#[derive(Parser, Debug)]
|
#[derive(Parser, Debug)]
|
||||||
struct Cli {
|
struct Cli {
|
||||||
|
@ -55,7 +55,7 @@ async fn main() {
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
struct ClientConnection<R: AsyncReadExt + AsyncWriteExt + Unpin> {
|
struct ClientConnection<R: AsyncReadExt + AsyncWriteExt + Unpin> {
|
||||||
pub conn: R,
|
pub conn: R,
|
||||||
pub version_minor: u64,
|
pub version: ProtocolVersion,
|
||||||
pub client_settings: Option<ClientSettings>,
|
pub client_settings: Option<ClientSettings>,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -70,7 +70,7 @@ where
|
||||||
Ok(client_protocol_version) => {
|
Ok(client_protocol_version) => {
|
||||||
let mut client_connection = ClientConnection {
|
let mut client_connection = ClientConnection {
|
||||||
conn,
|
conn,
|
||||||
version_minor: client_protocol_version,
|
version: client_protocol_version,
|
||||||
client_settings: None,
|
client_settings: None,
|
||||||
};
|
};
|
||||||
debug!("Client hanshake succeeded");
|
debug!("Client hanshake succeeded");
|
||||||
|
@ -106,8 +106,7 @@ async fn op_set_options<R>(conn: &mut ClientConnection<R>) -> std::io::Result<Cl
|
||||||
where
|
where
|
||||||
R: AsyncReadExt + AsyncWriteExt + Unpin + std::fmt::Debug,
|
R: AsyncReadExt + AsyncWriteExt + Unpin + std::fmt::Debug,
|
||||||
{
|
{
|
||||||
let settings =
|
let settings = worker_protocol::read_client_settings(&mut conn.conn, conn.version).await?;
|
||||||
worker_protocol::read_client_settings(&mut conn.conn, conn.version_minor).await?;
|
|
||||||
// The client expects us to send some logs when we're processing
|
// The client expects us to send some logs when we're processing
|
||||||
// the settings. Sending STDERR_LAST signal we're done processing.
|
// the settings. Sending STDERR_LAST signal we're done processing.
|
||||||
wire::write_u64(&mut conn.conn, worker_protocol::STDERR_LAST).await?;
|
wire::write_u64(&mut conn.conn, worker_protocol::STDERR_LAST).await?;
|
||||||
|
|
Loading…
Reference in a new issue