Flask Memory
Leaks

This is the single most expensive mistake you can make. It is also the most tempting, because the code looks completely reasonable:

def stream_proxy():
    upstream = requests.get(external_url, stream=True)

    def generate():
        for chunk in upstream.iter_content(chunk_size=4096):
            yield chunk

    return Response(stream_with_context(generate()))

The problem is not the code itself. The problem is what happens at the OS level. Each concurrent listener holds open two persistent TCP sockets: one from the external server to your Flask process, and one from your Flask process to the browser. Each socket has a kernel receive buffer that the OS allocates regardless of whether your Python code is reading it. On Linux, these buffers can reach 8 MB per socket under load.

# For HTTPS streams: redirect and let the browser connect directly
if parsed.scheme == 'https':
    return redirect(stream_url)

# For HTTP streams: refuse and tell the user to fix their upstream
return jsonify({'error': 'HTTP streams not supported. Use HTTPS.'}), 400

If you absolutely must proxy HTTP streams for a business reason you cannot work around, use a purpose-built reverse proxy (nginx, Caddy, HAProxy) in front of Flask, and route streaming paths there. These tools are built for persistent connections. Flask is not.

This one is subtle. You add a server-side cache to an expensive endpoint. You set a TTL. You add a client-side polling interval. You ship. Memory climbs. The problem: your cache TTL is shorter than your polling interval.

# Server: cache expires after 10 seconds
_cache[tenant_id] = {'data': result, 'expires': time.time() + 10}

# Client: polls every 15 seconds
setInterval(fetchNowPlaying, 15000);

Every single browser poll hits an expired cache. The cache is never warm when the client asks. You might as well have no cache at all. Every poll triggers the full expensive operation — which in this case was an outbound HTTP connection to an external server, with all the socket overhead that entails.

# Server: cache outlives the polling interval
_cache[tenant_id] = {'data': result, 'expires': time.time() + 45}

# Client: polls less frequently than cache expires
setInterval(fetchNowPlaying, 60000);

With the corrected values: 100 concurrent listeners on the same station all share one cached response. One external fetch per 45 seconds instead of 100. The connection overhead drops by two orders of magnitude.

This is the one nobody writes about. If your app lets users configure a URL that your backend then fetches (a webhook URL, a metadata endpoint, a feed URL), some users will paste the wrong thing. For audio-adjacent apps, they will paste their stream URL instead of their metadata API URL. Your backend will then faithfully fetch a never-ending audio stream on every request.

Validate user-supplied URLs before fetching them. At minimum, reject anything that looks like it will return binary streaming content:

_AUDIO_EXTENSIONS = ('.mp3', '.ogg', '.aac', '.m4a', '.opus')
_STREAM_PATH_SEGMENTS = ('/stream', '/live/', '/listen/', '/radio.mp3')

def looks_like_audio_stream(url: str, stream_url: str) -> bool:
    lower = url.lower().rstrip('/')
    return (
        url == stream_url  # user pasted stream URL into metadata field
        or any(lower.endswith(ext) for ext in _AUDIO_EXTENSIONS)
        or any(seg in lower for seg in _STREAM_PATH_SEGMENTS)
    )

if looks_like_audio_stream(metadata_url, stream_url):
    logger.warning("Rejecting audio stream URL from user: %s", metadata_url)
    return None  # skip the fetch entirely

Flask-Limiter is excellent. Its default storage backend in recent versions is memory://, which stores rate limit counters in process memory. This is fine for development. In production with gevent workers, it silently becomes a leak.

The memory:// backend uses a background cleanup timer to evict expired windows. Under gevent, this timer never fires correctly because gevent monkey-patches threading. Expired entries accumulate indefinitely. The dict grows without bound.

limiter = Limiter(
    key_func=get_remote_address,
    storage_uri="memory://"  # silent leak with gevent workers
)

redis_url = os.environ.get("REDIS_URL")

if not redis_url:
    logger.critical("REDIS_URL not set — rate limiting disabled to prevent memory leak")

limiter = Limiter(
    key_func=get_remote_address,
    storage_uri=redis_url or "memory://",
    enabled=bool(redis_url),  # disabled entirely without Redis
)

When memory climbs and you do not know why, work through this checklist in order:

import tracemalloc, gc

# Add to a protected admin route
def memory_debug():
    gc.collect()
    tracemalloc.start()
    # ... handle some requests ...
    snapshot = tracemalloc.take_snapshot()
    top = snapshot.statistics('lineno')[:20]
    return '\n'.join(str(s) for s in top)

with open('/proc/self/status') as f:
    for line in f:
        if line.startswith('VmRSS'):
            print(line.strip())  # Resident Set Size — what the OS sees

import sys

# Check sizes of known caches
print(len(_rate_windows), sys.getsizeof(_rate_windows))
print(len(_results_cache), sys.getsizeof(_results_cache))