Tolerate read-only database in API and content worker heartbeats

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.

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):

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:

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,16 +57,46 @@ 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
         )
     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):

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

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()