notfire/Iceshrimp.NET
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

[backend/queue] Add queue processor timeout

Browse source 
For now this is hardcoded per queue, but this will be configurable down the line.
This commit is contained in:
Laura Hausmann 2024-06-13 22:41:41 +02:00
parent a7dcfbe75a
commit 35148d7796
No known key found for this signature in database
GPG key ID: D044E84C5BE01605
5 changed files with 19 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
Iceshrimp.Backend/Core/Queues/BackgroundTaskQueue.cs
View file

@ -13,7 +13,8 @@ using JR = System.Text.Json.Serialization.JsonRequiredAttribute;
namespace Iceshrimp.Backend.Core.Queues;
public class BackgroundTaskQueue(int parallelism)
: PostgresJobQueue<BackgroundTaskJobData>("background-task", BackgroundTaskQueueProcessorDelegateAsync, parallelism)
: PostgresJobQueue<BackgroundTaskJobData>("background-task", BackgroundTaskQueueProcessorDelegateAsync,
parallelism, TimeSpan.FromSeconds(60))
{
private static async Task BackgroundTaskQueueProcessorDelegateAsync(
Job job,

3
Iceshrimp.Backend/Core/Queues/DeliverQueue.cs
View file

@ -8,7 +8,8 @@ using JR = System.Text.Json.Serialization.JsonRequiredAttribute;
namespace Iceshrimp.Backend.Core.Queues;
public class DeliverQueue(int parallelism)
: PostgresJobQueue<DeliverJobData>("deliver", DeliverQueueProcessorDelegateAsync, parallelism)
: PostgresJobQueue<DeliverJobData>("deliver", DeliverQueueProcessorDelegateAsync,
parallelism, TimeSpan.FromSeconds(60))
{
private static async Task DeliverQueueProcessorDelegateAsync(
Job job, DeliverJobData jobData, IServiceProvider scope, CancellationToken token

2
Iceshrimp.Backend/Core/Queues/InboxQueue.cs
View file

@ -11,7 +11,7 @@ using JR = System.Text.Json.Serialization.JsonRequiredAttribute;
namespace Iceshrimp.Backend.Core.Queues;
public class InboxQueue(int parallelism)
: PostgresJobQueue<InboxJobData>("inbox", InboxQueueProcessorDelegateAsync, parallelism)
: PostgresJobQueue<InboxJobData>("inbox", InboxQueueProcessorDelegateAsync, parallelism, TimeSpan.FromSeconds(60))
{
private static async Task InboxQueueProcessorDelegateAsync(
Job job,

3
Iceshrimp.Backend/Core/Queues/PreDeliverQueue.cs
View file

@ -14,7 +14,8 @@ using JR = System.Text.Json.Serialization.JsonRequiredAttribute;
namespace Iceshrimp.Backend.Core.Queues;
public class PreDeliverQueue(int parallelism)
: PostgresJobQueue<PreDeliverJobData>("pre-deliver", PreDeliverQueueProcessorDelegateAsync, parallelism)
: PostgresJobQueue<PreDeliverJobData>("pre-deliver", PreDeliverQueueProcessorDelegateAsync,
parallelism, TimeSpan.FromSeconds(60))
{
private static async Task PreDeliverQueueProcessorDelegateAsync(
Job job, PreDeliverJobData jobData, IServiceProvider scope, CancellationToken token

18
Iceshrimp.Backend/Core/Services/QueueService.cs
View file

@ -201,7 +201,8 @@ public interface IPostgresJobQueue
public class PostgresJobQueue<T>(
string name,
Func<Job, T, IServiceProvider, CancellationToken, Task> handler,
int parallelism
int parallelism,
TimeSpan timeout
) : IPostgresJobQueue where T : class
{
private readonly AsyncAutoResetEvent _delayedChannel = new();
@ -424,7 +425,7 @@ public class PostgresJobQueue<T>(
try
{
await handler(job, data, scope.ServiceProvider, token);
await handler(job, data, scope.ServiceProvider, token).WaitAsync(timeout, token);
}
catch (Exception e)
{
@ -436,13 +437,18 @@ public class PostgresJobQueue<T>(
var queueName = data is BackgroundTaskJobData ? name + $" ({data.GetType().Name})" : name;
if (e is GracefulException { Details: not null } ce)
{
logger.LogError("Failed to process job {id} in {queue} queue: {error} - {details}",
queueName, job.Id.ToStringLower(), ce.Message, ce.Details);
logger.LogError("Failed to process job {id} in queue {queue}: {error} - {details}",
job.Id.ToStringLower(), queueName, ce.Message, ce.Details);
}
else if (e is TimeoutException)
{
logger.LogError("Job {id} in queue {queue} didn't complete within the configured timeout ({timeout} seconds)",
job.Id.ToStringLower(), queueName, (int)timeout.TotalSeconds);
}
else
{
logger.LogError(e, "Failed to process job {id} in {queue} queue: {error}", job.Id.ToStringLower(),
queueName, e);
logger.LogError(e, "Failed to process job {id} in queue {queue}: {error}",
job.Id.ToStringLower(), queueName, e);
}
}