feat: deploy CI pipeline with ci-relay and local-runner

Add ci-relay (VPS webhook receiver) and local-runner (Thinkpad CI
executor) crates that communicate over iroh. Includes .ci.yml smoke
test pipeline, simulation tests, and development docs.

Authored by Claude, lovingly guided by Zachery Aaron Shores-Chmielewski
This commit is contained in:
Zachery Aaron Shores-Chmielewski 2026-02-16 01:26:20 +07:00
parent e9b93670eb
commit 0c6f9c9f39
19 changed files with 3739 additions and 0 deletions

8
.ci.yml Normal file
View file

@ -0,0 +1,8 @@
pipelines:
smoke:
triggers:
- event: push
branches: ["*"]
jobs:
hello:
run: echo "CI is alive"

50
Cargo.lock generated
View file

@ -397,6 +397,21 @@ version = "1.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6e4de3bc4ea267985becf712dc6d9eed8b04c953b3fcfb339ebc87acd9804901" checksum = "6e4de3bc4ea267985becf712dc6d9eed8b04c953b3fcfb339ebc87acd9804901"
[[package]]
name = "ci-relay"
version = "0.1.0"
dependencies = [
"clap",
"hex",
"hmac",
"iroh",
"serde_json",
"sha2 0.10.9",
"swactor-ci",
"tiny_http",
"tokio",
]
[[package]] [[package]]
name = "ciborium" name = "ciborium"
version = "0.2.2" version = "0.2.2"
@ -1063,6 +1078,7 @@ checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
dependencies = [ dependencies = [
"block-buffer 0.10.4", "block-buffer 0.10.4",
"crypto-common 0.1.7", "crypto-common 0.1.7",
"subtle",
] ]
[[package]] [[package]]
@ -1703,6 +1719,12 @@ version = "0.5.2"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c"
[[package]]
name = "hex"
version = "0.4.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70"
[[package]] [[package]]
name = "hickory-proto" name = "hickory-proto"
version = "0.25.2" version = "0.25.2"
@ -1756,6 +1778,15 @@ dependencies = [
"tracing", "tracing",
] ]
[[package]]
name = "hmac"
version = "0.12.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e"
dependencies = [
"digest 0.10.7",
]
[[package]] [[package]]
name = "http" name = "http"
version = "1.4.0" version = "1.4.0"
@ -2479,6 +2510,20 @@ version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "11d3d7f243d5c5a8b9bb5d6dd2b1602c0cb0b9db1621bafc7ed66e35ff9fe092" checksum = "11d3d7f243d5c5a8b9bb5d6dd2b1602c0cb0b9db1621bafc7ed66e35ff9fe092"
[[package]]
name = "local-runner"
version = "0.1.0"
dependencies = [
"clap",
"ctrlc",
"iroh",
"runtime-dashboard",
"serde_json",
"swactor",
"swactor-ci",
"tokio",
]
[[package]] [[package]]
name = "lock_api" name = "lock_api"
version = "0.4.14" version = "0.4.14"
@ -4408,10 +4453,15 @@ dependencies = [
name = "swactor-ci" name = "swactor-ci"
version = "0.1.0" version = "0.1.0"
dependencies = [ dependencies = [
"hex",
"hmac",
"serde", "serde",
"serde_json", "serde_json",
"serde_yaml", "serde_yaml",
"sha2 0.10.9",
"swactor", "swactor",
"tiny_http",
"ureq",
] ]
[[package]] [[package]]

View file

@ -13,6 +13,8 @@ members = [
"crates/crypto-wasm", "crates/crypto-wasm",
"tests/docker", "tests/docker",
"crates/ci", "crates/ci",
"crates/local-runner",
"crates/ci-relay",
"xtask", "xtask",
] ]
exclude = ["tools/depgraph"] exclude = ["tools/depgraph"]

357
HOW_TO_TEST_DRIVE.md Normal file
View file

@ -0,0 +1,357 @@
# How to Test Drive the Local CI Runner over LAN
This guide walks through running the local CI runner on the Thinkpad and triggering it from Forgejo on your main laptop. By the end, pushes to your Forgejo repos will automatically run CI pipelines on the Thinkpad.
## Network Layout
```
┌──────────────────────┐ LAN ┌──────────────────────┐
│ Main Laptop │◄───────────────────►│ Thinkpad │
│ │ │ │
│ Forgejo instance │ webhook POST ───► │ local-runner │
│ (e.g. :3000) │ ◄── status API ──── │ (e.g. :8787) │
│ │ │ dashboard (:9090) │
└──────────────────────┘ └──────────────────────┘
```
## Prerequisites
- **Main laptop**: Forgejo running and accessible on LAN (e.g. `http://192.168.1.100:3000`)
- **Thinkpad**: Rust toolchain installed, this repo cloned, network-reachable from the main laptop
- Both machines on the same LAN (or routable to each other)
Find each machine's LAN IP:
```bash
# On each machine
ip addr show | grep 'inet ' | grep -v 127.0.0.1
# or
hostname -I
```
We'll use these example IPs throughout the guide:
- Main laptop (Forgejo): `192.168.1.100`
- Thinkpad (CI runner): `192.168.1.200`
Replace them with your actual IPs.
---
## Step 1: Create a Forgejo API Token
On your main laptop, open Forgejo in a browser:
1. Go to **Settings > Applications** (top-right user menu > Settings > Applications)
2. Create a new token with at least these permissions:
- `repo`: read/write (needed to post commit statuses)
3. Copy the token (e.g. `abc123def456`)
---
## Step 2: Write a `.ci.yml` for Your Repository
Create a `.ci.yml` file on the Thinkpad. This defines what pipelines and jobs to run.
Example for a Rust project:
```yaml
pipelines:
check:
triggers:
- event: push
branches: ["*"]
exclude: ["master"]
jobs:
fmt:
run: cargo fmt -- --check
clippy:
run: cargo clippy -- -D warnings
test:
needs: [fmt, clippy]
run: cargo test
timeout: 300
release:
triggers:
- event: push
branches: ["master"]
jobs:
test:
run: cargo test --all-features
timeout: 600
bench:
needs: [test]
run: cargo bench
```
Key points:
- `triggers[].event`: one of `push`, `tag`, `merge`
- `triggers[].branches`: glob patterns (`"*"` matches all, `"feature-*"` matches prefixed)
- `triggers[].exclude`: branches to skip
- `jobs[].run`: a single command string, or a list of commands (run sequentially)
- `jobs[].needs`: list of jobs that must pass first (DAG dependencies)
- `jobs[].timeout`: max seconds before the job is killed (default: 300)
- `jobs[].env`: extra environment variables as key-value pairs
Save this file somewhere accessible on the Thinkpad, e.g. `/home/user/ci/my-project.ci.yml`.
---
## Step 3: Build the Local Runner on the Thinkpad
SSH into the Thinkpad (or work directly on it):
```bash
# Clone the repo if not already present
git clone <repo-url> ~/swactor
cd ~/swactor
# Build the local-runner binary
cargo build --release -p local-runner
```
The binary will be at `target/release/local-runner`.
---
## Step 4: Create a Working Directory
The runner clones your repo into a working directory for each pipeline. Create it:
```bash
mkdir -p ~/ci-work
```
---
## Step 5: Start the Local Runner
```bash
./target/release/local-runner \
--port 8787 \
--forgejo-url http://192.168.1.100:3000 \
--forgejo-token abc123def456 \
--secret my-webhook-secret \
--yaml /home/user/ci/my-project.ci.yml \
--work-dir /home/user/ci-work \
--repo-url http://192.168.1.100:3000/user/repo.git \
--dashboard-port 9090
```
| Flag | Description |
|------|-------------|
| `--port` | Port the webhook listener binds to (default: `8787`) |
| `--forgejo-url` | URL of your Forgejo instance on the main laptop |
| `--forgejo-token` | API token created in Step 1 |
| `--secret` | Webhook secret (must match what you set in Forgejo, see Step 6) |
| `--yaml` | Path to the `.ci.yml` file on disk |
| `--work-dir` | Base directory for git clones (one subdirectory per pipeline) |
| `--repo-url` | Git clone URL for the repository (HTTP or SSH) |
| `--dashboard-port` | Optional: enables the web dashboard on this port |
You should see:
```
Webhook listener on http://0.0.0.0:8787
Local CI runner started
Webhook: http://0.0.0.0:8787
YAML: /home/user/ci/my-project.ci.yml
Workdir: /home/user/ci-work
Dashboard: http://0.0.0.0:9090
```
The runner is now listening for webhooks.
---
## Step 6: Configure the Forgejo Webhook
On your main laptop, open Forgejo and go to the repo's settings:
1. Navigate to **Settings > Webhooks > Add Webhook > Forgejo**
2. Fill in:
- **Target URL**: `http://192.168.1.200:8787` (Thinkpad's LAN IP and runner port)
- **Secret**: `my-webhook-secret` (must match `--secret` from Step 5)
- **Trigger On**: Choose which events to send:
- **Push Events** (for `push` triggers)
- **Create Events** (for `tag` triggers)
- **Pull Request Events** (for `merge` triggers)
- **Branch filter**: leave blank to send all branches, or set a pattern
- **Active**: checked
3. Click **Add Webhook**
### Test the webhook connection
After adding the webhook, Forgejo shows a **Test Delivery** button. Click it to send a test ping. Check the Thinkpad terminal for output. Forgejo also shows the response status — you should see `200 OK`.
---
## Step 7: Push and Watch
From your main laptop (or anywhere with push access):
```bash
cd ~/my-project
git checkout -b test-ci
echo "// test" >> src/main.rs
git add src/main.rs
git commit -m "test CI"
git push origin test-ci
```
On the Thinkpad terminal, you'll see the runner:
1. Receive the webhook
2. Match triggers in `.ci.yml`
3. Queue the pipeline
4. Clone the repo and checkout the commit SHA
5. Execute jobs one at a time, respecting the DAG order
6. Stream stdout/stderr in real time
7. Report commit statuses back to Forgejo
Back in Forgejo, the commit will show status checks (pending, then success/failure) next to the SHA.
---
## Step 8: View the Dashboard (Optional)
If you started with `--dashboard-port 9090`, open a browser on any LAN machine:
```
http://192.168.1.200:9090
```
This shows the swactor runtime dashboard with CI-specific panels: active pipelines, recent pipelines, job statuses, and actor system metrics.
---
## Behavior Reference
### One-at-a-Time Execution
Jobs run serially — only one job executes at any moment across all pipelines. This guarantees benchmark isolation with no resource contention.
### Queue Supersede
If you push twice to the same branch quickly:
- **First push** is already running: it finishes normally
- **Second push** is queued: it runs after the first finishes
- **Third push** arrives while second is still queued: the second is **superseded** (marked as error, skipped), and the third takes its place in the queue
Only queued pipelines get superseded — a running pipeline always runs to completion.
### DAG Dependencies
Within a pipeline, jobs respect their `needs` dependencies. If `test` needs `[fmt, clippy]`, then `fmt` runs first, then `clippy`, then `test`. If `fmt` fails, `test` is skipped.
### CI Environment Variables
Every job command has these injected:
| Variable | Value |
|----------|-------|
| `CI` | `true` |
| `CI_COMMIT_SHA` | The commit being tested |
| `CI_BRANCH` | The branch name |
| `CI_PIPELINE_ID` | Numeric pipeline identifier |
| `CI_JOB_NAME` | Name of the current job |
Plus any `env` keys from the job definition in `.ci.yml`.
### Commit Status Reporting
The runner posts status updates to the Forgejo API for each pipeline and each job:
- `pending` when a pipeline/job is queued
- `success` when all jobs pass
- `failure` when a job fails
- `error` when a pipeline is superseded
These appear as commit status checks in Forgejo's UI.
---
## Troubleshooting
### Webhook not reaching the Thinkpad
- Verify the Thinkpad's firewall allows inbound on the webhook port:
```bash
# On Thinkpad
sudo ufw allow 8787/tcp # if using ufw
# or
sudo iptables -A INPUT -p tcp --dport 8787 -j ACCEPT
```
- Confirm connectivity from main laptop:
```bash
curl -v http://192.168.1.200:8787
# Should get "method not allowed" (405) — that means the server is reachable
```
### Signature mismatch (401)
- The `--secret` flag on the runner must exactly match the **Secret** field in Forgejo's webhook config
- If you don't want signature verification, set both to empty strings (omit `--secret` and leave Secret blank in Forgejo)
### Git clone fails
- Make sure `--repo-url` is reachable from the Thinkpad:
```bash
# On Thinkpad
git ls-remote http://192.168.1.100:3000/user/repo.git
```
- If the repo is private, use an authenticated URL:
```
http://user:password@192.168.1.100:3000/user/repo.git
```
Or use SSH:
```
git@192.168.1.100:user/repo.git
```
### Status updates not appearing in Forgejo
- Verify the API token has `repo` write permissions
- Check the Thinkpad terminal for `StatusReporter: failed to post status` errors
- Test the token manually:
```bash
curl -H "Authorization: token abc123def456" \
http://192.168.1.100:3000/api/v1/user
```
### Jobs failing unexpectedly
- Check that the Thinkpad has the necessary toolchain (cargo, rustup, etc.)
- The working directory for each pipeline is `{work-dir}/pipeline-{id}/` — you can inspect it
- Job stdout/stderr is streamed to the runner's terminal output
---
## Quick-Start Cheat Sheet
```bash
# === Thinkpad ===
cd ~/swactor
cargo build --release -p local-runner
mkdir -p ~/ci-work
./target/release/local-runner \
--port 8787 \
--forgejo-url http://LAPTOP_IP:3000 \
--forgejo-token YOUR_TOKEN \
--secret YOUR_SECRET \
--yaml /path/to/.ci.yml \
--work-dir ~/ci-work \
--repo-url http://LAPTOP_IP:3000/user/repo.git \
--dashboard-port 9090
# === Main Laptop (Forgejo) ===
# Repo > Settings > Webhooks > Add Webhook:
# URL: http://THINKPAD_IP:8787
# Secret: YOUR_SECRET
# Events: Push, Create, Pull Request
# === Test it ===
git push origin my-branch # triggers CI on the Thinkpad
```

View file

@ -0,0 +1,19 @@
[package]
name = "ci-relay"
version = "0.1.0"
edition = "2024"
[[bin]]
name = "ci-relay"
path = "src/main.rs"
[dependencies]
swactor-ci = { path = "../ci", features = ["local"] }
iroh = "0.96"
tokio = { version = "1", features = ["rt-multi-thread"] }
tiny_http = "0.12"
serde_json = "1"
hmac = "0.12"
sha2 = "0.10"
hex = "0.4"
clap = { version = "4", features = ["derive"] }

243
crates/ci-relay/src/main.rs Normal file
View file

@ -0,0 +1,243 @@
//! ci-relay — webhook relay for VPS side.
//!
//! Receives Forgejo webhook POSTs over HTTP, then forwards the parsed
//! `WebhookEvent` payloads to the Thinkpad local-runner over iroh.
use std::sync::Arc;
use std::time::Duration;
use clap::Parser;
use iroh::{Endpoint, RelayMode};
use tokio::sync::Mutex as TokioMutex;
use swactor_ci::webhook_server::parse_webhook_json;
use swactor_ci::{EventType, WebhookEvent};
/// ALPN protocol identifier for CI relay traffic over iroh.
const ALPN: &[u8] = b"swactor/ci/1";
/// Wire tag for WebhookEvent messages.
const WEBHOOK_TAG: &str = "ci::WebhookEvent";
#[derive(Parser)]
#[command(name = "ci-relay", about = "Webhook relay: Forgejo → iroh → local-runner")]
struct Args {
/// HTTP port for receiving Forgejo webhooks.
#[arg(long, default_value = "8787")]
port: u16,
/// Webhook secret for HMAC-SHA256 verification (empty to skip).
#[arg(long, default_value = "")]
secret: String,
}
fn main() {
let args = Args::parse();
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.expect("failed to build tokio runtime");
let endpoint = rt.block_on(async {
Endpoint::builder()
.alpns(vec![ALPN.to_vec()])
.relay_mode(RelayMode::Default)
.bind()
.await
.expect("failed to bind iroh endpoint")
});
let node_id = endpoint.id();
eprintln!("ci-relay started");
eprintln!(" Iroh Node ID: {node_id}");
eprintln!(" Webhook HTTP: http://0.0.0.0:{}", args.port);
eprintln!();
eprintln!("Waiting for runner to connect...");
// Shared state: the active connection from the Thinkpad runner.
let connection: Arc<TokioMutex<Option<iroh::endpoint::Connection>>> =
Arc::new(TokioMutex::new(None));
// Spawn a task that accepts inbound iroh connections from the runner.
{
let endpoint = endpoint.clone();
let connection = Arc::clone(&connection);
rt.spawn(async move {
loop {
match endpoint.accept().await {
Some(incoming) => match incoming.await {
Ok(conn) => {
let remote = conn.remote_id();
eprintln!("Runner connected: {remote}");
*connection.lock().await = Some(conn);
}
Err(e) => {
eprintln!("iroh accept error: {e}");
}
},
None => {
eprintln!("iroh endpoint closed");
break;
}
}
}
});
}
// Run the HTTP webhook listener on a standard thread (blocking).
let secret = args.secret.clone();
let server = tiny_http::Server::http(format!("0.0.0.0:{}", args.port))
.expect("failed to start HTTP server");
eprintln!("Listening for webhooks...");
for mut request in server.incoming_requests() {
let response = handle_webhook(&mut request, &secret, &connection, &rt);
let _ = request.respond(response);
}
}
/// Handle an incoming webhook HTTP request.
///
/// Parses and verifies the webhook, then forwards the event over iroh.
fn handle_webhook(
request: &mut tiny_http::Request,
secret: &str,
connection: &Arc<TokioMutex<Option<iroh::endpoint::Connection>>>,
rt: &tokio::runtime::Runtime,
) -> tiny_http::Response<std::io::Cursor<Vec<u8>>> {
use hmac::{Hmac, Mac};
use sha2::Sha256;
if request.method() != &tiny_http::Method::Post {
return tiny_http::Response::from_string("method not allowed").with_status_code(405);
}
// Read body.
let mut body = String::new();
if let Err(e) = std::io::Read::read_to_string(&mut request.as_reader(), &mut body) {
eprintln!("webhook: failed to read body: {e}");
return tiny_http::Response::from_string("bad request").with_status_code(400);
}
// Verify HMAC-SHA256 signature if secret is non-empty.
if !secret.is_empty() {
let sig_header = request
.headers()
.iter()
.find(|h| h.field.equiv("X-Forgejo-Signature"))
.map(|h| h.value.as_str().to_string());
match sig_header {
Some(sig_hex) => {
type HmacSha256 = Hmac<Sha256>;
let mut mac =
HmacSha256::new_from_slice(secret.as_bytes()).expect("HMAC key creation");
hmac::Mac::update(&mut mac, body.as_bytes());
let expected = hex::encode(mac.finalize().into_bytes());
if sig_hex != expected {
eprintln!("webhook: signature mismatch");
return tiny_http::Response::from_string("unauthorized").with_status_code(401);
}
}
None => {
eprintln!("webhook: missing signature header");
return tiny_http::Response::from_string("unauthorized").with_status_code(401);
}
}
}
// Determine event type from Forgejo header.
let event_header = request
.headers()
.iter()
.find(|h| h.field.equiv("X-Forgejo-Event"))
.map(|h| h.value.as_str().to_string())
.unwrap_or_default();
let event_type = match event_header.as_str() {
"push" => EventType::Push,
"create" => EventType::Tag,
"pull_request" => EventType::Merge,
other => {
eprintln!("webhook: ignoring event type '{other}'");
return tiny_http::Response::from_string("ignored").with_status_code(200);
}
};
// Parse JSON body.
let json: serde_json::Value = match serde_json::from_str(&body) {
Ok(v) => v,
Err(e) => {
eprintln!("webhook: failed to parse JSON: {e}");
return tiny_http::Response::from_string("bad json").with_status_code(400);
}
};
let webhook_event = match parse_webhook_json(&json, event_type) {
Some(e) => e,
None => {
eprintln!("webhook: could not extract fields from JSON");
return tiny_http::Response::from_string("bad payload").with_status_code(400);
}
};
eprintln!(
"webhook: {} {} on {}/{}",
webhook_event.commit_sha.get(..8).unwrap_or(&webhook_event.commit_sha),
webhook_event.branch,
webhook_event.repo_owner,
webhook_event.repo_name,
);
// Forward over iroh.
match forward_event(&webhook_event, connection, rt) {
Ok(()) => {
eprintln!(" → forwarded to runner");
tiny_http::Response::from_string("ok").with_status_code(200)
}
Err(e) => {
eprintln!(" → forward failed: {e}");
tiny_http::Response::from_string("relay error").with_status_code(502)
}
}
}
/// Serialize and send a WebhookEvent over the iroh connection.
fn forward_event(
event: &WebhookEvent,
connection: &Arc<TokioMutex<Option<iroh::endpoint::Connection>>>,
rt: &tokio::runtime::Runtime,
) -> Result<(), Box<dyn std::error::Error>> {
let payload = serde_json::to_vec(event)?;
rt.block_on(async {
let guard = connection.lock().await;
let conn = guard.as_ref().ok_or("no runner connected")?;
let mut send = conn.open_uni().await?;
write_tagged_message(&mut send, WEBHOOK_TAG.as_bytes(), &payload).await?;
send.finish()?;
// Wait briefly for the stream to flush.
tokio::time::sleep(Duration::from_millis(50)).await;
Ok(())
})
}
/// Write a tagged message to a QUIC send stream.
///
/// Frame format: `[4B tag_len][tag_bytes][payload_bytes]`
async fn write_tagged_message(
send: &mut iroh::endpoint::SendStream,
tag: &[u8],
payload: &[u8],
) -> Result<(), Box<dyn std::error::Error>> {
let tag_len = (tag.len() as u32).to_be_bytes();
send.write_all(&tag_len).await?;
send.write_all(tag).await?;
send.write_all(payload).await?;
Ok(())
}

View file

@ -9,4 +9,21 @@ serde = { version = "1", features = ["derive"] }
serde_yaml = "0.9" serde_yaml = "0.9"
serde_json = "1" serde_json = "1"
# Local runner library dependencies
tiny_http = { version = "0.12", optional = true }
ureq = { version = "2", optional = true }
hmac = { version = "0.12", optional = true }
sha2 = { version = "0.10", optional = true }
hex = { version = "0.4", optional = true }
[features]
default = []
local = [
"dep:tiny_http",
"dep:ureq",
"dep:hmac",
"dep:sha2",
"dep:hex",
]
[dev-dependencies] [dev-dependencies]

View file

@ -1,7 +1,11 @@
pub mod coordinator; pub mod coordinator;
pub mod local_coordinator;
pub mod local_runner;
pub mod pipeline; pub mod pipeline;
pub mod provisioner; pub mod provisioner;
pub mod runner; pub mod runner;
pub mod status_reporter;
pub mod webhook_server;
pub mod yaml; pub mod yaml;
use std::collections::HashMap; use std::collections::HashMap;
@ -200,6 +204,7 @@ pub struct JobSuccess;
pub enum JobFailure { pub enum JobFailure {
CommandFailed { exit_code: i32, last_lines: Vec<String> }, CommandFailed { exit_code: i32, last_lines: Vec<String> },
SshError(String), SshError(String),
ExecError(String),
Timeout, Timeout,
Interrupted, Interrupted,
} }
@ -211,6 +216,7 @@ impl fmt::Display for JobFailure {
write!(f, "command exited with code {exit_code}") write!(f, "command exited with code {exit_code}")
} }
JobFailure::SshError(msg) => write!(f, "SSH error: {msg}"), JobFailure::SshError(msg) => write!(f, "SSH error: {msg}"),
JobFailure::ExecError(msg) => write!(f, "exec error: {msg}"),
JobFailure::Timeout => write!(f, "job timed out"), JobFailure::Timeout => write!(f, "job timed out"),
JobFailure::Interrupted => write!(f, "spot instance interrupted"), JobFailure::Interrupted => write!(f, "spot instance interrupted"),
} }
@ -286,3 +292,23 @@ impl Default for CiConfig {
} }
} }
} }
// ─── Local CI Types ──────────────────────────────────────────────────────────
/// Job execution request for local runner (no InstanceReady needed).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LocalStartJob {
pub job_id: JobId,
pub work_dir: String,
pub job_def: JobDefinition,
pub env_overrides: HashMap<String, String>,
}
/// Configuration for the local CI runner.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LocalCiConfig {
pub ci: CiConfig,
pub repo_url: String,
pub work_dir: String,
pub ci_yaml_path: String,
}

View file

@ -0,0 +1,485 @@
//! LocalCoordinator actor: single-machine CI brain.
//!
//! Receives webhook events, queues pipelines, executes jobs one at a time
//! directly on the host. Supports branch-level supersede for queued pipelines.
use std::collections::{HashMap, VecDeque};
use std::process::Command;
use std::sync::{Arc, Mutex};
use swactor::actor::{ActorAddress, ActorInterface, Ctx};
use crate::pipeline::PipelineExecution;
use crate::status_reporter::StatusReporterMsg;
use crate::yaml::{self, CiYaml};
use crate::{
JobComplete, JobId, JobProgress, JobStatus, LocalCiConfig, LocalStartJob, PipelineId,
PipelineStatus, StatusUpdate, WebhookEvent,
};
/// Messages the LocalCoordinator can receive.
#[derive(Debug, Clone)]
pub enum LocalCoordinatorMsg {
Webhook(WebhookEvent),
SetCiYaml(CiYaml),
JobProgress(JobProgress),
JobComplete(JobComplete),
GitReady {
pipeline_id: PipelineId,
job_name: String,
work_dir: String,
},
}
/// Lightweight snapshot of coordinator state (no dashboard dependency).
/// The binary layer converts this to `CiSnapshot` for the dashboard.
#[derive(Debug, Clone, Default)]
pub struct LocalCiSnapshot {
pub active_pipelines: Vec<PipelineExecution>,
pub recent_pipelines: Vec<PipelineExecution>,
pub has_running_job: bool,
}
/// The LocalCoordinator actor state.
pub struct LocalCoordinator {
config: LocalCiConfig,
ci_yaml: Option<CiYaml>,
pipelines: HashMap<PipelineId, PipelineExecution>,
next_pipeline_id: u64,
/// FIFO queue of pipeline IDs awaiting execution.
queue: VecDeque<PipelineId>,
/// The pipeline currently being executed.
active_pipeline: Option<PipelineId>,
/// At most one running job (job_id, runner actor address).
running_job: Option<(JobId, ActorAddress)>,
/// Status reporter actor address.
status_reporter_addr: Option<ActorAddress>,
/// Bounded ring of finished pipelines.
completed: VecDeque<PipelineExecution>,
/// Shared snapshot for external consumers (e.g. dashboard binary).
ci_snapshot: Arc<Mutex<LocalCiSnapshot>>,
}
impl LocalCoordinator {
pub fn new(config: LocalCiConfig) -> Self {
Self {
config,
ci_yaml: None,
pipelines: HashMap::new(),
next_pipeline_id: 1,
queue: VecDeque::new(),
active_pipeline: None,
running_job: None,
status_reporter_addr: None,
completed: VecDeque::new(),
ci_snapshot: Arc::new(Mutex::new(LocalCiSnapshot::default())),
}
}
pub fn with_status_reporter(mut self, addr: ActorAddress) -> Self {
self.status_reporter_addr = Some(addr);
self
}
pub fn with_ci_yaml(mut self, yaml: CiYaml) -> Self {
self.ci_yaml = Some(yaml);
self
}
pub fn ci_snapshot(&self) -> Arc<Mutex<LocalCiSnapshot>> {
Arc::clone(&self.ci_snapshot)
}
pub fn pipelines(&self) -> &HashMap<PipelineId, PipelineExecution> {
&self.pipelines
}
pub fn completed(&self) -> &VecDeque<PipelineExecution> {
&self.completed
}
pub fn queue(&self) -> &VecDeque<PipelineId> {
&self.queue
}
pub fn active_pipeline(&self) -> Option<PipelineId> {
self.active_pipeline
}
pub fn running_job(&self) -> Option<&(JobId, ActorAddress)> {
self.running_job.as_ref()
}
fn handle_webhook(&mut self, ctx: &Ctx, event: WebhookEvent) {
let ci = match &self.ci_yaml {
Some(ci) => ci.clone(),
None => return,
};
let matched = yaml::matching_pipelines(&ci, &event);
for pipeline_name in matched {
let pipeline_def = &ci.pipelines[&pipeline_name];
let pipeline_id = PipelineId(self.next_pipeline_id);
self.next_pipeline_id += 1;
let job_defs: Vec<_> = pipeline_def
.jobs
.iter()
.map(|(name, def)| yaml::to_job_definition(name, def))
.collect();
let pipeline = PipelineExecution::new(
pipeline_id,
pipeline_name.clone(),
event.repo_owner.clone(),
event.repo_name.clone(),
event.commit_sha.clone(),
event.branch.clone(),
job_defs,
);
self.emit_status(
ctx,
StatusUpdate {
repo_owner: event.repo_owner.clone(),
repo_name: event.repo_name.clone(),
commit_sha: event.commit_sha.clone(),
state: "pending".into(),
context: format!("ci/{pipeline_name}"),
description: format!("Pipeline '{pipeline_name}' is pending"),
},
);
self.pipelines.insert(pipeline_id, pipeline);
self.enqueue_pipeline(pipeline_id, &event.branch);
}
self.try_schedule_next(ctx);
}
/// Enqueue a pipeline, superseding any queued pipeline for the same branch.
fn enqueue_pipeline(&mut self, pipeline_id: PipelineId, branch: &str) {
// Scan queue for entry with same branch (not the active pipeline).
let supersede_idx = self.queue.iter().position(|&qid| {
self.pipelines
.get(&qid)
.map(|p| p.branch == branch)
.unwrap_or(false)
});
if let Some(idx) = supersede_idx {
let old_id = self.queue[idx];
// Mark old pipeline as superseded.
if let Some(old_pipeline) = self.pipelines.get_mut(&old_id) {
old_pipeline.status = PipelineStatus::Error {
reason: "superseded".into(),
};
// Mark all pending jobs as skipped.
let job_names: Vec<String> = old_pipeline.jobs.keys().cloned().collect();
for name in job_names {
if old_pipeline.jobs[&name].status == JobStatus::Pending {
old_pipeline.set_job_status(&name, JobStatus::Skipped);
}
}
}
// Archive the superseded pipeline.
if let Some(old_pipeline) = self.pipelines.remove(&old_id) {
self.archive_pipeline(old_pipeline);
}
// Replace queue entry.
self.queue[idx] = pipeline_id;
} else {
self.queue.push_back(pipeline_id);
}
}
/// Core scheduling: one job at a time.
fn try_schedule_next(&mut self, ctx: &Ctx) {
// If a job is already running, nothing to do.
if self.running_job.is_some() {
return;
}
// If we have an active pipeline, try to find eligible jobs.
if let Some(active_id) = self.active_pipeline {
if let Some(pipeline) = self.pipelines.get(&active_id) {
let eligible = pipeline.eligible_jobs();
if !eligible.is_empty() {
let job_name = eligible[0].clone();
self.start_job(ctx, active_id, &job_name);
return;
}
// No eligible jobs — check if pipeline is terminal.
if pipeline.status.is_terminal() {
let pipeline = self.pipelines.remove(&active_id).unwrap();
self.emit_status(
ctx,
StatusUpdate {
repo_owner: pipeline.repo_owner.clone(),
repo_name: pipeline.repo_name.clone(),
commit_sha: pipeline.commit_sha.clone(),
state: pipeline.status.forgejo_state().into(),
context: format!("ci/{}", pipeline.pipeline_name),
description: format!(
"Pipeline '{}' {}",
pipeline.pipeline_name,
pipeline.status.forgejo_state()
),
},
);
self.archive_pipeline(pipeline);
self.active_pipeline = None;
// Recurse to pick next from queue.
self.try_schedule_next(ctx);
return;
}
}
// Pipeline exists but no eligible jobs and not terminal — waiting for running job.
return;
}
// No active pipeline — pop from queue.
if let Some(next_id) = self.queue.pop_front() {
self.active_pipeline = Some(next_id);
self.try_schedule_next(ctx);
}
}
fn start_job(&mut self, ctx: &Ctx, pipeline_id: PipelineId, job_name: &str) {
let pipeline = match self.pipelines.get_mut(&pipeline_id) {
Some(p) => p,
None => return,
};
let job = match pipeline.jobs.get_mut(job_name) {
Some(j) => j,
None => return,
};
job.status = JobStatus::Running;
let work_dir = format!(
"{}/pipeline-{}",
self.config.work_dir, pipeline_id.0
);
// Build CI env overrides.
let mut env_overrides = HashMap::new();
env_overrides.insert("CI".into(), "true".into());
env_overrides.insert("CI_COMMIT_SHA".into(), pipeline.commit_sha.clone());
env_overrides.insert("CI_BRANCH".into(), pipeline.branch.clone());
env_overrides.insert("CI_PIPELINE_ID".into(), pipeline_id.0.to_string());
env_overrides.insert("CI_JOB_NAME".into(), job_name.to_string());
let start_job = LocalStartJob {
job_id: job.job_id.clone(),
work_dir: work_dir.clone(),
job_def: job.definition.clone(),
env_overrides,
};
// Perform git checkout inline (blocks this worker, acceptable for local runner).
let sha = pipeline.commit_sha.clone();
let repo_url = self.config.repo_url.clone();
let git_ok = self.git_checkout(&repo_url, &sha, &work_dir);
if !git_ok {
if let Some(pipeline) = self.pipelines.get_mut(&pipeline_id) {
pipeline.set_job_status(
job_name,
JobStatus::Failed {
reason: "git checkout failed".into(),
},
);
}
self.try_schedule_next(ctx);
return;
}
// Emit per-job running status.
if let Some(pipeline) = self.pipelines.get(&pipeline_id) {
self.emit_status(
ctx,
StatusUpdate {
repo_owner: pipeline.repo_owner.clone(),
repo_name: pipeline.repo_name.clone(),
commit_sha: pipeline.commit_sha.clone(),
state: "pending".into(),
context: format!("ci/{job_name}"),
description: format!("Job '{job_name}' is running"),
},
);
}
// Spawn LocalRunner actor.
let runner =
crate::local_runner::LocalRunner::new(ctx.self_addr(), start_job);
match ctx.spawn(runner) {
Ok(runner_addr) => {
let job_id = JobId {
pipeline_id,
job_name: job_name.to_string(),
};
self.running_job = Some((job_id, runner_addr));
}
Err(_) => {
if let Some(pipeline) = self.pipelines.get_mut(&pipeline_id) {
pipeline.set_job_status(
job_name,
JobStatus::Failed {
reason: "failed to spawn runner".into(),
},
);
}
self.try_schedule_next(ctx);
}
}
}
fn git_checkout(&self, repo_url: &str, sha: &str, work_dir: &str) -> bool {
let path = std::path::Path::new(work_dir);
if path.join(".git").exists() {
// Already cloned — fetch and checkout.
let fetch = Command::new("git")
.args(["fetch", "origin"])
.current_dir(work_dir)
.output();
if fetch.is_err() || !fetch.unwrap().status.success() {
return false;
}
let checkout = Command::new("git")
.args(["checkout", sha])
.current_dir(work_dir)
.output();
checkout.map(|o| o.status.success()).unwrap_or(false)
} else {
// Fresh clone.
if let Some(parent) = path.parent() {
let _ = std::fs::create_dir_all(parent);
}
let clone = Command::new("git")
.args(["clone", repo_url, work_dir])
.output();
if clone.is_err() || !clone.as_ref().unwrap().status.success() {
return false;
}
let checkout = Command::new("git")
.args(["checkout", sha])
.current_dir(work_dir)
.output();
checkout.map(|o| o.status.success()).unwrap_or(false)
}
}
fn handle_job_complete(&mut self, ctx: &Ctx, complete: JobComplete) {
let pipeline_id = complete.job_id.pipeline_id;
let job_name = complete.job_id.job_name.clone();
let status = match complete.result {
Ok(_) => JobStatus::Passed,
Err(ref failure) => JobStatus::Failed {
reason: failure.to_string(),
},
};
// Emit per-job final status.
if let Some(pipeline) = self.pipelines.get(&pipeline_id) {
self.emit_status(
ctx,
StatusUpdate {
repo_owner: pipeline.repo_owner.clone(),
repo_name: pipeline.repo_name.clone(),
commit_sha: pipeline.commit_sha.clone(),
state: match &status {
JobStatus::Passed => "success".into(),
_ => "failure".into(),
},
context: format!("ci/{job_name}"),
description: format!("Job '{job_name}' completed"),
},
);
}
if let Some(pipeline) = self.pipelines.get_mut(&pipeline_id) {
pipeline.set_job_status(&job_name, status);
}
// Clear running job.
self.running_job = None;
// Schedule next.
self.try_schedule_next(ctx);
}
fn emit_status(&self, ctx: &Ctx, update: StatusUpdate) {
if let Some(reporter_addr) = self.status_reporter_addr {
let _ = ctx.send(
reporter_addr,
StatusReporterMsg::Report {
update,
forgejo_url: self.config.ci.forgejo_url.clone(),
forgejo_token: self.config.ci.forgejo_token.clone(),
},
);
}
}
fn archive_pipeline(&mut self, pipeline: PipelineExecution) {
self.completed.push_back(pipeline);
if self.completed.len() > 50 {
self.completed.pop_front();
}
}
fn update_snapshot(&self) {
let active: Vec<PipelineExecution> = self.pipelines.values().cloned().collect();
let recent: Vec<PipelineExecution> = self
.completed
.iter()
.rev()
.take(20)
.cloned()
.collect();
if let Ok(mut snap) = self.ci_snapshot.lock() {
snap.active_pipelines = active;
snap.recent_pipelines = recent;
snap.has_running_job = self.running_job.is_some();
}
}
}
impl ActorInterface for LocalCoordinator {
type Incoming = LocalCoordinatorMsg;
type Response = ();
fn handle(&mut self, ctx: &Ctx, msg: LocalCoordinatorMsg) {
match msg {
LocalCoordinatorMsg::Webhook(event) => self.handle_webhook(ctx, event),
LocalCoordinatorMsg::SetCiYaml(yaml) => {
self.ci_yaml = Some(yaml);
}
LocalCoordinatorMsg::JobProgress(progress) => {
if let Some(pipeline) = self.pipelines.get_mut(&progress.job_id.pipeline_id) {
if let Some(job) = pipeline.jobs.get_mut(&progress.job_id.job_name) {
job.output_lines.push(progress.output_line);
}
}
}
LocalCoordinatorMsg::JobComplete(complete) => {
self.handle_job_complete(ctx, complete);
}
LocalCoordinatorMsg::GitReady {
pipeline_id,
job_name,
work_dir,
} => {
// Git ready is used in the async variant; for now handled inline in start_job.
let _ = (pipeline_id, job_name, work_dir);
}
}
self.update_snapshot();
}
}

View file

@ -0,0 +1,237 @@
//! LocalRunner actor: executes job commands directly on the host via shell.
//!
//! Short-lived actor, one per job. Spawned by LocalCoordinator when a job
//! is ready to execute.
use std::io::BufRead;
use std::process::{Command, Stdio};
use std::sync::{Arc, Mutex};
use swactor::actor::{ActorAddress, ActorInterface, Ctx};
use crate::local_coordinator::LocalCoordinatorMsg;
use crate::{JobComplete, JobFailure, JobProgress, JobSuccess, LocalStartJob};
/// Messages the LocalRunner can receive.
#[derive(Debug, Clone)]
pub enum LocalRunnerMsg {
/// Begin executing the job (sent to self in on_start).
Execute,
/// Simulated: job completed (for testing without real shell).
SimComplete(Result<(), String>),
}
/// LocalRunner actor state.
pub struct LocalRunner {
coordinator_addr: ActorAddress,
start_job: LocalStartJob,
}
impl LocalRunner {
pub fn new(coordinator_addr: ActorAddress, start_job: LocalStartJob) -> Self {
Self {
coordinator_addr,
start_job,
}
}
/// Execute all commands in the job definition, streaming output back.
fn execute(&self, ctx: &Ctx) {
let job_id = &self.start_job.job_id;
let work_dir = &self.start_job.work_dir;
let timeout_secs = self.start_job.job_def.timeout_secs;
for cmd_str in &self.start_job.job_def.run {
// Send progress: command being run.
let _ = ctx.send(
self.coordinator_addr,
LocalCoordinatorMsg::JobProgress(JobProgress {
job_id: job_id.clone(),
output_line: format!("$ {cmd_str}"),
}),
);
let child_result = Command::new("sh")
.arg("-c")
.arg(cmd_str)
.current_dir(work_dir)
.envs(&self.start_job.env_overrides)
.envs(&self.start_job.job_def.env)
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn();
let mut child = match child_result {
Ok(c) => c,
Err(e) => {
let _ = ctx.send(
self.coordinator_addr,
LocalCoordinatorMsg::JobComplete(JobComplete {
job_id: job_id.clone(),
result: Err(JobFailure::ExecError(e.to_string())),
artifacts: Vec::new(),
}),
);
ctx.stop_self();
return;
}
};
// Timeout mechanism: share child handle, spawn thread that kills after timeout.
let kill_flag = Arc::new(Mutex::new(false));
let kill_flag_clone = Arc::clone(&kill_flag);
// Use a pipe to signal the timeout thread when the command finishes.
let (done_tx, done_rx) = std::sync::mpsc::channel::<()>();
let timeout_handle = std::thread::spawn(move || {
// Wait for either timeout or command completion.
if done_rx.recv_timeout(std::time::Duration::from_secs(timeout_secs)).is_err() {
*kill_flag_clone.lock().unwrap() = true;
}
});
// Read stdout line-by-line.
let stdout = child.stdout.take();
let stderr = child.stderr.take();
let mut last_lines: Vec<String> = Vec::new();
if let Some(stdout) = stdout {
let reader = std::io::BufReader::new(stdout);
for line in reader.lines() {
if let Ok(line) = line {
let _ = ctx.send(
self.coordinator_addr,
LocalCoordinatorMsg::JobProgress(JobProgress {
job_id: job_id.clone(),
output_line: line.clone(),
}),
);
last_lines.push(line);
if last_lines.len() > 50 {
last_lines.remove(0);
}
}
}
}
if let Some(stderr) = stderr {
let reader = std::io::BufReader::new(stderr);
for line in reader.lines() {
if let Ok(line) = line {
let _ = ctx.send(
self.coordinator_addr,
LocalCoordinatorMsg::JobProgress(JobProgress {
job_id: job_id.clone(),
output_line: format!("[stderr] {line}"),
}),
);
last_lines.push(line);
if last_lines.len() > 50 {
last_lines.remove(0);
}
}
}
}
let status = child.wait();
// Signal timeout thread that command finished.
let _ = done_tx.send(());
let _ = timeout_handle.join();
// Check if killed by timeout.
if *kill_flag.lock().unwrap() {
let _ = ctx.send(
self.coordinator_addr,
LocalCoordinatorMsg::JobComplete(JobComplete {
job_id: job_id.clone(),
result: Err(JobFailure::Timeout),
artifacts: Vec::new(),
}),
);
ctx.stop_self();
return;
}
match status {
Ok(exit) if exit.success() => {
// Command passed, continue to next.
}
Ok(exit) => {
let exit_code = exit.code().unwrap_or(-1);
let _ = ctx.send(
self.coordinator_addr,
LocalCoordinatorMsg::JobComplete(JobComplete {
job_id: job_id.clone(),
result: Err(JobFailure::CommandFailed {
exit_code,
last_lines,
}),
artifacts: Vec::new(),
}),
);
ctx.stop_self();
return;
}
Err(e) => {
let _ = ctx.send(
self.coordinator_addr,
LocalCoordinatorMsg::JobComplete(JobComplete {
job_id: job_id.clone(),
result: Err(JobFailure::ExecError(e.to_string())),
artifacts: Vec::new(),
}),
);
ctx.stop_self();
return;
}
}
}
// All commands passed.
let _ = ctx.send(
self.coordinator_addr,
LocalCoordinatorMsg::JobComplete(JobComplete {
job_id: job_id.clone(),
result: Ok(JobSuccess),
artifacts: Vec::new(),
}),
);
ctx.stop_self();
}
}
impl ActorInterface for LocalRunner {
type Incoming = LocalRunnerMsg;
type Response = ();
fn on_start(&mut self, ctx: &Ctx) {
let _ = ctx.send(ctx.self_addr(), LocalRunnerMsg::Execute);
}
fn handle(&mut self, ctx: &Ctx, msg: LocalRunnerMsg) {
match msg {
LocalRunnerMsg::Execute => {
self.execute(ctx);
}
LocalRunnerMsg::SimComplete(result) => {
let complete = JobComplete {
job_id: self.start_job.job_id.clone(),
result: match result {
Ok(()) => Ok(JobSuccess),
Err(msg) => Err(JobFailure::CommandFailed {
exit_code: 1,
last_lines: vec![msg],
}),
},
artifacts: Vec::new(),
};
let _ = ctx.send(
self.coordinator_addr,
LocalCoordinatorMsg::JobComplete(complete),
);
ctx.stop_self();
}
}
}
}

View file

@ -0,0 +1,75 @@
//! StatusReporter actor: fire-and-forget Forgejo commit status updates.
//!
//! Receives status update messages and POSTs them to the Forgejo API.
use swactor::actor::{ActorInterface, Ctx};
use crate::StatusUpdate;
/// Messages the StatusReporter can receive.
#[derive(Debug, Clone)]
pub enum StatusReporterMsg {
Report {
update: StatusUpdate,
forgejo_url: String,
forgejo_token: String,
},
}
/// StatusReporter actor state.
pub struct StatusReporter;
impl StatusReporter {
pub fn new() -> Self {
Self
}
#[cfg(feature = "local")]
fn post_status(update: &StatusUpdate, forgejo_url: &str, forgejo_token: &str) {
let url = format!(
"{}/api/v1/repos/{}/{}/statuses/{}",
forgejo_url.trim_end_matches('/'),
update.repo_owner,
update.repo_name,
update.commit_sha,
);
let body = serde_json::json!({
"state": update.state,
"context": update.context,
"description": update.description,
});
let result = ureq::post(&url)
.set("Authorization", &format!("token {forgejo_token}"))
.set("Content-Type", "application/json")
.send_string(&body.to_string());
if let Err(e) = result {
eprintln!("StatusReporter: failed to post status to {url}: {e}");
}
}
}
impl ActorInterface for StatusReporter {
type Incoming = StatusReporterMsg;
type Response = ();
fn handle(&mut self, _ctx: &Ctx, msg: StatusReporterMsg) {
match msg {
StatusReporterMsg::Report {
update,
forgejo_url,
forgejo_token,
} => {
#[cfg(feature = "local")]
Self::post_status(&update, &forgejo_url, &forgejo_token);
#[cfg(not(feature = "local"))]
{
let _ = (update, forgejo_url, forgejo_token);
}
}
}
}
}

View file

@ -0,0 +1,188 @@
//! Webhook HTTP listener: receives Forgejo webhook POSTs and forwards
//! them to the LocalCoordinator actor.
//!
//! Runs as a standard thread (not an actor) using `tiny_http`.
use crate::{EventType, WebhookEvent};
/// Start the webhook listener in a new thread.
///
/// Returns a join handle for the listener thread.
#[cfg(feature = "local")]
pub fn start_webhook_listener(
port: u16,
secret: String,
runtime: std::sync::Arc<swactor::runtime::Runtime>,
coordinator_addr: swactor::actor::ActorAddress,
) -> std::thread::JoinHandle<()> {
std::thread::Builder::new()
.name("webhook-listener".into())
.spawn(move || {
let server = tiny_http::Server::http(format!("0.0.0.0:{port}"))
.expect("failed to start webhook server");
eprintln!("Webhook listener on http://0.0.0.0:{port}");
for mut request in server.incoming_requests() {
let response = handle_request(&mut request, &secret, &runtime, coordinator_addr);
let _ = request.respond(response);
}
})
.expect("failed to spawn webhook listener thread")
}
#[cfg(feature = "local")]
fn handle_request(
request: &mut tiny_http::Request,
secret: &str,
runtime: &std::sync::Arc<swactor::runtime::Runtime>,
coordinator_addr: swactor::actor::ActorAddress,
) -> tiny_http::Response<std::io::Cursor<Vec<u8>>> {
use hmac::{Hmac, Mac};
use sha2::Sha256;
use crate::local_coordinator::LocalCoordinatorMsg;
// Only accept POST.
if request.method() != &tiny_http::Method::Post {
return tiny_http::Response::from_string("method not allowed")
.with_status_code(405);
}
// Read body.
let mut body = String::new();
if let Err(e) = std::io::Read::read_to_string(&mut request.as_reader(), &mut body) {
eprintln!("webhook: failed to read body: {e}");
return tiny_http::Response::from_string("bad request")
.with_status_code(400);
}
// Verify HMAC-SHA256 signature if secret is non-empty.
if !secret.is_empty() {
let sig_header = request
.headers()
.iter()
.find(|h| h.field.equiv("X-Forgejo-Signature"))
.map(|h| h.value.as_str().to_string());
match sig_header {
Some(sig_hex) => {
type HmacSha256 = Hmac<Sha256>;
let mut mac = HmacSha256::new_from_slice(secret.as_bytes())
.expect("HMAC key creation");
hmac::Mac::update(&mut mac, body.as_bytes());
let expected = hex::encode(mac.finalize().into_bytes());
if sig_hex != expected {
eprintln!("webhook: signature mismatch");
return tiny_http::Response::from_string("unauthorized")
.with_status_code(401);
}
}
None => {
eprintln!("webhook: missing signature header");
return tiny_http::Response::from_string("unauthorized")
.with_status_code(401);
}
}
}
// Determine event type from Forgejo header.
let event_header = request
.headers()
.iter()
.find(|h| h.field.equiv("X-Forgejo-Event"))
.map(|h| h.value.as_str().to_string())
.unwrap_or_default();
let event_type = match event_header.as_str() {
"push" => EventType::Push,
"create" => EventType::Tag,
"pull_request" => EventType::Merge,
other => {
eprintln!("webhook: ignoring event type '{other}'");
return tiny_http::Response::from_string("ignored").with_status_code(200);
}
};
// Parse JSON body to extract fields.
let json: serde_json::Value = match serde_json::from_str(&body) {
Ok(v) => v,
Err(e) => {
eprintln!("webhook: failed to parse JSON: {e}");
return tiny_http::Response::from_string("bad json").with_status_code(400);
}
};
let webhook_event = match parse_webhook_json(&json, event_type) {
Some(e) => e,
None => {
eprintln!("webhook: could not extract webhook fields from JSON");
return tiny_http::Response::from_string("bad payload").with_status_code(400);
}
};
// Send to coordinator.
let _ = runtime.send_to(coordinator_addr, LocalCoordinatorMsg::Webhook(webhook_event));
tiny_http::Response::from_string("ok").with_status_code(200)
}
/// Parse a Forgejo webhook JSON payload into a WebhookEvent.
pub fn parse_webhook_json(json: &serde_json::Value, event_type: EventType) -> Option<WebhookEvent> {
let repo = json.get("repository")?;
let repo_owner = repo
.get("owner")
.and_then(|o| o.get("login"))
.or_else(|| repo.get("owner").and_then(|o| o.get("username")))
.and_then(|v| v.as_str())?
.to_string();
let repo_name = repo.get("name").and_then(|v| v.as_str())?.to_string();
let (branch, commit_sha, tag) = match event_type {
EventType::Push => {
let reference = json.get("ref").and_then(|v| v.as_str()).unwrap_or("");
let branch = reference.strip_prefix("refs/heads/").unwrap_or(reference);
let sha = json
.get("after")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
(branch.to_string(), sha, None)
}
EventType::Tag => {
let reference = json.get("ref").and_then(|v| v.as_str()).unwrap_or("");
let tag_name = reference.strip_prefix("refs/tags/").unwrap_or(reference);
let sha = json
.get("sha")
.or_else(|| json.get("after"))
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
(String::new(), sha, Some(tag_name.to_string()))
}
EventType::Merge => {
let pr = json.get("pull_request")?;
let branch = pr
.get("head")
.and_then(|h| h.get("ref"))
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let sha = pr
.get("head")
.and_then(|h| h.get("sha"))
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
(branch, sha, None)
}
};
Some(WebhookEvent {
event_type,
repo_owner,
repo_name,
branch,
commit_sha,
tag,
})
}

View file

@ -0,0 +1,18 @@
[package]
name = "local-runner"
version = "0.1.0"
edition = "2024"
[[bin]]
name = "local-runner"
path = "src/main.rs"
[dependencies]
swactor = { path = "../..", features = ["serde"] }
swactor-ci = { path = "../ci", features = ["local"] }
runtime-dashboard = { path = "../runtime-dashboard", features = ["ci"] }
clap = { version = "4", features = ["derive"] }
ctrlc = "3"
iroh = "0.96"
tokio = { version = "1", features = ["rt-multi-thread"] }
serde_json = "1"

View file

@ -0,0 +1,361 @@
//! local-runner — single-machine CI runner for the Thinkpad.
//!
//! Receives Forgejo webhooks, queues pipelines, and executes jobs
//! one at a time for benchmark isolation.
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;
use clap::Parser;
use swactor::actor::ActorAddress;
use swactor::config::RuntimeConfig;
use swactor::runtime::Runtime;
use runtime_dashboard::ci_collector::{
CiSnapshot, CiStatsProvider, JobSnapshot, PipelineSnapshot, ProvisionerStatus,
};
use swactor_ci::local_coordinator::{LocalCiSnapshot, LocalCoordinator, LocalCoordinatorMsg};
use swactor_ci::pipeline::PipelineExecution;
use swactor_ci::status_reporter::StatusReporter;
use swactor_ci::webhook_server;
use swactor_ci::yaml;
use swactor_ci::{CiConfig, LocalCiConfig};
#[derive(Parser)]
#[command(name = "local-runner", about = "Swactor local CI runner")]
struct Args {
/// Webhook listen port.
#[arg(long, default_value = "8787")]
port: u16,
/// Forgejo instance URL.
#[arg(long, default_value = "")]
forgejo_url: String,
/// Forgejo API token.
#[arg(long, default_value = "")]
forgejo_token: String,
/// Webhook secret for HMAC verification (empty to skip).
#[arg(long, default_value = "")]
secret: String,
/// Path to .ci.yml file.
#[arg(long, default_value = ".ci.yml")]
yaml: String,
/// Base directory for git checkouts.
#[arg(long, default_value = "./ci-work")]
work_dir: String,
/// Git clone URL for the repository.
#[arg(long, default_value = "")]
repo_url: String,
/// Dashboard HTTP port (omit to disable).
#[arg(long)]
dashboard_port: Option<u16>,
/// Iroh Node ID of the ci-relay on the VPS (hex).
/// When set, webhooks arrive via iroh instead of HTTP.
#[arg(long)]
relay_node_id: Option<String>,
}
/// Bridge from LocalCiSnapshot to CiSnapshot for the dashboard.
struct LocalCiSnapshotProvider {
snapshot: Arc<Mutex<LocalCiSnapshot>>,
}
impl CiStatsProvider for LocalCiSnapshotProvider {
fn snapshot(&self) -> CiSnapshot {
let local = self.snapshot.lock().unwrap().clone();
CiSnapshot {
active_pipelines: local
.active_pipelines
.iter()
.map(pipeline_to_dashboard)
.collect(),
recent_pipelines: local
.recent_pipelines
.iter()
.map(pipeline_to_dashboard)
.collect(),
provisioner_status: ProvisionerStatus::Online,
active_instances: if local.has_running_job { 1 } else { 0 },
}
}
}
fn pipeline_to_dashboard(p: &PipelineExecution) -> PipelineSnapshot {
PipelineSnapshot {
pipeline_id: p.pipeline_id,
pipeline_name: p.pipeline_name.clone(),
repo_owner: p.repo_owner.clone(),
repo_name: p.repo_name.clone(),
commit_sha: p.commit_sha.clone(),
branch: p.branch.clone(),
status: p.status.clone(),
jobs: p
.jobs
.values()
.map(|j| JobSnapshot {
job_id: j.job_id.clone(),
job_name: j.definition.name.clone(),
status: j.status.clone(),
output_line_count: j.output_lines.len(),
})
.collect(),
}
}
fn main() {
let args = Args::parse();
let stop = Arc::new(AtomicBool::new(false));
// Signal handler.
{
let stop = Arc::clone(&stop);
ctrlc::set_handler(move || {
stop.store(true, Ordering::Relaxed);
})
.expect("failed to set signal handler");
}
// Optionally start dashboard.
let dash = args.dashboard_port.map(|port| {
let d = runtime_dashboard::start_dashboard(runtime_dashboard::DashboardConfig {
port,
..Default::default()
});
d.install_tracing();
d
});
// Create 2-thread runtime.
let num_threads = 2;
let collector = runtime_dashboard::collector::StatsCollector::new(num_threads);
let mut rt = Runtime::new(RuntimeConfig {
num_threads,
max_actors: 256,
channel_buffer_size: 2000,
..Default::default()
});
rt.set_stats_hook(collector.clone());
// Build config.
let ci_config = CiConfig {
webhook_port: args.port,
webhook_secret: args.secret.clone(),
forgejo_url: args.forgejo_url.clone(),
forgejo_token: args.forgejo_token.clone(),
data_dir: args.work_dir.clone(),
};
let local_config = LocalCiConfig {
ci: ci_config,
repo_url: args.repo_url.clone(),
work_dir: args.work_dir.clone(),
ci_yaml_path: args.yaml.clone(),
};
// Spawn StatusReporter.
let reporter_addr = rt
.spawn(StatusReporter::new())
.expect("failed to spawn StatusReporter");
// Spawn LocalCoordinator.
let coordinator = LocalCoordinator::new(local_config).with_status_reporter(reporter_addr);
let ci_snapshot = coordinator.ci_snapshot();
let coordinator_addr = rt
.spawn(coordinator)
.expect("failed to spawn LocalCoordinator");
// Load CI YAML from disk.
let yaml_content = std::fs::read_to_string(&args.yaml)
.unwrap_or_else(|e| panic!("failed to read {}: {e}", args.yaml));
let ci_yaml = yaml::parse_ci_yaml(&yaml_content)
.unwrap_or_else(|e| panic!("failed to parse CI YAML: {e}"));
// Start runtime.
let handle = rt.run().expect("failed to start runtime");
// Send CiYaml to coordinator.
let _ = handle
.runtime
.send_to(coordinator_addr, LocalCoordinatorMsg::SetCiYaml(ci_yaml));
// Wire dashboard.
if let Some(ref d) = dash {
d.set_runtime(Arc::clone(&handle.runtime), collector);
let provider = Arc::new(LocalCiSnapshotProvider {
snapshot: ci_snapshot,
});
d.set_ci(provider);
}
// Start webhook source: iroh relay or HTTP listener.
if let Some(ref relay_id_hex) = args.relay_node_id {
start_iroh_receiver(
relay_id_hex,
Arc::clone(&handle.runtime),
coordinator_addr,
Arc::clone(&stop),
);
} else {
let _webhook_handle = webhook_server::start_webhook_listener(
args.port,
args.secret,
Arc::clone(&handle.runtime),
coordinator_addr,
);
}
eprintln!("Local CI runner started");
if args.relay_node_id.is_some() {
eprintln!(" Webhook: via iroh relay");
} else {
eprintln!(" Webhook: http://0.0.0.0:{}", args.port);
}
eprintln!(" YAML: {}", args.yaml);
eprintln!(" Workdir: {}", args.work_dir);
if let Some(port) = args.dashboard_port {
eprintln!(" Dashboard: http://0.0.0.0:{port}");
}
// Main loop — just wait for ctrlc.
while !stop.load(Ordering::Relaxed) {
thread::sleep(Duration::from_millis(100));
}
eprintln!("\nShutting down...");
handle.shutdown();
if let Some(d) = dash {
d.shutdown();
}
handle.join();
}
// ─── Iroh Webhook Receiver ──────────────────────────────────────────────────
/// ALPN protocol identifier — must match ci-relay.
const CI_ALPN: &[u8] = b"swactor/ci/1";
/// Connect to the VPS ci-relay via iroh and receive WebhookEvents.
///
/// Runs in a background thread with its own tokio runtime.
fn start_iroh_receiver(
relay_id_hex: &str,
swactor_rt: Arc<Runtime>,
coordinator_addr: ActorAddress,
stop: Arc<AtomicBool>,
) {
let relay_key: iroh::PublicKey = relay_id_hex
.parse()
.unwrap_or_else(|e| panic!("invalid relay node ID '{relay_id_hex}': {e}"));
thread::Builder::new()
.name("iroh-receiver".into())
.spawn(move || {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("failed to build tokio runtime for iroh receiver");
rt.block_on(async move {
let endpoint = iroh::Endpoint::builder()
.alpns(vec![CI_ALPN.to_vec()])
.relay_mode(iroh::RelayMode::Default)
.bind()
.await
.expect("failed to bind iroh endpoint");
eprintln!(" Iroh local ID: {}", endpoint.id());
eprintln!(" Connecting to relay {relay_key}...");
let conn = endpoint
.connect(relay_key, CI_ALPN)
.await
.expect("failed to connect to ci-relay");
eprintln!(" Connected to relay!");
// Receive loop: the relay opens uni streams to send us events.
while !stop.load(Ordering::Relaxed) {
match tokio::time::timeout(Duration::from_secs(1), conn.accept_uni()).await {
Ok(Ok(mut recv)) => {
match read_tagged_message(&mut recv).await {
Ok((tag, payload)) => {
if tag == "ci::WebhookEvent" {
match serde_json::from_slice::<swactor_ci::WebhookEvent>(
&payload,
) {
Ok(event) => {
eprintln!(
"iroh: received webhook {} on {}",
event
.commit_sha
.get(..8)
.unwrap_or(&event.commit_sha),
event.branch,
);
let _ = swactor_rt.send_to(
coordinator_addr,
LocalCoordinatorMsg::Webhook(event),
);
}
Err(e) => {
eprintln!("iroh: failed to deserialize event: {e}")
}
}
} else {
eprintln!("iroh: unknown tag '{tag}', ignoring");
}
}
Err(e) => {
eprintln!("iroh: read error: {e}");
break;
}
}
}
Ok(Err(e)) => {
eprintln!("iroh: connection error: {e}");
break;
}
Err(_) => {
// Timeout — just loop and check stop flag.
}
}
}
endpoint.close().await;
});
})
.expect("failed to spawn iroh-receiver thread");
}
/// Read a tagged message from a QUIC recv stream.
///
/// Frame format: `[4B tag_len][tag_bytes][payload_bytes]`
async fn read_tagged_message(
recv: &mut iroh::endpoint::RecvStream,
) -> Result<(String, Vec<u8>), Box<dyn std::error::Error>> {
let mut tag_len_buf = [0u8; 4];
recv.read_exact(&mut tag_len_buf).await?;
let tag_len = u32::from_be_bytes(tag_len_buf) as usize;
if tag_len > 1024 {
return Err("tag too large".into());
}
let mut tag_buf = vec![0u8; tag_len];
recv.read_exact(&mut tag_buf).await?;
let tag = String::from_utf8(tag_buf)?;
let payload = recv.read_to_end(64 * 1024).await?;
Ok((tag, payload))
}

View file

@ -0,0 +1,552 @@
//! Local CI simulation: deterministic, round-based execution of the local
//! coordinator's queue + scheduling logic.
//!
//! No actors, no IO. Models the one-at-a-time scheduling with supersede.
//! Follows the same pattern as `sim.rs`.
use std::collections::{HashMap, VecDeque};
use swactor_ci::pipeline::PipelineExecution;
use swactor_ci::yaml::{self, CiYaml};
use swactor_ci::{JobId, JobStatus, PipelineId, PipelineStatus, StatusUpdate, WebhookEvent};
// ─── Simulation Config ──────────────────────────────────────────────────────
/// Configuration for a local CI simulation run.
#[derive(Debug, Clone)]
pub struct LocalSimConfig {
pub name: String,
pub num_rounds: usize,
pub ci_yaml: String,
pub webhook_schedule: Vec<(usize, WebhookEvent)>,
/// Rounds a job takes to execute.
pub job_duration: usize,
/// Force specific jobs to fail: (round, job_name_substring).
pub job_failure_schedule: Vec<(usize, String)>,
}
impl Default for LocalSimConfig {
fn default() -> Self {
Self {
name: "local-sim".into(),
num_rounds: 50,
ci_yaml: String::new(),
webhook_schedule: Vec::new(),
job_duration: 3,
job_failure_schedule: Vec::new(),
}
}
}
// ─── Simulation Trace ───────────────────────────────────────────────────────
#[derive(Debug, Clone)]
pub enum LocalSimEvent {
WebhookReceived { commit_sha: String },
PipelineCreated { pipeline_id: PipelineId, name: String },
PipelineSuperseded { pipeline_id: PipelineId },
PipelineCompleted { pipeline_id: PipelineId, status: PipelineStatus },
JobStarted { job_id: JobId },
JobCompleted { job_id: JobId, passed: bool },
JobSkipped { job_id: JobId },
}
#[derive(Debug, Clone)]
pub struct LocalSimSnapshot {
pub queued_pipelines: usize,
pub active_pipeline: Option<PipelineId>,
pub running_job: Option<JobId>,
pub completed_pipelines: usize,
}
#[derive(Debug, Clone)]
pub struct LocalSimTrace {
pub name: String,
pub events: Vec<(usize, LocalSimEvent)>,
pub snapshots: Vec<LocalSimSnapshot>,
pub status_updates: Vec<StatusUpdate>,
pub num_rounds: usize,
pub final_pipelines: Vec<PipelineExecution>,
}
// ─── Simulation State ───────────────────────────────────────────────────────
struct RunningJob {
job_id: JobId,
started_round: usize,
}
/// Run a local CI simulation and return the trace.
pub fn run_simulation(config: LocalSimConfig) -> LocalSimTrace {
let ci_yaml: CiYaml =
yaml::parse_ci_yaml(&config.ci_yaml).expect("LocalSimConfig.ci_yaml must be valid YAML");
let mut events: Vec<(usize, LocalSimEvent)> = Vec::new();
let mut snapshots: Vec<LocalSimSnapshot> = Vec::new();
let mut all_status_updates: Vec<StatusUpdate> = Vec::new();
// Coordinator state.
let mut pipelines: HashMap<PipelineId, PipelineExecution> = HashMap::new();
let mut next_pipeline_id: u64 = 1;
let mut queue: VecDeque<PipelineId> = VecDeque::new();
let mut active_pipeline: Option<PipelineId> = None;
let mut running_job: Option<RunningJob> = None;
let mut completed_pipelines: Vec<PipelineExecution> = Vec::new();
for round in 1..=config.num_rounds {
// 1. Inject webhook events for this round.
for (sched_round, event) in &config.webhook_schedule {
if *sched_round == round {
events.push((
round,
LocalSimEvent::WebhookReceived {
commit_sha: event.commit_sha.clone(),
},
));
let matched = yaml::matching_pipelines(&ci_yaml, event);
for pipeline_name in matched {
let pipeline_id = PipelineId(next_pipeline_id);
next_pipeline_id += 1;
let pipeline_def = &ci_yaml.pipelines[&pipeline_name];
let job_defs: Vec<_> = pipeline_def
.jobs
.iter()
.map(|(name, def)| yaml::to_job_definition(name, def))
.collect();
let pipeline = PipelineExecution::new(
pipeline_id,
pipeline_name.clone(),
event.repo_owner.clone(),
event.repo_name.clone(),
event.commit_sha.clone(),
event.branch.clone(),
job_defs,
);
events.push((
round,
LocalSimEvent::PipelineCreated {
pipeline_id,
name: pipeline_name.clone(),
},
));
all_status_updates.push(StatusUpdate {
repo_owner: event.repo_owner.clone(),
repo_name: event.repo_name.clone(),
commit_sha: event.commit_sha.clone(),
state: "pending".into(),
context: format!("ci/{pipeline_name}"),
description: format!("Pipeline '{pipeline_name}' is pending"),
});
pipelines.insert(pipeline_id, pipeline);
// Enqueue with supersede logic.
let supersede_idx = queue.iter().position(|&qid| {
pipelines
.get(&qid)
.map(|p| p.branch == event.branch)
.unwrap_or(false)
});
if let Some(idx) = supersede_idx {
let old_id = queue[idx];
if let Some(old_pipeline) = pipelines.get_mut(&old_id) {
old_pipeline.status = PipelineStatus::Error {
reason: "superseded".into(),
};
let job_names: Vec<String> =
old_pipeline.jobs.keys().cloned().collect();
for name in job_names {
if old_pipeline.jobs[&name].status == JobStatus::Pending {
old_pipeline.set_job_status(&name, JobStatus::Skipped);
}
}
}
events.push((
round,
LocalSimEvent::PipelineSuperseded {
pipeline_id: old_id,
},
));
if let Some(old_pipeline) = pipelines.remove(&old_id) {
// Emit terminal status for superseded pipeline.
all_status_updates.push(StatusUpdate {
repo_owner: old_pipeline.repo_owner.clone(),
repo_name: old_pipeline.repo_name.clone(),
commit_sha: old_pipeline.commit_sha.clone(),
state: "error".into(),
context: format!("ci/{}", old_pipeline.pipeline_name),
description: "superseded".into(),
});
completed_pipelines.push(old_pipeline);
}
queue[idx] = pipeline_id;
} else {
queue.push_back(pipeline_id);
}
}
}
}
// 2. Complete running job if it has reached duration.
if let Some(ref rj) = running_job {
if round - rj.started_round >= config.job_duration {
let job_id = rj.job_id.clone();
let should_fail = config
.job_failure_schedule
.iter()
.any(|(r, name_sub)| *r <= round && job_id.job_name.contains(name_sub.as_str()));
let passed = !should_fail;
if passed {
if let Some(pipeline) = pipelines.get_mut(&job_id.pipeline_id) {
pipeline.set_job_status(&job_id.job_name, JobStatus::Passed);
}
} else {
if let Some(pipeline) = pipelines.get_mut(&job_id.pipeline_id) {
pipeline.set_job_status(
&job_id.job_name,
JobStatus::Failed {
reason: "command failed".into(),
},
);
}
}
events.push((
round,
LocalSimEvent::JobCompleted {
job_id: job_id.clone(),
passed,
},
));
// Emit skipped events for any jobs that were skipped due to failure.
if !passed {
if let Some(pipeline) = pipelines.get(&job_id.pipeline_id) {
for (_name, job) in &pipeline.jobs {
if job.status == JobStatus::Skipped {
events.push((
round,
LocalSimEvent::JobSkipped {
job_id: job.job_id.clone(),
},
));
}
}
}
}
running_job = None;
}
}
// 3. Schedule next (one-at-a-time).
schedule_next(
&mut pipelines,
&mut queue,
&mut active_pipeline,
&mut running_job,
&mut completed_pipelines,
&mut events,
&mut all_status_updates,
round,
);
// 4. Snapshot.
snapshots.push(LocalSimSnapshot {
queued_pipelines: queue.len(),
active_pipeline,
running_job: running_job.as_ref().map(|rj| rj.job_id.clone()),
completed_pipelines: completed_pipelines.len(),
});
}
// Collect remaining active pipelines into final output.
let mut final_pipelines: Vec<PipelineExecution> = pipelines.into_values().collect();
final_pipelines.extend(completed_pipelines);
LocalSimTrace {
name: config.name,
events,
snapshots,
status_updates: all_status_updates,
num_rounds: config.num_rounds,
final_pipelines,
}
}
#[allow(clippy::too_many_arguments)]
fn schedule_next(
pipelines: &mut HashMap<PipelineId, PipelineExecution>,
queue: &mut VecDeque<PipelineId>,
active_pipeline: &mut Option<PipelineId>,
running_job: &mut Option<RunningJob>,
completed_pipelines: &mut Vec<PipelineExecution>,
events: &mut Vec<(usize, LocalSimEvent)>,
status_updates: &mut Vec<StatusUpdate>,
round: usize,
) {
// If a job is running, nothing to do.
if running_job.is_some() {
return;
}
// If we have an active pipeline, try eligible jobs.
if let Some(active_id) = *active_pipeline {
if let Some(pipeline) = pipelines.get(&active_id) {
let eligible = pipeline.eligible_jobs();
if !eligible.is_empty() {
let job_name = eligible[0].clone();
let job_id = JobId {
pipeline_id: active_id,
job_name: job_name.clone(),
};
// Mark as running.
if let Some(pipeline) = pipelines.get_mut(&active_id) {
if let Some(job) = pipeline.jobs.get_mut(&job_name) {
job.status = JobStatus::Running;
}
}
events.push((round, LocalSimEvent::JobStarted { job_id: job_id.clone() }));
*running_job = Some(RunningJob {
job_id,
started_round: round,
});
return;
}
// No eligible jobs — check terminal.
if pipeline.status.is_terminal() {
let pipeline = pipelines.remove(&active_id).unwrap();
let status = pipeline.status.clone();
status_updates.push(StatusUpdate {
repo_owner: pipeline.repo_owner.clone(),
repo_name: pipeline.repo_name.clone(),
commit_sha: pipeline.commit_sha.clone(),
state: pipeline.status.forgejo_state().into(),
context: format!("ci/{}", pipeline.pipeline_name),
description: format!(
"Pipeline '{}' {}",
pipeline.pipeline_name,
pipeline.status.forgejo_state()
),
});
events.push((
round,
LocalSimEvent::PipelineCompleted {
pipeline_id: active_id,
status,
},
));
completed_pipelines.push(pipeline);
*active_pipeline = None;
// Recurse.
schedule_next(
pipelines,
queue,
active_pipeline,
running_job,
completed_pipelines,
events,
status_updates,
round,
);
return;
}
}
// Pipeline exists but no eligible jobs and not terminal — waiting.
return;
}
// No active pipeline — pop from queue.
if let Some(next_id) = queue.pop_front() {
*active_pipeline = Some(next_id);
schedule_next(
pipelines,
queue,
active_pipeline,
running_job,
completed_pipelines,
events,
status_updates,
round,
);
}
}
// ─── Properties ─────────────────────────────────────────────────────────────
/// At most one job running in any snapshot.
pub fn check_one_at_a_time(trace: &LocalSimTrace) -> bool {
trace
.snapshots
.iter()
.all(|s| s.running_job.is_some() as usize <= 1)
}
/// Superseded pipelines never have a Running job.
pub fn check_superseded_no_running(trace: &LocalSimTrace) -> bool {
let superseded: Vec<PipelineId> = trace
.events
.iter()
.filter_map(|(_, e)| match e {
LocalSimEvent::PipelineSuperseded { pipeline_id } => Some(*pipeline_id),
_ => None,
})
.collect();
let started_jobs: Vec<&JobId> = trace
.events
.iter()
.filter_map(|(_, e)| match e {
LocalSimEvent::JobStarted { job_id } => Some(job_id),
_ => None,
})
.collect();
for pid in &superseded {
if started_jobs
.iter()
.any(|jid| jid.pipeline_id == *pid)
{
return false;
}
}
true
}
/// All non-superseded pipelines reach terminal status.
pub fn check_termination(trace: &LocalSimTrace) -> bool {
let superseded: Vec<PipelineId> = trace
.events
.iter()
.filter_map(|(_, e)| match e {
LocalSimEvent::PipelineSuperseded { pipeline_id } => Some(*pipeline_id),
_ => None,
})
.collect();
for pipeline in &trace.final_pipelines {
if superseded.contains(&pipeline.pipeline_id) {
continue;
}
if !pipeline.status.is_terminal() {
return false;
}
}
true
}
/// Within a pipeline, jobs respect dependency order.
pub fn check_dag_ordering(trace: &LocalSimTrace) -> bool {
let mut started: HashMap<(u64, &str), usize> = HashMap::new();
let mut completed: HashMap<(u64, &str), usize> = HashMap::new();
for (round, event) in &trace.events {
match event {
LocalSimEvent::JobStarted { job_id } => {
started.insert(
(job_id.pipeline_id.0, job_id.job_name.as_str()),
*round,
);
}
LocalSimEvent::JobCompleted { job_id, .. } => {
completed.insert(
(job_id.pipeline_id.0, job_id.job_name.as_str()),
*round,
);
}
_ => {}
}
}
for pipeline in &trace.final_pipelines {
for (name, job) in &pipeline.jobs {
if let Some(&start_round) = started.get(&(pipeline.pipeline_id.0, name.as_str())) {
for dep in &job.definition.needs {
if let Some(&dep_complete_round) =
completed.get(&(pipeline.pipeline_id.0, dep.as_str()))
{
if dep_complete_round > start_round {
return false;
}
}
}
}
}
}
true
}
/// Different branches execute in queue order (FIFO).
pub fn check_fifo_order(trace: &LocalSimTrace) -> bool {
// Collect pipeline creation order and first job start per pipeline.
let mut creation_order: Vec<PipelineId> = Vec::new();
let mut first_start: HashMap<PipelineId, usize> = HashMap::new();
for (round, event) in &trace.events {
if let LocalSimEvent::PipelineCreated { pipeline_id, .. } = event {
creation_order.push(*pipeline_id);
}
if let LocalSimEvent::JobStarted { job_id } = event {
first_start
.entry(job_id.pipeline_id)
.or_insert(*round);
}
}
// For each pair of pipelines created in order, if both started, the earlier-created
// one should have started no later.
for i in 0..creation_order.len() {
for j in (i + 1)..creation_order.len() {
let pid_a = creation_order[i];
let pid_b = creation_order[j];
if let (Some(&start_a), Some(&start_b)) =
(first_start.get(&pid_a), first_start.get(&pid_b))
{
if start_a > start_b {
return false;
}
}
}
}
true
}
/// Every webhook produces a terminal status (success/failure/error).
pub fn check_all_webhooks_terminate(trace: &LocalSimTrace) -> bool {
let webhook_commits: Vec<&str> = trace
.events
.iter()
.filter_map(|(_, e)| match e {
LocalSimEvent::WebhookReceived { commit_sha } => Some(commit_sha.as_str()),
_ => None,
})
.collect();
for sha in webhook_commits {
let has_terminal = trace.status_updates.iter().any(|u| {
u.commit_sha == sha
&& (u.state == "success" || u.state == "failure" || u.state == "error")
});
if !has_terminal {
return false;
}
}
true
}

View file

@ -1 +1,2 @@
pub mod local_sim;
pub mod sim; pub mod sim;

View file

@ -0,0 +1,256 @@
//! Property-based tests for the local CI simulation.
//!
//! These verify invariants that should hold across all possible simulation configurations.
use simulation::ci::local_sim::{self, LocalSimConfig};
use swactor_ci::{EventType, WebhookEvent};
fn simple_yaml() -> String {
r#"
pipelines:
check:
triggers:
- event: push
branches: ["*"]
jobs:
fmt:
run: cargo fmt -- --check
test:
needs: [fmt]
run: cargo test
"#
.into()
}
fn push(branch: &str, sha: &str) -> WebhookEvent {
WebhookEvent {
event_type: EventType::Push,
repo_owner: "user".into(),
repo_name: "repo".into(),
branch: branch.into(),
commit_sha: sha.into(),
tag: None,
}
}
// ─── Property: One-at-a-time ────────────────────────────────────────────────
#[test]
fn property_one_at_a_time_single_push() {
let config = LocalSimConfig {
name: "prop-1at1-single".into(),
num_rounds: 30,
ci_yaml: simple_yaml(),
webhook_schedule: vec![(1, push("main", "sha-1"))],
job_duration: 2,
..Default::default()
};
let trace = local_sim::run_simulation(config);
assert!(
local_sim::check_one_at_a_time(&trace),
"at most one job running at a time"
);
}
#[test]
fn property_one_at_a_time_burst() {
let config = LocalSimConfig {
name: "prop-1at1-burst".into(),
num_rounds: 100,
ci_yaml: simple_yaml(),
webhook_schedule: (1..=10)
.map(|i| (1, push(&format!("branch-{i}"), &format!("sha-{i}"))))
.collect(),
job_duration: 2,
..Default::default()
};
let trace = local_sim::run_simulation(config);
assert!(
local_sim::check_one_at_a_time(&trace),
"at most one job running at a time under burst"
);
}
#[test]
fn property_one_at_a_time_staggered() {
let config = LocalSimConfig {
name: "prop-1at1-stagger".into(),
num_rounds: 80,
ci_yaml: simple_yaml(),
webhook_schedule: vec![
(1, push("a", "sha-a")),
(3, push("b", "sha-b")),
(5, push("c", "sha-c")),
(7, push("d", "sha-d")),
],
job_duration: 3,
..Default::default()
};
let trace = local_sim::run_simulation(config);
assert!(
local_sim::check_one_at_a_time(&trace),
"at most one job running at a time under stagger"
);
}
// ─── Property: Supersede correctness ────────────────────────────────────────
#[test]
fn property_superseded_pipelines_never_run() {
let config = LocalSimConfig {
name: "prop-supersede".into(),
num_rounds: 60,
ci_yaml: simple_yaml(),
// Same branch, rapid pushes while jobs are long.
webhook_schedule: (1..=8)
.map(|i| (i, push("feature", &format!("sha-ss-{i}"))))
.collect(),
job_duration: 4,
..Default::default()
};
let trace = local_sim::run_simulation(config);
assert!(
local_sim::check_superseded_no_running(&trace),
"superseded pipelines should never have a running job"
);
}
// ─── Property: Termination ──────────────────────────────────────────────────
#[test]
fn property_all_non_superseded_terminate() {
let config = LocalSimConfig {
name: "prop-terminate".into(),
num_rounds: 100,
ci_yaml: simple_yaml(),
webhook_schedule: vec![
(1, push("a", "sha-t1")),
(2, push("b", "sha-t2")),
(3, push("a", "sha-t3")),
(10, push("c", "sha-t4")),
],
job_duration: 3,
..Default::default()
};
let trace = local_sim::run_simulation(config);
assert!(
local_sim::check_termination(&trace),
"all non-superseded pipelines must reach terminal status"
);
}
#[test]
fn property_all_webhooks_terminate() {
let config = LocalSimConfig {
name: "prop-wh-terminate".into(),
num_rounds: 100,
ci_yaml: simple_yaml(),
webhook_schedule: (1..=5)
.map(|i| (i * 3, push("main", &format!("sha-wh-{i}"))))
.collect(),
job_duration: 2,
..Default::default()
};
let trace = local_sim::run_simulation(config);
assert!(
local_sim::check_all_webhooks_terminate(&trace),
"every webhook must produce a terminal status"
);
}
// ─── Property: DAG ordering ─────────────────────────────────────────────────
#[test]
fn property_dag_ordering_deep_chain() {
let yaml = r#"
pipelines:
deep:
triggers:
- event: push
branches: ["*"]
jobs:
a:
run: echo a
b:
needs: [a]
run: echo b
c:
needs: [b]
run: echo c
d:
needs: [c]
run: echo d
"#;
let config = LocalSimConfig {
name: "prop-dag-deep".into(),
num_rounds: 50,
ci_yaml: yaml.into(),
webhook_schedule: vec![(1, push("main", "sha-dag"))],
job_duration: 2,
..Default::default()
};
let trace = local_sim::run_simulation(config);
assert!(
local_sim::check_dag_ordering(&trace),
"deep DAG ordering must be respected"
);
}
#[test]
fn property_dag_ordering_diamond() {
let yaml = r#"
pipelines:
diamond:
triggers:
- event: push
branches: ["*"]
jobs:
root:
run: echo root
left:
needs: [root]
run: echo left
right:
needs: [root]
run: echo right
merge:
needs: [left, right]
run: echo merge
"#;
let config = LocalSimConfig {
name: "prop-dag-diamond".into(),
num_rounds: 50,
ci_yaml: yaml.into(),
webhook_schedule: vec![(1, push("main", "sha-dia"))],
job_duration: 2,
..Default::default()
};
let trace = local_sim::run_simulation(config);
assert!(
local_sim::check_dag_ordering(&trace),
"diamond DAG ordering must be respected"
);
}
// ─── Property: FIFO across branches ─────────────────────────────────────────
#[test]
fn property_fifo_across_branches() {
let config = LocalSimConfig {
name: "prop-fifo".into(),
num_rounds: 80,
ci_yaml: simple_yaml(),
webhook_schedule: vec![
(1, push("a", "sha-f1")),
(2, push("b", "sha-f2")),
(3, push("c", "sha-f3")),
],
job_duration: 3,
..Default::default()
};
let trace = local_sim::run_simulation(config);
assert!(
local_sim::check_fifo_order(&trace),
"branches must execute in FIFO queue order"
);
}

View file

@ -0,0 +1,324 @@
//! Scenario tests for the local CI simulation.
//!
//! Each test tells a story: set up a scenario, run the simulation, verify outcomes.
use simulation::ci::local_sim::{self, LocalSimConfig, LocalSimEvent};
use swactor_ci::{EventType, WebhookEvent};
fn basic_ci_yaml() -> String {
r#"
pipelines:
check:
triggers:
- event: push
branches: ["*"]
exclude: ["master"]
jobs:
fmt:
run: cargo fmt -- --check
clippy:
run: cargo clippy
test:
needs: [fmt, clippy]
run: cargo test
full:
triggers:
- event: push
branches: ["master"]
jobs:
test:
run: cargo test --all-features
timeout: 600
bench:
needs: [test]
run: cargo bench
"#
.into()
}
fn push_event(branch: &str, sha: &str) -> WebhookEvent {
WebhookEvent {
event_type: EventType::Push,
repo_owner: "user".into(),
repo_name: "repo".into(),
branch: branch.into(),
commit_sha: sha.into(),
tag: None,
}
}
// ─── Scenario: Single push → single job → passes ────────────────────────────
#[test]
fn single_push_runs_and_completes() {
let yaml = r#"
pipelines:
check:
triggers:
- event: push
branches: ["*"]
jobs:
test:
run: cargo test
"#;
let config = LocalSimConfig {
name: "single-push".into(),
num_rounds: 20,
ci_yaml: yaml.into(),
webhook_schedule: vec![(1, push_event("main", "sha1"))],
job_duration: 2,
..Default::default()
};
let trace = local_sim::run_simulation(config);
let created = trace
.events
.iter()
.filter(|(_, e)| matches!(e, LocalSimEvent::PipelineCreated { .. }))
.count();
assert_eq!(created, 1);
let completed = trace
.events
.iter()
.filter(|(_, e)| matches!(e, LocalSimEvent::JobCompleted { passed: true, .. }))
.count();
assert_eq!(completed, 1);
assert!(local_sim::check_all_webhooks_terminate(&trace));
}
// ─── Scenario: Push A, push A again while queued → only latest runs ─────────
#[test]
fn supersede_queued_same_branch() {
let yaml = r#"
pipelines:
check:
triggers:
- event: push
branches: ["*"]
jobs:
test:
run: cargo test
"#;
let config = LocalSimConfig {
name: "supersede-queued".into(),
num_rounds: 30,
ci_yaml: yaml.into(),
// Push A at round 1 occupies the runner. Push A' at round 2 queues.
// Push A'' at round 3 should supersede A'.
webhook_schedule: vec![
(1, push_event("feature", "sha-a1")),
(2, push_event("feature", "sha-a2")),
(3, push_event("feature", "sha-a3")),
],
job_duration: 5,
..Default::default()
};
let trace = local_sim::run_simulation(config);
// sha-a2 should be superseded.
let superseded = trace
.events
.iter()
.filter(|(_, e)| matches!(e, LocalSimEvent::PipelineSuperseded { .. }))
.count();
assert!(superseded >= 1, "at least one pipeline should be superseded");
// sha-a2 should have an "error" status (superseded).
let a2_error = trace
.status_updates
.iter()
.any(|u| u.commit_sha == "sha-a2" && u.state == "error");
assert!(a2_error, "superseded pipeline should report error status");
// sha-a1 and sha-a3 should both reach terminal status.
let a1_terminal = trace
.status_updates
.iter()
.any(|u| u.commit_sha == "sha-a1" && (u.state == "success" || u.state == "failure"));
let a3_terminal = trace
.status_updates
.iter()
.any(|u| u.commit_sha == "sha-a3" && (u.state == "success" || u.state == "failure"));
assert!(a1_terminal, "first push should complete");
assert!(a3_terminal, "latest push should complete");
assert!(local_sim::check_superseded_no_running(&trace));
}
// ─── Scenario: Push A, push A while running → running completes, new queues ─
#[test]
fn push_while_running_does_not_supersede_active() {
let yaml = r#"
pipelines:
check:
triggers:
- event: push
branches: ["*"]
jobs:
test:
run: cargo test
"#;
let config = LocalSimConfig {
name: "no-supersede-active".into(),
num_rounds: 30,
ci_yaml: yaml.into(),
// Push A at round 1 starts running immediately.
// Push A' at round 2 should queue (not cancel the running job).
webhook_schedule: vec![
(1, push_event("feature", "sha-run1")),
(2, push_event("feature", "sha-run2")),
],
job_duration: 4,
..Default::default()
};
let trace = local_sim::run_simulation(config);
// Both should reach terminal status.
let run1_terminal = trace
.status_updates
.iter()
.any(|u| u.commit_sha == "sha-run1" && u.state == "success");
let run2_terminal = trace
.status_updates
.iter()
.any(|u| u.commit_sha == "sha-run2" && u.state == "success");
assert!(run1_terminal, "running pipeline should complete normally");
assert!(run2_terminal, "queued pipeline should run after");
assert!(local_sim::check_one_at_a_time(&trace));
}
// ─── Scenario: Push A, push B → both run in FIFO order ─────────────────────
#[test]
fn different_branches_run_fifo() {
let yaml = r#"
pipelines:
check:
triggers:
- event: push
branches: ["*"]
jobs:
test:
run: cargo test
"#;
let config = LocalSimConfig {
name: "fifo-branches".into(),
num_rounds: 30,
ci_yaml: yaml.into(),
webhook_schedule: vec![
(1, push_event("feature-a", "sha-fa")),
(1, push_event("feature-b", "sha-fb")),
],
job_duration: 3,
..Default::default()
};
let trace = local_sim::run_simulation(config);
// Both should complete.
assert!(local_sim::check_all_webhooks_terminate(&trace));
// FIFO order respected.
assert!(local_sim::check_fifo_order(&trace));
// One at a time.
assert!(local_sim::check_one_at_a_time(&trace));
}
// ─── Scenario: Job failure → dependents skipped → next pipeline starts ──────
#[test]
fn job_failure_skips_dependents_and_advances() {
let config = LocalSimConfig {
name: "failure-skip-advance".into(),
num_rounds: 40,
ci_yaml: basic_ci_yaml(),
webhook_schedule: vec![
(1, push_event("feature-x", "sha-fail")),
(2, push_event("feature-y", "sha-next")),
],
job_duration: 2,
// fmt fails.
job_failure_schedule: vec![(0, "fmt".into())],
..Default::default()
};
let trace = local_sim::run_simulation(config);
// test should be skipped in feature-x pipeline (needs fmt which fails).
let test_skipped = trace.events.iter().any(|(_, e)| {
matches!(e, LocalSimEvent::JobSkipped { job_id } if job_id.job_name == "test")
});
assert!(test_skipped, "test should be skipped when fmt fails");
// Both pipelines should reach terminal.
assert!(local_sim::check_all_webhooks_terminate(&trace));
// One at a time.
assert!(local_sim::check_one_at_a_time(&trace));
}
// ─── Scenario: Diamond DAG → jobs serialize respecting deps ─────────────────
#[test]
fn diamond_dag_serialized_with_deps() {
let yaml = r#"
pipelines:
diamond:
triggers:
- event: push
branches: ["*"]
jobs:
root:
run: echo root
left:
needs: [root]
run: echo left
right:
needs: [root]
run: echo right
merge:
needs: [left, right]
run: echo merge
"#;
let config = LocalSimConfig {
name: "diamond-dag".into(),
num_rounds: 50,
ci_yaml: yaml.into(),
webhook_schedule: vec![(1, push_event("main", "sha-diamond"))],
job_duration: 2,
..Default::default()
};
let trace = local_sim::run_simulation(config);
// All 4 jobs should complete.
let completed = trace
.events
.iter()
.filter(|(_, e)| matches!(e, LocalSimEvent::JobCompleted { .. }))
.count();
assert_eq!(completed, 4, "all 4 diamond jobs should complete");
// DAG ordering respected.
assert!(local_sim::check_dag_ordering(&trace));
// One at a time (serial).
assert!(local_sim::check_one_at_a_time(&trace));
// Pipeline should succeed.
let success = trace
.status_updates
.iter()
.any(|u| u.state == "success" && u.context == "ci/diamond");
assert!(success, "diamond pipeline should succeed");
}

View file

@ -0,0 +1,520 @@
# CI Webhook Relay via Iroh — Development History
> Covers the implementation of `ci-relay` and the iroh webhook receiver in
> `local-runner`, enabling Forgejo webhooks to reach a NAT'd CI runner via
> iroh's QUIC transport with automatic NAT traversal.
>
> ~3 files created · ~2 files modified · ~350 insertions
>
> *Branch: `spot-instance`*
---
## Table of Contents
1. [Problem & Motivation](#1-problem--motivation)
2. [Architecture](#2-architecture)
3. [What Was Built](#3-what-was-built)
4. [ci-relay Binary](#4-ci-relay-binary)
5. [local-runner Iroh Receiver](#5-local-runner-iroh-receiver)
6. [Wire Protocol](#6-wire-protocol)
7. [Connection Flow](#7-connection-flow)
8. [Design Decisions & Tradeoffs](#8-design-decisions--tradeoffs)
9. [Manual Testing Guide](#9-manual-testing-guide)
10. [Known Gaps & Future Improvements](#10-known-gaps--future-improvements)
---
## 1. Problem & Motivation
The CI runner (`local-runner`) was designed for same-LAN usage: Forgejo sends
webhooks over HTTP to the runner's listen port. In the real deployment:
- **Forgejo** runs on a VPS (`zachery.lol` / `139.59.195.69`)
- **CI runner** runs on a Thinkpad at home (`192.168.1.102`), behind NAT
The VPS cannot reach the Thinkpad directly — no inbound port is open, no
static IP, no UPnP. Traditional solutions (SSH reverse tunnel, VPN, port
forwarding on router) all require ongoing configuration and are fragile.
iroh is already integrated in swactor's distribution layer (`iroh_driver.rs`)
for SWIM protocol traffic. It provides QUIC connections with automatic NAT
traversal via relay servers — exactly what's needed to bridge the webhook gap.
### Why Not Just SSH Tunnel?
An SSH tunnel (`ssh -R 8787:localhost:8787 zachery.lol`) would work, but:
- Tunnels drop on network changes (laptop suspend, WiFi roaming)
- Requires autossh or systemd to keep alive
- Another moving part to debug when CI stops working
- Doesn't reuse any existing infrastructure
iroh handles reconnection, relay fallback, and NAT traversal automatically.
The implementation reuses the same tagged-message-over-QUIC-stream pattern
already proven in `iroh_driver.rs`.
---
## 2. Architecture
```
┌─────────────────────────────┐ ┌──────────────────────────────────┐
│ VPS (zachery.lol) │ │ Thinkpad (192.168.1.102) │
│ │ iroh │ │
│ Forgejo ──webhook──► Relay ├───────►│ local-runner │
│ :8787 │ QUIC │ (coordinator, runner, reporter) │
│ │ │ │
└─────────────────────────────┘ └──────────────────────────────────┘
```
**VPS side** — `ci-relay` binary:
- HTTP listener receives webhook POSTs from Forgejo (localhost only)
- iroh endpoint accepts the runner's inbound connection
- Forwards parsed `WebhookEvent` payloads over iroh uni streams
**Thinkpad side** — `local-runner` with `--relay-node-id`:
- Connects to the VPS relay's iroh endpoint on startup
- Receives `WebhookEvent` over iroh uni streams
- Feeds events into `LocalCoordinator` via existing `Webhook` message
- Status updates go directly Thinkpad → Forgejo API over HTTPS (no relay needed)
The relay is intentionally minimal — it's a bridge, not a CI component. All CI
logic stays in `local-runner`.
---
## 3. What Was Built
| Component | Location | Nature |
|-----------|----------|--------|
| ci-relay binary | `crates/ci-relay/Cargo.toml`, `src/main.rs` | **New** — VPS webhook relay |
| Iroh receiver | `crates/local-runner/src/main.rs` | **Modified** — iroh webhook source |
| Dependencies | `crates/local-runner/Cargo.toml` | **Modified** — added iroh, tokio, serde_json |
| Workspace | `Cargo.toml` | **Modified** — added ci-relay to members |
---
## 4. ci-relay Binary
### `crates/ci-relay/src/main.rs`
The relay runs two subsystems on a single process:
1. **iroh acceptor** (tokio task): accepts inbound connections from the runner,
caches the most recent one in `Arc<TokioMutex<Option<Connection>>>`
2. **HTTP listener** (main thread, blocking `tiny_http`): receives Forgejo
webhook POSTs, verifies HMAC, parses event, forwards over iroh
### Webhook Handling
Reuses the same verification and parsing logic as `webhook_server.rs`:
- HMAC-SHA256 verification via `X-Forgejo-Signature` header (skippable with empty secret)
- Event type from `X-Forgejo-Event` header: `push` → `Push`, `create` → `Tag`, `pull_request` → `Merge`
- JSON parsing via `parse_webhook_json()` (re-exported from `swactor-ci`)
The relay uses `parse_webhook_json` directly rather than duplicating parsing
logic. This keeps webhook interpretation consistent between HTTP and iroh paths.
### Forwarding
On webhook receipt, the relay:
1. Serializes the `WebhookEvent` to JSON
2. Opens a unidirectional QUIC stream on the cached connection
3. Writes the tagged message (`ci::WebhookEvent` tag + JSON payload)
4. Finishes the stream
If no runner is connected, the relay returns HTTP 502 to Forgejo. Forgejo will
retry the webhook per its configured retry policy.
### CLI
```
ci-relay [OPTIONS]
Options:
--port <PORT> HTTP port for Forgejo webhooks [default: 8787]
--secret <SECRET> HMAC-SHA256 secret [default: "" (no verification)]
```
On startup, the relay prints its iroh Node ID — this is the value the runner
needs for `--relay-node-id`.
---
## 5. local-runner Iroh Receiver
### New CLI Flag
```
--relay-node-id <HEX> Iroh Node ID of the VPS ci-relay
```
When `--relay-node-id` is provided:
- The HTTP webhook listener is **not started** (no port conflict, no exposure)
- An `iroh-receiver` thread starts instead
When omitted, behavior is unchanged — the HTTP listener starts on `--port`
as before.
### `start_iroh_receiver()`
Spawns a dedicated thread (`iroh-receiver`) with its own single-threaded tokio
runtime:
1. Creates an iroh `Endpoint` with ALPN `b"swactor/ci/1"`
2. Connects to the relay's `PublicKey` (parsed from the hex flag)
3. Enters a receive loop:
- `conn.accept_uni()` with 1-second timeout
- On stream: reads tagged message, deserializes `WebhookEvent`
- Sends `LocalCoordinatorMsg::Webhook(event)` to the coordinator via the swactor runtime
- On timeout: checks the `stop` flag (for graceful shutdown via Ctrl-C)
- On connection error: breaks and exits
The thread respects the same `AtomicBool` stop flag as the main loop, so
Ctrl-C cleanly shuts down both the swactor runtime and the iroh connection.
---
## 6. Wire Protocol
### ALPN
```rust
const CI_ALPN: &[u8] = b"swactor/ci/1";
```
Distinct from SWIM traffic (`b"swactor/swim/1"`). This allows both protocols
to coexist on the same iroh endpoint in the future if needed.
### Frame Format
Same tagged-message format as `iroh_driver.rs`:
```
[4 bytes: tag_len (big-endian u32)]
[tag_len bytes: tag string]
[remaining bytes: payload]
```
For webhook events:
- Tag: `"ci::WebhookEvent"` (17 bytes)
- Payload: JSON-serialized `WebhookEvent`
### Transport
Each webhook is one unidirectional QUIC stream. The relay opens the stream,
writes the tagged message, and finishes. The runner reads the message and the
stream closes. No persistent framing or multiplexing needed — QUIC streams
are lightweight.
---
## 7. Connection Flow
```
1. VPS starts ci-relay
→ iroh Endpoint binds
→ prints Node ID (ed25519 public key, hex)
→ HTTP listener starts on --port
→ waits for runner connection
2. Thinkpad starts local-runner --relay-node-id <hex>
→ iroh Endpoint binds
→ connects to relay's PublicKey
→ iroh handles NAT traversal (direct or via relay server)
→ relay logs "Runner connected: <runner-node-id>"
3. Forgejo sends webhook POST to localhost:8787 on VPS
→ relay verifies HMAC, parses event
→ relay opens uni stream on cached connection
→ writes tagged WebhookEvent
→ runner receives, deserializes, dispatches to coordinator
4. Coordinator triggers pipeline
→ StatusReporter posts status to Forgejo API directly
(Thinkpad → zachery.lol over HTTPS, no relay involvement)
```
The iroh connection is initiated by the runner (outbound from NAT), so no port
forwarding is needed. iroh's relay servers handle the initial rendezvous, then
attempt direct QUIC hole-punching for subsequent traffic.
---
## 8. Design Decisions & Tradeoffs
### 8.1 Separate Binary vs. Library Module
**Choice**: `ci-relay` is a standalone binary, not a module in `swactor-ci`.
**Why**: The relay runs on the VPS, which doesn't need swactor's runtime,
actors, or any CI execution logic. A small binary with minimal dependencies
deploys easily. It only depends on `swactor-ci` for `parse_webhook_json` and
the `WebhookEvent`/`EventType` types.
**Tradeoff**: Two binaries to build and deploy instead of one. Acceptable
given they run on different machines.
### 8.2 Runner Connects to Relay (Not Vice Versa)
**Choice**: The runner initiates the iroh connection to the relay.
**Why**: The runner is behind NAT. iroh can traverse NAT for established
connections, but the initial rendezvous requires at least one side to be
reachable. The VPS relay has a public IP and gets a stable relay URL from iroh's
infrastructure. The runner connects outbound, which always works regardless of
NAT type.
### 8.3 Single Cached Connection (Not Connection Pool)
**Choice**: The relay caches exactly one runner connection in
`Arc<TokioMutex<Option<Connection>>>`.
**Why**: There's one runner. If a new connection arrives (e.g., runner
restarts), it replaces the old one. No pool management needed.
**Tradeoff**: If multiple runners were needed, this would need a map. For
single-runner use, the simplicity is worth it.
### 8.4 Own Tokio Runtime Per Thread
**Choice**: The iroh-receiver thread creates its own single-threaded tokio
runtime rather than sharing the swactor runtime or the main thread's runtime.
**Why**: swactor's runtime is not tokio — it's a custom actor scheduler. The
iroh receiver needs async for QUIC operations. A dedicated single-threaded
runtime keeps the iroh I/O isolated from actor scheduling. Same pattern as
`IrohDriver` in the distribution layer (which owns a multi-thread runtime).
### 8.5 HTTP 502 When No Runner Connected
**Choice**: If Forgejo sends a webhook but no runner is connected, the relay
returns HTTP 502 (Bad Gateway).
**Why**: 502 tells Forgejo the upstream is unavailable. Forgejo will retry
the webhook according to its retry policy. This is better than 200 (silently
dropping) or 500 (suggesting a relay bug). When the runner reconnects, the
next webhook will succeed.
---
## 9. Manual Testing Guide
### Prerequisites
Build both binaries:
```bash
cargo build -p ci-relay -p local-runner
```
### 9.1 Local Smoke Test (Single Machine)
This tests the full relay path without needing two machines or Forgejo.
**Terminal 1 — Start the relay:**
```bash
./target/debug/ci-relay --port 9787
```
Output:
```
ci-relay started
Iroh Node ID: <NODE_ID_HEX>
Webhook HTTP: http://0.0.0.0:9787
Waiting for runner to connect...
Listening for webhooks...
```
Copy the Node ID.
**Terminal 2 — Start the runner:**
You need a `.ci.yml` file. Create a minimal one:
```yaml
# /tmp/test-ci.yml
pipelines:
test:
triggers:
- event: push
branches: ["*"]
jobs:
hello:
run: echo "hello from CI"
```
Then start:
```bash
./target/debug/local-runner \
--relay-node-id <NODE_ID_HEX> \
--yaml /tmp/test-ci.yml \
--work-dir /tmp/ci-work-test
```
You should see:
```
Iroh local ID: <RUNNER_ID>
Connecting to relay <NODE_ID>...
Connected to relay!
Local CI runner started
Webhook: via iroh relay
```
And in Terminal 1:
```
Runner connected: <RUNNER_ID>
```
**Terminal 3 — Send a fake webhook:**
```bash
curl -X POST http://localhost:9787 \
-H "Content-Type: application/json" \
-H "X-Forgejo-Event: push" \
-d '{
"ref": "refs/heads/main",
"after": "abc123def456789012345678901234567890abcd",
"repository": {
"name": "test-repo",
"owner": { "login": "testuser" }
}
}'
```
Expected output:
- **curl** returns: `ok`
- **Terminal 1** (relay):
```
webhook: abc123de main on testuser/test-repo
→ forwarded to runner
```
- **Terminal 2** (runner):
```
iroh: received webhook abc123de on main
```
The runner will also try to post status to Forgejo and log URL errors (since
we didn't pass `--forgejo-url`) — that's expected and confirms the event
reached the coordinator.
### 9.2 HMAC Verification Test
Start the relay with a secret:
```bash
./target/debug/ci-relay --port 9787 --secret mysecret
```
**Without signature — should be rejected (401):**
```bash
curl -v -X POST http://localhost:9787 \
-H "X-Forgejo-Event: push" \
-d '{"ref":"refs/heads/main","after":"abc123","repository":{"name":"r","owner":{"login":"u"}}}'
```
**With correct signature:**
```bash
# Compute HMAC-SHA256
BODY='{"ref":"refs/heads/main","after":"abc123","repository":{"name":"r","owner":{"login":"u"}}}'
SIG=$(echo -n "$BODY" | openssl dgst -sha256 -hmac "mysecret" | awk '{print $2}')
curl -X POST http://localhost:9787 \
-H "X-Forgejo-Event: push" \
-H "X-Forgejo-Signature: $SIG" \
-d "$BODY"
```
Should return `ok` and forward to the runner.
### 9.3 Runner Reconnection Test
1. Start relay and runner as in 9.1
2. Kill the runner (Ctrl-C in Terminal 2)
3. Restart the runner with the same `--relay-node-id`
4. The relay should log `Runner connected: <ID>` again
5. Send another webhook — it should flow through
### 9.4 No Runner Connected Test
1. Start the relay only (no runner)
2. Send a webhook via curl
3. Should get HTTP 502 and relay logs: `forward failed: no runner connected`
### 9.5 Full End-to-End with Forgejo
For a real deployment:
**On VPS:**
```bash
./ci-relay --port 8787 --secret <your-webhook-secret>
```
**On Thinkpad:**
```bash
./local-runner \
--relay-node-id <NODE_ID_FROM_VPS> \
--forgejo-url https://zachery.lol \
--forgejo-token <your-forgejo-api-token> \
--yaml .ci.yml \
--work-dir ~/ci-work \
--repo-url https://zachery.lol/<owner>/<repo>.git
```
**In Forgejo (repo settings → Webhooks):**
- Target URL: `http://localhost:8787`
- Secret: `<your-webhook-secret>`
- Events: Push, Create (tags), Pull Request
Push a commit and watch:
1. Relay logs the webhook and forwards it
2. Runner logs the received event and starts a pipeline
3. Forgejo shows commit status checks (pending → success/failure)
### 9.6 Inspecting Iroh Connectivity
Both binaries print their iroh Node ID on startup. To verify they're using
relay servers (expected when both are behind NAT or on different networks),
look for connection timing:
- **Fast connection (~1-3s)**: direct QUIC hole-punch succeeded
- **Slower connection (~5-10s)**: using iroh relay server fallback
If connection hangs indefinitely, check that both machines have internet
access and can reach iroh's relay servers (`https://relay.iroh.network`).
---
## 10. Known Gaps & Future Improvements
| Gap | Effort | Impact | Notes |
|-----|--------|--------|-------|
| Reconnection on runner side | Small | High | If the iroh connection drops mid-operation, the runner currently exits the receive loop. Should retry with backoff. |
| Multiple runner support | Medium | Medium | Relay caches one connection. For running CI on multiple machines, need a connection map keyed by runner identity. |
| Health check / heartbeat | Small | Medium | Neither side detects a silently dead connection until the next webhook. A periodic ping would surface stale connections faster. |
| Relay authentication | Small | Medium | Any iroh endpoint can connect to the relay. Should verify the runner's public key against an allowlist. |
| Binary size | Small | Low | ci-relay pulls in `swactor-ci` (which includes all CI types). A slimmer dependency with just `WebhookEvent` + `parse_webhook_json` would reduce the VPS binary. |
| Logging | Small | Low | Both binaries use `eprintln!`. Structured logging (tracing) would help in production. |
---
## Files Created/Modified
| Action | File | Purpose |
|--------|------|---------|
| Created | `crates/ci-relay/Cargo.toml` | Relay binary manifest |
| Created | `crates/ci-relay/src/main.rs` | Webhook relay: HTTP → iroh |
| Modified | `crates/local-runner/Cargo.toml` | Added iroh, tokio, serde_json deps |
| Modified | `crates/local-runner/src/main.rs` | Added `--relay-node-id` flag and iroh receiver |
| Modified | `Cargo.toml` (workspace root) | Added ci-relay to workspace members |