Multi-server data lifecycle, cache coherency, distributed locking and indexed leaderboards

README.md

@@ -367,12 +367,95 @@ new BinaryFormat() // compact, good for production
 
 Both implement `DataFormat` and can be used with any storage backend.
 
+## Multi-Server Lifecycle
+
+Across a fleet of servers a player (or a shared entity such as an island) is authoritative on
+exactly one node at a time. The lifecycle API lets a node warm that data into its cache before
+use and evict it afterwards, so a later visit to any node is never served a stale value.
+
+```java
+// Warm a player's whole document into this node in a single storage read.
+api.load(player);                       // synchronous
+api.loadAsync(player, executor);        // CompletableFuture<Void>
+api.isLoaded(player);                   // boolean
+
+// Persist pending changes and drop the player from this node's cache.
+api.unload(player);                     // flush + evict
+api.flush(player);                      // flush without evicting
+
+// Shared/linked entities have the same lifecycle.
+api.loadLink(ISLAND, islandId);
+api.unloadLink(ISLAND, islandId);
+```
+
+This is the primitive a proxy uses to implement "load the player's data on the target server
+*before* moving them there": the proxy asks the destination to `load(player)`, waits for the
+ack, then connects the player. Because the origin calls `unload(player)` on disconnect, the
+destination always starts from fresh storage.
+
+**Deferred persistence.** By default every write is flushed immediately. Pass `autoPersist = false`
+to buffer a whole play session in the cache and write it back once, on `flush`/`unload`
+(`shutdown` flushes everything so nothing is lost):
+
+```java
+DataAPI api = new DataAPIImpl(storage, new JsonFormat(), pubSub, /* autoPersist */ false);
+```
+
+**Cache coherency.** With a distributed event bus, a change made on another node to an entity that
+is currently loaded here updates the local view in place, so subscribed fields never go stale
+while a player is online. Eviction on `unload` handles the general case.
+
+## Distributed Locking
+
+Transactions guard a JVM-local lock by default, which only serialises threads within one process.
+Supply a `DistributedLock` to get true cross-node mutual exclusion — required when the same shared
+entity can be mutated from more than one server (e.g. a coop bank):
+
+```java
+DistributedLock lock = new RedisDistributedLock(jedisPool);   // or InMemoryDistributedLock for one node
+DataAPI api = new DataAPIImpl(storage, new JsonFormat(), pubSub, true, lock);
+
+// Transactions now take a cross-node lock keyed by the entity, in addition to the local monitor.
+api.transactionDirect(coopId, COOP, tx -> { tx.update(BANK, b -> b - 1000L); return null; });
+
+// The same primitive is available for app-level critical sections:
+try (var handle = ((DataAPIImpl) api).lock("coop-transfer:" + coopId, Duration.ofSeconds(5))) {
+    // ... multi-entity critical section ...
+}
+```
+
+The Redis implementation uses `SET NX PX` with a compare-and-delete release, so a lock is only
+released by its owner and a lease bounds a crashed holder.
+
+## Indexed Leaderboards
+
+Leaderboards are index-backed and **self-registering — no setup for numeric fields**. The first time
+you rank a field its index is built from existing data in one scan; from then on every node maintains
+it on write, and ranking reads only the requested slice. This requires a `LeaderboardIndex`-capable
+storage (`RedisDataStorage`, backed by sorted sets, or `InMemoryDataStorage` for single-node/tests):
+
+```java
+api.getTop(COINS, 10);         // just works — O(log N + page), no registration
+api.getTopPaged(COINS, 1, 50);
+```
+
+Only a **non-numeric** field needs a score function, since a sorted set ranks by a number:
+
+```java
+api.trackLeaderboard(NAME, String::length); // rank players by name length
+```
+
+`rebuildLeaderboard(field)` forces a rebuild from stored data if you ever need it. Storage backends
+that don't maintain an index (e.g. `FileDataStorage`) throw on ranking rather than silently scanning;
+`getTop(field, limit, comparator)` remains as the explicit scan-based escape hatch for ad-hoc custom
+orderings.
+
 ## Lifecycle
 
 Always shut down the API when done:
 
 ```java
-api.shutdown(); // stops expiration timers, closes Pub/Sub subscribers
+api.shutdown(); // flushes deferred writes, stops expiration timers, closes Pub/Sub subscribers
 ```
 
 For Redis storage, also close the storage:

src/main/java/net/swofty/DataAPI.java

@@ -6,7 +6,10 @@
 
 import java.time.Duration;
 import java.util.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
 import java.util.function.Predicate;
+import java.util.function.ToDoubleFunction;
 import java.util.function.UnaryOperator;
 
 public interface DataAPI {
@@ -69,4 +72,28 @@ public interface DataAPI {
     // Bulk operations - Linked
     <K, T extends Comparable<T>> List<LeaderboardEntry<T>> getTopLinked(LinkedField<K, T> field, int limit);
     <K, T> List<K> queryLinked(LinkedField<K, T> field, Predicate<T> filter);
+
+    // Leaderboard indexing - getTop/getTopPaged are index-backed and require a LeaderboardIndex-
+    // capable storage (e.g. Redis sorted sets). No registration is needed for numeric fields: the
+    // index self-builds on first rank and every node maintains it on write. trackLeaderboard only
+    // registers a score function so a NON-numeric field can be ranked; rebuildLeaderboard forces a
+    // rebuild from stored data.
+    <T> void trackLeaderboard(PlayerField<T> field, ToDoubleFunction<T> scorer);
+    <T> void rebuildLeaderboard(PlayerField<T> field);
+
+    // Lifecycle - warm a player's data into this node before use, evict it when done.
+    // This is the primitive a proxy uses to load a player's data on the target server
+    // BEFORE moving them there, and to evict it afterwards so a later visit is never stale.
+    void load(UUID player);
+    CompletableFuture<Void> loadAsync(UUID player, Executor executor);
+    void flush(UUID player);
+    void unload(UUID player);
+    boolean isLoaded(UUID player);
+    Set<UUID> loadedPlayers();
+
+    // Lifecycle - shared/linked entities (e.g. an island or coop shared across members)
+    <K> void loadLink(LinkType<K> type, K key);
+    <K> void flushLink(LinkType<K> type, K key);
+    <K> void unloadLink(LinkType<K> type, K key);
+    <K> boolean isLinkLoaded(LinkType<K> type, K key);
 }

src/main/java/net/swofty/api/BulkOperationExecutor.java

@@ -3,12 +3,13 @@
 import net.swofty.*;
 import net.swofty.event.EventBus;
 import net.swofty.storage.DataStorage;
+import net.swofty.storage.LeaderboardIndex;
 
 import java.util.*;
 import java.util.function.Predicate;
 import java.util.function.UnaryOperator;
 
-public class BulkOperationExecutor {
+class BulkOperationExecutor {
     private final PlayerDataManager playerData;
     private final LinkedDataManager linkedData;
     private final DataStorage storage;
@@ -23,7 +24,31 @@ public BulkOperationExecutor(PlayerDataManager playerData, LinkedDataManager lin
     }
 
     public <T extends Comparable<T>> List<LeaderboardEntry<T>> getTop(PlayerField<T> field, int limit) {
-        return getTop(field, limit, Comparator.reverseOrder());
+        LeaderboardIndex index = requireIndex();
+        playerData.ensureLeaderboardBuilt(field); // self-registers on first use, no-op thereafter
+        return fromIndex(field, index, 0, limit - 1);
+    }
+
+    // A leaderboard is always index-backed. There is no silent full-table scan: the index is built
+    // once on first use and maintained on every write, so ranking is O(log N + page).
+    private LeaderboardIndex requireIndex() {
+        if (!(storage instanceof LeaderboardIndex index)) {
+            throw new IllegalStateException("Storage " + storage.getClass().getSimpleName()
+                    + " does not support leaderboards; use a LeaderboardIndex-capable storage (e.g. RedisDataStorage)");
+        }
+        return index;
+    }
+
+    private <T> List<LeaderboardEntry<T>> fromIndex(PlayerField<T> field, LeaderboardIndex index,
+                                                    int start, int endInclusive) {
+        List<LeaderboardIndex.ScoreEntry> range = index.scoreRange(field.fullKey(), start, endInclusive, true);
+        List<LeaderboardEntry<T>> 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 <T> List<LeaderboardEntry<T>> getTop(PlayerField<T> field, int limit, Comparator<T> comparator) {
@@ -38,19 +63,12 @@ public <T> List<LeaderboardEntry<T>> getTop(PlayerField<T> field, int limit, Com
     }
 
     public <T extends Comparable<T>> Page<LeaderboardEntry<T>> getTopPaged(PlayerField<T> field, int page, int pageSize) {
-        List<Map.Entry<UUID, T>> entries = getAllPlayerValues(field);
-        entries.sort((a, b) -> b.getValue().compareTo(a.getValue()));
-
-        long total = entries.size();
+        LeaderboardIndex index = requireIndex();
+        playerData.ensureLeaderboardBuilt(field);
+        long total = index.leaderboardSize(field.fullKey());
         int totalPages = (int) Math.ceil((double) total / pageSize);
         int start = (page - 1) * pageSize;
-        int end = Math.min(start + pageSize, entries.size());
-
-        List<LeaderboardEntry<T>> content = new ArrayList<>();
-        for (int i = start; i < end; i++) {
-            Map.Entry<UUID, T> e = entries.get(i);
-            content.add(new LeaderboardEntry<>(e.getKey(), e.getValue(), i + 1));
-        }
+        List<LeaderboardEntry<T>> content = fromIndex(field, index, start, start + pageSize - 1);
         return new Page<>(content, page, totalPages, total);
     }
 

src/main/java/net/swofty/api/DataAPIImpl.java

@@ -4,13 +4,17 @@
 import net.swofty.data.DataFormat;
 import net.swofty.data.format.JsonFormat;
 import net.swofty.event.*;
+import net.swofty.lock.DistributedLock;
 import net.swofty.storage.DataStorage;
 import net.swofty.transaction.TransactionConsumer;
 import net.swofty.transaction.TransactionFunction;
 
 import java.time.Duration;
 import java.util.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
 import java.util.function.Predicate;
+import java.util.function.ToDoubleFunction;
 import java.util.function.UnaryOperator;
 
 public class DataAPIImpl implements DataAPI {
@@ -22,16 +26,50 @@ public class DataAPIImpl implements DataAPI {
     private final TransactionManager transactionManager;
     private final EventBus eventBus;
     private final BulkOperationExecutor bulkOperations;
+    private final DistributedLock distributedLock;
 
     public DataAPIImpl(DataStorage storage, DataFormat format, PubSubHandler pubSub) {
+        this(storage, format, pubSub, true);
+    }
+
+    public DataAPIImpl(DataStorage storage, DataFormat format, PubSubHandler pubSub, boolean autoPersist) {
+        this(storage, format, pubSub, autoPersist, null);
+    }
+
+    /**
+     * @param autoPersist     when true (the default) every write is flushed to storage immediately;
+     *                        when false writes stay in the cache until {@link #flush(UUID)} or
+     *                        {@link #unload(UUID)}, letting a node batch a play session into one write.
+     * @param distributedLock when non-null, transactions and {@link #lock(String, Duration)} use it for
+     *                        cross-node mutual exclusion; when null they fall back to a JVM-local lock.
+     */
+    public DataAPIImpl(DataStorage storage, DataFormat format, PubSubHandler pubSub, boolean autoPersist,
+                       DistributedLock distributedLock) {
         this.storage = storage;
+        this.distributedLock = distributedLock;
         this.eventBus = (pubSub != null) ? new DistributedEventBus(pubSub) : new EventBus();
         this.linkRegistry = new LinkRegistryImpl();
-        this.playerData = new PlayerDataManager(storage, format, eventBus);
-        this.linkedData = new LinkedDataManager(storage, format, eventBus, linkRegistry);
+        this.playerData = new PlayerDataManager(storage, format, eventBus, autoPersist);
+        this.linkedData = new LinkedDataManager(storage, format, eventBus, linkRegistry, autoPersist);
         this.expirationManager = new ExpirationManager();
-        this.transactionManager = new TransactionManager(playerData, linkedData, linkRegistry);
+        this.transactionManager = new TransactionManager(playerData, linkedData, linkRegistry,
+                distributedLock, Duration.ofSeconds(10));
         this.bulkOperations = new BulkOperationExecutor(playerData, linkedData, storage, eventBus);
+
+        // Keep locally cached containers coherent with changes made on other nodes.
+        if (eventBus instanceof DistributedEventBus distributed) {
+            distributed.setRemoteChangeHandler(new RemoteChangeHandler() {
+                @Override
+                public <T> void onPlayerChange(DataField<T> field, UUID player, T newValue) {
+                    playerData.applyRemote(field, player, newValue);
+                }
+
+                @Override
+                public <T> void onLinkedChange(DataField<T> field, String linkTypeName, String linkKey, T newValue) {
+                    linkedData.applyRemote(linkTypeName, linkKey, field, newValue);
+                }
+            });
+        }
     }
 
     public DataAPIImpl(DataStorage storage, DataFormat format) {
@@ -267,9 +305,84 @@ public <K, T> List<K> queryLinked(LinkedField<K, T> field, Predicate<T> filter)
         return bulkOperations.queryLinked(field, filter);
     }
 
+    @Override
+    public <T> void trackLeaderboard(PlayerField<T> field, ToDoubleFunction<T> scorer) {
+        playerData.trackLeaderboard(field, scorer);
+    }
+
+    @Override
+    public <T> void rebuildLeaderboard(PlayerField<T> field) {
+        playerData.rebuildLeaderboard(field);
+    }
+
     // ==================== Lifecycle ====================
 
+    @Override
+    public void load(UUID player) {
+        playerData.load(player);
+    }
+
+    @Override
+    public CompletableFuture<Void> 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<UUID> loadedPlayers() {
+        return playerData.loadedPlayers();
+    }
+
+    @Override
+    public <K> void loadLink(LinkType<K> type, K key) {
+        linkedData.loadLinked(type.name(), key);
+    }
+
+    @Override
+    public <K> void flushLink(LinkType<K> type, K key) {
+        linkedData.flushLinked(type.name(), key);
+    }
+
+    @Override
+    public <K> void unloadLink(LinkType<K> type, K key) {
+        linkedData.unloadLinked(type.name(), key);
+    }
+
+    @Override
+    public <K> boolean isLinkLoaded(LinkType<K> type, K key) {
+        return linkedData.isLinkedLoaded(type.name(), key);
+    }
+
+    /**
+     * Acquires the configured {@link DistributedLock} for an arbitrary key, for app-level critical
+     * sections that span more than one field or entity (e.g. transferring coins between two coops).
+     * Use with try-with-resources. Requires a distributed lock to have been supplied.
+     */
+    public DistributedLock.Handle lock(String key, Duration timeout) {
+        if (distributedLock == null) {
+            throw new IllegalStateException("No DistributedLock configured on this DataAPI");
+        }
+        return distributedLock.acquire(key, timeout);
+    }
+
     public void shutdown() {
+        // Flush any deferred writes so nothing is lost on a clean shutdown.
+        playerData.flushAll();
+        linkedData.flushAll();
         expirationManager.shutdown();
         if (eventBus instanceof DistributedEventBus deb) {
             deb.shutdown();