diff --git a/.claude/known_errors.md b/.claude/known_errors.md index 6ee4932..6f588ce 100644 --- a/.claude/known_errors.md +++ b/.claude/known_errors.md @@ -15,3 +15,21 @@ - **sqlite-jdbc:** never relocate `org.sqlite` when shading (JNI symbol names). Java 25 has no 32-bit x86 port (JEP 503), so those natives can be excluded from the jar. - **Local MySQL on macOS:** the Unix socket path must be ≤ 103 chars; use a short `--socket` path and connect over TCP. +- **Cache vs SQL drift under concurrency:** "decide → update cache → queue SQL write" must be one atomic step, or two + threads (accept vs remove, two `/f best` toggles) queue writes in a different order than they changed the cache. + Fix: one lock around every state change; SQL writes queued inside it. Found by `ConsistencyTest` (random stress). +- **Don't apply an event after its write commits (thenRun):** with crossing accepts, a delayed second `Befriended` fired + after a later removal and re-linked the pair. Apply immediately under the lock instead. +- **Cross-proxy ordering:** publish events *through the SQL queue* (after the writes queued before them), otherwise a + remote proxy can delete a friendship before its insert lands. Crossing accepts on two proxies need a Redis + `SET NX PX` claim on the pair, or the loser's lagging insert resurrects a removed friendship. +- **Stale profile loads:** a friendship change for a player whose profile is still loading is missed by the load (the + write is queued behind it) — track loading players and re-read if something changed. +- **Detect races deterministically:** a `LaggingDb` executor (queue SQL tasks, run random FIFO batches) reproduces + write-timing bugs every run; random multi-threaded stress only catches them ~10% of the time. +- **Test clocks vs wall time:** RedisNetwork turns absolute expiry into a TTL with real time; tests using a fake clock + must set it to `Instant.now()` or keys expire instantly. +- **Name shadowing:** a local variable named `redis` shadows the `redis.clients...` package in fully-qualified names. +- **Jedis 8:** `JedisPooled` → `RedisClient.builder()`; `ssl(boolean)` is deprecated → `sslOptions(SslOptions.defaults())`. +- **GitHub Actions YAML:** an unquoted `run:` containing `: ` (e.g. `...:root: -D...`) is parsed as a mapping and the + whole workflow is rejected (run with 0 jobs, named after the file path). Quote such commands. diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml new file mode 100644 index 0000000..565b3a6 --- /dev/null +++ b/.github/workflows/build.yml @@ -0,0 +1,42 @@ +name: build + +on: + push: + branches: [main] + pull_request: + +permissions: + contents: read + +jobs: + test: + runs-on: ubuntu-latest + services: + mysql: + image: mysql:8.4 + env: + MYSQL_ALLOW_EMPTY_PASSWORD: "yes" + MYSQL_DATABASE: friends_test + ports: ["3306:3306"] + options: >- + --health-cmd="mysqladmin ping -h 127.0.0.1" + --health-interval=5s --health-timeout=5s --health-retries=30 + redis: + image: redis:8 + ports: ["6379:6379"] + options: >- + --health-cmd="redis-cli ping" + --health-interval=5s --health-timeout=5s --health-retries=30 + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-java@v6 + with: + distribution: temurin + java-version: "25" + - uses: gradle/actions/setup-gradle@v6 + # Runs every suite: SQLite, MySQL and real Redis (throwaway tables / key namespaces). + - run: "./gradlew build -Dfriends.mysql=127.0.0.1:3306/friends_test:root: -Dfriends.redis=127.0.0.1:6379" + - uses: actions/upload-artifact@v7 + with: + name: friends-velocity + path: velocity/build/libs/friends-velocity-*.jar diff --git a/README.md b/README.md index 90801e4..d42bd4b 100644 --- a/README.md +++ b/README.md @@ -11,8 +11,8 @@ Modules: ## Build ```bash -./gradlew :velocity:shadowJar # -> velocity/build/libs/friends-velocity-0.2.0.jar -./gradlew :common:test # SQLite; add -Dfriends.mysql=host:port/db:user:password to also run on MySQL +./gradlew :velocity:shadowJar # -> velocity/build/libs/friends-velocity-0.3.0.jar +./gradlew :common:test # see "Tests" below for the MySQL and Redis suites ``` Drop the jar in Velocity's `plugins/` folder. Velocity 4.2 itself needs Java 25. sqlite-jdbc loads a native @@ -40,14 +40,34 @@ Permission: `friends.use` — allowed unless explicitly set to false (Velocity h ## Config (`plugins/friends/config.yml`) `storage.type` `sqlite` (single proxy) or `mysql` (MySQL/MariaDB), `request-expiry-minutes` (default 5), -`max-friends` (default 5000). +`max-friends` (default 5000), and the `redis` section for multi-proxy networks. + +## Multiple proxies + +Set `storage.type: mysql` on every proxy (same database), then `redis.enabled: true` with a unique, stable +`redis.proxy-id` per proxy. Friends then work across proxies: requests, accept/deny, join/leave messages, +online status, current server, `/f list` and tab-completion all see the whole network. + +- Redis holds only short-lived shared state: who is online where (`:online`), pending requests (one key + each, expiring with the request), a heartbeat per proxy and one pub/sub channel. MySQL stays the source of truth. +- A proxy that stops heartbeating (crash, freeze) is reaped by the others after ~30-40 s and its players are shown + offline. A restarted proxy clears what its previous run left behind. +- After a Redis reconnect every proxy re-reads the shared state and re-announces its own players. ## How it works - Each online player's friends and settings are loaded with one query at login and kept in memory, so lists, notifications and tab-completion never hit the database. -- Writes update the cache immediately and run on a single ordered database thread, never on proxy threads. -- Friend requests live in memory (they expire after minutes) and are lost on proxy restart. -- Single proxy only: multiple Velocity instances would need a shared cache/pub-sub (e.g. Redis). +- Every change is an event: applied under one lock (cache + queued SQL write together, so they can't drift), then + published through the SQL queue, so other proxies only see a change once it is stored. +- Writes run on a single ordered database thread per proxy, never on proxy threads. +- Friend requests live in memory (plus Redis when enabled) and expire after minutes; they are not kept in SQL. - Tables are `friends_players` and `friends_friendships`. Data from the old Bungee/Paper tables (`players`, `friends`, `requests`, `settings`) is not migrated. + +## Tests + +`./gradlew :common:test` runs everything that needs no services (SQLite, in-memory multi-proxy, randomized +cache-vs-database checks with a deliberately slow database, a concurrency stress test). Add +`-Dfriends.mysql=host:port/db:user:password` and/or `-Dfriends.redis=host:port` to also run the MySQL and +real-Redis suites (they use throwaway tables/key namespaces). diff --git a/build.gradle.kts b/build.gradle.kts index f24458a..cbd7452 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -4,7 +4,7 @@ plugins { allprojects { group = "com.friends" - version = "0.2.0" + version = "0.3.0" repositories { mavenCentral() diff --git a/common/build.gradle.kts b/common/build.gradle.kts index 85322e1..e6fb8c5 100644 --- a/common/build.gradle.kts +++ b/common/build.gradle.kts @@ -9,11 +9,13 @@ dependencies { compileOnly("net.kyori:adventure-text-minimessage") compileOnly("net.kyori:adventure-text-serializer-legacy") compileOnly("org.slf4j:slf4j-api:2.0.17") + compileOnly("com.google.code.gson:gson:2.14.0") // Shaded into the plugin jar. implementation("com.zaxxer:HikariCP:7.1.0") implementation("org.xerial:sqlite-jdbc:3.53.4.0") implementation("com.mysql:mysql-connector-j:26.7.0") + implementation("redis.clients:jedis:8.0.1") testImplementation(platform("net.kyori:adventure-bom:5.2.0")) testImplementation("net.kyori:adventure-api") @@ -21,6 +23,7 @@ dependencies { testImplementation("net.kyori:adventure-text-serializer-legacy") testImplementation("net.kyori:adventure-text-serializer-plain") testImplementation("org.slf4j:slf4j-api:2.0.17") + testImplementation("com.google.code.gson:gson:2.14.0") testImplementation(platform("org.junit:junit-bom:6.1.3")) testImplementation("org.junit.jupiter:junit-jupiter") testRuntimeOnly("org.junit.platform:junit-platform-launcher") @@ -30,4 +33,6 @@ tasks.test { jvmArgs("--enable-native-access=ALL-UNNAMED") // sqlite-jdbc loads a native library // Opt-in MySQL run: -Dfriends.mysql=host:port/db:user:password System.getProperty("friends.mysql")?.let { systemProperty("friends.mysql", it) } + // Opt-in Redis run: -Dfriends.redis=host:port (uses a throwaway key namespace) + System.getProperty("friends.redis")?.let { systemProperty("friends.redis", it) } } diff --git a/common/src/main/java/com/friends/common/Event.java b/common/src/main/java/com/friends/common/Event.java new file mode 100644 index 0000000..ce27cb9 --- /dev/null +++ b/common/src/main/java/com/friends/common/Event.java @@ -0,0 +1,34 @@ +package com.friends.common; + +import java.time.Instant; +import java.util.List; +import java.util.UUID; + +import com.friends.common.Storage.PlayerRow; + +/** + * A change other proxies must know about. The proxy where it happens applies it, then publishes it; every other + * proxy applies the same event, so single- and multi-proxy setups share one code path ({@code Friends.apply}). + */ +public sealed interface Event { + + /** Joined the network. */ + record Joined(UUID id, Presence presence) implements Event {} + + /** Changed server or status (no notification). */ + record Updated(UUID id, Presence presence) implements Event {} + + /** Left the network from {@code proxy} (ignored if they are already on another proxy). */ + record Left(UUID id, String proxy, Instant lastSeen) implements Event {} + + record RequestSent(Request request) implements Event {} + + record RequestDenied(UUID from, UUID to) implements Event {} + + record Befriended(PlayerRow a, PlayerRow b, Instant since) implements Event {} + + record Unfriended(UUID player, List friends) implements Event {} + + /** A proxy (re)started or stopped heartbeating: forget everyone who was on it. */ + record ProxyDown(String proxy) implements Event {} +} diff --git a/common/src/main/java/com/friends/common/FriendCommand.java b/common/src/main/java/com/friends/common/FriendCommand.java index 1105695..3e5f46e 100644 --- a/common/src/main/java/com/friends/common/FriendCommand.java +++ b/common/src/main/java/com/friends/common/FriendCommand.java @@ -17,11 +17,9 @@ public final class FriendCommand { private static final int MAX_SUGGESTIONS = 100; private final Friends friends; - private final Platform platform; - public FriendCommand(Friends friends, Platform platform) { + public FriendCommand(Friends friends) { this.friends = friends; - this.platform = platform; } public CompletableFuture execute(Online s, String[] args) { @@ -83,13 +81,17 @@ private static CompletableFuture usage(Online s, String usage) { public List suggest(Online s, String[] args) { if (args.length <= 1) { String prefix = args.length == 0 ? "" : args[0]; - return filter(Stream.concat(SUBCOMMANDS.stream(), platform.onlineNames().stream().filter(n -> !n.equalsIgnoreCase(s.name()))), prefix); + return filter(Stream.concat(SUBCOMMANDS.stream(), friends.onlineNames().stream().filter(n -> !n.equalsIgnoreCase(s.name()))), prefix); } String prefix = args[args.length - 1]; String sub = args[0].toLowerCase(Locale.ROOT); if (args.length > 2) return List.of(); return switch (sub) { - case "add" -> filter(platform.onlineNames().stream().filter(n -> !n.equalsIgnoreCase(s.name()) && !isFriend(s, n)), prefix); + case "add" -> { + var friendNames = friends.friendNames(s.id()).stream().map(n -> n.toLowerCase(Locale.ROOT)).collect(java.util.stream.Collectors.toSet()); + yield filter(friends.onlineNames().stream() + .filter(n -> !n.equalsIgnoreCase(s.name()) && !friendNames.contains(n.toLowerCase(Locale.ROOT))), prefix); + } case "accept", "deny", "decline" -> filter(friends.requesterNames(s.id()).stream(), prefix); case "remove", "delete", "best", "nickname", "nick" -> filter(friends.friendNames(s.id()).stream(), prefix); case "list" -> filter(Stream.of("best"), prefix); @@ -99,12 +101,9 @@ public List suggest(Online s, String[] args) { }; } - private boolean isFriend(Online s, String name) { - return platform.player(name).map(o -> friends.isFriend(s.id(), o.id())).orElse(false); - } - private static List filter(Stream options, String prefix) { String p = prefix.toLowerCase(Locale.ROOT); - return options.filter(o -> o.toLowerCase(Locale.ROOT).startsWith(p)).limit(MAX_SUGGESTIONS).toList(); + return options.filter(o -> o.toLowerCase(Locale.ROOT).startsWith(p)).distinct() + .sorted(String.CASE_INSENSITIVE_ORDER).limit(MAX_SUGGESTIONS).toList(); } } diff --git a/common/src/main/java/com/friends/common/Friends.java b/common/src/main/java/com/friends/common/Friends.java index 6dbf41e..e17eca0 100644 --- a/common/src/main/java/com/friends/common/Friends.java +++ b/common/src/main/java/com/friends/common/Friends.java @@ -7,14 +7,17 @@ import java.time.Duration; import java.time.Instant; import java.util.Comparator; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Set; import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executor; +import java.util.function.UnaryOperator; import java.util.regex.Pattern; import java.util.stream.Stream; @@ -27,34 +30,24 @@ import net.kyori.adventure.text.Component; /** - * Friends core: rules, per-online-player cache and messaging. Reads come from the cache; writes update the cache - * first and are queued on {@code db}, which must run tasks one at a time in order (writes, then later loads). - * ponytail: single proxy only, state lives in this JVM. Multi-proxy needs a shared cache + pub/sub (e.g. Redis). + * Friends core: rules, caches and messaging. Reads come from memory; SQL writes are queued on {@code db}, which must + * run tasks one at a time in order. Anything other proxies must see is an {@link Event}: applied here via + * {@link #apply}, then published on the {@link Network}, whose other proxies apply the same event. */ -public final class Friends { +public final class Friends implements Network.Listener { public static final int PAGE_SIZE = 10; private static final Pattern NAME = Pattern.compile("[\\w.*]{1,16}"); private static final Pattern NICKNAME = Pattern.compile("[\\p{L}\\p{N}_ ]{1,16}"); - record Key(UUID from, UUID to) {} - - record Request(PlayerRow from, PlayerRow to, Instant expires) { - Key key() { - return new Key(from.id(), to.id()); - } - } - - /** Cached state of an online player. {@code audience} is also the session identity. */ + /** Cached state of a player on this proxy. {@code audience} is also the session identity. */ static final class Profile { final Audience audience; final Map friends = new ConcurrentHashMap<>(); volatile boolean notifications; - volatile Status status; Profile(Audience audience, Storage.Loaded data) { this.audience = audience; this.notifications = data.notifications(); - this.status = data.status(); data.friends().forEach(f -> friends.put(f.id(), f)); } } @@ -69,18 +62,31 @@ private interface SqlRun { private final Storage storage; private final Platform platform; + private final Network network; + private final String proxy; private final Executor db; private final Clock clock; private final Duration requestExpiry; private final int maxFriends; private final Logger log; - private final Map profiles = new ConcurrentHashMap<>(); - // ponytail: requests live in memory (they expire after minutes anyway) and are lost on restart. - private final Map requests = new ConcurrentHashMap<>(); - - public Friends(Storage storage, Platform platform, Executor db, Clock clock, Duration requestExpiry, int maxFriends, Logger log) { + // Every state change (decide, update caches, queue the SQL write, publish) happens under this lock, so SQL writes + // are queued in the same order as the cache changes. Reads stay lock-free (concurrent maps). + // ponytail: one global lock; friend commands are rare and short. Per-player locks if contention ever shows. + private final Object lock = new Object(); + private final Map profiles = new ConcurrentHashMap<>(); // players on this proxy + private final Map presence = new ConcurrentHashMap<>(); // players on any proxy + // Players whose profile is being read; a friendship change for one of them makes the read stale (re-read it). + private final Set loading = ConcurrentHashMap.newKeySet(); + private final Set stale = ConcurrentHashMap.newKeySet(); + // Pending requests, network-wide. ponytail: memory/Redis only; they expire within minutes anyway. + private final Map requests = new ConcurrentHashMap<>(); + + public Friends(Storage storage, Platform platform, Network network, Executor db, Clock clock, + Duration requestExpiry, int maxFriends, Logger log) { this.storage = storage; this.platform = platform; + this.network = network; + this.proxy = network.proxyId(); this.db = db; this.clock = clock; this.requestExpiry = requestExpiry; @@ -88,21 +94,157 @@ public Friends(Storage storage, Platform platform, Executor db, Clock clock, Dur this.log = log; } + /** Joins the network (receives the shared state). Call once, before players connect. */ + public void start() { + network.start(this); + } + + // --- events --- + + /** Queue the event's SQL writes before calling this. */ + private void emit(Event event) { + synchronized (lock) { + apply(event); + // Publish through the DB queue: other proxies see an event only once the SQL writes queued before it are + // committed (so they can't act on a friendship that isn't stored yet), and in the order we applied them. + if (network != Network.LOCAL) db.execute(() -> network.publish(event)); + } + } + + @Override + public void event(Event event) { + apply(event); + } + + @Override + public void resync(Network.Snapshot snapshot) { + synchronized (lock) { + resyncLocked(snapshot); + } + } + + private void resyncLocked(Network.Snapshot snapshot) { + // Our own players win: their Redis writes may have been lost while it was unreachable, so re-announce them. + Map mine = new HashMap<>(); + profiles.keySet().forEach(id -> { + Presence p = presence.get(id); + if (p != null && p.proxy().equals(proxy)) mine.put(id, p); + }); + presence.clear(); + snapshot.online().forEach((id, p) -> { + if (!p.proxy().equals(proxy)) presence.put(id, p); // entries claiming this proxy are stale unless in `mine` + }); + presence.putAll(mine); + requests.clear(); + snapshot.requests().forEach(r -> requests.put(r.key(), r)); + mine.forEach((id, p) -> network.publish(new Event.Updated(id, p))); + } + + /** The only place shared state changes, for events from this proxy and from others alike. */ + void apply(Event event) { + synchronized (lock) { + applyLocked(event); + } + } + + private void applyLocked(Event event) { + switch (event) { + case Event.Joined(UUID id, Presence p) -> { + presence.put(id, p); + Instant now = clock.instant(); + forFriendsOf(id, f -> f.seen(p.name(), p.prefix(), now), p.status() != Status.OFFLINE, true); + } + case Event.Updated(UUID id, Presence p) -> + presence.compute(id, (_, old) -> old == null || old.proxy().equals(p.proxy()) ? p : old); + case Event.Left(UUID id, String from, Instant lastSeen) -> { + Presence p = presence.get(id); + if (p == null || !p.proxy().equals(from) || !presence.remove(id, p)) return; // moved proxies meanwhile + forFriendsOf(id, f -> f.seen(f.name(), f.prefix(), lastSeen), p.status() != Status.OFFLINE, false); + } + case Event.RequestSent(Request r) -> { + requests.put(r.key(), r); + message(r.to().id(), Messages.received(Messages.name(r.from()), r.from().name())); + } + case Event.RequestDenied(UUID from, UUID to) -> requests.remove(new Request.Key(from, to)); + case Event.Befriended(PlayerRow a, PlayerRow b, Instant since) -> { + requests.remove(new Request.Key(a.id(), b.id())); + requests.remove(new Request.Key(b.id(), a.id())); + link(a, b, since); + link(b, a, since); + } + case Event.Unfriended(UUID player, List others) -> { + Profile me = profiles.get(player); + if (me == null) markIfLoading(player); + for (UUID other : others) { + if (me != null) me.friends.remove(other); + Profile theirs = profiles.get(other); + if (theirs != null) theirs.friends.remove(player); + else markIfLoading(other); + } + } + case Event.ProxyDown(String down) -> { + Instant now = clock.instant(); + presence.forEach((id, p) -> { + if (p.proxy().equals(down)) applyLocked(new Event.Left(id, down, now)); + }); + } + } + } + + /** Refreshes every local player's view of {@code id}, optionally announcing the join/leave. */ + // ponytail: scans this proxy's players per network join/leave; add a reverse index if that ever shows up. + private void forFriendsOf(UUID id, UnaryOperator update, boolean announce, boolean joined) { + for (Profile other : profiles.values()) { + Friend view = other.friends.computeIfPresent(id, (_, f) -> update.apply(f)); + if (view == null || !announce || !other.notifications) continue; + Component name = Messages.friendName(view); + other.audience.sendMessage(joined ? Messages.joined(name) : Messages.left(name)); + } + } + + private void markIfLoading(UUID id) { + if (loading.contains(id)) stale.add(id); + } + + private void link(PlayerRow owner, PlayerRow friend, Instant since) { + Profile p = profiles.get(owner.id()); + if (p == null) markIfLoading(owner.id()); + // putIfAbsent: crossing accepts can befriend twice; keep the first entry (and its best/nickname flags). + if (p == null || p.friends.putIfAbsent(friend.id(), + new Friend(friend.id(), friend.name(), friend.prefix(), since, false, null, friend.lastSeen())) != null) return; + p.audience.sendMessage(Messages.nowFriends(Messages.name(friend))); + } + // --- lifecycle --- - /** Loads the player's data, then announces them to online friends. Never completes exceptionally. */ + /** Loads the player's data, then announces them. Never completes exceptionally. */ public CompletableFuture connect(Online p) { Instant now = clock.instant(); + synchronized (lock) { + loading.add(p.id()); + stale.remove(p.id()); + } return db(() -> { storage.savePlayer(p.id(), p.name(), p.prefix(), now); return storage.load(p.id()); - }).thenAccept(data -> { + }).thenCompose(data -> install(p, data)) + .exceptionally(_ -> { // logged by db(); the player just has no profile until they rejoin + loading.remove(p.id()); + return null; + }); + } + + private CompletableFuture install(Online p, Storage.Loaded data) { + synchronized (lock) { + // A friendship changed while we were reading: its write is queued behind us, so read again after it. + if (stale.remove(p.id())) return db(() -> storage.load(p.id())).thenCompose(fresh -> install(p, fresh)); + loading.remove(p.id()); // Left (or was replaced by a newer session) while loading: don't resurrect a stale profile. - if (platform.player(p.id()).filter(o -> o.audience() == p.audience()).isEmpty()) return; - Profile profile = new Profile(p.audience(), data); - profiles.put(p.id(), profile); - announce(p.id(), profile, f -> f.seen(p.name(), p.prefix(), now), true); - }).exceptionally(_ -> null); // logged by db(); the player just has no profile until they rejoin + if (platform.player(p.id()).filter(o -> o.audience() == p.audience()).isEmpty()) return completedFuture(null); + profiles.put(p.id(), new Profile(p.audience(), data)); + emit(new Event.Joined(p.id(), new Presence(p.name(), p.prefix(), proxy, p.server(), data.status()))); + return completedFuture(null); + } } /** Call once the player can receive chat (first backend server): tells them about pending requests. */ @@ -111,33 +253,35 @@ public void greet(Online p) { if (pending > 0) p.audience().sendMessage(Messages.pending(pending)); } - public void disconnect(Online p) { - Profile profile = profiles.get(p.id()); - // Only the session that loaded the profile may drop it (kick-existing-players can reorder events). - if (profile == null || profile.audience != p.audience() || !profiles.remove(p.id(), profile)) return; - Instant now = clock.instant(); - write(() -> storage.touch(p.id(), now)); - announce(p.id(), profile, f -> f.seen(f.name(), f.prefix(), now), false); + /** Call after a server switch so friends see the new server. */ + public void moved(Online p) { + synchronized (lock) { + Presence old = presence.get(p.id()); + if (profiles.containsKey(p.id()) && old != null) emit(new Event.Updated(p.id(), old.withServer(p.server()))); + } } - private void announce(UUID id, Profile profile, java.util.function.UnaryOperator update, boolean joined) { - for (UUID friendId : profile.friends.keySet()) { - Profile other = profiles.get(friendId); - if (other == null) continue; // offline friends have no profile - Friend view = other.friends.computeIfPresent(id, (_, f) -> update.apply(f)); - if (view == null || !other.notifications || profile.status == Status.OFFLINE) continue; - Component name = Messages.friendName(view); - other.audience.sendMessage(joined ? Messages.joined(name) : Messages.left(name)); + public void disconnect(Online p) { + synchronized (lock) { + Profile profile = profiles.get(p.id()); + // Only the session that loaded the profile may drop it (kick-existing-players can reorder events). + if (profile == null || profile.audience != p.audience() || !profiles.remove(p.id(), profile)) return; + Instant now = clock.instant(); + write(() -> storage.touch(p.id(), now)); + emit(new Event.Left(p.id(), proxy, now)); } } - /** Called periodically by the platform; also re-checked lazily, so timing isn't critical. */ + /** Called periodically; validity is also re-checked lazily, so timing isn't critical. */ public void expireRequests() { Instant now = clock.instant(); - for (Request r : requests.values()) { - if (now.isBefore(r.expires()) || !requests.remove(r.key(), r)) continue; - message(r.from().id(), Messages.expiredOutgoing(Messages.name(r.to()))); - message(r.to().id(), Messages.expiredIncoming(Messages.name(r.from()))); + synchronized (lock) { + for (Request r : requests.values()) { + if (now.isBefore(r.expires()) || !requests.remove(r.key(), r)) continue; + // Every proxy sweeps its own copy and only tells its own players, so each player hears it once. + message(r.from().id(), Messages.expiredOutgoing(Messages.name(r.to()))); + message(r.to().id(), Messages.expiredIncoming(Messages.name(r.from()))); + } } } @@ -154,18 +298,21 @@ public CompletableFuture add(Online s, String name) { return resolve(name).thenCompose(found -> { if (found.isEmpty()) return reply(s, Messages.notFound(name)); PlayerRow target = found.get(); - Component targetName = Messages.name(target); - if (me.friends.containsKey(target.id())) return reply(s, Messages.alreadyFriends(targetName)); - Request theirs = valid(requests.get(new Key(target.id(), s.id()))); - if (theirs != null) return accept(s, me, theirs); // both asked: just make them friends - if (valid(requests.get(new Key(s.id(), target.id()))) != null) return reply(s, Messages.alreadySent(targetName)); - if (me.friends.size() >= maxFriends) return reply(s, Messages.limitSelf(maxFriends)); return friendCount(target.id()).thenCompose(count -> { - if (count >= maxFriends) return reply(s, Messages.limitOther(targetName)); - PlayerRow from = row(s); - requests.put(new Key(s.id(), target.id()), new Request(from, target, clock.instant().plus(requestExpiry))); - message(target.id(), Messages.received(Messages.name(from), s.name())); - return reply(s, Messages.sent(targetName, requestExpiry.toMinutes())); + Request theirs; + synchronized (lock) { // all checks and the send in one step + Component targetName = Messages.name(target); + if (me.friends.containsKey(target.id())) return reply(s, Messages.alreadyFriends(targetName)); + theirs = valid(requests.get(new Request.Key(target.id(), s.id()))); + if (theirs == null) { + if (valid(requests.get(new Request.Key(s.id(), target.id()))) != null) return reply(s, Messages.alreadySent(targetName)); + if (me.friends.size() >= maxFriends) return reply(s, Messages.limitSelf(maxFriends)); + if (count >= maxFriends) return reply(s, Messages.limitOther(targetName)); + emit(new Event.RequestSent(new Request(row(s), target, clock.instant().plus(requestExpiry)))); + return reply(s, Messages.sent(targetName, requestExpiry.toMinutes())); + } + } + return accept(s, me, theirs); // both asked: just make them friends (outside the lock: it may claim) }); }); } @@ -182,27 +329,31 @@ private CompletableFuture accept(Online s, Profile me, Request r) { if (me.friends.size() >= maxFriends) return reply(s, Messages.limitSelf(maxFriends)); return friendCount(other.id()).thenCompose(count -> { if (count >= maxFriends) return reply(s, Messages.limitOther(Messages.name(other))); - if (!requests.remove(r.key(), r)) return reply(s, Messages.noRequest(other.name())); // expired/raced - Instant now = clock.instant(); - me.friends.put(other.id(), new Friend(other.id(), other.name(), other.prefix(), now, false, null, other.lastSeen())); - Profile theirs = profiles.get(other.id()); - if (theirs != null) theirs.friends.put(s.id(), new Friend(s.id(), s.name(), s.prefix(), now, false, null, now)); - write(() -> storage.addFriendship(s.id(), other.id(), now)); - message(other.id(), Messages.nowFriends(Messages.name(row(s)))); - return reply(s, Messages.nowFriends(Messages.name(other))); + if (!network.claim(s.id(), other.id())) return reply(s, Messages.noRequest(other.name())); // lost a crossing accept + synchronized (lock) { + if (!requests.remove(r.key(), r)) return reply(s, Messages.noRequest(other.name())); // expired/raced + if (me.friends.containsKey(other.id())) return completedFuture(null); // crossing accepts: already done + Instant now = clock.instant(); + write(() -> storage.addFriendship(s.id(), other.id(), now)); + emit(new Event.Befriended(other, row(s), now)); // tells both sides, wherever they are + return completedFuture(null); + } }); } public CompletableFuture deny(Online s, String name) { - Request r = incoming(s.id()).filter(x -> x.from().name().equalsIgnoreCase(name)).findFirst().orElse(null); - if (r == null || !requests.remove(r.key(), r)) return reply(s, Messages.noRequest(name)); - return reply(s, Messages.declined(Messages.name(r.from()))); + synchronized (lock) { + Request r = incoming(s.id()).filter(x -> x.from().name().equalsIgnoreCase(name)).findFirst().orElse(null); + if (r == null || !requests.remove(r.key(), r)) return reply(s, Messages.noRequest(name)); + emit(new Event.RequestDenied(r.from().id(), s.id())); + return reply(s, Messages.declined(Messages.name(r.from()))); + } } public CompletableFuture remove(Online s, String name) { - return withFriend(s, name, (me, f) -> { - unfriend(s.id(), me, f.id()); + return withFriend(s, name, (_, f) -> { write(() -> storage.removeFriendships(s.id(), List.of(f.id()))); + emit(new Event.Unfriended(s.id(), List.of(f.id()))); return reply(s, Messages.removed(Messages.name(f.prefix(), f.name()))); }); } @@ -210,18 +361,14 @@ public CompletableFuture remove(Online s, String name) { public CompletableFuture removeAll(Online s, boolean confirmed) { Profile me = profiles.get(s.id()); if (me == null) return reply(s, Messages.notLoaded()); - List victims = me.friends.values().stream().filter(f -> !f.best()).map(Friend::id).toList(); - if (victims.isEmpty()) return reply(s, Messages.nothingToRemove()); - if (!confirmed) return reply(s, Messages.removeAllConfirm(victims.size())); - victims.forEach(id -> unfriend(s.id(), me, id)); - write(() -> storage.removeFriendships(s.id(), victims)); - return reply(s, Messages.removedAll(victims.size())); - } - - private void unfriend(UUID self, Profile me, UUID friend) { - me.friends.remove(friend); - Profile theirs = profiles.get(friend); - if (theirs != null) theirs.friends.remove(self); + synchronized (lock) { + List victims = me.friends.values().stream().filter(f -> !f.best()).map(Friend::id).toList(); + if (victims.isEmpty()) return reply(s, Messages.nothingToRemove()); + if (!confirmed) return reply(s, Messages.removeAllConfirm(victims.size())); + write(() -> storage.removeFriendships(s.id(), victims)); + emit(new Event.Unfriended(s.id(), victims)); + return reply(s, Messages.removedAll(victims.size())); + } } public CompletableFuture best(Online s, String name) { @@ -246,20 +393,24 @@ public CompletableFuture nickname(Online s, String name, String nickname) public CompletableFuture toggleNotifications(Online s) { Profile me = profiles.get(s.id()); if (me == null) return reply(s, Messages.notLoaded()); - boolean enabled = !me.notifications; - me.notifications = enabled; - write(() -> storage.setNotifications(s.id(), enabled)); - return reply(s, Messages.notifications(enabled)); + synchronized (lock) { + boolean enabled = !me.notifications; + me.notifications = enabled; + write(() -> storage.setNotifications(s.id(), enabled)); + return reply(s, Messages.notifications(enabled)); + } } /** {@code status} null shows the current status with clickable options. */ public CompletableFuture status(Online s, Status status) { - Profile me = profiles.get(s.id()); - if (me == null) return reply(s, Messages.notLoaded()); - if (status == null) return reply(s, Messages.statusMenu(me.status)); - me.status = status; - write(() -> storage.setStatus(s.id(), status)); - return reply(s, Messages.statusSet(status)); + synchronized (lock) { + Presence current = presence.get(s.id()); + if (!profiles.containsKey(s.id()) || current == null) return reply(s, Messages.notLoaded()); + if (status == null) return reply(s, Messages.statusMenu(current.status())); + write(() -> storage.setStatus(s.id(), status)); + emit(new Event.Updated(s.id(), current.withStatus(status))); + return reply(s, Messages.statusSet(status)); + } } public CompletableFuture list(Online s, int page, boolean bestOnly) { @@ -286,9 +437,10 @@ public CompletableFuture list(Online s, int page, boolean bestOnly) { : a.friend().name().compareToIgnoreCase(b.friend().name())); private Messages.Entry entry(Friend f) { - Profile p = profiles.get(f.id()); - if (p == null || p.status == Status.OFFLINE) return new Messages.Entry(f, null, null); - return new Messages.Entry(f, p.status, platform.player(f.id()).map(Online::server).orElse(null)); + Presence p = presence.get(f.id()); + return p == null || p.status() == Status.OFFLINE + ? new Messages.Entry(f, null, null) + : new Messages.Entry(f, p.status(), p.server()); } public CompletableFuture requests(Online s) { @@ -309,9 +461,15 @@ public List requesterNames(UUID id) { return incoming(id).map(r -> r.from().name()).toList(); } - public boolean isFriend(UUID id, UUID other) { + /** A local player's cached friends (tests compare this with the database). */ + List cachedFriends(UUID id) { Profile p = profiles.get(id); - return p != null && p.friends.containsKey(other); + return p == null ? List.of() : List.copyOf(p.friends.values()); + } + + /** Everyone visibly online on any proxy (players appearing offline are left out). */ + public List onlineNames() { + return presence.values().stream().filter(p -> p.status() != Status.OFFLINE).map(Presence::name).toList(); } // --- helpers --- @@ -323,9 +481,11 @@ private interface FriendAction { private CompletableFuture withFriend(Online s, String name, FriendAction action) { Profile me = profiles.get(s.id()); if (me == null) return reply(s, Messages.notLoaded()); - return me.friends.values().stream().filter(f -> f.name().equalsIgnoreCase(name)).findFirst() - .map(f -> action.apply(me, f)) - .orElseGet(() -> reply(s, Messages.notFriend(name))); + synchronized (lock) { // look up and change in one step + return me.friends.values().stream().filter(f -> f.name().equalsIgnoreCase(name)).findFirst() + .map(f -> action.apply(me, f)) + .orElseGet(() -> reply(s, Messages.notFriend(name))); + } } // ponytail: linear scan over all pending requests; index by receiver if request volume ever matters. @@ -341,10 +501,15 @@ private PlayerRow row(Online p) { return new PlayerRow(p.id(), p.name(), p.prefix(), clock.instant()); } + /** Online anywhere on the network first (no query), then the database. */ private CompletableFuture> resolve(String name) { if (!NAME.matcher(name).matches()) return completedFuture(Optional.empty()); - Optional online = platform.player(name); - if (online.isPresent()) return completedFuture(Optional.of(row(online.get()))); + for (var e : presence.entrySet()) { // ponytail: O(online players), fine for a typed command + Presence p = e.getValue(); + if (p.name().equalsIgnoreCase(name)) { + return completedFuture(Optional.of(new PlayerRow(e.getKey(), p.name(), p.prefix(), clock.instant()))); + } + } return db(() -> storage.player(name)); } diff --git a/common/src/main/java/com/friends/common/Messages.java b/common/src/main/java/com/friends/common/Messages.java index eb71268..305c57c 100644 --- a/common/src/main/java/com/friends/common/Messages.java +++ b/common/src/main/java/com/friends/common/Messages.java @@ -15,7 +15,6 @@ import java.time.format.DateTimeFormatter; import java.util.List; -import com.friends.common.Friends.Request; import net.kyori.adventure.text.Component; import net.kyori.adventure.text.event.ClickEvent; diff --git a/common/src/main/java/com/friends/common/Network.java b/common/src/main/java/com/friends/common/Network.java new file mode 100644 index 0000000..8211bdd --- /dev/null +++ b/common/src/main/java/com/friends/common/Network.java @@ -0,0 +1,46 @@ +package com.friends.common; + +import java.util.List; +import java.util.Map; +import java.util.UUID; + +/** Connects proxies. {@link #LOCAL} is a single proxy; {@link RedisNetwork} shares state through Redis. */ +public interface Network extends AutoCloseable { + + /** The shared state a proxy starts from (and re-syncs to after a reconnect). */ + record Snapshot(Map online, List requests) {} + + interface Listener { + /** An event from another proxy. */ + void event(Event event); + + /** Replace mirrored shared state; called on start and after every reconnect. */ + void resync(Snapshot snapshot); + } + + String proxyId(); + + /** Start delivering to {@code listener}; it receives a first {@code resync} before any event. */ + void start(Listener listener); + + /** Share an event (already applied locally) with the other proxies. */ + void publish(Event event); + + /** + * Reserve befriending {@code a} and {@code b} network-wide for a moment. False if another proxy just did: e.g. both + * accepted each other's crossing requests at once, where a lagging insert could outlive a later removal. + */ + default boolean claim(UUID a, UUID b) { + return true; // one proxy: the local lock already serialises accepts + } + + @Override + void close(); + + Network LOCAL = new Network() { + @Override public String proxyId() { return "local"; } + @Override public void start(Listener listener) {} + @Override public void publish(Event event) {} + @Override public void close() {} + }; +} diff --git a/common/src/main/java/com/friends/common/Platform.java b/common/src/main/java/com/friends/common/Platform.java index 6e5e8d7..9f0e89c 100644 --- a/common/src/main/java/com/friends/common/Platform.java +++ b/common/src/main/java/com/friends/common/Platform.java @@ -1,6 +1,5 @@ package com.friends.common; -import java.util.Collection; import java.util.Optional; import java.util.UUID; @@ -9,13 +8,9 @@ /** What the core needs from the proxy/server it runs on. */ public interface Platform { + /** A player connected to this proxy. */ Optional player(UUID id); - /** Case-insensitive exact name lookup among online players. */ - Optional player(String name); - - Collection onlineNames(); - /** * A connected player. {@code audience} doubles as the session identity; {@code prefix} is a legacy * ('&') rank prefix or null; {@code server} is null while not connected to a backend. diff --git a/common/src/main/java/com/friends/common/Presence.java b/common/src/main/java/com/friends/common/Presence.java new file mode 100644 index 0000000..7004a7d --- /dev/null +++ b/common/src/main/java/com/friends/common/Presence.java @@ -0,0 +1,13 @@ +package com.friends.common; + +/** Where an online player is on the network. {@code server} is null until they reach a backend. */ +public record Presence(String name, String prefix, String proxy, String server, Status status) { + + Presence withServer(String server) { + return new Presence(name, prefix, proxy, server, status); + } + + Presence withStatus(Status status) { + return new Presence(name, prefix, proxy, server, status); + } +} diff --git a/common/src/main/java/com/friends/common/RedisNetwork.java b/common/src/main/java/com/friends/common/RedisNetwork.java new file mode 100644 index 0000000..1117829 --- /dev/null +++ b/common/src/main/java/com/friends/common/RedisNetwork.java @@ -0,0 +1,308 @@ +package com.friends.common; + +import java.io.IOException; +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.function.Function; +import java.util.stream.Collectors; + +import org.slf4j.Logger; + +import com.friends.common.Storage.PlayerRow; +import com.google.gson.Gson; +import com.google.gson.GsonBuilder; +import com.google.gson.JsonElement; +import com.google.gson.JsonParseException; +import com.google.gson.JsonPrimitive; +import com.google.gson.JsonDeserializer; +import com.google.gson.JsonSerializer; + +import redis.clients.jedis.DefaultJedisClientConfig; +import redis.clients.jedis.JedisPubSub; +import redis.clients.jedis.RedisClient; +import redis.clients.jedis.SslOptions; +import redis.clients.jedis.params.ScanParams; +import redis.clients.jedis.params.SetParams; +import redis.clients.jedis.resps.ScanResult; + +/** + * Multi-proxy {@link Network} over Redis. Shared state: a hash of online players ({@code :online}), one key per + * pending request with a TTL ({@code :request::}), and a heartbeat key per proxy. Events go through + * one pub/sub channel. A proxy that stops heartbeating is reaped by the others (its players are marked offline). + * The SQL database stays the source of truth for friendships and must be shared by every proxy (MySQL/MariaDB). + */ +public final class RedisNetwork implements Network { + private static final String LEAVE_IF_OWNED = """ + local v = redis.call('HGET', KEYS[1], ARGV[1]) + if v and cjson.decode(v).proxy == ARGV[2] then return redis.call('HDEL', KEYS[1], ARGV[1]) end + return 0"""; + private static final Map> TYPES = Arrays.stream(Event.class.getPermittedSubclasses()) + .collect(Collectors.toMap(Class::getSimpleName, Function.identity())); + + private record Envelope(String origin, String type, JsonElement event) {} + + private final RedisClient redis; + private final String proxy; + private final String channel; + private final String onlineKey; + private final String requestPrefix; + private final String claimPrefix; + private final String proxiesKey; + private final String heartbeatPrefix; + private final Duration beatEvery; + private final Duration beatTtl; + private final Logger log; + static final Gson GSON = new GsonBuilder() + .registerTypeAdapter(Instant.class, (JsonSerializer) (i, _, _) -> new JsonPrimitive(i.toEpochMilli())) + .registerTypeAdapter(Instant.class, (JsonDeserializer) (j, _, _) -> Instant.ofEpochMilli(j.getAsLong())) + .create(); + private final ExecutorService outbound = single("friends-redis-out"); // keeps our writes + publishes in order + private final ExecutorService inbound = single("friends-redis-in"); // keeps resyncs + received events in order + final ScheduledExecutorService heartbeat = Executors.newSingleThreadScheduledExecutor( + Thread.ofPlatform().name("friends-redis-heartbeat").daemon().factory()); + private volatile Listener listener; + private volatile JedisPubSub subscription; + private volatile boolean closed; + + public RedisNetwork(String host, int port, String password, int database, boolean ssl, String namespace, + String proxyId, Logger log) throws IOException { + this(host, port, password, database, ssl, namespace, proxyId, Duration.ofSeconds(10), Duration.ofSeconds(30), log); + } + + RedisNetwork(String host, int port, String password, int database, boolean ssl, String namespace, String proxyId, + Duration beatEvery, Duration beatTtl, Logger log) throws IOException { + this.proxy = proxyId; + this.channel = namespace + ":events"; + this.onlineKey = namespace + ":online"; + this.requestPrefix = namespace + ":request:"; + this.claimPrefix = namespace + ":claim:"; + this.proxiesKey = namespace + ":proxies"; + this.heartbeatPrefix = namespace + ":proxy:"; + this.beatEvery = beatEvery; + this.beatTtl = beatTtl; + this.log = log; + var config = DefaultJedisClientConfig.builder() + .password(password == null || password.isEmpty() ? null : password) + .database(database) + .clientName("friends-" + proxyId); + if (ssl) config.sslOptions(SslOptions.defaults()); // JDK trust store, full certificate + hostname checks + this.redis = RedisClient.builder().hostAndPort(host, port).clientConfig(config.build()).build(); + try { + redis.ping(); // fail fast on a wrong host/password, like the SQL pool does + } catch (RuntimeException e) { + redis.close(); + throw new IOException("could not connect to Redis at " + host + ":" + port, e); + } + } + + private static ExecutorService single(String name) { + return Executors.newSingleThreadExecutor(Thread.ofPlatform().name(name).daemon().factory()); + } + + @Override + public String proxyId() { + return proxy; + } + + /** Blocks (up to 5s) until the first snapshot has been delivered. */ + @Override + public void start(Listener listener) { + this.listener = listener; + CountDownLatch synced = new CountDownLatch(1); + // A previous run of this proxy may have crashed: drop what it left behind, and tell the others. + outbound.execute(() -> { + try { + forgetProxy(proxy); + redis.publish(channel, encode(new Event.ProxyDown(proxy))); + } catch (RuntimeException e) { + log.warn("Friends: could not clear this proxy's stale Redis state ({})", e.getMessage()); + } + }); + Thread.ofPlatform().name("friends-redis-sub").daemon().start(() -> subscribeLoop(synced)); + heartbeat.scheduleAtFixedRate(this::beat, 0, beatEvery.toMillis(), TimeUnit.MILLISECONDS); + try { + if (!synced.await(5, TimeUnit.SECONDS)) log.warn("Friends: no Redis snapshot yet, continuing; it will sync once connected"); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + private void subscribeLoop(CountDownLatch synced) { + while (!closed) { + JedisPubSub sub = new JedisPubSub() { + @Override + public void onSubscribe(String ch, int count) { + // Subscribed first, snapshot second: nothing published in between is lost. + inbound.execute(() -> { + resync(); + synced.countDown(); + }); + } + + @Override + public void onMessage(String ch, String message) { + inbound.execute(() -> receive(message)); + } + }; + subscription = sub; + try { + redis.subscribe(sub, channel); // blocks until unsubscribed or the connection drops + } catch (RuntimeException e) { + if (closed) return; + log.warn("Friends: lost the Redis subscription, retrying in 1s ({})", e.getMessage()); + try { + Thread.sleep(1000); + } catch (InterruptedException ie) { + return; + } + } + } + } + + private void resync() { + try { + listener.resync(snapshot()); + } catch (RuntimeException e) { + log.warn("Friends: could not read the shared state from Redis ({})", e.getMessage()); + } + } + + Snapshot snapshot() { + Map online = new HashMap<>(); + redis.hgetAll(onlineKey).forEach((id, json) -> online.put(UUID.fromString(id), GSON.fromJson(json, Presence.class))); + List requests = new ArrayList<>(); + ScanParams match = new ScanParams().match(requestPrefix + "*").count(500); + String cursor = ScanParams.SCAN_POINTER_START; + do { + ScanResult page = redis.scan(cursor, match); + cursor = page.getCursor(); + if (page.getResult().isEmpty()) continue; + for (String json : redis.mget(page.getResult().toArray(String[]::new))) { + if (json != null) requests.add(GSON.fromJson(json, Request.class)); // null: expired meanwhile + } + } while (!cursor.equals(ScanParams.SCAN_POINTER_START)); + return new Snapshot(online, requests); + } + + private void receive(String message) { + try { + Envelope envelope = GSON.fromJson(message, Envelope.class); + if (proxy.equals(envelope.origin())) return; // our own publish, already applied + Class type = TYPES.get(envelope.type()); + if (type == null) { + log.warn("Friends: ignoring unknown event type {} from proxy {}", envelope.type(), envelope.origin()); + return; + } + listener.event((Event) GSON.fromJson(envelope.event(), type)); + } catch (JsonParseException | IllegalArgumentException e) { + log.warn("Friends: ignoring a malformed Redis message ({})", e.getMessage()); + } + } + + String encode(Event event) { + return GSON.toJson(new Envelope(proxy, event.getClass().getSimpleName(), GSON.toJsonTree(event))); + } + + @Override + public void publish(Event event) { + outbound.execute(() -> { + try { + store(event); // state first, so a proxy snapshotting after the message never misses it + redis.publish(channel, encode(event)); + } catch (RuntimeException e) { + log.warn("Friends: could not publish {} to Redis ({})", event.getClass().getSimpleName(), e.getMessage()); + } + }); + } + + /** First proxy to claim a pair wins for 2s, which covers a normal SQL write queue. */ + @Override + public boolean claim(UUID a, UUID b) { + String pair = a.compareTo(b) < 0 ? a + ":" + b : b + ":" + a; + return redis.set(claimPrefix + pair, proxy, SetParams.setParams().nx().px(2000)) != null; // null: already taken + } + + private void store(Event event) { + switch (event) { + case Event.Joined(UUID id, Presence p) -> redis.hset(onlineKey, id.toString(), GSON.toJson(p)); + case Event.Updated(UUID id, Presence p) -> redis.hset(onlineKey, id.toString(), GSON.toJson(p)); + case Event.Left(UUID id, String from, Instant _) -> leaveIfOwned(id.toString(), from); + case Event.RequestSent(Request r) -> redis.set(requestKey(r.from().id(), r.to().id()), GSON.toJson(r), + SetParams.setParams().px(Math.max(1, Duration.between(Instant.now(), r.expires()).toMillis()))); + case Event.RequestDenied(UUID from, UUID to) -> redis.del(requestKey(from, to)); + case Event.Befriended(PlayerRow a, PlayerRow b, Instant _) -> + redis.del(requestKey(a.id(), b.id()), requestKey(b.id(), a.id())); + case Event.Unfriended _, Event.ProxyDown _ -> {} + } + } + + private String requestKey(UUID from, UUID to) { + return requestPrefix + from + ":" + to; + } + + /** Only removes the entry if the player is still on {@code from} (they may have moved proxies). */ + private void leaveIfOwned(String id, String from) { + redis.eval(LEAVE_IF_OWNED, List.of(onlineKey), List.of(id, from)); + } + + private void forgetProxy(String id) { + redis.hgetAll(onlineKey).forEach((player, json) -> { + if (id.equals(GSON.fromJson(json, Presence.class).proxy())) leaveIfOwned(player, id); + }); + } + + /** Refreshes our heartbeat and reaps proxies whose heartbeat expired. */ + private void beat() { + try { + boolean wasAlive = redis.exists(heartbeatPrefix + proxy); + redis.set(heartbeatPrefix + proxy, "1", SetParams.setParams().px(beatTtl.toMillis())); + redis.sadd(proxiesKey, proxy); + // We missed our own deadline (e.g. a long pause): others may have reaped us, so re-announce our players. + if (!wasAlive && listener != null) inbound.execute(this::resync); + for (String other : redis.smembers(proxiesKey)) { + if (other.equals(proxy) || redis.exists(heartbeatPrefix + other)) continue; + if (redis.srem(proxiesKey, other) == 0) continue; // another proxy reaped it first + log.warn("Friends: proxy {} stopped heartbeating; marking its players offline", other); + forgetProxy(other); + redis.publish(channel, encode(new Event.ProxyDown(other))); + inbound.execute(() -> listener.event(new Event.ProxyDown(other))); + } + } catch (RuntimeException e) { + log.warn("Friends: Redis heartbeat failed ({})", e.getMessage()); + } + } + + /** Flushes pending publishes, then tells the others we are gone. Call after players have disconnected. */ + @Override + public void close() { + closed = true; + heartbeat.shutdownNow(); + outbound.shutdown(); + try { + if (!outbound.awaitTermination(5, TimeUnit.SECONDS)) log.warn("Friends: gave up flushing Redis publishes"); + forgetProxy(proxy); + redis.publish(channel, encode(new Event.ProxyDown(proxy))); + redis.del(heartbeatPrefix + proxy); + redis.srem(proxiesKey, proxy); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (RuntimeException e) { + log.warn("Friends: could not clean up this proxy's Redis state ({})", e.getMessage()); + } + JedisPubSub sub = subscription; + if (sub != null && sub.isSubscribed()) sub.unsubscribe(); + inbound.shutdownNow(); + redis.close(); + } +} diff --git a/common/src/main/java/com/friends/common/Request.java b/common/src/main/java/com/friends/common/Request.java new file mode 100644 index 0000000..84dc9c0 --- /dev/null +++ b/common/src/main/java/com/friends/common/Request.java @@ -0,0 +1,16 @@ +package com.friends.common; + +import java.time.Instant; +import java.util.UUID; + +import com.friends.common.Storage.PlayerRow; + +/** A pending friend request. */ +public record Request(PlayerRow from, PlayerRow to, Instant expires) { + + record Key(UUID from, UUID to) {} + + Key key() { + return new Key(from.id(), to.id()); + } +} diff --git a/common/src/main/java/com/friends/common/Storage.java b/common/src/main/java/com/friends/common/Storage.java index a676e6b..845533b 100644 --- a/common/src/main/java/com/friends/common/Storage.java +++ b/common/src/main/java/com/friends/common/Storage.java @@ -57,6 +57,14 @@ public static Storage mysql(String host, int port, String database, String user, c.setUsername(user); c.setPassword(password); c.setMaximumPoolSize(2); + // HikariCP's recommended MySQL settings (statement/metadata caching), minus server-side prepared statements. + c.addDataSourceProperty("cachePrepStmts", "true"); + c.addDataSourceProperty("prepStmtCacheSize", "250"); + c.addDataSourceProperty("prepStmtCacheSqlLimit", "2048"); + c.addDataSourceProperty("useLocalSessionState", "true"); + c.addDataSourceProperty("cacheResultSetMetadata", "true"); + c.addDataSourceProperty("cacheServerConfiguration", "true"); + c.addDataSourceProperty("maintainTimeStats", "false"); return new Storage(c, true); } diff --git a/common/src/test/java/com/friends/common/ConsistencyTest.java b/common/src/test/java/com/friends/common/ConsistencyTest.java new file mode 100644 index 0000000..3e68d1f --- /dev/null +++ b/common/src/test/java/com/friends/common/ConsistencyTest.java @@ -0,0 +1,187 @@ +package com.friends.common; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +import com.friends.common.TestSupport.Harness; +import com.friends.common.TestSupport.Hub; +import com.friends.common.TestSupport.LaggingDb; +import com.friends.common.TestSupport.TestPlayer; + +/** + * Throws random command sequences (and concurrent ones) at the core, then checks the invariants the cache + * optimisation relies on: every online player's cached friends equal the database, and friendships are symmetric. + */ +class ConsistencyTest { + @TempDir + Path dir; + + static final List NAMES = List.of("Ann", "Ben", "Cat", "Dan", "Eve", "Fay", "Gus", "Hal", "Ivy", "Jon"); + + /** What must match between cache and database (lastSeen/name are refreshed lazily and may differ). */ + record Link(UUID friend, boolean best, String nickname) {} + + static Set links(List friends) { + return friends.stream().map(f -> new Link(f.id(), f.best(), f.nickname())).collect(Collectors.toSet()); + } + + static void assertConsistent(Storage storage, List proxies, Map online) throws Exception { + Map> db = new HashMap<>(); + for (String name : NAMES) { + UUID id = TestSupport.uuid(name); + db.put(id, storage.load(id).friends().stream().map(Friend::id).collect(Collectors.toSet())); + } + db.forEach((a, friends) -> friends.forEach(b -> + assertEquals(true, db.get(b).contains(a), "friendship " + a + " -> " + b + " is one-sided"))); + for (TestPlayer p : online.values()) { + List cached = proxies.stream().map(h -> h.friends.cachedFriends(p.id())).filter(l -> !l.isEmpty()) + .findFirst().orElse(List.of()); + assertEquals(links(storage.load(p.id()).friends()), links(cached), "cache drifted for " + p.online().name()); + } + } + + @ParameterizedTest(name = "{0} proxies, lagging db: {1}") + @CsvSource({"1, false, 1", "2, false, 1", "1, true, 1", "2, true, 1", "2, true, 2", "2, true, 3", "2, true, 4", "3, true, 5"}) + void randomCommandSequencesKeepCacheAndDatabaseInSync(int proxyCount, boolean lag, int seed) throws Exception { + Storage storage = Storage.sqlite(dir.resolve("friends.db")); + Hub hub = new Hub(); + List proxies = new ArrayList<>(); + List dbs = new ArrayList<>(); + for (int i = 0; i < proxyCount; i++) { + LaggingDb db = new LaggingDb(); + dbs.add(db); + Network network = proxyCount == 1 ? Network.LOCAL : hub.node("p" + i); + proxies.add(new Harness(storage, 5000, network, lag ? db : Runnable::run)); + } + Runnable drainAll = () -> { + while (dbs.stream().anyMatch(d -> !d.queue.isEmpty())) dbs.forEach(d -> { + while (!d.queue.isEmpty()) d.queue.poll().run(); + }); + hub.claims.clear(); // nothing in flight any more: like Redis' short claim TTL running out + }; + Map online = new HashMap<>(); + Map where = new HashMap<>(); + Random random = new Random(seed * 1000L + proxyCount + (lag ? 100 : 0)); + List crowd = NAMES.subList(0, 4); // few players, so the same pairs keep colliding + + for (int step = 0; step < 3000; step++) { + String name = crowd.get(random.nextInt(crowd.size())); + String other = crowd.get(random.nextInt(crowd.size())); + TestPlayer p = online.get(name); + if (p == null) { // offline players can only join + Harness h = proxies.get(random.nextInt(proxies.size())); + var o = new Platform.Online(TestSupport.uuid(name), name, null, "lobby", new TestSupport.Inbox()); + h.platform.online.put(o.id(), o); + var loaded = h.friends.connect(o); + drainAll.run(); + loaded.join(); + h.friends.greet(o); + online.put(name, new TestPlayer(o, (TestSupport.Inbox) o.audience())); + where.put(name, h); + continue; + } + Harness h = where.get(name); + String line = switch (random.nextInt(16)) { + case 0 -> null; + case 1, 2, 3, 4 -> "add " + other; + case 5, 6, 7, 8 -> "accept " + other; + case 9 -> "deny " + other; + case 10, 11 -> "remove " + other; + case 12 -> "best " + other; + case 13 -> random.nextBoolean() ? "nickname " + other + " N" + step % 7 : "nickname " + other; + case 14 -> random.nextInt(8) == 0 ? "removeall confirm" : "status " + List.of("online", "busy", "offline").get(random.nextInt(3)); + default -> "tick"; + }; + if (line == null) { + h.quit(p); + online.remove(name); + } else if (line.equals("tick")) { + Duration d = Duration.ofSeconds(random.nextInt(120)); + proxies.forEach(x -> { + x.clock.advance(d); + x.friends.expireRequests(); + }); + } else { + h.command.execute(p.online(), line.split(" ")); // not joined: with a lagging db it may still be queued + } + if (lag) dbs.forEach(d -> d.runSome(random)); + else hub.claims.clear(); + if (step % 100 == 0) { + drainAll.run(); + assertConsistent(storage, proxies, online); + } + } + drainAll.run(); + assertConsistent(storage, proxies, online); + storage.close(); + } + + @Test + void concurrentCommandsOnTheRealDatabaseThreadStayConsistent() throws Exception { + Storage storage = Storage.sqlite(dir.resolve("friends.db")); + ExecutorService db = Executors.newSingleThreadExecutor(); // like production: one ordered DB thread + Harness h = new Harness(storage, 5000, Network.LOCAL, db); + Map online = new HashMap<>(); + List crowd = NAMES.subList(0, 4); // few players = the same pairs collide constantly + for (String name : crowd) online.put(name, h.join(name)); + + ExecutorService pool = Executors.newFixedThreadPool(8); + List> commands = new ArrayList<>(); + Set errors = new HashSet<>(); + for (int t = 0; t < 8; t++) { + Random random = new Random(t); + pool.execute(() -> { + for (int i = 0; i < 2000; i++) { + TestPlayer p = online.get(crowd.get(random.nextInt(crowd.size()))); + String other = crowd.get(random.nextInt(crowd.size())); + String line = switch (random.nextInt(6)) { + case 0, 1 -> "add " + other; + case 2, 3 -> "accept " + other; + case 4 -> "remove " + other; + default -> "best " + other; + }; + var f = h.command.execute(p.online(), line.split(" ")); + synchronized (commands) { + commands.add(f); + } + } + }); + } + pool.shutdown(); + assertEquals(true, pool.awaitTermination(30, TimeUnit.SECONDS)); + CompletableFuture.allOf(commands.toArray(CompletableFuture[]::new)) + .exceptionally(e -> { + errors.add(e); + return null; + }).get(30, TimeUnit.SECONDS); + db.submit(() -> {}).get(30, TimeUnit.SECONDS); // drain queued writes + assertEquals(Set.of(), errors); + assertEquals(List.of(), online.values().stream().map(TestPlayer::last).filter(m -> m.contains("Something went wrong")).toList()); + + assertConsistent(storage, List.of(h), online); + db.shutdown(); + storage.close(); + } +} diff --git a/common/src/test/java/com/friends/common/MultiProxyTest.java b/common/src/test/java/com/friends/common/MultiProxyTest.java new file mode 100644 index 0000000..4b24b77 --- /dev/null +++ b/common/src/test/java/com/friends/common/MultiProxyTest.java @@ -0,0 +1,274 @@ +package com.friends.common; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.List; + + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import com.friends.common.TestSupport.Harness; +import com.friends.common.TestSupport.Hub; +import com.friends.common.TestSupport.TestPlayer; + +/** Two (or three) proxies sharing one database and an in-memory stand-in for Redis. */ +class MultiProxyTest { + @TempDir + Path dir; + Storage storage; + Hub hub; + Harness p1; + Harness p2; + + @BeforeEach + void setUp() throws Exception { + storage = Storage.sqlite(dir.resolve("friends.db")); + hub = new Hub(); + p1 = proxy("p1"); + p2 = proxy("p2"); + } + + Harness proxy(String id) { + return new Harness(storage, 5000, hub.node(id), Runnable::run); + } + + @AfterEach + void tearDown() { + storage.close(); + } + + private static void assertContains(String haystack, String needle) { + assertTrue(haystack.contains(needle), () -> "expected <" + needle + "> in:\n" + haystack); + } + + @Test + void requestAcceptAndListAcrossProxies() { + var alice = p1.join("Alice"); + var bob = p2.join("Bob"); + + p1.run(alice, "add Bob"); + assertContains(alice.last(), "You sent a friend request to Bob!"); + assertContains(bob.last(), "Friend request from Alice"); + + p2.run(bob, "accept Alice"); + assertContains(bob.last(), "You are now friends with Alice"); + assertContains(alice.last(), "You are now friends with Bob"); + + p1.run(alice, "list"); + assertContains(alice.last(), "Bob is in lobby"); + p2.run(bob, "list"); + assertContains(bob.last(), "Alice is in lobby"); + } + + @Test + void joinLeaveAndStatusReachFriendsOnOtherProxies() { + var alice = p1.join("Alice"); + var bob = p2.join("Bob"); + p1.run(alice, "add Bob"); + p2.run(bob, "accept Alice"); + + p2.quit(bob); + assertEquals("Friend > Bob left.", alice.last()); + p1.run(alice, "list"); + assertContains(alice.last(), "Bob is currently offline"); + + bob = p2.join("Bob"); + assertEquals("Friend > Bob joined.", alice.last()); + + p2.run(bob, "status busy"); + p1.run(alice, "list"); + assertContains(alice.last(), "Bob is busy"); + } + + @Test + void serverSwitchesAreVisibleEverywhere() { + var alice = p1.join("Alice"); + var bob = p2.join("Bob"); + p1.run(alice, "add Bob"); + p2.run(bob, "accept Alice"); + + var moved = new Platform.Online(bob.id(), "Bob", null, "bedwars-3", bob.inbox()); + p2.platform.online.put(bob.id(), moved); + p2.friends.moved(moved); + p1.run(alice, "list"); + assertContains(alice.last(), "Bob is in bedwars-3"); + } + + @Test + void removingOnOneProxyUpdatesTheOther() { + var alice = p1.join("Alice"); + var bob = p2.join("Bob"); + p1.run(alice, "add Bob"); + p2.run(bob, "accept Alice"); + + p1.run(alice, "remove Bob"); + assertEquals(List.of(), p2.friends.friendNames(bob.id())); + p2.run(bob, "list"); + assertContains(bob.last(), "You don't have any friends yet!"); + } + + @Test + void deniedRequestsDisappearEverywhere() { + var alice = p1.join("Alice"); + var bob = p2.join("Bob"); + p1.run(alice, "add Bob"); + p2.run(bob, "deny Alice"); + p1.run(alice, "requests"); + assertContains(alice.last(), "You don't have any pending friend requests."); + } + + @Test + void mutualRequestsFromDifferentProxiesBecomeFriends() { + var alice = p1.join("Alice"); + var bob = p2.join("Bob"); + p1.run(alice, "add Bob"); + p2.run(bob, "add Alice"); + assertContains(bob.last(), "You are now friends with Alice"); + assertEquals(List.of("Bob"), p1.friends.friendNames(alice.id())); + } + + @Test + void expiryIsAnnouncedOnceByEachPlayersOwnProxy() { + var alice = p1.join("Alice"); + var bob = p2.join("Bob"); + p1.run(alice, "add Bob"); + + p1.clock.advance(Duration.ofMinutes(6)); + p2.clock.advance(Duration.ofMinutes(6)); + p1.friends.expireRequests(); + p2.friends.expireRequests(); + assertEquals(1, alice.all().split("Your friend request to Bob has expired.", -1).length - 1); + assertEquals(1, bob.all().split("The friend request from Alice has expired.", -1).length - 1); + } + + @Test + void tabCompletionAndAddSeeTheWholeNetworkButNotInvisiblePlayers() { + var alice = p1.join("Alice"); + var bob = p2.join("Bob"); + p2.join("Carol"); + assertEquals(List.of("Bob", "Carol"), p1.suggest(alice, "add", "")); + + p2.run(bob, "status offline"); + assertEquals(List.of("Carol"), p1.suggest(alice, "add", "")); + p1.run(alice, "add bob"); // still addable by name, like an offline player + assertContains(alice.last(), "You sent a friend request to Bob!"); + } + + @Test + void aProxyStartedLaterGetsTheSharedStateFromTheSnapshot() { + var alice = p1.join("Alice"); + var bob = p2.join("Bob"); + p1.run(alice, "add Bob"); + p2.run(bob, "accept Alice"); + var carol = p1.join("Carol"); + p1.run(carol, "add Dave"); // Dave is unknown: nothing happens + p1.quit(p1.join("Dave")); // Dave has played before + p1.run(carol, "add Dave"); // request pending for an offline player + + var p3 = proxy("p3"); + var dave = p3.join("Dave"); + assertContains(dave.all(), "You have 1 pending friend request."); + p3.run(dave, "accept Carol"); + assertContains(carol.last(), "You are now friends with Dave"); + + p3.run(dave, "add Alice"); + p1.run(alice, "accept Dave"); + p3.run(dave, "list"); + assertContains(dave.last(), "Alice is in lobby"); + } + + @Test + void aLateLeaveFromTheOldProxyDoesNotHideAPlayerWhoMovedProxies() { + var alice = p1.join("Alice"); + var bob = p2.join("Bob"); + p1.run(alice, "add Bob"); + p2.run(bob, "accept Alice"); + + hub.hold = true; // Bob moves p2 -> p1, but p2's "left" is delivered after p1's "joined" + p2.quit(bob); + hub.hold = false; + bob = p1.join("Bob"); + hub.release(); + + p1.run(alice, "list"); + assertContains(alice.last(), "Bob is in lobby"); + var p3 = proxy("p3"); // and the shared state agrees + var carol = p3.join("Carol"); + assertTrue(p3.suggest(carol, "add", "").contains("Bob")); + } + + @Test + void aProxyGoingDownMarksItsPlayersOffline() { + var alice = p1.join("Alice"); + var bob = p2.join("Bob"); + p1.run(alice, "add Bob"); + p2.run(bob, "accept Alice"); + + p1.friends.event(new Event.ProxyDown("p2")); // what RedisNetwork delivers once p2's heartbeat expires + assertEquals("Friend > Bob left.", alice.last()); + p1.run(alice, "list"); + assertContains(alice.last(), "Bob is currently offline"); + } + + @Test + void crossingAcceptsOnTwoProxiesCannotResurrectARemovedFriendship() throws Exception { + // Slow databases, and hold deliveries so each step is explicit. + var db1 = new TestSupport.LaggingDb(); + var db2 = new TestSupport.LaggingDb(); + var q1 = new Harness(storage, 5000, hub.node("q1"), db1); + var q2 = new Harness(storage, 5000, hub.node("q2"), db2); + var alice = connect(q1, db1, "Alice"); + var bob = connect(q2, db2, "Bob"); + + hub.hold = true; // both ask at once: neither proxy has seen the other's request yet + q1.command.execute(alice.online(), "add Bob".split(" ")); + q2.command.execute(bob.online(), "add Alice".split(" ")); + db1.drain(); + db2.drain(); + hub.release(); // now both requests are known everywhere + + q1.command.execute(alice.online(), "accept Bob".split(" ")); + q2.command.execute(bob.online(), "accept Alice".split(" ")); + db2.queue.poll().run(); // q2 decides first, but its insert stays stuck behind a slow disk + db1.drain(); // q1 decides too (unless q2's claim stops it) and its insert lands + hub.release(); + q1.command.execute(alice.online(), "remove Bob".split(" ")); // no-op unless q1 befriended + db1.drain(); + hub.release(); + db2.drain(); // q2's late insert lands last + hub.release(); + hub.hold = false; + + // Whoever won, both caches must agree with the database. + List stored = storage.load(alice.id()).friends().stream().map(Friend::name).toList(); + assertEquals(stored, q1.friends.friendNames(alice.id()), "Alice's cache vs database"); + assertEquals(storage.load(bob.id()).friends().stream().map(Friend::name).toList(), + q2.friends.friendNames(bob.id()), "Bob's cache vs database"); + } + + private static TestPlayer connect(Harness h, TestSupport.LaggingDb db, String name) { + var o = new Platform.Online(TestSupport.uuid(name), name, null, "lobby", new TestSupport.Inbox()); + h.platform.online.put(o.id(), o); + var loaded = h.friends.connect(o); + db.drain(); + loaded.join(); + return new TestPlayer(o, (TestSupport.Inbox) o.audience()); + } + + @Test + void resyncKeepsAndReannouncesThisProxysOwnPlayers() { + var alice = p1.join("Alice"); + p2.join("Bob"); + + hub.online.clear(); // e.g. Redis restarted and lost its data + p1.friends.resync(hub.snapshot()); + assertTrue(hub.online.containsKey(alice.id()), "p1 re-published Alice"); + assertTrue(p2.suggest(p2.join("Carol"), "add", "").contains("Alice")); + } +} diff --git a/common/src/test/java/com/friends/common/RedisNetworkTest.java b/common/src/test/java/com/friends/common/RedisNetworkTest.java new file mode 100644 index 0000000..ab522b3 --- /dev/null +++ b/common/src/test/java/com/friends/common/RedisNetworkTest.java @@ -0,0 +1,187 @@ +package com.friends.common; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +import java.nio.file.Path; +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.slf4j.LoggerFactory; + +import com.friends.common.Storage.PlayerRow; +import com.friends.common.TestSupport.Harness; + +import redis.clients.jedis.Jedis; +import redis.clients.jedis.RedisClient; +import redis.clients.jedis.args.ClientType; +import redis.clients.jedis.params.ClientKillParams; + +/** Real-Redis checks; the Redis-backed ones need -Dfriends.redis=host:port (they use a throwaway namespace). */ +class RedisNetworkTest { + @TempDir + Path dir; + final String namespace = "friends-test-" + UUID.randomUUID().toString().substring(0, 8); + final List cleanup = new ArrayList<>(); + Storage storage; + + @AfterEach + void tearDown() throws Exception { + for (AutoCloseable c : cleanup.reversed()) c.close(); + if (storage != null) storage.close(); + } + + @Test + void everyEventSurvivesTheWireFormat() { + Instant t = Instant.ofEpochMilli(1_790_000_000_123L); + PlayerRow a = new PlayerRow(TestSupport.uuid("A"), "A", "&b[VIP] ", t); + PlayerRow b = new PlayerRow(TestSupport.uuid("B"), "B", null, t); + Presence p = new Presence("A", "&b[VIP] ", "p1", null, Status.AWAY); + List events = List.of( + new Event.Joined(a.id(), p), + new Event.Updated(a.id(), p.withServer("lobby")), + new Event.Left(a.id(), "p1", t), + new Event.RequestSent(new Request(a, b, t)), + new Event.RequestDenied(a.id(), b.id()), + new Event.Befriended(a, b, t), + new Event.Unfriended(a.id(), List.of(b.id())), + new Event.ProxyDown("p1")); + assertEquals(Event.class.getPermittedSubclasses().length, events.size(), "cover every event type"); + for (Event e : events) { + assertEquals(e, RedisNetwork.GSON.fromJson(RedisNetwork.GSON.toJson(e), e.getClass())); + } + } + + // --- against a real Redis --- + + String[] redisAddress() { + String spec = System.getProperty("friends.redis"); + assumeTrue(spec != null && !spec.isBlank(), "set -Dfriends.redis=host:port to run Redis tests"); + return spec.split(":", 2); + } + + RedisNetwork network(String proxy) throws Exception { + String[] hp = redisAddress(); + var n = new RedisNetwork(hp[0], Integer.parseInt(hp[1]), null, 0, false, namespace, proxy, + Duration.ofMillis(100), Duration.ofMillis(500), LoggerFactory.getLogger("redis-" + proxy)); + cleanup.add(n); + return n; + } + + Harness proxy(RedisNetwork network) throws Exception { + if (storage == null) storage = Storage.sqlite(dir.resolve("friends.db")); + var h = new Harness(storage, 5000, network, Runnable::run); + h.clock.now = Instant.now(); // Redis TTLs are computed from wall-clock time, like in production + return h; + } + + RedisClient client() { + String[] hp = redisAddress(); + RedisClient c = RedisClient.create(hp[0], Integer.parseInt(hp[1])); + cleanup.add(c); + return c; + } + + /** Events cross proxies asynchronously: retry the assertion for a few seconds. */ + static void eventually(Runnable assertion) throws InterruptedException { + long deadline = System.nanoTime() + Duration.ofSeconds(5).toNanos(); + while (true) { + try { + assertion.run(); + return; + } catch (AssertionError e) { + if (System.nanoTime() > deadline) throw e; + Thread.sleep(25); + } + } + } + + private static void assertContains(String haystack, String needle) { + assertTrue(haystack.contains(needle), () -> "expected <" + needle + "> in:\n" + haystack); + } + + @Test + void friendsAcrossTwoProxies() throws Exception { + var p1 = proxy(network("p1")); + var p2 = proxy(network("p2")); + var alice = p1.join("Alice"); + var bob = p2.join("Bob"); + eventually(() -> assertTrue(p1.suggest(alice, "add", "").contains("Bob"))); + + p1.run(alice, "add Bob"); + eventually(() -> assertContains(bob.all(), "Friend request from Alice")); + p2.run(bob, "accept Alice"); + eventually(() -> assertContains(alice.all(), "You are now friends with Bob")); + p1.run(alice, "list"); + assertContains(alice.last(), "Bob is in lobby"); + + p2.quit(bob); + eventually(() -> assertEquals("Friend > Bob left.", alice.last())); + } + + @Test + void onlyOneProxyCanClaimAPairAtATime() throws Exception { + var n1 = network("p1"); + var n2 = network("p2"); + UUID a = TestSupport.uuid("A"); + UUID b = TestSupport.uuid("B"); + assertTrue(n1.claim(a, b)); + assertTrue(!n2.claim(b, a), "same pair, either order"); + assertTrue(n2.claim(a, TestSupport.uuid("C"))); + eventually(() -> assertTrue(n2.claim(b, a), "claims expire after 2s")); // eventually() waits up to 5s + } + + @Test + void aLateProxyStartsFromTheSnapshot() throws Exception { + var p1 = proxy(network("p1")); + var alice = p1.join("Alice"); + p1.quit(p1.join("Dave")); + p1.run(alice, "add Dave"); + eventually(() -> assertEquals(1, client().keys(namespace + ":request:*").size())); + + var p3 = proxy(network("p3")); // start() waits for the first snapshot + var dave = p3.join("Dave"); + assertContains(dave.all(), "You have 1 pending friend request."); + assertTrue(p3.friends.onlineNames().contains("Alice")); + } + + @Test + void aProxyThatStopsHeartbeatingIsReaped() throws Exception { + var p1 = proxy(network("p1")); + var net2 = network("p2"); + var p2 = proxy(net2); + var alice = p1.join("Alice"); + var bob = p2.join("Bob"); + p1.run(alice, "add Bob"); + eventually(() -> assertContains(bob.all(), "Friend request from Alice")); + p2.run(bob, "accept Alice"); + eventually(() -> assertContains(alice.all(), "You are now friends with Bob")); + + net2.heartbeat.shutdownNow(); // p2 "freezes": no Left events, its heartbeat key expires + eventually(() -> assertEquals("Friend > Bob left.", alice.last())); + assertTrue(client().hget(namespace + ":online", bob.id().toString()) == null); + } + + @Test + void afterLosingTheSubscriptionAndTheDataAProxyReannouncesItsPlayers() throws Exception { + var p1 = proxy(network("p1")); + var alice = p1.join("Alice"); + var redis = client(); + eventually(() -> assertTrue(redis.hexists(namespace + ":online", alice.id().toString()))); + + redis.del(namespace + ":online"); // e.g. Redis restarted without persistence + String[] hp = redisAddress(); + try (var admin = new Jedis(hp[0], Integer.parseInt(hp[1]))) { // drop every subscription + admin.clientKill(ClientKillParams.clientKillParams() + .type(ClientType.PUBSUB)); + } + eventually(() -> assertTrue(redis.hexists(namespace + ":online", alice.id().toString()))); + } +} diff --git a/common/src/test/java/com/friends/common/TestSupport.java b/common/src/test/java/com/friends/common/TestSupport.java index e94cca0..36db9d7 100644 --- a/common/src/test/java/com/friends/common/TestSupport.java +++ b/common/src/test/java/com/friends/common/TestSupport.java @@ -11,14 +11,17 @@ import java.time.Instant; import java.time.ZoneId; import java.time.ZoneOffset; +import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.Executor; import org.slf4j.LoggerFactory; @@ -88,6 +91,27 @@ public void sendMessage(Component message) { } } + /** + * A single-threaded "slow database": SQL tasks queue up and run in random FIFO batches between commands, so any + * logic that depends on when a write actually lands (rather than on queue order) shows up deterministically. + */ + static final class LaggingDb implements Executor { + final ArrayDeque queue = new ArrayDeque<>(); + + @Override + public void execute(Runnable task) { + queue.add(task); + } + + void drain() { + while (!queue.isEmpty()) queue.poll().run(); + } + + void runSome(java.util.Random random) { + for (int n = random.nextInt(3); n > 0 && !queue.isEmpty(); n--) queue.poll().run(); // 0-2 tasks: writes pile up + } + } + static final class FakePlatform implements Platform { final Map online = new ConcurrentHashMap<>(); @@ -95,15 +119,81 @@ static final class FakePlatform implements Platform { public Optional player(UUID id) { return Optional.ofNullable(online.get(id)); } + } - @Override - public Optional player(String name) { - return online.values().stream().filter(o -> o.name().equalsIgnoreCase(name)).findFirst(); + /** + * In-memory stand-in for Redis: shared state with the same semantics as {@link RedisNetwork#publish}, and + * synchronous fan-out to the other proxies. {@code hold} queues deliveries so tests can reorder them. + */ + static final class Hub { + final Map online = new ConcurrentHashMap<>(); + final Map requests = new ConcurrentHashMap<>(); + final Set> claims = ConcurrentHashMap.newKeySet(); // like Redis' 2s claim keys; tests clear it + final List nodes = new CopyOnWriteArrayList<>(); + final List held = new ArrayList<>(); + boolean hold; + + Node node(String proxy) { + return new Node(proxy); } - @Override - public Collection onlineNames() { - return online.values().stream().map(Online::name).sorted().toList(); + void release() { + List batch = List.copyOf(held); + held.clear(); + batch.forEach(Runnable::run); + } + + Network.Snapshot snapshot() { + return new Network.Snapshot(Map.copyOf(online), List.copyOf(requests.values())); + } + + final class Node implements Network { + final String proxy; + Listener listener; + + Node(String proxy) { + this.proxy = proxy; + } + + @Override public String proxyId() { return proxy; } + + @Override + public boolean claim(UUID a, UUID b) { + return claims.add(Set.of(a, b)); + } + + @Override + public void start(Listener listener) { + this.listener = listener; + nodes.add(this); + listener.resync(snapshot()); + } + + @Override + public void publish(Event event) { + switch (event) { + case Event.Joined(UUID id, Presence p) -> online.put(id, p); + case Event.Updated(UUID id, Presence p) -> online.put(id, p); + case Event.Left(UUID id, String from, Instant _) -> online.computeIfPresent(id, (_, p) -> p.proxy().equals(from) ? null : p); + case Event.RequestSent(Request r) -> requests.put(r.key(), r); + case Event.RequestDenied(UUID from, UUID to) -> requests.remove(new Request.Key(from, to)); + case Event.Befriended(var a, var b, Instant _) -> { + requests.remove(new Request.Key(a.id(), b.id())); + requests.remove(new Request.Key(b.id(), a.id())); + } + case Event.Unfriended _, Event.ProxyDown _ -> {} + } + for (Node n : nodes) { + if (n == this) continue; + Runnable delivery = () -> n.listener.event(event); + if (hold) held.add(delivery); else delivery.run(); + } + } + + @Override + public void close() { + nodes.remove(this); + } } } @@ -138,10 +228,16 @@ static final class Harness implements AutoCloseable { final FriendCommand command; Harness(Storage storage, int maxFriends) { + this(storage, maxFriends, Network.LOCAL, Runnable::run); + } + + /** One proxy: its own players, clock and caches, sharing {@code storage} (and {@code network}) with others. */ + Harness(Storage storage, int maxFriends, Network network, java.util.concurrent.Executor db) { this.storage = storage; - this.friends = new Friends(storage, platform, Runnable::run, clock, Duration.ofMinutes(5), maxFriends, + this.friends = new Friends(storage, platform, network, db, clock, Duration.ofMinutes(5), maxFriends, LoggerFactory.getLogger(Harness.class)); - this.command = new FriendCommand(friends, platform); + this.command = new FriendCommand(friends); + friends.start(); } TestPlayer join(String name) { diff --git a/velocity/build.gradle.kts b/velocity/build.gradle.kts index 697dda3..c27d8f5 100644 --- a/velocity/build.gradle.kts +++ b/velocity/build.gradle.kts @@ -14,6 +14,7 @@ dependencies { configurations.runtimeClasspath { exclude(group = "org.slf4j") // provided by Velocity + exclude(group = "com.google.code.gson") // provided by Velocity exclude(group = "com.google.protobuf") // only used by MySQL's X DevAPI } @@ -23,6 +24,9 @@ tasks.shadowJar { // sqlite-jdbc is not relocated: its JNI symbols are bound to the org.sqlite package name. relocate("com.zaxxer.hikari", "com.friends.lib.hikari") relocate("com.mysql", "com.friends.lib.mysql") + relocate("redis.clients", "com.friends.lib.jedis") + relocate("org.apache.commons.pool2", "com.friends.lib.pool2") + relocate("org.json", "com.friends.lib.json") // No Java 25 runtime exists for 32-bit x86 (JEP 503/479) or 32-bit ARM Windows: drop those natives (~5 MB). exclude("org/sqlite/native/*/x86/**", "org/sqlite/native/Windows/armv7/**") mergeServiceFiles() diff --git a/velocity/src/main/java/com/friends/velocity/FriendsVelocity.java b/velocity/src/main/java/com/friends/velocity/FriendsVelocity.java index 22f1a4a..d6329d0 100644 --- a/velocity/src/main/java/com/friends/velocity/FriendsVelocity.java +++ b/velocity/src/main/java/com/friends/velocity/FriendsVelocity.java @@ -7,7 +7,6 @@ import java.sql.SQLException; import java.time.Clock; import java.time.Duration; -import java.util.Collection; import java.util.List; import java.util.Objects; import java.util.Optional; @@ -23,7 +22,9 @@ import com.friends.common.FriendCommand; import com.friends.common.Friends; +import com.friends.common.Network; import com.friends.common.Platform; +import com.friends.common.RedisNetwork; import com.friends.common.Storage; import com.google.inject.Inject; import com.velocitypowered.api.command.CommandManager; @@ -47,7 +48,7 @@ import net.luckperms.api.LuckPermsProvider; import net.luckperms.api.model.user.User; -@Plugin(id = "friends", name = "Friends", version = "0.2.0", +@Plugin(id = "friends", name = "Friends", version = "0.3.0", description = "Hypixel-style friends system", dependencies = @Dependency(id = "luckperms", optional = true)) public final class FriendsVelocity { @@ -56,6 +57,7 @@ public final class FriendsVelocity { private final Path dataDir; private ExecutorService db; private Storage storage; + private Network network; private Friends friends; private VelocityPlatform platform; @@ -72,24 +74,28 @@ public void onInit(ProxyInitializeEvent event) { try { config = loadConfig(); storage = openStorage(config); + network = openNetwork(config); } catch (IOException | SQLException e) { - log.error("Friends is disabled: could not load config or open storage", e); + log.error("Friends is disabled: {}", e.getMessage(), e); + if (storage != null) storage.close(); return; } // One thread keeps writes ordered and off the proxy's event/command threads. db = Executors.newSingleThreadExecutor(Thread.ofPlatform().name("friends-db").daemon().factory()); Function prefixes = proxy.getPluginManager().isLoaded("luckperms") ? LuckPermsPrefix::of : _ -> null; platform = new VelocityPlatform(proxy, prefixes); - friends = new Friends(storage, platform, db, Clock.systemUTC(), + friends = new Friends(storage, platform, network, db, Clock.systemUTC(), Duration.ofMinutes(config.node("request-expiry-minutes").getLong(5)), config.node("max-friends").getInt(5000), log); + friends.start(); - FriendCommand command = new FriendCommand(friends, platform); + FriendCommand command = new FriendCommand(friends); CommandManager commands = proxy.getCommandManager(); commands.register(commands.metaBuilder("friend").aliases("f", "friends").plugin(this).build(), new Cmd(command, platform, null)); commands.register(commands.metaBuilder("fl").plugin(this).build(), new Cmd(command, platform, "list")); commands.register(commands.metaBuilder("status").plugin(this).build(), new Cmd(command, platform, "status")); proxy.getScheduler().buildTask(this, friends::expireRequests).repeat(Duration.ofSeconds(1)).schedule(); + log.info("Friends enabled (proxy id: {})", network.proxyId()); } @Subscribe @@ -99,7 +105,10 @@ public EventTask onPostLogin(PostLoginEvent event) { @Subscribe public void onServerPostConnect(ServerPostConnectEvent event) { - if (friends != null && event.getPreviousServer() == null) friends.greet(platform.online(event.getPlayer())); + if (friends == null) return; + Platform.Online online = platform.online(event.getPlayer()); + friends.moved(online); + if (event.getPreviousServer() == null) friends.greet(online); } @Subscribe @@ -111,8 +120,9 @@ public void onDisconnect(DisconnectEvent event) { @Subscribe public void onShutdown(ProxyShutdownEvent event) throws InterruptedException { if (db == null) return; - db.shutdown(); + db.shutdown(); // players are already disconnected: finish their writes (and the publishes queued behind them) if (!db.awaitTermination(10, TimeUnit.SECONDS)) log.warn("Friends: gave up waiting for pending database writes"); + network.close(); storage.close(); } @@ -127,8 +137,12 @@ private ConfigurationNode loadConfig() throws IOException { return YamlConfigurationLoader.builder().path(file).build().load(); } + private static boolean mysql(ConfigurationNode config) { + return "mysql".equalsIgnoreCase(config.node("storage", "type").getString("sqlite")); + } + private Storage openStorage(ConfigurationNode config) throws SQLException { - if ("mysql".equalsIgnoreCase(config.node("storage", "type").getString("sqlite"))) { + if (mysql(config)) { ConfigurationNode m = config.node("mysql"); return Storage.mysql(m.node("host").getString("localhost"), m.node("port").getInt(3306), m.node("database").getString("friends"), m.node("user").getString("root"), @@ -137,6 +151,19 @@ private Storage openStorage(ConfigurationNode config) throws SQLException { return Storage.sqlite(dataDir.resolve(config.node("sqlite", "file").getString("friends.db"))); } + private Network openNetwork(ConfigurationNode config) throws IOException { + ConfigurationNode r = config.node("redis"); + if (!r.node("enabled").getBoolean(false)) return Network.LOCAL; + if (!mysql(config)) { + throw new IOException("redis (multi-proxy) needs storage.type: mysql, shared by every proxy"); + } + String id = r.node("proxy-id").getString(""); + if (id.isBlank()) id = UUID.randomUUID().toString().substring(0, 8); + return new RedisNetwork(r.node("host").getString("localhost"), r.node("port").getInt(6379), + r.node("password").getString(""), r.node("database").getInt(0), r.node("ssl").getBoolean(false), + r.node("namespace").getString("friends"), id, log); + } + record VelocityPlatform(ProxyServer proxy, Function prefixes) implements Platform { Online online(Player p) { return new Online(p.getUniqueId(), p.getUsername(), prefixes.apply(p.getUniqueId()), @@ -147,16 +174,6 @@ Online online(Player p) { public Optional player(UUID id) { return proxy.getPlayer(id).map(this::online); } - - @Override - public Optional player(String name) { - return proxy.getPlayer(name).map(this::online); - } - - @Override - public Collection onlineNames() { - return proxy.getAllPlayers().stream().map(Player::getUsername).toList(); - } } /** Routes /friend, /fl ("list") and /status ("status") into the shared command parser. */ diff --git a/velocity/src/main/resources/config.yml b/velocity/src/main/resources/config.yml index 8039d94..0a097fa 100644 --- a/velocity/src/main/resources/config.yml +++ b/velocity/src/main/resources/config.yml @@ -20,3 +20,17 @@ request-expiry-minutes: 5 # Maximum friends per player (Hypixel: 5000). max-friends: 5000 + +# Several proxies (multi-proxy network): share who is online, friend requests and notifications through Redis. +# Needs storage.type: mysql, with every proxy using the same database. +redis: + enabled: false + host: localhost + port: 6379 + password: "" + database: 0 + ssl: false + # Unique per proxy, stable across restarts (e.g. "proxy-1"). Empty = a random id on every start. + proxy-id: "" + # Prefix for keys and the pub/sub channel, so several networks can share one Redis. + namespace: friends