tds_driver/tds/codec/
transaction_manager.rs

1use super::{put_b_varbyte_utf16, AllHeaderTy, Encode, ALL_HEADERS_LEN_TX};
2use bytes::{BufMut, BytesMut};
3use std::borrow::Cow;
4
5uint_enum! {
6    /// The request type of a Transaction Manager request, as defined in
7    /// MS-TDS 2.2.6.9 (`TM_*` request kinds).
8    #[repr(u16)]
9    pub enum TransactionManagerRequestType {
10        /// Get the address of the Distributed Transaction Coordinator.
11        GetDtcAddress = 0,
12        /// Import an existing distributed transaction (propagate).
13        Propagate = 1,
14        /// Begin a new transaction (`TM_BEGIN_XACT`).
15        Begin = 5,
16        /// Promote a local transaction to a distributed one (`TM_PROMOTE_XACT`).
17        Promote = 6,
18        /// Commit the active transaction (`TM_COMMIT_XACT`).
19        Commit = 7,
20        /// Roll back the active transaction (`TM_ROLLBACK_XACT`).
21        Rollback = 8,
22        /// Create a savepoint in the active transaction (`TM_SAVE_XACT`).
23        Save = 9,
24    }
25}
26
27uint_enum! {
28    /// The transaction isolation level requested when beginning a transaction
29    /// through a Transaction Manager request (MS-TDS 2.2.6.9).
30    ///
31    /// Any level other than `Unspecified` is a session setting, like
32    /// `SET TRANSACTION ISOLATION LEVEL`: it stays in effect after the
33    /// transaction ends, for later transactions and for statements outside
34    /// any transaction.
35    #[repr(u8)]
36    pub enum IsolationLevel {
37        /// Keep the session's current isolation level ("No isolation level
38        /// change requested. Use current.", MS-TDS 2.2.6.9). That is READ
39        /// COMMITTED only on a session that never set another level: after a
40        /// SERIALIZABLE transaction, an `Unspecified` one is SERIALIZABLE too.
41        Unspecified = 0x00,
42        /// `READ UNCOMMITTED`.
43        ReadUncommitted = 0x01,
44        /// `READ COMMITTED`.
45        ReadCommitted = 0x02,
46        /// `REPEATABLE READ`.
47        RepeatableRead = 0x03,
48        /// `SERIALIZABLE`.
49        Serializable = 0x04,
50        /// `SNAPSHOT`.
51        Snapshot = 0x05,
52    }
53}
54
55/// A Transaction Manager request (packet type `0x0E`, MS-TDS 2.2.6.9).
56///
57/// These requests let the client begin, commit, roll back or create a
58/// savepoint in a transaction directly through the TDS protocol instead of
59/// issuing the equivalent T-SQL batch (`BEGIN TRAN`, `COMMIT`, ...).
60///
61/// Every request carries the current transaction descriptor in the request's
62/// `ALL_HEADERS` block so the server can associate it with the correct
63/// transaction.
64#[derive(Debug, Clone)]
65pub struct TransactionManagerRequest<'a> {
66    transaction_desc: [u8; 8],
67    body: TransactionRequestBody<'a>,
68}
69
70#[derive(Debug, Clone)]
71enum TransactionRequestBody<'a> {
72    Begin {
73        isolation_level: IsolationLevel,
74        name: Cow<'a, str>,
75    },
76    Commit {
77        name: Cow<'a, str>,
78    },
79    Rollback {
80        name: Cow<'a, str>,
81    },
82    Save {
83        name: Cow<'a, str>,
84    },
85}
86
87impl<'a> TransactionManagerRequest<'a> {
88    /// Build a `TM_BEGIN_XACT` request that begins a new transaction with the
89    /// given isolation level (see [`IsolationLevel`] for how long it lasts).
90    /// The (usually empty) transaction name is sent as a `B_VARBYTE`.
91    pub fn begin(
92        transaction_desc: [u8; 8],
93        isolation_level: IsolationLevel,
94        name: impl Into<Cow<'a, str>>,
95    ) -> Self {
96        Self {
97            transaction_desc,
98            body: TransactionRequestBody::Begin {
99                isolation_level,
100                name: name.into(),
101            },
102        }
103    }
104
105    /// Build a `TM_COMMIT_XACT` request that commits the active transaction.
106    pub fn commit(transaction_desc: [u8; 8], name: impl Into<Cow<'a, str>>) -> Self {
107        Self {
108            transaction_desc,
109            body: TransactionRequestBody::Commit { name: name.into() },
110        }
111    }
112
113    /// Build a `TM_ROLLBACK_XACT` request that rolls back the active
114    /// transaction (or to a savepoint of the given name).
115    pub fn rollback(transaction_desc: [u8; 8], name: impl Into<Cow<'a, str>>) -> Self {
116        Self {
117            transaction_desc,
118            body: TransactionRequestBody::Rollback { name: name.into() },
119        }
120    }
121
122    /// Build a `TM_SAVE_XACT` request that creates a savepoint with the given
123    /// name in the active transaction.
124    pub fn save(transaction_desc: [u8; 8], name: impl Into<Cow<'a, str>>) -> Self {
125        Self {
126            transaction_desc,
127            body: TransactionRequestBody::Save { name: name.into() },
128        }
129    }
130
131    fn request_type(&self) -> TransactionManagerRequestType {
132        match self.body {
133            TransactionRequestBody::Begin { .. } => TransactionManagerRequestType::Begin,
134            TransactionRequestBody::Commit { .. } => TransactionManagerRequestType::Commit,
135            TransactionRequestBody::Rollback { .. } => TransactionManagerRequestType::Rollback,
136            TransactionRequestBody::Save { .. } => TransactionManagerRequestType::Save,
137        }
138    }
139}
140
141/// What a request's names are, in the error for one too long. MS-TDS 2.2.6.9
142/// makes each a B_VARBYTE, counted in bytes; a B_VARCHAR's count of UTF-16
143/// units made the server reject any name of odd length (3976) and abort the
144/// transaction.
145const NAME: &str = "transaction or savepoint name";
146
147impl<'a> Encode<BytesMut> for TransactionManagerRequest<'a> {
148    fn encode(self, dst: &mut BytesMut) -> crate::Result<()> {
149        // ALL_HEADERS block carrying the transaction descriptor.
150        dst.put_u32_le(ALL_HEADERS_LEN_TX as u32);
151        dst.put_u32_le(ALL_HEADERS_LEN_TX as u32 - 4);
152        dst.put_u16_le(AllHeaderTy::TransactionDescriptor as u16);
153        dst.put_slice(&self.transaction_desc);
154        dst.put_u32_le(1);
155
156        // Request type (USHORT).
157        dst.put_u16_le(self.request_type() as u16);
158
159        match self.body {
160            TransactionRequestBody::Begin {
161                isolation_level,
162                name,
163            } => {
164                dst.put_u8(isolation_level as u8);
165                put_b_varbyte_utf16(dst, &name, NAME)?;
166            }
167            TransactionRequestBody::Commit { name } | TransactionRequestBody::Rollback { name } => {
168                put_b_varbyte_utf16(dst, &name, NAME)?;
169                // Flags byte: bit 0 (`fBeginXact`) unset — do not begin a new
170                // transaction after commit/rollback.
171                dst.put_u8(0);
172            }
173            TransactionRequestBody::Save { name } => put_b_varbyte_utf16(dst, &name, NAME)?,
174        }
175
176        Ok(())
177    }
178}
179
180#[cfg(test)]
181mod tests {
182    use super::*;
183
184    fn all_headers() -> Vec<u8> {
185        let mut v = Vec::new();
186        v.extend_from_slice(&(ALL_HEADERS_LEN_TX as u32).to_le_bytes());
187        v.extend_from_slice(&(ALL_HEADERS_LEN_TX as u32 - 4).to_le_bytes());
188        v.extend_from_slice(&(AllHeaderTy::TransactionDescriptor as u16).to_le_bytes());
189        v.extend_from_slice(&[1, 2, 3, 4, 5, 6, 7, 8]);
190        v.extend_from_slice(&1u32.to_le_bytes());
191        v
192    }
193
194    #[test]
195    fn encodes_begin_request() {
196        let desc = [1, 2, 3, 4, 5, 6, 7, 8];
197        let req = TransactionManagerRequest::begin(desc, IsolationLevel::ReadCommitted, "");
198
199        let mut buf = BytesMut::new();
200        req.encode(&mut buf).unwrap();
201
202        let mut expected = all_headers();
203        expected.extend_from_slice(&(TransactionManagerRequestType::Begin as u16).to_le_bytes());
204        expected.push(IsolationLevel::ReadCommitted as u8); // isolation level
205        expected.push(0); // B_VARBYTE length (empty name)
206
207        assert_eq!(&buf[..], &expected[..]);
208    }
209
210    #[test]
211    fn encodes_begin_request_with_name() {
212        let desc = [1, 2, 3, 4, 5, 6, 7, 8];
213        let req = TransactionManagerRequest::begin(desc, IsolationLevel::Serializable, "tx");
214
215        let mut buf = BytesMut::new();
216        req.encode(&mut buf).unwrap();
217
218        let mut expected = all_headers();
219        expected.extend_from_slice(&(TransactionManagerRequestType::Begin as u16).to_le_bytes());
220        expected.push(IsolationLevel::Serializable as u8);
221        expected.push(4); // bytes: two UTF-16 code units
222        expected.extend_from_slice(&b't'.to_le_bytes());
223        expected.push(0);
224        expected.extend_from_slice(&b'x'.to_le_bytes());
225        expected.push(0);
226
227        assert_eq!(&buf[..], &expected[..]);
228    }
229
230    #[test]
231    fn encodes_commit_request() {
232        let desc = [1, 2, 3, 4, 5, 6, 7, 8];
233        let req = TransactionManagerRequest::commit(desc, "");
234
235        let mut buf = BytesMut::new();
236        req.encode(&mut buf).unwrap();
237
238        let mut expected = all_headers();
239        expected.extend_from_slice(&(TransactionManagerRequestType::Commit as u16).to_le_bytes());
240        expected.push(0); // B_VARBYTE length (empty name)
241        expected.push(0); // flags: no new transaction
242
243        assert_eq!(&buf[..], &expected[..]);
244    }
245
246    #[test]
247    fn encodes_rollback_request() {
248        let desc = [8, 7, 6, 5, 4, 3, 2, 1];
249        let req = TransactionManagerRequest::rollback(desc, "");
250
251        let mut buf = BytesMut::new();
252        req.encode(&mut buf).unwrap();
253
254        let mut expected = Vec::new();
255        expected.extend_from_slice(&(ALL_HEADERS_LEN_TX as u32).to_le_bytes());
256        expected.extend_from_slice(&(ALL_HEADERS_LEN_TX as u32 - 4).to_le_bytes());
257        expected.extend_from_slice(&(AllHeaderTy::TransactionDescriptor as u16).to_le_bytes());
258        expected.extend_from_slice(&desc);
259        expected.extend_from_slice(&1u32.to_le_bytes());
260        expected.extend_from_slice(&(TransactionManagerRequestType::Rollback as u16).to_le_bytes());
261        expected.push(0); // B_VARBYTE length
262        expected.push(0); // flags
263
264        assert_eq!(&buf[..], &expected[..]);
265    }
266
267    #[test]
268    fn encodes_save_request() {
269        let desc = [1, 2, 3, 4, 5, 6, 7, 8];
270        let req = TransactionManagerRequest::save(desc, "sp1");
271
272        let mut buf = BytesMut::new();
273        req.encode(&mut buf).unwrap();
274
275        let mut expected = all_headers();
276        expected.extend_from_slice(&(TransactionManagerRequestType::Save as u16).to_le_bytes());
277        expected.push(6); // bytes: three UTF-16 code units
278        for unit in "sp1".encode_utf16() {
279            expected.extend_from_slice(&unit.to_le_bytes());
280        }
281
282        assert_eq!(&buf[..], &expected[..]);
283    }
284
285    #[test]
286    fn savepoint_name_over_127_units_is_an_error() {
287        // 128 units are 256 bytes, past the B_VARBYTE's u8 length.
288        for name in ["a".repeat(128), "\u{1F600}".repeat(64)] {
289            let err = TransactionManagerRequest::save([0; 8], name)
290                .encode(&mut BytesMut::new())
291                .unwrap_err();
292            assert!(matches!(err, crate::Error::Conversion(_)), "{err:?}");
293        }
294
295        let mut buf = BytesMut::new();
296        TransactionManagerRequest::save([0; 8], "a".repeat(127))
297            .encode(&mut buf)
298            .unwrap();
299        assert_eq!(buf[ALL_HEADERS_LEN_TX + 2], 254);
300    }
301}
tds_driver/tds/codec/transaction_manager.rs

tds_driver/tds/codec/
transaction_manager.rs
