From 51d11283de9727788e93af248d40675765381891 Mon Sep 17 00:00:00 2001 From: Lucas Cabello Date: Mon, 27 Jul 2026 11:29:56 -0400 Subject: [PATCH] fix: Tolerate read-only database in API and content worker heartbeats During zero-downtime deployments the database may temporarily switch to read-only mode. The heartbeat mechanism treated all write failures as fatal, causing API and content pods to crash-loop even though they could still serve read traffic (collection downloads, EE pulls, API GETs). Detect PostgreSQL SQLSTATE 25006 (ReadOnlySqlTransaction) and skip the heartbeat write instead of killing the worker. When the database becomes writable again, normal heartbeat behaviour resumes automatically. Fixes: AAP-80698 AI-ASSISTED-BY: Claude Opus 4.6 (1M context) Co-Authored-By: Claude Opus 4.6 (1M context) --- CHANGES/0000.bugfix | 3 + pulpcore/app/entrypoint.py | 44 +++++++++- pulpcore/app/util.py | 11 +++ pulpcore/content/__init__.py | 43 ++++++++- pulpcore/tests/unit/content/test_heartbeat.py | 77 +++++++++++++++- pulpcore/tests/unit/test_entrypoint.py | 88 +++++++++++++++++++ pulpcore/tests/unit/test_util.py | 51 +++++++++++ 7 files changed, 311 insertions(+), 6 deletions(-) create mode 100644 CHANGES/0000.bugfix create mode 100644 pulpcore/tests/unit/test_entrypoint.py diff --git a/CHANGES/0000.bugfix b/CHANGES/0000.bugfix new file mode 100644 index 00000000000..9d7bd7acb7c --- /dev/null +++ b/CHANGES/0000.bugfix @@ -0,0 +1,3 @@ +API and content workers now tolerate a read-only database by skipping heartbeat writes +instead of crash-looping. Workers continue serving read traffic and resume normal +heartbeats once the database becomes writable. diff --git a/pulpcore/app/entrypoint.py b/pulpcore/app/entrypoint.py index ab76e7e0fec..5f2605db15b 100644 --- a/pulpcore/app/entrypoint.py +++ b/pulpcore/app/entrypoint.py @@ -18,6 +18,7 @@ PulpcoreGunicornApplication, handle_control_interface_feature, ) +from pulpcore.app.util import _is_read_only_db_error logger = getLogger(__name__) @@ -52,6 +53,27 @@ def notify(self): os._exit(Arbiter.WORKER_BOOT_ERROR) def heartbeat(self): + if self.app_status is None: + from pulpcore.app.models import AppStatus + + try: + self.app_status = AppStatus.objects.create( + name=self.name, app_type="api", versions=self.versions + ) + except IntegrityError: + logger.error( + f"An API app with name {self.name} already exists in the database." + ) + raise + except (InterfaceError, DatabaseError) as e: + if _is_read_only_db_error(e): + connection.close_if_unusable_or_obsolete() + logger.debug( + f"Api App '{self.name}' heartbeat skipped (read-only transaction)." + ) + return + raise + try: self.app_status.save_heartbeat() logger.debug(self.beat_msg) @@ -61,6 +83,11 @@ def heartbeat(self): self.app_status.save_heartbeat() logger.debug(self.beat_msg) except (InterfaceError, DatabaseError) as e: + if _is_read_only_db_error(e): + logger.warning( + f"Api App '{self.name}' heartbeat skipped (read-only transaction)." + ) + return logger.error(f"{self.fail_beat_msg} Exception: {str(e)}") raise @@ -101,6 +128,15 @@ def init_process(self): except IntegrityError: logger.error(f"An API app with name {self.name} already exists in the database.") exit(Arbiter.WORKER_BOOT_ERROR) + except DatabaseError as e: + if _is_read_only_db_error(e): + logger.warning( + f"Api App '{self.name}' could not register in the database " + "(read-only transaction). Heartbeat will retry." + ) + self.app_status = None + else: + raise self.heartbeat_interval = settings.API_APP_TTL // 3 @@ -115,9 +151,13 @@ def run(self): try: super().run() finally: - # cleanup if self.app_status: - self.app_status.delete() + try: + self.app_status.delete() + except (InterfaceError, DatabaseError): + logger.info( + f"Api App '{self.name}' could not clean up its status record." + ) class PulpcoreApiApplication(PulpcoreGunicornApplication): diff --git a/pulpcore/app/util.py b/pulpcore/app/util.py index 76f2a5b47fb..66e597d6076 100644 --- a/pulpcore/app/util.py +++ b/pulpcore/app/util.py @@ -677,6 +677,17 @@ def get_worker_name(): ) +def _is_read_only_db_error(exc): + """Check if a DatabaseError was caused by a read-only database (SQLSTATE 25006).""" + cause = exc.__cause__ + while cause is not None: + pgcode = getattr(cause, "pgcode", None) or getattr(cause, "sqlstate", None) + if pgcode == "25006": + return True + cause = getattr(cause, "__cause__", None) + return False + + def normalize_http_status(status): """Convert the HTTP status code to 2xx, 3xx, etc., normalizing the last two digits.""" if 100 <= status < 200: diff --git a/pulpcore/content/__init__.py b/pulpcore/content/__init__.py index decf0957b0d..0464bc356d7 100644 --- a/pulpcore/content/__init__.py +++ b/pulpcore/content/__init__.py @@ -22,7 +22,7 @@ from pulpcore.app.apps import pulp_plugin_configs # noqa: E402 from pulpcore.app.models import AppStatus # noqa: E402 -from pulpcore.app.util import get_worker_name # noqa: E402 +from pulpcore.app.util import _is_read_only_db_error, get_worker_name # noqa: E402 from .authentication import authenticate, guid # noqa: E402 from .handler import Handler # noqa: E402 @@ -57,6 +57,7 @@ async def _heartbeat(): ) versions = {app.label: app.version for app in pulp_plugin_configs()} + app_status = None try: app_status = await AppStatus.objects.acreate( name=name, app_type="content", versions=versions @@ -64,9 +65,38 @@ async def _heartbeat(): except IntegrityError: log.error(f"A content app with name {name} already exists in the database.") exit(Arbiter.WORKER_BOOT_ERROR) + except DatabaseError as e: + if _is_read_only_db_error(e): + log.warning( + f"Content App '{name}' could not register in the database " + "(read-only transaction). Heartbeat will retry." + ) + else: + raise + try: while True: await asyncio.sleep(heartbeat_interval) + + if app_status is None: + try: + app_status = await AppStatus.objects.acreate( + name=name, app_type="content", versions=versions + ) + except IntegrityError: + log.error( + f"A content app with name {name} already exists in the database." + ) + raise + except (InterfaceError, DatabaseError) as e: + if _is_read_only_db_error(e): + await sync_to_async(Handler._reset_db_connection)() + log.debug( + f"Content App '{name}' heartbeat skipped (read-only transaction)." + ) + continue + raise + try: await app_status.asave_heartbeat() log.debug(msg) @@ -76,11 +106,20 @@ async def _heartbeat(): await app_status.asave_heartbeat() log.debug(msg) except (InterfaceError, DatabaseError) as e: + if _is_read_only_db_error(e): + log.warning( + f"Content App '{name}' heartbeat skipped " + "(read-only transaction)." + ) + continue log.error(f"{fail_msg} Exception: {str(e)}") exit(Arbiter.WORKER_BOOT_ERROR) finally: if app_status: - await app_status.adelete() + try: + await app_status.adelete() + except (InterfaceError, DatabaseError): + log.info(f"Content App '{name}' could not clean up its status record.") async def _heartbeat_ctx(app): diff --git a/pulpcore/tests/unit/content/test_heartbeat.py b/pulpcore/tests/unit/content/test_heartbeat.py index 5fff2076431..9de5e48b9a3 100644 --- a/pulpcore/tests/unit/content/test_heartbeat.py +++ b/pulpcore/tests/unit/content/test_heartbeat.py @@ -1,7 +1,7 @@ -from unittest.mock import AsyncMock, Mock, call +from unittest.mock import AsyncMock, Mock, call, patch import pytest -from django.db.utils import InterfaceError, OperationalError +from django.db.utils import InternalError, InterfaceError, OperationalError from pulpcore.app.models.status import AppStatus, AppStatusManager from pulpcore.content import _heartbeat @@ -12,6 +12,14 @@ class MockException(Exception): pass +def _make_read_only_error(): + cause = Exception("cannot execute UPDATE in a read-only transaction") + cause.pgcode = "25006" + exc = InternalError("read-only") + exc.__cause__ = cause + return exc + + @pytest.mark.parametrize("error_class", [InterfaceError, OperationalError]) @pytest.mark.asyncio async def test_db_connection_interface_error(monkeypatch, settings, error_class): @@ -35,3 +43,68 @@ async def test_db_connection_interface_error(monkeypatch, settings, error_class) mock_app_status.asave_heartbeat.assert_called() mock_reset_db.assert_has_calls([call()]) + + +@pytest.mark.asyncio +async def test_read_only_heartbeat_skips_without_exit(monkeypatch, settings): + """ + When the DB is read-only, heartbeat should skip without exiting. + """ + read_only_exc = _make_read_only_error() + + mock_app_status = AsyncMock() + mock_app_status.asave_heartbeat.side_effect = [read_only_exc, read_only_exc] + mock_acreate = AsyncMock() + mock_acreate.return_value = mock_app_status + monkeypatch.setattr(AppStatusManager, "acreate", mock_acreate) + monkeypatch.setattr(AppStatus, "objects", AppStatusManager()) + mock_reset_db = Mock() + monkeypatch.setattr(Handler, "_reset_db_connection", mock_reset_db) + settings.CONTENT_APP_TTL = 1 + + iteration_count = 0 + + original_sleep = __import__("asyncio").sleep + + async def counting_sleep(seconds): + nonlocal iteration_count + iteration_count += 1 + if iteration_count >= 2: + raise KeyboardInterrupt("stop test loop") + await original_sleep(0) + + with patch("asyncio.sleep", side_effect=counting_sleep): + with pytest.raises(KeyboardInterrupt): + await _heartbeat() + + mock_app_status.asave_heartbeat.assert_called() + mock_reset_db.assert_called() + + +@pytest.mark.asyncio +async def test_read_only_acreate_skips_without_exit(monkeypatch, settings): + """ + When acreate fails with read-only, _heartbeat should not exit. + """ + read_only_exc = _make_read_only_error() + + mock_acreate = AsyncMock(side_effect=read_only_exc) + monkeypatch.setattr(AppStatusManager, "acreate", mock_acreate) + monkeypatch.setattr(AppStatus, "objects", AppStatusManager()) + mock_reset_db = Mock() + monkeypatch.setattr(Handler, "_reset_db_connection", mock_reset_db) + settings.CONTENT_APP_TTL = 1 + + iteration_count = 0 + + async def counting_sleep(seconds): + nonlocal iteration_count + iteration_count += 1 + if iteration_count >= 2: + raise KeyboardInterrupt("stop test loop") + + with patch("asyncio.sleep", side_effect=counting_sleep): + with pytest.raises(KeyboardInterrupt): + await _heartbeat() + + assert mock_acreate.call_count >= 2 diff --git a/pulpcore/tests/unit/test_entrypoint.py b/pulpcore/tests/unit/test_entrypoint.py new file mode 100644 index 00000000000..fc0a5b53d54 --- /dev/null +++ b/pulpcore/tests/unit/test_entrypoint.py @@ -0,0 +1,88 @@ +from unittest.mock import MagicMock, patch + +import pytest +from django.db.utils import DatabaseError, InternalError + +from pulpcore.app.entrypoint import PulpApiWorker + +_mock_connection = patch("pulpcore.app.entrypoint.connection") + + +def _make_read_only_error(): + cause = Exception("cannot execute UPDATE in a read-only transaction") + cause.pgcode = "25006" + exc = InternalError("read-only") + exc.__cause__ = cause + return exc + + +def _make_worker(): + with patch.object(PulpApiWorker, "__init__", lambda self, *a, **kw: None): + worker = PulpApiWorker() + worker.name = "test-worker@host" + worker.versions = {"core": "3.115.0"} + worker.app_status = MagicMock() + worker.beat_msg = "heartbeat ok" + worker.fail_beat_msg = "heartbeat failed" + return worker + + +@_mock_connection +class TestHeartbeatReadOnly: + def test_heartbeat_skips_on_read_only(self, _conn): + worker = _make_worker() + read_only_exc = _make_read_only_error() + worker.app_status.save_heartbeat.side_effect = [read_only_exc, read_only_exc] + + worker.heartbeat() + + assert worker.app_status.save_heartbeat.call_count == 2 + + def test_heartbeat_raises_on_real_db_error(self, _conn): + worker = _make_worker() + db_error = DatabaseError("connection lost") + worker.app_status.save_heartbeat.side_effect = [db_error, db_error] + + with pytest.raises(DatabaseError): + worker.heartbeat() + + def test_heartbeat_retries_create_when_app_status_none(self, _conn): + worker = _make_worker() + worker.app_status = None + read_only_exc = _make_read_only_error() + + with patch("pulpcore.app.models.AppStatus") as mock_cls: + mock_cls.objects.create.side_effect = read_only_exc + worker.heartbeat() + + assert worker.app_status is None + + def test_heartbeat_creates_app_status_when_db_recovers(self, _conn): + worker = _make_worker() + worker.app_status = None + mock_status = MagicMock() + + with patch("pulpcore.app.models.AppStatus") as mock_cls: + mock_cls.objects.create.return_value = mock_status + worker.heartbeat() + + assert worker.app_status is mock_status + mock_status.save_heartbeat.assert_called_once() + + +class TestRunCleanup: + def test_run_finally_tolerates_db_error(self): + worker = _make_worker() + worker.app_status.delete.side_effect = DatabaseError("read-only") + + with patch.object(PulpApiWorker.__bases__[0], "run"): + worker.run() + + worker.app_status.delete.assert_called_once() + + def test_run_finally_skips_when_no_app_status(self): + worker = _make_worker() + worker.app_status = None + + with patch.object(PulpApiWorker.__bases__[0], "run"): + worker.run() diff --git a/pulpcore/tests/unit/test_util.py b/pulpcore/tests/unit/test_util.py index 6b2943ceae3..6f9172b3eb4 100644 --- a/pulpcore/tests/unit/test_util.py +++ b/pulpcore/tests/unit/test_util.py @@ -222,3 +222,54 @@ def test_large_chunk_rotation(self) -> None: self.assertEqual(len(writer.results), 4) self.assertEqual((self.test_dir / "export.tar.0003").stat().st_size, 5) + + +class TestIsReadOnlyDbError: + @staticmethod + def _make_cause(pgcode=None, sqlstate=None): + cause = Exception("db error") + if pgcode is not None: + cause.pgcode = pgcode + if sqlstate is not None: + cause.sqlstate = sqlstate + return cause + + def test_pgcode_25006(self): + from pulpcore.app.util import _is_read_only_db_error + + cause = self._make_cause(pgcode="25006") + exc = Exception("read-only") + exc.__cause__ = cause + assert _is_read_only_db_error(exc) is True + + def test_sqlstate_25006(self): + from pulpcore.app.util import _is_read_only_db_error + + cause = self._make_cause(sqlstate="25006") + exc = Exception("read-only") + exc.__cause__ = cause + assert _is_read_only_db_error(exc) is True + + def test_no_cause(self): + from pulpcore.app.util import _is_read_only_db_error + + exc = Exception("connection lost") + assert _is_read_only_db_error(exc) is False + + def test_different_pgcode(self): + from pulpcore.app.util import _is_read_only_db_error + + cause = self._make_cause(pgcode="08006") + exc = Exception("connection failure") + exc.__cause__ = cause + assert _is_read_only_db_error(exc) is False + + def test_nested_cause(self): + from pulpcore.app.util import _is_read_only_db_error + + inner = self._make_cause(pgcode="25006") + outer = Exception("wrapper") + outer.__cause__ = inner + exc = Exception("nested") + exc.__cause__ = outer + assert _is_read_only_db_error(exc) is True