30302568744

Committed 27 Jul 2026 08:26PM UTC coverage: 66.69%. First build

Build # 30302568744

Build Type

Pull #26

github

Committed by

web-flow

Commit Message

Merge 4a231468a into eac5c1044

Pull Request Pull Request #26: Add batches and continuation workflows

Coverage Stats

1239 of 1810 new or added lines in 17 files covered. (68.45%)

2855 of 4281 relevant lines covered (66.69%)

2.66 hits per line

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

43.1

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

ImmediatePlatform / Immediate.Jobs / 30302568744

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