Implement SQL storage for an "enable error recovery" scalar.

robot-server/robot_server/persistence/persistence_directory.py

@@ -52,6 +52,9 @@ async def prepare_active_subdirectory(prepared_root: Path) -> Path:
             v3_to_v4.Migration3to4(subdirectory="4"),
             v4_to_v5.Migration4to5(subdirectory="5"),
             v5_to_v6.Migration5to6(subdirectory="6"),
+            # TODO BEFORE MERGE: Double-check that I can make these changes in-place
+            # without breaking the robots that are on `edge` right now. Maybe rename
+            # LATEST_VERSION_DIRECTORY to 7.1/ and abandon the current 7/ directory.
             v6_to_v7.Migration6to7(subdirectory=LATEST_VERSION_DIRECTORY),
         ],
         temp_file_prefix="temp-",

robot-server/robot_server/persistence/tables/__init__.py

@@ -12,6 +12,7 @@
     action_table,
     run_csv_rtp_table,
     data_files_table,
+    enable_error_recovery_table,
     PrimitiveParamSQLEnum,
     ProtocolKindSQLEnum,
 )
@@ -28,6 +29,7 @@
     "action_table",
     "run_csv_rtp_table",
     "data_files_table",
+    "enable_error_recovery_table",
     "PrimitiveParamSQLEnum",
     "ProtocolKindSQLEnum",
 ]

robot-server/robot_server/persistence/tables/schema_7.py

@@ -240,6 +240,8 @@ class ProtocolKindSQLEnum(enum.Enum):
 )
 
 run_csv_rtp_table = sqlalchemy.Table(
+    # TODO BEFORE MERGE: If we need to rename the persistence directory anyway,
+    # take the opportunity to remove "table" from these SQL names?
     "run_csv_rtp_table",
     metadata,
     sqlalchemy.Column(
@@ -263,3 +265,22 @@ class ProtocolKindSQLEnum(enum.Enum):
         nullable=True,
     ),
 )
+
+# Single-row table to store the boolean "enable/disable error recovery" setting.
+enable_error_recovery_table = sqlalchemy.Table(
+    "enable_error_recovery",
+    metadata,
+    sqlalchemy.Column(
+        "id",
+        sqlalchemy.Integer,
+        # This `id=0` constraint, combined with the uniqueness constraint implicit in `primary_key=True`,
+        # is a trick to make sure this table only has at most 1 row.
+        sqlalchemy.CheckConstraint("id=0"),
+        primary_key=True,
+    ),
+    sqlalchemy.Column(
+        "enable_error_recovery",
+        sqlalchemy.Boolean,
+        nullable=True,
+    ),
+)

robot-server/robot_server/runs/error_recovery_setting_store.py

@@ -0,0 +1,34 @@
+# noqa: D100
+
+
+import sqlalchemy
+
+from robot_server.persistence.tables import enable_error_recovery_table
+
+
+class ErrorRecoverySettingStore:
+    """Persistently stores settings related to error recovery."""
+
+    def __init__(self, sql_engine: sqlalchemy.engine.Engine) -> None:
+        self._sql_engine = sql_engine
+
+    def get_is_enabled(self) -> bool | None:
+        """Get the value of the "error recovery enabled" setting.
+
+        `None` is the default, i.e. it's never been explicitly set one way or the other.
+        """
+        with self._sql_engine.begin() as transaction:
+            return transaction.execute(
+                sqlalchemy.select(enable_error_recovery_table.c.enable_error_recovery)
+            ).scalar_one_or_none()
+
+    def set_is_enabled(self, is_enabled: bool) -> None:
+        """Set the value of the "error recovery enabled" setting."""
+        with self._sql_engine.begin() as transaction:
+            transaction.execute(sqlalchemy.delete(enable_error_recovery_table))
+            transaction.execute(
+                sqlalchemy.insert(enable_error_recovery_table).values(
+                    id=0,  # id=0 to match the single-row constraint trick in the table declaration.
+                    enable_error_recovery=is_enabled,
+                )
+            )

robot-server/tests/runs/test_error_recovery_setting_store.py

@@ -0,0 +1,26 @@
+"""Tests for error_recovery_setting_store."""
+
+
+from robot_server.runs.error_recovery_setting_store import ErrorRecoverySettingStore
+
+import pytest
+import sqlalchemy
+
+
+@pytest.fixture
+def subject(
+    sql_engine: sqlalchemy.engine.Engine,
+) -> ErrorRecoverySettingStore:
+    """Return a test subject."""
+    return ErrorRecoverySettingStore(sql_engine=sql_engine)
+
+
+def test_error_recovery_setting_store(subject: ErrorRecoverySettingStore) -> None:
+    """Test `ErrorRecoverySettingStore`."""
+    assert subject.get_is_enabled() is None
+
+    subject.set_is_enabled(is_enabled=False)
+    assert subject.get_is_enabled() is False
+
+    subject.set_is_enabled(is_enabled=True)
+    assert subject.get_is_enabled() is True