Honor current TTL in DMap.Incr and DMap.Decr methods

integration_test.go

@@ -93,7 +93,7 @@ func TestIntegration_NodesJoinOrLeftDuringQuery(t *testing.T) {
 }
 
 func TestIntegration_DMap_Cache_Eviction_LRU_MaxKeys(t *testing.T) {
-	var maxKeys = 100000
+	maxKeys := 100000
 	newConfig := func() *config.Config {
 		c := config.New("local")
 		c.PartitionCount = config.DefaultPartitionCount
@@ -142,7 +142,7 @@ func TestIntegration_DMap_Cache_Eviction_LRU_MaxKeys(t *testing.T) {
 }
 
 func TestIntegration_DMap_Cache_Eviction_MaxKeys(t *testing.T) {
-	var maxKeys = 100000
+	maxKeys := 100000
 	newConfig := func() *config.Config {
 		c := config.New("local")
 		c.PartitionCount = config.DefaultPartitionCount
@@ -198,7 +198,7 @@ func TestIntegration_DMap_Cache_Eviction_MaxKeys(t *testing.T) {
 }
 
 func TestIntegration_DMap_Cache_Eviction_MaxIdleDuration(t *testing.T) {
-	var maxKeys = 100000
+	maxKeys := 100000
 	newConfig := func() *config.Config {
 		c := config.New("local")
 		c.PartitionCount = config.DefaultPartitionCount
@@ -247,7 +247,7 @@ func TestIntegration_DMap_Cache_Eviction_MaxIdleDuration(t *testing.T) {
 }
 
 func TestIntegration_DMap_Cache_Eviction_TTLDuration(t *testing.T) {
-	var maxKeys = 100000
+	maxKeys := 100000
 	newConfig := func() *config.Config {
 		c := config.New("local")
 		c.PartitionCount = config.DefaultPartitionCount
@@ -285,7 +285,7 @@ func TestIntegration_DMap_Cache_Eviction_TTLDuration(t *testing.T) {
 	var total int
 
 	for i := 0; i < maxKeys; i++ {
-		_, err = dm.Get(ctx, fmt.Sprintf("mykey-%d", i))
+		_, err := dm.Get(ctx, fmt.Sprintf("mykey-%d", i))
 		if err == ErrKeyNotFound {
 			err = nil
 			total++
@@ -296,7 +296,7 @@ func TestIntegration_DMap_Cache_Eviction_TTLDuration(t *testing.T) {
 }
 
 func TestIntegration_DMap_Cache_Eviction_LRU_MaxInuse(t *testing.T) {
-	var maxKeys = 100000
+	maxKeys := 100000
 	newConfig := func() *config.Config {
 		c := config.New("local")
 		c.PartitionCount = config.DefaultPartitionCount

internal/dmap/atomic.go

@@ -18,29 +18,31 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
+
 	"github.com/buraksezer/olric/internal/protocol"
 	"github.com/buraksezer/olric/internal/resp"
 	"github.com/buraksezer/olric/internal/util"
 	"github.com/buraksezer/olric/pkg/storage"
 )
 
-func (dm *DMap) loadCurrentAtomicInt(e *env) (int, error) {
+func (dm *DMap) loadCurrentAtomicInt(e *env) (int, int64, error) {
 	entry, err := dm.Get(e.ctx, e.key)
 	if errors.Is(err, ErrKeyNotFound) {
-		return 0, nil
+		return 0, 0, nil
 	}
 	if err != nil {
-		return 0, err
+		return 0, 0, err
 	}
 
 	if entry == nil {
-		return 0, nil
+		return 0, 0, nil
 	}
 	nr, err := util.ParseInt(entry.Value(), 10, 64)
 	if err != nil {
-		return 0, nil
+		return 0, 0, nil
 	}
-	return int(nr), nil
+	return int(nr), entry.TTL(), nil
 }
 
 func (dm *DMap) atomicIncrDecr(cmd string, e *env, delta int) (int, error) {
@@ -53,7 +55,7 @@ func (dm *DMap) atomicIncrDecr(cmd string, e *env, delta int) (int, error) {
 		}
 	}()
 
-	current, err := dm.loadCurrentAtomicInt(e)
+	current, ttl, err := dm.loadCurrentAtomicInt(e)
 	if err != nil {
 		return 0, err
 	}
@@ -79,6 +81,10 @@ func (dm *DMap) atomicIncrDecr(cmd string, e *env, delta int) (int, error) {
 		pool.Put(valueBuf)
 	}()
 
+	if ttl != 0 {
+		e.putConfig.HasPX = true
+		e.putConfig.PX = time.Until(time.UnixMilli(ttl))
+	}
 	err = dm.put(e)
 	if err != nil {
 		return 0, err

internal/dmap/atomic_test.go

@@ -20,6 +20,7 @@ import (
 	"sync"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/buraksezer/olric/internal/protocol"
 	"github.com/buraksezer/olric/internal/resp"
@@ -29,6 +30,36 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
+func TestDMap_loadCurrentAtomicInt(t *testing.T) {
+	cluster := testcluster.New(NewService)
+	s := cluster.AddMember(nil).(*Service)
+	defer cluster.Shutdown()
+
+	ctx := context.Background()
+	key := "incr"
+
+	ttlDuration := time.Second * 5
+	s.config.DMaps.TTLDuration = time.Second * 5
+
+	dm, err := s.NewDMap("atomic_test")
+	require.NoError(t, err)
+
+	_, err = dm.Incr(ctx, key, 1)
+	if err != nil {
+		s.log.V(2).Printf("[ERROR] Failed to call Incr: %v", err)
+		return
+	}
+
+	e := newEnv(ctx)
+	e.dmap = dm.name
+	e.key = key
+	_, ttl, err := dm.loadCurrentAtomicInt(e)
+	require.NoError(t, err)
+
+	<-time.After(time.Millisecond * 500)
+	require.WithinDuration(t, time.UnixMilli(ttl), time.Now(), ttlDuration)
+}
+
 func TestDMap_Atomic_Incr(t *testing.T) {
 	cluster := testcluster.New(NewService)
 	s := cluster.AddMember(nil).(*Service)

internal/dmap/get.go

@@ -249,7 +249,7 @@ func (dm *DMap) readRepair(winner *version, versions []*version) {
 			}
 
 			f.Lock()
-			e := newEnv(nil)
+			e := newEnv(context.Background())
 			e.hkey = hkey
 			e.fragment = f
 			err = dm.putEntryOnFragment(e, winner.entry)