30300691995

Committed 27 Jul 2026 08:00PM UTC coverage: 64.845%. First build

Build # 30300691995

Build Type

Pull #20

github

Committed by

web-flow

Commit Message

Merge a73d5f4ed into e8a73c280

Pull Request Pull Request #20: Initial draft

Coverage Stats

1300 of 1892 new or added lines in 29 files covered. (68.71%)

1649 of 2543 relevant lines covered (64.84%)

2.58 hits per line

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

81.82

/src/Immediate.Jobs.Shared/InMemoryJobStorage.cs

namespace Immediate.Jobs.Shared;

#pragma warning disable IDE0391 // Keep synchronous storage methods async for .NET 11 runtime-async state machines.

/// <summary>
/// A best-effort, non-durable, single-node provider intended for development and tests.
/// </summary>
public sealed class InMemoryJobStorage(TimeProvider timeProvider) : IJobStorage, IJobStorageReplica
{
        private readonly Lock _gate = new();
        private readonly Dictionary<string, JobRecord> _jobs = new(StringComparer.Ordinal);
        private readonly Dictionary<string, RecurringJobSchedule> _recurring = new(StringComparer.Ordinal);
        private readonly Dictionary<string, JobServerSnapshot> _servers = new(StringComparer.Ordinal);
        private readonly HashSet<string> _recurringKeys = new(StringComparer.Ordinal);

        /// <inheritdoc />
        public ValueTask InitializeAsync(CancellationToken cancellationToken = default)
        {
                cancellationToken.ThrowIfCancellationRequested();
                return ValueTask.CompletedTask;
        }

        /// <inheritdoc />
        public ValueTask EnqueueAsync(JobRecord job, CancellationToken cancellationToken = default)
        {
                ArgumentNullException.ThrowIfNull(job);
                cancellationToken.ThrowIfCancellationRequested();
                lock (_gate)
                {
                        if (!_jobs.TryAdd(job.Id, job))
                                throw new InvalidOperationException($"Job '{job.Id}' already exists.");
                }

                return ValueTask.CompletedTask;
        }

        /// <inheritdoc />
        public async ValueTask<IReadOnlyList<JobRecord>> AcquireDueJobsAsync(
                JobAcquisitionRequest request,
                CancellationToken cancellationToken = default
        )
        {
                ArgumentNullException.ThrowIfNull(request);
                cancellationToken.ThrowIfCancellationRequested();
                var now = timeProvider.GetUtcNow();
                lock (_gate)
                {
                        foreach (var expired in _jobs.Values.Where(x => x.State == JobState.Active && x.LeaseExpiresAt <= now).ToArray())
                        {
                                _jobs[expired.Id] = expired with
                                {
                                        State = JobState.Pending,
                                        WorkerId = null,
                                        LeaseExpiresAt = null,
                                };
                        }

                        var acquired = new List<JobRecord>(request.BatchSize);
                        foreach (var queue in request.Queues)
                        {
                                var queueCapacity = Math.Min(queue.Capacity, request.BatchSize - acquired.Count);
                                if (queueCapacity <= 0)
                                        continue;

                                var jobCapacities = queue.JobCapacities.ToDictionary(static pair => pair.Key, static pair => pair.Value, StringComparer.Ordinal);
                                foreach (var candidate in _jobs.Values
                                        .Where(job => job.QueueName == queue.QueueName &&
                                                jobCapacities.ContainsKey(job.JobName) &&
                                                job.State is JobState.Pending or JobState.Scheduled && job.DueAt <= now)
                                        .OrderBy(job => job.DueAt)
                                        .ThenBy(job => job.CreatedAt)
                                        .ThenBy(job => job.Id))
                                {
                                        if (queueCapacity == 0)
                                                break;
                                        if (jobCapacities[candidate.JobName] <= 0)
                                                continue;

                                        var job = candidate with
                                        {
                                                State = JobState.Active,
                                                Attempt = candidate.Attempt + 1,
                                                WorkerId = request.WorkerId,
                                                LeaseExpiresAt = now + request.Lease,
                                        };
                                        _jobs[job.Id] = job;
                                        acquired.Add(job);
                                        jobCapacities[job.JobName]--;
                                        queueCapacity--;
                                }
                        }

                        return acquired;
                }
        }

        /// <inheritdoc />
        public async ValueTask<IReadOnlyList<JobRecord>> AcquireJobsAsync(
                IReadOnlyCollection<string> jobIds,
                string workerId,
                TimeSpan lease,
                CancellationToken cancellationToken = default
        )
        {
                ArgumentNullException.ThrowIfNull(jobIds);
                ArgumentException.ThrowIfNullOrWhiteSpace(workerId);
                ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(lease, TimeSpan.Zero);
                cancellationToken.ThrowIfCancellationRequested();
                var now = timeProvider.GetUtcNow();
                lock (_gate)
                {
                        var acquired = new List<JobRecord>(jobIds.Count);
                        foreach (var id in jobIds)
                        {
                                if (!_jobs.TryGetValue(id, out var job))
                                        continue;
                                if (job.State == JobState.Active && job.LeaseExpiresAt <= now)
                                {
                                        job = job with { State = JobState.Pending, WorkerId = null, LeaseExpiresAt = null };
                                        _jobs[id] = job;
                                }

                                if (job.State is not (JobState.Pending or JobState.Scheduled) || job.DueAt > now)
                                        continue;

                                job = job with
                                {
                                        State = JobState.Active,
                                        Attempt = job.Attempt + 1,
                                        WorkerId = workerId,
                                        LeaseExpiresAt = now + lease,
                                };
                                _jobs[id] = job;
                                acquired.Add(job);
                        }

                        return acquired;
                }
        }

        /// <inheritdoc />
        public ValueTask RenewLeaseAsync(string jobId, string workerId, TimeSpan lease, CancellationToken cancellationToken = default)
        {
                cancellationToken.ThrowIfCancellationRequested();
                lock (_gate)
                {
                        var job = GetOwnedActive(jobId, workerId);
                        _jobs[jobId] = job with { LeaseExpiresAt = timeProvider.GetUtcNow() + lease };
                }

                return ValueTask.CompletedTask;
        }

        /// <inheritdoc />
        public ValueTask CompleteAsync(string jobId, string workerId, CancellationToken cancellationToken = default)
        {
                cancellationToken.ThrowIfCancellationRequested();
                lock (_gate)
                {
                        var job = GetOwnedActive(jobId, workerId);
                        _jobs[jobId] = job with
                        {
                                State = JobState.Succeeded,
                                WorkerId = null,
                                LeaseExpiresAt = null,
                                CompletedAt = timeProvider.GetUtcNow(),
                        };
                }

                return ValueTask.CompletedTask;
        }

        /// <inheritdoc />
        public ValueTask FailAsync(
                string jobId,
                string workerId,
                string error,
                DateTimeOffset? nextRetryAt,
                CancellationToken cancellationToken = default
        )
        {
                cancellationToken.ThrowIfCancellationRequested();
                lock (_gate)
                {
                        var job = GetOwnedActive(jobId, workerId);
                        _jobs[jobId] = job with
                        {
                                State = nextRetryAt.HasValue ? JobState.Scheduled : JobState.Failed,
                                DueAt = nextRetryAt ?? job.DueAt,
                                WorkerId = null,
                                LeaseExpiresAt = null,
                                LastError = error,
                                CompletedAt = nextRetryAt.HasValue ? null : timeProvider.GetUtcNow(),
                        };
                }

                return ValueTask.CompletedTask;
        }

        /// <inheritdoc />
        public ValueTask UpsertRecurringAsync(RecurringJobSchedule schedule, CancellationToken cancellationToken = default)
        {
                ArgumentNullException.ThrowIfNull(schedule);
                cancellationToken.ThrowIfCancellationRequested();
                lock (_gate)
                {
                        if (_recurring.TryGetValue(schedule.Name, out var current))
                        {
                                if (current.IsCodeDefined && !schedule.IsCodeDefined)
                                        throw new InvalidOperationException("Code-defined recurring schedules cannot be replaced by dynamic schedules.");

                                schedule = schedule with { IsPaused = current.IsPaused, LastRunAt = current.LastRunAt };
                        }

                        _recurring[schedule.Name] = schedule;
                }

                return ValueTask.CompletedTask;
        }

        /// <inheritdoc />
        public ValueTask RemoveObsoleteCodeDefinedRecurringAsync(
                IReadOnlyCollection<string> activeScheduleNames,
                CancellationToken cancellationToken = default
        )
        {
                ArgumentNullException.ThrowIfNull(activeScheduleNames);
                cancellationToken.ThrowIfCancellationRequested();
                var activeNames = activeScheduleNames.ToHashSet(StringComparer.Ordinal);
                lock (_gate)
                {
                        var obsoleteNames = _recurring
                                .Where(schedule => schedule.Value.IsCodeDefined && !activeNames.Contains(schedule.Key))
                                .Select(static schedule => schedule.Key)
                                .ToArray();
                        foreach (var name in obsoleteNames)
                                _ = _recurring.Remove(name);
                }

                return ValueTask.CompletedTask;
        }

        /// <inheritdoc />
        public ValueTask RemoveRecurringAsync(string name, CancellationToken cancellationToken = default)
        {
                cancellationToken.ThrowIfCancellationRequested();
                lock (_gate)
                {
                        if (_recurring.TryGetValue(name, out var schedule) && schedule.IsCodeDefined)
                                throw new InvalidOperationException("Code-defined recurring schedules cannot be deleted.");

                        _ = _recurring.Remove(name);
                }

                return ValueTask.CompletedTask;
        }

        /// <inheritdoc />
        public ValueTask PauseRecurringAsync(string name, CancellationToken cancellationToken = default) =>
                SetRecurringPaused(name, true, cancellationToken);

        /// <inheritdoc />
        public ValueTask ResumeRecurringAsync(string name, CancellationToken cancellationToken = default) =>
                SetRecurringPaused(name, false, cancellationToken);

        /// <inheritdoc />
        public async ValueTask<IReadOnlyList<RecurringJobSchedule>> GetDueRecurringAsync(
                DateTimeOffset now,
                int batchSize,
                CancellationToken cancellationToken = default
        )
        {
                cancellationToken.ThrowIfCancellationRequested();
                lock (_gate)
                {
                        return
                        [
                                .. _recurring.Values.Where(x => !x.IsPaused && x.NextRunAt <= now).OrderBy(x => x.NextRunAt).Take(batchSize),
                        ];
                }
        }

        /// <inheritdoc />
        public async ValueTask<bool> MaterializeRecurringAsync(
                RecurringJobSchedule schedule,
                JobRecord job,
                DateTimeOffset nextRunAt,
                CancellationToken cancellationToken = default
        )
        {
                ArgumentNullException.ThrowIfNull(schedule);
                ArgumentNullException.ThrowIfNull(job);
                cancellationToken.ThrowIfCancellationRequested();
                lock (_gate)
                {
                        if (!_recurring.TryGetValue(schedule.Name, out var current) || current.NextRunAt != schedule.NextRunAt)
                                return false;

                        var inserted = job.RecurringKey is null || _recurringKeys.Add(job.RecurringKey);
                        if (inserted)
                                _jobs[job.Id] = job;
                        _recurring[schedule.Name] = current with { LastRunAt = schedule.NextRunAt, NextRunAt = nextRunAt };
                        return inserted;
                }
        }

        /// <inheritdoc />
        public async ValueTask<JobMonitoringSnapshot> GetMonitoringSnapshotAsync(CancellationToken cancellationToken = default)
        {
                cancellationToken.ThrowIfCancellationRequested();
                lock (_gate)
                {
                        var counts = Enum.GetValues<JobState>().ToDictionary(state => state, state => _jobs.Values.LongCount(x => x.State == state));
                        var cutoff = timeProvider.GetUtcNow() - TimeSpan.FromMinutes(2);
                        IReadOnlyList<JobServerSnapshot> servers = [.. _servers.Values.Where(x => x.LastHeartbeat >= cutoff)];
                        return new(timeProvider.GetUtcNow(), counts, [.. _recurring.Values], servers);
                }
        }

        /// <inheritdoc />
        public async ValueTask<IReadOnlyList<JobRecord>> QueryJobsAsync(JobQuery query, CancellationToken cancellationToken = default)
        {
                ArgumentNullException.ThrowIfNull(query);
                cancellationToken.ThrowIfCancellationRequested();
                lock (_gate)
                {
                        var jobs = _jobs.Values.AsEnumerable();
                        if (query.Id is { } id)
                                jobs = jobs.Where(x => x.Id == id);
                        if (query.State is { } state)
                                jobs = jobs.Where(x => x.State == state);
                        if (!string.IsNullOrWhiteSpace(query.QueueName))
                                jobs = jobs.Where(x => x.QueueName == query.QueueName);

                        if (!string.IsNullOrWhiteSpace(query.Search))
                                jobs = jobs.Where(x => x.JobName.Contains(query.Search, StringComparison.OrdinalIgnoreCase));

                        return
                        [
                                .. jobs.OrderByDescending(x => x.CreatedAt).Skip(Math.Max(0, query.Skip)).Take(Math.Clamp(query.Take, 1, 1000)),
                        ];
                }
        }

        /// <inheritdoc />
        public ValueTask RetryAsync(string jobId, CancellationToken cancellationToken = default)
        {
                cancellationToken.ThrowIfCancellationRequested();
                lock (_gate)
                {
                        if (!_jobs.TryGetValue(jobId, out var job) || job.State != JobState.Failed)
                                throw new InvalidOperationException("Only failed jobs can be retried.");

                        _jobs[jobId] = job with { State = JobState.Pending, DueAt = timeProvider.GetUtcNow(), CompletedAt = null };
                }

                return ValueTask.CompletedTask;
        }

        /// <inheritdoc />
        public ValueTask DeleteAsync(string jobId, CancellationToken cancellationToken = default)
        {
                cancellationToken.ThrowIfCancellationRequested();
                lock (_gate)
                {
                        if (_jobs.TryGetValue(jobId, out var job) && job.State is JobState.Active or JobState.Pending or JobState.Scheduled)
                                throw new InvalidOperationException("Only terminal jobs can be deleted.");

                        _ = _jobs.Remove(jobId);
                }

                return ValueTask.CompletedTask;
        }

        /// <inheritdoc />
        public ValueTask PurgeAsync(TimeSpan succeededRetention, TimeSpan failedRetention, CancellationToken cancellationToken = default)
        {
                cancellationToken.ThrowIfCancellationRequested();
                var now = timeProvider.GetUtcNow();
                lock (_gate)
                {
                        foreach (var id in _jobs.Values
                                .Where(x => x.CompletedAt is { } completed &&
                                        (x.State == JobState.Succeeded && completed < now - succeededRetention ||
                                         x.State is JobState.Failed or JobState.Cancelled && completed < now - failedRetention))
                                .Select(x => x.Id)
                                .ToArray())
                        {
                                _ = _jobs.Remove(id);
                        }
                }

                return ValueTask.CompletedTask;
        }

        /// <inheritdoc />
        public ValueTask HeartbeatAsync(JobServerSnapshot server, CancellationToken cancellationToken = default)
        {
                ArgumentNullException.ThrowIfNull(server);
                cancellationToken.ThrowIfCancellationRequested();
                lock (_gate)
                        _servers[server.WorkerId] = server;
                return ValueTask.CompletedTask;
        }

        /// <inheritdoc />
        public async ValueTask<bool> IsHealthyAsync(CancellationToken cancellationToken = default)
        {
                cancellationToken.ThrowIfCancellationRequested();
                return true;
        }

        private JobRecord GetOwnedActive(string jobId, string workerId)
        {
                if (!_jobs.TryGetValue(jobId, out var job) || job.State != JobState.Active || job.WorkerId != workerId)
                        throw new InvalidOperationException($"Worker '{workerId}' does not own active job '{jobId}'.");
                return job;
        }

        private ValueTask SetRecurringPaused(string name, bool isPaused, CancellationToken cancellationToken)
        {
                cancellationToken.ThrowIfCancellationRequested();
                lock (_gate)
                {
                        if (!_recurring.TryGetValue(name, out var schedule))
                                throw new KeyNotFoundException($"Recurring schedule '{name}' was not found.");

                        _recurring[name] = schedule with { IsPaused = isPaused };
                }

                return ValueTask.CompletedTask;
        }
}

1	namespace Immediate.Jobs.Shared;
2
3	#pragma warning disable IDE0391 // Keep synchronous storage methods async for .NET 11 runtime-async state machines.
4
5	/// <summary>
6	/// A best-effort, non-durable, single-node provider intended for development and tests.
7	/// </summary>
8	public sealed class InMemoryJobStorage(TimeProvider timeProvider) : IJobStorage, IJobStorageReplica	4✔
9	{
10	private readonly Lock _gate = new();	4✔
11	private readonly Dictionary<string, JobRecord> _jobs = new(StringComparer.Ordinal);	4✔
12	private readonly Dictionary<string, RecurringJobSchedule> _recurring = new(StringComparer.Ordinal);	4✔
13	private readonly Dictionary<string, JobServerSnapshot> _servers = new(StringComparer.Ordinal);	4✔
14	private readonly HashSet<string> _recurringKeys = new(StringComparer.Ordinal);	4✔
15
16	/// <inheritdoc />
17	public ValueTask InitializeAsync(CancellationToken cancellationToken = default)
18	{
19	cancellationToken.ThrowIfCancellationRequested();	4✔
20	return ValueTask.CompletedTask;	4✔
21	}
22
23	/// <inheritdoc />
24	public ValueTask EnqueueAsync(JobRecord job, CancellationToken cancellationToken = default)
25	{
26	ArgumentNullException.ThrowIfNull(job);	4✔
27	cancellationToken.ThrowIfCancellationRequested();	4✔
28	lock (_gate)
29	{
30	if (!_jobs.TryAdd(job.Id, job))	4✔
NEW 31	throw new InvalidOperationException($"Job '{job.Id}' already exists.");	×
32	}	4✔
33
34	return ValueTask.CompletedTask;	4✔
35	}
36
37	/// <inheritdoc />
38	public async ValueTask<IReadOnlyList<JobRecord>> AcquireDueJobsAsync(
39	JobAcquisitionRequest request,
40	CancellationToken cancellationToken = default
41	)
42	{
43	ArgumentNullException.ThrowIfNull(request);	4✔
44	cancellationToken.ThrowIfCancellationRequested();	4✔
45	var now = timeProvider.GetUtcNow();	4✔
46	lock (_gate)
47	{
48	foreach (var expired in _jobs.Values.Where(x => x.State == JobState.Active && x.LeaseExpiresAt <= now).ToArray())	4✔
49	{
50	_jobs[expired.Id] = expired with	4✔
51	{	4✔
52	State = JobState.Pending,	4✔
53	WorkerId = null,	4✔
54	LeaseExpiresAt = null,	4✔
55	};	4✔
56	}
57
58	var acquired = new List<JobRecord>(request.BatchSize);	4✔
59	foreach (var queue in request.Queues)	4✔
60	{
61	var queueCapacity = Math.Min(queue.Capacity, request.BatchSize - acquired.Count);	4✔
62	if (queueCapacity <= 0)	4✔
63	continue;
64
65	var jobCapacities = queue.JobCapacities.ToDictionary(static pair => pair.Key, static pair => pair.Value, StringComparer.Ordinal);	4✔
66	foreach (var candidate in _jobs.Values	4✔
67	.Where(job => job.QueueName == queue.QueueName &&	4✔
68	jobCapacities.ContainsKey(job.JobName) &&	4✔
69	job.State is JobState.Pending or JobState.Scheduled && job.DueAt <= now)	4✔
70	.OrderBy(job => job.DueAt)	4✔
71	.ThenBy(job => job.CreatedAt)	4✔
72	.ThenBy(job => job.Id))	4✔
73	{
74	if (queueCapacity == 0)	4✔
75	break;	1✔
76	if (jobCapacities[candidate.JobName] <= 0)	4✔
77	continue;
78
79	var job = candidate with	4✔
80	{	4✔
81	State = JobState.Active,	4✔
82	Attempt = candidate.Attempt + 1,	4✔
83	WorkerId = request.WorkerId,	4✔
84	LeaseExpiresAt = now + request.Lease,	4✔
85	};	4✔
86	_jobs[job.Id] = job;	4✔
87	acquired.Add(job);	4✔
88	jobCapacities[job.JobName]--;	4✔
89	queueCapacity--;	4✔
90	}
91	}
92
93	return acquired;	4✔
94	}
95	}	4✔
96
97	/// <inheritdoc />
98	public async ValueTask<IReadOnlyList<JobRecord>> AcquireJobsAsync(
99	IReadOnlyCollection<string> jobIds,
100	string workerId,
101	TimeSpan lease,
102	CancellationToken cancellationToken = default
103	)
104	{
105	ArgumentNullException.ThrowIfNull(jobIds);	4✔
106	ArgumentException.ThrowIfNullOrWhiteSpace(workerId);	4✔
107	ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(lease, TimeSpan.Zero);	4✔
108	cancellationToken.ThrowIfCancellationRequested();	4✔
109	var now = timeProvider.GetUtcNow();	4✔
110	lock (_gate)
111	{
112	var acquired = new List<JobRecord>(jobIds.Count);	4✔
113	foreach (var id in jobIds)	4✔
114	{
115	if (!_jobs.TryGetValue(id, out var job))	4✔
116	continue;
117	if (job.State == JobState.Active && job.LeaseExpiresAt <= now)	4✔
118	{
119	job = job with { State = JobState.Pending, WorkerId = null, LeaseExpiresAt = null };	4✔
120	_jobs[id] = job;	4✔
121	}
122
123	if (job.State is not (JobState.Pending or JobState.Scheduled) \|\| job.DueAt > now)	4✔
124	continue;
125
126	job = job with	4✔
127	{	4✔
128	State = JobState.Active,	4✔
129	Attempt = job.Attempt + 1,	4✔
130	WorkerId = workerId,	4✔
131	LeaseExpiresAt = now + lease,	4✔
132	};	4✔
133	_jobs[id] = job;	4✔
134	acquired.Add(job);	4✔
135	}
136
137	return acquired;	4✔
138	}
139	}	4✔
140
141	/// <inheritdoc />
142	public ValueTask RenewLeaseAsync(string jobId, string workerId, TimeSpan lease, CancellationToken cancellationToken = default)
143	{
NEW 144	cancellationToken.ThrowIfCancellationRequested();	×
145	lock (_gate)
146	{
NEW 147	var job = GetOwnedActive(jobId, workerId);	×
NEW 148	_jobs[jobId] = job with { LeaseExpiresAt = timeProvider.GetUtcNow() + lease };	×
NEW 149	}	×
150
NEW 151	return ValueTask.CompletedTask;	×
152	}
153
154	/// <inheritdoc />
155	public ValueTask CompleteAsync(string jobId, string workerId, CancellationToken cancellationToken = default)
156	{
157	cancellationToken.ThrowIfCancellationRequested();	4✔
158	lock (_gate)
159	{
160	var job = GetOwnedActive(jobId, workerId);	4✔
161	_jobs[jobId] = job with	4✔
162	{	4✔
163	State = JobState.Succeeded,	4✔
164	WorkerId = null,	4✔
165	LeaseExpiresAt = null,	4✔
166	CompletedAt = timeProvider.GetUtcNow(),	4✔
167	};	4✔
168	}	4✔
169
170	return ValueTask.CompletedTask;	4✔
171	}
172
173	/// <inheritdoc />
174	public ValueTask FailAsync(
175	string jobId,
176	string workerId,
177	string error,
178	DateTimeOffset? nextRetryAt,
179	CancellationToken cancellationToken = default
180	)
181	{
182	cancellationToken.ThrowIfCancellationRequested();	4✔
183	lock (_gate)
184	{
185	var job = GetOwnedActive(jobId, workerId);	4✔
186	_jobs[jobId] = job with	4✔
187	{	4✔
188	State = nextRetryAt.HasValue ? JobState.Scheduled : JobState.Failed,	4✔
189	DueAt = nextRetryAt ?? job.DueAt,	4✔
190	WorkerId = null,	4✔
191	LeaseExpiresAt = null,	4✔
192	LastError = error,	4✔
193	CompletedAt = nextRetryAt.HasValue ? null : timeProvider.GetUtcNow(),	4✔
194	};	4✔
195	}	4✔
196
197	return ValueTask.CompletedTask;	4✔
198	}
199
200	/// <inheritdoc />
201	public ValueTask UpsertRecurringAsync(RecurringJobSchedule schedule, CancellationToken cancellationToken = default)
202	{
203	ArgumentNullException.ThrowIfNull(schedule);	4✔
204	cancellationToken.ThrowIfCancellationRequested();	4✔
205	lock (_gate)
206	{
207	if (_recurring.TryGetValue(schedule.Name, out var current))	4✔
208	{
209	if (current.IsCodeDefined && !schedule.IsCodeDefined)	4✔
210	throw new InvalidOperationException("Code-defined recurring schedules cannot be replaced by dynamic schedules.");	4✔
211
212	schedule = schedule with { IsPaused = current.IsPaused, LastRunAt = current.LastRunAt };	4✔
213	}
214
215	_recurring[schedule.Name] = schedule;	4✔
216	}	4✔
217
218	return ValueTask.CompletedTask;	4✔
219	}
220
221	/// <inheritdoc />
222	public ValueTask RemoveObsoleteCodeDefinedRecurringAsync(
223	IReadOnlyCollection<string> activeScheduleNames,
224	CancellationToken cancellationToken = default
225	)
226	{
227	ArgumentNullException.ThrowIfNull(activeScheduleNames);	4✔
228	cancellationToken.ThrowIfCancellationRequested();	4✔
229	var activeNames = activeScheduleNames.ToHashSet(StringComparer.Ordinal);	4✔
230	lock (_gate)
231	{
232	var obsoleteNames = _recurring	4✔
233	.Where(schedule => schedule.Value.IsCodeDefined && !activeNames.Contains(schedule.Key))	4✔
234	.Select(static schedule => schedule.Key)	4✔
235	.ToArray();	4✔
236	foreach (var name in obsoleteNames)	4✔
237	_ = _recurring.Remove(name);	4✔
238	}	4✔
239
240	return ValueTask.CompletedTask;	4✔
241	}
242
243	/// <inheritdoc />
244	public ValueTask RemoveRecurringAsync(string name, CancellationToken cancellationToken = default)
245	{
NEW 246	cancellationToken.ThrowIfCancellationRequested();	×
247	lock (_gate)
248	{
NEW 249	if (_recurring.TryGetValue(name, out var schedule) && schedule.IsCodeDefined)	×
NEW 250	throw new InvalidOperationException("Code-defined recurring schedules cannot be deleted.");	×
251
NEW 252	_ = _recurring.Remove(name);	×
NEW 253	}	×
254
NEW 255	return ValueTask.CompletedTask;	×
256	}
257
258	/// <inheritdoc />
259	public ValueTask PauseRecurringAsync(string name, CancellationToken cancellationToken = default) =>
NEW 260	SetRecurringPaused(name, true, cancellationToken);	×
261
262	/// <inheritdoc />
263	public ValueTask ResumeRecurringAsync(string name, CancellationToken cancellationToken = default) =>
NEW 264	SetRecurringPaused(name, false, cancellationToken);	×
265
266	/// <inheritdoc />
267	public async ValueTask<IReadOnlyList<RecurringJobSchedule>> GetDueRecurringAsync(
268	DateTimeOffset now,
269	int batchSize,
270	CancellationToken cancellationToken = default
271	)
272	{
273	cancellationToken.ThrowIfCancellationRequested();	4✔
274	lock (_gate)
275	{
276	return	4✔
277	[	4✔
278	.. _recurring.Values.Where(x => !x.IsPaused && x.NextRunAt <= now).OrderBy(x => x.NextRunAt).Take(batchSize),	4✔
279	];	4✔
280	}
281	}	4✔
282
283	/// <inheritdoc />
284	public async ValueTask<bool> MaterializeRecurringAsync(
285	RecurringJobSchedule schedule,
286	JobRecord job,
287	DateTimeOffset nextRunAt,
288	CancellationToken cancellationToken = default
289	)
290	{
291	ArgumentNullException.ThrowIfNull(schedule);	4✔
292	ArgumentNullException.ThrowIfNull(job);	4✔
293	cancellationToken.ThrowIfCancellationRequested();	4✔
294	lock (_gate)
295	{
296	if (!_recurring.TryGetValue(schedule.Name, out var current) \|\| current.NextRunAt != schedule.NextRunAt)	4✔
NEW 297	return false;	×
298
299	var inserted = job.RecurringKey is null \|\| _recurringKeys.Add(job.RecurringKey);	4✔
300	if (inserted)	4✔
301	_jobs[job.Id] = job;	4✔
302	_recurring[schedule.Name] = current with { LastRunAt = schedule.NextRunAt, NextRunAt = nextRunAt };	4✔
303	return inserted;	4✔
304	}
305	}	4✔
306
307	/// <inheritdoc />
308	public async ValueTask<JobMonitoringSnapshot> GetMonitoringSnapshotAsync(CancellationToken cancellationToken = default)
309	{
310	cancellationToken.ThrowIfCancellationRequested();	4✔
311	lock (_gate)
312	{
313	var counts = Enum.GetValues<JobState>().ToDictionary(state => state, state => _jobs.Values.LongCount(x => x.State == state));	4✔
314	var cutoff = timeProvider.GetUtcNow() - TimeSpan.FromMinutes(2);	4✔
315	IReadOnlyList<JobServerSnapshot> servers = [.. _servers.Values.Where(x => x.LastHeartbeat >= cutoff)];	4✔
316	return new(timeProvider.GetUtcNow(), counts, [.. _recurring.Values], servers);	4✔
317	}
318	}	4✔
319
320	/// <inheritdoc />
321	public async ValueTask<IReadOnlyList<JobRecord>> QueryJobsAsync(JobQuery query, CancellationToken cancellationToken = default)
322	{
323	ArgumentNullException.ThrowIfNull(query);	4✔
324	cancellationToken.ThrowIfCancellationRequested();	4✔
325	lock (_gate)
326	{
327	var jobs = _jobs.Values.AsEnumerable();	4✔
328	if (query.Id is { } id)	4✔
329	jobs = jobs.Where(x => x.Id == id);	4✔
330	if (query.State is { } state)	4✔
331	jobs = jobs.Where(x => x.State == state);	4✔
332	if (!string.IsNullOrWhiteSpace(query.QueueName))	4✔
333	jobs = jobs.Where(x => x.QueueName == query.QueueName);	4✔
334
335	if (!string.IsNullOrWhiteSpace(query.Search))	4✔
336	jobs = jobs.Where(x => x.JobName.Contains(query.Search, StringComparison.OrdinalIgnoreCase));	3✔
337
338	return	4✔
339	[	4✔
340	.. jobs.OrderByDescending(x => x.CreatedAt).Skip(Math.Max(0, query.Skip)).Take(Math.Clamp(query.Take, 1, 1000)),	4✔
341	];	4✔
342	}
343	}	4✔
344
345	/// <inheritdoc />
346	public ValueTask RetryAsync(string jobId, CancellationToken cancellationToken = default)
347	{
NEW 348	cancellationToken.ThrowIfCancellationRequested();	×
349	lock (_gate)
350	{
NEW 351	if (!_jobs.TryGetValue(jobId, out var job) \|\| job.State != JobState.Failed)	×
NEW 352	throw new InvalidOperationException("Only failed jobs can be retried.");	×
353
NEW 354	_jobs[jobId] = job with { State = JobState.Pending, DueAt = timeProvider.GetUtcNow(), CompletedAt = null };	×
NEW 355	}	×
356
NEW 357	return ValueTask.CompletedTask;	×
358	}
359
360	/// <inheritdoc />
361	public ValueTask DeleteAsync(string jobId, CancellationToken cancellationToken = default)
362	{
NEW 363	cancellationToken.ThrowIfCancellationRequested();	×
364	lock (_gate)
365	{
NEW 366	if (_jobs.TryGetValue(jobId, out var job) && job.State is JobState.Active or JobState.Pending or JobState.Scheduled)	×
NEW 367	throw new InvalidOperationException("Only terminal jobs can be deleted.");	×
368
NEW 369	_ = _jobs.Remove(jobId);	×
NEW 370	}	×
371
NEW 372	return ValueTask.CompletedTask;	×
373	}
374
375	/// <inheritdoc />
376	public ValueTask PurgeAsync(TimeSpan succeededRetention, TimeSpan failedRetention, CancellationToken cancellationToken = default)
377	{
378	cancellationToken.ThrowIfCancellationRequested();	4✔
379	var now = timeProvider.GetUtcNow();	4✔
380	lock (_gate)
381	{
382	foreach (var id in _jobs.Values	4✔
383	.Where(x => x.CompletedAt is { } completed &&	4✔
384	(x.State == JobState.Succeeded && completed < now - succeededRetention \|\|	4✔
385	x.State is JobState.Failed or JobState.Cancelled && completed < now - failedRetention))	4✔
NEW 386	.Select(x => x.Id)	×
387	.ToArray())	4✔
388	{
NEW 389	_ = _jobs.Remove(id);	×
390	}
391	}	4✔
392
393	return ValueTask.CompletedTask;	4✔
394	}
395
396	/// <inheritdoc />
397	public ValueTask HeartbeatAsync(JobServerSnapshot server, CancellationToken cancellationToken = default)
398	{
399	ArgumentNullException.ThrowIfNull(server);	4✔
400	cancellationToken.ThrowIfCancellationRequested();	4✔
401	lock (_gate)
402	_servers[server.WorkerId] = server;	4✔
403	return ValueTask.CompletedTask;	4✔
404	}
405
406	/// <inheritdoc />
407	public async ValueTask<bool> IsHealthyAsync(CancellationToken cancellationToken = default)
408	{
NEW 409	cancellationToken.ThrowIfCancellationRequested();	×
NEW 410	return true;	×
411	}
412
413	private JobRecord GetOwnedActive(string jobId, string workerId)
414	{
415	if (!_jobs.TryGetValue(jobId, out var job) \|\| job.State != JobState.Active \|\| job.WorkerId != workerId)	4✔
NEW 416	throw new InvalidOperationException($"Worker '{workerId}' does not own active job '{jobId}'.");	×
417	return job;	4✔
418	}
419
420	private ValueTask SetRecurringPaused(string name, bool isPaused, CancellationToken cancellationToken)
421	{
NEW 422	cancellationToken.ThrowIfCancellationRequested();	×
423	lock (_gate)
424	{
NEW 425	if (!_recurring.TryGetValue(name, out var schedule))	×
NEW 426	throw new KeyNotFoundException($"Recurring schedule '{name}' was not found.");	×
427
NEW 428	_recurring[name] = schedule with { IsPaused = isPaused };	×
NEW 429	}	×
430
NEW 431	return ValueTask.CompletedTask;	×
432	}
433	}

ImmediatePlatform / Immediate.Jobs / 30300691995

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