diff --git a/docs/en/framework/infrastructure/background-jobs/index.md b/docs/en/framework/infrastructure/background-jobs/index.md index cdc4a1b5fb..57bffefbe0 100644 --- a/docs/en/framework/infrastructure/background-jobs/index.md +++ b/docs/en/framework/infrastructure/background-jobs/index.md @@ -225,7 +225,7 @@ ABP includes a simple `IBackgroundJobManager` implementation that; - **Retries** job execution until the job **successfully runs** or **timeouts**. Default timeout is 2 days for a job. Logs all exceptions. - **Deletes** a job from the store (database) when it's successfully executed. If it's timed out, it sets it as **abandoned** and leaves it in the database. - **Increasingly waits between retries** for a job. It waits 1 minute for the first retry, 2 minutes for the second retry, 4 minutes for the third retry and so on. -- **Polls** the store for jobs in fixed intervals. It queries jobs, ordering by priority (asc) and then by try count (asc). +- **Polls** the store for jobs in fixed intervals. It queries jobs, ordering by priority (desc) and then by try count (asc). > `Volo.Abp.BackgroundJobs` nuget package contains the default background job manager and it is installed to the startup templates by default. @@ -248,11 +248,76 @@ public class MyModule : AbpModule ```` * `JobPollPeriod` is used to determine the interval between two job polling operations. Default is 5000 ms (5 seconds). -* `MaxJobFetchCount` is used to determine the maximum job count to fetch in a single polling operation. Default is 1000. +* `MaxJobFetchCount` is used to determine the maximum job count to fetch in a single polling operation. It is also used as the batch size for the retention cleanup deletions. Default is 1000. * `DefaultFirstWaitDuration` is used to determine the duration to wait before the first retry. Default is 60 seconds. * `DefaultTimeout` is used to determine the timeout duration for a job. Default is 172800 seconds (2 days). * `DefaultWaitFactor` is used to determine the factor to increase the wait duration between retries. Default is 2.0. * `DistributedLockName` is used to determine the distributed lock name to use. Default is `AbpBackgroundJobWorker`. +* `StoreSuccessfulJobs` is used to determine whether to keep successfully completed jobs in the store instead of deleting them. Default is `false`. See the *Storing Successful Jobs* section. +* `SuccessfulJobRetentionTime` is used to determine how long a kept job is retained before the cleanup deletes it. Default is 7 days. Set to `null` to keep completed jobs forever. Only relevant when `StoreSuccessfulJobs` is enabled. +* `CleanSuccessfulJobsPeriod` is used to determine the interval between cleanup runs that delete expired completed jobs. Default is 3600000 ms (1 hour). +* `CleanupDistributedLockName` is used to determine the distributed lock name for the cleanup worker. Default is `AbpBackgroundJobCleanup`. +* `MaxParallelJobExecutionCount` is used to determine the maximum number of jobs a worker executes in parallel within one poll cycle. Default is 1. See the *Parallel Job Execution* section. +* `PerJobDistributedLockPrefix` is used to determine the prefix of the per-job distributed lock name used when `MaxParallelJobExecutionCount` is greater than 1. Default is `AbpBackgroundJob:`. + +### Storing Successful Jobs + +By default, the background job manager deletes a job from the store as soon as it runs successfully. If you want to keep completed jobs (for auditing or history), enable `StoreSuccessfulJobs`: + +````csharp +Configure(options => +{ + options.StoreSuccessfulJobs = true; + options.SuccessfulJobRetentionTime = TimeSpan.FromDays(30); //null to keep forever +}); +```` + +When enabled, a successful job is not deleted; instead its `CompletionTime` is set and it stays in the store. Completed jobs are excluded from the waiting jobs query, so they are not executed again. A cleanup worker periodically deletes completed jobs older than `SuccessfulJobRetentionTime`. + +> **Note:** The `IBackgroundJobStore` interface has new overloads (a `GetWaitingJobsAsync` overload that takes a job name filter and a `DeleteAsync` overload for cleanup). If you have a custom `IBackgroundJobStore` implementation, you must implement them for your code to compile. The built-in stores already implement them. + +### Dedicated Workers per Job Type + +By default, a single worker processes all job types. If you want to process certain job types separately (for example, slow or high-volume jobs), you can register dedicated workers, each handling only the specified job argument types with its own distributed lock: + +````csharp +Configure(options => +{ + options.AddDedicatedWorker("NotificationWorkerLock"); + options.AddDedicatedWorker("ReportWorkerLock"); +}); +```` + +Each dedicated worker processes only its configured job types. An additional default worker is automatically started to process all the remaining job types. In sequential mode, each worker (including the default one) runs independently under its own distributed lock (see *Parallel Job Execution* for how this changes when running jobs in parallel). + +If you don't want to specify a lock name, use the overloads without the `lockName` parameter; a stable, length-bounded lock name is then derived from the job argument types: + +````csharp +Configure(options => +{ + options.AddDedicatedWorker(); + options.AddDedicatedWorker(); +}); +```` + +> **Note:** Each job type can be handled by only one dedicated worker, and each worker must have a unique lock name; `AddDedicatedWorker` throws if this is violated. Dedicated workers require an `IBackgroundJobStore` that can filter jobs by name (the built-in stores can). + +### Parallel Job Execution + +By default, a worker executes waiting jobs one by one under a single worker-level distributed lock, so only one job runs at a time across all application instances. If you want to execute multiple jobs concurrently, set `MaxParallelJobExecutionCount` to a value greater than 1: + +````csharp +Configure(options => +{ + options.MaxParallelJobExecutionCount = 4; +}); +```` + +When it is greater than 1, the worker-level lock is not used. Instead, each job is claimed with its own distributed lock, so multiple application instances can execute different jobs at the same time. With a properly configured distributed lock provider, a job is not executed by more than one instance at a time. + +`MaxParallelJobExecutionCount` is a per-worker, per-poll-cycle limit — it is not a cluster-wide limit. A worker first fetches up to `MaxJobFetchCount` waiting jobs, then executes up to `MaxParallelJobExecutionCount` of them in parallel, so a single worker runs up to `min(MaxJobFetchCount, MaxParallelJobExecutionCount)` jobs per cycle; keep `MaxJobFetchCount` at least as large as `MaxParallelJobExecutionCount` to avoid capping the parallelism. When you also configure dedicated workers, each worker runs its own timer and claims up to `MaxParallelJobExecutionCount` jobs, so the effective concurrency is up to (number of workers) × `MaxParallelJobExecutionCount` per application instance, and up to (number of application instances) × (number of workers) × `MaxParallelJobExecutionCount` across the whole cluster. + +> **Important:** Configure `MaxParallelJobExecutionCount` and `PerJobDistributedLockPrefix` consistently across all application instances. Mixing sequential (worker lock) and parallel (per-job lock) instances removes the common mutual exclusion, and a different prefix produces a different per-job lock name for the same job — either case may let the same job run on more than one instance. As with the sequential mode, configure a real [distributed lock](../distributed-locking.md) provider for clustered deployments. ### Data Store diff --git a/docs/en/modules/background-jobs.md b/docs/en/modules/background-jobs.md index 971284b7c9..2d57d0c907 100644 --- a/docs/en/modules/background-jobs.md +++ b/docs/en/modules/background-jobs.md @@ -33,6 +33,8 @@ Following custom repositories are defined for this module: - `IBackgroundJobRepository` +> `IBackgroundJobRepository` supports filtering the waiting jobs for dedicated workers and cleaning up retained completed jobs. See the *Dedicated Workers per Job Type* and *Storing Successful Jobs* sections of the [background jobs](../framework/infrastructure/background-jobs) document. + ### Database providers #### Common diff --git a/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/AbpBackgroundJobWorkerOptions.cs b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/AbpBackgroundJobWorkerOptions.cs index b339508028..f0dc0aa500 100644 --- a/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/AbpBackgroundJobWorkerOptions.cs +++ b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/AbpBackgroundJobWorkerOptions.cs @@ -1,4 +1,8 @@ -namespace Volo.Abp.BackgroundJobs; +using System; +using System.Collections.Generic; +using System.Linq; + +namespace Volo.Abp.BackgroundJobs; public class AbpBackgroundJobWorkerOptions { @@ -15,6 +19,7 @@ public class AbpBackgroundJobWorkerOptions /// /// Maximum count of jobs to fetch from data store in one loop. + /// Also used as the batch size for the retention cleanup deletions (see ). /// Default: 1000. /// public int MaxJobFetchCount { get; set; } @@ -42,7 +47,61 @@ public class AbpBackgroundJobWorkerOptions /// Distributed lock name for the worker. /// Default value: "AbpBackgroundJobWorker". /// - public string DistributedLockName { get; set; } + public string DistributedLockName { get; set; } + + /// + /// When set to true, a successfully completed job is kept (its is set) + /// instead of being deleted, so it can be used for auditing/history. Completed jobs are excluded from the + /// waiting jobs query and are removed by the retention cleanup (see ). + /// Default value: false (successful jobs are deleted). + /// + public bool StoreSuccessfulJobs { get; set; } + + /// + /// How long a kept (successfully completed) job is retained before the cleanup deletes it. + /// Only relevant when is true. Set to null to keep them forever (no automatic cleanup). + /// Default value: 7 days. + /// + public TimeSpan? SuccessfulJobRetentionTime { get; set; } + + /// + /// Interval (as milliseconds) between cleanup runs that delete retained successful jobs older than . + /// Default value: 3,600,000 (1 hour). + /// + public int CleanSuccessfulJobsPeriod { get; set; } + + /// + /// Distributed lock name for the cleanup worker. + /// Default value: "AbpBackgroundJobCleanup". + /// + public string CleanupDistributedLockName { get; set; } + + /// + /// Dedicated workers, each processing only the configured job types with its own distributed lock. + /// When this list is not empty, an additional default worker is started that processes all + /// other job types. When it is empty, a single default worker processes all job types (the default behavior). + /// Use to add a dedicated worker. + /// + public List WorkerConfigurations { get; } + + /// + /// Maximum number of jobs that a single worker executes in parallel within one poll cycle. + /// When it is 1 (default), jobs are executed sequentially under a single worker distributed lock + /// (the default behavior). When greater than 1, the worker distributed lock is not used; instead + /// each job is claimed with its own distributed lock so multiple application instances can execute + /// different jobs concurrently. + /// This value should be configured consistently across all application instances: mixing sequential + /// (worker lock) and parallel (per-job lock) instances removes the common mutual exclusion and may + /// let the same job run on more than one instance. + /// Default value: 1. + /// + public int MaxParallelJobExecutionCount { get; set; } + + /// + /// Prefix of the per-job distributed lock name used when is greater than 1. + /// Default value: "AbpBackgroundJob:". + /// + public string PerJobDistributedLockPrefix { get; set; } public AbpBackgroundJobWorkerOptions() { @@ -52,5 +111,97 @@ public class AbpBackgroundJobWorkerOptions DefaultTimeout = 172800; DefaultWaitFactor = 2.0; DistributedLockName = "AbpBackgroundJobWorker"; + WorkerConfigurations = new List(); + MaxParallelJobExecutionCount = 1; + PerJobDistributedLockPrefix = "AbpBackgroundJob:"; + SuccessfulJobRetentionTime = TimeSpan.FromDays(7); + CleanSuccessfulJobsPeriod = 3600000; + CleanupDistributedLockName = "AbpBackgroundJobCleanup"; + } + + /// + /// Adds a dedicated worker that processes only the given job argument types. + /// + /// A unique distributed lock name for this worker. + /// The job argument types handled exclusively by this worker. + public AbpBackgroundJobWorkerOptions AddDedicatedWorker(string lockName, params Type[] jobArgsTypes) + { + Check.NotNullOrEmpty(jobArgsTypes, nameof(jobArgsTypes)); + + var configuration = new BackgroundJobWorkerConfiguration(lockName, jobArgsTypes.Distinct().ToArray()); + + var duplicateType = configuration.JobArgsTypes.FirstOrDefault( + type => WorkerConfigurations.Any(c => c.JobArgsTypes.Contains(type))); + if (duplicateType != null) + { + throw new AbpException( + $"The background job args type '{duplicateType.FullName}' is already assigned to a dedicated worker. " + + $"Each job type can be handled by only one dedicated worker."); + } + + if (lockName == DistributedLockName || WorkerConfigurations.Any(c => c.LockName == lockName)) + { + throw new AbpException( + $"The distributed lock name '{lockName}' is already used by another background job worker. " + + $"Each worker must have a unique lock name."); + } + + WorkerConfigurations.Add(configuration); + return this; + } + + public AbpBackgroundJobWorkerOptions AddDedicatedWorker(string lockName) + { + return AddDedicatedWorker(lockName, typeof(TArgs)); + } + + public AbpBackgroundJobWorkerOptions AddDedicatedWorker(string lockName) + { + return AddDedicatedWorker(lockName, typeof(TArgs1), typeof(TArgs2)); + } + + public AbpBackgroundJobWorkerOptions AddDedicatedWorker(string lockName) + { + return AddDedicatedWorker(lockName, typeof(TArgs1), typeof(TArgs2), typeof(TArgs3)); + } + + /// + /// Adds a dedicated worker with a lock name derived from the given job argument type names. + /// Use the AddDedicatedWorker(string lockName, ...) overloads to set an explicit lock name. + /// + /// The job argument types handled exclusively by this worker. + public AbpBackgroundJobWorkerOptions AddDedicatedWorker(params Type[] jobArgsTypes) + { + return AddDedicatedWorker(GetDedicatedWorkerLockName(jobArgsTypes), jobArgsTypes); + } + + public AbpBackgroundJobWorkerOptions AddDedicatedWorker() + { + return AddDedicatedWorker(typeof(TArgs)); + } + + public AbpBackgroundJobWorkerOptions AddDedicatedWorker() + { + return AddDedicatedWorker(typeof(TArgs1), typeof(TArgs2)); + } + + public AbpBackgroundJobWorkerOptions AddDedicatedWorker() + { + return AddDedicatedWorker(typeof(TArgs1), typeof(TArgs2), typeof(TArgs3)); + } + + protected virtual string GetDedicatedWorkerLockName(Type[] jobArgsTypes) + { + Check.NotNullOrEmpty(jobArgsTypes, nameof(jobArgsTypes)); + + if (jobArgsTypes.Any(t => t == null)) + { + throw new ArgumentException("Job args types cannot contain null.", nameof(jobArgsTypes)); + } + + // Hash the (stable, sorted) full type names so the derived lock name stays short and within the + // length limits of distributed lock providers (e.g. SQL Server sp_getapplock is limited to 255 chars). + var key = string.Join(",", jobArgsTypes.Select(t => t.FullName).Distinct().OrderBy(n => n, StringComparer.Ordinal)); + return "AbpBackgroundJobDedicatedWorker:" + key.ToMd5(); } } diff --git a/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/AbpBackgroundJobsModule.cs b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/AbpBackgroundJobsModule.cs index b3601ad1be..0f4b20cd65 100644 --- a/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/AbpBackgroundJobsModule.cs +++ b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/AbpBackgroundJobsModule.cs @@ -1,4 +1,4 @@ -using System.Threading.Tasks; +using System.Threading.Tasks; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; using Volo.Abp.BackgroundWorkers; @@ -23,9 +23,15 @@ public class AbpBackgroundJobsModule : AbpModule { public override async Task OnApplicationInitializationAsync(ApplicationInitializationContext context) { - if (context.ServiceProvider.GetRequiredService>().Value.IsJobExecutionEnabled) + // The manager decides (based on options) whether and which workers to run. + await context.AddBackgroundWorkerAsync(); + + // Only register the cleanup worker when it has something to do (retained successful jobs). + var jobOptions = context.ServiceProvider.GetRequiredService>().Value; + var workerOptions = context.ServiceProvider.GetRequiredService>().Value; + if (jobOptions.IsJobExecutionEnabled && workerOptions.StoreSuccessfulJobs && workerOptions.SuccessfulJobRetentionTime != null) { - await context.AddBackgroundWorkerAsync(); + await context.AddBackgroundWorkerAsync(); } } diff --git a/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobCleanupWorker.cs b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobCleanupWorker.cs new file mode 100644 index 0000000000..72c36b8ffb --- /dev/null +++ b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobCleanupWorker.cs @@ -0,0 +1,66 @@ +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using Volo.Abp.BackgroundWorkers; +using Volo.Abp.DistributedLocking; +using Volo.Abp.Threading; +using Volo.Abp.Timing; + +namespace Volo.Abp.BackgroundJobs; + +/// +/// Periodically deletes retained successfully completed jobs older than +/// . +/// Only relevant when is enabled. +/// +public class BackgroundJobCleanupWorker : AsyncPeriodicBackgroundWorkerBase +{ + protected AbpBackgroundJobOptions JobOptions { get; } + + protected AbpBackgroundJobWorkerOptions WorkerOptions { get; } + + protected IAbpDistributedLock DistributedLock { get; } + + public BackgroundJobCleanupWorker( + AbpAsyncTimer timer, + IServiceScopeFactory serviceScopeFactory, + IOptions jobOptions, + IOptions workerOptions, + IAbpDistributedLock distributedLock) + : base(timer, serviceScopeFactory) + { + JobOptions = jobOptions.Value; + WorkerOptions = workerOptions.Value; + DistributedLock = distributedLock; + Timer.Period = WorkerOptions.CleanSuccessfulJobsPeriod; + } + + protected override async Task DoWorkAsync(PeriodicBackgroundWorkerContext workerContext) + { + if (!JobOptions.IsJobExecutionEnabled || + !WorkerOptions.StoreSuccessfulJobs || + WorkerOptions.SuccessfulJobRetentionTime == null) + { + return; + } + + var store = workerContext.ServiceProvider.GetRequiredService(); + var clock = workerContext.ServiceProvider.GetRequiredService(); + var completedBefore = clock.Now.Subtract(WorkerOptions.SuccessfulJobRetentionTime.Value); + + await using (var handle = await DistributedLock.TryAcquireAsync(WorkerOptions.CleanupDistributedLockName, cancellationToken: StoppingToken)) + { + if (handle == null) + { + return; + } + + int deletedCount; + do + { + deletedCount = await store.DeleteAsync(WorkerOptions.ApplicationName, completedBefore, WorkerOptions.MaxJobFetchCount, StoppingToken); + } + while (deletedCount > 0 && deletedCount >= WorkerOptions.MaxJobFetchCount && !StoppingToken.IsCancellationRequested); + } + } +} diff --git a/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobInfo.cs b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobInfo.cs index 4a9cec922c..51f73ae7b8 100644 --- a/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobInfo.cs +++ b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobInfo.cs @@ -50,6 +50,14 @@ public class BackgroundJobInfo /// public virtual bool IsAbandoned { get; set; } + /// + /// The time this job was completed successfully. + /// When set, the job is kept as history and excluded from the waiting jobs query. + /// It is only set when is enabled; + /// otherwise successfully completed jobs are deleted. + /// + public virtual DateTime? CompletionTime { get; set; } + /// /// Priority of this job. /// diff --git a/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobNameFilter.cs b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobNameFilter.cs new file mode 100644 index 0000000000..06c0ebf398 --- /dev/null +++ b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobNameFilter.cs @@ -0,0 +1,71 @@ +using System; +using System.Collections.Generic; +using System.Linq; + +namespace Volo.Abp.BackgroundJobs; + +/// +/// Filters the waiting jobs of a background job worker by job name. +/// A worker is exactly one of: no filter (), include-only (a dedicated worker) or +/// exclude-only (the default worker in a multi-worker setup) — the two can never be combined. +/// +public class BackgroundJobNameFilter +{ + /// + /// A filter that matches every job name. + /// + public static BackgroundJobNameFilter None { get; } = new(BackgroundJobNameFilterMode.None); + + public BackgroundJobNameFilterMode Mode { get; } + + public IReadOnlyList JobNames { get; } + + public BackgroundJobNameFilter(BackgroundJobNameFilterMode mode, IReadOnlyList? jobNames = null) + { + if (!Enum.IsDefined(typeof(BackgroundJobNameFilterMode), mode)) + { + throw new ArgumentException($"Invalid background job name filter mode: {mode}", nameof(mode)); + } + + var names = jobNames?.Where(x => !x.IsNullOrWhiteSpace()).Distinct(StringComparer.Ordinal).ToList() ?? new List(); + + if (mode == BackgroundJobNameFilterMode.None && names.Count > 0) + { + throw new ArgumentException("Job names must be empty when the filter mode is None.", nameof(jobNames)); + } + + if (mode != BackgroundJobNameFilterMode.None && names.Count == 0) + { + throw new ArgumentException("Job names cannot be empty when the filter mode is Include or Exclude.", nameof(jobNames)); + } + + Mode = mode; + JobNames = names.AsReadOnly(); + } + + public static BackgroundJobNameFilter Include(IReadOnlyList jobNames) + { + return new BackgroundJobNameFilter(BackgroundJobNameFilterMode.Include, jobNames); + } + + public static BackgroundJobNameFilter Exclude(IReadOnlyList jobNames) + { + return new BackgroundJobNameFilter(BackgroundJobNameFilterMode.Exclude, jobNames); + } + + /// + /// Whether the given job name passes this filter, using an ordinal (case-sensitive) comparison for the + /// in-memory eligibility re-check. The persistent stores translate and + /// into a database query instead, so their filtering follows the database collation. + /// Job names are expected to be unique beyond case (they are derived from the type name by default). + /// + public virtual bool IsMatch(string jobName) + { + return Mode switch + { + BackgroundJobNameFilterMode.Include => JobNames.Contains(jobName, StringComparer.Ordinal), + BackgroundJobNameFilterMode.Exclude => !JobNames.Contains(jobName, StringComparer.Ordinal), + _ => true + }; + } +} diff --git a/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobNameFilterMode.cs b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobNameFilterMode.cs new file mode 100644 index 0000000000..6ed2801b44 --- /dev/null +++ b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobNameFilterMode.cs @@ -0,0 +1,19 @@ +namespace Volo.Abp.BackgroundJobs; + +public enum BackgroundJobNameFilterMode : byte +{ + /// + /// No filter; all job names match. + /// + None = 0, + + /// + /// Only the job names in the filter match. + /// + Include = 1, + + /// + /// All job names except those in the filter match. + /// + Exclude = 2 +} diff --git a/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobWorker.cs b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobWorker.cs index a015e32d66..3053bb426f 100644 --- a/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobWorker.cs +++ b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobWorker.cs @@ -1,17 +1,22 @@ -using System; +using System; +using System.Collections.Generic; using System.Linq; +using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using Volo.Abp.BackgroundWorkers; +using Volo.Abp.DependencyInjection; using Volo.Abp.DistributedLocking; +using Volo.Abp.ExceptionHandling; using Volo.Abp.Threading; using Volo.Abp.Timing; namespace Volo.Abp.BackgroundJobs; -public class BackgroundJobWorker : AsyncPeriodicBackgroundWorkerBase, IBackgroundJobWorker +public class BackgroundJobWorker : IBackgroundJobWorker, ITransientDependency { protected AbpBackgroundJobOptions JobOptions { get; } @@ -19,95 +24,321 @@ public class BackgroundJobWorker : AsyncPeriodicBackgroundWorkerBase, IBackgroun protected IAbpDistributedLock DistributedLock { get; } + protected IServiceScopeFactory ServiceScopeFactory { get; } + + protected AbpAsyncTimer Timer { get; } + + public ILogger Logger { get; set; } + + protected string DistributedLockName { get; set; } = default!; + + protected BackgroundJobNameFilter JobNameFilter { get; set; } = BackgroundJobNameFilter.None; + + protected CancellationTokenSource StoppingTokenSource { get; } + + protected CancellationToken StoppingToken { get; } + public BackgroundJobWorker( AbpAsyncTimer timer, IOptions jobOptions, IOptions workerOptions, IServiceScopeFactory serviceScopeFactory, IAbpDistributedLock distributedLock) - : base( - timer, - serviceScopeFactory) { + Timer = timer; DistributedLock = distributedLock; + ServiceScopeFactory = serviceScopeFactory; WorkerOptions = workerOptions.Value; JobOptions = jobOptions.Value; + Logger = NullLogger.Instance; + Timer.Period = WorkerOptions.JobPollPeriod; + Timer.Elapsed = TimerOnElapsed; + + StoppingTokenSource = new CancellationTokenSource(); + StoppingToken = StoppingTokenSource.Token; + } + + public virtual Task StartAsync( + string? distributedLockName = null, + BackgroundJobNameFilter? jobNameFilter = null, + CancellationToken cancellationToken = default) + { + DistributedLockName = distributedLockName ?? WorkerOptions.DistributedLockName; + JobNameFilter = jobNameFilter ?? BackgroundJobNameFilter.None; + + Timer.Start(cancellationToken); + + return Task.CompletedTask; + } + + public virtual Task StopAsync(CancellationToken cancellationToken = default) + { + StoppingTokenSource.Cancel(); + Timer.Stop(cancellationToken); + StoppingTokenSource.Dispose(); + + return Task.CompletedTask; + } + + private async Task TimerOnElapsed(AbpAsyncTimer timer) + { + await RunAsync(); + } + + protected virtual async Task RunAsync() + { + using var scope = ServiceScopeFactory.CreateScope(); + + try + { + var workerContext = new PeriodicBackgroundWorkerContext(scope.ServiceProvider, StoppingToken); + + if (WorkerOptions.MaxParallelJobExecutionCount > 1) + { + await ExecuteJobsInParallelAsync(workerContext); + } + else + { + await ExecuteJobsWithWorkerLockAsync(workerContext); + } + } + catch (Exception ex) + { + await scope.ServiceProvider + .GetRequiredService() + .NotifyAsync(new ExceptionNotificationContext(ex)); + + Logger.LogException(ex); + } } - protected override async Task DoWorkAsync(PeriodicBackgroundWorkerContext workerContext) + protected virtual async Task ExecuteJobsWithWorkerLockAsync(PeriodicBackgroundWorkerContext workerContext) { - await using (var handler = await DistributedLock.TryAcquireAsync(WorkerOptions.DistributedLockName, cancellationToken: StoppingToken)) + await using (var handler = await DistributedLock.TryAcquireAsync(DistributedLockName, cancellationToken: StoppingToken)) { if (handler != null) { - var store = workerContext.ServiceProvider.GetRequiredService(); + await ExecuteWaitingJobsAsync(workerContext); + } + else + { + await WaitForNextTryAsync(); + } + } + } + + protected virtual async Task ExecuteWaitingJobsAsync(PeriodicBackgroundWorkerContext workerContext) + { + var store = workerContext.ServiceProvider.GetRequiredService(); + + var waitingJobs = await GetWaitingJobsAsync(workerContext, store); - var waitingJobs = await store.GetWaitingJobsAsync(WorkerOptions.ApplicationName, WorkerOptions.MaxJobFetchCount); + if (!waitingJobs.Any()) + { + return; + } - if (!waitingJobs.Any()) + var jobExecuter = workerContext.ServiceProvider.GetRequiredService(); + var clock = workerContext.ServiceProvider.GetRequiredService(); + var serializer = workerContext.ServiceProvider.GetRequiredService(); + + foreach (var jobInfo in waitingJobs) + { + await TryExecuteJobAsync(workerContext, store, jobInfo, jobExecuter, clock, serializer); + } + } + + /// + /// Executes waiting jobs in parallel across application instances, up to + /// jobs per cycle. + /// + protected virtual async Task ExecuteJobsInParallelAsync(PeriodicBackgroundWorkerContext workerContext) + { + var store = workerContext.ServiceProvider.GetRequiredService(); + + var waitingJobs = await GetWaitingJobsAsync(workerContext, store); + + if (!waitingJobs.Any()) + { + return; + } + + var runningTasks = new List(); + + // Await already-started jobs even if acquiring a lock for a later job throws, + // so no claimed job is left running detached from this cycle. + try + { + foreach (var jobInfo in waitingJobs) + { + if (runningTasks.Count >= WorkerOptions.MaxParallelJobExecutionCount || StoppingToken.IsCancellationRequested) { - return; + break; } - var jobExecuter = workerContext.ServiceProvider.GetRequiredService(); - var clock = workerContext.ServiceProvider.GetRequiredService(); - var serializer = workerContext.ServiceProvider.GetRequiredService(); - - foreach (var jobInfo in waitingJobs) + var handle = await DistributedLock.TryAcquireAsync(GetPerJobDistributedLockName(jobInfo), cancellationToken: StoppingToken); + if (handle == null) { - jobInfo.TryCount++; - jobInfo.LastTryTime = clock.Now; - - try - { - var jobConfiguration = JobOptions.GetJob(jobInfo.JobName); - var jobArgs = serializer.Deserialize(jobInfo.JobArgs, jobConfiguration.ArgsType); - var context = new JobExecutionContext( - workerContext.ServiceProvider, - jobConfiguration.JobType, - jobArgs, - workerContext.CancellationToken); - - try - { - await jobExecuter.ExecuteAsync(context); - - await store.DeleteAsync(jobInfo.Id); - } - catch (BackgroundJobExecutionException) - { - var nextTryTime = CalculateNextTryTime(jobInfo, clock); - - if (nextTryTime.HasValue) - { - jobInfo.NextTryTime = nextTryTime.Value; - } - else - { - jobInfo.IsAbandoned = true; - } - - await TryUpdateAsync(store, jobInfo); - } - } - catch (Exception ex) - { - Logger.LogException(ex); - jobInfo.IsAbandoned = true; - await TryUpdateAsync(store, jobInfo); - } + // Another instance is already processing this job. + continue; } + + runningTasks.Add(ExecuteClaimedJobAsync(jobInfo, handle)); } - else + } + finally + { + await Task.WhenAll(runningTasks); + } + } + + protected virtual async Task ExecuteClaimedJobAsync(BackgroundJobInfo jobInfo, IAbpDistributedLockHandle handle) + { + await using (handle) + { + // Each concurrently executed job runs in its own service scope so that scoped services + // (e.g. the DbContext and the unit of work) are not shared across parallel jobs. + using var scope = ServiceScopeFactory.CreateScope(); + + try { - try + var workerContext = new PeriodicBackgroundWorkerContext(scope.ServiceProvider, StoppingToken); + var store = scope.ServiceProvider.GetRequiredService(); + var clock = scope.ServiceProvider.GetRequiredService(); + + // Re-read under the lock: another instance may have completed, abandoned or rescheduled this job + // between fetching the waiting list and acquiring the per-job lock. + var currentJobInfo = await store.FindAsync(jobInfo.Id); + if (!IsJobEligible(currentJobInfo, clock)) { - await Task.Delay(WorkerOptions.JobPollPeriod * 12, StoppingToken); + return; } - catch (TaskCanceledException) { } + + var jobExecuter = scope.ServiceProvider.GetRequiredService(); + var serializer = scope.ServiceProvider.GetRequiredService(); + + await TryExecuteJobAsync(workerContext, store, currentJobInfo, jobExecuter, clock, serializer); + } + catch (Exception ex) + { + await scope.ServiceProvider + .GetRequiredService() + .NotifyAsync(new ExceptionNotificationContext(ex)); + + Logger.LogException(ex); + } + } + } + + protected virtual bool IsJobEligible(BackgroundJobInfo? jobInfo, IClock clock) + { + return jobInfo != null && + jobInfo.ApplicationName == WorkerOptions.ApplicationName && + !jobInfo.IsAbandoned && + jobInfo.CompletionTime == null && + jobInfo.NextTryTime <= clock.Now && + JobNameFilter.IsMatch(jobInfo.JobName); + } + + protected virtual string GetPerJobDistributedLockName(BackgroundJobInfo jobInfo) + { + return WorkerOptions.PerJobDistributedLockPrefix + jobInfo.Id; + } + + protected virtual async Task> GetWaitingJobsAsync( + PeriodicBackgroundWorkerContext workerContext, + IBackgroundJobStore store) + { + return await store.GetWaitingJobsAsync( + WorkerOptions.ApplicationName, + WorkerOptions.MaxJobFetchCount, + JobNameFilter); + } + + protected virtual async Task TryExecuteJobAsync( + PeriodicBackgroundWorkerContext workerContext, + IBackgroundJobStore store, + BackgroundJobInfo jobInfo, + IBackgroundJobExecuter jobExecuter, + IClock clock, + IBackgroundJobSerializer serializer) + { + jobInfo.TryCount++; + jobInfo.LastTryTime = clock.Now; + + try + { + var jobConfiguration = JobOptions.GetJob(jobInfo.JobName); + var jobArgs = serializer.Deserialize(jobInfo.JobArgs, jobConfiguration.ArgsType); + var context = new JobExecutionContext( + workerContext.ServiceProvider, + jobConfiguration.JobType, + jobArgs, + workerContext.CancellationToken); + + try + { + await jobExecuter.ExecuteAsync(context); + + await HandleJobSuccessAsync(store, jobInfo, clock); + } + catch (BackgroundJobExecutionException) + { + await HandleJobFailureAsync(store, jobInfo, clock); } } + catch (Exception ex) + { + await HandleJobErrorAsync(store, jobInfo, ex); + } + } + + protected virtual async Task HandleJobSuccessAsync(IBackgroundJobStore store, BackgroundJobInfo jobInfo, IClock clock) + { + if (WorkerOptions.StoreSuccessfulJobs) + { + // Keep the job as history: mark it completed instead of deleting. It is then excluded from the + // waiting jobs query and removed later by the retention cleanup. + jobInfo.CompletionTime = clock.Now; + await store.UpdateAsync(jobInfo); + } + else + { + await store.DeleteAsync(jobInfo.Id); + } + } + + protected virtual async Task HandleJobFailureAsync(IBackgroundJobStore store, BackgroundJobInfo jobInfo, IClock clock) + { + var nextTryTime = CalculateNextTryTime(jobInfo, clock); + + if (nextTryTime.HasValue) + { + jobInfo.NextTryTime = nextTryTime.Value; + } + else + { + jobInfo.IsAbandoned = true; + } + + await TryUpdateAsync(store, jobInfo); + } + + protected virtual async Task HandleJobErrorAsync(IBackgroundJobStore store, BackgroundJobInfo jobInfo, Exception ex) + { + Logger.LogException(ex); + jobInfo.IsAbandoned = true; + await TryUpdateAsync(store, jobInfo); + } + + protected virtual async Task WaitForNextTryAsync() + { + try + { + await Task.Delay(WorkerOptions.JobPollPeriod * 12, StoppingToken); + } + catch (TaskCanceledException) { } } protected virtual async Task TryUpdateAsync(IBackgroundJobStore store, BackgroundJobInfo jobInfo) diff --git a/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobWorkerConfiguration.cs b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobWorkerConfiguration.cs new file mode 100644 index 0000000000..848a2a6a1c --- /dev/null +++ b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobWorkerConfiguration.cs @@ -0,0 +1,37 @@ +using System; +using System.Collections.Generic; +using System.Linq; + +namespace Volo.Abp.BackgroundJobs; + +/// +/// Configuration of a dedicated that processes only specific job types. +/// +public class BackgroundJobWorkerConfiguration +{ + /// + /// A unique distributed lock name for this worker. It must be different from the names used by other workers. + /// It is used to serialize the worker across application instances when + /// is 1; in parallel mode + /// (greater than 1) jobs are claimed with per-job locks instead and this lock is not acquired. + /// + public string LockName { get; } + + /// + /// The job argument types that are processed exclusively by this worker. + /// + public IReadOnlyList JobArgsTypes { get; } + + public BackgroundJobWorkerConfiguration(string lockName, params Type[] jobArgsTypes) + { + LockName = Check.NotNullOrWhiteSpace(lockName, nameof(lockName)); + Check.NotNullOrEmpty(jobArgsTypes, nameof(jobArgsTypes)); + + if (jobArgsTypes.Any(t => t == null)) + { + throw new ArgumentException("Job args types cannot contain null.", nameof(jobArgsTypes)); + } + + JobArgsTypes = jobArgsTypes.ToList(); + } +} diff --git a/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobWorkerManager.cs b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobWorkerManager.cs new file mode 100644 index 0000000000..929c4803da --- /dev/null +++ b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/BackgroundJobWorkerManager.cs @@ -0,0 +1,131 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using Volo.Abp.BackgroundWorkers; + +namespace Volo.Abp.BackgroundJobs; + +/// +/// Owns and controls the background job workers. +/// When no is configured, a single +/// default worker processes all jobs. Otherwise, one dedicated worker is started per configuration +/// (each with its own distributed lock and job-type filter) plus a default worker for the remaining jobs. +/// The workers are resolved from DI, so a replaced is respected. +/// +public class BackgroundJobWorkerManager : IBackgroundWorker +{ + protected AbpBackgroundJobOptions JobOptions { get; } + + protected AbpBackgroundJobWorkerOptions WorkerOptions { get; } + + protected IServiceProvider ServiceProvider { get; } + + protected List Workers { get; } + + public BackgroundJobWorkerManager( + IOptions jobOptions, + IOptions workerOptions, + IServiceProvider serviceProvider) + { + JobOptions = jobOptions.Value; + WorkerOptions = workerOptions.Value; + ServiceProvider = serviceProvider; + Workers = new List(); + } + + public virtual async Task StartAsync(CancellationToken cancellationToken = default) + { + if (!JobOptions.IsJobExecutionEnabled) + { + return; + } + + if (!WorkerOptions.WorkerConfigurations.Any()) + { + await StartWorkerAsync(cancellationToken: cancellationToken); + return; + } + + // AddDedicatedWorker already rejects duplicate job types and lock names eagerly. This is the backstop + // for what can only be known here: two different args types that resolve to the same job name. + // Validate all configurations first, so a misconfiguration does not leave already-started workers running. + var dedicatedWorkers = new List(); + var allDedicatedJobNames = new List(); + + // The default worker uses WorkerOptions.DistributedLockName, so dedicated workers must not reuse it. + var lockNames = new List { WorkerOptions.DistributedLockName }; + + foreach (var configuration in WorkerOptions.WorkerConfigurations) + { + var jobNames = configuration.JobArgsTypes + .Select(GetJobName) + .Distinct() + .ToList(); + + var alreadyConfigured = jobNames.Intersect(allDedicatedJobNames).ToList(); + if (alreadyConfigured.Any()) + { + throw new AbpException( + $"The following background job(s) are configured for more than one dedicated worker: {string.Join(", ", alreadyConfigured)}. " + + $"Each job type can be handled by only one dedicated worker."); + } + + if (lockNames.Contains(configuration.LockName)) + { + throw new AbpException( + $"The distributed lock name '{configuration.LockName}' is used by more than one background job worker " + + $"(the default worker uses '{WorkerOptions.DistributedLockName}'). Each worker must have a unique lock name to run independently."); + } + + lockNames.Add(configuration.LockName); + allDedicatedJobNames.AddRange(jobNames); + dedicatedWorkers.Add(new DedicatedWorkerDefinition(configuration.LockName, jobNames)); + } + + foreach (var dedicatedWorker in dedicatedWorkers) + { + await StartWorkerAsync(dedicatedWorker.LockName, BackgroundJobNameFilter.Include(dedicatedWorker.JobNames), cancellationToken); + } + + // Default worker processes every job that is not handled by a dedicated worker. + await StartWorkerAsync(null, BackgroundJobNameFilter.Exclude(allDedicatedJobNames), cancellationToken); + } + + protected virtual string GetJobName(Type argsType) + { + try + { + return JobOptions.GetJob(argsType).JobName; + } + catch (AbpException ex) + { + throw new AbpException( + $"No background job is registered for the args type '{argsType.FullName}' configured via AddDedicatedWorker. " + + $"Register the job before configuring a dedicated worker for it.", ex); + } + } + + protected virtual async Task StartWorkerAsync( + string? distributedLockName = null, + BackgroundJobNameFilter? jobNameFilter = null, + CancellationToken cancellationToken = default) + { + var worker = ServiceProvider.GetRequiredService(); + await worker.StartAsync(distributedLockName, jobNameFilter, cancellationToken); + Workers.Add(worker); + } + + public virtual async Task StopAsync(CancellationToken cancellationToken = default) + { + foreach (var worker in Workers) + { + await worker.StopAsync(cancellationToken); + } + + Workers.Clear(); + } +} diff --git a/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/DedicatedWorkerDefinition.cs b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/DedicatedWorkerDefinition.cs new file mode 100644 index 0000000000..ba98c2c1dc --- /dev/null +++ b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/DedicatedWorkerDefinition.cs @@ -0,0 +1,27 @@ +using System.Collections.Generic; + +namespace Volo.Abp.BackgroundJobs; + +/// +/// A validated, ready-to-start dedicated worker: the distributed lock name it runs under and the +/// resolved job names it is responsible for. Built by from a +/// after all configurations have been validated. +/// +public class DedicatedWorkerDefinition +{ + /// + /// The distributed lock name this worker runs under. + /// + public string LockName { get; } + + /// + /// The resolved job names this worker is responsible for. + /// + public IReadOnlyList JobNames { get; } + + public DedicatedWorkerDefinition(string lockName, IReadOnlyList jobNames) + { + LockName = lockName; + JobNames = jobNames; + } +} diff --git a/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/IBackgroundJobStore.cs b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/IBackgroundJobStore.cs index 7fcf8cc5b6..421ec55434 100644 --- a/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/IBackgroundJobStore.cs +++ b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/IBackgroundJobStore.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Threading; using System.Threading.Tasks; namespace Volo.Abp.BackgroundJobs; @@ -24,7 +25,7 @@ public interface IBackgroundJobStore /// /// Gets waiting jobs. It should get jobs based on these: - /// Conditions: ApplicationName is applicationName And !IsAbandoned And NextTryTime <= Clock.Now. + /// Conditions: ApplicationName is applicationName And !IsAbandoned And CompletionTime == null And NextTryTime <= Clock.Now. /// Order by: Priority DESC, TryCount ASC, NextTryTime ASC. /// Maximum result: . /// @@ -32,12 +33,35 @@ public interface IBackgroundJobStore /// Maximum result count. Task> GetWaitingJobsAsync(string? applicationName, int maxResultCount); + /// + /// Gets waiting jobs (see ), additionally filtered by job name + /// according to . + /// + /// Application name. + /// Maximum result count. + /// Job name filter. When null, no job name filter is applied. + Task> GetWaitingJobsAsync( + string? applicationName, + int maxResultCount, + BackgroundJobNameFilter? jobNameFilter); + /// /// Deletes a job. /// /// The Job Unique Identifier. Task DeleteAsync(Guid jobId); + /// + /// Deletes successfully completed jobs ( is set) of the given + /// application that completed before . Used by the retention cleanup. + /// + /// The number of deleted jobs. + Task DeleteAsync( + string? applicationName, + DateTime completedBefore, + int maxResultCount, + CancellationToken cancellationToken = default); + /// /// Updates a job. /// diff --git a/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/IBackgroundJobWorker.cs b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/IBackgroundJobWorker.cs index ecae732b8d..9aa4ac25a8 100644 --- a/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/IBackgroundJobWorker.cs +++ b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/IBackgroundJobWorker.cs @@ -1,8 +1,27 @@ -using Volo.Abp.BackgroundWorkers; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; namespace Volo.Abp.BackgroundJobs; -public interface IBackgroundJobWorker : IBackgroundWorker +/// +/// A background job worker that polls and executes waiting jobs. +/// Instances are created, configured and started by . +/// +public interface IBackgroundJobWorker { + /// + /// Starts this worker. + /// + /// + /// Distributed lock name for this worker. When null, is used. + /// + /// Filters the jobs this worker processes by name. When null, all jobs are processed. + /// Cancellation token. + Task StartAsync( + string? distributedLockName = null, + BackgroundJobNameFilter? jobNameFilter = null, + CancellationToken cancellationToken = default); + Task StopAsync(CancellationToken cancellationToken = default); } diff --git a/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/InMemoryBackgroundJobStore.cs b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/InMemoryBackgroundJobStore.cs index 85916e8c37..d0fa34e7e8 100644 --- a/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/InMemoryBackgroundJobStore.cs +++ b/framework/src/Volo.Abp.BackgroundJobs/Volo/Abp/BackgroundJobs/InMemoryBackgroundJobStore.cs @@ -2,6 +2,7 @@ using System; using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; +using System.Threading; using System.Threading.Tasks; using Volo.Abp.DependencyInjection; using Volo.Abp.Timing; @@ -37,9 +38,20 @@ public class InMemoryBackgroundJobStore : IBackgroundJobStore, ISingletonDepende public virtual Task> GetWaitingJobsAsync(string? applicationName, int maxResultCount) { + return GetWaitingJobsAsync(applicationName, maxResultCount, null); + } + + public virtual Task> GetWaitingJobsAsync( + string? applicationName, + int maxResultCount, + BackgroundJobNameFilter? jobNameFilter) + { + var filter = jobNameFilter ?? BackgroundJobNameFilter.None; + var waitingJobs = _jobs.Values .Where(t => t.ApplicationName == applicationName) - .Where(t => !t.IsAbandoned && t.NextTryTime <= Clock.Now) + .Where(t => !t.IsAbandoned && t.CompletionTime == null && t.NextTryTime <= Clock.Now) + .Where(t => filter.IsMatch(t.JobName)) .OrderByDescending(t => t.Priority) .ThenBy(t => t.TryCount) .ThenBy(t => t.NextTryTime) @@ -57,6 +69,28 @@ public class InMemoryBackgroundJobStore : IBackgroundJobStore, ISingletonDepende return Task.CompletedTask; } + public virtual Task DeleteAsync( + string? applicationName, + DateTime completedBefore, + int maxResultCount, + CancellationToken cancellationToken = default) + { + var idsToDelete = _jobs.Values + .Where(t => t.ApplicationName == applicationName) + .Where(t => t.CompletionTime != null && t.CompletionTime < completedBefore) + .OrderBy(t => t.CompletionTime) + .Take(maxResultCount) + .Select(t => t.Id) + .ToList(); + + foreach (var id in idsToDelete) + { + _jobs.TryRemove(id, out _); + } + + return Task.FromResult(idsToDelete.Count); + } + public virtual Task UpdateAsync(BackgroundJobInfo jobInfo) { if (jobInfo.IsAbandoned) diff --git a/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpAutoLockNameWorkerTestModule.cs b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpAutoLockNameWorkerTestModule.cs new file mode 100644 index 0000000000..38b640b84b --- /dev/null +++ b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpAutoLockNameWorkerTestModule.cs @@ -0,0 +1,27 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Volo.Abp.Autofac; +using Volo.Abp.Modularity; + +namespace Volo.Abp.BackgroundJobs; + +[DependsOn( + typeof(AbpBackgroundJobsModule), + typeof(AbpAutofacModule), + typeof(AbpTestBaseModule) +)] +public class AbpAutoLockNameWorkerTestModule : AbpModule +{ + public override void ConfigureServices(ServiceConfigurationContext context) + { + context.Services.AddSingleton(); + context.Services.Replace(ServiceDescriptor.Transient()); + + // No lock name is given: it is derived from the job argument type names. + Configure(options => + { + options.AddDedicatedWorker(); + options.AddDedicatedWorker(); + }); + } +} diff --git a/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpBackgroundJobCleanupTestModule.cs b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpBackgroundJobCleanupTestModule.cs new file mode 100644 index 0000000000..b29c60784e --- /dev/null +++ b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpBackgroundJobCleanupTestModule.cs @@ -0,0 +1,23 @@ +using System; +using Volo.Abp.Autofac; +using Volo.Abp.Modularity; + +namespace Volo.Abp.BackgroundJobs; + +[DependsOn( + typeof(AbpBackgroundJobsModule), + typeof(AbpAutofacModule), + typeof(AbpTestBaseModule) +)] +public class AbpBackgroundJobCleanupTestModule : AbpModule +{ + public override void ConfigureServices(ServiceConfigurationContext context) + { + // IsJobExecutionEnabled is true by default, which the cleanup worker requires. + Configure(options => + { + options.StoreSuccessfulJobs = true; + options.SuccessfulJobRetentionTime = TimeSpan.FromDays(1); + }); + } +} diff --git a/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpBackgroundJobWorkerTestModule.cs b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpBackgroundJobWorkerTestModule.cs new file mode 100644 index 0000000000..db56843a4b --- /dev/null +++ b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpBackgroundJobWorkerTestModule.cs @@ -0,0 +1,25 @@ +using Microsoft.Extensions.DependencyInjection; +using Volo.Abp.Autofac; +using Volo.Abp.Modularity; + +namespace Volo.Abp.BackgroundJobs; + +[DependsOn( + typeof(AbpBackgroundJobsModule), + typeof(AbpAutofacModule), + typeof(AbpTestBaseModule) +)] +public class AbpBackgroundJobWorkerTestModule : AbpModule +{ + public override void ConfigureServices(ServiceConfigurationContext context) + { + // Drive the worker manually in tests; don't run the real periodic worker. + Configure(options => + { + options.IsJobExecutionEnabled = false; + }); + + context.Services.AddSingleton(); + context.Services.AddScoped(); + } +} diff --git a/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpDuplicateLockNameTestModule.cs b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpDuplicateLockNameTestModule.cs new file mode 100644 index 0000000000..6e35c81dfc --- /dev/null +++ b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpDuplicateLockNameTestModule.cs @@ -0,0 +1,27 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Volo.Abp.Autofac; +using Volo.Abp.Modularity; + +namespace Volo.Abp.BackgroundJobs; + +[DependsOn( + typeof(AbpBackgroundJobsModule), + typeof(AbpAutofacModule), + typeof(AbpTestBaseModule) +)] +public class AbpDuplicateLockNameTestModule : AbpModule +{ + public override void ConfigureServices(ServiceConfigurationContext context) + { + context.Services.AddSingleton(); + context.Services.Replace(ServiceDescriptor.Transient()); + + // Two workers with different job types but the same lock name, which must fail at initialization. + Configure(options => + { + options.AddDedicatedWorker("dup-lock"); + options.AddDedicatedWorker("dup-lock"); + }); + } +} diff --git a/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpDuplicateWorkerTestModule.cs b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpDuplicateWorkerTestModule.cs new file mode 100644 index 0000000000..f4d9aae208 --- /dev/null +++ b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpDuplicateWorkerTestModule.cs @@ -0,0 +1,27 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Volo.Abp.Autofac; +using Volo.Abp.Modularity; + +namespace Volo.Abp.BackgroundJobs; + +[DependsOn( + typeof(AbpBackgroundJobsModule), + typeof(AbpAutofacModule), + typeof(AbpTestBaseModule) +)] +public class AbpDuplicateWorkerTestModule : AbpModule +{ + public override void ConfigureServices(ServiceConfigurationContext context) + { + context.Services.AddSingleton(); + context.Services.Replace(ServiceDescriptor.Transient()); + + // The same job type is assigned to two dedicated workers, which must fail at initialization. + Configure(options => + { + options.AddDedicatedWorker("lock-a"); + options.AddDedicatedWorker("lock-b"); + }); + } +} diff --git a/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpMultiWorkerTestModule.cs b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpMultiWorkerTestModule.cs new file mode 100644 index 0000000000..d437c9325f --- /dev/null +++ b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpMultiWorkerTestModule.cs @@ -0,0 +1,28 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Volo.Abp.Autofac; +using Volo.Abp.Modularity; + +namespace Volo.Abp.BackgroundJobs; + +[DependsOn( + typeof(AbpBackgroundJobsModule), + typeof(AbpAutofacModule), + typeof(AbpTestBaseModule) +)] +public class AbpMultiWorkerTestModule : AbpModule +{ + public override void ConfigureServices(ServiceConfigurationContext context) + { + context.Services.AddSingleton(); + + // Replace the real worker with a recording one to assert how the manager resolves and starts workers. + context.Services.Replace(ServiceDescriptor.Transient()); + + Configure(options => + { + options.AddDedicatedWorker("lock-a"); + options.AddDedicatedWorker("lock-b"); + }); + } +} diff --git a/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpSameJobNameTestModule.cs b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpSameJobNameTestModule.cs new file mode 100644 index 0000000000..f4ddbb8be8 --- /dev/null +++ b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/AbpSameJobNameTestModule.cs @@ -0,0 +1,29 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Volo.Abp.Autofac; +using Volo.Abp.Modularity; + +namespace Volo.Abp.BackgroundJobs; + +[DependsOn( + typeof(AbpBackgroundJobsModule), + typeof(AbpAutofacModule), + typeof(AbpTestBaseModule) +)] +public class AbpSameJobNameTestModule : AbpModule +{ + public override void ConfigureServices(ServiceConfigurationContext context) + { + context.Services.AddSingleton(); + context.Services.Replace(ServiceDescriptor.Transient()); + + // Two dedicated workers with different args types that both resolve to "shared-job-name". + // AddDedicatedWorker's eager check compares by type (both pass); only the manager's backstop + // (which resolves job names) can catch this. + Configure(options => + { + options.AddDedicatedWorker("lock-a"); + options.AddDedicatedWorker("lock-b"); + }); + } +} diff --git a/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/BackgroundJobCleanupWorker_Tests.cs b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/BackgroundJobCleanupWorker_Tests.cs new file mode 100644 index 0000000000..a7a638ed7b --- /dev/null +++ b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/BackgroundJobCleanupWorker_Tests.cs @@ -0,0 +1,148 @@ +using System; +using System.Collections.Generic; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using Shouldly; +using Volo.Abp.BackgroundWorkers; +using Volo.Abp.DistributedLocking; +using Volo.Abp.Testing; +using Volo.Abp.Threading; +using Volo.Abp.Timing; +using Xunit; + +namespace Volo.Abp.BackgroundJobs; + +public class BackgroundJobCleanupWorker_Tests : AbpIntegratedTest +{ + private readonly IBackgroundJobStore _store; + private readonly IClock _clock; + private readonly AbpBackgroundJobWorkerOptions _workerOptions; + + public BackgroundJobCleanupWorker_Tests() + { + _store = GetRequiredService(); + _clock = GetRequiredService(); + _workerOptions = GetRequiredService>().Value; + } + + protected override void SetAbpApplicationCreationOptions(AbpApplicationCreationOptions options) + { + options.UseAutofac(); + } + + private TestableBackgroundJobCleanupWorker CreateWorker() + { + return new TestableBackgroundJobCleanupWorker( + GetRequiredService(), + GetRequiredService(), + GetRequiredService>(), + GetRequiredService>(), + GetRequiredService()); + } + + private Task RunCleanupAsync() + { + return CreateWorker().DoWorkPublicAsync(new PeriodicBackgroundWorkerContext(ServiceProvider)); + } + + private async Task InsertCompletedJobAsync(DateTime completionTime) + { + var id = Guid.NewGuid(); + await _store.InsertAsync(new BackgroundJobInfo + { + Id = id, + JobName = "job-a", + JobArgs = "{}", + CreationTime = _clock.Now, + NextTryTime = _clock.Now, + CompletionTime = completionTime + }); + return id; + } + + [Fact] + public async Task Should_Delete_Completed_Jobs_Older_Than_Retention() + { + // Retention is 1 day (configured in the module). + var oldJobId = await InsertCompletedJobAsync(_clock.Now.Subtract(TimeSpan.FromDays(2))); + var recentJobId = await InsertCompletedJobAsync(_clock.Now); + + await RunCleanupAsync(); + + (await _store.FindAsync(oldJobId)).ShouldBeNull(); // older than retention → deleted + (await _store.FindAsync(recentJobId)).ShouldNotBeNull(); // within retention → kept + } + + [Fact] + public async Task Should_Not_Delete_When_StoreSuccessfulJobs_Disabled() + { + _workerOptions.StoreSuccessfulJobs = false; + + var oldJobId = await InsertCompletedJobAsync(_clock.Now.Subtract(TimeSpan.FromDays(2))); + + await RunCleanupAsync(); + + (await _store.FindAsync(oldJobId)).ShouldNotBeNull(); + } + + [Fact] + public async Task Should_Not_Delete_When_Retention_Is_Null() + { + _workerOptions.SuccessfulJobRetentionTime = null; + + var oldJobId = await InsertCompletedJobAsync(_clock.Now.Subtract(TimeSpan.FromDays(2))); + + await RunCleanupAsync(); + + (await _store.FindAsync(oldJobId)).ShouldNotBeNull(); + } + + [Fact] + public async Task Should_Delete_All_Old_Jobs_In_Batches() + { + _workerOptions.MaxJobFetchCount = 2; + + var ids = new List(); + for (var i = 0; i < 5; i++) + { + ids.Add(await InsertCompletedJobAsync(_clock.Now.Subtract(TimeSpan.FromDays(2)))); + } + + await RunCleanupAsync(); + + foreach (var id in ids) + { + (await _store.FindAsync(id)).ShouldBeNull(); + } + } + + [Fact] + public async Task Should_Delete_Oldest_Completed_Jobs_First_When_Limited_By_MaxResultCount() + { + var oldest = await InsertCompletedJobAsync(_clock.Now.Subtract(TimeSpan.FromDays(5))); + var middle = await InsertCompletedJobAsync(_clock.Now.Subtract(TimeSpan.FromDays(3))); + var newest = await InsertCompletedJobAsync(_clock.Now.Subtract(TimeSpan.FromDays(1))); + + // All three are completed before now, but a single call may delete only two. + var deletedCount = await _store.DeleteAsync(null, _clock.Now, maxResultCount: 2); + + deletedCount.ShouldBe(2); + (await _store.FindAsync(oldest)).ShouldBeNull(); + (await _store.FindAsync(middle)).ShouldBeNull(); + (await _store.FindAsync(newest)).ShouldNotBeNull(); // newest survives, proving oldest-first deletion + } + + [Fact] + public async Task Should_Not_Loop_Forever_When_MaxJobFetchCount_Is_Zero() + { + _workerOptions.MaxJobFetchCount = 0; + + var oldJobId = await InsertCompletedJobAsync(_clock.Now.Subtract(TimeSpan.FromDays(2))); + + // Must return (not hang) even though nothing can be fetched/deleted with a zero page size. + await RunCleanupAsync(); + + (await _store.FindAsync(oldJobId)).ShouldNotBeNull(); + } +} diff --git a/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/BackgroundJobWorkerTestBase.cs b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/BackgroundJobWorkerTestBase.cs new file mode 100644 index 0000000000..2824c1a231 --- /dev/null +++ b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/BackgroundJobWorkerTestBase.cs @@ -0,0 +1,11 @@ +using Volo.Abp.Testing; + +namespace Volo.Abp.BackgroundJobs; + +public abstract class BackgroundJobWorkerTestBase : AbpIntegratedTest +{ + protected override void SetAbpApplicationCreationOptions(AbpApplicationCreationOptions options) + { + options.UseAutofac(); + } +} diff --git a/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/BackgroundJobWorker_AutoLockName_Tests.cs b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/BackgroundJobWorker_AutoLockName_Tests.cs new file mode 100644 index 0000000000..76e4734b05 --- /dev/null +++ b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/BackgroundJobWorker_AutoLockName_Tests.cs @@ -0,0 +1,34 @@ +using System; +using System.Linq; +using Shouldly; +using Volo.Abp.Testing; +using Xunit; + +namespace Volo.Abp.BackgroundJobs; + +public class BackgroundJobWorker_AutoLockName_Tests : AbpIntegratedTest +{ + protected override void SetAbpApplicationCreationOptions(AbpApplicationCreationOptions options) + { + options.UseAutofac(); + } + + [Fact] + public void Should_Derive_Bounded_Lock_Name_From_Job_Args_Types() + { + var records = GetRequiredService().Records; + + var dedicated = records.Where(r => r.JobNameFilter?.Mode == BackgroundJobNameFilterMode.Include).ToList(); + dedicated.Count.ShouldBe(2); + + // Lock name is derived (prefix + MD5 of the full type name), so it is stable and length-bounded. + var expectedA = "AbpBackgroundJobDedicatedWorker:" + typeof(WorkerJobAArgs).FullName!.ToMd5(); + var expectedB = "AbpBackgroundJobDedicatedWorker:" + typeof(WorkerJobBArgs).FullName!.ToMd5(); + + dedicated.ShouldContain(r => r.DistributedLockName == expectedA); + dedicated.ShouldContain(r => r.DistributedLockName == expectedB); + + // Bounded length regardless of how long the type names are. + dedicated.ShouldAllBe(r => r.DistributedLockName!.Length <= 64); + } +} diff --git a/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/BackgroundJobWorker_DuplicateConfiguration_Tests.cs b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/BackgroundJobWorker_DuplicateConfiguration_Tests.cs new file mode 100644 index 0000000000..bdd80efa7a --- /dev/null +++ b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/BackgroundJobWorker_DuplicateConfiguration_Tests.cs @@ -0,0 +1,63 @@ +using Microsoft.Extensions.DependencyInjection; +using Shouldly; +using Volo.Abp.Autofac; +using Xunit; + +namespace Volo.Abp.BackgroundJobs; + +public class BackgroundJobWorker_DuplicateConfiguration_Tests +{ + [Fact] + public void Should_Throw_Without_Starting_Any_Worker_When_A_Job_Type_Is_Assigned_To_Multiple_Workers() + { + using var application = AbpApplicationFactory.Create(options => + { + options.UseAutofac(); + }); + + var exception = Record.Exception(() => application.Initialize()); + + exception.ShouldNotBeNull(); + exception.ToString().ShouldContain("dedicated worker"); + + // Validation must happen before any worker is started. + var recorder = application.ServiceProvider.GetRequiredService(); + recorder.Records.ShouldBeEmpty(); + } + + [Fact] + public void Should_Throw_Without_Starting_Any_Worker_When_Two_Workers_Share_A_Lock_Name() + { + using var application = AbpApplicationFactory.Create(options => + { + options.UseAutofac(); + }); + + var exception = Record.Exception(() => application.Initialize()); + + exception.ShouldNotBeNull(); + exception.ToString().ShouldContain("lock name"); + + var recorder = application.ServiceProvider.GetRequiredService(); + recorder.Records.ShouldBeEmpty(); + } + + [Fact] + public void Should_Throw_Without_Starting_Any_Worker_When_Different_Args_Types_Resolve_To_The_Same_Job_Name() + { + using var application = AbpApplicationFactory.Create(options => + { + options.UseAutofac(); + }); + + // The two args types pass the eager (by-type) check but resolve to the same job name, + // so only the manager's backstop validation can reject them. + var exception = Record.Exception(() => application.Initialize()); + + exception.ShouldNotBeNull(); + exception.ToString().ShouldContain("more than one dedicated worker"); + + var recorder = application.ServiceProvider.GetRequiredService(); + recorder.Records.ShouldBeEmpty(); + } +} diff --git a/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/BackgroundJobWorker_MultiWorkerRegistration_Tests.cs b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/BackgroundJobWorker_MultiWorkerRegistration_Tests.cs new file mode 100644 index 0000000000..8279d2d493 --- /dev/null +++ b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/BackgroundJobWorker_MultiWorkerRegistration_Tests.cs @@ -0,0 +1,37 @@ +using System.Linq; +using Shouldly; +using Volo.Abp.Testing; +using Xunit; + +namespace Volo.Abp.BackgroundJobs; + +public class BackgroundJobWorker_MultiWorkerRegistration_Tests : AbpIntegratedTest +{ + protected override void SetAbpApplicationCreationOptions(AbpApplicationCreationOptions options) + { + options.UseAutofac(); + } + + [Fact] + public void Should_Start_Dedicated_Workers_And_A_Default_Worker() + { + // The workers are resolved from DI (RecordingBackgroundJobWorker replaces the real one), + // proving the manager honors the registered/replaced IBackgroundJobWorker. + var records = GetRequiredService().Records; + + records.Count.ShouldBe(3); + + var jobAName = BackgroundJobNameAttribute.GetName(); + var jobBName = BackgroundJobNameAttribute.GetName(); + + var dedicated = records.Where(r => r.JobNameFilter?.Mode == BackgroundJobNameFilterMode.Include).ToList(); + dedicated.Count.ShouldBe(2); + dedicated.ShouldContain(r => r.DistributedLockName == "lock-a" && r.JobNameFilter!.JobNames.Contains(jobAName)); + dedicated.ShouldContain(r => r.DistributedLockName == "lock-b" && r.JobNameFilter!.JobNames.Contains(jobBName)); + + var defaultWorker = records.Single(r => r.JobNameFilter?.Mode == BackgroundJobNameFilterMode.Exclude); + defaultWorker.DistributedLockName.ShouldBeNull(); + defaultWorker.JobNameFilter!.JobNames.ShouldContain(jobAName); + defaultWorker.JobNameFilter!.JobNames.ShouldContain(jobBName); + } +} diff --git a/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/BackgroundJobWorker_Tests.cs b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/BackgroundJobWorker_Tests.cs new file mode 100644 index 0000000000..4d8d506f19 --- /dev/null +++ b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/BackgroundJobWorker_Tests.cs @@ -0,0 +1,300 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using Shouldly; +using Volo.Abp.BackgroundWorkers; +using Volo.Abp.DependencyInjection; +using Volo.Abp.DistributedLocking; +using Volo.Abp.Threading; +using Volo.Abp.Timing; +using Xunit; + +// ReSharper disable PossibleMultipleEnumeration + +namespace Volo.Abp.BackgroundJobs; + +public class BackgroundJobWorker_Tests : BackgroundJobWorkerTestBase +{ + private readonly IBackgroundJobStore _store; + private readonly IClock _clock; + private readonly AbpBackgroundJobWorkerOptions _workerOptions; + + public BackgroundJobWorker_Tests() + { + _store = GetRequiredService(); + _clock = GetRequiredService(); + _workerOptions = GetRequiredService>().Value; + } + + private TestableBackgroundJobWorker CreateWorker() + { + return new TestableBackgroundJobWorker( + GetRequiredService(), + GetRequiredService>(), + GetRequiredService>(), + GetRequiredService(), + GetRequiredService()); + } + + private BackgroundJobInfo NewJob(string jobName) + { + return new BackgroundJobInfo + { + Id = Guid.NewGuid(), + JobName = jobName, + JobArgs = "{}", + CreationTime = _clock.Now, + NextTryTime = _clock.Now.AddMinutes(-1) + }; + } + + private PeriodicBackgroundWorkerContext Context() + { + return new PeriodicBackgroundWorkerContext(ServiceProvider); + } + + // Storing successful jobs + + [Fact] + public async Task Should_Keep_Successful_Job_As_History_When_Enabled() + { + _workerOptions.StoreSuccessfulJobs = true; + + var jobInfo = NewJob("job-a"); + await _store.InsertAsync(jobInfo); + + await CreateWorker().HandleJobSuccessPublicAsync(_store, jobInfo, _clock); + + // The job is kept (marked completed), not deleted. + var kept = await _store.FindAsync(jobInfo.Id); + kept.ShouldNotBeNull(); + kept.CompletionTime.ShouldNotBeNull(); + + // ...but it is excluded from the waiting jobs. + (await _store.GetWaitingJobsAsync(null, 1000)).ShouldNotContain(j => j.Id == jobInfo.Id); + } + + [Fact] + public async Task Should_Delete_Successful_Job_When_Disabled() + { + // StoreSuccessfulJobs is false by default. + var jobInfo = NewJob("job-a"); + await _store.InsertAsync(jobInfo); + + await CreateWorker().HandleJobSuccessPublicAsync(_store, jobInfo, _clock); + + (await _store.FindAsync(jobInfo.Id)).ShouldBeNull(); + } + + // Dedicated workers by job name + + [Fact] + public async Task Should_Return_Only_Included_Jobs() + { + await _store.InsertAsync(NewJob("job-a")); + await _store.InsertAsync(NewJob("job-a")); + await _store.InsertAsync(NewJob("job-b")); + + var worker = CreateWorker(); + worker.ConfigureTest(BackgroundJobNameFilter.Include(new[] { "job-a" })); + + var jobs = await worker.GetWaitingJobsPublicAsync(Context(), _store); + + jobs.Count.ShouldBe(2); + jobs.ShouldAllBe(j => j.JobName == "job-a"); + } + + [Fact] + public async Task Should_Exclude_Given_Jobs() + { + await _store.InsertAsync(NewJob("job-a")); + await _store.InsertAsync(NewJob("job-b")); + + var worker = CreateWorker(); + worker.ConfigureTest(BackgroundJobNameFilter.Exclude(new[] { "job-a" })); + + var jobs = await worker.GetWaitingJobsPublicAsync(Context(), _store); + + jobs.Count.ShouldBe(1); + jobs.Single().JobName.ShouldBe("job-b"); + } + + [Fact] + public void Should_Match_Job_Names_By_Filter() + { + BackgroundJobNameFilter.None.IsMatch("any").ShouldBeTrue(); + + var include = BackgroundJobNameFilter.Include(new[] { "job-a" }); + include.IsMatch("job-a").ShouldBeTrue(); + include.IsMatch("job-b").ShouldBeFalse(); + + var exclude = BackgroundJobNameFilter.Exclude(new[] { "job-a" }); + exclude.IsMatch("job-a").ShouldBeFalse(); + exclude.IsMatch("job-b").ShouldBeTrue(); + } + + [Fact] + public void BackgroundJobNameFilter_Should_Reject_Invalid_Mode_And_Names_Combinations() + { + Should.Throw(() => new BackgroundJobNameFilter(BackgroundJobNameFilterMode.Include)); + Should.Throw(() => new BackgroundJobNameFilter(BackgroundJobNameFilterMode.None, new[] { "job-a" })); + Should.Throw(() => new BackgroundJobNameFilter((BackgroundJobNameFilterMode)99, new[] { "job-a" })); + } + + // Parallel execution / eligibility + + [Fact] + public void Should_Evaluate_Job_Eligibility() + { + var worker = CreateWorker(); + + worker.IsJobEligiblePublic(null, _clock).ShouldBeFalse(); + + var future = NewJob("job-a"); + future.NextTryTime = _clock.Now.AddMinutes(5); + worker.IsJobEligiblePublic(future, _clock).ShouldBeFalse(); + + var abandoned = NewJob("job-a"); + abandoned.IsAbandoned = true; + worker.IsJobEligiblePublic(abandoned, _clock).ShouldBeFalse(); + + var completed = NewJob("job-a"); + completed.CompletionTime = _clock.Now; + worker.IsJobEligiblePublic(completed, _clock).ShouldBeFalse(); + + var eligible = NewJob("job-a"); + worker.IsJobEligiblePublic(eligible, _clock).ShouldBeTrue(); + } + + [Fact] + public void Should_Not_Be_Eligible_When_Job_Name_Filtered_Out() + { + var worker = CreateWorker(); + worker.ConfigureTest(BackgroundJobNameFilter.Include(new[] { "job-a" })); + + var otherJob = NewJob("job-b"); + worker.IsJobEligiblePublic(otherJob, _clock).ShouldBeFalse(); + } + + [Fact] + public void Should_Require_Job_Args_Types_For_A_Dedicated_Worker() + { + Should.Throw(() => new BackgroundJobWorkerConfiguration("lock-a")); + } + + [Fact] + public void AddDedicatedWorker_Should_Throw_At_Registration_When_A_Job_Type_Is_Added_Twice() + { + var options = new AbpBackgroundJobWorkerOptions(); + options.AddDedicatedWorker("lock-a"); + + Should.Throw(() => options.AddDedicatedWorker("lock-b")); + } + + [Fact] + public void AddDedicatedWorker_Should_Throw_At_Registration_When_A_Lock_Name_Is_Reused() + { + var options = new AbpBackgroundJobWorkerOptions(); + options.AddDedicatedWorker("dup-lock"); + + Should.Throw(() => options.AddDedicatedWorker("dup-lock")); + } + + [Fact] + public void AddDedicatedWorker_Should_Throw_At_Registration_When_Lock_Name_Equals_The_Default() + { + var options = new AbpBackgroundJobWorkerOptions(); + + Should.Throw(() => options.AddDedicatedWorker(options.DistributedLockName)); + } + + // Parallel execution + + [Fact] + public async Task Should_Execute_Multiple_Jobs_In_Parallel() + { + _workerOptions.MaxParallelJobExecutionCount = 3; + + var jobManager = GetRequiredService(); + await jobManager.EnqueueAsync(new ParallelTestJobArgs { Value = "1" }); + await jobManager.EnqueueAsync(new ParallelTestJobArgs { Value = "2" }); + await jobManager.EnqueueAsync(new ParallelTestJobArgs { Value = "3" }); + + await CreateWorker().ExecuteJobsInParallelPublicAsync(Context()); + + var tracker = GetRequiredService(); + tracker.Executed.Count.ShouldBe(3); + tracker.Executed.ShouldContain("1"); + tracker.Executed.ShouldContain("2"); + tracker.Executed.ShouldContain("3"); + + // Each job must run in its own service scope (isolated DbContext/UOW). + tracker.ScopeIds.Distinct().Count().ShouldBe(3); + + (await _store.GetWaitingJobsAsync(null, 1000)).ShouldBeEmpty(); + } + + [Fact] + public async Task Should_Execute_At_Most_MaxParallel_Jobs_Per_Cycle() + { + _workerOptions.MaxParallelJobExecutionCount = 2; + + var jobManager = GetRequiredService(); + for (var i = 0; i < 5; i++) + { + await jobManager.EnqueueAsync(new ParallelTestJobArgs { Value = i.ToString() }); + } + + await CreateWorker().ExecuteJobsInParallelPublicAsync(Context()); + + var tracker = GetRequiredService(); + tracker.Executed.Count.ShouldBe(2); + (await _store.GetWaitingJobsAsync(null, 1000)).Count.ShouldBe(3); + } + + [Fact] + public async Task Should_Skip_Job_Already_Claimed_By_Another_Instance() + { + _workerOptions.MaxParallelJobExecutionCount = 5; + + var jobManager = GetRequiredService(); + var lockedJobId = Guid.Parse(await jobManager.EnqueueAsync(new ParallelTestJobArgs { Value = "locked" })); + await jobManager.EnqueueAsync(new ParallelTestJobArgs { Value = "free" }); + + var distributedLock = GetRequiredService(); + var lockName = _workerOptions.PerJobDistributedLockPrefix + lockedJobId; + + await using (await distributedLock.TryAcquireAsync(lockName)) + { + await CreateWorker().ExecuteJobsInParallelPublicAsync(Context()); + } + + var tracker = GetRequiredService(); + tracker.Executed.ShouldContain("free"); + tracker.Executed.ShouldNotContain("locked"); + (await _store.FindAsync(lockedJobId)).ShouldNotBeNull(); + } + + [Fact] + public async Task Should_Mark_Job_Completed_On_Successful_Execution_When_Storing_Enabled() + { + _workerOptions.StoreSuccessfulJobs = true; + _workerOptions.MaxParallelJobExecutionCount = 2; + + var jobManager = GetRequiredService(); + var jobId = Guid.Parse(await jobManager.EnqueueAsync(new ParallelTestJobArgs { Value = "1" })); + + await CreateWorker().ExecuteJobsInParallelPublicAsync(Context()); + + GetRequiredService().Executed.ShouldContain("1"); + + // The job is kept (marked completed), not deleted, and excluded from the waiting list. + var job = await _store.FindAsync(jobId); + job.ShouldNotBeNull(); + job.CompletionTime.ShouldNotBeNull(); + (await _store.GetWaitingJobsAsync(null, 1000)).ShouldNotContain(j => j.Id == jobId); + } +} diff --git a/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/RecordingBackgroundJobWorker.cs b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/RecordingBackgroundJobWorker.cs new file mode 100644 index 0000000000..6acb5ca78d --- /dev/null +++ b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/RecordingBackgroundJobWorker.cs @@ -0,0 +1,51 @@ +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using Volo.Abp.DependencyInjection; + +namespace Volo.Abp.BackgroundJobs; + +/// +/// Records every call so multi-worker registration +/// (resolved from DI by ) can be asserted. Does not start any timer. +/// +public class WorkerStartRecorder +{ + public List Records { get; } = new List(); +} + +public class WorkerStartRecord +{ + public string? DistributedLockName { get; set; } + + public BackgroundJobNameFilter? JobNameFilter { get; set; } +} + +public class RecordingBackgroundJobWorker : IBackgroundJobWorker, ITransientDependency +{ + private readonly WorkerStartRecorder _recorder; + + public RecordingBackgroundJobWorker(WorkerStartRecorder recorder) + { + _recorder = recorder; + } + + public Task StartAsync( + string? distributedLockName = null, + BackgroundJobNameFilter? jobNameFilter = null, + CancellationToken cancellationToken = default) + { + _recorder.Records.Add(new WorkerStartRecord + { + DistributedLockName = distributedLockName, + JobNameFilter = jobNameFilter + }); + + return Task.CompletedTask; + } + + public Task StopAsync(CancellationToken cancellationToken = default) + { + return Task.CompletedTask; + } +} diff --git a/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/TestableBackgroundJobCleanupWorker.cs b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/TestableBackgroundJobCleanupWorker.cs new file mode 100644 index 0000000000..7c6cc16dec --- /dev/null +++ b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/TestableBackgroundJobCleanupWorker.cs @@ -0,0 +1,26 @@ +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using Volo.Abp.BackgroundWorkers; +using Volo.Abp.DistributedLocking; +using Volo.Abp.Threading; + +namespace Volo.Abp.BackgroundJobs; + +public class TestableBackgroundJobCleanupWorker : BackgroundJobCleanupWorker +{ + public TestableBackgroundJobCleanupWorker( + AbpAsyncTimer timer, + IServiceScopeFactory serviceScopeFactory, + IOptions jobOptions, + IOptions workerOptions, + IAbpDistributedLock distributedLock) + : base(timer, serviceScopeFactory, jobOptions, workerOptions, distributedLock) + { + } + + public Task DoWorkPublicAsync(PeriodicBackgroundWorkerContext workerContext) + { + return DoWorkAsync(workerContext); + } +} diff --git a/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/TestableBackgroundJobWorker.cs b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/TestableBackgroundJobWorker.cs new file mode 100644 index 0000000000..27274320fb --- /dev/null +++ b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/TestableBackgroundJobWorker.cs @@ -0,0 +1,55 @@ +#nullable enable +using System.Collections.Generic; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using Volo.Abp.BackgroundWorkers; +using Volo.Abp.DistributedLocking; +using Volo.Abp.Threading; +using Volo.Abp.Timing; + +namespace Volo.Abp.BackgroundJobs; + +/// +/// Exposes the protected members of for unit testing. +/// +public class TestableBackgroundJobWorker : BackgroundJobWorker +{ + public TestableBackgroundJobWorker( + AbpAsyncTimer timer, + IOptions jobOptions, + IOptions workerOptions, + IServiceScopeFactory serviceScopeFactory, + IAbpDistributedLock distributedLock) + : base(timer, jobOptions, workerOptions, serviceScopeFactory, distributedLock) + { + } + + public void ConfigureTest( + BackgroundJobNameFilter? jobNameFilter = null, + string? distributedLockName = null) + { + JobNameFilter = jobNameFilter ?? BackgroundJobNameFilter.None; + DistributedLockName = distributedLockName ?? WorkerOptions.DistributedLockName; + } + + public Task HandleJobSuccessPublicAsync(IBackgroundJobStore store, BackgroundJobInfo jobInfo, IClock clock) + { + return HandleJobSuccessAsync(store, jobInfo, clock); + } + + public Task ExecuteJobsInParallelPublicAsync(PeriodicBackgroundWorkerContext workerContext) + { + return ExecuteJobsInParallelAsync(workerContext); + } + + public Task> GetWaitingJobsPublicAsync(PeriodicBackgroundWorkerContext workerContext, IBackgroundJobStore store) + { + return GetWaitingJobsAsync(workerContext, store); + } + + public bool IsJobEligiblePublic(BackgroundJobInfo? jobInfo, IClock clock) + { + return IsJobEligible(jobInfo, clock); + } +} diff --git a/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/WorkerTestJobs.cs b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/WorkerTestJobs.cs new file mode 100644 index 0000000000..dd9feec657 --- /dev/null +++ b/framework/test/Volo.Abp.BackgroundJobs.Tests/Volo/Abp/BackgroundJobs/WorkerTestJobs.cs @@ -0,0 +1,99 @@ +using System; +using System.Collections.Concurrent; +using System.Threading.Tasks; +using Volo.Abp.DependencyInjection; + +namespace Volo.Abp.BackgroundJobs; + +public class ParallelJobTracker +{ + public ConcurrentBag Executed { get; } = new ConcurrentBag(); + + public ConcurrentBag ScopeIds { get; } = new ConcurrentBag(); +} + +/// +/// Scoped service used to verify that each parallel job runs in its own service scope. +/// +public class ScopeMarker +{ + public Guid Id { get; } = Guid.NewGuid(); +} + +public class ParallelTestJobArgs +{ + public string Value { get; set; } = default!; +} + +public class ParallelTestJob : AsyncBackgroundJob, ITransientDependency +{ + private readonly ParallelJobTracker _tracker; + private readonly ScopeMarker _scopeMarker; + + public ParallelTestJob(ParallelJobTracker tracker, ScopeMarker scopeMarker) + { + _tracker = tracker; + _scopeMarker = scopeMarker; + } + + public override Task ExecuteAsync(ParallelTestJobArgs args) + { + _tracker.Executed.Add(args.Value); + _tracker.ScopeIds.Add(_scopeMarker.Id); + return Task.CompletedTask; + } +} + +public class WorkerJobAArgs +{ + public string Value { get; set; } = default!; +} + +public class WorkerJobA : AsyncBackgroundJob, ITransientDependency +{ + public override Task ExecuteAsync(WorkerJobAArgs args) + { + return Task.CompletedTask; + } +} + +public class WorkerJobBArgs +{ + public string Value { get; set; } = default!; +} + +public class WorkerJobB : AsyncBackgroundJob, ITransientDependency +{ + public override Task ExecuteAsync(WorkerJobBArgs args) + { + return Task.CompletedTask; + } +} + +// Two different args types that resolve to the same job name, to exercise the manager's +// backstop validation (eager AddDedicatedWorker validation compares by type, not resolved name). +[BackgroundJobName("shared-job-name")] +public class SharedNameJobAArgs +{ +} + +[BackgroundJobName("shared-job-name")] +public class SharedNameJobBArgs +{ +} + +public class SharedNameJobA : AsyncBackgroundJob, ITransientDependency +{ + public override Task ExecuteAsync(SharedNameJobAArgs args) + { + return Task.CompletedTask; + } +} + +public class SharedNameJobB : AsyncBackgroundJob, ITransientDependency +{ + public override Task ExecuteAsync(SharedNameJobBArgs args) + { + return Task.CompletedTask; + } +} diff --git a/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/DemoAppModule.cs b/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/DemoAppModule.cs index 976705b07d..5569f6800a 100644 --- a/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/DemoAppModule.cs +++ b/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/DemoAppModule.cs @@ -1,5 +1,7 @@ -using System.Threading.Tasks; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; using Volo.Abp.Autofac; +using Volo.Abp.BackgroundJobs.DemoApp.Jobs; using Volo.Abp.BackgroundJobs.DemoApp.Shared; using Volo.Abp.BackgroundJobs.EntityFrameworkCore; using Volo.Abp.EntityFrameworkCore; @@ -32,17 +34,31 @@ public class DemoAppModule : AbpModule options.JobPollPeriod = 1000; options.DefaultFirstWaitDuration = 1; options.DefaultWaitFactor = 1; + + // Keep every successfully completed job as history (marks CompletionTime instead of deleting). + // Completed jobs are excluded from the waiting query and pruned after SuccessfulJobRetentionTime. + options.StoreSuccessfulJobs = true; + options.SuccessfulJobRetentionTime = System.TimeSpan.FromDays(1); + + // A dedicated worker (with its own distributed lock "DemoFeesWorkerLock") that only processes + // the slow fee-calculation jobs, so they don't block other jobs. A default worker is added automatically + // and processes all the remaining job types (e.g. SendEmailJob). + options.AddDedicatedWorker("DemoFeesWorkerLock"); + + // Let each worker execute up to 4 jobs in parallel (each job claimed with its own distributed lock, + // so multiple application instances can execute different jobs concurrently). + options.MaxParallelJobExecutionCount = 4; }); } - public override Task OnApplicationInitializationAsync(ApplicationInitializationContext context) + public override async Task OnApplicationInitializationAsync(ApplicationInitializationContext context) { - //TODO: Configure console logging - //context - // .ServiceProvider - // .GetRequiredService() - // .AddConsole(LogLevel.Debug); + // Enqueue a few demo jobs. The fee-calculation jobs are handled by the dedicated worker, + // while SendEmailJob is handled by the default worker. + var backgroundJobManager = context.ServiceProvider.GetRequiredService(); - return Task.CompletedTask; + await backgroundJobManager.EnqueueAsync(new CalculateAwsFeesJobArgs { AccountId = "acc-1" }); + await backgroundJobManager.EnqueueAsync(new CalculateAzureFeesJobArgs { SubscriptionId = "sub-1" }); + await backgroundJobManager.EnqueueAsync(new SendEmailJobArgs { To = "user@example.com", Subject = "Welcome" }); } } diff --git a/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/Jobs/CalculateAwsFeesJob.cs b/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/Jobs/CalculateAwsFeesJob.cs new file mode 100644 index 0000000000..2262bbc68d --- /dev/null +++ b/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/Jobs/CalculateAwsFeesJob.cs @@ -0,0 +1,20 @@ +using System.Threading.Tasks; +using Microsoft.Extensions.Logging; +using Volo.Abp.DependencyInjection; + +namespace Volo.Abp.BackgroundJobs.DemoApp.Jobs; + +public class CalculateAwsFeesJobArgs +{ + public string AccountId { get; set; } = default!; +} + +public class CalculateAwsFeesJob : AsyncBackgroundJob, ITransientDependency +{ + public override Task ExecuteAsync(CalculateAwsFeesJobArgs args) + { + // A slow, resource-intensive job that is isolated on a dedicated worker (see DemoAppModule). + Logger.LogInformation($"[AWS fees] Calculating fees for account '{args.AccountId}'..."); + return Task.CompletedTask; + } +} diff --git a/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/Jobs/CalculateAzureFeesJob.cs b/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/Jobs/CalculateAzureFeesJob.cs new file mode 100644 index 0000000000..d0dc43cd94 --- /dev/null +++ b/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/Jobs/CalculateAzureFeesJob.cs @@ -0,0 +1,20 @@ +using System.Threading.Tasks; +using Microsoft.Extensions.Logging; +using Volo.Abp.DependencyInjection; + +namespace Volo.Abp.BackgroundJobs.DemoApp.Jobs; + +public class CalculateAzureFeesJobArgs +{ + public string SubscriptionId { get; set; } = default!; +} + +public class CalculateAzureFeesJob : AsyncBackgroundJob, ITransientDependency +{ + public override Task ExecuteAsync(CalculateAzureFeesJobArgs args) + { + // Isolated on the same dedicated fees worker as CalculateAwsFeesJob. + Logger.LogInformation($"[Azure fees] Calculating fees for subscription '{args.SubscriptionId}'..."); + return Task.CompletedTask; + } +} diff --git a/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/Jobs/SendEmailJob.cs b/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/Jobs/SendEmailJob.cs new file mode 100644 index 0000000000..846b06f4fe --- /dev/null +++ b/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/Jobs/SendEmailJob.cs @@ -0,0 +1,23 @@ +using System.Threading.Tasks; +using Microsoft.Extensions.Logging; +using Volo.Abp.DependencyInjection; + +namespace Volo.Abp.BackgroundJobs.DemoApp.Jobs; + +public class SendEmailJobArgs +{ + public string To { get; set; } = default!; + + public string Subject { get; set; } = default!; +} + +public class SendEmailJob : AsyncBackgroundJob, ITransientDependency +{ + public override Task ExecuteAsync(SendEmailJobArgs args) + { + // A fast job. It is NOT configured for a dedicated worker, so the default worker processes it + // without waiting behind the slow fee-calculation jobs. + Logger.LogInformation($"[Email] Sending '{args.Subject}' to '{args.To}'..."); + return Task.CompletedTask; + } +} diff --git a/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/Migrations/20260701082002_Added_CompletionTime_To_BackgroundJobs.Designer.cs b/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/Migrations/20260701082002_Added_CompletionTime_To_BackgroundJobs.Designer.cs new file mode 100644 index 0000000000..01241ef94a --- /dev/null +++ b/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/Migrations/20260701082002_Added_CompletionTime_To_BackgroundJobs.Designer.cs @@ -0,0 +1,99 @@ +// +using System; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Metadata; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using Volo.Abp.BackgroundJobs.DemoApp.Db; +using Volo.Abp.EntityFrameworkCore; + +#nullable disable + +namespace Volo.Abp.BackgroundJobs.DemoApp.Migrations +{ + [DbContext(typeof(DemoAppDbContext))] + [Migration("20260701082002_Added_CompletionTime_To_BackgroundJobs")] + partial class Added_CompletionTime_To_BackgroundJobs + { + /// + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("_Abp_DatabaseProvider", EfCoreDatabaseProvider.SqlServer) + .HasAnnotation("ProductVersion", "10.0.9") + .HasAnnotation("Relational:MaxIdentifierLength", 128); + + SqlServerModelBuilderExtensions.UseIdentityColumns(modelBuilder); + + modelBuilder.Entity("Volo.Abp.BackgroundJobs.BackgroundJobRecord", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uniqueidentifier"); + + b.Property("ApplicationName") + .HasMaxLength(96) + .HasColumnType("nvarchar(96)"); + + b.Property("CompletionTime") + .HasColumnType("datetime2"); + + b.Property("ConcurrencyStamp") + .IsConcurrencyToken() + .IsRequired() + .HasMaxLength(40) + .HasColumnType("nvarchar(40)") + .HasColumnName("ConcurrencyStamp"); + + b.Property("CreationTime") + .HasColumnType("datetime2") + .HasColumnName("CreationTime"); + + b.Property("ExtraProperties") + .IsRequired() + .HasColumnType("nvarchar(max)") + .HasColumnName("ExtraProperties"); + + b.Property("IsAbandoned") + .ValueGeneratedOnAdd() + .HasColumnType("bit") + .HasDefaultValue(false); + + b.Property("JobArgs") + .IsRequired() + .HasMaxLength(1048576) + .HasColumnType("nvarchar(max)"); + + b.Property("JobName") + .IsRequired() + .HasMaxLength(128) + .HasColumnType("nvarchar(128)"); + + b.Property("LastTryTime") + .HasColumnType("datetime2"); + + b.Property("NextTryTime") + .HasColumnType("datetime2"); + + b.Property("Priority") + .ValueGeneratedOnAdd() + .HasColumnType("tinyint") + .HasDefaultValue((byte)15); + + b.Property("TryCount") + .ValueGeneratedOnAdd() + .HasColumnType("smallint") + .HasDefaultValue((short)0); + + b.HasKey("Id"); + + b.HasIndex("ApplicationName", "CompletionTime", "IsAbandoned", "NextTryTime"); + + b.ToTable("AbpBackgroundJobs", (string)null); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/Migrations/20260701082002_Added_CompletionTime_To_BackgroundJobs.cs b/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/Migrations/20260701082002_Added_CompletionTime_To_BackgroundJobs.cs new file mode 100644 index 0000000000..96f35f3379 --- /dev/null +++ b/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/Migrations/20260701082002_Added_CompletionTime_To_BackgroundJobs.cs @@ -0,0 +1,47 @@ +using System; +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace Volo.Abp.BackgroundJobs.DemoApp.Migrations +{ + /// + public partial class Added_CompletionTime_To_BackgroundJobs : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropIndex( + name: "IX_AbpBackgroundJobs_IsAbandoned_NextTryTime", + table: "AbpBackgroundJobs"); + + migrationBuilder.AddColumn( + name: "CompletionTime", + table: "AbpBackgroundJobs", + type: "datetime2", + nullable: true); + + migrationBuilder.CreateIndex( + name: "IX_AbpBackgroundJobs_ApplicationName_CompletionTime_IsAbandoned_NextTryTime", + table: "AbpBackgroundJobs", + columns: new[] { "ApplicationName", "CompletionTime", "IsAbandoned", "NextTryTime" }); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropIndex( + name: "IX_AbpBackgroundJobs_ApplicationName_CompletionTime_IsAbandoned_NextTryTime", + table: "AbpBackgroundJobs"); + + migrationBuilder.DropColumn( + name: "CompletionTime", + table: "AbpBackgroundJobs"); + + migrationBuilder.CreateIndex( + name: "IX_AbpBackgroundJobs_IsAbandoned_NextTryTime", + table: "AbpBackgroundJobs", + columns: new[] { "IsAbandoned", "NextTryTime" }); + } + } +} diff --git a/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/Migrations/DemoAppDbContextModelSnapshot.cs b/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/Migrations/DemoAppDbContextModelSnapshot.cs index ab91bc354c..af43eb368f 100644 --- a/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/Migrations/DemoAppDbContextModelSnapshot.cs +++ b/modules/background-jobs/app/Volo.Abp.BackgroundJobs.DemoApp/Migrations/DemoAppDbContextModelSnapshot.cs @@ -19,7 +19,7 @@ namespace Volo.Abp.BackgroundJobs.DemoApp.Migrations #pragma warning disable 612, 618 modelBuilder .HasAnnotation("_Abp_DatabaseProvider", EfCoreDatabaseProvider.SqlServer) - .HasAnnotation("ProductVersion", "10.0.2") + .HasAnnotation("ProductVersion", "10.0.9") .HasAnnotation("Relational:MaxIdentifierLength", 128); SqlServerModelBuilderExtensions.UseIdentityColumns(modelBuilder); @@ -34,6 +34,9 @@ namespace Volo.Abp.BackgroundJobs.DemoApp.Migrations .HasMaxLength(96) .HasColumnType("nvarchar(96)"); + b.Property("CompletionTime") + .HasColumnType("datetime2"); + b.Property("ConcurrencyStamp") .IsConcurrencyToken() .IsRequired() @@ -83,7 +86,7 @@ namespace Volo.Abp.BackgroundJobs.DemoApp.Migrations b.HasKey("Id"); - b.HasIndex("IsAbandoned", "NextTryTime"); + b.HasIndex("ApplicationName", "CompletionTime", "IsAbandoned", "NextTryTime"); b.ToTable("AbpBackgroundJobs", (string)null); }); diff --git a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobRecord.cs b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobRecord.cs index c01f933781..18862c8b1d 100644 --- a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobRecord.cs +++ b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobRecord.cs @@ -48,6 +48,12 @@ public class BackgroundJobRecord : AggregateRoot, IHasCreationTime /// public virtual bool IsAbandoned { get; set; } + /// + /// The time this job was completed successfully. When set, the job is kept as history and excluded + /// from the waiting jobs query (set only when successful job persistence is enabled). + /// + public virtual DateTime? CompletionTime { get; set; } + /// /// Priority of this job. /// diff --git a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobStore.cs b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobStore.cs index c7dbd6fbef..9d24c6c236 100644 --- a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobStore.cs +++ b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobStore.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Threading; using System.Threading.Tasks; using Volo.Abp.DependencyInjection; using Volo.Abp.ObjectMapping; @@ -22,9 +23,13 @@ public class BackgroundJobStore : IBackgroundJobStore, ITransientDependency public virtual async Task FindAsync(Guid jobId) { - return ObjectMapper.Map( - await BackgroundJobRepository.FindAsync(jobId) - ); + var backgroundJobRecord = await BackgroundJobRepository.FindAsync(jobId); + if (backgroundJobRecord == null) + { + return null!; + } + + return ObjectMapper.Map(backgroundJobRecord); } public virtual async Task InsertAsync(BackgroundJobInfo jobInfo) @@ -34,18 +39,37 @@ public class BackgroundJobStore : IBackgroundJobStore, ITransientDependency ); } - public virtual async Task> GetWaitingJobsAsync(string applicationName, int maxResultCount) + public virtual async Task> GetWaitingJobsAsync(string? applicationName, int maxResultCount) { return ObjectMapper.Map, List>( await BackgroundJobRepository.GetWaitingListAsync(applicationName, maxResultCount) ); } + public virtual async Task> GetWaitingJobsAsync( + string? applicationName, + int maxResultCount, + BackgroundJobNameFilter? jobNameFilter) + { + return ObjectMapper.Map, List>( + await BackgroundJobRepository.GetWaitingListAsync(applicationName, maxResultCount, jobNameFilter) + ); + } + public virtual async Task DeleteAsync(Guid jobId) { await BackgroundJobRepository.DeleteAsync(jobId); } + public virtual async Task DeleteAsync( + string? applicationName, + DateTime completedBefore, + int maxResultCount, + CancellationToken cancellationToken = default) + { + return await BackgroundJobRepository.DeleteAsync(applicationName, completedBefore, maxResultCount, cancellationToken); + } + public virtual async Task UpdateAsync(BackgroundJobInfo jobInfo) { var backgroundJobRecord = await BackgroundJobRepository.FindAsync(jobInfo.Id); diff --git a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobsDomainMapperlyMappers.cs b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobsDomainMapperlyMappers.cs index 9ac5cef9bf..a605582201 100644 --- a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobsDomainMapperlyMappers.cs +++ b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/BackgroundJobsDomainMapperlyMappers.cs @@ -5,7 +5,7 @@ using System.Threading.Tasks; using Riok.Mapperly.Abstractions; using Volo.Abp.Mapperly; -namespace Volo.Abp.BackgroundJobs; +namespace Volo.Abp.BackgroundJobs; [Mapper(RequiredMappingStrategy = RequiredMappingStrategy.Target)] public partial class BackgroundJobInfoToBackgroundJobRecordMapper @@ -31,6 +31,6 @@ public partial class BackgroundJobRecordToBackgroundJobInfoMapper : MapperBase { public override partial BackgroundJobInfo Map(BackgroundJobRecord source); - + public override partial void Map(BackgroundJobRecord source, BackgroundJobInfo destination); } diff --git a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/IBackgroundJobRepository.cs b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/IBackgroundJobRepository.cs index 7e189df5c8..d4e8df2d83 100644 --- a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/IBackgroundJobRepository.cs +++ b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.Domain/Volo/Abp/BackgroundJobs/IBackgroundJobRepository.cs @@ -2,12 +2,26 @@ using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; -using JetBrains.Annotations; using Volo.Abp.Domain.Repositories; namespace Volo.Abp.BackgroundJobs; public interface IBackgroundJobRepository : IBasicRepository { - Task> GetWaitingListAsync([CanBeNull] string applicationName, int maxResultCount, CancellationToken cancellationToken = default); + Task> GetWaitingListAsync( + string? applicationName, + int maxResultCount, + CancellationToken cancellationToken = default); + + Task> GetWaitingListAsync( + string? applicationName, + int maxResultCount, + BackgroundJobNameFilter? jobNameFilter, + CancellationToken cancellationToken = default); + + Task DeleteAsync( + string? applicationName, + DateTime completedBefore, + int maxResultCount, + CancellationToken cancellationToken = default); } diff --git a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.EntityFrameworkCore/Volo/Abp/BackgroundJobs/EntityFrameworkCore/BackgroundJobsDbContextModelCreatingExtensions.cs b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.EntityFrameworkCore/Volo/Abp/BackgroundJobs/EntityFrameworkCore/BackgroundJobsDbContextModelCreatingExtensions.cs index 53a2c53843..70a73a8562 100644 --- a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.EntityFrameworkCore/Volo/Abp/BackgroundJobs/EntityFrameworkCore/BackgroundJobsDbContextModelCreatingExtensions.cs +++ b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.EntityFrameworkCore/Volo/Abp/BackgroundJobs/EntityFrameworkCore/BackgroundJobsDbContextModelCreatingExtensions.cs @@ -29,9 +29,10 @@ public static class BackgroundJobsDbContextModelCreatingExtensions b.Property(x => x.NextTryTime); b.Property(x => x.LastTryTime); b.Property(x => x.IsAbandoned).HasDefaultValue(false); + b.Property(x => x.CompletionTime); b.Property(x => x.Priority).HasDefaultValue(BackgroundJobPriority.Normal).HasSentinel(BackgroundJobPriority.Normal); - b.HasIndex(x => new { x.IsAbandoned, x.NextTryTime }); + b.HasIndex(x => new { x.ApplicationName, x.CompletionTime, x.IsAbandoned, x.NextTryTime }); b.ApplyObjectExtensionMappings(); }); diff --git a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.EntityFrameworkCore/Volo/Abp/BackgroundJobs/EntityFrameworkCore/EfCoreBackgroundJobRepository.cs b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.EntityFrameworkCore/Volo/Abp/BackgroundJobs/EntityFrameworkCore/EfCoreBackgroundJobRepository.cs index a81de3cdfb..5b55a7adf8 100644 --- a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.EntityFrameworkCore/Volo/Abp/BackgroundJobs/EntityFrameworkCore/EfCoreBackgroundJobRepository.cs +++ b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.EntityFrameworkCore/Volo/Abp/BackgroundJobs/EntityFrameworkCore/EfCoreBackgroundJobRepository.cs @@ -3,7 +3,6 @@ using System.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; -using JetBrains.Annotations; using Microsoft.EntityFrameworkCore; using Volo.Abp.Domain.Repositories.EntityFrameworkCore; using Volo.Abp.EntityFrameworkCore; @@ -23,20 +22,71 @@ public class EfCoreBackgroundJobRepository : EfCoreRepository> GetWaitingListAsync([CanBeNull] string applicationName, int maxResultCount, CancellationToken cancellationToken = default) + public virtual Task> GetWaitingListAsync( + string? applicationName, + int maxResultCount, + CancellationToken cancellationToken = default) { - return await (await GetWaitingListQueryAsync(applicationName, maxResultCount)).ToListAsync(GetCancellationToken(cancellationToken)); + return GetWaitingListAsync(applicationName, maxResultCount, null, cancellationToken); } - protected virtual async Task> GetWaitingListQueryAsync([CanBeNull] string applicationName, int maxResultCount) + public virtual async Task> GetWaitingListAsync( + string? applicationName, + int maxResultCount, + BackgroundJobNameFilter? jobNameFilter, + CancellationToken cancellationToken = default) + { + return await (await GetWaitingListQueryAsync(applicationName, maxResultCount, jobNameFilter)) + .ToListAsync(GetCancellationToken(cancellationToken)); + } + + protected virtual async Task> GetWaitingListQueryAsync( + string? applicationName, + int maxResultCount, + BackgroundJobNameFilter? jobNameFilter = null) { var now = Clock.Now; - return (await GetDbSetAsync()) + var filter = jobNameFilter ?? BackgroundJobNameFilter.None; + var jobNames = filter.JobNames.ToList(); + + var query = (await GetDbSetAsync()) .Where(t => t.ApplicationName == applicationName) - .Where(t => !t.IsAbandoned && t.NextTryTime <= now) + .Where(t => !t.IsAbandoned && t.CompletionTime == null && t.NextTryTime <= now); + + if (filter.Mode == BackgroundJobNameFilterMode.Include) + { + query = query.Where(t => jobNames.Contains(t.JobName)); + } + else if (filter.Mode == BackgroundJobNameFilterMode.Exclude) + { + query = query.Where(t => !jobNames.Contains(t.JobName)); + } + + return query .OrderByDescending(t => t.Priority) .ThenBy(t => t.TryCount) .ThenBy(t => t.NextTryTime) .Take(maxResultCount); } + + public virtual async Task DeleteAsync(string? applicationName, DateTime completedBefore, int maxResultCount, CancellationToken cancellationToken = default) + { + var token = GetCancellationToken(cancellationToken); + var dbSet = await GetDbSetAsync(); + + var ids = await dbSet + .Where(t => t.ApplicationName == applicationName) + .Where(t => t.CompletionTime != null && t.CompletionTime < completedBefore) + .OrderBy(t => t.CompletionTime) + .Select(t => t.Id) + .Take(maxResultCount) + .ToListAsync(token); + + if (ids.Count == 0) + { + return 0; + } + + return await dbSet.Where(t => ids.Contains(t.Id)).ExecuteDeleteAsync(token); + } } diff --git a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.MongoDB/Volo/Abp/BackgroundJobs/MongoDB/MongoBackgroundJobRepository.cs b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.MongoDB/Volo/Abp/BackgroundJobs/MongoDB/MongoBackgroundJobRepository.cs index 71a5680524..af4ab7ef89 100644 --- a/modules/background-jobs/src/Volo.Abp.BackgroundJobs.MongoDB/Volo/Abp/BackgroundJobs/MongoDB/MongoBackgroundJobRepository.cs +++ b/modules/background-jobs/src/Volo.Abp.BackgroundJobs.MongoDB/Volo/Abp/BackgroundJobs/MongoDB/MongoBackgroundJobRepository.cs @@ -3,7 +3,6 @@ using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; using System.Linq; -using JetBrains.Annotations; using MongoDB.Driver; using MongoDB.Driver.Linq; using Volo.Abp.Domain.Repositories.MongoDB; @@ -24,20 +23,78 @@ public class MongoBackgroundJobRepository : MongoDbRepository> GetWaitingListAsync([CanBeNull] string applicationName, int maxResultCount, CancellationToken cancellationToken = default) + public virtual Task> GetWaitingListAsync( + string? applicationName, + int maxResultCount, + CancellationToken cancellationToken = default) { - return await (await GetWaitingListQuery(applicationName, maxResultCount, cancellationToken)).ToListAsync(GetCancellationToken(cancellationToken)); + return GetWaitingListAsync(applicationName, maxResultCount, null, cancellationToken); } - protected virtual async Task> GetWaitingListQuery([CanBeNull] string applicationName, int maxResultCount, CancellationToken cancellationToken = default) + public virtual async Task> GetWaitingListAsync( + string? applicationName, + int maxResultCount, + BackgroundJobNameFilter? jobNameFilter, + CancellationToken cancellationToken = default) + { + return await (await GetWaitingListQuery(applicationName, maxResultCount, jobNameFilter, cancellationToken)) + .ToListAsync(GetCancellationToken(cancellationToken)); + } + + protected virtual async Task> GetWaitingListQuery( + string? applicationName, + int maxResultCount, + BackgroundJobNameFilter? jobNameFilter = null, + CancellationToken cancellationToken = default) { var now = Clock.Now; - return (await GetQueryableAsync(cancellationToken)) + var filter = jobNameFilter ?? BackgroundJobNameFilter.None; + var jobNames = filter.JobNames.ToList(); + + var query = (await GetQueryableAsync(cancellationToken)) .Where(t => t.ApplicationName == applicationName) - .Where(t => !t.IsAbandoned && t.NextTryTime <= now) + .Where(t => !t.IsAbandoned && t.CompletionTime == null && t.NextTryTime <= now); + + if (filter.Mode == BackgroundJobNameFilterMode.Include) + { + query = query.Where(t => jobNames.Contains(t.JobName)); + } + else if (filter.Mode == BackgroundJobNameFilterMode.Exclude) + { + query = query.Where(t => !jobNames.Contains(t.JobName)); + } + + return query .OrderByDescending(t => t.Priority) .ThenBy(t => t.TryCount) .ThenBy(t => t.NextTryTime) .Take(maxResultCount); } + + public virtual async Task DeleteAsync(string? applicationName, DateTime completedBefore, int maxResultCount, CancellationToken cancellationToken = default) + { + var token = GetCancellationToken(cancellationToken); + + var ids = await (await GetQueryableAsync(token)) + .Where(t => t.ApplicationName == applicationName) + .Where(t => t.CompletionTime != null && t.CompletionTime < completedBefore) + .OrderBy(t => t.CompletionTime) + .Select(t => t.Id) + .Take(maxResultCount) + .ToListAsync(token); + + if (ids.Count == 0) + { + return 0; + } + + var dbContext = await GetDbContextAsync(token); + var collection = dbContext.Collection(); + + var result = dbContext.SessionHandle != null + ? await collection.DeleteManyAsync(dbContext.SessionHandle, x => ids.Contains(x.Id), cancellationToken: token) + : await collection.DeleteManyAsync(x => ids.Contains(x.Id), cancellationToken: token); + + return (int)result.DeletedCount; + } } diff --git a/modules/background-jobs/test/Volo.Abp.BackgroundJobs.TestBase/Volo/Abp/BackgroundJobs/BackgroundJobRepository_Tests.cs b/modules/background-jobs/test/Volo.Abp.BackgroundJobs.TestBase/Volo/Abp/BackgroundJobs/BackgroundJobRepository_Tests.cs index c62cb43309..b717d28d15 100644 --- a/modules/background-jobs/test/Volo.Abp.BackgroundJobs.TestBase/Volo/Abp/BackgroundJobs/BackgroundJobRepository_Tests.cs +++ b/modules/background-jobs/test/Volo.Abp.BackgroundJobs.TestBase/Volo/Abp/BackgroundJobs/BackgroundJobRepository_Tests.cs @@ -1,4 +1,6 @@ -using System.Linq; +using System; +using System.Collections.Generic; +using System.Linq; using System.Threading.Tasks; using Shouldly; using Volo.Abp.Modularity; @@ -36,4 +38,99 @@ public abstract class BackgroundJobRepository_Tests : Background backgroundJobs.Any(j => j.ApplicationName == "App2").ShouldBeFalse(); backgroundJobs.Any(j => j.ApplicationName == null).ShouldBeFalse(); } + + [Fact] + public async Task Should_Filter_Waiting_List_By_Included_Job_Names() + { + // App1 waiting jobs: two "TestJobName" + one "OtherJobName". + var testJobs = await _backgroundJobRepository.GetWaitingListAsync("App1", 10, BackgroundJobNameFilter.Include(new[] { "TestJobName" })); + testJobs.Count.ShouldBe(2); + testJobs.ShouldAllBe(j => j.JobName == "TestJobName"); + + var otherJobs = await _backgroundJobRepository.GetWaitingListAsync("App1", 10, BackgroundJobNameFilter.Include(new[] { "OtherJobName" })); + otherJobs.Count.ShouldBe(1); + otherJobs.Single().JobName.ShouldBe("OtherJobName"); + + var none = await _backgroundJobRepository.GetWaitingListAsync("App1", 10, BackgroundJobNameFilter.Include(new[] { "NonExistentJob" })); + none.ShouldBeEmpty(); + } + + [Fact] + public async Task Should_Filter_Waiting_List_By_Excluded_Job_Names() + { + var withoutTestJob = await _backgroundJobRepository.GetWaitingListAsync("App1", 10, BackgroundJobNameFilter.Exclude(new[] { "TestJobName" })); + withoutTestJob.Count.ShouldBe(1); + withoutTestJob.Single().JobName.ShouldBe("OtherJobName"); + + var withoutNonExistent = await _backgroundJobRepository.GetWaitingListAsync("App1", 10, BackgroundJobNameFilter.Exclude(new[] { "NonExistentJob" })); + withoutNonExistent.Count.ShouldBe(3); + } + + [Fact] + public async Task Should_Return_Null_From_Store_When_Job_Not_Found() + { + // The parallel worker re-reads a claimed job under the lock; a job removed by another instance must return null (not throw). + var store = GetRequiredService(); + var found = await store.FindAsync(Guid.NewGuid()); + found.ShouldBeNull(); + } + + [Fact] + public async Task Should_Exclude_Completed_Jobs_From_Waiting_List() + { + var completedJobId = Guid.NewGuid(); + await _backgroundJobRepository.InsertAsync( + new BackgroundJobRecord(completedJobId) + { + ApplicationName = "App1", + JobName = "TestJobName", + JobArgs = "{ value: 1 }", + NextTryTime = _clock.Now.Subtract(TimeSpan.FromMinutes(1)), + Priority = BackgroundJobPriority.Normal, + IsAbandoned = false, + CompletionTime = _clock.Now, + CreationTime = _clock.Now.Subtract(TimeSpan.FromMinutes(2)), + TryCount = 1 + }, + autoSave: true); + + var waitingJobs = await _backgroundJobRepository.GetWaitingListAsync("App1", 100); + waitingJobs.ShouldNotContain(j => j.Id == completedJobId); + } + + [Fact] + public async Task Should_Delete_Old_Successful_Jobs_Of_The_Given_Application_Only() + { + var oldApp1Id = Guid.NewGuid(); + await _backgroundJobRepository.InsertAsync(NewCompletedJob(oldApp1Id, "App1", _clock.Now.Subtract(TimeSpan.FromDays(2))), autoSave: true); + + var recentApp1Id = Guid.NewGuid(); + await _backgroundJobRepository.InsertAsync(NewCompletedJob(recentApp1Id, "App1", _clock.Now), autoSave: true); + + var oldApp2Id = Guid.NewGuid(); + await _backgroundJobRepository.InsertAsync(NewCompletedJob(oldApp2Id, "App2", _clock.Now.Subtract(TimeSpan.FromDays(2))), autoSave: true); + + var deleted = await _backgroundJobRepository.DeleteAsync("App1", _clock.Now.Subtract(TimeSpan.FromDays(1)), 100); + deleted.ShouldBe(1); + + (await _backgroundJobRepository.FindAsync(oldApp1Id)).ShouldBeNull(); // App1 old completed → deleted + (await _backgroundJobRepository.FindAsync(recentApp1Id)).ShouldNotBeNull(); // App1 recent → kept + (await _backgroundJobRepository.FindAsync(oldApp2Id)).ShouldNotBeNull(); // App2 old → not touched (isolation) + } + + private BackgroundJobRecord NewCompletedJob(Guid id, string applicationName, DateTime completionTime) + { + return new BackgroundJobRecord(id) + { + ApplicationName = applicationName, + JobName = "TestJobName", + JobArgs = "{ value: 1 }", + NextTryTime = _clock.Now.Subtract(TimeSpan.FromMinutes(1)), + Priority = BackgroundJobPriority.Normal, + IsAbandoned = false, + CompletionTime = completionTime, + CreationTime = _clock.Now.Subtract(TimeSpan.FromMinutes(5)), + TryCount = 1 + }; + } } diff --git a/modules/background-jobs/test/Volo.Abp.BackgroundJobs.TestBase/Volo/Abp/BackgroundJobs/BackgroundJobsTestData.cs b/modules/background-jobs/test/Volo.Abp.BackgroundJobs.TestBase/Volo/Abp/BackgroundJobs/BackgroundJobsTestData.cs index 0d8b2bfd29..cd01e1b09a 100644 --- a/modules/background-jobs/test/Volo.Abp.BackgroundJobs.TestBase/Volo/Abp/BackgroundJobs/BackgroundJobsTestData.cs +++ b/modules/background-jobs/test/Volo.Abp.BackgroundJobs.TestBase/Volo/Abp/BackgroundJobs/BackgroundJobsTestData.cs @@ -8,4 +8,5 @@ public class BackgroundJobsTestData : ISingletonDependency public Guid JobId1 { get; } = Guid.NewGuid(); public Guid JobId2 { get; } = Guid.NewGuid(); public Guid JobId3 { get; } = Guid.NewGuid(); + public Guid JobId4 { get; } = Guid.NewGuid(); } diff --git a/modules/background-jobs/test/Volo.Abp.BackgroundJobs.TestBase/Volo/Abp/BackgroundJobs/BackgroundJobsTestDataBuilder.cs b/modules/background-jobs/test/Volo.Abp.BackgroundJobs.TestBase/Volo/Abp/BackgroundJobs/BackgroundJobsTestDataBuilder.cs index f6f4c37e12..47ebf103e2 100644 --- a/modules/background-jobs/test/Volo.Abp.BackgroundJobs.TestBase/Volo/Abp/BackgroundJobs/BackgroundJobsTestDataBuilder.cs +++ b/modules/background-jobs/test/Volo.Abp.BackgroundJobs.TestBase/Volo/Abp/BackgroundJobs/BackgroundJobsTestDataBuilder.cs @@ -67,5 +67,21 @@ public class BackgroundJobsTestDataBuilder : ITransientDependency TryCount = 2 } ); + + // App1 waiting job with a different job name, to verify job-name filtering. + await _backgroundJobRepository.InsertAsync( + new BackgroundJobRecord(_testData.JobId4) + { + ApplicationName = "App1", + JobName = "OtherJobName", + JobArgs = "{ value: 4 }", + NextTryTime = _clock.Now.Subtract(TimeSpan.FromMinutes(1)), + Priority = BackgroundJobPriority.Normal, + IsAbandoned = false, + LastTryTime = null, + CreationTime = _clock.Now.Subtract(TimeSpan.FromMinutes(3)), + TryCount = 0 + } + ); } }