Design a Top-K Heavy Hitters Service

• Top K items for a window: K up to 1,000, for the last minute, the last hour, the last 24 hours and all time. Each entry is an item id with its count, in rank order; a request for the top 10 or 100 is a slice of the same list.
• Windows slide, they do not reset: “Last hour” at 10:42 means 09:42 to 10:42, not 10:00 to 10:42. The 1-minute list moves every 10 seconds and the 1-hour list every minute; the 24-hour list is minute-fresh at the head and drops its tail an hour at a time.
• Exact counts for the long windows: 24-hour and all-time counts must agree with a full recount of every view, except for the hour or so the recount has not reached yet, and the list says how big that part is.
• Every list says how fresh and how exact it is: as_of, the hour up to which counts are exact, and the largest possible over-count on the rest. Ranking, reporting and analytics teams consume these lists too, and they need to know what they are reading.
• Any countable item, not only videos: Hashtags from posts use the same pipeline with the key (type, id). They have far more distinct values than videos, and spammers can mint millions of new ones a minute.
• Removed videos drop out within seconds: A takedown or a video going private removes it from every list without waiting for its counts to age out. A video at number one all time would otherwise never leave.
• Out of scope: Deciding what counts as a view (watch-time rules, bot and fraud filtering happen upstream; we receive validated view events), personalised trending, per-country lists (a follow-up), and the view count on the watch page.

• Ingest rate (1 M views/s peak): About 30 B views a day. Popularity is heavily skewed: one video can be 5 % of all traffic during a live moment, and most videos get a handful of views.
• Freshness of the short windows (≤ 1 min): A 1-minute list that is five minutes old is a different product. This drives the hardest tradeoff in the design: at this rate, fresh and exact do not come from the same computation, so there are two paths.
• Accuracy (exact through the last recounted hour): Short windows may over-count by a stated bound and never under-count, so a true heavy hitter cannot be hidden. Long windows are exact except for the open hour or two, which is bounded the same way.
• Read latency (p99 < 50 ms): A read returns a list that already exists. Nothing is counted, merged or ranked on the read path.
• Read volume (~50 k reads/s, spiky): The trending shelf is on the home page. Almost every read is for one of eight lists, identical for every user, which is what makes reads cheap.
• Counting state bounded (independent of distinct items): Memory must not grow with the long tail or with a spam wave of a million junk hashtags. This is what pushes the fast path towards a sketch.
• Availability of reads (99.99 %): A list a minute old is fine; an empty shelf is not. When counting stalls, serve the last good list with its as_of rather than an error.

• View rate: ~350 k/s avg · 1 M/s peak. 30 B views/day ÷ 86 400 ≈ 350 k/s. Peak is about 3× the average, 1 M/s, so a peak minute is 60 M views and an average hour 1.25 B.
• Raw event volume: ~3 TB/day raw · ~30 TB kept. Event id, item id, timestamp, session hash and a few attributes: ~100 B. 30 B × 100 B = 3 TB/day raw, about 1 TB compressed in columnar files. Thirty days of raw views is ~30 TB, enough to recount any hour of the last month.
• Exact state for a sliding hour: ~15 GB. Assume ~5 M distinct videos in a peak minute at ~50 B per hash-map entry: 250 MB per minute bucket. A sliding hour keeps 60 buckets: 15 GB. A sliding day at minute resolution would be 360 GB. Affordable for an hour, not for a day, and unbounded once hashtag spam arrives.
• Count-min sketch size: 272 k × 5 · ~5.4 MB. For ε = 10⁻⁵ and δ = 1 %: width ⌈e/ε⌉ = 271 829, depth ⌈ln(1/δ)⌉ = 5. 1.36 M counters × 4 B = 5.4 MB per bucket, however many distinct items there are. 60 minute buckets for the hour: ~330 MB, about 45× less than exact.
• Sketch error bound, peak minute: ≤ 600 views. Over-count ≤ εN with probability 1 − δ: 10⁻⁵ × 60 M = 600 views. If popularity is roughly Zipf, the 1,000th video in a peak minute has ~4 k views, so the bound is ~15 % at the bottom of the list and negligible at the top, where the leaders have hundreds of thousands.
• A viral video after pre-aggregation: ~32 records/s. A video at 5 % of peak is 50 k views/s. With 64 pre-aggregators each emitting one combined count per item every 2 s, its counter partition receives 64 ÷ 2 = 32 records/s instead of 50 000. The hot key disappears; the total volume barely changes.
• Exact 24-hour recount: ~720 M rows. Each closed hour collapses 1.25 B views into ~30 M (item, hour) rows. Recomputing the 24-hour totals reads 24 × 30 M = 720 M rows: a few minutes of batch, cheap enough to recompute every hour instead of trusting a running total through late corrections.
• Origin reads for the lists: ~500/s. 8 lists (2 types × 4 windows) × 3 allowed sizes of k = 24 cacheable responses. With ~200 CDN PoPs and a 10 s TTL, the origin sees at most 200 × 24 ÷ 10 ≈ 480/s, whether users send 5 k or 500 k reads a second. Read volume is not a design driver.

• Video player: Web, mobile and TV players. When playback qualifies as a view, the player adds a view event with a client-generated event id to a batched beacon, and retries with the same id if the beacon fails. Playback never waits on it.
• Post service (hashtag uses): Another team’s system. Emits one event per hashtag per published post, which enters the same pipeline under the key (hashtag, tag). Hashtags have far more distinct values than videos and a spam problem, which is one reason counter memory is bounded.
• View ingest (edge · stateless): Accepts beacons at edge PoPs, checks signature and schema, and appends events to the view log in batches. Keys each event by session hash, not by video, so traffic spreads evenly whatever is popular. Holds no state, so it scales with requests and never sees a hot key.
• View log (Kafka · 256 partitions by session): The durability boundary and the fork between the two paths: the fast path reads it live, and an archive sink copies it to object storage. Keyed by session hash, so partitions stay balanced during a viral spike. Kept 3 days, so a broken consumer can rewind instead of losing views.
• Pre-aggregator (combine 2 s, then keyBy item): 64 stream tasks, each reading four log partitions, that sum views per (item, 10-second bucket) in a hash map and flush every 2 seconds. Only then is anything routed by item. The combine is what turns a video at 50 k views a second into 32 records a second at its counter partition.
• Window counters (count-min sketch + min-heap): 64 partitions, each owning a hash range of items, so an item’s whole count lives in one place. Per bucket, a count-min sketch and a min-heap of the partition’s top 1,000: six 10-second buckets for the sliding minute, a ring of 60 minute buckets for the sliding hour, and 24 hour sketches for open hours of the day. About 8 MB per partition, so a checkpoint every 10 s restores in about a second.
• Raw archive (object store · Parquet by hour): Every view event, written by a sink on its own consumer group and partitioned by event hour. An hour is closed 10 minutes after it ends, to let late beacons land. Kept 30 days: the input to every exact count and to any recount after a bug.
• Exact batch (Spark · hourly): For each closed hour: drop duplicate event ids, count per item with a map-side combine, and overwrite that hour’s partition of exact counts. Then recompute the 24-hour totals from 24 hourly partitions, apply the hour to the all-time totals, and write the exact candidates for both lists. Usually done by 20 past the hour.
• Exact counts (wide-column · keyed by item): Hourly counts per (item, hour), the exact 24-hour and all-time totals per item for point reads, and the exact top 3,000 candidates per window with the hour they are exact until. Written only by the batch, read by the list builder. Hourly partitions are overwritten whole, so a rerun replaces rather than adds.
• List builder (merge · stitch · publish): Merges the 64 partitions’ top lists into the global 1-minute list every 10 seconds and the 1-hour list every minute. Once a minute, builds the 24-hour and all-time lists by adding exact closed hours to the sketch’s estimate for the open ones. Hydrates titles, drops removed videos, and publishes each list as a new immutable version.
• Trending page: The home-page shelf, the trending tab, and internal consumers such as ranking and reporting jobs. Shows “updated 20 s ago” from as_of, and asks for k = 10, 100 or 1,000 so that every response is cacheable.
• CDN (10 s / 30 s TTL): Caches the 24 cacheable responses at every PoP: 10 seconds for the short windows, 30 for the long ones, with stale-while-revalidate. Origin traffic is set by PoPs and TTL, not by users: about 480 requests a second at worst. Takedowns purge the affected keys.
• Top-K API (stateless · two reads): Reads the current version pointer and the blob, slices the first k, and returns it with as_of, exact_until and max_overcount. No counting, merging or ranking happens here, which is what keeps p99 under 50 ms at any read rate and keeps the shelf up when counting stalls.
• Top-K lists (Redis · versioned blobs): One blob per (type, window, version), about 100 KB for 1,000 hydrated entries, plus a pointer to the current version of each list. The builder writes the new version and then moves the pointer, so a reader never sees a half-built list, and rollback is moving the pointer back.
• Video metadata (titles · status events): The video catalogue, owned by another team. The builder batch-reads titles and thumbnails for each list and subscribes to status changes, so a takedown or a video going private is filtered out of every list within seconds.

1. One view, from the player to the 1-minute list: The fast path. Note what is not on it: no per-view database write, nothing keyed by video until counts have been combined, and no ranking at read time.
   1. The player sends a batched beacon containing a qualified view and its event id.
   2. Ingest appends the events to the view log, keyed by session.
   3. Hashtag uses from the post service enter the same way, under their own item type.
   4. A pre-aggregator sums views per item and 10-second bucket for 2 seconds, then routes each sum to the item’s counter partition.
   5. The counter partition adds the count to that bucket’s sketch and offers the item to the bucket’s heap.
   6. Every 10 seconds, each partition reports its own top 1,000 for the last 60 seconds, and once a minute for the last hour.
   7. The builder merges the 64 lists, hydrates titles, drops removed videos and publishes a new version.
   8. The trending page reads the 1-minute list through the CDN.
2. Build and serve the 24-hour list: The read side picks per hour bucket: exact where the batch has finished, the sketch’s estimate where it has not. The work happens once a minute in the builder, never per read, and a takedown is applied at the same place.
   1. Once a minute, the builder reads the exact 24-hour candidates and the hour they are exact until.
   2. It asks the counters for open-hour estimates of those candidates, and takes their own top lists for the open hours.
   3. For open-hour risers that are not among the exact candidates, it point-reads their exact 24-hour base.
   4. It scores every candidate as exact closed hours plus the open-hour estimate, and keeps the top 1,000.
   5. It hydrates, filters and publishes the version with exact_until and the bound.
   6. A client asks for the top 100 over 24 hours, and the CDN usually answers.
   7. On a miss, the API reads the pointer and the blob and returns the first 100.
   8. A takedown arrives; the builder republishes every list without the video and purges cached copies.
3. A video goes viral at 50 k views a second: The scale-breaking case. Keying by item is what makes the merge correct, and it is also what would melt one partition. Combining before the shuffle is the fix.
   1. A video goes from 100 to 50 k views a second in five minutes.
   2. The view log stays balanced because nothing upstream is keyed by video.
   3. Each of the 64 pre-aggregators sees about 780 of its views a second and folds them into one running count.
   4. The video’s counter partition receives 32 partial counts a second instead of 50 k views.
   5. Its estimate passes the heap minimum, enters the bucket’s heap and climbs.
   6. It is in the next 1-minute list within seconds, and climbs the 1-hour list as minutes accumulate.
   7. Trending traffic triples as people come to look; the CDN absorbs it.
4. A counter partition dies mid-minute: The failure that tests the freshness promise. Lists must stay correct while they briefly stop moving, and recovery has to take seconds, which is one reason the counting state is kept small.
   1. The worker holding counter partition 17 dies; its counts since the last checkpoint are gone from memory.
   2. The job restores the partition from its last checkpoint, together with the log offsets that produced it.
   3. The pre-aggregators rewind to those offsets and replay about 10 seconds of views.
   4. Meanwhile the builder refuses to publish a merge that is missing a partition.
   5. The API keeps serving the last complete version with its as_of.
   6. Partition 17 reports again, and the next merge is complete.
5. An hour closes and becomes exact: The slow path. Raw events are recounted with deduplication and no approximation, the long lists swap an estimated hour for an exact one, and the fast path is graded against the result.
   1. The archive sink writes the hour’s views to object storage and closes the hour 10 minutes after it ends.
   2. The batch job drops duplicate event ids and counts views per item, combining on the map side.
   3. It overwrites hour 09 in the exact counts store.
   4. It recomputes the 24-hour totals from the last 24 hourly partitions and applies the hour to the all-time totals.
   5. It writes the exact top 3,000 candidates for both windows and moves exact_until to 10:00.
   6. The builder sees the new exact_until and rebuilds the long lists with one fewer estimated hour.
   7. The builder grades the fast path: the 1-hour list it published at 10:00 against the exact hour 09.

Ask: Why a count-min sketch? You said exact counts for an hour fit in 15 GB.

Good answers name: Count-min sketch per bucket plus a min-heap of the top 1,000, Exact hash maps per partition and bucket, Space-Saving or Misra-Gries counters, A Redis sorted set per window, ZINCRBY per view.

Our pick: A count-min sketch per bucket on each counter partition, sized for ε = 10⁻⁵ and δ = 1 %: 272 k counters wide and 5 rows deep, 5.4 MB per bucket for the whole cluster. Each partition sees 1/64 of the views, so it holds 1/64 of the width, 85 KB, for the same absolute bound. Beside each sketch is a min-heap of the partition’s top 1,000 by estimate: on every update, if the item’s estimate beats the heap minimum, it replaces it. The sketch bounds memory against the long tail and against spam, and its linearity is what makes sliding windows a matter of adding and subtracting buckets. Its error lands where it matters least: the leaders have hundreds of times the bound, and near rank 1,000 neighbouring items differ by a few views a minute, so their order is noise even with exact counts. Exactness is the batch path’s job, so the fast path only has to be close and fresh. None of this needs building from scratch: Redis ships a count-min sketch and a top-K structure as data types (the CMS.* and TOPK.* commands), and Apache DataSketches has a frequent-items sketch.

1. Walk me through the error bound. Where do e/ε and ln(1/δ) come from?
   In one row, the other items spread their N counts over w counters, so the expected extra on x’s counter is at most N/w. By Markov’s inequality, the chance that the extra exceeds e × N/w is at most 1/e, and with w = e/ε that threshold is εN. The rows hash independently, so the chance that all d rows exceed it is e to the minus d, which is δ when d = ln(1/δ). Width buys accuracy and depth buys confidence, and depth is cheap because it is logarithmic.
2. Can an obscure video show up at rank 990?
   For a video with almost no views to reach rank 990 in a peak minute, it needs an over-count of about 4,000, more than six times the bound, in all five rows at once. That is far less likely than δ. The realistic failure is milder: items near rank 1,000 that are within a few hundred views of each other swap places. We measure that against the exact counts every hour, and if it ever mattered, doubling the width halves the bound for 5.4 MB more per bucket.
3. What is conservative update, and would you use it?
   On an add, raise only the counters that are at the current minimum, instead of all d. On skewed traffic that cuts over-counting a lot, and it still never under-counts. The catch is that the sketch is no longer linear: adding two of them still gives a valid upper bound, but you can no longer subtract an expired bucket. The sliding ring subtracts, so it keeps the plain sketch; the hour sketches that are only ever summed could use conservative update.
4. Why not one global sketch on one machine? It is only 5.4 MB.
   The memory fits; the update rate and the failure domain do not. One process would receive every partial count, around half a million a second at peak, with no room for 10× growth, and it would be a single point of failure for every list. Splitting by item hash keeps each partition near 10 k updates a second, and because items never span partitions, only the heaps need merging.
5. A bot posts a million unique hashtags in a minute. What happens?
   To the sketch, nothing: memory is fixed, and each junk tag adds one to five counters. N grows by a million, so that minute’s bound grows by 10 views, which nobody will notice. With exact maps the same attack adds a million entries to every bucket it touches, for as long as the window lasts, which is how a counting tier gets taken down by someone who never calls it directly.

Ask: Each partition knows its own top 1,000. Why can’t you just merge them?

Good answers name: Partition by item key; each partition reports its own top 1,000, Random partitions, each keeping a top K′ > K, sketches merged to re-score, Merge each partition’s top K as if it were global, Ship every count to one ranking node.

Our pick: Route every partial count by hash(item) mod 64 after the pre-aggregator, so each item’s count lives on exactly one counter partition, and have each partition report its own top 1,000 rather than a 64th of that. The global top 1,000 must then be inside the union of the 64 × 1,000 candidates, and a 64-way merge of sorted lists finds it in microseconds. The extra candidates also make the sliding hour safe: for an item in the hour’s top 1,000 to be missed, it would have to rank outside its own partition’s top 1,000 in every one of the 60 minutes, with around 64,000 items above it each time. The cost of keying is the shuffle and the hot key, and the pre-aggregator pays for both. Random partitioning with K′ = 3K is the right tool where a shuffle is impossible, and it is worth saying so rather than pretending one answer fits everywhere.

1. Prove the keyed merge is exact. What if the global 1,000th item is 1,001st on its own partition?
   Then 1,000 items on that partition have higher counts, so at least 1,000 items globally are above it, and it is at best 1,001st globally, which contradicts the assumption. So every global top-1,000 item is in its own partition’s top 1,000, and the union of local lists contains all of them. The argument needs only that each item lives in one place, so it holds for exact counts and for sketch estimates alike.
2. Why does each partition report 1,000 and not 16?
   Because the top 1,000 are not spread evenly: by chance one partition can hold 30 of them, and a partition reporting 16 would drop 14. Reporting K per partition makes the proof work with no assumption about how items hash. It costs 64,000 entries every 10 seconds, a few MB, which is nothing next to being wrong.
3. With random partitions, how likely is a miss with K′ = 3K?
   It depends on the data, which is the problem. Each partition sees a random 1/64 of every item’s views, so an item’s local rank tracks its global rank with noise, and an item globally at 1,000 would have to fall below 3,000 on every partition at once, which essentially never happens while it has thousands of views. It becomes likely only for items with a handful of views per partition, where the noise is as large as the count. You measure recall against an exact count rather than argue it.
4. You go from 64 to 128 counter workers. What happens to the state?
   A sketch cannot be split by item, so the cheap way to rescale is never to split one. Hash items into many more logical shards than workers from day one, say 1,024, each with its own small sketch, and rescale by moving whole shards between workers. Each worker merges its own shards’ lists before reporting, which stays exact because items never span shards. Splitting shards later would mean starting a new layout empty at a bucket boundary and running both until the new one has an hour of history.
5. How do you know a partition’s report is missing or late?
   Each report carries its partition id and as_of. The builder publishes a merge only when all 64 reports for the same as_of are in; after a timeout it keeps serving the previous version and flags it stale, rather than publishing a list missing 1/64 of the catalogue. Missing reports per partition is a metric with an alert, because a silent partial merge looks completely normal to anyone reading it.

Ask: “Last hour” at 10:42 means 09:42 to 10:42. How do you keep that current every minute without counting every view sixty times?

Good answers name: Ring of buckets; window = running sum + newest − expired, Recompute the window from its buckets at each step, Native hopping windows in the stream processor, Exponentially decayed counters.

Our pick: Count each view once into its bucket and build windows from buckets, with each maintenance strategy used where it fits. The fast path keeps, per partition, six 10-second sketches for the sliding minute and a ring of 60 minute sketches for the sliding hour, each with a running sum: at every boundary it adds the bucket that just closed and subtracts the one that fell out, which costs the same whether 5 M or 50 M items were seen. Because a heap cannot be subtracted, each window’s list is rebuilt at every step by re-scoring the union of its buckets’ heap members against the window’s sketch. The fast path also keeps one sketch per closed hour for the last day, which is how it covers open hours for the 24-hour list. The exact path recomputes instead of subtracting: the 24-hour totals are summed from 24 hourly partitions every hour, so a corrected hour fixes itself, and all-time is a running total because it only ever grows. Say the granularity out loud: the 24-hour window is minute-fresh at its head and moves its tail an hour at a time.

1. Why does the heap have to be rebuilt? Can’t you just decrement?
   A min-heap of the top K knows only its members. When a bucket expires, members lose counts and may fall below items that were never in the heap, and the heap has no record of those items. So at each step the window’s candidates, the union of its buckets’ heap members, a few thousand items per partition, are re-scored from the window sketch, which takes milliseconds.
2. After a subtraction, is the hour sketch still a valid over-estimate?
   Yes, for a plain count-min sketch. The hour sketch is exactly the counter-wise sum of 60 minute sketches, which is the sketch you would have got by feeding it the hour’s views directly, so every guarantee holds for the hour’s N. It only stays true if every add to a bucket is also applied to the running sum, so a late view inside the ring updates both, and a view older than the ring is left to the batch.
3. What does the 24-hour window actually cover at 10:42?
   Twenty-three closed hours, from 11:00 yesterday to 10:00 today, plus the open hour from 10:00 to 10:42: 23 hours and 42 minutes. It moves at the head every minute and drops an hour at the tail on the hour, so it always covers between 23 and 24 hours. Nobody reading a trending list can tell, and it lets the day be built from exact hour buckets. If the product needs exactly 24 hours, keep minute buckets for the oldest hour too, which costs 60 more buckets.
4. Why not calendar windows, “today” rather than “the last 24 hours”?
   They are cheaper, because nothing ever expires: you add to today’s bucket and start a new one at midnight. But “today” at 00:05 is five minutes of data, so the list is noise for the first hour, and midnight is different in every time zone. Sliding windows cost a subtraction or a recompute and give a list that means the same thing at every moment, which is what a trending shelf needs.
5. Is the 1-minute list the last 60 seconds, or the last completed minute?
   The last 60 seconds, to within 10: it is built from six 10-second buckets and refreshed every 10 seconds, the same add-newest, subtract-oldest trick at a finer grain. The last completed minute would be simpler, but it is up to two minutes old by the time it is replaced, which misses the freshness target. The 1-hour list does not need this: refreshing it every minute moves it by under 2 %.

Ask: You keyed the counters by video. What happens when one video is 5 % of all views?

Good answers name: Pre-aggregate per (item, bucket) for 2 s before the keyed shuffle, Pre-aggregate on the ingest hosts at the edge, Split the hot key into N salted sub-keys and sum them, More partitions.

Our pick: Pre-aggregate before the keyed shuffle. 64 pre-aggregator tasks read the session-keyed log, sum views per (item, 10-second bucket) in a hash map, and flush every 2 seconds. Each item then reaches its counter partition as at most one record per task per flush, so the viral video arrives as 32 records a second instead of 50 k, while the total record count drops only modestly, because the tail is seen about once per flush anyway. The bucket is part of the combine key, so every partial count belongs to exactly one bucket and the counters can add it in the right place. The log upstream is keyed by session, so it never had a hot partition; routing by item happens exactly once, after the combine. Salting is the escalation if one key ever outgrows even this; here the combine is enough and needs no second stage.

1. How do you know a key is hot before it hurts?
   You mostly do not need to, which is the advantage of combining every key rather than only hot ones. For monitoring, each counter partition reports its input rate and consumer lag, and lag per partition is the alert: a hot partition shows up as one lagging consumer while the others idle. If salting were ever added, the counters’ own top list by rate would be the detector.
2. Why 2 seconds? Why not 10 and save more?
   At 2 s the viral video is already down to 32 records a second, which nothing downstream notices, so 10 s would buy little. It would cost 8 more seconds of the freshness budget, and since the 1-minute list refreshes every 10 seconds, every refresh would show data up to 20 seconds old. The flush interval should be small next to the refresh interval, and 2 s against 10 s is that.
3. A pre-aggregator crashes with 2 seconds of sums in memory.
   Those sums were never emitted, and the job restores from its last checkpoint, which includes the source offsets, so the views are re-read from the log and summed again. Sums emitted before the checkpoint are already in the counters’ checkpointed state, so they are not applied twice. Because the pre-aggregators and counters are one checkpointed job, the pair behaves as a single exactly-once unit.
4. Ten times the traffic, and one video is 20 % of it: 2 M views a second.
   The cap still holds, because records per key are set by the number of pre-aggregator tasks and the flush interval, not by the view rate. With 640 tasks at ten times the scale, that is 320 records a second, still trivial for one partition. What grows is the pre-aggregators’ own work, and that is spread evenly because the log is keyed by session.
5. Does the exact batch path have the same problem?
   Yes: grouping by item puts the viral video’s 180 M views for the hour on one reduce task. The fix is the same idea, a map-side combine, so each mapper sends one partial count per item. Spark does this by default for aggregations, which is why the job is written as an aggregation and not as a group-by that collects rows.

Ask: You have an approximate count and an exact count for the same video. Which one does the trending page show?

Good answers name: Stitch per hour bucket: exact where final, sketch estimate where open, Serve the fast list until the exact list exists, then switch, Exact streaming only: keyed exact counters for every window, Batch only, every few minutes.

Our pick: Two paths from one log, and a builder that picks per bucket. The fast path alone answers the 1-minute and 1-hour lists, labelled approximate with their bound. For 24 hours and all time, the builder takes exact totals for every hour before exact_until from the counts store, adds the counters’ estimates for the hours after it, and ranks the sums over two candidate sets: the batch’s exact top 3,000, and the counters’ top lists for the open hours. Every list carries as_of, exact_until and max_overcount, so a consumer can choose to wait for exactness. When an hour is finalised, the builder rebuilds with one fewer estimated hour and grades the 1-hour list it published at the top of that hour against the exact counts, recording recall and the largest over-count. If the batch falls behind, nothing breaks: more hours are estimated, the bound grows and is published, and exact_until lag pages someone long before the bound becomes embarrassing.

1. The batch is three hours late. What do users see?
   The same lists, slightly less exact. The builder estimates four open hours instead of one or two from the counters’ hour sketches, which cover a full day, and max_overcount on the 24-hour list grows to about 50 k views, a few percent of the 1,000th video’s count. exact_until lag alerts at two hours, long before anything looks wrong. If the batch were down for a day, the 24-hour list would be fully approximate: a degraded mode, not an outage.
2. And if the fast path is down?
   The 1-minute and 1-hour lists stop moving, are served with their as_of, and are marked stale after two minutes. The 24-hour and all-time lists fall back to exact-only lists through exact_until, which the batch writes anyway, so they are up to an hour behind but exact. Nothing on the read path depends on the counters being up.
3. Isn’t this the Lambda architecture everyone says to avoid?
   It is the same shape, and the usual complaint is maintaining two implementations of the same logic. Here the paths do genuinely different jobs: one ranks with a sketch in bounded memory, the other counts exactly with deduplication. What they must share is the definition of an item and a view, which lives in one schema and one library. The hourly comparison is what keeps them honest; without it, two paths drift apart silently.
4. What exactly do you compare every hour, and what are the thresholds?
   At the top of each hour, the sliding 1-hour list covers exactly the hour that just closed, so it can be graded against the exact counts once that hour is final. Recall at 100, recall at 1,000, the largest over-count against the published bound, and the largest rank displacement in the top 100. I would alert if recall at 100 drops below 99 % or any over-count exceeds the bound, because either one means a bug rather than bad luck.
5. Why not replay the log through the fast path to correct it, Kappa-style?
   A replay produces the same sketch with the same over-count, because the error comes from the structure, not from missing data. Being exact needs exact counting and whole-hour deduplication, which is a different computation, not a rerun. Kappa works when the streaming job is itself exact and the log retention covers the recount, and neither holds for a sketch over a 3-day log.

Ask: Why not compute the top K when someone asks for it?

Good answers name: Precomputed, versioned list blobs behind a CDN, A Redis sorted set per list, rewritten by the builder, Compute on read from the counters, Push lists to clients over WebSockets.

Our pick: The builder publishes each list as an immutable blob keyed by (type, window, version), about 100 KB for 1,000 hydrated entries, and then moves a pointer key to the new version, so readers see the old list or the new one and rollback is moving the pointer back. The API reads the pointer and the blob, slices the first k, and sets Cache-Control by window: 10 seconds for the short lists and 30 for the long ones, with stale-while-revalidate so a PoP never makes a user wait on the origin. k is restricted to 10, 100 or 1,000, which keeps the entire read surface to 24 cacheable responses. Takedowns republish every list from the builder’s extra candidates and purge the 12 affected keys, so a removed video is gone in seconds rather than at the next TTL. Nothing on the read path counts, ranks or merges, which is why it meets p99 < 50 ms and keeps serving the last good lists when everything behind it is down.

1. Why version the blob instead of overwriting one key?
   Overwriting one key is atomic, but a publish touches the list, its labels and several sizes of k, and a reader between those writes could get a mix. Writing a new version and then flipping one pointer makes the switch a single atomic write. It also gives rollback for free: if a bad build ships a list full of spam, the pointer goes back while the bug is found.
2. Someone wants the top 1,000 for every country. Does this still work?
   The shape does; the numbers change. 250 countries × 4 windows × 2 types is 2,000 lists, about 200 MB per set of current versions, which is still small. Most countries have far fewer than 1,000 items above noise, so the builder publishes shorter lists for them. The read side is unchanged, and 6,000 cache keys still fit easily in every PoP.
3. The CDN is down or bypassed. Can the origin take 50 k reads a second?
   Yes, with margin, because a read is two Redis gets and a slice. A handful of API servers that keep the current blob in process for a second would serve it, and Redis would see a few hundred reads a second. The CDN is there for latency and cost, not because the origin could not cope.
4. How do you make sure a removed video never reappears in a later list?
   The builder keeps a deny set fed by the metadata status stream and applies it at every publish, and the API checks a small in-memory copy as a second line. If the status stream lags, the builder also re-reads the status of the ids it publishes, so the exposure window is one publish interval. Counting continues untouched, because the views happened and the video may be restored.
5. What does a client do with exact_until and max_overcount?
   The trending shelf ignores them and shows the list. Internal consumers use them: a daily report waits for a list whose exact_until has passed the end of its day, and a recommendation model that uses trending counts as a feature can weight approximate ones less. Publishing the bound costs nothing, and it is what stops someone treating a provisional count as a fact.