fix(connectors): lock the connector definition row on edit and delete #1913
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -7,7 +7,7 @@ | |
|
|
||
| import logging | ||
| from datetime import datetime | ||
| from typing import Any, Dict, List, Optional, cast | ||
|
|
||
| from fastapi import APIRouter, Depends, HTTPException, status | ||
| from pydantic import BaseModel, Field | ||
|
|
@@ -286,7 +286,7 @@ async def get_custom_api( | |
|
|
||
|
|
||
| @custom_api_router.put("/{api_id}", response_model=CustomApiResponse) | ||
| def update_custom_api( | ||
| api_id: int, | ||
| api_data: CustomApiUpdate, | ||
| current_user: User = Depends(get_current_user), | ||
|
|
@@ -315,9 +315,81 @@ async def update_custom_api( | |
| detail="You do not have permission to edit this Custom API", | ||
| ) | ||
|
|
||
| # Which row a request writes decides which row it locks. Every field of | ||
| # ``CustomApiUpdate`` except ``is_active`` writes the shared | ||
| # ``CustomApi`` definition row; ``is_active`` writes this caller's own | ||
| # ``UserCustomApi`` link row and nothing else. Locking the definition | ||
| # row for a payload that never writes it made an activate/deactivate | ||
| # queue behind an unrelated edit of the same connector -- a wait that | ||
| # request has no write to justify, and one that surfaces as an error | ||
| # rather than a delay wherever a lock timeout is configured. | ||
| # | ||
| # ``model_fields_set`` decides this, not the values: an explicitly-null | ||
| # ``runtime_input_schema`` is written to the definition row below even | ||
| # though its value is ``None``, while an absent field is not written at | ||
| # all. The set is a superset of the writes below -- a payload carrying | ||
| # ``description=None`` is counted here and then skipped by the write at | ||
| # its own ``is not None`` guard -- so this takes the lock in a few | ||
| # cases that did not need it and skips it in none that do. | ||
| fields_set = api_data.model_fields_set | ||
| writes_definition_row = bool(fields_set - {"is_active"}) | ||
|
Collaborator
There was a problem hiding this comment. Minor; Blocking: no — same C7 root [prior]. I saw reply 3889610607 that
Collaborator
Author
There was a problem hiding this comment. Confirmed as described: a payload like I looked at moving to a "what will actually be written" classification instead of "which fields are present," and I'd rather not, for two reasons. First, it isn't a free win even on correctness: Second, the So I'm keeping the field-presence criterion as is. The extra locking it causes is confined to no-op companion payloads riding alongside
Collaborator
Author
There was a problem hiding this comment. Correcting one sentence in my earlier reply on this thread: I wrote that checking the definition's current name at the classifier "would mean reading that row once to check, then reading it again (locked) to write -- an extra query." That's wrong. By the time the classifier ( The conclusion doesn't change, but the reason does. The real problem with a value-based classifier isn't its cost — it's that the only name it could compare against is the one read at that earlier, unlocked point, and the write logic downstream compares against a different, later read instead (the fresh, locked-when-taken read at
AlexLiu190625 marked this conversation as resolved.
|
||
|
|
||
| # A fresh single-table read of the definition row, on both paths. The | ||
| # read above comes through the personal link row's relationship and | ||
| # cannot itself address just this table; this is a separate statement, | ||
| # so a row deleted between the two yields None here (handled as the | ||
| # same 404) rather than surfacing as an unrelated error out of the | ||
| # write path or out of ``db.refresh`` below. ``populate_existing()`` | ||
| # makes this statement's row the one the rest of this route reads and | ||
| # responds with: without it the already-identity-mapped instance the | ||
| # relationship loaded would be returned unrefreshed, and every field | ||
| # below would still be that earlier snapshot. | ||
| # | ||
| # ``FOR UPDATE`` is added only on the path that writes this row, so a | ||
| # request that writes it still waits for another request holding it. | ||
| # That clause is a PostgreSQL/MySQL row lock only: SQLAlchemy renders | ||
| # no locking clause at all on SQLite -- the statement it emits there is | ||
| # byte-for-byte the one it emits without this call -- so on a SQLite | ||
| # deployment the read-modify-write below is not serialized and two | ||
| # concurrent edits of one connector can still interleave. The | ||
| # single-statement conditional ``UPDATE``s elsewhere in this repository | ||
| # (services/task_interaction_staging.py, | ||
| # services/chat_history_service.py) are safe on SQLite without a lock | ||
| # because one statement is atomic there; that reasoning does not carry | ||
| # to this route, which reads the row, computes in Python, and writes it | ||
| # back. Closing the SQLite window needs the dual-dialect fence | ||
| # ``_lock_user_row_for_preferences_update`` (api/auth.py) and | ||
| # ``acquire_runtime_key_transition_fence`` (services/api_keys.py) use | ||
| # -- a no-op ``UPDATE`` that takes SQLite's writer lock -- and is left | ||
| # to a change of its own. | ||
| definition_query = ( | ||
| db.query(CustomApi).filter(CustomApi.id == api_id).populate_existing() | ||
| ) | ||
| if writes_definition_row: | ||
| definition_query = definition_query.with_for_update() | ||
|
AlexLiu190625 marked this conversation as resolved.
|
||
| current_api = definition_query.first() | ||
| if current_api is None: | ||
| raise HTTPException( | ||
| status_code=status.HTTP_404_NOT_FOUND, | ||
| detail="Custom API not found", | ||
| ) | ||
| api = current_api | ||
|
|
||
| # Read only after the statement above: rename_team_connector's "old" | ||
| # argument must be the name this transaction's own read established -- | ||
| # under the lock, on the path that renames -- not whatever the | ||
| # relationship read further up saw. A concurrent committed rename in | ||
| # between would otherwise make this stale, and the rewrite below would | ||
| # look for a name that no longer exists anywhere, leaving the previous | ||
| # renamer's selectors dangling with no error. | ||
| old_name = str(api.name) | ||
|
|
||
| # The row's declared type from here on is loosened for mypy's sake: the | ||
| # column-typed attributes below (name, description, env, ...) are all | ||
| # mutated directly by this route, exactly as they were when this local | ||
| # came off the relationship instead of off the definition query above. | ||
| mutable_api = cast(Any, api) | ||
|
|
||
| # Check name uniqueness if name is changed | ||
| if api_data.name and api_data.name != api.name: | ||
| existing = db.query(CustomApi).filter(CustomApi.name == api_data.name).first() | ||
|
|
@@ -326,33 +398,34 @@ async def update_custom_api( | |
| status_code=status.HTTP_400_BAD_REQUEST, | ||
| detail=f"Custom API with name '{api_data.name}' already exists", | ||
| ) | ||
| mutable_api.name = api_data.name | ||
|
|
||
| # Update fields | ||
| if api_data.description is not None: | ||
| mutable_api.description = api_data.description | ||
| if api_data.url is not None: | ||
| mutable_api.url = api_data.url | ||
| if api_data.method is not None: | ||
| mutable_api.method = api_data.method | ||
| if api_data.headers is not None: | ||
| mutable_api.headers = api_data.headers | ||
| if api_data.body is not None: | ||
| mutable_api.body = api_data.body | ||
|
|
||
| # Process env variables | ||
| if api_data.env is not None: | ||
| existing_env: Dict[str, str] = ( | ||
| mutable_api.env if isinstance(api.env, dict) else {} | ||
| ) | ||
| try: | ||
| processed_env = _process_env_vars(api_data.env, existing_env) | ||
| except ValueError as exc: | ||
| raise HTTPException( | ||
| status_code=status.HTTP_400_BAD_REQUEST, | ||
| detail=f"Invalid environment variables: {exc}", | ||
| ) from exc | ||
| mutable_api.env = processed_env | ||
|
|
||
| runtime_input_schema = ( | ||
| api_data.runtime_input_schema | ||
| if "runtime_input_schema" in fields_set | ||
|
|
@@ -374,19 +447,19 @@ async def update_custom_api( | |
| runtime_input_schema=runtime_input_schema, | ||
| runtime_bindings=runtime_bindings, | ||
| allow_delegated_authorization=allow_delegated_authorization, | ||
| static_headers=mutable_api.headers, | ||
| ) | ||
| except ValueError as exc: | ||
| raise HTTPException( | ||
| status_code=status.HTTP_400_BAD_REQUEST, | ||
| detail=f"Invalid runtime configuration: {exc}", | ||
| ) from exc | ||
| if "runtime_input_schema" in fields_set: | ||
| mutable_api.runtime_input_schema = runtime_input_schema | ||
| if "runtime_bindings" in fields_set: | ||
| mutable_api.runtime_bindings = runtime_bindings | ||
| if "allow_delegated_authorization" in fields_set: | ||
| mutable_api.allow_delegated_authorization = allow_delegated_authorization | ||
|
|
||
| from ..services.connector_team_scope import rename_team_connector | ||
|
|
||
|
|
@@ -410,7 +483,7 @@ async def update_custom_api( | |
|
|
||
|
|
||
| @custom_api_router.delete("/{api_id}", status_code=status.HTTP_204_NO_CONTENT) | ||
| def delete_custom_api( | ||
| api_id: int, | ||
| current_user: User = Depends(get_current_user), | ||
| db: Session = Depends(get_db), | ||
|
|
@@ -438,7 +511,65 @@ async def delete_custom_api( | |
| detail="You do not have permission to delete this Custom API", | ||
| ) | ||
|
|
||
| # One lock order across every row this pair of routes touches. An | ||
| # ``update_custom_api`` call that writes the definition row locks this | ||
| # same row first, calls ``rename_team_connector`` afterwards, and | ||
| # writes the ``UserCustomApi`` link row afterwards too; both branches below delete | ||
| # the link row first and the definition row second, inside one | ||
| # transaction. Taking the lock here -- before ``delete_team_connector`` | ||
| # rather than after it -- puts both routes in one order on both sides of | ||
| # the hook boundary: definition row first, then the link row and | ||
| # whatever rows a connector team hook locks. With the lock after the | ||
| # hook instead, the two routes waited in opposite directions across that | ||
| # boundary and a concurrent edit/delete pair on the same connector could | ||
| # deadlock (PostgreSQL 40P01, surfacing to the caller as HTTP 500). | ||
| # ``populate_existing`` matches the PUT's own lock: the row this | ||
| # transaction holds is the one the deletion below acts on, not whatever | ||
| # the relationship read above happened to see. This is also a fresh | ||
| # statement, so a row deleted between the access read above and here | ||
| # still yields None (handled as the same 404) rather than reaching | ||
| # ``db.delete`` with nothing to delete. | ||
| # | ||
| # Two costs come with taking it here rather than last. The two 403 | ||
| # refusals below read ``delete_team_connector``'s answer and so now | ||
| # happen after this statement: a request that is going to be refused | ||
| # does briefly hold this row, until the raised ``HTTPException`` | ||
| # propagates out and the request's session is closed without | ||
| # committing. And the hook's own work now runs inside the lock, so this | ||
| # route holds the row for longer than it did with the lock last. | ||
| # | ||
| # What this statement orders is this route against ``update_custom_api`` | ||
| # on the same connector. A PUT that writes the definition row cannot | ||
| # form a cycle with this route, whatever the hook locks, because both | ||
| # reach the hook with this row already held and so cannot be inside the | ||
| # hook at the same time. A PUT that writes only the caller's own | ||
| # ``UserCustomApi`` link row takes no lock on this row at all: it reads | ||
| # the definition row without locking it and waits only for the link | ||
| # row, so it has nothing this route waits for and cannot close a cycle | ||
| # either. A hook that additionally locks rows of its own in some other | ||
| # order relative to this repository's statements is outside what this | ||
| # statement arranges; only the installing application can order those. | ||
| # | ||
| # ``FOR UPDATE`` here is a PostgreSQL/MySQL row lock only: SQLAlchemy | ||
| # renders no locking clause at all on SQLite, so on a SQLite deployment | ||
| # this statement does not order this route against anything. Closing | ||
| # that window needs the dual-dialect fence | ||
| # ``acquire_runtime_key_transition_fence`` (services/api_keys.py) uses | ||
| # -- a no-op ``UPDATE`` that takes SQLite's writer lock -- and is left | ||
| # to a change of its own. | ||
| locked_api = ( | ||
| db.query(CustomApi) | ||
| .filter(CustomApi.id == api_id) | ||
| .populate_existing() | ||
| .with_for_update() | ||
|
AlexLiu190625 marked this conversation as resolved.
|
||
| .first() | ||
| ) | ||
| if locked_api is None: | ||
| raise HTTPException( | ||
| status_code=status.HTTP_404_NOT_FOUND, | ||
| detail="Custom API not found", | ||
| ) | ||
| api = locked_api | ||
|
|
||
| from ..services.connector_team_scope import delete_team_connector | ||
|
|
||
|
|
@@ -455,6 +586,7 @@ async def delete_custom_api( | |
| status_code=status.HTTP_403_FORBIDDEN, | ||
| detail="Only a team admin can delete a team Custom API", | ||
| ) | ||
|
|
||
| if team_delete.team_owned: | ||
| db.delete(user_api) | ||
| db.flush([user_api]) | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.