fix(index): use UNLINK instead of DEL in SearchIndex.drop_keys (#616)

## Summary

Switch `SearchIndex.drop_keys` and `AsyncSearchIndex.drop_keys` from
`DEL` to `UNLINK` so memory reclamation runs on a background thread.

Refs #600

## Motivation

`DEL` reclaims memory on the main thread, so a single `drop_keys` call
over a large key set stalls Redis proportionally to the freed keyspace.
For `SemanticCache` use cases where scope-targeted invalidation
routinely sweeps 10K to 1M+ keys (for example, a policy version rollover
in a multi-tenant deployment), this is a customer-visible latency event
on every invalidation. The issue thread on #600 lays this out in more
detail, including the path through `SemanticCache.drop()` which calls
`drop_keys` for the `keys=` argument.

`UNLINK` is a strict superset of `DEL` for this code path. It returns
the same count semantics and has been available since Redis 4, which is
well below the supported floor for current RedisVL targets. The only
observable difference is that reclaimed memory is reported lazily by
`MEMORY USAGE`, which is the point.

## Changes

- `redisvl/index/index.py`: `SearchIndex.drop_keys` (sync) now calls
`self._redis_client.unlink(...)` instead of
`self._redis_client.delete(...)`. Docstring updated to note the choice.
- `redisvl/index/index.py`: `AsyncSearchIndex.drop_keys` (async) now
calls `client.unlink(...)` instead of `client.delete(...)`. Docstring
updated to note the choice.
- `tests/unit/test_drop_keys_unlink.py`: new mock-based regression tests
covering single-key and list-of-keys paths on both sync and async
indexes, asserting `unlink` is called and `delete` is not.

`drop_documents` is intentionally left unchanged in this PR. It is a
related but separate API (it also applies the index prefix and validates
hash-tag co-location on cluster) and #601 is tracking the consistency
story there. Keeping this PR scoped to the `drop_keys` change matches
the framing in #600.

## Testing

All run locally with `python -m uv run pytest ...`. Docker was running
for `testcontainers`-backed integration tests.

- New regression suite (4 tests in
`tests/unit/test_drop_keys_unlink.py`):
- Fails on `upstream/main` with `AssertionError: Expected unlink to have
been [a]waited once. Called 0 times.` on all 4 cases.
  - Passes on this branch.
- Full `tests/unit/`: 836 passed, 1 skipped on baseline; 840 passed, 1
skipped on this branch. The +4 is the new regression suite. No
previously-passing test regressed and no new skips.
- `tests/integration/test_search_index.py` and
`tests/integration/test_async_search_index.py`: 93 passed (covers the
original `test_search_index_drop_keys` cases on both sync and async).
- `tests/integration/test_llmcache.py` and
`tests/integration/test_semantic_router.py`: 92 passed, 1 skipped. These
exercise `SemanticCache.drop()` and `SemanticRouter` which transit
`drop_keys` under the hood.
- `make`-equivalent linting: `isort --check-only`, `black --check`, and
`mypy redisvl` all clean.

<details>
<summary>Verification log tail</summary>

```
=== BASELINE UNIT TESTS ===
========== 836 passed, 1 skipped, 109 warnings in 250.94s (0:04:10) ===========

=== BASELINE INTEGRATION test_search_index drop_keys ===
================= 1 passed, 45 deselected, 1 warning in 6.21s =================

=== BASELINE INTEGRATION test_async_search_index drop_keys ===
================= 1 passed, 46 deselected, 1 warning in 5.59s =================

=== PATCHED: new regression test ===
tests/unit/test_drop_keys_unlink.py::TestDropKeysUsesUnlink::test_single_key_calls_unlink PASSED
tests/unit/test_drop_keys_unlink.py::TestDropKeysUsesUnlink::test_list_of_keys_calls_unlink PASSED
tests/unit/test_drop_keys_unlink.py::TestAsyncDropKeysUsesUnlink::test_single_key_calls_unlink PASSED
tests/unit/test_drop_keys_unlink.py::TestAsyncDropKeysUsesUnlink::test_list_of_keys_calls_unlink PASSED
======================== 4 passed, 1 warning in 5.07s =========================

=== format / lint (patched) ===
isort: would leave 186 files unchanged
black: 186 files would be left unchanged
mypy: Success: no issues found in 96 source files

=== PATCHED UNIT TESTS ===
========== 840 passed, 1 skipped, 108 warnings in 203.65s (0:03:23) ===========

=== PATCHED INTEGRATION drop_keys (sync + async) ===
tests/integration/test_search_index.py::test_search_index_drop_keys PASSED
tests/integration/test_async_search_index.py::test_search_index_drop_keys PASSED
================= 2 passed, 91 deselected, 1 warning in 6.47s =================

=== PATCHED INTEGRATION test_search_index + test_async_search_index (full) ===
======================= 93 passed, 1 warning in 10.75s ========================

=== PATCHED INTEGRATION llmcache + semantic_router (paths that go through drop_keys) ===
============ 92 passed, 1 skipped, 2 warnings in 372.78s (0:06:12) ============
```

</details>

## Notes

- Draft because I am not a regular contributor here and would rather
have a maintainer sanity-check the framing before unhiding it.
- Happy to fold in the `drop_documents` consistency work from #601 in a
separate PR, or to widen this one if you would prefer a single change
for the pair. I left them apart because the cluster hash-tag validation
in #601 is a different shape of fix.

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Low Risk**
> Low risk: a small, backwards-compatible change that swaps `DEL` for
`UNLINK` to reduce Redis blocking during bulk key deletion; behavior
differences are mostly limited to asynchronous memory reclamation
timing.
> 
> **Overview**
> **Switches `drop_keys` to non-blocking deletion.**
`SearchIndex.drop_keys` and `AsyncSearchIndex.drop_keys` now call Redis
`UNLINK` instead of `DEL`, and their docstrings document the rationale
(avoid blocking Redis when dropping large key sets).
> 
> Adds a unit regression suite (`tests/unit/test_drop_keys_unlink.py`)
asserting both sync and async `drop_keys` paths call `unlink` (single
key and list) and never call `delete`.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
4fbc650cdb230d56bc1ebe3b048a88c4cf07d678. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: Joshuaakaspace

redisvl/index/index.py

@@ -826,16 +826,21 @@ def invalidate_sql_schema_cache(self) -> None:
     def drop_keys(self, keys: str | list[str]) -> int:
         """Remove a specific entry or entries from the index by it's key ID.
 
+        Uses ``UNLINK`` rather than ``DEL`` so memory reclamation runs on a
+        background thread. This avoids blocking the main thread when a large
+        number of keys are dropped at once (for example, scope-targeted
+        ``SemanticCache`` invalidation). The returned count is unchanged.
+
         Args:
             keys (Union[str, List[str]]): The document ID or IDs to remove from the index.
 
         Returns:
             int: Count of records deleted from Redis.
         """
         if isinstance(keys, list):
-            return self._redis_client.delete(*keys)  # type: ignore
+            return self._redis_client.unlink(*keys)  # type: ignore
         else:
-            return self._redis_client.delete(keys)  # type: ignore
+            return self._redis_client.unlink(keys)  # type: ignore
 
     def drop_documents(self, ids: str | list[str]) -> int:
         """Remove documents from the index by their document IDs.
@@ -1779,6 +1784,11 @@ def invalidate_sql_schema_cache(self) -> None:
     async def drop_keys(self, keys: str | list[str]) -> int:
         """Remove a specific entry or entries from the index by it's key ID.
 
+        Uses ``UNLINK`` rather than ``DEL`` so memory reclamation runs on a
+        background thread. This avoids blocking the main thread when a large
+        number of keys are dropped at once (for example, scope-targeted
+        ``SemanticCache`` invalidation). The returned count is unchanged.
+
         Args:
             keys (Union[str, List[str]]): The document ID or IDs to remove from the index.
 
@@ -1787,9 +1797,9 @@ async def drop_keys(self, keys: str | list[str]) -> int:
         """
         client = await self._get_client()
         if isinstance(keys, list):
-            return await client.delete(*keys)
+            return await client.unlink(*keys)
         else:
-            return await client.delete(keys)
+            return await client.unlink(keys)
 
     async def drop_documents(self, ids: str | list[str]) -> int:
         """Remove documents from the index by their document IDs.

tests/unit/test_drop_keys_unlink.py

@@ -0,0 +1,82 @@
+"""Unit tests for SearchIndex/AsyncSearchIndex.drop_keys using UNLINK (issue #600)."""
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from redisvl.index import AsyncSearchIndex, SearchIndex
+from redisvl.schema import IndexSchema
+
+
+def _schema() -> IndexSchema:
+    return IndexSchema.from_dict(
+        {
+            "index": {"name": "drop_keys_test", "prefix": "drop_keys_test"},
+            "fields": [{"name": "id", "type": "tag"}],
+        }
+    )
+
+
+class TestDropKeysUsesUnlink:
+    """SearchIndex.drop_keys should issue UNLINK, not DEL.
+
+    UNLINK reclaims memory on a background thread; DEL reclaims on the main
+    thread and stalls the server when dropping a large key set (for example,
+    scope-targeted SemanticCache invalidation).
+    """
+
+    def test_single_key_calls_unlink(self):
+        client = MagicMock()
+        client.unlink.return_value = 1
+        client.delete.return_value = 1
+        index = SearchIndex(schema=_schema(), redis_client=client)
+
+        result = index.drop_keys("drop_keys_test:1")
+
+        assert result == 1
+        client.unlink.assert_called_once_with("drop_keys_test:1")
+        client.delete.assert_not_called()
+
+    def test_list_of_keys_calls_unlink(self):
+        client = MagicMock()
+        client.unlink.return_value = 3
+        client.delete.return_value = 3
+        index = SearchIndex(schema=_schema(), redis_client=client)
+
+        keys = ["drop_keys_test:1", "drop_keys_test:2", "drop_keys_test:3"]
+        result = index.drop_keys(keys)
+
+        assert result == 3
+        client.unlink.assert_called_once_with(*keys)
+        client.delete.assert_not_called()
+
+
+class TestAsyncDropKeysUsesUnlink:
+    """AsyncSearchIndex.drop_keys should issue UNLINK, not DEL."""
+
+    @pytest.mark.asyncio
+    async def test_single_key_calls_unlink(self):
+        client = MagicMock()
+        client.unlink = AsyncMock(return_value=1)
+        client.delete = AsyncMock(return_value=1)
+        index = AsyncSearchIndex(schema=_schema(), redis_client=client)
+
+        result = await index.drop_keys("drop_keys_test:1")
+
+        assert result == 1
+        client.unlink.assert_awaited_once_with("drop_keys_test:1")
+        client.delete.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_list_of_keys_calls_unlink(self):
+        client = MagicMock()
+        client.unlink = AsyncMock(return_value=3)
+        client.delete = AsyncMock(return_value=3)
+        index = AsyncSearchIndex(schema=_schema(), redis_client=client)
+
+        keys = ["drop_keys_test:1", "drop_keys_test:2", "drop_keys_test:3"]
+        result = await index.drop_keys(keys)
+
+        assert result == 3
+        client.unlink.assert_awaited_once_with(*keys)
+        client.delete.assert_not_awaited()