30302568744

Committed 27 Jul 2026 08:26PM UTC coverage: 66.69%. First build

Build # 30302568744

Build Type

Pull #26

github

Committed by

web-flow

Commit Message

Merge 4a231468a into eac5c1044

Pull Request Pull Request #26: Add batches and continuation workflows

Coverage Stats

1239 of 1810 new or added lines in 17 files covered. (68.45%)

2855 of 4281 relevant lines covered (66.69%)

2.66 hits per line

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

92.65

/src/Immediate.Jobs.Shared/JobBatchModels.cs

namespace Immediate.Jobs.Shared;

/// <summary>An opaque reference to a durable job invocation.</summary>
public readonly struct JobHandle : IEquatable<JobHandle>
{
        /// <summary>Creates a handle for an existing invocation identifier.</summary>
        public JobHandle(string id)
        {
                ArgumentException.ThrowIfNullOrWhiteSpace(id);
                Id = id;
        }

        internal JobHandle(string id, JobBatch? batch)
                : this(id) => Batch = batch;

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

        internal JobBatch? Batch { get; }

        /// <inheritdoc />
        public bool Equals(JobHandle other) => string.Equals(Id, other.Id, StringComparison.Ordinal);

        /// <inheritdoc />
        public override bool Equals(object? obj) => obj is JobHandle other && Equals(other);

        /// <inheritdoc />
        public override int GetHashCode() => Id is null ? 0 : StringComparer.Ordinal.GetHashCode(Id);

        /// <summary>Compares two handles by their opaque invocation identifier.</summary>
        public static bool operator ==(JobHandle left, JobHandle right) => left.Equals(right);

        /// <summary>Compares two handles by their opaque invocation identifier.</summary>
        public static bool operator !=(JobHandle left, JobHandle right) => !left.Equals(right);

        /// <inheritdoc />
        public override string ToString() => Id ?? string.Empty;
}

/// <summary>An opaque reference to a committed atomic batch.</summary>
public sealed record BatchHandle
{
        /// <summary>Creates a handle for an existing batch identifier.</summary>
        public BatchHandle(string id)
        {
                ArgumentException.ThrowIfNullOrWhiteSpace(id);
                Id = id;
        }

        /// <summary>The opaque batch identifier.</summary>
        public string Id { get; }
}

/// <summary>Determines how a continuation evaluates its parents.</summary>
public enum ContinuationTrigger
{
        /// <summary>Run only when every parent succeeds; otherwise cancel the continuation.</summary>
        AllSucceeded,

        /// <summary>Run after every parent reaches any terminal state.</summary>
        AllComplete,
}

/// <summary>Determines how work scheduled by a running job joins its workflow.</summary>
public enum ContinuationOptions
{
        /// <summary>Schedule outside the current batch and do not modify existing continuations.</summary>
        Detached,

        /// <summary>Add the job to the current batch as a parallel branch.</summary>
        BesideContinuations,

        /// <summary>Add the job to the current batch and make current waiters depend on it too.</summary>
        BeforeContinuations,
}

/// <summary>The durable lifecycle state of an atomic batch.</summary>
public enum BatchState
{
        /// <summary>At least one member has not reached a terminal state.</summary>
        Executing,
        /// <summary>Every member succeeded.</summary>
        Succeeded,
        /// <summary>At least one member failed.</summary>
        Failed,
        /// <summary>No member failed and at least one member was cancelled.</summary>
        Cancelled,
}

/// <summary>A durable atomic-batch header.</summary>
public sealed record JobBatchRecord
{
        /// <summary>The opaque batch identifier.</summary>
        public required string Id { get; init; }
        /// <summary>UTC time at which the batch was created.</summary>
        public required DateTimeOffset CreatedAt { get; init; }
        /// <summary>Total members added to the batch.</summary>
        public required int TotalJobs { get; init; }
        /// <summary>Members that have not reached a terminal state.</summary>
        public required int PendingCount { get; init; }
        /// <summary>Members that completed successfully.</summary>
        public int SucceededCount { get; init; }
        /// <summary>Members that exhausted their attempts.</summary>
        public int FailedCount { get; init; }
        /// <summary>Members cancelled by an explicit action or dependency violation.</summary>
        public int CancelledCount { get; init; }
        /// <summary>UTC time at which the first member was acquired.</summary>
        public DateTimeOffset? StartedAt { get; init; }
        /// <summary>UTC terminal completion time.</summary>
        public DateTimeOffset? CompletedAt { get; init; }
        /// <summary>The aggregate lifecycle state.</summary>
        public required BatchState State { get; init; }
}

/// <summary>A durable dependency edge from a job or batch to a child job.</summary>
public sealed record JobContinuationEdge
{
        /// <summary>The waiting child invocation.</summary>
        public required string ChildJobId { get; init; }
        /// <summary>The parent invocation for a job-to-job dependency.</summary>
        public string? ParentJobId { get; init; }
        /// <summary>The parent batch for a batch-to-job dependency.</summary>
        public string? ParentBatchId { get; init; }
        /// <summary>The condition under which the edge is satisfied.</summary>
        public ContinuationTrigger Trigger { get; init; } = ContinuationTrigger.AllSucceeded;
}

/// <summary>A continuation buffered during a running job attempt and committed only on success.</summary>
public sealed record JobContinuationAddition
{
        /// <summary>The fully serialized new invocation.</summary>
        public required JobRecord Job { get; init; }
        /// <summary>How the new invocation joins and modifies the current workflow.</summary>
        public required ContinuationOptions Options { get; init; }
        /// <summary>The dependency trigger from the running job to the new invocation.</summary>
        public ContinuationTrigger Trigger { get; init; } = ContinuationTrigger.AllSucceeded;
}

/// <summary>Per-attempt buffer used by generated schedulers during job execution.</summary>
public sealed class JobExecutionBuffer
{
        private readonly List<JobContinuationAddition> _additions = [];

        internal void Add(JobContinuationAddition addition) => _additions.Add(addition);

        internal IReadOnlyList<JobContinuationAddition> Snapshot() => _additions.Count == 0 ? [] : [.. _additions];
}

/// <summary>Aggregate progress for an atomic batch.</summary>
public sealed record BatchStatus(
        string Id,
        BatchState State,
        int Total,
        int Succeeded,
        int Failed,
        int Cancelled,
        int Remaining,
        DateTimeOffset CreatedAt,
        DateTimeOffset? StartedAt,
        DateTimeOffset? CompletedAt,
        double FractionSettled
);

/// <summary>Filters members returned from a batch.</summary>
public sealed record BatchMemberQuery
{
        /// <summary>Optional lifecycle-state filter.</summary>
        public JobState? State { get; init; }
        /// <summary>Number of members to skip.</summary>
        public int Skip { get; init; }
        /// <summary>Maximum members to return.</summary>
        public int Take { get; init; } = 100;
}

/// <summary>Filters batches for dashboard presentation.</summary>
public sealed record JobBatchQuery
{
        /// <summary>Optional aggregate-state filter.</summary>
        public BatchState? State { get; init; }
        /// <summary>Number of batches to skip.</summary>
        public int Skip { get; init; }
        /// <summary>Maximum batches to return.</summary>
        public int Take { get; init; } = 100;
}

/// <summary>Monitoring data for one batch member.</summary>
public sealed record BatchMemberStatus(
        string JobId,
        string JobName,
        string QueueName,
        JobState State,
        int Attempt,
        DateTimeOffset CreatedAt,
        DateTimeOffset? CompletedAt,
        string? LastError
);

/// <summary>A batch dependency graph.</summary>
public sealed record BatchGraph(
        string BatchId,
        IReadOnlyList<BatchGraphNode> Nodes,
        IReadOnlyList<BatchGraphEdge> Edges
);

/// <summary>A job node in a batch graph.</summary>
public sealed record BatchGraphNode(string JobId, string JobName, JobState State);

/// <summary>A dependency edge in a batch graph.</summary>
public sealed record BatchGraphEdge(
        string ChildJobId,
        string? ParentJobId,
        string? ParentBatchId,
        ContinuationTrigger Trigger
);

/// <summary>Monitoring data for one job.</summary>
public sealed record JobStatus(
        string JobId,
        string JobName,
        string QueueName,
        JobState State,
        int Attempt,
        int MaxAttempts,
        DateTimeOffset CreatedAt,
        DateTimeOffset DueAt,
        DateTimeOffset? CompletedAt,
        string? LastError,
        string? BatchId,
        IReadOnlyList<BatchGraphEdge> DependsOn
);

/// <summary>Read-only batch monitoring.</summary>
public interface IJobBatchMonitor
{
        /// <summary>Gets aggregate batch progress.</summary>
        ValueTask<BatchStatus?> GetStatusAsync(string batchId, CancellationToken cancellationToken = default);
        /// <summary>Queries batch members.</summary>
        ValueTask<IReadOnlyList<BatchMemberStatus>> QueryMembersAsync(
                string batchId,
                BatchMemberQuery query,
                CancellationToken cancellationToken = default
        );
        /// <summary>Gets the persisted dependency graph.</summary>
        ValueTask<BatchGraph?> GetGraphAsync(string batchId, CancellationToken cancellationToken = default);
}

/// <summary>Read-only single-job monitoring.</summary>
public interface IJobMonitor
{
        /// <summary>Gets one job and its incoming dependencies.</summary>
        ValueTask<JobStatus?> GetJobAsync(string jobId, CancellationToken cancellationToken = default);
}

1	namespace Immediate.Jobs.Shared;
2
3	/// <summary>An opaque reference to a durable job invocation.</summary>
4	public readonly struct JobHandle : IEquatable<JobHandle>
5	{
6	/// <summary>Creates a handle for an existing invocation identifier.</summary>
7	public JobHandle(string id)
8	{
9	ArgumentException.ThrowIfNullOrWhiteSpace(id);	4✔
10	Id = id;	4✔
11	}	4✔
12
13	internal JobHandle(string id, JobBatch? batch)
14	: this(id) => Batch = batch;	4✔
15
16	/// <summary>The opaque invocation identifier.</summary>
17	public string Id { get; }
18
19	internal JobBatch? Batch { get; }
20
21	/// <inheritdoc />
22	public bool Equals(JobHandle other) => string.Equals(Id, other.Id, StringComparison.Ordinal);	4✔
23
24	/// <inheritdoc />
NEW 25	public override bool Equals(object? obj) => obj is JobHandle other && Equals(other);	×
26
27	/// <inheritdoc />
NEW 28	public override int GetHashCode() => Id is null ? 0 : StringComparer.Ordinal.GetHashCode(Id);	×
29
30	/// <summary>Compares two handles by their opaque invocation identifier.</summary>
NEW 31	public static bool operator ==(JobHandle left, JobHandle right) => left.Equals(right);	×
32
33	/// <summary>Compares two handles by their opaque invocation identifier.</summary>
NEW 34	public static bool operator !=(JobHandle left, JobHandle right) => !left.Equals(right);	×
35
36	/// <inheritdoc />
NEW 37	public override string ToString() => Id ?? string.Empty;	×
38	}
39
40	/// <summary>An opaque reference to a committed atomic batch.</summary>
41	public sealed record BatchHandle
42	{
43	/// <summary>Creates a handle for an existing batch identifier.</summary>
44	public BatchHandle(string id)	4✔
45	{
46	ArgumentException.ThrowIfNullOrWhiteSpace(id);	4✔
47	Id = id;	4✔
48	}	4✔
49
50	/// <summary>The opaque batch identifier.</summary>
51	public string Id { get; }
52	}
53
54	/// <summary>Determines how a continuation evaluates its parents.</summary>
55	public enum ContinuationTrigger
56	{
57	/// <summary>Run only when every parent succeeds; otherwise cancel the continuation.</summary>
58	AllSucceeded,
59
60	/// <summary>Run after every parent reaches any terminal state.</summary>
61	AllComplete,
62	}
63
64	/// <summary>Determines how work scheduled by a running job joins its workflow.</summary>
65	public enum ContinuationOptions
66	{
67	/// <summary>Schedule outside the current batch and do not modify existing continuations.</summary>
68	Detached,
69
70	/// <summary>Add the job to the current batch as a parallel branch.</summary>
71	BesideContinuations,
72
73	/// <summary>Add the job to the current batch and make current waiters depend on it too.</summary>
74	BeforeContinuations,
75	}
76
77	/// <summary>The durable lifecycle state of an atomic batch.</summary>
78	public enum BatchState
79	{
80	/// <summary>At least one member has not reached a terminal state.</summary>
81	Executing,
82	/// <summary>Every member succeeded.</summary>
83	Succeeded,
84	/// <summary>At least one member failed.</summary>
85	Failed,
86	/// <summary>No member failed and at least one member was cancelled.</summary>
87	Cancelled,
88	}
89
90	/// <summary>A durable atomic-batch header.</summary>
91	public sealed record JobBatchRecord
92	{
93	/// <summary>The opaque batch identifier.</summary>
94	public required string Id { get; init; }
95	/// <summary>UTC time at which the batch was created.</summary>
96	public required DateTimeOffset CreatedAt { get; init; }
97	/// <summary>Total members added to the batch.</summary>
98	public required int TotalJobs { get; init; }
99	/// <summary>Members that have not reached a terminal state.</summary>
100	public required int PendingCount { get; init; }
101	/// <summary>Members that completed successfully.</summary>
102	public int SucceededCount { get; init; }
103	/// <summary>Members that exhausted their attempts.</summary>
104	public int FailedCount { get; init; }
105	/// <summary>Members cancelled by an explicit action or dependency violation.</summary>
106	public int CancelledCount { get; init; }
107	/// <summary>UTC time at which the first member was acquired.</summary>
108	public DateTimeOffset? StartedAt { get; init; }
109	/// <summary>UTC terminal completion time.</summary>
110	public DateTimeOffset? CompletedAt { get; init; }
111	/// <summary>The aggregate lifecycle state.</summary>
112	public required BatchState State { get; init; }
113	}
114
115	/// <summary>A durable dependency edge from a job or batch to a child job.</summary>
116	public sealed record JobContinuationEdge
117	{
118	/// <summary>The waiting child invocation.</summary>
119	public required string ChildJobId { get; init; }
120	/// <summary>The parent invocation for a job-to-job dependency.</summary>
121	public string? ParentJobId { get; init; }
122	/// <summary>The parent batch for a batch-to-job dependency.</summary>
123	public string? ParentBatchId { get; init; }
124	/// <summary>The condition under which the edge is satisfied.</summary>
125	public ContinuationTrigger Trigger { get; init; } = ContinuationTrigger.AllSucceeded;
126	}
127
128	/// <summary>A continuation buffered during a running job attempt and committed only on success.</summary>
129	public sealed record JobContinuationAddition
130	{
131	/// <summary>The fully serialized new invocation.</summary>
132	public required JobRecord Job { get; init; }
133	/// <summary>How the new invocation joins and modifies the current workflow.</summary>
134	public required ContinuationOptions Options { get; init; }
135	/// <summary>The dependency trigger from the running job to the new invocation.</summary>
136	public ContinuationTrigger Trigger { get; init; } = ContinuationTrigger.AllSucceeded;
137	}
138
139	/// <summary>Per-attempt buffer used by generated schedulers during job execution.</summary>
140	public sealed class JobExecutionBuffer
141	{
142	private readonly List<JobContinuationAddition> _additions = [];	4✔
143
144	internal void Add(JobContinuationAddition addition) => _additions.Add(addition);	4✔
145
146	internal IReadOnlyList<JobContinuationAddition> Snapshot() => _additions.Count == 0 ? [] : [.. _additions];	4✔
147	}
148
149	/// <summary>Aggregate progress for an atomic batch.</summary>
150	public sealed record BatchStatus(	4✔
151	string Id,	4✔
152	BatchState State,	4✔
153	int Total,	4✔
154	int Succeeded,	4✔
155	int Failed,	4✔
156	int Cancelled,	4✔
157	int Remaining,	4✔
158	DateTimeOffset CreatedAt,	4✔
159	DateTimeOffset? StartedAt,	4✔
160	DateTimeOffset? CompletedAt,	4✔
161	double FractionSettled	4✔
162	);	4✔
163
164	/// <summary>Filters members returned from a batch.</summary>
165	public sealed record BatchMemberQuery
166	{
167	/// <summary>Optional lifecycle-state filter.</summary>
168	public JobState? State { get; init; }
169	/// <summary>Number of members to skip.</summary>
170	public int Skip { get; init; }
171	/// <summary>Maximum members to return.</summary>
172	public int Take { get; init; } = 100;	4✔
173	}
174
175	/// <summary>Filters batches for dashboard presentation.</summary>
176	public sealed record JobBatchQuery
177	{
178	/// <summary>Optional aggregate-state filter.</summary>
179	public BatchState? State { get; init; }
180	/// <summary>Number of batches to skip.</summary>
181	public int Skip { get; init; }
182	/// <summary>Maximum batches to return.</summary>
183	public int Take { get; init; } = 100;	4✔
184	}
185
186	/// <summary>Monitoring data for one batch member.</summary>
187	public sealed record BatchMemberStatus(	4✔
188	string JobId,	4✔
189	string JobName,	4✔
190	string QueueName,	4✔
191	JobState State,	4✔
192	int Attempt,	4✔
193	DateTimeOffset CreatedAt,	4✔
194	DateTimeOffset? CompletedAt,	4✔
195	string? LastError	4✔
196	);	4✔
197
198	/// <summary>A batch dependency graph.</summary>
199	public sealed record BatchGraph(	4✔
200	string BatchId,	4✔
201	IReadOnlyList<BatchGraphNode> Nodes,	4✔
202	IReadOnlyList<BatchGraphEdge> Edges	4✔
203	);	4✔
204
205	/// <summary>A job node in a batch graph.</summary>
206	public sealed record BatchGraphNode(string JobId, string JobName, JobState State);	4✔
207
208	/// <summary>A dependency edge in a batch graph.</summary>
209	public sealed record BatchGraphEdge(	4✔
210	string ChildJobId,	4✔
211	string? ParentJobId,	4✔
212	string? ParentBatchId,	4✔
213	ContinuationTrigger Trigger	4✔
214	);	4✔
215
216	/// <summary>Monitoring data for one job.</summary>
217	public sealed record JobStatus(	4✔
218	string JobId,	4✔
219	string JobName,	4✔
220	string QueueName,	4✔
221	JobState State,	4✔
222	int Attempt,	4✔
223	int MaxAttempts,	4✔
224	DateTimeOffset CreatedAt,	4✔
225	DateTimeOffset DueAt,	4✔
226	DateTimeOffset? CompletedAt,	4✔
227	string? LastError,	4✔
228	string? BatchId,	4✔
229	IReadOnlyList<BatchGraphEdge> DependsOn	4✔
230	);	4✔
231
232	/// <summary>Read-only batch monitoring.</summary>
233	public interface IJobBatchMonitor
234	{
235	/// <summary>Gets aggregate batch progress.</summary>
236	ValueTask<BatchStatus?> GetStatusAsync(string batchId, CancellationToken cancellationToken = default);
237	/// <summary>Queries batch members.</summary>
238	ValueTask<IReadOnlyList<BatchMemberStatus>> QueryMembersAsync(
239	string batchId,
240	BatchMemberQuery query,
241	CancellationToken cancellationToken = default
242	);
243	/// <summary>Gets the persisted dependency graph.</summary>
244	ValueTask<BatchGraph?> GetGraphAsync(string batchId, CancellationToken cancellationToken = default);
245	}
246
247	/// <summary>Read-only single-job monitoring.</summary>
248	public interface IJobMonitor
249	{
250	/// <summary>Gets one job and its incoming dependencies.</summary>
251	ValueTask<JobStatus?> GetJobAsync(string jobId, CancellationToken cancellationToken = default);
252	}

ImmediatePlatform / Immediate.Jobs / 30302568744

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