From 8e9dc80f59185c15b00fdd3a35b28fa3c4641d3a Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 26 Oct 2026 22:06:04 -0400 Subject: [PATCH 1/4] feat(nats): wake the publisher over NATS so it needs no HTTP listener Signed-off-by: Yordis Prieto --- Cargo.lock | 1 + crates/walgit-nats/Cargo.toml | 1 + crates/walgit-nats/src/lib.rs | 6 +- crates/walgit-nats/src/service.rs | 76 +++++++ crates/walgit-nats/src/wake.rs | 135 +++++++++++ crates/walgit-nats/tests/jetstream.rs | 316 +++++++++++++++++++++++++- 6 files changed, 530 insertions(+), 5 deletions(-) create mode 100644 crates/walgit-nats/src/service.rs create mode 100644 crates/walgit-nats/src/wake.rs diff --git a/Cargo.lock b/Cargo.lock index 7785118..ec4a9aa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6004,6 +6004,7 @@ dependencies = [ "anyhow", "async-nats", "bytes", + "futures", "nkeys", "percent-encoding", "rcgen", diff --git a/crates/walgit-nats/Cargo.toml b/crates/walgit-nats/Cargo.toml index 1f4dde2..bd0a1d8 100644 --- a/crates/walgit-nats/Cargo.toml +++ b/crates/walgit-nats/Cargo.toml @@ -16,6 +16,7 @@ percent-encoding.workspace = true anyhow.workspace = true thiserror.workspace = true bytes.workspace = true +futures.workspace = true tokio.workspace = true tracing.workspace = true serde.workspace = true diff --git a/crates/walgit-nats/src/lib.rs b/crates/walgit-nats/src/lib.rs index 1ec5d82..e3b44bf 100644 --- a/crates/walgit-nats/src/lib.rs +++ b/crates/walgit-nats/src/lib.rs @@ -1,14 +1,16 @@ -//! `walgit nats`: the WAL's `ref` events (`docs/EVENTS.md`) published into NATS `JetStream`, -//! with the cursor in the stream. +//! `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`. pub mod env; pub mod message; pub mod publisher; +pub mod service; pub mod stream_shape; pub mod subject; #[cfg(feature = "testing")] pub mod testing; +pub mod wake; use anyhow::Context; use async_nats::jetstream::{self, stream}; diff --git a/crates/walgit-nats/src/service.rs b/crates/walgit-nats/src/service.rs new file mode 100644 index 0000000..5e311e4 --- /dev/null +++ b/crates/walgit-nats/src/service.rs @@ -0,0 +1,76 @@ +//! The NATS service API (`$SRV` micro protocol) in place of an HTTP listener: +//! `.events.catch-up` takes `{"repo":"owner/name"}` and answers with the catch-up report; +//! `$SRV.PING|INFO|STATS` come with the protocol (stats carry this publisher's counters). + +use std::str::FromStr; +use std::sync::Arc; + +use async_nats::service::ServiceExt; +use async_nats::service::error::Error as ServiceError; +use bytes::Bytes; +use futures::StreamExt; +use walgit_git::RepoId; + +use crate::publisher::Publisher; +use crate::subject::SubjectPrefix; + +pub const SERVICE_NAME: &str = "walgit-events"; + +#[derive(serde::Deserialize)] +struct CatchUpRequest { + repo: String, +} + +fn failure(code: usize, status: &impl ToString) -> ServiceError { + ServiceError { + status: status.to_string(), + code, + } +} + +async fn handle(publisher: &Publisher, payload: &[u8]) -> Result { + let req: CatchUpRequest = + serde_json::from_slice(payload).map_err(|e| failure(400, &format!("request: {e}")))?; + let id = RepoId::from_str(&req.repo).map_err(|e| failure(400, &e))?; + match publisher.catch_up_if_present(&id).await { + Ok(Some(report)) => serde_json::to_vec(&report) + .map(Bytes::from) + .map_err(|e| failure(500, &e)), + Ok(None) => Err(failure(404, &format!("no repository {id}"))), + Err(e) => Err(failure(500, &format!("{e:#}"))), + } +} + +/// Serve until the endpoint closes. Instances share requests through the service queue group. +pub async fn serve( + client: async_nats::Client, + prefix: &SubjectPrefix, + queue_group: &str, + publisher: Arc, +) -> anyhow::Result<()> { + let stats = publisher.clone(); + let service = client + .service_builder() + .description("walgit ref events into JetStream") + .queue_group(queue_group) + .stats_handler(move |_endpoint, _stats| stats.counters.snapshot()) + .start(SERVICE_NAME, env!("CARGO_PKG_VERSION")) + .await + .map_err(|e| anyhow::anyhow!("starting the {SERVICE_NAME} service: {e}"))?; + let mut endpoint = service + .group(prefix.child("events")) + .endpoint("catch-up") + .await + .map_err(|e| anyhow::anyhow!("registering catch-up: {e}"))?; + tracing::info!(subject = %prefix.child("events.catch-up"), "service ready"); + while let Some(request) = endpoint.next().await { + let publisher = publisher.clone(); + tokio::spawn(async move { + let answer = handle(&publisher, &request.message.payload).await; + if let Err(e) = request.respond(answer).await { + tracing::warn!(error = %e, "catch-up reply failed"); + } + }); + } + Ok(()) +} diff --git a/crates/walgit-nats/src/wake.rs b/crates/walgit-nats/src/wake.rs new file mode 100644 index 0000000..d10dc64 --- /dev/null +++ b/crates/walgit-nats/src/wake.rs @@ -0,0 +1,135 @@ +//! Wake-ups: a bucket notification on the notify subject, and the periodic sweep. +//! Both only ever call `catch_up`, so both are idempotent. + +use std::collections::BTreeSet; +use std::sync::Arc; +use std::time::Duration; + +use futures::StreamExt; +use walgit_events::{manifest_repo, notified_keys}; +use walgit_git::RepoId; + +use crate::publisher::Publisher; + +/// The repositories a notification body names: an S3 event (`Records[].s3.object.key`, what +/// `MinIO` and rustfs publish to NATS natively), a GCS envelope, `{"key": "..."}` or +/// `{"repo": "owner/name"}`. Anything else names nothing. +pub fn notified_repos(payload: &[u8], store_prefix: &str) -> BTreeSet { + let Ok(value) = serde_json::from_slice::(payload) else { + return BTreeSet::new(); + }; + notified_keys(&value, store_prefix) + .iter() + .filter_map(|k| manifest_repo(k, store_prefix)) + .collect() +} + +/// Subscribe to `subject` in `queue_group` (instances share the notifications) until the +/// subscription ends. A failed catch-up is logged, not retried: core NATS does not redeliver, and +/// the sweep is the backstop. +pub async fn notify_loop( + client: async_nats::Client, + subject: String, + queue_group: String, + store_prefix: String, + publisher: Arc, +) -> anyhow::Result<()> { + let mut sub = client.queue_subscribe(subject.clone(), queue_group).await?; + tracing::info!(%subject, "listening for bucket notifications"); + while let Some(msg) = sub.next().await { + for id in notified_repos(&msg.payload, &store_prefix) { + if let Err(e) = publisher.catch_up_if_present(&id).await { + tracing::warn!(repo = %id, error = %format!("{e:#}"), "notified catch-up failed; the sweep retries"); + } + } + } + Ok(()) +} + +/// One sweep at startup (quiet: it legitimately finds work), then every `every` (`0` = never again). +pub async fn sweep_loop(publisher: Arc, every: Duration) { + publisher.sweep(false).await; + if every.is_zero() { + return; + } + let mut tick = tokio::time::interval(every); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + tick.tick().await; + loop { + tick.tick().await; + publisher.sweep(true).await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn every_notification_shape_names_the_repo() { + let key = "pre/repos/acme/mono/manifest.pb"; + for body in [ + serde_json::json!({"Records": [{"eventName": "s3:ObjectCreated:Put", "s3": {"object": {"key": key}}}]}), + serde_json::json!({"key": key}), + serde_json::json!({"repo": "acme/mono"}), + ] { + let repos = notified_repos(body.to_string().as_bytes(), "pre/"); + let names: Vec<_> = repos.iter().map(ToString::to_string).collect(); + assert_eq!(names, ["acme/mono"], "{body}"); + } + assert!(notified_repos(b"not json", "pre/").is_empty()); + let other = serde_json::json!({"key": "pre/repos/acme/mono/wal/x.pack"}); + assert!(notified_repos(other.to_string().as_bytes(), "pre/").is_empty()); + } + + fn names(body: &serde_json::Value, prefix: &str) -> Vec { + notified_repos(body.to_string().as_bytes(), prefix) + .iter() + .map(ToString::to_string) + .collect() + } + + #[test] + fn an_s3_event_with_several_records_names_each_repo_once() { + let rec = |name: &str, key: &str| { + serde_json::json!({"eventName": name, "s3": {"object": {"key": key}}}) + }; + let body = serde_json::json!({"Records": [ + rec("s3:ObjectCreated:Put", "repos/a/one/manifest.pb"), + rec("s3:ObjectCreated:CompleteMultipartUpload", "repos/b/two/manifest.pb"), + rec("ObjectCreated:Put", "repos/c/three/manifest.pb"), + rec("s3:ObjectCreated:Put", "repos/a/one/manifest.pb"), + rec("s3:ObjectRemoved:Delete", "repos/d/gone/manifest.pb"), + rec("s3:ObjectCreated:Put", "repos/e/four/log/1.pb"), + rec("s3:ObjectCreated:Put", "repos/a%2Dx/five/manifest.pb"), + ]}); + assert_eq!(names(&body, ""), ["a/one", "a-x/five", "b/two", "c/three"]); + } + + #[test] + fn a_gcs_envelope_counts_only_finalize_events() { + let env = |event: &str, id: &str| { + serde_json::json!({"message": {"attributes": {"eventType": event, "objectId": id}}}) + }; + assert_eq!( + names(&env("OBJECT_FINALIZE", "repos/a/one/manifest.pb"), ""), + ["a/one"] + ); + for event in ["OBJECT_DELETE", "OBJECT_ARCHIVE", "OBJECT_METADATA_UPDATE", ""] { + assert!(names(&env(event, "repos/a/one/manifest.pb"), "").is_empty(), "{event}"); + } + } + + #[test] + fn the_store_prefix_scopes_what_is_notified() { + let key = |k: &str| serde_json::json!({"key": k}); + assert_eq!(names(&key("pre/fix/repos/a/one/manifest.pb"), "pre/fix/"), ["a/one"]); + assert!(names(&key("repos/a/one/manifest.pb"), "pre/fix/").is_empty()); + assert!(names(&key("other/repos/a/one/manifest.pb"), "pre/fix/").is_empty()); + assert_eq!(names(&serde_json::json!({"repo": "a/one"}), "pre/fix/"), ["a/one"]); + let s3 = serde_json::json!({"Records": [{"eventName": "s3:ObjectCreated:Put", + "s3": {"object": {"key": "pre/fix/repos/a/one/manifest.pb"}}}]}); + assert_eq!(names(&s3, "pre/fix/"), ["a/one"]); + assert!(names(&s3, "").is_empty()); + } +} diff --git a/crates/walgit-nats/tests/jetstream.rs b/crates/walgit-nats/tests/jetstream.rs index 261daa1..61afacd 100644 --- a/crates/walgit-nats/tests/jetstream.rs +++ b/crates/walgit-nats/tests/jetstream.rs @@ -2,14 +2,15 @@ #![allow(clippy::indexing_slicing, clippy::unwrap_used, clippy::expect_used)] //! The `JetStream` publisher against a real `nats-server -js` and the in-memory walgit store: -//! the stream is the cursor (resume without republishing) and crash-mid-entry completion. -//! Skips (passes) when `nats-server` is not on PATH. +//! the stream is the cursor (resume without republishing), crash-mid-entry completion, the +//! wake-ups and the service API. Skips (passes) when `nats-server` is not on PATH. use std::collections::HashMap; use std::sync::Arc; use std::time::{Duration, Instant}; use async_nats::jetstream; +use futures::StreamExt; use walgit_config::{Config, NatsStart}; use walgit_events::refs_from_entries; use walgit_git::{ObjectFormat, RepoId}; @@ -206,6 +207,32 @@ impl Rig { } } + fn spawn_notify(&self, publisher: &Arc) -> tokio::task::JoinHandle> { + let nats_cfg = &self.cfg.events.nats; + tokio::spawn(walgit_nats::wake::notify_loop( + self.client.clone(), + nats_cfg.notify_subject.clone(), + nats_cfg.queue_group.clone(), + self.cfg.store_prefix(), + publisher.clone(), + )) + } + + fn spawn_service(&self, publisher: &Arc) -> tokio::task::JoinHandle> { + let (client, prefix, publisher) = (self.client.clone(), self.prefix.clone(), publisher.clone()); + let queue_group = self.cfg.events.nats.queue_group.clone(); + tokio::spawn(async move { + walgit_nats::service::serve(client, &prefix, &queue_group, publisher).await + }) + } + + async fn notify(&self, body: &str) { + self.client + .publish(self.cfg.events.nats.notify_subject.clone(), body.to_string().into()) + .await + .unwrap(); + } + async fn stream(&self) -> jetstream::stream::Stream { self.js .get_stream(&self.cfg.events.nats.stream) @@ -217,6 +244,13 @@ impl Rig { self.stream().await.get_info().await.unwrap().state.messages } + async fn wait_for_messages(&self, n: u64) { + let t0 = Instant::now(); + while self.message_count().await < n { + assert!(t0.elapsed() < Duration::from_secs(10), "timed out waiting for {n} messages"); + tokio::time::sleep(Duration::from_millis(50)).await; + } + } } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] @@ -370,8 +404,86 @@ async fn a_second_publisher_that_moved_first_stops_this_one() { assert_eq!(rig.message_count().await, 1); } +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn notifications_and_the_service_api_wake_the_publisher() { + let Some(nats) = start_nats().await else { return }; + let rig = Rig::new(&nats, NatsStart::Oldest).await; + let id = RepoId::new("t", "r").unwrap(); + rig.registry.create(&id, ObjectFormat::Sha1).await.unwrap(); + let publisher = rig.publisher(NatsStart::Oldest).await; + + let nats_cfg = &rig.cfg.events.nats; + let notify = tokio::spawn(walgit_nats::wake::notify_loop( + rig.client.clone(), + nats_cfg.notify_subject.clone(), + nats_cfg.queue_group.clone(), + rig.cfg.store_prefix(), + publisher.clone(), + )); + let service = tokio::spawn({ + let (client, prefix, publisher) = + (rig.client.clone(), rig.prefix.clone(), publisher.clone()); + let queue_group = nats_cfg.queue_group.clone(); + async move { walgit_nats::service::serve(client, &prefix, &queue_group, publisher).await } + }); + tokio::time::sleep(Duration::from_millis(500)).await; + + // An S3 bucket notification, as the object store publishes it. + rig.push(&id, &[("refs/heads/main", ZERO, A)]).await; + let s3 = serde_json::json!({"Records": [{"eventName": "s3:ObjectCreated:Put", + "s3": {"object": {"key": "repos/t/r/manifest.pb"}}}]}); + rig.client + .publish(nats_cfg.notify_subject.clone(), s3.to_string().into()) + .await + .unwrap(); + rig.wait_for_messages(1).await; + + // {"repo": ...} through the same subject. + rig.push(&id, &[("refs/heads/main", A, B)]).await; + rig.client + .publish(nats_cfg.notify_subject.clone(), r#"{"repo":"t/r"}"#.into()) + .await + .unwrap(); + rig.wait_for_messages(2).await; + + // The service endpoint. + rig.push(&id, &[("refs/heads/main", B, C)]).await; + let reply = rig + .client + .request("walgit.events.catch-up", r#"{"repo":"t/r"}"#.into()) + .await + .unwrap(); + let report: serde_json::Value = serde_json::from_slice(&reply.payload).unwrap(); + assert_eq!((report["repo"].as_str(), report["emitted"].as_u64()), (Some("t/r"), Some(1))); + assert_eq!(rig.message_count().await, 3); + + let missing = rig + .client + .request("walgit.events.catch-up", r#"{"repo":"t/none"}"#.into()) + .await + .unwrap(); + let code = missing.headers.as_ref().and_then(|h| h.get("Nats-Service-Error-Code")); + assert_eq!(code.map(|c| c.as_str().to_string()), Some("404".into())); + + // The micro protocol's built-ins answer too. + let mut pings = rig.client.subscribe("inbox.ping").await.unwrap(); + rig.client + .publish_with_reply("$SRV.PING.walgit-events", "inbox.ping", "".into()) + .await + .unwrap(); + let pong = tokio::time::timeout(Duration::from_secs(5), pings.next()) + .await + .unwrap() + .unwrap(); + let pong: serde_json::Value = serde_json::from_slice(&pong.payload).unwrap(); + assert_eq!(pong["name"], "walgit-events"); + + notify.abort(); + service.abort(); +} + // --------------------------------------------------------------------------------------------- -// Dedup, races, wipes, gaps and outages. +// Dedup, races, wipes, gaps, outages and the wake-ups that must be ignored. // --------------------------------------------------------------------------------------------- async fn repo_with_three_events(rig: &Rig, owner: &str, name: &str) -> RepoId { @@ -642,6 +754,48 @@ async fn an_outage_mid_catch_up_loses_and_duplicates_nothing() { assert_eq!(ids.len(), REFS); } +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn notifications_that_name_nothing_are_ignored_and_the_loop_survives() { + let Some(nats) = start_nats().await else { return }; + let rig = Rig::new(&nats, NatsStart::Oldest).await; + let id = RepoId::new("t", "r").unwrap(); + rig.registry.create(&id, ObjectFormat::Sha1).await.unwrap(); + rig.push(&id, &[("refs/heads/main", ZERO, A)]).await; + let publisher = rig.publisher(NatsStart::Oldest).await; + let notify = rig.spawn_notify(&publisher); + tokio::time::sleep(Duration::from_millis(500)).await; + + for body in [ + r#"{"key":"repos/t/r/wal/abc.pack"}"#, + r#"{"key":"somewhere/else/manifest.pb"}"#, + r#"{"repo":"t/ghost"}"#, + r#"{"key":"repos/t/ghost/manifest.pb"}"#, + r#"{"repo":"not a repo id"}"#, + r#"{"Records":[{"s3":{"object":{"key":"repos/t/ghost/manifest.pb"}}}]}"#, + "not json", + "{", + "", + "[]", + "null", + "\"a string\"", + ] { + rig.notify(body).await; + } + rig.client.publish(rig.cfg.events.nats.notify_subject.clone(), vec![0xff, 0xfe, 0x00].into()).await.unwrap(); + rig.client.flush().await.unwrap(); + tokio::time::sleep(Duration::from_secs(1)).await; + assert!(!notify.is_finished(), "the notify loop must survive garbage"); + assert_eq!(rig.message_count().await, 0, "nothing was published for any of them"); + let repos: Vec = rig.registry.list().await.unwrap().iter().map(ToString::to_string).collect(); + assert_eq!(repos, ["t/r"], "an unknown repository is not created by being notified"); + + // A later valid notification still works. + rig.notify(r#"{"repo":"t/r"}"#).await; + rig.wait_for_messages(1).await; + assert!(!notify.is_finished()); + notify.abort(); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn known_limitation_head_start_skips_pushes_made_while_the_publisher_was_down() { // PINNED: docs/EVENTS.md says `start = "head"` publishes only pushes made after the publisher @@ -695,6 +849,162 @@ async fn a_foreign_message_on_a_repo_subject_fails_every_catch_up_with_a_clear_e assert_eq!(publisher.catch_up(&other).await.unwrap().emitted, 1); } +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_nats_server_restart_keeps_the_stream_the_subscriptions_and_the_service() { + let Some(mut nats) = start_nats().await else { return }; + let rig = Rig::new(&nats, NatsStart::Oldest).await; + let id = RepoId::new("t", "r").unwrap(); + rig.registry.create(&id, ObjectFormat::Sha1).await.unwrap(); + rig.push(&id, &[("refs/heads/main", ZERO, A)]).await; + let publisher = rig.publisher(NatsStart::Oldest).await; + let notify = rig.spawn_notify(&publisher); + let service = rig.spawn_service(&publisher); + tokio::time::sleep(Duration::from_millis(500)).await; + rig.notify(r#"{"repo":"t/r"}"#).await; + rig.wait_for_messages(1).await; + + nats.restart().await; + rig.wait_connected().await; + assert_eq!(rig.message_count().await, 1, "the stream survives the restart"); + + // The queue subscription is back: a notification after the restart triggers a catch-up. + rig.push(&id, &[("refs/heads/main", A, B)]).await; + rig.notify(r#"{"repo":"t/r"}"#).await; + rig.wait_for_messages(2).await; + assert_eq!(rig.ids().await, [msg_id(&id, 1, "refs/heads/main"), msg_id(&id, 2, "refs/heads/main")]); + assert!(!notify.is_finished() && !service.is_finished()); + + // The $SRV service is back too. + rig.push(&id, &[("refs/heads/main", B, C)]).await; + let reply = tokio::time::timeout( + Duration::from_secs(5), + rig.client.request("walgit.events.catch-up", r#"{"repo":"t/r"}"#.into()), + ) + .await + .unwrap() + .unwrap(); + let report: serde_json::Value = serde_json::from_slice(&reply.payload).unwrap(); + assert_eq!(report["emitted"], 1, "{report}"); + assert_eq!(rig.message_count().await, 3, "nothing was republished"); + notify.abort(); + service.abort(); +} + +fn error_code(reply: &async_nats::Message) -> Option { + reply + .headers + .as_ref() + .and_then(|h| h.get("Nats-Service-Error-Code")) + .map(|c| c.as_str().to_string()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn the_service_answers_bad_requests_and_reports_the_counters() { + let Some(nats) = start_nats().await else { return }; + let rig = Rig::new(&nats, NatsStart::Oldest).await; + let id = repo_with_three_events(&rig, "t", "r").await; + let publisher = rig.publisher(NatsStart::Oldest).await; + let service = rig.spawn_service(&publisher); + tokio::time::sleep(Duration::from_millis(500)).await; + + for (body, code) in [ + ("not json", "400"), + ("{}", "400"), + (r#"{"repo":42}"#, "400"), + (r#"{"repo":"no-slash"}"#, "400"), + (r#"{"repo":"a/b/c"}"#, "400"), + (r#"{"repo":"t/none"}"#, "404"), + ] { + let reply = rig.client.request("walgit.events.catch-up", body.into()).await.unwrap(); + assert_eq!(error_code(&reply).as_deref(), Some(code), "{body}"); + } + + // published: 3 through the endpoint. + let reply = rig.client.request("walgit.events.catch-up", r#"{"repo":"t/r"}"#.into()).await.unwrap(); + let report: serde_json::Value = serde_json::from_slice(&reply.payload).unwrap(); + assert_eq!((report["emitted"].as_u64(), report["head_seq"].as_u64()), (Some(3), Some(2))); + + // deduplicated: 2, from forcing a republish of the entry inside the window. + rig.inject(&id, 2, "refs/heads/not-in-the-entry").await; + let restarted = rig.publisher(NatsStart::Oldest).await; + restarted.catch_up(&id).await.unwrap(); + + // gap_entries: 3, on a second repository. + let gap = RepoId::new("t", "gap").unwrap(); + rig.registry.create(&gap, ObjectFormat::Sha1).await.unwrap(); + rig.push(&gap, &[("refs/heads/main", ZERO, A)]).await; + assert_eq!(publisher.catch_up(&gap).await.unwrap().emitted, 1); + for (old, new) in [(A, B), (B, C), (C, A)] { + rig.push(&gap, &[("refs/heads/main", old, new)]).await; + } + rig.registry.open(&gap).await.unwrap().write_checkpoint().await.unwrap(); + rig.push(&gap, &[("refs/heads/main", A, B)]).await; + assert_eq!(publisher.catch_up(&gap).await.unwrap().gap, Some([2, 4])); + + // The service's publisher is `publisher`: its counters are what $SRV.STATS reports. + let stats = rig.client.request("$SRV.STATS.walgit-events", "".into()).await.unwrap(); + let stats: serde_json::Value = serde_json::from_slice(&stats.payload).unwrap(); + eprintln!("stats: {stats}"); + let data = &stats["endpoints"][0]["data"]; + assert_eq!(stats["endpoints"][0]["subject"], "walgit.events.catch-up"); + assert_eq!(data["published"], 5, "3 + 1 + 1"); + assert_eq!(data["deduplicated"], 0, "the duplicates were counted by the other publisher"); + assert_eq!(data["gap_entries"], 3); + assert_eq!(data["conflicts"], 0); + assert_eq!(data["sweep_found"], 0); + assert_eq!(restarted.counters.snapshot()["deduplicated"], 2); + service.abort(); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn the_periodic_sweep_finds_a_push_nobody_notified() { + let Some(nats) = start_nats().await else { return }; + let rig = Rig::new(&nats, NatsStart::Oldest).await; + let id = RepoId::new("t", "r").unwrap(); + rig.registry.create(&id, ObjectFormat::Sha1).await.unwrap(); + rig.push(&id, &[("refs/heads/main", ZERO, A)]).await; + let publisher = rig.publisher(NatsStart::Oldest).await; + let sweeping = tokio::spawn(walgit_nats::wake::sweep_loop(publisher.clone(), Duration::from_millis(300))); + + // The startup sweep legitimately finds work and does not count it. + rig.wait_for_messages(1).await; + assert_eq!(publisher.counters.snapshot()["sweep_found"], 0); + + // No notification for this push: only the periodic sweep can find it. + rig.push(&id, &[("refs/heads/main", A, B)]).await; + rig.wait_for_messages(2).await; + // The sweep counts what it found once its catch-up returns, after the message is acknowledged. + let t0 = Instant::now(); + while publisher.counters.snapshot()["sweep_found"] == 0 && t0.elapsed() < Duration::from_secs(10) { + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert_eq!(publisher.counters.snapshot()["sweep_found"], 1); + sweeping.abort(); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_notification_inside_the_freshness_ttl_still_publishes_the_push() { + let Some(nats) = start_nats().await else { return }; + let rig = Rig::new(&nats, NatsStart::Oldest).await; + let id = RepoId::new("t", "r").unwrap(); + rig.registry.create(&id, ObjectFormat::Sha1).await.unwrap(); + rig.push(&id, &[("refs/heads/main", ZERO, A)]).await; + let (publisher, _registry, _cache) = rig + .operator_publisher(|c| c.wal.freshness_ttl = Duration::from_mins(10)) + .await; + let notify = rig.spawn_notify(&publisher); + tokio::time::sleep(Duration::from_millis(500)).await; + rig.notify(r#"{"repo":"t/r"}"#).await; + rig.wait_for_messages(1).await; + + // Pushed through another registry: the publisher only learns of it from the bucket. + rig.push(&id, &[("refs/heads/main", A, B)]).await; + rig.notify(r#"{"repo":"t/r"}"#).await; + rig.wait_for_messages(2).await; + assert_eq!(rig.ids().await, [msg_id(&id, 1, "refs/heads/main"), msg_id(&id, 2, "refs/heads/main")]); + notify.abort(); +} + fn pack_files(dir: &std::path::Path) -> Vec { let mut found = Vec::new(); let mut stack = vec![dir.to_path_buf()]; From 93641a4016f797c3b9faaa0a2e6bc6dddd6b3cf1 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 26 Oct 2026 22:23:37 -0400 Subject: [PATCH 2/4] test(nats): count the service's republish as the whole recreated repository Signed-off-by: Yordis Prieto --- crates/walgit-nats/tests/jetstream.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/walgit-nats/tests/jetstream.rs b/crates/walgit-nats/tests/jetstream.rs index 61afacd..fa6ec11 100644 --- a/crates/walgit-nats/tests/jetstream.rs +++ b/crates/walgit-nats/tests/jetstream.rs @@ -924,7 +924,7 @@ async fn the_service_answers_bad_requests_and_reports_the_counters() { let report: serde_json::Value = serde_json::from_slice(&reply.payload).unwrap(); assert_eq!((report["emitted"].as_u64(), report["head_seq"].as_u64()), (Some(3), Some(2))); - // deduplicated: 2, from forcing a republish of the entry inside the window. + // deduplicated: 3, from forcing a republish of the repository inside the window. rig.inject(&id, 2, "refs/heads/not-in-the-entry").await; let restarted = rig.publisher(NatsStart::Oldest).await; restarted.catch_up(&id).await.unwrap(); @@ -952,7 +952,7 @@ async fn the_service_answers_bad_requests_and_reports_the_counters() { assert_eq!(data["gap_entries"], 3); assert_eq!(data["conflicts"], 0); assert_eq!(data["sweep_found"], 0); - assert_eq!(restarted.counters.snapshot()["deduplicated"], 2); + assert_eq!(restarted.counters.snapshot()["deduplicated"], 3); service.abort(); } From 2387fc87ab1c32a5f81be2565ac9172a5b3ac633 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Tue, 27 Oct 2026 02:04:35 -0400 Subject: [PATCH 3/4] fix(nats): return from subscribing only once the server has the subscription Signed-off-by: Yordis Prieto --- crates/walgit-nats/src/service.rs | 33 +++++++++++++--- crates/walgit-nats/src/wake.rs | 22 +++++++---- crates/walgit-nats/tests/jetstream.rs | 54 +++++++++++---------------- 3 files changed, 64 insertions(+), 45 deletions(-) diff --git a/crates/walgit-nats/src/service.rs b/crates/walgit-nats/src/service.rs index 5e311e4..cb556bc 100644 --- a/crates/walgit-nats/src/service.rs +++ b/crates/walgit-nats/src/service.rs @@ -41,13 +41,21 @@ async fn handle(publisher: &Publisher, payload: &[u8]) -> Result, -) -> anyhow::Result<()> { + publisher: &Arc, +) -> anyhow::Result { let stats = publisher.clone(); let service = client .service_builder() @@ -57,12 +65,25 @@ pub async fn serve( .start(SERVICE_NAME, env!("CARGO_PKG_VERSION")) .await .map_err(|e| anyhow::anyhow!("starting the {SERVICE_NAME} service: {e}"))?; - let mut endpoint = service + let endpoint = service .group(prefix.child("events")) .endpoint("catch-up") .await .map_err(|e| anyhow::anyhow!("registering catch-up: {e}"))?; + client.flush().await?; tracing::info!(subject = %prefix.child("events.catch-up"), "service ready"); + Ok(Registered { + _service: service, + endpoint, + }) +} + +/// Serve until the endpoint closes. +pub async fn serve(registered: Registered, publisher: Arc) -> anyhow::Result<()> { + let Registered { + _service, + mut endpoint, + } = registered; while let Some(request) = endpoint.next().await { let publisher = publisher.clone(); tokio::spawn(async move { diff --git a/crates/walgit-nats/src/wake.rs b/crates/walgit-nats/src/wake.rs index d10dc64..3f39376 100644 --- a/crates/walgit-nats/src/wake.rs +++ b/crates/walgit-nats/src/wake.rs @@ -24,18 +24,26 @@ pub fn notified_repos(payload: &[u8], store_prefix: &str) -> BTreeSet { .collect() } -/// Subscribe to `subject` in `queue_group` (instances share the notifications) until the -/// subscription ends. A failed catch-up is logged, not retried: core NATS does not redeliver, and -/// the sweep is the backstop. -pub async fn notify_loop( - client: async_nats::Client, +/// Subscribe to `subject` in `queue_group` (instances share the notifications). Returns once the +/// server has registered the subscription, so a notification sent afterwards is not dropped. +pub async fn subscribe( + client: &async_nats::Client, subject: String, queue_group: String, +) -> anyhow::Result { + let sub = client.queue_subscribe(subject.clone(), queue_group).await?; + client.flush().await?; + tracing::info!(%subject, "listening for bucket notifications"); + Ok(sub) +} + +/// Consume `sub` until it ends. A failed catch-up is logged, not retried: core NATS does not +/// redeliver, and the sweep is the backstop. +pub async fn notify_loop( + mut sub: async_nats::Subscriber, store_prefix: String, publisher: Arc, ) -> anyhow::Result<()> { - let mut sub = client.queue_subscribe(subject.clone(), queue_group).await?; - tracing::info!(%subject, "listening for bucket notifications"); while let Some(msg) = sub.next().await { for id in notified_repos(&msg.payload, &store_prefix) { if let Err(e) = publisher.catch_up_if_present(&id).await { diff --git a/crates/walgit-nats/tests/jetstream.rs b/crates/walgit-nats/tests/jetstream.rs index fa6ec11..6990960 100644 --- a/crates/walgit-nats/tests/jetstream.rs +++ b/crates/walgit-nats/tests/jetstream.rs @@ -207,23 +207,29 @@ impl Rig { } } - fn spawn_notify(&self, publisher: &Arc) -> tokio::task::JoinHandle> { + async fn spawn_notify(&self, publisher: &Arc) -> tokio::task::JoinHandle> { let nats_cfg = &self.cfg.events.nats; - tokio::spawn(walgit_nats::wake::notify_loop( - self.client.clone(), + let sub = walgit_nats::wake::subscribe( + &self.client, nats_cfg.notify_subject.clone(), nats_cfg.queue_group.clone(), + ) + .await + .unwrap(); + tokio::spawn(walgit_nats::wake::notify_loop( + sub, self.cfg.store_prefix(), publisher.clone(), )) } - fn spawn_service(&self, publisher: &Arc) -> tokio::task::JoinHandle> { - let (client, prefix, publisher) = (self.client.clone(), self.prefix.clone(), publisher.clone()); - let queue_group = self.cfg.events.nats.queue_group.clone(); - tokio::spawn(async move { - walgit_nats::service::serve(client, &prefix, &queue_group, publisher).await - }) + async fn spawn_service(&self, publisher: &Arc) -> tokio::task::JoinHandle> { + let queue_group = &self.cfg.events.nats.queue_group; + let registered = + walgit_nats::service::register(&self.client, &self.prefix, queue_group, publisher) + .await + .unwrap(); + tokio::spawn(walgit_nats::service::serve(registered, publisher.clone())) } async fn notify(&self, body: &str) { @@ -413,20 +419,8 @@ async fn notifications_and_the_service_api_wake_the_publisher() { let publisher = rig.publisher(NatsStart::Oldest).await; let nats_cfg = &rig.cfg.events.nats; - let notify = tokio::spawn(walgit_nats::wake::notify_loop( - rig.client.clone(), - nats_cfg.notify_subject.clone(), - nats_cfg.queue_group.clone(), - rig.cfg.store_prefix(), - publisher.clone(), - )); - let service = tokio::spawn({ - let (client, prefix, publisher) = - (rig.client.clone(), rig.prefix.clone(), publisher.clone()); - let queue_group = nats_cfg.queue_group.clone(); - async move { walgit_nats::service::serve(client, &prefix, &queue_group, publisher).await } - }); - tokio::time::sleep(Duration::from_millis(500)).await; + let notify = rig.spawn_notify(&publisher).await; + let service = rig.spawn_service(&publisher).await; // An S3 bucket notification, as the object store publishes it. rig.push(&id, &[("refs/heads/main", ZERO, A)]).await; @@ -762,8 +756,7 @@ async fn notifications_that_name_nothing_are_ignored_and_the_loop_survives() { rig.registry.create(&id, ObjectFormat::Sha1).await.unwrap(); rig.push(&id, &[("refs/heads/main", ZERO, A)]).await; let publisher = rig.publisher(NatsStart::Oldest).await; - let notify = rig.spawn_notify(&publisher); - tokio::time::sleep(Duration::from_millis(500)).await; + let notify = rig.spawn_notify(&publisher).await; for body in [ r#"{"key":"repos/t/r/wal/abc.pack"}"#, @@ -857,9 +850,8 @@ async fn a_nats_server_restart_keeps_the_stream_the_subscriptions_and_the_servic rig.registry.create(&id, ObjectFormat::Sha1).await.unwrap(); rig.push(&id, &[("refs/heads/main", ZERO, A)]).await; let publisher = rig.publisher(NatsStart::Oldest).await; - let notify = rig.spawn_notify(&publisher); - let service = rig.spawn_service(&publisher); - tokio::time::sleep(Duration::from_millis(500)).await; + let notify = rig.spawn_notify(&publisher).await; + let service = rig.spawn_service(&publisher).await; rig.notify(r#"{"repo":"t/r"}"#).await; rig.wait_for_messages(1).await; @@ -904,8 +896,7 @@ async fn the_service_answers_bad_requests_and_reports_the_counters() { let rig = Rig::new(&nats, NatsStart::Oldest).await; let id = repo_with_three_events(&rig, "t", "r").await; let publisher = rig.publisher(NatsStart::Oldest).await; - let service = rig.spawn_service(&publisher); - tokio::time::sleep(Duration::from_millis(500)).await; + let service = rig.spawn_service(&publisher).await; for (body, code) in [ ("not json", "400"), @@ -992,8 +983,7 @@ async fn a_notification_inside_the_freshness_ttl_still_publishes_the_push() { let (publisher, _registry, _cache) = rig .operator_publisher(|c| c.wal.freshness_ttl = Duration::from_mins(10)) .await; - let notify = rig.spawn_notify(&publisher); - tokio::time::sleep(Duration::from_millis(500)).await; + let notify = rig.spawn_notify(&publisher).await; rig.notify(r#"{"repo":"t/r"}"#).await; rig.wait_for_messages(1).await; From 10484593a1823482bc136a4efd02e27e8943297c Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Tue, 27 Oct 2026 02:04:57 -0400 Subject: [PATCH 4/4] fix(nats): refuse catch-up requests beyond a bounded backlog Signed-off-by: Yordis Prieto --- crates/walgit-nats/src/publisher.rs | 2 ++ crates/walgit-nats/src/service.rs | 45 +++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+) diff --git a/crates/walgit-nats/src/publisher.rs b/crates/walgit-nats/src/publisher.rs index 523815d..5cfce31 100644 --- a/crates/walgit-nats/src/publisher.rs +++ b/crates/walgit-nats/src/publisher.rs @@ -55,6 +55,7 @@ pub struct Counters { pub conflicts: AtomicU64, pub rewinds: AtomicU64, pub sweep_found: AtomicU64, + pub refused: AtomicU64, } impl Counters { @@ -67,6 +68,7 @@ impl Counters { "conflicts": n(&self.conflicts), "rewinds": n(&self.rewinds), "sweep_found": n(&self.sweep_found), + "refused": n(&self.refused), }) } } diff --git a/crates/walgit-nats/src/service.rs b/crates/walgit-nats/src/service.rs index cb556bc..b5025cf 100644 --- a/crates/walgit-nats/src/service.rs +++ b/crates/walgit-nats/src/service.rs @@ -4,11 +4,13 @@ use std::str::FromStr; use std::sync::Arc; +use std::sync::atomic::Ordering; use async_nats::service::ServiceExt; use async_nats::service::error::Error as ServiceError; use bytes::Bytes; use futures::StreamExt; +use tokio::sync::{OwnedSemaphorePermit, Semaphore}; use walgit_git::RepoId; use crate::publisher::Publisher; @@ -16,6 +18,10 @@ use crate::subject::SubjectPrefix; pub const SERVICE_NAME: &str = "walgit-events"; +/// Catch-up requests in flight at once. They serialize on the publisher, so more than this only +/// queues payloads behind notification-driven catch-up. +const MAX_PENDING_REQUESTS: usize = 64; + #[derive(serde::Deserialize)] struct CatchUpRequest { repo: String, @@ -28,6 +34,15 @@ fn failure(code: usize, status: &impl ToString) -> ServiceError { } } +fn admit(slots: &Arc) -> Result { + slots.clone().try_acquire_owned().map_err(|_| { + failure( + 503, + &format!("busy: {MAX_PENDING_REQUESTS} catch-up requests already pending; retry"), + ) + }) +} + async fn handle(publisher: &Publisher, payload: &[u8]) -> Result { let req: CatchUpRequest = serde_json::from_slice(payload).map_err(|e| failure(400, &format!("request: {e}")))?; @@ -84,9 +99,21 @@ pub async fn serve(registered: Registered, publisher: Arc) -> anyhow: _service, mut endpoint, } = registered; + let slots = Arc::new(Semaphore::new(MAX_PENDING_REQUESTS)); while let Some(request) = endpoint.next().await { + let permit = match admit(&slots) { + Ok(permit) => permit, + Err(busy) => { + publisher.counters.refused.fetch_add(1, Ordering::Relaxed); + if let Err(e) = request.respond(Err(busy)).await { + tracing::warn!(error = %e, "catch-up reply failed"); + } + continue; + } + }; let publisher = publisher.clone(); tokio::spawn(async move { + let _permit = permit; let answer = handle(&publisher, &request.message.payload).await; if let Err(e) = request.respond(answer).await { tracing::warn!(error = %e, "catch-up reply failed"); @@ -95,3 +122,21 @@ pub async fn serve(registered: Registered, publisher: Arc) -> anyhow: } Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn requests_beyond_the_backlog_are_refused_until_one_finishes() { + let slots = Arc::new(Semaphore::new(MAX_PENDING_REQUESTS)); + let mut held: Vec<_> = (0..MAX_PENDING_REQUESTS) + .map(|_| admit(&slots).unwrap()) + .collect(); + let busy = admit(&slots).unwrap_err(); + assert_eq!(busy.code, 503); + assert!(busy.status.starts_with("busy"), "{}", busy.status); + held.pop(); + assert!(admit(&slots).is_ok()); + } +}