Remove-Get concurrency fix

common/errors.go

@@ -2,6 +2,7 @@ package common
 
 import (
 	"errors"
+
 	"github.com/multiversx/mx-chain-core-go/core"
 )
 

fifocache/fifocacheSharded.go

@@ -140,6 +140,11 @@ func (c *FIFOShardedCache) MaxSize() int {
 	return c.maxsize
 }
 
+// GetRemovalStatus will return the unknown status because this implementation does not track removed keys
+func (c *FIFOShardedCache) GetRemovalStatus(_ []byte) types.RemovalStatus {
+	return types.UnknownRemovalStatus
+}
+
 // Close does nothing for this cacher implementation
 func (c *FIFOShardedCache) Close() error {
 	return nil

fifocache/fifocacheSharded_test.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/multiversx/mx-chain-storage-go/fifocache"
+	"github.com/multiversx/mx-chain-storage-go/types"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -374,6 +375,22 @@ func TestFIFOShardedCache_CacherRegisterHasOrAddAddedDataHandlerNotAddedShouldNo
 	assert.Equal(t, 1, len(c.AddedDataHandlers()))
 }
 
+func TestFifoShardedCache_GetRemovalStatus(t *testing.T) {
+	t.Parallel()
+
+	key := []byte("key")
+	value := []byte("value")
+
+	cache, _ := fifocache.NewShardedCache(100, 2)
+	assert.Equal(t, types.UnknownRemovalStatus, cache.GetRemovalStatus(nil))
+
+	_ = cache.Put(key, value, 0)
+	assert.Equal(t, types.UnknownRemovalStatus, cache.GetRemovalStatus(nil))
+
+	cache.Remove(key)
+	assert.Equal(t, types.UnknownRemovalStatus, cache.GetRemovalStatus(nil))
+}
+
 func TestFifoShardedCache_ConcurrentOperation(t *testing.T) {
 	fifoCacher, _ := fifocache.NewShardedCache(10000, 3)
 
@@ -382,7 +399,7 @@ func TestFifoShardedCache_ConcurrentOperation(t *testing.T) {
 	wg.Add(numOperations)
 	for i := 0; i < numOperations; i++ {
 		go func(idx int) {
-			switch idx % 16 {
+			switch idx % 17 {
 			case 0:
 				fifoCacher.AddedDataHandlers()
 			case 1:
@@ -415,6 +432,8 @@ func TestFifoShardedCache_ConcurrentOperation(t *testing.T) {
 				fifoCacher.SizeInBytesContained()
 			case 15:
 				fifoCacher.UnRegisterHandler("id")
+			case 16:
+				fifoCacher.GetRemovalStatus(nil)
 			}
 
 			wg.Done()

go.mod

@@ -5,7 +5,7 @@ go 1.20
 require (
 	github.com/hashicorp/golang-lru v0.6.0
 	github.com/multiversx/concurrent-map v0.1.4
-	github.com/multiversx/mx-chain-core-go v1.2.16
+	github.com/multiversx/mx-chain-core-go v1.2.17-0.20230919104727-566f55d213ab
 	github.com/multiversx/mx-chain-logger-go v1.0.13
 	github.com/stretchr/testify v1.7.2
 	github.com/syndtr/goleveldb v1.0.1-0.20220721030215-126854af5e6d

go.sum

@@ -41,8 +41,8 @@ github.com/mr-tron/base58 v1.2.0 h1:T/HDJBh4ZCPbU39/+c3rRvE0uKBQlU27+QI8LJ4t64o=
 github.com/mr-tron/base58 v1.2.0/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc=
 github.com/multiversx/concurrent-map v0.1.4 h1:hdnbM8VE4b0KYJaGY5yJS2aNIW9TFFsUYwbO0993uPI=
 github.com/multiversx/concurrent-map v0.1.4/go.mod h1:8cWFRJDOrWHOTNSqgYCUvwT7c7eFQ4U2vKMOp4A/9+o=
-github.com/multiversx/mx-chain-core-go v1.2.16 h1:m0hUNmZQjGJxKDLQOHoM9jSaeDfVTbyd+mqiS8+NckE=
-github.com/multiversx/mx-chain-core-go v1.2.16/go.mod h1:BILOGHUOIG5dNNX8cgkzCNfDaVtoYrJRYcPnpxRMH84=
+github.com/multiversx/mx-chain-core-go v1.2.17-0.20230919104727-566f55d213ab h1:FO/YYB/cWG6ruHOrnQq6Dn4GEnK06I2KlrwtE2AUcKM=
+github.com/multiversx/mx-chain-core-go v1.2.17-0.20230919104727-566f55d213ab/go.mod h1:B5zU4MFyJezmEzCsAHE9YNULmGCm2zbPHvl9hazNxmE=
 github.com/multiversx/mx-chain-logger-go v1.0.13 h1:eru/TETo0MkO4ZTnXsQDKf4PBRpAXmqjT02klNT/JnY=
 github.com/multiversx/mx-chain-logger-go v1.0.13/go.mod h1:MZJhTAtZTJxT+yK2EHc4ZW3YOHUc1UdjCD0iahRNBZk=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=

immunitycache/cache.go

@@ -321,6 +321,11 @@ func (ic *ImmunityCache) Diagnose(_ bool) {
 	)
 }
 
+// GetRemovalStatus will return the unknown status because this implementation does not track removed keys
+func (ic *ImmunityCache) GetRemovalStatus(_ []byte) types.RemovalStatus {
+	return types.UnknownRemovalStatus
+}
+
 // Close does nothing for this cacher implementation
 func (ic *ImmunityCache) Close() error {
 	return nil

immunitycache/cache_test.go

@@ -9,6 +9,7 @@ import (
 
 	logger "github.com/multiversx/mx-chain-logger-go"
 	"github.com/multiversx/mx-chain-storage-go/common"
+	"github.com/multiversx/mx-chain-storage-go/types"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -336,6 +337,22 @@ func TestImmunityCache_ForgetCapacityHadBeenReachedInThePast(t *testing.T) {
 	require.Equal(t, uint64(0), cache.numCapacityReachedOccurrences.GetUint64())
 }
 
+func TestImmunityCache_GetRemovalStatus(t *testing.T) {
+	t.Parallel()
+
+	key := []byte("key")
+	value := []byte("value")
+
+	cache := newCacheToTest(1, 4, 1000)
+	assert.Equal(t, types.UnknownRemovalStatus, cache.GetRemovalStatus(nil))
+
+	_ = cache.Put(key, value, 0)
+	assert.Equal(t, types.UnknownRemovalStatus, cache.GetRemovalStatus(nil))
+
+	cache.Remove(key)
+	assert.Equal(t, types.UnknownRemovalStatus, cache.GetRemovalStatus(nil))
+}
+
 func newCacheToTest(numChunks uint32, maxNumItems uint32, numMaxBytes uint32) *ImmunityCache {
 	cache, err := NewImmunityCache(CacheConfig{
 		Name:                        "test",

leveldb/batch.go

@@ -10,19 +10,21 @@ import (
 var _ types.Batcher = (*batch)(nil)
 
 type batch struct {
-	batch       *leveldb.Batch
-	cachedData  map[string][]byte
-	removedData map[string]struct{}
-	mutBatch    sync.RWMutex
+	batch             *leveldb.Batch
+	cachedData        map[string][]byte
+	removedData       map[string]struct{}
+	previouslyRemoved map[string]struct{}
+	mutBatch          sync.RWMutex
 }
 
 // NewBatch creates a batch
-func NewBatch() *batch {
+func NewBatch(previouslyRemoved map[string]struct{}) *batch {
 	return &batch{
-		batch:       &leveldb.Batch{},
-		cachedData:  make(map[string][]byte),
-		removedData: make(map[string]struct{}),
-		mutBatch:    sync.RWMutex{},
+		batch:             &leveldb.Batch{},
+		cachedData:        make(map[string][]byte),
+		removedData:       make(map[string]struct{}),
+		previouslyRemoved: previouslyRemoved,
+		mutBatch:          sync.RWMutex{},
 	}
 }
 
@@ -69,6 +71,14 @@ func (b *batch) IsRemoved(key []byte) bool {
 	defer b.mutBatch.RUnlock()
 
 	_, found := b.removedData[string(key)]
+	if found {
+		return true
+	}
+	_, found = b.cachedData[string(key)]
+	if found {
+		return false
+	}
+	_, found = b.previouslyRemoved[string(key)]
 
 	return found
 }

leveldb/export_test.go

@@ -0,0 +1,6 @@
+package leveldb
+
+// PutBatch will call the unexported putBatch function
+func (s *SerialDB) PutBatch() {
+	_ = s.putBatch()
+}

leveldb/leveldb.go

@@ -220,7 +220,7 @@ func (s *DB) Has(key []byte) error {
 
 // CreateBatch returns a batcher to be used for batch writing data to the database
 func (s *DB) createBatch() types.Batcher {
-	return NewBatch()
+	return NewBatch(make(map[string]struct{}))
 }
 
 // putBatch writes the Batch data into the database

leveldb/leveldbSerial.go

@@ -80,7 +80,7 @@ func NewSerialDB(path string, batchDelaySeconds int, maxBatchSize int, maxOpenFi
 		closer:            closing.NewSafeChanCloser(),
 	}
 
-	dbStore.batch = NewBatch()
+	dbStore.batch = NewBatch(make(map[string]struct{}))
 
 	go dbStore.batchTimeoutHandle(ctx)
 	go dbStore.processLoop(ctx)
@@ -246,7 +246,7 @@ func (s *SerialDB) putBatch() error {
 		return common.ErrInvalidBatch
 	}
 	s.sizeBatch = 0
-	s.batch = NewBatch()
+	s.batch = NewBatch(dbBatch.removedData)
 	s.mutBatch.Unlock()
 
 	ch := make(chan error)

leveldb/leveldbSerial_test.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"math/big"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -358,3 +359,87 @@ func TestSerialDB_ConcurrentOperations(t *testing.T) {
 
 	wg.Wait()
 }
+
+func TestSerialDB_PutRemoveGet(t *testing.T) {
+	if testing.Short() {
+		t.Skip("this is not a short test")
+	}
+
+	ldb := createSerialLevelDb(t, 100000, 1000000, 10)
+
+	numKeys := 10000
+	for i := 0; i < numKeys; i++ {
+		_ = ldb.Put([]byte(fmt.Sprintf("key %d", i)), []byte("val"))
+	}
+
+	time.Sleep(time.Second * 2)
+
+	numErr := uint32(0)
+
+	for i := 0; i < numKeys; i++ {
+		key := []byte(fmt.Sprintf("key %d", i))
+
+		recoveredVal, _ := ldb.Get(key)
+		assert.NotEmpty(t, recoveredVal)
+
+		wg := &sync.WaitGroup{}
+		wg.Add(2)
+
+		// emulate the following scenario:
+		// the sequence Remove(key) -> Get(key) is done while the putBatch is called. So the actual edgecase is
+		// go routine 1: Remove(key) -----------------> Get(key)
+		// go routine 2:                putBatch()
+
+		go func() {
+			time.Sleep(time.Millisecond * 1)
+			ldb.PutBatch()
+			wg.Done()
+		}()
+		go func() {
+			_ = ldb.Remove(key)
+
+			time.Sleep(time.Millisecond * 1)
+
+			recoveredVal2, _ := ldb.Get(key)
+			if len(recoveredVal2) > 0 {
+				// the key-value was not removed
+				atomic.AddUint32(&numErr, 1)
+			}
+
+			wg.Done()
+		}()
+
+		wg.Wait()
+
+		require.Zero(t, atomic.LoadUint32(&numErr), "iteration %d out of %d", i, numKeys)
+	}
+
+	_ = ldb.Close()
+}
+
+func TestSerialDB_PutRemovePutHas(t *testing.T) {
+	ldb := createSerialLevelDb(t, 100000, 1000000, 10)
+
+	key := []byte("key")
+	value := []byte("value")
+
+	_ = ldb.Put(key, value)
+
+	// manually put the <key, value> pair in storage
+	ldb.PutBatch()
+	time.Sleep(time.Second)
+	assert.Nil(t, ldb.Has(key)) // key was found
+
+	// we now remove the key
+	_ = ldb.Remove(key)
+
+	// manually delete the key from the storage
+	ldb.PutBatch()
+	time.Sleep(time.Second)
+	assert.NotNil(t, ldb.Has(key)) // missing key
+
+	_ = ldb.Put(key, value)     // put the key again
+	assert.Nil(t, ldb.Has(key)) // key was found
+
+	_ = ldb.Close()
+}

lrucache/export_test.go

@@ -1,5 +1,6 @@
 package lrucache
 
+// AddedDataHandlers -
 func (c *lruCache) AddedDataHandlers() map[string]func(key []byte, value interface{}) {
 	return c.mapDataHandlers
 }

lrucache/lrucache.go

@@ -184,6 +184,11 @@ func (c *lruCache) MaxSize() int {
 	return c.maxsize
 }
 
+// GetRemovalStatus will return the unknown status because this implementation does not track removed keys
+func (c *lruCache) GetRemovalStatus(_ []byte) types.RemovalStatus {
+	return types.UnknownRemovalStatus
+}
+
 // Close does nothing for this cacher implementation
 func (c *lruCache) Close() error {
 	return nil

lrucache/lrucache_test.go

@@ -8,9 +8,9 @@ import (
 	"time"
 
 	"github.com/multiversx/mx-chain-core-go/core/check"
-	"github.com/multiversx/mx-chain-core-go/storage"
 	"github.com/multiversx/mx-chain-storage-go/common"
 	"github.com/multiversx/mx-chain-storage-go/lrucache"
+	"github.com/multiversx/mx-chain-storage-go/types"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -421,7 +421,7 @@ func TestLRUCache_CloseShouldNotErr(t *testing.T) {
 }
 
 type cacheWrapper struct {
-	c storage.Cacher
+	c types.Cacher
 }
 
 func newCacheWrapper() *cacheWrapper {
@@ -459,3 +459,19 @@ func TestLruCache_LenDuringEviction(t *testing.T) {
 		assert.Fail(t, "test failed, deadlock occurred")
 	}
 }
+
+func TestLruCache_GetRemovalStatus(t *testing.T) {
+	t.Parallel()
+
+	key := []byte("key")
+	value := []byte("value")
+
+	cache, _ := lrucache.NewCache(2)
+	assert.Equal(t, types.UnknownRemovalStatus, cache.GetRemovalStatus(nil))
+
+	_ = cache.Put(key, value, 0)
+	assert.Equal(t, types.UnknownRemovalStatus, cache.GetRemovalStatus(nil))
+
+	cache.Remove(key)
+	assert.Equal(t, types.UnknownRemovalStatus, cache.GetRemovalStatus(nil))
+}