Skip to content

Fix issues in valkeyClusterAsyncCallback #306

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bjosv merged 3 commits into from
Jun 5, 2026
Merged

Fix issues in valkeyClusterAsyncCallback #306

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0e554e4
Handle failures when initiating async connect during retry
bjosv Apr 20, 2026
025fccc
Remove unnecessary NULL check
bjosv Apr 20, 2026
9f34bf2
Simplify async callback and fix missed user notifications
bjosv Apr 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
82 changes: 34 additions & 48 deletions src/cluster.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2937,6 +2937,11 @@ static void throttledUpdateSlotMapAsync(valkeyClusterAsyncContext *acc,
}
}

/* Callback for async cluster commands. Handles redirects (MOVED/ASK),
* retries (TRYAGAIN/CLUSTERDOWN), and delivers the final reply to the user.
* On success or unrecoverable failure, the user callback is always invoked
* with the reply on success, or with NULL when an error occurred (check
* acc->err for details). */
static void valkeyClusterAsyncCallback(valkeyAsyncContext *ac, void *r,
void *privdata) {
int ret;
Expand All @@ -2948,27 +2953,26 @@ static void valkeyClusterAsyncCallback(valkeyAsyncContext *ac, void *r,
valkeyClusterNode *node;
struct cmd *command;

if (cad == NULL) {
goto error;
}
assert(cad != NULL);
if (cad == NULL)
return;
Comment on lines +2956 to +2958

@zuiderkwast zuiderkwast Apr 24, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

assert + if...? Pick one.

@bjosv bjosv Apr 24, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Its due to NDEBUG, get asserts in debug builds and tests, but don't crash in release builds.
Maybe its not worth it? I believe we have some legacy code like it

@zuiderkwast zuiderkwast May 8, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, make sense then! We can keep it.

@zuiderkwast zuiderkwast May 8, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's a little odd though. You get different behaviour with/without NDEBUG.


acc = cad->acc;
assert(acc != NULL);
if (acc == NULL) {
Comment thread
zuiderkwast marked this conversation as resolved.
goto error;
cluster_async_data_free(cad);
return;
}

cc = &acc->cc;
if (cc == NULL) {
goto error;
}

command = cad->command;
if (command == NULL) {
goto error;
}
assert(command != NULL);
if (command == NULL)
goto done;
Comment thread
zuiderkwast marked this conversation as resolved.

if (reply == NULL) {
/* Copy reply specific error from libvalkey */
/* Copy error from the underlying context. */
valkeyClusterAsyncSetError(acc, ac->err, ac->errstr);

node = (valkeyClusterNode *)ac->data;
Expand All @@ -2984,6 +2988,7 @@ static void valkeyClusterAsyncCallback(valkeyAsyncContext *ac, void *r,
if (cad->retry_count == NO_RETRY || cc->flags & VALKEY_FLAG_DISCONNECTING)
goto done;

/* Handle cluster redirect and retry errors. */
replyErrorType error_type = getReplyErrorType(reply);
if (error_type > CLUSTER_NO_ERROR && error_type < CLUSTER_ERR_OTHER) {
cad->retry_count++;
Expand All @@ -3009,9 +3014,12 @@ static void valkeyClusterAsyncCallback(valkeyAsyncContext *ac, void *r,
if (slot >= 0) {
cc->table[slot] = node;
}
ac_retry = valkeyClusterGetValkeyAsyncContext(acc, node);

ac_retry = valkeyClusterGetValkeyAsyncContext(acc, node);
if (ac_retry == NULL)
goto done;
break;

case CLUSTER_ERR_ASK:
node = getNodeFromRedirectReply(cc, &ac->c, reply, NULL);
if (node == NULL) {
Expand All @@ -3020,57 +3028,35 @@ static void valkeyClusterAsyncCallback(valkeyAsyncContext *ac, void *r,
}

ac_retry = valkeyClusterGetValkeyAsyncContext(acc, node);
if (ac_retry == NULL) {
/* Specific error already set */
if (ac_retry == NULL)
goto done;
}

ret =
valkeyAsyncCommand(ac_retry, NULL, NULL, VALKEY_COMMAND_ASKING);
if (ret != VALKEY_OK) {
goto error;
}

ret = valkeyAsyncCommand(ac_retry, NULL, NULL, VALKEY_COMMAND_ASKING);
if (ret != VALKEY_OK)
goto done;
break;

case CLUSTER_ERR_TRYAGAIN:
case CLUSTER_ERR_CLUSTERDOWN:
ac_retry = ac;

break;
default:

default:
goto done;
}

goto retry;
/* Retry the command on the selected connection. */
ret = valkeyAsyncFormattedCommand(ac_retry, valkeyClusterAsyncCallback,
cad, command->cmd, command->clen);
if (ret == VALKEY_OK)
return; /* Ownership of cad transferred to retry callback. */
/* Retry failed, fall through to notify the user. */
}

done:

if (acc->err) {
cad->callback(acc, NULL, cad->privdata);
} else {
cad->callback(acc, r, cad->privdata);
}

/* Deliver result to the user callback. */
cad->callback(acc, acc->err ? NULL : r, cad->privdata);
valkeyClusterAsyncClearError(acc);

cluster_async_data_free(cad);

return;

retry:

ret = valkeyAsyncFormattedCommand(ac_retry, valkeyClusterAsyncCallback, cad,
command->cmd, command->clen);
if (ret != VALKEY_OK) {
goto error;
}

return;

error:

cluster_async_data_free(cad);
}

Expand Down
Loading