feat(jobs): single-worker job execution (InMemory + Hangfire) #29
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,30 @@ | ||
| using Microsoft.Extensions.DependencyInjection; | ||
| using Microsoft.Extensions.DependencyInjection.Extensions; | ||
| using NeoReports.Abstractions; | ||
| using NeoReports.Jobs; | ||
|
|
||
| namespace NeoReports.Jobs.Hangfire.DependencyInjection; | ||
|
|
||
| /// <summary>DI entry points for the Hangfire job backend.</summary> | ||
| public static class ServiceCollectionExtensions | ||
| { | ||
| /// <summary> | ||
| /// Registers the Hangfire-backed scheduler plus the shared worker, invoker, and a no-op | ||
| /// checkpoint store. The caller is responsible for configuring Hangfire itself | ||
| /// (<c>AddHangfire(...)</c> with a storage provider and <c>AddHangfireServer()</c> for a single | ||
| /// server) and for registering the reports and core services (<c>AddReport</c> / <c>AddNeoReports</c>). | ||
| /// </summary> | ||
| /// <param name="services">The service collection.</param> | ||
| public static IServiceCollection AddNeoReportsHangfireJobs(this IServiceCollection services) | ||
| { | ||
| ArgumentNullException.ThrowIfNull(services); | ||
|
|
||
| services.TryAddSingleton<IJobStore, InMemoryJobStore>(); | ||
| services.TryAddSingleton<ICheckpointStore, NoOpCheckpointStore>(); | ||
| services.TryAddSingleton<ReportJobWorker>(); | ||
| services.TryAddSingleton<HangfireReportJobInvoker>(); | ||
| services.TryAddSingleton<IReportJobScheduler, HangfireJobScheduler>(); | ||
|
|
||
| return services; | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| using System.Collections.Concurrent; | ||
| using global::Hangfire; | ||
| using NeoReports.Abstractions; | ||
| using NeoReports.Jobs; | ||
|
|
||
| namespace NeoReports.Jobs.Hangfire; | ||
|
|
||
| /// <summary> | ||
| /// Single-server scheduler backed by Hangfire. Each enqueue creates a job record in the | ||
| /// <see cref="IJobStore"/> (status tracking) and an enqueued Hangfire background job (execution + | ||
| /// persistence). The two ids are mapped in-process so <see cref="CancelAsync"/> can abort the | ||
| /// running Hangfire job; the mapping is rebuilt per server run, matching the single-server model | ||
| /// (cross-restart cancellation is out of scope — a crashed job restarts from zero, D2). | ||
| /// </summary> | ||
| public sealed class HangfireJobScheduler : IReportJobScheduler | ||
| { | ||
| private readonly IBackgroundJobClient _client; | ||
| private readonly IJobStore _store; | ||
| private readonly ConcurrentDictionary<string, string> _hangfireIdByJobId = new(StringComparer.Ordinal); | ||
|
|
||
| /// <summary>Creates the scheduler.</summary> | ||
| /// <param name="client">Hangfire background job client.</param> | ||
| /// <param name="store">Store used to create and track jobs.</param> | ||
| public HangfireJobScheduler(IBackgroundJobClient client, IJobStore store) | ||
| { | ||
| _client = client; | ||
| _store = store; | ||
| } | ||
|
|
||
| /// <inheritdoc /> | ||
| public async Task<string> EnqueueAsync(ReportJobRequest request, CancellationToken cancellationToken) | ||
| { | ||
| ArgumentNullException.ThrowIfNull(request); | ||
|
|
||
| var job = await _store.CreateAsync(request, cancellationToken).ConfigureAwait(false); | ||
| var parametersJson = JobParameters.Serialize(request.Parameters); | ||
|
|
||
| // CancellationToken.None here is a placeholder; Hangfire substitutes a real token at run time. | ||
| var hangfireId = _client.Enqueue<HangfireReportJobInvoker>( | ||
| invoker => invoker.ExecuteAsync(job.Id, request.ReportName, parametersJson, CancellationToken.None)); | ||
|
|
||
| _hangfireIdByJobId[job.Id] = hangfireId; | ||
| return job.Id; | ||
| } | ||
|
|
||
| /// <inheritdoc /> | ||
| public Task<ReportJob?> GetAsync(string jobId, CancellationToken cancellationToken) => | ||
| _store.GetAsync(jobId, cancellationToken); | ||
|
|
||
| /// <inheritdoc /> | ||
| public Task<bool> CancelAsync(string jobId, CancellationToken cancellationToken) | ||
| { | ||
| if (_hangfireIdByJobId.TryGetValue(jobId, out var hangfireId)) | ||
| { | ||
| // Deleting the job trips the CancellationToken Hangfire injected into the invoker, so | ||
| // the pipeline stops cooperatively and the worker records a Cancelled status. | ||
| var deleted = _client.Delete(hangfireId); | ||
| return Task.FromResult(deleted); | ||
| } | ||
|
|
||
| return Task.FromResult(false); | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| using NeoReports.Jobs; | ||
|
|
||
| namespace NeoReports.Jobs.Hangfire; | ||
|
|
||
| /// <summary> | ||
| /// The unit of work Hangfire invokes for a report job. Hangfire resolves it from DI, persists its | ||
| /// arguments in storage (so the job survives restarts), and injects a <see cref="CancellationToken"/> | ||
| /// that is tripped on server shutdown or when the background job is aborted/deleted — which is how | ||
| /// cooperative cancellation reaches the pipeline. | ||
| /// </summary> | ||
| public sealed class HangfireReportJobInvoker | ||
| { | ||
| private readonly ReportJobWorker _worker; | ||
|
|
||
| /// <summary>Creates the invoker.</summary> | ||
| /// <param name="worker">The shared job worker.</param> | ||
| public HangfireReportJobInvoker(ReportJobWorker worker) => _worker = worker; | ||
|
|
||
| /// <summary> | ||
| /// Executes the job. Called by Hangfire; parameters arrive as a JSON string because Hangfire | ||
| /// serializes method arguments into its storage. | ||
| /// </summary> | ||
| /// <param name="jobId">The NeoReports job id (created in the store before enqueueing).</param> | ||
| /// <param name="reportName">The registered report to run.</param> | ||
| /// <param name="parametersJson">Parameters serialized by <see cref="JobParameters.Serialize"/>.</param> | ||
| /// <param name="cancellationToken">Injected by Hangfire; cancels on shutdown/abort.</param> | ||
| public Task ExecuteAsync(string jobId, string reportName, string parametersJson, CancellationToken cancellationToken) | ||
| { | ||
| var parameters = JobParameters.Deserialize(parametersJson); | ||
| return _worker.RunAsync(jobId, reportName, parameters, cancellationToken); | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,19 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk"> | ||
|
|
||
| <PropertyGroup> | ||
| <TargetFrameworks>net8.0;net9.0</TargetFrameworks> | ||
| <Description>Hangfire single-server job backend for NeoReports.</Description> | ||
| <PackageTags>reports;reporting;jobs;hangfire;background</PackageTags> | ||
| </PropertyGroup> | ||
|
|
||
| <ItemGroup> | ||
| <ProjectReference Include="..\NeoReports.Jobs\NeoReports.Jobs.csproj" /> | ||
| <ProjectReference Include="..\..\NeoReports.Abstractions\NeoReports.Abstractions.csproj" /> | ||
| </ItemGroup> | ||
|
|
||
| <ItemGroup> | ||
| <PackageReference Include="Hangfire.Core" /> | ||
| <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" /> | ||
| </ItemGroup> | ||
|
|
||
| </Project> |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,28 @@ | ||
| using Microsoft.Extensions.DependencyInjection; | ||
| using Microsoft.Extensions.DependencyInjection.Extensions; | ||
| using NeoReports.Abstractions; | ||
|
|
||
| namespace NeoReports.Jobs.DependencyInjection; | ||
|
|
||
| /// <summary>DI entry points for NeoReports job execution.</summary> | ||
| public static class ServiceCollectionExtensions | ||
| { | ||
| /// <summary> | ||
| /// Registers the in-memory job backend: <see cref="InMemoryJobStore"/>, | ||
| /// <see cref="NoOpCheckpointStore"/>, the shared <see cref="ReportJobWorker"/>, and the | ||
| /// in-process <see cref="InMemoryJobScheduler"/>. Assumes <c>AddNeoReports</c> and the reports | ||
| /// have already been registered so an <see cref="Core.Pipeline.IReportRunner"/> is available. | ||
| /// </summary> | ||
| /// <param name="services">The service collection.</param> | ||
| public static IServiceCollection AddNeoReportsInMemoryJobs(this IServiceCollection services) | ||
| { | ||
| ArgumentNullException.ThrowIfNull(services); | ||
|
|
||
| services.TryAddSingleton<IJobStore, InMemoryJobStore>(); | ||
| services.TryAddSingleton<ICheckpointStore, NoOpCheckpointStore>(); | ||
| services.TryAddSingleton<ReportJobWorker>(); | ||
| services.TryAddSingleton<IReportJobScheduler, InMemoryJobScheduler>(); | ||
|
|
||
| return services; | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,105 @@ | ||
| using System.Collections.Concurrent; | ||
| using NeoReports.Abstractions; | ||
|
|
||
| namespace NeoReports.Jobs; | ||
|
|
||
| /// <summary> | ||
| /// In-process scheduler that runs each enqueued job on a background <see cref="Task"/>. Suitable | ||
| /// for dev and tests; production uses the Hangfire single-server scheduler for persistence across | ||
| /// restarts. Cancellation is cooperative via a per-job <see cref="CancellationTokenSource"/>. | ||
| /// </summary> | ||
| public sealed class InMemoryJobScheduler : IReportJobScheduler, IAsyncDisposable | ||
| { | ||
| private readonly IJobStore _store; | ||
| private readonly ReportJobWorker _worker; | ||
| private readonly ConcurrentDictionary<string, CancellationTokenSource> _running = new(StringComparer.Ordinal); | ||
| private readonly ConcurrentDictionary<string, Task> _tasks = new(StringComparer.Ordinal); | ||
|
|
||
| /// <summary>Creates the scheduler.</summary> | ||
| /// <param name="store">Store used to create and track jobs.</param> | ||
| /// <param name="worker">Worker that executes each job.</param> | ||
| public InMemoryJobScheduler(IJobStore store, ReportJobWorker worker) | ||
| { | ||
| _store = store; | ||
| _worker = worker; | ||
| } | ||
|
|
||
| /// <inheritdoc /> | ||
| public async Task<string> EnqueueAsync(ReportJobRequest request, CancellationToken cancellationToken) | ||
| { | ||
| ArgumentNullException.ThrowIfNull(request); | ||
|
|
||
| var job = await _store.CreateAsync(request, cancellationToken).ConfigureAwait(false); | ||
| var cts = new CancellationTokenSource(); | ||
| _running[job.Id] = cts; | ||
|
|
||
| var task = Task.Run( | ||
| async () => | ||
| { | ||
| try | ||
| { | ||
| await _worker.RunAsync(job.Id, request.ReportName, request.Parameters, cts.Token) | ||
| .ConfigureAwait(false); | ||
| } | ||
| catch (Exception) | ||
| { | ||
| // The worker already recorded the failure in the store; swallow here so the | ||
| // background task never crashes the process. | ||
| } | ||
|
thiagoluga marked this conversation as resolved.
Dismissed
|
||
| finally | ||
| { | ||
| _running.TryRemove(job.Id, out _); | ||
| _tasks.TryRemove(job.Id, out _); | ||
| cts.Dispose(); | ||
| } | ||
| }, | ||
| CancellationToken.None); | ||
|
|
||
| _tasks[job.Id] = task; | ||
| return job.Id; | ||
| } | ||
|
|
||
| /// <inheritdoc /> | ||
| public Task<ReportJob?> GetAsync(string jobId, CancellationToken cancellationToken) => | ||
| _store.GetAsync(jobId, cancellationToken); | ||
|
|
||
| /// <inheritdoc /> | ||
| public Task<bool> CancelAsync(string jobId, CancellationToken cancellationToken) | ||
| { | ||
| if (_running.TryGetValue(jobId, out var cts)) | ||
| { | ||
| cts.Cancel(); | ||
| return Task.FromResult(true); | ||
| } | ||
|
|
||
| // Not running: either unknown or already finished — nothing to cancel. | ||
| return Task.FromResult(false); | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Awaits a job's background task (test/shutdown helper). Returns immediately if the job is | ||
| /// unknown or already finished. | ||
| /// </summary> | ||
| /// <param name="jobId">The job id.</param> | ||
| public Task WaitForCompletionAsync(string jobId) => | ||
| _tasks.TryGetValue(jobId, out var task) ? task : Task.CompletedTask; | ||
|
|
||
| /// <summary>Cancels all running jobs and waits for their background tasks to unwind.</summary> | ||
| public async ValueTask DisposeAsync() | ||
| { | ||
| foreach (var cts in _running.Values) | ||
| { | ||
| try { cts.Cancel(); } | ||
|
Check warning on line 92 in src/Jobs/NeoReports.Jobs/InMemoryJobScheduler.cs
|
||
| catch (ObjectDisposedException) { } | ||
|
Check warning on line 93 in src/Jobs/NeoReports.Jobs/InMemoryJobScheduler.cs
|
||
|
|
||
| } | ||
|
|
||
| try | ||
| { | ||
| await Task.WhenAll(_tasks.Values).ConfigureAwait(false); | ||
| } | ||
| catch (Exception) | ||
| { | ||
| // Best-effort drain on shutdown. | ||
| } | ||
|
thiagoluga marked this conversation as resolved.
Dismissed
|
||
| } | ||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.