1use super::*;
8
9#[cfg(test)]
10static RESERVATION_STATEMENT_COUNT: std::sync::atomic::AtomicUsize =
11 std::sync::atomic::AtomicUsize::new(0);
12
13#[cfg(test)]
14fn record_reservation_statement() {
15 RESERVATION_STATEMENT_COUNT.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
16}
17
18pub struct SqliteTriggerStore {
19 conn: SqliteConnection,
20 clock: Arc<dyn lash_core::Clock>,
21}
22
23impl SqliteTriggerStore {
24 pub async fn open(path: &Path) -> tokio_rusqlite::Result<Self> {
25 Self::open_with_clock(path, Arc::new(lash_core::SystemClock)).await
26 }
27
28 pub async fn open_with_clock(
29 path: &Path,
30 clock: Arc<dyn lash_core::Clock>,
31 ) -> tokio_rusqlite::Result<Self> {
32 let conn = SqliteConnection::open(path).await?;
33 ensure_trigger_schema(&conn).await?;
34 apply_pragmas(&conn, StoreBacking::File).await?;
35 Ok(Self { conn, clock })
36 }
37
38 pub async fn memory() -> tokio_rusqlite::Result<Self> {
39 Self::memory_with_clock(Arc::new(lash_core::SystemClock)).await
40 }
41
42 pub async fn memory_with_clock(
43 clock: Arc<dyn lash_core::Clock>,
44 ) -> tokio_rusqlite::Result<Self> {
45 let conn = SqliteConnection::open_in_memory().await?;
46 ensure_trigger_schema(&conn).await?;
47 apply_pragmas(&conn, StoreBacking::Memory).await?;
48 Ok(Self { conn, clock })
49 }
50
51 fn encode_json<T: serde::Serialize>(value: &T) -> Result<String, lash_core::PluginError> {
52 serde_json::to_string(value).map_err(|err| {
53 lash_core::PluginError::Session(format!("failed to encode trigger row: {err}"))
54 })
55 }
56
57 fn decode_subscription(
58 json: String,
59 ) -> Result<lash_core::TriggerSubscriptionRecord, lash_core::PluginError> {
60 serde_json::from_str(&json).map_err(|err| {
61 lash_core::PluginError::Session(format!(
62 "failed to decode trigger subscription row: {err}"
63 ))
64 })
65 }
66
67 fn decode_occurrence(
68 json: String,
69 ) -> Result<lash_core::TriggerOccurrenceRecord, lash_core::PluginError> {
70 serde_json::from_str(&json).map_err(|err| {
71 lash_core::PluginError::Session(format!(
72 "failed to decode trigger occurrence row: {err}"
73 ))
74 })
75 }
76
77 fn decode_delivery(
78 occurrence_json: String,
79 subscription_json: String,
80 process_id: String,
81 created_at_ms: i64,
82 reservation_status: lash_core::TriggerDeliveryReservationStatus,
83 ) -> Result<lash_core::TriggerDeliveryReservation, lash_core::PluginError> {
84 Ok(lash_core::TriggerDeliveryReservation {
85 occurrence: Self::decode_occurrence(occurrence_json)?,
86 subscription: Self::decode_subscription(subscription_json)?,
87 process_id,
88 created_at_ms: created_at_ms as u64,
89 reservation_status,
90 })
91 }
92
93 async fn list_deliveries_where(
94 &self,
95 where_clause: &'static str,
96 value: String,
97 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
98 self.conn
99 .call(move |conn| {
100 Ok((|| {
101 let sql = format!(
102 "SELECT d.process_id, d.created_at_ms, o.record_json, s.record_json
103 FROM trigger_deliveries d
104 JOIN trigger_occurrences o ON o.occurrence_id = d.occurrence_id
105 JOIN trigger_subscriptions s ON s.subscription_id = d.subscription_id
106 WHERE {where_clause}
107 ORDER BY d.created_at_ms ASC, d.occurrence_id ASC, d.subscription_id ASC"
108 );
109 let mut stmt = conn.prepare(&sql).map_err(process_sqlite_error)?;
110 let rows = stmt
111 .query_map(params![value.as_str()], |row| {
112 Ok((
113 row.get::<_, String>(0)?,
114 row.get::<_, i64>(1)?,
115 row.get::<_, String>(2)?,
116 row.get::<_, String>(3)?,
117 ))
118 })
119 .map_err(process_sqlite_error)?;
120 let mut deliveries = Vec::new();
121 for row in rows {
122 let (process_id, created_at_ms, occurrence_json, subscription_json) =
123 row.map_err(process_sqlite_error)?;
124 deliveries.push(Self::decode_delivery(
125 occurrence_json,
126 subscription_json,
127 process_id,
128 created_at_ms,
129 lash_core::TriggerDeliveryReservationStatus::AlreadyReserved,
130 )?);
131 }
132 Ok(deliveries)
133 })())
134 })
135 .await
136 .map_err(process_sqlite_error)?
137 }
138}
139
140fn trigger_tx_outcome<T>(
141 result: Result<T, lash_core::PluginError>,
142) -> TxOutcome<Result<T, lash_core::PluginError>> {
143 match result {
144 Ok(value) => TxOutcome::Commit(Ok(value)),
145 Err(err) => TxOutcome::Rollback(Err(err)),
146 }
147}
148
149#[async_trait::async_trait]
150impl lash_core::TriggerStore for SqliteTriggerStore {
151 fn durability_tier(&self) -> DurabilityTier {
152 DurabilityTier::Durable
153 }
154
155 async fn register_subscription(
156 &self,
157 draft: lash_core::TriggerSubscriptionDraft,
158 ) -> Result<lash_core::TriggerSubscriptionRecord, lash_core::PluginError> {
159 draft.validate()?;
160 let now = self.clock.timestamp_ms();
161 self.conn
162 .write_flow(move |tx| {
163 Ok(trigger_tx_outcome((|| {
164 tx.execute("INSERT INTO trigger_subscription_seq DEFAULT VALUES", [])
165 .map_err(process_sqlite_error)?;
166 let seq = tx.last_insert_rowid();
167 let handle = format!("trigger:{seq}");
168 let subscription_id = format!("subscription:{seq}");
169 let record = lash_core::TriggerSubscriptionRecord {
170 subscription_id: subscription_id.clone(),
171 registrant: draft.registrant,
172 env_ref: draft.env_ref,
173 wake_target: draft.wake_target,
174 handle,
175 name: draft.name,
176 source_type: draft.source_type,
177 source_key: draft.source_key,
178 source: draft.source,
179 payload_schema: draft.payload_schema,
180 target: draft.target,
181 target_identity: draft.target_identity,
182 event_types: draft.event_types,
183 input_template: draft.input_template,
184 target_label: draft.target_label,
185 enabled: true,
186 created_at_ms: now,
187 updated_at_ms: now,
188 };
189 tx.execute(
190 "INSERT INTO trigger_subscriptions (
191 subscription_id, registrant_scope_id, handle, source_type, source_key,
192 enabled, created_at_ms, updated_at_ms, record_json
193 )
194 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
195 params![
196 record.subscription_id.as_str(),
197 record.registrant_scope_id().as_str(),
198 record.handle.as_str(),
199 record.source_type.as_str(),
200 record.source_key.as_str(),
201 i64::from(record.enabled),
202 record.created_at_ms as i64,
203 record.updated_at_ms as i64,
204 Self::encode_json(&record)?,
205 ],
206 )
207 .map_err(process_sqlite_error)?;
208 Ok(record)
209 })()))
210 })
211 .await
212 .map_err(process_sqlite_error)?
213 }
214
215 async fn list_subscriptions(
216 &self,
217 filter: lash_core::TriggerSubscriptionFilter,
218 ) -> Result<Vec<lash_core::TriggerSubscriptionRecord>, lash_core::PluginError> {
219 self.conn
220 .call(move |conn| {
221 Ok((|| {
222 let mut sql =
223 "SELECT subscription_id, record_json FROM trigger_subscriptions WHERE 1 = 1"
224 .to_string();
225 let mut values = Vec::<rusqlite::types::Value>::new();
226 if let Some(registrant_scope_id) = filter.effective_registrant_scope_id() {
227 sql.push_str(" AND registrant_scope_id = ?");
228 values.push(registrant_scope_id.into());
229 }
230 if let Some(handle) = filter.handle.as_ref() {
231 sql.push_str(" AND handle = ?");
232 values.push(handle.clone().into());
233 }
234 if let Some(source_type) = filter.source_type.as_ref() {
235 sql.push_str(" AND source_type = ?");
236 values.push(source_type.clone().into());
237 }
238 if let Some(source_key) = filter.source_key.as_ref() {
239 sql.push_str(" AND source_key = ?");
240 values.push(source_key.clone().into());
241 }
242 if let Some(enabled) = filter.enabled {
243 sql.push_str(" AND enabled = ?");
244 values.push(i64::from(enabled).into());
245 }
246 sql.push_str(" ORDER BY registrant_scope_id ASC, handle ASC");
247 let mut stmt = conn.prepare(&sql).map_err(process_sqlite_error)?;
248 let rows = stmt
249 .query_map(rusqlite::params_from_iter(values.iter()), |row| {
250 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
251 })
252 .map_err(process_sqlite_error)?;
253 let mut records = Vec::new();
254 for row in rows {
255 let (subscription_id, json) = row.map_err(process_sqlite_error)?;
256 let record = match Self::decode_subscription(json) {
257 Ok(record) => record,
258 Err(err) => {
259 tracing::warn!(
260 error = %err,
261 subscription_id,
262 "skipping malformed trigger subscription during listing"
263 );
264 continue;
265 }
266 };
267 if filter.matches(&record) {
268 records.push(record);
269 }
270 }
271 Ok(records)
272 })())
273 })
274 .await
275 .map_err(process_sqlite_error)?
276 }
277
278 async fn cancel_subscription(
279 &self,
280 registrant_scope_id: &str,
281 handle: &str,
282 ) -> Result<bool, lash_core::PluginError> {
283 self.set_subscription_enabled(registrant_scope_id, handle, false)
284 .await
285 }
286
287 async fn set_subscription_enabled(
288 &self,
289 registrant_scope_id: &str,
290 handle: &str,
291 enabled: bool,
292 ) -> Result<bool, lash_core::PluginError> {
293 let registrant_scope_id = registrant_scope_id.to_string();
294 let handle = handle.to_string();
295 let updated_at_ms = self.clock.timestamp_ms();
296 self.conn
297 .write_flow(move |tx| {
298 Ok(trigger_tx_outcome((|| {
299 let selected: Option<(String, i64, String)> = tx
300 .query_row(
301 "SELECT subscription_id, enabled, record_json
302 FROM trigger_subscriptions
303 WHERE registrant_scope_id = ?1 AND handle = ?2",
304 params![registrant_scope_id.as_str(), handle.as_str()],
305 |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
306 )
307 .optional()
308 .map_err(process_sqlite_error)?;
309 let Some((subscription_id, stored_enabled, json)) = selected else {
310 return Ok(false);
311 };
312 let changed = (stored_enabled != 0) != enabled;
313 match Self::decode_subscription(json) {
314 Ok(mut record) => {
315 record.enabled = enabled;
316 record.updated_at_ms = updated_at_ms;
317 tx.execute(
318 "UPDATE trigger_subscriptions
319 SET enabled = ?3, updated_at_ms = ?4, record_json = ?5
320 WHERE subscription_id = ?1 AND handle = ?2",
321 params![
322 subscription_id.as_str(),
323 handle.as_str(),
324 i64::from(record.enabled),
325 record.updated_at_ms as i64,
326 Self::encode_json(&record)?,
327 ],
328 )
329 .map_err(process_sqlite_error)?;
330 }
331 Err(err) => {
332 tracing::warn!(
333 error = %err,
334 subscription_id,
335 handle,
336 "disabling malformed trigger subscription without rewriting record JSON"
337 );
338 tx.execute(
339 "UPDATE trigger_subscriptions
340 SET enabled = ?3, updated_at_ms = ?4
341 WHERE subscription_id = ?1 AND handle = ?2",
342 params![
343 subscription_id.as_str(),
344 handle.as_str(),
345 i64::from(enabled),
346 updated_at_ms as i64,
347 ],
348 )
349 .map_err(process_sqlite_error)?;
350 }
351 }
352 Ok(changed)
353 })()))
354 })
355 .await
356 .map_err(process_sqlite_error)?
357 }
358
359 async fn delete_subscription(
360 &self,
361 registrant_scope_id: &str,
362 handle: &str,
363 ) -> Result<bool, lash_core::PluginError> {
364 let registrant_scope_id = registrant_scope_id.to_string();
365 let handle = handle.to_string();
366 self.conn
367 .write_flow(move |tx| {
368 Ok(trigger_tx_outcome((|| {
369 let deleted = tx
370 .execute(
371 "DELETE FROM trigger_subscriptions
372 WHERE registrant_scope_id = ?1 AND handle = ?2",
373 params![registrant_scope_id.as_str(), handle.as_str()],
374 )
375 .map_err(process_sqlite_error)?;
376 Ok(deleted != 0)
377 })()))
378 })
379 .await
380 .map_err(process_sqlite_error)?
381 }
382
383 async fn delete_session_subscriptions(
384 &self,
385 session_id: &str,
386 ) -> Result<usize, lash_core::PluginError> {
387 let session_id = session_id.to_string();
388 self.conn
389 .write_flow(move |tx| {
390 Ok(trigger_tx_outcome((|| {
391 let mut stmt = tx
392 .prepare("SELECT subscription_id, record_json FROM trigger_subscriptions")
393 .map_err(process_sqlite_error)?;
394 let rows = stmt
395 .query_map([], |row| {
396 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
397 })
398 .map_err(process_sqlite_error)?;
399 let mut subscription_ids = Vec::new();
400 for row in rows {
401 let (subscription_id, json) = row.map_err(process_sqlite_error)?;
402 let record = match Self::decode_subscription(json) {
403 Ok(record) => record,
404 Err(err) => {
405 tracing::warn!(
406 error = %err,
407 subscription_id,
408 "skipping malformed trigger subscription during session delete"
409 );
410 continue;
411 }
412 };
413 if record.registrant_session_id() == Some(session_id.as_str()) {
414 subscription_ids.push(subscription_id);
415 }
416 }
417 drop(stmt);
418 let mut deleted = 0usize;
419 for subscription_id in subscription_ids {
420 deleted = deleted.saturating_add(
421 tx.execute(
422 "DELETE FROM trigger_subscriptions WHERE subscription_id = ?1",
423 params![subscription_id.as_str()],
424 )
425 .map_err(process_sqlite_error)?,
426 );
427 }
428 Ok(deleted)
429 })()))
430 })
431 .await
432 .map_err(process_sqlite_error)?
433 }
434
435 async fn record_occurrence(
436 &self,
437 request: lash_core::TriggerOccurrenceRequest,
438 ) -> Result<lash_core::TriggerOccurrenceRecord, lash_core::PluginError> {
439 lash_core::validate_trigger_occurrence_request(&request)?;
440 let request_hash = lash_core::trigger_occurrence_request_hash(&request)?;
441 let occurrence_id = lash_core::deterministic_occurrence_id(&request)?;
442 let occurred_at_ms = self.clock.timestamp_ms();
443 self.conn
444 .write_flow(move |tx| {
445 Ok(trigger_tx_outcome((|| {
446 let existing: Option<(String, String)> = tx
447 .query_row(
448 "SELECT request_hash, record_json
449 FROM trigger_occurrences
450 WHERE idempotency_key = ?1",
451 params![request.idempotency_key.as_str()],
452 |row| Ok((row.get(0)?, row.get(1)?)),
453 )
454 .optional()
455 .map_err(process_sqlite_error)?;
456 if let Some((existing_hash, existing_json)) = existing {
457 if existing_hash != request_hash {
458 return Err(lash_core::PluginError::Session(format!(
459 "trigger occurrence idempotency conflict for `{}`",
460 request.idempotency_key
461 )));
462 }
463 return Self::decode_occurrence(existing_json);
464 }
465 let record = lash_core::TriggerOccurrenceRecord {
466 occurrence_id: occurrence_id.clone(),
467 source_type: request.source_type,
468 source_key: request.source_key,
469 payload: request.payload,
470 idempotency_key: request.idempotency_key,
471 source: request.source,
472 session_id: request.session_id,
473 occurred_at_ms,
474 };
475 tx.execute(
476 "INSERT INTO trigger_occurrences (
477 occurrence_id, idempotency_key, request_hash, source_type,
478 source_key, occurred_at_ms, record_json
479 )
480 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
481 params![
482 record.occurrence_id.as_str(),
483 record.idempotency_key.as_str(),
484 request_hash.as_str(),
485 record.source_type.as_str(),
486 record.source_key.as_str(),
487 record.occurred_at_ms as i64,
488 Self::encode_json(&record)?,
489 ],
490 )
491 .map_err(process_sqlite_error)?;
492 Ok(record)
493 })()))
494 })
495 .await
496 .map_err(process_sqlite_error)?
497 }
498
499 async fn list_occurrences(
500 &self,
501 filter: lash_core::TriggerOccurrenceFilter,
502 ) -> Result<Vec<lash_core::TriggerOccurrenceRecord>, lash_core::PluginError> {
503 self.conn
504 .call(move |conn| {
505 Ok((|| {
506 let mut sql =
507 "SELECT occurrence_id, record_json FROM trigger_occurrences WHERE 1 = 1"
508 .to_string();
509 let mut values = Vec::<rusqlite::types::Value>::new();
510 if let Some(source_type) = filter.source_type.as_ref() {
511 sql.push_str(" AND source_type = ?");
512 values.push(source_type.clone().into());
513 }
514 if let Some(source_key) = filter.source_key.as_ref() {
515 sql.push_str(" AND source_key = ?");
516 values.push(source_key.clone().into());
517 }
518 if let Some(start_ms) = filter.occurred_at_start_ms {
519 sql.push_str(" AND occurred_at_ms >= ?");
520 values.push((start_ms as i64).into());
521 }
522 if let Some(end_ms) = filter.occurred_at_end_ms {
523 sql.push_str(" AND occurred_at_ms < ?");
524 values.push((end_ms as i64).into());
525 }
526 sql.push_str(" ORDER BY occurred_at_ms ASC, occurrence_id ASC");
527 let mut stmt = conn.prepare(&sql).map_err(process_sqlite_error)?;
528 let rows = stmt
529 .query_map(rusqlite::params_from_iter(values.iter()), |row| {
530 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
531 })
532 .map_err(process_sqlite_error)?;
533 let mut records = Vec::new();
534 for row in rows {
535 let (occurrence_id, json) = row.map_err(process_sqlite_error)?;
536 match Self::decode_occurrence(json) {
537 Ok(record) => records.push(record),
538 Err(err) => tracing::warn!(
539 error = %err,
540 occurrence_id,
541 "skipping malformed trigger occurrence during listing"
542 ),
543 }
544 }
545 Ok(records)
546 })())
547 })
548 .await
549 .map_err(process_sqlite_error)?
550 }
551
552 async fn reserve_matching_deliveries(
553 &self,
554 occurrence_id: &str,
555 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
556 let occurrence_id = occurrence_id.to_string();
557 let created_at_ms = self.clock.timestamp_ms();
558 self.conn
559 .write_flow(move |tx| {
560 Ok(trigger_tx_outcome((|| {
561 let occurrence_json: Option<String> = tx
562 .query_row(
563 "SELECT record_json
564 FROM trigger_occurrences
565 WHERE occurrence_id = ?1",
566 params![occurrence_id.as_str()],
567 |row| row.get(0),
568 )
569 .optional()
570 .map_err(process_sqlite_error)?;
571 let Some(occurrence_json) = occurrence_json else {
572 return Err(lash_core::PluginError::Session(format!(
573 "unknown trigger occurrence `{occurrence_id}`"
574 )));
575 };
576 let occurrence = Self::decode_occurrence(occurrence_json)?;
577 let subscriptions = {
578 let mut sql =
579 "SELECT subscription_id, record_json
580 FROM trigger_subscriptions
581 WHERE enabled = 1 AND source_type = ? AND source_key = ?"
582 .to_string();
583 let mut values: Vec<rusqlite::types::Value> = vec![
584 occurrence.source_type.clone().into(),
585 occurrence.source_key.clone().into(),
586 ];
587 if let Some(session_id) = occurrence.session_id.as_deref() {
588 let scope_id = format!("session:{session_id}");
589 sql.push_str(
590 " AND (registrant_scope_id = ? OR registrant_scope_id LIKE ? ESCAPE '\\')",
591 );
592 values.push(scope_id.clone().into());
593 values.push(
594 format!("{}/frame:%", escape_sqlite_like(&scope_id)).into(),
595 );
596 }
597 sql.push_str(" ORDER BY registrant_scope_id ASC, handle ASC");
598 let mut stmt = tx
599 .prepare(&sql)
600 .map_err(process_sqlite_error)?;
601 let rows = stmt
602 .query_map(rusqlite::params_from_iter(values.iter()), |row| {
603 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
604 })
605 .map_err(process_sqlite_error)?;
606 let mut subscriptions = Vec::new();
607 for row in rows {
608 let (subscription_id, json) = row.map_err(process_sqlite_error)?;
609 match Self::decode_subscription(json) {
610 Ok(subscription) => subscriptions.push(subscription),
611 Err(err) => tracing::warn!(
612 error = %err,
613 subscription_id,
614 occurrence_id = %occurrence.occurrence_id,
615 "skipping malformed trigger subscription during delivery reservation"
616 ),
617 }
618 }
619 subscriptions
620 };
621 let mut planned = Vec::with_capacity(subscriptions.len());
622 for subscription in subscriptions {
623 let process_id = lash_core::deterministic_delivery_process_id(
624 &occurrence.occurrence_id,
625 &subscription.subscription_id,
626 )?;
627 planned.push((subscription, process_id));
628 }
629 if planned.is_empty() {
630 return Ok(Vec::new());
631 }
632
633 let mut insert =
634 "INSERT INTO trigger_deliveries (
635 occurrence_id, subscription_id, process_id, created_at_ms
636 ) VALUES "
637 .to_string();
638 let mut insert_values: Vec<rusqlite::types::Value> =
639 Vec::with_capacity(planned.len() * 4);
640 for (index, (subscription, process_id)) in planned.iter().enumerate() {
641 if index > 0 {
642 insert.push_str(", ");
643 }
644 insert.push_str("(?, ?, ?, ?)");
645 insert_values.push(occurrence.occurrence_id.clone().into());
646 insert_values.push(subscription.subscription_id.clone().into());
647 insert_values.push(process_id.clone().into());
648 insert_values.push((created_at_ms as i64).into());
649 }
650 insert.push_str(
651 " ON CONFLICT DO NOTHING RETURNING subscription_id, created_at_ms",
652 );
653 #[cfg(test)]
654 record_reservation_statement();
655 let mut stmt = tx.prepare(&insert).map_err(process_sqlite_error)?;
656 let inserted_rows = stmt
657 .query_map(rusqlite::params_from_iter(insert_values.iter()), |row| {
658 Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
659 })
660 .map_err(process_sqlite_error)?
661 .collect::<Result<Vec<_>, _>>()
662 .map_err(process_sqlite_error)?;
663 drop(stmt);
664 let mut created_at_by_subscription =
665 std::collections::BTreeMap::from_iter(inserted_rows);
666 let inserted_subscription_ids = created_at_by_subscription
667 .keys()
668 .cloned()
669 .collect::<std::collections::BTreeSet<_>>();
670 let conflicted = planned
671 .iter()
672 .filter(|(subscription, _)| {
673 !inserted_subscription_ids.contains(&subscription.subscription_id)
674 })
675 .collect::<Vec<_>>();
676 if !conflicted.is_empty() {
677 let mut select =
678 "SELECT subscription_id, created_at_ms FROM trigger_deliveries
679 WHERE (occurrence_id, subscription_id) IN ("
680 .to_string();
681 let mut select_values: Vec<rusqlite::types::Value> =
682 Vec::with_capacity(conflicted.len() * 2);
683 for (index, (subscription, _)) in conflicted.iter().enumerate() {
684 if index > 0 {
685 select.push_str(", ");
686 }
687 select.push_str("(?, ?)");
688 select_values.push(occurrence.occurrence_id.clone().into());
689 select_values.push(subscription.subscription_id.clone().into());
690 }
691 select.push(')');
692 #[cfg(test)]
693 record_reservation_statement();
694 let mut stmt = tx.prepare(&select).map_err(process_sqlite_error)?;
695 let rows = stmt
696 .query_map(rusqlite::params_from_iter(select_values.iter()), |row| {
697 Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
698 })
699 .map_err(process_sqlite_error)?;
700 for row in rows {
701 let (subscription_id, stored_created_at_ms) =
702 row.map_err(process_sqlite_error)?;
703 created_at_by_subscription
704 .insert(subscription_id, stored_created_at_ms);
705 }
706 }
707
708 let mut reservations = Vec::with_capacity(planned.len());
709 for (subscription, process_id) in planned {
710 let inserted =
711 inserted_subscription_ids.contains(&subscription.subscription_id);
712 let stored_created_at_ms = created_at_by_subscription
713 .get(&subscription.subscription_id)
714 .copied()
715 .ok_or_else(|| {
716 lash_core::PluginError::Session(format!(
717 "trigger delivery `{}/{}` disappeared during reservation",
718 occurrence.occurrence_id, subscription.subscription_id
719 ))
720 })?;
721 reservations.push(lash_core::TriggerDeliveryReservation {
722 occurrence: occurrence.clone(),
723 subscription,
724 process_id,
725 created_at_ms: stored_created_at_ms as u64,
726 reservation_status: if inserted {
727 lash_core::TriggerDeliveryReservationStatus::Reserved
728 } else {
729 lash_core::TriggerDeliveryReservationStatus::AlreadyReserved
730 },
731 });
732 }
733 Ok(reservations)
734 })()))
735 })
736 .await
737 .map_err(process_sqlite_error)?
738 }
739
740 async fn list_deliveries_by_occurrence_id(
741 &self,
742 occurrence_id: &str,
743 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
744 self.list_deliveries_where("d.occurrence_id = ?1", occurrence_id.to_string())
745 .await
746 }
747
748 async fn list_deliveries_by_subscription_id(
749 &self,
750 subscription_id: &str,
751 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
752 self.list_deliveries_where("d.subscription_id = ?1", subscription_id.to_string())
753 .await
754 }
755
756 async fn list_deliveries_by_process_id(
757 &self,
758 process_id: &str,
759 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
760 self.list_deliveries_where("d.process_id = ?1", process_id.to_string())
761 .await
762 }
763}
764
765fn escape_sqlite_like(value: &str) -> String {
766 let mut escaped = String::with_capacity(value.len());
767 for ch in value.chars() {
768 if matches!(ch, '\\' | '%' | '_') {
769 escaped.push('\\');
770 }
771 escaped.push(ch);
772 }
773 escaped
774}
775
776#[cfg(test)]
777mod perf_tests {
778 use super::*;
779 use lash_core::TriggerStore;
780
781 fn subscription_draft(
782 session_id: &str,
783 source_key: &str,
784 index: usize,
785 ) -> lash_core::TriggerSubscriptionDraft {
786 let scope = lash_core::SessionScope::new(session_id);
787 lash_core::TriggerSubscriptionDraft {
788 registrant: lash_core::ProcessOriginator::session(scope.clone()),
789 env_ref: lash_core::ProcessExecutionEnvRef::new(format!("env:{index}")),
790 wake_target: Some(scope),
791 name: Some(format!("subscription-{index}")),
792 source_type: "perf.trigger".to_string(),
793 source_key: source_key.to_string(),
794 source: serde_json::json!({}),
795 payload_schema: lash_core::LashSchema::new(serde_json::json!({ "type": "object" })),
796 target: lash_core::ProcessInput::Engine {
797 kind: "test".to_string(),
798 payload: serde_json::json!({ "index": index }),
799 },
800 target_identity: lash_core::ProcessIdentity::new("test"),
801 event_types: Vec::new(),
802 input_template: BTreeMap::new(),
803 target_label: None,
804 }
805 }
806
807 #[tokio::test]
808 async fn reservation_statement_budget_is_constant_across_fanout() {
809 for subscription_count in [1, 8, 64] {
810 let store = SqliteTriggerStore::memory().await.expect("trigger store");
811 let source_key =
812 lash_core::empty_trigger_source_key("perf.trigger").expect("source key");
813 for index in 0..subscription_count {
814 store
815 .register_subscription(subscription_draft(
816 &format!("session-{index}"),
817 &source_key,
818 index,
819 ))
820 .await
821 .expect("register subscription");
822 }
823 let occurrence = store
824 .record_occurrence(lash_core::TriggerOccurrenceRequest::new(
825 "perf.trigger",
826 &source_key,
827 serde_json::json!({ "fanout": subscription_count }),
828 format!("fanout-{subscription_count}"),
829 ))
830 .await
831 .expect("record occurrence");
832
833 RESERVATION_STATEMENT_COUNT.store(0, std::sync::atomic::Ordering::SeqCst);
834 let fresh = store
835 .reserve_matching_deliveries(&occurrence.occurrence_id)
836 .await
837 .expect("fresh reservation");
838 let fresh_statements =
839 RESERVATION_STATEMENT_COUNT.load(std::sync::atomic::Ordering::SeqCst);
840 assert_eq!(fresh.len(), subscription_count);
841 assert_eq!(fresh_statements, 1);
842
843 RESERVATION_STATEMENT_COUNT.store(0, std::sync::atomic::Ordering::SeqCst);
844 let replay = store
845 .reserve_matching_deliveries(&occurrence.occurrence_id)
846 .await
847 .expect("replayed reservation");
848 let replay_statements =
849 RESERVATION_STATEMENT_COUNT.load(std::sync::atomic::Ordering::SeqCst);
850 assert_eq!(replay.len(), subscription_count);
851 assert_eq!(replay_statements, 2);
852 eprintln!(
853 "trigger fanout N={subscription_count}: old={} statements, fresh={fresh_statements}, replay={replay_statements}",
854 subscription_count * 2,
855 );
856 }
857 }
858}