fix(checkpointer): clear stale cubepi_schema_version rows on upgrade

INSERT ... ON CONFLICT DO NOTHING leaves prior version rows behind
because version is the primary key. _verify_schema does
SELECT version ... LIMIT 1 and could read the stale row, triggering a
spurious CubepiSchemaMismatch on upgrade. DELETE any non-current row
before INSERT — still idempotent.

Author: xfgong

cubepi/checkpointer/postgres/alembic_helpers.py

@@ -21,12 +21,19 @@ def create_message_partitions_op() -> str:
 
 
 def write_schema_version_op() -> str:
-    """Return SQL inserting the current schema version.
+    """Return SQL setting cubepi_schema_version to the current version.
 
     Call inside alembic upgrade() after CREATE TABLE cubepi_schema_version.
-    Idempotent via ON CONFLICT DO NOTHING.
+
+    Atomically clears any stale rows (from prior cubepi versions) and writes
+    the current version. ``version`` is the primary key, so a plain INSERT
+    would leave older rows in place and ``_verify_schema`` (which does
+    ``SELECT version ... LIMIT 1``) could read a stale value and falsely
+    report CubepiSchemaMismatch. Idempotent under repeated execution.
     """
     return (
+        f"DELETE FROM cubepi_schema_version "
+        f"WHERE version <> {EXPECTED_SCHEMA_VERSION}; "
         f"INSERT INTO cubepi_schema_version (version) "
         f"VALUES ({EXPECTED_SCHEMA_VERSION}) ON CONFLICT DO NOTHING;"
     )

tests/checkpointer/test_postgres.py

@@ -79,6 +79,18 @@ def test_write_schema_version_op_includes_expected_version() -> None:
     assert "ON CONFLICT" in sql
 
 
+def test_write_schema_version_op_clears_stale_rows() -> None:
+    """A prior version's row must be removed so _verify_schema sees the new one."""
+    from cubepi.checkpointer.postgres.alembic_helpers import write_schema_version_op
+
+    sql = write_schema_version_op()
+    # Must DELETE rows whose version is not the expected one before INSERT.
+    assert "DELETE FROM cubepi_schema_version" in sql
+    assert "WHERE version <> 1" in sql
+    # And the DELETE must come before the INSERT in the statement order.
+    assert sql.index("DELETE") < sql.index("INSERT")
+
+
 def test_schema_uninitialized_is_schema_error() -> None:
     from cubepi.checkpointer.postgres.exceptions import (
         CubepiSchemaError,