diff --git a/b2sdk/_internal/testing/helpers/bucket_manager.py b/b2sdk/_internal/testing/helpers/bucket_manager.py
index 014359452..bc76ef3b7 100644
--- a/b2sdk/_internal/testing/helpers/bucket_manager.py
+++ b/b2sdk/_internal/testing/helpers/bucket_manager.py
@@ -82,6 +82,11 @@ def new_bucket_info(self) -> dict:
             'created_by': NODE_DESCRIPTION,
         }
 
+    @tenacity.retry(
+        retry=tenacity.retry_if_exception_type(TooManyRequests),
+        wait=tenacity.wait_exponential(),
+        stop=tenacity.stop_after_attempt(8),
+    )
     def create_bucket(self, bucket_type: str = 'allPublic', **kwargs) -> Bucket:
         bucket_name = kwargs.pop('name', self.new_bucket_name())
 
diff --git a/b2sdk/_internal/transfer/outbound/upload_manager.py b/b2sdk/_internal/transfer/outbound/upload_manager.py
index b6c8dabd0..842d8c3f0 100644
--- a/b2sdk/_internal/transfer/outbound/upload_manager.py
+++ b/b2sdk/_internal/transfer/outbound/upload_manager.py
@@ -204,18 +204,17 @@ def _upload_small_file(
         content_length = upload_source.get_content_length()
         exception_info_list = []
         progress_listener.set_total_bytes(content_length)
-        for _ in range(self.MAX_UPLOAD_ATTEMPTS):
-            try:
-                with upload_source.open() as file:
-                    input_stream = ReadingStreamWithProgress(
-                        file, progress_listener, length=content_length
-                    )
-                    if upload_source.is_sha1_known():
-                        content_sha1 = upload_source.get_content_sha1()
-                    else:
-                        input_stream = StreamWithHash(input_stream, stream_length=content_length)
-                        content_sha1 = HEX_DIGITS_AT_END
-                    # it is important that `len()` works on `input_stream`
+        with upload_source.open() as file:
+            input_stream = ReadingStreamWithProgress(file, progress_listener, length=content_length)
+            if upload_source.is_sha1_known():
+                content_sha1 = upload_source.get_content_sha1()
+            else:
+                input_stream = StreamWithHash(input_stream, stream_length=content_length)
+                content_sha1 = HEX_DIGITS_AT_END
+            # it is important that `len()` works on `input_stream`
+
+            for _ in range(self.MAX_UPLOAD_ATTEMPTS):
+                try:
                     response = self.services.session.upload_file(
                         bucket_id,
                         file_name,
@@ -236,10 +235,10 @@ def _upload_small_file(
                     ), '{} != {}'.format(content_sha1, response['contentSha1'])
                     return self.services.api.file_version_factory.from_api_response(response)
 
-            except B2Error as e:
-                if not e.should_retry_upload():
-                    raise
-                exception_info_list.append(e)
-                self.account_info.clear_bucket_upload_data(bucket_id)
+                except B2Error as e:
+                    if not e.should_retry_upload():
+                        raise
+                    exception_info_list.append(e)
+                    self.account_info.clear_bucket_upload_data(bucket_id)
 
         raise MaxRetriesExceeded(self.MAX_UPLOAD_ATTEMPTS, exception_info_list)
diff --git a/changelog.d/+bucket-manager-create-retries.infrastructure.md b/changelog.d/+bucket-manager-create-retries.infrastructure.md
new file mode 100644
index 000000000..a86577a2b
--- /dev/null
+++ b/changelog.d/+bucket-manager-create-retries.infrastructure.md
@@ -0,0 +1 @@
+Add exponential retries for bucket creation in `testing.helpers.BucketManager`.
\ No newline at end of file
diff --git a/changelog.d/+upload_unbound_stream_retry_value_error.fixed.md b/changelog.d/+upload_unbound_stream_retry_value_error.fixed.md
new file mode 100644
index 000000000..1c430479b
--- /dev/null
+++ b/changelog.d/+upload_unbound_stream_retry_value_error.fixed.md
@@ -0,0 +1 @@
+Fixed a retry bug in `upload_unbound_stream()` small-file uploads where a retryable upload error could cause a one-shot buffered stream to be reopened after it was closed, raising `ValueError: I/O operation on closed file`.
diff --git a/test/unit/bucket/test_bucket.py b/test/unit/bucket/test_bucket.py
index 108c46efd..84030b2fe 100644
--- a/test/unit/bucket/test_bucket.py
+++ b/test/unit/bucket/test_bucket.py
@@ -1918,6 +1918,11 @@ def test_upload_one_retryable_error(self):
         data = b'hello world'
         self.bucket.upload_bytes(data, 'file1')
 
+    def test_upload_unbound_stream_one_retryable_error(self):
+        self.simulator.set_upload_errors([CanRetry(True)])
+        data = b'hello world'
+        self.bucket.upload_unbound_stream(io.BytesIO(data), 'file1')
+
     def test_upload_timeout(self):
         self.simulator.set_upload_errors([B2RequestTimeoutDuringUpload()])
         data = b'hello world'