Add client blocking mechanism for keys in use

[harrylin98]

Summary

This PR adds blockInuse, a server-initiated client blocking mechanism that prevents concurrent access to keys held exclusively by internal operations. When a client command targets a key that is currently in use, the client is blocked and its read handler is removed to prevent unbounded input buffering. Once the internal operation releases the key, the client is automatically resumed and its pending command is re-executed.

This is a building block for upcoming bgIteration that require exclusive key access without stalling the event loop.

Design Decisions

Isolated module

Current blocking in blocked.c is built around client-initiated command blocking, which is a coherent system where unblocking is driven by keyspace events. blockInuse is server-initiated blocking where unblocking is driven by explicit internal operation completion. Isolating blockInuse in its own module keeps the two unblock triggers separate, avoids embedding blockInuse state into client->bstate and serverDb structures that it has no logical
relationship with, and gives it a clearly defined API and invariants.

Removing READ handler

When a client is blocked_inuse blocked, its read handler is removed so the event loop stops monitoring read events for that connection, preventing new commands from being buffered into c->querybuf while the client is waiting. Clients would otherwise continue buffering incoming commands during the entire blocking period, leading to unbounded memory growth across all blocked clients. The read handler is restored in processUnblockedClients() when the client is unblocked.

The tradeoff is that removing the read handler also makes the event loop blind to TCP disconnects on that client connection. To avoid leaking zombie file descriptors, clientsCronTcpIsClosing() is added to detect and free connections that were closed by the remote side while the read handler was removed.

Observability

Clients blocked by blockInuse are visible as flag X in CLIENT LIST output.

Testing

Unit tests (src/unit/test_blockedInuse.cpp) cover each public API:

• blockInuse_blockClientOnKeys()
• blockInuse_unblockClientsOnKey()
• blockInuse_unblockClientsOnAllKeys()
• blockInuse_unlinkClient()

[codecov]

Codecov Report

❌ Patch coverage is 30.55556% with 125 lines in your changes missing coverage. Please review.
✅ Project coverage is 74.61%. Comparing base (543a6b8) to head (7c0bdb5).

Files with missing lines	Patch %	Lines
src/blocked_inuse.c	17.60%	117 Missing ⚠️
src/timeout.c	0.00%	5 Missing ⚠️
src/blocked.c	50.00%	3 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##           forkless    #3341      +/-   ##
============================================
+ Coverage     74.56%   74.61%   +0.05%     
============================================
  Files           130      131       +1     
  Lines         72730    72902     +172     
============================================
+ Hits          54228    54394     +166     
- Misses        18502    18508       +6     

Files with missing lines	Coverage Δ
src/connection.h	88.11% <ø> (ø)
src/networking.c	91.46% <100.00%> (+0.04%)	⬆️
src/rdma.c	100.00% <ø> (ø)
src/server.c	89.47% <100.00%> (-0.02%)	⬇️
src/server.h	100.00% <ø> (ø)
src/socket.c	94.94% <100.00%> (+0.14%)	⬆️
src/tls.c	17.64% <ø> (ø)
src/unix.c	78.31% <ø> (ø)
src/blocked.c	90.36% <50.00%> (-0.78%)	⬇️
src/timeout.c	88.46% <0.00%> (-1.15%)	⬇️
... and 1 more

... and 21 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[Copilot]

Pull request overview

This PR introduces a new internal “blocked-in-use” client blocking mechanism to prevent clients from accessing keys that are currently being operated on by internal/background work, and integrates it into the main server loop and unit tests.

Changes:

• Add blocked_inuse.{c,h} implementing client↔keys and key↔clients mappings, plus block/unblock/unlink APIs.
• Integrate the mechanism into client lifecycle paths (timeouts, unlinking, unblocked-client processing, shutdown init/release, and client info flags).
• Add GoogleTest coverage and wrapper hooks for mocking relevant server functions.

Reviewed changes

Copilot reviewed 11 out of 11 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
src/blocked_inuse.c	Implements the core block/unblock/unlink logic and internal hashtable mappings.
src/blocked_inuse.h	Declares the blockInuse public API and documents intended workflow.
src/server.h	Adds a new client flag bit (blockInuse_blocked).
src/server.c	Initializes/releases blockInuse; adds TCP close detection for blocked clients; asserts blocked-state invariants.
src/networking.c	Unlinks blockInuse-blocked clients during unlinkClient; prints new client flag X; adjusts pending-command processing behavior.
src/blocked.c	Updates processUnblockedClients() to support restoring read handlers for blockInuse-unblocked clients and adds pause/close_asap handling.
src/timeout.c	Exempts blockInuse-blocked clients from max-idle timeout enforcement.
src/unit/test_blockedInuse.cpp	Adds unit tests covering single/multi-key and multi-client scenarios, plus death tests.
src/unit/wrappers.h	Adds wrapper declarations needed to mock lookupKeyRead, processPendingCommandAndInputBuffer, and beforeNextClient in tests.
src/Makefile	Adds blocked_inuse.o to the server object list.
cmake/Modules/SourceFiles.cmake	Adds src/blocked_inuse.c to the CMake server sources list.

---

You can also share your feedback on Copilot code review. Take the survey.

[madolson]

This comment is wrong, since we support multiple back ends. This comment is also not very helpful, as it's just describing what is happening. There are a lot of "what" comments in this CR, which could probably get removed.

[murphyjacob4]

Is it resolved? I still see the comment

[ranshid]

.

[ranshid]

Overall I think this PR is missing capturing much of the motivation.
To me, it seems we wanted something like a combination of postpone+keys blocking type which also remove the read handler in order to avoid reading data and consuming more memory.
The entire blocked_inuse is what needs to be motivated IMO for example:

• why do we need a complete separated module and not use/extend the blocked.c to handle this block type?
• why do we need a new way to map keys to blocked clients and not use the blockForKeys?

[ranshid]

Can you better explain the motivation around this? why are we only targeting PAUSE_ACTIONS_CLIENT_ALL_SET and not PAUSE_ACTION_CLIENT_WRITE for example?

[harrylin98]

The motivation is to early-terminate processUnblockedClients when a full pause is active. Under PAUSE_ACTIONS_CLIENT_ALL_SET, no client commands should be "processed" at all, so there's no point iterating further. The early return avoids that churn and leaves all remaining clients in unblocked_clients to be picked up in a future beforeSleep() once the pause lifts.

Under PAUSE_ACTIONS_CLIENT_WRITE_SET, reads are still allowed to execute, so we let the loop continue — processCommand will postpone-block write commands and let read commands run normally.

Rephrase the comments in the code.

[ranshid]

I see. so it is strange this is placed inside the while loop. lets just take it outside.

[harrylin98]

Moving it outside is not quiet right...
Internal pause state could be updated by any of the unblocked client's processCommand, so it needs to be checked before each processCommand case.

I will clarify the comments for this.

[murphyjacob4]

Under PAUSE_ACTIONS_CLIENT_ALL_SET, no client commands should be "processed" at all, so there's no point iterating further.

I am not fully sure the behavior is identical to the previous behavior.

So consider this sequence of events:

1. Client blocked (e.g. by module)
2. Server paused with actions == ALL
3. Client unblocked

Before, we would go into processCommand, and in there we have various different checks execute:

valkey/src/server.c

Lines 4223 to 4550 in 9586093

	if (!scriptIsTimedout()) {
	/* Both EXEC and scripts call call() directly so there should be
	* no way in_exec or scriptIsRunning() is 1.
	* That is unless lua_timedout, in which case client may run
	* some commands. */
	serverAssert(!server.in_exec);
	serverAssert(!scriptIsRunning());
	}
	/* in case we are starting to ProcessCommand and we already have a command we assume
	* this is a reprocessing of this command, so we do not want to perform some of the actions again. */
	int client_reprocessing_command = c->cmd ? 1 : 0;
	/* only run command filter if not reprocessing command */
	if (!client_reprocessing_command) {
	moduleCallCommandFilters(c);
	reqresAppendRequest(c);
	}
	/* If we're inside a module blocked context yielding that wants to avoid
	* processing clients, postpone the command. */
	if (server.busy_module_yield_flags != BUSY_MODULE_YIELD_NONE &&
	!(server.busy_module_yield_flags & BUSY_MODULE_YIELD_CLIENTS)) {
	blockPostponeClient(c);
	return C_OK;
	}
	/* Now lookup the command and check ASAP about trivial error conditions
	* such as wrong arity, bad command name and so forth.
	* In case we are reprocessing a command after it was blocked,
	* we do not have to repeat the same checks */
	if (!client_reprocessing_command) {
	struct serverCommand *cmd = c->parsed_cmd;
	if (!cmd) {
	/* Handle possible security attacks. */
	if (!strcasecmp(objectGetVal(c->argv[0]), "host:") || !strcasecmp(objectGetVal(c->argv[0]), "post")) {
	securityWarningCommand(c);
	return C_ERR;
	}
	/* Check that the command lookup has been done before calling this
	* function, by calling prepareCommand(). */
	serverAssert(c->read_flags & READ_FLAGS_COMMAND_NOT_FOUND);
	}
	c->cmd = c->lastcmd = c->realcmd = cmd;
	if (authRequired(c)) {
	/* AUTH and HELLO and no auth commands are valid even in
	* non-authenticated state. */
	if (!c->cmd || !(c->cmd->flags & CMD_NO_AUTH)) {
	rejectCommand(c, shared.noautherr);
	return C_OK;
	}
	}
	c->flag.buffered_reply = 0;
	sds err;
	if (!commandCheckExistence(c, &err)) {
	rejectCommandSds(c, err);
	return C_OK;
	}
	if (c->read_flags & READ_FLAGS_BAD_ARITY) {
	/* Already detected this, but do it again just to get the error message. */
	serverAssert(!commandCheckArity(c->cmd, c->argc, &err));
	rejectCommandSds(c, err);
	return C_OK;
	}
	/* Check if the command is marked as protected and the relevant configuration allows it */
	if (c->cmd->flags & CMD_PROTECTED) {
	if ((c->cmd->proc == debugCommand && !allowProtectedAction(server.enable_debug_cmd, c)) ||
	(c->cmd->proc == moduleCommand && !allowProtectedAction(server.enable_module_cmd, c))) {
	rejectCommandFormat(c,
	"%s command not allowed. If the %s option is set to \"local\", "
	"you can run it from a local connection, otherwise you need to set this option "
	"in the configuration file, and then restart the server.",
	c->cmd->proc == debugCommand ? "DEBUG" : "MODULE",
	c->cmd->proc == debugCommand ? "enable-debug-command" : "enable-module-command");
	return C_OK;
	}
	}
	}
	uint64_t cmd_flags = getCommandFlags(c);
	int is_exec = (c->mstate && c->cmd->proc == execCommand);
	int ms_flags = is_exec ? c->mstate->cmd_flags : 0;
	int ms_inv_flags = is_exec ? c->mstate->cmd_inv_flags : 0;
	int combined_flags = cmd_flags | ms_flags;
	int combined_inv_flags = (~cmd_flags | ms_inv_flags);
	int is_read_command = (combined_flags & CMD_READONLY);
	int is_write_command = (combined_flags & CMD_WRITE);
	int is_denyoom_command = (combined_flags & CMD_DENYOOM);
	int is_denystale_command = (combined_inv_flags & CMD_STALE);
	int is_denyloading_command = (combined_inv_flags & CMD_LOADING);
	int is_may_replicate_command = (combined_flags & (CMD_WRITE | CMD_MAY_REPLICATE));
	int is_deny_async_loading_command = (combined_flags & CMD_NO_ASYNC_LOADING);
	const int obey_client = mustObeyClient(c);
	if (c->flag.multi && c->cmd->flags & CMD_NO_MULTI) {
	rejectCommandFormat(c, "Command '%s' not allowed inside a transaction", c->cmd->fullname);
	return C_OK;
	}
	/* Check if the user can run this command according to the current
	* ACLs. */
	int acl_errpos;
	int acl_retval = ACLCheckAllPerm(c, &acl_errpos);
	if (acl_retval != ACL_OK) {
	addACLLogEntry(c, acl_retval, (c->flag.multi) ? ACL_LOG_CTX_MULTI : ACL_LOG_CTX_TOPLEVEL, acl_errpos, NULL,
	NULL);
	sds msg = getAclErrorMessage(acl_retval, c->user, c->cmd, objectGetVal(c->argv[acl_errpos]), 0);
	rejectCommandFormat(c, "-NOPERM %s", msg);
	sdsfree(msg);
	return C_OK;
	}
	/* If cluster is enabled perform the cluster redirection here.
	* However we don't perform the redirection if:
	* 1) The sender of this command is our primary.
	* 2) The command has no key arguments. */
	if (server.cluster_enabled && !obey_client &&
	!(!(c->cmd->flags & CMD_MOVABLE_KEYS) && c->cmd->key_specs_num == 0 && c->cmd->proc != execCommand)) {
	int error_code;
	clusterNode *n = getNodeByQuery(c, &error_code);
	if (n == NULL || !clusterNodeIsMyself(n)) {
	if (c->cmd->proc == execCommand) {
	discardTransaction(c);
	} else {
	flagTransaction(c);
	}
	clusterRedirectClient(c, n, c->slot, error_code);
	c->duration = 0;
	c->cmd->rejected_calls++;
	return C_OK;
	}
	}
	if (clientSupportStandAloneRedirect(c) && !obey_client &&
	(is_write_command || (is_read_command && !c->flag.readonly))) {
	if (server.failover_state == FAILOVER_IN_PROGRESS) {
	/* During the FAILOVER process, when conditions are met (such as
	* when the force time is reached or the primary and replica offsets
	* are consistent), the primary actively becomes the replica and
	* transitions to the FAILOVER_IN_PROGRESS state.
	*
	* After the primary becomes the replica, and after handshaking
	* and other operations, it will eventually send the PSYNC FAILOVER
	* command to the replica, then the replica will become the primary.
	* This means that the upgrade of the replica to the primary is an
	* asynchronous operation, which implies that during the
	* FAILOVER_IN_PROGRESS state, there may be a period of time where
	* both nodes are replicas.
	*
	* In this scenario, if a -REDIRECT is returned, the request will be
	* redirected to the replica and then redirected back, causing back
	* and forth redirection. To avoid this situation, during the
	* FAILOVER_IN_PROGRESS state, we temporarily suspend the clients
	* that need to be redirected until the replica truly becomes the primary,
	* and then resume the execution. */
	blockPostponeClient(c);
	} else {
	if (c->cmd->proc == execCommand) {
	discardTransaction(c);
	} else {
	flagTransaction(c);
	}
	c->duration = 0;
	c->cmd->rejected_calls++;
	addReplyErrorSds(c, sdscatprintf(sdsempty(), "-REDIRECT %s:%d", server.primary_host, server.primary_port));
	}
	return C_OK;
	}
	/* Disconnect some clients if total clients memory is too high. We do this
	* before key eviction, after the last command was executed and consumed
	* some client output buffer memory. */
	evictClients();
	if (server.current_client == NULL) {
	/* If we evicted ourself then abort processing the command */
	return C_ERR;
	}
	/* Handle the maxmemory directive.
	*
	* Note that we do not want to reclaim memory if we are here re-entering
	* the event loop since there is a busy Lua script running in timeout
	* condition, to avoid mixing the propagation of scripts with the
	* propagation of DELs due to eviction. */
	if (server.maxmemory && !isInsideYieldingLongCommand()) {
	int out_of_memory = (performEvictions() == EVICT_FAIL);
	/* performEvictions may evict keys, so we need flush pending tracking
	* invalidation keys. If we don't do this, we may get an invalidation
	* message after we perform operation on the key, where in fact this
	* message belongs to the old value of the key before it gets evicted.*/
	trackingHandlePendingKeyInvalidations();
	/* performEvictions may flush replica output buffers. This may result
	* in a replica, that may be the active client, to be freed. */
	if (server.current_client == NULL) return C_ERR;
	if (out_of_memory && is_denyoom_command) {
	if (c->slot_migration_job != NULL) {
	clusterHandleSlotMigrationClientOOM(c->slot_migration_job);
	return C_ERR;
	}
	rejectCommand(c, shared.oomerr);
	return C_OK;
	}
	/* Save out_of_memory result at command start, otherwise if we check OOM
	* in the first write within script, memory used by lua stack and
	* arguments might interfere. We need to save it for EXEC and module
	* calls too, since these can call EVAL, but avoid saving it during an
	* interrupted / yielding busy script / module. */
	server.pre_command_oom_state = out_of_memory;
	}
	/* Make sure to use a reasonable amount of memory for client side
	* caching metadata. */
	if (server.tracking_clients) trackingLimitUsedSlots();
	/* Don't accept write commands if there are problems persisting on disk
	* unless coming from our primary, in which case check the replica ignore
	* disk write error config to either log or crash. */
	int deny_write_type = writeCommandsDeniedByDiskError();
	if (deny_write_type != DISK_ERROR_TYPE_NONE && (is_write_command || c->cmd->proc == pingCommand)) {
	if (obey_client) {
	if (!server.repl_ignore_disk_write_error && c->cmd->proc != pingCommand) {
	serverPanic("Replica was unable to write command to disk.");
	} else {
	static mstime_t last_log_time_ms = 0;
	const mstime_t log_interval_ms = 10000;
	if (server.mstime > last_log_time_ms + log_interval_ms) {
	last_log_time_ms = server.mstime;
	serverLog(LL_WARNING, "Replica is applying a command even though "
	"it is unable to write to disk.");
	}
	}
	} else {
	sds err = writeCommandsGetDiskErrorMessage(deny_write_type);
	/* remove the newline since rejectCommandSds adds it. */
	sdssubstr(err, 0, sdslen(err) - 2);
	rejectCommandSds(c, err);
	return C_OK;
	}
	}
	/* Don't accept write commands if there are not enough good replicas and
	* user configured the min-replicas-to-write option. */
	if (is_write_command && !checkGoodReplicasStatus()) {
	rejectCommand(c, shared.noreplicaserr);
	return C_OK;
	}
	/* Don't accept write commands if this is a read only replica. But
	* accept write commands if this is our primary. */
	if (server.primary_host && server.repl_replica_ro && !obey_client && is_write_command) {
	rejectCommand(c, shared.roreplicaerr);
	return C_OK;
	}
	/* Only allow a subset of commands in the context of Pub/Sub if the
	* connection is in RESP2 mode. With RESP3 there are no limits. */
	if ((c->flag.pubsub && c->resp == 2) && c->cmd->proc != pingCommand && c->cmd->proc != subscribeCommand &&
	c->cmd->proc != ssubscribeCommand && c->cmd->proc != unsubscribeCommand &&
	c->cmd->proc != sunsubscribeCommand && c->cmd->proc != psubscribeCommand &&
	c->cmd->proc != punsubscribeCommand && c->cmd->proc != quitCommand && c->cmd->proc != resetCommand) {
	rejectCommandFormat(c,
	"Can't execute '%s': only (P|S)SUBSCRIBE / "
	"(P|S)UNSUBSCRIBE / PING / QUIT / RESET are allowed in this context",
	c->cmd->fullname);
	return C_OK;
	}
	/* Only allow commands with flag "t", such as INFO, REPLICAOF and so on,
	* when replica-serve-stale-data is no and we are a replica with a broken
	* link with primary. */
	if (server.primary_host && server.repl_state != REPL_STATE_CONNECTED && server.repl_serve_stale_data == 0 &&
	is_denystale_command) {
	rejectCommand(c, shared.primarydownerr);
	return C_OK;
	}
	/* Loading DB? Return an error if the command has not the
	* CMD_LOADING flag. */
	if (server.loading && !server.async_loading && is_denyloading_command) {
	rejectCommand(c, shared.loadingerr);
	return C_OK;
	}
	/* During async-loading, block certain commands. */
	if (server.async_loading && is_deny_async_loading_command) {
	rejectCommand(c, shared.loadingerr);
	return C_OK;
	}
	/* when a busy job is being done (script / module)
	* Only allow a limited number of commands.
	* Note that we need to allow the transactions commands, otherwise clients
	* sending a transaction with pipelining without error checking, may have
	* the MULTI plus a few initial commands refused, then the timeout
	* condition resolves, and the bottom-half of the transaction gets
	* executed, see Github PR #7022. */
	if (isInsideYieldingLongCommand() && !(c->cmd->flags & CMD_ALLOW_BUSY)) {
	if (server.busy_module_yield_flags && server.busy_module_yield_reply) {
	rejectCommandFormat(c, "-BUSY %s", server.busy_module_yield_reply);
	} else if (server.busy_module_yield_flags) {
	rejectCommand(c, shared.slowmoduleerr);
	} else if (scriptIsEval()) {
	rejectCommand(c, shared.slowevalerr);
	} else {
	rejectCommand(c, shared.slowscripterr);
	}
	return C_OK;
	}
	/* Prevent a replica from sending commands that access the keyspace.
	* The main objective here is to prevent abuse of client pause check
	* from which replicas are exempt. */
	if (c->flag.replica && (is_may_replicate_command || is_write_command || is_read_command)) {
	rejectCommandFormat(c, "Replica can't interact with the keyspace");
	return C_OK;
	}

After, we would not go into processCommand, so all the linked lines will not execute.

Maybe you can help me understand why we are adding this change? If the reason is just optimization, I think we should skip this change. If the change is needed, we should double check that none of the functionality changes would be problematic

[harrylin98]

Yes it is an optimization for early return, and yes this will be processed later in processCommand.

I will remove it to keep the logic same.

[jjuleslasarte]

not specific to this code, but we should think through whether there are issues between this and #3381 -- does the code here need to handle blocked on keys?

[harrylin98]

Overall I think this PR is missing capturing much of the motivation. To me, it seems we wanted something like a combination of postpone+keys blocking type which also remove the read handler in order to avoid reading data and consuming more memory. The entire blocked_inuse is what needs to be motivated IMO for example:

Thanks for asking, I will also update in the PR description. Here are for the reply:

• why do we need a complete separated module and not use/extend the blocked.c to handle this block type?

Current blocking is built around client-initiated command blocking, which is a coherent system where unblocking is driven by keyspace events. blockInuse, however, is server-initiated blocking where unblocking is driven by explicit internal operation completion. Merging them would reduce cohesion — blocked.c would need to handle two unrelated unblock triggers — and increase coupling, since blockInuse state would be embedded into client->bstate and serverDb structures that it has no logical relationship with.

• why do we need a new way to map keys to blocked clients and not use the blockForKeys?

1. blockInuse is server-initiated blocking, and intentionally uses module-static hashtables rather than embedding mappings into each client's bstate and serverDb, keeping the state centralized and independent of client/db lifecycle.
2. blockForKeys feeds into signalKeyAsReady() for unblocking, signalKeyAsReady() is called during
   command execution, blockInuse instead unblocks clients after internal operation explicitly releases the key.

[murphyjacob4]

Current blocking is built around client-initiated command blocking, which is a coherent system where unblocking is driven by keyspace events. blockInuse, however, is server-initiated blocking where unblocking is driven by explicit internal operation completion

Is this true? Blocking is also used on server shutdown, on failover, on slot migration, on CLIENT PAUSE - there are many other cases where we use blocking for server-driven blocking causes and where unblocking is automatic (either by timeout, or by a server operation completion).

The only difference I can find is here we want to block by key, whereas the other server-driven blocking causes are blocking all operations at the server level. But we also have issues like #3406 where we would be doing a server-driven per-slot pause. Building a one-off parallel subsystem just for "in-use" keys creates technical debt and fragmentation. IMO we should continue to build out the existing blocking subsystem to handle these use cases rather than having competing subsystems.

[harrylin98]

Current blocking is built around client-initiated command blocking, which is a coherent system where unblocking is driven by keyspace events. blockInuse, however, is server-initiated blocking where unblocking is driven by explicit internal operation completion

Is this true? Blocking is also used on server shutdown, on failover, on slot migration, on CLIENT PAUSE - there are many other cases where we use blocking for server-driven blocking causes and where unblocking is automatic (either by timeout, or by a server operation completion).

The only difference I can find is here we want to block by key, whereas the other server-driven blocking causes are blocking all operations at the server level. But we also have issues like #3406 where we would be doing a server-driven per-slot pause. Building a one-off parallel subsystem just for "in-use" keys creates technical debt and fragmentation. IMO we should continue to build out the existing blocking subsystem to handle these use cases rather than having competing subsystems.

You're right that blocked.c handles server-driven blocking too, and as you noted, the key difference is per-key granularity. The existing per-key infrastructure blockForKeys() is tightly coupled to keyspace-mutation-driven unblocking — keys in db→blocking_keys feed signalKeyAsReady(), so any write to a key would incorrectly unblock a blockInuse client.

I would still stand by the parallel blockedInuse system. Extending blockForKeys() would require either a parallel db→blocking_keys_inuse dict, or filtering throughout signalKeyAsReady() to skip BLOCKED_INUSE clients — mixing unrelated unblock triggers in a hot path. The read handler removal is the other structural difference. Existing blocking keeps the read handler active. Adding conditional removal/restoration across unblockClient() and timeout handling, and the zombie-fd detection it requires would touch more shared code than the isolated module. Keeping blockInuse self-contained gives it high cohesion — its invariants are locally verifiable with no risk of interactions with the existing unblock paths, timeout handling, or bstate lifecycle.

[murphyjacob4]

Under PAUSE_ACTIONS_CLIENT_ALL_SET, no client commands should be "processed" at all, so there's no point iterating further.

I am not fully sure the behavior is identical to the previous behavior.

So consider this sequence of events:

1. Client blocked (e.g. by module)
2. Server paused with actions == ALL
3. Client unblocked

Before, we would go into processCommand, and in there we have various different checks execute:

valkey/src/server.c

Lines 4223 to 4550 in 9586093

	if (!scriptIsTimedout()) {
	/* Both EXEC and scripts call call() directly so there should be
	* no way in_exec or scriptIsRunning() is 1.
	* That is unless lua_timedout, in which case client may run
	* some commands. */
	serverAssert(!server.in_exec);
	serverAssert(!scriptIsRunning());
	}
	/* in case we are starting to ProcessCommand and we already have a command we assume
	* this is a reprocessing of this command, so we do not want to perform some of the actions again. */
	int client_reprocessing_command = c->cmd ? 1 : 0;
	/* only run command filter if not reprocessing command */
	if (!client_reprocessing_command) {
	moduleCallCommandFilters(c);
	reqresAppendRequest(c);
	}
	/* If we're inside a module blocked context yielding that wants to avoid
	* processing clients, postpone the command. */
	if (server.busy_module_yield_flags != BUSY_MODULE_YIELD_NONE &&
	!(server.busy_module_yield_flags & BUSY_MODULE_YIELD_CLIENTS)) {
	blockPostponeClient(c);
	return C_OK;
	}
	/* Now lookup the command and check ASAP about trivial error conditions
	* such as wrong arity, bad command name and so forth.
	* In case we are reprocessing a command after it was blocked,
	* we do not have to repeat the same checks */
	if (!client_reprocessing_command) {
	struct serverCommand *cmd = c->parsed_cmd;
	if (!cmd) {
	/* Handle possible security attacks. */
	if (!strcasecmp(objectGetVal(c->argv[0]), "host:") || !strcasecmp(objectGetVal(c->argv[0]), "post")) {
	securityWarningCommand(c);
	return C_ERR;
	}
	/* Check that the command lookup has been done before calling this
	* function, by calling prepareCommand(). */
	serverAssert(c->read_flags & READ_FLAGS_COMMAND_NOT_FOUND);
	}
	c->cmd = c->lastcmd = c->realcmd = cmd;
	if (authRequired(c)) {
	/* AUTH and HELLO and no auth commands are valid even in
	* non-authenticated state. */
	if (!c->cmd || !(c->cmd->flags & CMD_NO_AUTH)) {
	rejectCommand(c, shared.noautherr);
	return C_OK;
	}
	}
	c->flag.buffered_reply = 0;
	sds err;
	if (!commandCheckExistence(c, &err)) {
	rejectCommandSds(c, err);
	return C_OK;
	}
	if (c->read_flags & READ_FLAGS_BAD_ARITY) {
	/* Already detected this, but do it again just to get the error message. */
	serverAssert(!commandCheckArity(c->cmd, c->argc, &err));
	rejectCommandSds(c, err);
	return C_OK;
	}
	/* Check if the command is marked as protected and the relevant configuration allows it */
	if (c->cmd->flags & CMD_PROTECTED) {
	if ((c->cmd->proc == debugCommand && !allowProtectedAction(server.enable_debug_cmd, c)) ||
	(c->cmd->proc == moduleCommand && !allowProtectedAction(server.enable_module_cmd, c))) {
	rejectCommandFormat(c,
	"%s command not allowed. If the %s option is set to \"local\", "
	"you can run it from a local connection, otherwise you need to set this option "
	"in the configuration file, and then restart the server.",
	c->cmd->proc == debugCommand ? "DEBUG" : "MODULE",
	c->cmd->proc == debugCommand ? "enable-debug-command" : "enable-module-command");
	return C_OK;
	}
	}
	}
	uint64_t cmd_flags = getCommandFlags(c);
	int is_exec = (c->mstate && c->cmd->proc == execCommand);
	int ms_flags = is_exec ? c->mstate->cmd_flags : 0;
	int ms_inv_flags = is_exec ? c->mstate->cmd_inv_flags : 0;
	int combined_flags = cmd_flags | ms_flags;
	int combined_inv_flags = (~cmd_flags | ms_inv_flags);
	int is_read_command = (combined_flags & CMD_READONLY);
	int is_write_command = (combined_flags & CMD_WRITE);
	int is_denyoom_command = (combined_flags & CMD_DENYOOM);
	int is_denystale_command = (combined_inv_flags & CMD_STALE);
	int is_denyloading_command = (combined_inv_flags & CMD_LOADING);
	int is_may_replicate_command = (combined_flags & (CMD_WRITE | CMD_MAY_REPLICATE));
	int is_deny_async_loading_command = (combined_flags & CMD_NO_ASYNC_LOADING);
	const int obey_client = mustObeyClient(c);
	if (c->flag.multi && c->cmd->flags & CMD_NO_MULTI) {
	rejectCommandFormat(c, "Command '%s' not allowed inside a transaction", c->cmd->fullname);
	return C_OK;
	}
	/* Check if the user can run this command according to the current
	* ACLs. */
	int acl_errpos;
	int acl_retval = ACLCheckAllPerm(c, &acl_errpos);
	if (acl_retval != ACL_OK) {
	addACLLogEntry(c, acl_retval, (c->flag.multi) ? ACL_LOG_CTX_MULTI : ACL_LOG_CTX_TOPLEVEL, acl_errpos, NULL,
	NULL);
	sds msg = getAclErrorMessage(acl_retval, c->user, c->cmd, objectGetVal(c->argv[acl_errpos]), 0);
	rejectCommandFormat(c, "-NOPERM %s", msg);
	sdsfree(msg);
	return C_OK;
	}
	/* If cluster is enabled perform the cluster redirection here.
	* However we don't perform the redirection if:
	* 1) The sender of this command is our primary.
	* 2) The command has no key arguments. */
	if (server.cluster_enabled && !obey_client &&
	!(!(c->cmd->flags & CMD_MOVABLE_KEYS) && c->cmd->key_specs_num == 0 && c->cmd->proc != execCommand)) {
	int error_code;
	clusterNode *n = getNodeByQuery(c, &error_code);
	if (n == NULL || !clusterNodeIsMyself(n)) {
	if (c->cmd->proc == execCommand) {
	discardTransaction(c);
	} else {
	flagTransaction(c);
	}
	clusterRedirectClient(c, n, c->slot, error_code);
	c->duration = 0;
	c->cmd->rejected_calls++;
	return C_OK;
	}
	}
	if (clientSupportStandAloneRedirect(c) && !obey_client &&
	(is_write_command || (is_read_command && !c->flag.readonly))) {
	if (server.failover_state == FAILOVER_IN_PROGRESS) {
	/* During the FAILOVER process, when conditions are met (such as
	* when the force time is reached or the primary and replica offsets
	* are consistent), the primary actively becomes the replica and
	* transitions to the FAILOVER_IN_PROGRESS state.
	*
	* After the primary becomes the replica, and after handshaking
	* and other operations, it will eventually send the PSYNC FAILOVER
	* command to the replica, then the replica will become the primary.
	* This means that the upgrade of the replica to the primary is an
	* asynchronous operation, which implies that during the
	* FAILOVER_IN_PROGRESS state, there may be a period of time where
	* both nodes are replicas.
	*
	* In this scenario, if a -REDIRECT is returned, the request will be
	* redirected to the replica and then redirected back, causing back
	* and forth redirection. To avoid this situation, during the
	* FAILOVER_IN_PROGRESS state, we temporarily suspend the clients
	* that need to be redirected until the replica truly becomes the primary,
	* and then resume the execution. */
	blockPostponeClient(c);
	} else {
	if (c->cmd->proc == execCommand) {
	discardTransaction(c);
	} else {
	flagTransaction(c);
	}
	c->duration = 0;
	c->cmd->rejected_calls++;
	addReplyErrorSds(c, sdscatprintf(sdsempty(), "-REDIRECT %s:%d", server.primary_host, server.primary_port));
	}
	return C_OK;
	}
	/* Disconnect some clients if total clients memory is too high. We do this
	* before key eviction, after the last command was executed and consumed
	* some client output buffer memory. */
	evictClients();
	if (server.current_client == NULL) {
	/* If we evicted ourself then abort processing the command */
	return C_ERR;
	}
	/* Handle the maxmemory directive.
	*
	* Note that we do not want to reclaim memory if we are here re-entering
	* the event loop since there is a busy Lua script running in timeout
	* condition, to avoid mixing the propagation of scripts with the
	* propagation of DELs due to eviction. */
	if (server.maxmemory && !isInsideYieldingLongCommand()) {
	int out_of_memory = (performEvictions() == EVICT_FAIL);
	/* performEvictions may evict keys, so we need flush pending tracking
	* invalidation keys. If we don't do this, we may get an invalidation
	* message after we perform operation on the key, where in fact this
	* message belongs to the old value of the key before it gets evicted.*/
	trackingHandlePendingKeyInvalidations();
	/* performEvictions may flush replica output buffers. This may result
	* in a replica, that may be the active client, to be freed. */
	if (server.current_client == NULL) return C_ERR;
	if (out_of_memory && is_denyoom_command) {
	if (c->slot_migration_job != NULL) {
	clusterHandleSlotMigrationClientOOM(c->slot_migration_job);
	return C_ERR;
	}
	rejectCommand(c, shared.oomerr);
	return C_OK;
	}
	/* Save out_of_memory result at command start, otherwise if we check OOM
	* in the first write within script, memory used by lua stack and
	* arguments might interfere. We need to save it for EXEC and module
	* calls too, since these can call EVAL, but avoid saving it during an
	* interrupted / yielding busy script / module. */
	server.pre_command_oom_state = out_of_memory;
	}
	/* Make sure to use a reasonable amount of memory for client side
	* caching metadata. */
	if (server.tracking_clients) trackingLimitUsedSlots();
	/* Don't accept write commands if there are problems persisting on disk
	* unless coming from our primary, in which case check the replica ignore
	* disk write error config to either log or crash. */
	int deny_write_type = writeCommandsDeniedByDiskError();
	if (deny_write_type != DISK_ERROR_TYPE_NONE && (is_write_command || c->cmd->proc == pingCommand)) {
	if (obey_client) {
	if (!server.repl_ignore_disk_write_error && c->cmd->proc != pingCommand) {
	serverPanic("Replica was unable to write command to disk.");
	} else {
	static mstime_t last_log_time_ms = 0;
	const mstime_t log_interval_ms = 10000;
	if (server.mstime > last_log_time_ms + log_interval_ms) {
	last_log_time_ms = server.mstime;
	serverLog(LL_WARNING, "Replica is applying a command even though "
	"it is unable to write to disk.");
	}
	}
	} else {
	sds err = writeCommandsGetDiskErrorMessage(deny_write_type);
	/* remove the newline since rejectCommandSds adds it. */
	sdssubstr(err, 0, sdslen(err) - 2);
	rejectCommandSds(c, err);
	return C_OK;
	}
	}
	/* Don't accept write commands if there are not enough good replicas and
	* user configured the min-replicas-to-write option. */
	if (is_write_command && !checkGoodReplicasStatus()) {
	rejectCommand(c, shared.noreplicaserr);
	return C_OK;
	}
	/* Don't accept write commands if this is a read only replica. But
	* accept write commands if this is our primary. */
	if (server.primary_host && server.repl_replica_ro && !obey_client && is_write_command) {
	rejectCommand(c, shared.roreplicaerr);
	return C_OK;
	}
	/* Only allow a subset of commands in the context of Pub/Sub if the
	* connection is in RESP2 mode. With RESP3 there are no limits. */
	if ((c->flag.pubsub && c->resp == 2) && c->cmd->proc != pingCommand && c->cmd->proc != subscribeCommand &&
	c->cmd->proc != ssubscribeCommand && c->cmd->proc != unsubscribeCommand &&
	c->cmd->proc != sunsubscribeCommand && c->cmd->proc != psubscribeCommand &&
	c->cmd->proc != punsubscribeCommand && c->cmd->proc != quitCommand && c->cmd->proc != resetCommand) {
	rejectCommandFormat(c,
	"Can't execute '%s': only (P|S)SUBSCRIBE / "
	"(P|S)UNSUBSCRIBE / PING / QUIT / RESET are allowed in this context",
	c->cmd->fullname);
	return C_OK;
	}
	/* Only allow commands with flag "t", such as INFO, REPLICAOF and so on,
	* when replica-serve-stale-data is no and we are a replica with a broken
	* link with primary. */
	if (server.primary_host && server.repl_state != REPL_STATE_CONNECTED && server.repl_serve_stale_data == 0 &&
	is_denystale_command) {
	rejectCommand(c, shared.primarydownerr);
	return C_OK;
	}
	/* Loading DB? Return an error if the command has not the
	* CMD_LOADING flag. */
	if (server.loading && !server.async_loading && is_denyloading_command) {
	rejectCommand(c, shared.loadingerr);
	return C_OK;
	}
	/* During async-loading, block certain commands. */
	if (server.async_loading && is_deny_async_loading_command) {
	rejectCommand(c, shared.loadingerr);
	return C_OK;
	}
	/* when a busy job is being done (script / module)
	* Only allow a limited number of commands.
	* Note that we need to allow the transactions commands, otherwise clients
	* sending a transaction with pipelining without error checking, may have
	* the MULTI plus a few initial commands refused, then the timeout
	* condition resolves, and the bottom-half of the transaction gets
	* executed, see Github PR #7022. */
	if (isInsideYieldingLongCommand() && !(c->cmd->flags & CMD_ALLOW_BUSY)) {
	if (server.busy_module_yield_flags && server.busy_module_yield_reply) {
	rejectCommandFormat(c, "-BUSY %s", server.busy_module_yield_reply);
	} else if (server.busy_module_yield_flags) {
	rejectCommand(c, shared.slowmoduleerr);
	} else if (scriptIsEval()) {
	rejectCommand(c, shared.slowevalerr);
	} else {
	rejectCommand(c, shared.slowscripterr);
	}
	return C_OK;
	}
	/* Prevent a replica from sending commands that access the keyspace.
	* The main objective here is to prevent abuse of client pause check
	* from which replicas are exempt. */
	if (c->flag.replica && (is_may_replicate_command || is_write_command || is_read_command)) {
	rejectCommandFormat(c, "Replica can't interact with the keyspace");
	return C_OK;
	}

After, we would not go into processCommand, so all the linked lines will not execute.

Maybe you can help me understand why we are adding this change? If the reason is just optimization, I think we should skip this change. If the change is needed, we should double check that none of the functionality changes would be problematic

[murphyjacob4]

Is it resolved? I still see the comment

[murphyjacob4]

I get the argument to remove the read handler to prevent unbounded buffering. But why should we only remove the read handler for blocking for in use keys? Wouldn't this memory accumulation also be true for server blocks, like failover and slot migration?

If we don't need to have two different blocking behaviors (with read handler and without) then the system will be a lot less complex.

[harrylin98]

BLOCKED_POSTPONE blocks happen in the middle of the command exection — the client already sent a command, and the server defers its execution briefly. I think these are short-lived by design (failover completes, pause ends, migration finishes), so the read handler staying active doesn't cause meaningful memory accumulation.

[murphyjacob4]

But there are other types of blocks too that may benefit from this same behavior, right? What is special about blocking for in use keys when compared to say blocking for failover or blocking for slot migration.

[murphyjacob4]

I think these are short-lived by design (failover completes, pause ends, migration finishes), so the read handler staying active doesn't cause meaningful memory accumulation.

How long is something expected to be blocked for background processing by bgiterartor? Slot migration and failover can have blocks up to O(seconds) before timeout.

[harrylin98]

There is no accurate estimation for how long for a key to be blocked, caller needs to decided when to unblock. We can rework slot migration blocking to also remove the read handler in the future if needed.

[madolson]

Why is this a separate flag? Can we just the same 'b'