From 391ec3be591df0cf362c3506fdddc3db1896abc2 Mon Sep 17 00:00:00 2001 From: Kevin Montrose Date: Mon, 24 Aug 2026 10:24:04 -0400 Subject: [PATCH 1/2] with recovery cleanup, blocking during delete request is unnecessary - stressed tests to confirm --- .../Resp/Vector/VectorManager.Cleanup.cs | 40 +++---------------- libs/server/Resp/Vector/VectorManager.cs | 11 +---- .../VectorStore/VectorSessionFunctions.cs | 2 +- 3 files changed, 8 insertions(+), 45 deletions(-) diff --git a/libs/server/Resp/Vector/VectorManager.Cleanup.cs b/libs/server/Resp/Vector/VectorManager.Cleanup.cs index 72869a850a1..9f4278c9c55 100644 --- a/libs/server/Resp/Vector/VectorManager.Cleanup.cs +++ b/libs/server/Resp/Vector/VectorManager.Cleanup.cs @@ -87,7 +87,7 @@ public bool Reader(in TSourceLogRecord logRecord, RecordMetada } private readonly VectorSetCleanupWorkChannel cleanupTaskChannel; - private readonly VectorSetCleanupWorkChannel<(ulong Context, TaskCompletionSource MarkCompleted)> requestCleanupTaskChannel; + private readonly VectorSetCleanupWorkChannel requestCleanupTaskChannel; private readonly VectorSetCleanupWorkChannel requestDropTaskChannel; private readonly VectorSetCleanupWorkSet<(ulong Context, nint IndexPtr)> requestedDrops; private readonly ConcurrentDictionary potentiallyDeleted; @@ -203,8 +203,6 @@ private async Task RunRequestCleanupTaskAsync() // // The fact that we're in an OnDispose means Reset() isn't running. - var completions = new List(); - try { // TODO: this doesn't work with non-RESP impls... which maybe we don't care about? @@ -220,14 +218,9 @@ private async Task RunRequestCleanupTaskAsync() lock (this) { // Read all pending requests so we can do one update - while (requestCleanupTaskChannel.TryRead(out var t)) + while (requestCleanupTaskChannel.TryRead(out var context)) { - if (t.MarkCompleted != null) - { - completions.Add(t.MarkCompleted); - } - - var (contextIndex, contextValue) = ContextMetadata.DecomposeContext(t.Context); + var (contextIndex, contextValue) = ContextMetadata.DecomposeContext(context); if (!contextMetadatas[contextIndex].IsCleaningUp(contextIndex != 0, contextValue)) { contextMetadatas[contextIndex].MarkCleaningUp(contextIndex != 0, contextValue); @@ -246,35 +239,12 @@ private async Task RunRequestCleanupTaskAsync() ExceptionInjectionHelper.TriggerException(ExceptionInjectionType.VectorSet_Interrupt_Delete_3); - foreach (var completion in completions) - { - try - { - _ = completion.TrySetResult(); - } - catch (Exception innerE) - { - logger?.LogError(innerE, "While completing Vector Set cleanup request"); - } - } - // Pump the cleanup task once we're done _ = cleanupTaskChannel.TryPublish(); } catch (Exception e) { - foreach (var completion in completions) - { - try - { - _ = completion.TrySetException(e); - } - catch (Exception innerE) - { - // Best effort - logger?.LogError(innerE, "While cancelling Vector Set cleanup requests"); - } - } + logger?.LogError(e, "During request cleanup task"); } finally { @@ -504,7 +474,7 @@ static void QueueCleanups(VectorManager self) if (needsDelete) { // No need to wait for marking, since the record is already "deleted" - if (!self.requestCleanupTaskChannel.TryPublish((context, null))) + if (!self.requestCleanupTaskChannel.TryPublish(context)) { self.logger?.LogWarning("Could not request delete of abandoned Vector Set {key}", SpanByte.ToShortString(key)); } diff --git a/libs/server/Resp/Vector/VectorManager.cs b/libs/server/Resp/Vector/VectorManager.cs index 3d4d786cd20..66da206380d 100644 --- a/libs/server/Resp/Vector/VectorManager.cs +++ b/libs/server/Resp/Vector/VectorManager.cs @@ -666,7 +666,7 @@ internal bool TrySetAttribute(ReadOnlySpan indexValue, ReadOnlySpan /// /// Request deletion of a Vector Set given the VALUE of the index key. /// - internal void RequestDeletion(Span value) + internal void RequestDeletion(ReadOnlySpan value) { if (value.Length != IndexSize) { @@ -686,16 +686,11 @@ internal void RequestDeletion(Span value) return; } - var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - - if (!requestCleanupTaskChannel.TryPublish((context, tcs))) + if (!requestCleanupTaskChannel.TryPublish(context)) { throw new GarnetException("Could not submit request for Vector Set cleanup, aborting delete"); } - // Wait until the context is _marked_ for cleanup, but not the actual cleanup - AsyncUtils.BlockingWait(tcs.Task); - // Tell DiskANN to clean itself up DropIndex(value); } @@ -708,8 +703,6 @@ internal void RequestDeletion(Span value) /// There's subtlety here because the DiskANN index might be in use (on the current or other threads) /// and we can't allow the index to be recreated until any requested drops are processed. /// - /// - /// internal void RequestDropInMemoryIndex(ReadOnlySpan key, ReadOnlySpan value) { if (value.Length != IndexSize) diff --git a/libs/server/Storage/Functions/VectorStore/VectorSessionFunctions.cs b/libs/server/Storage/Functions/VectorStore/VectorSessionFunctions.cs index c05ca6a39e6..ad109abdfe9 100644 --- a/libs/server/Storage/Functions/VectorStore/VectorSessionFunctions.cs +++ b/libs/server/Storage/Functions/VectorStore/VectorSessionFunctions.cs @@ -369,7 +369,7 @@ public readonly bool CopyUpdater(in TSourceLogRecord srcLogRec Debug.Assert(input.WriteDesiredSize <= newValueAligned.Length, "Insufficient space for copy update, this should never happen"); Debug.Assert(input.WriteDesiredSize <= oldValueAligned.Length, "Insufficient space for copy update, this should never happen"); - oldValueAligned.CopyTo(newValueAligned); + oldValueAligned[..input.WriteDesiredSize].CopyTo(newValueAligned[..input.WriteDesiredSize]); unsafe { From 6f985ba8210b9e398e6c4623994172b363ba3fd1 Mon Sep 17 00:00:00 2001 From: Kevin Montrose Date: Mon, 24 Aug 2026 10:25:27 -0400 Subject: [PATCH 2/2] update vector-sets.md --- website/docs/dev/vector-sets.md | 1 - 1 file changed, 1 deletion(-) diff --git a/website/docs/dev/vector-sets.md b/website/docs/dev/vector-sets.md index 0d0f933f114..d6c2d3ec793 100644 --- a/website/docs/dev/vector-sets.md +++ b/website/docs/dev/vector-sets.md @@ -135,7 +135,6 @@ Deletion of Vector Sets is detected in the `GarnetTriggers.OnDispose` callback, This takes place in four steps: 1. The Vector Set context is marked for deletion from `GarnetTriggers.OnDispose` * A background task does this, as we do not have a usable storage session in the `GarnetTriggers` callback - * We _block_ on that background task, if an error occurs an exception is raised and the delete aborted 2. `GarnetTriggers.OnDispose` returns, deleting the index key 3. A background cleanup task scans the Tsavorite log for element keys, [see Cleanup](#cleanup) for more detail 4. The Vector Set context is marked available