diff --git a/memory-repository/argi-model-chat-memory-repository-redis/README.md b/memory-repository/argi-model-chat-memory-repository-redis/README.md index fc433b0e..0630326e 100644 --- a/memory-repository/argi-model-chat-memory-repository-redis/README.md +++ b/memory-repository/argi-model-chat-memory-repository-redis/README.md @@ -444,3 +444,48 @@ The integration tests verify actual keys and read/write/list/delete behavior for all nine client pairings, default and shared prefixes, and concurrent construction. They use unique test keys and never flush a database. These standalone tests do not certify TLS, Redis Cluster, or high-availability behavior. + +## Snapshot replacement with `saveAll` + +Jedis, Lettuce, and Redisson validate and serialize the entire replacement before +sending it to Redis. A nonempty snapshot is built in a unique temporary list and +published with `RENAME` inside one Lua invocation. Concurrent `saveAll` calls and +reads therefore see complete snapshots, without a delete/write gap or mixed lists. +The last publisher wins; an application's separate read-modify-write sequence is +still not atomic. + +Conversation keys, prefixes, message order, and list encoding remain unchanged. +The temporary key shares the destination's actual Cluster slot, including for +untagged keys and names containing braces. Redisson keeps a configured value codec +and destination name mapper. Successful replacement removes any previous TTL; +an empty snapshot deletes the conversation key. Local validation/serialization +failures and Redis command rejections before publication preserve the old values +and expiration, subject to natural expiry. + +Redis scripts do not roll back earlier commands on error. This script checks +cleanup permission before creating its temporary list, catches staging and +publication command errors, and deletes its temporary list on failure. There is +no client-side staging period, lease, or timeout retry in the repository. A +transport error or timeout can leave the publication outcome unknown: a completed +server-side publication is not promised to roll back. + +### Redis ACL compatibility + +In addition to the existing connection and read permissions, nonempty saves now +need `EVAL`/`EVALSHA`, `EXISTS`, `DEL`, `RPUSH`, and `RENAME`, plus access to both the +conversation namespace and temporary names matching `{argi-*}:snapshot:*`. +Restricting access to only the conversation prefix is insufficient. Review these +permissions before upgrading; a denied cleanup or script permission fails the +save before staging, and a denied publication command retains the old snapshot. +No conversation data migration is required. + +### Snapshot tests + +Run `RedisSnapshotIntegrationTest` against an isolated standalone Redis with +`-Dredis.test.host= -Dredis.test.port=`, or against a real Cluster with +`-Dredis.test.cluster=,,`. Without an explicit +endpoint it uses Testcontainers and skips when Docker is unavailable. The same +contract covers all three clients, concurrent readers/writers, legacy records, +TTL/empty snapshots, local failures, denied commands, staging collisions, and +Redisson's configured codec/name mapping. The endpoint must be test-owned: these +tests create and remove temporary ACL users on every supplied node. diff --git a/memory-repository/argi-model-chat-memory-repository-redis/src/main/java/io/github/agentic/ai/memory/redis/JedisRedisChatMemoryRepository.java b/memory-repository/argi-model-chat-memory-repository-redis/src/main/java/io/github/agentic/ai/memory/redis/JedisRedisChatMemoryRepository.java index a9689d38..71453348 100644 --- a/memory-repository/argi-model-chat-memory-repository-redis/src/main/java/io/github/agentic/ai/memory/redis/JedisRedisChatMemoryRepository.java +++ b/memory-repository/argi-model-chat-memory-repository-redis/src/main/java/io/github/agentic/ai/memory/redis/JedisRedisChatMemoryRepository.java @@ -183,18 +183,7 @@ public void saveAll(String conversationId, List messages) { Assert.noNullElements(messages, "messages cannot contain null elements"); String key = getKeyPrefix() + conversationId; List messageJsons = messages.stream().map(this::serializeMessage).toList(); - RedisConnectionFactory redisConnectionFactory = Objects.requireNonNull(redisTemplate.getConnectionFactory(), - "RedisConnectionFactory cannot be null"); - try (RedisConnection connection = redisConnectionFactory.getConnection()) { - connection.keyCommands().del(key.getBytes()); - if (!messageJsons.isEmpty()) { - byte[][] values = new byte[messageJsons.size()][]; - for (int i = 0; i < messageJsons.size(); i++) { - values[i] = messageJsons.get(i).getBytes(); - } - connection.listCommands().rPush(key.getBytes(), values); - } - } + RedisSnapshot.save(redisTemplate, key, messageJsons); } @Override diff --git a/memory-repository/argi-model-chat-memory-repository-redis/src/main/java/io/github/agentic/ai/memory/redis/LettuceRedisChatMemoryRepository.java b/memory-repository/argi-model-chat-memory-repository-redis/src/main/java/io/github/agentic/ai/memory/redis/LettuceRedisChatMemoryRepository.java index e11c1087..baa4e28a 100644 --- a/memory-repository/argi-model-chat-memory-repository-redis/src/main/java/io/github/agentic/ai/memory/redis/LettuceRedisChatMemoryRepository.java +++ b/memory-repository/argi-model-chat-memory-repository-redis/src/main/java/io/github/agentic/ai/memory/redis/LettuceRedisChatMemoryRepository.java @@ -213,18 +213,7 @@ public void saveAll(String conversationId, List messages) { Assert.noNullElements(messages, "messages cannot contain null elements"); String key = getKeyPrefix() + conversationId; List messageJsons = messages.stream().map(this::serializeMessage).toList(); - RedisConnectionFactory redisConnectionFactory = - Objects.requireNonNull(redisTemplate.getConnectionFactory(), "RedisConnectionFactory cannot be null"); - try (RedisConnection connection = redisConnectionFactory.getConnection()) { - connection.keyCommands().del(key.getBytes()); - if (!messageJsons.isEmpty()) { - byte[][] values = new byte[messageJsons.size()][]; - for (int i = 0; i < messageJsons.size(); i++) { - values[i] = messageJsons.get(i).getBytes(); - } - connection.listCommands().rPush(key.getBytes(), values); - } - } + RedisSnapshot.save(redisTemplate, key, messageJsons); } @Override diff --git a/memory-repository/argi-model-chat-memory-repository-redis/src/main/java/io/github/agentic/ai/memory/redis/RedisSnapshot.java b/memory-repository/argi-model-chat-memory-repository-redis/src/main/java/io/github/agentic/ai/memory/redis/RedisSnapshot.java new file mode 100644 index 00000000..8442bced --- /dev/null +++ b/memory-repository/argi-model-chat-memory-repository-redis/src/main/java/io/github/agentic/ai/memory/redis/RedisSnapshot.java @@ -0,0 +1,97 @@ +/* + * 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 + * + * https://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.github.agentic.ai.memory.redis; + +import java.util.List; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +import org.springframework.data.redis.connection.ClusterSlotHashUtil; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.core.script.DefaultRedisScript; + +/** + * Publishes a complete list without deleting the previous snapshot first. The temporary + * list exists only inside one script invocation: every command failure after creation is + * caught and cleans it up. DEL permission is checked before creating it, since Redis + * scripts do not roll back writes on error. No client-side staging or timeout retry is + * used. + * + *

+ * RENAME is the publication boundary and removes any previous destination expiration. + * Concurrent saves are last-publisher-wins; a read-modify-write sequence is not atomic. + */ +final class RedisSnapshot { + + static final String SCRIPT = """ + if #ARGV == 0 then + return redis.call('del', KEYS[1]) + end + if redis.call('exists', KEYS[2]) ~= 0 then + return redis.error_reply('Snapshot staging key already exists') + end + -- Check cleanup permission before any staging writes. Scripts cannot interleave + -- with ACL changes, and DEL is permitted even when Redis is out of memory. + redis.call('del', KEYS[2]) + for first = 1, #ARGV, 256 do + local chunk = {} + for i = first, math.min(first + 255, #ARGV) do + chunk[#chunk + 1] = ARGV[i] + end + local result = redis.pcall('rpush', KEYS[2], unpack(chunk)) + if type(result) == 'table' and result.err then + redis.call('del', KEYS[2]) + return redis.error_reply(result.err) + end + end + local result = redis.pcall('rename', KEYS[2], KEYS[1]) + if type(result) == 'table' and result.err then + redis.call('del', KEYS[2]) + return redis.error_reply(result.err) + end + return 1 + """; + + private static final DefaultRedisScript REDIS_SCRIPT = new DefaultRedisScript<>(SCRIPT, Long.class); + + private static final ConcurrentMap SLOT_TAGS = new ConcurrentHashMap<>(); + + private RedisSnapshot() { + } + + static void save(RedisTemplate template, String key, List messages) { + if (messages.isEmpty()) { + template.delete(key); + return; + } + template.execute(REDIS_SCRIPT, List.of(key, stagingKey(key)), messages.toArray()); + } + + static String stagingKey(String key) { + int slot = ClusterSlotHashUtil.calculateSlot(key); + String tag = SLOT_TAGS.computeIfAbsent(slot, target -> { + for (int candidate = 0;; candidate++) { + String value = "argi-" + candidate; + if (ClusterSlotHashUtil.calculateSlot(value) == target) { + return value; + } + } + }); + return "{" + tag + "}:snapshot:" + UUID.randomUUID(); + } + +} diff --git a/memory-repository/argi-model-chat-memory-repository-redis/src/main/java/io/github/agentic/ai/memory/redis/RedissonRedisChatMemoryRepository.java b/memory-repository/argi-model-chat-memory-repository-redis/src/main/java/io/github/agentic/ai/memory/redis/RedissonRedisChatMemoryRepository.java index 128d8e2c..953a91e2 100644 --- a/memory-repository/argi-model-chat-memory-repository-redis/src/main/java/io/github/agentic/ai/memory/redis/RedissonRedisChatMemoryRepository.java +++ b/memory-repository/argi-model-chat-memory-repository-redis/src/main/java/io/github/agentic/ai/memory/redis/RedissonRedisChatMemoryRepository.java @@ -20,6 +20,7 @@ import org.redisson.Redisson; import org.redisson.api.RKeys; import org.redisson.api.RList; +import org.redisson.api.RScript; import org.redisson.api.RedissonClient; import org.redisson.api.options.KeysScanOptions; import org.redisson.client.codec.StringCodec; @@ -32,6 +33,7 @@ import org.springframework.util.Assert; import org.springframework.util.StringUtils; +import java.nio.charset.StandardCharsets; import java.util.List; import java.util.Objects; import java.util.stream.Collectors; @@ -168,10 +170,22 @@ public void saveAll(String conversationId, List messages) { Assert.hasText(conversationId, "conversationId cannot be null or empty"); Assert.notNull(messages, "messages cannot be null"); Assert.noNullElements(messages, "messages cannot contain null elements"); - RList redisList = redissonClient.getList(getKeyPrefix() + conversationId); - redisList.delete(); List serializedMessages = messages.stream().map(this::serializeMessage).toList(); - redisList.addAll(serializedMessages); + String key = getKeyPrefix() + conversationId; + if (serializedMessages.isEmpty()) { + redissonClient.getList(key).delete(); + return; + } + String slotKey = key; + Config config = redissonClient.getConfig(); + if (config.isClusterConfig()) { + slotKey = config.useClusterServers().getNameMapper().map(key); + } + // RScript maps String keys. Pass the already slot-matched temporary name as + // bytes so a custom NameMapper cannot change its hash tag a second time. + byte[] stagingKey = RedisSnapshot.stagingKey(slotKey).getBytes(StandardCharsets.UTF_8); + redissonClient.getScript().eval(RScript.Mode.READ_WRITE, RedisSnapshot.SCRIPT, + RScript.ReturnType.INTEGER, List.of(key, stagingKey), serializedMessages.toArray()); } @Override diff --git a/memory-repository/argi-model-chat-memory-repository-redis/src/test/java/io/github/agentic/ai/memory/redis/RedisSnapshotIntegrationTest.java b/memory-repository/argi-model-chat-memory-repository-redis/src/test/java/io/github/agentic/ai/memory/redis/RedisSnapshotIntegrationTest.java new file mode 100644 index 00000000..9c170339 --- /dev/null +++ b/memory-repository/argi-model-chat-memory-repository-redis/src/test/java/io/github/agentic/ai/memory/redis/RedisSnapshotIntegrationTest.java @@ -0,0 +1,427 @@ +/* + * 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 + * + * https://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.github.agentic.ai.memory.redis; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.IntStream; + +import io.github.agentic.ai.memory.redis.builder.RedisChatMemoryBuilder; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; +import org.redisson.Redisson; +import org.redisson.api.NameMapper; +import org.redisson.codec.SerializationCodec; +import org.redisson.config.Config; +import org.springframework.ai.chat.messages.Message; +import org.springframework.ai.chat.messages.UserMessage; +import org.testcontainers.DockerClientFactory; +import org.testcontainers.containers.GenericContainer; +import redis.clients.jedis.HostAndPort; +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisCluster; +import redis.clients.jedis.JedisPooled; +import redis.clients.jedis.UnifiedJedis; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +/** + * Runs against an isolated standalone Redis by default. To exercise the identical + * contract against a real Cluster, set redis.test.cluster to comma-separated reachable + * master nodes. The endpoint must be test-owned: failure tests create and remove + * temporary ACL users. + */ +class RedisSnapshotIntegrationTest { + + private static GenericContainer redis; + + private static List nodes; + + private static String host; + + private static int port; + + private static UnifiedJedis admin; + + enum Client { + + JEDIS, LETTUCE, REDISSON + + } + + @BeforeAll + static void startRedis() { + String cluster = System.getProperty("redis.test.cluster"); + if (cluster != null) { + nodes = List.of(cluster.split(",")); + Set addresses = new HashSet<>(); + for (String node : nodes) { + addresses.add(HostAndPort.from(node)); + } + admin = new JedisCluster(addresses); + return; + } + host = System.getProperty("redis.test.host", "127.0.0.1"); + String externalPort = System.getProperty("redis.test.port"); + if (externalPort == null) { + assumeTrue(DockerClientFactory.instance().isDockerAvailable(), "Docker or an isolated Redis is required"); + redis = new GenericContainer<>("redis:7.0.15").withExposedPorts(6379); + redis.start(); + host = redis.getHost(); + port = redis.getMappedPort(6379); + } + else { + port = Integer.parseInt(externalPort); + } + admin = new JedisPooled(host, port); + } + + @AfterAll + static void stopRedis() { + if (admin != null) { + admin.close(); + } + if (redis != null) { + redis.stop(); + } + } + + private static > T configure(T builder, String prefix, String user) { + builder.keyPrefix(prefix).timeout(10000); + if (nodes != null) { + builder.nodes(nodes); + } + else { + builder.host(host).port(port); + } + if (user != null) { + builder.username(user).password("snapshot-test-password"); + } + return builder; + } + + private static BaseRedisChatMemoryRepository repository(Client client, String prefix, String user) { + return switch (client) { + case JEDIS -> configure(JedisRedisChatMemoryRepository.builder(), prefix, user).build(); + case LETTUCE -> configure(LettuceRedisChatMemoryRepository.builder(), prefix, user).build(); + case REDISSON -> configure(RedissonRedisChatMemoryRepository.builder(), prefix, user).build(); + }; + } + + private static String prefix() { + return "snapshot-test-" + UUID.randomUUID() + ":"; + } + + private static List snapshot(String name, int count) { + return IntStream.range(0, count).mapToObj(i -> (Message) new UserMessage(name + ":" + i)).toList(); + } + + private static List texts(List messages) { + return messages.stream().map(Message::getText).toList(); + } + + @ParameterizedTest + @EnumSource(Client.class) + void replacementPreservesLegacyKeysEncodingOrderAndTtlSemantics(Client writer) throws Exception { + String prefix = prefix(); + try (var repository = repository(writer, prefix, null)) { + for (String id : List.of("plain", "{tag}tail", "{}first{later}", "{{nested}", "unclosed{", "中文🙂")) { + String key = prefix + id; + try { + admin.rpush(key, repository.serializeMessage(new UserMessage("legacy"))); + admin.pexpire(key, 60000); + for (Client reader : Client.values()) { + try (var other = repository(reader, prefix, null)) { + assertThat(texts(other.findByConversationId(id))).containsExactly("legacy"); + } + } + var messages = snapshot("中文 replacement", 300); + repository.saveAll(id, messages); + assertThat(admin.lrange(key, 0, -1)) + .isEqualTo(messages.stream().map(repository::serializeMessage).toList()); + assertThat(admin.pttl(key)).isEqualTo(-1L); + for (Client reader : Client.values()) { + try (var other = repository(reader, prefix, null)) { + assertThat(texts(other.findByConversationId(id))).isEqualTo(texts(messages)); + } + } + repository.saveAll(id, List.of()); + assertThat(admin.exists(key)).isFalse(); + admin.psetex(key, 60000, "wrong type"); + repository.saveAll(id, messages); + assertThat(admin.type(key)).isEqualTo("list"); + assertThat(admin.pttl(key)).isEqualTo(-1L); + } + finally { + admin.del(key); + } + } + } + assertNoStagingKeys(); + } + + @ParameterizedTest + @EnumSource(Client.class) + void localFailuresRetainBytesAndExpiry(Client writer) throws Exception { + String prefix = prefix(); + String key = prefix + "history"; + try (var repository = repository(writer, prefix, null)) { + try { + repository.saveAll("history", snapshot("old", 2)); + admin.pexpire(key, 60000); + var old = admin.lrange(key, 0, -1); + assertThatThrownBy(() -> repository.saveAll(" ", snapshot("new", 1))) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> repository.saveAll("history", null)) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> repository.saveAll("history", Arrays.asList(new UserMessage("ok"), null))) + .isInstanceOf(IllegalArgumentException.class); + var broken = UserMessage.builder().text("bad").metadata(Map.of("cycle", new SelfReference())).build(); + for (var replacement : List.of(List.of(broken), + List.of(new UserMessage("ok"), broken))) { + assertThatThrownBy(() -> repository.saveAll("history", replacement)) + .isInstanceOf(RuntimeException.class) + .hasMessage("Error serializing message"); + } + assertThat(admin.lrange(key, 0, -1)).isEqualTo(old); + assertThat(admin.pttl(key)).isBetween(1L, 60000L); + } + finally { + admin.del(key); + } + } + assertNoStagingKeys(); + } + + @ParameterizedTest + @EnumSource(Client.class) + void rejectedCommandsDoNotPublishAndLeaveNoStagingKeys(Client writer) throws Exception { + String prefix = prefix(); + String key = prefix + "history"; + try (var seed = repository(Client.JEDIS, prefix, null)) { + try { + seed.saveAll("history", snapshot("old", 2)); + admin.pexpire(key, 60000); + var old = admin.lrange(key, 0, -1); + for (String denied : List.of("rpush", "rename", "del", "eval")) { + String user = "snapshot-" + UUID.randomUUID(); + if (denied.equals("eval")) { + acl(user, "on", ">snapshot-test-password", "~*", "+@all", "-eval", "-evalsha"); + } + else { + acl(user, "on", ">snapshot-test-password", "~*", "+@all", "-" + denied); + } + try (var repository = repository(writer, prefix, user)) { + assertThatThrownBy(() -> repository.saveAll("history", snapshot("new", 600))) + .isInstanceOf(RuntimeException.class); + assertThat(admin.lrange(key, 0, -1)).isEqualTo(old); + assertThat(admin.pttl(key)).isBetween(1L, 60000L); + assertNoStagingKeys(); + } + finally { + deleteUser(user); + } + } + } + finally { + admin.del(key); + } + } + } + + @Test + void concurrentReadersAndThreeWritersOnlyObserveWholeSnapshots() throws Exception { + String prefix = prefix(); + String id = "concurrent{}"; + String key = prefix + id; + List repositories = new ArrayList<>(); + var executor = Executors.newFixedThreadPool(6); + var running = new AtomicBoolean(true); + var ready = new CountDownLatch(6); + var start = new CountDownLatch(1); + try { + for (Client client : Client.values()) { + repositories.add(repository(client, prefix, null)); + } + var snapshots = List.of(snapshot("old", 300), snapshot("jedis", 300), snapshot("lettuce", 300), + snapshot("redisson", 300)); + Set> allowed = new HashSet<>(); + snapshots.forEach(messages -> allowed.add(texts(messages))); + repositories.get(0).saveAll(id, snapshots.get(0)); + var readers = new ArrayList>(); + var writers = new ArrayList>(); + for (int i = 0; i < 3; i++) { + final int index = i; + writers.add(executor.submit(() -> { + ready.countDown(); + assertThat(start.await(10, TimeUnit.SECONDS)).isTrue(); + for (int n = 0; n < 30; n++) { + repositories.get(index).saveAll(id, snapshots.get(index + 1)); + } + return null; + })); + readers.add(executor.submit(() -> { + ready.countDown(); + assertThat(start.await(10, TimeUnit.SECONDS)).isTrue(); + int samples = 0; + do { + assertThat(allowed).contains(texts(repositories.get(index).findByConversationId(id))); + samples++; + } + while (running.get() || samples < 30); + return samples; + })); + } + assertThat(ready.await(10, TimeUnit.SECONDS)).isTrue(); + start.countDown(); + for (var writer : writers) { + writer.get(60, TimeUnit.SECONDS); + } + running.set(false); + for (var reader : readers) { + assertThat(reader.get(60, TimeUnit.SECONDS)).isGreaterThanOrEqualTo(30); + } + assertThat(allowed).contains(texts(repositories.get(0).findByConversationId(id))); + } + finally { + running.set(false); + start.countDown(); + executor.shutdownNow(); + assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + for (var repository : repositories) { + repository.close(); + } + admin.del(key); + } + assertNoStagingKeys(); + } + + @Test + void redissonPreservesConfiguredCodecAndMappedDestination() throws Exception { + String prefix = prefix(); + String id = "custom-codec"; + String key = "mapped:" + prefix + id; + NameMapper mapper = new NameMapper() { + @Override + public String map(String name) { + return "mapped:" + name; + } + + @Override + public String unmap(String name) { + return name.substring("mapped:".length()); + } + }; + Config config = new Config().setCodec(new SerializationCodec()); + if (nodes == null) { + config.useSingleServer().setAddress("redis://" + host + ":" + port).setNameMapper(mapper); + } + else { + config.useClusterServers() + .addNodeAddress(nodes.stream().map(node -> "redis://" + node).toArray(String[]::new)) + .setNameMapper(mapper); + } + var legacy = Redisson.create(config); + try (var repository = RedissonRedisChatMemoryRepository.builder() + .redissonConfig(config) + .keyPrefix(prefix) + .build()) { + var message = new UserMessage("configured codec history"); + legacy.getList(prefix + id).add(repository.serializeMessage(message)); + var before = admin.lrange(key.getBytes(StandardCharsets.UTF_8), 0, -1); + repository.saveAll(id, List.of(message)); + assertThat(admin.lrange(key.getBytes(StandardCharsets.UTF_8), 0, -1)).usingRecursiveComparison() + .isEqualTo(before); + assertThat(texts(repository.findByConversationId(id))).containsExactly(message.getText()); + assertThat(legacy.getList(prefix + id).readAll()) + .containsExactly(repository.serializeMessage(message)); + } + finally { + legacy.shutdown(); + admin.del(key); + } + assertNoStagingKeys(); + } + + @Test + void stagingCollisionCannotOverwriteAnotherKey() { + String key = prefix() + "destination"; + String staging = RedisSnapshot.stagingKey(key); + try { + admin.rpush(key, "old"); + admin.set(staging, "other owner's value"); + assertThatThrownBy(() -> admin.eval(RedisSnapshot.SCRIPT, List.of(key, staging), List.of("new"))) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("staging key already exists"); + assertThat(admin.get(staging)).isEqualTo("other owner's value"); + assertThat(admin.lrange(key, 0, -1)).containsExactly("old"); + } + finally { + admin.del(key); + admin.del(staging); + } + } + + private static List endpoints() { + return nodes == null ? List.of(new HostAndPort(host, port)) : nodes.stream().map(HostAndPort::from).toList(); + } + + private static void acl(String user, String... rules) { + for (var endpoint : endpoints()) { + try (var connection = new Jedis(endpoint)) { + connection.aclSetUser(user, rules); + } + } + } + + private static void deleteUser(String user) { + for (var endpoint : endpoints()) { + try (var connection = new Jedis(endpoint)) { + connection.aclDelUser(user); + } + } + } + + private static void assertNoStagingKeys() { + for (var endpoint : endpoints()) { + try (var connection = new Jedis(endpoint)) { + assertThat(connection.keys("{argi-*}:snapshot:*")).isEmpty(); + } + } + } + + static class SelfReference { + + public final SelfReference self = this; + + } + +} diff --git a/memory-repository/argi-model-chat-memory-repository-redis/src/test/java/io/github/agentic/ai/memory/redis/RedissonSaveAllFailureTest.java b/memory-repository/argi-model-chat-memory-repository-redis/src/test/java/io/github/agentic/ai/memory/redis/RedissonSaveAllFailureTest.java new file mode 100644 index 00000000..85104b0d --- /dev/null +++ b/memory-repository/argi-model-chat-memory-repository-redis/src/test/java/io/github/agentic/ai/memory/redis/RedissonSaveAllFailureTest.java @@ -0,0 +1,154 @@ +/* + * 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 + * + * https://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.github.agentic.ai.memory.redis; + +import java.util.ArrayList; +import java.util.Arrays; +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.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.mockito.MockedStatic; +import org.redisson.Redisson; +import org.redisson.api.RList; +import org.redisson.api.RScript; +import org.redisson.api.RedissonClient; +import org.redisson.config.Config; +import org.springframework.ai.chat.messages.Message; +import org.springframework.ai.chat.messages.UserMessage; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +class RedissonSaveAllFailureTest { + + private final RedissonClient client = mock(RedissonClient.class); + + @SuppressWarnings("unchecked") + private final RList redisList = mock(RList.class); + + private final List stored = new ArrayList<>(); + + private MockedStatic redisson; + + private RedissonRedisChatMemoryRepository repository; + + @BeforeEach + void setUp() { + redisson = mockStatic(Redisson.class); + redisson.when(() -> Redisson.create(any(Config.class))).thenReturn(client); + repository = RedissonRedisChatMemoryRepository.builder().build(); + when(client.getList("argi_chat_memory:conversation")).thenReturn(redisList); + when(redisList.readAll()).thenAnswer(invocation -> new ArrayList<>(stored)); + when(redisList.delete()).thenAnswer(invocation -> { + boolean existed = !stored.isEmpty(); + stored.clear(); + return existed; + }); + doAnswer(invocation -> stored.addAll(invocation.getArgument(0))).when(redisList).addAll(any()); + RScript script = mock(RScript.class); + when(client.getScript()).thenReturn(script); + when(client.getConfig()).thenReturn(new Config()); + when(script.eval(eq(RScript.Mode.READ_WRITE), eq(RedisSnapshot.SCRIPT), eq(RScript.ReturnType.INTEGER), + anyList(), any(Object[].class))) + .thenAnswer(invocation -> { + stored.clear(); + for (Object value : (Object[]) invocation.getRawArguments()[4]) { + stored.add((String) value); + } + return 1L; + }); + stored.add(repository.serializeMessage(new UserMessage("old history"))); + } + + @AfterEach + void tearDown() { + redisson.close(); + } + + @ParameterizedTest + @ValueSource(ints = { 0, 1 }) + void serializationFailurePreservesOldHistory(int failureIndex) { + var replacement = new ArrayList(List.of(new UserMessage("valid replacement"))); + replacement.add(failureIndex, + UserMessage.builder() + .text("unserializable replacement") + .metadata(Map.of("cycle", new SelfReference())) + .build()); + + assertThatThrownBy(() -> repository.saveAll("conversation", replacement)).isInstanceOf(RuntimeException.class) + .hasMessage("Error serializing message"); + + assertThat(repository.findByConversationId("conversation")).extracting(Message::getText) + .containsExactly("old history"); + verify(client, never()).getScript(); + verify(redisList, never()).delete(); + verify(redisList, never()).addAll(any()); + } + + @Test + void invalidMessagesDoNotAccessRedis() { + assertThatThrownBy(() -> repository.saveAll("conversation", null)).isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> repository.saveAll("conversation", Arrays.asList(new UserMessage("valid"), null))) + .isInstanceOf(IllegalArgumentException.class); + verifyNoInteractions(client, redisList); + } + + @ParameterizedTest + @ValueSource(strings = { "", " " }) + void invalidConversationIdDoesNotAccessRedis(String conversationId) { + assertThatThrownBy(() -> repository.saveAll(conversationId, List.of(new UserMessage("valid")))) + .isInstanceOf(IllegalArgumentException.class); + verifyNoInteractions(client, redisList); + } + + @Test + void validMessagesReplaceOldHistoryInOrder() { + repository.saveAll("conversation", List.of(new UserMessage("first"), new UserMessage("second"))); + + assertThat(repository.findByConversationId("conversation")).extracting(Message::getText) + .containsExactly("first", "second"); + } + + @Test + void emptyMessagesClearOldHistory() { + repository.saveAll("conversation", List.of()); + + assertThat(repository.findByConversationId("conversation")).isEmpty(); + verify(redisList).delete(); + } + + static class SelfReference { + + public final SelfReference self = this; + + } + +} diff --git a/memory-repository/argi-model-chat-memory-repository-redis/src/test/java/io/github/agentic/ai/memory/redis/RedissonSaveAllIntegrationTest.java b/memory-repository/argi-model-chat-memory-repository-redis/src/test/java/io/github/agentic/ai/memory/redis/RedissonSaveAllIntegrationTest.java new file mode 100644 index 00000000..67342f0b --- /dev/null +++ b/memory-repository/argi-model-chat-memory-repository-redis/src/test/java/io/github/agentic/ai/memory/redis/RedissonSaveAllIntegrationTest.java @@ -0,0 +1,130 @@ +/* + * 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 + * + * https://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.github.agentic.ai.memory.redis; + +import java.util.List; +import java.util.Map; +import java.util.UUID; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.springframework.ai.chat.messages.Message; +import org.springframework.ai.chat.messages.UserMessage; +import org.testcontainers.DockerClientFactory; +import org.testcontainers.containers.GenericContainer; +import redis.clients.jedis.Jedis; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +class RedissonSaveAllIntegrationTest { + + private static GenericContainer redis; + + private static String host; + + private static int port; + + @BeforeAll + static void startRedis() { + String externalPort = System.getProperty("redis.test.port"); + if (externalPort != null) { + host = System.getProperty("redis.test.host", "127.0.0.1"); + port = Integer.parseInt(externalPort); + } + else { + assumeTrue(DockerClientFactory.instance().isDockerAvailable(), + "Docker or an explicit isolated Redis (redis.test.port) is required"); + redis = new GenericContainer<>("redis:7.0.15").withExposedPorts(6379); + redis.start(); + host = redis.getHost(); + port = redis.getMappedPort(6379); + } + } + + @AfterAll + static void stopRedis() { + if (redis != null) { + redis.stop(); + } + } + + @Test + void serializationFailureKeepsStoredHistoryAndExpiry() { + String conversationId = "saveall-failure-" + UUID.randomUUID(); + String key = "argi_chat_memory:" + conversationId; + try (var repository = RedissonRedisChatMemoryRepository.builder().host(host).port(port).build(); + var connection = new Jedis(host, port)) { + try { + repository.saveAll(conversationId, List.of(new UserMessage("old history"))); + connection.pexpire(key, 60000); + var oldRecords = connection.lrange(key, 0, -1); + var broken = UserMessage.builder() + .text("broken") + .metadata(Map.of("cycle", new SelfReference())) + .build(); + + assertThatThrownBy(() -> repository.saveAll(conversationId, + List.of(new UserMessage("valid first message"), broken))) + .isInstanceOf(RuntimeException.class) + .hasMessage("Error serializing message"); + + assertThat(connection.lrange(key, 0, -1)).isEqualTo(oldRecords); + assertThat(repository.findByConversationId(conversationId)).extracting(Message::getText) + .containsExactly("old history"); + assertThat(connection.pttl(key)).isBetween(1L, 60000L); + } + finally { + connection.del(key); + } + } + } + + @Test + void validReplacementAndEmptySnapshotKeepExistingSemantics() { + String conversationId = "saveall-valid-" + UUID.randomUUID(); + String key = "argi_chat_memory:" + conversationId; + try (var repository = RedissonRedisChatMemoryRepository.builder().host(host).port(port).build(); + var connection = new Jedis(host, port)) { + try { + repository.saveAll(conversationId, List.of(new UserMessage("old history"))); + connection.pexpire(key, 60000); + repository.saveAll(conversationId, List.of(new UserMessage("first"), new UserMessage("second"))); + + assertThat(repository.findByConversationId(conversationId)).extracting(Message::getText) + .containsExactly("first", "second"); + assertThat(connection.pttl(key)).isEqualTo(-1L); + + repository.saveAll(conversationId, List.of()); + + assertThat(repository.findByConversationId(conversationId)).isEmpty(); + assertThat(connection.exists(key)).isFalse(); + } + finally { + connection.del(key); + } + } + } + + static class SelfReference { + + public final SelfReference self = this; + + } + +}