phughk/ottyl
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Wrap the left-right index in IndexWriter/IndexReader types

Browse Source 
The engine actor and API handlers previously called left_right's
WriteHandle/ReadHandleFactory methods (append/publish/enter/handle)
directly. IndexWriter now exposes mutate()/peek() and IndexReader
exposes handle() (returning an IndexReadHandle with its own enter()),
so how the single-writer/multi-reader index is accessed is named and
scoped to these two types instead of being raw left_right calls
scattered through engine.rs/api.rs/profile.rs.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01A1ZJ3SrayzLzVndhVq9BmS
This commit is contained in:
Hugh Kaznowski
2026-09-06 17:10:23 +01:00
co-authored by Claude Sonnet 5
parent 3c2995acc1
commit 311713986a
3 changed files with 75 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api.rs
+1 -1
View File
@@ -217,7 +217,7 @@ fn mime_for_extension(extension: &str) -> &'static str {
impl EngineHandle {
async fn ask(&self, query: EngineQuery) -> Result<EngineResponse, StatusCode> {
let read_handle = self.reader_factory.handle();
let read_handle = self.reader.handle();
if let Some(sort_by) = query.sort_by() {
let needs_build = match read_handle.enter() {
Some(guard) => !guard.has_sort_helper_index(sort_by),
src/engine.rs
+73 -48
View File
@@ -149,14 +149,53 @@ impl Absorb<EngineOp> for EngineSnapshot {
}
}
struct IndexWriter {
handle: left_right::WriteHandle<EngineSnapshot, EngineOp>,
}
impl IndexWriter {
fn mutate(&mut self, op: impl FnMut(&mut EngineSnapshot) + Send + 'static) {
self.handle.append(EngineOp(Box::new(op)));
self.handle.publish();
}
fn peek(&mut self) -> Option<left_right::ReadGuard<'_, EngineSnapshot>> {
self.handle.enter()
}
}
pub struct IndexReadHandle(left_right::ReadHandle<EngineSnapshot>);
impl IndexReadHandle {
pub(crate) fn enter(&self) -> Option<left_right::ReadGuard<'_, EngineSnapshot>> {
self.0.enter()
}
}
pub struct IndexReader {
factory: left_right::ReadHandleFactory<EngineSnapshot>,
}
impl Clone for IndexReader {
fn clone(&self) -> Self {
IndexReader { factory: self.factory.clone() }
}
}
impl IndexReader {
pub(crate) fn handle(&self) -> IndexReadHandle {
IndexReadHandle(self.factory.handle())
}
}
pub struct EngineHandle {
pub tx: mpsc::Sender<EngineMessage>,
pub reader_factory: left_right::ReadHandleFactory<EngineSnapshot>,
pub reader: IndexReader,
}
impl Clone for EngineHandle {
fn clone(&self) -> Self {
EngineHandle { tx: self.tx.clone(), reader_factory: self.reader_factory.clone() }
EngineHandle { tx: self.tx.clone(), reader: self.reader.clone() }
}
}
@@ -172,23 +211,19 @@ impl EngineHandle {
pub fn spawn(settings: SettingsHandle, channel_capacity: usize) -> EngineHandle {
let (tx, rx) = mpsc::channel(channel_capacity);
let (write, read) = left_right::new::<EngineSnapshot, EngineOp>();
let reader_factory = read.factory();
tokio::spawn(run(rx, settings, write));
EngineHandle { tx, reader_factory }
let reader = IndexReader { factory: read.factory() };
tokio::spawn(run(rx, settings, IndexWriter { handle: write }));
EngineHandle { tx, reader }
}
#[cfg(test)]
pub(crate) fn test_handle(channel_capacity: usize) -> (EngineHandle, mpsc::Receiver<EngineMessage>) {
let (tx, rx) = mpsc::channel(channel_capacity);
let (_write, read) = left_right::new::<EngineSnapshot, EngineOp>();
(EngineHandle { tx, reader_factory: read.factory() }, rx)
(EngineHandle { tx, reader: IndexReader { factory: read.factory() } }, rx)
}
async fn run(
mut rx: mpsc::Receiver<EngineMessage>,
settings: SettingsHandle,
mut write: left_right::WriteHandle<EngineSnapshot, EngineOp>,
) {
async fn run(mut rx: mpsc::Receiver<EngineMessage>, settings: SettingsHandle, mut write: IndexWriter) {
let mut max_logs: Option<usize> = None;
let mut metric_sample_sequence: u64 = 0;
let mut alert_states: HashMap<String, AlertRuntimeState> = HashMap::new();
@@ -218,7 +253,7 @@ async fn run(
};
let text_for_op = text.clone();
write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| {
write.mutate(move |snap: &mut EngineSnapshot| {
snap.approx_bytes_total += approx_log_bytes(&text_for_op);
snap.index.insert(
IndexKey::IngestedLog { sequence, timestamp },
@@ -234,11 +269,10 @@ async fn run(
snap.purge_stale_sort_helper_indexes();
snap.evict_excess_logs(max_logs);
snap.ingest_lag_high_watermark_ms = ingest_lag_high_watermark_ms;
})));
write.publish();
});
let fired = write
.enter()
.peek()
.map(|guard| {
alerts::evaluate_alerts(&rules, &mut alert_states, &guard, sequence, timestamp, &text)
})
@@ -257,109 +291,100 @@ async fn run(
let sequence = metric_sample_sequence;
let timestamp = sample.timestamp;
let value = MetricPrimitive::Number(Number::Float(sample.value));
write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| {
write.mutate(move |snap: &mut EngineSnapshot| {
snap.push_bucket_value(&metric_name, labels.clone(), sequence, timestamp, value.clone());
})));
write.publish();
});
metric_sample_sequence += 1;
}
EngineMessage::SetMaxLogs(new_max) => {
max_logs = new_max;
write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| {
write.mutate(move |snap: &mut EngineSnapshot| {
snap.evict_excess_logs(new_max);
})));
write.publish();
});
}
EngineMessage::DeclareParam { request, respond_to } => {
let result = write
.enter()
.peek()
.map(|guard| guard.validate_param_declaration(&request))
.unwrap_or_else(|| Err("engine is shutting down".to_string()));
let succeeded = result.is_ok();
let name = request.name.clone();
let request_for_op = request.clone();
write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| {
write.mutate(move |snap: &mut EngineSnapshot| {
snap.sort_helper_indexes.remove(&name);
if succeeded {
snap.insert_param_declaration(request_for_op.clone());
}
})));
write.publish();
});
let _ = respond_to.send(result);
}
EngineMessage::RemoveParamDeclaration { name, respond_to } => {
let existed = write
.enter()
.peek()
.is_some_and(|guard| guard.index.contains_key(&IndexKey::ParamDeclaration { name: name.clone() }));
let name_for_op = name.clone();
write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| {
write.mutate(move |snap: &mut EngineSnapshot| {
snap.index.remove(&IndexKey::ParamDeclaration { name: name_for_op.clone() });
snap.sort_helper_indexes.remove(&name_for_op);
})));
write.publish();
});
let _ = respond_to.send(existed);
}
EngineMessage::ClearParams { respond_to } => {
let mut scratch = EngineSnapshot::from_index(
write.enter().map(|guard| guard.index.clone()).unwrap_or_default(),
write.peek().map(|guard| guard.index.clone()).unwrap_or_default(),
);
let removed = scratch.clear_all_param_declarations();
let (reextracted, logs_considered) = scratch.reextract_params_from_retained_logs();
write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| {
write.mutate(move |snap: &mut EngineSnapshot| {
snap.clear_all_param_declarations();
snap.reextract_params_from_retained_logs();
snap.sort_helper_indexes.clear();
})));
write.publish();
});
let _ = respond_to.send(ClearParamsResult { removed, reextracted, logs_considered });
}
EngineMessage::DeclareMetric { request, respond_to } => {
let result = write
.enter()
.peek()
.map(|guard| guard.validate_metric_request(&request).map(|_| ()))
.unwrap_or_else(|| Err("engine is shutting down".to_string()));
let succeeded = result.is_ok();
let request_for_op = request.clone();
write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| {
write.mutate(move |snap: &mut EngineSnapshot| {
if succeeded {
let _ = snap.declare_metric(request_for_op.clone());
}
})));
write.publish();
});
let _ = respond_to.send(result);
}
EngineMessage::UpdateMetric { name, request, respond_to } => {
let result = write
.enter()
.peek()
.map(|guard| guard.validate_update_metric(&name, &request).map(|_| ()))
.unwrap_or_else(|| Err("engine is shutting down".to_string()));
let succeeded = result.is_ok();
let existing_name = name.clone();
let request_for_op = request.clone();
write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| {
write.mutate(move |snap: &mut EngineSnapshot| {
if succeeded {
let _ = snap.update_metric(&existing_name, request_for_op.clone());
}
})));
write.publish();
});
let _ = respond_to.send(result);
}
EngineMessage::RemoveMetricDeclaration { name, respond_to } => {
let existed = write
.enter()
.peek()
.is_some_and(|guard| guard.index.contains_key(&IndexKey::MetricDeclaration { name: name.clone() }));
let name_for_op = name.clone();
write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| {
write.mutate(move |snap: &mut EngineSnapshot| {
snap.remove_metric_declaration(&name_for_op);
})));
write.publish();
});
let _ = respond_to.send(existed);
}
EngineMessage::EnsureSortHelperIndex { param_name, respond_to } => {
write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| {
write.mutate(move |snap: &mut EngineSnapshot| {
snap.ensure_sort_helper_index_built(&param_name);
})));
write.publish();
});
let _ = respond_to.send(());
}
EngineMessage::Barrier(respond_to) => {
src/profile.rs
+1 -1
View File
@@ -708,7 +708,7 @@ mod tests {
}
async fn ask(engine: &EngineHandle, query: EngineQuery) -> EngineResponse {
let read_handle = engine.reader_factory.handle();
let read_handle = engine.reader.handle();
let guard = read_handle.enter().expect("engine unavailable");
guard.handle_query_ro(query)
}