30665688881

Committed 31 Jul 2026 09:11PM UTC coverage: 83.969% (+0.1%) from 83.82%

Build # 30665688881

Build Type

Pull #86

github

Committed by

web-flow

Commit Message

Merge 76c6a3cba into c8caa3f4e

Pull Request Pull Request #86: Add job cancellation APIs and dashboard action

Coverage Stats

103 of 116 new or added lines in 9 files covered. (88.79%)

2 existing lines in 1 file now uncovered.

8213 of 9781 relevant lines covered (83.97%)

2.7 hits per line

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

92.81

/src/Immediate.Jobs.Shared/JobBatches.cs

namespace Immediate.Jobs.Shared;

/// <summary>Creates atomic batches of typed generated jobs.</summary>
public interface IJobBatchScheduler
{
        /// <summary>Cancels every non-terminal member of a committed batch.</summary>
        /// <param name="handle">The committed batch to cancel.</param>
        /// <param name="cancellationToken">A token that can cancel the storage operation.</param>
        /// <returns>A value task that represents the asynchronous cancellation.</returns>
        ValueTask CancelAsync(BatchHandle handle, CancellationToken cancellationToken = default) =>
                throw new NotSupportedException("This scheduler does not support cancelling batches.");

        /// <summary>Begins an in-memory batch buffer.</summary>
        /// <returns>The new batch buffer.</returns>
        JobBatch Begin();

        /// <summary>Begins a follow-up batch whose root members wait for a prior batch.</summary>
        /// <param name="after">The batch that must reach a terminal state before the follow-up roots are released.</param>
        /// <param name="on">The parent-batch outcome that releases the follow-up roots.</param>
        /// <returns>The new follow-up batch buffer.</returns>
        JobBatch Begin(BatchHandle after, ContinuationTrigger on = ContinuationTrigger.Success);

        /// <summary>Runs a batch body and commits it when the body succeeds.</summary>
        /// <param name="body">The callback that adds jobs and dependencies to the batch.</param>
        /// <param name="cancellationToken">A token that can cancel the commit operation.</param>
        /// <returns>A handle for the committed batch.</returns>
        ValueTask<BatchHandle> RunAsync(
                Func<JobBatch, ValueTask> body,
                CancellationToken cancellationToken = default
        );
}

/// <summary>Default scoped atomic-batch scheduler.</summary>
/// <param name="storage">The storage provider used to persist batch graphs.</param>
/// <param name="timeProvider">The clock used to timestamp batches and jobs.</param>
/// <param name="idGenerator">The generator used to create batch and job identifiers.</param>
public sealed class JobBatchScheduler(
        IJobStorage storage,
        TimeProvider timeProvider,
        IIdGenerator idGenerator
) : IJobBatchScheduler
{
        /// <inheritdoc />
        public ValueTask CancelAsync(BatchHandle handle, CancellationToken cancellationToken = default)
        {
                ArgumentNullException.ThrowIfNull(handle);
                return JobStorageCapabilityGuards.RequireGraph(storage).CancelBatchAsync(handle.Id, cancellationToken);
        }

        /// <inheritdoc />
        public JobBatch Begin() =>
                new(
                        JobStorageCapabilityGuards.RequireGraph(storage),
                        timeProvider,
                        idGenerator,
                        after: null,
                        ContinuationTrigger.Success
                );

        /// <inheritdoc />
        public JobBatch Begin(BatchHandle after, ContinuationTrigger on = ContinuationTrigger.Success)
        {
                ArgumentNullException.ThrowIfNull(after);
                return new(
                        JobStorageCapabilityGuards.RequireGraph(storage),
                        timeProvider,
                        idGenerator,
                        after,
                        on
                );
        }

        /// <inheritdoc />
        public async ValueTask<BatchHandle> RunAsync(
                Func<JobBatch, ValueTask> body,
                CancellationToken cancellationToken = default
        )
        {
                ArgumentNullException.ThrowIfNull(body);
                await using var batch = Begin();
                await body(batch).ConfigureAwait(false);
                return await batch.CommitAsync(cancellationToken).ConfigureAwait(false);
        }
}

/// <summary>An in-progress atomic batch buffer created by <see cref="IJobBatchScheduler"/>.</summary>
public sealed class JobBatch : IAsyncDisposable
{
        private enum Lifecycle
        {
                Open,
                Committing,
                Finished,
                Disposed,
        }

        private readonly Lock _gate = new();
        private readonly List<JobRecord> _jobs = [];
        private readonly List<JobContinuationEdge> _edges = [];
        private readonly IJobGraphStorage _storage;
        private readonly TimeProvider _timeProvider;
        private readonly BatchHandle? _after;
        private readonly ContinuationTrigger _trigger;
        private Lifecycle _lifecycle;

        internal JobBatch(
                IJobGraphStorage storage,
                TimeProvider timeProvider,
                IIdGenerator idGenerator,
                BatchHandle? after,
                ContinuationTrigger trigger
        )
        {
                _storage = storage;
                _timeProvider = timeProvider;
                _after = after;
                _trigger = trigger;
                Id = idGenerator.CreateId(IdKind.Batch);
        }

        /// <summary>The client-generated batch identifier.</summary>
        /// <value>The identifier assigned to the batch.</value>
        public string Id { get; }

        internal JobHandle Add(JobRecord record, ReadOnlySpan<JobHandle> parents, ContinuationTrigger on)
        {
                lock (_gate)
                {
                        EnsureOpenCore();
                        if (parents.IsEmpty)
                        {
                                _jobs.Add(record with { BatchId = Id });
                                return new(record.Id, this);
                        }

                        var parentIds = new HashSet<string>(StringComparer.Ordinal);
                        foreach (var parent in parents)
                        {
                                if (string.IsNullOrWhiteSpace(parent.Id))
                                        throw new ImmediateJobException("Continuation parent handles must have a non-empty identifier.");
                                if (!ReferenceEquals(parent.Batch, this))
                                        throw new ImmediateJobException("Continuation handles must belong to the same open batch.");
                                if (!parentIds.Add(parent.Id))
                                        throw new ImmediateJobException($"Duplicate continuation parent '{parent.Id}'.");
                        }

                        _jobs.Add(record with
                        {
                                BatchId = Id,
                                State = JobState.AwaitingContinuation,
                                RemainingDependencies = parentIds.Count,
                        });
                        foreach (var parentId in parentIds)
                        {
                                _edges.Add(new()
                                {
                                        ChildJobId = record.Id,
                                        ParentJobId = parentId,
                                        Trigger = on,
                                });
                        }

                        return new(record.Id, this);
                }
        }

        /// <summary>Atomically persists the buffered jobs and dependencies.</summary>
        /// <param name="cancellationToken">A token that can cancel the commit operation.</param>
        /// <returns>A handle for the committed batch.</returns>
        public async ValueTask<BatchHandle> CommitAsync(CancellationToken cancellationToken = default)
        {
                JobBatchRecord record;
                IReadOnlyList<JobRecord> jobs;
                IReadOnlyList<JobContinuationEdge> edges;
                lock (_gate)
                {
                        EnsureOpenCore();
                        if (_jobs.Count == 0)
                                throw new ImmediateJobException("An atomic batch cannot be committed without jobs.");

                        if (_after is { } parentBatch)
                        {
                                var children = _jobs.Select(static job => job.Id).ToHashSet(StringComparer.Ordinal);
                                foreach (var edge in _edges)
                                        _ = children.Remove(edge.ChildJobId);

                                foreach (var childId in children)
                                {
                                        var index = _jobs.FindIndex(job => string.Equals(job.Id, childId, StringComparison.Ordinal));
                                        var job = _jobs[index];
                                        _jobs[index] = job with
                                        {
                                                State = JobState.AwaitingContinuation,
                                                RemainingDependencies = job.RemainingDependencies + 1,
                                        };
                                        _edges.Add(new()
                                        {
                                                ChildJobId = childId,
                                                ParentBatchId = parentBatch.Id,
                                                Trigger = _trigger,
                                        });
                                }
                        }

                        record = new JobBatchRecord
                        {
                                Id = Id,
                                CreatedAt = _timeProvider.GetUtcNow(),
                                TotalJobs = _jobs.Count,
                                PendingCount = _jobs.Count,
                                State = BatchState.Executing,
                        };
                        jobs = Array.AsReadOnly(_jobs.ToArray());
                        edges = Array.AsReadOnly(_edges.ToArray());
                        _lifecycle = Lifecycle.Committing;
                }

                try
                {
                        await _storage.EnqueueBatchAsync(record, jobs, edges, cancellationToken).ConfigureAwait(false);
                        return new(Id);
                }
                finally
                {
                        lock (_gate)
                        {
                                if (_lifecycle == Lifecycle.Committing)
                                        _lifecycle = Lifecycle.Finished;
                        }
                }
        }

        /// <inheritdoc />
        public ValueTask DisposeAsync()
        {
                lock (_gate)
                {
                        if (_lifecycle == Lifecycle.Disposed)
                                return ValueTask.CompletedTask;
                        _lifecycle = Lifecycle.Disposed;
                        _jobs.Clear();
                        _edges.Clear();
                }

                return ValueTask.CompletedTask;
        }

        internal void EnsureOpen()
        {
                lock (_gate)
                        EnsureOpenCore();
        }

        private void EnsureOpenCore()
        {
                if (_lifecycle != Lifecycle.Open)
                        throw new ImmediateJobException("A batch or one of its handles was used after commit or disposal.");
        }
}

/// <summary>Storage-backed implementation of the public monitoring services.</summary>
/// <param name="storage">The storage provider queried for job and batch status.</param>
/// <param name="definitions">The generated job definitions used to enrich monitoring results.</param>
public sealed class JobMonitor(IJobStorage storage, IEnumerable<JobDefinition> definitions) : IJobBatchMonitor, IJobMonitor
{
        /// <inheritdoc />
        public ValueTask<BatchStatus?> GetStatusAsync(string batchId, CancellationToken cancellationToken = default) =>
                JobStorageCapabilityGuards.RequireGraph(storage).GetBatchStatusAsync(batchId, cancellationToken);

        /// <inheritdoc />
        public ValueTask<IReadOnlyList<BatchMemberStatus>> QueryMembersAsync(
                string batchId,
                BatchMemberQuery query,
                CancellationToken cancellationToken = default
        ) => JobStorageCapabilityGuards.RequireGraph(storage).QueryBatchMembersAsync(batchId, query, cancellationToken);

        /// <inheritdoc />
        public ValueTask<BatchGraph?> GetGraphAsync(string batchId, CancellationToken cancellationToken = default) =>
                JobStorageCapabilityGuards.RequireGraph(storage).GetBatchGraphAsync(batchId, cancellationToken);

        /// <inheritdoc />
        public async ValueTask<JobStatus?> GetJobAsync(string jobId, CancellationToken cancellationToken = default)
        {
                var status = await storage.GetJobStatusAsync(jobId, cancellationToken).ConfigureAwait(false);
                if (status is null)
                        return null;
                var definition = definitions.FirstOrDefault(candidate =>
                        string.Equals(candidate.Name, status.JobName, StringComparison.Ordinal));
                return definition is null ? status : status with { MaxAttempts = definition.MaxAttempts };
        }
}

1	namespace Immediate.Jobs.Shared;
2
3	/// <summary>Creates atomic batches of typed generated jobs.</summary>
4	public interface IJobBatchScheduler
5	{
6	/// <summary>Cancels every non-terminal member of a committed batch.</summary>
7	/// <param name="handle">The committed batch to cancel.</param>
8	/// <param name="cancellationToken">A token that can cancel the storage operation.</param>
9	/// <returns>A value task that represents the asynchronous cancellation.</returns>
10	ValueTask CancelAsync(BatchHandle handle, CancellationToken cancellationToken = default) =>
NEW 11	throw new NotSupportedException("This scheduler does not support cancelling batches.");	×
12
13	/// <summary>Begins an in-memory batch buffer.</summary>
14	/// <returns>The new batch buffer.</returns>
15	JobBatch Begin();
16
17	/// <summary>Begins a follow-up batch whose root members wait for a prior batch.</summary>
18	/// <param name="after">The batch that must reach a terminal state before the follow-up roots are released.</param>
19	/// <param name="on">The parent-batch outcome that releases the follow-up roots.</param>
20	/// <returns>The new follow-up batch buffer.</returns>
21	JobBatch Begin(BatchHandle after, ContinuationTrigger on = ContinuationTrigger.Success);
22
23	/// <summary>Runs a batch body and commits it when the body succeeds.</summary>
24	/// <param name="body">The callback that adds jobs and dependencies to the batch.</param>
25	/// <param name="cancellationToken">A token that can cancel the commit operation.</param>
26	/// <returns>A handle for the committed batch.</returns>
27	ValueTask<BatchHandle> RunAsync(
28	Func<JobBatch, ValueTask> body,
29	CancellationToken cancellationToken = default
30	);
31	}
32
33	/// <summary>Default scoped atomic-batch scheduler.</summary>
34	/// <param name="storage">The storage provider used to persist batch graphs.</param>
35	/// <param name="timeProvider">The clock used to timestamp batches and jobs.</param>
36	/// <param name="idGenerator">The generator used to create batch and job identifiers.</param>
37	public sealed class JobBatchScheduler(	4✔
38	IJobStorage storage,	4✔
39	TimeProvider timeProvider,	4✔
40	IIdGenerator idGenerator	4✔
41	) : IJobBatchScheduler	4✔
42	{
43	/// <inheritdoc />
44	public ValueTask CancelAsync(BatchHandle handle, CancellationToken cancellationToken = default)
45	{
46	ArgumentNullException.ThrowIfNull(handle);	4✔
47	return JobStorageCapabilityGuards.RequireGraph(storage).CancelBatchAsync(handle.Id, cancellationToken);	4✔
48	}
49
50	/// <inheritdoc />
51	public JobBatch Begin() =>
52	new(	4✔
53	JobStorageCapabilityGuards.RequireGraph(storage),	4✔
54	timeProvider,	4✔
55	idGenerator,	4✔
56	after: null,	4✔
57	ContinuationTrigger.Success	4✔
58	);	4✔
59
60	/// <inheritdoc />
61	public JobBatch Begin(BatchHandle after, ContinuationTrigger on = ContinuationTrigger.Success)
62	{
63	ArgumentNullException.ThrowIfNull(after);	4✔
64	return new(	4✔
65	JobStorageCapabilityGuards.RequireGraph(storage),	4✔
66	timeProvider,	4✔
67	idGenerator,	4✔
68	after,	4✔
69	on	4✔
70	);	4✔
71	}
72
73	/// <inheritdoc />
74	public async ValueTask<BatchHandle> RunAsync(
75	Func<JobBatch, ValueTask> body,
76	CancellationToken cancellationToken = default
77	)
78	{
79	ArgumentNullException.ThrowIfNull(body);	×
80	await using var batch = Begin();	×
81	await body(batch).ConfigureAwait(false);	×
82	return await batch.CommitAsync(cancellationToken).ConfigureAwait(false);	×
83	}
84	}
85
86	/// <summary>An in-progress atomic batch buffer created by <see cref="IJobBatchScheduler"/>.</summary>
87	public sealed class JobBatch : IAsyncDisposable
88	{
89	private enum Lifecycle
90	{
91	Open,
92	Committing,
93	Finished,
94	Disposed,
95	}
96
97	private readonly Lock _gate = new();	4✔
98	private readonly List<JobRecord> _jobs = [];	4✔
99	private readonly List<JobContinuationEdge> _edges = [];	4✔
100	private readonly IJobGraphStorage _storage;
101	private readonly TimeProvider _timeProvider;
102	private readonly BatchHandle? _after;
103	private readonly ContinuationTrigger _trigger;
104	private Lifecycle _lifecycle;
105
106	internal JobBatch(	4✔
107	IJobGraphStorage storage,	4✔
108	TimeProvider timeProvider,	4✔
109	IIdGenerator idGenerator,	4✔
110	BatchHandle? after,	4✔
111	ContinuationTrigger trigger	4✔
112	)	4✔
113	{
114	_storage = storage;	4✔
115	_timeProvider = timeProvider;	4✔
116	_after = after;	4✔
117	_trigger = trigger;	4✔
118	Id = idGenerator.CreateId(IdKind.Batch);	4✔
119	}	4✔
120
121	/// <summary>The client-generated batch identifier.</summary>
122	/// <value>The identifier assigned to the batch.</value>
123	public string Id { get; }
124
125	internal JobHandle Add(JobRecord record, ReadOnlySpan<JobHandle> parents, ContinuationTrigger on)
126	{	4✔
127	lock (_gate)
128	{
129	EnsureOpenCore();	4✔
130	if (parents.IsEmpty)	4✔
131	{
132	_jobs.Add(record with { BatchId = Id });	4✔
133	return new(record.Id, this);	4✔
134	}
135
136	var parentIds = new HashSet<string>(StringComparer.Ordinal);	4✔
137	foreach (var parent in parents)	4✔
138	{
139	if (string.IsNullOrWhiteSpace(parent.Id))	4✔
140	throw new ImmediateJobException("Continuation parent handles must have a non-empty identifier.");	×
141	if (!ReferenceEquals(parent.Batch, this))	4✔
142	throw new ImmediateJobException("Continuation handles must belong to the same open batch.");	×
143	if (!parentIds.Add(parent.Id))	4✔
144	throw new ImmediateJobException($"Duplicate continuation parent '{parent.Id}'.");	×
145	}
146
147	_jobs.Add(record with	4✔
148	{	4✔
149	BatchId = Id,	4✔
150	State = JobState.AwaitingContinuation,	4✔
151	RemainingDependencies = parentIds.Count,	4✔
152	});	4✔
153	foreach (var parentId in parentIds)	4✔
154	{
155	_edges.Add(new()	4✔
156	{	4✔
157	ChildJobId = record.Id,	4✔
158	ParentJobId = parentId,	4✔
159	Trigger = on,	4✔
160	});	4✔
161	}
162
163	return new(record.Id, this);	4✔
164	}
165	}	4✔
166
167	/// <summary>Atomically persists the buffered jobs and dependencies.</summary>
168	/// <param name="cancellationToken">A token that can cancel the commit operation.</param>
169	/// <returns>A handle for the committed batch.</returns>
170	public async ValueTask<BatchHandle> CommitAsync(CancellationToken cancellationToken = default)
171	{	4✔
172	JobBatchRecord record;
173	IReadOnlyList<JobRecord> jobs;
174	IReadOnlyList<JobContinuationEdge> edges;
175	lock (_gate)
176	{
177	EnsureOpenCore();	4✔
178	if (_jobs.Count == 0)	4✔
179	throw new ImmediateJobException("An atomic batch cannot be committed without jobs.");	×
180
181	if (_after is { } parentBatch)	4✔
182	{
183	var children = _jobs.Select(static job => job.Id).ToHashSet(StringComparer.Ordinal);	4✔
184	foreach (var edge in _edges)	4✔
185	_ = children.Remove(edge.ChildJobId);	4✔
186
187	foreach (var childId in children)	4✔
188	{
189	var index = _jobs.FindIndex(job => string.Equals(job.Id, childId, StringComparison.Ordinal));	4✔
190	var job = _jobs[index];	4✔
191	_jobs[index] = job with	4✔
192	{	4✔
193	State = JobState.AwaitingContinuation,	4✔
194	RemainingDependencies = job.RemainingDependencies + 1,	4✔
195	};	4✔
196	_edges.Add(new()	4✔
197	{	4✔
198	ChildJobId = childId,	4✔
199	ParentBatchId = parentBatch.Id,	4✔
200	Trigger = _trigger,	4✔
201	});	4✔
202	}
203	}
204
205	record = new JobBatchRecord	4✔
206	{	4✔
207	Id = Id,	4✔
208	CreatedAt = _timeProvider.GetUtcNow(),	4✔
209	TotalJobs = _jobs.Count,	4✔
210	PendingCount = _jobs.Count,	4✔
211	State = BatchState.Executing,	4✔
212	};	4✔
213	jobs = Array.AsReadOnly(_jobs.ToArray());	4✔
214	edges = Array.AsReadOnly(_edges.ToArray());	4✔
215	_lifecycle = Lifecycle.Committing;	4✔
216	}	4✔
217
218	try
219	{
220	await _storage.EnqueueBatchAsync(record, jobs, edges, cancellationToken).ConfigureAwait(false);	4✔
221	return new(Id);	4✔
222	}
223	finally
224	{	4✔
225	lock (_gate)
226	{
227	if (_lifecycle == Lifecycle.Committing)	4✔
228	_lifecycle = Lifecycle.Finished;	4✔
229	}	4✔
230	}	1✔
231	}	4✔
232
233	/// <inheritdoc />
234	public ValueTask DisposeAsync()
235	{	4✔
236	lock (_gate)
237	{
238	if (_lifecycle == Lifecycle.Disposed)	4✔
239	return ValueTask.CompletedTask;	4✔
240	_lifecycle = Lifecycle.Disposed;	4✔
241	_jobs.Clear();	4✔
242	_edges.Clear();	4✔
243	}	4✔
244
245	return ValueTask.CompletedTask;	4✔
246	}	4✔
247
248	internal void EnsureOpen()
249	{	4✔
250	lock (_gate)
251	EnsureOpenCore();	4✔
252	}	4✔
253
254	private void EnsureOpenCore()
255	{
256	if (_lifecycle != Lifecycle.Open)	4✔
257	throw new ImmediateJobException("A batch or one of its handles was used after commit or disposal.");	4✔
258	}	4✔
259	}
260
261	/// <summary>Storage-backed implementation of the public monitoring services.</summary>
262	/// <param name="storage">The storage provider queried for job and batch status.</param>
263	/// <param name="definitions">The generated job definitions used to enrich monitoring results.</param>
264	public sealed class JobMonitor(IJobStorage storage, IEnumerable<JobDefinition> definitions) : IJobBatchMonitor, IJobMonitor	4✔
265	{
266	/// <inheritdoc />
267	public ValueTask<BatchStatus?> GetStatusAsync(string batchId, CancellationToken cancellationToken = default) =>
268	JobStorageCapabilityGuards.RequireGraph(storage).GetBatchStatusAsync(batchId, cancellationToken);	4✔
269
270	/// <inheritdoc />
271	public ValueTask<IReadOnlyList<BatchMemberStatus>> QueryMembersAsync(
272	string batchId,
273	BatchMemberQuery query,
274	CancellationToken cancellationToken = default
275	) => JobStorageCapabilityGuards.RequireGraph(storage).QueryBatchMembersAsync(batchId, query, cancellationToken);	4✔
276
277	/// <inheritdoc />
278	public ValueTask<BatchGraph?> GetGraphAsync(string batchId, CancellationToken cancellationToken = default) =>
279	JobStorageCapabilityGuards.RequireGraph(storage).GetBatchGraphAsync(batchId, cancellationToken);	4✔
280
281	/// <inheritdoc />
282	public async ValueTask<JobStatus?> GetJobAsync(string jobId, CancellationToken cancellationToken = default)
283	{
284	var status = await storage.GetJobStatusAsync(jobId, cancellationToken).ConfigureAwait(false);	4✔
285	if (status is null)	4✔
286	return null;	×
287	var definition = definitions.FirstOrDefault(candidate =>	4✔
288	string.Equals(candidate.Name, status.JobName, StringComparison.Ordinal));	4✔
289	return definition is null ? status : status with { MaxAttempts = definition.MaxAttempts };	4✔
290	}	3✔
291	}

ImmediatePlatform / Immediate.Jobs / 30665688881

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