ruststream_sqlx_dialect/statement/
error.rs

1//! Why a dialect cannot build a statement, and the longest name a database keeps.
2
3use std::fmt::{self, Display, Formatter};
4
5use thiserror::Error;
6
7use crate::role::Role;
8
9/// The longest name a database keeps, in the unit it measures names by.
10///
11/// Postgres measures a name in bytes, MySQL and MariaDB in characters, so a name of accented
12/// letters can fit one database and not the other.
13///
14/// # Examples
15///
16/// ```
17/// # use std::num::NonZeroUsize;
18/// use ruststream_sqlx_dialect::{
19///     Dialect, NameLimit, Param, Statement, StatementError, TableSpec,
20/// };
21/// # use ruststream_sqlx_dialect::TableName;
22///
23/// /// SQL Server, a database without a built-in dialect.
24/// #[derive(Debug)]
25/// pub struct Mssql;
26///
27/// impl Mssql {
28///     // SQL Server keeps a name of up to 128 characters.
29///     fn named<'n>(&self, name: &'n str) -> Result<&'n str, StatementError> {
30///         if name.chars().count() <= 128 {
31///             return Ok(name);
32///         }
33///         Err(StatementError::IdentifierTooLong {
34///             dialect: self.name(),
35///             identifier: name.to_owned(),
36///             limit: NameLimit::Characters(128),
37///         })
38///     }
39/// }
40///
41/// impl Dialect for Mssql {
42///     fn name(&self) -> &'static str {
43///         "mssql"
44///     }
45///
46///     fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
47///         let mut sql = String::from("DELETE FROM ");
48///         self.quote_into(self.named(spec.table())?, &mut sql);
49///         sql.push_str(" WHERE ");
50///         self.quote_into(self.named(spec.id().name())?, &mut sql);
51///         sql.push_str(" = @p1");
52///         Ok(Statement::new(sql, [Param::Id]))
53///     }
54/// #     fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
55/// #     fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
56/// #     fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
57/// #     fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
58/// #     fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
59/// #     fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
60/// #     fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
61/// #     fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
62/// #     fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
63/// }
64/// ```
65// A `u16` holds every database's limit and keeps `StatementError` at 56 bytes: a `usize` would
66// grow it, and every error that carries it, by eight.
67#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
68pub enum NameLimit {
69    /// At most this many bytes of UTF-8.
70    Bytes(u16),
71    /// At most this many characters.
72    Characters(u16),
73}
74
75impl Display for NameLimit {
76    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
77        match self {
78            Self::Bytes(limit) => write!(f, "{limit} bytes"),
79            Self::Characters(limit) => write!(f, "{limit} characters"),
80        }
81    }
82}
83
84/// Why a dialect cannot build a statement for a table.
85///
86/// A broker builds a subscription's statements when the subscription starts, so a refusal stops
87/// it before it claims a row, and the message names the change to make.
88///
89/// # Examples
90///
91/// A dialect of the service's own refuses a table its claim does not serve:
92///
93/// ```
94/// # use std::num::NonZeroUsize;
95/// use ruststream_sqlx_dialect::{
96///     ClaimShape, Dialect, Isolation, Opening, Param, RowLock, Statement, StatementError,
97///     TableSpec,
98/// };
99/// # use ruststream_sqlx_dialect::TableName;
100///
101/// /// SQL Server, a database without a built-in dialect.
102/// #[derive(Debug)]
103/// pub struct Mssql;
104///
105/// impl RowLock for Mssql {
106///     // The claim of the service's one queue table, which it reads whole.
107///     fn lock_claim(
108///         &self,
109///         spec: &TableSpec<'_>,
110///         _shape: ClaimShape,
111///     ) -> Result<Statement, StatementError> {
112///         // This claim keeps no group in order.
113///         if spec.is_fifo() {
114///             return Err(StatementError::UnsupportedFifo { dialect: self.name() });
115///         }
116///
117///         // READPAST, which skips the rows another claim holds, refuses SERIALIZABLE.
118///         let opening = spec.opening();
119///         if opening == Opening::Isolation(Isolation::Serializable) {
120///             return Err(StatementError::UnsupportedOpening {
121///                 dialect: self.name(),
122///                 opening: opening.name(),
123///             });
124///         }
125///         Ok(Statement::new(
126///             "SELECT TOP (@p1) [job_id], [payload] FROM [email_jobs] \
127///              WITH (UPDLOCK, READPAST) ORDER BY [job_id]",
128///             [Param::Limit],
129///         ))
130///     }
131/// }
132/// # impl Dialect for Mssql {
133/// #     fn name(&self) -> &'static str { "mssql" }
134/// #     fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
135/// #     fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
136/// #     fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
137/// #     fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
138/// #     fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
139/// #     fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
140/// #     fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
141/// #     fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
142/// #     fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
143/// #     fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
144/// # }
145/// ```
146#[derive(Debug, Clone, PartialEq, Eq, Error)]
147#[non_exhaustive]
148pub enum StatementError {
149    /// The statement needs a column that plays `role`, and the table has none.
150    #[error(
151        "the {statement} statement needs a column playing `{role}`: add `#[field({role})]` to \
152         the struct"
153    )]
154    MissingRole {
155        /// The statement being built.
156        statement: &'static str,
157        /// The role it needs.
158        role: Role,
159    },
160    /// The dialect has no statements for the table's form.
161    #[error("the {dialect} dialect has no statements for the {form} form")]
162    UnsupportedForm {
163        /// The dialect's name.
164        dialect: &'static str,
165        /// The name of the table's form.
166        form: &'static str,
167    },
168    /// The statement belongs to one form, and the table takes rows in another: the row lock
169    /// claim asked for a lease table, a lease statement for a table without a lease, or an
170    /// advisory statement for a table without a lock key.
171    #[error("the {statement} statement does not serve a table in the {form} form")]
172    FormMismatch {
173        /// The statement being built.
174        statement: &'static str,
175        /// The name of the table's form.
176        form: &'static str,
177    },
178    /// The dialect has no claim that keeps a group in order. A dialect of the service's own
179    /// returns it from a claim of a table with FIFO groups when it builds no claim for them.
180    #[error("the {dialect} dialect has no claim for FIFO groups")]
181    UnsupportedFifo {
182        /// The dialect's name.
183        dialect: &'static str,
184    },
185    /// The table takes its rows by advisory lock and keeps its groups in order. In that form a
186    /// lock key on the group's column keeps a group in order, so the advisory claim and the take
187    /// refuse FIFO groups.
188    #[error(
189        "FIFO groups do not combine with the advisory lock form in the {dialect} dialect: drop \
190         `fifo` and put the group's column into the lock key"
191    )]
192    AdvisoryFifo {
193        /// The dialect's name.
194        dialect: &'static str,
195    },
196    /// A table, schema or column name is longer than the database allows. Postgres would cut it
197    /// short without a word and address another object; MySQL would refuse the statement.
198    #[error("`{identifier}` is longer than the {limit} the {dialect} dialect allows in a name")]
199    IdentifierTooLong {
200        /// The dialect's name.
201        dialect: &'static str,
202        /// The name that is too long.
203        identifier: String,
204        /// The longest name the database keeps, in the unit it measures names by.
205        limit: NameLimit,
206    },
207    /// The statement needs every column, and the struct flattens another whose columns the
208    /// description cannot see.
209    #[error(
210        "the {statement} statement needs every column, and a `#[sqlx(flatten)]` field hides some: \
211         write it in the service"
212    )]
213    Flattened {
214        /// The statement being built.
215        statement: &'static str,
216    },
217    /// The server is older than the dialect's statements need, as
218    /// [`Dialect::check_server`](crate::Dialect::check_server) found when the subscription
219    /// started.
220    #[error(
221        "the {dialect} dialect needs {required} or later for this form; the server reports \
222         `{server}`"
223    )]
224    ServerTooOld {
225        /// The dialect's name.
226        dialect: &'static str,
227        /// The version the server reports.
228        server: String,
229        /// The oldest server the dialect's statements run on.
230        required: &'static str,
231    },
232    /// The dialect cannot read rows by a list of ids, so a claim of the service's own needs a
233    /// fetch of its own beside it.
234    #[error(
235        "the {dialect} dialect cannot read rows by a list of ids: list `fetch` in `custom(..)` \
236         beside `claim`"
237    )]
238    UnsupportedFetch {
239        /// The dialect's name.
240        dialect: &'static str,
241    },
242    /// The table declares a lease and reads the database's clock. Settlement matches the expiry
243    /// the claim wrote, so the claim computes it from the crate's clock.
244    #[error(
245        "the lease form computes its expiry from the crate's clock: a table on `DatabaseClock` \
246         cannot declare `locked_until`"
247    )]
248    LeaseOnDatabaseClock {
249        /// The dialect's name.
250        dialect: &'static str,
251    },
252    /// The dialect opens no transaction at the table's isolation level or mode, as
253    /// [`Dialect::begin`](crate::Dialect::begin) found.
254    #[error("the {dialect} dialect opens no transaction at {opening}")]
255    UnsupportedOpening {
256        /// The dialect's name.
257        dialect: &'static str,
258        /// The opening, as [`Opening::name`](crate::Opening::name) names it.
259        opening: &'static str,
260    },
261    /// The table keeps its groups in order, takes its rows by row lock and opens its claims at
262    /// SERIALIZABLE, a level at which the dialect's [`fifo_guard`](crate::Dialect::fifo_guard)
263    /// would wait for the group's row in work instead of answering at once.
264    #[error(
265        "the {dialect} dialect claims FIFO groups below SERIALIZABLE: InnoDB turns every read of a \
266         SERIALIZABLE transaction into a locking read, so a claim would wait for the group's row in \
267         work; declare `isolation = repeatable_read` or a lower level"
268    )]
269    FifoAtSerializable {
270        /// The dialect's name.
271        dialect: &'static str,
272    },
273}
274
275#[cfg(test)]
276mod tests {
277    use super::{NameLimit, StatementError};
278    use crate::role::Role;
279
280    #[test]
281    fn errors_name_the_missing_piece() {
282        assert_eq!(
283            StatementError::UnsupportedForm {
284                dialect: "postgres",
285                form: "advisory lock",
286            }
287            .to_string(),
288            "the postgres dialect has no statements for the advisory lock form"
289        );
290        assert_eq!(
291            StatementError::UnsupportedFifo {
292                dialect: "postgres"
293            }
294            .to_string(),
295            "the postgres dialect has no claim for FIFO groups"
296        );
297        assert_eq!(
298            StatementError::MissingRole {
299                statement: "dead_letter_group",
300                role: Role::Group,
301            }
302            .to_string(),
303            "the dead_letter_group statement needs a column playing `group`: add `#[field(group)]` \
304             to the struct"
305        );
306    }
307
308    #[test]
309    fn the_new_errors_name_what_to_fix() {
310        assert_eq!(
311            StatementError::IdentifierTooLong {
312                dialect: "postgres",
313                identifier: "jobs".to_owned(),
314                limit: NameLimit::Bytes(63),
315            }
316            .to_string(),
317            "`jobs` is longer than the 63 bytes the postgres dialect allows in a name"
318        );
319        assert_eq!(
320            StatementError::IdentifierTooLong {
321                dialect: "mysql",
322                identifier: "jobs".to_owned(),
323                limit: NameLimit::Characters(64),
324            }
325            .to_string(),
326            "`jobs` is longer than the 64 characters the mysql dialect allows in a name"
327        );
328        assert_eq!(
329            StatementError::Flattened {
330                statement: "insert"
331            }
332            .to_string(),
333            "the insert statement needs every column, and a `#[sqlx(flatten)]` field hides some: \
334             write it in the service"
335        );
336        assert_eq!(
337            StatementError::FormMismatch {
338                statement: "extend",
339                form: "row lock",
340            }
341            .to_string(),
342            "the extend statement does not serve a table in the row lock form"
343        );
344        assert_eq!(
345            StatementError::UnsupportedOpening {
346                dialect: "postgres",
347                opening: "isolation `read_uncommitted`",
348            }
349            .to_string(),
350            "the postgres dialect opens no transaction at isolation `read_uncommitted`"
351        );
352        assert_eq!(
353            StatementError::FifoAtSerializable { dialect: "mysql" }.to_string(),
354            "the mysql dialect claims FIFO groups below SERIALIZABLE: InnoDB turns every read of a \
355             SERIALIZABLE transaction into a locking read, so a claim would wait for the group's \
356             row in work; declare `isolation = repeatable_read` or a lower level"
357        );
358    }
359}
ruststream_sqlx_dialect/statement/error.rs

ruststream_sqlx_dialect/statement/
error.rs
