Skip to main content

reifydb_transaction/transaction/
command.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use 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}