diff --git a/go/cpossum/c-possum.go b/go/cpossum/c-possum.go
index dfc32dd..0dd3155 100644
--- a/go/cpossum/c-possum.go
+++ b/go/cpossum/c-possum.go
@@ -58,7 +58,7 @@ func (me Stat) Size() int64 {
 	return int64(me.size)
 }
 
-type Handle = C.Handle
+type Handle = C.PossumHandle
 
 func NewHandle(dir string) *Handle {
 	cDir := C.CString(dir)
diff --git a/go/cpossum/possum.h b/go/cpossum/possum.h
index bb1d6f4..b84fa69 100644
--- a/go/cpossum/possum.h
+++ b/go/cpossum/possum.h
@@ -16,13 +16,7 @@ typedef enum {
 /**
  * Manages uncommitted writes
  */
-typedef struct BatchWriter BatchWriter;
-
-/**
- * Provides access to a storage directory. Manages manifest access, file cloning, file writers,
- * configuration, value eviction etc.
- */
-typedef struct Handle Handle;
+typedef struct BatchWriter_PossumHandle BatchWriter_PossumHandle;
 
 typedef struct PossumReader PossumReader;
 
@@ -31,9 +25,15 @@ typedef struct PossumReader PossumReader;
  */
 typedef struct PossumValue PossumValue;
 
+typedef struct Rc_RwLock_Handle Rc_RwLock_Handle;
+
 typedef struct ValueWriter ValueWriter;
 
-typedef BatchWriter PossumWriter;
+typedef Rc_RwLock_Handle PossumHandleRc;
+
+typedef PossumHandleRc PossumHandle;
+
+typedef BatchWriter_PossumHandle PossumWriter;
 
 typedef ValueWriter PossumValueWriter;
 
@@ -66,13 +66,13 @@ typedef struct {
   bool disable_hole_punching;
 } PossumLimits;
 
-Handle *possum_new(const char *path);
+PossumHandle *possum_new(const char *path);
 
 PossumError possum_start_new_value(PossumWriter *writer, PossumValueWriter **value);
 
 RawFileHandle possum_value_writer_fd(PossumValueWriter *value);
 
-PossumError possum_writer_rename(BatchWriter *writer, const PossumValue *value, PossumBuf new_key);
+PossumError possum_writer_rename(PossumWriter *writer, const PossumValue *value, PossumBuf new_key);
 
 PossumError possum_reader_add(PossumReader *reader, PossumBuf key, const PossumValue **value);
 
@@ -99,24 +99,24 @@ PossumError possum_writer_commit(PossumWriter *writer);
 
 PossumError possum_writer_stage(PossumWriter *writer, PossumBuf key, PossumValueWriter *value);
 
-void possum_drop(Handle *handle);
+void possum_drop(PossumHandle *handle);
 
-PossumError possum_set_instance_limits(Handle *handle, const PossumLimits *limits);
+PossumError possum_set_instance_limits(PossumHandle *handle, const PossumLimits *limits);
 
-PossumError possum_cleanup_snapshots(const Handle *handle);
+PossumError possum_cleanup_snapshots(const PossumHandle *handle);
 
-size_t possum_single_write_buf(Handle *handle, PossumBuf key, PossumBuf value);
+size_t possum_single_write_buf(PossumHandle *handle, PossumBuf key, PossumBuf value);
 
-PossumWriter *possum_new_writer(Handle *handle);
+PossumWriter *possum_new_writer(PossumHandle *handle);
 
-bool possum_single_stat(const Handle *handle, PossumBuf key, PossumStat *out_stat);
+bool possum_single_stat(const PossumHandle *handle, PossumBuf key, PossumStat *out_stat);
 
-PossumError possum_list_items(const Handle *handle,
+PossumError possum_list_items(const PossumHandle *handle,
                               PossumBuf prefix,
                               PossumItem **out_list,
                               size_t *out_list_len);
 
-PossumError possum_single_read_at(const Handle *handle,
+PossumError possum_single_read_at(const PossumHandle *handle,
                                   PossumBuf key,
                                   PossumBuf *buf,
                                   uint64_t offset);
@@ -124,10 +124,10 @@ PossumError possum_single_read_at(const Handle *handle,
 /**
  * stat is filled if non-null and a delete occurs. NoSuchKey is returned if the key does not exist.
  */
-PossumError possum_single_delete(const Handle *handle, PossumBuf key, PossumStat *stat);
+PossumError possum_single_delete(const PossumHandle *handle, PossumBuf key, PossumStat *stat);
 
-PossumError possum_reader_new(const Handle *handle, PossumReader **reader);
+PossumError possum_reader_new(const PossumHandle *handle, PossumReader **reader);
 
-PossumError possum_handle_move_prefix(Handle *handle, PossumBuf from, PossumBuf to);
+PossumError possum_handle_move_prefix(PossumHandle *handle, PossumBuf from, PossumBuf to);
 
-PossumError possum_handle_delete_prefix(Handle *handle, PossumBuf prefix);
+PossumError possum_handle_delete_prefix(PossumHandle *handle, PossumBuf prefix);
diff --git a/src/c_api/ext_fns/handle.rs b/src/c_api/ext_fns/handle.rs
index fe3271b..79888cb 100644
--- a/src/c_api/ext_fns/handle.rs
+++ b/src/c_api/ext_fns/handle.rs
@@ -1,42 +1,44 @@
 use libc::size_t;
 use positioned_io::ReadAt;
+use rusqlite::TransactionBehavior;
 
 use super::*;
 use crate::c_api::PossumError::NoError;
-use crate::Handle;
 
-// This drops the Handle Box. Instead, if this is hard to use correctly from C, it could drop a
-// top-level reference count for the box. i.e. If this one goes, there's no way to work with the
-// Handle, and when all other outstanding operations on the Handle complete, it will drop the Handle
-// for real.
+// This drops the PossumHandle Box. Instead, if this is hard to use correctly from C, it could drop
+// a top-level reference count for the box. i.e. If this one goes, there's no way to work with the
+// PossumHandle, and when all other outstanding operations on the PossumHandle complete, it will
+// drop the PossumHandle for real.
 #[no_mangle]
-pub extern "C" fn possum_drop(handle: *mut Handle) {
+pub extern "C" fn possum_drop(handle: *mut PossumHandle) {
     drop(unsafe { Box::from_raw(handle) })
 }
 
 #[no_mangle]
 pub extern "C" fn possum_set_instance_limits(
-    handle: *mut Handle,
+    handle: *mut PossumHandle,
     limits: *const PossumLimits,
 ) -> PossumError {
     let handle = unsafe { &mut *handle };
     let limits = unsafe { limits.read() };
     with_residual(|| {
         handle
+            .write()
+            .unwrap()
             .set_instance_limits(limits.into())
             .map_err(Into::into)
     })
 }
 
 #[no_mangle]
-pub extern "C" fn possum_cleanup_snapshots(handle: *const Handle) -> PossumError {
-    let handle = unsafe { &*handle };
-    with_residual(|| handle.cleanup_snapshots())
+pub extern "C" fn possum_cleanup_snapshots(handle: *const PossumHandle) -> PossumError {
+    let handle = unwrap_possum_handle(handle);
+    with_residual(|| handle.read().unwrap().cleanup_snapshots())
 }
 
 #[no_mangle]
 pub extern "C" fn possum_single_write_buf(
-    handle: *mut Handle,
+    handle: *mut PossumHandle,
     key: PossumBuf,
     value: PossumBuf,
 ) -> size_t {
@@ -44,7 +46,11 @@ pub extern "C" fn possum_single_write_buf(
     let value_slice = value.as_ref();
     const ERR_SENTINEL: usize = usize::MAX;
     let handle = unsafe { &*handle };
-    match handle.single_write_from(key_vec, value_slice) {
+    match handle
+        .read()
+        .unwrap()
+        .single_write_from(key_vec, value_slice)
+    {
         Err(_) => ERR_SENTINEL,
         Ok((n, _)) => {
             let n = n.try_into().unwrap();
@@ -55,18 +61,21 @@ pub extern "C" fn possum_single_write_buf(
 }
 
 #[no_mangle]
-pub extern "C" fn possum_new_writer(handle: *mut Handle) -> *mut PossumWriter {
-    let handle = unsafe { handle.as_ref() }.unwrap();
-    Box::into_raw(Box::new(handle.new_writer().unwrap()))
+pub extern "C" fn possum_new_writer(handle: *mut PossumHandle) -> *mut PossumWriter {
+    let handle = unwrap_possum_handle(handle);
+    let writer = BatchWriter::new(handle.clone());
+    Box::into_raw(Box::new(writer))
 }
 
 #[no_mangle]
 pub extern "C" fn possum_single_stat(
-    handle: *const Handle,
+    handle: *const PossumHandle,
     key: PossumBuf,
     out_stat: *mut PossumStat,
 ) -> bool {
     match unsafe { handle.as_ref() }
+        .unwrap()
+        .read()
         .unwrap()
         .read_single(key.as_ref())
         .unwrap()
@@ -82,12 +91,14 @@ pub extern "C" fn possum_single_stat(
 
 #[no_mangle]
 pub extern "C" fn possum_list_items(
-    handle: *const Handle,
+    handle: *const PossumHandle,
     prefix: PossumBuf,
     out_list: *mut *mut PossumItem,
     out_list_len: *mut size_t,
 ) -> PossumError {
     let items = match unsafe { handle.as_ref() }
+        .unwrap()
+        .read()
         .unwrap()
         .list_items(prefix.as_ref())
     {
@@ -100,13 +111,18 @@ pub extern "C" fn possum_list_items(
 
 #[no_mangle]
 pub extern "C" fn possum_single_read_at(
-    handle: *const Handle,
+    handle: *const PossumHandle,
     key: PossumBuf,
     buf: *mut PossumBuf,
     offset: u64,
 ) -> PossumError {
     let rust_key = key.as_ref();
-    let value = match unsafe { handle.as_ref() }.unwrap().read_single(rust_key) {
+    let value = match unsafe { handle.as_ref() }
+        .unwrap()
+        .read()
+        .unwrap()
+        .read_single(rust_key)
+    {
         Ok(Some(value)) => value,
         Ok(None) => return PossumError::NoSuchKey,
         Err(err) => return err.into(),
@@ -124,13 +140,13 @@ pub extern "C" fn possum_single_read_at(
 /// stat is filled if non-null and a delete occurs. NoSuchKey is returned if the key does not exist.
 #[no_mangle]
 pub extern "C" fn possum_single_delete(
-    handle: *const Handle,
+    handle: *const PossumHandle,
     key: PossumBuf,
     stat: *mut PossumStat,
 ) -> PossumError {
     with_residual(|| {
         let handle = unsafe { &*handle };
-        let value = match handle.single_delete(key.as_ref()) {
+        let value = match handle.read().unwrap().single_delete(key.as_ref()) {
             Ok(None) => return Err(crate::Error::NoSuchKey),
             Err(err) => return Err(err),
             Ok(Some(value)) => value,
@@ -144,15 +160,27 @@ pub extern "C" fn possum_single_delete(
 
 #[no_mangle]
 pub extern "C" fn possum_reader_new(
-    handle: *const Handle,
+    handle: *const PossumHandle,
     reader: *mut *mut PossumReader,
 ) -> PossumError {
-    let handle = unsafe { handle.as_ref() }.unwrap();
+    let handle = unwrap_possum_handle(handle).clone();
     let reader = unsafe { reader.as_mut() }.unwrap();
-    let rust_reader = match handle.read() {
+    let owned_tx_res = handle.start_transaction(
+        // This is copied from Handle::start_writable_transaction_with_behaviour and Handle::read
+        // until I make proper abstractions.
+        |conn, handle| {
+            let rtx = conn.transaction_with_behavior(TransactionBehavior::Deferred)?;
+            Ok(Transaction::new(rtx, handle))
+        },
+    );
+    let owned_tx = match owned_tx_res {
         Ok(ok) => ok,
         Err(err) => return err.into(),
     };
+    let rust_reader = Reader {
+        owned_tx,
+        reads: Default::default(),
+    };
     *reader = Box::into_raw(Box::new(PossumReader {
         rust_reader: Some(rust_reader),
         values: Default::default(),
@@ -162,13 +190,15 @@ pub extern "C" fn possum_reader_new(
 
 #[no_mangle]
 pub extern "C" fn possum_handle_move_prefix(
-    handle: *mut Handle,
+    handle: *mut PossumHandle,
     from: PossumBuf,
     to: PossumBuf,
 ) -> PossumError {
     let handle = unsafe { &mut *handle };
     with_residual(|| {
         handle
+            .read()
+            .unwrap()
             .move_prefix(from.as_ref(), to.as_ref())
             .map_err(Into::into)
     })
@@ -176,9 +206,15 @@ pub extern "C" fn possum_handle_move_prefix(
 
 #[no_mangle]
 pub extern "C" fn possum_handle_delete_prefix(
-    handle: *mut Handle,
+    handle: *mut PossumHandle,
     prefix: PossumBuf,
 ) -> PossumError {
     let handle = unsafe { &mut *handle };
-    with_residual(|| handle.delete_prefix(prefix.as_ref()).map_err(Into::into))
+    with_residual(|| {
+        handle
+            .read()
+            .unwrap()
+            .delete_prefix(prefix.as_ref())
+            .map_err(Into::into)
+    })
 }
diff --git a/src/c_api/ext_fns/mod.rs b/src/c_api/ext_fns/mod.rs
index 551399a..2472b65 100644
--- a/src/c_api/ext_fns/mod.rs
+++ b/src/c_api/ext_fns/mod.rs
@@ -3,6 +3,8 @@ mod handle;
 use std::ffi::{c_char, CStr};
 use std::path::PathBuf;
 use std::ptr::null_mut;
+use std::rc::Rc;
+use std::sync::RwLock;
 
 use libc::size_t;
 use positioned_io::ReadAt;
@@ -13,7 +15,7 @@ use crate::c_api::PossumError::{NoError, NoSuchKey};
 use crate::Handle;
 
 #[no_mangle]
-pub extern "C" fn possum_new(path: *const c_char) -> *mut Handle {
+pub extern "C" fn possum_new(path: *const c_char) -> *mut PossumHandle {
     if let Err(err) = env_logger::try_init() {
         warn!("error initing env_logger: {}", err);
     }
@@ -33,7 +35,7 @@ pub extern "C" fn possum_new(path: *const c_char) -> *mut Handle {
             return null_mut();
         }
     };
-    Box::into_raw(Box::new(handle))
+    Box::into_raw(Box::new(Rc::new(RwLock::new(handle))))
 }
 
 #[no_mangle]
@@ -78,7 +80,8 @@ pub extern "C" fn possum_reader_add(
     value: *mut *const PossumValue,
 ) -> PossumError {
     let reader = unsafe { reader.as_mut() }.unwrap();
-    let rust_value = match reader.rust_reader.as_mut().unwrap().add(key.as_ref()) {
+    let mut_rust_reader = reader.rust_reader.as_mut().unwrap();
+    let rust_value = match mut_rust_reader.add(key.as_ref()) {
         Ok(None) => return NoSuchKey,
         Ok(Some(value)) => value,
         Err(err) => return err.into(),
diff --git a/src/c_api/mod.rs b/src/c_api/mod.rs
index f750ad1..37c5aa5 100644
--- a/src/c_api/mod.rs
+++ b/src/c_api/mod.rs
@@ -9,7 +9,9 @@ use std::ffi::c_char;
 use std::mem::size_of;
 use std::pin::Pin;
 use std::ptr::copy_nonoverlapping;
+use std::rc::Rc;
 use std::slice;
+use std::sync::RwLockReadGuard;
 
 use libc::{calloc, malloc, size_t};
 
@@ -32,10 +34,15 @@ impl PossumBuf {
 struct PossumReader {
     // Removed when converted to a snapshot. Specific to the C API so as to not need to expose
     // Snapshot, and to convert Values automatically when a snapshot starts.
-    rust_reader: Option<Reader<OwnedTx<'static>>>,
+    rust_reader: Option<Reader<PossumReaderOwnedTransaction<'static>>>,
     values: Vec<Pin<Box<PossumValue>>>,
 }
 
+pub(crate) type PossumReaderOwnedTransaction<'a> = <PossumHandleRc as StartTransaction<
+    'a,
+    Transaction<'a, Rc<RwLockReadGuard<'a, Handle>>>,
+>>::Owned;
+
 use crate::c_api::PossumError::{AnyhowError, IoError, SqliteError};
 
 impl<V> From<V> for PossumStat
@@ -142,6 +149,10 @@ fn with_residual(f: impl FnOnce() -> PubResult<()>) -> PossumError {
     }
 }
 
+fn unwrap_possum_handle<'a>(handle: *const PossumHandle) -> &'a PossumHandle {
+    unsafe { handle.as_ref() }.unwrap()
+}
+
 impl From<PossumLimits> for handle::Limits {
     fn from(from: PossumLimits) -> Self {
         handle::Limits {
diff --git a/src/c_api/types.rs b/src/c_api/types.rs
index f0f42e8..1f62a91 100644
--- a/src/c_api/types.rs
+++ b/src/c_api/types.rs
@@ -12,7 +12,7 @@ use crate::{BatchWriter, Handle, ValueWriter};
 pub(crate) type PossumOffset = u64;
 
 pub type RawFileHandle = libc::intptr_t;
-pub type PossumWriter = BatchWriter<&'static Handle>;
+pub type PossumWriter = BatchWriter<PossumHandle>;
 
 #[repr(C)]
 #[derive(Debug)]
@@ -60,3 +60,5 @@ pub(crate) struct PossumLimits {
 pub(crate) type PossumValueWriter = ValueWriter;
 
 pub(crate) type PossumHandleRc = Rc<RwLock<Handle>>;
+
+pub(crate) type PossumHandle = PossumHandleRc;
diff --git a/src/handle.rs b/src/handle.rs
index a40da10..c2cf974 100644
--- a/src/handle.rs
+++ b/src/handle.rs
@@ -216,26 +216,9 @@ impl Handle {
     }
 
     pub fn new_writer(&self) -> Result<BatchWriter<&Handle>> {
-        Ok(BatchWriter {
-            handle: self,
-            exclusive_files: Default::default(),
-            pending_writes: Default::default(),
-            value_renames: Default::default(),
-        })
+        Ok(BatchWriter::new(self))
     }
 
-    // fn start_transaction<'h, T, O>(
-    //     &'h self,
-    //     make_tx: impl FnOnce(&'h mut Connection, &'h Handle) -> rusqlite::Result<T>,
-    // ) -> rusqlite::Result<O>
-    // where
-    //     O: From<OwnedTxInner<'h, T>>,
-    // {
-    //     self.star
-    //     let guard = self.conn.lock().unwrap();
-    //     Ok(owned_cell::OwnedCell::try_make(guard, |conn| make_tx(conn, self))?.into())
-    // }
-
     pub(crate) fn start_immediate_transaction(&self) -> rusqlite::Result<OwnedTx> {
         self.start_writable_transaction_with_behaviour(TransactionBehavior::Immediate)
     }
@@ -484,6 +467,7 @@ impl Handle {
 
 use item::Item;
 
+use crate::c_api::{PossumHandle, PossumHandleRc};
 use crate::dir::Dir;
 use crate::owned_cell::OwnedCell;
 use crate::ownedtx::{OwnedReadTx, OwnedTxInner};
@@ -531,16 +515,17 @@ impl<'h, T> StartTransaction<'h, T> for &'h Handle {
         make_tx: impl FnOnce(&'h mut Connection, Self::TxHandle) -> rusqlite::Result<T>,
     ) -> rusqlite::Result<Self::Owned> {
         let guard = self.conn.lock().unwrap();
-        owned_cell::OwnedCell::try_make_mut(guard, |conn| make_tx(conn, self))
+        OwnedCell::try_make_mut(guard, |conn| make_tx(conn, self))
     }
 }
 
-impl<'h, T> StartTransaction<'h, T> for Rc<RwLock<Handle>> {
+impl<'h, T> StartTransaction<'h, T> for PossumHandleRc {
     type Owned = OwnedCell<
         Self,
         OwnedCell<Rc<RwLockReadGuard<'h, Handle>>, OwnedCell<MutexGuard<'h, Connection>, T>>,
     >;
     type TxHandle = Rc<RwLockReadGuard<'h, Handle>>;
+    // type TxHandle = &'h Handle;
     fn start_transaction(
         self,
         make_tx: impl FnOnce(&'h mut Connection, Self::TxHandle) -> rusqlite::Result<T>,
@@ -565,3 +550,21 @@ impl WithHandle for &Handle {
         f(self)
     }
 }
+
+impl WithHandle for PossumHandle {
+    fn with_handle<R>(&self, f: impl FnOnce(&Handle) -> R) -> R {
+        f(&self.read().unwrap())
+    }
+}
+
+impl AsRef<Handle> for Handle {
+    fn as_ref(&self) -> &Handle {
+        &self
+    }
+}
+
+impl AsRef<Handle> for Rc<RwLockReadGuard<'_, Handle>> {
+    fn as_ref(&self) -> &Handle {
+        self.deref()
+    }
+}
diff --git a/src/lib.rs b/src/lib.rs
index 0eaf3c7..0b737bd 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -245,6 +245,20 @@ where
     value_renames: Vec<ValueRename>,
 }
 
+impl<H> BatchWriter<H>
+where
+    H: WithHandle,
+{
+    pub fn new(handle: H) -> Self {
+        Self {
+            handle,
+            exclusive_files: Default::default(),
+            pending_writes: Default::default(),
+            value_renames: Default::default(),
+        }
+    }
+}
+
 pub type TimestampInner = NaiveDateTime;
 
 #[derive(Debug, PartialEq, Copy, Clone, PartialOrd)]
diff --git a/src/owned_cell.rs b/src/owned_cell.rs
index c7e58e9..1c75452 100644
--- a/src/owned_cell.rs
+++ b/src/owned_cell.rs
@@ -14,7 +14,13 @@ use super::*;
 pub(crate) struct OwnedCell<O, D> {
     // The order here matters. dep must be dropped before owner.
     dep: D,
-    _owner: O,
+    owner: O,
+}
+
+impl<O, D> OwnedCell<O, D> {
+    pub fn owner(&self) -> &O {
+        &self.owner
+    }
 }
 
 impl<O, D> OwnedCell<O, D>
@@ -34,7 +40,7 @@ where
         // Deref knowing that when guard is moved, the deref will still be valid.
         let stable_deref: *mut O::Target = owner.deref_mut();
         Ok(Self {
-            _owner: owner,
+            owner,
             dep: make_dependent(unsafe { &mut *stable_deref })?,
         })
     }
@@ -58,7 +64,7 @@ where
         // Deref knowing that when guard is moved, the deref will still be valid.
         let stable_deref: *const O::Target = owner.deref();
         Ok(Self {
-            _owner: owner,
+            owner: owner,
             dep: make_dependent(unsafe { &*stable_deref })?,
         })
     }
diff --git a/src/ownedtx.rs b/src/ownedtx.rs
index 7198cb4..1e480c7 100644
--- a/src/ownedtx.rs
+++ b/src/ownedtx.rs
@@ -1,22 +1,27 @@
+use std::rc::Rc;
+use std::sync::RwLockReadGuard;
+
 use super::*;
+use crate::c_api::{PossumHandle, PossumReaderOwnedTransaction};
+use crate::handle::StartTransaction;
 use crate::tx::ReadTransactionOwned;
 
 /// A Sqlite Transaction and the mutex guard on the Connection it came from.
 // Not in the handle module since it can be owned by types other than Handle.
 pub struct OwnedTx<'handle> {
-    cell: OwnedTxInner<'handle, Transaction<'handle>>,
+    cell: OwnedTxInner<'handle, Transaction<'handle, &'handle Handle>>,
 }
 
 pub(crate) type OwnedTxInner<'h, T> = owned_cell::OwnedCell<MutexGuard<'h, Connection>, T>;
 
-impl<'a> From<OwnedTxInner<'a, Transaction<'a>>> for OwnedTx<'a> {
-    fn from(cell: OwnedTxInner<'a, Transaction<'a>>) -> Self {
+impl<'a> From<OwnedTxInner<'a, Transaction<'a, &'a Handle>>> for OwnedTx<'a> {
+    fn from(cell: OwnedTxInner<'a, Transaction<'a, &'a Handle>>) -> Self {
         Self { cell }
     }
 }
 
 impl<'a> Deref for OwnedTx<'a> {
-    type Target = Transaction<'a>;
+    type Target = Transaction<'a, &'a Handle>;
 
     fn deref(&self) -> &Self::Target {
         &self.cell
@@ -31,7 +36,7 @@ impl DerefMut for OwnedTx<'_> {
 
 impl<'a> OwnedTx<'a> {
     // Except for this move dependent dance it shouldn't be necessary to wrap the OwnedCell.
-    pub fn commit(self) -> Result<PostCommitWork<'a>> {
+    pub fn commit(self) -> Result<PostCommitWork<&'a Handle>> {
         self.cell.move_dependent(|tx| tx.commit())
     }
 }
@@ -65,38 +70,42 @@ pub(crate) trait OwnedTxTrait {
     fn transaction(&self) -> &Self::Tx;
 }
 
-// impl<'h, T> OwnedTxTrait for OwnedTxInner<'h, T> {
-//     type Tx = T;
-//
-//     fn end_tx<R>(self, take: impl FnOnce(Self::Tx)->R) ->R{
-//         self.move_dependent(take)
-//     }
-//
-//     fn as_handle(&self) -> &Handle {
-//         todo!()
-//     }
-//
-//     fn mut_transaction(&mut self) -> &mut Self::Tx {
-//         self.deref_mut()
-//     }
-// }
-
 impl<'h> OwnedTxTrait for OwnedTx<'h> {
-    type Tx = Transaction<'h>;
+    type Tx = Transaction<'h, &'h Handle>;
 
     fn end_tx<R>(self, take: impl FnOnce(Self::Tx) -> R) -> R {
-        todo!()
+        self.cell.move_dependent(take)
     }
 
     fn as_handle(&self) -> &Handle {
-        todo!()
+        self.cell.deref().handle()
     }
 
     fn mut_transaction(&mut self) -> &mut Self::Tx {
-        todo!()
+        self.deref_mut()
     }
 
     fn transaction(&self) -> &Self::Tx {
         self.cell.deref()
     }
 }
+
+impl OwnedTxTrait for PossumReaderOwnedTransaction<'static> {
+    type Tx = Transaction<'static, Rc<RwLockReadGuard<'static, Handle>>>;
+
+    fn end_tx<R>(self, take: impl FnOnce(Self::Tx) -> R) -> R {
+        self.move_dependent(|handle_guard|handle_guard.move_dependent(|conn_guard|conn_guard.move_dependent(take)))
+    }
+
+    fn as_handle(&self) -> &Handle {
+        self.deref().owner()
+    }
+
+    fn mut_transaction(&mut self) -> &mut Self::Tx {
+        self.deref_mut()
+    }
+
+    fn transaction(&self) -> &Self::Tx {
+        self.deref()
+    }
+}
diff --git a/src/reader.rs b/src/reader.rs
index fb9d4bd..215e79b 100644
--- a/src/reader.rs
+++ b/src/reader.rs
@@ -9,9 +9,10 @@ pub struct Reader<T> {
     pub(crate) reads: Reads,
 }
 
-impl<'a, T> Reader<T>
+impl<'a, T, H> Reader<T>
 where
-    T: OwnedTxTrait<Tx = Transaction<'a>>,
+    T: OwnedTxTrait<Tx = Transaction<'a, H>>,
+    H: AsRef<Handle>,
 {
     pub fn add(&mut self, key: &[u8]) -> rusqlite::Result<Option<Value>> {
         let res = self.owned_tx.mut_transaction().touch_for_read(key);
diff --git a/src/tx.rs b/src/tx.rs
index f866c79..2daa2b3 100644
--- a/src/tx.rs
+++ b/src/tx.rs
@@ -2,8 +2,8 @@ use super::*;
 
 /// This is more work to be done after the Handle conn mutex is released.
 #[must_use]
-pub(crate) struct PostCommitWork<'h> {
-    handle: &'h Handle,
+pub(crate) struct PostCommitWork<H> {
+    handle: H,
     deleted_values: Vec<NonzeroValueLocation>,
     altered_files: HashSet<FileId>,
 }
@@ -145,32 +145,37 @@ fn list_items_inner(
         .map_err(Into::into)
 }
 
-impl<'h> PostCommitWork<'h> {
+impl<H> PostCommitWork<H>
+where
+    H: AsRef<Handle>,
+{
     pub fn complete(self) {
         // This has to happen after exclusive files are flushed or there's a tendency for hole
         // punches to not persist. It doesn't fix the problem, but it significantly reduces it.
-        if !self.handle.instance_limits.disable_hole_punching {
-            self.handle.send_values_for_delete(self.deleted_values);
+        if !self.handle.as_ref().instance_limits.disable_hole_punching {
+            self.handle
+                .as_ref()
+                .send_values_for_delete(self.deleted_values);
         }
         // Forget any references to clones of files that have changed.
         for file_id in self.altered_files {
-            self.handle.clones.lock().unwrap().remove(&file_id);
+            self.handle.as_ref().clones.lock().unwrap().remove(&file_id);
         }
     }
 }
 
 // I can't work out how to have a reference to the Connection, and a transaction on it here at the
 // same time.
-pub struct Transaction<'h> {
+pub struct Transaction<'h, H> {
     tx: rusqlite::Transaction<'h>,
-    handle: &'h Handle,
+    handle: H,
     deleted_values: Vec<NonzeroValueLocation>,
     altered_files: HashSet<FileId>,
 }
 
 // TODO: Try doing this with a read trait that just requires a rusqlite::Transaction be available.
 
-impl<'t> ReadOnlyTransactionAccessor for Transaction<'t> {
+impl<'t, H> ReadOnlyTransactionAccessor for Transaction<'t, H> {
     fn readonly_transaction(&self) -> &rusqlite::Transaction {
         &self.tx
     }
@@ -178,17 +183,29 @@ impl<'t> ReadOnlyTransactionAccessor for Transaction<'t> {
 
 impl<T> ReadTransaction for T where T: ReadOnlyTransactionAccessor {}
 
-impl<'h> Transaction<'h> {
-    pub fn new(tx: rusqlite::Transaction<'h>, handle: &'h Handle) -> Self {
-        Self {
-            tx,
-            handle,
-            deleted_values: vec![],
-            altered_files: Default::default(),
-        }
+impl<'h, H> Transaction<'h, H> {
+    pub fn touch_for_read(&mut self, key: &[u8]) -> rusqlite::Result<Value> {
+        self.tx
+            .prepare_cached(&format!(
+                "update keys \
+                set last_used=cast(unixepoch('subsec')*1e3 as integer) \
+                where key=? \
+                returning {}",
+                value_columns_sql()
+            ))?
+            .query_row([key], Value::from_row)
+    }
+}
+
+impl<'h, H> Transaction<'h, H>
+where
+    H: AsRef<Handle>,
+{
+    pub fn handle(&self) -> &Handle {
+        self.handle.as_ref()
     }
 
-    pub(crate) fn commit(mut self) -> Result<PostCommitWork<'h>> {
+    pub(crate) fn commit(mut self) -> Result<PostCommitWork<H>> {
         self.apply_limits()?;
         self.tx.commit()?;
         Ok(PostCommitWork {
@@ -198,16 +215,30 @@ impl<'h> Transaction<'h> {
         })
     }
 
-    pub fn touch_for_read(&mut self, key: &[u8]) -> rusqlite::Result<Value> {
-        self.tx
-            .prepare_cached(&format!(
-                "update keys \
-                set last_used=cast(unixepoch('subsec')*1e3 as integer) \
-                where key=? \
-                returning {}",
-                value_columns_sql()
-            ))?
-            .query_row([key], Value::from_row)
+    pub fn apply_limits(&mut self) -> Result<()> {
+        if self.tx.transaction_state(None)? != rusqlite::TransactionState::Write {
+            return Ok(());
+        }
+        if let Some(max) = self.handle.as_ref().instance_limits.max_value_length_sum {
+            loop {
+                let actual = self
+                    .sum_value_length()
+                    .context("reading value_length sum")?;
+                if actual <= max {
+                    break;
+                }
+                self.evict_values(actual - max)?;
+            }
+        }
+        Ok(())
+    }
+    pub fn new(tx: rusqlite::Transaction<'h>, handle: H) -> Self {
+        Self {
+            tx,
+            handle,
+            deleted_values: vec![],
+            altered_files: Default::default(),
+        }
     }
 
     // TODO: Add a test for renaming onto itself.
@@ -332,24 +363,6 @@ impl<'h> Transaction<'h> {
         }
     }
 
-    pub fn apply_limits(&mut self) -> Result<()> {
-        if self.tx.transaction_state(None)? != rusqlite::TransactionState::Write {
-            return Ok(());
-        }
-        if let Some(max) = self.handle.instance_limits.max_value_length_sum {
-            loop {
-                let actual = self
-                    .sum_value_length()
-                    .context("reading value_length sum")?;
-                if actual <= max {
-                    break;
-                }
-                self.evict_values(actual - max)?;
-            }
-        }
-        Ok(())
-    }
-
     pub fn evict_values(&mut self, target_bytes: u64) -> Result<()> {
         let mut stmt = self.tx.prepare_cached(&format!(
             "delete from keys where key_id in (\