diff --git a/src/cluster.c b/src/cluster.c index 00f3c2d889..2f5952e3e1 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -1354,15 +1354,17 @@ void addNodeReplyForClusterSlot(client *c, clusterNode *node, int start_slot, in void clearCachedClusterSlotsResponse(void) { for (connTypeForCaching conn_type = CACHE_CONN_TCP; conn_type < CACHE_CONN_TYPE_MAX; conn_type++) { - if (server.cached_cluster_slot_info[conn_type]) { - sdsfree(server.cached_cluster_slot_info[conn_type]); - server.cached_cluster_slot_info[conn_type] = NULL; + for (int resp = 0; resp <= 3; resp++) { + if (server.cached_cluster_slot_info[conn_type][resp]) { + sdsfree(server.cached_cluster_slot_info[conn_type][resp]); + server.cached_cluster_slot_info[conn_type][resp] = NULL; + } } } } -sds generateClusterSlotResponse(void) { - client *recording_client = createCachedResponseClient(); +sds generateClusterSlotResponse(int resp) { + client *recording_client = createCachedResponseClient(resp); clusterNode *n = NULL; int num_primaries = 0, start = -1; void *slot_replylen = addReplyDeferredLen(recording_client); @@ -1392,8 +1394,8 @@ sds generateClusterSlotResponse(void) { return cluster_slot_response; } -int verifyCachedClusterSlotsResponse(sds cached_response) { - sds generated_response = generateClusterSlotResponse(); +int verifyCachedClusterSlotsResponse(sds cached_response, int resp) { + sds generated_response = generateClusterSlotResponse(resp); int is_equal = !sdscmp(generated_response, cached_response); /* Here, we use LL_WARNING so this gets printed when debug assertions are enabled and the system is about to crash. */ if (!is_equal) @@ -1413,16 +1415,16 @@ void clusterCommandSlots(client *c) { * 3) node ID * ... continued until done */ - connTypeForCaching conn_type = connIsTLS(c->conn); + connTypeForCaching conn_type = shouldReturnTlsInfo(); if (detectAndUpdateCachedNodeHealth()) clearCachedClusterSlotsResponse(); - sds cached_reply = server.cached_cluster_slot_info[conn_type]; + sds cached_reply = server.cached_cluster_slot_info[conn_type][c->resp]; if (!cached_reply) { - cached_reply = generateClusterSlotResponse(); - server.cached_cluster_slot_info[conn_type] = cached_reply; + cached_reply = generateClusterSlotResponse(c->resp); + server.cached_cluster_slot_info[conn_type][c->resp] = cached_reply; } else { - debugServerAssertWithInfo(c, NULL, verifyCachedClusterSlotsResponse(cached_reply) == 1); + debugServerAssertWithInfo(c, NULL, verifyCachedClusterSlotsResponse(cached_reply, c->resp) == 1); } addReplyProto(c, cached_reply, sdslen(cached_reply)); diff --git a/src/cluster.h b/src/cluster.h index f163e7f688..c056782cf2 100644 --- a/src/cluster.h +++ b/src/cluster.h @@ -104,7 +104,7 @@ const char *clusterNodePreferredEndpoint(clusterNode *n); long long clusterNodeReplOffset(clusterNode *node); clusterNode *clusterLookupNode(const char *name, int length); int detectAndUpdateCachedNodeHealth(void); -client *createCachedResponseClient(void); +client *createCachedResponseClient(int resp); void deleteCachedResponseClient(client *recording_client); void clearCachedClusterSlotsResponse(void); diff --git a/src/cluster_legacy.c b/src/cluster_legacy.c index 21aa620dd9..284530d15f 100644 --- a/src/cluster_legacy.c +++ b/src/cluster_legacy.c @@ -1033,7 +1033,9 @@ void clusterInit(void) { server.cluster->mf_end = 0; server.cluster->mf_replica = NULL; for (connTypeForCaching conn_type = CACHE_CONN_TCP; conn_type < CACHE_CONN_TYPE_MAX; conn_type++) { - server.cached_cluster_slot_info[conn_type] = NULL; + for (int resp = 0; resp <= 3; resp++) { + server.cached_cluster_slot_info[conn_type][resp] = NULL; + } } resetManualFailover(); clusterUpdateMyselfFlags(); diff --git a/src/networking.c b/src/networking.c index ecdeeb6588..7c90c69093 100644 --- a/src/networking.c +++ b/src/networking.c @@ -337,8 +337,9 @@ sds aggregateClientOutputBuffer(client *c) { * to initiate caching of any command response. * * It needs be paired with `deleteCachedResponseClient` function to stop caching. */ -client *createCachedResponseClient(void) { +client *createCachedResponseClient(int resp) { struct client *recording_client = createClient(NULL); + recording_client->resp = resp; /* Allocating the `conn` allows to prepare the caching client before adding * data to the clients output buffer by `prepareClientToWrite`. */ recording_client->conn = zcalloc(sizeof(connection)); diff --git a/src/server.h b/src/server.h index c4ce6f655e..da0efa15d6 100644 --- a/src/server.h +++ b/src/server.h @@ -2071,7 +2071,7 @@ struct valkeyServer { * dropping packets of a specific type */ /* Debug config that goes along with cluster_drop_packet_filter. When set, the link is closed on packet drop. */ uint32_t debug_cluster_close_link_on_packet_drop : 1; - sds cached_cluster_slot_info[CACHE_CONN_TYPE_MAX]; + sds cached_cluster_slot_info[CACHE_CONN_TYPE_MAX][4]; /* Align to RESP3 */ /* Scripting */ mstime_t busy_reply_threshold; /* Script / module timeout in milliseconds */ int pre_command_oom_state; /* OOM before command (script?) was started */ @@ -2734,7 +2734,7 @@ void initSharedQueryBuf(void); client *lookupClientByID(uint64_t id); int authRequired(client *c); void putClientInPendingWriteQueue(client *c); -client *createCachedResponseClient(void); +client *createCachedResponseClient(int resp); void deleteCachedResponseClient(client *recording_client); /* logreqres.c - logging of requests and responses */