prolly_store_mysql/
lib.rs

1#![doc = include_str!("../README.md")]
2
3pub use prolly::{
4    RemoteBatchOp, RemoteManifestUpdate, RemoteNamedRoot, RemoteProllyStore, RemoteRootCondition,
5    RemoteRootWrite, RemoteStoreBackend, RemoteTransactionConflict, RemoteTransactionUpdate,
6};
7
8/// MySQL adapter entry point.
9pub mod mysql {
10    use sqlx::{MySqlPool, Row};
11
12    use crate::{
13        RemoteBatchOp, RemoteManifestUpdate, RemoteNamedRoot, RemoteRootCondition, RemoteRootWrite,
14        RemoteStoreBackend, RemoteTransactionConflict, RemoteTransactionUpdate,
15    };
16
17    /// Store adapter for MySQL-backed prolly nodes and roots.
18    pub type MySqlStore = crate::RemoteProllyStore<MySqlBackend>;
19
20    /// SQLx-backed MySQL backend.
21    #[derive(Clone, Debug)]
22    pub struct MySqlBackend {
23        pool: MySqlPool,
24    }
25
26    impl MySqlBackend {
27        /// Create a backend from an existing SQLx pool.
28        pub fn new(pool: MySqlPool) -> Self {
29            Self { pool }
30        }
31
32        /// Connect to MySQL using `database_url`.
33        pub async fn connect(database_url: &str) -> Result<Self, sqlx::Error> {
34            Ok(Self::new(MySqlPool::connect(database_url).await?))
35        }
36
37        /// Borrow the underlying pool.
38        pub fn pool(&self) -> &MySqlPool {
39            &self.pool
40        }
41
42        /// Create the required tables if they do not already exist.
43        pub async fn initialize_schema(&self) -> Result<(), sqlx::Error> {
44            execute_statements(&self.pool, MYSQL_SCHEMA).await
45        }
46    }
47
48    impl RemoteStoreBackend for MySqlBackend {
49        type Error = sqlx::Error;
50
51        async fn get_node(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {
52            sqlx::query("SELECT node FROM prolly_nodes WHERE cid = ?")
53                .bind(key)
54                .fetch_optional(&self.pool)
55                .await?
56                .map(|row| row.try_get("node"))
57                .transpose()
58        }
59
60        async fn put_node(&self, key: &[u8], value: &[u8]) -> Result<(), Self::Error> {
61            sqlx::query(
62                "\
63                INSERT INTO prolly_nodes (cid, node) VALUES (?, ?) \
64                ON DUPLICATE KEY UPDATE node = VALUES(node)",
65            )
66            .bind(key)
67            .bind(value)
68            .execute(&self.pool)
69            .await?;
70            Ok(())
71        }
72
73        async fn delete_node(&self, key: &[u8]) -> Result<(), Self::Error> {
74            sqlx::query("DELETE FROM prolly_nodes WHERE cid = ?")
75                .bind(key)
76                .execute(&self.pool)
77                .await?;
78            Ok(())
79        }
80
81        async fn batch_nodes(&self, ops: &[RemoteBatchOp<'_>]) -> Result<(), Self::Error> {
82            let mut tx = self.pool.begin().await?;
83            for op in ops {
84                match op {
85                    RemoteBatchOp::Upsert { key, value } => {
86                        sqlx::query(
87                            "\
88                            INSERT INTO prolly_nodes (cid, node) VALUES (?, ?) \
89                            ON DUPLICATE KEY UPDATE node = VALUES(node)",
90                        )
91                        .bind(*key)
92                        .bind(*value)
93                        .execute(&mut *tx)
94                        .await?;
95                    }
96                    RemoteBatchOp::Delete { key } => {
97                        sqlx::query("DELETE FROM prolly_nodes WHERE cid = ?")
98                            .bind(*key)
99                            .execute(&mut *tx)
100                            .await?;
101                    }
102                }
103            }
104            tx.commit().await
105        }
106
107        async fn batch_get_nodes_ordered(
108            &self,
109            keys: &[&[u8]],
110        ) -> Result<Vec<Option<Vec<u8>>>, Self::Error> {
111            let mut values = Vec::with_capacity(keys.len());
112            for key in keys {
113                values.push(self.get_node(key).await?);
114            }
115            Ok(values)
116        }
117
118        async fn batch_put_nodes(&self, entries: &[(&[u8], &[u8])]) -> Result<(), Self::Error> {
119            let mut tx = self.pool.begin().await?;
120            for (key, value) in entries {
121                sqlx::query(
122                    "\
123                    INSERT INTO prolly_nodes (cid, node) VALUES (?, ?) \
124                    ON DUPLICATE KEY UPDATE node = VALUES(node)",
125                )
126                .bind(*key)
127                .bind(*value)
128                .execute(&mut *tx)
129                .await?;
130            }
131            tx.commit().await
132        }
133
134        async fn list_node_cids(&self) -> Result<Vec<Vec<u8>>, Self::Error> {
135            let rows = sqlx::query("SELECT cid FROM prolly_nodes ORDER BY cid")
136                .fetch_all(&self.pool)
137                .await?;
138            rows.into_iter().map(|row| row.try_get("cid")).collect()
139        }
140
141        fn prefers_batch_reads(&self) -> bool {
142            true
143        }
144
145        fn supports_hints(&self) -> bool {
146            true
147        }
148
149        async fn get_hint(
150            &self,
151            namespace: &[u8],
152            key: &[u8],
153        ) -> Result<Option<Vec<u8>>, Self::Error> {
154            sqlx::query("SELECT value FROM prolly_hints WHERE namespace = ? AND `key` = ?")
155                .bind(namespace)
156                .bind(key)
157                .fetch_optional(&self.pool)
158                .await?
159                .map(|row| row.try_get("value"))
160                .transpose()
161        }
162
163        async fn put_hint(
164            &self,
165            namespace: &[u8],
166            key: &[u8],
167            value: &[u8],
168        ) -> Result<(), Self::Error> {
169            sqlx::query(
170                "\
171                INSERT INTO prolly_hints (namespace, `key`, value) VALUES (?, ?, ?) \
172                ON DUPLICATE KEY UPDATE value = VALUES(value)",
173            )
174            .bind(namespace)
175            .bind(key)
176            .bind(value)
177            .execute(&self.pool)
178            .await?;
179            Ok(())
180        }
181
182        async fn batch_put_nodes_with_hint(
183            &self,
184            entries: &[(&[u8], &[u8])],
185            namespace: &[u8],
186            key: &[u8],
187            value: &[u8],
188        ) -> Result<(), Self::Error> {
189            let mut tx = self.pool.begin().await?;
190            for (key, value) in entries {
191                sqlx::query(
192                    "\
193                    INSERT INTO prolly_nodes (cid, node) VALUES (?, ?) \
194                    ON DUPLICATE KEY UPDATE node = VALUES(node)",
195                )
196                .bind(*key)
197                .bind(*value)
198                .execute(&mut *tx)
199                .await?;
200            }
201            sqlx::query(
202                "\
203                INSERT INTO prolly_hints (namespace, `key`, value) VALUES (?, ?, ?) \
204                ON DUPLICATE KEY UPDATE value = VALUES(value)",
205            )
206            .bind(namespace)
207            .bind(key)
208            .bind(value)
209            .execute(&mut *tx)
210            .await?;
211            tx.commit().await
212        }
213
214        async fn get_root_manifest(&self, name: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {
215            sqlx::query("SELECT manifest FROM prolly_roots WHERE name = ?")
216                .bind(name)
217                .fetch_optional(&self.pool)
218                .await?
219                .map(|row| row.try_get("manifest"))
220                .transpose()
221        }
222
223        async fn put_root_manifest(&self, name: &[u8], manifest: &[u8]) -> Result<(), Self::Error> {
224            sqlx::query(
225                "\
226                INSERT INTO prolly_roots (name, manifest) VALUES (?, ?) \
227                ON DUPLICATE KEY UPDATE manifest = VALUES(manifest)",
228            )
229            .bind(name)
230            .bind(manifest)
231            .execute(&self.pool)
232            .await?;
233            Ok(())
234        }
235
236        async fn delete_root_manifest(&self, name: &[u8]) -> Result<(), Self::Error> {
237            sqlx::query("DELETE FROM prolly_roots WHERE name = ?")
238                .bind(name)
239                .execute(&self.pool)
240                .await?;
241            Ok(())
242        }
243
244        async fn compare_and_swap_root_manifest(
245            &self,
246            name: &[u8],
247            expected: Option<&[u8]>,
248            new: Option<&[u8]>,
249        ) -> Result<RemoteManifestUpdate, Self::Error> {
250            let mut tx = self.pool.begin().await?;
251            let applied = match (expected, new) {
252                (None, Some(manifest)) => {
253                    sqlx::query("INSERT IGNORE INTO prolly_roots (name, manifest) VALUES (?, ?)")
254                        .bind(name)
255                        .bind(manifest)
256                        .execute(&mut *tx)
257                        .await?
258                        .rows_affected()
259                        == 1
260                }
261                (Some(expected), Some(manifest)) => {
262                    sqlx::query(
263                        "UPDATE prolly_roots SET manifest = ? WHERE name = ? AND manifest = ?",
264                    )
265                    .bind(manifest)
266                    .bind(name)
267                    .bind(expected)
268                    .execute(&mut *tx)
269                    .await?
270                    .rows_affected()
271                        == 1
272                }
273                (Some(expected), None) => {
274                    sqlx::query("DELETE FROM prolly_roots WHERE name = ? AND manifest = ?")
275                        .bind(name)
276                        .bind(expected)
277                        .execute(&mut *tx)
278                        .await?
279                        .rows_affected()
280                        == 1
281                }
282                (None, None) => {
283                    sqlx::query("SELECT manifest FROM prolly_roots WHERE name = ? FOR UPDATE")
284                        .bind(name)
285                        .fetch_optional(&mut *tx)
286                        .await?
287                        .is_none()
288                }
289            };
290
291            if applied {
292                tx.commit().await?;
293                return Ok(RemoteManifestUpdate::Applied);
294            }
295
296            let current = sqlx::query("SELECT manifest FROM prolly_roots WHERE name = ?")
297                .bind(name)
298                .fetch_optional(&mut *tx)
299                .await?
300                .map(|row| row.try_get("manifest"))
301                .transpose()?;
302            tx.rollback().await?;
303            Ok(RemoteManifestUpdate::Conflict { current })
304        }
305
306        async fn list_root_manifests(&self) -> Result<Vec<RemoteNamedRoot>, Self::Error> {
307            let rows = sqlx::query("SELECT name, manifest FROM prolly_roots ORDER BY name")
308                .fetch_all(&self.pool)
309                .await?;
310            rows.into_iter()
311                .map(|row| {
312                    Ok(RemoteNamedRoot::new(
313                        row.try_get("name")?,
314                        row.try_get("manifest")?,
315                    ))
316                })
317                .collect()
318        }
319
320        fn supports_transactions(&self) -> bool {
321            true
322        }
323
324        async fn commit_transaction(
325            &self,
326            node_writes: &[RemoteBatchOp<'_>],
327            root_conditions: &[RemoteRootCondition],
328            root_writes: &[RemoteRootWrite],
329        ) -> Result<RemoteTransactionUpdate, Self::Error> {
330            let mut tx = self.pool.begin().await?;
331
332            for condition in root_conditions {
333                let current =
334                    sqlx::query("SELECT manifest FROM prolly_roots WHERE name = ? FOR UPDATE")
335                        .bind(&condition.name)
336                        .fetch_optional(&mut *tx)
337                        .await?
338                        .map(|row| row.try_get("manifest"))
339                        .transpose()?;
340                if current != condition.expected {
341                    tx.rollback().await?;
342                    return Ok(RemoteTransactionUpdate::Conflict(
343                        RemoteTransactionConflict::new(
344                            condition.name.clone(),
345                            condition.expected.clone(),
346                            current,
347                        ),
348                    ));
349                }
350            }
351
352            for write in node_writes {
353                match write {
354                    RemoteBatchOp::Upsert { key, value } => {
355                        sqlx::query(
356                            "\
357                            INSERT INTO prolly_nodes (cid, node) VALUES (?, ?) \
358                            ON DUPLICATE KEY UPDATE node = VALUES(node)",
359                        )
360                        .bind(*key)
361                        .bind(*value)
362                        .execute(&mut *tx)
363                        .await?;
364                    }
365                    RemoteBatchOp::Delete { key } => {
366                        sqlx::query("DELETE FROM prolly_nodes WHERE cid = ?")
367                            .bind(*key)
368                            .execute(&mut *tx)
369                            .await?;
370                    }
371                }
372            }
373
374            for write in root_writes {
375                match write {
376                    RemoteRootWrite::Put { name, manifest } => {
377                        sqlx::query(
378                            "\
379                            INSERT INTO prolly_roots (name, manifest) VALUES (?, ?) \
380                            ON DUPLICATE KEY UPDATE manifest = VALUES(manifest)",
381                        )
382                        .bind(name)
383                        .bind(manifest)
384                        .execute(&mut *tx)
385                        .await?;
386                    }
387                    RemoteRootWrite::Delete { name } => {
388                        sqlx::query("DELETE FROM prolly_roots WHERE name = ?")
389                            .bind(name)
390                            .execute(&mut *tx)
391                            .await?;
392                    }
393                }
394            }
395
396            tx.commit().await?;
397            Ok(RemoteTransactionUpdate::Applied)
398        }
399    }
400
401    async fn execute_statements(pool: &MySqlPool, sql: &str) -> Result<(), sqlx::Error> {
402        for statement in sql
403            .split(';')
404            .map(str::trim)
405            .filter(|stmt| !stmt.is_empty())
406        {
407            sqlx::query(statement).execute(pool).await?;
408        }
409        Ok(())
410    }
411
412    /// Minimal table layout for MySQL implementations.
413    pub const MYSQL_SCHEMA: &str = "\
414CREATE TABLE IF NOT EXISTS prolly_nodes (
415  cid VARBINARY(32) PRIMARY KEY,
416  node LONGBLOB NOT NULL
417);
418CREATE TABLE IF NOT EXISTS prolly_hints (
419  namespace VARBINARY(255) NOT NULL,
420  `key` VARBINARY(255) NOT NULL,
421  value LONGBLOB NOT NULL,
422  PRIMARY KEY(namespace, `key`)
423);
424CREATE TABLE IF NOT EXISTS prolly_roots (
425  name VARBINARY(255) PRIMARY KEY,
426  manifest LONGBLOB NOT NULL
427);";
428}
429
430pub use mysql::*;
prolly_store_mysql/lib.rs

prolly_store_mysql/
lib.rs
