Skip to content

Commit 93c7d6f

Browse files
thiagolugathiagoluga
committed
feat(jobs): single-worker job execution (InMemory + Hangfire)
PR 6. Adds the job layer on top of the pipeline. NeoReports.Jobs (base): - ReportJobWorker: shared lifecycle core (running → completed/failed/cancelled, stats) used by every scheduler so semantics are identical. - InMemoryJobStore (thread-safe), NoOpCheckpointStore (v1 restarts from zero, D2). - InMemoryJobScheduler: runs each job on a background task; cooperative cancel via a per-job CancellationTokenSource. - JobParameters: JSON (de)serialization of run-time parameters for persistence. - DI: AddNeoReportsInMemoryJobs(). NeoReports.Jobs.Hangfire: - HangfireJobScheduler + HangfireReportJobInvoker, reusing ReportJobWorker. Params persisted as JSON args; cancellation via the CancellationToken Hangfire injects (CancelAsync deletes the background job). DI: AddNeoReportsHangfireJobs(). Idempotent restart (CA-16) is inherited from the pipeline: per-job temp staging, upload only after a fully successful run — a cancelled/crashed job publishes no partial file. Tests (18): store, JobParameters round-trip, lifecycle queued→running→completed, cancellation→cancelled (CA-15), source-failure→failed, cancelled-run-publishes- nothing + fresh-run-completes (CA-16), and Hangfire enqueue/cancel/invoker via real Hangfire.InMemory storage. Decisions D18 (jobs packaging) and D19 (worker/cancellation) recorded; plan.md PR 6 marked done.
1 parent 3dcb0e4 commit 93c7d6f

18 files changed

Lines changed: 1125 additions & 0 deletions

‎NeoReports.sln‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,12 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "02-sql-to-xlsx-s3", "sample
4343
EndProject
4444
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "NeoReports.Benchmarks", "benchmarks\NeoReports.Benchmarks\NeoReports.Benchmarks.csproj", "{C4AE3756-5EA7-48E1-AF38-0DE79AA4B0DC}"
4545
EndProject
46+
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "NeoReports.Jobs", "src\Jobs\NeoReports.Jobs\NeoReports.Jobs.csproj", "{55D991E5-0FB3-4989-8A1F-D9071F82280C}"
47+
EndProject
48+
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "NeoReports.Jobs.Hangfire", "src\Jobs\NeoReports.Jobs.Hangfire\NeoReports.Jobs.Hangfire.csproj", "{CDE09705-9C9A-4D5B-B8EF-A86BCD81CF28}"
49+
EndProject
50+
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "NeoReports.Jobs.UnitTests", "tests\NeoReports.Jobs.UnitTests\NeoReports.Jobs.UnitTests.csproj", "{98F4D154-BE1B-4CC5-9F75-51419E6F75BA}"
51+
EndProject
4652
Global
4753
GlobalSection(SolutionConfigurationPlatforms) = preSolution
4854
Debug|Any CPU = Debug|Any CPU
@@ -245,6 +251,42 @@ Global
245251
{C4AE3756-5EA7-48E1-AF38-0DE79AA4B0DC}.Release|x64.Build.0 = Release|Any CPU
246252
{C4AE3756-5EA7-48E1-AF38-0DE79AA4B0DC}.Release|x86.ActiveCfg = Release|Any CPU
247253
{C4AE3756-5EA7-48E1-AF38-0DE79AA4B0DC}.Release|x86.Build.0 = Release|Any CPU
254+
{55D991E5-0FB3-4989-8A1F-D9071F82280C}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
255+
{55D991E5-0FB3-4989-8A1F-D9071F82280C}.Debug|Any CPU.Build.0 = Debug|Any CPU
256+
{55D991E5-0FB3-4989-8A1F-D9071F82280C}.Debug|x64.ActiveCfg = Debug|Any CPU
257+
{55D991E5-0FB3-4989-8A1F-D9071F82280C}.Debug|x64.Build.0 = Debug|Any CPU
258+
{55D991E5-0FB3-4989-8A1F-D9071F82280C}.Debug|x86.ActiveCfg = Debug|Any CPU
259+
{55D991E5-0FB3-4989-8A1F-D9071F82280C}.Debug|x86.Build.0 = Debug|Any CPU
260+
{55D991E5-0FB3-4989-8A1F-D9071F82280C}.Release|Any CPU.ActiveCfg = Release|Any CPU
261+
{55D991E5-0FB3-4989-8A1F-D9071F82280C}.Release|Any CPU.Build.0 = Release|Any CPU
262+
{55D991E5-0FB3-4989-8A1F-D9071F82280C}.Release|x64.ActiveCfg = Release|Any CPU
263+
{55D991E5-0FB3-4989-8A1F-D9071F82280C}.Release|x64.Build.0 = Release|Any CPU
264+
{55D991E5-0FB3-4989-8A1F-D9071F82280C}.Release|x86.ActiveCfg = Release|Any CPU
265+
{55D991E5-0FB3-4989-8A1F-D9071F82280C}.Release|x86.Build.0 = Release|Any CPU
266+
{CDE09705-9C9A-4D5B-B8EF-A86BCD81CF28}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
267+
{CDE09705-9C9A-4D5B-B8EF-A86BCD81CF28}.Debug|Any CPU.Build.0 = Debug|Any CPU
268+
{CDE09705-9C9A-4D5B-B8EF-A86BCD81CF28}.Debug|x64.ActiveCfg = Debug|Any CPU
269+
{CDE09705-9C9A-4D5B-B8EF-A86BCD81CF28}.Debug|x64.Build.0 = Debug|Any CPU
270+
{CDE09705-9C9A-4D5B-B8EF-A86BCD81CF28}.Debug|x86.ActiveCfg = Debug|Any CPU
271+
{CDE09705-9C9A-4D5B-B8EF-A86BCD81CF28}.Debug|x86.Build.0 = Debug|Any CPU
272+
{CDE09705-9C9A-4D5B-B8EF-A86BCD81CF28}.Release|Any CPU.ActiveCfg = Release|Any CPU
273+
{CDE09705-9C9A-4D5B-B8EF-A86BCD81CF28}.Release|Any CPU.Build.0 = Release|Any CPU
274+
{CDE09705-9C9A-4D5B-B8EF-A86BCD81CF28}.Release|x64.ActiveCfg = Release|Any CPU
275+
{CDE09705-9C9A-4D5B-B8EF-A86BCD81CF28}.Release|x64.Build.0 = Release|Any CPU
276+
{CDE09705-9C9A-4D5B-B8EF-A86BCD81CF28}.Release|x86.ActiveCfg = Release|Any CPU
277+
{CDE09705-9C9A-4D5B-B8EF-A86BCD81CF28}.Release|x86.Build.0 = Release|Any CPU
278+
{98F4D154-BE1B-4CC5-9F75-51419E6F75BA}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
279+
{98F4D154-BE1B-4CC5-9F75-51419E6F75BA}.Debug|Any CPU.Build.0 = Debug|Any CPU
280+
{98F4D154-BE1B-4CC5-9F75-51419E6F75BA}.Debug|x64.ActiveCfg = Debug|Any CPU
281+
{98F4D154-BE1B-4CC5-9F75-51419E6F75BA}.Debug|x64.Build.0 = Debug|Any CPU
282+
{98F4D154-BE1B-4CC5-9F75-51419E6F75BA}.Debug|x86.ActiveCfg = Debug|Any CPU
283+
{98F4D154-BE1B-4CC5-9F75-51419E6F75BA}.Debug|x86.Build.0 = Debug|Any CPU
284+
{98F4D154-BE1B-4CC5-9F75-51419E6F75BA}.Release|Any CPU.ActiveCfg = Release|Any CPU
285+
{98F4D154-BE1B-4CC5-9F75-51419E6F75BA}.Release|Any CPU.Build.0 = Release|Any CPU
286+
{98F4D154-BE1B-4CC5-9F75-51419E6F75BA}.Release|x64.ActiveCfg = Release|Any CPU
287+
{98F4D154-BE1B-4CC5-9F75-51419E6F75BA}.Release|x64.Build.0 = Release|Any CPU
288+
{98F4D154-BE1B-4CC5-9F75-51419E6F75BA}.Release|x86.ActiveCfg = Release|Any CPU
289+
{98F4D154-BE1B-4CC5-9F75-51419E6F75BA}.Release|x86.Build.0 = Release|Any CPU
248290
EndGlobalSection
249291
GlobalSection(SolutionProperties) = preSolution
250292
HideSolutionNode = FALSE
@@ -266,5 +308,8 @@ Global
266308
{5A33C69B-40A0-49AF-87F4-40B75D0D53A5} = {22222222-2222-2222-2222-222222222222}
267309
{34079216-845C-417E-8511-FDD3F075CDCD} = {44444444-4444-4444-4444-444444444444}
268310
{C4AE3756-5EA7-48E1-AF38-0DE79AA4B0DC} = {33333333-3333-3333-3333-333333333333}
311+
{55D991E5-0FB3-4989-8A1F-D9071F82280C} = {11111111-1111-1111-1111-111111111111}
312+
{CDE09705-9C9A-4D5B-B8EF-A86BCD81CF28} = {11111111-1111-1111-1111-111111111111}
313+
{98F4D154-BE1B-4CC5-9F75-51419E6F75BA} = {22222222-2222-2222-2222-222222222222}
269314
EndGlobalSection
270315
EndGlobal

‎src/Jobs/NeoReports.Jobs.Hangfire/DependencyInjection/ServiceCollectionExtensions.cs‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
using Microsoft.Extensions.DependencyInjection;
2+
using Microsoft.Extensions.DependencyInjection.Extensions;
3+
using NeoReports.Abstractions;
4+
using NeoReports.Jobs;
5+
6+
namespace NeoReports.Jobs.Hangfire.DependencyInjection;
7+
8+
/// <summary>DI entry points for the Hangfire job backend.</summary>
9+
public static class ServiceCollectionExtensions
10+
{
11+
/// <summary>
12+
/// Registers the Hangfire-backed scheduler plus the shared worker, invoker, and a no-op
13+
/// checkpoint store. The caller is responsible for configuring Hangfire itself
14+
/// (<c>AddHangfire(...)</c> with a storage provider and <c>AddHangfireServer()</c> for a single
15+
/// server) and for registering the reports and core services (<c>AddReport</c> / <c>AddNeoReports</c>).
16+
/// </summary>
17+
/// <param name="services">The service collection.</param>
18+
public static IServiceCollection AddNeoReportsHangfireJobs(this IServiceCollection services)
19+
{
20+
ArgumentNullException.ThrowIfNull(services);
21+
22+
services.TryAddSingleton<IJobStore, InMemoryJobStore>();
23+
services.TryAddSingleton<ICheckpointStore, NoOpCheckpointStore>();
24+
services.TryAddSingleton<ReportJobWorker>();
25+
services.TryAddSingleton<HangfireReportJobInvoker>();
26+
services.TryAddSingleton<IReportJobScheduler, HangfireJobScheduler>();
27+
28+
return services;
29+
}
30+
}

‎src/Jobs/NeoReports.Jobs.Hangfire/HangfireJobScheduler.cs‎

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,63 @@
1+
using System.Collections.Concurrent;
2+
using global::Hangfire;
3+
using NeoReports.Abstractions;
4+
using NeoReports.Jobs;
5+
6+
namespace NeoReports.Jobs.Hangfire;
7+
8+
/// <summary>
9+
/// Single-server scheduler backed by Hangfire. Each enqueue creates a job record in the
10+
/// <see cref="IJobStore"/> (status tracking) and an enqueued Hangfire background job (execution +
11+
/// persistence). The two ids are mapped in-process so <see cref="CancelAsync"/> can abort the
12+
/// running Hangfire job; the mapping is rebuilt per server run, matching the single-server model
13+
/// (cross-restart cancellation is out of scope — a crashed job restarts from zero, D2).
14+
/// </summary>
15+
public sealed class HangfireJobScheduler : IReportJobScheduler
16+
{
17+
private readonly IBackgroundJobClient _client;
18+
private readonly IJobStore _store;
19+
private readonly ConcurrentDictionary<string, string> _hangfireIdByJobId = new(StringComparer.Ordinal);
20+
21+
/// <summary>Creates the scheduler.</summary>
22+
/// <param name="client">Hangfire background job client.</param>
23+
/// <param name="store">Store used to create and track jobs.</param>
24+
public HangfireJobScheduler(IBackgroundJobClient client, IJobStore store)
25+
{
26+
_client = client;
27+
_store = store;
28+
}
29+
30+
/// <inheritdoc />
31+
public async Task<string> EnqueueAsync(ReportJobRequest request, CancellationToken cancellationToken)
32+
{
33+
ArgumentNullException.ThrowIfNull(request);
34+
35+
var job = await _store.CreateAsync(request, cancellationToken).ConfigureAwait(false);
36+
var parametersJson = JobParameters.Serialize(request.Parameters);
37+
38+
// CancellationToken.None here is a placeholder; Hangfire substitutes a real token at run time.
39+
var hangfireId = _client.Enqueue<HangfireReportJobInvoker>(
40+
invoker => invoker.ExecuteAsync(job.Id, request.ReportName, parametersJson, CancellationToken.None));
41+
42+
_hangfireIdByJobId[job.Id] = hangfireId;
43+
return job.Id;
44+
}
45+
46+
/// <inheritdoc />
47+
public Task<ReportJob?> GetAsync(string jobId, CancellationToken cancellationToken) =>
48+
_store.GetAsync(jobId, cancellationToken);
49+
50+
/// <inheritdoc />
51+
public Task<bool> CancelAsync(string jobId, CancellationToken cancellationToken)
52+
{
53+
if (_hangfireIdByJobId.TryGetValue(jobId, out var hangfireId))
54+
{
55+
// Deleting the job trips the CancellationToken Hangfire injected into the invoker, so
56+
// the pipeline stops cooperatively and the worker records a Cancelled status.
57+
var deleted = _client.Delete(hangfireId);
58+
return Task.FromResult(deleted);
59+
}
60+
61+
return Task.FromResult(false);
62+
}
63+
}

‎src/Jobs/NeoReports.Jobs.Hangfire/HangfireReportJobInvoker.cs‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
using NeoReports.Jobs;
2+
3+
namespace NeoReports.Jobs.Hangfire;
4+
5+
/// <summary>
6+
/// The unit of work Hangfire invokes for a report job. Hangfire resolves it from DI, persists its
7+
/// arguments in storage (so the job survives restarts), and injects a <see cref="CancellationToken"/>
8+
/// that is tripped on server shutdown or when the background job is aborted/deleted — which is how
9+
/// cooperative cancellation reaches the pipeline.
10+
/// </summary>
11+
public sealed class HangfireReportJobInvoker
12+
{
13+
private readonly ReportJobWorker _worker;
14+
15+
/// <summary>Creates the invoker.</summary>
16+
/// <param name="worker">The shared job worker.</param>
17+
public HangfireReportJobInvoker(ReportJobWorker worker) => _worker = worker;
18+
19+
/// <summary>
20+
/// Executes the job. Called by Hangfire; parameters arrive as a JSON string because Hangfire
21+
/// serializes method arguments into its storage.
22+
/// </summary>
23+
/// <param name="jobId">The NeoReports job id (created in the store before enqueueing).</param>
24+
/// <param name="reportName">The registered report to run.</param>
25+
/// <param name="parametersJson">Parameters serialized by <see cref="JobParameters.Serialize"/>.</param>
26+
/// <param name="cancellationToken">Injected by Hangfire; cancels on shutdown/abort.</param>
27+
public Task ExecuteAsync(string jobId, string reportName, string parametersJson, CancellationToken cancellationToken)
28+
{
29+
var parameters = JobParameters.Deserialize(parametersJson);
30+
return _worker.RunAsync(jobId, reportName, parameters, cancellationToken);
31+
}
32+
}

‎src/Jobs/NeoReports.Jobs.Hangfire/NeoReports.Jobs.Hangfire.csproj‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
<Project Sdk="Microsoft.NET.Sdk">
2+
3+
<PropertyGroup>
4+
<TargetFrameworks>net8.0;net9.0</TargetFrameworks>
5+
<Description>Hangfire single-server job backend for NeoReports.</Description>
6+
<PackageTags>reports;reporting;jobs;hangfire;background</PackageTags>
7+
</PropertyGroup>
8+
9+
<ItemGroup>
10+
<ProjectReference Include="..\NeoReports.Jobs\NeoReports.Jobs.csproj" />
11+
<ProjectReference Include="..\..\NeoReports.Abstractions\NeoReports.Abstractions.csproj" />
12+
</ItemGroup>
13+
14+
<ItemGroup>
15+
<PackageReference Include="Hangfire.Core" />
16+
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" />
17+
</ItemGroup>
18+
19+
</Project>

‎src/Jobs/NeoReports.Jobs/DependencyInjection/ServiceCollectionExtensions.cs‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
using Microsoft.Extensions.DependencyInjection;
2+
using Microsoft.Extensions.DependencyInjection.Extensions;
3+
using NeoReports.Abstractions;
4+
5+
namespace NeoReports.Jobs.DependencyInjection;
6+
7+
/// <summary>DI entry points for NeoReports job execution.</summary>
8+
public static class ServiceCollectionExtensions
9+
{
10+
/// <summary>
11+
/// Registers the in-memory job backend: <see cref="InMemoryJobStore"/>,
12+
/// <see cref="NoOpCheckpointStore"/>, the shared <see cref="ReportJobWorker"/>, and the
13+
/// in-process <see cref="InMemoryJobScheduler"/>. Assumes <c>AddNeoReports</c> and the reports
14+
/// have already been registered so an <see cref="Core.Pipeline.IReportRunner"/> is available.
15+
/// </summary>
16+
/// <param name="services">The service collection.</param>
17+
public static IServiceCollection AddNeoReportsInMemoryJobs(this IServiceCollection services)
18+
{
19+
ArgumentNullException.ThrowIfNull(services);
20+
21+
services.TryAddSingleton<IJobStore, InMemoryJobStore>();
22+
services.TryAddSingleton<ICheckpointStore, NoOpCheckpointStore>();
23+
services.TryAddSingleton<ReportJobWorker>();
24+
services.TryAddSingleton<IReportJobScheduler, InMemoryJobScheduler>();
25+
26+
return services;
27+
}
28+
}

‎src/Jobs/NeoReports.Jobs/InMemoryJobScheduler.cs‎

Lines changed: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,105 @@
1+
using System.Collections.Concurrent;
2+
using NeoReports.Abstractions;
3+
4+
namespace NeoReports.Jobs;
5+
6+
/// <summary>
7+
/// In-process scheduler that runs each enqueued job on a background <see cref="Task"/>. Suitable
8+
/// for dev and tests; production uses the Hangfire single-server scheduler for persistence across
9+
/// restarts. Cancellation is cooperative via a per-job <see cref="CancellationTokenSource"/>.
10+
/// </summary>
11+
public sealed class InMemoryJobScheduler : IReportJobScheduler, IAsyncDisposable
12+
{
13+
private readonly IJobStore _store;
14+
private readonly ReportJobWorker _worker;
15+
private readonly ConcurrentDictionary<string, CancellationTokenSource> _running = new(StringComparer.Ordinal);
16+
private readonly ConcurrentDictionary<string, Task> _tasks = new(StringComparer.Ordinal);
17+
18+
/// <summary>Creates the scheduler.</summary>
19+
/// <param name="store">Store used to create and track jobs.</param>
20+
/// <param name="worker">Worker that executes each job.</param>
21+
public InMemoryJobScheduler(IJobStore store, ReportJobWorker worker)
22+
{
23+
_store = store;
24+
_worker = worker;
25+
}
26+
27+
/// <inheritdoc />
28+
public async Task<string> EnqueueAsync(ReportJobRequest request, CancellationToken cancellationToken)
29+
{
30+
ArgumentNullException.ThrowIfNull(request);
31+
32+
var job = await _store.CreateAsync(request, cancellationToken).ConfigureAwait(false);
33+
var cts = new CancellationTokenSource();
34+
_running[job.Id] = cts;
35+
36+
var task = Task.Run(
37+
async () =>
38+
{
39+
try
40+
{
41+
await _worker.RunAsync(job.Id, request.ReportName, request.Parameters, cts.Token)
42+
.ConfigureAwait(false);
43+
}
44+
catch (Exception)
45+
{
46+
// The worker already recorded the failure in the store; swallow here so the
47+
// background task never crashes the process.
48+
}
49+
finally
50+
{
51+
_running.TryRemove(job.Id, out _);
52+
_tasks.TryRemove(job.Id, out _);
53+
cts.Dispose();
54+
}
55+
},
56+
CancellationToken.None);
57+
58+
_tasks[job.Id] = task;
59+
return job.Id;
60+
}
61+
62+
/// <inheritdoc />
63+
public Task<ReportJob?> GetAsync(string jobId, CancellationToken cancellationToken) =>
64+
_store.GetAsync(jobId, cancellationToken);
65+
66+
/// <inheritdoc />
67+
public Task<bool> CancelAsync(string jobId, CancellationToken cancellationToken)
68+
{
69+
if (_running.TryGetValue(jobId, out var cts))
70+
{
71+
cts.Cancel();
72+
return Task.FromResult(true);
73+
}
74+
75+
// Not running: either unknown or already finished — nothing to cancel.
76+
return Task.FromResult(false);
77+
}
78+
79+
/// <summary>
80+
/// Awaits a job's background task (test/shutdown helper). Returns immediately if the job is
81+
/// unknown or already finished.
82+
/// </summary>
83+
/// <param name="jobId">The job id.</param>
84+
public Task WaitForCompletionAsync(string jobId) =>
85+
_tasks.TryGetValue(jobId, out var task) ? task : Task.CompletedTask;
86+
87+
/// <summary>Cancels all running jobs and waits for their background tasks to unwind.</summary>
88+
public async ValueTask DisposeAsync()
89+
{
90+
foreach (var cts in _running.Values)
91+
{
92+
try { cts.Cancel(); }
93+
catch (ObjectDisposedException) { }
94+
}
95+
96+
try
97+
{
98+
await Task.WhenAll(_tasks.Values).ConfigureAwait(false);
99+
}
100+
catch (Exception)
101+
{
102+
// Best-effort drain on shutdown.
103+
}
104+
}
105+
}

0 commit comments

Comments
 (0)