Faster upload token refresh on 5xx errors during uploads

b2sdk/_internal/b2http.py

@@ -42,6 +42,8 @@
     ClockSkew,
     ConnectionReset,
     PotentialS3EndpointPassedAsRealm,
+    ServiceError,
+    ServiceErrorDuringUpload,
     UnknownError,
     UnknownHost,
     interpret_b2_error,
@@ -363,6 +365,11 @@ def post_content_return_json(
             # this forces a token refresh, which is necessary if request is still alive
             # on the server but has terminated for some reason on the client. See #79
             raise B2RequestTimeoutDuringUpload()
+        except ServiceError as e:
+            # Convert ServiceError to ServiceErrorDuringUpload for upload operations.
+            # This disables HTTP-level retries and forces the upload manager to clear
+            # cached upload tokens and request fresh ones.
+            raise ServiceErrorDuringUpload(str(e))
 
     def post_json_return_json(self, url, headers, params, try_count: int = TRY_COUNT_OTHER):
         """

b2sdk/_internal/exception.py

@@ -434,6 +434,23 @@ class ServiceError(TransientErrorMixin, B2Error):
     """
 
 
+class ServiceErrorDuringUpload(ServiceError):
+    """
+    A :py:class:`b2sdk.v3.exception.ServiceError` that occurred during an upload operation.
+
+    This exception disables HTTP-level retries to force the error to propagate
+    back to the upload manager, which will clear cached upload tokens and request
+    fresh ones. This is necessary because 5xx errors may indicate the upload URL
+    is no longer valid (typically '503 Service Unavailable').
+
+    Similar to :py:class:`b2sdk.v3.exception.B2RequestTimeoutDuringUpload`, this ensures upload token refresh
+    happens when the server is experiencing issues.
+    """
+
+    def should_retry_http(self):
+        return False
+
+
 class CapExceeded(B2Error):
     def __str__(self):
         return 'Cap exceeded.'

b2sdk/_internal/raw_simulator.py

@@ -624,6 +624,9 @@ def _check_capability(self, account_auth_token, capability):
             # fortunately BucketSimulator makes it easy to retrieve the true account_auth_token
             # from an upload url
             real_auth_token = account_auth_token.split('/')[-1]
+            # Strip the _upload_<id> suffix if present to get the base account auth token
+            if '_upload_' in real_auth_token:
+                real_auth_token = real_auth_token.split('_upload_')[0]
             key = self.api.auth_token_to_key[real_auth_token]
         capabilities = key.get_allowed()['capabilities']
         return capability in capabilities
@@ -773,10 +776,13 @@ def get_file_info_by_name(self, account_auth_token, file_name):
 
     def get_upload_url(self, account_auth_token):
         upload_id = next(self.upload_url_counter)
+        # Generate a unique upload authorization token by appending upload_id to account token
+        # This ensures each upload URL has a different auth token, matching real B2 API behavior
+        unique_upload_token = '%s_upload_%d' % (account_auth_token, upload_id)
         upload_url = 'https://upload.example.com/%s/%d/%s' % (
             self.bucket_id,
             upload_id,
-            account_auth_token,
+            unique_upload_token,
         )
         return dict(bucketId=self.bucket_id, uploadUrl=upload_url, authorizationToken=upload_url)
 

changelog.d/+upload-503-token-refresh.fixed.md

@@ -0,0 +1 @@
+Refresh upload tokens faster when server responds with 5xx during an upload attempt. Fixes [B2_Command_Line_Tool#1118](https://github.com/Backblaze/B2_Command_Line_Tool/issues/1118)

doc/sqlite_account_info_schema.py

@@ -7,7 +7,8 @@
 # License https://www.backblaze.com/using_b2_code.html
 #
 ######################################################################
-""" generates a dot file with SqliteAccountInfo database structure """
+"""generates a dot file with SqliteAccountInfo database structure"""
+
 from __future__ import annotations
 
 import tempfile

test/integration/conftest.py

@@ -9,4 +9,4 @@
 ######################################################################
 
 
-pytest_plugins = ["b2sdk.v3.testing"]
+pytest_plugins = ['b2sdk.v3.testing']

test/unit/test_upload_503_token_refresh.py

@@ -0,0 +1,110 @@
+######################################################################
+#
+# File: test/unit/test_upload_503_token_refresh.py
+#
+# Copyright 2025 Backblaze Inc. All Rights Reserved.
+#
+# License https://www.backblaze.com/using_b2_code.html
+#
+######################################################################
+"""
+Test to verify that upload tokens are refreshed when a 503 error is returned.
+This test simulates the scenario where a server returns a 503 Service Unavailable
+error during file upload and verifies that the SDK properly requests new upload
+tokens for subsequent retry attempts.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+from b2sdk._internal.exception import ServiceError
+from b2sdk.v3 import B2Api, B2HttpApiConfig, DummyCache, RawSimulator, StubAccountInfo
+
+from .test_base import TestBase
+
+
+class TestUpload503TokenRefresh(TestBase):
+    """Test that 503 errors trigger upload token refresh using RawSimulator"""
+
+    def setUp(self):
+        self.bucket_name = 'test-bucket'
+        self.account_info = StubAccountInfo()
+        self.api = B2Api(
+            self.account_info,
+            cache=DummyCache(),
+            api_config=B2HttpApiConfig(_raw_api_class=RawSimulator),
+        )
+        self.simulator = self.api.session.raw_api
+        (self.account_id, self.master_key) = self.simulator.create_account()
+        self.api.authorize_account(
+            application_key_id=self.account_id,
+            application_key=self.master_key,
+            realm='production',
+        )
+        self.bucket = self.api.create_bucket(self.bucket_name, 'allPublic')
+
+    def test_upload_503_triggers_token_refresh(self):
+        """
+        Test that when a 503 error occurs during upload, the SDK:
+        1. Retries the upload
+        2. Uses a different upload token for the retry
+        """
+        # Track upload URLs/tokens used during upload attempts
+        upload_urls_used = []
+
+        # Wrap the upload_file method to track upload URLs
+        original_upload_file = self.simulator.upload_file
+
+        def tracked_upload_file(upload_url, *args, **kwargs):
+            upload_urls_used.append(upload_url)
+            return original_upload_file(upload_url, *args, **kwargs)
+
+        # Inject a 503 error for the first upload attempt
+        self.simulator.set_upload_errors(
+            [ServiceError('503 service_unavailable Service Unavailable')]
+        )
+
+        # Patch upload_file to track URLs
+        with patch.object(self.simulator, 'upload_file', side_effect=tracked_upload_file):
+            # Perform upload - should fail once with 503, then retry and succeed
+            data = b'test data for 503 error scenario'
+            file_name = 'test_503.txt'
+
+            self.bucket.upload_bytes(data, file_name)
+
+        # Verify the upload succeeded
+        file_info = self.bucket.get_file_info_by_name(file_name)
+        assert file_info is not None
+
+        # Verify that at least 2 upload attempts were made
+        assert (
+            len(upload_urls_used) >= 2
+        ), f'Expected at least 2 upload attempts, but got {len(upload_urls_used)}'
+
+        # Extract auth tokens from the URLs
+        # URL format: https://upload.example.com/bucket_id/upload_id/auth_token
+        first_url = upload_urls_used[0]
+        second_url = upload_urls_used[1]
+
+        first_auth_token = first_url.split('/')[-1]
+        second_auth_token = second_url.split('/')[-1]
+
+        print('\n✓ Upload token refresh test results:')
+        print(f'  First URL:   {first_url}')
+        print(f'  Second URL:  {second_url}')
+        print(f'  First auth token:  {first_auth_token}')
+        print(f'  Second auth token: {second_auth_token}')
+        print(f'  Total upload attempts: {len(upload_urls_used)}')
+
+        # Verify that auth tokens are different after a 503 error.
+        # This confirms that the SDK properly clears cached upload tokens
+        # and requests fresh ones when a 503 Service Error occurs.
+        assert first_auth_token != second_auth_token, (
+            f'BUG: Auth tokens are the same after 503 error!\n'
+            f'Expected different auth tokens, but got:\n'
+            f'  First:  {first_auth_token}\n'
+            f'  Second: {second_auth_token}\n'
+            f'This indicates the SDK is cycling through pre-fetched upload URLs\n'
+            f'instead of requesting fresh upload tokens after a 503 error.'
+        )