ruststream_sqlx_dialect/lease.rs
1//! The lease form: a claim that writes a lease into each row it takes and commits at once.
2
3use crate::dialect::Dialect;
4use crate::spec::TableSpec;
5use crate::statement::{ClaimShape, Statement, StatementError};
6
7/// The lease form of a dialect: the claim that writes a lease into each row it takes, the
8/// extension of a lease in work, and the stamp that leases a row a claim only selected.
9///
10/// A table with a `#[field(locked_until)]` field takes its rows this way. A claim writes the
11/// lease's expiry ([`Param::Lease`](crate::Param::Lease)) into each row it takes, counts the
12/// attempt and commits, so the handler runs outside any transaction. It skips every row whose
13/// lease has not ended by [`Param::LeaseNow`](crate::Param::LeaseNow). The expiry the claim wrote
14/// is the delivery's ownership token ([`Param::Held`](crate::Param::Held)): the extension and
15/// every settlement name the row and that token. A dialect implements this trait where it builds
16/// these statements, and a lease table does not compile against a dialect that does not. Every
17/// built-in dialect implements it.
18///
19/// Beside its statements the trait tells whether the claim writes the lease itself
20/// ([`claim_writes_lease`](Self::claim_writes_lease)), whether the rows it returns carry the
21/// claim's count of the attempt ([`claim_counts_attempt`](Self::claim_counts_attempt)), and how
22/// the transaction of a claim that only selects its rows opens
23/// ([`begin_lease_claim`](Self::begin_lease_claim)).
24///
25/// # Examples
26///
27/// A dialect for a database without a built-in one builds the lease form itself. SQL Server takes
28/// the claimable rows, writes their lease and counts the attempt in one update, and returns the
29/// rows as they were before:
30///
31/// ```
32/// # use std::num::NonZeroUsize;
33/// # use ruststream_sqlx_dialect::TableName;
34/// use ruststream_sqlx_dialect::{
35/// ClaimShape, Dialect, Lease, Param, Statement, StatementError, TableSpec,
36/// };
37///
38/// /// SQL Server, a database without a built-in dialect.
39/// #[derive(Debug)]
40/// pub struct Mssql;
41///
42/// // The service's one lease table, `email_jobs`, mounts on `Mssql`.
43/// impl Lease for Mssql {
44/// fn lease_claim(
45/// &self,
46/// _spec: &TableSpec<'_>,
47/// shape: ClaimShape,
48/// ) -> Result<Statement, StatementError> {
49/// let returned = match shape {
50/// ClaimShape::Rows => {
51/// "deleted.[job_id], deleted.[attempt], deleted.[locked_until], \\
52/// deleted.[payload]"
53/// }
54/// ClaimShape::Ids => "deleted.[job_id]",
55/// ClaimShape::Roles => {
56/// "deleted.[job_id] AS [id], deleted.[attempt] AS [attempt], \
57/// deleted.[payload] AS [payload]"
58/// }
59/// };
60/// Ok(Statement::new(
61/// format!(
62/// "WITH [claimed] AS (SELECT TOP (@p3) * FROM [email_jobs] \
63/// WITH (UPDLOCK, READPAST, ROWLOCK) \
64/// WHERE [locked_until] IS NULL OR [locked_until] <= @p2 ORDER BY [job_id]) \
65/// UPDATE [claimed] SET [locked_until] = @p1, [attempt] = [attempt] + 1 \
66/// OUTPUT {returned}"
67/// ),
68/// [Param::Lease, Param::LeaseNow, Param::Limit],
69/// ))
70/// }
71///
72/// fn extend(&self, _spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
73/// Ok(Statement::new(
74/// "UPDATE [email_jobs] SET [locked_until] = @p1 \
75/// WHERE [job_id] = @p2 AND [locked_until] = @p3",
76/// [Param::Lease, Param::Id, Param::Held],
77/// ))
78/// }
79///
80/// fn stamp(&self, _spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
81/// Ok(Statement::new(
82/// "UPDATE [email_jobs] SET [locked_until] = @p1, [attempt] = [attempt] + 1 \
83/// WHERE [job_id] = @p2 AND ([locked_until] IS NULL OR [locked_until] <= @p3)",
84/// [Param::Lease, Param::Id, Param::LeaseNow],
85/// ))
86/// }
87///
88/// // A claim of the service's own (`custom(claim)`) selects its rows, then stamps them, in
89/// // the transaction this opens. SQL Server starts one with `BEGIN TRANSACTION`.
90/// fn begin_lease_claim(&self) -> Option<&'static str> {
91/// Some("SET TRANSACTION ISOLATION LEVEL READ COMMITTED; BEGIN TRANSACTION")
92/// }
93/// }
94/// # impl Dialect for Mssql {
95/// # fn name(&self) -> &'static str { "mssql" }
96/// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
97/// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
98/// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
99/// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
100/// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
101/// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
102/// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
103/// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
104/// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
105/// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
106/// # }
107/// ```
108#[diagnostic::on_unimplemented(
109 message = "the `{Self}` dialect builds no lease claim, so a table on it cannot take its rows \
110 by lease",
111 label = "a table with `locked_until` takes its rows by lease",
112 note = "a dialect of the service's own takes the lease form by implementing `Lease`: the \
113 claim, the extension and the stamp"
114)]
115pub trait Lease: Dialect {
116 /// The statement that claims up to [`Param::Limit`](crate::Param::Limit) rows of the
117 /// subscription's group, in claim order, among the rows whose lease ended by
118 /// [`Param::LeaseNow`](crate::Param::LeaseNow).
119 ///
120 /// Where [`claim_writes_lease`](Self::claim_writes_lease) answers `true`, it also writes the
121 /// new lease ([`Param::Lease`](crate::Param::Lease)), counts the attempt, and returns the rows
122 /// as they were before.
123 ///
124 /// A table with FIFO groups keeps one row of a group in work. Its claim takes the group's
125 /// head, the first unfinished row of the group in claim order, and takes nothing while a row
126 /// of the group holds a lease or the head is not yet due, so a row that enters the group ahead
127 /// of the head in work waits too. It binds no [`Param::Limit`](crate::Param::Limit). It runs
128 /// in a transaction that first takes the group with the statement
129 /// [`fifo_guard`](Dialect::fifo_guard) gives, which keeps two claims apart until one of them
130 /// commits its lease.
131 ///
132 /// # Errors
133 ///
134 /// [`StatementError::FormMismatch`] for a table in another form;
135 /// [`StatementError::LeaseOnDatabaseClock`] for a lease table on the database's clock. A
136 /// dialect of the service's own that builds no claim for FIFO groups returns
137 /// [`StatementError::UnsupportedFifo`] for a table with them.
138 ///
139 /// # Examples
140 ///
141 /// ```
142 /// # use std::num::NonZeroUsize;
143 /// # use ruststream_sqlx_dialect::TableName;
144 /// use ruststream_sqlx_dialect::{
145 /// ClaimShape, Dialect, Lease, Param, Statement, StatementError, TableSpec,
146 /// };
147 ///
148 /// /// SQL Server, a database without a built-in dialect.
149 /// #[derive(Debug)]
150 /// pub struct Mssql;
151 ///
152 /// impl Lease for Mssql {
153 /// // One update takes the rows of the service's one lease table, `email_jobs`, in id
154 /// // order, skipping the rows another claim holds. It writes their lease, counts the
155 /// // attempt and returns them as they were before (`deleted`).
156 /// fn lease_claim(
157 /// &self,
158 /// _spec: &TableSpec<'_>,
159 /// shape: ClaimShape,
160 /// ) -> Result<Statement, StatementError> {
161 /// let returned = match shape {
162 /// ClaimShape::Rows => {
163 /// "deleted.[job_id], deleted.[attempt], deleted.[locked_until], \\
164 /// deleted.[payload]"
165 /// }
166 /// ClaimShape::Ids => "deleted.[job_id]",
167 /// ClaimShape::Roles => {
168 /// "deleted.[job_id] AS [id], deleted.[attempt] AS [attempt], \
169 /// deleted.[payload] AS [payload]"
170 /// }
171 /// };
172 /// Ok(Statement::new(
173 /// format!(
174 /// "WITH [claimed] AS (SELECT TOP (@p3) * FROM [email_jobs] \
175 /// WITH (UPDLOCK, READPAST, ROWLOCK) \
176 /// WHERE [locked_until] IS NULL OR [locked_until] <= @p2 ORDER BY [job_id]) \
177 /// UPDATE [claimed] SET [locked_until] = @p1, [attempt] = [attempt] + 1 \
178 /// OUTPUT {returned}"
179 /// ),
180 /// [Param::Lease, Param::LeaseNow, Param::Limit],
181 /// ))
182 /// }
183 /// # fn extend(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
184 /// # fn stamp(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
185 /// }
186 /// # impl Dialect for Mssql {
187 /// # fn name(&self) -> &'static str { "mssql" }
188 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
189 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
190 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
191 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
192 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
193 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
194 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
195 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
196 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
197 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
198 /// # }
199 /// ```
200 fn lease_claim(
201 &self,
202 spec: &TableSpec<'_>,
203 shape: ClaimShape,
204 ) -> Result<Statement, StatementError>;
205
206 /// The statement that extends a delivery's lease: it writes the new expiry
207 /// ([`Param::Lease`](crate::Param::Lease)) while the row still holds the delivery's token
208 /// ([`Param::Held`](crate::Param::Held)), so a handler that runs longer than one lease keeps
209 /// its row.
210 ///
211 /// # Errors
212 ///
213 /// [`StatementError::FormMismatch`] for a table in another form;
214 /// [`StatementError::LeaseOnDatabaseClock`] for a lease table on the database's clock.
215 ///
216 /// # Examples
217 ///
218 /// ```
219 /// # use std::num::NonZeroUsize;
220 /// # use ruststream_sqlx_dialect::TableName;
221 /// use ruststream_sqlx_dialect::{
222 /// ClaimShape, Dialect, Lease, Param, Statement, StatementError, TableSpec,
223 /// };
224 ///
225 /// /// SQL Server, a database without a built-in dialect.
226 /// #[derive(Debug)]
227 /// pub struct Mssql;
228 ///
229 /// impl Lease for Mssql {
230 /// // An extension of an email's lease also records when its handler was last seen. It
231 /// // changes the row only while the row holds the delivery's lease.
232 /// fn extend(&self, _spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
233 /// Ok(Statement::new(
234 /// "UPDATE [email_jobs] SET [locked_until] = @p1, [seen_at] = @p2 \
235 /// WHERE [job_id] = @p3 AND [locked_until] = @p4",
236 /// [Param::Lease, Param::Now, Param::Id, Param::Held],
237 /// ))
238 /// }
239 /// # fn lease_claim(&self, spec: &TableSpec<'_>, shape: ClaimShape) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
240 /// # fn stamp(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
241 /// }
242 /// # impl Dialect for Mssql {
243 /// # fn name(&self) -> &'static str { "mssql" }
244 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
245 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
246 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
247 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
248 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
249 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
250 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
251 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
252 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
253 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
254 /// # }
255 /// ```
256 fn extend(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>;
257
258 /// The statement that leases one claimed row: it writes the expiry
259 /// ([`Param::Lease`](crate::Param::Lease)) and counts the attempt, while no lease holds the
260 /// row at [`Param::LeaseNow`](crate::Param::LeaseNow).
261 ///
262 /// A claim that only selects its rows runs it for each of them inside its transaction: the
263 /// claim of a dialect whose [`claim_writes_lease`](Self::claim_writes_lease) answers `false`,
264 /// or a claim the service writes itself.
265 ///
266 /// # Errors
267 ///
268 /// [`StatementError::FormMismatch`] for a table in another form;
269 /// [`StatementError::LeaseOnDatabaseClock`] for a lease table on the database's clock.
270 ///
271 /// # Examples
272 ///
273 /// ```
274 /// # use std::num::NonZeroUsize;
275 /// # use ruststream_sqlx_dialect::TableName;
276 /// use ruststream_sqlx_dialect::{
277 /// ClaimShape, Dialect, Lease, Param, Statement, StatementError, TableSpec,
278 /// };
279 ///
280 /// /// SQL Server, a database without a built-in dialect.
281 /// #[derive(Debug)]
282 /// pub struct Mssql;
283 ///
284 /// impl Lease for Mssql {
285 /// // The claim only selects, so the broker stamps each row it took.
286 /// fn claim_writes_lease(&self) -> bool {
287 /// false
288 /// }
289 ///
290 /// fn begin_lease_claim(&self) -> Option<&'static str> {
291 /// Some("SET TRANSACTION ISOLATION LEVEL READ COMMITTED; BEGIN TRANSACTION")
292 /// }
293 ///
294 /// // The stamp of an email also records when it was claimed, on the server's clock.
295 /// fn stamp(&self, _spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
296 /// Ok(Statement::new(
297 /// "UPDATE [email_jobs] SET [locked_until] = @p1, [attempt] = [attempt] + 1, \
298 /// [claimed_at] = SYSUTCDATETIME() WHERE [job_id] = @p2 \
299 /// AND ([locked_until] IS NULL OR [locked_until] <= @p3)",
300 /// [Param::Lease, Param::Id, Param::LeaseNow],
301 /// ))
302 /// }
303 /// # fn lease_claim(&self, _spec: &TableSpec<'_>, _shape: ClaimShape) -> Result<Statement, StatementError> { Ok(Statement::new("SELECT TOP (@p2) [job_id], [attempt], [payload] FROM [email_jobs] WITH (UPDLOCK, READPAST, ROWLOCK) WHERE [locked_until] IS NULL OR [locked_until] <= @p1 ORDER BY [job_id]", [Param::LeaseNow, Param::Limit])) }
304 /// # fn extend(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
305 /// }
306 /// # impl Dialect for Mssql {
307 /// # fn name(&self) -> &'static str { "mssql" }
308 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
309 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
310 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
311 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
312 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
313 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
314 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
315 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
316 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
317 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
318 /// # }
319 /// ```
320 fn stamp(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>;
321
322 /// Whether the lease claim writes the lease itself. A dialect whose lease claim only selects
323 /// the rows answers `false`, and the claim's transaction then runs [`stamp`](Self::stamp) for
324 /// each claimed row before it commits.
325 ///
326 /// The provided method answers `true`, which suits a claim that writes the lease in its own
327 /// statement. A dialect whose claim only selects answers `false`, or the rows it selects reach
328 /// their handlers without a lease.
329 ///
330 /// # Examples
331 ///
332 /// ```
333 /// # use std::num::NonZeroUsize;
334 /// # use ruststream_sqlx_dialect::TableName;
335 /// use ruststream_sqlx_dialect::{
336 /// ClaimShape, Dialect, Lease, Param, Statement, StatementError, TableSpec,
337 /// };
338 ///
339 /// /// SQL Server, a database without a built-in dialect.
340 /// #[derive(Debug)]
341 /// pub struct Mssql;
342 ///
343 /// impl Lease for Mssql {
344 /// // The claim reads the claimable rows under an update lock and writes nothing.
345 /// fn lease_claim(
346 /// &self,
347 /// _spec: &TableSpec<'_>,
348 /// _shape: ClaimShape,
349 /// ) -> Result<Statement, StatementError> {
350 /// Ok(Statement::new(
351 /// "SELECT TOP (@p2) [job_id], [attempt], [payload] FROM [email_jobs] \
352 /// WITH (UPDLOCK, READPAST, ROWLOCK) \
353 /// WHERE [locked_until] IS NULL OR [locked_until] <= @p1 ORDER BY [job_id]",
354 /// [Param::LeaseNow, Param::Limit],
355 /// ))
356 /// }
357 ///
358 /// // So the broker stamps each row the claim took before the claim's transaction commits.
359 /// fn claim_writes_lease(&self) -> bool {
360 /// false
361 /// }
362 ///
363 /// fn begin_lease_claim(&self) -> Option<&'static str> {
364 /// Some("SET TRANSACTION ISOLATION LEVEL READ COMMITTED; BEGIN TRANSACTION")
365 /// }
366 /// # fn extend(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
367 /// # fn stamp(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
368 /// }
369 /// # impl Dialect for Mssql {
370 /// # fn name(&self) -> &'static str { "mssql" }
371 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
372 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
373 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
374 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
375 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
376 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
377 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
378 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
379 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
380 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
381 /// # }
382 /// ```
383 fn claim_writes_lease(&self) -> bool {
384 true
385 }
386
387 /// Whether the rows the lease claim of `spec` returns carry the attempt the claim counted. A
388 /// dialect whose claim returns the rows as they were before answers `false`; where it answers
389 /// `true`, a delivery reports one less than its row carries.
390 ///
391 /// The provided method answers `false`, which suits a claim that returns its rows as they were
392 /// before the count. A dialect whose claim returns them as it left them answers `true`, or each
393 /// delivery reports one attempt more than it is.
394 ///
395 /// # Examples
396 ///
397 /// ```
398 /// # use std::num::NonZeroUsize;
399 /// # use ruststream_sqlx_dialect::TableName;
400 /// use ruststream_sqlx_dialect::{
401 /// ClaimShape, Dialect, Lease, Param, Statement, StatementError, TableSpec,
402 /// };
403 ///
404 /// /// SQL Server, a database without a built-in dialect.
405 /// #[derive(Debug)]
406 /// pub struct Mssql;
407 ///
408 /// impl Lease for Mssql {
409 /// // One update takes the rows of `email_jobs`, writes their lease, counts the attempt and
410 /// // returns them as it left them (`inserted`).
411 /// fn lease_claim(
412 /// &self,
413 /// _spec: &TableSpec<'_>,
414 /// shape: ClaimShape,
415 /// ) -> Result<Statement, StatementError> {
416 /// let returned = match shape {
417 /// ClaimShape::Rows => {
418 /// "inserted.[job_id], inserted.[attempt], inserted.[locked_until], \
419 /// inserted.[payload]"
420 /// }
421 /// ClaimShape::Ids => "inserted.[job_id]",
422 /// ClaimShape::Roles => {
423 /// "inserted.[job_id] AS [id], inserted.[attempt] AS [attempt], \
424 /// inserted.[payload] AS [payload]"
425 /// }
426 /// };
427 /// Ok(Statement::new(
428 /// format!(
429 /// "WITH [claimed] AS (SELECT TOP (@p3) * FROM [email_jobs] \
430 /// WITH (UPDLOCK, READPAST, ROWLOCK) \
431 /// WHERE [locked_until] IS NULL OR [locked_until] <= @p2 ORDER BY [job_id]) \
432 /// UPDATE [claimed] SET [locked_until] = @p1, [attempt] = [attempt] + 1 \
433 /// OUTPUT {returned}"
434 /// ),
435 /// [Param::Lease, Param::LeaseNow, Param::Limit],
436 /// ))
437 /// }
438 ///
439 /// // So each delivery reports the attempt before the count.
440 /// fn claim_counts_attempt(&self, _spec: &TableSpec<'_>) -> bool {
441 /// true
442 /// }
443 /// # fn extend(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
444 /// # fn stamp(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
445 /// }
446 /// # impl Dialect for Mssql {
447 /// # fn name(&self) -> &'static str { "mssql" }
448 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
449 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
450 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
451 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
452 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
453 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
454 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
455 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
456 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
457 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
458 /// # }
459 /// ```
460 fn claim_counts_attempt(&self, spec: &TableSpec<'_>) -> bool {
461 let _ = spec;
462 false
463 }
464
465 /// The statement that opens the transaction of a claim that only selects its rows, in place
466 /// of `BEGIN`, or `None` when `BEGIN` opens it.
467 ///
468 /// The transaction holds the claim's select and the stamp of each row it took, and commits
469 /// before the handlers run. MySQL opens it at READ COMMITTED, as its row lock claim; SQLite
470 /// opens it with the write lock taken, so no other writer comes between the select and the
471 /// stamps. The provided method returns `None`: the transaction opens with a plain `BEGIN`, at
472 /// the connection's level.
473 ///
474 /// # Examples
475 ///
476 /// ```
477 /// # use std::num::NonZeroUsize;
478 /// # use ruststream_sqlx_dialect::TableName;
479 /// use ruststream_sqlx_dialect::{
480 /// ClaimShape, Dialect, Lease, Statement, StatementError, TableSpec,
481 /// };
482 ///
483 /// /// SQL Server, a database without a built-in dialect.
484 /// #[derive(Debug)]
485 /// pub struct Mssql;
486 ///
487 /// impl Lease for Mssql {
488 /// // A table with a claim of its own (`custom(claim)`) selects its rows, then stamps them,
489 /// // in the transaction this opens. SQL Server starts one with `BEGIN TRANSACTION`, at the
490 /// // level the statement names.
491 /// fn begin_lease_claim(&self) -> Option<&'static str> {
492 /// Some("SET TRANSACTION ISOLATION LEVEL READ COMMITTED; BEGIN TRANSACTION")
493 /// }
494 /// # fn lease_claim(&self, spec: &TableSpec<'_>, shape: ClaimShape) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
495 /// # fn extend(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
496 /// # fn stamp(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
497 /// }
498 /// # impl Dialect for Mssql {
499 /// # fn name(&self) -> &'static str { "mssql" }
500 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
501 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
502 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
503 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
504 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
505 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
506 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
507 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
508 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
509 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
510 /// # }
511 /// ```
512 fn begin_lease_claim(&self) -> Option<&'static str> {
513 None
514 }
515}