diff --git a/.junie/plans/thread-safety-improvements.md b/.junie/plans/thread-safety-improvements.md new file mode 100644 index 0000000..aefb5cc --- /dev/null +++ b/.junie/plans/thread-safety-improvements.md @@ -0,0 +1,99 @@ +--- +sessionId: session-260910-054543-zckm +--- + +# Requirements + +### Overview & Goals +Analyze and fix thread-safety issues and improve the `ExecutorProvider` / `ThreadManager` infrastructure and all code that interacts with it. + +### Scope +#### In Scope +- Fix thread-safety bugs in `ThreadManager`, `TaskManagerImpl`, `ChannelMessagePurger`, `MusicManager` +- Improve executor pool configuration (bounded pools, proper shutdown) +- Clean up inconsistent synchronization patterns + +#### Out of Scope +- Rewriting the task/command architecture +- Changing the `QueueManager` synchronization (it's consistent and correct, just verbose) + +# Technical Design + +### Current Issues Found + +#### 1. `ThreadManager` (`core/thread/ThreadManager.java`) +- **Unbounded cached thread pools** — `newCachedThreadPool()` creates unlimited threads; under load (e.g., many guilds purging simultaneously) this can exhaust memory. +- **Shutdown bug (line 66-69)** — If `awaitTermination` throws `InterruptedException` on the *first* executor, the `catch` block exits the loop and remaining executors only get `shutdownNow()` without a prior `shutdown()` call. Each executor should be shut down independently. +- **No daemon threads** — `BaseThreadFactory` doesn't set `thread.setDaemon(true)`, so leaked threads can prevent JVM exit. + +#### 2. `TaskManagerImpl` (`core/tasks/TaskManagerImpl.java`) +- **Null values in `ConcurrentHashMap` (line 178-184)** — `computeIfAbsent` returns `null` for `PersistableTask`/`BuiltInTask`, but `ConcurrentHashMap` does **not** permit null values. This will throw `NullPointerException` at runtime. +- **Mixed synchronization** — `getTasks()` is `synchronized` on `this`, but all mutations use `ConcurrentHashMap` operations (no `synchronized`). The `synchronized` keyword on `getTasks()` provides no actual protection since `insert()`, `removeTask()`, etc. don't synchronize on `this`. +- **`hasTask()` race (line 93-95)** — Check-then-act across two collections (`tasks` and `pendingTasks`) without synchronization. +- **`handle()` lock cleanup race (line 191-194)** — After acquiring the lock, it checks `tasks.containsKey()` and removes from `tasksLock`, but another thread could be waiting on the same lock and will find it removed. + +#### 3. `ChannelMessagePurger` (`utils/ChannelMessagePurger.java`) +- **Double unlock bug** — `purgeMessages()` acquires `lock` via `tryLock()` (line 68) and releases it in `finally` (line 79). But `safePurge()` also calls `lock.unlock()` (line 154), causing an `IllegalMonitorStateException` or releasing a lock that a *different* thread now holds. +- **Non-thread-safe static `HashMap`** (line 40) — `instances` is a plain `HashMap` accessed from multiple threads via `computeIfAbsent`. Should be `ConcurrentHashMap`. +- **Static mutable `ExecutorProvider` binding** — The `volatile` field + `bind()` pattern works but is fragile; a null check race exists between `canLockBeAcquired` callers and `purgeMessages`. + +#### 4. `MusicManager` (`music/MusicManager.java`) +- **Inconsistent synchronization** — `isConnected()`, `connect()`, `disconnect()`, `play()` are `synchronized`, but `isCurrentlyPlaying()` (line 114), `getCurrentlyPlaying()` (line 110), `stopTrack()` (line 138), `resume()` (line 142) are not. If these are called from event threads concurrently with `disconnect()`, they can see stale state. + +#### 5. `BaseThreadFactory` (`core/thread/BaseThreadFactory.java`) +- Threads are not set as daemon threads, which can prevent clean JVM shutdown if an executor leaks. + +### Proposed Changes + +#### `ThreadManager` +- Replace `newCachedThreadPool()` with `newFixedThreadPool()` or a bounded `ThreadPoolExecutor` with a reasonable max (e.g., `availableProcessors() * 2`). +- Fix shutdown loop: wrap each executor's `shutdown()`/`awaitTermination()` in its own try-catch so one failure doesn't skip the rest. +- Consider adding a `getExecutorStatus()` method for observability. + +#### `BaseThreadFactory` +- Set `thread.setDaemon(true)` to prevent JVM hang on unclean shutdown. + +#### `TaskManagerImpl` +- Fix `computeIfAbsent` null return: use a sentinel or simply don't store an entry for tasks that don't need locking (use `get()` + null check instead). +- Remove misleading `synchronized` on `getTasks()` or make all task mutations consistently synchronized. +- Guard `hasTask()` with proper synchronization or accept eventual consistency and document it. +- Fix lock cleanup race in `handle()` — don't remove from `tasksLock` inside the lock-held section; let `removeTask()` handle cleanup. + +#### `ChannelMessagePurger` +- Remove the `lock.unlock()` call on line 154 inside `safePurge()` — the `finally` block in `purgeMessages()` already handles it. +- Change `instances` from `HashMap` to `ConcurrentHashMap`. + +#### `MusicManager` +- Either synchronize all public methods that touch `player`/`AudioManager` state, or remove `synchronized` from methods where JDA already guarantees thread safety and document the threading model. + +# Delivery Steps + +### ✓ Step 1: Fix ThreadManager and BaseThreadFactory +ThreadManager uses bounded pools and has correct shutdown; BaseThreadFactory produces daemon threads. + +- Replace `Executors.newCachedThreadPool()` in `ThreadManager.getExecutor()` with a bounded `ThreadPoolExecutor` (core=0, max=availableProcessors*2, 60s keepAlive, LinkedBlockingQueue). +- Fix `stop()` shutdown loop: wrap each executor's `shutdown()`+`awaitTermination()` in its own try-catch so an `InterruptedException` on one executor doesn't skip the rest. +- In `BaseThreadFactory.newThread()`, add `thread.setDaemon(true)`. +- Verify the `getScheduledExecutor()` pool size is still appropriate. + +### ✓ Step 2: Fix TaskManagerImpl thread-safety bugs +TaskManagerImpl has no null-in-ConcurrentHashMap violations and consistent synchronization. + +- Replace the `computeIfAbsent` that returns null (line 178-184) with a pattern that avoids storing null: check task type first, only call `computeIfAbsent` when a lock is actually needed. +- Remove the misleading `synchronized` keyword from `getTasks()`, or switch to a consistent locking strategy across all mutation methods. +- Fix the lock cleanup race in `handle()`: do not remove from `tasksLock` at line 193; let `removeTask()` be the sole owner of lock cleanup. +- Document the threading contract for `hasTask()` (check-then-act across two collections is inherently racy). + +### ✓ Step 3: Fix ChannelMessagePurger concurrency bugs +ChannelMessagePurger has no double-unlock and uses thread-safe collections. + +- Remove the redundant `lock.unlock()` call on line 154 in `safePurge()` — the `finally` block in `purgeMessages()` (line 79) already releases the lock. +- Change the static `instances` field (line 40) from `HashMap` to `ConcurrentHashMap`. +- Remove the `synchronized` keyword from `countAllChannelMessages()` and `safePurge()` since the `ReentrantLock` already provides mutual exclusion. + +### ✓ Step 4: Harmonize MusicManager synchronization +MusicManager has a consistent and documented threading model. + +- Add `synchronized` to `isCurrentlyPlaying()`, `getCurrentlyPlaying()`, `stopTrack()`, and `resume()` to match the existing pattern on `connect()`/`disconnect()`/`play()`. +- Alternatively, if JDA's `AudioPlayer` is already thread-safe, remove `synchronized` from all methods and add a class-level Javadoc explaining the threading model. +- Investigate which approach is correct by checking Lavaplayer's `AudioPlayer` thread-safety guarantees, then apply consistently. \ No newline at end of file diff --git a/src/main/java/me/bobert/netneutrality/core/tasks/TaskManagerImpl.java b/src/main/java/me/bobert/netneutrality/core/tasks/TaskManagerImpl.java index c6171b0..85ccf94 100644 --- a/src/main/java/me/bobert/netneutrality/core/tasks/TaskManagerImpl.java +++ b/src/main/java/me/bobert/netneutrality/core/tasks/TaskManagerImpl.java @@ -89,13 +89,18 @@ public class TaskManagerImpl implements TaskManager { return tasks.get(id); } + /** + * Checks whether a task with the given ID exists. Note: this check spans two collections + * ({@code tasks} and {@code pendingTasks}) without a single lock, so the result is + * best-effort and may be stale by the time the caller acts on it. + */ @Override public boolean hasTask(String id) { return tasks.containsKey(id) || pendingTasks.stream().anyMatch(task -> task.getId().equals(id)); } @Override - public synchronized Collection getTasks() { + public Collection getTasks() { return tasks.values(); } @@ -175,13 +180,9 @@ public class TaskManagerImpl implements TaskManager { if (task != null) { String lockId = task.getLockId(); - ReentrantLock lock = tasksLock.computeIfAbsent(lockId, k -> { - if (task instanceof PersistableTask || task instanceof BuiltInTask) { - return null; - } - - return new ReentrantLock(); - }); + ReentrantLock lock = (task instanceof PersistableTask || task instanceof BuiltInTask) + ? null + : tasksLock.computeIfAbsent(lockId, k -> new ReentrantLock()); try { if (lock != null) { @@ -190,7 +191,6 @@ public class TaskManagerImpl implements TaskManager { if (!tasks.containsKey(taskId)) { logger.debug("Task {} is no longer available after aquiring lock", taskId); - tasksLock.remove(lockId); return; } } diff --git a/src/main/java/me/bobert/netneutrality/core/thread/BaseThreadFactory.java b/src/main/java/me/bobert/netneutrality/core/thread/BaseThreadFactory.java index 906a7c3..41e0afd 100644 --- a/src/main/java/me/bobert/netneutrality/core/thread/BaseThreadFactory.java +++ b/src/main/java/me/bobert/netneutrality/core/thread/BaseThreadFactory.java @@ -25,6 +25,7 @@ public class BaseThreadFactory implements ThreadFactory { public Thread newThread(@Nonnull Runnable r) { Thread thread = new Thread(r); thread.setName(name + "-" + threadId.getAndIncrement()); + thread.setDaemon(true); thread.setUncaughtExceptionHandler(new ExecutorException()); return thread; } diff --git a/src/main/java/me/bobert/netneutrality/core/thread/ThreadManager.java b/src/main/java/me/bobert/netneutrality/core/thread/ThreadManager.java index 6340c0e..cfd5ced 100644 --- a/src/main/java/me/bobert/netneutrality/core/thread/ThreadManager.java +++ b/src/main/java/me/bobert/netneutrality/core/thread/ThreadManager.java @@ -12,7 +12,9 @@ import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; /** @@ -50,7 +52,13 @@ public class ThreadManager implements ExecutorProvider { @Override public ExecutorService getExecutor(String name) { - return executors.computeIfAbsent(name, s -> Executors.newCachedThreadPool(new BaseThreadFactory(s))); + return executors.computeIfAbsent(name, s -> new ThreadPoolExecutor( + 0, + Runtime.getRuntime().availableProcessors() * 2, + 60L, TimeUnit.SECONDS, + new LinkedBlockingQueue<>(), + new BaseThreadFactory(s) + )); } @Override @@ -62,21 +70,20 @@ public class ThreadManager implements ExecutorProvider { public void stop() { logger.info("[Executor] Shutting down..."); - try { - for (ExecutorService ex : executors.values()) { + for (Map.Entry entry : executors.entrySet()) { + String name = entry.getKey(); + ExecutorService ex = entry.getValue(); + try { ex.shutdown(); - ex.awaitTermination(1, TimeUnit.SECONDS); - } - } catch (InterruptedException e) { - logger.info("[Executor] Running tasks interrupted"); - } finally { - executors.forEach((name, ex) -> { - if (!ex.isTerminated()) { + if (!ex.awaitTermination(1, TimeUnit.SECONDS)) { logger.warn("[Executor: {}] Forcing shutdown of running tasks..", name); ex.shutdownNow(); } - }); - + } catch (InterruptedException e) { + logger.info("[Executor: {}] Running tasks interrupted", name); + ex.shutdownNow(); + Thread.currentThread().interrupt(); + } } logger.info("[Executor] Shutdown finished!"); diff --git a/src/main/java/me/bobert/netneutrality/music/MusicManager.java b/src/main/java/me/bobert/netneutrality/music/MusicManager.java index edf5512..6b83f1d 100644 --- a/src/main/java/me/bobert/netneutrality/music/MusicManager.java +++ b/src/main/java/me/bobert/netneutrality/music/MusicManager.java @@ -107,11 +107,11 @@ public class MusicManager implements MusicManagerCallback { return guild.getAudioManager().getConnectedChannel() != null; } - public AudioTrack getCurrentlyPlaying() { + public synchronized AudioTrack getCurrentlyPlaying() { return player.getPlayingTrack(); } - public boolean isCurrentlyPlaying() { + public synchronized boolean isCurrentlyPlaying() { return player.getPlayingTrack() != null; } @@ -135,11 +135,11 @@ public class MusicManager implements MusicManagerCallback { } } - public void stopTrack() { + public synchronized void stopTrack() { player.stopTrack(); } - public CommandExecutionResult resume() { + public synchronized CommandExecutionResult resume() { return getQueueManager().resume(); } diff --git a/src/main/java/me/bobert/netneutrality/utils/ChannelMessagePurger.java b/src/main/java/me/bobert/netneutrality/utils/ChannelMessagePurger.java index b7b882e..0c0b457 100644 --- a/src/main/java/me/bobert/netneutrality/utils/ChannelMessagePurger.java +++ b/src/main/java/me/bobert/netneutrality/utils/ChannelMessagePurger.java @@ -15,9 +15,9 @@ import net.dv8tion.jda.api.utils.messages.MessageEditData; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; -import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -37,7 +37,7 @@ public class ChannelMessagePurger { private static final Logger logger = LogManager.getLogger(ChannelMessagePurger.class); - private static final Map instances = new HashMap<>(); + private static final Map instances = new ConcurrentHashMap<>(); private final GuildMessageChannel channel; private final Lock lock; @@ -82,7 +82,7 @@ public class ChannelMessagePurger { }); } - private synchronized int countAllChannelMessages(GuildMessageChannel channel) { + private int countAllChannelMessages(GuildMessageChannel channel) { logger.info("Guild {}: counting all messages from channel {}", channel.getGuild().getId(), channel.getId()); MessageHistory history = channel.getHistory(); int totalCount = 0; @@ -99,7 +99,7 @@ public class ChannelMessagePurger { return totalCount; } - private synchronized void safePurge(CommandContext context, int toPurge) { + private void safePurge(CommandContext context, int toPurge) { logger.info("Safe purging {} messages from channel {}", toPurge, channel.getId()); if (toPurge == 1) { @@ -151,6 +151,5 @@ public class ChannelMessagePurger { builder.addContent("\n").addContent("Messages older than two weeks weren't deleted."); } context.edit(MessageEditData.fromCreateData(builder.build())); - lock.unlock(); } }