Optimistic Transactions with BoostRedis

Introduction

Welcome back! In the previous lesson, you learned how to batch multiple commands into a single request using Boost.Redis to reduce network round trips. Batching improves performance, but it is not atomic — other clients may still change data between your batched reads and writes. This lesson introduces WATCH together with MULTI / EXEC, which you use to build conditional, optimistic-locking transactions. The idea is: MULTI / EXEC makes a group of writes execute atomically. WATCH monitors one or more keys and causes EXEC to abort if any of those keys change after WATCH and before EXEC . In short: batching is non-atomic; atomicity requires MULTI / EXEC . WATCH ensures the keys stay unchanged between your read and your EXEC .

What You'll Learn

By the end of this lesson, you will be able to: Send WATCH for one or more keys using request.push() . Combine WATCH with MULTI / EXEC to perform conditional updates. Detect conflicts via EXEC's reply (nil /aborted), not via exceptions. Follow the typical optimistic sequence (on the same connection): WATCH read compute MULTI write EXEC (retry if aborted) Remember: request batching by itself is not atomic. Transactions require MULTI / EXEC, and WATCH ensures the keys you care about didn't change.

The Problem: Race Conditions in Read-Modify-Write

Imagine you're building a banking application. You want to update a user's balance by adding money to their account. The naive approach looks like this: GET balance:1 \to returns "100" Calculate in C++: 100 + 50 = 150 SET balance:1 150 What's the problem? Between steps 1 and 3, another client might also read the balance, compute a new value, and write it back. You might lose updates: Client A reads 100, adds 50 \to wants to write 150 Client B reads 100, adds 30 \to wants to write 130 Client B writes 130 first Client A writes 150 second Result: Client B's +30 is lost! The balance should be 180, not 150. This is called a race condition . Even if you batch the commands, they're not atomic — other clients can still interleave their operations.

How WATCH Solves This

WATCH is Redis's optimistic locking mechanism. Here's how it works: Before reading, you tell Redis: "Watch this key for changes" Redis remembers which keys you're watching You read the key and compute your new value When you execute MULTI / EXEC, Redis checks: "Did any watched key change since WATCH?" If yes : EXEC returns nil (aborted), and nothing is written If no : Your transaction executes atomically This is called optimistic locking because you optimistically assume no conflict will occur. If one does, you detect it and retry. Key insight: WATCH doesn't lock the key or block other clients. It just monitors for changes. If a change happens, you find out during EXEC . Connection-specific (reliability) insight: WATCH is stored on the server per connection . That means: You must run the whole sequence (WATCH \to read \to MULTI / EXEC) on the same Redis connection. If the connection drops or is reset, the watched state is lost. In practice, you must treat a disconnect as “transaction attempt failed” and retry from the beginning after reconnecting. Don’t share one connection across concurrent optimistic transactions unless you strictly serialize them; otherwise one transaction’s WATCH can interfere with another’s.

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

C++

#include <boost/redis.hpp>
#include <boost/asio.hpp>
#include <boost/redis/src.hpp>

#include <iostream>
#include <string>
#include <vector>
#include <thread>

namespace asio  = boost::asio;
namespace redis = boost::redis;

using connection = redis::connection;

C++

void update_balance(connection& conn, int user_id, int increment) {
    const std::string key = "balance:" + std::to_string(user_id);

    while (true) {
        // We'll fill in the steps here
    }
}

C++

void update_balance(connection& conn, int user_id, int increment) {
    const std::string key = "balance:" + std::to_string(user_id);

    while (true) {
        // Send WATCH command
        redis::request watch_req;
        watch_req.push("WATCH", key);  // Tell Redis: monitor this key

        redis::response<std::string> watch_resp;
        conn.async_exec(watch_req, watch_resp, asio::use_future).get();
        // WATCH returns "OK", which we can ignore

        // Continue to next step...
    }
}

C++

void update_balance(connection& conn, int user_id, int increment) {
    const std::string key = "balance:" + std::to_string(user_id);

    while (true) {
        // WATCH step from above...
        {
            redis::request watch_req;
            watch_req.push("WATCH", key);
            redis::response<std::string> watch_resp;
            conn.async_exec(watch_req, watch_resp, asio::use_future).get();
        }

        // Read the current balance
        int balance = 0;
        {
            redis::request get_req;
            get_req.push("GET", key);  // Read the current value

            redis::response<std::string> get_resp;
            conn.async_exec(get_req, get_resp, asio::use_future).get();

            // Extract the result
            auto& slot = std::get<0>(get_resp);  // result<std::string, error>
            if (slot.has_value()) {
                // Key exists, parse the balance
                try {
                    balance = std::stoi(slot.value());
                } catch (...) {
                    balance = 0;  // Treat malformed value as 0
                }
            } else {
                // Key doesn't exist or adapter error -> treat as 0
                balance = 0;
            }
        }

        // Continue to transaction...
    }
}

C++

void update_balance(connection& conn, int user_id, int increment) {
    const std::string key = "balance:" + std::to_string(user_id);

    while (true) {
        // WATCH and GET steps from above...
        int balance = 0;
        {
            redis::request watch_req;
            watch_req.push("WATCH", key);
            redis::response<std::string> watch_resp;
            conn.async_exec(watch_req, watch_resp, asio::use_future).get();
        }
        {
            redis::request get_req;
            get_req.push("GET", key);
            redis::response<std::string> get_resp;
            conn.async_exec(get_req, get_resp, asio::use_future).get();
            auto& slot = std::get<0>(get_resp);
            if (slot.has_value()) {
                try {
                    balance = std::stoi(slot.value());
                } catch (...) {
                    balance = 0;
                }
            }
        }

        // Start building the transaction
        redis::request tx_req;
        tx_req.push("MULTI");  // Start transaction (commands queue after this)

        // Continue to SET and EXEC...
    }
}

C++

void update_balance(connection& conn, int user_id, int increment) {
    const std::string key = "balance:" + std::to_string(user_id);

    while (true) {
        // WATCH, GET, and MULTI steps from above...
        int balance = 0;
        {
            redis::request watch_req;
            watch_req.push("WATCH", key);
            redis::response<std::string> watch_resp;
            conn.async_exec(watch_req, watch_resp, asio::use_future).get();
        }
        {
            redis::request get_req;
            get_req.push("GET", key);
            redis::response<std::string> get_resp;
            conn.async_exec(get_req, get_resp, asio::use_future).get();
            auto& slot = std::get<0>(get_resp);
            if (slot.has_value()) {
                try {
                    balance = std::stoi(slot.value());
                } catch (...) {
                    balance = 0;
                }
            }
        }

        redis::request tx_req;
        tx_req.push("MULTI");
        tx_req.push("SET", key, std::to_string(balance + increment));  // Queue the write

        // Continue to EXEC...
    }
}

C++

void update_balance(connection& conn, int user_id, int increment) {
    const std::string key = "balance:" + std::to_string(user_id);

    while (true) {
        // All previous steps...
        int balance = 0;
        {
            redis::request watch_req;
            watch_req.push("WATCH", key);
            redis::response<std::string> watch_resp;
            conn.async_exec(watch_req, watch_resp, asio::use_future).get();
        }
        {
            redis::request get_req;
            get_req.push("GET", key);
            redis::response<std::string> get_resp;
            conn.async_exec(get_req, get_resp, asio::use_future).get();
            auto& slot = std::get<0>(get_resp);
            if (slot.has_value()) {
                try {
                    balance = std::stoi(slot.value());
                } catch (...) {
                    balance = 0;
                }
            }
        }
        // Note: WATCH and GET were sent in earlier async_exec calls because we must read
        // the current value and compute the new balance *before* we can build the MULTI/SET/EXEC
        // request. Here we pipeline MULTI + SET + EXEC together (single round-trip) for efficiency.
        // This still counts as one optimistic transaction attempt because it's all on the same
        // connection and WATCH remains active until EXEC.
        redis::request tx_req;
        tx_req.push("MULTI");
        tx_req.push("SET", key, std::to_string(balance + increment));
        tx_req.push("EXEC");  // Execute the transaction (or abort)

        // The response types:
        // - MULTI returns "OK" (string)
        // - SET returns "QUEUED" while in transaction (string)
        // - EXEC returns optional<vector<string>>:
        //     - nullopt if aborted (conflict detected!)
        //     - vector of results if successful
        redis::response<
            std::string,                                // MULTI reply
            std::string,                                // SET reply ("QUEUED")
            std::optional<std::vector<std::string>>     // EXEC result or nil
        > tx_resp;

        conn.async_exec(tx_req, tx_resp, asio::use_future).get();

        // Extract the EXEC result
        auto& exec_slot = std::get<2>(tx_resp);  // result<optional<vector<string>>, error>

        if (!exec_slot.has_value()) {
            // Protocol/adapter error (not a WATCH conflict)
            throw std::runtime_error("EXEC adapter error");
        }

        const auto& exec_opt = exec_slot.value();  // optional<vector<string>>

        // Check if transaction was aborted (conflict detected)
        if (exec_opt.has_value()) {
            // EXEC returned a non-nil array => transaction succeeded
            break;  // Exit the retry loop
        } else {
            // EXEC returned nil => transaction aborted because WATCH detected a change
            // Loop continues to retry the entire process
            continue;
        }
    }
}

C++

int main() {
    try {
        asio::io_context ioc;
        connection conn(ioc);

        redis::config cfg;
        cfg.addr.host = "127.0.0.1";
        cfg.addr.port = "6379";

        // Run the connection in the background
        conn.async_run(cfg, redis::logger{redis::logger::level::disabled}, asio::detached);

        std::thread io_thread([&ioc] {
            ioc.run();
        });

        // Initialize the balance to 100
        {
            redis::request set_req;
            set_req.push("SET", "balance:1", "100");
            redis::response<std::string> set_resp;
            conn.async_exec(set_req, set_resp, asio::use_future).get();
        }

        // Perform the optimistic update: add 50 to the balance
        update_balance(conn, 1, 50);

        // Verify and print the final balance
        {
            redis::request get_req;
            get_req.push("GET", "balance:1");

            redis::response<std::string> get_resp;
            conn.async_exec(get_req, get_resp, asio::use_future).get();

            auto& slot = std::get<0>(get_resp);  // result<std::string, error>
            std::string balance_str = slot.has_value() ? slot.value() : "0";

            std::cout << "Updated balance for user 1: " << balance_str << '\n';
        }

        conn.cancel();
        ioc.stop();
        io_thread.join();
    } catch (const std::exception& e) {
        std::cerr << "Error: " << e.what() << '\n';
        return 1;
    }

    return 0;
}

Introduction

What You'll Learn

The Problem: Race Conditions in Read-Modify-Write

How WATCH Solves This

Join the 1M+ learners on CodeSignal

Building the Solution: Step by Step

Step 1: Setting Up the Connection

Step 2: WATCH the Key

Step 3: GET the Current Value

Step 4: Start the Transaction with MULTI

Step 5: Queue the SET Command

Step 6: Execute with EXEC and Handle Conflicts

Step 7: The Main Function (Putting It All Together)

Why It Matters