This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 2fb232178f1 API: Return 503 when SQLite locks during backfill creation
(#67900)
2fb232178f1 is described below
commit 2fb232178f10c49b6ba6d5125280c87309913345
Author: Lohit Kolluri <[email protected]>
AuthorDate: Thu Jul 9 19:36:35 2026 +0530
API: Return 503 when SQLite locks during backfill creation (#67900)
SQLite backfill creation via the REST API could return HTTP 500 and
leave an orphan Backfill row when the scheduler and API server contend
for the database, blocking retries with 409. Map lock errors to HTTP 503
with a clear message and clean up partial DagRuns and TaskInstances so
users can retry against a clean slate.
Signed-off-by: Lohit Kolluri <[email protected]>
---
.../core_api/openapi/v2-rest-api-generated.yaml | 12 ++
.../core_api/routes/public/backfills.py | 42 +++-
airflow-core/src/airflow/models/backfill.py | 94 +++++----
.../ui/openapi-gen/requests/services.gen.ts | 6 +-
.../airflow/ui/openapi-gen/requests/types.gen.ts | 8 +
airflow-core/src/airflow/utils/sqlalchemy.py | 4 +
.../core_api/routes/public/test_backfills.py | 221 ++++++++++++++++++++-
7 files changed, 344 insertions(+), 43 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
index 97ae27faff5..b848b996298 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
+++
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
@@ -1173,6 +1173,12 @@ paths:
schema:
$ref: '#/components/schemas/HTTPExceptionResponse'
description: Conflict
+ '503':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Service Unavailable
'422':
description: Validation Error
content:
@@ -1432,6 +1438,12 @@ paths:
application/json:
schema:
$ref: '#/components/schemas/HTTPExceptionResponse'
+ '503':
+ description: Service Unavailable
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
'422':
description: Validation Error
content:
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py
index dcfba919465..c4971b883a1 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py
@@ -16,12 +16,13 @@
# under the License.
from __future__ import annotations
-from typing import Annotated
+from typing import Annotated, NoReturn
from fastapi import Depends, HTTPException, status
from fastapi.exceptions import RequestValidationError
from pydantic import NonNegativeInt
from sqlalchemy import select, update
+from sqlalchemy.exc import OperationalError
from sqlalchemy.orm import joinedload
from airflow._shared.timezones import timezone
@@ -60,11 +61,25 @@ from airflow.models.backfill import (
_create_backfill,
_do_dry_run,
)
+from airflow.utils.sqlalchemy import is_lock_not_available_error
from airflow.utils.state import DagRunState
backfills_router = AirflowRouter(tags=["Backfill"], prefix="/backfills")
+def _raise_locked_response_or_reraise(e: OperationalError, action: str) ->
NoReturn:
+ """Map a database lock OperationalError to HTTP 503, or re-raise if not a
lock error."""
+ if not is_lock_not_available_error(e):
+ raise
+ raise HTTPException(
+ status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+ detail=(
+ f"Database is locked. Backfill {action} is not supported on SQLite
"
+ "under concurrent access. Please use PostgreSQL or MySQL."
+ ),
+ )
+
+
@backfills_router.get(
path="",
dependencies=[
@@ -219,7 +234,12 @@ def cancel_backfill(backfill_id: NonNegativeInt, session:
SessionDep) -> Backfil
@backfills_router.post(
path="",
responses=create_openapi_http_exception_doc(
- [status.HTTP_400_BAD_REQUEST, status.HTTP_404_NOT_FOUND,
status.HTTP_409_CONFLICT]
+ [
+ status.HTTP_400_BAD_REQUEST,
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_409_CONFLICT,
+ status.HTTP_503_SERVICE_UNAVAILABLE,
+ ]
),
dependencies=[
Depends(action_logging()),
@@ -252,6 +272,8 @@ def create_backfill(
run_on_latest_version=resolved_run_on_latest,
)
return BackfillResponse.model_validate(backfill_obj)
+ except OperationalError as e:
+ _raise_locked_response_or_reraise(e, "creation")
except AlreadyRunningBackfill:
raise HTTPException(
@@ -283,7 +305,13 @@ def create_backfill(
@backfills_router.post(
path="/dry_run",
- responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND,
status.HTTP_409_CONFLICT]),
+ responses=create_openapi_http_exception_doc(
+ [
+ status.HTTP_404_NOT_FOUND,
+ status.HTTP_409_CONFLICT,
+ status.HTTP_503_SERVICE_UNAVAILABLE,
+ ]
+ ),
dependencies=[
Depends(requires_access_backfill(method="POST")),
],
@@ -307,13 +335,15 @@ def create_backfill_dry_run(
)
backfills = [
DryRunBackfillResponse(
- logical_date=d.logical_date, partition_key=d.partition_key,
partition_date=d.partition_date
+ logical_date=d.logical_date,
+ partition_key=d.partition_key,
+ partition_date=d.partition_date,
)
for d in backfills_dry_run
]
-
return DryRunBackfillCollectionResponse(backfills=backfills,
total_entries=len(backfills))
-
+ except OperationalError as e:
+ _raise_locked_response_or_reraise(e, "dry-run")
except DagNotFound:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
diff --git a/airflow-core/src/airflow/models/backfill.py
b/airflow-core/src/airflow/models/backfill.py
index 1f507de94bb..13b9e6d004f 100644
--- a/airflow-core/src/airflow/models/backfill.py
+++ b/airflow-core/src/airflow/models/backfill.py
@@ -36,17 +36,18 @@ from sqlalchemy import (
Integer,
String,
UniqueConstraint,
+ delete,
func,
select,
)
-from sqlalchemy.exc import IntegrityError
+from sqlalchemy.exc import IntegrityError, OperationalError
from sqlalchemy.orm import Mapped, mapped_column, relationship, validates
from airflow._shared.timezones import timezone
from airflow.exceptions import AirflowException, DagNotFound,
DagRunTypeNotAllowed
from airflow.models.base import Base, StringID
from airflow.utils.session import create_session
-from airflow.utils.sqlalchemy import UtcDateTime, with_row_locks
+from airflow.utils.sqlalchemy import UtcDateTime, is_lock_not_available_error,
with_row_locks
from airflow.utils.state import DagRunState
from airflow.utils.types import DagRunTriggeredByType, DagRunType
@@ -687,7 +688,7 @@ def _create_backfill(
f"No runs to create for Dag {dag_id} in the range
[{from_date}, {to_date}]"
)
- br = Backfill(
+ backfill = Backfill(
dag_id=dag_id,
from_date=from_date,
to_date=to_date,
@@ -697,33 +698,58 @@ def _create_backfill(
dag_model=dag,
triggering_user_name=triggering_user_name,
)
- session.add(br)
+ session.add(backfill)
+ # Commit immediately so the backfill is visible to concurrent requests
+ # checking num_active backfills, preventing duplicate active backfills
+ # for the same dag.
session.commit()
session.scalars(select(DagModel).where(DagModel.dag_id ==
dag_id)).one()
first_info = dagrun_info_list[0]
- if first_info.partition_key:
- _create_runs_partitioned(
- br=br,
- dag=dag,
- dagrun_info_list=dagrun_info_list,
- session=session,
- )
- else:
- _create_runs_non_partitioned(
- br=br,
- dag=dag,
- dagrun_info_list=dagrun_info_list,
- run_on_latest_version=run_on_latest_version,
- session=session,
- )
- return br
+ try:
+ if first_info.partition_key:
+ _create_runs_partitioned(
+ backfill=backfill,
+ dag=dag,
+ dagrun_info_list=dagrun_info_list,
+ session=session,
+ )
+ else:
+ _create_runs_non_partitioned(
+ backfill=backfill,
+ dag=dag,
+ dagrun_info_list=dagrun_info_list,
+ run_on_latest_version=run_on_latest_version,
+ session=session,
+ )
+ except OperationalError as e:
+ if is_lock_not_available_error(e):
+ # Lock error: clean up the orphan so the user can retry. The
+ # helper is best-effort; if it fails the original error still
+ # surfaces and the route returns 503.
+ _cleanup_partial_backfill(backfill, session)
+ raise
+ return backfill
+
+
+def _cleanup_partial_backfill(backfill: Backfill, session: Session) -> None:
+ """Best-effort removal of a partially-created backfill after a lock
error."""
+ from airflow.models.dagrun import DagRun
+
+ try:
+ session.rollback()
+
session.execute(delete(BackfillDagRun).where(BackfillDagRun.backfill_id ==
backfill.id))
+ session.execute(delete(DagRun).where(DagRun.backfill_id ==
backfill.id))
+ session.delete(backfill)
+ session.commit()
+ except Exception:
+ session.rollback()
def _create_runs_partitioned(
*,
- br: Backfill,
+ backfill: Backfill,
dag: SerializedDAG,
dagrun_info_list: list[DagRunInfo],
session: Session,
@@ -735,24 +761,24 @@ def _create_runs_partitioned(
_create_backfill_dag_run_partitioned(
dag=dag,
info=info,
- backfill_id=br.id,
- dag_run_conf=br.dag_run_conf,
- reprocess_behavior=ReprocessBehavior(br.reprocess_behavior),
+ backfill_id=backfill.id,
+ dag_run_conf=backfill.dag_run_conf,
+ reprocess_behavior=ReprocessBehavior(backfill.reprocess_behavior),
backfill_sort_ordinal=backfill_sort_ordinal,
- triggering_user_name=br.triggering_user_name,
+ triggering_user_name=backfill.triggering_user_name,
session=session,
)
log.info(
"Created backfill Dag run.",
dag_id=dag.dag_id,
- backfill_id=br.id,
- info=info,
+ backfill_id=backfill.id,
+ logical_date=info.logical_date,
)
def _create_runs_non_partitioned(
*,
- br: Backfill,
+ backfill: Backfill,
dag: SerializedDAG,
dagrun_info_list: list[DagRunInfo],
run_on_latest_version: bool,
@@ -766,17 +792,17 @@ def _create_runs_non_partitioned(
_create_backfill_dag_run_non_partitioned(
dag=dag,
info=info,
- backfill_id=br.id,
- dag_run_conf=br.dag_run_conf,
- reprocess_behavior=ReprocessBehavior(br.reprocess_behavior),
+ backfill_id=backfill.id,
+ dag_run_conf=backfill.dag_run_conf,
+ reprocess_behavior=ReprocessBehavior(backfill.reprocess_behavior),
backfill_sort_ordinal=backfill_sort_ordinal,
- triggering_user_name=br.triggering_user_name,
+ triggering_user_name=backfill.triggering_user_name,
run_on_latest_version=run_on_latest_version,
session=session,
)
log.info(
"Created backfill Dag run.",
dag_id=dag.dag_id,
- backfill_id=br.id,
- info=info,
+ backfill_id=backfill.id,
+ logical_date=info.logical_date,
)
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
index 78241796546..b674a72ccbd 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
@@ -479,7 +479,8 @@ export class BackfillService {
403: 'Forbidden',
404: 'Not Found',
409: 'Conflict',
- 422: 'Validation Error'
+ 422: 'Validation Error',
+ 503: 'Service Unavailable'
}
});
}
@@ -597,7 +598,8 @@ export class BackfillService {
403: 'Forbidden',
404: 'Not Found',
409: 'Conflict',
- 422: 'Validation Error'
+ 422: 'Validation Error',
+ 503: 'Service Unavailable'
}
});
}
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
index e1bf23b1602..d1a91b68a69 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -5164,6 +5164,10 @@ export type $OpenApiTs = {
* Validation Error
*/
422: HTTPValidationError;
+ /**
+ * Service Unavailable
+ */
+ 503: HTTPExceptionResponse;
};
};
};
@@ -5315,6 +5319,10 @@ export type $OpenApiTs = {
* Validation Error
*/
422: HTTPValidationError;
+ /**
+ * Service Unavailable
+ */
+ 503: HTTPExceptionResponse;
};
};
};
diff --git a/airflow-core/src/airflow/utils/sqlalchemy.py
b/airflow-core/src/airflow/utils/sqlalchemy.py
index 095d0de4741..7ca788ffe12 100644
--- a/airflow-core/src/airflow/utils/sqlalchemy.py
+++ b/airflow-core/src/airflow/utils/sqlalchemy.py
@@ -643,6 +643,10 @@ def is_lock_not_available_error(error: OperationalError):
# importing it. This doesn't
if db_err_code in ("55P03", 1205, 3572):
return True
+ # SQLite: `database is locked` (SQLITE_BUSY) — check the error text since
+ # sqlite3.OperationalError.args[0] is a human-readable string, not a
numeric code
+ if error.orig and "database is locked" in str(error.orig).lower():
+ return True
return False
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py
index d1bf2b51729..d0fdd77e764 100644
---
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py
+++
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py
@@ -17,16 +17,18 @@
from __future__ import annotations
import os
+import sqlite3
from datetime import datetime, timedelta
from unittest import mock
import pendulum
import pytest
from sqlalchemy import and_, func, select
+from sqlalchemy.exc import OperationalError, ProgrammingError
from airflow._shared.timezones import timezone
from airflow.dag_processing.dagbag import DagBag
-from airflow.models import DagModel, DagRun
+from airflow.models import DagModel, DagRun, TaskInstance
from airflow.models.backfill import (
Backfill,
BackfillDagRun,
@@ -41,6 +43,7 @@ from airflow.providers.standard.operators.python import
PythonOperator
from airflow.sdk import CronPartitionTimetable
from airflow.utils.session import provide_session
from airflow.utils.state import DagRunState
+from airflow.utils.types import DagRunType
from tests_common.test_utils.asserts import assert_queries_count
from tests_common.test_utils.db import (
@@ -360,6 +363,196 @@ class TestCreateBackfill(TestBackfillEndpoint):
== "Dag has tasks for which depends_on_past=True. You must
set reprocess behavior to reprocess completed or reprocess failed."
)
+ def test_create_backfill_database_locked(self, session, dag_maker,
test_client):
+ """SQLite 'database is locked' during backfill creation returns HTTP
503."""
+ with dag_maker(session=session, dag_id="TEST_DAG_1", schedule="0 * * *
*") as dag:
+ EmptyOperator(task_id="mytask")
+ session.scalars(select(DagModel)).all()
+ session.commit()
+
+ from_date = pendulum.parse("2024-01-01")
+ to_date = pendulum.parse("2024-02-01")
+
+ data = {
+ "dag_id": dag.dag_id,
+ "from_date": to_iso(from_date),
+ "to_date": to_iso(to_date),
+ "max_active_runs": 5,
+ "run_backwards": False,
+ "dag_run_conf": {},
+ }
+
+ with mock.patch(
+
"airflow.api_fastapi.core_api.routes.public.backfills._create_backfill",
+ side_effect=OperationalError(
+ "statement", "params", sqlite3.OperationalError("database is
locked")
+ ),
+ ):
+ response = test_client.post("/backfills", json=data)
+
+ # OperationalError with "database is locked" should return 503
+ assert response.status_code == 503
+ assert "database is locked" in response.json()["detail"].lower()
+
+ def test_create_backfill_cleans_up_orphan_on_lock_error(self, session,
dag_maker):
+ """The partial Backfill row is removed when the cleanup runs after a
lock error."""
+ from airflow.models.backfill import Backfill, _cleanup_partial_backfill
+
+ with dag_maker(session=session, dag_id="TEST_DAG_CLEANUP", schedule="0
* * * *") as dag:
+ EmptyOperator(task_id="mytask")
+ session.commit()
+
+ bf = Backfill(
+ dag_id=dag.dag_id,
+ from_date=pendulum.parse("2024-01-01"),
+ to_date=pendulum.parse("2024-02-01"),
+ max_active_runs=5,
+ dag_run_conf={},
+ reprocess_behavior="none",
+ dag_model=dag,
+ triggering_user_name="test",
+ )
+ session.add(bf)
+ session.commit()
+ bf_id = bf.id
+
+ assert
session.scalar(select(func.count()).select_from(Backfill).where(Backfill.id ==
bf_id)) == 1
+
+ _cleanup_partial_backfill(bf, session)
+
+ assert
session.scalar(select(func.count()).select_from(Backfill).where(Backfill.id ==
bf_id)) == 0
+
+ def test_create_backfill_cleans_up_after_failed_transaction(self, session,
dag_maker):
+ """The cleanup works when the session is in a deactivated state.
+
+ Mirrors the real flow inside ``_create_backfill`` after an
+ ``OperationalError``: SQLAlchemy deactivates the session until
+ an explicit ``rollback()``. The cleanup must call it before any
+ further operation; without that, ``session.execute()`` raises
+ ``InvalidRequestError`` and the cleanup is a silent no-op.
+ """
+ from sqlalchemy import text
+
+ from airflow.models.backfill import Backfill, _cleanup_partial_backfill
+
+ with dag_maker(session=session, dag_id="TEST_DAG_CLEANUP_DEACT",
schedule="0 * * * *") as dag:
+ EmptyOperator(task_id="mytask")
+ session.commit()
+
+ bf = Backfill(
+ dag_id=dag.dag_id,
+ from_date=pendulum.parse("2024-01-01"),
+ to_date=pendulum.parse("2024-02-01"),
+ max_active_runs=5,
+ dag_run_conf={},
+ reprocess_behavior="none",
+ dag_model=dag,
+ triggering_user_name="test",
+ )
+ session.add(bf)
+ session.commit()
+ bf_id = bf.id
+
+ # Force the session into a deactivated state (same shape as after
+ # a failed flush). SQLite raises OperationalError; Postgres/MySQL
raise ProgrammingError.
+ with pytest.raises((OperationalError, ProgrammingError)):
+ session.execute(text("INVALID SQL STATEMENT"))
+
+ _cleanup_partial_backfill(bf, session)
+
+ assert
session.scalar(select(func.count()).select_from(Backfill).where(Backfill.id ==
bf_id)) == 0
+
+ def test_create_backfill_cleanup_removes_partial_dag_runs(self, session,
dag_maker):
+ """Cleanup removes partial DagRuns, TIs, and BackfillDagRun rows
alongside the Backfill."""
+ from airflow.models.backfill import _cleanup_partial_backfill
+
+ with dag_maker(session=session, dag_id="TEST_DAG_CLEANUP_DR",
schedule="0 * * * *") as dag:
+ EmptyOperator(task_id="mytask")
+ session.commit()
+
+ bf = Backfill(
+ dag_id=dag.dag_id,
+ from_date=pendulum.parse("2024-01-01"),
+ to_date=pendulum.parse("2024-02-01"),
+ max_active_runs=5,
+ dag_run_conf={},
+ reprocess_behavior="none",
+ dag_model=dag,
+ triggering_user_name="test",
+ )
+ session.add(bf)
+ session.commit()
+
+ dr1 = dag_maker.create_dagrun(
+ logical_date=pendulum.parse("2024-01-01"),
+ run_type=DagRunType.BACKFILL_JOB,
+ backfill_id=bf.id,
+ state=DagRunState.QUEUED,
+ )
+ session.add(
+ BackfillDagRun(
+ backfill_id=bf.id,
+ dag_run_id=dr1.id,
+ logical_date=pendulum.parse("2024-01-01"),
+ sort_ordinal=1,
+ )
+ )
+
+ dr2 = dag_maker.create_dagrun(
+ logical_date=pendulum.parse("2024-01-02"),
+ run_type=DagRunType.BACKFILL_JOB,
+ backfill_id=bf.id,
+ state=DagRunState.QUEUED,
+ )
+ session.add(
+ BackfillDagRun(
+ backfill_id=bf.id,
+ dag_run_id=dr2.id,
+ logical_date=pendulum.parse("2024-01-02"),
+ sort_ordinal=2,
+ )
+ )
+ session.commit()
+
+ bf_id = bf.id
+ dr1_id = dr1.id
+ dr2_id = dr2.id
+ run_ids = [dr1.run_id, dr2.run_id]
+
+ assert
session.scalar(select(func.count()).select_from(Backfill).where(Backfill.id ==
bf_id)) == 1
+ assert
session.scalar(select(func.count()).select_from(DagRun).where(DagRun.id ==
dr1_id)) == 1
+ assert
session.scalar(select(func.count()).select_from(DagRun).where(DagRun.id ==
dr2_id)) == 1
+ assert (
+ session.scalar(
+
select(func.count()).select_from(BackfillDagRun).where(BackfillDagRun.backfill_id
== bf_id)
+ )
+ == 2
+ )
+ assert (
+ session.scalar(
+
select(func.count()).select_from(TaskInstance).where(TaskInstance.run_id.in_(run_ids))
+ )
+ >= 2
+ )
+
+ _cleanup_partial_backfill(bf, session)
+
+ assert
session.scalar(select(func.count()).select_from(Backfill).where(Backfill.id ==
bf_id)) == 0
+ assert
session.scalar(select(func.count()).select_from(DagRun).where(DagRun.id ==
dr1_id)) == 0
+ assert
session.scalar(select(func.count()).select_from(DagRun).where(DagRun.id ==
dr2_id)) == 0
+ assert (
+ session.scalar(
+
select(func.count()).select_from(BackfillDagRun).where(BackfillDagRun.backfill_id
== bf_id)
+ )
+ == 0
+ )
+ assert (
+ session.scalar(
+
select(func.count()).select_from(TaskInstance).where(TaskInstance.run_id.in_(run_ids))
+ )
+ == 0
+ )
+
@pytest.mark.parametrize(
"run_backwards",
[
@@ -632,6 +825,32 @@ class TestCreateBackfill(TestBackfillEndpoint):
class TestCreateBackfillDryRun(TestBackfillEndpoint):
+ def test_create_backfill_dry_run_database_locked(self, session, dag_maker,
test_client):
+ """SQLite 'database is locked' during backfill dry-run returns HTTP
503."""
+ with dag_maker(session=session, dag_id="TEST_DAG_DRY_LOCK",
schedule="0 * * * *") as dag:
+ EmptyOperator(task_id="mytask")
+ session.commit()
+
+ data = {
+ "dag_id": dag.dag_id,
+ "from_date": to_iso(pendulum.parse("2024-01-01")),
+ "to_date": to_iso(pendulum.parse("2024-02-01")),
+ "max_active_runs": 5,
+ "run_backwards": False,
+ "dag_run_conf": {},
+ }
+
+ with mock.patch(
+ "airflow.api_fastapi.core_api.routes.public.backfills._do_dry_run",
+ side_effect=OperationalError(
+ "statement", "params", sqlite3.OperationalError("database is
locked")
+ ),
+ ):
+ response = test_client.post("/backfills/dry_run", json=data)
+
+ assert response.status_code == 503
+ assert "database is locked" in response.json()["detail"].lower()
+
@pytest.mark.parametrize(
("reprocess_behavior", "expected_dates"),
[
