hydra-evaluator: Rewrite in Rust

Cargo.toml

@@ -1,7 +1,7 @@
 [workspace]
 
 resolver = "2"
-members = ["subprojects/hydra-builder", "subprojects/hydra-queue-runner", "subprojects/crates/*"]
+members = ["subprojects/hydra-builder", "subprojects/hydra-evaluator", "subprojects/hydra-queue-runner", "subprojects/crates/*"]
 
 [workspace.package]
 version = "0.1.0"

flake.nix

@@ -41,9 +41,12 @@
           };
           hydra-linters = self'.callPackage ./subprojects/hydra-linters/package.nix {
           };
-          hydra-queue-runner = self'.callPackage ./subprojects/hydra-queue-runner/package.nix {
+          hydra-rust = self'.callPackage ./subprojects/rust-package.nix {
             inherit nixComponents;
           };
+          hydra-queue-runner = self'.hydra-rust.queue_runner;
+          hydra-builder = self'.hydra-rust.builder;
+          hydra-evaluator = self'.hydra-rust.evaluator;
         });
     in
     rec {

nixos-modules/default.nix

@@ -6,6 +6,8 @@ rec {
     imports = [ ./web-app.nix ];
     services.hydra-dev.package =
       lib.mkDefault flakePackages.${pkgs.stdenv.hostPlatform.system}.hydra;
+    services.hydra-dev.evaluatorExecutable =
+      lib.mkDefault "${flakePackages.${pkgs.stdenv.hostPlatform.system}.hydra-evaluator}/bin/hydra-evaluator";
   };
 
   postgresql = ./postgresql.nix;
@@ -21,14 +23,14 @@ rec {
     _file = ./default.nix;
     imports = [ ./linux-builder-module.nix ];
     services.hydra-queue-builder-dev.package =
-      lib.mkDefault flakePackages.${pkgs.stdenv.hostPlatform.system}.hydra-queue-runner;
+      lib.mkDefault flakePackages.${pkgs.stdenv.hostPlatform.system}.hydra-builder;
   };
 
   darwin-builder = { pkgs, lib, ... }: {
     _file = ./default.nix;
     imports = [ ./darwin-builder-module.nix ];
     services.hydra-queue-builder-dev.package =
-      lib.mkDefault flakePackages.${pkgs.stdenv.hostPlatform.system}.hydra-queue-runner;
+      lib.mkDefault flakePackages.${pkgs.stdenv.hostPlatform.system}.hydra-builder;
   };
 
   hydra = { ... }: {

nixos-modules/web-app.nix

@@ -72,6 +72,11 @@ in
         description = "The Hydra package.";
       };
 
+      evaluatorExecutable = mkOption {
+        type = types.path;
+        description = "Path to the hydra-evaluator executable.";
+      };
+
       hydraURL = mkOption {
         type = types.str;
         description = ''
@@ -261,13 +266,13 @@ in
         requires = [ "hydra-init.service" ];
         restartTriggers = [ hydraConf ];
         after = [ "hydra-init.service" "network.target" ];
-        path = with pkgs; [ hostname-debian cfg.package ];
+        path = with pkgs; [ hostname-debian ];
         environment = env // {
           HYDRA_DBI = "${env.HYDRA_DBI};application_name=hydra-evaluator";
         };
         serviceConfig =
-          { ExecStart = "@${cfg.package}/bin/hydra-evaluator hydra-evaluator";
-            ExecStopPost = "${cfg.package}/bin/hydra-evaluator --unlock";
+          { ExecStart = "@${cfg.evaluatorExecutable} hydra-evaluator";
+            ExecStopPost = "${cfg.evaluatorExecutable} --unlock";
             User = "hydra";
             Restart = "always";
             WorkingDirectory = baseDir;

packaging/dev-shell.nix

@@ -47,7 +47,7 @@ hydra.overrideAttrs (finalAttrs: prevAttrs: {
   shellHook = ''
     pushd $(git rev-parse --show-toplevel) >/dev/null
 
-    PATH=$(pwd)/build/subprojects/hydra/hydra-evaluator:$(pwd)/subprojects/hydra/script:$PATH
+    PATH=$(pwd)/subprojects/hydra/script:$PATH
     PERL5LIB=$(pwd)/subprojects/hydra/lib:$PERL5LIB
     export HYDRA_HOME="$(pwd)/subprojects/hydra/"
     mkdir -p .hydra-data

subprojects/crates/db/src/lib.rs

@@ -36,6 +36,23 @@ impl Database {
         })
     }
 
+    pub async fn new_with_options(
+        options: sqlx::postgres::PgConnectOptions,
+        max_connections: u32,
+    ) -> Result<Self, Error> {
+        Ok(Self {
+            pool: sqlx::postgres::PgPoolOptions::new()
+                .max_connections(max_connections)
+                .connect_with(options)
+                .await?,
+        })
+    }
+
+    #[must_use]
+    pub fn pool(&self) -> &sqlx::PgPool {
+        &self.pool
+    }
+
     pub async fn get(&self) -> Result<Connection, Error> {
         let conn = self.pool.acquire().await?;
         Ok(Connection::new(conn))

subprojects/hydra-evaluator/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "hydra-evaluator"
+version.workspace = true
+edition = "2024"
+license = "GPL-3.0"
+rust-version.workspace = true
+
+[dependencies]
+tracing.workspace = true
+anyhow.workspace = true
+clap = { workspace = true, features = ["derive"] }
+tokio = { workspace = true, features = ["full"] }
+tokio-stream.workspace = true
+futures.workspace = true
+sqlx = { workspace = true, features = [
+  "runtime-tokio",
+  "tls-rustls-ring-webpki",
+  "postgres",
+] }
+
+db = { path = "../crates/db" }
+hydra-tracing = { path = "../crates/tracing" }

subprojects/hydra-evaluator/src/config.rs

@@ -0,0 +1,117 @@
+use std::collections::HashMap;
+
+use anyhow::{Context as _, bail};
+use sqlx::postgres::PgConnectOptions;
+
+#[derive(Debug)]
+pub(crate) struct HydraConfig {
+    options: HashMap<String, String>,
+}
+
+impl HydraConfig {
+    pub(crate) fn load() -> Self {
+        let mut options = HashMap::new();
+
+        let path = match std::env::var("HYDRA_CONFIG") {
+            Ok(p) if !p.is_empty() => p,
+            _ => return Self { options },
+        };
+
+        let contents = match std::fs::read_to_string(&path) {
+            Ok(c) => c,
+            Err(e) => {
+                tracing::warn!("could not read HYDRA_CONFIG at {path}: {e}");
+                return Self { options };
+            }
+        };
+
+        for line in contents.lines() {
+            // Strip comments
+            let line = match line.find('#') {
+                Some(pos) => &line[..pos],
+                None => line,
+            };
+            let line = line.trim();
+
+            let Some(eq) = line.find('=') else {
+                continue;
+            };
+
+            let key = line[..eq].trim();
+            let value = line[eq + 1..].trim();
+
+            if key.is_empty() {
+                continue;
+            }
+
+            options.insert(key.to_owned(), value.to_owned());
+        }
+
+        Self { options }
+    }
+
+    pub(crate) fn get_int(&self, key: &str, default: u64) -> u64 {
+        self.options
+            .get(key)
+            .and_then(|v| v.parse().ok())
+            .unwrap_or(default)
+    }
+}
+
+/// Parse a `HYDRA_DBI` environment variable into `PgConnectOptions`.
+///
+/// Accepts strings like `dbi:Pg:dbname=hydra;host=localhost;port=5432`.
+pub(crate) fn parse_hydra_dbi() -> anyhow::Result<PgConnectOptions> {
+    let dbi = std::env::var("HYDRA_DBI").unwrap_or_else(|_| "dbi:Pg:dbname=hydra;".to_owned());
+    parse_dbi(&dbi)
+}
+
+fn parse_dbi(dbi: &str) -> anyhow::Result<PgConnectOptions> {
+    let params = dbi
+        .strip_prefix("dbi:Pg:")
+        .or_else(|| dbi.strip_prefix("DBI:Pg:"))
+        .context("$HYDRA_DBI does not denote a PostgreSQL database")?;
+
+    let mut opts = PgConnectOptions::new();
+
+    for pair in params.split(';').filter(|s| !s.is_empty()) {
+        let (key, value) = pair
+            .split_once('=')
+            .with_context(|| format!("invalid DBI parameter: {pair}"))?;
+        match key.trim() {
+            "dbname" => opts = opts.database(value.trim()),
+            "host" => opts = opts.host(value.trim()),
+            "port" => {
+                opts = opts.port(
+                    value
+                        .trim()
+                        .parse()
+                        .with_context(|| format!("invalid port: {value}"))?,
+                );
+            }
+            "user" => opts = opts.username(value.trim()),
+            "password" => opts = opts.password(value.trim()),
+            "application_name" => opts = opts.application_name(value.trim()),
+            other => {
+                bail!("unknown DBI parameter: {other}");
+            }
+        }
+    }
+
+    Ok(opts)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_simple_dbi() {
+        parse_dbi("dbi:Pg:dbname=hydra;host=localhost;port=5432").unwrap();
+    }
+
+    #[test]
+    fn parse_dbi_uppercase() {
+        parse_dbi("DBI:Pg:dbname=testdb;").unwrap();
+    }
+}