//! the repo rows the relay changed that the indexer hasn't applied yet. in indexer builds the
//! relay doesn't write the row itself, it hands what it changed on with the message, and the
//! indexer puts it on the row in the txn that applies the message. a row the relay wrote first
//! would be ahead of the indexer after a crash between the two, and the replayed message would
//! look stale and be dropped. meanwhile the relay reads its own changes back from here
use std::sync::atomic::{AtomicU64, Ordering};
use jacquard_common::types::did::Did;
use miette::Result;
use crate::types::RepoState;
/// what one relay message changed on a repo row
#[derive(Debug)]
pub(crate) struct RowDelta {
/// the row the relay read, `None` when it made the row up for a repo it hadn't seen
pub(crate) before: Option<RepoState<'static>>,
pub(crate) after: RepoState<'static>,
/// which of the relay's changes this is, see [`Overlay::applied`]
pub(crate) version: u64,
}
/// where a [`RowDelta`] leaves the row
pub(crate) enum Landed {
Row(RepoState<'static>),
/// a new repo's row, which the repo count and a backfill still have to hear about
Made(RepoState<'static>),
/// the row was erased since the relay read it, so the change goes with it
Gone,
}
impl RowDelta {
pub(crate) fn land(&self, current: Option<RepoState<'static>>) -> Landed {
land(self.before.as_ref(), &self.after, current)
}
}
fn land(
before: Option<&RepoState<'static>>,
after: &RepoState<'static>,
current: Option<RepoState<'static>>,
) -> Landed {
match (before, current) {
(Some(before), Some(current)) => Landed::Row(after.clone().rebased(before, current)),
(Some(_), None) => Landed::Gone,
(None, None) => Landed::Made(after.clone()),
// a track, the crawler or a backfill made the row while the relay took the repo for
// new, and counted it and saw to its backfill. what the relay learned goes onto that
// row as if it had started from a blank one
(None, Some(current)) => {
Landed::Row(after.clone().rebased(&RepoState::backfilling(), current))
}
}
}
#[derive(Clone)]
struct Pending {
/// the row the first change not applied yet was made from
before: Option<RepoState<'static>>,
/// the row after the newest one
after: RepoState<'static>,
version: u64,
}
#[derive(Default)]
pub(crate) struct Overlay {
rows: scc::HashMap<Did, Pending>,
next_version: AtomicU64,
}
impl Overlay {
/// `did`'s row as the relay left it: what `read` finds durable, with the changes the
/// indexer hasn't applied yet on top
pub(crate) fn view(
&self,
did: &Did,
read: impl FnOnce() -> Result<Option<RepoState<'static>>>,
) -> Result<Option<RepoState<'static>>> {
// before the durable read: the indexer commits before it drops the entry, so a row
// read after finding none has everything in it already
let pending = self.rows.read_sync(did, |_, pending| pending.clone());
let durable = read()?;
let Some(pending) = pending else {
return Ok(durable);
};
Ok(
match land(pending.before.as_ref(), &pending.after, durable) {
Landed::Row(row) | Landed::Made(row) => Some(row),
Landed::Gone => None,
},
)
}
/// keeps a change the relay hands on, until [`Self::applied`]
pub(crate) fn stage(
&self,
did: &Did,
before: Option<RepoState<'static>>,
after: RepoState<'static>,
) -> RowDelta {
let version = self.next_version.fetch_add(1, Ordering::Relaxed);
let newest = after.clone();
self.rows
.entry_sync(did.clone())
.and_modify(|pending| {
pending.after = newest.clone();
pending.version = version;
})
.or_insert_with(|| Pending {
before: before.clone(),
after: newest,
version,
});
RowDelta {
before,
after,
version,
}
}
/// the indexer committed the change `version`, or dropped it. once that was the newest, the
/// durable row is all the relay has to read
pub(crate) fn applied(&self, did: &Did, version: u64) {
self.rows
.remove_if_sync(did, |pending| pending.version == version);
}
#[cfg(test)]
pub(crate) fn len(&self) -> usize {
self.rows.len()
}
}