Atomic Search Param operations

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

@@ -0,0 +1,60 @@
+# ADR 2603: 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.UnitTests/Features/Operations/BulkDelete/BulkDeleteProcessingJobTests.cs

@@ -15,7 +15,9 @@
 using Microsoft.Health.Fhir.Core.Features.Operations.BulkDelete;
 using Microsoft.Health.Fhir.Core.Features.Persistence;
 using Microsoft.Health.Fhir.Core.Features.Search;
+using Microsoft.Health.Fhir.Core.Features.Search.Parameters;
 using Microsoft.Health.Fhir.Core.Messages.Delete;
+using Microsoft.Health.Fhir.Core.Models;
 using Microsoft.Health.Fhir.Core.UnitTests.Extensions;
 using Microsoft.Health.Fhir.Tests.Common;
 using Microsoft.Health.JobManagement;
@@ -24,6 +26,8 @@
 using NSubstitute;
 using Xunit;
 
+using FhirJobConflictException = global::Microsoft.Health.Fhir.Core.Features.Operations.JobConflictException;
+
 namespace Microsoft.Health.Fhir.Core.UnitTests.Features.Operations.BulkDelete
 {
     [Trait(Traits.OwningTeam, OwningTeam.Fhir)]
@@ -32,6 +36,7 @@
     {
         private IDeletionService _deleter;
         private BulkDeleteProcessingJob _processingJob;
+        private ISearchParameterOperations _searchParameterOperations;
         private ISearchService _searchService;
         private IQueueClient _queueClient;
 
@@ -42,7 +47,8 @@
                 .Returns(Task.FromResult(new SearchResult(5, new List<Tuple<string, string>>())));
             _queueClient = Substitute.For<IQueueClient>();
             _deleter = Substitute.For<IDeletionService>();
-            _processingJob = new BulkDeleteProcessingJob(_deleter.CreateMockScopeFactory(), Substitute.For<RequestContextAccessor<IFhirRequestContext>>(), Substitute.For<IMediator>(), _searchService.CreateMockScopeFactory(), _queueClient);
+            _searchParameterOperations = Substitute.For<ISearchParameterOperations>();
+            _processingJob = new BulkDeleteProcessingJob(_deleter.CreateMockScopeFactory(), Substitute.For<RequestContextAccessor<IFhirRequestContext>>(), Substitute.For<IMediator>(), _searchParameterOperations, _searchService.CreateMockScopeFactory(), _queueClient);
         }
 
         [Fact]
@@ -103,5 +109,23 @@
             var actualDefinition = JsonConvert.DeserializeObject<BulkDeleteDefinition>(definitions[0]);
             Assert.Equal(2, actualDefinition.Type.SplitByOrSeparator().Count());
         }
+
+        [Fact]
+        public async Task GivenProcessingJobForSearchParameter_WhenReindexStartsBeforeExecution_ThenConflictIsThrown()
+        {
+            var definition = new BulkDeleteDefinition(JobType.BulkDeleteProcessing, DeleteOperation.HardDelete, KnownResourceTypes.SearchParameter, new List<Tuple<string, string>>(), new List<string>(), "https:\\test.com", "https:\\test.com", "test");
+            var jobInfo = new JobInfo
+            {
+                Id = 1,
+                Definition = JsonConvert.SerializeObject(definition),
+            };
+
+            _searchParameterOperations
+                .When(x => x.EnsureNoActiveReindexJobAsync(Arg.Any<CancellationToken>()))
+                .Do(_ => throw new FhirJobConflictException("reindex running"));
+
+            await Assert.ThrowsAsync<FhirJobConflictException>(() => _processingJob.ExecuteAsync(jobInfo, CancellationToken.None));
+            await _deleter.DidNotReceiveWithAnyArgs().DeleteMultipleAsync(default, default, default);
+        }
     }
 }

src/Microsoft.Health.Fhir.Core.UnitTests/Features/Search/Registry/SearchParameterStatusManagerTests.cs

@@ -205,5 +205,54 @@ public async Task GivenASPStatusManager_WhenInitializingAndResolverThrowsExcepti
             Assert.False(list[2].IsSupported);
             Assert.False(list[2].IsPartiallySupported);
         }
+
+        [Fact]
+        public async Task GivenASPStatusManager_WhenUpdatingStatus_ThenInMemoryCacheIsNotMutatedAndMediatorIsNotPublished()
+        {
+            // Arrange - Initialize so search parameters have known in-memory state
+            await _manager.EnsureInitializedAsync(CancellationToken.None);
+
+            var list = _searchParameterDefinitionManager.GetSearchParameters("Account").ToList();
+
+            // Capture initial in-memory state for the Enabled parameter (index 0: ResourceId)
+            bool initialIsSearchable = list[0].IsSearchable;
+            bool initialIsSupported = list[0].IsSupported;
+            bool initialIsPartiallySupported = list[0].IsPartiallySupported;
+            SortParameterStatus initialSortStatus = list[0].SortStatus;
+
+            // Clear any mediator calls from initialization
+            _mediator.ClearReceivedCalls();
+
+            // Act - Call UpdateSearchParameterStatusAsync to change status to Supported
+            await _manager.UpdateSearchParameterStatusAsync(
+                new[] { ResourceId },
+                SearchParameterStatus.Supported,
+                CancellationToken.None);
+
+            // Assert - DB write occurred
+            await _searchParameterStatusDataStore
+                .Received(1)
+                .UpsertStatuses(
+                    Arg.Is<List<ResourceSearchParameterStatus>>(statuses =>
+                        statuses.Count == 1 &&
+                        statuses[0].Uri.OriginalString == ResourceId &&
+                        statuses[0].Status == SearchParameterStatus.Supported),
+                    Arg.Any<CancellationToken>());
+
+            // Assert - In-memory SearchParameterInfo was NOT modified
+            // Re-fetch from the definition manager to make the assertion intent explicit
+            var refreshedList = _searchParameterDefinitionManager.GetSearchParameters("Account").ToList();
+            Assert.Equal(initialIsSearchable, refreshedList[0].IsSearchable);
+            Assert.Equal(initialIsSupported, refreshedList[0].IsSupported);
+            Assert.Equal(initialIsPartiallySupported, refreshedList[0].IsPartiallySupported);
+            Assert.Equal(initialSortStatus, refreshedList[0].SortStatus);
+
+            // Assert - Mediator was NOT called (no SearchParametersUpdatedNotification)
+            await _mediator
+                .DidNotReceive()
+                .Publish(
+                    Arg.Any<SearchParametersUpdatedNotification>(),
+                    Arg.Any<CancellationToken>());
+        }
     }
 }

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/BulkDelete/BulkDeleteProcessingJob.cs

@@ -19,7 +19,9 @@
 using Microsoft.Health.Fhir.Core.Features.Operations.BulkDelete.Messages;
 using Microsoft.Health.Fhir.Core.Features.Persistence;
 using Microsoft.Health.Fhir.Core.Features.Search;
+using Microsoft.Health.Fhir.Core.Features.Search.Parameters;
 using Microsoft.Health.Fhir.Core.Messages.Delete;
+using Microsoft.Health.Fhir.Core.Models;
 using Microsoft.Health.JobManagement;
 using Newtonsoft.Json;
 
@@ -31,19 +33,22 @@ public class BulkDeleteProcessingJob : IJob
         private readonly Func<IScoped<IDeletionService>> _deleterFactory;
         private readonly RequestContextAccessor<IFhirRequestContext> _contextAccessor;
         private readonly IMediator _mediator;
+        private readonly ISearchParameterOperations _searchParameterOperations;
         private readonly Func<IScoped<ISearchService>> _searchService;
         private readonly IQueueClient _queueClient;
 
         public BulkDeleteProcessingJob(
             Func<IScoped<IDeletionService>> deleterFactory,
             RequestContextAccessor<IFhirRequestContext> contextAccessor,
             IMediator mediator,
+            ISearchParameterOperations searchParameterOperations,
             Func<IScoped<ISearchService>> searchService,
             IQueueClient queueClient)
         {
             _deleterFactory = EnsureArg.IsNotNull(deleterFactory, nameof(deleterFactory));
             _contextAccessor = EnsureArg.IsNotNull(contextAccessor, nameof(contextAccessor));
             _mediator = EnsureArg.IsNotNull(mediator, nameof(mediator));
+            _searchParameterOperations = EnsureArg.IsNotNull(searchParameterOperations, nameof(searchParameterOperations));
             _searchService = EnsureArg.IsNotNull(searchService, nameof(searchService));
             _queueClient = EnsureArg.IsNotNull(queueClient, nameof(queueClient));
         }
@@ -78,6 +83,13 @@ public async Task<string> ExecuteAsync(JobInfo jobInfo, CancellationToken cancel
                 Exception exception = null;
                 List<string> types = definition.Type.SplitByOrSeparator().ToList();
 
+                if (types.Count > 0
+                    && string.Equals(types[0], KnownResourceTypes.SearchParameter, StringComparison.OrdinalIgnoreCase)
+                    && !(definition.ExcludedResourceTypes?.Any(x => string.Equals(x, KnownResourceTypes.SearchParameter, StringComparison.OrdinalIgnoreCase)) ?? false))
+                {
+                    await _searchParameterOperations.EnsureNoActiveReindexJobAsync(cancellationToken);
+                }
+
                 try
                 {
                     resourcesDeleted = await deleter.Value.DeleteMultipleAsync(

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

@@ -69,11 +69,6 @@ public async Task<CreateReindexResponse> Handle(CreateReindexRequest request, Ca
                 return new CreateReindexResponse(existingJob);
             }
 
-            // We need to pull in latest search parameter updates from the data store before creating a reindex job.
-            // There could be a potential delay of <see cref="ReindexJobConfiguration.JobPollingFrequency"/> before
-            // search parameter updates on one instance propagates to other instances.
-            await _searchParameterOperations.GetAndApplySearchParameterUpdates(cancellationToken);
-
             // What this handles is the scenario where a user is effectively forcing a reindex to run by passing
             // in a parameter of targetSearchParameterTypes. From those we can identify the base resource types.
             var searchParameterResourceTypes = new HashSet<string>();