Write-Behind View Counter with Safe Flushes

Video views arrive far too fast to write each one to the database, so they are buffered and flushed in batches (write-behind). Flushes can crash after claiming a batch, and a commit can be retried after a timeout. Implement the flushing protocol so no view is lost and none is counted twice.

ViewBuffer():

record(video, n) — add n views to the buffer.
flush() — if a claimed batch is still pending (its commit hasn't succeeded), return its id again without moving new data into it. Otherwise, if the buffer is empty return ""; else move the whole buffer into a new pending batch with id "b1", "b2", … and return the id.
commit(batchId) — the database transaction for a batch. If batchId is the pending batch, add its counts to the database totals, mark it committed, clear the pending batch and return true. Otherwise (already committed, or unknown) return false — committing twice must not double-count.
views(video) — the committed database total for the video (0 if none).

See Write-Through and Write-Behind.

Example 1

Input: ["ViewBuffer","record","record","flush","record","flush","commit","views","flush","commit","commit","views","views"]
[[],["v1",3],["v2",1],[],["v1",2],[],["b1"],["v1"],[],["b2"],["b2"],["v1"],["v2"]]
Output: [null,null,null,"b1",null,"b1",true,3,"b2",true,false,5,1]
Explanation: The second flush returns the still-pending b1; views recorded meanwhile go into b2. The retried commit of b2 is ignored.

Solution and explanation

Show the solution

Time O(1) per record, O(videos in batch) per commit · Space O(videos + batches)

Two rules make write-behind safe: claim, don't remove — a batch stays pending (re-flushable) until its database commit succeeds, so a crash between claim and commit loses nothing; and idempotent commits — remembering committed batch ids (in the database, in the same transaction) turns a retried commit into a no-op. Redis Streams provide the first rule (pending entries until XACK); a processed_batches table with a unique key provides the second.

Reference solution

class ViewBuffer {
  constructor() {
    this.buffer = new Map();
    this.pending = null;          // { id, counts: Map }
    this.committed = new Set();
    this.db = new Map();
    this.nextId = 1;
  }

  record(video, n) {
    this.buffer.set(video, (this.buffer.get(video) ?? 0) + n);
  }

  flush() {
    if (this.pending) return this.pending.id;
    if (this.buffer.size === 0) return "";
    this.pending = { id: `b${this.nextId++}`, counts: this.buffer };
    this.buffer = new Map();
    return this.pending.id;
  }

  commit(batchId) {
    if (!this.pending || this.pending.id !== batchId || this.committed.has(batchId)) return false;
    for (const [video, n] of this.pending.counts) this.db.set(video, (this.db.get(video) ?? 0) + n);
    this.committed.add(batchId);
    this.pending = null;
    return true;
  }

  views(video) {
    return this.db.get(video) ?? 0;
  }
}

class ViewBuffer {
  private buffer = new Map<string, number>();
  private pending: { id: string; counts: Map<string, number> } | null = null;
  private committed = new Set<string>();
  private db = new Map<string, number>();
  private nextId = 1;

  record(video: string, n: number): void {
    this.buffer.set(video, (this.buffer.get(video) ?? 0) + n);
  }

  flush(): string {
    if (this.pending) return this.pending.id;
    if (this.buffer.size === 0) return "";
    this.pending = { id: `b${this.nextId++}`, counts: this.buffer };
    this.buffer = new Map();
    return this.pending.id;
  }

  commit(batchId: string): boolean {
    if (!this.pending || this.pending.id !== batchId || this.committed.has(batchId)) return false;
    for (const [video, n] of this.pending.counts) this.db.set(video, (this.db.get(video) ?? 0) + n);
    this.committed.add(batchId);
    this.pending = null;
    return true;
  }

  views(video: string): number {
    return this.db.get(video) ?? 0;
  }
}

class ViewBuffer:
    def __init__(self):
        self.buffer: dict[str, int] = {}
        self.pending: tuple[str, dict[str, int]] | None = None
        self.committed: set[str] = set()
        self.db: dict[str, int] = {}
        self.next_id = 1

    def record(self, video: str, n: int) -> None:
        self.buffer[video] = self.buffer.get(video, 0) + n

    def flush(self) -> str:
        if self.pending:
            return self.pending[0]
        if not self.buffer:
            return ""
        self.pending = (f"b{self.next_id}", self.buffer)
        self.next_id += 1
        self.buffer = {}
        return self.pending[0]

    def commit(self, batch_id: str) -> bool:
        if not self.pending or self.pending[0] != batch_id or batch_id in self.committed:
            return False
        for video, n in self.pending[1].items():
            self.db[video] = self.db.get(video, 0) + n
        self.committed.add(batch_id)
        self.pending = None
        return True

    def views(self, video: str) -> int:
        return self.db.get(video, 0)

<?php

class ViewBuffer
{
    private array $buffer = [];
    private ?array $pending = null;   // ['id' => string, 'counts' => array]
    private array $committed = [];
    private array $db = [];
    private int $nextId = 1;

    public function record(string $video, int $n): void
    {
        $this->buffer[$video] = ($this->buffer[$video] ?? 0) + $n;
    }

    public function flush(): string
    {
        if ($this->pending !== null) {
            return $this->pending['id'];
        }
        if (!$this->buffer) {
            return '';
        }
        $this->pending = ['id' => 'b' . $this->nextId++, 'counts' => $this->buffer];
        $this->buffer = [];
        return $this->pending['id'];
    }

    public function commit(string $batchId): bool
    {
        if ($this->pending === null || $this->pending['id'] !== $batchId || isset($this->committed[$batchId])) {
            return false;
        }
        foreach ($this->pending['counts'] as $video => $n) {
            $this->db[$video] = ($this->db[$video] ?? 0) + $n;
        }
        $this->committed[$batchId] = true;
        $this->pending = null;
        return true;
    }

    public function views(string $video): int
    {
        return $this->db[$video] ?? 0;
    }
}

class ViewBuffer {
    private Map<String, Integer> buffer = new HashMap<>();
    private String pendingId = null;
    private Map<String, Integer> pendingCounts = null;
    private final Set<String> committed = new HashSet<>();
    private final Map<String, Integer> db = new HashMap<>();
    private int nextId = 1;

    public ViewBuffer() {}

    public void record(String video, int n) {
        buffer.merge(video, n, Integer::sum);
    }

    public String flush() {
        if (pendingId != null) return pendingId;
        if (buffer.isEmpty()) return "";
        pendingId = "b" + nextId++;
        pendingCounts = buffer;
        buffer = new HashMap<>();
        return pendingId;
    }

    public boolean commit(String batchId) {
        if (pendingId == null || !pendingId.equals(batchId) || committed.contains(batchId)) return false;
        pendingCounts.forEach((video, n) -> db.merge(video, n, Integer::sum));
        committed.add(batchId);
        pendingId = null;
        pendingCounts = null;
        return true;
    }

    public int views(String video) {
        return db.getOrDefault(video, 0);
    }
}

Common mistakes

Removing data from the buffer before the commit succeeds (lost views on crash).
Allowing a second commit of the same batch (double counting).
Moving new data into a batch that is already pending.

Follow-up questions

Persist committed batch ids in the database transaction.
Bound the buffer and apply backpressure when flushes fail.

Example 1

Hints

Write and run your solution

Solution and explanation

Reference solution

Common mistakes

Follow-up questions

Example 1

Hints

Write and run your solution

Solution and explanation

Reference solution

Common mistakes

Follow-up questions

More Design & Data Structures problems