Allow reads in txn

examples/sync-demo/src/colnDocType.ts

@@ -93,7 +93,7 @@ export function colnDocType<
     state: ColnState,
     body: (tx: ColnFfiTransaction<Ffi>) => void
   ): ColnState => {
-    const tx = state.store.beginTransaction()
+    const tx = state.store.transaction()
     try {
       const typedTx = new ffi.Transaction(state.store, tx)
       body(typedTx)

packages/coln-js-runtime/package.json

@@ -16,7 +16,7 @@
     "copy-license": "cp -r ../../LICENSES LICENSES",
     "build": "wasm-bodge build",
     "test": "npm run test:unit && npm run test:basic-ir",
-    "test:unit": "tsx --test tests/id-resolution.test.ts",
+    "test:unit": "tsx --test tests/id-resolution.test.ts tests/store-access.test.ts",
     "test:basic-ir": "npm run typecheck:basic-ir && npm run test:basic-ir:runtime",
     "typecheck:basic-ir": "tsx --tsconfig tests/basic-ir/tsconfig.json --test tests/basic-ir/typecheck.ts",
     "test:basic-ir:runtime": "tsx --tsconfig tests/basic-ir/tsconfig.json --test \"tests/basic-ir/*.test.ts\""

packages/coln-js-runtime/src/rust/handles.rs

@@ -5,7 +5,7 @@
 use coln_flir_rs::ir;
 use coln_store::{
     commit::{chunk::Chunk, hash::CommitHash as StoreCommitHash},
-    store::Store,
+    store::{Store, read::StoreRead},
     table::RowId as StoreRowId,
     txn::{OwnedTransaction, RowHandle as StoreRowHandle},
 };
@@ -43,6 +43,20 @@ pub struct TransactionHandle {
     pending_handles: Vec<(StoreRowHandle, JsValue)>,
 }
 
+impl TransactionHandle {
+    fn read_tx(&self) -> Result<&OwnedTransaction, JsValue> {
+        self.tx
+            .as_ref()
+            .ok_or_else(|| js_error("txn has been committed"))
+    }
+
+    fn write_tx(&mut self) -> Result<&mut OwnedTransaction, JsValue> {
+        self.tx
+            .as_mut()
+            .ok_or_else(|| js_error("txn has been committed"))
+    }
+}
+
 /*
 This function turns something like
 {
@@ -76,12 +90,37 @@ fn resolve_value_id(js_value: &JsValue, row_id: RowId) -> Result<(), JsValue> {
     Ok(())
 }
 
+// TODO we might want to distinguish between read and write txns
+// This is tricky because read transaction also needs an OwnedTransaction, which
+// does not make much sense.
 #[wasm_bindgen]
 impl TransactionHandle {
+    // read methods
+    #[wasm_bindgen(js_name = scanTable)]
+    pub fn scan_table(&self, path: String) -> Result<Vec<RowView>, JsValue> {
+        let path = ir::Path::from(path);
+        let rows = self
+            .read_tx()?
+            .scan_table(&path)
+            .map(|rows| rows.map(RowView::from).collect::<Vec<_>>())
+            .unwrap_or_default();
+
+        Ok(rows)
+    }
+
+    #[wasm_bindgen(js_name = rowById)]
+    pub fn row_by_id(&self, path: String, row_id: RowRef) -> Result<Option<RowView>, JsValue> {
+        let path = ir::Path::from(path);
+        let row_id = StoreRowId::try_from(row_id).map_err(js_error)?;
+
+        Ok(self.read_tx()?.row_by_id(&path, row_id).map(RowView::from))
+    }
+
+    // Write methods
     pub fn add(&mut self, path: String, values: Vec<Value>) -> Result<JsValue, JsValue> {
         let path = ir::Path::from(path);
         let values = values.into_iter().map(|v| v.into()).collect::<Vec<_>>();
-        let handle = self.tx()?.add(&path, values).map_err(js_error)?;
+        let handle = self.write_tx()?.add(&path, values).map_err(js_error)?;
 
         let (tx_id, counter) = handle.pending_ids().map_err(js_error)?;
         let temp_id = Value::temp_id(tx_id, counter);
@@ -116,14 +155,12 @@ impl TransactionHandle {
             Err((err, store)) => {
                 self.recovered_store = Some(store);
                 Err(js_error(format!(
-                    "{err}; recover the store with TransactionHandle.takeStore()"
+                    "{err}; recover the store with the read/write handle.take_store()"
                 )))
             }
         }
     }
 
-    // TODO adjust this API to not use take_store to recover but return the store
-    // after committing
     #[wasm_bindgen(js_name = takeStore)]
     pub fn take_store(&mut self) -> Result<StoreHandle, JsValue> {
         if let Some(tx) = self.tx.take() {
@@ -175,6 +212,18 @@ impl StoreHandle {
         self.store()?.json_ir().map_err(js_error)
     }
 
+    pub fn transaction(&mut self) -> Result<TransactionHandle, JsValue> {
+        let (store, pending_chunks) = self.owned_store()?;
+        Ok(TransactionHandle {
+            tx: Some(store.into_transaction()),
+            recovered_store: None,
+            pending_chunks,
+
+            pending_handles: Vec::new(),
+        })
+    }
+
+    // TODO DUPLICATE CODE! Remove after redesigning the RW interface
     #[wasm_bindgen(js_name = scanTable)]
     pub fn scan_table(&self, path: String) -> Result<Vec<RowView>, JsValue> {
         let path = ir::Path::from(path);
@@ -187,6 +236,7 @@ impl StoreHandle {
         Ok(rows)
     }
 
+    // TODO DUPLICATE CODE! Remove after redesigning the RW interface
     #[wasm_bindgen(js_name = rowById)]
     pub fn row_by_id(&self, path: String, row_id: RowRef) -> Result<Option<RowView>, JsValue> {
         let path = ir::Path::from(path);
@@ -195,8 +245,19 @@ impl StoreHandle {
         Ok(self.store()?.row_by_id(&path, row_id).map(RowView::from))
     }
 
-    #[wasm_bindgen(js_name = beginTransaction)]
-    pub fn begin_transaction(&mut self) -> Result<TransactionHandle, JsValue> {
+    fn store(&self) -> Result<&Store, JsValue> {
+        match &self.state {
+            StoreHandleState::Uninitialized { .. } => {
+                Err(js_error("store handle has not been initialized"))
+            }
+            StoreHandleState::Ready { store, .. } => Ok(store),
+            StoreHandleState::Moved => Err(js_error(
+                "store handle has already been moved into a transaction",
+            )),
+        }
+    }
+
+    fn owned_store(&mut self) -> Result<(Store, Vec<Vec<u8>>), JsValue> {
         let state = std::mem::replace(&mut self.state, StoreHandleState::Moved);
         let (store, pending_chunks) = match state {
             StoreHandleState::Ready {
@@ -214,13 +275,69 @@ impl StoreHandle {
             }
         };
 
-        Ok(TransactionHandle {
-            tx: Some(store.into_transaction()),
-            recovered_store: None,
-            pending_chunks,
+        Ok((store, pending_chunks))
+    }
+}
 
-            pending_handles: Vec::new(),
-        })
+impl StoreHandle {
+    fn ready(store: Store) -> Self {
+        Self::ready_with_pending(store, Vec::new())
+    }
+
+    fn ready_with_pending(store: Store, pending_chunks: Vec<Vec<u8>>) -> Self {
+        Self {
+            state: StoreHandleState::Ready {
+                store: Box::new(store),
+                pending_chunks,
+            },
+        }
+    }
+
+    // TODO this function is doing causal order delivery. This logic should NOT
+    // be here, and should be moved to somewhere else in the future.
+    fn apply_chunks(&mut self, chunk_bytes: Vec<Vec<u8>>) -> Result<(), String> {
+        match &mut self.state {
+            StoreHandleState::Uninitialized { chunks, has_root } => {
+                let decoded = chunk_bytes
+                    .iter()
+                    .map(|bytes| Chunk::decode(bytes))
+                    .collect::<Result<Vec<_>, _>>()
+                    .map_err(|error| error.to_string())?;
+                let previous_len = chunks.len();
+                let previously_had_root = *has_root;
+                *has_root |= decoded.iter().any(Chunk::is_root);
+                chunks.extend(chunk_bytes);
+                if *has_root {
+                    match Store::try_from_commit_bytes(chunks.iter()) {
+                        Ok((store, pending)) => {
+                            self.state = StoreHandle::ready_with_pending(store, pending).state
+                        }
+                        Err(error) => {
+                            chunks.truncate(previous_len);
+                            *has_root = previously_had_root;
+                            return Err(error.to_string());
+                        }
+                    }
+                }
+                Ok(())
+            }
+            StoreHandleState::Ready {
+                store,
+                pending_chunks,
+            } => {
+                pending_chunks.extend(chunk_bytes);
+                match store.apply_chunk_bytes(pending_chunks.iter().cloned()) {
+                    Ok(pending) => {
+                        *pending_chunks = pending;
+                        Ok(())
+                    }
+                    Err(error) => Err(error.to_string()),
+                }
+            }
+            StoreHandleState::Moved => {
+                Err("store handle has already been moved into a transaction".into())
+            }
+        }
     }
 }
 
@@ -292,89 +409,6 @@ impl CommitResult {
             .ok_or_else(|| js_error("commit result store has already been taken"))
     }
 }
-
-impl StoreHandle {
-    fn ready(store: Store) -> Self {
-        Self::ready_with_pending(store, Vec::new())
-    }
-
-    fn ready_with_pending(store: Store, pending_chunks: Vec<Vec<u8>>) -> Self {
-        Self {
-            state: StoreHandleState::Ready {
-                store: Box::new(store),
-                pending_chunks,
-            },
-        }
-    }
-
-    // TODO this function is doing causal order delivery. This logic should NOT
-    // be here, and should be moved to somewhere else in the future.
-    fn apply_chunks(&mut self, chunk_bytes: Vec<Vec<u8>>) -> Result<(), String> {
-        match &mut self.state {
-            StoreHandleState::Uninitialized { chunks, has_root } => {
-                let decoded = chunk_bytes
-                    .iter()
-                    .map(|bytes| Chunk::decode(bytes))
-                    .collect::<Result<Vec<_>, _>>()
-                    .map_err(|error| error.to_string())?;
-                let previous_len = chunks.len();
-                let previously_had_root = *has_root;
-                *has_root |= decoded.iter().any(Chunk::is_root);
-                chunks.extend(chunk_bytes);
-                if *has_root {
-                    match Store::try_from_commit_bytes(chunks.iter()) {
-                        Ok((store, pending)) => {
-                            self.state = StoreHandle::ready_with_pending(store, pending).state
-                        }
-                        Err(error) => {
-                            chunks.truncate(previous_len);
-                            *has_root = previously_had_root;
-                            return Err(error.to_string());
-                        }
-                    }
-                }
-                Ok(())
-            }
-            StoreHandleState::Ready {
-                store,
-                pending_chunks,
-            } => {
-                pending_chunks.extend(chunk_bytes);
-                match store.apply_chunk_bytes(pending_chunks.iter().cloned()) {
-                    Ok(pending) => {
-                        *pending_chunks = pending;
-                        Ok(())
-                    }
-                    Err(error) => Err(error.to_string()),
-                }
-            }
-            StoreHandleState::Moved => {
-                Err("store handle has already been moved into a transaction".into())
-            }
-        }
-    }
-
-    fn store(&self) -> Result<&Store, JsValue> {
-        match &self.state {
-            StoreHandleState::Uninitialized { .. } => {
-                Err(js_error("store handle has not been initialized"))
-            }
-            StoreHandleState::Ready { store, .. } => Ok(store),
-            StoreHandleState::Moved => Err(js_error(
-                "store handle has already been moved into a transaction",
-            )),
-        }
-    }
-}
-
-impl TransactionHandle {
-    fn tx(&mut self) -> Result<&mut OwnedTransaction, JsValue> {
-        self.tx
-            .as_mut()
-            .ok_or_else(|| js_error("transaction has already been committed"))
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use coln_flir_rs::ir::{
@@ -495,8 +529,12 @@ mod tests {
 
         let child = data.pop().expect("child commit");
         handle.apply_chunks(vec![child]).expect("buffer child");
-        let mut transaction = handle.begin_transaction().expect("begin transaction");
-        handle = transaction.take_store().expect("abort transaction");
+        let mut transaction = handle.transaction().expect("transaction");
+        handle = transaction
+            .commit()
+            .expect("empty commit")
+            .take_store()
+            .expect("store");
         handle.apply_chunks(data).expect("retry with parent");
 
         let table = handle
@@ -510,9 +548,9 @@ mod tests {
     #[test]
     fn active_transaction_can_return_its_store_without_committing() {
         let mut handle = StoreHandle::ready(source_store());
-        let mut transaction = handle.begin_transaction().expect("transaction");
+        let mut transaction = handle.transaction().expect("transaction");
         transaction
-            .tx()
+            .write_tx()
             .expect("owned transaction")
             .add(&Path::from("T"), vec![84_i64.into()])
             .expect("stage row");