use std::collections::BTreeMap;
use std::path::PathBuf;
use knot_types::{RepoDid, ServiceDid};
use crate::datastore::{Datastore, schema};
use serde::{Deserialize, Serialize};
const BACKOFF_BASE: i64 = 1;
const BACKOFF_CAP: i64 = 60;
const MAX_AGE: i64 = 3 * 24 * 60 * 60;
const MAX_ENTRIES: usize = 1_000;
// a dead spindle only evicts its own events, never another's.
const MAX_PER_SPINDLE: usize = 200;
#[derive(Serialize, Deserialize, Clone, Debug)]
pub(crate) struct Entry {
pub(crate) id: u64,
pub(crate) spindle: ServiceDid,
pub(crate) repo: RepoDid,
pub(crate) ref_name: String,
pub(crate) body: serde_json::Value,
queued_at: i64,
}
#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq)]
struct Backoff {
failures: u32,
retry_at: i64,
#[serde(default)]
claim: Option<u64>,
}
/// a dropped claim leaves its spindle waiting out the lease, and only logs.
#[must_use = "settle or release the claim, or its spindle waits out the lease"]
pub(crate) struct Claim {
pub(crate) entry: Entry,
token: u64,
resolved: bool,
}
impl Drop for Claim {
fn drop(&mut self) {
if !self.resolved {
tracing::warn!(
spindle = self.entry.spindle.as_str(),
ref_name = self.entry.ref_name.as_str(),
"a spindle claim went unresolved, so its spindle waits out the lease"
);
}
}
}
pub(crate) enum Outcome {
Delivered,
/// a 4xx except 408 and 429 is the spindle rejecting the event itself.
Rejected,
Retry,
}
#[derive(Serialize, Deserialize, Default, Debug)]
pub(crate) struct Queue {
next_id: u64,
#[serde(default)]
next_claim: u64,
entries: Vec<Entry>,
backoff: BTreeMap<String, Backoff>,
}
fn delay(failures: u32) -> i64 {
let doublings = failures.saturating_sub(1).min(31);
BACKOFF_BASE
.saturating_mul(1i64 << doublings)
.min(BACKOFF_CAP)
}
impl Queue {
pub(crate) fn enqueue(
&mut self,
spindle: &ServiceDid,
repo: &RepoDid,
events: Vec<(String, serde_json::Value)>,
now: i64,
) -> Vec<u64> {
let mut ids = Vec::new();
for (ref_name, body) in events {
self.make_room(spindle);
let id = self.next_id;
self.next_id += 1;
ids.push(id);
self.entries.push(Entry {
id,
spindle: spindle.clone(),
repo: repo.clone(),
ref_name,
body,
queued_at: now,
});
}
self.prune_backoff();
ids
}
fn make_room(&mut self, spindle: &ServiceDid) {
let held = self.entries.iter().filter(|e| &e.spindle == spindle);
if held.count() >= MAX_PER_SPINDLE {
if let Some(at) = self.entries.iter().position(|e| &e.spindle == spindle) {
self.evict(at, "its spindle has too many events queued");
}
} else if self.entries.len() >= MAX_ENTRIES {
self.evict(0, "the spindle outbox is full");
}
}
fn evict(&mut self, at: usize, why: &str) {
let dropped = self.entries.remove(at);
tracing::warn!(
repo = dropped.repo.as_str(),
ref_name = dropped.ref_name.as_str(),
spindle = dropped.spindle.as_str(),
"{why}, so its oldest push event is dropped"
);
}
fn prune_backoff(&mut self) {
let queued: Vec<&str> = self.entries.iter().map(|e| e.spindle.as_str()).collect();
self.backoff
.retain(|spindle, _| queued.contains(&spindle.as_str()));
}
fn expire(&mut self, now: i64) {
let oldest_kept = now - MAX_AGE;
self.entries.retain(|entry| {
let fresh = entry.queued_at >= oldest_kept;
if !fresh {
tracing::warn!(
repo = entry.repo.as_str(),
ref_name = entry.ref_name.as_str(),
spindle = entry.spindle.as_str(),
"giving up on a push event the spindle never took"
);
}
fresh
});
self.prune_backoff();
}
/// the lease has to outlast a delivery, or another claim re-sends the same event.
pub(crate) fn claim(&mut self, now: i64, lease: i64) -> Option<Claim> {
self.expire(now);
let entry = self
.entries
.iter()
.find(|entry| {
self.backoff
.get(entry.spindle.as_str())
.is_none_or(|backoff| backoff.retry_at <= now)
})?
.clone();
let token = self.next_claim;
self.next_claim += 1;
let held = self
.backoff
.entry(entry.spindle.to_string())
.or_insert(Backoff {
failures: 0,
retry_at: now,
claim: None,
});
held.retry_at = now + lease;
held.claim = Some(token);
Some(Claim {
entry,
token,
resolved: false,
})
}
/// after a takeover, a delivered event still goes and the backoff stays put.
pub(crate) fn settle(&mut self, mut claim: Claim, outcome: Outcome, now: i64) {
claim.resolved = true;
let spindle = claim.entry.spindle.to_string();
let holds = self
.backoff
.get(&spindle)
.is_some_and(|held| held.claim == Some(claim.token));
match outcome {
Outcome::Delivered | Outcome::Rejected => {
self.entries.retain(|queued| queued.id != claim.entry.id);
if holds {
self.backoff.remove(&spindle);
}
}
Outcome::Retry if holds => {
let failures = self.backoff[&spindle].failures.saturating_add(1);
let retry_at = now + delay(failures);
let backoff = Backoff {
failures,
retry_at,
claim: None,
};
self.backoff.insert(spindle, backoff);
}
Outcome::Retry => {}
}
self.prune_backoff();
}
pub(crate) fn release(&mut self, mut claim: Claim, now: i64) {
claim.resolved = true;
let spindle = claim.entry.spindle.to_string();
if let Some(held) = self
.backoff
.get_mut(&spindle)
.filter(|held| held.claim == Some(claim.token))
{
held.claim = None;
held.retry_at = now;
}
self.prune_backoff();
}
pub(crate) fn next_retry(&self) -> Option<i64> {
self.backoff
.values()
.filter(|backoff| backoff.claim.is_none())
.map(|backoff| backoff.retry_at)
.min()
}
pub(crate) fn is_empty(&self) -> bool {
self.entries.is_empty()
}
}
pub(crate) struct Outbox {
root: PathBuf,
}
impl Outbox {
pub(crate) fn at(root: &std::path::Path) -> Self {
Self {
root: root.to_owned(),
}
}
pub(crate) fn update<R>(&self, change: impl FnOnce(&mut Queue) -> R) -> anyhow::Result<R> {
Datastore::at(&self.root)
.record::<schema::SpindleOutbox>(&())
.update(change)
}
}
#[cfg(test)]
mod tests {
use super::*;
const LEASE: i64 = 90;
fn spindle(host: &str) -> ServiceDid {
ServiceDid::new(format!("did:web:{host}")).unwrap()
}
fn queue_of(spindle_host: &str, refs: &[&str], now: i64) -> Queue {
let mut queue = Queue::default();
let events = refs
.iter()
.map(|name| (name.to_string(), serde_json::json!({ "ref": name })))
.collect();
queue.enqueue(
&spindle(spindle_host),
&RepoDid::new("did:plc:squid").unwrap(),
events,
now,
);
queue
}
fn claimed_ref(queue: &mut Queue, now: i64) -> Option<String> {
queue
.claim(now, LEASE)
.map(|claim| claim.entry.ref_name.clone())
}
#[test]
fn a_spindles_events_go_out_one_at_a_time_in_order() {
let mut queue = queue_of("a.test", &["refs/heads/one", "refs/heads/two"], 100);
let first = queue.claim(100, LEASE).unwrap();
assert_eq!(first.entry.ref_name, "refs/heads/one");
assert_eq!(claimed_ref(&mut queue, 100), None, "held while one is out");
queue.settle(first, Outcome::Delivered, 100);
assert_eq!(
claimed_ref(&mut queue, 100).as_deref(),
Some("refs/heads/two")
);
}
#[test]
fn a_failed_attempt_backs_the_spindle_off_by_knot2s_doubling_up_to_its_cap() {
let mut queue = queue_of("a.test", &["refs/heads/one"], 100);
let mut now = 100;
let mut waits = Vec::new();
for _ in 0..8 {
let entry = queue.claim(now, LEASE).unwrap();
queue.settle(entry, Outcome::Retry, now);
let retry_at = queue.next_retry().unwrap();
assert!(
queue.claim(retry_at - 1, LEASE).is_none(),
"not before it is due"
);
waits.push(retry_at - now);
now = retry_at;
}
assert_eq!(waits, [1, 2, 4, 8, 16, 32, 60, 60]);
}
#[test]
fn one_spindle_backing_off_does_not_hold_another() {
let mut queue = queue_of("a.test", &["refs/heads/a"], 100);
queue.enqueue(
&spindle("b.test"),
&RepoDid::new("did:plc:squid").unwrap(),
vec![("refs/heads/b".into(), serde_json::json!({}))],
100,
);
let a = queue.claim(100, LEASE).unwrap();
queue.settle(a, Outcome::Retry, 100);
assert_eq!(
claimed_ref(&mut queue, 100).as_deref(),
Some("refs/heads/b")
);
}
#[test]
fn a_rejected_event_is_dropped_and_frees_the_spindle() {
let mut queue = queue_of("a.test", &["refs/heads/bad", "refs/heads/good"], 100);
let bad = queue.claim(100, LEASE).unwrap();
queue.settle(bad, Outcome::Rejected, 100);
assert_eq!(
claimed_ref(&mut queue, 100).as_deref(),
Some("refs/heads/good")
);
}
#[test]
fn a_claim_that_never_settles_frees_its_spindle_after_the_lease() {
let mut queue = queue_of("a.test", &["refs/heads/one"], 100);
drop(queue.claim(100, LEASE).unwrap());
assert_eq!(claimed_ref(&mut queue, 100 + LEASE - 1), None);
assert!(claimed_ref(&mut queue, 100 + LEASE).is_some());
}
#[test]
fn a_spindle_held_by_a_delivery_in_flight_sets_no_wakeup() {
let mut queue = queue_of("a.test", &["refs/heads/one"], 100);
let claim = queue.claim(100, LEASE).unwrap();
assert_eq!(queue.next_retry(), None);
queue.settle(claim, Outcome::Retry, 100);
assert_eq!(queue.next_retry(), Some(101));
}
#[test]
fn a_retry_after_the_lease_was_taken_over_leaves_the_new_holders_lease() {
let mut queue = queue_of("a.test", &["refs/heads/one", "refs/heads/two"], 100);
let slow = queue.claim(100, LEASE).unwrap();
let taken_over = queue.claim(100 + LEASE, LEASE).unwrap();
assert_eq!(taken_over.entry.id, slow.entry.id);
queue.settle(slow, Outcome::Retry, 100 + LEASE + 1);
assert_eq!(
queue.next_retry(),
None,
"its lease is still the new holder's"
);
assert!(queue.claim(100 + LEASE + 2, LEASE).is_none());
queue.settle(taken_over, Outcome::Delivered, 100 + LEASE + 3);
assert_eq!(
claimed_ref(&mut queue, 100 + LEASE + 3).as_deref(),
Some("refs/heads/two")
);
}
#[test]
fn a_delivery_by_a_claim_that_lost_its_lease_still_removes_its_event() {
let mut queue = queue_of("a.test", &["refs/heads/one", "refs/heads/two"], 100);
let slow = queue.claim(100, LEASE).unwrap();
let taken_over = queue.claim(100 + LEASE, LEASE).unwrap();
queue.settle(slow, Outcome::Delivered, 100 + LEASE + 1);
assert!(
queue.claim(100 + LEASE + 2, LEASE).is_none(),
"what the slow claim delivered is gone, the spindle is still held"
);
queue.settle(taken_over, Outcome::Delivered, 100 + LEASE + 3);
assert_eq!(
claimed_ref(&mut queue, 100 + LEASE + 3).as_deref(),
Some("refs/heads/two")
);
}
#[test]
fn a_released_claim_frees_its_spindle_without_counting_a_failure() {
let mut queue = queue_of("a.test", &["refs/heads/one"], 100);
let claim = queue.claim(100, LEASE).unwrap();
queue.release(claim, 100);
assert_eq!(
claimed_ref(&mut queue, 100).as_deref(),
Some("refs/heads/one"),
"the spindle is free at once"
);
}
#[test]
fn events_older_than_the_max_age_are_given_up_on() {
let mut queue = queue_of("a.test", &["refs/heads/old"], 100);
queue.enqueue(
&spindle("a.test"),
&RepoDid::new("did:plc:squid").unwrap(),
vec![("refs/heads/new".into(), serde_json::json!({}))],
100 + MAX_AGE,
);
assert_eq!(
claimed_ref(&mut queue, 100 + MAX_AGE + 1).as_deref(),
Some("refs/heads/new")
);
}
#[test]
fn a_dead_spindle_can_only_crowd_out_its_own_events() {
let mut queue = queue_of("dead.test", &[], 100);
queue.enqueue(
&spindle("live.test"),
&RepoDid::new("did:plc:squid").unwrap(),
vec![("refs/heads/live".into(), serde_json::json!({}))],
100,
);
for n in 0..MAX_PER_SPINDLE + 5 {
queue.enqueue(
&spindle("dead.test"),
&RepoDid::new("did:plc:squid").unwrap(),
vec![(format!("refs/heads/{n}"), serde_json::json!({}))],
100,
);
}
let held = |host: &str| {
queue
.entries
.iter()
.filter(|entry| entry.spindle == spindle(host))
.count()
};
assert_eq!(held("dead.test"), MAX_PER_SPINDLE);
assert_eq!(held("live.test"), 1);
assert_eq!(
queue
.entries
.iter()
.find(|e| e.spindle == spindle("dead.test"))
.unwrap()
.ref_name,
"refs/heads/5",
"the oldest went"
);
}
#[test]
fn the_whole_outbox_is_bounded() {
let mut queue = Queue::default();
for n in 0..MAX_ENTRIES + 3 {
queue.enqueue(
&spindle(&format!("s{n}.test")),
&RepoDid::new("did:plc:squid").unwrap(),
vec![("refs/heads/x".into(), serde_json::json!({}))],
100,
);
}
assert_eq!(queue.entries.len(), MAX_ENTRIES);
}
#[test]
fn the_queue_survives_the_process_that_wrote_it() {
let root = tempfile::tempdir().unwrap();
let ids = Outbox::at(root.path())
.update(|queue| {
queue.enqueue(
&spindle("a.test"),
&RepoDid::new("did:plc:squid").unwrap(),
vec![("refs/heads/one".into(), serde_json::json!({ "n": 1 }))],
100,
)
})
.unwrap();
let claim = Outbox::at(root.path())
.update(|queue| queue.claim(100, LEASE))
.unwrap()
.unwrap();
assert_eq!(
(claim.entry.id, claim.entry.body.clone()),
(ids[0], serde_json::json!({ "n": 1 }))
);
let held = Outbox::at(root.path())
.update(|queue| queue.claim(100, LEASE))
.unwrap();
assert!(held.is_none(), "the lease is stored too");
}
#[test]
fn an_outbox_that_does_not_decode_is_an_error_and_keeps_its_bytes() {
let root = tempfile::tempdir().unwrap();
let store = Datastore::at(root.path());
let path = store
.record::<schema::SpindleOutbox>(&())
.path()
.to_path_buf();
std::fs::write(&path, b"{\"entries\": [ truncated").unwrap();
let error = Outbox::at(root.path())
.update(|queue| {
queue.enqueue(
&spindle("a.test"),
&RepoDid::new("did:plc:squid").unwrap(),
vec![("refs/heads/one".into(), serde_json::json!({}))],
100,
)
})
.expect_err("junk is not an empty queue");
assert!(format!("{error:#}").contains("decode the spindle outbox"));
assert_eq!(
std::fs::read(&path).unwrap(),
b"{\"entries\": [ truncated",
"the queued events stay for a build that can read them"
);
}
#[test]
fn an_outbox_nobody_wrote_to_leaves_no_file() {
let root = tempfile::tempdir().unwrap();
assert!(
Outbox::at(root.path())
.update(|queue| queue.is_empty())
.unwrap()
);
let store = Datastore::at(root.path());
assert!(!store.record::<schema::SpindleOutbox>(&()).path().exists());
}
}