diff --git a/application/single_app/functions_data_management.py b/application/single_app/functions_data_management.py index 34f6dbc33..5b6681656 100644 --- a/application/single_app/functions_data_management.py +++ b/application/single_app/functions_data_management.py @@ -5,12 +5,14 @@ import base64 from email.utils import parsedate_to_datetime import hashlib +import io import json import logging import os import random import re import socket +import struct import time import tempfile import uuid @@ -21,7 +23,12 @@ from azure.core import MatchConditions from azure.core.credentials import AzureKeyCredential -from azure.core.exceptions import ResourceExistsError, ResourceNotFoundError +from azure.core.exceptions import ( + ResourceExistsError, + ResourceNotFoundError, + ServiceRequestError, + ServiceResponseError, +) import azure.cosmos as azure_cosmos from azure.cosmos import PartitionKey from azure.cosmos.exceptions import CosmosResourceNotFoundError @@ -29,7 +36,7 @@ from azure.search.documents import SearchClient from azure.search.documents.indexes import SearchIndexClient from azure.search.documents.indexes.models import SearchField, SearchFieldDataType, SearchIndex -from azure.storage.blob import BlobServiceClient, ContentSettings +from azure.storage.blob import BlobBlock, BlobServiceClient, ContentSettings from cryptography.fernet import Fernet import config as app_config @@ -173,6 +180,12 @@ DATA_MANAGEMENT_BACKUP_MAX_PARALLEL_OPERATIONS = 16 DATA_MANAGEMENT_BACKUP_DEFAULT_RETRY_COUNT = 5 DATA_MANAGEMENT_BACKUP_MAX_RETRY_COUNT = 10 +DATA_MANAGEMENT_BLOB_BACKUP_DEFAULT_PARALLEL_OPERATIONS = 4 +DATA_MANAGEMENT_BLOB_BACKUP_MAX_PARALLEL_OPERATIONS = 8 +DATA_MANAGEMENT_BLOB_BACKUP_DEFAULT_CHUNK_SIZE_MIB = 8 +DATA_MANAGEMENT_BLOB_BACKUP_MIN_CHUNK_SIZE_MIB = 1 +DATA_MANAGEMENT_BLOB_BACKUP_MAX_CHUNK_SIZE_MIB = 16 +DATA_MANAGEMENT_BLOB_BACKUP_CLEAN_TRANSFER_RECOVERY_COUNT = 3 DATA_MANAGEMENT_BACKUP_DEFAULT_SOURCE_RU = 10000 DATA_MANAGEMENT_BACKUP_MAX_SOURCE_RU = 10000 DATA_MANAGEMENT_BACKUP_CAPACITY_FAILURE_POLICY_FAIL = "fail" @@ -187,6 +200,11 @@ DATA_MANAGEMENT_SEARCH_BACKUP_CLEAN_PAGE_RECOVERY_COUNT = 3 DATA_MANAGEMENT_SEARCH_BACKUP_MAX_INDEX_CONCURRENCY = 3 DATA_MANAGEMENT_BLOB_SERVICE_TIMEOUT_SECONDS = 120 +DATA_MANAGEMENT_BLOB_BACKUP_ENCRYPTED_FORMAT = "fernet-chunked-v1" +DATA_MANAGEMENT_BLOB_BACKUP_RAW_FORMAT = "raw-v1" +DATA_MANAGEMENT_BLOB_BACKUP_ENCRYPTED_MAGIC = b"SCBF1\n" +DATA_MANAGEMENT_BLOB_BACKUP_ENCRYPTED_HEADER_SIZE = 49 +DATA_MANAGEMENT_BLOB_BACKUP_MAX_BLOCKS = 50000 DATA_MANAGEMENT_MIGRATION_REMOTE_REQUEST_TIMEOUT_SECONDS = 30 DATA_MANAGEMENT_MIGRATION_HEARTBEAT_INTERVAL_SECONDS = 2.0 DATA_MANAGEMENT_MIGRATION_HEARTBEAT_POLL_SECONDS = 1.0 @@ -257,6 +275,9 @@ "max_parallel_operations": 1, "backup_max_parallel_operations": DATA_MANAGEMENT_BACKUP_DEFAULT_PARALLEL_OPERATIONS, "backup_retry_count": DATA_MANAGEMENT_BACKUP_DEFAULT_RETRY_COUNT, + "backup_blob_max_parallel_operations": DATA_MANAGEMENT_BLOB_BACKUP_DEFAULT_PARALLEL_OPERATIONS, + "backup_blob_chunk_size_mib": DATA_MANAGEMENT_BLOB_BACKUP_DEFAULT_CHUNK_SIZE_MIB, + "backup_blob_retry_count": DATA_MANAGEMENT_BACKUP_DEFAULT_RETRY_COUNT, "backup_temporary_source_ru_enabled": False, "backup_temporary_source_ru": DATA_MANAGEMENT_BACKUP_DEFAULT_SOURCE_RU, "backup_capacity_failure_policy": DATA_MANAGEMENT_BACKUP_CAPACITY_FAILURE_POLICY_CONTINUE, @@ -712,6 +733,24 @@ def normalize_data_management_settings(payload=None, existing_settings=None, cur minimum=1, maximum=DATA_MANAGEMENT_BACKUP_MAX_RETRY_COUNT, ) + source["backup_blob_max_parallel_operations"] = _safe_int( + source.get("backup_blob_max_parallel_operations"), + default=DATA_MANAGEMENT_BLOB_BACKUP_DEFAULT_PARALLEL_OPERATIONS, + minimum=1, + maximum=DATA_MANAGEMENT_BLOB_BACKUP_MAX_PARALLEL_OPERATIONS, + ) + source["backup_blob_chunk_size_mib"] = _safe_int( + source.get("backup_blob_chunk_size_mib"), + default=DATA_MANAGEMENT_BLOB_BACKUP_DEFAULT_CHUNK_SIZE_MIB, + minimum=DATA_MANAGEMENT_BLOB_BACKUP_MIN_CHUNK_SIZE_MIB, + maximum=DATA_MANAGEMENT_BLOB_BACKUP_MAX_CHUNK_SIZE_MIB, + ) + source["backup_blob_retry_count"] = _safe_int( + source.get("backup_blob_retry_count"), + default=DATA_MANAGEMENT_BACKUP_DEFAULT_RETRY_COUNT, + minimum=1, + maximum=DATA_MANAGEMENT_BACKUP_MAX_RETRY_COUNT, + ) source["backup_temporary_source_ru_enabled"] = _safe_bool( source.get("backup_temporary_source_ru_enabled"), False, @@ -1320,6 +1359,67 @@ def _get_backup_search_retry_count(settings, backup_plan=None): ) +def _get_backup_blob_parallel_operations(settings, backup_plan=None): + execution = ( + (backup_plan or {}).get("source_blob_execution") + if isinstance(backup_plan, dict) else {} + ) + configured_value = ( + (execution or {}).get("max_parallel_operations") + if isinstance(execution, dict) and "max_parallel_operations" in execution else + (settings or {}).get("backup_blob_max_parallel_operations") + if backup_plan is None else + DATA_MANAGEMENT_BLOB_BACKUP_DEFAULT_PARALLEL_OPERATIONS + ) + return _safe_int( + configured_value, + default=DATA_MANAGEMENT_BLOB_BACKUP_DEFAULT_PARALLEL_OPERATIONS, + minimum=1, + maximum=DATA_MANAGEMENT_BLOB_BACKUP_MAX_PARALLEL_OPERATIONS, + ) + + +def _get_backup_blob_chunk_size_bytes(settings, backup_plan=None): + execution = ( + (backup_plan or {}).get("source_blob_execution") + if isinstance(backup_plan, dict) else {} + ) + configured_value = ( + (execution or {}).get("chunk_size_mib") + if isinstance(execution, dict) and "chunk_size_mib" in execution else + (settings or {}).get("backup_blob_chunk_size_mib") + if backup_plan is None else + DATA_MANAGEMENT_BLOB_BACKUP_DEFAULT_CHUNK_SIZE_MIB + ) + chunk_size_mib = _safe_int( + configured_value, + default=DATA_MANAGEMENT_BLOB_BACKUP_DEFAULT_CHUNK_SIZE_MIB, + minimum=DATA_MANAGEMENT_BLOB_BACKUP_MIN_CHUNK_SIZE_MIB, + maximum=DATA_MANAGEMENT_BLOB_BACKUP_MAX_CHUNK_SIZE_MIB, + ) + return chunk_size_mib * 1024 * 1024 + + +def _get_backup_blob_retry_count(settings, backup_plan=None): + execution = ( + (backup_plan or {}).get("source_blob_execution") + if isinstance(backup_plan, dict) else {} + ) + configured_value = ( + (execution or {}).get("retry_count") + if isinstance(execution, dict) and "retry_count" in execution else + (settings or {}).get("backup_blob_retry_count") + if backup_plan is None else + DATA_MANAGEMENT_BACKUP_DEFAULT_RETRY_COUNT + ) + return _safe_int( + configured_value, + default=DATA_MANAGEMENT_BACKUP_DEFAULT_RETRY_COUNT, + minimum=1, + maximum=DATA_MANAGEMENT_BACKUP_MAX_RETRY_COUNT, + ) + + def _get_backup_capacity_failure_policy(settings, backup_plan=None): execution = (backup_plan or {}).get("cosmos_execution") if isinstance(backup_plan, dict) else {} configured_value = ( @@ -1830,6 +1930,7 @@ def _update_migration_state_totals(state): "collision_count", "processed_count", "bytes", + "artifact_bytes", ): totals[field_name] += _safe_int(metrics.get(field_name), default=0, minimum=0) try: @@ -4673,6 +4774,10 @@ def _get_blob_properties_or_none(blob_client): return blob_client.get_blob_properties() except ResourceNotFoundError: return None + except Exception as exc: + if getattr(exc, "status_code", None) == 404: + return None + raise def _get_blob_content_md5(blob_properties): @@ -7514,6 +7619,14 @@ def _normalize_data_management_backup_plan(settings, backup_type, options=None, "page_size": DATA_MANAGEMENT_SEARCH_BACKUP_PAGE_SIZE, "clean_page_recovery_count": DATA_MANAGEMENT_SEARCH_BACKUP_CLEAN_PAGE_RECOVERY_COUNT, }, + "source_blob_execution": { + "max_parallel_operations": _get_backup_blob_parallel_operations(settings), + "chunk_size_mib": ( + _get_backup_blob_chunk_size_bytes(settings) // (1024 * 1024) + ), + "retry_count": _get_backup_blob_retry_count(settings), + "clean_transfer_recovery_count": DATA_MANAGEMENT_BLOB_BACKUP_CLEAN_TRANSFER_RECOVERY_COUNT, + }, "resource_contract": ["cosmos", "ai_search", "source_blobs"], } @@ -7938,14 +8051,18 @@ def sanitize_metric_number(value): "processed_count", "item_count", "blob_count", + "reused_count", "skipped_count", "failed_count", "bytes", + "artifact_bytes", "checkpoint_count", "retry_attempt_count", "throttle_count", "parallel_operations", "active_parallel_operations", + "chunk_size_bytes", + "in_flight_count", "source_page_count", "page_count", "current_page", @@ -7967,6 +8084,7 @@ def sanitize_metric_number(value): "current_container": _safe_text( progress.get("current_container") or progress.get("current_index") ), + "current_file": _safe_text(progress.get("current_file")), "cursor_committed": bool(progress.get("cursor_committed")), "checkpoint": { "next_batch_number": _safe_int(checkpoint.get("next_batch_number"), default=0, minimum=0), @@ -8056,6 +8174,7 @@ def sanitize_item_summary(summary): "skipped_count", "failed_count", "bytes", + "artifact_bytes", "checkpoint_count", "retry_attempt_count", "throttle_count", @@ -8067,20 +8186,28 @@ def sanitize_item_summary(summary): "elapsed_seconds", "records_per_second", "request_units_per_second", + "bytes_per_second", ) }, "telemetry": { "current_container": _safe_text(telemetry.get("current_container")), + "current_file": _safe_text(telemetry.get("current_file")), "current_page": _safe_int(telemetry.get("current_page"), default=0, minimum=0), "checkpoint_position": _safe_int(telemetry.get("checkpoint_position"), default=0, minimum=0), "records_processed": _safe_int(telemetry.get("records_processed"), default=0, minimum=0), "bytes": _safe_int(telemetry.get("bytes"), default=0, minimum=0), + "artifact_bytes": _safe_int( + telemetry.get("artifact_bytes"), + default=0, + minimum=0, + ), "request_units": sanitize_metric_number(telemetry.get("request_units")), "retries": _safe_int(telemetry.get("retries"), default=0, minimum=0), "throttles": _safe_int(telemetry.get("throttles"), default=0, minimum=0), "elapsed_seconds": sanitize_metric_number(telemetry.get("elapsed_seconds")), "records_per_second": sanitize_metric_number(telemetry.get("records_per_second")), "request_units_per_second": sanitize_metric_number(telemetry.get("request_units_per_second")), + "bytes_per_second": sanitize_metric_number(telemetry.get("bytes_per_second")), }, "source_capacity": public_source_capacity, "warnings": [ @@ -8635,7 +8762,10 @@ def _summarize_backup_artifact(artifact): "destination_failed_count", "destination_provenance_skip_count", "blob_count", + "reused_count", "encrypted", + "transfer_format", + "chunk_size_bytes", "container_name", "partition_key_path", "index_name", @@ -10292,49 +10422,522 @@ def _source_blob_container_names(): ] -def _copy_source_blob( +def _get_backup_blob_property(properties, field_name, default=None): + if isinstance(properties, dict): + return properties.get(field_name, default) + return getattr(properties, field_name, default) + + +def _build_backup_blob_source_item(source_container_name, blob_name, properties): + last_modified = _get_backup_blob_property(properties, "last_modified") + source_size = _safe_int( + _get_backup_blob_property(properties, "size"), + default=0, + minimum=0, + ) + source_etag = _safe_text(_get_backup_blob_property(properties, "etag")) + return { + "source_identity": _build_backup_source_identity( + "source_blobs", + source_container_name, + blob_name, + ), + "source_version": _build_backup_source_version({ + "etag": source_etag, + "last_modified": last_modified, + "size": source_size, + }), + "source_etag": source_etag, + "source_size": source_size, + "last_modified": last_modified, + } + + +def _get_backup_blob_metadata(properties): + metadata = _get_backup_blob_property(properties, "metadata", {}) + return metadata if isinstance(metadata, dict) else {} + + +def _build_backup_blob_target_metadata( + backup_job, + source_item, + transfer_format, + status, +): + return { + "simplechatbackupstatus": _safe_text(status), + "simplechatbackupjob": _safe_text((backup_job or {}).get("id")), + "simplechatbackupattempt": _safe_text( + (backup_job or {}).get("backup_attempt_id") + ), + "simplechatbackupleasegeneration": str( + _safe_int((backup_job or {}).get("lease_generation"), default=0, minimum=0) + ), + "simplechatbackupsourceversion": _safe_text( + (source_item or {}).get("source_version") + ), + "simplechatbackupsourcebytes": str( + _safe_int((source_item or {}).get("source_size"), default=0, minimum=0) + ), + "simplechatbackupformat": _safe_text(transfer_format), + } + + +def _is_verified_backup_blob_artifact( + target_properties, + backup_job, + source_item, + transfer_format, +): + if target_properties is None: + return False + metadata = _get_backup_blob_metadata(target_properties) + target_generation = _safe_int( + metadata.get("simplechatbackupleasegeneration"), + default=0, + minimum=0, + ) + current_generation = _safe_int( + (backup_job or {}).get("lease_generation"), + default=0, + minimum=0, + ) + if ( + _safe_text(metadata.get("simplechatbackupjob")) == + _safe_text((backup_job or {}).get("id")) and + target_generation > current_generation + ): + raise DataManagementBackupLeaseLostError( + "A newer backup attempt already finalized this source artifact." + ) + return ( + _safe_text(metadata.get("simplechatbackupstatus")).lower() == "succeeded" and + _safe_text(metadata.get("simplechatbackupjob")) == + _safe_text((backup_job or {}).get("id")) and + bool(_safe_text(metadata.get("simplechatbackupattempt"))) and + target_generation <= current_generation and + _safe_text(metadata.get("simplechatbackupsourceversion")) == + _safe_text((source_item or {}).get("source_version")) and + _safe_int(metadata.get("simplechatbackupsourcebytes"), default=-1) == + _safe_int((source_item or {}).get("source_size"), default=0, minimum=0) and + _safe_text(metadata.get("simplechatbackupformat")) == _safe_text(transfer_format) + ) + + +def _is_retryable_backup_blob_error(error): + status_code = _get_backup_exception_status_code(error) + return ( + isinstance( + error, + (TimeoutError, socket.timeout, ServiceRequestError, ServiceResponseError), + ) or + status_code in {408, 409, 412, 429} or + (status_code is not None and 500 <= status_code <= 599) + ) + + +def _read_backup_blob_range( + source_blob_client, + offset, + length, + source_etag, +): + download_kwargs = { + "offset": offset, + "length": length, + "max_concurrency": 1, + "timeout": DATA_MANAGEMENT_BLOB_SERVICE_TIMEOUT_SECONDS, + } + if source_etag: + download_kwargs.update({ + "etag": source_etag, + "match_condition": MatchConditions.IfNotModified, + }) + download = source_blob_client.download_blob(**download_kwargs) + buffer = io.BytesIO() + download.readinto(buffer) + return buffer.getvalue() + + +def _encode_backup_blob_chunk( + chunk, + fernet, + chunk_index, + total_chunks, + source_version, +): + if fernet is None: + return chunk + source_version_digest = hashlib.sha256( + _safe_text(source_version).encode("utf-8") + ).digest() + authenticated_chunk = ( + source_version_digest + + struct.pack( + ">QQ?", + int(chunk_index), + int(total_chunks), + int(chunk_index) == int(total_chunks) - 1, + ) + + chunk + ) + encrypted_chunk = fernet.encrypt(authenticated_chunk) + prefix = DATA_MANAGEMENT_BLOB_BACKUP_ENCRYPTED_MAGIC if chunk_index == 0 else b"" + return prefix + struct.pack(">I", len(encrypted_chunk)) + encrypted_chunk + + +def _backup_blob_block_id(block_index, transfer_identity): + block_value = f"{_safe_text(transfer_identity)[:24]}:{int(block_index):08d}" + return base64.b64encode(block_value.encode("ascii")).decode("ascii") + + +def _run_backup_source_blob_transfer(*args, **kwargs): + """Convert per-file setup and transfer failures into isolated worker outcomes.""" + source_item = kwargs.get("source_item") + target_blob_name = kwargs.get("target_blob_name") + fernet = kwargs.get("fernet") + if source_item is None and len(args) > 3: + source_item = args[3] + if target_blob_name is None and len(args) > 2: + target_blob_name = args[2] + if fernet is None and len(args) > 4: + fernet = args[4] + try: + return _transfer_backup_source_blob(*args, **kwargs) + except (DataManagementBackupCanceledError, DataManagementBackupLeaseLostError): + raise + except Exception as exc: + return { + "status": "failed", + "artifact_path": _encrypted_blob_name(target_blob_name, fernet), + "source_bytes": _safe_int( + (source_item or {}).get("source_size"), + default=0, + minimum=0, + ), + "artifact_bytes": 0, + "retry_attempt_count": 0, + "throttle_count": 0, + "throttle_delay_seconds": 0.0, + "transfer_format": ( + DATA_MANAGEMENT_BLOB_BACKUP_ENCRYPTED_FORMAT + if fernet else DATA_MANAGEMENT_BLOB_BACKUP_RAW_FORMAT + ), + "failure_summary": ( + _sanitize_data_management_backup_text(str(exc)) or + "Source blob transfer failed." + ), + } + + +def _transfer_backup_source_blob( target_container_client, source_blob_client, target_blob_name, + source_item, fernet=None, backup_job=None, - backup_settings=None, - heartbeat_message="", + chunk_size_bytes=None, + retry_count=None, + cancel_event=None, ): - def download(): - return source_blob_client.download_blob().readall() - - if isinstance(backup_job, dict): - blob_bytes = _run_backup_transfer_with_heartbeat( - backup_job, - backup_settings or {}, - heartbeat_message or "Downloading source blob for backup", - download, - ) - else: - blob_bytes = download() + """Transfer one source blob through bounded blocks and verify its durable target.""" final_blob_name = _encrypted_blob_name(target_blob_name, fernet) - if fernet: - blob_bytes = fernet.encrypt(blob_bytes) + transfer_format = ( + DATA_MANAGEMENT_BLOB_BACKUP_ENCRYPTED_FORMAT + if fernet else DATA_MANAGEMENT_BLOB_BACKUP_RAW_FORMAT + ) + target_blob_client = target_container_client.get_blob_client(final_blob_name) + existing_target = _get_blob_properties_or_none(target_blob_client) + if _is_verified_backup_blob_artifact( + existing_target, + backup_job, + source_item, + transfer_format, + ): + return { + "status": "reused", + "artifact_path": final_blob_name, + "source_bytes": _safe_int(source_item.get("source_size"), default=0, minimum=0), + "artifact_bytes": _safe_int( + _get_backup_blob_property(existing_target, "size"), + default=0, + minimum=0, + ), + "retry_attempt_count": 0, + "throttle_count": 0, + "throttle_delay_seconds": 0.0, + "transfer_format": transfer_format, + } - def upload(): - target_container_client.upload_blob( - name=final_blob_name, - data=blob_bytes, - overwrite=True, - content_settings=ContentSettings(content_type="application/octet-stream"), + configured_chunk_size = _safe_int( + chunk_size_bytes, + default=DATA_MANAGEMENT_BLOB_BACKUP_DEFAULT_CHUNK_SIZE_MIB * 1024 * 1024, + minimum=DATA_MANAGEMENT_BLOB_BACKUP_MIN_CHUNK_SIZE_MIB * 1024 * 1024, + maximum=DATA_MANAGEMENT_BLOB_BACKUP_MAX_CHUNK_SIZE_MIB * 1024 * 1024, + ) + source_size = _safe_int(source_item.get("source_size"), default=0, minimum=0) + required_chunk_size = ( + (source_size + DATA_MANAGEMENT_BLOB_BACKUP_MAX_BLOCKS - 1) // + DATA_MANAGEMENT_BLOB_BACKUP_MAX_BLOCKS + ) + if required_chunk_size > DATA_MANAGEMENT_BLOB_BACKUP_MAX_CHUNK_SIZE_MIB * 1024 * 1024: + raise ValueError( + "Source blob exceeds the configured bounded block-transfer size limit." ) + effective_chunk_size = max(configured_chunk_size, required_chunk_size) + attempts = _safe_int( + retry_count, + default=DATA_MANAGEMENT_BACKUP_DEFAULT_RETRY_COUNT, + minimum=1, + maximum=DATA_MANAGEMENT_BACKUP_MAX_RETRY_COUNT, + ) + source_etag = _safe_text(source_item.get("source_etag")) + total_chunks = max( + 1, + (source_size + effective_chunk_size - 1) // effective_chunk_size, + ) + transfer_identity = hashlib.sha256(json.dumps( + { + "job_id": _safe_text((backup_job or {}).get("id")), + "attempt_id": _safe_text((backup_job or {}).get("backup_attempt_id")), + "lease_generation": _safe_int( + (backup_job or {}).get("lease_generation"), + default=0, + minimum=0, + ), + "source_version": _safe_text(source_item.get("source_version")), + }, + ensure_ascii=True, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8")).hexdigest() + retry_attempt_count = 0 + throttle_count = 0 + throttle_delay_seconds = 0.0 + started_at = time.perf_counter() - if isinstance(backup_job, dict): - _run_backup_transfer_with_heartbeat( - backup_job, + for attempt in range(1, attempts + 1): + try: + if cancel_event is not None and cancel_event.is_set(): + raise DataManagementBackupCanceledError( + "Source blob transfer stopped after backup cancellation or lease loss." + ) + pending_metadata = _build_backup_blob_target_metadata( + backup_job, + source_item, + transfer_format, + "pending", + ) + current_target = _get_blob_properties_or_none(target_blob_client) + if _is_verified_backup_blob_artifact( + current_target, + backup_job, + source_item, + transfer_format, + ): + return { + "status": "reused", + "artifact_path": final_blob_name, + "source_bytes": source_size, + "artifact_bytes": _safe_int( + _get_backup_blob_property(current_target, "size"), + default=0, + minimum=0, + ), + "target_etag": _safe_text( + _get_backup_blob_property(current_target, "etag") + ), + "retry_attempt_count": retry_attempt_count, + "throttle_count": throttle_count, + "throttle_delay_seconds": round(throttle_delay_seconds, 3), + "elapsed_seconds": round(time.perf_counter() - started_at, 3), + "transfer_format": transfer_format, + } + target_condition = {} + current_target_etag = _safe_text( + _get_backup_blob_property(current_target, "etag") + ) + if current_target_etag: + target_condition.update({ + "etag": current_target_etag, + "match_condition": MatchConditions.IfNotModified, + }) + else: + target_condition["match_condition"] = MatchConditions.IfMissing + block_list = [] + artifact_bytes = 0 + if source_size == 0: + empty_payload = _encode_backup_blob_chunk( + b"", + fernet, + chunk_index=0, + total_chunks=1, + source_version=source_item.get("source_version"), + ) + target_blob_client.upload_blob( + data=empty_payload, + overwrite=True, + metadata=pending_metadata, + content_settings=ContentSettings(content_type="application/octet-stream"), + timeout=DATA_MANAGEMENT_BLOB_SERVICE_TIMEOUT_SECONDS, + **target_condition, + ) + artifact_bytes = len(empty_payload) + else: + block_index = 0 + for offset in range(0, source_size, effective_chunk_size): + if cancel_event is not None and cancel_event.is_set(): + raise DataManagementBackupCanceledError( + "Source blob transfer stopped after backup cancellation or lease loss." + ) + length = min(effective_chunk_size, source_size - offset) + source_chunk = _read_backup_blob_range( + source_blob_client, + offset, + length, + source_etag, + ) + if len(source_chunk) != length: + raise RuntimeError( + "Source blob range length changed during transfer." + ) + block_payload = _encode_backup_blob_chunk( + source_chunk, + fernet, + chunk_index=block_index, + total_chunks=total_chunks, + source_version=source_item.get("source_version"), + ) + block_id = _backup_blob_block_id(block_index, transfer_identity) + target_blob_client.stage_block( + block_id=block_id, + data=block_payload, + timeout=DATA_MANAGEMENT_BLOB_SERVICE_TIMEOUT_SECONDS, + ) + block_list.append(BlobBlock(block_id=block_id)) + artifact_bytes += len(block_payload) + block_index += 1 + target_blob_client.commit_block_list( + block_list, + metadata=pending_metadata, + content_settings=ContentSettings(content_type="application/octet-stream"), + timeout=DATA_MANAGEMENT_BLOB_SERVICE_TIMEOUT_SECONDS, + **target_condition, + ) + + target_properties = target_blob_client.get_blob_properties() + if _safe_int( + _get_backup_blob_property(target_properties, "size"), + default=-1, + ) != artifact_bytes: + raise RuntimeError( + "Backup artifact length did not match the committed transfer." + ) + current_source_properties = source_blob_client.get_blob_properties() + current_source_etag = _safe_text( + _get_backup_blob_property(current_source_properties, "etag") + ) + if source_etag and current_source_etag != source_etag: + raise RuntimeError("Source blob changed while it was being backed up.") + succeeded_metadata = _build_backup_blob_target_metadata( + backup_job, + source_item, + transfer_format, + "succeeded", + ) + metadata_kwargs = {"metadata": succeeded_metadata} + target_etag = _safe_text(_get_backup_blob_property(target_properties, "etag")) + if target_etag: + metadata_kwargs.update({ + "etag": target_etag, + "match_condition": MatchConditions.IfNotModified, + }) + target_blob_client.set_blob_metadata(**metadata_kwargs) + final_target_properties = target_blob_client.get_blob_properties() + return { + "status": "succeeded", + "artifact_path": final_blob_name, + "source_bytes": source_size, + "artifact_bytes": artifact_bytes, + "target_etag": _safe_text( + _get_backup_blob_property(final_target_properties, "etag") + ), + "retry_attempt_count": retry_attempt_count, + "throttle_count": throttle_count, + "throttle_delay_seconds": round(throttle_delay_seconds, 3), + "elapsed_seconds": round(time.perf_counter() - started_at, 3), + "transfer_format": transfer_format, + } + except (DataManagementBackupCanceledError, DataManagementBackupLeaseLostError): + raise + except Exception as exc: + status_code = _get_backup_exception_status_code(exc) + if status_code in {429, 503}: + throttle_count += 1 + if not _is_retryable_backup_blob_error(exc) or attempt >= attempts: + return { + "status": "failed", + "artifact_path": final_blob_name, + "source_bytes": source_size, + "artifact_bytes": 0, + "retry_attempt_count": retry_attempt_count, + "throttle_count": throttle_count, + "throttle_delay_seconds": round(throttle_delay_seconds, 3), + "elapsed_seconds": round(time.perf_counter() - started_at, 3), + "transfer_format": transfer_format, + "failure_summary": ( + _sanitize_data_management_backup_text(str(exc)) or + "Source blob transfer failed." + ), + } + retry_attempt_count += 1 + delay = _get_backup_retry_delay(exc, attempt) + throttle_delay_seconds += delay + if cancel_event is not None: + if cancel_event.wait(delay): + raise DataManagementBackupCanceledError( + "Source blob transfer stopped during retry backoff." + ) from exc + else: + time.sleep(delay) + raise RuntimeError("Source blob transfer exhausted its retry budget.") + + +def _copy_source_blob( + target_container_client, + source_blob_client, + target_blob_name, + fernet=None, + backup_job=None, + backup_settings=None, + heartbeat_message="", +): + """Compatibility wrapper for bounded source-blob transfer callers.""" + del heartbeat_message + source_properties = source_blob_client.get_blob_properties() + source_item = _build_backup_blob_source_item("", target_blob_name, source_properties) + result = _transfer_backup_source_blob( + target_container_client, + source_blob_client, + target_blob_name, + source_item, + fernet=fernet, + backup_job=backup_job, + chunk_size_bytes=_get_backup_blob_chunk_size_bytes( backup_settings or {}, - heartbeat_message or "Uploading source blob backup artifact", - upload, + (backup_job or {}).get("backup_plan") if isinstance(backup_job, dict) else None, + ), + retry_count=_get_backup_blob_retry_count( + backup_settings or {}, + (backup_job or {}).get("backup_plan") if isinstance(backup_job, dict) else None, + ), + ) + if result.get("status") == "failed": + raise RuntimeError( + result.get("failure_summary") or "Source blob transfer failed." ) - else: - upload() - return final_blob_name, len(blob_bytes) + return result["artifact_path"], result["artifact_bytes"] def _export_source_blob_artifacts(container_client, base_prefix, settings, fernet=None): @@ -10381,7 +10984,7 @@ def _export_source_blob_artifacts(container_client, base_prefix, settings, ferne artifact["bytes"] += uploaded_bytes except Exception as exc: artifact["status"] = "warning" - artifact["warning"] = str(exc) + artifact["warning"] = _sanitize_data_management_backup_text(str(exc)) artifacts.append(artifact) return artifacts @@ -10634,6 +11237,14 @@ def _sanitize_backup_manifest_entry(entry): "artifact_checkpoint_id", "artifact_path", "bytes", + "source_etag", + "source_last_modified", + "source_bytes", + "artifact_bytes", + "target_etag", + "retry_attempt_count", + "throttle_count", + "transfer_format", "recorded_at", "failure_summary", "skip_summary", @@ -10734,25 +11345,35 @@ def _update_backup_state_totals(state): "skipped_count": 0, "failed_count": 0, "bytes": 0, + "artifact_bytes": 0, "checkpoint_count": 0, "request_units": 0.0, "retry_attempt_count": 0, "throttle_count": 0, } current_container = "" + current_file = "" current_page = 0 started_at_values = [] for resource in (state.get("resources") or {}).values(): metrics = _backup_resource_metrics(resource) totals["processed_count"] += _safe_int(metrics.get("processed_count"), default=0, minimum=0) totals["exported_count"] += _safe_int( - metrics.get("item_count", metrics.get("exported_count")), + metrics.get( + "item_count", + metrics.get("exported_count", metrics.get("blob_count")), + ), default=0, minimum=0, ) totals["skipped_count"] += _safe_int(metrics.get("skipped_count"), default=0, minimum=0) totals["failed_count"] += _safe_int(metrics.get("failed_count"), default=0, minimum=0) totals["bytes"] += _safe_int(metrics.get("bytes"), default=0, minimum=0) + totals["artifact_bytes"] += _safe_int( + metrics.get("artifact_bytes"), + default=0, + minimum=0, + ) totals["checkpoint_count"] += _safe_int(metrics.get("checkpoint_count"), default=0, minimum=0) try: totals["request_units"] += max(0.0, float(metrics.get("request_units") or 0.0)) @@ -10780,6 +11401,7 @@ def _update_backup_state_totals(state): metrics.get("current_container") or metrics.get("current_index") ) current_page = _safe_int(metrics.get("current_page"), default=0, minimum=0) + current_file = _safe_text(metrics.get("current_file")) started_at = min(started_at_values) if started_at_values else None elapsed_seconds = max( 0.001, @@ -10788,19 +11410,23 @@ def _update_backup_state_totals(state): totals["request_units"] = round(totals["request_units"], 3) totals["elapsed_seconds"] = round(elapsed_seconds, 3) totals["records_per_second"] = round(totals["processed_count"] / elapsed_seconds, 3) + totals["bytes_per_second"] = round(totals["bytes"] / elapsed_seconds, 3) totals["request_units_per_second"] = round(totals["request_units"] / elapsed_seconds, 3) state["totals"] = totals state["telemetry"] = { "current_container": current_container, + "current_file": current_file, "current_page": current_page, "checkpoint_position": totals["checkpoint_count"], "records_processed": totals["processed_count"], "bytes": totals["bytes"], + "artifact_bytes": totals["artifact_bytes"], "request_units": totals["request_units"], "retries": totals["retry_attempt_count"], "throttles": totals["throttle_count"], "elapsed_seconds": totals["elapsed_seconds"], "records_per_second": totals["records_per_second"], + "bytes_per_second": totals["bytes_per_second"], "request_units_per_second": totals["request_units_per_second"], } return totals @@ -12417,6 +13043,9 @@ def export_batch(source_batch): resource_name, source_item, "skipped", + source_etag=source_item.get("source_etag"), + source_last_modified=source_item.get("last_modified"), + source_bytes=source_item.get("source_size"), skip_summary=skip_summary, )) _queue_backup_latest_item_state_update( @@ -12996,7 +13625,7 @@ def _execute_backup_source_blob_resource( source_container_client, source_container_name, ): - """Copy source blobs one at a time so each successful blob is durable on restart.""" + """Transfer source blobs concurrently while one coordinator commits checkpoints.""" resource_name = f"source_blobs:{source_container_name}" _sync_backup_latest_item_state_from_manifest(job, resource_name) existing_resource = get_backup_resource(state, resource_name) @@ -13005,12 +13634,39 @@ def _execute_backup_source_blob_resource( resource = start_backup_resource(state, resource_name, "source_blobs") previous_progress = resource.get("progress") if isinstance(resource.get("progress"), dict) else {} previous_checkpoint = resource.get("checkpoint") if isinstance(resource.get("checkpoint"), dict) else {} - copied_count = _safe_int(previous_progress.get("blob_count"), default=0, minimum=0) - skipped_count = _safe_int(previous_progress.get("skipped_count"), default=0, minimum=0) + copied_count = 0 + reused_count = 0 + skipped_count = 0 failed_count = 0 - byte_count = _safe_int(previous_progress.get("bytes"), default=0, minimum=0) + byte_count = 0 + artifact_byte_count = 0 checkpoint_count = _safe_int(previous_progress.get("checkpoint_count"), default=0, minimum=0) + source_read_count = _safe_int( + previous_progress.get("source_read_count"), + default=0, + minimum=0, + ) + retry_attempt_count = _safe_int( + previous_progress.get("retry_attempt_count"), + default=0, + minimum=0, + ) + throttle_count = _safe_int( + previous_progress.get("throttle_count"), + default=0, + minimum=0, + ) + throttle_delay_seconds = float(previous_progress.get("throttle_delay_seconds") or 0.0) batch_number = _safe_int(previous_checkpoint.get("next_batch_number"), default=1, minimum=1) + backup_plan = job.get("backup_plan") if isinstance(job.get("backup_plan"), dict) else {} + configured_parallelism = _get_backup_blob_parallel_operations(settings, backup_plan) + active_parallelism = configured_parallelism + chunk_size_bytes = _get_backup_blob_chunk_size_bytes(settings, backup_plan) + retry_count = _get_backup_blob_retry_count(settings, backup_plan) + clean_transfer_count = 0 + current_files = set() + started_at = time.perf_counter() + cancel_event = Event() append_manifest, flush_manifest, manifest_buffer = _create_backup_manifest_writer( job.get("id"), resource_name, @@ -13018,17 +13674,31 @@ def _execute_backup_source_blob_resource( pending_latest_state_updates = [] def persist(message): - nonlocal state + nonlocal state, previous_checkpoint _assert_backup_job_lease(job) if manifest_buffer: flush_manifest() + elapsed_seconds = max(0.001, time.perf_counter() - started_at) progress = { "processed_count": copied_count + skipped_count + failed_count, "blob_count": copied_count, + "reused_count": reused_count, "skipped_count": skipped_count, "failed_count": failed_count, "bytes": byte_count, + "artifact_bytes": artifact_byte_count, "checkpoint_count": checkpoint_count, + "source_read_count": source_read_count, + "retry_attempt_count": retry_attempt_count, + "throttle_count": throttle_count, + "throttle_delay_seconds": round(throttle_delay_seconds, 3), + "parallel_operations": configured_parallelism, + "active_parallel_operations": active_parallelism, + "chunk_size_bytes": chunk_size_bytes, + "in_flight_count": len(current_files), + "current_file": sorted(current_files)[0] if current_files else "", + "elapsed_seconds": round(elapsed_seconds, 3), + "bytes_per_second": round(byte_count / elapsed_seconds, 3), } state = _persist_backup_checkpoint( job, @@ -13046,172 +13716,312 @@ def persist(message): pending_latest_state_updates, ) - try: - blob_properties_iterable = source_container_client.list_blobs() - for blob_properties in blob_properties_iterable: - artifact_path = "" - blob_name = _safe_text(getattr(blob_properties, "name", None) or blob_properties.get("name")) + def record_skipped(source_item, latest_state): + nonlocal skipped_count + skipped_count += 1 + skip_summary = "Latest successful backup state matches the source version." + _append_backup_state_summary(state, "skipped_items", { + "service": "source_blobs", + "resource_name": resource_name, + "source_identity": source_item["source_identity"], + "skip_summary": skip_summary, + }) + append_manifest(_build_backup_manifest_entry( + job, + "source_blobs", + resource_name, + source_item, + "skipped", + skip_summary=skip_summary, + )) + _queue_backup_latest_item_state_update( + pending_latest_state_updates, + source_item, + "skipped", + checkpoint_id=_safe_text((latest_state or {}).get("artifact_checkpoint_id")), + artifact_path=_safe_text((latest_state or {}).get("artifact_path")), + skip_summary=skip_summary, + ) + + def next_transfer_candidate(source_iterator): + nonlocal source_read_count + while True: + blob_properties = next(source_iterator) + blob_name = _safe_text(_get_backup_blob_property(blob_properties, "name")) if not blob_name: continue - last_modified = getattr(blob_properties, "last_modified", None) - if last_modified is None and isinstance(blob_properties, dict): - last_modified = blob_properties.get("last_modified") + source_read_count += 1 + source_item = _build_backup_blob_source_item( + source_container_name, + blob_name, + blob_properties, + ) if not _is_backup_value_within_cutoff( - last_modified, - (job.get("backup_plan") or {}).get("source_cutoff_at"), + source_item.get("last_modified"), + backup_plan.get("source_cutoff_at"), ): continue - blob_etag = getattr(blob_properties, "etag", None) - if blob_etag is None and isinstance(blob_properties, dict): - blob_etag = blob_properties.get("etag") - blob_size = getattr(blob_properties, "size", None) - if blob_size is None and isinstance(blob_properties, dict): - blob_size = blob_properties.get("size") - source_item = { - "source_identity": _build_backup_source_identity( - "source_blobs", - source_container_name, - blob_name, - ), - "source_version": _build_backup_source_version({ - "etag": blob_etag, - "last_modified": last_modified, - "size": blob_size, - }), - } latest_state = _read_backup_latest_item_state( _get_backup_source_scope(job), - _build_backup_lineage_id(job.get("backup_plan")), + _build_backup_lineage_id(backup_plan), "source_blobs", resource_name, source_item["source_identity"], ) if not _is_backup_item_due_for_export( - job.get("backup_plan"), + backup_plan, latest_state, source_item["source_version"], backup_job_id=job.get("id"), ): - skipped_count += 1 - skip_summary = "Latest successful backup state matches the source version." - _append_backup_state_summary(state, "skipped_items", { - "service": "source_blobs", - "resource_name": resource_name, - "source_identity": source_item["source_identity"], - "skip_summary": skip_summary, - }) - append_manifest(_build_backup_manifest_entry( - job, - "source_blobs", - resource_name, - source_item, - "skipped", - skip_summary=skip_summary, - )) - _queue_backup_latest_item_state_update( - pending_latest_state_updates, - source_item, - "skipped", - checkpoint_id=_safe_text((latest_state or {}).get("artifact_checkpoint_id")), - artifact_path=_safe_text((latest_state or {}).get("artifact_path")), - skip_summary=skip_summary, - ) + if ( + _safe_text((latest_state or {}).get("job_id")) == + _safe_text(job.get("id")) and + _safe_text((latest_state or {}).get("status")).lower() == "succeeded" + ): + return { + "blob_name": blob_name, + "source_item": source_item, + "target_blob_name": ( + f"{base_prefix}/source_blobs/{source_container_name}/{blob_name}" + ), + } + record_skipped(source_item, latest_state) if len(manifest_buffer) >= DATA_MANAGEMENT_BACKUP_MANIFEST_BATCH_SIZE: persist(f"Checkpointed unchanged source blobs for {source_container_name}") continue + return { + "blob_name": blob_name, + "source_item": source_item, + "target_blob_name": ( + f"{base_prefix}/source_blobs/{source_container_name}/{blob_name}" + ), + } - checkpoint_id = _build_backup_checkpoint_id(job, resource_name, batch_number) - target_blob_name = f"{base_prefix}/source_blobs/{source_container_name}/{blob_name}" - _assert_backup_job_lease(job) + def record_transfer_result(candidate, transfer_result): + nonlocal copied_count, reused_count, failed_count, byte_count + nonlocal artifact_byte_count, checkpoint_count, batch_number + nonlocal previous_checkpoint, retry_attempt_count, throttle_count + nonlocal throttle_delay_seconds, active_parallelism, clean_transfer_count + source_item = candidate["source_item"] + artifact_path = _safe_text(transfer_result.get("artifact_path")) + result_status = _safe_text(transfer_result.get("status")).lower() + retry_attempt_count += _safe_int( + transfer_result.get("retry_attempt_count"), + default=0, + minimum=0, + ) + result_throttles = _safe_int( + transfer_result.get("throttle_count"), + default=0, + minimum=0, + ) + throttle_count += result_throttles + throttle_delay_seconds += float( + transfer_result.get("throttle_delay_seconds") or 0.0 + ) + if result_throttles: + active_parallelism = max(1, active_parallelism // 2) + clean_transfer_count = 0 + else: + clean_transfer_count += 1 + if ( + active_parallelism < configured_parallelism and + clean_transfer_count >= DATA_MANAGEMENT_BLOB_BACKUP_CLEAN_TRANSFER_RECOVERY_COUNT + ): + active_parallelism += 1 + clean_transfer_count = 0 + + checkpoint_id = _build_backup_checkpoint_id(job, resource_name, batch_number) + if result_status in {"succeeded", "reused"}: + copied_count += 1 + if result_status == "reused": + reused_count += 1 + source_bytes = _safe_int( + transfer_result.get("source_bytes"), + default=0, + minimum=0, + ) + artifact_bytes = _safe_int( + transfer_result.get("artifact_bytes"), + default=0, + minimum=0, + ) + byte_count += source_bytes + artifact_byte_count += artifact_bytes + append_manifest(_build_backup_manifest_entry( + job, + "source_blobs", + resource_name, + source_item, + "succeeded", + artifact_checkpoint_id=checkpoint_id, + artifact_path=artifact_path, + bytes=source_bytes, + source_etag=source_item.get("source_etag"), + source_last_modified=source_item.get("last_modified"), + source_bytes=source_bytes, + artifact_bytes=artifact_bytes, + target_etag=transfer_result.get("target_etag"), + retry_attempt_count=transfer_result.get("retry_attempt_count"), + throttle_count=transfer_result.get("throttle_count"), + transfer_format=transfer_result.get("transfer_format"), + )) + _queue_backup_latest_item_state_update( + pending_latest_state_updates, + source_item, + "succeeded", + checkpoint_id=checkpoint_id, + artifact_path=artifact_path, + ) + else: + failed_count += 1 + failure_summary = ( + _sanitize_data_management_backup_text( + transfer_result.get("failure_summary") + ) or + "Source blob transfer failed." + ) + _append_backup_state_summary(state, "failed_items", { + "service": "source_blobs", + "resource_name": resource_name, + "source_identity": source_item["source_identity"], + "failure_summary": failure_summary, + }) + append_manifest(_build_backup_manifest_entry( + job, + "source_blobs", + resource_name, + source_item, + "failed", + artifact_path=artifact_path, + source_etag=source_item.get("source_etag"), + source_last_modified=source_item.get("last_modified"), + source_bytes=source_item.get("source_size"), + retry_attempt_count=transfer_result.get("retry_attempt_count"), + throttle_count=transfer_result.get("throttle_count"), + transfer_format=transfer_result.get("transfer_format"), + failure_summary=failure_summary, + )) + _queue_backup_latest_item_state_update( + pending_latest_state_updates, + source_item, + "failed", + failure_summary=failure_summary, + ) + checkpoint_count += 1 + previous_checkpoint = _build_backup_checkpoint_summary( + previous_checkpoint, + batch_number, + artifact_path, + ) + batch_number += 1 + persist(f"Checkpointed source blob backup for {source_container_name}") + + try: + source_iterator = iter(source_container_client.list_blobs()) + source_exhausted = False + in_flight = {} + last_heartbeat_at = time.monotonic() + with ThreadPoolExecutor(max_workers=configured_parallelism) as executor: try: - source_blob_client = source_container_client.get_blob_client(blob_name) - artifact_path, uploaded_bytes = _copy_source_blob( - container_client, - source_blob_client, - target_blob_name, - fernet=fernet, - backup_job=job, - backup_settings=settings, - heartbeat_message=f"Copying source blob backup for {source_container_name}", - ) - _assert_backup_job_lease(job) - copied_count += 1 - byte_count += _safe_int(uploaded_bytes, default=0, minimum=0) - append_manifest(_build_backup_manifest_entry( - job, - "source_blobs", - resource_name, - source_item, - "succeeded", - artifact_checkpoint_id=checkpoint_id, - artifact_path=artifact_path, - bytes=uploaded_bytes, - )) - _queue_backup_latest_item_state_update( - pending_latest_state_updates, - source_item, - "succeeded", - checkpoint_id=checkpoint_id, - artifact_path=artifact_path, - ) + while not source_exhausted or in_flight: + while not source_exhausted and len(in_flight) < active_parallelism: + _assert_backup_job_lease(job) + try: + candidate = next_transfer_candidate(source_iterator) + except StopIteration: + source_exhausted = True + break + current_files.add(candidate["source_item"]["source_identity"]) + future = executor.submit( + _run_backup_source_blob_transfer, + container_client, + source_container_client.get_blob_client(candidate["blob_name"]), + candidate["target_blob_name"], + candidate["source_item"], + fernet, + job, + chunk_size_bytes, + retry_count, + cancel_event, + ) + in_flight[future] = candidate + if not in_flight: + continue + completed, _ = wait( + tuple(in_flight), + timeout=DATA_MANAGEMENT_BACKUP_HEARTBEAT_POLL_SECONDS, + return_when=FIRST_COMPLETED, + ) + if not completed: + if ( + time.monotonic() - last_heartbeat_at >= + DATA_MANAGEMENT_BACKUP_HEARTBEAT_INTERVAL_SECONDS + ): + _persist_backup_heartbeat( + job, + settings, + f"Transferring source blobs for {source_container_name}", + ) + last_heartbeat_at = time.monotonic() + else: + _assert_backup_job_lease(job) + continue + for future in completed: + candidate = in_flight.pop(future) + current_files.discard(candidate["source_item"]["source_identity"]) + transfer_result = future.result() + record_transfer_result(candidate, transfer_result) except (DataManagementBackupCanceledError, DataManagementBackupLeaseLostError): + cancel_event.set() raise - except Exception as exc: - failed_count += 1 - failure_summary = _safe_text(str(exc))[:500] or "Source blob copy failed." - _assert_backup_job_lease(job) - _append_backup_state_summary(state, "failed_items", { - "service": "source_blobs", - "resource_name": resource_name, - "source_identity": source_item["source_identity"], - "failure_summary": failure_summary, - }) - append_manifest(_build_backup_manifest_entry( - job, - "source_blobs", - resource_name, - source_item, - "failed", - failure_summary=failure_summary, - )) - _queue_backup_latest_item_state_update( - pending_latest_state_updates, - source_item, - "failed", - failure_summary=failure_summary, - ) - checkpoint_count += 1 - previous_checkpoint = _build_backup_checkpoint_summary( - previous_checkpoint, - batch_number, - artifact_path, - ) - batch_number += 1 - persist(f"Checkpointed source blob backup for {source_container_name}") except (DataManagementBackupCanceledError, DataManagementBackupLeaseLostError): + cancel_event.set() raise except Exception as exc: + cancel_event.set() failed_count += 1 + failure_summary = ( + _sanitize_data_management_backup_text(str(exc)) or + "Source blob enumeration failed." + ) _append_backup_state_summary(state, "failed_items", { "service": "source_blobs", "resource_name": resource_name, - "failure_summary": _safe_text(str(exc))[:500], + "failure_summary": failure_summary, }) if manifest_buffer or pending_latest_state_updates: persist(f"Finalized source blob checkpoints for {source_container_name}") + elapsed_seconds = max(0.001, time.perf_counter() - started_at) result = { "name": source_container_name, "type": "source_blob_container", "container_name": source_container_name, "status": "warning" if failed_count else "completed", "blob_count": copied_count, + "reused_count": reused_count, "skipped_count": skipped_count, "failed_count": failed_count, "bytes": byte_count, + "artifact_bytes": artifact_byte_count, "encrypted": bool(fernet), + "transfer_format": ( + DATA_MANAGEMENT_BLOB_BACKUP_ENCRYPTED_FORMAT + if fernet else DATA_MANAGEMENT_BLOB_BACKUP_RAW_FORMAT + ), "prefix": f"{base_prefix}/source_blobs/{source_container_name}/", "checkpoint_count": checkpoint_count, + "source_read_count": source_read_count, + "retry_attempt_count": retry_attempt_count, + "throttle_count": throttle_count, + "throttle_delay_seconds": round(throttle_delay_seconds, 3), + "parallel_operations": configured_parallelism, + "active_parallel_operations": active_parallelism, + "chunk_size_bytes": chunk_size_bytes, + "elapsed_seconds": round(elapsed_seconds, 3), + "bytes_per_second": round(byte_count / elapsed_seconds, 3), } if failed_count: _fail_backup_resource_checkpoint( diff --git a/application/single_app/static/js/admin/admin_data_management.js b/application/single_app/static/js/admin/admin_data_management.js index 9c835db57..0d45e4cff 100644 --- a/application/single_app/static/js/admin/admin_data_management.js +++ b/application/single_app/static/js/admin/admin_data_management.js @@ -74,6 +74,9 @@ function bindElements() { "data_management_encryption_enabled", "data_management_backup_max_parallel_operations", "data_management_backup_retry_count", + "data_management_backup_blob_max_parallel_operations", + "data_management_backup_blob_chunk_size_mib", + "data_management_backup_blob_retry_count", "data_management_backup_capacity_failure_policy", "data_management_backup_temporary_source_ru_enabled", "data-management-backup-temporary-ru-field", @@ -475,6 +478,9 @@ function populateSettings(settings) { setChecked(elements.datamanagementencryptionenabled, settings.encryption_enabled !== false); setValue(elements.datamanagementbackupmaxparalleloperations, settings.backup_max_parallel_operations || 4); setValue(elements.datamanagementbackupretrycount, settings.backup_retry_count || 5); + setValue(elements.datamanagementbackupblobmaxparalleloperations, settings.backup_blob_max_parallel_operations || 4); + setValue(elements.datamanagementbackupblobchunksizemib, settings.backup_blob_chunk_size_mib || 8); + setValue(elements.datamanagementbackupblobretrycount, settings.backup_blob_retry_count || 5); setValue(elements.datamanagementbackupcapacityfailurepolicy, settings.backup_capacity_failure_policy || "continue_without_boost"); setChecked(elements.datamanagementbackuptemporarysourceruenabled, Boolean(settings.backup_temporary_source_ru_enabled)); setValue(elements.datamanagementbackuptemporarysourceru, settings.backup_temporary_source_ru || 10000); @@ -667,6 +673,9 @@ function collectSettings() { encryption_enabled: Boolean(elements.datamanagementencryptionenabled?.checked), backup_max_parallel_operations: getNumberValue(elements.datamanagementbackupmaxparalleloperations, 4), backup_retry_count: getNumberValue(elements.datamanagementbackupretrycount, 5), + backup_blob_max_parallel_operations: getNumberValue(elements.datamanagementbackupblobmaxparalleloperations, 4), + backup_blob_chunk_size_mib: getNumberValue(elements.datamanagementbackupblobchunksizemib, 8), + backup_blob_retry_count: getNumberValue(elements.datamanagementbackupblobretrycount, 5), backup_capacity_failure_policy: getValue(elements.datamanagementbackupcapacityfailurepolicy) || "continue_without_boost", backup_temporary_source_ru_enabled: Boolean(elements.datamanagementbackuptemporarysourceruenabled?.checked), backup_temporary_source_ru: getNumberValue(elements.datamanagementbackuptemporarysourceru, 10000), diff --git a/application/single_app/templates/admin_settings.html b/application/single_app/templates/admin_settings.html index 3e8ac3bcd..1ac0942ac 100644 --- a/application/single_app/templates/admin_settings.html +++ b/application/single_app/templates/admin_settings.html @@ -5737,6 +5737,28 @@
Cosmos Backup Performance

Backups stream deterministic checkpoint batches and commit only verified work. Higher concurrency can increase source Cosmos cost and pressure.

+ +
+
+
Source Blob Backup Performance
+

Source files stream through bounded chunks and durable per-file checkpoints. Peak transfer buffering is bounded by concurrent transfers multiplied by chunk size.

+
+
+
+ + +
+
+ + +
+
+ + +
+
+
Defaults bound application transfer buffering to approximately 32 MiB, excluding Azure SDK overhead. Throttling temporarily reduces active transfers.
+
diff --git a/docs/explanation/features/DATA_MANAGEMENT_BACKUP_MIGRATION.md b/docs/explanation/features/DATA_MANAGEMENT_BACKUP_MIGRATION.md index dfcd9526b..2b9412f90 100644 --- a/docs/explanation/features/DATA_MANAGEMENT_BACKUP_MIGRATION.md +++ b/docs/explanation/features/DATA_MANAGEMENT_BACKUP_MIGRATION.md @@ -1,7 +1,7 @@ # Data Management Backup and Migration Implemented in version: **0.241.211** -Updated in version: **0.250.101** +Updated in version: **0.250.102** ## Overview @@ -67,6 +67,10 @@ Backup jobs write JSON/JSONL artifacts to the configured Azure Blob Storage cont - Cosmos DB app data for settings, users/groups/workspaces, conversations, documents, agents, actions, prompts, and workspace identities. - AI Search schemas and retrievable index documents for personal, group, and public indexes. Search documents are stored as deterministic page parts with schema fingerprints, source-window metadata, bounded part summaries, integrity status, and concurrent-write semantics in the artifact manifest. - Optional source document blob backup can be enabled from the admin UI. +- Source document blobs use bounded concurrent Azure SDK block transfer, + source/target version verification, durable per-file outcomes, adaptive + throttling, and `raw-v1` or authenticated `fernet-chunked-v1` artifacts. See + [Data Management Source Blob Backup Throughput](DATA_MANAGEMENT_BLOB_BACKUP_THROUGHPUT.md). - A manifest records artifact paths, app version, backup type, encryption status, and warnings. ### Job History and Backup Inventory @@ -119,6 +123,9 @@ For durable provenance, destination access probes, collision protection, checkpo - Source document blob backup is available only when Enhanced Citations is enabled. It defaults on when available and is disabled when Enhanced Citations is off. - Backup encryption keys can be stored in Key Vault when Key Vault secret storage is enabled. If Key Vault is unavailable, generated keys fall back to the Data Management settings document and the admin UI recommends enabling Key Vault. - Cosmos backup performance: bounded concurrent JSONL batch staging (default 4, maximum 16), retry attempts (default 5, maximum 10), and `continue_without_boost` or `fail` behavior when optional source capacity management cannot proceed. +- Source blob backup performance: concurrent file transfers (default 4, maximum + 8), bounded chunk size in MiB (default 8, range 1-16), and independent retry + attempts (default 5, maximum 10). - Temporary local/source Cosmos capacity boost: disabled by default; capped at 10,000 RU/s and restored from a durable source capacity snapshot. - Target Cosmos authentication: managed identity or account key. - Target Cosmos database name: always `SimpleChat`. @@ -158,6 +165,9 @@ For optional local/source Cosmos backup capacity management, assign the App Serv - Backup durability coverage: `functional_tests/test_data_management_backup_durability.py`. - Parallel Cosmos backup, retry/adaptive pressure, source capacity restoration/recovery, fencing, resume, and sanitized telemetry coverage: `functional_tests/test_data_management_backup_parallelism.py`. - AI Search keyset paging, >100,000 logical-result traversal, page concurrency, `429`/`503` recovery, resume, latest-state skips, schema/integrity validation, isolated index failure, and sanitized telemetry coverage: `functional_tests/test_data_management_ai_search_backup_export.py`. +- Source blob bounded transfer, encryption, resume, throttling, fencing, and + failure-isolation coverage: + `functional_tests/test_data_management_blob_backup_transfers.py`. - UI/template coverage: `ui_tests/test_admin_data_management_settings_ui.py`. - Scheduler/recovery, cancellation, retry, coordinator, provenance, and write-fence coverage remains in the focused `functional_tests/test_data_management_*` modules. - Syntax validation: `python -m py_compile` for modified backend modules and `node --check` for the admin browser module. @@ -173,5 +183,5 @@ For optional local/source Cosmos backup capacity management, assign the App Serv ## Version References -- Application version updated in `application/single_app/config.py` to `0.250.101`. +- Application version updated in `application/single_app/config.py` to `0.250.102`. - Functional and UI tests include the same implementation version. \ No newline at end of file diff --git a/docs/explanation/features/DATA_MANAGEMENT_BLOB_BACKUP_THROUGHPUT.md b/docs/explanation/features/DATA_MANAGEMENT_BLOB_BACKUP_THROUGHPUT.md new file mode 100644 index 000000000..09a0df8d5 --- /dev/null +++ b/docs/explanation/features/DATA_MANAGEMENT_BLOB_BACKUP_THROUGHPUT.md @@ -0,0 +1,197 @@ +# Data Management Source Blob Backup Throughput + +Implemented in version: **0.250.102** + +GitHub issue: [#1095](https://github.com/microsoft/simplechat/issues/1095) + +## Overview + +Data Management source-blob backups transfer Enhanced Citations files through +bounded Azure SDK blocks instead of downloading whole blobs with `readall()` or +copying every file serially. The worker preserves durable per-file outcomes, +optional backup encryption, source-version safety, cooperative cancellation, +and restart-safe target verification while allowing several files to transfer +concurrently. + +## Transport Evaluation + +| Candidate | Strengths | SimpleChat constraints | Decision | +|---|---|---|---| +| AzCopy | High aggregate throughput and mature transfer scheduling | Requires an additional platform-specific executable and deployment lifecycle. It cannot apply SimpleChat's framed Fernet backup encryption. Supplying source and destination authorization safely also varies by hosting platform. | Available in the operator benchmark harness, but not bundled or invoked by the application. | +| Blob server-side copy | Keeps unencrypted bytes off App Service and can be the fastest compatible cross-account path | The target storage service must be able to authorize the source URL. Existing managed-identity and connection-string configurations do not universally provide a delegated source URL. It cannot apply SimpleChat backup encryption. | Available in the benchmark harness when the operator supplies service-authorizable source access; not selected as the universal runtime path. | +| Azure SDK bounded block transfer | Works with the application's existing managed-identity and connection-string clients, supports conditional source reads, target verification, cancellation, durable retry, and application encryption | Bytes pass through App Service, so concurrency and chunk size must remain bounded. | Selected for encrypted and unencrypted application backups. | + +The selected path prioritizes a safe transport that works for every currently +supported SimpleChat authentication and encryption mode. The benchmark harness +keeps faster environment-specific alternatives measurable without silently +weakening encryption or requiring source SAS delegation in production. + +## Technical Specifications + +### Bounded Transfer Pipeline + +- The immutable job plan records concurrent blob transfers, chunk size, retry + attempts, and adaptive recovery behavior. +- Defaults are four file transfers with 8 MiB source chunks. Admin bounds are + one to eight transfers and 1 to 16 MiB chunks, so default application + buffering is approximately 32 MiB plus encryption and Azure SDK overhead. +- Each file uses ETag-conditional range downloads and deterministic staged + blocks. No source file is loaded into memory as a whole. +- Worker threads perform Blob operations only. The lease-owning coordinator + writes manifests, latest-item state, job progress, and resource checkpoints. +- `408`, `409`, `412`, `429`, `5xx`, Azure transport failures, and timeouts use + bounded exponential retry with jitter and service `Retry-After` guidance. +- `429` and `503` responses reduce active file concurrency. Three clean + transfers restore one slot until the immutable limit is reached. + +### Encryption Format + +Encrypted source artifacts use `fernet-chunked-v1`: + +1. The first block starts with the ASCII magic value `SCBF1` and a newline. +2. Every frame contains a four-byte big-endian Fernet token length. +3. The authenticated token payload contains a SHA-256 binding to the source + version, the chunk index, total chunk count, a final-chunk flag, and the + plaintext chunk. + +This format preserves bounded memory and authenticates content, source binding, +ordering, completeness, and finality. Empty encrypted files contain one +authenticated empty frame. Unencrypted artifacts use `raw-v1`. + +### Durable Resume and Fencing + +- Target metadata records pending or succeeded state, job and attempt identity, + lease generation, source version and length, and transfer format. +- Block IDs include a transfer-generation identity so concurrent attempts + cannot mix staged blocks from different source versions. +- Final commit uses target ETag or missing-target conditions. A stale attempt + cannot overwrite a newer commit, and seeing a newer lease generation stops + the stale worker. +- A succeeded target is reused only when job, source version, source length, + format, and generation rules verify. Uploaded-but-not-yet-checkpointed work is + therefore recovered without a duplicate target artifact. +- Each durable manifest outcome records hashed source identity, source ETag, + Last-Modified value, source and artifact bytes, target path and ETag, status, + retries, throttles, transfer format, job attempt, and lease generation. +- One unreadable or exhausted file becomes a sanitized failed outcome while + independent files continue. Enumeration failure remains resource-fatal + because a complete manifest cannot then be established. + +### Progress + +Per-container and aggregate progress includes source reads, completed/reused, +skipped and failed files, source and artifact bytes, retries, throttles, retry +delay, configured and active concurrency, chunk size, in-flight count, elapsed +time, bytes per second, hashed current-file identity, and last durable +checkpoint. Public progress never includes source content, account keys, +connection strings, SAS query strings, or raw token-bearing provider errors. + +## Configuration and Usage + +Open **Admin Settings > Data Management > Source Blob Backup Performance**: + +- **Concurrent blob transfers**: 1-8; default 4. +- **Transfer chunk size (MiB)**: 1-16; default 8. +- **Blob retry attempts**: 1-10; default 5. + +Settings are captured when a job is queued. Retry/Resume uses that immutable +plan rather than current admin values. + +## Reproducible Azure Benchmark + +The operator-run harness is +`functional_tests/benchmark_data_management_blob_backup.py`. It never writes +container URLs, endpoints, SAS values, source names, or exception messages to +its report. Run it only in an approved Azure environment; benchmark target +artifacts remain under a unique `simplechat-blob-benchmark/` prefix for operator +verification and cleanup. + +### Managed Identity SDK Comparison + +Assign the executing identity source Blob Data Reader and target Blob Data +Contributor, then set container URLs without SAS query strings: + +```powershell +$env:SIMPLECHAT_BLOB_BENCHMARK_AUTHENTICATION = "managed_identity" +$env:SIMPLECHAT_BLOB_BENCHMARK_SOURCE_CONTAINER_URL = "https://.blob.core.windows.net/" +$env:SIMPLECHAT_BLOB_BENCHMARK_TARGET_CONTAINER_URL = "https://.blob.core.windows.net/" +$env:SIMPLECHAT_BLOB_BENCHMARK_ENVIRONMENT = "approved-azure-test" +python functional_tests\benchmark_data_management_blob_backup.py --candidate sdk --parallelism 4 --chunk-size-mib 8 --output blob-benchmark-sdk.json +python functional_tests\benchmark_data_management_blob_backup.py --candidate sdk --parallelism 4 --chunk-size-mib 8 --encrypted --output blob-benchmark-sdk-encrypted.json +``` + +The SDK run records both the serial baseline and bounded parallel result. +Every completed candidate reports source bytes, artifact bytes, blob count, +elapsed time, and bytes per second. Server-side copy and AzCopy additionally +verify target count and length before the harness records completion. + +### Optional Candidate Comparison + +Server-side copy requires source authorization that the target storage service +can use. Configure SAS values only through the secure process environment and +never commit them: + +```powershell +$env:SIMPLECHAT_BLOB_BENCHMARK_AUTHENTICATION = "sas" +$env:SIMPLECHAT_BLOB_BENCHMARK_SOURCE_CONTAINER_URL = "https://.blob.core.windows.net/?" +$env:SIMPLECHAT_BLOB_BENCHMARK_TARGET_CONTAINER_URL = "https://.blob.core.windows.net/?" +python functional_tests\benchmark_data_management_blob_backup.py --candidate server-copy --parallelism 4 --chunk-size-mib 8 --output blob-benchmark-server-copy.json +``` + +Encrypted runs intentionally mark server-side copy and AzCopy incompatible +because neither candidate can apply the SimpleChat encryption format. + +AzCopy benchmark URLs must not contain SAS query strings because command +arguments are visible to local process inspection. Configure AzCopy managed +identity login for the approved host and use query-free container URLs: + +```powershell +$env:SIMPLECHAT_BLOB_BENCHMARK_AUTHENTICATION = "managed_identity" +$env:SIMPLECHAT_BLOB_BENCHMARK_SOURCE_CONTAINER_URL = "https://.blob.core.windows.net/" +$env:SIMPLECHAT_BLOB_BENCHMARK_TARGET_CONTAINER_URL = "https://.blob.core.windows.net/" +$env:SIMPLECHAT_BLOB_BENCHMARK_AZCOPY = "C:\approved-tools\azcopy.exe" +$env:AZCOPY_AUTO_LOGIN_TYPE = "MSI" +python functional_tests\benchmark_data_management_blob_backup.py --candidate azcopy --output blob-benchmark-azcopy.json +``` + +### Representative 50 GB Evidence + +The default harness threshold is 50 GiB. It refuses smaller source sets unless +`--allow-smaller` is supplied for harness validation. Record these environment +facts alongside the generated secret-free JSON: + +- App Service or runner SKU, region, operating system, and Python version. +- Source and target account regions, redundancy type, network path, and + private-endpoint configuration. +- Blob count and size distribution. +- Authentication mode, encryption mode, parallelism, and chunk size. +- Serial and bounded elapsed time and bytes per second. +- Any observed storage throttling. + +Throughput is environment-specific and is not a universal SimpleChat guarantee. +This repository does not contain fabricated 50 GiB results. Release evidence +must attach the JSON generated in the approved operator environment and show a +material elapsed-time improvement over that environment's serial baseline. + +## Testing and Validation + +- `functional_tests/test_data_management_blob_backup_transfers.py` +- `functional_tests/test_data_management_backup_durability.py` +- `functional_tests/test_data_management_security_patterns.py` +- `ui_tests/test_admin_data_management_settings_ui.py` + +Focused coverage validates chunk bounds, concurrent file limits, authenticated +encrypted framing, verified target reuse, source mutation, target-generation +fencing, Retry-After pressure, adaptive concurrency, isolated setup and transfer +failures, cancellation, durable manifests, and secret-safe failure summaries. + +## Limitations + +- The application does not bundle AzCopy or generate source SAS credentials. +- Server-side copy remains an operator benchmark candidate, not an automatic + production fallback. +- The application supports files up to the configured 50,000-block limit at the + maximum 16 MiB chunk size. A larger individual file is failed explicitly + rather than exceeding the configured memory bound. +- Live 50 GiB results require an operator-provided Azure environment and are not + produced by local functional tests. diff --git a/docs/explanation/release_notes.md b/docs/explanation/release_notes.md index 12a486646..284138567 100644 --- a/docs/explanation/release_notes.md +++ b/docs/explanation/release_notes.md @@ -6,6 +6,11 @@ For feature-focused and fix-focused drill-downs by version, see [Features by Ver #### New Features +* **High-Throughput Resumable Source Blob Backups** + * Source document backups now stream bounded Azure SDK blocks with configurable file concurrency and chunk size instead of buffering complete blobs or copying files serially. + * Added durable per-file verification and resume, source/target generation fencing, adaptive Retry-After-aware throttling, isolated file failures, authenticated chunked encryption, throughput telemetry, and a reproducible AzCopy/server-copy/SDK benchmark harness. + * (Ref: Closes #1095, `functions_data_management.py`, `test_data_management_blob_backup_transfers.py`, `benchmark_data_management_blob_backup.py`, `DATA_MANAGEMENT_BLOB_BACKUP_THROUGHPUT.md`) + * **Desktop Conversation Notifications** * Administrators can enable operating system notifications for completed AI responses, and users can manage their own preference from Profile. * Notifications appear only while SimpleChat is open in a hidden or unfocused tab, show the application and conversation titles without response content, and focus the existing tab when selected. diff --git a/functional_tests/benchmark_data_management_blob_backup.py b/functional_tests/benchmark_data_management_blob_backup.py new file mode 100644 index 000000000..2a1cd3b4c --- /dev/null +++ b/functional_tests/benchmark_data_management_blob_backup.py @@ -0,0 +1,509 @@ +#!/usr/bin/env python3 +# benchmark_data_management_blob_backup.py +""" +Azure benchmark harness for source-blob backup transport candidates. +Version: 0.250.102 +Implemented in: 0.250.102 + +This operator-run harness compares a serial SDK baseline, bounded parallel SDK +block transfer, Blob server-side copy, and AzCopy without printing or persisting +credentials, SAS query strings, endpoints, or source content. +""" + +import argparse +import base64 +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import datetime, timezone +import hashlib +import io +import json +import math +import os +from pathlib import Path +import shutil +import struct +import subprocess +import time +from urllib.parse import urlsplit, urlunsplit +import uuid + +from azure.core import MatchConditions +from azure.identity import DefaultAzureCredential +from azure.storage.blob import BlobBlock, ContainerClient, ContentSettings +from cryptography.fernet import Fernet + + +MIB = 1024 * 1024 +DEFAULT_EXPECTED_BYTES = 50 * 1024 * 1024 * 1024 +ENCRYPTED_MAGIC = b"SCBF1\n" +ENCRYPTED_HEADER_SIZE = 49 +SERVICE_TIMEOUT_SECONDS = 120 + + +def parse_args(): + parser = argparse.ArgumentParser( + description="Benchmark SimpleChat-compatible source-blob backup transports.", + ) + parser.add_argument( + "--candidate", + choices=("all", "sdk", "server-copy", "azcopy"), + default="all", + ) + parser.add_argument("--parallelism", type=int, default=4) + parser.add_argument("--chunk-size-mib", type=int, default=8) + parser.add_argument("--expected-bytes", type=int, default=DEFAULT_EXPECTED_BYTES) + parser.add_argument("--allow-smaller", action="store_true") + parser.add_argument("--encrypted", action="store_true") + parser.add_argument("--output", type=Path) + return parser.parse_args() + + +def require_environment(name): + value = os.getenv(name, "").strip() + if not value: + raise RuntimeError(f"Required benchmark environment variable is missing: {name}") + return value + + +def build_container_client(url, authentication_type): + credential = ( + DefaultAzureCredential() + if authentication_type == "managed_identity" else + None + ) + return ContainerClient.from_container_url(url, credential=credential) + + +def append_url_path(url, suffix): + parsed = urlsplit(url) + path = f"{parsed.path.rstrip('/')}/{suffix.strip('/')}" + return urlunsplit((parsed.scheme, parsed.netloc, path, parsed.query, parsed.fragment)) + + +def block_id(run_id, block_index): + raw = f"{run_id[:24]}:{int(block_index):08d}".encode("ascii") + return base64.b64encode(raw).decode("ascii") + + +def encode_chunk(chunk, fernet, chunk_index, total_chunks, source_version): + if fernet is None: + return chunk + source_digest = hashlib.sha256(source_version.encode("utf-8")).digest() + authenticated = ( + source_digest + + struct.pack( + ">QQ?", + chunk_index, + total_chunks, + chunk_index == total_chunks - 1, + ) + + chunk + ) + token = fernet.encrypt(authenticated) + prefix = ENCRYPTED_MAGIC if chunk_index == 0 else b"" + return prefix + struct.pack(">I", len(token)) + token + + +def source_version(properties): + payload = { + "etag": str(getattr(properties, "etag", "") or ""), + "last_modified": str(getattr(properties, "last_modified", "") or ""), + "size": int(getattr(properties, "size", 0) or 0), + } + encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def read_range(source_blob, offset, length, etag): + kwargs = { + "offset": offset, + "length": length, + "max_concurrency": 1, + "timeout": SERVICE_TIMEOUT_SECONDS, + } + if etag: + kwargs.update({ + "etag": etag, + "match_condition": MatchConditions.IfNotModified, + }) + buffer = io.BytesIO() + written = source_blob.download_blob(**kwargs).readinto(buffer) + if written != length: + raise RuntimeError("Source range length changed during benchmark transfer.") + return buffer.getvalue() + + +def copy_blob_with_sdk( + source_container, + target_container, + properties, + target_prefix, + chunk_size, + fernet, + run_id, +): + blob_name = properties.name + source_blob = source_container.get_blob_client(blob_name) + target_blob = target_container.get_blob_client(f"{target_prefix}/{blob_name}") + source_size = int(properties.size or 0) + etag = str(properties.etag or "") + version = source_version(properties) + total_chunks = max(1, math.ceil(source_size / chunk_size)) + blocks = [] + artifact_bytes = 0 + + if source_size == 0: + payload = encode_chunk(b"", fernet, 0, 1, version) + target_blob.upload_blob( + payload, + overwrite=True, + content_settings=ContentSettings(content_type="application/octet-stream"), + timeout=SERVICE_TIMEOUT_SECONDS, + ) + artifact_bytes = len(payload) + else: + for chunk_index, offset in enumerate(range(0, source_size, chunk_size)): + length = min(chunk_size, source_size - offset) + chunk = read_range(source_blob, offset, length, etag) + payload = encode_chunk( + chunk, + fernet, + chunk_index, + total_chunks, + version, + ) + current_block_id = block_id(run_id, chunk_index) + target_blob.stage_block( + block_id=current_block_id, + data=payload, + timeout=SERVICE_TIMEOUT_SECONDS, + ) + blocks.append(BlobBlock(block_id=current_block_id)) + artifact_bytes += len(payload) + target_blob.commit_block_list( + blocks, + content_settings=ContentSettings(content_type="application/octet-stream"), + timeout=SERVICE_TIMEOUT_SECONDS, + ) + + target_properties = target_blob.get_blob_properties() + if int(target_properties.size or 0) != artifact_bytes: + raise RuntimeError("Committed benchmark artifact length did not verify.") + current_source = source_blob.get_blob_properties() + if etag and str(current_source.etag or "") != etag: + raise RuntimeError("Source blob changed during benchmark transfer.") + return source_size, artifact_bytes + + +def run_sdk_candidate( + source_container, + target_container, + properties, + target_prefix, + parallelism, + chunk_size, + fernet, +): + run_id = uuid.uuid4().hex + started_at = time.perf_counter() + source_bytes = 0 + artifact_bytes = 0 + with ThreadPoolExecutor(max_workers=parallelism) as executor: + futures = [ + executor.submit( + copy_blob_with_sdk, + source_container, + target_container, + item, + target_prefix, + chunk_size, + fernet, + run_id, + ) + for item in properties + ] + for future in as_completed(futures): + copied_source_bytes, copied_artifact_bytes = future.result() + source_bytes += copied_source_bytes + artifact_bytes += copied_artifact_bytes + elapsed_seconds = max(0.001, time.perf_counter() - started_at) + return { + "status": "completed", + "blob_count": len(properties), + "source_bytes": source_bytes, + "artifact_bytes": artifact_bytes, + "elapsed_seconds": round(elapsed_seconds, 3), + "bytes_per_second": round(source_bytes / elapsed_seconds, 3), + "parallelism": parallelism, + "chunk_size_mib": chunk_size // MIB, + } + + +def copy_blob_server_side(source_container, target_container, properties, target_prefix): + source_blob = source_container.get_blob_client(properties.name) + target_blob = target_container.get_blob_client(f"{target_prefix}/{properties.name}") + response = target_blob.start_copy_from_url(source_blob.url) + copy_id = response.get("copy_id") if isinstance(response, dict) else None + while True: + target_properties = target_blob.get_blob_properties() + copy = getattr(target_properties, "copy", None) + status = str(getattr(copy, "status", "") or "").lower() + if status == "success": + break + if status in {"failed", "aborted"}: + raise RuntimeError("Blob server-side copy did not complete successfully.") + time.sleep(0.5) + if copy_id and str(getattr(copy, "id", "") or "") != str(copy_id): + raise RuntimeError("Blob server-side copy identity changed before verification.") + if int(target_properties.size or 0) != int(properties.size or 0): + raise RuntimeError("Blob server-side copy length did not verify.") + return int(properties.size or 0) + + +def run_server_copy_candidate( + source_container, + target_container, + properties, + target_prefix, + parallelism, +): + started_at = time.perf_counter() + copied_bytes = 0 + with ThreadPoolExecutor(max_workers=parallelism) as executor: + futures = [ + executor.submit( + copy_blob_server_side, + source_container, + target_container, + item, + target_prefix, + ) + for item in properties + ] + for future in as_completed(futures): + copied_bytes += future.result() + elapsed_seconds = max(0.001, time.perf_counter() - started_at) + return { + "status": "completed", + "blob_count": len(properties), + "source_bytes": copied_bytes, + "artifact_bytes": copied_bytes, + "elapsed_seconds": round(elapsed_seconds, 3), + "bytes_per_second": round(copied_bytes / elapsed_seconds, 3), + "parallelism": parallelism, + } + + +def run_azcopy_candidate( + source_url, + target_url, + target_prefix, + target_container=None, + source_properties=None, +): + if urlsplit(source_url).query or urlsplit(target_url).query: + return { + "status": "skipped", + "reason": "AzCopy SAS URLs are disabled because command arguments are process-visible.", + } + executable = os.getenv("SIMPLECHAT_BLOB_BENCHMARK_AZCOPY") or shutil.which("azcopy") + if not executable: + return { + "status": "skipped", + "reason": "AzCopy executable was not configured.", + } + started_at = time.perf_counter() + completed = subprocess.run( + [ + executable, + "copy", + source_url, + append_url_path(target_url, target_prefix), + "--recursive=true", + "--log-level=ERROR", + "--output-type=json", + ], + capture_output=True, + check=False, + text=True, + ) + elapsed_seconds = max(0.001, time.perf_counter() - started_at) + if completed.returncode != 0: + return { + "status": "failed", + "reason": f"AzCopy exited with code {completed.returncode}.", + "elapsed_seconds": round(elapsed_seconds, 3), + } + if target_container is None or source_properties is None: + return { + "status": "failed", + "reason": "AzCopy verification inputs were not supplied.", + "elapsed_seconds": round(elapsed_seconds, 3), + } + expected_blob_count = len(source_properties) + expected_bytes = sum(int(item.size or 0) for item in source_properties) + copied_properties = list( + target_container.list_blobs(name_starts_with=f"{target_prefix.strip('/')}/") + ) + copied_blob_count = len(copied_properties) + copied_bytes = sum(int(item.size or 0) for item in copied_properties) + if copied_blob_count != expected_blob_count or copied_bytes != expected_bytes: + return { + "status": "failed", + "reason": "AzCopy target count or length verification failed.", + "blob_count": copied_blob_count, + "source_bytes": expected_bytes, + "artifact_bytes": copied_bytes, + "elapsed_seconds": round(elapsed_seconds, 3), + } + return { + "status": "completed", + "blob_count": copied_blob_count, + "source_bytes": expected_bytes, + "artifact_bytes": copied_bytes, + "elapsed_seconds": round(elapsed_seconds, 3), + "bytes_per_second": round(expected_bytes / elapsed_seconds, 3), + } + + +def candidate_result(name, operation): + try: + result = operation() + return {"candidate": name, **result} + except Exception as exc: + return { + "candidate": name, + "status": "failed", + "error_type": type(exc).__name__, + } + + +def main(): + args = parse_args() + parallelism = max(1, min(8, args.parallelism)) + chunk_size_mib = max(1, min(16, args.chunk_size_mib)) + chunk_size = chunk_size_mib * MIB + authentication_type = os.getenv( + "SIMPLECHAT_BLOB_BENCHMARK_AUTHENTICATION", + "managed_identity", + ).strip().lower() + if authentication_type not in {"managed_identity", "sas"}: + raise RuntimeError("Benchmark authentication must be managed_identity or sas.") + + source_url = require_environment("SIMPLECHAT_BLOB_BENCHMARK_SOURCE_CONTAINER_URL") + target_url = require_environment("SIMPLECHAT_BLOB_BENCHMARK_TARGET_CONTAINER_URL") + source_container = build_container_client(source_url, authentication_type) + target_container = build_container_client(target_url, authentication_type) + properties = list(source_container.list_blobs()) + source_bytes = sum(int(item.size or 0) for item in properties) + if source_bytes < max(0, args.expected_bytes) and not args.allow_smaller: + raise RuntimeError( + "Benchmark source set is smaller than --expected-bytes; use --allow-smaller " + "only for harness validation." + ) + + run_prefix = ( + f"simplechat-blob-benchmark/" + f"{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')}-{uuid.uuid4().hex[:8]}" + ) + fernet = Fernet(Fernet.generate_key()) if args.encrypted else None + results = [] + if args.candidate in {"all", "sdk"}: + results.append(candidate_result( + "sdk-serial-encrypted" if fernet else "sdk-serial", + lambda: run_sdk_candidate( + source_container, + target_container, + properties, + f"{run_prefix}/sdk-serial", + 1, + chunk_size, + fernet, + ), + )) + results.append(candidate_result( + "sdk-bounded-encrypted" if fernet else "sdk-bounded", + lambda: run_sdk_candidate( + source_container, + target_container, + properties, + f"{run_prefix}/sdk-bounded", + parallelism, + chunk_size, + fernet, + ), + )) + if args.candidate in {"all", "server-copy"}: + if fernet: + results.append({ + "candidate": "server-copy", + "status": "skipped", + "reason": "Server-side copy cannot apply SimpleChat backup encryption.", + }) + elif authentication_type != "sas": + results.append({ + "candidate": "server-copy", + "status": "skipped", + "reason": "Server-side copy requires a source URL the storage service can authorize.", + }) + else: + results.append(candidate_result( + "server-copy", + lambda: run_server_copy_candidate( + source_container, + target_container, + properties, + f"{run_prefix}/server-copy", + parallelism, + ), + )) + if args.candidate in {"all", "azcopy"}: + if fernet: + results.append({ + "candidate": "azcopy", + "status": "skipped", + "reason": "AzCopy cannot apply SimpleChat backup encryption.", + }) + elif authentication_type != "managed_identity": + results.append({ + "candidate": "azcopy", + "status": "skipped", + "reason": "AzCopy requires managed identity so credentials are not placed in process arguments.", + }) + else: + results.append(candidate_result( + "azcopy", + lambda: run_azcopy_candidate( + source_url, + target_url, + f"{run_prefix}/azcopy", + target_container, + properties, + ), + )) + + report = { + "schema_version": 1, + "recorded_at": datetime.now(timezone.utc).isoformat(), + "environment": os.getenv( + "SIMPLECHAT_BLOB_BENCHMARK_ENVIRONMENT", + "operator-supplied", + )[:100], + "authentication_type": authentication_type, + "encrypted": bool(fernet), + "source_blob_count": len(properties), + "source_bytes": source_bytes, + "expected_bytes": max(0, args.expected_bytes), + "parallelism": parallelism, + "chunk_size_mib": chunk_size_mib, + "results": results, + } + serialized = json.dumps(report, indent=2, sort_keys=True) + if args.output: + args.output.write_text(serialized + "\n", encoding="utf-8") + print(serialized) + return 0 if all(result.get("status") != "failed" for result in results) else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/functional_tests/test_data_management_blob_backup_transfers.py b/functional_tests/test_data_management_blob_backup_transfers.py new file mode 100644 index 000000000..7954f59b6 --- /dev/null +++ b/functional_tests/test_data_management_blob_backup_transfers.py @@ -0,0 +1,651 @@ +#!/usr/bin/env python3 +# test_data_management_blob_backup_transfers.py +""" +Functional test for high-throughput resumable source-blob backups. +Version: 0.250.102 +Implemented in: 0.250.102 + +This test ensures source blobs use bounded block transfers, framed encryption, +verified resume, adaptive retry, isolated failures, and coordinator-owned +durable outcomes without leaking sensitive provider errors. +""" + +import copy +import importlib.util +import io +from pathlib import Path +import struct +import sys +import threading +import time +import types + +import pytest + + +REPO_ROOT = Path(__file__).resolve().parents[1] +APP_ROOT = REPO_ROOT / "application" / "single_app" +MODULE_PATH = APP_ROOT / "functions_data_management.py" +BENCHMARK_PATH = REPO_ROOT / "functional_tests" / "benchmark_data_management_blob_backup.py" +sys.path.insert(0, str(APP_ROOT)) + +from functions_data_management_backup_state import initialize_backup_state + + +class FakeBlobError(Exception): + """Expose Azure-like status and retry headers.""" + + def __init__(self, status_code, message="blob operation failed", headers=None): + super().__init__(message) + self.status_code = status_code + self.response = types.SimpleNamespace( + status_code=status_code, + headers=headers or {}, + ) + + +class FakeDownload: + """Write one bounded range into a caller-provided stream.""" + + def __init__(self, payload): + self.payload = payload + + def readinto(self, stream): + stream.write(self.payload) + return len(self.payload) + + +class FakeSourceBlobClient: + """Serve immutable ranges and optionally inject one provider failure.""" + + def __init__(self, data, etag, transient_error=None, permanent_error=None): + self.data = data + self.etag = etag + self.transient_error = transient_error + self.permanent_error = permanent_error + self.range_lengths = [] + self.download_calls = 0 + self.mutate_after_download = False + + def get_blob_properties(self): + return types.SimpleNamespace( + name="source.bin", + size=len(self.data), + etag=self.etag, + last_modified="2026-07-30T12:00:00+00:00", + ) + + def download_blob(self, **kwargs): + self.download_calls += 1 + if self.permanent_error is not None: + raise self.permanent_error + if self.transient_error is not None: + error = self.transient_error + self.transient_error = None + raise error + if kwargs.get("etag") and kwargs["etag"] != self.etag: + raise FakeBlobError(412, "source condition failed") + offset = int(kwargs.get("offset") or 0) + length = int(kwargs.get("length") or len(self.data)) + self.range_lengths.append(length) + payload = self.data[offset:offset + length] + if self.mutate_after_download and offset + length >= len(self.data): + self.etag = '"etag-mutated"' + return FakeDownload(payload) + + +class FakeSourceContainer: + """List source properties and return matching range clients.""" + + def __init__(self, blobs): + self.blobs = blobs + + def list_blobs(self): + for blob_name, blob_client in self.blobs.items(): + properties = blob_client.get_blob_properties() + properties.name = blob_name + yield properties + + def get_blob_client(self, blob_name): + return self.blobs[blob_name] + + +class FakeTargetBlobClient: + """Stage blocks and retain metadata for verification and resume.""" + + def __init__(self, container, blob_name): + self.container = container + self.blob_name = blob_name + self.uncommitted = {} + self.committed_blocks = [] + self.content = None + self.metadata = {} + self.etag_counter = 0 + + def _start_transfer(self): + self.container.start_transfer(self.blob_name) + + def _finish_transfer(self): + self.container.finish_transfer(self.blob_name) + + def get_blob_properties(self): + if self.blob_name in self.container.property_error_names: + raise FakeBlobError(500, "target lookup failed") + if self.content is None: + raise FakeBlobError(404, "missing") + return types.SimpleNamespace( + size=len(self.content), + metadata=copy.deepcopy(self.metadata), + etag=f'"target-{self.etag_counter}"', + ) + + def stage_block(self, block_id, data, **_kwargs): + self._start_transfer() + if self.container.stage_delay: + time.sleep(self.container.stage_delay) + payload = data.read() if hasattr(data, "read") else bytes(data) + self.uncommitted[block_id] = payload + + def commit_block_list(self, block_list, metadata=None, **_kwargs): + self._check_condition(_kwargs) + self.committed_blocks = [ + self.uncommitted[block.id] + for block in block_list + ] + self.content = b"".join(self.committed_blocks) + self.metadata = copy.deepcopy(metadata or {}) + self.etag_counter += 1 + + def upload_blob(self, data, metadata=None, **_kwargs): + self._check_condition(_kwargs) + self._start_transfer() + self.content = data.read() if hasattr(data, "read") else bytes(data) + self.committed_blocks = [self.content] + self.metadata = copy.deepcopy(metadata or {}) + self.etag_counter += 1 + + def set_blob_metadata(self, metadata, **_kwargs): + self._check_condition(_kwargs) + self.metadata = copy.deepcopy(metadata) + self.etag_counter += 1 + self._finish_transfer() + + def _check_condition(self, kwargs): + condition = kwargs.get("match_condition") + if condition is None: + return + condition_name = getattr(condition, "name", "") + if condition_name == "IfMissing" and self.content is not None: + raise FakeBlobError(412, "target already exists") + if condition_name == "IfNotModified": + current_etag = f'"target-{self.etag_counter}"' + if kwargs.get("etag") != current_etag: + raise FakeBlobError(412, "target changed") + + +class FakeTargetContainer: + """Track target content and maximum concurrent file transfers.""" + + def __init__(self, stage_delay=0.0, property_error_names=None): + self.clients = {} + self.stage_delay = stage_delay + self.active_names = set() + self.max_active_transfers = 0 + self.lock = threading.Lock() + self.property_error_names = set(property_error_names or []) + + def get_blob_client(self, blob_name): + self.clients.setdefault(blob_name, FakeTargetBlobClient(self, blob_name)) + return self.clients[blob_name] + + def start_transfer(self, blob_name): + with self.lock: + self.active_names.add(blob_name) + self.max_active_transfers = max( + self.max_active_transfers, + len(self.active_names), + ) + + def finish_transfer(self, blob_name): + with self.lock: + self.active_names.discard(blob_name) + + +class FakeCosmosContainer: + """Persist jobs, manifests, and latest-only state in memory.""" + + def __init__(self): + self.documents = {} + self.counter = 0 + + def create_item(self, body): + return self.upsert_item(body) + + def upsert_item(self, body): + self.counter += 1 + saved = copy.deepcopy(body) + saved["_etag"] = f"etag-{self.counter}" + partition_key = saved.get("source_scope") or saved.get("id") + self.documents[(partition_key, saved["id"])] = saved + return copy.deepcopy(saved) + + def read_item(self, item, partition_key): + document = self.documents.get((partition_key, item)) + if document is None: + raise FakeBlobError(404, "missing") + return copy.deepcopy(document) + + def query_items(self, **_kwargs): + return iter(copy.deepcopy(list(self.documents.values()))) + + +def load_module(monkeypatch): + """Load production helpers with in-memory configuration dependencies.""" + jobs = FakeCosmosContainer() + latest_state = FakeCosmosContainer() + config_module = types.ModuleType("config") + config_module.CLIENTS = {} + config_module.VERSION = "0.250.102" + config_module.cosmos_data_management_jobs_container = jobs + config_module.cosmos_data_management_job_items_container = jobs + config_module.cosmos_settings_container = jobs + config_module.cosmos_data_management_backup_item_states_container = latest_state + monkeypatch.setitem(sys.modules, "config", config_module) + + appinsights_module = types.ModuleType("functions_appinsights") + appinsights_module.log_event = lambda *_args, **_kwargs: None + monkeypatch.setitem(sys.modules, "functions_appinsights", appinsights_module) + + throughput_module = types.ModuleType("functions_cosmos_throughput") + throughput_module.CosmosThroughputError = RuntimeError + throughput_module.get_container_throughput = lambda *_args, **_kwargs: {} + throughput_module.get_database_throughput = lambda *_args, **_kwargs: {} + throughput_module.set_database_throughput = lambda *_args, **_kwargs: {} + monkeypatch.setitem(sys.modules, "functions_cosmos_throughput", throughput_module) + + module_name = "data_management_blob_backup_test_module" + spec = importlib.util.spec_from_file_location(module_name, MODULE_PATH) + module = importlib.util.module_from_spec(spec) + sys.modules[module_name] = module + assert spec and spec.loader + spec.loader.exec_module(module) + monkeypatch.delitem(sys.modules, module_name, raising=False) + return module, jobs, latest_state + + +def load_benchmark_module(): + """Load the operator harness without running its command-line entry point.""" + module_name = "data_management_blob_backup_benchmark_test_module" + spec = importlib.util.spec_from_file_location(module_name, BENCHMARK_PATH) + module = importlib.util.module_from_spec(spec) + assert spec and spec.loader + spec.loader.exec_module(module) + return module + + +def build_plan(parallel_operations=2, chunk_size_mib=1, retry_count=3): + """Build the immutable source-blob execution contract.""" + return { + "backup_type": "partial", + "source_scope": "simplechat-primary", + "source_cutoff_at": "2099-01-01T00:00:00+00:00", + "differential_mode": "latest_item_state", + "include_cosmos": False, + "include_ai_search": False, + "include_source_blobs": True, + "backup_storage_container_name": "simplechat-backups", + "backup_storage_path_prefix": "simplechat-backups", + "storage_identity": "test-storage", + "encryption_enabled": False, + "encryption_key_fingerprint": "", + "source_blob_execution": { + "max_parallel_operations": parallel_operations, + "chunk_size_mib": chunk_size_mib, + "retry_count": retry_count, + "clean_transfer_recovery_count": 3, + }, + "resource_contract": ["cosmos", "ai_search", "source_blobs"], + } + + +def build_job(module, plan): + """Create a running job accepted by the source-blob resource helper.""" + job_id = "11111111-1111-1111-1111-111111111111" + state = initialize_backup_state( + None, + job_id, + plan, + plan["source_scope"], + plan["source_cutoff_at"], + ) + return { + "id": job_id, + "operation": "backup", + "status": "running", + "backup_attempt_id": "attempt-1", + "lease_generation": 1, + "backup_plan": plan, + "backup_state": state, + } + + +def source_item(module, source_client, blob_name="source.bin"): + """Build source identity/version from the same properties used by production.""" + return module._build_backup_blob_source_item( + "documents", + blob_name, + source_client.get_blob_properties(), + ) + + +def decode_encrypted_blocks(module, fernet, blocks): + """Decode the framed Fernet block format without joining encrypted input first.""" + decoded = io.BytesIO() + expected_source_digest = None + for index, block in enumerate(blocks): + payload = block + if index == 0: + assert payload.startswith(module.DATA_MANAGEMENT_BLOB_BACKUP_ENCRYPTED_MAGIC) + payload = payload[len(module.DATA_MANAGEMENT_BLOB_BACKUP_ENCRYPTED_MAGIC):] + token_length = struct.unpack(">I", payload[:4])[0] + token = payload[4:4 + token_length] + assert len(token) == token_length + authenticated = fernet.decrypt(token) + source_digest = authenticated[:32] + chunk_index, total_chunks, is_final = struct.unpack( + ">QQ?", + authenticated[32:module.DATA_MANAGEMENT_BLOB_BACKUP_ENCRYPTED_HEADER_SIZE], + ) + expected_source_digest = expected_source_digest or source_digest + assert source_digest == expected_source_digest + assert chunk_index == index + assert total_chunks == len(blocks) + assert is_final is (index == len(blocks) - 1) + decoded.write(authenticated[module.DATA_MANAGEMENT_BLOB_BACKUP_ENCRYPTED_HEADER_SIZE:]) + return decoded.getvalue() + + +def test_transfer_chunks_and_reuses_verified_unencrypted_artifact(monkeypatch): + """Verify raw blobs never require whole-file buffering and resume avoids reads.""" + module, _jobs, _latest_state = load_module(monkeypatch) + data = (b"bounded-transfer-" * 200000) + b"tail" + source = FakeSourceBlobClient(data, '"etag-1"') + target = FakeTargetContainer() + job = build_job(module, build_plan()) + + result = module._transfer_backup_source_blob( + target, + source, + "backup/documents/source.bin", + source_item(module, source), + backup_job=job, + chunk_size_bytes=1024 * 1024, + retry_count=3, + ) + + assert result["status"] == "succeeded" + assert max(source.range_lengths) <= 1024 * 1024 + assert len(source.range_lengths) > 1 + target_blob = target.get_blob_client(result["artifact_path"]) + assert target_blob.content == data + assert target_blob.metadata["simplechatbackupstatus"] == "succeeded" + + download_calls = source.download_calls + reused = module._transfer_backup_source_blob( + target, + source, + "backup/documents/source.bin", + source_item(module, source), + backup_job=job, + chunk_size_bytes=1024 * 1024, + retry_count=3, + ) + assert reused["status"] == "reused" + assert source.download_calls == download_calls + + +def test_encrypted_transfer_uses_versioned_authenticated_frames(monkeypatch): + """Verify encrypted artifacts preserve plaintext through bounded Fernet frames.""" + module, _jobs, _latest_state = load_module(monkeypatch) + data = (b"encrypted-bounded-transfer-" * 100000) + b"tail" + source = FakeSourceBlobClient(data, '"etag-encrypted"') + target = FakeTargetContainer() + job = build_job(module, build_plan()) + fernet = module.Fernet(module.Fernet.generate_key()) + + result = module._transfer_backup_source_blob( + target, + source, + "backup/documents/encrypted.bin", + source_item(module, source, "encrypted.bin"), + fernet=fernet, + backup_job=job, + chunk_size_bytes=1024 * 1024, + retry_count=3, + ) + + assert result["transfer_format"] == "fernet-chunked-v1" + target_blob = target.get_blob_client(result["artifact_path"]) + assert decode_encrypted_blocks(module, fernet, target_blob.committed_blocks) == data + assert target_blob.metadata["simplechatbackupformat"] == "fernet-chunked-v1" + with pytest.raises(AssertionError): + decode_encrypted_blocks( + module, + fernet, + list(reversed(target_blob.committed_blocks)), + ) + + +def test_transfer_retries_throttle_and_sanitizes_terminal_failure(monkeypatch): + """Verify Retry-After pressure is counted and provider secrets never persist.""" + module, _jobs, _latest_state = load_module(monkeypatch) + monkeypatch.setattr(module, "_get_backup_retry_delay", lambda *_args: 0.001) + throttled_source = FakeSourceBlobClient( + b"retry-me", + '"etag-retry"', + transient_error=FakeBlobError( + 429, + "throttled token=must-not-leak", + {"x-ms-retry-after-ms": "1"}, + ), + ) + target = FakeTargetContainer() + job = build_job(module, build_plan()) + + result = module._transfer_backup_source_blob( + target, + throttled_source, + "backup/documents/retry.bin", + source_item(module, throttled_source, "retry.bin"), + backup_job=job, + chunk_size_bytes=1024 * 1024, + retry_count=3, + ) + assert result["status"] == "succeeded" + assert result["retry_attempt_count"] == 1 + assert result["throttle_count"] == 1 + + failed_source = FakeSourceBlobClient( + b"fail-me", + '"etag-fail"', + permanent_error=FakeBlobError(403, "denied token=must-not-leak"), + ) + failed = module._transfer_backup_source_blob( + target, + failed_source, + "backup/documents/fail.bin", + source_item(module, failed_source, "fail.bin"), + backup_job=job, + chunk_size_bytes=1024 * 1024, + retry_count=3, + ) + assert failed["status"] == "failed" + assert failed["failure_summary"] == "[redacted operational detail]" + assert "must-not-leak" not in str(failed) + + +def test_source_mutation_does_not_mark_artifact_succeeded(monkeypatch): + """Verify a changed ETag leaves a pending artifact eligible for safe retry.""" + module, _jobs, _latest_state = load_module(monkeypatch) + source = FakeSourceBlobClient(b"mutable", '"etag-before"') + source.mutate_after_download = True + target = FakeTargetContainer() + job = build_job(module, build_plan()) + + result = module._transfer_backup_source_blob( + target, + source, + "backup/documents/mutable.bin", + source_item(module, source, "mutable.bin"), + backup_job=job, + chunk_size_bytes=1024 * 1024, + retry_count=2, + ) + + assert result["status"] == "failed" + artifact = target.get_blob_client(result["artifact_path"]) + assert artifact.metadata["simplechatbackupstatus"] == "pending" + + +def test_newer_target_generation_fences_stale_attempt(monkeypatch): + """Verify an older worker cannot accept or overwrite a newer attempt artifact.""" + module, _jobs, _latest_state = load_module(monkeypatch) + source = FakeSourceBlobClient(b"fenced", '"etag-fenced"') + target = FakeTargetContainer() + plan = build_plan() + newer_job = build_job(module, plan) + newer_job["backup_attempt_id"] = "attempt-2" + newer_job["lease_generation"] = 2 + item = source_item(module, source, "fenced.bin") + succeeded = module._transfer_backup_source_blob( + target, + source, + "backup/documents/fenced.bin", + item, + backup_job=newer_job, + chunk_size_bytes=1024 * 1024, + retry_count=2, + ) + assert succeeded["status"] == "succeeded" + + stale_job = build_job(module, plan) + stale_job["backup_attempt_id"] = "attempt-1" + stale_job["lease_generation"] = 1 + with pytest.raises(module.DataManagementBackupLeaseLostError): + module._transfer_backup_source_blob( + target, + source, + "backup/documents/fenced.bin", + item, + backup_job=stale_job, + chunk_size_bytes=1024 * 1024, + retry_count=2, + ) + + +def test_resource_bounds_parallelism_and_isolates_failed_files(monkeypatch): + """Verify one failed file does not abort concurrent neighboring transfers.""" + module, jobs, _latest_state = load_module(monkeypatch) + monkeypatch.setattr(module, "_sync_backup_latest_item_state_from_manifest", lambda *_args: None) + monkeypatch.setattr(module, "_assert_backup_job_lease", lambda *_args, **_kwargs: None) + monkeypatch.setattr(module, "_persist_backup_heartbeat", lambda *_args, **_kwargs: None) + + def persist(_job, state, _settings, resource_name, progress, checkpoint, _message): + module.update_backup_resource(state, resource_name, progress, checkpoint=checkpoint) + return state + + def complete(_job, state, _settings, resource_name, result, _message): + module.complete_backup_resource(state, resource_name, result) + return state + + def fail(_job, state, _settings, resource_name, error_message, _message, result=None): + resource = module.fail_backup_resource(state, resource_name, error_message) + resource["result"] = copy.deepcopy(result or {}) + return state + + monkeypatch.setattr(module, "_persist_backup_checkpoint", persist) + monkeypatch.setattr(module, "_complete_backup_resource_checkpoint", complete) + monkeypatch.setattr(module, "_fail_backup_resource_checkpoint", fail) + monkeypatch.setattr(module, "_get_backup_retry_delay", lambda *_args: 0.001) + + plan = build_plan(parallel_operations=2) + job = build_job(module, plan) + source_container = FakeSourceContainer({ + "one.bin": FakeSourceBlobClient(b"one" * 400000, '"etag-one"'), + "two.bin": FakeSourceBlobClient( + b"two", + '"etag-two"', + transient_error=FakeBlobError(429, "pressure", {"retry-after": "0.001"}), + ), + "three.bin": FakeSourceBlobClient(b"three" * 300000, '"etag-three"'), + "failed.bin": FakeSourceBlobClient(b"failed", '"etag-failed"'), + }) + target = FakeTargetContainer( + stage_delay=0.01, + property_error_names={"simplechat-backups/job/source_blobs/documents/failed.bin"}, + ) + + result = module._execute_backup_source_blob_resource( + job, + job["backup_state"], + {}, + target, + "simplechat-backups/job", + None, + source_container, + "documents", + ) + + assert result["blob_count"] == 3 + assert result["failed_count"] == 1 + assert result["throttle_count"] == 1 + assert 2 <= target.max_active_transfers <= 2 + manifests = [ + document + for document in jobs.documents.values() + if document.get("type") == module.DATA_MANAGEMENT_BACKUP_MANIFEST_BATCH_TYPE + ] + outcomes = [ + entry["status"] + for manifest in manifests + for entry in manifest["entries"] + ] + assert outcomes.count("succeeded") == 3 + assert outcomes.count("failed") == 1 + + +def test_transfer_observes_cancellation_before_remote_work(monkeypatch): + """Verify cooperative cancellation prevents any source or target operation.""" + module, _jobs, _latest_state = load_module(monkeypatch) + source = FakeSourceBlobClient(b"cancel", '"etag-cancel"') + target = FakeTargetContainer() + job = build_job(module, build_plan()) + cancel_event = threading.Event() + cancel_event.set() + + with pytest.raises(module.DataManagementBackupCanceledError): + module._transfer_backup_source_blob( + target, + source, + "backup/documents/cancel.bin", + source_item(module, source, "cancel.bin"), + backup_job=job, + chunk_size_bytes=1024 * 1024, + retry_count=3, + cancel_event=cancel_event, + ) + assert source.download_calls == 0 + + +def test_benchmark_never_places_sas_values_in_azcopy_arguments(): + """Verify process-visible AzCopy URLs reject SAS-bearing candidates.""" + benchmark = load_benchmark_module() + result = benchmark.run_azcopy_candidate( + "https://source.example/container?sig=source-secret", + "https://target.example/container?sig=target-secret", + "benchmark/run", + ) + assert result["status"] == "skipped" + assert "process-visible" in result["reason"] + assert "source-secret" not in str(result) + assert "target-secret" not in str(result) diff --git a/functional_tests/test_data_management_security_patterns.py b/functional_tests/test_data_management_security_patterns.py index a82e663cb..126b821dd 100644 --- a/functional_tests/test_data_management_security_patterns.py +++ b/functional_tests/test_data_management_security_patterns.py @@ -2,9 +2,9 @@ # test_data_management_security_patterns.py """ Functional test for Data Management security patterns. -Version: 0.250.101 +Version: 0.250.102 Implemented in: 0.241.211 -Updated in: 0.250.101 +Updated in: 0.250.102 This test ensures Data Management admin routes require authenticated admin access, secrets stay redacted in frontend responses, and the admin browser @@ -69,7 +69,7 @@ def test_version_and_container_registration(): """Validate the Data Management version and Cosmos job container registrations.""" config_source = read_text(CONFIG_FILE) - assert 'VERSION = "0.250.101"' in config_source + assert 'VERSION = "0.250.102"' in config_source assert 'cosmos_data_management_jobs_container_name = "data_management_jobs"' in config_source assert 'partition_key=PartitionKey(path="/id")' in config_source assert 'cosmos_data_management_job_items_container_name = "data_management_job_items"' in config_source @@ -350,6 +350,9 @@ def test_admin_javascript_uses_safe_dom_patterns(): 'getBackupLiveMetrics', 'updateBackupCapacityVisibility', 'backup_max_parallel_operations', + 'backup_blob_max_parallel_operations', + 'backup_blob_chunk_size_mib', + 'backup_blob_retry_count', 'backup_temporary_source_ru_enabled', 'updateMigrationCapacityVisibility', 'updateMigrationModeVisibility', @@ -412,6 +415,9 @@ def test_admin_ui_exposes_data_management_without_external_assets(): 'id="data_management_migration_max_parallel_operations"', 'id="data_management_backup_max_parallel_operations"', 'id="data_management_backup_retry_count"', + 'id="data_management_backup_blob_max_parallel_operations"', + 'id="data_management_backup_blob_chunk_size_mib"', + 'id="data_management_backup_blob_retry_count"', 'id="data_management_backup_capacity_failure_policy"', 'id="data_management_backup_temporary_source_ru_enabled"', 'id="data_management_backup_temporary_source_ru"', diff --git a/ui_tests/test_admin_data_management_settings_ui.py b/ui_tests/test_admin_data_management_settings_ui.py index 1af014b80..2e4d5f253 100644 --- a/ui_tests/test_admin_data_management_settings_ui.py +++ b/ui_tests/test_admin_data_management_settings_ui.py @@ -1,10 +1,10 @@ # test_admin_data_management_settings_ui.py """ UI test for Admin Settings Data Management controls. -Version: 0.250.076 +Version: 0.250.102 Implemented in: 0.241.211 Updated in: 0.241.221 -Updated in: 0.250.076 +Updated in: 0.250.102 This test ensures admins can discover the Data Management tab, see the operational-business-hours warning, and access the backup, encryption, @@ -15,6 +15,7 @@ Version 0.250.071 adds resilient migration provenance, incremental modes, cutover reconciliation, and the external target Search writer freeze acknowledgement. Version 0.250.076 adds bounded parallel backup and source capacity controls. +Version 0.250.102 adds independently bounded source-blob transfer controls. """ import os @@ -74,6 +75,10 @@ def test_admin_data_management_controls_render_from_template(): "data-management-backup-performance-section", "data_management_backup_max_parallel_operations", "data_management_backup_retry_count", + "data-management-blob-backup-performance-section", + "data_management_backup_blob_max_parallel_operations", + "data_management_backup_blob_chunk_size_mib", + "data_management_backup_blob_retry_count", "data_management_backup_capacity_failure_policy", "data_management_backup_temporary_source_ru_enabled", "data-management-backup-temporary-ru-field", @@ -240,6 +245,9 @@ def test_admin_data_management_controls_render_from_template(): assert 'Skipped / failed' in js_source assert 'label: "Elapsed"' in js_source assert 'backup_max_parallel_operations' in js_source + assert 'backup_blob_max_parallel_operations' in js_source + assert 'backup_blob_chunk_size_mib' in js_source + assert 'backup_blob_retry_count' in js_source assert 'backup_temporary_source_ru_enabled' in js_source assert 'buildMigrationPlan' in js_source assert 'queueMigration(false)' in js_source @@ -336,6 +344,9 @@ def test_admin_data_management_tab_browser_workflow(): expect(page.locator("#data_management_migration_max_parallel_operations")).to_be_visible() expect(page.locator("#data_management_backup_max_parallel_operations")).to_be_visible() expect(page.locator("#data_management_backup_retry_count")).to_be_visible() + expect(page.locator("#data_management_backup_blob_max_parallel_operations")).to_be_visible() + expect(page.locator("#data_management_backup_blob_chunk_size_mib")).to_be_visible() + expect(page.locator("#data_management_backup_blob_retry_count")).to_be_visible() expect(page.get_by_label("Temporarily increase local source Cosmos capacity for this backup")).to_be_visible() expect(page.locator("#data_management_migration_retry_count")).to_be_visible() expect(page.locator("#data_management_migration_skip_recent_within_hours")).to_be_visible()