From cdd10a26f06c3db2dbacc9cd39e1d0b63bb4cbc6 Mon Sep 17 00:00:00 2001
From: Ben Johnson <benbjohnson@yahoo.com>
Date: Tue, 22 Sep 2020 13:15:29 -0600
Subject: [PATCH 01/12] Refactor RBF to use immutable list of WAL segments.

Previously, RBF shared a list of WAL segments between the DB & Tx.
However, this increased the need for mutexes to access the data.

WAL segments are effectively immutable on-disk so the list of segments
has been refactored so that changes to the segment list are done via
copy-on-write which allows read transactions to access segment data
without a mutex.

The database checkpointing can remove early, unused segments and
there is an update/add check to make sure that Tx segments pushed back
to the DB do not include removed segments.
---
 rbf/db.go       | 364 +++++++++++++++---------------------------------
 rbf/db_test.go  |   5 +-
 rbf/tx.go       | 182 +++++++++++++++++++++---
 rbf/wal.go      | 311 ++++++++++++++++++-----------------------
 rbf/wal_test.go |  20 +--
 5 files changed, 427 insertions(+), 455 deletions(-)

diff --git a/rbf/db.go b/rbf/db.go
index 1732eead5..e807956ec 100644
--- a/rbf/db.go
+++ b/rbf/db.go
@@ -41,12 +41,14 @@ const (
 type DB struct {
 	data        []byte           // mmap data
 	file        *os.File         // file descriptor
-	segments    []*WALSegment    // write-ahead log
 	rootRecords []*RootRecord    // cached root records
 	pageMap     *immutable.Map   // pgno-to-WALID mapping
 	txs         map[*Tx]struct{} // active transactions
 	opened      bool             // true if open
 
+	wcache   []byte       // wal write cache
+	segments []WALSegment // write-ahead log
+
 	mu     sync.RWMutex // general mutex
 	rwmu   sync.Mutex   // mutex for restricting single writer
 	exclmu sync.RWMutex // mutex for locking out everyone but a single writer
@@ -71,6 +73,7 @@ func NewDB(path string) *DB {
 	db := &DB{
 		txs:     make(map[*Tx]struct{}),
 		pageMap: immutable.NewMap(&uint32Hasher{}),
+		wcache:  make([]byte, MaxWALSegmentFileSize+PageSize),
 		Path:    path,
 		MaxSize: DefaultMaxSize,
 	}
@@ -84,12 +87,10 @@ func (db *DB) DataPath() string {
 
 // WALPath returns the path to the WAL directory.
 func (db *DB) WALPath() string {
-
 	return filepath.Join(db.Path, "wal")
 }
 
 func CreateDirIfNotExist(path string) {
-
 	dir := filepath.Dir(path)
 	if _, err := os.Stat(dir); os.IsNotExist(err) {
 		err = os.MkdirAll(dir, 0755)
@@ -102,7 +103,6 @@ func CreateDirIfNotExist(path string) {
 // Open opens a database with the file specified in Path.
 // Creates a new file if one does not already exist.
 func (db *DB) Open() (err error) {
-
 	db.mu.Lock()
 	defer db.mu.Unlock()
 
@@ -143,7 +143,7 @@ func (db *DB) Open() (err error) {
 	// Open write-ahead log & checkpoint to the end since no transactions are open.
 	if err := db.openWALSegments(); err != nil {
 		return fmt.Errorf("wal open: %w", err)
-	} else if err := db.checkpoint(true); err != nil {
+	} else if err := db.checkpoint(true, &nopLocker{}); err != nil {
 		return fmt.Errorf("checkpoint: %w", err)
 	}
 
@@ -171,27 +171,62 @@ func (db *DB) openWALSegments() error {
 	}
 
 	// Truncate everything after the last successful meta page.
-	if walID, err := db.findLastWALMetaPage(); err != nil {
+	if walID, err := findLastWALMetaPage(db.segments); err != nil {
 		return err
-	} else if err := db.truncateWALAfter(walID); err != nil {
+	} else if db.segments, err = truncateWALAfter(db.segments, walID); err != nil {
 		return err
 	}
 
 	return nil
 }
 
-// checkpoint copies pages from WAL segments into the main DB file. This can
+// updateWALSegment updates or adds a segment.
+func (db *DB) updateWALSegment(s WALSegment) {
+	segments := make([]WALSegment, len(db.segments), len(db.segments)+1)
+	copy(segments, db.segments)
+
+	// Find the matching segment using the path.
+	segment := walSegmentByPath(segments, s.Path)
+
+	// Update existing segment if it already exists.
+	// Otherwise append segment to the end.
+	if segment != nil {
+		*segment = s
+	} else {
+		assert(len(segments) == 0 || segments[len(segments)-1].MinWALID < s.MinWALID)
+		segments = append(segments, s)
+	}
+
+	// Replace DB segment list.
+	db.segments = segments
+}
+
+// Checkpoint copies pages from WAL segments into the main DB file. This can
 // only copy pages that aren't in use by an active transaction. The page map
 // is rebuilt as well for all WAL pages still in use.
 //
 // If exclusive is true, all WAL writes are flushed to disk.
-func (db *DB) checkpoint(exclusive bool) error {
-	if !db.opened {
+func (db *DB) Checkpoint() error {
+	return db.checkpoint(false, &db.mu)
+}
+
+// checkpoint moves WAL segments to the main DB file.
+//
+// Note that mu should db.mu when called through the external API but it
+// can be &nopLocker if called under lock.
+func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error {
+	// Obtain a snapshot of WAL segments at the start.
+	mu.Lock()
+	opened := db.opened
+	segments := db.segments
+	mu.Unlock()
+
+	if !opened {
 		return nil
 	}
 
 	// Determine last checkpointed WAL ID.
-	page, err := db.readPage(nil, 0)
+	page, err := db.readDBPage(0)
 	if err != nil {
 		return err
 	}
@@ -206,7 +241,7 @@ func (db *DB) checkpoint(exclusive bool) error {
 	pageMap := immutable.NewMap(&uint32Hasher{})
 	for {
 		// Determine last page of transaction.
-		metaWALID, err := db.findNextWALMetaPage(walID)
+		metaWALID, err := findNextWALMetaPage(segments, walID)
 		if err == io.EOF {
 			break
 		} else if err != nil {
@@ -217,7 +252,7 @@ func (db *DB) checkpoint(exclusive bool) error {
 		for ; walID <= metaWALID; walID++ {
 			canCheckpoint := exclusive || minActiveWALID == 0 || walID <= minActiveWALID
 
-			page, err := db.readWALPage(walID)
+			page, err := readWALPage(segments, walID)
 			if err != nil {
 				return err
 			}
@@ -242,13 +277,13 @@ func (db *DB) checkpoint(exclusive bool) error {
 			// Ensure we actually read the bitmap data in when we checkpoint.
 			// NOTE: The walID variable is incremented above in the pgno check.
 			if isBitmapHeader {
-				if page, err = db.readWALPage(walID); err != nil {
+				if page, err = readWALPage(segments, walID); err != nil {
 					return err
 				}
 			}
 
 			// Write page data into main db file.
-			if err := db.writePage(pgno, page); err != nil {
+			if err := db.writeDBPage(pgno, page); err != nil {
 				return err
 			}
 
@@ -261,98 +296,59 @@ func (db *DB) checkpoint(exclusive bool) error {
 
 	// Remove WAL segments that have been checkpointed.
 	if maxCheckpointedWALID != 0 {
-		for len(db.segments) > 0 {
-			segment := db.segments[0]
+		for _, segment := range segments {
 			if segment.MaxWALID() > maxCheckpointedWALID {
 				break
 			}
 
-			segpath := segment.Path()
-			if err := segment.Close(); err != nil {
-				return err
-			} else if err := os.Remove(segpath); err != nil {
+			if err := func() error {
+				mu.Lock()
+				defer mu.Unlock()
+				return db.removeWALSegment(segment.Path)
+			}(); err != nil {
 				return err
 			}
-			db.segments, db.segments[0] = db.segments[1:], nil
 		}
 	}
 
 	// Ensure all segments are flushed and there is no remapped pages.
 	if exclusive {
+		mu.Lock()
 		assert(len(db.segments) == 0)
 		assert(pageMap.Len() == 0)
+		mu.Unlock()
 	}
 
 	db.pageMap = pageMap
 	return nil
 }
 
-// truncateWALAfter removes all pages in the WAL after walID.
-func (db *DB) truncateWALAfter(walID int64) error {
-	for i := len(db.segments) - 1; i >= 0; i-- {
-		segment := db.segments[i]
-		if segment.MaxWALID() <= walID {
-			break
-		}
-
-		// Drop entire segment if all pages are after WAL ID.
-		if walID < segment.MinWALID() {
+// removeWALSegment closes and deletes the segment with the given path.
+//
+// The DB's segment list is entirely replaced so that transactions with
+// a reference to the old list can continue to use it without a lock.
+func (db *DB) removeWALSegment(path string) error {
+	newSegments := make([]WALSegment, 0, len(db.segments))
+	for _, segment := range db.segments {
+		// Close and remove if path matches.
+		if segment.Path == path {
 			if err := segment.Close(); err != nil {
 				return err
-			} else if err := os.Remove(segment.Path()); err != nil {
+			} else if err := os.Remove(segment.Path); err != nil {
 				return err
 			}
-			db.segments, db.segments[i] = db.segments[:len(db.segments)-1], nil
 			continue
 		}
 
-		// If we only remove some of the WAL pages then truncate and exit
-		// since segments before this will retain all their pages.
-		return segment.TruncateAfter(walID)
+		// Otherwise append to new slice of segments.
+		newSegments = append(newSegments, segment)
 	}
+
+	// Replace entire slice of segments.
+	db.segments = newSegments
 	return nil
 }
 
-func (db *DB) findNextWALMetaPage(walID int64) (metaWALID int64, err error) {
-	maxWALID := db.maxWALID()
-
-	for ; walID <= maxWALID; walID++ {
-		// Read page data from WAL and return if it is a meta page.
-		page, err := db.readWALPage(walID)
-		if err != nil {
-			return walID, err
-		} else if IsMetaPage(page) {
-			return walID, nil
-		}
-
-		// Skip over next page if this is a bitmap header.
-		if IsBitmapHeader(page) {
-			walID++
-		}
-	}
-
-	return -1, io.EOF
-}
-
-func (db *DB) findLastWALMetaPage() (walID int64, err error) {
-	if len(db.segments) == 0 {
-		return 0, nil
-	}
-
-	var maxMetaWALID int64
-	maxWALID := db.maxWALID()
-	for walID := db.minWALID(); walID <= maxWALID; walID++ {
-		if page, err := db.readWALPage(walID); err != nil {
-			return walID, err
-		} else if IsBitmapHeader(page) {
-			walID++ // skip next page for bitmap headers
-		} else if IsMetaPage(page) {
-			maxMetaWALID = walID // save max meta WAL ID
-		}
-	}
-	return maxMetaWALID, nil
-}
-
 // minActiveWALID returns the lowest WAL ID in use by any active transaction.
 // Returns 0 if no transactions are active.
 func (db *DB) minActiveWALID() int64 {
@@ -365,147 +361,8 @@ func (db *DB) minActiveWALID() int64 {
 	return walID
 }
 
-// ActiveWALSegment returns the most recent WAL segment.
-func (db *DB) ActiveWALSegment() *WALSegment {
-	db.mu.RLock()
-	defer db.mu.RUnlock()
-	return db.activeWALSegment()
-}
-
-func (db *DB) activeWALSegment() *WALSegment {
-	if len(db.segments) == 0 {
-		return nil
-	}
-	return db.segments[len(db.segments)-1]
-}
-
-// MinWALID returns the lowest WAL ID available in the WAL.
-func (db *DB) MinWALID() int64 {
-	db.mu.RLock()
-	defer db.mu.RUnlock()
-	return db.minWALID()
-}
-
-func (db *DB) minWALID() int64 {
-	if len(db.segments) == 0 {
-		return 0
-	}
-	return db.segments[0].MinWALID()
-}
-
-// MaxWALID returns the highest WAL ID available in the WAL.
-func (db *DB) MaxWALID() int64 {
-	db.mu.RLock()
-	defer db.mu.RUnlock()
-	return db.maxWALID()
-}
-
-func (db *DB) maxWALID() int64 {
-
-	if len(db.segments) == 0 {
-		return 0
-	}
-	s := db.segments[len(db.segments)-1]
-	return s.MaxWALID()
-}
-
-// WALPageN returns the number of pages across all segments.
-func (db *DB) WALPageN() int64 {
-	db.mu.RLock()
-	defer db.mu.RUnlock()
-
-	var n int64
-	for _, s := range db.segments {
-		n += int64(s.PageN())
-	}
-	return n
-}
-
-// SyncWAL flushes the active segment to disk.
-func (db *DB) SyncWAL() error {
-	if s := db.ActiveWALSegment(); s != nil {
-		return s.Sync()
-	}
-	return nil
-}
-
-// readWALPage reads a single page at the given WAL ID.
-func (db *DB) readWALPage(walID int64) ([]byte, error) {
-	// TODO(BBJ): Binary search for segment.
-	for _, s := range db.segments {
-		if walID >= s.MinWALID() && walID <= s.MaxWALID() {
-			return s.ReadWALPage(walID)
-		}
-	}
-	return nil, fmt.Errorf("cannot find segment containing WAL page: %d", walID)
-}
-
-func (db *DB) writeWALPage(page []byte, isMeta bool) (walID int64, err error) {
-	if err := db.ensureWritableWALSegment(); err != nil {
-		return 0, err
-	}
-	return db.activeWALSegment().WriteWALPage(page, isMeta)
-}
-
-func (db *DB) writeBitmapPage(pgno uint32, page []byte) (walID int64, err error) {
-
-	if err := db.ensureWritableWALSegment(); err != nil {
-		return 0, err
-	}
-
-	// Write header page for next bitmap page.
-	buf := make([]byte, PageSize)
-	writePageNo(buf[:], pgno)
-	writeFlags(buf[:], PageTypeBitmapHeader)
-	// TODO(BBJ): Write checksum.
-	if _, err := db.activeWALSegment().WriteWALPage(buf, false); err != nil {
-		return 0, fmt.Errorf("write bitmap header: %w", err)
-	}
-
-	// Write the bitmap page and return its WALID.
-	return db.activeWALSegment().WriteWALPage(page, false)
-}
-
-func (db *DB) ensureWritableWALSegment() error {
-	if s := db.activeWALSegment(); s != nil && s.Size() < MaxWALSegmentFileSize {
-		return nil
-	}
-	return db.addWALSegment()
-}
-
-// addWALSegment appends a new, writable segment and closing an existing segments for write.
-func (db *DB) addWALSegment() error {
-
-	// If we have a current active WAL segment then close it and start the
-	// next segment from the next WAL ID. If there is no existing WAL segments,
-	// read the last checkpointed WAL ID from the DB and start after that.
-	var base int64
-	if s := db.activeWALSegment(); s != nil {
-		base = s.MaxWALID() + 1
-		if err := s.CloseForWrite(); err != nil {
-			return err
-		}
-	} else {
-		page, err := db.readPage(db.pageMap, 0)
-		if err != nil {
-			return err
-		}
-		base = readMetaWALID(page) + 1
-	}
-
-	// Create new segment file.
-	s := NewWALSegment(filepath.Join(db.WALPath(), FormatWALSegmentPath(base)))
-	if err := s.Open(); err != nil {
-		return fmt.Errorf("add wal segment: %w", err)
-	}
-	db.segments = append(db.segments, s)
-
-	return nil
-}
-
 // Close closes the database.
 func (db *DB) Close() (err error) {
-
 	// TODO(bbj): Add wait group to hang until last Tx is complete.
 
 	// Wait for writer lock.
@@ -542,12 +399,12 @@ func (db *DB) Close() (err error) {
 
 // closeWALSegments closes the WAL and all its segments.
 func (db *DB) closeWALSegments() (err error) {
-
 	for _, s := range db.segments {
 		if e := s.Close(); e != nil && err == nil {
 			err = e
 		}
 	}
+	db.segments = nil
 	return err
 }
 
@@ -613,7 +470,6 @@ func (db *DB) HasData(requireOneHotBit bool) (hasAnyRecords bool, err error) {
 
 // Size returns the size of the database & WAL, in bytes.
 func (db *DB) Size() (int64, error) {
-
 	db.mu.RLock()
 	defer db.mu.RUnlock()
 
@@ -621,38 +477,27 @@ func (db *DB) Size() (int64, error) {
 	if err != nil {
 		return 0, err
 	}
-	return db.walSize() + fi.Size(), nil
+	return walSize(db.segments) + fi.Size(), nil
 }
 
 // WALSize returns the size of all WAL segments, in bytes.
 func (db *DB) WALSize() int64 {
-
 	db.mu.RLock()
 	defer db.mu.RUnlock()
-	return db.walSize()
-}
-
-func (db *DB) walSize() int64 {
-
-	var sz int64
-	for _, s := range db.segments {
-		sz += s.Size()
-	}
-	return sz
+	return walSize(db.segments)
 }
 
 // WALSegments returns the WAL segments currently on the DB.
-// This should only be used for debugging & testing purposes.
-func (db *DB) WALSegments() []*WALSegment {
-
+func (db *DB) WALSegments() []WALSegment {
 	db.mu.RLock()
 	defer db.mu.RUnlock()
-	return db.segments
+	other := make([]WALSegment, len(db.segments))
+	copy(other, db.segments)
+	return other
 }
 
 // init initializes a new database file.
 func (db *DB) init() error {
-
 	if err := db.initMetaPage(); err != nil {
 		return fmt.Errorf("meta: %w", err)
 	} else if err := db.initRootRecordPage(); err != nil {
@@ -713,6 +558,10 @@ func (db *DB) BeginWithExclusiveLock() (_ *Tx, err error) {
 }
 
 func (db *DB) begin(writable, exclusive bool) (_ *Tx, err error) {
+	if exclusive {
+		assert(writable) // exclusive transactions must be writable
+	}
+
 	if exclusive {
 		db.exclmu.Lock()
 	} else {
@@ -749,7 +598,7 @@ func (db *DB) begin(writable, exclusive bool) (_ *Tx, err error) {
 	// Flush all WAL writes to disk before an exclusive writer so that we can
 	// work directly with the on-disk database.
 	if exclusive {
-		if err := db.checkpoint(true); err != nil {
+		if err := db.checkpoint(true, &nopLocker{}); err != nil {
 			cleanup()
 			return nil, err
 		}
@@ -762,10 +611,21 @@ func (db *DB) begin(writable, exclusive bool) (_ *Tx, err error) {
 		writable:    writable,
 		exclusive:   exclusive,
 	}
+	if writable {
+		tx.wcache = db.wcache[:0]
+	}
+
+	// Copy list of WAL segments so they can be altered by the tx.
+	// Add last segment to the list of segments that will be updated/added.
+	if len(db.segments) != 0 {
+		tx.segments = make([]WALSegment, len(db.segments))
+		copy(tx.segments, db.segments)
+		tx.updatedSegmentPaths = []string{tx.segments[len(tx.segments)-1].Path}
+	}
 
 	// Copy meta page into transaction's buffer.
 	// This page is only written at the end of a dirty transaction.
-	page, err := db.readPage(db.pageMap, 0)
+	page, err := db.readMetaPage()
 	if err != nil {
 		tx.Rollback()
 		return nil, err
@@ -802,8 +662,8 @@ func (db *DB) removeTx(tx *Tx) error {
 	// Write pages from WAL to DB.
 	// TODO(bbj): Move this to an async goroutine.
 	if tx.writable {
-		if err := db.checkpoint(false); err != nil {
-			return err
+		if err := db.checkpoint(false, &nopLocker{}); err != nil {
+			return fmt.Errorf("checkpoint: %w", err)
 		}
 	}
 
@@ -824,21 +684,25 @@ func (db *DB) Check() error {
 	return tx.Check()
 }
 
-// writePage writes a page to the data file.
-func (db *DB) writePage(pgno uint32, page []byte) error {
+// writeDBPage writes a page to the data file.
+func (db *DB) writeDBPage(pgno uint32, page []byte) error {
 	_, err := db.file.WriteAt(page, int64(pgno)*PageSize)
 	return err
 }
 
-func (db *DB) readPage(pageMap *immutable.Map, pgno uint32) ([]byte, error) {
-	// Check if page is currently in WAL.
-	if pageMap != nil {
-		if walID, ok := pageMap.Get(pgno); ok {
-			return db.readWALPage(walID.(int64))
-		}
-	}
-
-	// Otherwise read from the data file.
+func (db *DB) readDBPage(pgno uint32) ([]byte, error) {
 	offset := int64(pgno) * PageSize
 	return db.data[offset : offset+PageSize], nil
 }
+
+func (db *DB) readMetaPage() ([]byte, error) {
+	if walID, ok := db.pageMap.Get(uint32(0)); ok {
+		return readWALPage(db.segments, walID.(int64))
+	}
+	return db.readDBPage(0)
+}
+
+type nopLocker struct{}
+
+func (*nopLocker) Lock()   {}
+func (*nopLocker) Unlock() {}
diff --git a/rbf/db_test.go b/rbf/db_test.go
index af0552c6f..a06cd26f6 100644
--- a/rbf/db_test.go
+++ b/rbf/db_test.go
@@ -112,10 +112,11 @@ func TestDB_Recovery(t *testing.T) {
 		tx1.Rollback()
 
 		// Close database & truncate WAL to remove commit page & bitmap data page.
-		segment := db.ActiveWALSegment()
+		segments := db.WALSegments()
+		segment := segments[len(segments)-1]
 		if err := db.Close(); err != nil {
 			t.Fatal(err)
-		} else if err := os.Truncate(segment.Path(), segment.Size()-(2*rbf.PageSize)); err != nil {
+		} else if err := os.Truncate(segment.Path, segment.Size()-(2*rbf.PageSize)); err != nil {
 			t.Fatal(err)
 		}
 
diff --git a/rbf/tx.go b/rbf/tx.go
index 5be536221..16e5505d0 100644
--- a/rbf/tx.go
+++ b/rbf/tx.go
@@ -17,9 +17,9 @@ import (
 	"fmt"
 	"io"
 	"math"
+	"os"
+	"path/filepath"
 	"sort"
-
-	//"strconv"
 	"strings"
 	"sync"
 
@@ -33,15 +33,19 @@ var _ = txkey.ToString
 
 // Tx represents a transaction.
 type Tx struct {
-	mu          sync.RWMutex
-	db          *DB            // parent db
-	meta        [PageSize]byte // copy of current meta page
-	walID       int64          // max WAL ID at start of tx
-	rootRecords []*RootRecord  // read-only cache of root records
-	pageMap     *immutable.Map // mapping of database pages to WAL IDs
-	writable    bool           // if true, tx can write
-	exclusive   bool           // if true, tx writes directly to db file (no wal)
-	dirty       bool           // if true, changes have been made
+	mu                  sync.RWMutex
+	db                  *DB            // parent db
+	segments            []WALSegment   // copy of WAL segments
+	updatedSegmentPaths []string       // updated or added segment paths
+	meta                [PageSize]byte // copy of current meta page
+	walID               int64          // max WAL ID at start of tx
+	rootRecords         []*RootRecord  // read-only cache of root records
+	pageMap             *immutable.Map // mapping of database pages to WAL IDs
+	writable            bool           // if true, tx can write
+	exclusive           bool           // if true, tx writes directly to db file (no wal)
+	dirty               bool           // if true, changes have been made
+
+	wcache []byte // write cache
 
 	// If Rollback() has already completed, don't do it again.
 	// Note db == nil means that commit has already been done.
@@ -76,7 +80,7 @@ func (tx *Tx) Commit() error {
 	if tx.dirty {
 		if err := tx.writeMetaPage(MetaPageFlagCommit); err != nil {
 			return err
-		} else if err := tx.db.SyncWAL(); err != nil {
+		} else if err := tx.flushWALWriter(); err != nil {
 			return err
 		}
 
@@ -89,6 +93,11 @@ func (tx *Tx) Commit() error {
 		tx.db.mu.Lock()
 		tx.db.rootRecords = tx.rootRecords
 		tx.db.pageMap = tx.pageMap
+		for _, path := range tx.updatedSegmentPaths {
+			segment := walSegmentByPath(tx.segments, path)
+			assert(segment != nil)
+			tx.db.updateWALSegment(*segment)
+		}
 		tx.db.mu.Unlock()
 	}
 
@@ -112,11 +121,12 @@ func (tx *Tx) Rollback() {
 
 	// TODO(bbj): Invalidate DB if rollback fails. Possibly attempt reopen?
 
-	// Remove all WAL pages that have been written by this transaction.
 	if tx.dirty {
-		if err := tx.db.truncateWALAfter(tx.walID); err != nil {
-			panic(err)
+		if _, err := truncateWALAfter(tx.segments, tx.walID); err != nil {
+			panicOn(err)
 		}
+		tx.segments = nil
+		tx.updatedSegmentPaths = nil
 	}
 
 	// Disconnect transaction from DB.
@@ -915,11 +925,28 @@ func (tx *Tx) readPage(pgno uint32) ([]byte, error) {
 		return tx.meta[:], nil
 	}
 
+	// Verify page number requested is within current size of database.
 	pageN := readMetaPageN(tx.meta[:])
 	if pgno > pageN {
 		return nil, fmt.Errorf("rbf: page read out of bounds: pgno=%d max=%d", pgno, pageN)
 	}
-	return tx.db.readPage(tx.pageMap, pgno)
+
+	// Check if page is remapped.
+	if walID, ok := tx.pageMap.Get(pgno); ok {
+		walID64 := walID.(int64)
+
+		// Read from write cache if not yet flushed to disk.
+		maxWALID := activeWALSegment(tx.segments).MaxWALID()
+		if walID64 > maxWALID {
+			offset := (walID64 - maxWALID - 1) * PageSize
+			return tx.wcache[offset : offset+PageSize], nil
+		}
+
+		// Otherwise return remapped page from WAL segment.
+		return readWALPage(tx.segments, walID64)
+	}
+
+	return tx.db.readDBPage(pgno)
 }
 
 func (tx *Tx) writePage(page []byte) error {
@@ -928,11 +955,11 @@ func (tx *Tx) writePage(page []byte) error {
 
 	// If we are running in exclusive mode, directly write page to database.
 	if tx.exclusive {
-		return tx.db.writePage(readPageNo(page), page)
+		return tx.db.writeDBPage(readPageNo(page), page)
 	}
 
 	// Write page to WAL and obtain position in WAL.
-	walID, err := tx.db.writeWALPage(page, false)
+	walID, err := tx.writeWALPage(page, false)
 	if err != nil {
 		return err
 	}
@@ -948,11 +975,11 @@ func (tx *Tx) writeBitmapPage(pgno uint32, page []byte) error {
 
 	// If we are running in exclusive mode, directly write page to database.
 	if tx.exclusive {
-		return tx.db.writePage(pgno, page)
+		return tx.db.writeDBPage(pgno, page)
 	}
 
 	// Write bitmap to WAL and obtain WAL position of the actual page data (not the prefix page).
-	walID, err := tx.db.writeBitmapPage(pgno, page)
+	walID, err := tx.writeBitmapWALPage(pgno, page)
 	if err != nil {
 		return err
 	}
@@ -968,11 +995,11 @@ func (tx *Tx) writeMetaPage(flag uint32) error {
 
 	// If we are running in exclusive mode, directly write page to database.
 	if tx.exclusive {
-		return tx.db.writePage(0, tx.meta[:])
+		return tx.db.writeDBPage(0, tx.meta[:])
 	}
 
 	// Write page to WAL and obtain position in WAL.
-	walID, err := tx.db.writeWALPage(tx.meta[:], true)
+	walID, err := tx.writeWALPage(tx.meta[:], true)
 	if err != nil {
 		return err
 	}
@@ -1562,3 +1589,114 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear
 	}
 	return
 }
+
+func (tx *Tx) flushWALWriter() error {
+	// Ignore if we have no data in the write cache.
+	if len(tx.wcache) == 0 {
+		return nil
+	}
+
+	// Determine active WAL segment.
+	assert(len(tx.segments) != 0)
+	segment := &tx.segments[len(tx.segments)-1]
+
+	// Open write handle to active segment.
+	w, err := os.OpenFile(segment.Path, os.O_WRONLY, 0666)
+	if err != nil {
+		return fmt.Errorf("open wal segment write handle: %w", err)
+	}
+	defer w.Close()
+
+	// Flush cache to writer.
+	if _, err := w.WriteAt(tx.wcache, int64(segment.PageN)*PageSize); err != nil {
+		return fmt.Errorf("write wal segment: %w", err)
+	} else if err := w.Sync(); err != nil {
+		return fmt.Errorf("sync wal segment: %w", err)
+	} else if err := w.Close(); err != nil {
+		return fmt.Errorf("close wal segment: %w", err)
+	}
+
+	// Increase the size of the last WAL segment & clear cache.
+	assert(len(tx.wcache)%PageSize == 0)
+	segment.PageN += len(tx.wcache) / PageSize
+	tx.wcache = tx.wcache[:0]
+
+	return nil
+}
+
+func (tx *Tx) writeWALPage(page []byte, isMeta bool) (walID int64, err error) {
+	if err := tx.ensureWritableWALSegment(); err != nil {
+		return 0, err
+	}
+
+	// Determine next WAL ID from cached meta page.
+	walID = readMetaWALID(tx.meta[:]) + 1
+
+	// Update WAL ID on cached meta page.
+	writeMetaWALID(tx.meta[:], walID)
+
+	// Append write to write buffer.
+	tx.wcache = append(tx.wcache, page...)
+
+	return walID, nil
+}
+
+func (tx *Tx) writeBitmapWALPage(pgno uint32, page []byte) (walID int64, err error) {
+	if err := tx.ensureWritableWALSegment(); err != nil {
+		return 0, err
+	}
+
+	// Write header page for next bitmap page.
+	buf := make([]byte, PageSize)
+	writePageNo(buf[:], pgno)
+	writeFlags(buf[:], PageTypeBitmapHeader)
+	// TODO(BBJ): Write checksum.
+	if _, err := tx.writeWALPage(buf, false); err != nil {
+		return 0, fmt.Errorf("write bitmap header: %w", err)
+	}
+
+	// Write the bitmap page and return its WALID.
+	return tx.writeWALPage(page, false)
+}
+
+func (tx *Tx) ensureWritableWALSegment() error {
+	writeCacheSize := int64(len(tx.wcache))
+	if len(tx.segments) != 0 && activeWALSegment(tx.segments).Size()+writeCacheSize < MaxWALSegmentFileSize {
+		return nil
+	}
+	return tx.addWALSegment()
+}
+
+// addWALSegment appends a new, writable segment and closing an existing segments for write.
+func (tx *Tx) addWALSegment() error {
+	if err := tx.flushWALWriter(); err != nil {
+		return err
+	}
+
+	// If we have a current active WAL segment then close it and start the
+	// next segment from the next WAL ID. If there is no existing WAL segments,
+	// read the last checkpointed WAL ID from the DB and start after that.
+	var base int64
+	if len(tx.segments) != 0 {
+		base = activeWALSegment(tx.segments).MaxWALID() + 1
+	} else {
+		page, err := tx.readPage(0)
+		if err != nil {
+			return err
+		}
+		base = readMetaWALID(page) + 1
+	}
+
+	// Create new segment file.
+	s := NewWALSegment(filepath.Join(tx.db.WALPath(), FormatWALSegmentPath(base)))
+	if err := s.Open(); err != nil {
+		return fmt.Errorf("add wal segment: %w", err)
+	}
+
+	// Track all segments that need to be added back to DB.
+	// The DB can remove segments in the background so we don't want to replace.
+	tx.segments = append(tx.segments, s)
+	tx.updatedSegmentPaths = append(tx.updatedSegmentPaths, s.Path)
+
+	return nil
+}
diff --git a/rbf/wal.go b/rbf/wal.go
index 7a0518d96..3a26c3bf6 100644
--- a/rbf/wal.go
+++ b/rbf/wal.go
@@ -16,9 +16,9 @@ package rbf
 
 import (
 	"fmt"
+	"io"
 	"os"
 	"path/filepath"
-	"sync"
 	"syscall"
 
 	"github.com/pilosa/pilosa/v2/syswrap"
@@ -26,66 +26,40 @@ import (
 
 // WALSegment represents a single file in the WAL.
 type WALSegment struct {
-	mu         sync.RWMutex
-	minWALID   int64    // base WALID; calculated from path
-	path       string   // path to file
-	w          *os.File // write handle
-	data       []byte   // read-only mmap data
-	writeCache []byte   // write buffer
-	pageN      int      // number of written pages
+	Path     string // path to file
+	MinWALID int64  // base WALID; calculated from path
+	PageN    int    // number of written pages
+
+	data []byte // read-only mmap data
 }
 
 // NewWALSegment returns a new instance of WALSegment for a given path.
-func NewWALSegment(path string) *WALSegment {
-	return &WALSegment{
-		path: path,
+func NewWALSegment(path string) WALSegment {
+	return WALSegment{
+		Path: path,
 	}
 }
 
-// Path returns the path the segment was initialized with.
-func (s *WALSegment) Path() string { return s.path }
-
-// MinWALID returns the initial WAL ID of the segment. Only available after Open().
-func (s *WALSegment) MinWALID() int64 {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-	return s.minWALID
-}
-
 // MaxWALID returns the maximum WAL ID of the segment. Only available after Open().
-func (s *WALSegment) MaxWALID() int64 {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-	return s.minWALID + int64(s.pageN) - 1
-}
-
-// PageN returns the number of pages in the segment.
-func (s *WALSegment) PageN() int {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-	return s.pageN
+func (s WALSegment) MaxWALID() int64 {
+	return s.MinWALID + int64(s.PageN) - 1
 }
 
 // Size returns the current size of the segment, in bytes.
-func (s *WALSegment) Size() int64 {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-	return int64(s.pageN) * PageSize
+func (s WALSegment) Size() int64 {
+	return int64(s.PageN) * PageSize
 }
 
 func (s *WALSegment) Open() (err error) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-
 	// Extract base WAL ID and validate path.
-	if s.minWALID, err = ParseWALSegmentPath(s.path); err != nil {
+	if s.MinWALID, err = ParseWALSegmentPath(s.Path); err != nil {
 		return err
 	}
 
 	// Determine file size & create if necessary.
 	var sz int64
-	if fi, err := os.Stat(s.path); os.IsNotExist(err) {
-		if f, err := os.OpenFile(s.path, os.O_RDWR|os.O_CREATE, 0666); err != nil {
+	if fi, err := os.Stat(s.Path); os.IsNotExist(err) {
+		if f, err := os.OpenFile(s.Path, os.O_RDWR|os.O_CREATE, 0666); err != nil {
 			return fmt.Errorf("touch wal segment file: %w", err)
 		} else if err := f.Close(); err != nil {
 			return fmt.Errorf("close touched wal segment file: %w", err)
@@ -97,10 +71,10 @@ func (s *WALSegment) Open() (err error) {
 	}
 
 	// Determine page count & truncate if a partial page is written.
-	s.pageN = int(sz / PageSize)
+	s.PageN = int(sz / PageSize)
 	if sz%PageSize != 0 {
-		sz = int64(s.pageN * PageSize)
-		if err := os.Truncate(s.path, sz); err != nil {
+		sz = int64(s.PageN * PageSize)
+		if err := os.Truncate(s.Path, sz); err != nil {
 			return fmt.Errorf("truncate wal segment file: %w", err)
 		}
 	}
@@ -113,7 +87,7 @@ func (s *WALSegment) Open() (err error) {
 	}
 
 	// Open file as a read-only memory map.
-	if f, err := os.OpenFile(s.path, os.O_RDONLY, 0666); err != nil {
+	if f, err := os.OpenFile(s.Path, os.O_RDONLY, 0666); err != nil {
 		return fmt.Errorf("open wal segment file: %w", err)
 	} else if s.data, err = syswrap.Mmap(int(f.Fd()), 0, int(mmapSize), syscall.PROT_READ, syscall.MAP_SHARED); err != nil {
 		f.Close()
@@ -127,12 +101,6 @@ func (s *WALSegment) Open() (err error) {
 
 // Close closes the write handle and the read-only mmap.
 func (s *WALSegment) Close() error {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-
-	if err := s.closeForWrite(); err != nil {
-		return err
-	}
 	if s.data != nil {
 		if err := syswrap.Munmap(s.data); err != nil {
 			return err
@@ -142,144 +110,141 @@ func (s *WALSegment) Close() error {
 	return nil
 }
 
-// CloseForWrite closes the write handle, if initialized.
-func (s *WALSegment) CloseForWrite() error {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	return s.closeForWrite()
-}
-
-func (s *WALSegment) closeForWrite() error {
-	// Ensure write buffer is flushed out.
-	if err := s.sync(); err != nil {
-		return err
-	}
-	s.writeCache = nil
-
-	// Close underlying file writer.
-	if s.w != nil {
-		if err := s.w.Close(); err != nil {
-			return err
-		}
-		s.w = nil
-	}
-	return nil
-}
-
 // ReadWALPage reads a single page at the given WAL ID.
 func (s *WALSegment) ReadWALPage(walID int64) ([]byte, error) {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-
 	// Ensure requested ID is contained in this file.
-	if walID < s.minWALID || walID > s.minWALID+int64(s.pageN) {
-		return nil, fmt.Errorf("wal segment page read out of range: id=%d base=%d pageN=%d", walID, s.minWALID, s.pageN)
+	if walID < s.MinWALID || walID > s.MinWALID+int64(s.PageN) {
+		return nil, fmt.Errorf("wal segment page read out of range: id=%d base=%d pageN=%d", walID, s.MinWALID, s.PageN)
 	}
 
-	offset := (walID - s.minWALID) * PageSize
-
-	// If offset is within write buffer, return from write buffer.
-	writeBufferOffset := int64((s.pageN * PageSize) - len(s.writeCache))
-	if offset >= writeBufferOffset {
-		buf := s.writeCache[offset-writeBufferOffset:]
-		return buf[:PageSize:PageSize], nil
-	}
-
-	// Otherwise return from on-disk mmap.
+	offset := (walID - s.MinWALID) * PageSize
 	return s.data[offset : offset+PageSize], nil
 }
 
-// WriteWALPage writes a single page to the WAL segment and returns its WAL identifier.
-func (s *WALSegment) WriteWALPage(page []byte, isMeta bool) (walID int64, err error) {
-	assert(len(page) == PageSize) // invalid page size
-
-	s.mu.Lock()
-	defer s.mu.Unlock()
-
-	// Initialize write file handle if not yet initialized.
-	if s.w == nil {
-		if s.w, err = os.OpenFile(s.path, os.O_WRONLY, 0666); err != nil {
-			return 0, fmt.Errorf("open wal segment write handle: %w", err)
+func walSegmentByPath(segments []WALSegment, path string) *WALSegment {
+	for _, segment := range segments {
+		if segment.Path == path {
+			return &segment
 		}
 	}
-
-	// Determine current WAL position.
-	walID = s.minWALID + int64(s.pageN)
-
-	// Write WAL ID if this is a meta page.
-	if isMeta {
-		writeMetaWALID(page, walID)
-		// TODO: Write meta page checksum
-	}
-
-	// Append write to write buffer & increment page count.
-	if s.writeCache == nil {
-		s.writeCache = make([]byte, 0, MaxWALSegmentFileSize+PageSize)
-	}
-	s.writeCache = append(s.writeCache, page...)
-	s.pageN++
-
-	return walID, nil
-}
-
-// TruncateAfter removes all pages after a given WAL ID.
-func (s *WALSegment) TruncateAfter(walID int64) error {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-
-	// Ensure this is a partial truncation. Full truncation of a segment
-	// should be performed by the DB since it needs to remove the segment.
-	assert(walID > s.minWALID)
-
-	// Update to new page size.
-	newPageN := int((walID - s.minWALID) + 1) // new page count of segment
-	truncPageN := s.pageN - newPageN          // number of pages removed
-	s.pageN = newPageN
-
-	// Check to see if we are only truncating from the write cache.
-	writeCachePageN := len(s.writeCache) / PageSize
-	if truncPageN <= int(writeCachePageN) {
-		s.writeCache = s.writeCache[:(writeCachePageN-truncPageN)*PageSize]
-		return nil
-	}
-
-	// Clear write cache.
-	s.writeCache = s.writeCache[:0]
-
-	// Remove on disk pages.
-	return os.Truncate(s.path, int64(s.pageN)*PageSize)
-}
-
-// Flush flushes the write buffer to the OS cache.
-func (s *WALSegment) Flush() error {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	return s.flush()
-}
-
-func (s *WALSegment) flush() error {
-	if _, err := s.w.WriteAt(s.writeCache, int64((s.pageN*PageSize)-len(s.writeCache))); err != nil {
-		return fmt.Errorf("wal segment write: %w", err)
-	}
-	s.writeCache = s.writeCache[:0]
 	return nil
 }
 
-// Sync flushes the write buffer and invokes a file sync to flush data to disk.
-func (s *WALSegment) Sync() error {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	return s.sync()
+func activeWALSegment(segments []WALSegment) WALSegment {
+	if len(segments) == 0 {
+		return WALSegment{}
+	}
+	return segments[len(segments)-1]
 }
 
-func (s *WALSegment) sync() error {
-	if s.w == nil {
-		return nil
+func minWALID(segments []WALSegment) int64 {
+	if len(segments) == 0 {
+		return 0
 	}
-	if err := s.flush(); err != nil {
-		return err
+	return segments[0].MinWALID
+}
+
+func maxWALID(segments []WALSegment) int64 {
+	if len(segments) == 0 {
+		return 0
 	}
-	return s.w.Sync()
+	s := segments[len(segments)-1]
+	return s.MaxWALID()
+}
+
+func walSize(segments []WALSegment) int64 {
+	var sz int64
+	for _, s := range segments {
+		sz += s.Size()
+	}
+	return sz
+}
+
+// readWALPage reads a single page at the given WAL ID.
+func readWALPage(segments []WALSegment, walID int64) ([]byte, error) {
+	// TODO(BBJ): Binary search for segment.
+	for _, s := range segments {
+		if walID >= s.MinWALID && walID <= s.MaxWALID() {
+			return s.ReadWALPage(walID)
+		}
+	}
+	return nil, fmt.Errorf("cannot find segment containing WAL page: %d", walID)
+}
+
+func findNextWALMetaPage(segments []WALSegment, walID int64) (metaWALID int64, err error) {
+	maxWALID := maxWALID(segments)
+
+	for ; walID <= maxWALID; walID++ {
+		// Read page data from WAL and return if it is a meta page.
+		page, err := readWALPage(segments, walID)
+		if err != nil {
+			return walID, err
+		} else if IsMetaPage(page) {
+			return walID, nil
+		}
+
+		// Skip over next page if this is a bitmap header.
+		if IsBitmapHeader(page) {
+			walID++
+		}
+	}
+
+	return -1, io.EOF
+}
+
+func findLastWALMetaPage(segments []WALSegment) (walID int64, err error) {
+	if len(segments) == 0 {
+		return 0, nil
+	}
+
+	var maxMetaWALID int64
+	maxWALID := maxWALID(segments)
+	for walID := minWALID(segments); walID <= maxWALID; walID++ {
+		if page, err := readWALPage(segments, walID); err != nil {
+			return walID, err
+		} else if IsBitmapHeader(page) {
+			walID++ // skip next page for bitmap headers
+		} else if IsMetaPage(page) {
+			maxMetaWALID = walID // save max meta WAL ID
+		}
+	}
+	return maxMetaWALID, nil
+}
+
+// truncateWALAfter removes all pages in the WAL after walID.
+func truncateWALAfter(segments []WALSegment, walID int64) ([]WALSegment, error) {
+	var newSegments []WALSegment
+
+	for i := range segments {
+		segment := &segments[i]
+
+		// Append entire segment if WAL range entirely before target WAL ID.
+		if walID > segment.MaxWALID() {
+			newSegments = append(newSegments, *segment)
+			continue
+		}
+
+		// If we only remove some of the WAL pages then truncate and append.
+		if segment.MinWALID < walID {
+			newSegment := *segment
+			newSegment.PageN = int((walID - newSegment.MinWALID) + 1)
+
+			if err := os.Truncate(newSegment.Path, int64(newSegment.PageN)*PageSize); err != nil {
+				return segments, err
+			}
+			newSegments = append(newSegments, newSegment)
+			continue
+		}
+
+		// Drop entire segment if all pages are after WAL ID.
+		if err := segment.Close(); err != nil {
+			return segments, err
+		} else if err := os.Remove(segment.Path); err != nil {
+			return segments, err
+		}
+	}
+
+	return newSegments, nil
 }
 
 // FormatWALSegmentPath returns a path for a WAL segment using a WAL ID.
diff --git a/rbf/wal_test.go b/rbf/wal_test.go
index b41e7e792..59463d435 100644
--- a/rbf/wal_test.go
+++ b/rbf/wal_test.go
@@ -15,10 +15,10 @@
 package rbf_test
 
 import (
-	"bytes"
-	"encoding/hex"
+	// "bytes"
+	// "encoding/hex"
 	"io/ioutil"
-	"math/rand"
+	// "math/rand"
 	"os"
 	"path/filepath"
 	"testing"
@@ -30,9 +30,9 @@ func TestWALSegment_Open(t *testing.T) {
 	t.Run("OK", func(t *testing.T) {
 		s := MustOpenWALSegment(t, 10)
 		defer MustCloseWALSegment(t, s)
-		if got, want := s.MinWALID(), int64(10); got != want {
+		if got, want := s.MinWALID, int64(10); got != want {
 			t.Fatalf("Base()=%d, want %d", got, want)
-		} else if got, want := s.PageN(), 0; got != want {
+		} else if got, want := s.PageN, 0; got != want {
 			t.Fatalf("PageN()=%d, want %d", got, want)
 		}
 	})
@@ -40,6 +40,7 @@ func TestWALSegment_Open(t *testing.T) {
 	// TODO(BBJ): Test open w/ partially written pages.
 }
 
+/*
 func TestWALSegment_WritePage(t *testing.T) {
 	rand := rand.New(rand.NewSource(0))
 	s := MustOpenWALSegment(t, 10)
@@ -84,6 +85,7 @@ func TestWALSegment_WritePage(t *testing.T) {
 		t.Fatal("unexpected second page")
 	}
 }
+*/
 
 func TestFormatWALSegmentPath(t *testing.T) {
 	if got, want := rbf.FormatWALSegmentPath(1234), "00000000000004d2.wal"; got != want {
@@ -107,6 +109,7 @@ func TestParseWALSegmentPath(t *testing.T) {
 	})
 }
 
+/*
 func BenchmarkWALSegment_WriteWALPage(b *testing.B) {
 	b.Run("8KB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, 8*(1<<10)) })
 	b.Run("16KB", func(b *testing.B) { benchmarkWALSegment_WriteWALPage(b, 16*(1<<10)) })
@@ -147,9 +150,10 @@ func benchmarkWALSegment_WriteWALPage(b *testing.B, flushSize int) {
 
 	b.SetBytes(rbf.MaxWALSegmentFileSize)
 }
+*/
 
 // MustOpenWALSegment opens a WAL segment in a temporary path. Fails on error.
-func MustOpenWALSegment(tb testing.TB, walID int64) *rbf.WALSegment {
+func MustOpenWALSegment(tb testing.TB, walID int64) rbf.WALSegment {
 	tb.Helper()
 
 	dir, err := ioutil.TempDir("", "")
@@ -169,11 +173,11 @@ func MustOpenWALSegment(tb testing.TB, walID int64) *rbf.WALSegment {
 }
 
 // MustCloseWALSegment closes s. Fails on error.
-func MustCloseWALSegment(tb testing.TB, s *rbf.WALSegment) {
+func MustCloseWALSegment(tb testing.TB, s rbf.WALSegment) {
 	tb.Helper()
 	if err := s.Close(); err != nil {
 		tb.Fatal(err)
-	} else if err := os.Remove(s.Path()); err != nil {
+	} else if err := os.Remove(s.Path); err != nil {
 		tb.Fatal(err)
 	}
 }

From d826cec4b395a77db7d026ea24f528b41317b0f1 Mon Sep 17 00:00:00 2001
From: Ben Johnson <benbjohnson@yahoo.com>
Date: Mon, 28 Sep 2020 13:57:33 -0600
Subject: [PATCH 02/12] fix RBF WAL segment reference error

---
 rbf/tx.go  |  6 ++----
 rbf/wal.go | 13 ++++++++++---
 2 files changed, 12 insertions(+), 7 deletions(-)

diff --git a/rbf/tx.go b/rbf/tx.go
index 16e5505d0..98bd999d4 100644
--- a/rbf/tx.go
+++ b/rbf/tx.go
@@ -1660,15 +1660,13 @@ func (tx *Tx) writeBitmapWALPage(pgno uint32, page []byte) (walID int64, err err
 }
 
 func (tx *Tx) ensureWritableWALSegment() error {
+	// Ignore if we still have space in the write cache.
 	writeCacheSize := int64(len(tx.wcache))
 	if len(tx.segments) != 0 && activeWALSegment(tx.segments).Size()+writeCacheSize < MaxWALSegmentFileSize {
 		return nil
 	}
-	return tx.addWALSegment()
-}
 
-// addWALSegment appends a new, writable segment and closing an existing segments for write.
-func (tx *Tx) addWALSegment() error {
+	// Flush write cache out to file before adding new segment.
 	if err := tx.flushWALWriter(); err != nil {
 		return err
 	}
diff --git a/rbf/wal.go b/rbf/wal.go
index 3a26c3bf6..bb9b53157 100644
--- a/rbf/wal.go
+++ b/rbf/wal.go
@@ -122,9 +122,9 @@ func (s *WALSegment) ReadWALPage(walID int64) ([]byte, error) {
 }
 
 func walSegmentByPath(segments []WALSegment, path string) *WALSegment {
-	for _, segment := range segments {
-		if segment.Path == path {
-			return &segment
+	for i := range segments {
+		if segments[i].Path == path {
+			return &segments[i]
 		}
 	}
 	return nil
@@ -247,6 +247,13 @@ func truncateWALAfter(segments []WALSegment, walID int64) ([]WALSegment, error)
 	return newSegments, nil
 }
 
+func DumpWALSegments(segments []WALSegment) {
+	fmt.Printf("WAL (%d segments)\n", len(segments))
+	for i, s := range segments {
+		fmt.Printf("[%d] WALIDs=(%d-%d) PageN=%d\n", i, s.MinWALID, s.MaxWALID(), s.PageN)
+	}
+}
+
 // FormatWALSegmentPath returns a path for a WAL segment using a WAL ID.
 func FormatWALSegmentPath(walID int64) string {
 	return fmt.Sprintf("%016x.wal", walID)

From 3429421148ce8f6e27ae1a4db789b294cb21784f Mon Sep 17 00:00:00 2001
From: Ben Johnson <benbjohnson@yahoo.com>
Date: Mon, 28 Sep 2020 14:46:27 -0600
Subject: [PATCH 03/12] Fix RBF checkpoint off-by-one WAL ID issue

---
 rbf/db.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/rbf/db.go b/rbf/db.go
index e807956ec..737d7793b 100644
--- a/rbf/db.go
+++ b/rbf/db.go
@@ -250,7 +250,7 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error {
 
 		// Loop over pages in the transaction.
 		for ; walID <= metaWALID; walID++ {
-			canCheckpoint := exclusive || minActiveWALID == 0 || walID <= minActiveWALID
+			canCheckpoint := exclusive || minActiveWALID == 0 || walID < minActiveWALID
 
 			page, err := readWALPage(segments, walID)
 			if err != nil {

From c51ba69c09051acdbd1dec034d69f74c3041db4c Mon Sep 17 00:00:00 2001
From: Ben Johnson <benbjohnson@yahoo.com>
Date: Wed, 30 Sep 2020 08:39:46 -0600
Subject: [PATCH 04/12] Fix RBF sync calls

---
 rbf/db.go  |  5 +++++
 rbf/rbf.go | 16 ++++++++++++++++
 rbf/wal.go |  6 +++---
 3 files changed, 24 insertions(+), 3 deletions(-)

diff --git a/rbf/db.go b/rbf/db.go
index 737d7793b..9ff40a204 100644
--- a/rbf/db.go
+++ b/rbf/db.go
@@ -294,6 +294,11 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error {
 		}
 	}
 
+	// Ensure WAL pages are fully copied & synced to DB file.
+	if err := db.file.Sync(); err != nil {
+		return fmt.Errorf("db file sync: %w", err)
+	}
+
 	// Remove WAL segments that have been checkpointed.
 	if maxCheckpointedWALID != 0 {
 		for _, segment := range segments {
diff --git a/rbf/rbf.go b/rbf/rbf.go
index f152e3499..a512fe237 100644
--- a/rbf/rbf.go
+++ b/rbf/rbf.go
@@ -648,3 +648,19 @@ func RowValues(b []uint64) []uint64 {
 // 	_, file, line, _ := runtime.Caller(skip + 1)
 // 	return fmt.Sprintf("%s:%d", file, line)
 // }
+
+// truncate truncates the file at path to sz bytes. File must exist.
+func truncate(path string, sz int64) error {
+	f, err := os.OpenFile(path, os.O_WRONLY, 0666)
+	if err != nil {
+		return fmt.Errorf("open file: %w", err)
+	}
+	defer f.Close()
+
+	if err := f.Truncate(sz); err != nil {
+		return fmt.Errorf("truncate: %w", err)
+	} else if err := f.Sync(); err != nil {
+		return fmt.Errorf("sync: %w", err)
+	}
+	return f.Close()
+}
diff --git a/rbf/wal.go b/rbf/wal.go
index bb9b53157..14e64a107 100644
--- a/rbf/wal.go
+++ b/rbf/wal.go
@@ -74,8 +74,8 @@ func (s *WALSegment) Open() (err error) {
 	s.PageN = int(sz / PageSize)
 	if sz%PageSize != 0 {
 		sz = int64(s.PageN * PageSize)
-		if err := os.Truncate(s.Path, sz); err != nil {
-			return fmt.Errorf("truncate wal segment file: %w", err)
+		if err := truncate(s.Path, sz); err != nil {
+			return fmt.Errorf("truncate wal file: %w", err)
 		}
 	}
 
@@ -229,7 +229,7 @@ func truncateWALAfter(segments []WALSegment, walID int64) ([]WALSegment, error)
 			newSegment := *segment
 			newSegment.PageN = int((walID - newSegment.MinWALID) + 1)
 
-			if err := os.Truncate(newSegment.Path, int64(newSegment.PageN)*PageSize); err != nil {
+			if err := truncate(newSegment.Path, int64(newSegment.PageN)*PageSize); err != nil {
 				return segments, err
 			}
 			newSegments = append(newSegments, newSegment)

From c978e2242edb1b6d47876b91272845261f7ed00f Mon Sep 17 00:00:00 2001
From: Ben Johnson <benbjohnson@yahoo.com>
Date: Wed, 30 Sep 2020 09:51:16 -0600
Subject: [PATCH 05/12] Skip some RBF tests during race detection

---
 fragment_internal_test.go           |  8 ++++++--
 rbf/rbf_norace.go => race/norace.go |  6 +++---
 rbf/rbf_race.go => race/race.go     |  6 +++---
 rbf/cursor_test.go                  |  9 +++++----
 rbf/tx_test.go                      |  5 +++--
 server/cluster_test.go              |  8 ++++++--
 server/server_test.go               | 11 +++++++++++
 7 files changed, 37 insertions(+), 16 deletions(-)
 rename rbf/rbf_norace.go => race/norace.go (86%)
 rename rbf/rbf_race.go => race/race.go (87%)

diff --git a/fragment_internal_test.go b/fragment_internal_test.go
index b33862d54..a47f9fa4b 100644
--- a/fragment_internal_test.go
+++ b/fragment_internal_test.go
@@ -33,13 +33,13 @@ import (
 	"testing"
 	"testing/quick"
 
-	"golang.org/x/sync/errgroup"
-
 	"github.com/davecgh/go-spew/spew"
 	"github.com/pilosa/pilosa/v2/pql"
+	"github.com/pilosa/pilosa/v2/race"
 	"github.com/pilosa/pilosa/v2/roaring"
 	"github.com/pilosa/pilosa/v2/testhook"
 	"github.com/pkg/errors"
+	"golang.org/x/sync/errgroup"
 )
 
 // Test flags
@@ -450,6 +450,10 @@ func TestFragment_SetValue(t *testing.T) {
 	})
 
 	t.Run("QuickCheck", func(t *testing.T) {
+		if os.Getenv("PILOSA_TXSRC") == "rbf" && race.Enabled {
+			t.Skip("race detection enabled, skipping for rbf")
+		}
+
 		if err := quick.Check(func(bitDepth uint, bitN uint64, values []uint64) bool {
 			// Limit bit depth & maximum values.
 			bitDepth = (bitDepth % 62) + 1
diff --git a/rbf/rbf_norace.go b/race/norace.go
similarity index 86%
rename from rbf/rbf_norace.go
rename to race/norace.go
index 15667ee93..6626c8380 100644
--- a/rbf/rbf_norace.go
+++ b/race/norace.go
@@ -14,7 +14,7 @@
 
 // +build !race
 
-package rbf
+package race
 
-// RaceEnabled is true if the -race flag is enabled.
-const RaceEnabled = false
+// Enabled is true if the -race flag is enabled.
+const Enabled = false
diff --git a/rbf/rbf_race.go b/race/race.go
similarity index 87%
rename from rbf/rbf_race.go
rename to race/race.go
index e178a736b..a6fe765bc 100644
--- a/rbf/rbf_race.go
+++ b/race/race.go
@@ -14,7 +14,7 @@
 
 // +build race
 
-package rbf
+package race
 
-// RaceEnabled is true if the -race flag is enabled.
-const RaceEnabled = true
+// Enabled is true if the -race flag is enabled.
+const Enabled = true
diff --git a/rbf/cursor_test.go b/rbf/cursor_test.go
index 40f2756f0..401117d06 100644
--- a/rbf/cursor_test.go
+++ b/rbf/cursor_test.go
@@ -23,6 +23,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/pilosa/pilosa/v2/race"
 	"github.com/pilosa/pilosa/v2/rbf"
 	"github.com/pilosa/pilosa/v2/roaring"
 )
@@ -76,7 +77,7 @@ func TestCursor_FirstNext_Quick(t *testing.T) {
 		t.Skip("-short enabled, skipping")
 	} else if is32Bit() {
 		t.Skip("32-bit build, skipping quick check tests")
-	} else if rbf.RaceEnabled {
+	} else if race.Enabled {
 		t.Skip("race detection enabled, skipping")
 	}
 
@@ -198,7 +199,7 @@ func TestCursor_LastPrev_Quick(t *testing.T) {
 		t.Skip("-short enabled, skipping")
 	} else if is32Bit() {
 		t.Skip("32-bit build, skipping quick check tests")
-	} else if rbf.RaceEnabled {
+	} else if race.Enabled {
 		t.Skip("race detection enabled, skipping")
 	}
 
@@ -312,7 +313,7 @@ func TestCursor_Union(t *testing.T) {
 			t.Skip("-short enabled, skipping")
 		} else if is32Bit() {
 			t.Skip("32-bit build, skipping quick check tests")
-		} else if rbf.RaceEnabled {
+		} else if race.Enabled {
 			t.Skip("race detection enabled, skipping")
 		}
 
@@ -393,7 +394,7 @@ func TestCursor_Intersect(t *testing.T) {
 			t.Skip("-short enabled, skipping")
 		} else if is32Bit() {
 			t.Skip("32-bit build, skipping quick check tests")
-		} else if rbf.RaceEnabled {
+		} else if race.Enabled {
 			t.Skip("race detection enabled, skipping")
 		}
 
diff --git a/rbf/tx_test.go b/rbf/tx_test.go
index defd0ac23..7e723aa8a 100644
--- a/rbf/tx_test.go
+++ b/rbf/tx_test.go
@@ -21,6 +21,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pilosa/pilosa/v2/race"
 	"github.com/pilosa/pilosa/v2/rbf"
 	"github.com/pilosa/pilosa/v2/txkey"
 )
@@ -218,7 +219,7 @@ func TestTx_Add_Quick(t *testing.T) {
 		t.Skip("-short enabled, skipping")
 	} else if is32Bit() {
 		t.Skip("32-bit build, skipping quick check tests")
-	} else if rbf.RaceEnabled {
+	} else if race.Enabled {
 		t.Skip("race detection enabled, skipping")
 	}
 
@@ -257,7 +258,7 @@ func TestTx_AddRemove_Quick(t *testing.T) {
 		t.Skip("-short enabled, skipping")
 	} else if is32Bit() {
 		t.Skip("32-bit build, skipping quick check tests")
-	} else if rbf.RaceEnabled {
+	} else if race.Enabled {
 		t.Skip("race detection enabled, skipping")
 	}
 
diff --git a/server/cluster_test.go b/server/cluster_test.go
index f1a73a921..7b17a253d 100644
--- a/server/cluster_test.go
+++ b/server/cluster_test.go
@@ -25,9 +25,9 @@ import (
 	"testing"
 	"time"
 
-	"github.com/pilosa/pilosa/v2/server"
-
 	"github.com/pilosa/pilosa/v2"
+	"github.com/pilosa/pilosa/v2/race"
+	"github.com/pilosa/pilosa/v2/server"
 	"github.com/pilosa/pilosa/v2/test"
 	"golang.org/x/sync/errgroup"
 )
@@ -634,6 +634,10 @@ func TestCluster_GossipMembership(t *testing.T) {
 }
 
 func TestClusterResize_RemoveNode(t *testing.T) {
+	if os.Getenv("PILOSA_TXSRC") == "rbf" && race.Enabled {
+		t.Skip("race detection enabled, skipping for rbf")
+	}
+
 	cluster := test.MustRunCluster(t, 3)
 	defer cluster.Close()
 	m0 := cluster.GetNode(0)
diff --git a/server/server_test.go b/server/server_test.go
index 270fdcaf5..02af30e59 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -34,6 +34,7 @@ import (
 	"github.com/pilosa/pilosa/v2"
 	"github.com/pilosa/pilosa/v2/http"
 	"github.com/pilosa/pilosa/v2/pql"
+	"github.com/pilosa/pilosa/v2/race"
 	"github.com/pilosa/pilosa/v2/roaring"
 	"github.com/pilosa/pilosa/v2/server"
 	"github.com/pilosa/pilosa/v2/test"
@@ -51,6 +52,8 @@ func init() { // nolint: gochecknoinits
 func TestMain_Set_Quick(t *testing.T) {
 	if testing.Short() {
 		t.Skip("short")
+	} else if os.Getenv("PILOSA_TXSRC") == "rbf" && race.Enabled {
+		t.Skip("race detection enabled, skipping for rbf")
 	}
 
 	for i := 0; i < 100; i++ {
@@ -947,6 +950,10 @@ func TestMain_ImportTimestampNoStandardView(t *testing.T) {
 }
 
 func TestClusterQueriesAfterRestart(t *testing.T) {
+	if os.Getenv("PILOSA_TXSRC") == "rbf" && race.Enabled {
+		t.Skip("race detection enabled, skipping for rbf")
+	}
+
 	cluster := test.MustRunCluster(t, 3)
 	defer cluster.Close()
 	cmd1 := cluster.GetNode(1)
@@ -1171,6 +1178,10 @@ func TestClusterExhaustingConnectionsImport(t *testing.T) {
 }
 
 func TestClusterMinMaxSumDecimal(t *testing.T) {
+	if os.Getenv("PILOSA_TXSRC") == "rbf" && race.Enabled {
+		t.Skip("race detection enabled, skipping for rbf")
+	}
+
 	cluster := test.MustRunCluster(t, 3)
 	defer cluster.Close()
 	cmd := cluster.GetNode(0)

From 3852ac79c4eac7c6df7a7d7124e18be627a0c25d Mon Sep 17 00:00:00 2001
From: Ben Johnson <benbjohnson@yahoo.com>
Date: Thu, 1 Oct 2020 09:47:32 -0600
Subject: [PATCH 06/12] Add rbf.DB.TxN() function and test check

---
 rbf/db.go       | 7 +++++++
 rbf/rbf_test.go | 2 ++
 2 files changed, 9 insertions(+)

diff --git a/rbf/db.go b/rbf/db.go
index 9ff40a204..5bbf117e9 100644
--- a/rbf/db.go
+++ b/rbf/db.go
@@ -100,6 +100,13 @@ func CreateDirIfNotExist(path string) {
 	}
 }
 
+// TxN returns the number of active transactions.
+func (db *DB) TxN() int {
+	db.mu.RLock()
+	defer db.mu.RUnlock()
+	return len(db.txs)
+}
+
 // Open opens a database with the file specified in Path.
 // Creates a new file if one does not already exist.
 func (db *DB) Open() (err error) {
diff --git a/rbf/rbf_test.go b/rbf/rbf_test.go
index 8b4768cd5..e4ea450e2 100644
--- a/rbf/rbf_test.go
+++ b/rbf/rbf_test.go
@@ -86,6 +86,8 @@ 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 {
+		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)
 	} else if err := os.RemoveAll(db.Path); err != nil {

From cf208cfaa32d7c2c64472cedc5140cae42f197bc Mon Sep 17 00:00:00 2001
From: Ben Johnson <benbjohnson@yahoo.com>
Date: Fri, 2 Oct 2020 09:53:05 -0600
Subject: [PATCH 07/12] Increase test http client timeout

The timeout was increased to allow additional time for RBF to process
a lot of individual `Set()` commands in `TestMain_RecalculateHashes`.
---
 test/pilosa.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/test/pilosa.go b/test/pilosa.go
index be0aaa1fc..d7fdf7aaa 100644
--- a/test/pilosa.go
+++ b/test/pilosa.go
@@ -302,7 +302,7 @@ func Do(t *testing.T, method, urlStr string, body string) *httpResponse {
 	// set a timeout instead of allowing gohttp.Defaultclient to
 	// potentially hang forever.
 	hc := &gohttp.Client{
-		Timeout: time.Second * 10,
+		Timeout: time.Second * 30,
 	}
 	resp, err := hc.Do(req)
 

From 265452cf3dcac45662a0c189cf02ab0ddb7c24d4 Mon Sep 17 00:00:00 2001
From: Ben Johnson <benbjohnson@yahoo.com>
Date: Mon, 5 Oct 2020 10:14:03 -0600
Subject: [PATCH 08/12] Add rbf.SyncEnabled

---
 rbf/db.go  |  2 +-
 rbf/rbf.go | 14 +++++++++++++-
 rbf/tx.go  |  2 +-
 3 files changed, 15 insertions(+), 3 deletions(-)

diff --git a/rbf/db.go b/rbf/db.go
index 5bbf117e9..970c66b0c 100644
--- a/rbf/db.go
+++ b/rbf/db.go
@@ -302,7 +302,7 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error {
 	}
 
 	// Ensure WAL pages are fully copied & synced to DB file.
-	if err := db.file.Sync(); err != nil {
+	if err := fsync(db.file); err != nil {
 		return fmt.Errorf("db file sync: %w", err)
 	}
 
diff --git a/rbf/rbf.go b/rbf/rbf.go
index a512fe237..b504e4fbc 100644
--- a/rbf/rbf.go
+++ b/rbf/rbf.go
@@ -91,6 +91,11 @@ var (
 // Debug is just a temporary flag used for debugging.
 var Debug bool
 
+// Testing constants.
+const (
+	SyncEnabled = true
+)
+
 // Magic32 returns the magic bytes as a big endian encoded uint32.
 func Magic32() uint32 {
 	return binary.BigEndian.Uint32([]byte(Magic))
@@ -659,8 +664,15 @@ func truncate(path string, sz int64) error {
 
 	if err := f.Truncate(sz); err != nil {
 		return fmt.Errorf("truncate: %w", err)
-	} else if err := f.Sync(); err != nil {
+	} else if err := fsync(f); err != nil {
 		return fmt.Errorf("sync: %w", err)
 	}
 	return f.Close()
 }
+
+func fsync(f *os.File) error {
+	if !SyncEnabled {
+		return nil
+	}
+	return f.Sync()
+}
diff --git a/rbf/tx.go b/rbf/tx.go
index 98bd999d4..c2647b20f 100644
--- a/rbf/tx.go
+++ b/rbf/tx.go
@@ -1610,7 +1610,7 @@ func (tx *Tx) flushWALWriter() error {
 	// Flush cache to writer.
 	if _, err := w.WriteAt(tx.wcache, int64(segment.PageN)*PageSize); err != nil {
 		return fmt.Errorf("write wal segment: %w", err)
-	} else if err := w.Sync(); err != nil {
+	} else if err := fsync(w); err != nil {
 		return fmt.Errorf("sync wal segment: %w", err)
 	} else if err := w.Close(); err != nil {
 		return fmt.Errorf("close wal segment: %w", err)

From 644969e6a94cdbd231108dac7a3ea28a12c9e5a6 Mon Sep 17 00:00:00 2001
From: Ben Johnson <benbjohnson@yahoo.com>
Date: Tue, 6 Oct 2020 09:27:48 -0600
Subject: [PATCH 09/12] Add log.* and tourna.* to gitignore

---
 .gitignore | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/.gitignore b/.gitignore
index bee19753e..84fdc73f4 100644
--- a/.gitignore
+++ b/.gitignore
@@ -7,3 +7,5 @@ build
 *~
 lattice
 release-pilosa-fsck.*.*.tar.gz
+/log.*
+/tourna.log.*
\ No newline at end of file

From 7b64abbb53d109df1ea93259811d1799fe522eed Mon Sep 17 00:00:00 2001
From: Ben Johnson <benbjohnson@yahoo.com>
Date: Tue, 6 Oct 2020 09:34:17 -0600
Subject: [PATCH 10/12] Reduce default RBF DB size to 4GB; remove race skips

---
 fragment_internal_test.go |  5 -----
 rbf/os.go                 |  2 +-
 server/cluster_test.go    |  5 -----
 server/server_test.go     | 11 -----------
 4 files changed, 1 insertion(+), 22 deletions(-)

diff --git a/fragment_internal_test.go b/fragment_internal_test.go
index a47f9fa4b..8514f7284 100644
--- a/fragment_internal_test.go
+++ b/fragment_internal_test.go
@@ -35,7 +35,6 @@ import (
 
 	"github.com/davecgh/go-spew/spew"
 	"github.com/pilosa/pilosa/v2/pql"
-	"github.com/pilosa/pilosa/v2/race"
 	"github.com/pilosa/pilosa/v2/roaring"
 	"github.com/pilosa/pilosa/v2/testhook"
 	"github.com/pkg/errors"
@@ -450,10 +449,6 @@ func TestFragment_SetValue(t *testing.T) {
 	})
 
 	t.Run("QuickCheck", func(t *testing.T) {
-		if os.Getenv("PILOSA_TXSRC") == "rbf" && race.Enabled {
-			t.Skip("race detection enabled, skipping for rbf")
-		}
-
 		if err := quick.Check(func(bitDepth uint, bitN uint64, values []uint64) bool {
 			// Limit bit depth & maximum values.
 			bitDepth = (bitDepth % 62) + 1
diff --git a/rbf/os.go b/rbf/os.go
index 7b2627f11..a76a550ef 100644
--- a/rbf/os.go
+++ b/rbf/os.go
@@ -19,4 +19,4 @@ package rbf
 // DefaultMaxSize is the default mmap size and therefore the maximum allowed
 // size of the database. The size can be increased by updating the DB.MaxSize
 // and reopening the database. This setting mainly affects virtual space usage.
-const DefaultMaxSize = 100 * (1 << 30) // 100GB
+const DefaultMaxSize = 4 * (1 << 30)
diff --git a/server/cluster_test.go b/server/cluster_test.go
index 7b17a253d..1cbda5ffc 100644
--- a/server/cluster_test.go
+++ b/server/cluster_test.go
@@ -26,7 +26,6 @@ import (
 	"time"
 
 	"github.com/pilosa/pilosa/v2"
-	"github.com/pilosa/pilosa/v2/race"
 	"github.com/pilosa/pilosa/v2/server"
 	"github.com/pilosa/pilosa/v2/test"
 	"golang.org/x/sync/errgroup"
@@ -634,10 +633,6 @@ func TestCluster_GossipMembership(t *testing.T) {
 }
 
 func TestClusterResize_RemoveNode(t *testing.T) {
-	if os.Getenv("PILOSA_TXSRC") == "rbf" && race.Enabled {
-		t.Skip("race detection enabled, skipping for rbf")
-	}
-
 	cluster := test.MustRunCluster(t, 3)
 	defer cluster.Close()
 	m0 := cluster.GetNode(0)
diff --git a/server/server_test.go b/server/server_test.go
index 02af30e59..270fdcaf5 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -34,7 +34,6 @@ import (
 	"github.com/pilosa/pilosa/v2"
 	"github.com/pilosa/pilosa/v2/http"
 	"github.com/pilosa/pilosa/v2/pql"
-	"github.com/pilosa/pilosa/v2/race"
 	"github.com/pilosa/pilosa/v2/roaring"
 	"github.com/pilosa/pilosa/v2/server"
 	"github.com/pilosa/pilosa/v2/test"
@@ -52,8 +51,6 @@ func init() { // nolint: gochecknoinits
 func TestMain_Set_Quick(t *testing.T) {
 	if testing.Short() {
 		t.Skip("short")
-	} else if os.Getenv("PILOSA_TXSRC") == "rbf" && race.Enabled {
-		t.Skip("race detection enabled, skipping for rbf")
 	}
 
 	for i := 0; i < 100; i++ {
@@ -950,10 +947,6 @@ func TestMain_ImportTimestampNoStandardView(t *testing.T) {
 }
 
 func TestClusterQueriesAfterRestart(t *testing.T) {
-	if os.Getenv("PILOSA_TXSRC") == "rbf" && race.Enabled {
-		t.Skip("race detection enabled, skipping for rbf")
-	}
-
 	cluster := test.MustRunCluster(t, 3)
 	defer cluster.Close()
 	cmd1 := cluster.GetNode(1)
@@ -1178,10 +1171,6 @@ func TestClusterExhaustingConnectionsImport(t *testing.T) {
 }
 
 func TestClusterMinMaxSumDecimal(t *testing.T) {
-	if os.Getenv("PILOSA_TXSRC") == "rbf" && race.Enabled {
-		t.Skip("race detection enabled, skipping for rbf")
-	}
-
 	cluster := test.MustRunCluster(t, 3)
 	defer cluster.Close()
 	cmd := cluster.GetNode(0)

From 2440413c49b0ec4e919aa6c05c7e9cf0057362d9 Mon Sep 17 00:00:00 2001
From: Ben Johnson <benbjohnson@yahoo.com>
Date: Tue, 6 Oct 2020 10:29:17 -0600
Subject: [PATCH 11/12] Remove test skips based on race detector

---
 race/norace.go     | 20 --------------------
 race/race.go       | 20 --------------------
 rbf/cursor_test.go |  9 ---------
 rbf/tx_test.go     |  5 -----
 4 files changed, 54 deletions(-)
 delete mode 100644 race/norace.go
 delete mode 100644 race/race.go

diff --git a/race/norace.go b/race/norace.go
deleted file mode 100644
index 6626c8380..000000000
--- a/race/norace.go
+++ /dev/null
@@ -1,20 +0,0 @@
-// 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.
-
-// +build !race
-
-package race
-
-// Enabled is true if the -race flag is enabled.
-const Enabled = false
diff --git a/race/race.go b/race/race.go
deleted file mode 100644
index a6fe765bc..000000000
--- a/race/race.go
+++ /dev/null
@@ -1,20 +0,0 @@
-// 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.
-
-// +build race
-
-package race
-
-// Enabled is true if the -race flag is enabled.
-const Enabled = true
diff --git a/rbf/cursor_test.go b/rbf/cursor_test.go
index 401117d06..9f134a92e 100644
--- a/rbf/cursor_test.go
+++ b/rbf/cursor_test.go
@@ -23,7 +23,6 @@ import (
 	"strings"
 	"testing"
 
-	"github.com/pilosa/pilosa/v2/race"
 	"github.com/pilosa/pilosa/v2/rbf"
 	"github.com/pilosa/pilosa/v2/roaring"
 )
@@ -77,8 +76,6 @@ func TestCursor_FirstNext_Quick(t *testing.T) {
 		t.Skip("-short enabled, skipping")
 	} else if is32Bit() {
 		t.Skip("32-bit build, skipping quick check tests")
-	} else if race.Enabled {
-		t.Skip("race detection enabled, skipping")
 	}
 
 	const n = 10000
@@ -199,8 +196,6 @@ func TestCursor_LastPrev_Quick(t *testing.T) {
 		t.Skip("-short enabled, skipping")
 	} else if is32Bit() {
 		t.Skip("32-bit build, skipping quick check tests")
-	} else if race.Enabled {
-		t.Skip("race detection enabled, skipping")
 	}
 
 	const n = 10000
@@ -313,8 +308,6 @@ func TestCursor_Union(t *testing.T) {
 			t.Skip("-short enabled, skipping")
 		} else if is32Bit() {
 			t.Skip("32-bit build, skipping quick check tests")
-		} else if race.Enabled {
-			t.Skip("race detection enabled, skipping")
 		}
 
 		QuickCheck(t, func(t *testing.T, rand *rand.Rand) {
@@ -394,8 +387,6 @@ func TestCursor_Intersect(t *testing.T) {
 			t.Skip("-short enabled, skipping")
 		} else if is32Bit() {
 			t.Skip("32-bit build, skipping quick check tests")
-		} else if race.Enabled {
-			t.Skip("race detection enabled, skipping")
 		}
 
 		QuickCheck(t, func(t *testing.T, rand *rand.Rand) {
diff --git a/rbf/tx_test.go b/rbf/tx_test.go
index 7e723aa8a..2d5505a4a 100644
--- a/rbf/tx_test.go
+++ b/rbf/tx_test.go
@@ -21,7 +21,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/pilosa/pilosa/v2/race"
 	"github.com/pilosa/pilosa/v2/rbf"
 	"github.com/pilosa/pilosa/v2/txkey"
 )
@@ -219,8 +218,6 @@ func TestTx_Add_Quick(t *testing.T) {
 		t.Skip("-short enabled, skipping")
 	} else if is32Bit() {
 		t.Skip("32-bit build, skipping quick check tests")
-	} else if race.Enabled {
-		t.Skip("race detection enabled, skipping")
 	}
 
 	QuickCheck(t, func(t *testing.T, rand *rand.Rand) {
@@ -258,8 +255,6 @@ func TestTx_AddRemove_Quick(t *testing.T) {
 		t.Skip("-short enabled, skipping")
 	} else if is32Bit() {
 		t.Skip("32-bit build, skipping quick check tests")
-	} else if race.Enabled {
-		t.Skip("race detection enabled, skipping")
 	}
 
 	QuickCheck(t, func(t *testing.T, rand *rand.Rand) {

From 3b7758a4f2a101dc27720b6d814e9df48b8025e9 Mon Sep 17 00:00:00 2001
From: Ben Johnson <benbjohnson@yahoo.com>
Date: Tue, 6 Oct 2020 10:36:57 -0600
Subject: [PATCH 12/12] Add comment explaining rbf.DB.checkpoint() args

---
 rbf/db.go | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/rbf/db.go b/rbf/db.go
index 970c66b0c..f60b12081 100644
--- a/rbf/db.go
+++ b/rbf/db.go
@@ -219,8 +219,12 @@ func (db *DB) Checkpoint() error {
 
 // checkpoint moves WAL segments to the main DB file.
 //
-// Note that mu should db.mu when called through the external API but it
-// can be &nopLocker if called under lock.
+// Note that mu should db.mu when called through DB.Checkpoint() but it
+// can be &nopLocker if called under lock. The external API will be used
+// to periodically checkpoint outside of a transaction and the locking
+// must be used only in the beginning (to obtain the segment list) and at
+// the end (when removing old segments from the list). If the entire function
+// were to obtain a lock then it would block all new read & write transactions.
 func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error {
 	// Obtain a snapshot of WAL segments at the start.
 	mu.Lock()