Redis Streams and the Unified Log

Redis Streams as a Foundation for the Unified Log

The "unified log" pattern — popularized by an influential LinkedIn engineering article — describes a simple but powerful idea: services write state changes to a shared, ordered log, and other services read from that log to stay in sync. Each record gets a unique ID. Consumers track the ID of the last record they processed, so they can resume from the right position after a crash. Because a log is durable and readable any number of times, it serves as a central ledger for all state changes across a distributed system.

This isn't a new concept. Journaling file systems use logs to guard against corruption, and Postgres uses its write-ahead log (WAL) to stream changes to replicas. Yet applying the same structure to service-to-service communication remains a compelling design choice. Chatty, synchronous message passing between services is slow, inefficient, and fragile. The log, by contrast, is asynchronous, batch-oriented, and resilient by construction.

Kafka is the most common realization of this pattern, but its operational overhead can be prohibitive. Redis' new streams data structure offers a much lighter path to the same architecture. Unlike a list, where positions are relative offsets, each record in a stream gets an addressable ID, which makes it easy for multiple consumers to track their own position by persisting the last ID they read. Commands like XADD append to a stream:

> XADD rocket-rides-log * id 123 distance 456.7
1506871964177.0

This appends a record with id = 123 and distance = 456.7 to the stream rocket-rides-log. Redis responds with a unique ID composed of the millisecond timestamp plus a sequence number (.0) to disambiguate entries within the same millisecond.

To read, use XRANGE:

> XRANGE rocket-rides-log - + COUNT 2
1) 1) 1506871964177.0
   2) 1) "id"
      2) "123"
      3) "distance"
      4) "456.7"
2) 1) 1506872463535.0
   2) 1) "id"
      2) "124"
      3) "distance"
      4) "89.0"

The tokens - and + are shorthand for the first and last records in the stream. Replace either with a literal ID such as 1506871964177.0 to start from or end at a specific point — precisely what a consumer needs to fetch only new records. Adding COUNT 2 bounds the size of each read, keeping processing efficient in batches.

Redis vs. Kafka for a Unified Log

For a full-featured deployment under substantial load, Kafka's sophisticated high-availability design makes it the stronger candidate. But Kafka is notoriously hard to configure and expensive to operate. Even a small hosted Kafka cluster starts at around $100 per month on Heroku, and running it in-house carries hidden server and personnel costs. For many projects, Redis is more practical: it is ubiquitous, easy to compile from source or acquire as a managed service, and supports countless other use cases — from caching to rate limiting to session storage.

Making Redis Durable

A unified log must survive host crashes without losing acknowledged writes. Redis defaults are not durable, which is acceptable for caching but not for the log role. The fix is the append-only file (AOF) mode: with appendonly enabled and appendfsync always set, Redis performs an fsync on every write to the AOF:

appendonly yes
appendfsync always

This is a performance tradeoff. More fsyncs means slower commands, though still fast in absolute terms. If Redis serves other, ephemeral roles in the same application, consider running two instances: one with AOF always-on for the log, another with default settings for cache-like data that can tolerate loss.

Designing for redelivery

In the Rocket Rides sample app, the API emits an event to a Redis stream whenever a new ride is booked. Each stream record carries the ride’s id and distance traveled; consumer processes read that stream and maintain running totals of distance per ride. All writes are wrapped in database transactions so that totals stay consistent across consumers even when components crash at arbitrary points.

The example works because producer and consumers are built around at-least once delivery. Under normal operation each message is handed off once, but any time the sender cannot confirm the delivery, it re-sends. That is the opposite of best-effort delivery, which drops messages under load, and different from exactly-once semantics, which would require coordination overhead that is rarely worth the cost. At-least once is practical as long as the rest of the system tolerates duplicates.

git clone https://github.com/brandur/rocket-rides-unified.git

Staging records before the log

The API does not write directly to Redis. Instead it inserts a row into a staged_log_records table inside the same Postgres transaction that creates the ride. Emitting only after commit prevents the stream from ever carrying data from an aborted transaction, the same pattern used for transactionally staged job drains.

CREATE TABLE staged_log_records (
    id     BIGSERIAL PRIMARY KEY,
    action TEXT      NOT NULL,
    data   JSONB     NOT NULL,
    object TEXT      NOT NULL
);

A separate streamer process polls that table, pulls staged rows in batches, and appends them to the Redis stream. It does not delete a staged record until Redis acknowledges receipt of the batch; on failure the entire batch is re-selected and re-sent, which is how duplicates enter the log. The streamer emits records ordered by ascending ride id, so the only time a consumer sees an out-of-order id is when a record was delivered twice.

def run_once
  num_streamed = 0

  # Need at least repeatable read isolation level so that our DELETE after
  # enqueueing will see the same records as the original SELECT.
  DB.transaction(isolation_level: :repeatable_read) do
    records = StagedLogRecord.order(:id).limit(BATCH_SIZE)

    unless records.empty?
      RDB.multi do
        records.each do |record|
          stream(record.data)
          num_streamed += 1

          $stdout.puts "Enqueued record: #{record.action} #{record.object}"
        end
      end

      StagedLogRecord.where(Sequel.lit("id <= ?", records.last.id)).delete
    end
  end

  num_streamed
end

#
# private
#

# Number of records to try to stream on each batch.
BATCH_SIZE = 1000
private_constant :BATCH_SIZE

private def stream(data)
  # XADD mystream MAXLEN ~ 10000  * data <JSON-encoded blob>
  #
  # MAXLEN ~ 10000 caps the stream at roughly that number (the "~" trades
  # precision for speed) so that it doesn't grow in a purely unbounded way.
  RDB.xadd(STREAM_NAME,
    "MAXLEN", "~", STREAM_MAXLEN,
    "*", "data", JSON.generate(data))
end

Logs grow without bound unless trimmed. The streamer passes MAXLEN to XADD with the ~ operator, telling Redis to truncate the stream to approximately that length by removing whole nodes rather than exact counts. That is fast but not safe for slow consumers: nothing prevents a lagging reader from losing records it has not fetched yet. A production deployment would track per-consumer progress and only trim records no remaining consumer needs.

Checkpointing inside a transaction

Consumers read from the stream in batches and process records one at a time. After a batch is handled successfully, the consumer records a checkpoint—the stream ID of the last record read—in Postgres. On restart the consumer resumes from that checkpoint rather than from the beginning of the log.

CREATE TABLE checkpoints (
    id            BIGSERIAL PRIMARY KEY,
    name          TEXT      NOT NULL UNIQUE,
    last_redis_id TEXT      NOT NULL,
    last_ride_id  BIGINT    NOT NULL
);

For the distance-tallying example, each consumer’s running total lives in a consumer_states row alongside its checkpoint.

CREATE TABLE consumer_states (
    id             BIGSERIAL        PRIMARY KEY,
    name           TEXT             NOT NULL UNIQUE,
    total_distance DOUBLE PRECISION NOT NULL
);

Updating total and checkpoint together inside one transaction is what makes crash recovery clean.

def run_once
  num_consumed = 0

  DB.transaction do
    checkpoint = Checkpoint.first(name: name)

    # "-" is a special symbol in Redis streams that dictates that we should
    # start from the earliest record in the stream. If we don't already have
    # a checkpoint, we start with that.
    start_id = "-"
    start_id = self.class.increment(checkpoint.last_redis_id) unless checkpoint.nil?

    checkpoint = Checkpoint.new(name: name, last_ride_id: 0) if checkpoint.nil?

    records = RDB.xrange(STREAM_NAME, start_id, "+", "COUNT", BATCH_SIZE)
    unless records.empty?
      # get or create a new state for this consumer
      state = ConsumerState.first(name: name)
      state = ConsumerState.new(name: name, total_distance: 0.0) if state.nil?

      records.each do |record|
        redis_id, fields = record

        # ["data", "{\"id\":123}"] -> {"data"=>"{\"id\":123}"}
        fields = Hash[*fields]

        data = JSON.parse(fields["data"])

        # if the ride's ID is lower or equal to one that we know we consumed,
        # skip it; this is a double send
        if data["id"] <= checkpoint.last_ride_id
          $stdout.puts "Skipped record: #{fields["data"]} " \
            "(already consumed this ride ID)"
          next
        end

        state.total_distance += data["distance"]

        $stdout.puts "Consumed record: #{fields["data"]} " \
          "total_distance=#{state.total_distance.round(1)}m"
        num_consumed += 1

        checkpoint.last_redis_id = redis_id
        checkpoint.last_ride_id = data["id"]
      end

      # now that all records for this round are consumed, persist state
      state.save

      # and persist the changes to the checkpoint
      checkpoint.save
    end
  end

  num_consumed
end

When an exception aborts that transaction, neither the distance total nor the checkpoint changes, so after a restart the consumer simply re-reads the same batch and applies it again with no ill effect. Because stream IDs are monotonic, the consumer can also skip any record whose ride id is less than or equal to the highest one it has already processed—a cheap defense against the streamer’s occasional double-send.

Fault injection

The example code deliberately breaks itself to prove the design works. The streamer double-sends every event in a batch about 10% of the time, simulating a mid-batch failure and retry. Each consumer likewise crashes after handling a batch but before committing, also about 10% of the time. Despite those injected faults, running the fleet with forego start shows the consumers staying roughly caught up and agreeing on total_distance for any given ride ID.

consumer0.1 | Consumed record: {"id":521,"distance":539.836923415231}
              total_distance=257721.7m
consumer1.1 | Consumed record: {"id":521,"distance":539.836923415231}
              total_distance=257721.7m

Comparing to alternatives

Transactional consumers are one option, not a requirement. A consumer whose work is fully idempotent can skip the transaction entirely. For example, a process that upserts warehouse rows instead of inserting, and tolerates deleting a missing row, can safely re-apply any batch. The distance tally in the example is not idempotent—reapplying a batch without checkpoints would double-count—so it needs the transaction to coordinate both the state change and the checkpoint.

The streamer plus staged-records pattern resembles Postgres 10 logical replication, but a Redis-backed log has practical advantages:

• Multiple producers can write to one stream without sharing a database.
• Producers can emit a public schema decoupled from their internal storage, so schema changes do not force consumer updates.
• Operators avoid the specialized knowledge needed to run replication slots and subscriptions correctly.

These guarantees are strong but not absolute. A bug that deletes staging code, for instance, would let rides fall out of the log unnoticed until a fix is deployed—by then the consumers’ totals would disagree with the canonical ride data. Systems that demand exact consistency need a periodic reconciliation job, such as a nightly sweep that recomputes total distance from all known rides and resets each consumer’s state to that snapshot before resuming stream consumption.