diff --git a/graph-persistence/argi-graph-persistence-redis/README.md b/graph-persistence/argi-graph-persistence-redis/README.md index 44a9d586..90204fbf 100644 --- a/graph-persistence/argi-graph-persistence-redis/README.md +++ b/graph-persistence/argi-graph-persistence-redis/README.md @@ -313,6 +313,243 @@ takeover verification are separate milestones (T11/T13) that build on this contract; the neutral interface carries only opaque payloads and state strings, leaving checkpoint serialization and run state machines to the caller. +## Opt-in single-primary advisory index repair (Refs #8) + +`RedisProgressIndexRepairer` is a small discovery/repair component, constructed +explicitly with a single-server Redisson client, the existing progress prefix, +an application state adapter, and a `RedisProgressAdvisoryIndex`. It does not +change the store API, keys, record format, lease scripts or default behavior. +It is independent of Core/Starter wiring and supplies no polling Worker. + +```java +var index = new RedisProgressAdvisoryIndex(redisson, "my-app:progress-hints:v1"); +var repairer = new RedisProgressIndexRepairer(redisson, + RedisLeasedProgressStore.DEFAULT_KEY_PREFIX, + metadata -> "APP_READY".equals(metadata.state()) + ? Optional.of(new RedisProgressAdvisoryIndex.Hint(metadata, 0)) + : Optional.empty(), + index); +var cursor = repairer.start(); +// Run one batch when the application's scheduler/capacity budget allows it. +var batch = repairer.repair(cursor, 25, 100); +cursor = batch.cursor(); // retain this entire continuation, including pending keys +var candidates = index.due(25); +``` + +`APP_READY` is an example application policy, not a built-in runnable state. +The adapter must explicitly interpret the application's opaque state and return +an absolute due time in milliseconds (zero means immediately due), or empty. +It must preserve the projected metadata and produce a deterministic decision/time +for a given revision. There is no generic event adapter: outbox payloads remain +opaque and are not read. Applications requiring event-only eligibility need a +separate bounded, versioned event projection; they must not infer runnable from +arbitrary state strings or the mere existence of an outbox. Due queries compare +scores with Redis `TIME`; retries, execution and heartbeat are not supplied. + +Discovery scans the authoritative progress keyspace, derives each session hash +key from the run key, and atomically projects the stored session/run IDs, schema +versions, revision and state. It verifies that the recovered IDs hash back to +the discovered key. It never relies on a prior global index and never reads +checkpoint/journal/outbox payloads or calls `snapshot`. Before `HGET`, `HSTRLEN` +checks enforce 1024 UTF-8 bytes per ID, 4096 bytes for state, one byte per schema +version and 19 bytes for revision. Oversized, wrong-type or corrupt metadata +stops the call with an error, leaving authority unchanged; applications need an +alert/reconciliation policy for such records. These are limits of this opt-in +reader, not new restrictions on the existing store. The projection also rejects +state fields at revision zero and missing/empty state at a committed revision, +matching the existing run-record format without interpreting application state. +It validates only the projected fields, not checkpoint/journal/outbox integrity. +No records are silently truncated or declared repaired. + +Each call processes at most `maxKeys` returned keys (1–1000), performs at most +one `SCAN`, and reads at most one small metadata projection per matching run. +`scanCount` (1–10000) is only Redis's work hint. **SCAN reply bytes, server CPU +and pending-page memory have no hard bound**: even COUNT=1 may return a large +page. The continuation retains that whole page with an offset; no same-page +keys are dropped and no new SCAN is issued until it is drained. Restart from +`start()` after a failed/unknown call; idempotent hints tolerate replay. Cursors +are in-memory and must not be moved to another database/keyspace. A completed +sweep covers retained keys in a quiescent keyspace; concurrent changes require +repeated sweeps. A hard discovery budget needs a future authoritative bounded +directory/protocol or server support and is not solved by this component. + +The reference sink is a separate durable ZSET, with a bounded encoded +`(sessionId, runId, revision)` member and absolute due-time score. `ZADD NX` makes +repeated/concurrent publication idempotent. Different revisions coexist, so a late +old publication cannot overwrite a newer revision. The two-argument constructor +preserves unbounded cardinality; the capacity overload and `poll`/`ack`/`retry` +protocol below provide bounded admission and visibility delivery. `due(limit)` +remains a diagnostic query and repeatedly returns unacknowledged candidates. +Only this schema may write the dedicated index and its auxiliary keys. + +A candidate is always advisory. The reference consumer below claims a session +lease and atomically revalidates authoritative metadata, reservation, owner/fence +and Redis TIME before invoking the application's executor. Neither discovery nor +the reference consumer is wired into Core/Starter or a production polling Worker. +No exactly-once or HA RPO0 guarantee is made. Only single-server Redisson client +configurations are accepted for discovery; the caller must supply a single +primary without automatic failover. The component cannot detect server topology +or failover policy. Cluster discovery and independent-JVM takeover are unverified. + +`RedisProgressIndexRepairerTest` uses real `valkey/valkey:8.1.2` with AOF and +`appendfsync always`. The failure-window fixture commits run/outbox, deliberately +never calls a publisher, shuts down the producing connection, and creates a new +repairer/connection. It is a skipped-publisher/connection-boundary simulation, +not an OS process kill, Redis crash/restart or two-JVM takeover test. The suite +also covers concurrent/repeated repair, oversized SCAN-page continuation, +explicit unknown/terminal-state policy, stale revisions, projection limits, +large historical payload isolation, server-time due queries, and Redis +wrong-type publication errors that preserve authoritative data. Testcontainers +skips without Docker must be reported as skips, not backend success. + +## Opt-in bounded membership directory and reference consumption (Refs #8) + +The SCAN repairer above remains available with its original soft bounds. For a +**new or explicitly migrated prefix**, all producers can instead opt into a +retained membership directory: + +```java +var store = RedisLeasedProgressStore.builder() + .redisson(redisson) + .keyPrefix("my-app:progress:v2") + .directoryCapacity(1000) + .build(); +var directory = new RedisProgressDirectory(redisson, "my-app:progress:v2", 1000); +var hints = new RedisProgressAdvisoryIndex(redisson, "my-app:hints:v2", 1000); +var projection = new RedisProgressIndexRepairer(redisson, "my-app:progress:v2", + m -> Optional.empty(), hints); +var batch = directory.repair("", 25, projection, + m -> "APP_READY".equals(m.state()) + ? Optional.of(new RedisProgressAdvisoryIndex.Hint(m, 0)) + : Optional.empty(), hints); +// Persist/report batch.issues(), then continue from batch.page().after(). +// Once page.done() is true, a later sweep starts again from "". +``` + +The directory is **authoritative for retained membership**, not for runnable +state, progress or lease ownership. It adds two dedicated keys, outside the +existing session hash tags: `:directory:v1:config` (capacity) and +`:directory:v1:members` (ZSET, all scores zero). `claim` validates directory +type/configuration/capacity before any authority write, runs the original claim +script, then registers membership in that **same Lua invocation** on success. +A process/connection loss after a successful claim cannot leave an accepted new +run unregistered. Commit updates the already registered run; there is no second +post-commit registration call. A full directory rejects a new identity before +creating session/run authority. Existing registered identities can still reclaim +leases. Busy/corrupt claims create no member. Redis errors/timeouts propagate; +timeout outcomes remain unknown and need re-reading. + +Capacity is immutable per prefix, 1..10000 retained session/run identities, +including terminal and revision-zero records. No directory GC/deletion is +provided: it never drops authority to admit more work. Exhaustion stops new +admission until the application explicitly provisions a new prefix or a future +retention protocol is selected. IDs are nonblank and at most 1024 UTF-8 bytes; +new commits through directory mode also require state <=4096 UTF-8 bytes. Default +store builders retain their old acceptance and behavior. Existing progress keys, +opaque state semantics, Redis TIME leases and exact int64 fences are unchanged. +The cross-session directory script requires a single primary and is deliberately +incompatible with Cluster/Sentinel/replicated client configurations. + +`page(after, limit)` uses exclusive lexicographic `ZRANGEBYLEX ... LIMIT 0 limit`, +with limit 1..1000. It returns at most that many members and a single bounded +cursor, with no overflow buffer. Immutable retained membership makes concurrent +sweeps safe; inserts behind an active cursor are visited on the next sweep. A +quiescent sweep covers all registered identities. There is no global snapshot +of concurrent submissions. + +Resource claims are specific to directory discovery and hint operations: + +- Directory members contain two Base64 URL IDs, at most 2733 ASCII bytes. A RESP2 + page response is at most `16 + 2742 * limit` bytes, excluding transport framing. + Retained member payload is at most `capacity * 2733` bytes. This is a payload + bound, not a bound on Redis allocator/index/client-object overhead. +- Directory lookup/insert is a fixed number of commands over at most capacity + members. Sorted-set range work is conventionally O(log capacity + limit); + compact encodings may scan up to capacity entries. It never scans the unrelated + database. A repair call performs one page query, at most limit projections and + at most limit publications. Application adapters must themselves be bounded. +- Each projection reads only the six length-checked fields described above; + its RESP2 metadata reply is below 6250 bytes. The leased variant additionally + length-checks owner/fence/expiry/reservation and validates them atomically with + metadata. Checkpoint, journal and outbox payloads are not projected. +- Hint members are at most 2753 ASCII bytes. A due reply is at most + `16 + 2762 * limit` RESP2 bytes. A poll reply interleaves members and exact int64 + delivery tokens, at most `16 + 2788 * limit` RESP2 bytes. At most capacity hint + members and their delivery-token fields are retained when the capacity overload + is used, plus fixed configuration and a retained sequence counter. + +These bounds require dedicated schema-owned keys, no eviction/TTL/deletion of +protocol keys and no bypass writers. They do **not** certify host CPU milliseconds, +RSS, Redis OOM/script failures or overall store claim/commit/snapshot cost. Existing +store scripts still read checkpoint/progress history; that cost is not made +bounded by enabling the directory. Lua isolates other clients but does not roll +back an unexpected runtime/OOM error after a write. Provision memory and preserve +the original single-primary durability assumptions; no performance or HA +certification is claimed. + +For migration, quiesce legacy producers, obtain and retain an operator-controlled +inventory of known `(sessionId, runId)` pairs, then call +`directory.registerExisting(sessionId, runId)` for each pair before enabling +polling. It validates the bounded stored identity and registers it atomically; +it is duplicate-safe and capacity checked. Audit the inventory against the source +of submissions, resolve any rejected identity, and require **every** future +producer to use directory mode. This cannot prove completeness for an unknown +legacy keyspace. Obtaining an inventory via SCAN is a separate, potentially +unbounded offline migration; no automatic full-keyspace bounded compatibility is +claimed. Oversized IDs require a migration decision rather than silent truncation. + +For registered records, bad/oversized/wrong-type metadata and missing authority +become bounded `RepairBatch.issues()` entries while the page cursor advances. +Persist or alert on these issues before persisting the new cursor; no record is +removed or declared repaired. Retry on later sweeps after operator correction. +A transport/publication exception aborts the call; retry the **input** cursor, +since partial successful publications are idempotent. A full hint sink likewise +requires consumption before retrying repair. Dedicated directory/index key +corruption by unsupported writers is a protocol failure, not silently skipped. + +`poll(limit, visibilityMillis)` atomically moves delivered hints to Redis +`TIME + visibilityMillis` and returns `(candidate, deliveryToken)`. Polling +workers do not each get the same first page. `ack(delivery)` removes exactly that +revision/delivery and its token; an old acknowledgement cannot delete a newer +revision or a redelivery. `retry(delivery, backoffMillis)` compares the same token, +postpones against Redis TIME and invalidates that token; repeated calls are no-ops. +Delays must be 1..86400000 ms. A lost response/crashed consumer redelivers after +visibility. Scores now encode visibility/retry, while initial publication remains +NX. Delivery sequence overflow fails before mutations. Terminal/stale hints are +pruned by ack; busy/error hints are deferred so existing due work can progress. +This is bounded-page progress for finite retained work, not a scheduler QoS +promise under unlimited competing publication. Repair may republish a pruned +old revision; authority validation makes it harmless. + +`RedisProgressReferenceConsumer` takes an explicit application runnable predicate +and an executor. Missing authority is not created from hints. For a present run +it claims the session, **then** atomically validates metadata/reservation/live +owner/fence/Redis TIME, rejects stale revisions and unknown/ineligible states, +and validates again immediately before dispatch. `Execution.guard()` renews the +lease and repeats that bounded metadata/lease validation; an executor should use +it before each effect. A successful executor returns a revision-matching +`ProgressCommit`; the store fences the commit, then the consumer acknowledges the +hint. The owner UUID must be unique per attempt. Exceptions schedule retry and +release the lease; an unknown commit result is reconciled by authority on +redelivery. Busy/lease-lost/metadata-error results defer the hint. Metadata errors +are returned for operator attention, not converted into runnable states. + +An external effect is outside the Lua transaction. A lease can expire after the +last guard, or an effect can succeed before a crash/timeout prevents its commit. +External fencing/idempotency/receipts and unsafe-recovery reconciliation remain +application responsibilities. The fake-executor test deliberately observes the +same effect twice across that boundary; this is not exactly-once execution or +timeout rollback. No automatic heartbeat thread, Core/Starter integration, +production Worker, event-only eligibility or independent-JVM takeover is supplied. + +`RedisProgressDirectoryTest` exercises real Valkey, concurrent capacity admission +and cursors, exact UTF-8/RESP page boundaries, explicit backfill, corrupt/large +records with unrelated progress, revision/delivery races, Redis-time visibility, +capacity exhaustion, competing consumers, expired/superseded leases and fake +execution failure windows. These tests run in one JVM. Graceful producer-client +shutdown and direct corruption/expiry fixtures do not imply OS kill, backend +restart, network-partition or independent-process takeover coverage. + ## Boundaries This module does not change legacy in-memory Redis-like Store classes in Core, diff --git a/graph-persistence/argi-graph-persistence-redis/src/main/java/io/github/agentic/ai/graph/persistence/redis/RedisLeasedProgressStore.java b/graph-persistence/argi-graph-persistence-redis/src/main/java/io/github/agentic/ai/graph/persistence/redis/RedisLeasedProgressStore.java index 4dfac6d3..a268154b 100644 --- a/graph-persistence/argi-graph-persistence-redis/src/main/java/io/github/agentic/ai/graph/persistence/redis/RedisLeasedProgressStore.java +++ b/graph-persistence/argi-graph-persistence-redis/src/main/java/io/github/agentic/ai/graph/persistence/redis/RedisLeasedProgressStore.java @@ -39,26 +39,26 @@ /** * Redis/Valkey implementation of {@link LeasedProgressStore}: one session lease - * (claim/renew/release) whose owner/fence guards run state, checkpoint, journal - * and outbox writes, so an independent run lock and checkpoint lock cannot drift - * apart. This is a new opt-in keyspace; {@link RedisRunStore}, - * {@link RedisLeasedCheckpointSaver} and the older savers are untouched. + * (claim/renew/release) whose owner/fence guards run state, checkpoint, journal and + * outbox writes, so an independent run lock and checkpoint lock cannot drift apart. This + * is a new opt-in keyspace; {@link RedisRunStore}, {@link RedisLeasedCheckpointSaver} and + * the older savers are untouched. *

- * Each session owns one hash (reservation, retained fence, live owner/expiry) - * plus three hashes per run (run record, journal, outbox), all sharing one cluster - * hash tag. Every mutation is a single primary-routed Lua script that re-checks - * reservation, owner, fence, Redis server {@code TIME} and run/checkpoint revision - * before its first write, so a rejected commit leaves no partial checkpoint or run - * progress. Scripts are sent without client-side retries, so a script runs at most - * once per call; after a transport error or timeout the outcome is unknown and the - * caller should re-read with {@link #snapshot(String, String)} before re-planning. + * Each session owns one hash (reservation, retained fence, live owner/expiry) plus three + * hashes per run (run record, journal, outbox), all sharing one cluster hash tag. Every + * mutation is a single primary-routed Lua script that re-checks reservation, owner, + * fence, Redis server {@code TIME} and run/checkpoint revision before its first write, so + * a rejected commit leaves no partial checkpoint or run progress. Scripts are sent + * without client-side retries, so a script runs at most once per call; after a transport + * error or timeout the outcome is unknown and the caller should re-read with + * {@link #snapshot(String, String)} before re-planning. *

* Lease expiry is measured only by the Redis server clock, so callers need no - * synchronized clocks. Keys never expire; fences, progress and checkpoints are - * retained. Asynchronous HA failover is not consensus fencing: a promoted replica - * that missed recent writes can re-issue a fencing token and lose acknowledged - * progress, and no RPO0 is promised. Same-slot layout and deployment support are - * verified at startup; a cross-slot or unsupported deployment fails fast. + * synchronized clocks. Keys never expire; fences, progress and checkpoints are retained. + * Asynchronous HA failover is not consensus fencing: a promoted replica that missed + * recent writes can re-issue a fencing token and lose acknowledged progress, and no RPO0 + * is promised. Same-slot layout and deployment support are verified at startup; a + * cross-slot or unsupported deployment fails fast. */ public final class RedisLeasedProgressStore implements LeasedProgressStore { @@ -70,10 +70,14 @@ public final class RedisLeasedProgressStore implements LeasedProgressStore { private final RedissonClient redisson; + private final String keyPrefix; + private final RedisProgressKeys keys; private final RScript script; + private final RedisProgressDirectory directory; + private RedisLeasedProgressStore(Builder builder) { this.redisson = requireNonNull(builder.redisson, "redisson cannot be null"); requireNonNull(builder.keyPrefix, "keyPrefix cannot be null"); @@ -81,18 +85,31 @@ private RedisLeasedProgressStore(Builder builder) { throw new IllegalArgumentException("keyPrefix must not be blank"); } if (builder.keyPrefix.indexOf('{') >= 0 || builder.keyPrefix.indexOf('}') >= 0) { - // The store owns the cluster hash tag; a brace in the prefix would change the slot. + // The store owns the cluster hash tag; a brace in the prefix would change the + // slot. throw new IllegalArgumentException("keyPrefix must not contain '{' or '}'"); } + this.keyPrefix = builder.keyPrefix; this.keys = new RedisProgressKeys(builder.keyPrefix); - // Redisson resends a command whose reply is late; a resent script would run twice and + // Redisson resends a command whose reply is late; a resent script would run twice + // and // could double-claim or double-commit, so retries are disabled. this.script = redisson.getScript(OptionalOptions.defaults().codec(StringCodec.INSTANCE).retryAttempts(0)); + this.directory = builder.directoryCapacity == 0 ? null + : new RedisProgressDirectory(redisson, builder.keyPrefix, builder.directoryCapacity); if (builder.verifySlotsAtStartup) { verifyDeployment(); } } + String keyPrefix() { + return keyPrefix; + } + + RedissonClient client() { + return redisson; + } + public static Builder builder() { return new Builder(); } @@ -104,8 +121,9 @@ public SessionLease claim(String sessionId, String runId, UUID ownerId, Duration requireNonNull(ownerId, "ownerId cannot be null"); long ttlMillis = ttlMillis(ttl); List sessionKeys = keys.keys(sessionId, runId); - List result = eval(RedisProgressScripts.CLAIM, sessionKeys.subList(0, 2), sessionId, runId, - ownerId.toString(), Long.toString(ttlMillis)); + List result = directory == null ? eval(RedisProgressScripts.CLAIM, sessionKeys.subList(0, 2), sessionId, + runId, ownerId.toString(), Long.toString(ttlMillis)) + : directory.claim(sessionId, runId, ownerId.toString(), ttlMillis); switch (status(result)) { case RedisProgressScripts.OK: return new SessionLease(sessionId, runId, ownerId, parseLong(result, 1), parseLong(result, 2)); @@ -171,6 +189,9 @@ public Optional snapshot(String sessionId, String runId) { public ProgressSnapshot commit(SessionLease lease, ProgressCommit commit) throws LeaseLostException { requireNonNull(lease, "lease cannot be null"); requireNonNull(commit, "commit cannot be null"); + if (directory != null && commit.state().getBytes(StandardCharsets.UTF_8).length > 4096) { + throw new IllegalArgumentException("Directory mode state must be <=4096 UTF-8 bytes"); + } requireNoDuplicateIds(commit.journal(), "journal"); requireNoDuplicateIds(commit.outbox(), "outbox"); long nextRevision = Math.addExact(commit.expectedRunRevision(), 1); @@ -202,7 +223,8 @@ public ProgressSnapshot commit(SessionLease lease, ProgressCommit commit) throws args.toArray()); switch (status(result)) { case RedisProgressScripts.OK: - // Only the live owner can write, so the follow-up read observes exactly this commit. + // Only the live owner can write, so the follow-up read observes exactly + // this commit. return snapshot(lease.sessionId(), lease.runId()) .orElseThrow(() -> new IllegalStateException("Committed run progress is missing")); case RedisProgressScripts.C_NOT_RESERVED: @@ -213,8 +235,8 @@ public ProgressSnapshot commit(SessionLease lease, ProgressCommit commit) throws throw new ProgressConflictException("Run revision does not match: expected " + commit.expectedRunRevision() + " but was " + string(result, 1)); case RedisProgressScripts.C_CP_REVISION: - throw new ProgressConflictException("Checkpoint revision does not match: stored revision is " - + string(result, 1)); + throw new ProgressConflictException( + "Checkpoint revision does not match: stored revision is " + string(result, 1)); case RedisProgressScripts.C_ENTRY: throw new ProgressConflictException("Append-only entry id already exists: " + string(result, 1)); default: @@ -223,9 +245,9 @@ public ProgressSnapshot commit(SessionLease lease, ProgressCommit commit) throws } /** - * Verifies the session key layout is single-slot and that the deployment - * accepts a multi-key script over it. A cross-slot layout, a cluster that - * splits the keys, or a proxy that cannot route the script fails at startup. + * Verifies the session key layout is single-slot and that the deployment accepts a + * multi-key script over it. A cross-slot layout, a cluster that splits the keys, or a + * proxy that cannot route the script fails at startup. */ private void verifyDeployment() { List probeKeys = keys.keys("argi:slot-probe:session", "argi:slot-probe:run"); @@ -257,7 +279,10 @@ private static int slot(String key) { return CRC16.crc16(hashTag(key).getBytes(StandardCharsets.UTF_8)) % 16384; } - /** Redis cluster hash-tag rule: the first non-empty {...} section, otherwise the whole key. */ + /** + * Redis cluster hash-tag rule: the first non-empty {...} section, otherwise the whole + * key. + */ private static String hashTag(String key) { int start = key.indexOf('{'); if (start >= 0) { @@ -301,8 +326,7 @@ private List> readSections(List result, int index) { } private List eval(String source, List scriptKeys, Object... args) { - List result = script.eval(RScript.Mode.READ_WRITE, source, RScript.ReturnType.MULTI, scriptKeys, - args); + List result = script.eval(RScript.Mode.READ_WRITE, source, RScript.ReturnType.MULTI, scriptKeys, args); if (result == null || result.isEmpty()) { throw new IllegalStateException("Redis progress store script returned no result"); } @@ -367,6 +391,17 @@ public static final class Builder { private boolean verifySlotsAtStartup = true; + private int directoryCapacity; + + /** Opt into the single-primary retained directory. All writers must opt in. */ + public Builder directoryCapacity(int capacity) { + if (capacity < 1 || capacity > 10000) { + throw new IllegalArgumentException("Directory capacity must be 1..10000"); + } + this.directoryCapacity = capacity; + return this; + } + private Builder() { } diff --git a/graph-persistence/argi-graph-persistence-redis/src/main/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressAdvisoryIndex.java b/graph-persistence/argi-graph-persistence-redis/src/main/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressAdvisoryIndex.java new file mode 100644 index 00000000..a876a81b --- /dev/null +++ b/graph-persistence/argi-graph-persistence-redis/src/main/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressAdvisoryIndex.java @@ -0,0 +1,235 @@ +/* + * 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.graph.persistence.redis; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Base64; +import java.util.List; +import java.util.Objects; + +import org.redisson.api.RScript; +import org.redisson.api.RedissonClient; +import org.redisson.api.options.OptionalOptions; +import org.redisson.client.codec.StringCodec; + +/** + * Revision-specific advisory hints, with optional capacity and visibility delivery. + * Authority lives in progress hashes, never here. Dedicated schema-owned keys only; + * poll/ack/retry are single-primary and make no external execution guarantee. + */ +public final class RedisProgressAdvisoryIndex { + + private static final long MAX_EXACT_SCORE = 9_007_199_254_740_991L; + + private static final String TIME = """ + local t = redis.call('TIME') + local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000) + """; + + private static final String POLL = TIME + """ + local members = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', now, 'LIMIT', 0, ARGV[1]) + local seq = redis.call('HGET', KEYS[2], '_sequence') or '0' + if not string.match(seq, '^%d+$') or #seq > 19 + or (#seq == 19 and seq > '9223372036854774807') then + return redis.error_reply('Delivery sequence invalid or exhausted') + end + local result = {} + for _, member in ipairs(members) do + redis.call('HINCRBY', KEYS[2], '_sequence', 1) + local token = redis.call('HGET', KEYS[2], '_sequence') + redis.call('HSET', KEYS[2], member, token) + redis.call('ZADD', KEYS[1], now + tonumber(ARGV[2]), member) + table.insert(result, member) + table.insert(result, token) + end + return result + """; + + private final RScript script; + + private final String key; + + private final String deliveries; + + private final String config; + + private final int capacity; + + /** Unbounded legacy reference sink; choose the capacity overload for bounded use. */ + public RedisProgressAdvisoryIndex(RedissonClient client, String key) { + this(client, key, 0); + } + + /** Capacity 1..10000; zero preserves the original unbounded publication contract. */ + public RedisProgressAdvisoryIndex(RedissonClient client, String key, int capacity) { + Objects.requireNonNull(client, "client"); + if (!client.getConfig().isSingleConfig()) { + throw new IllegalArgumentException("Reference index requires a single-server client"); + } + if (key == null || key.isBlank() || key.contains("{") || key.contains("}") || key.length() > 256 || capacity < 0 + || capacity > 10000) { + throw new IllegalArgumentException("Invalid dedicated hint key or capacity"); + } + this.key = key; + this.deliveries = key + ":deliveries:v1"; + this.config = key + ":config:v1"; + this.capacity = capacity; + this.script = client.getScript(OptionalOptions.defaults().codec(StringCodec.INSTANCE).retryAttempts(0)); + Number initialized = script.eval(RScript.Mode.READ_WRITE, """ + local previous = redis.call('HGET', KEYS[1], 'capacity') + if previous and previous ~= ARGV[1] then return 0 end + if tonumber(ARGV[1]) > 0 and redis.call('ZCARD', KEYS[2]) > tonumber(ARGV[1]) then return 0 end + redis.call('HSET', KEYS[1], 'capacity', ARGV[1]) + return 1 + """, RScript.ReturnType.INTEGER, List.of(config, key), Integer.toString(capacity)); + if (initialized.longValue() != 1) { + throw new IllegalStateException("Hint capacity differs from existing configuration/data"); + } + } + + /** + * Idempotent NX publication. A full index throws; authority can be repaired later. + */ + public boolean publish(Hint hint) { + Objects.requireNonNull(hint, "hint"); + RedisProgressIndexRepairer.Metadata m = Objects.requireNonNull(hint.metadata(), "metadata"); + if (hint.dueAtMillis() < 0 || hint.dueAtMillis() > MAX_EXACT_SCORE) { + throw new IllegalArgumentException("Invalid exact millisecond score"); + } + String member = member(new Candidate(m.sessionId(), m.runId(), m.revision())); + Number added = script.eval(RScript.Mode.READ_WRITE, """ + if redis.call('HGET', KEYS[2], 'capacity') ~= ARGV[3] then + return redis.error_reply('Hint configuration changed') + end + if redis.call('ZSCORE', KEYS[1], ARGV[2]) then return 0 end + if tonumber(ARGV[3]) > 0 and redis.call('ZCARD', KEYS[1]) >= tonumber(ARGV[3]) then + return redis.error_reply('Hint capacity exhausted; retry repair after consumption') + end + return redis.call('ZADD', KEYS[1], 'NX', ARGV[1], ARGV[2]) + """, RScript.ReturnType.INTEGER, List.of(key, config), Long.toString(hint.dueAtMillis()), member, + Integer.toString(capacity)); + return added.longValue() == 1; + } + + /** Diagnostic query; use poll for fair visibility delivery. */ + public List due(int limit) { + requireLimit(limit); + List members = script.eval(RScript.Mode.READ_WRITE, TIME + """ + return redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', now, 'LIMIT', 0, ARGV[1]) + """, RScript.ReturnType.MULTI, List.of(key), Integer.toString(limit)); + return members.stream().map(RedisProgressAdvisoryIndex::candidate).toList(); + } + + /** + * Atomically postpones each delivered member using Redis TIME and issues a unique + * exact int64 token. Busy/crashed consumers do not permanently pin the first page. A + * lost response redelivers after visibility; it does not delete authority or hints. + */ + public List poll(int limit, long visibilityMillis) { + requireLimit(limit); + requireDelay(visibilityMillis); + List raw = script.eval(RScript.Mode.READ_WRITE, POLL, RScript.ReturnType.MULTI, + List.of(key, deliveries), Integer.toString(limit), Long.toString(visibilityMillis)); + List result = new ArrayList<>(); + for (int i = 0; i < raw.size(); i += 2) { + result.add(new Delivery(candidate(raw.get(i)), Long.parseLong(raw.get(i + 1)))); + } + return List.copyOf(result); + } + + /** + * Deletes only this revision and this delivery; newer delivery/revision is untouched. + */ + public boolean ack(Delivery delivery) { + return finish(delivery, 0); + } + + /** Idempotent exact-delivery retry, scheduled against Redis TIME. */ + public boolean retry(Delivery delivery, long backoffMillis) { + requireDelay(backoffMillis); + return finish(delivery, backoffMillis); + } + + private boolean finish(Delivery delivery, long delay) { + Objects.requireNonNull(delivery, "delivery"); + if (delivery.token() <= 0) { + throw new IllegalArgumentException("Delivery token must be positive"); + } + Number changed = script.eval(RScript.Mode.READ_WRITE, TIME + """ + if redis.call('HGET', KEYS[2], ARGV[1]) ~= ARGV[2] then return 0 end + if tonumber(ARGV[3]) == 0 then + redis.call('ZREM', KEYS[1], ARGV[1]) + elseif redis.call('ZSCORE', KEYS[1], ARGV[1]) then + redis.call('ZADD', KEYS[1], now + tonumber(ARGV[3]), ARGV[1]) + end + redis.call('HDEL', KEYS[2], ARGV[1]) + return 1 + """, RScript.ReturnType.INTEGER, List.of(key, deliveries), member(delivery.candidate()), + Long.toString(delivery.token()), Long.toString(delay)); + return changed.longValue() == 1; + } + + private static void requireLimit(int limit) { + if (limit < 1 || limit > 1000) { + throw new IllegalArgumentException("limit must be between 1 and 1000"); + } + } + + private static void requireDelay(long delay) { + if (delay < 1 || delay > 86_400_000) { + throw new IllegalArgumentException("Delay must be 1..86400000 milliseconds"); + } + } + + private static String member(Candidate candidate) { + if (candidate.revision() < 0) { + throw new IllegalArgumentException("Revision must be nonnegative"); + } + return encode(candidate.sessionId()) + "." + encode(candidate.runId()) + "." + candidate.revision(); + } + + private static Candidate candidate(String member) { + String[] parts = member.split("\\.", -1); + if (parts.length != 3) { + throw new IllegalStateException("Invalid advisory member"); + } + return new Candidate(decode(parts[0]), decode(parts[1]), Long.parseLong(parts[2])); + } + + private static String encode(String value) { + Objects.requireNonNull(value, "identity"); + byte[] bytes = value.getBytes(StandardCharsets.UTF_8); + if (value.isBlank() || bytes.length > 1024) { + throw new IllegalArgumentException("Identity must be nonblank and at most 1024 UTF-8 bytes"); + } + return Base64.getUrlEncoder().withoutPadding().encodeToString(bytes); + } + + private static String decode(String value) { + return new String(Base64.getUrlDecoder().decode(value), StandardCharsets.UTF_8); + } + + public record Hint(RedisProgressIndexRepairer.Metadata metadata, long dueAtMillis) { + } + + public record Candidate(String sessionId, String runId, long revision) { + } + + public record Delivery(Candidate candidate, long token) { + } + +} diff --git a/graph-persistence/argi-graph-persistence-redis/src/main/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressDirectory.java b/graph-persistence/argi-graph-persistence-redis/src/main/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressDirectory.java new file mode 100644 index 00000000..513a5e03 --- /dev/null +++ b/graph-persistence/argi-graph-persistence-redis/src/main/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressDirectory.java @@ -0,0 +1,232 @@ +/* + * 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.graph.persistence.redis; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Base64; +import java.util.List; +import java.util.Objects; +import java.util.Optional; +import java.util.function.Function; + +import org.redisson.api.RScript; +import org.redisson.api.RedissonClient; +import org.redisson.api.options.OptionalOptions; +import org.redisson.client.codec.StringCodec; + +/** + * Opt-in retained authoritative membership directory. All producers must use the store's + * directoryCapacity option for this prefix. Registration and authority creation share one + * Lua invocation. No SCAN, deletion, automatic migration or Cluster support is supplied. + * Membership is authoritative; runnable state and lease ownership remain in progress + * hashes. + */ +public final class RedisProgressDirectory { + + private static final String GUARD = """ + local dt = redis.call('TYPE', KEYS[4]).ok + if dt ~= 'none' and dt ~= 'zset' then return {'DIRECTORY_TYPE'} end + if redis.call('TYPE', KEYS[3]).ok ~= 'hash' + or redis.call('HGET', KEYS[3], 'capacity') ~= ARGV[6] then + return {'DIRECTORY_CONFIG'} + end + local present = redis.call('ZSCORE', KEYS[4], ARGV[5]) + if not present and redis.call('ZCARD', KEYS[4]) >= tonumber(ARGV[6]) then + return {'DIRECTORY_FULL'} + end + """; + + private static final String CLAIM = GUARD + "local function claim_authority()\n" + + "local KEYS = {KEYS[1], KEYS[2]}\n" + RedisProgressScripts.CLAIM + "\nend\n" + """ + local result = claim_authority() + if result[1] == 'OK' then redis.call('ZADD', KEYS[4], 0, ARGV[5]) end + return result + """; + + private static final String BACKFILL = GUARD + """ + for i = 1, 2 do + if redis.call('TYPE', KEYS[i]).ok ~= 'hash' then return {'MISSING_OR_TYPE'} end + end + if redis.call('HSTRLEN', KEYS[1], 'session') > 1024 + or redis.call('HSTRLEN', KEYS[2], 'run') > 1024 then return {'IDENTITY_LIMIT'} end + if redis.call('HGET', KEYS[1], 'session') ~= ARGV[1] + or redis.call('HGET', KEYS[2], 'run') ~= ARGV[2] then return {'IDENTITY'} end + redis.call('ZADD', KEYS[4], 0, ARGV[5]) + return {'OK'} + """; + + private final RScript script; + + private final RedissonClient client; + + private final RedisProgressKeys keys; + + private final String prefix; + + private final String configKey; + + private final String directoryKey; + + private final int capacity; + + public RedisProgressDirectory(RedissonClient client, String prefix, int capacity) { + Objects.requireNonNull(client, "client"); + if (!client.getConfig().isSingleConfig() || prefix == null || prefix.isBlank() + || prefix.getBytes(StandardCharsets.UTF_8).length > 256 || prefix.contains("{") || prefix.contains("}") + || capacity < 1 || capacity > 10000) { + throw new IllegalArgumentException( + "Single-server directory requires prefix <=256 UTF-8 bytes and capacity 1..10000"); + } + this.client = client; + this.prefix = prefix; + this.capacity = capacity; + this.keys = new RedisProgressKeys(prefix); + this.configKey = prefix + ":directory:v1:config"; + this.directoryKey = prefix + ":directory:v1:members"; + this.script = client.getScript(OptionalOptions.defaults().codec(StringCodec.INSTANCE).retryAttempts(0)); + List result = script.eval(RScript.Mode.READ_WRITE, """ + local t = redis.call('TYPE', KEYS[1]).ok + if t ~= 'none' and t ~= 'hash' then return {'DIRECTORY_CONFIG'} end + local old = redis.call('HGET', KEYS[1], 'capacity') + if old and old ~= ARGV[1] then return {'DIRECTORY_CONFIG'} end + redis.call('HSET', KEYS[1], 'capacity', ARGV[1]) + return {'OK'} + """, RScript.ReturnType.MULTI, List.of(configKey), Integer.toString(capacity)); + requireOk(result); + } + + List claim(String session, String run, String owner, long ttl) { + List result = script.eval(RScript.Mode.READ_WRITE, CLAIM, RScript.ReturnType.MULTI, + scriptKeys(session, run), session, run, owner, Long.toString(ttl), member(session, run), + Integer.toString(capacity)); + if (String.valueOf(result.get(0)).startsWith("DIRECTORY_")) { + throw new IllegalStateException(String.valueOf(result.get(0))); + } + return result; + } + + /** + * Explicit one-record backfill from a caller-supplied inventory, with legacy writers + * quiesced. This registers identity even if other metadata needs operator repair; + * repair() will report that error while continuing past it. No full-keyspace claim. + */ + public void registerExisting(String session, String run) { + List result = script.eval(RScript.Mode.READ_WRITE, BACKFILL, RScript.ReturnType.MULTI, + scriptKeys(session, run), session, run, "", "", member(session, run), Integer.toString(capacity)); + requireOk(result); + } + + private List scriptKeys(String session, String run) { + return List.of(keys.sessionKey(session), keys.keys(session, run).get(1), configKey, directoryKey); + } + + /** + * Exclusive lexicographic continuation. Inserts behind the cursor appear next sweep. + */ + public Page page(String after, int limit) { + if (after == null || after.length() > 2733 + || (!after.isEmpty() && !after.matches("[A-Za-z0-9_-]+\\.[A-Za-z0-9_-]+")) || limit < 1 + || limit > 1000) { + throw new IllegalArgumentException("Invalid directory cursor or limit"); + } + List members = script.eval(RScript.Mode.READ_WRITE, + "return redis.call('ZRANGEBYLEX', KEYS[1], ARGV[1], '+', 'LIMIT', 0, ARGV[2])", + RScript.ReturnType.MULTI, List.of(directoryKey), after.isEmpty() ? "-" : "(" + after, + Integer.toString(limit)); + List ids = members.stream().map(RedisProgressDirectory::identity).toList(); + return new Page(ids, members.isEmpty() ? after : members.get(members.size() - 1), members.size() < limit); + } + + /** + * Bounded directory page plus one bounded metadata projection per entry. Bad metadata + * is returned as a queryable issue and the cursor advances; transport/publication + * failures propagate so retrying the input cursor republishes idempotently. + */ + public RepairBatch repair(String after, int limit, RedisProgressIndexRepairer projection, + Function> adapter, + RedisProgressAdvisoryIndex sink) { + if (!prefix.equals(projection.prefix()) || client != projection.client()) { + throw new IllegalArgumentException("Projection must share directory client and prefix"); + } + Page page = page(after, limit); + List issues = new ArrayList<>(); + int published = 0; + for (Identity id : page.entries()) { + Optional metadata; + try { + metadata = projection.metadata(id.sessionId(), id.runId()); + } + catch (RedisProgressIndexRepairer.MetadataException ex) { + issues.add(new Issue(id, ex.getMessage())); + continue; + } + if (metadata.isEmpty()) { + issues.add(new Issue(id, "MISSING_AUTHORITY")); + continue; + } + Optional hint = Objects.requireNonNull(adapter.apply(metadata.get())); + if (hint.isPresent()) { + if (!metadata.get().equals(hint.get().metadata())) { + throw new IllegalArgumentException("Adapter must preserve authority metadata"); + } + if (sink.publish(hint.get())) { + published++; + } + } + } + return new RepairBatch(page, published, List.copyOf(issues)); + } + + static String member(String session, String run) { + return encode(session) + "." + encode(run); + } + + private static String encode(String id) { + if (id == null || id.isBlank() || id.getBytes(StandardCharsets.UTF_8).length > 1024) { + throw new IllegalArgumentException("Directory identity must be nonblank and <=1024 UTF-8 bytes"); + } + return Base64.getUrlEncoder().withoutPadding().encodeToString(id.getBytes(StandardCharsets.UTF_8)); + } + + private static Identity identity(String member) { + String[] parts = member.split("\\.", -1); + if (parts.length != 2 || member.length() > 2733) { + throw new IllegalStateException("Invalid directory member; dedicated key has an unsupported writer"); + } + return new Identity(new String(Base64.getUrlDecoder().decode(parts[0]), StandardCharsets.UTF_8), + new String(Base64.getUrlDecoder().decode(parts[1]), StandardCharsets.UTF_8)); + } + + private static void requireOk(List result) { + if (!result.equals(List.of("OK"))) { + throw new IllegalStateException("Directory operation rejected: " + result.get(0)); + } + } + + public record Identity(String sessionId, String runId) { + } + + public record Page(List entries, String after, boolean done) { + } + + public record Issue(Identity identity, String error) { + } + + public record RepairBatch(Page page, int published, List issues) { + } + +} diff --git a/graph-persistence/argi-graph-persistence-redis/src/main/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressIndexRepairer.java b/graph-persistence/argi-graph-persistence-redis/src/main/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressIndexRepairer.java new file mode 100644 index 00000000..e642784f --- /dev/null +++ b/graph-persistence/argi-graph-persistence-redis/src/main/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressIndexRepairer.java @@ -0,0 +1,307 @@ +/* + * 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.graph.persistence.redis; + +import io.github.agentic.ai.graph.checkpoint.lease.LeaseLostException; + +import java.util.List; +import java.util.Objects; +import java.util.Optional; +import java.util.function.Function; +import java.util.regex.Pattern; + +import org.redisson.api.RScript; +import org.redisson.api.RedissonClient; +import org.redisson.api.options.OptionalOptions; +import org.redisson.client.codec.StringCodec; + +/** + * Explicitly constructed, single-primary discovery of authoritative progress metadata. + * Publishes advisory candidates only: it never claims leases or changes progress. A + * consumer must claim and then re-check authoritative state/revision before execution. No + * scheduler, terminal-state policy, outbox payload interpretation or exactly-once + * execution is supplied here. + */ +public final class RedisProgressIndexRepairer { + + private static final String SCAN = "return redis.call('SCAN', ARGV[1], 'MATCH', ARGV[2], 'COUNT', ARGV[3])"; + + // HSTRLEN checks run before HGET, so oversized opaque fields never cross the wire. + // No snapshot, checkpoint, journal entries or outbox payloads are read. + private static final String PROJECT = """ + for _, key in ipairs(KEYS) do + local t = redis.call('TYPE', key).ok + if t == 'none' then return {} end + if t ~= 'hash' then return {'ERROR', 'Invalid progress metadata type'} end + end + local fields = {{1, 'v', 1}, {1, 'session', 1024}, {2, 'v', 1}, + {2, 'run', 1024}, {2, 'revision', 19}, {2, 'state', 4096}} + for _, f in ipairs(fields) do + if redis.call('HSTRLEN', KEYS[f[1]], f[2]) > f[3] then + return {'ERROR', 'Progress metadata exceeds projection limit'} + end + end + local result = {} + for _, f in ipairs(fields) do + table.insert(result, redis.call('HGET', KEYS[f[1]], f[2]) or '') + end + local has_state = redis.call('HEXISTS', KEYS[2], 'state') == 1 + if (result[5] == '0' and has_state) or (result[5] ~= '0' and result[6] == '') then + return {'ERROR', 'Invalid progress metadata revision/state'} + end + return result + """; + + private static final String LEASE_PROJECT = "local function project()\n" + PROJECT + "\nend\n" + """ + local result = project() + if #result == 0 or result[1] == 'ERROR' then return result end + local fields = {{'owner', 36}, {'fence', 19}, {'expires', 19}, {'reservation', 1024}} + for _, f in ipairs(fields) do + if redis.call('HSTRLEN', KEYS[1], f[1]) > f[2] then + return {'ERROR', 'Progress lease exceeds projection limit'} + end + end + local lease = redis.call('HMGET', KEYS[1], 'owner', 'fence', 'expires', 'reservation') + if lease[1] ~= ARGV[1] or lease[2] ~= ARGV[2] or lease[4] ~= ARGV[3] then return {'LOST'} end + local expires = lease[3] + if not expires or (expires ~= '0' and not string.match(expires, '^[1-9][0-9]*$')) + or #expires > 19 or (#expires == 19 and expires > '9223372036854775807') then + return {'ERROR', 'Invalid progress lease expiry'} + end + local t = redis.call('TIME') + local now = tostring(tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000)) + if #now > #expires or (#now == #expires and now >= expires) then return {'LOST'} end + return result + """; + + private final RScript script; + + private final RedissonClient client; + + private final String prefix; + + private final Pattern runPattern; + + private final RedisProgressKeys keys; + + private final Function> adapter; + + private final RedisProgressAdvisoryIndex sink; + + /** + * The adapter owns the opaque state policy and must be deterministic for one + * revision. Return empty for states that should not produce a hint. Neither absence + * of a known terminal string nor presence of outbox data implies runnable. Prefix + * must be the same as the authoritative store; the sink is a separate key. + */ + public RedisProgressIndexRepairer(RedissonClient client, String prefix, + Function> adapter, RedisProgressAdvisoryIndex sink) { + Objects.requireNonNull(client, "client"); + if (!client.getConfig().isSingleConfig()) { + throw new IllegalArgumentException("Repair discovery requires an explicit single-server client"); + } + if (prefix == null || prefix.isBlank() || prefix.contains("{") || prefix.contains("}") + || prefix.length() > 256) { + throw new IllegalArgumentException( + "Progress prefix must be nonblank, at most 256 chars and contain no braces"); + } + this.client = client; + this.prefix = prefix; + this.runPattern = Pattern.compile(Pattern.quote(prefix) + ":\\{[A-Za-z0-9_-]{43}\\}:run:[A-Za-z0-9_-]{43}"); + this.keys = new RedisProgressKeys(prefix); + this.adapter = Objects.requireNonNull(adapter, "adapter"); + this.sink = Objects.requireNonNull(sink, "sink"); + this.script = client.getScript(OptionalOptions.defaults().codec(StringCodec.INSTANCE).retryAttempts(0)); + } + + /** Start a new sweep; restart from here if a call fails or its outcome is unknown. */ + public Cursor start() { + return new Cursor(prefix, "0", List.of(), 0, false); + } + + /** + * Inspects at most maxKeys returned keys and issues at most one SCAN per call. + * scanCount is a Redis work hint, NOT a hard response-size/CPU limit. A whole SCAN + * reply is retained in the continuation; even COUNT=1 can exceed one key. Thus key + * processing and metadata replies are bounded, but SCAN network bytes and cursor + * memory are soft bounds. Quiescent full sweeps discover retained records; concurrent + * mutations require repeated sweeps (normal SCAN semantics). + */ + public Batch repair(Cursor cursor, int maxKeys, int scanCount) { + Objects.requireNonNull(cursor, "cursor"); + if (!prefix.equals(cursor.prefix) || maxKeys < 1 || maxKeys > 1000 || scanCount < 1 || scanCount > 10000) { + throw new IllegalArgumentException("Invalid cursor or batch limits"); + } + if (cursor.done()) { + return new Batch(cursor, 0, 0); + } + List pending = cursor.pending; + int offset = cursor.offset; + String next = cursor.redisCursor; + if (cursor.pendingKeys() == 0) { + List reply = script.eval(RScript.Mode.READ_WRITE, SCAN, RScript.ReturnType.MULTI, List.of(), next, + globLiteral(prefix) + ":{*}:run:*", Integer.toString(scanCount)); + next = (String) reply.get(0); + pending = ((List) reply.get(1)).stream().map(String.class::cast).toList(); + offset = 0; + } + int inspected = Math.min(maxKeys, pending.size() - offset); + int published = 0; + for (String key : pending.subList(offset, offset + inspected)) { + if (runPattern.matcher(key).matches()) { + Optional metadata = project(key); + if (metadata.isPresent()) { + Optional hint = Objects + .requireNonNull(adapter.apply(metadata.get())); + if (hint.isPresent()) { + if (!hint.get().metadata().equals(metadata.get())) { + throw new IllegalArgumentException("Adapter must preserve projected authority metadata"); + } + if (sink.publish(hint.get())) { + published++; + } + } + } + } + } + return new Batch(new Cursor(prefix, next, pending, offset + inspected, true), inspected, published); + } + + /** + * Read bounded current metadata for a known ID, for candidate validation. No lease is + * acquired. + */ + public Optional metadata(String sessionId, String runId) { + Objects.requireNonNull(sessionId, "sessionId"); + Objects.requireNonNull(runId, "runId"); + return project((String) keys.keys(sessionId, runId).get(1)); + } + + /** + * Atomically validates current reservation, owner/fence and Redis TIME with metadata. + */ + public Optional metadata(LeasedProgressStore.SessionLease lease) { + Objects.requireNonNull(lease, "lease"); + String runKey = (String) keys.keys(lease.sessionId(), lease.runId()).get(1); + List result = script.eval(RScript.Mode.READ_WRITE, LEASE_PROJECT, RScript.ReturnType.MULTI, + List.of(keys.sessionKey(lease.sessionId()), runKey), lease.ownerId().toString(), + Long.toString(lease.fencingToken()), lease.runId()); + if (!result.isEmpty() && "LOST".equals(result.get(0))) { + throw new LeaseLostException(lease.sessionId(), lease.ownerId(), lease.fencingToken(), + "lease lost during metadata validation"); + } + return projected(runKey, result); + } + + RedissonClient client() { + return client; + } + + String prefix() { + return prefix; + } + + /** A bounded, known metadata fault; transport failures remain distinct. */ + public static final class MetadataException extends IllegalStateException { + + private static final long serialVersionUID = 1L; + + MetadataException(String message) { + super(message); + } + + } + + private Optional project(String runKey) { + String nsKey = runKey.substring(0, runKey.indexOf(":run:", prefix.length())) + ":ns"; + List result = script.eval(RScript.Mode.READ_WRITE, PROJECT, RScript.ReturnType.MULTI, + List.of(nsKey, runKey)); + return projected(runKey, result); + } + + private Optional projected(String runKey, List result) { + if (result.isEmpty()) { + return Optional.empty(); + } + if ("ERROR".equals(result.get(0))) { + throw new MetadataException(result.get(1)); + } + if (!"1".equals(result.get(0)) || !"1".equals(result.get(2)) || result.get(1).isBlank() + || result.get(3).isBlank() || !result.get(4).matches("0|[1-9][0-9]{0,18}")) { + throw new MetadataException("Invalid progress metadata"); + } + long revision; + try { + revision = Long.parseLong(result.get(4)); + } + catch (NumberFormatException ex) { + throw new MetadataException("Invalid progress metadata revision"); + } + Metadata metadata = new Metadata(result.get(1), result.get(3), revision, result.get(5)); + if (!keys.keys(metadata.sessionId(), metadata.runId()).get(1).equals(runKey)) { + throw new MetadataException("Progress identity does not match authoritative key"); + } + return Optional.of(metadata); + } + + private static String globLiteral(String value) { + return value.replace("\\", "\\\\").replace("*", "\\*").replace("?", "\\?").replace("[", "\\["); + } + + /** + * State is an opaque string; empty also represents the uncommitted revision-zero + * state. + */ + public record Metadata(String sessionId, String runId, long revision, String state) { + } + + /** + * In-memory continuation; do not reuse with a different Redis database or keyspace. + */ + public static final class Cursor { + + private final String prefix; + + private final String redisCursor; + + private final List pending; + + private final int offset; + + private final boolean started; + + private Cursor(String prefix, String redisCursor, List pending, int offset, boolean started) { + this.prefix = prefix; + this.redisCursor = redisCursor; + this.pending = List.copyOf(pending); + this.offset = offset; + this.started = started; + } + + public boolean done() { + return started && "0".equals(redisCursor) && pendingKeys() == 0; + } + + public int pendingKeys() { + return pending.size() - offset; + } + + } + + public record Batch(Cursor cursor, int inspectedKeys, int newHints) { + } + +} diff --git a/graph-persistence/argi-graph-persistence-redis/src/main/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressReferenceConsumer.java b/graph-persistence/argi-graph-persistence-redis/src/main/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressReferenceConsumer.java new file mode 100644 index 00000000..6cebd1c0 --- /dev/null +++ b/graph-persistence/argi-graph-persistence-redis/src/main/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressReferenceConsumer.java @@ -0,0 +1,173 @@ +/* + * 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.graph.persistence.redis; + +import java.time.Duration; +import java.util.Objects; +import java.util.UUID; +import java.util.function.Predicate; + +import io.github.agentic.ai.graph.checkpoint.lease.LeaseBusyException; +import io.github.agentic.ai.graph.checkpoint.lease.LeaseLostException; + +/** + * Minimal reference consumption contract, not Core/Starter Worker wiring. The application + * owns the state adapter and executor. Claim, re-read revision/state, then guard lease + * immediately before invoking execution. An external effect is not atomic with that + * guard: expiry after the guard still requires external fencing/idempotency and + * reconciliation. + */ +public final class RedisProgressReferenceConsumer { + + private final RedisLeasedProgressStore store; + + private final RedisProgressIndexRepairer projection; + + private final RedisProgressAdvisoryIndex hints; + + private final Predicate runnable; + + private final Duration ttl; + + private final long backoffMillis; + + public RedisProgressReferenceConsumer(RedisLeasedProgressStore store, RedisProgressIndexRepairer projection, + RedisProgressAdvisoryIndex hints, Predicate runnable, Duration ttl, + long backoffMillis) { + this.store = Objects.requireNonNull(store); + this.projection = Objects.requireNonNull(projection); + this.hints = Objects.requireNonNull(hints); + if (!store.keyPrefix().equals(projection.prefix()) || store.client() != projection.client()) { + throw new IllegalArgumentException("Consumer store and projection must share client and prefix"); + } + this.runnable = Objects.requireNonNull(runnable); + if (ttl == null || ttl.isZero() || ttl.isNegative() || ttl.compareTo(Duration.ofDays(1)) > 0 + || ttl.getNano() % 1000000 != 0 || backoffMillis < 1 || backoffMillis > 86400000) { + throw new IllegalArgumentException("Invalid lease TTL or backoff"); + } + this.ttl = ttl; + this.backoffMillis = backoffMillis; + } + + /** + * The owner ID must be unique per attempt. Any unexpected exception propagates after + * retry scheduling and lease release. An unknown commit outcome must be reconciled by + * re-reading authority on redelivery; a timeout does not roll back external effects. + */ + public Result consume(RedisProgressAdvisoryIndex.Delivery delivery, UUID owner, Executor executor) + throws Exception { + var candidate = delivery.candidate(); + // Do not bootstrap missing authority merely because an advisory hint exists. + try { + if (projection.metadata(candidate.sessionId(), candidate.runId()).isEmpty()) { + hints.ack(delivery); + return new Result("MISSING_AUTHORITY", null); + } + } + catch (RedisProgressIndexRepairer.MetadataException ex) { + hints.retry(delivery, backoffMillis); + return new Result("METADATA_ERROR", ex.getMessage()); + } + LeasedProgressStore.SessionLease lease; + try { + lease = store.claim(candidate.sessionId(), candidate.runId(), owner, ttl); + } + catch (LeaseBusyException ex) { + hints.retry(delivery, backoffMillis); + return new Result("BUSY", null); + } + boolean settled = false; + try { + var metadata = projection.metadata(lease); + if (metadata.isEmpty() || metadata.get().revision() != candidate.revision() + || !runnable.test(metadata.get())) { + hints.ack(delivery); + settled = true; + return new Result("STALE_OR_INELIGIBLE", null); + } + Execution execution = new Execution(lease, metadata.get()); + execution.guard(); + LeasedProgressStore.ProgressCommit commit = Objects.requireNonNull(executor.execute(execution)); + if (commit.expectedRunRevision() != candidate.revision()) { + throw new IllegalArgumentException("Executor commit must match the validated revision"); + } + store.commit(execution.lease, commit); + hints.ack(delivery); + settled = true; + return new Result("COMMITTED", null); + } + catch (LeaseLostException ex) { + return new Result("LEASE_LOST", null); + } + catch (RedisProgressIndexRepairer.MetadataException ex) { + return new Result("METADATA_ERROR", ex.getMessage()); + } + finally { + try { + if (!settled) { + hints.retry(delivery, backoffMillis); + } + } + finally { + store.release(lease); + } + } + } + + /** + * Per-attempt context. Call guard before each effect; it cannot fence that effect + * itself. + */ + public final class Execution { + + private LeasedProgressStore.SessionLease lease; + + private final RedisProgressIndexRepairer.Metadata metadata; + + private Execution(LeasedProgressStore.SessionLease lease, RedisProgressIndexRepairer.Metadata metadata) { + this.lease = lease; + this.metadata = metadata; + } + + public RedisProgressIndexRepairer.Metadata metadata() { + return metadata; + } + + public LeasedProgressStore.SessionLease lease() { + return lease; + } + + public void guard() { + lease = store.renew(lease, ttl); + var current = projection.metadata(lease); + if (current.isEmpty() || !metadata.equals(current.get())) { + throw new ProgressConflictException("Authority changed after claim"); + } + } + + } + + @FunctionalInterface + public interface Executor { + + LeasedProgressStore.ProgressCommit execute(Execution execution) throws Exception; + + } + + public record Result(String outcome, String error) { + } + +} diff --git a/graph-persistence/argi-graph-persistence-redis/src/test/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressDirectoryTest.java b/graph-persistence/argi-graph-persistence-redis/src/test/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressDirectoryTest.java new file mode 100644 index 00000000..f13e9169 --- /dev/null +++ b/graph-persistence/argi-graph-persistence-redis/src/test/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressDirectoryTest.java @@ -0,0 +1,640 @@ +/* + * 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.graph.persistence.redis; + +import java.io.ByteArrayOutputStream; +import java.io.InputStream; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Predicate; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.redisson.Redisson; +import org.redisson.api.RScript; +import org.redisson.api.RedissonClient; +import org.redisson.client.codec.StringCodec; +import org.redisson.config.Config; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.utility.DockerImageName; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Real Valkey, one JVM, explicit application adapter, fake executor. */ +@Testcontainers(disabledWithoutDocker = true) +class RedisProgressDirectoryTest { + + @Container + private static final GenericContainer VALKEY = new GenericContainer<>( + DockerImageName.parse("valkey/valkey:8.1.2")) + .withExposedPorts(6379) + .withCommand("valkey-server", "--appendonly", "yes", "--appendfsync", "always"); + + private static RedissonClient client; + + @BeforeAll + static void setup() { + client = newClient(); + } + + @AfterAll + static void close() { + client.shutdown(); + } + + @Test + void claimRegistersAtomicallyAndCommitBeforePublisherCrashIsRecoverable() { + String prefix = prefix(); + RedissonClient producer = newClient(); + try { + var store = store(producer, prefix, 4); + var lease = store.claim("s", "r", UUID.randomUUID(), Duration.ofSeconds(30)); + assertThat(new RedisProgressDirectory(client, prefix, 4).page("", 1).entries()).hasSize(1); + store.commit(lease, commit("READY", 0)); + store.release(lease); + // Graceful connection boundary, skipped publisher, NOT JVM/Redis kill. + } + finally { + producer.shutdown(); + } + var hints = new RedisProgressAdvisoryIndex(client, prefix + ":hints", 4); + assertThat(hints.due(4)).isEmpty(); + var directory = new RedisProgressDirectory(client, prefix, 4); + var result = repair(directory, "", 4, projection(prefix, hints), hints); + assertThat(result.published()).isEqualTo(1); + assertThat(result.issues()).isEmpty(); + assertThat(hints.due(4)).containsExactly(new RedisProgressAdvisoryIndex.Candidate("s", "r", 1)); + assertThat(repair(directory, "", 4, projection(prefix, hints), hints).published()).isZero(); + } + + @Test + void capacityAndWrongTypeRejectBeforeAuthorityCreationAndExistingMemberCanReclaim() { + String prefix = prefix(); + var store = store(client, prefix, 1); + var first = store.claim("s1", "r", UUID.randomUUID(), Duration.ofSeconds(30)); + store.release(first); + assertThatThrownBy(() -> store.claim("s2", "r", UUID.randomUUID(), Duration.ofSeconds(30))) + .hasMessageContaining("DIRECTORY_FULL"); + assertThat(store.snapshot("s2", "r")).isEmpty(); + assertThat(client.getKeys().countExists(new RedisProgressKeys(prefix).sessionKey("s2"))).isZero(); + var second = store.claim("s1", "r", UUID.randomUUID(), Duration.ofSeconds(30)); + assertThat(second.fencingToken()).isGreaterThan(first.fencingToken()); + store.release(second); + assertThatThrownBy(() -> new RedisProgressDirectory(client, prefix, 2)).hasMessageContaining("CONFIG"); + String broken = prefix(); + var brokenStore = store(client, broken, 1); + client.getBucket(broken + ":directory:v1:members", StringCodec.INSTANCE).set("wrong type"); + assertThatThrownBy(() -> brokenStore.claim("s", "r", UUID.randomUUID(), Duration.ofSeconds(30))) + .hasMessageContaining("DIRECTORY_TYPE"); + assertThat(brokenStore.snapshot("s", "r")).isEmpty(); + } + + @Test + void concurrentCapacityAdmissionHasOneWinnerAndNoOrphanAuthority() throws Exception { + String prefix = prefix(); + var store = store(client, prefix, 1); + var pool = Executors.newFixedThreadPool(2); + CountDownLatch start = new CountDownLatch(1); + try { + List> attempts = new ArrayList<>(); + for (String session : List.of("a", "b")) { + attempts.add(() -> { + start.await(); + try { + store.claim(session, "r", UUID.randomUUID(), Duration.ofSeconds(30)); + return true; + } + catch (IllegalStateException ex) { + assertThat(ex).hasMessageContaining("DIRECTORY_FULL"); + assertThat(store.snapshot(session, "r")).isEmpty(); + return false; + } + }); + } + var a = pool.submit(attempts.get(0)); + var b = pool.submit(attempts.get(1)); + start.countDown(); + assertThat(List.of(a.get(10, TimeUnit.SECONDS), b.get(10, TimeUnit.SECONDS))) + .containsExactlyInAnyOrder(true, false); + assertThat(new RedisProgressDirectory(client, prefix, 1).page("", 10).entries()).hasSize(1); + } + finally { + pool.shutdownNow(); + } + } + + @Test + void lexCursorHasNoOverflowAndConcurrentReadersCoverRetainedMembers() throws Exception { + String prefix = prefix(); + var store = store(client, prefix, 40); + for (int i = 0; i < 31; i++) { + var lease = store.claim("s" + i, "r", UUID.randomUUID(), Duration.ofSeconds(30)); + store.release(lease); + } + var directory = new RedisProgressDirectory(client, prefix, 40); + var pool = Executors.newFixedThreadPool(2); + try { + Callable> sweep = () -> { + var seen = new HashSet(); + String cursor = ""; + for (int i = 0; i < 40; i++) { + var page = directory.page(cursor, 1); + assertThat(page.entries()).hasSizeLessThanOrEqualTo(1); + for (var id : page.entries()) { + assertThat(seen.add(id)).isTrue(); + } + cursor = page.after(); + if (page.done()) { + return seen; + } + } + throw new AssertionError("Unbounded cursor"); + }; + var results = pool.invokeAll(List.of(sweep, sweep)); + assertThat(results.get(0).get()).hasSize(31).isEqualTo(results.get(1).get()); + } + finally { + pool.shutdownNow(); + } + } + + @Test + void utf8IdentityBoundaryAndActualRespPageBytesAreBounded() throws Exception { + String prefix = prefix(); + var store = store(client, prefix, 2); + String id = "界".repeat(341) + "a"; + var lease = store.claim(id, id, UUID.randomUUID(), Duration.ofSeconds(30)); + store.release(lease); + String encoded = RedisProgressDirectory.member(id, id); + assertThat(encoded).hasSize(2733); + try (Socket socket = new Socket(VALKEY.getHost(), VALKEY.getMappedPort(6379))) { + socket.setSoTimeout(10000); + String[] args = { "ZRANGEBYLEX", prefix + ":directory:v1:members", "-", "+", "LIMIT", "0", "1" }; + ByteArrayOutputStream request = new ByteArrayOutputStream(); + request.write(("*" + args.length + "\r\n").getBytes(StandardCharsets.US_ASCII)); + for (String arg : args) { + byte[] bytes = arg.getBytes(StandardCharsets.UTF_8); + request.write(("$" + bytes.length + "\r\n").getBytes(StandardCharsets.US_ASCII)); + request.write(bytes); + request.write("\r\n".getBytes(StandardCharsets.US_ASCII)); + } + socket.getOutputStream().write(request.toByteArray()); + InputStream in = socket.getInputStream(); + byte[] response = in.readNBytes(2746); + assertThat(response).hasSize(2746); + assertThat(new String(response, StandardCharsets.UTF_8)).isEqualTo("*1\r\n$2733\r\n" + encoded + "\r\n"); + } + assertThatThrownBy(() -> store.claim(id + "a", "x", UUID.randomUUID(), Duration.ofSeconds(30))) + .hasMessageContaining("1024"); + assertThat(store.snapshot(id + "a", "x")).isEmpty(); + assertThatThrownBy(() -> new RedisProgressDirectory(client, prefix, 2).page("x".repeat(2734), 1)) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void explicitBackfillIsRequiredAndBadLargeMissingRecordsDoNotBlockOtherAuthority() { + String prefix = prefix(); + var legacy = RedisLeasedProgressStore.builder().redisson(client).keyPrefix(prefix).build(); + for (String session : List.of("bad", "big", "missing", "good")) { + var lease = legacy.claim(session, "r", UUID.randomUUID(), Duration.ofSeconds(30)); + legacy.commit(lease, commit("READY", 0)); + legacy.release(lease); + } + var directory = new RedisProgressDirectory(client, prefix, 4); + assertThat(directory.page("", 10).entries()).isEmpty(); + for (String session : List.of("bad", "big", "missing", "good")) { + directory.registerExisting(session, "r"); + directory.registerExisting(session, "r"); + } + runMap(prefix, "bad").put("state", ""); + runMap(prefix, "big").put("state", "x".repeat(1000000)); + client.getKeys().delete((String) new RedisProgressKeys(prefix).keys("missing", "r").get(1)); + var hints = new RedisProgressAdvisoryIndex(client, prefix + ":hints", 4); + var result = repair(directory, "", 4, projection(prefix, hints), hints); + assertThat(result.issues()).hasSize(3); + assertThat(result.page().entries()).hasSize(4); + assertThat(result.published()).isEqualTo(1); + assertThat(hints.due(10)).containsExactly(new RedisProgressAdvisoryIndex.Candidate("good", "r", 1)); + assertThat(runMap(prefix, "big").get("state")).hasSize(1000000); + runMap(prefix, "big").put("state", "READY"); + runMap(prefix, "bad").put("state", "READY"); + assertThat(repair(directory, "", 4, projection(prefix, hints), hints).published()).isEqualTo(2); + } + + @Test + void visibilityAndRetryAdvancePastStaleBusyAndAbandonedFirstPage() throws Exception { + String prefix = prefix(); + var hints = new RedisProgressAdvisoryIndex(client, prefix + ":hints", 3); + for (int i = 1; i <= 3; i++) { + publish(hints, "s" + i, 1); + } + var first = hints.poll(1, 10000).get(0); + assertThat(hints.retry(first, 10000)).isTrue(); + assertThat(hints.retry(first, 10000)).isFalse(); + var second = hints.poll(1, 10000).get(0); + var third = hints.poll(1, 10000).get(0); + assertThat(List.of(first.candidate(), second.candidate(), third.candidate())).doesNotHaveDuplicates(); + assertThat(hints.poll(1, 10000)).isEmpty(); + assertThat(hints.ack(first)).isFalse(); + assertThat(hints.ack(second)).isTrue(); + assertThat(hints.ack(second)).isFalse(); + // Simulate expiry against Redis TIME, without depending on wall-clock sleeps. + makeDue(prefix, first); + assertThat(hints.poll(1, 10000).get(0).candidate()).isEqualTo(first.candidate()); + } + + @Test + void staleAckCannotDeleteNewRevisionOrNewDeliveryAndCapacityRecovers() { + String prefix = prefix(); + var hints = new RedisProgressAdvisoryIndex(client, prefix + ":hints", 2); + publish(hints, "s", 1); + var old = hints.poll(1, 10000).get(0); + publish(hints, "s", 2); + assertThatThrownBy(() -> publish(hints, "other", 1)).hasMessageContaining("capacity exhausted"); + assertThat(hints.ack(old)).isTrue(); + var newerRevision = hints.poll(1, 10000).get(0); + assertThat(newerRevision.candidate().revision()).isEqualTo(2); + makeDue(prefix, newerRevision); + var redelivery = hints.poll(1, 10000).get(0); + assertThat(redelivery.token()).isGreaterThan(newerRevision.token()); + assertThat(hints.ack(newerRevision)).isFalse(); + assertThat(hints.retry(newerRevision, 1)).isFalse(); + assertThat(hints.ack(redelivery)).isTrue(); + publish(hints, "other", 1); + assertThat(hints.due(10)).hasSize(1); + } + + @Test + void concurrentPollsAndAckPublicationRacePreserveNewRevision() throws Exception { + String prefix = prefix(); + var hints = new RedisProgressAdvisoryIndex(client, prefix + ":hints", 10); + publish(hints, "s", 1); + var pool = Executors.newFixedThreadPool(2); + try { + Callable> poll = () -> hints.poll(1, 10000); + var deliveries = pool.invokeAll(List.of(poll, poll)); + List all = new ArrayList<>(deliveries.get(0).get()); + all.addAll(deliveries.get(1).get()); + assertThat(all).hasSize(1); + var old = all.get(0); + Callable ack = () -> hints.ack(old); + Callable publish = () -> hints.publish(new RedisProgressAdvisoryIndex.Hint( + new RedisProgressIndexRepairer.Metadata("s", "r", 2, "READY"), 0)); + var results = pool.invokeAll(List.of(ack, publish)); + assertThat(results.get(0).get()).isTrue(); + assertThat(results.get(1).get()).isTrue(); + assertThat(hints.due(10)).containsExactly(new RedisProgressAdvisoryIndex.Candidate("s", "r", 2)); + } + finally { + pool.shutdownNow(); + } + } + + @Test + void terminalAndUnknownStateAreNotExecutedAndMissingHintCreatesNoAuthority() throws Exception { + String prefix = prefix(); + var store = store(client, prefix, 10); + var hints = new RedisProgressAdvisoryIndex(client, prefix + ":hints", 10); + var projection = projection(prefix, hints); + var consumer = consumer(store, projection, hints, m -> "READY".equals(m.state())); + for (String state : List.of("DONE", "FUTURE_UNKNOWN")) { + seed(store, state, state); + publish(hints, state, 1); + } + publish(hints, "never-created", 1); + for (var delivery : hints.poll(10, 10000)) { + assertThat(consumer.consume(delivery, UUID.randomUUID(), execution -> { + throw new AssertionError("Ineligible authority executed"); + }).outcome()).isIn("STALE_OR_INELIGIBLE", "MISSING_AUTHORITY"); + } + assertThat(store.snapshot("never-created", "r")).isEmpty(); + assertThat(hints.poll(10, 10000)).isEmpty(); + } + + @Test + void claimThenRevisionRecheckRejectsOldHintAndSuccessfulAttemptCommitsOnce() throws Exception { + String prefix = prefix(); + var store = store(client, prefix, 10); + seed(store, "s", "READY"); + var hints = new RedisProgressAdvisoryIndex(client, prefix + ":hints", 10); + publish(hints, "s", 0); + publish(hints, "s", 1); + var consumer = consumer(store, projection(prefix, hints), hints, m -> "READY".equals(m.state())); + AtomicInteger calls = new AtomicInteger(); + var result = new ArrayList(); + for (var delivery : hints.poll(10, 10000)) { + result.add(consumer.consume(delivery, UUID.randomUUID(), execution -> { + execution.guard(); + calls.incrementAndGet(); + return commit("DONE", execution.metadata().revision()); + }).outcome()); + } + assertThat(result).containsExactly("STALE_OR_INELIGIBLE", "COMMITTED"); + assertThat(calls).hasValue(1); + assertThat(store.snapshot("s", "r").orElseThrow().state()).isEqualTo("DONE"); + publish(hints, "s", 1); + assertThat(consumer.consume(hints.poll(1, 10000).get(0), UUID.randomUUID(), e -> { + throw new AssertionError("Old revision reran"); + }).outcome()).isEqualTo("STALE_OR_INELIGIBLE"); + } + + @Test + void twoConsumersCompeteForOneLeaseAndOnlyOneExecutorRuns() throws Exception { + String prefix = prefix(); + var store = store(client, prefix, 10); + seed(store, "s", "READY"); + var hints = new RedisProgressAdvisoryIndex(client, prefix + ":hints", 10); + publish(hints, "s", 1); + var delivery = hints.poll(1, 10000).get(0); + var consumer = consumer(store, projection(prefix, hints), hints, m -> "READY".equals(m.state())); + var pool = Executors.newFixedThreadPool(2); + CountDownLatch executing = new CountDownLatch(1); + CountDownLatch finish = new CountDownLatch(1); + AtomicInteger calls = new AtomicInteger(); + try { + var first = pool.submit(() -> consumer.consume(delivery, UUID.randomUUID(), e -> { + executing.countDown(); + assertThat(finish.await(10, TimeUnit.SECONDS)).isTrue(); + e.guard(); + calls.incrementAndGet(); + return commit("DONE", 1); + })); + assertThat(executing.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(consumer.consume(delivery, UUID.randomUUID(), e -> { + throw new AssertionError("Second owner executed"); + }).outcome()).isEqualTo("BUSY"); + finish.countDown(); + assertThat(first.get(10, TimeUnit.SECONDS).outcome()).isEqualTo("COMMITTED"); + assertThat(calls).hasValue(1); + // A competing retry may invalidate the old ack token; stale hint is harmless. + makeDue(prefix, delivery); + for (var retry : hints.poll(10, 10000)) { + assertThat(consumer.consume(retry, UUID.randomUUID(), e -> { + throw new AssertionError("Committed run reran"); + }).outcome()).isEqualTo("STALE_OR_INELIGIBLE"); + } + } + finally { + finish.countDown(); + pool.shutdownNow(); + } + } + + @Test + void leaseLostBeforeDispatchAndDuringExecutorBlocksEffectsAndCommit() throws Exception { + String prefix = prefix(); + var store = store(client, prefix, 10); + seed(store, "s", "READY"); + var hints = new RedisProgressAdvisoryIndex(client, prefix + ":hints", 10); + var projection = projection(prefix, hints); + publish(hints, "s", 1); + var delivery = hints.poll(1, 10000).get(0); + AtomicInteger effects = new AtomicInteger(); + var consumer = consumer(store, projection, hints, m -> { + expire(prefix, "s"); + return "READY".equals(m.state()); + }); + assertThat(consumer.consume(delivery, UUID.randomUUID(), e -> { + effects.incrementAndGet(); + return commit("DONE", 1); + }).outcome()).isEqualTo("LEASE_LOST"); + assertThat(effects).hasValue(0); + makeDue(prefix, delivery); + var another = hints.poll(1, 10000).get(0); + var normal = consumer(store, projection, hints, m -> "READY".equals(m.state())); + assertThat(normal.consume(another, UUID.randomUUID(), e -> { + expire(prefix, "s"); + var takeover = store.claim("s", "r", UUID.randomUUID(), Duration.ofSeconds(30)); + try { + e.guard(); + effects.incrementAndGet(); + return commit("DONE", 1); + } + finally { + store.release(takeover); + } + }).outcome()).isEqualTo("LEASE_LOST"); + assertThat(effects).hasValue(0); + assertThat(store.snapshot("s", "r").orElseThrow().runRevision()).isEqualTo(1); + } + + @Test + void executorFailureRetainsAuthorityAndCrashAfterEffectDoesNotPromiseExactlyOnce() throws Exception { + String prefix = prefix(); + var store = store(client, prefix, 10); + seed(store, "s", "READY"); + var hints = new RedisProgressAdvisoryIndex(client, prefix + ":hints", 10); + var consumer = consumer(store, projection(prefix, hints), hints, m -> "READY".equals(m.state())); + publish(hints, "s", 1); + var delivery = hints.poll(1, 10000).get(0); + AtomicInteger effects = new AtomicInteger(); + assertThatThrownBy(() -> consumer.consume(delivery, UUID.randomUUID(), e -> { + e.guard(); + effects.incrementAndGet(); + throw new IllegalStateException("fake crash after external effect before commit"); + })).hasMessageContaining("fake crash"); + assertThat(store.snapshot("s", "r").orElseThrow().runRevision()).isEqualTo(1); + makeDue(prefix, delivery); + assertThat(consumer.consume(hints.poll(1, 10000).get(0), UUID.randomUUID(), e -> { + e.guard(); + effects.incrementAndGet(); + return commit("DONE", 1); + }).outcome()).isEqualTo("COMMITTED"); + assertThat(effects).hasValue(2); // Explicit counterexample to external + // exactly-once. + } + + @Test + void discoveryContinuationDoesNotRetainAnUnboundedScanPage() { + String prefix = prefix(); + var store = store(client, prefix, 40); + for (int i = 0; i < 31; i++) { + var lease = store.claim("s" + i, "r", UUID.randomUUID(), Duration.ofSeconds(30)); + store.release(lease); + } + int retained; + if (Boolean.getBoolean("acceptance.legacyScan")) { + var hints = new RedisProgressAdvisoryIndex(client, prefix + ":hints"); + var repairer = projection(prefix, hints); + retained = repairer.repair(repairer.start(), 1, 10000).cursor().pendingKeys(); + } + else { + retained = new RedisProgressDirectory(client, prefix, 40).page("", 1).entries().size(); + } + assertThat(retained).as("one-item discovery call must not retain SCAN overflow").isLessThanOrEqualTo(1); + } + + @Test + void redisTimeVisibilityExpiresWithoutClientClockInput() throws Exception { + var hints = new RedisProgressAdvisoryIndex(client, prefix() + ":hints", 1); + publish(hints, "s", 1); + var first = hints.poll(1, 40).get(0); + assertThat(hints.poll(1, 40)).isEmpty(); + Thread.sleep(80); + var second = hints.poll(1, 10000).get(0); + assertThat(second.candidate()).isEqualTo(first.candidate()); + assertThat(second.token()).isGreaterThan(first.token()); + assertThat(hints.ack(first)).isFalse(); + assertThat(hints.ack(second)).isTrue(); + } + + @Test + void leasedMetadataValidationRejectsReleasedExpiredAndSupersededFences() { + String prefix = prefix(); + var store = store(client, prefix, 2); + var hints = new RedisProgressAdvisoryIndex(client, prefix + ":hints", 2); + var projection = projection(prefix, hints); + var first = store.claim("s", "r", UUID.randomUUID(), Duration.ofSeconds(30)); + store.commit(first, commit("READY", 0)); + assertThat(projection.metadata(first).orElseThrow().revision()).isEqualTo(1); + store.release(first); + assertThatThrownBy(() -> projection.metadata(first)) + .isInstanceOf(io.github.agentic.ai.graph.checkpoint.lease.LeaseLostException.class); + var second = store.claim("s", "r", UUID.randomUUID(), Duration.ofSeconds(30)); + assertThatThrownBy(() -> projection.metadata(first)) + .isInstanceOf(io.github.agentic.ai.graph.checkpoint.lease.LeaseLostException.class); + expire(prefix, "s"); + assertThatThrownBy(() -> projection.metadata(second)) + .isInstanceOf(io.github.agentic.ai.graph.checkpoint.lease.LeaseLostException.class); + assertThatThrownBy(() -> store.commit(second, commit("DONE", 1))) + .isInstanceOf(io.github.agentic.ai.graph.checkpoint.lease.LeaseLostException.class); + assertThat(store.snapshot("s", "r").orElseThrow().state()).isEqualTo("READY"); + } + + @Test + void directoryStateLimitIsOptInAndPublicationFailureCanBeRepaired() { + String prefix = prefix(); + var store = store(client, prefix, 2); + var lease = store.claim("s", "r", UUID.randomUUID(), Duration.ofSeconds(30)); + assertThatThrownBy(() -> store.commit(lease, commit("x".repeat(4097), 0))).hasMessageContaining("4096"); + assertThat(store.snapshot("s", "r").orElseThrow().runRevision()).isZero(); + store.commit(lease, commit("READY", 0)); + store.release(lease); + var hints = new RedisProgressAdvisoryIndex(client, prefix + ":hints", 2); + var directory = new RedisProgressDirectory(client, prefix, 2); + client.getBucket(prefix + ":hints", StringCodec.INSTANCE).set("wrong type"); + assertThatThrownBy(() -> repair(directory, "", 2, projection(prefix, hints), hints)) + .hasMessageContaining("WRONGTYPE"); + assertThat(store.snapshot("s", "r").orElseThrow().runRevision()).isEqualTo(1); + assertThat(directory.page("", 2).entries()).hasSize(1); + client.getKeys().delete(prefix + ":hints"); + assertThat(repair(directory, "", 2, projection(prefix, hints), hints).published()).isEqualTo(1); + String legacyPrefix = prefix(); + var legacy = RedisLeasedProgressStore.builder().redisson(client).keyPrefix(legacyPrefix).build(); + var oldLease = legacy.claim("s", "r", UUID.randomUUID(), Duration.ofSeconds(30)); + assertThat(legacy.commit(oldLease, commit("x".repeat(4097), 0)).state()).hasSize(4097); + legacy.release(oldLease); + } + + @Test + void deliveryTokensKeepExactInt64AndExhaustionMakesNoPartialDelivery() { + String prefix = prefix(); + var hints = new RedisProgressAdvisoryIndex(client, prefix + ":hints", 1); + publish(hints, "s", Long.MAX_VALUE); + var tokens = client.getMap(prefix + ":hints:deliveries:v1", StringCodec.INSTANCE); + tokens.put("_sequence", "9007199254740992"); + var delivery = hints.poll(1, 10000).get(0); + assertThat(delivery.token()).isEqualTo(9007199254740993L); + assertThat(delivery.candidate().revision()).isEqualTo(Long.MAX_VALUE); + assertThat(hints.ack(delivery)).isTrue(); + publish(hints, "s", Long.MAX_VALUE); + tokens.put("_sequence", Long.toString(Long.MAX_VALUE)); + assertThatThrownBy(() -> hints.poll(1, 10000)).hasMessageContaining("exhausted"); + assertThat(hints.due(1)).hasSize(1); + assertThat(tokens.size()).isEqualTo(1); + } + + private static RedisProgressDirectory.RepairBatch repair(RedisProgressDirectory directory, String after, int limit, + RedisProgressIndexRepairer projection, RedisProgressAdvisoryIndex hints) { + return directory.repair(after, limit, projection, m -> "READY".equals(m.state()) + ? Optional.of(new RedisProgressAdvisoryIndex.Hint(m, 0)) : Optional.empty(), hints); + } + + private static RedisProgressReferenceConsumer consumer(RedisLeasedProgressStore store, + RedisProgressIndexRepairer projection, RedisProgressAdvisoryIndex hints, + Predicate runnable) { + return new RedisProgressReferenceConsumer(store, projection, hints, runnable, Duration.ofSeconds(30), 10000); + } + + private static RedisProgressIndexRepairer projection(String prefix, RedisProgressAdvisoryIndex hints) { + return new RedisProgressIndexRepairer(client, prefix, m -> Optional.empty(), hints); + } + + private static RedisLeasedProgressStore store(RedissonClient client, String prefix, int capacity) { + return RedisLeasedProgressStore.builder() + .redisson(client) + .keyPrefix(prefix) + .directoryCapacity(capacity) + .build(); + } + + private static LeasedProgressStore.ProgressCommit commit(String state, long revision) { + return new LeasedProgressStore.ProgressCommit(state, revision, null, List.of(), List.of()); + } + + private static void seed(RedisLeasedProgressStore store, String session, String state) { + var lease = store.claim(session, "r", UUID.randomUUID(), Duration.ofSeconds(30)); + store.commit(lease, commit(state, 0)); + store.release(lease); + } + + private static void publish(RedisProgressAdvisoryIndex hints, String session, long revision) { + hints.publish(new RedisProgressAdvisoryIndex.Hint( + new RedisProgressIndexRepairer.Metadata(session, "r", revision, "READY"), 0)); + } + + private static void makeDue(String prefix, RedisProgressAdvisoryIndex.Delivery delivery) { + var c = delivery.candidate(); + String member = RedisProgressDirectory.member(c.sessionId(), c.runId()) + "." + c.revision(); + client.getScript(StringCodec.INSTANCE) + .eval(RScript.Mode.READ_WRITE, "return redis.call('ZADD', KEYS[1], 'XX', 0, ARGV[1])", + RScript.ReturnType.INTEGER, List.of(prefix + ":hints"), member); + } + + private static void expire(String prefix, String session) { + client.getMap(new RedisProgressKeys(prefix).sessionKey(session), StringCodec.INSTANCE).put("expires", "0"); + } + + private static org.redisson.api.RMap runMap(String prefix, String session) { + return client.getMap((String) new RedisProgressKeys(prefix).keys(session, "r").get(1), StringCodec.INSTANCE); + } + + private static String prefix() { + return "test:directory:" + UUID.randomUUID(); + } + + private static RedissonClient newClient() { + Config config = new Config(); + config.useSingleServer().setAddress("redis://" + VALKEY.getHost() + ":" + VALKEY.getMappedPort(6379)); + return Redisson.create(config); + } + +} diff --git a/graph-persistence/argi-graph-persistence-redis/src/test/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressIndexRepairerTest.java b/graph-persistence/argi-graph-persistence-redis/src/test/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressIndexRepairerTest.java new file mode 100644 index 00000000..bb3b2479 --- /dev/null +++ b/graph-persistence/argi-graph-persistence-redis/src/test/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressIndexRepairerTest.java @@ -0,0 +1,377 @@ +/* + * 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.graph.persistence.redis; + +import java.time.Duration; +import java.util.List; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.Callable; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.redisson.Redisson; +import org.redisson.api.RedissonClient; +import org.redisson.client.codec.StringCodec; +import org.redisson.config.Config; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.utility.DockerImageName; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Real single-primary storage tests; no model, executor or process-kill fixture. */ +@Testcontainers(disabledWithoutDocker = true) +class RedisProgressIndexRepairerTest { + + @Container + private static final GenericContainer VALKEY = new GenericContainer<>( + DockerImageName.parse("valkey/valkey:8.1.2")) + .withExposedPorts(6379) + .withCommand("valkey-server", "--appendonly", "yes", "--appendfsync", "always"); + + private static RedissonClient client; + + @BeforeAll + static void setup() { + client = newClient(); + } + + @AfterAll + static void close() { + if (client != null) { + client.shutdown(); + } + } + + @Test + void committedRunAndOutboxAreDiscoveredWithoutEverCallingPublisher() { + String prefix = prefix(); + String indexKey = prefix + ":hints"; + RedissonClient producer = newClient(); + try { + var store = store(producer, prefix); + var lease = store.claim("original-session", "original-run", UUID.randomUUID(), Duration.ofSeconds(30)); + store.commit(lease, + new LeasedProgressStore.ProgressCommit("READY", 0, null, + List.of(new LeasedProgressStore.Entry("journal", "opaque journal")), + List.of(new LeasedProgressStore.Entry("event", "opaque event")))); + // Simulate the boundary after commit and before publish: publisher is never + // called. + // Shutdown is graceful; this is NOT an OS kill or a Redis-restart test. + } + finally { + producer.shutdown(); + } + assertThat(client.getKeys().countExists(indexKey)).isZero(); + var sessionKey = new RedisProgressKeys(prefix).sessionKey("original-session"); + var leaseBeforeRepair = client.getMap(sessionKey, StringCodec.INSTANCE).readAllMap(); + RedissonClient recovery = newClient(); + try { + var sink = new RedisProgressAdvisoryIndex(recovery, indexKey); + var repairer = readyRepairer(recovery, prefix, sink); + assertThat(store(recovery, prefix).snapshot("original-session", "original-run").orElseThrow().outbox()) + .containsEntry("event", "opaque event"); + assertThat(sweep(repairer, 2, 10)).isEqualTo(1); + assertThat(sink.due(10)) + .containsExactly(new RedisProgressAdvisoryIndex.Candidate("original-session", "original-run", 1)); + assertThat(sweep(repairer, 2, 10)).isZero(); + assertThat(client.getMap(sessionKey, StringCodec.INSTANCE).readAllMap()).isEqualTo(leaseBeforeRepair); + assertThat(new RedisProgressAdvisoryIndex(client, indexKey).due(10)).isEqualTo(sink.due(10)); + assertThat(store(recovery, prefix).snapshot("original-session", "original-run").orElseThrow().runRevision()) + .isEqualTo(1); + } + finally { + recovery.shutdown(); + } + } + + @Test + void concurrentRepairersPublishOneRevisionMember() throws Exception { + String prefix = prefix(); + commit(prefix, "s", "r", "READY"); + var sink = new RedisProgressAdvisoryIndex(client, prefix + ":hints"); + RedissonClient second = newClient(); + var pool = Executors.newFixedThreadPool(2); + try { + Callable a = () -> sweep(readyRepairer(client, prefix, sink), 10, 100); + Callable b = () -> sweep( + readyRepairer(second, prefix, new RedisProgressAdvisoryIndex(second, prefix + ":hints")), 10, 100); + var results = pool.invokeAll(List.of(a, b)); + assertThat(results.get(0).get() + results.get(1).get()).isEqualTo(1); + assertThat(sink.due(100)).hasSize(1); + } + finally { + pool.shutdownNow(); + assertThat(pool.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + second.shutdown(); + } + } + + @Test + void continuationPreservesOversizedScanPageWhileProcessingOneKeyPerCall() { + String prefix = prefix(); + for (int i = 0; i < 30; i++) { + commit(prefix, "s" + i, "r" + i, "READY"); + } + var sink = new RedisProgressAdvisoryIndex(client, prefix + ":hints"); + var repairer = readyRepairer(client, prefix, sink); + var first = repairer.repair(repairer.start(), 1, 10000); + assertThat(first.inspectedKeys()).isEqualTo(1); + assertThat(first.cursor().pendingKeys()).isGreaterThan(1); + int added = first.newHints(); + var cursor = first.cursor(); + int calls = 0; + while (!cursor.done()) { + var batch = repairer.repair(cursor, 1, 10000); + assertThat(batch.inspectedKeys()).isLessThanOrEqualTo(1); + added += batch.newHints(); + cursor = batch.cursor(); + assertThat(++calls).isLessThan(1000); + } + assertThat(added).isEqualTo(30); + assertThat(sink.due(100)).hasSize(30); + assertThat(repairer.repair(cursor, 1, 1).inspectedKeys()).isZero(); + } + + @Test + void unknownAndTerminalStatesRequireAnExplicitAdapterAndOldHintsStayAdvisory() { + String prefix = prefix(); + var store = store(client, prefix); + var lease = store.claim("s", "r", UUID.randomUUID(), Duration.ofSeconds(30)); + store.commit(lease, new LeasedProgressStore.ProgressCommit("READY", 0, null, List.of(), List.of())); + var sink = new RedisProgressAdvisoryIndex(client, prefix + ":hints"); + var repairer = readyRepairer(client, prefix, sink); + assertThat(sweep(repairer, 10, 100)).isEqualTo(1); + var old = sink.due(10).get(0); + store.commit(lease, new LeasedProgressStore.ProgressCommit("DONE", 1, null, List.of(), List.of())); + commit(prefix, "other", "unknown", "SOME_FUTURE_STATE"); + assertThat(sweep(repairer, 10, 100)).isZero(); + assertThat(sink.due(10)).containsExactly(old); + var current = repairer.metadata(old.sessionId(), old.runId()).orElseThrow(); + assertThat(current.revision()).isNotEqualTo(old.revision()); + assertThat(current.state()).isEqualTo("DONE"); + // No consumer is supplied: this test rejects no execution and proves no terminal + // non-rerun claim. + } + + @Test + void newRevisionDoesNotLetOldPublicationOverwriteNewCandidate() { + String prefix = prefix(); + var store = store(client, prefix); + var lease = store.claim("s", "r", UUID.randomUUID(), Duration.ofSeconds(30)); + store.commit(lease, new LeasedProgressStore.ProgressCommit("READY", 0, null, List.of(), List.of())); + var sink = new RedisProgressAdvisoryIndex(client, prefix + ":hints"); + var repairer = readyRepairer(client, prefix, sink); + var old = repairer.metadata("s", "r").orElseThrow(); + store.commit(lease, new LeasedProgressStore.ProgressCommit("READY", 1, null, List.of(), List.of())); + assertThat(sweep(repairer, 10, 100)).isEqualTo(1); + assertThat(sink.publish(new RedisProgressAdvisoryIndex.Hint(old, 0))).isTrue(); + assertThat(sink.due(10)).extracting(RedisProgressAdvisoryIndex.Candidate::revision) + .containsExactlyInAnyOrder(1L, 2L); + } + + @Test + void projectionDoesNotReadLargeHistoricalPayloadsAndRejectsOversizedState() throws Exception { + String prefix = prefix(); + commit(prefix, "s", "r", "READY"); + var keys = new RedisProgressKeys(prefix).keys("s", "r"); + // Seed large historical data directly to isolate projection from commit's + // existing snapshot read. + for (int i = 2; i < 4; i++) { + client.getMap((String) keys.get(i), StringCodec.INSTANCE).put("large", "x".repeat(1_000_000)); + } + client.getMap((String) keys.get(1), StringCodec.INSTANCE).put("checkpoint", "x".repeat(1_000_000)); + var repairer = readyRepairer(client, prefix, new RedisProgressAdvisoryIndex(client, prefix + ":hints")); + String hgetallBefore = commandStats("hgetall"); + assertThat(sweep(repairer, 10, 100)).isEqualTo(1); + assertThat(repairer.metadata("s", "r").orElseThrow().state()).isEqualTo("READY"); + assertThat(commandStats("hgetall")).isEqualTo(hgetallBefore); + client.getMap((String) keys.get(1), StringCodec.INSTANCE).put("state", "x".repeat(4097)); + String hgetBeforeOversize = commandStats("hget"); + assertThatThrownBy(() -> repairer.metadata("s", "r")).hasMessageContaining("projection limit"); + assertThat(commandStats("hget")).isEqualTo(hgetBeforeOversize); + assertThat(client.getMap((String) keys.get(3), StringCodec.INSTANCE).size()).isEqualTo(1); + } + + @Test + void projectionRejectsInvalidRevisionStatePairsBeforeCallingAdapter() { + String prefix = prefix(); + var store = store(client, prefix); + var lease = store.claim("s", "r", UUID.randomUUID(), Duration.ofSeconds(30)); + var run = client.getMap((String) new RedisProgressKeys(prefix).keys("s", "r").get(1), + StringCodec.INSTANCE); + var sink = new RedisProgressAdvisoryIndex(client, prefix + ":hints"); + var repairer = new RedisProgressIndexRepairer(client, prefix, m -> { + throw new AssertionError("Corrupt metadata must not reach the application adapter"); + }, sink); + assertThat(repairer.metadata("s", "r").orElseThrow()) + .isEqualTo(new RedisProgressIndexRepairer.Metadata("s", "r", 0, "")); + // An absent state is valid at revision zero, but even a present empty state is + // not. + for (String state : List.of("", "READY")) { + run.put("state", state); + assertThatThrownBy(() -> store.snapshot("s", "r")).hasMessageContaining("corrupt"); + assertThatThrownBy(() -> sweep(repairer, 10, 100)).hasMessageContaining("Invalid progress metadata"); + assertThat(run.get("state")).isEqualTo(state); + } + run.remove("state"); + store.commit(lease, new LeasedProgressStore.ProgressCommit("READY", 0, null, List.of(), List.of())); + // A committed revision requires a nonempty state; missing and empty are corrupt. + for (boolean missing : List.of(true, false)) { + if (missing) { + run.remove("state"); + } + else { + run.put("state", ""); + } + assertThatThrownBy(() -> store.snapshot("s", "r")).hasMessageContaining("corrupt"); + assertThatThrownBy(() -> sweep(repairer, 10, 100)).hasMessageContaining("Invalid progress metadata"); + assertThat(run.containsKey("state")).isEqualTo(!missing); + assertThat(run.get("revision")).isEqualTo("1"); + } + assertThat(sink.due(10)).isEmpty(); + } + + @Test + void projectionPreservesOpaqueUtf8LimitsAndExactInt64Revision() { + String prefix = prefix(); + String session = "界".repeat(341) + "s"; // Exactly 1024 UTF-8 bytes. + String state = "界".repeat(1365) + "s"; // Exactly 4096 UTF-8 bytes. + commit(prefix, session, "r", state); + var run = client.getMap((String) new RedisProgressKeys(prefix).keys(session, "r").get(1), + StringCodec.INSTANCE); + run.put("revision", Long.toString(Long.MAX_VALUE)); + var sink = new RedisProgressAdvisoryIndex(client, prefix + ":hints"); + var repairer = new RedisProgressIndexRepairer(client, prefix, + m -> Optional.of(new RedisProgressAdvisoryIndex.Hint(m, 0)), sink); + assertThat(repairer.metadata(session, "r").orElseThrow()) + .isEqualTo(new RedisProgressIndexRepairer.Metadata(session, "r", Long.MAX_VALUE, state)); + assertThat(sweep(repairer, 10, 100)).isEqualTo(1); + assertThat(sink.due(10)) + .containsExactly(new RedisProgressAdvisoryIndex.Candidate(session, "r", Long.MAX_VALUE)); + assertThat(store(client, prefix).snapshot(session, "r").orElseThrow().runRevision()).isEqualTo(Long.MAX_VALUE); + // The established store accepts these IDs; only the opt-in reader imposes a byte + // cap. + String oversized = "界".repeat(342); + commit(prefix, oversized, "r", "READY"); + commit(prefix, "other", oversized, "READY"); + assertThatThrownBy(() -> repairer.metadata(oversized, "r")).hasMessageContaining("projection limit"); + assertThatThrownBy(() -> repairer.metadata("other", oversized)).hasMessageContaining("projection limit"); + assertThat(store(client, prefix).snapshot(oversized, "r").orElseThrow().state()).isEqualTo("READY"); + assertThat(store(client, prefix).snapshot("other", oversized).orElseThrow().state()).isEqualTo("READY"); + } + + @Test + void redisPublishErrorLeavesAuthorityIntactAndFreshSweepCanRetry() { + String prefix = prefix(); + commit(prefix, "s", "r", "READY"); + var before = store(client, prefix).snapshot("s", "r").orElseThrow(); + String indexKey = prefix + ":hints"; + client.getBucket(indexKey, StringCodec.INSTANCE).set("wrong-type"); + var repairer = readyRepairer(client, prefix, new RedisProgressAdvisoryIndex(client, indexKey)); + assertThatThrownBy(() -> sweep(repairer, 10, 100)).hasMessageContaining("WRONGTYPE"); + assertThat(store(client, prefix).snapshot("s", "r").orElseThrow()).isEqualTo(before); + // Fixture cleanup of its own advisory key only. + client.getBucket(indexKey).delete(); + assertThat(sweep(repairer, 10, 100)).isEqualTo(1); + } + + @Test + void prefixGlobCharactersDoNotDiscoverAnotherNamespace() { + String prefix = prefix() + "[*?]"; + commit(prefix, "s", "r", "READY"); + commit(prefix + "foreign", "foreign", "foreign", "READY"); + var sink = new RedisProgressAdvisoryIndex(client, prefix + ":hints"); + assertThat(sweep(readyRepairer(client, prefix, sink), 10, 100)).isEqualTo(1); + assertThat(sink.due(10)).containsExactly(new RedisProgressAdvisoryIndex.Candidate("s", "r", 1)); + } + + @Test + void dueQueriesUseServerTimeAndPublicationIsIdempotent() { + String prefix = prefix(); + var sink = new RedisProgressAdvisoryIndex(client, prefix + ":hints"); + var m = new RedisProgressIndexRepairer.Metadata("s", "r", 1, "RETRY"); + var future = new RedisProgressAdvisoryIndex.Hint(m, 9_007_199_254_740_991L); + assertThat(sink.publish(future)).isTrue(); + assertThat(sink.publish(future)).isFalse(); + assertThat(sink.due(10)).isEmpty(); + assertThat(sink.publish(new RedisProgressAdvisoryIndex.Hint( + new RedisProgressIndexRepairer.Metadata("s", "now", 1, "READY"), 0))) + .isTrue(); + assertThat(sink.due(1)).containsExactly(new RedisProgressAdvisoryIndex.Candidate("s", "now", 1)); + assertThatThrownBy(() -> sink.due(1001)).isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> sink.publish(new RedisProgressAdvisoryIndex.Hint(m, Long.MAX_VALUE))) + .isInstanceOf(IllegalArgumentException.class); + } + + private static String commandStats(String command) throws Exception { + var reply = VALKEY.execInContainer("valkey-cli", "INFO", "commandstats"); + assertThat(reply.getExitCode()).isZero(); + return reply.getStdout() + .lines() + .filter(line -> line.startsWith("cmdstat_" + command + ":")) + .findFirst() + .orElse(""); + } + + private static RedisProgressIndexRepairer readyRepairer(RedissonClient connection, String prefix, + RedisProgressAdvisoryIndex sink) { + return new RedisProgressIndexRepairer(connection, prefix, m -> "READY".equals(m.state()) + ? Optional.of(new RedisProgressAdvisoryIndex.Hint(m, 0)) : Optional.empty(), sink); + } + + private static int sweep(RedisProgressIndexRepairer repairer, int maxKeys, int count) { + var cursor = repairer.start(); + int hints = 0; + int calls = 0; + while (!cursor.done()) { + var result = repairer.repair(cursor, maxKeys, count); + assertThat(result.inspectedKeys()).isLessThanOrEqualTo(maxKeys); + hints += result.newHints(); + cursor = result.cursor(); + assertThat(++calls).isLessThan(10000); + } + return hints; + } + + private static void commit(String prefix, String session, String run, String state) { + var store = store(client, prefix); + var lease = store.claim(session, run, UUID.randomUUID(), Duration.ofSeconds(30)); + store.commit(lease, new LeasedProgressStore.ProgressCommit(state, 0, null, List.of(), List.of())); + store.release(lease); + } + + private static RedisLeasedProgressStore store(RedissonClient connection, String prefix) { + return RedisLeasedProgressStore.builder().redisson(connection).keyPrefix(prefix).build(); + } + + private static String prefix() { + return "argi:test:repair:" + UUID.randomUUID(); + } + + private static RedissonClient newClient() { + Config config = new Config(); + config.useSingleServer().setAddress("redis://" + VALKEY.getHost() + ":" + VALKEY.getMappedPort(6379)); + return Redisson.create(config); + } + +} diff --git a/graph-persistence/argi-graph-persistence-redis/src/test/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressProtocolReviewTest.java b/graph-persistence/argi-graph-persistence-redis/src/test/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressProtocolReviewTest.java new file mode 100644 index 00000000..39431016 --- /dev/null +++ b/graph-persistence/argi-graph-persistence-redis/src/test/java/io/github/agentic/ai/graph/persistence/redis/RedisProgressProtocolReviewTest.java @@ -0,0 +1,237 @@ +/* + * 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.graph.persistence.redis; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.redisson.Redisson; +import org.redisson.api.RedissonClient; +import org.redisson.client.codec.StringCodec; +import org.redisson.config.Config; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.utility.DockerImageName; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Independent adversarial review probes; real Valkey, one JVM, fake effects only. */ +@Testcontainers(disabledWithoutDocker = true) +class RedisProgressProtocolReviewTest { + + @Container + private static final GenericContainer VALKEY = new GenericContainer<>( + DockerImageName.parse("valkey/valkey:8.1.2")) + .withExposedPorts(6379) + .withCommand("valkey-server", "--appendonly", "yes", "--appendfsync", "always"); + + private static RedissonClient client; + + @BeforeAll + static void setup() { + Config config = new Config(); + config.useSingleServer().setAddress("redis://" + VALKEY.getHost() + ":" + VALKEY.getMappedPort(6379)); + client = Redisson.create(config); + } + + @AfterAll + static void close() { + if (client != null) { + client.shutdown(); + } + } + + @Test + void terminalAndUncommittedRecordsStillExhaustAdmissionWithoutLosingMembership() { + String prefix = prefix(); + var store = store(prefix, 2); + seed(store, "done", "DONE"); + var zero = store.claim("zero", "r", UUID.randomUUID(), Duration.ofSeconds(30)); + store.release(zero); + assertThatThrownBy(() -> store.claim("new", "r", UUID.randomUUID(), Duration.ofSeconds(30))) + .hasMessageContaining("DIRECTORY_FULL"); + assertThat(store.snapshot("new", "r")).isEmpty(); + assertThat(client.getKeys().countExists(new RedisProgressKeys(prefix).sessionKey("new"))).isZero(); + var directory = new RedisProgressDirectory(client, prefix, 2); + assertThat(directory.page("", 10).entries()).hasSize(2); + var reclaimed = store.claim("done", "r", UUID.randomUUID(), Duration.ofSeconds(30)); + assertThat(store.snapshot("done", "r").orElseThrow().state()).isEqualTo("DONE"); + store.release(reclaimed); + assertThat(directory.page("", 10).entries()).hasSize(2); + } + + @Test + void legacyAndMixedWritersRemainInvisibleUntilExplicitlyInventoried() { + String prefix = prefix(); + var legacy = store(prefix, 0); + seed(legacy, "old", "READY"); + var directory = new RedisProgressDirectory(client, prefix, 3); + var hints = new RedisProgressAdvisoryIndex(client, prefix + ":hints", 3); + var projection = projection(prefix, hints); + assertThat(directory.repair("", 10, projection, this::ready, hints).published()).isZero(); + directory.registerExisting("old", "r"); + directory.registerExisting("old", "r"); + assertThat(directory.repair("", 10, projection, this::ready, hints).published()).isEqualTo(1); + seed(legacy, "late-legacy", "READY"); + assertThat(directory.repair("", 10, projection, this::ready, hints).published()).isZero(); + assertThat(legacy.snapshot("late-legacy", "r")).isPresent(); + assertThat(directory.page("", 10).entries()).containsExactly(new RedisProgressDirectory.Identity("old", "r")); + } + + @Test + void stateAndRevisionChangedBetweenEligibilityAndDispatchCannotExecute() { + String prefix = prefix(); + var store = store(prefix, 2); + seed(store, "s", "READY"); + var hints = new RedisProgressAdvisoryIndex(client, prefix + ":hints", 2); + var projection = projection(prefix, hints); + hints.publish(new RedisProgressAdvisoryIndex.Hint(projection.metadata("s", "r").orElseThrow(), 0)); + AtomicInteger effects = new AtomicInteger(); + var consumer = new RedisProgressReferenceConsumer(store, projection, hints, metadata -> { + // Advance authority through the real fenced store API while this attempt owns + // it. + var ns = client.getMap(new RedisProgressKeys(prefix).sessionKey("s"), StringCodec.INSTANCE); + var lease = new LeasedProgressStore.SessionLease("s", "r", UUID.fromString(ns.get("owner")), + Long.parseLong(ns.get("fence")), Long.parseLong(ns.get("expires"))); + store.commit(lease, commit("DONE", metadata.revision())); + return true; + }, Duration.ofSeconds(30), 10000); + var delivery = hints.poll(1, 10000).get(0); + assertThatThrownBy(() -> consumer.consume(delivery, UUID.randomUUID(), execution -> { + effects.incrementAndGet(); + return commit("EXECUTED", 1); + })).isInstanceOf(ProgressConflictException.class).hasMessageContaining("Authority changed"); + assertThat(effects).hasValue(0); + assertThat(store.snapshot("s", "r").orElseThrow().runRevision()).isEqualTo(2); + assertThat( + client.getMap(new RedisProgressKeys(prefix).sessionKey("s"), StringCodec.INSTANCE).containsKey("owner")) + .isFalse(); + } + + @Test + void reservationIsCheckedIndependentlyOfMatchingOwnerAndFenceAndZeroExpiryIsLost() { + String prefix = prefix(); + var store = store(prefix, 2); + var lease = store.claim("s", "r", UUID.randomUUID(), Duration.ofSeconds(30)); + var hints = new RedisProgressAdvisoryIndex(client, prefix + ":hints", 2); + var projection = projection(prefix, hints); + var ns = client.getMap(new RedisProgressKeys(prefix).sessionKey("s"), StringCodec.INSTANCE); + ns.put("reservation", "other"); + assertThatThrownBy(() -> projection.metadata(lease)) + .isInstanceOf(io.github.agentic.ai.graph.checkpoint.lease.LeaseLostException.class); + ns.put("reservation", "r"); + ns.put("expires", "0"); + assertThatThrownBy(() -> projection.metadata(lease)) + .isInstanceOf(io.github.agentic.ai.graph.checkpoint.lease.LeaseLostException.class); + assertThatThrownBy(() -> store.commit(lease, commit("DONE", 0))) + .isInstanceOf(io.github.agentic.ai.graph.checkpoint.lease.LeaseLostException.class); + assertThat(ns.get("expires")).isEqualTo("0"); + } + + @Test + void oldDeliveryCannotAcknowledgeSameRevisionRepublishedAfterPruning() { + String prefix = prefix(); + var hints = new RedisProgressAdvisoryIndex(client, prefix + ":hints", 1); + var hint = new RedisProgressAdvisoryIndex.Hint(new RedisProgressIndexRepairer.Metadata("s", "r", 1, "READY"), + 0); + hints.publish(hint); + var first = hints.poll(1, 10000).get(0); + assertThat(hints.ack(first)).isTrue(); + assertThat(hints.publish(hint)).isTrue(); + assertThat(hints.ack(first)).isFalse(); + assertThat(hints.retry(first, 10000)).isFalse(); + var second = hints.poll(1, 10000).get(0); + assertThat(second.token()).isGreaterThan(first.token()); + assertThat(hints.ack(first)).isFalse(); + assertThat(hints.ack(second)).isTrue(); + assertThat(client.getMap(prefix + ":hints:deliveries:v1", StringCodec.INSTANCE).size()).isEqualTo(1); + } + + @Test + void oneEntryPagesAdvancePastBadAuthorityAndInsertionBehindCursorAppearsNextSweep() { + String prefix = prefix(); + var store = store(prefix, 8); + for (String id : List.of("m", "n", "o", "p")) { + seed(store, id, "READY"); + } + var keys = new RedisProgressKeys(prefix); + client.getMap((String) keys.keys("m", "r").get(1), StringCodec.INSTANCE).put("revision", "01"); + client.getMap((String) keys.keys("n", "r").get(1), StringCodec.INSTANCE).put("state", "x".repeat(1000000)); + client.getKeys().delete((String) keys.keys("o", "r").get(1)); + var directory = new RedisProgressDirectory(client, prefix, 8); + var hints = new RedisProgressAdvisoryIndex(client, prefix + ":hints", 8); + var projection = projection(prefix, hints); + String after = ""; + var issues = new ArrayList(); + int published = 0; + for (int i = 0; i < 5; i++) { + var result = directory.repair(after, 1, projection, this::ready, hints); + assertThat(result.page().entries()).hasSizeLessThanOrEqualTo(1); + issues.addAll(result.issues()); + published += result.published(); + after = result.page().after(); + if (result.page().done()) { + break; + } + } + assertThat(issues).hasSize(3); + assertThat(published).isEqualTo(1); + seed(store, "a", "READY"); // Its encoded member sorts behind the finished cursor. + assertThat(directory.page(after, 1).entries()).isEmpty(); + assertThat(directory.repair("", 8, projection, this::ready, hints).published()).isEqualTo(1); + assertThat(client.getMap((String) keys.keys("m", "r").get(1), StringCodec.INSTANCE).get("revision")) + .isEqualTo("01"); + assertThat(directory.page("", 8).entries()).hasSize(5); + } + + private Optional ready(RedisProgressIndexRepairer.Metadata metadata) { + return "READY".equals(metadata.state()) ? Optional.of(new RedisProgressAdvisoryIndex.Hint(metadata, 0)) + : Optional.empty(); + } + + private static RedisProgressIndexRepairer projection(String prefix, RedisProgressAdvisoryIndex hints) { + return new RedisProgressIndexRepairer(client, prefix, metadata -> Optional.empty(), hints); + } + + private static RedisLeasedProgressStore store(String prefix, int capacity) { + var builder = RedisLeasedProgressStore.builder().redisson(client).keyPrefix(prefix); + return (capacity == 0 ? builder : builder.directoryCapacity(capacity)).build(); + } + + private static LeasedProgressStore.ProgressCommit commit(String state, long revision) { + return new LeasedProgressStore.ProgressCommit(state, revision, null, List.of(), List.of()); + } + + private static void seed(RedisLeasedProgressStore store, String session, String state) { + var lease = store.claim(session, "r", UUID.randomUUID(), Duration.ofSeconds(30)); + store.commit(lease, commit(state, 0)); + store.release(lease); + } + + private static String prefix() { + return "review:extensions:" + UUID.randomUUID(); + } + +}