fix: Fix dispatcher race conditions and add sanitizers to CI

- Fix race condition in try_pop_task: hold lock for entire operation
- Fix lost tasks: re-queue extra FD tasks instead of discarding them
- Require unique_lock parameter to enforce locking at compile time
- Add ASan, TSan, and UBSan CI jobs to catch threading/memory issues

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: mhaynie

.github/workflows/ccpp.yml

@@ -124,15 +124,54 @@ jobs:
   #       port-name: mh-stuff
   #       workflow-pat: ${{ secrets.WORKFLOW_PAT }}
 
+  sanitizers:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: true
+      matrix:
+        sanitizer:
+          - name: asan
+            flags: -fsanitize=address -fno-omit-frame-pointer
+          - name: tsan
+            flags: -fsanitize=thread
+          - name: ubsan
+            flags: -fsanitize=undefined -fno-omit-frame-pointer
+
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Install tools
+      run: |
+        sudo apt-get update
+        sudo apt-get install -y clang-15 libc++-15-dev libc++abi-15-dev ninja-build libcurl4-openssl-dev
+
+    - name: Build with ${{ matrix.sanitizer.name }}
+      env:
+        CXX: clang++-15
+        CXXFLAGS: ${{ matrix.sanitizer.flags }} -g -O1
+      run: |
+        mkdir build
+        cd build
+        cmake -G Ninja -DMH_STUFF_COMPILE_LIBRARY=ON ../
+        cmake --build .
+
+    - name: Run tests with ${{ matrix.sanitizer.name }}
+      env:
+        ASAN_OPTIONS: detect_leaks=1:abort_on_error=1
+        TSAN_OPTIONS: abort_on_error=1
+        UBSAN_OPTIONS: print_stacktrace=1:abort_on_error=1
+      run: |
+        cd build
+        ctest --output-on-failure
+
   all-checks-passed:
     if: always()
-    needs: [build-linux]
+    needs: [build-linux, sanitizers]
     runs-on: ubuntu-latest
     steps:
     - name: Verify all checks passed
-      run: |
-        if [[ "${{ needs.build-linux.result }}" != "success" ]]; then
-          echo "build-linux failed: ${{ needs.build-linux.result }}"
-          exit 1
-        fi
-        echo "All checks passed!"
+      if: ${{ !(contains(needs.*.result, 'failure') || contains(needs.*.result, 'cancelled')) }}
+      run: echo "All checks passed!"
+    - name: Fail if any check failed
+      if: ${{ contains(needs.*.result, 'failure') || contains(needs.*.result, 'cancelled') }}
+      run: exit 1

cpp/include/mh/concurrency/dispatcher.inl

@@ -66,37 +66,40 @@ namespace mh
 
 			coro::coroutine_handle<> try_pop_task()
 			{
+				std::unique_lock lock(m_TasksMutex);
+
 				// Check for ready FDs first
-				auto ready_fd_tasks = check_fd_tasks();
+				auto ready_fd_tasks = check_fd_tasks_locked(lock);
 				if (!ready_fd_tasks.empty())
 				{
-					return ready_fd_tasks[0]; // Return first ready FD task
-				}
-
-				if (!m_Tasks.empty() || !m_DelayTasks.empty())
-				{
-					std::lock_guard lock(m_TasksMutex);
-
-					if (!m_DelayTasks.empty())
+					// Return first task, re-queue any extras
+					auto task = ready_fd_tasks[0];
+					for (size_t i = 1; i < ready_fd_tasks.size(); ++i)
 					{
-						auto now = clock_t::now();
-						const task_delay_data &taskDelayData = m_DelayTasks.front();
-						if (taskDelayData.m_DelayUntilTime <= now)
-						{
-							auto task = taskDelayData.m_Handle;
-							m_DelayTasks.pop();
-							return task;
-						}
+						m_Tasks.push(ready_fd_tasks[i]);
 					}
+					return task;
+				}
 
-					if (!m_Tasks.empty())
+				if (!m_DelayTasks.empty())
+				{
+					auto now = clock_t::now();
+					const task_delay_data &taskDelayData = m_DelayTasks.front();
+					if (taskDelayData.m_DelayUntilTime <= now)
 					{
-						auto task = m_Tasks.front();
-						m_Tasks.pop();
+						auto task = taskDelayData.m_Handle;
+						m_DelayTasks.pop();
 						return task;
 					}
 				}
 
+				if (!m_Tasks.empty())
+				{
+					auto task = m_Tasks.front();
+					m_Tasks.pop();
+					return task;
+				}
+
 				return nullptr;
 			}
 
@@ -152,10 +155,10 @@ namespace mh
 			}
 
 			// Check for ready FDs and return ready tasks
-			std::vector<coro::coroutine_handle<>> check_fd_tasks()
+			std::vector<coro::coroutine_handle<>> check_fd_tasks_locked(std::unique_lock<std::mutex>& lock)
 			{
+				assert(lock.owns_lock() && lock.mutex() == &m_TasksMutex);
 				std::vector<coro::coroutine_handle<>> ready_tasks;
-				std::lock_guard lock(m_TasksMutex);
 
 #ifdef _WIN32
 				// Windows: stub implementation for now