Skip to content

Commit a0a132f

Browse files
devigneddevigned
committed
add tests for sqlite store
Signed-off-by: David Justice <[email protected]>
1 parent ba4f3e0 commit a0a132f

File tree

7 files changed

+291
-45
lines changed

7 files changed

+291
-45
lines changed

‎crates/factor-key-value/src/host.rs‎

Lines changed: 42 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,11 @@
1+
use super::Cas;
12
use anyhow::{Context, Result};
23
use spin_core::{async_trait, wasmtime::component::Resource};
34
use spin_resource_table::Table;
45
use spin_world::v2::key_value;
56
use spin_world::wasi::keyvalue as wasi_keyvalue;
67
use std::{collections::HashSet, sync::Arc};
78
use tracing::{instrument, Level};
8-
use super::Cas;
99

1010
const DEFAULT_STORE_TABLE_CAPACITY: u32 = 256;
1111

@@ -32,15 +32,14 @@ pub trait Store: Sync + Send {
3232
async fn delete(&self, key: &str) -> Result<(), Error>;
3333
async fn exists(&self, key: &str) -> Result<bool, Error>;
3434
async fn get_keys(&self) -> Result<Vec<String>, Error>;
35-
async fn get_many(&self, keys: Vec<String>) -> Result<Vec<Option<(String, Vec<u8>)>>, Error>;
35+
async fn get_many(&self, keys: Vec<String>) -> Result<Vec<(String, Option<Vec<u8>>)>, Error>;
3636
async fn set_many(&self, key_values: Vec<(String, Vec<u8>)>) -> Result<(), Error>;
3737
async fn delete_many(&self, keys: Vec<String>) -> Result<(), Error>;
3838
async fn increment(&self, key: String, delta: i64) -> Result<i64, Error>;
39-
async fn new_compare_and_swap(&self, key: &str) -> Result<Arc<dyn Cas>, Error>;
39+
async fn new_compare_and_swap(&self, bucket_rep: u32, key: &str)
40+
-> Result<Arc<dyn Cas>, Error>;
4041
}
4142

42-
43-
4443
pub struct KeyValueDispatch {
4544
allowed_stores: HashSet<String>,
4645
manager: Arc<dyn StoreManager>,
@@ -282,7 +281,7 @@ impl wasi_keyvalue::batch::Host for KeyValueDispatch {
282281
&mut self,
283282
bucket: Resource<wasi_keyvalue::batch::Bucket>,
284283
keys: Vec<String>,
285-
) -> std::result::Result<Vec<Option<(String, Vec<u8>)>>, wasi_keyvalue::store::Error> {
284+
) -> std::result::Result<Vec<(String, Option<Vec<u8>>)>, wasi_keyvalue::store::Error> {
286285
let store = self.get_store_wasi(bucket)?;
287286
store
288287
.get_many(keys.iter().map(|k| k.to_string()).collect())
@@ -321,11 +320,20 @@ impl wasi_keyvalue::atomics::HostCas for KeyValueDispatch {
321320
bucket: Resource<wasi_keyvalue::atomics::Bucket>,
322321
key: String,
323322
) -> Result<Resource<wasi_keyvalue::atomics::Cas>, wasi_keyvalue::store::Error> {
323+
let bucket_rep = bucket.rep();
324+
let bucket: Resource<Bucket> = Resource::new_own(bucket_rep);
324325
let store = self.get_store_wasi(bucket)?;
325-
let cas = store.new_compare_and_swap(&key).await.map_err(to_wasi_err)?;
326+
let cas = store
327+
.new_compare_and_swap(bucket_rep, &key)
328+
.await
329+
.map_err(to_wasi_err)?;
326330
self.compare_and_swaps
327331
.push(cas)
328-
.map_err(|()| spin_world::wasi::keyvalue::store::Error::Other("too many compare_and_swaps opened".to_string()))
332+
.map_err(|()| {
333+
spin_world::wasi::keyvalue::store::Error::Other(
334+
"too many compare_and_swaps opened".to_string(),
335+
)
336+
})
329337
.map(Resource::new_own)
330338
}
331339

@@ -361,15 +369,32 @@ impl wasi_keyvalue::atomics::Host for KeyValueDispatch {
361369
#[instrument(name = "spin_key_value.swap", skip(self, cas_res, value), err(level = Level::INFO), fields(otel.kind = "client"))]
362370
async fn swap(
363371
&mut self,
364-
cas_res: Resource<wasi_keyvalue::atomics::Cas>,
372+
cas_res: Resource<atomics::Cas>,
365373
value: Vec<u8>,
366-
) -> Result<std::result::Result<(), wasi_keyvalue::atomics::CasError>> {
374+
) -> Result<std::result::Result<(), CasError>> {
375+
let cas_rep = cas_res.rep();
367376
let cas = self
368-
.get_cas(cas_res)
369-
.map_err(|e| wasi_keyvalue::atomics::CasError::StoreError(wasi_keyvalue::atomics::Error::Other(e.to_string())))?;
370-
Ok(cas.swap(value)
371-
.await
372-
.map_err(|e| wasi_keyvalue::atomics::CasError::StoreError(wasi_keyvalue::atomics::Error::Other(e.to_string()))))
377+
.get_cas(Resource::<Bucket>::new_own(cas_rep))
378+
.map_err(|e| CasError::StoreError(atomics::Error::Other(e.to_string())))?;
379+
380+
match cas.swap(value).await {
381+
Ok(cas) => Ok(Ok(())),
382+
Err(err) => {
383+
if err.to_string().contains("CAS_ERROR") {
384+
let bucket = Resource::new_own(cas.bucket_rep().await);
385+
let new_cas = self.new(bucket, cas.key().await).await?;
386+
let new_cas_rep = new_cas.rep();
387+
self.current(Resource::new_own(new_cas_rep)).await?;
388+
Err(anyhow::Error::new(CasError::CasFailed(Resource::new_own(
389+
new_cas_rep,
390+
))))
391+
} else {
392+
Err(anyhow::Error::new(CasError::StoreError(
393+
atomics::Error::Other(err.to_string()),
394+
)))
395+
}
396+
}
397+
}
373398
}
374399
}
375400

@@ -379,6 +404,8 @@ pub fn log_error(err: impl std::fmt::Debug) -> Error {
379404
}
380405

381406
use spin_world::v1::key_value::Error as LegacyError;
407+
use spin_world::wasi::keyvalue::atomics;
408+
use spin_world::wasi::keyvalue::atomics::{CasError, HostCas};
382409

383410
fn to_legacy_error(value: key_value::Error) -> LegacyError {
384411
match value {

‎crates/factor-key-value/src/lib.rs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -137,6 +137,8 @@ impl AppState {
137137
pub trait Cas: Sync + Send {
138138
async fn current(&self) -> anyhow::Result<Option<Vec<u8>>, Error>;
139139
async fn swap(&self, value: Vec<u8>) -> anyhow::Result<(), Error>;
140+
async fn bucket_rep(&self) -> u32;
141+
async fn key(&self) -> String;
140142
}
141143

142144
pub struct InstanceBuilder {

‎crates/factor-key-value/src/util.rs‎

Lines changed: 32 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -238,7 +238,33 @@ impl Store for CachingStore {
238238
.collect())
239239
}
240240

241-
async fn get_many(&self, keys: Vec<String>) -> anyhow::Result<Vec<Option<(String, Vec<u8>)>>, Error> {
241+
async fn get_many(
242+
&self,
243+
keys: Vec<String>,
244+
) -> anyhow::Result<Vec<(String, Option<Vec<u8>>)>, Error> {
245+
// // Retrieve the specified value from the cache, lazily populating the cache as necessary.
246+
// let mut state = self.state.lock().await;
247+
//
248+
// let mut keys_and_values: Vec<Option<(String, Vec<u8>)>> = Vec::new();
249+
// let mut keys_not_found: Vec<String> = Vec::new();
250+
// for key in keys {
251+
// match state.cache.get(key.as_str()).cloned() {
252+
// Some(value) => keys_and_values.push(Some((key, value))),
253+
// None => keys_not_found.push(key),
254+
// }
255+
// }
256+
//
257+
// // guarantee the guest will read its own writes even if entries have been popped off the end of the LRU
258+
// // cache prior to their corresponding writes reaching the backing store.
259+
// state.flush().await?;
260+
//
261+
// let value = self.inner.get(key).await?;
262+
//
263+
// state.cache.put(key.to_owned(), value.clone());
264+
//
265+
// Ok(value)
266+
//
267+
242268
todo!()
243269
}
244270

@@ -254,7 +280,11 @@ impl Store for CachingStore {
254280
todo!()
255281
}
256282

257-
async fn new_compare_and_swap(&self, key: &str) -> anyhow::Result<Arc<dyn Cas>, Error> {
283+
async fn new_compare_and_swap(
284+
&self,
285+
bucket_rep: u32,
286+
key: &str,
287+
) -> anyhow::Result<Arc<dyn Cas>, Error> {
258288
todo!()
259289
}
260290
}

‎crates/key-value-azure/src/store.rs‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -152,7 +152,7 @@ impl Store for AzureCosmosStore {
152152
self.get_keys().await
153153
}
154154

155-
async fn get_many(&self, keys: Vec<String>) -> Result<Vec<Option<(String, Vec<u8>)>>, Error> {
155+
async fn get_many(&self, keys: Vec<String>) -> Result<Vec<(String, Option<Vec<u8>>)>, Error> {
156156
todo!()
157157
}
158158

@@ -168,7 +168,11 @@ impl Store for AzureCosmosStore {
168168
todo!()
169169
}
170170

171-
async fn new_compare_and_swap(&self, key: &str) -> Result<Arc<dyn spin_factor_key_value::Cas>, Error> {
171+
async fn new_compare_and_swap(
172+
&self,
173+
bucket_rep: u32,
174+
key: &str,
175+
) -> Result<Arc<dyn spin_factor_key_value::Cas>, Error> {
172176
todo!()
173177
}
174178
}

‎crates/key-value-redis/src/store.rs‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -99,7 +99,7 @@ impl Store for RedisStore {
9999
.map_err(log_error)
100100
}
101101

102-
async fn get_many(&self, keys: Vec<String>) -> Result<Vec<Option<(String, Vec<u8>)>>, Error> {
102+
async fn get_many(&self, keys: Vec<String>) -> Result<Vec<(String, Option<Vec<u8>>)>, Error> {
103103
todo!()
104104
}
105105

@@ -115,7 +115,11 @@ impl Store for RedisStore {
115115
todo!()
116116
}
117117

118-
async fn new_compare_and_swap(&self, key: &str) -> Result<Arc<dyn spin_factor_key_value::Cas>, Error> {
118+
async fn new_compare_and_swap(
119+
&self,
120+
bucket_rep: u32,
121+
key: &str,
122+
) -> Result<Arc<dyn spin_factor_key_value::Cas>, Error> {
119123
todo!()
120124
}
121125
}

0 commit comments

Comments
 (0)