From fe97420cb4e2d3fa073b1882a3aca53994d07e84 Mon Sep 17 00:00:00 2001 From: Trezy Date: Mon, 3 Aug 2026 11:52:36 -0500 Subject: [PATCH] fix: prevent jetstream deletes from being skipped when a lexicon has no custom script fixes #80 Signed-off-by: Trezy --- .../content/docs/guides/record-scripts.md | 12 +- .../reference/script-examples/algolia-sync.md | 2 +- .../script-examples/meilisearch-sync.md | 2 +- src/admin/backfill.rs | 5 +- src/lib.rs | 2 + src/lua/mod.rs | 4 +- src/lua/scripts.rs | 53 +++-- src/record_handler.rs | 222 +++++++++++++++++- src/test_support.rs | 134 +++++++++++ tests/lua_linked_repos_api.rs | 8 +- 10 files changed, 406 insertions(+), 38 deletions(-) create mode 100644 src/test_support.rs diff --git a/packages/docs/content/docs/guides/record-scripts.md b/packages/docs/content/docs/guides/record-scripts.md index 23639a8..0cf3fad 100644 --- a/packages/docs/content/docs/guides/record-scripts.md +++ b/packages/docs/content/docs/guides/record-scripts.md @@ -81,7 +81,7 @@ The function is called once per event. | `true` | The original record is stored as-is | | *(no script)* | The original record is stored as-is | -On **delete** events, returning `nil` skips the delete (the record stays in the database). +On **delete** events, returning `nil` skips the delete (the record stays in the database). Because a delete event carries no `record`, `return record` returns `nil` there -- use `return true` to let a delete proceed. **Important:** If your script has side effects (e.g. syncing to a search index) but you want normal indexing to proceed, return `record` or `true` -- not nothing. A missing return statement returns `nil`, which **skips indexing**. @@ -183,7 +183,7 @@ Create a script with trigger `record.index:your.collection.nsid` to skip indexin ```lua function handle() if action == "delete" then - return record -- allow deletes to proceed + return true -- allow deletes to proceed end if record.title == nil or record.title == "" then @@ -201,7 +201,7 @@ Enrich a record with a computed field before it is stored: ```lua function handle() if action == "delete" then - return record + return true end record.slug = string.lower(string.gsub(record.title or "", "%s+", "-")) @@ -222,7 +222,7 @@ function handle() record = record }) }) - return record + return record or true -- `record` is nil on delete; `true` lets it proceed end ``` @@ -254,7 +254,7 @@ function handle() }) end - return record + return record or true -- `record` is nil on delete; `true` lets it proceed end ``` @@ -289,7 +289,7 @@ function handle() }) end - return record + return record or true -- `record` is nil on delete; `true` lets it proceed end ``` diff --git a/packages/docs/content/docs/reference/script-examples/algolia-sync.md b/packages/docs/content/docs/reference/script-examples/algolia-sync.md index d4594c6..22ee62f 100644 --- a/packages/docs/content/docs/reference/script-examples/algolia-sync.md +++ b/packages/docs/content/docs/reference/script-examples/algolia-sync.md @@ -30,7 +30,7 @@ function handle() }) end - return record + return record or true -- `record` is nil on delete; `true` lets it proceed end ``` diff --git a/packages/docs/content/docs/reference/script-examples/meilisearch-sync.md b/packages/docs/content/docs/reference/script-examples/meilisearch-sync.md index 2eb34a9..5ab93a6 100644 --- a/packages/docs/content/docs/reference/script-examples/meilisearch-sync.md +++ b/packages/docs/content/docs/reference/script-examples/meilisearch-sync.md @@ -31,7 +31,7 @@ function handle() }) end - return record + return record or true -- `record` is nil on delete; `true` lets it proceed end ``` diff --git a/src/admin/backfill.rs b/src/admin/backfill.rs index a1dfc9e..880b26c 100644 --- a/src/admin/backfill.rs +++ b/src/admin/backfill.rs @@ -1378,8 +1378,9 @@ async fn fetch_records_from_pds( ) .await { - None => continue, - Some(v) => v, + crate::lua::RecordHookOutcome::Skip => continue, + crate::lua::RecordHookOutcome::Replace(v) => v, + crate::lua::RecordHookOutcome::Proceed => entry.value.clone(), }; batch.push(PreparedRecord { diff --git a/src/lib.rs b/src/lib.rs index f3b410b..4ecd878 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -38,6 +38,8 @@ pub mod service_entries; pub mod service_identity; pub mod setup; pub mod spaces; +#[cfg(test)] +pub mod test_support; pub mod verification_methods; pub mod xrpc; diff --git a/src/lua/mod.rs b/src/lua/mod.rs index fab4340..d81946f 100644 --- a/src/lua/mod.rs +++ b/src/lua/mod.rs @@ -17,8 +17,8 @@ pub(crate) use context::SpaceContext; pub(crate) use execute::{execute_procedure_script, execute_query_script}; pub(crate) use sandbox::validate_script; pub use scripts::{ - LabelAppliedEvent, LabelHookOutcome, ParsedTrigger, RecordEventPayload, ResolvedScript, - ScriptLanguage, ScriptRow, TriggerKind, resolve, resolve_record_event, + LabelAppliedEvent, LabelHookOutcome, ParsedTrigger, RecordEventPayload, RecordHookOutcome, + ResolvedScript, ScriptLanguage, ScriptRow, TriggerKind, resolve, resolve_record_event, run_label_applied_script, run_record_event_once, run_record_event_script, trigger_for_label_uri, }; diff --git a/src/lua/scripts.rs b/src/lua/scripts.rs index 0edebeb..aa8d0b4 100644 --- a/src/lua/scripts.rs +++ b/src/lua/scripts.rs @@ -322,20 +322,37 @@ pub struct RecordEventPayload<'a> { pub record: Option<&'a Value>, } -/// Run the record-event script (if any) for a given event. Returns the -/// record body the indexer should store: `Some(record)` to proceed, -/// `None` to skip indexing. +/// What a record-event script chain decided for an event. +/// +/// Deliberately not `Option`. A delete carries no record body, so +/// "no script ran" and "the script returned `nil`" both used to spell +/// themselves `None`, and the delete path read the first as the second — +/// an instance with no scripts at all silently skipped every delete (#80). +#[derive(Debug, Clone, PartialEq)] +pub enum RecordHookOutcome { + /// Index the event with the body it arrived with. Either no script ran, + /// or the script waved the event through without rewriting it. + Proceed, + /// Index this body in place of the one that arrived. Only meaningful for + /// create/update — a delete has no body to replace. + Replace(Value), + /// Skip the event entirely — the script returned `nil`. Only ever + /// produced by a script that actually ran. + Skip, +} + +/// Run the record-event script (if any) for a given event. /// /// Failure mode is fail-open: a script that exhausts its retry budget is -/// dead-lettered and the indexer proceeds with the original record. +/// dead-lettered and the event proceeds as if no script had run. pub async fn run_record_event_script( state: &AppState, payload: RecordEventPayload<'_>, -) -> Option { +) -> RecordHookOutcome { let resolved = match resolve_record_event(state, payload.nsid, payload.action).await { Some(s) => s, - // No script for this trigger → indexer keeps the original record. - None => return payload.record.cloned(), + // No script for this trigger → index the event unchanged. + None => return RecordHookOutcome::Proceed, }; let host_id = format!("{}:{}", payload.nsid, payload.action); @@ -419,21 +436,23 @@ pub async fn run_record_event_script( ) .await; - // Fail-open: indexer proceeds with the original record. - payload.record.cloned() + // Fail-open: index the event as if no script had run. For a delete this + // means the delete still happens — the record body it lacks is not a + // reason to keep a record its PDS no longer has. + RecordHookOutcome::Proceed } /// Single attempt at the record-event Lua script. Used internally by the /// retry loop and externally by admin retry endpoints. /// -/// Returns `Ok(Some(value))` to continue indexing with `value`, -/// `Ok(None)` when the script returned `nil` (skip), or `Err(msg)` on -/// any execution failure. +/// Returns `Ok(Replace(value))` to continue indexing with `value`, +/// `Ok(Skip)` when the script returned `nil`, `Ok(Proceed)` when it waved +/// the event through, or `Err(msg)` on any execution failure. pub async fn run_record_event_once( state: &AppState, script: &ResolvedScript, payload: RecordEventPayload<'_>, -) -> Result, String> { +) -> Result { if script.language != ScriptLanguage::Lua { return Err(format!( "this binary cannot run {} scripts", @@ -493,15 +512,15 @@ pub async fn run_record_event_once( .map_err(|e| e.to_string())?; match result { - mlua::Value::Nil => Ok(None), + mlua::Value::Nil => Ok(RecordHookOutcome::Skip), mlua::Value::Table(_) => { let v: Value = lua .from_value(result) .map_err(|e| format!("convert lua return to JSON: {e}"))?; - Ok(Some(v)) + Ok(RecordHookOutcome::Replace(v)) } - // Non-nil, non-table return — pass-through: keep the original record. - _ => Ok(payload.record.cloned()), + // Non-nil, non-table return (`return true`) — pass-through. + _ => Ok(RecordHookOutcome::Proceed), } } diff --git a/src/record_handler.rs b/src/record_handler.rs index ea69a23..511c15e 100644 --- a/src/record_handler.rs +++ b/src/record_handler.rs @@ -7,6 +7,7 @@ use crate::AppState; use crate::db::{adapt_sql, now_rfc3339}; use crate::event_log::{EventLog, Severity, log_event}; use crate::lexicon::{LexiconType, ParsedLexicon, ProcedureAction}; +use crate::lua::RecordHookOutcome; /// The static collection we always include for lexicon schema updates. pub const LEXICON_SCHEMA_COLLECTION: &str = "com.atproto.lexicon.schema"; @@ -87,8 +88,9 @@ pub async fn handle_record_event(state: &AppState, record: &RecordEvent) { // Run record-event script (if any) before storing. The script's // return value determines what gets written: - // None → skip indexing entirely - // Some(record) → upsert with that record body + // Skip → skip indexing entirely + // Replace(record) → upsert with that record body + // Proceed → upsert with the record as it arrived // The dispatcher cascades `record.:` → // `record.index:`; failures are dead-lettered fail-open. let hook_result = crate::lua::run_record_event_script( @@ -104,7 +106,7 @@ pub async fn handle_record_event(state: &AppState, record: &RecordEvent) { ) .await; let rec_to_store = match hook_result { - None => { + RecordHookOutcome::Skip => { log_event( db, EventLog { @@ -124,7 +126,8 @@ pub async fn handle_record_event(state: &AppState, record: &RecordEvent) { .await; return; } - Some(v) => v, + RecordHookOutcome::Replace(v) => v, + RecordHookOutcome::Proceed => rec.clone(), }; let now = now_rfc3339(); @@ -209,8 +212,9 @@ pub async fn handle_record_event(state: &AppState, record: &RecordEvent) { "delete" => { let backend = state.db_backend; - // Run record-event script (if any) before deleting. A nil - // return aborts the delete; any other return continues. + // Run record-event script (if any) before deleting. Only a + // script that actually ran and returned `nil` aborts the + // delete — no script, or a dead-lettered one, proceeds. let hook_result = crate::lua::run_record_event_script( state, crate::lua::RecordEventPayload { @@ -223,7 +227,7 @@ pub async fn handle_record_event(state: &AppState, record: &RecordEvent) { }, ) .await; - if hook_result.is_none() { + if hook_result == RecordHookOutcome::Skip { log_event( db, EventLog { @@ -395,3 +399,207 @@ pub async fn handle_lexicon_schema_event(state: &AppState, did: &str, record: &R _ => {} } } + +#[cfg(test)] +mod tests { + use super::*; + + use crate::lexicon::ProcedureAction; + use crate::test_support::{memory_pool, test_state_with_pool}; + + const NSID: &str = "com.example.thing"; + const URI: &str = "at://did:plc:abc/com.example.thing/rkey1"; + + /// A state with the record/script/event-log tables and `NSID` registered as + /// a record-type lexicon, so `handle_record_event` treats it as tracked. + async fn tracked_state() -> AppState { + let pool = memory_pool().await; + for ddl in [ + "CREATE TABLE happyview_records ( + uri TEXT PRIMARY KEY, + did TEXT NOT NULL, + collection TEXT NOT NULL, + rkey TEXT NOT NULL, + record TEXT NOT NULL, + cid TEXT, + indexed_at TEXT NOT NULL, + created_at TEXT NOT NULL + )", + "CREATE TABLE happyview_scripts ( + id TEXT PRIMARY KEY, + body TEXT NOT NULL, + script_type TEXT NOT NULL DEFAULT 'lua' + )", + "CREATE TABLE happyview_event_logs ( + id TEXT PRIMARY KEY, + event_type TEXT NOT NULL, + severity TEXT NOT NULL, + actor_did TEXT, + subject TEXT, + detail TEXT, + created_at TEXT NOT NULL + )", + "CREATE TABLE happyview_record_refs ( + source_uri TEXT NOT NULL, + target_uri TEXT NOT NULL, + field TEXT NOT NULL + )", + ] { + crate::db::query(ddl) + .execute(&pool) + .await + .unwrap_or_else(|e| panic!("create table: {e}")); + } + + let state = test_state_with_pool(pool); + let parsed = ParsedLexicon::parse( + serde_json::json!({ + "lexicon": 1, + "id": NSID, + "defs": {"main": {"type": "record", "key": "tid"}}, + }), + 1, + Some(NSID.to_string()), + ProcedureAction::Upsert, + None, + ) + .expect("parse test lexicon"); + state.lexicons.upsert(parsed).await; + state + } + + async fn insert_record(state: &AppState) { + crate::db::query( + "INSERT INTO happyview_records (uri, did, collection, rkey, record, cid, indexed_at, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + ) + .bind(URI) + .bind("did:plc:abc") + .bind(NSID) + .bind("rkey1") + .bind(r#"{"text":"hello"}"#) + .bind("bafyreiabc") + .bind("2026-01-01T00:00:00+00:00") + .bind("2026-01-01T00:00:00+00:00") + .execute(&state.db) + .await + .expect("seed record"); + } + + async fn record_exists(state: &AppState) -> bool { + let row: Option<(String,)> = + crate::db::query_as("SELECT uri FROM happyview_records WHERE uri = ?") + .bind(URI) + .fetch_optional(&state.db) + .await + .expect("query record"); + row.is_some() + } + + fn delete_event() -> RecordEvent { + RecordEvent { + did: "did:plc:abc".to_string(), + collection: NSID.to_string(), + rkey: "rkey1".to_string(), + action: "delete".to_string(), + record: None, + cid: None, + } + } + + async fn install_script(state: &AppState, trigger: &str, body: &str) { + crate::db::query( + "INSERT INTO happyview_scripts (id, body, script_type) VALUES (?, ?, 'lua')", + ) + .bind(trigger) + .bind(body) + .execute(&state.db) + .await + .expect("install script"); + } + + /// Regression test for #80: a Jetstream delete for a tracked collection with + /// no registered script must delete the row. The "no script ran" and "the + /// script returned nil" signals used to be spelled the same way, so an + /// instance with no scripts at all skipped every delete. + #[tokio::test] + async fn delete_without_any_script_removes_the_record() { + let state = tracked_state().await; + insert_record(&state).await; + + handle_record_event(&state, &delete_event()).await; + + assert!( + !record_exists(&state).await, + "delete with no registered script must remove the record" + ); + } + + /// `return true` is the documented "proceed, I only had side effects" + /// return. On a delete it used to fall through to the original record + /// body — which is nil for a delete — and abort. + #[tokio::test] + async fn delete_with_a_script_returning_true_removes_the_record() { + let state = tracked_state().await; + install_script( + &state, + &format!("record.delete:{NSID}"), + "function handle() return true end", + ) + .await; + insert_record(&state).await; + + handle_record_event(&state, &delete_event()).await; + + assert!( + !record_exists(&state).await, + "a delete script returning true must let the delete proceed" + ); + } + + /// The documented delete gate: a script that runs and returns `nil` + /// still keeps the record. This is the one case that must NOT delete. + #[tokio::test] + async fn delete_with_a_script_returning_nil_keeps_the_record() { + let state = tracked_state().await; + install_script( + &state, + &format!("record.delete:{NSID}"), + "function handle() return nil end", + ) + .await; + insert_record(&state).await; + + handle_record_event(&state, &delete_event()).await; + + assert!( + record_exists(&state).await, + "a delete script returning nil must keep the record" + ); + } + + /// The create path's pass-through: no script means index the record as it + /// arrived, not skip it. + #[tokio::test] + async fn create_without_any_script_indexes_the_record() { + let state = tracked_state().await; + + handle_record_event( + &state, + &RecordEvent { + did: "did:plc:abc".to_string(), + collection: NSID.to_string(), + rkey: "rkey1".to_string(), + action: "create".to_string(), + record: Some(serde_json::json!({"text": "hello"})), + cid: None, + }, + ) + .await; + + assert!( + record_exists(&state).await, + "create with no registered script must index the record" + ); + } +} diff --git a/src/test_support.rs b/src/test_support.rs new file mode 100644 index 0000000..a489d0f --- /dev/null +++ b/src/test_support.rs @@ -0,0 +1,134 @@ +//! Shared fixtures for `cargo test --lib`. +//! +//! Building an `AppState` requires a fully-constructed OAuth client, which is +//! ~100 lines of boilerplate that several test modules had each copied. This +//! module exists so new tests can take a pool and get a usable state back. + +use std::sync::Arc; + +use tokio::sync::watch; + +use crate::AppState; +use crate::config::Config; +use crate::lexicon::LexiconRegistry; + +/// An in-memory SQLite pool. `max_connections(1)` is required: every +/// connection to `sqlite::memory:` gets its own private database, so a larger +/// pool would hand later queries an empty schema. +pub async fn memory_pool() -> sqlx::AnyPool { + sqlx::any::install_default_drivers(); + sqlx::pool::PoolOptions::::new() + .max_connections(1) + .connect("sqlite::memory:") + .await + .expect("connect to in-memory sqlite") +} + +/// Build an `AppState` backed by `pool`, wired for SQLite with no network +/// dependencies reachable (PLC and OAuth point at unroutable local ports). +pub fn test_state_with_pool(pool: sqlx::AnyPool) -> AppState { + let config = Config { + host: "127.0.0.1".into(), + port: 3000, + database_url: String::new(), + database_backend: crate::db::DatabaseBackend::Sqlite, + sqlite_journal_size_limit: crate::db::DEFAULT_JOURNAL_SIZE_LIMIT, + public_url: String::new(), + session_secret: "test-secret".into(), + jetstream_url: String::new(), + relay_url: String::new(), + plc_url: String::new(), + static_dir: String::new(), + base_path: None, + event_log_retention_days: 30, + app_name: None, + logo_uri: None, + tos_uri: None, + policy_uri: None, + token_encryption_key: None, + default_rate_limit_capacity: 100, + default_rate_limit_refill_rate: 2.0, + }; + let (tx, _) = watch::channel(vec![]); + let (labeler_tx, _) = watch::channel(()); + let backend = crate::db::DatabaseBackend::Sqlite; + let atrium_http = Arc::new(atrium_oauth::DefaultHttpClient::default()); + let did_resolver = atrium_identity::did::CommonDidResolver::new( + atrium_identity::did::CommonDidResolverConfig { + plc_directory_url: "https://plc.directory".into(), + http_client: Arc::clone(&atrium_http), + }, + ); + let handle_resolver = atrium_identity::handle::AtprotoHandleResolver::new( + atrium_identity::handle::AtprotoHandleResolverConfig { + dns_txt_resolver: crate::dns::NativeDnsResolver::new(), + http_client: atrium_http, + }, + ); + let scopes = vec![atrium_oauth::Scope::Known( + atrium_oauth::KnownScope::Atproto, + )]; + let oauth = atrium_oauth::OAuthClient::new(atrium_oauth::OAuthClientConfig { + client_metadata: atrium_oauth::AtprotoLocalhostClientMetadata { + redirect_uris: Some(vec!["http://127.0.0.1:0/auth/callback".into()]), + scopes: Some(scopes.clone()), + }, + keys: None, + state_store: crate::auth::oauth_store::DbStateStore::new(pool.clone(), backend), + session_store: crate::auth::oauth_store::DbSessionStore::new(pool.clone(), backend), + resolver: atrium_oauth::OAuthResolverConfig { + did_resolver, + handle_resolver, + authorization_server_metadata: Default::default(), + protected_resource_metadata: Default::default(), + }, + }) + .expect("test OAuth client"); + AppState { + config, + http: reqwest::Client::new(), + db: pool.clone(), + backfill_db: pool.clone(), + db_backend: backend, + domain_cache: crate::domain::DomainCache::new(), + lexicons: LexiconRegistry::new(), + collections_tx: tx, + labeler_subscriptions_tx: labeler_tx, + rate_limiter: crate::rate_limit::RateLimiter::new(crate::rate_limit::RateLimitDefaults { + query_cost: 1, + procedure_cost: 1, + proxy_cost: 1, + }), + oauth: Arc::new(crate::auth::OAuthClientRegistry::new(Arc::new(oauth))), + oauth_state_store: crate::auth::oauth_store::DbStateStore::new(pool.clone(), backend), + linked_repos_client: Arc::new( + crate::linked_repos::client::build( + "https://plc.directory", + "http://127.0.0.1:0/oauth-client-metadata.json", + "http://127.0.0.1:0", + "http://127.0.0.1:0/auth/callback".into(), + true, + scopes, + crate::auth::oauth_store::DbStateStore::new(pool.clone(), backend), + pool.clone(), + backend, + ) + .expect("test linked-repo OAuth client"), + ), + cookie_key: axum_extra::extract::cookie::Key::derive_from( + b"test-secret-that-is-at-least-32-bytes-long", + ), + plugin_registry: Arc::new(crate::plugin::PluginRegistry::new()), + wasm_runtime: Arc::new(crate::plugin::WasmRuntime::new().expect("wasm runtime")), + attestation_signer: None, + official_registry: Arc::new(tokio::sync::RwLock::new( + crate::plugin::official_registry::OfficialRegistryState::default(), + )), + official_registry_config: crate::plugin::official_registry::RegistryConfig::production(), + proxy_config: Arc::new(arc_swap::ArcSwap::new(Arc::new( + crate::proxy_config::ProxyConfig::default(), + ))), + backfill_events_tx: tokio::sync::broadcast::channel(16).0, + verbose_event_logging: Arc::new(std::sync::atomic::AtomicBool::new(false)), + } +} diff --git a/tests/lua_linked_repos_api.rs b/tests/lua_linked_repos_api.rs index 641c8aa..91257cc 100644 --- a/tests/lua_linked_repos_api.rs +++ b/tests/lua_linked_repos_api.rs @@ -594,8 +594,12 @@ async fn global_is_registered_for_record_event_scripts() { }, ) .await - .unwrap() - .expect("script returned a table"); + .unwrap(); + + let out = match out { + happyview::lua::RecordHookOutcome::Replace(v) => v, + other => panic!("expected the script's table back, got {other:?}"), + }; assert_eq!(out["kind"], "table"); assert_eq!(out["get"], "function"); -- 2.51.2