Garages3 #83

Merged
khwezi merged 3 commits from garages3 into main 2026-09-18 16:12:07 +02:00
7 changed files with 68 additions and 57 deletions
Showing only changes of commit b703b0c64d - Show all commits
+5 -3
View File
@@ -1,12 +1,14 @@
using Microsoft.FeatureManagement;
using PostFundManagement.Application.Extensions; using PostFundManagement.Application.Extensions;
using PostFundManagement.Application.Filters; using PostFundManagement.Application.Filters;
using PostFundManagement.Domain.Extensions;
using PostFundManagement.Domain.Health; using PostFundManagement.Domain.Health;
using PostFundManagement.Domain.Mediator; using PostFundManagement.Domain.Mediator;
using static PostFundManagement.Domain.Extensions.Constants; using static PostFundManagement.Domain.Extensions.Constants;
var builder = WebApplication.CreateBuilder(args); var builder = WebApplication.CreateBuilder(args);
builder.Services.AddFeatureManagement();
builder.Services.AddEndpointsApiExplorer(); builder.Services.AddEndpointsApiExplorer();
builder.Services.AddApiVersioning(options => builder.Services.AddApiVersioning(options =>
{ {
@@ -120,9 +122,9 @@ builder.Services.AddCors(options =>
var app = builder.Build(); var app = builder.Build();
var schedulerFactory = app.Services.GetRequiredService<ISchedulerFactory>(); var schedulerFactory = app.Services.GetRequiredService<ISchedulerFactory>();
var scheduler = await schedulerFactory.GetScheduler(DefaultSchedulerName); var scheduler = await schedulerFactory.GetScheduler();
if (!scheduler!.IsStarted) if (scheduler!.Status != SchedulerStatus.Running)
await scheduler.Start(); await scheduler.Start();
app.UseHsts(); app.UseHsts();
@@ -2,7 +2,7 @@ using PostFundManagement.Domain.Abstractions;
using PostFundManagement.Domain.Quartz; using PostFundManagement.Domain.Quartz;
using static PostFundManagement.Domain.Extensions.Constants; using static PostFundManagement.Domain.Extensions.Constants;
namespace PostFundManagement.Domain.Extensions; namespace PostFundManagement.Application.Extensions;
public static class Quartz public static class Quartz
{ {
@@ -10,35 +10,33 @@ public static class Quartz
{ {
var connectionString = configuration.GetConnectionString(DatabaseConfigName); var connectionString = configuration.GetConnectionString(DatabaseConfigName);
services.Configure<QuartzOptions>(options =>
{
options.Properties["quartz.scheduler.instanceName"] = schedulerName;
options.Properties["quartz.scheduler.instanceId"] = "AUTO";
});
services.ConfigureCommon(); services.ConfigureCommon();
services.AddQuartz(config => services.AddQuartz(config =>
{ {
config.SchedulerName = schedulerName;
config.SchedulerId = "AUTO";
config.UseSimpleTypeLoader(); config.UseSimpleTypeLoader();
config.UseDefaultThreadPool(options => options.MaxConcurrency = 0); config.UseDefaultThreadPool(options => options.MaxConcurrency = 0);
config.UseTimeZoneConverter(); config.UseTimeZoneConverter();
config.UsePersistentStore(storage => config.UsePersistentStore(storage =>
{ {
storage.PerformSchemaValidation = false; storage.ProvisionSchema();
storage.UsePostgres(options => options.ConnectionString = connectionString!);
storage.UseSystemTextJsonSerializer(); storage.UseSystemTextJsonSerializer();
storage.SetProperty("quartz.jobStore.clustered", "true");
storage.UsePostgres(options =>
{
options.ConnectionString = connectionString!;
options.TablePrefix = "quartz.qrtz_";
});
storage.UseClustering(cluster => storage.UseClustering(cluster =>
{ {
cluster.CheckinInterval = TimeSpan.FromSeconds(30); cluster.CheckinInterval = TimeSpan.FromSeconds(30);
cluster.CheckinMisfireThreshold = TimeSpan.FromSeconds(90); cluster.CheckinMisfireThreshold = TimeSpan.FromSeconds(90);
}); });
}); });
}); });
return services; return services;
@@ -48,36 +46,37 @@ public static class Quartz
{ {
var connectionString = configuration.GetConnectionString(DatabaseConfigName); var connectionString = configuration.GetConnectionString(DatabaseConfigName);
services.Configure<QuartzOptions>(options =>
{
options.Properties["quartz.scheduler.instanceName"] = schedulerName;
options.Properties["quartz.scheduler.instanceId"] = "AUTO";
options.Properties["quartz.scheduler.batchTriggerAcquisitionMaxCount"] = "5";
options.Properties["quartz.jobStore.misfireThreshold"] = TimeSpan.FromMinutes(2).TotalMilliseconds.ToString(CultureInfo.InvariantCulture);
options.Properties["quartz.plugin.jobHistory.type"] = "Quartz.Plugin.History.LoggingJobHistoryPlugin, Quartz.Plugins";
options.Properties["quartz.plugin.triggerHistory.type"] = "Quartz.Plugin.History.LoggingTriggerHistoryPlugin, Quartz.Plugins";
});
services.ConfigureCommon(); services.ConfigureCommon();
services.AddQuartzHostedService(options => options.WaitForJobsToComplete = true); services.AddQuartzHostedService(options =>
{
options.WaitForJobsToComplete = true;
});
services.AddQuartz(config => services.AddQuartz(config =>
{ {
config.SchedulerName = schedulerName;
config.SchedulerId = "AUTO";
config.InterruptJobsOnShutdown = true;
config.InterruptJobsOnShutdownWithWait = true;
config.MaxBatchSize = 5;
config.UseSimpleTypeLoader(); config.UseSimpleTypeLoader();
config.UseDefaultThreadPool(options => options.MaxConcurrency = 1); config.UseDefaultThreadPool(options => options.MaxConcurrency = 5);
config.UseTimeZoneConverter(); config.UseTimeZoneConverter();
config.SetProperty("quartz.jobStore.misfireThreshold", TimeSpan.FromMinutes(2).TotalMilliseconds.ToString(CultureInfo.InvariantCulture));
config.UsePersistentStore(storage => config.UsePersistentStore(storage =>
{ {
storage.PerformSchemaValidation = false; storage.ProvisionSchema();
storage.UsePostgres(options => options.ConnectionString = connectionString!);
storage.UseSystemTextJsonSerializer(); storage.UseSystemTextJsonSerializer();
storage.SetProperty("quartz.jobStore.clustered", "true");
storage.UsePostgres(options =>
{
options.ConnectionString = connectionString!;
options.TablePrefix = "quartz.qrtz_";
});
storage.UseClustering(cluster => storage.UseClustering(cluster =>
{ {
cluster.CheckinInterval = TimeSpan.FromSeconds(30); cluster.CheckinInterval = TimeSpan.FromSeconds(30);
@@ -91,18 +90,10 @@ public static class Quartz
private static IServiceCollection ConfigureCommon(this IServiceCollection services) private static IServiceCollection ConfigureCommon(this IServiceCollection services)
{ {
services.Configure<QuartzOptions>(options => services.Configure<QuartzOptions>(options => options.Scheduling.IgnoreDuplicates = true);
{
options.Scheduling.IgnoreDuplicates = true;
options.Scheduling.OverWriteExistingData = true;
options["quartz.plugin.jobHistory.type"] = "Quartz.Plugin.History.LoggingJobHistoryPlugin, Quartz.Plugins";
options["quartz.plugin.triggerHistory.type"] = "Quartz.Plugin.History.LoggingTriggerHistoryPlugin, Quartz.Plugins";
});
services.AddTransient<RetryJobListener>(); services.AddTransient<RetryJobListener>();
services.AddTransient<IJobOrchestrator, JobOrchestrator>(); services.AddTransient<IJobOrchestrator, JobOrchestrator>();
return services; return services;
} }
} }
@@ -8,15 +8,15 @@ public sealed class QuartzHealthCheck(ISchedulerFactory schedulerFactory) : IHea
{ {
try try
{ {
var scheduler = await schedulerFactory.GetScheduler(DefaultSchedulerName, cancellationToken); var scheduler = await schedulerFactory.GetScheduler(cancellationToken);
if(scheduler == null) if (scheduler == null)
return HealthCheckResult.Unhealthy($"Scheduler with name '{DefaultSchedulerName}' not found."); return HealthCheckResult.Unhealthy($"Scheduler with name '{DefaultSchedulerName}' not found.");
if (!scheduler.IsStarted) if (scheduler.Status != SchedulerStatus.Running)
return HealthCheckResult.Unhealthy($"{DefaultSchedulerName} Quartz scheduler is not running"); return HealthCheckResult.Unhealthy($"{DefaultSchedulerName} Quartz scheduler is not running");
await scheduler.CheckExists(new JobKey(Guid.NewGuid().ToString()), cancellationToken); await scheduler.Exists(new JobKey(Guid.NewGuid().ToString()), cancellationToken);
return HealthCheckResult.Healthy($"{DefaultSchedulerName} Quartz scheduler is ready"); return HealthCheckResult.Healthy($"{DefaultSchedulerName} Quartz scheduler is ready");
} }
@@ -26,7 +26,7 @@ public sealed class JobOrchestrator(ISchedulerFactory schedulerFactory) : IJobOr
.StartNow() .StartNow()
.Build(); .Build();
await scheduler.ScheduleJob(job, new List<ITrigger> { trigger }.AsReadOnly(), replace: true, cancellationToken); await scheduler.ScheduleJob(job, new List<ITrigger> { trigger }.AsReadOnly(), new ScheduleJobOptions { Replace = true }, cancellationToken);
} }
public async ValueTask ScheduleAsync<TNotification>(TNotification notification, string cronExpression, CancellationToken cancellationToken = default) public async ValueTask ScheduleAsync<TNotification>(TNotification notification, string cronExpression, CancellationToken cancellationToken = default)
@@ -51,17 +51,16 @@ public sealed class JobOrchestrator(ISchedulerFactory schedulerFactory) : IJobOr
var trigger = global::Quartz.TriggerBuilder.Create() var trigger = global::Quartz.TriggerBuilder.Create()
.WithIdentity(triggerKey) .WithIdentity(triggerKey)
.WithDescription($"Scheduled via Main Job at {now:g}") .WithDescription($"Scheduled via Main Job at {now:g}")
.WithCronSchedule(cronExpression, cron => cron .WithCronSchedule(cronExpression, cron => cron.WithMisfireInstruction(CronTriggerMisfireInstruction.IgnoreMisfires))
.WithMisfireHandlingInstructionIgnoreMisfires())
.StartAt((DateTimeOffset)now) .StartAt((DateTimeOffset)now)
.Build(); .Build();
await scheduler.AddJob(job, replace: true, cancellationToken); await scheduler.AddJob(job, new AddJobOptions { Replace = true }, cancellationToken);
if (await scheduler.CheckExists(triggerKey, cancellationToken)) if (await scheduler.Exists(triggerKey, cancellationToken))
await scheduler.RescheduleJob(triggerKey, trigger, cancellationToken); await scheduler.RescheduleJob(triggerKey, trigger, cancellationToken);
else else
await scheduler.ScheduleJob(job, new List<ITrigger> { trigger }.AsReadOnly(), replace: true, cancellationToken); await scheduler.ScheduleJob(job, new List<ITrigger> { trigger }.AsReadOnly(), new ScheduleJobOptions { Replace = true }, cancellationToken);
} }
public async ValueTask<bool> InterruptAsync(string eventName, string? correlationId = null, CancellationToken cancellationToken = default) public async ValueTask<bool> InterruptAsync(string eventName, string? correlationId = null, CancellationToken cancellationToken = default)
@@ -82,6 +81,6 @@ public sealed class JobOrchestrator(ISchedulerFactory schedulerFactory) : IJobOr
jobGroup = "scheduled-jobs"; jobGroup = "scheduled-jobs";
} }
return await scheduler.Interrupt(JobKey.Create(jobKeyName, jobGroup), cancellationToken); return await scheduler.Interrupt(new JobKey(jobKeyName, jobGroup), cancellationToken);
} }
} }
@@ -7,7 +7,8 @@ namespace PostFundManagement.Domain.Quartz;
[DisallowConcurrentExecution] [DisallowConcurrentExecution]
public sealed class MediatorJob<TNotification>(IMediator mediator) : IJob where TNotification : IEvent public sealed class MediatorJob<TNotification>(IMediator mediator) : IJob where TNotification : IEvent
{ {
public async Task Execute(IJobExecutionContext context)
public async ValueTask Execute(IJobExecutionContext context, CancellationToken cancellationToken = default)
{ {
if (context.Recovering) if (context.Recovering)
Trace.WriteLine($"CRITICAL RECOVERY: Resurrecting job '{typeof(TNotification).Name}' after a previous cluster node crashed mid-execution."); Trace.WriteLine($"CRITICAL RECOVERY: Resurrecting job '{typeof(TNotification).Name}' after a previous cluster node crashed mid-execution.");
@@ -43,7 +44,7 @@ public sealed class MediatorJob<TNotification>(IMediator mediator) : IJob where
catch (OperationCanceledException) when (context.CancellationToken.IsCancellationRequested) catch (OperationCanceledException) when (context.CancellationToken.IsCancellationRequested)
{ {
Trace.WriteLine($"Job '{typeof(TNotification).Name}' was gracefully interrupted by the cluster control plane."); Trace.WriteLine($"Job '{typeof(TNotification).Name}' was gracefully interrupted by the cluster control plane.");
activity?.SetStatus(ActivityStatusCode.Ok); activity?.SetStatus(ActivityStatusCode.Ok);
return; return;
@@ -16,6 +16,23 @@ public sealed class RetryJobListener : IJobListener
return; return;
if (jobException is not null && context.RefireCount < RetryCount) if (jobException is not null && context.RefireCount < RetryCount)
jobException.RefireImmediately = true; {
if (context.JobDetail is null)
return;
var triggers = await context.Scheduler.GetTriggersOfJob(context.JobDetail.Key, cancellationToken);
var oldTrigger = triggers?.FirstOrDefault();
if (oldTrigger is null)
return;
var newTrigger = global::Quartz.TriggerBuilder.Create()
.WithIdentity($"{oldTrigger.Key.Name}-retry-{context.RefireCount}", oldTrigger.Key.Group)
.StartNow()
.ForJob(context.JobDetail.Key)
.Build();
await context.Scheduler.RescheduleJob(oldTrigger.Key, newTrigger, cancellationToken);
}
} }
} }
+1
View File
@@ -25,6 +25,7 @@ services:
# Postgres Database Overrides # Postgres Database Overrides
- ConnectionStrings__PfmDatabase=Server=${POSTGRES_HOST:-192.168.1.170};Port=${POSTGRES_PORT:-5432};Database=${POSTGRES_DB:-pfm};User Id=${POSTGRES_USER:-pfm-user};Password=${POSTGRES_PASSWORD}; - ConnectionStrings__PfmDatabase=Server=${POSTGRES_HOST:-192.168.1.170};Port=${POSTGRES_PORT:-5432};Database=${POSTGRES_DB:-pfm};User Id=${POSTGRES_USER:-pfm-user};Password=${POSTGRES_PASSWORD};
- ConnectionStrings__PfmSchedulerDatabase=Server=${POSTGRES_HOST:-192.168.1.170};Port=${POSTGRES_PORT:-5432};Database=pfm-scheduler;User Id=${POSTGRES_USER:-pfm-user};Password=${POSTGRES_PASSWORD};
# Garage S3 Settings # Garage S3 Settings
- GeneralS3Settings__ServiceUrl=${GENERAL_S3_SERVICE_URL:-https://s3.garage.khongisa.co.za} - GeneralS3Settings__ServiceUrl=${GENERAL_S3_SERVICE_URL:-https://s3.garage.khongisa.co.za}