dynoxide/storage_backend/
rusqlite_impl.rs

1//! Native [`StorageBackend`] implementation backed by the rusqlite-typed
2//! [`Storage`].
3//!
4//! Each impl method is a thin async wrapper over the existing sync method on
5//! `Storage`: the body invokes the sync method synchronously and returns a
6//! ready future. No `spawn_blocking` is used, so today's behaviour of running
7//! rusqlite calls on the active executor thread is preserved exactly.
8//!
9//! # Load-bearing invariant: native futures never suspend
10//!
11//! Every method here resolves on the first poll (it runs synchronous work and
12//! returns `Ready`). The synchronous `Database` facade depends on this: it
13//! drives these futures with `block_on`, which is only safe to call from inside
14//! the tokio-based HTTP and MCP servers because an always-ready future never
15//! parks the worker thread. Do not introduce a real `.await` on external async
16//! I/O (a socket, a timer, a task spawn, a channel) into any method here
17//! without first moving the facade off `block_on`; otherwise the server worker
18//! threads will stall.
19
20use crate::errors::DynoxideError;
21use crate::storage::{
22    CreateTableMetadata, DatabaseInfo, QueryParams, ScanParams, Storage, StreamRecord,
23    TableMetadata, TableStats,
24};
25use crate::storage_backend::{BackendError, BaseItemRow, Clock, GsiItemRow, StorageBackend, error};
26use crate::types::Tag;
27
28/// Convert a [`DynoxideError`] into a [`BackendError`].
29///
30/// Storage's sync surface returns `Result<T, DynoxideError>`; the trait surface
31/// returns `Result<T, BackendError>`. The conversion preserves rusqlite error
32/// codes via [`error::from_rusqlite`]. A client-facing `ValidationException`
33/// (a backend method such as `set_tags` enforces the tag-count limit) is
34/// preserved as [`BackendError::Validation`] so the reverse conversion can
35/// restore its 400 envelope; any other variant falls through to
36/// [`BackendError::Other`] carrying the original `Display` output.
37fn dyno_to_backend(err: DynoxideError) -> BackendError {
38    match err {
39        DynoxideError::SqliteError(e) => error::from_rusqlite(e),
40        DynoxideError::ValidationException(msg) => BackendError::Validation(msg),
41        other => BackendError::Other(other.to_string()),
42    }
43}
44
45impl StorageBackend for Storage {
46    fn clock(&self) -> &dyn Clock {
47        Storage::clock(self)
48    }
49
50    async fn insert_table_metadata(&self, m: &CreateTableMetadata<'_>) -> Result<(), BackendError> {
51        Storage::insert_table_metadata(self, m).map_err(dyno_to_backend)
52    }
53
54    async fn get_table_metadata(
55        &self,
56        table_name: &str,
57    ) -> Result<Option<TableMetadata>, BackendError> {
58        // Counted at the trait boundary rather than inside `Storage`, because
59        // the figure that matters is how many times a caller asks. This backend
60        // answers most of them from its own cache; the wasm one crosses a bridge
61        // for every single call and caches nothing.
62        crate::bench_counters::record(&crate::bench_counters::METADATA_READS);
63        Storage::get_table_metadata(self, table_name).map_err(dyno_to_backend)
64    }
65
66    async fn delete_table_metadata(&self, table_name: &str) -> Result<bool, BackendError> {
67        Storage::delete_table_metadata(self, table_name).map_err(dyno_to_backend)
68    }
69
70    async fn update_table_metadata(
71        &self,
72        table_name: &str,
73        attribute_definitions: &str,
74        gsi_definitions: Option<&str>,
75    ) -> Result<(), BackendError> {
76        Storage::update_table_metadata(self, table_name, attribute_definitions, gsi_definitions)
77            .map_err(dyno_to_backend)
78    }
79
80    async fn update_vector_index_definitions(
81        &self,
82        table_name: &str,
83        vector_index_definitions: Option<&str>,
84    ) -> Result<(), BackendError> {
85        Storage::update_vector_index_definitions(self, table_name, vector_index_definitions)
86            .map_err(dyno_to_backend)
87    }
88
89    async fn update_provisioned_throughput(
90        &self,
91        table_name: &str,
92        provisioned_throughput: &str,
93    ) -> Result<(), BackendError> {
94        Storage::update_provisioned_throughput(self, table_name, provisioned_throughput)
95            .map_err(dyno_to_backend)
96    }
97
98    async fn clear_provisioned_throughput(&self, table_name: &str) -> Result<(), BackendError> {
99        Storage::clear_provisioned_throughput(self, table_name).map_err(dyno_to_backend)
100    }
101
102    async fn update_billing_mode(
103        &self,
104        table_name: &str,
105        billing_mode: &str,
106    ) -> Result<(), BackendError> {
107        Storage::update_billing_mode(self, table_name, billing_mode).map_err(dyno_to_backend)
108    }
109
110    async fn update_table_class(
111        &self,
112        table_name: &str,
113        table_class: &str,
114    ) -> Result<(), BackendError> {
115        Storage::update_table_class(self, table_name, table_class).map_err(dyno_to_backend)
116    }
117
118    async fn update_on_demand_throughput(
119        &self,
120        table_name: &str,
121        on_demand_throughput: &str,
122    ) -> Result<(), BackendError> {
123        Storage::update_on_demand_throughput(self, table_name, on_demand_throughput)
124            .map_err(dyno_to_backend)
125    }
126
127    async fn clear_on_demand_throughput(&self, table_name: &str) -> Result<(), BackendError> {
128        Storage::clear_on_demand_throughput(self, table_name).map_err(dyno_to_backend)
129    }
130
131    async fn get_tags(&self, table_name: &str) -> Result<Vec<Tag>, BackendError> {
132        Storage::get_tags(self, table_name).map_err(dyno_to_backend)
133    }
134
135    async fn set_tags(&self, table_name: &str, new_tags: &[Tag]) -> Result<(), BackendError> {
136        Storage::set_tags(self, table_name, new_tags).map_err(dyno_to_backend)
137    }
138
139    async fn update_deletion_protection(
140        &self,
141        table_name: &str,
142        enabled: bool,
143    ) -> Result<(), BackendError> {
144        Storage::update_deletion_protection(self, table_name, enabled).map_err(dyno_to_backend)
145    }
146
147    async fn remove_tags(&self, table_name: &str, keys: &[String]) -> Result<(), BackendError> {
148        Storage::remove_tags(self, table_name, keys).map_err(dyno_to_backend)
149    }
150
151    async fn list_table_names(&self) -> Result<Vec<String>, BackendError> {
152        Storage::list_table_names(self).map_err(dyno_to_backend)
153    }
154
155    async fn table_exists(&self, table_name: &str) -> Result<bool, BackendError> {
156        Storage::table_exists(self, table_name).map_err(dyno_to_backend)
157    }
158
159    async fn create_data_table(&self, table_name: &str) -> Result<(), BackendError> {
160        Storage::create_data_table(self, table_name).map_err(dyno_to_backend)
161    }
162
163    async fn drop_data_table(&self, table_name: &str) -> Result<(), BackendError> {
164        Storage::drop_data_table(self, table_name).map_err(dyno_to_backend)
165    }
166
167    async fn create_gsi_table(
168        &self,
169        table_name: &str,
170        index_name: &str,
171    ) -> Result<(), BackendError> {
172        Storage::create_gsi_table(self, table_name, index_name).map_err(dyno_to_backend)
173    }
174
175    async fn drop_gsi_table(&self, table_name: &str, index_name: &str) -> Result<(), BackendError> {
176        Storage::drop_gsi_table(self, table_name, index_name).map_err(dyno_to_backend)
177    }
178
179    async fn create_lsi_table(
180        &self,
181        table_name: &str,
182        index_name: &str,
183    ) -> Result<(), BackendError> {
184        Storage::create_lsi_table(self, table_name, index_name).map_err(dyno_to_backend)
185    }
186
187    async fn drop_lsi_table(&self, table_name: &str, index_name: &str) -> Result<(), BackendError> {
188        Storage::drop_lsi_table(self, table_name, index_name).map_err(dyno_to_backend)
189    }
190
191    async fn create_vector_table(
192        &self,
193        table_name: &str,
194        index_name: &str,
195    ) -> Result<(), BackendError> {
196        Storage::create_vector_table(self, table_name, index_name).map_err(dyno_to_backend)
197    }
198
199    async fn drop_vector_table(
200        &self,
201        table_name: &str,
202        index_name: &str,
203    ) -> Result<(), BackendError> {
204        Storage::drop_vector_table(self, table_name, index_name).map_err(dyno_to_backend)
205    }
206
207    async fn insert_vector_items(
208        &self,
209        table_name: &str,
210        index_name: &str,
211        rows: &[crate::storage_backend::VectorItemRow],
212    ) -> Result<(), BackendError> {
213        Storage::insert_vector_items(self, table_name, index_name, rows).map_err(dyno_to_backend)
214    }
215
216    async fn delete_vector_item(
217        &self,
218        table_name: &str,
219        index_name: &str,
220        table_pk: &str,
221        table_sk: &str,
222    ) -> Result<(), BackendError> {
223        Storage::delete_vector_item(self, table_name, index_name, table_pk, table_sk)
224            .map_err(dyno_to_backend)
225    }
226
227    async fn query_vector_candidates(
228        &self,
229        table_name: &str,
230        index_name: &str,
231        hash_value: Option<&str>,
232    ) -> Result<Vec<crate::storage_backend::VectorCandidateRow>, BackendError> {
233        Storage::query_vector_candidates(self, table_name, index_name, hash_value)
234            .map_err(dyno_to_backend)
235    }
236
237    async fn vector_items_for_keys(
238        &self,
239        table_name: &str,
240        index_name: &str,
241        keys: &[(String, String)],
242    ) -> Result<Vec<(String, String, String)>, BackendError> {
243        Storage::vector_items_for_keys(self, table_name, index_name, keys).map_err(dyno_to_backend)
244    }
245
246    async fn insert_gsi_item(
247        &self,
248        table_name: &str,
249        index_name: &str,
250        gsi_pk: &str,
251        gsi_sk: &str,
252        table_pk: &str,
253        table_sk: &str,
254        item_json: &str,
255    ) -> Result<(), BackendError> {
256        #[cfg(test)]
257        self.check_gsi_insert_fault().map_err(dyno_to_backend)?;
258        Storage::insert_gsi_item(
259            self, table_name, index_name, gsi_pk, gsi_sk, table_pk, table_sk, item_json,
260        )
261        .map_err(dyno_to_backend)
262    }
263
264    async fn insert_gsi_items(
265        &self,
266        table_name: &str,
267        index_name: &str,
268        rows: &[GsiItemRow],
269    ) -> Result<(), BackendError> {
270        Storage::insert_gsi_items(self, table_name, index_name, rows).map_err(dyno_to_backend)
271    }
272
273    async fn delete_gsi_item(
274        &self,
275        table_name: &str,
276        index_name: &str,
277        table_pk: &str,
278        table_sk: &str,
279    ) -> Result<(), BackendError> {
280        Storage::delete_gsi_item(self, table_name, index_name, table_pk, table_sk)
281            .map_err(dyno_to_backend)
282    }
283
284    async fn query_gsi_items(
285        &self,
286        table_name: &str,
287        index_name: &str,
288        gsi_pk: &str,
289        params: &QueryParams<'_>,
290    ) -> Result<Vec<(String, String, String)>, BackendError> {
291        Storage::query_gsi_items(self, table_name, index_name, gsi_pk, params)
292            .map_err(dyno_to_backend)
293    }
294
295    async fn scan_gsi_items(
296        &self,
297        table_name: &str,
298        index_name: &str,
299        params: &ScanParams<'_>,
300    ) -> Result<Vec<(String, String, String)>, BackendError> {
301        Storage::scan_gsi_items(self, table_name, index_name, params).map_err(dyno_to_backend)
302    }
303
304    async fn insert_lsi_item(
305        &self,
306        table_name: &str,
307        index_name: &str,
308        pk: &str,
309        sk: &str,
310        base_pk: &str,
311        base_sk: &str,
312        item_json: &str,
313    ) -> Result<(), BackendError> {
314        Storage::insert_lsi_item(
315            self, table_name, index_name, pk, sk, base_pk, base_sk, item_json,
316        )
317        .map_err(dyno_to_backend)
318    }
319
320    async fn delete_lsi_item(
321        &self,
322        table_name: &str,
323        index_name: &str,
324        base_pk: &str,
325        base_sk: &str,
326    ) -> Result<(), BackendError> {
327        Storage::delete_lsi_item(self, table_name, index_name, base_pk, base_sk)
328            .map_err(dyno_to_backend)
329    }
330
331    async fn query_lsi_items(
332        &self,
333        table_name: &str,
334        index_name: &str,
335        pk: &str,
336        params: &QueryParams<'_>,
337    ) -> Result<Vec<(String, String, String)>, BackendError> {
338        Storage::query_lsi_items(self, table_name, index_name, pk, params).map_err(dyno_to_backend)
339    }
340
341    async fn scan_lsi_items(
342        &self,
343        table_name: &str,
344        index_name: &str,
345        params: &ScanParams<'_>,
346    ) -> Result<Vec<(String, String, String)>, BackendError> {
347        Storage::scan_lsi_items(self, table_name, index_name, params).map_err(dyno_to_backend)
348    }
349
350    async fn begin_transaction(&self) -> Result<(), BackendError> {
351        Storage::begin_transaction(self).map_err(dyno_to_backend)
352    }
353
354    async fn commit(&self) -> Result<(), BackendError> {
355        Storage::commit(self).map_err(dyno_to_backend)
356    }
357
358    async fn rollback(&self) -> Result<(), BackendError> {
359        Storage::rollback(self).map_err(dyno_to_backend)
360    }
361
362    async fn enable_bulk_loading(&self) -> Result<(), BackendError> {
363        Storage::enable_bulk_loading(self).map_err(dyno_to_backend)
364    }
365
366    async fn disable_bulk_loading(&self) -> Result<(), BackendError> {
367        Storage::disable_bulk_loading(self).map_err(dyno_to_backend)
368    }
369
370    async fn put_item(
371        &self,
372        table_name: &str,
373        pk: &str,
374        sk: &str,
375        item_json: &str,
376        item_size: usize,
377    ) -> Result<Option<String>, BackendError> {
378        Storage::put_item(self, table_name, pk, sk, item_json, item_size).map_err(dyno_to_backend)
379    }
380
381    async fn put_item_with_hash(
382        &self,
383        table_name: &str,
384        pk: &str,
385        sk: &str,
386        item_json: &str,
387        item_size: usize,
388        hash_prefix: &str,
389    ) -> Result<Option<String>, BackendError> {
390        Storage::put_item_with_hash(self, table_name, pk, sk, item_json, item_size, hash_prefix)
391            .map_err(dyno_to_backend)
392    }
393
394    async fn put_base_items(
395        &self,
396        table_name: &str,
397        rows: &[BaseItemRow],
398    ) -> Result<(), BackendError> {
399        Storage::put_base_items(self, table_name, rows).map_err(dyno_to_backend)
400    }
401
402    async fn get_item(
403        &self,
404        table_name: &str,
405        pk: &str,
406        sk: &str,
407    ) -> Result<Option<String>, BackendError> {
408        Storage::get_item(self, table_name, pk, sk).map_err(dyno_to_backend)
409    }
410
411    async fn get_partition_size(&self, table_name: &str, pk: &str) -> Result<i64, BackendError> {
412        Storage::get_partition_size(self, table_name, pk).map_err(dyno_to_backend)
413    }
414
415    async fn get_lsi_partition_size(
416        &self,
417        table_name: &str,
418        index_name: &str,
419        pk: &str,
420    ) -> Result<i64, BackendError> {
421        Storage::get_lsi_partition_size(self, table_name, index_name, pk).map_err(dyno_to_backend)
422    }
423
424    async fn delete_item(
425        &self,
426        table_name: &str,
427        pk: &str,
428        sk: &str,
429    ) -> Result<Option<String>, BackendError> {
430        Storage::delete_item(self, table_name, pk, sk).map_err(dyno_to_backend)
431    }
432
433    async fn query_items(
434        &self,
435        table_name: &str,
436        pk: &str,
437        params: &QueryParams<'_>,
438    ) -> Result<Vec<(String, String, String)>, BackendError> {
439        Storage::query_items(self, table_name, pk, params).map_err(dyno_to_backend)
440    }
441
442    async fn scan_items(
443        &self,
444        table_name: &str,
445        params: &ScanParams<'_>,
446    ) -> Result<Vec<(String, String, String)>, BackendError> {
447        Storage::scan_items(self, table_name, params).map_err(dyno_to_backend)
448    }
449
450    async fn count_items(&self, table_name: &str) -> Result<i64, BackendError> {
451        Storage::count_items(self, table_name).map_err(dyno_to_backend)
452    }
453
454    async fn db_size_bytes(&self) -> Result<u64, BackendError> {
455        Storage::db_size_bytes(self).map_err(dyno_to_backend)
456    }
457
458    async fn table_count(&self) -> Result<usize, BackendError> {
459        Storage::table_count(self).map_err(dyno_to_backend)
460    }
461
462    async fn table_stats(&self) -> Result<Vec<TableStats>, BackendError> {
463        Storage::table_stats(self).map_err(dyno_to_backend)
464    }
465
466    async fn database_info(&self) -> Result<DatabaseInfo, BackendError> {
467        Storage::database_info(self).map_err(dyno_to_backend)
468    }
469
470    async fn vacuum(&self) -> Result<(), BackendError> {
471        Storage::vacuum(self).map_err(dyno_to_backend)
472    }
473
474    async fn enable_stream(
475        &self,
476        table_name: &str,
477        view_type: &str,
478        label: &str,
479    ) -> Result<(), BackendError> {
480        Storage::enable_stream(self, table_name, view_type, label).map_err(dyno_to_backend)
481    }
482
483    async fn disable_stream(&self, table_name: &str) -> Result<(), BackendError> {
484        Storage::disable_stream(self, table_name).map_err(dyno_to_backend)
485    }
486
487    async fn insert_stream_record(
488        &self,
489        table_name: &str,
490        event_name: &str,
491        keys_json: &str,
492        new_image: Option<&str>,
493        old_image: Option<&str>,
494        sequence_number: &str,
495        shard_id: &str,
496        created_at: i64,
497    ) -> Result<(), BackendError> {
498        Storage::insert_stream_record(
499            self,
500            table_name,
501            event_name,
502            keys_json,
503            new_image,
504            old_image,
505            sequence_number,
506            shard_id,
507            created_at,
508        )
509        .map_err(dyno_to_backend)
510    }
511
512    async fn insert_stream_record_with_identity(
513        &self,
514        table_name: &str,
515        event_name: &str,
516        keys_json: &str,
517        new_image: Option<&str>,
518        old_image: Option<&str>,
519        sequence_number: &str,
520        shard_id: &str,
521        created_at: i64,
522        user_identity: Option<&str>,
523    ) -> Result<(), BackendError> {
524        Storage::insert_stream_record_with_identity(
525            self,
526            table_name,
527            event_name,
528            keys_json,
529            new_image,
530            old_image,
531            sequence_number,
532            shard_id,
533            created_at,
534            user_identity,
535        )
536        .map_err(dyno_to_backend)
537    }
538
539    async fn next_stream_sequence_number(&self, table_name: &str) -> Result<i64, BackendError> {
540        Storage::next_stream_sequence_number(self, table_name).map_err(dyno_to_backend)
541    }
542
543    async fn get_stream_records(
544        &self,
545        table_name: &str,
546        shard_id: &str,
547        after_sequence: i64,
548        limit: usize,
549    ) -> Result<Vec<StreamRecord>, BackendError> {
550        Storage::get_stream_records(self, table_name, shard_id, after_sequence, limit)
551            .map_err(dyno_to_backend)
552    }
553
554    async fn list_stream_enabled_tables(&self) -> Result<Vec<TableMetadata>, BackendError> {
555        Storage::list_stream_enabled_tables(self).map_err(dyno_to_backend)
556    }
557
558    async fn update_ttl_config(
559        &self,
560        table_name: &str,
561        attribute_name: Option<&str>,
562        enabled: bool,
563    ) -> Result<(), BackendError> {
564        Storage::update_ttl_config(self, table_name, attribute_name, enabled)
565            .map_err(dyno_to_backend)
566    }
567
568    async fn list_ttl_enabled_tables(&self) -> Result<Vec<TableMetadata>, BackendError> {
569        Storage::list_ttl_enabled_tables(self).map_err(dyno_to_backend)
570    }
571
572    async fn get_shard_sequence_range(
573        &self,
574        table_name: &str,
575        shard_id: &str,
576    ) -> Result<(Option<String>, Option<String>), BackendError> {
577        Storage::get_shard_sequence_range(self, table_name, shard_id).map_err(dyno_to_backend)
578    }
579
580    async fn touch_cached_at(
581        &self,
582        table_name: &str,
583        pk: &str,
584        sk: &str,
585        timestamp: f64,
586    ) -> Result<(), BackendError> {
587        Storage::touch_cached_at(self, table_name, pk, sk, timestamp).map_err(dyno_to_backend)
588    }
589
590    async fn get_lru_items(
591        &self,
592        table_name: &str,
593        limit: usize,
594    ) -> Result<Vec<(String, String, i64)>, BackendError> {
595        Storage::get_lru_items(self, table_name, limit).map_err(dyno_to_backend)
596    }
597}
dynoxide/storage_backend/rusqlite_impl.rs

dynoxide/storage_backend/
rusqlite_impl.rs
