diff --git a/services/syncbase/store/leveldb/db.go b/services/syncbase/store/leveldb/db.go
index 028b1ad..0430561 100644
--- a/services/syncbase/store/leveldb/db.go
+++ b/services/syncbase/store/leveldb/db.go
@@ -11,16 +11,17 @@
 // #include "syncbase_leveldb.h"
 import "C"
 import (
-	"container/list"
 	"fmt"
 	"sync"
 	"unsafe"
 
 	"v.io/syncbase/x/ref/services/syncbase/store"
+	"v.io/syncbase/x/ref/services/syncbase/store/transactions"
 	"v.io/v23/verror"
 )
 
-// db is a wrapper around LevelDB that implements the store.Store interface.
+// db is a wrapper around LevelDB that implements the transactions.BatchStore
+// interface.
 type db struct {
 	// mu protects the state of the db.
 	mu   sync.RWMutex
@@ -30,21 +31,8 @@
 	readOptions  *C.leveldb_readoptions_t
 	writeOptions *C.leveldb_writeoptions_t
 	err          error
-
-	// txmu protects the transaction-related variables below, and is also held
-	// during transaction commits. It must always be acquired before mu.
-	txmu sync.Mutex
-	// txEvents is a queue of create/commit transaction events.
-	txEvents         *list.List
-	txSequenceNumber uint64
-	// txTable is a set of keys written by recent transactions. This set
-	// includes all write sets of transactions committed after the oldest living
-	// (in-flight) transaction.
-	txTable *trie
 }
 
-var _ store.Store = (*db)(nil)
-
 type OpenOptions struct {
 	CreateIfMissing bool
 	ErrorIfExists   bool
@@ -76,14 +64,12 @@
 	}
 	readOptions := C.leveldb_readoptions_create()
 	C.leveldb_readoptions_set_verify_checksums(readOptions, 1)
-	return &db{
+	return transactions.Wrap(&db{
 		node:         store.NewResourceNode(),
 		cDb:          cDb,
 		readOptions:  readOptions,
 		writeOptions: C.leveldb_writeoptions_create(),
-		txEvents:     list.New(),
-		txTable:      newTrie(),
-	}, nil
+	}), nil
 }
 
 // Close implements the store.Store interface.
@@ -91,7 +77,7 @@
 	d.mu.Lock()
 	defer d.mu.Unlock()
 	if d.err != nil {
-		return convertError(d.err)
+		return store.ConvertError(d.err)
 	}
 	d.node.Close()
 	C.leveldb_close(d.cDb)
@@ -130,37 +116,6 @@
 	return newStream(d, d.node, start, limit, d.readOptions)
 }
 
-// Put implements the store.StoreWriter interface.
-func (d *db) Put(key, value []byte) error {
-	write := store.WriteOp{
-		T:     store.PutOp,
-		Key:   key,
-		Value: value,
-	}
-	return d.write([]store.WriteOp{write}, d.writeOptions)
-}
-
-// Delete implements the store.StoreWriter interface.
-func (d *db) Delete(key []byte) error {
-	write := store.WriteOp{
-		T:   store.DeleteOp,
-		Key: key,
-	}
-	return d.write([]store.WriteOp{write}, d.writeOptions)
-}
-
-// NewTransaction implements the store.Store interface.
-func (d *db) NewTransaction() store.Transaction {
-	d.txmu.Lock()
-	defer d.txmu.Unlock()
-	d.mu.RLock()
-	defer d.mu.RUnlock()
-	if d.err != nil {
-		return &store.InvalidTransaction{Error: d.err}
-	}
-	return newTransaction(d, d.node)
-}
-
 // NewSnapshot implements the store.Store interface.
 func (d *db) NewSnapshot() store.Snapshot {
 	d.mu.RLock()
@@ -171,16 +126,8 @@
 	return newSnapshot(d, d.node)
 }
 
-// write writes a batch and adds all written keys to txTable.
-// TODO(rogulenko): remove this method.
-func (d *db) write(batch []store.WriteOp, cOpts *C.leveldb_writeoptions_t) error {
-	d.txmu.Lock()
-	defer d.txmu.Unlock()
-	return d.writeLocked(batch, cOpts)
-}
-
-// writeLocked is like write(), but it assumes txmu is held.
-func (d *db) writeLocked(batch []store.WriteOp, cOpts *C.leveldb_writeoptions_t) error {
+// WriteBatch implements the transactions.BatchStore interface.
+func (d *db) WriteBatch(batch ...transactions.WriteOp) error {
 	d.mu.Lock()
 	defer d.mu.Unlock()
 	if d.err != nil {
@@ -190,11 +137,11 @@
 	defer C.leveldb_writebatch_destroy(cBatch)
 	for _, write := range batch {
 		switch write.T {
-		case store.PutOp:
+		case transactions.PutOp:
 			cKey, cKeyLen := cSlice(write.Key)
 			cVal, cValLen := cSlice(write.Value)
 			C.leveldb_writebatch_put(cBatch, cKey, cKeyLen, cVal, cValLen)
-		case store.DeleteOp:
+		case transactions.DeleteOp:
 			cKey, cKeyLen := cSlice(write.Key)
 			C.leveldb_writebatch_delete(cBatch, cKey, cKeyLen)
 		default:
@@ -202,32 +149,8 @@
 		}
 	}
 	var cError *C.char
-	C.leveldb_write(d.cDb, cOpts, cBatch, &cError)
-	if err := goError(cError); err != nil {
-		return err
-	}
-	if d.txEvents.Len() == 0 {
-		return nil
-	}
-	d.trackBatch(batch)
-	return nil
-}
-
-// trackBatch writes the batch to txTable and adds a commit event to txEvents.
-func (d *db) trackBatch(batch []store.WriteOp) {
-	// TODO(rogulenko): do GC.
-	d.txSequenceNumber++
-	seq := d.txSequenceNumber
-	var keys [][]byte
-	for _, write := range batch {
-		d.txTable.add(write.Key, seq)
-		keys = append(keys, write.Key)
-	}
-	tx := &commitedTransaction{
-		seq:   seq,
-		batch: keys,
-	}
-	d.txEvents.PushBack(tx)
+	C.leveldb_write(d.cDb, d.writeOptions, cBatch, &cError)
+	return goError(cError)
 }
 
 // getWithOpts returns the value for the given key.
@@ -236,7 +159,7 @@
 	d.mu.RLock()
 	defer d.mu.RUnlock()
 	if d.err != nil {
-		return valbuf, convertError(d.err)
+		return valbuf, store.ConvertError(d.err)
 	}
 	var cError *C.char
 	var valLen C.size_t
diff --git a/services/syncbase/store/leveldb/snapshot.go b/services/syncbase/store/leveldb/snapshot.go
index f845fa6..a403127 100644
--- a/services/syncbase/store/leveldb/snapshot.go
+++ b/services/syncbase/store/leveldb/snapshot.go
@@ -50,7 +50,7 @@
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if s.err != nil {
-		return convertError(s.err)
+		return store.ConvertError(s.err)
 	}
 	s.node.Close()
 	C.leveldb_readoptions_destroy(s.cOpts)
@@ -66,7 +66,7 @@
 	s.mu.RLock()
 	defer s.mu.RUnlock()
 	if s.err != nil {
-		return valbuf, convertError(s.err)
+		return valbuf, store.ConvertError(s.err)
 	}
 	return s.d.getWithOpts(key, valbuf, s.cOpts)
 }
diff --git a/services/syncbase/store/leveldb/stream.go b/services/syncbase/store/leveldb/stream.go
index 3102d74..2d592b4 100644
--- a/services/syncbase/store/leveldb/stream.go
+++ b/services/syncbase/store/leveldb/stream.go
@@ -113,7 +113,7 @@
 func (s *stream) Err() error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	return convertError(s.err)
+	return store.ConvertError(s.err)
 }
 
 // Cancel implements the store.Stream interface.
diff --git a/services/syncbase/store/leveldb/util.go b/services/syncbase/store/leveldb/util.go
index 7d92527..dce69bb 100644
--- a/services/syncbase/store/leveldb/util.go
+++ b/services/syncbase/store/leveldb/util.go
@@ -44,7 +44,3 @@
 	})
 	return *(*[]byte)(ptr)
 }
-
-func convertError(err error) error {
-	return verror.Convert(verror.IDAction{}, nil, err)
-}
diff --git a/services/syncbase/store/memstore/snapshot.go b/services/syncbase/store/memstore/snapshot.go
index 7d0fff1..310f6e2 100644
--- a/services/syncbase/store/memstore/snapshot.go
+++ b/services/syncbase/store/memstore/snapshot.go
@@ -42,7 +42,7 @@
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if s.err != nil {
-		return convertError(s.err)
+		return store.ConvertError(s.err)
 	}
 	s.node.Close()
 	s.err = verror.New(verror.ErrCanceled, nil, store.ErrMsgAbortedSnapshot)
@@ -54,7 +54,7 @@
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if s.err != nil {
-		return valbuf, convertError(s.err)
+		return valbuf, store.ConvertError(s.err)
 	}
 	value, ok := s.data[string(key)]
 	if !ok {
diff --git a/services/syncbase/store/memstore/store.go b/services/syncbase/store/memstore/store.go
index e246fc4..15a2988 100644
--- a/services/syncbase/store/memstore/store.go
+++ b/services/syncbase/store/memstore/store.go
@@ -7,9 +7,11 @@
 package memstore
 
 import (
+	"fmt"
 	"sync"
 
 	"v.io/syncbase/x/ref/services/syncbase/store"
+	"v.io/syncbase/x/ref/services/syncbase/store/transactions"
 	"v.io/v23/verror"
 )
 
@@ -18,20 +20,14 @@
 	node *store.ResourceNode
 	data map[string][]byte
 	err  error
-	// Most recent sequence number handed out.
-	lastSeq uint64
-	// Value of lastSeq at the time of the most recent commit.
-	lastCommitSeq uint64
 }
 
-var _ store.Store = (*memstore)(nil)
-
 // New creates a new memstore.
 func New() store.Store {
-	return &memstore{
+	return transactions.Wrap(&memstore{
 		data: map[string][]byte{},
 		node: store.NewResourceNode(),
-	}
+	})
 }
 
 // Close implements the store.Store interface.
@@ -39,7 +35,7 @@
 	st.mu.Lock()
 	defer st.mu.Unlock()
 	if st.err != nil {
-		return convertError(st.err)
+		return store.ConvertError(st.err)
 	}
 	st.node.Close()
 	st.err = verror.New(verror.ErrCanceled, nil, store.ErrMsgClosedStore)
@@ -51,7 +47,7 @@
 	st.mu.Lock()
 	defer st.mu.Unlock()
 	if st.err != nil {
-		return valbuf, convertError(st.err)
+		return valbuf, store.ConvertError(st.err)
 	}
 	value, ok := st.data[string(key)]
 	if !ok {
@@ -71,31 +67,6 @@
 	return newSnapshot(st, st.node).Scan(start, limit)
 }
 
-// Put implements the store.StoreWriter interface.
-func (st *memstore) Put(key, value []byte) error {
-	return store.RunInTransaction(st, func(tx store.Transaction) error {
-		return tx.Put(key, value)
-	})
-}
-
-// Delete implements the store.StoreWriter interface.
-func (st *memstore) Delete(key []byte) error {
-	return store.RunInTransaction(st, func(tx store.Transaction) error {
-		return tx.Delete(key)
-	})
-}
-
-// NewTransaction implements the store.Store interface.
-func (st *memstore) NewTransaction() store.Transaction {
-	st.mu.Lock()
-	defer st.mu.Unlock()
-	if st.err != nil {
-		return &store.InvalidTransaction{Error: st.err}
-	}
-	st.lastSeq++
-	return newTransaction(st, st.node, st.lastSeq)
-}
-
 // NewSnapshot implements the store.Store interface.
 func (st *memstore) NewSnapshot() store.Snapshot {
 	st.mu.Lock()
@@ -105,3 +76,23 @@
 	}
 	return newSnapshot(st, st.node)
 }
+
+// WriteBatch implements the transactions.BatchStore interface.
+func (st *memstore) WriteBatch(batch ...transactions.WriteOp) error {
+	st.mu.Lock()
+	defer st.mu.Unlock()
+	if st.err != nil {
+		return store.ConvertError(st.err)
+	}
+	for _, write := range batch {
+		switch write.T {
+		case transactions.PutOp:
+			st.data[string(write.Key)] = write.Value
+		case transactions.DeleteOp:
+			delete(st.data, string(write.Key))
+		default:
+			panic(fmt.Sprintf("unknown write operation type: %v", write.T))
+		}
+	}
+	return nil
+}
diff --git a/services/syncbase/store/memstore/store_test.go b/services/syncbase/store/memstore/store_test.go
index fef07b1..0b04032 100644
--- a/services/syncbase/store/memstore/store_test.go
+++ b/services/syncbase/store/memstore/store_test.go
@@ -40,6 +40,10 @@
 	runTest(t, test.RunReadWriteRandomTest)
 }
 
+func TestConcurrentTransactions(t *testing.T) {
+	runTest(t, test.RunConcurrentTransactionsTest)
+}
+
 func TestTransactionState(t *testing.T) {
 	runTest(t, test.RunTransactionStateTest)
 }
diff --git a/services/syncbase/store/memstore/stream.go b/services/syncbase/store/memstore/stream.go
index a8780be..345ea93 100644
--- a/services/syncbase/store/memstore/stream.go
+++ b/services/syncbase/store/memstore/stream.go
@@ -87,7 +87,7 @@
 func (s *stream) Err() error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	return convertError(s.err)
+	return store.ConvertError(s.err)
 }
 
 // Cancel implements the store.Stream interface.
diff --git a/services/syncbase/store/memstore/transaction.go b/services/syncbase/store/memstore/transaction.go
deleted file mode 100644
index 5095a5f..0000000
--- a/services/syncbase/store/memstore/transaction.go
+++ /dev/null
@@ -1,168 +0,0 @@
-// Copyright 2015 The Vanadium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style
-// license that can be found in the LICENSE file.
-
-package memstore
-
-import (
-	"sync"
-	"time"
-
-	"v.io/syncbase/x/ref/services/syncbase/store"
-	"v.io/v23/verror"
-)
-
-const (
-	txnTimeout = time.Duration(5) * time.Second
-)
-
-type transaction struct {
-	mu   sync.Mutex
-	node *store.ResourceNode
-	st   *memstore
-	sn   *snapshot
-	// The following fields are used to determine whether method calls should
-	// error out.
-	err         error
-	seq         uint64
-	createdTime time.Time
-	// The following fields track writes performed against this transaction.
-	puts    map[string][]byte
-	deletes map[string]struct{}
-}
-
-var _ store.Transaction = (*transaction)(nil)
-
-func newTransaction(st *memstore, parent *store.ResourceNode, seq uint64) *transaction {
-	node := store.NewResourceNode()
-	tx := &transaction{
-		node:        node,
-		st:          st,
-		sn:          newSnapshot(st, node),
-		seq:         seq,
-		createdTime: time.Now(),
-		puts:        map[string][]byte{},
-		deletes:     map[string]struct{}{},
-	}
-	parent.AddChild(tx.node, func() {
-		tx.Abort()
-	})
-	return tx
-}
-
-func (tx *transaction) expired() bool {
-	return time.Now().After(tx.createdTime.Add(txnTimeout))
-}
-
-func (tx *transaction) error() error {
-	if tx.err != nil {
-		return convertError(tx.err)
-	}
-	if tx.expired() {
-		return verror.New(verror.ErrBadState, nil, store.ErrMsgExpiredTxn)
-	}
-	return nil
-}
-
-// Get implements the store.StoreReader interface.
-func (tx *transaction) Get(key, valbuf []byte) ([]byte, error) {
-	tx.mu.Lock()
-	defer tx.mu.Unlock()
-	if err := tx.error(); err != nil {
-		return valbuf, err
-	}
-
-	// Reflect the state of the transaction: the "puts" and "deletes"
-	// override the values in the transaction snapshot.
-	keyStr := string(key)
-	if val, ok := tx.puts[keyStr]; ok {
-		return val, nil
-	}
-	if _, ok := tx.deletes[keyStr]; ok {
-		return valbuf, verror.New(store.ErrUnknownKey, nil, keyStr)
-	}
-
-	return tx.sn.Get(key, valbuf)
-}
-
-// Scan implements the store.StoreReader interface.
-func (tx *transaction) Scan(start, limit []byte) store.Stream {
-	tx.mu.Lock()
-	defer tx.mu.Unlock()
-	if err := tx.error(); err != nil {
-		return &store.InvalidStream{Error: err}
-	}
-
-	// Create an array of store.WriteOps as it is needed to merge
-	// the snaphot stream with the uncommitted changes.
-	var writes []store.WriteOp
-	for k, v := range tx.puts {
-		writes = append(writes, store.WriteOp{T: store.PutOp, Key: []byte(k), Value: v})
-	}
-	for k, _ := range tx.deletes {
-		writes = append(writes, store.WriteOp{T: store.DeleteOp, Key: []byte(k), Value: []byte{}})
-	}
-
-	// Return a stream which merges the snaphot stream with the uncommitted changes.
-	return store.MergeWritesWithStream(tx.sn, writes, start, limit)
-}
-
-// Put implements the store.StoreWriter interface.
-func (tx *transaction) Put(key, value []byte) error {
-	tx.st.mu.Lock()
-	defer tx.st.mu.Unlock()
-	if err := tx.error(); err != nil {
-		return err
-	}
-	delete(tx.deletes, string(key))
-	tx.puts[string(key)] = value
-	return nil
-}
-
-// Delete implements the store.StoreWriter interface.
-func (tx *transaction) Delete(key []byte) error {
-	tx.st.mu.Lock()
-	defer tx.st.mu.Unlock()
-	if err := tx.error(); err != nil {
-		return err
-	}
-	delete(tx.puts, string(key))
-	tx.deletes[string(key)] = struct{}{}
-	return nil
-}
-
-// Commit implements the store.Transaction interface.
-func (tx *transaction) Commit() error {
-	tx.mu.Lock()
-	defer tx.mu.Unlock()
-	if err := tx.error(); err != nil {
-		return err
-	}
-	tx.err = verror.New(verror.ErrBadState, nil, store.ErrMsgCommittedTxn)
-	tx.node.Close()
-	tx.st.mu.Lock()
-	defer tx.st.mu.Unlock()
-	if tx.seq <= tx.st.lastCommitSeq {
-		return store.NewErrConcurrentTransaction(nil)
-	}
-	for k, v := range tx.puts {
-		tx.st.data[k] = v
-	}
-	for k := range tx.deletes {
-		delete(tx.st.data, k)
-	}
-	tx.st.lastCommitSeq = tx.st.lastSeq
-	return nil
-}
-
-// Abort implements the store.Transaction interface.
-func (tx *transaction) Abort() error {
-	tx.mu.Lock()
-	defer tx.mu.Unlock()
-	if err := tx.error(); err != nil {
-		return err
-	}
-	tx.node.Close()
-	tx.err = verror.New(verror.ErrCanceled, nil, store.ErrMsgAbortedTxn)
-	return nil
-}
diff --git a/services/syncbase/store/memstore/util.go b/services/syncbase/store/memstore/util.go
deleted file mode 100644
index 8dc3a7a..0000000
--- a/services/syncbase/store/memstore/util.go
+++ /dev/null
@@ -1,13 +0,0 @@
-// Copyright 2015 The Vanadium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style
-// license that can be found in the LICENSE file.
-
-package memstore
-
-import (
-	"v.io/v23/verror"
-)
-
-func convertError(err error) error {
-	return verror.Convert(verror.IDAction{}, nil, err)
-}
diff --git a/services/syncbase/store/transactions/manager.go b/services/syncbase/store/transactions/manager.go
new file mode 100644
index 0000000..254812f
--- /dev/null
+++ b/services/syncbase/store/transactions/manager.go
@@ -0,0 +1,194 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package transactions
+
+import (
+	"container/list"
+	"sync"
+
+	"v.io/syncbase/x/ref/services/syncbase/store"
+	"v.io/v23/verror"
+)
+
+// BatchStore is a CRUD-capable storage engine that supports atomic batch
+// writes. BatchStore doesn't support transactions.
+// This interface is a Go version of the C++ LevelDB interface. It serves as
+// an intermediate interface between store.Store and the LevelDB interface.
+type BatchStore interface {
+	store.StoreReader
+
+	// WriteBatch atomically writes a list of write operations to the database.
+	WriteBatch(batch ...WriteOp) error
+
+	// Close closes the store.
+	Close() error
+
+	// NewSnapshot creates a snapshot.
+	NewSnapshot() store.Snapshot
+}
+
+// manager handles transaction-related operations of the store.
+type manager struct {
+	BatchStore
+	// mu protects the variables below, and is also held during transaction
+	// commits. It must always be acquired before the store-level lock.
+	mu sync.Mutex
+	// events is a queue of create/commit transaction events.
+	events *list.List
+	seq    uint64
+	// txTable is a set of keys written by recent transactions. This set
+	// includes all write sets of transactions committed after the oldest living
+	// (in-flight) transaction.
+	txTable *trie
+}
+
+// commitedTransaction is only used as an element of manager.events.
+type commitedTransaction struct {
+	seq   uint64
+	batch [][]byte
+}
+
+// Wrap wraps the BatchStore with transaction functionality.
+func Wrap(bs BatchStore) store.Store {
+	return &manager{
+		BatchStore: bs,
+		events:     list.New(),
+		txTable:    newTrie(),
+	}
+}
+
+// Close implements the store.Store interface.
+func (mg *manager) Close() error {
+	mg.mu.Lock()
+	defer mg.mu.Unlock()
+	if mg.txTable == nil {
+		return verror.New(verror.ErrCanceled, nil, store.ErrMsgClosedStore)
+	}
+	mg.BatchStore.Close()
+	for event := mg.events.Front(); event != nil; event = event.Next() {
+		if tx, ok := event.Value.(*transaction); ok {
+			// tx.Abort() internally removes tx from the mg.events list under
+			// the mg.mu lock. To brake the cyclic dependency, we set tx.event
+			// to nil.
+			tx.mu.Lock()
+			tx.event = nil
+			tx.mu.Unlock()
+			tx.Abort()
+		}
+	}
+	mg.events = nil
+	mg.txTable = nil
+	return nil
+}
+
+// NewTransaction implements the store.Store interface.
+func (mg *manager) NewTransaction() store.Transaction {
+	mg.mu.Lock()
+	defer mg.mu.Unlock()
+	if mg.txTable == nil {
+		return &store.InvalidTransaction{
+			Error: verror.New(verror.ErrCanceled, nil, store.ErrMsgClosedStore),
+		}
+	}
+	return newTransaction(mg)
+}
+
+// Put implements the store.StoreWriter interface.
+func (mg *manager) Put(key, value []byte) error {
+	mg.mu.Lock()
+	defer mg.mu.Unlock()
+	if mg.txTable == nil {
+		return verror.New(verror.ErrCanceled, nil, store.ErrMsgClosedStore)
+	}
+	write := WriteOp{
+		T:     PutOp,
+		Key:   key,
+		Value: value,
+	}
+	if err := mg.BatchStore.WriteBatch(write); err != nil {
+		return err
+	}
+	mg.trackBatch(write)
+	return nil
+}
+
+// Delete implements the store.StoreWriter interface.
+func (mg *manager) Delete(key []byte) error {
+	mg.mu.Lock()
+	defer mg.mu.Unlock()
+	if mg.txTable == nil {
+		return verror.New(verror.ErrCanceled, nil, store.ErrMsgClosedStore)
+	}
+	write := WriteOp{
+		T:   DeleteOp,
+		Key: key,
+	}
+	if err := mg.BatchStore.WriteBatch(write); err != nil {
+		return err
+	}
+	mg.trackBatch(write)
+	return nil
+}
+
+// trackBatch writes the batch to txTable and adds a commit event to
+// the events queue.
+// Assumes mu is held.
+func (mg *manager) trackBatch(batch ...WriteOp) {
+	if mg.events.Len() == 0 {
+		return
+	}
+	// TODO(rogulenko): do GC.
+	mg.seq++
+	var keys [][]byte
+	for _, write := range batch {
+		mg.txTable.add(write.Key, mg.seq)
+		keys = append(keys, write.Key)
+	}
+	tx := &commitedTransaction{
+		seq:   mg.seq,
+		batch: keys,
+	}
+	mg.events.PushBack(tx)
+}
+
+//////////////////////////////////////////////////////////////
+// Read and Write types used for storing transcation reads
+// and uncommitted writes.
+
+type WriteType int
+
+const (
+	PutOp WriteType = iota
+	DeleteOp
+)
+
+type WriteOp struct {
+	T     WriteType
+	Key   []byte
+	Value []byte
+}
+
+type scanRange struct {
+	Start, Limit []byte
+}
+
+type readSet struct {
+	Keys   [][]byte
+	Ranges []scanRange
+}
+
+type writeOpArray []WriteOp
+
+func (a writeOpArray) Len() int {
+	return len(a)
+}
+
+func (a writeOpArray) Less(i, j int) bool {
+	return string(a[i].Key) < string(a[j].Key)
+}
+
+func (a writeOpArray) Swap(i, j int) {
+	a[i], a[j] = a[j], a[i]
+}
diff --git a/services/syncbase/store/merged_stream.go b/services/syncbase/store/transactions/merged_stream.go
similarity index 89%
rename from services/syncbase/store/merged_stream.go
rename to services/syncbase/store/transactions/merged_stream.go
index e8c1058..4ab10be 100644
--- a/services/syncbase/store/merged_stream.go
+++ b/services/syncbase/store/transactions/merged_stream.go
@@ -2,10 +2,12 @@
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.
 
-package store
+package transactions
 
 import (
 	"sort"
+
+	"v.io/syncbase/x/ref/services/syncbase/store"
 )
 
 //////////////////////////////////////////////////////////////
@@ -19,7 +21,7 @@
 // changes to be passed in as an array of WriteOp.
 
 // Create a new stream which merges a snapshot stream with an array of write operations.
-func MergeWritesWithStream(sn Snapshot, w []WriteOp, start, limit []byte) Stream {
+func mergeWritesWithStream(sn store.Snapshot, w []WriteOp, start, limit []byte) store.Stream {
 	// Collect writes with the range specified, then sort them.
 	// Note: Writes could contain more than one write for a given key.
 	//       The last write is the current state.
@@ -29,7 +31,7 @@
 			writesMap[string(write.Key)] = write
 		}
 	}
-	var writesArray WriteOpArray
+	var writesArray writeOpArray
 	for _, writeOp := range writesMap {
 		writesArray = append(writesArray, writeOp)
 	}
@@ -45,7 +47,7 @@
 }
 
 type mergedStream struct {
-	snapshotStream      Stream
+	snapshotStream      store.Stream
 	writesArray         []WriteOp
 	writesCursor        int
 	unusedSnapshotValue bool
@@ -94,8 +96,8 @@
 		s.writesCursor++
 		return false
 	}
-	s.key = CopyBytes(s.key, s.writesArray[s.writesCursor].Key)
-	s.value = CopyBytes(s.value, s.writesArray[s.writesCursor].Value)
+	s.key = store.CopyBytes(s.key, s.writesArray[s.writesCursor].Key)
+	s.value = store.CopyBytes(s.value, s.writesArray[s.writesCursor].Value)
 	s.writesCursor++
 	return true
 }
@@ -122,7 +124,7 @@
 	if !s.hasValue {
 		panic("nothing staged")
 	}
-	return CopyBytes(keybuf, s.key)
+	return store.CopyBytes(keybuf, s.key)
 }
 
 // Value implements the Stream interface.
@@ -130,7 +132,7 @@
 	if !s.hasValue {
 		panic("nothing staged")
 	}
-	return CopyBytes(valbuf, s.value)
+	return store.CopyBytes(valbuf, s.value)
 }
 
 // Err implements the Stream interface.
diff --git a/services/syncbase/store/leveldb/transaction.go b/services/syncbase/store/transactions/transaction.go
similarity index 67%
rename from services/syncbase/store/leveldb/transaction.go
rename to services/syncbase/store/transactions/transaction.go
index b8093f6..dcf7569 100644
--- a/services/syncbase/store/leveldb/transaction.go
+++ b/services/syncbase/store/transactions/transaction.go
@@ -2,10 +2,8 @@
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.
 
-package leveldb
+package transactions
 
-// #include "leveldb/c.h"
-import "C"
 import (
 	"bytes"
 	"container/list"
@@ -16,59 +14,41 @@
 	"v.io/x/lib/vlog"
 )
 
-// commitedTransaction is only used as an element of db.txEvents.
-type commitedTransaction struct {
-	seq   uint64
-	batch [][]byte
-}
-
-// transaction is a wrapper around LevelDB WriteBatch that implements
-// the store.Transaction interface.
+// transaction is a wrapper on top of a BatchWriter and a store.Snapshot that
+// implements the store.Transaction interface.
 type transaction struct {
 	// mu protects the state of the transaction.
 	mu       sync.Mutex
-	node     *store.ResourceNode
-	d        *db
+	mg       *manager
 	seq      uint64
 	event    *list.Element // pointer to element of db.txEvents
 	snapshot store.Snapshot
-	reads    store.ReadSet
-	writes   []store.WriteOp
-	cOpts    *C.leveldb_writeoptions_t
+	reads    readSet
+	writes   []WriteOp
 	err      error
 }
 
 var _ store.Transaction = (*transaction)(nil)
 
-func newTransaction(d *db, parent *store.ResourceNode) *transaction {
-	node := store.NewResourceNode()
-	snapshot := newSnapshot(d, node)
+func newTransaction(mg *manager) *transaction {
 	tx := &transaction{
-		node:     node,
-		d:        d,
-		snapshot: snapshot,
-		seq:      d.txSequenceNumber,
-		cOpts:    d.writeOptions,
+		mg:       mg,
+		snapshot: mg.BatchStore.NewSnapshot(),
+		seq:      mg.seq,
 	}
-	tx.event = d.txEvents.PushFront(tx)
-	parent.AddChild(tx.node, func() {
-		tx.Abort()
-	})
+	tx.event = mg.events.PushFront(tx)
 	return tx
 }
 
-// close frees allocated C objects and releases acquired locks.
+// close removes this transaction from the mg.events queue and aborts
+// the underlying snapshot.
 // Assumes mu is held.
 func (tx *transaction) close() {
 	tx.removeEvent()
-	tx.node.Close()
-	if tx.cOpts != tx.d.writeOptions {
-		C.leveldb_writeoptions_destroy(tx.cOpts)
-	}
-	tx.cOpts = nil
+	tx.snapshot.Abort()
 }
 
-// removeEvent removes this transaction from the db.txEvents queue.
+// removeEvent removes this transaction from the mg.events queue.
 // Assumes mu is held.
 func (tx *transaction) removeEvent() {
 	// This can happen if the transaction was committed, since Commit()
@@ -76,9 +56,9 @@
 	if tx.event == nil {
 		return
 	}
-	tx.d.txmu.Lock()
-	tx.d.txEvents.Remove(tx.event)
-	tx.d.txmu.Unlock()
+	tx.mg.mu.Lock()
+	tx.mg.events.Remove(tx.event)
+	tx.mg.mu.Unlock()
 	tx.event = nil
 }
 
@@ -87,7 +67,7 @@
 	tx.mu.Lock()
 	defer tx.mu.Unlock()
 	if tx.err != nil {
-		return valbuf, convertError(tx.err)
+		return valbuf, store.ConvertError(tx.err)
 	}
 	tx.reads.Keys = append(tx.reads.Keys, key)
 
@@ -99,7 +79,7 @@
 	for i := len(tx.writes) - 1; i >= 0; i-- {
 		op := &tx.writes[i]
 		if bytes.Equal(op.Key, key) {
-			if op.T == store.PutOp {
+			if op.T == PutOp {
 				return op.Value, nil
 			}
 			return valbuf, verror.New(store.ErrUnknownKey, nil, string(key))
@@ -117,13 +97,13 @@
 		return &store.InvalidStream{Error: tx.err}
 	}
 
-	tx.reads.Ranges = append(tx.reads.Ranges, store.ScanRange{
+	tx.reads.Ranges = append(tx.reads.Ranges, scanRange{
 		Start: start,
 		Limit: limit,
 	})
 
 	// Return a stream which merges the snaphot stream with the uncommitted changes.
-	return store.MergeWritesWithStream(tx.snapshot, tx.writes, start, limit)
+	return mergeWritesWithStream(tx.snapshot, tx.writes, start, limit)
 }
 
 // Put implements the store.StoreWriter interface.
@@ -131,10 +111,10 @@
 	tx.mu.Lock()
 	defer tx.mu.Unlock()
 	if tx.err != nil {
-		return convertError(tx.err)
+		return store.ConvertError(tx.err)
 	}
-	tx.writes = append(tx.writes, store.WriteOp{
-		T:     store.PutOp,
+	tx.writes = append(tx.writes, WriteOp{
+		T:     PutOp,
 		Key:   key,
 		Value: value,
 	})
@@ -146,10 +126,10 @@
 	tx.mu.Lock()
 	defer tx.mu.Unlock()
 	if tx.err != nil {
-		return convertError(tx.err)
+		return store.ConvertError(tx.err)
 	}
-	tx.writes = append(tx.writes, store.WriteOp{
-		T:   store.DeleteOp,
+	tx.writes = append(tx.writes, WriteOp{
+		T:   DeleteOp,
 		Key: key,
 	})
 	return nil
@@ -157,16 +137,16 @@
 
 // validateReadSet returns true iff the read set of this transaction has not
 // been invalidated by other transactions.
-// Assumes tx.d.txmu is held.
+// Assumes tx.mg.mu is held.
 func (tx *transaction) validateReadSet() bool {
 	for _, key := range tx.reads.Keys {
-		if tx.d.txTable.get(key) > tx.seq {
+		if tx.mg.txTable.get(key) > tx.seq {
 			vlog.VI(3).Infof("key conflict: %q", key)
 			return false
 		}
 	}
 	for _, r := range tx.reads.Ranges {
-		if tx.d.txTable.rangeMax(r.Start, r.Limit) > tx.seq {
+		if tx.mg.txTable.rangeMax(r.Start, r.Limit) > tx.seq {
 			vlog.VI(3).Infof("range conflict: {%q, %q}", r.Start, r.Limit)
 			return false
 		}
@@ -180,7 +160,7 @@
 	tx.mu.Lock()
 	defer tx.mu.Unlock()
 	if tx.err != nil {
-		return convertError(tx.err)
+		return store.ConvertError(tx.err)
 	}
 	tx.err = verror.New(verror.ErrBadState, nil, store.ErrMsgCommittedTxn)
 	// Explicitly remove this transaction from the event queue. If this was the
@@ -188,12 +168,16 @@
 	// not add this transaction's write set to txTable.
 	tx.removeEvent()
 	defer tx.close()
-	tx.d.txmu.Lock()
-	defer tx.d.txmu.Unlock()
+	tx.mg.mu.Lock()
+	defer tx.mg.mu.Unlock()
 	if !tx.validateReadSet() {
 		return store.NewErrConcurrentTransaction(nil)
 	}
-	return tx.d.writeLocked(tx.writes, tx.cOpts)
+	if err := tx.mg.BatchStore.WriteBatch(tx.writes...); err != nil {
+		return err
+	}
+	tx.mg.trackBatch(tx.writes...)
+	return nil
 }
 
 // Abort implements the store.Transaction interface.
@@ -201,7 +185,7 @@
 	tx.mu.Lock()
 	defer tx.mu.Unlock()
 	if tx.err != nil {
-		return convertError(tx.err)
+		return store.ConvertError(tx.err)
 	}
 	tx.err = verror.New(verror.ErrCanceled, nil, store.ErrMsgAbortedTxn)
 	tx.close()
diff --git a/services/syncbase/store/leveldb/trie.go b/services/syncbase/store/transactions/trie.go
similarity index 98%
rename from services/syncbase/store/leveldb/trie.go
rename to services/syncbase/store/transactions/trie.go
index 78a6982..51a99a3 100644
--- a/services/syncbase/store/leveldb/trie.go
+++ b/services/syncbase/store/transactions/trie.go
@@ -2,7 +2,7 @@
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.
 
-package leveldb
+package transactions
 
 import (
 	"fmt"
diff --git a/services/syncbase/store/util.go b/services/syncbase/store/util.go
index 72c4beb..19695ed 100644
--- a/services/syncbase/store/util.go
+++ b/services/syncbase/store/util.go
@@ -55,42 +55,7 @@
 	return dst
 }
 
-//////////////////////////////////////////////////////////////
-// Read and Write types used for storing transcation reads
-// and uncommitted writes.
-
-type ScanRange struct {
-	Start, Limit []byte
-}
-
-type ReadSet struct {
-	Keys   [][]byte
-	Ranges []ScanRange
-}
-
-type WriteType int
-
-const (
-	PutOp WriteType = iota
-	DeleteOp
-)
-
-type WriteOp struct {
-	T     WriteType
-	Key   []byte
-	Value []byte
-}
-
-type WriteOpArray []WriteOp
-
-func (a WriteOpArray) Len() int {
-	return len(a)
-}
-
-func (a WriteOpArray) Less(i, j int) bool {
-	return string(a[i].Key) < string(a[j].Key)
-}
-
-func (a WriteOpArray) Swap(i, j int) {
-	a[i], a[j] = a[j], a[i]
+// ConvertError returns a copy of the verror, appending the current stack to it.
+func ConvertError(err error) error {
+	return verror.Convert(verror.IDAction{}, nil, err)
 }