diff --git a/app/MindWork AI Studio/Assistants/I18N/allTexts.lua b/app/MindWork AI Studio/Assistants/I18N/allTexts.lua index 065f0acf..aa9acb00 100644 --- a/app/MindWork AI Studio/Assistants/I18N/allTexts.lua +++ b/app/MindWork AI Studio/Assistants/I18N/allTexts.lua @@ -9562,6 +9562,9 @@ UI_TEXT_CONTENT["AISTUDIO::PAGES::EMBEDDINGS::T1455637941"] = "Could not open th -- Failed: {0} UI_TEXT_CONTENT["AISTUDIO::PAGES::EMBEDDINGS::T1531389696"] = "Failed: {0}" +-- AI Studio indexes large data sources in steps of a few minutes, so that your other data sources get their turn in between. +UI_TEXT_CONTENT["AISTUDIO::PAGES::EMBEDDINGS::T1545941101"] = "AI Studio indexes large data sources in steps of a few minutes, so that your other data sources get their turn in between." + -- Open the settings UI_TEXT_CONTENT["AISTUDIO::PAGES::EMBEDDINGS::T1582896271"] = "Open the settings" diff --git a/app/MindWork AI Studio/Pages/Embeddings.razor b/app/MindWork AI Studio/Pages/Embeddings.razor index 0d365e86..33d60912 100644 --- a/app/MindWork AI Studio/Pages/Embeddings.razor +++ b/app/MindWork AI Studio/Pages/Embeddings.razor @@ -14,6 +14,9 @@ @T("AI Studio indexes local RAG data sources in the background. Finished files stay recorded so unchanged files can be skipped after a restart, while added or deleted files are detected during the next run. The same applies to documents without readable text, such as scanned pages: AI Studio remembers them and reads them again only once they change.") + + @T("AI Studio indexes large data sources in steps of a few minutes, so that your other data sources get their turn in between.") + @if (this.AreMailboxesEnabled) { diff --git a/app/MindWork AI Studio/Plugins/languages/de-de-43065dbc-78d0-45b7-92be-f14c2926e2dc/plugin.lua b/app/MindWork AI Studio/Plugins/languages/de-de-43065dbc-78d0-45b7-92be-f14c2926e2dc/plugin.lua index 206e5de3..8278f51e 100644 --- a/app/MindWork AI Studio/Plugins/languages/de-de-43065dbc-78d0-45b7-92be-f14c2926e2dc/plugin.lua +++ b/app/MindWork AI Studio/Plugins/languages/de-de-43065dbc-78d0-45b7-92be-f14c2926e2dc/plugin.lua @@ -9564,6 +9564,9 @@ UI_TEXT_CONTENT["AISTUDIO::PAGES::EMBEDDINGS::T1455637941"] = "Der Speicherort d -- Failed: {0} UI_TEXT_CONTENT["AISTUDIO::PAGES::EMBEDDINGS::T1531389696"] = "Fehlgeschlagen: {0}" +-- AI Studio indexes large data sources in steps of a few minutes, so that your other data sources get their turn in between. +UI_TEXT_CONTENT["AISTUDIO::PAGES::EMBEDDINGS::T1545941101"] = "AI Studio indexiert große Datenquellen in Etappen von wenigen Minuten, damit zwischendurch auch Ihre anderen Datenquellen an die Reihe kommen." + -- Open the settings UI_TEXT_CONTENT["AISTUDIO::PAGES::EMBEDDINGS::T1582896271"] = "Einstellungen öffnen" diff --git a/app/MindWork AI Studio/Plugins/languages/en-us-97dfb1ba-50c4-4440-8dfa-6575daf543c8/plugin.lua b/app/MindWork AI Studio/Plugins/languages/en-us-97dfb1ba-50c4-4440-8dfa-6575daf543c8/plugin.lua index 797327c7..234ff0a8 100644 --- a/app/MindWork AI Studio/Plugins/languages/en-us-97dfb1ba-50c4-4440-8dfa-6575daf543c8/plugin.lua +++ b/app/MindWork AI Studio/Plugins/languages/en-us-97dfb1ba-50c4-4440-8dfa-6575daf543c8/plugin.lua @@ -9564,6 +9564,9 @@ UI_TEXT_CONTENT["AISTUDIO::PAGES::EMBEDDINGS::T1455637941"] = "Could not open th -- Failed: {0} UI_TEXT_CONTENT["AISTUDIO::PAGES::EMBEDDINGS::T1531389696"] = "Failed: {0}" +-- AI Studio indexes large data sources in steps of a few minutes, so that your other data sources get their turn in between. +UI_TEXT_CONTENT["AISTUDIO::PAGES::EMBEDDINGS::T1545941101"] = "AI Studio indexes large data sources in steps of a few minutes, so that your other data sources get their turn in between." + -- Open the settings UI_TEXT_CONTENT["AISTUDIO::PAGES::EMBEDDINGS::T1582896271"] = "Open the settings" diff --git a/app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.State.cs b/app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.State.cs index fd381cdb..15d9e5d3 100644 --- a/app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.State.cs +++ b/app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.State.cs @@ -32,7 +32,7 @@ public sealed partial class DataSourceEmbeddingService dataSource.Id); var activeRun = this.CancelActiveDataSourceRun(dataSource); - this.ClearQueuedDataSourceState(dataSourceId); + this.runQueue.Clear(dataSourceId); if (activeRun is not null) await activeRun.Completion.Task; @@ -137,6 +137,8 @@ public sealed partial class DataSourceEmbeddingService IndexStoreClient? indexStore, CancellationToken token) { + // What failed while building the index which is thrown away now says nothing about the next one: + this.passes.TryRemove(dataSourceId, out _); await this.DeleteCollectionAsync(DataSourceEmbeddingNames.GetCollectionName(dataSourceId), vectorStore, token); indexStore ??= await databaseClientProvider.GetIndexStoreAsync(token); diff --git a/app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.cs b/app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.cs index 22e464f6..1a6f1dd3 100644 --- a/app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.cs +++ b/app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.cs @@ -1,6 +1,5 @@ using System.Collections.Concurrent; using System.Diagnostics.CodeAnalysis; -using System.Threading.Channels; using AIStudio.Provider; using AIStudio.Settings; @@ -33,13 +32,20 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM /// private readonly IReadOnlyList indexers = CreateIndexers(settingsManager, rustService, guardService, logger); - private readonly Channel queue = Channel.CreateUnbounded(); - private readonly ConcurrentDictionary queuedIds = new(StringComparer.OrdinalIgnoreCase); - private readonly ConcurrentDictionary runningIds = new(StringComparer.OrdinalIgnoreCase); - private readonly ConcurrentDictionary pendingRefreshModes = new(StringComparer.OrdinalIgnoreCase); + private readonly DataSourceRunQueue runQueue = new(); + + /// + /// Lets one run at a time embed, cf. ProcessDataSourceRunAsync. + /// + private readonly SemaphoreSlim runGate = new(1, 1); + private readonly ConcurrentDictionary activeRuns = new(StringComparer.OrdinalIgnoreCase); private readonly ConcurrentDictionary statuses = new(StringComparer.OrdinalIgnoreCase); - private readonly object queueStateLock = new(); + + /// + /// The pass of every data source whose last step left more to do, and of none else, cf. IndexingPass. + /// + private readonly ConcurrentDictionary passes = new(StringComparer.OrdinalIgnoreCase); /// /// Guards the run signature of every active run, cf. DataSourceRunControl.RunSignature. @@ -51,16 +57,6 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM private static string TB(string fallbackEN) => I18N.I.T(fallbackEN, typeof(DataSourceEmbeddingService).Namespace, nameof(DataSourceEmbeddingService)); - private enum DataSourceQueueRequestResult - { - QUEUED, - ALREADY_QUEUED, - RUNNING, - RUNNING_MARKED_PENDING, - } - - private sealed record DataSourceEmbeddingQueueItem(string DataSourceId, DataSourceEmbeddingRefreshMode RefreshMode); - private sealed record DataSourceRunControl(CancellationTokenSource TokenSource, TaskCompletionSource Completion) { /// @@ -458,7 +454,7 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM this.RefreshWatchers(); logger.LogDebug("Refreshed watcher state for data source '{DataSourceName}' ({DataSourceId}).", dataSource.Name, dataSource.Id); - var queueRequestResult = this.TryReserveDataSourceQueueSlot(dataSource.Id, queueAfterCurrentRun, refreshMode); + var queueRequestResult = this.runQueue.TryReserve(dataSource.Id, queueAfterCurrentRun, refreshMode); // // Every change of the embedding settings ends up here, whichever dialog made it. A run @@ -471,11 +467,11 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM switch (queueRequestResult) { case DataSourceQueueRequestResult.ALREADY_QUEUED: - logger.LogDebug("Data source '{DataSourceName}' ({DataSourceId}) is already queued for background embeddings. Ignoring duplicate queue request.", dataSource.Name, dataSource.Id); + logger.LogDebug("Data source '{DataSourceName}' ({DataSourceId}) is already queued for background embeddings. Merged the queue request into the queued run. RefreshMode={RefreshMode}.", dataSource.Name, dataSource.Id, refreshMode); return; case DataSourceQueueRequestResult.RUNNING: - logger.LogDebug("Data source '{DataSourceName}' ({DataSourceId}) is already being embedded. Ignoring duplicate queue request.", dataSource.Name, dataSource.Id); + logger.LogDebug("Data source '{DataSourceName}' ({DataSourceId}) is already being embedded. The queue request adds no further follow-up embedding run. RefreshMode={RefreshMode}.", dataSource.Name, dataSource.Id, refreshMode); return; case DataSourceQueueRequestResult.RUNNING_MARKED_PENDING: @@ -483,24 +479,18 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM return; } + var lane = this.GetQueueLane(dataSource.Id); logger.LogInformation( - "Queueing data source '{DataSourceName}' ({DataSourceId}) for background embedding hash check. RefreshMode={RefreshMode}.", + "Queueing data source '{DataSourceName}' ({DataSourceId}) for background embedding hash check. RefreshMode={RefreshMode}, Lane={Lane}.", dataSource.Name, dataSource.Id, - refreshMode); + refreshMode, + lane); if (!this.statuses.TryGetValue(dataSource.Id, out var currentStatus) || currentStatus.State is not DataSourceEmbeddingState.RUNNING) - { - this.UpsertStatus(this.CreateStatus( - dataSource, - DataSourceEmbeddingState.QUEUED, - currentStatus?.TotalDocuments ?? 0, - currentStatus?.IndexedDocuments ?? 0, - currentStatus?.FailedDocuments ?? 0, - failures: currentStatus?.Failures ?? [], - lastSyncUtc: currentStatus?.LastSyncUtc)); - } + this.UpsertStatus(this.GetQueuedStatus(dataSource)); + logger.LogDebug("Upserting status for data source '{DataSourceName}' ({DataSourceId}).", dataSource.Name, dataSource.Id); - await this.queue.Writer.WriteAsync(new DataSourceEmbeddingQueueItem(dataSource.Id, refreshMode)); + await this.runQueue.EnqueueAsync(dataSource.Id, lane); logger.LogDebug("Queued data source '{DataSourceName}' ({DataSourceId}).", dataSource.Name, dataSource.Id); } @@ -511,7 +501,7 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM this.RemoveWatcher(dataSource.Id); var activeRun = this.CancelActiveDataSourceRun(dataSource); - this.ClearQueuedDataSourceState(dataSource.Id); + this.runQueue.Clear(dataSource.Id); this.statuses.TryRemove(dataSource.Id, out _); if (activeRun is not null) { @@ -538,54 +528,81 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM while (!stoppingToken.IsCancellationRequested) { - var queueItem = await this.queue.Reader.ReadAsync(stoppingToken); - var dataSourceId = queueItem.DataSourceId; - this.MarkDataSourceRunStarted(dataSourceId); - - IIndexedDataSource? dataSource = null; + var runRequest = await this.runQueue.DequeueAsync(stoppingToken); try { - if (!this.TryGetConfiguredIndexedSource(dataSourceId, out dataSource)) - continue; - - await this.ProcessDataSourceRunAsync(dataSource, queueItem.RefreshMode, stoppingToken); + await this.RunDataSourceAsync(runRequest.DataSourceId, runRequest.RefreshMode, stoppingToken); } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { break; } - catch (VectorStoreUnreadableException exception) when (dataSource is not null) + } + } + + /// + /// Runs a data source which the queue marked as running, and queues its follow-up run afterward. + /// + /// + /// The one bracket around every run, whether it came out of the queue or from the startup check. + /// Whoever calls this has marked the run as running, cf. DataSourceRunQueue.DequeueAsync and + /// TryBeginRun. That is what turns every request during the run into exactly one follow-up run, + /// the next step of a data source which did its share of the work among them. It also shows the + /// data source as queued again once the run ends, instead of leaving it on running. + /// + /// The id of the data source. + /// Why the run was asked for. + /// The cancellation token. + private async Task RunDataSourceAsync(string dataSourceId, DataSourceEmbeddingRefreshMode refreshMode, CancellationToken token) + { + IIndexedDataSource? dataSource = null; + + try + { + if (!this.TryGetConfiguredIndexedSource(dataSourceId, out dataSource)) { - // - // Nothing is deleted and nothing is rebuilt here. The data source says what is - // wrong with it, stays out of the selection while it says so, and waits for the - // user to ask for the repair. - // - logger.LogError( - exception, - "The vector store of data source '{DataSourceName}' ({DataSourceId}) cannot be read. The data source is waiting for a repair.", - dataSource.Name, - dataSource.Id); - this.UpsertStatus(this.GetUnreadableVectorStoreStatus(dataSource)); + // Deleted, or no longer indexed since a preview was switched off. Its pass would lead nowhere: + this.passes.TryRemove(dataSourceId, out _); + return; } - catch (Exception exception) + + await this.ProcessDataSourceRunAsync(dataSource, refreshMode, token); + } + catch (OperationCanceledException) when (token.IsCancellationRequested) + { + throw; + } + catch (VectorStoreUnreadableException exception) when (dataSource is not null) + { + // + // Nothing is deleted and nothing is rebuilt here. The data source says what is + // wrong with it, stays out of the selection while it says so, and waits for the + // user to ask for the repair. + // + logger.LogError( + exception, + "The vector store of data source '{DataSourceName}' ({DataSourceId}) cannot be read. The data source is waiting for a repair.", + dataSource.Name, + dataSource.Id); + this.UpsertStatus(this.GetUnreadableVectorStoreStatus(dataSource)); + } + catch (Exception exception) + { + if (dataSource is null) { - if (dataSource is null) - { - logger.LogError(exception, "Background embedding failed for data source '{DataSourceId}'.", dataSourceId); - } - else - { - logger.LogError(exception, "Background embedding failed for data source '{DataSourceName}' ({DataSourceId}).", dataSource.Name, dataSource.Id); - this.UpsertStatus(this.GetFallbackStatus(dataSource, string.Format(TB("The data source '{0}' could not be processed. The log file holds the details."), dataSource.Name))); - } + logger.LogError(exception, "Background embedding failed for data source '{DataSourceId}'. RefreshMode={RefreshMode}.", dataSourceId, refreshMode); } - finally + else { - await this.QueuePendingDataSourceRunAsync(dataSourceId, stoppingToken); + logger.LogError(exception, "Background embedding failed for data source '{DataSourceName}' ({DataSourceId}). RefreshMode={RefreshMode}.", dataSource.Name, dataSource.Id, refreshMode); + this.UpsertStatus(this.GetFallbackStatus(dataSource, string.Format(TB("The data source '{0}' could not be processed. The log file holds the details."), dataSource.Name))); } } + finally + { + await this.QueuePendingDataSourceRunAsync(dataSourceId, token); + } } public override async Task StopAsync(CancellationToken cancellationToken) @@ -598,6 +615,7 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM { this.DisposeWatchers(); base.Dispose(); + this.runGate.Dispose(); } private async Task ProcessDataSourceRunAsync(IDataSourceBase requestedDataSource, DataSourceEmbeddingRefreshMode refreshMode, CancellationToken parentToken) @@ -630,7 +648,21 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM try { - await this.ProcessDataSourceAsync(dataSource, refreshMode, runControl, runTokenSource.Token); + // + // One run at a time, whichever way it started: the startup check goes on next to the + // queue once the automatic refresh is switched on while the app runs. The wait counts as + // part of the run, so deleting or repairing the data source ends the wait as well, + // instead of waiting for the run before it to finish first. + // + await this.runGate.WaitAsync(runTokenSource.Token); + try + { + await this.ProcessDataSourceAsync(dataSource, refreshMode, runControl, runTokenSource.Token); + } + finally + { + this.runGate.Release(); + } } catch (OperationCanceledException) when (!parentToken.IsCancellationRequested && runTokenSource.IsCancellationRequested) { @@ -641,23 +673,13 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM refreshMode); // - // A data source which is still there has its next run queued already. Its row says so - // instead of showing this run as running: a run of the startup check would otherwise - // keep that row until the check has worked through every other data source. + // A data source which is still there has its next run asked for already, which starts + // over with the current settings, in a pass of its own. Its row says so instead of + // showing this run as running. // + this.passes.TryRemove(dataSource.Id, out _); if (this.TryGetConfiguredIndexedSource(dataSource.Id, out var configuredDataSource)) - { - this.statuses.TryGetValue(configuredDataSource.Id, out var currentStatus); - this.UpsertStatus(this.CreateStatus( - configuredDataSource, - DataSourceEmbeddingState.QUEUED, - currentStatus?.TotalDocuments ?? 0, - currentStatus?.IndexedDocuments ?? 0, - currentStatus?.FailedDocuments ?? 0, - lastError: currentStatus?.LastError ?? string.Empty, - failures: currentStatus?.Failures ?? [], - lastSyncUtc: currentStatus?.LastSyncUtc)); - } + this.UpsertStatus(this.GetQueuedStatus(configuredDataSource)); } finally { @@ -678,10 +700,11 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM /// /// The data source to index. /// Why the run was started, for the log. + /// How much work the run does at most, as its kind of data source says. /// The run, which records the embedding settings it works for. /// The cancellation token. /// The context of the run, or null when there is no run. - private async Task PrepareIndexedRunAsync(IIndexedDataSource dataSource, DataSourceEmbeddingRefreshMode refreshMode, DataSourceRunControl runControl, CancellationToken token) + private async Task PrepareIndexedRunAsync(IIndexedDataSource dataSource, DataSourceEmbeddingRefreshMode refreshMode, RunShare share, DataSourceRunControl runControl, CancellationToken token) { logger.LogInformation( "Starting background embedding hash check for data source '{DataSourceName}' ({DataSourceId}). RefreshMode={RefreshMode}.", @@ -793,7 +816,22 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM var manifest = await this.EnsureCompatibleManifestAsync(dataSource, embeddingProvider, embeddingSignature, collectionName, vectorStore, indexStore, token); token.ThrowIfCancellationRequested(); - return new IndexedRunContext(dataSource, embeddingProvider, embeddingProvider.CreateProvider(), vectorStore, indexStore, manifest, settingsManager, this.UpsertStatus, logger); + // + // Only now, with the run signature recorded and checked, and with whatever an outdated + // manifest threw away: a pass belongs to the embedding settings it began with. + // + var lastPass = this.passes.GetValueOrDefault(dataSource.Id); + var pass = IndexingPass.ContinueOrBegin(lastPass, refreshMode, runSignature); + this.passes[dataSource.Id] = pass; + if (pass == lastPass) + logger.LogInformation( + "Carrying on the indexing pass of data source '{DataSourceName}' ({DataSourceId}). FailedEarlierInPass={FailedEarlierInPass}, RefreshMode={RefreshMode}.", + dataSource.Name, + dataSource.Id, + pass.FailedDocumentCount, + refreshMode); + + return new IndexedRunContext(dataSource, embeddingProvider, embeddingProvider.CreateProvider(), vectorStore, indexStore, manifest, share, pass, settingsManager, this.UpsertStatus, logger); } private async Task ProcessDataSourceAsync(IIndexedDataSource dataSource, DataSourceEmbeddingRefreshMode refreshMode, DataSourceRunControl runControl, CancellationToken token) @@ -807,18 +845,49 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM return; } - var context = await this.PrepareIndexedRunAsync(dataSource, refreshMode, runControl, token); - if (context is null) - return; + var outcome = IndexedRunOutcome.DONE; + try + { + var context = await this.PrepareIndexedRunAsync(dataSource, refreshMode, indexer.Share, runControl, token); + if (context is null) + return; - // - // Queued behind whatever else waits, so a data source which takes hours gives the others - // their turn. While this run is still active, that is the follow-up run every request - // during a run leaves behind. Should the user ask for a run of their own in the meantime, - // theirs comes first, and carries on just the same. - // - if (await indexer.ProcessAsync(context, refreshMode, token) is IndexedRunOutcome.MORE_TO_DO) - await this.QueueDataSourceAsync(dataSource, true, DataSourceEmbeddingRefreshMode.CONTINUATION); + var runStartedUtc = DateTimeOffset.UtcNow; + outcome = await indexer.ProcessAsync(context, refreshMode, token); + + // + // A kind of data source which never asks whether its run did its share holds up every + // other data source for as long as the run takes, and nothing fails to point that out: + // + var runDuration = DateTimeOffset.UtcNow - runStartedUtc; + if (context.ShareStartedUtc is null && runDuration > context.Share.MaxDuration * 2) + logger.LogWarning( + "The run of data source '{DataSourceName}' ({DataSourceId}) took {RunMinutes:F0} minutes without asking once whether it did its share of {ShareMinutes:F0} minutes, so every other data source waited that long. RefreshMode={RefreshMode}.", + dataSource.Name, + dataSource.Id, + runDuration.TotalMinutes, + context.Share.MaxDuration.TotalMinutes, + refreshMode); + + // + // While this run is still active, that is the follow-up run every request during a run + // leaves behind. It waits in the back lane, behind every data source which begins a + // pass and behind the other ones which carry on, so a data source which takes hours + // gives the others their turn, cf. GetQueueLane. Should the user ask for a run of their + // own in the meantime, the two merge into one. + // + if (outcome is IndexedRunOutcome.MORE_TO_DO) + await this.QueueDataSourceAsync(dataSource, true, DataSourceEmbeddingRefreshMode.CONTINUATION); + } + finally + { + // + // A pass lives only while the last step left more to do. Whatever else ended this run, + // the next one begins anew, and tries every document which failed so far once more: + // + if (outcome is not IndexedRunOutcome.MORE_TO_DO) + this.passes.TryRemove(dataSource.Id, out _); + } } private async Task DeleteCollectionAsync(string collectionName, VectorStoreClient? vectorStore, CancellationToken token) @@ -889,15 +958,7 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM if (this.statuses.TryGetValue(dataSource.Id, out var knownStatus) && knownStatus.State is DataSourceEmbeddingState.RUNNING) continue; - this.statuses[dataSource.Id] = this.CreateStatus( - dataSource, - DataSourceEmbeddingState.QUEUED, - knownStatus?.TotalDocuments ?? 0, - knownStatus?.IndexedDocuments ?? 0, - knownStatus?.FailedDocuments ?? 0, - failures: knownStatus?.Failures ?? [], - permanentlySkippedDocuments: knownStatus?.PermanentlySkippedDocuments ?? 0, - lastSyncUtc: knownStatus?.LastSyncUtc); + this.statuses[dataSource.Id] = this.GetQueuedStatus(dataSource); } // One message for the whole list, rather than one per data source: @@ -906,28 +967,18 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM foreach (var dataSource in supportedDataSources) { token.ThrowIfCancellationRequested(); - try + + // + // A data source which runs, or waits in the queue, gets its run that way already. + // Running it here as well would run it twice in a row, or next to the run going on. + // + if (!this.runQueue.TryBeginRun(dataSource.Id)) { - await this.ProcessDataSourceRunAsync(dataSource, DataSourceEmbeddingRefreshMode.STARTUP_HASH_CHECK, token); - } - catch (OperationCanceledException) when (token.IsCancellationRequested) - { - throw; - } - catch (VectorStoreUnreadableException exception) - { - logger.LogError( - exception, - "The vector store of data source '{DataSourceName}' ({DataSourceId}) cannot be read. The data source is waiting for a repair.", - dataSource.Name, - dataSource.Id); - this.UpsertStatus(this.GetUnreadableVectorStoreStatus(dataSource)); - } - catch (Exception exception) - { - logger.LogError(exception, "Initial embedding hash check failed for data source '{DataSourceName}' ({DataSourceId}).", dataSource.Name, dataSource.Id); - this.UpsertStatus(this.GetFallbackStatus(dataSource, string.Format(TB("The data source '{0}' could not be processed. The log file holds the details."), dataSource.Name))); + logger.LogDebug("Skipping the initial hash check of data source '{DataSourceName}' ({DataSourceId}) because it is running or queued already.", dataSource.Name, dataSource.Id); + continue; } + + await this.RunDataSourceAsync(dataSource.Id, DataSourceEmbeddingRefreshMode.STARTUP_HASH_CHECK, token); } if (!settingsManager.ConfigurationData.App.DataSourceIndexing.AutomaticRefresh) @@ -1117,6 +1168,44 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM LastSyncUtc: lastSyncUtc); } + /// + /// Shows a data source as waiting for its next run. + /// + /// The data source. + /// Its status so far as a queued one, or an empty queued status when it has none yet. + private DataSourceEmbeddingStatus GetQueuedStatus(IDataSourceBase dataSource) => this.statuses.TryGetValue(dataSource.Id, out var status) + ? AsQueued(status, dataSource) + : this.CreateStatus(dataSource, DataSourceEmbeddingState.QUEUED, 0, 0, 0); + + /// + /// Turns the status of a data source into one which waits for the next run. + /// + /// + /// Keeps what the row showed of the runs so far: the counts, the documents skipped for good, the + /// failures with the sentence about the data source as a whole, and the last sync. A data source + /// which pauses after its share of the work shows its progress that way, instead of a bar which + /// jumps back with every step. + /// + /// Clears what belongs to a run going on: the current document, and whatever the user was asked + /// to decide. The next run asks that anew when it still applies. The name comes from the data + /// source, since it may have been renamed in the meantime. + /// + /// The status so far. + /// The data source. + /// The queued status. + internal static DataSourceEmbeddingStatus AsQueued(DataSourceEmbeddingStatus status, IDataSourceBase dataSource) => status with + { + DataSourceName = dataSource.Name, + DataSourceType = dataSource.Type, + State = DataSourceEmbeddingState.QUEUED, + CurrentDocument = string.Empty, + CurrentDocumentBlock = null, + CurrentDocumentPage = null, + VectorStoreUnreadable = false, + Attention = DataSourceAttention.NONE, + PendingRemovalCount = null, + }; + private DataSourceEmbeddingStatus GetFallbackStatus(IDataSourceBase dataSource, string errorMessage) { return this.CreateStatus( @@ -1148,69 +1237,6 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM vectorStoreUnreadable: true); } - /// - /// A request arriving while the data source is being embedded leaves a mark for one follow-up - /// run, and the mark keeps why it was asked for. The first request decides that: any later one - /// only confirms that a follow-up is needed, which it already is. - /// - private DataSourceQueueRequestResult TryReserveDataSourceQueueSlot(string dataSourceId, bool queueAfterCurrentRun, DataSourceEmbeddingRefreshMode refreshMode) - { - lock (this.queueStateLock) - { - if (this.runningIds.ContainsKey(dataSourceId)) - { - if (queueAfterCurrentRun && this.pendingRefreshModes.TryAdd(dataSourceId, refreshMode)) - return DataSourceQueueRequestResult.RUNNING_MARKED_PENDING; - - return DataSourceQueueRequestResult.RUNNING; - } - - if (!this.queuedIds.TryAdd(dataSourceId, 0)) - return DataSourceQueueRequestResult.ALREADY_QUEUED; - - return DataSourceQueueRequestResult.QUEUED; - } - } - - private void MarkDataSourceRunStarted(string dataSourceId) - { - lock (this.queueStateLock) - { - this.queuedIds.TryRemove(dataSourceId, out _); - this.runningIds.TryAdd(dataSourceId, 0); - } - } - - private bool TryCompleteDataSourceRun(string dataSourceId, bool allowPendingRequeue, out DataSourceEmbeddingRefreshMode pendingRefreshMode) - { - lock (this.queueStateLock) - { - this.runningIds.TryRemove(dataSourceId, out _); - - if (!this.pendingRefreshModes.TryRemove(dataSourceId, out pendingRefreshMode)) - return false; - - return allowPendingRequeue && this.queuedIds.TryAdd(dataSourceId, 0); - } - } - - private void ReleaseQueuedDataSourceRun(string dataSourceId) - { - lock (this.queueStateLock) - { - this.queuedIds.TryRemove(dataSourceId, out _); - } - } - - private void ClearQueuedDataSourceState(string dataSourceId) - { - lock (this.queueStateLock) - { - this.queuedIds.TryRemove(dataSourceId, out _); - this.pendingRefreshModes.TryRemove(dataSourceId, out _); - } - } - private DataSourceRunControl? CancelActiveDataSourceRun(IDataSourceBase dataSource) { if (!this.activeRuns.TryGetValue(dataSource.Id, out var activeRun)) @@ -1314,38 +1340,46 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM { IIndexedDataSource? dataSource = null; var isConfigured = !token.IsCancellationRequested && this.TryGetConfiguredIndexedSource(dataSourceId, out dataSource); - if (!this.TryCompleteDataSourceRun(dataSourceId, isConfigured, out var refreshMode)) + if (!this.runQueue.TryComplete(dataSourceId, isConfigured, out var refreshMode)) return; if (dataSource is null) { - this.ReleaseQueuedDataSourceRun(dataSourceId); + this.runQueue.Release(dataSourceId); return; } - logger.LogInformation("Queueing one follow-up embedding run for data source '{DataSourceName}' ({DataSourceId}) after changes arrived during the previous run. RefreshMode={RefreshMode}.", dataSource.Name, dataSource.Id, refreshMode); + var lane = this.GetQueueLane(dataSourceId); + logger.LogInformation("Queueing one follow-up embedding run for data source '{DataSourceName}' ({DataSourceId}) after changes arrived during the previous run, or after it did its share of the work. RefreshMode={RefreshMode}, Lane={Lane}.", dataSource.Name, dataSource.Id, refreshMode, lane); - this.statuses.TryGetValue(dataSource.Id, out var currentStatus); - this.UpsertStatus(this.CreateStatus( - dataSource, - DataSourceEmbeddingState.QUEUED, - currentStatus?.TotalDocuments ?? 0, - currentStatus?.IndexedDocuments ?? 0, - currentStatus?.FailedDocuments ?? 0, - lastError: currentStatus?.LastError ?? string.Empty, - failures: currentStatus?.Failures ?? [], - lastSyncUtc: currentStatus?.LastSyncUtc)); + this.UpsertStatus(this.GetQueuedStatus(dataSource)); try { - await this.queue.Writer.WriteAsync(new DataSourceEmbeddingQueueItem(dataSourceId, refreshMode), token); + await this.runQueue.EnqueueAsync(dataSourceId, lane, token); } catch (OperationCanceledException) when (token.IsCancellationRequested) { - this.ReleaseQueuedDataSourceRun(dataSourceId); + this.runQueue.Release(dataSourceId); } } + /// + /// Picks the lane of the queue in which the next run of a data source waits. + /// + /// + /// A data source which carries on a pass waits behind every data source which begins one. A + /// mailbox at its interval therefore waits for one step at most, rather than for hours behind a + /// large folder. Since a pass begins without one, every pass gets one step in the front lane at + /// most: no data source keeps the front lane to itself, however long it takes. + /// + /// Asked when the run is enqueued, and only then. For a follow-up run, that is after the run + /// before it ended, which kept or dropped the pass by then. + /// + /// The id of the data source. + /// The back lane while the data source has a pass, the front lane otherwise. + private DataSourceQueueLane GetQueueLane(string dataSourceId) => this.passes.ContainsKey(dataSourceId) ? DataSourceQueueLane.BACK : DataSourceQueueLane.FRONT; + private void UpsertStatus(DataSourceEmbeddingStatus status) { this.statuses[status.DataSourceId] = status; diff --git a/app/MindWork AI Studio/Tools/Services/DataSourceQueueLane.cs b/app/MindWork AI Studio/Tools/Services/DataSourceQueueLane.cs new file mode 100644 index 00000000..9f981deb --- /dev/null +++ b/app/MindWork AI Studio/Tools/Services/DataSourceQueueLane.cs @@ -0,0 +1,23 @@ +namespace AIStudio.Tools.Services; + +/// +/// The lanes of the queue of indexing runs, cf. DataSourceRunQueue. +/// +/// +/// Every run in the front lane comes out before any run in the back lane. DataSourceEmbeddingService +/// picks the lane when it enqueues a run: a data source which carries on a pass of its indexing waits +/// in the back lane, cf. IndexingPass. A mailbox therefore waits for one step at most, rather than for +/// hours behind a large folder, and large data sources take turns with each other. +/// +internal enum DataSourceQueueLane +{ + /// + /// For a data source which begins a pass, e.g. a mailbox at its interval. + /// + FRONT, + + /// + /// For a data source which carries on a pass, after it did its share of the work. + /// + BACK, +} \ No newline at end of file diff --git a/app/MindWork AI Studio/Tools/Services/DataSourceQueueRequestResult.cs b/app/MindWork AI Studio/Tools/Services/DataSourceQueueRequestResult.cs new file mode 100644 index 00000000..0b0b953f --- /dev/null +++ b/app/MindWork AI Studio/Tools/Services/DataSourceQueueRequestResult.cs @@ -0,0 +1,27 @@ +namespace AIStudio.Tools.Services; + +/// +/// What became of a request to queue an indexing run, cf. DataSourceRunQueue.TryReserve. +/// +internal enum DataSourceQueueRequestResult +{ + /// + /// The data source has its place in the queue reserved, and whoever asked enqueues it now. + /// + QUEUED, + + /// + /// The data source waits in the queue already, and the request merged into that run. + /// + ALREADY_QUEUED, + + /// + /// The data source runs, and either has its follow-up run already or was asked for none. + /// + RUNNING, + + /// + /// The data source runs, and gets one follow-up run once the current one completes. + /// + RUNNING_MARKED_PENDING, +} \ No newline at end of file diff --git a/app/MindWork AI Studio/Tools/Services/DataSourceRunQueue.cs b/app/MindWork AI Studio/Tools/Services/DataSourceRunQueue.cs new file mode 100644 index 00000000..763af960 --- /dev/null +++ b/app/MindWork AI Studio/Tools/Services/DataSourceRunQueue.cs @@ -0,0 +1,268 @@ +using System.Diagnostics.CodeAnalysis; +using System.Threading.Channels; + +namespace AIStudio.Tools.Services; + +/// +/// The queue of indexing runs: which data sources wait for a run, which run, and which get one more run afterward. +/// +/// +/// A data source holds at most one place in the queue, and at most one follow-up run while it runs. +/// +/// The queue has two lanes, cf. DataSourceQueueLane. Every run in the front lane comes out before +/// any run in the back lane, and each lane hands out its runs in the order they came in. The data +/// sources in the back lane therefore take turns. +/// +/// The channel carries copies of ids, each with its lane and a number. Why a run was asked for is +/// kept next to it, together with the number of the copy which counts, and taking a run out of the +/// queue takes both along. Any other copy of the same id which is still in the channel then finds +/// nothing, or another number, and is skipped. That is how a data source which was deleted or +/// repaired while it waited leaves nothing behind: no run in an outdated mode, no second run +/// circling through the queue next to the one which was queued anew, and no copy in the front lane +/// which lets a later run of the data source overtake the back lane. +/// +/// Everything but the channel is guarded by one lock, so reserving, starting and completing a run +/// never see each other halfway. +/// +internal sealed class DataSourceRunQueue +{ + private readonly Channel channel = Channel.CreateUnboundedPrioritized(new UnboundedPrioritizedChannelOptions { Comparer = Comparer.Create(CompareCopies) }); + + /// + /// The data sources which wait for a run, each with why it was asked for. + /// + private readonly Dictionary queuedModes = new(StringComparer.OrdinalIgnoreCase); + + /// + /// The data sources which wait in the channel, each with the number of the copy which counts. + /// + /// + /// A data source whose place is reserved gets its number once it is enqueued. + /// + private readonly Dictionary queuedCopyNumbers = new(StringComparer.OrdinalIgnoreCase); + + /// + /// The data sources whose run has started and not completed yet. + /// + private readonly HashSet runningIds = new(StringComparer.OrdinalIgnoreCase); + + /// + /// The running data sources which get one more run once the current one completes, each with why it was asked for. + /// + private readonly Dictionary pendingModes = new(StringComparer.OrdinalIgnoreCase); + + private readonly Lock stateLock = new(); + + private long lastCopyNumber; + + /// + /// One copy of the id of a data source in the channel. + /// + /// The id of the data source. + /// The lane the copy waits in. + /// Counts up with every copy. It keeps each lane in the order its copies came in, and tells the copy which counts from those left behind. + private sealed record QueuedCopy(string DataSourceId, DataSourceQueueLane Lane, long Number); + + /// + /// Reserves a place in the queue for a data source, or one follow-up run while it runs. + /// + /// + /// A request for a data source which holds a place or a follow-up run already merges into it. + /// The first request decides why the run happens, and any later one only confirms that it is + /// needed. A manual retry is the exception and replaces any other mode: it is the only one which + /// signs in to a mailbox again after its server refused a sign-in, cf. MailboxIndexer. Behind a + /// sync at the interval which waits already, the retry the user asked for would be lost. + /// + /// The place is taken once its id is in the channel, cf. EnqueueAsync. + /// + /// The id of the data source. + /// Whether a request while the data source runs leaves a follow-up run. + /// Why the run is asked for. + /// QUEUED when the caller has to enqueue the data source now, or what became of the request instead. + public DataSourceQueueRequestResult TryReserve(string dataSourceId, bool queueAfterCurrentRun, DataSourceEmbeddingRefreshMode refreshMode) + { + lock (this.stateLock) + { + if (this.runningIds.Contains(dataSourceId)) + { + if (!queueAfterCurrentRun) + return DataSourceQueueRequestResult.RUNNING; + + return TryAddOrMerge(this.pendingModes, dataSourceId, refreshMode) + ? DataSourceQueueRequestResult.RUNNING_MARKED_PENDING + : DataSourceQueueRequestResult.RUNNING; + } + + return TryAddOrMerge(this.queuedModes, dataSourceId, refreshMode) + ? DataSourceQueueRequestResult.QUEUED + : DataSourceQueueRequestResult.ALREADY_QUEUED; + } + } + + /// + /// Puts a data source into the queue, once its place is reserved. + /// + /// The id of the data source. + /// The lane the data source waits in. + /// The cancellation token. + public ValueTask EnqueueAsync(string dataSourceId, DataSourceQueueLane lane, CancellationToken token = default) + { + QueuedCopy copy; + lock (this.stateLock) + { + // Cleared since its place was reserved, e.g. because the data source was deleted: + if (!this.queuedModes.ContainsKey(dataSourceId)) + return ValueTask.CompletedTask; + + copy = new(dataSourceId, lane, ++this.lastCopyNumber); + this.queuedCopyNumbers[dataSourceId] = copy.Number; + } + + return this.channel.Writer.WriteAsync(copy, token); + } + + /// + /// Waits for the next run, takes it out of the queue, and marks it as running. + /// + /// The cancellation token. + /// The run. + public async ValueTask DequeueAsync(CancellationToken token) + { + while (true) + { + if (this.TryDequeue(out var runRequest)) + return runRequest; + + if (!await this.channel.Reader.WaitToReadAsync(token)) + throw new ChannelClosedException(); + } + } + + /// + /// Takes the next run out of the queue, when there is one, and marks it as running. + /// + /// The run, when there is one. + /// True when a run was taken out of the queue. + public bool TryDequeue([NotNullWhen(true)] out DataSourceRunRequest? runRequest) + { + while (this.channel.Reader.TryRead(out var copy)) + { + lock (this.stateLock) + { + // Released or cleared while it waited, or enqueued anew since, with another copy which counts instead: + if (!this.queuedCopyNumbers.TryGetValue(copy.DataSourceId, out var number) || number != copy.Number) + continue; + + this.queuedCopyNumbers.Remove(copy.DataSourceId); + this.queuedModes.Remove(copy.DataSourceId, out var refreshMode); + this.runningIds.Add(copy.DataSourceId); + runRequest = new(copy.DataSourceId, refreshMode); + return true; + } + } + + runRequest = null; + return false; + } + + /// + /// Marks a data source as running which did not come out of the queue, unless it runs or waits in the queue already. + /// + /// + /// For the startup check, which works through every data source on its own. Its runs have to + /// count as running as well: a request during one of them then leaves one follow-up run, instead + /// of a second run next to it. + /// + /// The id of the data source. + /// True when the caller runs the data source now, and completes the run with TryComplete. + public bool TryBeginRun(string dataSourceId) + { + lock (this.stateLock) + { + if (this.queuedModes.ContainsKey(dataSourceId)) + return false; + + return this.runningIds.Add(dataSourceId); + } + } + + /// + /// Marks the run of a data source as completed, and reserves the place of its follow-up run when it has one. + /// + /// The id of the data source. + /// False drops the follow-up run, e.g. because the data source is gone. + /// Why the follow-up run was asked for. + /// True when the caller has to enqueue the follow-up run now. + public bool TryComplete(string dataSourceId, bool requeuePending, out DataSourceEmbeddingRefreshMode pendingRefreshMode) + { + lock (this.stateLock) + { + this.runningIds.Remove(dataSourceId); + + if (!this.pendingModes.Remove(dataSourceId, out pendingRefreshMode)) + return false; + + return requeuePending && TryAddOrMerge(this.queuedModes, dataSourceId, pendingRefreshMode); + } + } + + /// + /// Gives up the reserved place of a data source, because it could not be enqueued. + /// + /// The id of the data source. + public void Release(string dataSourceId) + { + lock (this.stateLock) + { + this.queuedModes.Remove(dataSourceId); + this.queuedCopyNumbers.Remove(dataSourceId); + } + } + + /// + /// Forgets the place and the follow-up run of a data source, e.g. because it was deleted. + /// + /// + /// A run which is going on stays marked as running until it completes. Whoever calls this stops + /// that run on their own. + /// + /// The id of the data source. + public void Clear(string dataSourceId) + { + lock (this.stateLock) + { + this.queuedModes.Remove(dataSourceId); + this.queuedCopyNumbers.Remove(dataSourceId); + this.pendingModes.Remove(dataSourceId); + } + } + + /// + /// Orders the copies in the channel: the front lane before the back lane, and each lane in the order its copies came in. + /// + private static int CompareCopies(QueuedCopy x, QueuedCopy y) + { + if (x.Lane != y.Lane) + return x.Lane is DataSourceQueueLane.FRONT ? -1 : 1; + + return x.Number.CompareTo(y.Number); + } + + /// + /// Records why a run of a data source was asked for, or merges the request into the one recorded already. + /// + /// The recorded runs. + /// The id of the data source. + /// Why the run is asked for. + /// True when there was no run recorded for the data source yet. + private static bool TryAddOrMerge(Dictionary modes, string dataSourceId, DataSourceEmbeddingRefreshMode refreshMode) + { + if (modes.TryAdd(dataSourceId, refreshMode)) + return true; + + if (refreshMode is DataSourceEmbeddingRefreshMode.MANUAL_RETRY) + modes[dataSourceId] = refreshMode; + + return false; + } +} \ No newline at end of file diff --git a/app/MindWork AI Studio/Tools/Services/DataSourceRunRequest.cs b/app/MindWork AI Studio/Tools/Services/DataSourceRunRequest.cs new file mode 100644 index 00000000..4f24a7d6 --- /dev/null +++ b/app/MindWork AI Studio/Tools/Services/DataSourceRunRequest.cs @@ -0,0 +1,8 @@ +namespace AIStudio.Tools.Services; + +/// +/// A run of a data source, as it comes out of the queue. +/// +/// The id of the data source. +/// Why the run was asked for. +internal sealed record DataSourceRunRequest(string DataSourceId, DataSourceEmbeddingRefreshMode RefreshMode); \ No newline at end of file diff --git a/app/MindWork AI Studio/Tools/Services/Indexing/AGENTS.md b/app/MindWork AI Studio/Tools/Services/Indexing/AGENTS.md index b7d00c5b..2cd6e724 100644 --- a/app/MindWork AI Studio/Tools/Services/Indexing/AGENTS.md +++ b/app/MindWork AI Studio/Tools/Services/Indexing/AGENTS.md @@ -21,7 +21,14 @@ The parts: attempt left behind. - **`EmbeddingDocument`** - one document: its key, its index row, its display name, and how to read its chunks. - **`DocumentRunProgress`** - counts the documents, records indexed and failed ones in the stores, publishes the - status and completes the run. + status, and completes or pauses the run. +- **`RunShare`** - how much one run may do before the other data sources get their turn: five minutes, and for + a mailbox at most 500 mails. A large data source is indexed in steps, one run each. A run which returns + `MORE_TO_DO` gets its next step in the back lane of `DataSourceRunQueue`, behind every data source which begins + a pass, so a mailbox waits for one step at most. +- **`IndexingPass`** - the steps of one data source, from its first step until one of them is done. It remembers + which documents failed during an earlier step, so the next steps do not read them again, and which failures + were reported to the user already. - **`TextChunker`** - cuts text into chunks the embedding provider accepts. Pick one of its strategies; do not write a chunker of your own. @@ -35,6 +42,13 @@ To add a kind of data source: 4. Keep whatever the kind has to remember beyond its documents in tables of its own in the index store, added by an EF Core migration (see `app/MindWork AI Studio/Tools/Databases/AGENTS.md`). 5. Report every status through `DocumentRunProgress`, so all rows of the embedding page behave alike. +6. Work in steps. Give the indexer its `Share`, and ask `DocumentRunProgress.HasDoneItsShare` before every + document. Leave out the documents which failed during an earlier step of the pass: either ask + `SkipIfFailedEarlierInPass` per document, or sort them out up front like `FileSourceIndexer.PartitionFiles` + and count them with `RecordFailedEarlierInPass`. Once the share is done, return what `PauseRunAsync` + returns: it decides whether the data source carries on in another run, or stops because its embedding + provider failed for everything. A run which takes more than twice its share without asking once ends up as + a warning in the log, while every other data source waits. Rules which are easy to break: - **A document key is not a path.** Only files use their full path as the key. Never pass a key through the @@ -45,3 +59,11 @@ Rules which are easy to break: deliberately: that rebuilds every index. - **The service decides when, the indexer decides how.** Whether changes are tracked at all depends on the automatic refresh setting and the startup hash check, and only the embedding service decides that. +- **A pause stores the source hash only when a partial index may be searched.** A stored source hash tells the + data source selection that a run got through the whole data source (`IsIndexAwaitingRebuild`). Files pass + `null` to `PauseRunAsync`, so a folder stays out of the selection until its first indexing or its rebuild is + complete. The mailbox passes its hash, since every mail it indexed can be searched right away. +- **A pass lives only while the last step left more to do.** The service drops the pass of a data source after + every other outcome and every exception, and when the stored index is reset. A retry the user asks for, and a + step with other embedding settings, begin a new pass (`IndexingPass.ContinueOrBegin`). A pass kept beyond that + leaves out documents which would work by now, for example once the embedding provider is reachable again. diff --git a/app/MindWork AI Studio/Tools/Services/Indexing/DocumentRunProgress.cs b/app/MindWork AI Studio/Tools/Services/Indexing/DocumentRunProgress.cs index c04a4397..0ad62ebb 100644 --- a/app/MindWork AI Studio/Tools/Services/Indexing/DocumentRunProgress.cs +++ b/app/MindWork AI Studio/Tools/Services/Indexing/DocumentRunProgress.cs @@ -14,7 +14,8 @@ namespace AIStudio.Tools.Services.Indexing; /// Counts the documents, keeps the list of failures and turns both into the status the embedding /// page shows. What happens to a document after it was indexed, or after it failed, is the same for /// every kind of data source, so that is decided here as well: the stores learn about it, the -/// manifest follows, and the user hears about it the way the failure deserves. +/// manifest follows, and the user hears about it the way the failure deserves. So is when a run did +/// its share of the work and leaves the rest to another one. /// /// The run the documents are indexed in. /// How many documents the data source has, readable or not. @@ -31,12 +32,20 @@ internal sealed class DocumentRunProgress(IndexedRunContext context, int totalDo private readonly List failures = inputFailures.ToList(); - // - // Which kinds of provider failure the user was already told about in this run. A rejected - // API key is the same problem for every one of a few thousand documents, and one message - // is what it takes to send the user to the settings. - // - private readonly HashSet reportedFailureReasons = []; + /// + /// The documents this run read, whether indexing them worked or not, cf. HasDoneItsShare. + /// + private int readDocuments; + + /// + /// The documents this run read which failed at the embedding provider, cf. PauseRunAsync. + /// + private int failedAtProvider; + + /// + /// The documents this run read which failed for any other reason which may pass, cf. PauseRunAsync. + /// + private int failedOtherwise; private static string TB(string fallbackEN) => I18N.I.T(fallbackEN, typeof(DocumentRunProgress).Namespace, nameof(DocumentRunProgress)); @@ -96,6 +105,38 @@ internal sealed class DocumentRunProgress(IndexedRunContext context, int totalDo this.failures.Add(new DataSourceEmbeddingFailure(documentKey, failure.Message, failure.OccurredAtUtc, ExtractionCode: failure.Code, IsPermanent: true)); } + /// + /// Skips a document which failed during an earlier step of the same pass, unless it changed since. + /// + /// + /// Such a document counts as failed, and its failure keeps its place in the list, the same as in + /// the step which read it. The counts stay right even when documents were deleted in between, + /// since only what the step comes across is counted. Should the document still be there once the + /// pass is through, the next pass tries it again. + /// + /// The key of the document. + /// The fingerprint of the document as it is now. + /// True when the document is skipped, and the caller moves on to the next one. + public bool SkipIfFailedEarlierInPass(string documentKey, string fingerprint) + { + if (!context.Pass.TryGetEarlierFailure(documentKey, fingerprint, out var failure)) + return false; + + this.RecordFailedEarlierInPass(failure); + return true; + } + + /// + /// Counts a document which was found to have failed during an earlier step of the same pass, cf. SkipIfFailedEarlierInPass. + /// + /// Why it failed back then, as IndexingPass.TryGetEarlierFailure tells. + public void RecordFailedEarlierInPass(DataSourceEmbeddingFailure failure) + { + this.FailedDocuments++; + this.LastError = failure.Reason; + this.failures.Add(failure); + } + /// /// Tells the user interface where the run stands. /// @@ -155,7 +196,9 @@ internal sealed class DocumentRunProgress(IndexedRunContext context, int totalDo await context.IndexStore.UpsertFileAsync(context.DataSource.Id, state, token); context.Manifest.Files[document.Key] = new EmbeddedFileRecord(state.Fingerprint, state.FileSize, state.LastWriteUtc, embeddedAtUtc, chunkCount); await context.ForgetPermanentFailureAsync(document.Key, token); + context.Pass.Forget(document.Key); + this.readDocuments++; this.IndexedDocuments++; if (isNew) this.NewDocuments++; @@ -175,6 +218,9 @@ internal sealed class DocumentRunProgress(IndexedRunContext context, int totalDo /// The cancellation token. public async Task RecordDocumentFailureAsync(EmbeddingDocument document, Exception exception, CancellationToken token) { + // Whichever way it failed, trying cost its time. A mail may even fail before its text was read: + this.readDocuments++; + var dataSource = context.DataSource; switch (exception) { @@ -186,8 +232,11 @@ internal sealed class DocumentRunProgress(IndexedRunContext context, int totalDo // needs to offer the matching way out. // this.FailedDocuments++; + this.failedAtProvider++; this.LastError = providerFailure.UserMessage; - this.failures.Add(new DataSourceEmbeddingFailure(document.Key, providerFailure.UserMessage, DateTimeOffset.UtcNow, providerFailure.FailureReason, providerFailure.StatusCode, context.EmbeddingProvider.Name, DisplayName: document.DisplayName)); + var failure = new DataSourceEmbeddingFailure(document.Key, providerFailure.UserMessage, DateTimeOffset.UtcNow, providerFailure.FailureReason, providerFailure.StatusCode, context.EmbeddingProvider.Name, DisplayName: document.DisplayName); + this.failures.Add(failure); + context.Pass.RecordFailure(document.Key, document.State.Fingerprint, failure); context.Manifest.Files.Remove(document.Key); await context.ForgetPermanentFailureAsync(document.Key, token); await context.CleanupFailedDocumentAsync(document.Key, token); @@ -202,8 +251,8 @@ internal sealed class DocumentRunProgress(IndexedRunContext context, int totalDo providerFailure.StatusCode); this.Publish(document.DisplayName); - // Once per kind of failure, not once per document: - if (this.reportedFailureReasons.Add(providerFailure.FailureReason)) + // Once per kind of failure and pass, not once per document or step: + if (context.Pass.TryMarkReported(providerFailure.FailureReason)) await MessageBus.INSTANCE.SendError(new(Icons.Material.Filled.CloudOff, providerFailure.UserMessage)); break; @@ -222,6 +271,7 @@ internal sealed class DocumentRunProgress(IndexedRunContext context, int totalDo var indexingMessage = this.GetFailureMessage(extractionFailure.Code, document); this.failures.Add(new DataSourceEmbeddingFailure(document.Key, indexingMessage, occurredAtUtc, ExtractionCode: extractionFailure.Code, IsPermanent: true, DisplayName: document.DisplayName)); context.Manifest.Files.Remove(document.Key); + context.Pass.Forget(document.Key); await context.CleanupFailedDocumentAsync(document.Key, token); var state = document.State; @@ -252,6 +302,7 @@ internal sealed class DocumentRunProgress(IndexedRunContext context, int totalDo // which would interrupt whatever the user is doing right now. // this.FailedDocuments++; + this.failedOtherwise++; var extractionCode = exception is FileExtractionException extractionFailure ? extractionFailure.Code : FileExtractionErrorCode.NONE; // @@ -260,7 +311,9 @@ internal sealed class DocumentRunProgress(IndexedRunContext context, int totalDo // var failureMessage = this.GetFailureMessage(extractionCode, document); this.LastError = failureMessage; - this.failures.Add(new DataSourceEmbeddingFailure(document.Key, failureMessage, DateTimeOffset.UtcNow, EmbeddingProviderName: context.EmbeddingProvider.Name, ExtractionCode: extractionCode, DisplayName: document.DisplayName)); + var failure = new DataSourceEmbeddingFailure(document.Key, failureMessage, DateTimeOffset.UtcNow, EmbeddingProviderName: context.EmbeddingProvider.Name, ExtractionCode: extractionCode, DisplayName: document.DisplayName); + this.failures.Add(failure); + context.Pass.RecordFailure(document.Key, document.State.Fingerprint, failure); context.Manifest.Files.Remove(document.Key); await context.ForgetPermanentFailureAsync(document.Key, token); await context.CleanupFailedDocumentAsync(document.Key, token); @@ -272,6 +325,30 @@ internal sealed class DocumentRunProgress(IndexedRunContext context, int totalDo } } + /// + /// Whether the run did its share of the work, so other data sources get their turn. + /// + /// + /// Asked before every document which is about to be read. The clock starts with the first + /// question, so every run reads at least one document. Counted are the documents which were + /// read, whether indexing them worked or not, cf. RunShare. + /// + /// The share only ends a run between two documents. A document which takes longer than the + /// whole share is read to its end, since its text arrives as a stream which cannot pick up where + /// it stopped. + /// + /// The current time, passed in so the share can be tested. + /// True when the run pauses now, cf. PauseRunAsync. + public bool HasDoneItsShare(DateTimeOffset nowUtc) + { + var shareStartedUtc = context.StartShareClock(nowUtc); + var share = context.Share; + if (share.MaxDocuments is { } maxDocuments && this.readDocuments >= maxDocuments) + return true; + + return nowUtc - shareStartedUtc >= share.MaxDuration; + } + /// /// Finishes the run: the collection is tidied up, the data source is marked as worked through, /// and the user interface learns how it went. @@ -286,40 +363,58 @@ internal sealed class DocumentRunProgress(IndexedRunContext context, int totalDo public async Task CompleteRunAsync(string sourceHash, string reason, CancellationToken token) { await this.StoreSourceHashAsync(sourceHash, reason, token); - - // - // Documents which were skipped for good do not make a run unsuccessful: nothing is left to - // try, and a data source made of nothing but scanned images would otherwise ask for - // attention forever. - // - var hasFailures = this.FailedDocuments > 0; - var lastError = hasFailures - ? string.IsNullOrWhiteSpace(this.LastError) - ? context.DataSource is DataSourceMailbox - ? TB("Some mails could not be indexed. The list below says which ones and why.") - : TB("Some files could not be indexed. The list below says which ones and why.") - : this.LastError - : string.Empty; - - context.PublishStatus(this.CreateStatus(hasFailures ? DataSourceEmbeddingState.FAILED : DataSourceEmbeddingState.COMPLETED, string.Empty, lastError, null, null)); + this.PublishCompletion(); } /// - /// Ends a run before it got through the whole data source, keeping what it did for the next one. + /// Ends a run which did its share of the work before it got through the whole data source, keeping what it did for the next one. /// /// - /// Stores everything CompleteRunAsync stores, the hash included. For a data source which is - /// worked through in several runs, the hash therefore says that its index can be searched, not - /// that the data source was worked through as a whole: that is what LastSyncUtc tells. Without - /// it, a mailbox would stay out of every search for the hours its first sync takes. + /// A run hands on to another one unless it indexed nothing and every document it read failed at + /// the embedding provider. Then the provider is most likely down, or refuses the API key, and + /// every further run would fail the same way, one document after the other. Such a run is + /// completed with its failures instead, and waits for the next time the data source is queued. + /// Any other failure is about its document, which the pass skips from now on, so the next run + /// gets further all the same. /// - /// The user interface learns nothing here. Whatever comes next tells it: the run being queued - /// again, or the question the user has to answer first. + /// The hash is stored only when one is given. It then says that the index of the data source can + /// be searched, not that the data source was worked through as a whole: that is what LastSyncUtc + /// tells. Without it, a mailbox would stay out of every search for the hours its first sync + /// takes. Without a hash, the stored one stays as it is, so a folder which is indexed for the + /// first time stays out of every search until its index is complete. + /// + /// When the run carries on, the user interface learns nothing here. Whatever comes next tells + /// it: the run being queued again, or the question the user has to answer first. /// - /// The hash of the data source as this run found it. - /// Why the collection is optimized now, for the log. + /// The hash of the data source as this run found it, when its index may be searched before it is complete. Null otherwise. + /// Why the run pauses, for the log. /// The cancellation token. - public Task PauseRunAsync(string sourceHash, string reason, CancellationToken token) => this.StoreSourceHashAsync(sourceHash, reason, token); + /// MORE_TO_DO when another run carries on, or DONE when this one was completed instead. + public async Task PauseRunAsync(string? searchableSourceHash, string reason, CancellationToken token) + { + if (searchableSourceHash is null) + await context.OptimizeCollectionIfNeededAsync(reason, token); + else + await this.StoreSourceHashAsync(searchableSourceHash, reason, token); + + // + // Only what this run read counts. Documents which could not even be looked at fail in every + // run, and would stop a folder with a single locked subfolder after every step. Neither do + // documents which failed for a reason of their own: they are never read again anyway, and + // never got as far as the provider. + // + if (this.IndexedDocuments > 0 || this.failedAtProvider is 0 || this.failedOtherwise > 0) + return IndexedRunOutcome.MORE_TO_DO; + + logger.LogWarning( + "Indexing data source '{DataSourceName}' ({DataSourceId}) stops without carrying on, since it indexed nothing and the embedding provider failed for all {FailedDocuments} document(s) it was sent.", + context.DataSource.Name, + context.DataSource.Id, + this.failedAtProvider); + + this.PublishCompletion(); + return IndexedRunOutcome.DONE; + } /// /// Tells the user interface how the data source stands as the index holds it, without a run. @@ -354,7 +449,15 @@ internal sealed class DocumentRunProgress(IndexedRunContext context, int totalDo /// /// Stores what a run found about the data source as a whole, and tidies up the collection. /// - private async Task StoreSourceHashAsync(string sourceHash, string reason, CancellationToken token) + /// + /// For a run which stops for a reason of its own and leaves an index which may be searched, + /// e.g. the sync of a mailbox which waits for the user to agree to a removal. The user interface + /// learns nothing here. + /// + /// The hash of the data source as this run found it. + /// Why the collection is optimized now, for the log. + /// The cancellation token. + public async Task StoreSourceHashAsync(string sourceHash, string reason, CancellationToken token) { context.Manifest.SourceHash = sourceHash; token.ThrowIfCancellationRequested(); @@ -365,6 +468,28 @@ internal sealed class DocumentRunProgress(IndexedRunContext context, int totalDo token.ThrowIfCancellationRequested(); } + /// + /// Tells the user interface that the run is over, and how it went. + /// + private void PublishCompletion() + { + // + // Documents which were skipped for good do not make a run unsuccessful: nothing is left to + // try, and a data source made of nothing but scanned images would otherwise ask for + // attention forever. + // + var hasFailures = this.FailedDocuments > 0; + var lastError = hasFailures + ? string.IsNullOrWhiteSpace(this.LastError) + ? context.DataSource is DataSourceMailbox + ? TB("Some mails could not be indexed. The list below says which ones and why.") + : TB("Some files could not be indexed. The list below says which ones and why.") + : this.LastError + : string.Empty; + + context.PublishStatus(this.CreateStatus(hasFailures ? DataSourceEmbeddingState.FAILED : DataSourceEmbeddingState.COMPLETED, string.Empty, lastError, null, null)); + } + /// /// What the user reads about a document which could not be indexed. /// diff --git a/app/MindWork AI Studio/Tools/Services/Indexing/FilePartition.cs b/app/MindWork AI Studio/Tools/Services/Indexing/FilePartition.cs new file mode 100644 index 00000000..04e7c757 --- /dev/null +++ b/app/MindWork AI Studio/Tools/Services/Indexing/FilePartition.cs @@ -0,0 +1,14 @@ +namespace AIStudio.Tools.Services.Indexing; + +/// +/// The files of a data source, sorted by what one run of the file indexer does with them. +/// +/// +/// Only the last group costs time. The others are counted before the first file is read, so the +/// progress of a step starts where the step before left off, rather than at zero. +/// +/// How many files were indexed and have not changed since. +/// The files which failed for a reason of their own and have not changed since, each with the stored reason. +/// Why each file failed which failed during an earlier step of the pass, and is not tried again before the next pass. +/// The files the run reads, in the order it reads them. +internal sealed record FilePartition(int UnchangedCount, IReadOnlyList> StillUnreadable, IReadOnlyList FailedEarlierInPass, IReadOnlyList ToRead); \ No newline at end of file diff --git a/app/MindWork AI Studio/Tools/Services/Indexing/FileSourceIndexer.cs b/app/MindWork AI Studio/Tools/Services/Indexing/FileSourceIndexer.cs index bb101058..7196235a 100644 --- a/app/MindWork AI Studio/Tools/Services/Indexing/FileSourceIndexer.cs +++ b/app/MindWork AI Studio/Tools/Services/Indexing/FileSourceIndexer.cs @@ -23,6 +23,12 @@ namespace AIStudio.Tools.Services.Indexing; /// /// Changes are noticed by a file system watcher per data source. A burst of changes is waited out /// before a run is asked for, so saving a document twice in a row costs one run, not two. +/// +/// A run reads files for its share of the time, cf. Share, and leaves the rest to the next one, so +/// a folder which takes hours does not hold up every other data source. Every such step lists the +/// folder anew, and counts what needs no reading before it reads the first file. A step stores no +/// hash, so a folder which is indexed for the first time stays out of every search until its index +/// is complete, cf. DocumentRunProgress.PauseRunAsync. /// /// The settings, read for whether local data sources refresh on their own. /// The runtime, which extracts the text of the files. @@ -49,6 +55,9 @@ internal sealed partial class FileSourceIndexer(SettingsManager settingsManager, /// public bool Supports(IDataSourceBase dataSource) => dataSource is DataSourceLocalDirectory or DataSourceLocalFile; + /// + public RunShare Share => new(RunShare.DEFAULT_DURATION); + /// public async Task ProcessAsync(IndexedRunContext context, DataSourceEmbeddingRefreshMode refreshMode, CancellationToken token) { @@ -57,7 +66,9 @@ internal sealed partial class FileSourceIndexer(SettingsManager settingsManager, var manifest = context.Manifest; + var enumerationStartedUtc = DateTimeOffset.UtcNow; var inputFiles = this.GetInputFiles(dataSource); + var enumerationDuration = DateTimeOffset.UtcNow - enumerationStartedUtc; var indexedFiles = inputFiles.Files; var totalFiles = indexedFiles.Count + inputFiles.FailedFiles; @@ -71,12 +82,17 @@ internal sealed partial class FileSourceIndexer(SettingsManager settingsManager, failure.Reason); } + // + // Listing the files is what every step of a large folder pays before it reads anything, + // hence the time it took: + // logger.LogInformation( - "Prepared data source '{DataSourceName}' ({DataSourceId}) for embedding. AccessibleFiles={AccessibleFiles}, FailedFiles={FailedFiles}, Collection='{CollectionName}'.", + "Prepared data source '{DataSourceName}' ({DataSourceId}) for embedding. AccessibleFiles={AccessibleFiles}, FailedFiles={FailedFiles}, EnumerationMs={EnumerationMs:F0}, Collection='{CollectionName}'.", dataSource.Name, dataSource.Id, indexedFiles.Count, inputFiles.FailedFiles, + enumerationDuration.TotalMilliseconds, context.CollectionName); var metadataSnapshot = this.BuildDataSourceMetadataSnapshot(dataSource, indexedFiles); @@ -116,60 +132,61 @@ internal sealed partial class FileSourceIndexer(SettingsManager settingsManager, } token.ThrowIfCancellationRequested(); + + // + // Everything which needs no reading is counted before the first file is read. Counted one + // by one, the progress of every step would start at zero, and climb through the files the + // steps before already did. + // + var partition = PartitionFiles(indexedFiles, metadataSnapshot.FileHashes, manifest, context.Pass); + progress.RecordUnchanged(partition.UnchangedCount); + foreach (var (filePath, permanentFailure) in partition.StillUnreadable) + { + logger.LogDebug( + "Skipping file '{FilePath}' for data source '{DataSourceName}' ({DataSourceId}) because reading it failed permanently before. FailureCode={FailureCode}, MetadataHashPrefix={MetadataHashPrefix}, OccurredAtUtc={OccurredAtUtc:O}.", + filePath, + dataSource.Name, + dataSource.Id, + permanentFailure.Code, + ShortHash(permanentFailure.Fingerprint), + permanentFailure.OccurredAtUtc); + progress.RecordStillUnreadable(filePath, permanentFailure); + } + + foreach (var failure in partition.FailedEarlierInPass) + progress.RecordFailedEarlierInPass(failure); + + logger.LogInformation( + "Sorted the files of data source '{DataSourceName}' ({DataSourceId}). Unchanged={UnchangedFiles}, PermanentlySkipped={PermanentlySkippedFiles}, FailedEarlierInPass={FailedEarlierInPass}, ToRead={FilesToRead}.", + dataSource.Name, + dataSource.Id, + partition.UnchangedCount, + partition.StillUnreadable.Count, + partition.FailedEarlierInPass.Count, + partition.ToRead.Count); progress.Publish(); // - // Everything the runtime filters out of these files is reported once for the whole data - // source. A run over a few thousand documents which removes something in forty of them - // is one thing that happened to the user, not forty. The scope ends with this method, so - // the report arrives when the run is finished rather than in the middle of it. + // Everything the runtime filters out of these files is reported once for the whole step. + // A step over a few thousand documents which removes something in forty of them is one + // thing that happened to the user, not forty. The scope ends with this method, so the + // report arrives when the step is over rather than in the middle of it. // await using var promptInjectionReportingScope = guardService.BeginAction(); - foreach (var file in indexedFiles) + for (var fileIndex = 0; fileIndex < partition.ToRead.Count; fileIndex++) { token.ThrowIfCancellationRequested(); + if (progress.HasDoneItsShare(DateTimeOffset.UtcNow)) + return await this.PauseAsync(dataSource, progress, refreshMode, partition.ToRead.Count - fileIndex, token); + var file = partition.ToRead[fileIndex]; var fingerprint = metadataSnapshot.FileHashes[file.FullName]; - if (manifest.Files.TryGetValue(file.FullName, out var existingRecord) && - string.Equals(existingRecord.Fingerprint, fingerprint, StringComparison.Ordinal)) - { - logger.LogDebug( - "Skipping unchanged file '{FilePath}' for data source '{DataSourceName}' ({DataSourceId}) because the persisted metadata hash matches. MetadataHashPrefix={MetadataHashPrefix}, LastWriteUtc={LastWriteUtc:O}, FileSize={FileSize}.", - file.FullName, - dataSource.Name, - dataSource.Id, - ShortHash(fingerprint), - file.LastWriteTimeUtc, - file.Length); - progress.RecordUnchanged(); - progress.Publish(); - continue; - } - - // - // A file which failed for a reason of its own is not read again until it changes. - // Without this, a folder holding hundreds of scanned documents without a text layer - // would spend half an hour on every start to arrive at the result we already have: - // - if (manifest.PermanentFailures.TryGetValue(file.FullName, out var permanentFailure) && - string.Equals(permanentFailure.Fingerprint, fingerprint, StringComparison.Ordinal)) - { - logger.LogDebug( - "Skipping file '{FilePath}' for data source '{DataSourceName}' ({DataSourceId}) because reading it failed permanently before. FailureCode={FailureCode}, MetadataHashPrefix={MetadataHashPrefix}, OccurredAtUtc={OccurredAtUtc:O}.", - file.FullName, - dataSource.Name, - dataSource.Id, - permanentFailure.Code, - ShortHash(fingerprint), - permanentFailure.OccurredAtUtc); - progress.RecordStillUnreadable(file.FullName, permanentFailure); - progress.Publish(); - continue; - } + var existingRecord = manifest.Files.GetValueOrDefault(file.FullName); var document = this.CreateFileDocument(context, dataSource, file, fingerprint); var reportBlockProgress = progress.BeginDocument(document); + var startedAtUtc = DateTimeOffset.UtcNow; try { @@ -182,7 +199,6 @@ internal sealed partial class FileSourceIndexer(SettingsManager settingsManager, ShortHash(fingerprint), progress.DoneDocuments + 1, totalFiles); - var startedAtUtc = DateTimeOffset.UtcNow; var chunkCount = await context.IndexDocumentAsync(document, reportBlockProgress, token); token.ThrowIfCancellationRequested(); var fingerprintAfterEmbedding = BuildFileMetadataHash(file); @@ -215,6 +231,21 @@ internal sealed partial class FileSourceIndexer(SettingsManager settingsManager, { await progress.RecordDocumentFailureAsync(document, exception, token); } + + // + // A file is read to its end, however long that takes: its text arrives as a stream from + // the runtime, which cannot pick up where it stopped. Every other data source waits + // for it, so it should at least show up in the log: + // + var fileDuration = DateTimeOffset.UtcNow - startedAtUtc; + if (fileDuration > context.Share.MaxDuration) + logger.LogWarning( + "Reading file '{FilePath}' of data source '{DataSourceName}' ({DataSourceId}) took {FileMinutes:F0} minutes, longer than the share of a whole run, which is {ShareMinutes:F0} minutes. Every other data source waited for it.", + file.FullName, + dataSource.Name, + dataSource.Id, + fileDuration.TotalMinutes, + context.Share.MaxDuration.TotalMinutes); } await progress.CompleteRunAsync(metadataSnapshot.SourceHash, "data source embedding run finished", token); @@ -236,6 +267,71 @@ internal sealed partial class FileSourceIndexer(SettingsManager settingsManager, return IndexedRunOutcome.DONE; } + /// + /// Sorts the files of a data source by what one run does with them. + /// + /// + /// A file which failed for a reason of its own is not read again until it changes. Without + /// that, a folder holding hundreds of scanned documents without a text layer would spend half + /// an hour on every start to arrive at the result we already have. A file which failed during + /// an earlier step of the pass waits for the next pass, unless it changed since, cf. IndexingPass. + /// + /// The files of the data source, in the order they are read. + /// The fingerprint of every one of them, by its full path. + /// What the index stores about the data source. + /// The pass the run is a step of. + /// The files, sorted. + internal static FilePartition PartitionFiles(IReadOnlyList files, IReadOnlyDictionary fingerprints, DataSourceEmbeddingManifest manifest, IndexingPass pass) + { + var unchangedCount = 0; + var stillUnreadable = new List>(); + var failedEarlierInPass = new List(); + var toRead = new List(); + + foreach (var file in files) + { + var fingerprint = fingerprints[file.FullName]; + if (manifest.Files.TryGetValue(file.FullName, out var existingRecord) && string.Equals(existingRecord.Fingerprint, fingerprint, StringComparison.Ordinal)) + unchangedCount++; + else if (manifest.PermanentFailures.TryGetValue(file.FullName, out var permanentFailure) && string.Equals(permanentFailure.Fingerprint, fingerprint, StringComparison.Ordinal)) + stillUnreadable.Add(new(file.FullName, permanentFailure)); + else if (pass.TryGetEarlierFailure(file.FullName, fingerprint, out var earlierFailure)) + failedEarlierInPass.Add(earlierFailure); + else + toRead.Add(file); + } + + return new(unchangedCount, stillUnreadable, failedEarlierInPass, toRead); + } + + /// + /// Ends a run which did its share of the work before it read every file. + /// + /// + /// Without a hash, cf. DocumentRunProgress.PauseRunAsync: a folder which is indexed for the + /// first time must not count as complete, and one which was complete before keeps the hash of + /// back then, which no longer matches, so the next run reads on. + /// + private async Task PauseAsync(IDataSource dataSource, DocumentRunProgress progress, DataSourceEmbeddingRefreshMode refreshMode, int filesLeftToRead, CancellationToken token) + { + var outcome = await progress.PauseRunAsync(null, "data source embedding run paused", token); + if (outcome is not IndexedRunOutcome.MORE_TO_DO) + return outcome; + + logger.LogInformation( + "Embedding data source '{DataSourceName}' ({DataSourceId}) carries on in another run. RefreshMode={RefreshMode}, Embedded={EmbeddedFiles}, Failed={FailedFiles}, Done={DoneFiles}/{TotalFiles}, LeftToRead={FilesLeftToRead}.", + dataSource.Name, + dataSource.Id, + refreshMode, + progress.IndexedDocuments, + progress.FailedDocuments, + progress.DoneDocuments, + progress.TotalDocuments, + filesLeftToRead); + + return IndexedRunOutcome.MORE_TO_DO; + } + /// /// Whether a file name marks a file which is never indexed, whatever its type. /// diff --git a/app/MindWork AI Studio/Tools/Services/Indexing/IIndexedSourceIndexer.cs b/app/MindWork AI Studio/Tools/Services/Indexing/IIndexedSourceIndexer.cs index 81e0c7fc..c976659c 100644 --- a/app/MindWork AI Studio/Tools/Services/Indexing/IIndexedSourceIndexer.cs +++ b/app/MindWork AI Studio/Tools/Services/Indexing/IIndexedSourceIndexer.cs @@ -24,13 +24,27 @@ internal interface IIndexedSourceIndexer : IDisposable /// True when this indexer is the one for it. public bool Supports(IDataSourceBase dataSource); + /// + /// How much work one run does at most, before it leaves the rest to another one. + /// + /// + /// All data sources share one queue, and only one run works at a time. A run which goes on for + /// hours holds up every other data source for as long, a mailbox whose sync is due among them. + /// So a run asks DocumentRunProgress.HasDoneItsShare before every document it is about to read, + /// and pauses through DocumentRunProgress.PauseRunAsync once the answer is yes. + /// + /// Nothing fails when a run forgets to ask. The embedding service therefore writes a warning into + /// the log when a run took more than twice its share without asking once. + /// + public RunShare Share { get; } + /// /// Works through the documents of the data source of a run, from finding them to completing the run. /// /// - /// A run may stop after its share of the work and leave the rest to another one, so a data source - /// which takes hours does not hold up all others for that long. Whether and when that run comes - /// is the embedding service's to decide. + /// A run stops after its share of the work, cf. Share, and leaves the rest to another one, so a + /// data source which takes hours does not hold up all others for that long. Whether and when that + /// run comes is the embedding service's to decide. /// /// The prepared run, whose data source this indexer supports. /// Why the run was started. diff --git a/app/MindWork AI Studio/Tools/Services/Indexing/IndexedRunContext.cs b/app/MindWork AI Studio/Tools/Services/Indexing/IndexedRunContext.cs index 656468fd..c854635c 100644 --- a/app/MindWork AI Studio/Tools/Services/Indexing/IndexedRunContext.cs +++ b/app/MindWork AI Studio/Tools/Services/Indexing/IndexedRunContext.cs @@ -24,10 +24,12 @@ namespace AIStudio.Tools.Services.Indexing; /// The vector store, known to be available. /// The index store, known to be available. /// What the index stores about the data source, made to match the current embedding configuration. +/// How much work the run does at most, as its kind of data source says. +/// The pass the run is a step of, which knows what failed during the steps before. /// The settings, which the provider reads when it embeds. /// Hands a new status of the data source to the embedding service. /// The logger of the embedding service, so the log reads the same whoever writes it. -internal sealed class IndexedRunContext(IIndexedDataSource dataSource, EmbeddingProvider embeddingProvider, IProvider provider, VectorStoreClient vectorStore, IndexStoreClient indexStore, DataSourceEmbeddingManifest manifest, SettingsManager settingsManager, Action publishStatus, ILogger logger) +internal sealed class IndexedRunContext(IIndexedDataSource dataSource, EmbeddingProvider embeddingProvider, IProvider provider, VectorStoreClient vectorStore, IndexStoreClient indexStore, DataSourceEmbeddingManifest manifest, RunShare share, IndexingPass pass, SettingsManager settingsManager, Action publishStatus, ILogger logger) { /// /// After how many stored chunks the collection is optimized while a run is still going. @@ -58,6 +60,26 @@ internal sealed class IndexedRunContext(IIndexedDataSource dataSource, Embedding public string CollectionName { get; } = DataSourceEmbeddingNames.GetCollectionName(dataSource.Id); + public RunShare Share => share; + + public IndexingPass Pass => pass; + + /// + /// When the run first asked whether it did its share of the work, or null while it never did. + /// + /// + /// The clock of the share starts then, cf. DocumentRunProgress.HasDoneItsShare. The embedding + /// service reads it once the run is over, to notice a kind of data source which never asks. + /// + public DateTimeOffset? ShareStartedUtc { get; private set; } + + /// + /// Starts the clock of the share of the work, unless it runs already. + /// + /// The current time. + /// When the clock started. + public DateTimeOffset StartShareClock(DateTimeOffset nowUtc) => this.ShareStartedUtc ??= nowUtc; + /// /// Tells the user interface about a new status of the data source. /// diff --git a/app/MindWork AI Studio/Tools/Services/Indexing/IndexingPass.cs b/app/MindWork AI Studio/Tools/Services/Indexing/IndexingPass.cs new file mode 100644 index 00000000..8239baef --- /dev/null +++ b/app/MindWork AI Studio/Tools/Services/Indexing/IndexingPass.cs @@ -0,0 +1,132 @@ +using System.Diagnostics.CodeAnalysis; + +using AIStudio.Provider; + +namespace AIStudio.Tools.Services.Indexing; + +/// +/// One way through a data source, from its first step to the step which gets through it. +/// +/// +/// A large data source is indexed in steps, each of them a run of its own, cf. RunShare. What one +/// step learns about documents which failed has to reach the next one, or every step reads them +/// again: that costs time, with a cloud provider money as well, and the list of failures would lose +/// whatever the steps before found. A pass keeps it for as long as the steps go on. +/// +/// The embedding service keeps a pass only while the last step of its data source left more to do. +/// Any other end of a run ends the pass, and so does a reset of the stored index. A manual retry +/// begins a new pass, as does a run for other embedding settings. +/// +/// Documents which failed for a reason of their own are none of its business: the index remembers +/// them for good, cf. PermanentIndexingFailure. Only the steps of one data source use a pass, one +/// after the other, so it needs no lock. +/// +/// The embedding settings the pass works for. +internal sealed class IndexingPass(DataSourceRunSignature runSignature) +{ + /// + /// How often a document is tried during one pass at most, whether it changed in between or not. + /// + /// + /// A file which changes faster than it can be read would otherwise be read in every step, e.g. + /// the log file of another program. + /// + private const int MAX_ATTEMPTS = 2; + + /// + /// A document which failed during this pass. + /// + /// The fingerprint of the document when it failed the last time. + /// How often it failed during this pass. + /// Why it failed the last time, as the list of failures shows it. + private sealed record FailedDocument(string Fingerprint, int Attempts, DataSourceEmbeddingFailure Failure); + + /// + /// The documents which failed during this pass for a reason which may pass, by their keys. + /// + private readonly Dictionary failedDocuments = new(StringComparer.OrdinalIgnoreCase); + + // + // Which kinds of provider failure the user was already told about in this pass. A rejected API + // key is the same problem for every one of a few thousand documents, and one message is what it + // takes to send the user to the settings. Kept for the whole pass, so a recurring failure such + // as a rate limit does not show up anew with every step. + // + private readonly HashSet reportedFailureReasons = []; + + public DataSourceRunSignature RunSignature => runSignature; + + /// + /// How many documents failed during this pass and are not tried again for now. + /// + public int FailedDocumentCount => this.failedDocuments.Count; + + /// + /// The pass a run works in: the one the last step of its data source left, or a new one. + /// + /// The pass the last step of the data source left, or null when there is none. + /// Why the run was asked for. + /// The embedding settings the run works for. + /// The pass of the run. + public static IndexingPass ContinueOrBegin(IndexingPass? current, DataSourceEmbeddingRefreshMode refreshMode, DataSourceRunSignature runSignature) + { + // + // A manual retry is the user asking for every failed document to be tried again, so it + // never carries on with what a pass knows. Neither does a run for other embedding settings: + // what failed with the old ones may well work with the new. + // + if (current is not null && refreshMode is not DataSourceEmbeddingRefreshMode.MANUAL_RETRY && current.RunSignature == runSignature) + return current; + + return new(runSignature); + } + + /// + /// Records that a document failed for a reason which may pass. + /// + /// The key of the document. + /// The fingerprint of the document as it was read. + /// Why it failed, as the list of failures shows it. + public void RecordFailure(string documentKey, string fingerprint, DataSourceEmbeddingFailure failure) + { + var attempts = this.failedDocuments.TryGetValue(documentKey, out var earlierFailure) ? earlierFailure.Attempts + 1 : 1; + this.failedDocuments[documentKey] = new(fingerprint, attempts, failure); + } + + /// + /// Forgets that a document failed, e.g. because it was indexed after all. + /// + /// The key of the document. + public void Forget(string documentKey) => this.failedDocuments.Remove(documentKey); + + /// + /// Whether a document failed during this pass and is not tried again before the next one. + /// + /// + /// A document which changed since it failed gets another try, as long as it did not fail + /// MAX_ATTEMPTS times during this pass already. + /// + /// The key of the document. + /// The fingerprint of the document as it is now. + /// Why it failed the last time, when it is not tried again. + /// True when the document is left alone. + public bool TryGetEarlierFailure(string documentKey, string fingerprint, [NotNullWhen(true)] out DataSourceEmbeddingFailure? failure) + { + failure = null; + if (!this.failedDocuments.TryGetValue(documentKey, out var failedDocument)) + return false; + + if (failedDocument.Attempts < MAX_ATTEMPTS && !string.Equals(failedDocument.Fingerprint, fingerprint, StringComparison.Ordinal)) + return false; + + failure = failedDocument.Failure; + return true; + } + + /// + /// Marks a kind of provider failure as reported to the user. + /// + /// The kind of provider failure. + /// True when the user was not told about it during this pass yet. + public bool TryMarkReported(ProviderRequestFailureReason reason) => this.reportedFailureReasons.Add(reason); +} \ No newline at end of file diff --git a/app/MindWork AI Studio/Tools/Services/Indexing/MailboxIndexer.Mails.cs b/app/MindWork AI Studio/Tools/Services/Indexing/MailboxIndexer.Mails.cs index 56b7a098..f1602346 100644 --- a/app/MindWork AI Studio/Tools/Services/Indexing/MailboxIndexer.Mails.cs +++ b/app/MindWork AI Studio/Tools/Services/Indexing/MailboxIndexer.Mails.cs @@ -69,6 +69,17 @@ internal sealed partial class MailboxIndexer return; } + // + // A mail which failed during an earlier step of this sync waits for the next sync, unless + // it changed since. Asked only here: the mail counts as encountered all the same, and one + // which merely moved was linked above. + // + if (progress.SkipIfFailedEarlierInPass(key, mailHash)) + { + progress.Publish(); + return; + } + // // Named after the subject as the server delivered it. The name is for the user alone, who // reads it on the embeddings page as in any mail program. A model only ever gets to read diff --git a/app/MindWork AI Studio/Tools/Services/Indexing/MailboxIndexer.cs b/app/MindWork AI Studio/Tools/Services/Indexing/MailboxIndexer.cs index 0907eb1f..ff38850e 100644 --- a/app/MindWork AI Studio/Tools/Services/Indexing/MailboxIndexer.cs +++ b/app/MindWork AI Studio/Tools/Services/Indexing/MailboxIndexer.cs @@ -28,10 +28,10 @@ namespace AIStudio.Tools.Services.Indexing; /// had its chance to link a mail which merely moved there. Even then, a mail is only orphaned, and /// deleted one run later, should it not turn up again by then. /// -/// A run reads a limited number of mails, for a limited time, and leaves the rest to the next run. -/// The first sync of a large mailbox takes hours, and the other data sources wait in the same -/// queue. Nothing is removed before a run got through the whole mailbox, and a removal of a large -/// part of it waits for the user to agree, cf. MailRemovalGuard. +/// A run reads a limited number of mails, for a limited time, cf. Share, and leaves the rest to the +/// next run. The first sync of a large mailbox takes hours, and the other data sources wait in the +/// same queue. Nothing is removed before a run got through the whole mailbox, and a removal of a +/// large part of it waits for the user to agree, cf. MailRemovalGuard. /// /// Signing in is guarded, because every refused attempt brings an account closer to being locked. /// A refused sign-in is stored, and no run signs in again until the user saves a new password, @@ -58,16 +58,6 @@ internal sealed partial class MailboxIndexer(SettingsManager settingsManager, Ru /// private const int SUMMARY_BATCH_SIZE = 100; - /// - /// How many mails one run reads at most, whether embedding them works or not. - /// - private const int MAX_READ_MAILS_PER_RUN = 500; - - /// - /// How long one run takes at most, before it leaves the rest to the next one. - /// - private static readonly TimeSpan MAX_RUN_DURATION = TimeSpan.FromMinutes(5); - /// /// How often every mailbox is synced while AI Studio runs. It goes by the clock, so a computer /// which slept longer than this syncs right after waking up. @@ -106,6 +96,13 @@ internal sealed partial class MailboxIndexer(SettingsManager settingsManager, Ru /// public bool Supports(IDataSourceBase dataSource) => dataSource is DataSourceMailbox; + /// + /// + /// Besides the time, a run reads at most 500 mails, whether embedding them works or not. + /// Linking a mail the index holds already costs next to nothing, so only the time limits that. + /// + public RunShare Share => new(RunShare.DEFAULT_DURATION, 500); + /// public async Task ProcessAsync(IndexedRunContext context, DataSourceEmbeddingRefreshMode refreshMode, CancellationToken token) { @@ -257,7 +254,7 @@ internal sealed partial class MailboxIndexer(SettingsManager settingsManager, Ru { foreach (var folder in plannedFolders) { - if (await this.SyncNewMailsAsync(context, mailbox, connector, folder, progress, encounteredKeys, runStartedUtc, token)) + if (await this.SyncNewMailsAsync(context, mailbox, connector, folder, progress, encounteredKeys, token)) continue; workedThrough = false; @@ -278,7 +275,7 @@ internal sealed partial class MailboxIndexer(SettingsManager settingsManager, Ru { logger.LogWarning("Syncing mailbox '{MailboxId}' would remove {RemovalCount} of {IndexedCount} location(s) from the index at once. The removal waits for the user to agree.", mailbox.Id, removals.Count, removals.IndexedCount); await context.IndexStore.HoldBackMailRemovalAsync(mailbox.Id, removals.Count, token); - await progress.PauseRunAsync(sourceHash, "mailbox sync waits for a removal to be approved", token); + await progress.StoreSourceHashAsync(sourceHash, "mailbox sync waits for a removal to be approved", token); progress.PublishRunFailure(GetRemovalMessage(removals.Count), DataSourceAttention.MASS_REMOVAL_PENDING, removals.Count); return IndexedRunOutcome.DONE; } @@ -344,21 +341,16 @@ internal sealed partial class MailboxIndexer(SettingsManager settingsManager, Ru /// Ends a run which did its share of the work before it got through the mailbox. /// /// - /// Only a run which got something done hands on to another one. When every mail it read - /// failed, e.g. because the embedding provider is down, the next run would fail all the same, - /// and the one after it, without end. Such a run ends with its failures instead, and waits for - /// the next time the mailbox is queued. + /// The hash goes along, so the mails indexed so far can be searched right away. Whether another + /// run carries on is decided by DocumentRunProgress.PauseRunAsync, the same for every kind of + /// data source. /// private async Task PauseAsync(DataSourceMailbox mailbox, DocumentRunProgress progress, string sourceHash, DataSourceEmbeddingRefreshMode refreshMode, CancellationToken token) { - if (progress.IndexedDocuments is 0 && progress.FailedDocuments > 0) - { - logger.LogWarning("Syncing mailbox '{MailboxId}' stops without carrying on, since none of the {FailedMails} mail(s) it read could be indexed.", mailbox.Id, progress.FailedDocuments); - await progress.CompleteRunAsync(sourceHash, "mailbox sync stopped without progress", token); - return IndexedRunOutcome.DONE; - } + var outcome = await progress.PauseRunAsync(sourceHash, "mailbox sync paused", token); + if (outcome is not IndexedRunOutcome.MORE_TO_DO) + return outcome; - await progress.PauseRunAsync(sourceHash, "mailbox sync paused", token); logger.LogInformation( "Syncing mailbox '{MailboxId}' carries on in another run. RefreshMode={RefreshMode}, Embedded={EmbeddedMails}, Failed={FailedMails}, Done={DoneMails}/{TotalMails}.", mailbox.Id, @@ -433,7 +425,7 @@ internal sealed partial class MailboxIndexer(SettingsManager settingsManager, Ru /// The second pass through one folder: links or indexes the mails the index does not hold there yet. /// /// True when the folder is through, false when the run did its share of the work before. - private async Task SyncNewMailsAsync(IndexedRunContext context, DataSourceMailbox mailbox, ImapMailboxConnector connector, PlannedFolder folder, DocumentRunProgress progress, ISet encounteredKeys, DateTimeOffset runStartedUtc, CancellationToken token) + private async Task SyncNewMailsAsync(IndexedRunContext context, DataSourceMailbox mailbox, ImapMailboxConnector connector, PlannedFolder folder, DocumentRunProgress progress, ISet encounteredKeys, CancellationToken token) { if (folder.Plan.NewUids.Count > 0) { @@ -451,13 +443,13 @@ internal sealed partial class MailboxIndexer(SettingsManager settingsManager, Ru foreach (var uidBatch in folder.Plan.NewUids.Chunk(SUMMARY_BATCH_SIZE)) { token.ThrowIfCancellationRequested(); - if (HasDoneItsShare(progress, runStartedUtc)) + if (progress.HasDoneItsShare(DateTimeOffset.UtcNow)) return false; var summaries = await connector.FetchSummariesAsync(uidBatch, token); foreach (var summary in summaries.OrderByDescending(summary => summary.UniqueId.Id)) { - if (HasDoneItsShare(progress, runStartedUtc)) + if (progress.HasDoneItsShare(DateTimeOffset.UtcNow)) return false; await this.SyncNewMailAsync(context, mailbox, connector, folder.Folder.FullName, summary, progress, encounteredKeys, token); @@ -479,16 +471,6 @@ internal sealed partial class MailboxIndexer(SettingsManager settingsManager, Ru return true; } - /// - /// Whether a run did its share of the work, so other data sources get their turn. - /// - /// - /// Counted are the mails which were read, whether embedding them worked or not. Linking a mail - /// the index holds already costs next to nothing, so only the time limits that. - /// - private static bool HasDoneItsShare(DocumentRunProgress progress, DateTimeOffset runStartedUtc) => - progress.IndexedDocuments + progress.FailedDocuments >= MAX_READ_MAILS_PER_RUN || DateTimeOffset.UtcNow - runStartedUtc >= MAX_RUN_DURATION; - /// /// Works out what a run which got through the whole mailbox would remove. /// diff --git a/app/MindWork AI Studio/Tools/Services/Indexing/RunShare.cs b/app/MindWork AI Studio/Tools/Services/Indexing/RunShare.cs new file mode 100644 index 00000000..4dee1746 --- /dev/null +++ b/app/MindWork AI Studio/Tools/Services/Indexing/RunShare.cs @@ -0,0 +1,20 @@ +namespace AIStudio.Tools.Services.Indexing; + +/// +/// How much work one indexing run does at most, before it leaves the rest to another run. +/// +/// +/// Only the work on documents counts, cf. DocumentRunProgress.HasDoneItsShare: the clock starts right +/// before the first one. Whatever a kind of data source does before, e.g. listing the files of a +/// folder, every run has to do anew, and counting it would leave a large folder little time for its +/// documents. +/// +/// How long a run works on documents at most. More than zero. +/// How many documents a run reads at most, whether that works or not. Null when only the time counts. +internal sealed record RunShare(TimeSpan MaxDuration, int? MaxDocuments = null) +{ + /// + /// How long a run works on documents, unless its kind of data source has a reason for another time. + /// + public static readonly TimeSpan DEFAULT_DURATION = TimeSpan.FromMinutes(5); +} \ No newline at end of file diff --git a/app/MindWork AI Studio/wwwroot/changelog/v26.10.1.md b/app/MindWork AI Studio/wwwroot/changelog/v26.10.1.md index 3024f9aa..1d494a13 100644 --- a/app/MindWork AI Studio/wwwroot/changelog/v26.10.1.md +++ b/app/MindWork AI Studio/wwwroot/changelog/v26.10.1.md @@ -15,6 +15,7 @@ - Improved the button in the chat for moving the chat to a workspace. You can now use it whenever the workspace list lets you move the chat as well, for example, right after you attached an audio or video file and before you sent your first message. - Improved how AI Studio keeps the message you have not sent yet. When AI Studio stores your chats automatically, your message and its attachments now stay with the chat, and they are back when you open the chat again, for example, after you switched to another chat. - Improved how AI Studio keeps the message of a new chat. When AI Studio stores your chats automatically and you leave a new chat before you send your message, AI Studio now stores the chat with your message. You find it among your disappearing chats, or in its workspace when you started it there. +- Improved how AI Studio indexes large local data sources. It now works through them in steps of a few minutes, so your other data sources, such as your mailboxes, stay up to date in between. - Changed the banner on the start page to a fresh design that shows what AI Studio can do today. - Fixed a warning about the confidence level that stayed in the dialog for adding a local data source, even after you chose an embedding provider that meets it. The dialog now checks again whenever you change either of them, so a warning also appears when you switch to a provider that falls short. - Fixed two confidence levels sharing the same name in the German user interface. Moderate is called "Mäßig" again, and Medium stays "Mittel". diff --git a/app/Tests/Tools/DataSourceRunQueueTests.cs b/app/Tests/Tools/DataSourceRunQueueTests.cs new file mode 100644 index 00000000..59f5bae2 --- /dev/null +++ b/app/Tests/Tools/DataSourceRunQueueTests.cs @@ -0,0 +1,313 @@ +using AIStudio.Tools.Services; + +namespace AIStudio.Tests.Tools; + +/// +/// Checks which runs the queue of the indexing runs hands out, and why each of them happens. +/// +/// +/// File watchers, intervals and clicks ask for runs, often many at once for the same data source. A +/// data source which runs twice embeds everything twice, at a cloud embedding provider for real +/// money. One which never runs again goes out of date without anybody noticing. A run in the wrong +/// mode is the subtle case: only a manual retry signs in to a mailbox again after its server refused +/// a sign-in. The order counts as well: a mailbox which waits behind a large folder misses its new +/// mails for hours. +/// +[TestFixture] +public sealed class DataSourceRunQueueTests +{ + private const string DATA_SOURCE_ID = "6f1d6a4e-6a5e-4c62-9a4f-0f2d2c8b7a11"; + + private const string OTHER_DATA_SOURCE_ID = "5a1e7c3d-9b2f-4d8e-a6c4-3f0b9d2e1c75"; + + private const string MAILBOX_ID = "0c8e2b4f-3d7a-4e19-b5c6-8a2f1d9e7b30"; + + [Test] + public async Task RequestsForAWaitingDataSourceMergeIntoOneRun() + { + var queue = new DataSourceRunQueue(); + + var first = await Request(queue, DataSourceEmbeddingRefreshMode.WATCHER_HASH_CHECK); + var second = await Request(queue, DataSourceEmbeddingRefreshMode.HASH_CHECK); + + Assert.Multiple(() => + { + Assert.That(first, Is.EqualTo(DataSourceQueueRequestResult.QUEUED)); + Assert.That(second, Is.EqualTo(DataSourceQueueRequestResult.ALREADY_QUEUED)); + Assert.That(DequeueAll(queue), Is.EqualTo(new[] { new DataSourceRunRequest(DATA_SOURCE_ID, DataSourceEmbeddingRefreshMode.WATCHER_HASH_CHECK) }), "The first request decides why the run happens."); + }); + } + + [Test] + public async Task RequestsDuringARunLeaveExactlyOneFollowUpRun() + { + var queue = new DataSourceRunQueue(); + await Request(queue, DataSourceEmbeddingRefreshMode.HASH_CHECK); + Assert.That(queue.TryDequeue(out _), Is.True); + + var first = await Request(queue, DataSourceEmbeddingRefreshMode.WATCHER_HASH_CHECK); + var second = await Request(queue, DataSourceEmbeddingRefreshMode.CONTINUATION); + var requeue = queue.TryComplete(DATA_SOURCE_ID, true, out var pendingRefreshMode); + if (requeue) + await queue.EnqueueAsync(DATA_SOURCE_ID, DataSourceQueueLane.FRONT); + + Assert.Multiple(() => + { + Assert.That(first, Is.EqualTo(DataSourceQueueRequestResult.RUNNING_MARKED_PENDING)); + Assert.That(second, Is.EqualTo(DataSourceQueueRequestResult.RUNNING)); + Assert.That(requeue, Is.True); + Assert.That(pendingRefreshMode, Is.EqualTo(DataSourceEmbeddingRefreshMode.WATCHER_HASH_CHECK), "The first request decides why the follow-up run happens."); + Assert.That(DequeueAll(queue), Is.EqualTo(new[] { new DataSourceRunRequest(DATA_SOURCE_ID, DataSourceEmbeddingRefreshMode.WATCHER_HASH_CHECK) }), "The running run may have read the data source before the changes, so one more run is needed, and one is enough."); + }); + } + + [Test] + public async Task ARunWithoutRequestsLeavesNoFollowUpRun() + { + var queue = new DataSourceRunQueue(); + await Request(queue, DataSourceEmbeddingRefreshMode.HASH_CHECK); + queue.TryDequeue(out _); + + var requeue = queue.TryComplete(DATA_SOURCE_ID, true, out _); + var nextRequest = queue.TryReserve(DATA_SOURCE_ID, true, DataSourceEmbeddingRefreshMode.HASH_CHECK); + + Assert.Multiple(() => + { + Assert.That(requeue, Is.False); + Assert.That(nextRequest, Is.EqualTo(DataSourceQueueRequestResult.QUEUED), "The data source must not count as running after its run completed."); + }); + } + + [Test] + public async Task AFollowUpRunOfADataSourceWhichIsGoneIsDropped() + { + var queue = new DataSourceRunQueue(); + await Request(queue, DataSourceEmbeddingRefreshMode.HASH_CHECK); + queue.TryDequeue(out _); + await Request(queue, DataSourceEmbeddingRefreshMode.WATCHER_HASH_CHECK); + + var requeue = queue.TryComplete(DATA_SOURCE_ID, false, out _); + var nextRequest = queue.TryReserve(DATA_SOURCE_ID, true, DataSourceEmbeddingRefreshMode.HASH_CHECK); + + Assert.Multiple(() => + { + Assert.That(requeue, Is.False); + Assert.That(nextRequest, Is.EqualTo(DataSourceQueueRequestResult.QUEUED), "Neither the run nor its follow-up run may linger."); + }); + } + + [Test] + public async Task ADataSourceClearedWhileItWaitedLeavesNoRunBehind() + { + var queue = new DataSourceRunQueue(); + await Request(queue, DataSourceEmbeddingRefreshMode.HASH_CHECK); + + queue.Clear(DATA_SOURCE_ID); + + Assert.That(queue.TryDequeue(out _), Is.False, "The data source was deleted while it waited. Its id is still in the channel, yet it must not run."); + } + + [Test] + public async Task ARepairedDataSourceRunsOnceAndInTheModeOfTheRepair() + { + var queue = new DataSourceRunQueue(); + await Request(queue, DataSourceEmbeddingRefreshMode.HASH_CHECK); + + // What the repair does: forget the waiting run, then ask for a manual retry. + queue.Clear(DATA_SOURCE_ID); + await Request(queue, DataSourceEmbeddingRefreshMode.MANUAL_RETRY); + + Assert.That(DequeueAll(queue), Is.EqualTo(new[] { new DataSourceRunRequest(DATA_SOURCE_ID, DataSourceEmbeddingRefreshMode.MANUAL_RETRY) }), "The id is in the channel twice now. Running both copies would embed the data source twice, and a run in the old mode would not sign in to a mailbox again."); + } + + [Test] + public async Task AManualRetryReplacesTheModeOfAWaitingRun() + { + var queue = new DataSourceRunQueue(); + await Request(queue, DataSourceEmbeddingRefreshMode.INTERVAL_CHECK); + + var retry = await Request(queue, DataSourceEmbeddingRefreshMode.MANUAL_RETRY); + var later = await Request(queue, DataSourceEmbeddingRefreshMode.TOOL_REQUEST); + + Assert.Multiple(() => + { + Assert.That(retry, Is.EqualTo(DataSourceQueueRequestResult.ALREADY_QUEUED)); + Assert.That(later, Is.EqualTo(DataSourceQueueRequestResult.ALREADY_QUEUED)); + Assert.That(DequeueAll(queue), Is.EqualTo(new[] { new DataSourceRunRequest(DATA_SOURCE_ID, DataSourceEmbeddingRefreshMode.MANUAL_RETRY) }), "Only a manual retry signs in to a mailbox again after its server refused a sign-in, and no later request may take that back."); + }); + } + + [Test] + public async Task AManualRetryReplacesTheModeOfAFollowUpRun() + { + var queue = new DataSourceRunQueue(); + await Request(queue, DataSourceEmbeddingRefreshMode.INTERVAL_CHECK); + queue.TryDequeue(out _); + + await Request(queue, DataSourceEmbeddingRefreshMode.INTERVAL_CHECK); + await Request(queue, DataSourceEmbeddingRefreshMode.MANUAL_RETRY); + await Request(queue, DataSourceEmbeddingRefreshMode.TOOL_REQUEST); + queue.TryComplete(DATA_SOURCE_ID, true, out var pendingRefreshMode); + + Assert.That(pendingRefreshMode, Is.EqualTo(DataSourceEmbeddingRefreshMode.MANUAL_RETRY), "Only a manual retry signs in to a mailbox again after its server refused a sign-in, and no later request may take that back."); + } + + [Test] + public async Task ARunOfTheStartupCheckGetsExactlyOneFollowUpRun() + { + var queue = new DataSourceRunQueue(); + var begun = queue.TryBeginRun(DATA_SOURCE_ID); + + var first = await Request(queue, DataSourceEmbeddingRefreshMode.CONTINUATION); + var second = await Request(queue, DataSourceEmbeddingRefreshMode.WATCHER_HASH_CHECK); + var requeue = queue.TryComplete(DATA_SOURCE_ID, true, out var pendingRefreshMode); + if (requeue) + await queue.EnqueueAsync(DATA_SOURCE_ID, DataSourceQueueLane.BACK); + + Assert.Multiple(() => + { + Assert.That(begun, Is.True); + Assert.That(first, Is.EqualTo(DataSourceQueueRequestResult.RUNNING_MARKED_PENDING), "The run began outside the queue, yet it runs all the same."); + Assert.That(second, Is.EqualTo(DataSourceQueueRequestResult.RUNNING)); + Assert.That(requeue, Is.True); + Assert.That(pendingRefreshMode, Is.EqualTo(DataSourceEmbeddingRefreshMode.CONTINUATION)); + Assert.That(DequeueAll(queue), Is.EqualTo(new[] { new DataSourceRunRequest(DATA_SOURCE_ID, DataSourceEmbeddingRefreshMode.CONTINUATION) }), "A data source which did its share of the work during the startup check must carry on afterward, exactly once."); + }); + } + + [Test] + public async Task TheStartupCheckLeavesOutADataSourceWhichRunsOrWaits() + { + var queue = new DataSourceRunQueue(); + await Request(queue, DataSourceEmbeddingRefreshMode.HASH_CHECK); + var running = queue.TryBeginRun(OTHER_DATA_SOURCE_ID); + + Assert.Multiple(() => + { + Assert.That(running, Is.True); + Assert.That(queue.TryBeginRun(DATA_SOURCE_ID), Is.False, "The data source waits in the queue, so it would run twice in a row."); + Assert.That(queue.TryBeginRun(OTHER_DATA_SOURCE_ID), Is.False, "The data source runs already, so it would run twice at once."); + }); + } + + [Test] + public async Task WaitingForTheNextRunSkipsWhatWasCleared() + { + var queue = new DataSourceRunQueue(); + await Request(queue, DataSourceEmbeddingRefreshMode.HASH_CHECK); + queue.Clear(DATA_SOURCE_ID); + + var nextRun = queue.DequeueAsync(CancellationToken.None).AsTask(); + var waitedForTheNextRun = !nextRun.IsCompleted; + await Request(queue, DataSourceEmbeddingRefreshMode.HASH_CHECK, OTHER_DATA_SOURCE_ID); + var run = await nextRun.WaitAsync(TimeSpan.FromSeconds(10)); + + Assert.Multiple(() => + { + Assert.That(waitedForTheNextRun, Is.True, "Nothing was left to run."); + Assert.That(run, Is.EqualTo(new DataSourceRunRequest(OTHER_DATA_SOURCE_ID, DataSourceEmbeddingRefreshMode.HASH_CHECK))); + }); + } + + [Test] + public async Task AMailboxWaitsForOneStepAtMost() + { + var queue = new DataSourceRunQueue(); + await Request(queue, DataSourceEmbeddingRefreshMode.CONTINUATION, DATA_SOURCE_ID, DataSourceQueueLane.BACK); + await Request(queue, DataSourceEmbeddingRefreshMode.CONTINUATION, OTHER_DATA_SOURCE_ID, DataSourceQueueLane.BACK); + Assert.That(queue.TryDequeue(out var step), Is.True); + + // The interval of the mailbox comes due while a large folder does its step: + await Request(queue, DataSourceEmbeddingRefreshMode.INTERVAL_CHECK, MAILBOX_ID); + await CarryOn(queue, step!.DataSourceId); + + Assert.Multiple(() => + { + Assert.That(step.DataSourceId, Is.EqualTo(DATA_SOURCE_ID)); + Assert.That(DequeueAll(queue).Select(run => run.DataSourceId), Is.EqualTo(new[] { MAILBOX_ID, OTHER_DATA_SOURCE_ID, DATA_SOURCE_ID }), "The mailbox must not wait behind every step of every large data source, which would add up to hours."); + }); + } + + [Test] + public async Task DataSourcesWhichCarryOnTakeTurns() + { + var queue = new DataSourceRunQueue(); + await Request(queue, DataSourceEmbeddingRefreshMode.CONTINUATION, DATA_SOURCE_ID, DataSourceQueueLane.BACK); + await Request(queue, DataSourceEmbeddingRefreshMode.CONTINUATION, OTHER_DATA_SOURCE_ID, DataSourceQueueLane.BACK); + + var steps = new List(); + for (var step = 0; step < 4; step++) + { + Assert.That(queue.TryDequeue(out var runRequest), Is.True); + steps.Add(runRequest!.DataSourceId); + await CarryOn(queue, runRequest.DataSourceId); + } + + Assert.That(steps, Is.EqualTo(new[] { DATA_SOURCE_ID, OTHER_DATA_SOURCE_ID, DATA_SOURCE_ID, OTHER_DATA_SOURCE_ID }), "A data source which takes hours must not hold up another one which takes hours as well."); + } + + [Test] + public async Task ADataSourceRepairedWhileItCarriedOnRunsOnceAndAtTheFront() + { + var queue = new DataSourceRunQueue(); + await Request(queue, DataSourceEmbeddingRefreshMode.CONTINUATION, DATA_SOURCE_ID, DataSourceQueueLane.BACK); + await Request(queue, DataSourceEmbeddingRefreshMode.CONTINUATION, OTHER_DATA_SOURCE_ID, DataSourceQueueLane.BACK); + + // What the repair does: forget the waiting run, which also drops the pass, then ask for a manual retry. + queue.Clear(DATA_SOURCE_ID); + await Request(queue, DataSourceEmbeddingRefreshMode.MANUAL_RETRY); + + Assert.That(DequeueAll(queue), Is.EqualTo(new[] { new DataSourceRunRequest(DATA_SOURCE_ID, DataSourceEmbeddingRefreshMode.MANUAL_RETRY), new DataSourceRunRequest(OTHER_DATA_SOURCE_ID, DataSourceEmbeddingRefreshMode.CONTINUATION) }), "The copy left behind in the back lane must not run the data source a second time."); + } + + [Test] + public async Task ACopyLeftBehindInTheFrontLaneLetsNoLaterRunOvertake() + { + var queue = new DataSourceRunQueue(); + await Request(queue, DataSourceEmbeddingRefreshMode.WATCHER_HASH_CHECK); + await Request(queue, DataSourceEmbeddingRefreshMode.CONTINUATION, OTHER_DATA_SOURCE_ID, DataSourceQueueLane.BACK); + + // The repair of a data source which waits in the front lane leaves a copy of its id behind there: + queue.Clear(DATA_SOURCE_ID); + await Request(queue, DataSourceEmbeddingRefreshMode.MANUAL_RETRY); + Assert.That(queue.TryDequeue(out var repair), Is.True); + await CarryOn(queue, repair!.DataSourceId); + + Assert.Multiple(() => + { + Assert.That(repair, Is.EqualTo(new DataSourceRunRequest(DATA_SOURCE_ID, DataSourceEmbeddingRefreshMode.MANUAL_RETRY))); + Assert.That(DequeueAll(queue).Select(run => run.DataSourceId), Is.EqualTo(new[] { OTHER_DATA_SOURCE_ID, DATA_SOURCE_ID }), "The next step of the repaired data source carries on its pass, so it waits its turn in the back lane."); + }); + } + + /// + /// Asks for a run the way DataSourceEmbeddingService does: reserve a place, and enqueue the data source when it got one. + /// + private static async Task Request(DataSourceRunQueue queue, DataSourceEmbeddingRefreshMode refreshMode, string dataSourceId = DATA_SOURCE_ID, DataSourceQueueLane lane = DataSourceQueueLane.FRONT) + { + var result = queue.TryReserve(dataSourceId, true, refreshMode); + if (result is DataSourceQueueRequestResult.QUEUED) + await queue.EnqueueAsync(dataSourceId, lane); + + return result; + } + + /// + /// Ends the step of a running data source the way DataSourceEmbeddingService does when the step left more to do. + /// + private static async Task CarryOn(DataSourceRunQueue queue, string dataSourceId) + { + await Request(queue, DataSourceEmbeddingRefreshMode.CONTINUATION, dataSourceId); + if (queue.TryComplete(dataSourceId, true, out _)) + await queue.EnqueueAsync(dataSourceId, DataSourceQueueLane.BACK); + } + + private static List DequeueAll(DataSourceRunQueue queue) + { + var runs = new List(); + while (queue.TryDequeue(out var runRequest)) + runs.Add(runRequest); + + return runs; + } +} \ No newline at end of file diff --git a/app/Tests/Tools/DocumentRunProgressTests.cs b/app/Tests/Tools/DocumentRunProgressTests.cs index bb359ec2..36194f33 100644 --- a/app/Tests/Tools/DocumentRunProgressTests.cs +++ b/app/Tests/Tools/DocumentRunProgressTests.cs @@ -7,6 +7,7 @@ using AIStudio.Tools.Databases.VectorStore; using AIStudio.Tools.Services; using AIStudio.Tools.Services.Indexing; +using Microsoft.AspNetCore.Components; using Microsoft.Extensions.Logging.Abstractions; namespace AIStudio.Tests.Tools; @@ -27,6 +28,8 @@ public sealed class DocumentRunProgressTests private const string EMBEDDING_ID = "b0a4c4d2-1f3e-4f0a-8c9d-5a6b7c8d9e01"; + private static readonly DateTimeOffset NOW = new(2026, 10, 11, 9, 0, 0, TimeSpan.Zero); + [Test] public async Task AProviderFailureKeepsTheDocumentsSkippedForGoodInTheStatus() { @@ -100,15 +103,234 @@ public sealed class DocumentRunProgressTests var (progress, statuses, manifest) = CreateProgress(totalDocuments: 10); progress.RecordUnchanged(4); - await progress.PauseRunAsync("SOURCE-HASH", "test", CancellationToken.None); + var outcome = await progress.PauseRunAsync("SOURCE-HASH", "test", CancellationToken.None); Assert.Multiple(() => { + Assert.That(outcome, Is.EqualTo(IndexedRunOutcome.MORE_TO_DO)); Assert.That(manifest.SourceHash, Is.EqualTo("SOURCE-HASH"), "Without the hash, a mailbox stays out of every search for the hours its first sync takes."); Assert.That(statuses, Is.Empty, "The status said the run is over, while the next one is about to carry on."); }); } + [Test] + public async Task APausedRunWithoutAHashLeavesTheStoredOneAsItIs() + { + var (progress, statuses, manifest) = CreateProgress(totalDocuments: 10); + manifest.SourceHash = "HASH-OF-THE-LAST-COMPLETE-RUN"; + + progress.RecordUnchanged(4); + var outcome = await progress.PauseRunAsync(null, "test", CancellationToken.None); + + Assert.Multiple(() => + { + Assert.That(outcome, Is.EqualTo(IndexedRunOutcome.MORE_TO_DO)); + Assert.That(manifest.SourceHash, Is.EqualTo("HASH-OF-THE-LAST-COMPLETE-RUN"), "The hash says that a run got through the whole folder. A half-indexed folder would count as complete, and the next run would skip it."); + Assert.That(statuses, Is.Empty); + }); + } + + [Test] + public async Task APauseAfterTheProviderFailedForEveryDocumentEndsTheRun() + { + var (progress, statuses, manifest) = CreateProgress(totalDocuments: 10); + var providerDown = new ProviderRequestException(ProviderRequestFailureReason.PROVIDER_UNAVAILABLE, "The embedding provider cannot be reached."); + + await progress.RecordDocumentFailureAsync(Document("/tmp/test-data/a.pdf"), providerDown, CancellationToken.None); + await progress.RecordDocumentFailureAsync(Document("/tmp/test-data/b.pdf"), providerDown, CancellationToken.None); + var outcome = await progress.PauseRunAsync(null, "test", CancellationToken.None); + + Assert.Multiple(() => + { + Assert.That(outcome, Is.EqualTo(IndexedRunOutcome.DONE), "The next run would fail all the same, and the one after it, without end."); + Assert.That(statuses[^1].State, Is.EqualTo(DataSourceEmbeddingState.FAILED), "The run is over, so its row has to say so, and why."); + Assert.That(statuses[^1].LastError, Is.EqualTo("The embedding provider cannot be reached.")); + Assert.That(manifest.SourceHash, Is.Empty, "A run which stopped halfway did not get through the folder."); + }); + } + + [Test] + public async Task DocumentsSkippedForGoodDoNotKeepARunWithAFailingProviderGoing() + { + var (progress, _, _) = CreateProgress(totalDocuments: 10); + + await progress.RecordDocumentFailureAsync(Document("/tmp/test-data/scan.pdf"), new FileExtractionException(FileExtractionErrorCode.NO_TEXT_EXTRACTED, "No text."), CancellationToken.None); + await progress.RecordDocumentFailureAsync(Document("/tmp/test-data/report.pdf"), new ProviderRequestException(ProviderRequestFailureReason.PROVIDER_UNAVAILABLE, "The embedding provider cannot be reached."), CancellationToken.None); + var outcome = await progress.PauseRunAsync(null, "test", CancellationToken.None); + + Assert.That(outcome, Is.EqualTo(IndexedRunOutcome.DONE), "A scanned document never gets as far as the provider, so it says nothing about whether the provider works."); + } + + [Test] + public async Task APauseAfterFailuresOfTheDocumentsThemselvesCarriesOn() + { + var (progress, statuses, _) = CreateProgress(totalDocuments: 10); + + await progress.RecordDocumentFailureAsync(Document("/tmp/test-data/a.pdf"), new IOException("The file changed."), CancellationToken.None); + await progress.RecordDocumentFailureAsync(Document("/tmp/test-data/b.pdf"), new ProviderRequestException(ProviderRequestFailureReason.TOO_MANY_REQUESTS, "Too many requests."), CancellationToken.None); + var outcome = await progress.PauseRunAsync(null, "test", CancellationToken.None); + + Assert.Multiple(() => + { + Assert.That(outcome, Is.EqualTo(IndexedRunOutcome.MORE_TO_DO), "The pass skips both documents from now on, so the next step gets further. Ending here left every file behind a block of failing files unindexed."); + Assert.That(statuses.Select(status => status.State), Has.None.EqualTo(DataSourceEmbeddingState.FAILED)); + }); + } + + [Test] + public async Task InputsWhichCouldNotBeLookedAtDoNotEndAPausedRun() + { + var (progress, statuses, _) = CreateProgress(totalDocuments: 10, failedInputs: 1); + + // A step which spent its share on a single scanned document without a text layer: + await progress.RecordDocumentFailureAsync(Document("/tmp/test-data/scan.pdf"), new FileExtractionException(FileExtractionErrorCode.NO_TEXT_EXTRACTED, "No text."), CancellationToken.None); + var outcome = await progress.PauseRunAsync(null, "test", CancellationToken.None); + + Assert.Multiple(() => + { + Assert.That(outcome, Is.EqualTo(IndexedRunOutcome.MORE_TO_DO), "A locked subfolder fails in every run, however much the run gets done otherwise, and the scanned document is never read again. Ending here left the rest of the folder unindexed."); + Assert.That(statuses.Select(status => status.State), Has.None.EqualTo(DataSourceEmbeddingState.FAILED)); + }); + } + + [Test] + public void TheShareCountsFromTheFirstQuestion() + { + var (progress, _, _) = CreateProgress(totalDocuments: 10, share: new RunShare(TimeSpan.FromMinutes(5))); + + var first = progress.HasDoneItsShare(NOW); + var shortlyBefore = progress.HasDoneItsShare(NOW + TimeSpan.FromMinutes(5) - TimeSpan.FromSeconds(1)); + var atTheEnd = progress.HasDoneItsShare(NOW + TimeSpan.FromMinutes(5)); + + Assert.Multiple(() => + { + Assert.That(first, Is.False, "However long finding the documents took, every run reads at least one. Otherwise, a folder whose files take longer to list than the share would never get anywhere."); + Assert.That(shortlyBefore, Is.False); + Assert.That(atTheEnd, Is.True); + }); + } + + [Test] + public async Task TheShareEndsAfterItsDocumentsWhetherIndexingThemWorkedOrNot() + { + var (progress, _, _) = CreateProgress(totalDocuments: 10, share: new RunShare(RunShare.DEFAULT_DURATION, 3)); + progress.HasDoneItsShare(NOW); + + await progress.RecordDocumentIndexedAsync(Document("/tmp/test-data/a.pdf"), 4, true, CancellationToken.None); + await progress.RecordDocumentFailureAsync(Document("/tmp/test-data/b.pdf"), new IOException("The file changed."), CancellationToken.None); + var afterTwo = progress.HasDoneItsShare(NOW); + await progress.RecordDocumentFailureAsync(Document("/tmp/test-data/scan.pdf"), new FileExtractionException(FileExtractionErrorCode.NO_TEXT_EXTRACTED, "No text."), CancellationToken.None); + var afterThree = progress.HasDoneItsShare(NOW); + + Assert.Multiple(() => + { + Assert.That(afterTwo, Is.False); + Assert.That(afterThree, Is.True, "Reading a document costs its time, whether indexing it worked or not."); + }); + } + + [Test] + public void DocumentsWhichWereNotReadDoNotCountTowardsTheShare() + { + var (progress, _, _) = CreateProgress(totalDocuments: 1_000, share: new RunShare(RunShare.DEFAULT_DURATION, 3)); + + progress.RecordUnchanged(900); + progress.RecordStillUnreadable("/tmp/test-data/scan.pdf", new PermanentIndexingFailureRecord("AAAA", FileExtractionErrorCode.NO_TEXT_EXTRACTED, "No text could be read.", DateTimeOffset.UtcNow)); + + Assert.That(progress.HasDoneItsShare(NOW), Is.False, "Leaving a document alone costs next to nothing. Counting it would end the step of a mailbox after its first few mails."); + } + + [Test] + public async Task AFailureOfAnEarlierStepKeepsItsPlaceInTheNextOne() + { + var pass = NewPass(); + var (firstStep, _, _) = CreateProgress(totalDocuments: 10, pass: pass); + await firstStep.RecordDocumentFailureAsync(Document("/tmp/test-data/report.pdf"), new IOException("The file changed."), CancellationToken.None); + + var (nextStep, statuses, _) = CreateProgress(totalDocuments: 10, pass: pass); + var skipped = nextStep.SkipIfFailedEarlierInPass("/tmp/test-data/report.pdf", "FINGERPRINT"); + nextStep.Publish(); + + Assert.Multiple(() => + { + Assert.That(skipped, Is.True, "Reading it once more costs time, with a cloud provider money as well, and ends the same way."); + Assert.That(statuses[^1].FailedDocuments, Is.EqualTo(1)); + Assert.That(statuses[^1].Failures.Select(failure => failure.DocumentKey), Is.EquivalentTo(["/tmp/test-data/report.pdf"]), "The list lost the failures of every earlier step."); + Assert.That(statuses[^1].LastError, Is.Not.Empty); + }); + } + + [Test] + public async Task ADocumentWhichChangedSinceItFailedIsTriedAgain() + { + var pass = NewPass(); + var (firstStep, _, _) = CreateProgress(totalDocuments: 10, pass: pass); + await firstStep.RecordDocumentFailureAsync(Document("/tmp/test-data/report.pdf"), new IOException("The file changed."), CancellationToken.None); + + var (nextStep, _, _) = CreateProgress(totalDocuments: 10, pass: pass); + + Assert.That(nextStep.SkipIfFailedEarlierInPass("/tmp/test-data/report.pdf", "CHANGED-FINGERPRINT"), Is.False, "The change may well be what makes it readable, e.g. the program which wrote it is done now."); + } + + [Test] + public async Task ADocumentIsTriedTwiceDuringAPassAtMost() + { + var pass = NewPass(); + var (firstStep, _, _) = CreateProgress(totalDocuments: 10, pass: pass); + await firstStep.RecordDocumentFailureAsync(Document("/tmp/test-data/app.log", "app.log", "FINGERPRINT-1"), new IOException("The file changed."), CancellationToken.None); + var (secondStep, _, _) = CreateProgress(totalDocuments: 10, pass: pass); + await secondStep.RecordDocumentFailureAsync(Document("/tmp/test-data/app.log", "app.log", "FINGERPRINT-2"), new IOException("The file changed."), CancellationToken.None); + + var (thirdStep, _, _) = CreateProgress(totalDocuments: 10, pass: pass); + + Assert.That(thirdStep.SkipIfFailedEarlierInPass("/tmp/test-data/app.log", "FINGERPRINT-3"), Is.True, "A file which changes faster than it can be read would be read in every step, and the pass would never end."); + } + + [Test] + public async Task ADocumentWhichWasIndexedAfterAllIsNoLongerSkipped() + { + var pass = NewPass(); + var (firstStep, _, _) = CreateProgress(totalDocuments: 10, pass: pass); + await firstStep.RecordDocumentFailureAsync(Document("/tmp/test-data/report.pdf"), new IOException("The file changed."), CancellationToken.None); + await firstStep.RecordDocumentIndexedAsync(Document("/tmp/test-data/report.pdf"), 4, false, CancellationToken.None); + + var (nextStep, _, _) = CreateProgress(totalDocuments: 10, pass: pass); + + Assert.That(nextStep.SkipIfFailedEarlierInPass("/tmp/test-data/report.pdf", "FINGERPRINT"), Is.False, "An indexed document showed up as failed for the rest of the pass."); + } + + [Test] + public async Task AProviderFailureIsReportedOncePerPass() + { + var message = $"Too many requests. Test {Guid.NewGuid()}"; + var collector = new ErrorCollector(message); + MessageBus.INSTANCE.RegisterComponent(collector); + MessageBus.INSTANCE.ApplyFilters(collector, [], [Event.SHOW_ERROR]); + + try + { + var pass = NewPass(); + var (firstStep, _, _) = CreateProgress(totalDocuments: 10, pass: pass); + await firstStep.RecordDocumentFailureAsync(Document("/tmp/test-data/a.pdf"), new ProviderRequestException(ProviderRequestFailureReason.TOO_MANY_REQUESTS, message), CancellationToken.None); + var (nextStep, _, _) = CreateProgress(totalDocuments: 10, pass: pass); + await nextStep.RecordDocumentFailureAsync(Document("/tmp/test-data/b.pdf"), new ProviderRequestException(ProviderRequestFailureReason.TOO_MANY_REQUESTS, message), CancellationToken.None); + var reportedDuringThePass = collector.Count; + + var (nextPass, _, _) = CreateProgress(totalDocuments: 10, pass: NewPass()); + await nextPass.RecordDocumentFailureAsync(Document("/tmp/test-data/a.pdf"), new ProviderRequestException(ProviderRequestFailureReason.TOO_MANY_REQUESTS, message), CancellationToken.None); + + Assert.Multiple(() => + { + Assert.That(reportedDuringThePass, Is.EqualTo(1), "A rate limit which lasts showed a message every five minutes."); + Assert.That(collector.Count, Is.EqualTo(2), "A new pass is a new attempt, which the user hears about should it fail again."); + }); + } + finally + { + MessageBus.INSTANCE.Unregister(collector); + } + } + [Test] public void EveryStatusOfARunCarriesTheLastSync() { @@ -178,7 +400,9 @@ public sealed class DocumentRunProgressTests Assert.That(statuses[^1].Failures[0].DisplayName, Is.EqualTo("report.pdf")); } - private static (DocumentRunProgress Progress, List Statuses, DataSourceEmbeddingManifest Manifest) CreateProgress(int totalDocuments, IIndexedDataSource? indexedDataSource = null) + private static IndexingPass NewPass() => new(new DataSourceRunSignature("TEST-EMBEDDING-SIGNATURE")); + + private static (DocumentRunProgress Progress, List Statuses, DataSourceEmbeddingManifest Manifest) CreateProgress(int totalDocuments, IIndexedDataSource? indexedDataSource = null, RunShare? share = null, int failedInputs = 0, IndexingPass? pass = null) { var statuses = new List(); var manifest = new DataSourceEmbeddingManifest(); @@ -201,18 +425,20 @@ public sealed class DocumentRunProgressTests new NoVectorStoreClient("Test vector store", "Not available in tests."), new NoIndexStoreClient("Test index store", "Not available in tests."), manifest, + share ?? new RunShare(RunShare.DEFAULT_DURATION), + pass ?? NewPass(), new SettingsManager(NullLogger.Instance, null!), statuses.Add, NullLogger.Instance); - return (new DocumentRunProgress(context, totalDocuments, 0, string.Empty, [], NullLogger.Instance), statuses, manifest); + return (new DocumentRunProgress(context, totalDocuments, failedInputs, string.Empty, [], NullLogger.Instance), statuses, manifest); } private static EmbeddingDocument Document(string path) => Document(path, Path.GetFileName(path)); - private static EmbeddingDocument Document(string key, string displayName) => new( + private static EmbeddingDocument Document(string key, string displayName, string fingerprint = "FINGERPRINT") => new( key, - new EmbeddingStateFile(IndexedDocumentIds.CreateParentId(DATA_SOURCE_ID, key), key, displayName, displayName, "pdf", "FINGERPRINT", 1024, DateTimeOffset.UnixEpoch, DateTimeOffset.UnixEpoch, DateTimeOffset.UnixEpoch, 0), + new EmbeddingStateFile(IndexedDocumentIds.CreateParentId(DATA_SOURCE_ID, key), key, displayName, displayName, "pdf", fingerprint, 1024, DateTimeOffset.UnixEpoch, DateTimeOffset.UnixEpoch, DateTimeOffset.UnixEpoch, 0), displayName, _ => NoChunks()); @@ -221,4 +447,24 @@ public sealed class DocumentRunProgressTests await Task.CompletedTask; yield break; } + + /// + /// Counts the error messages with one text, out of whatever other tests send meanwhile. + /// + private sealed class ErrorCollector(string message) : IMessageBusReceiver + { + private int count; + + public int Count => Volatile.Read(ref this.count); + + public Task ProcessMessage(ComponentBase? sendingComponent, Event triggeredEvent, T? data) + { + if (triggeredEvent is Event.SHOW_ERROR && data is DataErrorMessage error && error.Message == message) + Interlocked.Increment(ref this.count); + + return Task.CompletedTask; + } + + public Task> ProcessMessageWithResult(ComponentBase? sendingComponent, Event triggeredEvent, TPayload? data) => Task.FromResult(MessageBusAnswer.NONE); + } } \ No newline at end of file diff --git a/app/Tests/Tools/FilePartitionTests.cs b/app/Tests/Tools/FilePartitionTests.cs new file mode 100644 index 00000000..46c3959b --- /dev/null +++ b/app/Tests/Tools/FilePartitionTests.cs @@ -0,0 +1,69 @@ +using AIStudio.Tools; +using AIStudio.Tools.Services; +using AIStudio.Tools.Services.Indexing; + +namespace AIStudio.Tests.Tools; + +/// +/// Checks how a run of the file indexer sorts the files of a folder before it reads the first one. +/// +/// +/// A large folder is indexed in steps of a few minutes, and every step lists the folder anew. What a +/// step counts up front is what its progress starts with: a file counted as to be read lets the bar +/// jump back with every step, and costs the time to read it once more. A file left out wrongly is +/// never indexed at all. +/// +[TestFixture] +public sealed class FilePartitionTests +{ + private static readonly FileInfo UNCHANGED = new("/tmp/test-data/unchanged.pdf"); + private static readonly FileInfo CHANGED = new("/tmp/test-data/changed.pdf"); + private static readonly FileInfo ADDED = new("/tmp/test-data/added.pdf"); + private static readonly FileInfo SCANNED = new("/tmp/test-data/scanned.pdf"); + private static readonly FileInfo SCANNED_AND_RECOGNIZED = new("/tmp/test-data/scanned-and-recognized.pdf"); + private static readonly FileInfo FAILED_EARLIER = new("/tmp/test-data/failed-earlier.pdf"); + private static readonly FileInfo CHANGED_SINCE_IT_FAILED = new("/tmp/test-data/changed-since-it-failed.pdf"); + + private static readonly DataSourceEmbeddingFailure EARLIER_FAILURE = new(FAILED_EARLIER.FullName, "The file changed while it was being indexed.", DateTimeOffset.UnixEpoch); + + private FilePartition partition = null!; + + [SetUp] + public void SortTheFiles() + { + var files = new[] { UNCHANGED, CHANGED, ADDED, SCANNED, SCANNED_AND_RECOGNIZED, FAILED_EARLIER, CHANGED_SINCE_IT_FAILED }; + var fingerprints = files.ToDictionary(file => file.FullName, file => $"NOW:{file.Name}", StringComparer.OrdinalIgnoreCase); + + var manifest = new DataSourceEmbeddingManifest(); + manifest.Files[UNCHANGED.FullName] = Indexed($"NOW:{UNCHANGED.Name}"); + manifest.Files[CHANGED.FullName] = Indexed("BEFORE"); + manifest.PermanentFailures[SCANNED.FullName] = Unreadable($"NOW:{SCANNED.Name}"); + manifest.PermanentFailures[SCANNED_AND_RECOGNIZED.FullName] = Unreadable("BEFORE-OCR"); + + var pass = new IndexingPass(new DataSourceRunSignature("EMBEDDING-SIGNATURE")); + pass.RecordFailure(FAILED_EARLIER.FullName, $"NOW:{FAILED_EARLIER.Name}", EARLIER_FAILURE); + pass.RecordFailure(CHANGED_SINCE_IT_FAILED.FullName, "BEFORE", EARLIER_FAILURE with { DocumentKey = CHANGED_SINCE_IT_FAILED.FullName }); + + this.partition = FileSourceIndexer.PartitionFiles(files, fingerprints, manifest, pass); + } + + [Test] + public void UnchangedFilesAreCountedButNotRead() => + Assert.That(this.partition.UnchangedCount, Is.EqualTo(1)); + + [Test] + public void FilesSkippedForGoodAreNotReadUntilTheyChange() => + Assert.That(this.partition.StillUnreadable.Select(file => file.Key), Is.EqualTo(new[] { SCANNED.FullName }), "An OCR run over a scanned document changes it, which is exactly when it deserves another attempt."); + + [Test] + public void FilesWhichFailedEarlierInThePassAreNotReadUntilTheyChange() => + Assert.That(this.partition.FailedEarlierInPass, Is.EqualTo(new[] { EARLIER_FAILURE }), "The failure keeps its place in the list, or the list loses what the steps before found."); + + [Test] + public void EveryOtherFileIsReadInTheOrderOfTheFolder() => + Assert.That(this.partition.ToRead.Select(file => file.FullName), Is.EqualTo(new[] { CHANGED.FullName, ADDED.FullName, SCANNED_AND_RECOGNIZED.FullName, CHANGED_SINCE_IT_FAILED.FullName })); + + private static EmbeddedFileRecord Indexed(string fingerprint) => new(fingerprint, 1024, DateTimeOffset.UnixEpoch, DateTimeOffset.UnixEpoch, 4); + + private static PermanentIndexingFailureRecord Unreadable(string fingerprint) => new(fingerprint, FileExtractionErrorCode.NO_TEXT_EXTRACTED, "No text could be read.", DateTimeOffset.UnixEpoch); +} \ No newline at end of file diff --git a/app/Tests/Tools/IndexingPassTests.cs b/app/Tests/Tools/IndexingPassTests.cs new file mode 100644 index 00000000..61fd3f73 --- /dev/null +++ b/app/Tests/Tools/IndexingPassTests.cs @@ -0,0 +1,62 @@ +using AIStudio.Tools.Services; +using AIStudio.Tools.Services.Indexing; + +namespace AIStudio.Tests.Tools; + +/// +/// Checks when an indexing run carries on the pass of its data source, and when it begins a new one. +/// +/// +/// A pass decides which documents a run leaves alone, because they failed during an earlier step. A +/// run which carries on a pass it should not have left documents unindexed which would work now. A +/// run which begins anew when it should not have reads every failed document once more, with a cloud +/// provider for real money, and tells the user about the same failure once more. +/// +[TestFixture] +public sealed class IndexingPassTests +{ + private static readonly DataSourceRunSignature SIGNATURE = new("EMBEDDING-SIGNATURE"); + + [Test] + public void TheNextStepCarriesOnThePass() + { + var pass = new IndexingPass(SIGNATURE); + + // + // A request during a step merges into the next step, which then runs for the reason of + // that request: + // + Assert.Multiple(() => + { + foreach (var refreshMode in new[] { DataSourceEmbeddingRefreshMode.CONTINUATION, DataSourceEmbeddingRefreshMode.WATCHER_HASH_CHECK, DataSourceEmbeddingRefreshMode.INTERVAL_CHECK }) + Assert.That(IndexingPass.ContinueOrBegin(pass, refreshMode, SIGNATURE), Is.SameAs(pass), $"RefreshMode={refreshMode}"); + }); + } + + [Test] + public void AManualRetryBeginsANewPass() + { + var pass = new IndexingPass(SIGNATURE); + + Assert.That(IndexingPass.ContinueOrBegin(pass, DataSourceEmbeddingRefreshMode.MANUAL_RETRY, SIGNATURE), Is.Not.SameAs(pass), "The user asked for every failed document to be tried again."); + } + + [Test] + public void OtherEmbeddingSettingsBeginANewPass() + { + var pass = new IndexingPass(SIGNATURE); + var otherSignature = new DataSourceRunSignature("OTHER-EMBEDDING-SIGNATURE"); + + var nextPass = IndexingPass.ContinueOrBegin(pass, DataSourceEmbeddingRefreshMode.CONTINUATION, otherSignature); + + Assert.Multiple(() => + { + Assert.That(nextPass, Is.Not.SameAs(pass), "What failed with the old embedding provider may well work with the new one."); + Assert.That(nextPass.RunSignature, Is.EqualTo(otherSignature)); + }); + } + + [Test] + public void ARunWithoutAPassBeginsOne() => + Assert.That(IndexingPass.ContinueOrBegin(null, DataSourceEmbeddingRefreshMode.CONTINUATION, SIGNATURE).RunSignature, Is.EqualTo(SIGNATURE)); +} \ No newline at end of file diff --git a/app/Tests/Tools/QueuedStatusTests.cs b/app/Tests/Tools/QueuedStatusTests.cs new file mode 100644 index 00000000..f919a6d9 --- /dev/null +++ b/app/Tests/Tools/QueuedStatusTests.cs @@ -0,0 +1,95 @@ +using AIStudio.Settings.DataModel; +using AIStudio.Tools.Services; + +namespace AIStudio.Tests.Tools; + +/// +/// Checks what the row of a data source shows while it waits for its next run. +/// +/// +/// A large data source is indexed in steps of a few minutes and waits in the queue between them. +/// Its row must keep the progress and the failures of the steps so far, or the bar jumps back with +/// every step and the failures vanish for minutes. What belongs to a run going on must go, though: +/// a row which names a document while nothing runs, or offers a decision the next run may no longer +/// ask for, tells the user something which is not true. +/// +[TestFixture] +public sealed class QueuedStatusTests +{ + private static readonly DateTimeOffset NOW = new(2026, 10, 11, 9, 0, 0, TimeSpan.Zero); + + private static readonly DataSourceEmbeddingFailure FAILURE = new("/tmp/test-data/report.pdf", "The file could not be read.", NOW); + + [Test] + public void AQueuedDataSourceKeepsWhatItsRowShowedOfTheRunsSoFar() + { + var status = RunningStatus(); + + var queued = DataSourceEmbeddingService.AsQueued(status, DataSource()); + + Assert.Multiple(() => + { + Assert.That(queued.State, Is.EqualTo(DataSourceEmbeddingState.QUEUED)); + Assert.That(queued.TotalDocuments, Is.EqualTo(status.TotalDocuments)); + Assert.That(queued.IndexedDocuments, Is.EqualTo(status.IndexedDocuments)); + Assert.That(queued.FailedDocuments, Is.EqualTo(status.FailedDocuments)); + Assert.That(queued.PermanentlySkippedDocuments, Is.EqualTo(status.PermanentlySkippedDocuments), "Dropping them made the bar jump back."); + Assert.That(queued.ProgressPercent, Is.EqualTo(status.ProgressPercent), "The bar must not jump back between two steps of the same data source."); + Assert.That(queued.Failures, Is.EqualTo(new[] { FAILURE })); + Assert.That(queued.LastError, Is.EqualTo(status.LastError), "The sentence about the data source as a whole belongs next to its failures."); + Assert.That(queued.LastSyncUtc, Is.EqualTo(NOW)); + }); + } + + [Test] + public void AQueuedDataSourceForgetsWhatBelongsToARunGoingOn() + { + var queued = DataSourceEmbeddingService.AsQueued(RunningStatus(), DataSource()); + + Assert.Multiple(() => + { + Assert.That(queued.CurrentDocument, Is.Empty, "Nothing is being read while the data source waits."); + Assert.That(queued.CurrentDocumentBlock, Is.Null); + Assert.That(queued.CurrentDocumentPage, Is.Null); + Assert.That(queued.Attention, Is.EqualTo(DataSourceAttention.NONE), "The next run asks the user anew when the question still applies."); + Assert.That(queued.PendingRemovalCount, Is.Null, "The next run asks the user anew, with the number it finds then."); + Assert.That(queued.VectorStoreUnreadable, Is.False); + }); + } + + [Test] + public void AQueuedDataSourceShowsItsCurrentName() => + Assert.That(DataSourceEmbeddingService.AsQueued(RunningStatus(), DataSource()).DataSourceName, Is.EqualTo("Test data"), "The data source may have been renamed since its last run."); + + private static DataSourceEmbeddingStatus RunningStatus() => new( + "6f1d6a4e-6a5e-4c62-9a4f-0f2d2c8b7a11", + "Old name", + DataSourceType.LOCAL_DIRECTORY, + DataSourceEmbeddingState.RUNNING, + 1200, + 400, + 3, + "/tmp/test-data/notes.docx", + "Some files could not be read.", + [FAILURE], + PermanentlySkippedDocuments: 7, + CurrentDocumentBlock: 2, + CurrentDocumentPage: 5, + VectorStoreUnreadable: true, + Attention: DataSourceAttention.MASS_REMOVAL_PENDING, + PendingRemovalCount: 42, + LastSyncUtc: NOW); + + private static DataSourceLocalDirectory DataSource() => new() + { + Num = 1, + Id = "6f1d6a4e-6a5e-4c62-9a4f-0f2d2c8b7a11", + Name = "Test data", + Description = "Documents used by the tests.", + Type = DataSourceType.LOCAL_DIRECTORY, + EmbeddingId = "b0a4c4d2-1f3e-4f0a-8c9d-5a6b7c8d9e01", + MaxChunkTokenLength = 512, + ChunkOverlapTokenLength = 100, + Path = "/tmp/test-data", + }; +} \ No newline at end of file