diff --git a/acceptance/bundle/dms/declined-deploy/output.txt b/acceptance/bundle/dms/declined-deploy/output.txt index 0412b4b2f7..3b928d83ce 100644 --- a/acceptance/bundle/dms/declined-deploy/output.txt +++ b/acceptance/bundle/dms/declined-deploy/output.txt @@ -45,10 +45,11 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged "update_mask": "state,error_message,resource_id,status" }, "body": { - "state": "{\"state\":{\"catalog_name\":\"main\",\"name\":\"dms_declined_deploy_schema\"}}", + "error_message": "", "resource_id": "main.dms_declined_deploy_schema", - "status": "OPERATION_STATUS_SUCCEEDED", - "sequence_id": "0" + "sequence_id": "0", + "state": "{\"state\":{\"catalog_name\":\"main\",\"name\":\"dms_declined_deploy_schema\"}}", + "status": "OPERATION_STATUS_SUCCEEDED" } } { diff --git a/acceptance/bundle/dms/depends-on/output.txt b/acceptance/bundle/dms/depends-on/output.txt index ff24ec209f..93a63b305b 100644 --- a/acceptance/bundle/dms/depends-on/output.txt +++ b/acceptance/bundle/dms/depends-on/output.txt @@ -15,10 +15,11 @@ Resources: 2 created, 0 changed, 0 deleted, 0 unchanged "update_mask": "state,error_message,resource_id,status" }, "body": { - "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-depends-on/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"parent\",\"queue\":{\"enabled\":true}}}", + "error_message": "", "resource_id": "[NUMID]", - "status": "OPERATION_STATUS_SUCCEEDED", - "sequence_id": "0" + "sequence_id": "0", + "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-depends-on/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"parent\",\"queue\":{\"enabled\":true}}}", + "status": "OPERATION_STATUS_SUCCEEDED" } } { @@ -28,10 +29,11 @@ Resources: 2 created, 0 changed, 0 deleted, 0 unchanged "update_mask": "state,error_message,resource_id,status" }, "body": { - "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-depends-on/default/state/metadata.json\",\"version_id\":\"1\"},\"description\":\"depends on [NUMID]\",\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"child\",\"queue\":{\"enabled\":true}},\"depends_on\":[{\"node\":\"resources.jobs.parent\",\"label\":\"${resources.jobs.parent.id}\"}]}", + "error_message": "", "resource_id": "[NUMID]", - "status": "OPERATION_STATUS_SUCCEEDED", - "sequence_id": "0" + "sequence_id": "0", + "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-depends-on/default/state/metadata.json\",\"version_id\":\"1\"},\"description\":\"depends on [NUMID]\",\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"child\",\"queue\":{\"enabled\":true}},\"depends_on\":[{\"node\":\"resources.jobs.parent\",\"label\":\"${resources.jobs.parent.id}\"}]}", + "status": "OPERATION_STATUS_SUCCEEDED" } } diff --git a/acceptance/bundle/dms/emptied-resource/output.txt b/acceptance/bundle/dms/emptied-resource/output.txt index cbbea18ae8..59f1e3051a 100644 --- a/acceptance/bundle/dms/emptied-resource/output.txt +++ b/acceptance/bundle/dms/emptied-resource/output.txt @@ -50,10 +50,11 @@ Resources: 2 created, 0 changed, 0 deleted, 0 unchanged "update_mask": "state,error_message,resource_id,status" }, "body": { - "state": "{\"state\":{\"catalog_name\":\"main\",\"name\":\"dms_emptied_resource\"}}", + "error_message": "", "resource_id": "main.dms_emptied_resource", - "status": "OPERATION_STATUS_SUCCEEDED", - "sequence_id": "0" + "sequence_id": "0", + "state": "{\"state\":{\"catalog_name\":\"main\",\"name\":\"dms_emptied_resource\"}}", + "status": "OPERATION_STATUS_SUCCEEDED" } } { @@ -63,10 +64,11 @@ Resources: 2 created, 0 changed, 0 deleted, 0 unchanged "update_mask": "state,error_message,resource_id,status" }, "body": { - "state": "{\"state\":{\"securable_type\":\"schema\",\"full_name\":\"main.dms_emptied_resource\",\"__embed__\":[{\"principal\":\"someone@example.com\",\"privileges\":[\"USE_SCHEMA\"]}]},\"depends_on\":[{\"node\":\"resources.schemas.foo\",\"label\":\"${resources.schemas.foo.id}\"}]}", + "error_message": "", "resource_id": "schema/main.dms_emptied_resource", - "status": "OPERATION_STATUS_SUCCEEDED", - "sequence_id": "0" + "sequence_id": "0", + "state": "{\"state\":{\"securable_type\":\"schema\",\"full_name\":\"main.dms_emptied_resource\",\"__embed__\":[{\"principal\":\"someone@example.com\",\"privileges\":[\"USE_SCHEMA\"]}]},\"depends_on\":[{\"node\":\"resources.schemas.foo\",\"label\":\"${resources.schemas.foo.id}\"}]}", + "status": "OPERATION_STATUS_SUCCEEDED" } } { @@ -118,9 +120,11 @@ Resources: 0 created, 1 changed, 0 deleted, 1 unchanged "update_mask": "state,error_message,resource_id,status" }, "body": { + "error_message": "", "resource_id": "schema/main.dms_emptied_resource", - "status": "OPERATION_STATUS_SUCCEEDED", - "sequence_id": "0" + "sequence_id": "0", + "state": "", + "status": "OPERATION_STATUS_SUCCEEDED" } } { diff --git a/acceptance/bundle/dms/existing-state/output.txt b/acceptance/bundle/dms/existing-state/output.txt index cc0edf233e..586c918731 100644 --- a/acceptance/bundle/dms/existing-state/output.txt +++ b/acceptance/bundle/dms/existing-state/output.txt @@ -61,5 +61,5 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged >>> print_requests.py --dms //api/2.0/bundle --oneline {"method": "POST", "path": "/api/2.0/bundle/deployments", "body": {"initial_parent_path": "/Workspace/Users/[USERNAME]/.bundle/dms-existing-state/default/state", "target_name": "default"}} {"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DEPLOY", "target_name": "default", "display_name": "dms-existing-state", "workspace_info": {"file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-existing-state/default/files", "root_path": "/Workspace/Users/[USERNAME]/.bundle/dms-existing-state/default"}, "operations": [{"resource_key": "jobs.one", "action_type": "OPERATION_ACTION_TYPE_CREATE"}]}} -{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.one", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-existing-state/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"one\",\"queue\":{\"enabled\":true}}}", "resource_id": "[NUMID]", "status": "OPERATION_STATUS_SUCCEEDED", "sequence_id": "0"}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.one", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-existing-state/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"one\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}} {"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}} diff --git a/acceptance/bundle/dms/failed-delete/databricks.yml b/acceptance/bundle/dms/failed-delete/databricks.yml new file mode 100644 index 0000000000..a5fc958510 --- /dev/null +++ b/acceptance/bundle/dms/failed-delete/databricks.yml @@ -0,0 +1,10 @@ +bundle: + name: dms-failed-delete + +experimental: + record_deployment_history: true + +resources: + jobs: + stuck: + name: stuck diff --git a/acceptance/bundle/dms/failed-delete/out.test.toml b/acceptance/bundle/dms/failed-delete/out.test.toml new file mode 100644 index 0000000000..27ec2a7fcd --- /dev/null +++ b/acceptance/bundle/dms/failed-delete/out.test.toml @@ -0,0 +1,3 @@ +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] +EnvMatrix.DMS = [""] diff --git a/acceptance/bundle/dms/failed-delete/output.txt b/acceptance/bundle/dms/failed-delete/output.txt new file mode 100644 index 0000000000..9b61f6961c --- /dev/null +++ b/acceptance/bundle/dms/failed-delete/output.txt @@ -0,0 +1,150 @@ + +=== Deploy: the job is recorded +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-failed-delete/default/files... +Created jobs.stuck +Files: 5 uploaded, 0 deleted +Resources: 1 created, 0 changed, 0 deleted, 0 unchanged + +>>> print_requests.py --dms //api/2.0/bundle +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments", + "body": { + "initial_parent_path": "/Workspace/Users/[USERNAME]/.bundle/dms-failed-delete/default/state", + "target_name": "default" + } +} +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[NUMID]/versions", + "q": { + "version_id": "1" + }, + "body": { + "cli_version": "[CLI_VERSION]", + "version_type": "VERSION_TYPE_DEPLOY", + "target_name": "default", + "display_name": "dms-failed-delete", + "workspace_info": { + "file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-failed-delete/default/files", + "root_path": "/Workspace/Users/[USERNAME]/.bundle/dms-failed-delete/default" + }, + "operations": [ + { + "resource_key": "jobs.stuck", + "action_type": "OPERATION_ACTION_TYPE_CREATE" + } + ] + } +} +{ + "method": "PATCH", + "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.stuck", + "q": { + "update_mask": "state,error_message,resource_id,status" + }, + "body": { + "error_message": "", + "resource_id": "[NUMID]", + "sequence_id": "0", + "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-failed-delete/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"stuck\",\"queue\":{\"enabled\":true}}}", + "status": "OPERATION_STATUS_SUCCEEDED" + } +} +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/complete", + "body": { + "completion_reason": "VERSION_COMPLETE_SUCCESS" + } +} + +=== A resource that fails to delete is recorded as a failed operation carrying the error, so the history says why rather than showing the resource gone +>>> musterr [CLI] bundle destroy --auto-approve +The following resources will be deleted: + delete resources.jobs.stuck + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/dms-failed-delete/default + +Error: cannot delete resources.jobs.stuck: deleting id=[NUMID]: job is not deletable (400 INVALID_PARAMETER_VALUE) + +Endpoint: POST [DATABRICKS_URL]/api/2.2/jobs/delete +HTTP Status: 400 Bad Request +API error_code: INVALID_PARAMETER_VALUE +API message: job is not deletable + + +>>> print_requests.py --dms //api/2.0/bundle +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[NUMID]/versions", + "q": { + "version_id": "2" + }, + "body": { + "cli_version": "[CLI_VERSION]", + "version_type": "VERSION_TYPE_DESTROY", + "target_name": "default", + "display_name": "dms-failed-delete", + "previous_version_id": "1", + "workspace_info": { + "file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-failed-delete/default/files", + "root_path": "/Workspace/Users/[USERNAME]/.bundle/dms-failed-delete/default" + }, + "operations": [ + { + "resource_key": "jobs.stuck", + "action_type": "OPERATION_ACTION_TYPE_DELETE" + } + ] + } +} +{ + "method": "PATCH", + "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/operations/jobs.stuck", + "q": { + "update_mask": "error_message,resource_id,status" + }, + "body": { + "error_message": "deleting id=[NUMID]: job is not deletable (400 INVALID_PARAMETER_VALUE)", + "resource_id": "[NUMID]", + "sequence_id": "0", + "status": "OPERATION_STATUS_FAILED" + } +} +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/complete", + "body": { + "completion_reason": "VERSION_COMPLETE_FAILURE" + } +} + +>>> print_state.py +{ + "state_version": 2, + "cli_version": "[CLI_VERSION]", + "lineage": "[UUID]", + "serial": 1, + "state": { + "resources.jobs.stuck": { + "__id__": "[NUMID]", + "state": { + "deployment": { + "deployment_id": "[NUMID]", + "kind": "BUNDLE", + "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-failed-delete/default/state/metadata.json", + "version_id": "1" + }, + "edit_mode": "UI_LOCKED", + "format": "MULTI_TASK", + "max_concurrent_runs": 1, + "name": "stuck", + "queue": { + "enabled": true + } + } + } + } +} diff --git a/acceptance/bundle/dms/failed-delete/script b/acceptance/bundle/dms/failed-delete/script new file mode 100644 index 0000000000..5d2deabce5 --- /dev/null +++ b/acceptance/bundle/dms/failed-delete/script @@ -0,0 +1,9 @@ +title "Deploy: the job is recorded" +trace $CLI bundle deploy +trace print_requests.py --dms //api/2.0/bundle + +title "A resource that fails to delete is recorded as a failed operation carrying the error, so the history says why rather than showing the resource gone" +# The state entry survives too, so the next deploy still knows about the resource. +trace musterr $CLI bundle destroy --auto-approve +trace print_requests.py --dms //api/2.0/bundle +trace print_state.py diff --git a/acceptance/bundle/dms/failed-delete/test.toml b/acceptance/bundle/dms/failed-delete/test.toml new file mode 100644 index 0000000000..6f2ea43256 --- /dev/null +++ b/acceptance/bundle/dms/failed-delete/test.toml @@ -0,0 +1,6 @@ +# The job cannot be deleted, so the destroy fails and the operation is recorded as failed +# rather than left as if the resource were gone. +[[Server]] +Pattern = "POST /api/2.2/jobs/delete" +Response.StatusCode = 400 +Response.Body = '''{"error_code": "INVALID_PARAMETER_VALUE", "message": "job is not deletable"}''' diff --git a/acceptance/bundle/dms/failed-recreate/output.txt b/acceptance/bundle/dms/failed-recreate/output.txt index 408c04a233..4f05615b2f 100644 --- a/acceptance/bundle/dms/failed-recreate/output.txt +++ b/acceptance/bundle/dms/failed-recreate/output.txt @@ -57,21 +57,24 @@ Files: 3 uploaded, 0 deleted "update_mask": "state,error_message,resource_id,status" }, "body": { + "error_message": "", "resource_id": "main.dms_failed_recreate_schema", - "status": "OPERATION_STATUS_IN_PROGRESS", - "sequence_id": "0" + "sequence_id": "0", + "state": "", + "status": "OPERATION_STATUS_IN_PROGRESS" } } { "method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/operations/schemas.foo", "q": { - "update_mask": "error_message,status" + "update_mask": "error_message,resource_id,status" }, "body": { "error_message": "Fault injected by test. (400 INVALID_PARAMETER_VALUE)", - "status": "OPERATION_STATUS_FAILED", - "sequence_id": "1" + "resource_id": "", + "sequence_id": "1", + "status": "OPERATION_STATUS_FAILED" } } { diff --git a/acceptance/bundle/dms/failed-update/output.txt b/acceptance/bundle/dms/failed-update/output.txt index bb3a928e72..1448218f94 100644 --- a/acceptance/bundle/dms/failed-update/output.txt +++ b/acceptance/bundle/dms/failed-update/output.txt @@ -51,12 +51,13 @@ Files: 3 uploaded, 0 deleted "method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/operations/schemas.foo", "q": { - "update_mask": "error_message,status" + "update_mask": "error_message,resource_id,status" }, "body": { "error_message": "updating id=main.dms_failed_update_schema: Fault injected by test. (400 INVALID_PARAMETER_VALUE)", - "status": "OPERATION_STATUS_FAILED", - "sequence_id": "0" + "resource_id": "main.dms_failed_update_schema", + "sequence_id": "0", + "status": "OPERATION_STATUS_FAILED" } } { diff --git a/acceptance/bundle/dms/multiple-resources/output.txt b/acceptance/bundle/dms/multiple-resources/output.txt index ae8dc28844..bdb583d17b 100644 --- a/acceptance/bundle/dms/multiple-resources/output.txt +++ b/acceptance/bundle/dms/multiple-resources/output.txt @@ -11,11 +11,11 @@ Files: 4 uploaded, 0 deleted Resources: 5 created, 0 changed, 0 deleted, 0 unchanged >>> print_requests.py --dms //versions/1/operations --sort --del-body state --oneline -{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.five", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"resource_id": "[NUMID]", "status": "OPERATION_STATUS_SUCCEEDED", "sequence_id": "0"}} -{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.four", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"resource_id": "[NUMID]", "status": "OPERATION_STATUS_SUCCEEDED", "sequence_id": "0"}} -{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.one", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"resource_id": "[NUMID]", "status": "OPERATION_STATUS_SUCCEEDED", "sequence_id": "0"}} -{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.three", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"resource_id": "[NUMID]", "status": "OPERATION_STATUS_SUCCEEDED", "sequence_id": "0"}} -{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.two", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"resource_id": "[NUMID]", "status": "OPERATION_STATUS_SUCCEEDED", "sequence_id": "0"}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.five", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.four", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.one", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.three", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.two", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "status": "OPERATION_STATUS_SUCCEEDED"}} === Redeploy with no changes: nothing is applied, so no operations are recorded and only the version is opened and completed >>> [CLI] bundle deploy diff --git a/acceptance/bundle/dms/operation-upload-fails-once/databricks.yml b/acceptance/bundle/dms/operation-upload-fails-once/databricks.yml new file mode 100644 index 0000000000..df9f927f92 --- /dev/null +++ b/acceptance/bundle/dms/operation-upload-fails-once/databricks.yml @@ -0,0 +1,12 @@ +bundle: + name: dms-upload-fails-once + +experimental: + record_deployment_history: true + +resources: + schemas: + foo: + name: dms_upload_fails_once_schema + catalog_name: main + comment: v1 diff --git a/acceptance/bundle/dms/operation-upload-fails-once/out.test.toml b/acceptance/bundle/dms/operation-upload-fails-once/out.test.toml new file mode 100644 index 0000000000..27ec2a7fcd --- /dev/null +++ b/acceptance/bundle/dms/operation-upload-fails-once/out.test.toml @@ -0,0 +1,3 @@ +Cloud = false +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] +EnvMatrix.DMS = [""] diff --git a/acceptance/bundle/dms/operation-upload-fails-once/output.txt b/acceptance/bundle/dms/operation-upload-fails-once/output.txt new file mode 100644 index 0000000000..66e8586d55 --- /dev/null +++ b/acceptance/bundle/dms/operation-upload-fails-once/output.txt @@ -0,0 +1,87 @@ + +=== Deploy so there is a recorded resource to recreate +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-upload-fails-once/default/files... +Created schemas.foo +Files: 4 uploaded, 0 deleted +Resources: 1 created, 0 changed, 0 deleted, 0 unchanged + +=== A recreate records twice, and the first one fails: the second must still carry the sequence id staging left, because a rejected update earned no new one +>>> fault.py PATCH /api/2.0/bundle/* 500 0 1 + +>>> update_file.py databricks.yml catalog_name: main catalog_name: other + +>>> musterr [CLI] bundle deploy --auto-approve +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-upload-fails-once/default/files... + +This action will result in the deletion or recreation of the following UC schemas. Any underlying data may be lost: + recreate resources.schemas.foo +Error: recording operation for resources.schemas.foo with the deployment metadata service: Fault injected by test. (500 INJECTED) + +Endpoint: PATCH [DATABRICKS_URL]/api/2.0/bundle/deployments/[NUMID]/versions/2/operations/schemas.foo?update_mask=state%2Cerror_message%2Cresource_id%2Cstatus +HTTP Status: 500 Internal Server Error +API error_code: INJECTED +API message: Fault injected by test. + +Files: 3 uploaded, 0 deleted + +>>> print_requests.py --dms //api/2.0/bundle +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[NUMID]/versions", + "q": { + "version_id": "2" + }, + "body": { + "cli_version": "[CLI_VERSION]", + "version_type": "VERSION_TYPE_DEPLOY", + "target_name": "default", + "display_name": "dms-upload-fails-once", + "previous_version_id": "1", + "workspace_info": { + "file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-upload-fails-once/default/files", + "root_path": "/Workspace/Users/[USERNAME]/.bundle/dms-upload-fails-once/default" + }, + "operations": [ + { + "resource_key": "schemas.foo", + "action_type": "OPERATION_ACTION_TYPE_RECREATE" + } + ] + } +} +{ + "method": "PATCH", + "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/operations/schemas.foo", + "q": { + "update_mask": "state,error_message,resource_id,status" + }, + "body": { + "error_message": "", + "resource_id": "main.dms_upload_fails_once_schema", + "sequence_id": "0", + "state": "", + "status": "OPERATION_STATUS_IN_PROGRESS" + } +} +{ + "method": "PATCH", + "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/operations/schemas.foo", + "q": { + "update_mask": "state,error_message,resource_id,status" + }, + "body": { + "error_message": "", + "resource_id": "other.dms_upload_fails_once_schema", + "sequence_id": "0", + "state": "{\"state\":{\"catalog_name\":\"other\",\"comment\":\"v1\",\"name\":\"dms_upload_fails_once_schema\"}}", + "status": "OPERATION_STATUS_SUCCEEDED" + } +} +{ + "method": "POST", + "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/complete", + "body": { + "completion_reason": "VERSION_COMPLETE_FAILURE" + } +} diff --git a/acceptance/bundle/dms/operation-upload-fails-once/script b/acceptance/bundle/dms/operation-upload-fails-once/script new file mode 100644 index 0000000000..2040861723 --- /dev/null +++ b/acceptance/bundle/dms/operation-upload-fails-once/script @@ -0,0 +1,9 @@ +title "Deploy so there is a recorded resource to recreate" +trace $CLI bundle deploy +rm -f out.requests.txt + +title "A recreate records twice, and the first one fails: the second must still carry the sequence id staging left, because a rejected update earned no new one" +trace fault.py "PATCH /api/2.0/bundle/*" 500 0 1 +trace update_file.py databricks.yml "catalog_name: main" "catalog_name: other" +trace musterr $CLI bundle deploy --auto-approve +trace print_requests.py --dms //api/2.0/bundle diff --git a/acceptance/bundle/dms/provenance/output.txt b/acceptance/bundle/dms/provenance/output.txt index 61723573fc..2114c10bcf 100644 --- a/acceptance/bundle/dms/provenance/output.txt +++ b/acceptance/bundle/dms/provenance/output.txt @@ -43,10 +43,11 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged "update_mask": "state,error_message,resource_id,status" }, "body": { - "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-provenance/dev/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":4,\"name\":\"[dev [USERNAME]] foo\",\"queue\":{\"enabled\":true},\"tags\":{\"dev\":\"[USERNAME]\"}}}", + "error_message": "", "resource_id": "[NUMID]", - "status": "OPERATION_STATUS_SUCCEEDED", - "sequence_id": "0" + "sequence_id": "0", + "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-provenance/dev/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":4,\"name\":\"[dev [USERNAME]] foo\",\"queue\":{\"enabled\":true},\"tags\":{\"dev\":\"[USERNAME]\"}}}", + "status": "OPERATION_STATUS_SUCCEEDED" } } { diff --git a/acceptance/bundle/dms/record-failure/output.txt b/acceptance/bundle/dms/record-failure/output.txt index efea62c8a2..7a1ec70eb4 100644 --- a/acceptance/bundle/dms/record-failure/output.txt +++ b/acceptance/bundle/dms/record-failure/output.txt @@ -47,12 +47,13 @@ Files: 5 uploaded, 0 deleted "method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.doomed", "q": { - "update_mask": "error_message,status" + "update_mask": "error_message,resource_id,status" }, "body": { "error_message": "cluster spec is invalid (400 INVALID_PARAMETER_VALUE)", - "status": "OPERATION_STATUS_FAILED", - "sequence_id": "0" + "resource_id": "", + "sequence_id": "0", + "status": "OPERATION_STATUS_FAILED" } } { diff --git a/acceptance/bundle/dms/record/output.txt b/acceptance/bundle/dms/record/output.txt index 4f7a6f8bff..6462839f4b 100644 --- a/acceptance/bundle/dms/record/output.txt +++ b/acceptance/bundle/dms/record/output.txt @@ -45,10 +45,11 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged "update_mask": "state,error_message,resource_id,status" }, "body": { - "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-record/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"foo\",\"queue\":{\"enabled\":true}}}", + "error_message": "", "resource_id": "[NUMID]", - "status": "OPERATION_STATUS_SUCCEEDED", - "sequence_id": "0" + "sequence_id": "0", + "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-record/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"foo\",\"queue\":{\"enabled\":true}}}", + "status": "OPERATION_STATUS_SUCCEEDED" } } { @@ -143,9 +144,11 @@ Destroy: 1 deleted "update_mask": "state,error_message,resource_id,status" }, "body": { + "error_message": "", "resource_id": "[NUMID]", - "status": "OPERATION_STATUS_SUCCEEDED", - "sequence_id": "0" + "sequence_id": "0", + "state": "", + "status": "OPERATION_STATUS_SUCCEEDED" } } { @@ -215,10 +218,11 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged "update_mask": "state,error_message,resource_id,status" }, "body": { - "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-record/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"foo\",\"queue\":{\"enabled\":true}}}", + "error_message": "", "resource_id": "[NUMID]", - "status": "OPERATION_STATUS_SUCCEEDED", - "sequence_id": "0" + "sequence_id": "0", + "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-record/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"foo\",\"queue\":{\"enabled\":true}}}", + "status": "OPERATION_STATUS_SUCCEEDED" } } { diff --git a/acceptance/bundle/dms/resource-lifecycle/output.txt b/acceptance/bundle/dms/resource-lifecycle/output.txt index 7bdb9677c4..6d557430e6 100644 --- a/acceptance/bundle/dms/resource-lifecycle/output.txt +++ b/acceptance/bundle/dms/resource-lifecycle/output.txt @@ -45,10 +45,11 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged "update_mask": "state,error_message,resource_id,status" }, "body": { - "state": "{\"state\":{\"catalog_name\":\"main\",\"comment\":\"v1\",\"name\":\"dms_resource_lifecycle_schema\"}}", + "error_message": "", "resource_id": "main.dms_resource_lifecycle_schema", - "status": "OPERATION_STATUS_SUCCEEDED", - "sequence_id": "0" + "sequence_id": "0", + "state": "{\"state\":{\"catalog_name\":\"main\",\"comment\":\"v1\",\"name\":\"dms_resource_lifecycle_schema\"}}", + "status": "OPERATION_STATUS_SUCCEEDED" } } { @@ -103,9 +104,11 @@ Resources: 1 created, 0 changed, 1 deleted, 0 unchanged "update_mask": "state,error_message,resource_id,status" }, "body": { + "error_message": "", "resource_id": "main.dms_resource_lifecycle_schema", - "status": "OPERATION_STATUS_IN_PROGRESS", - "sequence_id": "0" + "sequence_id": "0", + "state": "", + "status": "OPERATION_STATUS_IN_PROGRESS" } } { @@ -115,10 +118,11 @@ Resources: 1 created, 0 changed, 1 deleted, 0 unchanged "update_mask": "state,error_message,resource_id,status" }, "body": { - "state": "{\"state\":{\"catalog_name\":\"other\",\"comment\":\"v1\",\"name\":\"dms_resource_lifecycle_schema\"}}", + "error_message": "", "resource_id": "other.dms_resource_lifecycle_schema", - "status": "OPERATION_STATUS_SUCCEEDED", - "sequence_id": "1" + "sequence_id": "1", + "state": "{\"state\":{\"catalog_name\":\"other\",\"comment\":\"v1\",\"name\":\"dms_resource_lifecycle_schema\"}}", + "status": "OPERATION_STATUS_SUCCEEDED" } } { @@ -173,9 +177,11 @@ Destroy: 1 deleted "update_mask": "state,error_message,resource_id,status" }, "body": { + "error_message": "", "resource_id": "other.dms_resource_lifecycle_schema", - "status": "OPERATION_STATUS_SUCCEEDED", - "sequence_id": "0" + "sequence_id": "0", + "state": "", + "status": "OPERATION_STATUS_SUCCEEDED" } } { diff --git a/bundle/direct/bundle_apply.go b/bundle/direct/bundle_apply.go index 454a4ac381..f97b404dcc 100644 --- a/bundle/direct/bundle_apply.go +++ b/bundle/direct/bundle_apply.go @@ -46,7 +46,7 @@ func (b *DeploymentBundle) Apply(ctx context.Context, client *databricks.Workspa // The state DB records every write with DMS from here on, so the service mirrors the WAL. // Writes go out on one background goroutine, off the apply path, and are drained below // once every worker has finished recording. - b.StateDB.StartRecording(ctx, b.OpRec) + b.StateDB.StartRecording(b.OpRec) g.Run(defaultParallelism, func(resourceKey string, failedDependency *string) bool { entry, err := plan.WriteLockEntry(resourceKey) diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go index 2823c4c070..0be6798d43 100644 --- a/bundle/direct/dstate/state.go +++ b/bundle/direct/dstate/state.go @@ -82,17 +82,15 @@ type DeploymentState struct { DMSDeploymentID string } -// StartRecording has every subsequent state write recorded with DMS through writer, so what -// the service holds mirrors the WAL. A nil writer records nothing, which is what a bundle that -// does not record deployment history passes. It is called once the version exists, which is why -// it is not an Open option, and ctx must outlive FinishRecording. -func (db *DeploymentState) StartRecording(ctx context.Context, writer dms.OperationWriter) { - if writer == nil { +// StartRecording has every subsequent state write recorded with DMS through sink, so what the +// service holds mirrors the WAL. A nil sink records nothing, which is what a bundle that does +// not record deployment history passes. It is called once the version exists, which is why it +// is not an Open option. +func (db *DeploymentState) StartRecording(sink *dms.OperationSink) { + if sink == nil { return } - sink := dms.NewOperationSink(ctx, writer) - db.mu.Lock() defer db.mu.Unlock() db.sink = sink diff --git a/bundle/direct/dstate/state_test.go b/bundle/direct/dstate/state_test.go index c54c1c1867..8cb5d0e74a 100644 --- a/bundle/direct/dstate/state_test.go +++ b/bundle/direct/dstate/state_test.go @@ -2,7 +2,6 @@ package dstate import ( "encoding/json" - "errors" "os" "path/filepath" "testing" @@ -18,25 +17,6 @@ func mustFinalize(t *testing.T, db *DeploymentState) { require.NoError(t, err) } -func TestStateWritesRecordNothingWithoutSink(t *testing.T) { - path := filepath.Join(t.TempDir(), "state.json") - - var db DeploymentState - require.NoError(t, db.Open(t.Context(), path, WithRecovery(true), WithWrite(true), nil)) - - // A bundle that does not record deployment history has no writer, so no sink is - // installed and every recording call is a no-op. - db.StartRecording(t.Context(), nil) - require.Nil(t, db.recorder()) - - require.NoError(t, db.SaveState(t.Context(), "jobs.my_job", "123", map[string]string{}, nil)) - require.NoError(t, db.DeleteState(t.Context(), "jobs.my_job")) - db.RecordFailure("jobs.my_job", "123", errors.New("boom")) - require.NoError(t, db.RecordingErr()) - require.NoError(t, db.FinishRecording()) - mustFinalize(t, &db) -} - func TestOpenSaveFinalizeRoundTrip(t *testing.T) { path := filepath.Join(t.TempDir(), "state.json") diff --git a/bundle/direct/pkg.go b/bundle/direct/pkg.go index 9d4618097d..ea9af780da 100644 --- a/bundle/direct/pkg.go +++ b/bundle/direct/pkg.go @@ -55,7 +55,7 @@ type DeploymentBundle struct { // OpRec records applied operations with DMS. Nil unless the bundle records deployment // history, in which case the deploy phase sets it once the version exists. Apply drains // it before returning. - OpRec dms.OperationWriter + OpRec *dms.OperationSink } // SetRemoteState updates the remote state with type validation and marks as fresh. diff --git a/bundle/phases/deploy.go b/bundle/phases/deploy.go index fed0de8560..81b418b5d8 100644 --- a/bundle/phases/deploy.go +++ b/bundle/phases/deploy.go @@ -355,7 +355,7 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand logdiag.LogError(ctx, err) return } - writer, err := recording.Start(ctx, staged) + sink, err := recording.Start(ctx, staged) if err != nil { logdiag.LogError(ctx, err) return @@ -363,7 +363,7 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand logDeploymentVersion(ctx, b, recording) // Record operations under that version, so DMS holds the deployed resource state. - b.DeploymentBundle.OpRec = writer + b.DeploymentBundle.OpRec = sink deployCore(ctx, b, plan, stateEngine, requestedEngine) if logdiag.HasError(ctx) { diff --git a/bundle/phases/destroy.go b/bundle/phases/destroy.go index 62fc0a7b09..5833d73cd5 100644 --- a/bundle/phases/destroy.go +++ b/bundle/phases/destroy.go @@ -277,12 +277,12 @@ func Destroy(ctx context.Context, b *bundle.Bundle, engine engine.EngineType) { logdiag.LogError(ctx, err) return } - writer, err := recording.Start(ctx, staged) + sink, err := recording.Start(ctx, staged) if err != nil { logdiag.LogError(ctx, err) return } - b.DeploymentBundle.OpRec = writer + b.DeploymentBundle.OpRec = sink destroyCore(ctx, b, plan, engine, recording) } else { cmdio.LogString(ctx, "Destroy cancelled!") diff --git a/libs/dms/client.go b/libs/dms/client.go index 4068fe40cd..af9c0f6b16 100644 --- a/libs/dms/client.go +++ b/libs/dms/client.go @@ -154,16 +154,6 @@ type StagedOperation struct { ActionType bundledeployments.OperationActionType `json:"action_type"` } -// updateOperationRequest carries the values an update writes. action_type and resource_key -// are left out: the service fixes them when the version stages the operation. -type updateOperationRequest struct { - State string `json:"state,omitempty"` - ErrorMessage string `json:"error_message,omitempty"` - ResourceId string `json:"resource_id,omitempty"` - Status bundledeployments.OperationStatus `json:"status,omitempty"` - SequenceId string `json:"sequence_id,omitempty"` -} - // operationResponse is the part of an operation response the CLI reads back. type operationResponse struct { // SequenceId is the concurrency token for the next update, typed as the service sends it. @@ -188,22 +178,25 @@ func (r *rawClient) CreateVersion(ctx context.Context, deploymentID, versionID s return &version, nil } -// newUpdateRequest builds the request body for update. Each field is sent because the mask -// names it: the service ignores the rest, and state is the largest field by far, so a -// failure that keeps the recorded state sends none of it. -func newUpdateRequest(update OperationUpdate, sequenceID string) updateOperationRequest { - body := updateOperationRequest{SequenceId: sequenceID} +// newUpdateRequest builds the request body for update. A field the mask names is always +// sent, even when empty: the service rejects a masked path the body omits, and an empty +// value is how a field is cleared. A field the mask leaves out is absent entirely - state +// is the largest by far, so a failure that keeps the recorded state sends none of it. +// action_type and resource_key are never sent: the service fixes them when the version +// stages the operation. +func newUpdateRequest(update OperationUpdate, sequenceID string) map[string]any { + body := map[string]any{"sequence_id": sequenceID} if update.Fields.Has(FieldState) { - body.State = string(update.State) + body[fieldNameState] = string(update.State) } if update.Fields.Has(FieldResourceID) { - body.ResourceId = update.ResourceID + body[fieldNameResourceID] = update.ResourceID } if update.Fields.Has(FieldErrorMessage) { - body.ErrorMessage = update.ErrorMessage + body[fieldNameErrorMessage] = update.ErrorMessage } if update.Fields.Has(FieldStatus) { - body.Status = update.Status + body[fieldNameStatus] = update.Status } return body } diff --git a/libs/dms/client_test.go b/libs/dms/client_test.go index f24db9f219..f1225062d0 100644 --- a/libs/dms/client_test.go +++ b/libs/dms/client_test.go @@ -108,55 +108,39 @@ func TestDeploymentIDFromName(t *testing.T) { } func TestUpdateRequestSendsAFieldOnlyWhenTheMaskNamesIt(t *testing.T) { - // Every case carries the same values, so what reaches the body is decided by the mask - // alone. A failure sending state would drop the resource from the deployment, and - // resource_id does not ride along with state. + // The masks the CLI builds are asserted on the wire by acceptance/bundle/dms. What that + // cannot show is that each field is gated on its own bit: state is the one whose absence + // would drop the resource from the deployment, and resource_id must not ride along with it. update := OperationUpdate{ + Fields: FieldResourceID | FieldStatus, State: json.RawMessage(`{"state":{"name":"foo"}}`), ResourceID: "job-1", Status: bundledeployments.OperationStatusOperationStatusSucceeded, ErrorMessage: "boom", } - tests := []struct { - name string - fields Fields - want updateOperationRequest - }{ - { - name: "a write that describes the resource", - fields: DescribesResource, - want: updateOperationRequest{ - State: `{"state":{"name":"foo"}}`, - ResourceId: "job-1", - Status: bundledeployments.OperationStatusOperationStatusSucceeded, - ErrorMessage: "boom", - SequenceId: "3", - }, - }, - { - name: "a failure that keeps the recorded state", - fields: KeepsState, - want: updateOperationRequest{ - Status: bundledeployments.OperationStatusOperationStatusSucceeded, - ErrorMessage: "boom", - SequenceId: "3", - }, - }, - { - name: "resource_id without state", - fields: FieldResourceID, - want: updateOperationRequest{ - ResourceId: "job-1", - SequenceId: "3", - }, - }, - } + assert.Equal(t, map[string]any{ + "resource_id": "job-1", + "status": bundledeployments.OperationStatusOperationStatusSucceeded, + "sequence_id": "3", + }, newUpdateRequest(update, "3")) +} - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - update.Fields = tt.fields - assert.Equal(t, tt.want, newUpdateRequest(update, "3")) - }) +func TestUpdateRequestSendsAMaskedFieldEvenWhenEmpty(t *testing.T) { + // The service rejects a masked path the body omits, so a success - which names + // error_message in its mask and has none - must still send it as an empty value. + update := OperationUpdate{ + Fields: DescribesResource, + State: json.RawMessage(`{"name":"foo"}`), + ResourceID: "job-1", + Status: bundledeployments.OperationStatusOperationStatusSucceeded, } + + assert.Equal(t, map[string]any{ + "state": `{"name":"foo"}`, + "resource_id": "job-1", + "error_message": "", + "status": bundledeployments.OperationStatusOperationStatusSucceeded, + "sequence_id": "0", + }, newUpdateRequest(update, "0")) } diff --git a/libs/dms/fields.go b/libs/dms/fields.go index e3bd271c3b..6188e2e3e0 100644 --- a/libs/dms/fields.go +++ b/libs/dms/fields.go @@ -17,20 +17,31 @@ const ( // an update may change. const DescribesResource = FieldState | FieldErrorMessage | FieldResourceID | FieldStatus -// KeepsState is what a failure claims: mark it failed and leave state alone. State means -// the resource is as it was written; no state means a delete went through and nothing -// replaced it, so the resource really is gone and the deployment should say so. -const KeepsState = FieldErrorMessage | FieldStatus +// KeepsState is what a failure claims: mark it failed, name the resource it failed on, and +// leave state alone. State means the resource is as it was written; no state means a delete +// went through and nothing replaced it, so the resource really is gone and the deployment +// should say so. The id is not optional: the service refuses an update to a delete operation +// that does not name the resource. +const KeepsState = FieldErrorMessage | FieldResourceID | FieldStatus + +// Field names on the wire, shared by the update mask and the update body so the two +// cannot drift: the service rejects a masked path the body leaves out. +const ( + fieldNameState = "state" + fieldNameErrorMessage = "error_message" + fieldNameResourceID = "resource_id" + fieldNameStatus = "status" +) // wireNames pairs each field with its name on the wire, in the order a mask lists them. var wireNames = []struct { field Fields name string }{ - {FieldState, "state"}, - {FieldErrorMessage, "error_message"}, - {FieldResourceID, "resource_id"}, - {FieldStatus, "status"}, + {FieldState, fieldNameState}, + {FieldErrorMessage, fieldNameErrorMessage}, + {FieldResourceID, fieldNameResourceID}, + {FieldStatus, fieldNameStatus}, } // Has reports whether f contains every field in other. diff --git a/libs/dms/fields_test.go b/libs/dms/fields_test.go index 198a55efca..a50bb81e30 100644 --- a/libs/dms/fields_test.go +++ b/libs/dms/fields_test.go @@ -9,7 +9,7 @@ import ( func TestFieldsMask(t *testing.T) { // The order is fixed, so the same set always sends the same mask. assert.Equal(t, "state,error_message,resource_id,status", DescribesResource.Mask()) - assert.Equal(t, "error_message,status", KeepsState.Mask()) + assert.Equal(t, "error_message,resource_id,status", KeepsState.Mask()) assert.Equal(t, "state", FieldState.Mask()) assert.Empty(t, Fields(0).Mask()) } diff --git a/libs/dms/operation_test.go b/libs/dms/operation_test.go index d48c63a334..ee28b3019c 100644 --- a/libs/dms/operation_test.go +++ b/libs/dms/operation_test.go @@ -12,45 +12,9 @@ import ( "github.com/stretchr/testify/require" ) -func TestNewStateUpdateRecordsEnvelopeAsIs(t *testing.T) { - // The state DB serializes the envelope (see dstate.SaveState); the update carries it - // through untouched, sensitive fields and all. - state := json.RawMessage(`{"state":{"name":"foo","token":"super-secret"}}`) - - update, err := NewStateUpdate("job-123", state, false) - require.NoError(t, err) - - assert.JSONEq(t, string(state), string(update.State)) - assert.Equal(t, bundledeployments.OperationStatusOperationStatusSucceeded, update.Status) - assert.Equal(t, DescribesResource, update.Fields) -} - -func TestNewStateUpdateInProgressIsNotFinished(t *testing.T) { - // A recreate's delete is half of a larger change, so an interrupted deploy must not - // leave the resource described as finished. - update, err := NewStateUpdate("", nil, true) - require.NoError(t, err) - - assert.Equal(t, StatusInProgress, update.Status) -} - -func TestNewStateUpdateRejectsOversizedState(t *testing.T) { - big := json.RawMessage(strings.Repeat("x", maxStateSize+1)) - - _, err := NewStateUpdate("job-123", big, false) - assert.ErrorContains(t, err, "exceeds the 65536 byte limit") -} - -func TestNewFailureUpdateRecordsError(t *testing.T) { - update := NewFailureUpdate("", errors.New("cluster spec is invalid")) - - assert.Equal(t, bundledeployments.OperationStatusOperationStatusFailed, update.Status) - assert.Equal(t, "cluster spec is invalid", update.ErrorMessage) - // The resource was never written, so there is no state to serve back for it. - assert.Nil(t, update.State) - // The update only marks the operation failed; see KeepsState. - assert.Equal(t, KeepsState, update.Fields) -} +// What these updates put on the wire - the mask, the status, the state a write carries and a +// failure leaves alone - is asserted by acceptance/bundle/dms. What is left here are the limits +// and the merge, which a deploy cannot reach. func TestNewFailureUpdateTruncatesLongError(t *testing.T) { // Truncated rather than rejected: a message over the limit would make recording @@ -90,18 +54,18 @@ func TestMergeLetsAWriteSupersedeAFailure(t *testing.T) { } func TestMergeKeepsTheWritesStateAndMask(t *testing.T) { - // A failure claims only status and error_message, so the write's state, id and mask - // survive. - write, err := NewStateUpdate("id-new", nil, false) + // A failure claims status, error_message and the id, so the state the write recorded and + // the mask that carries it survive: the resource stays listed as it was written. + write, err := NewStateUpdate("id-1", json.RawMessage(`{"state":{"name":"before"}}`), false) require.NoError(t, err) - failed := NewFailureUpdate("id-old", errors.New("boom")) + failed := NewFailureUpdate("id-1", errors.New("boom")) merged := write.Merge(failed) assert.Equal(t, bundledeployments.OperationStatusOperationStatusFailed, merged.Status) assert.Equal(t, "boom", merged.ErrorMessage) - assert.Equal(t, "id-new", merged.ResourceID) - assert.Nil(t, merged.State) + assert.Equal(t, "id-1", merged.ResourceID) + assert.JSONEq(t, `{"state":{"name":"before"}}`, string(merged.State)) assert.Equal(t, DescribesResource, merged.Fields) } diff --git a/libs/dms/recording.go b/libs/dms/recording.go index f717c7c081..50b9ac1209 100644 --- a/libs/dms/recording.go +++ b/libs/dms/recording.go @@ -41,11 +41,11 @@ type Recording interface { // Version is the version number Prepare claimed, and zero before it runs. Version() int64 - // Start creates the version, staging an operation for each resource, and returns the - // writer that fills them in - nil when nothing is recorded, which is what leaves the state - // DB without a sink. The staged set is fixed here: the service has no call to add one - // later, so a resource left out can never be recorded. - Start(ctx context.Context, staged []StagedOperation) (OperationWriter, error) + // Start creates the version, staging an operation for each resource, and returns the sink + // that fills them in - nil when nothing is recorded, which is what leaves the state DB + // without one. The staged set is fixed here: the service has no call to add one later, so + // a resource left out can never be recorded. + Start(ctx context.Context, staged []StagedOperation) (*OperationSink, error) // Finish completes the version. It is a no-op before Start, which is what lets a caller // defer it and still not complete a version a cancelled deploy never created, and it is @@ -102,7 +102,7 @@ func Disabled() Recording { } // disabled records nothing. Its Prepare leaves no deployment and no version, and its Start no -// writer, so a caller that stamps a version or installs the writer finds nothing to install. +// sink, so a caller that stamps a version or installs the sink finds nothing to install. type disabled struct{} func (disabled) Prepare(context.Context) error { return nil } @@ -110,7 +110,7 @@ func (disabled) DeploymentID() string { return "" } func (disabled) Version() int64 { return 0 } func (disabled) Finish(context.Context, bool) error { return nil } -func (disabled) Start(context.Context, []StagedOperation) (OperationWriter, error) { +func (disabled) Start(context.Context, []StagedOperation) (*OperationSink, error) { return nil, nil } @@ -158,7 +158,7 @@ func (r *recording) Prepare(ctx context.Context) error { } // Start implements Recording. -func (r *recording) Start(ctx context.Context, staged []StagedOperation) (OperationWriter, error) { +func (r *recording) Start(ctx context.Context, staged []StagedOperation) (*OperationSink, error) { // A deploy calls Prepare itself, because the version number is stamped onto every job and // pipeline before the plan is computed. A destroy creates a version too, but stamps // nothing, so it has no reason to settle the deployment any earlier than here. @@ -200,12 +200,7 @@ func (r *recording) Start(ctx context.Context, staged []StagedOperation) (Operat r.stopHeartbeat = startHeartbeat(ctx, r.client, r.deploymentID, r.versionNum) log.Infof(ctx, "Created deployment version: deployment=%s version=%s", r.deploymentID, version.VersionId) - return &operationWriter{ - client: r.client, - deploymentID: r.deploymentID, - version: r.versionNum, - sequenceIDs: make(map[ResourceKey]string), - }, nil + return newOperationSink(ctx, r.client, r.deploymentID, r.versionNum), nil } // Finish implements Recording. diff --git a/libs/dms/recording_test.go b/libs/dms/recording_test.go index 8339eda02f..57b584b02b 100644 --- a/libs/dms/recording_test.go +++ b/libs/dms/recording_test.go @@ -179,12 +179,12 @@ func TestDisabledRecordingIsNoOp(t *testing.T) { r := Disabled() require.NoError(t, r.Prepare(t.Context())) - writer, err := r.Start(t.Context(), []StagedOperation{{ResourceKey: "jobs.foo"}}) + sink, err := r.Start(t.Context(), []StagedOperation{{ResourceKey: "jobs.foo"}}) require.NoError(t, err) require.NoError(t, r.Finish(t.Context(), true)) assert.Empty(t, r.DeploymentID()) assert.Zero(t, r.Version()) - // No writer, which is what leaves the state DB without a sink and nothing to stamp. - assert.Nil(t, writer) + // No sink, which is what leaves the state DB recording nothing and nothing to stamp. + assert.Nil(t, sink) } diff --git a/libs/dms/sink.go b/libs/dms/sink.go index 8137a1e088..e59a60d32c 100644 --- a/libs/dms/sink.go +++ b/libs/dms/sink.go @@ -12,11 +12,18 @@ import ( // and DMS is what the next plan reads. const operationSinkQueueSize = 10 -// OperationSink writes operations one at a time on a background goroutine, so a deploy never -// waits on a round trip. queue holds bundle state keys, converted where they go on the wire, -// and pending the newest update per key, so a second write for a resource replaces the first. +// stagedSequenceID is what CreateVersion leaves on every operation it stages, and so the +// precondition for the first update of a resource. +const stagedSequenceID = "0" + +// OperationSink fills in the operations one version staged. It writes them one at a time on a +// background goroutine, so a deploy never waits on a round trip. queue holds bundle state keys, +// converted where they go on the wire, and pending the newest update per key, so a second write +// for a resource replaces the first. type OperationSink struct { - writer OperationWriter + client *Client + deploymentID string + version int64 // queue holds the keys that have something waiting. One slot per resource, so a full // queue means the deploy is that many resources ahead and the next write waits. Record @@ -29,6 +36,11 @@ type OperationSink struct { // stopQueue closes the queue, wrapped so Close can safely run twice. stopQueue func() + // sequenceIDs holds the token the last update for a resource returned. A resource absent + // from it has only what staging left, so its first update sends that. Unguarded: run is + // the only goroutine that writes, one update at a time. + sequenceIDs map[ResourceKey]string + // mu guards the fields below. mu sync.Mutex @@ -38,13 +50,16 @@ type OperationSink struct { err error } -// NewOperationSink starts the writer. ctx must outlive Close. -func NewOperationSink(ctx context.Context, writer OperationWriter) *OperationSink { +// newOperationSink starts the writer. ctx must outlive Close. +func newOperationSink(ctx context.Context, client *Client, deploymentID string, version int64) *OperationSink { s := &OperationSink{ - writer: writer, - queue: make(chan string, operationSinkQueueSize), - done: make(chan struct{}), - pending: make(map[string]OperationUpdate), + client: client, + deploymentID: deploymentID, + version: version, + queue: make(chan string, operationSinkQueueSize), + done: make(chan struct{}), + sequenceIDs: make(map[ResourceKey]string), + pending: make(map[string]OperationUpdate), } s.stopQueue = sync.OnceFunc(func() { close(s.queue) }) @@ -111,12 +126,30 @@ func (s *OperationSink) run(ctx context.Context) { } // Keep going after a failure, so one bad write does not drop everything behind it. - if err := s.writer.Write(ctx, KeyFromState(resourceKey), update); err != nil { + if err := s.write(ctx, KeyFromState(resourceKey), update); err != nil { s.setErr(fmt.Errorf("recording operation for %s with the deployment metadata service: %w", resourceKey, err)) } } } +// write sends one update, at the sequence id the resource is at. +func (s *OperationSink) write(ctx context.Context, key ResourceKey, update OperationUpdate) error { + sequenceID, written := s.sequenceIDs[key] + if !written { + sequenceID = stagedSequenceID + } + + next, err := s.client.UpdateOperation(ctx, s.deploymentID, s.version, key, sequenceID, update) + if err != nil { + return err + } + + // The next write for this resource echoes the sequence id this one earned. + s.sequenceIDs[key] = next + + return nil +} + // Close drains what is waiting and returns the first write error, which fails the deploy: // DMS is the source of truth, so a missing record would have the next deploy create a // resource that already exists. Safe to call twice. diff --git a/libs/dms/sink_test.go b/libs/dms/sink_test.go index 2ba76df690..234be54fa8 100644 --- a/libs/dms/sink_test.go +++ b/libs/dms/sink_test.go @@ -1,214 +1,74 @@ package dms import ( - "context" "encoding/json" - "errors" - "strconv" "strings" - "sync" "testing" - "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) -// fakeWriter records the writes it receives and optionally blocks until block is -// closed, so a test can hold the writer and observe what coalesces behind it. -// -// Keys arrive in the DMS form, which is what the sink puts on the wire. -type fakeWriter struct { - block chan struct{} - started chan ResourceKey - err error +// What the service ends up holding is asserted by acceptance/bundle/dms. What is left here is +// what a deploy cannot reach: the queue's coalescing and the size limit. No test here goes +// through the API - the queue is driven directly, so nothing waits on a request. - mu sync.Mutex - writes []string -} - -func (f *fakeWriter) Write(ctx context.Context, key ResourceKey, update OperationUpdate) error { - if f.started != nil { - f.started <- key +// queued builds a sink with nothing draining its queue, so a test drives record and take +// itself and nothing depends on when a background writer runs. +func queued() *OperationSink { + return &OperationSink{ + queue: make(chan string, operationSinkQueueSize), + pending: make(map[string]OperationUpdate), } - if f.block != nil { - <-f.block - } - - f.mu.Lock() - f.writes = append(f.writes, string(key)+"="+string(update.State)) - f.mu.Unlock() - - return f.err -} - -func (f *fakeWriter) recorded() []string { - f.mu.Lock() - defer f.mu.Unlock() - return append([]string(nil), f.writes...) } -// envelope is the serialized state the state DB hands the sink. The sink passes it through -// untouched, so the shape only has to look like what goes on the wire. -func envelope(name string) json.RawMessage { - return json.RawMessage(`{"state":{"name":"` + name + `"}}`) -} - -func recordState(t *testing.T, s *OperationSink, resourceKey, name string) { +func stateUpdate(t *testing.T, name string) OperationUpdate { t.Helper() - s.RecordOperation(t.Context(), resourceKey, false, "id-1", envelope(name)) + update, err := NewStateUpdate("id-1", json.RawMessage(`{"state":{"name":"`+name+`"}}`), false) + require.NoError(t, err) + return update } -func TestOperationSinkKeepsWritingAfterGoingIdle(t *testing.T) { - // The writer parks on an empty queue instead of returning. Apply can spend long stretches - // inside resource CRUD with nothing to record, and a writer that exited while idle would - // silently drop everything recorded after it. - f := &fakeWriter{} - s := NewOperationSink(t.Context(), f) - - recordState(t, s, "resources.jobs.foo", "v1") - require.Eventually(t, func() bool { return len(f.recorded()) == 1 }, 5*time.Second, time.Millisecond) - - // The queue is drained and the writer idle; what is recorded now still has to go. - recordState(t, s, "resources.jobs.bar", "v1") - require.NoError(t, s.Close()) - - assert.Len(t, f.recorded(), 2) -} - -func TestOperationSinkCoalescesWritesBehindAWrite(t *testing.T) { - // Hold the writer on the first write so the two behind it pile up. They carry - // the resource's full state, so only the newest needs to go: the resource costs - // two requests rather than three. - f := &fakeWriter{block: make(chan struct{}), started: make(chan ResourceKey, 2)} - s := NewOperationSink(t.Context(), f) - - recordState(t, s, "resources.jobs.foo", "v1") - assert.Equal(t, ResourceKey("jobs.foo"), <-f.started) - - recordState(t, s, "resources.jobs.foo", "v2") - recordState(t, s, "resources.jobs.foo", "v3") - - close(f.block) - require.NoError(t, s.Close()) - - assert.Equal(t, []string{ - `jobs.foo={"state":{"name":"v1"}}`, - `jobs.foo={"state":{"name":"v3"}}`, - }, f.recorded()) -} - -func TestOperationSinkRecordDuringWriteIsStillWritten(t *testing.T) { - f := &fakeWriter{block: make(chan struct{}), started: make(chan ResourceKey, 2)} - s := NewOperationSink(t.Context(), f) - - recordState(t, s, "resources.jobs.foo", "v1") - assert.Equal(t, ResourceKey("jobs.foo"), <-f.started) - - // The writer has taken this key off pending and is writing it right now. - recordState(t, s, "resources.jobs.foo", "v2") - - close(f.block) - require.NoError(t, s.Close()) - - // Two writes, in order: an in-flight request cannot be recalled, so v2 goes up - // after v1 rather than replacing it. The service ends up with the newest state. - assert.Equal(t, []string{ - `jobs.foo={"state":{"name":"v1"}}`, - `jobs.foo={"state":{"name":"v2"}}`, - }, f.recorded()) - assert.Empty(t, s.pending) -} - -func TestOperationSinkRecordWaitsWhenTheQueueIsFull(t *testing.T) { - // Recording holds the deploy back once every slot is taken. started is buffered for every - // write: nothing reads it after the first, and a writer blocked sending to it would - // never drain the queue. - f := &fakeWriter{block: make(chan struct{}), started: make(chan ResourceKey, operationSinkQueueSize+4)} - s := NewOperationSink(t.Context(), f) - - // One key is taken off the queue and stuck in the writer; the rest fill it. - recordState(t, s, "resources.jobs.busy", "v1") - assert.Equal(t, ResourceKey("jobs.busy"), <-f.started) - for i := range operationSinkQueueSize { - recordState(t, s, "resources.jobs.job"+strconv.Itoa(i), "v1") - } - - // The next distinct resource has nowhere to go until the writer moves on. Called - // directly rather than through recordState: its assertions may only run on the - // test's own goroutine. - late := envelope("v1") - blocked := make(chan struct{}) - go func() { - s.RecordOperation(t.Context(), "resources.jobs.late", false, "id-1", late) - close(blocked) - }() - - select { - case <-blocked: - t.Fatal("recording did not wait for a full queue, so the deploy can outrun the service") - case <-time.After(50 * time.Millisecond): - } - - close(f.block) - <-blocked - require.NoError(t, s.Close()) - - assert.Len(t, f.recorded(), operationSinkQueueSize+2) -} - -func TestOperationSinkReturnsWriteError(t *testing.T) { - writeErr := errors.New("boom") - f := &fakeWriter{err: writeErr} - s := NewOperationSink(t.Context(), f) - - assert.NoError(t, s.FirstErr()) - - recordState(t, s, "resources.jobs.foo", "v1") - - err := s.Close() - require.Error(t, err) - assert.ErrorIs(t, err, writeErr) - assert.ErrorContains(t, err, "resources.jobs.foo") - // Reported after the fact too, so apply can check between resources. - assert.Error(t, s.FirstErr()) -} - -func TestOperationSinkKeepsRecordingAfterWriteError(t *testing.T) { - // One failed write must not drop the records for everything behind it, so DMS - // ends up as close to reality as it can get. - f := &fakeWriter{err: errors.New("boom")} - s := NewOperationSink(t.Context(), f) - - recordState(t, s, "resources.jobs.foo", "v1") - recordState(t, s, "resources.jobs.bar", "v1") - - require.Error(t, s.Close()) - assert.Len(t, f.recorded(), 2) +func TestOperationSinkCoalescesWhileAKeyIsPending(t *testing.T) { + s := queued() + + // Two writes for one resource with nothing draining. They carry the resource's full + // state, so only the newest needs to go: one slot in the queue, not two. + s.record("resources.jobs.foo", stateUpdate(t, "v1")) + s.record("resources.jobs.foo", stateUpdate(t, "v2")) + + assert.Len(t, s.queue, 1) + update, ok := s.take("resources.jobs.foo") + require.True(t, ok) + assert.JSONEq(t, `{"state":{"name":"v2"}}`, string(update.State)) + + // Taken means a request has it, and an in-flight request cannot be recalled, so the next + // write gets its own slot rather than joining it. + s.record("resources.jobs.foo", stateUpdate(t, "v3")) + assert.Len(t, s.queue, 2) + update, ok = s.take("resources.jobs.foo") + require.True(t, ok) + assert.JSONEq(t, `{"state":{"name":"v3"}}`, string(update.State)) } func TestOperationSinkFailsOnOversizedState(t *testing.T) { // The service will not take a state this large, so the resource cannot be recorded. // Failing here says so, where reporting nothing would leave DMS without the resource // and the next plan would create it again. - f := &fakeWriter{} - s := NewOperationSink(t.Context(), f) + s := queued() - s.RecordOperation(t.Context(), "resources.jobs.foo", false, "id-1", envelope(strings.Repeat("x", maxStateSize))) + s.RecordOperation(t.Context(), "resources.jobs.foo", false, "id-1", json.RawMessage(strings.Repeat("x", maxStateSize+1))) - err := s.Close() - require.Error(t, err) - assert.ErrorContains(t, err, "exceeds the 65536 byte limit") - assert.Empty(t, f.recorded()) + assert.ErrorContains(t, s.FirstErr(), "exceeds the 65536 byte limit") + assert.Empty(t, s.queue) + assert.Empty(t, s.pending) } func TestOperationSinkCloseIsIdempotent(t *testing.T) { - f := &fakeWriter{} - s := NewOperationSink(t.Context(), f) - - recordState(t, s, "resources.jobs.foo", "v1") + // Nothing recorded, so no client is needed: this is the second close that must not panic + // on an already closed queue. + s := newOperationSink(t.Context(), nil, "dep-1", 2) require.NoError(t, s.Close()) require.NoError(t, s.Close()) - assert.Len(t, f.recorded(), 1) } diff --git a/libs/dms/writer.go b/libs/dms/writer.go deleted file mode 100644 index 6a59c64b45..0000000000 --- a/libs/dms/writer.go +++ /dev/null @@ -1,49 +0,0 @@ -package dms - -import ( - "context" - "sync" -) - -// stagedSequenceID is what CreateVersion leaves on every operation it stages, and so the -// precondition for the first update of a resource. -const stagedSequenceID = "0" - -// OperationWriter fills in the operations one version staged. Calls for different resources -// may run concurrently. -type OperationWriter interface { - Write(ctx context.Context, key ResourceKey, update OperationUpdate) error -} - -// operationWriter writes through the API, tracking the sequence id each resource is at. -type operationWriter struct { - client *Client - deploymentID string - version int64 - - mu sync.Mutex - // sequenceIDs holds the token the last update for a resource returned. A resource - // absent from it has only what staging left, so its first update sends that. - sequenceIDs map[ResourceKey]string -} - -func (w *operationWriter) Write(ctx context.Context, key ResourceKey, update OperationUpdate) error { - w.mu.Lock() - sequenceID, written := w.sequenceIDs[key] - w.mu.Unlock() - if !written { - sequenceID = stagedSequenceID - } - - next, err := w.client.UpdateOperation(ctx, w.deploymentID, w.version, key, sequenceID, update) - if err != nil { - return err - } - - // The next write for this resource echoes the sequence id this one earned. - w.mu.Lock() - w.sequenceIDs[key] = next - w.mu.Unlock() - - return nil -} diff --git a/libs/dms/writer_test.go b/libs/dms/writer_test.go deleted file mode 100644 index ba92475821..0000000000 --- a/libs/dms/writer_test.go +++ /dev/null @@ -1,81 +0,0 @@ -package dms - -import ( - "encoding/json" - "testing" - - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -// testWriter returns a writer for version 2 of dep-1, recording through raw. -func testWriter(raw *fakeRaw) OperationWriter { - return &operationWriter{ - client: &Client{raw: raw}, - deploymentID: "dep-1", - version: 2, - sequenceIDs: make(map[ResourceKey]string), - } -} - -func writeState(t *testing.T, w OperationWriter, key ResourceKey, resourceID string) { - t.Helper() - update, err := NewStateUpdate(resourceID, json.RawMessage(`{"state":{}}`), false) - require.NoError(t, err) - require.NoError(t, w.Write(t.Context(), key, update)) -} - -func TestWriterSendsTheStagedSequenceThenWhatTheServiceReturns(t *testing.T) { - // One operation per resource per version, so every write updates the same operation: the - // first at the sequence id staging left, each one after at the id the service returned. - f := newFakeRaw("7") - w := testWriter(f) - - update, err := NewStateUpdate("job-123", json.RawMessage(`{"state":{}}`), false) - require.NoError(t, err) - require.NoError(t, w.Write(t.Context(), "jobs.foo", update)) - writeState(t, w, "jobs.foo", "job-456") - - require.Len(t, f.updates, 2) - assert.Equal(t, updaterCall{ - deploymentID: "dep-1", - version: 2, - key: "jobs.foo", - sequenceID: stagedSequenceID, - update: update, - }, f.updates[0]) - assert.Equal(t, "7", f.updates[1].sequenceID) -} - -func TestWriterTracksSequencePerResource(t *testing.T) { - // Each resource has its own staged operation, so each one's first write echoes the staged - // sequence id rather than a sequence another resource earned. - f := newFakeRaw("7") - w := testWriter(f) - - writeState(t, w, "jobs.foo", "id-1") - writeState(t, w, "jobs.bar", "id-2") - - require.Len(t, f.updates, 2) - assert.Equal(t, stagedSequenceID, f.updates[0].sequenceID) - assert.Equal(t, stagedSequenceID, f.updates[1].sequenceID) -} - -func TestWriterErrorKeepsTheSequence(t *testing.T) { - // A failed write leaves the recorded sequence id alone, so the next write for that resource - // still carries the precondition the service last gave us rather than nothing. - f := newFakeRaw("9") - f.failOn = 1 - w := testWriter(f) - - writeState(t, w, "jobs.foo", "job-1") - - update, err := NewStateUpdate("job-2", json.RawMessage(`{"state":{}}`), false) - require.NoError(t, err) - require.ErrorContains(t, w.Write(t.Context(), "jobs.foo", update), "injected error") - - writeState(t, w, "jobs.foo", "job-3") - - require.Len(t, f.updates, 3) - assert.Equal(t, "9", f.updates[2].sequenceID) -} diff --git a/libs/testserver/bundledeployments.go b/libs/testserver/bundledeployments.go index f9f39dee1b..8c4f60bcdc 100644 --- a/libs/testserver/bundledeployments.go +++ b/libs/testserver/bundledeployments.go @@ -345,6 +345,11 @@ func (s *FakeWorkspace) UpdateOperation(req Request, deploymentID, versionID, re if !slices.Contains(dmsUpdatableOperationFields, path) { return dmsInvalidArgument("update_mask path " + path + " is not updatable") } + // A masked path must carry a value; omitting it is an error rather than a + // no-op, so a client cannot silently drop a field it claims to write. + if _, ok := raw[path]; !ok { + return dmsInvalidArgument(path + " is required when '" + path + "' is in update_mask (an empty value clears it)") + } update[path] = true }