30304771671

Committed 27 Jul 2026 08:56PM UTC coverage: 65.995%. First build

Build # 30304771671

Build Type

Pull #33

github

Committed by

web-flow

Commit Message

Merge 46c3047f1 into e9a147742

Pull Request Pull Request #33: Improve Storage API

Coverage Stats

1521 of 2481 new or added lines in 28 files covered. (61.31%)

4326 of 6555 relevant lines covered (66.0%)

2.22 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

47.2

/src/Immediate.Jobs.Testing/JobTestHarness.cs

using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Time.Testing;

namespace Immediate.Jobs.Testing;

/// <summary>
/// Hosts Immediate.Jobs with in-memory storage and a controllable clock without starting background threads.
/// </summary>
public sealed class JobTestHarness : IAsyncDisposable, IDisposable
{
        private readonly ServiceProvider _serviceProvider;
        private readonly IJobGraphStorage _graphStorage;
        private bool _disposed;

        /// <summary>Creates a harness at the Unix epoch.</summary>
        /// <param name="configureServices">
        /// Registers generated job definitions and the services used by their handlers. Calling the generated
        /// <c>AddImmediateJobs</c> method here is supported; the harness clock and in-memory provider remain authoritative.
        /// </param>
        public JobTestHarness(Action<IServiceCollection>? configureServices = null)
                : this(DateTimeOffset.UnixEpoch, configureServices)
        {
        }

        /// <summary>Creates a harness at a specified UTC instant.</summary>
        public JobTestHarness(
                DateTimeOffset start,
                Action<IServiceCollection>? configureServices = null
        )
        {
                TimeProvider = new(start);
                var services = new ServiceCollection();
                _ = services.AddSingleton<TimeProvider>(TimeProvider);
                _ = services.AddSingleton(TimeProvider);
                _ = services.AddSingleton(typeof(ILogger<>), typeof(NullLogger<>));
                _ = services.AddImmediateJobsCore(options =>
                {
                        _ = options.UseInMemory();
                        options.MaxParallelJobs = 1;
                });
                configureServices?.Invoke(services);
                _serviceProvider = services.BuildServiceProvider(new ServiceProviderOptions
                {
                        ValidateScopes = true,
                        ValidateOnBuild = true,
                });

                Services = _serviceProvider;
                Storage = _serviceProvider.GetRequiredService<IJobStorage>();
                _graphStorage = Storage as IJobGraphStorage
                        ?? throw new NotSupportedException(
                                "Batches & continuations require a graph-capable storage provider (a SQL database). " +
                                "The configured provider implements the queue capability only."
                        );
                Batches = new JobBatchScheduler(
                        Storage,
                        TimeProvider,
                        _serviceProvider.GetRequiredService<IIdGenerator>()
                );
                Scheduler = _serviceProvider.GetRequiredService<JobSchedulerService>();
        }

        /// <summary>The controllable clock used by schedulers, storage, retries, and cron materialization.</summary>
        public FakeTimeProvider TimeProvider { get; }

        /// <summary>The harness service provider.</summary>
        public IServiceProvider Services { get; }

        /// <summary>The in-memory durable-state abstraction.</summary>
        public IJobStorage Storage { get; }

        /// <summary>Builds atomic batches against the harness storage and fake clock.</summary>
        public IJobBatchScheduler Batches { get; }

        /// <summary>The production scheduler runner hosted by the harness.</summary>
        public JobSchedulerService Scheduler { get; }

        /// <summary>Runs every invocation currently due and returns when the due queue is empty.</summary>
        public ValueTask DrainAsync(CancellationToken cancellationToken = default) =>
                Scheduler.DrainAsync(cancellationToken);

        /// <summary>Advances fake time and then runs every invocation that became due.</summary>
        public async ValueTask AdvanceTimeAndDrainAsync(
                TimeSpan amount,
                CancellationToken cancellationToken = default
        )
        {
                if (amount < TimeSpan.Zero)
                        throw new ArgumentOutOfRangeException(nameof(amount), "Fake time cannot move backwards.");

                TimeProvider.Advance(amount);
                await DrainAsync(cancellationToken).ConfigureAwait(false);
        }

        /// <summary>Advances fake time to an absolute instant and drains newly due work.</summary>
        public ValueTask AdvanceTimeAndDrainAsync(
                DateTimeOffset instant,
                CancellationToken cancellationToken = default
        )
        {
                var amount = instant - TimeProvider.GetUtcNow();
                if (amount < TimeSpan.Zero)
                        throw new ArgumentOutOfRangeException(nameof(instant), "Fake time cannot move backwards.");
                return AdvanceTimeAndDrainAsync(amount, cancellationToken);
        }

        /// <summary>Returns persisted jobs matching a monitoring query.</summary>
        public ValueTask<IReadOnlyList<JobRecord>> QueryJobsAsync(
                JobQuery? query = null,
                CancellationToken cancellationToken = default
        ) => Storage.QueryJobsAsync(query ?? new() { Take = 1000 }, cancellationToken);

        /// <summary>Finds an invocation by identifier, or throws a test assertion exception.</summary>
        public async ValueTask<JobRecord> GetJobAsync(string jobId, CancellationToken cancellationToken = default)
        {
                var jobs = await Storage.QueryJobsAsync(new() { Take = 1000 }, cancellationToken).ConfigureAwait(false);
                return jobs.FirstOrDefault(job => job.Id == jobId)
                        ?? throw new JobTestAssertionException($"Expected job '{jobId}' to have been enqueued, but it was not found.");
        }

        /// <summary>Finds an invocation returned by a typed scheduler call.</summary>
        public ValueTask<JobRecord> GetJobAsync(JobHandle job, CancellationToken cancellationToken = default) =>
                GetJobAsync(job.Id, cancellationToken);

        /// <summary>Asserts and deserializes the invocation returned by a typed scheduler call.</summary>
        public async ValueTask<EnqueuedJob<TPayload>> AssertEnqueuedAsync<TPayload>(
                string jobId,
                JobState? expectedState = null,
                CancellationToken cancellationToken = default
        )
        {
                var job = await GetJobAsync(jobId, cancellationToken).ConfigureAwait(false);
                if (expectedState is { } state && job.State != state)
                {
                        throw new JobTestAssertionException(
                                $"Expected job '{jobId}' to be {state}, but it was {job.State}."
                        );
                }

                TPayload payload;
                try
                {
                        payload = _serviceProvider.GetRequiredService<IJobSerializer>().Deserialize<TPayload>(job.Payload);
                }
                catch (Exception exception)
                {
                        throw new JobTestAssertionException(
                                $"Job '{jobId}' did not contain a valid {typeof(TPayload).FullName} payload: {exception.Message}"
                        );
                }

                return new(job, payload);
        }

        /// <summary>Asserts and deserializes the invocation returned by a typed scheduler call.</summary>
        public ValueTask<EnqueuedJob<TPayload>> AssertEnqueuedAsync<TPayload>(
                JobHandle job,
                JobState? expectedState = null,
                CancellationToken cancellationToken = default
        ) => AssertEnqueuedAsync<TPayload>(job.Id, expectedState, cancellationToken);

        /// <summary>Asserts that a committed batch and exactly the expected number of members are visible together.</summary>
        public async ValueTask AssertBatchCommittedAtomicallyAsync(
                BatchHandle batch,
                int expectedMembers,
                CancellationToken cancellationToken = default
        )
        {
                ArgumentNullException.ThrowIfNull(batch);
                var status = await _graphStorage.GetBatchStatusAsync(batch.Id, cancellationToken).ConfigureAwait(false)
                        ?? throw new JobTestAssertionException($"Expected batch '{batch.Id}' to be committed, but it was not found.");
                var members = await _graphStorage.QueryBatchMembersAsync(
                        batch.Id,
                        new() { Take = Math.Max(1, expectedMembers + 1) },
                        cancellationToken
                ).ConfigureAwait(false);
                if (status.Total != expectedMembers || members.Count != expectedMembers)
                {
                        throw new JobTestAssertionException(
                                $"Expected batch '{batch.Id}' to contain {expectedMembers} jobs, but its header reports {status.Total} and {members.Count} members were found."
                        );
                }
        }

        /// <summary>Asserts that the child has a persisted dependency on the supplied parent.</summary>
        public async ValueTask AssertContinuationReleasedAfterAsync(
                JobHandle parent,
                JobHandle child,
                CancellationToken cancellationToken = default
        )
        {
                var childStatus = await Storage.GetJobStatusAsync(child.Id, cancellationToken).ConfigureAwait(false)
                        ?? throw new JobTestAssertionException($"Expected continuation '{child.Id}', but it was not found.");
                if (!childStatus.DependsOn.Any(edge => edge.ParentJobId == parent.Id))
                {
                        throw new JobTestAssertionException(
                                $"Expected job '{child.Id}' to depend on '{parent.Id}', but no such edge was persisted."
                        );
                }

                if (childStatus.State == JobState.AwaitingContinuation)
                        throw new JobTestAssertionException($"Expected continuation '{child.Id}' to be released, but it is still waiting.");
        }

        /// <summary>Asserts that every supplied invocation was cancelled by a dependency cascade.</summary>
        public async ValueTask AssertCascadeCancelledAsync(
                IReadOnlyCollection<JobHandle> subtree,
                CancellationToken cancellationToken = default
        )
        {
                ArgumentNullException.ThrowIfNull(subtree);
                foreach (var handle in subtree)
                {
                        var job = await GetJobAsync(handle, cancellationToken).ConfigureAwait(false);
                        if (job.State != JobState.Cancelled)
                                throw new JobTestAssertionException($"Expected job '{handle.Id}' to be cascade-cancelled, but it was {job.State}.");
                }
        }

        /// <summary>Runs one generated invoker, including its compile-time behavior pipeline, outside durable state.</summary>
        public async ValueTask RunThroughPipelineAsync<TPayload>(
                JobDefinition definition,
                TPayload payload,
                CancellationToken cancellationToken = default
        )
        {
                ArgumentNullException.ThrowIfNull(definition);
                var now = TimeProvider.GetUtcNow();
                var record = new JobRecord
                {
                        Id = _serviceProvider.GetRequiredService<IIdGenerator>().CreateId(IdKind.Job),
                        JobName = definition.Name,
                        QueueName = definition.Queue.Name,
                        Payload = _serviceProvider.GetRequiredService<IJobSerializer>().Serialize(payload),
                        State = JobState.Active,
                        DueAt = now,
                        CreatedAt = now,
                        Attempt = 1,
                };
                await using var scope = _serviceProvider.CreateAsyncScope();
                await definition.Invoker.InvokeAsync(
                        scope.ServiceProvider,
                        new(record, definition, cancellationToken)
                ).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public void Dispose()
        {
                if (_disposed)
                        return;
                _disposed = true;
                _serviceProvider.Dispose();
        }

        /// <inheritdoc />
        public async ValueTask DisposeAsync()
        {
                if (_disposed)
                        return;
                _disposed = true;
                await _serviceProvider.DisposeAsync().ConfigureAwait(false);
        }
}

/// <summary>A durable invocation paired with its strongly typed deserialized payload.</summary>
public sealed record EnqueuedJob<TPayload>(JobRecord Record, TPayload Payload);

1	using Microsoft.Extensions.DependencyInjection;
2	using Microsoft.Extensions.Logging;
3	using Microsoft.Extensions.Logging.Abstractions;
4	using Microsoft.Extensions.Time.Testing;
5
6	namespace Immediate.Jobs.Testing;
7
8	/// <summary>
9	/// Hosts Immediate.Jobs with in-memory storage and a controllable clock without starting background threads.
10	/// </summary>
11	public sealed class JobTestHarness : IAsyncDisposable, IDisposable
12	{
13	private readonly ServiceProvider _serviceProvider;
14	private readonly IJobGraphStorage _graphStorage;
15	private bool _disposed;
16
17	/// <summary>Creates a harness at the Unix epoch.</summary>
18	/// <param name="configureServices">
19	/// Registers generated job definitions and the services used by their handlers. Calling the generated
20	/// <c>AddImmediateJobs</c> method here is supported; the harness clock and in-memory provider remain authoritative.
21	/// </param>
22	public JobTestHarness(Action<IServiceCollection>? configureServices = null)
23	: this(DateTimeOffset.UnixEpoch, configureServices)	4✔
24	{
25	}	4✔
26
27	/// <summary>Creates a harness at a specified UTC instant.</summary>
28	public JobTestHarness(	4✔
29	DateTimeOffset start,	4✔
30	Action<IServiceCollection>? configureServices = null	4✔
31	)	4✔
32	{
33	TimeProvider = new(start);	4✔
34	var services = new ServiceCollection();	4✔
35	_ = services.AddSingleton<TimeProvider>(TimeProvider);	4✔
36	_ = services.AddSingleton(TimeProvider);	4✔
37	_ = services.AddSingleton(typeof(ILogger<>), typeof(NullLogger<>));	4✔
38	_ = services.AddImmediateJobsCore(options =>	4✔
39	{	4✔
40	_ = options.UseInMemory();	4✔
41	options.MaxParallelJobs = 1;	4✔
42	});	4✔
43	configureServices?.Invoke(services);	4✔
44	_serviceProvider = services.BuildServiceProvider(new ServiceProviderOptions	4✔
45	{	4✔
46	ValidateScopes = true,	4✔
47	ValidateOnBuild = true,	4✔
48	});	4✔
49
50	Services = _serviceProvider;	4✔
51	Storage = _serviceProvider.GetRequiredService<IJobStorage>();	4✔
52	_graphStorage = Storage as IJobGraphStorage	4✔
53	?? throw new NotSupportedException(	4✔
54	"Batches & continuations require a graph-capable storage provider (a SQL database). " +	4✔
55	"The configured provider implements the queue capability only."	4✔
56	);	4✔
57	Batches = new JobBatchScheduler(	4✔
58	Storage,	4✔
59	TimeProvider,	4✔
60	_serviceProvider.GetRequiredService<IIdGenerator>()	4✔
61	);	4✔
62	Scheduler = _serviceProvider.GetRequiredService<JobSchedulerService>();	4✔
63	}	4✔
64
65	/// <summary>The controllable clock used by schedulers, storage, retries, and cron materialization.</summary>
66	public FakeTimeProvider TimeProvider { get; }
67
68	/// <summary>The harness service provider.</summary>
69	public IServiceProvider Services { get; }
70
71	/// <summary>The in-memory durable-state abstraction.</summary>
72	public IJobStorage Storage { get; }
73
74	/// <summary>Builds atomic batches against the harness storage and fake clock.</summary>
75	public IJobBatchScheduler Batches { get; }
76
77	/// <summary>The production scheduler runner hosted by the harness.</summary>
78	public JobSchedulerService Scheduler { get; }
79
80	/// <summary>Runs every invocation currently due and returns when the due queue is empty.</summary>
81	public ValueTask DrainAsync(CancellationToken cancellationToken = default) =>
82	Scheduler.DrainAsync(cancellationToken);	4✔
83
84	/// <summary>Advances fake time and then runs every invocation that became due.</summary>
85	public async ValueTask AdvanceTimeAndDrainAsync(
86	TimeSpan amount,
87	CancellationToken cancellationToken = default
88	)
89	{
90	if (amount < TimeSpan.Zero)	4✔
91	throw new ArgumentOutOfRangeException(nameof(amount), "Fake time cannot move backwards.");	×
92
93	TimeProvider.Advance(amount);	4✔
94	await DrainAsync(cancellationToken).ConfigureAwait(false);	4✔
95	}	4✔
96
97	/// <summary>Advances fake time to an absolute instant and drains newly due work.</summary>
98	public ValueTask AdvanceTimeAndDrainAsync(
99	DateTimeOffset instant,
100	CancellationToken cancellationToken = default
101	)
102	{
103	var amount = instant - TimeProvider.GetUtcNow();	×
104	if (amount < TimeSpan.Zero)	×
105	throw new ArgumentOutOfRangeException(nameof(instant), "Fake time cannot move backwards.");	×
106	return AdvanceTimeAndDrainAsync(amount, cancellationToken);	×
107	}
108
109	/// <summary>Returns persisted jobs matching a monitoring query.</summary>
110	public ValueTask<IReadOnlyList<JobRecord>> QueryJobsAsync(
111	JobQuery? query = null,
112	CancellationToken cancellationToken = default
113	) => Storage.QueryJobsAsync(query ?? new() { Take = 1000 }, cancellationToken);	4✔
114
115	/// <summary>Finds an invocation by identifier, or throws a test assertion exception.</summary>
116	public async ValueTask<JobRecord> GetJobAsync(string jobId, CancellationToken cancellationToken = default)
117	{
118	var jobs = await Storage.QueryJobsAsync(new() { Take = 1000 }, cancellationToken).ConfigureAwait(false);	4✔
119	return jobs.FirstOrDefault(job => job.Id == jobId)	4✔
120	?? throw new JobTestAssertionException($"Expected job '{jobId}' to have been enqueued, but it was not found.");	4✔
121	}	4✔
122
123	/// <summary>Finds an invocation returned by a typed scheduler call.</summary>
124	public ValueTask<JobRecord> GetJobAsync(JobHandle job, CancellationToken cancellationToken = default) =>
125	GetJobAsync(job.Id, cancellationToken);	4✔
126
127	/// <summary>Asserts and deserializes the invocation returned by a typed scheduler call.</summary>
128	public async ValueTask<EnqueuedJob<TPayload>> AssertEnqueuedAsync<TPayload>(
129	string jobId,
130	JobState? expectedState = null,
131	CancellationToken cancellationToken = default
132	)
133	{
134	var job = await GetJobAsync(jobId, cancellationToken).ConfigureAwait(false);	4✔
135	if (expectedState is { } state && job.State != state)	4✔
136	{
137	throw new JobTestAssertionException(	×
138	$"Expected job '{jobId}' to be {state}, but it was {job.State}."	×
139	);	×
140	}
141
142	TPayload payload;
143	try
144	{
145	payload = _serviceProvider.GetRequiredService<IJobSerializer>().Deserialize<TPayload>(job.Payload);	4✔
146	}	4✔
147	catch (Exception exception)	×
148	{
149	throw new JobTestAssertionException(	×
150	$"Job '{jobId}' did not contain a valid {typeof(TPayload).FullName} payload: {exception.Message}"	×
151	);	×
152	}
153
154	return new(job, payload);	4✔
155	}	4✔
156
157	/// <summary>Asserts and deserializes the invocation returned by a typed scheduler call.</summary>
158	public ValueTask<EnqueuedJob<TPayload>> AssertEnqueuedAsync<TPayload>(
159	JobHandle job,
160	JobState? expectedState = null,
161	CancellationToken cancellationToken = default
162	) => AssertEnqueuedAsync<TPayload>(job.Id, expectedState, cancellationToken);	4✔
163
164	/// <summary>Asserts that a committed batch and exactly the expected number of members are visible together.</summary>
165	public async ValueTask AssertBatchCommittedAtomicallyAsync(
166	BatchHandle batch,
167	int expectedMembers,
168	CancellationToken cancellationToken = default
169	)
170	{
171	ArgumentNullException.ThrowIfNull(batch);	×
NEW 172	var status = await _graphStorage.GetBatchStatusAsync(batch.Id, cancellationToken).ConfigureAwait(false)	×
173	?? throw new JobTestAssertionException($"Expected batch '{batch.Id}' to be committed, but it was not found.");	×
NEW 174	var members = await _graphStorage.QueryBatchMembersAsync(	×
175	batch.Id,	×
176	new() { Take = Math.Max(1, expectedMembers + 1) },	×
177	cancellationToken	×
178	).ConfigureAwait(false);	×
179	if (status.Total != expectedMembers \|\| members.Count != expectedMembers)	×
180	{
181	throw new JobTestAssertionException(	×
182	$"Expected batch '{batch.Id}' to contain {expectedMembers} jobs, but its header reports {status.Total} and {members.Count} members were found."	×
183	);	×
184	}
185	}	×
186
187	/// <summary>Asserts that the child has a persisted dependency on the supplied parent.</summary>
188	public async ValueTask AssertContinuationReleasedAfterAsync(
189	JobHandle parent,
190	JobHandle child,
191	CancellationToken cancellationToken = default
192	)
193	{
194	var childStatus = await Storage.GetJobStatusAsync(child.Id, cancellationToken).ConfigureAwait(false)	×
195	?? throw new JobTestAssertionException($"Expected continuation '{child.Id}', but it was not found.");	×
196	if (!childStatus.DependsOn.Any(edge => edge.ParentJobId == parent.Id))	×
197	{
198	throw new JobTestAssertionException(	×
199	$"Expected job '{child.Id}' to depend on '{parent.Id}', but no such edge was persisted."	×
200	);	×
201	}
202
203	if (childStatus.State == JobState.AwaitingContinuation)	×
204	throw new JobTestAssertionException($"Expected continuation '{child.Id}' to be released, but it is still waiting.");	×
205	}	×
206
207	/// <summary>Asserts that every supplied invocation was cancelled by a dependency cascade.</summary>
208	public async ValueTask AssertCascadeCancelledAsync(
209	IReadOnlyCollection<JobHandle> subtree,
210	CancellationToken cancellationToken = default
211	)
212	{
213	ArgumentNullException.ThrowIfNull(subtree);	×
214	foreach (var handle in subtree)	×
215	{
216	var job = await GetJobAsync(handle, cancellationToken).ConfigureAwait(false);	×
217	if (job.State != JobState.Cancelled)	×
218	throw new JobTestAssertionException($"Expected job '{handle.Id}' to be cascade-cancelled, but it was {job.State}.");	×
219	}	×
220	}	×
221
222	/// <summary>Runs one generated invoker, including its compile-time behavior pipeline, outside durable state.</summary>
223	public async ValueTask RunThroughPipelineAsync<TPayload>(
224	JobDefinition definition,
225	TPayload payload,
226	CancellationToken cancellationToken = default
227	)
228	{
229	ArgumentNullException.ThrowIfNull(definition);	×
230	var now = TimeProvider.GetUtcNow();	×
231	var record = new JobRecord	×
232	{	×
NEW 233	Id = _serviceProvider.GetRequiredService<IIdGenerator>().CreateId(IdKind.Job),	×
234	JobName = definition.Name,	×
235	QueueName = definition.Queue.Name,	×
236	Payload = _serviceProvider.GetRequiredService<IJobSerializer>().Serialize(payload),	×
237	State = JobState.Active,	×
238	DueAt = now,	×
239	CreatedAt = now,	×
240	Attempt = 1,	×
241	};	×
242	await using var scope = _serviceProvider.CreateAsyncScope();	×
243	await definition.Invoker.InvokeAsync(	×
244	scope.ServiceProvider,	×
245	new(record, definition, cancellationToken)	×
246	).ConfigureAwait(false);	×
247	}	×
248
249	/// <inheritdoc />
250	public void Dispose()
251	{
252	if (_disposed)	×
253	return;	×
254	_disposed = true;	×
255	_serviceProvider.Dispose();	×
256	}	×
257
258	/// <inheritdoc />
259	public async ValueTask DisposeAsync()
260	{
261	if (_disposed)	4✔
262	return;	×
263	_disposed = true;	4✔
264	await _serviceProvider.DisposeAsync().ConfigureAwait(false);	4✔
265	}	4✔
266	}
267
268	/// <summary>A durable invocation paired with its strongly typed deserialized payload.</summary>
269	public sealed record EnqueuedJob<TPayload>(JobRecord Record, TPayload Payload);	4✔

ImmediatePlatform / Immediate.Jobs / 30304771671

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous