sources: refresh_chat, retire_chat, RoomHistory port, steward purge windows, and the tests that falsify them
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
ae4871b576
commit
e01cdfc4d9
6 changed files with 1435 additions and 14 deletions
|
|
@ -1303,6 +1303,22 @@ impl huginn_daemon::sources::SourceIndex for Desk {
|
|||
anyhow::bail!("a daemon never writes the source index")
|
||||
}
|
||||
|
||||
fn chat_tail(&mut self, _room_id: &str) -> anyhow::Result<Option<huginn_daemon::sources::StoredChat>> {
|
||||
anyhow::bail!("a daemon never reads the ingest job's side of the source index")
|
||||
}
|
||||
|
||||
fn chat_chunks_holding(&mut self, _room_id: &str, _event_ids: &[String]) -> anyhow::Result<Vec<huginn_daemon::sources::StoredChat>> {
|
||||
anyhow::bail!("a daemon never reads the ingest job's side of the source index")
|
||||
}
|
||||
|
||||
fn chat_chunks_overlapping(&mut self, _room_id: &str, _from_ms: u64, _to_ms: u64) -> anyhow::Result<Vec<huginn_daemon::sources::StoredChat>> {
|
||||
anyhow::bail!("a daemon never reads the ingest job's side of the source index")
|
||||
}
|
||||
|
||||
fn delete_chat(&mut self, _room_id: &str, _ids: Option<&[String]>) -> anyhow::Result<()> {
|
||||
anyhow::bail!("a daemon never writes the source index")
|
||||
}
|
||||
|
||||
fn chat_nearest(
|
||||
&mut self,
|
||||
_vector: &[f32],
|
||||
|
|
|
|||
|
|
@ -264,6 +264,8 @@ pub(crate) struct SourcesState {
|
|||
pub(crate) deletes: Vec<(String, Option<Vec<String>>)>,
|
||||
/// A delete naming this repo refuses.
|
||||
pub(crate) fail_deletes_of: Option<String>,
|
||||
/// Every cursor write refuses once this many have been written.
|
||||
pub(crate) fail_cursor_writes_after: Option<usize>,
|
||||
/// Every chat upsert refuses while this is set.
|
||||
pub(crate) fail_chat_upserts: bool,
|
||||
/// Every chat delete refuses while this is set.
|
||||
|
|
@ -332,6 +334,9 @@ impl SourceIndex for FakeSources {
|
|||
|
||||
fn set_cursors(&mut self, cursors: &BTreeMap<String, String>) -> Result<()> {
|
||||
let mut state = self.state.lock().unwrap();
|
||||
if state.fail_cursor_writes_after.is_some_and(|after| state.cursor_writes.len() >= after) {
|
||||
bail!("the vector store refused the cursor write");
|
||||
}
|
||||
state.cursor_writes.push(cursors.clone());
|
||||
let Some(DescribedSources::Labelled { cursors: held, .. }) = &mut state.described else { bail!("no source collection") };
|
||||
*held = cursors.clone();
|
||||
|
|
|
|||
|
|
@ -1,6 +1,28 @@
|
|||
//! Chat ingest: joined-room history into the shared source collection, beside
|
||||
//! the forge's repos. `refresh_chat` is the only writer of `corpus=chat` points
|
||||
//! and of the cursors object's chat keys.
|
||||
//! and of the cursors object's `chat:` keys; `chat_purge::request_purge` is the
|
||||
//! only writer of `chat-purge:` keys, and `refresh_chat` the only applier.
|
||||
//!
|
||||
//! The Matrix client that implements `RoomHistory` is the next cut's and holds no
|
||||
//! policy. What is in scope is `room_scope`'s; what may be embedded is
|
||||
//! `huginn_mind::redact::scrub`'s; this file folds events into chunks and writes
|
||||
//! them, write-then-delete and cursor-last, so any crash replays to the same state.
|
||||
|
||||
use std::collections::{BTreeMap, BTreeSet, HashSet};
|
||||
use std::time::Instant;
|
||||
|
||||
use anyhow::{Context, Result, bail, ensure};
|
||||
use huginn_mind::ChatMessage;
|
||||
use huginn_mind::redact::{ScrubKind, scrub};
|
||||
|
||||
use super::chat_chunk::{chat_version, chunk_id, chunk_messages, chunk_payload, cut_body};
|
||||
use super::chat_purge::{PurgeWindow, covers, parse_windows, purge_key};
|
||||
use super::room_scope::{Estate, Refusal, RoomFacts, admit_room, refuse_event};
|
||||
use super::{CORPUS, ChatPoint, DescribedSources, SourceIndex, StoredChat};
|
||||
use crate::index::{BATCH, BULK_EMBED_TIMEOUT, Embedder, INDEX_CALL_TIMEOUT};
|
||||
|
||||
/// Events asked of the port per page.
|
||||
pub const CHAT_PAGE: u32 = 200;
|
||||
|
||||
/// A chat cursor is the key `chat:<room_id>`; a forge repo name cannot contain
|
||||
/// a colon. The only definition of the prefix.
|
||||
|
|
@ -39,3 +61,488 @@ pub struct RoomEvent {
|
|||
pub content_keys: Vec<String>,
|
||||
pub kind: EventKind,
|
||||
}
|
||||
|
||||
/// One page of a room's events after a cursor event, oldest first.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct EventPage {
|
||||
pub events: Vec<RoomEvent>,
|
||||
/// More events follow this page.
|
||||
pub more: bool,
|
||||
}
|
||||
|
||||
/// The reader user's view of Synapse as ingest uses it. Holds no policy: it maps
|
||||
/// the server's answers to these types and filters nothing.
|
||||
pub trait RoomHistory {
|
||||
/// The rooms the reader has joined. The only source of rooms; there is no allow list.
|
||||
fn joined_rooms(&mut self) -> Result<Vec<String>>;
|
||||
fn facts(&mut self, room_id: &str) -> Result<RoomFacts>;
|
||||
/// Up to `limit` events after the event `after` (from the room's start when `None`).
|
||||
/// `Ok(None)` means `after` is unknown to the server.
|
||||
fn events_after(&mut self, room_id: &str, after: Option<&str>, limit: u32) -> Result<Option<EventPage>>;
|
||||
}
|
||||
|
||||
/// The cursors-object key of a room's chat cursor.
|
||||
pub fn cursor_key(room_id: &str) -> String {
|
||||
format!("{CHAT_CURSOR_PREFIX}{room_id}")
|
||||
}
|
||||
|
||||
/// The room a chat cursor key belongs to; `None` for any other key (a purge tombstone included).
|
||||
pub fn cursor_room(key: &str) -> Option<&str> {
|
||||
key.strip_prefix(CHAT_CURSOR_PREFIX)
|
||||
}
|
||||
|
||||
/// A cursor value, `<version>:<last event id>`; the event is empty before a room's first upsert.
|
||||
pub fn parse_cursor(value: &str) -> Option<(&str, &str)> {
|
||||
value.split_once(':')
|
||||
}
|
||||
|
||||
/// What one admitted room's run did, for the journal. Never carries a body or a sender.
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub struct ChatReport {
|
||||
pub room_id: String,
|
||||
pub alias: Option<String>,
|
||||
pub first: Option<String>,
|
||||
pub last: Option<String>,
|
||||
pub events: u32,
|
||||
pub chunks: u32,
|
||||
pub scrubbed: u32,
|
||||
pub purged: u32,
|
||||
pub seconds: f64,
|
||||
}
|
||||
|
||||
impl std::fmt::Display for ChatReport {
|
||||
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
write!(
|
||||
formatter,
|
||||
"{} ({}) {}..{} events={} chunks={} scrubbed={} purged={} seconds={:.1}",
|
||||
self.alias.as_deref().unwrap_or("-"),
|
||||
self.room_id,
|
||||
self.first.as_deref().unwrap_or("-"),
|
||||
self.last.as_deref().unwrap_or("-"),
|
||||
self.events,
|
||||
self.chunks,
|
||||
self.scrubbed,
|
||||
self.purged,
|
||||
self.seconds
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/// A joined room the scope refused, whole, and why; `event_id` when an event did it.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct RefusedRoom {
|
||||
pub room_id: String,
|
||||
pub reason: Refusal,
|
||||
pub event_id: Option<String>,
|
||||
}
|
||||
|
||||
/// A credential the scrub replaced: where, never what.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct ScrubHit {
|
||||
pub room_id: String,
|
||||
pub event_id: String,
|
||||
pub kind: ScrubKind,
|
||||
}
|
||||
|
||||
/// What a chat refresh did.
|
||||
#[derive(Clone, Debug, Default, PartialEq)]
|
||||
pub struct ChatRefreshed {
|
||||
pub rooms: Vec<ChatReport>,
|
||||
/// Rooms (or steps) that failed; the others went on.
|
||||
pub failed: Vec<String>,
|
||||
pub refused: Vec<RefusedRoom>,
|
||||
pub scrubbed: Vec<ScrubHit>,
|
||||
}
|
||||
|
||||
impl ChatRefreshed {
|
||||
/// Nothing failed and no joined room had to be refused: a joined estate, direct,
|
||||
/// encrypted or bridged room means the reader's roster is wrong. A scrub hit does not make a run unclean.
|
||||
pub fn is_clean(&self) -> bool {
|
||||
self.failed.is_empty() && self.refused.is_empty()
|
||||
}
|
||||
}
|
||||
|
||||
fn read_cursors<S: SourceIndex>(index: &mut S) -> Result<BTreeMap<String, String>> {
|
||||
match index.describe_sources(INDEX_CALL_TIMEOUT).context("describing the source collection")? {
|
||||
DescribedSources::Labelled { cursors, .. } => Ok(cursors),
|
||||
_ => bail!("the source collection is gone; the forge refresh creates it"),
|
||||
}
|
||||
}
|
||||
|
||||
fn put_cursor<S: SourceIndex>(index: &mut S, key: &str, value: &str) -> Result<()> {
|
||||
let mut cursors = read_cursors(index)?;
|
||||
if cursors.insert(key.to_owned(), value.to_owned()).as_deref() != Some(value) {
|
||||
index.set_cursors(&cursors).context("writing a chat cursor")?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn drop_cursor<S: SourceIndex>(index: &mut S, key: &str) -> Result<()> {
|
||||
let mut cursors = read_cursors(index)?;
|
||||
if cursors.remove(key).is_some() {
|
||||
index.set_cursors(&cursors).context("dropping a chat cursor")?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// A room's points, then its `chat:` cursor. Its purge tombstone stays.
|
||||
fn purge_room<S: SourceIndex>(index: &mut S, room_id: &str) -> Result<()> {
|
||||
index.delete_chat(room_id, None).with_context(|| format!("deleting the chunks of {room_id}"))?;
|
||||
drop_cursor(index, &cursor_key(room_id))
|
||||
}
|
||||
|
||||
/// Deletes the points and `chat:` cursor of every room not in `keep` (the reader left it,
|
||||
/// or it is refused or excluded), and returns those rooms. Purge tombstones and forge
|
||||
/// repos are never touched. With the empty set it removes the whole chat corpus.
|
||||
pub fn retire_chat<S: SourceIndex>(index: &mut S, keep: &BTreeSet<String>) -> Result<Vec<String>> {
|
||||
let rooms: Vec<String> = read_cursors(index)?.keys().filter_map(|key| cursor_room(key)).filter(|room| !keep.contains(*room)).map(str::to_owned).collect();
|
||||
for room in &rooms {
|
||||
purge_room(index, room)?;
|
||||
}
|
||||
Ok(rooms)
|
||||
}
|
||||
|
||||
/// A change to chunks already stored, in event order.
|
||||
enum Op {
|
||||
/// An edit by `sender` of `target`'s body; applies only to the original's sender.
|
||||
Edit { target: String, sender: String, body: String },
|
||||
/// A held message's body is what the page says it is (a replay).
|
||||
Set { target: String, body: String },
|
||||
Redact { target: String },
|
||||
}
|
||||
|
||||
impl Op {
|
||||
fn target(&self) -> &str {
|
||||
match self {
|
||||
Self::Edit { target, .. } | Self::Set { target, .. } | Self::Redact { target } => target,
|
||||
}
|
||||
}
|
||||
|
||||
/// Applies to `messages`; whether they changed.
|
||||
fn apply(&self, messages: &mut Vec<ChatMessage>) -> bool {
|
||||
let Some(at) = messages.iter().position(|message| message.event_id == self.target()) else { return false };
|
||||
match self {
|
||||
Self::Edit { sender, body, .. } if messages[at].sender == *sender && messages[at].body != *body => messages[at].body.clone_from(body),
|
||||
Self::Set { body, .. } if messages[at].body != *body => messages[at].body.clone_from(body),
|
||||
Self::Redact { .. } => {
|
||||
messages.remove(at);
|
||||
}
|
||||
_ => return false,
|
||||
}
|
||||
true
|
||||
}
|
||||
}
|
||||
|
||||
struct Folded {
|
||||
pending: Vec<ChatMessage>,
|
||||
ops: Vec<Op>,
|
||||
}
|
||||
|
||||
struct Room<'a> {
|
||||
facts: &'a RoomFacts,
|
||||
label: String,
|
||||
windows: Vec<PurgeWindow>,
|
||||
}
|
||||
|
||||
/// The page folded in order: the purge filter, then the credential scrub, then the
|
||||
/// event's effect. After this no raw body exists.
|
||||
fn fold(room: &Room, events: &[RoomEvent], report: &mut ChatReport, out: &mut ChatRefreshed) -> Folded {
|
||||
let mut folded = Folded { pending: Vec::new(), ops: Vec::new() };
|
||||
let mut redacted: HashSet<&str> = HashSet::new();
|
||||
let mut scrubbed = |event: &RoomEvent, body: &str| {
|
||||
let clean = scrub(body);
|
||||
for kind in clean.hits {
|
||||
report.scrubbed += 1;
|
||||
out.scrubbed.push(ScrubHit { room_id: room.facts.room_id.clone(), event_id: event.event_id.clone(), kind });
|
||||
}
|
||||
cut_body(&clean.text)
|
||||
};
|
||||
for event in events {
|
||||
match &event.kind {
|
||||
EventKind::Message { body } => {
|
||||
if covers(&room.windows, event.ts_ms) {
|
||||
report.purged += 1;
|
||||
} else if !redacted.contains(event.event_id.as_str()) {
|
||||
let body = scrubbed(event, body);
|
||||
if !body.trim().is_empty() {
|
||||
folded.pending.push(ChatMessage { event_id: event.event_id.clone(), sender: event.sender.clone(), ts_ms: event.ts_ms, body });
|
||||
}
|
||||
}
|
||||
}
|
||||
EventKind::Edit { target, body } => {
|
||||
if covers(&room.windows, event.ts_ms) {
|
||||
report.purged += 1;
|
||||
continue;
|
||||
}
|
||||
let body = scrubbed(event, body);
|
||||
match folded.pending.iter_mut().find(|message| message.event_id == *target) {
|
||||
Some(message) if message.sender == event.sender && !body.trim().is_empty() => message.body = body,
|
||||
Some(_) => {}
|
||||
None => folded.ops.push(Op::Edit { target: target.clone(), sender: event.sender.clone(), body }),
|
||||
}
|
||||
}
|
||||
EventKind::Redaction { target } => {
|
||||
folded.pending.retain(|message| message.event_id != *target);
|
||||
redacted.insert(target);
|
||||
folded.ops.push(Op::Redact { target: target.clone() });
|
||||
}
|
||||
EventKind::Other => {}
|
||||
}
|
||||
}
|
||||
folded
|
||||
}
|
||||
|
||||
/// Chunks to rewrite: the stored chunk ids they replace and the messages that now make them.
|
||||
type Rewrite = (Vec<String>, Vec<ChatMessage>);
|
||||
|
||||
/// Chunks `rewrites`, embeds and writes the new chunks (acknowledged), and only then deletes the old
|
||||
/// ids that are not among the new. Returns how many chunks were written.
|
||||
///
|
||||
/// A chunk written by an earlier run whose delete failed can still sit beside the chunk that replaced
|
||||
/// it, and a rewrite of the stale one then names the same first event, hence the same id, as the
|
||||
/// newer chunk. The newer chunk (`standing`, the other chunks the page found) stays; the stale one is deleted.
|
||||
fn write_rewrites<E: Embedder, S: SourceIndex>(
|
||||
embedder: &mut E,
|
||||
index: &mut S,
|
||||
dimensions: u32,
|
||||
room: &Room,
|
||||
rewrites: Vec<Rewrite>,
|
||||
standing: &BTreeSet<String>,
|
||||
) -> Result<u32> {
|
||||
let mut old: BTreeSet<String> = BTreeSet::new();
|
||||
let mut points: Vec<(String, super::ChatPayload)> = Vec::new();
|
||||
for (ids, messages) in rewrites {
|
||||
for group in chunk_messages(&room.label, &messages) {
|
||||
let id = chunk_id(&room.facts.room_id, &group[0].event_id);
|
||||
if standing.contains(&id) && !ids.contains(&id) {
|
||||
continue;
|
||||
}
|
||||
points.push((id, chunk_payload(&room.facts.room_id, room.facts.alias.as_deref(), &room.label, group)));
|
||||
}
|
||||
old.extend(ids);
|
||||
}
|
||||
let new: BTreeSet<&String> = points.iter().map(|(id, _)| id).collect();
|
||||
let stale: Vec<String> = old.iter().filter(|id| !new.contains(id)).cloned().collect();
|
||||
let written = u32::try_from(points.len()).unwrap_or(u32::MAX);
|
||||
for slice in points.chunks(BATCH) {
|
||||
let texts: Vec<String> = slice.iter().map(|(_, payload)| payload.text.clone()).collect();
|
||||
let vectors = embedder.embed(&texts, BULK_EMBED_TIMEOUT).context("embedding chat chunks")?;
|
||||
ensure!(vectors.len() == texts.len(), "the embedder returned {} vectors for {} chunks", vectors.len(), texts.len());
|
||||
let mut batch = Vec::with_capacity(slice.len());
|
||||
for ((id, payload), vector) in slice.iter().zip(vectors) {
|
||||
ensure!(vector.len() == dimensions as usize, "the embedder returned a vector of {} where the collection holds {dimensions}", vector.len());
|
||||
batch.push(ChatPoint { id: id.clone(), vector, payload: payload.clone() });
|
||||
}
|
||||
index.upsert_chat(&batch).context("writing chat chunks")?;
|
||||
}
|
||||
if !stale.is_empty() {
|
||||
index.delete_chat(&room.facts.room_id, Some(&stale)).context("deleting superseded chat chunks")?;
|
||||
}
|
||||
Ok(written)
|
||||
}
|
||||
|
||||
/// Applies the room's purge windows to the chunks it already holds.
|
||||
fn apply_purge<E: Embedder, S: SourceIndex>(embedder: &mut E, index: &mut S, dimensions: u32, room: &Room, report: &mut ChatReport) -> Result<()> {
|
||||
let mut hit: BTreeMap<String, StoredChat> = BTreeMap::new();
|
||||
for window in &room.windows {
|
||||
for chunk in index.chat_chunks_overlapping(&room.facts.room_id, window.from_ms, window.to_ms)? {
|
||||
hit.insert(chunk.id.clone(), chunk);
|
||||
}
|
||||
}
|
||||
let mut rewrites = Vec::new();
|
||||
for chunk in hit.into_values() {
|
||||
let kept: Vec<ChatMessage> = chunk.payload.messages.iter().filter(|message| !covers(&room.windows, message.ts_ms)).cloned().collect();
|
||||
let removed = chunk.payload.messages.len() - kept.len();
|
||||
if removed > 0 {
|
||||
report.purged += u32::try_from(removed).unwrap_or(u32::MAX);
|
||||
rewrites.push((vec![chunk.id], kept));
|
||||
}
|
||||
}
|
||||
report.chunks += write_rewrites(embedder, index, dimensions, room, rewrites, &BTreeSet::new())?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Folds one page into the stored chunks: edits and redactions rewrite the chunk that held
|
||||
/// the target alone, new messages extend the open tail.
|
||||
fn write_page<E: Embedder, S: SourceIndex>(embedder: &mut E, index: &mut S, dimensions: u32, room: &Room, mut folded: Folded) -> Result<u32> {
|
||||
let room_id = &room.facts.room_id;
|
||||
let mut asked: Vec<String> = folded.pending.iter().map(|message| message.event_id.clone()).collect();
|
||||
asked.extend(folded.ops.iter().map(|op| op.target().to_owned()));
|
||||
if asked.is_empty() {
|
||||
return Ok(0);
|
||||
}
|
||||
let holding = index.chat_chunks_holding(room_id, &asked)?;
|
||||
// A message the index already holds is a replay: its page body is asserted, never appended again.
|
||||
let held: HashSet<&str> = holding.iter().flat_map(|chunk| chunk.payload.messages.iter().map(|message| message.event_id.as_str())).collect();
|
||||
let (replayed, fresh): (Vec<ChatMessage>, Vec<ChatMessage>) = folded.pending.drain(..).partition(|message| held.contains(message.event_id.as_str()));
|
||||
let mut ops: Vec<Op> = replayed.into_iter().map(|message| Op::Set { target: message.event_id, body: message.body }).collect();
|
||||
ops.append(&mut folded.ops);
|
||||
|
||||
let tail = index.chat_tail(room_id)?;
|
||||
let tail_id = tail.as_ref().map(|tail| tail.id.clone());
|
||||
let standing: BTreeSet<String> = holding.iter().map(|chunk| chunk.id.clone()).chain(tail_id.clone()).collect();
|
||||
let mut tail_messages = tail.map(|tail| tail.payload.messages).unwrap_or_default();
|
||||
let mut tail_changed = false;
|
||||
let mut rewrites: Vec<Rewrite> = Vec::new();
|
||||
for chunk in holding {
|
||||
let mut messages = chunk.payload.messages.clone();
|
||||
let mut changed = false;
|
||||
for op in &ops {
|
||||
changed |= op.apply(&mut messages);
|
||||
}
|
||||
if Some(&chunk.id) == tail_id.as_ref() {
|
||||
(tail_messages, tail_changed) = (messages, changed);
|
||||
} else if changed {
|
||||
rewrites.push((vec![chunk.id], messages));
|
||||
}
|
||||
}
|
||||
if !fresh.is_empty() || tail_changed {
|
||||
tail_messages.extend(fresh);
|
||||
rewrites.push((tail_id.into_iter().collect(), tail_messages));
|
||||
}
|
||||
write_rewrites(embedder, index, dimensions, room, rewrites, &standing)
|
||||
}
|
||||
|
||||
enum Outcome {
|
||||
Done(ChatReport),
|
||||
Refused(RefusedRoom),
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn ingest_room<E: Embedder, S: SourceIndex, H: RoomHistory>(
|
||||
embedder: &mut E,
|
||||
index: &mut S,
|
||||
history: &mut H,
|
||||
dimensions: u32,
|
||||
facts: &RoomFacts,
|
||||
out: &mut ChatRefreshed,
|
||||
) -> Result<Outcome> {
|
||||
let started = Instant::now();
|
||||
let room_id = facts.room_id.as_str();
|
||||
let (key, version) = (cursor_key(room_id), chat_version());
|
||||
let cursors = read_cursors(index)?;
|
||||
let mut after: Option<String> = match cursors.get(&key).map(|value| parse_cursor(value)) {
|
||||
None => None,
|
||||
Some(Some((held, event))) if held == version => Some(event.to_owned()).filter(|event| !event.is_empty()),
|
||||
// Another chunker or scrub version, or a value nobody here wrote: this room alone starts over.
|
||||
Some(_) => {
|
||||
index.delete_chat(room_id, None).with_context(|| format!("deleting the chunks of {room_id} for a re-ingest"))?;
|
||||
None
|
||||
}
|
||||
};
|
||||
if after.is_none() {
|
||||
put_cursor(index, &key, &format!("{version}:"))?;
|
||||
}
|
||||
let windows = cursors.get(&purge_key(room_id)).map(|value| parse_windows(value)).transpose()?.unwrap_or_default();
|
||||
let label = facts.name.clone().or_else(|| facts.alias.clone()).unwrap_or_else(|| room_id.to_owned());
|
||||
let room = Room { facts, label, windows };
|
||||
let mut report = ChatReport { room_id: room_id.to_owned(), alias: facts.alias.clone(), first: None, last: None, events: 0, chunks: 0, scrubbed: 0, purged: 0, seconds: 0.0 };
|
||||
apply_purge(embedder, index, dimensions, &room, &mut report)?;
|
||||
let mut restarted = false;
|
||||
loop {
|
||||
let Some(page) = history.events_after(room_id, after.as_deref(), CHAT_PAGE)? else {
|
||||
ensure!(after.is_some() && !restarted, "the server does not know the start of {room_id}'s history");
|
||||
restarted = true;
|
||||
purge_room(index, room_id)?;
|
||||
put_cursor(index, &key, &format!("{version}:"))?;
|
||||
after = None;
|
||||
continue;
|
||||
};
|
||||
let Some(last) = page.events.last().map(|event| event.event_id.clone()) else { break };
|
||||
// A refusing event refuses the room before anything of the page is chunked or embedded.
|
||||
if let Some((event, reason)) = page.events.iter().find_map(|event| refuse_event(event).map(|reason| (event, reason))) {
|
||||
purge_room(index, room_id)?;
|
||||
return Ok(Outcome::Refused(RefusedRoom { room_id: room_id.to_owned(), reason, event_id: Some(event.event_id.clone()) }));
|
||||
}
|
||||
report.first.get_or_insert_with(|| page.events[0].event_id.clone());
|
||||
report.events += u32::try_from(page.events.len()).unwrap_or(u32::MAX);
|
||||
let folded = fold(&room, &page.events, &mut report, out);
|
||||
report.chunks += write_page(embedder, index, dimensions, &room, folded)?;
|
||||
put_cursor(index, &key, &format!("{version}:{last}"))?;
|
||||
report.last = Some(last.clone());
|
||||
after = Some(last);
|
||||
if !page.more {
|
||||
break;
|
||||
}
|
||||
}
|
||||
report.seconds = started.elapsed().as_secs_f64();
|
||||
Ok(Outcome::Done(report))
|
||||
}
|
||||
|
||||
/// Brings the chat corpus to the reader's joined rooms. The collection must already exist and be
|
||||
/// Huginn's, built under this embedder's model: the forge refresh creates and rebuilds it, and this never does.
|
||||
/// A joined room the scope refuses is purged and named; a room the reader left is retired; the rest
|
||||
/// are ingested from their cursors. One room's failure is journaled and the others go on.
|
||||
/// `journal` receives every line the run prints: reports, scrub hits, refusals and failures.
|
||||
pub fn refresh_chat<E: Embedder, S: SourceIndex, H: RoomHistory>(
|
||||
embedder: &mut E,
|
||||
index: &mut S,
|
||||
history: &mut H,
|
||||
estate: &Estate,
|
||||
journal: &mut dyn FnMut(String),
|
||||
) -> Result<ChatRefreshed> {
|
||||
let DescribedSources::Labelled { meta, .. } = index.describe_sources(INDEX_CALL_TIMEOUT).context("describing the source collection")? else {
|
||||
bail!("the source collection is absent or not Huginn's; the forge refresh creates it and chat never does");
|
||||
};
|
||||
ensure!(meta.corpus == CORPUS, "the source collection was built for corpus {} and chat ingest writes {CORPUS}; the forge refresh rebuilds it", meta.corpus);
|
||||
embedder.built_under(Some(&meta.model_identity()));
|
||||
let identity = embedder.model_identity(BULK_EMBED_TIMEOUT).context("reading the embedding model's identity")?;
|
||||
ensure!(meta.model_identity() == identity, "the source collection was built under {} {} and chat embeds under {} {}", meta.model, meta.model_digest, identity.name, identity.digest);
|
||||
|
||||
let mut out = ChatRefreshed::default();
|
||||
let mut joined = history.joined_rooms().context("listing the reader's joined rooms")?;
|
||||
let mut seen = HashSet::new();
|
||||
joined.retain(|room| seen.insert(room.clone()));
|
||||
let mut keep: BTreeSet<String> = joined.iter().cloned().collect();
|
||||
let mut admitted: Vec<RoomFacts> = Vec::new();
|
||||
for room_id in &joined {
|
||||
let verdict = if estate.holds_id(room_id) {
|
||||
Ok(Err(Refusal::Estate))
|
||||
} else {
|
||||
history.facts(room_id).map(|facts| admit_room(&facts, estate).map(|()| facts))
|
||||
};
|
||||
match verdict {
|
||||
Ok(Ok(facts)) => admitted.push(facts),
|
||||
Ok(Err(reason)) => {
|
||||
keep.remove(room_id);
|
||||
match purge_room(index, room_id) {
|
||||
Ok(()) => {}
|
||||
Err(error) => {
|
||||
journal(format!("huginn-sources: chat {room_id} could not be purged, the others go on: {error:#}"));
|
||||
out.failed.push(room_id.clone());
|
||||
}
|
||||
}
|
||||
journal(format!("refused: {room_id} {reason}"));
|
||||
out.refused.push(RefusedRoom { room_id: room_id.clone(), reason, event_id: None });
|
||||
}
|
||||
Err(error) => {
|
||||
journal(format!("huginn-sources: chat {room_id} facts failed, the others go on: {error:#}"));
|
||||
out.failed.push(room_id.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
if let Err(error) = retire_chat(index, &keep) {
|
||||
journal(format!("huginn-sources: retiring rooms the reader left failed, the others go on: {error:#}"));
|
||||
out.failed.push("retire".to_owned());
|
||||
}
|
||||
for facts in admitted {
|
||||
match ingest_room(embedder, index, history, identity.dimensions, &facts, &mut out) {
|
||||
Ok(Outcome::Done(report)) => {
|
||||
journal(report.to_string());
|
||||
out.rooms.push(report);
|
||||
}
|
||||
Ok(Outcome::Refused(refused)) => {
|
||||
journal(format!("refused: {} {} [{}]", refused.room_id, refused.reason, refused.event_id.as_deref().unwrap_or("")));
|
||||
out.refused.push(refused);
|
||||
}
|
||||
Err(error) => {
|
||||
journal(format!("huginn-sources: chat {} failed, the others go on: {error:#}", facts.room_id));
|
||||
out.failed.push(facts.room_id.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
for hit in &out.scrubbed {
|
||||
journal(format!("scrubbed: {} {} {}", hit.room_id, hit.event_id, hit.kind));
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
|
|
|
|||
103
crates/huginn-daemon/src/sources/chat_purge.rs
Normal file
103
crates/huginn-daemon/src/sources/chat_purge.rs
Normal file
|
|
@ -0,0 +1,103 @@
|
|||
//! A steward's purge of a room's time window. The tombstone is the one chat
|
||||
//! state that is not derived from Synapse: a durable request, kept in the
|
||||
//! collection's cursors object under `chat-purge:<room_id>`, that must outlive
|
||||
//! the points it removes, so a re-ingest never brings the window back.
|
||||
//!
|
||||
//! `request_purge` is the only writer; `refresh_chat` reads the tombstone and
|
||||
//! applies it at its next run. The code does not authenticate the steward: whoever
|
||||
//! can run the job as the unit's user (the host admin) may ask, and the request
|
||||
//! carries a ref in the job's journal for a trace. Tombstones live in the
|
||||
//! collection, so a full collection rebuild forgets them; re-issue them from
|
||||
//! their refs.
|
||||
|
||||
use anyhow::{Context, Result, bail};
|
||||
|
||||
use super::chat::CHAT_PURGE_PREFIX;
|
||||
use super::{DescribedSources, SourceIndex};
|
||||
use crate::index::INDEX_CALL_TIMEOUT;
|
||||
|
||||
/// The most windows a room's tombstone holds.
|
||||
pub const PURGE_WINDOWS_MAX: usize = 32;
|
||||
|
||||
/// Messages with `from_ms <= ts_ms < to_ms` are purged.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
|
||||
pub struct PurgeWindow {
|
||||
pub from_ms: u64,
|
||||
pub to_ms: u64,
|
||||
}
|
||||
|
||||
impl PurgeWindow {
|
||||
pub fn covers(&self, ts_ms: u64) -> bool {
|
||||
self.from_ms <= ts_ms && ts_ms < self.to_ms
|
||||
}
|
||||
}
|
||||
|
||||
/// The tombstone key of a room.
|
||||
pub fn purge_key(room_id: &str) -> String {
|
||||
format!("{CHAT_PURGE_PREFIX}{room_id}")
|
||||
}
|
||||
|
||||
/// A tombstone value, `<from>..<to>[,<from>..<to>...]`.
|
||||
pub fn parse_windows(value: &str) -> Result<Vec<PurgeWindow>> {
|
||||
value
|
||||
.split(',')
|
||||
.map(|part| {
|
||||
let (from, to) = part.split_once("..").with_context(|| format!("purge window {part:?} is not <from>..<to>"))?;
|
||||
let window = PurgeWindow { from_ms: from.parse().context("a purge window's start is not milliseconds")?, to_ms: to.parse().context("a purge window's end is not milliseconds")? };
|
||||
if window.from_ms >= window.to_ms {
|
||||
bail!("purge window {part:?} is empty");
|
||||
}
|
||||
Ok(window)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Windows sorted, with overlapping and touching ones joined.
|
||||
pub fn merge(mut windows: Vec<PurgeWindow>) -> Vec<PurgeWindow> {
|
||||
windows.sort();
|
||||
let mut merged: Vec<PurgeWindow> = Vec::new();
|
||||
for window in windows {
|
||||
match merged.last_mut() {
|
||||
Some(last) if window.from_ms <= last.to_ms => last.to_ms = last.to_ms.max(window.to_ms),
|
||||
_ => merged.push(window),
|
||||
}
|
||||
}
|
||||
merged
|
||||
}
|
||||
|
||||
pub fn encode(windows: &[PurgeWindow]) -> String {
|
||||
windows.iter().map(|window| format!("{}..{}", window.from_ms, window.to_ms)).collect::<Vec<_>>().join(",")
|
||||
}
|
||||
|
||||
/// Whether any window covers `ts_ms`.
|
||||
pub fn covers(windows: &[PurgeWindow], ts_ms: u64) -> bool {
|
||||
windows.iter().any(|window| window.covers(ts_ms))
|
||||
}
|
||||
|
||||
/// Records that `window` of `room_id` must leave the index and never return. Writes the cursors
|
||||
/// object only if the tombstone changed, and refuses a room's 33rd window. Writes no point and
|
||||
/// embeds nothing; the next `refresh_chat` applies it. The job's `chat-purge` subcommand calls
|
||||
/// it under the job's lock, because `set_cursors` replaces the cursors object whole.
|
||||
/// Returns whether the tombstone changed.
|
||||
pub fn request_purge<S: SourceIndex>(index: &mut S, room_id: &str, window: PurgeWindow) -> Result<bool> {
|
||||
if window.from_ms >= window.to_ms {
|
||||
bail!("a purge window must start before it ends");
|
||||
}
|
||||
let DescribedSources::Labelled { mut cursors, .. } = index.describe_sources(INDEX_CALL_TIMEOUT).context("describing the source collection")? else {
|
||||
bail!("the source collection is absent or not Huginn's; the forge refresh creates it");
|
||||
};
|
||||
let key = purge_key(room_id);
|
||||
let mut windows = cursors.get(&key).map(|held| parse_windows(held)).transpose()?.unwrap_or_default();
|
||||
windows.push(window);
|
||||
let windows = merge(windows);
|
||||
if windows.len() > PURGE_WINDOWS_MAX {
|
||||
bail!("room {room_id} already holds {PURGE_WINDOWS_MAX} purge windows; widen one instead");
|
||||
}
|
||||
let encoded = encode(&windows);
|
||||
if cursors.get(&key) == Some(&encoded) {
|
||||
return Ok(false);
|
||||
}
|
||||
cursors.insert(key, encoded);
|
||||
index.set_cursors(&cursors).context("writing the purge tombstone")?;
|
||||
Ok(true)
|
||||
}
|
||||
782
crates/huginn-daemon/src/sources/chat_tests.rs
Normal file
782
crates/huginn-daemon/src/sources/chat_tests.rs
Normal file
|
|
@ -0,0 +1,782 @@
|
|||
//! Chat ingest against a scripted room history, a fake vector store and the fake embedder.
|
||||
//! The refusals come first: what must never be read, embedded or stored.
|
||||
|
||||
use std::collections::{BTreeMap, BTreeSet};
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::{Result, bail};
|
||||
use huginn_mind::ChatMessage;
|
||||
use huginn_mind::redact::ScrubKind;
|
||||
|
||||
use super::chat::*;
|
||||
use super::chat_chunk::{CHAT_BODY_CHARS_MAX, chat_version, chunk_id};
|
||||
use super::chat_purge::{PURGE_WINDOWS_MAX, PurgeWindow, request_purge};
|
||||
use super::room_scope::{ESTATE_ALIASES, Estate, Refusal, RoomFacts};
|
||||
use super::*;
|
||||
use crate::index::MANAGED_BY;
|
||||
use crate::index::fakes::{FakeEmbedder, FakeSources};
|
||||
|
||||
// ---- the rig ----
|
||||
|
||||
struct Scripted {
|
||||
facts: RoomFacts,
|
||||
events: Vec<RoomEvent>,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct FakeHistory {
|
||||
joined: Vec<String>,
|
||||
rooms: BTreeMap<String, Scripted>,
|
||||
calls: Vec<String>,
|
||||
}
|
||||
|
||||
impl FakeHistory {
|
||||
fn room(&mut self, id: &str, events: Vec<RoomEvent>) -> &mut Self {
|
||||
self.rooms.insert(id.into(), Scripted { facts: RoomFacts { room_id: id.into(), ..RoomFacts::default() }, events });
|
||||
self.joined.push(id.into());
|
||||
self
|
||||
}
|
||||
|
||||
fn facts(&mut self, id: &str, edit: impl FnOnce(&mut RoomFacts)) -> &mut Self {
|
||||
edit(&mut self.rooms.get_mut(id).unwrap().facts);
|
||||
self
|
||||
}
|
||||
|
||||
fn push(&mut self, id: &str, events: Vec<RoomEvent>) {
|
||||
self.rooms.get_mut(id).unwrap().events.extend(events);
|
||||
}
|
||||
|
||||
fn called(&self, prefix: &str, room: &str) -> bool {
|
||||
self.calls.iter().any(|call| call == &format!("{prefix}:{room}"))
|
||||
}
|
||||
}
|
||||
|
||||
impl RoomHistory for FakeHistory {
|
||||
fn joined_rooms(&mut self) -> Result<Vec<String>> {
|
||||
self.calls.push("joined".into());
|
||||
Ok(self.joined.clone())
|
||||
}
|
||||
|
||||
fn facts(&mut self, room_id: &str) -> Result<RoomFacts> {
|
||||
self.calls.push(format!("facts:{room_id}"));
|
||||
match self.rooms.get(room_id) {
|
||||
Some(room) => Ok(room.facts.clone()),
|
||||
None => bail!("no such room"),
|
||||
}
|
||||
}
|
||||
|
||||
fn events_after(&mut self, room_id: &str, after: Option<&str>, limit: u32) -> Result<Option<EventPage>> {
|
||||
self.calls.push(format!("events:{room_id}"));
|
||||
let Some(room) = self.rooms.get(room_id) else { bail!("no such room") };
|
||||
let from = match after {
|
||||
None => 0,
|
||||
Some(after) => match room.events.iter().position(|event| event.event_id == after) {
|
||||
Some(at) => at + 1,
|
||||
None => return Ok(None),
|
||||
},
|
||||
};
|
||||
let rest = &room.events[from..];
|
||||
let take = rest.len().min(limit as usize);
|
||||
Ok(Some(EventPage { events: rest[..take].to_vec(), more: take < rest.len() }))
|
||||
}
|
||||
}
|
||||
|
||||
fn event(id: &str, sender: &str, ts_ms: u64, kind: EventKind) -> RoomEvent {
|
||||
RoomEvent { event_id: id.into(), sender: sender.into(), ts_ms, content_keys: vec!["body".into(), "msgtype".into()], kind }
|
||||
}
|
||||
|
||||
fn said(id: &str, ts_ms: u64, body: &str) -> RoomEvent {
|
||||
event(id, "@a:s", ts_ms, EventKind::Message { body: body.into() })
|
||||
}
|
||||
|
||||
fn edit(id: &str, sender: &str, ts_ms: u64, target: &str, body: &str) -> RoomEvent {
|
||||
event(id, sender, ts_ms, EventKind::Edit { target: target.into(), body: body.into() })
|
||||
}
|
||||
|
||||
fn redaction(id: &str, ts_ms: u64, target: &str) -> RoomEvent {
|
||||
event(id, "@a:s", ts_ms, EventKind::Redaction { target: target.into() })
|
||||
}
|
||||
|
||||
struct Rig {
|
||||
embedder: FakeEmbedder,
|
||||
index: FakeSources,
|
||||
estate: Estate,
|
||||
}
|
||||
|
||||
impl Rig {
|
||||
fn new() -> Self {
|
||||
let mut embedder = FakeEmbedder::new("d1");
|
||||
let index = FakeSources::default();
|
||||
let identity = crate::index::Embedder::model_identity(&mut embedder, Duration::from_secs(1)).unwrap();
|
||||
let meta = SourcesMeta {
|
||||
managed_by: MANAGED_BY.into(),
|
||||
corpus: CORPUS.into(),
|
||||
model: identity.name,
|
||||
model_digest: identity.digest,
|
||||
dimensions: identity.dimensions,
|
||||
chunker_version: chunk::CHUNKER_VERSION,
|
||||
};
|
||||
index.plant(DescribedSources::Labelled { meta, cursors: BTreeMap::new() });
|
||||
Self { embedder, index, estate: Estate::default() }
|
||||
}
|
||||
|
||||
fn run(&mut self, history: &mut FakeHistory) -> (Result<ChatRefreshed>, Vec<String>) {
|
||||
let mut journal = Vec::new();
|
||||
let out = refresh_chat(&mut self.embedder, &mut self.index, history, &self.estate, &mut |line| journal.push(line));
|
||||
(out, journal)
|
||||
}
|
||||
|
||||
fn ok(&mut self, history: &mut FakeHistory) -> ChatRefreshed {
|
||||
self.run(history).0.unwrap()
|
||||
}
|
||||
|
||||
fn points(&self) -> Vec<ChatPoint> {
|
||||
self.index.state.lock().unwrap().chat.values().cloned().collect()
|
||||
}
|
||||
|
||||
fn ids(&self) -> BTreeSet<String> {
|
||||
self.index.state.lock().unwrap().chat.keys().cloned().collect()
|
||||
}
|
||||
|
||||
fn bodies(&self) -> Vec<String> {
|
||||
let mut bodies: Vec<String> = self.points().iter().flat_map(|point| point.payload.messages.iter().map(|message| message.body.clone())).collect();
|
||||
bodies.sort();
|
||||
bodies
|
||||
}
|
||||
|
||||
fn held_ids(&self, room: &str) -> Vec<String> {
|
||||
let mut ids: Vec<String> =
|
||||
self.points().iter().filter(|point| point.payload.room_id == room).flat_map(|point| point.payload.messages.iter().map(|message| message.event_id.clone())).collect();
|
||||
ids.sort();
|
||||
ids
|
||||
}
|
||||
|
||||
fn cursor(&self, room: &str) -> Option<String> {
|
||||
self.index.cursors().get(&cursor_key(room)).cloned()
|
||||
}
|
||||
|
||||
fn embedded(&self) -> Vec<String> {
|
||||
self.embedder.texts()
|
||||
}
|
||||
|
||||
fn seed(&mut self, room: &str, event_id: &str, ts: u64) {
|
||||
let mut seeded = vec![ChatPoint {
|
||||
id: chunk_id(room, event_id),
|
||||
vector: vec![1.0; 4],
|
||||
payload: chunk_payload_of(room, &[(event_id, ts)]),
|
||||
}];
|
||||
self.index.upsert_chat(&seeded.split_off(0)).unwrap();
|
||||
let mut cursors = self.index.cursors();
|
||||
cursors.insert(cursor_key(room), format!("{}:{event_id}", chat_version()));
|
||||
self.index.set_cursors(&cursors).unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
fn chunk_payload_of(room: &str, messages: &[(&str, u64)]) -> ChatPayload {
|
||||
let messages: Vec<ChatMessage> = messages.iter().map(|(id, ts)| ChatMessage { event_id: (*id).into(), sender: "@a:s".into(), ts_ms: *ts, body: format!("seeded {id}") }).collect();
|
||||
ChatPayload { room_id: room.into(), room_alias: None, text: "seeded".into(), messages }
|
||||
}
|
||||
|
||||
fn tombstone(rig: &Rig, room: &str) -> Option<String> {
|
||||
rig.index.cursors().get(&format!("{CHAT_PURGE_PREFIX}{room}")).cloned()
|
||||
}
|
||||
|
||||
// ---- scope: refusals first ----
|
||||
|
||||
#[test]
|
||||
fn only_joined_rooms_are_looked_at_and_a_join_rule_is_not_the_test() {
|
||||
let mut history = FakeHistory::default();
|
||||
history.room("!a:s", vec![said("$a1", 1, "in a")]).room("!b:s", vec![said("$b1", 2, "in b")]);
|
||||
// A room the reader has not joined: scripted, but never listed.
|
||||
history.rooms.insert("!c:s".into(), Scripted { facts: RoomFacts { room_id: "!c:s".into(), ..RoomFacts::default() }, events: vec![said("$c1", 3, "in c")] });
|
||||
let mut rig = Rig::new();
|
||||
let done = rig.ok(&mut history);
|
||||
assert!(done.is_clean());
|
||||
assert!(!history.calls.iter().any(|call| call.contains("!c:s")), "{:?}", history.calls);
|
||||
assert_eq!(rig.held_ids("!a:s"), ["$a1"]);
|
||||
assert_eq!(rig.held_ids("!b:s"), ["$b1"]);
|
||||
assert!(rig.held_ids("!c:s").is_empty() && done.rooms.iter().all(|report| report.room_id != "!c:s"));
|
||||
let code = include_str!("room_scope.rs");
|
||||
assert!(!code[..code.find("#[cfg(test)]").unwrap()].contains(&["join", "_rule"].concat()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_dm_room_is_refused_unread_unindexed_and_purged() {
|
||||
let mut history = FakeHistory::default();
|
||||
history.room("!dm:s", vec![said("$d1", 1, "private")]).facts("!dm:s", |facts| facts.direct = true);
|
||||
let mut rig = Rig::new();
|
||||
rig.seed("!dm:s", "$old", 5);
|
||||
let (done, journal) = rig.run(&mut history);
|
||||
let done = done.unwrap();
|
||||
assert!(!history.called("events", "!dm:s"));
|
||||
assert!(rig.points().is_empty() && rig.cursor("!dm:s").is_none() && rig.embedded().is_empty());
|
||||
assert_eq!(done.refused, [RefusedRoom { room_id: "!dm:s".into(), reason: Refusal::Direct, event_id: None }]);
|
||||
assert!(!done.is_clean() && journal.iter().any(|line| line == "refused: !dm:s direct room"), "{journal:?}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn each_estate_room_is_refused_by_id_and_by_alias_and_an_excluded_room_in_the_index_is_purged() {
|
||||
let mut rig = Rig::new();
|
||||
rig.estate = Estate { ids: ["!byid:s".to_owned()].into() };
|
||||
let mut history = FakeHistory::default();
|
||||
history.room("!byid:s", vec![said("$i1", 1, "x")]);
|
||||
for (n, alias) in ESTATE_ALIASES.iter().map(|alias| alias.to_string()).chain(["#USERS:GAMECULT.ORG".to_owned()]).enumerate() {
|
||||
history.room(&format!("!al{n}:s"), vec![said("$x", 1, "x")]).facts(&format!("!al{n}:s"), |facts| facts.alias = Some(alias));
|
||||
}
|
||||
history.room("!alt:s", vec![said("$x", 1, "x")]).facts("!alt:s", |facts| facts.alt_aliases = vec!["#other:s".into(), "#patrons:gamecult.org".into()]);
|
||||
history.room("!ok:s", vec![said("$ok1", 1, "fine")]);
|
||||
history.room("!guide:s", vec![said("$g1", 1, "fine")]).facts("!guide:s", |facts| facts.alias = Some("#users-guide:gamecult.org".into()));
|
||||
history.room("!club:s", vec![said("$c1", 1, "fine")]).facts("!club:s", |facts| facts.alias = Some("#patrons-club:gamecult.org".into()));
|
||||
for room in ["!byid:s", "!al0:s", "!al1:s", "!al2:s", "!al3:s", "!alt:s"] {
|
||||
rig.seed(room, "$old", 5);
|
||||
}
|
||||
let mut cursors = rig.index.cursors();
|
||||
cursors.insert(format!("{CHAT_PURGE_PREFIX}!al0:s"), "1..2".into());
|
||||
rig.index.set_cursors(&cursors).unwrap();
|
||||
let done = rig.ok(&mut history);
|
||||
assert!(!history.called("facts", "!byid:s") && !history.called("events", "!byid:s"));
|
||||
for room in ["!al0:s", "!al1:s", "!al2:s", "!al3:s", "!alt:s"] {
|
||||
assert!(history.called("facts", room) && !history.called("events", room), "{room}");
|
||||
}
|
||||
let refused: BTreeSet<&str> = done.refused.iter().map(|refused| refused.room_id.as_str()).collect();
|
||||
assert_eq!(refused, ["!al0:s", "!al1:s", "!al2:s", "!al3:s", "!alt:s", "!byid:s"].into());
|
||||
assert!(done.refused.iter().all(|refused| refused.reason == Refusal::Estate) && !done.is_clean());
|
||||
let held: BTreeSet<String> = rig.points().iter().map(|point| point.payload.room_id.clone()).collect();
|
||||
assert_eq!(held, ["!guide:s".to_owned(), "!club:s".to_owned(), "!ok:s".to_owned()].into(), "lookalikes and plain rooms are in; every estate room is out");
|
||||
assert!(rig.index.cursors().keys().all(|key| !key.starts_with("chat:!al") && !key.starts_with("chat:!byid") && !key.starts_with("chat:!alt")));
|
||||
assert_eq!(tombstone(&rig, "!al0:s").as_deref(), Some("1..2"), "a tombstone is never touched by a refusal");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_encrypted_room_is_refused_and_an_encrypted_event_in_a_plain_room_is_never_stored() {
|
||||
let mut history = FakeHistory::default();
|
||||
history.room("!enc:s", vec![said("$e1", 1, "x")]).facts("!enc:s", |facts| facts.encrypted = true);
|
||||
history.room("!plain:s", vec![said("$p1", 1, "visible"), event("$p2", "@a:s", 2, EventKind::Other), said("$p3", 3, "also visible")]);
|
||||
let mut rig = Rig::new();
|
||||
let done = rig.ok(&mut history);
|
||||
assert_eq!(done.refused[0].reason, Refusal::Encrypted);
|
||||
assert!(!history.called("events", "!enc:s"));
|
||||
assert_eq!(rig.held_ids("!plain:s"), ["$p1", "$p3"]);
|
||||
assert_eq!(rig.cursor("!plain:s").unwrap(), format!("{}:$p3", chat_version()), "the cursor passes an Other event");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_bridged_room_is_refused_whole_before_any_event_is_read_and_its_chunks_are_purged() {
|
||||
let mut history = FakeHistory::default();
|
||||
for (n, state_type) in ["m.bridge", "uk.half-shot.bridge"].into_iter().enumerate() {
|
||||
let room = format!("!br{n}:s");
|
||||
history.room(&room, vec![said("$x", 1, "x")]).facts(&room, |facts| facts.bridge = Some(state_type.into()));
|
||||
}
|
||||
let mut rig = Rig::new();
|
||||
rig.seed("!br0:s", "$old", 5);
|
||||
let done = rig.ok(&mut history);
|
||||
assert_eq!(done.refused.len(), 2);
|
||||
assert!(done.refused.iter().all(|refused| matches!(refused.reason, Refusal::Bridged(_))));
|
||||
assert!(!history.calls.iter().any(|call| call.starts_with("events:")));
|
||||
assert!(rig.points().is_empty() && rig.index.cursors().is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_bridged_platforms_sender_or_content_refuses_the_room_before_anything_is_embedded_and_purges_what_it_held() {
|
||||
let cases = [
|
||||
event("$bad", "@discord_123:s", 2, EventKind::Message { body: "SECRET-BODY-TEXT".into() }),
|
||||
event("$bad", "@_discord_9:s", 2, EventKind::Message { body: "SECRET-BODY-TEXT".into() }),
|
||||
event("$bad", "@discordbot:s", 2, EventKind::Message { body: "SECRET-BODY-TEXT".into() }),
|
||||
RoomEvent { content_keys: vec!["body".into(), "fi.mau.discord".into()], ..said("$bad", 2, "SECRET-BODY-TEXT") },
|
||||
RoomEvent { content_keys: vec!["uk.half-shot.discord".into()], ..said("$bad", 2, "SECRET-BODY-TEXT") },
|
||||
];
|
||||
for bad in cases {
|
||||
let mut history = FakeHistory::default();
|
||||
history.room("!r:s", vec![said("$ok", 1, "before"), bad]);
|
||||
history.room("!other:s", vec![said("$o1", 1, "other room")]);
|
||||
let mut rig = Rig::new();
|
||||
rig.seed("!r:s", "$prev", 0);
|
||||
let (done, journal) = rig.run(&mut history);
|
||||
let done = done.unwrap();
|
||||
assert!(!rig.embedded().iter().any(|text| text.contains("SECRET-BODY-TEXT") || text.contains("before")), "nothing of the page is embedded");
|
||||
assert!(rig.held_ids("!r:s").is_empty() && rig.cursor("!r:s").is_none());
|
||||
assert_eq!(rig.held_ids("!other:s"), ["$o1"]);
|
||||
assert_eq!(done.refused.len(), 1);
|
||||
assert_eq!((done.refused[0].room_id.as_str(), done.refused[0].event_id.as_deref()), ("!r:s", Some("$bad")));
|
||||
assert!(!done.is_clean());
|
||||
assert!(journal.iter().all(|line| !line.contains("SECRET-BODY-TEXT")), "{journal:?}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_lookalike_of_the_bridged_platform_is_kept() {
|
||||
let mut history = FakeHistory::default();
|
||||
let kept = ["@discourse:s", "@disco:s", "@eureka:s"].map(|sender| event(&format!("${}", &sender[1..6]), sender, 1, EventKind::Message { body: "hi".into() }));
|
||||
let mut keyed = said("$k", 2, "hi");
|
||||
keyed.content_keys.push("discourse".into());
|
||||
let mut events = kept.to_vec();
|
||||
events.push(keyed);
|
||||
history.room("!r:s", events);
|
||||
let mut rig = Rig::new();
|
||||
assert!(rig.ok(&mut history).is_clean());
|
||||
assert_eq!(rig.held_ids("!r:s").len(), 4);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_collection_is_never_created_by_chat() {
|
||||
let mut history = FakeHistory::default();
|
||||
history.room("!r:s", vec![said("$a", 1, "x")]);
|
||||
for described in [None, Some(DescribedSources::Absent), Some(DescribedSources::Unlabelled)] {
|
||||
let mut rig = Rig::new();
|
||||
match described {
|
||||
None => rig.index.state.lock().unwrap().described = None,
|
||||
Some(described) => rig.index.plant(described),
|
||||
}
|
||||
let error = format!("{:#}", rig.run(&mut history).0.unwrap_err());
|
||||
assert!(error.contains("forge refresh creates it"), "{error}");
|
||||
assert_eq!(rig.index.state.lock().unwrap().recreated, 0);
|
||||
assert!(rig.points().is_empty() && rig.embedded().is_empty());
|
||||
}
|
||||
}
|
||||
|
||||
// ---- redactions, edits, replays ----
|
||||
|
||||
#[test]
|
||||
fn a_redacted_event_is_absent_after_the_next_refresh_and_a_first_or_only_message_changes_or_drops_its_chunk() {
|
||||
let mut history = FakeHistory::default();
|
||||
history.room("!r:s", vec![said("$a", 10, "alpha"), said("$b", 20, "beta"), said("$c", 30, "gamma")]);
|
||||
let mut rig = Rig::new();
|
||||
rig.ok(&mut history);
|
||||
let first_id = chunk_id("!r:s", "$a");
|
||||
assert_eq!(rig.ids(), [first_id.clone()].into());
|
||||
history.push("!r:s", vec![redaction("$r1", 40, "$b")]);
|
||||
rig.ok(&mut history);
|
||||
assert_eq!(rig.held_ids("!r:s"), ["$a", "$c"]);
|
||||
assert!(!rig.points()[0].payload.text.contains("beta"));
|
||||
history.push("!r:s", vec![redaction("$r2", 50, "$a")]);
|
||||
rig.ok(&mut history);
|
||||
assert_eq!(rig.ids(), [chunk_id("!r:s", "$c")].into(), "the first message left, so the id changed and the old one is gone");
|
||||
history.push("!r:s", vec![redaction("$r3", 60, "$c")]);
|
||||
rig.ok(&mut history);
|
||||
assert!(rig.points().is_empty(), "a chunk the redaction empties is deleted");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_redaction_in_the_same_page_as_its_message_never_reaches_the_embedder() {
|
||||
let mut history = FakeHistory::default();
|
||||
history.room("!r:s", vec![said("$a", 10, "keep me"), said("$gone", 20, "REDACTED-BODY"), redaction("$r", 30, "$gone")]);
|
||||
let mut rig = Rig::new();
|
||||
rig.ok(&mut history);
|
||||
assert!(rig.embedded().iter().all(|text| !text.contains("REDACTED-BODY")));
|
||||
assert_eq!(rig.held_ids("!r:s"), ["$a"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_edit_replaces_the_body_in_place_and_a_stranger_an_unknown_target_or_an_edit_of_an_edit_change_nothing() {
|
||||
let mut history = FakeHistory::default();
|
||||
history.room("!r:s", vec![said("$a", 10, "old words"), said("$b", 20, "other")]);
|
||||
let mut rig = Rig::new();
|
||||
rig.ok(&mut history);
|
||||
let before = rig.points();
|
||||
history.push("!r:s", vec![edit("$e1", "@a:s", 30, "$a", "new words")]);
|
||||
rig.ok(&mut history);
|
||||
let after = rig.points();
|
||||
assert_eq!(after.len(), 1);
|
||||
let first = &after[0].payload.messages[0];
|
||||
assert_eq!((first.event_id.as_str(), first.sender.as_str(), first.ts_ms, first.body.as_str()), ("$a", "@a:s", 10, "new words"));
|
||||
assert_eq!(after[0].id, before[0].id);
|
||||
assert!(after[0].payload.text.contains("new words") && !after[0].payload.text.contains("old words"));
|
||||
assert!(rig.embedded().last().unwrap().contains("new words"));
|
||||
let calls = rig.embedder.state.lock().unwrap().embedded.len();
|
||||
history.push("!r:s", vec![edit("$e2", "@stranger:s", 40, "$a", "hijack"), edit("$e3", "@a:s", 41, "$nothing", "lost"), edit("$e4", "@a:s", 42, "$e1", "edit of an edit")]);
|
||||
rig.ok(&mut history);
|
||||
assert_eq!(rig.points(), after);
|
||||
assert_eq!(rig.embedder.state.lock().unwrap().embedded.len(), calls, "no embed call for edits that change nothing");
|
||||
assert_eq!(rig.cursor("!r:s").unwrap(), format!("{}:$e4", chat_version()), "the cursor is the last event of any kind");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn replay_after_a_crash_between_the_upsert_and_the_cursor_neither_duplicates_nor_loses() {
|
||||
let mut history = FakeHistory::default();
|
||||
history.room("!r:s", vec![said("$a", 10, "alpha"), said("$b", 20, "beta"), edit("$e", "@a:s", 25, "$a", "alpha two"), said("$c", 30, "gamma")]);
|
||||
let mut rig = Rig::new();
|
||||
// One cursor write (the empty start) succeeds, then the page's upsert is acknowledged and the next write fails.
|
||||
rig.index.state.lock().unwrap().fail_cursor_writes_after = Some(1);
|
||||
assert!(!rig.run(&mut history).0.unwrap().failed.is_empty());
|
||||
let (ids, held) = (rig.ids(), rig.held_ids("!r:s"));
|
||||
assert_eq!(held, ["$a", "$b", "$c"]);
|
||||
rig.index.state.lock().unwrap().fail_cursor_writes_after = None;
|
||||
let done = rig.ok(&mut history);
|
||||
assert!(done.is_clean());
|
||||
assert_eq!((rig.ids(), rig.held_ids("!r:s")), (ids, held));
|
||||
assert!(rig.bodies().contains(&"alpha two".to_owned()));
|
||||
assert_eq!(rig.cursor("!r:s").unwrap(), format!("{}:$c", chat_version()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_failed_upsert_leaves_the_cursor_and_a_failed_embed_or_delete_deletes_nothing_the_index_held() {
|
||||
let mut history = FakeHistory::default();
|
||||
history.room("!r:s", vec![said("$a", 10, "alpha"), said("$b", 20, "beta")]);
|
||||
let mut rig = Rig::new();
|
||||
rig.ok(&mut history);
|
||||
let (cursor, points) = (rig.cursor("!r:s"), rig.points());
|
||||
history.push("!r:s", vec![said("$c", 30, "gamma")]);
|
||||
rig.index.state.lock().unwrap().fail_chat_upserts = true;
|
||||
assert_eq!(rig.run(&mut history).0.unwrap().failed, ["!r:s"]);
|
||||
assert_eq!((rig.cursor("!r:s"), rig.points()), (cursor.clone(), points.clone()));
|
||||
rig.index.state.lock().unwrap().fail_chat_upserts = false;
|
||||
rig.embedder.state.lock().unwrap().embed_fails = true;
|
||||
assert_eq!(rig.run(&mut history).0.unwrap().failed, ["!r:s"]);
|
||||
assert_eq!((rig.cursor("!r:s"), rig.points()), (cursor.clone(), points.clone()));
|
||||
rig.embedder.state.lock().unwrap().embed_fails = false;
|
||||
// A redaction of the first message writes a new chunk, then fails to delete the old one: the cursor stays.
|
||||
history.push("!r:s", vec![redaction("$r", 40, "$a")]);
|
||||
rig.index.state.lock().unwrap().fail_chat_deletes = true;
|
||||
assert_eq!(rig.run(&mut history).0.unwrap().failed, ["!r:s"]);
|
||||
assert_eq!(rig.cursor("!r:s"), cursor);
|
||||
rig.index.state.lock().unwrap().fail_chat_deletes = false;
|
||||
assert!(rig.ok(&mut history).is_clean());
|
||||
assert_eq!(rig.held_ids("!r:s"), ["$b", "$c"]);
|
||||
}
|
||||
|
||||
// ---- chunking through the job ----
|
||||
|
||||
#[test]
|
||||
fn appending_to_the_tail_over_many_runs_equals_one_run_over_all() {
|
||||
let all: Vec<RoomEvent> = (0..80).map(|n| said(&format!("$m{n:03}"), 1_000 * n + if n >= 40 { 5_000_000 } else { 0 }, &format!("message number {n} {}", "x".repeat((n as usize * 37) % 400)))).collect();
|
||||
let mut whole_history = FakeHistory::default();
|
||||
whole_history.room("!r:s", all.clone());
|
||||
let mut whole = Rig::new();
|
||||
whole.ok(&mut whole_history);
|
||||
let mut grown_history = FakeHistory::default();
|
||||
grown_history.room("!r:s", vec![]);
|
||||
let mut grown = Rig::new();
|
||||
for batch in all.chunks(7) {
|
||||
grown_history.push("!r:s", batch.to_vec());
|
||||
grown.ok(&mut grown_history);
|
||||
}
|
||||
assert!(whole.points().len() > 2);
|
||||
let shape = |rig: &Rig| rig.points().into_iter().map(|point| (point.id, point.payload)).collect::<Vec<_>>();
|
||||
assert_eq!(shape(&grown), shape(&whole));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_long_message_is_cut_with_a_marker_and_every_point_satisfies_the_payload_contract() {
|
||||
let mut history = FakeHistory::default();
|
||||
history.room("!r:s", vec![said("$a", 10, &"é".repeat(CHAT_BODY_CHARS_MAX * 2)), said("$b", 20, "short"), said("$c", 99_000_000, "later")]);
|
||||
history.facts("!r:s", |facts| {
|
||||
facts.alias = Some("#swarm:gamecult.org".into());
|
||||
facts.name = Some("Swarm".into());
|
||||
});
|
||||
let mut rig = Rig::new();
|
||||
rig.ok(&mut history);
|
||||
let points = rig.points();
|
||||
assert_eq!(points.len(), 2);
|
||||
for point in &points {
|
||||
let payload = &point.payload;
|
||||
assert_eq!((payload.room_id.as_str(), payload.room_alias.as_deref()), ("!r:s", Some("#swarm:gamecult.org")));
|
||||
assert_eq!(point.id, chunk_id("!r:s", &payload.messages[0].event_id));
|
||||
assert_eq!(payload.text, super::chat_chunk::embedded_text("Swarm", &payload.messages));
|
||||
assert!(!payload.messages.is_empty());
|
||||
}
|
||||
let long = &points.iter().find(|point| point.payload.messages[0].event_id == "$a").unwrap().payload.messages[0];
|
||||
assert_eq!(long.body.chars().count(), CHAT_BODY_CHARS_MAX);
|
||||
assert!(long.body.ends_with("[cut]"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_lost_cursor_event_or_a_version_bump_re_ingests_that_room_alone() {
|
||||
let mut history = FakeHistory::default();
|
||||
history.room("!r:s", vec![said("$a", 10, "alpha"), said("$b", 20, "beta")]).room("!q:s", vec![said("$q1", 10, "quiet")]);
|
||||
let mut rig = Rig::new();
|
||||
rig.ok(&mut history);
|
||||
let q_ids: BTreeSet<String> = rig.points().iter().filter(|point| point.payload.room_id == "!q:s").map(|point| point.id.clone()).collect();
|
||||
let recreated = rig.index.state.lock().unwrap().recreated;
|
||||
for bump in [Some("0.0".to_owned()), None] {
|
||||
let mut cursors = rig.index.cursors();
|
||||
let value = match &bump {
|
||||
Some(old) => format!("{old}:$b"),
|
||||
None => format!("{}:$gone-from-the-server", chat_version()),
|
||||
};
|
||||
cursors.insert(cursor_key("!r:s"), value);
|
||||
rig.index.set_cursors(&cursors).unwrap();
|
||||
let before = rig.embedded().len();
|
||||
assert!(rig.ok(&mut history).is_clean());
|
||||
assert_eq!(rig.held_ids("!r:s"), ["$a", "$b"]);
|
||||
assert!(rig.embedded().len() > before, "the room was re-embedded");
|
||||
assert_eq!(rig.points().iter().filter(|point| point.payload.room_id == "!q:s").map(|point| point.id.clone()).collect::<BTreeSet<_>>(), q_ids);
|
||||
assert_eq!(rig.cursor("!r:s").unwrap(), format!("{}:$b", chat_version()));
|
||||
}
|
||||
assert_eq!(rig.index.state.lock().unwrap().recreated, recreated);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn retire_chat_removes_rooms_no_longer_kept_and_given_none_all_chat_and_touches_nothing_else() {
|
||||
let mut rig = Rig::new();
|
||||
rig.index.upsert_sources(&[SourcePoint {
|
||||
id: "forge-1".into(),
|
||||
vector: vec![1.0; 4],
|
||||
payload: SourcePayload { repo: "Repo".into(), path: "a".into(), commit: "c".into(), blob: "b".into(), line_start: 1, line_end: 1, chunk_index: 0, ext: "rs".into(), text: "t".into() },
|
||||
}])
|
||||
.unwrap();
|
||||
for room in ["!a:s", "!b:s"] {
|
||||
rig.seed(room, "$e", 1);
|
||||
}
|
||||
let mut cursors = rig.index.cursors();
|
||||
cursors.insert("Repo".into(), "c".into());
|
||||
cursors.insert(format!("{CHAT_PURGE_PREFIX}!a:s"), "1..2".into());
|
||||
rig.index.set_cursors(&cursors).unwrap();
|
||||
let retired = retire_chat(&mut rig.index, &["!a:s".to_owned()].into()).unwrap();
|
||||
assert_eq!(retired, ["!b:s"]);
|
||||
assert_eq!(rig.held_ids("!a:s"), ["$e"]);
|
||||
assert!(rig.held_ids("!b:s").is_empty() && rig.cursor("!b:s").is_none());
|
||||
assert_eq!(retire_chat(&mut rig.index, &BTreeSet::new()).unwrap(), ["!a:s"]);
|
||||
assert!(rig.points().is_empty());
|
||||
assert_eq!(rig.index.state.lock().unwrap().points.len(), 1, "a forge point stays");
|
||||
let kept: Vec<String> = rig.index.cursors().into_keys().collect();
|
||||
assert_eq!(kept, ["Repo".to_owned(), format!("{CHAT_PURGE_PREFIX}!a:s")]);
|
||||
// A room the reader left is retired by the refresh itself.
|
||||
let mut history = FakeHistory::default();
|
||||
history.room("!r:s", vec![said("$a", 1, "x")]);
|
||||
rig.seed("!gone:s", "$e", 1);
|
||||
rig.ok(&mut history);
|
||||
assert!(rig.held_ids("!gone:s").is_empty());
|
||||
}
|
||||
|
||||
// ---- the credential scrub through the job ----
|
||||
|
||||
fn planted() -> Vec<(ScrubKind, String)> {
|
||||
vec![
|
||||
(ScrubKind::AuthorizationHeader, format!("Authorization: Bearer {}", "a1B2".repeat(8))),
|
||||
(ScrubKind::Bearer, format!("Bearer {}", "z9Y8".repeat(6))),
|
||||
(ScrubKind::KeyShape, format!("sk-{}", "q7W3".repeat(8))),
|
||||
(ScrubKind::KeyShape, format!("sk_live_{}", "m4N5".repeat(6))),
|
||||
(ScrubKind::KeyShape, format!("ghp_{}", "k2L9".repeat(9))),
|
||||
(ScrubKind::KeyShape, format!("github_pat_{}_{}", "A1b2C3".repeat(4) + "ab", "d4E5f6".repeat(9) + "abcde")),
|
||||
(ScrubKind::KeyShape, format!("glpat-{}", "r6T7".repeat(5))),
|
||||
(ScrubKind::KeyShape, "xoxb-123456789012-987654321098-aBcDeFgHiJkLmNoP".into()),
|
||||
(ScrubKind::KeyShape, "AKIA0123456789ABCDEF".into()),
|
||||
(ScrubKind::KeyShape, format!("AIza{}", "Sy1_".repeat(9))),
|
||||
(ScrubKind::Jwt, "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJmYWtlIn0.c2lnbmF0dXJlZmFrZQ".into()),
|
||||
(ScrubKind::PemBlock, "-----BEGIN RSA PRIVATE KEY-----\nMIIEowIBAAKCAQEAfake\n-----END RSA PRIVATE KEY-----".into()),
|
||||
(ScrubKind::PemBlock, "-----BEGIN PRIVATE KEY-----\nMIIBVQIBADANfakeNOEND".into()),
|
||||
(ScrubKind::UrlUserinfo, "https://user:hunter2hunter2@host.example/r.git".into()),
|
||||
(ScrubKind::UrlUserinfo, "https://tokenvalue12345@host.example/".into()),
|
||||
]
|
||||
}
|
||||
|
||||
/// The part of a planted secret no replacement can contain.
|
||||
fn distinct(secret: &str) -> String {
|
||||
let bare = secret.rsplit(['@', ' ', '\n']).next().filter(|_| !secret.contains("://")).unwrap_or(secret);
|
||||
let source = if secret.contains("://") { secret.split("://").nth(1).unwrap().split('@').next().unwrap().rsplit(':').next().unwrap() } else { bare };
|
||||
source.chars().rev().take(12).collect::<Vec<_>>().into_iter().rev().collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn planted_credentials_never_reach_the_chunk_the_embedder_or_the_journal_in_a_message_or_an_edit() {
|
||||
for (kind, secret) in planted() {
|
||||
let needle = distinct(&secret);
|
||||
assert!(needle.len() >= 8, "{secret}");
|
||||
for shape in ["see {} now", "see {}", "{}\nnext line"] {
|
||||
let (dirty, clean) = (shape.replace("{}", &secret), "nothing here".to_owned());
|
||||
let mut history = FakeHistory::default();
|
||||
history.room("!r:s", vec![said("$m", 10, &dirty), said("$e", 20, &clean), edit("$ed", "@a:s", 30, "$e", &dirty)]);
|
||||
let mut rig = Rig::new();
|
||||
let (done, journal) = rig.run(&mut history);
|
||||
let done = done.unwrap();
|
||||
assert!(done.is_clean(), "a scrub hit does not make a run unclean");
|
||||
let stored = format!("{:?}", rig.points().iter().map(|point| (&point.payload.messages, &point.payload.text)).collect::<Vec<_>>());
|
||||
for seen in [stored, rig.embedded().join("\n"), journal.join("\n")] {
|
||||
assert!(!seen.contains(&needle), "{kind}: {secret:?} leaked into {seen}");
|
||||
}
|
||||
assert!(rig.bodies().iter().filter(|body| body.contains("<redacted>")).count() >= 2, "{kind}: both bodies show the replacement");
|
||||
let hits: Vec<(&str, ScrubKind)> = done.scrubbed.iter().map(|hit| (hit.event_id.as_str(), hit.kind)).collect();
|
||||
assert!(hits.contains(&("$m", kind)) && hits.contains(&("$ed", kind)), "{kind}: {hits:?}");
|
||||
assert!(done.scrubbed.iter().all(|hit| hit.room_id == "!r:s"));
|
||||
assert!(journal.contains(&format!("scrubbed: !r:s $m {kind}")), "{journal:?}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_scrub_meets_edits_redactions_the_body_cut_replays_and_version_bumps() {
|
||||
let token = format!("ghp_{}", "k2L9".repeat(9));
|
||||
let mut history = FakeHistory::default();
|
||||
history.room(
|
||||
"!r:s",
|
||||
vec![
|
||||
said("$clean", 10, "all clean"),
|
||||
said("$dirty", 20, &format!("token {token} inside")),
|
||||
edit("$adds", "@a:s", 30, "$clean", &format!("now with {token}")),
|
||||
edit("$removes", "@a:s", 40, "$dirty", "token removed"),
|
||||
said("$gone", 50, &format!("secret {token}")),
|
||||
redaction("$r", 60, "$gone"),
|
||||
said("$straddle", 70, &format!("{} {token} {}", "x".repeat(CHAT_BODY_CHARS_MAX - 10), "y".repeat(100))),
|
||||
],
|
||||
);
|
||||
let mut rig = Rig::new();
|
||||
let done = rig.ok(&mut history);
|
||||
assert!(rig.bodies().iter().all(|body| !body.contains("k2L9k2L9")));
|
||||
assert!(rig.embedded().iter().all(|text| !text.contains("k2L9k2L9")));
|
||||
let by_id: BTreeMap<String, String> = rig.points().iter().flat_map(|point| point.payload.messages.iter().map(|message| (message.event_id.clone(), message.body.clone()))).collect();
|
||||
assert_eq!(by_id["$clean"], "now with <redacted>");
|
||||
assert_eq!(by_id["$dirty"], "token removed");
|
||||
assert!(!by_id.contains_key("$gone"));
|
||||
assert!(by_id["$straddle"].chars().count() <= CHAT_BODY_CHARS_MAX && !by_id["$straddle"].contains("ghp_"));
|
||||
assert!(done.scrubbed.iter().any(|hit| hit.event_id == "$adds") && done.scrubbed.iter().any(|hit| hit.event_id == "$gone"));
|
||||
let (ids, texts) = (rig.ids(), rig.points().into_iter().map(|point| point.payload.text).collect::<Vec<_>>());
|
||||
// Replay from the start, as after a lost cursor: identical ids and text.
|
||||
let mut cursors = rig.index.cursors();
|
||||
cursors.remove(&cursor_key("!r:s"));
|
||||
rig.index.set_cursors(&cursors).unwrap();
|
||||
rig.ok(&mut history);
|
||||
assert_eq!((rig.ids(), rig.points().into_iter().map(|point| point.payload.text).collect::<Vec<_>>()), (ids, texts));
|
||||
// An older scrub version re-ingests the room and scrubs it again from the history.
|
||||
let mut cursors = rig.index.cursors();
|
||||
cursors.insert(cursor_key("!r:s"), format!("{}.0:$straddle", huginn_mind::redact::SCRUB_VERSION - 1));
|
||||
rig.index.set_cursors(&cursors).unwrap();
|
||||
rig.ok(&mut history);
|
||||
assert!(rig.bodies().iter().all(|body| !body.contains("k2L9k2L9")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ordinary_text_is_stored_byte_for_byte() {
|
||||
let mut history = FakeHistory::default();
|
||||
let same = "risk-averse task-force sk- 3f2b8c1d9e7a4b5c6d0e1f2a3b4c5d6e7f8a9b0c !abc:s user@example.com ssh://git@host/r.git the bearer of bad news";
|
||||
history.room("!r:s", vec![said("$a", 1, same)]);
|
||||
let mut rig = Rig::new();
|
||||
let done = rig.ok(&mut history);
|
||||
assert_eq!(rig.bodies(), [same]);
|
||||
assert!(done.scrubbed.is_empty());
|
||||
}
|
||||
|
||||
// ---- steward purge ----
|
||||
|
||||
fn window(from_ms: u64, to_ms: u64) -> PurgeWindow {
|
||||
PurgeWindow { from_ms, to_ms }
|
||||
}
|
||||
|
||||
fn purge_history() -> FakeHistory {
|
||||
let mut history = FakeHistory::default();
|
||||
// t1 = 1000, t2 = 3000: $b at t1 and $c inside are purged, $d at t2 is kept.
|
||||
history.room("!r:s", vec![said("$a", 500, "before"), said("$b", 1000, "PURGED-B"), said("$c", 2000, "PURGED-C"), said("$d", 3000, "after"), said("$e", 4000, "later")]);
|
||||
history.room("!other:s", vec![said("$o1", 1000, "same time other room"), said("$o2", 2000, "same time")]);
|
||||
history
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_purge_window_removes_exactly_its_events_and_re_chunks_the_neighbours() {
|
||||
let mut history = purge_history();
|
||||
let mut rig = Rig::new();
|
||||
rig.ok(&mut history);
|
||||
let other_ids: BTreeSet<String> = rig.points().iter().filter(|point| point.payload.room_id == "!other:s").map(|point| point.id.clone()).collect();
|
||||
assert_eq!(rig.held_ids("!r:s"), ["$a", "$b", "$c", "$d", "$e"]);
|
||||
assert!(request_purge(&mut rig.index, "!r:s", window(1000, 3000)).unwrap());
|
||||
let embedded_before = rig.embedded().len();
|
||||
rig.ok(&mut history);
|
||||
assert_eq!(rig.held_ids("!r:s"), ["$a", "$d", "$e"], "t1 is inside, t2 is not");
|
||||
assert!(rig.embedded()[embedded_before..].iter().all(|text| !text.contains("PURGED")));
|
||||
assert!(rig.points().iter().all(|point| !point.payload.text.contains("PURGED")));
|
||||
assert_eq!(rig.points().iter().filter(|point| point.payload.room_id == "!other:s").map(|point| point.id.clone()).collect::<BTreeSet<_>>(), other_ids);
|
||||
assert_eq!(rig.held_ids("!other:s"), ["$o1", "$o2"]);
|
||||
assert_eq!(tombstone(&rig, "!r:s").as_deref(), Some("1000..3000"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_purged_first_message_changes_the_chunk_id_and_an_emptied_chunk_is_deleted() {
|
||||
let mut history = FakeHistory::default();
|
||||
history.room("!r:s", vec![said("$a", 1000, "x"), said("$b", 2000, "y"), said("$c", 99_000_000, "far")]);
|
||||
let mut rig = Rig::new();
|
||||
rig.ok(&mut history);
|
||||
assert_eq!(rig.ids().len(), 2);
|
||||
request_purge(&mut rig.index, "!r:s", window(900, 1500)).unwrap();
|
||||
rig.ok(&mut history);
|
||||
assert_eq!(rig.ids(), [chunk_id("!r:s", "$b"), chunk_id("!r:s", "$c")].into());
|
||||
request_purge(&mut rig.index, "!r:s", window(99_000_000, 99_000_001)).unwrap();
|
||||
rig.ok(&mut history);
|
||||
assert_eq!(rig.held_ids("!r:s"), ["$b"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_purged_window_does_not_come_back() {
|
||||
let mut history = purge_history();
|
||||
let mut rig = Rig::new();
|
||||
rig.ok(&mut history);
|
||||
request_purge(&mut rig.index, "!r:s", window(1000, 3000)).unwrap();
|
||||
rig.ok(&mut history);
|
||||
let want = ["$a", "$d", "$e"];
|
||||
// New events, an in-window edit of an outside message, a redaction.
|
||||
history.push("!r:s", vec![said("$f", 5000, "new"), edit("$ed", "@a:s", 2500, "$a", "EDIT-IN-WINDOW"), redaction("$rd", 6000, "$e")]);
|
||||
rig.ok(&mut history);
|
||||
assert_eq!(rig.held_ids("!r:s"), ["$a", "$d", "$f"]);
|
||||
assert!(rig.bodies().contains(&"before".to_owned()) && !rig.bodies().iter().any(|body| body.contains("EDIT-IN-WINDOW")), "the in-window edit is dropped and the original stays");
|
||||
// A lost cursor re-ingests from the start; a version bump re-ingests too.
|
||||
for cursor in [None, Some("0.0:$f".to_owned())] {
|
||||
let mut cursors = rig.index.cursors();
|
||||
match cursor {
|
||||
None => cursors.remove(&cursor_key("!r:s")),
|
||||
Some(old) => cursors.insert(cursor_key("!r:s"), old),
|
||||
};
|
||||
rig.index.set_cursors(&cursors).unwrap();
|
||||
rig.ok(&mut history);
|
||||
assert!(rig.bodies().iter().all(|body| !body.contains("PURGED")), "{:?}", rig.held_ids("!r:s"));
|
||||
assert_eq!(rig.held_ids("!r:s"), ["$a", "$d", "$f"], "$e stays redacted by the history; {want:?} before the new events");
|
||||
}
|
||||
// The reader leaves the room and joins again.
|
||||
let mut left = FakeHistory::default();
|
||||
left.room("!other:s", vec![said("$o1", 1000, "x")]);
|
||||
rig.ok(&mut left);
|
||||
assert!(rig.held_ids("!r:s").is_empty() && tombstone(&rig, "!r:s").is_some());
|
||||
rig.ok(&mut history);
|
||||
assert!(rig.bodies().iter().all(|body| !body.contains("PURGED")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_purge_of_an_unindexed_window_writes_only_the_tombstone_and_still_holds() {
|
||||
let mut rig = Rig::new();
|
||||
let writes = rig.index.state.lock().unwrap().cursor_writes.len();
|
||||
assert!(request_purge(&mut rig.index, "!never:s", window(10, 20)).unwrap());
|
||||
assert_eq!(rig.index.state.lock().unwrap().cursor_writes.len(), writes + 1);
|
||||
assert!(rig.points().is_empty() && rig.embedded().is_empty());
|
||||
let mut history = FakeHistory::default();
|
||||
history.room("!never:s", vec![said("$a", 5, "ok"), said("$b", 15, "PURGED"), said("$c", 25, "ok")]);
|
||||
rig.ok(&mut history);
|
||||
assert_eq!(rig.held_ids("!never:s"), ["$a", "$c"]);
|
||||
assert!(rig.embedded().iter().all(|text| !text.contains("PURGED")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_repeated_or_overlapping_request_changes_nothing_or_merges_and_the_33rd_window_is_refused() {
|
||||
let mut rig = Rig::new();
|
||||
assert!(request_purge(&mut rig.index, "!r:s", window(10, 20)).unwrap());
|
||||
let writes = rig.index.state.lock().unwrap().cursor_writes.len();
|
||||
assert!(!request_purge(&mut rig.index, "!r:s", window(10, 20)).unwrap());
|
||||
assert!(!request_purge(&mut rig.index, "!r:s", window(12, 18)).unwrap());
|
||||
assert_eq!(rig.index.state.lock().unwrap().cursor_writes.len(), writes, "no write for a request the tombstone already holds");
|
||||
assert!(request_purge(&mut rig.index, "!r:s", window(15, 30)).unwrap());
|
||||
assert_eq!(tombstone(&rig, "!r:s").as_deref(), Some("10..30"));
|
||||
for n in 0..(PURGE_WINDOWS_MAX as u64 - 1) {
|
||||
request_purge(&mut rig.index, "!r:s", window(100 + n * 10, 105 + n * 10)).unwrap();
|
||||
}
|
||||
let error = format!("{:#}", request_purge(&mut rig.index, "!r:s", window(10_000, 10_001)).unwrap_err());
|
||||
assert!(error.contains("32"), "{error}");
|
||||
assert!(request_purge(&mut Rig::new().index, "!r:s", window(5, 5)).is_err());
|
||||
}
|
||||
|
||||
// ---- the source stays in its lane ----
|
||||
|
||||
#[test]
|
||||
fn the_ingest_code_is_pure_over_its_ports_and_names_no_credential_shape_no_second_literal_and_no_join_rule() {
|
||||
let sources = [("chat.rs", include_str!("chat.rs")), ("chat_chunk.rs", include_str!("chat_chunk.rs")), ("chat_purge.rs", include_str!("chat_purge.rs")), ("room_scope.rs", include_str!("room_scope.rs"))];
|
||||
let process = ["Command", "::new"].concat();
|
||||
for (name, source) in sources {
|
||||
let code = source.split("#[cfg(test)]").next().unwrap();
|
||||
assert!(!code.contains(&process) && !code.contains("std::process"), "{name}");
|
||||
assert!(!code.to_ascii_lowercase().contains("join_rule"), "{name}");
|
||||
if name != "room_scope.rs" {
|
||||
assert!(!code.to_ascii_lowercase().contains("discord"), "{name}");
|
||||
}
|
||||
for shape in ["sk-", "sk_live_", "ghp_", "github_pat_", "glpat-", "AKIA", "xoxb-", "PRIVATE KEY", "Bearer"] {
|
||||
assert!(!code.contains(shape), "{name} names {shape}");
|
||||
}
|
||||
let defined = usize::from(name == "chat.rs");
|
||||
assert_eq!((code.matches("\"chat:").count(), code.matches("\"chat-purge:").count()), (defined, defined), "{name}: each prefix is defined once");
|
||||
}
|
||||
let chat = include_str!("chat.rs");
|
||||
assert_eq!(chat.matches("scrub(").count(), 1, "scrub is called in one place");
|
||||
}
|
||||
|
|
@ -3,29 +3,37 @@
|
|||
//! for semantic search. Each point's `corpus` payload (`forge` or `chat`) keeps
|
||||
//! the two apart; a forge search filters `forge` and a chat search `chat`.
|
||||
//!
|
||||
//! Two truths, one writer. The forge is the truth of which repos exist and what
|
||||
//! they hold; the collection is derived from it and can be deleted whenever.
|
||||
//! `refresh` is the only writer of the collection, run by the `huginn-sources`
|
||||
//! oneshot job: first build, incremental refresh, rebuild after a model or
|
||||
//! chunker change, and retirement are all the one routine (delete by filter,
|
||||
//! ingest at a commit, rewrite the cursors). The daemon only reads: `search`
|
||||
//! is its whole part, and nothing in `serve` or the worker calls a write.
|
||||
//! Two truths, one job. The forge is the truth of which repos exist and what
|
||||
//! they hold, Synapse of what the Minds' rooms said; the collection is derived
|
||||
//! from them and can be deleted whenever (but for a steward's purge tombstones, see
|
||||
//! `chat_purge`). `refresh` is the forge's writer and the only creator of the
|
||||
//! collection, run by the `huginn-sources` oneshot job: first build, incremental
|
||||
//! refresh, rebuild after a model or chunker change, and retirement are all the one
|
||||
//! routine (delete by filter, ingest at a commit, rewrite the cursors). `chat::refresh_chat`
|
||||
//! joins it in the same job and the same cursors object, as the only writer of chat points;
|
||||
//! it requires the collection to exist and never recreates it. The daemon only reads: `search`
|
||||
//! and `search_chat` are its whole part, and nothing in `serve` or the worker calls a write.
|
||||
//!
|
||||
//! A cursor is the commit whose chunks the collection holds for a repo. It
|
||||
//! lives in the collection's own metadata, so it is lost with the collection
|
||||
//! A cursor is the commit (or, for a room, the event) whose chunks the collection
|
||||
//! holds. It lives in the collection's own metadata, so it is lost with the collection
|
||||
//! and the two cannot disagree about what is held. It is advanced only after
|
||||
//! the repo's deletes and upserts are acknowledged: it may lag and never leads.
|
||||
//! It is dropped before the repo's first delete, so a run that fails between
|
||||
//! the delete and the upserts leaves no cursor, and the repo is ingested whole.
|
||||
//! The cursors object is written whole by this job alone, once per repo.
|
||||
//! the source's deletes and upserts are acknowledged: it may lag and never leads.
|
||||
//! A repo's is dropped before its first delete, so a run that fails between
|
||||
//! the delete and the upserts leaves no cursor, and the repo is ingested whole; a room's
|
||||
//! chat chunks are written before anything they replace is deleted. The cursors
|
||||
//! object is written whole by this job alone. A key is a repo name, or a chat key
|
||||
//! (`chat::is_chat_key`), never both.
|
||||
|
||||
pub mod chat;
|
||||
pub mod chat_chunk;
|
||||
pub mod chat_purge;
|
||||
pub mod chunk;
|
||||
pub mod forge;
|
||||
pub mod git;
|
||||
pub mod room_scope;
|
||||
|
||||
#[cfg(test)]
|
||||
mod chat_tests;
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue