• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

ImmediatePlatform / Immediate.Jobs / 30304771671

27 Jul 2026 08:56PM UTC coverage: 65.995%. First build
30304771671

Pull #33

github

web-flow
Merge 46c3047f1 into e9a147742
Pull Request #33: Improve Storage API

1521 of 2481 new or added lines in 28 files covered. (61.31%)

4326 of 6555 relevant lines covered (66.0%)

2.22 hits per line

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

57.1
/src/Immediate.Jobs.Shared/SingleServerJobStorage.cs
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 :
8
        IRecurringJobStorage,
9
        IJobGraphStorage,
10
        IAsyncDisposable,
11
        IDisposable
12
{
13
        private const int RecoveryBatchSize = 1000;
14
        private readonly TimeProvider _timeProvider;
15
        private readonly SemaphoreSlim _initialization = new(1, 1);
4✔
16
        private readonly IRecurringJobStorage _recurringDurableStorage;
17
        private readonly IJobGraphStorage _graphDurableStorage;
18
        private InMemoryJobStorage _primary;
19
        private bool _initialized;
20
        private bool _disposed;
21

22
        /// <summary>Creates a memory-primary store backed by the supplied durable replica.</summary>
23
        public SingleServerJobStorage(IJobStorage durableStorage, TimeProvider timeProvider)
4✔
24
        {
25
                ArgumentNullException.ThrowIfNull(durableStorage);
4✔
26
                ArgumentNullException.ThrowIfNull(timeProvider);
4✔
27
                if (durableStorage is SingleServerJobStorage)
4✔
28
                        throw new ArgumentException("A single-server store cannot be used as its own durable replica.", nameof(durableStorage));
×
29
                if (durableStorage is not IJobStorageReplica)
4✔
30
                        throw new ArgumentException("Single-server durable storage must implement IJobStorageReplica.", nameof(durableStorage));
×
31
                if (durableStorage is not IRecurringJobStorage recurringDurableStorage)
4✔
NEW
32
                        throw new ArgumentException("Single-server durable storage must support recurring jobs.", nameof(durableStorage));
×
33
                if (durableStorage is not IJobGraphStorage graphDurableStorage)
4✔
NEW
34
                        throw new ArgumentException("Single-server durable storage must support batches and continuations.", nameof(durableStorage));
×
35

36
                DurableStorage = durableStorage;
4✔
37
                _recurringDurableStorage = recurringDurableStorage;
4✔
38
                _graphDurableStorage = graphDurableStorage;
4✔
39
                _timeProvider = timeProvider;
4✔
40
                _primary = new(timeProvider);
4✔
41
        }
4✔
42

43
        /// <summary>The in-process authoritative store.</summary>
44
        public IJobStorage PrimaryStorage => _primary;
4✔
45

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

49
        /// <inheritdoc />
50
        public ValueTask InitializeAsync(CancellationToken cancellationToken = default) => EnsureInitializedAsync(cancellationToken);
4✔
51

52
        /// <inheritdoc />
53
        public async ValueTask EnqueueAsync(JobRecord job, CancellationToken cancellationToken = default)
54
        {
55
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
4✔
56
                await DurableStorage.EnqueueAsync(job, cancellationToken).ConfigureAwait(false);
4✔
57
                await _primary.EnqueueAsync(job, cancellationToken).ConfigureAwait(false);
4✔
58
        }
4✔
59

60
        /// <inheritdoc />
61
        public async ValueTask EnqueueContinuationAsync(
62
                JobRecord job,
63
                IReadOnlyList<JobContinuationEdge> edges,
64
                CancellationToken cancellationToken = default
65
        )
66
        {
67
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
NEW
68
                await _graphDurableStorage.EnqueueContinuationAsync(job, edges, cancellationToken).ConfigureAwait(false);
×
69
                await _primary.EnqueueContinuationAsync(job, edges, cancellationToken).ConfigureAwait(false);
×
70
        }
×
71

72
        /// <inheritdoc />
73
        public async ValueTask EnqueueBatchAsync(
74
                JobBatchRecord batch,
75
                IReadOnlyList<JobRecord> jobs,
76
                IReadOnlyList<JobContinuationEdge> edges,
77
                CancellationToken cancellationToken = default
78
        )
79
        {
80
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
NEW
81
                await _graphDurableStorage.EnqueueBatchAsync(batch, jobs, edges, cancellationToken).ConfigureAwait(false);
×
82
                await _primary.EnqueueBatchAsync(batch, jobs, edges, cancellationToken).ConfigureAwait(false);
×
83
        }
×
84

85
        /// <inheritdoc />
86
        public async ValueTask<IReadOnlyList<JobRecord>> AcquireDueJobsAsync(
87
                JobAcquisitionRequest request,
88
                CancellationToken cancellationToken = default
89
        )
90
        {
91
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
4✔
92
                var acquired = await _primary.AcquireDueJobsAsync(request, cancellationToken).ConfigureAwait(false);
4✔
93
                if (acquired.Count == 0)
4✔
94
                        return acquired;
4✔
95

96
                var replica = (IJobStorageReplica)DurableStorage;
4✔
97
                var replicated = await replica.AcquireJobsAsync(
4✔
98
                        [.. acquired.Select(x => x.Id)],
4✔
99
                        request.WorkerId,
4✔
100
                        request.Lease,
4✔
101
                        cancellationToken
4✔
102
                ).ConfigureAwait(false);
4✔
103
                if (acquired.Count != replicated.Count ||
4✔
104
                        !acquired.Select(x => x.Id).ToHashSet(StringComparer.Ordinal).SetEquals(replicated.Select(x => x.Id)))
4✔
105
                {
NEW
106
                        throw new ImmediateJobException(
×
107
                                "The durable job replica has drifted from the authoritative in-memory queue. " +
×
108
                                "Single-server mode must not be used by multiple scheduler processes."
×
109
                        );
×
110
                }
111

112
                return acquired;
4✔
113
        }
3✔
114

115
        /// <inheritdoc />
116
        public async ValueTask SetExecutionTelemetryAsync(
117
                string jobId,
118
                string workerId,
119
                string? traceId,
120
                string? spanId,
121
                DateTimeOffset startedAt,
122
                CancellationToken cancellationToken = default
123
        )
124
        {
125
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
4✔
126
                await DurableStorage.SetExecutionTelemetryAsync(
4✔
127
                        jobId,
4✔
128
                        workerId,
4✔
129
                        traceId,
4✔
130
                        spanId,
4✔
131
                        startedAt,
4✔
132
                        cancellationToken
4✔
133
                ).ConfigureAwait(false);
4✔
134
                await _primary.SetExecutionTelemetryAsync(
4✔
135
                        jobId,
4✔
136
                        workerId,
4✔
137
                        traceId,
4✔
138
                        spanId,
4✔
139
                        startedAt,
4✔
140
                        cancellationToken
4✔
141
                ).ConfigureAwait(false);
4✔
142
        }
4✔
143

144
        /// <inheritdoc />
145
        public async ValueTask RenewLeaseAsync(
146
                string jobId,
147
                string workerId,
148
                TimeSpan lease,
149
                CancellationToken cancellationToken = default
150
        )
151
        {
152
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
153
                await DurableStorage.RenewLeaseAsync(jobId, workerId, lease, cancellationToken).ConfigureAwait(false);
×
154
                await _primary.RenewLeaseAsync(jobId, workerId, lease, cancellationToken).ConfigureAwait(false);
×
155
        }
×
156

157
        /// <inheritdoc />
158
        public async ValueTask CompleteAsync(string jobId, string workerId, CancellationToken cancellationToken = default)
159
        {
160
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
161
                await DurableStorage.CompleteAsync(jobId, workerId, cancellationToken).ConfigureAwait(false);
×
162
                await _primary.CompleteAsync(jobId, workerId, cancellationToken).ConfigureAwait(false);
×
163
        }
×
164

165
        /// <inheritdoc />
166
        public async ValueTask CompleteWithContinuationsAsync(
167
                string jobId,
168
                string workerId,
169
                IReadOnlyList<JobContinuationAddition> additions,
170
                CancellationToken cancellationToken = default
171
        )
172
        {
173
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
NEW
174
                await _graphDurableStorage.CompleteWithContinuationsAsync(jobId, workerId, additions, cancellationToken)
×
175
                        .ConfigureAwait(false);
×
176
                await _primary.CompleteWithContinuationsAsync(jobId, workerId, additions, cancellationToken)
×
177
                        .ConfigureAwait(false);
×
178
        }
×
179

180
        /// <inheritdoc />
181
        public async ValueTask AddBatchJobAsync(
182
                string currentJobId,
183
                JobRecord job,
184
                ContinuationOptions options,
185
                CancellationToken cancellationToken = default
186
        )
187
        {
188
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
NEW
189
                await _graphDurableStorage.AddBatchJobAsync(currentJobId, job, options, cancellationToken).ConfigureAwait(false);
×
190
                await _primary.AddBatchJobAsync(currentJobId, job, options, cancellationToken).ConfigureAwait(false);
×
191
        }
×
192

193
        /// <inheritdoc />
194
        public async ValueTask FailAsync(
195
                string jobId,
196
                string workerId,
197
                string error,
198
                DateTimeOffset? nextRetryAt,
199
                CancellationToken cancellationToken = default
200
        )
201
        {
202
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
203
                await DurableStorage.FailAsync(jobId, workerId, error, nextRetryAt, cancellationToken).ConfigureAwait(false);
×
204
                await _primary.FailAsync(jobId, workerId, error, nextRetryAt, cancellationToken).ConfigureAwait(false);
×
205
        }
×
206

207
        /// <inheritdoc />
208
        public async ValueTask UpsertRecurringAsync(RecurringJobSchedule schedule, CancellationToken cancellationToken = default)
209
        {
210
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
4✔
211
                await _recurringDurableStorage.UpsertRecurringAsync(schedule, cancellationToken).ConfigureAwait(false);
4✔
212
                await _primary.UpsertRecurringAsync(schedule, cancellationToken).ConfigureAwait(false);
4✔
213
        }
4✔
214

215
        /// <inheritdoc />
216
        public async ValueTask RemoveObsoleteCodeDefinedRecurringAsync(
217
                IReadOnlyCollection<string> activeScheduleNames,
218
                CancellationToken cancellationToken = default
219
        )
220
        {
221
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
4✔
222
                await _recurringDurableStorage.RemoveObsoleteCodeDefinedRecurringAsync(activeScheduleNames, cancellationToken)
4✔
223
                        .ConfigureAwait(false);
4✔
224
                await _primary.RemoveObsoleteCodeDefinedRecurringAsync(activeScheduleNames, cancellationToken).ConfigureAwait(false);
4✔
225
        }
4✔
226

227
        /// <inheritdoc />
228
        public async ValueTask RemoveRecurringAsync(string name, CancellationToken cancellationToken = default)
229
        {
230
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
NEW
231
                await _recurringDurableStorage.RemoveRecurringAsync(name, cancellationToken).ConfigureAwait(false);
×
232
                await _primary.RemoveRecurringAsync(name, cancellationToken).ConfigureAwait(false);
×
233
        }
×
234

235
        /// <inheritdoc />
236
        public async ValueTask PauseRecurringAsync(string name, CancellationToken cancellationToken = default)
237
        {
238
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
NEW
239
                await _recurringDurableStorage.PauseRecurringAsync(name, cancellationToken).ConfigureAwait(false);
×
240
                await _primary.PauseRecurringAsync(name, cancellationToken).ConfigureAwait(false);
×
241
        }
×
242

243
        /// <inheritdoc />
244
        public async ValueTask ResumeRecurringAsync(string name, CancellationToken cancellationToken = default)
245
        {
246
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
NEW
247
                await _recurringDurableStorage.ResumeRecurringAsync(name, cancellationToken).ConfigureAwait(false);
×
248
                await _primary.ResumeRecurringAsync(name, cancellationToken).ConfigureAwait(false);
×
249
        }
×
250

251
        /// <inheritdoc />
252
        public async ValueTask<IReadOnlyList<RecurringJobSchedule>> GetDueRecurringAsync(
253
                DateTimeOffset now,
254
                int batchSize,
255
                CancellationToken cancellationToken = default
256
        )
257
        {
258
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
259
                return await _primary.GetDueRecurringAsync(now, batchSize, cancellationToken).ConfigureAwait(false);
×
260
        }
261

262
        /// <inheritdoc />
263
        public async ValueTask<bool> MaterializeRecurringAsync(
264
                RecurringJobSchedule schedule,
265
                JobRecord job,
266
                DateTimeOffset nextRunAt,
267
                CancellationToken cancellationToken = default
268
        )
269
        {
270
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
NEW
271
                if (!await _recurringDurableStorage.MaterializeRecurringAsync(schedule, job, nextRunAt, cancellationToken).ConfigureAwait(false))
×
272
                        return false;
×
273
                return await _primary.MaterializeRecurringAsync(schedule, job, nextRunAt, cancellationToken).ConfigureAwait(false);
×
274
        }
275

276
        /// <inheritdoc />
277
        public async ValueTask<JobMonitoringSnapshot> GetMonitoringSnapshotAsync(CancellationToken cancellationToken = default)
278
        {
279
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
4✔
280
                return await _primary.GetMonitoringSnapshotAsync(cancellationToken).ConfigureAwait(false);
4✔
281
        }
3✔
282

283
        /// <inheritdoc />
284
        public async ValueTask<IReadOnlyList<JobRecord>> QueryJobsAsync(
285
                JobQuery query,
286
                CancellationToken cancellationToken = default
287
        )
288
        {
289
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
4✔
290
                return await _primary.QueryJobsAsync(query, cancellationToken).ConfigureAwait(false);
4✔
291
        }
3✔
292

293
        /// <inheritdoc />
294
        public async ValueTask<BatchStatus?> GetBatchStatusAsync(
295
                string batchId,
296
                CancellationToken cancellationToken = default
297
        )
298
        {
299
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
300
                return await _primary.GetBatchStatusAsync(batchId, cancellationToken).ConfigureAwait(false);
×
301
        }
302

303
        /// <inheritdoc />
304
        public async ValueTask<IReadOnlyList<BatchStatus>> QueryBatchesAsync(
305
                JobBatchQuery query,
306
                CancellationToken cancellationToken = default
307
        )
308
        {
309
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
310
                return await _primary.QueryBatchesAsync(query, cancellationToken).ConfigureAwait(false);
×
311
        }
312

313
        /// <inheritdoc />
314
        public async ValueTask<IReadOnlyList<BatchMemberStatus>> QueryBatchMembersAsync(
315
                string batchId,
316
                BatchMemberQuery query,
317
                CancellationToken cancellationToken = default
318
        )
319
        {
320
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
321
                return await _primary.QueryBatchMembersAsync(batchId, query, cancellationToken).ConfigureAwait(false);
×
322
        }
323

324
        /// <inheritdoc />
325
        public async ValueTask<BatchGraph?> GetBatchGraphAsync(
326
                string batchId,
327
                CancellationToken cancellationToken = default
328
        )
329
        {
330
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
4✔
331
                return await _primary.GetBatchGraphAsync(batchId, cancellationToken).ConfigureAwait(false);
4✔
332
        }
3✔
333

334
        /// <inheritdoc />
335
        public async ValueTask<JobStatus?> GetJobStatusAsync(
336
                string jobId,
337
                CancellationToken cancellationToken = default
338
        )
339
        {
340
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
341
                return await _primary.GetJobStatusAsync(jobId, cancellationToken).ConfigureAwait(false);
×
342
        }
343

344
        /// <inheritdoc />
345
        public async ValueTask CancelBatchAsync(string batchId, CancellationToken cancellationToken = default)
346
        {
347
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
NEW
348
                await _graphDurableStorage.CancelBatchAsync(batchId, cancellationToken).ConfigureAwait(false);
×
349
                await _primary.CancelBatchAsync(batchId, cancellationToken).ConfigureAwait(false);
×
350
        }
×
351

352
        /// <inheritdoc />
353
        public async ValueTask DeleteBatchAsync(string batchId, CancellationToken cancellationToken = default)
354
        {
355
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
NEW
356
                await _graphDurableStorage.DeleteBatchAsync(batchId, cancellationToken).ConfigureAwait(false);
×
357
                await _primary.DeleteBatchAsync(batchId, cancellationToken).ConfigureAwait(false);
×
358
        }
×
359

360
        /// <inheritdoc />
361
        public async ValueTask RetryAsync(string jobId, CancellationToken cancellationToken = default)
362
        {
363
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
364
                await DurableStorage.RetryAsync(jobId, cancellationToken).ConfigureAwait(false);
×
365
                await _primary.RetryAsync(jobId, cancellationToken).ConfigureAwait(false);
×
366
        }
×
367

368
        /// <inheritdoc />
369
        public async ValueTask DeleteAsync(string jobId, CancellationToken cancellationToken = default)
370
        {
371
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
372
                await DurableStorage.DeleteAsync(jobId, cancellationToken).ConfigureAwait(false);
×
373
                await _primary.DeleteAsync(jobId, cancellationToken).ConfigureAwait(false);
×
374
        }
×
375

376
        /// <inheritdoc />
377
        public async ValueTask PurgeJobsAsync(
378
                TimeSpan succeededRetention,
379
                TimeSpan failedRetention,
380
                CancellationToken cancellationToken = default
381
        )
382
        {
383
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
NEW
384
                await DurableStorage.PurgeJobsAsync(
×
385
                        succeededRetention,
×
386
                        failedRetention,
×
387
                        cancellationToken
×
388
                ).ConfigureAwait(false);
×
NEW
389
                await _primary.PurgeJobsAsync(
×
390
                        succeededRetention,
×
391
                        failedRetention,
×
NEW
392
                        cancellationToken
×
NEW
393
                ).ConfigureAwait(false);
×
NEW
394
        }
×
395

396
        /// <inheritdoc />
397
        public async ValueTask PurgeBatchesAsync(
398
                TimeSpan batchSucceededRetention,
399
                TimeSpan batchFailedRetention,
400
                CancellationToken cancellationToken = default
401
        )
402
        {
NEW
403
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
NEW
404
                await _graphDurableStorage.PurgeBatchesAsync(
×
NEW
405
                        batchSucceededRetention,
×
NEW
406
                        batchFailedRetention,
×
NEW
407
                        cancellationToken
×
NEW
408
                ).ConfigureAwait(false);
×
NEW
409
                await _primary.PurgeBatchesAsync(
×
410
                        batchSucceededRetention,
×
411
                        batchFailedRetention,
×
412
                        cancellationToken
×
413
                ).ConfigureAwait(false);
×
414
        }
×
415

416
        /// <inheritdoc />
417
        public async ValueTask HeartbeatAsync(JobServerSnapshot server, CancellationToken cancellationToken = default)
418
        {
419
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
4✔
420
                await _primary.HeartbeatAsync(server, cancellationToken).ConfigureAwait(false);
4✔
421
        }
4✔
422

423
        /// <inheritdoc />
424
        public async ValueTask<bool> IsHealthyAsync(CancellationToken cancellationToken = default)
425
        {
426
                await EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
×
427
                return await DurableStorage.IsHealthyAsync(cancellationToken).ConfigureAwait(false);
×
428
        }
429

430
        /// <inheritdoc />
431
        public void Dispose()
432
        {
433
                if (_disposed)
4✔
434
                        return;
×
435
                _disposed = true;
4✔
436
                if (DurableStorage is IDisposable disposable)
4✔
437
                        disposable.Dispose();
×
438
                else if (DurableStorage is IAsyncDisposable asyncDisposable)
4✔
439
                        asyncDisposable.DisposeAsync().AsTask().GetAwaiter().GetResult();
×
440
                _initialization.Dispose();
4✔
441
        }
4✔
442

443
        /// <inheritdoc />
444
        public async ValueTask DisposeAsync()
445
        {
446
                if (_disposed)
×
447
                        return;
×
448
                _disposed = true;
×
449
                if (DurableStorage is IAsyncDisposable asyncDisposable)
×
450
                        await asyncDisposable.DisposeAsync().ConfigureAwait(false);
×
451
                else if (DurableStorage is IDisposable disposable)
×
452
                        disposable.Dispose();
×
453
                _initialization.Dispose();
×
454
        }
×
455

456
        private async ValueTask EnsureInitializedAsync(CancellationToken cancellationToken)
457
        {
458
                ObjectDisposedException.ThrowIf(_disposed, this);
4✔
459
                if (Volatile.Read(ref _initialized))
4✔
460
                        return;
4✔
461

462
                await _initialization.WaitAsync(cancellationToken).ConfigureAwait(false);
4✔
463
                try
464
                {
465
                        if (Volatile.Read(ref _initialized))
4✔
466
                                return;
×
467

468
                        await DurableStorage.InitializeAsync(cancellationToken).ConfigureAwait(false);
4✔
469
                        var recoveredPrimary = new InMemoryJobStorage(_timeProvider);
4✔
470
                        await recoveredPrimary.InitializeAsync(cancellationToken).ConfigureAwait(false);
4✔
471

472
                        var recoveredJobs = new List<JobRecord>();
4✔
473
                        foreach (var state in Enum.GetValues<JobState>())
4✔
474
                        {
475
                                var skip = 0;
4✔
476
                                while (true)
×
477
                                {
478
                                        var jobs = await DurableStorage.QueryJobsAsync(
4✔
479
                                                new() { State = state, Skip = skip, Take = RecoveryBatchSize },
4✔
480
                                                cancellationToken
4✔
481
                                        ).ConfigureAwait(false);
4✔
482
                                        recoveredJobs.AddRange(jobs);
4✔
483
                                        if (jobs.Count < RecoveryBatchSize)
4✔
484
                                                break;
485
                                        skip += jobs.Count;
×
486
                                }
487
                        }
488

489
                        var batchIds = recoveredJobs
4✔
490
                                .Where(static job => job.BatchId is not null)
4✔
491
                                .Select(static job => job.BatchId!)
4✔
492
                                .Distinct(StringComparer.Ordinal)
4✔
493
                                .ToArray();
4✔
494
                        var recoveredBatches = new Dictionary<string, RecoveredBatch>(batchIds.Length, StringComparer.Ordinal);
4✔
495
                        foreach (var batchId in batchIds)
4✔
496
                        {
497
                                var status = await _graphDurableStorage.GetBatchStatusAsync(batchId, cancellationToken).ConfigureAwait(false)
4✔
498
                                        ?? throw new ImmediateJobException($"Batch '{batchId}' has members but no durable batch header.");
4✔
499
                                var graph = await _graphDurableStorage.GetBatchGraphAsync(batchId, cancellationToken).ConfigureAwait(false)
4✔
500
                                        ?? throw new ImmediateJobException($"Batch '{batchId}' has members but no durable dependency graph.");
4✔
501
                                recoveredBatches.Add(batchId, new(
4✔
502
                                        new()
4✔
503
                                        {
4✔
504
                                                Id = status.Id,
4✔
505
                                                CreatedAt = status.CreatedAt,
4✔
506
                                                TotalJobs = status.Total,
4✔
507
                                                PendingCount = status.Remaining,
4✔
508
                                                SucceededCount = status.Succeeded,
4✔
509
                                                FailedCount = status.Failed,
4✔
510
                                                CancelledCount = status.Cancelled,
4✔
511
                                                StartedAt = status.StartedAt,
4✔
512
                                                CompletedAt = status.CompletedAt,
4✔
513
                                                State = status.State,
4✔
514
                                        },
4✔
515
                                        [.. recoveredJobs.Where(job => job.BatchId == batchId)],
4✔
516
                                        [.. graph.Edges.Select(ToContinuationEdge)]
4✔
517
                                ));
4✔
518
                        }
3✔
519

520
                        var restoredBatchIds = new HashSet<string>(StringComparer.Ordinal);
4✔
521
                        while (recoveredBatches.Count != 0)
4✔
522
                        {
523
                                var ready = recoveredBatches.Values
4✔
524
                                        .Where(batch => batch.Edges
4✔
525
                                                .Where(static edge => edge.ParentBatchId is not null)
4✔
526
                                                .All(edge => restoredBatchIds.Contains(edge.ParentBatchId!)))
4✔
527
                                        .OrderBy(static batch => batch.Record.CreatedAt)
1✔
528
                                        .ThenBy(static batch => batch.Record.Id, StringComparer.Ordinal)
1✔
529
                                        .ToArray();
4✔
530
                                if (ready.Length == 0)
4✔
531
                                {
532
                                        var unresolved = string.Join(", ", recoveredBatches.Keys.Order(StringComparer.Ordinal));
×
NEW
533
                                        throw new ImmediateJobException(
×
534
                                                $"Durable batches have cyclic or missing parent-batch dependencies: {unresolved}."
×
535
                                        );
×
536
                                }
537

538
                                foreach (var batch in ready)
4✔
539
                                {
540
                                        await recoveredPrimary.EnqueueBatchAsync(
4✔
541
                                                batch.Record,
4✔
542
                                                batch.Jobs,
4✔
543
                                                batch.Edges,
4✔
544
                                                cancellationToken
4✔
545
                                        ).ConfigureAwait(false);
4✔
546
                                        _ = recoveredBatches.Remove(batch.Record.Id);
4✔
547
                                        _ = restoredBatchIds.Add(batch.Record.Id);
4✔
548
                                }
3✔
549
                        }
550

551
                        foreach (var job in recoveredJobs.Where(static job => job.BatchId is null))
4✔
552
                        {
553
                                var status = await DurableStorage.GetJobStatusAsync(job.Id, cancellationToken).ConfigureAwait(false)
4✔
554
                                        ?? throw new ImmediateJobException($"Job '{job.Id}' was queried but has no durable status.");
4✔
555
                                if (status.DependsOn.Count == 0)
4✔
556
                                        await recoveredPrimary.EnqueueAsync(job, cancellationToken).ConfigureAwait(false);
4✔
557
                                else
558
                                        await recoveredPrimary.EnqueueContinuationAsync(
×
559
                                                job,
×
560
                                                [.. status.DependsOn.Select(ToContinuationEdge)],
×
561
                                                cancellationToken
×
562
                                        ).ConfigureAwait(false);
×
563
                        }
3✔
564

565
                        var snapshot = await DurableStorage.GetMonitoringSnapshotAsync(cancellationToken).ConfigureAwait(false);
4✔
566
                        foreach (var schedule in snapshot.Recurring)
4✔
567
                                await recoveredPrimary.UpsertRecurringAsync(schedule, cancellationToken).ConfigureAwait(false);
4✔
568

569
                        _primary = recoveredPrimary;
4✔
570
                        Volatile.Write(ref _initialized, true);
4✔
571
                }
4✔
572
                finally
573
                {
574
                        _ = _initialization.Release();
4✔
575
                }
1✔
576
        }
4✔
577

578
        private static JobContinuationEdge ToContinuationEdge(BatchGraphEdge edge) => new()
4✔
579
        {
4✔
580
                ChildJobId = edge.ChildJobId,
4✔
581
                ParentJobId = edge.ParentJobId,
4✔
582
                ParentBatchId = edge.ParentBatchId,
4✔
583
                Trigger = edge.Trigger,
4✔
584
        };
4✔
585

586
        private sealed record RecoveredBatch(
4✔
587
                JobBatchRecord Record,
4✔
588
                IReadOnlyList<JobRecord> Jobs,
4✔
589
                IReadOnlyList<JobContinuationEdge> Edges
4✔
590
        );
4✔
591
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE TRIAL · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc