1use std::{mem::take, ops::Bound, sync::Arc};
5
6use reifydb_codec::{key::encoded::EncodedKey, row::bytes::EncodedBytes};
7use reifydb_core::{
8 common::{CommitVersion, SourceVersion},
9 event::EventBus,
10 execution::ExecutionResult,
11 interface::{
12 WithEventBus,
13 catalog::{object::ObjectId, storage::StorageId},
14 change::{Change, ChangeOrigin, Diff},
15 store::{MultiVersionBatch, MultiVersionRow},
16 },
17 key::{
18 any::TaggedKey,
19 bound::TaggedKeyBoundRange,
20 row::{StoragePartitionedRowKey, StorageRowKey},
21 },
22};
23use reifydb_runtime::context::clock::Clock;
24use reifydb_value::{Result, error::Diagnostic, params::Params, reifydb_assertions, value::identity::IdentityId};
25use tracing::instrument;
26
27use crate::{
28 TransactionId,
29 accumulator::ChangeAccumulator,
30 change::{RowChange, TransactionalCatalogChanges},
31 dictionary::DictionaryAllocatorRegistry,
32 error::TransactionError,
33 interceptor::{
34 WithInterceptors,
35 authentication::{AuthenticationPostCreateInterceptor, AuthenticationPreDeleteInterceptor},
36 chain::InterceptorChain as Chain,
37 dictionary::{
38 DictionaryPostCreateInterceptor, DictionaryPostUpdateInterceptor,
39 DictionaryPreDeleteInterceptor, DictionaryPreUpdateInterceptor,
40 },
41 dictionary_row::{
42 DictionaryRowPostDeleteInterceptor, DictionaryRowPostInsertInterceptor,
43 DictionaryRowPostUpdateInterceptor, DictionaryRowPreDeleteInterceptor,
44 DictionaryRowPreInsertInterceptor, DictionaryRowPreUpdateInterceptor,
45 },
46 granted_role::{GrantedRolePostCreateInterceptor, GrantedRolePreDeleteInterceptor},
47 identity::{IdentityPostCreateInterceptor, IdentityPreDeleteInterceptor},
48 identity_attribute::{IdentityAttributePostCreateInterceptor, IdentityAttributePreDeleteInterceptor},
49 identity_attribute_value::{
50 IdentityAttributeValuePostCreateInterceptor, IdentityAttributeValuePreDeleteInterceptor,
51 },
52 interceptors::Interceptors,
53 namespace::{
54 NamespacePostCreateInterceptor, NamespacePostUpdateInterceptor, NamespacePreDeleteInterceptor,
55 NamespacePreUpdateInterceptor,
56 },
57 ringbuffer::{
58 RingBufferPostCreateInterceptor, RingBufferPostUpdateInterceptor,
59 RingBufferPreDeleteInterceptor, RingBufferPreUpdateInterceptor,
60 },
61 ringbuffer_row::{
62 RingBufferRowPostDeleteInterceptor, RingBufferRowPostInsertInterceptor,
63 RingBufferRowPostUpdateInterceptor, RingBufferRowPreDeleteInterceptor,
64 RingBufferRowPreInsertInterceptor, RingBufferRowPreUpdateInterceptor,
65 },
66 role::{RolePostCreateInterceptor, RolePreDeleteInterceptor},
67 series::{
68 SeriesPostCreateInterceptor, SeriesPostUpdateInterceptor, SeriesPreDeleteInterceptor,
69 SeriesPreUpdateInterceptor,
70 },
71 series_row::{
72 SeriesRowPostDeleteInterceptor, SeriesRowPostInsertInterceptor, SeriesRowPostUpdateInterceptor,
73 SeriesRowPreDeleteInterceptor, SeriesRowPreInsertInterceptor, SeriesRowPreUpdateInterceptor,
74 },
75 table::{
76 TablePostCreateInterceptor, TablePostUpdateInterceptor, TablePreDeleteInterceptor,
77 TablePreUpdateInterceptor,
78 },
79 table_row::{
80 TableRowPostDeleteInterceptor, TableRowPostInsertInterceptor, TableRowPostUpdateInterceptor,
81 TableRowPreDeleteInterceptor, TableRowPreInsertInterceptor, TableRowPreUpdateInterceptor,
82 },
83 transaction::{PostCommitContext, PostCommitInterceptor, PreCommitContext, PreCommitInterceptor},
84 view::{
85 ViewPostCreateInterceptor, ViewPostUpdateInterceptor, ViewPreDeleteInterceptor,
86 ViewPreUpdateInterceptor,
87 },
88 },
89 multi::{
90 RangeScope,
91 pending::PendingWrites,
92 transaction::{MultiTransaction, write::MultiWriteTransaction},
93 },
94 single::{SingleTransaction, read::SingleReadTransaction, write::SingleWriteTransaction},
95 transaction::{RqlExecutor, Transaction, apply_pre_commit_writes, collect_transaction_writes, write::Write},
96};
97
98pub struct CommandTransaction {
99 pub multi: MultiTransaction,
100 pub single: SingleTransaction,
101 state: TransactionState,
102
103 pub cmd: Option<MultiWriteTransaction>,
104 pub event_bus: EventBus,
105
106 pub(crate) row_changes: Vec<RowChange>,
107 pub(crate) interceptors: Interceptors,
108
109 pub(crate) accumulator: ChangeAccumulator,
110
111 pub identity: IdentityId,
112
113 pub(crate) executor: Option<Arc<dyn RqlExecutor>>,
114
115 pub(crate) dictionary_allocators: Option<DictionaryAllocatorRegistry>,
116
117 pub(crate) clock: Clock,
118
119 poison_cause: Option<Diagnostic>,
120}
121
122#[derive(Clone, Copy, PartialEq)]
123enum TransactionState {
124 Active,
125 Committed,
126 RolledBack,
127 Poisoned,
128}
129
130impl CommandTransaction {
131 #[instrument(name = "transaction::command::new", level = "debug", skip_all)]
132 pub fn new(
133 multi: MultiTransaction,
134 single: SingleTransaction,
135 event_bus: EventBus,
136 interceptors: Interceptors,
137 identity: IdentityId,
138 clock: Clock,
139 ) -> Result<Self> {
140 let cmd = multi.begin_command()?;
141 Ok(Self {
142 cmd: Some(cmd),
143 multi,
144 single,
145 state: TransactionState::Active,
146 event_bus,
147 interceptors,
148 row_changes: Vec::new(),
149 accumulator: ChangeAccumulator::new(),
150 identity,
151 executor: None,
152 dictionary_allocators: None,
153 clock,
154 poison_cause: None,
155 })
156 }
157
158 pub fn set_executor(&mut self, executor: Arc<dyn RqlExecutor>) {
159 self.executor = Some(executor);
160 }
161
162 pub fn set_dictionary_allocators(&mut self, registry: DictionaryAllocatorRegistry) {
163 self.dictionary_allocators = Some(registry);
164 }
165
166 pub fn dictionary_allocators(&self) -> Option<DictionaryAllocatorRegistry> {
167 self.dictionary_allocators.clone()
168 }
169
170 pub fn rql(&mut self, rql: &str, params: Params) -> ExecutionResult {
171 if let Err(e) = self.check_active() {
172 return ExecutionResult {
173 frames: vec![],
174 error: Some(e),
175 metrics: Default::default(),
176 };
177 }
178 let executor = self.executor.clone().expect("RqlExecutor not set");
179 let result = executor.rql(&mut Transaction::Command(self), rql, params);
180 if let Some(ref e) = result.error {
181 self.poison(*e.0.clone());
182 }
183 result
184 }
185
186 #[instrument(name = "transaction::command::event_bus", level = "trace", skip(self))]
187 pub fn event_bus(&self) -> &EventBus {
188 &self.event_bus
189 }
190
191 fn check_active(&self) -> Result<()> {
192 match self.state {
193 TransactionState::Active => Ok(()),
194 TransactionState::Committed => Err(TransactionError::AlreadyCommitted.into()),
195 TransactionState::RolledBack => Err(TransactionError::AlreadyRolledBack.into()),
196 TransactionState::Poisoned => Err(TransactionError::Poisoned {
197 cause: Box::new(self.poison_cause.clone().unwrap()),
198 }
199 .into()),
200 }
201 }
202
203 pub(crate) fn poison(&mut self, cause: Diagnostic) {
204 self.state = TransactionState::Poisoned;
205 self.poison_cause = Some(cause);
206 }
207
208 #[instrument(name = "transaction::command::commit", level = "debug", skip(self))]
209 pub fn commit(&mut self) -> Result<CommitVersion> {
210 self.check_active()?;
211 let mut ctx = self.build_pre_commit_context()?;
212 self.interceptors.pre_commit.execute(&mut ctx)?;
213 self.finalize_commit(ctx, false)
214 }
215
216 #[inline]
217 fn build_pre_commit_context(&mut self) -> Result<PreCommitContext> {
218 let transaction_writes = collect_transaction_writes(self.pending_writes());
219 Ok(PreCommitContext {
220 flow_changes: self.accumulator.take_changes(CommitVersion(0), self.clock.now())?,
221 pending_writes: Vec::new(),
222 transaction_writes,
223 view_entries: Vec::new(),
224 })
225 }
226
227 fn finalize_commit(&mut self, ctx: PreCommitContext, unchecked: bool) -> Result<CommitVersion> {
228 let Some(mut multi) = self.cmd.take() else {
229 unreachable!("Transaction state inconsistency")
230 };
231 reifydb_assertions! {
232 assert!(
233 self.state == TransactionState::Active,
234 "finalize_commit entered in non-Active state; commit()/commit_unchecked() must \
235 pass check_active() first, otherwise this double-commits or commits a \
236 rolled-back/poisoned transaction"
237 );
238 }
239 let id = self.apply_writes_and_mark_committed(&mut multi, &ctx)?;
240 let row_changes = take(&mut self.row_changes);
241 let flow_changes = self.merge_view_entries(ctx.flow_changes, ctx.view_entries)?;
242 let version = self.commit_and_post(multi, id, flow_changes, row_changes, unchecked)?;
243 Ok(version)
244 }
245
246 #[inline]
247 fn apply_writes_and_mark_committed(
248 &mut self,
249 multi: &mut MultiWriteTransaction,
250 ctx: &PreCommitContext,
251 ) -> Result<TransactionId> {
252 apply_pre_commit_writes(multi, &ctx.pending_writes)?;
253 let id = multi.id();
254 self.state = TransactionState::Committed;
255 Ok(id)
256 }
257
258 #[inline]
259 fn merge_view_entries(
260 &self,
261 mut flow_changes: Vec<Change>,
262 view_entries: Vec<(ObjectId, Diff)>,
263 ) -> Result<Vec<Change>> {
264 if !view_entries.is_empty() {
265 let mut accumulator = ChangeAccumulator::new();
266 for (object, diff) in view_entries {
267 accumulator.track(object, diff);
268 }
269 let changed_at = self.clock.now();
270 flow_changes.extend(accumulator.take_changes(CommitVersion(0), changed_at)?);
271 }
272 Ok(flow_changes)
273 }
274
275 #[inline]
276 fn commit_and_post(
277 &self,
278 mut multi: MultiWriteTransaction,
279 id: TransactionId,
280 flow_changes: Vec<Change>,
281 row_changes: Vec<RowChange>,
282 unchecked: bool,
283 ) -> Result<CommitVersion> {
284 let changes = TransactionalCatalogChanges::default();
285 let version = if unchecked {
286 multi.commit_unchecked(flow_changes)?
287 } else {
288 multi.commit(flow_changes)?
289 };
290 let _self_lease = multi.take_self_lease();
291 self.interceptors.post_commit.execute(PostCommitContext::new(id, version, changes, row_changes))?;
292 Ok(version)
293 }
294
295 pub fn execute_bulk_unchecked<F, R>(&mut self, body: F) -> Result<R>
296 where
297 F: FnOnce(&mut CommandTransaction) -> Result<R>,
298 {
299 self.disable_conflict_tracking()?;
300 let r = match body(self) {
301 Ok(r) => r,
302 Err(e) => {
303 let _ = self.rollback();
304 return Err(e);
305 }
306 };
307 self.commit_unchecked()?;
308 Ok(r)
309 }
310
311 #[instrument(name = "transaction::command::commit_unchecked", level = "debug", skip(self))]
312 pub fn commit_unchecked(&mut self) -> Result<CommitVersion> {
313 self.check_active()?;
314 let mut ctx = self.build_pre_commit_context()?;
315 self.interceptors.pre_commit.execute(&mut ctx)?;
316 self.finalize_commit(ctx, true)
317 }
318
319 #[instrument(name = "transaction::command::rollback", level = "debug", skip(self))]
320 pub fn rollback(&mut self) -> Result<()> {
321 self.check_active()?;
322 if let Some(mut multi) = self.cmd.take() {
323 self.state = TransactionState::RolledBack;
324 multi.rollback()
325 } else {
326 unreachable!("Transaction state inconsistency")
327 }
328 }
329
330 #[instrument(name = "transaction::command::pending_writes", level = "trace", skip(self))]
331 pub fn pending_writes(&self) -> &PendingWrites {
332 self.cmd.as_ref().unwrap().pending_writes()
333 }
334
335 #[instrument(name = "transaction::command::with_single_command", level = "trace", skip(self, keys, f))]
336 pub fn with_single_command<'a, I, F, R>(&self, keys: I, f: F) -> Result<R>
337 where
338 I: IntoIterator<Item = &'a EncodedKey> + Send,
339 F: FnOnce(&mut SingleWriteTransaction<'_>) -> Result<R> + Send,
340 R: Send,
341 {
342 self.check_active()?;
343 self.single.with_command(keys, f)
344 }
345
346 #[instrument(name = "transaction::command::begin_single_query", level = "trace", skip(self, keys))]
347 pub fn begin_single_query<'a, I>(&self, keys: I) -> Result<SingleReadTransaction<'_>>
348 where
349 I: IntoIterator<Item = &'a EncodedKey>,
350 {
351 self.check_active()?;
352 self.single.begin_query(keys)
353 }
354
355 #[instrument(name = "transaction::command::begin_single_command", level = "trace", skip(self, keys))]
356 pub fn begin_single_command<'a, I>(&self, keys: I) -> Result<SingleWriteTransaction<'_>>
357 where
358 I: IntoIterator<Item = &'a EncodedKey>,
359 {
360 self.check_active()?;
361 self.single.begin_command(keys)
362 }
363
364 pub fn track_row_change(&mut self, changes: &[RowChange]) {
365 self.row_changes.extend_from_slice(changes);
366 }
367
368 pub fn track_flow_change(&mut self, change: Change) {
369 if let ChangeOrigin::Object(id) = change.origin {
370 for diff in change.diffs {
371 self.accumulator.track(id, diff);
372 }
373 }
374 }
375
376 #[inline]
377 pub fn version(&self) -> CommitVersion {
378 self.cmd.as_ref().unwrap().version()
379 }
380
381 #[inline]
382 pub fn id(&self) -> TransactionId {
383 self.cmd.as_ref().unwrap().id()
384 }
385
386 #[inline]
387 pub fn get<K: Into<TaggedKey> + Clone>(&mut self, key: &K) -> Result<Option<MultiVersionRow<TaggedKey>>> {
388 self.check_active()?;
389 Ok(self.cmd.as_mut().unwrap().get(key)?.map(|v| v.into_multi_version_row()))
390 }
391
392 #[inline]
393 pub fn get_committed<K: Into<TaggedKey> + Clone>(
394 &mut self,
395 key: &K,
396 ) -> Result<Option<MultiVersionRow<TaggedKey>>> {
397 self.check_active()?;
398 Ok(self.cmd.as_mut().unwrap().get_committed(key)?.map(|v| v.into_multi_version_row()))
399 }
400
401 #[inline]
402 pub fn contains<K: Into<TaggedKey> + Clone>(&mut self, key: &K) -> Result<bool> {
403 self.check_active()?;
404 self.cmd.as_mut().unwrap().contains(key)
405 }
406
407 #[inline]
408 pub fn prefix(&mut self, prefix: &EncodedKey) -> Result<MultiVersionBatch<TaggedKey>> {
409 self.check_active()?;
410 self.cmd.as_mut().unwrap().prefix(prefix)
411 }
412
413 #[inline]
414 pub fn prefix_rev(&mut self, prefix: &EncodedKey) -> Result<MultiVersionBatch<TaggedKey>> {
415 self.check_active()?;
416 self.cmd.as_mut().unwrap().prefix_rev(prefix)
417 }
418
419 #[inline]
420 pub fn read_as_of_version_exclusive(&mut self, version: CommitVersion) -> Result<()> {
421 self.check_active()?;
422 self.cmd.as_mut().unwrap().read_as_of_version_exclusive(version);
423 Ok(())
424 }
425
426 pub fn stamp_source(&mut self, source: SourceVersion) -> Result<()> {
427 self.check_active()?;
428 self.cmd.as_mut().unwrap().stamp_source(source);
429 Ok(())
430 }
431
432 #[inline]
433 pub fn set<K: Into<TaggedKey> + Clone>(&mut self, key: &K, bytes: impl Into<EncodedBytes>) -> Result<()> {
434 self.check_active()?;
435 self.cmd.as_mut().unwrap().set(key, bytes.into())
436 }
437
438 #[inline]
439 pub fn reserve_writes(&mut self, additional: usize) -> Result<()> {
440 self.check_active()?;
441 self.cmd.as_mut().unwrap().reserve_writes(additional);
442 Ok(())
443 }
444
445 #[inline]
446 pub fn disable_conflict_tracking(&mut self) -> Result<()> {
447 self.check_active()?;
448 self.cmd.as_mut().unwrap().disable_conflict_tracking();
449 Ok(())
450 }
451
452 #[inline]
453 pub fn remove_with_pre<K: Into<TaggedKey> + Clone>(&mut self, key: &K, pre: EncodedBytes) -> Result<()> {
454 self.check_active()?;
455 self.cmd.as_mut().unwrap().remove_with_pre(key, pre)
456 }
457
458 #[inline]
459 pub fn remove<K: Into<TaggedKey> + Clone>(&mut self, key: &K) -> Result<()> {
460 self.check_active()?;
461 self.cmd.as_mut().unwrap().remove(key)
462 }
463
464 #[inline]
465 pub fn remove_unobserved<K: Into<TaggedKey> + Clone>(&mut self, key: &K) -> Result<()> {
466 self.check_active()?;
467 self.cmd.as_mut().unwrap().remove_unobserved(key)
468 }
469
470 #[inline]
471 pub fn remove_unobserved_with_pre<K: Into<TaggedKey> + Clone>(
472 &mut self,
473 key: &K,
474 pre: EncodedBytes,
475 ) -> Result<()> {
476 self.check_active()?;
477 self.cmd.as_mut().unwrap().remove_unobserved_with_pre(key, pre)
478 }
479
480 #[inline]
481 pub fn remove_silent<K: Into<TaggedKey> + Clone>(&mut self, key: &K) -> Result<()> {
482 self.check_active()?;
483 self.cmd.as_mut().unwrap().remove_silent(key)
484 }
485
486 #[inline]
487 pub fn mark_preexisting<K: Into<TaggedKey> + Clone>(&mut self, key: &K) -> Result<()> {
488 self.check_active()?;
489 self.cmd.as_mut().unwrap().mark_preexisting(key);
490 Ok(())
491 }
492
493 #[inline]
494 pub fn range(
495 &mut self,
496 range: TaggedKeyBoundRange,
497 scope: RangeScope,
498 batch_size: usize,
499 ) -> Result<Box<dyn Iterator<Item = Result<MultiVersionRow<TaggedKey>>> + Send + '_>> {
500 self.check_active()?;
501 Ok(self.cmd.as_mut().unwrap().range(range, scope, batch_size))
502 }
503
504 pub fn range_row(
505 &mut self,
506 storage: StorageId,
507 start: Bound<StorageRowKey>,
508 end: Bound<StorageRowKey>,
509 scope: RangeScope,
510 batch_size: usize,
511 ) -> Result<Box<dyn Iterator<Item = Result<MultiVersionRow<StorageRowKey>>> + Send + '_>> {
512 self.check_active()?;
513 Ok(self.cmd.as_mut().unwrap().range_row(storage, start, end, scope, batch_size))
514 }
515
516 #[inline]
517 pub fn range_partitioned_row(
518 &mut self,
519 storage: StorageId,
520 start: Bound<StoragePartitionedRowKey>,
521 end: Bound<StoragePartitionedRowKey>,
522 scope: RangeScope,
523 batch_size: usize,
524 ) -> Result<Box<dyn Iterator<Item = Result<MultiVersionRow<StoragePartitionedRowKey>>> + Send + '_>> {
525 self.check_active()?;
526 Ok(self.cmd.as_mut().unwrap().range_partitioned_row(storage, start, end, scope, batch_size))
527 }
528
529 #[inline]
530 pub fn range_persistence(
531 &mut self,
532 range: TaggedKeyBoundRange,
533 scope: RangeScope,
534 batch_size: usize,
535 ) -> Result<Box<dyn Iterator<Item = Result<MultiVersionRow<TaggedKey>>> + Send + '_>> {
536 self.check_active()?;
537 Ok(self.cmd.as_mut().unwrap().range_persistence(range, scope, batch_size))
538 }
539
540 #[inline]
541 pub fn range_rev(
542 &mut self,
543 range: TaggedKeyBoundRange,
544 scope: RangeScope,
545 batch_size: usize,
546 ) -> Result<Box<dyn Iterator<Item = Result<MultiVersionRow<TaggedKey>>> + Send + '_>> {
547 self.check_active()?;
548 Ok(self.cmd.as_mut().unwrap().range_rev(range, scope, batch_size))
549 }
550
551 #[inline]
552 pub fn range_rev_persistence(
553 &mut self,
554 range: TaggedKeyBoundRange,
555 scope: RangeScope,
556 batch_size: usize,
557 ) -> Result<Box<dyn Iterator<Item = Result<MultiVersionRow<TaggedKey>>> + Send + '_>> {
558 self.check_active()?;
559 Ok(self.cmd.as_mut().unwrap().range_rev_persistence(range, scope, batch_size))
560 }
561}
562
563impl WithEventBus for CommandTransaction {
564 fn event_bus(&self) -> &EventBus {
565 &self.event_bus
566 }
567}
568
569impl Write for CommandTransaction {
570 #[inline]
571 fn set(&mut self, key: &TaggedKey, bytes: EncodedBytes) -> Result<()> {
572 CommandTransaction::set(self, key, bytes)
573 }
574 #[inline]
575 fn remove_with_pre(&mut self, key: &TaggedKey, pre: EncodedBytes) -> Result<()> {
576 CommandTransaction::remove_with_pre(self, key, pre)
577 }
578 #[inline]
579 fn remove(&mut self, key: &TaggedKey) -> Result<()> {
580 CommandTransaction::remove(self, key)
581 }
582 #[inline]
583 fn mark_preexisting(&mut self, key: &TaggedKey) -> Result<()> {
584 CommandTransaction::mark_preexisting(self, key)
585 }
586 #[inline]
587 fn track_row_change(&mut self, changes: &[RowChange]) {
588 CommandTransaction::track_row_change(self, changes)
589 }
590 #[inline]
591 fn track_flow_change(&mut self, change: Change) {
592 CommandTransaction::track_flow_change(self, change)
593 }
594}
595
596impl WithInterceptors for CommandTransaction {
597 fn table_row_pre_insert_interceptors(&mut self) -> &mut Chain<dyn TableRowPreInsertInterceptor + Send + Sync> {
598 &mut self.interceptors.table_row_pre_insert
599 }
600
601 fn table_row_post_insert_interceptors(
602 &mut self,
603 ) -> &mut Chain<dyn TableRowPostInsertInterceptor + Send + Sync> {
604 &mut self.interceptors.table_row_post_insert
605 }
606
607 fn table_row_pre_update_interceptors(&mut self) -> &mut Chain<dyn TableRowPreUpdateInterceptor + Send + Sync> {
608 &mut self.interceptors.table_row_pre_update
609 }
610
611 fn table_row_post_update_interceptors(
612 &mut self,
613 ) -> &mut Chain<dyn TableRowPostUpdateInterceptor + Send + Sync> {
614 &mut self.interceptors.table_row_post_update
615 }
616
617 fn table_row_pre_delete_interceptors(&mut self) -> &mut Chain<dyn TableRowPreDeleteInterceptor + Send + Sync> {
618 &mut self.interceptors.table_row_pre_delete
619 }
620
621 fn table_row_post_delete_interceptors(
622 &mut self,
623 ) -> &mut Chain<dyn TableRowPostDeleteInterceptor + Send + Sync> {
624 &mut self.interceptors.table_row_post_delete
625 }
626
627 fn ringbuffer_row_pre_insert_interceptors(
628 &mut self,
629 ) -> &mut Chain<dyn RingBufferRowPreInsertInterceptor + Send + Sync> {
630 &mut self.interceptors.ringbuffer_row_pre_insert
631 }
632
633 fn ringbuffer_row_post_insert_interceptors(
634 &mut self,
635 ) -> &mut Chain<dyn RingBufferRowPostInsertInterceptor + Send + Sync> {
636 &mut self.interceptors.ringbuffer_row_post_insert
637 }
638
639 fn ringbuffer_row_pre_update_interceptors(
640 &mut self,
641 ) -> &mut Chain<dyn RingBufferRowPreUpdateInterceptor + Send + Sync> {
642 &mut self.interceptors.ringbuffer_row_pre_update
643 }
644
645 fn ringbuffer_row_post_update_interceptors(
646 &mut self,
647 ) -> &mut Chain<dyn RingBufferRowPostUpdateInterceptor + Send + Sync> {
648 &mut self.interceptors.ringbuffer_row_post_update
649 }
650
651 fn ringbuffer_row_pre_delete_interceptors(
652 &mut self,
653 ) -> &mut Chain<dyn RingBufferRowPreDeleteInterceptor + Send + Sync> {
654 &mut self.interceptors.ringbuffer_row_pre_delete
655 }
656
657 fn ringbuffer_row_post_delete_interceptors(
658 &mut self,
659 ) -> &mut Chain<dyn RingBufferRowPostDeleteInterceptor + Send + Sync> {
660 &mut self.interceptors.ringbuffer_row_post_delete
661 }
662
663 fn pre_commit_interceptors(&mut self) -> &mut Chain<dyn PreCommitInterceptor + Send + Sync> {
664 &mut self.interceptors.pre_commit
665 }
666
667 fn post_commit_interceptors(&mut self) -> &mut Chain<dyn PostCommitInterceptor + Send + Sync> {
668 &mut self.interceptors.post_commit
669 }
670
671 fn namespace_post_create_interceptors(
672 &mut self,
673 ) -> &mut Chain<dyn NamespacePostCreateInterceptor + Send + Sync> {
674 &mut self.interceptors.namespace_post_create
675 }
676
677 fn namespace_pre_update_interceptors(&mut self) -> &mut Chain<dyn NamespacePreUpdateInterceptor + Send + Sync> {
678 &mut self.interceptors.namespace_pre_update
679 }
680
681 fn namespace_post_update_interceptors(
682 &mut self,
683 ) -> &mut Chain<dyn NamespacePostUpdateInterceptor + Send + Sync> {
684 &mut self.interceptors.namespace_post_update
685 }
686
687 fn namespace_pre_delete_interceptors(&mut self) -> &mut Chain<dyn NamespacePreDeleteInterceptor + Send + Sync> {
688 &mut self.interceptors.namespace_pre_delete
689 }
690
691 fn table_post_create_interceptors(&mut self) -> &mut Chain<dyn TablePostCreateInterceptor + Send + Sync> {
692 &mut self.interceptors.table_post_create
693 }
694
695 fn table_pre_update_interceptors(&mut self) -> &mut Chain<dyn TablePreUpdateInterceptor + Send + Sync> {
696 &mut self.interceptors.table_pre_update
697 }
698
699 fn table_post_update_interceptors(&mut self) -> &mut Chain<dyn TablePostUpdateInterceptor + Send + Sync> {
700 &mut self.interceptors.table_post_update
701 }
702
703 fn table_pre_delete_interceptors(&mut self) -> &mut Chain<dyn TablePreDeleteInterceptor + Send + Sync> {
704 &mut self.interceptors.table_pre_delete
705 }
706
707 fn view_post_create_interceptors(&mut self) -> &mut Chain<dyn ViewPostCreateInterceptor + Send + Sync> {
708 &mut self.interceptors.view_post_create
709 }
710
711 fn view_pre_update_interceptors(&mut self) -> &mut Chain<dyn ViewPreUpdateInterceptor + Send + Sync> {
712 &mut self.interceptors.view_pre_update
713 }
714
715 fn view_post_update_interceptors(&mut self) -> &mut Chain<dyn ViewPostUpdateInterceptor + Send + Sync> {
716 &mut self.interceptors.view_post_update
717 }
718
719 fn view_pre_delete_interceptors(&mut self) -> &mut Chain<dyn ViewPreDeleteInterceptor + Send + Sync> {
720 &mut self.interceptors.view_pre_delete
721 }
722
723 fn ringbuffer_post_create_interceptors(
724 &mut self,
725 ) -> &mut Chain<dyn RingBufferPostCreateInterceptor + Send + Sync> {
726 &mut self.interceptors.ringbuffer_post_create
727 }
728
729 fn ringbuffer_pre_update_interceptors(
730 &mut self,
731 ) -> &mut Chain<dyn RingBufferPreUpdateInterceptor + Send + Sync> {
732 &mut self.interceptors.ringbuffer_pre_update
733 }
734
735 fn ringbuffer_post_update_interceptors(
736 &mut self,
737 ) -> &mut Chain<dyn RingBufferPostUpdateInterceptor + Send + Sync> {
738 &mut self.interceptors.ringbuffer_post_update
739 }
740
741 fn ringbuffer_pre_delete_interceptors(
742 &mut self,
743 ) -> &mut Chain<dyn RingBufferPreDeleteInterceptor + Send + Sync> {
744 &mut self.interceptors.ringbuffer_pre_delete
745 }
746
747 fn dictionary_row_pre_insert_interceptors(
748 &mut self,
749 ) -> &mut Chain<dyn DictionaryRowPreInsertInterceptor + Send + Sync> {
750 &mut self.interceptors.dictionary_row_pre_insert
751 }
752
753 fn dictionary_row_post_insert_interceptors(
754 &mut self,
755 ) -> &mut Chain<dyn DictionaryRowPostInsertInterceptor + Send + Sync> {
756 &mut self.interceptors.dictionary_row_post_insert
757 }
758
759 fn dictionary_row_pre_update_interceptors(
760 &mut self,
761 ) -> &mut Chain<dyn DictionaryRowPreUpdateInterceptor + Send + Sync> {
762 &mut self.interceptors.dictionary_row_pre_update
763 }
764
765 fn dictionary_row_post_update_interceptors(
766 &mut self,
767 ) -> &mut Chain<dyn DictionaryRowPostUpdateInterceptor + Send + Sync> {
768 &mut self.interceptors.dictionary_row_post_update
769 }
770
771 fn dictionary_row_pre_delete_interceptors(
772 &mut self,
773 ) -> &mut Chain<dyn DictionaryRowPreDeleteInterceptor + Send + Sync> {
774 &mut self.interceptors.dictionary_row_pre_delete
775 }
776
777 fn dictionary_row_post_delete_interceptors(
778 &mut self,
779 ) -> &mut Chain<dyn DictionaryRowPostDeleteInterceptor + Send + Sync> {
780 &mut self.interceptors.dictionary_row_post_delete
781 }
782
783 fn dictionary_post_create_interceptors(
784 &mut self,
785 ) -> &mut Chain<dyn DictionaryPostCreateInterceptor + Send + Sync> {
786 &mut self.interceptors.dictionary_post_create
787 }
788
789 fn dictionary_pre_update_interceptors(
790 &mut self,
791 ) -> &mut Chain<dyn DictionaryPreUpdateInterceptor + Send + Sync> {
792 &mut self.interceptors.dictionary_pre_update
793 }
794
795 fn dictionary_post_update_interceptors(
796 &mut self,
797 ) -> &mut Chain<dyn DictionaryPostUpdateInterceptor + Send + Sync> {
798 &mut self.interceptors.dictionary_post_update
799 }
800
801 fn dictionary_pre_delete_interceptors(
802 &mut self,
803 ) -> &mut Chain<dyn DictionaryPreDeleteInterceptor + Send + Sync> {
804 &mut self.interceptors.dictionary_pre_delete
805 }
806
807 fn series_row_pre_insert_interceptors(
808 &mut self,
809 ) -> &mut Chain<dyn SeriesRowPreInsertInterceptor + Send + Sync> {
810 &mut self.interceptors.series_row_pre_insert
811 }
812
813 fn series_row_post_insert_interceptors(
814 &mut self,
815 ) -> &mut Chain<dyn SeriesRowPostInsertInterceptor + Send + Sync> {
816 &mut self.interceptors.series_row_post_insert
817 }
818
819 fn series_row_pre_update_interceptors(
820 &mut self,
821 ) -> &mut Chain<dyn SeriesRowPreUpdateInterceptor + Send + Sync> {
822 &mut self.interceptors.series_row_pre_update
823 }
824
825 fn series_row_post_update_interceptors(
826 &mut self,
827 ) -> &mut Chain<dyn SeriesRowPostUpdateInterceptor + Send + Sync> {
828 &mut self.interceptors.series_row_post_update
829 }
830
831 fn series_row_pre_delete_interceptors(
832 &mut self,
833 ) -> &mut Chain<dyn SeriesRowPreDeleteInterceptor + Send + Sync> {
834 &mut self.interceptors.series_row_pre_delete
835 }
836
837 fn series_row_post_delete_interceptors(
838 &mut self,
839 ) -> &mut Chain<dyn SeriesRowPostDeleteInterceptor + Send + Sync> {
840 &mut self.interceptors.series_row_post_delete
841 }
842
843 fn series_post_create_interceptors(&mut self) -> &mut Chain<dyn SeriesPostCreateInterceptor + Send + Sync> {
844 &mut self.interceptors.series_post_create
845 }
846
847 fn series_pre_update_interceptors(&mut self) -> &mut Chain<dyn SeriesPreUpdateInterceptor + Send + Sync> {
848 &mut self.interceptors.series_pre_update
849 }
850
851 fn series_post_update_interceptors(&mut self) -> &mut Chain<dyn SeriesPostUpdateInterceptor + Send + Sync> {
852 &mut self.interceptors.series_post_update
853 }
854
855 fn series_pre_delete_interceptors(&mut self) -> &mut Chain<dyn SeriesPreDeleteInterceptor + Send + Sync> {
856 &mut self.interceptors.series_pre_delete
857 }
858
859 fn identity_post_create_interceptors(&mut self) -> &mut Chain<dyn IdentityPostCreateInterceptor + Send + Sync> {
860 &mut self.interceptors.identity_post_create
861 }
862
863 fn identity_pre_delete_interceptors(&mut self) -> &mut Chain<dyn IdentityPreDeleteInterceptor + Send + Sync> {
864 &mut self.interceptors.identity_pre_delete
865 }
866
867 fn role_post_create_interceptors(&mut self) -> &mut Chain<dyn RolePostCreateInterceptor + Send + Sync> {
868 &mut self.interceptors.role_post_create
869 }
870
871 fn role_pre_delete_interceptors(&mut self) -> &mut Chain<dyn RolePreDeleteInterceptor + Send + Sync> {
872 &mut self.interceptors.role_pre_delete
873 }
874
875 fn granted_role_post_create_interceptors(
876 &mut self,
877 ) -> &mut Chain<dyn GrantedRolePostCreateInterceptor + Send + Sync> {
878 &mut self.interceptors.granted_role_post_create
879 }
880
881 fn granted_role_pre_delete_interceptors(
882 &mut self,
883 ) -> &mut Chain<dyn GrantedRolePreDeleteInterceptor + Send + Sync> {
884 &mut self.interceptors.granted_role_pre_delete
885 }
886
887 fn identity_attribute_post_create_interceptors(
888 &mut self,
889 ) -> &mut Chain<dyn IdentityAttributePostCreateInterceptor + Send + Sync> {
890 &mut self.interceptors.identity_attribute_post_create
891 }
892
893 fn identity_attribute_pre_delete_interceptors(
894 &mut self,
895 ) -> &mut Chain<dyn IdentityAttributePreDeleteInterceptor + Send + Sync> {
896 &mut self.interceptors.identity_attribute_pre_delete
897 }
898
899 fn identity_attribute_value_post_create_interceptors(
900 &mut self,
901 ) -> &mut Chain<dyn IdentityAttributeValuePostCreateInterceptor + Send + Sync> {
902 &mut self.interceptors.identity_attribute_value_post_create
903 }
904
905 fn identity_attribute_value_pre_delete_interceptors(
906 &mut self,
907 ) -> &mut Chain<dyn IdentityAttributeValuePreDeleteInterceptor + Send + Sync> {
908 &mut self.interceptors.identity_attribute_value_pre_delete
909 }
910
911 fn authentication_post_create_interceptors(
912 &mut self,
913 ) -> &mut Chain<dyn AuthenticationPostCreateInterceptor + Send + Sync> {
914 &mut self.interceptors.authentication_post_create
915 }
916
917 fn authentication_pre_delete_interceptors(
918 &mut self,
919 ) -> &mut Chain<dyn AuthenticationPreDeleteInterceptor + Send + Sync> {
920 &mut self.interceptors.authentication_pre_delete
921 }
922}
923
924impl Drop for CommandTransaction {
925 fn drop(&mut self) {
926 if let Some(mut multi) = self.cmd.take()
927 && (self.state == TransactionState::Active || self.state == TransactionState::Poisoned)
928 {
929 let _ = multi.rollback();
930 }
931 }
932}