feat: add option to read memory from cgroup files

.github/workflows/_tests.yml

@@ -11,8 +11,6 @@ jobs:
         os: [ubuntu-latest]
         python-version: ["3.10", "3.11", "3.12", "3.13"]
         include:
-          - os: macos-15-intel
-            python-version: "3.13"
           - os: macos-latest
             python-version: "3.13"
           - os: windows-latest

.github/workflows/manual_tests.yml

@@ -17,8 +17,6 @@ jobs:
         include:
           - os: macos-latest
             python-version: "3.13"
-          - os: macos-15-intel
-            python-version: "3.13"
           - os: windows-latest
             python-version: "3.13"
     env:
@@ -48,7 +46,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: [ubuntu-latest, macos-15-intel, macos-latest, windows-latest]
+        os: [ubuntu-latest, macos-latest, windows-latest]
         python-version: ["3.10"]
     env:
       OS: ${{ matrix.os }}

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- False-positive memory limits in containerized environments by using cgroup-aware memory detection (fixes kraina-ai/quackosm#319)
+
 ## [2026.6.0] - 2026-06-16
 
 ### Added

pyproject.toml

@@ -132,4 +132,5 @@ test = [
     "osmnx>=1.3.0",
     "tqdm>=4.67.1",
     "overturemaps>=0.18.0",
+    "types-psutil",
 ]

rq_geo_toolkit/_system_memory.py

@@ -0,0 +1,241 @@
+"""Cgroup-aware system memory detection."""
+
+from __future__ import annotations
+
+import logging
+from functools import lru_cache
+from pathlib import Path
+from typing import Literal, NamedTuple
+
+import psutil
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class MemoryStatus(NamedTuple):
+    """Snapshot of system/container memory status."""
+
+    total_bytes: int
+    used_bytes: int
+    available_bytes: int
+    percent_used: float
+    source: Literal["cgroup_v2", "cgroup_v1", "psutil"]
+
+
+_CGROUP_V2_CONTROLLERS_PATH = Path("/sys/fs/cgroup/cgroup.controllers")
+_CGROUP_V2_MEMORY_MAX_PATH = Path("/sys/fs/cgroup/memory.max")
+_CGROUP_V2_MEMORY_CURRENT_PATH = Path("/sys/fs/cgroup/memory.current")
+_CGROUP_V2_MEMORY_STAT_PATH = Path("/sys/fs/cgroup/memory.stat")
+
+_CGROUP_V1_MEMORY_PATH = Path("/sys/fs/cgroup/memory")
+_CGROUP_V1_LIMIT_PATH = _CGROUP_V1_MEMORY_PATH / "memory.limit_in_bytes"
+_CGROUP_V1_USAGE_PATH = _CGROUP_V1_MEMORY_PATH / "memory.usage_in_bytes"
+_CGROUP_V1_STAT_PATH = _CGROUP_V1_MEMORY_PATH / "memory.stat"
+
+_CGROUP_UNLIMITED_SENTINEL_V1 = 9_223_372_036_854_771_712
+
+
+def _read_int_from_file(path: Path) -> int | None:
+    """Read an integer from a file, returning None on failure."""
+    try:
+        raw = path.read_text(encoding="utf-8").strip()
+        if not raw:
+            return None
+        return int(raw)
+    except OSError:
+        return None
+    except ValueError:
+        return None
+
+
+def _parse_memory_stat_inactive_file_v2(stat_text: str) -> int:
+    """Parse inactive_file value from cgroup v2 memory.stat."""
+    for line in stat_text.splitlines():
+        if line.startswith("inactive_file "):
+            _, value = line.split(" ", 1)
+            return int(value)
+    return 0
+
+
+def _parse_memory_stat_inactive_file_v1(stat_text: str) -> int:
+    """Parse total_inactive_file value from cgroup v1 memory.stat."""
+    for line in stat_text.splitlines():
+        if line.startswith("total_inactive_file "):
+            _, value = line.split(" ", 1)
+            return int(value)
+    return 0
+
+
+def _get_cgroup_v2_memory_status() -> MemoryStatus | None:
+    """Attempt to read memory status from cgroup v2."""
+    limit = _read_int_from_file(_CGROUP_V2_MEMORY_MAX_PATH)
+    if limit is None:
+        return None
+
+    if _CGROUP_V2_MEMORY_MAX_PATH.read_text(encoding="utf-8").strip() == "max":
+        return None
+
+    raw_usage = _read_int_from_file(_CGROUP_V2_MEMORY_CURRENT_PATH)
+    if raw_usage is None:
+        return None
+
+    inactive_file = 0
+    if _CGROUP_V2_MEMORY_STAT_PATH.is_file():
+        try:
+            stat_content = _CGROUP_V2_MEMORY_STAT_PATH.read_text(encoding="utf-8")
+            inactive_file = _parse_memory_stat_inactive_file_v2(stat_content)
+        except OSError:
+            pass
+
+    used_bytes = max(raw_usage - inactive_file, 0)
+    available_bytes = max(limit - used_bytes, 0)
+    percent_used = used_bytes / limit * 100 if limit > 0 else 0.0
+
+    return MemoryStatus(
+        total_bytes=limit,
+        used_bytes=used_bytes,
+        available_bytes=available_bytes,
+        percent_used=percent_used,
+        source="cgroup_v2",
+    )
+
+
+def _get_cgroup_v1_memory_status() -> MemoryStatus | None:
+    """Attempt to read memory status from cgroup v1."""
+    if not _CGROUP_V1_MEMORY_PATH.is_dir():
+        return None
+
+    limit = _read_int_from_file(_CGROUP_V1_LIMIT_PATH)
+    if limit is None:
+        return None
+
+    if limit >= _CGROUP_UNLIMITED_SENTINEL_V1:
+        host_total = psutil.virtual_memory().total
+        if limit >= host_total:
+            return None
+
+    raw_usage = _read_int_from_file(_CGROUP_V1_USAGE_PATH)
+    if raw_usage is None:
+        return None
+
+    inactive_file = 0
+    if _CGROUP_V1_STAT_PATH.is_file():
+        try:
+            stat_content = _CGROUP_V1_STAT_PATH.read_text(encoding="utf-8")
+            inactive_file = _parse_memory_stat_inactive_file_v1(stat_content)
+        except OSError:
+            pass
+
+    used_bytes = max(raw_usage - inactive_file, 0)
+    available_bytes = max(limit - used_bytes, 0)
+    percent_used = used_bytes / limit * 100 if limit > 0 else 0.0
+
+    return MemoryStatus(
+        total_bytes=limit,
+        used_bytes=used_bytes,
+        available_bytes=available_bytes,
+        percent_used=percent_used,
+        source="cgroup_v1",
+    )
+
+
+def _get_psutil_memory_status() -> MemoryStatus:
+    """Read memory status from psutil."""
+    mem = psutil.virtual_memory()
+    return MemoryStatus(
+        total_bytes=mem.total,
+        used_bytes=mem.used,
+        available_bytes=mem.available,
+        percent_used=mem.percent,
+        source="psutil",
+    )
+
+
+@lru_cache(maxsize=1)
+def _detect_memory_source() -> tuple[
+    Literal["cgroup_v2", "cgroup_v1", "psutil"],
+    Path | None,
+    Path | None,
+]:
+    """Detect which memory source to use and cache the decision."""
+    if _CGROUP_V2_CONTROLLERS_PATH.is_file():
+        v2_limit_path = _CGROUP_V2_MEMORY_MAX_PATH
+        v2_limit_text = ""
+        try:
+            v2_limit_text = v2_limit_path.read_text(encoding="utf-8").strip()
+        except OSError:
+            pass
+
+        if v2_limit_text and v2_limit_text != "max":
+            try:
+                limit = int(v2_limit_text)
+                if limit > 0:
+                    return "cgroup_v2", v2_limit_path, _CGROUP_V2_MEMORY_CURRENT_PATH
+            except ValueError:
+                pass
+
+    if _CGROUP_V1_MEMORY_PATH.is_dir():
+        v1_limit_path = _CGROUP_V1_LIMIT_PATH
+        v1_limit_text = ""
+        try:
+            v1_limit_text = v1_limit_path.read_text(encoding="utf-8").strip()
+        except OSError:
+            pass
+
+        if v1_limit_text:
+            try:
+                limit = int(v1_limit_text)
+                if 0 < limit < _CGROUP_UNLIMITED_SENTINEL_V1:
+                    return "cgroup_v1", v1_limit_path, _CGROUP_V1_USAGE_PATH
+                host_total = psutil.virtual_memory().total
+                if 0 < limit < host_total:
+                    return "cgroup_v1", v1_limit_path, _CGROUP_V1_USAGE_PATH
+            except (ValueError, OSError):
+                pass
+
+    return "psutil", None, None
+
+
+def get_memory_status(total_bytes_override: int | None = None) -> MemoryStatus:
+    """
+    Return current memory status, preferring cgroup-scoped values when available.
+
+    Args:
+        total_bytes_override: When provided, use this value as the total memory limit
+            instead of auto-detecting it from cgroup/psutil. Usage and percent are still
+            read from the best available source.
+
+    Returns:
+        MemoryStatus with total, used, available, percent, and the source used.
+    """
+    source, _, _ = _detect_memory_source()
+
+    if source == "cgroup_v2":
+        status = _get_cgroup_v2_memory_status()
+    elif source == "cgroup_v1":
+        status = _get_cgroup_v1_memory_status()
+    else:
+        status = None
+
+    if status is None:
+        status = _get_psutil_memory_status()
+
+    _LOGGER.debug(
+        "Memory status resolved: source=%s total_bytes=%d",
+        status.source,
+        status.total_bytes,
+    )
+
+    if total_bytes_override is not None:
+        used = status.used_bytes
+        available = max(total_bytes_override - used, 0)
+        percent = used / total_bytes_override * 100 if total_bytes_override > 0 else 0.0
+        status = MemoryStatus(
+            total_bytes=total_bytes_override,
+            used_bytes=used,
+            available_bytes=available,
+            percent_used=percent,
+            source=status.source,
+        )
+
+    return status

rq_geo_toolkit/duckdb.py

@@ -16,10 +16,10 @@
     from typing_extensions import TypedDict
 
 import duckdb
-import psutil
 from packaging import version
 from rich import print as rprint
 
+from rq_geo_toolkit._system_memory import get_memory_status
 from rq_geo_toolkit.constants import MEMORY_1GB
 from rq_geo_toolkit.multiprocessing_utils import WorkerProcess, run_process_with_memory_monitoring
 
@@ -133,13 +133,13 @@ def run_duckdb_query_function_with_memory_limit(
     duckdb_conn_kwargs: Optional[DuckDBConnKwargs] = None,
 ) -> tuple[float, int]:
     """Run function with duckdb query and limit threads automatically."""
-    current_memory_gb_limit = current_memory_gb_limit or ceil(
-        psutil.virtual_memory().total / MEMORY_1GB
-    )
-    current_threads_limit = (
-        current_threads_limit
-        or duckdb.sql("SELECT current_setting('threads') AS threads").fetchone()[0]
+    current_memory_gb_limit = float(
+        current_memory_gb_limit or ceil(get_memory_status().total_bytes / MEMORY_1GB)
     )
+    threads_result = duckdb.sql("SELECT current_setting('threads') AS threads").fetchone()
+    if threads_result is None:
+        raise RuntimeError("Failed to retrieve DuckDB threads setting.")
+    current_threads_limit = current_threads_limit or threads_result[0]
 
     while True:
         try:
@@ -153,7 +153,14 @@ def run_duckdb_query_function_with_memory_limit(
                     duckdb_conn_kwargs=duckdb_conn_kwargs,
                 )
                 process = WorkerProcess(target=f, args=args or (), kwargs=kwargs or {})
-                run_process_with_memory_monitoring(process)
+                override_bytes = (
+                    int(current_memory_gb_limit * MEMORY_1GB)
+                    if current_memory_gb_limit is not None
+                    else None
+                )
+                run_process_with_memory_monitoring(
+                    process, total_bytes_override=override_bytes
+                )
 
             return current_memory_gb_limit, current_threads_limit
         except (duckdb.OutOfMemoryException, MemoryError) as ex:
@@ -228,22 +235,29 @@ def run_query_with_memory_monitoring(
             duckdb_conn_kwargs=duckdb_conn_kwargs,
         )
     elif connection is not None:
-        current_memory_gb_limit = ceil(psutil.virtual_memory().total / MEMORY_1GB)
-        current_threads_limit = connection.sql(
+        current_memory_gb_limit = float(
+            ceil(get_memory_status().total_bytes / MEMORY_1GB)
+        )
+        current_threads_limit_result = connection.sql(
             "SELECT current_setting('threads') AS threads"
-        ).fetchone()[0]
+        ).fetchone()
+        if current_threads_limit_result is None:
+            raise RuntimeError("Failed to retrieve DuckDB threads setting.")
+        current_threads_limit = current_threads_limit_result[0]
 
         while True:
             try:
                 with ThreadPoolExecutor(max_workers=1) as executor:
                     connection.execute(f"SET memory_limit = '{current_memory_gb_limit}GB';")
                     connection.execute(f"SET threads = {current_threads_limit};")
 
-                    actual_memory = psutil.virtual_memory()
-                    percentage_threshold = 95
-                    if (actual_memory.total * 0.05) > MEMORY_1GB:  # pragma: no cover
+                    actual_memory = get_memory_status()
+                    percentage_threshold: float = 95
+                    if (actual_memory.total_bytes * 0.05) > MEMORY_1GB:  # pragma: no cover
                         percentage_threshold = (
-                            100 * (actual_memory.total - MEMORY_1GB) / actual_memory.total
+                            100
+                            * (actual_memory.total_bytes - MEMORY_1GB)
+                            / actual_memory.total_bytes
                         )
 
                     query_execution_future = executor.submit(
@@ -252,8 +266,8 @@ def run_query_with_memory_monitoring(
 
                     sleep_time = 0.1
                     while query_execution_future.running():
-                        actual_memory = psutil.virtual_memory()
-                        if actual_memory.percent > percentage_threshold:  # pragma: no cover
+                        actual_memory = get_memory_status()
+                        if actual_memory.percent_used > percentage_threshold:  # pragma: no cover
                             connection.interrupt()
                             query_execution_future.cancel()
                             raise MemoryError()