Uses a dynamic MinID for trimming

Signed-off-by: Elena Kolevska <[email protected]>

Author: elena-kolevska

common/component/redis/redis.go

@@ -82,7 +82,7 @@ type RedisClient interface {
 	ConfigurationSubscribe(ctx context.Context, args *ConfigurationSubscribeArgs)
 	SetNX(ctx context.Context, key string, value interface{}, expiration time.Duration) (*bool, error)
 	EvalInt(ctx context.Context, script string, keys []string, args ...interface{}) (*int, error, error)
-	XAdd(ctx context.Context, stream string, maxLenApprox int64, minIDApprox string, values map[string]interface{}) (string, error)
+	XAdd(ctx context.Context, stream string, maxLenApprox int64, streamTTL string, values map[string]interface{}) (string, error)
 	XGroupCreateMkStream(ctx context.Context, stream string, group string, start string) error
 	XAck(ctx context.Context, stream string, group string, messageID string) error
 	XReadGroupResult(ctx context.Context, group string, consumer string, streams []string, count int64, block time.Duration) ([]RedisXStream, error)

common/component/redis/settings.go

@@ -102,8 +102,8 @@ type Settings struct {
 	// The max len of stream
 	MaxLenApprox int64 `mapstructure:"maxLenApprox" mdonly:"pubsub"`
 
-	// Min ID of a stream (stream entry IDs lower than this threshold will be evicted)
-	MinIDApprox string `mapstructure:"minIDApprox" mdonly:"pubsub"`
+	// The TTL of stream entries
+	StreamTTL time.Duration `mapstructure:"streamTTL" mdonly:"pubsub"`
 
 	// EntraID / AzureAD Authentication based on the shared code which essentially uses the DefaultAzureCredential
 	// from the official Azure Identity SDK for Go
@@ -130,6 +130,15 @@ func (s *Settings) SetCertificate(fn func(cert *tls.Certificate)) error {
 	return nil
 }
 
+func (s *Settings) GetMinID(now time.Time) string {
+	// If StreamTTL is not set, return empty string (no trimming)
+	if s.StreamTTL == 0 {
+		return ""
+	}
+
+	return fmt.Sprintf("%d-1", now.Add(-s.StreamTTL).UnixMilli())
+}
+
 type Duration time.Duration
 
 func (r *Duration) DecodeString(value string) error {

common/component/redis/settings_test.go

@@ -3,6 +3,7 @@ package redis
 import (
 	"crypto/tls"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 )
@@ -39,4 +40,34 @@ func TestSettings(t *testing.T) {
 		require.NoError(t, err)
 		require.NotNil(t, c)
 	})
+
+	t.Run("stream TTL", func(t *testing.T) {
+		fixedTime := time.Date(2025, 3, 14, 0o1, 59, 26, 0, time.UTC)
+
+		tests := []struct {
+			name      string
+			streamTTL time.Duration
+			want      string
+		}{
+			{
+				name:      "with one hour TTL",
+				streamTTL: time.Hour,
+				want:      "1741913966000-1",
+			},
+			{
+				name:      "with zero TTL",
+				streamTTL: 0,
+				want:      "",
+			},
+		}
+
+		for _, tt := range tests {
+			t.Run(tt.name, func(t *testing.T) {
+				settings := &Settings{
+					StreamTTL: tt.streamTTL,
+				}
+				require.Equal(t, tt.want, settings.GetMinID(fixedTime))
+			})
+		}
+	})
 }

pubsub/redis/metadata.yaml

@@ -175,18 +175,18 @@ metadata:
     type: string
   - name: maxLenApprox
     required: false
-    description: Maximum number of items inside a stream.The old entries are automatically evicted when the specified length is reached, so that the stream is left at a constant size. Defaults to unlimited.
+    description: Maximum number of items inside a stream. The old entries are automatically evicted when the specified length is reached, so that the stream is left at a constant size. Defaults to unlimited.
     example: "10000"
     type: number
-  - name: minIDApprox
+  - name: streamTTL
     required: false
     description: |
-      Minimum ID threshold for the stream. Entries with IDs smaller than the specified threshold are evicted, 
-      keeping the stream trimmed to only contain entries with IDs greater than or equal to this value. 
-      This can be used for time-based eviction since Redis stream IDs are in the format {timestamp-sequence}, 
-      where timestamp is in milliseconds since epoch. Defaults to none.
-    example: "1742397526022-0"
-    type: string
+      TTL duration for stream entries. Entries older than this duration will be evicted.
+      This is an approximate value, as it's implemented using Redis stream's MINID trimming with the '~' modifier.
+      The actual retention may include slightly more entries than strictly defined by the TTL,
+      as Redis optimizes the trimming operation for efficiency by potentially keeping some additional entries.
+    example: "30d"
+    type: duration
 
 builtinAuthenticationProfiles:
   - name: "azuread"

pubsub/redis/redis.go

@@ -38,7 +38,7 @@ const (
 	queueDepth        = "queueDepth"
 	concurrency       = "concurrency"
 	maxLenApprox      = "maxLenApprox"
-	minIDApprox       = "minIDApprox"
+	streamTTL         = "streamTTL"
 )
 
 // redisStreams handles consuming from a Redis stream using
@@ -113,7 +113,7 @@ func (r *redisStreams) Publish(ctx context.Context, req *pubsub.PublishRequest)
 		redisPayload["metadata"] = serializedMetadata
 	}
 
-	_, err := r.client.XAdd(ctx, req.Topic, r.clientSettings.MaxLenApprox, r.clientSettings.MinIDApprox, redisPayload)
+	_, err := r.client.XAdd(ctx, req.Topic, r.clientSettings.MaxLenApprox, r.clientSettings.GetMinID(time.Now()), redisPayload)
 	if err != nil {
 		return fmt.Errorf("redis streams: error from publish: %s", err)
 	}

pubsub/redis/redis_test.go

@@ -19,6 +19,7 @@ import (
 	"strconv"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -35,7 +36,7 @@ func getFakeProperties() map[string]string {
 		consumerID:   "fakeConsumer",
 		enableTLS:    "true",
 		maxLenApprox: "1000",
-		minIDApprox:  "1742397526022-0",
+		streamTTL:    "1h",
 	}
 }
 
@@ -55,7 +56,7 @@ func TestParseRedisMetadata(t *testing.T) {
 		require.NoError(t, err)
 		assert.Equal(t, fakeProperties[consumerID], m.ConsumerID)
 		assert.Equal(t, int64(1000), m.MaxLenApprox)
-		assert.Equal(t, "1742397526022-0", m.MinIDApprox)
+		assert.Equal(t, 1*time.Hour, m.StreamTTL)
 	})
 
 	// TODO: fix the code to return the error for the missing property to make this test work