diff --git a/Makefile b/Makefile
index 8451ce00a..a5e9d0b6e 100644
--- a/Makefile
+++ b/Makefile
@@ -228,7 +228,8 @@ build-for-quick:
 docker-image-featurebase-quick: build-for-quick
 	docker build \
 	    --build-arg GO_VERSION=$(GO_VERSION) \
-	    --file Dockerfile-dax-quick ./.quick/
+	    --file Dockerfile-dax-quick \
+	    --tag dax/featurebase ./.quick/
 
 
 docker-image-datagen: vendor
diff --git a/api.go b/api.go
index f254dc568..f440df003 100644
--- a/api.go
+++ b/api.go
@@ -55,9 +55,7 @@ type API struct {
 
 	Serializer Serializer
 
-	writeLogReader     computer.WriteLogReader
-	writeLogWriter     computer.WriteLogWriter
-	snapshotReadWriter computer.SnapshotReadWriter
+	serverlessStorage *storage.ManagerManager
 
 	directiveWorkerPoolSize int
 
@@ -83,6 +81,13 @@ func OptAPIServer(s *Server) apiOption {
 	}
 }
 
+func OptAPIServerlessStorage(mm *storage.ManagerManager) apiOption {
+	return func(a *API) error {
+		a.serverlessStorage = mm
+		return nil
+	}
+}
+
 func OptAPIImportWorkerPoolSize(size int) apiOption {
 	return func(a *API) error {
 		a.importWorkerPoolSize = size
@@ -90,27 +95,6 @@ func OptAPIImportWorkerPoolSize(size int) apiOption {
 	}
 }
 
-func OptAPIWriteLogReader(wlr computer.WriteLogReader) apiOption {
-	return func(a *API) error {
-		a.writeLogReader = wlr
-		return nil
-	}
-}
-
-func OptAPIWriteLogWriter(wlw computer.WriteLogWriter) apiOption {
-	return func(a *API) error {
-		a.writeLogWriter = wlw
-		return nil
-	}
-}
-
-func OptAPISnapshotter(snap computer.SnapshotReadWriter) apiOption {
-	return func(a *API) error {
-		a.snapshotReadWriter = snap
-		return nil
-	}
-}
-
 func OptAPIDirectiveWorkerPoolSize(size int) apiOption {
 	return func(a *API) error {
 		a.directiveWorkerPoolSize = size
@@ -129,9 +113,6 @@ func OptAPIIsComputeNode(is bool) apiOption {
 func NewAPI(opts ...apiOption) (*API, error) {
 	api := &API{
 		importWorkerPoolSize: 2,
-		writeLogReader:       computer.NewNopWriteLogReader(),
-		writeLogWriter:       computer.NewNopWriteLogWriter(),
-		snapshotReadWriter:   computer.NewNopSnapshotReadWriter(),
 
 		directiveWorkerPoolSize: 2,
 	}
@@ -709,20 +690,20 @@ func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string,
 					Views:           req.Views,
 				}
 
-				// Get the current version for shard.
-				version, err := api.getOrCreateShardVersion(ctx, indexName, shard)
-				if err != nil {
-					return errors.Wrap(err, "get or creating shard version")
-				}
-
 				tkey := dax.TableKey(indexName)
 				qtid := tkey.QualifiedTableID()
 				partitionNum := dax.PartitionNum(partition)
 				shardNum := dax.ShardNum(shard)
 
-				api.server.logger.Debugf("importroaring writing to writelogger: %+v, %[1]T len(msg.Views): %d, table: %s", api.writeLogWriter, len(msg.Views), msg.Table)
-				if err := api.writeLogWriter.WriteShard(ctx, qtid, partitionNum, shardNum, version, msg); err != nil {
-					return err
+				b, err := computer.MarshalLogMessage(msg, computer.EncodeTypeJSON)
+				if err != nil {
+					return errors.Wrap(err, "marshalling log message")
+				}
+
+				mgr := api.serverlessStorage.GetShardManager(qtid, partitionNum, shardNum)
+				err = mgr.Append(b)
+				if err != nil {
+					return errors.Wrap(err, "appending shard data") // TODO do we need to set err0 or something?
 				}
 			}
 
@@ -1286,8 +1267,13 @@ func (api *API) DeleteView(ctx context.Context, indexName string, fieldName stri
 	return errors.Wrap(err, "sending DeleteView message")
 }
 
-// IndexShardSnapshot returns a reader that contains the contents of an RBF snapshot for an index/shard.
-func (api *API) IndexShardSnapshot(ctx context.Context, indexName string, shard uint64) (io.ReadCloser, error) {
+// IndexShardSnapshot returns a reader that contains the contents of
+// an RBF snapshot for an index/shard. When snapshotting for
+// serverless, we need to be able to transactionally move the write
+// log to the new version, so we expose writeTx to allow the caller to
+// request a write transaction for the snapshot even though we'll just
+// be reading inside RBF.
+func (api *API) IndexShardSnapshot(ctx context.Context, indexName string, shard uint64, writeTx bool) (io.ReadCloser, error) {
 	span, _ := tracing.StartSpanFromContext(ctx, "API.IndexShardSnapshot")
 	defer span.Finish()
 
@@ -1298,7 +1284,7 @@ func (api *API) IndexShardSnapshot(ctx context.Context, indexName string, shard
 	}
 
 	// Start transaction.
-	tx := index.holder.txf.NewTx(Txo{Index: index, Shard: shard})
+	tx := index.holder.txf.NewTx(Txo{Index: index, Shard: shard, Write: writeTx})
 
 	// Ensure transaction is an RBF transaction.
 	rtx, ok := tx.(*RBFTx)
@@ -1517,20 +1503,20 @@ func (api *API) Import(ctx context.Context, qcx *Qcx, req *ImportRequest, opts .
 	}
 
 	if api.isComputeNode && !options.suppressLog {
-		// Get the current version for shard.
-		version, err := api.getOrCreateShardVersion(ctx, req.Index, req.Shard)
-		if err != nil {
-			return errors.Wrap(err, "get or creating shard version")
-		}
-
 		tkey := dax.TableKey(req.Index)
 		qtid := tkey.QualifiedTableID()
 		partitionNum := dax.PartitionNum(partition)
 		shardNum := dax.ShardNum(req.Shard)
 
-		// Write the request to the write logger.
-		if err := api.writeLogWriter.WriteShard(ctx, qtid, partitionNum, shardNum, version, msg); err != nil {
-			return err
+		b, err := computer.MarshalLogMessage(msg, computer.EncodeTypeJSON)
+		if err != nil {
+			return errors.Wrap(err, "marshalling log message")
+		}
+
+		mgr := api.serverlessStorage.GetShardManager(qtid, partitionNum, shardNum)
+		err = mgr.Append(b)
+		if err != nil {
+			return errors.Wrap(err, "appending shard data") // TODO do we need to set err0 or something?
 		}
 	}
 
@@ -1765,21 +1751,21 @@ func (api *API) ImportRoaringShard(ctx context.Context, indexName string, shard
 				ClearRecords: view.ClearRecords,
 			}
 		}
-		// Get the current version for shard.
-		version, err := api.getOrCreateShardVersion(ctx, indexName, shard)
-		if err != nil {
-			err1 = errors.Wrap(err, "get or creating shard version")
-			return err1
-		}
+
 		tkey := dax.TableKey(indexName)
 		qtid := tkey.QualifiedTableID()
 		partitionNum := dax.PartitionNum(partition)
 		shardNum := dax.ShardNum(shard)
 
-		api.server.logger.Debugf("importroaringshard writing shard to writelogger: %+v, len(msg.Views): %d, table: %s", api.writeLogWriter, len(msg.Views), msg.Table)
+		b, err := computer.MarshalLogMessage(msg, computer.EncodeTypeJSON)
+		if err != nil {
+			err1 = errors.Wrap(err, "marshalling log message")
+			return err1
+		}
 
-		if err := api.writeLogWriter.WriteShard(ctx, qtid, partitionNum, shardNum, version, msg); err != nil {
-			err1 = errors.Wrap(err, "writing import-roaring-shard to writelogger")
+		mgr := api.serverlessStorage.GetShardManager(qtid, partitionNum, shardNum)
+		err1 = errors.Wrap(mgr.Append(b), "appending shard data")
+		if err1 != nil {
 			return err1
 		}
 	}
@@ -1860,20 +1846,21 @@ func (api *API) ImportValue(ctx context.Context, qcx *Qcx, req *ImportValueReque
 
 	if api.isComputeNode && !options.suppressLog {
 		// Get the current version for shard.
-		version, err := api.getOrCreateShardVersion(ctx, req.Index, req.Shard)
-		if err != nil {
-			return errors.Wrap(err, "get or creating shard version")
-		}
-
 		tkey := dax.TableKey(req.Index)
 		qtid := tkey.QualifiedTableID()
 		partitionNum := dax.PartitionNum(partition)
 		shardNum := dax.ShardNum(req.Shard)
-
-		// Write the request to the write logger.
-		if err := api.writeLogWriter.WriteShard(ctx, qtid, partitionNum, shardNum, version, msg); err != nil {
-			return errors.Wrap(err, "writing shard to write logger")
+		b, err := computer.MarshalLogMessage(msg, computer.EncodeTypeJSON)
+		if err != nil {
+			return errors.Wrap(err, "marshalling log message")
 		}
+
+		mgr := api.serverlessStorage.GetShardManager(qtid, partitionNum, shardNum)
+		err = mgr.Append(b)
+		if err != nil {
+			return errors.Wrap(err, "appending shard data") // TODO do we need to set err0 or something?
+		}
+
 	}
 
 	return nil
@@ -3107,47 +3094,33 @@ func (api *API) DirectiveApplied(ctx context.Context) (bool, error) {
 // SnapshotShardData triggers the node to perform a shard snapshot based on the
 // provided SnapshotShardDataRequest.
 func (api *API) SnapshotShardData(ctx context.Context, req *dax.SnapshotShardDataRequest) error {
-	qtid := req.TableKey.QualifiedTableID()
+	// TODO(jaffee) confirm this node is actually responsible for the given
+	// shard? Not sure we need to given that this request comes from
+	// MDS, but might be a belt&suspenders situation.
 
-	// Confirm that this node is currently responsible for table/shard/fromVersion.
-	var version int
-	if v, ok, err := api.holder.versionStore.ShardVersion(ctx, qtid, req.ShardNum); err != nil {
-		return err
-	} else if !ok {
-		return errors.Errorf("shard not managed by this node: %s, %d", req.TableKey, req.ShardNum)
-	} else if v != req.FromVersion {
-		return errors.Errorf("shard managed by this node is at version: %d, not: %d", v, req.FromVersion)
-	} else {
-		version = v
-	}
+	qtid := req.TableKey.QualifiedTableID()
 
 	partition := disco.ShardToShardPartition(string(req.TableKey), uint64(req.ShardNum), disco.DefaultPartitionN)
 	partitionNum := dax.PartitionNum(partition)
 
-	// Create the snapshot for the current version.
-	rc, err := api.IndexShardSnapshot(ctx, string(req.TableKey), uint64(req.ShardNum))
+	// Create the snapshot for the current version.  How do we ensure
+	// here that any new writes go to the new write log since we are
+	// now reading a version of the shard which exists only at this
+	// exact point in time? Ans: we'll cut over to the new storage
+	// manager and call IncrementWriteLogVersion while a write Tx is
+	// held on RBF.
+	rc, err := api.IndexShardSnapshot(ctx, string(req.TableKey), uint64(req.ShardNum), true)
 	if err != nil {
 		return errors.Wrap(err, "getting index/shard readcloser")
 	}
 
-	// The following closes rc, the ReadCloser.
-	if err := api.snapshotReadWriter.WriteShardData(ctx, qtid, partitionNum, req.ShardNum, version, rc); err != nil {
-		return errors.Wrap(err, "snapshotting shard data")
+	mgr := api.serverlessStorage.GetShardManager(qtid, partitionNum, req.ShardNum)
+	if err := mgr.IncrementWLVersion(); err != nil {
+		return errors.Wrap(err, "incrementing write log version")
 	}
-
-	// Increment the version of the shard managed by this node.
-	if err := api.holder.versionStore.AddShards(ctx, qtid,
-		dax.NewVersionedShard(req.ShardNum, req.ToVersion),
-	); err != nil {
-		return errors.Wrap(err, "incrementing shard version locally")
-	}
-
-	// Update the cached directive on the holder.
-	api.holder.SetDirective(&req.Directive)
-	api.holder.SetDirectiveApplied(true)
-
-	// Finally, delete the log file for the previous version.
-	return api.writeLogWriter.DeleteShard(ctx, qtid, partitionNum, req.ShardNum, req.FromVersion)
+	// TODO(jaffee) look into downgrading Tx on RBF to read lock here now that WL version is incremented.
+	err = mgr.Snapshot(rc)
+	return errors.Wrap(err, "snapshotting shard data")
 }
 
 // SnapshotTableKeys triggers the node to perform a table keys snapshot based on
@@ -3162,41 +3135,21 @@ func (api *API) SnapshotTableKeys(ctx context.Context, req *dax.SnapshotTableKey
 
 	qtid := req.TableKey.QualifiedTableID()
 
-	// Confirm that this node is currently responsible for table/partition/fromVersion.
-	var version int
-	if v, ok, err := api.holder.versionStore.PartitionVersion(ctx, qtid, req.PartitionNum); err != nil {
-		return err
-	} else if !ok {
-		return errors.Errorf("partition not managed by this node: %s, %d", req.TableKey, req.PartitionNum)
-	} else if v != req.FromVersion {
-		return errors.Errorf("partition managed by this node is at version: %d, not: %d", v, req.FromVersion)
-	} else {
-		version = v
-	}
-
 	// Create the snapshot for the current version.
 	wrTo, err := api.TranslateData(ctx, string(req.TableKey), int(req.PartitionNum))
 	if err != nil {
 		return errors.Wrapf(err, "getting index/partition writeto: %s/%d", req.TableKey, req.PartitionNum)
 	}
 
-	if err := api.snapshotReadWriter.WriteTableKeys(ctx, qtid, req.PartitionNum, version, wrTo); err != nil {
-		return errors.Wrap(err, "snapshotting table keys")
+	// TODO(jaffee) need to ensure writes to translation data can't
+	// occur while this is happening.
+	mgr := api.serverlessStorage.GetTableKeyManager(qtid, req.PartitionNum)
+	if err := mgr.IncrementWLVersion(); err != nil {
+		return errors.Wrap(err, "incrementing write log version")
 	}
-
-	// Increment the version of the partition managed by this node.
-	if err := api.holder.versionStore.AddPartitions(ctx, qtid,
-		dax.NewVersionedPartition(req.PartitionNum, req.ToVersion),
-	); err != nil {
-		return errors.Wrap(err, "incrementing partition version locally")
-	}
-
-	// Update the cached directive on the holder.
-	api.holder.SetDirective(&req.Directive)
-	api.holder.SetDirectiveApplied(true)
-
-	// Finally, delete the log file for the previous version.
-	return api.writeLogWriter.DeleteTableKeys(ctx, qtid, req.PartitionNum, req.FromVersion)
+	// TODO(jaffee) downgrade (currently non-existent) lock to read-only
+	err = mgr.SnapshotTo(wrTo)
+	return errors.Wrap(err, "snapshotting table keys")
 }
 
 // SnapshotFieldKeys triggers the node to perform a field keys snapshot based on
@@ -3204,41 +3157,20 @@ func (api *API) SnapshotTableKeys(ctx context.Context, req *dax.SnapshotTableKey
 func (api *API) SnapshotFieldKeys(ctx context.Context, req *dax.SnapshotFieldKeysRequest) error {
 	qtid := req.TableKey.QualifiedTableID()
 
-	// Confirm that this node is currently responsible for table/field/fromVersion.
-	var version int
-	if v, ok, err := api.holder.versionStore.FieldVersion(ctx, qtid, req.Field); err != nil {
-		return err
-	} else if !ok {
-		return errors.Errorf("field not managed by this node: %s, %s", req.TableKey, req.Field)
-	} else if v != req.FromVersion {
-		return errors.Errorf("field managed by this node is at version: %d, not: %d", v, req.FromVersion)
-	} else {
-		version = v
-	}
-
 	// Create the snapshot for the current version.
+	// TODO(jaffee) change this to get write lock
 	wrTo, err := api.FieldTranslateData(ctx, string(req.TableKey), string(req.Field))
 	if err != nil {
 		return errors.Wrap(err, "getting index/field writeto")
 	}
 
-	if err := api.snapshotReadWriter.WriteFieldKeys(ctx, qtid, req.Field, version, wrTo); err != nil {
-		return errors.Wrap(err, "snapshotting field keys")
+	mgr := api.serverlessStorage.GetFieldKeyManager(qtid, req.Field)
+	if err := mgr.IncrementWLVersion(); err != nil {
+		return errors.Wrap(err, "incrementing writelog version")
 	}
-
-	// Increment the version of the field managed by this node.
-	if err := api.holder.versionStore.AddFields(ctx, qtid,
-		dax.NewVersionedField(req.Field, req.ToVersion),
-	); err != nil {
-		return errors.Wrap(err, "incrementing field version locally")
-	}
-
-	// Update the cached directive on the holder.
-	api.holder.SetDirective(&req.Directive)
-	api.holder.SetDirectiveApplied(true)
-
-	// Finally, delete the log file for the previous version.
-	return api.writeLogWriter.DeleteFieldKeys(ctx, qtid, req.Field, req.FromVersion)
+	// TODO(jaffee) downgrade to read lock
+	err = mgr.SnapshotTo(wrTo)
+	return errors.Wrap(err, "snapshotTo in FieldKeys")
 }
 
 type serverInfo struct {
diff --git a/api_directive.go b/api_directive.go
index d7650c0b0..b4948e562 100644
--- a/api_directive.go
+++ b/api_directive.go
@@ -348,29 +348,26 @@ func (api *API) pushJobsTableKeys(ctx context.Context, jobs chan<- directiveJobT
 	}
 }
 
-func (api *API) loadTableKeys(ctx context.Context, idx *Index, tkey dax.TableKey, partition dax.VersionedPartition) error {
+func (api *API) loadTableKeys(ctx context.Context, idx *Index, tkey dax.TableKey, partition dax.PartitionNum) error {
 	qtid := tkey.QualifiedTableID()
 
-	// Load the previous snapshot. Version 0 doesn't have a snapshot
-	// file; it only has log entries.
-	if partition.Version > 0 {
-		// Load partition snapshot: version - 1
-		previousVersion := partition.Version - 1
-		rc, err := api.snapshotReadWriter.ReadTableKeys(ctx, qtid, partition.Num, previousVersion)
-		if err != nil {
-			return errors.Wrap(err, "reading table keys snapshot")
-		}
-		defer rc.Close()
+	mgr := api.serverlessStorage.GetTableKeyManager(qtid, partition)
+	rc, err := mgr.LoadLatestSnapshot()
+	if err != nil {
+		return errors.Wrap(err, "loading table key snapshot")
+	}
+	defer rc.Close()
+	if err := api.TranslateIndexDB(ctx, string(tkey), int(partition), rc); err != nil {
+		return errors.Wrap(err, "restoring table keys")
 
-		if err := api.TranslateIndexDB(ctx, string(tkey), int(partition.Num), rc); err != nil {
-			return errors.Wrap(err, "restoring table keys")
-
-		}
 	}
 
 	if err := func() error {
-		store := idx.TranslateStore(int(partition.Num))
-
+		store := idx.TranslateStore(int(partition))
+		reader, err := mgr.LoadWriteLog()
+		if err != nil {
+			return errors.Wrap(err, "loading write log table keys")
+		}
 		reader := api.writeLogReader.TableKeyReader(ctx, qtid, partition.Num, partition.Version)
 		if err := reader.Open(); err != nil {
 			// TODO: this log can be confusing because on a create
@@ -506,7 +503,7 @@ func (api *API) pushJobsShards(ctx context.Context, jobs chan<- directiveJobType
 	}
 }
 
-func (api *API) loadShard(ctx context.Context, tkey dax.TableKey, shard dax.VersionedShard) error {
+func (api *API) loadShard(ctx context.Context, tkey dax.TableKey, shard dax.ShardNum) error {
 	qtid := tkey.QualifiedTableID()
 
 	partition := disco.ShardToShardPartition(string(tkey), uint64(shard.Num), disco.DefaultPartitionN)
diff --git a/cluster.go b/cluster.go
index 86bd1c9cd..3ebd98013 100644
--- a/cluster.go
+++ b/cluster.go
@@ -4,6 +4,7 @@ package pilosa
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"log"
 	"sync"
@@ -72,8 +73,8 @@ type cluster struct { // nolint: maligned
 
 	partitionAssigner string
 
-	writeLogWriter computer.WriteLogWriter
-	versionStore   dax.VersionStore
+	serverlessStorage *storage.ManagerManager
+	versionStore      dax.VersionStore
 
 	// isComputeNode is set to true if this node is running as a DAX compute
 	// node.
@@ -100,8 +101,6 @@ func newCluster() *cluster {
 
 		disCo: disco.NopDisCo,
 		noder: disco.NewEmptyLocalNoder(),
-
-		writeLogWriter: computer.NewNopWriteLogWriter(),
 	}
 }
 
@@ -318,6 +317,44 @@ func (c *cluster) findFieldKeys(ctx context.Context, field *Field, keys ...strin
 	return translations, nil
 }
 
+func (c *cluster) appendFieldKeysWriteLog(ctx context.Context, qtid dax.QualifiedTableID, fieldName dax.FieldName, translations map[string]uint64) error {
+	// TODO move marshaling somewhere more centralized and less... explicitly json-y
+	msg := computer.FieldKeyMap{
+		TableKey:   qtid.Key(),
+		Field:      fieldName,
+		StringToID: translations,
+	}
+
+	b, err := json.Marshal(msg)
+	if err != nil {
+		return errors.Wrap(err, "marshalling field key map to json")
+	}
+	mgr := c.serverlessStorage.GetFieldKeyManager(qtid, fieldName)
+	err = mgr.Append(b)
+	if err != nil {
+		return errors.Wrap(err, "appending field keys")
+	}
+	return nil
+
+}
+
+func (c *cluster) appendTableKeysWriteLog(ctx context.Context, qtid dax.QualifiedTableID, partition dax.PartitionNum, translations map[string]uint64) error {
+	msg := computer.PartitionKeyMap{
+		TableKey:   qtid.Key(),
+		Partition:  partition,
+		StringToID: translations,
+	}
+
+	b, err := json.Marshal(msg)
+	if err != nil {
+		return errors.Wrap(err, "marshalling partition key map to json")
+	}
+
+	mgr := c.serverlessStorage.GetTableKeyManager(qtid, partition)
+	return errors.Wrap(mgr.Append(b), "appending table keys")
+
+}
+
 func (c *cluster) createFieldKeys(ctx context.Context, field *Field, keys ...string) (map[string]uint64, error) {
 	if idx := field.ForeignIndex(); idx != "" {
 		// The field uses foreign index keys.
@@ -350,17 +387,9 @@ func (c *cluster) createFieldKeys(ctx context.Context, field *Field, keys ...str
 		tkey := dax.TableKey(field.Index())
 		qtid := tkey.QualifiedTableID()
 		fieldName := dax.FieldName(field.Name())
-
-		// Get the current version for field.
-		version, found, err := c.versionStore.FieldVersion(ctx, qtid, fieldName)
+		err = c.appendFieldKeysWriteLog(ctx, qtid, fieldName, translations)
 		if err != nil {
-			return nil, errors.Wrap(err, "getting field version")
-		} else if !found {
-			return nil, errors.Errorf("no version found for table(%s) field(%s)", qtid, fieldName)
-		}
-
-		if err := c.writeLogWriter.CreateFieldKeys(ctx, qtid, fieldName, version, translations); err != nil {
-			return nil, errors.Errorf("logging field(%s/%s) keys(%v)", field.Index(), field.Name(), keys)
+			return nil, errors.Wrap(err, "appending to write log")
 		}
 
 		return translations, nil
@@ -754,16 +783,7 @@ func (c *cluster) createIndexKeys(ctx context.Context, indexName string, keys ..
 			tkey := dax.TableKey(idx.Name())
 			qtid := tkey.QualifiedTableID()
 			partitionNum := dax.PartitionNum(partitionID)
-
-			// Get the current version for partition.
-			version, found, err := c.versionStore.PartitionVersion(ctx, qtid, partitionNum)
-			if err != nil {
-				return errors.Wrap(err, "getting partition version")
-			} else if !found {
-				return errors.Errorf("no version found for table(%s) partition(%d)", qtid, partitionNum)
-			}
-
-			return c.writeLogWriter.CreateTableKeys(ctx, qtid, partitionNum, version, translations)
+			return c.appendTableKeysWriteLog(ctx, qtid, partitionNum, translations)
 		})
 	}
 
diff --git a/dax/Makefile b/dax/Makefile
index ca0cfb899..40c666b31 100644
--- a/dax/Makefile
+++ b/dax/Makefile
@@ -78,6 +78,9 @@ dc-logs-%:
 dc-prereqs:
 	mkdir -p ../.quick
 
+dc-cli:
+	featurebase cli --host localhost --port 8080 --org-id=testorg --db-id=testdb
+
 # This is just an example. For it to work, you'll first need to:
 #   featurebase cli --host localhost --port 8080 --org-id=testorg --db-id=testdb
 #   create table keysidstbl2 (_id string, slice idset);
diff --git a/dax/computer/interfaces.go b/dax/computer/interfaces.go
index f04e926cc..b6a423026 100644
--- a/dax/computer/interfaces.go
+++ b/dax/computer/interfaces.go
@@ -18,8 +18,12 @@ type Registrar interface {
 // These are typically implemented by the WriteLogger client.
 type WriteLogService interface {
 	AppendMessage(bucket string, key string, version int, msg []byte) error
-	LogReader(bucket string, key string, version int) (io.Reader, io.Closer, error)
+	LogReader(bucket string, key string, version int) (io.ReadCloser, error)
+	LogReaderFrom(bucket string, key string, version int, offset int) (io.ReadCloser, error)
 	DeleteLog(bucket string, key string, version int) error
+	List(bucket, key string) ([]WriteLogInfo, error)
+	Lock(bucket, key string) error
+	Unlock(bucket, key string) error
 }
 
 // SnapshotService represents the SnapshotService methods which Computer uses.
@@ -28,8 +32,89 @@ type SnapshotService interface {
 	Read(bucket string, key string, version int) (io.ReadCloser, error)
 	Write(bucket string, key string, version int, rc io.ReadCloser) error
 	WriteTo(bucket string, key string, version int, wrTo io.WriterTo) error
+	List(bucket, key string) ([]SnapInfo, error)
 }
 
+// ServerlessStorage is the interface to a particular shard or
+// translation store that contains both writelogger and
+// snapshotter. The interface is designed such that implementations
+// are expected to be stateful.
+//
+// One must not call LoadWriteLog until after calling
+// LoadLatestSnapshot. One must not call Append, IncrementWLVersion,
+// or Snapshot until after successfully calling Lock.
+type ServerlessStorage interface {
+	// LoadLatestSnapshot loads the latest available snapshot in the snapshot store.
+	LoadLatestSnapshot() (data io.ReadCloser, err error)
+
+	// // Potential future methods to support getting older versions. SnapInfo would have timestamp information as well.
+	//
+	// ListSnapshots() []SnapInfo
+	// LoadSnapshot(version int) (data io.ReadCloser, err error)
+
+	// LoadWriteLog can be called after LoadLatestSnapshot. It loads
+	// any writelog data which has been written since the latest
+	// snapshot. Subsequent calls to LoadWriteLog will only return new
+	// data that hasn't previously been returned from LoadWriteLog.
+	LoadWriteLog() (data io.ReadCloser, err error)
+
+	// Lock acquires an advisory lock for this resource which grants
+	// us exclusive access to write to it.  The normal pattern is to
+	// call:
+	//
+	// 1. LoadLatestSnapshot
+	// 2. LoadWriteLog
+	// 3. Lock
+	// 4. LoadWriteLog
+	//
+	// The second call to LoadWriteLog is necessary in case any writes
+	// occurred between the last load and acquiring the lock. Once the
+	// lock is acquired it should not be possible for any more writes
+	// to occur. Lock will error if (a) we fail to acquire the lock or
+	// (b) the state of the snapshot store for this resource is not
+	// identical to what is was before the lock was acquired. Case (b)
+	// means that quite a lot has happened in between LoadWriteLog and
+	// Lock, and we should probably just die and start over.
+	Lock() error
+
+	// Append appends the msg to the write log. It will fail if we
+	// haven't properly loaded and gotten a lock for the resource
+	// we're writing to.
+	Append(msg []byte) error
+
+	// IncrementWLVersion should be called during snapshotting with a
+	// write Tx held on the local resource. This ensures that any
+	// writes which completed prior to the snapshot are in the prior
+	// WL and any that complete after the snapshot are in the
+	// incremented WL.
+	IncrementWLVersion() error
+
+	// Snapshot takes a ReadCloser which has the contents of the
+	// resource being tracked at a particular point in time and writes
+	// them to the Snapshot Store. Upon a successful write it will
+	// truncate any write logs which are now incorporated into the
+	// snapshot.
+	Snapshot(rc io.ReadCloser) error
+	SnapshotTo(wt io.WriterTo) error
+
+	// Unlock releases the lock. This should be called if control of
+	// the underlying resource is being transitioned to another
+	// node. Ideally it's also called if the process crashes (e.g. via
+	// a defer), but an implementation based on filesystem locks
+	// should have those removed by the operating system when the
+	// process exits anyway.
+	Unlock() error
+}
+
+// SnapInfo holds metadata about a snapshot.
+type SnapInfo struct {
+	Version int
+	// Date    time.Time
+}
+
+// WriteLogInfo holds metadata about a write log.
+type WriteLogInfo SnapInfo
+
 // SnapshotReadWriter provides the interface for all snapshot read and writes in
 // FeatureBase.
 type SnapshotReadWriter interface {
diff --git a/dax/computer/writelog.go b/dax/computer/writelog.go
index df6999869..9c73d33e7 100644
--- a/dax/computer/writelog.go
+++ b/dax/computer/writelog.go
@@ -132,13 +132,13 @@ func newTableKeyReader(wl WriteLogService, qtid dax.QualifiedTableID, partition
 func (r *tableKeyReader) Open() error {
 	bucket := partitionBucket(r.table, r.partition)
 
-	reader, closer, err := r.wl.LogReader(bucket, keysFileName, r.version)
+	readcloser, err := r.wl.LogReader(bucket, keysFileName, r.version)
 	if err != nil {
 		return errors.Wrapf(err, "getting log reader: %s, %s, %d", bucket, keysFileName, r.version)
 	}
 
-	r.closer = closer
-	r.scanner = bufio.NewScanner(reader)
+	r.closer = readcloser
+	r.scanner = bufio.NewScanner(readcloser)
 
 	return nil
 }
@@ -201,13 +201,13 @@ func newFieldKeyReader(wl WriteLogService, qtid dax.QualifiedTableID, field dax.
 func (r *fieldKeyReader) Open() error {
 	bucket := fieldBucket(r.table, r.field)
 
-	reader, closer, err := r.wl.LogReader(bucket, keysFileName, r.version)
+	readcloser, err := r.wl.LogReader(bucket, keysFileName, r.version)
 	if err != nil {
 		return errors.Wrapf(err, "getting log reader: %s, %s, %d", bucket, keysFileName, r.version)
 	}
 
-	r.closer = closer
-	r.scanner = bufio.NewScanner(reader)
+	r.closer = readcloser
+	r.scanner = bufio.NewScanner(readcloser)
 
 	return nil
 }
@@ -273,13 +273,13 @@ func (r *shardReader) Open() error {
 	bucket := partitionBucket(r.table, r.partition)
 	shardKey := shardKey(r.shard)
 
-	reader, closer, err := r.wl.LogReader(bucket, shardKey, r.version)
+	readcloser, err := r.wl.LogReader(bucket, shardKey, r.version)
 	if err != nil {
 		return errors.Wrapf(err, "getting log reader: %s, %s, %d", bucket, shardKey, r.version)
 	}
 
-	r.closer = closer
-	r.scanner = bufio.NewScanner(reader)
+	r.closer = readcloser
+	r.scanner = bufio.NewScanner(readcloser)
 
 	return nil
 }
diff --git a/dax/mds/client/client.go b/dax/mds/client/client.go
index c4dafb7ea..9cc5b966a 100644
--- a/dax/mds/client/client.go
+++ b/dax/mds/client/client.go
@@ -451,7 +451,7 @@ func (c *Client) RegisterNode(ctx context.Context, node *dax.Node) error {
 
 	if resp.StatusCode != http.StatusOK {
 		b, _ := io.ReadAll(resp.Body)
-		return errors.Errorf("status code: %d: %s", resp.StatusCode, b)
+		return errors.Errorf("registration request to %s status code: %d: %s", url, resp.StatusCode, b)
 	}
 
 	return nil
diff --git a/dax/mds/controller/controller.go b/dax/mds/controller/controller.go
index aaae66944..f6989a5d0 100644
--- a/dax/mds/controller/controller.go
+++ b/dax/mds/controller/controller.go
@@ -1401,17 +1401,6 @@ func (c *Controller) SnapshotTable(ctx context.Context, qtid dax.QualifiedTableI
 // snapshot that shard, then increment its shard version for logs written to the
 // WriteLogger.
 func (c *Controller) SnapshotShardData(ctx context.Context, qtid dax.QualifiedTableID, shardNum dax.ShardNum) error {
-	// Confirm table/shard is being tracked; get the current shard.
-	fromShardVersion, ok, err := c.versionStore.ShardVersion(ctx, qtid, shardNum)
-	if err != nil {
-		return errors.Wrapf(err, "getting shard version: %s, %d", qtid, shardNum)
-	} else if !ok {
-		return NewErrInternal(
-			fmt.Sprintf("shard to snapshot not found: %s, %d", qtid, shardNum),
-		)
-	}
-	toShardVersion := fromShardVersion + 1
-
 	// Get the node responsible for the shard.
 	bal := c.ComputeBalancer
 
@@ -1428,54 +1417,17 @@ func (c *Controller) SnapshotShardData(ctx context.Context, qtid dax.QualifiedTa
 
 	addr := dax.Address(workers[0].ID)
 
-	// Make a copy of the controller's versionStore, and update the current
-	// shard so that the directive sent along with the SnapshotRequest reflects
-	// the state that we want after a successful snapshot.
-	versionStoreCopy, err := c.versionStore.Copy(ctx)
-	if err != nil {
-		return errors.Wrap(err, "copying version store")
-	}
-	if err := versionStoreCopy.AddShards(ctx, qtid,
-		dax.NewVersionedShard(shardNum, toShardVersion),
-	); err != nil {
-		return NewErrInternal(err.Error())
-	}
-
-	// Convert the address into a slice of addressMethod containing the
-	// appropriate method.
-	addressMethods := applyAddressMethod([]dax.Address{addr}, dax.DirectiveMethodSnapshot)
-
-	var toDirective dax.Directive
-	if directives, err := c.buildDirectives(ctx, addressMethods, versionStoreCopy); err != nil {
-		return NewErrInternal(err.Error())
-	} else if ld := len(directives); ld != 1 {
-		msg := fmt.Sprintf("buildDirectives returned invalid number of directives: %d", ld)
-		return NewErrInternal(msg)
-	} else {
-		toDirective = *directives[0]
-	}
-
 	// Send the node a snapshot request.
 	req := &dax.SnapshotShardDataRequest{
-		Address:     addr,
-		TableKey:    qtid.Key(),
-		ShardNum:    shardNum,
-		FromVersion: fromShardVersion,
-		ToVersion:   toShardVersion,
-		Directive:   toDirective,
+		Address:  addr,
+		TableKey: qtid.Key(),
+		ShardNum: shardNum,
 	}
 
 	if err := c.Director.SendSnapshotShardDataRequest(ctx, req); err != nil {
 		return NewErrInternal(err.Error())
 	}
 
-	// A successful request means the shard version can be incremented.
-	if err := c.versionStore.AddShards(ctx, qtid,
-		dax.NewVersionedShard(shardNum, toShardVersion),
-	); err != nil {
-		return NewErrInternal(err.Error())
-	}
-
 	return nil
 }
 
diff --git a/dax/snapshot.go b/dax/snapshot.go
index dd2d06a9c..5b1e04ef4 100644
--- a/dax/snapshot.go
+++ b/dax/snapshot.go
@@ -3,12 +3,8 @@ package dax
 type SnapshotShardDataRequest struct {
 	Address Address `json:"address"`
 
-	TableKey    TableKey `json:"table-key"`
-	ShardNum    ShardNum `json:"shard"`
-	FromVersion int      `json:"from-version"`
-	ToVersion   int      `json:"to-version"`
-
-	Directive Directive `json:"directive"`
+	TableKey TableKey `json:"table-key"`
+	ShardNum ShardNum `json:"shard"`
 }
 
 type SnapshotTableKeysRequest struct {
diff --git a/dax/snapshotter/snapshotter.go b/dax/snapshotter/snapshotter.go
index 7f46fc012..7fdd0b799 100644
--- a/dax/snapshotter/snapshotter.go
+++ b/dax/snapshotter/snapshotter.go
@@ -8,12 +8,17 @@ import (
 	"io/fs"
 	"os"
 	"path"
+	"strconv"
 	"sync"
+	"syscall"
 
 	"github.com/featurebasedb/featurebase/v3/errors"
 	"github.com/featurebasedb/featurebase/v3/logger"
 )
 
+// bucket = table + partition or table + field
+// key = "shard/num" or "keys"
+
 type Snapshotter struct {
 	mu sync.RWMutex
 
@@ -50,6 +55,29 @@ func (s *Snapshotter) Write(bucket string, key string, version int, rc io.ReadCl
 	return snapshotFile.Sync()
 }
 
+func (s *Snapshotter) List(bucket, key string) ([]computer.SnapInfo, error) {
+	dirpath := path.Join(s.dataDir, bucket, key)
+
+	entries, err := os.ReadDir(dirpath)
+	if err != nil {
+		if pe, ok := err.(*os.PathError); ok && pe.Err == syscall.ENOENT {
+			return nil, nil
+		}
+		return nil, errors.Wrap(err, "reading directory")
+	}
+	snaps := make([]computer.SnapInfo, len(entries))
+	for i, entry := range entries {
+		version, err := strconv.ParseInt(entry.Name(), 10, 64)
+		if err != nil {
+			return nil, errors.Wrapf(err, "filename '%s' could not be parsed to version number", entry.Name())
+		}
+		snaps[i] = computer.SnapInfo{
+			Version: int(version),
+		}
+	}
+	return snaps, nil
+}
+
 func (s *Snapshotter) Read(bucket string, key string, version int) (io.ReadCloser, error) {
 	_, filePath := s.paths(fullKey(bucket, key, version))
 	f, err := os.Open(filePath)
diff --git a/dax/storage/storage.go b/dax/storage/storage.go
new file mode 100644
index 000000000..b2e1df826
--- /dev/null
+++ b/dax/storage/storage.go
@@ -0,0 +1,320 @@
+package storage
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"path"
+	"sync"
+
+	"github.com/molecula/featurebase/v3/dax"
+	"github.com/molecula/featurebase/v3/dax/computer"
+	"github.com/molecula/featurebase/v3/errors"
+	"github.com/molecula/featurebase/v3/logger"
+)
+
+// ManagerManager holds all the various Managers each of which is
+// specific to a particular shard, table key partition or field, but
+// all of which use the same underlying snapshotter and writelogger.
+type ManagerManager struct {
+	Snapshotter computer.SnapshotService
+	WriteLogger computer.WriteLogService
+	Logger      logger.Logger
+
+	mu               sync.Mutex
+	shardManagers    map[shardK]*Manager
+	tableKeyManagers map[tableKeyK]*Manager
+	fieldKeyManagers map[fieldKeyK]*Manager
+}
+
+func NewManagerManager(s computer.SnapshotService, w computer.WriteLogService, l logger.Logger) *ManagerManager {
+	return &ManagerManager{
+		Snapshotter: s,
+		WriteLogger: w,
+		Logger:      l,
+
+		shardManagers:    make(map[shardK]*Manager),
+		tableKeyManagers: make(map[tableKeyK]*Manager),
+		fieldKeyManagers: make(map[fieldKeyK]*Manager),
+	}
+}
+
+// compound map keys
+
+type shardK struct {
+	qtid      dax.QualifiedTableID
+	partition dax.PartitionNum
+	shard     dax.ShardNum
+}
+
+type tableKeyK struct {
+	qtid      dax.QualifiedTableID
+	partition dax.PartitionNum
+}
+
+type fieldKeyK struct {
+	qtid  dax.QualifiedTableID
+	field dax.FieldName
+}
+
+func (mm *ManagerManager) GetShardManager(qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum) *Manager {
+	mm.mu.Lock()
+	defer mm.mu.Unlock()
+	key := shardK{qtid: qtid, partition: partition, shard: shard}
+	if m, ok := mm.shardManagers[key]; ok {
+		return m
+	}
+	mm.shardManagers[key] = (&Manager{
+		snapshotter: mm.Snapshotter,
+		writeLogger: mm.WriteLogger,
+		bucket:      partitionBucket(qtid.Key(), partition),
+		key:         shardKey(shard),
+		log:         mm.Logger,
+	}).initialize()
+
+	return mm.shardManagers[key]
+}
+
+func (mm *ManagerManager) GetTableKeyManager(qtid dax.QualifiedTableID, partition dax.PartitionNum) *Manager {
+	mm.mu.Lock()
+	defer mm.mu.Unlock()
+	key := tableKeyK{qtid: qtid, partition: partition}
+	if m, ok := mm.tableKeyManagers[key]; ok {
+		return m
+	}
+	mm.tableKeyManagers[key] = (&Manager{
+		snapshotter: mm.Snapshotter,
+		writeLogger: mm.WriteLogger,
+		bucket:      partitionBucket(qtid.Key(), partition),
+		key:         keysFileName,
+		log:         mm.Logger,
+	}).initialize()
+	return mm.tableKeyManagers[key]
+}
+
+func (mm *ManagerManager) GetFieldKeyManager(qtid dax.QualifiedTableID, field dax.FieldName) *Manager {
+	mm.mu.Lock()
+	defer mm.mu.Unlock()
+	key := fieldKeyK{qtid: qtid, field: field}
+	if m, ok := mm.fieldKeyManagers[key]; ok {
+		return m
+	}
+	mm.fieldKeyManagers[key] = (&Manager{
+		snapshotter: mm.Snapshotter,
+		writeLogger: mm.WriteLogger,
+		bucket:      fieldBucket(qtid.Key(), field),
+		key:         keysFileName,
+		log:         mm.Logger,
+	}).initialize()
+	return mm.fieldKeyManagers[key]
+}
+
+// Manager wraps the snapshotter and writelogger to implement the
+// ServerlessStorage interface and maintain messy state between
+// calls. Manager is *not* threadsafe, care should be taken that
+// concurrent calls are not made to Manager methods. The exception
+// being that Snapshot and Append are safe to call concurrently.
+type Manager struct {
+	snapshotter computer.SnapshotService
+	writeLogger computer.WriteLogService
+	bucket      string
+	key         string
+
+	log logger.Logger
+
+	loadWLsPastVersion int
+	latestWLVersion    int
+	lastWLPos          int
+
+	locked bool
+}
+
+func (m *Manager) initialize() *Manager {
+	m.loadWLsPastVersion = -2
+	m.latestWLVersion = -1
+	m.lastWLPos = -1
+	return m
+}
+
+func (m *Manager) LoadLatestSnapshot() (data io.ReadCloser, err error) {
+	snaps, err := m.snapshotter.List(m.bucket, m.key)
+	if err != nil {
+		return nil, errors.Wrap(err, "listing snapshots")
+	}
+	m.lastWLPos = 0
+
+	if len(snaps) == 0 {
+		m.loadWLsPastVersion = -1
+		return io.NopCloser(bytes.NewReader([]byte{})), nil
+	}
+	// assuming snapshots come back in sorted order
+	latest := snaps[len(snaps)-1]
+	m.loadWLsPastVersion = latest.Version
+
+	// TODO(jaffee): whatever is using the snapshot may discover that
+	// it is corrupted/incomplete. We don't want to separately check
+	// the checksum in here because then we'd have to read the whole
+	// snapshot twice. Need a way to catch the checksum error and tell
+	// Manager to mark that version as bad and remove it, then try
+	// LoadLatestSnapshot again.
+	return m.snapshotter.Read(m.bucket, m.key, latest.Version)
+}
+
+// // Potential future methods to support getting older versions. SnapInfo would have timestamp information as well.
+//
+// ListSnapshots() []SnapInfo
+// LoadSnapshot(version int) (data io.ReadCloser, err error)
+
+// LoadWriteLog can be called after LoadLatestSnapshot. It loads
+// any writelog data which has been written since the latest
+// snapshot. Subsequent calls to LoadWriteLog will only return new
+// data that hasn't previously been returned from LoadWriteLog.
+func (m *Manager) LoadWriteLog() (data io.ReadCloser, err error) {
+	if m.loadWLsPastVersion == -2 {
+		return nil, errors.New(errors.ErrUncoded, "LoadWriteLog called in inconsistent state, can't tell what version to load from")
+	}
+	wLogs, err := m.writeLogger.List(m.bucket, m.key)
+
+	versions := make([]int, 0, len(wLogs))
+	for _, log := range wLogs {
+		if log.Version > m.loadWLsPastVersion {
+			versions = append(versions, log.Version)
+		}
+	}
+
+	if len(versions) > 1 {
+		// TODO(jaffee) This can happen if there's a failure writing a
+		// snapshot. Need to implement a MultiReadCloser or similar
+		// that wraps all the latest write logs into one ReadCloser.
+		// It should only wrap the last one in a trackingReader.
+		return nil, errors.New(dax.ErrUnimplemented, "UNIMPLEMENTED: multiple write log versions ahead of latest snapshot.")
+	}
+
+	if len(versions) == 0 {
+		m.log.Debugf("LoadWriteLog: no logs after snapshot: %d on %s", m.loadWLsPastVersion, path.Join(m.bucket, m.key))
+		m.latestWLVersion = m.loadWLsPastVersion + 1
+		return io.NopCloser(bytes.NewReader([]byte{})), nil
+	}
+
+	if m.locked && m.latestWLVersion != versions[0] {
+		return nil, errors.New(errors.ErrUncoded, "write log version gone since locking")
+	}
+	m.latestWLVersion = versions[0]
+
+	r, err := m.writeLogger.LogReaderFrom(m.bucket, m.key, versions[0], m.lastWLPos)
+	if err != nil {
+		return nil, errors.Wrap(err, "getting writelog")
+	}
+	return &trackingReader{
+		r: r,
+		update: func(n int, err error) {
+			m.lastWLPos += n
+		},
+	}, nil
+}
+
+// Lock acquires an advisory lock for this resource which grants
+// us exclusive access to write to it.  The normal pattern is to
+// call:
+//
+// 1. LoadLatestSnapshot
+// 2. LoadWriteLog
+// 3. Lock
+// 4. LoadWriteLog
+//
+// The second call to LoadWriteLog is necessary in case any writes
+// occurred between the last load and acquiring the lock. Once the
+// lock is acquired it should not be possible for any more writes
+// to occur. Lock will error if (a) we fail to acquire the lock or
+// (b) the state of the snapshot store for this resource is not
+// identical to what is was before the lock was acquired. Case (b)
+// means that quite a lot has happened in between LoadWriteLog and
+// Lock, and we should probably just die and start over.
+func (m *Manager) Lock() error {
+	// lock is sort of arbitrarily on the write log interface
+	if err := m.writeLogger.Lock(m.bucket, m.key); err != nil {
+		return errors.Wrap(err, "acquiring lock")
+	}
+	m.locked = true
+	return nil
+}
+
+// Append appends the msg to the write log. It will fail if we
+// haven't properly loaded and gotten a lock for the resource
+// we're writing to.
+func (m *Manager) Append(msg []byte) error {
+	if m.latestWLVersion < 0 {
+		return errors.New(errors.ErrUncoded, "can't call append before loading and locking write log")
+	}
+	return m.writeLogger.AppendMessage(m.bucket, m.key, m.latestWLVersion, msg)
+}
+
+// IncrementWLVersion should be called during snapshotting with a
+// write Tx held on the local resource. This ensures that any
+// writes which completed prior to the snapshot are in the prior
+// WL and any that complete after the snapshot are in the
+// incremented WL.
+func (m *Manager) IncrementWLVersion() error {
+	m.latestWLVersion++
+	m.lastWLPos = -1
+	m.loadWLsPastVersion = -1
+	return nil
+}
+
+// Snapshot takes a ReadCloser which has the contents of the
+// resource being tracked at a particular point in time and writes
+// them to the Snapshot Store. Upon a successful write it will
+// truncate any write logs which are now incorporated into the
+// snapshot.
+func (m *Manager) Snapshot(rc io.ReadCloser) error {
+	// latestWLVersion has already been incremented at this point, so
+	// we write that version minus 1.
+	err := m.snapshotter.Write(m.bucket, m.key, m.latestWLVersion-1, rc)
+	if err != nil {
+		return errors.Wrap(err, "writing snapshot")
+	}
+	err = m.writeLogger.DeleteLog(m.bucket, m.key, m.latestWLVersion-1)
+	return errors.Wrap(err, "deleting old write log")
+}
+
+// SnapshotTo is Snapshot's ugly stepsister supporting the weirdness
+// of reading from translate stores who we're hoping to off in the
+// next season.
+func (m *Manager) SnapshotTo(wt io.WriterTo) error {
+	err := m.snapshotter.WriteTo(m.bucket, m.key, m.latestWLVersion-1, wt)
+	if err != nil {
+		return errors.Wrap(err, "writing snapshot SnapshotTo")
+	}
+	err = m.writeLogger.DeleteLog(m.bucket, m.key, m.latestWLVersion-1)
+	return errors.Wrap(err, "deleting old write log snapshotTo")
+}
+
+// Unlock releases the lock. This should be called if control of
+// the underlying resource is being transitioned to another
+// node. Ideally it's also called if the process crashes (e.g. via
+// a defer), but an implementation based on filesystem locks
+// should have those removed by the operating system when the
+// process exits anyway.
+func (m *Manager) Unlock() error {
+	if err := m.writeLogger.Unlock(m.bucket, m.key); err != nil {
+		return errors.Wrap(err, "unlocking")
+	}
+	m.locked = false
+	return nil
+}
+
+const (
+	keysFileName = "keys"
+)
+
+func partitionBucket(table dax.TableKey, partition dax.PartitionNum) string {
+	return path.Join(string(table), "partition", fmt.Sprintf("%d", partition))
+}
+
+func shardKey(shard dax.ShardNum) string {
+	return path.Join("shard", fmt.Sprintf("%d", shard))
+}
+
+func fieldBucket(table dax.TableKey, field dax.FieldName) string {
+	return path.Join(string(table), "field", string(field))
+}
diff --git a/dax/storage/storage_test.go b/dax/storage/storage_test.go
new file mode 100644
index 000000000..1bc33e3e6
--- /dev/null
+++ b/dax/storage/storage_test.go
@@ -0,0 +1,165 @@
+package storage
+
+import (
+	"bytes"
+	"io"
+	"os"
+
+	"testing"
+
+	"github.com/molecula/featurebase/v3/dax"
+	"github.com/molecula/featurebase/v3/dax/snapshotter"
+	"github.com/molecula/featurebase/v3/dax/writelogger"
+	"github.com/molecula/featurebase/v3/logger"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestManagerManager(t *testing.T) {
+	sdd, err := os.MkdirTemp("", "snaptest*")
+	assert.NoError(t, err)
+	wdd, err := os.MkdirTemp("", "wltest*")
+	assert.NoError(t, err)
+	defer func() {
+		os.RemoveAll(sdd)
+		os.RemoveAll(wdd)
+	}()
+
+	sn := snapshotter.New(snapshotter.Config{
+		DataDir: sdd,
+	})
+	wl := writelogger.New(writelogger.Config{
+		DataDir: wdd,
+	})
+
+	mm := NewManagerManager(sn, wl, logger.NewStandardLogger(os.Stderr))
+
+	qtid := dax.QualifiedTableID{
+		TableQualifier: dax.TableQualifier{
+			OrganizationID: dax.OrganizationID("org1"),
+			DatabaseID:     dax.DatabaseID("db1"),
+		},
+		ID:   dax.TableID("blah"),
+		Name: "blah",
+	}
+
+	// get a manager and perform normal startup routine on empty data
+	mgr := mm.GetShardManager(qtid, dax.PartitionNum(1), dax.ShardNum(1))
+
+	d, err := mgr.LoadLatestSnapshot()
+	assert.NoError(t, err)
+	n, err := d.Read(make([]byte, 8))
+	assert.Equal(t, 0, n)
+	assert.Equal(t, err, io.EOF)
+
+	wld, err := mgr.LoadWriteLog()
+	assert.NoError(t, err)
+	n, err = wld.Read(make([]byte, 8))
+	assert.Equal(t, 0, n)
+	assert.Equal(t, io.EOF, err)
+
+	err = mgr.Lock()
+	assert.NoError(t, err)
+
+	wld, err = mgr.LoadWriteLog()
+	assert.NoError(t, err)
+	n, err = wld.Read(make([]byte, 8))
+	assert.Equal(t, 0, n)
+	assert.Equal(t, io.EOF, err)
+
+	// append some data
+	err = mgr.Append([]byte("blahblah"))
+	assert.NoError(t, err)
+
+	mm2 := NewManagerManager(sn, wl, logger.NewStandardLogger(os.Stderr))
+	// get second manager for same stuff
+	mgr2 := mm2.GetShardManager(qtid, dax.PartitionNum(1), dax.ShardNum(1))
+	// load snapshot on 2nd manager (empty)
+	d, err = mgr2.LoadLatestSnapshot()
+	assert.NoError(t, err)
+	n, err = d.Read(make([]byte, 8))
+	assert.Equal(t, 0, n)
+	assert.Equal(t, io.EOF, err)
+
+	// load WL on 2nd manager (blahblah)
+	wld, err = mgr2.LoadWriteLog()
+	assert.NoError(t, err)
+	buf := make([]byte, 16)
+	n, err = wld.Read(buf)
+	assert.Equal(t, 9, n)
+	assert.Equal(t, "blahblah\n", string(buf[:9]))
+	n, err = wld.Read(buf)
+	assert.Equal(t, 0, n)
+	assert.Equal(t, io.EOF, err)
+
+	// begin snapshot procedure on 1st manager
+	err = mgr.IncrementWLVersion()
+	assert.NoError(t, err)
+
+	// do append on 1st manager mid-snapshot
+	err = mgr.Append([]byte("blahbla2"))
+	assert.NoError(t, err)
+
+	// snapshot 1st manager
+	rc := io.NopCloser(bytes.NewBufferString("hahaha"))
+	err = mgr.Snapshot(rc)
+	assert.NoError(t, err)
+
+	// append again on 1st manager
+	err = mgr.Append([]byte("blahbla3"))
+	assert.NoError(t, err)
+
+	// locking 2nd manager should fail
+	err = mgr2.Lock()
+	assert.NotNil(t, err)
+
+	// exit 1st manager
+	err = mgr.Unlock()
+	assert.NoError(t, err)
+
+	// locking 2nd manager should succeed
+	err = mgr2.Lock()
+	assert.NoError(t, err)
+
+	// loading write log should fail since there's been a snapshot
+	// between the last load and locking.
+	wld, err = mgr2.LoadWriteLog()
+	assert.NotNil(t, err)
+
+	// mgr2 dies due to error loading write lock
+	err = mgr2.Unlock()
+	assert.NoError(t, err)
+
+	// get third manager for same stuff
+	mm3 := NewManagerManager(sn, wl, logger.NewStandardLogger(os.Stderr))
+	mgr3 := mm3.GetShardManager(qtid, dax.PartitionNum(1), dax.ShardNum(1))
+	// load snapshot on 3nd manager
+	d, err = mgr3.LoadLatestSnapshot()
+	assert.NoError(t, err)
+	buf = make([]byte, 6)
+	n, err = d.Read(buf)
+	assert.Equal(t, 6, n)
+	assert.Equal(t, "hahaha", string(buf))
+	assert.Equal(t, nil, err)
+
+	// load write log on 3rd manager, get previous 2 writes
+	wld, err = mgr3.LoadWriteLog()
+	assert.NoError(t, err)
+	buf = make([]byte, 20)
+	n, _ = wld.Read(buf)
+	assert.Equal(t, 18, n)
+	assert.Equal(t, "blahbla2\nblahbla3\n", string(buf[:18]))
+	n, err = wld.Read(buf)
+	assert.Equal(t, 0, n)
+	assert.Equal(t, io.EOF, err)
+
+	// lock 3rd manager
+	err = mgr3.Lock()
+	assert.NoError(t, err)
+
+	// reload write log (should be empty)
+	wld, err = mgr3.LoadWriteLog()
+	assert.NoError(t, err)
+	n, err = wld.Read(make([]byte, 8))
+	assert.Equal(t, 0, n)
+	assert.Equal(t, io.EOF, err)
+}
diff --git a/dax/storage/util.go b/dax/storage/util.go
new file mode 100644
index 000000000..85c5e7adf
--- /dev/null
+++ b/dax/storage/util.go
@@ -0,0 +1,24 @@
+package storage
+
+import "io"
+
+// trackingReader wraps a Reader and calls an custom "update" function
+// whenever Read is called. Used by the storage layer to keep track of
+// how much of the writelog has been read.
+type trackingReader struct {
+	r      io.Reader
+	update func(int, error)
+}
+
+func (tr *trackingReader) Read(p []byte) (n int, err error) {
+	n, err = tr.r.Read(p)
+	tr.update(n, err)
+	return n, err
+}
+
+func (tr *trackingReader) Close() error {
+	if closer, ok := tr.r.(io.Closer); ok {
+		return closer.Close()
+	}
+	return nil
+}
diff --git a/dax/writelogger/writelogger.go b/dax/writelogger/writelogger.go
index db54f859b..ae7a8957a 100644
--- a/dax/writelogger/writelogger.go
+++ b/dax/writelogger/writelogger.go
@@ -7,7 +7,9 @@ import (
 	"io/fs"
 	"os"
 	"path"
+	"strconv"
 	"sync"
+	"syscall"
 
 	"github.com/featurebasedb/featurebase/v3/errors"
 	"github.com/featurebasedb/featurebase/v3/logger"
@@ -16,17 +18,19 @@ import (
 type WriteLogger struct {
 	mu sync.RWMutex
 
-	dataDir  string
-	logFiles map[string]*os.File
+	dataDir   string
+	logFiles  map[string]*os.File
+	lockFiles map[string]*os.File
 
 	logger logger.Logger
 }
 
 func New(cfg Config) *WriteLogger {
 	return &WriteLogger{
-		dataDir:  cfg.DataDir,
-		logFiles: make(map[string]*os.File),
-		logger:   logger.NopLogger,
+		dataDir:   cfg.DataDir,
+		logFiles:  make(map[string]*os.File),
+		lockFiles: make(map[string]*os.File),
+		logger:    logger.NopLogger,
 	}
 }
 
@@ -43,25 +47,58 @@ func (w *WriteLogger) AppendMessage(bucket string, key string, version int, mess
 		return errors.Wrapf(err, "getting log file by key: %s", fKey)
 	}
 
-	logFile.Write(append(message, "\n"...))
-	logFile.Sync()
-
-	return nil
+	_, err = logFile.Write(append(message, "\n"...))
+	if err != nil {
+		return errors.Wrapf(err, "writing to log file %s", logFile.Name())
+	}
+	err = logFile.Sync()
+	return errors.Wrapf(err, "syncing log file %s", logFile.Name())
 }
 
-func (w *WriteLogger) LogReader(bucket string, key string, version int) (io.Reader, io.Closer, error) {
+func (w *WriteLogger) List(bucket, key string) ([]computer.WriteLogInfo, error) {
+	dirpath := path.Join(w.dataDir, bucket, key)
+
+	entries, err := os.ReadDir(dirpath)
+	if err != nil {
+		if pe, ok := err.(*os.PathError); ok && pe.Err == syscall.ENOENT {
+			return nil, nil
+		}
+		return nil, errors.Wrap(err, "reading directory")
+	}
+
+	wLogs := make([]computer.WriteLogInfo, len(entries))
+	for i, entry := range entries {
+		version, err := strconv.ParseInt(entry.Name(), 10, 64)
+		if err != nil {
+			return nil, errors.Wrapf(err, "writelog filename '%s' could not be parsed to version number", entry.Name())
+		}
+		wLogs[i] = computer.WriteLogInfo{
+			Version: int(version),
+		}
+	}
+	return wLogs, nil
+}
+
+func (w *WriteLogger) LogReader(bucket, key string, version int) (io.ReadCloser, error) {
+	return w.LogReaderFrom(bucket, key, version, 0)
+}
+
+func (w *WriteLogger) LogReaderFrom(bucket string, key string, version int, offset int) (io.ReadCloser, error) {
 	_, filePath := w.paths(fullKey(bucket, key, version))
 
 	f, err := os.Open(filePath)
 	if err != nil {
 		if e, ok := err.(*fs.PathError); ok {
-			return nil, nil, e
+			return nil, e
 		}
-		return nil, nil, err
+		return nil, err
+	}
+	if offset > 0 {
+		f.Seek(int64(offset), io.SeekStart)
 	}
 	w.logger.Debugf("WriteLogger LogReader file: %s", f.Name())
 
-	return f, f, nil
+	return f, nil
 }
 
 func (w *WriteLogger) DeleteLog(bucket string, key string, version int) error {
@@ -84,6 +121,62 @@ func (w *WriteLogger) DeleteLog(bucket string, key string, version int) error {
 	return os.Remove(f.Name())
 }
 
+func (w *WriteLogger) lockFile(bucket, key string) (string, string) {
+	lockFile := path.Join(w.dataDir, bucket, fmt.Sprintf("_lock_%s", key))
+	return path.Dir(lockFile), lockFile
+}
+
+func (w *WriteLogger) Lock(bucket, key string) error {
+	lockDir, lockFile := w.lockFile(bucket, key)
+	fmt.Println("lock dir:", lockDir)
+	fmt.Println("lock fil:", lockFile)
+
+	if err := os.MkdirAll(lockDir, 0777); err != nil {
+		return errors.Wrapf(err, "lock dir %s", lockDir)
+	}
+	f, err := os.OpenFile(lockFile, os.O_CREATE|os.O_EXCL, 0644)
+	if err != nil {
+		return errors.Wrapf(err, "opening lock file: %s", lockFile)
+	}
+	w.lockFiles[lockFile] = f
+	// fd, err = syscall.Open(lockFile, syscall.O_RDWR|syscall.O_CREAT, 0644)
+	// if err != nil {
+	// 	return 0, errors.Wrapf(err, "syscall opening %s", lockFile)
+	// }
+	// err = syscall.FcntlFlock(uintptr(fd), syscall.F_SETLK, &syscall.Flock_t{
+	// 	Type: syscall.F_WRLCK,
+	// })
+	return errors.Wrap(err, "locking")
+
+}
+
+func (w *WriteLogger) Unlock(bucket, key string) error {
+	// TODO(jaffee) since the file isn't guaranteed to be removed if
+	// the process is killed, we should actually use flock instead of
+	// EXCL file creation. Problem with that is it makes testing
+	// tricky because file handles from the same process are able to
+	// acquire the flock simultaneously. Headache.
+	_, lockFile := w.lockFile(bucket, key)
+	f, ok := w.lockFiles[lockFile]
+	if !ok {
+		return errors.New(errors.ErrUncoded, "couldn't find file to unlock")
+	}
+	f.Close()
+	err := os.Remove(lockFile)
+	delete(w.lockFiles, lockFile)
+
+	// defer func() {
+	// 	err := syscall.Close(fd)
+	// 	if err != nil {
+	// 		w.logger.Printf("error closing lockfile %s", lockFile)
+	// 	}
+	// }()
+	// err := syscall.FcntlFlock(uintptr(fd), syscall.F_SETLK, &syscall.Flock_t{
+	// 	Type: syscall.F_UNLCK,
+	// })
+	return errors.Wrap(err, "closing lock file")
+}
+
 // paths takes a key and returns the full file path (including the root data
 // directory) as well as the full directory path (i.e. the file path without the
 // file portion).
@@ -116,6 +209,7 @@ func (w *WriteLogger) logFileByKey(key string) (*os.File, error) {
 	if err != nil {
 		return nil, errors.Wrapf(err, "opening file: %s", filePath)
 	}
+	fmt.Printf("opened %s for key %s\n", filePath, key)
 	w.logFiles[key] = f
 
 	return f, nil
diff --git a/http_handler.go b/http_handler.go
index 95cefe351..797f31125 100644
--- a/http_handler.go
+++ b/http_handler.go
@@ -2447,7 +2447,7 @@ func (h *Handler) handleGetIndexShardSnapshot(w http.ResponseWriter, r *http.Req
 		return
 	}
 
-	rc, err := h.api.IndexShardSnapshot(r.Context(), indexName, shard)
+	rc, err := h.api.IndexShardSnapshot(r.Context(), indexName, shard, false)
 	if err != nil {
 		switch errors.Cause(err) {
 		case ErrIndexNotFound:
diff --git a/server.go b/server.go
index 8fb55ca64..0644ec48d 100644
--- a/server.go
+++ b/server.go
@@ -97,9 +97,7 @@ type Server struct { // nolint: maligned
 
 	executionPlannerFn ExecutionPlannerFn
 
-	writeLogReader     computer.WriteLogReader
-	writeLogWriter     computer.WriteLogWriter
-	snapshotReadWriter computer.SnapshotReadWriter
+	serverlessStorage *daxstorage.ManagerManager
 
 	dataframeEnabled bool
 }
@@ -429,15 +427,6 @@ func OptServerPartitionAssigner(p string) ServerOption {
 	}
 }
 
-// OptServerWriteLogReader provides an implemenation of the WriteLogReader
-// interface.
-func OptServerWriteLogReader(wlr computer.WriteLogReader) ServerOption {
-	return func(s *Server) error {
-		s.writeLogReader = wlr
-		return nil
-	}
-}
-
 func OptServerExecutionPlannerFn(fn ExecutionPlannerFn) ServerOption {
 	return func(s *Server) error {
 		s.executionPlannerFn = fn
@@ -445,20 +434,9 @@ func OptServerExecutionPlannerFn(fn ExecutionPlannerFn) ServerOption {
 	}
 }
 
-// OptServerWriteLogWriter provides an implemenation of the WriteLogWriter
-// interface.
-func OptServerWriteLogWriter(wlw computer.WriteLogWriter) ServerOption {
+func OptServerServerlessStorage(mm *daxstorage.ManagerManager) ServerOption {
 	return func(s *Server) error {
-		s.writeLogWriter = wlw
-		return nil
-	}
-}
-
-// OptServerSnapshotReadWriter provides an implemenation of the
-// SnapshotReadWriter interface.
-func OptServerSnapshotReadWriter(snap computer.SnapshotReadWriter) ServerOption {
-	return func(s *Server) error {
-		s.snapshotReadWriter = snap
+		s.serverlessStorage = mm
 		return nil
 	}
 }
@@ -579,7 +557,7 @@ func NewServer(opts ...ServerOption) (*Server, error) {
 	s.cluster.disCo = s.disCo
 	s.cluster.noder = s.noder
 	s.cluster.sharder = s.sharder
-	s.cluster.writeLogWriter = s.writeLogWriter
+	s.cluster.serverlessStorage = s.serverlessStorage
 	s.cluster.versionStore = versionStore
 
 	// Append the NodeID tag to stats.
diff --git a/server/server.go b/server/server.go
index dcc37a016..50a122329 100644
--- a/server/server.go
+++ b/server/server.go
@@ -75,9 +75,10 @@ type Command struct {
 	logger         loggerLogger
 	queryLogger    loggerLogger
 
-	Registrar       computer.Registrar
-	writeLogService computer.WriteLogService
-	snapshotService computer.SnapshotService
+	Registrar         computer.Registrar
+	serverlessStorage *storage.ManagerManager
+	writeLogService   computer.WriteLogService
+	snapshotService   computer.SnapshotService
 
 	Handler      pilosa.HandlerI
 	httpHandler  http.Handler
@@ -550,19 +551,8 @@ func (m *Command) setupServer() error {
 		m.Config.Etcd.Dir = filepath.Join(path, pilosa.DiscoDir)
 	}
 
-	// WriteLogger setup.
-	var wlw computer.WriteLogWriter = computer.NewNopWriteLogWriter()
-	var wlr computer.WriteLogReader = computer.NewNopWriteLogReader()
-	if m.writeLogService != nil {
-		wlrw := computer.NewWriteLogReadWriter(m.writeLogService)
-		wlr = wlrw
-		wlw = wlrw
-	}
-
-	// Snapshotter setup.
-	var snap computer.SnapshotReadWriter = computer.NewNopSnapshotReadWriter()
-	if m.snapshotService != nil {
-		snap = computer.NewSnapshotReadWriter(m.snapshotService)
+	if m.writeLogService != nil && m.snapshotService != nil {
+		m.serverlessStorage = storage.NewManagerManager(m.snapshotService, m.writeLogService, m.logger)
 	}
 
 	executionPlannerFn := func(e pilosa.Executor, api *pilosa.API, sql string) sql3.CompilePlanner {
@@ -599,9 +589,7 @@ func (m *Command) setupServer() error {
 		pilosa.OptServerQueryHistoryLength(m.Config.QueryHistoryLength),
 		pilosa.OptServerPartitionAssigner(m.Config.Cluster.PartitionToNodeAssignment),
 		pilosa.OptServerExecutionPlannerFn(executionPlannerFn),
-		pilosa.OptServerWriteLogReader(wlr),
-		pilosa.OptServerWriteLogWriter(wlw),
-		pilosa.OptServerSnapshotReadWriter(snap),
+		pilosa.OptServerServerlessStorage(m.serverlessStorage),
 		pilosa.OptServerIsDataframeEnabled(m.Config.Dataframe.Enable),
 	}
 
@@ -646,9 +634,7 @@ func (m *Command) setupServer() error {
 	m.API, err = pilosa.NewAPI(
 		pilosa.OptAPIServer(m.Server),
 		pilosa.OptAPIImportWorkerPoolSize(m.Config.ImportWorkerPoolSize),
-		pilosa.OptAPIWriteLogReader(wlr),
-		pilosa.OptAPIWriteLogWriter(wlw),
-		pilosa.OptAPISnapshotter(snap),
+		pilosa.OptAPIServerlessStorage(m.serverlessStorage),
 		pilosa.OptAPIDirectiveWorkerPoolSize(m.Config.DirectiveWorkerPoolSize),
 		pilosa.OptAPIIsComputeNode(m.isComputeNode),
 	)