Designing Memory-Stable Stream Pipelines with Bounded Buffers
When a streaming backend is designed well, it feels effortless: memory stays flat, latency is predictable, and reconnect storms are just another Tuesday. Even as traffic grows and features pile up, the pipeline keeps behaving like a well-tuned instrument instead of a fragile chain of patches.
A pattern that consistently delivers that kind of stability in real-time, event-driven Python systems combines bounded ring buffers, interval-driven coalescing, and timestamp-aligned batching. Together they provide explicit memory limits, stable and predictable latency, and graceful handling of reconnects and bursts, whether the events are IMU samples from a wearable, weight readings from a smart bowl, or telemetry from long-lived app sessions.
Why stream backends drift into trouble
The healthiest stream backends aren't defined by how they handle explosions, they're defined by how they handle the subtle stuff. Unbounded queues and invisible memory creep show up like this: the ingestion layer receives events over HTTP, WebSocket, or MQTT, each event gets appended into an in-memory list or queue, and a background worker periodically drains that structure and writes to storage.
This feels fine in development. Then production happens, a few devices reconnect and replay buffered data, network hiccups cause short high-intensity bursts of traffic, and product features gradually increase the event rate per device. If that buffer is unbounded, your memory usage becomes a function of burst size, worker throughput, and how long issues go unnoticed. By the time you see RSS steadily climbing in your graphs, the process is often one reconnect storm away from getting killed.
Another subtle failure is timing drift between how you think batching works and how it actually behaves. In production, the worker sometimes starts late, processing time varies with payload size, and load patterns change across the day. Your "5 second" cadence turns into "somewhere between 3 and 15 seconds," which breaks assumptions for user-facing freshness, downstream analytics that expect stable windows, and alert logic assuming consistent latency bounds.
Reconnect storms and replays are normal with mobile clients, wearables, or smart devices, devices move through patchy coverage, apps get killed and relaunched by the OS, and firmware buffers and replays events after outages. If your pipeline isn't explicit about per-device limits and what to do with replays, a reconnect storm can inject a large backlog of old events into your hot path, starve fresher, more relevant data, and inflate the memory footprint and processing time for every tick.
The smell in observability includes memory graphs that never fully flatten after load spikes, latency spikes aligning with a spike in active sessions, and coalescing or flush jobs whose run duration varies wildly.
The pattern: bounded, time-aware stream processing
The architecture we want can be summarized in three ideas: bounded per-stream ring buffers, interval-driven coalescing (a metronome), and timestamp-aligned batching (stable windows). This gives you explicit control over how much history you keep in memory, when work happens, and how events map into logical time windows.
For each logical stream (device_id, user_id, bowl_id), maintain a fixed-capacity ring buffer. Capacity is expressed as "N most recent events," and when full, new events replace or evict older ones according to a policy:
from collections import deque
from dataclasses import dataclass
from typing import Any
@dataclass
class Event:
ts: float # event timestamp in seconds (device or server time)
payload: Any
class RingBuffer:
def __init__(self, capacity: int):
self.capacity = capacity
self._data = deque(maxlen=capacity)
def append(self, event: Event):
self._data.append(event) # oldest entries overwritten automatically
def snapshot(self) -> list[Event]:
return list(self._data) # shallow copy for safe iterationPer-stream memory usage becomes a simple product: max events per stream equals capacity, and max events overall equals capacity times active streams. Now you can reason about RAM in advance instead of discovering it in your alerts.
Instead of flushing whenever enough events accumulate, introduce a coalescer loop firing at a fixed interval, say every 2-5 seconds. At each tick it snapshots the buffers, partitions events into windows, builds batches, and emits them to storage or downstream services. This gives predictable CPU utilization and I/O patterns, a natural place to enforce cluster-wide constraints, and a clean mental model: every interval, each active stream gets processed.
async def coalescer_loop(buffers, window_s: int, interval_s: float):
while True:
now_ts = time()
batches = build_batches(buffers, window_s, now_ts)
for batch in batches:
await emit_batch(batch) # write to DB, queue, or cache
await asyncio.sleep(interval_s)Instead of "whatever arrived between ticks," align events to explicit time windows, fixed size (5 seconds, 30 seconds, or a minute), derived from event timestamps or arrival time. Windows then become first-class, window_id maps to a start/end range, stored alongside aggregates, and used as keys in databases or downstream queries. This is essential for time-series analytics, ML feature engineering, and consistent user-facing updates.
Implementation process in Python
Let's turn the pattern into a concrete pipeline. Assumptions: a Python service receives events via HTTP or WebSockets, events are tagged with a stream identifier and timestamp, and we maintain in-memory bounded buffers, periodically persisting windowed batches to storage.
Step 1 is defining your time semantics. Two key decisions: window size (short windows like 5s mean fresher UI updates but more batches, longer windows like 30-60s mean fewer writes but coarser granularity) and time source (event time, the device timestamp, for analytics and ML, versus arrival time, the server timestamp, for freshness and alerting guarantees). You can store both, but pick one as the canonical dimension for batching, event time windows with arrival time SLAs is a good compromise for many health or telemetry systems.
Step 2 sizes ring buffers from traffic and SLAs. Estimate per-stream peak: r equals maximum events per second per device you're willing to support, H equals horizon in seconds of history you want to keep in memory, so capacity per stream is roughly r times H. Say up to 40 events per second from a device, kept for 30 seconds of history, gives capacity_per_stream of about 1200 events. For 1000 active devices, that's max in-memory events around 1.2 million, and with small payloads you can decide if that fits your RAM budget, otherwise reduce H or aggregate earlier.
Step 3 is ingress, writing into bounded buffers, keeping a map from stream_id to RingBuffer:
from collections import defaultdict
buffers: dict[str, RingBuffer] = defaultdict(lambda: RingBuffer(capacity=1200))
def ingest_event(stream_id: str, ts: float, payload: dict):
evt = Event(ts=ts, payload=payload)
buffers[stream_id].append(evt)You'd call ingest_event from your FastAPI/Starlette/WebSocket handler. For more control over overflow, you can reject new events if the buffer is full, drop oldest events explicitly if they're older than a cutoff, or drop events belonging to already-finalized windows via watermarking.
Step 4 builds batches per window, not per tick. At each coalescer tick, compute window IDs for events in each buffer, group by stream_id and window_id, and build compact batch records:
from collections import defaultdict
def build_batches(buffers, window_s: int, now_ts: float):
grouped = defaultdict(list)
for stream_id, buf in buffers.items():
for evt in buf.snapshot():
wid = window_id(evt.ts, window_s)
grouped[(stream_id, wid)].append(evt)
batches = []
for (stream_id, wid), events in grouped.items():
t_start = wid * window_s
t_end = (wid + 1) * window_s
# Example aggregation; customize for your signals
batch = {
"stream_id": stream_id,
"window_id": wid,
"t_start": t_start,
"t_end": t_end,
"count": len(events),
"events": events, # or aggregated stats only
}
batches.append(batch)
return batchesYou can trim events down to aggregated metrics, min/mean/max, or last value, to reduce write volume, and apply a max batch size if needed. The important property: batches correspond to clear, repeatable windows, not arbitrary boundaries.
Step 5 garbage-collects buffer contents. Even with maxlen, you don't want to retain events longer than necessary, so keep only events newer than now_ts minus horizon:
from collections import deque
def gc_buffers(buffers, horizon_s: int, now_ts: float):
cutoff = now_ts - horizon_s
for buf in buffers.values():
buf._data = deque(
[evt for evt in buf._data if evt.ts >= cutoff],
maxlen=buf.capacity
)This enforces two invariants: no stream buffer holds more than capacity events, and no event older than H seconds stays in memory. Both are easy to reason about for capacity planning.
Step 6 uses watermarks for reconnects and replays. Maintain a per-stream watermark recording the largest finalized time you've fully processed:
from collections import defaultdict
last_processed_ts: dict[str, float] = defaultdict(float)
def ingest_event_with_watermark(stream_id: str, ts: float, payload: dict):
# Ignore events that fall entirely before our last processed time
if ts <= last_processed_ts[stream_id]:
return # too old; already covered by previous windows
ingest_event(stream_id, ts, payload)After emitting batches, update watermarks:
def update_watermarks_from_batches(batches):
for batch in batches:
stream_id = batch["stream_id"]
last_processed_ts[stream_id] = max(
last_processed_ts[stream_id],
batch["t_end"],
)Now when a device reconnects and replays events, old samples get dropped at ingress, only genuinely new windows contribute to memory and processing, and backward time travel from skewed device clocks becomes visible in metrics rather than silently corrupting state. In ecosystems where a wearable or bowl might buffer multiple minutes of data offline, this strategy is critical, you absorb history without threatening your live pipeline.
Step 7 writes to storage and downstream consumers. Batches can be written to a hot read store for app dashboards, a time-series or analytics store for aggregation and ML, or a queue or topic for downstream jobs. Typical key structure uses stream_id as partition key and t_start or window_id as sort key. Every record now represents a well-defined time window, and you can reason about retention, backfills, and reprocessing on a window-by-window basis rather than at random offsets.
What "good" looks like in production
Once you migrate to a bounded, time-aware pipeline, your observability story should change in specific ways. Memory usage plateaus, RSS per process rises from startup to a stable band and stays there under steady load, with bursts producing temporary bumps that return cleanly to baseline. Batch durations show low variance, P50/P95 durations of the coalescer loop stay within a narrow envelope, with no sudden spikes unless something truly exceptional happens. End-to-end latency becomes predictable, bounded by a function of window size, interval, and processing time, stretching gracefully rather than explosively during reconnect storms.
You'll also want metrics like number of active streams, events ingested per second, events dropped by reason, batches emitted per tick, and coalescer loop duration and jitter.
Capacity planning becomes math, not folklore, you can estimate worst-case memory from capacity_per_stream times max_streams rather than finger-crossing. Incidents have crisp root causes, "we mis-sized the horizon" or "window size too small for peak rate" is far clearer than a mysterious OOM. Product experiments become safer, adding new event types or increasing sampling rates is less risky because core invariants around memory and timing are enforced by design.
Hoomanely's mission is helping pet parents keep their pets healthier and happier using continuous, trustworthy data. That data arrives as streams: motion and posture from wearables, weight and consumption patterns from smart bowls, and app interactions from humans in the loop. From the system's perspective, a dog's day is thousands of tiny events, micro-movements distinguishing rest from agitation, subtle bowl weight changes marking snacking versus a full meal, and evening patterns distinguishing normal pacing from potential discomfort.
If the backend leaks memory under reconnect storms, introduces random latency penalties, or handles replays inconsistently, higher-level experiences degrade, insights lag or become noisy, alerts either over-fire or miss important episodes, and long-term behavior models get biased by skewed or duplicated data. Memory-stable, time-aware stream pipelines are one of the quiet foundations of Hoomanely's stack, when you open the app the latest bowl or activity summaries appear with a predictable freshness bound, and when a wearable or bowl comes back online after being offline, the backlog gets ingested without destabilizing other pets' data.
Key takeaways
Bound your memory explicitly, use per-stream ring buffers with fixed capacity, making "how much history do we keep" a deliberate parameter, not an accident. Let a metronome drive your pipeline, run a coalescer loop at a fixed interval rather than tying batching purely to accumulated events. Align everything around windows, treat window_id and the time range as first-class, storing and indexing them and designing downstream logic around them. Treat reconnects and replays as a normal case, maintaining per-stream watermarks so old data is intentionally ignored rather than silently double-counted. And instrument shape, not just averages, watching memory plateaus, batch timing distributions, and drop reasons to tune window size, buffer capacity, and horizons. If you already have a streaming backend, start by wrapping existing queues in ring buffers with clear capacities, add a coalescer task that builds time-aligned batches, and gradually migrate storage and downstream jobs to read from windowed batches rather than ad-hoc lists. Over time your pipeline will feel less like an accumulation of handlers and more like a well-defined, time-aware machine, one that's easier to scale, easier to debug, and aligned with the real-time expectations your users and products demand.
