Designing a Distributed Counter: Sharded Writes, Approximate Reads, and Consistency Trade-offs at Scale

Counting things at scale is one of those problems that looks trivial until it isn’t. A like counter on a viral post, a view counter on a video that just got shared by someone with 10 million followers, an inventory decrement for a flash sale, an analytics event pipeline tracking 50,000 events per second: all of these converge on the same fundamental problem. Incrementing a single number, reliably, fast, from many concurrent writers.

The naive solution works until it doesn’t.

Why a Single Atomic Counter Breaks

A single INCR command in Redis is atomic. For low-traffic counters that is fine. But under sustained high write volume, you hit three walls fast.

Throughput ceiling. A single Redis node handles around 100,000 operations per second under ideal conditions. A viral video receiving 500,000 concurrent viewers, each triggering a view event, blows past that ceiling immediately.

Replication lag. If you replicate that counter across regions, the primary becomes a write bottleneck. Followers lag. A counter that reads 4,200,000 in us-east-1 might read 4,180,000 in eu-west-1 a second later.

Lock contention in relational databases. If you store the counter in Postgres with a row-level lock, every increment serializes. At 10,000 writes per second you will see lock wait timeouts and connection pool exhaustion before you see accurate counts.

The problem compounds when you add the read side. If a user just liked a post, they expect to see their like reflected immediately. But with eventual consistency across shards and regions, “immediately” is ambiguous.

There is no single correct architecture. The right approach depends on whether you need exact counts or approximate, how stale reads can be, and what you are willing to pay in write throughput and operational complexity.

Sharded Write Paths with Rollup Aggregation

The most common production approach for high-throughput exact counting is write sharding with periodic rollup.

The idea: instead of one counter key, maintain N shard keys. Each write goes to one shard, selected by a hash or randomly. A background job periodically sums all shards into an authoritative total.

// Counter shard selection
const COUNTER_SHARDS = 64;

function getShardKey(counterId: string): string {
  const shardIndex = Math.floor(Math.random() * COUNTER_SHARDS);
  return `counter:${counterId}:shard:${shardIndex}`;
}

async function incrementCounter(
  redis: Redis,
  counterId: string
): Promise<void> {
  const shardKey = getShardKey(counterId);
  await redis.incr(shardKey);
}

// Rollup job: sum all shards into a single authoritative value
async function rollupCounter(
  redis: Redis,
  counterId: string
): Promise<number> {
  const shardKeys = Array.from(
    { length: COUNTER_SHARDS },
    (_, i) => `counter:${counterId}:shard:${i}`
  );

  const pipeline = redis.pipeline();
  shardKeys.forEach((key) => pipeline.get(key));
  const results = await pipeline.exec();

  const total = results.reduce((sum, [err, val]) => {
    if (err) return sum;
    return sum + (parseInt(val as string, 10) || 0);
  }, 0);

  await redis.set(`counter:${counterId}:total`, total);
  return total;
}

The read path queries counter:${counterId}:total, which is updated on whatever rollup schedule fits your staleness budget. A news feed that shows “1.2M views” can tolerate 60-second-old data. An inventory counter for flash-sale stock cannot.

The tradeoff is that reads are always slightly stale by design. The rollup frequency determines the staleness window. Aggressive rollup (every second) increases read accuracy but adds background job pressure. Running rollup every 30 seconds is usually a reasonable default for social counters.

One important implementation detail: rollup writes should use SET with the computed total, not further INCR operations on the total key. Mixing increment operations on the total key with rollup writes creates race conditions that can undercount.

CRDT-Based Commutative Counters

CRDTs (Conflict-free Replicated Data Types) take a different approach. Rather than coordinating writes through a single shard owner or serialized rollup, they let any node accept writes independently and merge state later without conflicts.

G-Counter (Grow-Only Counter): Each node maintains its own counter. The global count is the sum of all per-node values. Merging two replicas takes the max per node, so no coordination is needed.

interface GCounter {
  nodeId: string;
  counts: Map<string, number>;
}

function increment(counter: GCounter): GCounter {
  const current = counter.counts.get(counter.nodeId) ?? 0;
  return {
    ...counter,
    counts: new Map(counter.counts).set(counter.nodeId, current + 1),
  };
}

function merge(a: GCounter, b: GCounter): GCounter {
  const merged = new Map<string, number>();
  const allNodes = new Set([...a.counts.keys(), ...b.counts.keys()]);

  for (const nodeId of allNodes) {
    const aVal = a.counts.get(nodeId) ?? 0;
    const bVal = b.counts.get(nodeId) ?? 0;
    merged.set(nodeId, Math.max(aVal, bVal));
  }

  return { nodeId: a.nodeId, counts: merged };
}

function value(counter: GCounter): number {
  let total = 0;
  for (const v of counter.counts.values()) {
    total += v;
  }
  return total;
}

PN-Counter (Positive-Negative Counter): For counters that need decrements (inventory, net votes), you maintain two G-Counters: one for increments, one for decrements. The net value is value(P) - value(N).

interface PNCounter {
  nodeId: string;
  positive: GCounter;
  negative: GCounter;
}

function pnIncrement(counter: PNCounter): PNCounter {
  return { ...counter, positive: increment(counter.positive) };
}

function pnDecrement(counter: PNCounter): PNCounter {
  return { ...counter, negative: increment(counter.negative) };
}

function pnValue(counter: PNCounter): number {
  return value(counter.positive) - value(counter.negative);
}

function pnMerge(a: PNCounter, b: PNCounter): PNCounter {
  return {
    nodeId: a.nodeId,
    positive: merge(a.positive, b.positive),
    negative: merge(a.negative, b.negative),
  };
}

The advantage of CRDTs is that they are designed for multi-region active-active setups. Any datacenter can accept writes. Merges are deterministic and commutative: the order does not matter and there are no conflicts to resolve.

The cost is storage proportional to the number of nodes. A G-Counter with 50 nodes stores 50 integers per counter key. At millions of distinct counter IDs, this adds up. CRDT-based counters also require a gossip or sync mechanism to propagate state between nodes, which adds network and serialization overhead that pure Redis INCR does not have.

Approximate Counting: Probabilistic Structures

For analytics workloads where exact counts are not required, probabilistic data structures offer dramatically better throughput and lower memory at the cost of a small, bounded error.

HyperLogLog for cardinality estimation. If your question is “how many unique users viewed this page?” rather than “how many total views?”, HyperLogLog gives you an estimate with a standard error of 0.81% using at most 12KB of memory regardless of input cardinality.

Redis exposes HyperLogLog natively via PFADD and PFCOUNT. The operational model is simple: add user identifiers as they arrive, query the estimated distinct count. No deduplication logic, no secondary storage for seen-user tracking.

async function recordUniqueView(
  redis: Redis,
  pageId: string,
  userId: string
): Promise<void> {
  await redis.pfadd(`hll:views:${pageId}`, userId);
}

async function getUniqueViewCount(
  redis: Redis,
  pageId: string
): Promise<number> {
  return redis.pfcount(`hll:views:${pageId}`);
}

Count-Min Sketch for frequency estimation. HyperLogLog answers “how many distinct?” but not “how many times did this specific item appear?”. Count-Min Sketch answers the frequency query with a fixed error bound.

The structure is a 2D array of d hash functions by w counters. Each increment hashes the item with each hash function and increments the corresponding cell. To query a frequency, you take the minimum across all d hash rows, which bounds the overcounting introduced by hash collisions.

class CountMinSketch {
  private table: number[][];
  private readonly width: number;
  private readonly depth: number;

  constructor(width = 1000, depth = 5) {
    this.width = width;
    this.depth = depth;
    this.table = Array.from({ length: depth }, () =>
      new Array<number>(width).fill(0)
    );
  }

  private hash(item: string, seed: number): number {
    let h = seed * 31;
    for (let i = 0; i < item.length; i++) {
      h = (Math.imul(h, 31) + item.charCodeAt(i)) >>> 0;
    }
    return h % this.width;
  }

  increment(item: string): void {
    for (let i = 0; i < this.depth; i++) {
      const col = this.hash(item, i + 1);
      this.table[i][col]++;
    }
  }

  estimate(item: string): number {
    let min = Infinity;
    for (let i = 0; i < this.depth; i++) {
      const col = this.hash(item, i + 1);
      min = Math.min(min, this.table[i][col]);
    }
    return min;
  }
}

Width and depth parameters control the accuracy-memory tradeoff. Larger width reduces collision probability. More depth rows reduce the chance that all rows collide simultaneously. A depth of 5 and width of 1000 gives roughly 1% error with 40KB of memory for 32-bit integers.

Count-Min Sketch is particularly useful for leaderboards (“top trending items in the last hour”) where you want approximate frequencies across millions of candidate keys without storing exact counts for each one.

Read-Your-Writes Consistency

The hardest consistency challenge in distributed counters is the user who just performed an action. If a user likes a post and immediately reloads their feed, they expect to see their like reflected. If the read returns stale data showing their like was not counted, that is a trust-eroding bug even if the underlying count is correct by the time anyone else sees it.

The standard solution is to combine a session token with a read-path bypass for recent writes.

interface CounterSession {
  userId: string;
  localAdjustment: Map<string, number>;
  expiresAt: number;
}

async function getCounterWithReadYourWrites(
  redis: Redis,
  counterId: string,
  session: CounterSession
): Promise<number> {
  const stored = await redis.get(`counter:${counterId}:total`);
  const base = parseInt(stored ?? "0", 10);

  // Apply the local adjustment from this session
  const localDelta = session.localAdjustment.get(counterId) ?? 0;
  if (Date.now() > session.expiresAt) {
    return base;
  }
  return base + localDelta;
}

async function incrementWithSession(
  redis: Redis,
  counterId: string,
  session: CounterSession
): Promise<void> {
  await incrementCounter(redis, counterId);

  // Track the delta locally so the user sees their write immediately
  const current = session.localAdjustment.get(counterId) ?? 0;
  session.localAdjustment.set(counterId, current + 1);
  session.expiresAt = Date.now() + 30_000; // 30-second read-your-writes window
}

This approach is simple and does not require any cross-service coordination. The 30-second window covers rollup lag in typical deployments. After the window expires, the session adjustment is dropped and reads fall through to the rolled-up total, which by then includes the write.

Hot-Key Mitigation with Jittered Shard Selection

Deterministic shard selection (hashing the counter ID to a fixed shard) defeats the purpose of sharding for popular counters. If counter:post:12345678 always maps to shard 7, shard 7 becomes the hot node.

Random shard selection spreads write load uniformly but makes rollup slightly more expensive (you always read all shards). A middle ground is jittered shard selection: use a time-bucketed seed that changes every few seconds to spread writes across shards without pure randomness.

function getJitteredShardKey(
  counterId: string,
  shardCount: number,
  jitterWindowMs = 2000
): string {
  // Quantize current time into buckets, then add randomness within the bucket
  const timeBucket = Math.floor(Date.now() / jitterWindowMs);
  const jitter = Math.floor(Math.random() * Math.ceil(shardCount / 4));
  const shardIndex = (timeBucket + jitter) % shardCount;
  return `counter:${counterId}:shard:${shardIndex}`;
}

The jitter window ensures that writes in the same time window are distributed across roughly a quarter of the shard space, preventing any single shard from absorbing all traffic while keeping the number of hot shards predictable.

For extreme hot keys (a global trending counter hit by millions of concurrent users), you can also introduce a local write buffer at the application tier. Batch 100 increments client-side and flush as a single INCRBY to the shard, reducing Redis commands by two orders of magnitude.

Tradeoffs Table

Approach	Accuracy	Write Throughput	Read Latency	Storage Cost	Consistency Model
Single atomic counter	Exact	Low (single key bottleneck)	Fast (single key read)	Minimal	Strong (serialized)
Sharded with rollup	Exact	High (N keys in parallel)	Fast (read pre-rolled total)	Low-moderate	Eventual (rollup lag)
CRDT G-Counter / PN-Counter	Exact	Very high (no coordination)	Moderate (sum all nodes)	High (per-node vectors)	Eventual (gossip merge)
HyperLogLog	Approximate (~0.81% error)	High	Fast	Very low (12KB fixed)	Eventual
Count-Min Sketch	Approximate (bounded error)	High	Fast	Low (configurable)	Eventual

The right choice is rarely one approach across the entire system. A social platform might use sharded-with-rollup for like counts (exact, eventually consistent), HyperLogLog for unique viewer counts (approximate, high cardinality), and a single atomic counter with Redis for inventory decrements where low write volume and exact semantics are both required.

Production Considerations

Rollup frequency and cache coherence. The rollup job is your primary lever for accuracy. Running it every second gives near-real-time totals but increases background job load. Running it every 60 seconds means reads can lag by up to a minute. The right frequency is the maximum staleness your product can tolerate, not the minimum technically achievable. Instrument the lag metric explicitly: measure the difference between the rolled-up total and the live shard sum on every rollup cycle.

Counter overflow and wraparound. Redis INCR uses 64-bit signed integers, giving a maximum value of 9,223,372,036,854,775,807. For most counters this is not a practical concern, but for high-frequency analytics event counters that shard to many keys you can hit per-shard integer limits faster than expected. Monitor shard values and alert before they approach INT64_MAX. For probabilistic structures implemented outside Redis, use 64-bit unsigned integers and add overflow detection in your increment implementation.

Sharding rebalance during scaling events. When you double your shard count from 64 to 128, in-flight writes go to the new keyspace while rollup jobs may still be reading the old keyspace. Coordinate this transition explicitly: run both shard counts simultaneously during a migration window, rolling up from both old and new shard keys until all historical writes have drained from the old shards. A phased cutover with a configurable shard count in a feature flag is safer than a hard cutover.

Decrement correctness with PN-Counters. The PN-Counter model can produce negative values if your decrement node is ahead of your increment node in the merge graph. For inventory counters where negative stock is invalid, add a floor check at the application layer before accepting a decrement, rather than relying on the data structure to enforce it. The CRDT guarantees merge correctness, not business rule invariants.

TTL and counter expiry. Shard keys for ephemeral counters (daily active counts, per-request rate limits) should carry TTLs. A background rollup job that iterates over all counter keys can accumulate unbounded memory if old keys are not expired. Use EXPIREAT set to the end of the counter’s relevant window when writing shard keys for time-bounded counters.

Observability. Track rollup duration, shard skew (the variance across shard values for the same counter as a hot-key indicator), and rollup lag as dedicated metrics. When a single shard is receiving 80% of the writes for a given counter, you have a hot-key problem regardless of how the architecture is designed on paper.

Closing

Distributed counting is not a solved problem with a universal answer. A single atomic operation is correct and simple until throughput makes it impractical. Sharding with rollup handles most production cases but requires careful migration planning at scale. CRDTs eliminate coordination at the cost of storage and operational complexity. Probabilistic structures are the right tool when exact numbers are less important than cardinality or frequency rank. The production insight is that most systems need two or three of these approaches running side by side, each handling a different counting semantic, not a single unified counter abstraction.