fix: prevent delayed CAS deletes from lowering fences

storage/cas.go

@@ -699,6 +699,12 @@ func (s *Storage) DeleteIfVersion(key string, expected uint64) (retErr error) {
 		return storageErrors.NewInternalError("DeleteIfVersion", err)
 	}
 
+	if s.beforeMetaCommit != nil {
+		if err := s.beforeMetaCommit(); err != nil {
+			return mapRocksDBError("DeleteIfVersion", key, err)
+		}
+	}
+
 	// Every eviction-index entry of the value being deleted — the captured
 	// one, and any the asynchronous LRU refresh flushes for a read that
 	// preceded the delete — carries a write time before this instant; a

storage/cas_fence_test.go

@@ -7,6 +7,7 @@ import (
 	"bytes"
 	"os"
 	"path/filepath"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -40,6 +41,116 @@ func mismatchWith(t *testing.T, err error) uint64 {
 	return vm.CurrentVersion
 }
 
+// TestCAS_Fence_DelayedDeleteMustNotLowerFence forces a delete that issued an
+// older stamp to merge after a newer invalidation. A refill holding the token
+// observed between the two deletes must still lose to the newer fence.
+func TestCAS_Fence_DelayedDeleteMustNotLowerFence(t *testing.T) {
+	s, cleanup := createCASTestStorage(t, func(config *StorageConfig) {
+		config.FenceRetention = time.Hour
+	})
+	defer cleanup()
+
+	// Keep the fence well inside its retention horizon throughout this schedule.
+	started := time.Now()
+
+	const key = "delayed-delete"
+	deleteAAtMerge := make(chan uint64, 1)
+	resumeDeleteA := make(chan struct{})
+	resumed := false
+	resumeA := func() {
+		if !resumed {
+			close(resumeDeleteA)
+			resumed = true
+		}
+	}
+	var firstDelete atomic.Bool
+	s.beforeMetaCommit = func() error {
+		if firstDelete.CompareAndSwap(false, true) {
+			// Delete A has issued its stamp and built its operand, but has not
+			// submitted the metadata merge yet.
+			deleteAAtMerge <- s.lastVersion.Load()
+			<-resumeDeleteA
+		}
+		return nil
+	}
+
+	deleteAResult := make(chan error, 1)
+	go func() { deleteAResult <- s.DeleteIfVersion(key, 0) }()
+	deleteAJoined := false
+	defer func() {
+		resumeA()
+		if !deleteAJoined {
+			<-deleteAResult
+		}
+	}()
+
+	var a uint64
+	select {
+	case a = <-deleteAAtMerge:
+	case err := <-deleteAResult:
+		deleteAJoined = true
+		t.Fatalf("Delete A returned before reaching the merge barrier: %v", err)
+	case <-time.After(10 * time.Second):
+		t.Fatal("Delete A did not reach the merge barrier")
+	}
+	require.NotZero(t, a)
+
+	// A is paused before Merge, so this absent read receives a fresh token b.
+	b := absentToken(t, s, key)
+	require.Greater(t, b, a)
+
+	// B must commit before A resumes. Its tombstone carries the newer fence c.
+	require.NoError(t, s.DeleteIfVersion(key, 0))
+	c := absentToken(t, s, key)
+	require.Greater(t, c, b)
+	row, found, err := s.readRowForCAS(keys.MakeMetadataKey(key))
+	require.NoError(t, err)
+	require.True(t, found)
+	require.Equal(t, int64(1), row.Expiry)
+	require.Equal(t, c, row.Version)
+	require.Empty(t, row.RawFilePath)
+	require.Empty(t, row.SegmentPath)
+
+	resumeA()
+	errA := <-deleteAResult
+	deleteAJoined = true
+	aMismatch, aLost := storageErrors.IsVersionMismatch(errA)
+	if errA != nil && !aLost {
+		t.Fatalf("Delete A returned an unexpected operation error: %v", errA)
+	}
+	aCurrent := uint64(0)
+	if aLost {
+		aCurrent = aMismatch.CurrentVersion
+	}
+
+	retainedToken := absentToken(t, s, key)
+	row, found, err = s.readRowForCAS(keys.MakeMetadataKey(key))
+	require.NoError(t, err)
+	require.True(t, found)
+	require.Equal(t, int64(1), row.Expiry)
+	require.Empty(t, row.RawFilePath)
+	require.Empty(t, row.SegmentPath)
+
+	_, refillErr := s.PutIfVersion(key, bytes.NewReader([]byte("stale")), 0, b)
+	refillMismatch, refillLost := storageErrors.IsVersionMismatch(refillErr)
+	if refillErr != nil && !refillLost {
+		t.Fatalf("the ordered refill returned an unexpected operation error: %v", refillErr)
+	}
+	refillCurrent := uint64(0)
+	if refillLost {
+		refillCurrent = refillMismatch.CurrentVersion
+	}
+	_, finalToken, found, err := s.GetWithVersion(key)
+	require.NoError(t, err)
+	require.Less(t, time.Since(started), s.cleaner.fenceRetention, "the fence must remain within its retention horizon")
+
+	if !(aLost && aCurrent == c && retainedToken == c && row.Version == c &&
+		refillLost && refillCurrent == c && finalToken == c && !found) {
+		t.Errorf("delayed CAS delete must preserve the newer retained fence: a=%d b=%d c=%d, delete mismatch=%t current=%d, retained=%d, refill mismatch=%t current=%d, final=%d found=%t",
+			a, b, c, aLost, aCurrent, retainedToken, refillLost, refillCurrent, finalToken, found)
+	}
+}
+
 // TestCAS_Fence_PopulateThatObservedAbsenceBeforeDeleteLoses: the invalidate-
 // vs-populate race. A populate reads absence (token t), an invalidation lands
 // on the still-missing key (fence S > t), and the populate's write carrying t

storage/cas_test.go

@@ -22,10 +22,10 @@ import (
 	pb "github.com/tigrisdata/ocache/storage/proto"
 )
 
-func createCASTestStorage(t *testing.T) (*Storage, func()) {
+func createCASTestStorage(t *testing.T, options ...func(*StorageConfig)) (*Storage, func()) {
 	// Cap set so eviction indexing runs; thresholds small enough to exercise the
 	// raw-file path with modest payloads.
-	return createTestStorage(t, 3600, 1024, 4*1024, 16*1024*1024, 1000, 1<<30)
+	return createTestStorage(t, 3600, 1024, 4*1024, 16*1024*1024, 1000, 1<<30, options...)
 }
 
 func readAllString(t *testing.T, r io.Reader) string {

storage/merge/cas_operator_test.go

@@ -232,6 +232,48 @@ func TestMergeCAS_FenceBump(t *testing.T) {
 	assert.Equal(t, uint64(400), got.Version)
 }
 
+// TestMergeCAS_DelayedDeleteDoesNotLowerFence covers a later database arrival
+// whose CAS_DELETE stamp was issued before the tombstone it meets. The older
+// delete must not lower the fence and admit a refill that observed the interval.
+func TestMergeCAS_DelayedDeleteDoesNotLowerFence(t *testing.T) {
+	op := NewMultiplexOperator()
+	operands := [][]byte{
+		casDeleteOperand(t, 0, 300),         // B: newer invalidation arrives first
+		casDeleteOperand(t, 0, 100),         // A: earlier stamp arrives later
+		casPutOperand(t, 200, 400, "stale"), // refill observed between A and B
+	}
+
+	allAtOnce, ok := op.FullMerge(casMetaKey, nil, operands)
+	require.True(t, ok)
+	fence := mustUnmarshal(t, allAtOnce)
+	require.True(t, IsTombstone(fence))
+	assert.Equal(t, uint64(300), fence.Version, "the delayed delete must not lower B's fence")
+
+	// RocksDB may resolve the same sequence with a prefix folded into a base.
+	// Every split must agree with FullMerge over the full operand sequence.
+	for split := 1; split < len(operands); split++ {
+		folded, ok := op.FullMerge(casMetaKey, nil, operands[:split])
+		require.True(t, ok)
+		rest, ok := op.FullMerge(casMetaKey, folded, operands[split:])
+		require.True(t, ok)
+		assert.Equal(t, allAtOnce, rest, "split at %d diverged from single-pass resolution", split)
+	}
+
+	// A newer delete still advances the fence; a fresh or explicitly unordered
+	// refill retains its existing behavior.
+	advanced, ok := op.FullMerge(casMetaKey, allAtOnce, [][]byte{casDeleteOperand(t, 0, 500)})
+	require.True(t, ok)
+	assert.Equal(t, uint64(500), mustUnmarshal(t, advanced).Version)
+
+	fresh, ok := op.FullMerge(casMetaKey, allAtOnce, [][]byte{casPutOperand(t, 300, 600, "fresh")})
+	require.True(t, ok)
+	assert.Equal(t, []byte("fresh"), mustUnmarshal(t, fresh).Data)
+
+	unordered, ok := op.FullMerge(casMetaKey, allAtOnce, [][]byte{casPutOperand(t, 0, 700, "unordered")})
+	require.True(t, ok)
+	assert.Equal(t, []byte("unordered"), mustUnmarshal(t, unordered).Data)
+}
+
 // TestMergeCAS_Determinism asserts read-time and compaction-time resolution
 // agree: folding a prefix of operands into a new base and then applying the
 // rest must produce byte-identical results to applying all operands at once.

storage/merge/operator.go

@@ -290,10 +290,12 @@ func (m *MultiplexOperator) mergeMetadataCAS(key, existingValue []byte, operands
 				}
 				hadBase = true
 			case base.Expiry == tombstoneExpiry:
-				// Deleting a dead key again bumps the fence to a newer stamp, so
-				// a second invalidation orders AFTER anything that observed the
-				// first. Same admission rule as a put over a fence.
-				if fenceAdmits(&base, op.CasExpectedVersion) {
+				// Deleting a dead key again normally bumps the fence, so a second
+				// invalidation orders after anything that observed the first. But
+				// stamp issuance can precede the merge by an arbitrary delay: an
+				// otherwise-admitted older operand must not lower this newer fence.
+				// Keeping the base makes the delete's read-back report a mismatch.
+				if fenceAdmits(&base, op.CasExpectedVersion) && op.Version >= base.Version {
 					base = pb.ValueMessage{Expiry: tombstoneExpiry, Version: op.Version}
 				}
 			default:

storage/storage.go

@@ -186,9 +186,9 @@ type Storage struct {
 	// covers a file only while it is being written, and this covers the gap
 	// from that unlock to the commit that makes the file referenced (#156).
 	inflightRaw sync.Map
-	// beforeMetaCommit is set only by package tests: it runs inside putLow just
-	// before the metadata batch is written, and a non-nil error stands in for a
-	// RocksDB write failure so the callers' failure paths can be exercised.
+	// beforeMetaCommit is set only by package tests: it runs before putLow's
+	// metadata batch and DeleteIfVersion's metadata merge, and a non-nil error
+	// stands in for a RocksDB write failure so failure paths can be exercised.
 	beforeMetaCommit func() error
 	evictionPolicy   string        // "lru" or "fifo"; governs whether reads refresh access time
 	closed           atomic.Bool   // True when storage has been closed

storage/test_helpers.go

@@ -9,9 +9,10 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// createTestStorage creates a new storage instance for testing and returns a cleanup function
+// createTestStorage creates a new storage instance for testing and returns a cleanup function.
+// Options customize StorageConfig before construction starts background workers.
 func createTestStorage(t testing.TB, ttl int, inlineThreshold int, compactThreshold int64,
-	segmentSize int64, fdCacheSize int, maxDiskUsage int64,
+	segmentSize int64, fdCacheSize int, maxDiskUsage int64, options ...func(*StorageConfig),
 ) (*Storage, func()) {
 	dir := t.TempDir()
 
@@ -26,6 +27,9 @@ func createTestStorage(t testing.TB, ttl int, inlineThreshold int, compactThresh
 		FragThreshold:       DefaultFragmentationThreshold,
 		DisableRecompaction: true,
 	}
+	for _, option := range options {
+		option(config)
+	}
 	s, err := NewStorageWithConfig(config)
 	require.NoError(t, err, "failed to create storage")