What an overflowing buffer loses

# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""What an overflowing buffer loses: drop-newest, drop-oldest, drop-oldest per key.

A keyed stream (one update per record for one of K entities, such as a product's
price) overloads a bounded buffer: the producer is faster than the consumer for a
while, then stops, and the consumer drains what is left. Three full-policies:
drop-newest (reject the arrival), drop-oldest (evict the head), and drop-oldest per
key (a newer update for a key replaces the queued one; only a new key evicts the head).
An exact discrete-event simulation. At the end the consumer's view of every entity is
compared with the truth.
"""
import argparse, json, random
from collections import OrderedDict, deque

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--produce", type=float, default=1500, help="Updates produced per second during the overload.")
p.add_argument("--consume", type=float, default=1000, help="Updates the consumer applies per second.")
p.add_argument("--keys", type=int, default=1000, help="Number of entities (keys).")
p.add_argument("--buffer", type=int, default=2000, help="Buffer capacity (records).")
p.add_argument("--seconds", type=float, default=60, help="Seconds of overload; then the producer stops and the consumer drains.")
p.add_argument("--seed", type=int, default=1, help="Random seed for the key sequence.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()


def run(policy):
    rnd = random.Random(args.seed)
    truth, seen = {}, {}                        # key -> latest value produced / applied
    fifo, keyed = deque(), OrderedDict()        # (key, value, t) for the FIFO policies; key -> (value, t) for per key
    produced = dropped = 0
    ages = []
    t_arr, t_srv, value = 0.0, 0.0, 0
    while True:
        size = len(keyed) if policy == "drop-oldest per key" else len(fifo)
        arriving = t_arr < args.seconds
        if arriving and t_arr <= t_srv:
            t, key = t_arr, rnd.randrange(args.keys)
            value += 1; produced += 1; truth[key] = value
            if policy == "drop-oldest per key":
                if key in keyed:
                    keyed[key] = (value, t); dropped += 1          # the queued, older update is superseded
                else:
                    if len(keyed) >= args.buffer: keyed.popitem(last=False); dropped += 1
                    keyed[key] = (value, t)
            elif size < args.buffer:
                fifo.append((key, value, t))
            elif policy == "drop-newest":
                dropped += 1
            else:                                                 # drop-oldest
                fifo.popleft(); fifo.append((key, value, t)); dropped += 1
            t_arr += 1 / args.produce
        elif size:
            t = max(t_srv, 0.0)
            if policy == "drop-oldest per key":
                key, (v, made) = keyed.popitem(last=False)
            else:
                key, v, made = fifo.popleft()
            seen[key] = v; ages.append(t - made)
            t_srv = t + 1 / args.consume
        elif arriving:
            t_srv = t_arr
        else:
            break
    ages.sort()
    wrong = sum(1 for k, v in truth.items() if seen.get(k) != v)
    return {"policy": policy, "produced": produced, "dropped": dropped, "dropped_pct": 100 * dropped / produced,
            "wrong_final": wrong, "wrong_final_pct": 100 * wrong / len(truth),
            "p50_age_s": ages[len(ages) // 2], "p99_age_s": ages[int(len(ages) * 0.99)]}


rows = [run(pol) for pol in ("drop-newest", "drop-oldest", "drop-oldest per key")]
for r in rows:
    print(f"{r['policy']:<20} dropped {r['dropped']:>7,} ({r['dropped_pct']:4.1f}%)  entities with a wrong final value "
          f"{r['wrong_final']:>4} ({r['wrong_final_pct']:4.1f}%)  age at delivery p50 {r['p50_age_s']:.3f} s  p99 {r['p99_age_s']:.3f} s")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"policies": rows}}, f)