Repository: valkey-io/valkey
File: src/server.c (C)

Code before:
static void propagateNow(int dbid, robj **argv, int argc, int target) {
    if (!shouldPropagate(target)) return;

    /* This needs to be unreachable since the dataset should be fixed during
     * replica pause (otherwise data may be lost during a failover).
     *
     * Thought, there are exceptions:
     *
     * 1. We allow write commands that were queued up before and after to
     *    execute, if a CLIENT PAUSE executed during a transaction, we will
     *    track the state, the CLIENT PAUSE takes effect only after a transaction
     *    has finished.
     * 2. Primary loses a slot during the pause, deletes all keys and replicates
     *    DEL to its replicas. In this case, we will track the state, the dirty
     *    slots will be deleted in the end without affecting the data consistency.
     *
     * Note that case 2 can happen in one of the following scenarios:
     * 1) The primary waits for the replica to replicate before exiting, see
     *    shutdown-timeout in conf for more details. In this case, primary lost
     *    a slot during the SIGTERM waiting.
     * 2) The primary waits for the replica to replicate during a manual failover.
     *    In this case, primary lost a slot during the pausing.
     * 3) The primary was paused by CLIENT PAUSE, and lost a slot during the
     *    pausing. */
    serverAssert(!isPausedActions(PAUSE_ACTION_REPLICA) || server.client_pause_in_transaction ||
                 server.server_del_keys_in_slot);

    if (server.aof_state != AOF_OFF && target & PROPAGATE_AOF) feedAppendOnlyFile(dbid, argv, argc);
    if (target & PROPAGATE_REPL) replicationFeedReplicas(dbid, argv, argc);
}

/* Used inside commands to schedule the propagation of additional commands
 * after the current command is propagated to AOF / Replication.
 *
 * dbid is the database ID the command should be propagated into.
 * Arguments of the command to propagate are passed as an array of
 * Objects pointers of len 'argc', using the 'argv' vector.
 *
 * The function does not take a reference to the passed 'argv' vector,
 * so it is up to the caller to release the passed argv (but it is usually
 * stack allocated).  The function automatically increments ref count of
 * passed objects, so the caller does not need to. */
void alsoPropagate(int dbid, robj **argv, int argc, int target) {
    robj **argvcopy;
    int j;

    if (!shouldPropagate(target)) return;

    argvcopy = zmalloc(sizeof(robj *) * argc);
    for (j = 0; j < argc; j++) {
        argvcopy[j] = argv[j];

Review comment (performance):
You're right, it's good to introduce this variable if we want to have a correct assert. It is very similar to `client_pause_in_transaction`. Actually they have the same purpose: to bypass this assert.

I don't really like that propagateNow has to be aware of special features like cluster. Maybe we can use just one variable `allow_writes_during_client_pause` for both of these?

I want to avoid complexity, but I'm not sure what is the best way to do it. Can we just remove the assert and use test cases instead to check that eviction and expire don't replicate during client pause?

Code after:
static void propagateNow(int dbid, robj **argv, int argc, int target) {
    if (!shouldPropagate(target)) return;

    /* This needs to be unreachable since the dataset should be fixed during
     * replica pause (otherwise data may be lost during a failover).
     *
     * Though, there are exceptions:
     *
     * 1. We allow write commands that were queued up before and after to
     *    execute, if a CLIENT PAUSE executed during a transaction, we will
     *    track the state, the CLIENT PAUSE takes effect only after a transaction
     *    has finished.
     * 2. Primary loses a slot during the pause, deletes all keys and replicates
     *    DEL to its replicas. In this case, we will track the state, the dirty
     *    slots will be deleted in the end without affecting the data consistency.
     *
     * Note that case 2 can happen in one of the following scenarios:
     * 1) The primary waits for the replica to replicate before exiting, see
     *    shutdown-timeout in conf for more details. In this case, primary lost
     *    a slot during the SIGTERM waiting.
     * 2) The primary waits for the replica to replicate during a manual failover.
     *    In this case, primary lost a slot during the pausing.
     * 3) The primary was paused by CLIENT PAUSE, and lost a slot during the
     *    pausing. */
    serverAssert(!isPausedActions(PAUSE_ACTION_REPLICA) || server.client_pause_in_transaction ||
                 server.server_del_keys_in_slot);

    if (server.aof_state != AOF_OFF && target & PROPAGATE_AOF) feedAppendOnlyFile(dbid, argv, argc);
    if (target & PROPAGATE_REPL) replicationFeedReplicas(dbid, argv, argc);
}

/* Used inside commands to schedule the propagation of additional commands
 * after the current command is propagated to AOF / Replication.
 *
 * dbid is the database ID the command should be propagated into.
 * Arguments of the command to propagate are passed as an array of
 * Objects pointers of len 'argc', using the 'argv' vector.
 *
 * The function does not take a reference to the passed 'argv' vector,
 * so it is up to the caller to release the passed argv (but it is usually
 * stack allocated).  The function automatically increments ref count of
 * passed objects, so the caller does not need to. */
void alsoPropagate(int dbid, robj **argv, int argc, int target) {
    robj **argvcopy;
    int j;

    if (!shouldPropagate(target)) return;

    argvcopy = zmalloc(sizeof(robj *) * argc);
    for (j = 0; j < argc; j++) {
        argvcopy[j] = argv[j];