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

55.48

/src/Immediate.Jobs.Shared/SingleServerJobStorage.cs

namespace Immediate.Jobs.Shared;

/// <summary>
/// A single-server storage topology that executes against an authoritative in-process store while
/// synchronously replicating changes to durable storage and restoring them when the process starts.
/// </summary>
public sealed class SingleServerJobStorage : IJobStorage, IAsyncDisposable, IDisposable
{
        private const int RecoveryBatchSize = 1000;
        private readonly TimeProvider _timeProvider;
        private readonly SemaphoreSlim _initialization = new(1, 1);
        private InMemoryJobStorage _primary;
        private bool _initialized;
        private bool _disposed;

        /// <summary>Creates a memory-primary store backed by the supplied durable replica.</summary>
        public SingleServerJobStorage(IJobStorage durableStorage, TimeProvider timeProvider)
        {
                ArgumentNullException.ThrowIfNull(durableStorage);
                ArgumentNullException.ThrowIfNull(timeProvider);
                if (durableStorage is SingleServerJobStorage)
                        throw new ArgumentException("A single-server store cannot be used as its own durable replica.", nameof(durableStorage));
                if (durableStorage is not IJobStorageReplica)
                        throw new ArgumentException("Single-server durable storage must implement IJobStorageReplica.", nameof(durableStorage));

                DurableStorage = durableStorage;
                _timeProvider = timeProvider;
                _primary = new(timeProvider);
        }

        /// <summary>The in-process authoritative store.</summary>
        public IJobStorage PrimaryStorage => _primary;

        /// <summary>The durable write-through replica.</summary>
        public IJobStorage DurableStorage { get; }

        /// <inheritdoc />
        public ValueTask InitializeAsync(CancellationToken cancellationToken = default) => EnsureInitializedAsync(cancellationToken);

        /// <inheritdoc />
        public async ValueTask EnqueueAsync(JobRecord job, CancellationToken cancellationToken = default)
        {
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
                await DurableStorage.EnqueueAsync(job, cancellationToken).ConfigureAwait(false);
                await _primary.EnqueueAsync(job, cancellationToken).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public async ValueTask<IReadOnlyList<JobRecord>> AcquireDueJobsAsync(
                JobAcquisitionRequest request,
                CancellationToken cancellationToken = default
        )
        {
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
                var acquired = await _primary.AcquireDueJobsAsync(request, cancellationToken).ConfigureAwait(false);
                if (acquired.Count == 0)
                        return acquired;

                var replica = (IJobStorageReplica)DurableStorage;
                var replicated = await replica.AcquireJobsAsync(
                        [.. acquired.Select(x => x.Id)],
                        request.WorkerId,
                        request.Lease,
                        cancellationToken
                ).ConfigureAwait(false);
                if (acquired.Count != replicated.Count ||
                        !acquired.Select(x => x.Id).ToHashSet(StringComparer.Ordinal).SetEquals(replicated.Select(x => x.Id)))
                {
                        throw new InvalidOperationException(
                                "The durable job replica has drifted from the authoritative in-memory queue. " +
                                "Single-server mode must not be used by multiple scheduler processes."
                        );
                }

                return acquired;
        }

        /// <inheritdoc />
        public async ValueTask RenewLeaseAsync(
                string jobId,
                string workerId,
                TimeSpan lease,
                CancellationToken cancellationToken = default
        )
        {
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
                await DurableStorage.RenewLeaseAsync(jobId, workerId, lease, cancellationToken).ConfigureAwait(false);
                await _primary.RenewLeaseAsync(jobId, workerId, lease, cancellationToken).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public async ValueTask CompleteAsync(string jobId, string workerId, CancellationToken cancellationToken = default)
        {
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
                await DurableStorage.CompleteAsync(jobId, workerId, cancellationToken).ConfigureAwait(false);
                await _primary.CompleteAsync(jobId, workerId, cancellationToken).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public async ValueTask FailAsync(
                string jobId,
                string workerId,
                string error,
                DateTimeOffset? nextRetryAt,
                CancellationToken cancellationToken = default
        )
        {
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
                await DurableStorage.FailAsync(jobId, workerId, error, nextRetryAt, cancellationToken).ConfigureAwait(false);
                await _primary.FailAsync(jobId, workerId, error, nextRetryAt, cancellationToken).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public async ValueTask UpsertRecurringAsync(RecurringJobSchedule schedule, CancellationToken cancellationToken = default)
        {
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
                await DurableStorage.UpsertRecurringAsync(schedule, cancellationToken).ConfigureAwait(false);
                await _primary.UpsertRecurringAsync(schedule, cancellationToken).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public async ValueTask RemoveObsoleteCodeDefinedRecurringAsync(
                IReadOnlyCollection<string> activeScheduleNames,
                CancellationToken cancellationToken = default
        )
        {
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
                await DurableStorage.RemoveObsoleteCodeDefinedRecurringAsync(activeScheduleNames, cancellationToken)
                        .ConfigureAwait(false);
                await _primary.RemoveObsoleteCodeDefinedRecurringAsync(activeScheduleNames, cancellationToken).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public async ValueTask RemoveRecurringAsync(string name, CancellationToken cancellationToken = default)
        {
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
                await DurableStorage.RemoveRecurringAsync(name, cancellationToken).ConfigureAwait(false);
                await _primary.RemoveRecurringAsync(name, cancellationToken).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public async ValueTask PauseRecurringAsync(string name, CancellationToken cancellationToken = default)
        {
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
                await DurableStorage.PauseRecurringAsync(name, cancellationToken).ConfigureAwait(false);
                await _primary.PauseRecurringAsync(name, cancellationToken).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public async ValueTask ResumeRecurringAsync(string name, CancellationToken cancellationToken = default)
        {
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
                await DurableStorage.ResumeRecurringAsync(name, cancellationToken).ConfigureAwait(false);
                await _primary.ResumeRecurringAsync(name, cancellationToken).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public async ValueTask<IReadOnlyList<RecurringJobSchedule>> GetDueRecurringAsync(
                DateTimeOffset now,
                int batchSize,
                CancellationToken cancellationToken = default
        )
        {
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
                return await _primary.GetDueRecurringAsync(now, batchSize, cancellationToken).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public async ValueTask<bool> MaterializeRecurringAsync(
                RecurringJobSchedule schedule,
                JobRecord job,
                DateTimeOffset nextRunAt,
                CancellationToken cancellationToken = default
        )
        {
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
                if (!await DurableStorage.MaterializeRecurringAsync(schedule, job, nextRunAt, cancellationToken).ConfigureAwait(false))
                        return false;
                return await _primary.MaterializeRecurringAsync(schedule, job, nextRunAt, cancellationToken).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public async ValueTask<JobMonitoringSnapshot> GetMonitoringSnapshotAsync(CancellationToken cancellationToken = default)
        {
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
                return await _primary.GetMonitoringSnapshotAsync(cancellationToken).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public async ValueTask<IReadOnlyList<JobRecord>> QueryJobsAsync(
                JobQuery query,
                CancellationToken cancellationToken = default
        )
        {
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
                return await _primary.QueryJobsAsync(query, cancellationToken).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public async ValueTask RetryAsync(string jobId, CancellationToken cancellationToken = default)
        {
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
                await DurableStorage.RetryAsync(jobId, cancellationToken).ConfigureAwait(false);
                await _primary.RetryAsync(jobId, cancellationToken).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public async ValueTask DeleteAsync(string jobId, CancellationToken cancellationToken = default)
        {
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
                await DurableStorage.DeleteAsync(jobId, cancellationToken).ConfigureAwait(false);
                await _primary.DeleteAsync(jobId, cancellationToken).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public async ValueTask PurgeAsync(
                TimeSpan succeededRetention,
                TimeSpan failedRetention,
                CancellationToken cancellationToken = default
        )
        {
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
                await DurableStorage.PurgeAsync(succeededRetention, failedRetention, cancellationToken).ConfigureAwait(false);
                await _primary.PurgeAsync(succeededRetention, failedRetention, cancellationToken).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public async ValueTask HeartbeatAsync(JobServerSnapshot server, CancellationToken cancellationToken = default)
        {
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
                await _primary.HeartbeatAsync(server, cancellationToken).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public async ValueTask<bool> IsHealthyAsync(CancellationToken cancellationToken = default)
        {
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
                return await DurableStorage.IsHealthyAsync(cancellationToken).ConfigureAwait(false);
        }

        /// <inheritdoc />
        public void Dispose()
        {
                if (_disposed)
                        return;
                _disposed = true;
                if (DurableStorage is IDisposable disposable)
                        disposable.Dispose();
                else if (DurableStorage is IAsyncDisposable asyncDisposable)
                        asyncDisposable.DisposeAsync().AsTask().GetAwaiter().GetResult();
                _initialization.Dispose();
        }

        /// <inheritdoc />
        public async ValueTask DisposeAsync()
        {
                if (_disposed)
                        return;
                _disposed = true;
                if (DurableStorage is IAsyncDisposable asyncDisposable)
                        await asyncDisposable.DisposeAsync().ConfigureAwait(false);
                else if (DurableStorage is IDisposable disposable)
                        disposable.Dispose();
                _initialization.Dispose();
        }

        private async ValueTask EnsureInitializedAsync(CancellationToken cancellationToken)
        {
                ObjectDisposedException.ThrowIf(_disposed, this);
                if (Volatile.Read(ref _initialized))
                        return;

                await _initialization.WaitAsync(cancellationToken).ConfigureAwait(false);
                try
                {
                        if (Volatile.Read(ref _initialized))
                                return;

                        await DurableStorage.InitializeAsync(cancellationToken).ConfigureAwait(false);
                        var recoveredPrimary = new InMemoryJobStorage(_timeProvider);
                        await recoveredPrimary.InitializeAsync(cancellationToken).ConfigureAwait(false);

                        foreach (var state in Enum.GetValues<JobState>())
                        {
                                var skip = 0;
                                while (true)
                                {
                                        var jobs = await DurableStorage.QueryJobsAsync(
                                                new() { State = state, Skip = skip, Take = RecoveryBatchSize },
                                                cancellationToken
                                        ).ConfigureAwait(false);
                                        foreach (var job in jobs)
                                                await recoveredPrimary.EnqueueAsync(job, cancellationToken).ConfigureAwait(false);
                                        if (jobs.Count < RecoveryBatchSize)
                                                break;
                                        skip += jobs.Count;
                                }
                        }

                        var snapshot = await DurableStorage.GetMonitoringSnapshotAsync(cancellationToken).ConfigureAwait(false);
                        foreach (var schedule in snapshot.Recurring)
                                await recoveredPrimary.UpsertRecurringAsync(schedule, cancellationToken).ConfigureAwait(false);

                        _primary = recoveredPrimary;
                        Volatile.Write(ref _initialized, true);
                }
                finally
                {
                        _ = _initialization.Release();
                }
        }
}

1	namespace Immediate.Jobs.Shared;
2
3	/// <summary>
4	/// A single-server storage topology that executes against an authoritative in-process store while
5	/// synchronously replicating changes to durable storage and restoring them when the process starts.
6	/// </summary>
7	public sealed class SingleServerJobStorage : IJobStorage, IAsyncDisposable, IDisposable
8	{
9	private const int RecoveryBatchSize = 1000;
10	private readonly TimeProvider _timeProvider;
11	private readonly SemaphoreSlim _initialization = new(1, 1);	4✔
12	private InMemoryJobStorage _primary;
13	private bool _initialized;
14	private bool _disposed;
15
16	/// <summary>Creates a memory-primary store backed by the supplied durable replica.</summary>
17	public SingleServerJobStorage(IJobStorage durableStorage, TimeProvider timeProvider)	4✔
18	{
19	ArgumentNullException.ThrowIfNull(durableStorage);	4✔
20	ArgumentNullException.ThrowIfNull(timeProvider);	4✔
21	if (durableStorage is SingleServerJobStorage)	4✔
NEW 22	throw new ArgumentException("A single-server store cannot be used as its own durable replica.", nameof(durableStorage));	×
23	if (durableStorage is not IJobStorageReplica)	4✔
NEW 24	throw new ArgumentException("Single-server durable storage must implement IJobStorageReplica.", nameof(durableStorage));	×
25
26	DurableStorage = durableStorage;	4✔
27	_timeProvider = timeProvider;	4✔
28	_primary = new(timeProvider);	4✔
29	}	4✔
30
31	/// <summary>The in-process authoritative store.</summary>
32	public IJobStorage PrimaryStorage => _primary;	4✔
33
34	/// <summary>The durable write-through replica.</summary>
35	public IJobStorage DurableStorage { get; }
36
37	/// <inheritdoc />
38	public ValueTask InitializeAsync(CancellationToken cancellationToken = default) => EnsureInitializedAsync(cancellationToken);	4✔
39
40	/// <inheritdoc />
41	public async ValueTask EnqueueAsync(JobRecord job, CancellationToken cancellationToken = default)
42	{
43	await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);	4✔
44	await DurableStorage.EnqueueAsync(job, cancellationToken).ConfigureAwait(false);	4✔
45	await _primary.EnqueueAsync(job, cancellationToken).ConfigureAwait(false);	4✔
46	}	4✔
47
48	/// <inheritdoc />
49	public async ValueTask<IReadOnlyList<JobRecord>> AcquireDueJobsAsync(
50	JobAcquisitionRequest request,
51	CancellationToken cancellationToken = default
52	)
53	{
54	await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);	4✔
55	var acquired = await _primary.AcquireDueJobsAsync(request, cancellationToken).ConfigureAwait(false);	4✔
56	if (acquired.Count == 0)	4✔
57	return acquired;	4✔
58
59	var replica = (IJobStorageReplica)DurableStorage;	4✔
60	var replicated = await replica.AcquireJobsAsync(	4✔
61	[.. acquired.Select(x => x.Id)],	4✔
62	request.WorkerId,	4✔
63	request.Lease,	4✔
64	cancellationToken	4✔
65	).ConfigureAwait(false);	4✔
66	if (acquired.Count != replicated.Count \|\|	4✔
67	!acquired.Select(x => x.Id).ToHashSet(StringComparer.Ordinal).SetEquals(replicated.Select(x => x.Id)))	4✔
68	{
NEW 69	throw new InvalidOperationException(	×
NEW 70	"The durable job replica has drifted from the authoritative in-memory queue. " +	×
NEW 71	"Single-server mode must not be used by multiple scheduler processes."	×
NEW 72	);	×
73	}
74
75	return acquired;	4✔
76	}	3✔
77
78	/// <inheritdoc />
79	public async ValueTask RenewLeaseAsync(
80	string jobId,
81	string workerId,
82	TimeSpan lease,
83	CancellationToken cancellationToken = default
84	)
85	{
NEW 86	await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);	×
NEW 87	await DurableStorage.RenewLeaseAsync(jobId, workerId, lease, cancellationToken).ConfigureAwait(false);	×
NEW 88	await _primary.RenewLeaseAsync(jobId, workerId, lease, cancellationToken).ConfigureAwait(false);	×
NEW 89	}	×
90
91	/// <inheritdoc />
92	public async ValueTask CompleteAsync(string jobId, string workerId, CancellationToken cancellationToken = default)
93	{
NEW 94	await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);	×
NEW 95	await DurableStorage.CompleteAsync(jobId, workerId, cancellationToken).ConfigureAwait(false);	×
NEW 96	await _primary.CompleteAsync(jobId, workerId, cancellationToken).ConfigureAwait(false);	×
NEW 97	}	×
98
99	/// <inheritdoc />
100	public async ValueTask FailAsync(
101	string jobId,
102	string workerId,
103	string error,
104	DateTimeOffset? nextRetryAt,
105	CancellationToken cancellationToken = default
106	)
107	{
NEW 108	await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);	×
NEW 109	await DurableStorage.FailAsync(jobId, workerId, error, nextRetryAt, cancellationToken).ConfigureAwait(false);	×
NEW 110	await _primary.FailAsync(jobId, workerId, error, nextRetryAt, cancellationToken).ConfigureAwait(false);	×
NEW 111	}	×
112
113	/// <inheritdoc />
114	public async ValueTask UpsertRecurringAsync(RecurringJobSchedule schedule, CancellationToken cancellationToken = default)
115	{
116	await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);	4✔
117	await DurableStorage.UpsertRecurringAsync(schedule, cancellationToken).ConfigureAwait(false);	4✔
118	await _primary.UpsertRecurringAsync(schedule, cancellationToken).ConfigureAwait(false);	4✔
119	}	4✔
120
121	/// <inheritdoc />
122	public async ValueTask RemoveObsoleteCodeDefinedRecurringAsync(
123	IReadOnlyCollection<string> activeScheduleNames,
124	CancellationToken cancellationToken = default
125	)
126	{
127	await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);	4✔
128	await DurableStorage.RemoveObsoleteCodeDefinedRecurringAsync(activeScheduleNames, cancellationToken)	4✔
129	.ConfigureAwait(false);	4✔
130	await _primary.RemoveObsoleteCodeDefinedRecurringAsync(activeScheduleNames, cancellationToken).ConfigureAwait(false);	4✔
131	}	4✔
132
133	/// <inheritdoc />
134	public async ValueTask RemoveRecurringAsync(string name, CancellationToken cancellationToken = default)
135	{
NEW 136	await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);	×
NEW 137	await DurableStorage.RemoveRecurringAsync(name, cancellationToken).ConfigureAwait(false);	×
NEW 138	await _primary.RemoveRecurringAsync(name, cancellationToken).ConfigureAwait(false);	×
NEW 139	}	×
140
141	/// <inheritdoc />
142	public async ValueTask PauseRecurringAsync(string name, CancellationToken cancellationToken = default)
143	{
NEW 144	await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);	×
NEW 145	await DurableStorage.PauseRecurringAsync(name, cancellationToken).ConfigureAwait(false);	×
NEW 146	await _primary.PauseRecurringAsync(name, cancellationToken).ConfigureAwait(false);	×
NEW 147	}	×
148
149	/// <inheritdoc />
150	public async ValueTask ResumeRecurringAsync(string name, CancellationToken cancellationToken = default)
151	{
NEW 152	await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);	×
NEW 153	await DurableStorage.ResumeRecurringAsync(name, cancellationToken).ConfigureAwait(false);	×
NEW 154	await _primary.ResumeRecurringAsync(name, cancellationToken).ConfigureAwait(false);	×
NEW 155	}	×
156
157	/// <inheritdoc />
158	public async ValueTask<IReadOnlyList<RecurringJobSchedule>> GetDueRecurringAsync(
159	DateTimeOffset now,
160	int batchSize,
161	CancellationToken cancellationToken = default
162	)
163	{
NEW 164	await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);	×
NEW 165	return await _primary.GetDueRecurringAsync(now, batchSize, cancellationToken).ConfigureAwait(false);	×
166	}
167
168	/// <inheritdoc />
169	public async ValueTask<bool> MaterializeRecurringAsync(
170	RecurringJobSchedule schedule,
171	JobRecord job,
172	DateTimeOffset nextRunAt,
173	CancellationToken cancellationToken = default
174	)
175	{
NEW 176	await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);	×
NEW 177	if (!await DurableStorage.MaterializeRecurringAsync(schedule, job, nextRunAt, cancellationToken).ConfigureAwait(false))	×
NEW 178	return false;	×
NEW 179	return await _primary.MaterializeRecurringAsync(schedule, job, nextRunAt, cancellationToken).ConfigureAwait(false);	×
180	}
181
182	/// <inheritdoc />
183	public async ValueTask<JobMonitoringSnapshot> GetMonitoringSnapshotAsync(CancellationToken cancellationToken = default)
184	{
185	await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);	4✔
186	return await _primary.GetMonitoringSnapshotAsync(cancellationToken).ConfigureAwait(false);	4✔
187	}	3✔
188
189	/// <inheritdoc />
190	public async ValueTask<IReadOnlyList<JobRecord>> QueryJobsAsync(
191	JobQuery query,
192	CancellationToken cancellationToken = default
193	)
194	{
195	await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);	4✔
196	return await _primary.QueryJobsAsync(query, cancellationToken).ConfigureAwait(false);	4✔
197	}	3✔
198
199	/// <inheritdoc />
200	public async ValueTask RetryAsync(string jobId, CancellationToken cancellationToken = default)
201	{
NEW 202	await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);	×
NEW 203	await DurableStorage.RetryAsync(jobId, cancellationToken).ConfigureAwait(false);	×
NEW 204	await _primary.RetryAsync(jobId, cancellationToken).ConfigureAwait(false);	×
NEW 205	}	×
206
207	/// <inheritdoc />
208	public async ValueTask DeleteAsync(string jobId, CancellationToken cancellationToken = default)
209	{
NEW 210	await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);	×
NEW 211	await DurableStorage.DeleteAsync(jobId, cancellationToken).ConfigureAwait(false);	×
NEW 212	await _primary.DeleteAsync(jobId, cancellationToken).ConfigureAwait(false);	×
NEW 213	}	×
214
215	/// <inheritdoc />
216	public async ValueTask PurgeAsync(
217	TimeSpan succeededRetention,
218	TimeSpan failedRetention,
219	CancellationToken cancellationToken = default
220	)
221	{
NEW 222	await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);	×
NEW 223	await DurableStorage.PurgeAsync(succeededRetention, failedRetention, cancellationToken).ConfigureAwait(false);	×
NEW 224	await _primary.PurgeAsync(succeededRetention, failedRetention, cancellationToken).ConfigureAwait(false);	×
NEW 225	}	×
226
227	/// <inheritdoc />
228	public async ValueTask HeartbeatAsync(JobServerSnapshot server, CancellationToken cancellationToken = default)
229	{
230	await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);	4✔
231	await _primary.HeartbeatAsync(server, cancellationToken).ConfigureAwait(false);	4✔
232	}	4✔
233
234	/// <inheritdoc />
235	public async ValueTask<bool> IsHealthyAsync(CancellationToken cancellationToken = default)
236	{
NEW 237	await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);	×
NEW 238	return await DurableStorage.IsHealthyAsync(cancellationToken).ConfigureAwait(false);	×
239	}
240
241	/// <inheritdoc />
242	public void Dispose()
243	{
244	if (_disposed)	4✔
NEW 245	return;	×
246	_disposed = true;	4✔
247	if (DurableStorage is IDisposable disposable)	4✔
NEW 248	disposable.Dispose();	×
249	else if (DurableStorage is IAsyncDisposable asyncDisposable)	4✔
NEW 250	asyncDisposable.DisposeAsync().AsTask().GetAwaiter().GetResult();	×
251	_initialization.Dispose();	4✔
252	}	4✔
253
254	/// <inheritdoc />
255	public async ValueTask DisposeAsync()
256	{
NEW 257	if (_disposed)	×
NEW 258	return;	×
NEW 259	_disposed = true;	×
NEW 260	if (DurableStorage is IAsyncDisposable asyncDisposable)	×
NEW 261	await asyncDisposable.DisposeAsync().ConfigureAwait(false);	×
NEW 262	else if (DurableStorage is IDisposable disposable)	×
NEW 263	disposable.Dispose();	×
NEW 264	_initialization.Dispose();	×
NEW 265	}	×
266
267	private async ValueTask EnsureInitializedAsync(CancellationToken cancellationToken)
268	{
269	ObjectDisposedException.ThrowIf(_disposed, this);	4✔
270	if (Volatile.Read(ref _initialized))	4✔
271	return;	4✔
272
273	await _initialization.WaitAsync(cancellationToken).ConfigureAwait(false);	4✔
274	try
275	{
276	if (Volatile.Read(ref _initialized))	4✔
NEW 277	return;	×
278
279	await DurableStorage.InitializeAsync(cancellationToken).ConfigureAwait(false);	4✔
280	var recoveredPrimary = new InMemoryJobStorage(_timeProvider);	4✔
281	await recoveredPrimary.InitializeAsync(cancellationToken).ConfigureAwait(false);	4✔
282
283	foreach (var state in Enum.GetValues<JobState>())	4✔
284	{
285	var skip = 0;	4✔
NEW 286	while (true)	×
287	{
288	var jobs = await DurableStorage.QueryJobsAsync(	4✔
289	new() { State = state, Skip = skip, Take = RecoveryBatchSize },	4✔
290	cancellationToken	4✔
291	).ConfigureAwait(false);	4✔
292	foreach (var job in jobs)	4✔
293	await recoveredPrimary.EnqueueAsync(job, cancellationToken).ConfigureAwait(false);	4✔
294	if (jobs.Count < RecoveryBatchSize)	4✔
295	break;
NEW 296	skip += jobs.Count;	×
297	}
298	}
299
300	var snapshot = await DurableStorage.GetMonitoringSnapshotAsync(cancellationToken).ConfigureAwait(false);	4✔
301	foreach (var schedule in snapshot.Recurring)	4✔
302	await recoveredPrimary.UpsertRecurringAsync(schedule, cancellationToken).ConfigureAwait(false);	4✔
303
304	_primary = recoveredPrimary;	4✔
305	Volatile.Write(ref _initialized, true);	4✔
306	}	4✔
307	finally
308	{
309	_ = _initialization.Release();	4✔
310	}	1✔
311	}	4✔
312	}

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