You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by wy...@apache.org on 2026/10/16 14:36:51 UTC
(maka) branch rust-preview updated: perf(event-log): hash exact source bytes on SQLite workers
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch rust-preview
in repository https://gitbox.apache.org/repos/asf/maka.git
The following commit(s) were added to refs/heads/rust-preview by this push:
new 847d12d247 perf(event-log): hash exact source bytes on SQLite workers
847d12d247 is described below
commit 847d12d247c2cdccd57146851a29b5a511263dfc
Author: Wang <m4...@gmail.com>
AuthorDate: Fri Oct 16 22:36:17 2026 +0800
perf(event-log): hash exact source bytes on SQLite workers
Reduce source digest transport and use partial indexes for invocation boundaries, safety checks, and streamed step reconstruction. Preserve same-snapshot raw-byte authentication and typed validation errors.
Generated-by: Codex
---
.../migrations/0024_invocation_boundary_lookup.sql | 24 ++++
.../event-log/migrations/0025_safety_lookup.sql | 26 ++++
platform/event-log/src/context/evidence.rs | 101 ++------------
platform/event-log/src/context/evidence/scan.rs | 147 +++++++++++++++++++++
platform/event-log/src/context/evidence/tests.rs | 48 ++++---
platform/event-log/src/context/safety.rs | 2 +
platform/event-log/src/model_items.rs | 4 +-
platform/event-log/src/sqlite_functions.rs | 7 +-
8 files changed, 249 insertions(+), 110 deletions(-)
diff --git a/platform/event-log/migrations/0024_invocation_boundary_lookup.sql b/platform/event-log/migrations/0024_invocation_boundary_lookup.sql
new file mode 100644
index 0000000000..8e57d27342
--- /dev/null
+++ b/platform/event-log/migrations/0024_invocation_boundary_lookup.sql
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+-- SQLite can prove a single kind matches an OR term, but not IN membership.
+-- Keep exactly the same rows and uniqueness while enabling point lookups.
+DROP INDEX invocation_boundary;
+CREATE UNIQUE INDEX invocation_boundary ON event_log(invocation_id, kind)
+WHERE kind = 'invocation_opened' OR kind = 'invocation_ended';
diff --git a/platform/event-log/migrations/0025_safety_lookup.sql b/platform/event-log/migrations/0025_safety_lookup.sql
new file mode 100644
index 0000000000..63b1e0dabb
--- /dev/null
+++ b/platform/event-log/migrations/0025_safety_lookup.sql
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+-- Skip observation payloads when checking execution boundaries and effects.
+-- OR terms also let single-kind queries prove they match this partial index.
+CREATE INDEX safety_session_kind_sequence ON event_log(
+ json_extract(event_json, '$.invocation.session_id'), kind, sequence
+) WHERE kind = 'invocation_opened' OR kind = 'tool_dispatched'
+ OR kind = 'model_requested' OR kind = 'model_completed'
+ OR kind = 'model_interrupted';
diff --git a/platform/event-log/src/context/evidence.rs b/platform/event-log/src/context/evidence.rs
index dde982efd7..b3abdb3802 100644
--- a/platform/event-log/src/context/evidence.rs
+++ b/platform/event-log/src/context/evidence.rs
@@ -20,10 +20,11 @@
use super::selection::Selection;
use super::{SourceEvidence, invalid};
use crate::{StoreError, sequence_number};
-use futures_util::TryStreamExt;
-use sha2::{Digest, Sha256};
use sqlx::SqliteConnection;
+mod scan;
+pub(crate) use scan::register;
+
/// The same scoped digest as complete prefix(), without retaining source rows.
/// Even a single old giant observation is hashed in bounded SQL BLOB slices.
pub(super) async fn evidence(
@@ -58,96 +59,22 @@ pub(super) async fn digests(
selection: &Selection,
throughs: &[u64],
) -> Result<Vec<String>, StoreError> {
- let Some(through) = throughs.last().copied() else {
+ if throughs.is_empty() {
return Ok(Vec::new());
- };
+ }
let session = selection
.session
.as_deref()
.ok_or_else(|| invalid("context requires a Session"))?;
- let encoded = serde_json::to_vec(&selection.scope)?;
- let mut digest = Sha256::new();
- digest.update(b"maka.log-prefix.v1\0");
- digest.update((encoded.len() as u64).to_be_bytes());
- digest.update(encoded);
- let mut boundaries = throughs.iter().copied().peekable();
- let mut result = Vec::new();
- let mut finish = |digest: &Sha256, sequence: u64| -> Result<(), StoreError> {
- if let Some(&boundary) = boundaries.peek() {
- if sequence > boundary {
- return Err(invalid("coverage is not a selected source high-water"));
- }
- if sequence == boundary {
- let mut prefix = digest.clone();
- // The high-water suffix lets overlapping prefixes share the
- // stream while still binding each digest to its exact cut.
- prefix.update(boundary.to_be_bytes());
- result.push(format!("sha256:{:x}", prefix.finalize()));
- boundaries.next();
- }
- }
- Ok(())
- };
- finish(&digest, 0)?;
- let filter = Selection::predicate("e", "?4");
- let columns = "e.sequence, length(CAST(e.event_json AS BLOB)),
- CASE WHEN length(CAST(e.event_json AS BLOB)) <= 65536 THEN CAST(e.event_json AS BLOB) END";
- let mut after = 0i64;
- loop {
- let mut oversized = None;
- {
- let mut rows =
- sqlx::query_as::<_, (i64, i64, Option<Vec<u8>>)>(sqlx::AssertSqlSafe(format!(
- "SELECT {columns} FROM runtime_events e
- WHERE e.sequence > ?1 AND e.sequence <= ?2
- AND json_extract(e.event_json,'$.invocation.session_id') = ?3 AND {filter}
- UNION ALL SELECT {columns} FROM session_history_members h
- JOIN runtime_events e ON e.sequence = h.sequence
- WHERE h.session_id = ?3 AND h.sequence > ?1 AND h.sequence <= ?2 AND {filter}
- ORDER BY sequence"
- )))
- .bind(after)
- .bind(through as i64)
- .bind(session)
- .bind(&selection.lineage)
- .fetch(&mut *connection);
- while let Some((sequence, bytes, payload)) = rows.try_next().await? {
- if sequence <= after {
- return Err(invalid("source selection repeats or reorders a record"));
- }
- let bytes = sequence_number(bytes)?;
- digest.update(sequence_number(sequence)?.to_be_bytes());
- digest.update(bytes.to_be_bytes());
- after = sequence;
- if let Some(payload) = payload {
- digest.update(payload);
- finish(&digest, sequence_number(sequence)?)?;
- } else {
- oversized = Some((sequence, bytes));
- break;
- }
- }
- }
- let Some((sequence, bytes)) = oversized else {
- break;
- };
- let mut offset = 0u64;
- while offset < bytes {
- let slice: Vec<u8> = sqlx::query_scalar(
- "SELECT substr(CAST(event_json AS BLOB), ?, 65536) FROM runtime_events WHERE sequence = ?",
- ).bind((offset + 1) as i64).bind(sequence).fetch_one(&mut *connection).await?;
- if slice.is_empty() {
- return Err(invalid("source bytes disappeared"));
- }
- offset += slice.len() as u64;
- digest.update(slice);
- }
- finish(&digest, sequence_number(sequence)?)?;
- }
- if boundaries.peek().is_some() {
- return Err(invalid("coverage is not a selected source high-water"));
- }
- Ok(result)
+ let result: String = sqlx::query_scalar("SELECT context_source_digests(?, ?, ?, ?)")
+ .bind(session)
+ .bind(&selection.lineage)
+ .bind(serde_json::to_vec(&selection.scope)?)
+ .bind(serde_json::to_string(throughs)?)
+ .fetch_one(connection)
+ .await?;
+ serde_json::from_str::<Result<Vec<String>, String>>(&result)?
+ .map_err(StoreError::InvalidTransition)
}
pub(super) async fn high_water(
diff --git a/platform/event-log/src/context/evidence/scan.rs b/platform/event-log/src/context/evidence/scan.rs
new file mode 100644
index 0000000000..caaa7ef33d
--- /dev/null
+++ b/platform/event-log/src/context/evidence/scan.rs
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+//! Reduce exact source bytes on SQLite's worker, returning only digest cuts.
+//! No authenticated state survives this function or the calling transaction.
+
+use crate::context::{invalid, selection::Selection};
+use crate::{StoreError, sequence_number};
+use rusqlite::{Connection, functions::FunctionFlags, params, types::ValueRef};
+use sha2::{Digest, Sha256};
+
+pub(crate) fn register(connection: &Connection) -> Result<(), StoreError> {
+ connection.create_scalar_function(
+ "context_source_digests",
+ 4,
+ FunctionFlags::SQLITE_UTF8 | FunctionFlags::SQLITE_DIRECTONLY,
+ |context| {
+ let session: String = context.get(0)?;
+ let lineage: Option<String> = context.get(1)?;
+ let scope: Vec<u8> = context.get(2)?;
+ let cuts: String = context.get(3)?;
+ let cuts: Vec<u64> = serde_json::from_str(&cuts)
+ .map_err(|error| rusqlite::Error::UserFunctionError(Box::new(error)))?;
+ // SAFETY: synchronous read-only use on SQLite's current worker and
+ // transaction. No handle, row, or reference escapes the callback;
+ // it never awaits, changes threads, or recursively invokes this UDF.
+ let connection = unsafe { context.get_connection() }?;
+ let result = match digests(&connection, &session, lineage.as_deref(), &scope, &cuts) {
+ Ok(digests) => Ok(digests),
+ // Keep domain failures typed across the SQLx/UDF boundary.
+ Err(StoreError::InvalidTransition(message)) => Err(message),
+ Err(StoreError::Sqlite(error)) => return Err(error),
+ Err(error) => return Err(rusqlite::Error::UserFunctionError(Box::new(error))),
+ };
+ serde_json::to_string(&result)
+ .map_err(|error| rusqlite::Error::UserFunctionError(Box::new(error)))
+ },
+ )?;
+ Ok(())
+}
+
+fn digests(
+ connection: &Connection,
+ session: &str,
+ lineage: Option<&str>,
+ scope: &[u8],
+ throughs: &[u64],
+) -> Result<Vec<String>, StoreError> {
+ let Some(through) = throughs.last().copied() else {
+ return Ok(Vec::new());
+ };
+ let mut digest = Sha256::new();
+ digest.update(b"maka.log-prefix.v1\0");
+ digest.update((scope.len() as u64).to_be_bytes());
+ digest.update(scope);
+ let mut boundaries = throughs.iter().copied().peekable();
+ let mut result = Vec::new();
+ let mut finish = |digest: &Sha256, sequence: u64| -> Result<(), StoreError> {
+ if let Some(&boundary) = boundaries.peek() {
+ if sequence > boundary {
+ return Err(invalid("coverage is not a selected source high-water"));
+ }
+ if sequence == boundary {
+ let mut prefix = digest.clone();
+ prefix.update(boundary.to_be_bytes());
+ result.push(format!("sha256:{:x}", prefix.finalize()));
+ boundaries.next();
+ }
+ }
+ Ok(())
+ };
+ finish(&digest, 0)?;
+ let filter = Selection::predicate("e", "?4");
+ let payload = "length(CAST(e.event_json AS BLOB)),
+ CASE WHEN length(CAST(e.event_json AS BLOB)) <= 65536 THEN CAST(e.event_json AS BLOB) END";
+ let sql = format!(
+ "SELECT e.sequence AS sequence, {payload} FROM runtime_events e
+ WHERE e.sequence > ?1 AND e.sequence <= ?2
+ AND json_extract(e.event_json,'$.invocation.session_id') = ?3 AND {filter}
+ UNION ALL SELECT h.sequence AS sequence, {payload} FROM session_history_members h
+ JOIN runtime_events e ON e.sequence = h.sequence
+ WHERE h.session_id = ?3 AND h.sequence > ?1 AND h.sequence <= ?2 AND {filter}
+ ORDER BY sequence"
+ );
+ let mut statement = connection.prepare(&sql)?;
+ let mut slices = connection.prepare(
+ "SELECT substr(CAST(event_json AS BLOB), ?, 65536) FROM runtime_events WHERE sequence = ?",
+ )?;
+ let mut previous = 0i64;
+ let mut rows = statement.query(params![0i64, through as i64, session, lineage])?;
+ while let Some(row) = rows.next()? {
+ let sequence: i64 = row.get(0)?;
+ let bytes: i64 = row.get(1)?;
+ if sequence <= previous {
+ return Err(invalid("source selection repeats or reorders a record"));
+ }
+ let bytes = sequence_number(bytes)?;
+ digest.update(sequence_number(sequence)?.to_be_bytes());
+ digest.update(bytes.to_be_bytes());
+ previous = sequence;
+ match row.get_ref(2)? {
+ ValueRef::Blob(payload) => digest.update(payload),
+ ValueRef::Null => {
+ // Nested reads stay on this worker and snapshot. Keep the
+ // outer cursor so even oversized duplicate rows are checked.
+ let mut offset = 0u64;
+ while offset < bytes {
+ let mut parts = slices.query(params![(offset + 1) as i64, sequence])?;
+ let part = parts
+ .next()?
+ .ok_or_else(|| invalid("source bytes disappeared"))?;
+ let slice = part
+ .get_ref(0)?
+ .as_blob()
+ .map_err(|_| invalid("source bytes are not a BLOB"))?;
+ if slice.is_empty() {
+ return Err(invalid("source bytes disappeared"));
+ }
+ offset += slice.len() as u64;
+ digest.update(slice);
+ }
+ }
+ _ => return Err(invalid("source bytes are not a BLOB")),
+ }
+ finish(&digest, sequence_number(sequence)?)?;
+ }
+ if boundaries.peek().is_some() {
+ return Err(invalid("coverage is not a selected source high-water"));
+ }
+ Ok(result)
+}
diff --git a/platform/event-log/src/context/evidence/tests.rs b/platform/event-log/src/context/evidence/tests.rs
index a0288fea9a..a692d7764f 100644
--- a/platform/event-log/src/context/evidence/tests.rs
+++ b/platform/event-log/src/context/evidence/tests.rs
@@ -91,6 +91,13 @@ async fn streamed_boundaries_match_complete_prefixes_and_reject_unselected_cuts(
log.connection
.run(move |connection| {
Box::pin(async move {
+ // A UDF's temporary rusqlite view must not remove hooks owned
+ // by SQLx when it releases the borrowed connection.
+ let updates = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
+ let observed = updates.clone();
+ connection.lock_handle().await?.set_update_hook(move |_| {
+ observed.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+ });
let mut tx = connection.begin().await?;
let selection = Selection::session("session");
let throughs: Vec<_> = expected.iter().map(|prefix| prefix.high_water).collect();
@@ -108,9 +115,10 @@ async fn streamed_boundaries_match_complete_prefixes_and_reject_unselected_cuts(
[actual.last().unwrap().clone()]
);
assert!(
- digests(&mut tx, &selection, &[foreign, last])
- .await
- .is_err(),
+ matches!(
+ digests(&mut tx, &selection, &[foreign, last]).await,
+ Err(StoreError::InvalidTransition(_))
+ ),
"a valid last cut must not hide an unselected intermediate cut"
);
assert!(digests(&mut tx, &selection, &[last + 1]).await.is_err());
@@ -118,6 +126,7 @@ async fn streamed_boundaries_match_complete_prefixes_and_reject_unselected_cuts(
.bind(large_event)
.execute(&mut *tx)
.await?;
+ assert!(updates.load(std::sync::atomic::Ordering::Relaxed) > 0);
let changed = digests(&mut tx, &selection, &throughs).await?;
assert_eq!(
changed[0], actual[0],
@@ -179,7 +188,7 @@ async fn duplicate_copied_membership_at_a_cut_cannot_reuse_its_original_digest()
configuration: None,
input: InvocationInput::Message {
source_messages: Vec::new(),
- content: "native copied-session message".into(),
+ content: "native copied-session message".repeat(3000).into(),
request_fingerprint: None,
},
},
@@ -192,7 +201,7 @@ async fn duplicate_copied_membership_at_a_cut_cannot_reuse_its_original_digest()
.unwrap();
}
let prefix = log
- .scoped_prefix(Selection::session("copy").scope, 10, 8192)
+ .scoped_prefix(Selection::session("copy").scope, 10, 256 * 1024)
.await
.unwrap();
log.connection
@@ -206,19 +215,22 @@ async fn duplicate_copied_membership_at_a_cut_cannot_reuse_its_original_digest()
.digest,
prefix.digest
);
- sqlx::query(
- "INSERT INTO session_history_members(session_id,sequence) VALUES('copy',?)",
- )
- .bind(prefix.high_water as i64)
- .execute(&mut *tx)
- .await?;
- let error = selected(&mut tx, &selection, prefix.high_water)
- .await
- .unwrap_err();
- assert!(
- error.to_string().contains("source selection repeats"),
- "{error}"
- );
+ // Both the small terminal at the requested cut and the
+ // oversized opening before it must reject repeated membership.
+ for duplicate in [prefix.high_water, prefix.events[0].sequence] {
+ sqlx::query(
+ "INSERT INTO session_history_members(session_id,sequence) VALUES('copy',?)",
+ )
+ .bind(duplicate as i64)
+ .execute(&mut *tx)
+ .await?;
+ let error = selected(&mut tx, &selection, prefix.high_water)
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("source selection repeats"), "{error}");
+ sqlx::query("DELETE FROM session_history_members WHERE session_id='copy' AND sequence=?")
+ .bind(duplicate as i64).execute(&mut *tx).await?;
+ }
tx.rollback().await?;
Ok(())
})
diff --git a/platform/event-log/src/context/safety.rs b/platform/event-log/src/context/safety.rs
index a8b51bdc30..433d26d061 100644
--- a/platform/event-log/src/context/safety.rs
+++ b/platform/event-log/src/context/safety.rs
@@ -96,6 +96,8 @@ async fn require_safe_through_policy(
crate::model_items::accepted_calls!(), "
SELECT EXISTS(SELECT 1 FROM runtime_events e
WHERE json_extract(e.event_json, '$.invocation.session_id') = ?1
+ AND (e.kind = 'invocation_opened' OR e.kind = 'tool_dispatched'
+ OR e.kind = 'model_requested' OR e.kind = 'model_completed')
AND (?2 IS NULL OR e.invocation_id != ?2) AND (
(e.kind = 'invocation_opened' AND NOT EXISTS(SELECT 1 FROM runtime_events t
WHERE t.invocation_id = e.invocation_id AND t.kind = 'invocation_ended'))
diff --git a/platform/event-log/src/model_items.rs b/platform/event-log/src/model_items.rs
index 27f36aaed9..efadb42728 100644
--- a/platform/event-log/src/model_items.rs
+++ b/platform/event-log/src/model_items.rs
@@ -37,7 +37,7 @@ pub(crate) async fn read(
return Ok(None);
}
let mut builder = StepBuilder::for_step(step).map_err(invalid)?;
- let mut rows = sqlx::query_scalar::<_, String>("SELECT event_json FROM runtime_events WHERE invocation_id=? AND kind='model_observed' AND json_extract(event_json,'$.fact.step_id')=? ORDER BY sequence")
+ let mut rows = sqlx::query_scalar::<_, String>("SELECT event_json FROM runtime_events WHERE invocation_id=? AND kind='model_observed' AND CAST(json_extract(event_json,'$.fact.step_id') AS TEXT)=? ORDER BY sequence")
.bind(invocation).bind(step).fetch(connection);
while let Some(json) = rows.try_next().await? {
let event: RuntimeEvent = serde_json::from_str(&json)?;
@@ -91,7 +91,7 @@ pub(crate) async fn project(
let mut builder = StepBuilder::for_step(&step_id).map_err(invalid)?;
let mut items = Vec::new();
let mut scratch = 0usize;
- let mut rows = sqlx::query_as::<_, (i64, String)>("SELECT sequence,event_json FROM runtime_events WHERE invocation_id=?1 AND kind='model_observed' AND json_extract(event_json,'$.fact.step_id')=?2 AND sequence<?3 ORDER BY sequence")
+ let mut rows = sqlx::query_as::<_, (i64, String)>("SELECT sequence,event_json FROM runtime_events WHERE invocation_id=?1 AND kind='model_observed' AND CAST(json_extract(event_json,'$.fact.step_id') AS TEXT)=?2 AND sequence<?3 ORDER BY sequence")
.bind(&invocation.invocation_id).bind(&step_id).bind(sequence as i64).fetch(connection);
while let Some((position, json)) = rows.try_next().await? {
let event: RuntimeEvent = serde_json::from_str(&json)?;
diff --git a/platform/event-log/src/sqlite_functions.rs b/platform/event-log/src/sqlite_functions.rs
index 6ddd5406f9..c2beab4dd4 100644
--- a/platform/event-log/src/sqlite_functions.rs
+++ b/platform/event-log/src/sqlite_functions.rs
@@ -34,9 +34,10 @@ pub(crate) async fn register(connection: &mut SqliteConnection) -> Result<(), St
let borrowed = unsafe { rusqlite::Connection::from_handle(locked.as_raw_handle().as_ptr()) }?;
let registered = crate::sessions::register_functions(&borrowed)
.and_then(|()| crate::message_identity::register(&borrowed))
- .and_then(|()| crate::transcript::navigation::register(&borrowed));
- // With rusqlite's optional hooks features, wrapper Drop also removes hooks.
- // Register once at startup before any hooks exist, never during operations.
+ .and_then(|()| crate::transcript::navigation::register(&borrowed))
+ .and_then(|()| crate::context::evidence::register(&borrowed));
+ // rusqlite 0.39 removes hooks only for an owned connection; this borrowed
+ // wrapper leaves SQLx's connection and hooks intact.
drop(borrowed);
registered
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org