From 770a66f45a50b10e4a0441904337e87ea11c7f98 Mon Sep 17 00:00:00 2001 From: Denis Bilenko Date: Tue, 20 Oct 2026 15:21:33 +0200 Subject: [PATCH 1/2] Retry more transient errors on reads, lock release and deployment-service calls Retry the idempotent calls that failed on transient backend errors in an integration run: deploy lock stat/read/delete, state file reads, remote reads and waits after create/update, and the deployment-service get, list, update and delete calls. Reads also retry "unexpected EOF" transport errors. Co-authored-by: Isaac --- bundle/deploy/filer.go | 4 +- bundle/direct/apply.go | 8 ++-- bundle/direct/bundle_plan.go | 4 +- bundle/direct/dstate/state.go | 7 ++- bundle/phases/destroy.go | 5 +-- bundle/phases/dms.go | 28 ++++++++++-- cmd/bundle/utils/process.go | 12 +++-- libs/locker/locker.go | 3 +- libs/retry/filer.go | 40 +++++++++++++++-- libs/retry/filer_test.go | 85 +++++++++++++++++++++++++++++++++++ libs/retry/retry.go | 22 ++++++--- libs/retry/retry_test.go | 4 ++ 12 files changed, 193 insertions(+), 29 deletions(-) diff --git a/bundle/deploy/filer.go b/bundle/deploy/filer.go index 1db8819d3ff..bd1171ed3e7 100644 --- a/bundle/deploy/filer.go +++ b/bundle/deploy/filer.go @@ -43,7 +43,9 @@ func (s stateFiler) Read(ctx context.Context, path string) (io.ReadCloser, error // No get-status first: a missing file is the common first-deploy case and export's 404 // reports it in one round trip. A directory at a state path is not a state file; its // export fails to parse as one. - r, err := s.workspaceClient.Workspace.Download(ctx, absPath) + r, err := retry.With(ctx, retry.IsTransientRead, func() (io.ReadCloser, error) { + return s.workspaceClient.Workspace.Download(ctx, absPath) + }) if errors.Is(err, apierr.ErrNotFound) { return nil, fmt.Errorf("file does not exist: %s: %w", absPath, fs.ErrNotExist) } diff --git a/bundle/direct/apply.go b/bundle/direct/apply.go index 57367883c6c..d4f4a2ffb41 100644 --- a/bundle/direct/apply.go +++ b/bundle/direct/apply.go @@ -91,7 +91,7 @@ func (d *DeploymentUnit) Create(ctx context.Context, db *dstate.DeploymentState, } waitRemoteState, err := waitCapped(ctx, d.MaxWait, "creation of "+d.ResourceKey, func(ctx context.Context) (any, error) { - return retry.With(ctx, retry.IsTransient, func() (any, error) { + return retry.With(ctx, retry.IsTransientRead, func() (any, error) { return d.Adapter.WaitAfterCreate(ctx, newID, newState) }) }) @@ -189,7 +189,7 @@ func (d *DeploymentUnit) Update(ctx context.Context, db *dstate.DeploymentState, } } - waitRemoteState, err := retry.With(ctx, retry.IsTransient, func() (any, error) { + waitRemoteState, err := retry.With(ctx, retry.IsTransientRead, func() (any, error) { return d.Adapter.WaitAfterUpdate(ctx, id, newState) }) if err != nil { @@ -233,7 +233,7 @@ func (d *DeploymentUnit) UpdateWithID(ctx context.Context, db *dstate.Deployment return fmt.Errorf("saving state id=%s: %w", oldID, err) } - waitRemoteState, err := retry.With(ctx, retry.IsTransient, func() (any, error) { + waitRemoteState, err := retry.With(ctx, retry.IsTransientRead, func() (any, error) { return d.Adapter.WaitAfterUpdate(ctx, newID, newState) }) if err != nil { @@ -375,7 +375,7 @@ func (d *DeploymentUnit) refreshRemoteState(ctx context.Context, id string) erro if d.RemoteState != nil { return nil } - remoteState, err := retry.With(ctx, retry.IsTransient, func() (any, error) { + remoteState, err := retry.With(ctx, retry.IsTransientRead, func() (any, error) { return d.Adapter.DoRead(ctx, id) }) if err != nil { diff --git a/bundle/direct/bundle_plan.go b/bundle/direct/bundle_plan.go index 98da4f1376d..abf10376039 100644 --- a/bundle/direct/bundle_plan.go +++ b/bundle/direct/bundle_plan.go @@ -237,7 +237,7 @@ func (b *DeploymentBundle) CalculatePlan(ctx context.Context, client *databricks return true } - remoteState, err := retry.With(ctx, retry.IsTransient, func() (any, error) { + remoteState, err := retry.With(ctx, retry.IsTransientRead, func() (any, error) { return adapter.DoRead(ctx, id) }) if err != nil { @@ -315,7 +315,7 @@ func (b *DeploymentBundle) CalculatePlan(ctx context.Context, client *databricks return false } - remoteState, err := retry.With(ctx, retry.IsTransient, func() (any, error) { + remoteState, err := retry.With(ctx, retry.IsTransientRead, func() (any, error) { return adapter.DoRead(ctx, dbentry.ID) }) if err != nil { diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go index ac8f8226c4b..0bdd2bac037 100644 --- a/bundle/direct/dstate/state.go +++ b/bundle/direct/dstate/state.go @@ -23,6 +23,7 @@ import ( "github.com/databricks/cli/libs/cmdctx" "github.com/databricks/cli/libs/dms" "github.com/databricks/cli/libs/log" + "github.com/databricks/cli/libs/retry" "github.com/databricks/databricks-sdk-go/service/bundledeployments" "github.com/google/uuid" ) @@ -584,8 +585,10 @@ Run "databricks bundle destroy" first, then deploy again with deployment history if dmsDeployment.DeploymentID != "" { w := cmdctx.WorkspaceClient(ctx) - recorded, err := w.BundleDeployments.ListResourcesAll(ctx, bundledeployments.ListResourcesRequest{ - Parent: dms.DeploymentName(dmsDeployment.DeploymentID), + recorded, err := retry.With(ctx, retry.IsTransientRead, func() ([]bundledeployments.Resource, error) { + return w.BundleDeployments.ListResourcesAll(ctx, bundledeployments.ListResourcesRequest{ + Parent: dms.DeploymentName(dmsDeployment.DeploymentID), + }) }) if err != nil { return fmt.Errorf("listing resources from the deployment history service: %w", err) diff --git a/bundle/phases/destroy.go b/bundle/phases/destroy.go index 8b5497d626a..07bbbbca385 100644 --- a/bundle/phases/destroy.go +++ b/bundle/phases/destroy.go @@ -20,7 +20,6 @@ import ( "github.com/databricks/cli/libs/log" "github.com/databricks/cli/libs/logdiag" "github.com/databricks/databricks-sdk-go/apierr" - "github.com/databricks/databricks-sdk-go/service/bundledeployments" ) func assertRootPathExists(ctx context.Context, b *bundle.Bundle) (bool, error) { @@ -168,7 +167,7 @@ func destroyCore(ctx context.Context, b *bundle.Bundle, plan *deployplan.Plan) { if completed { deploymentID := b.DeploymentBundle.StateDB.DeploymentID w := b.WorkspaceClient(ctx) - if err := w.BundleDeployments.DeleteDeployment(ctx, bundledeployments.DeleteDeploymentRequest{Name: dms.DeploymentName(deploymentID)}); err != nil { + if err := deleteDeployment(ctx, w, deploymentID); err != nil { logdiag.LogError(ctx, fmt.Errorf("failed to delete deployment: %w", err)) return } @@ -301,7 +300,7 @@ func Destroy(ctx context.Context, b *bundle.Bundle) { } else if completed { deploymentID := b.DeploymentBundle.StateDB.DeploymentID w := b.WorkspaceClient(ctx) - if err := w.BundleDeployments.DeleteDeployment(ctx, bundledeployments.DeleteDeploymentRequest{Name: dms.DeploymentName(deploymentID)}); err != nil { + if err := deleteDeployment(ctx, w, deploymentID); err != nil { logdiag.LogError(ctx, fmt.Errorf("failed to delete deployment: %w", err)) } } diff --git a/bundle/phases/dms.go b/bundle/phases/dms.go index abd89f03f2f..a2399c9e3a1 100644 --- a/bundle/phases/dms.go +++ b/bundle/phases/dms.go @@ -17,7 +17,10 @@ import ( "github.com/databricks/cli/libs/dms" "github.com/databricks/cli/libs/log" "github.com/databricks/cli/libs/logdiag" + "github.com/databricks/cli/libs/retry" "github.com/databricks/cli/libs/workspaceurls" + "github.com/databricks/databricks-sdk-go" + "github.com/databricks/databricks-sdk-go/apierr" "github.com/databricks/databricks-sdk-go/common/types/fieldmask" "github.com/databricks/databricks-sdk-go/service/bundledeployments" ) @@ -95,10 +98,12 @@ func createOrUpdateDeployment(ctx context.Context, b *bundle.Bundle, current *bu } db.StateDB.DeploymentID = deploymentID } else if mask := metadata.StaleFields(current); mask != "" { - _, err := w.BundleDeployments.UpdateDeployment(ctx, bundledeployments.UpdateDeploymentRequest{ - Name: dms.DeploymentName(deploymentID), - Deployment: dms.DeploymentUpdate(metadata, mask), - UpdateMask: fieldmask.FieldMask{Paths: strings.Split(mask, ",")}, + _, err := retry.With(ctx, retry.IsTransient, func() (*bundledeployments.Deployment, error) { + return w.BundleDeployments.UpdateDeployment(ctx, bundledeployments.UpdateDeploymentRequest{ + Name: dms.DeploymentName(deploymentID), + Deployment: dms.DeploymentUpdate(metadata, mask), + UpdateMask: fieldmask.FieldMask{Paths: strings.Split(mask, ",")}, + }) }) if err != nil { logdiag.LogError(ctx, fmt.Errorf("failed to update deployment: %w", err)) @@ -112,6 +117,21 @@ func createOrUpdateDeployment(ctx context.Context, b *bundle.Bundle, current *bu }) } +// deleteDeployment deletes the deployment record. A retry that finds it missing succeeds: the +// earlier attempt may have deleted it before its response was lost. +func deleteDeployment(ctx context.Context, w *databricks.WorkspaceClient, deploymentID string) error { + retried := false + _, err := retry.With(ctx, retry.IsTransient, func() (struct{}, error) { + err := w.BundleDeployments.DeleteDeployment(ctx, bundledeployments.DeleteDeploymentRequest{Name: dms.DeploymentName(deploymentID)}) + if retried && apierr.IsMissing(err) { + return struct{}{}, nil + } + retried = true + return struct{}{}, err + }) + return err +} + // startVersion claims the version the run settled on and opens the buffer that records // each state write under it. Called after approval, so a declined deploy never claims a number. // A no-op when the bundle does not record deployment history. diff --git a/cmd/bundle/utils/process.go b/cmd/bundle/utils/process.go index 6ccb3367db7..21c3008f719 100644 --- a/cmd/bundle/utils/process.go +++ b/cmd/bundle/utils/process.go @@ -31,12 +31,14 @@ import ( "github.com/databricks/cli/libs/dyn" "github.com/databricks/cli/libs/log" "github.com/databricks/cli/libs/logdiag" + "github.com/databricks/cli/libs/retry" "github.com/databricks/cli/libs/sync" "github.com/databricks/cli/libs/telemetry" "github.com/databricks/cli/libs/telemetry/protos" "github.com/databricks/databricks-sdk-go" "github.com/databricks/databricks-sdk-go/apierr" "github.com/databricks/databricks-sdk-go/service/bundledeployments" + "github.com/databricks/databricks-sdk-go/service/workspace" "github.com/databricks/databricks-sdk-go/useragent" "github.com/spf13/cobra" "golang.org/x/mod/semver" @@ -597,7 +599,9 @@ func ValidateEngineSetting(ctx context.Context, b *bundle.Bundle) error { func fetchDeploymentFromStatePath(ctx context.Context, w *databricks.WorkspaceClient, statePath string) (string, *bundledeployments.Deployment, error) { nodePath := path.Join(statePath, dms.DeploymentNodeName) - obj, err := w.Workspace.GetStatusByPath(ctx, nodePath) + obj, err := retry.With(ctx, retry.IsTransientRead, func() (*workspace.ObjectInfo, error) { + return w.Workspace.GetStatusByPath(ctx, nodePath) + }) if errors.Is(err, apierr.ErrNotFound) || errors.Is(err, apierr.ErrResourceDoesNotExist) { return "", nil, nil } @@ -605,8 +609,10 @@ func fetchDeploymentFromStatePath(ctx context.Context, w *databricks.WorkspaceCl return "", nil, fmt.Errorf("looking up deployment at %s: %w", nodePath, err) } deploymentID := strconv.FormatInt(obj.ObjectId, 10) - deployment, err := w.BundleDeployments.GetDeployment(ctx, bundledeployments.GetDeploymentRequest{ - Name: dms.DeploymentName(deploymentID), + deployment, err := retry.With(ctx, retry.IsTransientRead, func() (*bundledeployments.Deployment, error) { + return w.BundleDeployments.GetDeployment(ctx, bundledeployments.GetDeploymentRequest{ + Name: dms.DeploymentName(deploymentID), + }) }) if err != nil { return "", nil, err diff --git a/libs/locker/locker.go b/libs/locker/locker.go index 0de7fdb4d96..a63bc71b626 100644 --- a/libs/locker/locker.go +++ b/libs/locker/locker.go @@ -12,6 +12,7 @@ import ( "time" "github.com/databricks/cli/libs/filer" + "github.com/databricks/cli/libs/retry" "github.com/databricks/databricks-sdk-go" "github.com/google/uuid" ) @@ -209,7 +210,7 @@ func CreateLocker(user, targetDir string, w *databricks.WorkspaceClient) (*Locke } locker := &Locker{ - filer: filer, + filer: retry.Filer(filer), TargetDir: targetDir, Active: false, diff --git a/libs/retry/filer.go b/libs/retry/filer.go index 091b4c73952..11b991b18fc 100644 --- a/libs/retry/filer.go +++ b/libs/retry/filer.go @@ -3,21 +3,24 @@ package retry import ( "bytes" "context" + "errors" "io" + "io/fs" "slices" "github.com/databricks/cli/libs/filer" ) -// retryingFiler retries overwriting writes that fail with a transient server error. +// retryingFiler retries calls that fail with a transient server error and are safe to repeat. type retryingFiler struct { filer.Filer } -// Filer wraps f so that Write retries on transient server errors. +// Filer wraps f so that Write, Read, ReadDir, Stat and Delete retry on transient errors. // Only writes with [filer.OverwriteIfExists] are retried: they are idempotent, whereas repeating // a write that does not overwrite could fail on the file its own earlier attempt created. -// All other methods are passed through. +// A retried Delete treats a missing file as success, since the earlier attempt may have removed it. +// Mkdir is passed through. func Filer(f filer.Filer) filer.Filer { return &retryingFiler{Filer: f} } @@ -55,3 +58,34 @@ func (r *retryingFiler) Write(ctx context.Context, path string, reader io.Reader }) return err } + +func (r *retryingFiler) Read(ctx context.Context, path string) (io.ReadCloser, error) { + return With(ctx, IsTransientRead, func() (io.ReadCloser, error) { + return r.Filer.Read(ctx, path) + }) +} + +func (r *retryingFiler) ReadDir(ctx context.Context, path string) ([]fs.DirEntry, error) { + return With(ctx, IsTransientRead, func() ([]fs.DirEntry, error) { + return r.Filer.ReadDir(ctx, path) + }) +} + +func (r *retryingFiler) Stat(ctx context.Context, path string) (fs.FileInfo, error) { + return With(ctx, IsTransientRead, func() (fs.FileInfo, error) { + return r.Filer.Stat(ctx, path) + }) +} + +func (r *retryingFiler) Delete(ctx context.Context, path string, mode ...filer.DeleteMode) error { + retried := false + _, err := With(ctx, IsTransient, func() (struct{}, error) { + err := r.Filer.Delete(ctx, path, mode...) + if retried && errors.Is(err, fs.ErrNotExist) { + return struct{}{}, nil + } + retried = true + return struct{}{}, err + }) + return err +} diff --git a/libs/retry/filer_test.go b/libs/retry/filer_test.go index a264ce47066..90e57faedb9 100644 --- a/libs/retry/filer_test.go +++ b/libs/retry/filer_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "io" + "io/fs" "net/http" "strings" "testing" @@ -81,3 +82,87 @@ func TestFilerDoesNotRetryWithoutOverwrite(t *testing.T) { require.Error(t, err) assert.Len(t, f.bodies, 1) } + +// readFiler fails the first `failures` calls of every read-side method with err. +type readFiler struct { + filer.Filer + failures int + err error + calls int +} + +func (f *readFiler) next() error { + f.calls++ + if f.calls <= f.failures { + return f.err + } + return nil +} + +func (f *readFiler) Read(ctx context.Context, path string) (io.ReadCloser, error) { + return io.NopCloser(strings.NewReader("x")), f.next() +} + +func (f *readFiler) ReadDir(ctx context.Context, path string) ([]fs.DirEntry, error) { + return nil, f.next() +} + +func (f *readFiler) Stat(ctx context.Context, path string) (fs.FileInfo, error) { + return nil, f.next() +} + +func (f *readFiler) Delete(ctx context.Context, path string, mode ...filer.DeleteMode) error { + return f.next() +} + +func TestFilerRetriesReads(t *testing.T) { + ctx := testContext(t) + calls := map[string]func(filer.Filer) error{ + "Read": func(f filer.Filer) error { _, err := f.Read(ctx, "a"); return err }, + "ReadDir": func(f filer.Filer) error { _, err := f.ReadDir(ctx, "a"); return err }, + "Stat": func(f filer.Filer) error { _, err := f.Stat(ctx, "a"); return err }, + } + for name, call := range calls { + t.Run(name, func(t *testing.T) { + f := &readFiler{failures: 1, err: errTransient} + require.NoError(t, call(Filer(f))) + assert.Equal(t, 2, f.calls) + }) + } +} + +func TestFilerDoesNotRetryNotExist(t *testing.T) { + f := &readFiler{failures: 1, err: fs.ErrNotExist} + _, err := Filer(f).Stat(testContext(t), "a") + require.ErrorIs(t, err, fs.ErrNotExist) + assert.Equal(t, 1, f.calls) +} + +func TestFilerDeleteRetries(t *testing.T) { + f := &readFiler{failures: 1, err: errTransient} + require.NoError(t, Filer(f).Delete(testContext(t), "a")) + assert.Equal(t, 2, f.calls) +} + +func TestFilerDeleteMissingAfterRetryIsSuccess(t *testing.T) { + // The first attempt hits a transient error, the retry finds the file gone. + f := &sequenceFiler{errs: []error{errTransient, fs.ErrNotExist}} + require.NoError(t, Filer(f).Delete(testContext(t), "a")) +} + +func TestFilerDeleteMissingOnFirstAttemptIsError(t *testing.T) { + f := &sequenceFiler{errs: []error{fs.ErrNotExist}} + require.ErrorIs(t, Filer(f).Delete(testContext(t), "a"), fs.ErrNotExist) +} + +// sequenceFiler returns errs from successive Delete calls. +type sequenceFiler struct { + filer.Filer + errs []error +} + +func (f *sequenceFiler) Delete(ctx context.Context, path string, mode ...filer.DeleteMode) error { + err := f.errs[0] + f.errs = f.errs[1:] + return err +} diff --git a/libs/retry/retry.go b/libs/retry/retry.go index 97ac0f0d6d6..86dd9ec232c 100644 --- a/libs/retry/retry.go +++ b/libs/retry/retry.go @@ -7,6 +7,7 @@ import ( "fmt" "net/http" "net/url" + "slices" "strconv" "strings" "time" @@ -52,19 +53,28 @@ func IsTransient(ctx context.Context, err error) bool { return apiErr.StatusCode >= 500 || apiErr.StatusCode == http.StatusRequestTimeout } +// transientReadMessages are the transport errors, matched by message, that are worth retrying on a read: +// - REFUSED_STREAM, the HTTP/2 refusal in issue #6472, where the server refused the stream before +// processing the request. +// - unexpected EOF, where the connection was cut mid-response. Observed on GET /clusters/get. +var transientReadMessages = []string{"REFUSED_STREAM", "unexpected EOF"} + // IsTransientRead returns true for errors worth retrying on an idempotent read: the transient -// server errors IsTransient covers, plus the HTTP/2 REFUSED_STREAM in issue #6472, where the -// server refused the stream before processing the request so the read is safe to replay. The -// SDK already retries the transport errors it recognizes (connection reset/refused, i/o timeout, +// server errors IsTransient covers, plus the transport errors in transientReadMessages. The SDK +// already retries the transport errors it recognizes (connection reset/refused, i/o timeout, // TLS handshake) for its full RetryTimeout, so retrying every *url.Error would restart that -// multi-minute loop; REFUSED_STREAM is the gap it leaves. net/http bundles unexported http2 +// multi-minute loop; these are the gaps it leaves. net/http bundles unexported http2 // types, so we match the message, as the SDK does for its own transient url errors. func IsTransientRead(ctx context.Context, err error) bool { if IsTransient(ctx, err) { return true } - _, ok := errors.AsType[*url.Error](err) - return ok && strings.Contains(err.Error(), "REFUSED_STREAM") + if _, ok := errors.AsType[*url.Error](err); !ok { + return false + } + return slices.ContainsFunc(transientReadMessages, func(m string) bool { + return strings.Contains(err.Error(), m) + }) } // With retries fn while check returns true for the error, up to MaxRetries times. diff --git a/libs/retry/retry_test.go b/libs/retry/retry_test.go index 0a9006cbedc..0568fc82b09 100644 --- a/libs/retry/retry_test.go +++ b/libs/retry/retry_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "io" "net/http" "net/url" "testing" @@ -35,6 +36,9 @@ func TestIsTransientRead(t *testing.T) { // REFUSED_STREAM is the transport failure the SDK leaves unretried, so we replay the read (issue #6472). refusedStream := &url.Error{Op: "Get", URL: "https://example.test/api/2.0/workspace/get-status", Err: errors.New("stream error: stream ID 1; REFUSED_STREAM; received from peer")} assert.True(t, IsTransientRead(ctx, refusedStream)) + // The connection cut mid-response, seen on GET /clusters/get. + unexpectedEOF := &url.Error{Op: "Get", URL: "https://example.test/api/2.1/clusters/get", Err: io.ErrUnexpectedEOF} + assert.True(t, IsTransientRead(ctx, unexpectedEOF)) // Transport errors the SDK already retries for its full RetryTimeout must not be retried again. connRefused := &url.Error{Op: "Get", URL: "https://example.test/api/2.0/workspace/get-status", Err: errors.New("dial tcp 10.0.0.1:443: connect: connection refused")} assert.False(t, IsTransientRead(ctx, connRefused)) From f98fd07332d59bac06d9e7f6b2a913fb50cd1c2d Mon Sep 17 00:00:00 2001 From: Denis Bilenko Date: Tue, 20 Oct 2026 15:37:21 +0200 Subject: [PATCH 2/2] Add changelog fragment Co-authored-by: Isaac --- .nextchanges/bundles/7129.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 .nextchanges/bundles/7129.md diff --git a/.nextchanges/bundles/7129.md b/.nextchanges/bundles/7129.md new file mode 100644 index 00000000000..866e916c8ab --- /dev/null +++ b/.nextchanges/bundles/7129.md @@ -0,0 +1 @@ +* Retry more transient server errors during `bundle deploy` and `bundle destroy`, including when releasing the deploy lock and reading state. ([#7129](https://github.com/databricks/cli/pull/7129))