From 8e2f559db58ba8e401f11885942e1b45a9f1d617 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8C=97=E5=85=89c?= <2533995180@qq.com> Date: Mon, 13 Jul 2026 16:33:44 +0800 Subject: [PATCH 1/2] fix ERR 'EVAL' command keys must in same slot` in Redis Cluster --- .../extensions/redis/store/RedisStore.java | 27 +- .../redis/store/RedisStoreTest.java | 589 ++++++++++++++++++ 2 files changed, 611 insertions(+), 5 deletions(-) create mode 100644 agentscope-extensions/agentscope-extensions-redis/src/test/java/io/agentscope/extensions/redis/store/RedisStoreTest.java diff --git a/agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/store/RedisStore.java b/agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/store/RedisStore.java index ce027641c5..f6f1cc8b97 100644 --- a/agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/store/RedisStore.java +++ b/agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/store/RedisStore.java @@ -33,21 +33,27 @@ *

For each item with namespace {@code [a, b, c]} and key {@code k}, two Redis keys are used: * *

* *

{@code } is the namespace components joined with {@code "\0"}. * + *

The namespace path is wrapped in Redis hash tags ({@code {…}}) so that the item hash and the + * namespace index always map to the same hash slot, making Lua scripts compatible with Redis + * Cluster. Namespace segments must not contain {@code '{'} or {@code '}'}. + * *

Concurrency

* *

{@link #put} and {@link #putIfVersion} both run as a single Lua {@code EVAL}, making the * version read + hash write + index update one atomic Redis operation. This makes * {@link #putIfVersion} safe to use as a distributed compare-and-swap primitive across multiple - * processes sharing the same Redis instance. + * processes sharing the same Redis instance. Both keys referenced by each Lua script share the same + * hash tag, ensuring they reside on the same cluster slot and making the design compatible with + * Redis Cluster. * *

{@link #search} is not in the same transaction as {@link #put}: index members may * temporarily refer to an item whose hash has not yet been written (window between {@code ZADD} @@ -232,11 +238,13 @@ private Map deserialize(String json) { } private String itemKey(List namespace, String key) { - return keyPrefix + "item:" + namespacePath(namespace) + NS_SEPARATOR + key; + String ns = namespacePath(namespace); + return keyPrefix + "item:{" + ns + "}" + NS_SEPARATOR + key; } private String indexKey(List namespace) { - return keyPrefix + "idx:" + namespacePath(namespace); + String ns = namespacePath(namespace); + return keyPrefix + "idx:{" + ns + "}"; } private static String namespacePath(List namespace) { @@ -247,6 +255,15 @@ private static String namespacePath(List namespace) { if (segment == null) { throw new IllegalArgumentException("namespace segment must not be null"); } + if (segment.indexOf('{') >= 0 || segment.indexOf('}') >= 0) { + throw new IllegalArgumentException( + "namespace segment must not contain '{' or '}'" + + " (incompatible with Redis Cluster hash tags)"); + } + if (segment.indexOf('\0') >= 0) { + throw new IllegalArgumentException( + "namespace segment must not contain the NUL character"); + } if (i > 0) { sb.append(NS_SEPARATOR); } diff --git a/agentscope-extensions/agentscope-extensions-redis/src/test/java/io/agentscope/extensions/redis/store/RedisStoreTest.java b/agentscope-extensions/agentscope-extensions-redis/src/test/java/io/agentscope/extensions/redis/store/RedisStoreTest.java new file mode 100644 index 0000000000..a6e38fc977 --- /dev/null +++ b/agentscope-extensions/agentscope-extensions-redis/src/test/java/io/agentscope/extensions/redis/store/RedisStoreTest.java @@ -0,0 +1,589 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.agentscope.extensions.redis.store; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.fasterxml.jackson.databind.ObjectMapper; +import io.agentscope.harness.agent.filesystem.remote.store.StoreItem; +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; +import redis.clients.jedis.UnifiedJedis; + +class RedisStoreTest { + + @Mock private UnifiedJedis jedis; + + private AutoCloseable mocks; + + @BeforeEach + void setUp() { + mocks = MockitoAnnotations.openMocks(this); + } + + @AfterEach + void tearDown() throws Exception { + reset(jedis); + if (mocks != null) { + mocks.close(); + } + } + + // ------------------------------------------------------------------------- + // Constructor tests + // ------------------------------------------------------------------------- + + @Test + void constructorRejectsNullJedis() { + assertThrows(NullPointerException.class, () -> new RedisStore(null)); + } + + @Test + void constructorRejectsNullObjectMapper() { + assertThrows(NullPointerException.class, () -> new RedisStore(jedis, "prefix", null)); + } + + @Test + void constructorWithDefaultPrefix() { + RedisStore store = new RedisStore(jedis); + assertNotNull(store); + } + + @Test + void constructorNormalizesPrefixWithoutTrailingColon() { + RedisStore store = new RedisStore(jedis, "myprefix"); + assertNotNull(store); + // Verify via key generation by doing a put — the key should use "myprefix:" + when(jedis.eval(anyString(), anyList(), anyList())).thenReturn("1"); + store.put(List.of("ns"), "key", Map.of("k", "v")); + verify(jedis) + .eval( + anyString(), + eq(List.of("myprefix:item:{ns}\0key", "myprefix:idx:{ns}")), + anyList()); + } + + @Test + void constructorNormalizesPrefixWithTrailingColon() { + RedisStore store = new RedisStore(jedis, "myprefix:"); + when(jedis.eval(anyString(), anyList(), anyList())).thenReturn("1"); + store.put(List.of("ns"), "key", Map.of("k", "v")); + verify(jedis) + .eval( + anyString(), + eq(List.of("myprefix:item:{ns}\0key", "myprefix:idx:{ns}")), + anyList()); + } + + @Test + void constructorFallsBackToDefaultOnBlankPrefix() { + RedisStore store = new RedisStore(jedis, ""); + when(jedis.eval(anyString(), anyList(), anyList())).thenReturn("1"); + store.put(List.of("ns"), "key", Map.of("k", "v")); + verify(jedis) + .eval( + anyString(), + eq(List.of("agentscope:store:item:{ns}\0key", "agentscope:store:idx:{ns}")), + anyList()); + } + + // ------------------------------------------------------------------------- + // Hash tag / key generation tests + // ------------------------------------------------------------------------- + + @Test + void itemKeyContainsHashTag() { + RedisStore store = new RedisStore(jedis); + when(jedis.eval(anyString(), anyList(), anyList())).thenReturn("1"); + store.put(List.of("a", "b"), "mykey", Map.of()); + + // item key = prefix + "item:{" + ns + "}\0" + key + // idx key = prefix + "idx:{" + ns + "}" + verify(jedis) + .eval( + anyString(), + eq( + List.of( + "agentscope:store:item:{a\0b}\0mykey", + "agentscope:store:idx:{a\0b}")), + anyList()); + } + + @Test + void itemKeyAndIndexKeyShareSameHashTag() { + RedisStore store = new RedisStore(jedis); + when(jedis.eval(anyString(), anyList(), anyList())).thenReturn("1"); + store.put(List.of("sessions", "sid123", "fs"), "file.txt", Map.of("data", 1)); + + List expectedKeys = + List.of( + "agentscope:store:item:{sessions\0sid123\0fs}\0file.txt", + "agentscope:store:idx:{sessions\0sid123\0fs}"); + verify(jedis).eval(anyString(), eq(expectedKeys), anyList()); + } + + @Test + void hashTagContainsOnlyNamespacePath() { + // Verify that the hash tag wraps only the namespace, not the key + RedisStore store = new RedisStore(jedis); + when(jedis.eval(anyString(), anyList(), anyList())).thenReturn("1"); + store.put(List.of("ns"), "k", Map.of()); + + // The "{" starts right after "item:" and "}" is before "\0k" + verify(jedis) + .eval( + anyString(), + eq(List.of("agentscope:store:item:{ns}\0k", "agentscope:store:idx:{ns}")), + anyList()); + } + + @Test + void singleComponentNamespaceHasHashTag() { + RedisStore store = new RedisStore(jedis); + when(jedis.eval(anyString(), anyList(), anyList())).thenReturn("1"); + store.put(List.of("single"), "k", Map.of()); + + verify(jedis) + .eval( + anyString(), + eq( + List.of( + "agentscope:store:item:{single}\0k", + "agentscope:store:idx:{single}")), + anyList()); + } + + // ------------------------------------------------------------------------- + // Namespace validation tests + // ------------------------------------------------------------------------- + + @Test + void namespaceSegmentRejectsOpenBrace() { + RedisStore store = new RedisStore(jedis); + assertThrows( + IllegalArgumentException.class, + () -> store.put(List.of("ns{inner"), "k", Map.of())); + } + + @Test + void namespaceSegmentRejectsCloseBrace() { + RedisStore store = new RedisStore(jedis); + assertThrows( + IllegalArgumentException.class, () -> store.put(List.of("ns}end"), "k", Map.of())); + } + + @Test + void namespaceSegmentRejectsBothBraces() { + RedisStore store = new RedisStore(jedis); + assertThrows( + IllegalArgumentException.class, () -> store.put(List.of("{full}"), "k", Map.of())); + } + + @Test + void namespaceSegmentRejectsBracesInMiddleSegment() { + RedisStore store = new RedisStore(jedis); + assertThrows( + IllegalArgumentException.class, + () -> store.put(Arrays.asList("valid", "in{valid", "also-valid"), "k", Map.of())); + } + + @Test + void namespaceCannotBeNull() { + RedisStore store = new RedisStore(jedis); + assertThrows(NullPointerException.class, () -> store.put(null, "k", Map.of())); + } + + @Test + void namespaceSegmentCannotBeNull() { + RedisStore store = new RedisStore(jedis); + assertThrows( + IllegalArgumentException.class, + () -> store.put(Arrays.asList("a", null), "k", Map.of())); + } + + @Test + void namespaceSegmentCannotContainNul() { + RedisStore store = new RedisStore(jedis); + assertThrows( + IllegalArgumentException.class, () -> store.put(List.of("a\0b"), "k", Map.of())); + } + + // ------------------------------------------------------------------------- + // Key validation tests + // ------------------------------------------------------------------------- + + @Test + void keyCannotBeNull() { + RedisStore store = new RedisStore(jedis); + assertThrows( + IllegalArgumentException.class, () -> store.put(List.of("ns"), null, Map.of())); + } + + @Test + void keyCannotBeEmpty() { + RedisStore store = new RedisStore(jedis); + assertThrows(IllegalArgumentException.class, () -> store.put(List.of("ns"), "", Map.of())); + } + + @Test + void keyCannotContainNul() { + RedisStore store = new RedisStore(jedis); + assertThrows( + IllegalArgumentException.class, + () -> store.put(List.of("ns"), "bad\0key", Map.of())); + } + + // ------------------------------------------------------------------------- + // putIfVersion validation tests + // ------------------------------------------------------------------------- + + @Test + void putIfVersionRejectsNegativeExpectedVersion() { + RedisStore store = new RedisStore(jedis); + assertThrows( + IllegalArgumentException.class, + () -> store.putIfVersion(List.of("ns"), "k", Map.of(), -1L)); + } + + // ------------------------------------------------------------------------- + // get tests + // ------------------------------------------------------------------------- + + @Test + void getReturnsItem() { + Map hash = new LinkedHashMap<>(); + hash.put("value", "{\"k\":\"v\"}"); + hash.put("version", "3"); + when(jedis.hgetAll("agentscope:store:item:{ns}\0mykey")).thenReturn(hash); + + RedisStore store = new RedisStore(jedis); + StoreItem item = store.get(List.of("ns"), "mykey"); + + assertNotNull(item); + assertEquals("mykey", item.key()); + assertEquals("v", item.value().get("k")); + assertEquals(3L, item.version()); + } + + @Test + void getReturnsNullWhenNotFound() { + when(jedis.hgetAll(anyString())).thenReturn(Collections.emptyMap()); + + RedisStore store = new RedisStore(jedis); + assertNull(store.get(List.of("ns"), "nope")); + } + + @Test + void getReturnsNullWhenRedisReturnsNull() { + when(jedis.hgetAll(anyString())).thenReturn(null); + + RedisStore store = new RedisStore(jedis); + assertNull(store.get(List.of("ns"), "nope")); + } + + @Test + void getToleratesInvalidVersion() { + Map hash = new LinkedHashMap<>(); + hash.put("value", "{}"); + hash.put("version", "not-a-number"); + when(jedis.hgetAll(anyString())).thenReturn(hash); + + RedisStore store = new RedisStore(jedis); + StoreItem item = store.get(List.of("ns"), "k"); + + assertNotNull(item); + assertEquals(0L, item.version()); + } + + // ------------------------------------------------------------------------- + // put tests + // ------------------------------------------------------------------------- + + @Test + void putEvalsScriptWithCorrectKeys() { + when(jedis.eval(anyString(), anyList(), anyList())).thenReturn("1"); + + RedisStore store = new RedisStore(jedis); + store.put(List.of("a", "b"), "k", Map.of("x", 1)); + + verify(jedis) + .eval( + eq(RedisStoreHelper.PUT_SCRIPT), + eq( + List.of( + "agentscope:store:item:{a\0b}\0k", + "agentscope:store:idx:{a\0b}")), + anyList()); + } + + @Test + void putSerializesNullValueAsEmptyMap() { + when(jedis.eval(anyString(), anyList(), anyList())).thenReturn("1"); + + RedisStore store = new RedisStore(jedis); + store.put(List.of("ns"), "k", null); + + verify(jedis).eval(anyString(), anyList(), eq(List.of("{}", "k"))); + } + + // ------------------------------------------------------------------------- + // putIfVersion tests + // ------------------------------------------------------------------------- + + @Test + void putIfVersionReturnsTrueOnSuccess() { + when(jedis.eval(anyString(), anyList(), anyList())).thenReturn("5"); + + RedisStore store = new RedisStore(jedis); + assertTrue(store.putIfVersion(List.of("ns"), "k", Map.of(), 4L)); + } + + @Test + void putIfVersionReturnsFalseOnVersionMismatch() { + when(jedis.eval(anyString(), anyList(), anyList())).thenReturn("0"); + + RedisStore store = new RedisStore(jedis); + assertFalse(store.putIfVersion(List.of("ns"), "k", Map.of(), 999L)); + } + + @Test + void putIfVersionPassesExpectedVersionAsArg() { + when(jedis.eval(anyString(), anyList(), anyList())).thenReturn("1"); + + RedisStore store = new RedisStore(jedis); + store.putIfVersion(List.of("ns"), "k", Map.of("v", 1), 0L); + + verify(jedis) + .eval( + eq(RedisStoreHelper.PUT_IF_VERSION_SCRIPT), + eq(List.of("agentscope:store:item:{ns}\0k", "agentscope:store:idx:{ns}")), + eq(List.of("{\"v\":1}", "k", "0"))); + } + + // ------------------------------------------------------------------------- + // search tests + // ------------------------------------------------------------------------- + + @Test + void searchReturnsEmptyForNonPositiveLimit() { + RedisStore store = new RedisStore(jedis); + assertTrue(store.search(List.of("ns"), 0, 0).isEmpty()); + assertTrue(store.search(List.of("ns"), -1, 0).isEmpty()); + } + + @Test + void searchReturnsEmptyWhenNoMembers() { + when(jedis.zrangeByLex(anyString(), anyString(), anyString(), anyInt(), anyInt())) + .thenReturn(Collections.emptyList()); + + RedisStore store = new RedisStore(jedis); + assertTrue(store.search(List.of("ns"), 10, 0).isEmpty()); + } + + @Test + void searchSkipsStaleIndexEntries() { + when(jedis.zrangeByLex(anyString(), anyString(), anyString(), anyInt(), anyInt())) + .thenReturn(List.of("k1", "k2")); + when(jedis.hgetAll("agentscope:store:item:{ns}\0k1")) + .thenReturn(Map.of("value", "{\"a\":1}", "version", "1")); + when(jedis.hgetAll("agentscope:store:item:{ns}\0k2")).thenReturn(Collections.emptyMap()); + + RedisStore store = new RedisStore(jedis); + List items = store.search(List.of("ns"), 10, 0); + + assertEquals(1, items.size()); + assertEquals("k1", items.get(0).key()); + } + + @Test + void searchAppliesOffsetAndLimit() { + when(jedis.zrangeByLex(anyString(), anyString(), anyString(), anyInt(), anyInt())) + .thenReturn(List.of("k1")); + when(jedis.hgetAll("agentscope:store:item:{ns}\0k1")) + .thenReturn(Map.of("value", "{}", "version", "1")); + + RedisStore store = new RedisStore(jedis); + store.search(List.of("ns"), 5, 3); + + verify(jedis).zrangeByLex("agentscope:store:idx:{ns}", "-", "+", 3, 5); + } + + @Test + void searchHandlesNullReturnFromZrange() { + when(jedis.zrangeByLex(anyString(), anyString(), anyString(), anyInt(), anyInt())) + .thenReturn(null); + + RedisStore store = new RedisStore(jedis); + assertTrue(store.search(List.of("ns"), 10, 0).isEmpty()); + } + + @Test + void searchUsesHashTagKeyForIndex() { + when(jedis.zrangeByLex(anyString(), anyString(), anyString(), anyInt(), anyInt())) + .thenReturn(Collections.emptyList()); + + RedisStore store = new RedisStore(jedis); + store.search(List.of("a", "b"), 10, 0); + + verify(jedis).zrangeByLex("agentscope:store:idx:{a\0b}", "-", "+", 0, 10); + } + + // ------------------------------------------------------------------------- + // delete tests + // ------------------------------------------------------------------------- + + @Test + void deleteEvalsScriptWithCorrectKeys() { + when(jedis.eval(anyString(), anyList(), anyList())).thenReturn(1L); + + RedisStore store = new RedisStore(jedis); + store.delete(List.of("ns"), "k"); + + verify(jedis) + .eval( + eq(RedisStoreHelper.DELETE_SCRIPT), + eq(List.of("agentscope:store:item:{ns}\0k", "agentscope:store:idx:{ns}")), + eq(List.of("k"))); + } + + // ------------------------------------------------------------------------- + // Serialization edge cases + // ------------------------------------------------------------------------- + + @Test + void deserializeHandlesNullJson() { + Map hash = new LinkedHashMap<>(); + hash.put("value", null); + hash.put("version", "1"); + when(jedis.hgetAll(anyString())).thenReturn(hash); + + RedisStore store = new RedisStore(jedis); + StoreItem item = store.get(List.of("ns"), "k"); + + assertNotNull(item); + assertTrue(item.value().isEmpty()); + } + + @Test + void deserializeHandlesEmptyJson() { + Map hash = new LinkedHashMap<>(); + hash.put("value", ""); + hash.put("version", "1"); + when(jedis.hgetAll(anyString())).thenReturn(hash); + + RedisStore store = new RedisStore(jedis); + StoreItem item = store.get(List.of("ns"), "k"); + + assertNotNull(item); + assertTrue(item.value().isEmpty()); + } + + @Test + void deserializeInvalidJsonThrows() { + Map hash = new LinkedHashMap<>(); + hash.put("value", "not-json"); + hash.put("version", "1"); + when(jedis.hgetAll(anyString())).thenReturn(hash); + + RedisStore store = new RedisStore(jedis); + assertThrows(IllegalStateException.class, () -> store.get(List.of("ns"), "k")); + } + + @Test + void serializeFailureThrows() { + ObjectMapper failingMapper = + new ObjectMapper() { + @Override + public String writeValueAsString(Object value) + throws com.fasterxml.jackson.core.JsonProcessingException { + throw new com.fasterxml.jackson.core.JsonProcessingException("fail") {}; + } + }; + RedisStore store = new RedisStore(jedis, "prefix", failingMapper); + assertThrows( + IllegalArgumentException.class, + () -> store.put(List.of("ns"), "k", Map.of("x", "y"))); + } + + // ------------------------------------------------------------------------- + // Custom ObjectMapper test + // ------------------------------------------------------------------------- + + @Test + void customObjectMapperRoundTrip() { + ObjectMapper mapper = new ObjectMapper(); + Map hash = new LinkedHashMap<>(); + hash.put("value", "{\"custom\":true}"); + hash.put("version", "7"); + when(jedis.hgetAll("agentscope:store:item:{ns}\0k")).thenReturn(hash); + when(jedis.eval(anyString(), anyList(), anyList())).thenReturn("1"); + + RedisStore store = new RedisStore(jedis, "agentscope:store:", mapper); + store.put(List.of("ns"), "k", Map.of("custom", true)); + StoreItem item = store.get(List.of("ns"), "k"); + + assertEquals(Boolean.TRUE, item.value().get("custom")); + assertEquals(7L, item.version()); + } + + /** + * Helper to access private script constants for verification in tests. Uses reflection because + * the scripts are private static fields in {@link RedisStore}. + */ + static final class RedisStoreHelper { + static final String PUT_SCRIPT; + static final String PUT_IF_VERSION_SCRIPT; + static final String DELETE_SCRIPT; + + static { + try { + var putField = RedisStore.class.getDeclaredField("PUT_SCRIPT"); + putField.setAccessible(true); + PUT_SCRIPT = (String) putField.get(null); + + var casField = RedisStore.class.getDeclaredField("PUT_IF_VERSION_SCRIPT"); + casField.setAccessible(true); + PUT_IF_VERSION_SCRIPT = (String) casField.get(null); + + var delField = RedisStore.class.getDeclaredField("DELETE_SCRIPT"); + delField.setAccessible(true); + DELETE_SCRIPT = (String) delField.get(null); + } catch (Exception e) { + throw new RuntimeException("Failed to access RedisStore script fields", e); + } + } + } +} From da37a72a1984c29e60fa004b101d1c2e533e8422 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8C=97=E5=85=89c?= <2533995180@qq.com> Date: Tue, 14 Jul 2026 23:51:09 +0800 Subject: [PATCH 2/2] [feature] expose typed events --- .../extensions/redis/store/RedisStore.java | 171 +++++++++++++ .../redis/store/RedisStoreTest.java | 240 ++++++++++++++++++ 2 files changed, 411 insertions(+) diff --git a/agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/store/RedisStore.java b/agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/store/RedisStore.java index f6f1cc8b97..323387af80 100644 --- a/agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/store/RedisStore.java +++ b/agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/store/RedisStore.java @@ -59,6 +59,14 @@ * temporarily refer to an item whose hash has not yet been written (window between {@code ZADD} * and {@code HSET}, which is closed by the Lua atomicity above) or to an item that was * concurrently deleted. The implementation tolerates the latter by skipping missing items. + * + *

Migration from pre-hash-tag format

+ * + *

Versions prior to the Redis Cluster fix used key patterns {@code item:\0} and + * {@code idx:} (no hash tags). This version uses {@code item:{}\0} and + * {@code idx:{}}. Existing data stored under the old format will not be visible to the + * new code. Use {@link #migrateFromLegacyFormat(UnifiedJedis, String)} to rename old keys to the + * new format before deploying this version. */ public class RedisStore implements BaseStore { @@ -128,6 +136,11 @@ public RedisStore(UnifiedJedis jedis, String keyPrefix) { public RedisStore(UnifiedJedis jedis, String keyPrefix, ObjectMapper objectMapper) { this.jedis = Objects.requireNonNull(jedis, "jedis must not be null"); this.keyPrefix = normalizePrefix(keyPrefix); + if (this.keyPrefix.indexOf('{') >= 0 || this.keyPrefix.indexOf('}') >= 0) { + throw new IllegalArgumentException( + "keyPrefix must not contain '{' or '}'" + + " (incompatible with Redis Cluster hash tags)"); + } this.objectMapper = Objects.requireNonNull(objectMapper, "objectMapper must not be null"); } @@ -291,6 +304,164 @@ private static String asString(Object eval) { return eval.toString(); } + // ------------------------------------------------------------------------- + // Migration + // ------------------------------------------------------------------------- + + /** + * Migrates keys from the pre-hash-tag format to the current hash-tag format. + * + *

Before this version, keys were stored as {@code item:\0} and {@code + * idx:}. The current format wraps the namespace in Redis hash tags: {@code + * item:{}\0} and {@code idx:{}}. This method scans for old-format + * keys and renames them using the Redis {@code RENAME} command. + * + *

Important: this method must be called before any {@link RedisStore} + * instance using the same prefix starts writing data, to avoid conflicts. It is the caller's + * responsibility to ensure no concurrent writes are happening during migration. + * + *

In a Redis Cluster deployment, each {@code RENAME} only works when both the old and new + * key happen to be on the same slot. Since the old keys had no hash tags, they may reside on + * different slots than the new keys. This method handles this by falling back to a copy-then- + * delete strategy when {@code RENAME} fails with a cross-slot error. + * + * @param jedis initialized Jedis client; must not be {@code null} + * @param keyPrefix the key prefix used by the store; if blank, {@link #DEFAULT_KEY_PREFIX} is + * used + * @return the number of keys migrated + */ + public static long migrateFromLegacyFormat(UnifiedJedis jedis, String keyPrefix) { + Objects.requireNonNull(jedis, "jedis must not be null"); + String prefix = normalizePrefix(keyPrefix); + if (prefix.indexOf('{') >= 0 || prefix.indexOf('}') >= 0) { + throw new IllegalArgumentException( + "keyPrefix must not contain '{' or '}'" + + " (incompatible with Redis Cluster hash tags)"); + } + + String itemPattern = prefix + "item:*"; + String idxPattern = prefix + "idx:*"; + + long migrated = 0; + migrated += migrateKeysByPattern(jedis, itemPattern, prefix, "item:"); + migrated += migrateKeysByPattern(jedis, idxPattern, prefix, "idx:"); + return migrated; + } + + /** + * Scans keys matching the given glob pattern and renames them from the old format (no hash + * tags) to the new format (with hash tags around the namespace portion). + * + *

For an item key like {@code prefix:item:ns1\0ns2\0mykey}, the namespace portion is + * {@code ns1\0ns2} and the key suffix is {@code \0mykey}. The renamed key becomes {@code + * prefix:item:{ns1\0ns2}\0mykey}. + * + *

For an index key like {@code prefix:idx:ns1\0ns2}, the namespace portion is {@code + * ns1\0ns2} with no suffix. The renamed key becomes {@code prefix:idx:{ns1\0ns2}}. + * + *

Keys that already contain a hash tag (i.e. contain {@code '{'}) are skipped, as they are + * assumed to already be in the new format. + */ + private static long migrateKeysByPattern( + UnifiedJedis jedis, String pattern, String prefix, String typeTag) { + long migrated = 0; + String cursor = "0"; + do { + var scanResult = + jedis.scan(cursor, new redis.clients.jedis.params.ScanParams().match(pattern)); + cursor = scanResult.getCursor(); + for (String oldKey : scanResult.getResult()) { + if (oldKey == null || oldKey.indexOf('{') >= 0) { + // Already in new format or invalid — skip + continue; + } + String newKey = computeNewKey(oldKey, prefix, typeTag); + if (newKey == null || newKey.equals(oldKey)) { + continue; + } + try { + jedis.rename(oldKey, newKey); + } catch (redis.clients.jedis.exceptions.JedisDataException e) { + // Cross-slot RENAME in cluster — fall back to copy + delete + if (isCrossSlotError(e)) { + copyAndDelete(jedis, oldKey, newKey); + } else { + throw e; + } + } + migrated++; + } + } while (!"0".equals(cursor)); + return migrated; + } + + /** + * Computes the new-format key from an old-format key. + * + *

For item keys ({@code typeTag = "item:"}): the old format is {@code + * item:\0}, where {@code } may itself contain {@code \0} as a namespace + * segment separator. The last {@code \0} separates the namespace from the key suffix. + * + *

For index keys ({@code typeTag = "idx:"}): the old format is {@code idx:}, + * where the entire portion after the type tag is the namespace path (no key suffix). + */ + static String computeNewKey(String oldKey, String prefix, String typeTag) { + String afterPrefix = oldKey.substring(prefix.length() + typeTag.length()); + if (afterPrefix.isEmpty()) { + return null; + } + String ns; + String suffix; + if ("idx:".equals(typeTag)) { + // Index keys: the entire afterPrefix is the namespace, no key suffix + ns = afterPrefix; + suffix = ""; + } else { + // Item keys: the last \0 separates namespace from key suffix + int lastNul = afterPrefix.lastIndexOf('\0'); + if (lastNul >= 0) { + ns = afterPrefix.substring(0, lastNul); + suffix = afterPrefix.substring(lastNul); // includes the \0 + } else { + // Edge case: item key with no NUL — treat entire string as namespace + ns = afterPrefix; + suffix = ""; + } + } + return prefix + typeTag + "{" + ns + "}" + suffix; + } + + private static boolean isCrossSlotError(Exception e) { + String msg = e.getMessage(); + return msg != null && msg.contains("CROSSSLOT"); + } + + /** + * Copies data from oldKey to newKey and deletes oldKey. Handles both hash and zset types since + * item keys are hashes and index keys are sorted sets. + */ + private static void copyAndDelete(UnifiedJedis jedis, String oldKey, String newKey) { + String type = jedis.type(oldKey); + if ("hash".equalsIgnoreCase(type)) { + Map data = jedis.hgetAll(oldKey); + if (data != null && !data.isEmpty()) { + jedis.hset(newKey, data); + jedis.del(oldKey); + } + } else if ("zset".equalsIgnoreCase(type)) { + var members = jedis.zrangeWithScores(oldKey, 0, -1); + if (members != null && !members.isEmpty()) { + for (var tuple : members) { + jedis.zadd(newKey, tuple.getScore(), tuple.getElement()); + } + jedis.del(oldKey); + } + } else { + // Unknown type or empty — just delete the old key + jedis.del(oldKey); + } + } + private static String normalizePrefix(String prefix) { if (prefix == null || prefix.isBlank()) { return DEFAULT_KEY_PREFIX; diff --git a/agentscope-extensions/agentscope-extensions-redis/src/test/java/io/agentscope/extensions/redis/store/RedisStoreTest.java b/agentscope-extensions/agentscope-extensions-redis/src/test/java/io/agentscope/extensions/redis/store/RedisStoreTest.java index a6e38fc977..d330ce34d1 100644 --- a/agentscope-extensions/agentscope-extensions-redis/src/test/java/io/agentscope/extensions/redis/store/RedisStoreTest.java +++ b/agentscope-extensions/agentscope-extensions-redis/src/test/java/io/agentscope/extensions/redis/store/RedisStoreTest.java @@ -21,10 +21,13 @@ import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.reset; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -559,6 +562,243 @@ void customObjectMapperRoundTrip() { assertEquals(7L, item.version()); } + // ------------------------------------------------------------------------- + // Prefix validation tests + // ------------------------------------------------------------------------- + + @Test + void constructorRejectsPrefixWithOpenBrace() { + assertThrows(IllegalArgumentException.class, () -> new RedisStore(jedis, "my{app")); + } + + @Test + void constructorRejectsPrefixWithCloseBrace() { + assertThrows(IllegalArgumentException.class, () -> new RedisStore(jedis, "my}app")); + } + + @Test + void constructorRejectsPrefixWithBothBraces() { + assertThrows(IllegalArgumentException.class, () -> new RedisStore(jedis, "my{app}:")); + } + + @Test + void constructorRejectsPrefixWithBracesAfterColon() { + assertThrows( + IllegalArgumentException.class, () -> new RedisStore(jedis, "myprefix:{shard}")); + } + + @Test + void constructorAcceptsPrefixWithoutBraces() { + RedisStore store = new RedisStore(jedis, "clean-prefix:"); + assertNotNull(store); + } + + // ------------------------------------------------------------------------- + // Migration tests + // ------------------------------------------------------------------------- + + @Test + void migrateRejectsNullJedis() { + assertThrows( + NullPointerException.class, + () -> RedisStore.migrateFromLegacyFormat(null, "prefix")); + } + + @Test + void migrateRejectsPrefixWithBraces() { + assertThrows( + IllegalArgumentException.class, + () -> RedisStore.migrateFromLegacyFormat(jedis, "my{app}")); + } + + @Test + void computeNewKeyItemKeyWithSingleNamespace() { + String result = + RedisStore.computeNewKey( + "agentscope:store:item:ns\0mykey", "agentscope:store:", "item:"); + assertEquals("agentscope:store:item:{ns}\0mykey", result); + } + + @Test + void computeNewKeyItemKeyWithMultiNamespace() { + String result = + RedisStore.computeNewKey( + "agentscope:store:item:a\0b\0c\0mykey", "agentscope:store:", "item:"); + assertEquals("agentscope:store:item:{a\0b\0c}\0mykey", result); + } + + @Test + void computeNewKeyIndexKey() { + String result = + RedisStore.computeNewKey("agentscope:store:idx:a\0b", "agentscope:store:", "idx:"); + assertEquals("agentscope:store:idx:{a\0b}", result); + } + + @Test + void computeNewKeyIndexKeySingleNamespace() { + String result = + RedisStore.computeNewKey("agentscope:store:idx:myns", "agentscope:store:", "idx:"); + assertEquals("agentscope:store:idx:{myns}", result); + } + + @Test + void computeNewKeyReturnsNullForEmptyNamespace() { + // Edge case: key like "prefix:item:" with nothing after "item:" + String result = + RedisStore.computeNewKey("agentscope:store:item:", "agentscope:store:", "item:"); + assertNull(result); + } + + @Test + @SuppressWarnings("unchecked") + void migrateSkipsKeysAlreadyWithHashTags() { + var scanResult = mock(redis.clients.jedis.resps.ScanResult.class); + when(scanResult.getCursor()).thenReturn("0"); + when(scanResult.getResult()).thenReturn(List.of("agentscope:store:item:{ns}\0key")); + when(jedis.scan(anyString(), any())).thenReturn(scanResult); + + long count = RedisStore.migrateFromLegacyFormat(jedis, "agentscope:store:"); + + assertEquals(0, count); + verify(jedis, never()).rename(anyString(), anyString()); + } + + @Test + @SuppressWarnings("unchecked") + void migrateRenamesOldFormatItemKey() { + var scanResultItem = mock(redis.clients.jedis.resps.ScanResult.class); + when(scanResultItem.getCursor()).thenReturn("0"); + when(scanResultItem.getResult()).thenReturn(List.of("agentscope:store:item:ns\0mykey")); + + var scanResultIdx = mock(redis.clients.jedis.resps.ScanResult.class); + when(scanResultIdx.getCursor()).thenReturn("0"); + when(scanResultIdx.getResult()).thenReturn(List.of()); + + when(jedis.scan(anyString(), any())) + .thenReturn(scanResultItem) // first scan for item:* + .thenReturn(scanResultIdx); // second scan for idx:* + when(jedis.rename("agentscope:store:item:ns\0mykey", "agentscope:store:item:{ns}\0mykey")) + .thenReturn("OK"); + + long count = RedisStore.migrateFromLegacyFormat(jedis, "agentscope:store:"); + + assertEquals(1, count); + verify(jedis) + .rename("agentscope:store:item:ns\0mykey", "agentscope:store:item:{ns}\0mykey"); + } + + @Test + @SuppressWarnings("unchecked") + void migrateRenamesOldFormatIndexKey() { + var scanResultItem = mock(redis.clients.jedis.resps.ScanResult.class); + when(scanResultItem.getCursor()).thenReturn("0"); + when(scanResultItem.getResult()).thenReturn(List.of()); + + var scanResultIdx = mock(redis.clients.jedis.resps.ScanResult.class); + when(scanResultIdx.getCursor()).thenReturn("0"); + when(scanResultIdx.getResult()).thenReturn(List.of("agentscope:store:idx:a\0b")); + + when(jedis.scan(anyString(), any())) + .thenReturn(scanResultItem) // first scan for item:* + .thenReturn(scanResultIdx); // second scan for idx:* + when(jedis.rename("agentscope:store:idx:a\0b", "agentscope:store:idx:{a\0b}")) + .thenReturn("OK"); + + long count = RedisStore.migrateFromLegacyFormat(jedis, "agentscope:store:"); + + assertEquals(1, count); + verify(jedis).rename("agentscope:store:idx:a\0b", "agentscope:store:idx:{a\0b}"); + } + + @Test + @SuppressWarnings("unchecked") + void migrateFallsBackToCopyAndDeleteOnCrossSlotError() { + var scanResultItem = mock(redis.clients.jedis.resps.ScanResult.class); + when(scanResultItem.getCursor()).thenReturn("0"); + when(scanResultItem.getResult()).thenReturn(List.of("agentscope:store:item:ns\0k")); + + var scanResultIdx = mock(redis.clients.jedis.resps.ScanResult.class); + when(scanResultIdx.getCursor()).thenReturn("0"); + when(scanResultIdx.getResult()).thenReturn(List.of()); + + when(jedis.scan(anyString(), any())).thenReturn(scanResultItem).thenReturn(scanResultIdx); + when(jedis.rename("agentscope:store:item:ns\0k", "agentscope:store:item:{ns}\0k")) + .thenThrow( + new redis.clients.jedis.exceptions.JedisDataException( + "CROSSSLOT Keys in request don't hash to the same slot")); + when(jedis.type("agentscope:store:item:ns\0k")).thenReturn("hash"); + when(jedis.hgetAll("agentscope:store:item:ns\0k")) + .thenReturn(Map.of("value", "{\"x\":1}", "version", "3")); + when(jedis.hset( + "agentscope:store:item:{ns}\0k", + Map.of("value", "{\"x\":1}", "version", "3"))) + .thenReturn(1L); + when(jedis.del("agentscope:store:item:ns\0k")).thenReturn(1L); + + long count = RedisStore.migrateFromLegacyFormat(jedis, "agentscope:store:"); + + assertEquals(1, count); + verify(jedis) + .hset( + "agentscope:store:item:{ns}\0k", + Map.of("value", "{\"x\":1}", "version", "3")); + verify(jedis).del("agentscope:store:item:ns\0k"); + } + + @Test + @SuppressWarnings("unchecked") + void migrateFallsBackToCopyAndDeleteForZset() { + var scanResultItem = mock(redis.clients.jedis.resps.ScanResult.class); + when(scanResultItem.getCursor()).thenReturn("0"); + when(scanResultItem.getResult()).thenReturn(List.of()); + + var scanResultIdx = mock(redis.clients.jedis.resps.ScanResult.class); + when(scanResultIdx.getCursor()).thenReturn("0"); + when(scanResultIdx.getResult()).thenReturn(List.of("agentscope:store:idx:ns")); + + when(jedis.scan(anyString(), any())).thenReturn(scanResultItem).thenReturn(scanResultIdx); + when(jedis.rename("agentscope:store:idx:ns", "agentscope:store:idx:{ns}")) + .thenThrow( + new redis.clients.jedis.exceptions.JedisDataException( + "CROSSSLOT Keys in request don't hash to the same slot")); + when(jedis.type("agentscope:store:idx:ns")).thenReturn("zset"); + when(jedis.zrangeWithScores("agentscope:store:idx:ns", 0, -1)) + .thenReturn( + List.of( + new redis.clients.jedis.resps.Tuple("k1", 0.0), + new redis.clients.jedis.resps.Tuple("k2", 0.0))); + when(jedis.del("agentscope:store:idx:ns")).thenReturn(1L); + + long count = RedisStore.migrateFromLegacyFormat(jedis, "agentscope:store:"); + + assertEquals(1, count); + verify(jedis).zadd(eq("agentscope:store:idx:{ns}"), eq(0.0), eq("k1")); + verify(jedis).zadd(eq("agentscope:store:idx:{ns}"), eq(0.0), eq("k2")); + verify(jedis).del("agentscope:store:idx:ns"); + } + + @Test + @SuppressWarnings("unchecked") + void migratePropagatesNonCrossSlotErrors() { + var scanResultItem = mock(redis.clients.jedis.resps.ScanResult.class); + when(scanResultItem.getCursor()).thenReturn("0"); + when(scanResultItem.getResult()).thenReturn(List.of("agentscope:store:item:ns\0k")); + + var scanResultIdx = mock(redis.clients.jedis.resps.ScanResult.class); + when(scanResultIdx.getCursor()).thenReturn("0"); + when(scanResultIdx.getResult()).thenReturn(List.of()); + + when(jedis.scan(anyString(), any())).thenReturn(scanResultItem).thenReturn(scanResultIdx); + when(jedis.rename(anyString(), anyString())) + .thenThrow( + new redis.clients.jedis.exceptions.JedisDataException( + "ERR some other error")); + + assertThrows( + redis.clients.jedis.exceptions.JedisDataException.class, + () -> RedisStore.migrateFromLegacyFormat(jedis, "agentscope:store:")); + } + /** * Helper to access private script constants for verification in tests. Uses reflection because * the scripts are private static fields in {@link RedisStore}.