From 70b191e84984684d58ac5b677367c58b16021f36 Mon Sep 17 00:00:00 2001 From: Paul Lizer Date: Mon, 17 Aug 2026 17:20:35 -0400 Subject: [PATCH] Fix Cosmos backup continuation token failure Data Management backups silently omitted every Cosmos container holding more than one page of documents. Affected containers failed with "BadRequest: Invalid Continuation Token" and were dropped from the backup artifact set while the job still reported completion with warnings, so in most deployments personal conversations and personal messages were never backed up. _iter_backup_cosmos_source_items rebuilt the cross-partition query for each page and replayed the previous pager's continuation token. Cross-partition results are merged client side, so that token cannot resume the merged query, and rebuilding query_items discarded the SDK execution context. Page one always succeeded and page two always failed. Drain a single pager to exhaustion instead, retrying on the same pager so the execution context survives. Also close two observability gaps found while diagnosing this: - Source blob transfer failures produced no log output at all. Log the first failure per resource plus a bounded rollup of distinct failure reasons and counts at resource completion. - Application Insights received the constant [SIMPLE_CHAT_LOG_EVENT] with every string property reduced to a length. Retain sanitized message text and an allowlist of non-sensitive diagnostic keys. Sensitive keys still collapse to a presence flag and secret redaction is unchanged. Fixes #1258 --- application/single_app/config.py | 2 +- .../single_app/functions_appinsights.py | 33 +- .../single_app/functions_data_management.py | 245 +++++++----- .../COSMOS_BACKUP_CONTINUATION_TOKEN_FIX.md | 119 ++++++ docs/explanation/release_notes.md | 20 + ...ata_management_backup_cosmos_pagination.py | 355 ++++++++++++++++++ 6 files changed, 682 insertions(+), 92 deletions(-) create mode 100644 docs/explanation/fixes/COSMOS_BACKUP_CONTINUATION_TOKEN_FIX.md create mode 100644 functional_tests/test_data_management_backup_cosmos_pagination.py diff --git a/application/single_app/config.py b/application/single_app/config.py index 4ce4fc421..675ef29d8 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.208" +VERSION = "0.250.209" IS_DEVELOPMENT = is_development_env_enabled() SESSION_COOKIE_SAMESITE = os.getenv('SESSION_COOKIE_SAMESITE', 'Lax') diff --git a/application/single_app/functions_appinsights.py b/application/single_app/functions_appinsights.py index 92af07399..6d1a1c6e2 100644 --- a/application/single_app/functions_appinsights.py +++ b/application/single_app/functions_appinsights.py @@ -75,6 +75,28 @@ LOGGER_FALLBACK_MESSAGE = "[SIMPLE_CHAT_LOG_FALLBACK]" LOGGER_EXTERNAL_EVENT_MESSAGE = "[SIMPLE_CHAT_EXTERNAL_EVENT]" MAX_EXTERNAL_EVENT_STRING_LENGTH = 256 +LOGGER_SAFE_TEXT_MAX_LENGTH = 1024 +# Diagnostic keys whose sanitized text may be retained; everything else stays a length. +LOGGER_SAFE_TEXT_KEYS = frozenset({ + "attempt", + "container", + "containername", + "error", + "errortype", + "failurereasons", + "failuresummary", + "indexname", + "jobid", + "operation", + "reason", + "resource", + "resourcename", + "service", + "stage", + "status", + "step", + "taskname", +}) def _format_message(message: Any, message_args: Optional[Tuple[Any, ...]] = None) -> str: @@ -113,6 +135,10 @@ def _is_sensitive_external_event_key(key: Any) -> bool: ) +def _is_safe_log_text_key(key: Any) -> bool: + return _normalize_log_key(key) in LOGGER_SAFE_TEXT_KEYS + + def sanitize_log_message(message: Any) -> str: """Redact secret-like values from log messages while preserving diagnostic text.""" message_text = str(message) @@ -176,8 +202,9 @@ def _build_logger_extra( message: Any, extra: Optional[Dict[str, Any]] = None, ) -> Optional[Dict[str, Any]]: - """Build log-record properties without clear-text user-controlled values.""" + """Build log-record properties, keeping only sanitized allowlisted diagnostic text.""" logger_extra: Dict[str, Any] = { + "sc_message": sanitize_log_message(message or "")[:MAX_LOG_STRING_LENGTH], "sc_message_length": len(str(message or "")), } @@ -193,6 +220,10 @@ def _build_logger_extra( logger_extra[f"{normalized_key}_count"] = len(value) elif isinstance(value, str): logger_extra[f"{normalized_key}_length"] = len(value) + if _is_safe_log_text_key(key): + logger_extra[normalized_key] = ( + sanitize_log_message(value)[:LOGGER_SAFE_TEXT_MAX_LENGTH] + ) else: logger_extra[normalized_key] = _logger_safe_scalar(value) diff --git a/application/single_app/functions_data_management.py b/application/single_app/functions_data_management.py index 21f3ff89f..5c2f32dee 100644 --- a/application/single_app/functions_data_management.py +++ b/application/single_app/functions_data_management.py @@ -211,6 +211,7 @@ DATA_MANAGEMENT_MIGRATION_MANIFEST_BATCH_SIZE = 100 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_MAX_RECENT_CHECKPOINTS = 20 DATA_MANAGEMENT_BACKUP_DEFAULT_PARALLEL_OPERATIONS = 4 DATA_MANAGEMENT_BACKUP_MAX_PARALLEL_OPERATIONS = 16 @@ -14408,6 +14409,29 @@ def _append_backup_state_summary(state, field_name, summary): del values[:-DATA_MANAGEMENT_BACKUP_MAX_PUBLIC_ITEM_SUMMARIES] +def _record_backup_failure_reason(reason_counts, failure_summary): + """Tally distinct failure reasons so per-item faults log as one bounded rollup.""" + if not isinstance(reason_counts, dict): + return + reason = _safe_text(failure_summary)[:200] or "Unspecified backup failure." + if reason in reason_counts: + reason_counts[reason] += 1 + elif len(reason_counts) < DATA_MANAGEMENT_BACKUP_MAX_LOGGED_FAILURE_REASONS: + reason_counts[reason] = 1 + else: + reason_counts["Other backup failures."] = ( + reason_counts.get("Other backup failures.", 0) + 1 + ) + + +def _summarize_backup_failure_reasons(reason_counts): + """Render the most frequent failure reasons as a single log-safe string.""" + if not isinstance(reason_counts, dict) or not reason_counts: + return "" + ordered = sorted(reason_counts.items(), key=lambda entry: (-entry[1], entry[0])) + return "; ".join(f"{count}x {reason}" for reason, count in ordered) + + def _build_backup_checkpoint_id(job, resource_name, batch_number): return ( f"{_safe_text(job.get('id'))}:" @@ -14830,102 +14854,101 @@ def normalize_item(raw_item): "source_version": source_version, } - continuation_token = None - while True: + def finalize_page(normalized_page): + normalized_page.sort( + key=lambda item: (item["source_identity"], item["source_version"]), + ) + report_telemetry({"source_page_count": 1}) + return normalized_page + + def normalize_page(raw_page): + return finalize_page([ + item for item in (normalize_item(raw_item) for raw_item in raw_page) + if item is not None + ]) + + def assert_not_canceled(): if cancel_event is not None and cancel_event.is_set(): raise DataManagementBackupCanceledError( "Cosmos source export stopped after backup cancellation or lease loss." ) - page_completed = False - for attempt in range(1, retry_count + 1): - try: - source_iterable = container.query_items( - query=query, - parameters=parameters, - enable_cross_partition_query=True, - max_item_count=DATA_MANAGEMENT_BACKUP_MANIFEST_BATCH_SIZE, - populate_query_metrics=True, - response_hook=response_hook, - ) - if hasattr(source_iterable, "by_page"): - page_iterator = source_iterable.by_page( - continuation_token=continuation_token, - ) - try: - source_page = list(next(page_iterator)) - except StopIteration: - return - normalized_page = [ - item for item in (normalize_item(raw_item) for raw_item in source_page) - if item is not None - ] - normalized_page.sort( - key=lambda item: (item["source_identity"], item["source_version"]), - ) - report_telemetry({"source_page_count": 1}) - for source_item in normalized_page: - yield source_item - continuation_token = getattr(page_iterator, "continuation_token", None) - page_completed = True - break - # Lightweight test doubles may not expose Cosmos paging. Keep their - # staging bounded even though production uses continuation pages. - buffered_items = [] - for raw_item in source_iterable: - normalized_item = normalize_item(raw_item) - if normalized_item is None: - continue - buffered_items.append(normalized_item) - if len(buffered_items) < DATA_MANAGEMENT_BACKUP_MANIFEST_BATCH_SIZE: - continue - buffered_items.sort( - key=lambda item: (item["source_identity"], item["source_version"]), - ) - report_telemetry({"source_page_count": 1}) - for source_item in buffered_items: - yield source_item - buffered_items = [] - if buffered_items: - buffered_items.sort( - key=lambda item: (item["source_identity"], item["source_version"]), - ) - report_telemetry({"source_page_count": 1}) - for source_item in buffered_items: - yield source_item - return - except (DataManagementBackupCanceledError, DataManagementBackupLeaseLostError): - raise - except Exception as exc: - if not _is_retryable_backup_cosmos_error(exc) or attempt >= retry_count: - log_event( - "[DATA_MANAGEMENT] Cosmos backup source page read failed.", - { - "container": container_name, - "status_code": _get_backup_exception_status_code(exc), - "error": str(exc), - }, - level=logging.WARNING, - ) + def wait_before_page_retry(exc, attempt): + if not _is_retryable_backup_cosmos_error(exc) or attempt >= retry_count: + log_event( + "[DATA_MANAGEMENT] Cosmos backup source page read failed.", + { + "container": container_name, + "status_code": _get_backup_exception_status_code(exc), + "attempt": attempt, + "error": str(exc), + }, + level=logging.WARNING, + ) + raise exc + status_code = _get_backup_exception_status_code(exc) + retry_delay = _get_backup_retry_delay(exc, attempt) + report_telemetry({ + "source_retry_attempt_count": 1, + "source_throttle_count": 1 if status_code in {429, 449} else 0, + "last_retry_delay_seconds": round(retry_delay, 3), + }) + if cancel_event is not None: + if cancel_event.wait(retry_delay): + raise DataManagementBackupCanceledError( + "Cosmos source export stopped during retry backoff." + ) from exc + else: + time.sleep(retry_delay) + + assert_not_canceled() + source_iterable = container.query_items( + query=query, + parameters=parameters, + enable_cross_partition_query=True, + max_item_count=DATA_MANAGEMENT_BACKUP_MANIFEST_BATCH_SIZE, + populate_query_metrics=True, + response_hook=response_hook, + ) + + if hasattr(source_iterable, "by_page"): + # Cross-partition results are merged client side, so the surfaced continuation + # token cannot be replayed against a rebuilt query. Drain a single pager. + page_iterator = source_iterable.by_page() + while True: + assert_not_canceled() + raw_page = None + for attempt in range(1, retry_count + 1): + try: + raw_page = list(next(page_iterator)) + break + except StopIteration: + return + except (DataManagementBackupCanceledError, DataManagementBackupLeaseLostError): raise - status_code = _get_backup_exception_status_code(exc) - retry_delay = _get_backup_retry_delay(exc, attempt) - report_telemetry({ - "source_retry_attempt_count": 1, - "source_throttle_count": 1 if status_code in {429, 449} else 0, - "last_retry_delay_seconds": round(retry_delay, 3), - }) - if cancel_event is not None: - if cancel_event.wait(retry_delay): - raise DataManagementBackupCanceledError( - "Cosmos source export stopped during retry backoff." - ) - else: - time.sleep(retry_delay) - if not page_completed: - return - if not continuation_token: - return + except Exception as exc: + wait_before_page_retry(exc, attempt) + if raw_page is None: + return + for source_item in normalize_page(raw_page): + yield source_item + + # Lightweight test doubles may not expose Cosmos paging. Keep their staging + # bounded even though production drains continuation pages. + buffered_items = [] + for raw_item in source_iterable: + normalized_item = normalize_item(raw_item) + if normalized_item is None: + continue + buffered_items.append(normalized_item) + if len(buffered_items) < DATA_MANAGEMENT_BACKUP_MANIFEST_BATCH_SIZE: + continue + for source_item in finalize_page(buffered_items): + yield source_item + buffered_items = [] + if buffered_items: + for source_item in finalize_page(buffered_items): + yield source_item def _escape_backup_search_filter_value(value): @@ -16629,6 +16652,7 @@ def _execute_backup_source_blob_resource( retry_count = _get_backup_blob_retry_count(settings, backup_plan) clean_transfer_count = 0 current_files = set() + failure_reason_counts = {} started_at = time.perf_counter() cancel_event = Event() append_manifest, flush_manifest, manifest_buffer = _create_backup_manifest_writer( @@ -16846,6 +16870,20 @@ def record_transfer_result(candidate, transfer_result): ) or "Source blob transfer failed." ) + _record_backup_failure_reason(failure_reason_counts, failure_summary) + if failed_count == 1: + log_event( + "[DATA_MANAGEMENT] Source blob backup item failed.", + { + "job_id": job.get("id"), + "resource": resource_name, + "container": source_container_name, + "failure_summary": failure_summary, + "retry_attempt_count": transfer_result.get("retry_attempt_count"), + "throttle_count": transfer_result.get("throttle_count"), + }, + level=logging.WARNING, + ) _append_backup_state_summary(state, "failed_items", { "service": "source_blobs", "resource_name": resource_name, @@ -16950,12 +16988,39 @@ def record_transfer_result(candidate, transfer_result): _sanitize_data_management_backup_text(str(exc)) or "Source blob enumeration failed." ) + _record_backup_failure_reason(failure_reason_counts, failure_summary) + log_event( + "[DATA_MANAGEMENT] Source blob backup enumeration failed.", + { + "job_id": job.get("id"), + "resource": resource_name, + "container": source_container_name, + "failure_summary": failure_summary, + }, + level=logging.WARNING, + ) _append_backup_state_summary(state, "failed_items", { "service": "source_blobs", "resource_name": resource_name, "failure_summary": failure_summary, }) + if failed_count: + log_event( + "[DATA_MANAGEMENT] Source blob backup completed with failures.", + { + "job_id": job.get("id"), + "resource": resource_name, + "container": source_container_name, + "failed_count": failed_count, + "copied_count": copied_count, + "skipped_count": skipped_count, + "source_read_count": source_read_count, + "failure_reasons": _summarize_backup_failure_reasons(failure_reason_counts), + }, + level=logging.WARNING, + ) + 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) diff --git a/docs/explanation/fixes/COSMOS_BACKUP_CONTINUATION_TOKEN_FIX.md b/docs/explanation/fixes/COSMOS_BACKUP_CONTINUATION_TOKEN_FIX.md new file mode 100644 index 000000000..ce2eb8171 --- /dev/null +++ b/docs/explanation/fixes/COSMOS_BACKUP_CONTINUATION_TOKEN_FIX.md @@ -0,0 +1,119 @@ +# Cosmos Backup Continuation Token Fix + +Fixed in version: **0.250.209** +Tracking issue: [#1258](https://github.com/microsoft/simplechat/issues/1258) + +## Issue Description + +Scheduled and manual Data Management backups silently omitted every Cosmos container that held more than one page of documents. Affected containers were recorded as failed resources and were absent from the backup artifact set entirely, while the job itself reported `completed_with_warnings`. + +In practice this meant the two largest containers in most deployments — `personal_conversations` and `personal_messages` — were never backed up, and the failure was easy to overlook because the overall job still reported success-with-warnings. + +Observed in App Service console logs: + +``` +[DATA_MANAGEMENT] Cosmos backup source page read failed. -- {'container': 'conversations', 'status_code': 400, + 'error': '(BadRequest) Invalid Continuation Token ActivityId: d009399c-468b-44b2-94a8-8c4515e0cf79, + Microsoft.Azure.Documents.Common/2.14.0'} +``` + +## Root Cause Analysis + +`_iter_backup_cosmos_source_items` rebuilt the Cosmos query on every page and replayed the previous pager's continuation token into the new pager: + +```python +continuation_token = None +while True: + source_iterable = container.query_items( # new query object per page + query="SELECT * FROM c ORDER BY c.id", + enable_cross_partition_query=True, + max_item_count=DATA_MANAGEMENT_BACKUP_MANIFEST_BATCH_SIZE, # 100 + ...) + page_iterator = source_iterable.by_page(continuation_token=continuation_token) + ... + continuation_token = getattr(page_iterator, "continuation_token", None) +``` + +A cross-partition Cosmos query is merged client side by the SDK. The continuation token surfaced by the pager belongs to the underlying per-partition request and is not valid for resuming the merged query. Rebuilding `query_items(...)` each iteration additionally discarded the SDK's cross-partition execution context, so the merge state could not be restored regardless of the token. + +Page one always succeeded; page two always failed with `BadRequest: Invalid Continuation Token`. Because `400` is non-retryable in `_is_retryable_backup_cosmos_error`, the resource failed immediately with zero retries. + +The 100-document boundary matched `DATA_MANAGEMENT_BACKUP_MANIFEST_BATCH_SIZE` exactly: + +| Container | Documents | Pages | Result | +|---|---|---|---| +| `public_documents` | 84 | 1 | completed | +| `personal_documents` | 59 | 1 | completed | +| `group_documents` | 39 | 1 | completed | +| `collaboration_messages` | 21 | 1 | completed | +| all other containers | <= 21 | 1 | completed | +| `personal_conversations` | > 100 | 2+ | failed | +| `personal_messages` | > 100 | 2+ | failed | + +### Contributing observability gaps + +Two logging gaps made the defect far harder to diagnose than it should have been: + +1. Per-item source blob transfer failures were never logged. `_run_backup_source_blob_transfer` folded every exception into a `failure_summary` string written only to the backup state document (last 50 entries) and the encrypted manifest. A run with 19,978 failed blobs produced zero log lines; a 47 MB console log export covering the whole backup window contained 72,465 `[DEBUG][Log]` lines and none containing "backup". +2. Application Insights carried no message text. `_build_logger_extra` reduced every string property to a length and the emitted trace message was always the constant `[SIMPLE_CHAT_LOG_EVENT]`, so traces could not identify which event fired. + +## Technical Details + +### Files Modified + +| File | Change | +|---|---| +| `application/single_app/functions_data_management.py` | Drain a single Cosmos pager; add source blob failure logging and bounded failure-reason rollup helpers | +| `application/single_app/functions_appinsights.py` | Retain sanitized message text and allowlisted diagnostic keys in log properties | +| `application/single_app/config.py` | Version bump to `0.250.209` | +| `functional_tests/test_data_management_backup_cosmos_pagination.py` | New regression coverage | + +### Code Changes Summary + +**Cosmos source pagination.** The query is now built once and a single pager is drained to exhaustion. Retries call `next()` on the same pager so the SDK's cross-partition execution context is preserved, and no continuation token is ever passed to `by_page()`. Page normalization, client-side sorting, cutoff filtering, cancellation checks, and RU/retry telemetry are unchanged. The non-paged fallback used by lightweight test doubles is retained and now shares the same page-finalization helper. + +**Source blob failure logging.** `_execute_backup_source_blob_resource` now logs the first failure for a resource, logs enumeration failures, and emits one rollup at resource completion containing failure counts and the distinct failure reasons. Two new helpers keep this bounded: + +- `_record_backup_failure_reason` tallies distinct reasons up to `DATA_MANAGEMENT_BACKUP_MAX_LOGGED_FAILURE_REASONS` (10) and aggregates the remainder into an `Other backup failures.` bucket. +- `_summarize_backup_failure_reasons` renders the tally as a single frequency-ordered string. + +This keeps 19,000 failures at a handful of log lines rather than one line per item. + +**Application Insights properties.** `_build_logger_extra` now emits `sc_message` containing the sanitized message text, and preserves the sanitized value of an explicit allowlist of non-sensitive diagnostic keys (`job_id`, `resource`, `resource_name`, `container`, `container_name`, `error`, `error_type`, `failure_summary`, `failure_reasons`, `index_name`, `operation`, `service`, `stage`, `status`, `step`, `task_name`, `reason`, `attempt`). Values are truncated to `LOGGER_SAFE_TEXT_MAX_LENGTH` (1024) and message text to `MAX_LOG_STRING_LENGTH` (8192). Sensitive keys still collapse to a `_present` boolean, all other strings still reduce to a length only, and `sanitize_log_message` redaction is unchanged. + +### Testing Approach + +`functional_tests/test_data_management_backup_cosmos_pagination.py` uses a fake pager that raises `400 Invalid Continuation Token` if any continuation token is supplied, reproducing the production failure directly. + +## Validation + +Focused suite: + +```bash +python -m pytest -q functional_tests/test_data_management_backup_cosmos_pagination.py +``` + +| Test | Assertion | +|---|---| +| `test_multi_page_cosmos_source_drains_a_single_pager` | 300 documents across 3 pages stream fully; query built exactly once; `by_page` only ever receives `None` | +| `test_cutoff_and_unpaged_fallback_still_stream` | Cutoff epoch drops exactly the out-of-range document; non-paged fallback streams every item | +| `test_failure_reason_rollup_is_bounded` | Distinct reasons stay bounded, overflow aggregates, summary is frequency-ordered | +| `test_logger_extra_retains_sanitized_diagnostics` | Message text and allowlisted diagnostics retained; secret value never emitted | +| `test_version_is_at_least_fix_version` | `config.py` version floor | + +Regression suite across all `functional_tests/test_data_management_*.py` files: **147 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` by stashing these changes and re-running it. It is unrelated to this fix. + +### Before / After + +| Behavior | Before | After | +|---|---|---| +| Container with <= 100 documents | Backed up | Backed up | +| Container with > 100 documents | `400 Invalid Continuation Token`, resource failed, absent from artifact set | Fully streamed and backed up | +| Blob transfer failures | No log output at all | First failure logged plus bounded rollup with counts and distinct reasons | +| App Insights trace | `[SIMPLE_CHAT_LOG_EVENT]` with lengths only | Sanitized message text plus allowlisted diagnostic values | + +## Cross-References + +- Issue: [#1258](https://github.com/microsoft/simplechat/issues/1258) +- Feature documentation: `docs/explanation/features/DATA_MANAGEMENT_BACKUP_MIGRATION.md` +- Functional test: `functional_tests/test_data_management_backup_cosmos_pagination.py` diff --git a/docs/explanation/release_notes.md b/docs/explanation/release_notes.md index 84ae66d20..7122cd9b5 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.209)** + +#### Bug Fixes + +* **Cosmos Backup Continuation Token Failure** + * Fixed Data Management backups silently omitting every Cosmos container that held more than one page of documents, which in most deployments meant personal conversations and personal messages were never backed up. + * Affected containers failed with `BadRequest: Invalid Continuation Token` and were dropped from the backup artifact set while the job still reported completion with warnings. + * Root cause was rebuilding the cross-partition query for each page and replaying the previous pager's continuation token; the backup now drains a single pager so the SDK's cross-partition execution context is preserved. + * (Ref: #1258, `functions_data_management.py`, Cosmos backup source paging) + +* **Missing Backup Failure Diagnostics** + * Source blob transfer failures previously produced no log output at all, so a run with nearly 20,000 failed blobs left no trace in App Service logs. + * Backups now log the first failure for each resource plus a bounded rollup of distinct failure reasons and counts when the resource finishes. + * (Ref: #1258, `functions_data_management.py`, source blob backup logging) + +* **Application Insights Log Message Text** + * Structured log events reached Application Insights as the constant `[SIMPLE_CHAT_LOG_EVENT]` with every string property reduced to a character count, making traces unusable for diagnosis. + * Traces now carry the sanitized message text and an allowlist of non-sensitive diagnostic values such as job ID, resource, container, status code, and error. Sensitive keys still collapse to a presence flag and secret redaction is unchanged. + * (Ref: #1258, `functions_appinsights.py`, log event properties) + ### **(v0.250.208)** #### User Interface Enhancements diff --git a/functional_tests/test_data_management_backup_cosmos_pagination.py b/functional_tests/test_data_management_backup_cosmos_pagination.py new file mode 100644 index 000000000..d7aac8987 --- /dev/null +++ b/functional_tests/test_data_management_backup_cosmos_pagination.py @@ -0,0 +1,355 @@ +#!/usr/bin/env python3 +# test_data_management_backup_cosmos_pagination.py +""" +Functional test for Cosmos backup source pagination and backup failure logging. +Version: 0.250.209 +Implemented in: 0.250.209 + +This test ensures Cosmos backup source reads drain a single cross-partition +pager instead of replaying a continuation token against a rebuilt query, which +previously failed every container holding more than one page with +"BadRequest: Invalid Continuation Token". It also covers the bounded failure +reason rollup used for source blob backup logging and the sanitized diagnostic +text now retained in Application Insights log properties. +""" + +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 + + +class FakeCosmosBadRequest(Exception): + """Mimic the non-retryable Cosmos error raised for replayed continuation tokens.""" + + def __init__(self, message="(BadRequest) Invalid Continuation Token"): + super().__init__(message) + self.status_code = 400 + + +class FakePageIterator: + """Serve ordered pages and reject any attempt to resume from a token.""" + + def __init__(self, pages, continuation_token=None): + if continuation_token is not None: + raise FakeCosmosBadRequest() + self._pages = list(pages) + self._index = 0 + self.continuation_token = "opaque-token" + + def __iter__(self): + return self + + def __next__(self): + if self._index >= len(self._pages): + raise StopIteration + page = self._pages[self._index] + self._index += 1 + return list(page) + + +class FakeItemPaged: + """Expose the by_page contract used by the production source reader.""" + + def __init__(self, pages, call_log): + self._pages = pages + self._call_log = call_log + + def by_page(self, continuation_token=None): + self._call_log.append(continuation_token) + return FakePageIterator(self._pages, continuation_token=continuation_token) + + def __iter__(self): + for page in self._pages: + for item in page: + yield item + + +class FakePagedContainer: + """Return a paged query result and count how often the query is rebuilt.""" + + def __init__(self, pages): + self._pages = pages + self.query_calls = 0 + self.by_page_tokens = [] + + def query_items(self, **kwargs): + self.query_calls += 1 + return FakeItemPaged(self._pages, self.by_page_tokens) + + +class FakeUnpagedContainer: + """Model lightweight test doubles that do not implement Cosmos paging.""" + + def __init__(self, items): + self._items = items + self.query_calls = 0 + + def query_items(self, **kwargs): + self.query_calls += 1 + return list(self._items) + + +ARTIFACT = { + "name": "personal_conversations", + "container_attr": "cosmos_conversations_container", + "container_name_attr": "cosmos_conversations_container_name", + "partition_key_path": "/id", + "category": "conversations", +} + + +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.209" + config_module.cosmos_data_management_jobs_container = None + config_module.cosmos_data_management_job_items_container = None + config_module.cosmos_settings_container = None + config_module.cosmos_conversations_container_name = "conversations" + sys.modules["config"] = config_module + + appinsights_module = types.ModuleType("functions_appinsights") + logged_events = [] + appinsights_module.log_event = ( + lambda message, extra=None, **_kwargs: logged_events.append((message, extra or {})) + ) + 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_pagination_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) + module.logged_events = logged_events + return module + + +def build_pages(page_count, page_size): + """Build sequential Cosmos-shaped pages with unique ids per document.""" + pages = [] + for page_index in range(page_count): + pages.append([ + { + "id": f"doc-{page_index * page_size + offset:05d}", + "_ts": 1700000000, + "_etag": f"etag-{page_index}-{offset}", + "payload": "value", + } + for offset in range(page_size) + ]) + return pages + + +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 + + +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 + + +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 + + +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 + + +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 + + +if __name__ == "__main__": + tests = [ + test_multi_page_cosmos_source_drains_a_single_pager, + test_cutoff_and_unpaged_fallback_still_stream, + test_failure_reason_rollup_is_bounded, + test_logger_extra_retains_sanitized_diagnostics, + test_version_is_at_least_fix_version, + ] + results = [] + 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)