fix(memory-monitor): lower fail-fast thresholds to catch OOM before cliff

Co-Authored-By: patrick.nilan@airbyte.io <patrick.nilan@airbyte.io>

Author: devin-ai-integration[bot]

airbyte_cdk/utils/memory_monitor.py

@@ -30,13 +30,19 @@
 #   2. anonymous memory >= anon-share threshold of *current cgroup usage*
 # Comparing anon to usage (not limit) answers the more relevant question:
 # "is most of the near-OOM memory actually process-owned anonymous memory?"
-_CRITICAL_THRESHOLD = 0.98
+#
+# Thresholds are deliberately set below the OOM cliff to leave headroom for
+# the check-interval race window: between two checks, allocations can jump
+# a container past any gate directly into kernel OOM-kill. Firing the fail-
+# fast trace well before the cliff is what makes the failure visible to the
+# platform instead of appearing as a silent exit.
+_CRITICAL_THRESHOLD = 0.95
 _ANON_SHARE_OF_USAGE_THRESHOLD = 0.85
 
 # Check interval (every N messages) — tightens after crossing high-pressure threshold
 _DEFAULT_CHECK_INTERVAL = 5000
 _HIGH_PRESSURE_CHECK_INTERVAL = 100
-_HIGH_PRESSURE_THRESHOLD = 0.95
+_HIGH_PRESSURE_THRESHOLD = 0.90
 
 
 def _read_cgroup_v2_anon_bytes() -> Optional[int]:
@@ -90,21 +96,21 @@ class MemoryMonitor:
 
     **Logging (event-based, not periodic):**
 
-    - One INFO when high-pressure mode activates (usage first crosses 95%)
-    - One INFO/WARNING when critical threshold (98%) is crossed but we do
+    - One INFO when high-pressure mode activates (usage first crosses 90%)
+    - One INFO/WARNING when critical threshold (95%) is crossed but we do
       *not* raise (either anon share is below the fail-fast gate or the
       anonymous memory signal is unavailable)
     - No repeated per-check warnings — logging is driven by state
       transitions, not periodic sampling
 
-    **High-pressure polling:** Once cgroup usage first crosses 95%, the check
+    **High-pressure polling:** Once cgroup usage first crosses 90%, the check
     interval permanently tightens from 5000 to 100 messages to narrow the race
     window near OOM.
 
     **Fail-fast:** Raises ``AirbyteTracedException`` with
     ``FailureType.system_error`` when *both*:
 
-    1. Cgroup usage >= 98% of the container limit (container is near OOM-kill)
+    1. Cgroup usage >= 95% of the container limit (container is near OOM-kill)
     2. Anonymous memory >= 85% of *current cgroup usage* (most of the charged
        memory is process-private anonymous pages, not file-backed cache)
 
@@ -209,7 +215,7 @@ def check_memory_usage(self) -> None:
 
         Intended to be called on every message. The monitor internally tracks
         a message counter and only reads cgroup files every ``check_interval``
-        messages (default 5000). Once usage crosses 95%, the interval tightens
+        messages (default 5000). Once usage crosses 90%, the interval tightens
         to 100 messages for the remainder of the sync.
 
         Logging is event-based (one-shot on state transitions), not periodic.

unit_tests/utils/test_memory_monitor.py

@@ -23,31 +23,31 @@
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 
 _MOCK_USAGE_BELOW = "500000000\n"  # 50% of 1 GB
-_MOCK_USAGE_AT_90 = "910000000\n"  # 91% of 1 GB  (below 95% logging threshold)
-_MOCK_USAGE_AT_95 = "960000000\n"  # 96% of 1 GB  (above 95% logging threshold)
-_MOCK_USAGE_AT_98 = "980000000\n"  # 98% of 1 GB  (at critical threshold)
+_MOCK_USAGE_AT_85 = "850000000\n"  # 85% of 1 GB  (below 90% high-pressure threshold)
+_MOCK_USAGE_AT_92 = "920000000\n"  # 92% of 1 GB  (above 90% high-pressure, below 95% critical)
+_MOCK_USAGE_AT_96 = "960000000\n"  # 96% of 1 GB  (above 95% critical threshold)
 _MOCK_LIMIT = "1000000000\n"  # 1 GB
 
 # cgroup v2 memory.stat mock content.
 # The "anon" field is what we parse for the cgroup-level anonymous memory signal.
 _MOCK_MEMORY_STAT_ANON_HIGH = (
-    "anon 860000000\n"  # 860 MB — 87.7% of 980 MB usage (above 85% threshold)
+    "anon 840000000\n"  # 840 MB — 87.5% of 960 MB usage (above 85% threshold)
     "file 100000000\n"
     "kernel 20000000\n"
 )
 _MOCK_MEMORY_STAT_ANON_LOW = (
-    "anon 300000000\n"  # 300 MB — 30.6% of 980 MB usage (below 85% threshold)
-    "file 650000000\n"
+    "anon 300000000\n"  # 300 MB — 31.25% of 960 MB usage (below 85% threshold)
+    "file 630000000\n"
     "kernel 30000000\n"
 )
 _MOCK_MEMORY_STAT_NO_ANON = (
-    "file 650000000\n"  # malformed: missing anon line
+    "file 630000000\n"  # malformed: missing anon line
     "kernel 30000000\n"
 )
 
 # /proc/self/status mock values (fallback when cgroup v2 memory.stat is unavailable).
-_MOCK_PROC_ANON_HIGH = "RssAnon:\t   840000 kB\n"  # ~860 MB — 87.7% of 980 MB usage
-_MOCK_PROC_ANON_LOW = "RssAnon:\t   300000 kB\n"  # ~307 MB — 31.3% of 980 MB usage
+_MOCK_PROC_ANON_HIGH = "RssAnon:\t   820313 kB\n"  # ~840 MB — 87.5% of 960 MB usage
+_MOCK_PROC_ANON_LOW = "RssAnon:\t   300000 kB\n"  # ~307 MB — 32.0% of 960 MB usage
 
 
 def _v2_exists(self: Path) -> bool:
@@ -134,12 +134,12 @@ def test_noop_when_limit_is_zero(caplog: pytest.LogCaptureFixture) -> None:
 
 
 def test_no_log_below_threshold(caplog: pytest.LogCaptureFixture) -> None:
-    """No log should be emitted when usage is below 95%."""
+    """No log should be emitted when usage is below the 90% high-pressure threshold."""
     monitor = MemoryMonitor(check_interval=1)
     with (
         caplog.at_level(logging.DEBUG, logger="airbyte"),
         patch.object(Path, "exists", _v2_exists),
-        patch.object(Path, "read_text", _v2_mock_read(usage=_MOCK_USAGE_AT_90)),
+        patch.object(Path, "read_text", _v2_mock_read(usage=_MOCK_USAGE_AT_85)),
     ):
         monitor.check_memory_usage()
     # Only the debug probe message, no info/warning
@@ -152,11 +152,11 @@ def test_no_log_below_threshold(caplog: pytest.LogCaptureFixture) -> None:
 
 
 def test_cgroup_v1_activates_high_pressure_mode(caplog: pytest.LogCaptureFixture) -> None:
-    """Memory reading works with cgroup v1 paths and activates high-pressure mode at 95%."""
+    """Memory reading works with cgroup v1 paths and activates high-pressure mode at 90%."""
 
     def mock_read_text(self: Path) -> str:
         if self == _CGROUP_V1_USAGE:
-            return _MOCK_USAGE_AT_95
+            return _MOCK_USAGE_AT_92
         if self == _CGROUP_V1_LIMIT:
             return _MOCK_LIMIT
         return ""
@@ -185,7 +185,7 @@ def test_check_interval_skips_intermediate_calls(caplog: pytest.LogCaptureFixtur
     with (
         caplog.at_level(logging.INFO, logger="airbyte"),
         patch.object(Path, "exists", _v2_exists),
-        patch.object(Path, "read_text", _v2_mock_read(usage=_MOCK_USAGE_AT_95)),
+        patch.object(Path, "read_text", _v2_mock_read(usage=_MOCK_USAGE_AT_92)),
     ):
         # First 4999 calls should be skipped
         for _ in range(4999):
@@ -290,7 +290,7 @@ def test_read_cgroup_v2_anon_bytes_parses_anon_field() -> None:
     """Correctly parses the 'anon' field from cgroup v2 memory.stat."""
     with patch.object(Path, "read_text", return_value=_MOCK_MEMORY_STAT_ANON_HIGH):
         result = _read_cgroup_v2_anon_bytes()
-    assert result == 860000000
+    assert result == 840000000
 
 
 def test_read_cgroup_v2_anon_bytes_returns_none_when_anon_absent() -> None:
@@ -314,7 +314,7 @@ def raise_oserror(self: Path) -> str:
 # ---------------------------------------------------------------------------
 
 
-def _v2_full_mock(usage: str = _MOCK_USAGE_AT_98, memory_stat: str = _MOCK_MEMORY_STAT_ANON_HIGH):
+def _v2_full_mock(usage: str = _MOCK_USAGE_AT_96, memory_stat: str = _MOCK_MEMORY_STAT_ANON_HIGH):
     """Return a mock read_text that serves cgroup v2 current/max AND memory.stat."""
 
     def mock_read_text(self: Path) -> str:
@@ -330,7 +330,7 @@ def mock_read_text(self: Path) -> str:
 
 
 def test_raises_when_cgroup_critical_and_anon_share_of_usage_above_threshold() -> None:
-    """Fail-fast raises when cgroup >= 98% and anon >= 85% of current usage."""
+    """Fail-fast raises when cgroup >= 95% and anon >= 85% of current usage."""
     monitor = MemoryMonitor(check_interval=1)
     with (
         patch.object(Path, "exists", _v2_exists),
@@ -340,14 +340,14 @@ def test_raises_when_cgroup_critical_and_anon_share_of_usage_above_threshold() -
             monitor.check_memory_usage()
     assert exc_info.value.failure_type == FailureType.system_error
     assert "critical threshold" in (exc_info.value.message or "")
-    assert "98%" in (exc_info.value.message or "")
+    assert "96%" in (exc_info.value.message or "")
     assert "anon share of usage" in (exc_info.value.internal_message or "")
 
 
 def test_no_raise_when_cgroup_critical_but_anon_share_of_usage_below_threshold(
     caplog: pytest.LogCaptureFixture,
 ) -> None:
-    """No exception when cgroup >= 98% but anon < 85% of usage; logs once then silences."""
+    """No exception when cgroup >= 95% but anon < 85% of usage; logs once then silences."""
     monitor = MemoryMonitor(check_interval=1)
     with (
         caplog.at_level(logging.INFO, logger="airbyte"),
@@ -368,7 +368,7 @@ def test_falls_back_to_process_rssanon_when_cgroup_v2_anon_unavailable() -> None
 
     def mock_read_text(self: Path) -> str:
         if self == _CGROUP_V2_CURRENT:
-            return _MOCK_USAGE_AT_98
+            return _MOCK_USAGE_AT_96
         if self == _CGROUP_V2_MAX:
             return _MOCK_LIMIT
         if self == _CGROUP_V2_STAT:
@@ -394,13 +394,13 @@ def test_falls_back_to_process_rssanon_low_and_does_not_raise(
 
     def mock_read_text(self: Path) -> str:
         if self == _CGROUP_V2_CURRENT:
-            return _MOCK_USAGE_AT_98
+            return _MOCK_USAGE_AT_96
         if self == _CGROUP_V2_MAX:
             return _MOCK_LIMIT
         if self == _CGROUP_V2_STAT:
             return _MOCK_MEMORY_STAT_NO_ANON  # anon line missing
         if self == _PROC_SELF_STATUS:
-            return _MOCK_PROC_ANON_LOW  # ~307 MB — 31.3% of 980 MB usage (below 85%)
+            return _MOCK_PROC_ANON_LOW  # ~307 MB — 32.0% of 960 MB usage (below 85%)
         return ""
 
     monitor = MemoryMonitor(check_interval=1)
@@ -421,7 +421,7 @@ def test_no_raise_when_anonymous_memory_signal_unavailable_at_critical_usage(
 
     def mock_read_text(self: Path) -> str:
         if self == _CGROUP_V2_CURRENT:
-            return _MOCK_USAGE_AT_98
+            return _MOCK_USAGE_AT_96
         if self == _CGROUP_V2_MAX:
             return _MOCK_LIMIT
         if self == _CGROUP_V2_STAT:
@@ -444,17 +444,17 @@ def mock_read_text(self: Path) -> str:
     assert monitor._critical_logged
 
 
-def test_switches_to_high_pressure_check_interval_after_crossing_95_percent(
+def test_switches_to_high_pressure_check_interval_after_crossing_90_percent(
     caplog: pytest.LogCaptureFixture,
 ) -> None:
-    """Once usage crosses 95%, the monitor tightens polling from 5000 to 100 messages."""
+    """Once usage crosses 90%, the monitor tightens polling from 5000 to 100 messages."""
     monitor = MemoryMonitor(check_interval=5000)
     assert not monitor._high_pressure_mode
 
     with (
         caplog.at_level(logging.INFO, logger="airbyte"),
         patch.object(Path, "exists", _v2_exists),
-        patch.object(Path, "read_text", _v2_mock_read(usage=_MOCK_USAGE_AT_95)),
+        patch.object(Path, "read_text", _v2_mock_read(usage=_MOCK_USAGE_AT_92)),
     ):
         # Pump 5000 messages to trigger the first real check
         for _ in range(5000):