storage: test that batch commit doesn't touch SSTs

This commit adds simple test where it injects errors on any operation
touching any SSTable file before attempting to commit a batch, and
asserts that the batch can commit successfully without needing any
SSTable.

References: #143135

Release note: None

Author: iskettaneh

pkg/storage/BUILD.bazel

@@ -217,6 +217,7 @@ go_test(
         "@com_github_cockroachdb_pebble//sstable",
         "@com_github_cockroachdb_pebble//sstable/block",
         "@com_github_cockroachdb_pebble//vfs",
+        "@com_github_cockroachdb_pebble//vfs/errorfs",
         "@com_github_cockroachdb_redact//:redact",
         "@com_github_kr_pretty//:pretty",
         "@com_github_stretchr_testify//assert",

pkg/storage/batch_test.go

@@ -12,6 +12,7 @@ import (
 	"fmt"
 	"reflect"
 	"strconv"
+	"strings"
 	"sync/atomic"
 	"testing"
 
@@ -27,6 +28,8 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble"
+	"github.com/cockroachdb/pebble/vfs"
+	"github.com/cockroachdb/pebble/vfs/errorfs"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -1010,3 +1013,61 @@ func TestBatchReader(t *testing.T) {
 	require.False(t, r.Next())
 	require.NoError(t, r.Error())
 }
+
+// TestBatchCommitDoesntTouchSST tests that committing a writeBatch doesn't
+// touch SST files.
+func TestBatchCommitDoesntTouchSST(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	// Create an atomic variable that will cause an error when SST operations are
+	// performed.
+	var failSSTOps atomic.Bool
+
+	// Create a custom injector that blocks SST operations when failSSTOps is
+	// true.
+	injector := errorfs.InjectorFunc(func(op errorfs.Op) error {
+		if strings.Contains(op.Path, ".sst") && failSSTOps.Load() {
+			return errors.Newf("blocking SST operation: %+v", op)
+		}
+		return nil
+	})
+
+	// Create the wrapped filesystem, and create a db.
+	memFS := vfs.NewMem()
+	wrappedFS := errorfs.Wrap(memFS, injector)
+	env := mustInitTestEnv(t, wrappedFS, "")
+	db, err := Open(context.Background(), env, cluster.MakeClusterSettings())
+	require.NoError(t, err)
+	defer db.Close()
+
+	// Initialize the db with some data.
+	initBatch := db.NewBatch()
+	defer initBatch.Close()
+
+	// Perform some operations.
+	require.NoError(t, initBatch.PutUnversioned(mvccKey("key1").Key, []byte("val1")))
+	require.NoError(t, initBatch.PutUnversioned(mvccKey("key2").Key, []byte("val2")))
+	require.NoError(t, initBatch.PutUnversioned(mvccKey("key3").Key, []byte("val3")))
+	require.NoError(t, initBatch.PutUnversioned(mvccKey("key4").Key, []byte("val4")))
+	require.NoError(t, initBatch.Commit(true /* sync */))
+
+	// Force a flush to create an SST file.
+	require.NoError(t, db.Flush())
+
+	// Create a new batch for testing.
+	testingBatch := db.NewBatch()
+	defer testingBatch.Close()
+
+	// Perform some operations.
+	require.Equal(t, []byte("val1"), mvccGetRaw(t, testingBatch, mvccKey("key1")))
+	require.Equal(t, []byte(nil), mvccGetRaw(t, testingBatch, mvccKey("non-existent-key")))
+	_, err = Scan(context.Background(), testingBatch, localMax, roachpb.KeyMax, 0)
+	require.NoError(t, err)
+	require.NoError(t, testingBatch.ClearUnversioned(mvccKey("key4").Key, ClearOptions{}))
+
+	// Before committing, enable SST operation errors and make sure the commit
+	// succeeds.
+	failSSTOps.Store(true)
+	require.NoError(t, testingBatch.Commit(true /* sync */))
+}