fix: make DB migrations idempotent and stamp alembic after init_database

charts/model-engine/Chart.yaml

@@ -15,7 +15,7 @@ type: application
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
 # Versions are expected to follow Semantic Versioning (https://semver.org/)
-version: 0.2.7
+version: 0.2.8
 
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application. Versions are not expected to

charts/model-engine/templates/database_init_job.yaml

@@ -10,7 +10,7 @@ metadata:
     "helm.sh/hook-weight": "-2"
     "helm.sh/hook-delete-policy": before-hook-creation,hook-succeeded
 spec:
-  backoffLimit: 0
+  backoffLimit: 2
   activeDeadlineSeconds: 600
   template:
     metadata:

charts/model-engine/templates/database_migration_job.yaml

@@ -10,7 +10,7 @@ metadata:
     "helm.sh/hook-weight": "-1"
     "helm.sh/hook-delete-policy": before-hook-creation,hook-succeeded
 spec:
-  backoffLimit: 0
+  backoffLimit: 2
   activeDeadlineSeconds: 600
   template:
     metadata:

model-engine/model_engine_server/db/migrations/README

@@ -19,6 +19,32 @@ alembic revision -m “initial”
 alembic stamp fa3267c80731
 ```
 
+# Idempotency and adoption of pre-alembic databases
+
+Migrations are self-adopting and safe to re-run:
+
+1. The initial revision (fa3267c80731) checks whether
+   hosted_model_inference.endpoints already exists. If it does (e.g. the
+   database was initialized via the init_database entrypoint's create_all
+   before alembic was ever run), it skips initial.sql and just records the
+   revision instead of failing on CREATE SCHEMA.
+2. Subsequent add-column revisions check for column existence before
+   adding/dropping, so they no-op on schemas that already have the columns.
+3. The init_database entrypoint stamps `alembic head` after a successful
+   create_all, but ONLY when this run created the schema from scratch:
+   it skips stamping if the database already has an alembic revision
+   recorded (public.alembic_version_model_engine has a row), or if
+   hosted_model_inference.endpoints existed before create_all ran.
+   Stamping an already-stamped database would fast-forward the version
+   table past unapplied migrations and silently skip them; stamping a
+   pre-existing unstamped schema would over-claim, since create_all only
+   creates missing tables and never adds columns to existing ones. In both
+   cases the migration job's `alembic upgrade head` does the right thing
+   (applies pending revisions / adopts the schema via the guards above).
+
+Manual stamping (stamp_initial_schema.sh) is therefore only needed for
+databases created before this behavior existed.
+
 # Steps to make generic database schema changes
 
 Steps can be found here: https://alembic.sqlalchemy.org/en/latest/tutorial.html#running-our-second-migration

model-engine/model_engine_server/db/migrations/alembic/versions/2024_09_09_1736-fa3267c80731_initial.py

@@ -22,6 +22,17 @@
 
 
 def upgrade() -> None:
+    # If the schema already exists (e.g. the database was initialized by
+    # init_database's create_all before alembic was stamped), adopt the
+    # existing schema instead of replaying initial.sql, whose bare
+    # CREATE SCHEMA statements would fail with DuplicateSchema.
+    inspector = sa.inspect(op.get_bind())
+    if inspector.has_table("endpoints", schema="hosted_model_inference"):
+        print(
+            "Table hosted_model_inference.endpoints already exists; "
+            "adopting existing schema and skipping initial.sql."
+        )
+        return
     with open(INITIAL_MIGRATION_PATH) as fd:
         op.execute(fd.read())
 

model-engine/model_engine_server/db/migrations/alembic/versions/2024_09_09_1831-b574e9711e35_chat_completion_add_extra_routes.py

@@ -16,17 +16,24 @@
 depends_on = None
 
 
+def _has_column(table: str, column: str, schema: str) -> bool:
+    inspector = sa.inspect(op.get_bind())
+    return any(col["name"] == column for col in inspector.get_columns(table, schema=schema))
+
+
 def upgrade() -> None:
-    op.add_column(
-        "bundles",
-        sa.Column("runnable_image_extra_routes", ARRAY(sa.Text), nullable=True),
-        schema="hosted_model_inference",
-    )
+    if not _has_column("bundles", "runnable_image_extra_routes", "hosted_model_inference"):
+        op.add_column(
+            "bundles",
+            sa.Column("runnable_image_extra_routes", ARRAY(sa.Text), nullable=True),
+            schema="hosted_model_inference",
+        )
 
 
 def downgrade():
-    op.drop_column(
-        "bundles",
-        "runnable_image_extra_routes",
-        schema="hosted_model_inference",
-    )
+    if _has_column("bundles", "runnable_image_extra_routes", "hosted_model_inference"):
+        op.drop_column(
+            "bundles",
+            "runnable_image_extra_routes",
+            schema="hosted_model_inference",
+        )

model-engine/model_engine_server/db/migrations/alembic/versions/2024_09_24_1456-f55525c81eb5_multinode_bundle.py

@@ -16,27 +16,36 @@
 depends_on = None
 
 
+def _has_column(table: str, column: str, schema: str) -> bool:
+    inspector = sa.inspect(op.get_bind())
+    return any(col["name"] == column for col in inspector.get_columns(table, schema=schema))
+
+
 def upgrade() -> None:
-    op.add_column(
-        "bundles",
-        sa.Column("runnable_image_worker_command", ARRAY(sa.Text), nullable=True),
-        schema="hosted_model_inference",
-    )
-    op.add_column(
-        "bundles",
-        sa.Column("runnable_image_worker_env", sa.JSON, nullable=True),
-        schema="hosted_model_inference",
-    )
+    if not _has_column("bundles", "runnable_image_worker_command", "hosted_model_inference"):
+        op.add_column(
+            "bundles",
+            sa.Column("runnable_image_worker_command", ARRAY(sa.Text), nullable=True),
+            schema="hosted_model_inference",
+        )
+    if not _has_column("bundles", "runnable_image_worker_env", "hosted_model_inference"):
+        op.add_column(
+            "bundles",
+            sa.Column("runnable_image_worker_env", sa.JSON, nullable=True),
+            schema="hosted_model_inference",
+        )
 
 
 def downgrade() -> None:
-    op.drop_column(
-        "bundles",
-        "runnable_image_worker_command",
-        schema="hosted_model_inference",
-    )
-    op.drop_column(
-        "bundles",
-        "runnable_image_worker_env",
-        schema="hosted_model_inference",
-    )
+    if _has_column("bundles", "runnable_image_worker_command", "hosted_model_inference"):
+        op.drop_column(
+            "bundles",
+            "runnable_image_worker_command",
+            schema="hosted_model_inference",
+        )
+    if _has_column("bundles", "runnable_image_worker_env", "hosted_model_inference"):
+        op.drop_column(
+            "bundles",
+            "runnable_image_worker_env",
+            schema="hosted_model_inference",
+        )

model-engine/model_engine_server/db/migrations/alembic/versions/2025_09_16_1741-e580182d6bfd_add_passthrough_forwarder.py

@@ -15,17 +15,24 @@
 depends_on = None
 
 
+def _has_column(table: str, column: str, schema: str) -> bool:
+    inspector = sa.inspect(op.get_bind())
+    return any(col["name"] == column for col in inspector.get_columns(table, schema=schema))
+
+
 def upgrade() -> None:
-    op.add_column(
-        "bundles",
-        sa.Column("runnable_image_forwarder_type", sa.String(), nullable=True),
-        schema="hosted_model_inference",
-    )
+    if not _has_column("bundles", "runnable_image_forwarder_type", "hosted_model_inference"):
+        op.add_column(
+            "bundles",
+            sa.Column("runnable_image_forwarder_type", sa.String(), nullable=True),
+            schema="hosted_model_inference",
+        )
 
 
 def downgrade() -> None:
-    op.drop_column(
-        "bundles",
-        "runnable_image_forwarder_type",
-        schema="hosted_model_inference",
-    )
+    if _has_column("bundles", "runnable_image_forwarder_type", "hosted_model_inference"):
+        op.drop_column(
+            "bundles",
+            "runnable_image_forwarder_type",
+            schema="hosted_model_inference",
+        )

model-engine/model_engine_server/db/migrations/alembic/versions/2025_09_25_1940-221aa19d3f32_add_routes_column.py

@@ -15,17 +15,24 @@
 depends_on = None
 
 
+def _has_column(table: str, column: str, schema: str) -> bool:
+    inspector = sa.inspect(op.get_bind())
+    return any(col['name'] == column for col in inspector.get_columns(table, schema=schema))
+
+
 def upgrade() -> None:
-    op.add_column(
-        'bundles',
-        sa.Column('runnable_image_routes', sa.ARRAY(sa.Text), nullable=True),
-        schema='hosted_model_inference',
-    )
+    if not _has_column('bundles', 'runnable_image_routes', 'hosted_model_inference'):
+        op.add_column(
+            'bundles',
+            sa.Column('runnable_image_routes', sa.ARRAY(sa.Text), nullable=True),
+            schema='hosted_model_inference',
+        )
 
 
 def downgrade() -> None:
-    op.drop_column(
-        'bundles',
-        'runnable_image_routes',
-        schema='hosted_model_inference',
-    )
+    if _has_column('bundles', 'runnable_image_routes', 'hosted_model_inference'):
+        op.drop_column(
+            'bundles',
+            'runnable_image_routes',
+            schema='hosted_model_inference',
+        )

model-engine/model_engine_server/db/migrations/alembic/versions/2026_02_10_1920-62da4f8b3403_add_task_expires_seconds_column.py

@@ -15,17 +15,24 @@
 depends_on = None
 
 
+def _has_column(table: str, column: str, schema: str) -> bool:
+    inspector = sa.inspect(op.get_bind())
+    return any(col['name'] == column for col in inspector.get_columns(table, schema=schema))
+
+
 def upgrade() -> None:
-    op.add_column(
-        'endpoints',
-        sa.Column('task_expires_seconds', sa.Integer, nullable=True),
-        schema='hosted_model_inference',
-    )
+    if not _has_column('endpoints', 'task_expires_seconds', 'hosted_model_inference'):
+        op.add_column(
+            'endpoints',
+            sa.Column('task_expires_seconds', sa.Integer, nullable=True),
+            schema='hosted_model_inference',
+        )
 
 
 def downgrade() -> None:
-    op.drop_column(
-        'endpoints',
-        'task_expires_seconds',
-        schema='hosted_model_inference',
-    )
+    if _has_column('endpoints', 'task_expires_seconds', 'hosted_model_inference'):
+        op.drop_column(
+            'endpoints',
+            'task_expires_seconds',
+            schema='hosted_model_inference',
+        )

model-engine/model_engine_server/db/migrations/alembic/versions/2026_02_20_1200-a1b2c3d4e5f6_add_queue_message_timeout_seconds_column.py

@@ -15,17 +15,24 @@
 depends_on = None
 
 
+def _has_column(table: str, column: str, schema: str) -> bool:
+    inspector = sa.inspect(op.get_bind())
+    return any(col['name'] == column for col in inspector.get_columns(table, schema=schema))
+
+
 def upgrade() -> None:
-    op.add_column(
-        'endpoints',
-        sa.Column('queue_message_timeout_seconds', sa.Integer, nullable=True),
-        schema='hosted_model_inference',
-    )
+    if not _has_column('endpoints', 'queue_message_timeout_seconds', 'hosted_model_inference'):
+        op.add_column(
+            'endpoints',
+            sa.Column('queue_message_timeout_seconds', sa.Integer, nullable=True),
+            schema='hosted_model_inference',
+        )
 
 
 def downgrade() -> None:
-    op.drop_column(
-        'endpoints',
-        'queue_message_timeout_seconds',
-        schema='hosted_model_inference',
-    )
+    if _has_column('endpoints', 'queue_message_timeout_seconds', 'hosted_model_inference'):
+        op.drop_column(
+            'endpoints',
+            'queue_message_timeout_seconds',
+            schema='hosted_model_inference',
+        )

model-engine/model_engine_server/db/migrations/alembic/versions/2026_06_16_1200-c4d5e6f7a8b9_add_status_reason_column.py

@@ -15,17 +15,24 @@
 depends_on = None
 
 
+def _has_column(table: str, column: str, schema: str) -> bool:
+    inspector = sa.inspect(op.get_bind())
+    return any(col['name'] == column for col in inspector.get_columns(table, schema=schema))
+
+
 def upgrade() -> None:
-    op.add_column(
-        'endpoints',
-        sa.Column('status_reason', sa.Text, nullable=True),
-        schema='hosted_model_inference',
-    )
+    if not _has_column('endpoints', 'status_reason', 'hosted_model_inference'):
+        op.add_column(
+            'endpoints',
+            sa.Column('status_reason', sa.Text, nullable=True),
+            schema='hosted_model_inference',
+        )
 
 
 def downgrade() -> None:
-    op.drop_column(
-        'endpoints',
-        'status_reason',
-        schema='hosted_model_inference',
-    )
+    if _has_column('endpoints', 'status_reason', 'hosted_model_inference'):
+        op.drop_column(
+            'endpoints',
+            'status_reason',
+            schema='hosted_model_inference',
+        )