//! events go through the outbox. one the spindle refuses is retried by the next `post-activate`.
pub(crate) mod outbox;
use std::time::Duration;
use knot_atproto::{Atproto, AtprotoError, ServiceAuth, SpindleAnswer};
use knot_git::RefClass;
use knot_runtime::{OsEntropy, ReqwestHttp, SystemClock};
use knot_types::{AccountDid, RepoDid, ServiceDid};
use tokio::time::Instant;
use crate::config::ValidatedConfig;
use crate::crawl;
use outbox::{Entry, Outbox, Outcome};
const REPLY_WAIT: Duration = Duration::from_secs(2);
const MAX_LINES: usize = 10;
const MAX_CHARS: usize = 1000;
/// events are a few hundred bytes. this only cuts off ref names built to bloat the queue.
const MAX_EVENT_BYTES: usize = 8 * 1024;
const SERVED_WAIT: Duration = Duration::from_secs(2);
const LEASE_MARGIN: u64 = 30;
const DRAIN_LINGER: Duration = Duration::from_secs(60);
pub(crate) struct RefMove {
name: String,
old: Option<String>,
new: Option<String>,
}
pub(crate) fn moves(before: &str, after: &str) -> Vec<RefMove> {
let parse = |snapshot: &'_ str| -> std::collections::BTreeMap<String, String> {
snapshot
.lines()
.filter_map(|line| line.split_once(' '))
.map(|(oid, name)| (name.to_owned(), oid.to_owned()))
.collect()
};
let (before, after) = (parse(before), parse(after));
let names: std::collections::BTreeSet<_> = before.keys().chain(after.keys()).collect();
names
.into_iter()
.filter(|name| before.get(*name) != after.get(*name))
.map(|name| RefMove {
name: name.clone(),
old: before.get(name).cloned(),
new: after.get(name).cloned(),
})
.collect()
}
fn push_event(
null: &str,
repo: &RepoDid,
actor: &AccountDid,
moved: &RefMove,
) -> serde_json::Value {
serde_json::json!({
"repo": repo,
"event": {
"$type": "org.tangled.event.push",
"repo": repo,
"ref": moved.name,
"oldSha": moved.old.as_deref().unwrap_or(null),
"newSha": moved.new.as_deref().unwrap_or(null),
"actor": actor,
},
})
}
fn events(
null: &str,
repo: &RepoDid,
actor: &AccountDid,
moved: &[RefMove],
) -> Vec<(String, serde_json::Value)> {
moved
.iter()
.filter(|moved| RefClass::of_revspec(&moved.name) == RefClass::Public)
.map(|moved| (moved.name.clone(), push_event(null, repo, actor, moved)))
.filter(|(_, body)| body.to_string().len() <= MAX_EVENT_BYTES)
.collect()
}
async fn enqueue(
config: &ValidatedConfig,
repo: &RepoDid,
actor: &AccountDid,
moved: &[RefMove],
) -> anyhow::Result<Vec<u64>> {
let Some(spindle) = spindle_of(config, repo).await? else {
return Ok(Vec::new());
};
let null = config.object_format.null_oid().to_hex();
let events = events(&null, repo, actor, moved);
if events.is_empty() {
return Ok(Vec::new());
}
let now = crate::atproto::unix_now().get();
Outbox::at(&config.output_root).update(|queue| queue.enqueue(&spindle, repo, events, now))
}
async fn spindle_of(
config: &ValidatedConfig,
repo: &RepoDid,
) -> anyhow::Result<Option<ServiceDid>> {
crate::atproto::published_spindle(&config.output_root.join("atproto"), repo).await
}
#[derive(Default)]
pub(crate) struct Heard {
pub(crate) said: Vec<String>,
pub(crate) pending: bool,
}
pub(crate) async fn tell(
config: &ValidatedConfig,
published: &[String],
repo: &RepoDid,
actor: &AccountDid,
moved: &[RefMove],
) -> Heard {
if let Ok(None) = spindle_of(config, repo).await {
return Heard::default();
}
if crawl::wait_served(config, published, SERVED_WAIT).await {
return announce(config, repo, actor, moved).await;
}
Heard {
said: Vec::new(),
pending: defer(config, repo, actor, moved).await,
}
}
pub(crate) async fn defer(
config: &ValidatedConfig,
repo: &RepoDid,
actor: &AccountDid,
moved: &[RefMove],
) -> bool {
match enqueue(config, repo, actor, moved).await {
Ok(ids) => !ids.is_empty(),
Err(error) => {
tracing::warn!(%repo, "the spindle doesn't hear of this push: {error:#}");
false
}
}
}
async fn announce(
config: &ValidatedConfig,
repo: &RepoDid,
actor: &AccountDid,
moved: &[RefMove],
) -> Heard {
let ids = match enqueue(config, repo, actor, moved).await {
Ok(ids) if ids.is_empty() => return Heard::default(),
Ok(ids) => ids,
Err(error) => {
tracing::warn!(%repo, "the spindle doesn't hear of this push: {error:#}");
return Heard::default();
}
};
let replies = pump(config, Some(Instant::now() + REPLY_WAIT))
.await
.unwrap_or_else(|error| {
tracing::warn!(%repo, "the spindle didn't take this push yet: {error:#}");
Vec::new()
});
let mut said: Vec<String> = Vec::new();
for text in replies
.into_iter()
.filter(|(id, _)| ids.contains(id))
.filter_map(|(_, message)| message)
{
if !said.contains(&text) {
said.push(text);
}
}
let pending = Outbox::at(&config.output_root)
.update(|queue| !queue.is_empty())
.unwrap_or(true);
Heard {
said: said.iter().flat_map(|text| terminal_lines(text)).collect(),
pending,
}
}
pub(crate) async fn drain(config: &ValidatedConfig) {
let until = Instant::now() + DRAIN_LINGER;
loop {
if let Err(error) = pump(config, None).await {
tracing::warn!("couldn't deliver the queued push events: {error:#}");
return;
}
let next = Outbox::at(&config.output_root).update(|queue| queue.next_retry());
let Ok(Some(at)) = next else { return };
let wait = Duration::from_secs((at - crate::atproto::unix_now().get()).max(1) as u64);
if Instant::now() + wait > until {
return;
}
tokio::time::sleep(wait).await;
}
}
async fn pump(
config: &ValidatedConfig,
deadline: Option<Instant>,
) -> anyhow::Result<Vec<(u64, Option<String>)>> {
let outbox = Outbox::at(&config.output_root);
if outbox.update(|queue| queue.is_empty())? {
return Ok(Vec::new());
}
let signer = super::identity::load_knot_signer(config)?;
let limits = config.http_limits();
// a delivery makes two requests, the did document and the event.
let lease = (2 * limits.request_timeout.as_secs() + LEASE_MARGIN) as i64;
let atproto = config.atproto(ReqwestHttp::new(limits)?)?;
let auth = ServiceAuth::new(&signer, &OsEntropy);
let mut replies = Vec::new();
while deadline.is_none_or(|deadline| Instant::now() < deadline) {
let now = crate::atproto::unix_now().get();
let Some(claim) = outbox.update(|queue| queue.claim(now, lease))? else {
break;
};
if deadline.is_some_and(|deadline| Instant::now() >= deadline) {
// out of time to try. the event is released, not counted as a failure.
outbox.update(move |queue| queue.release(claim, now))?;
break;
}
let entry = &claim.entry;
let attempt = deliver(&atproto, &auth, entry);
let result = match deadline {
// a notify cut off here may have reached the spindle. it is sent again.
// spindles dedupe on repo, ref, old and new.
Some(deadline) => tokio::time::timeout_at(deadline, attempt)
.await
.unwrap_or(Err(Failure::Transient("no answer in time".into()))),
None => attempt.await,
};
let now = crate::atproto::unix_now().get();
let outcome = match result {
Ok(message) => {
replies.push((entry.id, message));
Outcome::Delivered
}
Err(Failure::Permanent(error)) => {
tracing::warn!(repo = entry.repo.as_str(), ref_name = entry.ref_name, %error, "the spindle rejected the push event, so it's dropped");
Outcome::Rejected
}
Err(Failure::Transient(error)) => {
tracing::warn!(repo = entry.repo.as_str(), ref_name = entry.ref_name, %error, "the spindle didn't take the push event, retrying");
Outcome::Retry
}
};
outbox.update(move |queue| queue.settle(claim, outcome, now))?;
}
Ok(replies)
}
enum Failure {
Permanent(String),
Transient(String),
}
impl From<AtprotoError> for Failure {
fn from(error: AtprotoError) -> Self {
let rejected = matches!(
&error,
AtprotoError::NotifyEvent { status, .. }
if (400..500).contains(&status.get()) && ![408, 429].contains(&status.get())
);
let error = error.to_string();
if rejected {
Self::Permanent(error)
} else {
Self::Transient(error)
}
}
}
async fn deliver(
atproto: &Atproto<ReqwestHttp, SystemClock>,
auth: &ServiceAuth<'_>,
entry: &Entry,
) -> Result<Option<String>, Failure> {
let answer = atproto
.notify_spindle(auth, &entry.spindle, entry.body.to_string().into())
.await?;
Ok(match answer {
SpindleAnswer::Scheduled { message, .. } => message,
SpindleAnswer::Nothing => None,
SpindleAnswer::Malformed(problem) => {
tracing::warn!(
spindle = entry.spindle.as_str(),
problem,
"the spindle's answer to the push wasn't readable, counting it delivered"
);
None
}
})
}
// spindle text is untrusted. only plain printable lines reach the terminal.
fn terminal_lines(text: &str) -> Vec<String> {
let mut plain = String::new();
let mut chars = text.chars().take(MAX_CHARS).peekable();
while let Some(c) = chars.next() {
match c {
'\u{1b}' => skip_escape(&mut chars),
'\n' => plain.push(c),
c if c.is_control() => {}
c => plain.push(c),
}
}
plain.lines().take(MAX_LINES).map(str::to_string).collect()
}
// csi ends on a byte from `@` to `~`, osc on bel or `ESC \`.
fn skip_escape(chars: &mut std::iter::Peekable<impl Iterator<Item = char>>) {
match chars.next() {
Some('[') => {
chars.find(|c| ('@'..='~').contains(c));
}
Some(']') if chars.find(|c| matches!(c, '\u{7}' | '\u{1b}')) == Some('\u{1b}') => {
chars.next_if_eq(&'\\');
}
_ => {}
}
}
#[cfg(test)]
mod tests {
use super::*;
const NULL: &str = "0000000000000000000000000000000000000000";
fn moved(name: &str, old: Option<&str>, new: Option<&str>) -> RefMove {
RefMove {
name: name.into(),
old: old.map(Into::into),
new: new.map(Into::into),
}
}
fn event(moved: &RefMove) -> serde_json::Value {
push_event(
NULL,
&RepoDid::new("did:plc:squid").unwrap(),
&AccountDid::new("did:plc:pusher").unwrap(),
moved,
)
}
#[test]
fn a_push_event_is_knot2s_body() {
let sha = "ab".repeat(20);
assert_eq!(
event(&moved("refs/heads/main", Some(NULL), Some(&sha))),
serde_json::json!({
"repo": "did:plc:squid",
"event": {
"$type": "org.tangled.event.push",
"repo": "did:plc:squid",
"ref": "refs/heads/main",
"oldSha": NULL,
"newSha": sha,
"actor": "did:plc:pusher",
},
})
);
}
#[test]
fn a_ref_that_did_not_exist_or_no_longer_does_has_the_null_oid() {
let sha = "ab".repeat(20);
let created = event(&moved("refs/tags/v1", None, Some(&sha)));
assert_eq!(created["event"]["oldSha"], NULL);
let deleted = event(&moved("refs/tags/v1", Some(&sha), None));
assert_eq!(deleted["event"]["newSha"], NULL);
}
#[test]
fn moves_are_the_refs_that_differ_between_snapshots() {
let before = "1111 refs/heads/main\n2222 refs/heads/gone\n3333 refs/tags/same\n";
let after = "4444 refs/heads/main\n3333 refs/tags/same\n5555 refs/heads/new\n";
let found: Vec<_> = moves(before, after)
.into_iter()
.map(|m| (m.name, m.old, m.new))
.collect();
let some = |oid: &str| Some(oid.to_string());
assert_eq!(
found,
[
("refs/heads/gone".to_string(), some("2222"), None),
("refs/heads/main".to_string(), some("1111"), some("4444")),
("refs/heads/new".to_string(), None, some("5555")),
]
);
}
#[test]
fn only_public_refs_make_events() {
let sha = "ab".repeat(20);
let moved: Vec<_> = [
"refs/heads/main",
"refs/cobs/org.tangled.track.ticket/1",
"refs/cob-checkpoints/x",
"refs/tngl/keep/abc",
"refs/hidden/x",
"refs/atproto/x",
"refs/tags/v1",
]
.map(|name| self::moved(name, None, Some(&sha)))
.into();
let sent: Vec<_> = events(
NULL,
&RepoDid::new("did:plc:squid").unwrap(),
&AccountDid::new("did:plc:pusher").unwrap(),
&moved,
)
.into_iter()
.map(|(name, _)| name)
.collect();
assert_eq!(sent, ["refs/heads/main", "refs/tags/v1"]);
}
#[test]
fn an_event_too_big_to_queue_is_dropped() {
let long = format!("refs/heads/{}", "x".repeat(MAX_EVENT_BYTES));
let sent = events(
NULL,
&RepoDid::new("did:plc:squid").unwrap(),
&AccountDid::new("did:plc:pusher").unwrap(),
&[moved(&long, None, Some(NULL))],
);
assert!(sent.is_empty());
}
#[test]
fn a_spindles_words_reach_the_terminal_as_plain_lines() {
let said = "\u{1b}[31mred\u{1b}[0m\u{7} line\n\u{1b}]0;title\u{7}two\r\n";
assert_eq!(terminal_lines(said), ["red line", "two"]);
let many = "line\n".repeat(MAX_LINES + 5);
assert_eq!(terminal_lines(&many).len(), MAX_LINES);
assert_eq!(
terminal_lines(&"x".repeat(MAX_CHARS + 50))[0].len(),
MAX_CHARS
);
}
}