fix: seven bugs found in a whole-branch review, each with a failing-before test

.sdkharness/tests/conformance/client.sync

@@ -9,6 +9,10 @@ claims it):
             policy; sync down into a fresh directory
     Assert  all 3 land; the remote matches the local side after the second
             pass; the round-tripped bytes are identical
+    Leg     reporter=None (sync_folders documents `SyncReport | None`): sync
+            down from the bucket, and sync up from a directory holding a
+            dangling symlink; neither may crash, and the unreadable entry is
+            skipped, not uploaded
 
 NO DOCUMENTED CONTRACT FOUND -- this is SDK-side/ecosystem tooling, not a B2 API contract.
 B2 documents only sync's building blocks. "Remote matches local" is read
@@ -257,7 +261,16 @@ def write(path, data):
         handle.write(data)
 
 
-def sync(b2, syncer, source, destination):
+def sync(b2, syncer, source, destination, silent=False):
+    """Run one sync pass. `silent` passes reporter=None, which sync_folders documents as valid."""
+    if silent:
+        syncer.sync_folders(
+            source_folder=source,
+            dest_folder=destination,
+            now_millis=int(time.time() * 1000),
+            reporter=None,
+        )
+        return
     reporter = b2.SyncReport(io.StringIO(), True)
     try:
         syncer.sync_folders(
@@ -353,6 +366,62 @@ def run(target: str) -> None:
         extra = sorted(set(os.listdir(down_dir)) - set(contents))
         if extra:
             raise Failure('sync down', f'{len(extra)} unexpected file(s) came down')
+
+        # Leg: reporter=None. sync_folders documents `SyncReport | None`, so a caller
+        # that wants no progress output must get a normal sync, never an AttributeError
+        # (b2-source path) and never a crash on an inaccessible local file.
+        silent_down = tempfile.mkdtemp(prefix='sdkharness-conf-silent-down-')
+        silent_up = tempfile.mkdtemp(prefix='sdkharness-conf-silent-up-')
+        try:
+            step(
+                'sync down with reporter=None',
+                lambda: sync(
+                    b2,
+                    syncer,
+                    b2.parse_folder(remote_url, api),
+                    b2.parse_folder(silent_down, api),
+                    silent=True,
+                ),
+            )
+            if sorted(os.listdir(silent_down)) != sorted(contents):
+                raise Failure('sync down with reporter=None', 'the files did not all come down')
+            for name, expected in contents.items():
+                with open(os.path.join(silent_down, name), 'rb') as handle:
+                    if handle.read() != expected:
+                        raise Failure('sync down with reporter=None', "a file's bytes differ")
+
+            # A local source holding a file that cannot be read. A dangling symlink is
+            # inaccessible whoever runs the check (chmod 000 is not for root).
+            write(os.path.join(silent_up, 'good.txt'), payload_of(300))
+            os.symlink(
+                os.path.join(silent_up, 'does-not-exist'), os.path.join(silent_up, 'a-dangling')
+            )
+            silent_url = f'b2://{bucket.name}/{REMOTE_PREFIX}-silent'
+            step(
+                'sync up with reporter=None past an inaccessible file',
+                lambda: sync(
+                    b2,
+                    syncer,
+                    b2.parse_folder(silent_up, api),
+                    b2.parse_folder(silent_url, api),
+                    silent=True,
+                ),
+            )
+            silent_landed = step(
+                'list the reporter=None upload',
+                lambda: sorted(
+                    v.file_name.rsplit('/', 1)[-1]
+                    for v, _folder in bucket.ls(REMOTE_PREFIX + '-silent/', recursive=True)
+                ),
+            )
+            if silent_landed != ['good.txt']:
+                raise Failure(
+                    'sync up with reporter=None past an inaccessible file',
+                    f'the remote holds {len(silent_landed)} file(s), expected only the readable one',
+                )
+        finally:
+            shutil.rmtree(silent_down, ignore_errors=True)
+            shutil.rmtree(silent_up, ignore_errors=True)
     finally:
         teardown(api, bucket)
         shutil.rmtree(up_dir, ignore_errors=True)

b2sdk/_internal/account_info/sqlite_account_info.py

@@ -441,9 +441,8 @@ def _migrate_allowed_to_multi_bucket(self, *, version: int):
             allowed['buckets'] = None
 
         allowed_text = json.dumps(allowed)
-        stmt = f"UPDATE account SET allowed = ('{allowed_text}');"
 
-        self._perform_update(version, [stmt])
+        self._perform_update(version, [('UPDATE account SET allowed = ?;', (allowed_text,))])
 
     def _ensure_update(self, update_number, update_commands: list[str]):
         """
@@ -466,17 +465,21 @@ def _get_update_count(self, update_number: int):
             )
             return cursor.fetchone()[0]
 
-    def _perform_update(self, update_number, update_commands: list[str]):
+    def _perform_update(self, update_number, update_commands: list[str | tuple[str, tuple]]):
         with self._get_connection() as conn:
             conn.execute('BEGIN')
             for command in update_commands:
-                conn.execute(command)
+                if isinstance(command, str):
+                    conn.execute(command)
+                else:
+                    conn.execute(*command)
             conn.execute('INSERT INTO update_done (update_number) VALUES (?);', (update_number,))
 
     def clear(self):
         """
         Remove all info about accounts and buckets.
         """
+        self._reset_upload_pools()
         with self._get_connection() as conn:
             conn.execute('DELETE FROM account;')
             conn.execute('DELETE FROM bucket;')

b2sdk/_internal/bucket.py

@@ -1485,6 +1485,10 @@ def copy(
             file_retention=file_retention,
             legal_hold=legal_hold,
             cache_control=cache_control,
+            expires=expires,
+            content_disposition=content_disposition,
+            content_encoding=content_encoding,
+            content_language=content_language,
             min_part_size=min_part_size,
             max_part_size=max_part_size,
         )

b2sdk/_internal/raw_api.py

@@ -392,6 +392,7 @@ def start_large_file(
         server_side_encryption: EncryptionSetting | None = None,
         file_retention: FileRetentionSetting | None = None,
         legal_hold: LegalHold | None = None,
+        custom_upload_timestamp: int | None = None,
     ):
         pass
 
@@ -880,7 +881,7 @@ def start_large_file(
             kwargs['fileRetention'] = file_retention.serialize_to_json_for_request()
 
         if custom_upload_timestamp is not None:
-            kwargs['custom_upload_timestamp'] = custom_upload_timestamp
+            kwargs['customUploadTimestamp'] = custom_upload_timestamp
 
         return self._post_json(
             api_url,

b2sdk/_internal/replication/monitoring.py

@@ -12,7 +12,8 @@
 import sys
 from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
-from queue import Queue
+from queue import Full, Queue
+from threading import Event
 from typing import ClassVar, Iterator
 
 from ..api import B2Api
@@ -195,29 +196,49 @@ def scan(self, scan_destination: bool = True) -> AbstractScanReport:
         """
         report = self.REPORT_CLASS()
         queue = Queue(maxsize=self.QUEUE_SIZE)
+        stop = Event()  # set when the consumer dies, so the producer never blocks on a full queue
+
+        def put(item) -> bool:
+            while not stop.is_set():
+                try:
+                    queue.put(item, timeout=0.1)
+                    return True
+                except Full:
+                    pass
+            return False
 
         if not scan_destination:
 
             def fill_queue():
-                for path in self.source_folder.all_files(
-                    policies_manager=self.scan_policies_manager,
-                    reporter=self.report,
-                ):
-                    queue.put((path,), block=True)
-                queue.put(None, block=True)
+                try:
+                    for path in self.source_folder.all_files(
+                        policies_manager=self.scan_policies_manager,
+                        reporter=self.report,
+                    ):
+                        if not put((path,)):
+                            return
+                finally:
+                    put(None)
         else:
 
             def fill_queue():
-                for pair in self.iter_pairs():
-                    queue.put(pair, block=True)
-                queue.put(None, block=True)
+                try:
+                    for pair in self.iter_pairs():
+                        if not put(pair):
+                            return
+                finally:
+                    put(None)
 
         def consume_queue():
             while True:
                 items = queue.get(block=True)
                 if items is None:  # using None as "end of queue" marker
                     break
-                report.add(*items)
+                try:
+                    report.add(*items)
+                except BaseException:
+                    stop.set()
+                    raise
 
         with ThreadPoolExecutor(max_workers=2) as thread_pool:
             futures = [

b2sdk/_internal/scan/folder.py

@@ -289,7 +289,7 @@ def _walk_relative_paths(
                 except OSError:
                     if reporter is not None:
                         reporter.local_access_error(str(local_path))
-                        continue
+                    continue
 
                 local_scan_path = LocalPath(
                     absolute_path=self.make_full_path(str(relative_file_path)),
@@ -303,7 +303,7 @@ def _walk_relative_paths(
                 if not _file_read_access(local_path):
                     if reporter is not None:
                         reporter.local_permission_error(str(local_path))
-                        continue
+                    continue
 
                 yield local_scan_path
 

b2sdk/_internal/sync/sync.py

@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 import concurrent.futures as futures
+import io
 import logging
 from enum import Enum, unique
 from typing import cast
@@ -202,7 +203,11 @@ def sync_folders(
         queue_limit = self.max_workers + 1000
         sync_executor = BoundedQueueExecutor(unbounded_executor, queue_limit=queue_limit)
 
-        if source_type == 'local' and reporter is not None:
+        if reporter is None:
+            # silent reporter so that downstream code can call it unconditionally
+            reporter = SyncReport(io.StringIO(), True)
+
+        if source_type == 'local':
             # Start the thread that counts the local files. That's the operation
             # that should be fastest, and it provides scale for the progress reporting.
             sync_executor.submit(count_files, source_folder, reporter, self.policies_manager)

b2sdk/_internal/transfer/inbound/downloader/parallel.py

@@ -156,7 +156,7 @@ def _finish_hashing(self, first_part, file, hasher, content_length):
                 stop = True
             else:
                 to_hash = data
-            hasher.update(data)
+            hasher.update(to_hash)
             current_offset += len(to_hash)
             if stop:
                 break
@@ -248,6 +248,8 @@ def __init__(self, file, max_queue_depth):
         self.file = file
         self.queue = queue.Queue(max_queue_depth)
         self.total = 0
+        self.error: BaseException | None = None
+        self._stopped = threading.Event()
         self.stats_collector = StatsCollector(str(self.file), 'writer', 'seek')
         super().__init__()
 
@@ -266,11 +268,18 @@ def run(self):
                 if shutdown:
                     break
 
-                with stats_collector_other:
-                    file.seek(offset)
+                if self.error is not None:
+                    continue  # keep draining so that producers never block
 
-                with stats_collector_write:
-                    file.write(data)
+                try:
+                    with stats_collector_other:
+                        file.seek(offset)
+
+                    with stats_collector_write:
+                        file.write(data)
+                except Exception as e:
+                    self.error = e
+                    continue
 
                 self.total += len(data)
 
@@ -279,12 +288,25 @@ def __enter__(self):
         return self
 
     def queue_write(self, offset: int, data: bytes) -> None:
-        self.queue.put((False, offset, data))
+        item = (False, offset, data)
+        while True:
+            if self.error is not None:
+                raise self.error
+            if self._stopped.is_set():
+                raise RuntimeError('writer has been stopped')
+            try:
+                self.queue.put(item, timeout=0.1)
+                return
+            except queue.Full:
+                pass
 
     def __exit__(self, exc_type, exc_val, exc_tb):
-        self.queue.put((True, None, None))
+        self._stopped.set()  # make straggling producers fail instead of blocking forever
+        self.queue.put((True, None, None))  # writer keeps draining, so this cannot block for long
         self.join()
         self.stats_collector.report()
+        if exc_type is None and self.error is not None:
+            raise self.error
 
 
 def download_first_part(

changelog.d/+copy-file-headers.fixed.md

@@ -0,0 +1 @@
+Pass `expires`, `content_disposition`, `content_encoding` and `content_language` through `Bucket.copy` to the created file instead of dropping them.

changelog.d/+custom-upload-timestamp.fixed.md

@@ -0,0 +1 @@
+Send `customUploadTimestamp` under its B2 wire name when starting a large file, so custom upload timestamps are no longer ignored.

changelog.d/+local-scan-errors.fixed.md

@@ -0,0 +1 @@
+Keep scanning a local folder after a file cannot be read or accessed when no reporter is given, instead of failing on that file.

changelog.d/+parallel-download-writer.fixed.md

@@ -0,0 +1 @@
+Parallel download: hash the data that was actually written, and surface a failing writer to producers instead of blocking them or losing the error.

changelog.d/+replication-monitor-hang.fixed.md

@@ -0,0 +1 @@
+Stop the replication monitor from hanging when its consumer fails: the producer no longer blocks on a full queue and always signals the end of input.

changelog.d/+sqlite-allowed-param.fixed.md

@@ -0,0 +1 @@
+Store the account's allowed capabilities in SQLite with a bound parameter instead of string formatting, and reset upload pools after an account-info schema update.

changelog.d/+sync-silent-reporter.fixed.md

@@ -0,0 +1 @@
+Accept `reporter=None` in `Synchronizer.sync_folders` by using a silent reporter instead of failing.

test/unit/account_info/test_account_info.py

@@ -217,6 +217,18 @@ def test_set_auth_data_compatibility(self, account_info_default_data, allowed):
         )
         assert account_info.get_allowed() == allowed
 
+    def test_clear_drops_cached_upload_urls(self, account_info_default_data):
+        """After ``clear()`` (log out) no upload URL or token of the cleared account may be handed out."""
+        account_info = self._make_info()
+        account_info.set_auth_data(**account_info_default_data)
+        account_info.put_bucket_upload_url('bucket-0', 'http://bucket-0', 'bucket-0_auth')
+        account_info.put_large_file_upload_url('file_0', 'http://file_0', 'auth_0')
+
+        account_info.clear()
+
+        assert (None, None) == account_info.take_bucket_upload_url('bucket-0')
+        assert (None, None) == account_info.take_large_file_upload_url('file_0')
+
     def test_clear_bucket_upload_data(self):
         account_info = self._make_info()
         account_info.put_bucket_upload_url('bucket-0', 'http://bucket-0', 'bucket-0_auth')