diff --git a/README.md b/README.md index 2f0077c..5d7cceb 100644 --- a/README.md +++ b/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,7 +134,7 @@ Maven: nl.hauntedmc.dataprovider dataprovider-api - 3.4.1 + 3.4.2 provided ``` @@ -139,7 +142,7 @@ Maven: 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) diff --git a/dataprovider-core/src/test/java/nl/hauntedmc/dataprovider/core/integration/RedisCoordinationFencingIT.java b/dataprovider-core/src/test/java/nl/hauntedmc/dataprovider/core/integration/RedisCoordinationFencingIT.java new file mode 100644 index 0000000..ab043e8 --- /dev/null +++ b/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. + * + *

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.

+ */ +@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 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(); + } +} diff --git a/docs/COORDINATION.md b/docs/COORDINATION.md new file mode 100644 index 0000000..5607f72 --- /dev/null +++ b/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 +/ +``` + +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. diff --git a/docs/README.md b/docs/README.md index bd9adc3..a717564 100644 --- a/docs/README.md +++ b/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) diff --git a/pom.xml b/pom.xml index a8614df..2b6619a 100644 --- a/pom.xml +++ b/pom.xml @@ -7,7 +7,7 @@ nl.hauntedmc.platform haunted-library-parent - 1.3.0 + 1.5.0 @@ -39,7 +39,7 @@ scm:git:git://github.com/HauntedMC/DataProvider.git scm:git:ssh://git@github.com/HauntedMC/DataProvider.git https://github.com/HauntedMC/DataProvider - v3.4.1 + v3.4.2 @@ -60,7 +60,7 @@ - 3.4.1 + 3.4.2 26.2.build.112-stable @@ -119,7 +119,6 @@ org.jacoco jacoco-maven-plugin - ${jacoco.version} enforce-coverage-minimum @@ -145,7 +144,6 @@ org.basepom.maven duplicate-finder-maven-plugin - ${duplicate.finder.version} reject-duplicate-classes @@ -161,30 +159,21 @@ - - org.apache.maven.plugins - maven-compiler-plugin - ${maven.compiler.plugin.version} - - ${maven.compiler.release} - true - - maven-surefire-plugin - ${maven.surefire.version} @{argLine} -javaagent:${settings.localRepository}/org/mockito/mockito-core/${mockito.version}/mockito-core-${mockito.version}.jar --enable-native-access=ALL-UNNAMED + + + + org.mockito + mockito-core + ${mockito.version} + + - maven-jar-plugin${maven.jar.version} - maven-clean-plugin${maven.clean.version} - maven-resources-plugin${maven.resources.version} - maven-install-plugin${maven.install.version} - maven-deploy-plugin${maven.deploy.version} - maven-site-plugin${maven.site.version} - maven-shade-plugin${maven.shade.version}