A 6-hour nightly webhook-log replay dies on memory - how do you restructure it to stream?

A nightly replay job walks six hours of webhook delivery records, re-sends the failures, and dies partway through with the machine out of memory. The fix is almost never a bigger box; it is finding the stage that accumulates. ## Locate the materializing stage Read the job backwards from the point of peak memory and look for anything that holds the whole run at once: - `rows = handle.readlines()` - the entire file as a list of strings, up front. - `records = [json.loads(line) for line in rows]` - now the same data again, as parsed objects, which are several times larger than the text they came from. - `failed = [r for r in records if r["status"] >= 500]` - a third copy. - `results = []` with `results.append(...)` in the loop, written only after the loop ends. - A `sorted(...)` or a growing `set` used for deduplication. Any one of these turns a job whose *working set* is one record into a job whose working set is the whole night. Six hours of deliveries at a modest rate is millions of records, and parsed objects carry per-object overhead that raw text does not. ## Make every stage a generator The streaming shape keeps the same logical steps but changes each from a collection into a stage: ```python def records(handle): for line in handle: # the file object yields one line at a time yield json.loads(line) def failures(rows): for row in rows: if row["status"] >= 500: yield row with open(path, encoding="utf-8") as handle: for row in failures(records(handle)): resend(row) ``` Now exactly one line, one parsed record and one in-flight send exist at a time. Peak memory becomes a property of the largest single record, not of the run length - which is the whole point: the job's memory no longer scales with how long the window is, so extending it from six hours to a day changes nothing. Two habits keep it that way. Write results out inside the loop instead of collecting them, and never insert a `list()` "just to check something" - a debugging line like `print(len(list(rows)))` both re-materializes the stream and consumes it. ## Batching, and the boundary bug Downstream sinks usually want groups, not single records - a batch endpoint, a bulk insert, a chunked upload. Hand-rolled batching is where this job breaks in the most annoying way: ```python batch = [] for row in rows: batch.append(row) if len(batch) == 500: send(batch) batch = [] # if the loop ends here, the last partial batch is silently dropped ``` If the record count is not an exact multiple of 500, the final short batch never ships. Nothing raises, nothing logs, and the loss is invisible until someone reconciles counts weeks later. The mirror-image bug is flushing inside the loop *and* after it without clearing, which sends the boundary batch twice - duplicate deliveries, which for a webhook replay is worse than a drop. Since Python 3.12 the stdlib does this correctly: `itertools.batched(rows, 500)` yields tuples and keeps the short final batch, and passing `strict=True` raises instead if you require exact multiples. If you must hand-roll, the flush after the loop guarded by `if batch:` is the line that must exist. ## What cannot stream, and how to bound it Some requirements genuinely need to see everything: global ordering, deduplication across the whole window, "the last event per key". These are barriers, and pretending otherwise just moves the memory spike. The options are to bound the state instead of the data - a dedupe set holding fixed-width keys rather than whole records is often two orders of magnitude smaller - to sort within batches when global order is not truly required, or to push the operation into a database or an external sort. Aggregates should be running values: a `collections.Counter` of status codes costs one entry per distinct code, regardless of how many records were counted. ## Confirming the fix Once every stage is a generator, memory should be flat and roughly independent of input size - so a run over ten minutes of logs and a run over six hours should look the same in RSS. If it still grows, the accumulator survived somewhere: a cache, a list built in an inner loop, or a stage that sorts. Checkpointing progress also becomes possible in the streaming shape - because records are processed one at a time, a crashed run can resume from the last acknowledged offset rather than starting the six hours again.