Repository: polarsignals/frostdb (go)

Issue:
wal: ensure wal can be appended to after loading snapshots
I believe that if we load a snapshot at txn `n`, but we have a truncated wal with the last index at txn `n-k`, we could get into a situation where no new records are appended to the WAL, which is not good since the log request queue could grow unboundedly given the wal's nextTx is at `n-k`, but this transaction will never be appended to the log.

I think this is currently happening in production given that the WALs are tiny.

Pull request description:
*: recovery fixes
Closes #390 

The main fix is to overwrite `roTables` wal when loading from bucket storage. This was previously not done after calling `openWAL` on the `DB`, making it so that after recovery, the WAL was never appended to since these tables were created with a no-op wal.

Patch:
diff --git a/db.go b/db.go
index 730dad6..89662d1 100644
--- a/db.go
+++ b/db.go
@@ -379,6 +379,19 @@ func (s *ColumnStore) DB(ctx context.Context, name string) (*DB, error) {
 			if err != nil {
 				return err
 			}
+			// WAL pointers of tables need to be updated to the DB WAL since
+			// they are loaded from object storage and snapshots with a no-op
+			// WAL by default.
+			for _, table := range db.tables {
+				if !table.config.DisableWal {
+					table.wal = db.wal
+				}
+			}
+			for _, table := range db.roTables {
+				if !table.config.DisableWal {
+					table.wal = db.wal
+				}
+			}
 		}
 
 		// Register metrics last to avoid duplicate registration should and of the WAL or storage replay errors occur
@@ -468,16 +481,6 @@ func (db *DB) recover(ctx context.Context, wal WAL) error {
 	}
 	snapshotLogArgs := make([]any, 0)
 	if snapshotTx != 0 {
-		db.mtx.Lock()
-		// WAL pointers of tables loaded on snapshot need to be updated to the
-		// DB WAL.
-		for _, table := range db.tables {
-			if !table.config.DisableWal {
-				table.wal = wal
-			}
-		}
-		db.mtx.Unlock()
-
 		snapshotLogArgs = append(
 			snapshotLogArgs,
 			"snapshot_tx", snapshotTx,
@@ -614,7 +617,7 @@ func (db *DB) recover(ctx context.Context, wal WAL) error {
 			// If we get to this point it means a block was finished but did
 			// not get persisted.
 			table.pendingBlocks[table.active] = struct{}{}
-			go table.writeBlock(ctx, table.active)
+			go table.writeBlock(table.active, false /* snapshotDB */)
 
 			protoEqual := false
 			switch schema.(type) {
@@ -731,17 +734,19 @@ func (db *DB) Close() error {
 	for _, table := range db.tables {
 		table.close()
 		if db.bucket != nil {
-			table.writeBlock(context.TODO(), table.ActiveBlock())
+			// Write the blocks but no snapshots since they are long-running
+			// jobs.
+			table.writeBlock(table.ActiveBlock(), false /* snapshotDB */)
 		}
 	}
 
 	if db.bucket != nil {
 		// If we've successfully persisted all the table blocks we can remove
 		// the wal and snapshots.
-		if err := os.RemoveAll(db.walDir()); err != nil {
+		if err := os.RemoveAll(db.snapshotsDir()); err != nil {
 			return err
 		}
-		if err := os.RemoveAll(db.snapshotsDir()); err != nil {
+		if err := os.RemoveAll(db.walDir()); err != nil {
 			return err
 		}
 	}
diff --git a/table.go b/table.go
index 6bf7976..c2572dc 100644
--- a/table.go
+++ b/table.go
@@ -387,7 +387,7 @@ func (t *Table) dropPendingBlock(block *TableBlock) {
 	delete(t.pendingBlocks, block)
 }
 
-func (t *Table) writeBlock(ctx context.Context, block *TableBlock) {
+func (t *Table) writeBlock(block *TableBlock, snapshotDB bool) {
 	level.Debug(t.logger).Log("msg", "syncing block")
 	block.pendingWritersWg.Wait()
 
@@ -446,7 +446,7 @@ func (t *Table) writeBlock(ctx context.Context, block *TableBlock) {
 	}
 	t.mtx.Unlock()
 	t.db.maintainWAL()
-	if t.db.columnStore.snapshotTriggerSize != 0 && t.db.columnStore.enableWAL {
+	if snapshotDB && t.db.columnStore.snapshotTriggerSize != 0 && t.db.columnStore.enableWAL {
 		func() {
 			if !t.db.snapshotInProgress.CompareAndSwap(false, true) {
 				// Snapshot already in progress. This could lead to duplicate
@@ -499,7 +499,7 @@ func (t *Table) RotateBlock(ctx context.Context, block *TableBlock) error {
 	t.metrics.numParts.Set(float64(0))
 
 	t.pendingBlocks[block] = struct{}{}
-	go t.writeBlock(ctx, block)
+	go t.writeBlock(block, true /* snapshotDB */)
 
 	return nil
 }


Test patch:
diff --git a/db_test.go b/db_test.go
index f2ea8b0..0af8472 100644
--- a/db_test.go
+++ b/db_test.go
@@ -1613,6 +1613,7 @@ func TestDBRecover(t *testing.T) {
 			WithStoragePath(dir),
 			WithWAL(),
 			WithSnapshotTriggerSize(1),
+			WithBucketStorage(bucket),
 		)
 		require.NoError(t, err)
 		defer c.Close()