feat(native-retries): add custom should-retry logic for mount initialization (#4805)

* feat(native-retries): add custom should-retry logic for mount initialization

### Description
- Implemented `ShouldRetryOnMount`, `ShouldRetryOnMountWithRetryContext`, and `ShouldRetryOnMountWithMonitoringAndRetryContext` in `internal/storage/storageutil/custom_retry.go`.
- Scoped mount initialization retries to standard transient errors (`ShouldRetryWithoutLogging`), missing buckets (`404` / `NotFound` with "bucket does not exist", case-insensitive), and `403` / `PermissionDenied`.
- Configured retry logging in `ShouldRetryOnMountWithRetryContext` to use `logger.Errorf` to ensure visibility across standard error logs and GKE error files.

### Link to the issue in case of a bug fix.
NA

### Testing details
1. Manual - NA
2. Unit tests - Added 5 table-driven test suites (19 cases) in `custom_retry_test.go` verifying nil guards, standard transient errors, thread-safe `logBuffer` concurrency, ERROR-level log output, case-insensitive missing bucket matching, and error categorization (`STALLED_READ_REQUEST` vs `OTHER_ERRORS`). Ran full package tests (`go test ./internal/storage/storageutil/...`).
3. Integration tests - NA

### Any backward incompatible change? If so, please explain.
No

* test(storageutil): add unit tests for wrapped gRPC and permanent errors on mount

* refactor(native-retries): centralize mount retry error classification in determineRetryAction

Author: dangarhs

internal/storage/storageutil/custom_retry.go

@@ -17,6 +17,7 @@ package storageutil
 import (
 	"context"
 	"errors"
+	"strings"
 
 	"cloud.google.com/go/storage"
 	"github.com/googlecloudplatform/gcsfuse/v3/internal/logger"
@@ -38,41 +39,69 @@ const (
 	retry401
 	// retryUnauthenticated indicates a gRPC Unauthenticated error which requires a retry due to credentials refresh.
 	retryUnauthenticated
+	// retry404BucketDoesNotExist indicates an HTTP 404 error where the bucket was not found during mount.
+	retry404BucketDoesNotExist
+	// retryNotFoundBucketDoesNotExist indicates a gRPC NotFound error where the bucket was not found during mount.
+	retryNotFoundBucketDoesNotExist
+	// retry403 indicates an HTTP 403 Permission Denied error during mount.
+	retry403
+	// retryPermissionDenied indicates a gRPC PermissionDenied error during mount.
+	retryPermissionDenied
 )
 
+const errStrBucketNotExist = "bucket does not exist"
+
 func determineRetryAction(err error) retryAction {
 	if storage.ShouldRetry(err) {
 		return retryTransient
 	}
 
-	// HTTP 401 errors - Invalid Credentials
-	// This is a work-around to fix the corner case where GCSFuse checks the token
-	// as valid but GCS says invalid. This might be due to client-server timer
-	// issues. Actual fix will be refresh the token earlier than 1 hr.
-	// Changes will be done post resolution of the below issue:
-	// https://github.com/golang/oauth2/issues/623
-	// TODO(b/518674297): Please incorporate the correct fix post resolution of the above issue.
-	if typed, ok := err.(*googleapi.Error); ok {
-		if typed.Code == 401 {
+	var apiErr *googleapi.Error
+	if errors.As(err, &apiErr) {
+		// HTTP 401 errors - Invalid Credentials
+		// This is a work-around to fix the corner case where GCSFuse checks the token
+		// as valid but GCS says invalid. This might be due to client-server timer
+		// issues. Actual fix will be refresh the token earlier than 1 hr.
+		// Changes will be done post resolution of the below issue:
+		// https://github.com/golang/oauth2/issues/623
+		// TODO(b/518674297): Please incorporate the correct fix post resolution of the above issue.
+		if apiErr.Code == 401 {
 			return retry401
 		}
+		if apiErr.Code == 403 {
+			return retry403
+		}
+		if apiErr.Code == 404 && strings.Contains(strings.ToLower(apiErr.Message), errStrBucketNotExist) {
+			return retry404BucketDoesNotExist
+		}
 	}
 
-	// This is the same case as above, but for gRPC UNAUTHENTICATED errors. See
-	// https://github.com/golang/oauth2/issues/623
-	// TODO(b/518674297): Please incorporate the correct fix post resolution of the above issue.
 	if status, ok := status.FromError(err); ok {
+		// This is the same case as above, but for gRPC UNAUTHENTICATED errors. See
+		// https://github.com/golang/oauth2/issues/623
+		// TODO(b/518674297): Please incorporate the correct fix post resolution of the above issue.
 		if status.Code() == codes.Unauthenticated {
 			return retryUnauthenticated
 		}
+		if status.Code() == codes.PermissionDenied {
+			return retryPermissionDenied
+		}
+		if status.Code() == codes.NotFound && strings.Contains(strings.ToLower(status.Message()), errStrBucketNotExist) {
+			return retryNotFoundBucketDoesNotExist
+		}
 	}
 	return noRetry
 }
 
 // ShouldRetryWithoutLogging checks if the error is transient and should be retried.
 // This method is same as ShouldRetry except it doesn't add warning logs.
 func ShouldRetryWithoutLogging(err error) bool {
-	return determineRetryAction(err) != noRetry
+	switch determineRetryAction(err) {
+	case retryTransient, retry401, retryUnauthenticated:
+		return true
+	default:
+		return false
+	}
 }
 
 // ShouldRetryWithRetryContext checks if the given error is transient and should be retried,
@@ -114,3 +143,9 @@ func ShouldRetryWithMonitoringAndRetryContext(
 	metricHandle.GcsRetryCount(1, val)
 	return retry
 }
+
+// ShouldRetryOnMount checks if the error is retryable during mount initialization.
+// In addition to standard transient errors, it retries HTTP 403/404 and gRPC PermissionDenied/NotFound errors.
+func ShouldRetryOnMount(err error) bool {
+	return determineRetryAction(err) != noRetry
+}

internal/storage/storageutil/custom_retry_test.go

@@ -18,6 +18,7 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"net"
 	"net/url"
@@ -180,6 +181,21 @@ func TestShouldRetryWithoutLogging(t *testing.T) {
 			},
 			expectedResult: false,
 		},
+		{
+			name: "403 error - non-retryable for regular ops",
+			err: &googleapi.Error{
+				Code: 403,
+			},
+			expectedResult: false,
+		},
+		{
+			name: "404 bucket missing error - non-retryable for regular ops",
+			err: &googleapi.Error{
+				Code:    404,
+				Message: "The specified bucket does not exist.",
+			},
+			expectedResult: false,
+		},
 	}
 
 	for _, tc := range testCases {
@@ -233,7 +249,22 @@ func TestDetermineRetryAction(t *testing.T) {
 		{
 			name:     "PermissionDeniedGrpcError",
 			err:      status.Error(codes.PermissionDenied, "permission denied"),
-			expected: noRetry,
+			expected: retryPermissionDenied,
+		},
+		{
+			name:     "GoogleApiError403",
+			err:      &googleapi.Error{Code: 403},
+			expected: retry403,
+		},
+		{
+			name:     "GoogleApiError404BucketNotExist",
+			err:      &googleapi.Error{Code: 404, Message: "The specified bucket does not exist."},
+			expected: retry404BucketDoesNotExist,
+		},
+		{
+			name:     "GrpcNotFoundBucketNotExist",
+			err:      status.Error(codes.NotFound, "The specified bucket does not exist."),
+			expected: retryNotFoundBucketDoesNotExist,
 		},
 		{
 			name:     "UnexpectedEOF",
@@ -423,3 +454,90 @@ func TestShouldRetryWithMonitoringForRetryableErrors(t *testing.T) {
 		})
 	}
 }
+
+func TestShouldRetryOnMount(t *testing.T) {
+	testCases := []struct {
+		name     string
+		err      error
+		expected bool
+	}{
+		{
+			name:     "nil error",
+			err:      nil,
+			expected: false,
+		},
+		{
+			name:     "standard transient error 502",
+			err:      &googleapi.Error{Code: 502},
+			expected: true,
+		},
+		{
+			name:     "standard transient error 401",
+			err:      &googleapi.Error{Code: 401},
+			expected: true,
+		},
+		{
+			name:     "HTTP 403 Forbidden",
+			err:      &googleapi.Error{Code: 403, Message: "Permission denied on resource"},
+			expected: true,
+		},
+		{
+			name:     "HTTP 404 missing bucket",
+			err:      &googleapi.Error{Code: 404, Message: "The specified bucket does not exist."},
+			expected: true,
+		},
+		{
+			name:     "HTTP 404 missing bucket mixed case",
+			err:      &googleapi.Error{Code: 404, Message: "The Specified Bucket Does Not Exist."},
+			expected: true,
+		},
+		{
+			name:     "HTTP 404 missing object",
+			err:      &googleapi.Error{Code: 404, Message: "No such object: my-bucket/test-object"},
+			expected: false,
+		},
+		{
+			name:     "gRPC PermissionDenied",
+			err:      status.Error(codes.PermissionDenied, "caller does not have required permission"),
+			expected: true,
+		},
+		{
+			name:     "gRPC NotFound missing bucket",
+			err:      status.Error(codes.NotFound, "The specified bucket does not exist."),
+			expected: true,
+		},
+		{
+			name:     "gRPC NotFound missing bucket mixed case",
+			err:      status.Error(codes.NotFound, "The Specified Bucket Does Not Exist."),
+			expected: true,
+		},
+		{
+			name:     "gRPC NotFound missing object",
+			err:      status.Error(codes.NotFound, "No such object: my-bucket/test-object"),
+			expected: false,
+		},
+		{
+			name:     "permanent error HTTP 400",
+			err:      &googleapi.Error{Code: 400, Message: "Bad Request"},
+			expected: false,
+		},
+		{
+			name:     "permanent error gRPC InvalidArgument",
+			err:      status.Error(codes.InvalidArgument, "invalid bucket name"),
+			expected: false,
+		},
+		{
+			name:     "wrapped gRPC PermissionDenied",
+			err:      fmt.Errorf("mount failed: %w", status.Error(codes.PermissionDenied, "caller does not have required permission")),
+			expected: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			result := ShouldRetryOnMount(tc.err)
+
+			assert.Equal(t, tc.expected, result)
+		})
+	}
+}