Commit 055958aa authored by antirez's avatar antirez
Browse files

Cluster: make getNodeByQuery() responsible of -CLUSTERDOWN errors.

This fixes a bug introduced by d827dbfd, and makes the code consistent
with the logic of always allowing, while the cluster is down, commands
that don't target any key.

As a side effect the code is also simpler now.
parent 1d8078b7
......@@ -4933,7 +4933,10 @@ void readwriteCommand(redisClient *c) {
* REDIS_CLUSTER_REDIR_DOWN_UNBOUND if the request addresses a slot which is
* not bound to any node. In this case the cluster global state should be
* already "down" but it is fragile to rely on the update of the global state,
* so we also handle it here. */
* so we also handle it here.
*
* REDIS_CLUSTER_REDIR_DOWN_STATE if the cluster is down but the user attempts
* to execute a command that addresses one or more keys. */
clusterNode *getNodeByQuery(redisClient *c, struct redisCommand *cmd, robj **argv, int argc, int *hashslot, int *error_code) {
clusterNode *n = NULL;
robj *firstkey = NULL;
......@@ -5040,9 +5043,15 @@ clusterNode *getNodeByQuery(redisClient *c, struct redisCommand *cmd, robj **arg
}
/* No key at all in command? then we can serve the request
* without redirections or errors. */
* without redirections or errors in all the cases. */
if (n == NULL) return myself;
/* Cluster is globally down but we got keys? We can't serve the request. */
if (server.cluster->state != REDIS_CLUSTER_OK) {
if (error_code) *error_code = REDIS_CLUSTER_REDIR_DOWN_STATE;
return NULL;
}
/* Return the hashslot by reference. */
if (hashslot) *hashslot = slot;
......
......@@ -2191,23 +2191,17 @@ int processCommand(redisClient *c) {
c->cmd->proc != execCommand))
{
int hashslot;
if (server.cluster->state != REDIS_CLUSTER_OK) {
flagTransaction(c);
clusterRedirectClient(c,NULL,0,REDIS_CLUSTER_REDIR_DOWN_STATE);
return REDIS_OK;
} else {
int error_code;
clusterNode *n = getNodeByQuery(c,c->cmd,c->argv,c->argc,&hashslot,&error_code);
if (n == NULL || n != server.cluster->myself) {
if (c->cmd->proc == execCommand) {
discardTransaction(c);
} else {
flagTransaction(c);
}
clusterRedirectClient(c,n,hashslot,error_code);
return REDIS_OK;
int error_code;
clusterNode *n = getNodeByQuery(c,c->cmd,c->argv,c->argc,
&hashslot,&error_code);
if (n == NULL || n != server.cluster->myself) {
if (c->cmd->proc == execCommand) {
discardTransaction(c);
} else {
flagTransaction(c);
}
clusterRedirectClient(c,n,hashslot,error_code);
return REDIS_OK;
}
}
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment