From 78b77726f406ef39ea0438deb684f6d8172c8b16 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Tue, 27 Oct 2026 01:38:33 -0400 Subject: [PATCH] feat(nats): ship the JetStream sink as its own binary The sink needs no HTTP server or web UI, so it should not pay for them in size or attack surface. Signed-off-by: Yordis Prieto --- AGENTS.md | 5 +++ Cargo.lock | 4 ++- Containerfile | 7 ++-- crates/walgit-cli/Cargo.toml | 1 - crates/walgit-cli/src/bin/walgit.rs | 2 +- crates/walgit-cli/src/lib.rs | 3 -- crates/walgit-config/src/lib.rs | 4 +-- crates/walgit-nats/Cargo.toml | 7 ++++ crates/walgit-nats/src/bin/walgit-nats.rs | 32 +++++++++++++++++++ crates/walgit-nats/src/env.rs | 4 +-- crates/walgit-nats/src/lib.rs | 6 ++-- crates/walgit-nats/src/stream_shape.rs | 2 +- crates/walgit-nats/tests/env.rs | 2 +- crates/walgit-nats/tests/jetstream.rs | 2 +- crates/walgit-server/tests/events_nats.rs | 2 +- docs/EVENTS.md | 27 +++++++++------- flake.nix | 6 ++-- justfile | 4 +-- tests/e2e.sh | 39 +++++++++++++---------- walgit.example.toml | 4 +-- 20 files changed, 108 insertions(+), 55 deletions(-) create mode 100644 crates/walgit-nats/src/bin/walgit-nats.rs diff --git a/AGENTS.md b/AGENTS.md index 1c6fc398..2208064d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -495,6 +495,11 @@ full cold-read/resource acceptance gates listed in `docs/spec/README.md`. crate shared with the webhook bridge, whose bucket cursor (`events/cursor.json`) is unchanged. Wake-ups arrive over NATS and the periodic sweep stays the backstop; the service API is the NATS micro protocol, no HTTP. +- **D51 (2026-10-27): The JetStream sink ships as its own `walgit-nats` binary.** This amends D9's one-binary + rule for this sink only: it carries no HTTP server or web UI and depends on neither `walgit-server` nor + `walgit-cli`. It shares config loading and telemetry with the other binaries through `walgit-telemetry`, and + reads the same `walgit.toml` as serve or a smaller file holding only the sections it uses (`docs/EVENTS.md`). + ## 5. Working rules - **No backwards compatibility (pre-1.0, banner at top):** change the shape and delete the old one in the same diff --git a/Cargo.lock b/Cargo.lock index dee6b506..89ef81df 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5926,7 +5926,6 @@ dependencies = [ "tracing", "walgit-config", "walgit-git", - "walgit-nats", "walgit-proto", "walgit-server", "walgit-store", @@ -6004,7 +6003,9 @@ dependencies = [ "anyhow", "async-nats", "bytes", + "clap", "futures", + "mimalloc", "nkeys", "percent-encoding", "rcgen", @@ -6021,6 +6022,7 @@ dependencies = [ "walgit-nats", "walgit-proto", "walgit-store", + "walgit-telemetry", "walgit-wal", ] diff --git a/Containerfile b/Containerfile index 806d8ac1..f1ce67ee 100644 --- a/Containerfile +++ b/Containerfile @@ -37,9 +37,10 @@ ARG WALGIT_BUILD_SHA=dev ENV WALGIT_BUILD_SHA=${WALGIT_BUILD_SHA} RUN --mount=type=cache,target=/usr/local/cargo/registry \ --mount=type=cache,target=/src/target \ - cargo build --release --locked -p walgit-cli \ + cargo build --release --locked -p walgit-cli -p walgit-nats \ && install -D target/release/walgit /out/bin/walgit \ - && install -D target/release/walgit-server /out/bin/walgit-server + && install -D target/release/walgit-server /out/bin/walgit-server \ + && install -D target/release/walgit-nats /out/bin/walgit-nats # ---- 3. runtime ----------------------------------------------------------------------------- # trixie ships git 2.47+: walgit wants >= 2.47 on the server (pack.writeReverseIndex, @@ -50,7 +51,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends git git-lfs cor && git --version RUN useradd --uid 1000 --create-home --shell /bin/sh walgit \ && mkdir -p /etc/walgit /var/lib/walgit && chown walgit:walgit /var/lib/walgit -COPY --from=build /out/bin/walgit /out/bin/walgit-server /usr/local/bin/ +COPY --from=build /out/bin/walgit /out/bin/walgit-server /out/bin/walgit-nats /usr/local/bin/ COPY walgit.example.toml /etc/walgit/walgit.toml ENV RUST_LOG=info,walgit=debug \ WALGIT_CONFIG=/etc/walgit/walgit.toml \ diff --git a/crates/walgit-cli/Cargo.toml b/crates/walgit-cli/Cargo.toml index f41c149b..3b2a1d62 100644 --- a/crates/walgit-cli/Cargo.toml +++ b/crates/walgit-cli/Cargo.toml @@ -25,7 +25,6 @@ walgit-telemetry.workspace = true walgit-git.workspace = true walgit-wal.workspace = true walgit-server.workspace = true -walgit-nats.workspace = true gix-hash.workspace = true anyhow.workspace = true tokio.workspace = true diff --git a/crates/walgit-cli/src/bin/walgit.rs b/crates/walgit-cli/src/bin/walgit.rs index a15663ad..9035dc22 100644 --- a/crates/walgit-cli/src/bin/walgit.rs +++ b/crates/walgit-cli/src/bin/walgit.rs @@ -1,4 +1,4 @@ -//! `walgit`: the full CLI (serve | compact | repo | wal | nats | synth | import | mirror | config). +//! `walgit`: the full CLI (serve | compact | repo | wal | synth | import | mirror | config). fn main() -> anyhow::Result<()> { walgit_cli::main() } diff --git a/crates/walgit-cli/src/lib.rs b/crates/walgit-cli/src/lib.rs index 8de3549c..08b288c5 100644 --- a/crates/walgit-cli/src/lib.rs +++ b/crates/walgit-cli/src/lib.rs @@ -108,8 +108,6 @@ enum Command { #[command(subcommand)] action: RepoAction, }, - /// Publish the WAL's ref events into NATS `JetStream` (`[events.nats]`, `docs/EVENTS.md`). - Nats, /// WAL inspection and rewind. Wal { #[command(subcommand)] @@ -441,7 +439,6 @@ async fn dispatch(command: Command, cfg: Config) -> Result<()> { } => compact::run(repo, all, once, base, &cfg).await, Command::Repo { action } => repo::run(action, &cfg).await, Command::Wal { action } => wal_cmd::run(action, &cfg).await, - Command::Nats => walgit_nats::run(cfg.clone()).await, Command::Mirror { from, to, diff --git a/crates/walgit-config/src/lib.rs b/crates/walgit-config/src/lib.rs index b4976071..c4743aec 100644 --- a/crates/walgit-config/src/lib.rs +++ b/crates/walgit-config/src/lib.rs @@ -637,7 +637,7 @@ pub struct EventsConfig { /// warns when a sweep finds unpublished entries. `0` = off. #[serde(with = "humantime_serde")] pub sweep_interval: Duration, - /// The `JetStream` publisher (`walgit nats`, `docs/EVENTS.md`). Read only by that subcommand. + /// The `JetStream` publisher (`walgit-nats`, `docs/EVENTS.md`). Read only by that binary. pub nats: NatsEventsConfig, } @@ -663,7 +663,7 @@ pub enum NatsStart { Head, } -/// `[events.nats]`: `walgit nats` publishes the bridge's `ref` events into `JetStream`. +/// `[events.nats]`: `walgit-nats` publishes the bridge's `ref` events into `JetStream`. /// The resume cursor is the last message on each repository's subject, so there is no cursor /// object in the bucket. #[derive(Debug, Clone, Serialize, Deserialize)] diff --git a/crates/walgit-nats/Cargo.toml b/crates/walgit-nats/Cargo.toml index bd0a1d8a..770537a3 100644 --- a/crates/walgit-nats/Cargo.toml +++ b/crates/walgit-nats/Cargo.toml @@ -4,12 +4,19 @@ version.workspace = true edition.workspace = true license.workspace = true +[[bin]] +name = "walgit-nats" +path = "src/bin/walgit-nats.rs" + [dependencies] walgit-events.workspace = true walgit-config.workspace = true walgit-store.workspace = true walgit-git.workspace = true walgit-wal.workspace = true +walgit-telemetry.workspace = true +clap.workspace = true +mimalloc.workspace = true async-nats.workspace = true nkeys.workspace = true percent-encoding.workspace = true diff --git a/crates/walgit-nats/src/bin/walgit-nats.rs b/crates/walgit-nats/src/bin/walgit-nats.rs new file mode 100644 index 00000000..234d5c8e --- /dev/null +++ b/crates/walgit-nats/src/bin/walgit-nats.rs @@ -0,0 +1,32 @@ +//! `walgit-nats --config walgit.toml`: the WAL's `ref` events published into NATS `JetStream` +//! (D50, D51), without the server or the web UI. `--config` may name the file `walgit serve` +//! reads or a smaller one holding only `[store]`, `[cache]`, `[wal]`, `[events]` and `[telemetry]`. + +use std::path::PathBuf; +use std::sync::Arc; + +use clap::Parser; + +#[global_allocator] +static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc; + +#[derive(Parser)] +#[command( + name = "walgit-nats", + version = walgit_telemetry::BUILD_SHA, + about = "Publish walgit's ref events into NATS JetStream" +)] +struct Cli { + /// Path to the configuration file. + #[arg(long, env = "WALGIT_CONFIG", default_value = "walgit.toml")] + config: PathBuf, +} + +fn main() -> anyhow::Result<()> { + let cli = Cli::parse(); + let cfg = walgit_telemetry::startup(&cli.config, "walgit-nats"); + tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build()? + .block_on(walgit_nats::run(Arc::new(cfg))) +} diff --git a/crates/walgit-nats/src/env.rs b/crates/walgit-nats/src/env.rs index 8f74ddba..0e9e734d 100644 --- a/crates/walgit-nats/src/env.rs +++ b/crates/walgit-nats/src/env.rs @@ -1,4 +1,4 @@ -//! How `walgit nats` reaches NATS, read from the same `NATS_*` environment variables the `nats` CLI +//! How `walgit-nats` reaches NATS, read from the same `NATS_*` environment variables the `nats` CLI //! reads and combined the way it combines them, so one shell environment drives both and secrets //! never sit in `walgit.toml`. @@ -107,7 +107,7 @@ impl NatsEnv { for unsupported in [ENV_NATS_CONTEXT, ENV_NATS_SOCKS_PROXY] { anyhow::ensure!( var(unsupported).is_none(), - "{unsupported} is set but walgit nats does not support it; unset it and use the other NATS_* variables" + "{unsupported} is set but walgit-nats does not support it; unset it and use the other NATS_* variables" ); } let home = var("HOME"); diff --git a/crates/walgit-nats/src/lib.rs b/crates/walgit-nats/src/lib.rs index 02c52f1b..268d2713 100644 --- a/crates/walgit-nats/src/lib.rs +++ b/crates/walgit-nats/src/lib.rs @@ -1,4 +1,4 @@ -//! `walgit nats`: the WAL's `ref` events (`docs/EVENTS.md`) published into NATS `JetStream`, with +//! `walgit-nats`: the WAL's `ref` events (`docs/EVENTS.md`) published into NATS `JetStream`, with //! the cursor in the stream, wake-ups over NATS and a `$SRV` service API instead of HTTP. //! The event model is shared with the webhook bridge through `walgit-events`. @@ -50,7 +50,7 @@ pub async fn open_stream( ) -> anyhow::Result { let stream = js.get_stream(&cfg.stream).await.with_context(|| { format!( - "stream {} (walgit nats never creates it; provision it as docs/EVENTS.md shows)", + "stream {} (walgit-nats never creates it; provision it as docs/EVENTS.md shows)", cfg.stream ) })?; @@ -98,7 +98,7 @@ pub fn publisher_config(cfg: &Config) -> Config { cfg } -/// `walgit nats`: publish until SIGTERM/ctrl-c, or fail when a worker stops. +/// `walgit-nats`: publish until SIGTERM/ctrl-c, or fail when a worker stops. pub async fn run(cfg: Arc) -> anyhow::Result<()> { let nats = &cfg.events.nats; let prefix = SubjectPrefix::new(&nats.subject_prefix)?; diff --git a/crates/walgit-nats/src/stream_shape.rs b/crates/walgit-nats/src/stream_shape.rs index 67fc5e7f..f541ee88 100644 --- a/crates/walgit-nats/src/stream_shape.rs +++ b/crates/walgit-nats/src/stream_shape.rs @@ -1,4 +1,4 @@ -//! What `walgit nats` needs from the stream it publishes into. The operator provisions the stream +//! What `walgit-nats` needs from the stream it publishes into. The operator provisions the stream //! (`docs/EVENTS.md`); walgit only checks it, so it never needs the permission to create one. use std::fmt; diff --git a/crates/walgit-nats/tests/env.rs b/crates/walgit-nats/tests/env.rs index 3f14ff4d..cb87702d 100644 --- a/crates/walgit-nats/tests/env.rs +++ b/crates/walgit-nats/tests/env.rs @@ -1,4 +1,4 @@ -//! `walgit nats` connects with whatever the `NATS_*` variables resolve to, against a real +//! `walgit-nats` connects with whatever the `NATS_*` variables resolve to, against a real //! `nats-server` that requires it. use std::time::Duration; diff --git a/crates/walgit-nats/tests/jetstream.rs b/crates/walgit-nats/tests/jetstream.rs index 69909601..a1de204e 100644 --- a/crates/walgit-nats/tests/jetstream.rs +++ b/crates/walgit-nats/tests/jetstream.rs @@ -96,7 +96,7 @@ impl Rig { ) } - /// What `walgit nats` runs: its own registry over the same store (it never sees the + /// What `walgit-nats` runs: its own registry over the same store (it never sees the /// writer's pushes except through the bucket) on the operator's `[wal]` settings, as /// `run` adapts them. Returns the registry's cache dir guard too. async fn operator_publisher( diff --git a/crates/walgit-server/tests/events_nats.rs b/crates/walgit-server/tests/events_nats.rs index 0e8ae32f..26c72d5f 100644 --- a/crates/walgit-server/tests/events_nats.rs +++ b/crates/walgit-server/tests/events_nats.rs @@ -1,7 +1,7 @@ // Test fixtures use panics to fail the test, including shared helper functions. #![allow(clippy::indexing_slicing, clippy::unwrap_used)] -//! Events (docs/EVENTS.md, `walgit nats`): real `git push`es against the harness server, published +//! Events (docs/EVENTS.md, `walgit-nats`): real `git push`es against the harness server, published //! into a real `nats-server -js` by a `walgit_nats` publisher over the same registry. Skips (passes) //! when `nats-server` is not on PATH. mod harness; diff --git a/docs/EVENTS.md b/docs/EVENTS.md index f2069bf3..69add7fa 100644 --- a/docs/EVENTS.md +++ b/docs/EVENTS.md @@ -115,15 +115,18 @@ webhook_secret = "…" # env: WALGIT__EVENTS__WEBHOOK_SECRET sweep_interval = "5m" ``` -## Delivery: NATS JetStream (`walgit nats`) +## Delivery: NATS JetStream (`walgit-nats`) -`walgit nats` is a second sink for the same `ref` events, a subcommand of the normal CLI -(`walgit nats --config walgit.toml`, `[events.nats]`) built on the `walgit-nats` crate, which depends on -`walgit-events` (the shared event model) and never on `walgit-server`. The webhook bridge is unchanged and the two can run side by side. +`walgit-nats` is a second sink for the same `ref` events, its own binary (`walgit-nats --config walgit.toml`, +`[events.nats]`) built from the `walgit-nats` crate. It depends on `walgit-events` (the shared event model) and +`walgit-telemetry` (config loading and logging), never on `walgit-server` or `walgit-cli`, so it carries no HTTP +server and no web UI. The webhook bridge is unchanged and the two can run side by side. Like the bridge it is a reader of the WAL, never a step of a push (D32). It keeps its local copies under `/.nats`, so it can share a `walgit.toml` and a host with `walgit serve` without touching serve's. +`--config` (or `WALGIT_CONFIG`) may name that same file, or a second, smaller one holding only `[store]`, +`[cache]`, `[wal]`, `[events]` and `[telemetry]`. -- **The operator provisions the stream; `walgit nats` never creates it**, so its credentials need no +- **The operator provisions the stream; `walgit-nats` never creates it**, so its credentials need no stream-admin permission, only stream info and message reads on the stream and publishes on its subjects. The stream it expects, for the default `stream` and `subject_prefix`: @@ -152,7 +155,7 @@ Like the bridge it is a reader of the WAL, never a step of a push (D32). It keep shapes as `POST /_events/notify` (S3 event JSON, a GCS envelope, `{"key": ...}`, `{"repo": ...}`), and `events.sweep_interval` is the backstop with the same warning when it finds unpublished entries. Gap detection matches the bridge: a log gap is logged and the entries that are readable are published. -- **Service API.** `walgit nats` registers a NATS micro service `walgit-events` (`$SRV.PING|INFO|STATS`, built in) +- **Service API.** `walgit-nats` registers a NATS micro service `walgit-events` (`$SRV.PING|INFO|STATS`, built in) with one endpoint, `.events.catch-up`. Request `{"repo": "o/r"}`, reply the catch-up report (`repo`, `from_seq`, `head_seq`, `emitted`, `conflict`); an unknown repository is a 404 service error. Gap, conflict and sweep counters are in the stats response. There is no HTTP listener. @@ -181,19 +184,19 @@ Like the bridge it is a reader of the WAL, never a step of a push (D32). It keep The webhook bridge has the same ambiguity from the other side: its cursor lives under the repository's prefix and is deleted with it, so the new incarnation is published from seq 1 and the dedup key `(repo, seq, ref_name)` collides with the old incarnation's events. -- **Config adaptation.** `walgit nats` always revalidates the manifest (`wal.freshness_ttl` is forced to zero) and +- **Config adaptation.** `walgit-nats` always revalidates the manifest (`wal.freshness_ttl` is forced to zero) and never prefetches packs (`wal.prefetch_packs` is forced off), whatever `[wal]` says, because it never applies a push itself and only reads manifest and log (`walgit_nats::publisher_config`). - **Cost.** Every wake-up for a repository reads the last message on its subject (one JetStream request) before deciding there is nothing to do, so a wiped stream is noticed without a restart. Each behaviour above has a test against a real `nats-server -js` (`crates/walgit-nats/tests/jetstream.rs`, and -`crates/walgit-server/tests/events_nats.rs` with real git pushes); `tests/e2e.sh` runs `walgit nats` as a process. +`crates/walgit-server/tests/events_nats.rs` with real git pushes); `tests/e2e.sh` runs `walgit-nats` as a process. | Behaviour | Test | |---|---| | Push becomes one message per ref with the bridge's event shape | `pushes_become_events_with_the_right_shape`, `an_atomic_multi_ref_push_is_one_message_per_ref_in_order` | -| The webhook bridge and `walgit nats` emit the same events | `the_webhook_bridge_and_the_jetstream_publisher_agree` | +| The webhook bridge and `walgit-nats` emit the same events | `the_webhook_bridge_and_the_jetstream_publisher_agree` | | Names are encoded in the subject | `a_dotted_repository_name_is_encoded_in_the_subject` | | The stream is the cursor; a restart resumes from it | `stream_is_the_cursor`, `a_restarted_publisher_resumes_from_the_stream` | | A crash mid-entry finishes that entry only | `a_crash_mid_entry_finishes_that_entry_only` | @@ -231,8 +234,8 @@ urls = ["nats://nats.internal:4222"] start = "oldest" ``` -Credentials are never in `walgit.toml`. `walgit nats` reads the variables the `nats` CLI reads and combines -them the way it does, so a shell where `nats` works runs `walgit nats` against the same server as the same +Credentials are never in `walgit.toml`. `walgit-nats` reads the variables the `nats` CLI reads and combines +them the way it does, so a shell where `nats` works runs `walgit-nats` against the same server as the same user. An empty variable counts as unset. | Variable | Effect | @@ -247,7 +250,7 @@ user. An empty variable counts as unset. The identity is the first of user and password, `NATS_CREDS`, `NATS_NKEY`, then `NATS_JWT`, the order `nats` uses. A server URL written `user:password@host` or `token@host` replaces the user, password and token, as in -`nats`; servers that carry credentials must carry the same ones. `NATS_CONTEXT` and `NATS_SOCKS_PROXY` stop `walgit nats` at startup: it reads no context files and has no +`nats`; servers that carry credentials must carry the same ones. `NATS_CONTEXT` and `NATS_SOCKS_PROXY` stop `walgit-nats` at startup: it reads no context files and has no SOCKS dialer, and ignoring either would connect differently from `nats`. `NATS_TIMEOUT` and `NATS_COLOR` shape the CLI's own requests and output and have no effect here. Only the methods and the URLs without their credentials are logged, never a secret. diff --git a/flake.nix b/flake.nix index c3a1e7e2..470d981b 100644 --- a/flake.nix +++ b/flake.nix @@ -2,7 +2,7 @@ # walgit — a git server that is one binary in front of an object store. # # nix run github:/walgit -- --config walgit.toml # run it - # nix build .#walgit # result/bin/{walgit,walgit-server} + # nix build .#walgit # result/bin/{walgit,walgit-server,walgit-nats} # nix build .#image && podman load < result # an OCI image (same bytes as the Containerfile) # nix develop # rust, protobuf, pnpm, git, just, rustfs-compatible tools # @@ -86,7 +86,7 @@ src = rustSrc; strictDeps = true; nativeBuildInputs = with pkgs; [ protobuf pkg-config cmake perl python3 ]; - cargoExtraArgs = "-p walgit-cli --locked"; + cargoExtraArgs = "-p walgit-cli -p walgit-nats --locked"; doCheck = false; }; cargoArtifacts = craneLib.buildDepsOnly commonArgs; @@ -101,7 +101,7 @@ # `walgit serve` shells out to git and GNU sort/comm for bounded pack planning. nativeBuildInputs = commonArgs.nativeBuildInputs ++ [ pkgs.makeWrapper ]; postInstall = '' - for b in walgit walgit-server; do + for b in walgit walgit-server walgit-nats; do wrapProgram "$out/bin/$b" --prefix PATH : ${lib.makeBinPath [ pkgs.git pkgs.git-lfs pkgs.coreutils ]} done ''; diff --git a/justfile b/justfile index 64b4e580..6c53b328 100644 --- a/justfile +++ b/justfile @@ -149,8 +149,8 @@ sim: # Standalone CLI/server against memory; add WALGIT_TEST_S3_ENDPOINT for the local rig. smoke: - {{t15}} cargo build -p walgit-cli - WALGIT="$(realpath "${CARGO_TARGET_DIR:-target}/debug/walgit")" {{t15}} tests/e2e.sh + {{t15}} cargo build -p walgit-cli -p walgit-nats + WALGIT="$(realpath "${CARGO_TARGET_DIR:-target}/debug/walgit")" WALGIT_NATS="$(realpath "${CARGO_TARGET_DIR:-target}/debug/walgit-nats")" {{t15}} tests/e2e.sh # Slow tier: #[ignore]d benches/soaks (20k-ref push, 466k-ref render, ...). test-slow: diff --git a/tests/e2e.sh b/tests/e2e.sh index 2e20f699..81bcf4f6 100755 --- a/tests/e2e.sh +++ b/tests/e2e.sh @@ -14,8 +14,9 @@ # tests/e2e.sh # WALGIT_TEST_S3_ENDPOINT=http://localhost:9000 tests/e2e.sh # WALGIT=/absolute/path/to/freshly/built/walgit tests/e2e.sh +# WALGIT_NATS=/absolute/path/to/freshly/built/walgit-nats (phase 3; default target/release/walgit-nats) # -# Phase 3 (S3/rustfs plus `nats-server` and the `nats` CLI on PATH): `walgit nats` as a process next to +# Phase 3 (S3/rustfs plus `nats-server` and the `nats` CLI on PATH): `walgit-nats` as a process next to # `walgit serve`; real pushes become JetStream events; a restart republishes nothing; wake-ups work. # # Requires: cargo, git, curl, python3 (for random port), ripgrep (S3 maintenance receipt). @@ -25,6 +26,7 @@ set -euo pipefail SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" ROOT="$(cd "$SCRIPT_DIR/.." && pwd)" WALGIT="${WALGIT:-}" +WALGIT_NATS="${WALGIT_NATS:-}" # --- helpers ----------------------------------------------------------------- @@ -71,9 +73,12 @@ step "e2e: walgit end-to-end test" if [[ -z "$WALGIT" ]]; then step "building walgit (release)..." - (cd "$ROOT" && cargo build -p walgit-cli --release) + (cd "$ROOT" && cargo build -p walgit-cli -p walgit-nats --release) WALGIT="$ROOT/target/release/walgit" fi +if [[ -z "$WALGIT_NATS" ]]; then + WALGIT_NATS="$ROOT/target/release/walgit-nats" +fi [[ -x "$WALGIT" ]] || fail "WALGIT must name an executable built from this checkout" TMP="$(mktemp -d)" @@ -348,17 +353,19 @@ EOF kill "$SERVER_A_PID" "$SERVER_B_PID" 2>/dev/null || true fi -# --- phase 3: `walgit nats` as processes (optional, local only) -------------- +# --- phase 3: `walgit-nats` as processes (optional, local only) -------------- -# Real processes end to end: nats-server -js, `walgit serve` and `walgit nats` sharing the rustfs -# bucket, real git pushes, assertions through the `nats` CLI. The first walgit nats is woken by the +# Real processes end to end: nats-server -js, `walgit serve` and `walgit-nats` sharing the rustfs +# bucket, real git pushes, assertions through the `nats` CLI. The first walgit-nats is woken by the # sweep (and one notification); after a kill and restart the sweep is off, so a push there can only # arrive through the notify subject. if [[ -z "${WALGIT_E2E_BASE_URL:-}" && -n "${WALGIT_TEST_S3_ENDPOINT:-}" ]]; then if ! command -v nats-server >/dev/null 2>&1 || ! command -v nats >/dev/null 2>&1; then - step "phase 3: walgit nats (skipped: nats-server and the nats CLI must be on PATH)" + step "phase 3: walgit-nats (skipped: nats-server and the nats CLI must be on PATH)" + elif [[ ! -x "$WALGIT_NATS" ]]; then + fail "phase 3 needs the walgit-nats binary: build it (cargo build -p walgit-nats --release) or set WALGIT_NATS" else - step "phase 3: walgit nats against nats-server -js ($WALGIT_TEST_S3_ENDPOINT)" + step "phase 3: walgit-nats against nats-server -js ($WALGIT_TEST_S3_ENDPOINT)" export AWS_ACCESS_KEY_ID="${AWS_ACCESS_KEY_ID:-walgit-dev}" AWS_SECRET_ACCESS_KEY="${AWS_SECRET_ACCESS_KEY:-walgit-dev-secret}" NATS_PORT="$(rand_port)" @@ -436,15 +443,15 @@ print(m["subject"], body["action"], body["ref_type"], body["ref_name"], body["_w local log="$1"; shift : > "$log" RUST_LOG=info env "WALGIT__CACHE__DIR=$TMP/cache-nats-pub" "$@" \ - "$WALGIT" --config "$TMP/cfg-nats.toml" nats > "$log" 2>&1 & + "$WALGIT_NATS" --config "$TMP/cfg-nats.toml" > "$log" 2>&1 & NATS_PUB_PID=$! PIDS+=("$NATS_PUB_PID") for ((i=0; i<30; i++)); do - kill -0 "$NATS_PUB_PID" 2>/dev/null || { cat "$log"; fail "walgit nats exited"; } + kill -0 "$NATS_PUB_PID" 2>/dev/null || { cat "$log"; fail "walgit-nats exited"; } grep -q "service ready" "$log" && return 0 sleep 1 done - cat "$log"; fail "walgit nats did not become ready" + cat "$log"; fail "walgit-nats did not become ready" } RUST_LOG=info env "WALGIT__CACHE__DIR=$TMP/cache-nats-serve" \ @@ -453,7 +460,7 @@ print(m["subject"], body["action"], body["ref_type"], body["ref_name"], body["_w PIDS+=("$SERVER_N_PID") wait_http "$BASE_N/healthz" 15 || fail "walgit serve (nats phase) not ready" start_walgit_nats "$TMP/walgit-nats-1.log" - pass "walgit serve on $PORT_N and walgit nats ready" + pass "walgit serve on $PORT_N and walgit-nats ready" curl_auth -sf -X PUT "$BASE_N/$REPO3.git" -o /dev/null || fail "create $REPO3" git clone -q "$SYNTH_DIR" "$TMP/work-nats" @@ -480,7 +487,7 @@ print(m["subject"], body["action"], body["ref_type"], body["ref_name"], body["_w [[ "$(echo "$EVENTS" | sed -n 4p)" == "$SUBJECT3 delete tag refs/tags/v1.0.0 4 $REPO3:4:refs/tags/v1.0.0" ]] || fail "event 4" pass "create, update, tag create and delete on $SUBJECT3, in WAL order" - step "nats: kill walgit nats, restart (sweep off): no duplicates, a new push arrives by notification" + step "nats: kill walgit-nats, restart (sweep off): no duplicates, a new push arrives by notification" kill "$NATS_PUB_PID" 2>/dev/null || true wait "$NATS_PUB_PID" 2>/dev/null || true start_walgit_nats "$TMP/walgit-nats-2.log" WALGIT__EVENTS__SWEEP_INTERVAL=0s @@ -508,20 +515,20 @@ r = json.load(sys.stdin) assert r["repo"] == sys.argv[1] and r["head_seq"] == 5 and r["emitted"] == 0, r' "$REPO3" || fail "catch-up report" pass "catch-up reports head 5, nothing to emit" - step "nats: SIGTERM makes walgit nats exit 0 promptly" + step "nats: SIGTERM makes walgit-nats exit 0 promptly" kill -TERM "$NATS_PUB_PID" for ((i=0; i<50; i++)); do kill -0 "$NATS_PUB_PID" 2>/dev/null || break sleep 0.1 done if kill -0 "$NATS_PUB_PID" 2>/dev/null; then - cat "$TMP/walgit-nats-2.log"; fail "walgit nats still running 5s after SIGTERM" + cat "$TMP/walgit-nats-2.log"; fail "walgit-nats still running 5s after SIGTERM" fi NATS_RC=0 wait "$NATS_PUB_PID" 2>/dev/null || NATS_RC=$? - [[ "$NATS_RC" -eq 0 ]] || { cat "$TMP/walgit-nats-2.log"; fail "walgit nats exited $NATS_RC after SIGTERM"; } + [[ "$NATS_RC" -eq 0 ]] || { cat "$TMP/walgit-nats-2.log"; fail "walgit-nats exited $NATS_RC after SIGTERM"; } grep -q "shutting down" "$TMP/walgit-nats-2.log" || fail "no shutdown line after SIGTERM" - pass "SIGTERM: walgit nats exited 0 within 5s" + pass "SIGTERM: walgit-nats exited 0 within 5s" kill "$SERVER_N_PID" 2>/dev/null || true curl_auth -s -X DELETE "$BASE_N/$REPO3.git" -o /dev/null || true diff --git a/walgit.example.toml b/walgit.example.toml index 9f7e8c98..e89854f6 100644 --- a/walgit.example.toml +++ b/walgit.example.toml @@ -203,7 +203,7 @@ lock_wait_warn = "1s" # WARN "lock wait" (lock, repo, wait_ms, reque # webhook_secret = "" # X-Walgit-Signature: sha256=; unset = unsigned # sweep_interval = "5m" # backstop + health check (warns when a sweep finds unpublished entries); 0 = off -[events.nats] # docs/EVENTS.md: read only by `walgit nats --config walgit.toml` +[events.nats] # docs/EVENTS.md: read only by `walgit-nats --config walgit.toml` # its local copies live under /.nats, apart from walgit serve's on the same config # urls = ["nats://127.0.0.1:4222"] # the client connects to the first reachable server # the connection reads the `nats` CLI's variables the way it does (empty = unset): NATS_URL replaces urls; @@ -212,7 +212,7 @@ lock_wait_warn = "1s" # WARN "lock wait" (lock, repo, wait_ms, reque # user:password@ or token@ in a server URL replaces the user, password and token; # NATS_CERT + NATS_KEY, NATS_CA for TLS; NATS_CONTEXT and NATS_SOCKS_PROXY are refused; secrets are never logged # stream = "WALGIT_EVENTS" # the JetStream stream holding the events (subjects ".ref.>"); it is the resume cursor -# walgit nats never creates the stream; provision it (docs/EVENTS.md), so it needs no stream-admin permission; +# walgit-nats never creates the stream; provision it (docs/EVENTS.md), so it needs no stream-admin permission; # it fails at boot unless the stream is file storage, limits retention, captures the subjects and dedups # subject_prefix = "walgit" # events go to ".ref.."; metacharacters in names are %XX encoded # notify_subject = "walgit.notify" # core subject bucket notifications are published to (S3 event JSON, {"key"} or {"repo"})