Add snapshot offload substrate

adapter/redis.go

@@ -120,9 +120,10 @@ const (
 	// blocking-command wait loops when no in-process write signal arrives.
 	// Signals cover normal XADD / ZADD / ZINCRBY wakeups immediately; this
 	// interval only bounds missed-signal, wrong-type, and BLOCK-deadline checks.
-	// Keep it below common sub-second BLOCK budgets while reducing idle
-	// fallback scans versus the previous 100ms cadence.
-	defaultRedisBlockWaitFallback = 250 * time.Millisecond
+	// Keep normal producer wakeups event-driven while reducing idle fallback scans
+	// from blocked consumers waiting on empty collections.
+	defaultRedisBlockWaitFallback = time.Second
+	redisFinalTypeCheckTimeout    = 100 * time.Millisecond
 	redisFlushLegacyTimeout       = 10 * time.Minute
 	redisRelayPublishTimeout      = 2 * time.Second
 	redisTraceArgLimit            = 6

adapter/redis_compat_commands_stream_test.go

@@ -185,6 +185,64 @@ func TestRedis_StreamXReadShortBlockReturnsNullNotError(t *testing.T) {
 	}
 }
 
+func TestRedis_StreamXReadBlockChecksWrongTypeAtDeadline(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 3)
+	defer shutdown(nodes)
+
+	rdbReader := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress})
+	defer func() { _ = rdbReader.Close() }()
+	rdbWriter := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress})
+	defer func() { _ = rdbWriter.Close() }()
+	ctx := context.Background()
+
+	key := "stream-block-wrongtype"
+	_, err := rdbWriter.XAdd(ctx, &redis.XAddArgs{
+		Stream: key,
+		ID:     "1-0",
+		Values: []string{"k", "v"},
+	}).Result()
+	require.NoError(t, err)
+
+	type readResult struct {
+		streams []redis.XStream
+		err     error
+	}
+	resultCh := make(chan readResult, 1)
+	go func() {
+		streams, err := rdbReader.XRead(ctx, &redis.XReadArgs{
+			Streams: []string{key, "$"},
+			Count:   1,
+			Block:   2 * time.Second,
+		}).Result()
+		resultCh <- readResult{streams: streams, err: err}
+	}()
+
+	requireStreamWaiterRegistered(t, nodes[0].redisServer.streamWaiters, key)
+	require.NoError(t, rdbWriter.Set(ctx, key, "now-a-string", 0).Err())
+
+	select {
+	case res := <-resultCh:
+		require.Error(t, res.err)
+		require.Contains(t, res.err.Error(), "WRONGTYPE")
+		require.Empty(t, res.streams)
+	case <-time.After(4 * time.Second):
+		t.Fatal("XREAD BLOCK did not return after wrong-type overwrite")
+	}
+}
+
+func requireStreamWaiterRegistered(t *testing.T, reg *keyWaiterRegistry, key string) {
+	t.Helper()
+	require.Eventually(t, func() bool {
+		if reg == nil {
+			return false
+		}
+		reg.mu.Lock()
+		defer reg.mu.Unlock()
+		return len(reg.waiters[key]) > 0
+	}, 2*time.Second, 10*time.Millisecond)
+}
+
 // TestRedis_StreamCommandsRejectWrongType locks down the wrongType
 // detection on the stream fast path: keyTypeAtExpect short-circuits to
 // the slow path when the expected (stream) prefixes return empty, so

adapter/redis_peer_limiter.go

@@ -9,10 +9,12 @@ import (
 )
 
 const (
-	redisPerPeerLimitEnv             = "ELASTICKV_REDIS_PER_PEER_CONNECTIONS"
-	defaultRedisPerPeerConnectionCap = 8
-	redisPeerLimitError              = "ERR max connections per client exceeded"
-	unknownRedisPeer                 = "unknown"
+	redisPerPeerLimitEnv              = "ELASTICKV_REDIS_PER_PEER_CONNECTIONS"
+	defaultRedisProxyPoolPeerCap      = 64
+	defaultRedisDedicatedPeerHeadroom = 64
+	defaultRedisPerPeerConnectionCap  = defaultRedisProxyPoolPeerCap + defaultRedisDedicatedPeerHeadroom
+	redisPeerLimitError               = "ERR max connections per client exceeded"
+	unknownRedisPeer                  = "unknown"
 )
 
 type redisPeerLimiter struct {

adapter/redis_peer_limiter_test.go

@@ -3,13 +3,21 @@ package adapter
 import (
 	"testing"
 
+	"github.com/bootjp/elastickv/proxy"
 	"github.com/stretchr/testify/require"
 )
 
 const (
 	testPeerLimit = 2
 )
 
+func TestRedisPeerLimiterDefaultMatchesProxyPool(t *testing.T) {
+	t.Setenv(redisPerPeerLimitEnv, "")
+	limiter := newDefaultRedisPeerLimiter()
+	require.NotNil(t, limiter)
+	require.Equal(t, proxy.DefaultElasticKVBackendOptions().PoolSize+defaultRedisDedicatedPeerHeadroom, limiter.limit)
+}
+
 func TestRedisPeerLimiterRejectsAndReleases(t *testing.T) {
 	server := NewRedisServer(nil, "", nil, nil, nil, nil, WithRedisPerPeerConnectionLimit(testPeerLimit))
 	c1 := &remoteCommandRecorder{remote: "192.168.0.64:10001"}

adapter/redis_stream_cmds.go

@@ -1275,6 +1275,46 @@ func isXReadIterCtxError(err error) bool {
 	}
 }
 
+func (r *RedisServer) xreadFinalTypeCheck(conn redcon.Conn, req xreadRequest) bool {
+	ctx, cancel := context.WithTimeout(r.handlerContext(), redisFinalTypeCheckTimeout)
+	defer cancel()
+	var err error
+	if ok := r.runWithHeavyCommandSlot(func() {
+		err = r.xreadCheckTypes(ctx, req)
+	}); !ok {
+		return false
+	}
+	if err != nil {
+		if isXReadIterCtxError(err) {
+			return false
+		}
+		writeRedisError(conn, err)
+		return true
+	}
+	return false
+}
+
+func (r *RedisServer) xreadCheckTypes(ctx context.Context, req xreadRequest) error {
+	for _, key := range req.keys {
+		readTS := r.readTS()
+		typ, err := r.keyTypeAtExpect(ctx, key, readTS, redisTypeStream)
+		if err != nil {
+			return err
+		}
+		if typ != redisTypeNone && typ != redisTypeStream {
+			return wrongTypeError()
+		}
+	}
+	return nil
+}
+
+func (r *RedisServer) writeXReadFinalOrNull(conn redcon.Conn, req xreadRequest) {
+	if r.xreadFinalTypeCheck(conn, req) {
+		return
+	}
+	conn.WriteNull()
+}
+
 func (r *RedisServer) xread(conn redcon.Conn, cmd redcon.Command) {
 	req, err := parseXReadRequest(cmd.Args)
 	if err != nil {
@@ -1350,7 +1390,7 @@ func (r *RedisServer) xreadBusyPoll(conn redcon.Conn, req xreadRequest, deadline
 		// return DeadlineExceeded, which we'd then surface as an error.
 		iterTimeout := time.Until(deadline)
 		if iterTimeout <= 0 {
-			conn.WriteNull()
+			r.writeXReadFinalOrNull(conn, req)
 			return
 		}
 		// Cap each iteration at redisDispatchTimeout to avoid holding
@@ -1394,7 +1434,7 @@ func (r *RedisServer) xreadBusyPoll(conn redcon.Conn, req xreadRequest, deadline
 		}
 
 		if !time.Now().Before(deadline) {
-			conn.WriteNull()
+			r.writeXReadFinalOrNull(conn, req)
 			return
 		}
 		waitForBlockedCommandUpdate(handlerCtx, w, deadline, r.blockWaitFallback)