1use std::collections::{hash_map::Entry, HashMap};
4use std::path::Path;
5use std::sync::{Mutex, MutexGuard};
6use std::time::Duration;
7
8use rusqlite::{params, Connection, OptionalExtension};
9
10use prolly::{
11 BatchOp, Cid, Error, ManifestStore, ManifestStoreScan, ManifestUpdate, NamedRootManifest,
12 NodeStoreScan, RootCondition, RootManifest, RootWrite, Store, TransactionConflict,
13 TransactionNodeWrite, TransactionUpdate, TransactionalStore,
14};
15
16struct OrderedBatchReadPlan<'a> {
17 unique_keys: Vec<&'a [u8]>,
18 positions: Option<Vec<usize>>,
19}
20
21impl<'a> OrderedBatchReadPlan<'a> {
22 fn new(keys: &[&'a [u8]]) -> Self {
23 let mut unique_indexes = HashMap::with_capacity(keys.len());
24 let mut unique_keys = Vec::with_capacity(keys.len());
25 let mut positions = None;
26 for key in keys {
27 match unique_indexes.entry(*key) {
28 Entry::Occupied(entry) => positions
29 .get_or_insert_with(|| (0..unique_keys.len()).collect::<Vec<_>>())
30 .push(*entry.get()),
31 Entry::Vacant(entry) => {
32 let index = unique_keys.len();
33 unique_keys.push(*key);
34 if let Some(positions) = positions.as_mut() {
35 positions.push(index);
36 }
37 entry.insert(index);
38 }
39 }
40 }
41 Self {
42 unique_keys,
43 positions,
44 }
45 }
46
47 fn unique_keys(&self) -> &[&'a [u8]] {
48 &self.unique_keys
49 }
50
51 fn expand_owned<T: Clone>(&self, values: Vec<Option<T>>) -> Vec<Option<T>> {
52 match &self.positions {
53 Some(positions) => positions
54 .iter()
55 .map(|&index| values[index].clone())
56 .collect(),
57 None => values,
58 }
59 }
60}
61
62fn cid_from_store_key(key: &[u8], context: &str) -> Result<Cid, String> {
63 let bytes: [u8; 32] = key.try_into().map_err(|_| {
64 format!(
65 "{context} key has invalid CID length {}, expected 32",
66 key.len()
67 )
68 })?;
69 Ok(Cid(bytes))
70}
71
72fn sort_cids(cids: &mut [Cid]) {
73 cids.sort_by(|left, right| left.as_bytes().cmp(right.as_bytes()));
74}
75
76fn sort_named_root_manifests(roots: &mut [NamedRootManifest]) {
77 roots.sort_by(|left, right| left.name.cmp(&right.name));
78}
79
80const CREATE_TABLE_SQL: &str = "\
81CREATE TABLE IF NOT EXISTS prolly_nodes (
82 cid BLOB PRIMARY KEY NOT NULL,
83 node BLOB NOT NULL
84) WITHOUT ROWID;";
85
86const CREATE_HINTS_TABLE_SQL: &str = "\
87CREATE TABLE IF NOT EXISTS prolly_hints (
88 namespace BLOB NOT NULL,
89 key BLOB NOT NULL,
90 value BLOB NOT NULL,
91 PRIMARY KEY (namespace, key)
92) WITHOUT ROWID;";
93
94const CREATE_ROOTS_TABLE_SQL: &str = "\
95CREATE TABLE IF NOT EXISTS prolly_roots (
96 name BLOB PRIMARY KEY NOT NULL,
97 manifest BLOB NOT NULL
98) WITHOUT ROWID;";
99
100const SELECT_SQL: &str = "SELECT node FROM prolly_nodes WHERE cid = ?1";
101const SELECT_NODE_CIDS_SQL: &str = "SELECT cid FROM prolly_nodes ORDER BY cid";
102const UPSERT_SQL: &str = "\
103INSERT INTO prolly_nodes (cid, node)
104VALUES (?1, ?2)
105ON CONFLICT(cid) DO UPDATE SET node = excluded.node";
106const DELETE_SQL: &str = "DELETE FROM prolly_nodes WHERE cid = ?1";
107const SELECT_ROOT_SQL: &str = "SELECT manifest FROM prolly_roots WHERE name = ?1";
108const SELECT_ROOTS_SQL: &str = "SELECT name, manifest FROM prolly_roots ORDER BY name";
109const UPSERT_ROOT_SQL: &str = "\
110INSERT INTO prolly_roots (name, manifest)
111VALUES (?1, ?2)
112ON CONFLICT(name) DO UPDATE SET manifest = excluded.manifest";
113const DELETE_ROOT_SQL: &str = "DELETE FROM prolly_roots WHERE name = ?1";
114
115#[derive(Debug, Clone)]
117pub struct SqliteStoreConfig {
118 pub busy_timeout_ms: u64,
120 pub enable_wal: bool,
122 pub synchronous_normal: bool,
124}
125
126impl Default for SqliteStoreConfig {
127 fn default() -> Self {
128 Self {
129 busy_timeout_ms: 5_000,
130 enable_wal: true,
131 synchronous_normal: true,
132 }
133 }
134}
135
136#[derive(Debug)]
138pub struct SqliteStoreError {
139 message: String,
140 source: Option<rusqlite::Error>,
141}
142
143impl SqliteStoreError {
144 pub fn new(message: impl Into<String>) -> Self {
146 Self {
147 message: message.into(),
148 source: None,
149 }
150 }
151
152 pub fn from_sqlite(err: rusqlite::Error, context: impl Into<String>) -> Self {
154 Self {
155 message: format!("{}: {}", context.into(), err),
156 source: Some(err),
157 }
158 }
159}
160
161impl std::fmt::Display for SqliteStoreError {
162 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
163 write!(f, "SQLite error: {}", self.message)
164 }
165}
166
167impl std::error::Error for SqliteStoreError {
168 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
169 self.source
170 .as_ref()
171 .map(|e| e as &(dyn std::error::Error + 'static))
172 }
173}
174
175impl From<rusqlite::Error> for SqliteStoreError {
176 fn from(err: rusqlite::Error) -> Self {
177 Self {
178 message: err.to_string(),
179 source: Some(err),
180 }
181 }
182}
183
184pub struct SqliteStore {
189 conn: Mutex<Connection>,
190}
191
192impl SqliteStore {
193 pub fn open<P: AsRef<Path>>(path: P) -> Result<Self, SqliteStoreError> {
195 Self::open_with_config(path, SqliteStoreConfig::default())
196 }
197
198 pub fn open_with_config<P: AsRef<Path>>(
200 path: P,
201 config: SqliteStoreConfig,
202 ) -> Result<Self, SqliteStoreError> {
203 let conn = Connection::open(path.as_ref()).map_err(|e| {
204 SqliteStoreError::from_sqlite(
205 e,
206 format!("Failed to open database at {:?}", path.as_ref()),
207 )
208 })?;
209 Self::from_connection(conn, config)
210 }
211
212 pub fn open_in_memory() -> Result<Self, SqliteStoreError> {
214 let conn = Connection::open_in_memory()
215 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to open in-memory database"))?;
216 Self::from_connection(conn, SqliteStoreConfig::default())
217 }
218
219 fn from_connection(
220 conn: Connection,
221 config: SqliteStoreConfig,
222 ) -> Result<Self, SqliteStoreError> {
223 conn.busy_timeout(Duration::from_millis(config.busy_timeout_ms))
224 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to set busy timeout"))?;
225
226 if config.enable_wal {
227 conn.pragma_update(None, "journal_mode", "WAL")
228 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to enable WAL mode"))?;
229 }
230 if config.synchronous_normal {
231 conn.pragma_update(None, "synchronous", "NORMAL")
232 .map_err(|e| {
233 SqliteStoreError::from_sqlite(e, "Failed to set synchronous=NORMAL")
234 })?;
235 }
236 conn.pragma_update(None, "temp_store", "MEMORY")
237 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to set temp_store=MEMORY"))?;
238 conn.execute_batch(CREATE_TABLE_SQL)
239 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to initialize schema"))?;
240 conn.execute_batch(CREATE_HINTS_TABLE_SQL)
241 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to initialize hint schema"))?;
242 conn.execute_batch(CREATE_ROOTS_TABLE_SQL)
243 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to initialize root schema"))?;
244
245 Ok(Self {
246 conn: Mutex::new(conn),
247 })
248 }
249
250 fn connection(&self) -> Result<MutexGuard<'_, Connection>, SqliteStoreError> {
251 self.conn
252 .lock()
253 .map_err(|e| SqliteStoreError::new(format!("lock poisoned: {}", e)))
254 }
255}
256
257impl Store for SqliteStore {
258 type Error = SqliteStoreError;
259
260 fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {
261 let conn = self.connection()?;
262 conn.query_row(SELECT_SQL, params![key], |row| row.get(0))
263 .optional()
264 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to read key"))
265 }
266
267 fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Self::Error> {
268 let conn = self.connection()?;
269 conn.execute(UPSERT_SQL, params![key, value])
270 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to write key"))?;
271 Ok(())
272 }
273
274 fn delete(&self, key: &[u8]) -> Result<(), Self::Error> {
275 let conn = self.connection()?;
276 conn.execute(DELETE_SQL, params![key])
277 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to delete key"))?;
278 Ok(())
279 }
280
281 fn batch(&self, ops: &[BatchOp]) -> Result<(), Self::Error> {
282 let mut conn = self.connection()?;
283 let tx = conn
284 .transaction()
285 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to start transaction"))?;
286
287 {
288 let mut upsert = tx
289 .prepare_cached(UPSERT_SQL)
290 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to prepare batch write"))?;
291 let mut delete = tx
292 .prepare_cached(DELETE_SQL)
293 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to prepare batch delete"))?;
294
295 for op in ops {
296 match op {
297 BatchOp::Upsert { key, value } => {
298 upsert.execute(params![key, value]).map_err(|e| {
299 SqliteStoreError::from_sqlite(e, "Failed to write key in batch")
300 })?;
301 }
302 BatchOp::Delete { key } => {
303 delete.execute(params![key]).map_err(|e| {
304 SqliteStoreError::from_sqlite(e, "Failed to delete key in batch")
305 })?;
306 }
307 }
308 }
309 }
310
311 tx.commit()
312 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to commit transaction"))
313 }
314
315 fn batch_get(&self, keys: &[&[u8]]) -> Result<HashMap<Vec<u8>, Vec<u8>>, Self::Error> {
316 let conn = self.connection()?;
317 let mut stmt = conn
318 .prepare_cached(SELECT_SQL)
319 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to prepare batch read"))?;
320 let plan = OrderedBatchReadPlan::new(keys);
321 let mut results = HashMap::with_capacity(plan.unique_keys().len());
322
323 for key in plan.unique_keys() {
324 if let Some(value) = stmt
325 .query_row(params![key], |row| row.get(0))
326 .optional()
327 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to read key in batch"))?
328 {
329 results.insert(key.to_vec(), value);
330 }
331 }
332
333 Ok(results)
334 }
335
336 fn batch_get_ordered(&self, keys: &[&[u8]]) -> Result<Vec<Option<Vec<u8>>>, Self::Error> {
337 let conn = self.connection()?;
338 let mut stmt = conn.prepare_cached(SELECT_SQL).map_err(|e| {
339 SqliteStoreError::from_sqlite(e, "Failed to prepare ordered batch read")
340 })?;
341 let plan = OrderedBatchReadPlan::new(keys);
342 let mut unique_values = Vec::with_capacity(plan.unique_keys().len());
343
344 for key in plan.unique_keys() {
345 let value = stmt
346 .query_row(params![key], |row| row.get(0))
347 .optional()
348 .map_err(|e| {
349 SqliteStoreError::from_sqlite(e, "Failed to read key in ordered batch")
350 })?;
351 unique_values.push(value);
352 }
353
354 Ok(plan.expand_owned(unique_values))
355 }
356
357 fn batch_get_ordered_unique(
358 &self,
359 keys: &[&[u8]],
360 ) -> Result<Vec<Option<Vec<u8>>>, Self::Error> {
361 let conn = self.connection()?;
362 let mut stmt = conn.prepare_cached(SELECT_SQL).map_err(|e| {
363 SqliteStoreError::from_sqlite(e, "Failed to prepare unique ordered batch read")
364 })?;
365 let mut values = Vec::with_capacity(keys.len());
366
367 for key in keys {
368 let value = stmt
369 .query_row(params![key], |row| row.get(0))
370 .optional()
371 .map_err(|e| {
372 SqliteStoreError::from_sqlite(e, "Failed to read key in unique ordered batch")
373 })?;
374 values.push(value);
375 }
376
377 Ok(values)
378 }
379
380 fn batch_put(&self, entries: &[(&[u8], &[u8])]) -> Result<(), Self::Error> {
381 let mut conn = self.connection()?;
382 let tx = conn
383 .transaction()
384 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to start transaction"))?;
385
386 {
387 let mut stmt = tx.prepare_cached(UPSERT_SQL).map_err(|e| {
388 SqliteStoreError::from_sqlite(e, "Failed to prepare batch_put write")
389 })?;
390 for (key, value) in entries {
391 stmt.execute(params![key, value]).map_err(|e| {
392 SqliteStoreError::from_sqlite(e, "Failed to write key in batch_put")
393 })?;
394 }
395 }
396
397 tx.commit()
398 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to commit transaction"))
399 }
400
401 fn supports_hints(&self) -> bool {
402 true
403 }
404
405 fn get_hint(&self, namespace: &[u8], key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {
406 let conn = self.connection()?;
407 conn.query_row(
408 "SELECT value FROM prolly_hints WHERE namespace = ?1 AND key = ?2",
409 params![namespace, key],
410 |row| row.get(0),
411 )
412 .optional()
413 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to read hint"))
414 }
415
416 fn put_hint(&self, namespace: &[u8], key: &[u8], value: &[u8]) -> Result<(), Self::Error> {
417 let conn = self.connection()?;
418 conn.execute(
419 "\
420 INSERT INTO prolly_hints (namespace, key, value) \
421 VALUES (?1, ?2, ?3) \
422 ON CONFLICT(namespace, key) DO UPDATE SET value = excluded.value",
423 params![namespace, key, value],
424 )
425 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to write hint"))?;
426 Ok(())
427 }
428
429 fn batch_put_with_hint(
430 &self,
431 entries: &[(&[u8], &[u8])],
432 namespace: &[u8],
433 key: &[u8],
434 value: &[u8],
435 ) -> Result<(), Self::Error> {
436 let mut conn = self.connection()?;
437 let tx = conn
438 .transaction()
439 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to start transaction"))?;
440
441 {
442 let mut upsert_node = tx.prepare_cached(UPSERT_SQL).map_err(|e| {
443 SqliteStoreError::from_sqlite(e, "Failed to prepare batch_put write")
444 })?;
445 for (key, value) in entries {
446 upsert_node.execute(params![key, value]).map_err(|e| {
447 SqliteStoreError::from_sqlite(e, "Failed to write key in batch_put")
448 })?;
449 }
450 }
451
452 tx.execute(
453 "\
454 INSERT INTO prolly_hints (namespace, key, value) \
455 VALUES (?1, ?2, ?3) \
456 ON CONFLICT(namespace, key) DO UPDATE SET value = excluded.value",
457 params![namespace, key, value],
458 )
459 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to write hint in batch_put"))?;
460
461 tx.commit()
462 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to commit transaction"))
463 }
464}
465
466impl NodeStoreScan for SqliteStore {
467 type Error = SqliteStoreError;
468
469 fn list_node_cids(&self) -> Result<Vec<Cid>, Self::Error> {
470 let conn = self.connection()?;
471 let mut stmt = conn
472 .prepare_cached(SELECT_NODE_CIDS_SQL)
473 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to prepare node CID listing"))?;
474 let rows = stmt
475 .query_map([], |row| row.get::<_, Vec<u8>>(0))
476 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to list node CIDs"))?;
477
478 let mut cids = Vec::new();
479 for row in rows {
480 let key = row
481 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to read listed node CID"))?;
482 cids.push(cid_from_store_key(&key, "SQLite node").map_err(SqliteStoreError::new)?);
483 }
484 sort_cids(&mut cids);
485 Ok(cids)
486 }
487}
488
489impl ManifestStore for SqliteStore {
490 type Error = SqliteStoreError;
491
492 fn get_root(&self, name: &[u8]) -> Result<Option<RootManifest>, Self::Error> {
493 let conn = self.connection()?;
494 let bytes = conn
495 .query_row(SELECT_ROOT_SQL, params![name], |row| row.get(0))
496 .optional()
497 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to read root manifest"))?;
498 decode_root_manifest(bytes)
499 }
500
501 fn put_root(&self, name: &[u8], manifest: &RootManifest) -> Result<(), Self::Error> {
502 let conn = self.connection()?;
503 let bytes = encode_root_manifest(manifest)?;
504 conn.execute(UPSERT_ROOT_SQL, params![name, bytes])
505 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to write root manifest"))?;
506 Ok(())
507 }
508
509 fn delete_root(&self, name: &[u8]) -> Result<(), Self::Error> {
510 let conn = self.connection()?;
511 conn.execute(DELETE_ROOT_SQL, params![name])
512 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to delete root manifest"))?;
513 Ok(())
514 }
515
516 fn compare_and_swap_root(
517 &self,
518 name: &[u8],
519 expected: Option<&RootManifest>,
520 new: Option<&RootManifest>,
521 ) -> Result<ManifestUpdate, Self::Error> {
522 let expected_bytes = expected.map(encode_root_manifest).transpose()?;
523 let new_bytes = new.map(encode_root_manifest).transpose()?;
524
525 let mut conn = self.connection()?;
526 let tx = conn
527 .transaction()
528 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to start root transaction"))?;
529
530 let current_bytes = tx
531 .query_row(SELECT_ROOT_SQL, params![name], |row| row.get(0))
532 .optional()
533 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to read root manifest"))?;
534
535 if current_bytes.as_deref() != expected_bytes.as_deref() {
536 return Ok(ManifestUpdate::Conflict {
537 current: decode_root_manifest(current_bytes)?,
538 });
539 }
540
541 match new_bytes {
542 Some(bytes) => {
543 tx.execute(UPSERT_ROOT_SQL, params![name, bytes])
544 .map_err(|e| {
545 SqliteStoreError::from_sqlite(e, "Failed to write root manifest")
546 })?;
547 }
548 None => {
549 tx.execute(DELETE_ROOT_SQL, params![name]).map_err(|e| {
550 SqliteStoreError::from_sqlite(e, "Failed to delete root manifest")
551 })?;
552 }
553 }
554
555 tx.commit()
556 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to commit root transaction"))?;
557 Ok(ManifestUpdate::Applied)
558 }
559}
560
561impl ManifestStoreScan for SqliteStore {
562 fn list_roots(&self) -> Result<Vec<NamedRootManifest>, Self::Error> {
563 let conn = self.connection()?;
564 let mut stmt = conn.prepare_cached(SELECT_ROOTS_SQL).map_err(|e| {
565 SqliteStoreError::from_sqlite(e, "Failed to prepare root manifest listing")
566 })?;
567 let rows = stmt
568 .query_map([], |row| {
569 Ok((row.get::<_, Vec<u8>>(0)?, row.get::<_, Vec<u8>>(1)?))
570 })
571 .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to list root manifests"))?;
572
573 let mut roots = Vec::new();
574 for row in rows {
575 let (name, bytes) = row.map_err(|e| {
576 SqliteStoreError::from_sqlite(e, "Failed to read listed root manifest")
577 })?;
578 let manifest = RootManifest::from_bytes(&bytes)
579 .map_err(|err| SqliteStoreError::new(err.to_string()))?;
580 roots.push(NamedRootManifest::new(name, manifest));
581 }
582 sort_named_root_manifests(&mut roots);
583 Ok(roots)
584 }
585}
586
587impl TransactionalStore for SqliteStore {
588 fn supports_transactions(&self) -> bool {
589 true
590 }
591
592 fn commit_transaction(
593 &self,
594 node_writes: &[TransactionNodeWrite],
595 root_conditions: &[RootCondition],
596 root_writes: &[RootWrite],
597 ) -> Result<TransactionUpdate, Error> {
598 let mut conn = self
599 .connection()
600 .map_err(|err| Error::Store(Box::new(err)))?;
601 let tx = conn.transaction().map_err(|err| {
602 Error::Store(Box::new(SqliteStoreError::from_sqlite(
603 err,
604 "Failed to start transaction commit",
605 )))
606 })?;
607
608 for condition in root_conditions {
609 let current_bytes = tx
610 .query_row(SELECT_ROOT_SQL, params![condition.name], |row| row.get(0))
611 .optional()
612 .map_err(|err| {
613 Error::Store(Box::new(SqliteStoreError::from_sqlite(
614 err,
615 "Failed to read root manifest during transaction commit",
616 )))
617 })?;
618 let current =
619 decode_root_manifest(current_bytes).map_err(|err| Error::Store(Box::new(err)))?;
620 if current != condition.expected {
621 return Ok(TransactionUpdate::Conflict(TransactionConflict::new(
622 condition.name.clone(),
623 condition.expected.clone(),
624 current,
625 )));
626 }
627 }
628
629 {
630 let mut upsert_node = tx.prepare_cached(UPSERT_SQL).map_err(|err| {
631 Error::Store(Box::new(SqliteStoreError::from_sqlite(
632 err,
633 "Failed to prepare transaction node write",
634 )))
635 })?;
636 let mut delete_node = tx.prepare_cached(DELETE_SQL).map_err(|err| {
637 Error::Store(Box::new(SqliteStoreError::from_sqlite(
638 err,
639 "Failed to prepare transaction node delete",
640 )))
641 })?;
642 for write in node_writes {
643 match write {
644 TransactionNodeWrite::Upsert { key, value } => {
645 upsert_node.execute(params![key, value]).map_err(|err| {
646 Error::Store(Box::new(SqliteStoreError::from_sqlite(
647 err,
648 "Failed to write node during transaction commit",
649 )))
650 })?;
651 }
652 TransactionNodeWrite::Delete { key } => {
653 delete_node.execute(params![key]).map_err(|err| {
654 Error::Store(Box::new(SqliteStoreError::from_sqlite(
655 err,
656 "Failed to delete node during transaction commit",
657 )))
658 })?;
659 }
660 }
661 }
662 }
663
664 {
665 let mut upsert_root = tx.prepare_cached(UPSERT_ROOT_SQL).map_err(|err| {
666 Error::Store(Box::new(SqliteStoreError::from_sqlite(
667 err,
668 "Failed to prepare transaction root write",
669 )))
670 })?;
671 let mut delete_root = tx.prepare_cached(DELETE_ROOT_SQL).map_err(|err| {
672 Error::Store(Box::new(SqliteStoreError::from_sqlite(
673 err,
674 "Failed to prepare transaction root delete",
675 )))
676 })?;
677 for write in root_writes {
678 match write {
679 RootWrite::Put { name, manifest } => {
680 let bytes = encode_root_manifest(manifest)
681 .map_err(|err| Error::Store(Box::new(err)))?;
682 upsert_root.execute(params![name, bytes]).map_err(|err| {
683 Error::Store(Box::new(SqliteStoreError::from_sqlite(
684 err,
685 "Failed to write root during transaction commit",
686 )))
687 })?;
688 }
689 RootWrite::Delete { name } => {
690 delete_root.execute(params![name]).map_err(|err| {
691 Error::Store(Box::new(SqliteStoreError::from_sqlite(
692 err,
693 "Failed to delete root during transaction commit",
694 )))
695 })?;
696 }
697 }
698 }
699 }
700
701 tx.commit().map_err(|err| {
702 Error::Store(Box::new(SqliteStoreError::from_sqlite(
703 err,
704 "Failed to commit transaction",
705 )))
706 })?;
707 Ok(TransactionUpdate::Applied {
708 nodes_written: node_writes.len(),
709 roots_written: root_writes.len(),
710 })
711 }
712}
713
714fn encode_root_manifest(manifest: &RootManifest) -> Result<Vec<u8>, SqliteStoreError> {
715 manifest
716 .to_bytes()
717 .map_err(|e| SqliteStoreError::new(format!("failed to encode root manifest: {e}")))
718}
719
720fn decode_root_manifest(bytes: Option<Vec<u8>>) -> Result<Option<RootManifest>, SqliteStoreError> {
721 bytes
722 .as_deref()
723 .map(RootManifest::from_bytes)
724 .transpose()
725 .map_err(|e| SqliteStoreError::new(format!("failed to decode root manifest: {e}")))
726}
727
728#[cfg(test)]
729mod tests {
730 use super::*;
731
732 #[test]
733 fn sqlite_store_put_get_delete() {
734 let store = SqliteStore::open_in_memory().unwrap();
735
736 store.put(b"key", b"value").unwrap();
737 assert_eq!(store.get(b"key").unwrap(), Some(b"value".to_vec()));
738
739 store.delete(b"key").unwrap();
740 assert_eq!(store.get(b"key").unwrap(), None);
741 }
742
743 #[test]
744 fn sqlite_store_batch_is_order_preserving_for_reads() {
745 let store = SqliteStore::open_in_memory().unwrap();
746 let ops = vec![
747 BatchOp::Upsert {
748 key: b"a",
749 value: b"1",
750 },
751 BatchOp::Upsert {
752 key: b"b",
753 value: b"2",
754 },
755 BatchOp::Upsert {
756 key: b"c",
757 value: b"3",
758 },
759 ];
760
761 store.batch(&ops).unwrap();
762
763 let keys: Vec<&[u8]> = vec![b"c", b"missing", b"a", b"c", b"missing", b"b"];
764 assert_eq!(
765 store.batch_get_ordered(&keys).unwrap(),
766 vec![
767 Some(b"3".to_vec()),
768 None,
769 Some(b"1".to_vec()),
770 Some(b"3".to_vec()),
771 None,
772 Some(b"2".to_vec())
773 ]
774 );
775 }
776
777 #[test]
778 fn sqlite_store_batch_put_updates_existing_keys() {
779 let store = SqliteStore::open_in_memory().unwrap();
780
781 store.put(b"a", b"old").unwrap();
782 store
783 .batch_put(&[(b"a".as_slice(), b"new".as_slice()), (b"b", b"2")])
784 .unwrap();
785
786 assert_eq!(store.get(b"a").unwrap(), Some(b"new".to_vec()));
787 assert_eq!(store.get(b"b").unwrap(), Some(b"2".to_vec()));
788 }
789
790 #[test]
791 fn sqlite_store_persists_hints_separately_from_nodes() {
792 let store = SqliteStore::open_in_memory().unwrap();
793
794 store.put_hint(b"rightmost", b"root", b"hint-v1").unwrap();
795 assert_eq!(
796 store.get_hint(b"rightmost", b"root").unwrap(),
797 Some(b"hint-v1".to_vec())
798 );
799 assert_eq!(store.get_hint(b"rightmost", b"missing").unwrap(), None);
800 assert_eq!(store.get(b"root").unwrap(), None);
801
802 store.put_hint(b"rightmost", b"root", b"hint-v2").unwrap();
803 assert_eq!(
804 store.get_hint(b"rightmost", b"root").unwrap(),
805 Some(b"hint-v2".to_vec())
806 );
807 }
808}