more patches

.sops.yaml

@@ -11,7 +11,7 @@ keys:
   - &system_brain age1jhf7vm0005j60mjq63696frrmjhpy8kpc2d66mw044lqap5mjv4snmwvwm # cspell:disable-line
   - &system_jeeves age13lmqgc3jvkyah5e3vcwmj4s5wsc2akctcga0lpc0x8v8du3fxprqp4ldkv # cspell:disable-line
   - &system_rhapsody age1ufnewppysaq2wwcl4ugngjz8pfzc5a35yg7luq0qmuqvctajcycs5lf6k4 # cspell:disable-line
-  - &system_portal_1 age1cm7ujz4t9cyc3nyy0w30l2addggk0gy8h3shd9yq47jg845tndnsa3utfl # cspell:disable-line
+  - &system_portal_1 age1vyav6kxtvt3z4vtnkkjj38eu8hlts5m7ygyckhskvalg2gpjk52su53d0a # cspell:disable-line
 
 creation_rules:
   - path_regex: users/secrets\.yaml$
@@ -23,33 +23,3 @@ creation_rules:
           - *system_jeeves
           - *system_rhapsody
           - *system_portal_1
-
-  - path_regex: systems/bob/secrets\.yaml$
-    key_groups:
-      - age:
-          - *admin_richie
-          - *system_bob
-
-  - path_regex: systems/brain/secrets\.yaml$
-    key_groups:
-      - age:
-          - *admin_richie
-          - *system_brain
-
-  - path_regex: systems/jeeves/secrets\.yaml$
-    key_groups:
-      - age:
-          - *admin_richie
-          - *system_jeeves
-
-  - path_regex: systems/rhapsody-in-green/secrets\.yaml$
-    key_groups:
-      - age:
-          - *admin_richie
-          - *system_rhapsody
-
-  - path_regex: systems/portal-1/secrets\.yaml$
-    key_groups:
-      - age:
-          - *admin_richie
-          - *system_portal_1

Cargo.toml

@@ -2,66 +2,3 @@
 resolver = "2"
 
 members = ["rust/*"]
-
-[workspace.dependencies]
-anyhow = "1.0"
-axum = "0.8"
-bytes = "1.10"
-chrono = "0.4"
-clap = { version = "4.5", features = ["derive", "env"] }
-env_logger = "0.11"
-garde = { version = "0.23", features = ["derive", "serde"] }
-hex = "0.4"
-log = "0.4"
-prost = "0.14"
-prost-build = "0.14"
-protoc-bin-vendored = "3.2"
-# HTTP clients use the shared tls crate to install AWS-LC before building a reqwest client.
-reqwest = { version = "0.12", default-features = false, features = ["rustls-tls-webpki-roots-no-provider"] }
-rustls = { version = "0.23", default-features = false, features = ["aws_lc_rs", "std", "tls12"] }
-serde = { version = "1.0", features = ["derive"] }
-serde_json = "1.0"
-sha2 = "0.10"
-signal = { path = "rust/signal" }
-sqlx = { version = "0.9", default-features = false, features = [
-    "chrono",
-    "postgres",
-    "runtime-tokio",
-    "tls-none",
-] }
-subtle = "2.6"
-systemd = { version = "0.10", features = ["journal"] }
-thiserror = "2.0"
-tls = { path = "rust/tls" }
-tokio = { version = "1.47", features = [
-    "macros",
-    "net",
-    "rt-multi-thread",
-    "signal",
-    "time",
-] }
-tokio-retry = "0.3"
-tower-http = { version = "0.6", features = ["limit", "trace"] }
-tracing = "0.1"
-tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
-zfs = { path = "rust/zfs" }
-zstd = "0.13"
-
-[workspace.lints.clippy]
-all = { level = "deny", priority = -1 }
-pedantic = { level = "deny", priority = -1 } # UM, ACTUALLY
-nursery = { level = "deny", priority = -1 }  # BETA LINTS
-# DENY PANICS
-unwrap_used = "deny"
-expect_used = "deny"
-indexing_slicing = "deny"
-arithmetic_side_effects = "deny"
-unreachable = "deny"
-unimplemented = "deny"
-unchecked_time_subtraction = "deny"
-todo = "deny"
-string_slice = "deny"
-panic_in_result_fn = "deny"
-panic = "deny"
-exit = "deny"
-as_conversions = "deny"

clippy.toml

@@ -1,4 +0,0 @@
-allow-unwrap-in-tests = true
-allow-expect-in-tests = true
-allow-panic-in-tests = true
-allow-indexing-slicing-in-tests = true

common/global/default.nix

@@ -21,7 +21,7 @@
 
   boot = {
     tmp.useTmpfs = lib.mkDefault true;
-    kernelPackages = lib.mkDefault pkgs.linuxPackages_6_18;
+    kernelPackages = lib.mkDefault pkgs.linuxPackages_6_12;
   };
 
   hardware.enableRedistributableFirmware = true;

common/optional/desktop.nix

@@ -0,0 +1,42 @@
+{ pkgs, ... }:
+{
+  boot = {
+    kernelPackages = pkgs.linuxPackages_6_18;
+    zfs.package = pkgs.zfs_2_4;
+  };
+
+  hardware.bluetooth = {
+    enable = true;
+    powerOnBoot = true;
+  };
+
+  # rtkit is optional but recommended for pipewire
+  security.rtkit.enable = true;
+
+  services = {
+    displayManager.sddm = {
+      enable = true;
+      wayland.enable = true;
+    };
+
+    desktopManager.plasma6.enable = true;
+
+    xserver = {
+      enable = true;
+      xkb = {
+        layout = "us";
+        variant = "";
+      };
+    };
+
+    pulseaudio.enable = false;
+
+    pipewire = {
+      enable = true;
+      alsa.enable = true;
+      alsa.support32Bit = true;
+      pulse.enable = true;
+      wireplumber.enable = true;
+    };
+  };
+}

common/optional/desktop/default.nix

@@ -1,17 +0,0 @@
-{ pkgs, ... }:
-{
-  imports = [
-    ./kde.nix
-    ./pipewire.nix
-  ];
-
-  boot = {
-    kernelPackages = pkgs.linuxPackages_6_18;
-    zfs.package = pkgs.zfs_2_4;
-  };
-
-  hardware.bluetooth = {
-    enable = true;
-    powerOnBoot = true;
-  };
-}

common/optional/desktop/kde.nix

@@ -1,45 +0,0 @@
-{ pkgs, ... }:
-{
-  environment.plasma6.excludePackages = with pkgs.kdePackages; [
-    aurorae # theme
-    elisa # music player
-    kate # text editor
-    kconfig # config editor
-    khelpcenter # help center
-    kinfocenter # system info
-    konsole # terminal
-    krdp # remote desktop
-    ktexteditor # thing for kate
-    okular # pdf reader
-    plasma-keyboard # used by plasma-keyboard KCM
-    plasma-sdk # plasma development tools
-    plasma-workspace-wallpapers # wallpapers
-    qrca # qr code scanner
-    qtvirtualkeyboard # virtual keyboard
-    union # theme
-  ];
-
-  # disable KDE PIM (Personal Information Management) applications
-  programs.kde-pim.enable = false;
-
-  services = {
-    # removes screen reader and speech dispatcher from the system
-    orca.enable = false;
-    speechd.enable = false;
-
-    displayManager.sddm = {
-      enable = true;
-      wayland.enable = true;
-    };
-
-    desktopManager.plasma6.enable = true;
-
-    xserver = {
-      enable = true;
-      xkb = {
-        layout = "us";
-        variant = "";
-      };
-    };
-  };
-}

common/optional/desktop/pipewire.nix

@@ -1,16 +0,0 @@
-{
-  # rtkit is optional but recommended for pipewire
-  security.rtkit.enable = true;
-
-  services = {
-    pulseaudio.enable = false;
-
-    pipewire = {
-      enable = true;
-      alsa.enable = true;
-      alsa.support32Bit = true;
-      pulse.enable = true;
-      wireplumber.enable = true;
-    };
-  };
-}

common/optional/steam.nix

@@ -1,6 +1,7 @@
 { pkgs, ... }:
 {
   environment.systemPackages = with pkgs; [
+    mangohud
     steam-run
   ];
   hardware.steam-hardware.enable = true;

common/optional/x86-64-v3/patches/cuda/default.nix

@@ -0,0 +1,19 @@
+{ prev }:
+let
+  buildRedistHook = prev.runCommand "buildRedistHook.bash" { } ''
+    cp ${prev.path}/pkgs/development/cuda-modules/buildRedist/buildRedistHook.bash "$out"
+    chmod u+w "$out"
+    patch "$out" < ${./propagated-build-output-arrays.patch}
+  '';
+in
+prev.cudaPackages_12_9.overrideScope (
+  _cudaFinal: cudaPrev: {
+    buildRedist =
+      args:
+      (cudaPrev.buildRedist args).overrideAttrs (old: {
+        nativeBuildInputs = map (
+          input: if builtins.baseNameOf input == "buildRedistHook.bash" then buildRedistHook else input
+        ) old.nativeBuildInputs;
+      });
+  }
+)

common/optional/x86-64-v3/patches/cuda/propagated-build-output-arrays.patch

@@ -0,0 +1,44 @@
+Subject: [PATCH] Preserve structured CUDA output arrays
+
+The multiple-outputs hook now accepts arrays. The old compatibility export
+collapses include and lib into one array element, breaking indirect expansion.
+Keep the array and iterate over each output when propagating to out.
+
+--- a/buildRedistHook.bash
++++ b/buildRedistHook.bash
+@@ -23,9 +23,6 @@
+ 
+   postInstallCheckHooks+=(checkCudaNonEmptyOutputs)
+   nixLog "added checkCudaNonEmptyOutputs to postInstallCheckHooks"
+-
+-  preFixupHooks+=(fixupPropagatedBuildOutputsForMultipleOutputs)
+-  nixLog "added fixupPropagatedBuildOutputsForMultipleOutputs to preFixupHooks"
+ 
+   postFixupHooks+=(fixupCudaPropagatedBuildOutputsToOut)
+   nixLog "added fixupCudaPropagatedBuildOutputsToOut to postFixupHooks"
+@@ -193,16 +190,6 @@
+   return 0
+ }
+ 
+-# TODO(@connorbaker): https://github.com/NixOS/nixpkgs/issues/323126.
+-# _multioutPropagateDev() currently expects a space-separated string rather than an array.
+-# NOTE: Because _multioutPropagateDev is a postFixup hook, we correct it in preFixup.
+-fixupPropagatedBuildOutputsForMultipleOutputs() {
+-  nixLog "converting propagatedBuildOutputs to a space-separated string"
+-  # shellcheck disable=SC2124
+-  export propagatedBuildOutputs="${propagatedBuildOutputs[@]}"
+-  return 0
+-}
+-
+ # The multiple outputs setup hook only propagates build outputs to dev.
+ # We want to propagate them to out as well, in case the user interpolates
+ # the package into a string -- in such a case, the dev output is not selected
+@@ -217,7 +204,7 @@
+   mkdir -p "${out:?}/nix-support"
+ 
+   # NOTE: We must use printWords to ensure the output is a single line.
+-  for output in $propagatedBuildOutputs; do
++  for output in "${propagatedBuildOutputs[@]}"; do
+     # Propagate the other components to the out output
+     nixLog "adding ${!output:?} to propagatedBuildInputs of ${out:?}"
+     printWords "${!output:?}" >>"${out:?}/nix-support/propagated-build-inputs"

common/optional/x86-64-v3/patches/default.nix

@@ -1,6 +1,8 @@
 _final: prev:
 (import ./abseil { inherit prev; })
 // {
+  cudaPackages_12_9 = import ./cuda { inherit prev; };
+  trunk = import ./trunk { inherit (prev) trunk jq; };
   gnutls = import ./gnutls { inherit (prev) gnutls; };
   prometheus = import ./prometheus { inherit (prev) prometheus; };
   zopfli = import ./zopfli { inherit (prev) zopfli; };

common/optional/x86-64-v3/patches/trunk/default.nix

@@ -0,0 +1,21 @@
+{ trunk, jq }:
+trunk.overrideAttrs (old: {
+  nativeBuildInputs = (old.nativeBuildInputs or [ ]) ++ [ jq ];
+
+  # Patch the bundled libdeflate before Cargo validates its vendor checksums.
+  postPatch = (old.postPatch or "") + ''
+    libdeflateVendors=( "$cargoDepsCopy"/source-*/libdeflate-sys-1.23.1 )
+    if [ "''${#libdeflateVendors[@]}" -ne 1 ] || [ ! -d "''${libdeflateVendors[0]}" ]; then
+      echo "Expected exactly one vendored libdeflate-sys 1.23.1" >&2
+      exit 1
+    fi
+    libdeflateVendor="''${libdeflateVendors[0]}"
+    patch -d "$libdeflateVendor/libdeflate" -p1 < ${./libdeflate-gcc16.patch}
+    libdeflateHeader=libdeflate/lib/x86/cpu_features.h
+    libdeflateChecksum=$(sha256sum "$libdeflateVendor/$libdeflateHeader" | cut -d ' ' -f 1)
+    jq --arg path "$libdeflateHeader" --arg hash "$libdeflateChecksum" \
+      '.files[$path] = $hash' "$libdeflateVendor/.cargo-checksum.json" \
+      > "$libdeflateVendor/.cargo-checksum.json.new"
+    mv "$libdeflateVendor/.cargo-checksum.json.new" "$libdeflateVendor/.cargo-checksum.json"
+  '';
+})

common/optional/x86-64-v3/patches/trunk/libdeflate-gcc16.patch

@@ -0,0 +1,33 @@
+From 2a3762cf736aff9aa5dda329fdec3c08a816eb9b Mon Sep 17 00:00:00 2001
+From: Sergei Trofimovich <slyich@gmail.com>
+Date: Tue, 20 May 2025 21:18:28 +0100
+Subject: [PATCH] lib/x86/cpu_features.h: drop evex512 on gcc-16
+
+gcc-16 `master` branch dropped support for `evex512` as:
+    https://gcc.gnu.org/git/?p=gcc.git;a=commitdiff;h=c052a6f4a1c803cb92147ff98fb91cf3511e0856
+
+As a result libdeflate build started failing as:
+
+    In file included from /build/source/lib/adler32.c:28:
+    /build/source/lib/x86/adler32_template.h:135:12: error: attribute 'target' argument 'no-evex512' is unknown
+      135 | ADD_SUFFIX(reduce_to_32bits)(vec_t v_s1, vec_t v_s2, u32 *s1_p, u32 *s2_p)
+          |            ^~~~~~~~~~~~~~~~
+
+The change adds upper build to `evex512` for `gcc`.
+---
+ lib/x86/cpu_features.h | 2 +-
+ 1 file changed, 1 insertion(+), 1 deletion(-)
+
+diff --git a/lib/x86/cpu_features.h b/lib/x86/cpu_features.h
+index fbfdb0eb..f9d69d72 100644
+--- a/lib/x86/cpu_features.h
++++ b/lib/x86/cpu_features.h
+@@ -165,7 +165,7 @@ static inline u32 get_x86_cpu_features(void) { return 0; }
+ #  define HAVE_AVXVNNI(features)	((features) & X86_CPU_FEATURE_AVXVNNI)
+ #endif
+ 
+-#if (GCC_PREREQ(14, 0) || CLANG_PREREQ(18, 0, 18000000)) \
++#if ((GCC_PREREQ(14, 0) && !(GCC_PREREQ(16, 0))) || CLANG_PREREQ(18, 0, 18000000)) \
+ 	&& !defined(__EVEX512__) /* avoid subtracting the evex512 feature */
+ #  define EVEX512	",evex512"	/* needed to override potential -mno-evex512 */
+ #  define NO_EVEX512	",no-evex512"

common/optional/zfs/default.nix

@@ -1,4 +1,5 @@
 {
+  inputs,
   lib,
   pkgs,
   ...
@@ -14,6 +15,7 @@
   services = {
     snapshot_manager = {
       enable = lib.mkDefault true;
+      PYTHONPATH = "${inputs.self}/";
     };
 
     zfs = {

common/optional/zfs/snapshot.nix

@@ -1,4 +1,5 @@
 {
+  inputs,
   pkgs,
   lib,
   config,
@@ -6,18 +7,23 @@
 }:
 let
   cfg = config.services.snapshot_manager;
-  snapshotManager = pkgs.callPackage ../../../rust/snapshot_manager/package.nix { };
+  snapshotManagerPackages = ps: with ps; [ typer ];
 in
 {
   options = {
     services.snapshot_manager = {
       enable = lib.mkEnableOption "ZFS snapshot manager";
-      signalAlerts = lib.mkEnableOption "Signal failure alerts for the snapshot manager";
       path = lib.mkOption {
         type = lib.types.path;
         default = ./snapshot_config.toml;
         description = "Path to the snapshot_manager TOML config.";
       };
+      PYTHONPATH = lib.mkOption {
+        type = lib.types.str;
+        description = ''
+          the PYTHONPATH to use for the snapshot_manager service.
+        '';
+      };
       EnvironmentFile = lib.mkOption {
         type = lib.types.nullOr (lib.types.coercedTo lib.types.path toString lib.types.str);
         default = null;
@@ -30,7 +36,40 @@ in
   };
 
   config = lib.mkIf cfg.enable {
-    system.checks = [ snapshotManager ];
+    nixpkgs.overlays = [
+      (final: _prev: {
+        snapshot_manager_python = final.python314.withPackages snapshotManagerPackages;
+        snapshot_manager_test_python = final.python314.withPackages (
+          ps:
+          snapshotManagerPackages ps
+          ++ (with ps; [
+            pyfakefs
+            pytest
+            pytest-asyncio
+            pytest-mock
+            pytest-xdist
+          ])
+        );
+        snapshot_manager_tests =
+          final.runCommand "snapshot-manager-tests"
+            {
+              nativeBuildInputs = [ final.snapshot_manager_test_python ];
+            }
+            ''
+              export HOME="$TMPDIR"
+              cd ${inputs.self}
+              pytest \
+                -o cache_dir="$TMPDIR/pytest-cache" \
+                tests/test_common.py \
+                tests/test_signal_alert.py \
+                tests/test_snapshot_manager.py \
+                tests/test_zfs.py
+              touch "$out"
+            '';
+      })
+    ];
+
+    system.checks = [ pkgs.snapshot_manager_tests ];
 
     systemd = {
       services.snapshot_manager = {
@@ -38,9 +77,12 @@ in
         requires = [ "zfs-import.target" ];
         after = [ "zfs-import.target" ];
         path = [ pkgs.zfs ];
+        environment = {
+          PYTHONPATH = cfg.PYTHONPATH;
+        };
         serviceConfig = {
           Type = "oneshot";
-          ExecStart = "${snapshotManager}/bin/snapshot-manager ${lib.escapeShellArg cfg.path} --signal-alerts ${lib.boolToString cfg.signalAlerts}";
+          ExecStart = "${pkgs.snapshot_manager_python}/bin/python -m python.tools.snapshot_manager ${lib.escapeShellArg cfg.path}";
         }
         // lib.optionalAttrs (cfg.EnvironmentFile != null) {
           EnvironmentFile = cfg.EnvironmentFile;

flake.lock

@@ -45,11 +45,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1791293500,
-        "narHash": "sha256-3hwJd2/vG/VYV1P/g7oDgdbNsWODhN5F//Od+hPMm2k=",
+        "lastModified": 1790819692,
+        "narHash": "sha256-rvubAPbh3QCliic3sAK+CXTrPCGgJstY52OTs/5H1J4=",
         "owner": "nix-community",
         "repo": "home-manager",
-        "rev": "0e11b7bcba1d74ef8ea10bfdfb1e1a5811a21087",
+        "rev": "c8ecc29e5175452bee4a2aa1ba2383856d795338",
         "type": "github"
       },
       "original": {
@@ -67,11 +67,11 @@
         "rust-overlay": "rust-overlay"
       },
       "locked": {
-        "lastModified": 1791127101,
-        "narHash": "sha256-SmrBBhcv9iNHA2vlafi7Zpe8bkcsMjVNSmCCVf5/jrQ=",
+        "lastModified": 1790427786,
+        "narHash": "sha256-8+Ip0HFW2oZiRr0zCZmPPIHfTdxxHy4CmSF1uOkM3Bw=",
         "owner": "LLukas22",
         "repo": "Jellyswarrm",
-        "rev": "0b4301c6c74efff9c3634296913032579c98faf6",
+        "rev": "9e67c5ad12116add867ba6c1bf69f767879e9147",
         "type": "github"
       },
       "original": {
@@ -85,11 +85,11 @@
         "nixpkgs": "nixpkgs"
       },
       "locked": {
-        "lastModified": 1791128201,
-        "narHash": "sha256-rlsYY/Dg/LhbIj51Aa6TLvRpZeb3KZgUTd7llQ5XcJE=",
+        "lastModified": 1790722812,
+        "narHash": "sha256-XwFt1uLyaBCQU1/nUeGqeBnqNXPVgGRnz20YVYdmF6Y=",
         "owner": "nixos",
         "repo": "nixos-hardware",
-        "rev": "31cc5f4d9b9ba601071e8b8504601b9b176e2756",
+        "rev": "06f9ecaea5f64b6ff61cf42cb32f21621c4fa14a",
         "type": "github"
       },
       "original": {
@@ -114,11 +114,11 @@
     },
     "nixpkgs-master": {
       "locked": {
-        "lastModified": 1791296764,
-        "narHash": "sha256-N2b6Uxdr1MFhscEPqDyntTTtGZASgq3XYE0ujRzfeNI=",
+        "lastModified": 1790862528,
+        "narHash": "sha256-00jiK7QCLGuVe4mj3kDa6P/iDFKb87aK9ZtJDZCOvTc=",
         "owner": "nixos",
         "repo": "nixpkgs",
-        "rev": "a15afac8691bd24f746255ffbd76306693e4cf7a",
+        "rev": "86f3f30260edec38090733d0667e818e6e2ea6f5",
         "type": "github"
       },
       "original": {
@@ -146,11 +146,11 @@
     },
     "nixpkgs_2": {
       "locked": {
-        "lastModified": 1791260994,
-        "narHash": "sha256-Miqqk/ammqnTUxaoCyvtwoPeLbI1ksFyLxi/CazZpWY=",
+        "lastModified": 1790689126,
+        "narHash": "sha256-ilerN1WLSvF+HMjziC/Wv99J5y02maDH+6hZPwsORKg=",
         "owner": "nixos",
         "repo": "nixpkgs",
-        "rev": "151fa4e8ddfdd8dd25d945ad94ed54a13de9f6e4",
+        "rev": "b4fd65b198c599cbe814fcb9f42d25d021595ec9",
         "type": "github"
       },
       "original": {
@@ -201,11 +201,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1791103873,
-        "narHash": "sha256-nFxM+pKoZ8LJAEnUXARyCaOAloWgaW9kZQOSjWzKcTE=",
+        "lastModified": 1790498116,
+        "narHash": "sha256-rs9meAYxW3zzrh43yaW7htrqCD+X9+pupDPHN86fumI=",
         "owner": "Mic92",
         "repo": "sops-nix",
-        "rev": "dcd241ba97088c22569d1573286e1b9daad340c0",
+        "rev": "5efb5a6f4f5ab192817d28557dd4d650fa14d866",
         "type": "github"
       },
       "original": {

python/tools/snapshot_manager.py

@@ -0,0 +1,161 @@
+"""snapshot_manager."""
+
+from __future__ import annotations
+
+import logging
+import sys
+import tomllib
+from functools import cache
+from os import getenv
+from pathlib import Path  # noqa: TC003 This is required for the typer CLI
+from re import compile as re_compile
+from re import search
+
+import typer
+
+from python.common import configure_logger, utcnow
+from python.signal_alert import signal_alert
+from python.zfs import Dataset, get_datasets
+
+logger = logging.getLogger(__name__)
+
+
+def main(config_file: Path) -> None:
+    """Main."""
+    configure_logger(level="DEBUG")
+    logger.info("Starting snapshot_manager")
+
+    from_phone = getenv("SIGNAL_ALERT_FROM_PHONE")
+    to_phone = getenv("SIGNAL_ALERT_TO_PHONE")
+    if not from_phone:
+        error = "SIGNAL_ALERT_FROM_PHONE environment variable is not set"
+        raise ValueError(error)
+    if not to_phone:
+        error = "SIGNAL_ALERT_TO_PHONE environment variable is not set"
+        raise ValueError(error)
+
+    try:
+        time_stamp = get_time_stamp()
+
+        for dataset in get_datasets():
+            status = dataset.create_snapshot(time_stamp)
+            logger.debug(f"{status=}")
+            if status != "snapshot created":
+                msg = f"{dataset.name} failed to create snapshot {time_stamp}"
+                logger.error(msg)
+                signal_alert(msg, from_phone, to_phone)
+                continue
+            count_lookup = get_count_lookup(config_file, dataset.name)
+            logger.info(f"using {count_lookup} for {dataset.name}")
+            get_snapshots_to_delete(dataset, count_lookup, from_phone, to_phone)
+    except Exception:
+        logger.exception("snapshot_manager failed")
+        signal_alert("snapshot_manager failed", from_phone, to_phone)
+        sys.exit(1)
+    else:
+        logger.info("snapshot_manager completed")
+
+
+def get_count_lookup(config_file: Path, dataset_name: str) -> dict[str, int]:
+    """Get the count lookup.
+
+    Args:
+        config_file (Path): The path to the configuration file.
+        dataset_name (str): The name of the dataset.
+
+    Returns:
+        dict[str, int]: The count lookup.
+    """
+    config_data = load_config_data(config_file)
+
+    return config_data.get(dataset_name, get_default_config(config_data))
+
+
+def get_default_config(config_data: dict[str, dict[str, int]]) -> dict[str, int]:
+    """Get the default configuration.
+
+    Args:
+        config_data (dict[str, dict[str, int]]): The configuration data.
+
+    Returns:
+        dict[str, int]: The default configuration.
+    """
+    return config_data.get(
+        "default",
+        {"15_min": 4, "hourly": 12, "daily": 0, "monthly": 0},
+    )
+
+
+@cache
+def load_config_data(config_file: Path) -> dict[str, dict[str, int]]:
+    """Load a TOML configuration file.
+
+    Args:
+        config_file (Path): The path to the configuration file.
+
+    Returns:
+        dict: The configuration data.
+    """
+    return tomllib.loads(config_file.read_text())
+
+
+def get_snapshots_to_delete(
+    dataset: Dataset,
+    count_lookup: dict[str, int],
+    from_phone: str,
+    to_phone: str,
+) -> None:
+    """Get snapshots to delete.
+
+    Args:
+        dataset (Dataset): the dataset
+        count_lookup (dict[str, int]): the count lookup
+        from_phone (str): The Signal account sending alerts.
+        to_phone (str): The Signal account receiving alerts.
+    """
+    snapshots = dataset.get_snapshots()
+
+    logger.info(f"calculating snapshots for {dataset.name} to be deleted")
+    if not snapshots:
+        logger.info(f"{dataset.name} has no snapshots")
+        return
+
+    filters = (
+        ("15_min", re_compile(r"auto_\d{10}(?:15|30|45)")),
+        ("hourly", re_compile(r"auto_\d{8}(?!00)\d{2}00")),
+        ("daily", re_compile(r"auto_\d{6}(?!01)\d{2}0000")),
+        ("monthly", re_compile(r"auto_\d{6}010000")),
+    )
+
+    for filter_name, snapshot_filter in filters:
+        logger.debug(f"{filter_name=}\n{snapshot_filter=}")
+
+        filtered_snapshots = sorted(snapshot.name for snapshot in snapshots if search(snapshot_filter, snapshot.name))
+
+        logger.debug(f"{filtered_snapshots=}")
+
+        snapshots_wanted = count_lookup[filter_name]
+        snapshots_being_deleted = filtered_snapshots[:-snapshots_wanted] if snapshots_wanted > 0 else filtered_snapshots
+
+        logger.info(f"{snapshots_being_deleted} are being deleted")
+        for snapshot in snapshots_being_deleted:
+            if error := dataset.delete_snapshot(snapshot):
+                error_message = f"{dataset.name}@{snapshot} failed to delete: {error}"
+                signal_alert(error_message, from_phone, to_phone)
+                logger.error(error_message)
+
+
+def get_time_stamp() -> str:
+    """Get the time stamp."""
+    now = utcnow()
+    nearest_15_min = now.replace(minute=(now.minute - (now.minute % 15)))
+    return nearest_15_min.strftime("auto_%Y%m%d%H%M")
+
+
+def cli() -> None:
+    """CLI."""
+    typer.run(main)
+
+
+if __name__ == "__main__":
+    cli()

rust/haproxy_ingest_proto/Cargo.toml

@@ -1,17 +0,0 @@
-[package]
-name = "haproxy-ingest-proto"
-version = "0.1.0"
-edition = "2024"
-description = "Shared protocol types and limits for HAProxy ingestion"
-license = "MIT"
-build = "build.rs"
-
-[lints]
-workspace = true
-
-[dependencies]
-prost.workspace = true
-
-[build-dependencies]
-prost-build.workspace = true
-protoc-bin-vendored.workspace = true

rust/haproxy_ingest_proto/README.md

@@ -1,96 +0,0 @@
-# HAProxy cloud log ingestion
-
-This workspace is split into three crates:
-
-- `haproxy-ingest-proto` owns the shared Protobuf schema and protocol limits.
-- `haproxy-log-shipper` is a short-lived client launched by a systemd timer on
-  `portal-1`. It reads only local `haproxy.service` journal records and drains
-  them in cursor order.
-- `haproxy-ingest-server` is the private HTTP API on Jeeves. It authenticates
-  the source, commits records to PostgreSQL, and advances the journal cursor
-  atomically.
-
-The wire contract is in `proto/haproxy_ingest.proto`; the initial PostgreSQL
-schema is in `../haproxy_ingest_server/migrations/001_initial.sql`.
-
-## One-time setup
-
-The shipper endpoint is supplied by the `haproxy-log-shipper` service as
-`http://jeeves.taile39c31.ts.net:8444`; the environment file only needs the
-shared token. Ingestion uses HTTP over Tailscale and requires no certificates.
-
-Create one random token and install the same value on the shipper and server.
-A VM rebuild with a new `/etc/machine-id` is still treated as a new log source.
-
-On `portal-1`, create `/var/lib/haproxy-log-shipper/environment`:
-
-```text
-HAPROXY_INGEST_TOKEN=<at-least-32-character-random-token>
-```
-
-Set the file ownership and mode to `root:root` and `0600`.
-
-On Jeeves, create
-`/zfs/storage/secrets/services/haproxy-ingest-server`:
-
-```text
-HAPROXY_INGEST_TOKEN=<the-same-random-token>
-```
-
-Set this file to `root:root` and `0600`. systemd loads it before switching the
-API to its unprivileged service account.
-
-A suitable token can be generated without writing it to shell history:
-
-```console
-openssl rand -hex 32
-```
-
-Deploy Jeeves before `portal-1`, then verify:
-
-```console
-systemctl status haproxy-ingest-server.service
-systemctl status haproxy-log-shipper.timer
-journalctl -u haproxy-log-shipper.service
-```
-
-The server listens over IPv4 on TCP 8444, which Jeeves' firewall permits only
-through `tailscale0`. Both service sandboxes disable IPv6. Tailscale encrypts
-traffic between the hosts. The bearer token authorizes access to the ingestion API;
-the machine ID identifies a log source but is not an authorization boundary.
-
-## Protocol behavior
-
-- Maximum sender batch: 1,000 records or 20 MiB serialized Protobuf.
-- Maximum receiver request: 24 MiB compressed, 20 MiB decompressed, 10,000 records.
-- Compression: one Zstandard frame over the complete Protobuf body.
-- Request timeouts: 3 seconds to connect and 10 seconds total.
-- Retry delays: 1, 2, and 4 seconds plus bounded jitter.
-- `409`: fetch the authoritative server cursor and rebuild.
-- `413`: fail because a sender batch within the configured limits was rejected.
-- `422`: replace only the reported offending record with a durable rejection.
-- Ambiguous failures: query the status endpoint before retrying.
-
-Journal retention on `portal-1` is persistent, bounded to 512 MiB, and retained
-for up to 14 days. If an acknowledged cursor has already been vacuumed, the
-shipper records a `JournalGap` before resuming from the oldest retained matching
-entry.
-
-## Database inspection
-
-The API owns the `haproxy_ingest` database and creates daily
-`haproxy_logs_YYYYMMDD` partitions before using chunked SQLx inserts.
-Useful checks include:
-
-```sql
-SELECT machine_id, hostname, last_journal_timestamp, last_commit_time
-FROM ingest_source_state;
-
-SELECT journal_timestamp, client_ip, http_method, request_path, status_code
-FROM haproxy_logs
-ORDER BY journal_timestamp DESC
-LIMIT 100;
-
-SELECT * FROM ingest_journal_gaps ORDER BY detected_at DESC;
-SELECT * FROM ingest_rejected_events ORDER BY ingest_time DESC;
-```

rust/haproxy_ingest_proto/build.rs

@@ -1,11 +0,0 @@
-fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let protoc = protoc_bin_vendored::protoc_bin_path()?;
-    unsafe {
-        std::env::set_var("PROTOC", protoc);
-    }
-    let mut config = prost_build::Config::new();
-    config.type_attribute(".haproxy.ingest.v1.LogBatch", "#[derive(Eq)]");
-    config.compile_protos(&["proto/haproxy_ingest.proto"], &["proto"])?;
-    println!("cargo:rerun-if-changed=proto/haproxy_ingest.proto");
-    Ok(())
-}

rust/haproxy_ingest_proto/proto/haproxy_ingest.proto

@@ -1,71 +0,0 @@
-syntax = "proto3";
-
-package haproxy.ingest.v1;
-
-message LogBatch {
-  uint32 schema_version = 1;
-  string machine_id = 2;
-  string hostname = 3;
-  optional string previous_cursor = 4;
-  string last_cursor = 5;
-  int64 first_journal_timestamp_us = 6;
-  int64 last_journal_timestamp_us = 7;
-  repeated LogRecord records = 8;
-}
-
-message LogRecord {
-  int64 journal_timestamp_us = 1;
-
-  oneof payload {
-    HAProxyEvent event = 2;
-    RejectedEvent rejected = 3;
-    JournalGap gap = 4;
-  }
-}
-
-message HAProxyEvent {
-  reserved 22;
-
-  optional string client_ip = 1;
-  optional uint32 client_port = 2;
-  optional string frontend_name = 3;
-  optional string backend_name = 4;
-  optional string server_name = 5;
-  optional string http_method = 6;
-  optional string request_path = 7;
-  optional string http_version = 8;
-  int32 status_code = 9;
-  uint64 bytes_sent = 10;
-  int32 time_request_ms = 11;
-  int32 time_wait_ms = 12;
-  int32 time_connect_ms = 13;
-  int32 time_response_ms = 14;
-  int32 time_total_ms = 15;
-  optional string termination_state = 16;
-  uint32 retries = 17;
-  optional string tls_sni = 18;
-  optional string tls_version = 19;
-  optional string tls_cipher = 20;
-  optional bool tls_resumed = 21;
-  repeated string normalization_flags = 23;
-  optional uint32 active_connections = 24;
-  optional uint32 frontend_connections = 25;
-  optional uint32 backend_connections = 26;
-  optional uint32 server_connections = 27;
-  optional uint32 server_queue = 28;
-  optional uint32 backend_queue = 29;
-}
-
-message RejectedEvent {
-  string reason = 1;
-  bytes message_sha256 = 2;
-  optional string raw_message_prefix = 3;
-  bool raw_message_truncated = 4;
-}
-
-message JournalGap {
-  string expected_cursor = 1;
-  string first_available_cursor = 2;
-  optional int64 first_available_timestamp_us = 3;
-  string reason = 4;
-}

rust/haproxy_ingest_proto/src/lib.rs

@@ -1,12 +0,0 @@
-pub mod proto {
-    #![allow(clippy::large_enum_variant)]
-
-    include!(concat!(env!("OUT_DIR"), "/haproxy.ingest.v1.rs"));
-}
-
-pub const SCHEMA_VERSION: u32 = 1;
-pub const MAX_BATCH_RECORDS: usize = 1_000;
-pub const MAX_BATCH_PROTO_BYTES: usize = 20 * 1024 * 1024;
-pub const MAX_COMPRESSED_BYTES: usize = 24 * 1024 * 1024;
-pub const MAX_DECOMPRESSED_BYTES: usize = MAX_BATCH_PROTO_BYTES;
-pub const SERVER_MAX_RECORDS: usize = 10_000;

rust/haproxy_ingest_server/Cargo.toml

@@ -1,29 +0,0 @@
-[package]
-name = "haproxy-ingest-server"
-version = "0.1.0"
-edition = "2024"
-description = "HAProxy log ingestion API"
-license = "MIT"
-
-[lints]
-workspace = true
-
-[dependencies]
-anyhow.workspace = true
-axum.workspace = true
-bytes.workspace = true
-chrono = { workspace = true, features = ["serde"] }
-clap.workspace = true
-haproxy-ingest-proto = { path = "../haproxy_ingest_proto" }
-hex.workspace = true
-prost.workspace = true
-serde.workspace = true
-sha2.workspace = true
-sqlx.workspace = true
-subtle.workspace = true
-thiserror.workspace = true
-tokio.workspace = true
-tower-http.workspace = true
-tracing.workspace = true
-tracing-subscriber.workspace = true
-zstd.workspace = true

rust/haproxy_ingest_server/migrations/001_initial.sql

@@ -1,102 +0,0 @@
-CREATE TABLE IF NOT EXISTS ingest_source_state (
-    machine_id                TEXT PRIMARY KEY,
-    hostname                  TEXT NOT NULL,
-    last_cursor               TEXT,
-    last_journal_timestamp    TIMESTAMPTZ,
-    last_source_ip            INET,
-    last_commit_time          TIMESTAMPTZ
-);
-
-CREATE TABLE IF NOT EXISTS haproxy_logs (
-    id                        BIGINT GENERATED ALWAYS AS IDENTITY,
-    machine_id                TEXT NOT NULL,
-    hostname                  TEXT NOT NULL,
-    source_ip                 INET NOT NULL,
-    journal_timestamp         TIMESTAMPTZ NOT NULL,
-    request_time              TIMESTAMPTZ,
-    client_ip                 INET,
-    client_port               INTEGER,
-    frontend_name             TEXT,
-    backend_name              TEXT,
-    server_name               TEXT,
-    http_method               TEXT,
-    request_path              TEXT,
-    http_version              TEXT,
-    status_code               INTEGER NOT NULL,
-    bytes_sent                BIGINT NOT NULL,
-    time_request_ms           INTEGER NOT NULL,
-    time_wait_ms              INTEGER NOT NULL,
-    time_connect_ms           INTEGER NOT NULL,
-    time_response_ms          INTEGER NOT NULL,
-    time_total_ms             INTEGER NOT NULL,
-    termination_state         TEXT,
-    retries                   INTEGER NOT NULL,
-    tls_sni                   TEXT,
-    tls_version               TEXT,
-    tls_cipher                TEXT,
-    tls_resumed               BOOLEAN,
-    normalization_flags       TEXT[] NOT NULL DEFAULT '{}',
-    active_connections        INTEGER,
-    frontend_connections      INTEGER,
-    backend_connections       INTEGER,
-    server_connections        INTEGER,
-    server_queue              INTEGER,
-    backend_queue             INTEGER,
-    ingest_time               TIMESTAMPTZ NOT NULL DEFAULT now(),
-    PRIMARY KEY (journal_timestamp, id)
-) PARTITION BY RANGE (journal_timestamp);
-
-CREATE INDEX IF NOT EXISTS haproxy_logs_machine_time_idx
-ON haproxy_logs (machine_id, journal_timestamp);
-
-CREATE INDEX IF NOT EXISTS haproxy_logs_status_time_idx
-ON haproxy_logs (status_code, journal_timestamp);
-
-CREATE TABLE IF NOT EXISTS ingest_rejected_events (
-    id                        BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
-    machine_id                TEXT NOT NULL,
-    hostname                  TEXT NOT NULL,
-    journal_timestamp         TIMESTAMPTZ NOT NULL,
-    reason                    TEXT NOT NULL,
-    message_sha256            BYTEA NOT NULL,
-    raw_message_prefix        TEXT,
-    raw_message_truncated     BOOLEAN NOT NULL,
-    source_ip                 INET NOT NULL,
-    ingest_time               TIMESTAMPTZ NOT NULL DEFAULT now()
-);
-
-CREATE INDEX IF NOT EXISTS ingest_rejected_events_machine_time_idx
-ON ingest_rejected_events (machine_id, journal_timestamp);
-
-CREATE TABLE IF NOT EXISTS ingest_journal_gaps (
-    id                         BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
-    machine_id                 TEXT NOT NULL,
-    hostname                   TEXT NOT NULL,
-    expected_cursor            TEXT NOT NULL,
-    first_available_cursor     TEXT NOT NULL,
-    first_available_timestamp  TIMESTAMPTZ,
-    reason                     TEXT NOT NULL,
-    source_ip                  INET NOT NULL,
-    detected_at                TIMESTAMPTZ NOT NULL DEFAULT now()
-);
-
-CREATE INDEX IF NOT EXISTS ingest_journal_gaps_machine_time_idx
-ON ingest_journal_gaps (machine_id, detected_at);
-
-CREATE TABLE IF NOT EXISTS ingest_batches (
-    batch_id                   TEXT PRIMARY KEY,
-    machine_id                 TEXT NOT NULL,
-    hostname                   TEXT NOT NULL,
-    source_ip                  INET NOT NULL,
-    previous_cursor            TEXT,
-    last_cursor                TEXT NOT NULL,
-    first_journal_timestamp    TIMESTAMPTZ,
-    last_journal_timestamp     TIMESTAMPTZ,
-    record_count               INTEGER NOT NULL,
-    compressed_bytes           BIGINT NOT NULL,
-    uncompressed_bytes         BIGINT NOT NULL,
-    received_at                TIMESTAMPTZ NOT NULL DEFAULT now()
-);
-
-CREATE INDEX IF NOT EXISTS ingest_batches_machine_time_idx
-ON ingest_batches (machine_id, received_at);

rust/haproxy_ingest_server/package.nix

@@ -1,19 +0,0 @@
-{
-  lib,
-  rustPlatform,
-}:
-rustPlatform.buildRustPackage {
-  pname = "haproxy-ingest-server";
-  version = "0.1.0";
-  src = import ../rust-source.nix { inherit lib; };
-
-  cargoLock.lockFile = ../../Cargo.lock;
-  cargoBuildFlags = [ "-p haproxy-ingest-server" ];
-  cargoTestFlags = [ "-p haproxy-ingest-server" ];
-
-  meta = {
-    description = "HAProxy log ingestion API";
-    license = lib.licenses.mit;
-    platforms = lib.platforms.linux;
-  };
-}

rust/haproxy_ingest_server/src/db.rs

@@ -1,376 +0,0 @@
-//! `SQLx` statements and Rust-to-PostgreSQL value conversion.
-
-use std::net::IpAddr;
-
-use anyhow::{Context, Result};
-use chrono::{DateTime, NaiveDate, Utc};
-use haproxy_ingest_proto::proto::{HaProxyEvent, JournalGap, LogBatch, RejectedEvent, log_record};
-use sha2::{Digest, Sha256};
-use sqlx::postgres::PgRow;
-use sqlx::{AssertSqlSafe, Decode, PgPool, Postgres, QueryBuilder, Row, Transaction, Type};
-
-use crate::{ApiError, timestamp};
-
-const MIGRATION: &str = include_str!("../migrations/001_initial.sql");
-const EVENT_INSERT_CHUNK_SIZE: usize = 1_000;
-
-pub struct SourceStatus {
-    pub hostname: Option<String>,
-    pub last_cursor: Option<String>,
-    pub last_journal_timestamp: Option<DateTime<Utc>>,
-    pub last_commit_time: Option<DateTime<Utc>>,
-}
-
-pub async fn apply_ingestion_schema(pool: &PgPool) -> Result<()> {
-    sqlx::raw_sql(MIGRATION)
-        .execute(pool)
-        .await
-        .context("failed to apply ingestion schema")?;
-    Ok(())
-}
-
-pub async fn select_source_status(
-    pool: &PgPool,
-    machine_id: &str,
-) -> Result<Option<SourceStatus>, ApiError> {
-    let row = sqlx::query(
-        "SELECT hostname, last_cursor, last_journal_timestamp, last_commit_time \
-         FROM ingest_source_state WHERE machine_id = $1",
-    )
-    .bind(machine_id)
-    .fetch_optional(pool)
-    .await
-    .map_err(ApiError::internal)?;
-    row.map(|row| {
-        Ok(SourceStatus {
-            hostname: get(&row, "hostname")?,
-            last_cursor: get(&row, "last_cursor")?,
-            last_journal_timestamp: get(&row, "last_journal_timestamp")?,
-            last_commit_time: get(&row, "last_commit_time")?,
-        })
-    })
-    .transpose()
-}
-
-fn get<'r, T>(row: &'r PgRow, column: &str) -> Result<T, ApiError>
-where
-    T: Decode<'r, Postgres> + Type<Postgres>,
-{
-    row.try_get(column).map_err(ApiError::internal)
-}
-
-pub async fn set_lock_timeout(transaction: &mut Transaction<'_, Postgres>) -> Result<(), ApiError> {
-    sqlx::query("SET LOCAL lock_timeout = '5s'")
-        .execute(&mut **transaction)
-        .await
-        .map_err(ApiError::internal)?;
-    Ok(())
-}
-
-pub async fn set_statement_timeout(
-    transaction: &mut Transaction<'_, Postgres>,
-) -> Result<(), ApiError> {
-    sqlx::query("SET LOCAL statement_timeout = '8s'")
-        .execute(&mut **transaction)
-        .await
-        .map_err(ApiError::internal)?;
-    Ok(())
-}
-
-pub async fn insert_source_state(
-    transaction: &mut Transaction<'_, Postgres>,
-    batch: &LogBatch,
-) -> Result<(), ApiError> {
-    sqlx::query(
-        "INSERT INTO ingest_source_state (machine_id, hostname) VALUES ($1, $2) \
-         ON CONFLICT (machine_id) DO NOTHING",
-    )
-    .bind(&batch.machine_id)
-    .bind(&batch.hostname)
-    .execute(&mut **transaction)
-    .await
-    .map_err(ApiError::internal)?;
-    Ok(())
-}
-
-pub async fn select_committed_cursor(
-    transaction: &mut Transaction<'_, Postgres>,
-    machine_id: &str,
-) -> Result<Option<String>, ApiError> {
-    sqlx::query_scalar(
-        "SELECT last_cursor FROM ingest_source_state WHERE machine_id = $1 FOR UPDATE",
-    )
-    .bind(machine_id)
-    .fetch_one(&mut **transaction)
-    .await
-    .map_err(ApiError::internal)
-}
-
-pub async fn create_daily_partition(
-    transaction: &mut Transaction<'_, Postgres>,
-    date: NaiveDate,
-    next: NaiveDate,
-) -> Result<(), ApiError> {
-    let name = partition_name(date);
-    let statement = format!(
-        "CREATE TABLE IF NOT EXISTS {name} PARTITION OF haproxy_logs \
-         FOR VALUES FROM ('{date}') TO ('{next}')"
-    );
-    sqlx::query(AssertSqlSafe(statement.as_str()))
-        .execute(&mut **transaction)
-        .await
-        .map_err(ApiError::internal)?;
-    Ok(())
-}
-
-fn partition_name(date: NaiveDate) -> String {
-    format!("haproxy_logs_{}", date.format("%Y%m%d"))
-}
-
-struct PreparedEvent<'a> {
-    event: &'a HaProxyEvent,
-    journal_timestamp: DateTime<Utc>,
-    client_ip: Option<String>,
-    client_port: Option<i32>,
-    bytes_sent: i64,
-    retries: i32,
-    active_connections: Option<i32>,
-    frontend_connections: Option<i32>,
-    backend_connections: Option<i32>,
-    server_connections: Option<i32>,
-    server_queue: Option<i32>,
-    backend_queue: Option<i32>,
-}
-
-fn prepare_events(batch: &LogBatch) -> Result<Vec<PreparedEvent<'_>>, ApiError> {
-    let mut prepared = Vec::new();
-    for record in &batch.records {
-        let Some(log_record::Payload::Event(event)) = &record.payload else {
-            continue;
-        };
-        prepared.push(PreparedEvent {
-            event,
-            journal_timestamp: timestamp(record.journal_timestamp_us)?,
-            client_ip: event
-                .client_ip
-                .as_deref()
-                .and_then(|value| value.parse::<IpAddr>().ok())
-                .map(|value| value.to_string()),
-            client_port: event
-                .client_port
-                .map(i32::try_from)
-                .transpose()
-                .map_err(ApiError::internal)?,
-            bytes_sent: i64::try_from(event.bytes_sent).map_err(ApiError::internal)?,
-            retries: i32::try_from(event.retries).map_err(ApiError::internal)?,
-            active_connections: sql_i32(event.active_connections).map_err(ApiError::internal)?,
-            frontend_connections: sql_i32(event.frontend_connections)
-                .map_err(ApiError::internal)?,
-            backend_connections: sql_i32(event.backend_connections).map_err(ApiError::internal)?,
-            server_connections: sql_i32(event.server_connections).map_err(ApiError::internal)?,
-            server_queue: sql_i32(event.server_queue).map_err(ApiError::internal)?,
-            backend_queue: sql_i32(event.backend_queue).map_err(ApiError::internal)?,
-        });
-    }
-    Ok(prepared)
-}
-
-#[allow(clippy::too_many_lines)]
-pub async fn insert_events(
-    transaction: &mut Transaction<'_, Postgres>,
-    batch: &LogBatch,
-    source_ip: IpAddr,
-) -> Result<(), ApiError> {
-    let prepared = prepare_events(batch)?;
-    let source_ip = source_ip.to_string();
-    for chunk in prepared.chunks(EVENT_INSERT_CHUNK_SIZE) {
-        let mut query = QueryBuilder::<Postgres>::new(
-            "INSERT INTO haproxy_logs (machine_id,hostname,source_ip,journal_timestamp,client_ip,\
-             client_port,frontend_name,backend_name,server_name,http_method,request_path,http_version,\
-             status_code,bytes_sent,time_request_ms,time_wait_ms,time_connect_ms,time_response_ms,\
-             time_total_ms,termination_state,retries,tls_sni,tls_version,tls_cipher,tls_resumed,\
-             normalization_flags,active_connections,frontend_connections,\
-             backend_connections,server_connections,server_queue,backend_queue) ",
-        );
-        query.push_values(chunk, |mut values, row| {
-            values
-                .push_bind(&batch.machine_id)
-                .push_bind(&batch.hostname)
-                .push_bind(source_ip.as_str())
-                .push_unseparated("::inet")
-                .push_bind(row.journal_timestamp)
-                .push_bind(row.client_ip.as_deref())
-                .push_unseparated("::inet")
-                .push_bind(row.client_port)
-                .push_bind(row.event.frontend_name.as_deref())
-                .push_bind(row.event.backend_name.as_deref())
-                .push_bind(row.event.server_name.as_deref())
-                .push_bind(row.event.http_method.as_deref())
-                .push_bind(row.event.request_path.as_deref())
-                .push_bind(row.event.http_version.as_deref())
-                .push_bind(row.event.status_code)
-                .push_bind(row.bytes_sent)
-                .push_bind(row.event.time_request_ms)
-                .push_bind(row.event.time_wait_ms)
-                .push_bind(row.event.time_connect_ms)
-                .push_bind(row.event.time_response_ms)
-                .push_bind(row.event.time_total_ms)
-                .push_bind(row.event.termination_state.as_deref())
-                .push_bind(row.retries)
-                .push_bind(row.event.tls_sni.as_deref())
-                .push_bind(row.event.tls_version.as_deref())
-                .push_bind(row.event.tls_cipher.as_deref())
-                .push_bind(row.event.tls_resumed)
-                .push_bind(row.event.normalization_flags.as_slice())
-                .push_bind(row.active_connections)
-                .push_bind(row.frontend_connections)
-                .push_bind(row.backend_connections)
-                .push_bind(row.server_connections)
-                .push_bind(row.server_queue)
-                .push_bind(row.backend_queue);
-        });
-        query
-            .build()
-            .execute(&mut **transaction)
-            .await
-            .map_err(ApiError::internal)?;
-    }
-    Ok(())
-}
-
-pub async fn insert_rejected_event(
-    transaction: &mut Transaction<'_, Postgres>,
-    batch: &LogBatch,
-    source_ip: IpAddr,
-    journal_timestamp: DateTime<Utc>,
-    rejected: &RejectedEvent,
-) -> Result<(), ApiError> {
-    sqlx::query(
-        "INSERT INTO ingest_rejected_events (machine_id,hostname,journal_timestamp,reason,\
-         message_sha256,raw_message_prefix,raw_message_truncated,source_ip) \
-         VALUES ($1,$2,$3,$4,$5,$6,$7,$8::inet)",
-    )
-    .bind(&batch.machine_id)
-    .bind(&batch.hostname)
-    .bind(journal_timestamp)
-    .bind(&rejected.reason)
-    .bind(&rejected.message_sha256)
-    .bind(&rejected.raw_message_prefix)
-    .bind(rejected.raw_message_truncated)
-    .bind(source_ip.to_string())
-    .execute(&mut **transaction)
-    .await
-    .map_err(ApiError::internal)?;
-    Ok(())
-}
-
-pub async fn insert_journal_gap(
-    transaction: &mut Transaction<'_, Postgres>,
-    batch: &LogBatch,
-    source_ip: IpAddr,
-    first_available_timestamp: Option<DateTime<Utc>>,
-    gap: &JournalGap,
-) -> Result<(), ApiError> {
-    sqlx::query(
-        "INSERT INTO ingest_journal_gaps (machine_id,hostname,expected_cursor,\
-         first_available_cursor,first_available_timestamp,reason,source_ip) \
-         VALUES ($1,$2,$3,$4,$5,$6,$7::inet)",
-    )
-    .bind(&batch.machine_id)
-    .bind(&batch.hostname)
-    .bind(&gap.expected_cursor)
-    .bind(&gap.first_available_cursor)
-    .bind(first_available_timestamp)
-    .bind(&gap.reason)
-    .bind(source_ip.to_string())
-    .execute(&mut **transaction)
-    .await
-    .map_err(ApiError::internal)?;
-    Ok(())
-}
-
-pub async fn insert_ingest_batch(
-    transaction: &mut Transaction<'_, Postgres>,
-    batch: &LogBatch,
-    source_ip: IpAddr,
-    compressed_bytes: usize,
-    uncompressed_bytes: usize,
-) -> Result<(), ApiError> {
-    let first_timestamp = timestamp(batch.first_journal_timestamp_us)?;
-    let last_timestamp = timestamp(batch.last_journal_timestamp_us)?;
-    let batch_id = batch_id(batch);
-    let record_count = i32::try_from(batch.records.len()).map_err(ApiError::internal)?;
-    let compressed_bytes = i64::try_from(compressed_bytes).map_err(ApiError::internal)?;
-    let uncompressed_bytes = i64::try_from(uncompressed_bytes).map_err(ApiError::internal)?;
-    sqlx::query(
-        "INSERT INTO ingest_batches (batch_id, machine_id, hostname, source_ip, previous_cursor, \
-         last_cursor, first_journal_timestamp, last_journal_timestamp, record_count, \
-         compressed_bytes, uncompressed_bytes) \
-         VALUES ($1,$2,$3,$4::inet,$5,$6,$7,$8,$9,$10,$11) ON CONFLICT (batch_id) DO NOTHING",
-    )
-    .bind(&batch_id)
-    .bind(&batch.machine_id)
-    .bind(&batch.hostname)
-    .bind(source_ip.to_string())
-    .bind(&batch.previous_cursor)
-    .bind(&batch.last_cursor)
-    .bind(first_timestamp)
-    .bind(last_timestamp)
-    .bind(record_count)
-    .bind(compressed_bytes)
-    .bind(uncompressed_bytes)
-    .execute(&mut **transaction)
-    .await
-    .map_err(ApiError::internal)?;
-    Ok(())
-}
-
-pub async fn update_source_state(
-    transaction: &mut Transaction<'_, Postgres>,
-    batch: &LogBatch,
-    source_ip: IpAddr,
-) -> Result<(), ApiError> {
-    let last_timestamp = timestamp(batch.last_journal_timestamp_us)?;
-    sqlx::query(
-        "UPDATE ingest_source_state SET hostname=$2, last_cursor=$3, \
-         last_journal_timestamp=$4, last_source_ip=$5::inet, last_commit_time=now() \
-         WHERE machine_id=$1",
-    )
-    .bind(&batch.machine_id)
-    .bind(&batch.hostname)
-    .bind(&batch.last_cursor)
-    .bind(last_timestamp)
-    .bind(source_ip.to_string())
-    .execute(&mut **transaction)
-    .await
-    .map_err(ApiError::internal)?;
-    Ok(())
-}
-
-fn batch_id(batch: &LogBatch) -> String {
-    let mut hash = Sha256::new();
-    hash.update(batch.machine_id.as_bytes());
-    hash.update([0]);
-    if let Some(previous) = &batch.previous_cursor {
-        hash.update(previous.as_bytes());
-    }
-    hash.update([0]);
-    hash.update(batch.last_cursor.as_bytes());
-    hex::encode(hash.finalize())
-}
-
-fn sql_i32(value: Option<u32>) -> Result<Option<i32>, std::num::TryFromIntError> {
-    value.map(i32::try_from).transpose()
-}
-
-#[cfg(test)]
-#[allow(clippy::unwrap_used)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn partition_names_are_date_only() {
-        let date = NaiveDate::from_ymd_opt(2026, 8, 29).unwrap();
-        assert_eq!(partition_name(date), "haproxy_logs_20260829");
-    }
-}

rust/haproxy_ingest_server/src/main.rs

@@ -1,530 +0,0 @@
-use std::{
-    collections::BTreeSet,
-    io::Read,
-    net::{IpAddr, SocketAddr},
-};
-
-use anyhow::{Context, Result, bail};
-use axum::{
-    Router,
-    body::Bytes,
-    extract::{ConnectInfo, DefaultBodyLimit, Path, State},
-    http::{HeaderMap, StatusCode, header},
-    response::{IntoResponse, Response},
-    routing::{get as get_route, post},
-};
-use chrono::{DateTime, Days, Utc};
-use clap::Parser;
-use haproxy_ingest_proto::{
-    MAX_COMPRESSED_BYTES, MAX_DECOMPRESSED_BYTES,
-    proto::{LogBatch, log_record},
-};
-use prost::Message;
-use serde::Serialize;
-use sha2::{Digest, Sha256};
-use sqlx::{PgPool, Postgres, Transaction, postgres::PgPoolOptions};
-use subtle::ConstantTimeEq;
-use tower_http::trace::TraceLayer;
-use tracing::{info, warn};
-use tracing_subscriber::EnvFilter;
-
-mod db;
-mod validation;
-
-use validation::{ValidationError, validate_batch};
-
-#[derive(Debug, Parser)]
-#[command(about, version)]
-struct Args {
-    #[arg(long, env = "HAPROXY_INGEST_LISTEN", default_value = "0.0.0.0:8444")]
-    listen: SocketAddr,
-
-    #[arg(long, env = "HAPROXY_INGEST_DATABASE_URL", hide_env_values = true)]
-    database_url: String,
-
-    #[arg(long, env = "HAPROXY_INGEST_TOKEN", hide_env_values = true)]
-    token: String,
-}
-
-#[derive(Clone)]
-struct AppState {
-    pool: PgPool,
-    auth: SourceToken,
-}
-
-#[derive(Clone)]
-struct SourceToken {
-    hash: [u8; 32],
-}
-
-impl SourceToken {
-    fn parse(value: &str) -> Result<Self> {
-        let token = value.trim();
-        if token.len() < 32 {
-            bail!("source token must be at least 32 characters");
-        }
-        Ok(Self {
-            hash: Sha256::digest(token.as_bytes()).into(),
-        })
-    }
-
-    fn authenticate(&self, headers: &HeaderMap) -> Result<(), ApiError> {
-        let authorization = headers
-            .get(header::AUTHORIZATION)
-            .and_then(|value| value.to_str().ok())
-            .and_then(|value| value.strip_prefix("Bearer "))
-            .ok_or_else(|| ApiError::new(StatusCode::UNAUTHORIZED, "missing bearer token"))?;
-        let presented: [u8; 32] = Sha256::digest(authorization.as_bytes()).into();
-        (presented.ct_eq(&self.hash).unwrap_u8() == 1)
-            .then_some(())
-            .ok_or_else(|| ApiError::new(StatusCode::UNAUTHORIZED, "invalid bearer token"))
-    }
-}
-
-#[derive(Debug, Serialize)]
-struct StatusResponse {
-    machine_id: String,
-    hostname: Option<String>,
-    last_cursor: Option<String>,
-    last_journal_timestamp: Option<DateTime<Utc>>,
-    last_commit_time: Option<DateTime<Utc>>,
-}
-
-#[derive(Debug, Serialize)]
-struct CommitResponse {
-    machine_id: String,
-    last_cursor: String,
-    records_committed: usize,
-    already_committed: bool,
-}
-
-#[derive(Debug, Serialize)]
-struct ErrorResponse {
-    error: String,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    record_index: Option<usize>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    last_cursor: Option<String>,
-}
-
-#[derive(Debug)]
-struct ApiError {
-    status: StatusCode,
-    message: String,
-    record_index: Option<usize>,
-    last_cursor: Option<String>,
-}
-
-impl ApiError {
-    fn new(status: StatusCode, message: impl Into<String>) -> Self {
-        Self {
-            status,
-            message: message.into(),
-            record_index: None,
-            last_cursor: None,
-        }
-    }
-
-    fn invalid(error: &ValidationError) -> Self {
-        Self {
-            status: StatusCode::UNPROCESSABLE_ENTITY,
-            record_index: error.record_index(),
-            message: error.to_string(),
-            last_cursor: None,
-        }
-    }
-
-    fn conflict(last_cursor: Option<String>) -> Self {
-        Self {
-            status: StatusCode::CONFLICT,
-            message: "batch previous_cursor does not match the committed cursor".to_owned(),
-            record_index: None,
-            last_cursor,
-        }
-    }
-
-    fn internal(error: impl std::fmt::Display) -> Self {
-        warn!(error = %error, "ingestion request failed");
-        Self::new(
-            StatusCode::SERVICE_UNAVAILABLE,
-            "ingestion backend unavailable",
-        )
-    }
-}
-
-impl IntoResponse for ApiError {
-    fn into_response(self) -> Response {
-        (
-            self.status,
-            axum::Json(ErrorResponse {
-                error: self.message,
-                record_index: self.record_index,
-                last_cursor: self.last_cursor,
-            }),
-        )
-            .into_response()
-    }
-}
-
-#[derive(Debug, Eq, PartialEq)]
-enum CursorDecision {
-    Accept,
-    AlreadyCommitted,
-    Conflict,
-}
-
-#[tokio::main]
-async fn main() -> Result<()> {
-    tracing_subscriber::fmt()
-        .json()
-        .with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()))
-        .init();
-    let args = Args::parse();
-    let pool = PgPoolOptions::new()
-        .max_connections(8)
-        .connect(&args.database_url)
-        .await
-        .context("failed to connect to PostgreSQL")?;
-
-    db::apply_ingestion_schema(&pool).await?;
-
-    let state = AppState {
-        pool,
-        auth: SourceToken::parse(&args.token)?,
-    };
-
-    let app = Router::new()
-        .route("/v1/logs/status/{machine_id}", get_route(get_status))
-        .route("/v1/logs/batch", post(post_batch))
-        .layer(DefaultBodyLimit::max(MAX_COMPRESSED_BYTES))
-        .layer(TraceLayer::new_for_http())
-        .with_state(state);
-    let listener = tokio::net::TcpListener::bind(args.listen)
-        .await
-        .context("failed to bind ingestion HTTP listener")?;
-
-    info!(listen = %args.listen, "HAProxy ingestion API listening over HTTP");
-    axum::serve(
-        listener,
-        app.into_make_service_with_connect_info::<SocketAddr>(),
-    )
-    .await
-    .context("ingestion server failed")
-}
-
-async fn get_status(
-    State(state): State<AppState>,
-    Path(machine_id): Path<String>,
-    headers: HeaderMap,
-) -> Result<axum::Json<StatusResponse>, ApiError> {
-    state.auth.authenticate(&headers)?;
-    let status = db::select_source_status(&state.pool, &machine_id).await?;
-    let response = match status {
-        None => StatusResponse {
-            machine_id: machine_id.clone(),
-            hostname: None,
-            last_cursor: None,
-            last_journal_timestamp: None,
-            last_commit_time: None,
-        },
-        Some(status) => StatusResponse {
-            machine_id: machine_id.clone(),
-            hostname: status.hostname,
-            last_cursor: status.last_cursor,
-            last_journal_timestamp: status.last_journal_timestamp,
-            last_commit_time: status.last_commit_time,
-        },
-    };
-    Ok(axum::Json(response))
-}
-
-async fn post_batch(
-    State(state): State<AppState>,
-    ConnectInfo(peer): ConnectInfo<SocketAddr>,
-    headers: HeaderMap,
-    body: Bytes,
-) -> Result<axum::Json<CommitResponse>, ApiError> {
-    state.auth.authenticate(&headers)?;
-    check_protobuf(&headers)?;
-    check_zstd(&headers)?;
-    let compressed_bytes = body.len();
-    let decoded = decompress_bounded(&body)?;
-    let uncompressed_bytes = decoded.len();
-    let batch = LogBatch::decode(decoded.as_slice()).map_err(|error| {
-        ApiError::new(
-            StatusCode::BAD_REQUEST,
-            format!("invalid protobuf: {error}"),
-        )
-    })?;
-    validate_batch(&batch).map_err(|error| ApiError::invalid(&error))?;
-    let response = commit_batch(
-        &state.pool,
-        &batch,
-        peer.ip(),
-        compressed_bytes,
-        uncompressed_bytes,
-    )
-    .await?;
-    Ok(axum::Json(response))
-}
-
-fn check_protobuf(headers: &HeaderMap) -> Result<(), ApiError> {
-    if headers
-        .get(header::CONTENT_TYPE)
-        .and_then(|value| value.to_str().ok())
-        != Some("application/x-protobuf")
-    {
-        return Err(ApiError::new(
-            StatusCode::UNSUPPORTED_MEDIA_TYPE,
-            "Content-Type must be application/x-protobuf",
-        ));
-    }
-    Ok(())
-}
-
-fn check_zstd(headers: &HeaderMap) -> Result<(), ApiError> {
-    if headers
-        .get(header::CONTENT_ENCODING)
-        .and_then(|value| value.to_str().ok())
-        != Some("zstd")
-    {
-        return Err(ApiError::new(
-            StatusCode::UNSUPPORTED_MEDIA_TYPE,
-            "Content-Encoding must be zstd",
-        ));
-    }
-    Ok(())
-}
-
-fn decompress_bounded(body: &[u8]) -> Result<Vec<u8>, ApiError> {
-    let decoder = zstd::stream::read::Decoder::new(body).map_err(|error| {
-        ApiError::new(
-            StatusCode::BAD_REQUEST,
-            format!("invalid zstd body: {error}"),
-        )
-    })?;
-    let mut output = Vec::new();
-    let read_limit = u64::try_from(MAX_DECOMPRESSED_BYTES)
-        .map_err(ApiError::internal)?
-        .saturating_add(1);
-    decoder
-        .take(read_limit)
-        .read_to_end(&mut output)
-        .map_err(|error| {
-            ApiError::new(
-                StatusCode::BAD_REQUEST,
-                format!("invalid zstd body: {error}"),
-            )
-        })?;
-    if output.len() > MAX_DECOMPRESSED_BYTES {
-        return Err(ApiError::new(
-            StatusCode::PAYLOAD_TOO_LARGE,
-            "decompressed protobuf exceeds 20 MiB",
-        ));
-    }
-    Ok(output)
-}
-
-async fn commit_batch(
-    pool: &PgPool,
-    batch: &LogBatch,
-    source_ip: IpAddr,
-    compressed_bytes: usize,
-    uncompressed_bytes: usize,
-) -> Result<CommitResponse, ApiError> {
-    let mut transaction = pool.begin().await.map_err(ApiError::internal)?;
-    db::set_lock_timeout(&mut transaction).await?;
-    db::set_statement_timeout(&mut transaction).await?;
-    db::insert_source_state(&mut transaction, batch).await?;
-    let committed_cursor = db::select_committed_cursor(&mut transaction, &batch.machine_id).await?;
-    match classify_cursor_chain(
-        committed_cursor.as_deref(),
-        batch.previous_cursor.as_deref(),
-        &batch.last_cursor,
-    ) {
-        CursorDecision::Accept => {}
-        CursorDecision::AlreadyCommitted => {
-            transaction.commit().await.map_err(ApiError::internal)?;
-            return Ok(CommitResponse {
-                machine_id: batch.machine_id.clone(),
-                last_cursor: batch.last_cursor.clone(),
-                records_committed: batch.records.len(),
-                already_committed: true,
-            });
-        }
-        CursorDecision::Conflict => return Err(ApiError::conflict(committed_cursor)),
-    }
-
-    ensure_daily_partitions(&mut transaction, batch).await?;
-    db::insert_events(&mut transaction, batch, source_ip).await?;
-    insert_non_event_records(&mut transaction, batch, source_ip).await?;
-    db::insert_ingest_batch(
-        &mut transaction,
-        batch,
-        source_ip,
-        compressed_bytes,
-        uncompressed_bytes,
-    )
-    .await?;
-    db::update_source_state(&mut transaction, batch, source_ip).await?;
-    transaction.commit().await.map_err(ApiError::internal)?;
-    info!(
-        machine_id = %batch.machine_id,
-        records = batch.records.len(),
-        last_cursor = %batch.last_cursor,
-        "committed HAProxy log batch"
-    );
-    Ok(CommitResponse {
-        machine_id: batch.machine_id.clone(),
-        last_cursor: batch.last_cursor.clone(),
-        records_committed: batch.records.len(),
-        already_committed: false,
-    })
-}
-
-async fn ensure_daily_partitions(
-    transaction: &mut Transaction<'_, Postgres>,
-    batch: &LogBatch,
-) -> Result<(), ApiError> {
-    let mut dates = BTreeSet::new();
-    for record in &batch.records {
-        if matches!(record.payload, Some(log_record::Payload::Event(_))) {
-            dates.insert(timestamp(record.journal_timestamp_us)?.date_naive());
-        }
-    }
-    for date in dates {
-        let next = date.checked_add_days(Days::new(1)).ok_or_else(|| {
-            ApiError::new(StatusCode::UNPROCESSABLE_ENTITY, "partition date overflow")
-        })?;
-        db::create_daily_partition(transaction, date, next).await?;
-    }
-    Ok(())
-}
-
-async fn insert_non_event_records(
-    transaction: &mut Transaction<'_, Postgres>,
-    batch: &LogBatch,
-    source_ip: IpAddr,
-) -> Result<(), ApiError> {
-    for record in &batch.records {
-        match &record.payload {
-            Some(log_record::Payload::Rejected(rejected)) => {
-                let journal_timestamp = timestamp(record.journal_timestamp_us)?;
-                db::insert_rejected_event(
-                    transaction,
-                    batch,
-                    source_ip,
-                    journal_timestamp,
-                    rejected,
-                )
-                .await?;
-            }
-            Some(log_record::Payload::Gap(gap)) => {
-                let first_available_timestamp = gap
-                    .first_available_timestamp_us
-                    .map(timestamp)
-                    .transpose()?;
-                db::insert_journal_gap(
-                    transaction,
-                    batch,
-                    source_ip,
-                    first_available_timestamp,
-                    gap,
-                )
-                .await?;
-            }
-            Some(log_record::Payload::Event(_)) | None => {}
-        }
-    }
-    Ok(())
-}
-
-fn timestamp(micros: i64) -> Result<DateTime<Utc>, ApiError> {
-    DateTime::from_timestamp_micros(micros).ok_or_else(|| {
-        ApiError::new(
-            StatusCode::UNPROCESSABLE_ENTITY,
-            "journal timestamp is outside supported range",
-        )
-    })
-}
-
-fn classify_cursor_chain(
-    committed: Option<&str>,
-    previous: Option<&str>,
-    last: &str,
-) -> CursorDecision {
-    if committed == Some(last) {
-        CursorDecision::AlreadyCommitted
-    } else if committed == previous {
-        CursorDecision::Accept
-    } else {
-        CursorDecision::Conflict
-    }
-}
-
-#[cfg(test)]
-#[allow(clippy::unwrap_used)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn starts_without_certificate_arguments() {
-        Args::try_parse_from([
-            "haproxy-ingest-server",
-            "--database-url",
-            "postgres://localhost/haproxy_ingest",
-            "--token",
-            "0123456789abcdef0123456789abcdef",
-        ])
-        .unwrap();
-    }
-
-    #[test]
-    fn shared_token_authenticates_requests() {
-        let token = SourceToken::parse("0123456789abcdef0123456789abcdef").unwrap();
-        let mut headers = HeaderMap::new();
-        headers.insert(
-            header::AUTHORIZATION,
-            "Bearer 0123456789abcdef0123456789abcdef".parse().unwrap(),
-        );
-        assert!(token.authenticate(&headers).is_ok());
-        headers.insert(
-            header::AUTHORIZATION,
-            "Bearer fedcba9876543210fedcba9876543210".parse().unwrap(),
-        );
-        assert!(token.authenticate(&headers).is_err());
-    }
-
-    #[test]
-    fn decompression_limit_is_enforced() {
-        let input = vec![0_u8; MAX_DECOMPRESSED_BYTES + 1];
-        let compressed = zstd::stream::encode_all(input.as_slice(), 1).unwrap();
-        assert!(matches!(
-            decompress_bounded(&compressed),
-            Err(ApiError {
-                status: StatusCode::PAYLOAD_TOO_LARGE,
-                ..
-            })
-        ));
-    }
-
-    #[test]
-    fn cursor_chain_handles_bootstrap_continuation_duplicate_and_conflict() {
-        assert_eq!(
-            classify_cursor_chain(None, None, "c1"),
-            CursorDecision::Accept
-        );
-        assert_eq!(
-            classify_cursor_chain(Some("c1"), Some("c1"), "c2"),
-            CursorDecision::Accept
-        );
-        assert_eq!(
-            classify_cursor_chain(Some("c2"), Some("c1"), "c2"),
-            CursorDecision::AlreadyCommitted
-        );
-        assert_eq!(
-            classify_cursor_chain(Some("other"), Some("c1"), "c2"),
-            CursorDecision::Conflict
-        );
-    }
-}

rust/haproxy_ingest_server/src/validation.rs

@@ -1,214 +0,0 @@
-use std::net::IpAddr;
-
-use chrono::{DateTime, Utc};
-use haproxy_ingest_proto::{
-    SCHEMA_VERSION, SERVER_MAX_RECORDS,
-    proto::{LogBatch, LogRecord, log_record},
-};
-use thiserror::Error;
-
-#[derive(Debug, Error)]
-pub enum ValidationError {
-    #[error("unsupported schema_version {0}")]
-    SchemaVersion(u32),
-    #[error("machine_id must be 32 lowercase hexadecimal characters")]
-    MachineId,
-    #[error("hostname is empty or longer than 255 bytes")]
-    Hostname,
-    #[error("last_cursor is empty")]
-    LastCursor,
-    #[error("batch contains no records or more than {SERVER_MAX_RECORDS} records")]
-    RecordCount,
-    #[error("batch journal timestamp bounds are invalid")]
-    TimestampBounds,
-    #[error("record {index}: {reason}")]
-    Record { index: usize, reason: String },
-}
-
-impl ValidationError {
-    #[must_use]
-    pub const fn record_index(&self) -> Option<usize> {
-        match self {
-            Self::Record { index, .. } => Some(*index),
-            _ => None,
-        }
-    }
-}
-
-/// Validates protocol-level and per-record invariants before database access.
-///
-/// # Errors
-///
-/// Returns [`ValidationError`] when the batch cannot be committed safely.
-pub fn validate_batch(batch: &LogBatch) -> Result<(), ValidationError> {
-    if batch.schema_version != SCHEMA_VERSION {
-        return Err(ValidationError::SchemaVersion(batch.schema_version));
-    }
-    if batch.machine_id.len() != 32
-        || !batch
-            .machine_id
-            .bytes()
-            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
-    {
-        return Err(ValidationError::MachineId);
-    }
-    if batch.hostname.is_empty() || batch.hostname.len() > 255 {
-        return Err(ValidationError::Hostname);
-    }
-    if batch.last_cursor.is_empty() {
-        return Err(ValidationError::LastCursor);
-    }
-    if batch.records.is_empty() || batch.records.len() > SERVER_MAX_RECORDS {
-        return Err(ValidationError::RecordCount);
-    }
-    // Journal cursors define record order; wall-clock timestamps may move backward.
-    if batch
-        .records
-        .first()
-        .map(|record| record.journal_timestamp_us)
-        != Some(batch.first_journal_timestamp_us)
-        || batch
-            .records
-            .last()
-            .map(|record| record.journal_timestamp_us)
-            != Some(batch.last_journal_timestamp_us)
-    {
-        return Err(ValidationError::TimestampBounds);
-    }
-
-    for (index, record) in batch.records.iter().enumerate() {
-        validate_record(index, record)?;
-    }
-    Ok(())
-}
-
-fn validate_record(index: usize, record: &LogRecord) -> Result<(), ValidationError> {
-    if DateTime::<Utc>::from_timestamp_micros(record.journal_timestamp_us).is_none() {
-        return record_error(
-            index,
-            "journal timestamp is outside PostgreSQL's supported range",
-        );
-    }
-    let Some(payload) = &record.payload else {
-        return record_error(index, "payload is absent");
-    };
-    match payload {
-        log_record::Payload::Event(event) => {
-            if event.bytes_sent > i64::MAX.cast_unsigned() {
-                return record_error(index, "bytes_sent exceeds PostgreSQL BIGINT");
-            }
-            if event.retries > i32::MAX.cast_unsigned()
-                || [
-                    event.active_connections,
-                    event.frontend_connections,
-                    event.backend_connections,
-                    event.server_connections,
-                    event.server_queue,
-                    event.backend_queue,
-                ]
-                .into_iter()
-                .flatten()
-                .any(|value| value > i32::MAX.cast_unsigned())
-            {
-                return record_error(index, "counter exceeds PostgreSQL INTEGER");
-            }
-            if event.client_port.is_some_and(|port| port > u16::MAX.into()) {
-                return record_error(index, "client_port exceeds 65535");
-            }
-            if let Some(ip) = &event.client_ip
-                && ip.parse::<IpAddr>().is_err()
-            {
-                return record_error(index, "client_ip is invalid");
-            }
-        }
-        log_record::Payload::Rejected(rejected) => {
-            if rejected.reason.is_empty() || rejected.message_sha256.len() != 32 {
-                return record_error(index, "rejected event reason or SHA-256 is invalid");
-            }
-        }
-        log_record::Payload::Gap(gap) => {
-            if gap.expected_cursor.is_empty()
-                || gap.first_available_cursor.is_empty()
-                || gap.reason.is_empty()
-            {
-                return record_error(index, "journal gap fields are incomplete");
-            }
-        }
-    }
-    Ok(())
-}
-
-fn record_error<T>(index: usize, reason: &str) -> Result<T, ValidationError> {
-    Err(ValidationError::Record {
-        index,
-        reason: reason.to_owned(),
-    })
-}
-
-#[cfg(test)]
-#[allow(clippy::indexing_slicing, clippy::unwrap_used)]
-mod tests {
-    use haproxy_ingest_proto::proto::{HaProxyEvent, LogRecord, log_record};
-
-    use super::*;
-
-    fn batch() -> LogBatch {
-        LogBatch {
-            schema_version: 1,
-            machine_id: "0123456789abcdef0123456789abcdef".to_owned(),
-            hostname: "portal-1".to_owned(),
-            previous_cursor: None,
-            last_cursor: "cursor-1".to_owned(),
-            first_journal_timestamp_us: 1_700_000_000_000_000,
-            last_journal_timestamp_us: 1_700_000_000_000_000,
-            records: vec![LogRecord {
-                journal_timestamp_us: 1_700_000_000_000_000,
-                payload: Some(log_record::Payload::Event(HaProxyEvent::default())),
-            }],
-        }
-    }
-
-    #[test]
-    fn accepts_a_valid_batch() {
-        validate_batch(&batch()).unwrap();
-    }
-
-    #[test]
-    fn accepts_a_batch_with_decreasing_journal_timestamps() {
-        let mut batch = batch();
-        batch.previous_cursor = Some("cursor-0".to_owned());
-        batch.last_cursor = "cursor-2".to_owned();
-        batch.last_journal_timestamp_us = 1_699_999_999_000_000;
-        batch.records.push(LogRecord {
-            journal_timestamp_us: batch.last_journal_timestamp_us,
-            payload: Some(log_record::Payload::Event(HaProxyEvent::default())),
-        });
-
-        validate_batch(&batch).unwrap();
-    }
-
-    #[test]
-    fn rejects_batch_timestamps_that_do_not_match_the_endpoint_records() {
-        for (first, last) in [
-            (1_699_999_999_000_000, 1_700_000_000_000_000),
-            (1_700_000_000_000_000, 1_700_000_001_000_000),
-        ] {
-            let mut batch = batch();
-            batch.first_journal_timestamp_us = first;
-            batch.last_journal_timestamp_us = last;
-
-            assert!(matches!(
-                validate_batch(&batch),
-                Err(ValidationError::TimestampBounds)
-            ));
-        }
-    }
-
-    #[test]
-    fn reports_the_invalid_record_index() {
-        let mut batch = batch();
-        batch.records[0].payload = None;
-        let error = validate_batch(&batch).unwrap_err();
-        assert_eq!(error.record_index(), Some(0));
-    }
-}

rust/haproxy_log_shipper/Cargo.toml

@@ -1,28 +0,0 @@
-[package]
-name = "haproxy-log-shipper"
-version = "0.1.0"
-edition = "2024"
-description = "Ordered HAProxy journald ingestion client"
-license = "MIT"
-
-[lints]
-workspace = true
-
-[dependencies]
-anyhow.workspace = true
-chrono = { workspace = true, features = ["serde"] }
-clap.workspace = true
-garde.workspace = true
-haproxy-ingest-proto = { path = "../haproxy_ingest_proto" }
-prost.workspace = true
-reqwest = { workspace = true, features = ["json"] }
-serde.workspace = true
-serde_json.workspace = true
-sha2.workspace = true
-systemd.workspace = true
-tls.workspace = true
-tokio.workspace = true
-tokio-retry.workspace = true
-tracing.workspace = true
-tracing-subscriber.workspace = true
-zstd.workspace = true

rust/haproxy_log_shipper/package.nix

@@ -1,24 +0,0 @@
-{
-  lib,
-  pkg-config,
-  rustPlatform,
-  systemd,
-}:
-rustPlatform.buildRustPackage {
-  pname = "haproxy-log-shipper";
-  version = "0.1.0";
-  src = import ../rust-source.nix { inherit lib; };
-
-  cargoLock.lockFile = ../../Cargo.lock;
-  cargoBuildFlags = [ "-p haproxy-log-shipper" ];
-  cargoTestFlags = [ "-p haproxy-log-shipper" ];
-
-  nativeBuildInputs = [ pkg-config ];
-  buildInputs = [ systemd ];
-
-  meta = {
-    description = "Ordered HAProxy journald ingestion client";
-    license = lib.licenses.mit;
-    platforms = lib.platforms.linux;
-  };
-}

rust/haproxy_log_shipper/src/api.rs

@@ -1,360 +0,0 @@
-use std::time::Duration;
-
-use anyhow::{Context, Result, bail};
-use haproxy_ingest_proto::{
-    SCHEMA_VERSION,
-    proto::{LogBatch, LogRecord, log_record},
-};
-use prost::Message;
-use reqwest::{Client, StatusCode, header};
-use serde::Deserialize;
-use tokio_retry::strategy::{ExponentialBackoff, jitter};
-use tracing::{Instrument, info, warn};
-
-use crate::mapper::rejected_event;
-
-const INITIAL_RETRIES: usize = 3;
-
-#[derive(Debug, Deserialize)]
-pub struct StatusResponse {
-    machine_id: String,
-    pub last_cursor: Option<String>,
-    last_journal_timestamp: Option<chrono::DateTime<chrono::Utc>>,
-}
-
-#[derive(Debug, Deserialize)]
-struct CommitResponse {
-    last_cursor: String,
-    records_committed: usize,
-    already_committed: bool,
-}
-
-#[derive(Debug, Deserialize)]
-struct ErrorResponse {
-    error: String,
-    record_index: Option<usize>,
-    last_cursor: Option<String>,
-}
-
-#[derive(Clone, Debug)]
-pub struct PendingRecord {
-    pub record: LogRecord,
-    pub cursor: String,
-    pub raw_message: Option<String>,
-}
-
-#[derive(Debug)]
-pub struct PendingBatch {
-    pub machine_id: String,
-    pub hostname: String,
-    pub previous_cursor: Option<String>,
-    pub records: Vec<PendingRecord>,
-}
-
-impl PendingBatch {
-    fn last_cursor(&self) -> Result<&str> {
-        self.records
-            .last()
-            .map(|record| record.cursor.as_str())
-            .context("pending batch is empty")
-    }
-
-    pub fn to_wire(&self) -> Result<LogBatch> {
-        let first_journal_timestamp_us = self
-            .records
-            .first()
-            .context("pending batch is empty")?
-            .record
-            .journal_timestamp_us;
-        let last_journal_timestamp_us = self
-            .records
-            .last()
-            .context("pending batch is empty")?
-            .record
-            .journal_timestamp_us;
-        Ok(LogBatch {
-            schema_version: SCHEMA_VERSION,
-            machine_id: self.machine_id.clone(),
-            hostname: self.hostname.clone(),
-            previous_cursor: self.previous_cursor.clone(),
-            last_cursor: self.last_cursor()?.to_owned(),
-            first_journal_timestamp_us,
-            last_journal_timestamp_us,
-            records: self
-                .records
-                .iter()
-                .map(|pending| pending.record.clone())
-                .collect(),
-        })
-    }
-
-    fn replace_with_rejection(&mut self, index: usize, reason: &str) -> Result<()> {
-        let pending = self
-            .records
-            .get_mut(index)
-            .with_context(|| format!("receiver returned out-of-range record index {index}"))?;
-        if matches!(
-            pending.record.payload,
-            Some(log_record::Payload::Rejected(_))
-        ) {
-            bail!("receiver rejected an already-isolated RejectedEvent at index {index}");
-        }
-        let raw = pending
-            .raw_message
-            .as_deref()
-            .context("receiver rejected a journal gap marker")?;
-        pending.record.payload = Some(log_record::Payload::Rejected(rejected_event(
-            raw,
-            &format!("receiver validation rejected record: {reason}"),
-        )));
-        Ok(())
-    }
-}
-
-pub enum SendResult {
-    Committed(String),
-    Resync,
-}
-
-enum PostBatchResult {
-    Complete(SendResult),
-    RetryNow,
-    RetryAfterStatusCheck,
-}
-
-enum StatusRequestResult {
-    Complete(StatusResponse),
-    Retry,
-}
-
-pub async fn post_batch_with_retry(
-    client: &Client,
-    endpoint: &str,
-    pending: &mut PendingBatch,
-) -> Result<SendResult> {
-    for (attempt, delay) in retry_schedule().enumerate() {
-        if let Some(delay) = delay {
-            tokio::time::sleep(delay).await;
-        }
-        let last_cursor = loop {
-            let batch = pending.to_wire()?;
-            let last_cursor = batch.last_cursor.clone();
-            let encoded = batch.encode_to_vec();
-            let compressed = zstd::stream::encode_all(encoded.as_slice(), 3)
-                .context("failed to compress protobuf batch")?;
-            let result = post_batch(client, endpoint, pending, compressed, &last_cursor)
-                .instrument(tracing::info_span!("post_batch", attempt))
-                .await?;
-            match result {
-                PostBatchResult::Complete(send_result) => return Ok(send_result),
-                PostBatchResult::RetryNow => {}
-                PostBatchResult::RetryAfterStatusCheck => break last_cursor,
-            }
-        };
-
-        if let Ok(status) = get_status_with_retry(client, endpoint, &pending.machine_id).await {
-            if status.last_cursor.as_deref() == Some(last_cursor.as_str()) {
-                info!(last_cursor, "status check confirmed ambiguous batch commit");
-                return Ok(SendResult::Committed(last_cursor));
-            }
-            if status.last_cursor != pending.previous_cursor {
-                return Ok(SendResult::Resync);
-            }
-        }
-    }
-    bail!("batch failed after initial attempt and {INITIAL_RETRIES} retries")
-}
-
-async fn post_batch(
-    client: &Client,
-    endpoint: &str,
-    pending: &mut PendingBatch,
-    compressed: Vec<u8>,
-    last_cursor: &str,
-) -> Result<PostBatchResult> {
-    let response = client
-        .post(format!("{endpoint}/v1/logs/batch"))
-        .header(header::CONTENT_TYPE, "application/x-protobuf")
-        .header(header::CONTENT_ENCODING, "zstd")
-        .body(compressed)
-        .send()
-        .await;
-
-    match response {
-        Ok(response) if response.status().is_success() => {
-            let committed: CommitResponse = response
-                .json()
-                .await
-                .context("receiver returned an invalid commit response")?;
-            if committed.last_cursor != last_cursor
-                || committed.records_committed != pending.records.len()
-            {
-                bail!("receiver commit response did not match the sent batch");
-            }
-            info!(
-                already_committed = committed.already_committed,
-                "batch acknowledged"
-            );
-            Ok(PostBatchResult::Complete(SendResult::Committed(
-                last_cursor.to_owned(),
-            )))
-        }
-        Ok(response) if response.status() == StatusCode::CONFLICT => {
-            let error = decode_error(response).await;
-            warn!(server_cursor = ?error.last_cursor, error = %error.error, "receiver rejected cursor chain");
-            Ok(PostBatchResult::Complete(SendResult::Resync))
-        }
-        Ok(response) if response.status() == StatusCode::PAYLOAD_TOO_LARGE => {
-            let error = decode_error(response).await;
-            bail!(
-                "receiver rejected a batch within the sender's configured limits as too large: {}",
-                error.error
-            )
-        }
-        Ok(response) if response.status() == StatusCode::UNPROCESSABLE_ENTITY => {
-            let error = decode_error(response).await;
-            let index = error
-                .record_index
-                .context("receiver returned 422 without an offending record index")?;
-            pending.replace_with_rejection(index, &error.error)?;
-            warn!(record_index = index, error = %error.error, "isolated receiver-rejected record");
-            Ok(PostBatchResult::RetryNow)
-        }
-        Ok(response)
-            if matches!(
-                response.status(),
-                StatusCode::UNAUTHORIZED
-                    | StatusCode::FORBIDDEN
-                    | StatusCode::BAD_REQUEST
-                    | StatusCode::UNSUPPORTED_MEDIA_TYPE
-            ) =>
-        {
-            let status = response.status();
-            let error = decode_error(response).await;
-            bail!("permanent receiver error {status}: {}", error.error);
-        }
-        Ok(response) if !response.status().is_server_error() => {
-            let status = response.status();
-            let error = decode_error(response).await;
-            bail!("unexpected receiver response {status}: {}", error.error);
-        }
-        Ok(response) => {
-            warn!(status = %response.status(), "transient receiver failure");
-            Ok(PostBatchResult::RetryAfterStatusCheck)
-        }
-        Err(error) => {
-            warn!(error = %error, "ambiguous batch request failure");
-            Ok(PostBatchResult::RetryAfterStatusCheck)
-        }
-    }
-}
-
-pub async fn get_status_with_retry(
-    client: &Client,
-    endpoint: &str,
-    machine_id: &str,
-) -> Result<StatusResponse> {
-    for (attempt, delay) in retry_schedule().enumerate() {
-        if let Some(delay) = delay {
-            tokio::time::sleep(delay).await;
-        }
-        let result = get_status(client, endpoint, machine_id)
-            .instrument(tracing::info_span!("get_status", attempt))
-            .await?;
-        match result {
-            StatusRequestResult::Complete(status) => return Ok(status),
-            StatusRequestResult::Retry => {}
-        }
-    }
-    bail!("status request failed after initial attempt and {INITIAL_RETRIES} retries")
-}
-
-async fn get_status(
-    client: &Client,
-    endpoint: &str,
-    machine_id: &str,
-) -> Result<StatusRequestResult> {
-    let response = match client
-        .get(format!("{endpoint}/v1/logs/status/{machine_id}"))
-        .send()
-        .await
-    {
-        Ok(response) => response,
-        Err(error) => {
-            warn!(error = %error, "transient receiver status failure");
-            return Ok(StatusRequestResult::Retry);
-        }
-    };
-    if !response.status().is_success() {
-        let status = response.status();
-        let error = decode_error(response).await;
-        if status.is_server_error() {
-            warn!(status = %status, error = %error.error, "transient receiver status failure");
-            return Ok(StatusRequestResult::Retry);
-        }
-        bail!(
-            "receiver status request failed with {status}: {}",
-            error.error
-        );
-    }
-    let status: StatusResponse = response
-        .json()
-        .await
-        .context("receiver returned an invalid status response")?;
-    if status.machine_id != machine_id {
-        bail!("receiver status response contains the wrong machine_id");
-    }
-    info!(
-        last_cursor = ?status.last_cursor,
-        last_journal_timestamp = ?status.last_journal_timestamp,
-        "loaded receiver checkpoint"
-    );
-    Ok(StatusRequestResult::Complete(status))
-}
-
-fn retry_delays() -> impl Iterator<Item = Duration> {
-    // The library raises the base to each attempt, then multiplies by the factor.
-    // 2^n * 500 milliseconds gives delays of 1, 2, and 4 seconds.
-    ExponentialBackoff::from_millis(2)
-        .factor(500)
-        .take(INITIAL_RETRIES)
-}
-
-fn retry_schedule() -> impl Iterator<Item = Option<Duration>> {
-    std::iter::once(None).chain(retry_delays().map(jitter).map(Some))
-}
-
-async fn decode_error(response: reqwest::Response) -> ErrorResponse {
-    response.json().await.unwrap_or_else(|error| ErrorResponse {
-        error: format!("response body was not valid error JSON: {error}"),
-        record_index: None,
-        last_cursor: None,
-    })
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn retries_use_one_two_and_four_second_delays() {
-        assert_eq!(
-            retry_delays().collect::<Vec<_>>(),
-            [
-                Duration::from_secs(1),
-                Duration::from_secs(2),
-                Duration::from_secs(4),
-            ]
-        );
-    }
-
-    #[test]
-    fn retry_schedule_starts_immediately_and_bounds_jitter() {
-        let mut schedule = retry_schedule();
-        assert_eq!(schedule.next(), Some(None));
-        for limit in retry_delays() {
-            assert!(matches!(schedule.next(), Some(Some(delay)) if delay <= limit));
-        }
-        assert_eq!(schedule.next(), None);
-    }
-}

rust/haproxy_log_shipper/src/main.rs

@@ -1,271 +0,0 @@
-use std::{fs, time::Duration};
-
-use anyhow::{Context, Result, anyhow, bail};
-use clap::Parser;
-use haproxy_ingest_proto::{
-    MAX_BATCH_PROTO_BYTES, MAX_BATCH_RECORDS,
-    proto::{JournalGap, LogRecord, log_record},
-};
-use prost::Message;
-use reqwest::{Client, header};
-use systemd::journal::{Journal, OpenOptions};
-use tls::install_tls_provider;
-use tracing::{info, warn};
-use tracing_subscriber::EnvFilter;
-
-mod api;
-mod mapper;
-
-use api::{PendingBatch, PendingRecord, SendResult, get_status_with_retry, post_batch_with_retry};
-use mapper::map_message;
-
-#[derive(Debug, Parser)]
-#[command(about, version)]
-struct Args {
-    #[arg(long, env = "HAPROXY_INGEST_TOKEN", hide_env_values = true)]
-    token: String,
-
-    #[arg(long, env = "HAPROXY_INGEST_ENDPOINT")]
-    endpoint: String,
-
-    #[arg(long, env = "HAPROXY_INGEST_UNIT", default_value = "haproxy.service")]
-    unit: String,
-}
-
-#[tokio::main]
-async fn main() -> Result<()> {
-    tracing_subscriber::fmt()
-        .json()
-        .with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()))
-        .init();
-    let args = Args::parse();
-    install_tls_provider()?;
-    let machine_id = get_valid_machine_id()?;
-    let hostname = read_trimmed("/etc/hostname")?;
-    let client = build_client(&args.token)?;
-
-    let mut committed_cursor = get_status_with_retry(&client, &args.endpoint, &machine_id)
-        .await?
-        .last_cursor;
-    loop {
-        let Some(mut batch) =
-            read_batch(&machine_id, &hostname, committed_cursor.clone(), &args.unit)?
-        else {
-            info!(machine_id, "no uncommitted HAProxy journal records");
-            return Ok(());
-        };
-
-        committed_cursor = match post_batch_with_retry(&client, &args.endpoint, &mut batch).await? {
-            SendResult::Committed(cursor) => {
-                info!(
-                    records = batch.records.len(),
-                    last_cursor = %cursor,
-                    "receiver committed HAProxy batch"
-                );
-                Some(cursor)
-            }
-            SendResult::Resync => {
-                let status = get_status_with_retry(&client, &args.endpoint, &machine_id).await?;
-
-                warn!(
-                    server_cursor = ?status.last_cursor,
-                    "cursor conflict; rebuilding from receiver checkpoint"
-                );
-
-                status.last_cursor
-            }
-        };
-    }
-}
-
-fn build_client(token: &str) -> Result<Client> {
-    let authorization = format!("Bearer {token}")
-        .parse()
-        .context("HAPROXY_INGEST_TOKEN contains invalid HTTP header characters")?;
-    Client::builder()
-        .connect_timeout(Duration::from_secs(3))
-        .timeout(Duration::from_secs(10))
-        .default_headers({
-            let mut headers = reqwest::header::HeaderMap::new();
-            headers.insert(header::AUTHORIZATION, authorization);
-            headers
-        })
-        .build()
-        .context("failed to build HTTP client")
-}
-
-fn read_batch(
-    machine_id: &str,
-    hostname: &str,
-    previous_cursor: Option<String>,
-    unit: &str,
-) -> Result<Option<PendingBatch>> {
-    let (mut journal, setup) = setup_journal(machine_id, unit, previous_cursor.as_deref())?;
-    let mut pending_batch = PendingBatch {
-        machine_id: machine_id.to_owned(),
-        hostname: hostname.to_owned(),
-        previous_cursor,
-        records: Vec::new(),
-    };
-    match setup {
-        JournalSetup::Empty => return Ok(None),
-        JournalSetup::Ready => {}
-        JournalSetup::Gap(pending_record) => pending_batch.records.push(*pending_record),
-    }
-    while pending_batch.records.len() < MAX_BATCH_RECORDS {
-        let Some(entry) = journal
-            .next_entry()
-            .context("failed to read journal entry")?
-        else {
-            break;
-        };
-        let cursor = journal.cursor().context("journal entry has no cursor")?;
-        let journal_timestamp_us = i64::try_from(
-            journal
-                .timestamp_usec()
-                .context("journal entry has no timestamp")?,
-        )
-        .context("journal timestamp exceeds int64")?;
-        let message = entry.get("MESSAGE").cloned().unwrap_or_default();
-        pending_batch.records.push(PendingRecord {
-            record: map_message(&message, journal_timestamp_us),
-            cursor,
-            raw_message: Some(message),
-        });
-        if pending_batch.to_wire()?.encoded_len() > MAX_BATCH_PROTO_BYTES {
-            if pending_batch.records.len() == 1 {
-                bail!("single mapped journal record exceeds the 20 MiB protobuf limit");
-            }
-            pending_batch.records.pop();
-            break;
-        }
-    }
-    if pending_batch.records.is_empty() {
-        Ok(None)
-    } else {
-        Ok(Some(pending_batch))
-    }
-}
-
-enum JournalSetup {
-    Empty,
-    Ready,
-    Gap(Box<PendingRecord>),
-}
-
-fn setup_journal(
-    machine_id: &str,
-    unit: &str,
-    previous_cursor: Option<&str>,
-) -> Result<(Journal, JournalSetup)> {
-    let mut journal = OpenOptions::default()
-        .system(true)
-        .local_only(true)
-        .open()
-        .context("failed to open the local system journal")?;
-    journal
-        .match_add("_SYSTEMD_UNIT", unit.as_bytes())
-        .context("failed to filter journal by systemd unit")?;
-    journal
-        .match_add("_MACHINE_ID", machine_id.as_bytes())
-        .context("failed to filter journal by machine ID")?;
-    let setup = position_journal(&mut journal, previous_cursor)?;
-    Ok((journal, setup))
-}
-
-fn position_journal(journal: &mut Journal, previous_cursor: Option<&str>) -> Result<JournalSetup> {
-    let Some(cursor) = previous_cursor else {
-        journal
-            .seek_head()
-            .context("failed to seek to journal head")?;
-        return Ok(JournalSetup::Ready);
-    };
-    let exact = journal.seek_cursor(cursor).is_ok()
-        && journal
-            .next()
-            .context("failed to move to saved journal cursor")?
-            > 0
-        && journal
-            .test_cursor(cursor)
-            .context("failed to test saved journal cursor")?;
-    if exact {
-        return Ok(JournalSetup::Ready);
-    }
-
-    journal
-        .seek_head()
-        .context("failed to seek to retained journal head")?;
-    if journal
-        .next()
-        .context("failed to read retained journal head")?
-        == 0
-    {
-        return Ok(JournalSetup::Empty);
-    }
-    let first_cursor = journal
-        .cursor()
-        .context("retained journal entry has no cursor")?;
-    let first_timestamp = i64::try_from(
-        journal
-            .timestamp_usec()
-            .context("retained journal entry has no timestamp")?,
-    )
-    .context("journal timestamp exceeds int64")?;
-    let gap = PendingRecord {
-        record: LogRecord {
-            journal_timestamp_us: first_timestamp,
-            payload: Some(log_record::Payload::Gap(JournalGap {
-                expected_cursor: cursor.to_owned(),
-                first_available_cursor: first_cursor.clone(),
-                first_available_timestamp_us: Some(first_timestamp),
-                reason: "saved cursor is no longer present in the filtered journal".to_owned(),
-            })),
-        },
-        cursor: first_cursor,
-        raw_message: None,
-    };
-    journal
-        .seek_head()
-        .context("failed to rewind retained journal")?;
-    Ok(JournalSetup::Gap(Box::new(gap)))
-}
-
-fn get_valid_machine_id() -> Result<String> {
-    let machine_id = read_trimmed("/etc/machine-id")?;
-    if machine_id.len() != 32
-        || !machine_id
-            .bytes()
-            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
-    {
-        bail!("/etc/machine-id is not 32 lowercase hexadecimal characters");
-    }
-    Ok(machine_id)
-}
-
-fn read_trimmed(path: &str) -> Result<String> {
-    let value = fs::read_to_string(path).with_context(|| format!("failed to read {path}"))?;
-    let value = value.trim();
-    if value.is_empty() {
-        return Err(anyhow!("{path} is empty"));
-    }
-    Ok(value.to_owned())
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[tokio::test]
-    async fn builds_http_client_without_certificate() -> Result<()> {
-        let args = Args::try_parse_from([
-            "haproxy-log-shipper",
-            "--endpoint",
-            "http://127.0.0.1:8444",
-            "--token",
-            "test-token",
-        ])?;
-        install_tls_provider()?;
-        build_client(&args.token)?;
-        Ok(())
-    }
-}

rust/haproxy_log_shipper/src/mapper.rs

@@ -1,354 +0,0 @@
-use std::{fmt::Display, net::IpAddr, str::FromStr};
-
-use anyhow::{Context, Result};
-use garde::{Unvalidated, Valid};
-use haproxy_ingest_proto::proto::{HaProxyEvent, LogRecord, RejectedEvent, log_record};
-use serde::{Deserialize, Deserializer, de::Error as _};
-use sha2::{Digest, Sha256};
-
-const RAW_PREFIX_LIMIT: usize = 4 * 1024;
-
-#[derive(Debug, Deserialize, garde::Validate)]
-#[garde(allow_unvalidated)]
-struct RawHaProxyEvent {
-    #[serde(default, deserialize_with = "deserialize_optional_string")]
-    client_ip: Option<String>,
-    #[serde(default, deserialize_with = "deserialize_optional_number")]
-    #[garde(range(max = 65_535))]
-    client_port: Option<u32>,
-    #[serde(default, deserialize_with = "deserialize_optional_string")]
-    frontend_name: Option<String>,
-    #[serde(default, deserialize_with = "deserialize_optional_string")]
-    backend_name: Option<String>,
-    #[serde(default, deserialize_with = "deserialize_optional_string")]
-    server_name: Option<String>,
-    #[serde(default, deserialize_with = "deserialize_optional_string")]
-    http_method: Option<String>,
-    #[serde(default, deserialize_with = "deserialize_optional_string")]
-    request_path: Option<String>,
-    #[serde(default, deserialize_with = "deserialize_optional_string")]
-    http_version: Option<String>,
-    #[serde(default, deserialize_with = "deserialize_optional_number")]
-    #[garde(required, range(min = 100, max = 599))]
-    status_code: Option<i32>,
-    #[serde(default, deserialize_with = "deserialize_optional_number")]
-    #[garde(required, range(max = 9_223_372_036_854_775_807))]
-    bytes_sent: Option<u64>,
-    #[serde(default, deserialize_with = "deserialize_optional_number")]
-    #[garde(required, range(min = -1))]
-    time_request_ms: Option<i32>,
-    #[serde(default, deserialize_with = "deserialize_optional_number")]
-    #[garde(required, range(min = -1))]
-    time_wait_ms: Option<i32>,
-    #[serde(default, deserialize_with = "deserialize_optional_number")]
-    #[garde(required, range(min = -1))]
-    time_connect_ms: Option<i32>,
-    #[serde(default, deserialize_with = "deserialize_optional_number")]
-    #[garde(required, range(min = -1))]
-    time_response_ms: Option<i32>,
-    #[serde(default, deserialize_with = "deserialize_optional_number")]
-    #[garde(required, range(min = -1))]
-    time_total_ms: Option<i32>,
-    #[serde(default, deserialize_with = "deserialize_optional_string")]
-    termination_state: Option<String>,
-    #[serde(default, deserialize_with = "deserialize_optional_number")]
-    #[garde(range(max = 2_147_483_647))]
-    retries: Option<u32>,
-    #[serde(default, deserialize_with = "deserialize_optional_string")]
-    tls_sni: Option<String>,
-    #[serde(default, deserialize_with = "deserialize_optional_string")]
-    tls_version: Option<String>,
-    #[serde(default, deserialize_with = "deserialize_optional_string")]
-    tls_cipher: Option<String>,
-    #[serde(default, deserialize_with = "deserialize_optional_bool")]
-    tls_resumed: Option<bool>,
-    #[serde(default, deserialize_with = "deserialize_optional_number")]
-    #[garde(range(max = 2_147_483_647))]
-    active_connections: Option<u32>,
-    #[serde(default, deserialize_with = "deserialize_optional_number")]
-    #[garde(range(max = 2_147_483_647))]
-    frontend_connections: Option<u32>,
-    #[serde(default, deserialize_with = "deserialize_optional_number")]
-    #[garde(range(max = 2_147_483_647))]
-    backend_connections: Option<u32>,
-    #[serde(default, deserialize_with = "deserialize_optional_number")]
-    #[garde(range(max = 2_147_483_647))]
-    server_connections: Option<u32>,
-    #[serde(default, deserialize_with = "deserialize_optional_number")]
-    #[garde(range(max = 2_147_483_647))]
-    server_queue: Option<u32>,
-    #[serde(default, deserialize_with = "deserialize_optional_number")]
-    #[garde(range(max = 2_147_483_647))]
-    backend_queue: Option<u32>,
-}
-
-#[must_use]
-pub fn map_message(message: &str, journal_timestamp_us: i64) -> LogRecord {
-    let payload = match parse_event(message) {
-        Ok(event) => log_record::Payload::Event(event),
-        Err(error) => log_record::Payload::Rejected(rejected_event(message, &format!("{error:#}"))),
-    };
-    LogRecord {
-        journal_timestamp_us,
-        payload: Some(payload),
-    }
-}
-
-fn parse_event(message: &str) -> Result<HaProxyEvent> {
-    let event: Unvalidated<RawHaProxyEvent> =
-        serde_json::from_str(message).context("invalid HAProxy JSON")?;
-    let event = event.validate().context("invalid HAProxy event")?;
-    normalize_event(event)
-}
-
-fn normalize_event(event: Valid<RawHaProxyEvent>) -> Result<HaProxyEvent> {
-    let event = event.into_inner();
-    let mut normalization_flags = Vec::new();
-    let client_ip = event.client_ip.and_then(|raw| {
-        raw.parse::<IpAddr>().map_or_else(
-            |_| {
-                normalization_flags.push("invalid_client_ip".to_owned());
-                None
-            },
-            |ip| Some(ip.to_string()),
-        )
-    });
-
-    Ok(HaProxyEvent {
-        client_ip,
-        client_port: event.client_port,
-        frontend_name: event.frontend_name,
-        backend_name: event.backend_name,
-        server_name: event.server_name,
-        http_method: event.http_method,
-        request_path: event.request_path,
-        http_version: event.http_version,
-        status_code: event
-            .status_code
-            .context("validated status_code is absent")?,
-        bytes_sent: event.bytes_sent.context("validated bytes_sent is absent")?,
-        time_request_ms: event
-            .time_request_ms
-            .context("validated time_request_ms is absent")?,
-        time_wait_ms: event
-            .time_wait_ms
-            .context("validated time_wait_ms is absent")?,
-        time_connect_ms: event
-            .time_connect_ms
-            .context("validated time_connect_ms is absent")?,
-        time_response_ms: event
-            .time_response_ms
-            .context("validated time_response_ms is absent")?,
-        time_total_ms: event
-            .time_total_ms
-            .context("validated time_total_ms is absent")?,
-        termination_state: event.termination_state,
-        retries: event.retries.unwrap_or_default(),
-        tls_sni: event.tls_sni,
-        tls_version: event.tls_version,
-        tls_cipher: event.tls_cipher,
-        tls_resumed: event.tls_resumed,
-        normalization_flags,
-        active_connections: event.active_connections,
-        frontend_connections: event.frontend_connections,
-        backend_connections: event.backend_connections,
-        server_connections: event.server_connections,
-        server_queue: event.server_queue,
-        backend_queue: event.backend_queue,
-    })
-}
-
-#[must_use]
-pub fn rejected_event(message: &str, reason: &str) -> RejectedEvent {
-    let mut hasher = Sha256::new();
-    hasher.update(message.as_bytes());
-    let prefix = truncate_utf8(message, RAW_PREFIX_LIMIT);
-    RejectedEvent {
-        reason: truncate_utf8(reason, 1_024).to_owned(),
-        message_sha256: hasher.finalize().to_vec(),
-        raw_message_prefix: Some(prefix.to_owned()),
-        raw_message_truncated: prefix.len() != message.len(),
-    }
-}
-
-fn deserialize_optional_string<'de, D>(deserializer: D) -> Result<Option<String>, D::Error>
-where
-    D: Deserializer<'de>,
-{
-    match Option::<String>::deserialize(deserializer)? {
-        Some(value) if is_empty_or_dash(&value) => Ok(None),
-        value => Ok(value),
-    }
-}
-
-#[derive(Deserialize)]
-#[serde(untagged)]
-enum NumberOrString<T> {
-    Number(T),
-    String(String),
-}
-
-fn deserialize_optional_number<'de, D, T>(deserializer: D) -> Result<Option<T>, D::Error>
-where
-    D: Deserializer<'de>,
-    T: Deserialize<'de> + FromStr,
-    T::Err: Display,
-{
-    match Option::<NumberOrString<T>>::deserialize(deserializer)? {
-        None => Ok(None),
-        Some(NumberOrString::Number(value)) => Ok(Some(value)),
-        Some(NumberOrString::String(value)) if is_empty_or_dash(&value) => Ok(None),
-        Some(NumberOrString::String(value)) => value.parse().map(Some).map_err(D::Error::custom),
-    }
-}
-
-#[derive(Deserialize)]
-#[serde(untagged)]
-enum BoolValue {
-    Bool(bool),
-    Integer(i64),
-    String(String),
-}
-
-fn deserialize_optional_bool<'de, D>(deserializer: D) -> Result<Option<bool>, D::Error>
-where
-    D: Deserializer<'de>,
-{
-    match Option::<BoolValue>::deserialize(deserializer)? {
-        None => Ok(None),
-        Some(BoolValue::Bool(value)) => Ok(Some(value)),
-        Some(BoolValue::Integer(0)) => Ok(Some(false)),
-        Some(BoolValue::Integer(1)) => Ok(Some(true)),
-        Some(BoolValue::String(value)) if is_empty_or_dash(&value) => Ok(None),
-        Some(BoolValue::String(value)) if matches!(value.as_str(), "0" | "false") => {
-            Ok(Some(false))
-        }
-        Some(BoolValue::String(value)) if matches!(value.as_str(), "1" | "true") => Ok(Some(true)),
-        Some(BoolValue::Integer(_) | BoolValue::String(_)) => {
-            Err(D::Error::custom("expected a boolean, 0, 1, null, or '-'"))
-        }
-    }
-}
-
-fn is_empty_or_dash(value: &str) -> bool {
-    value.is_empty() || value == "-"
-}
-
-fn truncate_utf8(value: &str, max_bytes: usize) -> &str {
-    if value.len() <= max_bytes {
-        return value;
-    }
-    let mut end = max_bytes;
-    while !value.is_char_boundary(end) {
-        end = end.saturating_sub(1);
-    }
-    value.get(..end).unwrap_or_default()
-}
-
-#[cfg(test)]
-#[allow(clippy::panic, clippy::unwrap_used)]
-mod tests {
-    use haproxy_ingest_proto::proto::log_record;
-
-    use super::*;
-
-    fn valid_json(extra: &str) -> String {
-        format!(
-            r#"{{"client_ip":"192.0.2.10","status_code":200,"bytes_sent":12,"time_request_ms":-1,"time_wait_ms":0,"time_connect_ms":3,"time_response_ms":4,"time_total_ms":7{extra}}}"#
-        )
-    }
-
-    #[test]
-    fn preserves_negative_timer_sentinels() {
-        let record = map_message(&valid_json(""), 10);
-        let Some(log_record::Payload::Event(event)) = record.payload else {
-            panic!("expected mapped event");
-        };
-        assert_eq!(event.time_request_ms, -1);
-    }
-
-    #[test]
-    fn serde_parses_the_full_u64_range() {
-        let message =
-            valid_json("").replace("\"bytes_sent\":12", "\"bytes_sent\":18446744073709551615");
-        let event: RawHaProxyEvent = serde_json::from_str(&message).unwrap();
-        assert_eq!(event.bytes_sent, Some(u64::MAX));
-    }
-
-    #[test]
-    fn serde_coerces_haproxy_values() {
-        let message = r#"{
-            "client_port":"443",
-            "frontend_name":"-",
-            "status_code":"200",
-            "bytes_sent":"12",
-            "time_request_ms":"-1",
-            "time_wait_ms":"0",
-            "time_connect_ms":"3",
-            "time_response_ms":"4",
-            "time_total_ms":"7",
-            "retries":"1",
-            "tls_resumed":"true",
-            "server_queue":null
-        }"#;
-        let record = map_message(message, 10);
-        let Some(log_record::Payload::Event(event)) = record.payload else {
-            panic!("expected mapped event");
-        };
-        assert_eq!(event.client_port, Some(443));
-        assert_eq!(event.frontend_name, None);
-        assert_eq!(event.status_code, 200);
-        assert_eq!(event.time_request_ms, -1);
-        assert_eq!(event.retries, 1);
-        assert_eq!(event.tls_resumed, Some(true));
-        assert_eq!(event.server_queue, None);
-    }
-
-    #[test]
-    fn garde_rejects_invalid_semantics() {
-        let messages = [
-            valid_json("").replace("\"status_code\":200", "\"status_code\":99"),
-            valid_json("").replace("\"time_total_ms\":7", "\"time_total_ms\":-2"),
-            valid_json(",\"client_port\":65536"),
-        ];
-        for message in messages {
-            let record = map_message(&message, 10);
-            let Some(log_record::Payload::Rejected(rejected)) = record.payload else {
-                panic!("expected rejected event");
-            };
-            assert!(rejected.reason.contains("invalid HAProxy event"));
-        }
-    }
-
-    #[test]
-    fn invalid_ip_is_non_fatal_and_flagged() {
-        let message = valid_json("").replace("192.0.2.10", "not-an-ip");
-        let record = map_message(&message, 10);
-        let Some(log_record::Payload::Event(event)) = record.payload else {
-            panic!("expected mapped event");
-        };
-        assert_eq!(event.client_ip, None);
-        assert_eq!(event.normalization_flags, ["invalid_client_ip"]);
-    }
-
-    #[test]
-    fn preserves_request_path() {
-        let path = format!("/{}é", "a".repeat(20_000));
-        let message = valid_json(&format!(",\"request_path\":{path:?}"));
-        let record = map_message(&message, 10);
-        let Some(log_record::Payload::Event(event)) = record.payload else {
-            panic!("expected mapped event");
-        };
-        assert_eq!(event.request_path.as_deref(), Some(path.as_str()));
-    }
-
-    #[test]
-    fn malformed_event_becomes_durable_rejection() {
-        let record = map_message("not-json", 10);
-        let Some(log_record::Payload::Rejected(rejected)) = record.payload else {
-            panic!("expected rejected event");
-        };
-        assert_eq!(rejected.message_sha256.len(), 32);
-        assert!(rejected.reason.contains("invalid HAProxy JSON"));
-    }
-}

rust/rust-source.nix

@@ -1,10 +0,0 @@
-{ lib }:
-lib.fileset.toSource {
-  root = ../.;
-  fileset = lib.fileset.unions [
-    ../Cargo.toml
-    ../Cargo.lock
-    ../clippy.toml
-    ./.
-  ];
-}

rust/signal/Cargo.toml

@@ -1,17 +0,0 @@
-[package]
-name = "signal"
-version = "0.1.0"
-edition = "2024"
-description = "Send Signal alerts through the local signal-cli JSON-RPC daemon"
-license = "MIT"
-publish = false
-
-[dependencies]
-anyhow.workspace = true
-serde_json.workspace = true
-
-[target.'cfg(target_os = "linux")'.dev-dependencies]
-libc = "0.2"
-
-[lints]
-workspace = true

rust/signal/src/lib.rs

@@ -1,78 +0,0 @@
-//! Send alerts through the local signal-cli daemon on Unix systems.
-//!
-//! ```no_run
-//! use signal::Signal;
-//!
-//! let client = Signal {
-//!     from_phone: "+15555550100".into(),
-//! };
-//! client.send("+15555550101", "Service failed")?;
-//! client.send("+15555550102", "Service recovered")?;
-//! # Ok::<(), anyhow::Error>(())
-//! ```
-
-use std::{
-    io::{BufRead, BufReader, Write},
-    os::unix::net::UnixStream,
-    time::Duration,
-};
-
-use anyhow::{Context, Result, bail};
-use serde_json::{Value, json};
-
-/// A sending account for alerts through `/run/signal-cli/socket`.
-pub struct Signal {
-    /// The phone number of an account registered with the local daemon.
-    pub from_phone: String,
-}
-
-impl Signal {
-    /// Send a message to `to_phone` and wait for the corresponding JSON-RPC response.
-    ///
-    /// Socket reads and writes use a four-second timeout.
-    ///
-    /// # Errors
-    ///
-    /// Returns an error if connecting, writing, or reading fails, the daemon
-    /// returns invalid JSON or an RPC error, or it closes before responding.
-    pub fn send(&self, to_phone: &str, body: &str) -> Result<()> {
-        let mut stream = UnixStream::connect("/run/signal-cli/socket")
-            .context("Failed to connect to Signal RPC socket")?;
-        let timeout = Duration::from_secs(4);
-        stream.set_read_timeout(Some(timeout))?;
-        stream.set_write_timeout(Some(timeout))?;
-        let request = json!({
-            "jsonrpc": "2.0",
-            "method": "send",
-            "params": {
-                "account": self.from_phone,
-                "recipient": [to_phone],
-                "message": body,
-            },
-            "id": "signal-alert",
-        });
-        writeln!(stream, "{request}").context("Failed to send Signal RPC request")?;
-        let mut reader = BufReader::new(stream);
-        let mut line = String::new();
-        loop {
-            line.clear();
-            if reader
-                .read_line(&mut line)
-                .context("Failed to read Signal RPC response")?
-                == 0
-                || !line.ends_with('\n')
-            {
-                bail!("Signal RPC socket closed before returning a response");
-            }
-            let response: Value =
-                serde_json::from_str(&line).context("Invalid Signal RPC response")?;
-            if response.get("id").and_then(Value::as_str) != Some("signal-alert") {
-                continue;
-            }
-            if let Some(error) = response.get("error").filter(|value| !value.is_null()) {
-                bail!("Signal RPC returned an error: {error}");
-            }
-            return Ok(());
-        }
-    }
-}

rust/signal/tests/send.rs

@@ -1,147 +0,0 @@
-#![cfg(test)]
-#![cfg(target_os = "linux")]
-
-use std::{
-    cell::RefCell,
-    io::{self, BufRead, BufReader, Write},
-    os::{fd::AsRawFd, unix::net::UnixStream},
-    thread,
-    time::{Duration, Instant},
-};
-
-use serde_json::{Value, json};
-use signal::Signal;
-
-thread_local! {
-    static CONNECTION: RefCell<Option<UnixStream>> = const { RefCell::new(None) };
-}
-
-// Patch libc's connect only in this integration-test executable. Each test
-// supplies its own socket pair; no call can reach the real Signal daemon.
-// SAFETY: This symbol uses libc's connect ABI and preserves its return/errno
-// convention. It never delegates to a real connection attempt.
-#[unsafe(no_mangle)]
-unsafe extern "C" fn connect(
-    socket: libc::c_int,
-    address: *const libc::sockaddr,
-    length: libc::socklen_t,
-) -> libc::c_int {
-    let path_offset = std::mem::offset_of!(libc::sockaddr_un, sun_path);
-    let expected = b"/run/signal-cli/socket\0";
-    if !address.is_null()
-        && let Some(path_end) = path_offset.checked_add(expected.len())
-        && usize::try_from(length).is_ok_and(|length| length >= path_end)
-    {
-        // SAFETY: connect's caller supplies an address valid for length bytes.
-        // The bounds check above covers the entire pathname slice.
-        let path = unsafe {
-            std::slice::from_raw_parts(address.cast::<u8>().add(path_offset), expected.len())
-        };
-        if path == expected
-            && let Some(stream) = CONNECTION.with_borrow_mut(Option::take)
-        {
-            // SAFETY: Both descriptors are live and owned by this thread.
-            // dup2 replaces the unconnected socket with the test endpoint.
-            return if unsafe { libc::dup2(stream.as_raw_fd(), socket) } < 0 {
-                -1
-            } else {
-                0
-            };
-        }
-    }
-    // SAFETY: __errno_location returns this thread's valid errno pointer.
-    unsafe { *libc::__errno_location() = libc::ECONNREFUSED };
-    -1
-}
-
-fn test_connection() -> UnixStream {
-    let (client, server) = UnixStream::pair().unwrap();
-    CONNECTION.with_borrow_mut(|connection| assert!(connection.replace(client).is_none()));
-    server
-}
-
-fn signal() -> Signal {
-    Signal {
-        from_phone: "1234567890".into(),
-    }
-}
-
-#[test]
-fn reuses_sender_for_different_recipients_and_ignores_unrelated_responses() {
-    let client = signal();
-    for (to_phone, body) in [
-        ("0987654321", "failure\nwith details"),
-        ("+15555550102", "another failure"),
-    ] {
-        let server = test_connection();
-        let task = thread::spawn(move || {
-            let mut reader = BufReader::new(server);
-            let mut line = String::new();
-            reader.read_line(&mut line).unwrap();
-            let request: Value = serde_json::from_str(&line).unwrap();
-            assert_eq!(
-                request,
-                json!({
-                    "jsonrpc": "2.0", "method": "send", "id": "signal-alert",
-                    "params": { "account": "1234567890", "recipient": [to_phone], "message": body },
-                })
-            );
-            reader
-                .get_mut()
-                .write_all(b"{\"method\":\"receive\"}\n{\"id\":\"other\"}\n{\"id\":\"signal-")
-                .unwrap();
-            reader
-                .get_mut()
-                .write_all(b"alert\",\"result\":{}}\n")
-                .unwrap();
-        });
-        client.send(to_phone, body).unwrap();
-        task.join().unwrap();
-    }
-}
-
-#[test]
-fn rejects_rpc_errors_invalid_json_and_premature_close() {
-    for (reply, expected) in [
-        (
-            "{\"id\":\"signal-alert\",\"error\":{\"code\":-1}}\n",
-            "Signal RPC returned an error",
-        ),
-        ("invalid\n", "Invalid Signal RPC response"),
-        ("", "socket closed"),
-        ("{\"id\":\"signal-alert\"}", "socket closed"),
-    ] {
-        let server = test_connection();
-        let task = thread::spawn(move || {
-            let mut reader = BufReader::new(server);
-            reader.read_line(&mut String::new()).unwrap();
-            reader.get_mut().write_all(reply.as_bytes()).unwrap();
-        });
-        let error = signal().send("0987654321", "failure").unwrap_err();
-        assert!(format!("{error:#}").contains(expected), "{error:#}");
-        task.join().unwrap();
-    }
-}
-
-#[test]
-fn reports_connection_failure() {
-    let error = signal().send("0987654321", "failure").unwrap_err();
-    assert_eq!(error.to_string(), "Failed to connect to Signal RPC socket");
-    assert_eq!(
-        error.downcast_ref::<io::Error>().unwrap().kind(),
-        io::ErrorKind::ConnectionRefused
-    );
-}
-
-#[test]
-fn unresponsive_server_times_out() {
-    let _server = test_connection();
-    let start = Instant::now();
-    let error = signal().send("0987654321", "failure").unwrap_err();
-    assert_eq!(error.to_string(), "Failed to read Signal RPC response");
-    assert!(matches!(
-        error.downcast_ref::<io::Error>().unwrap().kind(),
-        io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock
-    ));
-    assert!(start.elapsed() >= Duration::from_secs(4));
-}

rust/snapshot_manager/Cargo.toml

@@ -1,23 +0,0 @@
-[package]
-name = "snapshot-manager"
-version = "0.1.0"
-edition = "2024"
-description = "Create and rotate ZFS snapshots with Signal failure alerts"
-license = "MIT"
-
-[dependencies]
-anyhow.workspace = true
-chrono.workspace = true
-clap = { workspace = true, features = ["derive", "env"] }
-env_logger.workspace = true
-log.workspace = true
-serde = { workspace = true, features = ["derive"] }
-signal.workspace = true
-toml = "0.9"
-zfs.workspace = true
-
-[dev-dependencies]
-tempfile = "3"
-
-[lints]
-workspace = true

rust/snapshot_manager/package.nix

@@ -1,25 +0,0 @@
-{ lib, rustPlatform }:
-rustPlatform.buildRustPackage {
-  pname = "snapshot-manager";
-  version = "0.1.0";
-  src = import ../rust-source.nix { inherit lib; };
-
-  cargoLock.lockFile = ../../Cargo.lock;
-  cargoBuildFlags = [
-    "-p"
-    "snapshot-manager"
-  ];
-  cargoTestFlags = [
-    "-p"
-    "snapshot-manager"
-    "-p"
-    "signal"
-    "-p"
-    "zfs"
-  ];
-
-  meta = {
-    description = "ZFS snapshot manager with Signal failure alerts";
-    mainProgram = "snapshot-manager";
-  };
-}

rust/snapshot_manager/src/main.rs

@@ -1,383 +0,0 @@
-use std::{
-    collections::HashMap,
-    fs,
-    path::{Path, PathBuf},
-    process::ExitCode,
-};
-
-use anyhow::{Context, Result};
-use chrono::{DateTime, Datelike, NaiveDate, NaiveTime, Timelike, Utc};
-use clap::Parser;
-use log::{error, info};
-use serde::Deserialize;
-use signal::Signal;
-
-#[derive(Parser)]
-#[command(version, about)]
-struct Args {
-    /// TOML file containing default and per-dataset snapshot retention counts.
-    config_file: PathBuf,
-
-    /// Send Signal failure alerts (requires both phone numbers).
-    #[arg(long, env = "SIGNAL_ALERTS", default_value_t = false, action = clap::ArgAction::Set)]
-    signal_alerts: bool,
-
-    /// Signal account sending failure alerts.
-    #[arg(
-        long,
-        env = "SIGNAL_ALERT_FROM_PHONE",
-        hide_env_values = true,
-        required_if_eq("signal_alerts", "true"),
-        value_parser = clap::builder::NonEmptyStringValueParser::new()
-    )]
-    signal_alert_from_phone: Option<String>,
-
-    /// Signal recipient for failure alerts.
-    #[arg(
-        long,
-        env = "SIGNAL_ALERT_TO_PHONE",
-        hide_env_values = true,
-        required_if_eq("signal_alerts", "true"),
-        value_parser = clap::builder::NonEmptyStringValueParser::new()
-    )]
-    signal_alert_to_phone: Option<String>,
-}
-
-#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq)]
-#[serde(deny_unknown_fields)]
-struct Retention {
-    #[serde(rename = "15_min")]
-    quarter_hourly: usize,
-    hourly: usize,
-    daily: usize,
-    monthly: usize,
-}
-
-impl Default for Retention {
-    fn default() -> Self {
-        Self {
-            quarter_hourly: 4,
-            hourly: 12,
-            daily: 0,
-            monthly: 0,
-        }
-    }
-}
-
-#[derive(Debug, Deserialize)]
-#[serde(transparent)]
-struct Config(HashMap<String, Retention>);
-
-impl Config {
-    fn load(path: &Path) -> Result<Self> {
-        let contents = fs::read_to_string(path)
-            .with_context(|| format!("Failed to read configuration {}", path.display()))?;
-        toml::from_str(&contents)
-            .with_context(|| format!("Invalid configuration {}", path.display()))
-    }
-
-    fn retention_for(&self, dataset: &str) -> Retention {
-        self.0
-            .get(dataset)
-            .or_else(|| self.0.get("default"))
-            .copied()
-            .unwrap_or_default()
-    }
-}
-
-fn main() -> ExitCode {
-    let args = Args::parse();
-    env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init();
-    info!("Starting snapshot_manager");
-    let signal = if args.signal_alerts {
-        args.signal_alert_from_phone
-            .map(|from_phone| Signal { from_phone })
-    } else {
-        None
-    };
-    let alerts = signal.as_ref().zip(args.signal_alert_to_phone.as_deref());
-    match run(&args.config_file, alerts) {
-        Ok(()) => {
-            info!("snapshot_manager completed");
-            ExitCode::SUCCESS
-        }
-        Err(error) => {
-            error!("{error:#}");
-            ExitCode::FAILURE
-        }
-    }
-}
-
-fn run(config_file: &Path, alerts: Option<(&Signal, &str)>) -> Result<()> {
-    let result = (|| {
-        // Validate all retention settings before making any changes to ZFS.
-        let config = Config::load(config_file)?;
-        let snapshot_time = timestamp(Utc::now());
-        for dataset_name in zfs::list_datasets()? {
-            if let Err(error) = zfs::create_snapshot(&dataset_name, &snapshot_time) {
-                let message = format!("{dataset_name} failed to create snapshot {snapshot_time}");
-                error!("{message}: {error:#}");
-                send_alert(alerts, &message)?;
-                continue;
-            }
-            remove_old_snapshots(&config, alerts, &dataset_name)
-                .context("Failed to remove old snapshots")?;
-        }
-        Ok(())
-    })();
-    if let Err(error) = &result {
-        error!("snapshot_manager failed: {error:#}");
-        if let Err(alert_error) = send_alert(alerts, "snapshot_manager failed") {
-            error!("Failed to send failure alert: {alert_error:#}");
-        }
-    }
-    result
-}
-
-fn timestamp(now: DateTime<Utc>) -> String {
-    let minute = now.minute().saturating_sub(now.minute() % 15);
-    format!("auto_{}{minute:02}", now.format("%Y%m%d%H"))
-}
-
-fn remove_old_snapshots(
-    config: &Config,
-    alerts: Option<(&Signal, &str)>,
-    dataset_name: &str,
-) -> Result<()> {
-    let retention = config.retention_for(dataset_name);
-    info!("Using {retention:?} for {dataset_name}");
-    let snapshots = zfs::list_snapshots(dataset_name)?;
-    for snapshot in snapshots_to_delete(&snapshots, retention) {
-        info!("Deleting {dataset_name}@{snapshot}");
-        if let Some(error) = zfs::delete_snapshot(dataset_name, snapshot)? {
-            let message = format!("{dataset_name}@{snapshot} failed to delete: {error}");
-            error!("{message}");
-            send_alert(alerts, &message)?;
-        }
-    }
-    Ok(())
-}
-
-fn snapshots_to_delete(snapshots: &[String], retention: Retention) -> Vec<&str> {
-    let mut tiers: [Vec<&str>; 4] = Default::default();
-    for snapshot in snapshots {
-        if let Some(index) = bucket(snapshot)
-            && let Some(tier) = tiers.get_mut(index)
-        {
-            tier.push(snapshot);
-        }
-    }
-
-    let counts = [
-        retention.quarter_hourly,
-        retention.hourly,
-        retention.daily,
-        retention.monthly,
-    ];
-    let mut deletions = Vec::new();
-    for (mut tier, count) in tiers.into_iter().zip(counts) {
-        tier.sort_unstable();
-        let delete_count = tier.len().saturating_sub(count);
-        deletions.extend(tier.into_iter().take(delete_count));
-    }
-    deletions
-}
-
-fn bucket(name: &str) -> Option<usize> {
-    let digits = name.strip_prefix("auto_")?;
-    if digits.len() != 12 || !digits.bytes().all(|byte| byte.is_ascii_digit()) {
-        return None;
-    }
-    let date = NaiveDate::from_ymd_opt(
-        digits.get(..4)?.parse().ok()?,
-        digits.get(4..6)?.parse().ok()?,
-        digits.get(6..8)?.parse().ok()?,
-    )?;
-    let time = NaiveTime::from_hms_opt(
-        digits.get(8..10)?.parse().ok()?,
-        digits.get(10..)?.parse().ok()?,
-        0,
-    )?;
-
-    // The tiers are disjoint: midnight belongs to daily, except on the first
-    // of the month, when it belongs to monthly.
-    match (time.minute(), time.hour(), date.day()) {
-        (15 | 30 | 45, _, _) => Some(0),
-        (0, 1..=23, _) => Some(1),
-        (0, 0, 2..=31) => Some(2),
-        (0, 0, 1) => Some(3),
-        _ => None,
-    }
-}
-
-fn send_alert(alerts: Option<(&Signal, &str)>, body: &str) -> Result<()> {
-    if let Some((signal, to_phone)) = alerts {
-        signal.send(to_phone, body)?;
-    }
-    Ok(())
-}
-
-#[cfg(test)]
-mod tests {
-    use chrono::TimeZone;
-
-    use super::*;
-
-    #[test]
-    fn dataset_overrides_default_and_other_datasets_fall_back() {
-        let config: Config = toml::from_str(
-            r#"
-            [default]
-            15_min = 8
-            hourly = 24
-            daily = 0
-            monthly = 0
-
-            ["pool/home"]
-            15_min = 2
-            hourly = 3
-            daily = 4
-            monthly = 5
-            "#,
-        )
-        .unwrap();
-        assert_eq!(
-            config.retention_for("pool/home"),
-            Retention {
-                quarter_hourly: 2,
-                hourly: 3,
-                daily: 4,
-                monthly: 5,
-            }
-        );
-        assert_eq!(config.retention_for("pool/other").hourly, 24);
-        assert_eq!(config.retention_for("pool/home/child").quarter_hourly, 8);
-    }
-
-    #[test]
-    fn absent_default_uses_original_builtin_counts() {
-        let config: Config = toml::from_str("").unwrap();
-        assert_eq!(config.retention_for("pool/home"), Retention::default());
-    }
-
-    #[test]
-    fn rejects_incomplete_negative_and_unknown_retention_settings() {
-        for settings in [
-            "15_min = 1\nhourly = 2\ndaily = 3",
-            "15_min = -1\nhourly = 2\ndaily = 3\nmonthly = 4",
-            "15_min = 1\nhourly = 2\ndaily = 3\nmonthly = 4\nweekly = 5",
-            "15_min = 1.5\nhourly = 2\ndaily = 3\nmonthly = 4",
-        ] {
-            assert!(toml::from_str::<Config>(&format!("[default]\n{settings}")).is_err());
-        }
-    }
-
-    #[test]
-    fn timestamps_round_down_at_quarter_hour_boundaries() {
-        for (minute, expected) in [
-            (0, "00"),
-            (14, "00"),
-            (15, "15"),
-            (29, "15"),
-            (30, "30"),
-            (44, "30"),
-            (45, "45"),
-            (59, "45"),
-        ] {
-            let now = Utc.with_ymd_and_hms(2026, 1, 1, 0, minute, 59).unwrap();
-            assert_eq!(timestamp(now), format!("auto_2026010100{expected}"));
-        }
-    }
-
-    #[test]
-    fn tiers_are_disjoint_at_day_and_month_boundaries() {
-        for (name, expected) in [
-            ("auto_202602010015", 0),
-            ("auto_202602010030", 0),
-            ("auto_202602010045", 0),
-            ("auto_202602010100", 1),
-            ("auto_202602012300", 1),
-            ("auto_202602020000", 2),
-            ("auto_202402290000", 2),
-            ("auto_202602010000", 3),
-            ("auto_202601010000", 3),
-        ] {
-            assert_eq!(bucket(name), Some(expected), "{name}");
-        }
-    }
-
-    #[test]
-    fn only_exact_valid_manager_names_are_eligible() {
-        for name in [
-            "manual",
-            "manual_auto_202602010015",
-            "auto_202602010015_backup",
-            "auto_20260201001",
-            "auto_202602300000",
-            "auto_202602012400",
-            "auto_202602010060",
-            "auto_202602010005",
-            "auto_２０２６02010015",
-        ] {
-            assert_eq!(bucket(name), None, "{name}");
-        }
-    }
-
-    #[test]
-    fn keeps_newest_in_each_tier_regardless_of_input_order() {
-        let snapshots = [
-            "auto_202601010000",
-            "auto_202602020000",
-            "auto_202602020115",
-            "auto_202602020215",
-            "auto_202602020200",
-            "auto_202602020100",
-            "auto_202602030000",
-            "auto_202602010000",
-            "manual",
-        ]
-        .map(String::from);
-        let retention = Retention {
-            quarter_hourly: 1,
-            hourly: 1,
-            daily: 1,
-            monthly: 1,
-        };
-        assert_eq!(
-            snapshots_to_delete(&snapshots, retention),
-            [
-                "auto_202602020115",
-                "auto_202602020100",
-                "auto_202602020000",
-                "auto_202601010000",
-            ]
-        );
-    }
-
-    #[test]
-    fn zero_counts_delete_all_managed_snapshots_and_large_counts_keep_them() {
-        let snapshots = [
-            "auto_202601010015",
-            "auto_202601010100",
-            "auto_202601020000",
-            "auto_202601010000",
-            "manual",
-        ]
-        .map(String::from);
-        let none = Retention {
-            quarter_hourly: 0,
-            hourly: 0,
-            daily: 0,
-            monthly: 0,
-        };
-        assert_eq!(snapshots_to_delete(&snapshots, none).len(), 4);
-        let all = Retention {
-            quarter_hourly: 100,
-            hourly: 100,
-            daily: 100,
-            monthly: 100,
-        };
-        assert!(snapshots_to_delete(&snapshots, all).is_empty());
-        assert!(snapshots_to_delete(&[], none).is_empty());
-    }
-}

rust/snapshot_manager/tests/cli.rs

@@ -1,202 +0,0 @@
-#![cfg(test)]
-
-use std::{fs, os::unix::fs::symlink, path::Path, process::Command};
-
-use tempfile::tempdir;
-
-fn command() -> Command {
-    let mut command = Command::new(env!("CARGO_BIN_EXE_snapshot-manager"));
-    command
-        .env_remove("SIGNAL_ALERTS")
-        .env_remove("SIGNAL_ALERT_FROM_PHONE")
-        .env_remove("SIGNAL_ALERT_TO_PHONE")
-        .env_remove("SNAPSHOT_TEST_SCRIPT")
-        .env_remove("RUST_LOG");
-    command
-}
-
-#[test]
-fn help_works_without_phone_numbers() {
-    let output = command().arg("--help").output().unwrap();
-    assert!(output.status.success());
-    assert!(
-        String::from_utf8(output.stdout)
-            .unwrap()
-            .contains("<CONFIG_FILE>")
-    );
-}
-
-#[test]
-fn help_documents_environment_variables_without_exposing_phone_numbers() {
-    let output = command()
-        .env("SIGNAL_ALERT_FROM_PHONE", "+15555550101")
-        .env("SIGNAL_ALERT_TO_PHONE", "+15555550102")
-        .arg("--help")
-        .output()
-        .unwrap();
-    assert!(output.status.success());
-    let stdout = String::from_utf8(output.stdout).unwrap();
-    for name in [
-        "SIGNAL_ALERTS",
-        "SIGNAL_ALERT_FROM_PHONE",
-        "SIGNAL_ALERT_TO_PHONE",
-    ] {
-        assert!(stdout.contains(name), "{stdout}");
-    }
-    assert!(!stdout.contains("+15555550101"), "{stdout}");
-    assert!(!stdout.contains("+15555550102"), "{stdout}");
-}
-
-#[test]
-fn config_argument_is_required() {
-    let output = command().output().unwrap();
-    assert_eq!(output.status.code(), Some(2));
-}
-
-#[test]
-fn enabled_alerts_require_phone_numbers_before_running() {
-    for use_env in [false, true] {
-        for (from, to, expected) in [
-            (None, None, "--signal-alert-from-phone"),
-            (Some(""), None, "--signal-alert-from-phone"),
-            (Some("+15555550101"), None, "--signal-alert-to-phone"),
-            (Some("+15555550101"), Some(""), "--signal-alert-to-phone"),
-            (None, Some("+15555550102"), "--signal-alert-from-phone"),
-        ] {
-            let mut command = command();
-            if use_env {
-                command.env("SIGNAL_ALERTS", "true");
-            } else {
-                command.args(["--signal-alerts", "true"]);
-            }
-            if let Some(from) = from {
-                command.env("SIGNAL_ALERT_FROM_PHONE", from);
-            }
-            if let Some(to) = to {
-                command.env("SIGNAL_ALERT_TO_PHONE", to);
-            }
-            let output = command.arg("unused.toml").output().unwrap();
-            let stderr = String::from_utf8(output.stderr).unwrap();
-            assert_eq!(output.status.code(), Some(2), "{stderr}");
-            assert!(stderr.contains(expected), "{stderr}");
-            assert!(!stderr.contains("Starting snapshot_manager"), "{stderr}");
-        }
-    }
-}
-
-#[test]
-fn enabled_alerts_accept_environment_variables_and_cli_overrides() {
-    let dir = tempdir().unwrap();
-    let config = dir.path().join("config.toml");
-    let scenario = dir.path().join("scenario");
-    fs::write(&config, "").unwrap();
-    // An empty dataset list completes successfully without sending an alert.
-    fs::write(&scenario, "").unwrap();
-    symlink(
-        Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/zfs"),
-        dir.path().join("zfs"),
-    )
-    .unwrap();
-
-    for use_cli in [false, true] {
-        let mut command = command();
-        if use_cli {
-            command
-                .env("SIGNAL_ALERTS", "invalid")
-                .env("SIGNAL_ALERT_FROM_PHONE", "")
-                .env("SIGNAL_ALERT_TO_PHONE", "")
-                .args([
-                    "--signal-alerts",
-                    "true",
-                    "--signal-alert-from-phone",
-                    "+15555550101",
-                    "--signal-alert-to-phone",
-                    "+15555550102",
-                ]);
-        } else {
-            command
-                .env("SIGNAL_ALERTS", "true")
-                .env("SIGNAL_ALERT_FROM_PHONE", "+15555550101")
-                .env("SIGNAL_ALERT_TO_PHONE", "+15555550102");
-        }
-        let output = command
-            .current_dir(dir.path())
-            .env("PATH", dir.path())
-            .env("SNAPSHOT_TEST_SCRIPT", &scenario)
-            .arg(&config)
-            .output()
-            .unwrap();
-        let stderr = String::from_utf8(output.stderr).unwrap();
-        assert!(output.status.success(), "{stderr}");
-        assert!(stderr.contains("snapshot_manager completed"), "{stderr}");
-    }
-}
-
-#[test]
-fn disabled_alerts_allow_snapshot_failures_without_phone_numbers() {
-    let dir = tempdir().unwrap();
-    let config = dir.path().join("config.toml");
-    fs::write(&config, "").unwrap();
-    symlink(
-        Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/zfs"),
-        dir.path().join("zfs"),
-    )
-    .unwrap();
-
-    for (env_value, args) in [
-        (None, vec![]),
-        (Some("false"), vec![]),
-        (None, vec!["--signal-alerts", "false"]),
-        (Some("true"), vec!["--signal-alerts", "false"]),
-    ] {
-        let mut command = command();
-        if let Some(value) = env_value {
-            command.env("SIGNAL_ALERTS", value);
-        }
-        let output = command
-            .env("PATH", dir.path())
-            .args(args)
-            .arg(&config)
-            .output()
-            .unwrap();
-        let stderr = String::from_utf8(output.stderr).unwrap();
-        assert!(output.status.success(), "{stderr}");
-        assert!(
-            stderr.contains("pool/home failed to create snapshot"),
-            "{stderr}"
-        );
-        assert!(stderr.contains("snapshot_manager completed"), "{stderr}");
-        assert!(!stderr.contains("Signal"), "{stderr}");
-    }
-}
-
-#[test]
-fn disabled_alerts_preserve_fatal_errors_without_phone_numbers() {
-    let dir = tempdir().unwrap();
-    for args in [vec![], vec!["--signal-alerts", "false"]] {
-        let output = command()
-            .args(args)
-            .arg(dir.path().join("missing.toml"))
-            .output()
-            .unwrap();
-        let stderr = String::from_utf8(output.stderr).unwrap();
-        assert_eq!(output.status.code(), Some(1), "{stderr}");
-        assert!(stderr.contains("Failed to read configuration"), "{stderr}");
-        assert!(!stderr.contains("SIGNAL_ALERT_"), "{stderr}");
-        assert!(!stderr.contains("Failed to send failure alert"), "{stderr}");
-    }
-}
-
-#[test]
-fn signal_alerts_requires_a_boolean_value() {
-    for use_env in [false, true] {
-        let mut command = command();
-        if use_env {
-            command.env("SIGNAL_ALERTS", "invalid");
-        } else {
-            command.args(["--signal-alerts", "invalid"]);
-        }
-        let output = command.arg("unused.toml").output().unwrap();
-        assert_eq!(output.status.code(), Some(2));
-    }
-}

rust/snapshot_manager/tests/fixtures/zfs

@@ -1,12 +0,0 @@
-#!/bin/sh
-if [ -n "${SNAPSHOT_TEST_SCRIPT:-}" ]; then
-  printf '%s\n' "$*" >> calls
-  . "$SNAPSHOT_TEST_SCRIPT"
-  exit
-fi
-
-case "$*" in
-  'list -Hp -t filesystem -o name') printf '%s\n' pool/home ;;
-  'snapshot pool/home@'*) printf '%s\n' 'permission denied' >&2; exit 1 ;;
-  *) exit 99 ;;
-esac

rust/snapshot_manager/tests/manager.rs

@@ -1,290 +0,0 @@
-#![cfg(test)]
-
-use std::{
-    fs,
-    os::unix::fs::symlink,
-    path::{Path, PathBuf},
-    process::{Command, Output},
-};
-
-use tempfile::{TempDir, tempdir};
-
-struct Fixture {
-    dir: TempDir,
-    config: PathBuf,
-}
-
-impl Fixture {
-    fn new(script: &str) -> Self {
-        let dir = tempdir().unwrap();
-        // Only the child process sees this PATH. The real zfs cannot be found.
-        symlink(
-            Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/zfs"),
-            dir.path().join("zfs"),
-        )
-        .unwrap();
-        // The executable is a checked-in fixture; scenario files are read by
-        // its shell, so parallel tests never execute a file being written.
-        fs::write(dir.path().join("scenario"), script).unwrap();
-        let config = dir.path().join("config.toml");
-        fs::write(
-            &config,
-            "[default]\n15_min = 1\nhourly = 0\ndaily = 0\nmonthly = 0\n",
-        )
-        .unwrap();
-        Self { dir, config }
-    }
-
-    fn calls(&self) -> Vec<String> {
-        fs::read_to_string(self.dir.path().join("calls"))
-            .unwrap_or_default()
-            .lines()
-            .map(String::from)
-            .collect()
-    }
-
-    fn normalized_calls(&self) -> Vec<String> {
-        self.calls()
-            .into_iter()
-            .map(|call| {
-                if call.starts_with("snapshot ") {
-                    let (dataset, name) = call.split_once('@').unwrap();
-                    assert!(name.starts_with("auto_"), "{call}");
-                    format!("{dataset}@<timestamp>")
-                } else {
-                    call
-                }
-            })
-            .collect()
-    }
-
-    fn run(&self) -> Output {
-        Command::new(env!("CARGO_BIN_EXE_snapshot-manager"))
-            .current_dir(self.dir.path())
-            .env("PATH", self.dir.path())
-            .env("SNAPSHOT_TEST_SCRIPT", self.dir.path().join("scenario"))
-            .env_remove("SIGNAL_ALERTS")
-            .env_remove("SIGNAL_ALERT_FROM_PHONE")
-            .env_remove("SIGNAL_ALERT_TO_PHONE")
-            .env_remove("RUST_LOG")
-            .arg(&self.config)
-            .output()
-            .unwrap()
-    }
-}
-
-fn stderr(output: &Output) -> &str {
-    std::str::from_utf8(&output.stderr).unwrap()
-}
-
-#[test]
-fn creates_then_prunes_only_this_datasets_managed_snapshots() {
-    let fixture = Fixture::new(
-        r#"
-case "$*" in
-  'list -Hp -t filesystem -o name')
-    printf '%s\n' pool pool/home
-    printf '%s\n' 'warning on stderr' >&2 ;;
-  'snapshot pool/home@auto_'*) ;;
-  'list -Hp -t snapshot -o name -d 1 pool/home')
-    printf '%s\n' pool/home@auto_200002020145 pool/home@auto_200002020115 pool/home@manual pool/home@manual_auto_200001010015 pool/home/child@auto_200001010015 pool/other@auto_200001010015 ;;
-  'destroy pool/home@auto_200002020115') ;;
-  *) exit 99 ;;
-esac
-"#,
-    );
-    let output = fixture.run();
-    assert!(output.status.success(), "{}", stderr(&output));
-    assert_eq!(
-        fixture.normalized_calls(),
-        [
-            "list -Hp -t filesystem -o name",
-            "snapshot pool/home@<timestamp>",
-            "list -Hp -t snapshot -o name -d 1 pool/home",
-            "destroy pool/home@auto_200002020115",
-        ]
-    );
-}
-
-#[test]
-fn failed_or_duplicate_creation_skips_pruning_but_continues() {
-    for error in ["snapshot already exists", "permission denied"] {
-        let fixture = Fixture::new(&format!(
-            r#"
-case "$*" in
-  'list -Hp -t filesystem -o name') printf '%s\n' pool/home pool/other ;;
-  'snapshot pool/home@auto_'*) printf '%s\n' '{error}' >&2; exit 1 ;;
-  'snapshot pool/other@auto_'*) ;;
-  'list -Hp -t snapshot -o name -d 1 pool/other') ;;
-  *) exit 99 ;;
-esac
-"#
-        ));
-        let output = fixture.run();
-        assert!(output.status.success(), "{}", stderr(&output));
-        assert!(stderr(&output).contains("pool/home failed to create snapshot auto_"));
-        assert!(stderr(&output).contains(error));
-        assert_eq!(
-            fixture.normalized_calls(),
-            [
-                "list -Hp -t filesystem -o name",
-                "snapshot pool/home@<timestamp>",
-                "snapshot pool/other@<timestamp>",
-                "list -Hp -t snapshot -o name -d 1 pool/other",
-            ]
-        );
-        let calls = fixture.calls();
-        assert_eq!(
-            calls[1].split_once('@').unwrap().1,
-            calls[2].split_once('@').unwrap().1
-        );
-    }
-}
-
-#[test]
-fn dependent_clone_failure_is_reported_and_remaining_deletions_continue() {
-    let fixture = Fixture::new(
-        r#"
-case "$*" in
-  'list -Hp -t filesystem -o name') printf '%s\n' pool/home ;;
-  'snapshot pool/home@auto_'*) ;;
-  'list -Hp -t snapshot -o name -d 1 pool/home')
-    printf '%s\n' pool/home@auto_200002020115 pool/home@auto_200002020130 pool/home@auto_200002020145 ;;
-  'destroy pool/home@auto_200002020115')
-    printf '%s\n' "cannot destroy 'pool/home@auto_200002020115': snapshot has dependent clones" >&2; exit 1 ;;
-  'destroy pool/home@auto_200002020130') ;;
-  *) exit 99 ;;
-esac
-"#,
-    );
-    let output = fixture.run();
-    assert!(output.status.success(), "{}", stderr(&output));
-    assert!(
-        stderr(&output).contains(
-            "pool/home@auto_200002020115 failed to delete: snapshot has dependent clones"
-        )
-    );
-    assert_eq!(
-        fixture.calls().last().unwrap(),
-        "destroy pool/home@auto_200002020130"
-    );
-}
-
-#[test]
-fn unexpected_destroy_failure_aborts_and_reports_the_error() {
-    let fixture = Fixture::new(
-        r#"
-case "$1" in
-  list)
-    if [ "$4" = filesystem ]; then
-      printf '%s\n' pool/home
-    else
-      printf '%s\n' pool/home@auto_200002020115 pool/home@auto_200002020130 pool/home@auto_200002020145
-    fi ;;
-  snapshot) ;;
-  destroy) printf '%s\n' 'permission denied' >&2; exit 1 ;;
-esac
-"#,
-    );
-    let output = fixture.run();
-    assert_eq!(output.status.code(), Some(1), "{}", stderr(&output));
-    assert!(stderr(&output).contains("permission denied"));
-    assert!(stderr(&output).contains("snapshot_manager failed"));
-    assert_eq!(
-        fixture.calls().last().unwrap(),
-        "destroy pool/home@auto_200002020115"
-    );
-}
-
-#[test]
-fn failed_list_never_uses_partial_output() {
-    for stream in ["", " >&2"] {
-        let fixture = Fixture::new(&format!(
-            "printf '%s\\n' pool/home\nprintf '%s\\n' 'cannot list datasets'{stream}\nexit 1"
-        ));
-        let output = fixture.run();
-        assert_eq!(output.status.code(), Some(1), "{}", stderr(&output));
-        assert!(stderr(&output).contains("cannot list datasets"));
-        assert!(stderr(&output).contains("snapshot_manager failed"));
-        assert_eq!(fixture.calls(), ["list -Hp -t filesystem -o name"]);
-    }
-}
-
-#[test]
-fn failed_snapshot_listing_aborts_without_destroying_anything() {
-    let fixture = Fixture::new(
-        r#"
-case "$*" in
-  'list -Hp -t filesystem -o name') printf '%s\n' pool/home ;;
-  'snapshot pool/home@auto_'*) ;;
-  *) printf '%s\n' pool/home@auto_200002020115; exit 1 ;;
-esac
-"#,
-    );
-    let output = fixture.run();
-    assert_eq!(output.status.code(), Some(1), "{}", stderr(&output));
-    assert!(stderr(&output).contains("snapshot_manager failed"));
-    assert_eq!(fixture.calls().len(), 3);
-}
-
-#[test]
-fn invalid_config_stops_before_any_zfs_commands() {
-    let fixture = Fixture::new("exit 99");
-    fs::write(
-        &fixture.config,
-        "[default]\n15_min = -1\nhourly = 1\ndaily = 0\nmonthly = 0",
-    )
-    .unwrap();
-    let output = fixture.run();
-    assert_eq!(output.status.code(), Some(1), "{}", stderr(&output));
-    assert!(stderr(&output).contains("Invalid configuration"));
-    assert!(fixture.calls().is_empty());
-}
-
-#[test]
-fn missing_zfs_executable_fails_and_reports_the_error() {
-    let fixture = Fixture::new("exit 99");
-    fs::remove_file(fixture.dir.path().join("zfs")).unwrap();
-    let output = fixture.run();
-    assert_eq!(output.status.code(), Some(1), "{}", stderr(&output));
-    assert!(stderr(&output).contains("Failed to execute zfs"));
-    assert!(stderr(&output).contains("snapshot_manager failed"));
-    assert!(fixture.calls().is_empty());
-}
-
-#[test]
-fn dataset_names_are_passed_as_one_argument_without_a_shell() {
-    let fixture = Fixture::new(
-        r#"
-case "$1" in
-  list)
-    [ "$*" = 'list -Hp -t filesystem -o name' ] || exit 98
-    printf '%s\n' 'pool/two words;touch injected' ;;
-  snapshot)
-    [ "$#" -eq 2 ] || exit 97
-    case "$2" in
-      'pool/two words;touch injected@auto_'*) printf '%s\n' 'argument preserved' >&2; exit 1 ;;
-      *) exit 96 ;;
-    esac ;;
-  *) exit 99 ;;
-esac
-"#,
-    );
-    let output = fixture.run();
-    assert!(output.status.success(), "{}", stderr(&output));
-    assert!(stderr(&output).contains("argument preserved"));
-    assert_eq!(
-        fixture.normalized_calls(),
-        [
-            "list -Hp -t filesystem -o name",
-            "snapshot pool/two words;touch injected@<timestamp>"
-        ]
-    );
-    assert!(fs::read_dir(fixture.dir.path()).unwrap().all(|entry| {
-        !entry
-            .unwrap()
-            .file_name()
-            .to_string_lossy()
-            .starts_with("injected")
-    }));
-}

rust/tls/Cargo.toml

@@ -1,14 +0,0 @@
-[package]
-name = "tls"
-version = "0.1.0"
-edition = "2024"
-description = "Shared AWS-LC TLS provider initialization"
-license = "MIT"
-publish = false
-
-[dependencies]
-anyhow.workspace = true
-rustls.workspace = true
-
-[lints]
-workspace = true

rust/tls/src/lib.rs

@@ -1,16 +0,0 @@
-//! Shared TLS configuration for workspace applications.
-
-use anyhow::{Result, anyhow};
-
-/// Install AWS-LC as the process-wide default TLS crypto provider.
-///
-/// Call once at startup, before creating HTTP clients.
-///
-/// # Errors
-///
-/// Returns an error if a TLS crypto provider has already been initialized.
-pub fn install_tls_provider() -> Result<()> {
-    rustls::crypto::aws_lc_rs::default_provider()
-        .install_default()
-        .map_err(|_| anyhow!("TLS crypto provider was already initialized"))
-}

rust/van_weather/Cargo.toml

@@ -6,16 +6,16 @@ description = "Fetch privacy-masked weather for a van and publish it to Home Ass
 license = "MIT"
 
 [dependencies]
-anyhow.workspace = true
-chrono.workspace = true
-clap = { workspace = true, features = ["derive", "env"] }
-env_logger.workspace = true
-log.workspace = true
-num-traits = "0.2"
-reqwest = { workspace = true, features = ["blocking", "json"] }
-serde = { workspace = true, features = ["derive"] }
-serde_json.workspace = true
-tls.workspace = true
+anyhow = "1.0"
+chrono = "0.4"
+clap = { version = "4.5", features = ["derive", "env"] }
+env_logger = "0.11"
+log = "0.4"
+reqwest = { version = "0.12", default-features = false, features = ["blocking", "json", "rustls-tls"] }
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+
+[lints.clippy]
+all = "deny"
+pedantic = "deny"
 
-[lints]
-workspace = true

rust/van_weather/package.nix

@@ -1,18 +1,10 @@
-{ lib, rustPlatform }:
+{ rustPlatform }:
 rustPlatform.buildRustPackage {
   pname = "van-weather";
   version = "0.1.0";
-  src = import ../rust-source.nix { inherit lib; };
+  src = ./.;
 
-  cargoLock.lockFile = ../../Cargo.lock;
-  cargoBuildFlags = [
-    "-p"
-    "van-weather"
-  ];
-  cargoTestFlags = [
-    "-p"
-    "van-weather"
-  ];
+  cargoLock.lockFile = ./Cargo.lock;
 
   meta = {
     description = "Privacy-masked van weather publisher for Home Assistant";

rust/van_weather/src/main.rs

@@ -4,7 +4,6 @@ use anyhow::{Context, Result, bail};
 use chrono::{DateTime, Utc};
 use clap::Parser;
 use log::{error, info};
-use num_traits::ToPrimitive;
 use reqwest::{
     StatusCode, Url,
     blocking::{Client, ClientBuilder},
@@ -13,7 +12,6 @@ use reqwest::{
 };
 use serde::Deserialize;
 use serde_json::{Value, json};
-use tls::install_tls_provider;
 
 const LAT_ENTITY: &str = "sensor.van_last_known_latitude";
 const LON_ENTITY: &str = "sensor.van_last_known_longitude";
@@ -27,7 +25,7 @@ const fn decimal_factor(decimals: u32) -> f64 {
     let mut remaining = decimals;
     while remaining > 0 {
         factor *= 10.0;
-        remaining = remaining.saturating_sub(1);
+        remaining -= 1;
     }
     factor
 }
@@ -160,7 +158,6 @@ fn main() -> Result<()> {
     } = Args::parse();
     env_logger::Builder::from_env(env_logger::Env::default().default_filter_or(log_level.as_str()))
         .init();
-    install_tls_provider()?;
 
     let ha_url = ha_url.trim_end_matches('/').to_owned();
     let ha_host = Url::parse(&ha_url)
@@ -471,39 +468,24 @@ fn weather_updates(weather: &Weather) -> Vec<(String, Value)> {
 }
 
 fn sensor(entity_id: &str, state: Option<Value>, attributes: Value) -> Option<(String, Value)> {
-    let state = state?;
-    let mut payload = serde_json::Map::new();
-    payload.insert("state".to_owned(), state);
-    payload.insert("attributes".to_owned(), attributes);
-    Some((entity_id.to_owned(), Value::Object(payload)))
+    state.map(|state| {
+        let mut payload = serde_json::Map::new();
+        payload.insert("state".to_owned(), state);
+        payload.insert("attributes".to_owned(), attributes);
+        (entity_id.to_owned(), Value::Object(payload))
+    })
 }
 
+#[allow(clippy::cast_possible_truncation)]
 fn percent(value: Option<f64>) -> i64 {
     // Preserve Python's int(probability * 100) behavior for Home Assistant.
-    let value = value.unwrap_or_default() * 100.0;
-    value.to_i64().unwrap_or_else(|| {
-        if value.is_nan() {
-            0
-        } else if value.is_sign_negative() {
-            i64::MIN
-        } else {
-            i64::MAX
-        }
-    })
+    (value.unwrap_or_default() * 100.0) as i64
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
 
-    #[test]
-    fn builds_http_clients_with_aws_lc() -> Result<()> {
-        install_tls_provider()?;
-        build_client("home-assistant.example", Some("test-token"))?;
-        build_client(PIRATE_WEATHER_HOST, None)?;
-        Ok(())
-    }
-
     #[test]
     fn maps_conditions_for_home_assistant() {
         assert_eq!(condition(Some("clear-day")), "sunny");
@@ -580,21 +562,4 @@ mod tests {
         assert!((mask_coordinate(37.7749) - 37.8).abs() < f64::EPSILON);
         assert!((mask_coordinate(-122.4194) - (-122.4)).abs() < f64::EPSILON);
     }
-
-    #[test]
-    fn percentages_truncate_and_preserve_saturation_for_extreme_values() {
-        for (value, expected) in [
-            (None, 0),
-            (Some(0.129), 12),
-            (Some(-0.129), -12),
-            (Some(1.0), 100),
-            (Some(f64::NAN), 0),
-            (Some(f64::MAX), i64::MAX),
-            (Some(f64::MIN), i64::MIN),
-            (Some(f64::INFINITY), i64::MAX),
-            (Some(f64::NEG_INFINITY), i64::MIN),
-        ] {
-            assert_eq!(percent(value), expected);
-        }
-    }
 }

rust/zfs/Cargo.toml

@@ -1,13 +0,0 @@
-[package]
-name = "zfs"
-version = "0.1.0"
-edition = "2024"
-description = "List ZFS datasets and manage snapshots through the zfs executable"
-license = "MIT"
-publish = false
-
-[dependencies]
-anyhow.workspace = true
-
-[lints]
-workspace = true

rust/zfs/src/lib.rs

@@ -1,100 +0,0 @@
-//! Manage ZFS datasets and snapshots using the `zfs` executable on `PATH`.
-//!
-//! ```no_run
-//! for snapshot in zfs::list_snapshots("pool/home")? {
-//!     println!("{snapshot}");
-//! }
-//! # Ok::<(), anyhow::Error>(())
-//! ```
-
-use std::process::Command;
-
-use anyhow::{Context, Result, bail};
-
-const ZFS_EXECUTABLE: &str = "zfs";
-
-fn zfs_command(args: &[&str]) -> Result<String> {
-    let output = Command::new(ZFS_EXECUTABLE)
-        .args(args)
-        .env("LC_ALL", "C")
-        .output()
-        .context("Failed to execute zfs")?;
-    if !output.status.success() {
-        let message = if output.stderr.is_empty() {
-            &output.stdout
-        } else {
-            &output.stderr
-        };
-        bail!(
-            "zfs {} failed ({}): {}",
-            args.join(" "),
-            output.status,
-            String::from_utf8_lossy(message).trim()
-        );
-    }
-    String::from_utf8(output.stdout).context("zfs returned invalid UTF-8")
-}
-
-/// List filesystem datasets, excluding pool roots.
-///
-/// # Errors
-///
-/// Returns an error if executing ZFS or decoding its output fails.
-pub fn list_datasets() -> Result<Vec<String>> {
-    let output = zfs_command(&["list", "-Hp", "-t", "filesystem", "-o", "name"])?;
-    Ok(output
-        .lines()
-        .filter(|name| name.contains('/'))
-        .map(String::from)
-        .collect())
-}
-
-/// List snapshot names belonging directly to `dataset`, excluding child datasets.
-///
-/// Returned names omit the `dataset@` prefix.
-///
-/// # Errors
-///
-/// Returns an error if executing ZFS or decoding its output fails.
-pub fn list_snapshots(dataset: &str) -> Result<Vec<String>> {
-    let output = zfs_command(&[
-        "list", "-Hp", "-t", "snapshot", "-o", "name", "-d", "1", dataset,
-    ])?;
-    Ok(output
-        .lines()
-        .filter_map(|name| {
-            let (owner, snapshot) = name.split_once('@')?;
-            (owner == dataset).then(|| snapshot.to_owned())
-        })
-        .collect())
-}
-
-/// Create `dataset@snapshot`.
-///
-/// # Errors
-///
-/// Returns an error if executing ZFS or decoding its output fails.
-pub fn create_snapshot(dataset: &str, snapshot: &str) -> Result<()> {
-    zfs_command(&["snapshot", &format!("{dataset}@{snapshot}")])?;
-    Ok(())
-}
-
-/// Delete `dataset@snapshot`.
-///
-/// Returns `None` on success, or `Some("snapshot has dependent clones")` when
-/// dependent clones prevent deletion.
-///
-/// # Errors
-///
-/// Returns an error for other ZFS failures, execution failures, or invalid UTF-8 output.
-pub fn delete_snapshot(dataset: &str, snapshot: &str) -> Result<Option<&'static str>> {
-    if let Err(error) = zfs_command(&["destroy", &format!("{dataset}@{snapshot}")]) {
-        if error.to_string().contains(&format!(
-            "cannot destroy '{dataset}@{snapshot}': snapshot has dependent clones"
-        )) {
-            return Ok(Some("snapshot has dependent clones"));
-        }
-        return Err(error);
-    }
-    Ok(None)
-}

systems/bob/secrets.yaml

@@ -1,25 +0,0 @@
-system_token: ENC[AES256_GCM,data:zpyosU2oJTljDBbjeQs79gpzPtfL6Y41nZ3X1d+IgdI1fczx/F6MUOaBN6xVkVGYNHhaaCKaPmS/X7J8TDJHQGWPxM0ZsPluGGCnxhQjsqyY4oc3jyBkye7IbnnWmj1yhisq2V1ti5W7uewCQBioGqJYCNo4PfEjp6ipcgaI5no=,iv:Ee6nw88SUL8+80RR0YA4vAAGilGf7bY6EPhE81jD1Ic=,tag:NtRwWqCntdQN9BlakxvslQ==,type:str]
-sops:
-    age:
-        - enc: |
-            -----BEGIN AGE ENCRYPTED FILE-----
-            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBtdjNXQ2VpUlBYMjZJbmRT
-            ajk2WEY5VndyU0o0aWdYbXRUSnE0QWxGNlYwCnBkOVd1VkMweTdIZHZBTmp5akMv
-            TE5SRStHYndXWm0wOG82TDlRa2dHV1kKLS0tIEwveXkxMkpULzhBZWQ5WXJQOGFm
-            TjVmQUkzQmJHYXgwSjF1T1dqMlNUVUEKCz8bv8tdtNPRLSXCNoyEY9yTdpFBg1uG
-            EpZ/VgsBV0LktJkqkv6ZSRWp2HiEC61305GEFxrE3u4w7ha6FLJMrQ==
-            -----END AGE ENCRYPTED FILE-----
-          recipient: age1u8zj599elqqvcmhxn8zuwrufsz8w8w366d3ayrljjejljt2q45kq8mxw9c
-        - enc: |
-            -----BEGIN AGE ENCRYPTED FILE-----
-            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBrLzVQSFZQWTZ2ZnVLdEta
-            S0tkMElVMzZJQ2xSeElQL3N3dy9hQWxzVld3ClcySjhvT0RyUUk4SlNPd2Jpdzhv
-            QU5samVSQnFEdk80M3diazVEVUVFT1UKLS0tIEpLdE82dUFuT0t2NGxqYmxSbis1
-            MFAxN3BtRHVKMWxTeVBpblJDTUcvK0UKZTfYCDTAijn0UPK+Z7UjWOoODPf746Q9
-            ya085jHs39Z1597tgJzWIgydkHl6x9VYUkDu+1Yq4LOeyHFscELnJQ==
-            -----END AGE ENCRYPTED FILE-----
-          recipient: age1q47vup0tjhulkg7d6xwmdsgrw64h4ax3la3evzqpxyy4adsmk9fs56qz3y
-    lastmodified: "2026-10-14T02:07:23Z"
-    mac: ENC[AES256_GCM,data:o21tNt4Cs5DpCMcWCFjCQ3DRZZBCftFgm2P80y849sFF8rfFVZy99Rda0fiPpohRs45GsvDxPaDAxcCJl2N31pQ3Zg8VCKNS1LbBoDI3iqhuUtmKeudGfOUY6Gd1+j/E4ML29Cfrzm6r3WiXNPa7q8Jmq/kspwoubpQF2q9xp3o=,iv:6hbw+cRlEAdSaPvfh1sMoORI1uhQkqmpOK0aIhL+f+c=,tag:wTa/39ToJSOmn7b31+m34Q==,type:str]
-    unencrypted_suffix: _unencrypted
-    version: 3.13.3

systems/brain/secrets.yaml

@@ -1,25 +0,0 @@
-system_token: ENC[AES256_GCM,data:qub1LgQ89K1EmKk8y/sE1OcPdEB5Zu/S/iZENBA+5d3NQUxJwhX3f3xTo4syBsYFk3b4X1AVx8X0JK27fyr36f9QKPc8IR3Kt4foNp+96gJs9wzskJ188yGFGn1Tle9BLkF1/l5Pzt8Xwe4VCOGuT9uh74q2sCUgcC52u+aez6E=,iv:AbAunuRTY9lhcnd2MwOyVv4in0MfyxwH+LsNK0DoMbQ=,tag:VAmVTSpPh734cZt5Vy0mTg==,type:str]
-sops:
-    age:
-        - enc: |
-            -----BEGIN AGE ENCRYPTED FILE-----
-            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBjWGNFR01XRXBJUDZ0cFd6
-            dzdWVUlHT1NQU2VrV3VRSlF2bThYdWRtTERJClI1cFFTZzRNSlc1QmlnQUNxSHZB
-            dVhqejBMcDZVVXdIeU0yM3NxelBhMTgKLS0tIFJWRHdqTW5UY3ZOZUZ5dXJKRity
-            YXhHOWl0Y0dQOS9uN0VCSmJHWkxXd0EKV5LbdNZ5oujX+rey8i/XUxoOKw99JKJJ
-            N1sIIc4LmXrqeqRRhXAakQYoWEqQa12Vu3OZrPXkvoQ0KQHKcKbPPA==
-            -----END AGE ENCRYPTED FILE-----
-          recipient: age1u8zj599elqqvcmhxn8zuwrufsz8w8w366d3ayrljjejljt2q45kq8mxw9c
-        - enc: |
-            -----BEGIN AGE ENCRYPTED FILE-----
-            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBhVkNnUkZEbk8rcm90dFAy
-            Ty9vYWN4YkIxdVUyTlJCak4yMmx2dkZXUFFRCmZTRkJReGRwTXB1ZThYSjVmUC9D
-            L1JkVEhHUjRjenpBS0hkQWRiVXY1dEEKLS0tIGM1S2VYbHBjbDU2RG5MUC9HcDNW
-            VEFCR3ZXVzNOMVY3Q0VEM0Y4dnROZUEKKpULO8QALBkItUgkT7IQgPsC/3gQKfid
-            1dgv+CpFGLXDtS8Ocm/r0y6WL2ds1E5rwoQ4UphnxWJGLIAo8cs36g==
-            -----END AGE ENCRYPTED FILE-----
-          recipient: age1jhf7vm0005j60mjq63696frrmjhpy8kpc2d66mw044lqap5mjv4snmwvwm
-    lastmodified: "2026-10-14T02:08:28Z"
-    mac: ENC[AES256_GCM,data:fBhvyzmnzQtLZ7QNELrJvFfnfymLjfzb7Bi61+6dCqNNgrMnYI4gU+9xDSepHskWQ+9ZS3fHZ8ZQCMC+xjhEvI8PWVe4nQYFKHkNOOrfZSBb8osWyzChbQCQXJtK1zalfo/Yk9gvOzeGhqM+XOqMt15Wr56eETSPpPNwVdG2QPY=,iv:TRCrl90dFGJiP0RsYpEIuvknCh8tslIVWTu+Loeklbw=,tag:LkyzzSEnytMkz40IPeBM6g==,type:str]
-    unencrypted_suffix: _unencrypted
-    version: 3.13.3

systems/brain/services/home_assistant.nix

@@ -13,10 +13,6 @@
   services = {
     home-assistant = {
       enable = true;
-      # Keep Home Assistant's Python dependencies on baseline x86-64.
-      package = pkgs.x86-v1.home-assistant.overrideAttrs (_: {
-        doInstallCheck = false;
-      });
       config = {
         homeassistant = {
           time_zone = "America/New_York";
@@ -83,7 +79,7 @@
         "isal"
         "modbus" # for victron modbus integration
       ];
-      customComponents = with pkgs.x86-v1.home-assistant-custom-components; [
+      customComponents = with pkgs.home-assistant-custom-components; [
         garmin_connect
       ];
 

systems/jeeves/default.nix

@@ -38,5 +38,10 @@ in
     };
   };
 
+  users.groups = {
+    nornsight = { };
+    nornsight-admin = { };
+  };
+
   system.stateVersion = "24.05";
 }

systems/jeeves/networking/firewall.nix

@@ -6,10 +6,10 @@ let
     gitea_http = 6443;
     gitea_ssh = 2223;
     grafana = 3000;
-    haproxy_ingest = 8444;
     jellyfin_http = 8096;
     jellyswarrm = 3100;
     nix_binary_cache = 5000;
+    nornsight = 8001;
     ollama = 11434;
     open_webui = 8080;
     postgresql = 5432;
@@ -27,10 +27,10 @@ in
     interfaces.tailscale0.allowedTCPPorts = [
       ports.audiobookshelf
       ports.gems
-      ports.haproxy_ingest
       ports.jellyfin_http
       ports.jellyswarrm
       ports.nix_binary_cache
+      ports.nornsight
       ports.share
     ];
     allowedTCPPorts = [

systems/jeeves/runners/nix_builder.nix

@@ -107,7 +107,6 @@ let
           nixos-rebuild
           nodejs
           ruff
-          rustfmt
           treefmt
           wget
         ];

systems/jeeves/secrets.yaml

@@ -1,25 +0,0 @@
-system_token: ENC[AES256_GCM,data:aDNBcQuFE3ImFMOs83cmyeWcANavFoYFy564zSq8XcgFL2VUho2D8UtW/e+hTdJBEjeNTd7gSrtFd/Ved6QgZHgPCyxUmNkPcr6XqTe6a+AFWCx7tVSWYSweKt7BI5GcZ4oA7oeOJ+/9Tefbo5Yg0QI1HgKcS3D7BHEdr5iQi4E=,iv:lvBbttBqzIMvsFobUZQfbaYCyDrjztB/cb3r2P1TtwA=,tag:AEI1Rj6/A6f0K9MBx9Ai0w==,type:str]
-sops:
-    age:
-        - enc: |
-            -----BEGIN AGE ENCRYPTED FILE-----
-            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBJQnVqUkxRUm5PQ21lV01o
-            OEJuMWtSWlE1Z09RaUxNeHpycU1DcnM1dGlVCmEvUUlmcmxnWHRYS016U3RhYkw1
-            enZQcnJrSUwybGpKWkx4bm9ra2lVWGcKLS0tIDRDKzl0bEp6YnhweTN4YW9LU2hI
-            alF5V3BIQ0R5MXZFSmt6dnRXanp0Ym8KQhrtXzQLEG2pXzKs1Z8i9DPigWKpiovZ
-            dfAw2Tzp4B8kTHuo4PuYU/S37HoYYhf5ZV5vRiUoNYhrr6Q8wwu1Iw==
-            -----END AGE ENCRYPTED FILE-----
-          recipient: age1u8zj599elqqvcmhxn8zuwrufsz8w8w366d3ayrljjejljt2q45kq8mxw9c
-        - enc: |
-            -----BEGIN AGE ENCRYPTED FILE-----
-            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBod2dvaVhxNW1tWld0eW1x
-            TWZuNFJWWndQSDVKWU9WdjlTWW13RU9RZ2gwCjJPb0VUTmlMWGFJTEFFdE1jbHRT
-            YllSQXBNeUFFSk9JYmdJSGVKQVhSU0EKLS0tIG16T2R6MzJKVTE3ZDhCVU93WXlB
-            QW13QVc5dUpVTFFMZ0l3dFlPY2NOYXcK1IdjpBBhT2l5FvFNwLR+XH3AqbUivYNd
-            pQwOr6T7ysB70h12EKZ5V4Ti4BT8ETeQUgFC2JbX2OZe0cTrxtacQw==
-            -----END AGE ENCRYPTED FILE-----
-          recipient: age13lmqgc3jvkyah5e3vcwmj4s5wsc2akctcga0lpc0x8v8du3fxprqp4ldkv
-    lastmodified: "2026-10-14T02:08:46Z"
-    mac: ENC[AES256_GCM,data:4eVms5du2CbLqdY+kQYo6kcDRIZD+h950tInjESug9kjkSBQhWCpJLrIM1Ud9FYtqPE59rsd4s9wlDRY1+hpvBg3YgdcKjSBR9sNXL6D5cCE+2aYfTrkVgkcEwWeFc2OauXjkrqR2QXpq5ltq/3He6Hxdr08kC37GunvZEfJwzg=,iv:3qkVvRPUP/C3s5+a7dpOmoqaXbnWKxgM7OF8noxxMpw=,tag:jUJsLpLr8rxii0Hh6E9bXw==,type:str]
-    unencrypted_suffix: _unencrypted
-    version: 3.13.3

systems/jeeves/services/gitea.nix

@@ -23,6 +23,7 @@ in
       };
       service.DISABLE_REGISTRATION = true;
       server = {
+        DOMAIN = "gitea.tmmworkshop.com";
         ROOT_URL = "https://gitea.tmmworkshop.com/";
         HTTP_PORT = 6443;
         BUILTIN_SSH_SERVER_USER = "gitea";

systems/jeeves/services/haproxy_ingest.nix

@@ -1,82 +0,0 @@
-{
-  pkgs,
-  ...
-}:
-let
-  vars = import ../vars.nix;
-  haproxy_ingest_server = pkgs.callPackage ../../../rust/haproxy_ingest_server/package.nix { };
-in
-{
-  users = {
-    groups.haproxy-ingest = { };
-    users.haproxy-ingest = {
-      isSystemUser = true;
-      group = "haproxy-ingest";
-    };
-  };
-
-  systemd = {
-    services = {
-      haproxy-ingest-server = {
-        description = "HAProxy log ingestion API";
-        documentation = [ "file://${../../../rust/haproxy_ingest_proto/README.md}" ];
-        wantedBy = [ "multi-user.target" ];
-        after = [
-          "network-online.target"
-          "postgresql.service"
-          "tailscaled-autoconnect.service"
-        ];
-        wants = [
-          "network-online.target"
-          "postgresql.service"
-          "tailscaled-autoconnect.service"
-        ];
-
-        serviceConfig = {
-          Type = "simple";
-          User = "haproxy-ingest";
-          Group = "haproxy-ingest";
-          ExecStart = "${haproxy_ingest_server}/bin/haproxy-ingest-server";
-          Environment = [
-            "HAPROXY_INGEST_DATABASE_URL=postgres://haproxy_ingest@localhost/haproxy_ingest?host=/run/postgresql&sslmode=disable"
-            "HAPROXY_INGEST_LISTEN=0.0.0.0:8444"
-            "RUST_LOG=info,tower_http=warn"
-          ];
-          EnvironmentFile = "${vars.secrets}/services/haproxy-ingest-server";
-          Restart = "on-failure";
-          RestartSec = "5s";
-
-          CapabilityBoundingSet = [ "" ];
-          DeviceAllow = [ "" ];
-          LockPersonality = true;
-          MemoryDenyWriteExecute = true;
-          NoNewPrivileges = true;
-          PrivateDevices = true;
-          PrivateTmp = true;
-          ProtectClock = true;
-          ProtectControlGroups = true;
-          ProtectHome = true;
-          ProtectHostname = true;
-          ProtectKernelLogs = true;
-          ProtectKernelModules = true;
-          ProtectKernelTunables = true;
-          ProtectSystem = "strict";
-          RemoveIPC = true;
-          RestrictAddressFamilies = [
-            "AF_INET"
-            "AF_UNIX"
-          ];
-          RestrictNamespaces = true;
-          RestrictRealtime = true;
-          RestrictSUIDSGID = true;
-          SystemCallArchitectures = "native";
-          SystemCallFilter = [
-            "@system-service"
-            "~@privileged"
-          ];
-          UMask = "0077";
-        };
-      };
-    };
-  };
-}

systems/jeeves/services/nornsight.nix

@@ -0,0 +1,107 @@
+{ pkgs, ... }:
+let
+  vars = import ../vars.nix;
+  stateDir = "${vars.services}/nornsight";
+  appDir = "${stateDir}/app";
+  binPath = pkgs.lib.makeBinPath [
+    pkgs.binutils
+    pkgs.libpq
+    pkgs.postgresql
+    pkgs.stdenv.cc
+  ];
+  libraryPath = pkgs.lib.makeLibraryPath [
+    pkgs.libpq
+    pkgs.postgresql.lib
+  ];
+in
+{
+  systemd.tmpfiles.rules = [
+    "d ${stateDir} 0750 nornsight nornsight - -"
+  ];
+
+  users.users.nornsight = {
+    isSystemUser = true;
+    group = "nornsight";
+    home = stateDir;
+  };
+
+  systemd.services.nornsight = {
+    description = "Norn Sight";
+    after = [ "network-online.target" ];
+    wants = [ "network-online.target" ];
+    wantedBy = [ "multi-user.target" ];
+
+    environment = {
+      HOME = stateDir;
+      UV_CACHE_DIR = "${stateDir}/.cache/uv";
+      UV_PROJECT_ENVIRONMENT = "${appDir}/.venv";
+      UV_PYTHON = "${pkgs.python313}/bin/python3.13";
+      UV_PYTHON_DOWNLOADS = "never";
+      LD_LIBRARY_PATH = libraryPath;
+      LIBRARY_PATH = libraryPath;
+      PSYCOPG_IMPL = "python";
+    };
+
+    path = with pkgs; [
+      bash
+      coreutils
+      git
+      uv
+    ];
+
+    serviceConfig = {
+      Type = "simple";
+      User = "nornsight";
+      Group = "nornsight";
+      EnvironmentFile = "-${vars.secrets}/services/nornsight";
+      WorkingDirectory = stateDir;
+      Restart = "on-failure";
+      RestartSec = "5s";
+      StandardOutput = "journal";
+      StandardError = "journal";
+      NoNewPrivileges = true;
+      PrivateTmp = true;
+      ProtectHome = true;
+      ProtectSystem = "strict";
+      ReadWritePaths = [ stateDir ];
+    };
+
+    script = ''
+      set -eu
+      export PATH="${binPath}:$PATH"
+      export LD_LIBRARY_PATH="${libraryPath}:''${LD_LIBRARY_PATH:-}"
+      export LIBRARY_PATH="${libraryPath}:''${LIBRARY_PATH:-}"
+
+      : "''${NORN_SIGHT_REPO_URL:?NORN_SIGHT_REPO_URL is required}"
+      branch="''${NORN_SIGHT_BRANCH:-main}"
+
+      if [ -d "${appDir}/.git" ]; then
+        current_origin="$(git -C "${appDir}" remote get-url origin)"
+        if [ "$current_origin" != "$NORN_SIGHT_REPO_URL" ]; then
+          rm -rf "${appDir}"
+        fi
+      fi
+
+      if [ ! -d "${appDir}/.git" ]; then
+        git clone --branch "$branch" "$NORN_SIGHT_REPO_URL" "${appDir}"
+      else
+        cd "${appDir}"
+        git fetch origin "$branch"
+        git checkout "$branch"
+        git pull --ff-only origin "$branch"
+      fi
+
+      cd "${appDir}"
+      uv sync --upgrade
+      uv run python - <<'PY'
+      import ctypes.util
+      import os
+
+      print(f"LD_LIBRARY_PATH={os.environ.get('LD_LIBRARY_PATH')}")
+      print(f"LIBRARY_PATH={os.environ.get('LIBRARY_PATH')}")
+      print(f"libpq={ctypes.util.find_library('pq')}")
+      PY
+      exec uv run uvicorn pipelines.web.main:app --host 0.0.0.0 --port 8001
+    '';
+  };
+}

systems/jeeves/services/postgress.nix

@@ -36,9 +36,6 @@ in
       # signalbot
       local signalbot  signalbot   trust
 
-      # HAProxy cloud log ingestion API
-      local haproxy_ingest haproxy_ingest trust
-
       # math
       local postgres  math   trust
       host  postgres  math   127.0.0.1/32    trust
@@ -100,13 +97,6 @@ in
           replication = true;
         };
       }
-      {
-        name = "haproxy_ingest";
-        ensureDBOwnership = true;
-        ensureClauses = {
-          login = true;
-        };
-      }
       {
         name = "signalbot";
         ensureDBOwnership = true;
@@ -118,7 +108,6 @@ in
     ensureDatabases = [
       "data_science_dev"
       "gitea"
-      "haproxy_ingest"
       "math"
       "n8n"
       "richie"

systems/portal-1/haproxy/default.nix

@@ -1,13 +1,14 @@
 let
   domains = [
     "audiobookshelf"
-    "auth"
     "cache"
     "gitea"
     "gems"
     "jellyfin"
     "share"
   ];
+  extraDomains = [ "www.norn-sight.com" ];
+
   makeCert = name: {
     name = "${name}.tmmworkshop.com";
     value = {
@@ -17,11 +18,20 @@ let
     };
   };
 
-  acmeServices = map (domain: "acme-${domain}.tmmworkshop.com.service") domains;
+  makeExtraCert = name: {
+    inherit name;
+    value = {
+      webroot = "/var/lib/acme/.challenges";
+      group = "acme";
+      reloadServices = [ "haproxy.service" ];
+    };
+  };
+
+  acmeServices =
+    map (domain: "acme-${domain}.tmmworkshop.com.service") domains
+    ++ map (domain: "acme-${domain}.service") extraDomains;
 in
 {
-  imports = [ ./ingest.nix ];
-
   networking.firewall.allowedTCPPorts = [
     80
     443
@@ -55,7 +65,7 @@ in
   security.acme = {
     acceptTerms = true;
     defaults.email = "Richie@tmmworkshop.com";
-    certs = builtins.listToAttrs ((map makeCert domains));
+    certs = builtins.listToAttrs ((map makeCert domains) ++ (map makeExtraCert extraDomains));
   };
 
   systemd = {

systems/portal-1/haproxy/haproxy.cfg

@@ -1,21 +1,12 @@
 global
   log stdout format raw local0
-  # Structured HTTP log fields, grouped so the format remains editable here.
-  setenv HAPROXY_JSON_CLIENT "%(client_ip)ci %(client_port)cp"
-  setenv HAPROXY_JSON_ROUTING "%(frontend_name)ft %(backend_name)b %(server_name)s"
-  setenv HAPROXY_JSON_HTTP "%(http_method)HM %(request_path)HP %(http_version)HV %(status_code)ST %(bytes_sent)B"
-  setenv HAPROXY_JSON_TIMINGS "%(time_request_ms)TR %(time_wait_ms)Tw %(time_connect_ms)Tc %(time_response_ms)Tr %(time_total_ms)Ta"
-  setenv HAPROXY_JSON_RESULT "%(termination_state)tsc %(retries)rc"
-  setenv HAPROXY_JSON_CONNECTIONS "%(active_connections)ac %(frontend_connections)fc %(backend_connections)bc %(server_connections)sc %(server_queue)sq %(backend_queue)bq"
-  setenv HAPROXY_JSON_TLS "%(tls_sni)[ssl_fc_sni] %(tls_version)sslv %(tls_cipher)sslc %(tls_resumed)[ssl_fc_is_resumed]"
-  setenv HAPROXY_JSON_LOG "${HAPROXY_JSON_CLIENT} ${HAPROXY_JSON_ROUTING} ${HAPROXY_JSON_HTTP} ${HAPROXY_JSON_TIMINGS} ${HAPROXY_JSON_RESULT} ${HAPROXY_JSON_CONNECTIONS} ${HAPROXY_JSON_TLS}"
   # stats socket /run/haproxy/admin.sock mode 660 level admin expose-fd listeners
   stats timeout 30s
 
 defaults
   log global
   mode http
-  log-format "%{+json}o ${HAPROXY_JSON_LOG}"
+  option httplog
   retries 3
   maxconn 2000
   timeout connect 5s
@@ -32,7 +23,7 @@ defaults
 #Application Setup
 frontend ContentSwitching
   bind [::]:80 v4v6
-  bind [::]:443 v4v6 ssl crt /var/lib/acme/audiobookshelf.tmmworkshop.com/full.pem crt /var/lib/acme/cache.tmmworkshop.com/full.pem crt /var/lib/acme/gems.tmmworkshop.com/full.pem crt /var/lib/acme/jellyfin.tmmworkshop.com/full.pem crt /var/lib/acme/share.tmmworkshop.com/full.pem crt /var/lib/acme/gitea.tmmworkshop.com/full.pem crt
+  bind [::]:443 v4v6 ssl crt /var/lib/acme/audiobookshelf.tmmworkshop.com/full.pem crt /var/lib/acme/cache.tmmworkshop.com/full.pem crt /var/lib/acme/gems.tmmworkshop.com/full.pem crt /var/lib/acme/jellyfin.tmmworkshop.com/full.pem crt /var/lib/acme/share.tmmworkshop.com/full.pem crt /var/lib/acme/gitea.tmmworkshop.com/full.pem crt /var/lib/acme/www.norn-sight.com/full.pem
   mode  http
 
   # ACME challenge routing (must be first)
@@ -45,6 +36,7 @@ frontend ContentSwitching
   acl host_share  hdr(host) -i share.tmmworkshop.com
   acl host_gitea  hdr(host) -i gitea.tmmworkshop.com
   acl host_gems  hdr(host) -i gems.tmmworkshop.com
+  acl host_norn_sight  hdr(host) -i www.norn-sight.com
 
   # --- Request logging ---
   # Capture the Host header and User-Agent so the httplog shows who is
@@ -116,6 +108,7 @@ frontend ContentSwitching
   use_backend share_nodes  if host_share
   use_backend gitea  if host_gitea
   use_backend gems  if host_gems
+  use_backend norn_sight  if host_norn_sight
 
 # Proxy Gitea's built-in SSH server over Tailscale to Jeeves. Port 22 remains
 # available to Endlessh on Portal.
@@ -167,3 +160,7 @@ backend gems
   option forwardfor
   timeout server 1h
   server gems jeeves:8002
+
+backend norn_sight
+  mode http
+  server server jeeves:8001

systems/portal-1/haproxy/ingest.nix

@@ -1,103 +0,0 @@
-{ pkgs, ... }:
-let
-  haproxy_log_shipper = pkgs.callPackage ../../../rust/haproxy_log_shipper/package.nix { };
-  endpoint = "http://jeeves.taile39c31.ts.net:8444";
-  environment_file = "/var/lib/haproxy-log-shipper/environment";
-in
-{
-  services.journald.settings.Journal = {
-    Storage = "persistent";
-    SystemMaxUse = "512M";
-    MaxRetentionSec = "14day";
-  };
-
-  users = {
-    groups.haproxy-log-shipper = { };
-    users.haproxy-log-shipper = {
-      isSystemUser = true;
-      group = "haproxy-log-shipper";
-      extraGroups = [ "systemd-journal" ];
-    };
-  };
-
-  systemd = {
-    services = {
-      haproxy = {
-        serviceConfig = {
-          StandardOutput = "journal";
-          StandardError = "journal";
-        };
-      };
-
-      haproxy-log-shipper = {
-        description = "Ship ordered HAProxy journal logs to Jeeves";
-        documentation = [ "file://${../../../rust/haproxy_ingest_proto/README.md}" ];
-        after = [
-          "network-online.target"
-          "tailscaled-autoconnect.service"
-        ];
-        wants = [
-          "network-online.target"
-          "tailscaled-autoconnect.service"
-        ];
-
-        serviceConfig = {
-          Type = "oneshot";
-          User = "haproxy-log-shipper";
-          Group = "haproxy-log-shipper";
-          SupplementaryGroups = [ "systemd-journal" ];
-          ExecStart = "${haproxy_log_shipper}/bin/haproxy-log-shipper";
-          Environment = [
-            "HAPROXY_INGEST_ENDPOINT=${endpoint}"
-            "RUST_LOG=info"
-          ];
-          EnvironmentFile = environment_file;
-
-          CapabilityBoundingSet = [ "" ];
-          DeviceAllow = [ "" ];
-          LockPersonality = true;
-          MemoryDenyWriteExecute = true;
-          NoNewPrivileges = true;
-          PrivateDevices = true;
-          PrivateTmp = true;
-          ProtectClock = true;
-          ProtectControlGroups = true;
-          ProtectHome = true;
-          ProtectHostname = true;
-          ProtectKernelLogs = true;
-          ProtectKernelModules = true;
-          ProtectKernelTunables = true;
-          ProtectSystem = "strict";
-          RemoveIPC = true;
-          RestrictAddressFamilies = [
-            "AF_INET"
-            "AF_UNIX"
-          ];
-          RestrictNamespaces = true;
-          RestrictRealtime = true;
-          RestrictSUIDSGID = true;
-          SystemCallArchitectures = "native";
-          SystemCallFilter = [
-            "@system-service"
-            "~@privileged"
-          ];
-          UMask = "0077";
-        };
-      };
-    };
-
-    timers.haproxy-log-shipper = {
-      description = "Run the HAProxy log shipper after each completed catch-up cycle";
-      wantedBy = [ "timers.target" ];
-      timerConfig = {
-        OnBootSec = "30s";
-        OnUnitInactiveSec = "60s";
-        Unit = "haproxy-log-shipper.service";
-      };
-    };
-
-    tmpfiles.rules = [
-      "d /var/lib/haproxy-log-shipper 0755 root root - -"
-    ];
-  };
-}

systems/portal-1/install.md

@@ -1,29 +0,0 @@
-# install
-
-## preinstall steps
-
-remove old tailscale entry
-
-## commands
-
-```sh
-scp root@149.28.55.47:/etc/ssh/ssh_host_ed25519_key.pub ./temp.pub
-nix-shell -p ssh-to-age --run 'cat temp.pub | ssh-to-age'
-rm temp.pub
-nix-shell -p sops --run "sops updatekeys users/secrets.yaml"
-nix run github:nix-community/nixos-anywhere -- --ssh-port 22 --copy-host-keys --build-on local --flake .#portal-1 root@149.28.55.47
-ssh richie@149.28.55.47 -p 278
-kitty +kitten ssh richie@149.28.55.47 -p 278
-```
-
-## Small VM
-
-This was required for a 2gb vm
-
-```sh
-sudo tee /etc/default/grub.d/99-nixos-anywhere-disable-kho.cfg >/dev/null <<'EOF'
-GRUB_CMDLINE_LINUX_DEFAULT="${GRUB_CMDLINE_LINUX_DEFAULT} kho=off"
-EOF
-
-sudo update-grub && sudo reboot
-```

systems/portal-1/secrets.yaml

@@ -1,25 +0,0 @@
-system_token: ENC[AES256_GCM,data:Yc6m0q2xX3n+dcMZ7NMn4c4UPBmGK0EjL6E4fMdxnNHcELpc+BEywCLOECJQUNw1jgMb35P6uuyEXrT+CsW9WKzbXE07T9clWo4sMkaxLugb/AklXCvMt7Ci0ObIaUZuoeKOs685rMUZLS6/er8dS0kWoBvuWfCq6pzfmC61wOQ=,iv:YfdJ1bvLsqzFloDyHG6+Epcf6jRMv6FAJSZRSHfAG6k=,tag:BlGBjxCXGZUd/+XKQ1mAkA==,type:str]
-sops:
-    age:
-        - enc: |
-            -----BEGIN AGE ENCRYPTED FILE-----
-            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBYaXFMVDBDV1lCYThDY1dh
-            U1VWbDFvd2huc04yQXF3RDl5TkE4UTR2RjFJCkJDbkdoWFJDVVlubWZTQVpGK09W
-            WkRJL0F3cjdNUmVQSlZiSE5QUmMzRk0KLS0tIFBST1JxdHpRSjlTU3c2UGNUNzJO
-            YlpJbENvV21GeEJtSm1PMVExK0JlV28KGMo+3gMjtR7722AWZpT8YuEJT0cxdQXR
-            Jf9ZCzmY54mBolyUeh6iBfNqdP6Ao3TeOGQcKFO1d0mLAWfAqdaCVw==
-            -----END AGE ENCRYPTED FILE-----
-          recipient: age1u8zj599elqqvcmhxn8zuwrufsz8w8w366d3ayrljjejljt2q45kq8mxw9c
-        - enc: |
-            -----BEGIN AGE ENCRYPTED FILE-----
-            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBaOTlxZjdHdnZwVGlRY2FW
-            WlU5R09LaW1rbzNmRUFWSkhtcHoyQWpJdGh3CkdFTUNBZFZJY0FjTVcxV0QvYWc2
-            SjI5Mkwwa3lTQTRTOWtXNXpaWHRITEEKLS0tIDlOS0tKNVRZOXdKRTNKa1pOZ2d3
-            MlEyZTZ0MXNkWXlSN2JlUHpKanpETkkKzWULPeeI6VH1GjKm/V/kEAc7DL5TzDbM
-            M700P1MqiCAyki6gO5BhmjdPspQFrc24z+hXL1WXJgKny7EmT9hUhw==
-            -----END AGE ENCRYPTED FILE-----
-          recipient: age1cm7ujz4t9cyc3nyy0w30l2addggk0gy8h3shd9yq47jg845tndnsa3utfl
-    lastmodified: "2026-10-14T02:09:14Z"
-    mac: ENC[AES256_GCM,data:s6KwNHefexYx1kFBOIg1fA8gMFkZFPDAS2ElvjEMM1Yfv6Sf8E3+opq5TWxAzm5i9+wR7/0QjmGVxHmtHqUxhO+TWke4te4HPf3etgGFHOZtVApc3DBGywO+g78XbyjbvoOALQmCTNrm7hHkhPbHFJWap/4DhChK4GEDUJHi8Zs=,iv:eCNA1ZWoLHWVeQi2o3maILQgoc18OrGnOHJDvPHUrfA=,tag:1YNztKsLAFtn5pkGMLueHA==,type:str]
-    unencrypted_suffix: _unencrypted
-    version: 3.13.3

systems/rhapsody-in-green/default.nix

@@ -3,15 +3,17 @@
   imports = [
     "${inputs.self}/users/richie"
     "${inputs.self}/common/global"
-    "${inputs.self}/common/optional/desktop"
+    "${inputs.self}/common/optional/desktop.nix"
     "${inputs.self}/common/optional/docker.nix"
     "${inputs.self}/common/optional/steam.nix"
     "${inputs.self}/common/optional/syncthing_base.nix"
     "${inputs.self}/common/optional/systemd-boot.nix"
     "${inputs.self}/common/optional/tailscale.nix"
+    "${inputs.self}/common/optional/yubikey.nix"
     "${inputs.self}/common/optional/zfs"
     ./hardware.nix
     ./programs.nix
+    ./qmk.nix
     ./syncthing.nix
     inputs.nixos-hardware.nixosModules.framework-13-7040-amd
   ];

systems/rhapsody-in-green/secrets.yaml

@@ -1,25 +0,0 @@
-system_token: ENC[AES256_GCM,data:785g7cl+2IOLFnx0b4m+Eh2OyjPoKMDOHc4L9bI7t1JdRZgyeeGPuEvwI8Pd8NCpZV7vlibxT3RE5+7fjmUakRn6lTbTQ/U1rKIvKxPSS5ocPC6WXZLx4QRBvKzD6I8FFO45CkZ7GeVJ69j3wXWkyteHcCF0R+8r/VAAPTfBZG4=,iv:iO8oCSPEKMvLMRMNrGfgpwCBuMKx/cffksEGjoAlh4g=,tag:y8j4MPK0tjVcUkH0xJxEzQ==,type:str]
-sops:
-    age:
-        - enc: |
-            -----BEGIN AGE ENCRYPTED FILE-----
-            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBIMmdsaWhPeHBTUWRqVWMz
-            aVB4RmRJRWFqSTdIYTMxc0hpeFV6UlFvM2dNCjdkTEhXeVFhcERxd2x0OUZTZTQz
-            QitzMjJ1UHFPMmlqUGdNd2FYRWEvWFUKLS0tIDVicThCam9hM0tSOE5hc1JlVEtZ
-            Wis4TTlqV1hiWXpuNjNnQ0w5aWVOOEUKzYOt27fcJ1OcdJNGMHnbHfhNd5KhCWsm
-            Bc3o9js8QxZYHGgiyq9RFenwpT1HSQp7Oj4clwLv9JtCCrTXpJXq5g==
-            -----END AGE ENCRYPTED FILE-----
-          recipient: age1u8zj599elqqvcmhxn8zuwrufsz8w8w366d3ayrljjejljt2q45kq8mxw9c
-        - enc: |
-            -----BEGIN AGE ENCRYPTED FILE-----
-            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBaTTI5aTZQaTlqVzBWNjc1
-            dmFXV3VyanFvVjh3cmpDVEFrTFUvNmxPcXgwCmhtSmE3bHJLWTNHNVQyQlBBUTVG
-            TzdveVNYSWFMZ3k2NjdVQ1VqcjFCajQKLS0tIG11VnNlVElUN3RjNmtsZjRIL1h5
-            MGgwbHZnWWVkSEFyYWd6ajgyWVVVM1EKd0SlmGctAWcxYn03cqM752vzmZdeg63O
-            N8kb1I+QaeTeUrYrCCfTl2iEspyJx1G9QPEfdvbHYKpgNfdJBDUyHQ==
-            -----END AGE ENCRYPTED FILE-----
-          recipient: age1ufnewppysaq2wwcl4ugngjz8pfzc5a35yg7luq0qmuqvctajcycs5lf6k4
-    lastmodified: "2026-10-14T02:09:45Z"
-    mac: ENC[AES256_GCM,data:NHkNpGH1b3srra8cc4WAq+npFIMuXi+2JNMquYNVoOS9VTyDXfOxSeMVAHjpWErRnDo+wPU26RHQXapsNa+MdrVYzCbhSib5tRNZCMysnl7l8nTGpeNAITCF+45++AaWhC9iljVkB8zQaFFZak8KHH4EXih8yHuojadVMwZZuEs=,iv:wG+H+s+vbN+l4ojrw/+aU6o9tS6UMIE674OHOI2e5eM=,tag:XRUi0fqlCWMaJmiLFheKDA==,type:str]
-    unencrypted_suffix: _unencrypted
-    version: 3.13.3

tests/test_snapshot_manager.py

@@ -0,0 +1,204 @@
+"""test_snapshot_manager."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+import pytest
+
+from python.tools.snapshot_manager import get_snapshots_to_delete, get_time_stamp, load_config_data, main
+from python.zfs.dataset import Dataset, Snapshot
+
+if TYPE_CHECKING:
+    from pyfakefs.fake_filesystem import FakeFilesystem
+    from pytest_mock import MockerFixture
+
+SNAPSHOT_MANAGER = "python.tools.snapshot_manager"
+FROM_PHONE = "1234567890"
+TO_PHONE = "0987654321"
+
+
+def patch_phone_numbers(mocker: MockerFixture) -> None:
+    """Patch the Signal phone number environment variables."""
+    phone_numbers = {
+        "SIGNAL_ALERT_FROM_PHONE": FROM_PHONE,
+        "SIGNAL_ALERT_TO_PHONE": TO_PHONE,
+    }
+    mocker.patch(f"{SNAPSHOT_MANAGER}.getenv", side_effect=phone_numbers.get)
+
+
+def patch_utcnow(mocker: MockerFixture, datetime_value: datetime) -> None:
+    """patch_utcnow."""
+    mocker.patch("python.tools.snapshot_manager.utcnow", return_value=datetime_value)
+
+
+def create_mock_snapshot(mocker: MockerFixture, name: str) -> Snapshot:
+    """create_mock_snapshot."""
+    mock_snapshot = mocker.MagicMock(spec=Snapshot)
+    mock_snapshot.name = name
+
+    return mock_snapshot
+
+
+def test_main(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    """Test main."""
+    load_config_data.cache_clear()
+    patch_phone_numbers(mocker)
+
+    mocker.patch(f"{SNAPSHOT_MANAGER}.get_time_stamp", return_value="2023-01-01T00:00:00")
+
+    mock_dataset = mocker.MagicMock(spec=Dataset)
+    mock_dataset.name = "test_dataset"
+    mock_dataset.create_snapshot.return_value = "snapshot created"
+    mock_get_datasets = mocker.patch(f"{SNAPSHOT_MANAGER}.get_datasets", return_value=(mock_dataset,))
+
+    mock_get_snapshots_to_delete = mocker.patch(f"{SNAPSHOT_MANAGER}.get_snapshots_to_delete")
+    mock_signal_alert = mocker.patch(f"{SNAPSHOT_MANAGER}.signal_alert")
+    mock_snapshot_config_toml = '["default"]\n15_min = 8\nhourly = 24\ndaily = 0\nmonthly = 0\n'
+    fs.create_file("/mock_snapshot_config.toml", contents=mock_snapshot_config_toml)
+    main(Path("/mock_snapshot_config.toml"))
+
+    mock_signal_alert.assert_not_called()
+    mock_get_datasets.assert_called_once()
+    mock_get_snapshots_to_delete.assert_called_once_with(
+        mock_dataset,
+        {
+            "15_min": 8,
+            "hourly": 24,
+            "daily": 0,
+            "monthly": 0,
+        },
+        FROM_PHONE,
+        TO_PHONE,
+    )
+
+
+def test_main_create_snapshot_failure(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    """Test main."""
+    load_config_data.cache_clear()
+    patch_phone_numbers(mocker)
+
+    mocker.patch(f"{SNAPSHOT_MANAGER}.get_time_stamp", return_value="2023-01-01T00:00:00")
+
+    mock_dataset = mocker.MagicMock(spec=Dataset)
+    mock_dataset.name = "test_dataset"
+    mock_dataset.create_snapshot.return_value = "snapshot not created"
+    mock_get_datasets = mocker.patch(f"{SNAPSHOT_MANAGER}.get_datasets", return_value=(mock_dataset,))
+
+    mock_get_snapshots_to_delete = mocker.patch(f"{SNAPSHOT_MANAGER}.get_snapshots_to_delete")
+    mock_signal_alert = mocker.patch(f"{SNAPSHOT_MANAGER}.signal_alert")
+    mock_snapshot_config_toml = '["default"]\n15_min = 8\nhourly = 24\ndaily = 0\nmonthly = 0\n'
+    fs.create_file("/mock_snapshot_config.toml", contents=mock_snapshot_config_toml)
+    main(Path("/mock_snapshot_config.toml"))
+
+    mock_signal_alert.assert_called_once_with(
+        "test_dataset failed to create snapshot 2023-01-01T00:00:00",
+        FROM_PHONE,
+        TO_PHONE,
+    )
+    mock_get_datasets.assert_called_once()
+    mock_get_snapshots_to_delete.assert_not_called()
+
+
+def test_main_exception(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    """Test main."""
+    load_config_data.cache_clear()
+    patch_phone_numbers(mocker)
+
+    mocker.patch(f"{SNAPSHOT_MANAGER}.get_time_stamp", return_value="2023-01-01T00:00:00")
+
+    mock_dataset = mocker.MagicMock(spec=Dataset)
+    mock_dataset.name = "test_dataset"
+    mock_dataset.create_snapshot.return_value = "snapshot created"
+    mock_get_datasets = mocker.patch(f"{SNAPSHOT_MANAGER}.get_datasets", side_effect=Exception("test"))
+
+    mock_get_snapshots_to_delete = mocker.patch(f"{SNAPSHOT_MANAGER}.get_snapshots_to_delete")
+    mock_signal_alert = mocker.patch(f"{SNAPSHOT_MANAGER}.signal_alert")
+    mock_snapshot_config_toml = '["default"]\n15_min = 8\nhourly = 24\ndaily = 0\nmonthly = 0\n'
+    fs.create_file("/mock_snapshot_config.toml", contents=mock_snapshot_config_toml)
+    with pytest.raises(SystemExit) as pytest_wrapped_e:
+        main(Path("/mock_snapshot_config.toml"))
+
+    assert isinstance(pytest_wrapped_e.value, SystemExit)
+    assert pytest_wrapped_e.value.code == 1
+    mock_signal_alert.assert_called_once_with("snapshot_manager failed", FROM_PHONE, TO_PHONE)
+    mock_get_datasets.assert_called_once()
+    mock_get_snapshots_to_delete.assert_not_called()
+
+
+def test_get_snapshots_to_delete(mocker: MockerFixture) -> None:
+    """test_get_snapshots_to_delete."""
+    mock_snapshot_0 = create_mock_snapshot(mocker, "auto_202509150415")
+    mock_snapshot_1 = create_mock_snapshot(mocker, "auto_202509150415")
+
+    mock_dataset = mocker.MagicMock(spec=Dataset)
+    mock_dataset.name = "test_dataset"
+    mock_dataset.get_snapshots.return_value = (mock_snapshot_0, mock_snapshot_1)
+    mock_dataset.delete_snapshot.return_value = None
+
+    mock_signal_alert = mocker.patch(f"{SNAPSHOT_MANAGER}.signal_alert")
+
+    get_snapshots_to_delete(
+        mock_dataset,
+        {"15_min": 1, "hourly": 0, "daily": 0, "monthly": 0},
+        FROM_PHONE,
+        TO_PHONE,
+    )
+
+    mock_signal_alert.assert_not_called()
+    mock_dataset.delete_snapshot.assert_called_once_with("auto_202509150415")
+
+
+def test_get_snapshots_to_delete_no_snapshot(mocker: MockerFixture) -> None:
+    """test_get_snapshots_to_delete_no_snapshot."""
+    mock_dataset = mocker.MagicMock(spec=Dataset)
+    mock_dataset.name = "test_dataset"
+    mock_dataset.get_snapshots.return_value = ()
+    mock_dataset.delete_snapshot.return_value = None
+
+    mock_signal_alert = mocker.patch(f"{SNAPSHOT_MANAGER}.signal_alert")
+
+    get_snapshots_to_delete(
+        mock_dataset,
+        {"15_min": 1, "hourly": 0, "daily": 0, "monthly": 0},
+        FROM_PHONE,
+        TO_PHONE,
+    )
+
+    mock_signal_alert.assert_not_called()
+    mock_dataset.delete_snapshot.assert_not_called()
+
+
+def test_get_snapshots_to_delete_errored(mocker: MockerFixture) -> None:
+    """test_get_snapshots_to_delete_errored."""
+    mock_snapshot_0 = create_mock_snapshot(mocker, "auto_202509150415")
+    mock_snapshot_1 = create_mock_snapshot(mocker, "auto_202509150415")
+
+    mock_dataset = mocker.MagicMock(spec=Dataset)
+    mock_dataset.name = "test_dataset"
+    mock_dataset.get_snapshots.return_value = (mock_snapshot_0, mock_snapshot_1)
+    mock_dataset.delete_snapshot.return_value = "snapshot has dependent clones"
+
+    mock_signal_alert = mocker.patch(f"{SNAPSHOT_MANAGER}.signal_alert")
+
+    get_snapshots_to_delete(
+        mock_dataset,
+        {"15_min": 1, "hourly": 0, "daily": 0, "monthly": 0},
+        FROM_PHONE,
+        TO_PHONE,
+    )
+
+    mock_signal_alert.assert_called_once_with(
+        "test_dataset@auto_202509150415 failed to delete: snapshot has dependent clones",
+        FROM_PHONE,
+        TO_PHONE,
+    )
+    mock_dataset.delete_snapshot.assert_called_once_with("auto_202509150415")
+
+
+def test_get_time_stamp(mocker: MockerFixture) -> None:
+    """Test get_time_stamp."""
+    patch_utcnow(mocker, datetime(2023, 1, 1, 0, 0, 0, tzinfo=UTC))
+    assert get_time_stamp() == "auto_202301010000"

treefmt.toml

@@ -13,11 +13,6 @@ command = "nixfmt"
 # Glob pattern of files to include
 includes = ["*.nix"]
 
-[formatter.rustfmt]
-command = "rustfmt"
-options = ["--edition", "2024"]
-includes = ["*.rs"]
-
 [formatter.ruff-format]
 command = "ruff"
 options = ["format"]

users/math/default.nix

@@ -36,6 +36,8 @@ in
         "hass"
         "libvirtd"
         "networkmanager"
+        "nornsight"
+        "nornsight-admin"
         "plugdev"
         "scanner"
         "transmission"

users/richie/default.nix

@@ -36,6 +36,8 @@ in
         "hass"
         "libvirtd"
         "networkmanager"
+        "nornsight"
+        "nornsight-admin"
         "ollama"
         "plugdev"
         "scanner"

users/richie/home/gui/default.nix

@@ -21,7 +21,6 @@
     vlc
     # browser
     brave
-    firefox
     # dev tools
     gparted
     jetbrains.datagrip

users/richie/home/gui/llm_tools.nix

@@ -1,7 +1,9 @@
 { pkgs, ... }:
 {
   home.packages = [
+    pkgs.master.claude-code
     pkgs.master.codex
     pkgs.master.opencode
+    pkgs.master.pi-coding-agent
   ];
 }

users/richie/home/programs.nix

@@ -42,9 +42,6 @@
     rustfmt
     trunk
     wasm-pack
-    # protobuf
-    protobuf
-    buf
     # nix
     nix-init
     nix-output-monitor

users/secrets.yaml

@@ -6,58 +6,58 @@ sops:
     age:
         - enc: |
             -----BEGIN AGE ENCRYPTED FILE-----
-            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSArMVJ6Rm1QQlAvR2FKRmFu
-            dlBianhsU051dnh4Y3Zqdkt2RHVQTy9KVVZFClVCMXZmR3pSR0hxNzB2RnFuS2Fs
-            OGZCQjdJTmhTa1NDZ1B4L2tJSnFmdmcKLS0tIEhTQTRNZUdlNy9XcC9EbnVrcmZQ
-            Tzh5SUt0K3hUVll2dDZNR1ZnbzVPRFUKxTLOmsr5zMKCuH7KDqs1itkPLn+2kzSh
-            0s5C5XOIsMwUrkA/uIFrOWaP57YBVWPM2Ry+GAp4Bbf+y8C1en5HUw==
+            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBEMjQvSzZSY08vRlI0dzBP
+            YytpVlpPSGFHaTI5b2xzUkFFdEY0bVhQdGgwClM3UmM1OElpcEZTcTB0NHpSZ2d6
+            aVBpcjVIaDk3Tko1cm4xYjdkYVQ1L1EKLS0tIE1aQW95U0JIVGc4MGloSXRXc3ZS
+            a3EvU1dJdmxOeVBIS28yT3dHd2ZEYWcKv5TIqRRQs3AXER1WHEx8wotyXr2Ktxsu
+            dwldwEHyDNqwf7UgGOZu7JueeZ7y2KC8KZ7ixIdrhVaoxuLEvCg7VA==
             -----END AGE ENCRYPTED FILE-----
           recipient: age1u8zj599elqqvcmhxn8zuwrufsz8w8w366d3ayrljjejljt2q45kq8mxw9c
         - enc: |
             -----BEGIN AGE ENCRYPTED FILE-----
-            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBmaTd1dU0zUGZVSkJJOUQw
-            OThIMDVSM1FKdVptaDNsMlFOYThpVmFnN1JvCmpqZlBrY2hVQ2grWkJOZTJkay9T
-            Q0VBVzRVYk5vU1BGMFhOckl3TDJDNjQKLS0tIDVqQTlJaHBuY0tHLzEwKzh6QStZ
-            a0ZqZmdOTnNEdEp6Nk5sN0NyT05MWmMKgP43hrlXzBpp+hELBaOJdYdsU3AmDvPf
-            FOh89Fn9Zp+Lm1shQTF6GcACTAr04+7MPfLveyUFt4vbHxr9ISYaOg==
+            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSByaE5jb3p3VzZ1dE05Wmlx
+            c1ZoWGhTcTRHRE9oMWtkaCtBY1l1a2QyWWlvCk9MM2FmcEZubXR1REpOcHZ6aHVE
+            UEkzbllZdkxPYzlTMHNod3VWNnhoaTgKLS0tIGVCdEpqUzhFM0swRjVjb0Z6ckFN
+            UzlYb1U4UGJqWE9oczdlNVh1S1lqL3cKSKxaJFhun2ee45I1jC55YKNdfYB8MsS+
+            dgFOFCvn/4uEcUyKZbUf0STjbIZ5eMHl8NVRTupI4tftGiE559M5Wg==
             -----END AGE ENCRYPTED FILE-----
           recipient: age1q47vup0tjhulkg7d6xwmdsgrw64h4ax3la3evzqpxyy4adsmk9fs56qz3y
         - enc: |
             -----BEGIN AGE ENCRYPTED FILE-----
-            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSAvWUZRME43WmxLd1BHaHFm
-            dXBib3NLcnI2K3M0cnl3NHJxSXU4Q253cW1ZCmlva2RReGcyTFN0TURCbkJFcHRW
-            Skd6aVVhcFlOYXkvTVVBRzZaYUpSUHcKLS0tIHlxMHQ1d21TSDBHMmtraUM3S0F2
-            VUUwOFJtSkM3V2tESnExR1hUSCt0TjgKOnhAVyikMjsc/lTKJrkbhQCw/EgiKuUg
-            DmzoBvVYmBfAJR74VVOdex2t0Ieg9SuHtCO7dQ7h82woY9Z8UudWDQ==
+            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSA2U01jTVZQT1RRRU1nQmVv
+            Z0lnOGJLN1c0UkNLbnJjV29na3Rub0xzbXhNCjY0aFNuUnFDUnZheDRHT3R5L3Fv
+            ZkJ3QkF2RENKRVdUYjRzRXMrNklvN2cKLS0tIFJzK2o1MmVTWHZMWmVDWlgzeGY5
+            VDg4OGl0NGFYY2VNZGY5UkR1cVdWY0UK1qa3swdfgUw1tnnThRf1sl9f2zTP/d/d
+            iwcenOEryXatVMJjZVWrLC5C7MwtKz+pIBXa98yaSpCDLqjAMwOCPg==
             -----END AGE ENCRYPTED FILE-----
           recipient: age1jhf7vm0005j60mjq63696frrmjhpy8kpc2d66mw044lqap5mjv4snmwvwm
         - enc: |
             -----BEGIN AGE ENCRYPTED FILE-----
-            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBMazg0K1o0WkU0ZHRZdWpV
-            UGxMV0cwaW9ZOG5PWVI4ZXZrSldIc2xuMDNzCmlhYjUwMXF5U1g0ODlMZmVVVXdj
-            K3hOM1Arc3VNTHNKMkozRXpYSG5RQWcKLS0tIEhrZFF6K1ZpYjNiMEFyYUpGejNM
-            Q3Q2cFhaVHhoOTZzM3pKeDBFam9PUVEK4XPeeoMYFWCr/LBTA+ZrnMqz+mE3IaKo
-            OQNVGW84fk7SZ4lWWU8l6jZmmX1Sf9aqyFad1hoI9N3AEZmhiiyhlA==
+            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBTREI2UEZwS0Y0SmEvK1F4
+            Y1hZZjRVWWxMT29mc3ZXcHFDclRUYi96LzA4CkIwZ2h4T2JQRW1OL2hmRTFHWkFY
+            RHZqQUFqU3hvNlRBZHRiSThPVUE3TG8KLS0tIHh0Z2tlNXREczJsa1g2ekZyZjBt
+            dW41aVA3VEVqa1NDMHM0T3FKZFNTN0kKGQPRUMSRtibFVFRZR/igsjgxgDo4ogga
+            dUHBRU65r41er+Wk3lKdtbXJX1g83K8ay7540hNXheOhxFi57trrzA==
             -----END AGE ENCRYPTED FILE-----
           recipient: age13lmqgc3jvkyah5e3vcwmj4s5wsc2akctcga0lpc0x8v8du3fxprqp4ldkv
         - enc: |
             -----BEGIN AGE ENCRYPTED FILE-----
-            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBoRHF5R2lWYWF1dUM3a3ZJ
-            RWZDVFBjRVhRaE5PRGtlbmROZm5IOWdNQkYwCldSWisvaVZDLzdwRDFCRkprM1Vo
-            NnFXUU1FWWIwZ3Nja3Mya2l3U1lROGcKLS0tIGRlZmhFQlBCZEJxYnRTUkcrSlVq
-            RC9jY0NINCtKTHRmMGRmd0N1WUVHWWcKrBjys6OKDuAELbpFpLOvV5GOXuVW9STX
-            ndHNw1rI14QJ7lZ54gAfdAtNFl1NzHDsaUuv/HIW9i2jhkv+APKtuA==
+            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBrV1Z2MGNPK28wTStRSUZW
+            dmFMaUt5MENHeUpVcDVxQnZhd1o5RmJwOWwwCittaGNrcW1hV25VaGJaTEhZRFBJ
+            Y2toWEpFWFZYT0plb2lkTlJCS2Mwb3MKLS0tIHlkVjVESWFsbWlVQ0xVQ0llclF2
+            L1FMbC91YjJsR3BGRjJ0QVQ1MDdBbEUKNnOI9+H1LJcvVcwl9uQMmFp35y5RB+Zl
+            Mo3z3unweQtS+jiUn7RyVs5daG1E2tfihu2CQDWaktuCuunMpt3Wew==
             -----END AGE ENCRYPTED FILE-----
           recipient: age1ufnewppysaq2wwcl4ugngjz8pfzc5a35yg7luq0qmuqvctajcycs5lf6k4
         - enc: |
             -----BEGIN AGE ENCRYPTED FILE-----
-            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSA5Ti9laG9aTlpSSFR5Vmk0
-            MEVFY2pJdDJaK0RZQ1dUd2dRMDBSOVNNWEhBCmdsSDl3a1RPTE9xSlRSZTBhNFVB
-            bzFoVlZzQjQ3QWJoZkwrUWFucTlRN1kKLS0tIEozRUwwTnlpMHg4eGdLa2h5U3hF
-            ekEwYTBKWmRiK29JMXVqbU1DWVVWNGcKY6kkh7poDQHDdZdFdqzEI12P9UMq2saQ
-            gQXljjqcamqZ96KwFmuZm9RB/6OJJAPY8MBSqlrPTSjcnAqikXWpEA==
+            YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSBqR0hUT3hxTzRwTVhuTm9E
+            TUFtNFFCYWNUNUNVbHNaNXQ0aEVBa04vSEU0CkRPVzhuYUsrM0wxZjRDdkxvTVA2
+            UW9CSjlIQW5yenhaUi9mbEdLYlhWYk0KLS0tIGRXMm9Wa3l4UWxqcGl6ZlBZV3E0
+            Z3YrbUYrSmV0NHNTeTdhMkJEdytwSmMKGLpe5+p4FGJNHy6SVgnsDbEsiP0J8lW/
+            0+3X3LQoL0f1S2L5DjexF/H0NGvWjAPaGzDNB6V50VxGDabEIVqjvQ==
             -----END AGE ENCRYPTED FILE-----
-          recipient: age1cm7ujz4t9cyc3nyy0w30l2addggk0gy8h3shd9yq47jg845tndnsa3utfl
+          recipient: age1vyav6kxtvt3z4vtnkkjj38eu8hlts5m7ygyckhskvalg2gpjk52su53d0a
     lastmodified: "2026-08-25T15:35:53Z"
     mac: ENC[AES256_GCM,data:4/WuSmUb2ONlB56dgbwwekZ8h7E1vvPzPxYCvNM8N68nt1nb9JxoMeAX9PktmljAarYLQJd1O5+49ZEph3Gzx9LrwvWnm9asq0UeeJi08x7wdQDcawNFlBFzXF2qlgxU1JssNHZZOB+5nYH4t8vFzPTq/9BqoifPR2+pLBYhUQI=,iv:N/gVgeWT2/1bfHuR6TanyDBijWmXAqyACXpnwfwDuVA=,tag:pQ4ct4G7SARz2wyjOc7AKQ==,type:str]
     unencrypted_suffix: _unencrypted