ruststream_sqlx_dialect/row_lock.rs
1//! The row lock form: a claim that locks its rows for the transaction its handler settles in.
2
3use crate::dialect::Dialect;
4use crate::spec::TableSpec;
5use crate::statement::{ClaimShape, Statement, StatementError};
6
7/// The row lock form of a dialect: the claim that locks its rows until its transaction ends.
8///
9/// A table that declares neither `#[field(locked_until)]` nor `advisory_lock` takes its rows
10/// this way. The claim selects the claimable rows under a lock, the handler runs while the
11/// transaction holds them, and the settlement ends the transaction; after a crash the database
12/// rolls back and the rows return at once. The transaction opens at the table's isolation level,
13/// with the statement [`Dialect::begin`] gives. A dialect implements this trait where its
14/// database locks rows for a transaction, and a table in the row lock form does not compile
15/// against a dialect that does not. [`Postgres`](crate::Postgres) and [`MySql`](crate::MySql)
16/// implement it; [`Sqlite`](crate::Sqlite) does not, as its writer locks the whole database.
17///
18/// # Examples
19///
20/// ```
21/// # use std::num::NonZeroUsize;
22/// # use ruststream_sqlx_dialect::TableName;
23/// use ruststream_sqlx_dialect::{
24/// ClaimShape, Dialect, Param, RowLock, Statement, StatementError, TableSpec,
25/// };
26///
27/// /// SQL Server, a database without a built-in dialect.
28/// #[derive(Debug)]
29/// pub struct Mssql;
30///
31/// // A table without `locked_until` or `advisory_lock` mounts on `Mssql`. The claim holds an
32/// // update lock on each row it takes until its transaction ends, and skips the rows another claim
33/// // holds.
34/// impl RowLock for Mssql {
35/// fn lock_claim(
36/// &self,
37/// _spec: &TableSpec<'_>,
38/// shape: ClaimShape,
39/// ) -> Result<Statement, StatementError> {
40/// let selected = match shape {
41/// ClaimShape::Rows => "[job_id], [attempt], [payload]",
42/// ClaimShape::Ids => "[job_id]",
43/// ClaimShape::Roles => {
44/// "[job_id] AS [id], [attempt] AS [attempt], [payload] AS [payload]"
45/// }
46/// };
47/// Ok(Statement::new(
48/// format!(
49/// "SELECT TOP (@p1) {selected} FROM [email_jobs] \
50/// WITH (UPDLOCK, READPAST, ROWLOCK) ORDER BY [job_id]"
51/// ),
52/// [Param::Limit],
53/// ))
54/// }
55/// }
56/// # impl Dialect for Mssql {
57/// # fn name(&self) -> &'static str { "mssql" }
58/// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
59/// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
60/// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
61/// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
62/// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
63/// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
64/// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
65/// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
66/// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
67/// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
68/// # }
69/// ```
70#[diagnostic::on_unimplemented(
71 message = "the `{Self}` dialect builds no row lock claim, so a table on it cannot take its \
72 rows by row lock",
73 label = "a table without `locked_until` or `advisory_lock` takes its rows by row lock",
74 note = "declare `#[field(locked_until)]` (the lease form) or `#[inbox(advisory_lock = \"..\")]` \
75 (the advisory lock form) on the struct",
76 note = "a dialect of the service's own serves the row lock form by implementing `RowLock`, \
77 where its database locks rows"
78)]
79pub trait RowLock: Dialect {
80 /// The statement that claims up to [`Param::Limit`](crate::Param::Limit) rows of the
81 /// subscription's group, in claim order, and locks them until the claim's transaction ends.
82 ///
83 /// A table with FIFO groups keeps one row of a group in work. Its claim takes the group's
84 /// head, the first unfinished row of the group in claim order, and takes nothing while the
85 /// head is in work (another claim's transaction holds it) or not yet due. It binds no
86 /// [`Param::Limit`](crate::Param::Limit). The claim's transaction first takes the group with
87 /// the statement [`fifo_guard`](Dialect::fifo_guard) gives and holds it until the delivery
88 /// settles, so a row that enters the group ahead of the head in work waits too.
89 ///
90 /// # Errors
91 ///
92 /// [`StatementError::FormMismatch`] for a table in another form. A dialect of the service's
93 /// own that builds no claim for FIFO groups returns [`StatementError::UnsupportedFifo`] for a
94 /// table with them.
95 ///
96 /// # Examples
97 ///
98 /// ```
99 /// # use std::num::NonZeroUsize;
100 /// # use ruststream_sqlx_dialect::TableName;
101 /// use ruststream_sqlx_dialect::{
102 /// ClaimShape, Dialect, Param, RowLock, Statement, StatementError, TableSpec,
103 /// };
104 ///
105 /// /// SQL Server, a database without a built-in dialect.
106 /// #[derive(Debug)]
107 /// pub struct Mssql;
108 ///
109 /// impl RowLock for Mssql {
110 /// // The due rows of the subscription's group in `email_jobs`, in claim order. `UPDLOCK`
111 /// // keeps two claims apart until the transaction ends, `READPAST` skips the rows another
112 /// // claim holds, and `ROWLOCK` keeps the lock to the rows themselves.
113 /// fn lock_claim(
114 /// &self,
115 /// spec: &TableSpec<'_>,
116 /// _shape: ClaimShape,
117 /// ) -> Result<Statement, StatementError> {
118 /// // This claim keeps no group in order.
119 /// if spec.is_fifo() {
120 /// return Err(StatementError::UnsupportedFifo { dialect: self.name() });
121 /// }
122 /// Ok(Statement::new(
123 /// "SELECT TOP (@p3) [job_id], [attempt], [payload] FROM [email_jobs] \
124 /// WITH (UPDLOCK, READPAST, ROWLOCK) \
125 /// WHERE [name] = @p1 AND [retry_after] <= @p2 \
126 /// ORDER BY [priority], [retry_after], [job_id]",
127 /// [Param::Group, Param::Now, Param::Limit],
128 /// ))
129 /// }
130 /// }
131 /// # impl Dialect for Mssql {
132 /// # fn name(&self) -> &'static str { "mssql" }
133 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
134 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
135 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
136 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
137 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
138 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
139 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
140 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
141 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
142 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
143 /// # }
144 /// ```
145 fn lock_claim(
146 &self,
147 spec: &TableSpec<'_>,
148 shape: ClaimShape,
149 ) -> Result<Statement, StatementError>;
150}