feat(haproxy): implement HAProxy log shipper with initial configuration and dependencies

- Introduced a new Rust-based service for ingesting HAProxy logs, including the necessary database schema and API endpoints.
- Implemented database interactions for managing source status, inserting events, and handling ingestion batches.
- Added validation for incoming log batches to ensure data integrity.
- Configured systemd service for the HAProxy ingestion server with appropriate security settings.
- Updated firewall rules to allow traffic on the ingestion server's port (8444).
- Created necessary PostgreSQL database and user configurations for the ingestion service.

Cargo.toml

@@ -5,14 +5,47 @@ members = ["rust/*"]
 
 [workspace.dependencies]
 anyhow = "1.0"
+axum = "0.8"
+bytes = "1.10"
 chrono = "0.4"
-clap = "4.5"
+clap = { version = "4.5", features = ["derive", "env"] }
 env_logger = "0.11"
+garde = { version = "0.23", features = ["derive", "serde"] }
+hex = "0.4"
 log = "0.4"
-serde = "1.0"
+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 }

rust/haproxy_ingest_proto/Cargo.toml

@@ -0,0 +1,17 @@
+[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

@@ -0,0 +1,96 @@
+# 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

@@ -0,0 +1,11 @@
+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

@@ -0,0 +1,71 @@
+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

@@ -0,0 +1,12 @@
+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

@@ -0,0 +1,29 @@
+[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

@@ -0,0 +1,102 @@
+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

@@ -0,0 +1,19 @@
+{
+  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

@@ -0,0 +1,376 @@
+//! `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

@@ -0,0 +1,530 @@
+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

@@ -0,0 +1,214 @@
+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

@@ -0,0 +1,28 @@
+[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

@@ -0,0 +1,24 @@
+{
+  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

@@ -0,0 +1,360 @@
+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

@@ -0,0 +1,271 @@
+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

@@ -0,0 +1,354 @@
+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/tls/Cargo.toml

@@ -0,0 +1,14 @@
+[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

@@ -0,0 +1,16 @@
+//! 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

@@ -12,9 +12,10 @@ clap = { workspace = true, features = ["derive", "env"] }
 env_logger.workspace = true
 log.workspace = true
 num-traits = "0.2"
-reqwest = { version = "0.12", default-features = false, features = ["blocking", "json", "rustls-tls"] }
+reqwest = { workspace = true, features = ["blocking", "json"] }
 serde = { workspace = true, features = ["derive"] }
 serde_json.workspace = true
+tls.workspace = true
 
 [lints]
 workspace = true

rust/van_weather/src/main.rs

@@ -13,6 +13,7 @@ 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";
@@ -159,6 +160,7 @@ 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)
@@ -494,6 +496,14 @@ fn percent(value: Option<f64>) -> i64 {
 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");

systems/jeeves/networking/firewall.nix

@@ -6,6 +6,7 @@ let
     gitea_http = 6443;
     gitea_ssh = 2223;
     grafana = 3000;
+    haproxy_ingest = 8444;
     jellyfin_http = 8096;
     jellyswarrm = 3100;
     nix_binary_cache = 5000;
@@ -26,6 +27,7 @@ in
     interfaces.tailscale0.allowedTCPPorts = [
       ports.audiobookshelf
       ports.gems
+      ports.haproxy_ingest
       ports.jellyfin_http
       ports.jellyswarrm
       ports.nix_binary_cache

systems/jeeves/services/haproxy_ingest.nix

@@ -0,0 +1,82 @@
+{
+  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/postgress.nix

@@ -36,6 +36,9 @@ 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
@@ -97,6 +100,13 @@ in
           replication = true;
         };
       }
+      {
+        name = "haproxy_ingest";
+        ensureDBOwnership = true;
+        ensureClauses = {
+          login = true;
+        };
+      }
       {
         name = "signalbot";
         ensureDBOwnership = true;
@@ -108,6 +118,7 @@ in
     ensureDatabases = [
       "data_science_dev"
       "gitea"
+      "haproxy_ingest"
       "math"
       "n8n"
       "richie"

systems/portal-1/haproxy/default.nix

@@ -1,6 +1,7 @@
 let
   domains = [
     "audiobookshelf"
+    "auth"
     "cache"
     "gitea"
     "gems"
@@ -19,6 +20,8 @@ let
   acmeServices = map (domain: "acme-${domain}.tmmworkshop.com.service") domains;
 in
 {
+  imports = [ ./ingest.nix ];
+
   networking.firewall.allowedTCPPorts = [
     80
     443

systems/portal-1/haproxy/ingest.nix

@@ -0,0 +1,103 @@
+{ 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 - -"
+    ];
+  };
+}