maurice/snix
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(nix-daemon): Implement client handler.

Browse source 
This change includes only the basic nix handshake protocol handling and
sets up a client session. The only supported operation at this point is
SetOptions.

Additional operations will be implemented in subsequent cls.

Change-Id: I3eccd9e0ceb270c3865929543c702f1491768852
Reviewed-on: https://cl.tvl.fyi/c/depot/+/12743
Autosubmit: Vladimir Kryachko <v.kryachko@gmail.com>
Tested-by: BuildkiteCI
Reviewed-by: flokli <flokli@flokli.de>
Reviewed-by: edef <edef@edef.eu>
Reviewed-by: Brian Olsen <me@griff.name>
This commit is contained in:
Vova Kryachko 2024-11-08 10:44:27 -05:00 • committed by Vladimir Kryachko
parent 72bc4e0270
commit b564ed9d43
25 changed files with 1822 additions and 253 deletions
Download patch file Download diff file Expand all files Collapse all files

31
tvix/Cargo.lock generated
View file

@ -2300,6 +2300,7 @@ dependencies = [
"nix-compat-derive",
"nom",
"num-traits",
"num_enum",
"pin-project-lite",
"pretty_assertions",
"proptest",
@ -2423,6 +2424,27 @@ dependencies = [
"libc",
]
[[package]]
name = "num_enum"
version = "0.7.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4e613fc340b2220f734a8595782c551f1250e969d87d3be1ae0579e8d4065179"
dependencies = [
"num_enum_derive",
]
[[package]]
name = "num_enum_derive"
version = "0.7.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "af1844ef2428cc3e1cb900be36181049ef3d3193c63e43026cfe202983b27a56"
dependencies = [
"proc-macro-crate",
"proc-macro2",
"quote",
"syn 2.0.79",
]
[[package]]
name = "number_prefix"
version = "0.4.0"
@ -2833,6 +2855,15 @@ dependencies = [
"syn 2.0.79",
]
[[package]]
name = "proc-macro-crate"
version = "3.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8ecf48c7ca261d60b74ab1a7b20da18bede46776b2e55535cb958eb595c5fa7b"
dependencies = [
"toml_edit 0.22.22",
]
[[package]]
name = "proc-macro-error-attr2"
version = "2.0.0"

101
tvix/Cargo.nix
View file

@ -7243,6 +7243,10 @@ rec {
name = "num-traits";
packageId = "num-traits";
}
{
name = "num_enum";
packageId = "num_enum";
}
{
name = "pin-project-lite";
packageId = "pin-project-lite";
@ -7269,7 +7273,7 @@ rec {
name = "tokio";
packageId = "tokio";
optional = true;
features = [ "io-util" "macros" ];
features = [ "io-util" "macros" "sync" ];
}
{
name = "tracing";
@ -7328,13 +7332,14 @@ rec {
features = {
"async" = [ "tokio" ];
"bytes" = [ "dep:bytes" ];
"default" = [ "async" "wire" "nix-compat-derive" ];
"daemon" = [ "tokio" "nix-compat-derive" ];
"default" = [ "async" "daemon" "wire" "nix-compat-derive" ];
"nix-compat-derive" = [ "dep:nix-compat-derive" ];
"pin-project-lite" = [ "dep:pin-project-lite" ];
"tokio" = [ "dep:tokio" ];
"wire" = [ "tokio" "pin-project-lite" "bytes" ];
};
resolvedDefaultFeatures = [ "async" "bytes" "default" "nix-compat-derive" "pin-project-lite" "test" "tokio" "wire" ];
resolvedDefaultFeatures = [ "async" "bytes" "daemon" "default" "nix-compat-derive" "pin-project-lite" "test" "tokio" "wire" ];
};
"nix-compat-derive" = rec {
crateName = "nix-compat-derive";
@ -7453,6 +7458,7 @@ rec {
}
];
src = lib.cleanSourceWith { filter = sourceFilter; src = ./nix-daemon; };
libName = "nix_daemon";
dependencies = [
{
name = "async-trait";
@ -7656,6 +7662,76 @@ rec {
];
};
"num_enum" = rec {
crateName = "num_enum";
version = "0.7.3";
edition = "2021";
sha256 = "0yai0vafhy85mvhknzfqd7lm04hzaln7i5c599rhy8mj831kyqaf";
authors = [
"Daniel Wagner-Hall <dawagner@gmail.com>"
"Daniel Henry-Mantilla <daniel.henry.mantilla@gmail.com>"
"Vincent Esche <regexident@gmail.com>"
];
dependencies = [
{
name = "num_enum_derive";
packageId = "num_enum_derive";
usesDefaultFeatures = false;
}
];
features = {
"complex-expressions" = [ "num_enum_derive/complex-expressions" ];
"default" = [ "std" ];
"std" = [ "num_enum_derive/std" ];
};
resolvedDefaultFeatures = [ "default" "std" ];
};
"num_enum_derive" = rec {
crateName = "num_enum_derive";
version = "0.7.3";
edition = "2021";
sha256 = "0mksna1jj87ydh146gn6jcqkvvs920c3dgh0p4f3xk184kpl865g";
procMacro = true;
authors = [
"Daniel Wagner-Hall <dawagner@gmail.com>"
"Daniel Henry-Mantilla <daniel.henry.mantilla@gmail.com>"
"Vincent Esche <regexident@gmail.com>"
];
dependencies = [
{
name = "proc-macro-crate";
packageId = "proc-macro-crate";
optional = true;
}
{
name = "proc-macro2";
packageId = "proc-macro2";
}
{
name = "quote";
packageId = "quote";
}
{
name = "syn";
packageId = "syn 2.0.79";
features = [ "parsing" ];
}
];
devDependencies = [
{
name = "syn";
packageId = "syn 2.0.79";
features = [ "extra-traits" "parsing" ];
}
];
features = {
"complex-expressions" = [ "syn/full" ];
"default" = [ "std" ];
"proc-macro-crate" = [ "dep:proc-macro-crate" ];
"std" = [ "proc-macro-crate" ];
};
resolvedDefaultFeatures = [ "proc-macro-crate" "std" ];
};
"number_prefix" = rec {
crateName = "number_prefix";
version = "0.4.0";
@ -9033,6 +9109,23 @@ rec {
"verbatim" = [ "syn/parsing" ];
};
};
"proc-macro-crate" = rec {
crateName = "proc-macro-crate";
version = "3.2.0";
edition = "2021";
sha256 = "0yzsqnavb3lmrcsmbrdjfrky9vcbl46v59xi9avn0796rb3likwf";
libName = "proc_macro_crate";
authors = [
"Bastian Köcher <git@kchr.de>"
];
dependencies = [
{
name = "toml_edit";
packageId = "toml_edit 0.22.22";
}
];
};
"proc-macro-error-attr2" = rec {
crateName = "proc-macro-error-attr2";
version = "2.0.0";
@ -13746,7 +13839,7 @@ rec {
"perf" = [ "dep:kstring" ];
"serde" = [ "dep:serde" "toml_datetime/serde" "dep:serde_spanned" ];
};
resolvedDefaultFeatures = [ "display" "parse" "serde" ];
resolvedDefaultFeatures = [ "default" "display" "parse" "serde" ];
};
"tonic" = rec {
crateName = "tonic";

8
tvix/nix-compat/Cargo.toml
View file

@ -8,10 +8,13 @@ edition = "2021"
async = ["tokio"]
# code emitting low-level packets used in the daemon protocol.
wire = ["tokio", "pin-project-lite", "bytes"]
# nix-daemon protocol handling
daemon = ["tokio", "nix-compat-derive"]
test = []
# Enable all features by default.
default = ["async", "wire", "nix-compat-derive"]
default = ["async", "daemon", "wire", "nix-compat-derive"]
[dependencies]
bitflags = { workspace = true }
@ -30,8 +33,9 @@ sha2 = { workspace = true }
thiserror = { workspace = true }
tracing = { workspace = true }
bytes = { workspace = true, optional = true }
tokio = { workspace = true, features = ["io-util", "macros"], optional = true }
tokio = { workspace = true, features = ["io-util", "macros", "sync"], optional = true }
pin-project-lite = { workspace = true, optional = true }
num_enum = "0.7.3"
[dependencies.nix-compat-derive]
path = "../nix-compat-derive"

4
tvix/nix-compat/src/lib.rs
View file

@ -14,7 +14,7 @@ pub mod store_path;
#[cfg(feature = "wire")]
pub mod wire;
#[cfg(feature = "wire")]
#[cfg(feature = "daemon")]
pub mod nix_daemon;
#[cfg(feature = "wire")]
#[cfg(feature = "daemon")]
pub use nix_daemon::worker_protocol;

2
tvix/nix-compat/src/nar/wire/mod.rs
View file

@ -91,9 +91,11 @@ pub const TOK_ENT: [u8; 48] = *b"\x05\0\0\0\0\0\0\0entry\0\0\0\x01\0\0\0\0\0\0\0
pub const TOK_NOD: [u8; 48] = *b"\x04\0\0\0\0\0\0\0node\0\0\0\0\x01\0\0\0\0\0\0\0(\0\0\0\0\0\0\0\x04\0\0\0\0\0\0\0type\0\0\0\0";
pub const TOK_PAR: [u8; 16] = *b"\x01\0\0\0\0\0\0\0)\0\0\0\0\0\0\0";
#[cfg(feature = "async")]
#[allow(dead_code)]
const TOK_PAD_PAR: [u8; 24] = *b"\0\0\0\0\0\0\0\0\x01\0\0\0\0\0\0\0)\0\0\0\0\0\0\0";
#[cfg(feature = "async")]
#[allow(dead_code)]
#[derive(Debug)]
pub(crate) enum PadPar {}

229
tvix/nix-compat/src/nix_daemon/handler.rs Normal file
View file

@ -0,0 +1,229 @@
use std::{future::Future, sync::Arc};
use tokio::{
io::{split, AsyncReadExt, AsyncWriteExt, ReadHalf, WriteHalf},
sync::Mutex,
};
use tracing::debug;
use super::{
worker_protocol::{server_handshake_client, ClientSettings, Operation, Trust, STDERR_LAST},
NixDaemonIO,
};
use crate::wire::{
de::{NixRead, NixReader},
ser::{NixSerialize, NixWrite, NixWriter, NixWriterBuilder},
ProtocolVersion,
};
use crate::{nix_daemon::types::NixError, worker_protocol::STDERR_ERROR};
/// Handles a single connection with a nix client.
///
/// As part of its [`initialization`] it performs the handshake with the client
/// and determines the [ProtocolVersion] and [ClientSettings] to use for the remainder of the session.
///
/// Once initialized, [`handle_client`] needs to be called to handle the rest of the session,
/// it delegates all operation handling to an instance of [NixDaemonIO].
///
/// [`initialization`]: NixDaemon::initialize
#[allow(dead_code)]
pub struct NixDaemon<IO, R, W> {
io: Arc<IO>,
protocol_version: ProtocolVersion,
client_settings: ClientSettings,
reader: NixReader<R>,
writer: Arc<Mutex<NixWriter<W>>>,
}
impl<IO, R, W> NixDaemon<IO, R, W>
where
IO: NixDaemonIO + Sync + Send,
{
pub fn new(
io: Arc<IO>,
protocol_version: ProtocolVersion,
client_settings: ClientSettings,
reader: NixReader<R>,
writer: NixWriter<W>,
) -> Self {
Self {
io,
protocol_version,
client_settings,
reader,
writer: Arc::new(Mutex::new(writer)),
}
}
}
impl<IO, RW> NixDaemon<IO, ReadHalf<RW>, WriteHalf<RW>>
where
RW: AsyncReadExt + AsyncWriteExt + Send + Unpin + 'static,
IO: NixDaemonIO + Sync + Send,
{
/// Async constructor for NixDaemon.
///
/// Performs the initial handshake with the client and retrieves the client's preferred
/// settings.
///
/// The resulting daemon can handle the client session by calling [NixDaemon::handle_client].
pub async fn initialize(io: Arc<IO>, mut connection: RW) -> Result<Self, std::io::Error>
where
RW: AsyncReadExt + AsyncWriteExt + Send + Unpin,
{
let protocol_version =
server_handshake_client(&mut connection, "2.18.2", Trust::Trusted).await?;
connection.write_u64_le(STDERR_LAST).await?;
let (reader, writer) = split(connection);
let mut reader = NixReader::builder()
.set_version(protocol_version)
.build(reader);
let mut writer = NixWriterBuilder::default()
.set_version(protocol_version)
.build(writer);
// The first op is always SetOptions
let operation: Operation = reader.read_value().await?;
if operation != Operation::SetOptions {
return Err(std::io::Error::other(
"Expected SetOptions operation, but got {operation}",
));
}
let client_settings: ClientSettings = reader.read_value().await?;
writer.write_number(STDERR_LAST).await?;
writer.flush().await?;
Ok(Self::new(
io,
protocol_version,
client_settings,
reader,
writer,
))
}
/// Main client connection loop, reads client's requests and responds to them accordingly.
pub async fn handle_client(&mut self) -> Result<(), std::io::Error> {
loop {
let op_code = self.reader.read_number().await?;
match TryInto::<Operation>::try_into(op_code) {
Ok(operation) => match operation {
Operation::SetOptions => {
self.client_settings = self.reader.read_value().await?;
self.handle(async { Ok(()) }).await?
}
_ => {
return Err(std::io::Error::other(format!(
"Operation {operation:?} is not implemented"
)));
}
},
_ => {
return Err(std::io::Error::other(format!(
"Unknown operation code received: {op_code}"
)));
}
}
}
}
/// Handles the operation and sends the response or error to the client.
///
/// As per nix daemon protocol, after sending the request, the client expects zero or more
/// log lines/activities followed by either
/// * STDERR_LAST and the response bytes
/// * STDERR_ERROR and the error
///
/// This is a helper method, awaiting on the passed in future and then
/// handling log lines/activities as described above.
async fn handle<T>(
&mut self,
future: impl Future<Output = std::io::Result<T>>,
) -> Result<(), std::io::Error>
where
T: NixSerialize + Send,
{
let result = future.await;
let mut writer = self.writer.lock().await;
match result {
Ok(r) => {
// the protocol requires that we first indicate that we are done sending logs
// by sending STDERR_LAST and then the response.
writer.write_number(STDERR_LAST).await?;
writer.write_value(&r).await?;
writer.flush().await
}
Err(e) => {
debug!(err = ?e, "IO error");
writer.write_number(STDERR_ERROR).await?;
writer.write_value(&NixError::new(format!("{e:?}"))).await?;
writer.flush().await
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Arc;
use tokio::io::AsyncWriteExt;
use crate::{
wire::ProtocolVersion,
worker_protocol::{ClientSettings, WORKER_MAGIC_1, WORKER_MAGIC_2},
};
struct MockDaemonIO {}
impl NixDaemonIO for MockDaemonIO {}
#[tokio::test]
async fn test_daemon_initialization() {
let mut builder = tokio_test::io::Builder::new();
let test_conn = builder
.read(&WORKER_MAGIC_1.to_le_bytes())
.write(&WORKER_MAGIC_2.to_le_bytes())
// Our version is 1.37
.write(&[37, 1, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00])
// The client's versin is 1.35
.read(&[35, 1, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00])
// cpu affinity
.read(&[0; 8])
// reservespace
.read(&[0; 8])
// version (size)
.write(&[0x06, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00])
// version (data == 2.18.2 + padding)
.write(&[50, 46, 49, 56, 46, 50, 0, 0])
// Trusted (1 == client trusted)
.write(&[1, 0, 0, 0, 0, 0, 0, 0])
// STDERR_LAST
.write(&[115, 116, 108, 97, 0, 0, 0, 0]);
let mut bytes = Vec::new();
let mut writer = NixWriter::new(&mut bytes);
writer
.write_value(&ClientSettings::default())
.await
.unwrap();
writer.flush().await.unwrap();
let test_conn = test_conn
// SetOptions op
.read(&[19, 0, 0, 0, 0, 0, 0, 0])
.read(&bytes)
// STDERR_LAST
.write(&[115, 116, 108, 97, 0, 0, 0, 0])
.build();
let daemon = NixDaemon::initialize(Arc::new(MockDaemonIO {}), test_conn)
.await
.unwrap();
assert_eq!(daemon.client_settings, ClientSettings::default());
assert_eq!(daemon.protocol_version, ProtocolVersion::from_parts(1, 35));
}
}

7
tvix/nix-compat/src/nix_daemon/mod.rs
View file

@ -1 +1,8 @@
pub mod handler;
pub mod types;
pub mod worker_protocol;
/// Represents all possible operations over the nix-daemon protocol.
pub trait NixDaemonIO {
// TODO add methods to it.
}

62
tvix/nix-compat/src/nix_daemon/types.rs Normal file
View file

@ -0,0 +1,62 @@
use nix_compat_derive::{NixDeserialize, NixSerialize};
/// Marker type that consumes/sends and ignores a u64.
#[derive(Clone, Debug, NixDeserialize, NixSerialize)]
#[nix(from = "u64", into = "u64")]
pub struct IgnoredZero;
impl From<u64> for IgnoredZero {
fn from(_: u64) -> Self {
IgnoredZero
}
}
impl From<IgnoredZero> for u64 {
fn from(_: IgnoredZero) -> Self {
0
}
}
#[derive(Debug, NixSerialize)]
pub struct TraceLine {
have_pos: IgnoredZero,
hint: String,
}
/// Represents an error returned by the nix-daemon to its client.
///
/// Adheres to the format described in serialization.md
#[derive(NixSerialize)]
pub struct NixError {
#[nix(version = "26..")]
type_: &'static str,
#[nix(version = "26..")]
level: u64,
#[nix(version = "26..")]
name: &'static str,
msg: String,
#[nix(version = "26..")]
have_pos: IgnoredZero,
#[nix(version = "26..")]
traces: Vec<TraceLine>,
#[nix(version = "..=25")]
exit_status: u64,
}
impl NixError {
pub fn new(msg: String) -> Self {
Self {
type_: "Error",
level: 0, // error
name: "Error",
msg,
have_pos: IgnoredZero {},
traces: vec![],
exit_status: 1,
}
}
}

203
tvix/nix-compat/src/nix_daemon/worker_protocol.rs
View file

@ -1,20 +1,21 @@
use std::{
cmp::min,
collections::HashMap,
collections::BTreeMap,
io::{Error, ErrorKind},
};
use enum_primitive_derive::Primitive;
use num_traits::{FromPrimitive, ToPrimitive};
use nix_compat_derive::{NixDeserialize, NixSerialize};
use num_enum::{FromPrimitive, IntoPrimitive, TryFromPrimitive};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use crate::wire;
use crate::wire::ProtocolVersion;
static WORKER_MAGIC_1: u64 = 0x6e697863; // "nixc"
static WORKER_MAGIC_2: u64 = 0x6478696f; // "dxio"
pub(crate) static WORKER_MAGIC_1: u64 = 0x6e697863; // "nixc"
pub(crate) static WORKER_MAGIC_2: u64 = 0x6478696f; // "dxio"
pub static STDERR_LAST: u64 = 0x616c7473; // "alts"
pub(crate) static STDERR_ERROR: u64 = 0x63787470; // "cxtp"
/// | Nix version | Protocol |
/// |-----------------|----------|
@ -55,7 +56,11 @@ pub static MAX_SETTING_SIZE: usize = 1024;
/// Note: for now, we're using the Nix 2.20 operation description. The
/// operations marked as obsolete are obsolete for Nix 2.20, not
/// necessarily for Nix 2.3. We'll revisit this later on.
#[derive(Debug, PartialEq, Primitive)]
#[derive(
Clone, Debug, PartialEq, TryFromPrimitive, IntoPrimitive, NixDeserialize, NixSerialize,
)]
#[nix(try_from = "u64", into = "u64")]
#[repr(u64)]
pub enum Operation {
IsValidPath = 1,
HasSubstitutes = 3,
@ -106,8 +111,13 @@ pub enum Operation {
/// Log verbosity. In the Nix wire protocol, the client requests a
/// verbosity level to the daemon, which in turns does not produce any
/// log below this verbosity.
#[derive(Debug, PartialEq, Primitive)]
#[derive(
Debug, PartialEq, FromPrimitive, IntoPrimitive, NixDeserialize, NixSerialize, Default, Clone,
)]
#[nix(from = "u64", into = "u64")]
#[repr(u64)]
pub enum Verbosity {
#[default]
LvlError = 0,
LvlWarn = 1,
LvlNotice = 2,
@ -120,7 +130,7 @@ pub enum Verbosity {
/// Settings requested by the client. These settings are applied to a
/// connection to between the daemon and a client.
#[derive(Debug, PartialEq)]
#[derive(Debug, PartialEq, NixDeserialize, NixSerialize, Default)]
pub struct ClientSettings {
pub keep_failed: bool,
pub keep_going: bool,
@ -128,70 +138,21 @@ pub struct ClientSettings {
pub verbosity: Verbosity,
pub max_build_jobs: u64,
pub max_silent_time: u64,
pub verbose_build: bool,
pub use_build_hook: bool,
pub verbose_build: u64,
pub log_type: u64,
pub print_build_trace: u64,
pub build_cores: u64,
pub use_substitutes: bool,
/// Key/Value dictionary in charge of overriding the settings set
/// by the Nix config file.
///
/// Some settings can be safely overidden,
/// some other require the user running the Nix client to be part
/// of the trusted users group.
pub overrides: HashMap<String, String>,
}
/// Reads the client settings from the wire.
///
/// Note: this function **only** reads the settings. It does not
/// manage the log state with the daemon. You'll have to do that on
/// your own. A minimal log implementation will consist in sending
/// back [STDERR_LAST] to the client after reading the client
/// settings.
///
/// FUTUREWORK: write serialization.
pub async fn read_client_settings<R: AsyncReadExt + Unpin>(
r: &mut R,
client_version: ProtocolVersion,
) -> std::io::Result<ClientSettings> {
let keep_failed = r.read_u64_le().await? != 0;
let keep_going = r.read_u64_le().await? != 0;
let try_fallback = r.read_u64_le().await? != 0;
let verbosity_uint = r.read_u64_le().await?;
let verbosity = Verbosity::from_u64(verbosity_uint).ok_or_else(|| {
Error::new(
ErrorKind::InvalidData,
format!("Can't convert integer {} to verbosity", verbosity_uint),
)
})?;
let max_build_jobs = r.read_u64_le().await?;
let max_silent_time = r.read_u64_le().await?;
_ = r.read_u64_le().await?; // obsolete useBuildHook
let verbose_build = r.read_u64_le().await? != 0;
_ = r.read_u64_le().await?; // obsolete logType
_ = r.read_u64_le().await?; // obsolete printBuildTrace
let build_cores = r.read_u64_le().await?;
let use_substitutes = r.read_u64_le().await? != 0;
let mut overrides = HashMap::new();
if client_version.minor() >= 12 {
let num_overrides = r.read_u64_le().await?;
for _ in 0..num_overrides {
let name = wire::read_string(r, 0..=MAX_SETTING_SIZE).await?;
let value = wire::read_string(r, 0..=MAX_SETTING_SIZE).await?;
overrides.insert(name, value);
}
}
Ok(ClientSettings {
keep_failed,
keep_going,
try_fallback,
verbosity,
max_build_jobs,
max_silent_time,
verbose_build,
build_cores,
use_substitutes,
overrides,
})
#[nix(version = "12..")]
pub overrides: BTreeMap<String, String>,
}
/// Performs the initial handshake the server is sending to a connecting client.
@ -269,18 +230,17 @@ where
/// Read a worker [Operation] from the wire.
pub async fn read_op<R: AsyncReadExt + Unpin>(r: &mut R) -> std::io::Result<Operation> {
let op_number = r.read_u64_le().await?;
Operation::from_u64(op_number).ok_or(Error::new(
ErrorKind::InvalidData,
format!("Invalid OP number {}", op_number),
))
Operation::try_from(op_number).map_err(|_| {
Error::new(
ErrorKind::InvalidData,
format!("Invalid OP number {}", op_number),
)
})
}
/// Write a worker [Operation] to the wire.
pub async fn write_op<W: AsyncWriteExt + Unpin>(w: &mut W, op: &Operation) -> std::io::Result<()> {
let op = Operation::to_u64(op).ok_or(Error::new(
ErrorKind::Other,
format!("Can't convert the OP {:?} to u64", op),
))?;
pub async fn write_op<W: AsyncWriteExt + Unpin>(w: &mut W, op: Operation) -> std::io::Result<()> {
let op: u64 = op.into();
w.write_u64(op).await
}
@ -309,8 +269,6 @@ where
#[cfg(test)]
mod tests {
use super::*;
use hex_literal::hex;
use tokio_test::io::Builder;
#[tokio::test]
async fn test_init_hanshake() {
@ -391,99 +349,4 @@ mod tests {
assert_eq!(picked_version, ProtocolVersion::from_parts(1, 24))
}
#[tokio::test]
async fn test_read_client_settings_without_overrides() {
// Client settings bits captured from a Nix 2.3.17 run w/ sockdump (protocol version 21).
let wire_bits = hex!(
"00 00 00 00 00 00 00 00 \
00 00 00 00 00 00 00 00 \
00 00 00 00 00 00 00 00 \
02 00 00 00 00 00 00 00 \
10 00 00 00 00 00 00 00 \
00 00 00 00 00 00 00 00 \
01 00 00 00 00 00 00 00 \
00 00 00 00 00 00 00 00 \
00 00 00 00 00 00 00 00 \
00 00 00 00 00 00 00 00 \
00 00 00 00 00 00 00 00 \
01 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 settings = read_client_settings(&mut mock, ProtocolVersion::from_parts(1, 21))
.await
.expect("should parse");
let expected = ClientSettings {
keep_failed: false,
keep_going: false,
try_fallback: false,
verbosity: Verbosity::LvlNotice,
max_build_jobs: 16,
max_silent_time: 0,
verbose_build: false,
build_cores: 0,
use_substitutes: true,
overrides: HashMap::new(),
};
assert_eq!(settings, expected);
}
#[tokio::test]
async fn test_read_client_settings_with_overrides() {
// Client settings bits captured from a Nix 2.3.17 run w/ sockdump (protocol version 21).
let wire_bits = hex!(
"00 00 00 00 00 00 00 00 \
00 00 00 00 00 00 00 00 \
00 00 00 00 00 00 00 00 \
02 00 00 00 00 00 00 00 \
10 00 00 00 00 00 00 00 \
00 00 00 00 00 00 00 00 \
01 00 00 00 00 00 00 00 \
00 00 00 00 00 00 00 00 \
00 00 00 00 00 00 00 00 \
00 00 00 00 00 00 00 00 \
00 00 00 00 00 00 00 00 \
01 00 00 00 00 00 00 00 \
02 00 00 00 00 00 00 00 \
0c 00 00 00 00 00 00 00 \
61 6c 6c 6f 77 65 64 2d \
75 72 69 73 00 00 00 00 \
1e 00 00 00 00 00 00 00 \
68 74 74 70 73 3a 2f 2f \
62 6f 72 64 65 61 75 78 \
2e 67 75 69 78 2e 67 6e \
75 2e 6f 72 67 2f 00 00 \
0d 00 00 00 00 00 00 00 \
61 6c 6c 6f 77 65 64 2d \
75 73 65 72 73 00 00 00 \
0b 00 00 00 00 00 00 00 \
6a 65 61 6e 20 70 69 65 \
72 72 65 00 00 00 00 00"
);
let mut mock = Builder::new().read(&wire_bits).build();
let settings = read_client_settings(&mut mock, ProtocolVersion::from_parts(1, 21))
.await
.expect("should parse");
let overrides = HashMap::from([
(
String::from("allowed-uris"),
String::from("https://bordeaux.guix.gnu.org/"),
),
(String::from("allowed-users"), String::from("jean pierre")),
]);
let expected = ClientSettings {
keep_failed: false,
keep_going: false,
try_fallback: false,
verbosity: Verbosity::LvlNotice,
max_build_jobs: 16,
max_silent_time: 0,
verbose_build: false,
build_cores: 0,
use_substitutes: true,
overrides,
};
assert_eq!(settings, expected);
}
}

9
tvix/nix-compat/src/wire/ser/bytes.rs
View file

@ -38,6 +38,15 @@ impl NixSerialize for str {
}
}
impl NixSerialize for &str {
async fn serialize<W>(&self, writer: &mut W) -> Result<(), W::Error>
where
W: NixWrite,
{
writer.write_slice(self.as_bytes()).await
}
}
#[cfg(test)]
mod test {
use hex_literal::hex;

10
tvix/nix-compat/src/wire/ser/mod.rs
View file

@ -122,3 +122,13 @@ pub trait NixSerialize {
where
W: NixWrite;
}
// Noop
impl NixSerialize for () {
async fn serialize<W>(&self, _writer: &mut W) -> Result<(), W::Error>
where
W: NixWrite,
{
Ok(())
}
}

39
tvix/nix-daemon/src/bin/nix-daemon.rs
View file

@ -1,8 +1,10 @@
use clap::Parser;
use mimalloc::MiMalloc;
use std::error::Error;
use tokio::io::AsyncWriteExt;
use nix_compat::nix_daemon::handler::NixDaemon;
use nix_daemon::TvixDaemon;
use std::{error::Error, sync::Arc};
use tokio_listener::SystemOptions;
use tracing::error;
use tvix_store::utils::{construct_services, ServiceUrlsGrpc};
#[global_allocator]
@ -58,7 +60,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
}
async fn run(cli: Cli) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let (_blob_service, _directory_service, _path_info_service, _nar_calculation_service) =
let (_blob_service, _directory_service, path_info_service, _nar_calculation_service) =
construct_services(cli.service_addrs).await?;
let listen_address = cli.listen_args.listen_address.unwrap_or_else(|| {
@ -74,16 +76,29 @@ async fn run(cli: Cli) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
)
.await?;
while let Ok((mut connection, _)) = listener.accept().await {
tokio::spawn(async move {
let ucred = connection
.try_borrow_unix()
.and_then(|u| u.peer_cred().ok());
let io = Arc::new(TvixDaemon::new(path_info_service));
// For now we just write the connected process credentials into the connection.
let _ = connection
.write_all(format!("Hello {:?}", ucred).as_bytes())
.await;
while let Ok((connection, _)) = listener.accept().await {
let io = io.clone();
tokio::spawn(async move {
match NixDaemon::initialize(io.clone(), connection).await {
Ok(mut daemon) => {
if let Err(error) = daemon.handle_client().await {
match error.kind() {
std::io::ErrorKind::UnexpectedEof => {
// client disconnected, nothing to do
}
_ => {
// otherwise log the error and disconnect
error!(error=?error, "client error");
}
}
}
}
Err(error) => {
error!(error=?error, "nix-daemon handshake failed");
}
}
});
}
Ok(())

18
tvix/nix-daemon/src/lib.rs Normal file
View file

@ -0,0 +1,18 @@
use std::sync::Arc;
use nix_compat::nix_daemon::NixDaemonIO;
use tvix_store::pathinfoservice::PathInfoService;
#[allow(dead_code)]
pub struct TvixDaemon {
path_info_service: Arc<dyn PathInfoService>,
}
impl TvixDaemon {
pub fn new(path_info_service: Arc<dyn PathInfoService>) -> Self {
Self { path_info_service }
}
}
/// Implements [NixDaemonIO] backed by tvix services.
impl NixDaemonIO for TvixDaemon {}