From 55217f8736e9b4a7a61f873d9eb9bdc8e4385e02 Mon Sep 17 00:00:00 2001 From: Shreyas Goenka Date: Mon, 24 Aug 2026 16:53:38 +0000 Subject: [PATCH 1/2] dms: fold the operation writer into the sink The sink was the writer's only caller, so the interface bought an indirection and nothing else. Start returns the sink now, which also drops the ctx from StartRecording - the goroutine belongs to whoever created the version. Writes were already serialized by the queue, so the sequence-id map loses its mutex. No sink test goes through the API any more: the queue is driven directly, and what the service sees is asserted by acceptance/bundle/dms - including two new tests, for a delete that fails and for a recording that is rejected once. The first of those found a bug. A failure update claimed no resource_id, so the id it was handed never went out and the service refused the update to a delete operation. KeepsState names the field now: a failure says which resource it failed on, and says so even when there is none to name - a create that never got an id, or a recreate whose delete already dropped it. Co-authored-by: Isaac --- .../bundle/dms/failed-delete/databricks.yml | 10 + .../bundle/dms/failed-delete/out.test.toml | 3 + .../bundle/dms/failed-delete/output.txt | 149 ++++++++++++ acceptance/bundle/dms/failed-delete/script | 9 + acceptance/bundle/dms/failed-delete/test.toml | 6 + .../bundle/dms/failed-recreate/output.txt | 2 +- .../bundle/dms/failed-update/output.txt | 3 +- .../databricks.yml | 12 + .../operation-upload-fails-once/out.test.toml | 3 + .../operation-upload-fails-once/output.txt | 84 +++++++ .../dms/operation-upload-fails-once/script | 9 + .../bundle/dms/record-failure/output.txt | 2 +- bundle/direct/bundle_apply.go | 2 +- bundle/direct/dstate/state.go | 14 +- bundle/direct/dstate/state_test.go | 20 -- bundle/direct/pkg.go | 2 +- bundle/phases/deploy.go | 4 +- bundle/phases/destroy.go | 4 +- libs/dms/client_test.go | 53 +---- libs/dms/fields.go | 10 +- libs/dms/fields_test.go | 2 +- libs/dms/operation_test.go | 54 +---- libs/dms/recording.go | 23 +- libs/dms/recording_test.go | 6 +- libs/dms/sink.go | 55 ++++- libs/dms/sink_test.go | 222 ++++-------------- libs/dms/writer.go | 49 ---- libs/dms/writer_test.go | 81 ------- 28 files changed, 423 insertions(+), 470 deletions(-) create mode 100644 acceptance/bundle/dms/failed-delete/databricks.yml create mode 100644 acceptance/bundle/dms/failed-delete/out.test.toml create mode 100644 acceptance/bundle/dms/failed-delete/output.txt create mode 100644 acceptance/bundle/dms/failed-delete/script create mode 100644 acceptance/bundle/dms/failed-delete/test.toml create mode 100644 acceptance/bundle/dms/operation-upload-fails-once/databricks.yml create mode 100644 acceptance/bundle/dms/operation-upload-fails-once/out.test.toml create mode 100644 acceptance/bundle/dms/operation-upload-fails-once/output.txt create mode 100644 acceptance/bundle/dms/operation-upload-fails-once/script delete mode 100644 libs/dms/writer.go delete mode 100644 libs/dms/writer_test.go 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..9f0e09019a --- /dev/null +++ b/acceptance/bundle/dms/failed-delete/output.txt @@ -0,0 +1,149 @@ + +=== 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": { + "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}}}", + "resource_id": "[NUMID]", + "status": "OPERATION_STATUS_SUCCEEDED", + "sequence_id": "0" + } +} +{ + "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]", + "status": "OPERATION_STATUS_FAILED", + "sequence_id": "0" + } +} +{ + "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..ac634f0dca 100644 --- a/acceptance/bundle/dms/failed-recreate/output.txt +++ b/acceptance/bundle/dms/failed-recreate/output.txt @@ -66,7 +66,7 @@ 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": "Fault injected by test. (400 INVALID_PARAMETER_VALUE)", diff --git a/acceptance/bundle/dms/failed-update/output.txt b/acceptance/bundle/dms/failed-update/output.txt index bb3a928e72..b7bf222599 100644 --- a/acceptance/bundle/dms/failed-update/output.txt +++ b/acceptance/bundle/dms/failed-update/output.txt @@ -51,10 +51,11 @@ 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)", + "resource_id": "main.dms_failed_update_schema", "status": "OPERATION_STATUS_FAILED", "sequence_id": "0" } 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..196ac8f038 --- /dev/null +++ b/acceptance/bundle/dms/operation-upload-fails-once/output.txt @@ -0,0 +1,84 @@ + +=== 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": { + "resource_id": "main.dms_upload_fails_once_schema", + "status": "OPERATION_STATUS_IN_PROGRESS", + "sequence_id": "0" + } +} +{ + "method": "PATCH", + "path": "/api/2.0/bundle/deployments/[NUMID]/versions/2/operations/schemas.foo", + "q": { + "update_mask": "state,error_message,resource_id,status" + }, + "body": { + "state": "{\"state\":{\"catalog_name\":\"other\",\"comment\":\"v1\",\"name\":\"dms_upload_fails_once_schema\"}}", + "resource_id": "other.dms_upload_fails_once_schema", + "status": "OPERATION_STATUS_SUCCEEDED", + "sequence_id": "0" + } +} +{ + "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/record-failure/output.txt b/acceptance/bundle/dms/record-failure/output.txt index efea62c8a2..7e963b0a39 100644 --- a/acceptance/bundle/dms/record-failure/output.txt +++ b/acceptance/bundle/dms/record-failure/output.txt @@ -47,7 +47,7 @@ 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)", 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_test.go b/libs/dms/client_test.go index f24db9f219..6a9c9302f8 100644 --- a/libs/dms/client_test.go +++ b/libs/dms/client_test.go @@ -108,55 +108,20 @@ 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", - }, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - update.Fields = tt.fields - assert.Equal(t, tt.want, newUpdateRequest(update, "3")) - }) - } + assert.Equal(t, updateOperationRequest{ + ResourceId: "job-1", + Status: bundledeployments.OperationStatusOperationStatusSucceeded, + SequenceId: "3", + }, newUpdateRequest(update, "3")) } diff --git a/libs/dms/fields.go b/libs/dms/fields.go index e3bd271c3b..7f4364ec24 100644 --- a/libs/dms/fields.go +++ b/libs/dms/fields.go @@ -17,10 +17,12 @@ 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 // wireNames pairs each field with its name on the wire, in the order a mask lists them. var wireNames = []struct { 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) -} From 64c8bd3e7c91cc4307e9d421435f5524680ab703 Mon Sep 17 00:00:00 2001 From: Shreyas Goenka Date: Mon, 24 Aug 2026 21:26:08 +0000 Subject: [PATCH 2/2] dms: send every field the update mask names UpdateOperation listed error_message in its update_mask but marked the body field omitempty, so a successful apply - which has no error - dropped it from the body while still claiming to write it. The service rejects that: error_message is required when 'error_message' is in update_mask (an empty value clears it) (400 INVALID_PARAMETER_VALUE) Verified against a live staging workspace, where every deploy failed on the first operation. The fake server accepted it because it treated a masked-but-absent field as "set to empty", so it now returns the same 400 and the local suite covers this. Co-authored-by: Isaac --- .../bundle/dms/declined-deploy/output.txt | 7 +++-- acceptance/bundle/dms/depends-on/output.txt | 14 +++++---- .../bundle/dms/emptied-resource/output.txt | 20 +++++++----- .../bundle/dms/existing-state/output.txt | 2 +- .../bundle/dms/failed-delete/output.txt | 11 ++++--- .../bundle/dms/failed-recreate/output.txt | 11 ++++--- .../bundle/dms/failed-update/output.txt | 4 +-- .../bundle/dms/multiple-resources/output.txt | 10 +++--- .../operation-upload-fails-once/output.txt | 13 +++++--- acceptance/bundle/dms/provenance/output.txt | 7 +++-- .../bundle/dms/record-failure/output.txt | 5 +-- acceptance/bundle/dms/record/output.txt | 20 +++++++----- .../bundle/dms/resource-lifecycle/output.txt | 26 ++++++++++------ libs/dms/client.go | 31 +++++++------------ libs/dms/client_test.go | 27 +++++++++++++--- libs/dms/fields.go | 17 +++++++--- libs/testserver/bundledeployments.go | 5 +++ 17 files changed, 141 insertions(+), 89 deletions(-) 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/output.txt b/acceptance/bundle/dms/failed-delete/output.txt index 9f0e09019a..9b61f6961c 100644 --- a/acceptance/bundle/dms/failed-delete/output.txt +++ b/acceptance/bundle/dms/failed-delete/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-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}}}", + "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-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" } } { @@ -108,8 +109,8 @@ API message: job is not deletable "body": { "error_message": "deleting id=[NUMID]: job is not deletable (400 INVALID_PARAMETER_VALUE)", "resource_id": "[NUMID]", - "status": "OPERATION_STATUS_FAILED", - "sequence_id": "0" + "sequence_id": "0", + "status": "OPERATION_STATUS_FAILED" } } { diff --git a/acceptance/bundle/dms/failed-recreate/output.txt b/acceptance/bundle/dms/failed-recreate/output.txt index ac634f0dca..4f05615b2f 100644 --- a/acceptance/bundle/dms/failed-recreate/output.txt +++ b/acceptance/bundle/dms/failed-recreate/output.txt @@ -57,9 +57,11 @@ 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" } } { @@ -70,8 +72,9 @@ Files: 3 uploaded, 0 deleted }, "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 b7bf222599..1448218f94 100644 --- a/acceptance/bundle/dms/failed-update/output.txt +++ b/acceptance/bundle/dms/failed-update/output.txt @@ -56,8 +56,8 @@ Files: 3 uploaded, 0 deleted "body": { "error_message": "updating id=main.dms_failed_update_schema: Fault injected by test. (400 INVALID_PARAMETER_VALUE)", "resource_id": "main.dms_failed_update_schema", - "status": "OPERATION_STATUS_FAILED", - "sequence_id": "0" + "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/output.txt b/acceptance/bundle/dms/operation-upload-fails-once/output.txt index 196ac8f038..66e8586d55 100644 --- a/acceptance/bundle/dms/operation-upload-fails-once/output.txt +++ b/acceptance/bundle/dms/operation-upload-fails-once/output.txt @@ -57,9 +57,11 @@ Files: 3 uploaded, 0 deleted "update_mask": "state,error_message,resource_id,status" }, "body": { + "error_message": "", "resource_id": "main.dms_upload_fails_once_schema", - "status": "OPERATION_STATUS_IN_PROGRESS", - "sequence_id": "0" + "sequence_id": "0", + "state": "", + "status": "OPERATION_STATUS_IN_PROGRESS" } } { @@ -69,10 +71,11 @@ Files: 3 uploaded, 0 deleted "update_mask": "state,error_message,resource_id,status" }, "body": { - "state": "{\"state\":{\"catalog_name\":\"other\",\"comment\":\"v1\",\"name\":\"dms_upload_fails_once_schema\"}}", + "error_message": "", "resource_id": "other.dms_upload_fails_once_schema", - "status": "OPERATION_STATUS_SUCCEEDED", - "sequence_id": "0" + "sequence_id": "0", + "state": "{\"state\":{\"catalog_name\":\"other\",\"comment\":\"v1\",\"name\":\"dms_upload_fails_once_schema\"}}", + "status": "OPERATION_STATUS_SUCCEEDED" } } { 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 7e963b0a39..7a1ec70eb4 100644 --- a/acceptance/bundle/dms/record-failure/output.txt +++ b/acceptance/bundle/dms/record-failure/output.txt @@ -51,8 +51,9 @@ Files: 5 uploaded, 0 deleted }, "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/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 6a9c9302f8..f1225062d0 100644 --- a/libs/dms/client_test.go +++ b/libs/dms/client_test.go @@ -119,9 +119,28 @@ func TestUpdateRequestSendsAFieldOnlyWhenTheMaskNamesIt(t *testing.T) { ErrorMessage: "boom", } - assert.Equal(t, updateOperationRequest{ - ResourceId: "job-1", - Status: bundledeployments.OperationStatusOperationStatusSucceeded, - SequenceId: "3", + assert.Equal(t, map[string]any{ + "resource_id": "job-1", + "status": bundledeployments.OperationStatusOperationStatusSucceeded, + "sequence_id": "3", }, 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 7f4364ec24..6188e2e3e0 100644 --- a/libs/dms/fields.go +++ b/libs/dms/fields.go @@ -24,15 +24,24 @@ const DescribesResource = FieldState | FieldErrorMessage | FieldResourceID | Fie // 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/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 }