diff --git a/application/single_app/config.py b/application/single_app/config.py index 09cae9063..cfe20b596 100644 --- a/application/single_app/config.py +++ b/application/single_app/config.py @@ -96,7 +96,7 @@ EXECUTOR_TYPE = 'thread' EXECUTOR_MAX_WORKERS = 30 SESSION_TYPE = 'filesystem' -VERSION = "0.250.219" +VERSION = "0.250.220" IS_DEVELOPMENT = is_development_env_enabled() SESSION_COOKIE_SAMESITE = os.getenv('SESSION_COOKIE_SAMESITE', 'Lax') diff --git a/application/single_app/functions_data_management.py b/application/single_app/functions_data_management.py index 5c2f32dee..8661905b0 100644 --- a/application/single_app/functions_data_management.py +++ b/application/single_app/functions_data_management.py @@ -212,6 +212,7 @@ DATA_MANAGEMENT_BACKUP_MANIFEST_BATCH_SIZE = 100 DATA_MANAGEMENT_BACKUP_MAX_PUBLIC_ITEM_SUMMARIES = 50 DATA_MANAGEMENT_BACKUP_MAX_LOGGED_FAILURE_REASONS = 10 +DATA_MANAGEMENT_BACKUP_CHECKPOINT_INTERVAL_SECONDS = 15 DATA_MANAGEMENT_BACKUP_MAX_RECENT_CHECKPOINTS = 20 DATA_MANAGEMENT_BACKUP_DEFAULT_PARALLEL_OPERATIONS = 4 DATA_MANAGEMENT_BACKUP_MAX_PARALLEL_OPERATIONS = 16 @@ -13393,6 +13394,14 @@ def _get_backup_blob_property(properties, field_name, default=None): return getattr(properties, field_name, default) +def _normalize_backup_etag(value): + """Strip transport quoting so list_blobs and get_blob_properties ETags compare equal.""" + normalized = _safe_text(value).strip() + if normalized[:2].upper() == "W/": + normalized = normalized[2:].strip() + return normalized.strip('"') + + 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( @@ -13658,6 +13667,7 @@ def _transfer_backup_source_blob( maximum=DATA_MANAGEMENT_BACKUP_MAX_RETRY_COUNT, ) source_etag = _safe_text(source_item.get("source_etag")) + normalized_source_etag = _normalize_backup_etag(source_etag) total_chunks = max( 1, (source_size + effective_chunk_size - 1) // effective_chunk_size, @@ -13800,10 +13810,10 @@ def _transfer_backup_source_blob( "Backup artifact length did not match the committed transfer." ) current_source_properties = source_blob_client.get_blob_properties() - current_source_etag = _safe_text( + current_source_etag = _normalize_backup_etag( _get_backup_blob_property(current_source_properties, "etag") ) - if source_etag and current_source_etag != source_etag: + if normalized_source_etag and current_source_etag != normalized_source_etag: raise RuntimeError("Source blob changed while it was being backed up.") succeeded_metadata = _build_backup_blob_target_metadata( backup_job, @@ -16660,10 +16670,12 @@ def _execute_backup_source_blob_resource( resource_name, ) pending_latest_state_updates = [] + last_persist_at = time.monotonic() def persist(message): - nonlocal state, previous_checkpoint + nonlocal state, previous_checkpoint, last_persist_at _assert_backup_job_lease(job) + last_persist_at = time.monotonic() if manifest_buffer: flush_manifest() elapsed_seconds = max(0.001, time.perf_counter() - started_at) @@ -16704,6 +16716,16 @@ def persist(message): pending_latest_state_updates, ) + def maybe_persist(message): + """Checkpoint on batch or interval so each transfer is not its own Cosmos write.""" + if ( + len(manifest_buffer) >= DATA_MANAGEMENT_BACKUP_MANIFEST_BATCH_SIZE or + time.monotonic() - last_persist_at >= DATA_MANAGEMENT_BACKUP_CHECKPOINT_INTERVAL_SECONDS + ): + persist(message) + else: + _assert_backup_job_lease(job) + def record_skipped(source_item, latest_state): nonlocal skipped_count skipped_count += 1 @@ -16918,7 +16940,7 @@ def record_transfer_result(candidate, transfer_result): artifact_path, ) batch_number += 1 - persist(f"Checkpointed source blob backup for {source_container_name}") + maybe_persist(f"Checkpointed source blob backup for {source_container_name}") try: source_iterator = iter(source_container_client.list_blobs()) diff --git a/docs/explanation/fixes/SOURCE_BLOB_BACKUP_ETAG_FIX.md b/docs/explanation/fixes/SOURCE_BLOB_BACKUP_ETAG_FIX.md new file mode 100644 index 000000000..15ccf9255 --- /dev/null +++ b/docs/explanation/fixes/SOURCE_BLOB_BACKUP_ETAG_FIX.md @@ -0,0 +1,149 @@ +> Root cause: every source blob backup compared a `list_blobs()` ETag against a +> `get_blob_properties()` ETag. Azure returns those in different transport formats, +> so the equality check never held and no source blob was ever backed up. + +# Source Blob Backup ETag Fix + +Fixed in version: **0.250.220** +Tracking issue: [#1271](https://github.com/microsoft/simplechat/issues/1271) + +## Issue Description + +Every source blob in every Data Management backup failed with: + +``` +Source blob changed while it was being backed up. +``` + +The failure rate was exactly 100% across all source blob containers, with zero retries. Backups reported `completed_with_warnings`, so the condition was easy to miss. + +Production evidence from job `data_management_partial_20260818T0300Z`: + +| Container | Blobs read | Copied | Failed | +|---|---|---|---| +| `user-documents` | 71 | 0 | 71 | +| `group-documents` | 427 | 0 | 427 | +| `public-documents` | 95 | 0 | 95 | +| `personal-chat` | 19,394 | 0 | 19,394 | + +Net effect: user documents, group documents, public documents, and all chat attachments had **never** been backed up. + +## Root Cause Analysis + +`_transfer_backup_source_blob` performs a post-upload consistency check that the source did not change mid-transfer: + +```python +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.") +``` + +The two operands originate from different Azure SDK code paths that format ETags differently: + +| Value | Origin | SDK path | Format | +|---|---|---|---| +| `source_etag` | `list_blobs()` via `_build_backup_blob_source_item` | `get_blob_properties_from_generated_code()` reads the XML `` element | `0x8DE...` | +| `current_source_etag` | `get_blob_properties()` | `BlobProperties(**headers)` reads the HTTP `ETag` header | `"0x8DE..."` | + +The HTTP `ETag` response header is an RFC 7232 quoted-string; the List Blobs XML element is not. `azure-storage-blob==12.24.1` performs no normalization in either direction, so the comparison was effectively `0x8DE... != "0x8DE..."`, which is always true. + +`RuntimeError` is not retryable under `_is_retryable_backup_blob_error`, so each blob failed immediately. Job telemetry reported `Retries / throttles: 0 / 0`, which corroborates a non-transient, first-attempt failure. + +### Why the failure was expensive + +The guard runs at the *end* of the transfer, after ranged reads, block staging, and `commit_block_list`. Every blob was therefore fully downloaded, optionally encrypted, and uploaded to the backup container before being discarded as failed. Those artifacts were committed with `pending` metadata and never promoted to `succeeded`, leaving orphaned pending artifacts behind on each run. + +The ranged reads also send that same unquoted ETag as an `If-Match` precondition and Azure accepted them, which independently confirms the blobs did not actually change and that only the Python-side string comparison was wrong. + +## Technical Details + +### Files Modified + +| File | Change | +|---|---| +| `application/single_app/functions_data_management.py` | ETag normalization for the equality check; batched source blob checkpoints | +| `application/single_app/config.py` | Version bump to `0.250.220` | +| `functional_tests/test_data_management_backup_source_blob_etag.py` | New regression coverage | +| `functional_tests/test_data_management_backup_cosmos_pagination.py` | Converted to real assertions so pytest reports failures | + +### Code Changes Summary + +**ETag normalization.** A new `_normalize_backup_etag` helper strips transport quoting and the optional `W/` weak-validator prefix. It is applied only at the comparison site: + +```python +source_etag = _safe_text(source_item.get("source_etag")) +normalized_source_etag = _normalize_backup_etag(source_etag) +... +current_source_etag = _normalize_backup_etag( + _get_backup_blob_property(current_source_properties, "etag") +) +if normalized_source_etag and current_source_etag != normalized_source_etag: + raise RuntimeError("Source blob changed while it was being backed up.") +``` + +`source_item["source_etag"]` deliberately retains the exact value returned by `list_blobs()`, so the `If-Match` precondition sent on ranged reads is byte-for-byte unchanged from current production behavior. The guard itself is preserved; a genuine mid-transfer source change still fails. + +**Checkpoint batching.** `record_transfer_result` previously called `persist()` for every item, producing one Cosmos write per blob — 19,394 writes for `personal-chat` alone, which capped throughput at roughly 6 items/second and accounted for the 74-minute runtime. A new `maybe_persist` helper checkpoints when the manifest buffer reaches `DATA_MANAGEMENT_BACKUP_MANIFEST_BATCH_SIZE` (100) or when `DATA_MANAGEMENT_BACKUP_CHECKPOINT_INTERVAL_SECONDS` (15) has elapsed, whichever comes first. The job lease is still asserted on every item, and the existing tail `persist()` continues to flush the final partial batch. Worst-case re-work after an interrupted run stays bounded at 100 items or 15 seconds. + +### Scope Check + +The migration path contains a visually similar comparison in `_copy_source_blobs_to_target`, but it sources `source_properties` from `get_blob_properties()`, so both operands are already quoted. Migration is **not** affected and was left unchanged. + +## Validation + +```bash +python -m pytest -q functional_tests/test_data_management_backup_source_blob_etag.py +``` + +| Test | Assertion | +|---|---| +| `test_etag_normalization_strips_transport_quoting` | Quoted, unquoted, weak, and padded ETags all normalize identically | +| `test_listed_and_fetched_etags_compare_equal` | The raw listed ETag is preserved for `If-Match`, and both formats compare equal once normalized | +| `test_transfer_succeeds_across_list_and_get_etag_formats` | The production transfer path succeeds end to end when the listing is unquoted and the fetch is quoted | +| `test_genuinely_changed_source_blob_still_fails` | A source blob whose ETag changes after download is still rejected | +| `test_verified_artifact_matches_source_version` | Reuse detection still keys off the recorded source version | +| `test_checkpoint_interval_is_bounded` | Checkpoint interval and batch size stay within safe bounds | +| `test_version_is_at_least_fix_version` | `config.py` version floor | + +**Regression probe.** Neutralizing `_normalize_backup_etag` causes four of the seven tests to fail, including the end-to-end transfer, which reports the exact production message: + +``` +Transfer must succeed, got 'failed' ('Source blob changed while it was being backed up.') +``` + +**Full Data Management suite:** 154 passed, 1 failed. The single failure, `test_data_management_backup_durability.py::test_backup_recovery_and_admin_progress_are_bounded_and_sanitized`, was confirmed pre-existing on `origin/Development` and is unrelated. + +### Test reliability fix + +While validating, the repository's standard functional-test template was found to hide failures under pytest. Tests written as: + +```python +def test_x(): + try: + assert ... + return True + except Exception: + return False +``` + +return a value rather than raising, and pytest reports them as **passed** with only a `PytestReturnNotNoneWarning`. A deliberately broken build reported `7 passed` while the underlying transfer was failing. + +Both backup test files added in this and the preceding fix now assert directly and return `None`, with the `__main__` block wrapping each call to preserve standalone console output and exit codes. Suite warnings dropped from 13 to 1. + +### Before / After + +| Behavior | Before | After | +|---|---|---| +| Source blob backup | 100% failure, `Source blob changed while it was being backed up.` | Succeeds | +| Genuine mid-transfer change | Rejected | Still rejected | +| `If-Match` precondition value | Raw listed ETag | Unchanged | +| Checkpoint writes | One Cosmos write per blob | Batched per 100 items or 15 seconds | +| Failing tests under pytest | Reported as passed | Reported as failed | + +## Cross-References + +- Issue: [#1271](https://github.com/microsoft/simplechat/issues/1271) +- Follow-up (cosmetic): [#1272](https://github.com/microsoft/simplechat/issues/1272) +- Preceding fix: `docs/explanation/fixes/COSMOS_BACKUP_CONTINUATION_TOKEN_FIX.md` +- Functional test: `functional_tests/test_data_management_backup_source_blob_etag.py` diff --git a/docs/explanation/release_notes.md b/docs/explanation/release_notes.md index c661ac6a5..71afe9473 100644 --- a/docs/explanation/release_notes.md +++ b/docs/explanation/release_notes.md @@ -2,6 +2,26 @@ For feature-focused and fix-focused drill-downs by version, see [Features by Version](/explanation/features/) and [Fixes by Version](/explanation/fixes/). +### **(v0.250.220)** + +#### Bug Fixes + +* **Source Blob Backup ETag Failure** + * Fixed every source blob failing backup with "Source blob changed while it was being backed up", which meant user documents, group documents, public documents, and chat attachments were never actually backed up. + * Root cause was comparing an ETag from `list_blobs()` (unquoted XML element) against one from `get_blob_properties()` (RFC-quoted HTTP header); the two never matched, so the post-transfer consistency check always failed after the blob had already been downloaded and uploaded. + * Both values are now normalized before comparison. The precondition sent to Azure is unchanged, and a genuine mid-transfer source change is still rejected. + * (Ref: #1271, `functions_data_management.py`, source blob transfer verification) + +* **Source Blob Backup Checkpoint Throughput** + * Source blob backups previously wrote one Cosmos checkpoint per blob, capping throughput at roughly six items per second and stretching a single container to over an hour. + * Checkpoints are now batched per 100 items or 15 seconds, whichever comes first, while still asserting the job lease on every item. + * (Ref: #1271, `functions_data_management.py`, source blob checkpointing) + +* **Functional Tests Silently Passing Under pytest** + * Backup functional tests written with the try/except and `return False` template were reported as passed by pytest because they returned a value instead of raising. + * The backup ETag and Cosmos pagination test files now assert directly, so real failures are reported by both pytest and standalone execution. + * (Ref: #1271, `test_data_management_backup_source_blob_etag.py`, `test_data_management_backup_cosmos_pagination.py`) + ### **(v0.250.219)** #### Bug Fixes diff --git a/functional_tests/test_data_management_backup_cosmos_pagination.py b/functional_tests/test_data_management_backup_cosmos_pagination.py index d7aac8987..d588a9eb4 100644 --- a/functional_tests/test_data_management_backup_cosmos_pagination.py +++ b/functional_tests/test_data_management_backup_cosmos_pagination.py @@ -165,177 +165,128 @@ def build_pages(page_count, page_size): def test_multi_page_cosmos_source_drains_a_single_pager(): """Containers larger than one page must stream fully without a token replay.""" - print("Testing Cosmos backup multi-page source reads...") - try: - module = load_data_management_module() - page_size = module.DATA_MANAGEMENT_BACKUP_MANIFEST_BATCH_SIZE - pages = build_pages(3, page_size) - container = FakePagedContainer(pages) - - items = list(module._iter_backup_cosmos_source_items( - container, - ARTIFACT, - {"cosmos_source_cutoff_epoch": 0}, - )) - - expected_count = 3 * page_size - assert len(items) == expected_count, ( - f"Expected {expected_count} streamed items, got {len(items)}" - ) - assert container.query_calls == 1, ( - f"Query must be built once, was built {container.query_calls} times" - ) - assert container.by_page_tokens == [None], ( - f"by_page must never receive a continuation token, got {container.by_page_tokens}" - ) - - identities = {item["source_identity"] for item in items} - assert len(identities) == expected_count, "Streamed items must be unique" - - print("Test passed!") - return True - except Exception as exc: - print(f"Test failed: {exc}") - import traceback - traceback.print_exc() - return False + module = load_data_management_module() + page_size = module.DATA_MANAGEMENT_BACKUP_MANIFEST_BATCH_SIZE + pages = build_pages(3, page_size) + container = FakePagedContainer(pages) + + items = list(module._iter_backup_cosmos_source_items( + container, + ARTIFACT, + {"cosmos_source_cutoff_epoch": 0}, + )) + + expected_count = 3 * page_size + assert len(items) == expected_count, ( + f"Expected {expected_count} streamed items, got {len(items)}" + ) + assert container.query_calls == 1, ( + f"Query must be built once, was built {container.query_calls} times" + ) + assert container.by_page_tokens == [None], ( + f"by_page must never receive a continuation token, got {container.by_page_tokens}" + ) + + identities = {item["source_identity"] for item in items} + assert len(identities) == expected_count, "Streamed items must be unique" def test_cutoff_and_unpaged_fallback_still_stream(): """The cutoff filter and non-paged test-double path must keep working.""" - print("Testing Cosmos backup cutoff filtering and unpaged fallback...") - try: - module = load_data_management_module() - page_size = module.DATA_MANAGEMENT_BACKUP_MANIFEST_BATCH_SIZE - - pages = build_pages(2, page_size) - pages[1][0]["_ts"] = 1900000000 - container = FakePagedContainer(pages) - items = list(module._iter_backup_cosmos_source_items( - container, - ARTIFACT, - {"cosmos_source_cutoff_epoch": 1800000000}, - )) - assert len(items) == (2 * page_size) - 1, ( - f"Cutoff must drop exactly one item, got {len(items)}" - ) - - flat_items = [item for page in build_pages(2, page_size) for item in page] - unpaged = FakeUnpagedContainer(flat_items) - unpaged_items = list(module._iter_backup_cosmos_source_items( - unpaged, - ARTIFACT, - {"cosmos_source_cutoff_epoch": 0}, - )) - assert len(unpaged_items) == 2 * page_size, ( - f"Unpaged fallback must stream every item, got {len(unpaged_items)}" - ) - - print("Test passed!") - return True - except Exception as exc: - print(f"Test failed: {exc}") - import traceback - traceback.print_exc() - return False + module = load_data_management_module() + page_size = module.DATA_MANAGEMENT_BACKUP_MANIFEST_BATCH_SIZE + + pages = build_pages(2, page_size) + pages[1][0]["_ts"] = 1900000000 + container = FakePagedContainer(pages) + items = list(module._iter_backup_cosmos_source_items( + container, + ARTIFACT, + {"cosmos_source_cutoff_epoch": 1800000000}, + )) + assert len(items) == (2 * page_size) - 1, ( + f"Cutoff must drop exactly one item, got {len(items)}" + ) + + flat_items = [item for page in build_pages(2, page_size) for item in page] + unpaged = FakeUnpagedContainer(flat_items) + unpaged_items = list(module._iter_backup_cosmos_source_items( + unpaged, + ARTIFACT, + {"cosmos_source_cutoff_epoch": 0}, + )) + assert len(unpaged_items) == 2 * page_size, ( + f"Unpaged fallback must stream every item, got {len(unpaged_items)}" + ) def test_failure_reason_rollup_is_bounded(): """Per-item failures must collapse into a bounded, ordered log summary.""" - print("Testing backup failure reason rollup...") - try: - module = load_data_management_module() - limit = module.DATA_MANAGEMENT_BACKUP_MAX_LOGGED_FAILURE_REASONS - - counts = {} - for index in range(limit + 25): - module._record_backup_failure_reason(counts, f"Failure kind {index}") - for _ in range(5): - module._record_backup_failure_reason(counts, "Failure kind 0") - - assert len(counts) <= limit + 1, ( - f"Distinct reasons must stay bounded, got {len(counts)}" - ) - assert "Other backup failures." in counts, "Overflow reasons must be aggregated" - assert counts["Failure kind 0"] == 6, ( - f"Repeat reasons must accumulate, got {counts['Failure kind 0']}" - ) - - summary = module._summarize_backup_failure_reasons(counts) - rendered_counts = [int(entry.split("x ", 1)[0]) for entry in summary.split("; ")] - assert rendered_counts == sorted(rendered_counts, reverse=True), ( - f"Summary must be ordered by frequency, got {summary!r}" - ) - assert "25x Other backup failures." in summary, ( - f"Overflow bucket must be reported, got {summary!r}" - ) - assert "6x Failure kind 0" in summary, ( - f"Repeat reason counts must be reported, got {summary!r}" - ) - assert module._summarize_backup_failure_reasons({}) == "", ( - "Empty rollups must render as an empty string" - ) - - print("Test passed!") - return True - except Exception as exc: - print(f"Test failed: {exc}") - import traceback - traceback.print_exc() - return False + module = load_data_management_module() + limit = module.DATA_MANAGEMENT_BACKUP_MAX_LOGGED_FAILURE_REASONS + + counts = {} + for index in range(limit + 25): + module._record_backup_failure_reason(counts, f"Failure kind {index}") + for _ in range(5): + module._record_backup_failure_reason(counts, "Failure kind 0") + + assert len(counts) <= limit + 1, ( + f"Distinct reasons must stay bounded, got {len(counts)}" + ) + assert "Other backup failures." in counts, "Overflow reasons must be aggregated" + assert counts["Failure kind 0"] == 6, ( + f"Repeat reasons must accumulate, got {counts['Failure kind 0']}" + ) + + summary = module._summarize_backup_failure_reasons(counts) + rendered_counts = [int(entry.split("x ", 1)[0]) for entry in summary.split("; ")] + assert rendered_counts == sorted(rendered_counts, reverse=True), ( + f"Summary must be ordered by frequency, got {summary!r}" + ) + assert "25x Other backup failures." in summary, ( + f"Overflow bucket must be reported, got {summary!r}" + ) + assert "6x Failure kind 0" in summary, ( + f"Repeat reason counts must be reported, got {summary!r}" + ) + assert module._summarize_backup_failure_reasons({}) == "", ( + "Empty rollups must render as an empty string" + ) def test_logger_extra_retains_sanitized_diagnostics(): """App Insights properties must carry message text without leaking secrets.""" - print("Testing App Insights logger extras...") - try: - sys.modules.pop("functions_appinsights", None) - sys.modules.setdefault("app_settings_cache", types.ModuleType("app_settings_cache")) - import functions_appinsights - - extra = functions_appinsights._build_logger_extra( - "[DATA_MANAGEMENT] Cosmos backup source page read failed.", - { - "job_id": "data_management_partial_20260817T0300Z", - "container": "conversations", - "status_code": 400, - "error": "(BadRequest) Invalid Continuation Token", - "api_key": "super-secret-value", - }, - ) - - assert "Cosmos backup source page read failed" in extra["sc_message"], ( - "Sanitized message text must reach App Insights" - ) - assert extra["sc_error"] == "(BadRequest) Invalid Continuation Token", ( - f"Allowlisted diagnostics must retain text, got {extra.get('sc_error')!r}" - ) - assert extra["sc_container"] == "conversations" - assert extra["sc_status_code"] == 400 - assert extra["sc_api_key_present"] is True, "Sensitive keys must collapse to presence" - assert "super-secret-value" not in str(extra), "Secret values must never be emitted" - - print("Test passed!") - return True - except Exception as exc: - print(f"Test failed: {exc}") - import traceback - traceback.print_exc() - return False + sys.modules.pop("functions_appinsights", None) + sys.modules.setdefault("app_settings_cache", types.ModuleType("app_settings_cache")) + import functions_appinsights + + extra = functions_appinsights._build_logger_extra( + "[DATA_MANAGEMENT] Cosmos backup source page read failed.", + { + "job_id": "data_management_partial_20260817T0300Z", + "container": "conversations", + "status_code": 400, + "error": "(BadRequest) Invalid Continuation Token", + "api_key": "super-secret-value", + }, + ) + + assert "Cosmos backup source page read failed" in extra["sc_message"], ( + "Sanitized message text must reach App Insights" + ) + assert extra["sc_error"] == "(BadRequest) Invalid Continuation Token", ( + f"Allowlisted diagnostics must retain text, got {extra.get('sc_error')!r}" + ) + assert extra["sc_container"] == "conversations" + assert extra["sc_status_code"] == 400 + assert extra["sc_api_key_present"] is True, "Sensitive keys must collapse to presence" + assert "super-secret-value" not in str(extra), "Secret values must never be emitted" def test_version_is_at_least_fix_version(): """The shipped app version must include this fix.""" - print("Testing config version floor...") - try: - assert_app_version_at_least("0.250.209") - print("Test passed!") - return True - except Exception as exc: - print(f"Test failed: {exc}") - import traceback - traceback.print_exc() - return False + assert_app_version_at_least("0.250.209") if __name__ == "__main__": @@ -346,10 +297,17 @@ def test_version_is_at_least_fix_version(): test_logger_extra_retains_sanitized_diagnostics, test_version_is_at_least_fix_version, ] - results = [] + failures = 0 for test in tests: print(f"\nRunning {test.__name__}...") - results.append(test()) - - print(f"\nResults: {sum(results)}/{len(results)} tests passed") - sys.exit(0 if all(results) else 1) + try: + test() + print("Test passed!") + except Exception as exc: + failures += 1 + print(f"Test failed: {exc}") + import traceback + traceback.print_exc() + + print(f"\nResults: {len(tests) - failures}/{len(tests)} tests passed") + sys.exit(1 if failures else 0) diff --git a/functional_tests/test_data_management_backup_source_blob_etag.py b/functional_tests/test_data_management_backup_source_blob_etag.py new file mode 100644 index 000000000..47bb23946 --- /dev/null +++ b/functional_tests/test_data_management_backup_source_blob_etag.py @@ -0,0 +1,339 @@ +#!/usr/bin/env python3 +# test_data_management_backup_source_blob_etag.py +""" +Functional test for source blob backup ETag normalization and checkpoint batching. +Version: 0.250.220 +Implemented in: 0.250.220 + +This test ensures source blob backups survive the ETag quoting difference between +list_blobs() (XML element, unquoted) and get_blob_properties() (HTTP ETag +header, RFC quoted). That mismatch previously failed every source blob with +"Source blob changed while it was being backed up." and zero retries, so no +source document or chat attachment was ever backed up. + +Test functions assert directly and return None so pytest reports real failures. +""" + +import importlib.util +from pathlib import Path +import sys +import types + +REPO_ROOT = Path(__file__).resolve().parents[1] +APP_ROOT = REPO_ROOT / "application" / "single_app" +MODULE_PATH = APP_ROOT / "functions_data_management.py" +sys.path.insert(0, str(APP_ROOT)) +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from test_support.versioning import assert_app_version_at_least + +RAW_ETAG = "0x8DE9A1B2C3D4E5F" + + +class FakeListedBlobProperties: + """Model a list_blobs() result, whose XML ETag element carries no quotes.""" + + def __init__(self, name, size, etag): + self.name = name + self.size = size + self.etag = etag + self.last_modified = "2026-08-18T03:00:00+00:00" + + +class FakeFetchedBlobProperties: + """Model a get_blob_properties() result, whose HTTP ETag header is RFC quoted.""" + + def __init__(self, size, etag, metadata=None): + self.size = size + self.etag = f'"{etag}"' + self.metadata = metadata or {} + + +class FakeNotFound(Exception): + """Stand in for a missing destination blob or a failed precondition.""" + + status_code = 404 + + +class FakeDownload: + """Return a fixed payload for a ranged source read.""" + + def __init__(self, payload): + self._payload = payload + + def readinto(self, buffer): + buffer.write(self._payload) + return len(self._payload) + + +class FakeSourceBlobClient: + """Serve ranged reads and a quoted ETag, mirroring real Blob Storage behavior.""" + + def __init__(self, payload, etag, etag_after_download=None): + self._payload = payload + self._etag = etag + self._etag_after_download = etag_after_download + self._downloaded = False + self.download_calls = [] + + def download_blob(self, offset=0, length=None, **kwargs): + condition_etag = kwargs.get("etag") + self.download_calls.append(condition_etag) + # Azure accepts If-Match with or without transport quoting. + if condition_etag and condition_etag.strip('"') != self._etag.strip('"'): + raise FakeNotFound() + self._downloaded = True + end = offset + (length if length is not None else len(self._payload)) + return FakeDownload(self._payload[offset:end]) + + def get_blob_properties(self): + current_etag = self._etag + if self._downloaded and self._etag_after_download: + current_etag = self._etag_after_download + return FakeFetchedBlobProperties(len(self._payload), current_etag) + + +class FakeTargetBlobClient: + """Accept staged blocks and record the committed artifact.""" + + def __init__(self): + self.blocks = {} + self.committed = None + self.metadata = {} + + def get_blob_properties(self): + if self.committed is None: + raise FakeNotFound() + return FakeFetchedBlobProperties(len(self.committed), "0xTARGETETAG", self.metadata) + + def upload_blob(self, data=None, **kwargs): + self.committed = data + self.metadata = dict(kwargs.get("metadata") or {}) + + def stage_block(self, block_id=None, data=None, **_kwargs): + self.blocks[block_id] = data + + def commit_block_list(self, block_list, **kwargs): + self.committed = b"".join(self.blocks[block.id] for block in block_list) + self.metadata = dict(kwargs.get("metadata") or {}) + + def set_blob_metadata(self, metadata=None, **_kwargs): + self.metadata = dict(metadata or {}) + + +class FakeTargetContainerClient: + """Hand out a single reusable destination blob client.""" + + def __init__(self, blob_client): + self._blob_client = blob_client + + def get_blob_client(self, _name): + return self._blob_client + + +def load_data_management_module(): + """Load production backup helpers with stubbed infrastructure dependencies.""" + config_module = types.ModuleType("config") + config_module.CLIENTS = {} + config_module.VERSION = "0.250.220" + config_module.cosmos_data_management_jobs_container = None + config_module.cosmos_data_management_job_items_container = None + config_module.cosmos_settings_container = None + sys.modules["config"] = config_module + + appinsights_module = types.ModuleType("functions_appinsights") + appinsights_module.log_event = lambda *_a, **_k: None + sys.modules["functions_appinsights"] = appinsights_module + + throughput_module = types.ModuleType("functions_cosmos_throughput") + + class FakeCosmosThroughputError(Exception): + pass + + throughput_module.CosmosThroughputError = FakeCosmosThroughputError + throughput_module.get_container_throughput = lambda *_a, **_k: {} + throughput_module.get_database_throughput = lambda *_a, **_k: {} + throughput_module.set_database_throughput = lambda *_a, **_k: {} + sys.modules["functions_cosmos_throughput"] = throughput_module + + module_name = "data_management_backup_etag_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) + sys.modules.pop(module_name, None) + return module + + +def run_transfer(module, source_client, source_item, retry_count=1): + """Run the production transfer against in-memory blob clients.""" + target_client = FakeTargetBlobClient() + result = module._transfer_backup_source_blob( + FakeTargetContainerClient(target_client), + source_client, + "backups/source_blobs/personal-chat/attachment.bin", + source_item, + fernet=None, + backup_job={"id": "job-1", "backup_attempt_id": "attempt-1", "lease_generation": 1}, + chunk_size_bytes=None, + retry_count=retry_count, + ) + return result, target_client + + +def test_etag_normalization_strips_transport_quoting(): + """Quoted, unquoted, and weak ETags must normalize to the same value.""" + module = load_data_management_module() + normalize = module._normalize_backup_etag + + assert normalize(RAW_ETAG) == RAW_ETAG + assert normalize(f'"{RAW_ETAG}"') == RAW_ETAG, "Quoted header ETag must normalize" + assert normalize(f'W/"{RAW_ETAG}"') == RAW_ETAG, "Weak validator must normalize" + assert normalize(f' "{RAW_ETAG}" ') == RAW_ETAG, "Surrounding space must normalize" + assert normalize(None) == "", "Missing ETag must normalize to empty" + assert normalize("") == "" + + +def test_listed_and_fetched_etags_compare_equal(): + """A list_blobs ETag must match the same blob's get_blob_properties ETag.""" + module = load_data_management_module() + + listed = FakeListedBlobProperties("chat/abc.png", 2048, RAW_ETAG) + source_item = module._build_backup_blob_source_item("personal-chat", listed.name, listed) + + assert source_item["source_etag"] == RAW_ETAG, ( + "The conditional-header ETag must be preserved exactly as listed" + ) + + fetched = FakeFetchedBlobProperties(2048, RAW_ETAG) + assert fetched.etag != source_item["source_etag"], ( + "This test is meaningless unless the two transport formats differ" + ) + assert ( + module._normalize_backup_etag(fetched.etag) == + module._normalize_backup_etag(source_item["source_etag"]) + ), "Listed and fetched ETags must compare equal after normalization" + + +def test_transfer_succeeds_across_list_and_get_etag_formats(): + """The real transfer path must not treat quoting differences as a changed blob.""" + module = load_data_management_module() + + payload = b"simplechat source blob payload" * 64 + listed = FakeListedBlobProperties("chat/attachment.bin", len(payload), RAW_ETAG) + source_item = module._build_backup_blob_source_item("personal-chat", listed.name, listed) + source_client = FakeSourceBlobClient(payload, RAW_ETAG) + + result, target_client = run_transfer(module, source_client, source_item) + + assert result["status"] == "succeeded", ( + f"Transfer must succeed, got {result['status']!r} ({result.get('failure_summary')!r})" + ) + assert result["source_bytes"] == len(payload) + assert target_client.committed == payload, "Committed artifact must match the source" + assert target_client.metadata.get("simplechatbackupstatus") == "succeeded", ( + "Artifact metadata must be promoted from pending to succeeded" + ) + assert result["retry_attempt_count"] == 0, "A clean transfer must not retry" + + +def test_genuinely_changed_source_blob_still_fails(): + """A real mid-transfer source change must still be rejected.""" + module = load_data_management_module() + + payload = b"payload" + listed = FakeListedBlobProperties("chat/changed.bin", len(payload), RAW_ETAG) + source_item = module._build_backup_blob_source_item("personal-chat", listed.name, listed) + source_client = FakeSourceBlobClient( + payload, + RAW_ETAG, + etag_after_download="0xCHANGEDMIDTRANSFER", + ) + + result, _target_client = run_transfer(module, source_client, source_item) + + assert result["status"] == "failed", ( + f"A changed source blob must still fail, got {result['status']!r}" + ) + assert "changed while it was being backed up" in result["failure_summary"], ( + f"Failure summary must name the source change, got {result['failure_summary']!r}" + ) + + +def test_verified_artifact_matches_source_version(): + """Reused-artifact detection must still key off the recorded source version.""" + module = load_data_management_module() + + listed = FakeListedBlobProperties("docs/report.pdf", 4096, RAW_ETAG) + source_item = module._build_backup_blob_source_item("user-documents", listed.name, listed) + backup_job = {"id": "job-1", "backup_attempt_id": "attempt-1", "lease_generation": 2} + transfer_format = module.DATA_MANAGEMENT_BLOB_BACKUP_RAW_FORMAT + + metadata = module._build_backup_blob_target_metadata( + backup_job, + source_item, + transfer_format, + "succeeded", + ) + target_properties = {"metadata": metadata, "size": 4096} + + assert module._is_verified_backup_blob_artifact( + target_properties, + backup_job, + source_item, + transfer_format, + ), "A succeeded artifact for the same source version must verify as reusable" + + changed = module._build_backup_blob_source_item( + "user-documents", + listed.name, + FakeListedBlobProperties(listed.name, 4096, "0xDIFFERENTETAG"), + ) + assert not module._is_verified_backup_blob_artifact( + target_properties, + backup_job, + changed, + transfer_format, + ), "A genuinely changed source version must not verify as reusable" + + +def test_checkpoint_interval_is_bounded(): + """Checkpoint batching must stay bounded so resume work stays small.""" + module = load_data_management_module() + interval = module.DATA_MANAGEMENT_BACKUP_CHECKPOINT_INTERVAL_SECONDS + batch_size = module.DATA_MANAGEMENT_BACKUP_MANIFEST_BATCH_SIZE + + assert 1 <= interval <= 60, f"Checkpoint interval must stay within a minute, got {interval}" + assert 1 <= batch_size <= 500, f"Manifest batch size must stay bounded, got {batch_size}" + + +def test_version_is_at_least_fix_version(): + """The shipped app version must include this fix.""" + assert_app_version_at_least("0.250.220") + + +if __name__ == "__main__": + tests = [ + test_etag_normalization_strips_transport_quoting, + test_listed_and_fetched_etags_compare_equal, + test_transfer_succeeds_across_list_and_get_etag_formats, + test_genuinely_changed_source_blob_still_fails, + test_verified_artifact_matches_source_version, + test_checkpoint_interval_is_bounded, + test_version_is_at_least_fix_version, + ] + failures = 0 + for test in tests: + print(f"\nRunning {test.__name__}...") + try: + test() + print("Test passed!") + except Exception as exc: + failures += 1 + print(f"Test failed: {exc}") + import traceback + traceback.print_exc() + + print(f"\nResults: {len(tests) - failures}/{len(tests)} tests passed") + sys.exit(1 if failures else 0)