diff --git a/README.md b/README.md index fc09a05..75a0811 100644 --- a/README.md +++ b/README.md @@ -367,12 +367,95 @@ new BinaryFormat() // compact, good for production Both implement `DataFormat` and can be used with any storage backend. +## Multi-Server Lifecycle + +Across a fleet of servers a player (or a shared entity such as an island) is authoritative on +exactly one node at a time. The lifecycle API lets a node warm that data into its cache before +use and evict it afterwards, so a later visit to any node is never served a stale value. + +```java +// Warm a player's whole document into this node in a single storage read. +api.load(player); // synchronous +api.loadAsync(player, executor); // CompletableFuture +api.isLoaded(player); // boolean + +// Persist pending changes and drop the player from this node's cache. +api.unload(player); // flush + evict +api.flush(player); // flush without evicting + +// Shared/linked entities have the same lifecycle. +api.loadLink(ISLAND, islandId); +api.unloadLink(ISLAND, islandId); +``` + +This is the primitive a proxy uses to implement "load the player's data on the target server +*before* moving them there": the proxy asks the destination to `load(player)`, waits for the +ack, then connects the player. Because the origin calls `unload(player)` on disconnect, the +destination always starts from fresh storage. + +**Deferred persistence.** By default every write is flushed immediately. Pass `autoPersist = false` +to buffer a whole play session in the cache and write it back once, on `flush`/`unload` +(`shutdown` flushes everything so nothing is lost): + +```java +DataAPI api = new DataAPIImpl(storage, new JsonFormat(), pubSub, /* autoPersist */ false); +``` + +**Cache coherency.** With a distributed event bus, a change made on another node to an entity that +is currently loaded here updates the local view in place, so subscribed fields never go stale +while a player is online. Eviction on `unload` handles the general case. + +## Distributed Locking + +Transactions guard a JVM-local lock by default, which only serialises threads within one process. +Supply a `DistributedLock` to get true cross-node mutual exclusion — required when the same shared +entity can be mutated from more than one server (e.g. a coop bank): + +```java +DistributedLock lock = new RedisDistributedLock(jedisPool); // or InMemoryDistributedLock for one node +DataAPI api = new DataAPIImpl(storage, new JsonFormat(), pubSub, true, lock); + +// Transactions now take a cross-node lock keyed by the entity, in addition to the local monitor. +api.transactionDirect(coopId, COOP, tx -> { tx.update(BANK, b -> b - 1000L); return null; }); + +// The same primitive is available for app-level critical sections: +try (var handle = ((DataAPIImpl) api).lock("coop-transfer:" + coopId, Duration.ofSeconds(5))) { + // ... multi-entity critical section ... +} +``` + +The Redis implementation uses `SET NX PX` with a compare-and-delete release, so a lock is only +released by its owner and a lease bounds a crashed holder. + +## Indexed Leaderboards + +Leaderboards are index-backed and **self-registering — no setup for numeric fields**. The first time +you rank a field its index is built from existing data in one scan; from then on every node maintains +it on write, and ranking reads only the requested slice. This requires a `LeaderboardIndex`-capable +storage (`RedisDataStorage`, backed by sorted sets, or `InMemoryDataStorage` for single-node/tests): + +```java +api.getTop(COINS, 10); // just works — O(log N + page), no registration +api.getTopPaged(COINS, 1, 50); +``` + +Only a **non-numeric** field needs a score function, since a sorted set ranks by a number: + +```java +api.trackLeaderboard(NAME, String::length); // rank players by name length +``` + +`rebuildLeaderboard(field)` forces a rebuild from stored data if you ever need it. Storage backends +that don't maintain an index (e.g. `FileDataStorage`) throw on ranking rather than silently scanning; +`getTop(field, limit, comparator)` remains as the explicit scan-based escape hatch for ad-hoc custom +orderings. + ## Lifecycle Always shut down the API when done: ```java -api.shutdown(); // stops expiration timers, closes Pub/Sub subscribers +api.shutdown(); // flushes deferred writes, stops expiration timers, closes Pub/Sub subscribers ``` For Redis storage, also close the storage: diff --git a/src/main/java/net/swofty/DataAPI.java b/src/main/java/net/swofty/DataAPI.java index 83650f5..e855101 100644 --- a/src/main/java/net/swofty/DataAPI.java +++ b/src/main/java/net/swofty/DataAPI.java @@ -6,7 +6,10 @@ import java.time.Duration; import java.util.*; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executor; import java.util.function.Predicate; +import java.util.function.ToDoubleFunction; import java.util.function.UnaryOperator; public interface DataAPI { @@ -69,4 +72,28 @@ public interface DataAPI { // Bulk operations - Linked > List> getTopLinked(LinkedField field, int limit); List queryLinked(LinkedField field, Predicate filter); + + // Leaderboard indexing - getTop/getTopPaged are index-backed and require a LeaderboardIndex- + // capable storage (e.g. Redis sorted sets). No registration is needed for numeric fields: the + // index self-builds on first rank and every node maintains it on write. trackLeaderboard only + // registers a score function so a NON-numeric field can be ranked; rebuildLeaderboard forces a + // rebuild from stored data. + void trackLeaderboard(PlayerField field, ToDoubleFunction scorer); + void rebuildLeaderboard(PlayerField field); + + // Lifecycle - warm a player's data into this node before use, evict it when done. + // This is the primitive a proxy uses to load a player's data on the target server + // BEFORE moving them there, and to evict it afterwards so a later visit is never stale. + void load(UUID player); + CompletableFuture loadAsync(UUID player, Executor executor); + void flush(UUID player); + void unload(UUID player); + boolean isLoaded(UUID player); + Set loadedPlayers(); + + // Lifecycle - shared/linked entities (e.g. an island or coop shared across members) + void loadLink(LinkType type, K key); + void flushLink(LinkType type, K key); + void unloadLink(LinkType type, K key); + boolean isLinkLoaded(LinkType type, K key); } diff --git a/src/main/java/net/swofty/api/BulkOperationExecutor.java b/src/main/java/net/swofty/api/BulkOperationExecutor.java index 2fe40ce..9c1c6db 100644 --- a/src/main/java/net/swofty/api/BulkOperationExecutor.java +++ b/src/main/java/net/swofty/api/BulkOperationExecutor.java @@ -3,12 +3,13 @@ import net.swofty.*; import net.swofty.event.EventBus; import net.swofty.storage.DataStorage; +import net.swofty.storage.LeaderboardIndex; import java.util.*; import java.util.function.Predicate; import java.util.function.UnaryOperator; -public class BulkOperationExecutor { +class BulkOperationExecutor { private final PlayerDataManager playerData; private final LinkedDataManager linkedData; private final DataStorage storage; @@ -23,7 +24,31 @@ public BulkOperationExecutor(PlayerDataManager playerData, LinkedDataManager lin } public > List> getTop(PlayerField field, int limit) { - return getTop(field, limit, Comparator.reverseOrder()); + LeaderboardIndex index = requireIndex(); + playerData.ensureLeaderboardBuilt(field); // self-registers on first use, no-op thereafter + return fromIndex(field, index, 0, limit - 1); + } + + // A leaderboard is always index-backed. There is no silent full-table scan: the index is built + // once on first use and maintained on every write, so ranking is O(log N + page). + private LeaderboardIndex requireIndex() { + if (!(storage instanceof LeaderboardIndex index)) { + throw new IllegalStateException("Storage " + storage.getClass().getSimpleName() + + " does not support leaderboards; use a LeaderboardIndex-capable storage (e.g. RedisDataStorage)"); + } + return index; + } + + private List> fromIndex(PlayerField field, LeaderboardIndex index, + int start, int endInclusive) { + List range = index.scoreRange(field.fullKey(), start, endInclusive, true); + List> result = new ArrayList<>(range.size()); + int rank = start + 1; + for (LeaderboardIndex.ScoreEntry entry : range) { + UUID id = UUID.fromString(entry.id()); + result.add(new LeaderboardEntry<>(id, playerData.getFieldValue(id, field), rank++)); + } + return result; } public List> getTop(PlayerField field, int limit, Comparator comparator) { @@ -38,19 +63,12 @@ public List> getTop(PlayerField field, int limit, Com } public > Page> getTopPaged(PlayerField field, int page, int pageSize) { - List> entries = getAllPlayerValues(field); - entries.sort((a, b) -> b.getValue().compareTo(a.getValue())); - - long total = entries.size(); + LeaderboardIndex index = requireIndex(); + playerData.ensureLeaderboardBuilt(field); + long total = index.leaderboardSize(field.fullKey()); int totalPages = (int) Math.ceil((double) total / pageSize); int start = (page - 1) * pageSize; - int end = Math.min(start + pageSize, entries.size()); - - List> content = new ArrayList<>(); - for (int i = start; i < end; i++) { - Map.Entry e = entries.get(i); - content.add(new LeaderboardEntry<>(e.getKey(), e.getValue(), i + 1)); - } + List> content = fromIndex(field, index, start, start + pageSize - 1); return new Page<>(content, page, totalPages, total); } diff --git a/src/main/java/net/swofty/api/DataAPIImpl.java b/src/main/java/net/swofty/api/DataAPIImpl.java index a83ecd8..571d085 100644 --- a/src/main/java/net/swofty/api/DataAPIImpl.java +++ b/src/main/java/net/swofty/api/DataAPIImpl.java @@ -4,13 +4,17 @@ import net.swofty.data.DataFormat; import net.swofty.data.format.JsonFormat; import net.swofty.event.*; +import net.swofty.lock.DistributedLock; import net.swofty.storage.DataStorage; import net.swofty.transaction.TransactionConsumer; import net.swofty.transaction.TransactionFunction; import java.time.Duration; import java.util.*; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executor; import java.util.function.Predicate; +import java.util.function.ToDoubleFunction; import java.util.function.UnaryOperator; public class DataAPIImpl implements DataAPI { @@ -22,16 +26,50 @@ public class DataAPIImpl implements DataAPI { private final TransactionManager transactionManager; private final EventBus eventBus; private final BulkOperationExecutor bulkOperations; + private final DistributedLock distributedLock; public DataAPIImpl(DataStorage storage, DataFormat format, PubSubHandler pubSub) { + this(storage, format, pubSub, true); + } + + public DataAPIImpl(DataStorage storage, DataFormat format, PubSubHandler pubSub, boolean autoPersist) { + this(storage, format, pubSub, autoPersist, null); + } + + /** + * @param autoPersist when true (the default) every write is flushed to storage immediately; + * when false writes stay in the cache until {@link #flush(UUID)} or + * {@link #unload(UUID)}, letting a node batch a play session into one write. + * @param distributedLock when non-null, transactions and {@link #lock(String, Duration)} use it for + * cross-node mutual exclusion; when null they fall back to a JVM-local lock. + */ + public DataAPIImpl(DataStorage storage, DataFormat format, PubSubHandler pubSub, boolean autoPersist, + DistributedLock distributedLock) { this.storage = storage; + this.distributedLock = distributedLock; this.eventBus = (pubSub != null) ? new DistributedEventBus(pubSub) : new EventBus(); this.linkRegistry = new LinkRegistryImpl(); - this.playerData = new PlayerDataManager(storage, format, eventBus); - this.linkedData = new LinkedDataManager(storage, format, eventBus, linkRegistry); + this.playerData = new PlayerDataManager(storage, format, eventBus, autoPersist); + this.linkedData = new LinkedDataManager(storage, format, eventBus, linkRegistry, autoPersist); this.expirationManager = new ExpirationManager(); - this.transactionManager = new TransactionManager(playerData, linkedData, linkRegistry); + this.transactionManager = new TransactionManager(playerData, linkedData, linkRegistry, + distributedLock, Duration.ofSeconds(10)); this.bulkOperations = new BulkOperationExecutor(playerData, linkedData, storage, eventBus); + + // Keep locally cached containers coherent with changes made on other nodes. + if (eventBus instanceof DistributedEventBus distributed) { + distributed.setRemoteChangeHandler(new RemoteChangeHandler() { + @Override + public void onPlayerChange(DataField field, UUID player, T newValue) { + playerData.applyRemote(field, player, newValue); + } + + @Override + public void onLinkedChange(DataField field, String linkTypeName, String linkKey, T newValue) { + linkedData.applyRemote(linkTypeName, linkKey, field, newValue); + } + }); + } } public DataAPIImpl(DataStorage storage, DataFormat format) { @@ -267,9 +305,84 @@ public List queryLinked(LinkedField field, Predicate filter) return bulkOperations.queryLinked(field, filter); } + @Override + public void trackLeaderboard(PlayerField field, ToDoubleFunction scorer) { + playerData.trackLeaderboard(field, scorer); + } + + @Override + public void rebuildLeaderboard(PlayerField field) { + playerData.rebuildLeaderboard(field); + } + // ==================== Lifecycle ==================== + @Override + public void load(UUID player) { + playerData.load(player); + } + + @Override + public CompletableFuture loadAsync(UUID player, Executor executor) { + return CompletableFuture.runAsync(() -> playerData.load(player), executor); + } + + @Override + public void flush(UUID player) { + playerData.flush(player); + } + + @Override + public void unload(UUID player) { + playerData.unload(player); + } + + @Override + public boolean isLoaded(UUID player) { + return playerData.isLoaded(player); + } + + @Override + public Set loadedPlayers() { + return playerData.loadedPlayers(); + } + + @Override + public void loadLink(LinkType type, K key) { + linkedData.loadLinked(type.name(), key); + } + + @Override + public void flushLink(LinkType type, K key) { + linkedData.flushLinked(type.name(), key); + } + + @Override + public void unloadLink(LinkType type, K key) { + linkedData.unloadLinked(type.name(), key); + } + + @Override + public boolean isLinkLoaded(LinkType type, K key) { + return linkedData.isLinkedLoaded(type.name(), key); + } + + /** + * Acquires the configured {@link DistributedLock} for an arbitrary key, for app-level critical + * sections that span more than one field or entity (e.g. transferring coins between two coops). + * Use with try-with-resources. Requires a distributed lock to have been supplied. + */ + public DistributedLock.Handle lock(String key, Duration timeout) { + if (distributedLock == null) { + throw new IllegalStateException("No DistributedLock configured on this DataAPI"); + } + return distributedLock.acquire(key, timeout); + } + public void shutdown() { + // Flush any deferred writes so nothing is lost on a clean shutdown. + playerData.flushAll(); + linkedData.flushAll(); expirationManager.shutdown(); if (eventBus instanceof DistributedEventBus deb) { deb.shutdown(); diff --git a/src/main/java/net/swofty/api/DataContainer.java b/src/main/java/net/swofty/api/DataContainer.java index 586b75b..62f0226 100644 --- a/src/main/java/net/swofty/api/DataContainer.java +++ b/src/main/java/net/swofty/api/DataContainer.java @@ -3,15 +3,34 @@ import net.swofty.DataField; import net.swofty.data.DataFormat; import net.swofty.data.DataReader; -import net.swofty.data.DataWriter; -import net.swofty.data.format.JsonDataWriter; +import java.util.LinkedHashMap; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; -public class DataContainer { +/** + * The live, in-memory view of one entity's data on this node. + * + *

Only the fields actually read or written this session are materialised in + * {@link #data}. To avoid clobbering fields that were never touched, the container + * also retains the full {@link #backingDocument} it last saw in storage and, on + * {@link #serialize(DataFormat)}, merges the touched fields over it. This is what + * makes a partial write safe: previously an untouched field would be silently + * dropped the first time any other field was persisted. + */ +class DataContainer { private final ConcurrentHashMap data = new ConcurrentHashMap<>(); + // Fields explicitly cleared (set to null) this session. Suppressed from the merged + // output so a null-set actually deletes a field that still exists in the backing document. + private final Set tombstones = ConcurrentHashMap.newKeySet(); + + // The full serialized document as last seen in storage (null = no stored document). + private volatile byte[] backingDocument; + private volatile boolean documentLoaded; + private volatile boolean dirty; + @SuppressWarnings("unchecked") public T get(DataField field) { Object value = data.get(field.fullKey()); @@ -21,35 +40,82 @@ public T get(DataField field) { public void set(DataField field, T value) { if (value == null) { data.remove(field.fullKey()); + tombstones.add(field.fullKey()); } else { data.put(field.fullKey(), value); + tombstones.remove(field.fullKey()); } + dirty = true; } public boolean has(String fullKey) { return data.containsKey(fullKey); } - public void loadField(DataField field, DataFormat format, byte[] raw) { - if (raw == null || data.containsKey(field.fullKey())) return; - DataReader reader = format.createReader(raw); + // ---- Document lifecycle ------------------------------------------------- + + /** Records the full backing document so later partial writes do not drop untouched fields. */ + public void loadDocument(DataFormat format, byte[] raw) { + this.backingDocument = raw; + this.documentLoaded = true; + } + + public boolean isDocumentLoaded() { + return documentLoaded; + } + + public boolean isDirty() { + return dirty; + } + + /** Lazily deserialises a single field out of the backing document into the live view. */ + public void ensureField(DataField field, DataFormat format) { + if (data.containsKey(field.fullKey()) || tombstones.contains(field.fullKey())) return; + if (backingDocument == null) return; + DataReader reader = format.createReader(backingDocument); if (reader.hasKey(field.fullKey())) { - DataReader section = reader.readSection(field.fullKey()); - Object value = field.codec().read(section); + Object value = field.codec().read(reader.readSection(field.fullKey())); if (value != null) { data.put(field.fullKey(), value); } } } + /** Back-compat entry point: warm the document (if needed) then pull a single field out of it. */ + public void loadField(DataField field, DataFormat format, byte[] raw) { + if (!documentLoaded) loadDocument(format, raw); + ensureField(field, format); + } + + /** Merges the touched fields over the backing document so nothing untouched is lost. */ public byte[] serialize(DataFormat format) { - DataWriter writer = format.createWriter(); - if (writer instanceof JsonDataWriter jsonWriter) { - for (Map.Entry entry : data.entrySet()) { - jsonWriter.getData().put(entry.getKey(), entry.getValue()); - } + Map merged = new LinkedHashMap<>(); + if (backingDocument != null) { + merged.putAll(format.readRaw(backingDocument)); + } + merged.putAll(data); + for (String tombstone : tombstones) { + merged.remove(tombstone); } - return format.toBytes(writer); + return format.writeRaw(merged); + } + + /** + * Applies a value that another node has already persisted. Updates the live view so + * local reads are fresh, but preserves the prior dirty state so a clean (e.g. read-only) + * container does not get marked dirty and re-persist stale sibling fields on unload. + */ + public void applyRemote(DataField field, T value) { + boolean wasDirty = this.dirty; + set(field, value); + this.dirty = wasDirty; + } + + /** Records the bytes just written to storage as the new backing document and clears the dirty flag. */ + public void markPersisted(byte[] bytes) { + this.backingDocument = bytes; + this.documentLoaded = true; + this.dirty = false; } ConcurrentHashMap rawData() { diff --git a/src/main/java/net/swofty/api/ExpirationManager.java b/src/main/java/net/swofty/api/ExpirationManager.java index 98763e8..dba0076 100644 --- a/src/main/java/net/swofty/api/ExpirationManager.java +++ b/src/main/java/net/swofty/api/ExpirationManager.java @@ -12,7 +12,7 @@ import java.util.UUID; import java.util.concurrent.*; -public class ExpirationManager { +class ExpirationManager { private final ConcurrentHashMap expirations = new ConcurrentHashMap<>(); private final ScheduledExecutorService scheduler; diff --git a/src/main/java/net/swofty/api/LinkRegistryImpl.java b/src/main/java/net/swofty/api/LinkRegistryImpl.java index dabb92f..97c7f24 100644 --- a/src/main/java/net/swofty/api/LinkRegistryImpl.java +++ b/src/main/java/net/swofty/api/LinkRegistryImpl.java @@ -5,7 +5,7 @@ import java.util.*; import java.util.concurrent.ConcurrentHashMap; -public class LinkRegistryImpl { +class LinkRegistryImpl { private final ConcurrentHashMap> playerLinks = new ConcurrentHashMap<>(); private final ConcurrentHashMap> reverseIndex = new ConcurrentHashMap<>(); diff --git a/src/main/java/net/swofty/api/LinkedDataManager.java b/src/main/java/net/swofty/api/LinkedDataManager.java index 95b04f5..9e00662 100644 --- a/src/main/java/net/swofty/api/LinkedDataManager.java +++ b/src/main/java/net/swofty/api/LinkedDataManager.java @@ -11,19 +11,26 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.function.UnaryOperator; -public class LinkedDataManager { +class LinkedDataManager { private final DataStorage storage; private final DataFormat format; private final EventBus eventBus; private final LinkRegistryImpl linkRegistry; private final ConcurrentHashMap cache = new ConcurrentHashMap<>(); private final ConcurrentHashMap locks = new ConcurrentHashMap<>(); + private final boolean autoPersist; public LinkedDataManager(DataStorage storage, DataFormat format, EventBus eventBus, LinkRegistryImpl linkRegistry) { + this(storage, format, eventBus, linkRegistry, true); + } + + public LinkedDataManager(DataStorage storage, DataFormat format, EventBus eventBus, + LinkRegistryImpl linkRegistry, boolean autoPersist) { this.storage = storage; this.format = format; this.eventBus = eventBus; this.linkRegistry = linkRegistry; + this.autoPersist = autoPersist; } Object getLock(String compositeKey) { @@ -107,8 +114,8 @@ T getFieldValue(String linkTypeName, Object key, DataField field) { String ck = compositeKey(linkTypeName, key); DataContainer container = getContainer(ck); if (!container.has(field.fullKey())) { - byte[] raw = storage.load("linked/" + linkTypeName, key.toString()); - container.loadField(field, format, raw); + ensureDocumentLoaded(linkTypeName, key, container); + container.ensureField(field, format); } return container.get(field); } @@ -116,9 +123,97 @@ T getFieldValue(String linkTypeName, Object key, DataField field) { void setFieldValue(String linkTypeName, Object key, DataField field, T value) { String ck = compositeKey(linkTypeName, key); DataContainer container = getContainer(ck); + ensureDocumentLoaded(linkTypeName, key, container); container.set(field, value); + if (autoPersist) { + persistLinked(linkTypeName, key, container); + } + } + + private void ensureDocumentLoaded(String linkTypeName, Object key, DataContainer container) { + if (!container.isDocumentLoaded()) { + container.loadDocument(format, storage.load("linked/" + linkTypeName, key.toString())); + } + } + + private void persistLinked(String linkTypeName, String keyString, DataContainer container) { byte[] bytes = container.serialize(format); - storage.save("linked/" + linkTypeName, key.toString(), bytes); + storage.save("linked/" + linkTypeName, keyString, bytes); + container.markPersisted(bytes); + } + + private void persistLinked(String linkTypeName, Object key, DataContainer container) { + persistLinked(linkTypeName, key.toString(), container); + } + + // ---- Lifecycle ---------------------------------------------------------- + + /** Warms a shared entity's whole document into this node's cache in a single storage read. */ + public void loadLinked(String linkTypeName, Object key) { + String ck = compositeKey(linkTypeName, key); + synchronized (getLock(ck)) { + DataContainer container = getContainer(ck); + if (!container.isDocumentLoaded()) { + container.loadDocument(format, storage.load("linked/" + linkTypeName, key.toString())); + } + } + } + + public void flushLinked(String linkTypeName, Object key) { + String ck = compositeKey(linkTypeName, key); + synchronized (getLock(ck)) { + DataContainer container = cache.get(ck); + if (container != null && container.isDirty()) { + persistLinked(linkTypeName, key, container); + } + } + } + + public void unloadLinked(String linkTypeName, Object key) { + String ck = compositeKey(linkTypeName, key); + synchronized (getLock(ck)) { + DataContainer container = cache.get(ck); + if (container != null && container.isDirty()) { + persistLinked(linkTypeName, key, container); + } + cache.remove(ck); + } + locks.remove(ck); + } + + public boolean isLinkedLoaded(String linkTypeName, Object key) { + return cache.containsKey(compositeKey(linkTypeName, key)); + } + + /** Flushes every cached shared entity. Used on shutdown so deferred writes are not lost. */ + public void flushAll() { + for (String ck : cache.keySet()) { + int colon = ck.indexOf(':'); + if (colon < 0) continue; + String linkTypeName = ck.substring(0, colon); + String keyString = ck.substring(colon + 1); + synchronized (getLock(ck)) { + DataContainer container = cache.get(ck); + if (container != null && container.isDirty()) { + persistLinked(linkTypeName, keyString, container); + } + } + } + } + + /** + * Applies a change that originated on another node to a locally cached shared entity, + * without re-persisting or re-firing events. Only touches entities currently loaded here. + */ + void applyRemote(String linkTypeName, Object key, DataField field, T newValue) { + String ck = compositeKey(linkTypeName, key); + DataContainer container = cache.get(ck); + if (container == null) return; + synchronized (getLock(ck)) { + container = cache.get(ck); + if (container == null) return; + container.applyRemote(field, newValue); + } } public List listLinkedIds(String linkTypeName) { diff --git a/src/main/java/net/swofty/api/PlayerDataManager.java b/src/main/java/net/swofty/api/PlayerDataManager.java index 2033bea..ce4cc3d 100644 --- a/src/main/java/net/swofty/api/PlayerDataManager.java +++ b/src/main/java/net/swofty/api/PlayerDataManager.java @@ -6,22 +6,34 @@ import net.swofty.data.DataFormat; import net.swofty.event.EventBus; import net.swofty.storage.DataStorage; +import net.swofty.storage.LeaderboardIndex; import java.util.*; import java.util.concurrent.ConcurrentHashMap; +import java.util.function.ToDoubleFunction; import java.util.function.UnaryOperator; -public class PlayerDataManager { +// Internal to net.swofty.api — reach it through DataAPI / DataAPIImpl, not directly. +class PlayerDataManager { private final DataStorage storage; private final DataFormat format; private final EventBus eventBus; private final ConcurrentHashMap cache = new ConcurrentHashMap<>(); private final ConcurrentHashMap locks = new ConcurrentHashMap<>(); + // Optional custom score functions, keyed by field. Numeric fields need no entry here — they + // are scored automatically — so leaderboards require no registration in the common case. + private final ConcurrentHashMap> scorers = new ConcurrentHashMap<>(); + private final boolean autoPersist; public PlayerDataManager(DataStorage storage, DataFormat format, EventBus eventBus) { + this(storage, format, eventBus, true); + } + + public PlayerDataManager(DataStorage storage, DataFormat format, EventBus eventBus, boolean autoPersist) { this.storage = storage; this.format = format; this.eventBus = eventBus; + this.autoPersist = autoPersist; } public Object getLock(UUID player) { @@ -64,16 +76,27 @@ public void update(UUID player, PlayerField field, UnaryOperator updat T getFieldValue(UUID player, DataField field) { DataContainer container = getContainer(player); if (!container.has(field.fullKey())) { - byte[] raw = storage.load("players", player.toString()); - container.loadField(field, format, raw); + ensureDocumentLoaded(player, container); + container.ensureField(field, format); } return container.get(field); } void setFieldValue(UUID player, DataField field, T value) { DataContainer container = getContainer(player); + // Warm the backing document first so serialize() merges over it and never + // drops fields that were never read this session. + ensureDocumentLoaded(player, container); container.set(field, value); - persist(player); + if (autoPersist) { + persist(player); + } + } + + private void ensureDocumentLoaded(UUID player, DataContainer container) { + if (!container.isDocumentLoaded()) { + container.loadDocument(format, storage.load("players", player.toString())); + } } void persist(UUID player) { @@ -81,6 +104,143 @@ void persist(UUID player) { if (container != null) { byte[] bytes = container.serialize(format); storage.save("players", player.toString(), bytes); + container.markPersisted(bytes); + updateLeaderboards(player, container); + } + } + + // ---- Leaderboard indexing ---------------------------------------------- + // + // Leaderboards need no registration. The first time a field is ranked, its index is built by + // a one-time scan (ensureLeaderboardBuilt); from then on the index EXISTS in shared storage, + // and every node maintains it on write via updateScoreIfPresent. A field that is never ranked + // has no index, so its writes cost nothing. Numeric fields are scored automatically; a custom + // scorer is only needed to rank a non-numeric field. + + private LeaderboardIndex leaderboardIndex() { + return storage instanceof LeaderboardIndex index ? index : null; + } + + private LeaderboardIndex requireLeaderboardIndex() { + LeaderboardIndex index = leaderboardIndex(); + if (index == null) { + throw new IllegalStateException("Storage " + storage.getClass().getSimpleName() + + " does not support leaderboards; use a LeaderboardIndex-capable storage" + + " (e.g. RedisDataStorage or InMemoryDataStorage)"); + } + return index; + } + + /** Optional: register a score function so a non-numeric field can be ranked. */ + public void trackLeaderboard(PlayerField field, ToDoubleFunction scorer) { + requireLeaderboardIndex(); + scorers.put(field.fullKey(), scorer); + } + + @SuppressWarnings("unchecked") + private Double scoreOf(String fullKey, Object value) { + if (value == null) return null; + ToDoubleFunction scorer = (ToDoubleFunction) scorers.get(fullKey); + if (scorer != null) return scorer.applyAsDouble(value); + if (value instanceof Number number) return number.doubleValue(); + return null; // not rankable without a scorer + } + + // Maintains only leaderboards that already exist, so unranked fields cost nothing. + private void updateLeaderboards(UUID player, DataContainer container) { + LeaderboardIndex index = leaderboardIndex(); + if (index == null) return; + for (Map.Entry entry : container.rawData().entrySet()) { + Double score = scoreOf(entry.getKey(), entry.getValue()); + if (score != null) { + index.updateScoreIfPresent(entry.getKey(), player.toString(), score); + } + } + } + + /** Builds the index on first use by scanning existing players once; a no-op once it exists. */ + void ensureLeaderboardBuilt(PlayerField field) { + LeaderboardIndex index = requireLeaderboardIndex(); + if (index.leaderboardExists(field.fullKey())) return; + rebuildLeaderboard(field); + } + + /** Rebuilds a field's index from stored data. Called automatically on first rank; also public. */ + public void rebuildLeaderboard(PlayerField field) { + LeaderboardIndex index = requireLeaderboardIndex(); + for (String id : storage.listIds("players")) { + UUID player = UUID.fromString(id); + T value = getFieldValue(player, field); + Double score = scoreOf(field.fullKey(), value); + if (score == null) { + throw new IllegalStateException("Leaderboard field '" + field.fullKey() + + "' is not numeric; register a score function with trackLeaderboard(field, scorer)"); + } + index.updateScore(field.fullKey(), id, score); + } + } + + // ---- Lifecycle ---------------------------------------------------------- + + /** Warms the player's whole document into this node's cache in a single storage read. */ + public void load(UUID player) { + synchronized (getLock(player)) { + DataContainer container = getContainer(player); + if (!container.isDocumentLoaded()) { + container.loadDocument(format, storage.load("players", player.toString())); + } + } + } + + /** Persists pending changes for a player if the cache holds unsaved edits. */ + public void flush(UUID player) { + synchronized (getLock(player)) { + DataContainer container = cache.get(player); + if (container != null && container.isDirty()) { + persist(player); + } + } + } + + /** Flushes pending changes then evicts the player from this node's cache. */ + public void unload(UUID player) { + synchronized (getLock(player)) { + DataContainer container = cache.get(player); + if (container != null && container.isDirty()) { + persist(player); + } + cache.remove(player); + } + locks.remove(player); + } + + public boolean isLoaded(UUID player) { + return cache.containsKey(player); + } + + public Set loadedPlayers() { + return new HashSet<>(cache.keySet()); + } + + /** Flushes every cached player. Used on shutdown so deferred writes are not lost. */ + public void flushAll() { + for (UUID player : cache.keySet()) { + flush(player); + } + } + + /** + * Applies a change that originated on another node to the locally cached container, + * without re-persisting or re-firing events. Only touches players that are currently + * loaded here, so it never resurrects an evicted or never-loaded entity. + */ + void applyRemote(DataField field, UUID player, T newValue) { + DataContainer container = cache.get(player); + if (container == null) return; + synchronized (getLock(player)) { + container = cache.get(player); + if (container == null) return; + container.applyRemote(field, newValue); } } diff --git a/src/main/java/net/swofty/api/TransactionManager.java b/src/main/java/net/swofty/api/TransactionManager.java index 663b310..af27ab0 100644 --- a/src/main/java/net/swofty/api/TransactionManager.java +++ b/src/main/java/net/swofty/api/TransactionManager.java @@ -3,71 +3,96 @@ import net.swofty.LinkedField; import net.swofty.PlayerField; import net.swofty.LinkType; +import net.swofty.lock.DistributedLock; import net.swofty.transaction.*; +import java.time.Duration; import java.util.HashMap; import java.util.Map; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.function.UnaryOperator; -public class TransactionManager { +class TransactionManager { + private static final DistributedLock.Handle NO_OP = () -> {}; + private final PlayerDataManager playerData; private final LinkedDataManager linkedData; private final LinkRegistryImpl linkRegistry; + private final DistributedLock distributedLock; + private final Duration lockTimeout; public TransactionManager(PlayerDataManager playerData, LinkedDataManager linkedData, LinkRegistryImpl linkRegistry) { + this(playerData, linkedData, linkRegistry, null, Duration.ofSeconds(10)); + } + + public TransactionManager(PlayerDataManager playerData, LinkedDataManager linkedData, LinkRegistryImpl linkRegistry, + DistributedLock distributedLock, Duration lockTimeout) { this.playerData = playerData; this.linkedData = linkedData; this.linkRegistry = linkRegistry; + this.distributedLock = distributedLock; + this.lockTimeout = lockTimeout; + } + + // Cross-node mutual exclusion when a distributed lock is configured; a no-op handle + // otherwise, leaving the JVM-local monitor as the only guard (single-node behaviour). + private DistributedLock.Handle acquire(String key) { + return distributedLock == null ? NO_OP : distributedLock.acquire(key, lockTimeout); } public R execute(UUID player, TransactionFunction action) { - synchronized (playerData.getLock(player)) { - TransactionContext tx = new TransactionContext(player); - try { - R result = action.apply(tx); - tx.commit(); - return result; - } catch (TransactionAbortException e) { - tx.rollback(); - return null; - } catch (Exception e) { - tx.rollback(); - throw e; + try (DistributedLock.Handle ignored = acquire("player:" + player)) { + synchronized (playerData.getLock(player)) { + TransactionContext tx = new TransactionContext(player); + try { + R result = action.apply(tx); + tx.commit(); + return result; + } catch (TransactionAbortException e) { + tx.rollback(); + return null; + } catch (Exception e) { + tx.rollback(); + throw e; + } } } } public void execute(UUID player, TransactionConsumer action) { - synchronized (playerData.getLock(player)) { - TransactionContext tx = new TransactionContext(player); - try { - action.accept(tx); - tx.commit(); - } catch (TransactionAbortException e) { - tx.rollback(); - } catch (Exception e) { - tx.rollback(); - throw e; + try (DistributedLock.Handle ignored = acquire("player:" + player)) { + synchronized (playerData.getLock(player)) { + TransactionContext tx = new TransactionContext(player); + try { + action.accept(tx); + tx.commit(); + } catch (TransactionAbortException e) { + tx.rollback(); + } catch (Exception e) { + tx.rollback(); + throw e; + } } } } public R executeDirect(K key, LinkType type, TransactionFunction action) { String ck = LinkedDataManager.compositeKey(type.name(), key); - synchronized (linkedData.getLock(ck)) { - TransactionContext tx = new TransactionContext(null); - try { - R result = action.apply(tx); - tx.commit(); - return result; - } catch (TransactionAbortException e) { - tx.rollback(); - return null; - } catch (Exception e) { - tx.rollback(); - throw e; + try (DistributedLock.Handle ignored = acquire("linked:" + ck)) { + synchronized (linkedData.getLock(ck)) { + TransactionContext tx = new TransactionContext(null); + try { + R result = action.apply(tx); + tx.commit(); + return result; + } catch (TransactionAbortException e) { + tx.rollback(); + return null; + } catch (Exception e) { + tx.rollback(); + throw e; + } } } } diff --git a/src/main/java/net/swofty/api/Validation.java b/src/main/java/net/swofty/api/Validation.java index a112cd7..a69d303 100644 --- a/src/main/java/net/swofty/api/Validation.java +++ b/src/main/java/net/swofty/api/Validation.java @@ -7,7 +7,7 @@ import net.swofty.validation.ValidationResult; import net.swofty.validation.Validator; -public final class Validation { +final class Validation { private Validation() {} @SuppressWarnings("unchecked") diff --git a/src/main/java/net/swofty/data/DataFormat.java b/src/main/java/net/swofty/data/DataFormat.java index c882d3a..c6d6fa6 100644 --- a/src/main/java/net/swofty/data/DataFormat.java +++ b/src/main/java/net/swofty/data/DataFormat.java @@ -1,7 +1,27 @@ package net.swofty.data; +import java.util.Map; + public interface DataFormat { DataReader createReader(byte[] data); DataWriter createWriter(); byte[] toBytes(DataWriter writer); + + /** + * Parses a stored document into its top-level {@code fullKey -> value} map. + * Used to merge a partial in-memory view back over the full stored document so + * fields that were never read this session are not dropped on save. Formats that + * are not keyed (e.g. a purely sequential binary format) may leave this + * unsupported, in which case they cannot back multi-field container storage. + */ + default Map readRaw(byte[] data) { + throw new UnsupportedOperationException( + getClass().getSimpleName() + " does not support whole-document (readRaw) access"); + } + + /** Serializes a top-level {@code fullKey -> value} map back into stored bytes. */ + default byte[] writeRaw(Map data) { + throw new UnsupportedOperationException( + getClass().getSimpleName() + " does not support whole-document (writeRaw) access"); + } } diff --git a/src/main/java/net/swofty/data/format/JsonFormat.java b/src/main/java/net/swofty/data/format/JsonFormat.java index 267b2ea..3b8a940 100644 --- a/src/main/java/net/swofty/data/format/JsonFormat.java +++ b/src/main/java/net/swofty/data/format/JsonFormat.java @@ -39,4 +39,18 @@ public byte[] toBytes(DataWriter writer) { String json = GSON.toJson(jsonWriter.getData()); return json.getBytes(StandardCharsets.UTF_8); } + + @Override + public Map readRaw(byte[] data) { + if (data == null || data.length == 0) { + return new LinkedHashMap<>(); + } + Map map = GSON.fromJson(new String(data, StandardCharsets.UTF_8), MAP_TYPE); + return map == null ? new LinkedHashMap<>() : map; + } + + @Override + public byte[] writeRaw(Map data) { + return GSON.toJson(data).getBytes(StandardCharsets.UTF_8); + } } diff --git a/src/main/java/net/swofty/event/DistributedEventBus.java b/src/main/java/net/swofty/event/DistributedEventBus.java index 18fcd4a..ad1b317 100644 --- a/src/main/java/net/swofty/event/DistributedEventBus.java +++ b/src/main/java/net/swofty/event/DistributedEventBus.java @@ -7,6 +7,7 @@ import net.swofty.ExpiringField; import net.swofty.ExpiringLinkedField; import net.swofty.LinkType; +import net.swofty.LinkedField; import net.swofty.codec.Codec; import net.swofty.data.DataReader; import net.swofty.data.DataWriter; @@ -27,6 +28,13 @@ public class DistributedEventBus extends EventBus { private final ConcurrentHashMap> fieldRegistry = new ConcurrentHashMap<>(); private final ConcurrentHashMap> linkTypeRegistry = new ConcurrentHashMap<>(); + private volatile RemoteChangeHandler remoteChangeHandler; + + /** Registers the cache-coherency hook. Called by the API implementation after construction. */ + public void setRemoteChangeHandler(RemoteChangeHandler handler) { + this.remoteChangeHandler = handler; + } + public DistributedEventBus(PubSubHandler pubSubHandler) { this(pubSubHandler, UUID.randomUUID().toString()); } @@ -162,6 +170,10 @@ private void handlePlayerDataChanged(EventMessage msg) { UUID player = UUID.fromString((String) msg.data.get("player")); Object oldValue = deserializeValue(field.codec(), msg.data.get("oldValue")); Object newValue = deserializeValue(field.codec(), msg.data.get("newValue")); + RemoteChangeHandler handler = remoteChangeHandler; + if (handler != null) { + handler.onPlayerChange(field, player, newValue); + } super.firePlayerDataChanged(field, player, oldValue, newValue); } @@ -173,6 +185,10 @@ private void handleLinkedDataChanged(EventMessage msg) { Object oldValue = deserializeValue(field.codec(), msg.data.get("oldValue")); Object newValue = deserializeValue(field.codec(), msg.data.get("newValue")); Set affected = listToUuidSet(msg.data.get("affected")); + RemoteChangeHandler handler = remoteChangeHandler; + if (handler != null && field instanceof LinkedField linkedField) { + handler.onLinkedChange(field, linkedField.linkType().name(), linkKey, newValue); + } super.fireLinkedDataChanged(field, linkKey, oldValue, newValue, affected); } diff --git a/src/main/java/net/swofty/event/RemoteChangeHandler.java b/src/main/java/net/swofty/event/RemoteChangeHandler.java new file mode 100644 index 0000000..75806a0 --- /dev/null +++ b/src/main/java/net/swofty/event/RemoteChangeHandler.java @@ -0,0 +1,17 @@ +package net.swofty.event; + +import net.swofty.DataField; + +import java.util.UUID; + +/** + * Applies changes that arrive from another node (over the distributed event bus) to this + * node's local caches, so a container that is currently loaded here does not serve a stale + * value after a peer mutates it. Wired by the API implementation; the event bus itself has + * no knowledge of how data is cached. + */ +public interface RemoteChangeHandler { + void onPlayerChange(DataField field, UUID player, T newValue); + + void onLinkedChange(DataField field, String linkTypeName, String linkKey, T newValue); +} diff --git a/src/main/java/net/swofty/lock/DistributedLock.java b/src/main/java/net/swofty/lock/DistributedLock.java new file mode 100644 index 0000000..303343c --- /dev/null +++ b/src/main/java/net/swofty/lock/DistributedLock.java @@ -0,0 +1,25 @@ +package net.swofty.lock; + +import java.time.Duration; + +/** + * A mutual-exclusion lock that spans server instances. Supplying one to the API upgrades + * transactions from a JVM-local lock (which only serialises threads within a single process) + * to true cross-node mutual exclusion — required when the same entity (e.g. a coop or island + * shared across servers) can be mutated from more than one node at once. + */ +public interface DistributedLock { + + /** + * Acquires the named lock, blocking up to {@code timeout}. The returned handle must be + * closed to release the lock; use it with try-with-resources. Throws + * {@link LockAcquisitionException} if the lock cannot be acquired within the timeout. + */ + Handle acquire(String key, Duration timeout); + + /** A held lock. Closing it releases the lock and never throws a checked exception. */ + interface Handle extends AutoCloseable { + @Override + void close(); + } +} diff --git a/src/main/java/net/swofty/lock/InMemoryDistributedLock.java b/src/main/java/net/swofty/lock/InMemoryDistributedLock.java new file mode 100644 index 0000000..8943533 --- /dev/null +++ b/src/main/java/net/swofty/lock/InMemoryDistributedLock.java @@ -0,0 +1,29 @@ +package net.swofty.lock; + +import java.time.Duration; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.ReentrantLock; + +/** + * A single-JVM {@link DistributedLock} backed by {@link ReentrantLock}. Useful for tests and + * single-node deployments; it does not coordinate across processes. Being reentrant, a thread + * that already holds a key can acquire it again (e.g. a nested transaction on the same entity). + */ +public class InMemoryDistributedLock implements DistributedLock { + private final ConcurrentHashMap locks = new ConcurrentHashMap<>(); + + @Override + public Handle acquire(String key, Duration timeout) { + ReentrantLock lock = locks.computeIfAbsent(key, k -> new ReentrantLock()); + try { + if (!lock.tryLock(timeout.toMillis(), TimeUnit.MILLISECONDS)) { + throw new LockAcquisitionException("Timed out acquiring lock: " + key); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new LockAcquisitionException("Interrupted acquiring lock: " + key); + } + return lock::unlock; + } +} diff --git a/src/main/java/net/swofty/lock/LockAcquisitionException.java b/src/main/java/net/swofty/lock/LockAcquisitionException.java new file mode 100644 index 0000000..02bac0f --- /dev/null +++ b/src/main/java/net/swofty/lock/LockAcquisitionException.java @@ -0,0 +1,8 @@ +package net.swofty.lock; + +/** Thrown when a {@link DistributedLock} cannot be acquired within its timeout. */ +public class LockAcquisitionException extends RuntimeException { + public LockAcquisitionException(String message) { + super(message); + } +} diff --git a/src/main/java/net/swofty/lock/RedisDistributedLock.java b/src/main/java/net/swofty/lock/RedisDistributedLock.java new file mode 100644 index 0000000..bea25cd --- /dev/null +++ b/src/main/java/net/swofty/lock/RedisDistributedLock.java @@ -0,0 +1,67 @@ +package net.swofty.lock; + +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisPool; +import redis.clients.jedis.params.SetParams; + +import java.time.Duration; +import java.util.Collections; +import java.util.UUID; + +/** + * A Redis-backed {@link DistributedLock} using {@code SET key token NX PX } to acquire + * and a compare-and-delete Lua script to release, so a lock is only ever released by its owner. + * A lease time bounds how long a crashed holder can block others. Not reentrant across nodes: + * re-acquiring a key already held (even by the same thread) blocks until the lease or timeout. + */ +public class RedisDistributedLock implements DistributedLock { + private static final String UNLOCK_SCRIPT = + "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end"; + + private final JedisPool pool; + private final String prefix; + private final Duration leaseTime; + private final long retryDelayMillis; + + public RedisDistributedLock(JedisPool pool) { + this(pool, "swofty:lock", Duration.ofSeconds(30), 50); + } + + public RedisDistributedLock(JedisPool pool, String prefix, Duration leaseTime, long retryDelayMillis) { + this.pool = pool; + this.prefix = prefix; + this.leaseTime = leaseTime; + this.retryDelayMillis = retryDelayMillis; + } + + @Override + public Handle acquire(String key, Duration timeout) { + String redisKey = prefix + ":" + key; + String token = UUID.randomUUID().toString(); + long deadline = System.nanoTime() + timeout.toNanos(); + SetParams params = new SetParams().nx().px(leaseTime.toMillis()); + + while (true) { + try (Jedis jedis = pool.getResource()) { + if ("OK".equals(jedis.set(redisKey, token, params))) { + return () -> release(redisKey, token); + } + } + if (System.nanoTime() >= deadline) { + throw new LockAcquisitionException("Timed out acquiring lock: " + key); + } + try { + Thread.sleep(retryDelayMillis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new LockAcquisitionException("Interrupted acquiring lock: " + key); + } + } + } + + private void release(String redisKey, String token) { + try (Jedis jedis = pool.getResource()) { + jedis.eval(UNLOCK_SCRIPT, Collections.singletonList(redisKey), Collections.singletonList(token)); + } + } +} diff --git a/src/main/java/net/swofty/storage/InMemoryDataStorage.java b/src/main/java/net/swofty/storage/InMemoryDataStorage.java index 64842e8..8086fe8 100644 --- a/src/main/java/net/swofty/storage/InMemoryDataStorage.java +++ b/src/main/java/net/swofty/storage/InMemoryDataStorage.java @@ -3,8 +3,9 @@ import java.util.*; import java.util.concurrent.ConcurrentHashMap; -public class InMemoryDataStorage implements DataStorage { +public class InMemoryDataStorage implements DataStorage, LeaderboardIndex { private final ConcurrentHashMap> data = new ConcurrentHashMap<>(); + private final ConcurrentHashMap> leaderboards = new ConcurrentHashMap<>(); @Override public byte[] load(String type, String id) { @@ -37,4 +38,56 @@ public boolean exists(String type, String id) { return bucket != null && bucket.containsKey(id); } + // ---- LeaderboardIndex --------------------------------------------------- + + @Override + public void updateScore(String leaderboard, String id, double score) { + leaderboards.computeIfAbsent(leaderboard, k -> new ConcurrentHashMap<>()).put(id, score); + } + + @Override + public void updateScoreIfPresent(String leaderboard, String id, double score) { + ConcurrentHashMap board = leaderboards.get(leaderboard); + if (board != null) { + board.put(id, score); + } + } + + @Override + public boolean leaderboardExists(String leaderboard) { + ConcurrentHashMap board = leaderboards.get(leaderboard); + return board != null && !board.isEmpty(); + } + + @Override + public void removeFromLeaderboard(String leaderboard, String id) { + ConcurrentHashMap board = leaderboards.get(leaderboard); + if (board != null) { + board.remove(id); + } + } + + @Override + public List scoreRange(String leaderboard, int start, int endInclusive, boolean descending) { + ConcurrentHashMap board = leaderboards.get(leaderboard); + if (board == null) { + return List.of(); + } + List sorted = new ArrayList<>(); + board.forEach((id, score) -> sorted.add(new ScoreEntry(id, score))); + sorted.sort(descending + ? Comparator.comparingDouble(ScoreEntry::score).reversed() + : Comparator.comparingDouble(ScoreEntry::score)); + if (start >= sorted.size() || start < 0) { + return List.of(); + } + int end = Math.min(endInclusive, sorted.size() - 1); + return new ArrayList<>(sorted.subList(start, end + 1)); + } + + @Override + public long leaderboardSize(String leaderboard) { + ConcurrentHashMap board = leaderboards.get(leaderboard); + return board == null ? 0 : board.size(); + } } diff --git a/src/main/java/net/swofty/storage/LeaderboardIndex.java b/src/main/java/net/swofty/storage/LeaderboardIndex.java new file mode 100644 index 0000000..87f4102 --- /dev/null +++ b/src/main/java/net/swofty/storage/LeaderboardIndex.java @@ -0,0 +1,43 @@ +package net.swofty.storage; + +import java.util.List; + +/** + * A capability a {@link DataStorage} may implement to maintain sorted indexes (e.g. Redis sorted + * sets) for leaderboard fields. It is the sole backing for {@code getTop}/{@code getTopPaged}, which + * read the requested slice directly ({@code O(log N + page)}) rather than scanning and deserializing + * every stored entity. Indexes are self-registering: built once on first rank ({@link #updateScore}) + * and maintained on every write ({@link #updateScoreIfPresent}), so ranking needs no registration for + * numeric fields and unranked fields cost nothing. Storage that lacks this capability cannot rank. + */ +public interface LeaderboardIndex { + + /** A ranked member: its id and score. */ + record ScoreEntry(String id, double score) {} + + /** Records or updates a member's score in the named leaderboard, creating it if needed. */ + void updateScore(String leaderboard, String id, double score); + + /** + * Records a member's score only if the leaderboard already exists. This is how every node + * maintains a leaderboard once any node has built it, without resurrecting indexes for fields + * that are never ranked. Must be atomic with respect to the existence check. + */ + void updateScoreIfPresent(String leaderboard, String id, double score); + + /** Whether the named leaderboard has been built (has any members). */ + boolean leaderboardExists(String leaderboard); + + /** Removes a member from the named leaderboard. */ + void removeFromLeaderboard(String leaderboard, String id); + + /** Returns the {@code [start, endInclusive]} rank slice, ordered by score. */ + List scoreRange(String leaderboard, int start, int endInclusive, boolean descending); + + /** Number of members in the named leaderboard. */ + long leaderboardSize(String leaderboard); + + default List topScores(String leaderboard, int limit, boolean descending) { + return scoreRange(leaderboard, 0, limit - 1, descending); + } +} diff --git a/src/main/java/net/swofty/storage/RedisDataStorage.java b/src/main/java/net/swofty/storage/RedisDataStorage.java index 0d982d2..52f3cc3 100644 --- a/src/main/java/net/swofty/storage/RedisDataStorage.java +++ b/src/main/java/net/swofty/storage/RedisDataStorage.java @@ -3,12 +3,13 @@ import redis.clients.jedis.Jedis; import redis.clients.jedis.JedisPool; import redis.clients.jedis.JedisPoolConfig; +import redis.clients.jedis.resps.Tuple; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.List; -public class RedisDataStorage implements DataStorage { +public class RedisDataStorage implements DataStorage, LeaderboardIndex { private final JedisPool pool; private final String prefix; @@ -70,6 +71,67 @@ public boolean exists(String type, String id) { } } + // ---- LeaderboardIndex (Redis sorted sets) ------------------------------- + + private String leaderboardKey(String leaderboard) { + return prefix + ":lb:" + leaderboard; + } + + private static final String ZADD_IF_EXISTS = + "if redis.call('exists', KEYS[1]) == 1 then return redis.call('zadd', KEYS[1], ARGV[1], ARGV[2]) else return 0 end"; + + @Override + public void updateScore(String leaderboard, String id, double score) { + try (Jedis jedis = pool.getResource()) { + jedis.zadd(leaderboardKey(leaderboard), score, id); + } + } + + @Override + public void updateScoreIfPresent(String leaderboard, String id, double score) { + try (Jedis jedis = pool.getResource()) { + jedis.eval(ZADD_IF_EXISTS, + java.util.List.of(leaderboardKey(leaderboard)), + java.util.List.of(Double.toString(score), id)); + } + } + + @Override + public boolean leaderboardExists(String leaderboard) { + try (Jedis jedis = pool.getResource()) { + return jedis.exists(leaderboardKey(leaderboard)); + } + } + + @Override + public void removeFromLeaderboard(String leaderboard, String id) { + try (Jedis jedis = pool.getResource()) { + jedis.zrem(leaderboardKey(leaderboard), id); + } + } + + @Override + public List scoreRange(String leaderboard, int start, int endInclusive, boolean descending) { + String key = leaderboardKey(leaderboard); + try (Jedis jedis = pool.getResource()) { + List tuples = descending + ? jedis.zrevrangeWithScores(key, start, endInclusive) + : jedis.zrangeWithScores(key, start, endInclusive); + List result = new ArrayList<>(tuples.size()); + for (Tuple tuple : tuples) { + result.add(new ScoreEntry(tuple.getElement(), tuple.getScore())); + } + return result; + } + } + + @Override + public long leaderboardSize(String leaderboard) { + try (Jedis jedis = pool.getResource()) { + return jedis.zcard(leaderboardKey(leaderboard)); + } + } + public JedisPool getPool() { return pool; } diff --git a/src/test/java/net/swofty/BulkOperationsTest.java b/src/test/java/net/swofty/BulkOperationsTest.java index 918cff3..20c380b 100644 --- a/src/test/java/net/swofty/BulkOperationsTest.java +++ b/src/test/java/net/swofty/BulkOperationsTest.java @@ -2,21 +2,16 @@ import net.swofty.api.DataAPIImpl; import net.swofty.codec.Codecs; -import net.swofty.data.format.JsonFormat; -import net.swofty.storage.FileDataStorage; +import net.swofty.storage.InMemoryDataStorage; import org.junit.jupiter.api.*; -import org.junit.jupiter.api.io.TempDir; -import java.nio.file.Path; import java.util.*; import static org.junit.jupiter.api.Assertions.*; class BulkOperationsTest { - @TempDir - Path tempDir; private DataAPIImpl api; private static final PlayerField COINS = PlayerField.create("test", "coins", Codecs.INT, 0); @@ -27,7 +22,8 @@ class BulkOperationsTest { @BeforeEach void setUp() { - api = new DataAPIImpl(new FileDataStorage(tempDir, new JsonFormat(), ".json"), new JsonFormat()); + // InMemory storage maintains a leaderboard index; numeric leaderboards need no registration. + api = new DataAPIImpl(new InMemoryDataStorage()); } @AfterEach diff --git a/src/test/java/net/swofty/DistributedLockTest.java b/src/test/java/net/swofty/DistributedLockTest.java new file mode 100644 index 0000000..3ce81f8 --- /dev/null +++ b/src/test/java/net/swofty/DistributedLockTest.java @@ -0,0 +1,77 @@ +package net.swofty; + +import net.swofty.api.DataAPIImpl; +import net.swofty.codec.Codecs; +import net.swofty.data.format.JsonFormat; +import net.swofty.lock.DistributedLock; +import net.swofty.lock.InMemoryDistributedLock; +import net.swofty.lock.LockAcquisitionException; +import net.swofty.storage.InMemoryDataStorage; + +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.*; + +class DistributedLockTest { + + private static final PlayerField COINS = PlayerField.create("game", "coins", Codecs.INT, 0); + + @Test + void concurrentTransactionsSerialiseUnderTheLock() throws InterruptedException { + DistributedLock lock = new InMemoryDistributedLock(); + // Shared lock instance across two "nodes" so they contend as they would over Redis. + DataAPIImpl node = new DataAPIImpl(new InMemoryDataStorage(), new JsonFormat(), null, true, lock); + UUID player = UUID.randomUUID(); + node.set(player, COINS, 0); + + int threads = 8, incrementsEach = 200; + CountDownLatch start = new CountDownLatch(1); + CountDownLatch done = new CountDownLatch(threads); + for (int i = 0; i < threads; i++) { + new Thread(() -> { + try { + start.await(); + for (int j = 0; j < incrementsEach; j++) { + node.transaction(player, tx -> { + tx.set(COINS, tx.get(COINS) + 1); + return null; + }); + } + } catch (InterruptedException ignored) { + } finally { + done.countDown(); + } + }).start(); + } + start.countDown(); + done.await(); + + assertEquals(threads * incrementsEach, node.get(player, COINS), "no lost updates under contention"); + node.shutdown(); + } + + @Test + void heldLockBlocksAnotherAcquirerUntilTimeout() throws InterruptedException { + DistributedLock lock = new InMemoryDistributedLock(); + AtomicInteger failures = new AtomicInteger(); + + DistributedLock.Handle held = lock.acquire("k", Duration.ofSeconds(1)); + Thread other = new Thread(() -> { + try (DistributedLock.Handle ignored = lock.acquire("k", Duration.ofMillis(100))) { + // should not get here while held + } catch (LockAcquisitionException e) { + failures.incrementAndGet(); + } + }); + other.start(); + other.join(); + held.close(); + + assertEquals(1, failures.get(), "second acquirer times out while the lock is held"); + } +} diff --git a/src/test/java/net/swofty/LeaderboardIndexTest.java b/src/test/java/net/swofty/LeaderboardIndexTest.java new file mode 100644 index 0000000..ba741ee --- /dev/null +++ b/src/test/java/net/swofty/LeaderboardIndexTest.java @@ -0,0 +1,103 @@ +package net.swofty; + +import net.swofty.api.DataAPIImpl; +import net.swofty.codec.Codecs; +import net.swofty.data.format.JsonFormat; +import net.swofty.storage.FileDataStorage; +import net.swofty.storage.InMemoryDataStorage; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.util.List; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Leaderboards are index-backed and self-registering: no trackLeaderboard call is needed for a + * numeric field. The index builds on first rank and is maintained on every subsequent write. + */ +class LeaderboardIndexTest { + + @TempDir + Path tempDir; + + private static final PlayerField COINS = PlayerField.create("game", "coins", Codecs.INT, 0); + + @Test + void numericLeaderboardWorksWithNoRegistration() { + DataAPIImpl api = new DataAPIImpl(new InMemoryDataStorage()); + UUID a = UUID.randomUUID(), b = UUID.randomUUID(), c = UUID.randomUUID(); + api.set(a, COINS, 300); + api.set(b, COINS, 100); + api.set(c, COINS, 200); + + List> top = api.getTop(COINS, 3); // no trackLeaderboard + assertEquals(List.of(a, c, b), top.stream().map(LeaderboardEntry::playerId).toList()); + assertEquals(List.of(300, 200, 100), top.stream().map(LeaderboardEntry::value).toList()); + assertEquals(1, top.get(0).rank()); + api.shutdown(); + } + + @Test + void firstRankBackfillsExistingDataThenWritesKeepItCurrent() { + InMemoryDataStorage storage = new InMemoryDataStorage(); + + // Data written before any ranking happened. + DataAPIImpl seed = new DataAPIImpl(storage); + UUID a = UUID.randomUUID(), b = UUID.randomUUID(); + seed.set(a, COINS, 5); + seed.set(b, COINS, 9); + seed.shutdown(); + + DataAPIImpl api = new DataAPIImpl(storage); + // First rank self-builds the index from the pre-existing data. + assertEquals(List.of(b, a), api.getTop(COINS, 2).stream().map(LeaderboardEntry::playerId).toList()); + + // A later write is reflected without any re-registration. + api.update(a, COINS, v -> v + 100); // a -> 105, overtakes b + assertEquals(a, api.getTop(COINS, 1).get(0).playerId()); + api.shutdown(); + } + + @Test + void paging() { + DataAPIImpl api = new DataAPIImpl(new InMemoryDataStorage()); + for (int i = 1; i <= 25; i++) { + api.set(UUID.randomUUID(), COINS, i * 10); + } + Page> page = api.getTopPaged(COINS, 1, 10); + assertEquals(10, page.content().size()); + assertEquals(3, page.totalPages()); + assertEquals(25, page.totalElements()); + assertEquals(250, page.content().get(0).value()); + api.shutdown(); + } + + @Test + void nonNumericFieldRequiresAScorer() { + PlayerField NAME = PlayerField.create("game", "name", Codecs.STRING, ""); + DataAPIImpl api = new DataAPIImpl(new InMemoryDataStorage()); + api.set(UUID.randomUUID(), NAME, "abc"); + + // Cannot auto-score a String; ranking it without a scorer fails with a clear message. + assertThrows(IllegalStateException.class, () -> api.getTop(NAME, 5)); + + // Registering a scorer makes it rankable. + UUID longName = UUID.randomUUID(); + api.set(longName, NAME, "a-very-long-name"); + api.trackLeaderboard(NAME, String::length); + assertEquals(longName, api.getTop(NAME, 5).get(0).playerId()); + api.shutdown(); + } + + @Test + void storageWithoutAnIndexCannotRank() { + DataAPIImpl api = new DataAPIImpl(new FileDataStorage(tempDir, new JsonFormat(), ".json"), new JsonFormat()); + assertThrows(IllegalStateException.class, () -> api.getTop(COINS, 10)); + assertThrows(IllegalStateException.class, () -> api.trackLeaderboard(COINS, Integer::doubleValue)); + api.shutdown(); + } +} diff --git a/src/test/java/net/swofty/LifecycleTest.java b/src/test/java/net/swofty/LifecycleTest.java new file mode 100644 index 0000000..d886f4a --- /dev/null +++ b/src/test/java/net/swofty/LifecycleTest.java @@ -0,0 +1,105 @@ +package net.swofty; + +import net.swofty.api.DataAPIImpl; +import net.swofty.codec.Codecs; +import net.swofty.data.format.JsonFormat; +import net.swofty.storage.DataStorage; +import net.swofty.storage.InMemoryDataStorage; + +import org.junit.jupiter.api.Test; + +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Exercises the multi-node cache lifecycle: warming data before use (the primitive a + * proxy uses to load a player on the target server before sending them), evicting it so a + * later visit is never stale, and deferring writes until flush/unload. + */ +class LifecycleTest { + + private static final PlayerField COINS = PlayerField.create("game", "coins", Codecs.INT, 0); + + @Test + void unloadEvictsSoALaterReadIsNotStale() { + DataStorage storage = new InMemoryDataStorage(); + UUID player = UUID.randomUUID(); + + // Two independent nodes over shared storage. + DataAPIImpl nodeA = new DataAPIImpl(storage); + DataAPIImpl nodeB = new DataAPIImpl(storage); + + nodeA.set(player, COINS, 100); + assertEquals(100, nodeA.get(player, COINS)); // now cached on A + + // Player moves to B and earns coins there. + nodeB.set(player, COINS, 500); + + // A still has the stale cached value until the player is unloaded from A. + assertEquals(100, nodeA.get(player, COINS), "A holds its cached value"); + nodeA.unload(player); + assertFalse(nodeA.isLoaded(player)); + assertEquals(500, nodeA.get(player, COINS), "after eviction A reloads the fresh value"); + + nodeA.shutdown(); + nodeB.shutdown(); + } + + @Test + void loadWarmsDataThenUnloadEvicts() { + DataStorage storage = new InMemoryDataStorage(); + UUID player = UUID.randomUUID(); + + DataAPIImpl writer = new DataAPIImpl(storage); + writer.set(player, COINS, 7); + writer.shutdown(); + + // Target node warms the player's data up front, as a proxy would before sending them. + DataAPIImpl target = new DataAPIImpl(storage); + assertFalse(target.isLoaded(player)); + target.load(player); + assertTrue(target.isLoaded(player), "warmed into cache"); + assertEquals(7, target.get(player, COINS)); + + target.unload(player); + assertFalse(target.isLoaded(player), "evicted after use"); + target.shutdown(); + } + + @Test + void deferredPersistenceHoldsWritesUntilFlush() { + DataStorage storage = new InMemoryDataStorage(); + UUID player = UUID.randomUUID(); + + DataAPIImpl node = new DataAPIImpl(storage, new JsonFormat(), null, /* autoPersist */ false); + node.set(player, COINS, 42); + + // Not yet visible to another node — the write is still buffered. + DataAPIImpl observer = new DataAPIImpl(storage); + assertEquals(0, observer.get(player, COINS), "write is deferred, not yet persisted"); + + node.flush(player); + + DataAPIImpl observer2 = new DataAPIImpl(storage); + assertEquals(42, observer2.get(player, COINS), "flush persists the buffered write"); + + node.shutdown(); + observer.shutdown(); + observer2.shutdown(); + } + + @Test + void shutdownFlushesDeferredWrites() { + DataStorage storage = new InMemoryDataStorage(); + UUID player = UUID.randomUUID(); + + DataAPIImpl node = new DataAPIImpl(storage, new JsonFormat(), null, false); + node.set(player, COINS, 13); + node.shutdown(); // must flush before exiting + + DataAPIImpl observer = new DataAPIImpl(storage); + assertEquals(13, observer.get(player, COINS)); + observer.shutdown(); + } +} diff --git a/src/test/java/net/swofty/MultiFieldPersistenceTest.java b/src/test/java/net/swofty/MultiFieldPersistenceTest.java new file mode 100644 index 0000000..8aea5bd --- /dev/null +++ b/src/test/java/net/swofty/MultiFieldPersistenceTest.java @@ -0,0 +1,72 @@ +package net.swofty; + +import net.swofty.api.DataAPIImpl; +import net.swofty.codec.Codecs; +import net.swofty.storage.DataStorage; +import net.swofty.storage.InMemoryDataStorage; + +import org.junit.jupiter.api.Test; + +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Regression tests for the "partial write drops untouched fields" bug. A node that + * writes one field without ever having read another must not wipe the unread field + * from storage — otherwise moving a player between server instances loses data. + */ +class MultiFieldPersistenceTest { + + private static final PlayerField COINS = PlayerField.create("game", "coins", Codecs.INT, 0); + private static final PlayerField GEMS = PlayerField.create("game", "gems", Codecs.INT, 0); + private static final PlayerField NAME = PlayerField.create("game", "name", Codecs.STRING, ""); + + @Test + void writingOneFieldDoesNotDropUntouchedFields() { + DataStorage storage = new InMemoryDataStorage(); + UUID player = UUID.randomUUID(); + + // Session 1: populate several fields. + DataAPIImpl api1 = new DataAPIImpl(storage); + api1.set(player, COINS, 100); + api1.set(player, GEMS, 50); + api1.set(player, NAME, "swofty"); + api1.shutdown(); + + // Session 2 (fresh node, empty cache): touch ONLY coins, never read gems/name. + DataAPIImpl api2 = new DataAPIImpl(storage); + api2.set(player, COINS, 999); + api2.shutdown(); + + // Session 3: the untouched fields must survive. + DataAPIImpl api3 = new DataAPIImpl(storage); + assertEquals(999, api3.get(player, COINS), "updated field"); + assertEquals(50, api3.get(player, GEMS), "untouched field must not be dropped"); + assertEquals("swofty", api3.get(player, NAME), "untouched field must not be dropped"); + api3.shutdown(); + } + + @Test + void nullSetDeletesFromBackingDocumentInsteadOfResurrecting() { + DataStorage storage = new InMemoryDataStorage(); + UUID player = UUID.randomUUID(); + + // A value exists in the backing document. + DataAPIImpl api1 = new DataAPIImpl(storage); + api1.set(player, COINS, 7); + api1.set(player, GEMS, 5); + api1.shutdown(); + + // A fresh node clears coins. The merge must honour the deletion (tombstone) + // rather than resurrecting the old value from the backing document. + DataAPIImpl api2 = new DataAPIImpl(storage); + api2.set(player, COINS, null); + api2.shutdown(); + + DataAPIImpl api3 = new DataAPIImpl(storage); + assertEquals(0, api3.get(player, COINS), "cleared field falls back to default, not the old value"); + assertEquals(5, api3.get(player, GEMS), "sibling field survives the clear"); + api3.shutdown(); + } +}