diff --git a/src/Microsoft.Health.Fhir.Core.UnitTests/Features/Operations/BulkDelete/BulkDeleteProcessingJobTests.cs b/src/Microsoft.Health.Fhir.Core.UnitTests/Features/Operations/BulkDelete/BulkDeleteProcessingJobTests.cs index 482a682b11..22c59cc660 100644 --- a/src/Microsoft.Health.Fhir.Core.UnitTests/Features/Operations/BulkDelete/BulkDeleteProcessingJobTests.cs +++ b/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>()))); _queueClient = Substitute.For(); _deleter = Substitute.For(); - _searchParameterOperations = Substitute.For(); - _processingJob = new BulkDeleteProcessingJob(_deleter.CreateMockScopeFactory(), Substitute.For>(), Substitute.For(), _searchParameterOperations, _searchService.CreateMockScopeFactory(), _queueClient); + _processingJob = new BulkDeleteProcessingJob(_deleter.CreateMockScopeFactory(), Substitute.For>(), Substitute.For(), _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>(), new List(), "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>(), new List(), "https:\\\\test.com", "https:\\\\test.com", "test"); + var jobInfo = new JobInfo() { Id = 1, Definition = JsonConvert.SerializeObject(definition) }; - _searchParameterOperations - .When(x => x.EnsureNoActiveReindexJobAsync(Arg.Any())) - .Do(_ => throw new FhirJobConflictException("reindex running")); + // Simulate the database layer throwing JobConflictException when trying to delete SearchParameter during active reindex + _deleter.DeleteMultipleAsync(Arg.Any(), Arg.Any(), Arg.Any>()).Returns>>(x => throw new FhirJobConflictException("A reindex job is currently running.")); - await Assert.ThrowsAsync(() => _processingJob.ExecuteAsync(jobInfo, CancellationToken.None)); - await _deleter.DidNotReceiveWithAnyArgs().DeleteMultipleAsync(default, default, default); + await Assert.ThrowsAsync(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>(), new List(), "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>(), new List(), "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(), Arg.Any(), Arg.Any>()).Returns>>(x => throw new FhirJobConflictException("A reindex job is currently running.")); + + // The conflict should propagate through the workflow, preventing any deletions + await Assert.ThrowsAsync(async () => await _processingJob.ExecuteAsync(jobInfo, CancellationToken.None)); + } - _searchParameterOperations - .When(x => x.EnsureNoActiveReindexJobAsync(Arg.Any())) - .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>(), new List(), "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 { { "Patient", 5 } }; + _deleter.DeleteMultipleAsync(Arg.Any(), Arg.Any(), Arg.Any>()).Returns(Task.FromResult>(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(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(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(), Arg.Any(), Arg.Any>()).Returns>>(x => throw new FhirJobConflictException("A reindex job is currently running.")); - await Assert.ThrowsAsync(() => _processingJob.ExecuteAsync(jobInfo, CancellationToken.None)); - await _deleter.DidNotReceiveWithAnyArgs().DeleteMultipleAsync(default, default, default); + var followUpJobInfo = new JobInfo() { Id = 2, Definition = definitions[0] }; + await Assert.ThrowsAsync(async () => await _processingJob.ExecuteAsync(followUpJobInfo, CancellationToken.None)); } } } diff --git a/src/Microsoft.Health.Fhir.Core.UnitTests/Features/Search/Registry/SearchParameterCacheRefreshBackgroundServiceTests.cs b/src/Microsoft.Health.Fhir.Core.UnitTests/Features/Search/Registry/SearchParameterCacheRefreshBackgroundServiceTests.cs index 939be58b4a..e1e0a0a5f7 100644 --- a/src/Microsoft.Health.Fhir.Core.UnitTests/Features/Search/Registry/SearchParameterCacheRefreshBackgroundServiceTests.cs +++ b/src/Microsoft.Health.Fhir.Core.UnitTests/Features/Search/Registry/SearchParameterCacheRefreshBackgroundServiceTests.cs @@ -413,10 +413,8 @@ public async Task WhenBackgroundIsBlockedByAPI_BackgroundShouldSkipRefresh() Substitute.For(), Substitute.For(), Substitute.For(), - () => Substitute.For>(), Substitute.For>>(), Substitute.For>(), - Substitute.For(), Substitute.For>()); // Start a long-running API call that holds the semaphore @@ -464,10 +462,8 @@ public async Task WhenAPIIsBlockedByBackground_ApiShouldWait() Substitute.For(), Substitute.For(), Substitute.For(), - () => Substitute.For>(), Substitute.For>>(), Substitute.For>(), - Substitute.For(), Substitute.For>()); var mockLogger = Substitute.For>(); diff --git a/src/Microsoft.Health.Fhir.Core/Features/Operations/BulkDelete/BulkDeleteProcessingJob.cs b/src/Microsoft.Health.Fhir.Core/Features/Operations/BulkDelete/BulkDeleteProcessingJob.cs index 59aeb4d8d1..a6c3eb8800 100644 --- a/src/Microsoft.Health.Fhir.Core/Features/Operations/BulkDelete/BulkDeleteProcessingJob.cs +++ b/src/Microsoft.Health.Fhir.Core/Features/Operations/BulkDelete/BulkDeleteProcessingJob.cs @@ -33,7 +33,6 @@ public class BulkDeleteProcessingJob : IJob private readonly Func> _deleterFactory; private readonly RequestContextAccessor _contextAccessor; private readonly IMediator _mediator; - private readonly ISearchParameterOperations _searchParameterOperations; private readonly Func> _searchService; private readonly IQueueClient _queueClient; @@ -41,14 +40,12 @@ public BulkDeleteProcessingJob( Func> deleterFactory, RequestContextAccessor contextAccessor, IMediator mediator, - ISearchParameterOperations searchParameterOperations, Func> 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 ExecuteAsync(JobInfo jobInfo, CancellationToken cancel Exception exception = null; List 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 ExecuteAsync(JobInfo jobInfo, CancellationToken cancel catch (IncompleteOperationException> 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> 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 ExecuteAsync(JobInfo jobInfo, CancellationToken cancel _contextAccessor.RequestContext = existingFhirRequestContext; } } - - private static bool CanAffectSearchParameters(IReadOnlyCollection resourceTypes, IList 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)); - } } } diff --git a/src/Microsoft.Health.Fhir.Core/Features/Operations/BulkDelete/Handlers/GetBulkDeleteHandler.cs b/src/Microsoft.Health.Fhir.Core/Features/Operations/BulkDelete/Handlers/GetBulkDeleteHandler.cs index f9fd4f3a22..c601e3e5d5 100644 --- a/src/Microsoft.Health.Fhir.Core/Features/Operations/BulkDelete/Handlers/GetBulkDeleteHandler.cs +++ b/src/Microsoft.Health.Fhir.Core/Features/Operations/BulkDelete/Handlers/GetBulkDeleteHandler.cs @@ -84,7 +84,15 @@ public async Task 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 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) { diff --git a/src/Microsoft.Health.Fhir.Core/Features/Operations/Reindex/CreateReindexRequestHandler.cs b/src/Microsoft.Health.Fhir.Core/Features/Operations/Reindex/CreateReindexRequestHandler.cs index 959b27291f..533eaed410 100644 --- a/src/Microsoft.Health.Fhir.Core/Features/Operations/Reindex/CreateReindexRequestHandler.cs +++ b/src/Microsoft.Health.Fhir.Core/Features/Operations/Reindex/CreateReindexRequestHandler.cs @@ -3,8 +3,6 @@ // 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; @@ -12,10 +10,7 @@ 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 _authorizationService; private readonly ReindexJobConfiguration _reindexJobConfiguration; - private readonly ISearchParameterDefinitionManager _searchParameterDefinitionManager; - private readonly ISearchParameterOperations _searchParameterOperations; public CreateReindexRequestHandler( IFhirOperationDataStore fhirOperationDataStore, IAuthorizationService authorizationService, - IOptions reindexJobConfiguration, - ISearchParameterDefinitionManager searchParameterDefinitionManager, - ISearchParameterOperations searchParameterOperations) + IOptions 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 Handle(CreateReindexRequest request, CancellationToken cancellationToken) diff --git a/src/Microsoft.Health.Fhir.Core/Features/Search/Parameters/ISearchParameterOperations.cs b/src/Microsoft.Health.Fhir.Core/Features/Search/Parameters/ISearchParameterOperations.cs index 07acdb427b..11dbd57673 100644 --- a/src/Microsoft.Health.Fhir.Core/Features/Search/Parameters/ISearchParameterOperations.cs +++ b/src/Microsoft.Health.Fhir.Core/Features/Search/Parameters/ISearchParameterOperations.cs @@ -23,8 +23,6 @@ public interface ISearchParameterOperations Task UpdateSearchParameterStatusAsync(IReadOnlyCollection searchParameterUris, SearchParameterStatus status, CancellationToken cancellationToken, bool ignoreSearchParameterNotSupportedException = false); - Task EnsureNoActiveReindexJobAsync(CancellationToken cancellationToken); - /// /// This method should be called to get any updates to search param cache /// diff --git a/src/Microsoft.Health.Fhir.Core/Features/Search/Parameters/SearchParameterOperations.cs b/src/Microsoft.Health.Fhir.Core/Features/Search/Parameters/SearchParameterOperations.cs index 48262cdcf4..b0617a1d6b 100644 --- a/src/Microsoft.Health.Fhir.Core/Features/Search/Parameters/SearchParameterOperations.cs +++ b/src/Microsoft.Health.Fhir.Core/Features/Search/Parameters/SearchParameterOperations.cs @@ -6,7 +6,6 @@ using System; using System.Collections.Generic; using System.Linq; -using System.Text; using System.Threading; using System.Threading.Tasks; using EnsureThat; @@ -16,12 +15,10 @@ using Microsoft.Health.Extensions.DependencyInjection; using Microsoft.Health.Fhir.Core.Exceptions; using Microsoft.Health.Fhir.Core.Extensions; -using Microsoft.Health.Fhir.Core.Features; using Microsoft.Health.Fhir.Core.Features.Definition; using Microsoft.Health.Fhir.Core.Features.Definition.BundleWrappers; using Microsoft.Health.Fhir.Core.Features.Operations; using Microsoft.Health.Fhir.Core.Features.Persistence; -using Microsoft.Health.Fhir.Core.Features.Search; using Microsoft.Health.Fhir.Core.Features.Search.Registry; using Microsoft.Health.Fhir.Core.Models; using Newtonsoft.Json.Linq; @@ -35,10 +32,8 @@ public class SearchParameterOperations : ISearchParameterOperations, IDisposable private readonly IModelInfoProvider _modelInfoProvider; private readonly ISearchParameterSupportResolver _searchParameterSupportResolver; private readonly IDataStoreSearchParameterValidator _dataStoreSearchParameterValidator; - private readonly Func> _fhirOperationDataStoreFactory; private readonly Func> _searchServiceFactory; private readonly IScopeProvider _fhirDataStoreFactory; - private readonly IResourceWrapperFactory _resourceWrapperFactory; private readonly ILogger _logger; private DateTimeOffset? _searchParamLastUpdated; private readonly SemaphoreSlim _refreshSemaphore; @@ -50,10 +45,8 @@ public SearchParameterOperations( IModelInfoProvider modelInfoProvider, ISearchParameterSupportResolver searchParameterSupportResolver, IDataStoreSearchParameterValidator dataStoreSearchParameterValidator, - Func> fhirOperationDataStoreFactory, Func> searchServiceFactory, IScopeProvider fhirDataStoreFactory, - IResourceWrapperFactory resourceWrapperFactory, ILogger logger) { EnsureArg.IsNotNull(searchParameterStatusManager, nameof(searchParameterStatusManager)); @@ -61,10 +54,8 @@ public SearchParameterOperations( EnsureArg.IsNotNull(modelInfoProvider, nameof(modelInfoProvider)); EnsureArg.IsNotNull(searchParameterSupportResolver, nameof(searchParameterSupportResolver)); EnsureArg.IsNotNull(dataStoreSearchParameterValidator, nameof(dataStoreSearchParameterValidator)); - EnsureArg.IsNotNull(fhirOperationDataStoreFactory, nameof(fhirOperationDataStoreFactory)); EnsureArg.IsNotNull(searchServiceFactory, nameof(searchServiceFactory)); EnsureArg.IsNotNull(fhirDataStoreFactory, nameof(fhirDataStoreFactory)); - EnsureArg.IsNotNull(resourceWrapperFactory, nameof(resourceWrapperFactory)); EnsureArg.IsNotNull(logger, nameof(logger)); _searchParameterStatusManager = searchParameterStatusManager; @@ -72,10 +63,8 @@ public SearchParameterOperations( _modelInfoProvider = modelInfoProvider; _searchParameterSupportResolver = searchParameterSupportResolver; _dataStoreSearchParameterValidator = dataStoreSearchParameterValidator; - _fhirOperationDataStoreFactory = fhirOperationDataStoreFactory; _searchServiceFactory = searchServiceFactory; _fhirDataStoreFactory = fhirDataStoreFactory; - _resourceWrapperFactory = resourceWrapperFactory; _logger = logger; _refreshSemaphore = new SemaphoreSlim(1, 1); } @@ -107,17 +96,6 @@ public string GetSearchParameterHash(string resourceType) } } - public async Task EnsureNoActiveReindexJobAsync(CancellationToken cancellationToken) - { - using IScoped fhirOperationDataStore = _fhirOperationDataStoreFactory(); - (bool found, string id) activeReindexJob = await fhirOperationDataStore.Value.CheckActiveReindexJobsAsync(cancellationToken); - - if (activeReindexJob.found) - { - throw new JobConflictException(string.Format(Core.Resources.ChangesToSearchParametersNotAllowedWhileReindexing, activeReindexJob.id)); - } - } - public async Task ValidateSearchParameterAsync(ITypedElement searchParam, CancellationToken cancellationToken, DateTimeOffset? lastUpdated = null) { var searchParameterWrapper = new SearchParameterWrapper(searchParam); @@ -199,8 +177,6 @@ public async Task DeleteSearchParameterAsync(RawResource searchParamResource, Ca try { - await EnsureNoActiveReindexJobAsync(cancellationToken); - _logger.LogInformation("DeleteSearchParameterAsync: Refreshing cache"); await GetAndApplySearchParameterUpdates(cancellationToken); var status = isHardDelete ? SearchParameterStatus.PendingHardDelete : SearchParameterStatus.PendingDelete; @@ -232,7 +208,6 @@ public async Task DeleteSearchParameterAsync(RawResource searchParamResource, Ca public async Task UpdateSearchParameterStatusAsync(IReadOnlyCollection searchParameterUris, SearchParameterStatus status, CancellationToken cancellationToken, bool ignoreSearchParameterNotSupportedException = false) { - await EnsureNoActiveReindexJobAsync(cancellationToken); await _searchParameterStatusManager.UpdateSearchParameterStatusAsync(searchParameterUris, status, cancellationToken, ignoreSearchParameterNotSupportedException); } diff --git a/src/Microsoft.Health.Fhir.Core/Resources.Designer.cs b/src/Microsoft.Health.Fhir.Core/Resources.Designer.cs index d8170bd831..fad9a8f430 100644 --- a/src/Microsoft.Health.Fhir.Core/Resources.Designer.cs +++ b/src/Microsoft.Health.Fhir.Core/Resources.Designer.cs @@ -19,7 +19,7 @@ namespace Microsoft.Health.Fhir.Core { // class via a tool like ResGen or Visual Studio. // To add or remove a member, edit your .ResX file then rerun ResGen // with the /str option, or rebuild your VS project. - [global::System.CodeDom.Compiler.GeneratedCodeAttribute("System.Resources.Tools.StronglyTypedResourceBuilder", "18.0.0.0")] + [global::System.CodeDom.Compiler.GeneratedCodeAttribute("System.Resources.Tools.StronglyTypedResourceBuilder", "17.0.0.0")] [global::System.Diagnostics.DebuggerNonUserCodeAttribute()] [global::System.Runtime.CompilerServices.CompilerGeneratedAttribute()] internal class Resources { @@ -160,7 +160,7 @@ internal static string ChainedResourceTypesNotAllowedDueToScope { } /// - /// Looks up a localized string similar to Changes to search parameters is not allowed while a reindex job is ongoing. Wait for the reindex job with Id: {0} to finish, or cancel it.. + /// Looks up a localized string similar to Changes to search parameters are not allowed while a reindex job is in progress. Wait for the reindex job with Id: {0} to finish, or cancel it.. /// internal static string ChangesToSearchParametersNotAllowedWhileReindexing { get { diff --git a/src/Microsoft.Health.Fhir.Core/Resources.resx b/src/Microsoft.Health.Fhir.Core/Resources.resx index 292a276b16..f36b370c3e 100644 --- a/src/Microsoft.Health.Fhir.Core/Resources.resx +++ b/src/Microsoft.Health.Fhir.Core/Resources.resx @@ -500,7 +500,7 @@ The search parameter with Uri '{0}' is defined by the FHIR specification and cannot be updated or deleted. Custom search parameters must use a different URL. - Changes to search parameters is not allowed while a reindex job is ongoing. Wait for the reindex job with Id: {0} to finish, or cancel it. + Changes to search parameters are not allowed while a reindex job is in progress. Wait for the reindex job with Id: {0} to finish, or cancel it. {0} Id of currently active Reindex job. diff --git a/src/Microsoft.Health.Fhir.CosmosDb/Features/Storage/Registry/CosmosDbSearchParameterStatusDataStore.cs b/src/Microsoft.Health.Fhir.CosmosDb/Features/Storage/Registry/CosmosDbSearchParameterStatusDataStore.cs index fbf380f082..e001d31ad5 100644 --- a/src/Microsoft.Health.Fhir.CosmosDb/Features/Storage/Registry/CosmosDbSearchParameterStatusDataStore.cs +++ b/src/Microsoft.Health.Fhir.CosmosDb/Features/Storage/Registry/CosmosDbSearchParameterStatusDataStore.cs @@ -13,8 +13,10 @@ using Microsoft.Health.Core; using Microsoft.Health.Extensions.DependencyInjection; using Microsoft.Health.Fhir.Core.Extensions; +using Microsoft.Health.Fhir.Core.Features.Operations; using Microsoft.Health.Fhir.Core.Features.Search.Registry; using Microsoft.Health.Fhir.CosmosDb.Core.Configs; +using Microsoft.Health.JobManagement; namespace Microsoft.Health.Fhir.CosmosDb.Features.Storage.Registry { @@ -22,19 +24,23 @@ public sealed class CosmosDbSearchParameterStatusDataStore : ISearchParameterSta { private readonly Func> _containerScopeFactory; private readonly ICosmosQueryFactory _queryFactory; + private readonly IQueueClient _queueClient; private readonly SemaphoreSlim _statusListSemaphore = new(1, 1); public CosmosDbSearchParameterStatusDataStore( Func> containerScopeFactory, CosmosDataStoreConfiguration cosmosDataStoreConfiguration, - ICosmosQueryFactory queryFactory) + ICosmosQueryFactory queryFactory, + IQueueClient queueClient) { EnsureArg.IsNotNull(containerScopeFactory, nameof(containerScopeFactory)); EnsureArg.IsNotNull(cosmosDataStoreConfiguration, nameof(cosmosDataStoreConfiguration)); EnsureArg.IsNotNull(queryFactory, nameof(queryFactory)); + EnsureArg.IsNotNull(queueClient, nameof(queueClient)); _containerScopeFactory = containerScopeFactory; _queryFactory = queryFactory; + _queueClient = queueClient; } public string SearchParamCacheUpdateProcessName => null; @@ -96,6 +102,17 @@ public async Task UpsertStatuses(IReadOnlyList st return; } + // Check for active reindex jobs unless this call is from a reindex job itself + if (!reindexId.HasValue || reindexId.Value <= 0) + { + var activeJobs = await _queueClient.GetActiveJobsByQueueTypeAsync((byte)QueueType.Reindex, returnParentOnly: true, cancellationToken); + if (activeJobs.Count > 0) + { + var jobId = activeJobs[0].Id; + throw new Fhir.Core.Features.Operations.JobConflictException(string.Format(Fhir.Core.Resources.ChangesToSearchParametersNotAllowedWhileReindexing, jobId)); + } + } + foreach (IEnumerable statusBatch in statuses.TakeBatch(100)) { using IScoped clientScope = _containerScopeFactory.Invoke(); diff --git a/src/Microsoft.Health.Fhir.Shared.Api.UnitTests/Controllers/BulkDeleteControllerTests.cs b/src/Microsoft.Health.Fhir.Shared.Api.UnitTests/Controllers/BulkDeleteControllerTests.cs index 278b145834..8a32362f53 100644 --- a/src/Microsoft.Health.Fhir.Shared.Api.UnitTests/Controllers/BulkDeleteControllerTests.cs +++ b/src/Microsoft.Health.Fhir.Shared.Api.UnitTests/Controllers/BulkDeleteControllerTests.cs @@ -47,12 +47,10 @@ public class BulkDeleteControllerTests private readonly BulkDeleteController _controller; private readonly HttpRequest _httpRequest; private readonly IMediator _mediator; - private readonly ISearchParameterOperations _searchParameterOperations; public BulkDeleteControllerTests() { _mediator = Substitute.For(); - _searchParameterOperations = Substitute.For(); _mediator.Send( Arg.Any(), Arg.Any()) @@ -81,16 +79,8 @@ public BulkDeleteControllerTests() Arg.Any()) .Returns(new Uri(OperationResultUrl)); - var searchService = Substitute.For(); - searchService.GetUsedResourceTypes(Arg.Any()) - .Returns(Task.FromResult>(new List { KnownResourceTypes.Patient })); - var scopedSearchService = Substitute.For>(); - scopedSearchService.Value.Returns(searchService); - _controller = new BulkDeleteController( _mediator, - _searchParameterOperations, - () => scopedSearchService, urlResolver); _controller.ControllerContext = new ControllerContext( new ActionContext( @@ -201,30 +191,6 @@ await Run( purgeHistory)); } - [Fact] - public async Task GivenBulkDeleteForSearchParameter_WhenReindexIsRunning_ThenConflictIsThrown() - { - _searchParameterOperations - .When(x => x.EnsureNoActiveReindexJobAsync(Arg.Any())) - .Do(_ => throw new FhirJobConflictException("reindex running")); - - await Assert.ThrowsAsync(() => _controller.BulkDeleteByResourceType(KnownResourceTypes.SearchParameter, new HardDeleteModel(), false, false)); - await _mediator.DidNotReceiveWithAnyArgs().Send(default, default); - } - - [Fact] - public async Task GivenBulkDeleteWithSearchParameterExcluded_WhenReindexIsRunning_ThenRequestStillSucceeds() - { - _searchParameterOperations - .When(x => x.EnsureNoActiveReindexJobAsync(Arg.Any())) - .Do(_ => throw new FhirJobConflictException("reindex running")); - - var response = await _controller.BulkDelete(new HardDeleteModel(), false, false, KnownResourceTypes.SearchParameter); - - Assert.IsType(response); - await _searchParameterOperations.DidNotReceive().EnsureNoActiveReindexJobAsync(Arg.Any()); - } - [Theory] [InlineData(0, HttpStatusCode.Accepted)] [InlineData(1, HttpStatusCode.OK)] diff --git a/src/Microsoft.Health.Fhir.Shared.Api/Controllers/BulkDeleteController.cs b/src/Microsoft.Health.Fhir.Shared.Api/Controllers/BulkDeleteController.cs index 15df6132bf..d31a47a01c 100644 --- a/src/Microsoft.Health.Fhir.Shared.Api/Controllers/BulkDeleteController.cs +++ b/src/Microsoft.Health.Fhir.Shared.Api/Controllers/BulkDeleteController.cs @@ -11,7 +11,6 @@ using MediatR; using Microsoft.AspNetCore.Mvc; using Microsoft.Health.Api.Features.Audit; -using Microsoft.Health.Extensions.DependencyInjection; using Microsoft.Health.Fhir.Api.Extensions; using Microsoft.Health.Fhir.Api.Features.ActionResults; using Microsoft.Health.Fhir.Api.Features.Filters; @@ -26,8 +25,6 @@ using Microsoft.Health.Fhir.Core.Features.Operations.BulkDelete; using Microsoft.Health.Fhir.Core.Features.Operations.BulkDelete.Messages; using Microsoft.Health.Fhir.Core.Features.Routing; -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.ValueSets; @@ -40,8 +37,6 @@ namespace Microsoft.Health.Fhir.Api.Controllers public class BulkDeleteController : Controller { private readonly IMediator _mediator; - private readonly ISearchParameterOperations _searchParameterOperations; - private readonly Func> _searchService; private readonly IUrlResolver _urlResolver; private readonly HashSet _excludedParameters = new(new PropertyEqualityComparer(StringComparison.OrdinalIgnoreCase, s => s)) @@ -55,13 +50,9 @@ public class BulkDeleteController : Controller public BulkDeleteController( IMediator mediator, - ISearchParameterOperations searchParameterOperations, - Func> searchService, IUrlResolver urlResolver) { _mediator = EnsureArg.IsNotNull(mediator, nameof(mediator)); - _searchParameterOperations = EnsureArg.IsNotNull(searchParameterOperations, nameof(searchParameterOperations)); - _searchService = EnsureArg.IsNotNull(searchService, nameof(searchService)); _urlResolver = EnsureArg.IsNotNull(urlResolver, nameof(urlResolver)); } @@ -158,33 +149,11 @@ private async Task SendDeleteRequest(string typeParameter, bool h excludedResourceTypesList = excludedResourceTypes.Split(',').ToList(); } - if (await CanAffectSearchParametersAsync(typeParameter, excludedResourceTypesList)) - { - await _searchParameterOperations.EnsureNoActiveReindexJobAsync(HttpContext.RequestAborted); - } - CreateBulkDeleteResponse result = await _mediator.BulkDeleteAsync(deleteOperation, typeParameter, searchParameters, softDeleteCleanup, excludedResourceTypesList, removeReferences, HttpContext.RequestAborted); var response = JobResult.Accepted(); response.SetContentLocationHeader(_urlResolver, OperationsConstants.BulkDelete, result.Id.ToString()); return response; } - - private async Task CanAffectSearchParametersAsync(string resourceType, IList excludedResourceTypes) - { - if (excludedResourceTypes?.Any(x => string.Equals(x, KnownResourceTypes.SearchParameter, StringComparison.OrdinalIgnoreCase)) == true) - { - return false; - } - - if (!string.IsNullOrWhiteSpace(resourceType)) - { - return resourceType.SplitByOrSeparator().Any(x => string.Equals(x, KnownResourceTypes.SearchParameter, StringComparison.OrdinalIgnoreCase)); - } - - using IScoped searchService = _searchService.Invoke(); - IReadOnlyList usedResourceTypes = await searchService.Value.GetUsedResourceTypes(HttpContext.RequestAborted); - return usedResourceTypes.Any(x => string.Equals(x, KnownResourceTypes.SearchParameter, StringComparison.OrdinalIgnoreCase)); - } } } diff --git a/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Operations/SearchParameterState/SearchParameterStateUpdateHandlerTests.cs b/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Operations/SearchParameterState/SearchParameterStateUpdateHandlerTests.cs index e21caf1fdc..798c19e58d 100644 --- a/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Operations/SearchParameterState/SearchParameterStateUpdateHandlerTests.cs +++ b/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Operations/SearchParameterState/SearchParameterStateUpdateHandlerTests.cs @@ -111,7 +111,6 @@ public SearchParameterStateUpdateHandlerTests() _cancellationToken = CancellationToken.None; _authorizationService.CheckAccess(DataActions.SearchParameter, _cancellationToken).Returns(DataActions.SearchParameter); - _searchParameterOperations.EnsureNoActiveReindexJobAsync(Arg.Any()).Returns(Task.CompletedTask); _searchParameterOperations.UpdateSearchParameterStatusAsync(Arg.Any>(), Arg.Any(), Arg.Any(), Arg.Any()).Returns(Task.CompletedTask); var searchParamDefinitionStore = new List @@ -335,16 +334,6 @@ public async Task GivenARequestToUpdateSearchParameterStatus_WhenRequestIsValied Assert.Contains("Status=PendingDisable", loggers.logger.LogRecords[0].State.ToString()); } - [Fact] - public async Task GivenReindexRunningBeforeStatusUpdate_WhenHandlingRequest_ThenJobConflictIsThrown() - { - _searchParameterOperations - .When(x => x.EnsureNoActiveReindexJobAsync(Arg.Any())) - .Do(_ => throw new FhirJobConflictException("reindex running")); - - await Assert.ThrowsAsync(() => _searchParameterStateUpdateHandler.Handle(new SearchParameterStateUpdateRequest(new List>()), default)); - } - private (IAuditLogger auditLogger, TestLogger logger) CreateTestAuditLogger() { IOptions optionsConfig = Substitute.For>(); diff --git a/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Resources/Delete/DeletionServiceTests.cs b/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Resources/Delete/DeletionServiceTests.cs index 37cfe4631e..5a10e96a5c 100644 --- a/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Resources/Delete/DeletionServiceTests.cs +++ b/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Resources/Delete/DeletionServiceTests.cs @@ -266,7 +266,7 @@ public async Task GivenSearchParameterDelete_WhenConcurrencyConflictExhaustsRetr .DeleteSearchParameterAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) .Returns(_ => throw new BadRequestException(Core.Resources.SearchParameterConcurrencyConflict)); - var exception = await Assert.ThrowsAsync>>(async () => + var exception = await Assert.ThrowsAsync>>(async () => await _service.DeleteMultipleAsync(request, CancellationToken.None)); Assert.Contains(" Deletion.3", exception.InnerException.Message); diff --git a/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Search/SearchParameters/SearchParameterDefinitionManagerTests.cs b/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Search/SearchParameters/SearchParameterDefinitionManagerTests.cs index 394f643b9d..f6469a79ca 100644 --- a/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Search/SearchParameters/SearchParameterDefinitionManagerTests.cs +++ b/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Search/SearchParameters/SearchParameterDefinitionManagerTests.cs @@ -148,7 +148,6 @@ public SearchParameterDefinitionManagerTests() var searchParameterDataStoreValidator = Substitute.For(); searchParameterDataStoreValidator.ValidateSearchParameter(Arg.Any(), out Arg.Any()).Returns(true, null); - var fhirOperationDataStore = Substitute.For(); var searchService = Substitute.For(); var fhirDataStore = Substitute.For(); @@ -158,10 +157,8 @@ public SearchParameterDefinitionManagerTests() ModelInfoProvider.Instance, _searchParameterSupportResolver, searchParameterDataStoreValidator, - () => fhirOperationDataStore.CreateMockScope(), () => searchService.CreateMockScope(), fhirDataStore.CreateMockScopeProvider(), - Substitute.For(), NullLogger.Instance); } diff --git a/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Search/SearchParameters/SearchParameterValidatorTests.cs b/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Search/SearchParameters/SearchParameterValidatorTests.cs index 7ae8ef6fd1..89d9eb17c1 100644 --- a/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Search/SearchParameters/SearchParameterValidatorTests.cs +++ b/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Search/SearchParameters/SearchParameterValidatorTests.cs @@ -68,7 +68,6 @@ public SearchParameterValidatorTests() x[3] = searchParameterInfo; return true; }); - _searchParameterOperations.EnsureNoActiveReindexJobAsync(CancellationToken.None).Returns(Task.CompletedTask); _searchParameterOperations.SearchParamLastUpdated.Returns(System.DateTimeOffset.UtcNow); } @@ -318,18 +317,6 @@ public async Task GivenCustomSearchParameter_WhenUpdating_ThenNoExceptionThrown( await validator.ValidateSearchParameterInput(searchParam, "PUT", CancellationToken.None); } - [Fact] - public async Task GivenActiveReindexJob_WhenValidatingSearchParam_ThenJobConflictExceptionThrown() - { - _searchParameterOperations - .When(x => x.EnsureNoActiveReindexJobAsync(Arg.Any())) - .Do(_ => throw new System.Exception("reindex running")); - - var exception = await Assert.ThrowsAsync(() => CreateValidator().ValidateSearchParameterInput(new SearchParameter { Url = "http://unique" }, "POST", CancellationToken.None)); - - Assert.Equal("reindex running", exception.Message); - } - private SearchParameterValidator CreateValidator() => new SearchParameterValidator( _authorizationService, diff --git a/src/Microsoft.Health.Fhir.Shared.Core/Features/Operations/SearchParameterState/SearchParameterStateUpdateHandler.cs b/src/Microsoft.Health.Fhir.Shared.Core/Features/Operations/SearchParameterState/SearchParameterStateUpdateHandler.cs index 977a212fff..6a76bd1af2 100644 --- a/src/Microsoft.Health.Fhir.Shared.Core/Features/Operations/SearchParameterState/SearchParameterStateUpdateHandler.cs +++ b/src/Microsoft.Health.Fhir.Shared.Core/Features/Operations/SearchParameterState/SearchParameterStateUpdateHandler.cs @@ -59,8 +59,6 @@ public async Task Handle(SearchParameterStat await _authorizationService.CheckAccess(DataActions.SearchParameter, true, cancellationToken); - await _searchParameterOperations.EnsureNoActiveReindexJobAsync(cancellationToken); - _resourceSearchParameterStatus = await _searchParameterStatusManager.GetAllSearchParameterStatus(cancellationToken); Dictionary> searchParametersToUpdate = ParseRequestForUpdate(request, out List invalidSearchParameters); @@ -120,7 +118,7 @@ private Dictionary> ParseRequestForUpdate(Se return searchParametersToUpdate; } - private SearchParameterStateUpdateResponse CreateBundleResponse(Dictionary> searchParametersToUpdate, List invalidSearchParameters, bool isReindexRunning = false) + private SearchParameterStateUpdateResponse CreateBundleResponse(Dictionary> searchParametersToUpdate, List invalidSearchParameters) { // Create the bundle from the result. var bundle = new Bundle(); @@ -193,29 +191,6 @@ private SearchParameterStateUpdateResponse CreateBundleResponse(Dictionary - { - new OperationOutcome.IssueComponent - { - Severity = OperationOutcome.IssueSeverity.Error, - Code = OperationOutcome.IssueType.Conflict, - Details = new CodeableConcept - { - Text = Core.Resources.ReindexRunningException, - }, - }, - }, - }, - }); - } - bundle.Type = Bundle.BundleType.BatchResponse; bundle.Total = invalidSearchParameters?.Count + searchParametersToUpdate?.Count; bundle.Meta = new Meta diff --git a/src/Microsoft.Health.Fhir.Shared.Core/Features/Resources/Delete/DeletionService.cs b/src/Microsoft.Health.Fhir.Shared.Core/Features/Resources/Delete/DeletionService.cs index 04b50be38b..0d78a17357 100644 --- a/src/Microsoft.Health.Fhir.Shared.Core/Features/Resources/Delete/DeletionService.cs +++ b/src/Microsoft.Health.Fhir.Shared.Core/Features/Resources/Delete/DeletionService.cs @@ -364,7 +364,7 @@ private async Task> DeleteMultipleAsyncInternal(Condit } }); var aggregateException = new AggregateException(exceptions); - throw new IncompleteOperationException>(aggregateException, resourceTypesDeleted); + throw new IncompleteOperationException>(aggregateException, resourceTypesDeleted); } return resourceTypesDeleted; diff --git a/src/Microsoft.Health.Fhir.Shared.Core/Features/Search/Parameters/SearchParameterValidator.cs b/src/Microsoft.Health.Fhir.Shared.Core/Features/Search/Parameters/SearchParameterValidator.cs index c656183392..9e8bb21c79 100644 --- a/src/Microsoft.Health.Fhir.Shared.Core/Features/Search/Parameters/SearchParameterValidator.cs +++ b/src/Microsoft.Health.Fhir.Shared.Core/Features/Search/Parameters/SearchParameterValidator.cs @@ -73,8 +73,6 @@ public SearchParameterValidator( { await _authorizationService.CheckAccess(DataActions.Reindex, true, cancellationToken); - await _searchParameterOperations.EnsureNoActiveReindexJobAsync(cancellationToken); - if (string.IsNullOrEmpty(searchParam.Url) && (method.Equals(HttpDeleteName, StringComparison.Ordinal) || method.Equals(HttpPatchName, StringComparison.Ordinal))) { // Return out if this is delete OR patch call and no Url so FHIRController can move to next action diff --git a/src/Microsoft.Health.Fhir.SqlServer/Features/ExceptionExtension.cs b/src/Microsoft.Health.Fhir.SqlServer/Features/ExceptionExtension.cs index d907d13e2d..2f50794775 100644 --- a/src/Microsoft.Health.Fhir.SqlServer/Features/ExceptionExtension.cs +++ b/src/Microsoft.Health.Fhir.SqlServer/Features/ExceptionExtension.cs @@ -35,6 +35,14 @@ public static bool IsSearchParameterConcurrencyConflict(this Exception e) && e.Message.Contains("expected last updated", StringComparison.OrdinalIgnoreCase); } + public static bool IsReindexJobConflict(this Exception e) + { + var sqlEx = e as SqlException; + return sqlEx != null + && sqlEx.Number == 50002 + && e.Message.Contains("reindex job is in progress", StringComparison.OrdinalIgnoreCase); + } + private static bool HasDeadlockErrorPattern(string str) { return str.Contains("deadlock", StringComparison.OrdinalIgnoreCase); diff --git a/src/Microsoft.Health.Fhir.SqlServer/Features/Storage/Registry/SqlServerSearchParameterStatusDataStore.cs b/src/Microsoft.Health.Fhir.SqlServer/Features/Storage/Registry/SqlServerSearchParameterStatusDataStore.cs index 0fd095aa98..5f136eaeb6 100644 --- a/src/Microsoft.Health.Fhir.SqlServer/Features/Storage/Registry/SqlServerSearchParameterStatusDataStore.cs +++ b/src/Microsoft.Health.Fhir.SqlServer/Features/Storage/Registry/SqlServerSearchParameterStatusDataStore.cs @@ -139,7 +139,11 @@ public async Task UpsertStatuses(IReadOnlyList st using var cmd = new SqlCommand(); cmd.CommandType = CommandType.StoredProcedure; cmd.CommandText = "dbo.MergeSearchParams"; - cmd.Parameters.AddWithValue("@ReindexId", reindexId ?? 0); + if (reindexId.HasValue) + { + cmd.Parameters.AddWithValue("@ReindexId", reindexId.Value); + } + new SearchParamListTableValuedParameterDefinition("@SearchParams").AddParameter(cmd.Parameters, new SearchParamListRowGenerator().GenerateRows(statuses)); try @@ -151,6 +155,11 @@ public async Task UpsertStatuses(IReadOnlyList st _logger.LogWarning(ex, $"Optimistic concurrency conflict occurred while calling dbo.MergeSearchParams. ReindexId={reindexId ?? 0}"); throw new BadRequestException(Core.Resources.SearchParameterConcurrencyConflict); } + catch (SqlException ex) when (ex.IsReindexJobConflict()) + { + _logger.LogWarning(ex, $"Error calling dbo.MergeSearchParams. {ex.Message}"); + throw new JobConflictException(ex.Message); + } } // Synchronize the FHIR model dictionary with the data in SQL search parameter status table diff --git a/src/Microsoft.Health.Fhir.SqlServer/Features/Storage/SqlServerFhirDataStore.cs b/src/Microsoft.Health.Fhir.SqlServer/Features/Storage/SqlServerFhirDataStore.cs index 04e4c46db3..18202bc4eb 100644 --- a/src/Microsoft.Health.Fhir.SqlServer/Features/Storage/SqlServerFhirDataStore.cs +++ b/src/Microsoft.Health.Fhir.SqlServer/Features/Storage/SqlServerFhirDataStore.cs @@ -217,6 +217,11 @@ public async Task MergeAsync(IReadOnlyList x.Resource).ToList(); - - await CleanupAsync(); - - await CreateAsync(); - - await EnsureCreateAsync(); - - await CheckSearchParameterStatusAsync(SearchParameterStatus.Supported, TimeSpan.FromSeconds(searchParameterStatusTimeoutSeconds)); - - var queryParams = new Dictionary { { KnownQueryParameterNames.BulkHardDelete, hardDelete ? "true" : "false" } }; - using var request = GenerateBulkDeleteRequest(tag, $"{ResourceType.SearchParameter}/$bulk-delete", queryParams); - using var response = await _httpClient.SendAsync(request); - Assert.Equal(HttpStatusCode.Accepted, response.StatusCode); - - await CheckBulkDeleteStatusAsync(response.Content.Headers.ContentLocation, bundle.Entry.Count); - - await EnsureBulkDeleteAsync(); - - var expectedStatus = hardDelete ? SearchParameterStatus.PendingHardDelete : SearchParameterStatus.PendingDelete; - await CheckSearchParameterStatusAsync(expectedStatus, TimeSpan.FromSeconds(searchParameterStatusTimeoutSeconds)); - - async Task CleanupAsync() - { - foreach (var resource in resources) - { - var status = (await _fhirClient.HardDeleteAsync(resource, false)).StatusCode; - Assert.True(status == HttpStatusCode.NotFound || status == HttpStatusCode.NoContent, $"expected=({HttpStatusCode.NotFound},{HttpStatusCode.NoContent}) actual={status}"); - } - } - - async Task CreateAsync() - { - foreach (var resource in resources) - { - var status = (await _fhirClient.UpdateAsync(resource)).StatusCode; - Assert.True(status == HttpStatusCode.Created || status == HttpStatusCode.OK, $"expected=({HttpStatusCode.Created},{HttpStatusCode.OK}) actual={status}"); - } - } - - async Task EnsureCreateAsync() - { - foreach (var resource in resources) - { - var status = (await _fhirClient.ReadAsync($"{ResourceType.SearchParameter}/{resource.Id}")).StatusCode; - Assert.True(status == HttpStatusCode.OK, $"expected={HttpStatusCode.OK} actual={status}"); - } - } - - async Task CheckBulkDeleteStatusAsync(Uri location, long expectedCount) - { - var result = (await _fhirClient.WaitForBulkJobStatus("Bulk Delete", location)).Resource; - var actualCount = 0L; - var issuesChecked = 0; - foreach (var parameter in result.Parameter) - { - if (parameter.Name == "Issues") - { - issuesChecked++; - } - else if (parameter.Name == "ResourceDeletedCount") - { - foreach (var part in parameter.Part) - { - Assert.True(part.Name == KnownResourceTypes.SearchParameter, $"Unexpected type={part.Name}"); - actualCount = (long)((Integer64)part.Value).Value; - } - } - else - { - throw new Exception($"Unexpected parameter {parameter.Name}"); - } - } - - Assert.True(issuesChecked == 0, $"issues={issuesChecked}"); - Assert.True(expectedCount == actualCount, $"expected={expectedCount} actual={actualCount}"); - } - - async Task EnsureBulkDeleteAsync() - { - // SearchParameter resources should still exist after bulk delete - only status is updated - var response = await _fhirClient.SearchAsync(ResourceType.SearchParameter, $"_tag={tag}"); - Assert.Equal(HttpStatusCode.OK, response.StatusCode); - var count = response.Resource?.Entry.Count ?? 0; - Assert.True(count == resources.Count, $"Expected {resources.Count} search parameters to still exist (only status updated), but found {count}."); - } - - async Task CheckSearchParameterStatusAsync(SearchParameterStatus expectedStatus, TimeSpan timeout) - { - if (_fixture.DataStore == DataStore.CosmosDb) - { - return; - } - - var expectedStatusText = expectedStatus.ToString(); - var pollingInterval = TimeSpan.FromSeconds(2); - var stopwatch = Stopwatch.StartNew(); - string lastObservedState = "No status responses recorded."; - - while (stopwatch.Elapsed < timeout) - { - var statusesMatched = true; - var observedStates = new List(); - - foreach (var url in resources.Select(resource => ((SearchParameter)resource).Url)) - { - try - { - var response = await _fhirClient.ReadAsync($"{ResourceType.SearchParameter}/$status?url={url}"); - if (response.StatusCode != HttpStatusCode.OK) - { - statusesMatched = false; - observedStates.Add($"url={url} response={response.StatusCode}"); - continue; - } - - var part = response.Resource.Parameter - .FirstOrDefault(x => x.Part.Any(p => string.Equals(p.Name, "url", StringComparison.OrdinalIgnoreCase) && string.Equals(p.Value?.ToString(), url, StringComparison.OrdinalIgnoreCase))); - - var status = part?.Part - .Where(x => string.Equals(x.Name, "status", StringComparison.OrdinalIgnoreCase)) - .Select(x => x.Value?.ToString()) - .FirstOrDefault(); - - observedStates.Add($"url={url} actual={status ?? ""}"); - - if (part == null || !string.Equals(status, expectedStatusText, StringComparison.Ordinal)) - { - statusesMatched = false; - } - } - catch (Exception ex) - { - statusesMatched = false; - observedStates.Add($"url={url} error={ex.GetType().Name}: {ex.Message}"); - } - } - - lastObservedState = string.Join("; ", observedStates); - - if (statusesMatched) - { - return; - } - - await Task.Delay(pollingInterval); - } - - throw new XunitException( - $"Timed out after {timeout.TotalSeconds:F0}s waiting for SearchParameter status '{expectedStatusText}'. Last observed state: {lastObservedState}"); - } - } - [SkippableFact] #pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously public async Task GivenABulkDeleteJob_WhenRemovingReferences_ThenReferencesAreRemoved() diff --git a/test/Microsoft.Health.Fhir.Shared.Tests.E2E/Rest/Reindex/ReindexTests.cs b/test/Microsoft.Health.Fhir.Shared.Tests.E2E/Rest/Reindex/ReindexTests.cs index 22efcd761e..724d4d3bb0 100644 --- a/test/Microsoft.Health.Fhir.Shared.Tests.E2E/Rest/Reindex/ReindexTests.cs +++ b/test/Microsoft.Health.Fhir.Shared.Tests.E2E/Rest/Reindex/ReindexTests.cs @@ -1261,7 +1261,13 @@ public async Task GivenSearchParamBulkDelete_ThenConflictWhenReindexAndSuccessOn using var bulkDeleteRequest = new HttpRequestMessage(HttpMethod.Delete, deleteUrl); bulkDeleteRequest.Headers.Add("Prefer", "respond-async"); var response = await _fixture.HttpClient.SendAsync(bulkDeleteRequest); - Assert.Equal(HttpStatusCode.Conflict, response.StatusCode); + Assert.Equal(HttpStatusCode.Accepted, response.StatusCode); + var location = response.Content.Headers.ContentLocation; + Assert.NotNull(location); + + // Wait for bulk delete job to complete - it should fail due to reindex conflict during execution + var jobResult = await _fixture.TestFhirClient.WaitForBulkJobStatus("Bulk delete", location); + Assert.Equal(HttpStatusCode.Conflict, jobResult.Response.StatusCode); var reindexStatus = await WaitForJobCompletionAsync(reindex.uri, TimeSpan.FromSeconds(300)); Assert.Equal(OperationStatus.Completed, reindexStatus); @@ -1274,11 +1280,11 @@ public async Task GivenSearchParamBulkDelete_ThenConflictWhenReindexAndSuccessOn bulkDeleteRequest2.Headers.Add("Prefer", "respond-async"); response = await _fixture.HttpClient.SendAsync(bulkDeleteRequest2); Assert.Equal(HttpStatusCode.Accepted, response.StatusCode); - var location = response.Content.Headers.ContentLocation; - Assert.NotNull(location); + var location2 = response.Content.Headers.ContentLocation; + Assert.NotNull(location2); - var jobResult = await _fixture.TestFhirClient.WaitForBulkJobStatus("Bulk delete", location); - Assert.Equal(HttpStatusCode.OK, jobResult.Response.StatusCode); + var jobResult2 = await _fixture.TestFhirClient.WaitForBulkJobStatus("Bulk delete", location2); + Assert.Equal(HttpStatusCode.OK, jobResult2.Response.StatusCode); // resource not touched resource = await _fixture.TestFhirClient.ReadAsync($"SearchParameter/{code}"); diff --git a/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Features/Operations/FhirOperationDataStoreExportTests.cs b/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Features/Operations/FhirOperationDataStoreExportTests.cs index c881568706..a311bcabad 100644 --- a/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Features/Operations/FhirOperationDataStoreExportTests.cs +++ b/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Features/Operations/FhirOperationDataStoreExportTests.cs @@ -33,19 +33,21 @@ public class FhirOperationDataStoreExportTests : IClassFixture, IAsyncLifetime { private readonly IFhirOperationDataStore _operationDataStore; - private readonly IFhirStorageTestHelper _testHelper; public FhirOperationDataStoreReindexTests(FhirStorageTestsFixture fixture) { - _operationDataStore = fixture.OperationDataStore; - _testHelper = fixture.TestHelper; + _operationDataStore = fixture.TestSqlServerOperationDataStore ?? fixture.OperationDataStore; } public async Task InitializeAsync() { - await _testHelper.DeleteAllReindexJobRecordsAsync(); - await CancelActiveReindexJobIfExists(); - GetTestQueueClient().ClearJobs(); await AssertNoReindexJobsExist(); diff --git a/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Features/Operations/Reindex/ReindexJobTests.cs b/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Features/Operations/Reindex/ReindexJobTests.cs index 4d307a8fad..3718fcc07a 100644 --- a/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Features/Operations/Reindex/ReindexJobTests.cs +++ b/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Features/Operations/Reindex/ReindexJobTests.cs @@ -84,14 +84,12 @@ public class ReindexJobTests : IClassFixture, IAsyncLif private readonly ISearchIndexer _searchIndexer = Substitute.For(); private ISupportedSearchParameterDefinitionManager _supportedSearchParameterDefinitionManager; private SearchParameterStatusManager _searchParameterStatusManager; - private ISupportedSearchParameterDefinitionManager _supportedSearchParameterDefinitionManager2; private SearchParameterStatusManager _searchParameterStatusManager2; private readonly ISearchParameterSupportResolver _searchParameterSupportResolver = Substitute.For(); private readonly ITestOutputHelper _output; private IScoped _searchService; - private readonly RequestContextAccessor _contextAccessor = Substitute.For>(); private ISearchParameterOperations _searchParameterOperations = null; private ISearchParameterOperations _searchParameterOperations2 = null; private readonly IDataStoreSearchParameterValidator _dataStoreSearchParameterValidator = Substitute.For(); @@ -151,10 +149,8 @@ public async Task InitializeAsync() ModelInfoProvider.Instance, _searchParameterSupportResolver, _dataStoreSearchParameterValidator, - () => _fhirOperationDataStore.CreateMockScope(), () => _searchService, _scopedDataStore.CreateMockScopeProviderFromScoped(), - _resourceWrapperFactory, NullLogger.Instance); // Start background service so it triggers GetAndApplySearchParameterUpdates which signals the TCS. @@ -166,9 +162,7 @@ public async Task InitializeAsync() _createReindexRequestHandler = new CreateReindexRequestHandler( _fhirOperationDataStore, DisabledFhirAuthorizationService.Instance, - _optionsReindexConfig, - _searchParameterDefinitionManager, - _searchParameterOperations); + _optionsReindexConfig); _reindexSingleResourceRequestHandler = new ReindexSingleResourceRequestHandler( DisabledFhirAuthorizationService.Instance, @@ -1524,7 +1518,6 @@ private async Task InitializeSecondFHIRService() _fixture.DataStore.CreateMockScopeProvider(), NullLogger.Instance); await _searchParameterDefinitionManager2.EnsureInitializedAsync(CancellationToken.None); - _supportedSearchParameterDefinitionManager2 = new SupportedSearchParameterDefinitionManager(_searchParameterDefinitionManager2); _searchParameterStatusManager2 = new SearchParameterStatusManager(_fixture.SearchParameterStatusDataStore, _searchParameterDefinitionManager2, _searchParameterSupportResolver, mediator, NullLogger.Instance); await _searchParameterStatusManager2.EnsureInitializedAsync(CancellationToken.None); @@ -1535,10 +1528,8 @@ private async Task InitializeSecondFHIRService() ModelInfoProvider.Instance, _searchParameterSupportResolver, _dataStoreSearchParameterValidator, - () => _fhirOperationDataStore.CreateMockScope(), () => _searchService, _scopedDataStore.CreateMockScopeProviderFromScoped(), - _resourceWrapperFactory, NullLogger.Instance); } diff --git a/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/CosmosDbFhirStorageTestsFixture.cs b/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/CosmosDbFhirStorageTestsFixture.cs index 1c38f43830..353a65e476 100644 --- a/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/CosmosDbFhirStorageTestsFixture.cs +++ b/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/CosmosDbFhirStorageTestsFixture.cs @@ -78,6 +78,7 @@ public class CosmosDbFhirStorageTestsFixture : IServiceProvider, IAsyncLifetime private SearchParameterStatusManager _searchParameterStatusManager; private CosmosClient _cosmosClient; private CosmosQueueClient _queueClient; + private TestQueueClient _testQueueClient; private CosmosFhirOperationDataStore _cosmosFhirOperationDataStore; private ISearchIndexer _searchIndexer; @@ -248,10 +249,15 @@ public virtual async Task InitializeAsync() var documentClient = new NonDisposingScope(_container); + var mockQueueClientForSearchParams = Substitute.For(); + mockQueueClientForSearchParams.GetActiveJobsByQueueTypeAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(Task.FromResult((IReadOnlyList)Array.Empty())); + _searchParameterStatusDataStore = new CosmosDbSearchParameterStatusDataStore( () => documentClient, _cosmosDataStoreConfiguration, - cosmosDocumentQueryFactory); + cosmosDocumentQueryFactory, + mockQueueClientForSearchParams); var bundleConfiguration = new BundleConfiguration() { SupportsBundleOrchestrator = true }; var bundleOptions = Substitute.For>(); @@ -330,6 +336,7 @@ public virtual async Task InitializeAsync() await _searchParameterDefinitionManager.EnsureInitializedAsync(CancellationToken.None); var queueClient = new TestQueueClient(); + _testQueueClient = queueClient; _fhirOperationDataStore = new CosmosFhirOperationDataStore( queueClient, documentClient, @@ -447,6 +454,11 @@ object IServiceProvider.GetService(Type serviceType) return _searchIndexer; } + if (serviceType == typeof(TestQueueClient)) + { + return _testQueueClient; + } + return null; } } diff --git a/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/FhirStorageTestsFixture.cs b/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/FhirStorageTestsFixture.cs index 74ea56cce6..1c53995b21 100644 --- a/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/FhirStorageTestsFixture.cs +++ b/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/FhirStorageTestsFixture.cs @@ -182,6 +182,19 @@ internal IFhirOperationDataStore SqlServerOperationDataStore } } + internal IFhirOperationDataStore TestSqlServerOperationDataStore + { + get + { + if (_fixture is SqlServerFhirStorageTestsFixture sqlFixture) + { + return sqlFixture.TestSqlServerOperationDataStore; + } + + return null; + } + } + public IQueueClient QueueClient => _fixture.GetRequiredService(); public IServiceProvider Service => _fixture; @@ -331,10 +344,8 @@ public async Task InitializeAsync() ModelInfoProvider.Instance, searchParameterSupportResolver, dataStoreSearchParameterValidator, - () => OperationDataStore.CreateMockScope(), () => SearchService.CreateMockScope(), DataStore.CreateMockScopeProvider(), - resourceWrapperFactory, NullLogger.Instance); var deleter = new DeletionService( diff --git a/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/SqlServerFhirStorageTestsFixture.cs b/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/SqlServerFhirStorageTestsFixture.cs index f6865c655d..82c21d74ab 100644 --- a/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/SqlServerFhirStorageTestsFixture.cs +++ b/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/SqlServerFhirStorageTestsFixture.cs @@ -69,6 +69,7 @@ public class SqlServerFhirStorageTestsFixture : IServiceProvider, IAsyncLifetime private SqlServerFhirDataStore _fhirDataStore; private IFhirOperationDataStore _fhirOperationDataStore; private SqlServerFhirOperationDataStore _sqlServerFhirOperationDataStore; + private SqlServerFhirOperationDataStore _testSqlServerFhirOperationDataStore; private SqlServerFhirStorageTestHelper _testHelper; private SchemaUpgradeRunner _schemaUpgradeRunner; private FilebasedSearchParameterStatusDataStore _filebasedSearchParameterStatusDataStore; @@ -77,6 +78,7 @@ public class SqlServerFhirStorageTestsFixture : IServiceProvider, IAsyncLifetime private SupportedSearchParameterDefinitionManager _supportedSearchParameterDefinitionManager; private SearchParameterStatusManager _searchParameterStatusManager; private SqlQueueClient _sqlQueueClient; + private TestQueueClient _testQueueClient; private FhirSqlServerConfiguration _fhirSqlConfiguration; public SqlServerFhirStorageTestsFixture() @@ -110,6 +112,8 @@ internal SqlServerFhirStorageTestsFixture(int maximumSupportedSchemaVersion, str internal SqlServerFhirOperationDataStore SqlServerOperationDataStore => _sqlServerFhirOperationDataStore; + internal SqlServerFhirOperationDataStore TestSqlServerOperationDataStore => _testSqlServerFhirOperationDataStore; + internal SqlTransactionHandler SqlTransactionHandler { get; private set; } internal SqlConnectionWrapperFactory SqlConnectionWrapperFactory { get; private set; } @@ -184,8 +188,12 @@ public async Task InitializeAsync() NullLogger.Instance); SqlServerFhirModel = sqlServerFhirModel; - // the test queue client may not be enough for these tests. will need to look back into this. - var queueClient = new TestQueueClient(); + // Create the real SqlQueueClient to use throughout the fixture + var sqlQueueClient = new SqlQueueClient(SchemaInformation, SqlRetryService, NullLogger.Instance); + _sqlQueueClient = sqlQueueClient; + + // TestQueueClient is still used by export tests + _testQueueClient = new TestQueueClient(); // Add custom logic to set up the AzurePipelinesCredential if we are running in Azure Pipelines string federatedClientID = EnvironmentVariables.GetEnvironmentVariable(KnownEnvironmentVariableNames.AzureSubscriptionClientId); @@ -199,7 +207,7 @@ public async Task InitializeAsync() SqlAuthenticationProvider.SetProvider(SqlAuthenticationMethod.ActiveDirectoryWorkloadIdentity, new SqlAzurePipelinesWorkloadIdentityAuthenticationProvider(azurePipelinesCredential)); } - _testHelper = new SqlServerFhirStorageTestHelper(_initialConnectionString, MasterDatabaseName, sqlServerFhirModel, SqlConnectionBuilder, queueClient, SchemaInformation); + _testHelper = new SqlServerFhirStorageTestHelper(_initialConnectionString, MasterDatabaseName, sqlServerFhirModel, SqlConnectionBuilder, _testQueueClient, SchemaInformation); await _testHelper.CreateAndInitializeDatabase(_databaseName, _maximumSupportedSchemaVersion, CancellationToken.None); var searchParameterToSearchValueTypeMap = new SearchParameterToSearchValueTypeMap(); @@ -247,11 +255,15 @@ public async Task InitializeAsync() importErrorSerializer, new SqlStoreClient(SqlRetryService, NullLogger.Instance, SchemaInformation)); - _fhirOperationDataStore = new SqlServerFhirOperationDataStore(SqlConnectionWrapperFactory, queueClient, NullLogger.Instance, NullLoggerFactory.Instance); - - var sqlQueueClient = new SqlQueueClient(SchemaInformation, SqlRetryService, NullLogger.Instance); + // Create operation data store with real SqlQueueClient for reindex tests _sqlServerFhirOperationDataStore = new SqlServerFhirOperationDataStore(SqlConnectionWrapperFactory, sqlQueueClient, NullLogger.Instance, NullLoggerFactory.Instance); + // Create operation data store with TestQueueClient for export tests + _testSqlServerFhirOperationDataStore = new SqlServerFhirOperationDataStore(SqlConnectionWrapperFactory, _testQueueClient, NullLogger.Instance, NullLoggerFactory.Instance); + + // Use the real SqlQueueClient for IFhirOperationDataStore + _fhirOperationDataStore = _sqlServerFhirOperationDataStore; + _fhirRequestContextAccessor.RequestContext.CorrelationId.Returns(Guid.NewGuid().ToString()); _fhirRequestContextAccessor.RequestContext.RouteName.Returns("routeName"); @@ -410,6 +422,11 @@ object IServiceProvider.GetService(Type serviceType) return _sqlQueueClient; } + if (serviceType == typeof(TestQueueClient)) + { + return _testQueueClient; + } + if (serviceType == typeof(TestSqlHashCalculator)) { return SqlQueryHashCalculator as TestSqlHashCalculator; diff --git a/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/SqlServerSearchParameterStatusDataStoreTests.cs b/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/SqlServerSearchParameterStatusDataStoreTests.cs index 871a41af47..24b668afeb 100644 --- a/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/SqlServerSearchParameterStatusDataStoreTests.cs +++ b/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/SqlServerSearchParameterStatusDataStoreTests.cs @@ -15,8 +15,11 @@ using Microsoft.Health.Fhir.SqlServer.Features.Storage.Registry; using Microsoft.Health.Fhir.Tests.Common; using Microsoft.Health.Fhir.Tests.Common.FixtureParameters; +using Microsoft.Health.JobManagement; using Microsoft.Health.Test.Utilities; +using Newtonsoft.Json; using Xunit; +using JobConflictException = Microsoft.Health.Fhir.Core.Features.Operations.JobConflictException; namespace Microsoft.Health.Fhir.Tests.Integration.Persistence { @@ -580,5 +583,139 @@ public async Task GivenGetSearchParameterStatuses_WhenCalled_ThenReturnsConsiste Assert.True(sqlStatus.Id > 0, "SqlServerResourceSearchParameterStatus should have a valid Id"); }); } + + [Fact] + public async Task GivenActiveReindexJob_WhenUpsertingSearchParameterStatus_ThenJobConflictExceptionIsThrown() + { + // This test validates that search parameter status updates are blocked when a reindex job is running + // This ensures bulk delete operations targeting SearchParameter resources will fail during reindex + + await _fixture.SearchParameterOperations.GetAndApplySearchParameterUpdates(CancellationToken.None); + + // Arrange - Create a test search parameter status + var testUri = "http://hl7.org/fhir/SearchParameter/Test-ReindexConflict-" + Guid.NewGuid(); + var status = new ResourceSearchParameterStatus + { + Uri = new Uri(testUri), + Status = SearchParameterStatus.Enabled, + IsPartiallySupported = false, + LastUpdated = _fixture.SearchParameterOperations.SearchParamLastUpdated, + }; + + // Create a reindex job using OperationDataStore + var reindexJobRecord = new Core.Features.Operations.Reindex.Models.ReindexJobRecord(new List()); + var reindexJobWrapper = await _fixture.OperationDataStore.CreateReindexJobAsync(reindexJobRecord, CancellationToken.None); + var jobId = long.Parse(reindexJobWrapper.JobRecord.Id); + + try + { + // Act & Assert - Upserting without reindexId should throw JobConflictException + var exception = await Assert.ThrowsAsync(async () => + { + await _fixture.SearchParameterStatusDataStore.UpsertStatuses([status], CancellationToken.None); + }); + + Assert.Contains("reindex", exception.Message, StringComparison.OrdinalIgnoreCase); + Assert.Contains(jobId.ToString(), exception.Message); + } + finally + { + // Cleanup - Cancel the reindex job + var jobInfo = await _fixture.QueueClient.DequeueAsync((byte)QueueType.Reindex, "test-cleanup", 0, CancellationToken.None, jobId); + if (jobInfo != null) + { + jobInfo.Status = JobStatus.Cancelled; + await _fixture.QueueClient.CompleteJobAsync(jobInfo, false, CancellationToken.None); + } + } + } + + [Fact] + public async Task GivenActiveReindexJob_WhenUpsertingWithReindexId_ThenSucceeds() + { + // This test validates that the reindex job itself can update search parameter statuses + // by passing its job ID + + // Arrange - Create a test search parameter status + var testUri = "http://hl7.org/fhir/SearchParameter/Test-ReindexBypass-" + Guid.NewGuid(); + var status = new ResourceSearchParameterStatus + { + Uri = new Uri(testUri), + Status = SearchParameterStatus.Enabled, + IsPartiallySupported = false, + LastUpdated = DateTimeOffset.UtcNow, // no need to set value to SearchParamLastUpdated because reindex is excluded from concurrency checks + }; + + // Start a reindex job using OperationDataStore + var reindexJobRecord = new Core.Features.Operations.Reindex.Models.ReindexJobRecord(new List()); + var reindexJobWrapper = await _fixture.OperationDataStore.CreateReindexJobAsync(reindexJobRecord, CancellationToken.None); + var jobId = long.Parse(reindexJobWrapper.JobRecord.Id); + var jobInfo = await _fixture.QueueClient.DequeueAsync((byte)QueueType.Reindex, "test", 0, CancellationToken.None, jobId); + Assert.NotNull(jobInfo); + + try + { + // Act - Upserting with reindexId should succeed (bypass the concurrency check) + await _fixture.SearchParameterStatusDataStore.UpsertStatuses([status], CancellationToken.None, reindexId: jobId); + + // Assert - Verify the status was created + var allStatuses = await _fixture.SearchParameterStatusDataStore.GetSearchParameterStatuses(CancellationToken.None); + var createdStatus = allStatuses.FirstOrDefault(s => s.Uri.OriginalString == testUri); + Assert.NotNull(createdStatus); + Assert.Equal(SearchParameterStatus.Enabled, createdStatus.Status); + } + finally + { + // Cleanup + await _testHelper.DeleteSearchParameterStatusAsync(testUri); + if (jobInfo != null) + { + jobInfo.Status = JobStatus.Cancelled; + await _fixture.QueueClient.CompleteJobAsync(jobInfo, false, CancellationToken.None); + } + } + } + + [Fact] + public async Task GivenNoActiveReindexJob_WhenUpsertingSearchParameterStatus_ThenSucceeds() + { + // This test validates that when no reindex is running, search parameter updates work normally + // This ensures bulk delete operations can proceed when no reindex is active + + // Arrange - Clean up any active reindex jobs + var activeJobs = await _fixture.QueueClient.GetActiveJobsByQueueTypeAsync((byte)QueueType.Reindex, false, CancellationToken.None); + foreach (var activeJob in activeJobs) + { + activeJob.Status = JobStatus.Cancelled; + await _fixture.QueueClient.CompleteJobAsync(activeJob, false, CancellationToken.None); + } + + await _fixture.SearchParameterOperations.GetAndApplySearchParameterUpdates(CancellationToken.None); + + var testUri = "http://hl7.org/fhir/SearchParameter/Test-NoReindex-" + Guid.NewGuid(); + var status = new ResourceSearchParameterStatus + { + Uri = new Uri(testUri), + Status = SearchParameterStatus.Enabled, + IsPartiallySupported = false, + LastUpdated = _fixture.SearchParameterOperations.SearchParamLastUpdated, + }; + + try + { + // Act - Should succeed without any reindex job running + await _fixture.SearchParameterStatusDataStore.UpsertStatuses([status], CancellationToken.None); + + // Assert - Verify the status was created + var allStatuses = await _fixture.SearchParameterStatusDataStore.GetSearchParameterStatuses(CancellationToken.None); + var createdStatus = allStatuses.FirstOrDefault(s => s.Uri.OriginalString == testUri); + Assert.NotNull(createdStatus); + Assert.Equal(SearchParameterStatus.Enabled, createdStatus.Status); + } + finally + { + await _testHelper.DeleteSearchParameterStatusAsync(testUri); + } + } } }