From 04b8c744fb7bc5f16494193287b696cb0d925aff Mon Sep 17 00:00:00 2001 From: Yue Chao Qin Date: Mon, 24 Aug 2026 16:09:13 -0700 Subject: [PATCH] Extract `_create_in_transaction` so a caller can own the pipeline-run transaction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## What this changes `PipelineRunsApiService_Sql.create()` is split in two. A new private `_create_in_transaction()` does the work — build the execution-node tree, insert the `PipelineRun`, flush, mirror the system annotations — and returns the `bts.PipelineRun`. `create()` keeps its signature, its return type and its transaction, and is now a thin wrapper around it. ``` before after ────── ───── create() create() with session.begin(): with session.begin(): ...build and insert... _create_in_transaction() ← the work, no commit session.commit() ← redundant session.refresh(); return session.refresh(); return ``` Purely additive for existing callers: same signature, same transaction ownership, same response object. The one behavioural tidy-up is the `session.commit()` that sat *inside* `with session.begin():` — the block already commits on exit, so it is gone. ## Why Starting a pipeline run is a same-database insert, not a remote call, so it can share a caller's transaction. Today it cannot: `create()` calls `session.begin()`, and SQLAlchemy refuses that on a session that already has a transaction open. That blocks any caller that has to write a run **atomically with its own rows**. Ours is an event-driven trigger: it claims a "this cycle has fired" fence row and starts a run, and the two must commit or roll back together — otherwise a crash between them either fires the same trigger twice or drops the run silently. With `_create_in_transaction()` the caller keeps one transaction around both. ## Private on purpose The underscore is deliberate: this is an internal seam, not a new public API, so nothing here is promised to stay. Happy to make it public if other consumers want the same guarantee — that is a question for reviewers, and it changes nothing about the code. ## Tests `TestCreateInTransaction` pins the contract, since a private method has nothing else protecting it: - the run is flushed, so the caller can use its ID inside the transaction; - an outer rollback leaves **zero** `pipeline_run` and **zero** `execution_node` rows; - a caller holding `with session.begin():` — the same shape `create()` uses — gets a durable run once the block exits. The last two are mutation-checked: putting `session.commit()` back inside `_create_in_transaction` turns **both** red, so the "never commits" rule has two independent guards. Existing coverage of `create()` is unchanged — 50 tests across 8 classes reach it, all against a real SQLite engine with no mocks, and they too go red if `create()` stops committing. Full suite: 471 passed. Assisted-By: devx/11fb0c55-5ff5-402f-abc7-1a722b0b09cb --- cloud_pipelines_backend/api_server_sql.py | 82 +++++++++++++++-------- tests/test_api_server_sql.py | 50 ++++++++++++++ 2 files changed, 104 insertions(+), 28 deletions(-) diff --git a/cloud_pipelines_backend/api_server_sql.py b/cloud_pipelines_backend/api_server_sql.py index cf71a165..451f0796 100644 --- a/cloud_pipelines_backend/api_server_sql.py +++ b/cloud_pipelines_backend/api_server_sql.py @@ -106,7 +106,7 @@ def _fail_if_changing_system_annotation(self, *, key: str) -> None: if key.startswith(filter_query_sql.SYSTEM_KEY_PREFIX): raise errors.ApiValidationError(self._SYSTEM_KEY_RESERVED_MSG) - def create( + def _create_in_transaction( self, session: orm.Session, root_task: structures.TaskSpec, @@ -115,44 +115,70 @@ def create( # Arbitrary metadata. Can be used to specify user. annotations: Optional[dict[str, Any]] = None, created_by: str | None = None, - ) -> PipelineRunResponse: + ) -> bts.PipelineRun: + """Creates a pipeline run inside a transaction the caller already owns. + + Flushes, so the returned run has its ID populated, but never commits: + the caller decides when the work becomes durable. Use this when a run + must be written atomically with the caller's own rows. Callers that just + want a run created should use `create` instead. + """ # TODO: Validate the pipeline spec # TODO: Load and validate all components # TODO: Fetch missing components and populate component specs pipeline_name = root_task.component_ref.spec.name - with session.begin(): + root_execution_node = _recursively_create_all_executions_and_artifacts_root( + session=session, + root_task_spec=root_task, + ) - root_execution_node = _recursively_create_all_executions_and_artifacts_root( - session=session, - root_task_spec=root_task, - ) + # Store into DB. + current_time = _get_current_time() + pipeline_run = bts.PipelineRun( + root_execution=root_execution_node, + created_at=current_time, + updated_at=current_time, + annotations=annotations, + created_by=created_by, + extra_data={ + self._PIPELINE_NAME_EXTRA_DATA_KEY: pipeline_name, + }, + ) + session.add(pipeline_run) + # Flush to populate pipeline_run.id (server-generated) before inserting annotation FKs. + # TODO: Use ORM relationship instead of explicit flush + manual FK assignment. + session.flush() + _mirror_system_annotations( + session=session, + pipeline_run_id=pipeline_run.id, + created_by=created_by, + pipeline_name=pipeline_name, + annotations=annotations, + ) + return pipeline_run - # Store into DB. - current_time = _get_current_time() - pipeline_run = bts.PipelineRun( - root_execution=root_execution_node, - created_at=current_time, - updated_at=current_time, - annotations=annotations, - created_by=created_by, - extra_data={ - self._PIPELINE_NAME_EXTRA_DATA_KEY: pipeline_name, - }, - ) - session.add(pipeline_run) - # Flush to populate pipeline_run.id (server-generated) before inserting annotation FKs. - # TODO: Use ORM relationship instead of explicit flush + manual FK assignment. - session.flush() - _mirror_system_annotations( + def create( + self, + session: orm.Session, + root_task: structures.TaskSpec, + # Component library to avoid repeating component specs inside task specs + components: Optional[list[structures.ComponentReference]] = None, + # Arbitrary metadata. Can be used to specify user. + annotations: Optional[dict[str, Any]] = None, + created_by: str | None = None, + ) -> PipelineRunResponse: + # `session.begin()` commits when the block exits, so no explicit commit + # is needed here. + with session.begin(): + pipeline_run = self._create_in_transaction( session=session, - pipeline_run_id=pipeline_run.id, - created_by=created_by, - pipeline_name=pipeline_name, + root_task=root_task, + components=components, annotations=annotations, + created_by=created_by, ) - session.commit() session.refresh(pipeline_run) return PipelineRunResponse.from_db(pipeline_run) diff --git a/tests/test_api_server_sql.py b/tests/test_api_server_sql.py index f8f23b98..900f775c 100644 --- a/tests/test_api_server_sql.py +++ b/tests/test_api_server_sql.py @@ -433,6 +433,56 @@ def test_create_mirrors_absent_values_as_empty_string( ) +def _count_rows(*, session: orm.Session, table: type) -> int: + return session.scalar(sqlalchemy.select(sqlalchemy.func.count()).select_from(table)) + + +class TestCreateInTransaction: + """Pins the contract of `_create_in_transaction` for callers that own the transaction. + + The method is private, so nothing outside this file is promised it exists or + that it stays free of an internal commit. These tests are what turns that + into a promise: a rename, a removal, or a commit creeping back in fails here + rather than in a consumer that batches the run with its own rows. + """ + + def test_flushes_so_the_caller_can_use_the_run_id(self, session_factory, service): + with session_factory() as session: + session.begin() + pipeline_run = service._create_in_transaction( + session, root_task=_make_task_spec("in-transaction") + ) + assert pipeline_run.id is not None + assert pipeline_run.root_execution_id is not None + session.rollback() + + def test_rollback_leaves_no_rows(self, session_factory, service): + with session_factory() as session: + session.begin() + service._create_in_transaction( + session, root_task=_make_task_spec("rolled-back") + ) + session.rollback() + + with session_factory() as session: + assert _count_rows(session=session, table=bts.PipelineRun) == 0 + assert _count_rows(session=session, table=bts.ExecutionNode) == 0 + + def test_the_callers_commit_makes_the_run_durable(self, session_factory, service): + with session_factory() as session: + # The same shape `create` uses: the block commits on exit, so the + # test never commits by hand. + with session.begin(): + pipeline_run = service._create_in_transaction( + session, root_task=_make_task_spec("committed-by-caller") + ) + run_id = pipeline_run.id + + with session_factory() as session: + assert session.get(bts.PipelineRun, run_id) is not None + assert _count_rows(session=session, table=bts.PipelineRun) == 1 + + class TestCreateMirrorsUserAnnotations: def test_create_mirrors_user_annotations( self,