feat(nats): wake the publisher over NATS so it needs no HTTP listener

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

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};

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),
         })
     }
 }

crates/walgit-nats/src/service.rs

@@ -0,0 +1,142 @@
+//! The NATS service API (`$SRV` micro protocol) in place of an HTTP listener:
+//! `<prefix>.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 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;
+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,
+}
+
+fn failure(code: usize, status: &impl ToString) -> ServiceError {
+    ServiceError {
+        status: status.to_string(),
+        code,
+    }
+}
+
+fn admit(slots: &Arc<Semaphore>) -> Result<OwnedSemaphorePermit, ServiceError> {
+    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<Bytes, ServiceError> {
+    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:#}"))),
+    }
+}
+
+/// A registered service: the handle keeps `$SRV` answering, the endpoint carries the requests.
+pub struct Registered {
+    _service: async_nats::service::Service,
+    endpoint: async_nats::service::endpoint::Endpoint,
+}
+
+/// Register the service and its endpoint. Returns once the server has processed the
+/// subscriptions, so a request sent afterwards finds a responder. Instances share requests
+/// through the service queue group.
+pub async fn register(
+    client: &async_nats::Client,
+    prefix: &SubjectPrefix,
+    queue_group: &str,
+    publisher: &Arc<Publisher>,
+) -> anyhow::Result<Registered> {
+    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 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<Publisher>) -> anyhow::Result<()> {
+    let Registered {
+        _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");
+            }
+        });
+    }
+    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());
+    }
+}

crates/walgit-nats/src/wake.rs

@@ -0,0 +1,143 @@
+//! 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<RepoId> {
+    let Ok(value) = serde_json::from_slice::<serde_json::Value>(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). 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<async_nats::Subscriber> {
+    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<Publisher>,
+) -> anyhow::Result<()> {
+    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<Publisher>, 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<String> {
+        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());
+    }
+}