diff --git a/knot2/crates/knot-record/src/chain.rs b/knot2/crates/knot-record/src/chain.rs index aaeb48348..59cb08d29 100644 --- a/knot2/crates/knot-record/src/chain.rs +++ b/knot2/crates/knot-record/src/chain.rs @@ -32,7 +32,7 @@ pub const MAX_COMMIT_BLOCKS_BYTES: usize = 2_000_000; pub trait Emit { fn rev(&mut self, prev: Option<&Tid>) -> Tid; fn frame(&self, chunk: &CommitChunk) -> Option; - fn committed(&mut self, frame: Option); + fn committed(&mut self, chunk: &CommitChunk, frame: Option); } impl Emit for Ticker { @@ -44,7 +44,7 @@ impl Emit for Ticker { None } - fn committed(&mut self, _frame: Option) {} + fn committed(&mut self, _chunk: &CommitChunk, _frame: Option) {} } #[derive(Debug, thiserror::Error)] @@ -482,7 +482,7 @@ pub async fn initialize( Some(blocks_tree), frame.clone(), )?; - emit.committed(frame); + emit.committed(&chunk, frame); Ok(chunk) } @@ -879,7 +879,7 @@ fn write_chunk(plan: &mut Plan<'_, S>, chunk: &CommitChunk) -> Result<(), Cha )?; plan.parent = Some(written); plan.blocks_tree = Some(grown); - plan.emit.committed(frame); + plan.emit.committed(chunk, frame); Ok(()) } @@ -1565,7 +1565,7 @@ mod tests { Some(Bytes::from(chunk.rev.to_string())) } - fn committed(&mut self, frame: Option) { + fn committed(&mut self, _chunk: &CommitChunk, frame: Option) { self.frames.push(frame); } } @@ -1807,7 +1807,7 @@ mod tests { )) } - fn committed(&mut self, _frame: Option) {} + fn committed(&mut self, _chunk: &CommitChunk, _frame: Option) {} } #[tokio::test] @@ -2056,7 +2056,7 @@ mod tests { None } - fn committed(&mut self, _frame: Option) {} + fn committed(&mut self, _chunk: &CommitChunk, _frame: Option) {} } fn faces(chunks: &[CommitChunk]) -> Vec<(CommitChunk, OpMap)> { diff --git a/knot2/crates/knot-server/src/main.rs b/knot2/crates/knot-server/src/main.rs index f83091f6b..0894390cc 100644 --- a/knot2/crates/knot-server/src/main.rs +++ b/knot2/crates/knot-server/src/main.rs @@ -606,6 +606,7 @@ async fn run(config: knot_config::Validated) -> anyhow::Result<()> { let slots = knot_resource::Slots::for_machine(); let (imports, import_jobs) = knot_xrpc::imports::ImportQueue::channel(); + let (notify, notices) = knot_xrpc::notify::NotifyQueue::channel(&syncers); let ssh_base = knot_ssh::SshState::new(knot_ssh::SshConfig { repos: repos.clone(), @@ -670,6 +671,7 @@ async fn run(config: knot_config::Validated) -> anyhow::Result<()> { catalog: Arc::clone(&catalog), rkeys: Default::default(), syncers, + notify, }); let ssh_base = ssh_base.with_ref_surface(Arc::new(knot_xrpc::RefProjection { state: Arc::clone(&xrpc_state), @@ -793,6 +795,23 @@ async fn run(config: knot_config::Validated) -> anyhow::Result<()> { )); tracing::info!("import worker running"); } + { + let (notify_stop, notify_stop_rx) = tokio::sync::watch::channel(false); + let notify_shutdown = shutdown.clone(); + tokio::spawn(async move { + notify_shutdown.cancelled().await; + let _ = notify_stop.send(true); + }); + tokio::spawn(knot_xrpc::notify::run( + Arc::clone(&xrpc_state), + notices, + notify_stop_rx, + )); + tokio::spawn(knot_xrpc::notify::run_sweeper( + Arc::clone(&xrpc_state), + shutdown.clone(), + )); + } tokio::spawn(allocator::govern_decay(shutdown.clone())); let keyfill_task = knot_keyfill::spawn( Arc::clone(&index), diff --git a/knot2/crates/knot-sim/src/harness.rs b/knot2/crates/knot-sim/src/harness.rs index 681942a62..2e531589a 100644 --- a/knot2/crates/knot-sim/src/harness.rs +++ b/knot2/crates/knot-sim/src/harness.rs @@ -882,6 +882,7 @@ fn assemble_router(parts: StateParts) -> (Router, FakeState) { catalog: Arc::new(knot_messages::Catalog::defaults()), rkeys: Default::default(), syncers: Default::default(), + notify: knot_xrpc::notify::NotifyQueue::channel(&Default::default()).0, }); let resolver = knot_pack::index_resolver(Arc::clone(&index)); ( diff --git a/knot2/crates/knot-sim/tests/lfs_roundtrip.rs b/knot2/crates/knot-sim/tests/lfs_roundtrip.rs index 524a3c2aa..9fdabe41e 100644 --- a/knot2/crates/knot-sim/tests/lfs_roundtrip.rs +++ b/knot2/crates/knot-sim/tests/lfs_roundtrip.rs @@ -435,6 +435,7 @@ async fn spawn(published_line: String, with_h3: bool) -> World { catalog: Arc::new(knot_messages::Catalog::defaults()), rkeys: Default::default(), syncers: Default::default(), + notify: knot_xrpc::notify::NotifyQueue::channel(&Default::default()).0, }); let (_imports_stop, imports_halt) = tokio::sync::watch::channel(false); diff --git a/knot2/crates/knot-ssh/tests/ssh_push.rs b/knot2/crates/knot-ssh/tests/ssh_push.rs index 93b852086..8bbc4fddf 100644 --- a/knot2/crates/knot-ssh/tests/ssh_push.rs +++ b/knot2/crates/knot-ssh/tests/ssh_push.rs @@ -487,6 +487,7 @@ async fn spawn_server_core( )), rkeys: Default::default(), syncers: Default::default(), + notify: knot_xrpc::notify::NotifyQueue::channel(&Default::default()).0, }); let firehose = Arc::clone(&state.firehose); (Some(state), Some(firehose)) diff --git a/knot2/crates/knot-xrpc/src/atproto.rs b/knot2/crates/knot-xrpc/src/atproto.rs index 8ad3dc33f..a4f24ac06 100644 --- a/knot2/crates/knot-xrpc/src/atproto.rs +++ b/knot2/crates/knot-xrpc/src/atproto.rs @@ -46,7 +46,7 @@ fn repo_not_found(did: &str) -> XrpcError { ) } -fn record_value(bytes: &[u8]) -> Result { +pub(crate) fn record_value(bytes: &[u8]) -> Result { let decoded: Ipld = serde_ipld_dagcbor::from_slice(bytes) .map_err(|error| XrpcError::internal(error.to_string()))?; json_of(&decoded).map_err(XrpcError::internal) diff --git a/knot2/crates/knot-xrpc/src/firehose.rs b/knot2/crates/knot-xrpc/src/firehose.rs index 0502f3fd1..a8d75b985 100644 --- a/knot2/crates/knot-xrpc/src/firehose.rs +++ b/knot2/crates/knot-xrpc/src/firehose.rs @@ -27,6 +27,7 @@ use serde::{Deserialize, Serialize}; use crate::XrpcState; use crate::error::XrpcError; +use crate::notify::{NotifyQueue, WriteNotice}; use crate::out::datetime_of; use Source::{Chains, Ring}; @@ -51,13 +52,17 @@ pub fn feed_of( Resolved::Ready(Some(visibility)) => visibility, _ => Visibility::Private, }; - feed_for(&state.firehose, visibility) + feed_for(&state.firehose, &state.notify, visibility) } -pub(crate) fn feed_for(log: &Arc>, visibility: Visibility) -> Arc { +pub(crate) fn feed_for( + log: &Arc>, + notify: &Arc, + visibility: Visibility, +) -> Arc { match visibility { Visibility::Public => Arc::new(PublicFeed(Arc::clone(log))), - Visibility::Private => Arc::new(PrivateFeed), + Visibility::Private => Arc::new(PrivateFeed(Arc::clone(notify))), } } @@ -79,13 +84,7 @@ pub trait Feed: Send + Sync { now: UnixSeconds, ); - fn publish_account( - &self, - did: &RepoDid, - active: bool, - status: Option<&'static str>, - now: UnixSeconds, - ); + fn publish_deleted(&self, did: &RepoDid, now: UnixSeconds); // counts as a publish because the tombstone goes on the firehose and answers getRepoStatus fn record_deletion( @@ -127,14 +126,8 @@ impl Feed for PublicFeed { publish_push(&self.0, did, pusher, applied, options, signer, now); } - fn publish_account( - &self, - did: &RepoDid, - active: bool, - status: Option<&'static str>, - now: UnixSeconds, - ) { - publish_account(&self.0, did, active, status, now); + fn publish_deleted(&self, did: &RepoDid, now: UnixSeconds) { + publish_deleted(&self.0, did, now); } fn record_deletion( @@ -147,11 +140,11 @@ impl Feed for PublicFeed { } } -pub struct PrivateFeed; +pub struct PrivateFeed(Arc); impl Feed for PrivateFeed { fn emit<'a>(&'a self, did: &'a RepoDid, now: UnixSeconds) -> Box { - Box::new(PrivateEmit::new(did, now)) + Box::new(PrivateEmit::new(did, now, &self.0)) } fn reserve_birth(&self) -> Birth { @@ -171,13 +164,8 @@ impl Feed for PrivateFeed { ) { } - fn publish_account( - &self, - _did: &RepoDid, - _active: bool, - _status: Option<&'static str>, - _now: UnixSeconds, - ) { + fn publish_deleted(&self, did: &RepoDid, _now: UnixSeconds) { + self.0.space_deleted(did.clone()); } fn record_deletion( @@ -201,19 +189,21 @@ impl Birth { } } -// still frames every commit so the chain on disk looks public, nothing gets sent +// still frames every commit so the chain on disk matches a public repo's but nothing goes out pub struct PrivateEmit<'a> { did: &'a RepoDid, now: UnixSeconds, last: Option, + notify: &'a NotifyQueue, } impl<'a> PrivateEmit<'a> { - pub(crate) fn new(did: &'a RepoDid, now: UnixSeconds) -> Self { + pub(crate) fn new(did: &'a RepoDid, now: UnixSeconds, notify: &'a NotifyQueue) -> Self { Self { did, now, last: None, + notify, } } } @@ -230,15 +220,26 @@ impl Emit for PrivateEmit<'_> { } fn frame(&self, chunk: &CommitChunk) -> Option { - commit_frame( + let (commit, car) = commit_car(self.did, chunk)?; + Some(commit_frame_of( self.did, chunk, + commit, + car, UnixMicros::new(chunk.rev.timestamp()), self.now, - ) + )) } - fn committed(&mut self, _frame: Option) {} + fn committed(&mut self, chunk: &CommitChunk, frame: Option) { + if frame.is_none() || !self.notify.watching_writes() { + return; + } + if let Some((_, car)) = commit_car(self.did, chunk) { + self.notify + .write(WriteNotice::of(self.did, chunk, Bytes::from(car))); + } + } } impl ChainEmit for PrivateEmit<'_> { @@ -330,7 +331,7 @@ impl Emit for FirehoseEmit<'_, C> { ) } - fn committed(&mut self, frame: Option) { + fn committed(&mut self, _chunk: &CommitChunk, frame: Option) { self.frames.push(frame); } } @@ -368,14 +369,14 @@ fn fulfill_account( ); } -fn publish_account( - log: &EventLog, - did: &RepoDid, - active: bool, - status: Option<&'static str>, - now: UnixSeconds, -) { - fulfill_account(log.reserve(), did, active, status, now); +fn publish_deleted(log: &EventLog, did: &RepoDid, now: UnixSeconds) { + fulfill_account( + log.reserve(), + did, + false, + Some(crate::plc::STATUS_DELETED), + now, + ); } pub(crate) fn account_deleted_frame(did: &RepoDid, deleted_at: UnixMicros) -> Bytes { @@ -386,7 +387,7 @@ pub(crate) fn account_deleted_frame(did: &RepoDid, deleted_at: UnixMicros) -> By did: did.clone(), time: datetime_of(deleted_at.seconds()), active: false, - status: Some("deleted"), + status: Some(crate::plc::STATUS_DELETED), }, ) } @@ -703,6 +704,11 @@ fn commit_frame( seq: UnixMicros, now: UnixSeconds, ) -> Option { + let (commit_cid, car) = commit_car(did, chunk)?; + Some(commit_frame_of(did, chunk, commit_cid, car, seq, now)) +} + +fn commit_car(did: &RepoDid, chunk: &CommitChunk) -> Option<(Cid, Vec)> { let commit_cid = match chain::commit_cid(&chunk.commit) { Ok(cid) => cid, Err(error) => { @@ -712,8 +718,18 @@ fn commit_frame( }; let mut blocks = chunk.blocks.clone(); blocks.insert(commit_cid, Bytes::from(chunk.commit.clone())); - let car = knot_record::car_bytes(commit_cid, &blocks); - Some(frame( + Some((commit_cid, knot_record::car_bytes(commit_cid, &blocks))) +} + +fn commit_frame_of( + did: &RepoDid, + chunk: &CommitChunk, + commit_cid: Cid, + car: Vec, + seq: UnixMicros, + now: UnixSeconds, +) -> Bytes { + frame( FrameKind::Commit, &CommitMessage { seq, @@ -742,7 +758,7 @@ fn commit_frame( prev_data: chunk.prev_data, time: datetime_of(now), }, - )) + ) } #[derive(serde::Deserialize)] @@ -1612,7 +1628,7 @@ mod tests { .await .unwrap(); emit.finish().unwrap(); - publish_account(&first, &did, false, Some("deleted"), UnixSeconds::new(1)); + publish_deleted(&first, &did, UnixSeconds::new(1)); persist_seq(&first, &repos, &knot); assert!( chunks[0].rev.timestamp() < first.last_issued().get(), @@ -1880,7 +1896,7 @@ mod tests { let untouched = log().last_issued(); let log = Arc::new(log()); let did = did(); - let feed = PrivateFeed; + let feed = PrivateFeed(NotifyQueue::channel(&Default::default()).0); let mut emit = feed.emit(&did, UnixSeconds::new(1_000)); let chunks = futures::executor::block_on(chain::materialize( &repo, @@ -1917,7 +1933,8 @@ mod tests { fn a_private_repo_keeps_its_revs_rising() { let (_dir, _layout, _did, repo, signer) = rig(); let did = did(); - let mut emit = PrivateEmit::new(&did, UnixSeconds::new(1_000)); + let notify = NotifyQueue::channel(&Default::default()).0; + let mut emit = PrivateEmit::new(&did, UnixSeconds::new(1_000), ¬ify); let first = futures::executor::block_on(chain::materialize( &repo, &did, @@ -1928,7 +1945,7 @@ mod tests { )) .unwrap(); // the clock goes backwards here but the rev still has to move forward - let mut emit = PrivateEmit::new(&did, UnixSeconds::new(1)); + let mut emit = PrivateEmit::new(&did, UnixSeconds::new(1), ¬ify); let second = futures::executor::block_on(chain::sync( &repo, &did, @@ -1952,11 +1969,11 @@ mod tests { fn standalone_frames_of_a_private_repo_are_never_published() { let log = Arc::new(log()); let did = did(); - let feed = PrivateFeed; + let feed = PrivateFeed(NotifyQueue::channel(&Default::default()).0); let signer = knot_runtime::K256Signer::from_slice(&[7u8; 32]).unwrap(); let now = UnixSeconds::new(1); feed.reserve_birth().announce(&did, now); - feed.publish_account(&did, false, Some("deleted"), now); + feed.publish_deleted(&did, now); feed.publish_sync(&did, &Tid::from_time(1, 0), &[], now); feed.publish_push( &did, diff --git a/knot2/crates/knot-xrpc/src/lib.rs b/knot2/crates/knot-xrpc/src/lib.rs index 07f921e31..c4dbe0f07 100644 --- a/knot2/crates/knot-xrpc/src/lib.rs +++ b/knot2/crates/knot-xrpc/src/lib.rs @@ -22,6 +22,7 @@ mod members; mod merge; pub mod migrate; mod mint; +pub mod notify; mod out; pub mod plc; mod query; @@ -225,6 +226,7 @@ pub struct XrpcState { pub firehose: Arc>, pub rkeys: RecordKeys, pub syncers: Syncers, + pub notify: Arc, } pub fn knot_repo(knot_did: &KnotId) -> Result { diff --git a/knot2/crates/knot-xrpc/src/notify.rs b/knot2/crates/knot-xrpc/src/notify.rs new file mode 100644 index 000000000..4306edbb6 --- /dev/null +++ b/knot2/crates/knot-xrpc/src/notify.rs @@ -0,0 +1,412 @@ +use std::collections::{BTreeSet, HashMap}; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Duration; + +use base64::Engine as _; +use bytes::Bytes; +use jacquard_common::types::tid::Tid; +use knot_index::Resolved; +use knot_record::chain::{CommitChunk, Op}; +use knot_runtime::{Clock, HttpTransport}; +use knot_types::{Nsid, RepoDid, ServiceDid, Syncers, Visibility}; +use serde_json::{Value, json}; +use tokio::sync::{mpsc, watch}; +use tokio::task::JoinSet; +use tokio_util::sync::CancellationToken; + +use crate::XrpcState; + +const NOTIFY_WRITE: &str = "org.tangled.space.notifyWrite"; +const NOTIFY_SPACE_DELETED: &str = "org.tangled.space.notifySpaceDeleted"; +const NOTIFY_LEASE_LAPSED: &str = "org.tangled.knot.notifyLeaseLapsed"; + +const COVERED_REF: &str = "refs/knot/notify-covered"; +const SWEEP_INTERVAL: Duration = Duration::from_secs(60); +const DELIVERY_TIMEOUT: Duration = Duration::from_secs(10); +const LANE_DEPTH: usize = 256; +const MAX_LAPSED_PER_NOTICE: usize = 100; + +fn method(name: &'static str) -> Nsid { + Nsid::new_static(name).expect("literal NSID parses") +} + +pub struct WriteNotice { + repo: RepoDid, + rev: Tid, + prev_rev: Option, + ops: Vec<(Op, Option)>, + blocks: Bytes, +} + +impl WriteNotice { + pub(crate) fn of(repo: &RepoDid, chunk: &CommitChunk, blocks: Bytes) -> Self { + Self { + repo: repo.clone(), + rev: chunk.rev.clone(), + prev_rev: chunk.prev_rev.clone(), + ops: chunk + .ops + .iter() + .map(|op| { + let record = op.cid().and_then(|cid| chunk.blocks.get(&cid)).cloned(); + (op.clone(), record) + }) + .collect(), + blocks, + } + } + + fn body(&self) -> Result { + let ops = self + .ops + .iter() + .map(|(op, record)| op_json(op, record.as_deref())) + .collect::, _>>()?; + let mut body = json!({ + "space": self.repo, + "rev": self.rev, + "ops": ops, + "blocks": { + "$bytes": base64::engine::general_purpose::STANDARD_NO_PAD.encode(&self.blocks) + }, + }); + if let Some(prev) = &self.prev_rev { + body["prevRev"] = json!(prev); + } + Ok(body) + } +} + +fn op_json(op: &Op, record: Option<&[u8]>) -> Result { + let (collection, rkey) = op + .key() + .split_once('/') + .ok_or_else(|| format!("chain key {:?} has no collection", op.key()))?; + let action = match op { + Op::Create { .. } => "create", + Op::Update { .. } => "update", + Op::Delete { .. } => "delete", + }; + let mut body = json!({ "action": action, "collection": collection, "rkey": rkey }); + if let Some(cid) = op.cid() { + body["cid"] = json!(cid.to_string()); + let record = record.ok_or_else(|| format!("{} has no record block", op.key()))?; + body["value"] = crate::atproto::record_value(record).map_err(|error| error.to_string())?; + } + if let Some(prev) = op.prev() { + body["prev"] = json!(prev.to_string()); + } + Ok(body) +} + +pub enum Notice { + Write(WriteNotice), + SpaceDeleted(RepoDid), + LeaseLapsed(Vec), +} + +impl Notice { + fn method(&self) -> Nsid { + method(match self { + Notice::Write(_) => NOTIFY_WRITE, + Notice::SpaceDeleted(_) => NOTIFY_SPACE_DELETED, + Notice::LeaseLapsed(_) => NOTIFY_LEASE_LAPSED, + }) + } + + fn body(&self) -> Result { + match self { + Notice::Write(write) => write.body(), + Notice::SpaceDeleted(repo) => Ok(json!({ "space": repo })), + Notice::LeaseLapsed(repos) => Ok(json!({ "spaces": repos })), + } + } +} + +pub struct NotifyQueue { + notices: mpsc::UnboundedSender, + watching_writes: bool, + settled: AtomicUsize, +} + +impl NotifyQueue { + pub fn channel(syncers: &Syncers) -> (Arc, mpsc::UnboundedReceiver) { + let (notices, jobs) = mpsc::unbounded_channel(); + ( + Arc::new(Self { + notices, + watching_writes: !recipients(syncers, &method(NOTIFY_WRITE)).is_empty(), + settled: AtomicUsize::new(0), + }), + jobs, + ) + } + + // callers still hold cob locks so this only queues and never waits on the network + pub(crate) fn write(&self, notice: WriteNotice) { + let _ = self.notices.send(Notice::Write(notice)); + } + + pub(crate) fn space_deleted(&self, repo: RepoDid) { + let _ = self.notices.send(Notice::SpaceDeleted(repo)); + } + + pub(crate) fn lease_lapsed(&self, repos: Vec) { + let _ = self.notices.send(Notice::LeaseLapsed(repos)); + } + + // private writes check this to skip copying blocks nobody wants + pub(crate) fn watching_writes(&self) -> bool { + self.watching_writes + } + + // finished deliveries plus notices nobody got so tests can wait on it instead of sleeping + pub fn settled(&self) -> usize { + self.settled.load(Ordering::SeqCst) + } + + fn settle(&self) { + self.settled.fetch_add(1, Ordering::SeqCst); + } +} + +fn recipients(syncers: &Syncers, method: &Nsid) -> Vec { + syncers + .iter() + .filter(|syncer| syncer.allows(method)) + .map(|syncer| syncer.did().clone()) + .collect() +} + +struct Delivery { + method: Nsid, + body: Value, +} + +async fn deliver( + state: &XrpcState, + syncer: &ServiceDid, + delivery: &Delivery, +) { + let signer = match state.secrets.signer(&state.knot_did) { + Ok(signer) => signer, + Err(error) => { + tracing::error!(%error, "the knot signer is unavailable, so syncers aren't notified"); + return; + } + }; + let sent = state.atproto.notify_syncer( + &signer, + &*state.entropy, + syncer, + &delivery.method, + &delivery.body, + ); + match tokio::time::timeout(DELIVERY_TIMEOUT, sent).await { + Ok(Ok(())) => {} + Ok(Err(error)) => { + tracing::warn!(%syncer, method = %delivery.method, %error, "a syncer wasn't notified") + } + Err(_) => tracing::warn!(%syncer, method = %delivery.method, "a syncer timed out"), + } +} + +// use a queue per syncer, a dead one only blocks itself +struct Lanes { + senders: HashMap>>, + tasks: JoinSet<()>, +} + +impl Lanes { + fn send( + &mut self, + state: &Arc>, + syncer: &ServiceDid, + delivery: &Arc, + ) { + let lane = self + .senders + .entry(syncer.as_str().to_owned()) + .or_insert_with(|| { + let (sender, mut queued) = mpsc::channel::>(LANE_DEPTH); + let (state, syncer) = (Arc::clone(state), syncer.clone()); + self.tasks.spawn(async move { + while let Some(delivery) = queued.recv().await { + deliver(&state, &syncer, &delivery).await; + state.notify.settle(); + } + }); + sender + }); + if lane.try_send(Arc::clone(delivery)).is_err() { + tracing::warn!(%syncer, "a syncer's notify lane is full or gone, so it recovers through listRepoOps"); + state.notify.settle(); + } + } +} + +async fn dispatch( + state: &Arc>, + lanes: &mut Lanes, + notice: Notice, +) { + let method = notice.method(); + let syncers = recipients(&state.syncers, &method); + let uncovered = match ¬ice { + Notice::Write(write) => { + state.index.lease_covers(&write.repo, state.now()) != Resolved::Ready(true) + } + Notice::SpaceDeleted(_) | Notice::LeaseLapsed(_) => false, + }; + if syncers.is_empty() || uncovered { + state.notify.settle(); + return; + } + let body = match notice.body() { + Ok(body) => body, + Err(error) => { + tracing::error!(%error, %method, "a notice wouldn't encode, so no syncer gets it"); + state.notify.settle(); + return; + } + }; + let delivery = Arc::new(Delivery { method, body }); + for syncer in &syncers { + lanes.send(state, syncer, &delivery); + } +} + +pub async fn run( + state: Arc>, + mut notices: mpsc::UnboundedReceiver, + mut shutdown: watch::Receiver, +) { + let mut lanes = Lanes { + senders: HashMap::new(), + tasks: JoinSet::new(), + }; + loop { + tokio::select! { + notice = notices.recv() => match notice { + Some(notice) => dispatch(&state, &mut lanes, notice).await, + None => return, + }, + _ = shutdown.changed() => return, + } + } +} + +fn read_covered( + state: &XrpcState, +) -> Result>, String> { + let meta = state + .repos + .open_meta(&state.knot_did) + .map_err(|error| error.to_string())?; + let stored = + crate::plc::load::>(&meta, COVERED_REF).map_err(|error| error.to_string())?; + Ok(stored.map(|repos| repos.into_iter().collect())) +} + +fn write_covered( + state: &XrpcState, + covered: &BTreeSet, +) -> Result<(), String> { + let repos: Vec<&RepoDid> = covered.iter().collect(); + let meta = state + .repos + .open_meta(&state.knot_did) + .map_err(|error| error.to_string())?; + crate::plc::store(&meta, COVERED_REF, Some(&repos)).map_err(|error| error.to_string()) +} + +// public repos are still readable and deleted ones already got their own notice +fn lapsed_for_real(state: &XrpcState, repo: &RepoDid) -> bool { + state.index.visibility_of(repo) == Resolved::Ready(Some(Visibility::Private)) +} + +#[derive(Default)] +pub struct SweepMemory { + stored: Option>, + seen: Option>, +} + +// the covered set is kept on a meta ref so lapses from while the knot was down still get reported +pub async fn sweep( + state: &Arc>, + memory: &mut SweepMemory, +) { + let Resolved::Ready(covered) = state.index.covered_private_repos(state.now()) else { + return; + }; + if memory.stored.is_none() { + let reading = Arc::clone(state); + let read = tokio::task::spawn_blocking(move || read_covered(&reading)) + .await + .map_err(|error| error.to_string()) + .and_then(|read| read); + match read { + Ok(stored) => memory.stored = Some(stored.unwrap_or_default()), + Err(error) => { + tracing::warn!(%error, "the covered repos wouldn't read, will retry before reporting any lapse"); + return; + } + } + } + let previous = memory + .seen + .take() + .or_else(|| memory.stored.clone()) + .unwrap_or_default(); + // a repo whose readers won't load stays covered because we can't tell if it lapsed + let unknown = covered + .unknown + .into_iter() + .filter(|repo| previous.contains(repo)); + let covered: BTreeSet = covered + .live + .into_iter() + .map(|(repo, _)| repo) + .chain(unknown) + .collect(); + let lapsed: Vec = previous + .difference(&covered) + .filter(|repo| lapsed_for_real(state, repo)) + .cloned() + .collect(); + for repos in lapsed.chunks(MAX_LAPSED_PER_NOTICE) { + state.notify.lease_lapsed(repos.to_vec()); + } + memory.seen = Some(covered.clone()); + if memory.stored.as_ref() != Some(&covered) { + let writing = Arc::clone(state); + let target = covered.clone(); + let written = tokio::task::spawn_blocking(move || write_covered(&writing, &target)) + .await + .map_err(|error| error.to_string()) + .and_then(|written| written); + match written { + Ok(()) => memory.stored = Some(covered), + Err(error) => { + tracing::warn!(%error, "the covered repos wouldn't persist, will retry"); + } + } + } +} + +pub async fn run_sweeper( + state: Arc>, + shutdown: CancellationToken, +) { + let mut tick = tokio::time::interval(SWEEP_INTERVAL); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let _first = tick.tick().await; + let mut memory = SweepMemory::default(); + loop { + tokio::select! { + () = shutdown.cancelled() => return, + _ = tick.tick() => sweep(&state, &mut memory).await, + } + } +} diff --git a/knot2/crates/knot-xrpc/src/repos.rs b/knot2/crates/knot-xrpc/src/repos.rs index 5e124de7d..e67c1c273 100644 --- a/knot2/crates/knot-xrpc/src/repos.rs +++ b/knot2/crates/knot-xrpc/src/repos.rs @@ -388,7 +388,7 @@ pub(crate) async fn create_repo( let knot_signer = state.secrets.signer(&state.knot_did)?; let repos = state.repos.clone(); let cob_locks = Arc::clone(&state.cob_locks); - let feed = crate::firehose::feed_for(&state.firehose, visibility); + let feed = crate::firehose::feed_for(&state.firehose, &state.notify, visibility); let born = Arc::clone(&feed); let did = repo_did.clone(); let initialized = created.clone(); @@ -771,7 +771,7 @@ pub(crate) async fn delete_repo( }) .await?; - feed.publish_account(&repo_did, false, Some("deleted"), state.now()); + feed.publish_deleted(&repo_did, state.now()); Ok(ok_empty()) } diff --git a/knot2/crates/knot-xrpc/src/tests.rs b/knot2/crates/knot-xrpc/src/tests.rs index 30e6a81a6..10b0a8242 100644 --- a/knot2/crates/knot-xrpc/src/tests.rs +++ b/knot2/crates/knot-xrpc/src/tests.rs @@ -404,6 +404,7 @@ fn state_from( )), rkeys: Default::default(), syncers: Default::default(), + notify: crate::notify::NotifyQueue::channel(&Default::default()).0, }); (state, clock) } diff --git a/knot2/crates/knot-xrpc/tests/common/mod.rs b/knot2/crates/knot-xrpc/tests/common/mod.rs index aa86ddfee..7eb3054a0 100644 --- a/knot2/crates/knot-xrpc/tests/common/mod.rs +++ b/knot2/crates/knot-xrpc/tests/common/mod.rs @@ -45,7 +45,7 @@ pub async fn serve_local_plc(world: &World) -> std::net::SocketAddr { let endpoint = endpoint.clone(); async move { let did = Did::new_owned(did).expect("the plc request path is a DID"); - let doc = did_doc(&did, &multikey, Some(&endpoint)); + let doc = did_doc(&did, &multikey, Some(&endpoint), &[]); let mut headers = axum::http::HeaderMap::new(); headers.insert( axum::http::header::CONTENT_TYPE, @@ -445,6 +445,14 @@ pub fn fixture_record(created_at: UnixSeconds) -> knot_types::RecordBody { pub type Responder = Box Result + Send + Sync>; +pub struct Sent { + pub url: url::Url, + pub authorization: Option, + pub body: serde_json::Value, +} + +type Published = std::collections::HashMap>; + struct Terms { admins: BTreeSet, policy: knot_types::ContributionPolicy, @@ -472,9 +480,23 @@ pub struct World { pub router: Router, pub state: Arc, ManualClock>>, plc_pds: Arc>>, + services: Arc>, + sent: Arc>>, } impl World { + pub fn publish_service(&self, did: &Did, id: &str, endpoint: &str) { + self.services + .lock() + .unwrap() + .entry(did.clone()) + .or_default() + .push((id.to_string(), endpoint.to_string())); + } + + pub fn sent(&self) -> Vec { + std::mem::take(&mut *self.sent.lock().unwrap()) + } pub fn publish_plc_pds(&self, did: &Did, pds: &KnotServiceUrl) { self.plc_pds .lock() @@ -567,6 +589,25 @@ impl World { ) } + pub fn declaring_administered(syncers: Syncers) -> Self { + Self::with_terms(Terms { + admins: BTreeSet::from([owner()]), + syncers, + ..Terms::default() + }) + } + + fn with_terms(terms: Terms) -> Self { + Self::assemble( + true, + ByteLimits::default(), + ObjectFormat::SHA1, + LimitConfig::default(), + true, + terms, + ) + } + fn build(rebuilt: bool, byte_limits: ByteLimits, object_format: ObjectFormat) -> Self { Self::build_with_limits(rebuilt, byte_limits, object_format, LimitConfig::default()) } @@ -629,7 +670,26 @@ impl World { let plc_pds = Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())); let pds_of = Arc::clone(&plc_pds); + let services = Arc::new(std::sync::Mutex::new(Published::new())); + let services_of = Arc::clone(&services); + let sent = Arc::new(std::sync::Mutex::new(Vec::new())); + let sent_of = Arc::clone(&sent); let responder: Responder = Box::new(move |request| { + if request.method == http::Method::POST { + sent_of.lock().unwrap().push(Sent { + url: request.url.clone(), + authorization: request + .headers + .get(header::AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + .map(str::to_string), + body: request + .body + .as_ref() + .and_then(|body| serde_json::from_slice(body).ok()) + .unwrap_or_default(), + }); + } let host = request.url.host_str().unwrap_or_default(); let did = match host { "plc.directory" => Some(request.url.path().trim_start_matches('/').to_string()), @@ -650,6 +710,12 @@ impl World { .cloned() .or_else(|| (!did.as_str().starts_with("did:web:")).then(default_pds)) .as_ref(), + &services_of + .lock() + .unwrap() + .get(&did) + .cloned() + .unwrap_or_default(), ), None => Bytes::new(), }; @@ -690,6 +756,7 @@ impl World { let lfs = lfs_handle.map(|handle| knot_xrpc::LfsWeb::new(knot_acl::Lfs::new(handle), 8)); let (imports, import_jobs) = knot_xrpc::imports::ImportQueue::channel(); + let (notify, notices) = knot_xrpc::notify::NotifyQueue::channel(&syncers); let shutdown = tokio::sync::watch::channel(false); let state = Arc::new(XrpcState { imports, @@ -758,11 +825,17 @@ impl World { )), rkeys: Default::default(), syncers, + notify, }); // the runner lives like it does in the server, so wakes from tests are processed tokio::spawn(knot_xrpc::imports::run( Arc::clone(&state), import_jobs, + shutdown.1.clone(), + )); + tokio::spawn(knot_xrpc::notify::run( + Arc::clone(&state), + notices, shutdown.1, )); let did_document = knot_atproto::knot_did_document( @@ -786,6 +859,8 @@ impl World { router, state, plc_pds, + services, + sent, } } @@ -993,7 +1068,16 @@ fn pds_service(pds: Option<&KnotServiceUrl>) -> Vec { .collect() } -fn did_doc(did: &Did, multikey: &ActorId, pds: Option<&KnotServiceUrl>) -> serde_json::Value { +fn did_doc( + did: &Did, + multikey: &ActorId, + pds: Option<&KnotServiceUrl>, + extra: &[(String, String)], +) -> serde_json::Value { + let mut services = pds_service(pds); + services.extend(extra.iter().map(|(id, endpoint)| { + serde_json::json!({ "id": id, "type": "TangledSyncer", "serviceEndpoint": endpoint }) + })); serde_json::json!({ "id": did.as_str(), "alsoKnownAs": [], @@ -1003,18 +1087,18 @@ fn did_doc(did: &Did, multikey: &ActorId, pds: Option<&KnotServiceUrl>) -> serde "controller": did.as_str(), "publicKeyMultibase": multikey.as_str() }], - "service": pds_service(pds) + "service": services }) } -fn did_doc_at(did: &Did, pds: Option<&KnotServiceUrl>) -> Bytes { +fn did_doc_at(did: &Did, pds: Option<&KnotServiceUrl>, extra: &[(String, String)]) -> Bytes { let sec1 = actor_key() .verifying_key() .to_encoded_point(true) .as_bytes() .to_vec(); let multikey = ActorId::new(knot_types::crypto::multikey(0xe7, &sec1)).unwrap(); - let doc = did_doc(did, &multikey, pds); + let doc = did_doc(did, &multikey, pds, extra); Bytes::from(serde_json::to_vec(&doc).unwrap()) } @@ -1101,6 +1185,7 @@ pub fn repo_did_document(world: &World, did: &RepoDid) -> serde_json::Value { &Did::new_owned(did.as_str()).unwrap(), &knot_multikey(world), Some(&world.state.knot_service_url), + &[], ) } diff --git a/knot2/crates/knot-xrpc/tests/notify.rs b/knot2/crates/knot-xrpc/tests/notify.rs new file mode 100644 index 000000000..9b19b792d --- /dev/null +++ b/knot2/crates/knot-xrpc/tests/notify.rs @@ -0,0 +1,438 @@ +mod common; + +use std::time::Duration; + +use base64::Engine as _; +use base64::engine::general_purpose::URL_SAFE_NO_PAD; +use jacquard_common::types::tid::Tid; +use k256::ecdsa::signature::Verifier; +use knot_git::{RefUpdate, Repo}; +use knot_types::{ + AccountDid, Did, RecordRkey, RefName, RepoDid, RepoName, RepoRkey, Syncers, UnixSeconds, + Visibility, +}; + +use common::{Sent, World, declared, nsid, owner, post_authed}; +use knot_xrpc::notify::SweepMemory; + +const WRITE: &str = "org.tangled.space.notifyWrite"; +const DELETED: &str = "org.tangled.space.notifySpaceDeleted"; +const LAPSED: &str = "org.tangled.knot.notifyLeaseLapsed"; +const DESCRIBE: &str = "com.atproto.repo.describeRepo"; + +fn bobbin() -> Did { + Did::new_owned("did:web:bobbin.example").unwrap() +} + +fn deliberi() -> Did { + Did::new_owned("did:web:deliberi.example").unwrap() +} + +fn squid() -> AccountDid { + AccountDid::new("did:plc:squid").unwrap() +} + +// every syncer publishes its endpoint at https://, like did:web +fn world_with(syncers: &[(&Did, &[&str])]) -> World { + let declared = syncers + .iter() + .map(|(did, lxm)| declared(did.as_str(), lxm)) + .collect(); + let world = World::declaring_administered(Syncers::new(declared).unwrap()); + for (did, _) in syncers { + let host = did.as_str().strip_prefix("did:web:").unwrap(); + world.publish_service(did, "#tangled_syncer", &format!("https://{host}")); + } + world +} + +fn repo(world: &World, name: &str, visibility: Visibility) -> RepoDid { + let did = RepoDid::new(format!("did:plc:{name}fixture")).unwrap(); + world.layout.create(&did).unwrap(); + world.register_as( + &did, + &RepoName::new(name).unwrap(), + &RepoRkey::new(name).unwrap(), + visibility, + ); + world.state.index.ensure_collaborators(&did).unwrap(); + did +} + +fn write(world: &World, did: &RepoDid, rkeys: &[&str], at: i64) { + let path = world.layout.repo_path(did).unwrap(); + let rkeys: Vec = rkeys + .iter() + .map(|rkey| RecordRkey::new(*rkey).unwrap()) + .collect(); + world.publish(did, &path, &rkeys, UnixSeconds::new(at)); +} + +async fn settled(world: &World, count: usize) { + tokio::time::timeout(Duration::from_secs(10), async { + while world.state.notify.settled() < count { + tokio::time::sleep(Duration::from_millis(1)).await; + } + }) + .await + .expect("the notifier settles"); +} + +// for asserting nothing got sent, there's no signal to wait on +async fn quiet(world: &World) { + tokio::time::sleep(Duration::from_millis(50)).await; + assert!(world.sent().is_empty(), "a notification went out"); +} + +fn endpoint(method: &str, host: &str) -> String { + format!("https://{host}/xrpc/{method}") +} + +fn covered_owner(world: &World) { + world.state.index.renew_lease(&owner(), world.state.now()); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_private_write_goes_to_a_covered_syncer_that_lists_the_nsid() { + let (bobbin, deliberi) = (bobbin(), deliberi()); + let world = world_with(&[(&bobbin, &[WRITE]), (&deliberi, &[DESCRIBE])]); + let hidden = repo(&world, "hidden", Visibility::Private); + covered_owner(&world); + + write(&world, &hidden, &["reef"], 1_000); + settled(&world, 1).await; + + let sent = world.sent(); + assert_eq!( + sent.len(), + 1, + "only the syncer that lists the nsid hears it" + ); + assert_eq!(sent[0].url.as_str(), endpoint(WRITE, "bobbin.example")); + let first = &sent[0].body; + assert_eq!(first["space"], hidden.as_str()); + let rev: Tid = first["rev"].as_str().unwrap().parse().unwrap(); + assert!( + first.get("prevRev").is_none(), + "the first commit has no previous rev" + ); + let ops = first["ops"].as_array().unwrap(); + assert_eq!(ops.len(), 1); + assert_eq!(ops[0]["action"], "create"); + assert_eq!(ops[0]["collection"], "org.tangled.test.record"); + assert_eq!(ops[0]["rkey"], "reef"); + assert!(ops[0]["cid"].as_str().unwrap().starts_with("bafy")); + assert_eq!(ops[0]["value"]["$type"], "org.tangled.test.record"); + assert!(ops[0].get("prev").is_none()); + assert!(!first["blocks"]["$bytes"].as_str().unwrap().is_empty()); + + write(&world, &hidden, &["reef"], 2_000); + settled(&world, 2).await; + let sent = world.sent(); + let second = &sent[0].body; + assert_eq!(second["prevRev"], rev.as_str()); + let op = &second["ops"][0]; + assert_eq!(op["action"], "update"); + assert_eq!(op["prev"], ops[0]["cid"]); + assert_ne!(op["cid"], ops[0]["cid"]); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_delete_op_carries_no_value() { + let bobbin = bobbin(); + let world = world_with(&[(&bobbin, &[WRITE])]); + let hidden = repo(&world, "hidden", Visibility::Private); + covered_owner(&world); + write(&world, &hidden, &["reef"], 1_000); + settled(&world, 1).await; + world.sent(); + + write(&world, &hidden, &[], 2_000); + settled(&world, 2).await; + + let op = &world.sent()[0].body["ops"][0]; + assert_eq!(op["action"], "delete"); + assert_eq!(op["rkey"], "reef"); + assert!(op.get("cid").is_none() && op.get("value").is_none()); + assert!(op["prev"].is_string()); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_write_with_no_covering_lease_is_not_sent() { + let bobbin = bobbin(); + let world = world_with(&[(&bobbin, &[WRITE])]); + let hidden = repo(&world, "hidden", Visibility::Private); + + write(&world, &hidden, &["reef"], 1_000); + settled(&world, 1).await; + + quiet(&world).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_public_repo_write_is_not_sent() { + let bobbin = bobbin(); + let world = world_with(&[(&bobbin, &[WRITE])]); + let open = repo(&world, "open", Visibility::Public); + covered_owner(&world); + + write(&world, &open, &["reef"], 1_000); + + quiet(&world).await; + assert_eq!(world.state.notify.settled(), 0, "nothing was even queued"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_write_nobody_listens_for_is_not_sent() { + let deliberi = deliberi(); + let world = world_with(&[(&deliberi, &[DESCRIBE])]); + let hidden = repo(&world, "hidden", Visibility::Private); + covered_owner(&world); + + write(&world, &hidden, &["reef"], 1_000); + + quiet(&world).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_syncer_without_an_endpoint_entry_is_skipped() { + let (bobbin, deliberi) = (bobbin(), deliberi()); + let declared = [&bobbin, &deliberi].map(|did| declared(did.as_str(), &[WRITE])); + let world = World::declaring_administered(Syncers::new(declared.to_vec()).unwrap()); + world.publish_service(&bobbin, "#tangled_syncer", "https://bobbin.example"); + let hidden = repo(&world, "hidden", Visibility::Private); + covered_owner(&world); + + write(&world, &hidden, &["reef"], 1_000); + settled(&world, 2).await; + + let sent = world.sent(); + assert_eq!( + sent.len(), + 1, + "deliberi lists no endpoint, so it is skipped" + ); + assert_eq!(sent[0].url.host_str(), Some("bobbin.example")); +} + +async fn delete(world: &World, did: &RepoDid) { + let (status, body) = post_authed( + world, + &nsid("org.tangled.repo.deleteRepo"), + &owner(), + serde_json::json!({ "repo": did, "force": true }), + ) + .await; + assert_eq!(status, http::StatusCode::OK, "{body}"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn deleting_a_private_repo_sends_space_deleted_without_asking_for_a_lease() { + let (bobbin, deliberi) = (bobbin(), deliberi()); + let world = world_with(&[(&bobbin, &[DELETED]), (&deliberi, &[WRITE])]); + let hidden = repo(&world, "hidden", Visibility::Private); + + delete(&world, &hidden).await; + settled(&world, 1).await; + + let sent = world.sent(); + assert_eq!(sent.len(), 1); + assert_eq!(sent[0].url.as_str(), endpoint(DELETED, "bobbin.example")); + assert_eq!(sent[0].body, serde_json::json!({ "space": hidden })); +} + +#[tokio::test(flavor = "multi_thread")] +async fn deleting_a_public_repo_notifies_no_syncer() { + let bobbin = bobbin(); + let world = world_with(&[(&bobbin, &[DELETED])]); + let open = repo(&world, "open", Visibility::Public); + + delete(&world, &open).await; + + quiet(&world).await; +} + +fn lease_for_a_collaborator(world: &World, hidden: &RepoDid) { + world.add_collaborator(hidden, &squid(), &owner(), UnixSeconds::new(1_000)); + world.state.index.renew_lease(&squid(), world.state.now()); +} + +async fn sweep(world: &World, memory: &mut SweepMemory) { + knot_xrpc::notify::sweep(&world.state, memory).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_lapsed_lease_is_reported_once() { + let bobbin = bobbin(); + let world = world_with(&[(&bobbin, &[LAPSED])]); + let hidden = repo(&world, "hidden", Visibility::Private); + lease_for_a_collaborator(&world, &hidden); + let mut last = SweepMemory::default(); + + sweep(&world, &mut last).await; + quiet(&world).await; + + world.remove_collaborator(&hidden, &squid(), UnixSeconds::new(2_000)); + sweep(&world, &mut last).await; + settled(&world, 1).await; + let sent = world.sent(); + assert_eq!(sent.len(), 1); + assert_eq!(sent[0].url.as_str(), endpoint(LAPSED, "bobbin.example")); + assert_eq!(sent[0].body, serde_json::json!({ "spaces": [hidden] })); + + sweep(&world, &mut last).await; + quiet(&world).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_lapse_while_the_knot_was_down_is_still_reported() { + let bobbin = bobbin(); + let world = world_with(&[(&bobbin, &[LAPSED])]); + let hidden = repo(&world, "hidden", Visibility::Private); + lease_for_a_collaborator(&world, &hidden); + sweep(&world, &mut SweepMemory::default()).await; + + world.remove_collaborator(&hidden, &squid(), UnixSeconds::new(2_000)); + sweep(&world, &mut SweepMemory::default()).await; + settled(&world, 1).await; + + assert_eq!( + world.sent()[0].body, + serde_json::json!({ "spaces": [hidden] }) + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_deleted_repo_is_not_reported_as_lapsed() { + let bobbin = bobbin(); + let world = world_with(&[(&bobbin, &[DELETED, LAPSED])]); + let hidden = repo(&world, "hidden", Visibility::Private); + lease_for_a_collaborator(&world, &hidden); + let mut last = SweepMemory::default(); + sweep(&world, &mut last).await; + + delete(&world, &hidden).await; + settled(&world, 1).await; + sweep(&world, &mut last).await; + + let sent = world.sent(); + assert_eq!(sent.len(), 1, "{} sends", sent.len()); + assert_eq!(sent[0].url.as_str(), endpoint(DELETED, "bobbin.example")); +} + +const COVERED_REF: &str = "refs/knot/notify-covered"; + +fn covered_ref() -> RefName { + RefName::new(COVERED_REF).unwrap() +} + +#[tokio::test(flavor = "multi_thread")] +async fn an_unreadable_stored_set_is_kept_and_a_later_read_still_reports_the_lapse() { + let bobbin = bobbin(); + let world = world_with(&[(&bobbin, &[LAPSED])]); + let hidden = repo(&world, "hidden", Visibility::Private); + lease_for_a_collaborator(&world, &hidden); + sweep(&world, &mut SweepMemory::default()).await; + + let meta = Repo::open(&world.state.meta_path).unwrap(); + let good = meta.find_ref(&covered_ref()).unwrap().unwrap(); + let junk = meta.git().write_blob(b"{ not a set").unwrap().detach(); + let junk = knot_types::Oid::from(junk); + meta.update_ref(&RefUpdate::Update { + name: covered_ref(), + old: good, + new: junk, + }) + .unwrap(); + + world.remove_collaborator(&hidden, &squid(), UnixSeconds::new(2_000)); + let mut memory = SweepMemory::default(); + sweep(&world, &mut memory).await; + quiet(&world).await; + assert_eq!(meta.find_ref(&covered_ref()).unwrap(), Some(junk)); + + meta.update_ref(&RefUpdate::Update { + name: covered_ref(), + old: junk, + new: good, + }) + .unwrap(); + sweep(&world, &mut memory).await; + settled(&world, 1).await; + assert_eq!( + world.sent()[0].body, + serde_json::json!({ "spaces": [hidden] }) + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_failed_write_is_retried_on_the_next_sweep() { + let bobbin = bobbin(); + let world = world_with(&[(&bobbin, &[LAPSED])]); + let hidden = repo(&world, "hidden", Visibility::Private); + lease_for_a_collaborator(&world, &hidden); + let meta = Repo::open(&world.state.meta_path).unwrap(); + let blob = meta.git().write_blob(b"[]").unwrap().detach(); + let blocker = RefName::new(format!("{COVERED_REF}/blocker")).unwrap(); + meta.update_ref(&RefUpdate::Create { + name: blocker.clone(), + new: knot_types::Oid::from(blob), + }) + .unwrap(); + + let mut memory = SweepMemory::default(); + sweep(&world, &mut memory).await; + assert_eq!(meta.find_ref(&covered_ref()).unwrap(), None); + + meta.update_ref(&RefUpdate::Delete { + name: blocker, + old: knot_types::Oid::from(blob), + }) + .unwrap(); + sweep(&world, &mut memory).await; + assert!(meta.find_ref(&covered_ref()).unwrap().is_some()); + + world.remove_collaborator(&hidden, &squid(), UnixSeconds::new(2_000)); + sweep(&world, &mut SweepMemory::default()).await; + settled(&world, 1).await; +} + +fn claims_of(sent: &Sent) -> (serde_json::Value, String, Vec) { + let token = sent + .authorization + .as_deref() + .and_then(|header| header.strip_prefix("Bearer ")) + .expect("the notification carries a bearer token"); + let (signing_input, signature) = token.rsplit_once('.').unwrap(); + let payload = signing_input.split_once('.').unwrap().1; + let claims = serde_json::from_slice(&URL_SAFE_NO_PAD.decode(payload).unwrap()).unwrap(); + ( + claims, + signing_input.to_string(), + URL_SAFE_NO_PAD.decode(signature).unwrap(), + ) +} + +#[tokio::test(flavor = "multi_thread")] +async fn the_notification_is_a_service_token_from_the_knot_to_the_syncer() { + let bobbin = bobbin(); + let world = world_with(&[(&bobbin, &[DELETED])]); + let hidden = repo(&world, "hidden", Visibility::Private); + delete(&world, &hidden).await; + settled(&world, 1).await; + let sent = world.sent(); + let (claims, signing_input, signature) = claims_of(&sent[0]); + + assert_eq!(claims["iss"], world.state.knot_did.as_str()); + assert_eq!(claims["aud"], bobbin.as_str()); + assert_eq!(claims["lxm"], DELETED); + assert!(claims["jti"].as_str().is_some_and(|jti| !jti.is_empty())); + let lifetime = claims["exp"].as_i64().unwrap() - claims["iat"].as_i64().unwrap(); + assert!((0..=300).contains(&lifetime), "a short lived token"); + let signer = world.state.secrets.signer(&world.state.knot_did).unwrap(); + let public = knot_runtime::Signer::public_key(&signer); + let key = k256::ecdsa::VerifyingKey::from_sec1_bytes(public.as_bytes()).unwrap(); + let signature = k256::ecdsa::Signature::from_slice(&signature).unwrap(); + key.verify(signing_input.as_bytes(), &signature) + .expect("the knot's own key signed it"); +}