[feat] swap: mark unsupported commands (XADD non-deterministic varian…

src/commands_swap/hexpire.json.merge

@@ -0,0 +1,6 @@
+{
+    "HEXPIRE": {
+        "swap_command_flags": ["SWAP_HARD_BLOCKED"],
+        "intention": "SWAP_NOP"
+    }
+}

src/commands_swap/hexpireat.json.merge

@@ -0,0 +1,6 @@
+{
+    "HEXPIREAT": {
+        "swap_command_flags": ["SWAP_HARD_BLOCKED"],
+        "intention": "SWAP_NOP"
+    }
+}

src/commands_swap/hgetex.json.merge

@@ -0,0 +1,6 @@
+{
+    "HGETEX": {
+        "swap_command_flags": ["SWAP_HARD_BLOCKED"],
+        "intention": "SWAP_NOP"
+    }
+}

src/commands_swap/hpersist.json.merge

@@ -0,0 +1,6 @@
+{
+    "HPERSIST": {
+        "swap_command_flags": ["SWAP_HARD_BLOCKED"],
+        "intention": "SWAP_NOP"
+    }
+}

src/commands_swap/hpexpire.json.merge

@@ -0,0 +1,6 @@
+{
+    "HPEXPIRE": {
+        "swap_command_flags": ["SWAP_HARD_BLOCKED"],
+        "intention": "SWAP_NOP"
+    }
+}

src/commands_swap/hpexpireat.json.merge

@@ -0,0 +1,6 @@
+{
+    "HPEXPIREAT": {
+        "swap_command_flags": ["SWAP_HARD_BLOCKED"],
+        "intention": "SWAP_NOP"
+    }
+}

src/commands_swap/hsetex.json.merge

@@ -0,0 +1,6 @@
+{
+    "HSETEX": {
+        "swap_command_flags": ["SWAP_HARD_BLOCKED"],
+        "intention": "SWAP_NOP"
+    }
+}

src/commands_swap/xack.json.merge

@@ -1,6 +1,6 @@
 {
     "XACK": {
-        "swap_command_flags": [],
+        "swap_command_flags": ["SWAP_SOFT_BLOCKED"],
         "intention": "SWAP_NOP"
     }
 }

src/commands_swap/xackdel.json.merge

@@ -1,6 +1,6 @@
 {
     "XACKDEL": {
-        "swap_command_flags": [],
+        "swap_command_flags": ["SWAP_HARD_BLOCKED"],
         "intention": "SWAP_NOP"
     }
 }

src/commands_swap/xadd.json.merge

@@ -1,6 +1,8 @@
 {
     "XADD": {
-        "swap_command_flags": [],
+        "swap_command_flags": [
+            "SWAP_SOFT_BLOCKED"
+        ],
         "intention": "SWAP_NOP"
     }
 }

src/commands_swap/xautoclaim.json.merge

@@ -1,6 +1,6 @@
 {
     "XAUTOCLAIM": {
-        "swap_command_flags": [],
+        "swap_command_flags": ["SWAP_SOFT_BLOCKED"],
         "intention": "SWAP_NOP"
     }
 }

src/commands_swap/xclaim.json.merge

@@ -1,6 +1,6 @@
 {
     "XCLAIM": {
-        "swap_command_flags": [],
+        "swap_command_flags": ["SWAP_SOFT_BLOCKED"],
         "intention": "SWAP_NOP"
     }
 }

src/commands_swap/xdel.json.merge

@@ -1,6 +1,6 @@
 {
     "XDEL": {
-        "swap_command_flags": [],
+        "swap_command_flags": ["SWAP_SOFT_BLOCKED"],
         "intention": "SWAP_NOP"
     }
 }

src/commands_swap/xdelex.json.merge

@@ -1,6 +1,6 @@
 {
     "XDELEX": {
-        "swap_command_flags": [],
+        "swap_command_flags": ["SWAP_HARD_BLOCKED"],
         "intention": "SWAP_NOP"
     }
 }

src/commands_swap/xgroup.json.merge

@@ -1,6 +1,6 @@
 {
     "XGROUP": {
-        "swap_command_flags": [],
+        "swap_command_flags": ["SWAP_SOFT_BLOCKED"],
         "intention": "SWAP_NOP"
     }
 }

src/commands_swap/xreadgroup.json.merge

@@ -1,6 +1,6 @@
 {
     "XREADGROUP": {
-        "swap_command_flags": [],
+        "swap_command_flags": ["SWAP_SOFT_BLOCKED"],
         "intention": "SWAP_NOP"
     }
 }

src/commands_swap/xrevrange.json.merge

@@ -1,6 +1,6 @@
 {
     "XREVRANGE": {
-        "swap_command_flags": [],
+        "swap_command_flags": ["SWAP_SOFT_BLOCKED"],
         "intention": "SWAP_NOP"
     }
 }

src/commands_swap/xsetid.json.merge

@@ -1,6 +1,6 @@
 {
     "XSETID": {
-        "swap_command_flags": [],
+        "swap_command_flags": ["SWAP_SOFT_BLOCKED"],
         "intention": "SWAP_NOP"
     }
 }

src/commands_swap/xtrim.json.merge

@@ -1,6 +1,6 @@
 {
     "XTRIM": {
-        "swap_command_flags": [],
+        "swap_command_flags": ["SWAP_SOFT_BLOCKED"],
         "intention": "SWAP_NOP"
     }
 }

src/config.c

@@ -3588,6 +3588,7 @@ standardConfig static_configs[] = {
     createBoolConfig("swap-rdb-bitmap-encode-enabled", NULL, MODIFIABLE_CONFIG, server.swap_rdb_bitmap_encode_enabled, 1, NULL, NULL),
     createBoolConfig("swap-bitmap-subkeys-enabled", NULL, MODIFIABLE_CONFIG, server.swap_bitmap_subkeys_enabled, 1, NULL, NULL),
     createBoolConfig("swap-ttl-compact-enabled", NULL, MODIFIABLE_CONFIG, server.swap_ttl_compact_enabled, 1, NULL, NULL),
+    createBoolConfig("swap-soft-block-cmd-enabled", NULL, MODIFIABLE_CONFIG, server.swap_soft_block_cmd_enabled, 0, NULL, NULL),
     createBoolConfig("rocksdb.data.cache_index_and_filter_blocks", "rocksdb.cache_index_and_filter_blocks", IMMUTABLE_CONFIG, server.rocksdb_data_cache_index_and_filter_blocks, 0, NULL, NULL),
     createBoolConfig("rocksdb.meta.cache_index_and_filter_blocks", NULL, IMMUTABLE_CONFIG, server.rocksdb_meta_cache_index_and_filter_blocks, 0, NULL, NULL),
     createBoolConfig("rocksdb.enable_pipelined_write", NULL, IMMUTABLE_CONFIG, server.rocksdb_enable_pipelined_write, 0, NULL, NULL),

src/ctrip_swap.h

@@ -2093,6 +2093,10 @@ void swapApplySwapInfo(int swap_info_argc, sds *swap_info_argv);
 int submitReplClientRequests(client *c);
 sds genSwapReplInfoString(sds info);
 
+/* Command blocking */
+sds genSwapCommandInfoString(sds info);
+int swapCmdShouldBlock(client *c, struct redisCommand *cmd);
+
 /* Swap */
 void swapInit(void);
 int dbSwap(client *c);

src/ctrip_swap_cmd.c

@@ -1417,6 +1417,36 @@ void swapInfoCommand(client *c) {
     addReply(c,shared.ok);
 }
 
+/* Return 1 when cmd must be rejected. The swap check lives inside so
+ * callers do not wrap the call in ENABLE_SWAP. */
+int swapCmdShouldBlock(client *c, struct redisCommand *cmd) {
+
+    if (!cmd ||
+        !(cmd->flags & (CMD_SWAP_HARD_BLOCKED|CMD_SWAP_SOFT_BLOCKED)))
+        return 0;
+
+    if (c && (c->flags & CLIENT_MASTER)) {
+        server.swap_master_allowed_cmd_count++;
+        return 0;
+    }
+    /* AOF replay applies commands that were already accepted. */
+    if (c && c->id == CLIENT_ID_AOF)
+        return 0;
+
+    /* reject_blocked: local client, command refused.
+     * allowed_blocked: local client, soft-blocked command executed. */
+    if (cmd->flags & CMD_SWAP_HARD_BLOCKED) {
+        server.swap_reject_blocked_cmd_count++;
+        return 1;
+    }
+    if (server.swap_soft_block_cmd_enabled) {
+        server.swap_reject_blocked_cmd_count++;
+        return 1;
+    }
+    server.swap_allowed_blocked_cmd_count++;
+    return 0;
+}
+
 #ifdef REDIS_TEST
 
 void rewriteResetClientCommandCString(client *c, int argc, ...) {

src/ctrip_swap_server.h

@@ -38,6 +38,8 @@
 #define CMD_SWAP_DATATYPE_ZSET (1ULL<<35)
 #define CMD_SWAP_DATATYPE_LIST (1ULL<<36)
 #define CMD_SWAP_DATATYPE_BITMAP (1ULL<<37)
+#define CMD_SWAP_HARD_BLOCKED (1ULL<<38)
+#define CMD_SWAP_SOFT_BLOCKED (1ULL<<39)
 
 /* CHECK: CLIENT_REPL_RDBONLY is the last CLIENT_xx flag */
 #define CLIENT_SWAPPING (1ULL<<53) /* The client is waiting swap. */
@@ -310,6 +312,10 @@ typedef struct swapBatchLimitsConfig {
     /* for swap.info command, which propagate system info to replica */ \
     int swap_swap_info_supported; \
     int swap_swap_info_propagate_mode; \
+    int swap_soft_block_cmd_enabled; \
+    long long swap_reject_blocked_cmd_count; \
+    long long swap_allowed_blocked_cmd_count; \
+    long long swap_master_allowed_cmd_count; \
     unsigned long long swap_swap_info_slave_period;     /* Master send cmd swap.info to the slave every N seconds */
 
 #ifdef __APPLE__

src/ctrip_swap_stat.c

@@ -348,6 +348,18 @@ void trackSwapInstantaneousMetrics(long long current_base, long long factor) {
     trackSwapRateLimitInstantaneousMetrics(current_base, factor);
 }
 
+sds genSwapCommandInfoString(sds info) {
+    info = sdscatprintf(info,
+        "swap_cmd_stats:"
+        "reject_blocked=%lld,"
+        "allowed_blocked=%lld,"
+        "master_allowed=%lld\r\n",
+        server.swap_reject_blocked_cmd_count,
+        server.swap_allowed_blocked_cmd_count,
+        server.swap_master_allowed_cmd_count);
+    return info;
+}
+
 sds genSwapInfoString(sds info) {
     info = genSwapStorageInfoString(info);
     info = genSwapHitInfoString(info);
@@ -364,6 +376,7 @@ sds genSwapInfoString(sds info) {
     info = genSwapPersistInfoString(info);
     info = genSwapBitmapStringSwitchedInfoString(info);
     info = genSwapTtlCompactInfoString(info);
+    info = genSwapCommandInfoString(info);
     return info;
 }
 

src/module.c

@@ -1167,6 +1167,10 @@ int64_t commandFlagsFromString(char *s) {
         else if (!strcasecmp(t,"allow-busy")) flags |= CMD_ALLOW_BUSY;
         else if (!strcasecmp(t, "internal")) flags |= (CMD_INTERNAL|CMD_NOSCRIPT); /* We also disallow internal commands in scripts. */
         else if (!strcasecmp(t,"rewrite-command")) flags |= CMD_GTID_NON_DETERMINISM;
+#ifdef ENABLE_SWAP
+        else if (!strcasecmp(t, "swap-hard-blocked")) flags |= CMD_SWAP_HARD_BLOCKED;   
+        else if (!strcasecmp(t, "swap-soft-blocked")) flags |= CMD_SWAP_SOFT_BLOCKED;
+#endif
         else break;
     }
     sdsfreesplitres(tokens,count);

src/script.c

@@ -634,6 +634,13 @@ void scriptCall(scriptRunCtx *run_ctx, sds *err) {
         goto error;
     }
 
+#ifdef ENABLE_SWAP
+    if (swapCmdShouldBlock(c, cmd)) {
+        *err = sdsnew("Can't execute this command: data type is not supported by swap");
+        goto error;
+    }
+#endif
+
     if (scriptVerifyAllowStale(c, err) != C_OK) {
         goto error;
     }

src/server.c

@@ -4632,6 +4632,13 @@ int processCommand(client *c) {
         rejectCommandFormat(c, "Previous master draining.");
         return C_OK;
     }
+
+    if (swapCmdShouldBlock(c, c->cmd)) {
+        rejectCommandFormat(c,
+            "-ERR Can't execute '%s' command: data type is not supported by swap",
+            c->cmd->fullname);
+        return C_OK;
+    }
 #endif
 
     /* Only allow a subset of commands in the context of Pub/Sub if the
@@ -5296,6 +5303,10 @@ void addReplyFlagsForCommand(client *c, struct redisCommand *cmd) {
         {CMD_ALLOW_BUSY,        "allow_busy"},
         /* {CMD_TOUCHES_ARBITRARY_KEYS,  "TOUCHES_ARBITRARY_KEYS"}, Hidden on purpose */
         {CMD_GTID_NON_DETERMINISM,   "rewrite_command"},
+#ifdef ENABLE_SWAP
+        {CMD_SWAP_HARD_BLOCKED,      "swap_hard_blocked"},
+        {CMD_SWAP_SOFT_BLOCKED,      "swap_soft_blocked"},
+#endif
         {0,NULL}
     };
     addReplyCommandFlags(c, cmd->flags, flagNames);

tests/support/util.tcl

@@ -756,7 +756,11 @@ proc generate_fuzzy_traffic_on_key {key type duration} {
     set zset_commands {ZADD ZCARD ZCOUNT ZINCRBY ZINTERSTORE ZLEXCOUNT ZPOPMAX ZPOPMIN ZRANGE ZRANGEBYLEX ZRANGEBYSCORE ZRANK ZREM ZREMRANGEBYLEX ZREMRANGEBYRANK ZREMRANGEBYSCORE ZREVRANGE ZREVRANGEBYLEX ZREVRANGEBYSCORE ZREVRANK ZSCAN ZSCORE ZUNIONSTORE ZRANDMEMBER}
     set list_commands {LINDEX LINSERT LLEN LPOP LPOS LPUSH LPUSHX LRANGE LREM LSET LTRIM RPOP RPOPLPUSH RPUSH RPUSHX}
     set set_commands {SADD SCARD SDIFF SDIFFSTORE SINTER SINTERSTORE SISMEMBER SMEMBERS SMOVE SPOP SRANDMEMBER SREM SSCAN SUNION SUNIONSTORE}
-    set stream_commands {XACK XADD XCLAIM XDEL XGROUP XINFO XLEN XPENDING XRANGE XREAD XREADGROUP XREVRANGE XTRIM XDELEX XACKDEL}
+    set stream_commands {XACK XADD XCLAIM XDEL XGROUP XINFO XLEN XPENDING XRANGE XREAD XREADGROUP XREVRANGE XTRIM}
+    # XDELEX and XACKDEL are hard blocked on swap.
+    if {!$::swap} {
+        lappend stream_commands XDELEX XACKDEL
+    }
     set vset_commands {VADD VREM}
     set commands [dict create string $string_commands hash $hash_commands zset $zset_commands list $list_commands set $set_commands stream $stream_commands vectorset $vset_commands]