Port durable batched search metadata sync to React v2

application/single_app/background_tasks.py

@@ -36,6 +36,7 @@
 )
 from functions_debug import debug_print
 from functions_data_management import check_due_data_management_jobs_once
+from functions_documents import process_due_document_search_metadata_syncs
 from functions_file_sync import check_due_file_sync_sources_once
 from functions_keyvault_reminders import (
     KEY_VAULT_SECRET_REMINDER_LOCK_NAME,
@@ -981,6 +982,36 @@
         time.sleep(60)
 
 
+def run_document_search_metadata_sync_loop():
+    """Retry search chunk metadata syncs that failed or never ran in their original process."""
+    while True:
+        lock_document = None
+        try:
+            lock_document = acquire_distributed_task_lock('document_search_metadata_sync_scan', lease_seconds=600)
+            if lock_document:
+                summary = process_due_document_search_metadata_syncs()
+                if summary.get('due'):
+                    log_event(
+                        '[DOCUMENT_SEARCH_SYNC] Reconciled due search metadata syncs.',
+                        extra=summary,
+                        level=logging.INFO,
+                    )
+            else:
+                debug_print('Skipping search metadata sync reconciliation because another worker holds the lease.')
+        except Exception as exc:
+            log_event(
+                '[DOCUMENT_SEARCH_SYNC] Search metadata sync reconciliation failed.',
+                extra={'error_type': type(exc).__name__},
+                level=logging.ERROR,
+                exceptionTraceback=True,
+            )
+        finally:
+            if lock_document:
+                release_distributed_task_lock(lock_document)
+
+        time.sleep(60)
+
+
 def run_content_screening_scheduler_loop(app=None):
     """Recover completion work even when enrollment of new scans is disabled."""
     while True:
@@ -1086,6 +1117,7 @@
         ('Orchestration scheduler background task started.', lambda: run_orchestration_scheduler_loop(app=app)),
         ('Workflow chat delivery background task started.', lambda: run_workflow_chat_delivery_loop(app=app)),
         ('Data Management scheduler background task started.', lambda: run_data_management_scheduler_loop(app=app)),
+        ('Document search metadata sync background task started.', run_document_search_metadata_sync_loop),
         ('Content screening scheduler background task started.', lambda: run_content_screening_scheduler_loop(app=app)),
         ('App maintenance background task started.', run_app_maintenance_loop),
         ('Key Vault secret reminder background task started.', run_key_vault_secret_reminder_loop),

application/single_app/config.py

@@ -101,7 +101,7 @@
 EXECUTOR_TYPE = 'thread'
 EXECUTOR_MAX_WORKERS = 30
 SESSION_TYPE = 'filesystem'
-VERSION = "0.261.267"
+VERSION = "0.261.268"
 IS_DEVELOPMENT = is_development_env_enabled()
 
 # Opt-out for deployments where App Service Easy Auth is active but the platform

application/single_app/functions_group_document_collaboration.py

@@ -437,6 +437,47 @@ def _search_acl(document, *, operation_guard):
         )
 
 
+def _search_acl_is_current(document):
+    """Whether every search chunk of the document carries the group access list Cosmos records."""
+    client = _get_search_client(group_id=document["group_id"])
+    escaped_id = document["id"].replace("'", "''")
+    available = _available(document) and is_current_group_document(document)
+    expected = sorted(set(document.get("shared_group_ids") or [])) if available else []
+    for chunk in client.search(search_text="*", filter=f"document_id eq '{escaped_id}'", select=["id", "shared_group_ids"]):
+        if sorted(set(chunk.get("shared_group_ids") or [])) != expected:
+            return False
+    return True
+
+
+def _repair_approved_share_search_acl(user_id, group_id, document):
+    """Re-run the search effect when an approved share's chunks miss the recorded access list.
+
+    Approving a share that is already approved completes an approval whose search update did not
+    finish, such as one made through the classic workspace or before this repair existed. Chunks
+    that already match are left alone, so a repeat has no effects. Returns receipt errors, if any.
+    """
+    def guard():
+        current = read_group_document_record(document["id"])
+        _require_share_action(user_id, group_id, current, "approve_share", group_id)
+        return current
+
+    try:
+        if _search_acl_is_current(document):
+            return []
+        _search_acl(document, operation_guard=guard)
+    except Exception as error:
+        log_event(
+            "[DOCUMENT_SHARING] An approved group share could not be projected to search again.",
+            extra={"document_id": document["id"], "group_id": document["group_id"], "exception_type": type(error).__name__},
+            level=logging.WARNING,
+        )
+        return [{
+            "stage": "search", "code": "search_repair_required",
+            "message": "The search update could not be confirmed. Refresh before retrying.",
+        }]
+    return []
+
+
 def _notification_effect(document, operation, *, operation_guard):
     target_id = operation["target_group_id"]
     source_id = document["group_id"]
@@ -606,6 +647,10 @@ def change_group_document_share(user_id, group_id, document_id, action, payload,
     )
     unchanged = action == "share" and before in {"approved", "not_approved"} or action == "approve_share" and before == "approved"
     if unchanged:
+        if action == "approve_share":
+            errors = _repair_approved_share_search_acl(user_id, group_id, document)
+            if errors:
+                return _operation_receipt(group_id, document, action, target_group_id, "partial", state, errors), 207
         return _operation_receipt(group_id, document, action, target_group_id, "unchanged", state, []), 200
     if action == "unshare" and before is None and existing and existing.get("target_group_id") == target_group_id and not _unfinished(existing):
         return _operation_receipt(group_id, document, action, target_group_id, "unchanged", "removed", []), 200

application/single_app/functions_group_document_management.py

@@ -32,6 +32,7 @@
     process_document_upload_background,
     process_metadata_extraction_background,
     select_current_documents,
+    summarize_document_search_sync,
     update_document,
     validate_tag_color,
     validate_tags,
@@ -272,9 +273,10 @@ def update_group_document_metadata(
         authorize_group_document_operation, user_id, group_id, document_id, operation,
         expected_version=document.get("version"),
     )
-    saved = update_document(
+    saved, search_sync = update_document(
         document_id=document_id, user_id=user_id, group_id=group_id,
-        strict=True, expected_etag=document.get("_etag"), operation_guard=guard, **changes,
+        strict=True, expected_etag=document.get("_etag"), operation_guard=guard,
+        return_search_sync=True, **changes,
     )
     if not isinstance(saved, dict) or saved.get("id") != document_id or saved.get("group_id") != group_id:
         raise DocumentMutationPropagationError("The scoped document update could not be confirmed.")
@@ -285,11 +287,13 @@ def update_group_document_metadata(
         updated_fields={name: "[updated]" for name in changes} if SCREENING_FIELD in document else changes,
         file_type=document.get("file_type"),
     )
-    # Metadata edits apply directly; they never start a new screening hold.
+    # Metadata edits apply directly; they never start a new screening hold. Search chunks pick the
+    # change up from a background sync, which search_sync reports.
     return {
         "message": "Group document metadata updated.",
         "document_id": document_id, "group_id": group_id,
         "updated_fields": list(changes), "status": "updated",
+        "search_sync": search_sync,
     }
 
 
@@ -313,6 +317,7 @@ def tag_group_documents(user_id, group_id, payload):
         # the whole batch with no document written.
         _ensure_document_tag_definitions(user_id, group_id, tags)
     result = {"success": [], "errors": []}
+    search_syncs = []
     for document_id in document_ids:
         try:
             document = authorize_group_document_operation(user_id, group_id, document_id, "tag_documents")
@@ -323,14 +328,16 @@ def tag_group_documents(user_id, group_id, payload):
                 updated_tags = [tag for tag in current_tags if normalize_tag(tag) not in tags]
             else:
                 updated_tags = tags
-            update_group_document_metadata(
+            receipt = update_group_document_metadata(
                 user_id, group_id, document_id, {"tags": updated_tags}, operation="tag_documents",
                 ensure_definitions=False,
             )
+            search_syncs.append(receipt.get("search_sync"))
             result["success"].append({"document_id": document_id, "tags": updated_tags})
         except Exception as error:
             failure, _status = group_operation_error(error, "tag_documents", document_id=document_id, group_id=group_id)
             result["errors"].append(failure)
+    result["search_sync"] = summarize_document_search_sync(search_syncs)
     return result, 207 if result["errors"] else 200
 
 
@@ -388,6 +395,7 @@ def change_group_document_tag(user_id, group_id, tag_name, payload=None, *, dele
             definition["color"] = _new_tag_definition(old_name, new_color)["color"]
         _patch_tag_definitions(user_id, group_id, group, {old_name: definition}, removals=old_keys)
         result["tag"] = {"name": old_name, "color": definition["color"]}
+        result["search_sync"] = summarize_document_search_sync([])
         return result, 200
 
     changes = {}
@@ -403,6 +411,7 @@ def change_group_document_tag(user_id, group_id, tag_name, payload=None, *, dele
         changes[new_name] = target_definition
         result["tag"] = {"name": new_name, "color": target_definition.get("color") or get_default_tag_color(new_name)}
     group = _patch_tag_definitions(user_id, group_id, group, changes)
+    search_syncs = []
     for target in targets:
         document_id = target["id"]
         try:
@@ -414,14 +423,16 @@ def change_group_document_tag(user_id, group_id, tag_name, payload=None, *, dele
                 if not (delete and normalize_tag(tag) == old_name)
             ]
             updated_tags = list(dict.fromkeys(updated_tags))
-            update_group_document_metadata(
+            receipt = update_group_document_metadata(
                 user_id, group_id, document_id, {"tags": updated_tags}, operation="tag_documents", ensure_definitions=False,
             )
+            search_syncs.append(receipt.get("search_sync"))
             result["success"].append({"document_id": document_id, "tags": updated_tags})
         except Exception as error:
             failure, _status = group_operation_error(error, "manage_tags", document_id=document_id, group_id=group_id)
             result["errors"].append(failure)
     result["documents_updated"] = len(result["success"])
+    result["search_sync"] = summarize_document_search_sync(search_syncs)
     if not result["errors"]:
         try:
             _patch_tag_definitions(user_id, group_id, group, {}, removals=old_keys)

application/single_app/functions_public_document_management.py

@@ -37,6 +37,7 @@
     process_document_upload_background,
     process_metadata_extraction_background,
     select_current_documents,
+    summarize_document_search_sync,
     update_document,
     validate_tag_color,
     validate_tags,
@@ -278,9 +279,10 @@ def update_public_document_metadata(
         authorize_public_document_operation, user_id, workspace_id, document_id, operation,
         expected_version=document.get("version"),
     )
-    saved = update_document(
+    saved, search_sync = update_document(
         document_id=document_id, user_id=user_id, public_workspace_id=workspace_id,
-        strict=True, expected_etag=document.get("_etag"), operation_guard=guard, **changes,
+        strict=True, expected_etag=document.get("_etag"), operation_guard=guard,
+        return_search_sync=True, **changes,
     )
     if not isinstance(saved, dict) or saved.get("id") != document_id or saved.get("public_workspace_id") != workspace_id:
         raise DocumentMutationPropagationError("The scoped document update could not be confirmed.")
@@ -291,11 +293,13 @@ def update_public_document_metadata(
         updated_fields={name: "[updated]" for name in changes} if SCREENING_FIELD in document else changes,
         file_type=document.get("file_type"),
     )
-    # Metadata edits apply directly; they never start a new screening hold.
+    # Metadata edits apply directly; they never start a new screening hold. Search chunks pick the
+    # change up from a background sync, which search_sync reports.
     return {
         "message": "Public document metadata updated.",
         "document_id": document_id, "public_workspace_id": workspace_id,
         "updated_fields": list(changes), "status": "updated",
+        "search_sync": search_sync,
     }
 
 
@@ -319,6 +323,7 @@ def tag_public_documents(user_id, workspace_id, payload):
         # the whole batch with no document written.
         _ensure_document_tag_definitions(user_id, workspace_id, tags)
     result = {"success": [], "errors": []}
+    search_syncs = []
     for document_id in document_ids:
         try:
             document = authorize_public_document_operation(user_id, workspace_id, document_id, "tag_documents")
@@ -329,14 +334,16 @@ def tag_public_documents(user_id, workspace_id, payload):
                 updated_tags = [tag for tag in current_tags if normalize_tag(tag) not in tags]
             else:
                 updated_tags = tags
-            update_public_document_metadata(
+            receipt = update_public_document_metadata(
                 user_id, workspace_id, document_id, {"tags": updated_tags}, operation="tag_documents",
                 ensure_definitions=False,
             )
+            search_syncs.append(receipt.get("search_sync"))
             result["success"].append({"document_id": document_id, "tags": updated_tags})
         except Exception as error:
             failure, _status = public_operation_error(error, "tag_documents", document_id=document_id, workspace_id=workspace_id)
             result["errors"].append(failure)
+    result["search_sync"] = summarize_document_search_sync(search_syncs)
     return result, 207 if result["errors"] else 200
 
 
@@ -394,6 +401,7 @@ def change_public_document_tag(user_id, workspace_id, tag_name, payload=None, *,
             definition["color"] = _new_tag_definition(old_name, new_color)["color"]
         _patch_tag_definitions(user_id, workspace_id, workspace, {old_name: definition}, removals=old_keys)
         result["tag"] = {"name": old_name, "color": definition["color"]}
+        result["search_sync"] = summarize_document_search_sync([])
         return result, 200
 
     changes = {}
@@ -409,6 +417,7 @@ def change_public_document_tag(user_id, workspace_id, tag_name, payload=None, *,
         changes[new_name] = target_definition
         result["tag"] = {"name": new_name, "color": target_definition.get("color") or get_default_tag_color(new_name)}
     workspace = _patch_tag_definitions(user_id, workspace_id, workspace, changes)
+    search_syncs = []
     for target in targets:
         document_id = target["id"]
         try:
@@ -420,14 +429,16 @@ def change_public_document_tag(user_id, workspace_id, tag_name, payload=None, *,
                 if not (delete and normalize_tag(tag) == old_name)
             ]
             updated_tags = list(dict.fromkeys(updated_tags))
-            update_public_document_metadata(
+            receipt = update_public_document_metadata(
                 user_id, workspace_id, document_id, {"tags": updated_tags}, operation="tag_documents", ensure_definitions=False,
             )
+            search_syncs.append(receipt.get("search_sync"))
             result["success"].append({"document_id": document_id, "tags": updated_tags})
         except Exception as error:
             failure, _status = public_operation_error(error, "manage_tags", document_id=document_id, workspace_id=workspace_id)
             result["errors"].append(failure)
     result["documents_updated"] = len(result["success"])
+    result["search_sync"] = summarize_document_search_sync(search_syncs)
     if not result["errors"]:
         try:
             _patch_tag_definitions(user_id, workspace_id, workspace, {}, removals=old_keys)