38501404688

Committed 14 Oct 2026 02:30AM UTC coverage: 91.206% (+0.4%) from 90.78%

Build # 38501404688

Build Type

push

github

Committed by

web-flow

Commit Message

Resolve job queues from definitions at query time (#219)

* Resolve job queues from persisted definitions at query time

* Expose job record conversion on JobSubmission

Coverage Stats

1116 of 1133 new or added lines in 54 files covered. (98.5%)

14 existing lines in 4 files now uncovered.

14821 of 16250 relevant lines covered (91.21%)

2.95 hits per line

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

94.87

/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.Enqueue.cs

using Immediate.Jobs.Shared.Apis;
using System.Data;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;

namespace Immediate.Jobs.EntityFrameworkCore;

internal sealed partial class EntityFrameworkCoreJobStorage<TContext>
        where TContext : DbContext
{
        /// <inheritdoc />
        public async ValueTask EnqueueAsync(JobSubmission job, CancellationToken cancellationToken = default)
        {
                EnqueueAsyncCalled(job.JobHandle);
                cancellationToken.ThrowIfCancellationRequested();
                await TaskScheduler.Yield();

                await RetryConcurrencyAsync(
                        operationCancellationToken => EnqueueCoreAsync(job, operationCancellationToken),
                        cancellationToken
                );
        }

        /// <inheritdoc />
        public async ValueTask EnqueueContinuationAsync(
                JobSubmission job,
                IReadOnlyList<JobContinuationEdge> edges,
                CancellationToken cancellationToken = default
        )
        {
                EnqueueContinuationAsyncCalled(job.JobHandle, edges.Count);
                cancellationToken.ThrowIfCancellationRequested();
                await TaskScheduler.Yield();

                await ExecuteGraphInsertAsync(batch: null, [job], edges, cancellationToken);
        }

        /// <inheritdoc />
        public async ValueTask EnqueueBatchAsync(
                BatchRecord batch,
                IReadOnlyList<JobSubmission> jobs,
                IReadOnlyList<JobContinuationEdge> edges,
                CancellationToken cancellationToken = default
        )
        {
                EnqueueBatchAsyncCalled(batch.BatchHandle, jobs.Count, edges.Count);
                cancellationToken.ThrowIfCancellationRequested();
                await TaskScheduler.Yield();

                await ExecuteGraphInsertAsync(batch, jobs, edges, cancellationToken);
        }

        /// <inheritdoc />
        public async ValueTask AddBatchJobAsync(
                JobHandle currentJobHandle,
                int executionNumber,
                JobSubmission job,
                ContinuationOptions options,
                CancellationToken cancellationToken = default
        )
        {
                AddBatchJobAsyncCalled(job.JobHandle, executionNumber);
                cancellationToken.ThrowIfCancellationRequested();
                await TaskScheduler.Yield();

                await RetryConcurrencyAsync(
                        operationCancellationToken => AddBatchJobCoreAsync(
                                currentJobHandle,
                                executionNumber,
                                job,
                                options,
                                operationCancellationToken
                        ),
                        cancellationToken
                );
        }

        private async Task EnqueueCoreAsync(JobSubmission job, CancellationToken cancellationToken)
        {
                cancellationToken.ThrowIfCancellationRequested();

                await using var context = await contextFactory.CreateDbContextAsync(cancellationToken);
                await using var transaction = await context.Database.BeginTransactionAsync(IsolationLevel.Serializable, cancellationToken);
                var persisted = (await CreateRecordsAsync(context, [job], cancellationToken))[0];

                _ = context.Set<ImmediateJobEntity>().Add(ToEntity(persisted));
                _ = await context.SaveChangesAsync(cancellationToken);
                await transaction.CommitAsync(cancellationToken);
        }

        private ValueTask ExecuteGraphInsertAsync(
                BatchRecord? batch,
                IReadOnlyList<JobSubmission> jobs,
                IReadOnlyList<JobContinuationEdge> edges,
                CancellationToken cancellationToken
        ) => RetryConcurrencyAsync(
                operationCancellationToken => InsertGraphCoreAsync(batch, jobs, edges, operationCancellationToken),
                cancellationToken
        );

        private async Task InsertGraphCoreAsync(
                BatchRecord? batch,
                IReadOnlyList<JobSubmission> jobs,
                IReadOnlyList<JobContinuationEdge> edges,
                CancellationToken cancellationToken
        )
        {
                var jobHandles = jobs.Select(static job => job.JobHandle.Value).ToHashSet(StringComparer.Ordinal);
                if (jobHandles.Count != jobs.Count)
                        throw new ImmediateJobException("A batch or continuation insert contains duplicate job identifiers.");
                if (batch is not null && jobs.Any(job => job.BatchHandle != batch.BatchHandle))
                        throw new ImmediateJobException("Every atomic batch member must carry the committed batch identifier.");

                var edgeEntities = edges.Select(ToEntity).ToList();
                if (edgeEntities.Any(edge => !jobHandles.Contains(edge.ChildJobHandle)))
                        throw new ImmediateJobException("Every continuation edge must target a job inserted by the same operation.");
                if (edgeEntities.DistinctBy(static edge => (edge.ChildJobHandle, edge.ParentKind, edge.ParentId)).Count() != edgeEntities.Count)
                        throw new ImmediateJobException("Duplicate continuation edges are not allowed.");
                ThrowIfCyclic(jobHandles, edgeEntities);

                await using var context = await contextFactory.CreateDbContextAsync(cancellationToken);
                await using var transaction = await context.Database.BeginTransactionAsync(IsolationLevel.Serializable, cancellationToken);
                var records = await CreateRecordsAsync(context, jobs, cancellationToken);

                var jobEntities = records.Select(ToEntity).ToDictionary(static job => job.Id, StringComparer.Ordinal);
                await EvaluateInitialDependenciesAsync(
                        context,
                        jobEntities,
                        edgeEntities,
                        _timeProvider.GetUtcNow(),
                        cancellationToken
                );

                if (batch is not null)
                {
                        var terminal = jobEntities.Values.Where(static job => IsTerminal(job.State)).ToList();
                        var pending = jobEntities.Count - terminal.Count;
                        var succeeded = terminal.Count(static job => job.State == JobState.Succeeded);
                        var failed = terminal.Count(static job => job.State == JobState.Failed);
                        var cancelled = terminal.Count(static job => job.State == JobState.Cancelled);
                        var skipped = terminal.Count(static job => job.State == JobState.Skipped);
                        _ = context.Add(new ImmediateJobBatchEntity
                        {
                                Id = batch.BatchHandle.Value,
                                CreatedAt = batch.CreatedAt,
                                TotalJobs = jobEntities.Count,
                                PendingCount = pending,
                                SucceededCount = succeeded,
                                FailedCount = failed,
                                CancelledCount = cancelled,
                                SkippedCount = skipped,
                                StartedAt = batch.StartedAt,
                                CompletedAt = pending == 0 ? batch.CompletedAt ?? _timeProvider.GetUtcNow() : null,
                                State = pending == 0
                                        ? GetTerminalBatchState(failed, cancelled)
                                        : batch.State == BatchState.WaitingForTrigger ? BatchState.WaitingForTrigger : BatchState.Executing,
                                ConcurrencyStamp = Guid.NewGuid(),
                        });
                }

                context.AddRange(jobEntities.Values);
                context.AddRange(edgeEntities);
                _ = await context.SaveChangesAsync(cancellationToken);
                await transaction.CommitAsync(cancellationToken);
        }

        private async Task AddBatchJobCoreAsync(
                JobHandle currentJobHandle,
                int executionNumber,
                JobSubmission record,
                ContinuationOptions options,
                CancellationToken cancellationToken
        )
        {
                await using var context = await contextFactory.CreateDbContextAsync(cancellationToken);
                await using var transaction = await context.Database.BeginTransactionAsync(IsolationLevel.Serializable, cancellationToken);
                var current = await context.Set<ImmediateJobEntity>()
                        .SingleOrDefaultAsync(job => job.Id == currentJobHandle.Value && job.Attempt == executionNumber && job.State == JobState.Active, cancellationToken)
                        ?? throw new ImmediateJobException($"The current active job '{currentJobHandle}' was not found.");
                if (current.BatchHandle is not { } batchHandle)
                        throw new ImmediateJobException("The current job does not belong to a batch.");
                if (options is not (ContinuationOptions.BesideContinuations or ContinuationOptions.BeforeContinuations))
                        throw new ArgumentOutOfRangeException(nameof(options));
                if (!string.Equals(record.BatchHandle?.Value, batchHandle, StringComparison.Ordinal))
                        throw new ImmediateJobException("The new job must belong to the current job's batch.");
                if (record.State is not (JobState.Pending or JobState.Scheduled))
                        throw new ImmediateJobException($"Concurrent batch member '{record.JobHandle}' has invalid state '{record.State}'.");

                var batch = await context.Set<ImmediateJobBatchEntity>()
                        .SingleAsync(item => item.Id == batchHandle && item.State == BatchState.Executing, cancellationToken);
                var persisted = (await CreateRecordsAsync(context, [record], cancellationToken))[0];
                var job = ToEntity(persisted);
                _ = context.Add(job);
                batch.TotalJobs++;
                batch.PendingCount++;
                batch.ConcurrencyStamp = Guid.NewGuid();

                if (options == ContinuationOptions.BeforeContinuations)
                {
                        var waiters = await GetActiveWaitersAsync(context, currentJobHandle, cancellationToken);
                        foreach (var waiter in waiters)
                        {
                                _ = context.Add(new ImmediateJobContinuationEntity
                                {
                                        ChildJobHandle = waiter.Id,
                                        ParentKind = ContinuationParentKind.Job,
                                        ParentId = job.Id,
                                        Delay = 0,
                                        Trigger = ContinuationTrigger.Success,
                                });
                                waiter.RemainingDependencies++;
                                waiter.ConcurrencyStamp = Guid.NewGuid();
                        }
                }

                _ = await context.SaveChangesAsync(cancellationToken);
                await transaction.CommitAsync(cancellationToken);
        }

        private static async Task<Dictionary<string, string>> GetDefinitionNamesAsync(
                TContext context,
                List<string> names,
                CancellationToken cancellationToken
        )
        {
                if (names.Count == 0)
                        return [with(StringComparer.OrdinalIgnoreCase)];

                var query = context.Set<ImmediateJobDefinitionMetadataEntity>().AsNoTracking();
                // SQLite's NOCASE collation only handles ASCII; compare Unicode aliases in the client.
                if (context.Database.ProviderName?.Contains("Sqlite", StringComparison.Ordinal) != true)
                        query = query.Where(definition => names.Contains(definition.Name));

                var definitions = await query
                        .Select(definition => definition.Name)
                        .ToListAsync(cancellationToken);

                return definitions.ToDictionary(
                        static name => name,
                        static name => name,
                        StringComparer.OrdinalIgnoreCase
                );
        }

        private static async Task<IReadOnlyList<JobRecord>> CreateRecordsAsync(
                TContext context,
                IReadOnlyList<JobSubmission> submissions,
                CancellationToken cancellationToken
        )
        {
                var names = submissions.Select(static job => job.JobName).Distinct(StringComparer.OrdinalIgnoreCase).ToList();
                var definitions = await GetDefinitionNamesAsync(context, names, cancellationToken);
                return submissions.Select(job =>
                {
                        if (!definitions.TryGetValue(job.JobName, out var definition))
                                throw new KeyNotFoundException($"Job definition '{job.JobName}' was not found.");

                        return job.ToJobRecord(definition);
                }).ToList();
        }

        [LoggerMessage(
                EventId = LibraryEventIds.EnqueueAsyncCalled,
                EventName = "Immediate.Jobs.EntityFrameworkCore.EnqueueAsyncCalled",
                Level = LogLevel.Debug,
                Message = "EnqueueAsync called (JobHandle={JobHandle})"
        )]
        private partial void EnqueueAsyncCalled(JobHandle jobHandle);

        [LoggerMessage(
                EventId = LibraryEventIds.EnqueueContinuationAsyncCalled,
                EventName = "Immediate.Jobs.EntityFrameworkCore.EnqueueContinuationAsyncCalled",
                Level = LogLevel.Debug,
                Message = "EnqueueContinuationAsync called (JobHandle={JobHandle}, Edges={Edges})"
        )]
        private partial void EnqueueContinuationAsyncCalled(JobHandle jobHandle, int edges);

        [LoggerMessage(
                EventId = LibraryEventIds.EnqueueBatchAsyncCalled,
                EventName = "Immediate.Jobs.EntityFrameworkCore.EnqueueBatchAsyncCalled",
                Level = LogLevel.Debug,
                Message = "EnqueueBatchAsync called (BatchHandle={BatchHandle}, Jobs={Jobs}, Edges={Edges})"
        )]
        private partial void EnqueueBatchAsyncCalled(BatchHandle batchHandle, int jobs, int edges);

        [LoggerMessage(
                EventId = LibraryEventIds.AddBatchJobAsyncCalled,
                EventName = "Immediate.Jobs.EntityFrameworkCore.AddBatchJobAsyncCalled",
                Level = LogLevel.Debug,
                Message = "AddBatchJobAsync called (JobHandle={JobHandle}, Execution={Execution})"
        )]
        private partial void AddBatchJobAsyncCalled(JobHandle jobHandle, int execution);
}

1	using Immediate.Jobs.Shared.Apis;
2	using System.Data;
3	using Microsoft.EntityFrameworkCore;
4	using Microsoft.Extensions.Logging;
5
6	namespace Immediate.Jobs.EntityFrameworkCore;
7
8	internal sealed partial class EntityFrameworkCoreJobStorage<TContext>
9	where TContext : DbContext
10	{
11	/// <inheritdoc />
12	public async ValueTask EnqueueAsync(JobSubmission job, CancellationToken cancellationToken = default)
13	{
14	EnqueueAsyncCalled(job.JobHandle);	3 ✔
15	cancellationToken.ThrowIfCancellationRequested();	3 ✔
16	await TaskScheduler.Yield();	3 ✔
17
18	await RetryConcurrencyAsync(	3 ✔
19	operationCancellationToken => EnqueueCoreAsync(job, operationCancellationToken),	3 ✔
20	cancellationToken	3 ✔
21	);	3 ✔
22	}	3 ✔
23
24	/// <inheritdoc />
25	public async ValueTask EnqueueContinuationAsync(
26	JobSubmission job,
27	IReadOnlyList<JobContinuationEdge> edges,
28	CancellationToken cancellationToken = default
29	)
30	{
31	EnqueueContinuationAsyncCalled(job.JobHandle, edges.Count);	3 ✔
32	cancellationToken.ThrowIfCancellationRequested();	3 ✔
33	await TaskScheduler.Yield();	3 ✔
34
35	await ExecuteGraphInsertAsync(batch: null, [job], edges, cancellationToken);	3 ✔
36	}	3 ✔
37
38	/// <inheritdoc />
39	public async ValueTask EnqueueBatchAsync(
40	BatchRecord batch,
41	IReadOnlyList<JobSubmission> jobs,
42	IReadOnlyList<JobContinuationEdge> edges,
43	CancellationToken cancellationToken = default
44	)
45	{
46	EnqueueBatchAsyncCalled(batch.BatchHandle, jobs.Count, edges.Count);	3 ✔
47	cancellationToken.ThrowIfCancellationRequested();	3 ✔
48	await TaskScheduler.Yield();	3 ✔
49
50	await ExecuteGraphInsertAsync(batch, jobs, edges, cancellationToken);	3 ✔
51	}	3 ✔
52
53	/// <inheritdoc />
54	public async ValueTask AddBatchJobAsync(
55	JobHandle currentJobHandle,
56	int executionNumber,
57	JobSubmission job,
58	ContinuationOptions options,
59	CancellationToken cancellationToken = default
60	)
61	{
62	AddBatchJobAsyncCalled(job.JobHandle, executionNumber);	3 ✔
63	cancellationToken.ThrowIfCancellationRequested();	3 ✔
64	await TaskScheduler.Yield();	3 ✔
65
66	await RetryConcurrencyAsync(	3 ✔
67	operationCancellationToken => AddBatchJobCoreAsync(	3 ✔
68	currentJobHandle,	3 ✔
69	executionNumber,	3 ✔
70	job,	3 ✔
71	options,	3 ✔
72	operationCancellationToken	3 ✔
73	),	3 ✔
74	cancellationToken	3 ✔
75	);	3 ✔
76	}	3 ✔
77
78	private async Task EnqueueCoreAsync(JobSubmission job, CancellationToken cancellationToken)
79	{
80	cancellationToken.ThrowIfCancellationRequested();	3 ✔
81
82	await using var context = await contextFactory.CreateDbContextAsync(cancellationToken);	3 ✔
83	await using var transaction = await context.Database.BeginTransactionAsync(IsolationLevel.Serializable, cancellationToken);	3 ✔
84	var persisted = (await CreateRecordsAsync(context, [job], cancellationToken))[0];	3 ✔
85
86	_ = context.Set<ImmediateJobEntity>().Add(ToEntity(persisted));	3 ✔
87	_ = await context.SaveChangesAsync(cancellationToken);	3 ✔
88	await transaction.CommitAsync(cancellationToken);	3 ✔
89	}	3 ✔
90
91	private ValueTask ExecuteGraphInsertAsync(
92	BatchRecord? batch,
93	IReadOnlyList<JobSubmission> jobs,
94	IReadOnlyList<JobContinuationEdge> edges,
95	CancellationToken cancellationToken
96	) => RetryConcurrencyAsync(	3 ✔
97	operationCancellationToken => InsertGraphCoreAsync(batch, jobs, edges, operationCancellationToken),	3 ✔
98	cancellationToken	3 ✔
99	);	3 ✔
100
101	private async Task InsertGraphCoreAsync(
102	BatchRecord? batch,
103	IReadOnlyList<JobSubmission> jobs,
104	IReadOnlyList<JobContinuationEdge> edges,
105	CancellationToken cancellationToken
106	)
107	{
108	var jobHandles = jobs.Select(static job => job.JobHandle.Value).ToHashSet(StringComparer.Ordinal);	3 ✔
109	if (jobHandles.Count != jobs.Count)	3 ✔
110	throw new ImmediateJobException("A batch or continuation insert contains duplicate job identifiers.");	×
111	if (batch is not null && jobs.Any(job => job.BatchHandle != batch.BatchHandle))	3 ✔
112	throw new ImmediateJobException("Every atomic batch member must carry the committed batch identifier.");	×
113
114	var edgeEntities = edges.Select(ToEntity).ToList();	3 ✔
115	if (edgeEntities.Any(edge => !jobHandles.Contains(edge.ChildJobHandle)))	3 ✔
116	throw new ImmediateJobException("Every continuation edge must target a job inserted by the same operation.");	×
117	if (edgeEntities.DistinctBy(static edge => (edge.ChildJobHandle, edge.ParentKind, edge.ParentId)).Count() != edgeEntities.Count)	3 ✔
118	throw new ImmediateJobException("Duplicate continuation edges are not allowed.");	×
119	ThrowIfCyclic(jobHandles, edgeEntities);	3 ✔
120
121	await using var context = await contextFactory.CreateDbContextAsync(cancellationToken);	3 ✔
122	await using var transaction = await context.Database.BeginTransactionAsync(IsolationLevel.Serializable, cancellationToken);	3 ✔
123	var records = await CreateRecordsAsync(context, jobs, cancellationToken);	3 ✔
124
125	var jobEntities = records.Select(ToEntity).ToDictionary(static job => job.Id, StringComparer.Ordinal);	3 ✔
126	await EvaluateInitialDependenciesAsync(	3 ✔
127	context,	3 ✔
128	jobEntities,	3 ✔
129	edgeEntities,	3 ✔
130	_timeProvider.GetUtcNow(),	3 ✔
131	cancellationToken	3 ✔
132	);	3 ✔
133
134	if (batch is not null)	3 ✔
135	{
136	var terminal = jobEntities.Values.Where(static job => IsTerminal(job.State)).ToList();	3 ✔
137	var pending = jobEntities.Count - terminal.Count;	3 ✔
138	var succeeded = terminal.Count(static job => job.State == JobState.Succeeded);	2 ✔
139	var failed = terminal.Count(static job => job.State == JobState.Failed);	2 ✔
140	var cancelled = terminal.Count(static job => job.State == JobState.Cancelled);	2 ✔
141	var skipped = terminal.Count(static job => job.State == JobState.Skipped);	2 ✔
142	_ = context.Add(new ImmediateJobBatchEntity	3 ✔
143	{	3 ✔
144	Id = batch.BatchHandle.Value,	3 ✔
145	CreatedAt = batch.CreatedAt,	3 ✔
146	TotalJobs = jobEntities.Count,	3 ✔
147	PendingCount = pending,	3 ✔
148	SucceededCount = succeeded,	3 ✔
149	FailedCount = failed,	3 ✔
150	CancelledCount = cancelled,	3 ✔
151	SkippedCount = skipped,	3 ✔
152	StartedAt = batch.StartedAt,	3 ✔
153	CompletedAt = pending == 0 ? batch.CompletedAt ?? _timeProvider.GetUtcNow() : null,	3 ✔
154	State = pending == 0	3 ✔
155	? GetTerminalBatchState(failed, cancelled)	3 ✔
156	: batch.State == BatchState.WaitingForTrigger ? BatchState.WaitingForTrigger : BatchState.Executing,	3 ✔
157	ConcurrencyStamp = Guid.NewGuid(),	3 ✔
158	});	3 ✔
159	}
160
161	context.AddRange(jobEntities.Values);	3 ✔
162	context.AddRange(edgeEntities);	3 ✔
163	_ = await context.SaveChangesAsync(cancellationToken);	3 ✔
164	await transaction.CommitAsync(cancellationToken);	3 ✔
165	}	3 ✔
166
167	private async Task AddBatchJobCoreAsync(
168	JobHandle currentJobHandle,
169	int executionNumber,
170	JobSubmission record,
171	ContinuationOptions options,
172	CancellationToken cancellationToken
173	)
174	{
175	await using var context = await contextFactory.CreateDbContextAsync(cancellationToken);	3 ✔
176	await using var transaction = await context.Database.BeginTransactionAsync(IsolationLevel.Serializable, cancellationToken);	3 ✔
177	var current = await context.Set<ImmediateJobEntity>()	3 ✔
178	.SingleOrDefaultAsync(job => job.Id == currentJobHandle.Value && job.Attempt == executionNumber && job.State == JobState.Active, cancellationToken)	3 ✔
179	?? throw new ImmediateJobException($"The current active job '{currentJobHandle}' was not found.");	3 ✔
180	if (current.BatchHandle is not { } batchHandle)	3 ✔
181	throw new ImmediateJobException("The current job does not belong to a batch.");	×
182	if (options is not (ContinuationOptions.BesideContinuations or ContinuationOptions.BeforeContinuations))	3 ✔
183	throw new ArgumentOutOfRangeException(nameof(options));	×
184	if (!string.Equals(record.BatchHandle?.Value, batchHandle, StringComparison.Ordinal))	3 ✔
185	throw new ImmediateJobException("The new job must belong to the current job's batch.");	×
186	if (record.State is not (JobState.Pending or JobState.Scheduled))	3 ✔
187	throw new ImmediateJobException($"Concurrent batch member '{record.JobHandle}' has invalid state '{record.State}'.");	3 ✔
188
189	var batch = await context.Set<ImmediateJobBatchEntity>()	3 ✔
190	.SingleAsync(item => item.Id == batchHandle && item.State == BatchState.Executing, cancellationToken);	3 ✔
191	var persisted = (await CreateRecordsAsync(context, [record], cancellationToken))[0];	3 ✔
192	var job = ToEntity(persisted);	3 ✔
193	_ = context.Add(job);	3 ✔
194	batch.TotalJobs++;	3 ✔
195	batch.PendingCount++;	3 ✔
196	batch.ConcurrencyStamp = Guid.NewGuid();	3 ✔
197
198	if (options == ContinuationOptions.BeforeContinuations)	3 ✔
199	{
200	var waiters = await GetActiveWaitersAsync(context, currentJobHandle, cancellationToken);	3 ✔
201	foreach (var waiter in waiters)	3 ✔
202	{
203	_ = context.Add(new ImmediateJobContinuationEntity	3 ✔
204	{	3 ✔
205	ChildJobHandle = waiter.Id,	3 ✔
206	ParentKind = ContinuationParentKind.Job,	3 ✔
207	ParentId = job.Id,	3 ✔
208	Delay = 0,	3 ✔
209	Trigger = ContinuationTrigger.Success,	3 ✔
210	});	3 ✔
211	waiter.RemainingDependencies++;	3 ✔
212	waiter.ConcurrencyStamp = Guid.NewGuid();	3 ✔
213	}
214	}
215
216	_ = await context.SaveChangesAsync(cancellationToken);	3 ✔
217	await transaction.CommitAsync(cancellationToken);	3 ✔
218	}	3 ✔
219
220	private static async Task<Dictionary<string, string>> GetDefinitionNamesAsync(
221	TContext context,
222	List<string> names,
223	CancellationToken cancellationToken
224	)
225	{
226	if (names.Count == 0)	3 ✔
NEW 227	return [with(StringComparer.OrdinalIgnoreCase)];	×
228
229	var query = context.Set<ImmediateJobDefinitionMetadataEntity>().AsNoTracking();	3 ✔
230	// SQLite's NOCASE collation only handles ASCII; compare Unicode aliases in the client.
231	if (context.Database.ProviderName?.Contains("Sqlite", StringComparison.Ordinal) != true)	3 ✔
232	query = query.Where(definition => names.Contains(definition.Name));	3 ✔
233
234	var definitions = await query	3 ✔
235	.Select(definition => definition.Name)	3 ✔
236	.ToListAsync(cancellationToken);	3 ✔
237
238	return definitions.ToDictionary(	3 ✔
239	static name => name,	3 ✔
240	static name => name,	3 ✔
241	StringComparer.OrdinalIgnoreCase	3 ✔
242	);	3 ✔
243	}	2 ✔
244
245	private static async Task<IReadOnlyList<JobRecord>> CreateRecordsAsync(
246	TContext context,
247	IReadOnlyList<JobSubmission> submissions,
248	CancellationToken cancellationToken
249	)
250	{
251	var names = submissions.Select(static job => job.JobName).Distinct(StringComparer.OrdinalIgnoreCase).ToList();	3 ✔
252	var definitions = await GetDefinitionNamesAsync(context, names, cancellationToken);	3 ✔
253	return submissions.Select(job =>	3 ✔
254	{	3 ✔
255	if (!definitions.TryGetValue(job.JobName, out var definition))	3 ✔
256	throw new KeyNotFoundException($"Job definition '{job.JobName}' was not found.");	3 ✔
257		3 ✔
258	return job.ToJobRecord(definition);	3 ✔
259	}).ToList();	3 ✔
260	}	2 ✔
261
262	[LoggerMessage(
263	EventId = LibraryEventIds.EnqueueAsyncCalled,
264	EventName = "Immediate.Jobs.EntityFrameworkCore.EnqueueAsyncCalled",
265	Level = LogLevel.Debug,
266	Message = "EnqueueAsync called (JobHandle={JobHandle})"
267	)]
268	private partial void EnqueueAsyncCalled(JobHandle jobHandle);
269
270	[LoggerMessage(
271	EventId = LibraryEventIds.EnqueueContinuationAsyncCalled,
272	EventName = "Immediate.Jobs.EntityFrameworkCore.EnqueueContinuationAsyncCalled",
273	Level = LogLevel.Debug,
274	Message = "EnqueueContinuationAsync called (JobHandle={JobHandle}, Edges={Edges})"
275	)]
276	private partial void EnqueueContinuationAsyncCalled(JobHandle jobHandle, int edges);
277
278	[LoggerMessage(
279	EventId = LibraryEventIds.EnqueueBatchAsyncCalled,
280	EventName = "Immediate.Jobs.EntityFrameworkCore.EnqueueBatchAsyncCalled",
281	Level = LogLevel.Debug,
282	Message = "EnqueueBatchAsync called (BatchHandle={BatchHandle}, Jobs={Jobs}, Edges={Edges})"
283	)]
284	private partial void EnqueueBatchAsyncCalled(BatchHandle batchHandle, int jobs, int edges);
285
286	[LoggerMessage(
287	EventId = LibraryEventIds.AddBatchJobAsyncCalled,
288	EventName = "Immediate.Jobs.EntityFrameworkCore.AddBatchJobAsyncCalled",
289	Level = LogLevel.Debug,
290	Message = "AddBatchJobAsync called (JobHandle={JobHandle}, Execution={Execution})"
291	)]
292	private partial void AddBatchJobAsyncCalled(JobHandle jobHandle, int execution);
293	}

ImmediatePlatform / Immediate.Jobs / 38501404688

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