Fix redis crash when duplicate messages are appended to log. (#3316)

src/ray/gcs/redis_module/ray_redis_module.cc

@@ -311,7 +311,23 @@ int TableAppend_DoWrite(RedisModuleCtx *ctx, RedisModuleString **argv, int argc,
     // must be unique, or else we will have gaps in the log.
     // TODO(rkn): We need to get rid of this uniqueness requirement. We can
     // easily have multiple log events with the same message.
-    RAY_CHECK(flags == REDISMODULE_ZADD_ADDED) << "Appended a duplicate entry";
+    if (flags != REDISMODULE_ZADD_ADDED) {
+      // The following code is a workaround to store the data at a new unique
+      // key. This is so redis doesn't crash (we currently have duplicate keys
+      // for error conditions, which get delivered via pubsub).
+      size_t len;
+      const char *id_str = RedisModule_StringPtrLen(id, &len);
+      RAY_LOG(INFO) << "Duplicate key: " << std::string(id_str, len);
+      // Store the value into a unique new key, just to keep track of it and
+      // make sure the log size grows.
+      std::string postfix = std::to_string(index);
+      RedisModuleString *new_id =
+          RedisString_Format(ctx, "%S:%b", id, postfix.data(), postfix.size());
+      RedisModuleKey *new_key = OpenPrefixedKey(
+          ctx, prefix_str, new_id, REDISMODULE_READ | REDISMODULE_WRITE, mutated_key_str);
+      RedisModule_ZsetAdd(new_key, index, data, &flags);
+      RAY_CHECK(flags == REDISMODULE_ZADD_ADDED);
+    }
     return REDISMODULE_OK;
   } else {
     // The requested index did not match the current length of the log. Return

test/runtest.py

@@ -2402,6 +2402,29 @@ def test_ray_setproctitle(shutdown_only):
     ray.get(unique_1.remote())
 
 
+def test_duplicate_error_messages(shutdown_only):
+    ray.init(num_cpus=0)
+
+    driver_id = ray.ray_constants.NIL_JOB_ID.id()
+    error_data = ray.gcs_utils.construct_error_message(driver_id, "test",
+                                                       "message", 0)
+
+    # Push the same message to the GCS twice (they are the same because we
+    # do not include a timestamp).
+
+    r = ray.worker.global_worker.redis_client
+
+    r.execute_command("RAY.TABLE_APPEND", ray.gcs_utils.TablePrefix.ERROR_INFO,
+                      ray.gcs_utils.TablePubsub.ERROR_INFO, driver_id,
+                      error_data)
+
+    # Before https://github.com/ray-project/ray/pull/3316 this would
+    # give an error
+    r.execute_command("RAY.TABLE_APPEND", ray.gcs_utils.TablePrefix.ERROR_INFO,
+                      ray.gcs_utils.TablePubsub.ERROR_INFO, driver_id,
+                      error_data)
+
+
 @pytest.mark.skipif(
     os.getenv("TRAVIS") is None,
     reason="This test should only be run on Travis.")