Move check for "running reindex" in the data store

src/Microsoft.Health.Fhir.Core.UnitTests/Features/Operations/BulkDelete/BulkDeleteProcessingJobTests.cs

@@ -36,7 +36,6 @@ public class BulkDeleteProcessingJobTests
     {
         private IDeletionService _deleter;
         private BulkDeleteProcessingJob _processingJob;
-        private ISearchParameterOperations _searchParameterOperations;
         private ISearchService _searchService;
         private IQueueClient _queueClient;
 
@@ -47,8 +46,7 @@ public BulkDeleteProcessingJobTests()
                 .Returns(Task.FromResult(new SearchResult(5, new List<Tuple<string, string>>())));
             _queueClient = Substitute.For<IQueueClient>();
             _deleter = Substitute.For<IDeletionService>();
-            _searchParameterOperations = Substitute.For<ISearchParameterOperations>();
-            _processingJob = new BulkDeleteProcessingJob(_deleter.CreateMockScopeFactory(), Substitute.For<RequestContextAccessor<IFhirRequestContext>>(), Substitute.For<IMediator>(), _searchParameterOperations, _searchService.CreateMockScopeFactory(), _queueClient);
+            _processingJob = new BulkDeleteProcessingJob(_deleter.CreateMockScopeFactory(), Substitute.For<RequestContextAccessor<IFhirRequestContext>>(), Substitute.For<IMediator>(), _searchService.CreateMockScopeFactory(), _queueClient);
         }
 
         [Fact]
@@ -113,37 +111,56 @@ public async Task GivenProcessingJob_WhenJobIsRunWithMultipleResourceTypes_ThenF
         [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),
-            };
+            var definition = new BulkDeleteDefinition(JobType.BulkDeleteProcessing, DeleteOperation.HardDelete, "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"));
+            // Simulate the database layer throwing JobConflictException when trying to delete SearchParameter during active reindex
+            _deleter.DeleteMultipleAsync(Arg.Any<ConditionalDeleteResourceRequest>(), Arg.Any<CancellationToken>(), Arg.Any<IList<string>>()).Returns<Task<IDictionary<string, long>>>(x => throw new FhirJobConflictException("A reindex job is currently running."));
 
-            await Assert.ThrowsAsync<FhirJobConflictException>(() => _processingJob.ExecuteAsync(jobInfo, CancellationToken.None));
-            await _deleter.DidNotReceiveWithAnyArgs().DeleteMultipleAsync(default, default, default);
+            await Assert.ThrowsAsync<FhirJobConflictException>(async () => await _processingJob.ExecuteAsync(jobInfo, CancellationToken.None));
         }
 
         [Fact]
-        public async Task GivenProcessingJobWithSearchParameterLaterInTheChain_WhenReindexStartsBeforeExecution_ThenConflictIsThrownBeforeAnyDelete()
+        public async Task GivenBulkDeleteWithSearchParameterFirst_WhenReindexIsActive_ThenConflictIsThrownImmediately()
         {
-            var definition = new BulkDeleteDefinition(JobType.BulkDeleteProcessing, DeleteOperation.HardDelete, "Patient,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),
-            };
+            var definition = new BulkDeleteDefinition(JobType.BulkDeleteProcessing, DeleteOperation.HardDelete, "SearchParameter,Patient", 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) };
+
+            // SearchParameter processed first - database layer throws conflict due to active reindex
+            _deleter.DeleteMultipleAsync(Arg.Any<ConditionalDeleteResourceRequest>(), Arg.Any<CancellationToken>(), Arg.Any<IList<string>>()).Returns<Task<IDictionary<string, long>>>(x => throw new FhirJobConflictException("A reindex job is currently running."));
+
+            // The conflict should propagate through the workflow, preventing any deletions
+            await Assert.ThrowsAsync<FhirJobConflictException>(async () => await _processingJob.ExecuteAsync(jobInfo, CancellationToken.None));
+        }
 
-            _searchParameterOperations
-                .When(x => x.EnsureNoActiveReindexJobAsync(Arg.Any<CancellationToken>()))
-                .Do(_ => throw new FhirJobConflictException("reindex running"));
+        [Fact]
+        public async Task GivenBulkDeleteWithSearchParameterSecond_WhenReindexStartsBetweenDeletes_ThenFollowUpJobFails()
+        {
+            var definition = new BulkDeleteDefinition(JobType.BulkDeleteProcessing, DeleteOperation.HardDelete, "Patient,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) };
+
+            // Patient processed first and succeeds (reindex hasn't started yet)
+            var patientResults = new Dictionary<string, long> { { "Patient", 5 } };
+            _deleter.DeleteMultipleAsync(Arg.Any<ConditionalDeleteResourceRequest>(), Arg.Any<CancellationToken>(), Arg.Any<IList<string>>()).Returns(Task.FromResult<IDictionary<string, long>>(patientResults));
+
+            // Execute the first job - should succeed and enqueue follow-up job for SearchParameter
+            var result = await _processingJob.ExecuteAsync(jobInfo, CancellationToken.None);
+            var bulkDeleteResult = JsonConvert.DeserializeObject<BulkDeleteResult>(result);
+            Assert.Equal(5, bulkDeleteResult.ResourcesDeleted["Patient"]);
+
+            // Verify a follow-up job was queued for SearchParameter
+            var calls = _queueClient.ReceivedCalls();
+            var definitions = (string[])calls.First().GetArguments()[1];
+            Assert.Single(definitions);
+            var followUpDefinition = JsonConvert.DeserializeObject<BulkDeleteDefinition>(definitions[0]);
+            Assert.Equal("SearchParameter", followUpDefinition.Type);
+
+            // Now simulate the follow-up job running, but reindex has started between the two jobs
+            _deleter.ClearReceivedCalls();
+            _deleter.DeleteMultipleAsync(Arg.Any<ConditionalDeleteResourceRequest>(), Arg.Any<CancellationToken>(), Arg.Any<IList<string>>()).Returns<Task<IDictionary<string, long>>>(x => throw new FhirJobConflictException("A reindex job is currently running."));
 
-            await Assert.ThrowsAsync<FhirJobConflictException>(() => _processingJob.ExecuteAsync(jobInfo, CancellationToken.None));
-            await _deleter.DidNotReceiveWithAnyArgs().DeleteMultipleAsync(default, default, default);
+            var followUpJobInfo = new JobInfo() { Id = 2, Definition = definitions[0] };
+            await Assert.ThrowsAsync<FhirJobConflictException>(async () => await _processingJob.ExecuteAsync(followUpJobInfo, CancellationToken.None));
         }
     }
 }

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

@@ -413,10 +413,8 @@ public async Task WhenBackgroundIsBlockedByAPI_BackgroundShouldSkipRefresh()
                 Substitute.For<IModelInfoProvider>(),
                 Substitute.For<ISearchParameterSupportResolver>(),
                 Substitute.For<IDataStoreSearchParameterValidator>(),
-                () => Substitute.For<IScoped<IFhirOperationDataStore>>(),
                 Substitute.For<Func<IScoped<ISearchService>>>(),
                 Substitute.For<IScopeProvider<IFhirDataStore>>(),
-                Substitute.For<IResourceWrapperFactory>(),
                 Substitute.For<ILogger<SearchParameterOperations>>());
 
             // Start a long-running API call that holds the semaphore
@@ -464,10 +462,8 @@ public async Task WhenAPIIsBlockedByBackground_ApiShouldWait()
                 Substitute.For<IModelInfoProvider>(),
                 Substitute.For<ISearchParameterSupportResolver>(),
                 Substitute.For<IDataStoreSearchParameterValidator>(),
-                () => Substitute.For<IScoped<IFhirOperationDataStore>>(),
                 Substitute.For<Func<IScoped<ISearchService>>>(),
                 Substitute.For<IScopeProvider<IFhirDataStore>>(),
-                Substitute.For<IResourceWrapperFactory>(),
                 Substitute.For<ILogger<SearchParameterOperations>>());
 
             var mockLogger = Substitute.For<ILogger<SearchParameterCacheRefreshBackgroundService>>();

src/Microsoft.Health.Fhir.Core/Features/Operations/BulkDelete/BulkDeleteProcessingJob.cs

@@ -33,22 +33,19 @@ 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));
         }
@@ -83,11 +80,6 @@ public async Task<string> ExecuteAsync(JobInfo jobInfo, CancellationToken cancel
                 Exception exception = null;
                 List<string> types = definition.Type.SplitByOrSeparator().ToList();
 
-                if (CanAffectSearchParameters(types, definition.ExcludedResourceTypes))
-                {
-                    await _searchParameterOperations.EnsureNoActiveReindexJobAsync(cancellationToken);
-                }
-
                 try
                 {
                     resourcesDeleted = await deleter.Value.DeleteMultipleAsync(
@@ -106,8 +98,36 @@ public async Task<string> ExecuteAsync(JobInfo jobInfo, CancellationToken cancel
                 catch (IncompleteOperationException<IDictionary<string, long>> ex)
                 {
                     resourcesDeleted = ex.PartialResults;
-                    result.Issues.Add(ex.Message);
-                    exception = ex;
+                    bool conflictFound = false;
+
+                    if (ex.InnerException is AggregateException aggEx)
+                    {
+                        foreach (var innerEx in aggEx.InnerExceptions)
+                        {
+                            if (innerEx is IncompleteOperationException<Dictionary<string, long>> incompleteEx && incompleteEx.InnerException is JobConflictException conflictEx)
+                            {
+                                result.Issues.Add($"JobConflictException: {conflictEx.Message}");
+                                exception = conflictEx;
+                                conflictFound = true;
+                            }
+                            else if (innerEx is JobConflictException directConflictEx)
+                            {
+                                result.Issues.Add($"JobConflictException: {directConflictEx.Message}");
+                                exception = directConflictEx;
+                                conflictFound = true;
+                            }
+                            else
+                            {
+                                result.Issues.Add($"{innerEx.GetType().Name}: {innerEx.Message}");
+                            }
+                        }
+                    }
+
+                    if (!conflictFound)
+                    {
+                        result.Issues.Add(ex.Message);
+                        exception = ex;
+                    }
                 }
 
                 foreach (var (key, value) in resourcesDeleted)
@@ -144,15 +164,5 @@ public async Task<string> ExecuteAsync(JobInfo jobInfo, CancellationToken cancel
                 _contextAccessor.RequestContext = existingFhirRequestContext;
             }
         }
-
-        private static bool CanAffectSearchParameters(IReadOnlyCollection<string> resourceTypes, IList<string> excludedResourceTypes)
-        {
-            if (excludedResourceTypes?.Any(x => string.Equals(x, KnownResourceTypes.SearchParameter, StringComparison.OrdinalIgnoreCase)) == true)
-            {
-                return false;
-            }
-
-            return resourceTypes.Any(x => string.Equals(x, KnownResourceTypes.SearchParameter, StringComparison.OrdinalIgnoreCase));
-        }
     }
 }

src/Microsoft.Health.Fhir.Core/Features/Operations/BulkDelete/Handlers/GetBulkDeleteHandler.cs

@@ -84,7 +84,15 @@ public async Task<GetBulkDeleteResponse> Handle(GetBulkDeleteRequest request, Ca
                             }
                             else
                             {
-                                issues.Add(new OperationOutcomeIssue(OperationOutcomeConstants.IssueSeverity.Error, OperationOutcomeConstants.IssueType.Exception, detailsText: issue));
+                                if (issue.StartsWith("JobConflictException:", StringComparison.Ordinal))
+                                {
+                                    failureResultCode = HttpStatusCode.Conflict;
+                                    issues.Add(new OperationOutcomeIssue(OperationOutcomeConstants.IssueSeverity.Error, OperationOutcomeConstants.IssueType.Conflict, detailsText: issue.Substring("JobConflictException:".Length).TrimStart()));
+                                }
+                                else
+                                {
+                                    issues.Add(new OperationOutcomeIssue(OperationOutcomeConstants.IssueSeverity.Error, OperationOutcomeConstants.IssueType.Exception, detailsText: issue));
+                                }
                             }
                         }
                     }
@@ -94,7 +102,10 @@ public async Task<GetBulkDeleteResponse> Handle(GetBulkDeleteRequest request, Ca
                     }
 
                     // Need a way to get a failure result code. Most likely will be 503, 400, or 403
-                    failureResultCode = HttpStatusCode.InternalServerError;
+                    if (failureResultCode == HttpStatusCode.OK)
+                    {
+                        failureResultCode = HttpStatusCode.InternalServerError;
+                    }
                 }
                 else if (job.Status == JobStatus.Cancelled)
                 {

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

@@ -3,19 +3,14 @@
 // Licensed under the MIT License (MIT). See LICENSE in the repo root for license information.
 // -------------------------------------------------------------------------------------------------
 
-using System.Collections.Generic;
-using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using EnsureThat;
 using MediatR;
 using Microsoft.Extensions.Options;
 using Microsoft.Health.Core.Features.Security.Authorization;
 using Microsoft.Health.Fhir.Core.Configs;
-using Microsoft.Health.Fhir.Core.Exceptions;
-using Microsoft.Health.Fhir.Core.Features.Definition;
 using Microsoft.Health.Fhir.Core.Features.Operations.Reindex.Models;
-using Microsoft.Health.Fhir.Core.Features.Search.Parameters;
 using Microsoft.Health.Fhir.Core.Features.Security;
 using Microsoft.Health.Fhir.Core.Features.Security.Authorization;
 using Microsoft.Health.Fhir.Core.Messages.Reindex;
@@ -28,27 +23,19 @@ public class CreateReindexRequestHandler : IRequestHandler<CreateReindexRequest,
         private readonly IFhirOperationDataStore _fhirOperationDataStore;
         private readonly IAuthorizationService<DataActions> _authorizationService;
         private readonly ReindexJobConfiguration _reindexJobConfiguration;
-        private readonly ISearchParameterDefinitionManager _searchParameterDefinitionManager;
-        private readonly ISearchParameterOperations _searchParameterOperations;
 
         public CreateReindexRequestHandler(
             IFhirOperationDataStore fhirOperationDataStore,
             IAuthorizationService<DataActions> authorizationService,
-            IOptions<ReindexJobConfiguration> reindexJobConfiguration,
-            ISearchParameterDefinitionManager searchParameterDefinitionManager,
-            ISearchParameterOperations searchParameterOperations)
+            IOptions<ReindexJobConfiguration> reindexJobConfiguration)
         {
             EnsureArg.IsNotNull(fhirOperationDataStore, nameof(fhirOperationDataStore));
             EnsureArg.IsNotNull(authorizationService, nameof(authorizationService));
             EnsureArg.IsNotNull(reindexJobConfiguration, nameof(reindexJobConfiguration));
-            EnsureArg.IsNotNull(searchParameterDefinitionManager, nameof(searchParameterDefinitionManager));
-            EnsureArg.IsNotNull(searchParameterOperations, nameof(searchParameterOperations));
 
             _fhirOperationDataStore = fhirOperationDataStore;
             _authorizationService = authorizationService;
             _reindexJobConfiguration = reindexJobConfiguration.Value;
-            _searchParameterDefinitionManager = searchParameterDefinitionManager;
-            _searchParameterOperations = searchParameterOperations;
         }
 
         public async Task<CreateReindexResponse> Handle(CreateReindexRequest request, CancellationToken cancellationToken)

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

@@ -23,8 +23,6 @@ public interface ISearchParameterOperations
 
         Task UpdateSearchParameterStatusAsync(IReadOnlyCollection<string> searchParameterUris, SearchParameterStatus status, CancellationToken cancellationToken, bool ignoreSearchParameterNotSupportedException = false);
 
-        Task EnsureNoActiveReindexJobAsync(CancellationToken cancellationToken);
-
         /// <summary>
         /// This method should be called to get any updates to search param cache
         /// </summary>