DEV Community: Onion Lee

Designing Distributed Cron: At-Most-Once Execution

Onion Lee — Fri, 30 May 2025 06:01:35 +0000

Introduction

Inspired by Google’s “Reliable Cron across the Planet”, I set out to design my own version of a distributed cron system. The core goal was simple: execute a binary at scheduled times - but do so in a scalable and fault-tolerant way. In other words, my system should continue operating even when nodes crash.

To guide design decisions and avoid unbounded scaling of my service, I set a few baseline expectations for system load:

500–1,000 worker nodes
Thousands of scheduled tasks per day
Task duration: between 1–30 seconds

One more thing is that I deliberately trade off liveness (okay to skip a launch) for safety (never duplicate work).

Overall Architecture

Centralized state store: PostgreSQL (replicated for HA)
Coordinator-based dispatch: Coordinators own all DB I/O and task routing via gRPC.
Workers: Push heartbeats; receive tasks over RPC.
Scheduler: Separate component ingests user cron specs and writes them to the DB.

Naive Solution:

The naive solution for a fault-tolerant cron is to let each worker node independently poll the database for pending tasks and execute them when found. While conceptually simple, this approach does not scale. With hundreds of workers polling every few seconds, the database becomes a central bottleneck, overwhelmed by read contention.

Moreover, implementing heartbeats under this decentralized model only amplifies database load, as each worker must constantly read and write metadata about its own liveness or task state.

Using a Coordinator:

To avoid overloading the database, I adopted a coordinator-based design. In this model, a centralized coordinator node (or set of nodes) handles all database interactions — dispatching tasks to workers to updating task's metadata. Workers do not poll the database directly; instead, they receive tasks pushed from the coordinator via RPC. This architecture offloads and shifts the burden of task coordination from the database.

Additionally, worker heartbeats are sent to the coordinator (not the database), further reducing I/O pressure on persistent storage.

For fault tolerance, the system supports multiple coordinator instances. While leader election via consensus protocols like Raft (my own implementation) or Paxos can ensure a single active coordinator, I opted for a simpler approach: allowing multiple coordinators to run concurrently. To avoid race conditions, PostgreSQL advisory locks are used during task table scans, ensuring mutual exclusion. This design keeps coordinators stateless and enables horizontal scalability without complex coordination logic.

Worker Nodes

The coordinator is responsible for distributing tasks to worker nodes and maintaining visibility into their availability. A naive strategy might involve the coordinator pinging workers on-demand when a task needs dispatching. However, this has major drawbacks:

It fails to proactively detect crashes or network partitions — if a node silently dies, the system won’t know until it tries (and fails) to assign work.
It offers no insight into system load, making it hard to balance task distribution based on available capacity.

To address these limitations, I would use a heartbeat mechanism. Here, each worker node periodically sends a heartbeat RPC to the coordinator, reporting its liveness and current workload. This allows the coordinator to maintain an up-to-date in-memory view of cluster health and dispatch tasks intelligently if needed.

A simplified version of the heartbeat payload looks like this:

{
  "worker_id": "string",        // worker ID
  "timestamp": 1234567890,      // Unix epoch (in ms) when this heartbeat was sent
  "status": "idle"|"busy",      // state flag
  "load": {
    "running_tasks":            // number of tasks currently executing
    "max_capacity":             // total parallel slots this worker can handle
  }
}

Coordinator

To track worker node health, the coordinator maintains an in-memory map of active nodes, keyed by node ID, with status and last heartbeat timestamp. A background reaper loop periodically scans this map to detect unresponsive nodes. If multiple coordinators may be active, this map would typically be backed by a centralized store such as Redis for shared access.

Task Dispatching:

To assign tasks, the coordinator periodically scans the database for eligible tasks (e.g., scheduled and uncompleted). A naive approach might wrap the scan, RPC dispatch, and database update in a single transaction, but this is inefficient—it holds database locks during potentially slow RPC calls, causing contention under load.

Ensuring At-Most-Once Delivery:

Nonetheless, handling this is very tricky because this is essentially a matter of exactly-once delivery semantics. If coordinator’s task dispatching RPC reaches the worker node but the response partitions, coordinator may think the task hasn’t been dispatched and dispatches it on the next scan, resulting in a duplicate processing. To craft a solution, let me first define our business requirement discussed earlier: favor skipping task launches rather than risking duplicate executions.

Two-Phase Dispatch Protocol:

To guard against RPC/network partitions, I designed a lightweight, two-phase dispatch protocol (inspired by 2PC’s prepare/commit) that enforces at-most-once task execution:

The flow is straightforward:

Phase 1: Pre-Assignment
- Worker immediately replies “yes”/“no.”
- yes → mark task TEMPORARY_ASSIGNED; no → leave NEW.
Phase 2: Completion Confirmation
- After work, worker calls ConfirmDone(taskID) (retry up to N times).
- If task is still TEMPORARY_ASSIGNED, coordinator marks it DONE and returns “ok.”
- Otherwise returns “not ok,” and worker aborts.

In the face of lost RPCs, a task may be retried later (skipped once), but never executed twice. This achieves at-most-once semantics with minimal overhead compared to full 2PC.

Worker Node Crashes:

When a worker completes its task, it sends a response back to the coordinator, which then updates the database to mark the task as completed.

If a worker crashes or becomes unresponsive, the reaper loop detects this via the in-memory heartbeat map. The coordinator then queries the database for any active tasks previously assigned to that node and logs them for further action — such as for re-dispatching, retries, or administrative reporting.

Google’s approach, as noted earlier, employs sophisticated checkpointing to ensure idempotent task processing and seamless failover for running tasks. To keep my system straightforward, I opted for a simpler strategy: record crash context for manual or automated remediation.

Scheduler

We have coordinator and worker nodes. The final component of the service is scheduler, which accepts user cron definitions (e.g. “0 0 * * *”), and writes them into the DB as NEW tasks.

A naive approach would have the coordinator handle client requests directly. While feasible, this is uncommon in production systems, as it makes it difficult to scale components independently. Instead, introducing a separate scheduler node enables a clean separation of responsibilities and improves scalability.

Conclusion

Ensuring exactly-once task delivery in distributed systems is inherently challenging. My design consciously favors occasional task skipping over duplicate execution, striking a balance between reliability and simplicity. While it doesn’t aim to meet the scale of Google’s stringent SLAs, I hope it provides a practical and effective solution for moderate workloads.

JavaLSM: An LSM Tree Based Key-Value Storage in Java

Onion Lee — Sun, 11 May 2025 14:01:52 +0000

In this post, I’ll share how I built my own LSM-based storage engine from scratch in Java. I focused on implementing the full data pipeline correctly, while using existing solutions for low-level details such as balanced search tree or probabilistic data structures.

seungwonlee003 / JavaLSM

An LSM Tree based key-value storage engine written in Java

JavaLSM: An LSM Tree Based Key-Value Storage in Java

Introduction

Blog Post

JavaLSM is an LSM tree based storage engine written in Java.

It offers a simple key-value interface and is designed to be used as an embedded storage engine.

Features

Simple key-value interface: put(key, value), get(key), delete(key)
LSM Tree Architecture: In-memory buffer + disk-based SSTables
Fast Reads: Powered by Bloom filters and block indexes
Concurrent Reads: Fine-grained locking allows reads during compaction
Automatic Compaction: Full-level leveled compaction strategy optimizes storage
Crash Recovery: Write-Ahead Log (WAL) ensures durability of in-memory write buffer

Usage

Opening a DB instance

Create a DB instance by calling its constructor. No arguments are required. By default, Write-Ahead Logging (WAL) is enabled, and both the memtable and SSTable sizes are set to 16GB.

DB db = new DB();

Writing to the DB

Use the put method to insert or update entries in the DB. It accepts…

View on GitHub

Basics

LSM tree-based storage is a write-optimized design that uses an in-memory buffer to store initial writes and immutable sorted string tables (SSTables) for persistent storage on disk. The in-memory buffer, called a memtable, typically uses a balanced search tree as its underlying data structure. This keeps entries sorted, enabling efficient in-memory writes and reads with O(log n) complexity.

An SSTable is an immutable, sorted, and append-only structure stored on disk. Since the memtable is already sorted, flushing it to disk to create a new SSTable is a relatively cheap O(n) operation. To optimize space and remove old or duplicate entries, background compaction runs periodically, merging SSTables and discarding obsolete data.

Memtable

With the basics in place, I implemented the memtable using Java's built-in TreeMap, which is backed by a red-black tree. I kept both memtables and SSTables at a fixed size, ensuring that flushing a memtable to disk never required splitting SSTables—this simplified the process.

Block Index

In LevelDB, data is read and written in ~4KB blocks to minimize costly disk I/O. I adopted a similar approach by maintaining an in-memory index per block that stores the first key, block offset, and block size. This was stored in Java’s TreeMap. Assuming each entry fits within a block (i.e., <4KB) and that block indexes fit in memory, this reduces lookup time from O(N) to O(log N) per SSTable by avoiding full table scans. See below for the SSTable’s get method:


public String get(String key) {
    if (key.compareTo(minKey) < 0 || key.compareTo(maxKey) > 0) {
        return null;
    }

    if (!bloomFilterUtil.mightContain(key)) {
        return null;
    }

    Map.Entry<String, BlockInfo> indexEntry = index.floorEntry(key);
    if (indexEntry == null) {
        return null;
    }

    BlockInfo blockInfo = indexEntry.getValue();

    try (RandomAccessFile file = new RandomAccessFile(filePath, "r")) {
        file.seek(blockInfo.offset);
        byte[] blockData = new byte[(int) blockInfo.length];
        file.readFully(blockData);

        try (ByteArrayInputStream bais = new ByteArrayInputStream(blockData);
             DataInputStream dis = new DataInputStream(bais)) {
            while (dis.available() > 0) {
                int keyLength = dis.readInt();
                byte[] keyBytes = new byte[keyLength];
                dis.readFully(keyBytes);
                String currentKey = new String(keyBytes, StandardCharsets.UTF_8);

                int valueLength = dis.readInt();
                byte[] valueBytes = new byte[valueLength];
                dis.readFully(valueBytes);

                if (currentKey.equals(key)) {
                    return IOUtils.deserializeValue(valueBytes);
                }
                if (currentKey.compareTo(key) > 0) {
                    return null;
                }
            }
        }
    } catch (IOException e) {
        throw new RuntimeException("Failed to read SSTable: " + filePath, e);
    }

    return null;
}

Compaction

With efficient reads and writes in place, I implemented a simple compaction to control the ever-growing log of SSTables. I used a k-way merge algorithm with min-heap to combine multiple SSTables into new, smaller SSTables without duplicates. Here's how it works: a compaction thread acquires a read lock and merges data between level n and level n+1. It combines all SSTables from level n with selected SSTables from level n+1, maintaining a list ordered by recency. An iterator is created for each SSTable, and their first entries are inserted into a min-heap. The heap is sorted by keys, breaking ties by favoring entries from more recent SSTables (which reflect newer writes). As elements are popped from the heap, they're added to the new SSTable only if they aren't marked as deleted (tombstoned) and haven't already been written. Duplicates are skipped, and iterators advance. The heap maintains at most k elements at a time. If the current SSTable exceeds its size threshold during compaction, a new SSTable file is created.

Full-level Leveled Compaction

This strategy is known as full-level leveled compaction. It simplifies comparator logic and merging while maintaining a time complexity of O(N log k), where k is the number of input SSTables and N is the total number of entries. I implemented fine-grained locking so that reads can continue on both memtables and SSTables even during this CPU-intensive compaction process. A write lock is only briefly acquired on the manifest file when swapping out old SSTables for newly compacted ones.

In production systems, a more advanced approach called partial compaction compacts only subsets of SSTables over time, reducing write amplification. However, I didn’t implement this, as compaction wasn’t the bottleneck in my case.

Bloom filter

I added an optimization for searching the keys that do not exist. In such cases, without optimization, the database would have to check every SSTable at each level, which is expensive. I experimented with Guava’s Bloom filter to measure their impact on reducing unnecessary disk lookups. In the benchmark for negative gets (where the key is absent), performance improved by a factor of 10 compared to random access get benchmarks.

Write-ahead Log

To ensure durability of writes buffered in the memtable, I implemented a write-ahead log (WAL), an append-only log that guarantees persistence even if the system crashes before flushing data to disk.

Finally, here’s an overview diagram of my LSM storage engine architecture:

Initially, I thought the concept behind LSM storage engines was simple, but building a complete pipeline was much more challenging. There are far more subtle details and possible optimizations than I expected. Nonetheless, through this project, I gained a much deeper understanding of how LSM-tree-based storage engines work—and I hope this knowledge will help me grow into a better engineer.

JavaRaft: Raft Distributed Consensus Protocol

Onion Lee — Sun, 11 May 2025 02:23:05 +0000

While studying for System Design interviews, I picked up some basic knowledge and got interested in distributed systems. After some studies, I still found many distributed systems algorithms a bit abstract and hand-wavy. To gain some hands-on experience, I decided to build Raft consensus algorithm from scratch.

It took me a few weeks to fully understand the Raft paper and implement the core algorithm. In this post, I want to reiterate on my learning of some key details of my Raft-based distributed key-value store.

seungwonlee003 / JavaRaft

Java-based Raft consensus implementation with Spring Boot RPC and in-memory key-value store.

Introduction

Demo Video

Blog Post

A Java-based implementation of the Raft consensus algorithm, designed to support consistent (CP) distributed systems like distributed key-value stores.

This project includes:

The core Raft algorithm
A set of pluggable interfaces for state machines and persistence
A Spring Boot-based RPC layer for inter-node communication
A simple in-memory key-value store with HTTP endpoints for PUT, GET, and DELETE operations (as a reference implementation)

Developers can extend this framework by implementing custom state machines and storage backends to build their own distributed services.

Built with Java, Spring Boot, Lombok, and SLF4J.

Features

Features reference the section number of the Raft paper:

Leader election (§5.2)
Log replication (§5.3)
Election restriction (§5.4.1)
Committing entries from previous terms (§5.4.2)
Follower and candidate crashes (§5.5)
Implementing linearizable semantics (§8)

Usage

Try out the distributed key-value API.

Configure application properties for nodes and build the JAR (skipping tests):

mvn clean package -DskipTests

…

View on GitHub

While my implementation is far from production-ready, my goal was to build a system that works under basic conditions and demonstrates the core ideas.

To keep things focused, I won’t go into line-by-line code or the fine-grained interactions between components. Instead, I’ll walk through a high-level overview of how I implemented the core Raft algorithm.

Leader election (§5.2):

The first module I developed is the leader election. In Raft, all nodes begin as followers, and a leader is elected when a follower’s election timer expires.

To enable this, I implemented an election timer that triggers an election upon expiration. It is called when the node transition its state to a follower.

// If election timeout elapses without receiving AppendEntries RPC from current leader or granting vote to candidate:
// convert to candidate (§5.2)
public void reset() {
    cancel();

    long minTimeout = config.getElectionTimeoutMillisMin();
    long maxTimeout = config.getElectionTimeoutMillisMax();
    long timeout = minTimeout + random.nextInt((int)(maxTimeout - minTimeout));

    electionFuture = scheduler.schedule(() -> {
        electionManager.startElection();
    }, timeout, TimeUnit.MILLISECONDS);
}

The ElectionManager class includes two primary methods: startElection() and handleVoteRequest(). The startElection() method allocates threads from a pool based on the number of peers and sends asynchronous vote requests to each follower via their /requestVote endpoint. Then, the main thread completes the method by resetting the election timer. Here, a timeout, configured in the RestTemplate, ensures responses arrive within a duration shorter than the election timer, preventing overlapping elections on the same node.

If a peer’s response indicates a higher term, the node immediately steps down to the follower state, and the election concludes. If less than majority of responses are received, the election timer resets. Otherwise, the node becomes a leader, cancels the ongoing election timer to prevent any new elections, and starts sending heartbeats to the follower nodes. HandleVoteRequest method implementation is rather straightforward following the Raft paper, so will be skipped.

/**
 * On conversion to candidate, start election:
 * Increment currentTerm, vote for self, reset election timer, send RequestVote RPCs to all other servers.
 */
public void startElection() {
    lockManager.getStateWriteLock().lock();
    lockManager.getLogReadLock().lock();
    try {
        if (nodeState.getCurrentRole() == Role.LEADER) {
            return;
        }

        nodeState.setCurrentRole(Role.CANDIDATE);
        nodeState.incrementTerm();
        nodeState.setVotedFor(nodeState.getNodeId());

        int currentTerm = nodeState.getCurrentTerm();
        int lastLogIndex = raftLog.getLastIndex();
        int lastLogTerm = raftLog.getLastTerm();

        List<CompletableFuture<VoteResponseDto>> voteFutures = new ArrayList<>();
        ExecutorService executor = Executors.newCachedThreadPool();

        for (String peerUrl : config.getPeerUrls().values()) {
            CompletableFuture<VoteResponseDto> voteFuture = CompletableFuture
                    .supplyAsync(() -> requestVote(
                            currentTerm,
                            nodeState.getNodeId(),
                            lastLogIndex,
                            lastLogTerm,
                            peerUrl
                    ), executor)
                    .orTimeout(config.getElectionRpcTimeoutMillis(), TimeUnit.MILLISECONDS)
                    .exceptionally(throwable -> {
                        return new VoteResponseDto(currentTerm, false);
                    });
            voteFutures.add(voteFuture);
        }

        int majority = (config.getPeerUrls().size() + 1) / 2 + 1;
        AtomicInteger voteCount = new AtomicInteger(1);

        // If votes received from majority of servers: become leader (§5.2).
        for (CompletableFuture<VoteResponseDto> future : voteFutures) {
            future.thenAccept(response -> {
                lockManager.getStateWriteLock().lock();
                try {
                    if (nodeState.getCurrentRole() != Role.CANDIDATE || nodeState.getCurrentTerm() != currentTerm) {
                        return;
                    }
                    if (response != null && response.isVoteGranted()) {
                        int newVoteCount = voteCount.incrementAndGet();
                        if (newVoteCount >= majority) {
                            stateManager.becomeLeader();
                        }
                    }
                } finally {
                    lockManager.getStateWriteLock().unlock();
                }
            });
        }
        stateManager.resetElectionTimer();
    } finally {
        lockManager.getLogReadLock().unlock();
        lockManager.getStateWriteLock().unlock();
    }
}

Log replication (§5.3):

The second module I implemented is the log replication. Given the complexity of this logic, I splitted the responsibilities into three separate classes: one for handling client write requests, another for initiating log replication, and lastly for processing append entries.

AppendEntriesHandler.java
ClientRequestHandler.java
LogReplicator.java
RaftReplicationManager.java

To manage these interactions, I used the Facade design pattern, introducing a centralized ReplicationManager class to delegate calls to the appropriate classes.

...
public class RaftReplicationManager {
    ...

    public boolean handleClientRequest(LogEntry entry) {
        return clientRequestHandler.handle(entry);
    }

    public void startLogReplication() {
        logReplicator.start();
    }

    public AppendEntriesResponseDTO handleAppendEntries(AppendEntriesDTO dto) {
        return appendEntriesHandler.handle(dto);
    }

    public void initializeIndices() {
        logReplicator.initializeIndices();
    }
}

The replication process begins with the start() method in the LogReplicator class. This method relies on several key data structures defined by Raft: nextIndex[] and matchIndex[], which track replication progress for each follower, and a pendingReplication[] map to manage ongoing replication tasks.

When a node becomes the leader after winning an election, this method iterates over all followers in the pendingReplication[] map and launches asynchronous log replication for each one if not already launched, utilizing threads from a fixed thread pool.

// Upon election: send initial empty AppendEntries RPCs (heartbeat) to each server; repeat during idle periods
// to prevent election timeouts (§5.2)
public void start() {
    lockManager.getStateReadLock().lock();
    try {
        if (nodeState.getCurrentRole() != Role.LEADER) return;
    } finally {
        lockManager.getStateReadLock().unlock();
    }

    for (String peer : config.getPeerUrls().values()) {
        if (!pendingReplication.getOrDefault(peer, false)) {
            pendingReplication.put(peer, true);
            executor.submit(() -> replicateLoop(peer));
        }
    }
}

For each follower, an asynchronous thread executes the replicateLoop() method. This method runs a while loop that continuously replicates new log entries to the /appendEntries endpoint of the follower, catches followers up on missing entries, or sends empty log entries as heartbeats to maintain leadership. The loop’s back-off period aligns with the heartbeat interval.

// Sends heartbeats/log entries to peer, adjusting sleep time to match heartbeat interval
private void replicateLoop(String peer) {
    int backoff = config.getHeartbeatIntervalMillis();
    while (true) {
        lockManager.getStateReadLock().lock();
        try {
            if (nodeState.getCurrentRole() != Role.LEADER) break;
        } finally {
            lockManager.getStateReadLock().unlock();
        }

        lockManager.getLogWriteLock().lock();
        lockManager.getStateWriteLock().lock();
        lockManager.getStateMachineWriteLock().lock();
        long startTime = System.currentTimeMillis();
        try {
            boolean ok = replicate(peer);
            if (ok) updateCommitIndex();
        } finally {
            lockManager.getStateMachineWriteLock().unlock();
            lockManager.getStateWriteLock().unlock();
            lockManager.getLogWriteLock().unlock();
        }
        long duration = System.currentTimeMillis() - startTime;
        long sleepTime = Math.max(0, backoff - duration);
        try {
            Thread.sleep(sleepTime);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            break;
        }
    }
    pendingReplication.put(peer, false);
}

Here, the RestTemplate timeout is configured to prevent the loop from stalling indefinitely, which could otherwise trigger a unneccessary election on a follower.

When the replication loop succeeds—assuming no follower responds with a higher term—the leader updates the corresponding nextIndex[] and matchIndex[] values and invokes updateCommitIndex() method.

// If last log index >= nextIndex for a follower: send AppendEntries RPC with log entries starting at nextIndex.
private boolean replicate(String peer) {
    int ni = nextIndex.get(peer);
    int prevIdx = ni - 1;
    int prevTerm = (prevIdx >= 0) ? raftLog.getTermAt(prevIdx) : 0;
    List<LogEntry> entries = raftLog.getEntriesFrom(ni);

    AppendEntriesRequestDto dto = new AppendEntriesRequestDto(
            nodeState.getCurrentTerm(), nodeState.getNodeId(),
            prevIdx, prevTerm, entries, raftLog.getCommitIndex()
    );

    String peerId = config.getPeerUrls().entrySet().stream()
            .filter(entry -> entry.getValue().equals(peer))
            .map(Map.Entry::getKey)
            .map(String::valueOf)
            .findFirst()
            .orElse("unknown");

    try {
        String url = peer + "/raft/appendEntries";
        ResponseEntity<AppendEntriesResponseDto> res = restTemplate.postForEntity(url, dto, AppendEntriesResponseDto.class);
        AppendEntriesResponseDto body = res.getBody() != null ? res.getBody() : new AppendEntriesResponseDto(-1, false);
        if (body.getTerm() > nodeState.getCurrentTerm()) {
            nodeStateManager.becomeFollower(body.getTerm());
            return false;
        }
        if (body.isSuccess()) {
            nextIndex.put(peer, ni + entries.size());
            matchIndex.put(peer, ni + entries.size() - 1);
            return true;
        } else {
            int newNextIndex = Math.max(0, ni - 1);
            nextIndex.put(peer, newNextIndex);
            return false;
        }
    } catch (ResourceAccessException e) {
        return false;
    } catch (Exception e) {
        return false;
    }
}

This, in turn, triggers applyEntries() method, which applies log entries up to the commit index to the data store. Additionally, when advancing the commit index, it ensures that only entries from the leader’s current term are committed. This upholds Raft’s safety guarantee, preventing the application of stale entries from prior terms. This is shown below:

private void updateCommitIndex() {
    int majority = (config.getPeerUrlList().size() + 1) / 2 + 1;
    int term = nodeState.getCurrentTerm();
    for (int i = log.getLastIndex(); i > log.getCommitIndex(); i--) {
        int count = 1;
        for (int idx : matchIndex.values()) {
            if (idx >= i) count++;
        }
        if (count >= majority && log.getTermAt(i) == term) {
            log.setCommitIndex(i);
            applyEntries();
            break;
        }
    }
}

From the follower’s perspective, the handle() method in the AppendEntriesHandler class processes the AppendEntriesDTO received from the leader.

It first verifies the sender’s leadership by checking the term. Next, it compares its log with the leader’s nextIndex[] for that follower. If the follower’s log is not up-to-date due to a mismatch in prior entries, it rejects the request, prompting the leader to decrement the follower’s nextIndex[] by one and retry.

If the log matches, the follower accepts the entries according to the log matching property (§5.2), updates its commit index, and applies the entries to its state machine up to the new commit index.

public AppendEntriesResponseDto handle(AppendEntriesRequestDto dto) {
    lockManager.getLogWriteLock().lock();
    lockManager.getStateWriteLock().lock();
    lockManager.getStateMachineWriteLock().lock();
    try {
        int term = nodeState.getCurrentTerm();
        int leaderTerm = dto.getTerm();

        // Reply false if term < currentTerm (§5.1)
        if (leaderTerm < term) return new AppendEntriesResponseDto(term, false);

        // If RPC request or response contains term T > currentTerm: set currentTerm = T, convert to follower (§5.1)
        if (leaderTerm > term) {
            stateManager.becomeFollower(leaderTerm);
            term = leaderTerm;
        }

        nodeState.setCurrentLeader(dto.getLeaderId());

        // Reply false if log doesn't contain an entry at prevLogIndex whose term matches prevLogTerm (§5.3)
        if (dto.getPrevLogIndex() > 0 &&
                (!raftLog.containsEntryAt(dto.getPrevLogIndex()) ||
                        raftLog.getTermAt(dto.getPrevLogIndex()) != dto.getPrevLogTerm())) {
            stateManager.resetElectionTimer();
            return new AppendEntriesResponseDto(term, false);
        }

        // If an existing entry conflicts with a new one (same index but different terms), delete the existing
        // entry and all that follow it (§5.3). Append any new entries not already in the log.
        int index = dto.getPrevLogIndex() + 1;
        List<LogEntry> entries = dto.getEntries();
        if (!entries.isEmpty()) {
            for (int i = 0; i < entries.size(); i++) {
                int logIndex = index + i;
                if (raftLog.containsEntryAt(logIndex) && raftLog.getTermAt(logIndex) != entries.get(i).getTerm()) {
                    raftLog.deleteFrom(logIndex);
                    raftLog.appendAll(entries.subList(i, entries.size()));
                    break;
                }
            }
            if (!raftLog.containsEntryAt(index)) {
                raftLog.appendAll(entries);
            }
        }

        // If leaderCommit > commitIndex, set commitIndex = min(leaderCommit, index of the last new entry)
        if (dto.getLeaderCommit() > raftLog.getCommitIndex()) {
            int lastNew = dto.getPrevLogIndex() + entries.size();
            raftLog.setCommitIndex(Math.min(dto.getLeaderCommit(), lastNew));
            applyEntries();
        }
        stateManager.resetElectionTimer();
        return new AppendEntriesResponseDto(term, true);
    } finally {
        lockManager.getStateMachineWriteLock().unlock();
        lockManager.getStateWriteLock().unlock();
        lockManager.getLogWriteLock().unlock();
    }
}

Finally, the ClientRequestHandler manages client writes to the database. Its handle() method appends the entry to the leader’s log for replication and monitors the entry’s index. It returns a success response only when the last applied index surpasses the entry’s index, confirming that the write has been applied to the state machine.

This mechanism supports linearizability, ensuring that subsequent client reads following a successful write reflect the updated state machine.

// If command received from the client: append entry to local log, respond after entry applied
// to state machine (§5.3)
public boolean handle(LogEntry clientEntry) {
    lockManager.getStateReadLock().lock();
    lockManager.getLogWriteLock().lock();
    int entryIndex;
    try {
        if (!raftNodeState.isLeader()) return false;
        raftLog.append(clientEntry);
        entryIndex = raftLog.getLastIndex();
    } finally {
        lockManager.getLogWriteLock().unlock();
        lockManager.getStateReadLock().unlock();
    }

    long start = System.currentTimeMillis();
    long timeoutMillis = raftConfig.getClientRequestTimeoutMillis();

    while (true) {
        lockManager.getLogReadLock().lock();
        lockManager.getStateReadLock().lock();
        try {
            if (raftNodeState.getCurrentRole() != Role.LEADER) {
                return false;
            }
            if (raftNodeState.getLastApplied() >= entryIndex) {
                return true;
            }
        } finally {
            lockManager.getStateReadLock().unlock();
            lockManager.getLogReadLock().unlock();
        }

        if (System.currentTimeMillis() - start > timeoutMillis) {
            return false;
        }
        try {
            Thread.sleep(50);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        }
    }
}

Here, a slight latency is introduced because client writes are aligned with the heartbeat cycle. If the heartbeat interval is 1 second, the leader may wait up to 1 second before replicating the write. While the Raft protocol allows independent AppendEntries RPCs for writes (separate from heartbeats), I chose to synchronize client writes with the heartbeat cycle to simplify concurrency handling and to take advantage of batched replication.

Before concluding, I want to address concurrency concerns. In Raft, concurrent threads handling RequestVote, AppendEntries, and election timeouts can introduce race conditions by simultaneously modifying shared node state (such as currentTerm, votedFor, and the log). To ensure serializability and prevent such race conditions, I implemented a centralized read-write lock that guards access to the node’s state, log, and state machine.

This concludes the explanation of the core Raft logics. Now, I want to talk about the fundamental concern that I addressed per the Raft paper known as "Implementing linearizable semantics" (§8).

Ensuring Lineraizable Semantics

In Raft, writes are committed once they reach a majority quorum, and the leader is guaranteed to hold all committed log entries. However, this does not inherently ensure that reads routed to the leader always reflect the most up-to-date state, which is necessary for linearizability.

Consider a scenario where a client is connected to a leader that loses its leadership due to a network partition, isolating it from the majority of the cluster. If the client reconnects to this former leader before the partition heals, the node may still consider itself the leader and serve stale data from its state machine data that could have been superseded by a new leader. This violates linearizability, as the read does not reflect the latest committed writes.

Raft paper proposes different solutions to this and I implemented a mechanism where the client get operation will always send no-op entry to the followers to confirm leadership. If majority of nodes accepts the entry through /appendEntries endpoint logic, it guarantees that the leader is not stale, thereby confirming reads as linearizable. While this introduces extra network round trip, it provides just-enough solution given my current implementation.

More efficient alternatives, such as ReadIndex or lease-based reads, exist, but I chose this straightforward approach. Sorry for the messy code below: this was because I didn’t plan to extend this class any further.

@GetMapping("/get")
public ResponseEntity<String> get(@RequestParam String key) {
    final String NO_OP_ENTRY = "NO_OP_ENTRY";
    try {
        handleRead(new WriteRequestDto(NO_OP_ENTRY, Long.MAX_VALUE, NO_OP_ENTRY, NO_OP_ENTRY));
    } catch (Exception e) {
        return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                .body("Failed to process read request: " + e.getMessage());
    }

    lockManager.getStateMachineReadLock().lock();
    try {
        String value = kvStore.get(key);
        if (value == null) {
            return ResponseEntity.status(HttpStatus.NOT_FOUND)
                    .body("Key not found: " + key);
        }
        return ResponseEntity.ok(value);
    } catch (Exception e) {
        return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                .body("Error retrieving key: " + e.getMessage());
    } finally {
        lockManager.getStateMachineReadLock().unlock();
    }
}

private void handleRead(WriteRequestDto request) {
    LogEntry entry = new LogEntry(
            nodeState.getCurrentTerm(),
            request.getKey(),
            request.getValue(),
            LogEntry.Operation.PUT,
            request.getClientId(),
            request.getSequenceNumber()
    );
    boolean committed = replicationManager.handleClientRequest(entry);
    if (!committed) throw new RuntimeException("Can't process the read at this moment");
}

Ensuring Lineraizable Semantics 2

Raft paper discusses a scenario where “if the leader crashes after committing the log entry but before responding to the client, the client will retry the command with a new leader, causing it to be executed a second time”. Client-side deduplication solves this by attaching a unique identifier (e.g., a sequence number) to each request. The server tracks these IDs, rejecting duplicates. This preserves both correctness and linearizability, addressing one of the key challenges.

In the implementation, I modified the apply() method in the StateMachine class. Before applying a write, it queries the database (in-memory) to compare the log entry’s sequence number with the last recorded sequence number for that client ID. The write is applied only if the new sequence number is greater, confirming it as a fresh operation rather than a duplicate.

@Override
public void apply(LogEntry entry) {
    if (entry == null) {
        throw new IllegalArgumentException("Log entry cannot be null");
    }

    // deduplication check
    String clientId = entry.getClientId();
    long sequenceNumber = entry.getSequenceNumber();
    Long lastSequenceNumber = kvStore.getLastSequenceNumber(clientId);
    if (lastSequenceNumber != null && sequenceNumber <= lastSequenceNumber) {
        return;
    }
...
}

Persistence

The final requirement is persistence. The state machine must ensure durability, meaning no data is lost in the event of a crash. While I initially tested Raft’s core components with an in-memory implementation, the system was designed around an abstraction layer to allow for disk-based persistence.

If I were to build a distributed key-value store on top of this, I’d use my custom-built embedded database as the state machine, paired with a file-backed, append-only log to serve as the durable Raft log.

With these components in place, I’ve addressed the essential elements of my Raft-based key-value store. I believe my implementation isn’t perfect and there should be far more tedious conditions and corner cases that I simply am not aware of. Numerous ways for subtle bugs.

Nonetheless, this was quite an exciting journey as it taught me areas to consider in designing a distributed systems.

Cruise Booking System Redesign

Onion Lee — Fri, 25 Apr 2025 07:30:34 +0000

Motivation

I'm currently serving in the military on a remote island, and I often need to take a cruise home during breaks. Unfortunately, the cruise booking app had some critical issues that caused delays and disrupted my plans.

1. Race Condition: During seat selection and payment, the system allowed multiple users to select the same seat at the same time. The seat would ultimately go to whoever completed payment first—even if someone else had selected it earlier.

2. Slow Response Time: When weather or rough seas triggered mass rebookings, the system's response time dropped drastically. Seat reservation operations would lag or fail completely, often causing payment attempts to be aborted.

Because of these issues, I set out to redesign the booking API with a focus on robust, reliable seat reservation logic.

Assumptions

To keep the project scoped around the core booking logic, I made several simplifying assumptions:

Users are assumed to be authenticated.
All cruise information, including seat maps and address details, is predefined (no admin panel).
Payment processing is simulated, including webhooks and is simplified.

Two-Phase Booking Design

To address the first problem - race condition - I explored the timer-based approach. A countdown timer starts as soon as the user selects a seat. If the user fails to complete the checkout process within the allotted time, the seat is released for other users.

Using a Status Column

I tried to achieve this in the relational database layer using a status column. The process goes as follows:

A user sends a POST request to the /trips/reserve endpoint.
The system checks the status of the seat in the database - if status is available, the system updates it to pending.

However, this had several issues, where the system needs to clean up stale entries periodically with a cron job (e.g., every 1 minute). This caused a stale time window between the periodic sweeping intervals. This was unacceptable as the seat should become reservable immediately when the timeout expires, reflecting the real-time status of the seat.

Using an Expiration Column

A more efficient approach exists using expiration column:

Expiration column stores expiration timestamp
A POST request checks the status and the current time against the expiration column. If the status = pending but the expiration time has passed, the system considers the seat available for booking.

This solution is much elegant with no stale time window, perfectly solving our first problem.

Concurrency Issues

Now, I went to design the seat reservation endpoint. When I first designed the seat reservation endpoint, it seemed straightforward: let users pick a seat and write to the database. However, upon running simple concurrency tests, I discovered a last-writer-wins issue when multiple users simultaneously attempted to reserve the same seat. In practice, all users received success notifications—even though only the last update was actually committed—causing data inconsistencies and confusion.

I built a seat reservation service to assign seats to users with a time-based expiration. The core logic is:

@Transactional
public SeatReservation reserveSeat(Long seatId, Long userId){
    SeatReservation seatReservation = seatReservationRepository.findById(seatId).orElseThrow(() -> new RuntimeException("Seat Not Found"));

    if(seatReservation.getExpirationTime() != null && seatReservation.getExpirationTime().isBefore(LocalDateTime.now())) throw new RuntimeException("Another user has reserved it already");

    seatReservation.setReservationStatus(ReservationStatus.PENDING);
    seatReservation.setReservedByUserId(userId);
    seatReservation.setExpirationTime(LocalDateTime.now().plusMinutes(EXPIRATION_TIME));

    return seatReservation;
}

Sequential Testing

To ensure the service worked as intended, I tested it with 20 users reserving a seat one-by-one:

@Test
void shouldReserveSeatSequentiallyWithOnlyOneSuccess() {
    final int userCount = 20;
    final AtomicInteger successCount = new AtomicInteger();
    final AtomicInteger failureCount = new AtomicInteger();

    for (int i = 0; i < userCount; i++) {
        final long userId = i + 1;
        try {
            seatService.reserveSeat(seatId, userId);
            successCount.incrementAndGet();
        } catch (Exception ex) {
            failureCount.incrementAndGet();
        }
    }

    System.out.println("Success: " + successCount.get());
    System.out.println("Failure: " + failureCount.get());

    assertThat(successCount.get()).isEqualTo(1);
}

The test passed—only one user got the seat, as expected.

Success: 1
Failure: 19

In production, users might hit the endpoint /trips/.../reserve simultaneously for popular seats. I simulated this with 20 concurrent users using threads:

@Test
void shouldReserveSeatConcurrentlyWithOnlyOneSuccess() throws InterruptedException {
    final int userCount = 20;
    final ExecutorService executorService = Executors.newFixedThreadPool(userCount);
    final CountDownLatch countDownLatch = new CountDownLatch(userCount);

    final AtomicInteger successCount = new AtomicInteger();
    final AtomicInteger failureCount = new AtomicInteger();

    for (int i = 0; i < userCount; i++) {
        final long userId = i + 1;
        executorService.submit(() -> {
            try {
                seatService.reserveSeat(seatId, userId);
                successCount.incrementAndGet();
            } catch (Exception ex) {
                failureCount.incrementAndGet();
            } finally {
                countDownLatch.countDown();
            }
        });
    }

    countDownLatch.await();
    executorService.shutdown();

    System.out.println("Success: " + successCount.get());
    System.out.println("Failure: " + failureCount.get());

    assertThat(successCount.get()).isEqualTo(1);
}

The test failed—10 users succeeded instead of 1. I diagnosed this as a "last writer wins" race condition: concurrent threads overwrote each other’s updates before the expiration check kicked in.

org.opentest4j.AssertionFailedError:
expected: 1
but was: 10
Expected: 1
Actual: 10

"Last-Writer-Wins" Scenario

To solve this anomaly, I looked into the transactional isolation level in MySQL and found out the issue. Using MySQL’s REPEATABLE READ isolation, transactions operate on independent snapshots. For example, User2 reads an outdated seat state, sees it as available, and updates it—unaware of User1’s concurrent reservation. This allows multiple successes where only one should occur.

My goal was clear: ensure that the first user to hit /trips/.../reserve reserves the seat if it’s available, while subsequent users fail without overwriting the reservation during the concurrent environment.

To fix this, I devised five different approaches including:

SERIALIZABLE Isolation Level
In-Memory Critical Section
Optimistic Locking (OCC)
Pessimistic Locking
Atomic Database Operation

SERIALIZABLE Isolation Level

This is the most straightforward approach: bumping the isolation level to SERIALIZABLE, which prevents concurrent modifications to the seat during both reads and writes. Specifically, MySQL achieves searizliable isolation level using locks, where reads and writes block each other. Below is the code for changing isolation level using JPA annotation.

@Transactional(isolation = Isolation.SERIALIZABLE)
public SeatReservation reserveSeat(Long seatId, Long userId){

The 20-user concurrency test confirmed it worked:

result:

Success: 1
Failure: 19

However, upon researching, decreased read/write throughput due to blocking read/write and deadlocks concerned me as seat booking must be highly concurrent. So I explored different approaches as well.

In addition to this, I thought of using PostgreSQL instead of MySQL because it supports serializable snapshot isolation via lock-free MVCC. MySQL achieves similar behavior via record locks and gap locks, with reduced concurrency. However, I decided to stick with MySQL to tackle this challenge.

In-Memory Critical Section

As an alternative to tweaking database isolation levels, I explored an in-memory solution. I added the synchronized keyword to the reserveSeat method to limit access to one thread at a time, expecting it to prevent concurrency issues:

@Transactional
public synchronized SeatReservation reserveSeat(Long seatId, Long userId){
    SeatReservation seatReservation = seatReservationRepository.findById(seatId).orElseThrow(() -> new RuntimeException("Seat Not Found"));

    if(seatReservation.getExpirationTime() != null && seatReservation.getExpirationTime().isBefore(LocalDateTime.now())) throw new RuntimeException("Another user has reserved it already");

    seatReservation.setReservationStatus(ReservationStatus.PENDING);
    seatReservation.setReservedByUserId(userId);
    seatReservation.setExpirationTime(LocalDateTime.now().plusMinutes(EXPIRATION_TIME));

    return seatReservation;
}

The test failed unexpectedly:

org.opentest4j.AssertionFailedError:
expected: 1
but was: 7
Expected: 1
Actual: 7

Instead of one success, seven threads succeeded, revealing a flaw in my approach. Digging deeper, I realized the issue stemmed from the interplay between @Transactional and synchronized block.

Spring’s transactional proxy wraps the method, managing transaction begin and commit outside the synchronized block.

The synchronized keyword locks only the method’s execution, not the transaction commit phase. This allowed another thread to enter while the first was committing, leading to multiple successes. A diagram I sketched illustrates the race condition:

Even if this had worked locally, I recognized a major limitation: synchronized is ineffective in a horizontally scaled environment. With multiple servers, in-memory locks don’t coordinate across instances, making this approach unscalable. Therefore, I moved on to try other approaches.

Optimistic locks

Using Hibernate, I added a version column to the SeatReservation entity:

@Version
private Long version;

Hibernate automatically manages versioning. On read, it fetched the row with its version atomically:

Hibernate: select sr1_0.seat_id,sr1_0.expiration_time,sr1_0.payment_intent_id,sr1_0.reservation_status,sr1_0.reserved_by_user_id,sr1_0.transaction_id,sr1_0.trip_id,sr1_0.version from seat_reservation sr1_0 where sr1_0.seat_id=?

On write, it updated only if the version hadn’t changed:

Hibernate: update seat_reservation set expiration_time=?,payment_intent_id=?,reservation_status=?,reserved_by_user_id=?,transaction_id=?,trip_id=?,version=? where seat_id=? and version=?

I ran my concurrency test with 20 users:

test result:
Success: 1
Failure: 19

The test passed—only one user reserved the seat, as Hibernate detected version mismatches and rolled back conflicting updates.

Encouraged by the result, I tested with a smaller scale—such as three concurrent users. However, a conflict occurred where the seat remained unreserved by any of them as all transactions failed simultaneously and no retries occurred. Adding retry logic was an option, but it clashed with the business rule:

“The first user to call the seat reservation endpoint (e.g., /trips/.../reserve) successfully reserves the seat if it is available.”

Retries shuffle the order, meaning the "first" user might not win—success would hinge on retry timing, not request order. This violated the requirement. Therefore, I went on to explore other solutions.

Pessimistic locks

I implemented it using JPA’s @lock annotation and a custom query:

@Lock(LockModeType.PESSIMISTIC_WRITE)
@Query("SELECT sr FROM SeatReservation sr WHERE sr.seatId = :id")
Optional<SeatReservation> findByIdWithPessimisticLock(@Param("id") Long id);

Hibernate automatically appended a FOR UPDATE clause to the SQL query, ensuring the row is locked since the read:

Hibernate: select sr1_0.seat_id,sr1_0.expiration_time,sr1_0.payment_intent_id,sr1_0.reservation_status,sr1_0.reserved_by_user_id,sr1_0.transaction_id,sr1_0.trip_id,sr1_0.version from seat_reservation sr1_0 where sr1_0.seat_id=? for update

Hibernate: update seat_reservation set expiration_time=?,payment_intent_id=?,reservation_status=?,reserved_by_user_id=?,transaction_id=?,trip_id=?,version=? where seat_id=? and version=?

My 20-user concurrency test confirmed success:

result:
Success: 1
Failure: 19

Pessimistic locking looks effective, guaranteeing order and consistency in our current architecture. It is a strong fit for our system unless we scale to a multi-leader architecture, which would require reevaluating our approach.

Atomic Database Operation

Pessimistic locking worked, and quite nicely, but since it locks the row from read until the commit, I sought further optimization.

Upon some thinking, I realized that the SELECT and UPDATE operations could be combined into one single atomic query. This leverages the query’s inherent atomicity, cuts database round-trips, and simplifies the operation.

To implement this, I crafted a JPQL query in SeatReservationRepository and adjusted the service layer:

@Transactional
  public SeatReservation reserveSeat(Long seatId, Long userId) {
      LocalDateTime now = LocalDateTime.now();
      LocalDateTime newExpirationTime = now.plusMinutes(EXPIRATION_TIME);

      // Try to update the seat in one atomic statement.
      int updatedRows = seatReservationRepository.updateSeatReservationAtomically(
              seatId,
              userId,
              ReservationStatus.PENDING,
              newExpirationTime,
              now
      );

      // If no rows were updated, seat was not free (already reserved by someone else).
      if (updatedRows == 0) {
          throw new RuntimeException("Another user has reserved it already or seat not found.");
      }

      // If here, seat was successfully updated.
      // Now we can fetch the seat from DB to return it (and to record the new state).
      SeatReservation seatReservation = seatReservationRepository.findById(seatId)
              .orElseThrow(() -> new RuntimeException("Seat Not Found after update!"));

      return seatReservation;
  }

@Modifying
@Query("UPDATE SeatReservation sr " +
        "SET sr.reservationStatus = :newStatus, " +
        "    sr.reservedByUserId = :userId, " +
        "    sr.expirationTime = :expirationTime " +
        "WHERE sr.seatId = :seatId " +
        "  AND (sr.expirationTime IS NULL OR sr.expirationTime < :currentTime)")
int updateSeatReservationAtomically(@Param("seatId") Long seatId,
                                    @Param("userId") Long userId,
                                    @Param("newStatus") ReservationStatus newStatus,
                                    @Param("expirationTime") LocalDateTime expirationTime,
                                    @Param("currentTime") LocalDateTime currentTime);

The 20-user concurrency test succeeded and only one user successfully reserved the seat, as the atomic update prevented concurrent modifications.

This approach shines for our simple case, reducing overhead. However, if future logic spans multiple tables or adds complex checks between read and write, a single query might not scale. For now, I deemed it aa efficient and meets our needs.

Tests

To see the average execution time of each approach, I also measured execution time locally, focusing on relative performance between the approaches:

@Test
void shouldReserveSeatConcurrentlyWithOnlyOneSuccess() throws InterruptedException {
        ...
    // Start timing
    long startTime = System.nanoTime();
    for (int i = 0; i < userCount; i++) {
            ...
    }
    // Stop timing
    long endTime = System.nanoTime();
    long elapsedNanos = endTime - startTime;
    long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(elapsedNanos);

    System.out.println("Elapsed Time: " + elapsedMillis + " ms");
        ...
}

Result:

Serializable Isolation Level: Elapsed Time: 450 ms
Optimistic locks: Elapsed Time: 83 ms
Pessimistic locks: Elapsed Time: 130 ms
Atomic Query: 105 ms

Serializable isolation level was the slowest, as expected, while optimistic locks were fast but impractical for high contention scenario. Pessimistic locks worked but added latency. The atomic query struck a balance—fast, scalable, and reliable.

Webhook Issues

Payment gateways, including Stripe, operate on an at-least-once delivery model for webhook events. This means that if our application fails to acknowledge a webhook request with an HTTP 200 OK response (due to high traffic, server issues, etc.), the payment gateway will retry the webhook delivery as shown below:

At first, I thought, "Wouldn't it be fine if I just checked the latest payment status?" But then I realized:

Retries could create duplicate processing if not handled properly.
Repeated webhook events for the same transaction may overwrite the original timestamp in the SeatReservation table.

To prevent double-processing, I designed the system to handle webhooks idempotently. That is, every time a webhook is received, the system checks:

If the payment_intent_id (or equivalent unique transaction reference) already exists in the database and has been marked as PAID.
If it does, the system simply acknowledges the webhook without making any changes and sends an HTTP 200 OK response back to the gateway.
If it doesn’t, it updates the reservation status accordingly and records the change.

This way, no matter how many times Stripe sends a webhook event, the same transaction is never processed twice.

Proposed Solution

I have adjusted my database schema to ensure smooth handling of these scenarios.

When a webhook request arrives with a given payment_intent_id, the system checks the SeatReservation row to see if that payment_intent_id is already present.
If it does not exist, the system updates the seat reservation to reflect the new payment status (e.g., PAID), records a timestamp (if needed), and logs the update.
If a duplicate webhook is detected (i.e., the payment_intent_id is already present), the system preserves the original timestamp and record by simply returning an HTTP 200 response, thus preventing unnecessary retries.

Another Webhook Issues

As I fleshed out the payment processing logic, another tricky edge case caught my attention: delayed webhook deliveries. What happens if a payment gateway like Stripe processes a payment successfully, but the webhook arrives after the seat reservation’s timer expires? I investigated this to ensure reservation validity remained intact.

After some reflection, I mapped out the fallout:

Payment Succeeds, Reservation Expires: The user pays, but the webhook lags, arriving post-timeout. The seat’s reservation is released, despite a valid payment.
System Misalignment: Since the expired reservation is no longer linked to the payment, the system cannot assign the seat.
Cleanup Needed: This leaves a paid-but-unreserved seat, likely requiring a refund despite a successful transaction.

If a delayed webhook arrives for a reservation that has expired, the system must issue a refund in the system, while also making sure no seat is reserved.

Now, to sum up, I have designed the below logic to incorporate both the idempotency in payment processing and the refund logic.

Verify Webhook Request:
- Validate using secret keys/signatures.
Acquire a Lock / Begin Transaction:
- Check if the seat exists.
- Lock the reservation record (or use an optimistic locking strategy).
Perform an Idempotence Check:
Check UserId and Reservation Expiry:
- Check to ensure the user matches the reserved user
- Check for expired reservation
- If Expired:
  - Log the refund action so it can later be aggregated for the refund processing.
- If Not Expired:
  - Process the webhook payload:
    - SUCCESS: Update reservationStatus to PAID.
    - FAILED/CANCELLED: write a structured log about the message.
  - Ensure updates and history logging occur in one transaction.
Release the Lock / Commit Transaction:
- Ensure all changes are committed atomically.
Return an HTTP 200 Response:
- Acknowledge receipt of the webhook to prevent further retries.

Conclusion

In designing a more robust seat reservation API, key issues from my initial experience—race conditions and slow response times—were addressed to improve the booking experience.

To resolve the race condition issue, I introduced a two-phase booking process. This approach allowed a user to acquire a timed lock on a row when selecting a seat, preventing scenarios where a faster payment response inadvertently claims the seat.

Improving response times was challenging due to limited access to Ticket30's codebase. However, through concurrency testing, I was about to somewhat approximate the potential bottleneck and address them. Multiple approaches were considered, and atomic database operation was deemed to be most suitable yet efficient. This solution significantly improved performance while avoiding locking overhead.

During the implementation stage, additional edge cases, such as duplicate webhook events and delayed payment confirmations, were identified. These were carefully addressed to ensure system reliability and resilience under peak load.

By resolving these challenges, I believe the new booking system ensures a more efficient, scalable, and user-friendly experience.