Harden distributed coordination for DataProvider 3.4.2

README.md

@@ -25,6 +25,7 @@ It gives you one clean API for MySQL, MongoDB, Redis, and Redis messaging so you
 - Platform support: Velocity + Bukkit/Paper
 - Optional Hibernate ORM support for relational workflows (`nl.hauntedmc.dataprovider.api.orm.ORMContext`)
 - Disposable Pub/Sub plus capability-discoverable durable acknowledged Redis messaging
+- Atomic Redis coordination with renewable fenced leases, monotonic fencing tokens, fenced writes/deletes, and compare-and-set operations
 
 ## Requirements
 
@@ -73,6 +74,8 @@ api.unregisterDatabase(DatabaseType.MYSQL, "example");
 
 If you maintain multiple plugins, this gives your team one standard integration model instead of backend-specific code per project.
 
+For distributed ownership/fencing semantics, see [Distributed coordination](docs/COORDINATION.md).
+
 ## Admin Commands
 
 Paper uses `/dataprovider` (alias `/dp`); Velocity uses `/dataproviderproxy` (alias `/dp`). Both use native,
@@ -131,15 +134,15 @@ Maven:
 <dependency>
   <groupId>nl.hauntedmc.dataprovider</groupId>
   <artifactId>dataprovider-api</artifactId>
-  <version>3.4.1</version>
+  <version>3.4.2</version>
   <scope>provided</scope>
 </dependency>
 ```
 
 Gradle (Groovy):
 
 ```groovy
-compileOnly "nl.hauntedmc.dataprovider:dataprovider-api:3.4.1"
+compileOnly "nl.hauntedmc.dataprovider:dataprovider-api:3.4.2"
 ```
 
 GitHub Packages authentication details are in the docs.
@@ -172,6 +175,7 @@ Build outputs:
 - [Documentation index](docs/README.md)
 - [Architecture](docs/ARCHITECTURE.md)
 - [Usage guide](docs/USAGE_GUIDE.md)
+- [Distributed coordination](docs/COORDINATION.md)
 - [Configuration](docs/CONFIGURATION.md)
 - [Development](docs/DEVELOPMENT.md)
 - [Testing and CI](docs/TESTING_AND_CI.md)

dataprovider-core/src/test/java/nl/hauntedmc/dataprovider/core/integration/RedisCoordinationFencingIT.java

@@ -0,0 +1,218 @@
+package nl.hauntedmc.dataprovider.core.integration;
+
+import nl.hauntedmc.dataprovider.core.database.keyvalue.impl.redis.RedisDatabase;
+import nl.hauntedmc.dataprovider.core.testutil.RecordingLoggerAdapter;
+import nl.hauntedmc.dataprovider.database.coordination.FencedLease;
+import org.junit.jupiter.api.Test;
+import org.spongepowered.configurate.CommentedConfigurationNode;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Qualifies the Redis coordination guarantees consumed by higher-level multi-process runtimes.
+ *
+ * <p>The tests intentionally model owners as {@code logical-node-id/process-incarnation}. A restarted
+ * process must never be able to revive a former lease merely because its logical node name is unchanged.</p>
+ */
+@Testcontainers(disabledWithoutDocker = true)
+class RedisCoordinationFencingIT {
+
+    private static final String REDIS_PASSWORD = "coordination-secret";
+    private static final Duration NORMAL_TTL = Duration.ofSeconds(5);
+
+    @Container
+    private static final GenericContainer<?> REDIS = new GenericContainer<>(DockerImageName.parse("redis:7.4-alpine"))
+            .withExposedPorts(6379)
+            .withCommand("redis-server", "--requirepass", REDIS_PASSWORD);
+
+    @Test
+    void acquireRenewAndReleaseRequireExactLeaseOwnership() throws Exception {
+        RedisDatabase database = database();
+        try {
+            database.connect();
+            assertTrue(database.isConnected());
+            var coordination = database.getCoordinationDataAccess();
+            String resource = resource("exact-owner");
+            String ownerA = owner("proxy-01");
+            String ownerB = owner("proxy-02");
+
+            FencedLease first = coordination.acquire(resource, ownerA, NORMAL_TTL).join().orElseThrow();
+            assertTrue(coordination.acquire(resource, ownerB, NORMAL_TTL).join().isEmpty());
+
+            Thread.sleep(20L);
+            FencedLease renewed = coordination.renew(first, NORMAL_TTL).join().orElseThrow();
+            assertEquals(first.owner(), renewed.owner());
+            assertEquals(first.fencingToken(), renewed.fencingToken());
+            assertTrue(renewed.expiresAt().isAfter(first.expiresAt()));
+
+            FencedLease wrongOwner = new FencedLease(
+                    resource, ownerB, renewed.fencingToken(), renewed.expiresAt());
+            assertTrue(coordination.renew(wrongOwner, NORMAL_TTL).join().isEmpty());
+            assertFalse(coordination.release(wrongOwner).join());
+
+            FencedLease wrongToken = new FencedLease(
+                    resource, ownerA, renewed.fencingToken() + 1_000L, renewed.expiresAt());
+            assertTrue(coordination.renew(wrongToken, NORMAL_TTL).join().isEmpty());
+            assertFalse(coordination.release(wrongToken).join());
+
+            assertTrue(coordination.release(renewed).join());
+            FencedLease next = coordination.acquire(resource, ownerB, NORMAL_TTL).join().orElseThrow();
+            assertTrue(next.fencingToken() > renewed.fencingToken());
+            assertTrue(coordination.release(next).join());
+        } finally {
+            database.disconnect();
+        }
+    }
+
+    @Test
+    void fencingTokensStrictlyIncreaseAcrossProcessIncarnations() {
+        RedisDatabase database = database();
+        try {
+            database.connect();
+            assertTrue(database.isConnected());
+            var coordination = database.getCoordinationDataAccess();
+            String resource = resource("monotonic");
+            List<Long> tokens = new ArrayList<>();
+
+            for (String logicalNode : List.of("proxy-01", "proxy-02", "proxy-01", "proxy-03")) {
+                FencedLease lease = coordination.acquire(resource, owner(logicalNode), NORMAL_TTL)
+                        .join().orElseThrow();
+                tokens.add(lease.fencingToken());
+                assertTrue(coordination.release(lease).join());
+            }
+
+            for (int index = 1; index < tokens.size(); index++) {
+                assertTrue(tokens.get(index) > tokens.get(index - 1),
+                        () -> "Fencing tokens must be strictly increasing: " + tokens);
+            }
+        } finally {
+            database.disconnect();
+        }
+    }
+
+    @Test
+    void authoritativeClaimImmediatelyFencesTheFormerOwner() {
+        RedisDatabase database = database();
+        try {
+            database.connect();
+            assertTrue(database.isConnected());
+            var coordination = database.getCoordinationDataAccess();
+            String resource = resource("claim");
+            String ownerA = owner("proxy-01");
+            String ownerB = owner("proxy-02");
+
+            FencedLease first = coordination.acquire(resource, ownerA, NORMAL_TTL).join().orElseThrow();
+            var claim = coordination.claim(resource, ownerB, NORMAL_TTL).join();
+            FencedLease second = claim.lease();
+
+            assertEquals(ownerA, claim.previousOwner().orElseThrow());
+            assertEquals(first.fencingToken(), claim.previousFencingToken());
+            assertTrue(second.fencingToken() > first.fencingToken());
+            assertTrue(coordination.renew(first, NORMAL_TTL).join().isEmpty());
+            assertFalse(coordination.release(first).join());
+            assertTrue(coordination.renew(second, NORMAL_TTL).join().isPresent());
+            assertTrue(coordination.release(second).join());
+        } finally {
+            database.disconnect();
+        }
+    }
+
+    @Test
+    void staleWriterCannotOverwriteOrDeleteNewerOwnersValue() {
+        RedisDatabase database = database();
+        try {
+            database.connect();
+            assertTrue(database.isConnected());
+            var coordination = database.getCoordinationDataAccess();
+            var values = database.getDataAccess();
+            String resource = resource("stale-writer");
+            String valueKey = "coordination:test:value:" + UUID.randomUUID();
+
+            FencedLease first = coordination.acquire(resource, owner("proxy-01"), NORMAL_TTL)
+                    .join().orElseThrow();
+            assertTrue(coordination.writeFenced(first, valueKey, "generation-a", NORMAL_TTL).join());
+            assertEquals("generation-a", values.getKey(valueKey).join());
+
+            FencedLease second = coordination.claim(resource, owner("proxy-02"), NORMAL_TTL).join().lease();
+            assertTrue(second.fencingToken() > first.fencingToken());
+            assertTrue(coordination.writeFenced(second, valueKey, "generation-b", NORMAL_TTL).join());
+
+            assertFalse(coordination.writeFenced(first, valueKey, "stale-generation", NORMAL_TTL).join());
+            assertFalse(coordination.deleteFenced(first, valueKey).join());
+            assertEquals("generation-b", values.getKey(valueKey).join());
+
+            assertTrue(coordination.deleteFenced(second, valueKey).join());
+            assertEquals(null, values.getKey(valueKey).join());
+            assertTrue(coordination.release(second).join());
+        } finally {
+            database.disconnect();
+        }
+    }
+
+    @Test
+    void reconnectAfterLeaseExpiryDoesNotResurrectOldProcessAuthority() throws Exception {
+        RedisDatabase database = database();
+        try {
+            database.connect();
+            assertTrue(database.isConnected());
+            String resource = resource("reconnect");
+            String valueKey = "coordination:test:reconnect:" + UUID.randomUUID();
+            Duration shortTtl = Duration.ofMillis(500);
+
+            FencedLease former = database.getCoordinationDataAccess()
+                    .acquire(resource, owner("proxy-01"), shortTtl).join().orElseThrow();
+
+            // Drop this client's connection while Redis remains authoritative and lets the lease expire.
+            database.disconnect();
+            Thread.sleep(800L);
+            database.connect();
+            assertTrue(database.isConnected());
+
+            var coordination = database.getCoordinationDataAccess();
+            FencedLease replacement = coordination.acquire(resource, owner("proxy-01"), NORMAL_TTL)
+                    .join().orElseThrow();
+
+            assertTrue(replacement.fencingToken() > former.fencingToken());
+            assertTrue(coordination.renew(former, NORMAL_TTL).join().isEmpty());
+            assertFalse(coordination.writeFenced(former, valueKey, "stale", NORMAL_TTL).join());
+            assertTrue(coordination.writeFenced(replacement, valueKey, "current", NORMAL_TTL).join());
+            assertEquals("current", database.getDataAccess().getKey(valueKey).join());
+            assertTrue(coordination.release(replacement).join());
+        } finally {
+            database.disconnect();
+        }
+    }
+
+    private static RedisDatabase database() {
+        CommentedConfigurationNode config = CommentedConfigurationNode.root();
+        try {
+            config.node("host").set(REDIS.getHost());
+            config.node("port").set(REDIS.getMappedPort(6379));
+            config.node("password").set(REDIS_PASSWORD);
+            config.node("database").set(0);
+            config.node("network_namespace").set("coordination-fencing-it");
+        } catch (org.spongepowered.configurate.serialize.SerializationException exception) {
+            throw new IllegalStateException("Could not build Redis integration configuration.", exception);
+        }
+        return new RedisDatabase(config, new RecordingLoggerAdapter());
+    }
+
+    private static String resource(String purpose) {
+        return "featureframework:test:" + purpose + ':' + UUID.randomUUID();
+    }
+
+    private static String owner(String logicalNodeId) {
+        return logicalNodeId + '/' + UUID.randomUUID();
+    }
+}

docs/COORDINATION.md

@@ -0,0 +1,124 @@
+# Distributed coordination
+
+DataProvider exposes a small, platform-neutral coordination API for consumers that need atomic ownership, fencing, compare-and-set, or indexed TTL state on Redis.
+
+The public entry point is `KeyValueDatabaseProvider#getCoordinationDataAccess()`. Coordination is generic infrastructure: DataProvider does not define application leaders, replica groups, failover policy, or feature placement.
+
+## Lease model
+
+A `FencedLease` contains:
+
+- `resource`: the logical resource being coordinated;
+- `owner`: the exact process incarnation that currently owns it;
+- `fencingToken`: a positive, monotonically increasing ownership generation for that resource;
+- `expiresAt`: the expiry calculated from Redis' authoritative server clock.
+
+Use process-incarnation owners rather than a reusable logical service name. A recommended shape is:
+
+```text
+<logical-node-id>/<unique-process-id>
+```
+
+For example:
+
+```text
+worker-01/93286f21-54e8-45dd-a24b-20b4fb98ef94
+```
+
+A restart of `worker-01` must use a new process ID. This prevents a restarted process from being mistaken for the JVM/process that acquired an older lease.
+
+## Operations
+
+### `acquire(resource, owner, ttl)`
+
+Acquires only an unowned or expired resource. A live owner is never displaced. Successful acquisitions issue a new fencing token.
+
+Use `acquire` for normal static ownership or leader eligibility where an existing live owner must win.
+
+### `renew(lease, ttl)`
+
+Renews only the exact resource/owner/fencing-token tuple. Renewal extends the lease but keeps the same fencing token because ownership has not changed.
+
+A failed or empty renewal means the caller can no longer prove that lease is current. Application code must not infer authority from a locally cached `FencedLease` after renewal fails.
+
+### `release(lease)`
+
+Releases only the exact current resource/owner/fencing-token tuple. A stale owner or stale token cannot release a newer owner's lease.
+
+### `claim(resource, owner, ttl)`
+
+Performs an explicit latest-owner-wins takeover and always issues a newer fencing token. The former owner is immediately stale even if it still has a locally cached lease object.
+
+`claim` is stronger than normal acquisition. Consumers should reserve it for workflows where forced takeover is intentional rather than using it as a retry mechanism for `acquire`.
+
+## Fencing tokens
+
+Every successful new acquisition or claim for one resource advances that resource's fencing generation. Renewals do not.
+
+A typical sequence is:
+
+```text
+process A acquires -> token 41
+process A renews  -> token 41
+process A releases
+process B acquires -> token 42
+```
+
+Consumers that protect state outside Redis should persist/compare the fencing token at their own write boundary. A lower token must never be allowed to overwrite work already accepted under a higher token.
+
+## Fenced Redis values
+
+`writeFenced`, `deleteFenced`, `writeFencedIndexed`, and `deleteFencedIndexed` first verify the exact current lease owner and fencing token atomically in Redis.
+
+After another owner has acquired or claimed a newer generation, an old `FencedLease` cannot mutate those fenced values.
+
+Indexed variants additionally maintain an explicit coordination index. `readIndexedValues` returns live values and prunes members whose TTL value has expired.
+
+## Compare-and-set operations
+
+`compareAndSetWithTtl` atomically writes only when the current value matches the expected value. Passing a null expected value means the key must be absent.
+
+`compareAndDelete` deletes only when the current value exactly matches the supplied expected value.
+
+These operations are independent from lease ownership and are useful for small atomic state transitions that do not require a fencing generation.
+
+## Time and expiry
+
+Lease scripts use Redis `TIME`; the Redis server clock is authoritative for the returned `expiresAt` value. Clients should not manufacture lease expiry timestamps locally.
+
+A cached lease does not guarantee current ownership. Network failure, process suspension, lease expiry, or an explicit `claim` may make it stale. Consumers that depend on continuing authority should renew well before expiry and define a local safety margin for stopping authoritative work when renewal can no longer be proven.
+
+## Connection interruption
+
+Redis reconnect does not revive an expired lease. The lease key may expire while a client is disconnected, while the fencing counter remains. A subsequent acquisition receives a newer token and the former lease remains stale.
+
+DataProvider reports coordination operation success or failure; availability policy belongs to the consumer. For example, an application may continue ordinary read-only/local behavior while disabling singleton work whenever lease renewal cannot be proven.
+
+## Scoped ownership
+
+Long-lived subsystems should acquire their backend through a dedicated `DataProviderScope`:
+
+```java
+DataProviderScope scope = api.scope("component.coordination");
+KeyValueDatabaseProvider redis = scope.registerDatabaseOrThrow(
+        DatabaseType.REDIS,
+        "coordination",
+        KeyValueDatabaseProvider.class
+);
+CoordinationDataAccess coordination = redis.getCoordinationDataAccess();
+```
+
+Closing the scope releases that subsystem's DataProvider registrations without affecting unrelated scopes or plugins.
+
+## What DataProvider deliberately does not decide
+
+DataProvider does not define:
+
+- which nodes are eligible to own a resource;
+- automatic or static leader election policy;
+- replica-group membership;
+- application configuration replication;
+- application fail-open/fail-closed behavior;
+- application-specific cluster tables.
+
+Those policies belong to the consuming application or framework. DataProvider provides the generic atomic primitives they can build on.

docs/README.md

@@ -8,6 +8,7 @@ This directory contains developer and operational notes for DataProvider.
 - [Usage Guide](USAGE_GUIDE.md)
 - [Best Practices](BEST_PRACTICES.md)
 - [Scoped Lifecycle](SCOPED_LIFECYCLE.md)
+- [Distributed Coordination](COORDINATION.md)
 - [Structured Exceptions](EXCEPTIONS.md)
 - [Operation Observation](OBSERVATION.md)
 - [Configuration](CONFIGURATION.md)