diff --git a/build/jobs/provision-deploy.yml b/build/jobs/provision-deploy.yml index 2186eb5abd..3b74801655 100644 --- a/build/jobs/provision-deploy.yml +++ b/build/jobs/provision-deploy.yml @@ -71,7 +71,8 @@ jobs: $additionalProperties["SqlServer__AllowDatabaseCreation"] = "true" # Cosmos DB autoscale is configured in the ARM template (10,000 RU max) $additionalProperties["TaskHosting__PollingFrequencyInSeconds"] = 1 - $additionalProperties["FhirServer__CoreFeatures__SearchParameterCacheRefreshIntervalSeconds"] = 2 + $additionalProperties["FhirServer__CoreFeatures__SearchParameterCacheRefreshIntervalSeconds"] = 1 + $additionalProperties["FhirServer__Operations__Reindex__CacheRefreshWaitMultiplier"] = 6 $additionalProperties["ASPNETCORE_FORWARDEDHEADERS_ENABLED"] = "true" $webAppName = "${{ parameters.webAppName }}".ToLower() @@ -90,7 +91,7 @@ jobs: enableExport = $true enableConvertData = $true enableImport = $true - backgroundTaskCount = 2 + backgroundTaskCount = 4 enableReindex = if ("${{ parameters.reindexEnabled }}" -eq "true") { $true } else { $false } registryName = '$(azureContainerRegistry)' imageTag = '${{ parameters.imageTag }}' diff --git a/src/Microsoft.Health.Fhir.Api/Resources.Designer.cs b/src/Microsoft.Health.Fhir.Api/Resources.Designer.cs index 18b1d96a4c..1a3d8f8d57 100644 --- a/src/Microsoft.Health.Fhir.Api/Resources.Designer.cs +++ b/src/Microsoft.Health.Fhir.Api/Resources.Designer.cs @@ -222,6 +222,33 @@ public static string CustomAuditHeaderTooLarge { } } + /// + /// Looks up a localized string similar to Input search parameters have duplicate codes [{0}] and Urls [{1}].. + /// + public static string DuplicateSearchParamCodesAndUrlsInBundle { + get { + return ResourceManager.GetString("DuplicateSearchParamCodesAndUrlsInBundle", resourceCulture); + } + } + + /// + /// Looks up a localized string similar to Input search parameters have duplicate codes [{0}]. + /// + public static string DuplicateSearchParamCodesInBundle { + get { + return ResourceManager.GetString("DuplicateSearchParamCodesInBundle", resourceCulture); + } + } + + /// + /// Looks up a localized string similar to Input search parameters have duplicate Urls [{0}].. + /// + public static string DuplicateSearchParamUrlsInBundle { + get { + return ResourceManager.GetString("DuplicateSearchParamUrlsInBundle", resourceCulture); + } + } + /// /// Looks up a localized string similar to One or more invalid parameters are found: {0}. /// @@ -637,29 +664,29 @@ public static string NotAbleToCreateTheFinalResultsOfAnOperation { } /// - /// Looks up a localized string similar to Content-Type must be application/x-www-form-urlencoded.. + /// Looks up a localized string similar to The requested route was not found.. /// - public static string OAuth2ContentTypeMustBeFormUrlEncoded { + public static string NotFoundException { get { - return ResourceManager.GetString("OAuth2ContentTypeMustBeFormUrlEncoded", resourceCulture); + return ResourceManager.GetString("NotFoundException", resourceCulture); } } /// - /// Looks up a localized string similar to token parameter is required.. + /// Looks up a localized string similar to Content-Type must be application/x-www-form-urlencoded.. /// - public static string OAuth2TokenParameterRequired { + public static string OAuth2ContentTypeMustBeFormUrlEncoded { get { - return ResourceManager.GetString("OAuth2TokenParameterRequired", resourceCulture); + return ResourceManager.GetString("OAuth2ContentTypeMustBeFormUrlEncoded", resourceCulture); } } /// - /// Looks up a localized string similar to The requested route was not found.. + /// Looks up a localized string similar to token parameter is required.. /// - public static string NotFoundException { + public static string OAuth2TokenParameterRequired { get { - return ResourceManager.GetString("NotFoundException", resourceCulture); + return ResourceManager.GetString("OAuth2TokenParameterRequired", resourceCulture); } } diff --git a/src/Microsoft.Health.Fhir.Api/Resources.resx b/src/Microsoft.Health.Fhir.Api/Resources.resx index f7ab007328..1d4503ea85 100644 --- a/src/Microsoft.Health.Fhir.Api/Resources.resx +++ b/src/Microsoft.Health.Fhir.Api/Resources.resx @@ -459,4 +459,15 @@ The number of the parameter must not be more than one: {0} - + + Input search parameters have duplicate codes [{0}] + {0} is comma delimited list of dup codes + + + Input search parameters have duplicate Urls [{0}]. + {0} is comma delimited list of duplicate Urls + + + Input search parameters have duplicate codes [{0}] and Urls [{1}]. + + \ No newline at end of file diff --git a/src/Microsoft.Health.Fhir.Core/Configs/ReindexJobConfiguration.cs b/src/Microsoft.Health.Fhir.Core/Configs/ReindexJobConfiguration.cs index b413158ce2..d1395cad81 100644 --- a/src/Microsoft.Health.Fhir.Core/Configs/ReindexJobConfiguration.cs +++ b/src/Microsoft.Health.Fhir.Core/Configs/ReindexJobConfiguration.cs @@ -44,10 +44,22 @@ public ReindexJobConfiguration() /// /// Controls the multiplier applied to the SearchParameterCacheRefreshIntervalSeconds - /// to determine time to wait for search param cache refresh + /// to determine time to wait for search param cache refresh. Relevant for Cosmos only. /// public int CacheRefreshWaitMultiplier { get; set; } = 3; + /// + /// Controls the multiplier applied to the SearchParameterCacheRefreshIntervalSeconds + /// to determine max time to wait for search param cache refresh. Relevant for SQL only. + /// + public int CacheUpdateMaxWaitMultiplier { get; set; } = 20; + + /// + /// Controls the multiplier applied to the SearchParameterCacheRefreshIntervalSeconds + /// to determine the time interval to retrieve active host names. Relevant for SQL only. + /// + public int ActiveHostsEventsMultiplier { get; set; } = 9; + /// /// Controls how many surrogate ID ranges are fetched per database call when calculating /// job ranges. Uses batched calls to avoid timeout on large tables. diff --git a/src/Microsoft.Health.Fhir.Core/Features/Conformance/GetSmartConfigurationHandler.cs b/src/Microsoft.Health.Fhir.Core/Features/Conformance/GetSmartConfigurationHandler.cs index fc2bfaf5f2..2ac45cdf98 100644 --- a/src/Microsoft.Health.Fhir.Core/Features/Conformance/GetSmartConfigurationHandler.cs +++ b/src/Microsoft.Health.Fhir.Core/Features/Conformance/GetSmartConfigurationHandler.cs @@ -52,15 +52,20 @@ public async Task Handle(GetSmartConfigurationReq Uri authorizationEndpoint; Uri tokenEndpoint; + string issuer; + string jwksUri; if (_securityConfiguration.EnableAadSmartOnFhirProxy) { authorizationEndpoint = new Uri(request.BaseUri, "AadSmartOnFhirProxy/authorize"); tokenEndpoint = new Uri(request.BaseUri, "AadSmartOnFhirProxy/token"); + + // Still resolve issuer and jwks_uri from OIDC discovery + (_, _, issuer, jwksUri) = await _oidcDiscoveryService.ResolveEndpointsAsync(baseEndpoint, cancellationToken); } else { - (authorizationEndpoint, tokenEndpoint) = await _oidcDiscoveryService.ResolveEndpointsAsync(baseEndpoint, cancellationToken); + (authorizationEndpoint, tokenEndpoint, issuer, jwksUri) = await _oidcDiscoveryService.ResolveEndpointsAsync(baseEndpoint, cancellationToken); } ICollection capabilities = new List( @@ -110,6 +115,10 @@ public async Task Handle(GetSmartConfigurationReq "code", }; + string introspectionEndpoint = !string.IsNullOrEmpty(_smartIdentityProviderConfiguration.Introspection) + ? _smartIdentityProviderConfiguration.Introspection + : new Uri(request.BaseUri, "connect/introspect").ToString(); + return new GetSmartConfigurationResponse( authorizationEndpoint, tokenEndpoint, @@ -119,9 +128,11 @@ public async Task Handle(GetSmartConfigurationReq grantTypesSupported, tokenEndpointAuthMethodsSupported, responseTypesSupported, - _smartIdentityProviderConfiguration.Introspection, + introspectionEndpoint, _smartIdentityProviderConfiguration.Management, - _smartIdentityProviderConfiguration.Revocation); + _smartIdentityProviderConfiguration.Revocation, + issuer, + jwksUri); } catch (Exception e) when (e is ArgumentNullException || e is UriFormatException) { diff --git a/src/Microsoft.Health.Fhir.Core/Features/Definition/SearchParameterDefinitionManager.cs b/src/Microsoft.Health.Fhir.Core/Features/Definition/SearchParameterDefinitionManager.cs index 045d9a800f..6a7ff93952 100644 --- a/src/Microsoft.Health.Fhir.Core/Features/Definition/SearchParameterDefinitionManager.cs +++ b/src/Microsoft.Health.Fhir.Core/Features/Definition/SearchParameterDefinitionManager.cs @@ -573,6 +573,11 @@ private bool TryGetFromTypeLookup(string resourceType, string code, out SearchPa [System.Diagnostics.CodeAnalysis.SuppressMessage("Performance", "CA1859:Use concrete types when possible for improved performance", Justification = "Collection defined on model")] private ICollection GetDerivedResourceTypes(IReadOnlyCollection resourceTypes) + { + return GetDerivedResourceTypes(_modelInfoProvider, resourceTypes); + } + + public static ICollection GetDerivedResourceTypes(IModelInfoProvider modelInfoProvider, IReadOnlyCollection resourceTypes) { var completeResourceList = new HashSet(resourceTypes); @@ -580,7 +585,7 @@ private ICollection GetDerivedResourceTypes(IReadOnlyCollection { if (baseResourceType == KnownResourceTypes.Resource) { - completeResourceList.UnionWith(_modelInfoProvider.GetResourceTypeNames().ToHashSet()); + completeResourceList.UnionWith(modelInfoProvider.GetResourceTypeNames().ToHashSet()); // We added all possible resource types, so no need to continue break; @@ -588,7 +593,7 @@ private ICollection GetDerivedResourceTypes(IReadOnlyCollection if (baseResourceType == KnownResourceTypes.DomainResource) { - var domainResourceChildResourceTypes = _modelInfoProvider.GetResourceTypeNames().ToHashSet(); + var domainResourceChildResourceTypes = modelInfoProvider.GetResourceTypeNames().ToHashSet(); // Remove types that inherit from Resource directly domainResourceChildResourceTypes.Remove(KnownResourceTypes.Binary); diff --git a/src/Microsoft.Health.Fhir.Core/Features/Operations/Reindex/ReindexOrchestratorJob.cs b/src/Microsoft.Health.Fhir.Core/Features/Operations/Reindex/ReindexOrchestratorJob.cs index 7162064d78..d6bdf3ddca 100644 --- a/src/Microsoft.Health.Fhir.Core/Features/Operations/Reindex/ReindexOrchestratorJob.cs +++ b/src/Microsoft.Health.Fhir.Core/Features/Operations/Reindex/ReindexOrchestratorJob.cs @@ -6,12 +6,14 @@ using System; using System.Collections.Generic; using System.Collections.Immutable; +using System.Diagnostics; using System.Linq; using System.Security.Cryptography; using System.Threading; using System.Threading.Tasks; using EnsureThat; using Hl7.Fhir.Model; +using Hl7.Fhir.Utility; using Microsoft.AspNetCore.JsonPatch.Internal; using Microsoft.Data.SqlClient; using Microsoft.Extensions.Logging; @@ -120,7 +122,7 @@ public ReindexOrchestratorJob( _searchParameterStatusManager = searchParameterStatusManager; _searchParameterOperations = searchParameterOperations; _operationsConfiguration = operationsConfiguration.Value; - _searchParameterCacheRefreshIntervalSeconds = Math.Max(1, coreFeatureConfiguration.Value.SearchParameterCacheRefreshIntervalSeconds); + _searchParameterCacheRefreshIntervalSeconds = coreFeatureConfiguration.Value.SearchParameterCacheRefreshIntervalSeconds; // Determine support for surrogate ID ranging once // This is to ensure Gen1 Reindex still works as expected but we still maintain perf on job inseration to SQL @@ -191,45 +193,64 @@ public async Task ExecuteAsync(JobInfo jobInfo, CancellationToken cancel private async Task RefreshSearchParameterCache(bool isReindexStart) { - // Wait for the background cache refresh service to complete N successful refresh cycles. - // This ensures all instances (including processing pods) have the latest search parameter definitions. var suffix = isReindexStart ? "Start" : "End"; _logger.LogJobInformation(_jobInfo, $"Reindex orchestrator job started cache refresh at the {suffix}."); await TryLogEvent($"ReindexOrchestratorJob={_jobInfo.Id}.ExecuteAsync.{suffix}", "Warn", "Started", null, _cancellationToken); - // First, wait for the local background refresh service to complete N cycles. - // This ensures _searchParamLastUpdated is up-to-date on THIS instance before - // we use it as the convergence target for cross-instance checks. - await _searchParameterOperations.WaitForRefreshCyclesAsync(_operationsConfiguration.Reindex.CacheRefreshWaitMultiplier, _cancellationToken); - if (_isSurrogateIdRangingSupported) { - // SQL Server: After local refresh, verify ALL instances have converged to - // the same SearchParamLastUpdated via the EventLog table. This prevents the + // SQL Server: Wait for all instances to update their cache. This prevents the // orchestrator from creating reindex ranges while other instances still have // stale search parameter caches and would write resources with wrong hashes. - // Use the same lookback as active host detection so we do not miss qualifying - // refresh events that occurred shortly before this instance entered the wait. - var activeHostsSince = DateTime.UtcNow.AddSeconds(-20 * _searchParameterCacheRefreshIntervalSeconds); - var syncStartDate = activeHostsSince; - await _searchParameterOperations.WaitForAllInstancesCacheConsistencyAsync(syncStartDate, activeHostsSince, _cancellationToken); + var updateEventsSince = isReindexStart ? _jobInfo.StartDate.Value : DateTime.UtcNow; + var isConsistent = await WaitForAllInstancesCacheSyncAsync(updateEventsSince, _cancellationToken); + if (!isConsistent) + { + var msg = "Unable to sync search parameter cache. Please resubmit reindex. If issue persists please contact your administrator."; + _logger.LogJobError(_jobInfo, msg); + await TryLogEvent($"ReindexOrchestratorJob={_jobInfo.Id}.ExecuteAsync.{suffix}", "Error", msg, null, _cancellationToken); + throw new JobExecutionException(msg, false); + } } else { // Cosmos DB: There is no EventLog-based convergence tracking, so wait a fixed - // delay to allow all instances to refresh their search parameter caches from - // the shared Cosmos container. + // delay to allow all instances to refresh their search parameter caches. var delayMs = _operationsConfiguration.Reindex.CacheRefreshWaitMultiplier * _searchParameterCacheRefreshIntervalSeconds * 1000; - _logger.LogJobInformation(_jobInfo, "Cosmos DB detected — waiting {DelayMs}ms for cache propagation across instances.", delayMs); + _logger.LogJobInformation(_jobInfo, $"Cosmos DB detected — waiting {delayMs}ms for cache propagation across instances."); await Task.Delay(delayMs, _cancellationToken); } - // Update the reindex job record with the latest hash map var currentDate = _searchParameterOperations.SearchParamLastUpdated.HasValue ? _searchParameterOperations.SearchParamLastUpdated.Value : DateTimeOffset.MinValue; _searchParamLastUpdated = currentDate; _logger.LogJobInformation(_jobInfo, $"Reindex orchestrator job completed cache refresh at the {suffix}: SearchParamLastUpdated {_searchParamLastUpdated}"); await TryLogEvent($"ReindexOrchestratorJob={_jobInfo.Id}.ExecuteAsync.{suffix}", "Warn", $"SearchParamLastUpdated={_searchParamLastUpdated.ToString("yyyy-MM-dd HH:mm:ss.fff")}", null, _cancellationToken); + + async Task WaitForAllInstancesCacheSyncAsync(DateTime updateEventsSince, CancellationToken cancellationToken) + { + var start = Stopwatch.StartNew(); + var maxWaitTime = TimeSpan.FromSeconds(_operationsConfiguration.Reindex.CacheUpdateMaxWaitMultiplier * _searchParameterCacheRefreshIntervalSeconds); + var waitInterval = TimeSpan.FromSeconds(_searchParameterCacheRefreshIntervalSeconds); + var activeHostsSince = DateTime.UtcNow.AddSeconds((-1) * _operationsConfiguration.Reindex.ActiveHostsEventsMultiplier * _searchParameterCacheRefreshIntervalSeconds); + CacheConsistencyResult result = null; + while (start.Elapsed < maxWaitTime) + { + result = await _searchParameterStatusManager.CheckCacheConsistencyAsync(updateEventsSince, activeHostsSince, cancellationToken); + + if (result.IsConsistent) + { + var logDate = _searchParameterOperations.SearchParamLastUpdated.HasValue ? _searchParameterOperations.SearchParamLastUpdated.Value : DateTimeOffset.MinValue; + _logger.LogJobInformation(_jobInfo, $"Cache sync check: All {result.ActiveHosts} active host(s) have converged to SearchParamLastUpdated={logDate.ToString("yyyy-MM-dd HH:mm:ss.fff")}."); + break; + } + + _logger.LogJobInformation(_jobInfo, $"Cache sync check: {result.ConvergedHosts}/{result.ActiveHosts} hosts synced. Waiting..."); + await Task.Delay(waitInterval, cancellationToken); + } + + return result != null && result.IsConsistent; + } } private async Task> CreateReindexProcessingJobsAsync(CancellationToken cancellationToken) diff --git a/src/Microsoft.Health.Fhir.Core/Features/Operations/SmartConfigurationResult.cs b/src/Microsoft.Health.Fhir.Core/Features/Operations/SmartConfigurationResult.cs index 843d78a116..a925107bb8 100644 --- a/src/Microsoft.Health.Fhir.Core/Features/Operations/SmartConfigurationResult.cs +++ b/src/Microsoft.Health.Fhir.Core/Features/Operations/SmartConfigurationResult.cs @@ -37,7 +37,9 @@ public SmartConfigurationResult( ICollection responseTypesSupported = null, string introspectionEndpoint = null, string managementEndpoint = null, - string revocationEndpoint = null) + string revocationEndpoint = null, + string issuer = null, + string jwksUri = null) { EnsureArg.IsNotNull(authorizationEndpoint, nameof(authorizationEndpoint)); EnsureArg.IsNotNull(tokenEndpoint, nameof(tokenEndpoint)); @@ -54,6 +56,8 @@ public SmartConfigurationResult( IntrospectionEndpoint = introspectionEndpoint; ManagementEndpoint = managementEndpoint; RevocationEndpoint = revocationEndpoint; + Issuer = issuer; + JwksUri = jwksUri; } [JsonConstructor] @@ -93,5 +97,13 @@ public SmartConfigurationResult() [JsonProperty("revocation_endpoint")] public string RevocationEndpoint { get; } + + [JsonProperty("issuer")] + public string Issuer { get; } + +#pragma warning disable CA1056 // URI-like properties should not be strings + [JsonProperty("jwks_uri")] + public string JwksUri { get; } +#pragma warning restore CA1056 // URI-like properties should not be strings } } 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 7978224e24..2185401829 100644 --- a/src/Microsoft.Health.Fhir.Core/Features/Search/Parameters/ISearchParameterOperations.cs +++ b/src/Microsoft.Health.Fhir.Core/Features/Search/Parameters/ISearchParameterOperations.cs @@ -36,25 +36,5 @@ public interface ISearchParameterOperations Task GetAndApplySearchParameterUpdates(CancellationToken cancellationToken, bool forceFullRefresh = false); string GetSearchParameterHash(string resourceType); - - /// - /// Waits for the specified number of successful cache refresh cycles to complete. - /// Each cycle corresponds to a successful execution of the background cache refresh service. - /// - /// The number of successful refresh cycles to wait for. If zero or negative, returns immediately. - /// Cancellation token. - /// A task that completes when the requested number of refresh cycles have occurred. - Task WaitForRefreshCyclesAsync(int cycleCount, CancellationToken cancellationToken); - - /// - /// Waits until all active server instances have converged their search parameter caches - /// to the current instance's SearchParamLastUpdated timestamp. For SQL, this verifies via - /// the EventLog table. For Cosmos/File-based, this returns immediately. - /// - /// Only cache refresh sync records on or after this time are considered for convergence. - /// Only active-host evidence on or after this time is considered. - /// Cancellation token. - /// A task that completes when all instances have consistent caches. - Task WaitForAllInstancesCacheConsistencyAsync(DateTime syncStartDate, DateTime activeHostsSince, CancellationToken cancellationToken); } } 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 f8e9eb462b..d2ec72555d 100644 --- a/src/Microsoft.Health.Fhir.Core/Features/Search/Parameters/SearchParameterOperations.cs +++ b/src/Microsoft.Health.Fhir.Core/Features/Search/Parameters/SearchParameterOperations.cs @@ -5,6 +5,7 @@ using System; using System.Collections.Generic; +using System.Diagnostics; using System.Linq; using System.Threading; using System.Threading.Tasks; @@ -36,8 +37,6 @@ public class SearchParameterOperations : ISearchParameterOperations private readonly Func> _searchServiceFactory; private readonly ILogger _logger; private DateTimeOffset? _searchParamLastUpdated; - private int _activeConsistencyWaiters; - private volatile TaskCompletionSource _refreshSignal = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); public SearchParameterOperations( SearchParameterStatusManager searchParameterStatusManager, @@ -84,135 +83,6 @@ public string GetSearchParameterHash(string resourceType) } } - /// - public async Task WaitForRefreshCyclesAsync(int cycleCount, CancellationToken cancellationToken) - { - if (cycleCount <= 0) - { - return; - } - - // Safety net: if the background service is not publishing notifications (e.g. crashed), - // fail fast rather than blocking indefinitely. Under normal conditions with a 20-second - // refresh interval, even 3 cycles would complete in ~60 seconds. - using var timeoutCts = new CancellationTokenSource(TimeSpan.FromMinutes(5)); - using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeoutCts.Token); - - for (int i = 0; i < cycleCount; i++) - { - linkedCts.Token.ThrowIfCancellationRequested(); - - // Capture the current signal before awaiting - var currentSignal = _refreshSignal; - using var registration = linkedCts.Token.Register(() => currentSignal.TrySetCanceled()); - - try - { - await currentSignal.Task; - } - catch (TaskCanceledException) when (timeoutCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested) - { - throw new TimeoutException( - $"SearchParameterCacheRefreshBackgroundService did not complete {cycleCount} refresh cycle(s) within 5 minutes. The server may be in an unhealthy state."); - } - } - } - - /// - public async Task WaitForAllInstancesCacheConsistencyAsync(DateTime syncStartDate, DateTime activeHostsSince, CancellationToken cancellationToken) - { - if (!_searchParamLastUpdated.HasValue) - { - _logger.LogInformation("SearchParamLastUpdated is null — skipping cross-instance cache consistency check."); - return; - } - - var targetTimestamp = _searchParamLastUpdated.Value.ToString("yyyy-MM-dd HH:mm:ss.fffffff"); - var waitStart = DateTime.UtcNow; - - Interlocked.Increment(ref _activeConsistencyWaiters); - - try - { - await TryLogConsistencyWaitEventAsync( - "Warn", - $"Target={targetTimestamp} PollIntervalSeconds=30 TimeoutMinutes=10 SyncStartDate={syncStartDate:O} ActiveHostsSince={activeHostsSince:O}", - null, - cancellationToken); - - // Poll with a timeout — same 10-minute safety net as WaitForRefreshCyclesAsync - using var timeoutCts = new CancellationTokenSource(TimeSpan.FromMinutes(10)); - using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeoutCts.Token); - - while (!linkedCts.Token.IsCancellationRequested) - { - var result = await _searchParameterStatusManager.CheckCacheConsistencyAsync(targetTimestamp, syncStartDate, activeHostsSince, linkedCts.Token); - - if (result.IsConsistent) - { - var elapsedMilliseconds = (long)(DateTime.UtcNow - waitStart).TotalMilliseconds; - - _logger.LogInformation( - "All {TotalActiveHosts} active host(s) have converged to SearchParamLastUpdated={Target}.", - result.TotalActiveHosts, - targetTimestamp); - - await TryLogConsistencyWaitEventAsync( - "Warn", - $"Target={targetTimestamp} TotalActiveHosts={result.TotalActiveHosts} ConvergedHosts={result.ConvergedHosts} ElapsedMs={elapsedMilliseconds}", - waitStart, - cancellationToken); - - return; - } - - _logger.LogInformation( - "Cache consistency check: {ConvergedHosts}/{TotalActiveHosts} hosts converged to SearchParamLastUpdated={Target}. Waiting...", - result.ConvergedHosts, - result.TotalActiveHosts, - targetTimestamp); - - try - { - await Task.Delay(TimeSpan.FromSeconds(30), linkedCts.Token); - } - catch (OperationCanceledException) when (timeoutCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested) - { - var elapsedMilliseconds = (long)(DateTime.UtcNow - waitStart).TotalMilliseconds; - var timeoutMessage = $"Target={targetTimestamp} TotalActiveHosts={result.TotalActiveHosts} ConvergedHosts={result.ConvergedHosts} ElapsedMs={elapsedMilliseconds}"; - - await TryLogConsistencyWaitEventAsync( - "Error", - timeoutMessage, - waitStart, - CancellationToken.None); - - throw new TimeoutException( - $"Not all instances converged to SearchParamLastUpdated={targetTimestamp} within 10 minutes. The server may be in an unhealthy state."); - } - } - - cancellationToken.ThrowIfCancellationRequested(); - } - finally - { - Interlocked.Decrement(ref _activeConsistencyWaiters); - } - } - - private async Task TryLogConsistencyWaitEventAsync(string status, string text, DateTime? startDate, CancellationToken cancellationToken) - { - try - { - using IScoped searchService = _searchServiceFactory(); - await searchService.Value.TryLogEvent(nameof(WaitForAllInstancesCacheConsistencyAsync), status, text, startDate, cancellationToken); - } - catch (Exception ex) - { - _logger.LogWarning(ex, "Failed to log WaitForAllInstancesCacheConsistencyAsync event."); - } - } - public async Task EnsureNoActiveReindexJobAsync(CancellationToken cancellationToken) { using IScoped fhirOperationDataStore = _fhirOperationDataStoreFactory(); @@ -435,47 +305,19 @@ public async Task GetAndApplySearchParameterUpdates(CancellationToken cancellati var inCache = ParametersAreInCache(statusesToFetch, cancellationToken); - DateTimeOffset? searchParamLastUpdatedToLog = null; - // If cache is updated directly and not from the database not all will have corresponding resources. // Do not advance or log the timestamp unless the cache contents are conclusive for this cycle. - if (inCache && allHaveResources) + if (inCache && allHaveResources && results.LastUpdated.HasValue) { - if (results.LastUpdated.HasValue) // this should be the only place in the code to assign last updated - { - _searchParamLastUpdated = results.LastUpdated.Value; - } - - searchParamLastUpdatedToLog = _searchParamLastUpdated; + _searchParamLastUpdated = results.LastUpdated.Value; // this should be the only place in the code to assign last updated } - if (searchParamLastUpdatedToLog.HasValue - && (results.LastUpdated.HasValue || Volatile.Read(ref _activeConsistencyWaiters) > 0)) + if (_searchParamLastUpdated.HasValue) { // Log to EventLog for cross-instance convergence tracking (SQL only; Cosmos/File are no-ops). - // Emit the current cache timestamp for no-op refresh cycles only while a consistency waiter is active. - try - { - var lastUpdatedText = searchParamLastUpdatedToLog.Value.ToString("yyyy-MM-dd HH:mm:ss.fffffff"); - using IScoped searchService = _searchServiceFactory(); - await searchService.Value.TryLogEvent( - "SearchParameterCacheRefresh", - "End", - $"SearchParamLastUpdated={lastUpdatedText}", - null, - cancellationToken); - } - catch (Exception ex) - { - _logger.LogWarning(ex, "Failed to log SearchParameterCacheRefresh event. Cross-instance convergence checks may be affected."); - } + var lastUpdatedText = _searchParamLastUpdated.Value.ToString("yyyy-MM-dd HH:mm:ss.fffffff"); + await _searchParameterStatusManager.TryLogEvent(_searchParameterStatusManager.SearchParamCacheUpdateProcessName, "Warn", lastUpdatedText, null, cancellationToken); } - - // Signal waiters that a refresh cycle has completed. - // This fires every cycle (even when no changes are found) because - // WaitForRefreshCyclesAsync counts completed cycles, not cycles with changes. - var previous = Interlocked.Exchange(ref _refreshSignal, new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously)); - previous.TrySetResult(true); } // This should handle racing condition between saving new parameter on one VM and refreshing cache on the other, diff --git a/src/Microsoft.Health.Fhir.Core/Features/Search/Registry/CacheConsistencyResult.cs b/src/Microsoft.Health.Fhir.Core/Features/Search/Registry/CacheConsistencyResult.cs index 216e93db9d..c288792411 100644 --- a/src/Microsoft.Health.Fhir.Core/Features/Search/Registry/CacheConsistencyResult.cs +++ b/src/Microsoft.Health.Fhir.Core/Features/Search/Registry/CacheConsistencyResult.cs @@ -17,9 +17,9 @@ public class CacheConsistencyResult public bool IsConsistent { get; set; } /// - /// Gets or sets the total number of active hosts discovered. + /// Gets or sets the number of active hosts discovered. /// - public int TotalActiveHosts { get; set; } + public int ActiveHosts { get; set; } /// /// Gets or sets the number of hosts that have converged to the target version. diff --git a/src/Microsoft.Health.Fhir.Core/Features/Search/Registry/FilebasedSearchParameterStatusDataStore.cs b/src/Microsoft.Health.Fhir.Core/Features/Search/Registry/FilebasedSearchParameterStatusDataStore.cs index 6aeb4cb248..3ba8b9e715 100644 --- a/src/Microsoft.Health.Fhir.Core/Features/Search/Registry/FilebasedSearchParameterStatusDataStore.cs +++ b/src/Microsoft.Health.Fhir.Core/Features/Search/Registry/FilebasedSearchParameterStatusDataStore.cs @@ -37,6 +37,8 @@ public FilebasedSearchParameterStatusDataStore( public delegate ISearchParameterStatusDataStore Resolver(); + public string SearchParamCacheUpdateProcessName => null; + public async Task TryLogEvent(string process, string status, string text, DateTime? startDate, CancellationToken cancellationToken) { await Task.CompletedTask; // noop @@ -98,10 +100,9 @@ public void SyncStatuses(IReadOnlyCollection stat // Do nothing. This is only required for SQL. } - public Task CheckCacheConsistencyAsync(string targetSearchParamLastUpdated, DateTime syncStartDate, DateTime activeHostsSince, CancellationToken cancellationToken) + public Task CheckCacheConsistencyAsync(DateTime updateEventsSince, DateTime activeHostsSince, CancellationToken cancellationToken) { - // File-based registry is single-instance only. Always consistent. - return Task.FromResult(new CacheConsistencyResult { IsConsistent = true, TotalActiveHosts = 1, ConvergedHosts = 1 }); + throw new NotSupportedException("Cache sync is not supported for file-based storage."); } public async Task GetMaxLastUpdatedAsync(CancellationToken cancellationToken) diff --git a/src/Microsoft.Health.Fhir.Core/Features/Search/Registry/ISearchParameterStatusDataStore.cs b/src/Microsoft.Health.Fhir.Core/Features/Search/Registry/ISearchParameterStatusDataStore.cs index dbe92be6d2..bb808b15b5 100644 --- a/src/Microsoft.Health.Fhir.Core/Features/Search/Registry/ISearchParameterStatusDataStore.cs +++ b/src/Microsoft.Health.Fhir.Core/Features/Search/Registry/ISearchParameterStatusDataStore.cs @@ -12,6 +12,8 @@ namespace Microsoft.Health.Fhir.Core.Features.Search.Registry { public interface ISearchParameterStatusDataStore { + string SearchParamCacheUpdateProcessName { get; } + Task> GetSearchParameterStatuses(CancellationToken cancellationToken, DateTimeOffset? startLastUpdated = null); Task UpsertStatuses(IReadOnlyCollection statuses, CancellationToken cancellationToken); @@ -21,14 +23,12 @@ public interface ISearchParameterStatusDataStore Task TryLogEvent(string process, string status, string text, DateTime? startDate, CancellationToken cancellationToken); /// - /// Checks whether all active instances have converged their search parameter caches - /// to at least the specified target timestamp. + /// Checks whether all active instances have updated their search parameter caches /// - /// The target SearchParamLastUpdated timestamp to check for. - /// Only cache refresh sync records on or after this time are considered for convergence. - /// Only active-host evidence on or after this time is considered. + /// Only cache update records after this time are considered for convergence. + /// Only active hosts after this time are considered. /// Cancellation token. /// A indicating convergence status. - Task CheckCacheConsistencyAsync(string targetSearchParamLastUpdated, DateTime syncStartDate, DateTime activeHostsSince, CancellationToken cancellationToken); + Task CheckCacheConsistencyAsync(DateTime updateEventsSince, DateTime activeHostsSince, CancellationToken cancellationToken); } } diff --git a/src/Microsoft.Health.Fhir.Core/Features/Search/Registry/ISearchParameterStatusManager.cs b/src/Microsoft.Health.Fhir.Core/Features/Search/Registry/ISearchParameterStatusManager.cs index cb7ced63e0..a965f39450 100644 --- a/src/Microsoft.Health.Fhir.Core/Features/Search/Registry/ISearchParameterStatusManager.cs +++ b/src/Microsoft.Health.Fhir.Core/Features/Search/Registry/ISearchParameterStatusManager.cs @@ -25,6 +25,6 @@ public interface ISearchParameterStatusManager Task UpdateSearchParameterStatusAsync(IReadOnlyCollection searchParameterUris, SearchParameterStatus status, CancellationToken cancellationToken, bool ignoreSearchParameterNotSupportedException = false); - Task CheckCacheConsistencyAsync(string targetSearchParamLastUpdated, DateTime syncStartDate, DateTime activeHostsSince, CancellationToken cancellationToken); + Task CheckCacheConsistencyAsync(DateTime updateEventsSince, DateTime activeHostsSince, CancellationToken cancellationToken); } } diff --git a/src/Microsoft.Health.Fhir.Core/Features/Search/Registry/SearchParameterStatusManager.cs b/src/Microsoft.Health.Fhir.Core/Features/Search/Registry/SearchParameterStatusManager.cs index 51c1cf1255..777c482195 100644 --- a/src/Microsoft.Health.Fhir.Core/Features/Search/Registry/SearchParameterStatusManager.cs +++ b/src/Microsoft.Health.Fhir.Core/Features/Search/Registry/SearchParameterStatusManager.cs @@ -49,6 +49,8 @@ public SearchParameterStatusManager( _logger = logger; } + public string SearchParamCacheUpdateProcessName => _searchParameterStatusDataStore.SearchParamCacheUpdateProcessName; + internal async Task EnsureInitializedAsync(CancellationToken cancellationToken) { var updated = new List(); @@ -265,9 +267,14 @@ private static TempStatus EvaluateSearchParamStatus(ResourceSearchParameterStatu return tempStatus; } - public async Task CheckCacheConsistencyAsync(string targetSearchParamLastUpdated, DateTime syncStartDate, DateTime activeHostsSince, CancellationToken cancellationToken) + public async Task CheckCacheConsistencyAsync(DateTime updateEventsSince, DateTime activeHostsSince, CancellationToken cancellationToken) + { + return await _searchParameterStatusDataStore.CheckCacheConsistencyAsync(updateEventsSince, activeHostsSince, cancellationToken); + } + + public async Task TryLogEvent(string process, string status, string text, DateTime? startDate, CancellationToken cancellationToken) { - return await _searchParameterStatusDataStore.CheckCacheConsistencyAsync(targetSearchParamLastUpdated, syncStartDate, activeHostsSince, cancellationToken); + await _searchParameterStatusDataStore.TryLogEvent(process, status, text, startDate, cancellationToken); } private struct TempStatus diff --git a/src/Microsoft.Health.Fhir.Core/Features/Security/Constants.cs b/src/Microsoft.Health.Fhir.Core/Features/Security/Constants.cs index 4399ec22a7..2e916317c0 100644 --- a/src/Microsoft.Health.Fhir.Core/Features/Security/Constants.cs +++ b/src/Microsoft.Health.Fhir.Core/Features/Security/Constants.cs @@ -3,6 +3,7 @@ // Licensed under the MIT License (MIT). See LICENSE in the repo root for license information. // ------------------------------------------------------------------------------------------------- +using System; using Hl7.Fhir.Model; namespace Microsoft.Health.Fhir.Core.Features.Security @@ -54,10 +55,7 @@ public static class Constants "authorize-post", }; - public static readonly string[] SmartCapabilityThirdPartyContexts = new[] - { - "context-ehr-encounter", - }; + public static readonly string[] SmartCapabilityThirdPartyContexts = Array.Empty(); public static ref readonly Coding RestfulSecurityServiceOAuth => ref RestfulSecurityServiceOAuthCodeableConcept; diff --git a/src/Microsoft.Health.Fhir.Core/Features/Security/IOidcDiscoveryService.cs b/src/Microsoft.Health.Fhir.Core/Features/Security/IOidcDiscoveryService.cs index cab66809e3..8a91fb07a7 100644 --- a/src/Microsoft.Health.Fhir.Core/Features/Security/IOidcDiscoveryService.cs +++ b/src/Microsoft.Health.Fhir.Core/Features/Security/IOidcDiscoveryService.cs @@ -16,13 +16,13 @@ namespace Microsoft.Health.Fhir.Core.Features.Security public interface IOidcDiscoveryService { /// - /// Resolves the authorization and token endpoints for the given OIDC authority. + /// Resolves the authorization and token endpoints, issuer, and JWKS URI for the given OIDC authority. /// Fetches the OpenID Connect discovery document at {authority}/.well-known/openid-configuration. /// If discovery fails, falls back to Entra ID URL pattern ({authority}/oauth2/v2.0/authorize and /token). /// /// The OIDC authority URL. /// Cancellation token. - /// A tuple containing the authorization and token endpoint URIs. - Task<(Uri AuthorizationEndpoint, Uri TokenEndpoint)> ResolveEndpointsAsync(string authority, CancellationToken cancellationToken = default); + /// A tuple containing the authorization endpoint, token endpoint, issuer, and JWKS URI. + Task<(Uri AuthorizationEndpoint, Uri TokenEndpoint, string Issuer, string JwksUri)> ResolveEndpointsAsync(string authority, CancellationToken cancellationToken = default); } } diff --git a/src/Microsoft.Health.Fhir.Core/Features/Security/OidcDiscoveryService.cs b/src/Microsoft.Health.Fhir.Core/Features/Security/OidcDiscoveryService.cs index cdee749ffa..959018da0d 100644 --- a/src/Microsoft.Health.Fhir.Core/Features/Security/OidcDiscoveryService.cs +++ b/src/Microsoft.Health.Fhir.Core/Features/Security/OidcDiscoveryService.cs @@ -31,7 +31,7 @@ public class OidcDiscoveryService : IOidcDiscoveryService private readonly IHttpClientFactory _httpClientFactory; private readonly ILogger _logger; private readonly TimeSpan _cacheDuration; - private readonly ConcurrentDictionary _cache = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary _cache = new(StringComparer.Ordinal); private readonly AsyncRetryPolicy _retryPolicy; @@ -70,7 +70,7 @@ internal OidcDiscoveryService(IHttpClientFactory httpClientFactory, ILogger ResolveEndpointsAsync(string authority, CancellationToken cancellationToken = default) + public async Task<(Uri AuthorizationEndpoint, Uri TokenEndpoint, string Issuer, string JwksUri)> ResolveEndpointsAsync(string authority, CancellationToken cancellationToken = default) { EnsureArg.IsNotNullOrWhiteSpace(authority, nameof(authority)); @@ -80,18 +80,18 @@ internal OidcDiscoveryService(IHttpClientFactory httpClientFactory, ILogger DiscoverEndpointsAsync(string authority, CancellationToken cancellationToken) + private async Task<(Uri AuthorizationEndpoint, Uri TokenEndpoint, string Issuer, string JwksUri)> DiscoverEndpointsAsync(string authority, CancellationToken cancellationToken) { try { @@ -108,10 +108,12 @@ internal OidcDiscoveryService(IHttpClientFactory httpClientFactory, ILogger diff --git a/src/Microsoft.Health.Fhir.Core/Messages/Get/GetSmartConfigurationResponse.cs b/src/Microsoft.Health.Fhir.Core/Messages/Get/GetSmartConfigurationResponse.cs index 1a5c0b8b61..dd8bd77af0 100644 --- a/src/Microsoft.Health.Fhir.Core/Messages/Get/GetSmartConfigurationResponse.cs +++ b/src/Microsoft.Health.Fhir.Core/Messages/Get/GetSmartConfigurationResponse.cs @@ -33,7 +33,9 @@ public GetSmartConfigurationResponse( ICollection responseTypesSupported = null, string introspectionEndpoint = null, string managementEndpoint = null, - string revocationEndpoint = null) + string revocationEndpoint = null, + string issuer = null, + string jwksUri = null) { EnsureArg.IsNotNull(authorizationEndpoint, nameof(authorizationEndpoint)); EnsureArg.IsNotNull(tokenEndpoint, nameof(tokenEndpoint)); @@ -50,6 +52,8 @@ public GetSmartConfigurationResponse( IntrospectionEndpoint = introspectionEndpoint; ManagementEndpoint = managementEndpoint; RevocationEndpoint = revocationEndpoint; + Issuer = issuer; + JwksUri = jwksUri; } public Uri AuthorizationEndpoint { get; } @@ -73,5 +77,11 @@ public GetSmartConfigurationResponse( public string ManagementEndpoint { get; } public string RevocationEndpoint { get; } + + public string Issuer { get; } + +#pragma warning disable CA1056 // URI-like properties should not be strings + public string JwksUri { get; } +#pragma warning restore CA1056 // URI-like properties should not be strings } } diff --git a/src/Microsoft.Health.Fhir.CosmosDb/Features/Storage/Queues/CosmosQueueClient.cs b/src/Microsoft.Health.Fhir.CosmosDb/Features/Storage/Queues/CosmosQueueClient.cs index 393ce883d8..acd73ccd43 100644 --- a/src/Microsoft.Health.Fhir.CosmosDb/Features/Storage/Queues/CosmosQueueClient.cs +++ b/src/Microsoft.Health.Fhir.CosmosDb/Features/Storage/Queues/CosmosQueueClient.cs @@ -163,7 +163,7 @@ AND ARRAY_CONTAINS([{string.Join(",", definitionHashes.Select(x => $"'{x.Key}'") return jobInfos; } - public async Task EnqueueWithStatusAsync(byte queueType, long groupId, string definition, JobStatus jobStatus, string result, DateTime? startDate, CancellationToken cancellationToken) + public async Task EnqueueWithStatusAsync(byte queueType, long? groupId, string definition, JobStatus jobStatus, string result, DateTime? startDate, CancellationToken cancellationToken) { // This is a special case, we are adding this function to support Handle Bulk data access 2.0 // Only limited changes are made in order to get Cosmos working 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 ef7a3e7667..7ff4b32cbb 100644 --- a/src/Microsoft.Health.Fhir.CosmosDb/Features/Storage/Registry/CosmosDbSearchParameterStatusDataStore.cs +++ b/src/Microsoft.Health.Fhir.CosmosDb/Features/Storage/Registry/CosmosDbSearchParameterStatusDataStore.cs @@ -37,6 +37,8 @@ public CosmosDbSearchParameterStatusDataStore( _queryFactory = queryFactory; } + public string SearchParamCacheUpdateProcessName => null; + public async Task TryLogEvent(string process, string status, string text, DateTime? startDate, CancellationToken cancellationToken) { await Task.CompletedTask; // noop @@ -114,11 +116,9 @@ public void SyncStatuses(IReadOnlyCollection stat // Do nothing. This is only required for SQL. } - public Task CheckCacheConsistencyAsync(string targetSearchParamLastUpdated, DateTime syncStartDate, DateTime activeHostsSince, CancellationToken cancellationToken) + public Task CheckCacheConsistencyAsync(DateTime updateEventsSince, DateTime activeHostsSince, CancellationToken cancellationToken) { - // Cosmos DB does not use EventLog-based convergence tracking. - // Return immediately as consistent since each instance refreshes from the same Cosmos container. - return Task.FromResult(new CacheConsistencyResult { IsConsistent = true, TotalActiveHosts = 1, ConvergedHosts = 1 }); + throw new NotSupportedException("Cache sync is not supported for Cosmos DB storage."); } public void Dispose() diff --git a/src/Microsoft.Health.Fhir.Shared.Api.UnitTests/Features/Resources/Bundle/BundleHandlerEdgeCaseTests.cs b/src/Microsoft.Health.Fhir.Shared.Api.UnitTests/Features/Resources/Bundle/BundleHandlerEdgeCaseTests.cs index b16d90ffe3..14928492bf 100644 --- a/src/Microsoft.Health.Fhir.Shared.Api.UnitTests/Features/Resources/Bundle/BundleHandlerEdgeCaseTests.cs +++ b/src/Microsoft.Health.Fhir.Shared.Api.UnitTests/Features/Resources/Bundle/BundleHandlerEdgeCaseTests.cs @@ -31,6 +31,7 @@ using Microsoft.Health.Fhir.Core.Features.Search.Registry; using Microsoft.Health.Fhir.Core.Features.Security.Authorization; using Microsoft.Health.Fhir.Core.Features.Validation; +using Microsoft.Health.Fhir.Core.Models; using Microsoft.Health.Fhir.Core.UnitTests.Features.Context; using Microsoft.Health.Fhir.Tests.Common; using Microsoft.Health.Test.Utilities; @@ -165,7 +166,8 @@ private IFhirRequestContext CreateRequestContextForBundleHandlerProcessing(Bundl router, profilesResolver, Substitute.For(), - NullLogger.Instance); + NullLogger.Instance, + Substitute.For()); return fhirRequestContextAccessor.RequestContext; } diff --git a/src/Microsoft.Health.Fhir.Shared.Api.UnitTests/Features/Resources/Bundle/BundleHandlerTests.cs b/src/Microsoft.Health.Fhir.Shared.Api.UnitTests/Features/Resources/Bundle/BundleHandlerTests.cs index f6c1f42c20..ef75b0d6ef 100644 --- a/src/Microsoft.Health.Fhir.Shared.Api.UnitTests/Features/Resources/Bundle/BundleHandlerTests.cs +++ b/src/Microsoft.Health.Fhir.Shared.Api.UnitTests/Features/Resources/Bundle/BundleHandlerTests.cs @@ -141,7 +141,8 @@ public BundleHandlerTests() _router, _profilesResolver, Substitute.For(), - NullLogger.Instance); + NullLogger.Instance, + Substitute.For()); } [Fact] diff --git a/src/Microsoft.Health.Fhir.Shared.Api.UnitTests/Features/Security/SecurityProviderTests.cs b/src/Microsoft.Health.Fhir.Shared.Api.UnitTests/Features/Security/SecurityProviderTests.cs index e6b94759d5..483e5de5db 100644 --- a/src/Microsoft.Health.Fhir.Shared.Api.UnitTests/Features/Security/SecurityProviderTests.cs +++ b/src/Microsoft.Health.Fhir.Shared.Api.UnitTests/Features/Security/SecurityProviderTests.cs @@ -57,7 +57,7 @@ public SecurityProviderTests() _oidcDiscoveryService = Substitute.For(); _oidcDiscoveryService.ResolveEndpointsAsync(Arg.Any(), Arg.Any()) - .Returns((new Uri(OpenIdAuthorizationEndpointUri), new Uri(OpenIdTokenEndpointUri))); + .Returns((new Uri(OpenIdAuthorizationEndpointUri), new Uri(OpenIdTokenEndpointUri), null, null)); _urlResolver = Substitute.For(); _urlResolver.ResolveRouteNameUrl( diff --git a/src/Microsoft.Health.Fhir.Shared.Api/Features/Resources/Bundle/BundleHandler.cs b/src/Microsoft.Health.Fhir.Shared.Api/Features/Resources/Bundle/BundleHandler.cs index 06745778a3..3159485db9 100644 --- a/src/Microsoft.Health.Fhir.Shared.Api/Features/Resources/Bundle/BundleHandler.cs +++ b/src/Microsoft.Health.Fhir.Shared.Api/Features/Resources/Bundle/BundleHandler.cs @@ -10,7 +10,6 @@ using System.IO; using System.Linq; using System.Net; -using System.Reflection.Metadata; using System.Threading; using System.Threading.Tasks; using System.Transactions; @@ -43,6 +42,7 @@ using Microsoft.Health.Fhir.Core.Extensions; using Microsoft.Health.Fhir.Core.Features; using Microsoft.Health.Fhir.Core.Features.Context; +using Microsoft.Health.Fhir.Core.Features.Definition; using Microsoft.Health.Fhir.Core.Features.Persistence; using Microsoft.Health.Fhir.Core.Features.Persistence.Orchestration; using Microsoft.Health.Fhir.Core.Features.Resources; @@ -93,6 +93,7 @@ public partial class BundleHandler : IRequestHandler logger) + ILogger logger, + IModelInfoProvider modelInfoProvider) : this() { EnsureArg.IsNotNull(httpContextAccessor, nameof(httpContextAccessor)); @@ -165,6 +167,7 @@ public BundleHandler( _profilesResolver = EnsureArg.IsNotNull(profilesResolver, nameof(profilesResolver)); _searchParameterStatusDataStore = EnsureArg.IsNotNull(searchParameterStatusDataStore, nameof(searchParameterStatusDataStore)); _logger = EnsureArg.IsNotNull(logger, nameof(logger)); + _modelInfoProvider = EnsureArg.IsNotNull(modelInfoProvider, nameof(modelInfoProvider)); // Not all versions support the same enum values, so do the dictionary creation in the version specific partial. _requests = _verbExecutionOrder.ToDictionary(verb => verb, _ => new List()); @@ -221,6 +224,11 @@ public async Task Handle(BundleRequest request, CancellationToke Type = BundleType.BatchResponse, }; + if (bundleProcessingLogic == BundleProcessingLogic.Parallel) + { + CheckConflictsAcrossInputSearchParams(bundleResource); + } + await ProcessAllResourcesInABundleAsRequestsAsync(responseBundle, bundleProcessingLogic, cancellationToken); var response = new BundleResponse( @@ -253,6 +261,8 @@ public async Task Handle(BundleRequest request, CancellationToke } } + CheckConflictsAcrossInputSearchParams(bundleResource); + var responseBundle = new Hl7.Fhir.Model.Bundle { Type = BundleType.TransactionResponse, @@ -279,6 +289,44 @@ public async Task Handle(BundleRequest request, CancellationToke } } + private void CheckConflictsAcrossInputSearchParams(Hl7.Fhir.Model.Bundle bundle) + { + var codes = new HashSet<(string Type, string Code)>(); + var urls = new HashSet(); + var dupCodes = new HashSet<(string Type, string Code)>(); + var dupUrls = new HashSet(); + foreach (var param in bundle.Entry.Select(_ => _.Resource as SearchParameter).Where(_ => _ != null)) + { + if (param.Code != null && param.Base != null) + { + var allResourceTypes = SearchParameterDefinitionManager.GetDerivedResourceTypes(_modelInfoProvider, param.Base.Where(_ => _ != null).Select(_ => _.Value.ToString()).ToList()); + foreach (var resourceType in allResourceTypes.Where(_ => !codes.Add((_, param.Code)))) + { + dupCodes.Add((resourceType, param.Code)); + } + } + + if (param.Url != null && !urls.Add(param.Url)) + { + dupUrls.Add(param.Url); + } + } + + if (dupCodes.Count > 0 || dupUrls.Count > 0) + { + if (dupCodes.Count == 0) + { + throw new RequestNotValidException(string.Format(Api.Resources.DuplicateSearchParamUrlsInBundle, string.Join(", ", dupUrls))); + } + else if (dupUrls.Count == 0) + { + throw new RequestNotValidException(string.Format(Api.Resources.DuplicateSearchParamCodesInBundle, string.Join(", ", dupCodes))); + } + + throw new RequestNotValidException(string.Format(Api.Resources.DuplicateSearchParamCodesAndUrlsInBundle, string.Join(", ", dupCodes), string.Join(", ", dupUrls))); + } + } + private static bool SetRequestContextWithOptimizedQuerying(HttpContext outerHttpContext, IFhirRequestContext fhirRequestContext, ILogger logger) { try diff --git a/src/Microsoft.Health.Fhir.Shared.Api/Features/Security/SecurityProvider.cs b/src/Microsoft.Health.Fhir.Shared.Api/Features/Security/SecurityProvider.cs index 53d8c2bf84..77d1908404 100644 --- a/src/Microsoft.Health.Fhir.Shared.Api/Features/Security/SecurityProvider.cs +++ b/src/Microsoft.Health.Fhir.Shared.Api/Features/Security/SecurityProvider.cs @@ -69,7 +69,7 @@ public async Task BuildAsync(ICapabilityStatementBuilder builder, CancellationTo } else { - var (authorizationEndpoint, tokenEndpoint) = await _oidcDiscoveryService.ResolveEndpointsAsync( + var (authorizationEndpoint, tokenEndpoint, _, _) = await _oidcDiscoveryService.ResolveEndpointsAsync( _securityConfiguration.Authentication.Authority, cancellationToken); diff --git a/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Conformance/GetSmartConfigurationHandlerTests.cs b/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Conformance/GetSmartConfigurationHandlerTests.cs index 9c9f947f85..133c830992 100644 --- a/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Conformance/GetSmartConfigurationHandlerTests.cs +++ b/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Conformance/GetSmartConfigurationHandlerTests.cs @@ -49,7 +49,7 @@ private static GetSmartConfigurationHandler CreateHandler( .Returns(callInfo => { string authority = callInfo.ArgAt(0).TrimEnd('/'); - return (new Uri(authority + "/oauth2/v2.0/authorize"), new Uri(authority + "/oauth2/v2.0/token")); + return (new Uri(authority + "/oauth2/v2.0/authorize"), new Uri(authority + "/oauth2/v2.0/token"), authority + "/v2.0", authority + "/discovery/v2.0/keys"); }); } @@ -92,6 +92,8 @@ public async Task GivenASmartConfigurationHandler_WhenSecurityConfigurationEnabl Assert.Equal(baseEndpoint + "/oauth2/v2.0/authorize", response.AuthorizationEndpoint.ToString()); Assert.Equal(baseEndpoint + "/oauth2/v2.0/token", response.TokenEndpoint.ToString()); Assert.Equal(ExpectedBaseCapabilities, response.Capabilities); + Assert.Equal(baseEndpoint + "/v2.0", response.Issuer); + Assert.Equal(baseEndpoint + "/discovery/v2.0/keys", response.JwksUri); // Verify SMART v2 scopes are included Assert.NotNull(response.ScopesSupported); @@ -99,6 +101,9 @@ public async Task GivenASmartConfigurationHandler_WhenSecurityConfigurationEnabl Assert.NotNull(response.GrantTypesSupported); Assert.NotNull(response.TokenEndpointAuthMethodsSupported); Assert.NotNull(response.ResponseTypesSupported); + + // Verify auto-constructed introspection endpoint + Assert.Equal("https://fhir.example.com/connect/introspect", response.IntrospectionEndpoint); } [Fact] @@ -115,7 +120,7 @@ public async Task GivenASmartConfigurationHandler_WhenBaseEndpointIsInvalid_Then // Make the discovery service throw UriFormatException for invalid authorities var oidcService = Substitute.For(); oidcService.ResolveEndpointsAsync(Arg.Any(), Arg.Any()) - .Returns<(Uri, Uri)>(x => throw new UriFormatException("Invalid URI")); + .Returns<(Uri, Uri, string, string)>(x => throw new UriFormatException("Invalid URI")); var handler = CreateHandler(securityConfiguration, oidcDiscoveryService: oidcService); @@ -154,7 +159,16 @@ public async Task GivenASmartConfigurationHandler_WhenOtherEndpointsAreSpecified Assert.Equal(baseEndpoint + "/oauth2/v2.0/token", response.TokenEndpoint.ToString()); // Verify SMART v2 endpoints - Assert.Equal(introspectionEndpoint, response.IntrospectionEndpoint); + if (introspectionEndpoint is not null) + { + Assert.Equal(introspectionEndpoint, response.IntrospectionEndpoint); + } + else + { + // When not configured, defaults to built-in introspection endpoint + Assert.Equal("https://fhir.example.com/connect/introspect", response.IntrospectionEndpoint); + } + Assert.Equal(managementEndpoint, response.ManagementEndpoint); Assert.Equal(revocationEndpoint, response.RevocationEndpoint); } @@ -178,6 +192,8 @@ public async Task GivenASmartConfigurationHandler_WhenAadSmartOnFhirProxyEnabled Assert.Equal("https://fhir.example.com/AadSmartOnFhirProxy/authorize", response.AuthorizationEndpoint.ToString()); Assert.Equal("https://fhir.example.com/AadSmartOnFhirProxy/token", response.TokenEndpoint.ToString()); Assert.Equal(ExpectedBaseCapabilities, response.Capabilities); + Assert.Equal(baseEndpoint + "/v2.0", response.Issuer); + Assert.Equal(baseEndpoint + "/discovery/v2.0/keys", response.JwksUri); // Verify SMART v2 scopes are included Assert.NotNull(response.ScopesSupported); diff --git a/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Operations/Reindex/ReindexOrchestratorJobTests.cs b/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Operations/Reindex/ReindexOrchestratorJobTests.cs index 15647d721f..40ae4e68d9 100644 --- a/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Operations/Reindex/ReindexOrchestratorJobTests.cs +++ b/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Operations/Reindex/ReindexOrchestratorJobTests.cs @@ -48,7 +48,7 @@ public class ReindexOrchestratorJobTests private readonly ISearchParameterStatusManager _searchParameterStatusManager; private readonly ISearchParameterDefinitionManager _searchDefinitionManager; private readonly ISearchParameterOperations _searchParameterOperations; - private IQueueClient _queueClient; + private readonly IQueueClient _queueClient; private readonly CancellationToken _cancellationToken; public ReindexOrchestratorJobTests() @@ -60,6 +60,8 @@ public ReindexOrchestratorJobTests() _searchDefinitionManager = Substitute.For(); _searchParameterStatusManager = Substitute.For(); _searchParameterOperations = Substitute.For(); + _searchParameterStatusManager.CheckCacheConsistencyAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(new CacheConsistencyResult { IsConsistent = true, ActiveHosts = 1, ConvergedHosts = 1 }); // Initialize a fresh queue client for each test _queueClient = new TestQueueClient(); @@ -76,17 +78,15 @@ private void Dispose() _cancellationTokenSource?.Dispose(); } - private ReindexOrchestratorJob CreateReindexOrchestratorJob(IFhirRuntimeConfiguration runtimeConfig = null, int waitMultiplier = 0) + private ReindexOrchestratorJob CreateReindexOrchestratorJob() { - runtimeConfig ??= new AzureHealthDataServicesRuntimeConfiguration(); + var runtimeConfig = new AzureHealthDataServicesRuntimeConfiguration(); var coreFeatureConfig = Substitute.For>(); - coreFeatureConfig.Value.Returns(new CoreFeatureConfiguration()); + coreFeatureConfig.Value.Returns(new CoreFeatureConfiguration { SearchParameterCacheRefreshIntervalSeconds = 1 }); var operationsConfig = Substitute.For>(); - var conf = new OperationsConfiguration(); - conf.Reindex.CacheRefreshWaitMultiplier = waitMultiplier; - operationsConfig.Value.Returns(conf); + operationsConfig.Value.Returns(new OperationsConfiguration { Reindex = new ReindexJobConfiguration { CacheUpdateMaxWaitMultiplier = 1 } }); return new ReindexOrchestratorJob( _queueClient, @@ -124,6 +124,7 @@ private async Task CreateReindexJobRecord( // Return the enqueued job with Running status var jobInfo = orchestratorJobs.First(); jobInfo.Status = JobStatus.Running; + jobInfo.StartDate = DateTime.UtcNow; return jobInfo; } @@ -257,6 +258,28 @@ private async Task> WaitForJobsAsync(long groupId, TimeSpan timeou return finalJobs.Where(j => j.Id != groupId).ToList(); } + [Fact] + public async Task ExecuteAsync_WhenCacheIsNotSynced_ReturnsErrorResult() + { + _searchParameterStatusManager.CheckCacheConsistencyAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(new CacheConsistencyResult { IsConsistent = false, ActiveHosts = 1, ConvergedHosts = 1 }); + + var jobInfo = await CreateReindexJobRecord(); + var orchestrator = CreateReindexOrchestratorJob(); + + // Act + var result = await orchestrator.ExecuteAsync(jobInfo, CancellationToken.None); + var jobResult = JsonConvert.DeserializeObject(result); + + // Assert + Assert.NotNull(jobResult); + Assert.NotNull(jobResult.Error); + Assert.Contains(jobResult.Error, e => e.Diagnostics.Contains("Unable to sync search parameter cache")); + + _searchParameterStatusManager.CheckCacheConsistencyAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(new CacheConsistencyResult { IsConsistent = true, ActiveHosts = 1, ConvergedHosts = 1 }); + } + [Fact] public async Task ExecuteAsync_WhenCancellationRequested_ReturnsJobCancelledResult() { @@ -264,16 +287,17 @@ public async Task ExecuteAsync_WhenCancellationRequested_ReturnsJobCancelledResu var cancellationTokenSource = new CancellationTokenSource(); cancellationTokenSource.CancelAfter(10); // Cancel after short delay - // Make WaitForRefreshCyclesAsync block until cancellation, simulating a real wait - _searchParameterOperations.WaitForRefreshCyclesAsync(Arg.Any(), Arg.Any()) - .Returns(callInfo => + // Make CheckCacheConsistencyAsync block until cancellation, simulating a real wait + _searchParameterStatusManager.CheckCacheConsistencyAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(async callInfo => { - var ct = callInfo.ArgAt(1); - return Task.Delay(Timeout.Infinite, ct); + var ct = callInfo.ArgAt(2); + await Task.Delay(Timeout.Infinite, ct); + return new CacheConsistencyResult { IsConsistent = true, ActiveHosts = 1, ConvergedHosts = 1 }; }); var jobInfo = await CreateReindexJobRecord(); - var orchestrator = CreateReindexOrchestratorJob(waitMultiplier: 1); + var orchestrator = CreateReindexOrchestratorJob(); // Act var result = await orchestrator.ExecuteAsync(jobInfo, cancellationTokenSource.Token); @@ -283,6 +307,9 @@ public async Task ExecuteAsync_WhenCancellationRequested_ReturnsJobCancelledResu Assert.NotNull(jobResult); Assert.NotNull(jobResult.Error); Assert.Contains(jobResult.Error, e => e.Diagnostics.Contains("cancelled")); + + _searchParameterStatusManager.CheckCacheConsistencyAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(new CacheConsistencyResult { IsConsistent = true, ActiveHosts = 1, ConvergedHosts = 1 }); } [Fact] @@ -708,7 +735,7 @@ public async Task EnqueueQueryProcessingJobsAsync_WithValidSearchParameters_Crea }); var jobInfo = await CreateReindexJobRecord(); - var orchestrator = CreateReindexOrchestratorJob(new AzureHealthDataServicesRuntimeConfiguration()); + var orchestrator = CreateReindexOrchestratorJob(); // Act: Fire off execute asynchronously without awaiting var executeTask = orchestrator.ExecuteAsync(jobInfo, _cancellationToken); @@ -1692,7 +1719,7 @@ public async Task CreateReindexProcessingJobsAsync_WithTargetResourceTypes_Filte }); var jobInfo = await CreateReindexJobRecord(targetResourceTypes: targetResourceTypes); - var orchestrator = CreateReindexOrchestratorJob(new AzureHealthDataServicesRuntimeConfiguration()); + var orchestrator = CreateReindexOrchestratorJob(); // Act _ = orchestrator.ExecuteAsync(jobInfo, _cancellationToken); @@ -2050,39 +2077,18 @@ public async Task CheckForCompletionAsync_WithCaseVariantUrls_MaintainsBothVaria [Fact] public async Task RefreshSearchParameterCache_WaitsForConfiguredNumberOfCacheRefreshCycles() { - // Arrange - int configuredMultiplier = 3; - - var emptyStatus = new List(); - _searchParameterStatusManager.GetAllSearchParameterStatus(Arg.Any()) - .Returns(emptyStatus); - - var jobInfo = await CreateReindexJobRecord(); - var orchestrator = CreateReindexOrchestratorJob(waitMultiplier: configuredMultiplier); - - // Act - var result = await orchestrator.ExecuteAsync(jobInfo, _cancellationToken); - - // Assert - WaitForRefreshCyclesAsync should have been called twice (Start and End) with the configured multiplier - await _searchParameterOperations.Received(2).WaitForRefreshCyclesAsync(configuredMultiplier, Arg.Any()); - } - - [Fact] - public async Task RefreshSearchParameterCache_WithZeroMultiplier_DoesNotWait() - { - // Arrange var emptyStatus = new List(); _searchParameterStatusManager.GetAllSearchParameterStatus(Arg.Any()) .Returns(emptyStatus); var jobInfo = await CreateReindexJobRecord(); - var orchestrator = CreateReindexOrchestratorJob(waitMultiplier: 0); + var orchestrator = CreateReindexOrchestratorJob(); // Act var result = await orchestrator.ExecuteAsync(jobInfo, _cancellationToken); - // Assert - WaitForRefreshCyclesAsync should have been called with 0 (returns immediately) - await _searchParameterOperations.Received(2).WaitForRefreshCyclesAsync(0, Arg.Any()); + // Assert - CheckCacheConsistencyAsync should have been called twice (Start and End) with the configured multiplier + await _searchParameterStatusManager.Received(2).CheckCacheConsistencyAsync(Arg.Any(), Arg.Any(), Arg.Any()); } } } diff --git a/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Operations/Reindex/ReindexProcessingJobTests.cs b/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Operations/Reindex/ReindexProcessingJobTests.cs index 23711e1fa2..1403134788 100644 --- a/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Operations/Reindex/ReindexProcessingJobTests.cs +++ b/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Operations/Reindex/ReindexProcessingJobTests.cs @@ -1355,7 +1355,7 @@ public async Task CheckDiscrepancies_WhenHashMismatch_ThrowsReindexJobException( async () => await _reindexProcessingJobTaskFactory().ExecuteAsync(jobInfo, _cancellationToken)); Assert.Contains($"ResourceType={expectedResourceType} SearchParameterHash: Requested={requestedHash} != Current={staleHash}", exception.Message); - await _searchParameterOperations.DidNotReceive().WaitForRefreshCyclesAsync(Arg.Any(), Arg.Any()); + ////await _searchParameterOperations.DidNotReceive().WaitForRefreshCyclesAsync(Arg.Any(), Arg.Any()); } [Fact] @@ -1400,7 +1400,7 @@ public async Task CheckDiscrepancies_WhenHashMismatch_DoesNotWaitForRefresh() async () => await _reindexProcessingJobTaskFactory().ExecuteAsync(jobInfo, _cancellationToken)); Assert.Contains($"ResourceType={expectedResourceType} SearchParameterHash: Requested={requestedHash} != Current={staleHash}", exception.Message); - await _searchParameterOperations.DidNotReceive().WaitForRefreshCyclesAsync(Arg.Any(), Arg.Any()); + ////await _searchParameterOperations.DidNotReceive().WaitForRefreshCyclesAsync(Arg.Any(), Arg.Any()); } [Fact] @@ -1460,7 +1460,7 @@ public async Task CheckDiscrepancies_WhenHashMatches_DoesNotWaitForRefresh() // Assert - Job succeeded and WaitForRefreshCyclesAsync was NOT called Assert.Equal(_mockedSearchCount, result.SucceededResourceCount); - await _searchParameterOperations.DidNotReceive().WaitForRefreshCyclesAsync(Arg.Any(), Arg.Any()); + ////await _searchParameterOperations.DidNotReceive().WaitForRefreshCyclesAsync(Arg.Any(), Arg.Any()); } [Fact] @@ -1507,7 +1507,7 @@ public async Task CheckDiscrepancies_WhenSearchParamLastUpdatedIsStale_ThrowsRei async () => await _reindexProcessingJobTaskFactory().ExecuteAsync(jobInfo, _cancellationToken)); Assert.Contains("SearchParamLastUpdated: Requested=2026-01-01 00:00:01.000 > Current=2026-01-01 00:00:00.000", exception.Message); - await _searchParameterOperations.DidNotReceive().WaitForRefreshCyclesAsync(Arg.Any(), Arg.Any()); + ////await _searchParameterOperations.DidNotReceive().WaitForRefreshCyclesAsync(Arg.Any(), Arg.Any()); } } } diff --git a/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Security/OidcDiscoveryServiceTests.cs b/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Security/OidcDiscoveryServiceTests.cs index 1b8aa0c5f2..99a34a4ced 100644 --- a/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Security/OidcDiscoveryServiceTests.cs +++ b/src/Microsoft.Health.Fhir.Shared.Core.UnitTests/Features/Security/OidcDiscoveryServiceTests.cs @@ -30,10 +30,12 @@ public async Task GivenValidAuthority_WhenDiscoverySucceeds_ThenCorrectEndpoints var service = CreateService(new OidcDiscoveryMessageHandler(expectedAuth, expectedToken)); - var (authEndpoint, tokenEndpoint) = await service.ResolveEndpointsAsync(authority); + var (authEndpoint, tokenEndpoint, issuer, jwksUri) = await service.ResolveEndpointsAsync(authority); Assert.Equal(expectedAuth, authEndpoint.ToString()); Assert.Equal(expectedToken, tokenEndpoint.ToString()); + Assert.Equal("https://example.com", issuer); + Assert.Equal("https://example.com/.well-known/keys", jwksUri); } [Fact] @@ -43,10 +45,12 @@ public async Task GivenValidAuthority_WhenDiscoveryFails_ThenEntraIdFallbackUsed var service = CreateService(new FailingMessageHandler()); - var (authEndpoint, tokenEndpoint) = await service.ResolveEndpointsAsync(authority); + var (authEndpoint, tokenEndpoint, issuer, jwksUri) = await service.ResolveEndpointsAsync(authority); Assert.Equal(authority + "/oauth2/v2.0/authorize", authEndpoint.ToString()); Assert.Equal(authority + "/oauth2/v2.0/token", tokenEndpoint.ToString()); + Assert.Equal(authority + "/v2.0", issuer); + Assert.Equal(authority + "/discovery/v2.0/keys", jwksUri); } [Fact] @@ -56,10 +60,12 @@ public async Task GivenValidAuthority_WhenDiscoveryReturnsBadContent_ThenEntraId var service = CreateService(new EmptyDiscoveryMessageHandler()); - var (authEndpoint, tokenEndpoint) = await service.ResolveEndpointsAsync(authority); + var (authEndpoint, tokenEndpoint, issuer, jwksUri) = await service.ResolveEndpointsAsync(authority); Assert.Equal(authority + "/oauth2/v2.0/authorize", authEndpoint.ToString()); Assert.Equal(authority + "/oauth2/v2.0/token", tokenEndpoint.ToString()); + Assert.Equal(authority + "/v2.0", issuer); + Assert.Equal(authority + "/discovery/v2.0/keys", jwksUri); } [Fact] @@ -104,7 +110,7 @@ public async Task GivenAuthorityWithTrailingSlash_WhenResolved_ThenNormalizedCor var service = CreateService(new OidcDiscoveryMessageHandler(expectedAuth, expectedToken)); - var (authEndpoint, tokenEndpoint) = await service.ResolveEndpointsAsync(authority); + var (authEndpoint, tokenEndpoint, _, _) = await service.ResolveEndpointsAsync(authority); Assert.Equal(expectedAuth, authEndpoint.ToString()); Assert.Equal(expectedToken, tokenEndpoint.ToString()); @@ -176,7 +182,8 @@ protected override Task SendAsync(HttpRequestMessage reques string json = $@"{{ ""authorization_endpoint"": ""{_authorizationEndpoint}"", ""token_endpoint"": ""{_tokenEndpoint}"", - ""issuer"": ""https://example.com"" + ""issuer"": ""https://example.com"", + ""jwks_uri"": ""https://example.com/.well-known/keys"" }}"; return Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) diff --git a/src/Microsoft.Health.Fhir.Shared.Core/Extensions/FhirMediatorExtensions.cs b/src/Microsoft.Health.Fhir.Shared.Core/Extensions/FhirMediatorExtensions.cs index b8cb1e98e2..cb88d0f76a 100644 --- a/src/Microsoft.Health.Fhir.Shared.Core/Extensions/FhirMediatorExtensions.cs +++ b/src/Microsoft.Health.Fhir.Shared.Core/Extensions/FhirMediatorExtensions.cs @@ -184,7 +184,9 @@ public static async Task GetSmartConfigurationAsync(th response.ResponseTypesSupported, response.IntrospectionEndpoint, response.ManagementEndpoint, - response.RevocationEndpoint); + response.RevocationEndpoint, + response.Issuer, + response.JwksUri); } public static async Task GetOperationVersionsAsync(this IMediator mediator, CancellationToken cancellationToken = default) diff --git a/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Migrations/109.diff.sql b/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Migrations/109.diff.sql index b734c39720..47cd2bc071 100644 --- a/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Migrations/109.diff.sql +++ b/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Migrations/109.diff.sql @@ -1,30 +1,34 @@ IF object_id('UpsertSearchParamsWithOptimisticConcurrency') IS NOT NULL DROP PROCEDURE UpsertSearchParamsWithOptimisticConcurrency +IF object_id('AcquireReindexJobs') IS NOT NULL DROP PROCEDURE AcquireReindexJobs +IF object_id('UpdateReindexJob') IS NOT NULL DROP PROCEDURE UpdateReindexJob +IF object_id('GetSearchParamMaxLastUpdated') IS NOT NULL DROP PROCEDURE GetSearchParamMaxLastUpdated GO -ALTER PROCEDURE dbo.MergeSearchParams @SearchParams dbo.SearchParamList READONLY - ,@IsResourceChangeCaptureEnabled bit = 0 - ,@TransactionId bigint = NULL - ,@Resources dbo.ResourceList READONLY - ,@ResourceWriteClaims dbo.ResourceWriteClaimList READONLY - ,@ReferenceSearchParams dbo.ReferenceSearchParamList READONLY - ,@TokenSearchParams dbo.TokenSearchParamList READONLY - ,@TokenTexts dbo.TokenTextList READONLY - ,@StringSearchParams dbo.StringSearchParamList READONLY - ,@UriSearchParams dbo.UriSearchParamList READONLY - ,@NumberSearchParams dbo.NumberSearchParamList READONLY - ,@QuantitySearchParams dbo.QuantitySearchParamList READONLY - ,@DateTimeSearchParms dbo.DateTimeSearchParamList READONLY - ,@ReferenceTokenCompositeSearchParams dbo.ReferenceTokenCompositeSearchParamList READONLY - ,@TokenTokenCompositeSearchParams dbo.TokenTokenCompositeSearchParamList READONLY - ,@TokenDateTimeCompositeSearchParams dbo.TokenDateTimeCompositeSearchParamList READONLY - ,@TokenQuantityCompositeSearchParams dbo.TokenQuantityCompositeSearchParamList READONLY - ,@TokenStringCompositeSearchParams dbo.TokenStringCompositeSearchParamList READONLY - ,@TokenNumberNumberCompositeSearchParams dbo.TokenNumberNumberCompositeSearchParamList READONLY +CREATE OR ALTER PROCEDURE dbo.MergeResourcesAndSearchParams + @SearchParams dbo.SearchParamList READONLY + ,@IsResourceChangeCaptureEnabled bit = 0 + ,@TransactionId bigint = NULL + ,@Resources dbo.ResourceList READONLY + ,@ResourceWriteClaims dbo.ResourceWriteClaimList READONLY + ,@ReferenceSearchParams dbo.ReferenceSearchParamList READONLY + ,@TokenSearchParams dbo.TokenSearchParamList READONLY + ,@TokenTexts dbo.TokenTextList READONLY + ,@StringSearchParams dbo.StringSearchParamList READONLY + ,@UriSearchParams dbo.UriSearchParamList READONLY + ,@NumberSearchParams dbo.NumberSearchParamList READONLY + ,@QuantitySearchParams dbo.QuantitySearchParamList READONLY + ,@DateTimeSearchParms dbo.DateTimeSearchParamList READONLY + ,@ReferenceTokenCompositeSearchParams dbo.ReferenceTokenCompositeSearchParamList READONLY + ,@TokenTokenCompositeSearchParams dbo.TokenTokenCompositeSearchParamList READONLY + ,@TokenDateTimeCompositeSearchParams dbo.TokenDateTimeCompositeSearchParamList READONLY + ,@TokenQuantityCompositeSearchParams dbo.TokenQuantityCompositeSearchParamList READONLY + ,@TokenStringCompositeSearchParams dbo.TokenStringCompositeSearchParamList READONLY + ,@TokenNumberNumberCompositeSearchParams dbo.TokenNumberNumberCompositeSearchParamList READONLY AS set nocount on DECLARE @SP varchar(100) = object_name(@@procid) ,@Mode varchar(200) = 'Cnt='+convert(varchar,(SELECT count(*) FROM @SearchParams)) ,@st datetime = getUTCdate() - ,@LastUpdated datetimeoffset(7) = sysdatetimeoffset() + ,@LastUpdated datetimeoffset(7) = convert(datetimeoffset(7), sysUTCdatetime()) ,@msg varchar(4000) ,@Rows int ,@AffectedRows int = 0 @@ -36,13 +40,11 @@ INSERT INTO @SearchParamsCopy SELECT * FROM @SearchParams WHILE EXISTS (SELECT * FROM @SearchParamsCopy) BEGIN SELECT TOP 1 @Uri = Uri, @Status = Status FROM @SearchParamsCopy - SET @msg = 'Uri='+@Uri+' Status='+@Status + SET @msg = 'Status='+@Status+' Uri='+@Uri EXECUTE dbo.LogEvent @Process=@SP,@Mode=@Mode,@Status='Start',@Text=@msg DELETE FROM @SearchParamsCopy WHERE Uri = @Uri END -DECLARE @SummaryOfChanges TABLE (Uri varchar(128) COLLATE Latin1_General_100_CS_AS NOT NULL, Operation varchar(20) NOT NULL) - BEGIN TRY SET TRANSACTION ISOLATION LEVEL SERIALIZABLE @@ -63,27 +65,27 @@ BEGIN TRY IF EXISTS (SELECT * FROM @Resources) BEGIN EXECUTE dbo.MergeResources - @AffectedRows = @AffectedRows OUTPUT - ,@RaiseExceptionOnConflict = 1 - ,@IsResourceChangeCaptureEnabled = @IsResourceChangeCaptureEnabled - ,@TransactionId = @TransactionId - ,@SingleTransaction = 1 - ,@Resources = @Resources - ,@ResourceWriteClaims = @ResourceWriteClaims - ,@ReferenceSearchParams = @ReferenceSearchParams - ,@TokenSearchParams = @TokenSearchParams - ,@TokenTexts = @TokenTexts - ,@StringSearchParams = @StringSearchParams - ,@UriSearchParams = @UriSearchParams - ,@NumberSearchParams = @NumberSearchParams - ,@QuantitySearchParams = @QuantitySearchParams - ,@DateTimeSearchParms = @DateTimeSearchParms - ,@ReferenceTokenCompositeSearchParams = @ReferenceTokenCompositeSearchParams - ,@TokenTokenCompositeSearchParams = @TokenTokenCompositeSearchParams - ,@TokenDateTimeCompositeSearchParams = @TokenDateTimeCompositeSearchParams - ,@TokenQuantityCompositeSearchParams = @TokenQuantityCompositeSearchParams - ,@TokenStringCompositeSearchParams = @TokenStringCompositeSearchParams - ,@TokenNumberNumberCompositeSearchParams = @TokenNumberNumberCompositeSearchParams; + @AffectedRows = @AffectedRows OUTPUT + ,@RaiseExceptionOnConflict = 1 + ,@IsResourceChangeCaptureEnabled = @IsResourceChangeCaptureEnabled + ,@TransactionId = @TransactionId + ,@SingleTransaction = 1 + ,@Resources = @Resources + ,@ResourceWriteClaims = @ResourceWriteClaims + ,@ReferenceSearchParams = @ReferenceSearchParams + ,@TokenSearchParams = @TokenSearchParams + ,@TokenTexts = @TokenTexts + ,@StringSearchParams = @StringSearchParams + ,@UriSearchParams = @UriSearchParams + ,@NumberSearchParams = @NumberSearchParams + ,@QuantitySearchParams = @QuantitySearchParams + ,@DateTimeSearchParms = @DateTimeSearchParms + ,@ReferenceTokenCompositeSearchParams = @ReferenceTokenCompositeSearchParams + ,@TokenTokenCompositeSearchParams = @TokenTokenCompositeSearchParams + ,@TokenDateTimeCompositeSearchParams = @TokenDateTimeCompositeSearchParams + ,@TokenQuantityCompositeSearchParams = @TokenQuantityCompositeSearchParams + ,@TokenStringCompositeSearchParams = @TokenStringCompositeSearchParams + ,@TokenNumberNumberCompositeSearchParams = @TokenNumberNumberCompositeSearchParams; SET @Rows = @Rows + @AffectedRows; END @@ -97,16 +99,9 @@ BEGIN TRY ,IsPartiallySupported = I.IsPartiallySupported WHEN NOT MATCHED BY TARGET THEN INSERT ( Uri, Status, LastUpdated, IsPartiallySupported) - VALUES (I.Uri, I.Status, @LastUpdated, I.IsPartiallySupported) - OUTPUT I.Uri, $action INTO @SummaryOfChanges; - SET @Rows = @@rowcount + VALUES (I.Uri, I.Status, @LastUpdated, I.IsPartiallySupported); - SELECT S.SearchParamId - ,S.Uri - ,S.LastUpdated - FROM dbo.SearchParam S JOIN @SummaryOfChanges C ON C.Uri = S.Uri - WHERE C.Operation = 'INSERT' - SET @msg = 'LastUpdated='+substring(convert(varchar,@LastUpdated),1,23)+' INSERT='+convert(varchar,@@rowcount) + SET @msg = 'LastUpdated='+convert(varchar(23),@LastUpdated,126)+' Merged='+convert(varchar,@@rowcount) COMMIT TRANSACTION @@ -118,48 +113,22 @@ BEGIN CATCH THROW END CATCH GO -INSERT INTO Parameters (Id,Char) SELECT 'MergeSearchParams','LogEvent' WHERE NOT EXISTS (SELECT * FROM dbo.Parameters WHERE Id = 'MergeSearchParams') -GO --- Enable event logging for DequeueJob to allow active host discovery via EventLog.HostName -INSERT INTO dbo.Parameters (Id, Char) SELECT 'DequeueJob', 'LogEvent' WHERE NOT EXISTS (SELECT * FROM dbo.Parameters WHERE Id = 'DequeueJob') -GO --- Enable event logging for cache refresh convergence tracking and diagnostics -INSERT INTO dbo.Parameters (Id, Char) SELECT 'SearchParameterCacheRefresh', 'LogEvent' WHERE NOT EXISTS (SELECT * FROM dbo.Parameters WHERE Id = 'SearchParameterCacheRefresh') +INSERT INTO Parameters (Id,Char) SELECT 'MergeResourcesAndSearchParams','LogEvent' GO -CREATE OR ALTER PROCEDURE dbo.CheckSearchParamCacheConsistency - @TargetSearchParamLastUpdated varchar(100) - ,@SyncStartDate datetime2(7) - ,@ActiveHostsSince datetime2(7) - ,@StalenessThresholdMinutes int = 10 +CREATE OR ALTER PROCEDURE dbo.GetSearchParamCacheUpdateEvents @UpdateProcess varchar(100), @UpdateEventsSince datetime, @ActiveHostsSince datetime AS set nocount on -SELECT HostName - ,CAST(NULL AS datetime2(7)) AS SyncEventDate - ,CAST(NULL AS nvarchar(3500)) AS EventText - FROM dbo.EventLog - WHERE EventDate >= @ActiveHostsSince - AND HostName IS NOT NULL - AND Process = 'DequeueJob' - -UNION ALL +DECLARE @SP varchar(100) = object_name(@@procid) + ,@Mode varchar(200) = 'Process='+@UpdateProcess+' EventsSince='+convert(varchar(23),@UpdateEventsSince,126)+' HostsSince='+convert(varchar(23),@ActiveHostsSince,126) + ,@st datetime = getUTCdate() -SELECT HostName - ,EventDate - ,EventText +SELECT EventDate + ,EventText = CASE WHEN Process = @UpdateProcess AND EventDate > @UpdateEventsSince THEN EventText ELSE NULL END + ,HostName FROM dbo.EventLog - WHERE EventDate >= @SyncStartDate - AND HostName IS NOT NULL - AND Process = 'SearchParameterCacheRefresh' - AND Status = 'End' + WHERE EventDate > @ActiveHostsSince + +EXECUTE dbo.LogEvent @Process=@SP,@Mode=@Mode,@Status='End',@Rows=@@rowcount,@Start=@st +GO +INSERT INTO dbo.Parameters (Id, Char) SELECT 'GetSearchParamCacheUpdateEvents', 'LogEvent' GO ---DECLARE @SearchParams dbo.SearchParamList ---INSERT INTO @SearchParams --- --SELECT 'http://example.org/fhir/SearchParameter/custom-mixed-base-d9e18fc8', 'Enabled', 0, '2026-01-26 17:15:43.0364438 -08:00' --- SELECT 'Test', 'Enabled', 0, '2026-01-26 17:15:43.0364438 -08:00' ---INSERT INTO @SearchParams --- SELECT 'Test2', 'Enabled', 0, '2026-01-26 17:15:43.0364438 -08:00' ---SELECT * FROM @SearchParams ---EXECUTE dbo.MergeSearchParams @SearchParams ---SELECT TOP 100 * FROM SearchParam ORDER BY SearchParamId DESC ---DELETE FROM SearchParam WHERE Uri LIKE 'Test%' ---SELECT TOP 10 * FROM EventLog ORDER BY EventDate DESC diff --git a/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Migrations/109.sql b/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Migrations/109.sql index 5d0a84638b..e708901f03 100644 --- a/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Migrations/109.sql +++ b/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Migrations/109.sql @@ -1016,49 +1016,6 @@ CREATE TABLE dbo.WatchdogLeases ( ); COMMIT -GO -CREATE PROCEDURE dbo.AcquireReindexJobs -@jobHeartbeatTimeoutThresholdInSeconds BIGINT, @maximumNumberOfConcurrentJobsAllowed INT -AS -SET NOCOUNT ON; -SET XACT_ABORT ON; -SET TRANSACTION ISOLATION LEVEL SERIALIZABLE; -BEGIN TRANSACTION; -DECLARE @expirationDateTime AS DATETIME2 (7); -SELECT @expirationDateTime = DATEADD(second, -@jobHeartbeatTimeoutThresholdInSeconds, SYSUTCDATETIME()); -DECLARE @numberOfRunningJobs AS INT; -SELECT @numberOfRunningJobs = COUNT(*) -FROM dbo.ReindexJob WITH (TABLOCKX) -WHERE Status = 'Running' - AND HeartbeatDateTime > @expirationDateTime; -DECLARE @limit AS INT = @maximumNumberOfConcurrentJobsAllowed - @numberOfRunningJobs; -IF (@limit > 0) - BEGIN - DECLARE @availableJobs TABLE ( - Id VARCHAR (64) COLLATE Latin1_General_100_CS_AS NOT NULL, - JobVersion BINARY (8) NOT NULL); - INSERT INTO @availableJobs - SELECT TOP (@limit) Id, - JobVersion - FROM dbo.ReindexJob - WHERE (Status = 'Queued' - OR (Status = 'Running' - AND HeartbeatDateTime <= @expirationDateTime)) - ORDER BY HeartbeatDateTime; - DECLARE @heartbeatDateTime AS DATETIME2 (7) = SYSUTCDATETIME(); - UPDATE dbo.ReindexJob - SET Status = 'Running', - HeartbeatDateTime = @heartbeatDateTime, - RawJobRecord = JSON_MODIFY(RawJobRecord, '$.status', 'Running') - OUTPUT inserted.RawJobRecord, inserted.JobVersion - FROM dbo.ReindexJob AS job - INNER JOIN - @availableJobs AS availableJob - ON job.Id = availableJob.Id - AND job.JobVersion = availableJob.JobVersion; - END -COMMIT TRANSACTION; - GO CREATE PROCEDURE dbo.AcquireWatchdogLease @Watchdog VARCHAR (100), @Worker VARCHAR (100), @AllowRebalance BIT=1, @ForceAcquire BIT=0, @LeasePeriodSec FLOAT, @WorkerIsRunning BIT=0, @LeaseEndTime DATETIME OUTPUT, @IsAcquired BIT OUTPUT, @CurrentLeaseHolder VARCHAR (100)=NULL OUTPUT @@ -1368,43 +1325,6 @@ WHERE Status = 'Running' OR Status = 'Queued' OR Status = 'Paused'; -GO -CREATE OR ALTER PROCEDURE dbo.CheckSearchParamCacheConsistency -@TargetSearchParamLastUpdated VARCHAR (100), @SyncStartDate DATETIME2 (7), @ActiveHostsSince DATETIME2 (7), @StalenessThresholdMinutes INT=10 -AS -SET NOCOUNT ON; -SELECT HostName, - CAST (NULL AS DATETIME2 (7)) AS SyncEventDate, - CAST (NULL AS NVARCHAR (3500)) AS EventText -FROM dbo.EventLog -WHERE EventDate >= @ActiveHostsSince - AND HostName IS NOT NULL - AND Process = 'DequeueJob' -UNION ALL -SELECT HostName, - EventDate, - EventText -FROM dbo.EventLog -WHERE EventDate >= @SyncStartDate - AND HostName IS NOT NULL - AND Process = 'SearchParameterCacheRefresh' - AND Status = 'End'; - - -GO -INSERT INTO dbo.Parameters (Id, Char) -SELECT 'DequeueJob', - 'LogEvent' -WHERE NOT EXISTS (SELECT * - FROM dbo.Parameters - WHERE Id = 'DequeueJob'); - - -GO -INSERT INTO Parameters (Id, Char) -SELECT 'SearchParameterCacheRefresh', - 'LogEvent'; - GO CREATE PROCEDURE dbo.CleanupEventLog WITH EXECUTE AS 'dbo' @@ -1995,12 +1915,6 @@ BEGIN CATCH THROW; END CATCH - -GO -INSERT INTO Parameters (Id, Char) -SELECT 'DequeueJob', - 'LogEvent'; - GO CREATE PROCEDURE dbo.DisableIndex @tableName NVARCHAR (128), @indexName NVARCHAR (128) @@ -3295,21 +3209,24 @@ BEGIN CATCH END CATCH GO -CREATE PROCEDURE dbo.GetSearchParamMaxLastUpdated +CREATE PROCEDURE dbo.GetSearchParamCacheUpdateEvents +@UpdateProcess VARCHAR (100), @UpdateEventsSince DATETIME, @ActiveHostsSince DATETIME AS SET NOCOUNT ON; -DECLARE @SP AS VARCHAR (100) = object_name(@@procid), @Mode AS VARCHAR (200) = 'SearchParam MaxLastUpdated Query', @st AS DATETIME = getUTCdate(), @MaxLastUpdated AS DATETIMEOFFSET (7); -BEGIN TRY - EXECUTE dbo.LogEvent @Process = @SP, @Mode = @Mode, @Status = 'Start', @Start = @st; - SELECT @MaxLastUpdated = MAX(LastUpdated) - FROM dbo.SearchParam; - SELECT @MaxLastUpdated AS MaxLastUpdated; - EXECUTE dbo.LogEvent @Process = @SP, @Mode = @Mode, @Status = 'End', @Start = @st, @Rows = @@ROWCOUNT; -END TRY -BEGIN CATCH - EXECUTE dbo.LogEvent @Process = @SP, @Mode = @Mode, @Status = 'Error', @Start = @st; - THROW; -END CATCH +DECLARE @SP AS VARCHAR (100) = object_name(@@procid), @Mode AS VARCHAR (200) = 'Process=' + @UpdateProcess + ' EventsSince=' + CONVERT (VARCHAR (23), @UpdateEventsSince, 126) + ' HostsSince=' + CONVERT (VARCHAR (23), @ActiveHostsSince, 126), @st AS DATETIME = getUTCdate(); +SELECT EventDate, + CASE WHEN Process = @UpdateProcess + AND EventDate > @UpdateEventsSince THEN EventText ELSE NULL END AS EventText, + HostName +FROM dbo.EventLog +WHERE EventDate > @ActiveHostsSince; +EXECUTE dbo.LogEvent @Process = @SP, @Mode = @Mode, @Status = 'End', @Rows = @@rowcount, @Start = @st; + + +GO +INSERT INTO dbo.Parameters (Id, Char) +SELECT 'GetSearchParamCacheUpdateEvents', + 'LogEvent'; GO CREATE PROCEDURE dbo.GetSearchParamStatuses @@ -4476,6 +4393,74 @@ BEGIN CATCH THROW; END CATCH +GO +CREATE PROCEDURE dbo.MergeResourcesAndSearchParams +@SearchParams dbo.SearchParamList READONLY, @IsResourceChangeCaptureEnabled BIT=0, @TransactionId BIGINT=NULL, @Resources dbo.ResourceList READONLY, @ResourceWriteClaims dbo.ResourceWriteClaimList READONLY, @ReferenceSearchParams dbo.ReferenceSearchParamList READONLY, @TokenSearchParams dbo.TokenSearchParamList READONLY, @TokenTexts dbo.TokenTextList READONLY, @StringSearchParams dbo.StringSearchParamList READONLY, @UriSearchParams dbo.UriSearchParamList READONLY, @NumberSearchParams dbo.NumberSearchParamList READONLY, @QuantitySearchParams dbo.QuantitySearchParamList READONLY, @DateTimeSearchParms dbo.DateTimeSearchParamList READONLY, @ReferenceTokenCompositeSearchParams dbo.ReferenceTokenCompositeSearchParamList READONLY, @TokenTokenCompositeSearchParams dbo.TokenTokenCompositeSearchParamList READONLY, @TokenDateTimeCompositeSearchParams dbo.TokenDateTimeCompositeSearchParamList READONLY, @TokenQuantityCompositeSearchParams dbo.TokenQuantityCompositeSearchParamList READONLY, @TokenStringCompositeSearchParams dbo.TokenStringCompositeSearchParamList READONLY, @TokenNumberNumberCompositeSearchParams dbo.TokenNumberNumberCompositeSearchParamList READONLY +AS +SET NOCOUNT ON; +DECLARE @SP AS VARCHAR (100) = object_name(@@procid), @Mode AS VARCHAR (200) = 'Cnt=' + CONVERT (VARCHAR, (SELECT count(*) + FROM @SearchParams)), @st AS DATETIME = getUTCdate(), @LastUpdated AS DATETIMEOFFSET (7) = CONVERT (DATETIMEOFFSET (7), sysUTCdatetime()), @msg AS VARCHAR (4000), @Rows AS INT, @AffectedRows AS INT = 0, @Uri AS VARCHAR (4000), @Status AS VARCHAR (20); +DECLARE @SearchParamsCopy AS dbo.SearchParamList; +INSERT INTO @SearchParamsCopy +SELECT * +FROM @SearchParams; +WHILE EXISTS (SELECT * + FROM @SearchParamsCopy) + BEGIN + SELECT TOP 1 @Uri = Uri, + @Status = Status + FROM @SearchParamsCopy; + SET @msg = 'Status=' + @Status + ' Uri=' + @Uri; + EXECUTE dbo.LogEvent @Process = @SP, @Mode = @Mode, @Status = 'Start', @Text = @msg; + DELETE @SearchParamsCopy + WHERE Uri = @Uri; + END +BEGIN TRY + SET TRANSACTION ISOLATION LEVEL SERIALIZABLE; + BEGIN TRANSACTION; + SELECT TOP 60 @msg = string_agg(S.Uri, ', ') + FROM @SearchParams AS I + INNER JOIN + dbo.SearchParam AS S + ON S.Uri = I.Uri + WHERE I.LastUpdated != S.LastUpdated; + IF @msg IS NOT NULL + BEGIN + SET @msg = concat('Optimistic concurrency conflict detected for search parameters: ', @msg); + ROLLBACK; + THROW 50001, @msg, 1; + END + IF EXISTS (SELECT * + FROM @Resources) + BEGIN + EXECUTE dbo.MergeResources @AffectedRows = @AffectedRows OUTPUT, @RaiseExceptionOnConflict = 1, @IsResourceChangeCaptureEnabled = @IsResourceChangeCaptureEnabled, @TransactionId = @TransactionId, @SingleTransaction = 1, @Resources = @Resources, @ResourceWriteClaims = @ResourceWriteClaims, @ReferenceSearchParams = @ReferenceSearchParams, @TokenSearchParams = @TokenSearchParams, @TokenTexts = @TokenTexts, @StringSearchParams = @StringSearchParams, @UriSearchParams = @UriSearchParams, @NumberSearchParams = @NumberSearchParams, @QuantitySearchParams = @QuantitySearchParams, @DateTimeSearchParms = @DateTimeSearchParms, @ReferenceTokenCompositeSearchParams = @ReferenceTokenCompositeSearchParams, @TokenTokenCompositeSearchParams = @TokenTokenCompositeSearchParams, @TokenDateTimeCompositeSearchParams = @TokenDateTimeCompositeSearchParams, @TokenQuantityCompositeSearchParams = @TokenQuantityCompositeSearchParams, @TokenStringCompositeSearchParams = @TokenStringCompositeSearchParams, @TokenNumberNumberCompositeSearchParams = @TokenNumberNumberCompositeSearchParams; + SET @Rows = @Rows + @AffectedRows; + END + MERGE INTO dbo.SearchParam + AS S + USING @SearchParams AS I ON I.Uri = S.Uri + WHEN MATCHED THEN UPDATE + SET Status = I.Status, + LastUpdated = @LastUpdated, + IsPartiallySupported = I.IsPartiallySupported + WHEN NOT MATCHED BY TARGET THEN INSERT (Uri, Status, LastUpdated, IsPartiallySupported) VALUES (I.Uri, I.Status, @LastUpdated, I.IsPartiallySupported); + SET @msg = 'LastUpdated=' + CONVERT (VARCHAR (23), @LastUpdated, 126) + ' Merged=' + CONVERT (VARCHAR, @@rowcount); + COMMIT TRANSACTION; + EXECUTE dbo.LogEvent @Process = @SP, @Mode = @Mode, @Status = 'End', @Start = @st, @Action = 'Merge', @Rows = @Rows, @Text = @msg; +END TRY +BEGIN CATCH + IF @@trancount > 0 + ROLLBACK; + EXECUTE dbo.LogEvent @Process = @SP, @Mode = @Mode, @Status = 'Error', @Start = @st; + THROW; +END CATCH + + +GO +INSERT INTO Parameters (Id, Char) +SELECT 'MergeResourcesAndSearchParams', + 'LogEvent'; + GO CREATE PROCEDURE dbo.MergeResourcesBeginTransaction @Count INT, @TransactionId BIGINT OUTPUT, @SequenceRangeFirstValue INT=NULL OUTPUT, @HeartbeatDate DATETIME=NULL, @EnableThrottling BIT=0 @@ -4679,11 +4664,11 @@ END CATCH GO CREATE PROCEDURE dbo.MergeSearchParams -@SearchParams dbo.SearchParamList READONLY, @IsResourceChangeCaptureEnabled BIT=0, @TransactionId BIGINT=NULL, @Resources dbo.ResourceList READONLY, @ResourceWriteClaims dbo.ResourceWriteClaimList READONLY, @ReferenceSearchParams dbo.ReferenceSearchParamList READONLY, @TokenSearchParams dbo.TokenSearchParamList READONLY, @TokenTexts dbo.TokenTextList READONLY, @StringSearchParams dbo.StringSearchParamList READONLY, @UriSearchParams dbo.UriSearchParamList READONLY, @NumberSearchParams dbo.NumberSearchParamList READONLY, @QuantitySearchParams dbo.QuantitySearchParamList READONLY, @DateTimeSearchParms dbo.DateTimeSearchParamList READONLY, @ReferenceTokenCompositeSearchParams dbo.ReferenceTokenCompositeSearchParamList READONLY, @TokenTokenCompositeSearchParams dbo.TokenTokenCompositeSearchParamList READONLY, @TokenDateTimeCompositeSearchParams dbo.TokenDateTimeCompositeSearchParamList READONLY, @TokenQuantityCompositeSearchParams dbo.TokenQuantityCompositeSearchParamList READONLY, @TokenStringCompositeSearchParams dbo.TokenStringCompositeSearchParamList READONLY, @TokenNumberNumberCompositeSearchParams dbo.TokenNumberNumberCompositeSearchParamList READONLY +@SearchParams dbo.SearchParamList READONLY AS SET NOCOUNT ON; DECLARE @SP AS VARCHAR (100) = object_name(@@procid), @Mode AS VARCHAR (200) = 'Cnt=' + CONVERT (VARCHAR, (SELECT count(*) - FROM @SearchParams)), @st AS DATETIME = getUTCdate(), @LastUpdated AS DATETIMEOFFSET (7) = sysdatetimeoffset(), @msg AS VARCHAR (4000), @Rows AS INT, @AffectedRows AS INT = 0, @Uri AS VARCHAR (4000), @Status AS VARCHAR (20); + FROM @SearchParams)), @st AS DATETIME = getUTCdate(), @LastUpdated AS DATETIMEOFFSET (7) = sysdatetimeoffset(), @msg AS VARCHAR (4000), @Rows AS INT, @Uri AS VARCHAR (4000), @Status AS VARCHAR (20); DECLARE @SearchParamsCopy AS dbo.SearchParamList; INSERT INTO @SearchParamsCopy SELECT * @@ -4717,12 +4702,6 @@ BEGIN TRY ROLLBACK; THROW 50001, @msg, 1; END - IF EXISTS (SELECT * - FROM @Resources) - BEGIN - EXECUTE dbo.MergeResources @AffectedRows = @AffectedRows OUTPUT, @RaiseExceptionOnConflict = 1, @IsResourceChangeCaptureEnabled = @IsResourceChangeCaptureEnabled, @TransactionId = @TransactionId, @SingleTransaction = 1, @Resources = @Resources, @ResourceWriteClaims = @ResourceWriteClaims, @ReferenceSearchParams = @ReferenceSearchParams, @TokenSearchParams = @TokenSearchParams, @TokenTexts = @TokenTexts, @StringSearchParams = @StringSearchParams, @UriSearchParams = @UriSearchParams, @NumberSearchParams = @NumberSearchParams, @QuantitySearchParams = @QuantitySearchParams, @DateTimeSearchParms = @DateTimeSearchParms, @ReferenceTokenCompositeSearchParams = @ReferenceTokenCompositeSearchParams, @TokenTokenCompositeSearchParams = @TokenTokenCompositeSearchParams, @TokenDateTimeCompositeSearchParams = @TokenDateTimeCompositeSearchParams, @TokenQuantityCompositeSearchParams = @TokenQuantityCompositeSearchParams, @TokenStringCompositeSearchParams = @TokenStringCompositeSearchParams, @TokenNumberNumberCompositeSearchParams = @TokenNumberNumberCompositeSearchParams; - SET @Rows = @Rows + @AffectedRows; - END MERGE INTO dbo.SearchParam AS S USING @SearchParams AS I ON I.Uri = S.Uri @@ -5255,34 +5234,6 @@ BEGIN VALUES (@CheckpointId, @LastProcessedDateTime, @LastProcessedIdentifier, sysutcdatetime()); END -GO -CREATE PROCEDURE dbo.UpdateReindexJob -@id VARCHAR (64), @status VARCHAR (10), @rawJobRecord VARCHAR (MAX), @jobVersion BINARY (8) -AS -SET NOCOUNT ON; -SET XACT_ABORT ON; -BEGIN TRANSACTION; -DECLARE @currentJobVersion AS BINARY (8); -SELECT @currentJobVersion = JobVersion -FROM dbo.ReindexJob WITH (UPDLOCK, HOLDLOCK) -WHERE Id = @id; -IF (@currentJobVersion IS NULL) - BEGIN - THROW 50404, 'Reindex job record not found', 1; - END -IF (@jobVersion <> @currentJobVersion) - BEGIN - THROW 50412, 'Precondition failed', 1; - END -DECLARE @heartbeatDateTime AS DATETIME2 (7) = SYSUTCDATETIME(); -UPDATE dbo.ReindexJob -SET Status = @status, - HeartbeatDateTime = @heartbeatDateTime, - RawJobRecord = @rawJobRecord -WHERE Id = @id; -SELECT @@DBTS; -COMMIT TRANSACTION; - GO CREATE PROCEDURE dbo.UpdateResourceSearchParams @FailedResources INT=0 OUTPUT, @Resources dbo.ResourceList READONLY, @ResourceWriteClaims dbo.ResourceWriteClaimList READONLY, @ReferenceSearchParams dbo.ReferenceSearchParamList READONLY, @TokenSearchParams dbo.TokenSearchParamList READONLY, @TokenTexts dbo.TokenTextList READONLY, @StringSearchParams dbo.StringSearchParamList READONLY, @UriSearchParams dbo.UriSearchParamList READONLY, @NumberSearchParams dbo.NumberSearchParamList READONLY, @QuantitySearchParams dbo.QuantitySearchParamList READONLY, @DateTimeSearchParams dbo.DateTimeSearchParamList READONLY, @ReferenceTokenCompositeSearchParams dbo.ReferenceTokenCompositeSearchParamList READONLY, @TokenTokenCompositeSearchParams dbo.TokenTokenCompositeSearchParamList READONLY, @TokenDateTimeCompositeSearchParams dbo.TokenDateTimeCompositeSearchParamList READONLY, @TokenQuantityCompositeSearchParams dbo.TokenQuantityCompositeSearchParamList READONLY, @TokenStringCompositeSearchParams dbo.TokenStringCompositeSearchParamList READONLY, @TokenNumberNumberCompositeSearchParams dbo.TokenNumberNumberCompositeSearchParamList READONLY diff --git a/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/AcquireReindexJobs.sql b/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/AcquireReindexJobs.sql deleted file mode 100644 index 60a5dbfae9..0000000000 --- a/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/AcquireReindexJobs.sql +++ /dev/null @@ -1,71 +0,0 @@ --- --- STORED PROCEDURE --- Acquires reindex jobs. --- --- DESCRIPTION --- Timestamps the available reindex jobs and sets their statuses to running. --- --- PARAMETERS --- @jobHeartbeatTimeoutThresholdInSeconds --- * The number of seconds that must pass before a reindex job is considered stale --- @maximumNumberOfConcurrentJobsAllowed --- * The maximum number of running jobs we can have at once --- --- RETURN VALUE --- The updated jobs that are now running. --- -CREATE PROCEDURE dbo.AcquireReindexJobs - @jobHeartbeatTimeoutThresholdInSeconds bigint, - @maximumNumberOfConcurrentJobsAllowed int -AS -SET NOCOUNT ON; -SET XACT_ABORT ON; -SET TRANSACTION ISOLATION LEVEL SERIALIZABLE; -BEGIN TRANSACTION; - --- We will consider a job to be stale if its timestamp is smaller than or equal to this. -DECLARE @expirationDateTime AS DATETIME2 (7); -SELECT @expirationDateTime = DATEADD(second, -@jobHeartbeatTimeoutThresholdInSeconds, SYSUTCDATETIME()); - --- Get the number of jobs that are running and not stale. --- Acquire and hold an exclusive table lock for the entire transaction to prevent jobs from being created, updated or deleted during acquisitions. -DECLARE @numberOfRunningJobs AS INT; -SELECT @numberOfRunningJobs = COUNT(*) -FROM dbo.ReindexJob WITH (TABLOCKX) -WHERE Status = 'Running' - AND HeartbeatDateTime > @expirationDateTime; - --- Determine how many available jobs we can pick up. -DECLARE @limit AS INT = @maximumNumberOfConcurrentJobsAllowed - @numberOfRunningJobs; -IF (@limit > 0) - BEGIN - DECLARE @availableJobs TABLE ( - Id VARCHAR (64) COLLATE Latin1_General_100_CS_AS NOT NULL, - JobVersion BINARY (8) NOT NULL); - - -- Get the available jobs, which are reindex jobs that are queued or stale. - -- Older jobs will be prioritized over newer ones. - INSERT INTO @availableJobs - SELECT TOP (@limit) Id, - JobVersion - FROM dbo.ReindexJob - WHERE (Status = 'Queued' - OR (Status = 'Running' - AND HeartbeatDateTime <= @expirationDateTime)) - ORDER BY HeartbeatDateTime; - DECLARE @heartbeatDateTime AS DATETIME2 (7) = SYSUTCDATETIME(); - - -- Update each available job's status to running both in the reindex table's status column and in the raw reindex job record JSON. - UPDATE dbo.ReindexJob - SET Status = 'Running', - HeartbeatDateTime = @heartbeatDateTime, - RawJobRecord = JSON_MODIFY(RawJobRecord, '$.status', 'Running') - OUTPUT inserted.RawJobRecord, inserted.JobVersion - FROM dbo.ReindexJob AS job - INNER JOIN - @availableJobs AS availableJob - ON job.Id = availableJob.Id - AND job.JobVersion = availableJob.JobVersion; - END -COMMIT TRANSACTION; -GO diff --git a/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/CheckSearchParamCacheConsistency.sql b/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/CheckSearchParamCacheConsistency.sql deleted file mode 100644 index c635d3837f..0000000000 --- a/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/CheckSearchParamCacheConsistency.sql +++ /dev/null @@ -1,32 +0,0 @@ ---DROP PROCEDURE dbo.CheckSearchParamCacheConsistency -GO -CREATE OR ALTER PROCEDURE dbo.CheckSearchParamCacheConsistency - @TargetSearchParamLastUpdated varchar(100) - ,@SyncStartDate datetime2(7) - ,@ActiveHostsSince datetime2(7) - ,@StalenessThresholdMinutes int = 10 -AS -set nocount on -SELECT HostName - ,CAST(NULL AS datetime2(7)) AS SyncEventDate - ,CAST(NULL AS nvarchar(3500)) AS EventText - FROM dbo.EventLog - WHERE EventDate >= @ActiveHostsSince - AND HostName IS NOT NULL - AND Process = 'DequeueJob' - -UNION ALL - -SELECT HostName - ,EventDate - ,EventText - FROM dbo.EventLog - WHERE EventDate >= @SyncStartDate - AND HostName IS NOT NULL - AND Process = 'SearchParameterCacheRefresh' - AND Status = 'End' -GO -INSERT INTO dbo.Parameters (Id, Char) SELECT 'DequeueJob', 'LogEvent' WHERE NOT EXISTS (SELECT * FROM dbo.Parameters WHERE Id = 'DequeueJob') -GO -INSERT INTO Parameters (Id,Char) SELECT 'SearchParameterCacheRefresh','LogEvent' -GO diff --git a/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/DequeueJob.sql b/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/DequeueJob.sql index ebbfe6b175..b7c3c46453 100644 --- a/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/DequeueJob.sql +++ b/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/DequeueJob.sql @@ -155,5 +155,3 @@ BEGIN CATCH THROW END CATCH GO -INSERT INTO Parameters (Id,Char) SELECT 'DequeueJob','LogEvent' -GO diff --git a/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/GetSearchParamCacheUpdateEvents.sql b/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/GetSearchParamCacheUpdateEvents.sql new file mode 100644 index 0000000000..f5736f2c8a --- /dev/null +++ b/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/GetSearchParamCacheUpdateEvents.sql @@ -0,0 +1,19 @@ +--DROP PROCEDURE dbo.GetSearchParamCacheUpdateEvents +GO +CREATE PROCEDURE dbo.GetSearchParamCacheUpdateEvents @UpdateProcess varchar(100), @UpdateEventsSince datetime, @ActiveHostsSince datetime +AS +set nocount on +DECLARE @SP varchar(100) = object_name(@@procid) + ,@Mode varchar(200) = 'Process='+@UpdateProcess+' EventsSince='+convert(varchar(23),@UpdateEventsSince,126)+' HostsSince='+convert(varchar(23),@ActiveHostsSince,126) + ,@st datetime = getUTCdate() + +SELECT EventDate + ,EventText = CASE WHEN Process = @UpdateProcess AND EventDate > @UpdateEventsSince THEN EventText ELSE NULL END + ,HostName + FROM dbo.EventLog + WHERE EventDate > @ActiveHostsSince + +EXECUTE dbo.LogEvent @Process=@SP,@Mode=@Mode,@Status='End',@Rows=@@rowcount,@Start=@st +GO +INSERT INTO dbo.Parameters (Id, Char) SELECT 'GetSearchParamCacheUpdateEvents', 'LogEvent' +GO diff --git a/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/GetSearchParamMaxLastUpdated.sql b/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/GetSearchParamMaxLastUpdated.sql deleted file mode 100644 index 6b09aec83f..0000000000 --- a/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/GetSearchParamMaxLastUpdated.sql +++ /dev/null @@ -1,41 +0,0 @@ -/************************************************************* - Stored procedure for getting max LastUpdated timestamp from SearchParam table -**************************************************************/ --- --- STORED PROCEDURE --- GetSearchParamMaxLastUpdated --- --- DESCRIPTION --- Gets the maximum LastUpdated timestamp from the SearchParam table. --- This is used for efficient cache validation without retrieving all records. --- Includes error handling and EventLog logging. --- --- RETURN VALUE --- The maximum LastUpdated timestamp or NULL if no records exist. --- -CREATE PROCEDURE dbo.GetSearchParamMaxLastUpdated -AS -SET NOCOUNT ON - -DECLARE @SP varchar(100) = object_name(@@procid) - ,@Mode varchar(200) = 'SearchParam MaxLastUpdated Query' - ,@st datetime = getUTCdate() - ,@MaxLastUpdated datetimeoffset(7) - -BEGIN TRY - EXECUTE dbo.LogEvent @Process=@SP,@Mode=@Mode,@Status='Start',@Start=@st - - -- Get the maximum LastUpdated timestamp from SearchParam table - SELECT @MaxLastUpdated = MAX(LastUpdated) - FROM dbo.SearchParam - - -- Return the result - SELECT @MaxLastUpdated AS MaxLastUpdated - - EXECUTE dbo.LogEvent @Process=@SP,@Mode=@Mode,@Status='End',@Start=@st,@Rows=@@ROWCOUNT -END TRY -BEGIN CATCH - EXECUTE dbo.LogEvent @Process=@SP,@Mode=@Mode,@Status='Error',@Start=@st; - THROW -END CATCH -GO diff --git a/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/MergeResourcesAndSearchParams.sql b/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/MergeResourcesAndSearchParams.sql new file mode 100644 index 0000000000..2f65ddba4e --- /dev/null +++ b/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/MergeResourcesAndSearchParams.sql @@ -0,0 +1,123 @@ +CREATE PROCEDURE dbo.MergeResourcesAndSearchParams + @SearchParams dbo.SearchParamList READONLY + ,@IsResourceChangeCaptureEnabled bit = 0 + ,@TransactionId bigint = NULL + ,@Resources dbo.ResourceList READONLY + ,@ResourceWriteClaims dbo.ResourceWriteClaimList READONLY + ,@ReferenceSearchParams dbo.ReferenceSearchParamList READONLY + ,@TokenSearchParams dbo.TokenSearchParamList READONLY + ,@TokenTexts dbo.TokenTextList READONLY + ,@StringSearchParams dbo.StringSearchParamList READONLY + ,@UriSearchParams dbo.UriSearchParamList READONLY + ,@NumberSearchParams dbo.NumberSearchParamList READONLY + ,@QuantitySearchParams dbo.QuantitySearchParamList READONLY + ,@DateTimeSearchParms dbo.DateTimeSearchParamList READONLY + ,@ReferenceTokenCompositeSearchParams dbo.ReferenceTokenCompositeSearchParamList READONLY + ,@TokenTokenCompositeSearchParams dbo.TokenTokenCompositeSearchParamList READONLY + ,@TokenDateTimeCompositeSearchParams dbo.TokenDateTimeCompositeSearchParamList READONLY + ,@TokenQuantityCompositeSearchParams dbo.TokenQuantityCompositeSearchParamList READONLY + ,@TokenStringCompositeSearchParams dbo.TokenStringCompositeSearchParamList READONLY + ,@TokenNumberNumberCompositeSearchParams dbo.TokenNumberNumberCompositeSearchParamList READONLY +AS +set nocount on +DECLARE @SP varchar(100) = object_name(@@procid) + ,@Mode varchar(200) = 'Cnt='+convert(varchar,(SELECT count(*) FROM @SearchParams)) + ,@st datetime = getUTCdate() + ,@LastUpdated datetimeoffset(7) = convert(datetimeoffset(7), sysUTCdatetime()) + ,@msg varchar(4000) + ,@Rows int + ,@AffectedRows int = 0 + ,@Uri varchar(4000) + ,@Status varchar(20) + +DECLARE @SearchParamsCopy dbo.SearchParamList +INSERT INTO @SearchParamsCopy SELECT * FROM @SearchParams +WHILE EXISTS (SELECT * FROM @SearchParamsCopy) +BEGIN + SELECT TOP 1 @Uri = Uri, @Status = Status FROM @SearchParamsCopy + SET @msg = 'Status='+@Status+' Uri='+@Uri + EXECUTE dbo.LogEvent @Process=@SP,@Mode=@Mode,@Status='Start',@Text=@msg + DELETE FROM @SearchParamsCopy WHERE Uri = @Uri +END + +BEGIN TRY + SET TRANSACTION ISOLATION LEVEL SERIALIZABLE + + BEGIN TRANSACTION + + -- Check for concurrency conflicts first using LastUpdated + -- Only the top 60 are included in the message to avoid hitting the 8000 character limit, but all conflicts will cause the transaction to roll back + SELECT TOP 60 @msg = string_agg(S.Uri, ', ') + FROM @SearchParams I JOIN dbo.SearchParam S ON S.Uri = I.Uri + WHERE I.LastUpdated != S.LastUpdated + IF @msg IS NOT NULL + BEGIN + SET @msg = concat('Optimistic concurrency conflict detected for search parameters: ', @msg) + ROLLBACK TRANSACTION; + THROW 50001, @msg, 1 + END + + IF EXISTS (SELECT * FROM @Resources) + BEGIN + EXECUTE dbo.MergeResources + @AffectedRows = @AffectedRows OUTPUT + ,@RaiseExceptionOnConflict = 1 + ,@IsResourceChangeCaptureEnabled = @IsResourceChangeCaptureEnabled + ,@TransactionId = @TransactionId + ,@SingleTransaction = 1 + ,@Resources = @Resources + ,@ResourceWriteClaims = @ResourceWriteClaims + ,@ReferenceSearchParams = @ReferenceSearchParams + ,@TokenSearchParams = @TokenSearchParams + ,@TokenTexts = @TokenTexts + ,@StringSearchParams = @StringSearchParams + ,@UriSearchParams = @UriSearchParams + ,@NumberSearchParams = @NumberSearchParams + ,@QuantitySearchParams = @QuantitySearchParams + ,@DateTimeSearchParms = @DateTimeSearchParms + ,@ReferenceTokenCompositeSearchParams = @ReferenceTokenCompositeSearchParams + ,@TokenTokenCompositeSearchParams = @TokenTokenCompositeSearchParams + ,@TokenDateTimeCompositeSearchParams = @TokenDateTimeCompositeSearchParams + ,@TokenQuantityCompositeSearchParams = @TokenQuantityCompositeSearchParams + ,@TokenStringCompositeSearchParams = @TokenStringCompositeSearchParams + ,@TokenNumberNumberCompositeSearchParams = @TokenNumberNumberCompositeSearchParams; + + SET @Rows = @Rows + @AffectedRows; + END + + MERGE INTO dbo.SearchParam S + USING @SearchParams I ON I.Uri = S.Uri + WHEN MATCHED THEN + UPDATE + SET Status = I.Status + ,LastUpdated = @LastUpdated + ,IsPartiallySupported = I.IsPartiallySupported + WHEN NOT MATCHED BY TARGET THEN + INSERT ( Uri, Status, LastUpdated, IsPartiallySupported) + VALUES (I.Uri, I.Status, @LastUpdated, I.IsPartiallySupported); + + SET @msg = 'LastUpdated='+convert(varchar(23),@LastUpdated,126)+' Merged='+convert(varchar,@@rowcount) + + COMMIT TRANSACTION + + EXECUTE dbo.LogEvent @Process=@SP,@Mode=@Mode,@Status='End',@Start=@st,@Action='Merge',@Rows=@Rows,@Text=@msg +END TRY +BEGIN CATCH + IF @@trancount > 0 ROLLBACK TRANSACTION; + EXECUTE dbo.LogEvent @Process=@SP,@Mode=@Mode,@Status='Error',@Start=@st; + THROW +END CATCH +GO +INSERT INTO Parameters (Id,Char) SELECT 'MergeResourcesAndSearchParams','LogEvent' +GO +--DECLARE @SearchParams dbo.SearchParamList +--INSERT INTO @SearchParams +-- --SELECT 'http://example.org/fhir/SearchParameter/custom-mixed-base-d9e18fc8', 'Enabled', 0, '2026-01-26 17:15:43.0364438 -08:00' +-- SELECT 'Test', 'Enabled', 0, '2026-01-26 17:15:43.0364438 -08:00' +--INSERT INTO @SearchParams +-- SELECT 'Test2', 'Enabled', 0, '2026-01-26 17:15:43.0364438 -08:00' +--SELECT * FROM @SearchParams +--EXECUTE dbo.MergeResourcesAndSearchParams @SearchParams +--SELECT TOP 100 * FROM SearchParam ORDER BY SearchParamId DESC +--DELETE FROM SearchParam WHERE Uri LIKE 'Test%' +--SELECT TOP 10 * FROM EventLog ORDER BY EventDate DESC diff --git a/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/MergeSearchParams.sql b/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/MergeSearchParams.sql index 9bc4ddfbf6..5347f2e50b 100644 --- a/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/MergeSearchParams.sql +++ b/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/MergeSearchParams.sql @@ -1,22 +1,4 @@ CREATE PROCEDURE dbo.MergeSearchParams @SearchParams dbo.SearchParamList READONLY - ,@IsResourceChangeCaptureEnabled bit = 0 - ,@TransactionId bigint = NULL - ,@Resources dbo.ResourceList READONLY - ,@ResourceWriteClaims dbo.ResourceWriteClaimList READONLY - ,@ReferenceSearchParams dbo.ReferenceSearchParamList READONLY - ,@TokenSearchParams dbo.TokenSearchParamList READONLY - ,@TokenTexts dbo.TokenTextList READONLY - ,@StringSearchParams dbo.StringSearchParamList READONLY - ,@UriSearchParams dbo.UriSearchParamList READONLY - ,@NumberSearchParams dbo.NumberSearchParamList READONLY - ,@QuantitySearchParams dbo.QuantitySearchParamList READONLY - ,@DateTimeSearchParms dbo.DateTimeSearchParamList READONLY - ,@ReferenceTokenCompositeSearchParams dbo.ReferenceTokenCompositeSearchParamList READONLY - ,@TokenTokenCompositeSearchParams dbo.TokenTokenCompositeSearchParamList READONLY - ,@TokenDateTimeCompositeSearchParams dbo.TokenDateTimeCompositeSearchParamList READONLY - ,@TokenQuantityCompositeSearchParams dbo.TokenQuantityCompositeSearchParamList READONLY - ,@TokenStringCompositeSearchParams dbo.TokenStringCompositeSearchParamList READONLY - ,@TokenNumberNumberCompositeSearchParams dbo.TokenNumberNumberCompositeSearchParamList READONLY AS set nocount on DECLARE @SP varchar(100) = object_name(@@procid) @@ -25,7 +7,6 @@ DECLARE @SP varchar(100) = object_name(@@procid) ,@LastUpdated datetimeoffset(7) = sysdatetimeoffset() ,@msg varchar(4000) ,@Rows int - ,@AffectedRows int = 0 ,@Uri varchar(4000) ,@Status varchar(20) @@ -58,34 +39,6 @@ BEGIN TRY THROW 50001, @msg, 1 END - IF EXISTS (SELECT * FROM @Resources) - BEGIN - EXECUTE dbo.MergeResources - @AffectedRows = @AffectedRows OUTPUT - ,@RaiseExceptionOnConflict = 1 - ,@IsResourceChangeCaptureEnabled = @IsResourceChangeCaptureEnabled - ,@TransactionId = @TransactionId - ,@SingleTransaction = 1 - ,@Resources = @Resources - ,@ResourceWriteClaims = @ResourceWriteClaims - ,@ReferenceSearchParams = @ReferenceSearchParams - ,@TokenSearchParams = @TokenSearchParams - ,@TokenTexts = @TokenTexts - ,@StringSearchParams = @StringSearchParams - ,@UriSearchParams = @UriSearchParams - ,@NumberSearchParams = @NumberSearchParams - ,@QuantitySearchParams = @QuantitySearchParams - ,@DateTimeSearchParms = @DateTimeSearchParms - ,@ReferenceTokenCompositeSearchParams = @ReferenceTokenCompositeSearchParams - ,@TokenTokenCompositeSearchParams = @TokenTokenCompositeSearchParams - ,@TokenDateTimeCompositeSearchParams = @TokenDateTimeCompositeSearchParams - ,@TokenQuantityCompositeSearchParams = @TokenQuantityCompositeSearchParams - ,@TokenStringCompositeSearchParams = @TokenStringCompositeSearchParams - ,@TokenNumberNumberCompositeSearchParams = @TokenNumberNumberCompositeSearchParams; - - SET @Rows = @Rows + @AffectedRows; - END - MERGE INTO dbo.SearchParam S USING @SearchParams I ON I.Uri = S.Uri WHEN MATCHED THEN diff --git a/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/UpdateReindexJob.sql b/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/UpdateReindexJob.sql deleted file mode 100644 index 2f2fd23c83..0000000000 --- a/src/Microsoft.Health.Fhir.SqlServer/Features/Schema/Sql/Sprocs/UpdateReindexJob.sql +++ /dev/null @@ -1,58 +0,0 @@ -/************************************************************* - Stored procedures - UpdateReindexJob -**************************************************************/ --- --- STORED PROCEDURE --- Updates a reindex job. --- --- DESCRIPTION --- Modifies an existing job in the ReindexJob table. --- --- PARAMETERS --- @id --- * The ID of the reindex job record --- @status --- * The status of the reindex job --- @rawJobRecord --- * A JSON document --- @jobVersion --- * The version of the job to update must match this --- --- RETURN VALUE --- The row version of the updated reindex job. --- -CREATE PROCEDURE dbo.UpdateReindexJob - @id varchar(64), - @status varchar(10), - @rawJobRecord varchar(max), - @jobVersion binary(8) -AS -SET NOCOUNT ON; -SET XACT_ABORT ON; -BEGIN TRANSACTION; -DECLARE @currentJobVersion AS BINARY (8); - --- Acquire and hold an update lock on a row in the ReindexJob table for the entire transaction. --- This ensures the version check and update occur atomically. -SELECT @currentJobVersion = JobVersion -FROM dbo.ReindexJob WITH (UPDLOCK, HOLDLOCK) -WHERE Id = @id; -IF (@currentJobVersion IS NULL) - BEGIN - THROW 50404, 'Reindex job record not found', 1; - END -IF (@jobVersion <> @currentJobVersion) - BEGIN - THROW 50412, 'Precondition failed', 1; - END - --- We will timestamp the jobs when we update them to track stale jobs. -DECLARE @heartbeatDateTime AS DATETIME2 (7) = SYSUTCDATETIME(); -UPDATE dbo.ReindexJob -SET Status = @status, - HeartbeatDateTime = @heartbeatDateTime, - RawJobRecord = @rawJobRecord -WHERE Id = @id; -SELECT @@DBTS; -COMMIT TRANSACTION; -GO 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 8fb763aba0..7aa6f89558 100644 --- a/src/Microsoft.Health.Fhir.SqlServer/Features/Storage/Registry/SqlServerSearchParameterStatusDataStore.cs +++ b/src/Microsoft.Health.Fhir.SqlServer/Features/Storage/Registry/SqlServerSearchParameterStatusDataStore.cs @@ -31,7 +31,7 @@ namespace Microsoft.Health.Fhir.SqlServer.Features.Storage.Registry { internal class SqlServerSearchParameterStatusDataStore : ISearchParameterStatusDataStore { - private const string SearchParamLastUpdatedPrefix = "SearchParamLastUpdated="; + public const string SearchParamCacheUpdateProcess = "SearchParamCacheUpdate"; private readonly ISqlRetryService _sqlRetryService; private readonly SchemaInformation _schemaInformation; @@ -59,6 +59,8 @@ public SqlServerSearchParameterStatusDataStore( _logger = logger; } + public string SearchParamCacheUpdateProcessName => SearchParamCacheUpdateProcess; + public async Task TryLogEvent(string process, string status, string text, DateTime? startDate, CancellationToken cancellationToken) { await _sqlRetryService.TryLogEvent(process, status, text, startDate, cancellationToken); @@ -187,52 +189,18 @@ private async Task UpsertStatusesInternal(IReadOnlyCollection= 109) { - cmd.Parameters.AddWithValue("@IsResourceChangeCaptureEnabled", false); - cmd.Parameters.Add(new SqlParameter("@TransactionId", SqlDbType.BigInt) { Value = DBNull.Value }); - - new ResourceListTableValuedParameterDefinition("@Resources").AddParameter(cmd.Parameters, Array.Empty()); - new ResourceWriteClaimListTableValuedParameterDefinition("@ResourceWriteClaims").AddParameter(cmd.Parameters, Array.Empty()); - new ReferenceSearchParamListTableValuedParameterDefinition("@ReferenceSearchParams").AddParameter(cmd.Parameters, Array.Empty()); - new TokenSearchParamListTableValuedParameterDefinition("@TokenSearchParams").AddParameter(cmd.Parameters, Array.Empty()); - new TokenTextListTableValuedParameterDefinition("@TokenTexts").AddParameter(cmd.Parameters, Array.Empty()); - new StringSearchParamListTableValuedParameterDefinition("@StringSearchParams").AddParameter(cmd.Parameters, Array.Empty()); - new UriSearchParamListTableValuedParameterDefinition("@UriSearchParams").AddParameter(cmd.Parameters, Array.Empty()); - new NumberSearchParamListTableValuedParameterDefinition("@NumberSearchParams").AddParameter(cmd.Parameters, Array.Empty()); - new QuantitySearchParamListTableValuedParameterDefinition("@QuantitySearchParams").AddParameter(cmd.Parameters, Array.Empty()); - new DateTimeSearchParamListTableValuedParameterDefinition("@DateTimeSearchParms").AddParameter(cmd.Parameters, Array.Empty()); - new ReferenceTokenCompositeSearchParamListTableValuedParameterDefinition("@ReferenceTokenCompositeSearchParams").AddParameter(cmd.Parameters, Array.Empty()); - new TokenTokenCompositeSearchParamListTableValuedParameterDefinition("@TokenTokenCompositeSearchParams").AddParameter(cmd.Parameters, Array.Empty()); - new TokenDateTimeCompositeSearchParamListTableValuedParameterDefinition("@TokenDateTimeCompositeSearchParams").AddParameter(cmd.Parameters, Array.Empty()); - new TokenQuantityCompositeSearchParamListTableValuedParameterDefinition("@TokenQuantityCompositeSearchParams").AddParameter(cmd.Parameters, Array.Empty()); - new TokenStringCompositeSearchParamListTableValuedParameterDefinition("@TokenStringCompositeSearchParams").AddParameter(cmd.Parameters, Array.Empty()); - new TokenNumberNumberCompositeSearchParamListTableValuedParameterDefinition("@TokenNumberNumberCompositeSearchParams").AddParameter(cmd.Parameters, Array.Empty()); + cmd.CommandText = "dbo.MergeResourcesAndSearchParams"; } - - var results = await cmd.ExecuteReaderAsync( - _sqlRetryService, - (reader) => { return reader.ReadRow(VLatest.SearchParam.SearchParamId, VLatest.SearchParam.Uri, VLatest.SearchParam.LastUpdated); }, - _logger, - cancellationToken); - - foreach (var result in results) + else { - (short searchParamId, string searchParamUri, DateTimeOffset lastUpdated) = result; + cmd.CommandText = "dbo.MergeSearchParams"; + } - // Add the new search parameters to the FHIR model dictionary. - _fhirModel.TryAddSearchParamIdToUriMapping(searchParamUri, searchParamId); + new SearchParamListTableValuedParameterDefinition("@SearchParams").AddParameter(cmd.Parameters, new SearchParamListRowGenerator().GenerateRows(statuses.ToList())); - // Update the LastUpdated in our original collection for future operations - var matchingStatus = statuses.FirstOrDefault(s => s.Uri.OriginalString == searchParamUri); - if (matchingStatus != null) - { - matchingStatus.LastUpdated = lastUpdated; - } - } + await cmd.ExecuteNonQueryAsync(_sqlRetryService, _logger, cancellationToken); } // Synchronize the FHIR model dictionary with the data in SQL search parameter status table @@ -247,84 +215,56 @@ public void SyncStatuses(IReadOnlyCollection stat } } - public async Task CheckCacheConsistencyAsync(string targetSearchParamLastUpdated, DateTime syncStartDate, DateTime activeHostsSince, CancellationToken cancellationToken) + public async Task CheckCacheConsistencyAsync(DateTime updateEventsSince, DateTime activeHostsSince, CancellationToken cancellationToken) { - EnsureArg.IsNotNullOrWhiteSpace(targetSearchParamLastUpdated, nameof(targetSearchParamLastUpdated)); - - if (_schemaInformation.Current < (int)SchemaVersion.V108) + if (_schemaInformation.Current < (int)SchemaVersion.V109) // Pre-V109 schemas don't have the sproc; assume inconsistent { - // Pre-V108 schemas don't have the sproc; assume consistent - return new CacheConsistencyResult { IsConsistent = true, TotalActiveHosts = 1, ConvergedHosts = 1 }; + return new CacheConsistencyResult { IsConsistent = false, ActiveHosts = 0, ConvergedHosts = 0 }; } using var cmd = new SqlCommand(); cmd.CommandType = CommandType.StoredProcedure; - cmd.CommandText = "dbo.CheckSearchParamCacheConsistency"; - cmd.Parameters.AddWithValue("@TargetSearchParamLastUpdated", targetSearchParamLastUpdated); - cmd.Parameters.AddWithValue("@SyncStartDate", syncStartDate); + cmd.CommandText = "dbo.GetSearchParamCacheUpdateEvents"; + cmd.Parameters.AddWithValue("@UpdateProcess", SearchParamCacheUpdateProcess); + cmd.Parameters.AddWithValue("@UpdateEventsSince", updateEventsSince); cmd.Parameters.AddWithValue("@ActiveHostsSince", activeHostsSince); var results = await cmd.ExecuteReaderAsync( _sqlRetryService, (reader) => { - var hostName = reader.GetString(0); - var syncEventDate = reader.IsDBNull(1) ? (DateTime?)null : reader.GetDateTime(1); - var eventText = reader.IsDBNull(2) ? null : reader.GetString(2); - return (hostName, syncEventDate, eventText); + var eventDate = reader.GetDateTime(0); + var lastUpdated = reader.IsDBNull(1) ? null : reader.GetString(1); // Use event text as-is because date is saved in a sortable format. + var hostName = reader.GetString(2); + return (eventDate, lastUpdated, hostName); }, _logger, cancellationToken); - if (results.Count == 0) - { - // If no results, assume consistent (could be a fresh database) - return new CacheConsistencyResult { IsConsistent = true, TotalActiveHosts = 0, ConvergedHosts = 0 }; - } - - var activeHosts = new HashSet(StringComparer.OrdinalIgnoreCase); - var latestSyncByHost = new Dictionary(StringComparer.OrdinalIgnoreCase); + var activeHosts = results.Select(r => r.hostName).ToHashSet(); - foreach (var (hostName, syncEventDate, eventText) in results) + // Taking 2 latest events should guarantee that cache completed at least one full update cycle after updateEventsSince. + var eventsByHosts = results.Where(_ => !string.IsNullOrEmpty(_.lastUpdated)) + .GroupBy(_ => _.hostName) + .Select(g => new { g.Key, Value = g.OrderByDescending(_ => _.eventDate).Take(2).ToList() }) + .ToDictionary(_ => _.Key, _ => _.Value); + var updatedHosts = new Dictionary(); + foreach (var hostName in activeHosts) { - activeHosts.Add(hostName); - - if (syncEventDate.HasValue && !string.IsNullOrEmpty(eventText)) + // There is always a time gap of several milliseconds to several seconds between search indexes generation and surrogate ids assignment. + // We need to make sure that during this time search param cache does not change. + // Hence we check that last updated is identical on last two update events. + if (eventsByHosts.TryGetValue(hostName, out var value) && value.Count == 2 && value[0].lastUpdated == value[1].lastUpdated) { - if (!latestSyncByHost.TryGetValue(hostName, out var existingSync) - || syncEventDate.Value > existingSync.SyncEventDate) - { - latestSyncByHost[hostName] = (syncEventDate.Value, eventText); - } + updatedHosts.Add(hostName, value[0].lastUpdated); } } - int totalActiveHosts = activeHosts.Count; - int totalConvergedHosts = activeHosts.Count(hostName => - latestSyncByHost.TryGetValue(hostName, out var latestSync) - && TryGetLoggedSearchParamLastUpdated(latestSync.EventText, out string loggedSearchParamLastUpdated) - && StringComparer.Ordinal.Compare(loggedSearchParamLastUpdated, targetSearchParamLastUpdated) >= 0); - - return new CacheConsistencyResult - { - IsConsistent = totalActiveHosts > 0 && totalConvergedHosts >= totalActiveHosts, - TotalActiveHosts = totalActiveHosts, - ConvergedHosts = totalConvergedHosts, - }; - } - - private static bool TryGetLoggedSearchParamLastUpdated(string eventText, out string loggedSearchParamLastUpdated) - { - if (!string.IsNullOrEmpty(eventText) - && eventText.StartsWith(SearchParamLastUpdatedPrefix, StringComparison.Ordinal) - && eventText.Length > SearchParamLastUpdatedPrefix.Length) - { - loggedSearchParamLastUpdated = eventText.Substring(SearchParamLastUpdatedPrefix.Length); - return true; - } - - loggedSearchParamLastUpdated = null; - return false; + var maxLastUpdated = updatedHosts.Values.Max(_ => _); + var convergedHosts = updatedHosts.Where(_ => _.Value == maxLastUpdated).Select(_ => _.Key).ToList(); + var isConsistent = convergedHosts.Count > 0 && convergedHosts.Count == activeHosts.Count; + await TryLogEvent("CheckCacheConsistency", "Warn", $"isConsistent={isConsistent} ActiveHosts={activeHosts.Count} ConvergedHosts={convergedHosts.Count}", null, cancellationToken); + return new CacheConsistencyResult { IsConsistent = isConsistent, ActiveHosts = activeHosts.Count, ConvergedHosts = convergedHosts.Count }; } } } diff --git a/src/Microsoft.Health.Fhir.SqlServer/Features/Storage/SqlQueueClient.cs b/src/Microsoft.Health.Fhir.SqlServer/Features/Storage/SqlQueueClient.cs index 316a979c19..8e85924bbf 100644 --- a/src/Microsoft.Health.Fhir.SqlServer/Features/Storage/SqlQueueClient.cs +++ b/src/Microsoft.Health.Fhir.SqlServer/Features/Storage/SqlQueueClient.cs @@ -176,12 +176,16 @@ public async Task> EnqueueAsync(byte queueType, string[] } } - public async Task EnqueueWithStatusAsync(byte queueType, long groupId, string definition, JobStatus jobStatus, string result, DateTime? startDate, CancellationToken cancellationToken) + public async Task EnqueueWithStatusAsync(byte queueType, long? groupId, string definition, JobStatus jobStatus, string result, DateTime? startDate, CancellationToken cancellationToken) { using var sqlCommand = new SqlCommand() { CommandText = "dbo.EnqueueJobs", CommandType = CommandType.StoredProcedure, CommandTimeout = 300 }; sqlCommand.Parameters.AddWithValue("@QueueType", queueType); new StringListTableValuedParameterDefinition("@Definitions").AddParameter(sqlCommand.Parameters, [new StringListRow(definition)]); - sqlCommand.Parameters.AddWithValue("@GroupId", groupId); + if (groupId.HasValue) + { + sqlCommand.Parameters.AddWithValue("@GroupId", groupId.Value); + } + sqlCommand.Parameters.AddWithValue("@Status", jobStatus); if (startDate.HasValue) { diff --git a/src/Microsoft.Health.Fhir.SqlServer/Features/Storage/SqlServerFhirDataStore.cs b/src/Microsoft.Health.Fhir.SqlServer/Features/Storage/SqlServerFhirDataStore.cs index 622c378060..c5b4c3653e 100644 --- a/src/Microsoft.Health.Fhir.SqlServer/Features/Storage/SqlServerFhirDataStore.cs +++ b/src/Microsoft.Health.Fhir.SqlServer/Features/Storage/SqlServerFhirDataStore.cs @@ -801,9 +801,9 @@ internal async Task MergeResourcesWrapperAsync(long transactionId, bool singleTr cmd.CommandType = CommandType.StoredProcedure; bool hasPendingStatuses = pendingStatuses?.Count > 0; - if (hasPendingStatuses) + if (hasPendingStatuses && _schemaInformation.Current >= 109) { - cmd.CommandText = "dbo.MergeSearchParams"; + cmd.CommandText = "dbo.MergeResourcesAndSearchParams"; new SearchParamListTableValuedParameterDefinition("@SearchParams").AddParameter(cmd.Parameters, new SearchParamListRowGenerator().GenerateRows(pendingStatuses.ToList())); } else diff --git a/src/Microsoft.Health.TaskManagement.UnitTests/TestQueueClient.cs b/src/Microsoft.Health.TaskManagement.UnitTests/TestQueueClient.cs index ae2ea91670..68b71b0dd9 100644 --- a/src/Microsoft.Health.TaskManagement.UnitTests/TestQueueClient.cs +++ b/src/Microsoft.Health.TaskManagement.UnitTests/TestQueueClient.cs @@ -150,11 +150,12 @@ public Task DequeueAsync(byte queueType, string worker, int heartbeatTi job = jobInfos.FirstOrDefault(t => t.QueueType == queueType && (t.Status == JobStatus.Created || - (t.Status == JobStatus.Running && (DateTime.Now - t.HeartbeatDateTime) > TimeSpan.FromSeconds(heartbeatTimeoutSec)))); + (t.Status == JobStatus.Running && (DateTime.UtcNow - t.HeartbeatDateTime) > TimeSpan.FromSeconds(heartbeatTimeoutSec)))); if (job != null) { job.Status = JobStatus.Running; - job.HeartbeatDateTime = DateTime.Now; + job.StartDate = DateTime.UtcNow; + job.HeartbeatDateTime = DateTime.UtcNow; } } @@ -182,15 +183,11 @@ public Task> EnqueueAsync(byte queueType, string[] defini Id = largestId, GroupId = gId, Status = JobStatus.Created, + CreateDate = DateTime.UtcNow, HeartbeatDateTime = DateTime.UtcNow, QueueType = queueType, }; - if (newJob.Status == JobStatus.Created) - { - newJob.CreateDate = DateTime.UtcNow; - } - result.Add(newJob); jobInfos.Add(newJob); } @@ -201,7 +198,7 @@ public Task> EnqueueAsync(byte queueType, string[] defini return Task.FromResult>(result); } - public Task EnqueueWithStatusAsync(byte queueType, long groupId, string definition, JobStatus jobStatus, string result, DateTime? startDate, CancellationToken cancellationToken) + public Task EnqueueWithStatusAsync(byte queueType, long? groupId, string definition, JobStatus jobStatus, string result, DateTime? startDate, CancellationToken cancellationToken) { var response = new List(); @@ -216,17 +213,15 @@ public Task EnqueueWithStatusAsync(byte queueType, long groupId, string { Definition = definition, Id = largestId, - GroupId = groupId, + GroupId = groupId.HasValue ? groupId.Value : largestId, Status = jobStatus, + CreateDate = DateTime.UtcNow, + StartDate = jobStatus == JobStatus.Running ? DateTime.UtcNow : null, + EndDate = jobStatus == JobStatus.Completed || jobStatus == JobStatus.Failed ? DateTime.UtcNow : null, HeartbeatDateTime = DateTime.UtcNow, QueueType = queueType, }; - if (newJob.Status == JobStatus.Created) - { - newJob.CreateDate = DateTime.UtcNow; - } - response.Add(newJob); jobInfos.Add(newJob); } @@ -285,7 +280,7 @@ public Task PutJobHeartbeatAsync(JobInfo jobInfo, CancellationToken cancel throw new JobNotExistException("not exist"); } - job.HeartbeatDateTime = DateTime.Now; + job.HeartbeatDateTime = DateTime.UtcNow; job.Result = jobInfo.Result; cancel = job.CancelRequested; diff --git a/src/Microsoft.Health.TaskManagement/IQueueClient.cs b/src/Microsoft.Health.TaskManagement/IQueueClient.cs index 017abc7a54..2f4a63930a 100644 --- a/src/Microsoft.Health.TaskManagement/IQueueClient.cs +++ b/src/Microsoft.Health.TaskManagement/IQueueClient.cs @@ -38,7 +38,7 @@ public interface IQueueClient /// or status to track table-level defragmentation /// progress within a coordinator job group. /// - public Task EnqueueWithStatusAsync(byte queueType, long groupId, string definition, JobStatus jobStatus, string result, DateTime? startDate, CancellationToken cancellationToken); + public Task EnqueueWithStatusAsync(byte queueType, long? groupId, string definition, JobStatus jobStatus, string result, DateTime? startDate, CancellationToken cancellationToken); /// /// Dequeue multiple jobs 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 d871cef997..d9ba5eab58 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 @@ -8,6 +8,7 @@ using System.Diagnostics; using System.Linq; using System.Net; +using System.Reflection.Metadata; using System.Security.Cryptography; using System.Threading; using System.Threading.Tasks; @@ -44,6 +45,254 @@ public ReindexTests(HttpIntegrationTestFixture fixture, ITestOutputHelper output _output = output; } + [Fact] + public async Task GivenTwoSearchParamsWithCodeConflictOnDerived_ThenBadRequestIsReturned() + { + await CancelAnyRunningReindexJobsAsync(); +#if R5 + var personTypes = new List() { VersionIndependentResourceTypesAll.Person }; + var resourceTypes = new List() { VersionIndependentResourceTypesAll.Resource }; +#else + var personTypes = new List() { ResourceType.Person}; + var resourceTypes = new List() { ResourceType.Resource }; +#endif + const string urlPrefix = "http://my.org/"; + var ids = new List { "c-id-1", "c-id-2" }; + var code = "same-code"; + try + { + var bundle = new Bundle { Type = Bundle.BundleType.Batch, Entry = [] }; + + var id = ids[0]; + var searchParam = new SearchParameter + { + Id = id, + Url = $"{urlPrefix}c-1", + Name = code, + Code = code, + Status = PublicationStatus.Active, + Type = SearchParamType.Token, + Expression = "Person.id", + Description = "any", + Base = personTypes, + }; + + bundle.Entry.Add(new EntryComponent { Request = new RequestComponent { Method = Bundle.HTTPVerb.PUT, Url = $"SearchParameter/{id}" }, Resource = searchParam }); + + id = ids[1]; + searchParam = new SearchParameter + { + Id = id, + Url = $"{urlPrefix}c-2", + Name = code, + Code = code, + Status = PublicationStatus.Active, + Type = SearchParamType.Token, + Expression = "Resource.id", + Description = "any", + Base = resourceTypes, + }; + + bundle.Entry.Add(new EntryComponent { Request = new RequestComponent { Method = Bundle.HTTPVerb.PUT, Url = $"SearchParameter/{id}" }, Resource = searchParam }); + + await _fixture.TestFhirClient.PostBundleAsync(bundle, new FhirBundleOptions { BundleProcessingLogic = FhirBundleProcessingLogic.Parallel }); + Assert.Fail("This point should not be reached."); + } + catch (FhirClientException ex) + { + Assert.Equal(HttpStatusCode.BadRequest, ex.StatusCode); + Assert.Contains($"Input search parameters have duplicate codes [(Person, {code})]", ex.Message); + } + finally + { + await DeleteSearchParamsAsync(ids); + } + } + + [Fact] + public async Task GivenTwoSearchParamsForDifferentResourceTypesUsingSameCode_ThenBothCreated() + { + await CancelAnyRunningReindexJobsAsync(); +#if R5 + var personTypes = new List() { VersionIndependentResourceTypesAll.Person }; + var supplyDeliveryTypes = new List() { VersionIndependentResourceTypesAll.SupplyDelivery }; +#else + var personTypes = new List() { ResourceType.Person }; + var supplyDeliveryTypes = new List() { ResourceType.SupplyDelivery }; +#endif + const string urlPrefix = "http://my.org/"; + var ids = new List { "c-id-1", "c-id-2" }; + try + { + var bundle = new Bundle { Type = Bundle.BundleType.Batch, Entry = [] }; + + var code = "same-code"; + var id = ids[0]; + var searchParam = new SearchParameter + { + Id = id, + Url = $"{urlPrefix}c-1", + Name = code, + Code = code, + Status = PublicationStatus.Active, + Type = SearchParamType.Token, + Expression = "Person.id", + Description = "any", + Base = personTypes, + }; + + bundle.Entry.Add(new EntryComponent { Request = new RequestComponent { Method = Bundle.HTTPVerb.PUT, Url = $"SearchParameter/{id}" }, Resource = searchParam }); + + id = ids[1]; + searchParam = new SearchParameter + { + Id = id, + Url = $"{urlPrefix}c-2", + Name = code, + Code = code, + Status = PublicationStatus.Active, + Type = SearchParamType.Token, + Expression = "SupplyDelivery.id", + Description = "any", + Base = supplyDeliveryTypes, + }; + + bundle.Entry.Add(new EntryComponent { Request = new RequestComponent { Method = Bundle.HTTPVerb.PUT, Url = $"SearchParameter/{id}" }, Resource = searchParam }); + + var response = await _fixture.TestFhirClient.PostBundleAsync(bundle, new FhirBundleOptions { BundleProcessingLogic = FhirBundleProcessingLogic.Parallel }); + Assert.Equal(2, response.Resource.Entry.Count); + Assert.All(response.Resource.Entry, _ => Assert.NotNull(_.Resource as SearchParameter)); + } + finally + { + await DeleteSearchParamsAsync(ids); + } + } + + [Theory] + [InlineData(true, false, true, false)] + [InlineData(true, false, false, false)] + [InlineData(false, true, true, false)] + [InlineData(false, true, false, false)] + [InlineData(true, false, true, true)] + [InlineData(true, false, false, true)] + [InlineData(false, true, true, true)] + [InlineData(false, true, false, true)] + [InlineData(true, true, false, true)] + [InlineData(true, true, false, false)] + [InlineData(true, true, true, true)] + [InlineData(true, true, true, false)] + public async Task GivenTwoPersonSearchParams_ThenBadRequestIsReturnedForConflicts(bool dupCodes, bool dupUrls, bool isParallel, bool isBatch) + { + if (!_isSql && !isBatch) // cosmos does not support transaction bundles + { + return; + } + + await CancelAnyRunningReindexJobsAsync(); + + const string urlPrefix = "http://my.org/"; + var codes = new List(); + var urls = new List(); + var ids = new List(); + try + { + ids.Add("c-id-x"); + codes.Add("c-code-x"); + urls.Add($"{urlPrefix}c-code-x"); + + ids.Add("c-id-y"); + if (dupCodes && dupUrls) + { + codes.Add("c-code-x"); + urls.Add($"{urlPrefix}c-code-x"); + } + else if (dupCodes) + { + codes.Add("c-code-x"); + urls.Add($"{urlPrefix}c-code-y"); + } + else if (dupUrls) + { + codes.Add("c-code-y"); + urls.Add($"{urlPrefix}c-code-x"); + } + + var response = await CreatePersonSearchParamsAsync(); + if (!isParallel && isBatch) // sequential batch + { + Assert.Equal(2, response.Entry.Count); + + // both search params ingested. this is not correct, and need to be fixed when dedupping bug is fixed. + Assert.All(response.Entry, _ => Assert.NotNull(_.Resource as SearchParameter)); + } + else + { + Assert.Fail("This point should not be reached."); + } + } + catch (FhirClientException ex) + { + Assert.Equal(HttpStatusCode.BadRequest, ex.StatusCode); + Assert.Contains("Input search parameters have duplicate", ex.Message); + if (dupCodes) + { + Assert.Contains("codes", ex.Message); + } + + if (dupUrls) + { + Assert.Contains("Urls", ex.Message); + } + } + finally + { + await DeleteSearchParamsAsync(ids); + } + + async Task CreatePersonSearchParamsAsync() + { + var bundle = new Bundle { Type = isBatch ? Bundle.BundleType.Batch : Bundle.BundleType.Transaction, Entry = [] }; +#if R5 + var resourceTypes = new List() { VersionIndependentResourceTypesAll.Person }; +#else + var resourceTypes = new List() { ResourceType.Person }; +#endif + for (var i = 0; i < ids.Count; i++) + { + var code = codes[i]; + var id = ids[i]; + var searchParam = new SearchParameter + { + Id = id, + Url = urls[i], + Name = code, + Code = code, + Status = PublicationStatus.Active, + Type = SearchParamType.Token, + Expression = "Person.id", + Description = "any", + Base = resourceTypes, + }; + + bundle.Entry.Add(new EntryComponent { Request = new RequestComponent { Method = Bundle.HTTPVerb.PUT, Url = $"SearchParameter/{id}" }, Resource = searchParam }); + } + + var result = await _fixture.TestFhirClient.PostBundleAsync(bundle, new FhirBundleOptions { BundleProcessingLogic = isParallel ? FhirBundleProcessingLogic.Parallel : FhirBundleProcessingLogic.Sequential }); + return result; + } + } + + private async Task DeleteSearchParamsAsync(List ids) + { + foreach (var id in ids) + { + var bundle = new Bundle { Type = Bundle.BundleType.Batch, Entry = new List() }; + bundle.Entry.Add(new EntryComponent { Request = new RequestComponent { Method = Bundle.HTTPVerb.DELETE, Url = $"SearchParameter/{id}" } }); + await _fixture.TestFhirClient.PostBundleAsync(bundle, new FhirBundleOptions { BundleProcessingLogic = FhirBundleProcessingLogic.Parallel }); + } + } + [Fact] public async Task GivenReindexJobWithConcurrentUpdates_ThenReportedCountsAreLessThanOriginal() { @@ -94,7 +343,7 @@ public async Task GivenReindexJobWithConcurrentUpdates_ThenReportedCountsAreLess [Fact] public async Task GivenReindexJobWithMixedZeroAndNonZeroCountResources_WhenReindexCompletes_ThenSearchParametersShouldWork() { - var storageMultiplier = _isSql ? 1 : 50; // allows to keep settings for cosmos and optimize sql + var storageMultiplier = (_isSql || _fixture.IsUsingInProcTestServer) ? 1 : 50; // allows to keep settings for cosmos and optimize sql // Cancel any running reindex jobs before starting this test await CancelAnyRunningReindexJobsAsync(); @@ -962,7 +1211,7 @@ private async Task VerifySearchParameterIsWorkingAsync( { Exception lastException = null; - var maxRetries = _isSql ? 5 : 25; + var maxRetries = _isSql ? 1 : 25; var retryDelayMs = 1000; for (int attempt = 1; attempt <= maxRetries; attempt++) { 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 fabbe5b77f..60497a916d 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 @@ -62,8 +62,9 @@ namespace Microsoft.Health.Fhir.Tests.Integration.Features.Operations.Reindex public class ReindexJobTests : IClassFixture, IAsyncLifetime { private JobHosting _jobHosting; - private CancellationTokenSource _jobHostingCts; + private CancellationTokenSource _backgroundCts; private Task _jobHostingTask; + private Task _cacheUpdateTask; private IQueueClient _queueClient; private IJobFactory _jobFactory; @@ -95,7 +96,7 @@ public class ReindexJobTests : IClassFixture, IAsyncLif private readonly IDataStoreSearchParameterValidator _dataStoreSearchParameterValidator = Substitute.For(); private readonly IOptions _optionsReindexConfig = Substitute.For>(); private readonly IOptions _coreFeatureConfig = Substitute.For>(); - private SearchParameterCacheRefreshBackgroundService _cacheRefreshBackgroundService; + private readonly IOptions _operationsConfig = Substitute.For>(); public ReindexJobTests(FhirStorageTestsFixture fixture, ITestOutputHelper output) { @@ -106,6 +107,9 @@ public ReindexJobTests(FhirStorageTestsFixture fixture, ITestOutputHelper output public async Task InitializeAsync() { + _operationsConfig.Value.Returns(new OperationsConfiguration()); + _coreFeatureConfig.Value.Returns(new CoreFeatureConfiguration { SearchParameterCacheRefreshIntervalSeconds = 1 }); + // Initialize critical fields first before cleanup _fhirOperationDataStore = _fixture.OperationDataStore; _fhirStorageTestHelper = _fixture.TestHelper; @@ -155,15 +159,6 @@ public async Task InitializeAsync() { SearchParameterCacheRefreshIntervalSeconds = 1, }); - _cacheRefreshBackgroundService = new SearchParameterCacheRefreshBackgroundService( - _searchParameterStatusManager, - (SearchParameterOperations)_searchParameterOperations, - _coreFeatureConfig, - NullLogger.Instance); - - // Start the primary background service and trigger initialization so it begins refreshing immediately. - await _cacheRefreshBackgroundService.StartAsync(CancellationToken.None); - await _cacheRefreshBackgroundService.Handle(new SearchParametersInitializedNotification(), CancellationToken.None); _createReindexRequestHandler = new CreateReindexRequestHandler( _fhirOperationDataStore, @@ -185,27 +180,22 @@ public async Task InitializeAsync() // Initialize second FHIR service. await InitializeSecondFHIRService(); - await InitializeJobHosting(); + InitializeJobHosting(); + + StartCacheUpdateTask(_backgroundCts.Token); } public async Task DisposeAsync() { - if (_cacheRefreshBackgroundService != null) - { - await _cacheRefreshBackgroundService.StopAsync(CancellationToken.None); - } - // Clean up resources before finishing test class await DeleteTestResources(); - await StopJobHostingBackgroundServiceAsync(); + await StopBackgroundTasksAsync(); return; } -#pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously - private async Task InitializeJobHosting() -#pragma warning restore CS1998 // Async method lacks 'await' operators and will run synchronously + private void InitializeJobHosting() { // Get the actual queue client from the operation datastore implementation var operationDataStoreBase = _fhirOperationDataStore as FhirOperationDataStoreBase; @@ -248,10 +238,6 @@ private async Task InitializeJobHosting() if (typeId == (int)JobType.ReindexOrchestrator) { - // Create a mock OperationsConfiguration for the test - var operationsConfig = Substitute.For>(); - operationsConfig.Value.Returns(new OperationsConfiguration()); - job = new ReindexOrchestratorJob( _queueClient, () => _searchService, @@ -262,11 +248,11 @@ private async Task InitializeJobHosting() _fixture.FhirRuntimeConfiguration, NullLoggerFactory.Instance, _coreFeatureConfig, - operationsConfig); + _operationsConfig); } else if (typeId == (int)JobType.ReindexProcessing) { - Func> fhirDataStoreScope = () => _scopedDataStore.Value.CreateMockScope(); + Func> fhirDataStoreScope = () => _scopedDataStore.Value.CreateMockScope(); job = new ReindexProcessingJob( () => _searchService, fhirDataStoreScope, @@ -298,21 +284,21 @@ private async Task InitializeJobHosting() _jobHosting.JobHeartbeatTimeoutThresholdInSeconds = 30; _jobHosting.JobHeartbeatIntervalInSeconds = 5; - _jobHostingCts = new CancellationTokenSource(); + _backgroundCts = new CancellationTokenSource(); // Run this on a separate thread to avoid blocking the test _jobHostingTask = Task.Run(() => _jobHosting.ExecuteAsync( (byte)QueueType.Reindex, // Use the correct queue type - runningJobCount: 5, + runningJobCount: 2, workerName: "ReindexTestWorker", - cancellationTokenSource: _jobHostingCts)); + cancellationTokenSource: _backgroundCts)); } - private async Task StopJobHostingBackgroundServiceAsync() + private async Task StopBackgroundTasksAsync() { - if (_jobHostingCts != null) + if (_backgroundCts != null) { - _jobHostingCts.Cancel(); + _backgroundCts.Cancel(); if (_jobHostingTask != null) { @@ -326,25 +312,43 @@ private async Task StopJobHostingBackgroundServiceAsync() } } - _jobHostingCts.Dispose(); - _jobHostingCts = null; + if (_cacheUpdateTask != null) + { + try + { + await _cacheUpdateTask.ConfigureAwait(false); + } + catch (OperationCanceledException) + { + // Expected when cancellation occurs + } + } + + _backgroundCts.Dispose(); + _backgroundCts = null; } } private void StartCacheUpdateTask(CancellationToken cancellationToken) { - var task = new Task( - _ => + _cacheUpdateTask = new Task( + async _ => { while (!cancellationToken.IsCancellationRequested) { - var allSearchParameterStatus = _searchParameterStatusManager.GetAllSearchParameterStatus(cancellationToken).Result; - _searchParameterStatusManager.ApplySearchParameterStatus(allSearchParameterStatus, cancellationToken).Wait(); - Thread.Sleep(_coreFeatureConfig.Value.SearchParameterCacheRefreshIntervalSeconds * 1000); + try + { + await _searchParameterOperations.GetAndApplySearchParameterUpdates(cancellationToken); + } + catch (Exception) + { + } + + Thread.Sleep(TimeSpan.FromSeconds(_coreFeatureConfig.Value.SearchParameterCacheRefreshIntervalSeconds)); } }, cancellationToken); - task.Start(); + _cacheUpdateTask.Start(); } [Fact] @@ -482,7 +486,7 @@ public async Task GivenSearchParametersToReindex_ThenReindexJobShouldComplete() try { - var reindexJobWorker = await PerformReindexingOperation(response, OperationStatus.Completed, cancellationTokenSource); + var reindexJobWorker = await WaitForReindexCompletionAsync(response, cancellationTokenSource); Assert.True(reindexJobWorker.JobRecord.ResourceCounts.Count > 0); Assert.True(reindexJobWorker.JobRecord.Progress > 0); @@ -532,7 +536,7 @@ public async Task GivenNoSupportedSearchParameters_WhenRunningReindexJob_ThenJob try { - await PerformReindexingOperation(response, OperationStatus.Completed, cancellationTokenSource); + await WaitForReindexCompletionAsync(response, cancellationTokenSource); } finally { @@ -557,7 +561,7 @@ public async Task GivenNoMatchingResources_WhenRunningReindexJob_ThenJobIsComple try { - await PerformReindexingOperation(response, OperationStatus.Completed, cancellationTokenSource); + await WaitForReindexCompletionAsync(response, cancellationTokenSource); var updateSearchParamList = await _searchParameterStatusManager.GetAllSearchParameterStatus(default); Assert.Equal(SearchParameterStatus.Enabled, updateSearchParamList.Where(sp => sp.Uri.OriginalString == searchParam.Url.OriginalString).First().Status); @@ -606,7 +610,7 @@ public async Task GivenNewSearchParamCreatedBeforeResourcesToBeIndexed_WhenReind try { - await PerformReindexingOperation(response, OperationStatus.Completed, cancellationTokenSource); + await WaitForReindexCompletionAsync(response, cancellationTokenSource); // Rerun the same search as above searchResults = await _searchService.Value.SearchAsync("Patient", queryParams, CancellationToken.None); @@ -754,7 +758,7 @@ public async Task GivenNewSearchParamCreatedAfterResourcesToBeIndexed_WhenReinde try { - await PerformReindexingOperation(response, OperationStatus.Completed, cancellationTokenSource); + await WaitForReindexCompletionAsync(response, cancellationTokenSource); // Rerun the same search as above searchResults = await _searchService.Value.SearchAsync("Patient", queryParams, CancellationToken.None); @@ -822,7 +826,7 @@ public async Task GivenSecondFHIRServiceSynced_WhenReindexJobCompleted_ThenSecon try { - await PerformReindexingOperation(response, OperationStatus.Completed, cancellationTokenSource); + await WaitForReindexCompletionAsync(response, cancellationTokenSource); ResourceSearchParameterStatus syncedStatus = null; bool hasPrimaryDefinition = false; @@ -965,11 +969,7 @@ public async Task GivenNewSearchParamWithResourceBaseType_WhenReindexJobComplete try { - await PerformReindexingOperation(response, OperationStatus.Completed, cancellationTokenSource); - - // CRITICAL: Force the search parameter definition manager to refresh/sync - // This is the missing piece - the search service needs to know about status changes - await _searchParameterOperations.GetAndApplySearchParameterUpdates(CancellationToken.None, true); + await WaitForReindexCompletionAsync(response, cancellationTokenSource); // Now test the actual search functionality // Rerun the same search as above @@ -1105,14 +1105,6 @@ public async Task GivenFailedProcessingJobs_WhenOrchestratorProcessesResults_The await SeedProcessingJobAsync(orchestratorGroupId, supplyDeliveryJobDefinition, supplyJob1Result, JobStatus.Failed, null); await SeedProcessingJobAsync(orchestratorGroupId, supplyDeliveryJobDefinition, supplyJob2Result, JobStatus.Failed, null); - // Initialize the OperationsConfiguration for the orchestrator - var operationsConfig = Substitute.For>(); - operationsConfig.Value.Returns(new OperationsConfiguration()); - _coreFeatureConfig.Value.Returns(new CoreFeatureConfiguration - { - SearchParameterCacheRefreshIntervalSeconds = 1, - }); - // Create orchestrator and execute var runtimeConfiguration = Substitute.For(); runtimeConfiguration.IsSurrogateIdRangingSupported.Returns(false); @@ -1127,7 +1119,7 @@ public async Task GivenFailedProcessingJobs_WhenOrchestratorProcessesResults_The runtimeConfiguration, NullLoggerFactory.Instance, _coreFeatureConfig, - operationsConfig); + _operationsConfig); // Execute the orchestrator - it will load all processing jobs and extract errors var orchestratorResult = await orchestrator.ExecuteAsync(orchestratorJobInfo, CancellationToken.None); @@ -1271,7 +1263,7 @@ public async Task GivenSurrogateRangeFetchOom_WhenProcessingJobRuns_ThenSplitUse Definition = JsonConvert.SerializeObject(jobDefinition), }; - Func> dataStoreScope = () => _scopedDataStore.Value.CreateMockScope(); + Func> dataStoreScope = () => _scopedDataStore.Value.CreateMockScope(); var processingJob = new ReindexProcessingJob( () => _searchService, dataStoreScope, @@ -1300,9 +1292,7 @@ await mockedSearchService.Received(1).GetSurrogateIdRanges( private async Task SeedOrchestratorJobAsync(ReindexJobRecord jobRecord) { var definition = JsonConvert.SerializeObject(jobRecord); - - var enqueued = await _queueClient.EnqueueAsync((byte)QueueType.Reindex, new[] { definition }, null, false, CancellationToken.None); - return enqueued.Single(); + return await _queueClient.EnqueueWithStatusAsync((byte)QueueType.Reindex, null, definition, JobStatus.Running, null, null, CancellationToken.None); } private async Task SeedProcessingJobAsync(long groupId, ReindexProcessingJobDefinition jobDefinition, ReindexProcessingJobResult jobResult, JobStatus status, int? data) @@ -1320,78 +1310,36 @@ private async Task SeedProcessingJobAsync(long groupId, ReindexProcessingJobDefi }) : JsonConvert.SerializeObject(jobResult); - var enqueued = await _queueClient.EnqueueAsync((byte)QueueType.Reindex, new[] { definition }, groupId, false, CancellationToken.None); - var processingJob = enqueued.Single(); - processingJob.Status = status; - processingJob.Data = data; - processingJob.Result = result; - await _queueClient.CompleteJobAsync(processingJob, false, CancellationToken.None); + var job = await _queueClient.EnqueueWithStatusAsync((byte)QueueType.Reindex, groupId, definition, JobStatus.Running, null, null, CancellationToken.None); + job.Status = status; + job.Data = data; + job.Result = result; + await _queueClient.CompleteJobAsync(job, false, CancellationToken.None); } - private async Task PerformReindexingOperation( - CreateReindexResponse response, - OperationStatus operationStatus, - CancellationTokenSource cancellationTokenSource, - int delay = 1000) + private async Task WaitForReindexCompletionAsync(CreateReindexResponse response, CancellationTokenSource cancellationTokenSource) { - StartCacheUpdateTask(cancellationTokenSource.Token); - - const int MaxNumberOfAttempts = 120; - - ReindexJobWrapper reindexJobWrapper = await _fhirOperationDataStore.GetReindexJobByIdAsync(response.Job.JobRecord.Id, cancellationTokenSource.Token); - - int delayCount = 0; - - Stopwatch stopwatch = Stopwatch.StartNew(); - while (reindexJobWrapper.JobRecord.Status != operationStatus && delayCount < MaxNumberOfAttempts) + var stopwatch = Stopwatch.StartNew(); + ReindexJobWrapper orchestrator = null; + while (stopwatch.Elapsed.TotalSeconds < 120) { - // Check for any processing jobs that need to be executed - var allJobs = await _queueClient.GetJobByGroupIdAsync((byte)QueueType.Reindex, response.Job.JobRecord.GroupId, true, cancellationTokenSource.Token); - var processingJobs = allJobs.Where(j => j.Status == JobStatus.Created || j.Status == JobStatus.Running).ToList(); - - // Execute any pending processing jobs - foreach (var processingJob in processingJobs.Where(j => j.Status == JobStatus.Created)) + orchestrator = await _fhirOperationDataStore.GetReindexJobByIdAsync(response.Job.JobRecord.Id, cancellationTokenSource.Token); + if (orchestrator.JobRecord.Status == OperationStatus.Failed + || orchestrator.JobRecord.Status == OperationStatus.Canceled + || orchestrator.JobRecord.Status == OperationStatus.Completed) { - try - { - var scopedJob = _jobFactory.Create(processingJob); - var result = await scopedJob.Value.ExecuteAsync(processingJob, cancellationTokenSource.Token); - processingJob.Status = JobStatus.Completed; - processingJob.Result = result; - await _queueClient.CompleteJobAsync(processingJob, false, cancellationTokenSource.Token); - } - catch (Exception ex) - { - processingJob.Status = JobStatus.Failed; - processingJob.Result = JsonConvert.SerializeObject(new { Error = ex.Message }); - await _queueClient.CompleteJobAsync(processingJob, false, cancellationTokenSource.Token); - } + break; } - await Task.Delay(delay); - delayCount++; - reindexJobWrapper = await _fhirOperationDataStore.GetReindexJobByIdAsync(response.Job.JobRecord.Id, cancellationTokenSource.Token); - - if (operationStatus == OperationStatus.Completed && - (reindexJobWrapper.JobRecord.Status == OperationStatus.Failed || reindexJobWrapper.JobRecord.Status == OperationStatus.Canceled)) - { - var failureReason = reindexJobWrapper.JobRecord.FailureDetails?.FailureReason; - Assert.Fail($"Fail-fast. Current job status '{reindexJobWrapper.JobRecord.Status}'. Expected job status '{operationStatus}'. Number of attempts: {MaxNumberOfAttempts}. Time elapsed: {stopwatch.Elapsed}. FailureDetails: '{failureReason}'."); - } + await Task.Delay(1000); } - // If we maxed out attempts and did not reach the expected status, clean up any active jobs - if (reindexJobWrapper.JobRecord.Status != operationStatus) - { - Assert.True( - operationStatus == reindexJobWrapper.JobRecord.Status, - $"Current job status '{reindexJobWrapper.JobRecord.Status}'. Expected job status '{operationStatus}'. Number of attempts: {delayCount}. Time elapsed: {stopwatch.Elapsed}."); - } + Assert.Equal(OperationStatus.Completed, orchestrator.JobRecord.Status); var serializer = new FhirJsonSerializer(); - _output.WriteLine(serializer.SerializeToString(reindexJobWrapper.ToParametersResourceElement().ToPoco())); + _output.WriteLine(serializer.SerializeToString(orchestrator.ToParametersResourceElement().ToPoco())); - return reindexJobWrapper; + return orchestrator; } private async Task SetUpForReindexing(CreateReindexRequest request = null) diff --git a/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/FhirStorageTests.cs b/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/FhirStorageTests.cs index b13b681f63..6f71076443 100644 --- a/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/FhirStorageTests.cs +++ b/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/FhirStorageTests.cs @@ -17,6 +17,7 @@ using Microsoft.Data.SqlClient; using Microsoft.Health.Abstractions.Exceptions; using Microsoft.Health.Abstractions.Features.Transactions; +using Microsoft.Health.Extensions.DependencyInjection; using Microsoft.Health.Extensions.Xunit; using Microsoft.Health.Fhir.Core; using Microsoft.Health.Fhir.Core.Exceptions; @@ -1205,6 +1206,8 @@ private async Task CreatePatientSearchParam(string searchParamN // Add the search parameter to the datastore await _fixture.SearchParameterStatusManager.UpdateSearchParameterStatusAsync(new List { searchParam.Url }, SearchParameterStatus.Supported, CancellationToken.None); + await _fixture.SearchParameterOperations.GetAndApplySearchParameterUpdates(); + return searchParam; } 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 27db98b04b..c83fa58190 100644 --- a/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/FhirStorageTestsFixture.cs +++ b/test/Microsoft.Health.Fhir.Shared.Tests.Integration/Persistence/FhirStorageTestsFixture.cs @@ -22,6 +22,7 @@ using Microsoft.Extensions.Options; using Microsoft.Health.Abstractions.Features.Transactions; using Microsoft.Health.Core.Features.Context; +using Microsoft.Health.Extensions.DependencyInjection; using Microsoft.Health.Fhir.Api.Features.Bundle; using Microsoft.Health.Fhir.Api.Features.Routing; using Microsoft.Health.Fhir.Core.Configs; @@ -69,6 +70,7 @@ public class FhirStorageTestsFixture : IAsyncLifetime, IDisposable private readonly ResourceIdProvider _resourceIdProvider; private readonly DataResourceFilter _dataResourceFilter; private readonly IFhirRuntimeConfiguration _fhirRuntimeConfiguration; + private SearchParameterOperations _searchParameterOperations; public FhirStorageTestsFixture(DataStore dataStore) : this(dataStore switch @@ -133,6 +135,8 @@ internal FhirStorageTestsFixture(IServiceProvider fixture) public SearchParameterDefinitionManager SearchParameterDefinitionManager => _fixture.GetRequiredService(); + public SearchParameterOperations SearchParameterOperations => _searchParameterOperations; + public SupportedSearchParameterDefinitionManager SupportedSearchParameterDefinitionManager => _fixture.GetRequiredService(); public SchemaInitializer SchemaInitializer => _fixture.GetRequiredService(); @@ -242,6 +246,16 @@ public async Task InitializeAsync() ServiceProvider services = collection.BuildServiceProvider(); + _searchParameterOperations = new SearchParameterOperations( + SearchParameterStatusManager, + SearchParameterDefinitionManager, + ModelInfoProvider.Instance, + Substitute.For(), + Substitute.For(), + () => Substitute.For>(), + () => Substitute.For>(), + NullLogger.Instance); + Mediator = new Mediator(services); }