Atomic Search Param operations

docs/arch/adr-2602-atomic-searchparameter-crud-and-cache-refresh-ownership.md

@@ -0,0 +1,60 @@
+# ADR 2602: Atomic SearchParameter CRUD Operations
+Labels: [SQL](https://github.com/microsoft/fhir-server/labels/Area-SQL) | [Core](https://github.com/microsoft/fhir-server/labels/Area-Core) | [SearchParameter](https://github.com/microsoft/fhir-server/labels/Area-SearchParameter)
+
+## Context
+SearchParameter create, update, and delete operations require two coordinated writes: the SearchParameter status row (`dbo.SearchParam` table) and the resource itself (`dbo.Resource` via `dbo.MergeResources`). Previously, these writes occurred in separate steps in the request pipeline, creating partial-failure windows where one could succeed while the other fails — producing orphaned or inconsistent state.
+
+Composing these writes into a single atomic SQL operation introduced a new problem for transaction bundles: `dbo.MergeSearchParams` acquires an exclusive lock on `dbo.SearchParam` per entry, and that lock is held by the outer bundle transaction. When the next entry's behavior pipeline calls `GetAllSearchParameterStatus` on a separate connection, it blocks on the same table, causing a timeout. This required a deferred-flush approach for transaction bundles.
+
+Additionally, SearchParameter CRUD behaviors previously performed direct in-memory cache mutations (`AddNewSearchParameters`, `DeleteSearchParameter`, etc.) during the request pipeline. This duplicated responsibility with the `SearchParameterCacheRefreshBackgroundService`, which already polls the database and applies cache updates across all instances.
+
+Key considerations:
+- Eliminating partial-commit windows between status and resource persistence.
+- Handling the lock contention on `dbo.SearchParam` introduced by composed writes in transaction bundles.
+- Simplifying cache ownership by removing direct cache mutations from the CRUD request path.
+- Preserving existing behavior for non-SearchParameter resources and Cosmos DB paths.
+
+## Decision
+We implement three complementary changes:
+
+### 1. Composed writes for single operations (SQL)
+For individual SearchParameter CRUD, behaviors queue pending status updates in request context (`SearchParameter.PendingStatusUpdates`) instead of persisting directly. `SqlServerFhirDataStore` detects pending statuses and calls `dbo.MergeSearchParams` (which internally calls `dbo.MergeResources`) so both writes execute in one stored-procedure-owned transaction.
+
+### 2. Deferred flush for transaction bundles
+For transaction bundles, per-entry resource upserts call `dbo.MergeResources` only (no SearchParam table touch), avoiding the exclusive lock. `BundleHandler` accumulates pending statuses across all entries and flushes them in a single `dbo.MergeSearchParams` call at the end of the bundle, still within the outer transaction scope.
+
+```mermaid
+graph TD;
+  A[SearchParameter Operation] -->|Single operation| B[Behavior queues pending status in request context];
+  B --> C[SqlServerFhirDataStore calls MergeSearchParams];
+  C --> D[MergeSearchParams: status + MergeResources in one transaction];
+  A -->|Transaction bundle| E[Per-entry: Behavior queues status, UpsertAsync calls MergeResources only];
+  E --> F[BundleHandler drains pending statuses after each entry];
+  F --> G[After all entries: single MergeSearchParams flush within outer transaction];
+  G --> H[Transaction commits atomically];
+```
+
+### 3. Cache update removal from CRUD path
+SearchParameter CRUD behaviors no longer perform direct in-memory cache mutations. All cache updates (`AddNewSearchParameters`, `DeleteSearchParameter`, `UpdateSearchParameterStatus`) are now solely owned by the `SearchParameterCacheRefreshBackgroundService`, which periodically polls the database via `GetAndApplySearchParameterUpdates`. This simplifies the CRUD path and ensures consistent cache convergence across distributed instances.
+
+### Scope
+- **SQL Server**: Full atomic guarantees for create, update, and delete of SearchParameter resources, both single and in transaction bundles.
+- **Cosmos DB**: Pending statuses are flushed after resource upsert (improved sequencing, not a single transactional unit).
+- **Unchanged**: Non-SearchParameter CRUD, existing SearchParameter status lifecycle states, cache convergence model.
+
+## Status
+Pending acceptance
+
+## Consequences
+- **Positive Impacts:**
+  - Eliminates orphaned status/resource records from partial commits.
+  - Clarifies ownership: behaviors queue intent, data stores persist atomically, background service owns cache.
+  - Deferred-flush approach avoids lock contention introduced by composed writes in transaction bundles.
+  - Removing cache mutations from CRUD simplifies the request path and eliminates a class of cache-divergence bugs.
+
+- **Potential Drawbacks:**
+  - Increased complexity in request context, data store, and bundle handler coordination.
+  - SQL schema migration required (`MergeSearchParams` expanded to accept resource TVPs).
+  - Eventual consistency window: cache may lag behind database until the next background refresh cycle.
+  - Cosmos DB path remains best-effort sequencing rather than true atomic commit.
+

src/Microsoft.Health.Fhir.Core/Features/Definition/SearchParameterDefinitionBuilder.cs

@@ -1,4 +1,4 @@
-// -------------------------------------------------------------------------------------------------
+// -------------------------------------------------------------------------------------------------
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License (MIT). See LICENSE in the repo root for license information.
 // -------------------------------------------------------------------------------------------------
@@ -63,7 +63,8 @@ internal static void Build(
                 searchParameters,
                 uriDictionary,
                 modelInfoProvider,
-                isSystemDefined).ToLookup(
+                isSystemDefined,
+                logger).ToLookup(
                     entry => entry.ResourceType,
                     entry => entry.SearchParameter);
 
@@ -121,7 +122,8 @@ private static SearchParameterInfo GetOrCreateSearchParameterInfo(SearchParamete
             IReadOnlyCollection<ITypedElement> searchParamCollection,
             ConcurrentDictionary<string, SearchParameterInfo> uriDictionary,
             IModelInfoProvider modelInfoProvider,
-            bool isSystemDefined = false)
+            bool isSystemDefined,
+            ILogger logger)
         {
             var issues = new List<OperationOutcomeIssue>();
             var searchParameters = searchParamCollection.Select((x, entryIndex) =>
@@ -151,8 +153,27 @@ private static SearchParameterInfo GetOrCreateSearchParameterInfo(SearchParamete
                 {
                     SearchParameterInfo searchParameterInfo = GetOrCreateSearchParameterInfo(searchParameter, uriDictionary);
 
-                    // Mark spec-defined search parameters as system-defined
-                    searchParameterInfo.IsSystemDefined = isSystemDefined;
+                    // Mark spec-defined search parameters as system-defined.
+                    // Once marked, this should remain true across subsequent Build calls.
+                    bool wasSystemDefined = searchParameterInfo.IsSystemDefined;
+                    searchParameterInfo.IsSystemDefined |= isSystemDefined;
+
+                    if (!wasSystemDefined && searchParameterInfo.IsSystemDefined)
+                    {
+                        logger.LogDebug(
+                            "SearchParameter IsSystemDefined enabled: Url={Url}, Code={Code}, BuildIsSystemDefined={BuildIsSystemDefined}",
+                            searchParameterInfo.Url?.OriginalString,
+                            searchParameterInfo.Code,
+                            isSystemDefined);
+                    }
+                    else if (wasSystemDefined && !isSystemDefined)
+                    {
+                        logger.LogWarning(
+                            "SearchParameter IsSystemDefined downgrade ignored: Url={Url}, Code={Code}, BuildIsSystemDefined={BuildIsSystemDefined}",
+                            searchParameterInfo.Url?.OriginalString,
+                            searchParameterInfo.Code,
+                            isSystemDefined);
+                    }
 
                     if (searchParameterInfo.Code == "_profile" && searchParameterInfo.Type == SearchParamType.Reference)
                     {
@@ -174,6 +195,8 @@ private static SearchParameterInfo GetOrCreateSearchParameterInfo(SearchParamete
 
             EnsureNoIssues();
 
+            SearchParameterInfo.ResourceTypeSearchParameter.IsSystemDefined = true;
+
             var validatedSearchParameters = new List<(string ResourceType, SearchParameterInfo SearchParameter)>
             {
                 // _type is currently missing from the search params definition bundle, so we inject it in here.

src/Microsoft.Health.Fhir.Core/Features/Operations/Reindex/ReindexOrchestratorJob.cs

@@ -77,6 +77,12 @@ public sealed class ReindexOrchestratorJob : IJob
         /// </summary>
         private static readonly AsyncPolicy _searchParameterStatusRetries = Policy.WrapAsync(_requestRateRetries, _timeoutRetries);
 
+        /// <summary>
+        /// Retry policy for reindex query execution.
+        /// Handles transient SQL timeouts and Cosmos DB request rate limiting.
+        /// </summary>
+        private static readonly AsyncPolicy _reindexQueryRetries = Policy.WrapAsync(_requestRateRetries, _timeoutRetries);
+
         private HashSet<long> _processedJobIds = new HashSet<long>();
         private HashSet<string> _processedSearchParameters = new HashSet<string>();
         private List<JobInfo> _jobsToProcess;
@@ -727,7 +733,8 @@ private async Task<SearchResult> GetResourceCountForQueryAsync(ReindexJobQuerySt
             {
                 try
                 {
-                    return await searchService.Value.SearchForReindexAsync(queryParametersList, searchParameterHash, countOnly: countOnly, cancellationToken, true);
+                    return await _reindexQueryRetries.ExecuteAsync(
+                        async () => await searchService.Value.SearchForReindexAsync(queryParametersList, searchParameterHash, countOnly: countOnly, cancellationToken, true));
                 }
                 catch (Exception ex)
                 {

src/Microsoft.Health.Fhir.Core/Features/Persistence/ResourceWrapperOperation.cs

@@ -3,7 +3,9 @@
 // Licensed under the MIT License (MIT). See LICENSE in the repo root for license information.
 // -------------------------------------------------------------------------------------------------
 
+using System.Collections.Generic;
 using EnsureThat;
+using Microsoft.Health.Fhir.Core.Features.Search.Registry;
 using Microsoft.Health.Fhir.Core.Models;
 
 namespace Microsoft.Health.Fhir.Core.Features.Persistence
@@ -46,6 +48,8 @@ public ResourceWrapperOperation(
 
         public BundleResourceContext BundleResourceContext { get; }
 
+        public IReadOnlyList<ResourceSearchParameterStatus> PendingSearchParameterStatuses { get; internal set; }
+
 #pragma warning disable CA1024 // Use properties where appropriate
         public DataStoreOperationIdentifier GetIdentifier()
         {

src/Microsoft.Health.Fhir.Core/Features/Search/Parameters/CreateOrUpdateSearchParameterBehavior.cs

@@ -4,12 +4,19 @@
 // -------------------------------------------------------------------------------------------------
 
 using System;
+using System.Collections.Generic;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using EnsureThat;
+using Hl7.Fhir.ElementModel;
 using MediatR;
+using Microsoft.Health.Core.Features.Context;
 using Microsoft.Health.Fhir.Core.Exceptions;
+using Microsoft.Health.Fhir.Core.Extensions;
+using Microsoft.Health.Fhir.Core.Features.Context;
 using Microsoft.Health.Fhir.Core.Features.Persistence;
+using Microsoft.Health.Fhir.Core.Features.Search.Registry;
 using Microsoft.Health.Fhir.Core.Messages.Create;
 using Microsoft.Health.Fhir.Core.Messages.Upsert;
 using Microsoft.Health.Fhir.Core.Models;
@@ -19,25 +26,41 @@ namespace Microsoft.Health.Fhir.Core.Features.Search.Parameters
     public class CreateOrUpdateSearchParameterBehavior<TCreateResourceRequest, TUpsertResourceResponse> : IPipelineBehavior<CreateResourceRequest, UpsertResourceResponse>,
         IPipelineBehavior<UpsertResourceRequest, UpsertResourceResponse>
     {
-        private ISearchParameterOperations _searchParameterOperations;
-        private IFhirDataStore _fhirDataStore;
+        private readonly ISearchParameterOperations _searchParameterOperations;
+        private readonly IFhirDataStore _fhirDataStore;
+        private readonly ISearchParameterStatusManager _searchParameterStatusManager;
+        private readonly RequestContextAccessor<IFhirRequestContext> _requestContextAccessor;
+        private readonly IModelInfoProvider _modelInfoProvider;
 
-        public CreateOrUpdateSearchParameterBehavior(ISearchParameterOperations searchParameterOperations, IFhirDataStore fhirDataStore)
+        public CreateOrUpdateSearchParameterBehavior(
+            ISearchParameterOperations searchParameterOperations,
+            IFhirDataStore fhirDataStore,
+            ISearchParameterStatusManager searchParameterStatusManager,
+            RequestContextAccessor<IFhirRequestContext> requestContextAccessor,
+            IModelInfoProvider modelInfoProvider)
         {
             EnsureArg.IsNotNull(searchParameterOperations, nameof(searchParameterOperations));
             EnsureArg.IsNotNull(fhirDataStore, nameof(fhirDataStore));
+            EnsureArg.IsNotNull(searchParameterStatusManager, nameof(searchParameterStatusManager));
+            EnsureArg.IsNotNull(requestContextAccessor, nameof(requestContextAccessor));
+            EnsureArg.IsNotNull(modelInfoProvider, nameof(modelInfoProvider));
 
             _searchParameterOperations = searchParameterOperations;
             _fhirDataStore = fhirDataStore;
+            _searchParameterStatusManager = searchParameterStatusManager;
+            _requestContextAccessor = requestContextAccessor;
+            _modelInfoProvider = modelInfoProvider;
         }
 
         public async Task<UpsertResourceResponse> Handle(CreateResourceRequest request, RequestHandlerDelegate<UpsertResourceResponse> next, CancellationToken cancellationToken)
         {
             if (request.Resource.InstanceType.Equals(KnownResourceTypes.SearchParameter, StringComparison.Ordinal))
             {
-                // Before committing the SearchParameter resource to the data store, add it to the SearchParameterDefinitionManager
-                // and parse the fhirPath, as well as validate the parameter type
-                await _searchParameterOperations.AddSearchParameterAsync(request.Resource.Instance, cancellationToken);
+                // Before committing the SearchParameter resource to the data store, validate the parameter type
+                await _searchParameterOperations.ValidateSearchParameterAsync(request.Resource.Instance, cancellationToken);
+
+                var url = request.Resource.Instance.GetStringScalar("url");
+                await QueueStatusAsync(url, SearchParameterStatus.Supported, cancellationToken);
             }
 
             // Allow the resource to be updated with the normal handler
@@ -67,19 +90,70 @@ public async Task<UpsertResourceResponse> Handle(UpsertResourceRequest request,
 
                 if (prevSearchParamResource != null && prevSearchParamResource.IsDeleted == false)
                 {
-                    // Update the SearchParameterDefinitionManager with the new SearchParameter in order to validate any changes
-                    // to the fhirpath or the datatype
-                    await _searchParameterOperations.UpdateSearchParameterAsync(request.Resource.Instance, prevSearchParamResource.RawResource, cancellationToken);
+                    // Validate any changes to the fhirpath or the datatype
+                    await _searchParameterOperations.ValidateSearchParameterAsync(request.Resource.Instance, cancellationToken);
+
+                    var previousUrl = _modelInfoProvider.ToTypedElement(prevSearchParamResource.RawResource).GetStringScalar("url");
+                    var newUrl = request.Resource.Instance.GetStringScalar("url");
+
+                    if (!string.IsNullOrWhiteSpace(previousUrl) && !previousUrl.Equals(newUrl, StringComparison.Ordinal))
+                    {
+                        await QueueStatusAsync(previousUrl, SearchParameterStatus.Deleted, cancellationToken);
+                    }
+
+                    await QueueStatusAsync(newUrl, SearchParameterStatus.Supported, cancellationToken);
                 }
                 else
                 {
                     // No previous version exists or it was deleted, so add it as a new SearchParameter
-                    await _searchParameterOperations.AddSearchParameterAsync(request.Resource.Instance, cancellationToken);
+                    await _searchParameterOperations.ValidateSearchParameterAsync(request.Resource.Instance, cancellationToken);
+
+                    var url = request.Resource.Instance.GetStringScalar("url");
+                    await QueueStatusAsync(url, SearchParameterStatus.Supported, cancellationToken);
                 }
             }
 
             // Now allow the resource to updated per the normal behavior
             return await next(cancellationToken);
         }
+
+        private async Task QueueStatusAsync(string url, SearchParameterStatus status, CancellationToken cancellationToken)
+        {
+            if (string.IsNullOrWhiteSpace(url))
+            {
+                return;
+            }
+
+            var context = _requestContextAccessor.RequestContext;
+            if (context == null)
+            {
+                return;
+            }
+
+            if (!context.Properties.TryGetValue(SearchParameterRequestContextPropertyNames.PendingStatusUpdates, out var value) ||
+                value is not List<ResourceSearchParameterStatus> pendingStatuses)
+            {
+                pendingStatuses = new List<ResourceSearchParameterStatus>();
+                context.Properties[SearchParameterRequestContextPropertyNames.PendingStatusUpdates] = pendingStatuses;
+            }
+
+            var currentStatuses = await _searchParameterStatusManager.GetAllSearchParameterStatus(cancellationToken);
+            var existing = currentStatuses.FirstOrDefault(s => string.Equals(s.Uri?.OriginalString, url, StringComparison.Ordinal));
+
+            var update = new ResourceSearchParameterStatus
+            {
+                Uri = new Uri(url),
+                Status = status,
+                LastUpdated = existing?.LastUpdated ?? DateTimeOffset.UtcNow,
+                IsPartiallySupported = existing?.IsPartiallySupported ?? false,
+                SortStatus = existing?.SortStatus ?? SortParameterStatus.Disabled,
+            };
+
+            lock (pendingStatuses)
+            {
+                pendingStatuses.RemoveAll(s => string.Equals(s.Uri?.OriginalString, url, StringComparison.Ordinal));
+                pendingStatuses.Add(update);
+            }
+        }
     }
 }