Sync document search metadata in the background with batched merges

application/single_app/background_tasks.py

@@ -34,6 +34,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,
@@ -881,6 +882,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_app_maintenance_loop():
     """Run idempotent app maintenance tasks under a distributed lock."""
     while True:
@@ -956,6 +987,7 @@
         ('File Sync scheduler background task started.', run_file_sync_scheduler_loop),
         ('Tabular generated-output scheduler background task started.', run_tabular_generated_output_scheduler_loop),
         ('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),
         ('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

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

application/single_app/route_backend_group_documents.py

@@ -127,6 +127,15 @@
     ]
 
 
+def _group_share_search_acl_error_response(error):
+    """Return the safe response for a sharing change the search index could not enforce."""
+    response = jsonify({'error': str(error)})
+    if isinstance(error, DocumentSearchAclProjectionDeferredError):
+        response.headers['Retry-After'] = '150'
+        return response, 503
+    return response, 500
+
+
 def _get_group_name(group_doc, fallback='Unknown Group'):
     return str((group_doc or {}).get('name') or fallback).strip()
 
@@ -942,79 +951,40 @@
         if error_response:
             return error_response
 
-        data = request.get_json()
+        data = request.get_json() or {}
 
         # Track which fields were updated
         updated_fields = {}
+        document_updates = {}
 
         try:
             if 'title' in data:
-                update_document(
-                    document_id=document_id,
-                    group_id=active_group_id,
-                    user_id=user_id,
-                    title=data['title']
-                )
+                document_updates['title'] = data['title']
                 updated_fields['title'] = data['title']
             if 'abstract' in data:
-                update_document(
-                    document_id=document_id,
-                    group_id=active_group_id,
-                    user_id=user_id,
-                    abstract=data['abstract']
-                )
+                document_updates['abstract'] = data['abstract']
                 updated_fields['abstract'] = data['abstract']
             if 'keywords' in data:
                 if isinstance(data['keywords'], list):
-                    update_document(
-                        document_id=document_id,
-                        group_id=active_group_id,
-                        user_id=user_id,
-                        keywords=data['keywords']
-                    )
+                    document_updates['keywords'] = data['keywords']
                     updated_fields['keywords'] = data['keywords']
                 else:
                     keywords_list = [kw.strip() for kw in data['keywords'].split(',')]
-                    update_document(
-                        document_id=document_id,
-                        group_id=active_group_id,
-                        user_id=user_id,
-                        keywords=keywords_list
-                    )
+                    document_updates['keywords'] = keywords_list
                     updated_fields['keywords'] = keywords_list
             if 'publication_date' in data:
-                update_document(
-                    document_id=document_id,
-                    group_id=active_group_id,
-                    user_id=user_id,
-                    publication_date=data['publication_date']
-                )
+                document_updates['publication_date'] = data['publication_date']
                 updated_fields['publication_date'] = data['publication_date']
             if 'document_classification' in data:
-                update_document(
-                    document_id=document_id,
-                    group_id=active_group_id,
-                    user_id=user_id,
-                    document_classification=data['document_classification']
-                )
+                document_updates['document_classification'] = data['document_classification']
                 updated_fields['document_classification'] = data['document_classification']
             if 'authors' in data:
                 if isinstance(data['authors'], list):
-                    update_document(
-                        document_id=document_id,
-                        group_id=active_group_id,
-                        user_id=user_id,
-                        authors=data['authors']
-                    )
+                    document_updates['authors'] = data['authors']
                     updated_fields['authors'] = data['authors']
                 else:
                     authors_list = [data['authors']]
-                    update_document(
-                        document_id=document_id,
-                        group_id=active_group_id,
-                        user_id=user_id,
-                        authors=authors_list
-                    )
+                    document_updates['authors'] = authors_list
                     updated_fields['authors'] = authors_list
 
             if 'tags' in data:
@@ -1025,13 +995,18 @@
                     return jsonify({'error': error_msg}), 400
                 for tag in normalized_tags:
                     get_or_create_tag_definition(user_id, tag, workspace_type='group', group_id=active_group_id)
-                update_document(
+                document_updates['tags'] = normalized_tags
+                updated_fields['tags'] = normalized_tags
+
+            # Save every change in one update; search chunks are synced in the background
+            update_result = {}
+            if document_updates:
+                update_result = update_document(
                     document_id=document_id,
                     group_id=active_group_id,
                     user_id=user_id,
-                    tags=normalized_tags
-                )
-                updated_fields['tags'] = normalized_tags
+                    **document_updates,
+                ) or {}
 
             # Log the metadata update transaction if any fields were updated
             if updated_fields:
@@ -1060,9 +1035,18 @@
                         group_id=active_group_id
                     )
 
-            return jsonify({'message': 'Group document metadata updated successfully'}), 200
+            response_payload = {'message': 'Group document metadata updated successfully'}
+            if isinstance(update_result, dict) and update_result.get('search_sync'):
+                response_payload['search_sync'] = update_result['search_sync']
+
+            return jsonify(response_payload), 200
         except Exception as e:
-            return jsonify({'error': str(e)}), 500
+            log_event(
+                '[DOCUMENT_METADATA] Failed to update group document metadata',
+                {'document_id': document_id, 'group_id': active_group_id, 'error_type': type(e).__name__},
+                level=logging.ERROR,
+            )
+            return jsonify({'error': 'Unable to update document metadata'}), 500
 
     @bp.route('/api/group_documents/<document_id>', methods=['DELETE'])
     @swagger_route(security=get_auth_security())
@@ -1527,6 +1511,8 @@
                 invalidate_group_search_cache(active_group_id)
 
             return jsonify({'message': 'Share approved' if updated else 'Already approved'}), 200
+        except (DocumentSearchAclProjectionDeferredError, DocumentSearchAclProjectionError) as exc:
+            return _group_share_search_acl_error_response(exc)
         except Exception as e:
             return jsonify({'error': f'Error approving shared document: {str(e)}'}), 500
 
@@ -1895,6 +1881,8 @@
                 'document_id': document_id,
                 'shared_with_group': target_group_id
             }), 200
+        except (DocumentSearchAclProjectionDeferredError, DocumentSearchAclProjectionError) as exc:
+            return _group_share_search_acl_error_response(exc)
         except Exception as e:
             return jsonify({'error': f'Error sharing document: {str(e)}'}), 500
 
@@ -1964,6 +1952,8 @@
                 'document_id': document_id,
                 'unshared_with_group': target_group_id
             }), 200
+        except (DocumentSearchAclProjectionDeferredError, DocumentSearchAclProjectionError) as exc:
+            return _group_share_search_acl_error_response(exc)
         except Exception as e:
             return jsonify({'error': f'Error unsharing document: {str(e)}'}), 500
 
@@ -2041,6 +2031,8 @@
             return jsonify({
                 'message': 'Share denied' if was_pending else 'Successfully removed group from shared document'
             }), 200
+        except (DocumentSearchAclProjectionDeferredError, DocumentSearchAclProjectionError) as exc:
+            return _group_share_search_acl_error_response(exc)
         except Exception as e:
             return jsonify({'error': f'Error removing group from shared document: {str(e)}'}), 500
 
@@ -2218,7 +2210,7 @@
 
         from functions_documents import (
             validate_tags, update_document,
-            propagate_tags_to_chunks, get_or_create_tag_definition
+            get_or_create_tag_definition
         )
 
         is_valid, error_msg, normalized_tags = validate_tags(tags_input)
@@ -2268,18 +2260,14 @@
                     elif action == 'set_tags':
                         new_tags = normalized_tags
 
+                    # Search chunks and blob tags sync in the background
                     update_document(
                         document_id=doc_id,
                         group_id=active_group_id,
                         user_id=user_id,
                         tags=new_tags
                     )
 
-                    try:
-                        propagate_tags_to_chunks(doc_id, new_tags, user_id, group_id=active_group_id)
-                    except Exception:
-                        pass
-
                     results['success'].append({
                         'document_id': doc_id,
                         'tags': new_tags
@@ -2331,7 +2319,7 @@
         new_name = data.get('new_name')
         new_color = data.get('color')
 
-        from functions_documents import normalize_tag, validate_tag_color, validate_tags, update_document, propagate_tags_to_chunks
+        from functions_documents import normalize_tag, validate_tag_color, validate_tags, update_document
 
         try:
             normalized_old_tag = normalize_tag(tag_name)
@@ -2363,18 +2351,14 @@
                         current_tags = doc['tags']
                         new_tags = [normalized_new_tag if t == normalized_old_tag else t for t in current_tags]
 
+                        # Search chunks and blob tags sync in the background
                         update_document(
                             document_id=doc['id'],
                             group_id=active_group_id,
                             user_id=user_id,
                             tags=new_tags
                         )
 
-                        try:
-                            propagate_tags_to_chunks(doc['id'], new_tags, user_id, group_id=active_group_id)
-                        except Exception:
-                            pass
-
                         updated_count += 1
 
                 tag_defs = group_doc.get('tag_definitions', {})
@@ -2442,7 +2426,7 @@
         if error_response:
             return error_response
 
-        from functions_documents import normalize_tag, update_document, propagate_tags_to_chunks
+        from functions_documents import normalize_tag, update_document
 
         try:
             normalized_tag = normalize_tag(tag_name)
@@ -2466,18 +2450,14 @@
                 if normalized_tag in doc.get('tags', []):
                     new_tags = [t for t in doc['tags'] if t != normalized_tag]
 
+                    # Search chunks and blob tags sync in the background
                     update_document(
                         document_id=doc['id'],
                         group_id=active_group_id,
                         user_id=user_id,
                         tags=new_tags
                     )
 
-                    try:
-                        propagate_tags_to_chunks(doc['id'], new_tags, user_id, group_id=active_group_id)
-                    except Exception:
-                        pass
-
                     updated_count += 1
 
             tag_defs = group_doc.get('tag_definitions', {})