How to Debug Memory Leaks in Python 3.14 Apps Using Memray 1.12 and PyPy 7.3

import sys
import os
import time
import typing
from dataclasses import dataclass
from typing import Dict, List, Optional
import json

# Global cache with intentional leak: never evicts entries
LEAKY_CACHE: Dict[str, "IngestedRecord"] = {}
MAX_CACHE_SIZE = 10_000  # Intentionally ignored to trigger leak

@dataclass
class IngestedRecord:
    """Container for processed data records"""
    record_id: str
    payload: Dict[str, typing.Any]
    ingest_ts: float  # Unix timestamp of ingestion
    processed: bool = False

def load_records_from_disk(file_path: str) -> List[IngestedRecord]:
    """Load raw JSON records from a file, return parsed IngestedRecord objects.

    Args:
        file_path: Absolute path to JSONL file with one record per line.

    Returns:
        List of IngestedRecord instances.

    Raises:
        FileNotFoundError: If file_path does not exist.
        json.JSONDecodeError: If any line is invalid JSON.
    """
    if not os.path.exists(file_path):
        raise FileNotFoundError(f"Record file not found: {file_path}")

    records: List[IngestedRecord] = []
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            for line_num, line in enumerate(f, 1):
                line = line.strip()
                if not line:
                    continue
                try:
                    raw = json.loads(line)
                    record = IngestedRecord(
                        record_id=raw.get("id", f"unknown_{line_num}"),
                        payload=raw,
                        ingest_ts=time.time()
                    )
                    records.append(record)
                except json.JSONDecodeError as e:
                    print(f"Skipping invalid JSON at line {line_num}: {e}", file=sys.stderr)
                    continue
    except Exception as e:
        print(f"Unexpected error loading records: {e}", file=sys.stderr)
        raise

    return records

def process_record(record: IngestedRecord) -> None:
    """Process a single record and add to leaky global cache.

    NOTE: This function is the root cause of the memory leak: it never evicts
    entries from LEAKY_CACHE, even when size exceeds MAX_CACHE_SIZE.
    """
    global LEAKY_CACHE
    # Intentionally skip eviction logic to trigger leak
    LEAKY_CACHE[record.record_id] = record
    record.processed = True
    # Simulate small processing overhead
    time.sleep(0.001)

def run_ingestion_cycle(file_path: str, cycles: int = 5) -> None:
    """Run multiple ingestion cycles to exacerbate memory leak.

    Args:
        file_path: Path to input JSONL file.
        cycles: Number of times to reload and process records.
    """
    for cycle in range(cycles):
        print(f"Starting ingestion cycle {cycle + 1}/{cycles}")
        try:
            records = load_records_from_disk(file_path)
        except FileNotFoundError:
            print(f"Cycle {cycle + 1} failed: input file missing", file=sys.stderr)
            continue
        except Exception as e:
            print(f"Cycle {cycle + 1} failed: {e}", file=sys.stderr)
            continue

        for record in records:
            try:
                process_record(record)
            except Exception as e:
                print(f"Failed to process record {record.record_id}: {e}", file=sys.stderr)

        print(f"Cycle {cycle + 1} complete. Cache size: {len(LEAKY_CACHE)}")
        # Simulate time between cycles
        time.sleep(1)

if __name__ == "__main__":
    # Default to sample data if no path provided
    input_path = sys.argv[1] if len(sys.argv) > 1 else "sample_records.jsonl"
    # Run 10 cycles to clearly show memory growth
    run_ingestion_cycle(input_path, cycles=10)

import subprocess
import sys
import os
import json
import typing
from pathlib import Path

# Configuration for Memray 1.12 profiling run
MEMRAY_VERSION = "1.12.0"
PROFILE_OUTPUT_DIR = Path("./memray_profiles")
APP_ENTRY_POINT = "leaky_app.py"
APP_ARGS = ["sample_records.jsonl"]

def check_memray_installed() -> bool:
    """Verify Memray 1.12+ is installed in the current environment."""
    try:
        result = subprocess.run(
            [sys.executable, "-m", "memray", "--version"],
            capture_output=True,
            text=True,
            check=True
        )
        # Parse version string (format: memray 1.12.0)
        version_str = result.stdout.strip().split()[-1]
        major, minor, patch = map(int, version_str.split("."))
        if major == 1 and minor >= 12:
            print(f"Memray {version_str} detected, compatible with this tutorial")
            return True
        else:
            print(f"Unsupported Memray version: {version_str}. Requires 1.12+")
            return False
    except subprocess.CalledProcessError:
        print("Memray not installed. Install with: pip install memray==1.12.0")
        return False
    except Exception as e:
        print(f"Error checking Memray installation: {e}", file=sys.stderr)
        return False

def run_memray_profile() -> Path:
    """Run Memray 1.12 profile on the target application, return path to output file."""
    PROFILE_OUTPUT_DIR.mkdir(exist_ok=True)
    output_file = PROFILE_OUTPUT_DIR / "leaky_app_profile.bin"

    # Build memray run command: memray run -o output.bin --live -q app.py args
    cmd = [
        sys.executable, "-m", "memray", "run",
        "-o", str(output_file),
        "--live",  # Enable live mode for real-time monitoring
        "-q",  # Quiet mode to reduce noise
        APP_ENTRY_POINT
    ] + APP_ARGS

    print(f"Running Memray profile with command: {' '.join(cmd)}")
    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=300  # 5 minute timeout for profile run
        )
        if result.returncode != 0:
            print(f"Profile run failed with return code {result.returncode}", file=sys.stderr)
            print(f"Stdout: {result.stdout}", file=sys.stderr)
            print(f"Stderr: {result.stderr}", file=sys.stderr)
            raise RuntimeError("Memray profile run failed")
        print(f"Profile complete. Output saved to {output_file}")
        return output_file
    except subprocess.TimeoutExpired:
        print("Profile run timed out after 300 seconds", file=sys.stderr)
        raise
    except Exception as e:
        print(f"Unexpected error running profile: {e}", file=sys.stderr)
        raise

def generate_memray_report(profile_path: Path) -> None:
    """Generate HTML and flamegraph reports from Memray profile output."""
    # Generate HTML report
    html_report = PROFILE_OUTPUT_DIR / "leaky_app_report.html"
    cmd_html = [
        sys.executable, "-m", "memray", "report",
        "-o", str(html_report),
        str(profile_path)
    ]

    # Generate flamegraph
    flamegraph_report = PROFILE_OUTPUT_DIR / "leaky_app_flamegraph.html"
    cmd_flamegraph = [
        sys.executable, "-m", "memray", "flamegraph",
        "-o", str(flamegraph_report),
        str(profile_path)
    ]

    for cmd, report_name in [(cmd_html, "HTML report"), (cmd_flamegraph, "Flamegraph")]:
        try:
            result = subprocess.run(cmd, capture_output=True, text=True, check=True)
            print(f"Generated {report_name}: {report_name.replace(' ', '_').lower()}")
        except subprocess.CalledProcessError as e:
            print(f"Failed to generate {report_name}: {e.stderr}", file=sys.stderr)
        except Exception as e:
            print(f"Unexpected error generating {report_name}: {e}", file=sys.stderr)

def parse_memray_stats(profile_path: Path) -> Dict[str, typing.Any]:
    """Parse Memray stats output to extract key memory metrics."""
    cmd = [sys.executable, "-m", "memray", "stats", str(profile_path)]
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, check=True)
        stats = {}
        for line in result.stdout.splitlines():
            if ":" in line:
                key, val = line.split(":", 1)
                stats[key.strip()] = val.strip()
        return stats
    except Exception as e:
        print(f"Error parsing Memray stats: {e}", file=sys.stderr)
        return {}

if __name__ == "__main__":
    if not check_memray_installed():
        sys.exit(1)

    # Create sample input file if it doesn't exist
    sample_file = Path("sample_records.jsonl")
    if not sample_file.exists():
        print("Creating sample input file with 1000 records...")
        with open(sample_file, 'w', encoding='utf-8') as f:
            for i in range(1000):
                record = {"id": f"record_{i}", "data": "x" * 1024}  # 1KB per record
                f.write(json.dumps(record) + "\n")

    try:
        profile_path = run_memray_profile()
        generate_memray_report(profile_path)
        stats = parse_memray_stats(profile_path)
        print("\nKey Memray Stats:")
        for k, v in stats.items():
            print(f"  {k}: {v}")
    except Exception as e:
        print(f"Profiling workflow failed: {e}", file=sys.stderr)
        sys.exit(1)

import sys
import os
import time
import typing
import subprocess
from pathlib import Path
from dataclasses import dataclass

# PyPy 7.3 configuration
PYPY_VERSION_TARGET = "7.3"
FIXED_CACHE_SIZE = 10_000  # Max entries for fixed cache
LEAKY_CACHE: typing.Dict[str, "IngestedRecord"] = {}
FIXED_CACHE: typing.Dict[str, "IngestedRecord"] = {}

@dataclass
class IngestedRecord:
    record_id: str
    payload: typing.Dict[str, typing.Any]
    ingest_ts: float
    processed: bool = False

def check_pypy_version() -> bool:
    """Verify we're running under PyPy 7.3+"""
    if "PyPy" not in sys.version:
        print(f"Not running under PyPy. Current interpreter: {sys.version}")
        return False
    # Parse PyPy version (format: PyPy 7.3.15 ...)
    version_str = sys.version.split()[1]
    major, minor = map(int, version_str.split(".")[:2])
    if major == 7 and minor >= 3:
        print(f"Running under PyPy {version_str}, compatible with this tutorial")
        return True
    else:
        print(f"Unsupported PyPy version: {version_str}. Requires 7.3+")
        return False

def fixed_process_record(record: IngestedRecord) -> None:
    """Fixed version of process_record that evicts old entries when cache is full."""
    global FIXED_CACHE
    # Evict oldest entry if cache exceeds max size
    if len(FIXED_CACHE) >= FIXED_CACHE_SIZE:
        # Sort by ingest_ts to evict oldest first
        oldest_key = min(FIXED_CACHE.items(), key=lambda x: x[1].ingest_ts)[0]
        del FIXED_CACHE[oldest_key]
    FIXED_CACHE[record.record_id] = record
    record.processed = True
    time.sleep(0.001)

def run_pypy_comparison(file_path: str, cycles: int = 5) -> None:
    """Run leaky vs fixed app under PyPy 7.3, compare memory usage."""
    if not os.path.exists(file_path):
        print(f"Input file not found: {file_path}", file=sys.stderr)
        return

    # Track memory for leaky version
    leaky_memory: typing.List[float] = []
    # Track memory for fixed version
    fixed_memory: typing.List[float] = []

    print("Running leaky version under PyPy 7.3...")
    for cycle in range(cycles):
        # Simulate loading records (reuse logic from first example)
        records = []
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                for line_num, line in enumerate(f, 1):
                    line = line.strip()
                    if not line:
                        continue
                    try:
                        raw = json.loads(line)
                        record = IngestedRecord(
                            record_id=raw.get("id", f"unknown_{line_num}"),
                            payload=raw,
                            ingest_ts=time.time()
                        )
                        records.append(record)
                    except json.JSONDecodeError as e:
                        print(f"Skipping invalid JSON at line {line_num}: {e}", file=sys.stderr)
        except Exception as e:
            print(f"Error loading records: {e}", file=sys.stderr)
            return

        # Process with leaky function
        for record in records:
            try:
                # Leaky function (same as first example)
                global LEAKY_CACHE
                LEAKY_CACHE[record.record_id] = record
                record.processed = True
                time.sleep(0.001)
            except Exception as e:
                print(f"Error processing record: {e}", file=sys.stderr)

        # Get current memory usage (PyPy specific: use resource module)
        try:
            import resource
            mem_usage = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024  # Convert to MB
            leaky_memory.append(mem_usage)
            print(f"Cycle {cycle + 1} leaky memory: {mem_usage:.2f} MB. Cache size: {len(LEAKY_CACHE)}")
        except ImportError:
            print("resource module not available, skipping memory tracking")

    print("\nRunning fixed version under PyPy 7.3...")
    for cycle in range(cycles):
        # Reload records (same as above)
        records = []
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                for line_num, line in enumerate(f, 1):
                    line = line.strip()
                    if not line:
                        continue
                    try:
                        raw = json.loads(line)
                        record = IngestedRecord(
                            record_id=raw.get("id", f"unknown_{line_num}"),
                            payload=raw,
                            ingest_ts=time.time()
                        )
                        records.append(record)
                    except json.JSONDecodeError as e:
                        print(f"Skipping invalid JSON at line {line_num}: {e}", file=sys.stderr)
        except Exception as e:
            print(f"Error loading records: {e}", file=sys.stderr)
            return

        # Process with fixed function
        for record in records:
            try:
                fixed_process_record(record)
            except Exception as e:
                print(f"Error processing record: {e}", file=sys.stderr)

        # Get memory usage
        try:
            import resource
            mem_usage = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024
            fixed_memory.append(mem_usage)
            print(f"Cycle {cycle + 1} fixed memory: {mem_usage:.2f} MB. Cache size: {len(FIXED_CACHE)}")
        except ImportError:
            print("resource module not available, skipping memory tracking")

    # Print comparison
    print("\n=== PyPy 7.3 Memory Comparison (Leaky vs Fixed) ===")
    print(f"Leaky final memory: {leaky_memory[-1]:.2f} MB")
    print(f"Fixed final memory: {fixed_memory[-1]:.2f} MB")
    print(f"Memory saved: {leaky_memory[-1] - fixed_memory[-1]:.2f} MB")
    print(f"Leaky cache size: {len(LEAKY_CACHE)}")
    print(f"Fixed cache size: {len(FIXED_CACHE)}")

if __name__ == "__main__":
    if not check_pypy_version():
        sys.exit(1)

    input_path = sys.argv[1] if len(sys.argv) > 1 else "sample_records.jsonl"
    # Create sample file if not exists
    sample_file = Path(input_path)
    if not sample_file.exists():
        print("Creating sample input file with 1000 records...")
        import json
        with open(sample_file, 'w', encoding='utf-8') as f:
            for i in range(1000):
                record = {"id": f"record_{i}", "data": "x" * 1024}
                f.write(json.dumps(record) + "\n")

    run_pypy_comparison(input_path, cycles=5)

# Correct Memray command for PyPy 7.3 profiling with native JIT tracking
pypy3 -m memray run --native -o pypy_profile.bin -q leaky_app.py sample_records.jsonl

# Run Memray with live dashboard for real-time leak detection
python3.14 -m memray run --live -o live_profile.bin leaky_app.py sample_records.jsonl
# Open http://localhost:8080 in your browser to view the dashboard

import gc

# Find all referrers to a leaked object from Memray report
leaked_obj = FIXED_CACHE["record_0"]  # Example leaked object
referrers = gc.get_referrers(leaked_obj)
print(f"Leaked object has {len(referrers)} referrers:")
for ref in referrers:
    print(f"  - {type(ref)}: {ref}")

python-memray-pypy-debugging/
├── leaky_app.py                # Sample leaky application (Code Example 1)
├── memray_profiler.py          # Memray 1.12 profiling script (Code Example 2)
├── pypy_comparison.py          # PyPy 7.3 leaky vs fixed comparison (Code Example 3)
├── sample_records.jsonl        # Sample 1KB-per-record input file
├── requirements.txt            # Dependencies: memray==1.12.0, flask==3.0.0
├── .github/
│   └── workflows/
│       └── memray-ci.yml       # CI integration for memory leak checks
└── README.md                   # Tutorial summary and setup instructions