30304771671

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

Build # 30304771671

Build Type

Pull #33

github

Committed by

web-flow

Commit Message

Merge 46c3047f1 into e9a147742

Pull Request Pull Request #33: Improve Storage API

Coverage Stats

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

82.14

/src/Immediate.Jobs.Shared/JobModels.cs

using System.Diagnostics;

namespace Immediate.Jobs.Shared;

/// <summary>The durable lifecycle state of a job.</summary>
public enum JobState
{
        /// <summary>The job is parked until all continuation dependencies are satisfied.</summary>
        AwaitingContinuation,
        /// <summary>Reserved for jobs parked until required inputs are supplied.</summary>
        AwaitingParameters,
        /// <summary>The job is delayed until its due time.</summary>
        Scheduled,
        /// <summary>The job is ready for acquisition.</summary>
        Pending,
        /// <summary>A worker owns the job lease.</summary>
        Active,
        /// <summary>The job completed successfully.</summary>
        Succeeded,
        /// <summary>The job exhausted all attempts.</summary>
        Failed,
        /// <summary>The job was cancelled.</summary>
        Cancelled,
}

/// <summary>A storage-neutral durable job record.</summary>
public sealed record JobRecord
{
        /// <summary>The stable persisted queue name.</summary>
        public string QueueName { get; init; } = JobQueueDefinition.DefaultName;

        /// <summary>The unique opaque invocation identifier.</summary>
        public required string Id { get; init; }

        /// <summary>The generated stable job name.</summary>
        public required string JobName { get; init; }

        /// <summary>The serialized payload.</summary>
        public required string Payload { get; init; }

        /// <summary>Serialized ambient-context envelope captured while enqueueing.</summary>
        public string? Context { get; init; }

        /// <summary>Optional fairness group key scoped within the queue. Null means an independent tenant.</summary>
        public string? GroupId { get; init; }

        /// <summary>The current lifecycle state.</summary>
        public required JobState State { get; init; }

        /// <summary>UTC time at which the job may be acquired.</summary>
        public required DateTimeOffset DueAt { get; init; }

        /// <summary>UTC time at which the invocation was created.</summary>
        public required DateTimeOffset CreatedAt { get; init; }

        /// <summary>Number of attempts already started.</summary>
        public int Attempt { get; init; }

        /// <summary>The worker currently owning the record.</summary>
        public string? WorkerId { get; init; }

        /// <summary>UTC lease expiry for active work.</summary>
        public DateTimeOffset? LeaseExpiresAt { get; init; }

        /// <summary>The latest failure text.</summary>
        public string? LastError { get; init; }

        /// <summary>UTC terminal completion time.</summary>
        public DateTimeOffset? CompletedAt { get; init; }

        /// <summary>Unique recurring materialization key.</summary>
        public string? RecurringKey { get; init; }

        /// <summary>W3C trace parent captured while enqueueing.</summary>
        public string? TraceParent { get; init; }

        /// <summary>W3C trace state captured while enqueueing.</summary>
        public string? TraceState { get; init; }

        /// <summary>Trace identifier created for the latest execution attempt.</summary>
        public string? ExecutionTraceId { get; init; }

        /// <summary>Span identifier created for the latest execution attempt.</summary>
        public string? ExecutionSpanId { get; init; }

        /// <summary>UTC start time of the latest execution attempt.</summary>
        public DateTimeOffset? ExecutionStartedAt { get; init; }

        /// <summary>The atomic batch containing this invocation, if any.</summary>
        public string? BatchId { get; init; }

        /// <summary>Number of incoming continuation dependencies not yet satisfied.</summary>
        public int RemainingDependencies { get; init; }

        /// <summary>Number of settled incoming continuation dependencies whose parent failed.</summary>
        public int FailedDependencies { get; init; }
}

/// <summary>A persisted recurring schedule.</summary>
public sealed record RecurringJobSchedule
{
        /// <summary>Unique schedule identity.</summary>
        public required string Name { get; init; }

        /// <summary>The generated job definition name.</summary>
        public required string JobName { get; init; }

        /// <summary>A five- or six-field cron expression.</summary>
        public required string Cron { get; init; }

        /// <summary>An IANA time-zone identifier.</summary>
        public required string TimeZone { get; init; }

        /// <summary>Whether this schedule originated in compiled code.</summary>
        public required bool IsCodeDefined { get; init; }

        /// <summary>Whether future scheduled occurrences are paused.</summary>
        public bool IsPaused { get; init; }

        /// <summary>The next scheduled occurrence in UTC.</summary>
        public required DateTimeOffset NextRunAt { get; init; }

        /// <summary>The most recently materialized scheduled occurrence in UTC.</summary>
        public DateTimeOffset? LastRunAt { get; init; }
}

/// <summary>Immutable generated execution settings.</summary>
public sealed record JobDefinition
{
        /// <summary>The queue used by newly-created invocations.</summary>
        public JobQueueDefinition Queue { get; init; } = JobQueueDefinition.Default;

        /// <summary>The stable job name.</summary>
        public required string Name { get; init; }

        /// <summary>The generated invoker.</summary>
        public required IJobInvoker Invoker { get; init; }

        /// <summary>The job CLR type, used only for logging and diagnostics.</summary>
        public required Type JobType { get; init; }

        /// <summary>The optional code-defined cron.</summary>
        public string? Cron { get; init; }

        /// <summary>The cron time-zone identifier.</summary>
        public string TimeZone { get; init; } = "UTC";

        /// <summary>Total allowed attempts.</summary>
        public int MaxAttempts { get; init; } = 3;

        /// <summary>Optional per-attempt timeout.</summary>
        public TimeSpan? Timeout { get; init; }

        /// <summary>Maximum executions of this job per node. Zero is unbounded.</summary>
        public int MaxConcurrency { get; init; }

        /// <summary>Recurring overlap behavior.</summary>
        public OverlapPolicy OverlapPolicy { get; init; } = OverlapPolicy.Skip;

        /// <summary>Retry-delay algorithm.</summary>
        public BackoffStrategy Backoff { get; init; } = BackoffStrategy.ExponentialJitter;

        /// <summary>Retry base delay.</summary>
        public TimeSpan BackoffBase { get; init; } = TimeSpan.FromSeconds(5);
}

/// <summary>A monitoring query.</summary>
public sealed record JobQuery
{
        /// <summary>Optional exact invocation identifier.</summary>
        public string? Id { get; init; }

        /// <summary>Optional state filter.</summary>
        public JobState? State { get; init; }

        /// <summary>Optional exact queue name filter.</summary>
        public string? QueueName { get; init; }

        /// <summary>Optional case-insensitive job name search.</summary>
        public string? Search { get; init; }

        /// <summary>Number of records to skip.</summary>
        public int Skip { get; init; }

        /// <summary>Maximum records to return.</summary>
        public int Take { get; init; } = 100;
}

/// <summary>Immutable compile-time queue settings.</summary>
public sealed record JobQueueDefinition
{
        /// <summary>The built-in queue used by jobs without <c>UsesQueue</c>.</summary>
        public const string DefaultName = "default";

        /// <summary>The built-in default queue definition.</summary>
        public static JobQueueDefinition Default { get; } = new() { Name = DefaultName };

        /// <summary>The stable persisted queue name.</summary>
        public required string Name { get; init; }

        /// <summary>The dispatch priority. Larger values are dispatched first.</summary>
        public int Priority { get; init; }

        /// <summary>Maximum in-flight jobs on one scheduler node. Zero means unbounded.</summary>
        public int Concurrency { get; init; }
}

/// <summary>Describes the remaining acquisition capacity for one queue.</summary>
public sealed record JobQueueAcquisition
{
        /// <summary>The persisted queue name.</summary>
        public required string QueueName { get; init; }

        /// <summary>Maximum records to acquire from this queue.</summary>
        public required int Capacity { get; init; }

        /// <summary>Remaining acquisition capacity by stable job name.</summary>
        public required IReadOnlyDictionary<string, int> JobCapacities { get; init; }
}

/// <summary>Immutable fair queue settings applied by storage during one acquisition request.</summary>
public sealed record FairQueuePolicy(
        double ConcurrencyShareThreshold,
        int MinInflightForNoisy,
        bool GroupRoundRobin
);

/// <summary>A priority-ordered, node-local storage acquisition request.</summary>
public sealed record JobAcquisitionRequest
{
        /// <summary>The worker node taking ownership.</summary>
        public required string WorkerId { get; init; }

        /// <summary>The lease assigned to acquired records.</summary>
        public required TimeSpan Lease { get; init; }

        /// <summary>The maximum total records to acquire.</summary>
        public required int BatchSize { get; init; }

        /// <summary>Queues in dispatch order, with their remaining capacities.</summary>
        public required IReadOnlyList<JobQueueAcquisition> Queues { get; init; }

        /// <summary>Fair queue policy for this acquisition, or <see langword="null"/> when fairness is disabled.</summary>
        public FairQueuePolicy? FairQueues { get; init; }
}

/// <summary>Queue totals for monitoring and health endpoints.</summary>
public sealed record JobMonitoringSnapshot(
        DateTimeOffset CapturedAt,
        IReadOnlyDictionary<JobState, long> Counts,
        IReadOnlyList<RecurringJobSchedule> Recurring,
        IReadOnlyList<JobServerSnapshot> Servers
)
{
        /// <summary>Capabilities implemented by the active storage provider.</summary>
        public StorageCapabilities Capabilities { get; init; } = StorageCapabilities.Queue;
}

/// <summary>A live scheduler-node heartbeat.</summary>
public sealed record JobServerSnapshot(
        string WorkerId,
        DateTimeOffset LastHeartbeat,
        int ActiveWorkers,
        int MaxWorkers
);

internal static class TraceContextCapture
{
        public static (string? Parent, string? State) Current()
        {
                var activity = Activity.Current;
                return (activity?.Id, activity?.TraceStateString);
        }
}

1	using System.Diagnostics;
2
3	namespace Immediate.Jobs.Shared;
4
5	/// <summary>The durable lifecycle state of a job.</summary>
6	public enum JobState
7	{
8	/// <summary>The job is parked until all continuation dependencies are satisfied.</summary>
9	AwaitingContinuation,
10	/// <summary>Reserved for jobs parked until required inputs are supplied.</summary>
11	AwaitingParameters,
12	/// <summary>The job is delayed until its due time.</summary>
13	Scheduled,
14	/// <summary>The job is ready for acquisition.</summary>
15	Pending,
16	/// <summary>A worker owns the job lease.</summary>
17	Active,
18	/// <summary>The job completed successfully.</summary>
19	Succeeded,
20	/// <summary>The job exhausted all attempts.</summary>
21	Failed,
22	/// <summary>The job was cancelled.</summary>
23	Cancelled,
24	}
25
26	/// <summary>A storage-neutral durable job record.</summary>
27	public sealed record JobRecord
28	{
29	/// <summary>The stable persisted queue name.</summary>
30	public string QueueName { get; init; } = JobQueueDefinition.DefaultName;	5✔
31
32	/// <summary>The unique opaque invocation identifier.</summary>
33	public required string Id { get; init; }
34
35	/// <summary>The generated stable job name.</summary>
36	public required string JobName { get; init; }
37
38	/// <summary>The serialized payload.</summary>
39	public required string Payload { get; init; }
40
41	/// <summary>Serialized ambient-context envelope captured while enqueueing.</summary>
42	public string? Context { get; init; }
43
44	/// <summary>Optional fairness group key scoped within the queue. Null means an independent tenant.</summary>
45	public string? GroupId { get; init; }
46
47	/// <summary>The current lifecycle state.</summary>
48	public required JobState State { get; init; }
49
50	/// <summary>UTC time at which the job may be acquired.</summary>
51	public required DateTimeOffset DueAt { get; init; }
52
53	/// <summary>UTC time at which the invocation was created.</summary>
54	public required DateTimeOffset CreatedAt { get; init; }
55
56	/// <summary>Number of attempts already started.</summary>
57	public int Attempt { get; init; }
58
59	/// <summary>The worker currently owning the record.</summary>
60	public string? WorkerId { get; init; }
61
62	/// <summary>UTC lease expiry for active work.</summary>
63	public DateTimeOffset? LeaseExpiresAt { get; init; }
64
65	/// <summary>The latest failure text.</summary>
66	public string? LastError { get; init; }
67
68	/// <summary>UTC terminal completion time.</summary>
69	public DateTimeOffset? CompletedAt { get; init; }
70
71	/// <summary>Unique recurring materialization key.</summary>
72	public string? RecurringKey { get; init; }
73
74	/// <summary>W3C trace parent captured while enqueueing.</summary>
75	public string? TraceParent { get; init; }
76
77	/// <summary>W3C trace state captured while enqueueing.</summary>
78	public string? TraceState { get; init; }
79
80	/// <summary>Trace identifier created for the latest execution attempt.</summary>
81	public string? ExecutionTraceId { get; init; }
82
83	/// <summary>Span identifier created for the latest execution attempt.</summary>
84	public string? ExecutionSpanId { get; init; }
85
86	/// <summary>UTC start time of the latest execution attempt.</summary>
87	public DateTimeOffset? ExecutionStartedAt { get; init; }
88
89	/// <summary>The atomic batch containing this invocation, if any.</summary>
90	public string? BatchId { get; init; }
91
92	/// <summary>Number of incoming continuation dependencies not yet satisfied.</summary>
93	public int RemainingDependencies { get; init; }
94
95	/// <summary>Number of settled incoming continuation dependencies whose parent failed.</summary>
96	public int FailedDependencies { get; init; }
97	}
98
99	/// <summary>A persisted recurring schedule.</summary>
100	public sealed record RecurringJobSchedule
101	{
102	/// <summary>Unique schedule identity.</summary>
103	public required string Name { get; init; }
104
105	/// <summary>The generated job definition name.</summary>
106	public required string JobName { get; init; }
107
108	/// <summary>A five- or six-field cron expression.</summary>
109	public required string Cron { get; init; }
110
111	/// <summary>An IANA time-zone identifier.</summary>
112	public required string TimeZone { get; init; }
113
114	/// <summary>Whether this schedule originated in compiled code.</summary>
115	public required bool IsCodeDefined { get; init; }
116
117	/// <summary>Whether future scheduled occurrences are paused.</summary>
118	public bool IsPaused { get; init; }
119
120	/// <summary>The next scheduled occurrence in UTC.</summary>
121	public required DateTimeOffset NextRunAt { get; init; }
122
123	/// <summary>The most recently materialized scheduled occurrence in UTC.</summary>
124	public DateTimeOffset? LastRunAt { get; init; }
125	}
126
127	/// <summary>Immutable generated execution settings.</summary>
128	public sealed record JobDefinition
129	{
130	/// <summary>The queue used by newly-created invocations.</summary>
131	public JobQueueDefinition Queue { get; init; } = JobQueueDefinition.Default;	4✔
132
133	/// <summary>The stable job name.</summary>
134	public required string Name { get; init; }
135
136	/// <summary>The generated invoker.</summary>
137	public required IJobInvoker Invoker { get; init; }
138
139	/// <summary>The job CLR type, used only for logging and diagnostics.</summary>
140	public required Type JobType { get; init; }
141
142	/// <summary>The optional code-defined cron.</summary>
143	public string? Cron { get; init; }
144
145	/// <summary>The cron time-zone identifier.</summary>
146	public string TimeZone { get; init; } = "UTC";	4✔
147
148	/// <summary>Total allowed attempts.</summary>
149	public int MaxAttempts { get; init; } = 3;	4✔
150
151	/// <summary>Optional per-attempt timeout.</summary>
152	public TimeSpan? Timeout { get; init; }
153
154	/// <summary>Maximum executions of this job per node. Zero is unbounded.</summary>
155	public int MaxConcurrency { get; init; }
156
157	/// <summary>Recurring overlap behavior.</summary>
158	public OverlapPolicy OverlapPolicy { get; init; } = OverlapPolicy.Skip;
159
160	/// <summary>Retry-delay algorithm.</summary>
161	public BackoffStrategy Backoff { get; init; } = BackoffStrategy.ExponentialJitter;	4✔
162
163	/// <summary>Retry base delay.</summary>
164	public TimeSpan BackoffBase { get; init; } = TimeSpan.FromSeconds(5);	4✔
165	}
166
167	/// <summary>A monitoring query.</summary>
168	public sealed record JobQuery
169	{
170	/// <summary>Optional exact invocation identifier.</summary>
171	public string? Id { get; init; }
172
173	/// <summary>Optional state filter.</summary>
174	public JobState? State { get; init; }
175
176	/// <summary>Optional exact queue name filter.</summary>
177	public string? QueueName { get; init; }
178
179	/// <summary>Optional case-insensitive job name search.</summary>
180	public string? Search { get; init; }
181
182	/// <summary>Number of records to skip.</summary>
183	public int Skip { get; init; }
184
185	/// <summary>Maximum records to return.</summary>
186	public int Take { get; init; } = 100;	5✔
187	}
188
189	/// <summary>Immutable compile-time queue settings.</summary>
190	public sealed record JobQueueDefinition
191	{
192	/// <summary>The built-in queue used by jobs without <c>UsesQueue</c>.</summary>
193	public const string DefaultName = "default";
194
195	/// <summary>The built-in default queue definition.</summary>
196	public static JobQueueDefinition Default { get; } = new() { Name = DefaultName };	4✔
197
198	/// <summary>The stable persisted queue name.</summary>
199	public required string Name { get; init; }
200
201	/// <summary>The dispatch priority. Larger values are dispatched first.</summary>
202	public int Priority { get; init; }
203
204	/// <summary>Maximum in-flight jobs on one scheduler node. Zero means unbounded.</summary>
205	public int Concurrency { get; init; }
206	}
207
208	/// <summary>Describes the remaining acquisition capacity for one queue.</summary>
209	public sealed record JobQueueAcquisition
210	{
211	/// <summary>The persisted queue name.</summary>
212	public required string QueueName { get; init; }
213
214	/// <summary>Maximum records to acquire from this queue.</summary>
215	public required int Capacity { get; init; }
216
217	/// <summary>Remaining acquisition capacity by stable job name.</summary>
218	public required IReadOnlyDictionary<string, int> JobCapacities { get; init; }
219	}
220
221	/// <summary>Immutable fair queue settings applied by storage during one acquisition request.</summary>
NEW 222	public sealed record FairQueuePolicy(	×
NEW 223	double ConcurrencyShareThreshold,	×
NEW 224	int MinInflightForNoisy,	×
NEW 225	bool GroupRoundRobin	×
NEW 226	);	×
227
228	/// <summary>A priority-ordered, node-local storage acquisition request.</summary>
229	public sealed record JobAcquisitionRequest
230	{
231	/// <summary>The worker node taking ownership.</summary>
232	public required string WorkerId { get; init; }
233
234	/// <summary>The lease assigned to acquired records.</summary>
235	public required TimeSpan Lease { get; init; }
236
237	/// <summary>The maximum total records to acquire.</summary>
238	public required int BatchSize { get; init; }
239
240	/// <summary>Queues in dispatch order, with their remaining capacities.</summary>
241	public required IReadOnlyList<JobQueueAcquisition> Queues { get; init; }
242
243	/// <summary>Fair queue policy for this acquisition, or <see langword="null"/> when fairness is disabled.</summary>
244	public FairQueuePolicy? FairQueues { get; init; }
245	}
246
247	/// <summary>Queue totals for monitoring and health endpoints.</summary>
248	public sealed record JobMonitoringSnapshot(	5✔
249	DateTimeOffset CapturedAt,	5✔
250	IReadOnlyDictionary<JobState, long> Counts,	5✔
251	IReadOnlyList<RecurringJobSchedule> Recurring,	5✔
252	IReadOnlyList<JobServerSnapshot> Servers	5✔
253	)	5✔
254	{
255	/// <summary>Capabilities implemented by the active storage provider.</summary>
256	public StorageCapabilities Capabilities { get; init; } = StorageCapabilities.Queue;	5✔
257	}
258
259	/// <summary>A live scheduler-node heartbeat.</summary>
260	public sealed record JobServerSnapshot(	4✔
261	string WorkerId,	4✔
262	DateTimeOffset LastHeartbeat,	4✔
263	int ActiveWorkers,	4✔
264	int MaxWorkers	4✔
265	);	4✔
266
267	internal static class TraceContextCapture
268	{
269	public static (string? Parent, string? State) Current()
270	{
271	var activity = Activity.Current;	4✔
272	return (activity?.Id, activity?.TraceStateString);	4✔
273	}
274	}

ImmediatePlatform / Immediate.Jobs / 30304771671

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