From 9a24f26da68c2cba6224c7b9e0007e3dbbd67725 Mon Sep 17 00:00:00 2001 From: jilio <22717553+jilio@users.noreply.github.com> Date: Sun, 11 Oct 2026 23:19:43 +1100 Subject: [PATCH] =?UTF-8?q?Resolve=20a=20{"repo":=20=E2=80=A6}=20notify=20?= =?UTF-8?q?under=20a=20store=20prefix?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit POST /_events/notify turns a plain {"repo": "o/r"} into the key repos/o/r/manifest.pb and resolves it like a bucket notification, which must start with store.prefix. With a non-empty prefix the key never matches: the notify is acked with an empty report and the bridge waits for its sweep (default 5 min), silently. Bucket notifications are fine: they carry the prefixed object key. Build the key the repository's manifest has in this bucket, prefix included. Keys in bucket notifications and {"key": …} stay bucket object keys; EVENTS.md now says so. The test harness now applies store.prefix the way walgit_store::from_config does for real backends, so a server test can run under a prefix. notify_wakes_the_bridge_under_a_store_prefix sends all four shapes to a prefixed server (before: {"repo": "t/r"} answered [] and woke nothing); the bridge unit test resolves them under a prefix. Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/walgit-server/src/bridge.rs | 31 +++++++++++-- crates/walgit-server/tests/events.rs | 67 +++++++++++++++++++++++++++ crates/walgit-server/tests/harness.rs | 8 +++- docs/EVENTS.md | 4 +- 4 files changed, 103 insertions(+), 7 deletions(-) diff --git a/crates/walgit-server/src/bridge.rs b/crates/walgit-server/src/bridge.rs index 349ff181..870b340a 100644 --- a/crates/walgit-server/src/bridge.rs +++ b/crates/walgit-server/src/bridge.rs @@ -14,7 +14,8 @@ //! `…/manifest.pb` — the commit point itself as a notification. Accepted //! shapes: a GCS Pub/Sub push envelope (`message.attributes.objectId`), an S3 //! event notification (`Records[].s3.object.key`), or a plain `{"key": "…"}` -//! / `{"repo": "owner/name"}`. A non-2xx here is meant to be redelivered; +//! (all bucket keys, `store.prefix` included) / `{"repo": "owner/name"}` +//! (whatever the prefix). A non-2xx here is meant to be redelivered; //! * the sweep (`events.sweep_interval`): every repo — the backstop. A sweep //! that finds unpublished entries means the notifications are not flowing //! and says so (`events_bridge_sweep_found_total`, warn). @@ -258,8 +259,9 @@ impl Bridge { } } -/// The object keys (or repository ids) a notification body names. Store-agnostic. -fn notified_keys(v: &serde_json::Value) -> Vec { +/// The bucket object keys a notification body names, `store.prefix` included as the +/// notifications carry it. Store-agnostic. +fn notified_keys(v: &serde_json::Value, store_prefix: &str) -> Vec { let mut keys = Vec::new(); // GCS → Pub/Sub push envelope. let attrs = &v["message"]["attributes"]; @@ -299,8 +301,9 @@ fn notified_keys(v: &serde_json::Value) -> Vec { if let Some(k) = v["key"].as_str() { keys.push(k.to_string()); } + // A repository id names no object: use the key its manifest has in this bucket. if let Some(r) = v["repo"].as_str() { - keys.push(format!("repos/{r}/manifest.pb")); + keys.push(format!("{store_prefix}repos/{r}/manifest.pb")); } keys } @@ -324,7 +327,7 @@ pub async fn http_notify( let v: serde_json::Value = serde_json::from_slice(&bytes) .map_err(|e| ApiError::BadRequest(format!("notify body: {e}")))?; let mut reports = Vec::new(); - for key in notified_keys(&v) { + for key in notified_keys(&v, &bridge.store_prefix) { match bridge.object_finalized(&key).await { Ok(Some(report)) => reports.push(report), Ok(None) => {} @@ -390,7 +393,25 @@ mod tests { .manifest_repo("prefix/repos/acme/monorepo/manifest.pb") .unwrap(); assert_eq!(id.to_string(), "acme/monorepo"); + // Every body shape names the same repository under a non-empty prefix. + for body in [ + serde_json::json!({"repo": "acme/monorepo"}), + serde_json::json!({"key": "prefix/repos/acme/monorepo/manifest.pb"}), + serde_json::json!({"message": {"attributes": { + "eventType": "OBJECT_FINALIZE", + "objectId": "prefix/repos/acme/monorepo/manifest.pb"}}}), + serde_json::json!({"Records": [{"eventName": "ObjectCreated:Put", + "s3": {"object": {"key": "prefix/repos/acme/monorepo/manifest.pb"}}}]}), + ] { + let ids: Vec<_> = notified_keys(&body, &b.store_prefix) + .iter() + .filter_map(|k| b.manifest_repo(k)) + .map(|id| id.to_string()) + .collect(); + assert_eq!(ids, ["acme/monorepo"], "{body}"); + } for other in [ + "repos/t/r/manifest.pb", "prefix/repos/t/r/wal/abc.pack", "prefix/repos/t/r/events/cursor.json", "walgit-go/repos/t/r/manifest.pb", diff --git a/crates/walgit-server/tests/events.rs b/crates/walgit-server/tests/events.rs index dbc9898c..92084ef9 100644 --- a/crates/walgit-server/tests/events.rs +++ b/crates/walgit-server/tests/events.rs @@ -206,6 +206,73 @@ async fn bridge_publishes_from_cursor_exactly_once() -> TestResult { Ok(()) } +/// With a bucket prefix (`store.prefix`), every notification shape wakes the bridge: bucket +/// notifications carry the prefixed object key, and a plain `{"repo": …}` names the repository +/// whatever the prefix. A key outside this walgit's prefix is acked and ignored. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn notify_wakes_the_bridge_under_a_store_prefix() -> TestResult { + use walgit_store::ObjectStoreExt; + let (url, captured) = webhook().await; + let server = Server::start_with_tweak(|c| { + bridge_cfg(&url, Duration::ZERO)(c); + c.store.prefix = "tenant-a".into(); + }) + .await?; + server.put_repo("t", "r").await?; + assert!( + server + .store + .exists("tenant-a/repos/t/r/manifest.pb") + .await?, + "the repository lives under the prefix" + ); + let src = TestRepo::synthetic(1, 1)?; + git_in(&src, &["commit", "--allow-empty", "-m", "a"])?; + git_in(&src, &["branch", "-M", "main"])?; + git_in( + &src, + &["remote", "add", "origin", &server.repo_url("t", "r")], + )?; + git_in(&src, &["push", "-u", "origin", "main"])?; + + let client = reqwest::Client::new(); + let notify = |body: serde_json::Value| { + let client = client.clone(); + let url = format!("{}/_events/notify", server.base_url); + async move { + let resp = client.post(url).json(&body).send().await?; + anyhow::ensure!(resp.status() == 200, "{body}: {}", resp.status()); + anyhow::Ok(resp.json::().await?) + } + }; + // Outside the prefix: not this walgit's object. + let report = notify(serde_json::json!({"key": "repos/t/r/manifest.pb"})).await?; + assert_eq!(report, serde_json::json!([])); + assert!(captured.lock().unwrap().is_empty()); + + let mut events = 0; + for body in [ + serde_json::json!({"repo": "t/r"}), + gcs_notification("tenant-a/repos/t/r/manifest.pb", "OBJECT_FINALIZE"), + serde_json::json!({"Records": [{"eventName": "ObjectCreated:Put", "s3": {"object": {"key": "tenant-a/repos/t/r/manifest.pb"}}}]}), + serde_json::json!({"key": "tenant-a/repos/t/r/manifest.pb"}), + ] { + git_in(&src, &["commit", "--allow-empty", "-m", "next"])?; + git_in(&src, &["push"])?; + events += 1; + let report = notify(body.clone()).await?; + assert_eq!(report[0]["repo"], "t/r", "{body}: {report}"); + assert_eq!( + report[0]["emitted"], + if events == 1 { 2 } else { 1 }, + "{body}: {report}" + ); + } + let got = wait_for(&captured, 5).await; + assert_eq!(got.len(), 5); + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn bridge_sweep_timer_publishes_without_notifications() -> TestResult { let (url, captured) = webhook().await; diff --git a/crates/walgit-server/tests/harness.rs b/crates/walgit-server/tests/harness.rs index da9ea20f..520e1b5a 100644 --- a/crates/walgit-server/tests/harness.rs +++ b/crates/walgit-server/tests/harness.rs @@ -109,7 +109,13 @@ impl Server { cfg.server.public_url = Some(base_url.clone()); cfg.validate().context("config validate")?; - let dyn_store: DynStore = store.clone(); + // `store.prefix` scopes every key, as `walgit_store::from_config` does for real backends. + let prefix = cfg.store_prefix(); + let dyn_store: DynStore = if prefix.is_empty() { + store.clone() + } else { + Arc::new(walgit_store::Prefixed::new(store.clone(), prefix)) + }; let state = AppState::new(Arc::new(cfg), dyn_store).await?; let registry = state.registry.clone(); diff --git a/docs/EVENTS.md b/docs/EVENTS.md index 6ca6b97f..8f52b709 100644 --- a/docs/EVENTS.md +++ b/docs/EVENTS.md @@ -97,7 +97,9 @@ Wake-ups (both idempotent; they only ever call `catch_up`): itself as the notification. Accepted bodies: a GCS Pub/Sub push envelope (`message.attributes.eventType = OBJECT_FINALIZE`, `objectId`), an S3 event notification (`Records[].eventName = ObjectCreated:*`, `s3.object.key`; MinIO, rustfs and Ceph emit the same shape), or your own glue's `{"key": "repos/o/r/manifest.pb"}` - / `{"repo": "o/r"}`. Everything else is acked and ignored; a webhook failure answers 503 so the notifier + / `{"repo": "o/r"}`. Keys are bucket object keys as notifications carry them, `store.prefix` included + (`{"key": "/repos/o/r/manifest.pb"}`); `{"repo": …}` names the repository whatever the prefix. + Everything else is acked and ignored; a webhook failure answers 503 so the notifier redelivers. Authenticated like every route (`require_read`): give the notifier a token. - The sweep (`events.sweep_interval`, default 5 min): `list` + one conditional manifest GET per repo. Not needed for correctness; it is the backstop *and the health check* — a sweep that publishes anything means