Optimistic Transactions with BoostRedis

Optimistic Transactions with BoostRedis | CodeSignal Learn

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

Step 1: Setting Up the Connection

First, let's set up our basic includes and connection boilerplate: 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;#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; We'll use a function to encapsulate the update logic with automatic retry: 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 } }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 } } The while (true) loop handles retries when conflicts are detected. Let's break down what happens inside.

Step 2: WATCH the Key

Input: Key name (balance:1) Output: Redis returns "OK" Purpose: Tell Redis to monitor this key for changes until EXEC 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... } }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... } } What's happening: We send WATCH balance:1. From this point forward, Redis tracks if any other client modifies this key. The command returns "OK" to confirm.

Step 3: GET the Current Value

Input: Key name (balance:1) Output: String value ("100") or nil if key doesn't exist Purpose: Read the current balance so we can compute the new value 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... } }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... } } What's happening: We GET the current value. We use .has_value() to check if the key exists, and parse the result. At this point, the key is being watched, but we haven't written anything yet. We compute the new balance in C++: balance + increment.

Step 4: Start the Transaction with MULTI

Input: None (command has no arguments) Output: "OK" Purpose: Start queuing commands; nothing executes yet 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... } }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... } } What's happening: MULTI tells Redis: "Queue the following commands. Don't execute them yet." It returns "OK" immediately.

Step 5: Queue the SET Command

Input: Key name and new value (balance:1, "150") Output: "QUEUED" (not executed yet) Purpose: Queue the write operation to update the balance 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... } }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... } } What's happening: We queue SET balance:1 150. Because we're inside a MULTI block, this command doesn't execute yet. Redis just returns "QUEUED" to confirm it's in the queue.

Step 6: Execute with EXEC and Handle Conflicts

Input: None (command has no arguments) Output: Either nil (conflict detected) or array of command results (success) Purpose: Execute all queued commands atomically, or abort if watched keys changed 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; } } }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; } } } What's happening: EXEC checks: "Did any watched key change since WATCH?" If yes: EXEC returns nil (represented as std::nullopt in C++), and nothing is written. We detect this and continue to retry from WATCH. If no: All queued commands execute atomically, and EXEC returns their results. We break out of the retry loop. Conflict detection: We check if exec_opt.has_value(). If it's std::nullopt, we detected a conflict and retry. EXEC automatically clears the WATCH state, so we must WATCH again.

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

Finally, here's the main function that initializes the balance, performs the update, and verifies the result: 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; }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; } What's happening here: We set up the connection and run io_context in a background thread We initialize balance:1 to 100 We call update_balance(conn, 1, 50) which uses WATCH + MULTI/EXEC to safely add 50 We verify the final balance and print it (should be 150) We clean up by canceling the connection and stopping the io_context

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