diff --git a/packages/coln-js-runtime/src/rust/handles.rs b/packages/coln-js-runtime/src/rust/handles.rs
index fa475898..6446a8f6 100644
--- a/packages/coln-js-runtime/src/rust/handles.rs
+++ b/packages/coln-js-runtime/src/rust/handles.rs
@@ -4,7 +4,7 @@
 
 use coln_flir_rs::ir;
 use coln_store::{
-    commit::hash::CommitHash as StoreCommitHash,
+    commit::{chunk::Chunk, hash::CommitHash as StoreCommitHash},
     store::Store,
     table::RowId as StoreRowId,
     txn::{OwnedTransaction, RowHandle as StoreRowHandle},
@@ -19,13 +19,26 @@ use wasm_bindgen::prelude::wasm_bindgen;
 
 #[wasm_bindgen]
 pub struct StoreHandle {
-    store: Option<Store>,
+    state: StoreHandleState,
+}
+
+enum StoreHandleState {
+    Uninitialized {
+        chunks: Vec<Vec<u8>>,
+        has_root: bool,
+    },
+    Ready {
+        store: Box<Store>,
+        pending_chunks: Vec<Vec<u8>>,
+    },
+    Moved,
 }
 
 #[wasm_bindgen]
 pub struct TransactionHandle {
     tx: Option<OwnedTransaction>,
     recovered_store: Option<Store>,
+    pending_chunks: Vec<Vec<u8>>,
 
     pending_handles: Vec<(StoreRowHandle, JsValue)>,
 }
@@ -94,7 +107,10 @@ impl TransactionHandle {
 
                 Ok(CommitResult {
                     commit: commit.to_string(),
-                    store: Some(StoreHandle { store: Some(store) }),
+                    store: Some(StoreHandle::ready_with_pending(
+                        store,
+                        std::mem::take(&mut self.pending_chunks),
+                    )),
                 })
             }
             Err((err, store)) => {
@@ -110,12 +126,21 @@ impl TransactionHandle {
     // after committing
     #[wasm_bindgen(js_name = takeStore)]
     pub fn take_store(&mut self) -> Result<StoreHandle, JsValue> {
+        if let Some(tx) = self.tx.take() {
+            return Ok(StoreHandle::ready_with_pending(
+                tx.abort(),
+                std::mem::take(&mut self.pending_chunks),
+            ));
+        }
         let store = self
             .recovered_store
             .take()
             .ok_or_else(|| js_error("transaction does not have a recovered store"))?;
 
-        Ok(StoreHandle { store: Some(store) })
+        Ok(StoreHandle::ready_with_pending(
+            store,
+            std::mem::take(&mut self.pending_chunks),
+        ))
     }
 }
 
@@ -127,13 +152,22 @@ pub struct CommitResult {
 
 #[wasm_bindgen]
 impl StoreHandle {
+    pub fn empty() -> StoreHandle {
+        Self {
+            state: StoreHandleState::Uninitialized {
+                chunks: Vec::new(),
+                has_root: false,
+            },
+        }
+    }
+
     #[wasm_bindgen(js_name = fromTheory)]
     pub fn from_theory(flat_theory_json: String) -> Result<StoreHandle, JsValue> {
         let theory = serde_json::from_str::<ir::FlatRealm>(&flat_theory_json)
             .map_err(|err| js_error(format!("invalid flat theory JSON: {err}")))?;
         let store = Store::try_from_ir(theory).map_err(js_error)?;
 
-        Ok(Self { store: Some(store) })
+        Ok(Self::ready(store))
     }
 
     #[wasm_bindgen(js_name = jsonIR)]
@@ -163,14 +197,27 @@ impl StoreHandle {
 
     #[wasm_bindgen(js_name = beginTransaction)]
     pub fn begin_transaction(&mut self) -> Result<TransactionHandle, JsValue> {
-        let store = self
-            .store
-            .take()
-            .ok_or_else(|| js_error("store handle has already been moved into a transaction"))?;
+        let state = std::mem::replace(&mut self.state, StoreHandleState::Moved);
+        let (store, pending_chunks) = match state {
+            StoreHandleState::Ready {
+                store,
+                pending_chunks,
+            } => (*store, pending_chunks),
+            state @ StoreHandleState::Uninitialized { .. } => {
+                self.state = state;
+                return Err(js_error("store handle has not been initialized"));
+            }
+            StoreHandleState::Moved => {
+                return Err(js_error(
+                    "store handle has already been moved into a transaction",
+                ));
+            }
+        };
 
         Ok(TransactionHandle {
             tx: Some(store.into_transaction()),
             recovered_store: None,
+            pending_chunks,
 
             pending_handles: Vec::new(),
         })
@@ -182,12 +229,19 @@ impl StoreHandle {
     // For automerge-repo interfacing
 
     pub fn heads(&self) -> Result<Vec<CommitHash>, JsValue> {
-        let heads = self
-            .store()?
-            .heads()
-            .into_iter()
-            .map(CommitHash::from)
-            .collect::<Vec<_>>();
+        let heads = match &self.state {
+            StoreHandleState::Uninitialized { .. } => return Ok(Vec::new()),
+            StoreHandleState::Ready { store, .. } => store,
+            StoreHandleState::Moved => {
+                return Err(js_error(
+                    "store handle has already been moved into a transaction",
+                ));
+            }
+        }
+        .heads()
+        .into_iter()
+        .map(CommitHash::from)
+        .collect::<Vec<_>>();
 
         Ok(heads)
     }
@@ -197,6 +251,9 @@ impl StoreHandle {
         &self,
         have_heads: Vec<CommitHash>,
     ) -> Result<Vec<CommitChunk>, JsValue> {
+        if matches!(self.state, StoreHandleState::Uninitialized { .. }) {
+            return Ok(Vec::new());
+        }
         let have_heads = have_heads
             .into_iter()
             .map(StoreCommitHash::try_from)
@@ -217,10 +274,7 @@ impl StoreHandle {
     pub fn apply_chunk_bytes(&mut self, chunk_bytes: JsValue) -> Result<(), JsValue> {
         let chunk_bytes =
             serde_wasm_bindgen::from_value::<Vec<Vec<u8>>>(chunk_bytes).map_err(js_error)?;
-
-        self.store_mut()?
-            .apply_chunk_bytes(chunk_bytes)
-            .map_err(js_error)
+        self.apply_chunks(chunk_bytes).map_err(js_error)
     }
 }
 
@@ -240,16 +294,76 @@ impl CommitResult {
 }
 
 impl StoreHandle {
-    fn store(&self) -> Result<&Store, JsValue> {
-        self.store
-            .as_ref()
-            .ok_or_else(|| js_error("store handle has been moved into a transaction"))
+    fn ready(store: Store) -> Self {
+        Self::ready_with_pending(store, Vec::new())
     }
 
-    fn store_mut(&mut self) -> Result<&mut Store, JsValue> {
-        self.store
-            .as_mut()
-            .ok_or_else(|| js_error("store handle has been moved into a transaction"))
+    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",
+            )),
+        }
     }
 }
 
@@ -260,3 +374,184 @@ impl TransactionHandle {
             .ok_or_else(|| js_error("transaction has already been committed"))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use coln_flir_rs::ir::{
+        BuiltinTy, ColType, ColumnEntry, EntityVariant, FlatRealm, Path, Schema, TableEntry,
+    };
+
+    use super::*;
+
+    fn source_store() -> Store {
+        let theory = FlatRealm {
+            tables: vec![TableEntry {
+                path: Path::from("T"),
+                table: Schema {
+                    entity_variant: EntityVariant::Table,
+                    columns: vec![ColumnEntry {
+                        path: Path::from("value"),
+                        col_type: ColType::BuiltinTy {
+                            builtin_ty: BuiltinTy::BuiltinInt,
+                        },
+                    }],
+                    primary_key: None,
+                },
+            }],
+            rules: vec![],
+        };
+        let mut store = Store::try_from_ir(theory).expect("store");
+        let mut transaction = store.transaction();
+        transaction
+            .add(&Path::from("T"), vec![42_i64.into()])
+            .expect("add row");
+        transaction.commit().expect("commit");
+        store
+    }
+
+    #[test]
+    fn empty_handle_buffers_data_until_root_arrives() {
+        let source = source_store();
+        let (root, data): (Vec<_>, Vec<_>) = source
+            .commit_chunks_after(&[])
+            .into_iter()
+            .map(|chunk| chunk.bytes)
+            .partition(|bytes| Chunk::decode(bytes).expect("chunk").is_root());
+        let mut handle = StoreHandle::empty();
+
+        handle.apply_chunks(data).expect("buffer data");
+        assert!(matches!(
+            handle.state,
+            StoreHandleState::Uninitialized { .. }
+        ));
+        assert!(handle.heads().expect("heads").is_empty());
+
+        handle.apply_chunks(root).expect("apply root");
+        let store = handle.store().expect("initialized store");
+        let table = store.table_at(&Path::from("T")).expect("table");
+        assert_eq!(table.row_count(), 1);
+    }
+
+    #[test]
+    fn empty_handle_retries_bootstrap_when_missing_parent_arrives() {
+        let mut source = source_store();
+        let mut transaction = source.transaction();
+        transaction
+            .add(&Path::from("T"), vec![84_i64.into()])
+            .expect("add second row");
+        transaction.commit().expect("second commit");
+
+        let mut root = None;
+        let mut data = Vec::new();
+        for chunk in source.commit_chunks_after(&[]) {
+            if Chunk::decode(&chunk.bytes).expect("chunk").is_root() {
+                root = Some(chunk.bytes);
+            } else {
+                data.push(chunk.bytes);
+            }
+        }
+        assert_eq!(data.len(), 2);
+
+        let mut handle = StoreHandle::empty();
+        let child = data.pop().expect("child commit");
+        handle
+            .apply_chunks(vec![root.expect("root"), child])
+            .expect("buffer child");
+
+        handle.apply_chunks(data).expect("retry with parent");
+        let table = handle
+            .store()
+            .expect("initialized store")
+            .table_at(&Path::from("T"))
+            .expect("table");
+        assert_eq!(table.row_count(), 2);
+    }
+
+    #[test]
+    fn ready_handle_retries_commit_when_missing_parent_arrives() {
+        let mut source = source_store();
+        let mut transaction = source.transaction();
+        transaction
+            .add(&Path::from("T"), vec![84_i64.into()])
+            .expect("add second row");
+        transaction.commit().expect("second commit");
+
+        let mut root = None;
+        let mut data = Vec::new();
+        for chunk in source.commit_chunks_after(&[]) {
+            if Chunk::decode(&chunk.bytes).expect("chunk").is_root() {
+                root = Some(chunk.bytes);
+            } else {
+                data.push(chunk.bytes);
+            }
+        }
+        assert_eq!(data.len(), 2);
+
+        let mut handle = StoreHandle::empty();
+        handle
+            .apply_chunks(vec![root.expect("root")])
+            .expect("apply root");
+        assert!(matches!(handle.state, StoreHandleState::Ready { .. }));
+
+        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");
+        handle.apply_chunks(data).expect("retry with parent");
+
+        let table = handle
+            .store()
+            .expect("initialized store")
+            .table_at(&Path::from("T"))
+            .expect("table");
+        assert_eq!(table.row_count(), 2);
+    }
+
+    #[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");
+        transaction
+            .tx()
+            .expect("owned transaction")
+            .add(&Path::from("T"), vec![84_i64.into()])
+            .expect("stage row");
+
+        let recovered = transaction.take_store().expect("recover store");
+        let table = recovered
+            .store()
+            .expect("store")
+            .table_at(&Path::from("T"))
+            .expect("table");
+        assert_eq!(table.row_count(), 1);
+    }
+
+    #[test]
+    fn malformed_batch_does_not_poison_later_bootstrap() {
+        let source = source_store();
+        let chunks = source
+            .commit_chunks_after(&[])
+            .into_iter()
+            .map(|chunk| chunk.bytes)
+            .collect::<Vec<_>>();
+        let root = chunks
+            .iter()
+            .find(|bytes| Chunk::decode(bytes).expect("chunk").is_root())
+            .expect("root")
+            .clone();
+        let mut handle = StoreHandle::empty();
+
+        assert!(handle.apply_chunks(vec![root, vec![0xff]]).is_err());
+        handle.apply_chunks(chunks).expect("valid retry");
+
+        assert_eq!(
+            handle
+                .store()
+                .expect("initialized store")
+                .table_at(&Path::from("T"))
+                .expect("table")
+                .row_count(),
+            1
+        );
+    }
+}
diff --git a/packages/coln-js-runtime/src/ts/RealmBindings.ts b/packages/coln-js-runtime/src/ts/RealmBindings.ts
new file mode 100644
index 00000000..88516c95
--- /dev/null
+++ b/packages/coln-js-runtime/src/ts/RealmBindings.ts
@@ -0,0 +1,16 @@
+// SPDX-FileCopyrightText: 2026 Coln contributors
+//
+// SPDX-License-Identifier: Apache-2.0 OR MIT
+
+import type { StoreHandle, TransactionHandle } from "#wasm-bodge/bindings"
+
+export interface ColnSchema {
+  entities: readonly unknown[]
+  rules: readonly unknown[]
+}
+
+export interface RealmBindings<ViewRoot = unknown, TransactionRoot = unknown> {
+  schema: ColnSchema
+  View: new (store: StoreHandle) => { root: ViewRoot }
+  Transaction: new (store: StoreHandle, transaction: TransactionHandle) => { root: TransactionRoot }
+}
diff --git a/packages/coln-js-runtime/src/ts/index.ts b/packages/coln-js-runtime/src/ts/index.ts
index 9d936a44..19e74047 100644
--- a/packages/coln-js-runtime/src/ts/index.ts
+++ b/packages/coln-js-runtime/src/ts/index.ts
@@ -5,6 +5,8 @@
 export type { CommitChunk, RowRef, RowView, Value } from "#wasm-bodge/bindings";
 export { CommitResult, StoreHandle, TransactionHandle, valueEqual } from "#wasm-bodge/bindings"
 
+export type { RealmBindings, ColnSchema } from "./RealmBindings"
+
 export * as ColnSet from "./ColnSet";
 
 export * as ColnRef from "./ColnRef";
diff --git a/packages/coln-js-runtime/tests/basic-ir/helpers.ts b/packages/coln-js-runtime/tests/basic-ir/helpers.ts
index fb0aeb10..be9e2327 100644
--- a/packages/coln-js-runtime/tests/basic-ir/helpers.ts
+++ b/packages/coln-js-runtime/tests/basic-ir/helpers.ts
@@ -2,19 +2,10 @@
 //
 // SPDX-License-Identifier: Apache-2.0 OR MIT
 
-import { StoreHandle, type TransactionHandle } from "@coln-project/runtime";
-
-interface RealmModule<ViewRoot, TransactionRoot> {
-  schema: unknown;
-  View: new (store: StoreHandle) => { root: ViewRoot };
-  Transaction: new (
-    store: StoreHandle,
-    transaction: TransactionHandle,
-  ) => { root: TransactionRoot };
-}
+import { StoreHandle, type RealmBindings } from "@coln-project/runtime";
 
 export function beginRealm<ViewRoot, TransactionRoot>(
-  realm: RealmModule<ViewRoot, TransactionRoot>,
+  realm: RealmBindings<ViewRoot, TransactionRoot>,
 ) {
   let store = StoreHandle.fromTheory(JSON.stringify(realm.schema));
   const transaction = store.beginTransaction();
diff --git a/packages/coln-store/src/commit/chunk.rs b/packages/coln-store/src/commit/chunk.rs
index 39789cb4..3b8f368f 100644
--- a/packages/coln-store/src/commit/chunk.rs
+++ b/packages/coln-store/src/commit/chunk.rs
@@ -27,6 +27,8 @@ impl From<ChunkType> for u8 {
     }
 }
 
+/// A chunk what the external world sees as a "chunk" of data, right now it is
+/// just a commit, but in the future it might be a sedimemtree fragment.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum Chunk {
     Commit { header: Header, payload: Vec<u8> },
@@ -53,6 +55,10 @@ impl Chunk {
         Ok(chunk)
     }
 
+    pub fn is_root(&self) -> bool {
+        self.chunk_type() == ChunkType::Root
+    }
+
     pub(crate) fn chunk_type(&self) -> ChunkType {
         match self {
             Chunk::Commit { header, .. } | Chunk::Root { header, .. } => header.chunk_type,
diff --git a/packages/coln-store/src/commit/error.rs b/packages/coln-store/src/commit/error.rs
index a70a34bc..6a77d69f 100644
--- a/packages/coln-store/src/commit/error.rs
+++ b/packages/coln-store/src/commit/error.rs
@@ -13,6 +13,7 @@ pub enum CodecError {
     IOError(io::Error),
     SchemaError(String),
     DataFormatError(String),
+    DataContentError(String),
     ChecksumMismatch,
     DecodeError(hexane::PackError),
     ChunkMismatch { expected: ChunkType, got: ChunkType },
@@ -26,6 +27,7 @@ impl fmt::Display for CodecError {
             CodecError::IOError(err) => write!(f, "io error: {err}"),
             CodecError::SchemaError(msg) => write!(f, "schema error: {msg}"),
             CodecError::DataFormatError(msg) => write!(f, "data format error: {msg}"),
+            CodecError::DataContentError(msg) => write!(f, "data content error: {msg}"),
             CodecError::ChecksumMismatch => write!(f, "chunk checksum mismatch"),
             CodecError::DecodeError(err) => write!(f, "decode error: {err:?}"),
             CodecError::ChunkMismatch { expected, got } => write!(
diff --git a/packages/coln-store/src/commit/pst.rs b/packages/coln-store/src/commit/pst.rs
index 592ad05d..3c2e923e 100644
--- a/packages/coln-store/src/commit/pst.rs
+++ b/packages/coln-store/src/commit/pst.rs
@@ -5,7 +5,7 @@
 use std::io::Write;
 
 use crate::commit::Commit;
-use crate::commit::chunk::{Chunk, ChunkType};
+use crate::commit::chunk::Chunk;
 use crate::commit::error::CodecError;
 use crate::commit::leb128 as commit_leb128;
 use crate::store::Store;
@@ -38,7 +38,14 @@ pub fn encode_store(store: &Store) -> Result<Vec<u8>, CodecError> {
 /// Decode a store from bytes produced by [`encode_store`].
 pub fn decode_store(data: &[u8]) -> Result<Store, StoreError> {
     let encoded = read_store_envelope(data)?;
-    decode_store_chunks(encoded.chunks)
+    let (store, pending) = Store::try_from_chunks(encoded.chunks)?;
+    if !pending.is_empty() {
+        return Err(CodecError::DataContentError(
+            "store snapshot contains commits that could not be applied".into(),
+        )
+        .into());
+    }
+    Ok(store)
 }
 
 struct EncodedStore {
@@ -83,43 +90,6 @@ fn write_commit_chunk(buf: &mut Vec<u8>, commit: &Commit<'_>) {
     Chunk::from(commit).write(buf)
 }
 
-fn decode_store_chunks(chunks: Vec<Chunk>) -> Result<Store, StoreError> {
-    let roots = chunks
-        .iter()
-        .filter(|chunk| chunk.chunk_type() == ChunkType::Root)
-        .collect::<Vec<_>>();
-    if roots.is_empty() {
-        return Err(CodecError::DataFormatError("commit graph has no root commit".into()).into());
-    }
-    if roots.len() > 1 {
-        return Err(
-            CodecError::DataFormatError("commit graph has multiple root commits".into()).into(),
-        );
-    }
-
-    let root_commit = Commit::from_chunk((*roots[0]).clone(), |_| None)?;
-    let root_payload = root_commit.root_payload()?;
-    let mut store = Store::try_from_ir(root_payload)?;
-
-    let mut commits = Vec::new();
-    for chunk in chunks {
-        if chunk.chunk_type() == ChunkType::Root {
-            continue;
-        }
-
-        let commit = Commit::from_chunk(chunk, |path| {
-            store
-                .resolve_table(path)
-                .and_then(|oid| store.table_meta(oid))
-        })?;
-        commits.push(commit);
-    }
-
-    store.apply_commits(commits)?;
-
-    Ok(store)
-}
-
 #[cfg(test)]
 mod tests {
     use coln_flir_rs::ir::{BuiltinTy, ColType, ColumnEntry, EntityVariant};
@@ -209,13 +179,13 @@ mod tests {
             .expect("encoded store contains a chunk magic")
     }
 
-    fn is_missing_dep_error(result: Result<Store, StoreError>) -> bool {
+    fn is_data_content_error(result: Result<Store, StoreError>) -> bool {
         matches!(
             result,
             Err(err)
                 if matches!(
                     &err,
-                    StoreError::Commit(crate::store::error::CommitApplyError::MissingDep)
+                    StoreError::Encode(CodecError::DataContentError(_))
                 )
         )
     }
@@ -352,7 +322,7 @@ mod tests {
             frame_chunk(ChunkType::Commit, commit.payload()),
         ]);
 
-        assert!(is_missing_dep_error(decode_store(&bytes)));
+        assert!(is_data_content_error(decode_store(&bytes)));
     }
 
     #[test]
diff --git a/packages/coln-store/src/store/error.rs b/packages/coln-store/src/store/error.rs
index bfb53b9d..e2ecd889 100644
--- a/packages/coln-store/src/store/error.rs
+++ b/packages/coln-store/src/store/error.rs
@@ -4,6 +4,7 @@
 
 use crate::commit::error::CodecError;
 use crate::commit::graph::CommitGraphError;
+use crate::commit::hash::CommitHash;
 use crate::solver::compile::CompileError;
 use crate::solver::validate::RuleViolation;
 use crate::table::ValidationError;
@@ -27,15 +28,10 @@ pub enum StoreError {
 
 #[derive(Debug, thiserror::Error)]
 pub enum CommitApplyError {
-    #[error("missing commit dependency")]
-    MissingDep,
-    #[error("disconnected commit")]
-    DisconnectedCommit,
-    // A commit that should definitely exist but is missing
-    #[error("missing commit")]
-    MissingCommit,
+    #[error("A commit {0} with no dependency")]
+    DanglingCommit(CommitHash),
     #[error("An existing commit has conflict payload")]
-    ConflictPayload,
-    #[error("Root commit cannot be applied")]
-    RootCommit,
+    ConflictPayload(CommitHash),
+    #[error("Root commit {0} cannot be applied")]
+    RootCommit(CommitHash),
 }
diff --git a/packages/coln-store/src/store/mod.rs b/packages/coln-store/src/store/mod.rs
index 8da3ec6a..26df187b 100644
--- a/packages/coln-store/src/store/mod.rs
+++ b/packages/coln-store/src/store/mod.rs
@@ -203,47 +203,6 @@ impl Store {
         Ok(graph)
     }
 
-    #[cfg(feature = "native")]
-    // used in SQL mode only
-    pub(crate) fn create_table(
-        &mut self,
-        path: ir::Path,
-        schema: ir::Schema,
-    ) -> Result<TableOid, StoreError> {
-        let oid = self.tables.len();
-        self.path_to_oid.insert(path.clone(), oid);
-        self.tables.insert(oid, Table::new(path, oid, schema));
-
-        let mut tables: Vec<_> = self
-            .tables
-            .values()
-            .map(|table| {
-                (
-                    table.oid(),
-                    ir::TableEntry {
-                        path: table.path().clone(),
-                        table: table.schema().clone(),
-                    },
-                )
-            })
-            .collect();
-        tables.sort_by_key(|(oid, _)| *oid);
-        let ir = FlatRealm {
-            tables: tables.into_iter().map(|(_, entry)| entry).collect(),
-            rules: self.rule_entries.clone(),
-        };
-        self.commits = Self::graph_with_root_commit(&ir)?;
-        Ok(oid)
-    }
-
-    pub fn transaction(&mut self) -> Transaction<'_> {
-        Transaction::new(self)
-    }
-
-    pub fn into_transaction(self) -> OwnedTransaction {
-        OwnedTransaction::new(self)
-    }
-
     /// Builds an empty column store per `theory.tables` and keeps only `theory.rules`
     /// (schemas are stored on each [`Table`]).
     pub fn try_from_ir(ir: FlatRealm) -> Result<Self, StoreError> {
@@ -279,6 +238,18 @@ impl Store {
     }
 }
 
+impl Store {
+    // transactions
+
+    pub fn transaction(&mut self) -> Transaction<'_> {
+        Transaction::new(self)
+    }
+
+    pub fn into_transaction(self) -> OwnedTransaction {
+        OwnedTransaction::new(self)
+    }
+}
+
 impl Store {
     // Dealing with rules
     fn compile_rules(rules: &[RuleEntry]) -> Result<Vec<CompRule>, CompileError> {
@@ -370,21 +341,31 @@ impl Store {
             .collect()
     }
 
+    /// This will try to merge the `other` store as much as possible into this store
+    // TODO need to rethink `merge` more carefully
     pub fn merge(&mut self, other: &Self) -> Result<Vec<CommitHash>, StoreError> {
         let commits = self.commits_added(other);
         self.apply_commits(commits)?;
         Ok(self.heads())
     }
 
-    pub fn apply_commit(&mut self, commit: Commit<'static>) -> Result<(), StoreError> {
+    /// Apply a single commit, respect its dependency.
+    /// Return the commit if it cannot be applied due to missing deps
+    pub fn apply_commit(
+        &mut self,
+        commit: Commit<'static>,
+    ) -> Result<Option<Commit<'static>>, StoreError> {
         // This needs to call apply_commits because it needs to do dependency check
-        self.apply_commits([commit])
+        self.apply_commits([commit]).map(|mut h| h.pop())
     }
 
+    /// Apply as many commits as possible respecting their dependencies. Return the
+    /// commit hashes that are NOT applied, so the caller knows which ones they
+    /// need to retry.
     pub fn apply_commits(
         &mut self,
         commits: impl IntoIterator<Item = Commit<'static>>,
-    ) -> Result<(), StoreError> {
+    ) -> Result<Vec<Commit<'static>>, StoreError> {
         let mut pending = HashMap::new();
 
         for commit in commits {
@@ -393,17 +374,21 @@ impl Store {
                 continue;
             }
 
+            // We assume that the root commit has been used to construct the store
+            // and therefore must have been applied
             if commit.is_root() {
-                return Err(CommitApplyError::RootCommit.into());
+                return Err(CommitApplyError::RootCommit(commit.hash()).into());
             }
+
+            // We assume that all commits will have deps
             if commit.deps.is_empty() {
-                return Err(CommitApplyError::MissingDep.into());
+                return Err(CommitApplyError::DanglingCommit(commit.hash()).into());
             }
 
             if let Some(existing) = pending.get(&hash) {
                 let existing: &Commit<'static> = existing;
                 if *existing != commit {
-                    return Err(CommitApplyError::ConflictPayload.into());
+                    return Err(CommitApplyError::ConflictPayload(commit.hash()).into());
                 }
                 continue;
             }
@@ -430,8 +415,12 @@ impl Store {
                     count += 1;
                     waiting_on.entry(*dep).or_default().push(*hash);
                 } else {
-                    // deps is not in pending or applied commits
-                    return Err(CommitApplyError::MissingDep.into());
+                    tracing::info!(
+                        commit_hash = %commit.hash(),
+                        missing_dep = %dep,
+                        "skipping commit with dependency that is neither applied nor pending"
+                    );
+                    count += 1;
                 }
             }
 
@@ -445,14 +434,14 @@ impl Store {
         while let Some(hash) = ready.pop_first() {
             let commit = pending
                 .remove(&hash)
-                .ok_or(CommitApplyError::MissingCommit)?;
+                .expect("hash in ready should also exist in pending");
 
             self.apply_commit_atomic(commit)?;
             if let Some(waitings) = waiting_on.remove(&hash) {
                 for wh in waitings {
                     let count = unsatisfied
                         .get_mut(&wh)
-                        .ok_or(CommitApplyError::MissingCommit)?;
+                        .expect("A commit that is waiting must be in unsatisfied");
                     *count -= 1;
                     if *count == 0 {
                         unsatisfied.remove(&wh).unwrap();
@@ -462,12 +451,7 @@ impl Store {
             }
         }
 
-        // pending is not empty, but there is no commit to apply
-        if !pending.is_empty() {
-            return Err(CommitApplyError::DisconnectedCommit.into());
-        }
-
-        Ok(())
+        Ok(pending.into_values().collect())
     }
 
     fn apply_commit_atomic(&mut self, commit: Commit<'static>) -> Result<(), StoreError> {
@@ -644,10 +628,11 @@ impl Store {
     }
 
     /// Apply the bytes received by interpreting them as chunks, for syncing purposes
+    /// Return chunk bytes that cannot be applied yet.
     pub fn apply_chunk_bytes(
         &mut self,
         chunk_bytes: impl IntoIterator<Item = Vec<u8>>,
-    ) -> Result<(), StoreError> {
+    ) -> Result<Vec<Vec<u8>>, StoreError> {
         let commits = chunk_bytes
             .into_iter()
             .map(|bytes| Chunk::decode(&bytes))
@@ -661,12 +646,104 @@ impl Store {
             })
             .collect::<Result<Vec<_>, _>>()?;
 
-        self.apply_commits(commits)
+        Ok(Self::commits_to_chunk_bytes(self.apply_commits(commits)?))
+    }
+
+    /// Build a store from commit chunks from scratch, assuming that the input
+    /// `chunk_bytes` contains a valid root commit.
+    pub fn try_from_commit_bytes(
+        chunk_bytes: impl IntoIterator<Item = impl AsRef<[u8]>>,
+    ) -> Result<(Self, Vec<Vec<u8>>), StoreError> {
+        let chunks = chunk_bytes
+            .into_iter()
+            .map(|bytes| Chunk::decode(bytes.as_ref()))
+            .collect::<Result<Vec<_>, _>>()?;
+        Self::try_from_chunks(chunks)
+    }
+
+    /// Create a store from the commit chunks, assuming the chunk contains a valid root
+    pub(crate) fn try_from_chunks(chunks: Vec<Chunk>) -> Result<(Self, Vec<Vec<u8>>), StoreError> {
+        let roots = chunks
+            .iter()
+            .filter(|chunk| chunk.is_root())
+            .collect::<Vec<_>>();
+        if roots.is_empty() {
+            return Err(
+                CodecError::DataFormatError("commit graph has no root commit".into()).into(),
+            );
+        }
+        if roots.len() > 1 {
+            return Err(CodecError::DataFormatError(
+                "commit graph has multiple root commits".into(),
+            )
+            .into());
+        }
+
+        let root_commit = Commit::from_chunk((*roots[0]).clone(), |_| None)?;
+        let root_payload = root_commit.root_payload()?;
+        let mut store = Store::try_from_ir(root_payload)?;
+
+        let mut commits = Vec::new();
+        for chunk in chunks {
+            if chunk.is_root() {
+                continue;
+            }
+
+            let commit = Commit::from_chunk(chunk, |path| {
+                store
+                    .resolve_table(path)
+                    .and_then(|oid| store.table_meta(oid))
+            })?;
+            commits.push(commit);
+        }
+
+        let pending_bytes = Self::commits_to_chunk_bytes(store.apply_commits(commits)?);
+        Ok((store, pending_bytes))
+    }
+
+    fn commits_to_chunk_bytes(commits: Vec<Commit<'static>>) -> Vec<Vec<u8>> {
+        commits
+            .into_iter()
+            .map(|commit| Chunk::from(commit).encoded())
+            .collect()
     }
 }
 
 impl Store {
-    // for debugging and testing
+    // for debugging and testing and experiments
+
+    #[cfg(feature = "native")]
+    // used in SQL mode only
+    pub(crate) fn create_table(
+        &mut self,
+        path: ir::Path,
+        schema: ir::Schema,
+    ) -> Result<TableOid, StoreError> {
+        let oid = self.tables.len();
+        self.path_to_oid.insert(path.clone(), oid);
+        self.tables.insert(oid, Table::new(path, oid, schema));
+
+        let mut tables: Vec<_> = self
+            .tables
+            .values()
+            .map(|table| {
+                (
+                    table.oid(),
+                    ir::TableEntry {
+                        path: table.path().clone(),
+                        table: table.schema().clone(),
+                    },
+                )
+            })
+            .collect();
+        tables.sort_by_key(|(oid, _)| *oid);
+        let ir = FlatRealm {
+            tables: tables.into_iter().map(|(_, entry)| entry).collect(),
+            rules: self.rule_entries.clone(),
+        };
+        self.commits = Self::graph_with_root_commit(&ir)?;
+        Ok(oid)
+    }
 
     /// Dump every table in the store for debugging, in ascending [`TableOid`] order,
     /// separated by a blank line.
diff --git a/packages/coln-store/src/store/tests.rs b/packages/coln-store/src/store/tests.rs
index 47efaf04..4a706547 100644
--- a/packages/coln-store/src/store/tests.rs
+++ b/packages/coln-store/src/store/tests.rs
@@ -826,6 +826,41 @@ mod commits {
         assert_eq!(hashes, vec![commit]);
     }
 
+    #[test]
+    fn commit_chunks_create_empty_store() {
+        let source = Store::new();
+        let chunks = source
+            .commit_chunks_after(&[])
+            .into_iter()
+            .map(|chunk| chunk.bytes)
+            .collect::<Vec<_>>();
+
+        let (restored, pending) = Store::try_from_commit_bytes(chunks).expect("store from chunks");
+
+        assert!(pending.is_empty());
+        assert_eq!(restored.table_count(), 0);
+        assert_eq!(restored.heads(), source.heads());
+    }
+
+    #[test]
+    fn commit_chunks_create_store_from_out_of_order_data() {
+        let mut source = single_int_store();
+        let commit = commit_int(&mut source, 99);
+        let mut chunks = source
+            .commit_chunks_after(&[])
+            .into_iter()
+            .map(|chunk| chunk.bytes)
+            .collect::<Vec<_>>();
+        chunks.reverse();
+
+        let (restored, pending) = Store::try_from_commit_bytes(chunks).expect("store from chunks");
+
+        assert!(pending.is_empty());
+        let table = restored.table_at(&Path::from("T")).expect("table");
+        assert_eq!(table.cell_at(0, 0), Some(CellValue::Int(99)));
+        assert_eq!(restored.heads(), vec![commit]);
+    }
+
     #[test]
     fn apply_commits_applies_rows_and_updates_heads() {
         let mut source = single_int_store();
@@ -882,7 +917,7 @@ mod commits {
     }
 
     #[test]
-    fn apply_commits_rejects_missing_dependency_without_changing_store() {
+    fn apply_commits_skips_missing_dependency_without_changing_store() {
         let mut source = single_int_store();
         let mut target = single_int_store();
         commit_int(&mut source, 1);
@@ -892,12 +927,65 @@ mod commits {
             .expect("second commit")
             .clone();
 
-        let err = target.apply_commits([second_commit]).unwrap_err();
+        let leftover = target
+            .apply_commits([second_commit])
+            .expect("skip missing dependency");
+        let leftover_hashes: Vec<_> = leftover.iter().map(Commit::hash).collect();
 
-        assert!(matches!(
-            err,
-            StoreError::Commit(CommitApplyError::MissingDep)
-        ));
+        assert_eq!(leftover_hashes, vec![second]);
+        assert_eq!(
+            target
+                .table_at(&Path::from("T"))
+                .expect("table")
+                .row_count(),
+            0
+        );
+    }
+
+    #[test]
+    fn apply_commits_applies_ready_commits_and_returns_blocked_ones() {
+        let mut source = single_int_store();
+        let mut target = single_int_store();
+        let first = commit_int(&mut source, 1);
+        commit_int(&mut source, 2);
+        let third = commit_int(&mut source, 3);
+        let first_commit = source.commit_by_hash(&first).expect("first commit").clone();
+        let third_commit = source.commit_by_hash(&third).expect("third commit").clone();
+
+        let leftover = target
+            .apply_commits([first_commit, third_commit])
+            .expect("apply ready commits");
+        let leftover_hashes: Vec<_> = leftover.iter().map(Commit::hash).collect();
+
+        assert_eq!(leftover_hashes, vec![third]);
+        assert_eq!(
+            target
+                .table_at(&Path::from("T"))
+                .expect("table")
+                .row_count(),
+            1
+        );
+        assert_eq!(target.heads(), vec![first]);
+    }
+
+    #[test]
+    fn apply_chunk_bytes_retries_leftover_when_missing_parent_arrives() {
+        let mut source = single_int_store();
+        let mut target = single_int_store();
+        commit_int(&mut source, 1);
+        let second = commit_int(&mut source, 2);
+
+        let chunks: Vec<Vec<u8>> = source
+            .commit_chunks_after(&target.heads())
+            .into_iter()
+            .map(|chunk| chunk.bytes)
+            .collect();
+        assert_eq!(chunks.len(), 2);
+
+        let leftover = target
+            .apply_chunk_bytes([chunks[1].clone()])
+            .expect("skip child without parent");
+        assert_eq!(leftover.len(), 1);
         assert_eq!(
             target
                 .table_at(&Path::from("T"))
@@ -905,6 +993,21 @@ mod commits {
                 .row_count(),
             0
         );
+
+        let leftover = target
+            .apply_chunk_bytes(
+                leftover
+                    .into_iter()
+                    .chain(std::iter::once(chunks[0].clone())),
+            )
+            .expect("retry leftover with parent");
+        assert!(leftover.is_empty());
+
+        let table = target.table_at(&Path::from("T")).expect("table");
+        assert_eq!(table.row_count(), 2);
+        assert_eq!(table.cell_at(0, 0), Some(CellValue::Int(1)));
+        assert_eq!(table.cell_at(1, 0), Some(CellValue::Int(2)));
+        assert_eq!(target.heads(), vec![second]);
     }
 }
 
diff --git a/packages/coln-store/src/txn/inner.rs b/packages/coln-store/src/txn/inner.rs
index ac02ac33..f75b6920 100644
--- a/packages/coln-store/src/txn/inner.rs
+++ b/packages/coln-store/src/txn/inner.rs
@@ -31,7 +31,7 @@ pub(crate) struct TxnInner {
 }
 
 impl TxnInner {
-    pub(crate) fn new(deps: Vec<CommitHash>) -> Self {
+    pub(super) fn new(deps: Vec<CommitHash>) -> Self {
         Self {
             deps,
             author: Author::foo(),
@@ -66,7 +66,7 @@ impl TxnInner {
         Ok(temp_id)
     }
 
-    pub(crate) fn add(
+    pub(super) fn add(
         &mut self,
         store: &Store,
         table: &ir::Path,
@@ -108,7 +108,7 @@ impl TxnInner {
         });
     }
 
-    pub(crate) fn commit(self, store: &mut Store) -> Result<CommitHash, StoreError> {
+    pub(super) fn commit(self, store: &mut Store) -> Result<CommitHash, StoreError> {
         info!(op_count = self.pending.len(), "commit txn");
         let TxnInner {
             deps,
@@ -133,22 +133,22 @@ impl TxnInner {
 
         let h = cmt.hash();
         match store.apply_commit(cmt) {
-            Ok(()) => {
+            Ok(None) => {
+                // Everything applied successfully
                 Self::finalize_handles(pending_handles, h, store);
                 Ok(h)
             }
+            Ok(Some(_)) => {
+                unreachable!("commit a local transaction should always succeed");
+            }
             Err(err) => {
                 Self::invalidate_handles(pending_handles, "txn commit failed");
                 Err(err)
             }
         }
-        // 1. validate full batch (PK conflicts including intra-batch)
-        // 2. compute hash: blake3(deps || timestamp || message || canonical(ops))
-        // 3. resolve: TxnRowId(k) -> RowId { commit: hash, counter: k }
-        //             CellValue::TxnId(k) -> CellValue::Id(RowId { commit: hash, counter: k })
-        // 4. apply resolved Ops to tables via table.insert_row
-        // 5. check_rules
-        // 6. push CommitMeta into store.commit_graph, advance heads
-        // 7. return hash
+    }
+
+    pub(super) fn abort(self) {
+        Self::invalidate_handles(self.pending_handles, "txn abort");
     }
 }
diff --git a/packages/coln-store/src/txn/mod.rs b/packages/coln-store/src/txn/mod.rs
index bc17adf6..30061f30 100644
--- a/packages/coln-store/src/txn/mod.rs
+++ b/packages/coln-store/src/txn/mod.rs
@@ -55,6 +55,10 @@ impl<'a> Transaction<'a> {
         self.inner.commit(self.store)
     }
     // pub fn commit_with(mut self, opts: CommitOptions) -> Result<CommitHash, StoreIntError> { ... }
+
+    pub fn abort(self) {
+        self.inner.abort()
+    }
 }
 
 pub struct OwnedTransaction {
@@ -79,6 +83,11 @@ impl OwnedTransaction {
         self.inner.add(&self.store, table, values)
     }
 
+    pub fn abort(self) -> Store {
+        self.inner.abort();
+        self.store
+    }
+
     // We need to return Store to the user for roll back purposes, so the Err variant must be large
     #[allow(clippy::result_large_err)]
     pub fn commit(mut self) -> Result<(CommitHash, Store), (StoreError, Store)> {
@@ -225,6 +234,39 @@ mod tests {
         assert_eq!(edge.cell_at(0, 0), Some(CellValue::Id(node_id)));
     }
 
+    #[test]
+    fn abort_invalidates_returned_row_handles() {
+        let nodes = Path::from("Nodes");
+        let edges = Path::from("Edges");
+        let mut store = Store::new();
+        store
+            .create_table(nodes.clone(), table_schema(vec![], None))
+            .expect("create nodes table");
+        store
+            .create_table(
+                edges.clone(),
+                table_schema(vec![row_id_col("node", nodes.clone())], None),
+            )
+            .expect("create edges table");
+        let mut tx = store.transaction();
+        let node = tx.add(&nodes, vec![]).expect("add node");
+        tx.abort();
+        let err = node.row_id().expect_err("abort invalidates handle");
+        assert!(matches!(
+            err,
+            StoreError::Validation(ValidationError::InvalidRowHandle { .. })
+        ));
+        let mut tx = store.transaction();
+        let err = tx
+            .add(&edges, vec![node.into()])
+            .expect_err("aborted handle cannot be reused");
+        assert!(matches!(
+            err,
+            StoreError::Validation(ValidationError::InvalidRowHandle { .. })
+        ));
+        assert_eq!(store.table_at(&nodes).expect("Nodes").row_count(), 0);
+    }
+
     #[test]
     fn failed_transaction_invalidates_returned_row_handles() {
         let nodes = Path::from("Nodes");