From c8c8379b29ad072128e02e03ea985d14c4b8a9c9 Mon Sep 17 00:00:00 2001 From: Swofty-Developments Date: Thu, 23 Jul 2026 23:26:19 +1000 Subject: [PATCH 1/8] fix: preserve untouched fields on partial writes Previously a container serialized only the fields touched this session and saved that as a full overwrite, so the first write from a node that had not read every field silently dropped the unread fields from storage. When a player moved between server instances this lost data. The container now retains the full backing document it last saw in storage and, on serialize, merges the touched fields over it, so untouched fields are preserved. A null-set records a tombstone that suppresses the field from the merged output, so clearing a value still deletes it instead of resurrecting the stored one. DataFormat gains readRaw/writeRaw for whole-document access, implemented for the JSON format, and the managers warm the backing document once per entity rather than reloading storage on every field access. --- .../java/net/swofty/api/DataContainer.java | 81 ++++++++++++++++--- .../net/swofty/api/LinkedDataManager.java | 12 ++- .../net/swofty/api/PlayerDataManager.java | 14 +++- src/main/java/net/swofty/data/DataFormat.java | 20 +++++ .../net/swofty/data/format/JsonFormat.java | 14 ++++ .../net/swofty/MultiFieldPersistenceTest.java | 72 +++++++++++++++++ 6 files changed, 196 insertions(+), 17 deletions(-) create mode 100644 src/test/java/net/swofty/MultiFieldPersistenceTest.java diff --git a/src/main/java/net/swofty/api/DataContainer.java b/src/main/java/net/swofty/api/DataContainer.java index 586b75b..7ec2c6f 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; +/** + * 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. + */ public 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,71 @@ 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); + } + + /** 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/LinkedDataManager.java b/src/main/java/net/swofty/api/LinkedDataManager.java index 95b04f5..d4c1882 100644 --- a/src/main/java/net/swofty/api/LinkedDataManager.java +++ b/src/main/java/net/swofty/api/LinkedDataManager.java @@ -107,8 +107,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 +116,17 @@ 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); byte[] bytes = container.serialize(format); storage.save("linked/" + linkTypeName, key.toString(), bytes); + container.markPersisted(bytes); + } + + private void ensureDocumentLoaded(String linkTypeName, Object key, DataContainer container) { + if (!container.isDocumentLoaded()) { + container.loadDocument(format, storage.load("linked/" + linkTypeName, key.toString())); + } } 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..cb7f572 100644 --- a/src/main/java/net/swofty/api/PlayerDataManager.java +++ b/src/main/java/net/swofty/api/PlayerDataManager.java @@ -64,23 +64,33 @@ 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); } + private void ensureDocumentLoaded(UUID player, DataContainer container) { + if (!container.isDocumentLoaded()) { + container.loadDocument(format, storage.load("players", player.toString())); + } + } + void persist(UUID player) { DataContainer container = cache.get(player); if (container != null) { byte[] bytes = container.serialize(format); storage.save("players", player.toString(), bytes); + container.markPersisted(bytes); } } 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/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(); + } +} From 512020aba7d49dcbee6d81c3f3da50fe4995b04c Mon Sep 17 00:00:00 2001 From: Swofty-Developments Date: Thu, 23 Jul 2026 23:32:06 +1000 Subject: [PATCH 2/8] feat: add multi-node cache lifecycle and coherency Adds an explicit lifecycle to the data API so a node can warm an entity's whole document into its cache in a single storage read, flush pending changes, and evict the entity when finished. load/loadAsync/flush/unload/isLoaded/loadedPlayers cover players and loadLink/flushLink/unloadLink/isLinkLoaded cover shared entities such as an island or coop. Eviction is what keeps a multi-server deployment correct: without it a container lingered in a node's cache forever, so a player who changed data on another server and later returned would be served the stale cached value and, on the next write, overwrite the fresh data. unload now flushes and drops the container so a later visit reloads from storage. load is also the primitive a proxy needs to prepare a player's data on the target server before moving them there. When a distributed event bus is present the API now keeps loaded containers coherent with peer writes: a remote change to a currently-loaded entity updates the local view in place, preserving the prior dirty state so a read-only node does not re-persist stale sibling fields. Coherency applies to fields that have a subscription registered on the receiving node. An optional deferred-persistence mode buffers writes until flush or unload so a play session can be written back once instead of per field; shutdown flushes everything so no buffered write is lost. --- src/main/java/net/swofty/DataAPI.java | 18 +++ src/main/java/net/swofty/api/DataAPIImpl.java | 83 +++++++++++++- .../java/net/swofty/api/DataContainer.java | 11 ++ .../net/swofty/api/LinkedDataManager.java | 93 +++++++++++++++- .../net/swofty/api/PlayerDataManager.java | 74 +++++++++++- .../net/swofty/event/DistributedEventBus.java | 16 +++ .../net/swofty/event/RemoteChangeHandler.java | 17 +++ src/test/java/net/swofty/LifecycleTest.java | 105 ++++++++++++++++++ 8 files changed, 411 insertions(+), 6 deletions(-) create mode 100644 src/main/java/net/swofty/event/RemoteChangeHandler.java create mode 100644 src/test/java/net/swofty/LifecycleTest.java diff --git a/src/main/java/net/swofty/DataAPI.java b/src/main/java/net/swofty/DataAPI.java index 83650f5..86a204f 100644 --- a/src/main/java/net/swofty/DataAPI.java +++ b/src/main/java/net/swofty/DataAPI.java @@ -6,6 +6,8 @@ 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.UnaryOperator; @@ -69,4 +71,20 @@ public interface DataAPI { // Bulk operations - Linked > List> getTopLinked(LinkedField field, int limit); List queryLinked(LinkedField field, Predicate filter); + + // 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/DataAPIImpl.java b/src/main/java/net/swofty/api/DataAPIImpl.java index a83ecd8..426d1a8 100644 --- a/src/main/java/net/swofty/api/DataAPIImpl.java +++ b/src/main/java/net/swofty/api/DataAPIImpl.java @@ -10,6 +10,8 @@ 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.UnaryOperator; @@ -24,14 +26,38 @@ public class DataAPIImpl implements DataAPI { private final BulkOperationExecutor bulkOperations; public DataAPIImpl(DataStorage storage, DataFormat format, PubSubHandler pubSub) { + this(storage, format, pubSub, true); + } + + /** + * @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. + */ + public DataAPIImpl(DataStorage storage, DataFormat format, PubSubHandler pubSub, boolean autoPersist) { this.storage = storage; 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.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) { @@ -269,7 +295,60 @@ public List queryLinked(LinkedField field, Predicate filter) // ==================== 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); + } + 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 7ec2c6f..bff796f 100644 --- a/src/main/java/net/swofty/api/DataContainer.java +++ b/src/main/java/net/swofty/api/DataContainer.java @@ -100,6 +100,17 @@ public byte[] serialize(DataFormat format) { 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; diff --git a/src/main/java/net/swofty/api/LinkedDataManager.java b/src/main/java/net/swofty/api/LinkedDataManager.java index d4c1882..d7ca878 100644 --- a/src/main/java/net/swofty/api/LinkedDataManager.java +++ b/src/main/java/net/swofty/api/LinkedDataManager.java @@ -18,12 +18,19 @@ public class LinkedDataManager { 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) { @@ -118,9 +125,9 @@ void setFieldValue(String linkTypeName, Object key, DataField field, T va DataContainer container = getContainer(ck); ensureDocumentLoaded(linkTypeName, key, container); container.set(field, value); - byte[] bytes = container.serialize(format); - storage.save("linked/" + linkTypeName, key.toString(), bytes); - container.markPersisted(bytes); + if (autoPersist) { + persistLinked(linkTypeName, key, container); + } } private void ensureDocumentLoaded(String linkTypeName, Object key, DataContainer container) { @@ -129,6 +136,86 @@ private void ensureDocumentLoaded(String linkTypeName, Object key, DataContainer } } + private void persistLinked(String linkTypeName, String keyString, DataContainer container) { + byte[] bytes = container.serialize(format); + 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) { return storage.listIds("linked/" + linkTypeName); } diff --git a/src/main/java/net/swofty/api/PlayerDataManager.java b/src/main/java/net/swofty/api/PlayerDataManager.java index cb7f572..07b3b9d 100644 --- a/src/main/java/net/swofty/api/PlayerDataManager.java +++ b/src/main/java/net/swofty/api/PlayerDataManager.java @@ -17,11 +17,17 @@ public class PlayerDataManager { private final EventBus eventBus; private final ConcurrentHashMap cache = new ConcurrentHashMap<>(); private final ConcurrentHashMap locks = 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) { @@ -76,7 +82,9 @@ void setFieldValue(UUID player, DataField field, T value) { // 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) { @@ -94,6 +102,70 @@ void persist(UUID player) { } } + // ---- 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); + } + } + public List listPlayerIds() { return storage.listIds("players"); } 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/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(); + } +} From 19a7f7aa384a0436a0dfdb5f9397766c70ea1b16 Mon Sep 17 00:00:00 2001 From: Swofty-Developments Date: Thu, 23 Jul 2026 23:35:10 +1000 Subject: [PATCH 3/8] feat: add optional distributed locking for transactions Transactions previously guarded only a JVM-local monitor, so two servers mutating the same shared entity (e.g. a coop or island) could interleave and lose updates despite the API presenting transactions as atomic. A DistributedLock can now be supplied; when present, transactions take a cross-node lock keyed by the entity in addition to the local monitor, and a public lock(key, timeout) helper exposes the same primitive for app-level critical sections that span more than one field or entity. When absent, behaviour is unchanged. Ships a Redis implementation using 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, plus a reentrant in-memory implementation for single-node use and tests. --- src/main/java/net/swofty/api/DataAPIImpl.java | 33 ++++++- .../net/swofty/api/TransactionManager.java | 93 ++++++++++++------- .../java/net/swofty/lock/DistributedLock.java | 25 +++++ .../swofty/lock/InMemoryDistributedLock.java | 29 ++++++ .../swofty/lock/LockAcquisitionException.java | 8 ++ .../net/swofty/lock/RedisDistributedLock.java | 67 +++++++++++++ .../java/net/swofty/DistributedLockTest.java | 77 +++++++++++++++ 7 files changed, 293 insertions(+), 39 deletions(-) create mode 100644 src/main/java/net/swofty/lock/DistributedLock.java create mode 100644 src/main/java/net/swofty/lock/InMemoryDistributedLock.java create mode 100644 src/main/java/net/swofty/lock/LockAcquisitionException.java create mode 100644 src/main/java/net/swofty/lock/RedisDistributedLock.java create mode 100644 src/test/java/net/swofty/DistributedLockTest.java diff --git a/src/main/java/net/swofty/api/DataAPIImpl.java b/src/main/java/net/swofty/api/DataAPIImpl.java index 426d1a8..feaa12b 100644 --- a/src/main/java/net/swofty/api/DataAPIImpl.java +++ b/src/main/java/net/swofty/api/DataAPIImpl.java @@ -4,6 +4,7 @@ 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; @@ -24,24 +25,34 @@ 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 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) { + 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, 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. @@ -345,6 +356,18 @@ 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(); diff --git a/src/main/java/net/swofty/api/TransactionManager.java b/src/main/java/net/swofty/api/TransactionManager.java index 663b310..b4a067b 100644 --- a/src/main/java/net/swofty/api/TransactionManager.java +++ b/src/main/java/net/swofty/api/TransactionManager.java @@ -3,8 +3,10 @@ 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; @@ -12,62 +14,85 @@ import java.util.function.UnaryOperator; public 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/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/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"); + } +} From e8bc5d5f8636b15c2a971e5109b15e68074aca80 Mon Sep 17 00:00:00 2001 From: Swofty-Developments Date: Thu, 23 Jul 2026 23:39:45 +1000 Subject: [PATCH 4/8] feat: add sorted-index leaderboards to avoid full scans getTop, getTopPaged and the query helpers previously loaded and deserialized every stored player on each call, an O(N) scan that does not hold up for a live leaderboard. Storage backends may now implement the optional LeaderboardIndex capability (Redis sorted sets, or an in-memory index for single-node and tests). A field is opted in with trackLeaderboard and a scorer; each persisted write to it updates the index, so getTop and getTopPaged read the ranked slice directly and then load only that page's values, turning the query into O(log N + page). rebuildLeaderboard backfills the index from data written before the field was tracked. Backends without the capability, or untracked fields, fall through to the existing scan, so the change is purely additive. --- src/main/java/net/swofty/DataAPI.java | 7 ++ .../net/swofty/api/BulkOperationExecutor.java | 34 ++++++++ src/main/java/net/swofty/api/DataAPIImpl.java | 11 +++ .../net/swofty/api/PlayerDataManager.java | 47 +++++++++++ .../swofty/storage/InMemoryDataStorage.java | 41 +++++++++- .../net/swofty/storage/LeaderboardIndex.java | 32 ++++++++ .../net/swofty/storage/RedisDataStorage.java | 45 ++++++++++- .../java/net/swofty/LeaderboardIndexTest.java | 77 +++++++++++++++++++ 8 files changed, 292 insertions(+), 2 deletions(-) create mode 100644 src/main/java/net/swofty/storage/LeaderboardIndex.java create mode 100644 src/test/java/net/swofty/LeaderboardIndexTest.java diff --git a/src/main/java/net/swofty/DataAPI.java b/src/main/java/net/swofty/DataAPI.java index 86a204f..c3e9c41 100644 --- a/src/main/java/net/swofty/DataAPI.java +++ b/src/main/java/net/swofty/DataAPI.java @@ -9,6 +9,7 @@ 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 { @@ -72,6 +73,12 @@ public interface DataAPI { > List> getTopLinked(LinkedField field, int limit); List queryLinked(LinkedField field, Predicate filter); + // Leaderboard indexing - when the storage backend maintains sorted indexes (e.g. Redis + // sorted sets), track a field so getTop/getTopPaged read a ranked slice instead of scanning + // every stored player. rebuildLeaderboard backfills the index from existing data once. + void trackLeaderboard(PlayerField field, ToDoubleFunction scorer); + void rebuildLeaderboard(PlayerField field, ToDoubleFunction scorer); + // 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. diff --git a/src/main/java/net/swofty/api/BulkOperationExecutor.java b/src/main/java/net/swofty/api/BulkOperationExecutor.java index 2fe40ce..8be0585 100644 --- a/src/main/java/net/swofty/api/BulkOperationExecutor.java +++ b/src/main/java/net/swofty/api/BulkOperationExecutor.java @@ -3,6 +3,7 @@ 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; @@ -23,9 +24,34 @@ public BulkOperationExecutor(PlayerDataManager playerData, LinkedDataManager lin } public > List> getTop(PlayerField field, int limit) { + LeaderboardIndex index = indexFor(field); + if (index != null) { + return fromIndex(field, index, 0, limit - 1); + } return getTop(field, limit, Comparator.reverseOrder()); } + // The sorted index only exists when the backend maintains one AND the field is tracked; + // otherwise we fall back to the scan below, so this is purely an accelerator. + private LeaderboardIndex indexFor(PlayerField field) { + if (storage instanceof LeaderboardIndex index && playerData.isLeaderboardTracked(field.fullKey())) { + return index; + } + return null; + } + + 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) { List> entries = getAllPlayerValues(field); entries.sort((a, b) -> comparator.compare(a.getValue(), b.getValue())); @@ -38,6 +64,14 @@ public List> getTop(PlayerField field, int limit, Com } public > Page> getTopPaged(PlayerField field, int page, int pageSize) { + LeaderboardIndex index = indexFor(field); + if (index != null) { + long total = index.leaderboardSize(field.fullKey()); + int totalPages = (int) Math.ceil((double) total / pageSize); + int start = (page - 1) * pageSize; + List> content = fromIndex(field, index, start, start + pageSize - 1); + return new Page<>(content, page, totalPages, total); + } List> entries = getAllPlayerValues(field); entries.sort((a, b) -> b.getValue().compareTo(a.getValue())); diff --git a/src/main/java/net/swofty/api/DataAPIImpl.java b/src/main/java/net/swofty/api/DataAPIImpl.java index feaa12b..ae321ea 100644 --- a/src/main/java/net/swofty/api/DataAPIImpl.java +++ b/src/main/java/net/swofty/api/DataAPIImpl.java @@ -14,6 +14,7 @@ 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 { @@ -304,6 +305,16 @@ 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, ToDoubleFunction scorer) { + playerData.rebuildLeaderboard(field, scorer); + } + // ==================== Lifecycle ==================== @Override diff --git a/src/main/java/net/swofty/api/PlayerDataManager.java b/src/main/java/net/swofty/api/PlayerDataManager.java index 07b3b9d..7a40811 100644 --- a/src/main/java/net/swofty/api/PlayerDataManager.java +++ b/src/main/java/net/swofty/api/PlayerDataManager.java @@ -6,9 +6,11 @@ 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 { @@ -17,8 +19,11 @@ public class PlayerDataManager { private final EventBus eventBus; private final ConcurrentHashMap cache = new ConcurrentHashMap<>(); private final ConcurrentHashMap locks = new ConcurrentHashMap<>(); + private final ConcurrentHashMap> tracked = new ConcurrentHashMap<>(); private final boolean autoPersist; + private record TrackedLeaderboard(PlayerField field, ToDoubleFunction scorer) {} + public PlayerDataManager(DataStorage storage, DataFormat format, EventBus eventBus) { this(storage, format, eventBus, true); } @@ -99,6 +104,48 @@ void persist(UUID player) { byte[] bytes = container.serialize(format); storage.save("players", player.toString(), bytes); container.markPersisted(bytes); + updateLeaderboards(player, container); + } + } + + // ---- Leaderboard indexing ---------------------------------------------- + + private LeaderboardIndex leaderboardIndex() { + return storage instanceof LeaderboardIndex index ? index : null; + } + + boolean isLeaderboardTracked(String fullKey) { + return tracked.containsKey(fullKey); + } + + public void trackLeaderboard(PlayerField field, ToDoubleFunction scorer) { + tracked.put(field.fullKey(), new TrackedLeaderboard<>(field, scorer)); + } + + private void updateLeaderboards(UUID player, DataContainer container) { + LeaderboardIndex index = leaderboardIndex(); + if (index == null || tracked.isEmpty()) return; + for (TrackedLeaderboard t : tracked.values()) { + // Only index a field once it is materialised this session, so we never write a + // default score over a real one for a field that was never touched. + if (container.has(t.field().fullKey())) { + index.updateScore(t.field().fullKey(), player.toString(), score(t, container)); + } + } + } + + @SuppressWarnings("unchecked") + private double score(TrackedLeaderboard t, DataContainer container) { + return t.scorer().applyAsDouble((T) container.get(t.field())); + } + + /** Backfills the index for a tracked field by scanning existing stored players once. */ + public void rebuildLeaderboard(PlayerField field, ToDoubleFunction scorer) { + LeaderboardIndex index = leaderboardIndex(); + if (index == null) return; + for (String id : storage.listIds("players")) { + UUID player = UUID.fromString(id); + index.updateScore(field.fullKey(), id, scorer.applyAsDouble(getFieldValue(player, field))); } } diff --git a/src/main/java/net/swofty/storage/InMemoryDataStorage.java b/src/main/java/net/swofty/storage/InMemoryDataStorage.java index 64842e8..eff7be4 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,42 @@ 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 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..7a7253a --- /dev/null +++ b/src/main/java/net/swofty/storage/LeaderboardIndex.java @@ -0,0 +1,32 @@ +package net.swofty.storage; + +import java.util.List; + +/** + * An optional capability a {@link DataStorage} may implement to maintain sorted indexes + * (e.g. Redis sorted sets) for leaderboard fields. When present, {@code getTop}/{@code getTopPaged} + * for a tracked field read the ranked slice directly instead of scanning and deserializing every + * stored entity, turning an O(N) query into O(log N + page). When absent, the API falls back to + * the scan-based implementation, so this is purely additive. + */ +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. */ + void updateScore(String leaderboard, String id, double score); + + /** 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..d26b074 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,48 @@ public boolean exists(String type, String id) { } } + // ---- LeaderboardIndex (Redis sorted sets) ------------------------------- + + private String leaderboardKey(String leaderboard) { + return prefix + ":lb:" + leaderboard; + } + + @Override + public void updateScore(String leaderboard, String id, double score) { + try (Jedis jedis = pool.getResource()) { + jedis.zadd(leaderboardKey(leaderboard), score, id); + } + } + + @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/LeaderboardIndexTest.java b/src/test/java/net/swofty/LeaderboardIndexTest.java new file mode 100644 index 0000000..c4ab0c7 --- /dev/null +++ b/src/test/java/net/swofty/LeaderboardIndexTest.java @@ -0,0 +1,77 @@ +package net.swofty; + +import net.swofty.api.DataAPIImpl; +import net.swofty.codec.Codecs; +import net.swofty.storage.InMemoryDataStorage; + +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * The indexed leaderboard path (Redis sorted sets in production, an in-memory index here) + * must return the same ranking as the scan-based fallback, without scanning every player. + */ +class LeaderboardIndexTest { + + private static final PlayerField COINS = PlayerField.create("game", "coins", Codecs.INT, 0); + + @Test + void indexedGetTopMatchesInsertionRanking() { + DataAPIImpl api = new DataAPIImpl(new InMemoryDataStorage()); + api.trackLeaderboard(COINS, Integer::doubleValue); + + 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); + 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()); + assertEquals(3, top.get(2).rank()); + api.shutdown(); + } + + @Test + void indexReflectsUpdatesAndPaging() { + DataAPIImpl api = new DataAPIImpl(new InMemoryDataStorage()); + api.trackLeaderboard(COINS, Integer::doubleValue); + + UUID a = UUID.randomUUID(), b = UUID.randomUUID(); + api.set(a, COINS, 10); + api.set(b, COINS, 20); + api.update(a, COINS, c -> c + 100); // a jumps to 110, overtaking b + + assertEquals(a, api.getTop(COINS, 1).get(0).playerId()); + + Page> page = api.getTopPaged(COINS, 1, 1); + assertEquals(2, page.totalElements()); + assertEquals(2, page.totalPages()); + assertEquals(a, page.content().get(0).playerId()); + api.shutdown(); + } + + @Test + void rebuildBackfillsExistingData() { + InMemoryDataStorage storage = new InMemoryDataStorage(); + + // Data written before the leaderboard was tracked. + 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); + api.trackLeaderboard(COINS, Integer::doubleValue); + api.rebuildLeaderboard(COINS, Integer::doubleValue); + + assertEquals(List.of(b, a), api.getTop(COINS, 2).stream().map(LeaderboardEntry::playerId).toList()); + api.shutdown(); + } +} From 4c778bc10bb68418b2395056d47b11a00e003d8a Mon Sep 17 00:00:00 2001 From: Swofty-Developments Date: Thu, 23 Jul 2026 23:40:28 +1000 Subject: [PATCH 5/8] docs: document multi-server lifecycle, locking and indexed leaderboards --- README.md | 79 ++++++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 78 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index fc09a05..eac48ce 100644 --- a/README.md +++ b/README.md @@ -367,12 +367,89 @@ 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 + +`getTop`/`getTopPaged` scan every stored player by default. When the backend maintains sorted +indexes (Redis sorted sets, or the in-memory index for single-node/tests), track a field so the +ranked slice is read directly instead: + +```java +api.trackLeaderboard(COINS, Integer::doubleValue); // maintain an index on every write +api.rebuildLeaderboard(COINS, Integer::doubleValue); // one-time backfill of pre-existing data + +api.getTop(COINS, 10); // now O(log N + page) instead of an O(N) scan +api.getTopPaged(COINS, 1, 50); +``` + +Untracked fields and backends without the capability fall through to the scan, so this is purely +an accelerator. + ## 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: From d0bd537cf875773f206c3d5ca2ddac9acee46602 Mon Sep 17 00:00:00 2001 From: Swofty-Developments Date: Thu, 23 Jul 2026 23:48:53 +1000 Subject: [PATCH 6/8] refactor: make leaderboards index-only instead of falling back to a scan The indexed leaderboard previously kept a scan-based path and chose it whenever the field was not tracked or the storage lacked an index, so forgetting to register a leaderboard silently restored the O(N) scan the index was meant to replace. Leaderboards are now index-only: getTop and getTopPaged require a registered field and read the sorted index directly, and both trackLeaderboard (on a storage without an index) and ranking an unregistered field fail fast with a clear error. getTop(field, limit, comparator) remains as the explicit scan-based escape hatch for ad-hoc custom orderings. --- README.md | 16 +++---- src/main/java/net/swofty/DataAPI.java | 7 ++-- .../net/swofty/api/BulkOperationExecutor.java | 42 ++++++------------- .../net/swofty/api/PlayerDataManager.java | 5 +++ .../net/swofty/storage/LeaderboardIndex.java | 10 ++--- .../java/net/swofty/BulkOperationsTest.java | 11 ++--- src/test/java/net/swofty/DataHandlerTest.java | 1 + .../java/net/swofty/LeaderboardIndexTest.java | 23 ++++++++++ 8 files changed, 63 insertions(+), 52 deletions(-) diff --git a/README.md b/README.md index eac48ce..e7edf5d 100644 --- a/README.md +++ b/README.md @@ -429,20 +429,22 @@ released by its owner and a lease bounds a crashed holder. ## Indexed Leaderboards -`getTop`/`getTopPaged` scan every stored player by default. When the backend maintains sorted -indexes (Redis sorted sets, or the in-memory index for single-node/tests), track a field so the -ranked slice is read directly instead: +Leaderboards are always index-backed — there is no silent full-table scan. Register the field once +(which requires a `LeaderboardIndex`-capable storage such as `RedisDataStorage`, backed by sorted +sets, or `InMemoryDataStorage` for single-node/tests); every write then maintains the index and +ranking reads only the requested slice: ```java -api.trackLeaderboard(COINS, Integer::doubleValue); // maintain an index on every write +api.trackLeaderboard(COINS, Integer::doubleValue); // required; throws if the storage can't index api.rebuildLeaderboard(COINS, Integer::doubleValue); // one-time backfill of pre-existing data -api.getTop(COINS, 10); // now O(log N + page) instead of an O(N) scan +api.getTop(COINS, 10); // O(log N + page) api.getTopPaged(COINS, 1, 50); ``` -Untracked fields and backends without the capability fall through to the scan, so this is purely -an accelerator. +`getTop`/`getTopPaged` on a field that was never tracked throw `IllegalStateException` rather than +quietly falling back to an O(N) scan. If you genuinely need an ad-hoc full sort by a custom +ordering, `getTop(field, limit, comparator)` is the explicit scan-based escape hatch. ## Lifecycle diff --git a/src/main/java/net/swofty/DataAPI.java b/src/main/java/net/swofty/DataAPI.java index c3e9c41..dd47eba 100644 --- a/src/main/java/net/swofty/DataAPI.java +++ b/src/main/java/net/swofty/DataAPI.java @@ -73,9 +73,10 @@ public interface DataAPI { > List> getTopLinked(LinkedField field, int limit); List queryLinked(LinkedField field, Predicate filter); - // Leaderboard indexing - when the storage backend maintains sorted indexes (e.g. Redis - // sorted sets), track a field so getTop/getTopPaged read a ranked slice instead of scanning - // every stored player. rebuildLeaderboard backfills the index from existing data once. + // Leaderboard indexing - a leaderboard field MUST be registered here (which requires a + // LeaderboardIndex-capable storage, e.g. Redis sorted sets); getTop/getTopPaged then read a + // ranked slice from the index and throw for an unregistered field rather than scanning every + // stored player. rebuildLeaderboard backfills the index from existing data once. void trackLeaderboard(PlayerField field, ToDoubleFunction scorer); void rebuildLeaderboard(PlayerField field, ToDoubleFunction scorer); diff --git a/src/main/java/net/swofty/api/BulkOperationExecutor.java b/src/main/java/net/swofty/api/BulkOperationExecutor.java index 8be0585..16163e0 100644 --- a/src/main/java/net/swofty/api/BulkOperationExecutor.java +++ b/src/main/java/net/swofty/api/BulkOperationExecutor.java @@ -24,20 +24,18 @@ public BulkOperationExecutor(PlayerDataManager playerData, LinkedDataManager lin } public > List> getTop(PlayerField field, int limit) { - LeaderboardIndex index = indexFor(field); - if (index != null) { - return fromIndex(field, index, 0, limit - 1); - } - return getTop(field, limit, Comparator.reverseOrder()); + return fromIndex(field, requireIndex(field), 0, limit - 1); } - // The sorted index only exists when the backend maintains one AND the field is tracked; - // otherwise we fall back to the scan below, so this is purely an accelerator. - private LeaderboardIndex indexFor(PlayerField field) { - if (storage instanceof LeaderboardIndex index && playerData.isLeaderboardTracked(field.fullKey())) { - return index; + // A leaderboard is always index-backed. The field must be registered with trackLeaderboard, + // which also guarantees the storage maintains the sorted index — so there is no silent + // full-table scan hiding behind a forgotten registration. + private LeaderboardIndex requireIndex(PlayerField field) { + if (!playerData.isLeaderboardTracked(field.fullKey())) { + throw new IllegalStateException("Leaderboard field '" + field.fullKey() + + "' must be registered with trackLeaderboard(field, scorer) before it can be ranked"); } - return null; + return (LeaderboardIndex) storage; } private List> fromIndex(PlayerField field, LeaderboardIndex index, @@ -64,27 +62,11 @@ public List> getTop(PlayerField field, int limit, Com } public > Page> getTopPaged(PlayerField field, int page, int pageSize) { - LeaderboardIndex index = indexFor(field); - if (index != null) { - long total = index.leaderboardSize(field.fullKey()); - int totalPages = (int) Math.ceil((double) total / pageSize); - int start = (page - 1) * pageSize; - List> content = fromIndex(field, index, start, start + pageSize - 1); - return new Page<>(content, page, totalPages, total); - } - List> entries = getAllPlayerValues(field); - entries.sort((a, b) -> b.getValue().compareTo(a.getValue())); - - long total = entries.size(); + LeaderboardIndex index = requireIndex(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/PlayerDataManager.java b/src/main/java/net/swofty/api/PlayerDataManager.java index 7a40811..5b6efd0 100644 --- a/src/main/java/net/swofty/api/PlayerDataManager.java +++ b/src/main/java/net/swofty/api/PlayerDataManager.java @@ -119,6 +119,11 @@ boolean isLeaderboardTracked(String fullKey) { } public void trackLeaderboard(PlayerField field, ToDoubleFunction scorer) { + if (leaderboardIndex() == null) { + throw new IllegalStateException("Storage " + storage.getClass().getSimpleName() + + " does not maintain a leaderboard index; use a LeaderboardIndex-capable storage" + + " (e.g. RedisDataStorage or InMemoryDataStorage)"); + } tracked.put(field.fullKey(), new TrackedLeaderboard<>(field, scorer)); } diff --git a/src/main/java/net/swofty/storage/LeaderboardIndex.java b/src/main/java/net/swofty/storage/LeaderboardIndex.java index 7a7253a..22af7de 100644 --- a/src/main/java/net/swofty/storage/LeaderboardIndex.java +++ b/src/main/java/net/swofty/storage/LeaderboardIndex.java @@ -3,11 +3,11 @@ import java.util.List; /** - * An optional capability a {@link DataStorage} may implement to maintain sorted indexes - * (e.g. Redis sorted sets) for leaderboard fields. When present, {@code getTop}/{@code getTopPaged} - * for a tracked field read the ranked slice directly instead of scanning and deserializing every - * stored entity, turning an O(N) query into O(log N + page). When absent, the API falls back to - * the scan-based implementation, so this is purely additive. + * 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}: + * registering a field with {@code trackLeaderboard} requires an index-capable storage, and ranking + * reads the requested slice directly ({@code O(log N + page)}) rather than scanning and + * deserializing every stored entity. A field that was never registered throws instead of scanning. */ public interface LeaderboardIndex { diff --git a/src/test/java/net/swofty/BulkOperationsTest.java b/src/test/java/net/swofty/BulkOperationsTest.java index 918cff3..2906e89 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,9 @@ class BulkOperationsTest { @BeforeEach void setUp() { - api = new DataAPIImpl(new FileDataStorage(tempDir, new JsonFormat(), ".json"), new JsonFormat()); + // InMemory storage maintains a leaderboard index; leaderboards are index-only. + api = new DataAPIImpl(new InMemoryDataStorage()); + api.trackLeaderboard(COINS, Integer::doubleValue); } @AfterEach diff --git a/src/test/java/net/swofty/DataHandlerTest.java b/src/test/java/net/swofty/DataHandlerTest.java index adaa662..007f938 100644 --- a/src/test/java/net/swofty/DataHandlerTest.java +++ b/src/test/java/net/swofty/DataHandlerTest.java @@ -32,6 +32,7 @@ class DataHandlerTest { @BeforeEach void setUp() { api = new DataAPIImpl(new InMemoryDataStorage()); + api.trackLeaderboard(COINS, Integer::doubleValue); } @AfterEach diff --git a/src/test/java/net/swofty/LeaderboardIndexTest.java b/src/test/java/net/swofty/LeaderboardIndexTest.java index c4ab0c7..31ff147 100644 --- a/src/test/java/net/swofty/LeaderboardIndexTest.java +++ b/src/test/java/net/swofty/LeaderboardIndexTest.java @@ -2,10 +2,14 @@ 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; @@ -17,8 +21,27 @@ */ class LeaderboardIndexTest { + @TempDir + Path tempDir; + private static final PlayerField COINS = PlayerField.create("game", "coins", Codecs.INT, 0); + @Test + void rankingAnUntrackedFieldFailsFast() { + DataAPIImpl api = new DataAPIImpl(new InMemoryDataStorage()); + // No trackLeaderboard call — ranking must throw instead of silently scanning every player. + assertThrows(IllegalStateException.class, () -> api.getTop(COINS, 10)); + assertThrows(IllegalStateException.class, () -> api.getTopPaged(COINS, 1, 10)); + api.shutdown(); + } + + @Test + void trackingOnStorageWithoutAnIndexFailsFast() { + DataAPIImpl api = new DataAPIImpl(new FileDataStorage(tempDir, new JsonFormat(), ".json"), new JsonFormat()); + assertThrows(IllegalStateException.class, () -> api.trackLeaderboard(COINS, Integer::doubleValue)); + api.shutdown(); + } + @Test void indexedGetTopMatchesInsertionRanking() { DataAPIImpl api = new DataAPIImpl(new InMemoryDataStorage()); From bd155109cc5c39650dc2073e1d4aa3228fa5753f Mon Sep 17 00:00:00 2001 From: Swofty-Developments Date: Fri, 24 Jul 2026 00:03:39 +1000 Subject: [PATCH 7/8] refactor: make numeric leaderboards self-registering instead of opt-in Ranking a field no longer requires a trackLeaderboard call. The first getTop/getTopPaged for a field builds its index from existing data in one scan, and from then on the index exists in shared storage so every node maintains it on write via a new updateScoreIfPresent (an atomic exists-then-add). Numeric values are scored automatically, so the common case needs zero setup; a field that is never ranked never gets an index and its writes cost nothing. trackLeaderboard now only registers a score function for ranking a non-numeric field, and rebuildLeaderboard(field) forces a rebuild. Storage without an index still throws on ranking rather than silently scanning. --- README.md | 26 +++--- src/main/java/net/swofty/DataAPI.java | 11 ++- .../net/swofty/api/BulkOperationExecutor.java | 22 +++-- src/main/java/net/swofty/api/DataAPIImpl.java | 4 +- .../net/swofty/api/PlayerDataManager.java | 75 ++++++++++----- .../swofty/storage/InMemoryDataStorage.java | 14 +++ .../net/swofty/storage/LeaderboardIndex.java | 21 ++++- .../net/swofty/storage/RedisDataStorage.java | 19 ++++ .../java/net/swofty/BulkOperationsTest.java | 3 +- src/test/java/net/swofty/DataHandlerTest.java | 1 - .../java/net/swofty/LeaderboardIndexTest.java | 93 ++++++++++--------- 11 files changed, 183 insertions(+), 106 deletions(-) diff --git a/README.md b/README.md index e7edf5d..75a0811 100644 --- a/README.md +++ b/README.md @@ -429,22 +429,26 @@ released by its owner and a lease bounds a crashed holder. ## Indexed Leaderboards -Leaderboards are always index-backed — there is no silent full-table scan. Register the field once -(which requires a `LeaderboardIndex`-capable storage such as `RedisDataStorage`, backed by sorted -sets, or `InMemoryDataStorage` for single-node/tests); every write then maintains the index and -ranking reads only the requested slice: +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.trackLeaderboard(COINS, Integer::doubleValue); // required; throws if the storage can't index -api.rebuildLeaderboard(COINS, Integer::doubleValue); // one-time backfill of pre-existing data - -api.getTop(COINS, 10); // O(log N + page) +api.getTop(COINS, 10); // just works — O(log N + page), no registration api.getTopPaged(COINS, 1, 50); ``` -`getTop`/`getTopPaged` on a field that was never tracked throw `IllegalStateException` rather than -quietly falling back to an O(N) scan. If you genuinely need an ad-hoc full sort by a custom -ordering, `getTop(field, limit, comparator)` is the explicit scan-based escape hatch. +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 diff --git a/src/main/java/net/swofty/DataAPI.java b/src/main/java/net/swofty/DataAPI.java index dd47eba..e855101 100644 --- a/src/main/java/net/swofty/DataAPI.java +++ b/src/main/java/net/swofty/DataAPI.java @@ -73,12 +73,13 @@ public interface DataAPI { > List> getTopLinked(LinkedField field, int limit); List queryLinked(LinkedField field, Predicate filter); - // Leaderboard indexing - a leaderboard field MUST be registered here (which requires a - // LeaderboardIndex-capable storage, e.g. Redis sorted sets); getTop/getTopPaged then read a - // ranked slice from the index and throw for an unregistered field rather than scanning every - // stored player. rebuildLeaderboard backfills the index from existing data once. + // 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, 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 diff --git a/src/main/java/net/swofty/api/BulkOperationExecutor.java b/src/main/java/net/swofty/api/BulkOperationExecutor.java index 16163e0..66799ba 100644 --- a/src/main/java/net/swofty/api/BulkOperationExecutor.java +++ b/src/main/java/net/swofty/api/BulkOperationExecutor.java @@ -24,18 +24,19 @@ public BulkOperationExecutor(PlayerDataManager playerData, LinkedDataManager lin } public > List> getTop(PlayerField field, int limit) { - return fromIndex(field, requireIndex(field), 0, limit - 1); + 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. The field must be registered with trackLeaderboard, - // which also guarantees the storage maintains the sorted index — so there is no silent - // full-table scan hiding behind a forgotten registration. - private LeaderboardIndex requireIndex(PlayerField field) { - if (!playerData.isLeaderboardTracked(field.fullKey())) { - throw new IllegalStateException("Leaderboard field '" + field.fullKey() - + "' must be registered with trackLeaderboard(field, scorer) before it can be ranked"); + // 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 (LeaderboardIndex) storage; + return index; } private List> fromIndex(PlayerField field, LeaderboardIndex index, @@ -62,7 +63,8 @@ public List> getTop(PlayerField field, int limit, Com } public > Page> getTopPaged(PlayerField field, int page, int pageSize) { - LeaderboardIndex index = requireIndex(field); + 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; diff --git a/src/main/java/net/swofty/api/DataAPIImpl.java b/src/main/java/net/swofty/api/DataAPIImpl.java index ae321ea..571d085 100644 --- a/src/main/java/net/swofty/api/DataAPIImpl.java +++ b/src/main/java/net/swofty/api/DataAPIImpl.java @@ -311,8 +311,8 @@ public void trackLeaderboard(PlayerField field, ToDoubleFunction score } @Override - public void rebuildLeaderboard(PlayerField field, ToDoubleFunction scorer) { - playerData.rebuildLeaderboard(field, scorer); + public void rebuildLeaderboard(PlayerField field) { + playerData.rebuildLeaderboard(field); } // ==================== Lifecycle ==================== diff --git a/src/main/java/net/swofty/api/PlayerDataManager.java b/src/main/java/net/swofty/api/PlayerDataManager.java index 5b6efd0..63159b7 100644 --- a/src/main/java/net/swofty/api/PlayerDataManager.java +++ b/src/main/java/net/swofty/api/PlayerDataManager.java @@ -19,11 +19,11 @@ public class PlayerDataManager { private final EventBus eventBus; private final ConcurrentHashMap cache = new ConcurrentHashMap<>(); private final ConcurrentHashMap locks = new ConcurrentHashMap<>(); - private final ConcurrentHashMap> tracked = 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; - private record TrackedLeaderboard(PlayerField field, ToDoubleFunction scorer) {} - public PlayerDataManager(DataStorage storage, DataFormat format, EventBus eventBus) { this(storage, format, eventBus, true); } @@ -109,48 +109,73 @@ void persist(UUID player) { } // ---- 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; } - boolean isLeaderboardTracked(String fullKey) { - return tracked.containsKey(fullKey); - } - - public void trackLeaderboard(PlayerField field, ToDoubleFunction scorer) { - if (leaderboardIndex() == null) { + private LeaderboardIndex requireLeaderboardIndex() { + LeaderboardIndex index = leaderboardIndex(); + if (index == null) { throw new IllegalStateException("Storage " + storage.getClass().getSimpleName() - + " does not maintain a leaderboard index; use a LeaderboardIndex-capable storage" + + " does not support leaderboards; use a LeaderboardIndex-capable storage" + " (e.g. RedisDataStorage or InMemoryDataStorage)"); } - tracked.put(field.fullKey(), new TrackedLeaderboard<>(field, scorer)); + 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 || tracked.isEmpty()) return; - for (TrackedLeaderboard t : tracked.values()) { - // Only index a field once it is materialised this session, so we never write a - // default score over a real one for a field that was never touched. - if (container.has(t.field().fullKey())) { - index.updateScore(t.field().fullKey(), player.toString(), score(t, container)); + 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); } } } - @SuppressWarnings("unchecked") - private double score(TrackedLeaderboard t, DataContainer container) { - return t.scorer().applyAsDouble((T) container.get(t.field())); + /** 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); } - /** Backfills the index for a tracked field by scanning existing stored players once. */ - public void rebuildLeaderboard(PlayerField field, ToDoubleFunction scorer) { - LeaderboardIndex index = leaderboardIndex(); - if (index == null) return; + /** 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); - index.updateScore(field.fullKey(), id, scorer.applyAsDouble(getFieldValue(player, field))); + 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); } } diff --git a/src/main/java/net/swofty/storage/InMemoryDataStorage.java b/src/main/java/net/swofty/storage/InMemoryDataStorage.java index eff7be4..8086fe8 100644 --- a/src/main/java/net/swofty/storage/InMemoryDataStorage.java +++ b/src/main/java/net/swofty/storage/InMemoryDataStorage.java @@ -45,6 +45,20 @@ 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); diff --git a/src/main/java/net/swofty/storage/LeaderboardIndex.java b/src/main/java/net/swofty/storage/LeaderboardIndex.java index 22af7de..87f4102 100644 --- a/src/main/java/net/swofty/storage/LeaderboardIndex.java +++ b/src/main/java/net/swofty/storage/LeaderboardIndex.java @@ -4,19 +4,30 @@ /** * 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}: - * registering a field with {@code trackLeaderboard} requires an index-capable storage, and ranking - * reads the requested slice directly ({@code O(log N + page)}) rather than scanning and - * deserializing every stored entity. A field that was never registered throws instead of scanning. + * 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. */ + /** 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); diff --git a/src/main/java/net/swofty/storage/RedisDataStorage.java b/src/main/java/net/swofty/storage/RedisDataStorage.java index d26b074..52f3cc3 100644 --- a/src/main/java/net/swofty/storage/RedisDataStorage.java +++ b/src/main/java/net/swofty/storage/RedisDataStorage.java @@ -77,6 +77,9 @@ 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()) { @@ -84,6 +87,22 @@ public void updateScore(String leaderboard, String id, double score) { } } + @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()) { diff --git a/src/test/java/net/swofty/BulkOperationsTest.java b/src/test/java/net/swofty/BulkOperationsTest.java index 2906e89..20c380b 100644 --- a/src/test/java/net/swofty/BulkOperationsTest.java +++ b/src/test/java/net/swofty/BulkOperationsTest.java @@ -22,9 +22,8 @@ class BulkOperationsTest { @BeforeEach void setUp() { - // InMemory storage maintains a leaderboard index; leaderboards are index-only. + // InMemory storage maintains a leaderboard index; numeric leaderboards need no registration. api = new DataAPIImpl(new InMemoryDataStorage()); - api.trackLeaderboard(COINS, Integer::doubleValue); } @AfterEach diff --git a/src/test/java/net/swofty/DataHandlerTest.java b/src/test/java/net/swofty/DataHandlerTest.java index 007f938..adaa662 100644 --- a/src/test/java/net/swofty/DataHandlerTest.java +++ b/src/test/java/net/swofty/DataHandlerTest.java @@ -32,7 +32,6 @@ class DataHandlerTest { @BeforeEach void setUp() { api = new DataAPIImpl(new InMemoryDataStorage()); - api.trackLeaderboard(COINS, Integer::doubleValue); } @AfterEach diff --git a/src/test/java/net/swofty/LeaderboardIndexTest.java b/src/test/java/net/swofty/LeaderboardIndexTest.java index 31ff147..ba741ee 100644 --- a/src/test/java/net/swofty/LeaderboardIndexTest.java +++ b/src/test/java/net/swofty/LeaderboardIndexTest.java @@ -16,8 +16,8 @@ import static org.junit.jupiter.api.Assertions.*; /** - * The indexed leaderboard path (Redis sorted sets in production, an in-memory index here) - * must return the same ranking as the scan-based fallback, without scanning every player. + * 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 { @@ -27,74 +27,77 @@ class LeaderboardIndexTest { private static final PlayerField COINS = PlayerField.create("game", "coins", Codecs.INT, 0); @Test - void rankingAnUntrackedFieldFailsFast() { + void numericLeaderboardWorksWithNoRegistration() { DataAPIImpl api = new DataAPIImpl(new InMemoryDataStorage()); - // No trackLeaderboard call — ranking must throw instead of silently scanning every player. - assertThrows(IllegalStateException.class, () -> api.getTop(COINS, 10)); - assertThrows(IllegalStateException.class, () -> api.getTopPaged(COINS, 1, 10)); - api.shutdown(); - } - - @Test - void trackingOnStorageWithoutAnIndexFailsFast() { - DataAPIImpl api = new DataAPIImpl(new FileDataStorage(tempDir, new JsonFormat(), ".json"), new JsonFormat()); - assertThrows(IllegalStateException.class, () -> api.trackLeaderboard(COINS, Integer::doubleValue)); - api.shutdown(); - } - - @Test - void indexedGetTopMatchesInsertionRanking() { - DataAPIImpl api = new DataAPIImpl(new InMemoryDataStorage()); - api.trackLeaderboard(COINS, Integer::doubleValue); - 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); + 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()); - assertEquals(3, top.get(2).rank()); api.shutdown(); } @Test - void indexReflectsUpdatesAndPaging() { - DataAPIImpl api = new DataAPIImpl(new InMemoryDataStorage()); - api.trackLeaderboard(COINS, Integer::doubleValue); + void firstRankBackfillsExistingDataThenWritesKeepItCurrent() { + InMemoryDataStorage storage = new InMemoryDataStorage(); + // Data written before any ranking happened. + DataAPIImpl seed = new DataAPIImpl(storage); UUID a = UUID.randomUUID(), b = UUID.randomUUID(); - api.set(a, COINS, 10); - api.set(b, COINS, 20); - api.update(a, COINS, c -> c + 100); // a jumps to 110, overtaking b + 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(); + } - Page> page = api.getTopPaged(COINS, 1, 1); - assertEquals(2, page.totalElements()); - assertEquals(2, page.totalPages()); - assertEquals(a, page.content().get(0).playerId()); + @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 rebuildBackfillsExistingData() { - InMemoryDataStorage storage = new InMemoryDataStorage(); + void nonNumericFieldRequiresAScorer() { + PlayerField NAME = PlayerField.create("game", "name", Codecs.STRING, ""); + DataAPIImpl api = new DataAPIImpl(new InMemoryDataStorage()); + api.set(UUID.randomUUID(), NAME, "abc"); - // Data written before the leaderboard was tracked. - DataAPIImpl seed = new DataAPIImpl(storage); - UUID a = UUID.randomUUID(), b = UUID.randomUUID(); - seed.set(a, COINS, 5); - seed.set(b, COINS, 9); - seed.shutdown(); + // Cannot auto-score a String; ranking it without a scorer fails with a clear message. + assertThrows(IllegalStateException.class, () -> api.getTop(NAME, 5)); - DataAPIImpl api = new DataAPIImpl(storage); - api.trackLeaderboard(COINS, Integer::doubleValue); - api.rebuildLeaderboard(COINS, Integer::doubleValue); + // 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(); + } - assertEquals(List.of(b, a), api.getTop(COINS, 2).stream().map(LeaderboardEntry::playerId).toList()); + @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(); } } From 9b83ec53bc1036caa1dc0bdc853039e96428846b Mon Sep 17 00:00:00 2001 From: Swofty-Developments Date: Fri, 24 Jul 2026 00:08:34 +1000 Subject: [PATCH 8/8] refactor: stop exporting internal impl classes as public API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The net.swofty.api package held its plumbing (PlayerDataManager, LinkedDataManager, DataContainer, BulkOperationExecutor, TransactionManager, ExpirationManager, LinkRegistryImpl, Validation) as public classes, so with no module descriptor they were importable from the published jar and invited callers to bypass DataAPI and reach into internals. Nothing outside the package references them — only DataAPIImpl is used externally — so they are now package-private. DataAPIImpl and the DataAPI interface are unchanged; this only removes accidental public surface and drops the internals from the generated Javadoc. --- src/main/java/net/swofty/api/BulkOperationExecutor.java | 2 +- src/main/java/net/swofty/api/DataContainer.java | 2 +- src/main/java/net/swofty/api/ExpirationManager.java | 2 +- src/main/java/net/swofty/api/LinkRegistryImpl.java | 2 +- src/main/java/net/swofty/api/LinkedDataManager.java | 2 +- src/main/java/net/swofty/api/PlayerDataManager.java | 3 ++- src/main/java/net/swofty/api/TransactionManager.java | 2 +- src/main/java/net/swofty/api/Validation.java | 2 +- 8 files changed, 9 insertions(+), 8 deletions(-) diff --git a/src/main/java/net/swofty/api/BulkOperationExecutor.java b/src/main/java/net/swofty/api/BulkOperationExecutor.java index 66799ba..9c1c6db 100644 --- a/src/main/java/net/swofty/api/BulkOperationExecutor.java +++ b/src/main/java/net/swofty/api/BulkOperationExecutor.java @@ -9,7 +9,7 @@ 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; diff --git a/src/main/java/net/swofty/api/DataContainer.java b/src/main/java/net/swofty/api/DataContainer.java index bff796f..62f0226 100644 --- a/src/main/java/net/swofty/api/DataContainer.java +++ b/src/main/java/net/swofty/api/DataContainer.java @@ -19,7 +19,7 @@ * makes a partial write safe: previously an untouched field would be silently * dropped the first time any other field was persisted. */ -public class DataContainer { +class DataContainer { private final ConcurrentHashMap data = new ConcurrentHashMap<>(); // Fields explicitly cleared (set to null) this session. Suppressed from the merged 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 d7ca878..9e00662 100644 --- a/src/main/java/net/swofty/api/LinkedDataManager.java +++ b/src/main/java/net/swofty/api/LinkedDataManager.java @@ -11,7 +11,7 @@ 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; diff --git a/src/main/java/net/swofty/api/PlayerDataManager.java b/src/main/java/net/swofty/api/PlayerDataManager.java index 63159b7..ce4cc3d 100644 --- a/src/main/java/net/swofty/api/PlayerDataManager.java +++ b/src/main/java/net/swofty/api/PlayerDataManager.java @@ -13,7 +13,8 @@ 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; diff --git a/src/main/java/net/swofty/api/TransactionManager.java b/src/main/java/net/swofty/api/TransactionManager.java index b4a067b..af27ab0 100644 --- a/src/main/java/net/swofty/api/TransactionManager.java +++ b/src/main/java/net/swofty/api/TransactionManager.java @@ -13,7 +13,7 @@ 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; 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")