Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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=<host> -Dredis.test.port=<port>`, or against a real Cluster with
`-Dredis.test.cluster=<host1:port1>,<host2:port2>,<host3:port3>`. 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.
Original file line number Diff line number Diff line change
Expand Up @@ -183,18 +183,7 @@ public void saveAll(String conversationId, List<Message> messages) {
Assert.noNullElements(messages, "messages cannot contain null elements");
String key = getKeyPrefix() + conversationId;
List<String> 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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -213,18 +213,7 @@ public void saveAll(String conversationId, List<Message> messages) {
Assert.noNullElements(messages, "messages cannot contain null elements");
String key = getKeyPrefix() + conversationId;
List<String> 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
Expand Down
Original file line number Diff line number Diff line change
@@ -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.
*
* <p>
* 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<Long> REDIS_SCRIPT = new DefaultRedisScript<>(SCRIPT, Long.class);

private static final ConcurrentMap<Integer, String> SLOT_TAGS = new ConcurrentHashMap<>();

private RedisSnapshot() {
}

static void save(RedisTemplate<String, String> template, String key, List<String> 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();
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -168,10 +170,22 @@ public void saveAll(String conversationId, List<Message> 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<String> redisList = redissonClient.getList(getKeyPrefix() + conversationId);
redisList.delete();
List<String> 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
Expand Down
Loading
Loading