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.


Discussion:
I think this could be the first reason we need to fork the wal library, in order to force the `lastIndex` to be a specific integer (the `txn` we're resuming from on reload). However, before doing that, I want to figure out how we got into this state in the first place.
I think I see what happened. https://github.com/polarsignals/frostdb/pull/383 removed the backwards compatibility test we had for WALs and deprecated the schema field. This causes a WAL replay to panic when replaying old WALs since the accessing the new field will panic with a nil pointer. The wal is then truncated due to our recovery mechanism, which on an old wal will truncate at index 0, causing an empty WAL to be used and never appended to in the future.

We probably need to add the backwards incompatibility test with a new WAL since old WALs could have created more issues (no config was persisted) to protect from other proto changes in the future.
The problem still remains that if we ever `TruncateBack` here: https://github.com/polarsignals/frostdb/blob/eb1037fad04c7300cefcc25cb6ed3499cc0ccaa8/wal/wal.go#L417

We will never be able to accept new writes to the WAL.
#393 fixes this issue because when loading from a snapshot at txn `n` and a wal at txn `n+k`, the tables would be created with a noop wal that was never updated to point to the db wal. This means that no more records will ever be logged to the WAL.

The issue that is mentioned here that somehow we get into a state with a WAL that has a `lastIndex < highWatermark` is theoretically impossible. WALs will only ever be replayed from a snapshot at txn `n`, and `TruncateBack` will therefore only ever be called on txn `n+i`, but `n+i >= n` is always true, so we will never load a snapshot to update our last index past the end of the WAL (unless we manually delete files).
Reopening this issue because although we fixed some cases where this could happen, it still seems to be happening on prod.

Patch:
diff --git a/db.go b/db.go
--- 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
--- 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
--- 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()
diff --git a/testdata/oldwal/00000000000000000001 b/testdata/oldwal/databases/test/wal/00000000000000000001
similarity index 100%
rename from testdata/oldwal/00000000000000000001
rename to testdata/oldwal/databases/test/wal/00000000000000000001