Fixed the scheduler to handle inline tasks and to not create deadlock on threads of TickerQ

Author: arcenox

src/TickerQ.EntityFrameworkCore/Infrastructure/BasePersistenceProvider.cs

@@ -68,16 +68,16 @@ public async IAsyncEnumerable<TimeTickerEntity> QueueTimedOutTimeTickers([Enumer
         await using var dbContext = await DbContextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false);;
         var context = dbContext.Set<TTimeTicker>();
         var now = _clock.UtcNow;
-        var fallbackThreshold = now.AddMilliseconds(-100);  // Fallback picks up tasks overdue by > 100ms
+        var fallbackThreshold = now.AddSeconds(-1);  // Fallback picks up tasks older than main 1-second window
 
         var timeTickersToUpdate =  await context
             .AsNoTracking()
             .Where(x => x.ExecutionTime != null)
             .Where(x => x.Status == TickerStatus.Idle || x.Status == TickerStatus.Queued)
-            .Where(x => x.ExecutionTime <= fallbackThreshold)  // Only tasks overdue by more than 100ms
+            .Where(x => x.ExecutionTime <= fallbackThreshold)  // Only tasks older than 1 second
             .Include(x => x.Children.Where(y => y.ExecutionTime == null))
             .Select(MappingExtensions.ForQueueTimeTickers<TTimeTicker>())
-            .ToArrayAsync(cancellationToken).ConfigureAwait(false);;
+            .ToArrayAsync(cancellationToken).ConfigureAwait(false);
 
         foreach (var timeTicker in timeTickersToUpdate)
         {
@@ -273,7 +273,7 @@ await dbContext.Set<CronTickerOccurrenceEntity<TCronTicker>>()
     public async IAsyncEnumerable<CronTickerOccurrenceEntity<TCronTicker>> QueueTimedOutCronTickerOccurrences([EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
         var now = _clock.UtcNow;
-        var fallbackThreshold = now.AddMilliseconds(-100);  // Fallback picks up tasks overdue by > 100ms
+        var fallbackThreshold = now.AddSeconds(-1);  // Fallback picks up tasks older than main 1-second window
 
         await using var dbContext = await DbContextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false);;
         var context = dbContext.Set<CronTickerOccurrenceEntity<TCronTicker>>();
@@ -282,7 +282,7 @@ public async IAsyncEnumerable<CronTickerOccurrenceEntity<TCronTicker>> QueueTime
             .AsNoTracking()
             .Include(x => x.CronTicker)
             .Where(x => x.Status == TickerStatus.Idle || x.Status == TickerStatus.Queued)
-            .Where(x => x.ExecutionTime <= fallbackThreshold)  // Only tasks overdue by more than 100ms
+            .Where(x => x.ExecutionTime <= fallbackThreshold)  // Only tasks older than 1 second
             .Select(MappingExtensions.ForQueueCronTickerOccurrence<CronTickerOccurrenceEntity<TCronTicker>, TCronTicker>())
             .ToArrayAsync(cancellationToken).ConfigureAwait(false);
 
@@ -438,13 +438,13 @@ public async IAsyncEnumerable<CronTickerOccurrenceEntity<TCronTicker>> QueueCron
     public async Task<CronTickerOccurrenceEntity<TCronTicker>> GetEarliestAvailableCronOccurrence(Guid[] ids, CancellationToken cancellationToken = default)
     {
         var now = _clock.UtcNow;
-        var mainSchedulerThreshold = now.AddMilliseconds(-now.Millisecond);
+        var mainSchedulerThreshold = now.AddSeconds(-1);
         await using var dbContext = await DbContextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false);;
         return await dbContext.Set<CronTickerOccurrenceEntity<TCronTicker>>()
             .AsNoTracking()
             .Include(x => x.CronTicker)
             .Where(x => ids.Contains(x.CronTickerId))
-            .Where(x => x.ExecutionTime >= mainSchedulerThreshold)  // Only recent/upcoming tasks (not heavily overdue)
+            .Where(x => x.ExecutionTime >= mainSchedulerThreshold)  // Only items within the 1-second main scheduler window
             .WhereCanAcquire(_lockHolder)
             .OrderBy(x => x.ExecutionTime)
             .Select(MappingExtensions.ForLatestQueuedCronTickerOccurrence<CronTickerOccurrenceEntity<TCronTicker>, TCronTicker>())
@@ -475,4 +475,4 @@ await dbContext.Set<CronTickerOccurrenceEntity<TCronTicker>>()
     }
 
     #endregion
-}
+}

src/TickerQ/Exceptions/TerminateExecutionException.cs

@@ -7,8 +7,9 @@ internal class TerminateExecutionException : Exception
     {
         internal readonly TickerStatus Status = TickerStatus.Skipped;
         public TerminateExecutionException(string message) : base(message) { }
-        public TerminateExecutionException(TickerStatus tickerType, string message) : base(message) {}
-        public TerminateExecutionException(string message, Exception innerException) : base(message, innerException) {}
+        public TerminateExecutionException(TickerStatus tickerType, string message) : base(message)
+            => Status = tickerType;
+        public TerminateExecutionException(string message, Exception innerException) : base(message, innerException) { }
         public TerminateExecutionException(TickerStatus tickerType, string message, Exception innerException) : base(message, innerException)
             => Status = tickerType;
     }

src/TickerQ/Src/BackgroundServices/TickerQSchedulerBackgroundService.cs

@@ -106,7 +106,7 @@ private async Task RunTickerQSchedulerAsync(CancellationToken stoppingToken, Can
                 await _internalTickerManager.SetTickersInProgress(_executionContext.Functions, cancellationToken);
 
                 foreach (var function in _executionContext.Functions.OrderBy(x => x.CachedPriority))
-                    await _taskScheduler.QueueAsync(async ct => await _taskHandler.ExecuteTaskAsync(function,false, ct), function.CachedPriority, stoppingToken);
+                    _ = _taskScheduler.QueueAsync(async ct => await _taskHandler.ExecuteTaskAsync(function,false, ct), function.CachedPriority, stoppingToken);
             }
 
             var (timeRemaining, functions) =

src/TickerQ/Src/Provider/TickerInMemoryPersistenceProvider.cs

@@ -81,13 +81,13 @@ public async IAsyncEnumerable<TimeTickerEntity> QueueTimeTickers(TimeTickerEntit
         public async IAsyncEnumerable<TimeTickerEntity> QueueTimedOutTimeTickers([EnumeratorCancellation] CancellationToken cancellationToken = default)
         {
             var now = _clock.UtcNow;
-            var fallbackThreshold = now.AddMilliseconds(-100);  // Fallback picks up tasks overdue by > 100ms
+            var fallbackThreshold = now.AddSeconds(-1);  // Fallback picks up tasks older than main 1-second window
 
             // First, get the time tickers that need to be updated (matching EF query)
             var timeTickersToUpdate = TimeTickers.Values
                 .Where(x => x.ExecutionTime != null)
                 .Where(x => x.Status == TickerStatus.Idle || x.Status == TickerStatus.Queued)
-                .Where(x => x.ExecutionTime <= fallbackThreshold)  // Only tasks overdue by more than 100ms
+                .Where(x => x.ExecutionTime <= fallbackThreshold)  // Only tasks older than 1 second
                 .Select(x => ForQueueTimeTickers(x))  // Map to TimeTickerEntity with children, matching EF's Select
                 .ToArray();
 
@@ -149,13 +149,15 @@ public Task ReleaseAcquiredTimeTickers(Guid[] timeTickerIds, CancellationToken c
         public Task<TimeTickerEntity[]> GetEarliestTimeTickers(CancellationToken cancellationToken = default)
         {
             var now = _clock.UtcNow;
-            var mainSchedulerThreshold = now.AddMilliseconds(-100);  // Main scheduler handles tasks up to 100ms overdue
+            
+            // Define the window: ignore anything older than 1 second ago
+            var oneSecondAgo = now.AddSeconds(-1);
 
             // Build base query matching EF Core's approach
             var baseQuery = TimeTickers.Values
                 .Where(x => x.ExecutionTime != null)
                 .Where(CanAcquire)
-                .Where(x => x.ExecutionTime >= mainSchedulerThreshold);  // Only recent/upcoming tasks (not heavily overdue)
+                .Where(x => x.ExecutionTime >= oneSecondAgo);  // Ignore old tickers (fallback handles them)
 
             // Get minimum execution time (matching EF's approach)
             var minExecutionTime = baseQuery
@@ -166,12 +168,22 @@ public Task<TimeTickerEntity[]> GetEarliestTimeTickers(CancellationToken cancell
             if (minExecutionTime == null)
                 return Task.FromResult(Array.Empty<TimeTickerEntity>());
 
-            // Get tasks within 50ms window of the earliest task for batching efficiency
-            var batchWindow = minExecutionTime.Value.AddMilliseconds(50);
+            // Round the minimum execution time down to its second
+            var minSecond = new DateTime(
+                minExecutionTime.Value.Year,
+                minExecutionTime.Value.Month,
+                minExecutionTime.Value.Day,
+                minExecutionTime.Value.Hour,
+                minExecutionTime.Value.Minute,
+                minExecutionTime.Value.Second,
+                DateTimeKind.Utc);
+
+            // Fetch all tickers within that complete second (this ensures we get all tickers in the same second)
+            var maxExecutionTime = minSecond.AddSeconds(1);
 
-            // Final query with mapping (matching EF's approach)
             var result = baseQuery
-                .Where(x => x.ExecutionTime.Value <= batchWindow)
+                .Where(x => x.ExecutionTime >= minSecond && x.ExecutionTime < maxExecutionTime)
+                .OrderBy(x => x.ExecutionTime)
                 .Select(ForQueueTimeTickers)  // Use same mapping as EF Core
                 .ToArray();
 
@@ -598,7 +610,7 @@ public Task<int> RemoveCronTickers(Guid[] cronTickerIds, CancellationToken cance
         public Task<CronTickerOccurrenceEntity<TCronTicker>> GetEarliestAvailableCronOccurrence(Guid[] ids, CancellationToken cancellationToken = default)
         {
             var now = _clock.UtcNow;
-            var mainSchedulerThreshold = now.AddMilliseconds(-100);  // Main scheduler handles tasks up to 100ms overdue
+            var mainSchedulerThreshold = now.AddSeconds(-1);  // Main scheduler handles items within the 1-second window
 
             var query = CronOccurrences.Values.AsEnumerable();
 
@@ -673,11 +685,11 @@ public async IAsyncEnumerable<CronTickerOccurrenceEntity<TCronTicker>> QueueCron
         public async IAsyncEnumerable<CronTickerOccurrenceEntity<TCronTicker>> QueueTimedOutCronTickerOccurrences([EnumeratorCancellation] CancellationToken cancellationToken = default)
         {
             var now = _clock.UtcNow;
-            var fallbackThreshold = now.AddMilliseconds(-100);  // Fallback picks up tasks overdue by > 100ms
+            var fallbackThreshold = now.AddSeconds(-1);  // Fallback picks up tasks older than main 1-second window
 
             var occurrencesToUpdate = CronOccurrences.Values
                 .Where(x => x.Status == TickerStatus.Idle || x.Status == TickerStatus.Queued)
-                .Where(x => x.ExecutionTime <= fallbackThreshold)  // Only tasks overdue by more than 100ms
+                .Where(x => x.ExecutionTime <= fallbackThreshold)  // Only tasks older than 1 second
                 .ToArray();
 
             foreach (var occurrence in occurrencesToUpdate)
@@ -1004,4 +1016,4 @@ private void ApplyFunctionContextToCronOccurrence(CronTickerOccurrenceEntity<TCr
 
         #endregion
     }
-}
+}

src/TickerQ/Src/TickerQThreadPool/TickerQTaskScheduler.cs

@@ -53,8 +53,11 @@ public TickerQTaskScheduler(
             _workerQueues[i] = new ConcurrentQueue<WorkItem>();
         }
 
-        // Start at least one worker immediately to handle incoming tasks
-        TryStartWorker();
+        // Start all workers upfront to honor maxConcurrency
+        for (int i = 0; i < _maxConcurrency; i++)
+        {
+            TryStartWorker();
+        }
     }
 
     /// <summary>
@@ -149,20 +152,14 @@ private void EnsureWorkerAvailable()
             TryStartWorker();
             return;
         }
-        
-        // Start more workers if we have queued tasks
+
+        // If there is queued work and we still have capacity, start another worker
         var totalQueued = _totalQueuedTasks;
         var activeWorkers = _activeWorkers;
-        
-        // If we have tasks but not enough workers, start more
+
         if (totalQueued > 0 && activeWorkers < _maxConcurrency)
         {
-            // Start workers proportional to load
-            var desiredWorkers = Math.Min(totalQueued, _maxConcurrency);
-            if (desiredWorkers > activeWorkers)
-            {
-                TryStartWorker();
-            }
+            TryStartWorker();
         }
     }
 
@@ -253,24 +250,7 @@ private async Task WorkerLoopCoreAsync(int workerId)
                 // No work found - check if we should exit
                 if (DateTime.UtcNow - lastWorkTime > _idleWorkerTimeout)
                 {
-                    // Check ALL queues for any remaining work before exiting
-                    bool anyWorkRemaining = false;
-                    for (int i = 0; i < _maxConcurrency; i++)
-                    {
-                        if (_workerQueues[i].Count > 0)
-                        {
-                            anyWorkRemaining = true;
-                            break;
-                        }
-                    }
-                    
-                    // Only exit if there's really no work and we have minimum workers
-                    if (!anyWorkRemaining && _totalQueuedTasks == 0 && _activeWorkers > 1)
-                    {
-                        break; // Exit this worker
-                    }
-                    
-                    // Reset timer if we need to stay
+                    // Keep workers alive to maintain maxConcurrency; just reset timer
                     lastWorkTime = DateTime.UtcNow;
                 }
 
@@ -337,8 +317,43 @@ private async Task ExecuteWorkAsync(WorkItem workItem)
             // Check cancellation before executing
             if (!workItem.UserToken.IsCancellationRequested && !_shutdownCts.Token.IsCancellationRequested)
             {
-                // Execute the work asynchronously - this won't block the worker
-                await workItem.Work(workItem.UserToken).ConfigureAwait(false);
+                // Start the work without awaiting it so this worker
+                // can continue processing other items while the task awaits.
+                var task = workItem.Work(workItem.UserToken);
+                
+                if (task == null)
+                    return;
+
+                if (!task.IsCompleted)
+                {
+                    // Observe completion and exceptions without blocking the worker loop
+                    _ = task.ContinueWith(t =>
+                    {
+                        try
+                        {
+                            if (t.IsFaulted)
+                            {
+                                _ = t.Exception;
+                            }
+                        }
+                        catch
+                        {
+                            // Swallow continuation exceptions
+                        }
+                    }, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
+                }
+                else
+                {
+                    // Task already completed synchronously – observe any exception
+                    try
+                    {
+                        await task.ConfigureAwait(false);
+                    }
+                    catch
+                    {
+                        // Swallow exceptions to keep worker alive
+                    }
+                }
             }
         }
         catch (OperationCanceledException)