test(jepsen): add ZSet safety workload with model-based checker

.github/workflows/jepsen-test-scheduled.yml

@@ -147,6 +147,16 @@ jobs:
             --max-txn-length ${{ inputs.max-txn-length || '4' }} \
             --ports 63791,63792,63793 \
             --host 127.0.0.1
+      - name: Run Redis ZSet safety Jepsen workload against elastickv
+        working-directory: jepsen
+        timeout-minutes: 10
+        run: |
+          timeout 480 ~/lein run -m elastickv.redis-zset-safety-workload \
+            --time-limit ${{ inputs.time-limit || '150' }} \
+            --rate ${{ inputs.rate || '10' }} \
+            --concurrency ${{ inputs.concurrency || '8' }} \
+            --ports 63791,63792,63793 \
+            --host 127.0.0.1
       - name: Run DynamoDB Jepsen workload against elastickv
         working-directory: jepsen
         timeout-minutes: 10

.github/workflows/jepsen-test.yml

@@ -122,6 +122,11 @@ jobs:
         timeout-minutes: 3
         run: |
           timeout 120 ~/lein run -m elastickv.redis-workload --time-limit 5 --rate 5 --concurrency 5 --ports 63791,63792,63793 --host 127.0.0.1
+      - name: Run Redis ZSet safety Jepsen workload against elastickv
+        working-directory: jepsen
+        timeout-minutes: 3
+        run: |
+          timeout 120 ~/lein run -m elastickv.redis-zset-safety-workload --time-limit 5 --rate 5 --concurrency 5 --ports 63791,63792,63793 --host 127.0.0.1
       - name: Run DynamoDB Jepsen workload against elastickv
         working-directory: jepsen
         timeout-minutes: 3

adapter/redis_compat_commands_stream_test.go

@@ -284,20 +284,21 @@ func TestRedis_StreamXReadLatencyIsConstant(t *testing.T) {
 		total  = 10_000
 		probes = 100
 	)
+	lastID := ""
 	for i := range total {
-		_, err := rdb.XAdd(ctx, &redis.XAddArgs{
+		id, err := rdb.XAdd(ctx, &redis.XAddArgs{
 			Stream: "stream-lat",
-			ID:     fmt.Sprintf("%d-0", 1_000_000+i),
+			ID:     "*",
 			Values: []string{"i", fmt.Sprint(i)},
 		}).Result()
 		require.NoError(t, err)
+		lastID = id
 	}
 
-	afterID := fmt.Sprintf("%d-0", 1_000_000+total-1)
 	measure := func() time.Duration {
 		start := time.Now()
 		streams, err := rdb.XRead(ctx, &redis.XReadArgs{
-			Streams: []string{"stream-lat", afterID},
+			Streams: []string{"stream-lat", lastID},
 			Count:   10,
 			Block:   10 * time.Millisecond,
 		}).Result()

adapter/redis_lua_context.go

@@ -2623,7 +2623,10 @@ func (c *luaScriptContext) cmdZRangeByScoreSlow(key []byte, options luaZRangeByS
 // the whole offset+limit budget on those filtered-out rows and miss
 // the real matches at score > value.
 //
-// The score index is lex-sorted by (userKey, sortableScore, member).
+// The score index is grouped by (userKey, sortableScore). Member bytes
+// follow the score, but the MVCC timestamp suffix means equal-score
+// member ordering is normalized by zsetRangeByScoreFast rather than
+// trusted directly from physical scan order.
 // Conventions:
 //
 //	minBound = -Inf              -> startKey = ZSetScoreScanPrefix(key)
@@ -3738,26 +3741,37 @@ func parseZScoreBound(raw string) (zScoreBound, error) {
 	return zScoreBound{kind: zBoundValue, score: score, inclusive: inclusive}, nil
 }
 
-func scoreInRange(score float64, minBound zScoreBound, maxBound zScoreBound) bool {
-	if minBound.kind == zBoundValue {
-		if minBound.inclusive {
-			if score < minBound.score {
-				return false
-			}
-		} else if score <= minBound.score {
-			return false
+func scoreSatisfiesMinBound(score float64, bound zScoreBound) bool {
+	switch bound.kind {
+	case zBoundValue:
+		if bound.inclusive {
+			return score >= bound.score
 		}
+		return score > bound.score
+	case zBoundPosInf:
+		return math.IsInf(score, +1)
+	default:
+		return true
 	}
-	if maxBound.kind == zBoundValue {
-		if maxBound.inclusive {
-			if score > maxBound.score {
-				return false
-			}
-		} else if score >= maxBound.score {
-			return false
+}
+
+func scoreSatisfiesMaxBound(score float64, bound zScoreBound) bool {
+	switch bound.kind {
+	case zBoundNegInf:
+		return math.IsInf(score, -1)
+	case zBoundValue:
+		if bound.inclusive {
+			return score <= bound.score
 		}
+		return score < bound.score
+	default:
+		return true
 	}
-	return true
+}
+
+func scoreInRange(score float64, minBound zScoreBound, maxBound zScoreBound) bool {
+	return scoreSatisfiesMinBound(score, minBound) &&
+		scoreSatisfiesMaxBound(score, maxBound)
 }
 
 func cloneSetMembers(in map[string]struct{}) map[string]struct{} {

adapter/redis_retry_test.go

@@ -219,6 +219,50 @@ func TestRedisExecLuaCompatRetriesWriteConflict(t *testing.T) {
 	require.Equal(t, 1.0, zset.Entries[0].Score)
 }
 
+func TestRedisZRemWideColumnRemovesMemberAndScoreIndex(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	st := store.NewMVCCStore()
+	coord := newRetryOnceCoordinator(st)
+
+	srv := &RedisServer{
+		store:       st,
+		coordinator: coord,
+		scriptCache: map[string]string{},
+	}
+
+	key := []byte("retry:zrem-wide")
+	addConn := &recordingConn{}
+	srv.zadd(addConn, redcon.Command{Args: [][]byte{
+		[]byte(cmdZAdd), key,
+		[]byte("55"), []byte("m8"),
+		[]byte("-49"), []byte("m6"),
+	}})
+	require.Empty(t, addConn.err)
+	require.Equal(t, int64(2), addConn.int)
+
+	remConn := &recordingConn{}
+	srv.zrem(remConn, redcon.Command{Args: [][]byte{
+		[]byte(cmdZRem), key, []byte("m8"),
+	}})
+	require.Empty(t, remConn.err)
+	require.Equal(t, int64(1), remConn.int)
+
+	readTS := snapshotTS(coord.clock, st)
+	zset, exists, err := srv.loadZSetAt(ctx, key, readTS)
+	require.NoError(t, err)
+	require.True(t, exists)
+	require.Equal(t, []redisZSetEntry{{Member: "m6", Score: -49}}, zset.Entries)
+
+	memberExists, err := st.ExistsAt(ctx, store.ZSetMemberKey(key, []byte("m8")), readTS)
+	require.NoError(t, err)
+	require.False(t, memberExists)
+	scoreExists, err := st.ExistsAt(ctx, store.ZSetScoreKey(key, 55, []byte("m8")), readTS)
+	require.NoError(t, err)
+	require.False(t, scoreExists)
+}
+
 func TestRedisEvalRetriesWriteConflict(t *testing.T) {
 	t.Parallel()