{"header":{"message":"Basic CI infra (no runner yet)","description":null,"timestamp":"2026-03-04T17:58:49.106504753Z","authors":[{}]},"hash":"YJLEG4WRYKKQJMMDRKIRG4T4L7ZZ7N777EKIRSHHWHNZYWZJ7G5AC","authors":[{"login":"pmeunier","key":"8ES6e1QmLjuQn1DzS2DQRKebzCKPHpdf4E9hfwMqQoe7"}],"deps":{"hashes":[["YJLEG4WRYKKQJMMDRKIRG4T4L7ZZ7N777EKIRSHHWHNZYWZJ7G5AC","Basic CI infra (no runner yet)"],["AA",null],["2QMA3JQXSSDMQPXI2G73GAP3JINFXRG74XPSQXJMJ6FYPHVDYQ4QC",""]],"deps":["2QMA3JQXSSDMQPXI2G73GAP3JINFXRG74XPSQXJMJ6FYPHVDYQ4QC"],"known":[]},"hunks":[{"FileAdd":{"meta":{"basename":"ci","metadata":576},"context":[{"change":2,"pos":1}],"contents":[]}},{"FileAdd":{"meta":{"basename":"src","metadata":576},"context":[{"change":0,"pos":14}],"contents":[]}},{"FileAdd":{"meta":{"basename":"main.rs","metadata":0},"context":[{"change":0,"pos":31}],"contents":[{"Add":"use ci::Message;"},{"Add":"use serde_derive::*;"},{"Add":"use std::io::Read;"},{"Add":"use std::net::{Ipv6Addr, SocketAddr, ToSocketAddrs};"},{"Add":"use std::path::{Path, PathBuf};"},{"Add":"use std::sync::{Arc, Mutex};"},{"Add":"use std::time::{Duration, SystemTime};"},{"Add":"use tracing::*;"},{"Add":"use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};"},{"Add":""},{"Add":"#[derive(Serialize, Deserialize)]"},{"Add":"struct ConfigFile {"},{"Add":" key_path: String,"},{"Add":" port: u16,"},{"Add":" timeout_secs: usize,"},{"Add":" server_public_keys: Vec,"},{"Add":" log_path: String,"},{"Add":" tarball_path: String,"},{"Add":"}"},{"Add":""},{"Add":"#[derive(Serialize, Deserialize)]"},{"Add":"struct BuildResult {"},{"Add":" finished: chrono::DateTime,"},{"Add":" status: Option,"},{"Add":" link: Option,"},{"Add":" job: ci::Job,"},{"Add":"}"},{"Add":""},{"Add":"use clap::*;"},{"Add":""},{"Add":"#[derive(Debug, Parser)]"},{"Add":"pub struct App {"},{"Add":" #[arg(short, long)]"},{"Add":" config: PathBuf,"},{"Add":"}"},{"Add":""},{"Add":"#[tokio::main]"},{"Add":"async fn main() {"},{"Add":" tracing_subscriber::registry()"},{"Add":" .with("},{"Add":" tracing_subscriber::EnvFilter::try_from_default_env()"},{"Add":" .unwrap_or_else(|_| \"ci=debug\".into()),"},{"Add":" )"},{"Add":" .with(tracing_subscriber::fmt::layer())"},{"Add":" .init();"},{"Add":" let matches = App::parse();"},{"Add":" let conf: ConfigFile ="},{"Add":" toml::from_str(&std::fs::read_to_string(&matches.config).unwrap()).unwrap();"},{"Add":" let config = Arc::new(thrussh::client::Config::default());"},{"Add":" let addr = Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 0);"},{"Add":" let addr = (addr, conf.port).to_socket_addrs().unwrap().next().unwrap();"},{"Add":" let key = Arc::new(thrussh_keys::load_secret_key(&conf.key_path, None).unwrap());"},{"Add":" let (sender, mut receiver) = tokio::sync::mpsc::channel(1);"},{"Add":" let client = CiClient {"},{"Add":" process: Arc::new(Mutex::new(Process::default())),"},{"Add":" log_path: Path::new(&conf.log_path).to_path_buf(),"},{"Add":" tarball_path: Path::new(&conf.tarball_path).to_path_buf(),"},{"Add":" last_window_adjustment: SystemTime::now(),"},{"Add":" server_public_keys: Arc::new("},{"Add":" conf.server_public_keys"},{"Add":" .iter()"},{"Add":" .map(|p| thrussh_keys::parse_public_key_base64(p).unwrap())"},{"Add":" .collect(),"},{"Add":" ),"},{"Add":" sender,"},{"Add":" };"},{"Add":" loop {"},{"Add":" if let Err(e) = client"},{"Add":" .protocol(&addr, config.clone(), key.clone(), &mut receiver)"},{"Add":" .await"},{"Add":" {"},{"Add":" error!(\"restarting because of error: {:?}\", e)"},{"Add":" }"},{"Add":" tokio::time::sleep(std::time::Duration::from_secs(1)).await;"},{"Add":" }"},{"Add":"}"},{"Add":""},{"Add":"#[derive(Clone, Debug)]"},{"Add":"pub struct CiClient {"},{"Add":" process: Arc>,"},{"Add":" tarball_path: PathBuf,"},{"Add":" log_path: PathBuf,"},{"Add":" last_window_adjustment: SystemTime,"},{"Add":" server_public_keys: Arc>,"},{"Add":" sender: tokio::sync::mpsc::Sender<(ci::Job, Option, Option)>,"},{"Add":"}"},{"Add":""},{"Add":"#[derive(Debug, Default)]"},{"Add":"struct Process {"},{"Add":" child: Option>>,"},{"Add":" job: Option,"},{"Add":" tarball: Option<(std::fs::File, usize)>,"},{"Add":"}"},{"Add":""},{"Add":"impl Process {"},{"Add":" fn is_ready(&self) -> bool {"},{"Add":" self.child.is_none() && self.job.is_none()"},{"Add":" }"},{"Add":"}"},{"Add":""},{"Add":"impl CiClient {"},{"Add":" pub async fn protocol("},{"Add":" &self,"},{"Add":" addr: &SocketAddr,"},{"Add":" config: Arc,"},{"Add":" key: Arc,"},{"Add":" receiver: &mut tokio::sync::mpsc::Receiver<(ci::Job, Option, Option)>,"},{"Add":" ) -> Result<(), anyhow::Error> {"},{"Add":" let mut h = thrussh::client::connect(config, &addr, self.clone()).await?;"},{"Add":" debug!(\"Opening session\");"},{"Add":" if !h.authenticate_publickey(\"ci\", key).await? {"},{"Add":" return Ok(());"},{"Add":" }"},{"Add":" let mut channel = h.channel_open_session().await?;"},{"Add":" channel"},{"Add":" .data("},{"Add":" &bincode::serialize(&Message::Handshake {"},{"Add":" version: ci::VERSION,"},{"Add":" id: 0,"},{"Add":" })"},{"Add":" .unwrap()[..],"},{"Add":" )"},{"Add":" .await?;"},{"Add":" debug!(\"handshake done\");"},{"Add":" 'outer: loop {"},{"Add":" if self.process.lock().unwrap().is_ready() {"},{"Add":" channel"},{"Add":" .data(&bincode::serialize(&Message::Ready).unwrap()[..])"},{"Add":" .await?;"},{"Add":" debug!(\"ready\");"},{"Add":" }"},{"Add":" loop {"},{"Add":" tokio::select! {"},{"Add":" msg = channel.wait() => {"},{"Add":" debug!(\"msg = {:?}\", msg);"},{"Add":" if !self.handle_msg(&mut channel, &self.sender, msg).await? {"},{"Add":" break 'outer"},{"Add":" }"},{"Add":" }"},{"Add":" msg = receiver.recv() => {"},{"Add":" debug!(\"message {:#?}\", msg);"},{"Add":" if let Some(p) = self.process.lock().unwrap().child.take() {"},{"Add":" p.await??"},{"Add":" }"},{"Add":" if let Some((job, exit_status, path)) = msg {"},{"Add":" self.send_log(&mut channel, job, exit_status, path).await?"},{"Add":" }"},{"Add":" channel.data(&bincode::serialize(&Message::Ready).unwrap()[..]).await?;"},{"Add":" }"},{"Add":" }"},{"Add":" }"},{"Add":" }"},{"Add":" Ok(())"},{"Add":" }"},{"Add":""},{"Add":" async fn handle_msg("},{"Add":" &self,"},{"Add":" channel: &mut thrussh::client::Channel,"},{"Add":" sender: &tokio::sync::mpsc::Sender<(ci::Job, Option, Option)>,"},{"Add":" msg: Option,"},{"Add":" ) -> Result {"},{"Add":" match msg {"},{"Add":" Some(thrussh::ChannelMsg::Data { data }) => {"},{"Add":" let mut proc = self.process.lock().unwrap();"},{"Add":" if let Some((mut f, mut len)) = proc.tarball.take() {"},{"Add":" debug!(\"len = {:?}\", len);"},{"Add":" use std::io::Write;"},{"Add":" f.write_all(&data)?;"},{"Add":" len -= data.len();"},{"Add":" if len > 0 {"},{"Add":" proc.tarball = Some((f, len));"},{"Add":" }"},{"Add":" return Ok(true);"},{"Add":" }"},{"Add":" let msg = bincode::deserialize::(&data);"},{"Add":" debug!(\"msg = {:?}\", msg);"},{"Add":" match msg {"},{"Add":" Ok(Message::Job(job)) => {"},{"Add":" self.handle_job(channel, sender.clone(), &mut proc, job)"},{"Add":" .await?"},{"Add":" }"},{"Add":" Ok(Message::Chunk { id, len, .. }) => {"},{"Add":" let p = self.tarball_path.join(&format!(\"{}.tar.gz.tmp\", id));"},{"Add":" if len == 0 {"},{"Add":" let p2 = self.tarball_path.join(&format!(\"{}.tar.gz\", id));"},{"Add":" std::fs::rename(&p, &p2)?;"},{"Add":" proc.tarball = None;"},{"Add":" let job = proc.job.take().unwrap();"},{"Add":" self.handle_job(channel, sender.clone(), &mut proc, job)"},{"Add":" .await?;"},{"Add":" return Ok(true);"},{"Add":" }"},{"Add":" let file = std::fs::OpenOptions::new()"},{"Add":" .write(true)"},{"Add":" .create(true)"},{"Add":" .append(true)"},{"Add":" .open(&p)"},{"Add":" .unwrap();"},{"Add":" proc.tarball = Some((file, len as usize));"},{"Add":" }"},{"Add":" Ok(msg) => {"},{"Add":" debug!(\"msg = {:?}\", msg);"},{"Add":" }"},{"Add":" _ => return Ok(false),"},{"Add":" }"},{"Add":" }"},{"Add":" None => return Ok(false),"},{"Add":" msg => debug!(\"{:?}\", msg),"},{"Add":" }"},{"Add":" Ok(true)"},{"Add":" }"},{"Add":""},{"Add":" async fn handle_job("},{"Add":" &self,"},{"Add":" channel: &mut thrussh::client::Channel,"},{"Add":" sender: tokio::sync::mpsc::Sender<(ci::Job, Option, Option)>,"},{"Add":" process: &mut Process,"},{"Add":" job: ci::Job,"},{"Add":" ) -> Result<(), anyhow::Error> {"},{"Add":" let p = self.tarball_path.join(&format!(\"{}.tar.gz\", job.id));"},{"Add":" debug!(\"p = {:?}\", p);"},{"Add":" if std::fs::metadata(&p).is_err() {"},{"Add":" debug!(\"getting tarball\");"},{"Add":" channel"},{"Add":" .data(&bincode::serialize(&Message::GetTarball { id: job.id }).unwrap()[..])"},{"Add":" .await?;"},{"Add":" process.job = Some(job);"},{"Add":" return Ok(());"},{"Add":" }"},{"Add":" debug!(\"tar = {:?}\", p);"},{"Add":" let status = std::process::Command::new(\"tar\")"},{"Add":" .args(&[\"-xf\", p.to_str().unwrap()])"},{"Add":" .current_dir(&self.tarball_path)"},{"Add":" .status()"},{"Add":" .unwrap();"},{"Add":" debug!(\"nix: {:?}\", status);"},{"Add":""},{"Add":" let tarballp = self.tarball_path.join(job.id.to_string());"},{"Add":" let logp = self.log_path.clone();"},{"Add":""},{"Add":" let result_path = logp.join(&format!(\"{}.result\", job.id));"},{"Add":" if let Ok(mut f) = std::fs::File::open(&result_path) {"},{"Add":" if let Ok(build_result) = serde_json::from_reader::<_, BuildResult>(&mut f) {"},{"Add":" sender"},{"Add":" .send((build_result.job, build_result.status, build_result.link))"},{"Add":" .await?;"},{"Add":" return Ok(());"},{"Add":" }"},{"Add":" }"},{"Add":""},{"Add":" debug!(\"p = {:?}\", tarballp);"},{"Add":" process.child = Some(tokio::task::spawn(async move {"},{"Add":" let mut process = tokio::process::Command::new(\"nix-build\")"},{"Add":" .arg(\"default.nix\")"},{"Add":" .current_dir(&tarballp)"},{"Add":" .stdin(std::process::Stdio::null())"},{"Add":" .stdout(std::process::Stdio::piped())"},{"Add":" .stderr(std::process::Stdio::piped())"},{"Add":" .spawn()"},{"Add":" .unwrap();"},{"Add":" let stdout = process.stdout.as_mut().unwrap();"},{"Add":" let stderr = process.stderr.as_mut().unwrap();"},{"Add":" let mut fstdout ="},{"Add":" tokio::fs::File::create(logp.join(&format!(\"{}.stdout\", job.id))).await?;"},{"Add":" let mut fstderr ="},{"Add":" tokio::fs::File::create(logp.join(&format!(\"{}.stderr\", job.id))).await?;"},{"Add":" let (a, b) = futures::future::join("},{"Add":" tokio::io::copy(stdout, &mut fstdout),"},{"Add":" tokio::io::copy(stderr, &mut fstderr),"},{"Add":" )"},{"Add":" .await;"},{"Add":" a?;"},{"Add":" b?;"},{"Add":" let status = process.wait().await?;"},{"Add":" debug!(\"status = {:?}\", status);"},{"Add":""},{"Add":" let mut result_file = std::fs::File::create(&result_path)?;"},{"Add":" let link = std::fs::read_link(&tarballp.join(\"result\")).ok();"},{"Add":" serde_json::to_writer("},{"Add":" &mut result_file,"},{"Add":" &BuildResult {"},{"Add":" finished: chrono::Utc::now(),"},{"Add":" status: status.code(),"},{"Add":" job: job.clone(),"},{"Add":" link: link.clone(),"},{"Add":" },"},{"Add":" )?;"},{"Add":""},{"Add":" sender.send((job, status.code(), link)).await?;"},{"Add":""},{"Add":" std::fs::remove_dir_all(&tarballp).unwrap_or(());"},{"Add":" std::fs::remove_file(&p)?;"},{"Add":""},{"Add":" Ok(())"},{"Add":" }));"},{"Add":" Ok(())"},{"Add":" }"},{"Add":""},{"Add":" async fn send_log("},{"Add":" &self,"},{"Add":" channel: &mut thrussh::client::Channel,"},{"Add":" job: ci::Job,"},{"Add":" exit_status: Option,"},{"Add":" path: Option,"},{"Add":" ) -> Result<(), anyhow::Error> {"},{"Add":" let id = job.id;"},{"Add":" let msg = Message::Log {"},{"Add":" job,"},{"Add":" exit_status,"},{"Add":" path,"},{"Add":" };"},{"Add":" channel.data(&bincode::serialize(&msg).unwrap()[..]).await?;"},{"Add":""},{"Add":" let mut buf = Vec::with_capacity(4096);"},{"Add":" debug!("},{"Add":" \"stdout: {:?}\","},{"Add":" self.log_path.join(&format!(\"{}.stdout\", id))"},{"Add":" );"},{"Add":" if let Ok(ref mut stdout) ="},{"Add":" std::fs::File::open(&self.log_path.join(&format!(\"{}.stdout\", id)))"},{"Add":" {"},{"Add":" let len = channel.writable_packet_size().min(MAX_BUF_SIZE);"},{"Add":" buf.resize(len, 0);"},{"Add":" while let Ok(n) = stdout.read(&mut buf) {"},{"Add":" if n == 0 {"},{"Add":" channel"},{"Add":" .data("},{"Add":" &bincode::serialize(&Message::Chunk {"},{"Add":" id,"},{"Add":" stderr: false,"},{"Add":" len: 0,"},{"Add":" })"},{"Add":" .unwrap()[..],"},{"Add":" )"},{"Add":" .await?;"},{"Add":" break;"},{"Add":" }"},{"Add":" channel"},{"Add":" .data("},{"Add":" &bincode::serialize(&Message::Chunk {"},{"Add":" id,"},{"Add":" stderr: false,"},{"Add":" len: n as u32,"},{"Add":" })"},{"Add":" .unwrap()[..],"},{"Add":" )"},{"Add":" .await?;"},{"Add":" channel.data(&buf[..n]).await?"},{"Add":" }"},{"Add":" }"},{"Add":" if let Ok(ref mut stdout) ="},{"Add":" std::fs::File::open(&self.log_path.join(&format!(\"{}.stderr\", id)))"},{"Add":" {"},{"Add":" let len = channel.writable_packet_size().min(MAX_BUF_SIZE);"},{"Add":" buf.resize(len, 0);"},{"Add":" while let Ok(n) = stdout.read(&mut buf) {"},{"Add":" if n == 0 {"},{"Add":" channel"},{"Add":" .data("},{"Add":" &bincode::serialize(&Message::Chunk {"},{"Add":" id,"},{"Add":" stderr: true,"},{"Add":" len: 0,"},{"Add":" })"},{"Add":" .unwrap()[..],"},{"Add":" )"},{"Add":" .await?;"},{"Add":" break;"},{"Add":" }"},{"Add":" channel"},{"Add":" .data("},{"Add":" &bincode::serialize(&Message::Chunk {"},{"Add":" id,"},{"Add":" stderr: true,"},{"Add":" len: n as u32,"},{"Add":" })"},{"Add":" .unwrap()[..],"},{"Add":" )"},{"Add":" .await?;"},{"Add":" channel.data(&buf[..n]).await?"},{"Add":" }"},{"Add":" }"},{"Add":" Ok(())"},{"Add":" }"},{"Add":"}"},{"Add":""},{"Add":"const MAX_BUF_SIZE: usize = 1 << 16;"},{"Add":""},{"Add":"impl thrussh::client::Handler for CiClient {"},{"Add":" type Error = anyhow::Error;"},{"Add":""},{"Add":" fn check_server_key("},{"Add":" self,"},{"Add":" server_public_key: &thrussh_keys::key::PublicKey,"},{"Add":" ) -> impl futures::Future> {"},{"Add":" let valid = self"},{"Add":" .server_public_keys"},{"Add":" .iter()"},{"Add":" .any(|p| p == server_public_key);"},{"Add":" futures::future::ready(Ok((self, valid)))"},{"Add":" }"},{"Add":""},{"Add":" fn adjust_window(&mut self, _channel: thrussh::ChannelId, target: u32) -> u32 {"},{"Add":" let elapsed = self.last_window_adjustment.elapsed().unwrap();"},{"Add":" self.last_window_adjustment = SystemTime::now();"},{"Add":" if target >= 10_000_000 {"},{"Add":" return target;"},{"Add":" }"},{"Add":" if elapsed < Duration::from_secs(2) {"},{"Add":" target * 2"},{"Add":" } else if elapsed > Duration::from_secs(8) {"},{"Add":" target / 2"},{"Add":" } else {"},{"Add":" target"},{"Add":" }"},{"Add":" }"},{"Add":"}"}]}},{"FileAdd":{"meta":{"basename":"lib.rs","metadata":0},"context":[{"change":0,"pos":31}],"contents":[{"Add":"use serde_derive::*;"},{"Add":""},{"Add":"pub const VERSION: u16 = 0;"},{"Add":""},{"Add":"#[derive(Debug, Clone, Serialize, Deserialize)]"},{"Add":"pub struct Job {"},{"Add":" pub id: uuid::Uuid,"},{"Add":" pub repo: uuid::Uuid,"},{"Add":" pub channel: String,"},{"Add":" pub state: libpijul::pristine::Merkle,"},{"Add":" pub extra_patches: Vec,"},{"Add":"}"},{"Add":""},{"Add":"#[derive(Debug, Serialize, Deserialize)]"},{"Add":"pub enum Message {"},{"Add":" Handshake {"},{"Add":" version: u16,"},{"Add":" id: i64,"},{"Add":" },"},{"Add":" Ready,"},{"Add":" Job(Job),"},{"Add":" GetTarball {"},{"Add":" id: uuid::Uuid,"},{"Add":" },"},{"Add":" InvalidInput {"},{"Add":" input: String,"},{"Add":" },"},{"Add":" Log {"},{"Add":" job: Job,"},{"Add":" exit_status: Option,"},{"Add":" path: Option,"},{"Add":" },"},{"Add":" Chunk {"},{"Add":" id: uuid::Uuid,"},{"Add":" stderr: bool,"},{"Add":" len: u32,"},{"Add":" },"},{"Add":"}"}]}},{"FileAdd":{"meta":{"basename":"Cargo.toml","metadata":0},"context":[{"change":0,"pos":14}],"contents":[{"Add":"[package]"},{"Add":"authors = [\"Pierre-Étienne Meunier \"]"},{"Add":"edition = \"2021\""},{"Add":"name = \"ci\""},{"Add":"version = \"0.1.0\""},{"Add":""},{"Add":"[dependencies]"},{"Add":"anyhow = \"1.0.100\""},{"Add":"bincode = \"1.0\""},{"Add":"chrono = \"0.4.42\""},{"Add":"clap = { version = \"4.5.53\", features = [\"derive\"] }"},{"Add":"env_logger = \"0.11.8\""},{"Add":"futures = \"0.3.31\""},{"Add":"log = \"0.4.28\""},{"Add":"serde = \"1.0.228\""},{"Add":"serde_derive = \"1.0.228\""},{"Add":"serde_json = \"1.0.145\""},{"Add":"thrussh = \"0.40\""},{"Add":"thrussh-keys = \"0.23\""},{"Add":""},{"Add":"libpijul = { version = \"1.0.0-beta.10\", features = [\"tarball\"] }"},{"Add":""},{"Add":"tokio = { version = \"1.48\", features = [ \"process\", \"fs\" ] }"},{"Add":"toml = { version = \"0.9.8\" }"},{"Add":"uuid = { version = \"1.21.0\", features = [\"serde\" ] }"},{"Add":"tracing = \"0.1.41\""},{"Add":"tracing-subscriber = { version = \"0.3.20\", features = [\"env-filter\"] }"}]}}],"owner":"pijul","repo":"nest","token":"oVbrJzocO1sJfHqCagXAtlQwuXxsfdaiBnMfiA3xf68=","login":null}

…

Dependencies

Change contents