Added a pause between indexing steps for shared embedding servers (#1114)

app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.cs

@@ -17,7 +17,7 @@ using static AIStudio.Tools.Services.Indexing.IndexingLogFormat;
 namespace AIStudio.Tools.Services;
 
 public sealed partial class DataSourceEmbeddingService(SettingsManager settingsManager, RustService rustService, DatabaseClientProvider databaseClientProvider,
-    PromptInjectionGuardService guardService, MediaTranscriptionService mediaTranscriptionService, ILogger<DataSourceEmbeddingService> logger) : BackgroundService, IMessageBusReceiver
+    PromptInjectionGuardService guardService, MediaTranscriptionService mediaTranscriptionService, ThreadSafeRandom rng, ILogger<DataSourceEmbeddingService> logger) : BackgroundService, IMessageBusReceiver
 {
     /// <summary>
     /// How long the re-index check waits for the index database before it gives up.
@@ -73,6 +73,11 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM
     /// </summary>
     private readonly IndexingPauses pauses = new();
 
+    /// <summary>
+    /// The pause between two steps, cf. WaitForPauseBetweenStepsAsync.
+    /// </summary>
+    private readonly PauseBetweenSteps pauseBetweenSteps = new();
+
     /// <summary>
     /// An active run.
     /// </summary>
@@ -680,13 +685,31 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM
             // 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. A paused indexing holds
-            // every run back in front of the gate, cf. PauseIndexingAsync:
+            // every run back in front of the gate, cf. PauseIndexingAsync. Behind the gate, the run
+            // keeps its distance to the step before, cf. WaitForPauseBetweenStepsAsync:
             //
             await this.pauses.WaitWhilePausedAsync(runTokenSource.Token);
             await this.runGate.WaitAsync(runTokenSource.Token);
             try
             {
-                await this.ProcessDataSourceAsync(dataSource, refreshMode, runControl, runTokenSource.Token);
+                await this.WaitForPauseBetweenStepsAsync(dataSource, refreshMode, runTokenSource.Token);
+
+                //
+                // Waiting may have taken minutes, while the user could change the data source. The
+                // run has resolved nothing yet, so it starts with the data source as it is now,
+                // instead of finding itself outdated once it records its signature:
+                //
+                if (!this.TryGetConfiguredIndexedSource(dataSource.Id, out var currentDataSource))
+                {
+                    logger.LogDebug(
+                        "Skipping embedding run for data source '{DataSourceName}' ({DataSourceId}) because it is no longer configured. RefreshMode={RefreshMode}.",
+                        dataSource.Name,
+                        dataSource.Id,
+                        refreshMode);
+                    return;
+                }
+
+                await this.ProcessDataSourceAsync(currentDataSource, refreshMode, runControl, runTokenSource.Token);
             }
             finally
             {
@@ -718,6 +741,37 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM
         }
     }
 
+    /// <summary>
+    /// Waits for what is left of the pause between the last step and the run which is about to start, cf. PauseBetweenSteps.
+    /// </summary>
+    /// <remarks>
+    /// The pause spares an embedding server which many people share, so it holds whichever data
+    /// source comes next, the same one included. It is waited for behind the run gate: only one run
+    /// waits, and its token ends the wait, when the app shuts down, when the data source is deleted
+    /// or repaired, or when a backup pauses the indexing. Meanwhile, the data source shows as queued,
+    /// since nothing published a status for this run yet, and every other request is queued as usual.
+    /// </remarks>
+    /// <param name="dataSource">The data source of the run, for the log.</param>
+    /// <param name="refreshMode">Why the run was asked for, for the log.</param>
+    /// <param name="token">The token of the run.</param>
+    private async Task WaitForPauseBetweenStepsAsync(IIndexedDataSource dataSource, DataSourceEmbeddingRefreshMode refreshMode, CancellationToken token)
+    {
+        var range = PauseRange.FromSettings(settingsManager.ConfigurationData.App.DataSourceIndexing);
+        var pause = this.pauseBetweenSteps.GetPauseBeforeNextStep(DateTimeOffset.UtcNow, range, rng);
+        if (pause.Remaining <= TimeSpan.Zero)
+            return;
+
+        logger.LogInformation(
+            "Pausing {PauseSeconds:F0} seconds between two indexing steps. Data source '{DataSourceName}' ({DataSourceId}) starts in {RemainingSeconds:F0} seconds. RefreshMode={RefreshMode}.",
+            pause.Pause.TotalSeconds,
+            dataSource.Name,
+            dataSource.Id,
+            pause.Remaining.TotalSeconds,
+            refreshMode);
+
+        await Task.Delay(pause.Remaining, token);
+    }
+
     /// <summary>
     /// Works out everything an indexing run needs, before anything is read from the data source.
     /// </summary>
@@ -891,6 +945,7 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM
         }
 
         var outcome = IndexedRunOutcome.DONE;
+        IndexedRunContext? context = null;
         try
         {
             //
@@ -899,7 +954,7 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM
             // removes something in forty of them is one thing that happened to the user, not forty.
             //
             await using var promptInjectionScope = guardService.BeginAction();
-            var context = await this.PrepareIndexedRunAsync(dataSource, refreshMode, indexer.Share, runControl, promptInjectionScope, token);
+            context = await this.PrepareIndexedRunAsync(dataSource, refreshMode, indexer.Share, runControl, promptInjectionScope, token);
             if (context is null)
                 return;
 
@@ -932,6 +987,13 @@ public sealed partial class DataSourceEmbeddingService(SettingsManager settingsM
         }
         finally
         {
+            //
+            // Only a run which worked on documents put load on the embedding provider, so only such
+            // a step makes the next one wait, even when it failed or stopped halfway:
+            //
+            if (context?.ShareStartedUtc is not null)
+                this.pauseBetweenSteps.RecordStepEnded(DateTimeOffset.UtcNow);
+
             //
             // 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:

app/MindWork AI Studio/Tools/Services/Indexing/AGENTS.md

@@ -40,6 +40,10 @@ The parts:
   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.
+- **`PauseBetweenSteps`** - a gap after every step which worked on documents, so that many people can share one
+  embedding server. The next run, of whichever data source, waits for a random pause between the bounds of the
+  settings (`PauseRange`: 0 to 300 seconds, none by default), less the time which passed since the step ended.
+  Not to be confused with the pause of the indexing as a whole (`IndexingPauses`), which a backup takes.
 - **`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.
@@ -98,6 +102,13 @@ Rules which are easy to break:
   holds two stores which do not fit together. Discarding the stored index of a data source, when it is deleted
   or repaired, waits for the end of a pause as well (`WaitWhilePausedAsync`); the user interface asks
   `IsIndexingPaused` first and tells the user instead of waiting.
+- **The pause between steps lies behind the run gate.** `ProcessDataSourceRunAsync` waits for it right after
+  `runGate.WaitAsync`, before the run resolves anything: only one run waits, its token ends the wait, and its
+  data source still shows as queued. Never wait for it in an indexer, where the run would hold its signature and
+  its stores, and show as running. Afterward, the run looks up its data source anew, since the user may have
+  changed it meanwhile. A step opens a gap only once it asked `HasDoneItsShare` (`IndexedRunContext.ShareStartedUtc`),
+  so a run which found nothing to read leaves none, and a startup check over unchanged data sources does not
+  slow down.
 - **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 or another selection of documents, begin a new pass

app/MindWork AI Studio/Tools/Services/Indexing/PauseBeforeStep.cs

@@ -0,0 +1,14 @@
+namespace AIStudio.Tools.Services.Indexing;
+
+/// <summary>
+/// How long the indexing step which is about to start waits, cf. PauseBetweenSteps.GetPauseBeforeNextStep.
+/// </summary>
+/// <param name="Pause">The pause drawn for the gap since the last step.</param>
+/// <param name="Remaining">What is left of the pause now. More than zero, unless this is NONE.</param>
+internal sealed record PauseBeforeStep(TimeSpan Pause, TimeSpan Remaining)
+{
+    /// <summary>
+    /// The step starts right away.
+    /// </summary>
+    public static readonly PauseBeforeStep NONE = new(TimeSpan.Zero, TimeSpan.Zero);
+}

app/MindWork AI Studio/Tools/Services/Indexing/PauseBetweenSteps.cs

@@ -0,0 +1,75 @@
+namespace AIStudio.Tools.Services.Indexing;
+
+/// <summary>
+/// The pause between two indexing steps, which spares an embedding server that many people share.
+/// </summary>
+/// <remarks>
+/// Not to be confused with the pauses of the indexing as a whole, e.g., while a backup copies the
+/// indexes, cf. IndexingPauses.<br/><br/>
+///
+/// The pause is a gap between the end of one step and the start of the next, whichever data source
+/// either of them belongs to: it is about the load on the embedding server, not about the order of
+/// the data sources. Only a step which worked on documents opens a gap. Its pause is drawn once the
+/// next step is about to start, from the bounds which apply then, and the time which passed in the
+/// meantime counts toward it. When no step comes, nothing is drawn.
+/// </remarks>
+internal sealed class PauseBetweenSteps
+{
+    private readonly Lock stateLock = new();
+
+    /// <summary>
+    /// When the last step which worked on documents ended. Null before the first one.
+    /// </summary>
+    private DateTimeOffset? lastStepEndedUtc;
+
+    /// <summary>
+    /// The pause drawn for the gap since the last step. Null until a step is about to start.
+    /// </summary>
+    private TimeSpan? drawnPause;
+
+    /// <summary>
+    /// Notes that a step which worked on documents ended, which opens a new gap.
+    /// </summary>
+    /// <param name="nowUtc">The current time, passed in so the pause can be tested.</param>
+    public void RecordStepEnded(DateTimeOffset nowUtc)
+    {
+        lock (this.stateLock)
+        {
+            this.lastStepEndedUtc = nowUtc;
+            this.drawnPause = null;
+        }
+    }
+
+    /// <summary>
+    /// Works out how long the step which is about to start waits.
+    /// </summary>
+    /// <remarks>
+    /// The first step which asks after a gap opened draws its pause, with both bounds included. A step
+    /// which stops while it waits, e.g., because its data source was deleted, leaves the same pause
+    /// to the step after it, which waits only for the rest.
+    /// </remarks>
+    /// <param name="nowUtc">The current time, passed in so the pause can be tested.</param>
+    /// <param name="range">The bounds which apply now.</param>
+    /// <param name="random">Draws the pause.</param>
+    /// <returns>The pause and what is left of it, or NONE when the step starts right away.</returns>
+    public PauseBeforeStep GetPauseBeforeNextStep(DateTimeOffset nowUtc, PauseRange range, Random random)
+    {
+        lock (this.stateLock)
+        {
+            if (this.lastStepEndedUtc is not { } stepEndedUtc)
+                return PauseBeforeStep.NONE;
+
+            if (this.drawnPause is null)
+            {
+                // Nothing is drawn, so bounds which are set before the next step still apply to this gap:
+                if (range.IsNone)
+                    return PauseBeforeStep.NONE;
+
+                this.drawnPause = TimeSpan.FromSeconds(random.Next(range.MinimumSeconds, range.MaximumSeconds + 1));
+            }
+
+            var remaining = stepEndedUtc + this.drawnPause.Value - nowUtc;
+            return remaining > TimeSpan.Zero ? new(this.drawnPause.Value, remaining) : PauseBeforeStep.NONE;
+        }
+    }
+}

app/MindWork AI Studio/Tools/Services/Indexing/PauseRange.cs

@@ -0,0 +1,68 @@
+using AIStudio.Settings;
+using AIStudio.Settings.DataModel;
+
+namespace AIStudio.Tools.Services.Indexing;
+
+/// <summary>
+/// The bounds of the random pause between two indexing steps, cf. PauseBetweenSteps.
+/// </summary>
+/// <param name="MinimumSeconds">The shortest pause. Between zero and MaximumSeconds.</param>
+/// <param name="MaximumSeconds">The longest pause, which may be drawn as well. At most DataDataSourceIndexing.MAX_PAUSE_BETWEEN_STEPS_SECONDS.</param>
+internal sealed record PauseRange(int MinimumSeconds, int MaximumSeconds)
+{
+    /// <summary>
+    /// No pause at all, which is the default.
+    /// </summary>
+    public static readonly PauseRange NONE = new(0, 0);
+
+    /// <summary>
+    /// Whether the next step starts right away.
+    /// </summary>
+    public bool IsNone => this.MaximumSeconds is 0;
+
+    /// <summary>
+    /// The bounds which apply now, as the user set them and as configuration plugins lock them.
+    /// </summary>
+    /// <param name="settings">The settings of the indexing.</param>
+    /// <returns>The bounds which apply.</returns>
+    public static PauseRange FromSettings(DataDataSourceIndexing settings)
+    {
+        var minimumSeconds = settings.MinimumPauseBetweenStepsSeconds;
+        var maximumSeconds = settings.MaximumPauseBetweenStepsSeconds;
+
+        //
+        // Which bound a configuration plugin locks matters only while the bounds are the wrong way
+        // round. Asking reads the managed state of the whole app, which bounds in order do without:
+        //
+        if (minimumSeconds <= maximumSeconds)
+            return GetEffective(minimumSeconds, false, maximumSeconds, false);
+
+        var isMinimumLocked = ManagedConfiguration.TryGet(x => x.App.DataSourceIndexing, x => x.MinimumPauseBetweenStepsSeconds, out var minimumMeta) && minimumMeta.IsLocked;
+        var isMaximumLocked = ManagedConfiguration.TryGet(x => x.App.DataSourceIndexing, x => x.MaximumPauseBetweenStepsSeconds, out var maximumMeta) && maximumMeta.IsLocked;
+        return GetEffective(minimumSeconds, isMinimumLocked, maximumSeconds, isMaximumLocked);
+    }
+
+    /// <summary>
+    /// Works out the bounds which apply, so that the lower one never exceeds the upper one.
+    /// </summary>
+    /// <remarks>
+    /// A bound which a configuration plugin locks wins, and the other one adjusts to it. A bound the
+    /// plugin only presets counts like one the user set. Otherwise, the lower bound wins, i.e., the
+    /// longer pause: it spares the embedding server, which is what the pause is for. That holds as
+    /// well when a plugin locks both bounds the wrong way round.
+    /// </remarks>
+    /// <param name="minimumSeconds">The lower bound as set.</param>
+    /// <param name="isMinimumLocked">Whether a configuration plugin locks the lower bound.</param>
+    /// <param name="maximumSeconds">The upper bound as set.</param>
+    /// <param name="isMaximumLocked">Whether a configuration plugin locks the upper bound.</param>
+    /// <returns>The bounds which apply.</returns>
+    public static PauseRange GetEffective(int minimumSeconds, bool isMinimumLocked, int maximumSeconds, bool isMaximumLocked)
+    {
+        var minimum = Math.Clamp(minimumSeconds, 0, DataDataSourceIndexing.MAX_PAUSE_BETWEEN_STEPS_SECONDS);
+        var maximum = Math.Clamp(maximumSeconds, 0, DataDataSourceIndexing.MAX_PAUSE_BETWEEN_STEPS_SECONDS);
+        if (isMaximumLocked && !isMinimumLocked)
+            return new(Math.Min(minimum, maximum), maximum);
+
+        return new(minimum, Math.Max(minimum, maximum));
+    }
+}