Fix WAL ID not found panic.

This commit changes the checkpointing to determine a minimum WAL ID
for readers and a max ID based on the writer. Pages are checkpointed
from the WAL up to the writer's max WAL ID but segments are removed
only up to the reader's minimum WAL ID. This ensures that WAL pages
are not removed out from under current read transactions.

go.sum

@@ -161,6 +161,7 @@ github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
 github.com/pilosa/memberlist v0.1.4-0.20190415211605-f6512523c021 h1:ERLyN4p3KS5Fk2ADsDENm2cq0+Lx6sF1sG8uwRlySpU=
 github.com/pilosa/memberlist v0.1.4-0.20190415211605-f6512523c021/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I=
+github.com/pilosa/pilosa v1.4.1 h1:zSNyS/MqXTfRDNRBBNCApsOu2oUnTEj2uymCaqYSNx8=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=

rbf/db.go

@@ -238,10 +238,9 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error {
 		return err
 	}
 	walID := readMetaWALID(page)
-	maxCheckpointedWALID := walID
 
 	// Determine the high water mark for WAL pages that can be copied.
-	minActiveWALID := db.minActiveWALID()
+	writerWALID := db.writerWALID()
 
 	// Loop over each transaction
 	walID++
@@ -257,7 +256,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 || writerWALID == 0 || walID < writerWALID
 
 			page, err := readWALPage(segments, walID)
 			if err != nil {
@@ -293,11 +292,6 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error {
 			if err := db.writeDBPage(pgno, page); err != nil {
 				return err
 			}
-
-			// Track highest WALID that has been checkpointed back to disk.
-			if IsMetaPage(page) {
-				maxCheckpointedWALID = walID
-			}
 		}
 	}
 
@@ -305,21 +299,23 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error {
 	if err := db.fsync(db.file); err != nil {
 		return fmt.Errorf("db file sync: %w", err)
 	}
+	mu.Lock()
+	db.pageMap = pageMap
+	mu.Unlock()
 
 	// Remove WAL segments that have been checkpointed.
-	if maxCheckpointedWALID != 0 {
-		for _, segment := range segments {
-			if segment.MaxWALID() > maxCheckpointedWALID {
-				break
-			}
+	minPageMapWALID := db.minPageMapWALID()
+	for _, segment := range segments {
+		if minPageMapWALID != 0 && segment.MaxWALID() > minPageMapWALID {
+			break
+		}
 
-			if err := func() error {
-				mu.Lock()
-				defer mu.Unlock()
-				return db.removeWALSegment(segment.Path)
-			}(); err != nil {
-				return err
-			}
+		if err := func() error {
+			mu.Lock()
+			defer mu.Unlock()
+			return db.removeWALSegment(segment.Path)
+		}(); err != nil {
+			return err
 		}
 	}
 
@@ -331,7 +327,6 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error {
 		mu.Unlock()
 	}
 
-	db.pageMap = pageMap
 	return nil
 }
 
@@ -361,16 +356,38 @@ func (db *DB) removeWALSegment(path string) error {
 	return 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 {
-	var walID int64
+// minPageMapWALID returns the lowest WAL ID referenced by an active page map.
+func (db *DB) minPageMapWALID() int64 {
+	// Use the db's page map because that is the state of the map when the
+	// writer transaction started. We can't use the writer transaction's map
+	// because it can change.
+	min := pageMapMinWALID(db.pageMap)
+
 	for tx := range db.txs {
-		if walID == 0 || walID > tx.walID {
-			walID = tx.walID
+		// If a write transaction is active, ensure the min is at least the starting WAL.
+		if tx.writable {
+			if min == 0 || tx.walID < min {
+				min = tx.walID
+			}
+			continue
+		}
+
+		// Record the min WAL ID referenced by the reader's page map.
+		if walID := pageMapMinWALID(tx.pageMap); min == 0 || walID < min {
+			min = walID
 		}
 	}
-	return walID
+	return min
+}
+
+// writerWALID returns the starting WAL ID of the active writer tx.
+func (db *DB) writerWALID() int64 {
+	for tx := range db.txs {
+		if tx.writable {
+			return tx.walID
+		}
+	}
+	return 0
 }
 
 // Close closes the database.
@@ -738,3 +755,15 @@ type nopLocker struct{}
 
 func (*nopLocker) Lock()   {}
 func (*nopLocker) Unlock() {}
+
+// pageMapMinWALID returns the lowest WAL ID
+func pageMapMinWALID(m *immutable.Map) int64 {
+	var min int64
+	for itr := m.Iterator(); !itr.Done(); {
+		_, v := itr.Next()
+		if walID := v.(int64); min == 0 || walID < min {
+			min = walID
+		}
+	}
+	return min
+}

rbf/db_test.go

@@ -15,6 +15,7 @@
 package rbf_test
 
 import (
+	"context"
 	"fmt"
 	"math/rand"
 	"net"
@@ -24,6 +25,8 @@ import (
 	"time"
 
 	"github.com/pilosa/pilosa/v2/rbf"
+	rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg"
+	"golang.org/x/sync/errgroup"
 	_ "net/http/pprof"
 )
 
@@ -290,6 +293,81 @@ func TestDB_HasData(t *testing.T) {
 	}
 }
 
+// Ensures the DB can continuously write while readers are executing.
+func TestDB_MultiTx(t *testing.T) {
+	if testing.Short() {
+		t.Skip("-short enabled, skipping")
+	}
+
+	cfg := rbfcfg.NewDefaultConfig()
+	cfg.CheckpointEveryDur = 1 * time.Millisecond
+	db := MustOpenDB(t, cfg)
+	defer MustCloseDB(t, db)
+
+	// Run multiple readers in separate goroutines.
+	ctx, cancel := context.WithCancel(context.Background())
+	g, ctx := errgroup.WithContext(ctx)
+	for i := 0; i < 10; i++ {
+		g.Go(func() error {
+			for {
+				if ctx.Err() != nil {
+					return nil // cancelled, return no error
+				} else if err := testDB_MultiTx_reader(db); err != nil {
+					return err
+				}
+			}
+		})
+	}
+
+	// Continuously set/clear bits while readers are executing.
+	for i := 0; i < 1000; i++ {
+		func() {
+			tx, err := db.Begin(true)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer tx.Rollback()
+
+			for j := 0; j < rand.Intn(10); j++ {
+				v := rand.Intn(1 << 20)
+				if _, err := tx.Add("x", uint64(v)); err != nil {
+					t.Fatal(err)
+				}
+			}
+
+			if err := tx.Commit(); err != nil {
+				t.Fatal(err)
+			}
+		}()
+	}
+
+	// Stop readers & wait.
+	cancel()
+	if err := g.Wait(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// testDB_MultiTx_reader checks if a bitmap contains a random set of bits.
+func testDB_MultiTx_reader(db *rbf.DB) error {
+	tx, err := db.Begin(false)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
+
+	for i := 0; i < rand.Intn(1000); i++ {
+		v := rand.Intn(1 << 20)
+		if _, err := tx.Contains("x", uint64(v)); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
 // better diagnosis of deadlocks/hung situations versus just really slow "Quick" tests.
 func TestMain(m *testing.M) {
 	port := getAvailPort()

rbf/rbf_test.go

@@ -26,6 +26,7 @@ import (
 	"testing"
 
 	"github.com/pilosa/pilosa/v2/rbf"
+	rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg"
 )
 
 var quickCheckN *int = flag.Int("quickchecks", 10, "The number of iterations for each quickcheck")
@@ -60,20 +61,24 @@ func TestReadWriteRootRecord(t *testing.T) {
 }
 
 // NewDB returns a new instance of DB with a temporary path.
-func NewDB() *rbf.DB {
+func NewDB(cfg ...*rbfcfg.Config) *rbf.DB {
 	path, err := ioutil.TempDir("", "")
 	if err != nil {
 		panic(err)
 	}
 
-	db := rbf.NewDB(path, nil)
+	var cfg0 *rbfcfg.Config
+	if len(cfg) > 0 {
+		cfg0 = cfg[0]
+	}
+	db := rbf.NewDB(path, cfg0)
 	return db
 }
 
 // MustOpenDB returns a db opened on a temporary file. On error, fail test.
-func MustOpenDB(tb testing.TB) *rbf.DB {
+func MustOpenDB(tb testing.TB, cfg ...*rbfcfg.Config) *rbf.DB {
 	tb.Helper()
-	db := NewDB()
+	db := NewDB(cfg...)
 	if err := db.Open(); err != nil {
 		tb.Fatal(err)
 	}