Foreign key violation error when starting flow with de-registered but active agent

[alex-hsp]

Description

When trying to submit a flow run without registered agent (i removed agent via ui) (but agent was still running though), prefect server fails with cryptic exception shown to user:

[{'message': 'Foreign key violation.', 'locations': [{'line': 2, 'column': 5}], 'path': ['set_flow_run_states'], 'extensions': {'code': 'INTERNAL_SERVER_ERROR', 'exception': {'message': 'Foreign key violation.'}}}]

Expected Behavior

Do not schedule flow run into de-registered agents

Reproduction

1. Create a minimal flow
2. Spin up prefect server setup
3. Start any agent (i was using docker agent)
4. Wait until agent shows up, and then remove it from UI
5. Start a flow

Environment

Some docker-compose logs

hs-databricks-agent_1  | [2020-10-23 08:21:00,939] DEBUG - hs-databricks agent | Found flow runs ['b13ab3e8-b781-429c-a65b-96a551683dda']
hs-databricks-agent_1  | [2020-10-23 08:21:00,940] DEBUG - hs-databricks agent | Querying flow run metadata
hs-databricks-agent_1  | [2020-10-23 08:21:00,958] INFO - hs-databricks agent | Found 1 flow run(s) to submit for execution.
hs-databricks-agent_1  | [2020-10-23 08:21:00,959] DEBUG - hs-databricks agent | Next query for flow runs in 0.25 seconds
hs-databricks-agent_1  | [2020-10-23 08:21:00,959] DEBUG - hs-databricks agent | Updating states for flow run b13ab3e8-b781-429c-a65b-96a551683dda
hs-databricks-agent_1  | [2020-10-23 08:21:00,961] DEBUG - hs-databricks agent | Flow run b13ab3e8-b781-429c-a65b-96a551683dda is in a Scheduled state, updating to Submitted
postgres_1             | 2020-10-23 08:21:00.998 UTC [4573] ERROR:  insert or update on table "flow_run" violates foreign key constraint "flow_run_agent_id_fkey"
postgres_1             | 2020-10-23 08:21:00.998 UTC [4573] DETAIL:  Key (agent_id)=(c42a93cb-8e89-4d16-b6e3-01e8c08e0c01) is not present in table "agent".
postgres_1             | 2020-10-23 08:21:00.998 UTC [4573] STATEMENT:  WITH "flow_run__mutation_result_alias" AS (UPDATE "public"."flow_run" SET "agent_id" = ($1)::uuid  WHERE (('true') AND ((((("public"."flow_run"."id") = (($2)::uuid)) AND ('true')) AND ('true')) AND ('true'))) RETURNING * , CASE WHEN 'true' THEN NULL ELSE "hdb_catalog"."check_violation"('update check constraint failed')  END ), "flow_run__all_columns_alias" AS (SELECT  "id" , "tenant_id" , "created" , "flow_id" , "parameters" , "scheduled_start_time" , "auto_scheduled" , "heartbeat" , "start_time" , "end_time" , "version" , "state" , "state_timestamp" , "state_message" , "state_result" , "state_start_time" , "serialized_state" , "name" , "context" , "times_resurrected" , "updated" , "idempotency_key" , "agent_id" , "labels"  FROM "flow_run__mutation_result_alias"      ) SELECT  json_build_object('affected_rows', (SELECT  COUNT(*)  FROM "flow_run__all_columns_alias"      ) )        
hasura_1               | {"type":"http-log","timestamp":"2020-10-23T08:21:00.929+0000","level":"error","detail":{"operation":{"user_vars":{"x-hasura-role":"admin"},"error":{"path":"$","error":"Foreign key violation. insert or update on table \"flow_run\" violates foreign key constraint \"flow_run_agent_id_fkey\"","code":"constraint-violation"},"request_id":"16fac4d9-b87e-41a2-ae98-f48b0e6c615c","response_size":173,"query":{"variables":{"update_set":{"agent_id":"c42a93cb-8e89-4d16-b6e3-01e8c08e0c01"},"update_where":{"id":{"_eq":"b13ab3e8-b781-429c-a65b-96a551683dda"}}},"query":"mutation($update_where: flow_run_bool_exp!, $update_set: flow_run_set_input) {\n    update: update_flow_run(where: $update_where, _set: $update_set) {\n        affected_rows\n    }\n}"}},"http_info":{"status":400,"http_version":"HTTP/1.1","url":"/v1alpha1/graphql","ip":"172.18.0.4","method":"POST","content_encoding":null}}}
graphql_1              | Foreign key violation.
graphql_1              | 
graphql_1              | GraphQL request:2:3
graphql_1              | 1 | mutation ($input: set_flow_run_states_input!) {
graphql_1              | 2 |   set_flow_run_states(input: $input) {
graphql_1              |   |   ^
graphql_1              | 3 |     states {
graphql_1              | Traceback (most recent call last):
graphql_1              |   File "/prefect-server/src/prefect_server/database/hasura.py", line 85, in execute
graphql_1              |     as_box=as_box,
graphql_1              |   File "/prefect-server/src/prefect_server/utilities/graphql.py", line 80, in execute
graphql_1              |     raise ValueError(result["errors"])
graphql_1              | ValueError: [{'extensions': {'path': '$', 'code': 'constraint-violation'}, 'message': 'Foreign key violation. insert or update on table "flow_run" violates foreign key constraint "flow_run_agent_id_fkey"'}]
graphql_1              | 
graphql_1              | During handling of the above exception, another exception occurred:
graphql_1              | 
graphql_1              | Traceback (most recent call last):
graphql_1              |   File "/usr/local/lib/python3.7/site-packages/graphql/execution/execute.py", line 628, in await_result
graphql_1              |     return await result
graphql_1              |   File "/prefect-server/src/prefect_server/graphql/extensions.py", line 52, in resolve
graphql_1              |     result = await result
graphql_1              |   File "/prefect-server/src/prefect_server/graphql/states.py", line 45, in resolve_set_flow_run_states
graphql_1              |     *[check_size_and_set_state(state_input) for state_input in input["states"]]
graphql_1              |   File "/prefect-server/src/prefect_server/graphql/states.py", line 39, in check_size_and_set_state
graphql_1              |     agent_id=agent_id,
graphql_1              |   File "/prefect-server/src/prefect_server/api/states.py", line 133, in set_flow_run_state
graphql_1              |     await api.runs.update_flow_run_agent(flow_run_id=flow_run_id, agent_id=agent_id)
graphql_1              |   File "/prefect-server/src/prefect_server/api/runs.py", line 410, in update_flow_run_agent
graphql_1              |     set={"agent_id": agent_id}
graphql_1              |   File "/prefect-server/src/prefect_server/database/orm.py", line 406, in update
graphql_1              |     run_mutation=run_mutation,
graphql_1              |   File "/prefect-server/src/prefect_server/database/hasura.py", line 391, in update
graphql_1              |     result = await self.execute_mutations_in_transaction(mutations=[graphql])
graphql_1              |   File "/prefect-server/src/prefect_server/database/hasura.py", line 165, in execute_mutations_in_transaction
graphql_1              |     as_box=as_box,
graphql_1              |   File "/prefect-server/src/prefect_server/database/hasura.py", line 91, in execute
graphql_1              |     raise ValueError("Foreign key violation.")
graphql_1              | ValueError: Foreign key violation.
graphql_1              | 
graphql_1              | The above exception was the direct cause of the following exception:
graphql_1              | 
graphql_1              | Traceback (most recent call last):
graphql_1              |   File "/usr/local/lib/python3.7/site-packages/graphql/execution/execute.py", line 674, in await_completed
graphql_1              |     return await completed
graphql_1              |   File "/usr/local/lib/python3.7/site-packages/graphql/execution/execute.py", line 659, in await_result
graphql_1              |     return_type, field_nodes, info, path, await result
graphql_1              |   File "/usr/local/lib/python3.7/site-packages/graphql/execution/execute.py", line 733, in complete_value
graphql_1              |     raise result
graphql_1              |   File "/usr/local/lib/python3.7/site-packages/graphql/execution/execute.py", line 628, in await_result
graphql_1              |     return await result
graphql_1              |   File "/prefect-server/src/prefect_server/graphql/extensions.py", line 52, in resolve
graphql_1              |     result = await result
graphql_1              |   File "/prefect-server/src/prefect_server/graphql/states.py", line 45, in resolve_set_flow_run_states
graphql_1              |     *[check_size_and_set_state(state_input) for state_input in input["states"]]
graphql_1              |   File "/prefect-server/src/prefect_server/graphql/states.py", line 39, in check_size_and_set_state
graphql_1              |     agent_id=agent_id,
graphql_1              |   File "/prefect-server/src/prefect_server/api/states.py", line 133, in set_flow_run_state
graphql_1              |     await api.runs.update_flow_run_agent(flow_run_id=flow_run_id, agent_id=agent_id)
graphql_1              |   File "/prefect-server/src/prefect_server/api/runs.py", line 410, in update_flow_run_agent
graphql_1              |     set={"agent_id": agent_id}
graphql_1              |   File "/prefect-server/src/prefect_server/database/orm.py", line 406, in update
graphql_1              |     run_mutation=run_mutation,
graphql_1              |   File "/prefect-server/src/prefect_server/database/hasura.py", line 391, in update
graphql_1              |     result = await self.execute_mutations_in_transaction(mutations=[graphql])
graphql_1              |   File "/prefect-server/src/prefect_server/database/hasura.py", line 165, in execute_mutations_in_transaction
graphql_1              |     as_box=as_box,
graphql_1              |   File "/prefect-server/src/prefect_server/database/hasura.py", line 91, in execute
graphql_1              |     raise ValueError("Foreign key violation.")
graphql_1              | graphql.error.graphql_error.GraphQLError: Foreign key violation.
graphql_1              | 
graphql_1              | GraphQL request:2:3
graphql_1              | 1 | mutation ($input: set_flow_run_states_input!) {
graphql_1              | 2 |   set_flow_run_states(input: $input) {
graphql_1              |   |   ^
graphql_1              | 3 |     states {
hs-databricks-agent_1  | [2020-10-23 08:21:01,013] ERROR - hs-databricks agent | Logging platform error for flow run b13ab3e8-b781-429c-a65b-96a551683dda
graphql_1              | INFO:     172.18.0.7:39012 - "POST /graphql/ HTTP/1.1" 200 OK
graphql_1              | INFO:     172.18.0.7:39016 - "POST /graphql/ HTTP/1.1" 200 OK
graphql_1              | INFO:     172.18.0.7:39020 - "POST /graphql/ HTTP/1.1" 200 OK
hs-databricks-agent_1  | [2020-10-23 08:21:01,066] ERROR - hs-databricks agent | Error while deploying flow: ClientError([{'message': 'Foreign key violation.', 'locations': [{'line': 2, 'column': 5}], 'path': ['set_flow_run_states'], 'extensions': {'code': 'INTERNAL_SERVER_ERROR', 'exception': {'message': 'Foreign key violation.'}}}])
hs-databricks-agent_1  | [2020-10-23 08:21:01,067] DEBUG - hs-databricks agent | Completed flow run submission (id: b13ab3e8-b781-429c-a65b-96a551683dda)

• postgres:11
• hasura/graphql-engine:v1.3.0
• prefecthq/server:2020-10-08
• prefecthq/apollo:2020-10-07
• prefecthq/ui:2020-10-05
• prefecthq/prefect:0.13.10-python3.7 as agent

[joshmeek]

Thanks for opening @alex-hsp and outlining how to easily reproduce!