Merge branch 'master' into 54mir/authentication

2026-09-07 17:15:56 +00:00 · 2021-12-20 16:42:11 -06:00 · 2021-12-20 16:42:11 -06:00 · 3b374a62bf
commit 3b374a62bf
parent a24a1e8922 18d2ef3d53
37 changed files with 1699 additions and 618 deletions
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@ -144,16 +144,6 @@ jobs:
      - run:
          command: make test-external-lookup EXTERNAL_LOOKUP_DSN=postgresql://postgres:password@localhost/circle_test?sslmode=disable
          no_output_timeout: 30m
-  test-backup-restore:
-    executor:
-      name: golang
-    steps:
-      - checkout-plus
-      - skip-if-root-unchanged
-      - setup_remote_docker
-      - run: echo -n $DOCKER_PASS | docker login -u $DOCKER_USER --password-stdin
-      - run: make backuptests-build 
-      - run: make backuptests
  cluster-tests:
    executor:
      name: golang
@ -273,10 +263,6 @@ workflows:
          context: molecula
          requires:
            - setup
-      - test-backup-restore:
-          context: molecula
-          requires:
-            - setup
      - cluster-tests:
          context: molecula
          requires:
@ -292,8 +278,6 @@ workflows:
          filters:
            tags:
              only: /^v.*/
-            branches:
-              only: master
      - publish_release:
          context: molecula
          requires:
--- a/Dockerfile.pilosa
+++ b/Dockerfile.pilosa
@ -1,37 +0,0 @@
-ARG GO_VERSION=latest
-
-######################
-### Pilosa builder ###
-######################
-
-FROM golang:${GO_VERSION} as pilosa-builder
-ARG MAKE_FLAGS
-WORKDIR /pilosa
-
-COPY . ./
-
-RUN make build FLAGS="-o build/featurebase" ${MAKE_FLAGS}
-
-#####################
-### Pilosa runner ###
-#####################
-
-FROM alpine:3.13.2 as runner
-
-LABEL maintainer "dev@molecula.com"
-
-RUN apk add --no-cache curl jq
-
-COPY --from=pilosa-builder /pilosa/build/featurebase /
-
-COPY NOTICE /NOTICE
-
-EXPOSE 10101
-VOLUME /data
-
-ENV PILOSA_DATA_DIR /data
-ENV PILOSA_BIND 0.0.0.0:10101
-ENV PILOSA_BIND_GRPC 0.0.0.0:20101
-
-ENTRYPOINT ["/featurebase"]
-CMD ["server"]
--- a/Dockerfile.runner
+++ b/Dockerfile.runner
@ -1,23 +0,0 @@
-ARG GO_VERSION=latest
-
-######################
-### Pilosa builder ###
-######################
-
-FROM golang:${GO_VERSION} as pilosa-builder
-ARG MAKE_FLAGS
-WORKDIR /pilosa
-
-COPY . ./
-
-RUN make build FLAGS="-o build/featurebase" ${MAKE_FLAGS}
-
-FROM moleculacorp/idk as idk
-LABEL maintainer "dev@molecula.com"
-RUN apt-get update -y
-RUN apt-get install -y bash curl jq
-
-
-COPY --from=pilosa-builder /pilosa/build/featurebase /
-COPY testBackupRestore.sh /
-CMD ["bash","/testBackupRestore.sh"]
--- a/9
+++ b/9
@ -155,15 +155,6 @@ clustertests: vendor
 clustertests-build: vendor
 	docker-compose -f $(DOCKER_COMPOSE) down -v
 	docker-compose -f $(DOCKER_COMPOSE) up --exit-code-from=client1 --build
-# Test Cluster backup and restore
-backuptests-build: vendor
-	docker-compose -f docker-compose-3.yml down
-	docker-compose -f docker-compose-3.yml build 
-
-backuptests: vendor
-	docker-compose -f docker-compose-3.yml down -v
-	docker-compose -f docker-compose-3.yml up --exit-code-from=client1  --abort-on-container-exit
-

 # Install Pilosa
 install:
--- a/api.go
+++ b/api.go
@ -834,6 +834,15 @@ func (api *API) FragmentData(ctx context.Context, indexName, fieldName, viewName
 	return f, nil
 }

+type RedirectError struct {
+	HostPort string
+	error    string
+}
+
+func (r RedirectError) Error() string {
+	return r.error
+}
+
 // TranslateData returns all translation data in the specified partition.
 func (api *API) TranslateData(ctx context.Context, indexName string, partition int) (io.WriterTo, error) {
 	span, _ := tracing.StartSpanFromContext(ctx, "API.TranslateData")
@ -849,6 +858,15 @@ func (api *API) TranslateData(ctx context.Context, indexName string, partition i
 		return nil, newNotFoundError(ErrIndexNotFound, indexName)
 	}

+	snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
+	nodes := snap.PartitionNodes(partition)
+	if nodes[0].ID != api.server.NodeID() {
+		return nil, RedirectError{
+			HostPort: nodes[0].URI.HostPort(),
+			error:    fmt.Sprintf("can't translate data, this node(%s) does not partition %d", api.server.uri, partition),
+		}
+	}
+
 	// Retrieve translatestore from holder.
 	store := idx.TranslateStore(partition)
 	if store == nil {
@ -978,7 +996,10 @@ func (api *API) Usage(ctx context.Context, remote bool) (map[string]NodeUsage, e
 		return resp, nil
 	}

-	if api.usageCache.lastCalcDuration < usageCacheMinDuration {
+	api.usageCache.muAssign.Lock()
+	lastCalc := api.usageCache.lastCalcDuration
+	api.usageCache.muAssign.Unlock()
+	if lastCalc < usageCacheMinDuration {
 		err := api.ResetUsageCache()
 		if err != nil {
 			api.server.logger.Infof("could not reset usageCache: %s", err)
@ -2732,8 +2753,8 @@ func (api *API) RestoreShard(ctx context.Context, indexName string, shard uint64

 	for _, flv := range flvs {
 		fld := idx.field(flv.Field)
-		view, ok := fld.viewMap[flv.View]
-		if !ok {
+		view := fld.view(flv.View)
+		if view == nil {
 			view, err = fld.createViewIfNotExists(flv.View)
 			if err != nil {
 				return err
@ -2747,6 +2768,14 @@ func (api *API) RestoreShard(ctx context.Context, indexName string, shard uint64
 		if err != nil {
 			return err
 		}
+		bd, err := view.bitDepth([]uint64{shard})
+		if err != nil {
+			return err
+		}
+		err = fld.cacheBitDepth(bd)
+		if err != nil {
+			return err
+		}
 	}

 	return nil
--- a/authz/authorization.go
+++ b/authz/authorization.go
@ -0,0 +1,150 @@
+// Copyright 2017 Pilosa Corp.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package authz
+
+import (
+	"fmt"
+	"io"
+	"io/ioutil"
+
+	"gopkg.in/yaml.v2"
+)
+
+type Auth struct {
+	// Enable AuthZ/AuthN for featurebase server
+	Enable bool `toml:"enable"`
+
+	// Application/Client ID
+	ClientId string `toml:"client-id"`
+
+	// Client Secret
+	ClientSecret string `toml:"client-secret"`
+
+	// Authorize URL
+	AuthorizeURL string `toml:"authorize-url"`
+
+	// Token URL
+	TokenURL string `toml:"token-url"`
+
+	// Group Endpoint URL
+	GroupEndpointURL string `toml:"group-endpoint-url"`
+
+	// Scope URL
+	ScopeURL string `toml:"scope-url"`
+
+	// Permissions file for groups
+	PermissionsFile string `toml:"permissions"`
+}
+
+type GroupPermissions struct {
+	Permissions map[string]map[string]string `yaml:"user-groups"`
+	Admin       string                       `yaml:"admin"`
+}
+
+type Group struct {
+	UserID    string
+	GroupID   string `json:"id"`
+	GroupName string `json:"displayName"`
+}
+
+func (p *GroupPermissions) ReadPermissionsFile(permsFile io.Reader) (err error) {
+	permsData, err := ioutil.ReadAll(permsFile)
+
+	if err != nil {
+		return fmt.Errorf("reading permissions failed with error: %s", err)
+	}
+
+	err = yaml.UnmarshalStrict(permsData, &p)
+	if err != nil {
+		return fmt.Errorf("unmarshalling permissions failed with error: %s", err)
+	}
+
+	return
+}
+
+func (p *GroupPermissions) GetPermissions(groups []Group, index string) (permission string, errors error) {
+
+	if admin := p.IsAdmin(groups); admin {
+		return "admin", nil
+	}
+
+	allPermissions := map[string]bool{
+		"write": false,
+		"read":  false,
+	}
+
+	if len(groups) == 0 {
+		return "", fmt.Errorf("user is not part of any groups in identity provider")
+	}
+
+	var groupsDenied []string
+	for _, group := range groups {
+		if _, ok := p.Permissions[group.GroupID]; ok {
+			if perm, ok := p.Permissions[group.GroupID][index]; ok {
+				allPermissions[perm] = true
+			} else {
+				return "", fmt.Errorf("user %s does not have permission to index %s", group.UserID, index)
+			}
+		} else {
+			groupsDenied = append(groupsDenied, group.GroupID)
+		}
+	}
+
+	if len(groupsDenied) == len(groups) {
+		return "", fmt.Errorf("group(s) %s does not have permission to FeatureBase", groupsDenied)
+	}
+
+	if allPermissions["write"] {
+		return "write", nil
+	} else if allPermissions["read"] {
+		return "read", nil
+	} else {
+		return "", fmt.Errorf("no permissions found")
+	}
+}
+
+func (p *GroupPermissions) IsAdmin(groups []Group) bool {
+	for _, group := range groups {
+		if p.Admin == group.GroupID {
+			return true
+		}
+	}
+	return false
+}
+
+func (p *GroupPermissions) GetAuthorizedIndexList(groups []Group, desiredPermission string) (indexList []string) {
+	// if user is admin, find all indexes in permissions file and return them
+	if admin := p.IsAdmin(groups); admin {
+		for groupId := range p.Permissions {
+			for index := range p.Permissions[groupId] {
+				indexList = append(indexList, index)
+			}
+		}
+		return indexList
+	}
+
+	for _, group := range groups {
+		if _, ok := p.Permissions[group.GroupID]; ok {
+			for index, permission := range p.Permissions[group.GroupID] {
+				if permission == desiredPermission {
+					indexList = append(indexList, index)
+				} else if permission == "write" && desiredPermission == "read" {
+					indexList = append(indexList, index)
+				}
+			}
+		}
+	}
+	return indexList
+}
--- a/authz/authorization_test.go
+++ b/authz/authorization_test.go
@ -0,0 +1,314 @@
+// Copyright 2017 Pilosa Corp.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package authz_test
+
+import (
+	"fmt"
+	"reflect"
+	"sort"
+	"strings"
+	"testing"
+
+	"github.com/molecula/featurebase/v2/authz"
+)
+
+func TestAuth_ReadPermissionsFile(t *testing.T) {
+
+	singleInput := `user-groups:
+  "dca35310-ecda-4f23-86cd-876aee55906b":
+    "test": "read"
+admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"`
+
+	multiInput := `user-groups:
+  "dca35310-ecda-4f23-86cd-876aee55906b":
+    "test": "read"
+    "test2": "write"
+  "dca35310-ecda-4f23-86cd-876aee559900":
+    "test": "write"
+admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"`
+
+	singlePermission := authz.GroupPermissions{
+		Permissions: map[string]map[string]string{
+			"dca35310-ecda-4f23-86cd-876aee55906b": {"test": "read"},
+		},
+		Admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe",
+	}
+
+	multiPermission := authz.GroupPermissions{
+		Permissions: map[string]map[string]string{
+			"dca35310-ecda-4f23-86cd-876aee55906b": {"test": "read", "test2": "write"},
+			"dca35310-ecda-4f23-86cd-876aee559900": {"test": "write"}},
+		Admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe",
+	}
+
+	tests := []struct {
+		input  string
+		output authz.GroupPermissions
+	}{
+		{singleInput, singlePermission},
+		{multiInput, multiPermission},
+	}
+
+	for i, test := range tests {
+		t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
+			permFile := strings.NewReader(test.input)
+
+			var p authz.GroupPermissions
+			err := p.ReadPermissionsFile(permFile)
+			if err != nil {
+				t.Fatalf("readPermissionsFile error: %s", err)
+			}
+
+			if !reflect.DeepEqual(p, test.output) {
+				t.Fatalf("expected output %s, but got %s", test.output, p)
+			}
+		},
+		)
+	}
+}
+
+func TestAuth_GetPermissions(t *testing.T) {
+
+	// initializes different example of permissions file in yaml
+	permissions1 := `"user-groups":
+  "dca35310-ecda-4f23-86cd-876aee55906b":
+    "test": "read"
+admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"`
+
+	permissions2 := `"user-groups":
+  "dca35310-ecda-4f23-86cd-876aee559900":
+    "test": "write"
+admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"`
+
+	permissions3 := `"user-groups":
+  "dca35310-ecda-4f23-86cd-876aee55906b":
+    "test": "write"
+    "test2": "read"
+  "dca35310-ecda-4f23-86cd-876aee559900":
+    "test": "read"
+admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"`
+
+	permissions4 := `"user-groups":
+  "dca35310-ecda-4f23-86cd-876aee559900":
+    "test": ""
+admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"`
+
+	// initializes groups that are returned from identity provider
+	groupName := "name"
+	userId := "user-id"
+	groupsList1 := []authz.Group{}
+	groupsList2 := []authz.Group{{userId, "fake-group", groupName}}
+	groupsList3 := []authz.Group{
+		{userId, "dca35310-ecda-4f23-86cd-876aee55906b", groupName},
+		{userId, "dca35310-ecda-4f23-86cd-876aee559900", groupName},
+	}
+	groupsList4 := []authz.Group{{userId, "ac97c9e2-346b-42a2-b6da-18bcb61a32fe", groupName}}
+
+	tests := []struct {
+		yamlData   string
+		groups     []authz.Group
+		index      string
+		userAccess string
+		err        string
+	}{
+		{
+			permissions1,
+			groupsList1,
+			"test",
+			"",
+			"user is not part of any groups in identity provider",
+		},
+		{
+			permissions1,
+			groupsList3,
+			"test1",
+			"",
+			"does not have permission to index",
+		},
+		{
+			permissions2,
+			groupsList2,
+			"test",
+			"",
+			"does not have permission to FeatureBase",
+		},
+		{
+			permissions1,
+			groupsList3,
+			"test",
+			"read",
+			"",
+		},
+		{
+			permissions2,
+			groupsList3,
+			"test",
+			"write",
+			"",
+		},
+		{
+			permissions3,
+			groupsList4,
+			"test",
+			"admin",
+			"",
+		},
+		{
+			permissions4,
+			groupsList3,
+			"test",
+			"",
+			"no permissions found",
+		},
+	}
+
+	for i, test := range tests {
+		t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
+
+			permFile := strings.NewReader(test.yamlData)
+
+			var p authz.GroupPermissions
+			if err := p.ReadPermissionsFile(permFile); err != nil {
+				t.Errorf("Error: %s", err)
+			}
+
+			p1, err := p.GetPermissions(test.groups, test.index)
+
+			if p1 != test.userAccess {
+				t.Errorf("expected permission to be %s, but got %s", test.userAccess, p1)
+			}
+
+			if err != nil {
+				if !strings.Contains(err.Error(), test.err) {
+					t.Errorf("expected error to contain %s, but got %s", test.err, err.Error())
+				}
+			}
+
+		})
+	}
+}
+
+func TestAuth_IsAdmin(t *testing.T) {
+
+	group1 := []authz.Group{
+		{"admin-user-id", "ac97c9e2-346b-42a2-b6da-18bcb61a32fe", "admin-group"},
+	}
+
+	group2 := []authz.Group{
+		{"user-id", "dca35310-ecda-4f23-86cd-876aee55906b", "group-name"},
+	}
+
+	groupPermissions := authz.GroupPermissions{
+		Permissions: map[string]map[string]string{
+			"dca35310-ecda-4f23-86cd-876aee55906b": {"test": "write"},
+		},
+		Admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe",
+	}
+
+	tests := []struct {
+		groups           []authz.Group
+		groupPermissions authz.GroupPermissions
+		output           bool
+	}{
+		{
+			group1, groupPermissions, true,
+		},
+		{
+			group2, groupPermissions, false,
+		},
+	}
+
+	for i, test := range tests {
+		t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
+			p := test.groupPermissions
+			resp := p.IsAdmin(test.groups)
+			if resp != test.output {
+				t.Errorf("expected %t, but got %t", test.output, resp)
+			}
+		})
+	}
+}
+
+func TestAuth_GetAuthorizedIndexList(t *testing.T) {
+
+	group1 := []authz.Group{
+		{"user-id", "dca35310-ecda-4f23-86cd-876aee55906b", "group-name"},
+	}
+
+	group2 := []authz.Group{
+		{"admin-user-id", "ac97c9e2-346b-42a2-b6da-18bcb61a32fe", "admin-group"},
+	}
+
+	group3 := []authz.Group{
+		{"user-id", "dca35310-ecda-4f23-86cd-876aee559900", "group-name"},
+	}
+
+	p := authz.GroupPermissions{
+		Permissions: map[string]map[string]string{
+			"dca35310-ecda-4f23-86cd-876aee55906b": {
+				"test1": "read",
+				"test2": "write",
+			},
+			"dca35310-ecda-4f23-86cd-876aee559900": {
+				"test3": "read",
+			},
+		},
+		Admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe",
+	}
+
+	tests := []struct {
+		groups     []authz.Group
+		permission string
+		output     []string
+	}{
+		{
+			group1,
+			"read",
+			[]string{"test1", "test2"},
+		},
+		{
+			group1,
+			"write",
+			[]string{"test2"},
+		},
+		{
+			group3,
+			"write",
+			nil,
+		},
+		{
+			group2,
+			"read",
+			[]string{"test1", "test2", "test3"},
+		},
+		{
+			group2,
+			"write",
+			[]string{"test1", "test2", "test3"},
+		},
+	}
+
+	for i, test := range tests {
+		t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
+
+			indexList := p.GetAuthorizedIndexList(test.groups, test.permission)
+			sort.Strings(indexList)
+
+			if !reflect.DeepEqual(indexList, test.output) {
+				t.Errorf("expected %s, but got %s", test.output, indexList)
+			}
+		})
+	}
+
+}
--- a/client/client.go
+++ b/client/client.go
@ -808,7 +808,6 @@ func (c *Client) shardsMax() (map[string]uint64, error) {
 }

 // HTTPRequest sends an HTTP request to the Pilosa server (used by idk)
-// nolint: deadcode
 func (c *Client) HTTPRequest(method string, path string, data []byte, headers map[string]string) (status int, body []byte, err error) {
 	span := c.tracer.StartSpan("Client.HTTPRequest")

--- a/cmd/roaring-migrate/main.go
+++ b/cmd/roaring-migrate/main.go
@ -273,21 +273,19 @@ func Migrate(dataDir, backupPath string) error {
 	})
 	//raw is now sorted by shard

-	// need index/field/shard
-	// make rbf file in backup
-	rowSize := uint64(0) //?
-	clear := false
-	log := false
 	cache := &rbfFile{
 		temp: filepath.Join(backupPath, "_SCRATCH"),
 	}
+	bm := roaring.NewSliceBitmap()
 	for _, filename := range raw {
 		index, field, view, shard := Extract(filename)
+
 		content, err := ioutil.ReadFile(dataDir + filename)
 		if err != nil {
 			return err
 		}
-		itr, err := roaring.NewRoaringIterator(content)
+		err = bm.UnmarshalBinary(content)
+
 		if err != nil {
 			return err
 		}
@ -300,10 +298,13 @@ func Migrate(dataDir, backupPath string) error {
 			return err
 		}
 		key := string(txkey.Prefix(index, field, view, shard))
-		_, _, err = tx.ImportRoaringBits(key, itr, clear, log, rowSize)
-		if err != nil {
-			tx.Rollback()
-			return err
+		itr, ok := bm.Containers.Iterator(0)
+		if ok {
+			for itr.Next() {
+				k, v := itr.Value()
+				tx.PutContainer(key, k, v)
+
+			}
 		}
 		err = tx.Commit()
 		if err != nil {
--- a/ctl/backup.go
+++ b/ctl/backup.go
@ -18,7 +18,7 @@ import (
 	"golang.org/x/sync/errgroup"
 )

-// BackupCommand represents a command for backing up a Pilosa node.
+// BackupCommand represents a command for backing up a FeatureBase node.
 type BackupCommand struct { // nolint: maligned
 	tlsConfig *tls.Config

@ -183,12 +183,17 @@ func (cmd *BackupCommand) backupIDAllocData(ctx context.Context) error {
 func (cmd *BackupCommand) backupIndexTranslation(ctx context.Context, ii *pilosa.IndexInfo) error {
 	logger := cmd.Logger()
 	logger.Printf("backing up index translation: %q", ii.Name)
-	if err := cmd.backupIndexTranslateData(ctx, ii.Name); err != nil {
-		return err
+	if ii.Options.Keys {
+		if err := cmd.backupIndexTranslateData(ctx, ii.Name); err != nil {
+			return err
+		}
 	}

 	// Back up field translation data.
 	for _, fi := range ii.Fields {
+		if !fi.Options.Keys {
+			continue
+		}
 		if err := cmd.backupFieldTranslateData(ctx, ii.Name, fi.Name); err != nil {
 			return fmt.Errorf("cannot backup field translation data for field %q on index %q: %w", fi.Name, ii.Name, err)
 		}
@ -286,7 +291,6 @@ func (cmd *BackupCommand) backupShardNode(ctx context.Context, indexName string,
 func (cmd *BackupCommand) backupIndexTranslateData(ctx context.Context, name string) error {
 	partitionN := topology.DefaultPartitionN

-	// Back up all bitmap data for the index.
 	ch := make(chan int, partitionN)
 	for partitionID := 0; partitionID < partitionN; partitionID++ {
 		ch <- partitionID
@ -318,9 +322,7 @@ func (cmd *BackupCommand) backupIndexPartitionTranslateData(ctx context.Context,
 	logger.Printf("backing up index translation data: %s/%d", name, partitionID)

 	rc, err := cmd.client.IndexTranslateDataReader(ctx, name, partitionID)
-	if err == pilosa.ErrTranslateStoreNotFound {
-		return nil
-	} else if err != nil {
+	if err != nil {
 		return fmt.Errorf("fetching translate data reader: %w", err)
 	}
 	defer rc.Close()
@ -349,9 +351,7 @@ func (cmd *BackupCommand) backupFieldTranslateData(ctx context.Context, indexNam
 	logger.Printf("backing up field translation data: %s/%s", indexName, fieldName)

 	rc, err := cmd.client.FieldTranslateDataReader(ctx, indexName, fieldName)
-	if err == pilosa.ErrTranslateStoreNotFound {
-		return nil
-	} else if err != nil {
+	if err != nil {
 		return fmt.Errorf("fetching translate data reader: %w", err)
 	}
 	defer rc.Close()
--- a/ctl/chksum.go
+++ b/ctl/chksum.go
@ -8,7 +8,7 @@ import (
 	"io"

 	"github.com/cespare/xxhash"
-	"github.com/molecula/featurebase/v2"
+	pilosa "github.com/molecula/featurebase/v2"
 	"github.com/molecula/featurebase/v2/server"
 )

@ -61,14 +61,19 @@ func (cmd *ChkSumCommand) Run(ctx context.Context) (err error) {
 	h := xxhash.New()

 	for _, ii := range schema.Indexes {
-		qa := &pilosa.QueryRequest{Index: ii.Name, Query: "Count(All())"}
+		qa := &pilosa.QueryRequest{Index: ii.Name, Query: "All()"}
 		rs, err := client.Query(ctx, ii.Name, qa)
 		if err != nil {
 			return err
 		}
-		all := rs.Results[0].(uint64)
-		as := fmt.Sprintf("all=%v", all)
-		_, _ = h.Write([]byte(as))
+
+		all := rs.Results[0].(*pilosa.Row)
+		if len(all.Keys) > 0 {
+			allString := fmt.Sprintf("%v", all.Keys)
+			_, _ = h.Write([]byte(allString))
+		} else {
+			_, _ = h.Write(all.Roaring())
+		}

 		for _, field := range ii.Fields {
 			switch field.Options.Type {
@ -92,7 +97,7 @@ func (cmd *ChkSumCommand) Run(ctx context.Context) (err error) {
 				}
 				for _, item := range res.Results {
 					rowids := item.(*pilosa.RowIdentifiers)
-					//either rowids or keys
+					// either rowids or keys
 					for _, row := range rowids.Keys {
 						countPql := fmt.Sprintf(`Count(Row(%v="%v"))`, field.Name, row)
 						qr := &pilosa.QueryRequest{Index: ii.Name, Query: countPql}
@ -121,7 +126,7 @@ func (cmd *ChkSumCommand) Run(ctx context.Context) (err error) {
 			}

 		}
-		fmt.Printf("hash:%x\n", h.Sum(nil))
+		fmt.Fprintf(cmd.Stdout, "hash:%x\n", h.Sum(nil))
 	}

 	return nil
--- a/ctl/rbf_pages.go
+++ b/ctl/rbf_pages.go
@ -69,9 +69,9 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error {

 	// Print one line for each page.
 	for pgno, info := range infos {
+		fmt.Fprintf(cmd.Stdout, "%-8d ", pgno)
 		switch info := info.(type) {
 		case *rbf.MetaPageInfo:
-			fmt.Fprintf(cmd.Stdout, "%-8d ", pgno)
 			fmt.Fprintf(cmd.Stdout, "%-10s ", "meta")
 			if cmd.WithTree {
 				fmt.Fprintf(cmd.Stdout, "%-30q ", "")
@ -79,7 +79,6 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error {
 			fmt.Fprintf(cmd.Stdout, "pageN=%d,walid=%d,rootrec=%d,freelist=%d\n", info.PageN, info.WALID, info.RootRecordPageNo, info.FreelistPageNo)

 		case *rbf.RootRecordPageInfo:
-			fmt.Fprintf(cmd.Stdout, "%-8d ", pgno)
 			fmt.Fprintf(cmd.Stdout, "%-10s ", "rootrec")
 			if cmd.WithTree {
 				fmt.Fprintf(cmd.Stdout, "%-30q ", "")
@ -87,7 +86,6 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error {
 			fmt.Fprintf(cmd.Stdout, "next=%d\n", info.Next)

 		case *rbf.LeafPageInfo:
-			fmt.Fprintf(cmd.Stdout, "%-8d ", pgno)
 			fmt.Fprintf(cmd.Stdout, "%-10s ", "leaf")
 			if cmd.WithTree {
 				fmt.Fprintf(cmd.Stdout, "%-30q ", prefixToString(info.Tree))
@ -95,7 +93,6 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error {
 			fmt.Fprintf(cmd.Stdout, "flags=x%x,celln=%d\n", info.Flags, info.CellN)

 		case *rbf.BranchPageInfo:
-			fmt.Fprintf(cmd.Stdout, "%-8d ", pgno)
 			fmt.Fprintf(cmd.Stdout, "%-10s ", "branch")
 			if cmd.WithTree {
 				fmt.Fprintf(cmd.Stdout, "%-30q ", prefixToString(info.Tree))
@ -103,7 +100,6 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error {
 			fmt.Fprintf(cmd.Stdout, "flags=x%x,celln=%d\n", info.Flags, info.CellN)

 		case *rbf.BitmapPageInfo:
-			fmt.Fprintf(cmd.Stdout, "%-8d ", pgno)
 			fmt.Fprintf(cmd.Stdout, "%-10s ", "bitmap")
 			if cmd.WithTree {
 				fmt.Fprintf(cmd.Stdout, "%-30q ", prefixToString(info.Tree))
@ -111,7 +107,6 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error {
 			fmt.Fprintf(cmd.Stdout, "-\n")

 		case *rbf.FreePageInfo:
-			fmt.Fprintf(cmd.Stdout, "%-8d ", pgno)
 			fmt.Fprintf(cmd.Stdout, "%-10s ", "free")
 			if cmd.WithTree {
 				fmt.Fprintf(cmd.Stdout, "%-30q ", "")
@ -119,7 +114,7 @@ func (cmd *RBFPagesCommand) Run(ctx context.Context) error {
 			fmt.Fprintf(cmd.Stdout, "-\n")

 		default:
-			panic(fmt.Sprintf("unexpected page info type %T", info))
+			fmt.Fprintf(cmd.Stdout, "unknown [%T]\n", info)
 		}
 	}

--- a/ctl/server.go
+++ b/ctl/server.go
@ -85,7 +85,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
 	flags.BoolVar(&srv.Config.Storage.FsyncEnabled, "storage.fsync", true, "enable fsync fully safe flush-to-disk")

 	// RowcacheOn
-	flags.BoolVar((&srv.Config.RowcacheOn), "rowcache-on", srv.Config.RowcacheOn, "turn on the rowcache for all backends (may speed some queries)")
+	flags.BoolVar((&srv.Config.RowcacheOn), "rowcache-on", srv.Config.RowcacheOn, "Do not use, permanently disabled. Flag exists for backwards compatibility and will be removed.")

 	// RBF specific flags. See pilosa/rbf/cfg/cfg.go for definitions.
 	srv.Config.RBFConfig.DefineFlags(flags)
@ -120,5 +120,6 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
 	flags.StringSliceVar(&srv.Config.Auth.Scopes, "auth.scopes", srv.Config.Auth.Scopes, "Comma separated list of scopes obtained from IdP")
 	flags.StringVar(&srv.Config.Auth.HashKey, "auth.hash-key", srv.Config.Auth.HashKey, "First Secret for Auth.")
 	flags.StringVar(&srv.Config.Auth.BlockKey, "auth.block-key", srv.Config.Auth.BlockKey, "Second Secret for Auth.")
+	flags.StringVar(&srv.Config.Auth.PermissionsFile, "auth.permissions", srv.Config.Auth.PermissionsFile, "Permissions' file with group authorization.")

 }
--- a/docker-compose-3.yml
+++ b/docker-compose-3.yml
@ -1,97 +0,0 @@
-version: "3"
-services:
-  pilosa0:
-    image: build/pilosa                                                                                                                                                     
-    build:                                                                                                                                                                 
-      context: .                                                                                                                                                           
-      dockerfile: Dockerfile.pilosa              
-    environment:
-      PILOSA_ADVERTISE: pilosa0:10101
-      PILOSA_ADVERTISE_GRPC: pilosa0:20101
-      PILOSA_CLUSTER_REPLICAS: 1
-      PILOSA_DATA_DIR: /data/pilosa0
-      PILOSA_ETCD_ADVERTISE_CLIENT_ADDRESS: http://pilosa0:10201
-      PILOSA_ETCD_ADVERTISE_PEER_ADDRESS: http://pilosa0:10301
-      PILOSA_ETCD_INITIAL_CLUSTER: pilosa0=http://pilosa0:10301,pilosa1=http://pilosa1:10301,pilosa2=http://pilosa2:10301
-      PILOSA_ETCD_LISTEN_CLIENT_ADDRESS: http://0.0.0.0:10201
-      PILOSA_ETCD_LISTEN_PEER_ADDRESS: http://0.0.0.0:10301
-      PILOSA_NAME: pilosa0
-      PILOSA_STORAGE_BACKEND: ${PILOSA_STORAGE_BACKEND:-rbf}
-    volumes:
-    - data:/data
-    healthcheck:
-      test: x=$$(curl -s localhost:10101/status | jq -r ".state") && [[ "$$x" == "NORMAL"  ]] || $$(exit 1)
-      interval: 10s
-      timeout: 5s
-      retries: 5
-  pilosa1:
-    image: build/pilosa                                                                                                                                                     
-    build:                                                                                                                                                                 
-      context: .                                                                                                                                                           
-      dockerfile: Dockerfile.pilosa              
-    environment:
-      PILOSA_ADVERTISE: pilosa1:10101
-      PILOSA_ADVERTISE_GRPC: pilosa1:20101
-      PILOSA_CLUSTER_REPLICAS: 1
-      PILOSA_DATA_DIR: /data/pilosa1
-      PILOSA_ETCD_ADVERTISE_CLIENT_ADDRESS: http://pilosa1:10201
-      PILOSA_ETCD_ADVERTISE_PEER_ADDRESS: http://pilosa1:10301
-      PILOSA_ETCD_INITIAL_CLUSTER: pilosa0=http://pilosa0:10301,pilosa1=http://pilosa1:10301,pilosa2=http://pilosa2:10301
-      PILOSA_ETCD_LISTEN_CLIENT_ADDRESS: http://0.0.0.0:10201
-      PILOSA_ETCD_LISTEN_PEER_ADDRESS: http://0.0.0.0:10301
-      PILOSA_NAME: pilosa1
-      PILOSA_STORAGE_BACKEND: ${PILOSA_STORAGE_BACKEND:-rbf}
-    volumes:
-    - data:/data
-  pilosa2:
-    image: build/pilosa                                                                                                                                                     
-    build:                                                                                                                                                                 
-      context: .                                                                                                                                                           
-      dockerfile: Dockerfile.pilosa              
-    environment:
-      PILOSA_ADVERTISE: pilosa2:10101
-      PILOSA_ADVERTISE_GRPC: pilosa2:20101
-      PILOSA_CLUSTER_REPLICAS: 1
-      PILOSA_DATA_DIR: /data/pilosa2
-      PILOSA_ETCD_ADVERTISE_CLIENT_ADDRESS: http://pilosa2:10201
-      PILOSA_ETCD_ADVERTISE_PEER_ADDRESS: http://pilosa2:10301
-      PILOSA_ETCD_INITIAL_CLUSTER: pilosa0=http://pilosa0:10301,pilosa1=http://pilosa1:10301,pilosa2=http://pilosa2:10301
-      PILOSA_ETCD_LISTEN_CLIENT_ADDRESS: http://0.0.0.0:10201
-      PILOSA_ETCD_LISTEN_PEER_ADDRESS: http://0.0.0.0:10301
-      PILOSA_NAME: pilosa2
-      PILOSA_STORAGE_BACKEND: ${PILOSA_STORAGE_BACKEND:-rbf}
-    volumes:
-    - data:/data
-  pilosax:
-    image: build/pilosa                                                                                                                                                     
-    build:                                                                                                                                                                 
-      context: .                                                                                                                                                           
-      dockerfile: Dockerfile.pilosa              
-    environment:
-      PILOSA_ADVERTISE: pilosax:10101
-      PILOSA_ADVERTISE_GRPC: pilosax:20101
-      PILOSA_CLUSTER_REPLICAS: 1
-      PILOSA_DATA_DIR: /data/pilosax
-      PILOSA_ETCD_ADVERTISE_CLIENT_ADDRESS: http://pilosax:10201
-      PILOSA_ETCD_ADVERTISE_PEER_ADDRESS: http://pilosax:10301
-      PILOSA_ETCD_INITIAL_CLUSTER: pilosax=http://pilosax:10301
-      PILOSA_ETCD_LISTEN_CLIENT_ADDRESS: http://0.0.0.0:10201
-      PILOSA_ETCD_LISTEN_PEER_ADDRESS: http://0.0.0.0:10301
-      PILOSA_NAME: pilosax
-      PILOSA_STORAGE_BACKEND: ${PILOSA_STORAGE_BACKEND:-rbf}
-    volumes:
-    - data:/data
-  client1:
-    image: tgruben/bash
-    build:
-      context: .
-      dockerfile: Dockerfile.runner
-    environment:
-        - GO111MODULE=on
-    volumes:
-        - /var/run/docker.sock:/var/run/docker.sock
-    depends_on: 
-      - pilosa0
-    
-volumes:
-    data:
--- a/executor.go
+++ b/executor.go
@ -51,6 +51,9 @@ type executor struct {
 	Node    *topology.Node
 	Cluster *cluster

+	// how many jobs the work queue has seen
+	workCounter uint64
+
 	// Client used for remote requests.
 	client InternalQueryClient

@ -61,6 +64,7 @@ type executor struct {
 	workMu         sync.RWMutex
 	workersWG      sync.WaitGroup
 	workerPoolSize int
+	currentWorkers int64
 	work           chan job

 	// Maximum per-request memory usage (Extract() only)
@ -128,15 +132,61 @@ func newExecutor(opts ...executorOption) *executor {
 	e.work = make(chan job, e.workerPoolSize)
 	_ = testhook.Opened(NewAuditor(), e, nil)
 	for i := 0; i < e.workerPoolSize; i++ {
-		e.workersWG.Add(1)
-		go func() {
-			defer e.workersWG.Done()
-			worker(e.work)
-		}()
+		e.addWorker()
 	}
+	go func() {
+		// background task: every so often, check to see whether we have
+		// work in the queue but none has been taken for a while. if so, we
+		// need more workers.
+		prev := atomic.LoadUint64(&e.workCounter)
+		periodic := time.NewTicker(50 * time.Millisecond)
+		defer periodic.Stop()
+		running := true
+		idle := 0
+		for running {
+			<-periodic.C
+			func() {
+				e.workMu.RLock()
+				defer e.workMu.RUnlock()
+				if e.shutdown {
+					running = false
+					return
+				}
+				if len(e.work) == 0 {
+					idle++
+					if idle > 10 && atomic.LoadInt64(&e.currentWorkers) > int64(e.workerPoolSize*2) {
+						select {
+						case e.work <- job{idleHands: true}:
+							// we closed an excess worker
+						default:
+							// somehow between our test above and now the work
+							// queue FILLED UP and we stoically accept this
+						}
+						idle = 0
+					}
+					return
+				}
+				next := atomic.LoadUint64(&e.workCounter)
+				if next == prev {
+					e.addWorker()
+				}
+				prev = next
+			}()
+		}
+	}()
 	return e
 }

+func (e *executor) addWorker() {
+	e.workersWG.Add(1)
+	atomic.AddInt64(&e.currentWorkers, 1)
+	go func() {
+		defer e.workersWG.Done()
+		e.worker(e.work)
+		atomic.AddInt64(&e.currentWorkers, -1)
+	}()
+}
+
 func (e *executor) Close() error {
 	e.workMu.Lock()
 	defer e.workMu.Unlock()
@ -4493,7 +4543,11 @@ func (e *executor) executeRowShard(ctx context.Context, qcx *Qcx, index string,
 			return nil, err
 		}
 		defer finisher(&err0)
-		return frag.row(tx, rowID)
+		row, err := frag.row(tx, rowID)
+		if qcx.write && err == nil {
+			row = row.Clone()
+		}
+		return row, err
 	}

 	// If no quantum exists then return an empty bitmap.
@ -4532,15 +4586,21 @@ func (e *executor) executeRowShard(ctx context.Context, qcx *Qcx, index string,
 	if len(rows) == 0 {
 		return &Row{}, nil
 	} else if len(rows) == 1 {
+		if qcx.write {
+			return rows[0].Clone(), nil
+		}
 		return rows[0], nil
 	}
 	row := rows[0].Union(rows[1:]...)
+	if qcx.write {
+		row = row.Clone()
+	}
 	return row, nil

 }

 // executeRowBSIGroupShard executes a range(bsiGroup) call for a local shard.
-func (e *executor) executeRowBSIGroupShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (_ *Row, err0 error) {
+func (e *executor) executeRowBSIGroupShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (cloneable *Row, err0 error) {
 	span, _ := tracing.StartSpanFromContext(ctx, "Executor.executeRowBSIGroupShard")
 	defer span.Finish()

@ -4572,6 +4632,11 @@ func (e *executor) executeRowBSIGroupShard(ctx context.Context, qcx *Qcx, index
 		return nil, err
 	}
 	defer finisher(&err0)
+	defer func() {
+		if qcx.write && cloneable != nil {
+			cloneable = cloneable.Clone()
+		}
+	}()

 	// EQ null           _exists - frag.NotNull()
 	// NEQ null          frag.NotNull()
@ -4822,6 +4887,9 @@ func (e *executor) executeNotShard(ctx context.Context, qcx *Qcx, index string,
 		if existenceRow, err = existenceFrag.row(tx, 0); err != nil {
 			return nil, err
 		}
+		if qcx.write {
+			existenceRow = existenceRow.Clone()
+		}
 	}
 	// the finishers returned by a write tx, which we might be in if there's
 	// a higher-level write in this call OR ANY OTHER CALL, are safe to
@ -5915,10 +5983,15 @@ type job struct {
 	ctx             context.Context
 	memoryAvailable *int64 // shared, atomic value
 	resultChan      chan mapResponse
+	idleHands       bool
 }

-func worker(work chan job) {
+func (e *executor) worker(work chan job) {
 	for j := range work {
+		atomic.AddUint64(&e.workCounter, 1)
+		if j.idleHands {
+			return
+		}
 		// Skip out early if the context is done, but still send
 		// an ack so mapperLocal can be sure we aren't about to
 		// work on something it sent us.
--- a/executor_test.go
+++ b/executor_test.go
@ -27,6 +27,7 @@ import (
 	"github.com/google/go-cmp/cmp/cmpopts"
 	pilosa "github.com/molecula/featurebase/v2"
 	"github.com/molecula/featurebase/v2/boltdb"
+	"github.com/molecula/featurebase/v2/ctl"
 	"github.com/molecula/featurebase/v2/disco"
 	"github.com/molecula/featurebase/v2/http"
 	"github.com/molecula/featurebase/v2/pql"
@ -6748,7 +6749,7 @@ func variousQueriesCountDistinctTimestamp(t *testing.T, c *test.Cluster) {
 	field := "ts"

 	// create an index and timestamp field
-	c.CreateField(t, index, pilosa.IndexOptions{}, field, pilosa.OptFieldTypeTimestamp(time.Unix(0, 0), "s"))
+	c.CreateField(t, index, pilosa.IndexOptions{TrackExistence: true}, field, pilosa.OptFieldTypeTimestamp(time.Unix(0, 0), "s"))

 	// add some data
 	data := []string{"2010-01-02T12:32:00Z", "2010-04-20T12:32:00Z", "2011-04-20T12:32:00Z"}
@ -7015,19 +7016,90 @@ func TestMissingKeyRegression(t *testing.T) {
 // (single and multi-node clusters, different endpoints for the
 // queries (HTTP, GRPC, Postgres), etc.).
 func TestVariousQueries(t *testing.T) {
-	for _, clusterSize := range []int{1, 3, 4, 7} {
+	for _, clusterSize := range []int{1, 3, 5} {
 		clusterSize := clusterSize
 		t.Run(fmt.Sprintf("%d-node", clusterSize), func(t *testing.T) {
 			c := test.MustRunCluster(t, clusterSize)
 			defer c.Close()
+
+			// put a variety of data into the cluster
+			populateTestData(t, c)
+			backupTest(t, c, usersIndex)
+
 			variousQueries(t, c)
 			variousQueriesOnTimeFields(t, c)
 			variousQueriesOnPercentiles(t, c)
 			variousQueriesCountDistinctTimestamp(t, c)
+			backupTest(t, c, "") // test backup/restore of all indexes
 		})
 	}
 }

+func backupTest(t *testing.T, c *test.Cluster, index string) {
+	// should this really be in executor? No. But all these
+	// integration-y query tests probably shouldn't be either. My goal
+	// putting this here is to take advantage of already-existing
+	// clusters and data.
+
+	sum := chkSumCluster(t, c)
+
+	backupDir := backupCluster(t, c, index)
+
+	cnew := test.MustRunCluster(t, 3) // this way we test 1->3 3->3 5->3
+	defer cnew.Close()
+
+	restoreCluster(t, backupDir, cnew)
+
+	sumNew := chkSumCluster(t, cnew)
+
+	if sum != sumNew {
+		t.Fatalf("old/new checksum mismatch, old:\n%s\nnew:\n%s", sum, sumNew)
+	}
+}
+
+func chkSumCluster(t *testing.T, c *test.Cluster) string {
+	buf := &bytes.Buffer{}
+
+	chkSum := ctl.NewChkSumCommand(nil, buf, buf)
+	chkSum.Host = c.Nodes[len(c.Nodes)-1].URL()
+	if err := chkSum.Run(context.Background()); err != nil {
+		t.Fatalf("running checksum: %v", err)
+	}
+
+	return buf.String()
+}
+
+func backupCluster(t *testing.T, c *test.Cluster, index string) (backupDir string) {
+	td, err := testhook.TempDir(t, "backupTest")
+	if err != nil {
+		t.Fatalf("can't even get a temp dir, what a ripoff: %v", err)
+	}
+	td = td + "/backupTest"
+
+	buf := &bytes.Buffer{}
+	backupCommand := ctl.NewBackupCommand(nil, buf, buf)
+	backupCommand.Host = c.Nodes[len(c.Nodes)-1].URL() // don't pick node 0 so we don't always get primary (better code coverage)
+	backupCommand.Index = index
+	backupCommand.OutputDir = td
+
+	if err := backupCommand.Run(context.Background()); err != nil {
+		t.Log(buf.String())
+		t.Fatalf("running backup: %v", err)
+	}
+	return td
+}
+
+func restoreCluster(t *testing.T, backupDir string, c *test.Cluster) {
+	buf := &bytes.Buffer{}
+
+	restore := ctl.NewRestoreCommand(nil, buf, buf)
+	restore.Host = c.Nodes[len(c.Nodes)-1].URL()
+	restore.Path = backupDir
+	if err := restore.Run(context.Background()); err != nil {
+		t.Fatalf("restoring: %v", err)
+	}
+}
+
 // tests for abbreviating time values in queries
 func variousQueriesOnPercentiles(t *testing.T, c *test.Cluster) {
 	// todo, make rand more random, 42 isnt the answer to everything
@ -7332,10 +7404,12 @@ func variousQueriesOnTimeFields(t *testing.T, c *test.Cluster) {
 	}
 }

-func variousQueries(t *testing.T, c *test.Cluster) {
+var usersIndex = "users"
+
+func populateTestData(t *testing.T, c *test.Cluster) {
 	// Create and populate "likenums" similar to "likes", but without keys on the field.
-	c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "likenums")
-	c.ImportIDKey(t, "users", "likenums", []test.KeyID{
+	c.CreateField(t, usersIndex, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "likenums")
+	c.ImportIDKey(t, usersIndex, "likenums", []test.KeyID{
 		{ID: 1, Key: "userA"},
 		{ID: 2, Key: "userB"},
 		{ID: 3, Key: "userC"},
@ -7353,8 +7427,8 @@ func variousQueries(t *testing.T, c *test.Cluster) {
 	})

 	// Create and populate "likes" field.
-	c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "likes", pilosa.OptFieldKeys())
-	c.ImportKeyKey(t, "users", "likes", [][2]string{
+	c.CreateField(t, usersIndex, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "likes", pilosa.OptFieldKeys())
+	c.ImportKeyKey(t, usersIndex, "likes", [][2]string{
 		{"molecula", "userA"},
 		{"pilosa", "userB"},
 		{"pangolin", "userC"},
@ -7370,8 +7444,8 @@ func variousQueries(t *testing.T, c *test.Cluster) {
 	})

 	// Create and populate "dinner" field.
-	c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "dinner", pilosa.OptFieldKeys())
-	c.ImportKeyKey(t, "users", "dinner", [][2]string{
+	c.CreateField(t, usersIndex, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "dinner", pilosa.OptFieldKeys())
+	c.ImportKeyKey(t, usersIndex, "dinner", [][2]string{
 		{"leftovers", "userB"},
 		{"pizza", "userA"},
 		{"pizza", "userB"},
@ -7381,11 +7455,11 @@ func variousQueries(t *testing.T, c *test.Cluster) {
 	})

 	// Create and populate "places_visited" time field.
-	c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "places_visited", pilosa.OptFieldKeys(), pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YM")))
+	c.CreateField(t, usersIndex, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "places_visited", pilosa.OptFieldKeys(), pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YM")))
 	ts2019Jan01 := int64(1546300800) * 1e+9 // 2019 January 1st 0:00:00
 	ts2019Aug01 := int64(1564617600) * 1e+9 // 2019 August  1st 0:00:00
 	ts2020Jan01 := int64(1577836800) * 1e+9 // 2020 January 1st 0:00:00
-	c.ImportTimeQuantumKey(t, "users", "places_visited", []test.TimeQuantumKey{
+	c.ImportTimeQuantumKey(t, usersIndex, "places_visited", []test.TimeQuantumKey{
 		// 2019 January: nairobi, paris, austin, toronto
 		{RowKey: "nairobi", ColKey: "userB", Ts: ts2019Jan01},
 		{RowKey: "paris", ColKey: "userC", Ts: ts2019Jan01},
@ -7405,8 +7479,8 @@ func variousQueries(t *testing.T, c *test.Cluster) {
 	})

 	// Create and populate "affinity" int field with negative, positive, zero and null values.
-	c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "affinity", pilosa.OptFieldTypeInt(-1000, 1000))
-	c.ImportIntKey(t, "users", "affinity", []test.IntKey{
+	c.CreateField(t, usersIndex, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "affinity", pilosa.OptFieldTypeInt(-1000, 1000))
+	c.ImportIntKey(t, usersIndex, "affinity", []test.IntKey{
 		{Val: 10, Key: "userA"},
 		{Val: -10, Key: "userB"},
 		{Val: 5, Key: "userC"},
@ -7415,8 +7489,8 @@ func variousQueries(t *testing.T, c *test.Cluster) {
 	})

 	// Create and populate "net_worth" int field with positive values.
-	c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "net_worth", pilosa.OptFieldTypeInt(-100000000, 100000000))
-	c.ImportIntKey(t, "users", "net_worth", []test.IntKey{
+	c.CreateField(t, usersIndex, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "net_worth", pilosa.OptFieldTypeInt(-100000000, 100000000))
+	c.ImportIntKey(t, usersIndex, "net_worth", []test.IntKey{
 		{Val: 1, Key: "userA"},
 		{Val: 10, Key: "userB"},
 		{Val: 100, Key: "userC"},
@ -7425,8 +7499,8 @@ func variousQueries(t *testing.T, c *test.Cluster) {
 		{Val: 100000, Key: "userF"},
 	})

-	c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "zip_code", pilosa.OptFieldTypeInt(0, 100000))
-	c.ImportIntKey(t, "users", "zip_code", []test.IntKey{
+	c.CreateField(t, usersIndex, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "zip_code", pilosa.OptFieldTypeInt(0, 100000))
+	c.ImportIntKey(t, usersIndex, "zip_code", []test.IntKey{
 		{Val: 78739, Key: "userA"},
 		{Val: 78739, Key: "userB"},
 		{Val: 19707, Key: "userC"},
@ -7434,7 +7508,12 @@ func variousQueries(t *testing.T, c *test.Cluster) {
 		{Val: 86753, Key: "userE"},
 		{Val: 78739, Key: "userG"},
 	})
+}

+func variousQueries(t *testing.T, c *test.Cluster) {
+	// NOTE: this relies on populateTestData being called first
+
+	// define and run a bunch of tests
 	tests := []struct {
 		query       string
 		qrVerifier  func(t *testing.T, resp pilosa.QueryResponse)
@ -7781,8 +7860,8 @@ leftovers,1

 	for i, tst := range tests {
 		t.Run(fmt.Sprintf("%d-%s", i, tst.query), func(t *testing.T) {
-			resp := c.Query(t, "users", tst.query)
-			tr := c.QueryGRPC(t, "users", tst.query)
+			resp := c.Query(t, usersIndex, tst.query)
+			tr := c.QueryGRPC(t, usersIndex, tst.query)
 			if tst.qrVerifier != nil {
 				tst.qrVerifier(t, resp)
 			}
@ -8352,7 +8431,7 @@ func MinMaxTimestampNodeTester(t *testing.T, numNodes int) {
 	defer c.Close()

 	// create an index and timestamp field
-	c.CreateField(t, index, pilosa.IndexOptions{}, field, pilosa.OptFieldTypeTimestamp(time.Unix(0, 0), "s"))
+	c.CreateField(t, index, pilosa.IndexOptions{TrackExistence: true}, field, pilosa.OptFieldTypeTimestamp(time.Unix(0, 0), "s"))

 	// add some data
 	expected := "2010-01-02T12:32:00Z"
--- a/field.go
+++ b/field.go
@ -705,8 +705,11 @@ func (f *Field) cacheBitDepth(bd uint64) error {
 	f.mu.Lock()
 	defer f.mu.Unlock()

-	f.options.BitDepth = bd
-	if bsig != nil {
+	if f.options.BitDepth < bd {
+		f.options.BitDepth = bd
+	}
+
+	if bsig != nil && bsig.BitDepth < bd {
 		bsig.BitDepth = bd
 	}

@ -1884,7 +1887,7 @@ func applyDefaultOptions(o *FieldOptions) FieldOptions {
 // are included.
 func (o *FieldOptions) MarshalJSON() ([]byte, error) {
 	switch o.Type {
-	case FieldTypeSet:
+	case FieldTypeSet, "":
 		return json.Marshal(struct {
 			Type      string `json:"type"`
 			CacheType string `json:"cacheType"`
@ -1975,7 +1978,7 @@ func (o *FieldOptions) MarshalJSON() ([]byte, error) {
 			o.Type,
 		})
 	}
-	return nil, errors.New("invalid field type")
+	return nil, errors.Errorf("invalid field type: '%s'", o.Type)
 }

 // MinTimestamp returns the minimum value for a timestamp field.
--- a/fragment.go
+++ b/fragment.go
@ -218,6 +218,8 @@ func newFragment(holder *Holder, spec fragSpec, shard uint64, flags byte) *fragm
 func (f *fragment) cachePath() string { return f.path() + cacheExt }

 func (f *fragment) bitDepth() (uint64, error) {
+	f.mu.RLock()
+	defer f.mu.RUnlock()
 	tx, err := f.holder.BeginTx(false, f.idx, f.shard)
 	if err != nil {
 		return 0, errors.Wrapf(err, "beginning new tx(false, %s, %d)", f.index(), f.shard)
@ -593,8 +595,8 @@ func (f *fragment) mutexCheck(tx Tx, details bool, limit int) (map[uint64][]uint

 // row returns a row by ID.
 func (f *fragment) row(tx Tx, rowID uint64) (*Row, error) {
-	f.mu.Lock()
-	defer f.mu.Unlock()
+	f.mu.RLock()
+	defer f.mu.RUnlock()
 	return f.unprotectedRow(tx, rowID)
 }

@ -937,9 +939,12 @@ func (f *fragment) unprotectedClearRow(tx Tx, rowID uint64) (changed bool, err e
 	return changed, nil
 }

-// unprotectedClearBlock clears all rows for a given block.
+// clearBlock clears all rows for a given block.
 // This updates both the on-disk storage and the in-cache bitmap.
-func (f *fragment) unprotectedClearBlock(tx Tx, block int) (changed bool, err error) {
+func (f *fragment) clearBlock(tx Tx, block int) (changed bool, err error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
 	firstRow := uint64(block * HashBlockSize)
 	var wp *io.Writer
 	if f.storage != nil {
@ -2708,20 +2713,24 @@ func (f *fragment) importValue(tx Tx, columnIDs []uint64, values []int64, bitDep
 func (f *fragment) importRoaring(ctx context.Context, tx Tx, data []byte, clear bool) error {
 	span, ctx := tracing.StartSpanFromContext(ctx, "fragment.importRoaring")
 	defer span.Finish()
-	span, ctx = tracing.StartSpanFromContext(ctx, "importRoaring.AcquireFragmentLock")
-	f.mu.Lock()
-	defer f.mu.Unlock()
-	span.Finish()

-	return f.unprotectedImportRoaring(ctx, tx, data, clear)
+	rowSet, updateCache, err := f.doImportRoaring(ctx, tx, data, clear)
+	if err != nil {
+		return errors.Wrap(err, "doImportRoaring")
+	}
+	if updateCache {
+		return f.updateCachePostImport(ctx, rowSet)
+	}
+	return nil
 }

-func (f *fragment) unprotectedImportRoaring(ctx context.Context, tx Tx, data []byte, clear bool) error {
+func (f *fragment) doImportRoaring(ctx context.Context, tx Tx, data []byte, clear bool) (map[uint64]int, bool, error) {
+	f.mu.RLock()
+	defer f.mu.RUnlock()
 	rowSize := uint64(1 << shardVsContainerExponent)
 	span, ctx := tracing.StartSpanFromContext(ctx, "importRoaring.ImportRoaringBits")
+	defer span.Finish()

-	useRowCache := storage.RowCacheEnabled()
-	var changed int
 	var rowSet map[uint64]int
 	var wp *io.Writer
 	if f.storage != nil {
@ -2734,37 +2743,37 @@ func (f *fragment) unprotectedImportRoaring(ctx context.Context, tx Tx, data []b
 			return err
 		}

-		changed, rowSet, err = tx.ImportRoaringBits(f.index(), f.field(), f.view(), f.shard, rit, clear, true, rowSize)
+		_, rowSet, err = tx.ImportRoaringBits(f.index(), f.field(), f.view(), f.shard, rit, clear, true, rowSize)
 		return err
 	})

-	span.Finish()
 	if err != nil {
-		return err
+		return nil, false, err
 	}

 	updateCache := f.CacheType != CacheTypeNone
+	return rowSet, updateCache, err
+}
+
+func (f *fragment) updateCachePostImport(ctx context.Context, rowSet map[uint64]int) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
 	anyChanged := false

 	for rowID, changes := range rowSet {
 		if changes == 0 {
 			continue
 		}
-		if useRowCache && f.rowCache != nil {
-			f.rowCache.Add(rowID, nil)
-		}
-		if updateCache {
-			anyChanged = true
-			if changes < 0 {
-				absChanges := uint64(-1 * changes)
-				if absChanges <= f.cache.Get(rowID) {
-					f.cache.BulkAdd(rowID, f.cache.Get(rowID)-absChanges)
-				} else {
-					f.cache.BulkAdd(rowID, 0)
-				}
+		anyChanged = true
+		if changes < 0 {
+			absChanges := uint64(-1 * changes)
+			if absChanges <= f.cache.Get(rowID) {
+				f.cache.BulkAdd(rowID, f.cache.Get(rowID)-absChanges)
 			} else {
-				f.cache.BulkAdd(rowID, f.cache.Get(rowID)+uint64(changes))
+				f.cache.BulkAdd(rowID, 0)
 			}
+		} else {
+			f.cache.BulkAdd(rowID, f.cache.Get(rowID)+uint64(changes))
 		}
 	}
 	// we only set this if we need to update the cache
@ -2772,26 +2781,18 @@ func (f *fragment) unprotectedImportRoaring(ctx context.Context, tx Tx, data []b
 		f.cache.Invalidate()
 	}

-	span, _ = tracing.StartSpanFromContext(ctx, "importRoaring.incrementOpN")
-
-	f.incrementOpN(changed)
-
-	span.Finish()
 	return nil
 }

 // importRoaringOverwrite overwrites the specified block with the provided data.
 func (f *fragment) importRoaringOverwrite(ctx context.Context, tx Tx, data []byte, block int) error {
-	f.mu.Lock()
-	defer f.mu.Unlock()
-
 	// Clear the existing data from fragment block.
-	if _, err := f.unprotectedClearBlock(tx, block); err != nil {
+	if _, err := f.clearBlock(tx, block); err != nil {
 		return errors.Wrapf(err, "clearing block: %d", block)
 	}

 	// Union the new block data with the fragment data.
-	return f.unprotectedImportRoaring(ctx, tx, data, false)
+	return f.importRoaring(ctx, tx, data, false)
 }

 // incrementOpN increase the operation count by one.
--- a/go.mod
+++ b/go.mod
@ -57,7 +57,7 @@ require (
 	golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
 	google.golang.org/grpc v1.28.0
-	gopkg.in/yaml.v2 v2.3.0 // indirect
+	gopkg.in/yaml.v2 v2.3.0
 	modernc.org/mathutil v1.0.0
 	modernc.org/strutil v1.0.0
 	sigs.k8s.io/yaml v1.2.0 // indirect
--- a/http/handler.go
+++ b/http/handler.go
@ -2342,6 +2342,12 @@ func (h *Handler) handleGetTranslateData(w http.ResponseWriter, r *http.Request)

 	// Retrieve partition data from holder.
 	p, err := h.api.TranslateData(r.Context(), q.Get("index"), int(partition))
+	if redir, ok := err.(pilosa.RedirectError); ok {
+		newURL := *r.URL
+		newURL.Host = redir.HostPort
+		http.Redirect(w, r, newURL.String(), http.StatusSeeOther)
+		return
+	}
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusNotFound)
 		return
--- a/install/featurebase.conf
+++ b/install/featurebase.conf
@ -384,4 +384,5 @@ log-path = "/var/log/molecula/featurebase.log"
 #  logout-url = ""
 #  scopes = ["", ""]
 #  hash-key = ""
-#  block-key = ""
+#  block-key = ""
+#  permissions = ""
--- a/rbf/cfg/cfg.go
+++ b/rbf/cfg/cfg.go
@ -2,6 +2,7 @@
 package cfg

 import (
+	"github.com/molecula/featurebase/v2/logger"
 	"github.com/spf13/pflag"
 )

@ -35,6 +36,11 @@ type Config struct {
 	// CursorCacheSize is the number of copies of Cursor{} to keep in our
 	// readyCursorCh arena to avoid GC pressure.
 	CursorCacheSize int64 `toml:"cursor-cache-size"`
+
+	// Logger specifies a logger for asynchronous errors, such as
+	// background checkpoints. It cannot be set from toml. The default is
+	// to use stderr.
+	Logger logger.Logger `toml:"-"`
 }

 func NewDefaultConfig() *Config {
--- a/rbf/cursor.go
+++ b/rbf/cursor.go
@ -774,6 +774,25 @@ func (c *Cursor) deleteBranchCell(stackIndex int, key uint64) (err error) {
 	cells[len(cells)-1] = branchCell{}
 	cells = cells[:len(cells)-1]

+	// Branches are not allowed to have zero element so we must remove the page
+	// or, in the case of the root page, convert to a leaf page.
+	if len(cells) == 0 {
+		// If this is the root page, convert to leaf page.
+		if stackIndex == 0 {
+			var buf [PageSize]byte
+			writePageNo(buf[:], elem.pgno)
+			writeFlags(buf[:], PageTypeLeaf)
+			writeCellN(buf[:], len(cells))
+			return c.tx.writePage(buf[:])
+		}
+
+		// If this is a non-root page, free and remove from parent.
+		if err := c.tx.freePgno(elem.pgno); err != nil {
+			return err
+		}
+		return c.deleteBranchCell(stackIndex-1, oldPageKey)
+	}
+
 	// If the root only has one node, replace it with its child.
 	if stackIndex == 0 && len(cells) == 1 {
 		target, _, err := c.tx.readPage(cells[0].ChildPgno)
@ -802,6 +821,9 @@ func (c *Cursor) deleteBranchCell(stackIndex int, key uint64) (err error) {
 		writeBranchCell(buf[:], j, offset, cell)
 		offset += align8(branchCellSize)
 	}
+
+	assert(readCellN(buf[:]) > 0) // must have at least one cell
+
 	if err := c.tx.writePage(buf[:]); err != nil {
 		return err
 	}
--- a/rbf/cursor_test.go
+++ b/rbf/cursor_test.go
@ -973,8 +973,8 @@ func TestCursor_SplitBranchCells(t *testing.T) {
 	}
 	//
 	c, _ := tx.Cursor("x") //added just for dot code coverage
-	c.Dump("ignore for coverage")
-
+	c.Dump("test.dump")
+	os.Remove("test.dump")
 }

 func TestCursor_RemoveCells(t *testing.T) {
--- a/rbf/db.go
+++ b/rbf/db.go
@ -11,6 +11,7 @@ import (
 	"syscall"

 	"github.com/benbjohnson/immutable"
+	"github.com/molecula/featurebase/v2/logger"
 	rbfcfg "github.com/molecula/featurebase/v2/rbf/cfg"
 	"github.com/molecula/featurebase/v2/syswrap"
 )
@ -27,6 +28,16 @@ var cursorSyncPool = &sync.Pool{
 	},
 }

+// txWaiter is a representation of "i need to wait for txs to complete".
+// it is created with a function, and will run that function, with the db
+// lock held, at some point after every Tx that was open when it was created
+// has closed. WARNING: A txWaiter may hold db.rwmu.
+type txWaiter struct {
+	ready     chan struct{}
+	waitingOn map[*Tx]struct{}
+	callback  func()
+}
+
 // DB options like 	MaxSize, FsyncEnabled, DoAllocZero
 // can be set before calling DB.Open().
 type DB struct {
@ -38,15 +49,21 @@ type DB struct {
 	pageMap     *PageMap             // pgno-to-WALID mapping
 	txs         map[*Tx]struct{}     // active transactions
 	opened      bool                 // true if open
+	logger      logger.Logger        // for diagnostics from async things

-	wal      []byte   // wal mmap
-	walFile  *os.File // wal file descriptor
-	walPageN int      // wal page count
+	wal       []byte   // wal mmap
+	walFile   *os.File // wal file descriptor
+	walPageN  int      // wal page count
+	baseWALID int64    // WAL ID of first page

 	mu       sync.RWMutex // general mutex
 	rwmu     sync.Mutex   // mutex for restricting single writer
 	haltCond *sync.Cond   // condition for resuming txs after checkpoint

+	txWaiters []*txWaiter // things waiting for Txs to close
+
+	isDead error // this database died in an unrecoverable way, error out opens
+
 	// Path represents the path to the database file.
 	Path string
 }
@ -62,6 +79,11 @@ func NewDB(path string, cfg *rbfcfg.Config) *DB {
 		txs:     make(map[*Tx]struct{}),
 		pageMap: NewPageMap(),
 		Path:    path,
+		logger:  cfg.Logger,
+	}
+	if db.logger == nil {
+		// default to writing to stdout if not told otherwise
+		db.logger = logger.NewStandardLogger(os.Stderr)
 	}
 	db.haltCond = sync.NewCond(&db.mu)

@ -133,8 +155,12 @@ func (db *DB) Open() (err error) {
 	// Open write-ahead log & checkpoint to the end since no transactions are open.
 	if err := db.openWAL(); err != nil {
 		return fmt.Errorf("wal open: %w", err)
-	} else if err := db.checkpoint(); err != nil {
-		return fmt.Errorf("checkpoint: %w", err)
+	} else {
+		// checkpoint wants to hold the rwmu lock.
+		db.rwmu.Lock()
+		if err := db.checkpoint(); err != nil {
+			return fmt.Errorf("startup checkpoint: %w", err)
+		}
 	}

 	return nil
@ -158,10 +184,12 @@ func (db *DB) openWAL() (err error) {

 	// Determine the number of whole pages in the WAL.
 	var pageN int
+	var fileSize int64
 	if fi, err := db.walFile.Stat(); err != nil {
 		return fmt.Errorf("wal stat: %w", err)
 	} else {
-		pageN = int(fi.Size() / PageSize)
+		fileSize = fi.Size()
+		pageN = int(fileSize / PageSize)
 	}

 	// Read backwards through the WAL to find the last valid meta page.
@ -169,28 +197,96 @@ func (db *DB) openWAL() (err error) {
 		if page, err := db.readWALPageAt(pageN - 1); err != nil {
 			return err
 		} else if IsMetaPage(page) {
+			// We now face a challenge. Probably this is a meta page.
+			// But consider a sequence of pages written which gets
+			// interrupted right before the meta page is written.
+			// If the last page is a bitmap page, it could LOOK LIKE a meta
+			// page. So we have to check the page before it. If that page
+			// is a bitmap header, then actually this is a bitmap page, right?
+			// If that page doesn't exist, of course, we're fine, except
+			// for the philosophical question of why we wrote a meta page
+			// when no pages had changed.
+			if pageN > 1 {
+				if page, err = db.readWALPageAt(pageN - 2); err != nil {
+					return err
+				}
+				if IsBitmapHeader(page) {
+					// But wait!
+					// What if this *is* a meta page, and the page before it is
+					// actually a *bitmap page* that looks like a bitmap header? And
+					// so on.
+					//
+					// Rather than try to resolve this, in this insanely unlikely
+					// situation, we read from the beginning which allows us to
+					// always know what we're seeing, because every bitmap page
+					// comes *after* a bitmap header page, and thus, we know when
+					// we might be seeing one.
+					pageN, err = db.methodicalWALPageN(pageN)
+					if err != nil {
+						return err
+					}
+				}
+			}
 			break
 		}
 	}
-
-	// Truncate WAL to the last valid meta page.
-	if err := db.walFile.Truncate(int64(pageN * PageSize)); err != nil {
-		return fmt.Errorf("wal truncate: %w", err)
-	} else if _, err := db.walFile.Seek(int64(pageN*PageSize), io.SeekStart); err != nil {
+	if fileSize != int64(pageN*PageSize) {
+		if err := db.walFile.Truncate(int64(pageN * PageSize)); err != nil {
+			return fmt.Errorf("wal truncate: %w", err)
+		}
+	}
+	if _, err := db.walFile.Seek(int64(pageN*PageSize), io.SeekStart); err != nil {
 		return fmt.Errorf("wal seek: %w", err)
 	}
 	db.walPageN = pageN
+	db.baseWALID = readMetaWALID(db.data)

 	return nil
 }

-// checkpoint moves all WAL pages to the main DB file.
-// Must be called by a write transaction while under db.mu lock.
-func (db *DB) checkpoint() error {
+// methodicalWALPageN tries to determine the last meta page in a very reliable
+// but slow way. This handles the theoretical but hard to imagine creating
+// edge case where we have a bitmap page which happens to look like a meta
+// page, and the write got interrupted before the meta page got written.
+func (db *DB) methodicalWALPageN(pageN int) (lastMeta int, err error) {
+	for i := 0; i < pageN; i++ {
+		var page []byte
+		if page, err = db.readWALPageAt(i); err != nil {
+			return -1, err
+		}
+		switch {
+		case IsMetaPage(page):
+			lastMeta = i
+		case IsBitmapHeader(page):
+			// skip the bitmap page, which we can't usefully evaluate
+			i++
+		}
+	}
+	return lastMeta, nil
+}
+
+// Checkpoint performs a manual checkpoint. This is not necessary except for tests.
+func (db *DB) Checkpoint() error {
+	db.mu.Lock()
+	defer db.mu.Unlock()
+	db.rwmu.Lock()
+	return db.checkpoint()
+}
+
+// checkpoint moves all WAL pages to the main DB file. Must be called
+// while holding both db.mu and db.rwmu. Should release db.rwmu, but not
+// db.mu.
+func (db *DB) checkpoint() (err error) {
+	// if we don't spin off a possible async waiter, we should release the
+	// write lock, if we do, that will release it.
+	releaseLock := true
+	defer func() {
+		if releaseLock {
+			db.rwmu.Unlock()
+		}
+	}()
 	if !db.opened {
 		return nil
-	} else if len(db.txs) > 0 {
-		return nil // skip if transactions open
 	}

 	// Check if there are any WAL pages, if not do nothing as
@ -199,48 +295,112 @@ func (db *DB) checkpoint() error {
 	if db.walPageN == 0 {
 		return nil
 	}
-
-	for i := 0; i < db.walPageN; i++ {
-		page, err := db.readWALPageAt(i)
-		if err != nil {
-			return err
+	// wake up things waiting on haltCond when we're done, even if we fail.
+	// Otherwise, we deadlock with them all stuck waiting on that forever.
+	defer func() {
+		if err != nil && db.isDead == nil {
+			db.isDead = err
 		}
+		db.haltCond.Broadcast()
+	}()

-		// Determine page number. Meta pages are always on zero & bitmap
-		// headers specify the page number of the next page in the WAL.
-		// All other pages have their page number in the page data.
-		var pgno uint32
-		if IsBitmapHeader(page) {
-			pgno = readPageNo(page)
-			if page, err = db.readWALPageAt(i + 1); err != nil {
-				return err
+	// Copy the pages from the WAL back to the database outside of the lock.
+	if err := func() error {
+		db.mu.Unlock() // This is intentionally reversed so run w/o lock
+		defer db.mu.Lock()
+
+		var page []byte
+		// We might have either a *PageMap or just the file. If we have the file,
+		// building the PageMap is fairly expensive because it's fancy and immutable.
+		// If we have the PageMap *or* some other map, that's two different things
+		// to iterate. If we have the PageMap, building a map from it is relatively
+		// cheap, so we'll do it that way.
+		pages := make(map[uint32]int)
+
+		if db.pageMap.size == 0 {
+			// you'd think we're done, but actually this PROBABLY means that
+			// this is initial startup, and we haven't read the file yet. We scan
+			// the file for pages, because it turns out most of them probably
+			// got overwritten.
+			for i := 0; i < db.walPageN; i++ {
+				page, err = db.readWALPageAt(i)
+				if err != nil {
+					return fmt.Errorf("reading WAL page %d: %w", i, err)
+				}
+
+				// Determine page number. Meta pages are always on zero & bitmap
+				// headers specify the page number of the next page in the WAL.
+				// All other pages have their page number in the page data.
+				var pgno uint32
+				if IsBitmapHeader(page) {
+					pgno = readPageNo(page)
+					if i+1 < db.walPageN {
+						if page, err = db.readWALPageAt(i + 1); err != nil {
+							return err
+						}
+					} else {
+						return fmt.Errorf("last page of WAL file (%d) is bitmap header", i)
+					}
+					i++ // bitmaps in WAL are two pages
+				} else if !IsMetaPage(page) {
+					pgno = readPageNo(page)
+				}
+				// record where in the file we have this page
+				pages[pgno] = i
+			}
+		} else {
+			itr := db.pageMap.Iterator()
+			itr.First()
+			for k, v, ok := itr.Next(); ok; k, v, ok = itr.Next() {
+				pages[k] = int(v - db.baseWALID - 1)
 			}
-			i++ // bitmaps in WAL are two pages
-		} else if !IsMetaPage(page) {
-			pgno = readPageNo(page)
 		}

-		// Write data to the data file.
-		if err := db.writeDBPage(pgno, page); err != nil {
-			return err
+		// fmt.Printf("checkpoint: walPageN %d, PageMap size %d\n", db.walPageN, db.pageMap.size)
+		for pgno, walID := range pages {
+			page, err = db.readWALPageAt(walID)
+			if err != nil {
+				return fmt.Errorf("reading page %d [page number %d]: %v", walID, pgno, err)
+			}
+
+			// Write data to the data file.
+			if err = db.writeDBPage(pgno, page); err != nil {
+				return fmt.Errorf("writing page %d: %v", pgno, err)
+			}
 		}
+
+		// Ensure database file is synced and then truncate the WAL file.
+		if err = db.fsync(db.file); err != nil {
+			return fmt.Errorf("db file sync: %w", err)
+		}
+
+		return nil
+	}(); err != nil {
+		return err
 	}

-	// Ensure database file is synced and then truncate the WAL file.
-	if err := db.fsync(db.file); err != nil {
-		return fmt.Errorf("db file sync: %w", err)
-	} else if err := db.walFile.Truncate(0); err != nil {
-		return fmt.Errorf("truncate wal file: %w", err)
-	} else if err := db.fsync(db.walFile); err != nil {
-		return fmt.Errorf("wal file sync: %w", err)
-	} else if _, err := db.walFile.Seek(0, io.SeekStart); err != nil {
-		return fmt.Errorf("seek wal file: %w", err)
-	}
+	// now we've updated the file. There are existing transactions that are still
+	// using the WAL, though. So we wait for them to terminate before we unlock
+	// the rwmu and update the metadata about the WAL.
+	releaseLock = false
 	db.walPageN = 0
 	db.pageMap = NewPageMap()

-	// Notify halted transactions that the WAL has been checkpointed.
-	db.haltCond.Broadcast()
+	db.afterCurrentTx(func() {
+		defer db.rwmu.Unlock()
+		db.baseWALID = readMetaWALID(db.data)
+		db.mu.Unlock()
+		defer db.mu.Lock()
+
+		if err = db.walFile.Truncate(0); err != nil {
+			db.logger.Errorf("truncate wal file: %w", err)
+		} else if err = db.fsync(db.walFile); err != nil {
+			db.logger.Errorf("wal file sync: %w", err)
+		} else if _, err = db.walFile.Seek(0, io.SeekStart); err != nil {
+			db.logger.Errorf("seek wal file: %w", err)
+		}
+
+	})

 	return nil
 }
@ -444,21 +604,31 @@ func (db *DB) Begin(writable bool) (_ *Tx, err error) {
 	}

 	db.mu.Lock()
-	// note: We cannot defer db.mu.Unlock() here because
-	// we call tx.Rollback() before if db.readMetaPage
-	// returns an error, and thus we will deadlock against
-	// ourselves when the Rollback tries to acquire the db.mu.
-	// This is why db.mu.Unlock() is done manually below.
+	defer db.mu.Unlock()

 	if !db.opened {
 		cleanup()
-		db.mu.Unlock()
 		return nil, ErrClosed
 	}
+	if db.isDead != nil {
+		err := db.isDead
+		cleanup()
+		return nil, err
+	}

-	// Wait for WAL size to be below threshold.
-	for int64(db.walPageN*PageSize) > db.cfg.MaxWALCheckpointSize {
-		db.haltCond.Wait()
+	// Wait for WAL size to be below threshold, if we're going to write.
+	// Reads don't care.
+	if writable {
+		for int64(db.walPageN*PageSize) > db.cfg.MaxWALCheckpointSize {
+			if db.isDead != nil {
+				err := db.isDead
+				cleanup()
+				return nil, err
+			}
+			// This implicitly releases db.mu.Lock and comes back with it
+			// held again.
+			db.haltCond.Wait()
+		}
 	}

 	tx := &Tx{
@ -470,6 +640,11 @@ func (db *DB) Begin(writable bool) (_ *Tx, err error) {

 		DeleteEmptyContainer: true,
 	}
+	defer func() {
+		if err != nil {
+			tx.rollback(true)
+		}
+	}()

 	if writable {
 		tx.dirtyPages = make(map[uint32][]byte)
@ -480,10 +655,6 @@ func (db *DB) Begin(writable bool) (_ *Tx, err error) {
 	// This page is only written at the end of a dirty transaction.
 	page, err := db.readMetaPage()
 	if err != nil {
-		// we will deadlock in tx.Rollback()
-		// on db.mu.Lock unless we manually db.mu.Unlock first.
-		db.mu.Unlock()
-		tx.Rollback()
 		return nil, err
 	}
 	copy(tx.meta[:], page)
@ -499,36 +670,102 @@ func (db *DB) Begin(writable bool) (_ *Tx, err error) {
 	// this avoids recomputing the cache if there are no write txs for a while.
 	if db.rootRecords == nil {
 		if db.rootRecords, err = tx.RootRecords(); err != nil {
-			db.mu.Unlock()
-			tx.Rollback()
 			return nil, err
 		}
 	}

-	db.mu.Unlock()
 	return tx, nil
 }

-// removeTx removes an active transaction from the database.
-func (db *DB) removeTx(tx *Tx) error {
-	// Release writer lock if tx is writable.
-	if tx.writable {
-		tx.db.rwmu.Unlock()
+// afterCurrentTx produces runs the provided callback, with the db lock
+// held, after all current Tx terminate. It should be called with the db
+// lock held.
+func (db *DB) afterCurrentTx(callback func()) {
+	if len(db.txs) == 0 {
+		callback()
+		return
 	}
+	txw := &txWaiter{}
+	txw.ready = make(chan struct{})
+	txw.callback = callback
+	txw.waitingOn = make(map[*Tx]struct{}, len(db.txs))
+	for k := range db.txs {
+		txw.waitingOn[k] = struct{}{}
+	}
+	db.txWaiters = append(db.txWaiters, txw)
+	go func() {
+		<-txw.ready
+		// fmt.Printf("afterCurrentTx: locking db\n")
+		db.mu.Lock()
+		defer db.mu.Unlock()
+		// fmt.Printf("afterCurrentTx: running callback\n")
+		txw.callback()
+	}()
+	return
+}

+// removeTx removes an active transaction from the database. it obtains
+// the db lock, and currently drops it, but will later possibly be leaving
+// it retained by an asynchronous op that wants to happen before we start
+// running new tx.
+func (db *DB) removeTx(tx *Tx) error {
+	// We might want to trigger a checkpoint. Only for writable
+	// transactions, and only when either there's nothing else open or we
+	// really need to.
+	checkpoint := false
+	if tx.writable {
+		walSize := db.walSize()
+		if walSize > db.cfg.MinWALCheckpointSize {
+			// Might be a good time for a checkpoint. We'll do a checkpoint
+			// if we're the only transaction, or if we have to.
+			if len(db.txs) == 1 || walSize > db.cfg.MaxWALCheckpointSize {
+				checkpoint = true
+			}
+		}
+		// During checkpointing, we'll be preventing writes, but allowing reads.
+		if !checkpoint {
+			tx.db.rwmu.Unlock()
+		}
+	}
+	// remove ourselves from the list of transactions the db is keeping.
 	delete(tx.db.txs, tx)
+	for i := 0; i < len(tx.db.txWaiters); i++ {
+		txw := tx.db.txWaiters[i]
+		// in practice this probably never matters, but theoretically the
+		// goroutine that's waiting on the condition variable may
+		// not have performed its first test on len(txw.waitingOn) yet.
+		delete(txw.waitingOn, tx)
+		// let it know we're done. we've still got db.mu.lock, so it won't
+		// happen just yet, but it'll be able to continue.
+		if len(txw.waitingOn) == 0 {
+			// remove us from the db's list
+			copy(db.txWaiters[i:], db.txWaiters[i+1:])
+			db.txWaiters = db.txWaiters[:len(db.txWaiters)-1]
+			close(txw.ready)
+			// decrement i so we don't skip an entry we just copied in to [i]
+			i--
+		}
+	}

 	// Disassociate from db.
 	tx.db = nil

-	// Write pages from WAL to DB.
-	// TODO(bbj): Move this to an async goroutine.
-	if len(db.txs) == 0 && db.walSize() > db.cfg.MinWALCheckpointSize {
-		if err := db.checkpoint(); err != nil {
-			return fmt.Errorf("checkpoint: %w", err)
-		}
+	if checkpoint {
+		// We need to run a checkpoint. This can be semi-asynchronous.
+		// It needs to wait until every existing transaction has finished,
+		// because every existing transaction could want to look up pages
+		// which are in the database before our operations, but which should
+		// now be in the WAL. We want them to use the WAL instead.
+		// fmt.Printf("possibly-async checkpoint...\n")
+		db.afterCurrentTx(func() {
+			// We still hold db.rwmu here. checkpoint unlocks it when it's
+			// ready.
+			// fmt.Printf("checkpoint starting\n")
+			if err := db.checkpoint(); err != nil {
+				db.logger.Errorf("async checkpoint: %v", err)
+			}
+		})
 	}
-
 	return nil
 }

@ -554,14 +791,9 @@ func (db *DB) readDBPage(pgno uint32) ([]byte, error) {
 	return db.data[offset : offset+PageSize], nil
 }

-// baseWALID returns the WAL ID stored in the database file meta page.
-func (db *DB) baseWALID() int64 {
-	return readMetaWALID(db.data)
-}
-
 // readWALPageByID reads a WAL page by WAL ID.
 func (db *DB) readWALPageByID(id int64) ([]byte, error) {
-	return db.readWALPageAt(int(id - db.baseWALID() - 1))
+	return db.readWALPageAt(int(id - db.baseWALID - 1))
 }

 // readWALPageAt reads the i-th page in the WAL file.
--- a/rbf/db_test.go
+++ b/rbf/db_test.go
@ -13,6 +13,7 @@ import (

 	_ "net/http/pprof"

+	"github.com/felixge/fgprof"
 	"github.com/molecula/featurebase/v2/rbf"
 	rbfcfg "github.com/molecula/featurebase/v2/rbf/cfg"
 	"golang.org/x/sync/errgroup"
@ -290,7 +291,8 @@ func TestDB_MultiTx(t *testing.T) {

 					time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)

-					for i := 0; i < rand.Intn(1000); i++ {
+					n := rand.Intn(500) + 500
+					for i := 0; i < n; i++ {
 						v := rand.Intn(1 << 20)
 						if _, err := tx.Contains("x", uint64(v)); err != nil {
 							return err
@ -315,7 +317,8 @@ func TestDB_MultiTx(t *testing.T) {
 			}
 			defer tx.Rollback()

-			for j := 0; j < rand.Intn(100); j++ {
+			n := rand.Intn(90) + 10
+			for j := 0; j < n; j++ {
 				v := rand.Intn(1 << 20)
 				if _, err := tx.Add("x", uint64(v)); err != nil {
 					t.Fatal(err)
@ -336,6 +339,119 @@ func TestDB_MultiTx(t *testing.T) {
 	}
 }

+// premake pool of random values
+const randPool = (1 << 18)
+
+// benchmarkOneCheckpoint
+func benchmarkOneCheckpoint(b *testing.B, randInts []int) {
+	cfg := rbfcfg.NewDefaultConfig()
+	// extremely low to force checkpointing
+	cfg.MinWALCheckpointSize = rbf.PageSize * 16
+	cfg.MaxWALCheckpointSize = rbf.PageSize * 64
+	var _ rbfcfg.Config
+	db := MustOpenDB(b, cfg)
+	defer MustCloseDB(b, db)
+
+	// Run multiple readers in separate goroutines.
+	ctx, cancel := context.WithCancel(context.Background())
+	g, ctx := errgroup.WithContext(ctx)
+	for i := 0; i < 8; i++ {
+		i := i
+		g.Go(func() error {
+			for {
+				if ctx.Err() != nil {
+					return nil // cancelled, return no error
+				} else if err := func() error {
+					tx, err := db.Begin(false)
+					if err != nil {
+						return err
+					}
+					defer tx.Rollback()
+
+					time.Sleep(time.Duration(rand.Intn(int(3 * time.Millisecond))))
+
+					times := rand.Intn(1000) + 1
+					for j := 0; j < times; j++ {
+						v := randInts[((i<<10)+j)%(randPool-1)]
+						if _, err := tx.Contains("x", uint64(v)); err != nil {
+							return err
+						}
+					}
+					return nil
+				}(); err != nil {
+					return err
+				}
+				// time.Sleep(time.Duration(rand.Intn(int(3 * time.Millisecond))))
+			}
+		})
+	}
+
+	// Continuously set/clear bits while readers are executing.
+	next := 0
+	for i := 0; i < 1000; i++ {
+		func() {
+			tx, err := db.Begin(true)
+			if err != nil {
+				b.Fatal(err)
+			}
+			defer tx.Rollback()
+
+			times := rand.Intn(100)
+			for j := 0; j < times; j++ {
+				v := randInts[next]
+				next = (next + 1) % (randPool - 1)
+				if j&7 == 0 {
+					// some removes but they're less frequent
+					if _, err := tx.Remove("x", uint64(v)); err != nil {
+						b.Fatal(err)
+					}
+				} else {
+					if _, err := tx.Add("x", uint64(v)); err != nil {
+						b.Fatal(err)
+					}
+				}
+
+			}
+			if err := tx.Commit(); err != nil {
+				b.Fatal(err)
+			}
+		}()
+	}
+
+	// Stop readers & wait.
+	cancel()
+	if err := g.Wait(); err != nil {
+		b.Fatal(err)
+	}
+}
+
+func BenchmarkDbCheckpoint(b *testing.B) {
+	out, err := os.Create("cp.out")
+	if err != nil {
+		b.Fatalf("creating log file: %v", err)
+	}
+	done := fgprof.Start(out, fgprof.FormatPprof)
+	b.StopTimer()
+	// premake these because otherwise it's >5% of CPU in the reads
+	randInts := make([]int, randPool)
+	for i := range randInts {
+		v1, v2 := rand.Intn(1<<24), rand.Intn(1<<24)
+		// minimum gives us a skewed distribution which makes lower values more
+		// likely than higher values, so we get a mix of container types
+		if v1 < v2 {
+			randInts[i] = v1
+		} else {
+			randInts[i] = v2
+		}
+	}
+	b.StartTimer()
+	for i := 0; i < b.N; i++ {
+		benchmarkOneCheckpoint(b, randInts)
+	}
+	b.StopTimer()
+	done()
+}
+
 // better diagnosis of deadlocks/hung situations versus just really slow "Quick" tests.
 func TestMain(m *testing.M) {
 	l, err := net.Listen("tcp", ":0")
--- a/rbf/rbf_test.go
+++ b/rbf/rbf_test.go
@ -11,6 +11,7 @@ import (
 	"sort"
 	"testing"

+	"github.com/molecula/featurebase/v2/logger"
 	"github.com/molecula/featurebase/v2/rbf"
 	rbfcfg "github.com/molecula/featurebase/v2/rbf/cfg"
 	"github.com/molecula/featurebase/v2/testhook"
@ -65,6 +66,13 @@ func NewDB(tb testing.TB, cfg ...*rbfcfg.Config) *rbf.DB {
 // MustOpenDB returns a db opened on a temporary file. On error, fail test.
 func MustOpenDB(tb testing.TB, cfg ...*rbfcfg.Config) *rbf.DB {
 	tb.Helper()
+	if len(cfg) == 0 || cfg[0] == nil {
+		newconf := rbfcfg.NewDefaultConfig()
+		newconf.Logger = logger.NewLogfLogger(tb)
+		cfg = []*rbfcfg.Config{newconf}
+	} else if cfg[0].Logger == nil {
+		cfg[0].Logger = logger.NewLogfLogger(tb)
+	}
 	db := NewDB(tb, cfg...)
 	if err := db.Open(); err != nil {
 		tb.Fatal(err)
@ -78,7 +86,14 @@ func MustCloseDB(tb testing.TB, db *rbf.DB) {
 	tb.Helper()
 	if err := db.Check(); err != nil && err != rbf.ErrClosed {
 		tb.Fatal(err)
-	} else if n := db.TxN(); n != 0 {
+	}
+	MustCloseDBNoCheck(tb, db)
+}
+
+// MustCloseDBNoCheck closes db. On error, fail test.
+func MustCloseDBNoCheck(tb testing.TB, db *rbf.DB) {
+	tb.Helper()
+	if n := db.TxN(); n != 0 {
 		tb.Fatalf("db still has %d active transactions; must closed before closing db", n)
 	} else if err := db.Close(); err != nil && err != rbf.ErrClosed {
 		tb.Fatal(err)
--- a/rbf/tx.go
+++ b/rbf/tx.go
@ -109,24 +109,31 @@ func (tx *Tx) Commit() error {
 		// future plan: after checkpoint is moved to background
 		// or not every removeTx, then we can move the
 		// tx.db.rootRecords = tx.rootRecords into removeTx().
-
+		//
+		// ... or maybe not: let's do that part here, and then removeTx
+		// may or may not start a checkpoint, possibly asynchronously.
+		//
 		// avoid race detector firing on a write race here
-		// vs the read of rootRecords at db.Begin()
+		// vs the read of rootRecords at db.Begin(), then release
+		// the lock, because we need removeTx to grab the lock to
+		// work, but if it wants to checkpoint, it wants to be able to return
+		// to us here and still be holding the lock.
 		tx.db.mu.Lock()
-		defer tx.db.mu.Unlock()
 		tx.db.rootRecords = tx.rootRecords
 		tx.db.pageMap = tx.pageMap
 		tx.db.walPageN = tx.walPageN
-		return tx.db.removeTx(tx)
+		tx.db.mu.Unlock()
 	}

-	// Disconnect transaction from DB.
 	tx.db.mu.Lock()
 	defer tx.db.mu.Unlock()
+	// Disconnect transaction from DB.
 	return tx.db.removeTx(tx)
 }

-func (tx *Tx) Rollback() {
+func (tx *Tx) Rollback() { tx.rollback(false) }
+
+func (tx *Tx) rollback(hasDBLock bool) {
 	tx.mu.Lock()
 	defer tx.mu.Unlock()

@ -141,8 +148,10 @@ func (tx *Tx) Rollback() {
 	}

 	// Disconnect transaction from DB.
-	tx.db.mu.Lock()
-	defer tx.db.mu.Unlock()
+	if !hasDBLock {
+		tx.db.mu.Lock()
+		defer tx.db.mu.Unlock()
+	}
 	vprint.PanicOn(tx.db.removeTx(tx))
 }

@ -732,6 +741,27 @@ func (tx *Tx) Check() error {
 	return nil
 }

+func (tx *Tx) checkPage(pgno, parent, typ uint32) error {
+	switch typ {
+	case PageTypeBranch:
+		return tx.checkBranchPage(pgno, parent, typ)
+	default:
+		return nil
+	}
+}
+
+func (tx *Tx) checkBranchPage(pgno, parent, typ uint32) error {
+	page, _, err := tx.readPage(pgno)
+	if err != nil {
+		return err
+	}
+
+	if readCellN(page) == 0 {
+		return fmt.Errorf("branch page %d is empty", pgno)
+	}
+	return nil
+}
+
 // checkPageAllocations ensures that all pages are either in-use or on the freelist.
 func (tx *Tx) checkPageAllocations() error {
 	freePageSet, err := tx.freePageSet()
@ -821,7 +851,7 @@ func (tx *Tx) inusePageSet() (map[uint32]struct{}, error) {
 	// Traverse freelist and mark pages as in-use.
 	if err := tx.walkTree(readMetaFreelistPageNo(tx.meta[:]), 0, func(pgno, parent, typ uint32) error {
 		m[pgno] = struct{}{}
-		return nil
+		return tx.checkPage(pgno, parent, typ)
 	}); err != nil {
 		return m, err
 	}
@ -837,7 +867,8 @@ func (tx *Tx) inusePageSet() (map[uint32]struct{}, error) {

 		if err := tx.walkTree(pgno.(uint32), 0, func(pgno, parent, typ uint32) error {
 			m[pgno] = struct{}{}
-			return nil
+
+			return tx.checkPage(pgno, parent, typ)
 		}); err != nil {
 			return m, err
 		}
--- a/rbf/tx_test.go
+++ b/rbf/tx_test.go
@ -2,8 +2,11 @@
 package rbf_test

 import (
+	"encoding/binary"
 	"fmt"
 	"math/rand"
+	"os"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@ -433,6 +436,52 @@ func TestTx_DeallocateToFreeList(t *testing.T) {
 	}
 }

+func TestTx_Remove(t *testing.T) {
+	t.Parallel()
+
+	db := MustOpenDB(t)
+	defer MustCloseDB(t, db)
+
+	tx := MustBegin(t, db, true)
+	defer tx.Rollback()
+
+	if err := tx.CreateBitmap("x"); err != nil {
+		t.Fatal(err)
+	}
+
+	// Insert large array values.
+	var values []uint64
+	for i := 0; i < 1000; i++ {
+		for j := 0; j < rbf.ArrayMaxSize; j++ {
+			v := uint64((i << 16) + j)
+			values = append(values, v)
+
+			if _, err := tx.Add("x", v); err != nil {
+				t.Fatalf("Add(%d) err=%q", v, err)
+			}
+		}
+	}
+
+	if err := tx.Commit(); err != nil {
+		t.Fatal(err)
+	}
+
+	tx = MustBegin(t, db, true)
+	defer tx.Rollback()
+
+	// Remove all array values.
+	for _, i := range rand.Perm(len(values)) {
+		v := values[i]
+		if _, err := tx.Remove("x", v); err != nil {
+			t.Fatalf("Remove(%d) err=%q", v, err)
+		}
+	}
+
+	if err := tx.Commit(); err != nil {
+		t.Fatal(err)
+	}
+}
+
 func TestTx_AddRemove_Quick(t *testing.T) {
 	if testing.Short() {
 		t.Skip("-short enabled, skipping")
@ -770,3 +819,83 @@ func TestTx_DeleteBitmapsWithPrefix(t *testing.T) {
 	checkInfos()

 }
+
+func TestTx_Check(t *testing.T) {
+	t.Run("EmptyBranchPage", func(t *testing.T) {
+		t.Parallel()
+
+		db := MustOpenDB(t)
+		defer MustCloseDBNoCheck(t, db)
+		tx := MustBegin(t, db, true)
+		defer tx.Rollback()
+
+		if err := tx.CreateBitmap("x"); err != nil {
+			t.Fatal(err)
+		}
+
+		// Insert enough array containers to split page.
+		for i := 0; i < 1000; i++ {
+			if _, err := tx.Add("x", uint64(i<<16)); err != nil {
+				t.Fatalf("Add(%d) err=%q", i<<16, err)
+			}
+		}
+
+		// Read page types for all pages.
+		infos, err := tx.PageInfos()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		// Commit & checkpoint to flush to the data file.
+		if err := tx.Commit(); err != nil {
+			t.Fatal(err)
+		} else if err := db.Checkpoint(); err != nil {
+			t.Fatal(err)
+		}
+
+		// Corrupt first branch page found by zeroing out the cell count.
+		var pgno uint32
+		for _, info := range infos {
+			if info, ok := info.(*rbf.BranchPageInfo); ok {
+				pgno = info.Pgno
+				page := mustReadPage(t, db.DataPath(), pgno)
+				binary.BigEndian.PutUint16(page[8:10], 0) // zero cell count
+				mustWritePage(t, db.DataPath(), pgno, page)
+				break
+			}
+		}
+
+		// Verify that check now returns an error.
+		if err := db.Check(); err == nil || !strings.Contains(err.Error(), fmt.Sprintf("branch page %d is empty", pgno)) {
+			t.Fatalf("unexpected error: %#v", err)
+		}
+	})
+}
+
+func mustReadPage(tb testing.TB, path string, pgno uint32) []byte {
+	tb.Helper()
+	f, err := os.Open(path)
+	if err != nil {
+		tb.Fatal(err)
+	}
+	defer f.Close()
+
+	buf := make([]byte, rbf.PageSize)
+	if _, err := f.ReadAt(buf, int64(pgno)*rbf.PageSize); err != nil {
+		tb.Fatal(err)
+	}
+	return buf
+}
+
+func mustWritePage(tb testing.TB, path string, pgno uint32, buf []byte) {
+	tb.Helper()
+	f, err := os.OpenFile(path, os.O_WRONLY, 0666)
+	if err != nil {
+		tb.Fatal(err)
+	}
+	defer f.Close()
+
+	if _, err := f.WriteAt(buf, int64(pgno)*rbf.PageSize); err != nil {
+		tb.Fatal(err)
+	}
+}
--- a/server.go
+++ b/server.go
@ -476,7 +476,6 @@ func NewServer(opts ...ServerOption) (*Server, error) {
 	}
 	s.holder = NewHolder(path, s.holderConfig)
 	s.holder.Stats.SetLogger(s.logger)
-	s.holder.Logger.Infof("RowCacheOn: %v", s.holderConfig.RowcacheOn)
 	cwd, err := os.Getwd()
 	if err != nil {
 		return nil, err
--- a/server/config.go
+++ b/server/config.go
@ -4,14 +4,19 @@ package server
 import (
 	"context"
 	"fmt"
+	"io"
 	"log"
 	"net"
 	"net/url"
+	"os"
+	"path/filepath"
 	"runtime"
 	"strconv"
 	"strings"
 	"time"

+
+	"github.com/molecula/featurebase/v2/authz"
 	petcd "github.com/molecula/featurebase/v2/etcd"
 	rbfcfg "github.com/molecula/featurebase/v2/rbf/cfg"
 	"github.com/molecula/featurebase/v2/storage"
@ -199,9 +204,9 @@ type Config struct {
 	// "rbf".
 	Storage *storage.Config `toml:"storage"`

-	// RowcacheOn, if true, turns on the row cache for all storage backends.
-	// The default is now off because it makes rbf queries faster and uses
-	// much less memory.
+	// RowcacheOn permanently disabled. No longer useful w/ RBF. Left
+	// for backward compatibility but will be removed in a future
+	// version.
 	RowcacheOn bool `toml:"rowcache-on"`

 	// RBFConfig defines all externally configurable RBF flags.
@ -229,16 +234,15 @@ type Config struct {
 	// Toggles /schema/details endpoint. If off, it returns empty.
 	SchemaDetailsOn bool `toml:"schema-details-on"`

+
 	Auth Auth
 }

 type Auth struct {
 	// Enable AuthZ/AuthN for featurebase server
 	Enable bool `toml:"enable"`
-
-	// Application/Client ID
+  
 	ClientId string `toml:"client-id"`
-
 	ClientSecret     string   `toml:"client-secret"`
 	AuthorizeURL     string   `toml:"authorize-url"`
 	TokenURL         string   `toml:"token-url"`
@ -247,6 +251,8 @@ type Auth struct {
 	Scopes           []string `toml:"scopes"`
 	HashKey          string   `toml:"hash-key"`
 	BlockKey         string   `toml:"block-key"`
+  PermissionsFile string `toml:"permissions"`
+	Auth authz.Auth `toml:"auth"`
 }

 // Namespace returns the namespace to use based on the Future flag.
@ -611,9 +617,9 @@ func lookupAddr(ctx context.Context, resolver *net.Resolver, host string) (strin
 	return addrs[0].String(), nil
 }

-func (c *Config) ValidateAuth() ([]error, error) {
+func (c *Config) ValidateAuth() (errors []error) {
 	if !c.Auth.Enable {
-		return []error{}, nil
+		return
 	}
 	authConfig := map[string]string{
 		"ClientId":         c.Auth.ClientId,
@ -626,7 +632,6 @@ func (c *Config) ValidateAuth() ([]error, error) {
 		"BlockKey":         c.Auth.BlockKey,
 	}

-	errors := make([]error, 0)
 	for name, value := range authConfig {
 		if value == "" {
 			errors = append(errors, fmt.Errorf("empty string for auth config %s", name))
@ -647,20 +652,101 @@ func (c *Config) ValidateAuth() ([]error, error) {
 			}
 		}
 	}
+
 	if len(c.Auth.Scopes) == 0 {
 		errors = append(errors, fmt.Errorf("must provide scope for authentication with IdP - for access and refresh token"))
 	}
-	if len(errors) > 0 {
-		return errors, fmt.Errorf("there were errors validating config")
+
+	return errors
+}
+
+func (c *Config) ValidatePermissions(permsFile io.Reader) (errors []error) {
+
+	var p authz.GroupPermissions
+	if err := p.ReadPermissionsFile(permsFile); err != nil {
+		return append(errors, err)
 	}
-	return errors, nil
+
+	if len(p.Permissions) == 0 {
+		return append(errors, fmt.Errorf("no group permissions found in permissions file: %s", c.Auth.PermissionsFile))
+	}
+
+	for groupId, indexPerm := range p.Permissions {
+		if groupId == "" {
+			errors = append(errors, fmt.Errorf("empty string for group id in permissions file %s", c.Auth.PermissionsFile))
+			continue
+		}
+
+		for index, perm := range indexPerm {
+			if index == "" {
+				errors = append(errors, fmt.Errorf("empty string for index for group id %s in permissions file %s ", groupId, c.Auth.PermissionsFile))
+				continue
+			}
+
+			if perm == "" {
+				errors = append(errors, fmt.Errorf("empty string for permission for group id %s and index %s in permissions file %s", groupId, index, c.Auth.PermissionsFile))
+				continue
+			}
+
+			if !((perm == "write") || (perm == "read")) {
+				errors = append(errors, fmt.Errorf("not a valid permission %s for group id %s and index %s in permissions file %s; expected permissions are read or write", perm, groupId, index, c.Auth.PermissionsFile))
+				continue
+			}
+		}
+	}
+
+	if p.Admin == "" {
+		errors = append(errors, fmt.Errorf("empty string for admin in permissions file: %s", c.Auth.PermissionsFile))
+
+	}
+
+	return errors
+}
+
+func (c *Config) ValidatePermissionsFile() (err error) {
+
+	if c.Auth.PermissionsFile == "" {
+		return fmt.Errorf("empty string for auth config permissions file")
+	}
+
+	fileExt := filepath.Ext(c.Auth.PermissionsFile)
+	if (fileExt != ".yaml") && (fileExt != ".yml") {
+		return fmt.Errorf("invalid file extension for auth config permissions file: %s", c.Auth.PermissionsFile)
+	}
+	return
 }

 func (c *Config) MustValidateAuth() {
-	if errors, err := c.ValidateAuth(); err != nil {
-		for _, e := range errors {
+
+	errorsAuth := c.ValidateAuth()
+	if len(errorsAuth) > 0 {
+		for _, e := range errorsAuth {
 			log.Println(e)
 		}
-		log.Fatal(err)
+	}
+
+	var errorsPerm []error
+	errorsPermFile := c.ValidatePermissionsFile()
+	if errorsPermFile == nil {
+		permsFile, err := os.Open(c.Auth.PermissionsFile)
+		if err != nil {
+			log.Println(err)
+		}
+
+		defer permsFile.Close()
+
+		errorsPerm = c.ValidatePermissions(permsFile)
+		if len(errorsPerm) > 0 {
+			for _, e := range errorsPerm {
+				log.Println(e)
+			}
+		}
+
+	} else {
+		log.Println(errorsPermFile)
+	}
+
+	if len(errorsAuth) > 0 || len(errorsPerm) > 0 || errorsPermFile != nil {
+		log.Fatal(fmt.Errorf("there were errors validating authN/authZ config and/or permissions"))
 	}
 }
--- a/server/config_internal_test.go
+++ b/server/config_internal_test.go
@ -8,6 +8,7 @@ import (
 	"os"
 	"strings"
 	"testing"
+	"github.com/molecula/featurebase/v2/authz"
 )

 type addrs struct{ bind, advertise string }
@ -280,11 +281,12 @@ func TestConfig_validateAuth(t *testing.T) {
 	errorMesgEmpty := "empty string"
 	errorMesgURL := "invalid URL"
 	errorMesgScope := "must provide scope"
+  errorMesgKey := "invalid key length"
 	validTestURL := "https://url.com/"
 	validClientID := "clientid"
 	validClientSecret := "clientSecret"
 	validKey := "3db6665be8b860af422155acf2346d4fcb46678fca42e60d934abe0b7ce43600"
-	notValidURL := "not-a-url"
+	invalidURL := "not-a-url"
 	emptyString := ""
 	validStringSlice := []string{"https://graph.microsoft.com/.default", "offline_access"}
 	validString := "asdfqwer1234asdfzxcv"
@ -297,7 +299,6 @@ func TestConfig_validateAuth(t *testing.T) {
 		expErrs []string
 		input   Auth
 	}{
-
 		{
 			// Auth enabled, all configs are set to empty string
 			[]string{
@ -309,7 +310,6 @@ func TestConfig_validateAuth(t *testing.T) {
 				errorMesgEmpty,
 				errorMesgEmpty,
 				errorMesgEmpty,
-				errorMesgScope,
 			},
 			Auth{
 				Enable:           enable,
@ -319,168 +319,65 @@ func TestConfig_validateAuth(t *testing.T) {
 				TokenURL:         emptyString,
 				GroupEndpointURL: emptyString,
 				LogoutURL:        emptyString,
-				Scopes:           emptySlice,
+				Scopes:           validStringSlice,
 				HashKey:          emptyString,
 				BlockKey:         emptyString,
 			},
 		},
 		{
-			// Auth enabled, some configs are set to empty string
+			// Auth enabled, keys are invalid length
 			[]string{
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgScope,
-			},
-			Auth{
-				Enable:           enable,
-				ClientId:         validClientID,
-				ClientSecret:     emptyString,
-				AuthorizeURL:     emptyString,
-				TokenURL:         emptyString,
-				GroupEndpointURL: emptyString,
-				LogoutURL:        emptyString,
-				Scopes:           emptySlice,
-				HashKey:          emptyString,
-				BlockKey:         emptyString,
-			},
-		},
-		{
-			// Auth enabled, some configs are set to empty string
-			[]string{
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgScope,
-			},
-			Auth{
-				Enable:           enable,
-				ClientId:         emptyString,
-				ClientSecret:     validClientSecret,
-				AuthorizeURL:     emptyString,
-				TokenURL:         emptyString,
-				GroupEndpointURL: emptyString,
-				LogoutURL:        emptyString,
-				Scopes:           emptySlice,
-				HashKey:          emptyString,
-				BlockKey:         emptyString,
-			},
-		},
-		{
-			// Auth enabled, some configs are set to empty string
-			[]string{
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgURL,
-				errorMesgScope,
-			},
-			Auth{
-				Enable:           enable,
-				ClientId:         validClientID,
-				ClientSecret:     validClientSecret,
-				AuthorizeURL:     emptyString,
-				TokenURL:         emptyString,
-				GroupEndpointURL: emptyString,
-				LogoutURL:        notValidURL,
-				Scopes:           emptySlice,
-				HashKey:          emptyString,
-				BlockKey:         emptyString,
-			},
-		},
-		{
-			// Auth enabled, some configs are set to empty string
-			[]string{
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgScope,
+				errorMesgKey,
+				errorMesgKey,
 			},
 			Auth{
 				Enable:           enable,
 				ClientId:         validClientID,
 				ClientSecret:     validClientSecret,
 				AuthorizeURL:     validTestURL,
-				TokenURL:         emptyString,
-				GroupEndpointURL: emptyString,
-				LogoutURL:        emptyString,
-				Scopes:           emptySlice,
-				HashKey:          emptyString,
-				BlockKey:         emptyString,
-			},
-		},
-		{
-			// Auth enabled, some configs are set to empty string
-			[]string{
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgScope,
-			},
-			Auth{
-				Enable:           enable,
-				ClientId:         validClientID,
-				ClientSecret:     validClientSecret,
-				AuthorizeURL:     validTestURL,
-				TokenURL:         validTestURL,
-				GroupEndpointURL: emptyString,
-				LogoutURL:        validTestURL,
-				Scopes:           emptySlice,
-				HashKey:          emptyString,
-				BlockKey:         emptyString,
-			},
-		},
-		{
-			// Auth enabled, some strings are set to invalid URL
-			[]string{
-				errorMesgEmpty,
-				errorMesgEmpty,
-				errorMesgURL,
-				errorMesgURL,
-			},
-			Auth{
-				Enable:           enable,
-				ClientId:         validClientID,
-				ClientSecret:     validClientSecret,
-				AuthorizeURL:     notValidURL,
 				TokenURL:         validTestURL,
 				GroupEndpointURL: validTestURL,
-				LogoutURL:        notValidURL,
+				LogoutURL:        validTestURL,
 				Scopes:           validStringSlice,
-				HashKey:          emptyString,
-				BlockKey:         emptyString,
+				HashKey:          validString,
+				BlockKey:         validString,
 			},
 		},
 		{
-			// Auth enabled, some strings are set to invalid URL
+			// Auth enabled, some URLs are set to invalid URL
 			[]string{
 				errorMesgURL,
 				errorMesgURL,
 				errorMesgURL,
-				errorMesgEmpty,
 			},
 			Auth{
 				Enable:           enable,
 				ClientId:         validClientID,
 				ClientSecret:     validClientSecret,
 				AuthorizeURL:     validTestURL,
-				TokenURL:         notValidURL,
-				GroupEndpointURL: notValidURL,
-				LogoutURL:        notValidURL,
+				TokenURL:         invalidURL,
+				GroupEndpointURL: invalidURL,
+				LogoutURL:        invalidURL,
 				Scopes:           validStringSlice,
-				HashKey:          emptyString,
+				HashKey:          validKey,
+				BlockKey:         validKey,
+			},
+		},
+		{
+			// Auth enabled, all configs are set properly except scope
+			[]string{
+        errorMesgScope,
+      },
+			Auth{
+				Enable:           enable,
+				ClientId:         validClientID,
+				ClientSecret:     validClientSecret,
+				AuthorizeURL:     validTestURL,
+				TokenURL:         validTestURL,
+				GroupEndpointURL: validTestURL,
+				LogoutURL:        validTestURL,
+				Scopes:           emptySlice,
+				HashKey:          validKey,
 				BlockKey:         validKey,
 			},
 		},
@ -501,19 +398,19 @@ func TestConfig_validateAuth(t *testing.T) {
 			},
 		},
 		{
-			// Auth disabled, all configs are set to some values
+			// Auth disabled, some configs are set to values
 			[]string{},
 			Auth{
 				Enable:           disable,
-				ClientId:         validString,
+				ClientId:         emptyString,
 				ClientSecret:     validString,
-				AuthorizeURL:     validString,
-				TokenURL:         validTestURL,
-				GroupEndpointURL: validTestURL,
+				AuthorizeURL:     emptyString,
+				TokenURL:         emptyString,
+				GroupEndpointURL: invalidURL,
 				LogoutURL:        validTestURL,
 				Scopes:           validStringSlice,
 				HashKey:          validKey,
-				BlockKey:         validKey,
+				BlockKey:         emptyString,
 			},
 		},
 	}
@ -523,9 +420,9 @@ func TestConfig_validateAuth(t *testing.T) {
 			c := NewConfig()
 			c.Auth = test.input

-			errors, err := c.ValidateAuth()
+			errors := c.ValidateAuth()
 			if len(test.expErrs) > 0 {
-				if err == nil {
+				if errors == nil {
 					t.Fatal("expected errors, but none were found")
 				}
 			}
@ -535,11 +432,121 @@ func TestConfig_validateAuth(t *testing.T) {
 				t.Fatalf("expected %v errors but got %v", len(test.expErrs), len(errors))
 			}

-			// for i, e := range errors {
-			// 	if !strings.Contains(e.Error(), test.expErrs[i]) {
-			// 		t.Errorf("expected error to contain %s, but got %s", test.expErrs[i], e.Error())
-			// 	}
-			// }
+			for i, e := range errors {
+			 	if !strings.Contains(e.Error(), test.expErrs[i]) {
+			 		t.Errorf("expected error to contain %s, but got %s", test.expErrs[i], e.Error())
+			 	}
+			}
+		})
+	}
+}
+
+func TestConfig_validatePermissions(t *testing.T) {
+	permissions0 := ``
+
+	permissions1 := `user-groups:
+  "":
+    "test": "read"
+admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"`
+
+	permissions2 := `user-groups:
+  "dca35310-ecda-4f23-86cd-876aee559900":
+    "": "write"
+admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"`
+
+	permissions3 := `user-groups:
+  "dca35310-ecda-4f23-86cd-876aee559900":
+    "test": ""
+admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"`
+
+	permissions4 := `user-groups:
+  "dca35310-ecda-4f23-86cd-876aee559900":
+    "test": "readwrite"
+admin: "ac97c9e2-346b-42a2-b6da-18bcb61a32fe"`
+
+	permissions5 := `user-groups:
+  "dca35310-ecda-4f23-86cd-876aee559900":
+    "test": "read"`
+
+	tests := []struct {
+		err   string
+		input string
+	}{
+		{
+			"no group permissions found in permissions file",
+			permissions0,
+		},
+		{
+			"empty string for group id",
+			permissions1,
+		},
+		{
+			"empty string for index",
+			permissions2,
+		},
+		{
+			"empty string for permission",
+			permissions3,
+		},
+		{
+			"not a valid permission",
+			permissions4,
+		},
+		{
+			"empty string for admin in permissions file",
+			permissions5,
+		},
+	}
+
+	for i, test := range tests {
+		t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
+
+			c := NewConfig()
+			c.Auth.PermissionsFile = "test.yaml"
+
+			permFile := strings.NewReader(test.input)
+			errors := c.ValidatePermissions(permFile)
+
+			if errors == nil {
+				t.Fatal("expected errors, but none were found")
+			}
+
+			for _, err := range errors {
+				if !strings.Contains(err.Error(), test.err) {
+					t.Errorf("expected error to contain %s, but got %s", test.err, err.Error())
+
+				}
+			}
+		})
+	}
+}
+
+func TestConfig_validatePermissionsFilename(t *testing.T) {
+
+	tests := []struct {
+		err   string
+		input string
+	}{
+		{
+			"empty string for auth config permissions file",
+			"",
+		},
+		{
+			"invalid file extension for auth config permissions file",
+			"permissions.txt",
+		},
+	}
+
+	for i, test := range tests {
+		t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
+			c := NewConfig()
+			c.Auth.PermissionsFile = test.input
+
+			if err := c.ValidatePermissionsFile(); err != nil {
+				if !strings.Contains(err.Error(), test.err) {
+					t.Errorf("expected error to contain %s, but got %s", test.err, err.Error())
+				}
+			}
 		})
 	}
 }
--- a/server/server.go
+++ b/server/server.go
@ -30,6 +30,7 @@ import (

 	pilosa "github.com/molecula/featurebase/v2"
 	"github.com/molecula/featurebase/v2/authn"
+	"github.com/molecula/featurebase/v2/authz"
 	"github.com/molecula/featurebase/v2/boltdb"
 	"github.com/molecula/featurebase/v2/encoding/proto"
 	petcd "github.com/molecula/featurebase/v2/etcd"
@ -481,7 +482,7 @@ func (m *Command) SetupServer() error {
 		pilosa.OptServerClusterName(m.Config.Cluster.Name),
 		pilosa.OptServerSerializer(proto.Serializer{}),
 		pilosa.OptServerStorageConfig(m.Config.Storage),
-		pilosa.OptServerRowcacheOn(m.Config.RowcacheOn),
+		pilosa.OptServerRowcacheOn(false),
 		pilosa.OptServerRBFConfig(m.Config.RBFConfig),
 		pilosa.OptServerMaxQueryMemory(m.Config.MaxQueryMemory),
 		pilosa.OptServerQueryHistoryLength(m.Config.QueryHistoryLength),
@ -524,12 +525,22 @@ func (m *Command) SetupServer() error {

 	if m.Config.Auth.Enable {
 		m.Config.MustValidateAuth()
+		permsFile, err := os.Open(m.Config.Auth.PermissionsFile)
+		if err != nil {
+			return err
+		}
+		defer permsFile.Close()
+
+		var p authz.GroupPermissions
+		if err = p.ReadPermissionsFile(permsFile); err != nil {
+			return err
+		}
+    
 		ac := m.Config.Auth
 		m.auth, err = authn.NewAuth(m.logger, m.listenURI.String(), ac.Scopes, ac.AuthorizeURL, ac.TokenURL, ac.GroupEndpointURL, ac.LogoutURL, ac.ClientId, ac.ClientSecret, ac.HashKey, ac.BlockKey)
 		if err != nil {
 			return errors.Wrap(err, "instantiating authN object")
 		}
-
 	}

 	m.Handler, err = http.NewHandler(
--- a/testBackupRestore.sh
+++ b/testBackupRestore.sh
@ -1,57 +0,0 @@
-#!/bin/bash
-
-set -eux
-
-declare STATUS="NORMAL"
-declare TIMEOUT=30
-sleep 4
-STATUS=$STATUS timeout -s TERM $TIMEOUT bash -c \
-    'while [[ ${STATUS_RECEIVED} != ${STATUS} ]];\
-    	do STATUS_RECEIVED=$(curl --connect-timeout 1 -s pilosa0:10101/status | jq  -r ".state") && \
-        echo "received status: $STATUS_RECEIVED" && \
-        sleep 1;\
-    done;'
-echo "NOW DO STUFF"
-datagen --source kitchensink_keyed -e 9999 --pilosa.index sink --pilosa.batch-size 10000 --pilosa.hosts pilosa0:10101
-before=$(/featurebase chksum --host pilosa0:10101)
-/featurebase backup -o backupdir --host pilosa0:10101
-curl -X DELETE -s pilosa0:10101/index/sink
-/featurebase restore -s backupdir --host pilosa0:10101
-after=$(/featurebase chksum --host pilosa0:10101)
-if [ "$before" = "$after" ]; then
-	echo "PASS Cluster"
-else
-	echo "FAIL Single"
-	exit 1
-fi
-/featurebase restore -s backupdir --host pilosax:10101
-single=$(/featurebase chksum --host pilosax:10101)
-if [ "$before" = "$single" ]; then
-	echo "PASS Single"
-	exit 0
-else
-	echo "FAIL Single"
-	exit 1
-fi
-
-datagen --source texas_health -e 9999 --pilosa.index newsink --pilosa.batch-size 10000 --pilosa.hosts pilosa0:10101
-before=$(/featurebase chksum --host pilosa0:10101)
-/featurebase backup -o newbackupdir --host pilosa0:10101 --index newsink
-curl -X DELETE -s pilosa0:10101/index/newsink
-/featurebase restore -s newbackupdir --host pilosa0:10101
-after=$(/featurebase chksum --host pilosa0:10101)
-if [ "$before" = "$after" ]; then
-	echo "PASS Cluster Table"
-else
-	echo "FAIL Single Table"
-	exit 1
-fi
-/featurebase restore -s newbackupdir --host pilosax:10101
-single=$(/featurebase chksum --host pilosax:10101)
-if [ "$before" = "$single" ]; then
-	echo "PASS Single Table"
-	exit 0
-else
-	echo "FAIL Single Table"
-	exit 1
-fi
--- a/translate.go
+++ b/translate.go
@ -568,6 +568,8 @@ func (s *InMemTranslateStore) WriteTo(w io.Writer) (int64, error) {
 // don't expect to use InMemTranslateStore much, it's mostly there to
 // avoid disk load during testing.
 func (s *InMemTranslateStore) ReadFrom(r io.Reader) (count int64, err error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
 	var bytes []byte
 	bytes, err = ioutil.ReadAll(r)
 	count = int64(len(bytes))
--- a/txfactory.go
+++ b/txfactory.go
@ -242,10 +242,15 @@ func (qcx *Qcx) GetTx(o Txo) (tx Tx, finisher func(perr *error), err error) {
 	}

 	// qcx.write reflects the top executor determination
-	// if a write will be done at the end, so we upgrade
-	// the "local" read Tx to be writes, so that they
-	// don't deadlock against themselves.
-	o.Write = o.Write || qcx.write
+	// if a write will be happen at some point, in which case, to avoid
+	// locking problems with multi-shard things, we (probably incorrectly)
+	// treat every Tx as its own individual separate Tx.
+	//
+	// But we still want to open non-write transactions individually, we
+	// just can't recycle them (because write operations will come in and
+	// we want them to work and commit right away so we're not holding a write
+	// lock for long).
+	writeLogic := o.Write || qcx.write

 	// In general, we make ALL write transactions local, and never reuse them
 	// below. Previously this was to help lmdb.
@ -273,7 +278,7 @@ func (qcx *Qcx) GetTx(o Txo) (tx Tx, finisher func(perr *error), err error) {
 		return *qcx.RequiredForAtomicWriteTx, NoopFinisher, nil
 	}

-	if !o.Write && qcx.Grp != nil {
+	if !writeLogic && qcx.Grp != nil {
 		// read, with a group in place.
 		finisher = func(perr *error) {} // finisher is a returned value

--- a/view.go
+++ b/view.go
@ -619,7 +619,9 @@ func (v *view) bitDepth(shards []uint64) (uint64, error) {
 	var maxBitDepth uint64

 	for _, shard := range shards {
+		v.mu.RLock()
 		frag, ok := v.fragments[shard]
+		v.mu.RUnlock()
 		if !ok || frag == nil {
 			continue
 		}