//! the messages a relay shard holds for a did while that did's lookup runs. the shard keeps
//! working through every other did meanwhile, and replays these in order once the answer is in
use std::collections::{HashMap, VecDeque};
use std::time::{Duration, Instant};
use jacquard_common::types::did::Did;
use smol_str::SmolStr;
use super::WorkerMessage;
use super::lookup::{LookupKind, Lookups};
use crate::ingest::cursor::Ticket;
use crate::ingest::stream::SubscribeReposMessage;
const MESSAGES_PER_DID: usize = 64;
const BYTES_PER_DID: usize = 2 * 1024 * 1024;
const DIDS_PER_SHARD: usize = 1024;
/// a did's held messages are dropped this long after the first one, whatever its lookup is
/// doing, checked whenever the shard wakes. the lookup's own deadline normally ends it well
/// before
pub(crate) const HOLD_MAX: Duration = Duration::from_secs(30);
pub(crate) struct Held {
pub(crate) msg: WorkerMessage,
pub(crate) ticket: Ticket,
}
impl Held {
/// about what holding it costs. frames are a few KB, a commit's blocks are most of it
fn size(&self) -> usize {
const OVERHEAD: usize = 512;
OVERHEAD
+ match &self.msg.msg {
SubscribeReposMessage::Commit(commit) => commit.blocks.len(),
SubscribeReposMessage::Sync(sync) => sync.blocks.len(),
_ => 0,
}
}
}
struct Queue {
held: VecDeque<Held>,
/// the lookup the first message waits for
kind: LookupKind,
bytes: usize,
since: Instant,
/// the did:web host whose park slot this queue takes
host: Option<SmolStr>,
}
#[derive(Default)]
pub(crate) struct Parked {
dids: HashMap<Did, Queue>,
/// when each queue started, oldest first. a queue taken early leaves its entry here until
/// it reaches the front, which is at most [`HOLD_MAX`] later
started: VecDeque<(Instant, Did)>,
}
impl Parked {
pub(crate) fn kind(&self, did: &Did) -> Option<LookupKind> {
self.dids.get(did).map(|queue| queue.kind)
}
pub(crate) fn len(&self) -> usize {
self.dids.len()
}
/// starts holding messages for `did`, or hands `held` back when a cap says no
pub(crate) fn start(
&mut self,
lookups: &Lookups,
did: Did,
kind: LookupKind,
held: Held,
host: Option<SmolStr>,
) -> Result<(), Held> {
let bytes = held.size();
if self.dids.len() >= DIDS_PER_SHARD || !lookups.reserve_park(host.as_ref(), bytes) {
return Err(held);
}
let since = Instant::now();
self.started.push_back((since, did.clone()));
self.dids.insert(
did,
Queue {
held: VecDeque::from([held]),
kind,
bytes,
since,
host,
},
);
Ok(())
}
/// holds `held` behind what `did` already waits with, so its messages never reorder. past
/// a cap the whole queue comes back, keeping a tail would reorder too. `did` must be held
pub(crate) fn push(
&mut self,
lookups: &Lookups,
did: &Did,
held: Held,
) -> Result<(), VecDeque<Held>> {
let queue = self.dids.get_mut(did).expect("the did is held");
let bytes = held.size();
if queue.held.len() >= MESSAGES_PER_DID
|| queue.bytes + bytes > BYTES_PER_DID
|| !lookups.grow_park(bytes)
{
let mut dropped = self.take(lookups, did).expect("just found");
dropped.push_back(held);
return Err(dropped);
}
queue.bytes += bytes;
queue.held.push_back(held);
Ok(())
}
pub(crate) fn take(&mut self, lookups: &Lookups, did: &Did) -> Option<VecDeque<Held>> {
let queue = self.dids.remove(did)?;
lookups.release_park(queue.host.as_ref(), queue.bytes);
Some(queue.held)
}
/// the queues held past [`HOLD_MAX`], with the lookup each waited for
pub(crate) fn expired(&mut self, lookups: &Lookups) -> Vec<(Did, LookupKind, VecDeque<Held>)> {
let mut expired = Vec::new();
while let Some((since, _)) = self.started.front()
&& since.elapsed() >= HOLD_MAX
{
let (since, did) = self.started.pop_front().expect("just looked");
if let Some(kind) = self
.dids
.get(&did)
.filter(|queue| queue.since == since)
.map(|queue| queue.kind)
{
let held = self.take(lookups, &did).expect("just found");
expired.push((did, kind, held));
}
}
expired
}
}