fix: improve reindex reliability with distributed locking and observability

.gitignore

@@ -60,3 +60,4 @@ _NCrunch_*
 **/.idea/**/contentModel.xml
 **/.idea/**/modules.xml
 **/.idea/copilot/chatSessions/
+.idea/

docs/guide/index-management.md

@@ -851,3 +851,30 @@ AddReindexScript(2, @"
 - [Migrations](/guide/migrations) - Document migrations
 - [Jobs](/guide/jobs) - Index maintenance jobs
 - [Elasticsearch Setup](/guide/elasticsearch-setup) - Connection configuration
+
+## Concurrency Safety
+
+Reindexing is protected by a distributed lock keyed on the index alias to prevent concurrent reindex operations from corrupting data.
+
+### Lock Strategy
+
+- **Lock key**: `reindex:{alias}` (e.g., `reindex:employees`)
+- **Lock TTL**: 20 minutes, auto-renewed during long-running operations
+- Both direct (`VersionedIndex.ReindexAsync`) and work-item (`ReindexWorkItemHandler`) paths use the same lock
+- Only one reindex per logical index can run at a time — subsequent version transitions wait for the current one to complete
+
+### Why Alias-Only Keys
+
+Using the alias as the lock key ensures that sequential version transitions (v1→v2, then v2→v3) cannot overlap. If v2→v3 started before v1→v2 completed, v3 would contain incomplete data from v2.
+
+### Lock Renewal for Long-Running Reindexes
+
+For indexes with millions of documents that take hours to reindex, the lock is automatically renewed on every progress callback (every 1-10 seconds during the polling loop). This prevents lock expiration during legitimate long-running operations.
+
+### Crash Recovery
+
+If an instance crashes mid-reindex, the lock expires after 20 minutes. Another instance can then retry the reindex. `VersionedIndex.ReindexAsync()` is resume-safe — it picks up from the last document using timestamp-based or ID-based range queries.
+
+### Unique Index Names
+
+`ElasticConfiguration.AddIndex()` enforces unique index names (case-insensitive). Registering two indexes with the same alias throws an `ArgumentException` at startup, preventing conflicts before they can cause data corruption.

samples/Foundatio.SampleApp/Server/Foundatio.SampleApp.Server.csproj

@@ -7,7 +7,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Foundatio.Extensions.Hosting" Version="13.0.1-preview.0.11" />
+    <PackageReference Include="Foundatio.Extensions.Hosting" Version="13.0.1-preview.0.12" />
     <PackageReference Include="Microsoft.AspNetCore.Components.WebAssembly.Server" Version="10.0.7" />
   </ItemGroup>
 

src/Foundatio.Repositories.Elasticsearch/Configuration/ElasticConfiguration.cs

@@ -13,6 +13,7 @@
 using Foundatio.Parsers.ElasticQueries;
 using Foundatio.Queues;
 using Foundatio.Repositories.Elasticsearch.CustomFields;
+using Foundatio.Repositories.Elasticsearch.Jobs;
 using Foundatio.Repositories.Elasticsearch.Queries.Builders;
 using Foundatio.Repositories.Extensions;
 using Foundatio.Resilience;
@@ -88,6 +89,7 @@ protected virtual void ConfigureSettings(ConnectionSettings settings)
     public IElasticClient Client => _client.Value;
     public ICacheClient Cache { get; }
     public IMessageBus MessageBus { get; }
+    public ILockProvider LockProvider => _lockProvider;
     public ILoggerFactory LoggerFactory { get; }
     public IResiliencePolicyProvider ResiliencePolicyProvider { get; }
     public IResiliencePolicy ResiliencePolicy { get; }
@@ -116,6 +118,9 @@ public void AddIndex(IIndex index)
         if (_frozenIndexes.IsValueCreated)
             throw new InvalidOperationException("Can't add indexes after the list has been frozen.");
 
+        if (_indexes.Any(i => String.Equals(i.Name, index.Name, StringComparison.OrdinalIgnoreCase)))
+            throw new ArgumentException($"An index with name '{index.Name}' has already been registered.", nameof(index));
+
         _indexes.Add(index);
     }
 
@@ -187,8 +192,7 @@ private async Task ConfigureIndexInternalAsync(IIndex idx, bool beginReindexingO
             throw new InvalidOperationException("Must specify work item queue and lock provider in order to migrate index versions.");
 
         var reindexWorkItem = versionedIndex.CreateReindexWorkItem(currentVersion);
-        bool isReindexing = await _lockProvider.IsLockedAsync(String.Join(":", "reindex", reindexWorkItem.Alias,
-            reindexWorkItem.OldIndex, reindexWorkItem.NewIndex)).AnyContext();
+        bool isReindexing = await _lockProvider.IsLockedAsync(ElasticReindexer.GetLockName(versionedIndex.Name)).AnyContext();
         if (isReindexing)
             return;
 

src/Foundatio.Repositories.Elasticsearch/Configuration/IElasticConfiguration.cs

@@ -2,6 +2,7 @@
 using System.Collections.Generic;
 using System.Threading.Tasks;
 using Foundatio.Caching;
+using Foundatio.Lock;
 using Foundatio.Messaging;
 using Foundatio.Parsers.ElasticQueries;
 using Foundatio.Repositories.Elasticsearch.CustomFields;
@@ -17,6 +18,7 @@ public interface IElasticConfiguration : IDisposable
     IElasticClient Client { get; }
     ICacheClient Cache { get; }
     IMessageBus MessageBus { get; }
+    ILockProvider LockProvider { get; }
     ILoggerFactory LoggerFactory { get; }
     IResiliencePolicyProvider ResiliencePolicyProvider { get; }
     TimeProvider TimeProvider { get; set; }

src/Foundatio.Repositories.Elasticsearch/Configuration/Index.cs

@@ -351,6 +351,7 @@ public virtual Task ReindexAsync(Func<int, string?, Task>? progressCallbackAsync
         {
             OldIndex = Name,
             NewIndex = Name,
+            Alias = Name,
             DeleteOld = false,
             TimestampField = GetTimeStampField()
         };

src/Foundatio.Repositories.Elasticsearch/Configuration/VersionedIndex.cs

@@ -5,6 +5,7 @@
 using System.Linq.Expressions;
 using System.Text;
 using System.Threading.Tasks;
+using Foundatio.Lock;
 using Foundatio.Parsers.ElasticQueries;
 using Foundatio.Parsers.ElasticQueries.Extensions;
 using Foundatio.Repositories.Elasticsearch.Extensions;
@@ -250,9 +251,33 @@ public override async Task ReindexAsync(Func<int, string?, Task>? progressCallba
         if (currentVersion < 0 || currentVersion >= Version)
             return;
 
+        string lockKey = ElasticReindexer.GetLockName(Name);
+        await using var reindexLock = await Configuration.LockProvider.AcquireAsync(lockKey, TimeSpan.FromMinutes(20), TimeSpan.FromMinutes(30)).AnyContext();
+        if (reindexLock is null)
+            throw new InvalidOperationException($"Unable to acquire reindex lock for '{Name}' after 30 minutes.");
+
+        currentVersion = await GetCurrentVersionAsync().AnyContext();
+        if (currentVersion < 0 || currentVersion >= Version)
+            return;
+
         var reindexWorkItem = CreateReindexWorkItem(currentVersion);
+
+        Func<int, string?, Task> wrappedCallback = async (progress, message) =>
+        {
+            await reindexLock.RenewAsync().AnyContext();
+
+            if (progressCallbackAsync is not null)
+            {
+                await progressCallbackAsync(progress, message).AnyContext();
+            }
+            else
+            {
+                _logger.LogInformation("Reindex Progress {Progress:F1}%: {Message}", progress, message);
+            }
+        };
+
         var reindexer = new ElasticReindexer(Configuration.Client, _logger);
-        await reindexer.ReindexAsync(reindexWorkItem, progressCallbackAsync).AnyContext();
+        await reindexer.ReindexAsync(reindexWorkItem, wrappedCallback).AnyContext();
     }
 
     public override async Task MaintainAsync(bool includeOptionalTasks = true)

src/Foundatio.Repositories.Elasticsearch/Jobs/ReindexWorkItem.cs

@@ -6,7 +6,7 @@ public record ReindexWorkItem
 {
     public required string OldIndex { get; init; }
     public required string NewIndex { get; init; }
-    public string? Alias { get; init; }
+    public required string Alias { get; init; }
     public string? Script { get; init; }
     public bool DeleteOld { get; set; }
     public string? TimestampField { get; init; }

src/Foundatio.Repositories.Elasticsearch/Jobs/ReindexWorkItemHandler.cs

@@ -25,7 +25,7 @@ public ReindexWorkItemHandler(IElasticClient client, ILockProvider lockProvider,
         if (workItem is not ReindexWorkItem reindexWorkItem)
             return Task.FromResult<ILock?>(null);
 
-        return _lockProvider.TryAcquireAsync(String.Join(":", "reindex", reindexWorkItem.Alias, reindexWorkItem.OldIndex, reindexWorkItem.NewIndex), TimeSpan.FromMinutes(20), cancellationToken);
+        return _lockProvider.TryAcquireAsync(ElasticReindexer.GetLockName(reindexWorkItem.Alias), TimeSpan.FromMinutes(20), cancellationToken);
     }
 
     public override Task HandleItemAsync(WorkItemContext context)

src/Foundatio.Repositories.Elasticsearch/Repositories/ElasticReindexer.cs

@@ -5,6 +5,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using Elasticsearch.Net;
+using Foundatio.Parsers.ElasticQueries.Extensions;
 using Foundatio.Repositories.Elasticsearch.Extensions;
 using Foundatio.Repositories.Elasticsearch.Jobs;
 using Foundatio.Repositories.Extensions;
@@ -27,6 +28,16 @@ public class ElasticReindexer
     private const string ID_FIELD = "id";
     private const int MAX_STATUS_FAILS = 10;
 
+    /// <summary>
+    /// Returns the distributed lock resource name for serializing reindex operations on the given alias.
+    /// </summary>
+    public static string GetLockName(string alias)
+    {
+        ArgumentException.ThrowIfNullOrEmpty(alias);
+
+        return String.Concat("reindex:", alias);
+    }
+
     public ElasticReindexer(IElasticClient client, ILogger? logger = null) : this(client, TimeProvider.System, logger)
     {
     }
@@ -62,7 +73,14 @@ public async Task ReindexAsync(ReindexWorkItem workItem, Func<int, string?, Task
             };
         }
 
-        _logger.LogInformation("Received reindex work item for new index: {NewIndex}", workItem.NewIndex);
+        using var _ = _logger.BeginScope(new Dictionary<string, object>
+        {
+            [nameof(workItem.OldIndex)] = workItem.OldIndex,
+            [nameof(workItem.NewIndex)] = workItem.NewIndex,
+            [nameof(workItem.Alias)] = workItem.Alias
+        });
+
+        _logger.LogInformation("Received reindex work item for {OldIndex} -> {NewIndex}", workItem.OldIndex, workItem.NewIndex);
         var startTime = _timeProvider.GetUtcNow().UtcDateTime.AddSeconds(-1);
         await progressCallbackAsync(0, "Starting reindex...").AnyContext();
         var firstPassResult = await InternalReindexAsync(workItem, progressCallbackAsync, 0, 90, workItem.StartUtc).AnyContext();
@@ -96,7 +114,7 @@ public async Task ReindexAsync(ReindexWorkItem workItem, Func<int, string?, Task
         var refreshResponse = await _client.Indices.RefreshAsync(Indices.All).AnyContext();
         _logger.LogRequest(refreshResponse);
         if (!refreshResponse.IsValid)
-            _logger.LogWarning("Failed to refresh indices before second reindex pass: {Error}", refreshResponse.ServerError);
+            _logger.LogWarning("Failed to refresh indices before second reindex pass for {OldIndex} -> {NewIndex}: {Error}", workItem.OldIndex, workItem.NewIndex, refreshResponse.GetErrorMessage());
 
         ReindexResult? secondPassResult = null;
         if (!String.IsNullOrEmpty(workItem.TimestampField))
@@ -117,34 +135,34 @@ public async Task ReindexAsync(ReindexWorkItem workItem, Func<int, string?, Task
             refreshResponse = await _client.Indices.RefreshAsync(Indices.All).AnyContext();
             _logger.LogRequest(refreshResponse);
             if (!refreshResponse.IsValid)
-                _logger.LogWarning("Failed to refresh indices before doc count comparison: {Error}", refreshResponse.ServerError);
+                _logger.LogWarning("Failed to refresh indices before doc count comparison for {OldIndex} -> {NewIndex}: {Error}", workItem.OldIndex, workItem.NewIndex, refreshResponse.GetErrorMessage());
 
             var newDocCountResponse = await _client.CountAsync<object>(d => d.Index(workItem.NewIndex)).AnyContext();
             _logger.LogRequest(newDocCountResponse);
             if (!newDocCountResponse.IsValid)
-                _logger.LogWarning("Failed to get new index doc count: {Error}", newDocCountResponse.ServerError);
+                _logger.LogWarning("Failed to get new index doc count for {NewIndex}: {Error}", workItem.NewIndex, newDocCountResponse.GetErrorMessage());
 
             var oldDocCountResponse = await _client.CountAsync<object>(d => d.Index(workItem.OldIndex)).AnyContext();
             _logger.LogRequest(oldDocCountResponse);
             if (!oldDocCountResponse.IsValid)
-                _logger.LogWarning("Failed to get old index doc count: {Error}", oldDocCountResponse.ServerError);
+                _logger.LogWarning("Failed to get old index doc count for {OldIndex}: {Error}", workItem.OldIndex, oldDocCountResponse.GetErrorMessage());
 
             await progressCallbackAsync(98, $"Old Docs: {oldDocCountResponse.Count} New Docs: {newDocCountResponse.Count}").AnyContext();
             if (newDocCountResponse.IsValid && oldDocCountResponse.IsValid && newDocCountResponse.Count >= oldDocCountResponse.Count)
             {
                 var deleteIndexResponse = await _client.Indices.DeleteAsync(Indices.Index(workItem.OldIndex)).AnyContext();
                 _logger.LogRequest(deleteIndexResponse);
                 if (!deleteIndexResponse.IsValid)
-                    _logger.LogWarning("Failed to delete old index {OldIndex}: {Error}", workItem.OldIndex, deleteIndexResponse.ServerError);
+                    _logger.LogWarning("Failed to delete old index {OldIndex}: {Error}", workItem.OldIndex, deleteIndexResponse.GetErrorMessage());
 
                 if (deleteIndexResponse.IsValid)
                     await progressCallbackAsync(99, $"Deleted index: {workItem.OldIndex}").AnyContext();
                 else
-                    await progressCallbackAsync(99, $"Failed to delete old index {workItem.OldIndex}: {deleteIndexResponse.ServerError}").AnyContext();
+                    await progressCallbackAsync(99, $"Failed to delete old index {workItem.OldIndex}: {deleteIndexResponse.GetErrorMessage()}").AnyContext();
             }
         }
 
-        await progressCallbackAsync(100, null).AnyContext();
+        await progressCallbackAsync(100, "Reindex complete").AnyContext();
     }
 
     private async Task<ReindexResult> InternalReindexAsync(ReindexWorkItem workItem, Func<int, string?, Task> progressCallbackAsync, int startProgress = 0, int endProgress = 100, DateTime? startTime = null, CancellationToken cancellationToken = default)
@@ -173,7 +191,7 @@ private async Task<ReindexResult> InternalReindexAsync(ReindexWorkItem workItem,
             return response;
         }, cancellationToken).AnyContext();
 
-        _logger.LogInformation("Reindex Task Id: {TaskId}", result.Task.FullyQualifiedId);
+        _logger.LogInformation("Reindex Task Id: {ReindexTaskId}", result.Task.FullyQualifiedId);
         _logger.LogRequest(result);
         long totalDocs = result.Total;
 
@@ -196,7 +214,8 @@ private async Task<ReindexResult> InternalReindexAsync(ReindexWorkItem workItem,
 
                 if (statusGetFails > MAX_STATUS_FAILS)
                 {
-                    _logger.LogError("Failed to get the status {FailureCount} times in a row", MAX_STATUS_FAILS);
+                    _logger.LogError("Failed to get the status {FailureCount} times in a row for reindex task {ReindexTaskId} reindexing {OldIndex} -> {NewIndex}",
+                        statusGetFails, result.Task.FullyQualifiedId, workItem.OldIndex, workItem.NewIndex);
                     break;
                 }
 
@@ -221,7 +240,7 @@ private async Task<ReindexResult> InternalReindexAsync(ReindexWorkItem workItem,
                 sw.Restart();
             lastProgress = lastCompleted;
 
-            string lastMessage = $"Total: {status.Task.Status.Total:N0} Completed: {lastCompleted:N0} VersionConflicts: {status.Task.Status.VersionConflicts:N0}";
+            string lastMessage = $"[{workItem.NewIndex}] Total: {status.Task.Status.Total:N0} Completed: {lastCompleted:N0} VersionConflicts: {status.Task.Status.VersionConflicts:N0}";
             await progressCallbackAsync(CalculateProgress(status.Task.Status.Total, lastCompleted, startProgress, endProgress), lastMessage).AnyContext();
 
             if (status.Completed && response?.Error == null)
@@ -245,6 +264,22 @@ private async Task<ReindexResult> InternalReindexAsync(ReindexWorkItem workItem,
         } while (!cancellationToken.IsCancellationRequested);
         sw.Stop();
 
+        if (!taskSuccess)
+        {
+            if (cancellationToken.IsCancellationRequested)
+            {
+                _logger.LogWarning("Reindex cancelled for {OldIndex} -> {NewIndex}. ReindexTaskId: {ReindexTaskId}, LastProgress: {LastProgress}, TotalDocs: {TotalDocs}, Elapsed: {Elapsed}",
+                    workItem.OldIndex, workItem.NewIndex, result.Task.FullyQualifiedId, lastProgress, totalDocs, sw.Elapsed);
+            }
+            else
+            {
+                _logger.LogError("Reindex abandoned for {OldIndex} -> {NewIndex}. ReindexTaskId: {ReindexTaskId}, StatusFails: {StatusFails}, LastProgress: {LastProgress}, TotalDocs: {TotalDocs}, Elapsed: {Elapsed}",
+                    workItem.OldIndex, workItem.NewIndex, result.Task.FullyQualifiedId, statusGetFails, lastProgress, totalDocs, sw.Elapsed);
+            }
+
+            await TryCancelTaskAsync(result.Task, workItem.OldIndex, workItem.NewIndex).AnyContext();
+        }
+
         long failures = 0;
         if (lastReindexResponse?.Failures != null && lastReindexResponse.Failures.Count > 0)
         {
@@ -318,6 +353,31 @@ private async Task HandleFailureAsync(ReindexWorkItem workItem, BulkIndexByScrol
             _logger.LogErrorRequest(indexResponse, "Error indexing document {Index}/{Id}", workItem.NewIndex + "-error", gr.Id);
     }
 
+    /// <summary>
+    /// Attempts to cancel the Elasticsearch server-side reindex task. Best-effort — failures are logged but not propagated.
+    /// </summary>
+    private async Task TryCancelTaskAsync(TaskId reindexTaskId, string oldIndex, string newIndex)
+    {
+        try
+        {
+            using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+            var response = await _client.Tasks.CancelAsync(c => c.TaskId(reindexTaskId), cts.Token).AnyContext();
+            if (response.IsValid)
+            {
+                _logger.LogRequest(response);
+                _logger.LogInformation("Cancelled reindex task {ReindexTaskId} for {OldIndex} -> {NewIndex}", reindexTaskId.FullyQualifiedId, oldIndex, newIndex);
+            }
+            else
+            {
+                _logger.LogErrorRequest(response, "Failed to cancel reindex task {ReindexTaskId} for {OldIndex} -> {NewIndex}", reindexTaskId.FullyQualifiedId, oldIndex, newIndex);
+            }
+        }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, "Exception cancelling reindex task {ReindexTaskId} for {OldIndex} -> {NewIndex}", reindexTaskId.FullyQualifiedId, oldIndex, newIndex);
+        }
+    }
+
     private async Task<List<string>> GetIndexAliasesAsync(string index)
     {
         var aliasesResponse = await _client.Indices.GetAliasAsync(index).AnyContext();

src/Foundatio.Repositories/Foundatio.Repositories.csproj

@@ -1,9 +1,9 @@
 <Project Sdk="Microsoft.NET.Sdk">
   <ItemGroup>
-    <PackageReference Include="Foundatio" Version="13.0.1-preview.0.11" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
+    <PackageReference Include="Foundatio" Version="13.0.1-preview.0.12" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
     <ProjectReference Include="$(FoundatioProjectsDir)Foundatio\src\Foundatio\Foundatio.csproj" Condition="'$(ReferenceFoundatioSource)' == 'true'" />
 
-    <PackageReference Include="Foundatio.JsonNet" Version="13.0.1-preview.0.11" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
+    <PackageReference Include="Foundatio.JsonNet" Version="13.0.1-preview.0.12" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
     <ProjectReference Include="$(FoundatioProjectsDir)Foundatio\src\Foundatio.JsonNet\Foundatio.JsonNet.csproj" Condition="'$(ReferenceFoundatioSource)' == 'true'" />
   </ItemGroup>
 </Project>