diff --git a/go.mod b/go.mod index 8bc328bb6..ba9563ab3 100644 --- a/go.mod +++ b/go.mod @@ -7,7 +7,7 @@ toolchain go1.26.5 require ( buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.11-20260709200747-435963d16310.1 buf.build/go/protovalidate v1.2.0 - github.com/cenkalti/backoff/v4 v4.3.0 + github.com/cenkalti/backoff/v7 v7.0.0 github.com/fsnotify/fsnotify v1.10.1 github.com/gabriel-vasile/mimetype v1.4.13 github.com/go-resty/resty/v2 v2.17.2 @@ -134,6 +134,7 @@ require ( github.com/bytedance/gopkg v0.1.3 // indirect github.com/bytedance/sonic v1.15.0 // indirect github.com/bytedance/sonic/loader v0.5.0 // indirect + github.com/cenkalti/backoff/v4 v4.3.0 // indirect github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cloudwego/base64x v0.1.6 // indirect diff --git a/go.sum b/go.sum index 22915952a..8c1a66e6b 100644 --- a/go.sum +++ b/go.sum @@ -138,6 +138,8 @@ github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK3 github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= +github.com/cenkalti/backoff/v7 v7.0.0 h1:ZP+QAaaOnVUHo+ufFpZ835hbT3x2fy+h2lecVEosZ6A= +github.com/cenkalti/backoff/v7 v7.0.0/go.mod h1:qcKBGwsu4hpxHtQ8tWYsQ+ifzx2+sS+Xx/3jfe30lI8= github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU= github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= diff --git a/internal/backoff/backoff.go b/internal/backoff/backoff.go index f42b14d52..26811ae9b 100644 --- a/internal/backoff/backoff.go +++ b/internal/backoff/backoff.go @@ -7,8 +7,9 @@ package backoff import ( "context" + "time" - "github.com/cenkalti/backoff/v4" + "github.com/cenkalti/backoff/v7" "github.com/nginx/agent/v3/internal/config" ) @@ -46,14 +47,51 @@ import ( // 10 19.210 {stop} // // Information from https://pkg.go.dev/github.com/cenkalti/backoff/v4#section-readme +// RetryOptions builds a slice of backoff.RetryOption from the config.BackOff +// settings and the context. Call sites can pass the returned slice to backoff.Retry as +// variadic options. +// +// If the context has a deadline, the elapsed time limit is set to the remaining +// time until that deadline (if it's less restrictive than the config-based +// MaxElapsedTime). This mirrors the v4 API behavior where WithContext would +// respect the context timeout. If the context has no deadline, the config-based +// MaxElapsedTime is used. +func RetryOptions(ctx context.Context, backoffSettings *config.BackOff) []backoff.RetryOption { + eb := backoff.NewExponentialBackOff() + eb.InitialInterval = backoffSettings.InitialInterval + eb.MaxInterval = backoffSettings.MaxInterval + eb.RandomizationFactor = backoffSettings.RandomizationFactor + eb.Multiplier = backoffSettings.Multiplier + + maxElapsedTime := backoffSettings.MaxElapsedTime + if deadline, ok := ctx.Deadline(); ok { + timeUntilDeadline := time.Until(deadline) + // Use the smaller of the two timeouts, but only if deadline is positive + if timeUntilDeadline > 0 && (maxElapsedTime == 0 || timeUntilDeadline < maxElapsedTime) { + maxElapsedTime = timeUntilDeadline + } + } + + return []backoff.RetryOption{ + backoff.WithBackOff(eb), + backoff.WithMaxElapsedTime(maxElapsedTime), + } +} + +// WaitUntil retries a no-result operation until it succeeds, a permanent +// error is returned, or the retry options elapse. It adapts the operation to +// the generic backoff.Retry API. func WaitUntil( ctx context.Context, backoffSettings *config.BackOff, - operation backoff.Operation, + operation func() error, ) error { - backoffWithContext := Context(ctx, backoffSettings) + retryOpts := RetryOptions(ctx, backoffSettings) + _, err := backoff.Retry(ctx, func() (struct{}, error) { + return struct{}{}, operation() + }, retryOpts...) - return backoff.Retry(operation, backoffWithContext) + return err } // WaitUntilWithData Implementation of backoff operations that increases the back off period for each retry @@ -63,21 +101,7 @@ func WaitUntil( func WaitUntilWithData[T any]( ctx context.Context, backoffSettings *config.BackOff, - operation backoff.OperationWithData[T], + operation backoff.Operation[T], ) (T, error) { - backoffWithContext := Context(ctx, backoffSettings) - - return backoff.RetryWithData(operation, backoffWithContext) -} - -//nolint:ireturn // must return an interface -func Context(ctx context.Context, backoffSettings *config.BackOff) backoff.BackOffContext { - eb := backoff.NewExponentialBackOff() - eb.InitialInterval = backoffSettings.InitialInterval - eb.MaxInterval = backoffSettings.MaxInterval - eb.MaxElapsedTime = backoffSettings.MaxElapsedTime - eb.RandomizationFactor = backoffSettings.RandomizationFactor - eb.Multiplier = backoffSettings.Multiplier - - return backoff.WithContext(eb, ctx) + return backoff.Retry[T](ctx, operation, RetryOptions(ctx, backoffSettings)...) } diff --git a/internal/backoff/backoff_test.go b/internal/backoff/backoff_test.go index 60cbc75a9..e663df907 100644 --- a/internal/backoff/backoff_test.go +++ b/internal/backoff/backoff_test.go @@ -11,7 +11,6 @@ import ( "testing" "time" - "github.com/cenkalti/backoff/v4" "github.com/nginx/agent/v3/internal/config" "github.com/stretchr/testify/assert" ) @@ -21,7 +20,7 @@ func TestWaitUntil(t *testing.T) { invocations := 0 tests := []struct { context context.Context - operation backoff.Operation + operation func() error name string initialInterval time.Duration maxInterval time.Duration @@ -106,7 +105,7 @@ func TestWaitUntilWithData(t *testing.T) { invocations := -1 tests := []struct { context context.Context - operation backoff.OperationWithData[int] + operation func() (int, error) name string initialInterval time.Duration maxInterval time.Duration @@ -194,8 +193,6 @@ func TestContext(t *testing.T) { RandomizationFactor: config.DefBackoffRandomizationFactor, Multiplier: config.DefBackoffMultiplier, } - - backoffCtx := Context(context.Background(), settings) - + backoffCtx := RetryOptions(context.Background(), settings) assert.NotEmpty(t, backoffCtx) } diff --git a/internal/command/command_service.go b/internal/command/command_service.go index edf36f755..99cd3b63c 100644 --- a/internal/command/command_service.go +++ b/internal/command/command_service.go @@ -13,7 +13,7 @@ import ( "sync" "sync/atomic" - "github.com/cenkalti/backoff/v4" + "github.com/cenkalti/backoff/v7" mpi "github.com/nginx/agent/v3/api/grpc/mpi/v1" "github.com/nginx/agent/v3/internal/config" @@ -116,9 +116,10 @@ func (cs *CommandService) UpdateDataPlaneStatus( return response, nil } - response, err := backoff.RetryWithData( + response, err := backoff.Retry( + backOffCtx, sendDataPlaneStatus, - backoffHelpers.Context(backOffCtx, cfg.Client.Backoff), + backoffHelpers.RetryOptions(backOffCtx, cfg.Client.Backoff)..., ) if err != nil { return err @@ -152,9 +153,10 @@ func (cs *CommandService) UpdateDataPlaneHealth(ctx context.Context, instanceHea backOffCtx, backoffCancel := context.WithTimeout(ctx, cfg.Client.Backoff.MaxElapsedTime) defer backoffCancel() - response, err := backoff.RetryWithData( + response, err := backoff.Retry( + backOffCtx, cs.dataPlaneHealthCallback(ctx, request), - backoffHelpers.Context(backOffCtx, cfg.Client.Backoff), + backoffHelpers.RetryOptions(backOffCtx, cfg.Client.Backoff)..., ) if err != nil { return err @@ -167,7 +169,6 @@ func (cs *CommandService) UpdateDataPlaneHealth(ctx context.Context, instanceHea func (cs *CommandService) SendDataPlaneResponse(ctx context.Context, response *mpi.DataPlaneResponse) error { slog.DebugContext(ctx, "Sending data plane response", "response", response) - cfg := cs.config() backOffCtx, backoffCancel := context.WithTimeout(ctx, cfg.Client.Backoff.MaxElapsedTime) defer backoffCancel() @@ -177,9 +178,10 @@ func (cs *CommandService) SendDataPlaneResponse(ctx context.Context, response *m return err } - return backoff.Retry( + return backoffHelpers.WaitUntil( + backOffCtx, + cfg.Client.Backoff, cs.sendDataPlaneResponseCallback(ctx, response), - backoffHelpers.Context(backOffCtx, cfg.Client.Backoff), ) } @@ -209,7 +211,7 @@ func (cs *CommandService) Subscribe(ctx context.Context) { case <-ctx.Done(): return default: - retryError := backoff.Retry(cs.receiveCallback(ctx), backoffHelpers.Context(ctx, commonSettings)) + retryError := backoffHelpers.WaitUntil(ctx, commonSettings, cs.receiveCallback(ctx)) if retryError != nil { slog.WarnContext(ctx, "Failed to receive messages from subscribe stream", "error", retryError) } @@ -281,9 +283,10 @@ func (cs *CommandService) createConnectionCall(ctx context.Context) (*mpi.Create } slog.DebugContext(ctx, "Sending create connection request", "request", request) - resp, err := backoff.RetryWithData( + resp, err := backoff.Retry( + ctx, cs.connectCallback(ctx, request), - backoffHelpers.Context(ctx, commonSettings), + backoffHelpers.RetryOptions(ctx, commonSettings)..., ) if err != nil { cs.isConnected.Store(false) @@ -378,12 +381,12 @@ func (cs *CommandService) handleConfigApplyResponse( } slog.DebugContext(ctx, "Sending data plane response for queued config apply request", "response", newResponse) - backOffCtx, backoffCancel := context.WithTimeout(ctx, cfg.Client.Backoff.MaxElapsedTime) - err := backoff.Retry( + err := backoffHelpers.WaitUntil( + backOffCtx, + cfg.Client.Backoff, cs.sendDataPlaneResponseCallback(ctx, newResponse), - backoffHelpers.Context(backOffCtx, cfg.Client.Backoff), ) backoffCancel() diff --git a/internal/file/file_service_operator.go b/internal/file/file_service_operator.go index c2c1717a9..de04bc3ab 100644 --- a/internal/file/file_service_operator.go +++ b/internal/file/file_service_operator.go @@ -19,7 +19,7 @@ import ( "sync" "sync/atomic" - "github.com/cenkalti/backoff/v4" + "github.com/cenkalti/backoff/v7" mpi "github.com/nginx/agent/v3/api/grpc/mpi/v1" backoffHelpers "github.com/nginx/agent/v3/internal/backoff" "github.com/nginx/agent/v3/internal/config" @@ -105,9 +105,10 @@ func (fso *FileServiceOperator) File( return response, nil } - getFileResp, getFileErr := backoff.RetryWithData( + getFileResp, getFileErr := backoff.Retry( + backOffCtx, getFile, - backoffHelpers.Context(backOffCtx, fso.agentConfig.Client.Backoff), + backoffHelpers.RetryOptions(backOffCtx, fso.agentConfig.Client.Backoff)..., ) if getFileErr != nil { @@ -215,9 +216,10 @@ func (fso *FileServiceOperator) UpdateOverview( return response, nil } - response, err := backoff.RetryWithData( + response, err := backoff.Retry( + newCtx, sendUpdateOverview, - backoffHelpers.Context(newCtx, backoffSettings), + backoffHelpers.RetryOptions(newCtx, backoffSettings)..., ) if err != nil { return err @@ -399,7 +401,6 @@ func (fso *FileServiceOperator) sendUpdateFileRequest( }, MessageMeta: messageMeta, } - backOffCtx, backoffCancel := context.WithTimeout(ctx, fso.agentConfig.Client.Backoff.MaxElapsedTime) defer backoffCancel() @@ -431,9 +432,10 @@ func (fso *FileServiceOperator) sendUpdateFileRequest( return response, nil } - response, err := backoff.RetryWithData( + response, err := backoff.Retry( + backOffCtx, sendUpdateFile, - backoffHelpers.Context(backOffCtx, fso.agentConfig.Client.Backoff), + backoffHelpers.RetryOptions(backOffCtx, fso.agentConfig.Client.Backoff)..., ) if err != nil { return err @@ -522,7 +524,7 @@ func (fso *FileServiceOperator) sendUpdateFileStreamHeader( return nil } - return backoff.Retry(sendUpdateFileHeader, backoffHelpers.Context(backOffCtx, fso.agentConfig.Client.Backoff)) + return backoffHelpers.WaitUntil(backOffCtx, fso.agentConfig.Client.Backoff, sendUpdateFileHeader) } func (fso *FileServiceOperator) sendFileUpdateStreamChunks( @@ -616,7 +618,7 @@ func (fso *FileServiceOperator) sendFileUpdateStreamChunk( return nil } - return backoff.Retry(sendUpdateFileChunk, backoffHelpers.Context(backOffCtx, fso.agentConfig.Client.Backoff)) + return backoffHelpers.WaitUntil(backOffCtx, fso.agentConfig.Client.Backoff, sendUpdateFileChunk) } func (fso *FileServiceOperator) setupIdentifiers(ctx context.Context, iteration int) (context.Context, string) { diff --git a/internal/file/file_service_operator_test.go b/internal/file/file_service_operator_test.go index 3d635f778..f1dd4eeb7 100644 --- a/internal/file/file_service_operator_test.go +++ b/internal/file/file_service_operator_test.go @@ -14,6 +14,7 @@ import ( "testing" "time" + "github.com/cenkalti/backoff/v7" mpi "github.com/nginx/agent/v3/api/grpc/mpi/v1" "github.com/nginx/agent/v3/api/grpc/mpi/v1/v1fakes" "github.com/nginx/agent/v3/pkg/files" @@ -127,7 +128,7 @@ func TestFileServiceOperator_UpdateOverview_NoConnection(t *testing.T) { }, }, filePath, 0) - assert.ErrorIs(t, err, context.DeadlineExceeded) + assert.ErrorIs(t, err, backoff.ErrMaxElapsedTime) } func TestFileManagerService_UpdateFile(t *testing.T) { diff --git a/internal/grpc/grpc.go b/internal/grpc/grpc.go index 50990651c..eff2a5cde 100644 --- a/internal/grpc/grpc.go +++ b/internal/grpc/grpc.go @@ -20,7 +20,7 @@ import ( "github.com/nginx/agent/v3/internal/datasource/file" - "github.com/cenkalti/backoff/v4" + "github.com/cenkalti/backoff/v7" grpcRetry "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/retry" "buf.build/go/protovalidate" diff --git a/internal/grpc/grpc_test.go b/internal/grpc/grpc_test.go index 1308e885b..33278f95b 100644 --- a/internal/grpc/grpc_test.go +++ b/internal/grpc/grpc_test.go @@ -13,7 +13,7 @@ import ( "fmt" "testing" - "github.com/cenkalti/backoff/v4" + "github.com/cenkalti/backoff/v7" "github.com/nginx/agent/v3/test/stub" "google.golang.org/grpc" "google.golang.org/grpc/codes" @@ -391,7 +391,7 @@ func Test_ValidateGrpcError(t *testing.T) { require.ErrorIs(t, TestError{}, result) } result = ValidateGrpcError(status.Errorf(codes.InvalidArgument, "error")) - require.ErrorIs(t, &backoff.PermanentError{}, result) + require.ErrorIs(t, result, backoff.ErrPermanent) } func Test_transportCredentials(t *testing.T) { diff --git a/internal/nginx/nginx_instance_operator.go b/internal/nginx/nginx_instance_operator.go index e85f743c1..bcb1aac42 100644 --- a/internal/nginx/nginx_instance_operator.go +++ b/internal/nginx/nginx_instance_operator.go @@ -14,7 +14,8 @@ import ( "path/filepath" "time" - "github.com/nginx/agent/v3/internal/backoff" + "github.com/cenkalti/backoff/v7" + backoffHelpers "github.com/nginx/agent/v3/internal/backoff" "github.com/nginx/agent/v3/pkg/nginxprocess" "github.com/nginx/agent/v3/pkg/host/exec" @@ -146,7 +147,7 @@ func (i *NginxInstanceOperator) checkWorkers(ctx context.Context, instanceID str slog.DebugContext(ctx, "Found parent process ID, checking NGINX worker processes have reloaded", "process_id", newPid) - err := backoff.WaitUntil(ctx, backoffSettings, func() error { + err := backoffHelpers.WaitUntil(ctx, backoffSettings, func() error { currentWorkers := i.nginxProcessOperator.NginxWorkerProcesses(ctx, newPid) if len(currentWorkers) == 0 { return errors.New("waiting for NGINX worker processes") @@ -160,6 +161,10 @@ func (i *NginxInstanceOperator) checkWorkers(ctx context.Context, instanceID str return fmt.Errorf("waiting for NGINX worker to be newer than %v", createdTime) }) + + if re := backoff.AsRetryError(err); re != nil { + err = re.LastErr + } if err != nil { slog.WarnContext( ctx, diff --git a/test/mock/grpc/mock_management_file_service.go b/test/mock/grpc/mock_management_file_service.go index 14f686f21..f05a298ed 100644 --- a/test/mock/grpc/mock_management_file_service.go +++ b/test/mock/grpc/mock_management_file_service.go @@ -18,7 +18,6 @@ import ( "github.com/nginx/agent/v3/pkg/files" - "github.com/cenkalti/backoff/v4" backoffHelpers "github.com/nginx/agent/v3/internal/backoff" "github.com/nginx/agent/v3/internal/config" internalgrpc "github.com/nginx/agent/v3/internal/grpc" @@ -347,7 +346,7 @@ func (mgs *FileService) sendGetFileStreamHeader(ctx context.Context, return nil } - return backoff.Retry(sendGetFileHeader, backoffHelpers.Context(backOffCtx, mgs.agentConfig.Client.Backoff)) + return backoffHelpers.WaitUntil(backOffCtx, mgs.agentConfig.Client.Backoff, sendGetFileHeader) } func (mgs *FileService) sendGetFileStreamChunk(ctx context.Context, chunk v1.FileDataChunk_Content, @@ -380,7 +379,7 @@ func (mgs *FileService) sendGetFileStreamChunk(ctx context.Context, chunk v1.Fil return nil } - return backoff.Retry(sendGetFileChunk, backoffHelpers.Context(backOffCtx, mgs.agentConfig.Client.Backoff)) + return backoffHelpers.WaitUntil(backOffCtx, mgs.agentConfig.Client.Backoff, sendGetFileChunk) } func readChunk(