From 07d0dd86ad1fd6d5dbffc694fbdff9ae33c4b449 Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Fri, 9 Oct 2026 09:27:31 +0000 Subject: [PATCH 1/3] test: reproduce delayed CAS fence regression Signed-off-by: Perfloop Agent --- storage/cas.go | 6 ++ storage/cas_fence_test.go | 110 +++++++++++++++++++++++++++++ storage/merge/cas_operator_test.go | 42 +++++++++++ storage/storage.go | 6 +- 4 files changed, 161 insertions(+), 3 deletions(-) diff --git a/storage/cas.go b/storage/cas.go index 159d8a4..b50254f 100644 --- a/storage/cas.go +++ b/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 diff --git a/storage/cas_fence_test.go b/storage/cas_fence_test.go index 86cfe92..b4afa49 100644 --- a/storage/cas_fence_test.go +++ b/storage/cas_fence_test.go @@ -7,6 +7,7 @@ import ( "bytes" "os" "path/filepath" + "sync/atomic" "testing" "time" @@ -40,6 +41,115 @@ 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) + defer cleanup() + + // Keep the fence well inside its retention horizon throughout this schedule. + started := time.Now() + s.cleaner.fenceRetention = time.Hour + + 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 diff --git a/storage/merge/cas_operator_test.go b/storage/merge/cas_operator_test.go index 9651c77..4e5bba9 100644 --- a/storage/merge/cas_operator_test.go +++ b/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. diff --git a/storage/storage.go b/storage/storage.go index eee53e1..0b740c3 100644 --- a/storage/storage.go +++ b/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 From d268e2408a8035bf8f17a4c83b27332648c12a89 Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Fri, 9 Oct 2026 09:31:59 +0000 Subject: [PATCH 2/3] fix: keep CAS delete fences monotonic Signed-off-by: Perfloop Agent --- storage/merge/operator.go | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/storage/merge/operator.go b/storage/merge/operator.go index dfba14b..689c234 100644 --- a/storage/merge/operator.go +++ b/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: From dc311146591a373d47565ffdca405527860b481e Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Fri, 9 Oct 2026 11:35:08 +0000 Subject: [PATCH 3/3] test: configure CAS fence retention before startup Signed-off-by: Perfloop Agent --- storage/cas_fence_test.go | 5 +++-- storage/cas_test.go | 4 ++-- storage/test_helpers.go | 8 ++++++-- 3 files changed, 11 insertions(+), 6 deletions(-) diff --git a/storage/cas_fence_test.go b/storage/cas_fence_test.go index b4afa49..0ca651b 100644 --- a/storage/cas_fence_test.go +++ b/storage/cas_fence_test.go @@ -45,12 +45,13 @@ func mismatchWith(t *testing.T, err error) uint64 { // 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) + 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() - s.cleaner.fenceRetention = time.Hour const key = "delayed-delete" deleteAAtMerge := make(chan uint64, 1) diff --git a/storage/cas_test.go b/storage/cas_test.go index 3dc3807..ec62746 100644 --- a/storage/cas_test.go +++ b/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 { diff --git a/storage/test_helpers.go b/storage/test_helpers.go index 7703a6e..d419034 100644 --- a/storage/test_helpers.go +++ b/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")