From 0473024504e914890f12654263ef9252b6ae15d7 Mon Sep 17 00:00:00 2001 From: Brian Tyler Date: Thu, 20 Aug 2026 13:50:04 +0100 Subject: [PATCH] feat: add task continuation --- README.md | 16 ++ .../Components/Pages/JobDetail.razor | 26 +++ .../Components/Pages/Jobs.razor | 2 + .../Internal/DashboardFormat.cs | 1 + .../Operations/EnqueueJobOperation.cs | 129 +++++++++++++++ .../PostgresJobInspectionOperation.cs | 153 ++++++++++++++++++ .../PostgresJobRetentionOperation.cs | 17 +- .../Operations/TryClaimNextJobOperation.cs | 14 ++ .../Internal/PostgresMigrator.cs | 10 ++ .../Internal/PostgresNames.cs | 5 +- src/Sheddueller.Testing/FakeEnqueuedJob.cs | 9 +- src/Sheddueller.Testing/FakeJobEnqueuer.cs | 85 ++++++++-- src/Sheddueller/Enqueueing/JobEnqueuer.cs | 80 ++++++++- .../Inspection/Jobs/JobInspectionDetail.cs | 10 ++ .../Inspection/Jobs/JobQueuePositionKind.cs | 5 + src/Sheddueller/JobEnqueueItem.cs | 44 +++++ src/Sheddueller/Storage/EnqueueJobRequest.cs | 3 +- .../JobRetentionStoreContractTests.cs | 45 ++++++ .../JobStoreContractTests.cs | 55 ++++++- .../FakeJobEnqueuerTests.cs | 19 +++ test/Sheddueller.Tests/JobEnqueuerTests.cs | 75 +++++++++ 21 files changed, 780 insertions(+), 23 deletions(-) diff --git a/README.md b/README.md index af2b37a..13d9242 100644 --- a/README.md +++ b/README.md @@ -178,6 +178,22 @@ Set `ShedduellerOptions.EnableJobLogCapture = true` to enable durable capture of Use `NotBeforeUtc` for delayed jobs. Use `JobIdempotencyKind.MethodAndArguments` to reuse an existing queued job with the same target method and serialized arguments. +Submit dependency graphs as one atomic batch. `DependsOn` blocks a job until every referenced job in the same batch is terminal; completed, failed, and canceled prerequisites all satisfy the dependency. Graphs may contain arbitrary fan-out, fan-in, and depth. + +```csharp +var fetchProfile = JobEnqueueItem.Create( + (jobs, ct) => jobs.FetchProfileAsync(managerId, ct)); +var fetchRates = JobEnqueueItem.Create( + (jobs, ct) => jobs.FetchRatesAsync(managerId, ct)); +var aggregate = JobEnqueueItem.Create( + (jobs, ct) => jobs.AggregateAsync(managerId, ct)) + .DependsOn([fetchProfile, fetchRates]); + +await enqueuer.EnqueueManyAsync([fetchProfile, fetchRates, aggregate], cancellationToken); +``` + +Every prerequisite must be present in the submitted batch. Cycles, self-dependencies, duplicate items, missing prerequisites, and idempotency within dependency graphs are rejected before enqueueing. + ## Recurring Schedules Recurring schedules are keyed definitions. Calling `CreateOrUpdateAsync` at startup is the intended reconciliation model. diff --git a/src/Sheddueller.Dashboard/Components/Pages/JobDetail.razor b/src/Sheddueller.Dashboard/Components/Pages/JobDetail.razor index 781c06d..2f475c2 100644 --- a/src/Sheddueller.Dashboard/Components/Pages/JobDetail.razor +++ b/src/Sheddueller.Dashboard/Components/Pages/JobDetail.razor @@ -213,6 +213,32 @@ HrefFactory="GroupFilterHref" LinkAriaLabelPrefix="Filter jobs by group" /> + + @if (_detail.PrerequisiteJobIds.Count > 0) + { +
+ Prerequisites +
+ @foreach (var prerequisiteJobId in _detail.PrerequisiteJobIds) + { + + } +
+
+ } + + @if (_detail.DependentJobIds.Count > 0) + { +
+ Dependents +
+ @foreach (var dependentJobId in _detail.DependentJobIds) + { + + } +
+
+ }
diff --git a/src/Sheddueller.Dashboard/Components/Pages/Jobs.razor b/src/Sheddueller.Dashboard/Components/Pages/Jobs.razor index c70a94b..7da158f 100644 --- a/src/Sheddueller.Dashboard/Components/Pages/Jobs.razor +++ b/src/Sheddueller.Dashboard/Components/Pages/Jobs.razor @@ -1261,6 +1261,7 @@ { Kind: JobQueuePositionKind.Claimable, Position: { } position } => string.Create(CultureInfo.InvariantCulture, $"#{position}"), { Kind: JobQueuePositionKind.Claimable } => "Ready", { Kind: JobQueuePositionKind.Claimed } => "Running", + { Kind: JobQueuePositionKind.WaitingForDependencies } => "Dependencies", { Kind: JobQueuePositionKind.BlockedByConcurrency } => "Blocked", { Kind: JobQueuePositionKind.RetryWaiting } => "Retry", { Kind: JobQueuePositionKind.Delayed } => "Delayed", @@ -1275,6 +1276,7 @@ { JobQueuePositionKind.Claimable => "claimable", JobQueuePositionKind.Claimed => "claimed", + JobQueuePositionKind.WaitingForDependencies => "waiting", JobQueuePositionKind.BlockedByConcurrency => "blocked", JobQueuePositionKind.RetryWaiting => "waiting", JobQueuePositionKind.Delayed => "delayed", diff --git a/src/Sheddueller.Dashboard/Internal/DashboardFormat.cs b/src/Sheddueller.Dashboard/Internal/DashboardFormat.cs index 8508df6..6de13db 100644 --- a/src/Sheddueller.Dashboard/Internal/DashboardFormat.cs +++ b/src/Sheddueller.Dashboard/Internal/DashboardFormat.cs @@ -149,6 +149,7 @@ public static string QueueKind( JobQueuePositionKind.Claimable => "claimable", JobQueuePositionKind.Delayed => "delayed", JobQueuePositionKind.RetryWaiting => "retry_waiting", + JobQueuePositionKind.WaitingForDependencies => "waiting_for_dependencies", JobQueuePositionKind.BlockedByConcurrency => "blocked_by_concurrency", JobQueuePositionKind.Claimed => "running_active", JobQueuePositionKind.Terminal => "terminal", diff --git a/src/Sheddueller.Postgres/Internal/Operations/EnqueueJobOperation.cs b/src/Sheddueller.Postgres/Internal/Operations/EnqueueJobOperation.cs index e5a5f75..c084226 100644 --- a/src/Sheddueller.Postgres/Internal/Operations/EnqueueJobOperation.cs +++ b/src/Sheddueller.Postgres/Internal/Operations/EnqueueJobOperation.cs @@ -33,6 +33,7 @@ public static async ValueTask> ExecuteManyAsync( } var requestSnapshot = requests.ToArray(); + ValidateDependencyGraph(requestSnapshot); foreach (var request in requestSnapshot) { ArgumentNullException.ThrowIfNull(request); @@ -50,6 +51,7 @@ public static async ValueTask> ExecuteManyAsync( await CreateStagingTablesAsync(connection, transaction, cancellationToken).ConfigureAwait(false); await CopyJobsAsync(connection, requestSnapshot, cancellationToken).ConfigureAwait(false); await CopyGroupsAsync(connection, requestSnapshot, cancellationToken).ConfigureAwait(false); + await CopyDependenciesAsync(connection, requestSnapshot, cancellationToken).ConfigureAwait(false); await CopyTagsAsync(connection, requestSnapshot, cancellationToken).ConfigureAwait(false); await CopyEventsAsync(connection, requestSnapshot, cancellationToken).ConfigureAwait(false); await EnsureNoDuplicateJobIdsAsync(context, connection, transaction, cancellationToken).ConfigureAwait(false); @@ -57,6 +59,7 @@ public static async ValueTask> ExecuteManyAsync( var results = await InsertStagedJobsAsync(context, connection, transaction, cancellationToken).ConfigureAwait(false); await InsertStagedGroupsAsync(context, connection, transaction, cancellationToken).ConfigureAwait(false); + await InsertStagedDependenciesAsync(context, connection, transaction, cancellationToken).ConfigureAwait(false); await InsertStagedTagsAsync(context, connection, transaction, cancellationToken).ConfigureAwait(false); await InsertStagedEventsAsync(context, connection, transaction, cancellationToken).ConfigureAwait(false); await PostgresMetricsRollups.RecordStagedQueuedJobsAsync(context, connection, transaction, cancellationToken) @@ -106,6 +109,11 @@ create temp table sheddueller_enqueue_groups ( group_key text not null ) on commit drop; + create temp table sheddueller_enqueue_dependencies ( + job_id uuid not null, + prerequisite_job_id uuid not null + ) on commit drop; + create temp table sheddueller_enqueue_tags ( job_id uuid not null, ordinal integer not null, @@ -265,6 +273,34 @@ from stdin (format binary) await importer.CompleteAsync(cancellationToken).ConfigureAwait(false); } + private static async ValueTask CopyDependenciesAsync( + NpgsqlConnection connection, + EnqueueJobRequest[] requests, + CancellationToken cancellationToken) + { + await using var importer = await connection.BeginBinaryImportAsync( + """ + copy sheddueller_enqueue_dependencies ( + job_id, + prerequisite_job_id) + from stdin (format binary) + """, + cancellationToken) + .ConfigureAwait(false); + + foreach (var request in requests) + { + foreach (var prerequisiteJobId in request.PrerequisiteJobIds ?? []) + { + await importer.StartRowAsync(cancellationToken).ConfigureAwait(false); + await importer.WriteAsync(request.JobId, NpgsqlDbType.Uuid, cancellationToken).ConfigureAwait(false); + await importer.WriteAsync(prerequisiteJobId, NpgsqlDbType.Uuid, cancellationToken).ConfigureAwait(false); + } + } + + await importer.CompleteAsync(cancellationToken).ConfigureAwait(false); + } + private static async ValueTask CopyEventsAsync( NpgsqlConnection connection, EnqueueJobRequest[] requests, @@ -537,6 +573,28 @@ from sheddueller_enqueue_tags tag cancellationToken) .ConfigureAwait(false); + private static async ValueTask InsertStagedDependenciesAsync( + PostgresOperationContext context, + NpgsqlConnection connection, + NpgsqlTransaction transaction, + CancellationToken cancellationToken) + => await PostgresOperationContext.ExecuteCountAsync( + connection, + transaction, + $""" + insert into {context.Names.JobDependencies} (job_id, prerequisite_job_id) + select dependency.job_id, dependency.prerequisite_job_id + from sheddueller_enqueue_dependencies dependency + join sheddueller_enqueue_results dependent_result on dependent_result.job_id = dependency.job_id + join sheddueller_enqueue_results prerequisite_result on prerequisite_result.job_id = dependency.prerequisite_job_id + where dependent_result.was_enqueued = true + and prerequisite_result.was_enqueued = true + on conflict (job_id, prerequisite_job_id) do nothing; + """, + static _ => { }, + cancellationToken) + .ConfigureAwait(false); + private static async ValueTask InsertStagedEventsAsync( PostgresOperationContext context, NpgsqlConnection connection, @@ -618,6 +676,77 @@ private static async ValueTask WriteNullableAsync( await importer.WriteAsync(value.Value, dbType, cancellationToken).ConfigureAwait(false); } + private static void ValidateDependencyGraph(IReadOnlyList requests) + { + var requestsById = new Dictionary(); + foreach (var request in requests) + { + ArgumentNullException.ThrowIfNull(request); + if (!requestsById.TryAdd(request.JobId, request)) + { + throw new InvalidOperationException($"Job '{request.JobId}' appears more than once in the batch."); + } + } + + var hasDependencies = requests.Any(static request => request.PrerequisiteJobIds is { Count: > 0 }); + if (hasDependencies && requests.Any(static request => request.IdempotencyKey is not null)) + { + throw new ArgumentException("Jobs in a dependency graph cannot use idempotency.", nameof(requests)); + } + + foreach (var request in requests) + { + var prerequisites = request.PrerequisiteJobIds ?? []; + if (prerequisites.Count != prerequisites.Distinct().Count()) + { + throw new ArgumentException($"Job '{request.JobId}' contains duplicate prerequisites.", nameof(requests)); + } + + foreach (var prerequisiteJobId in prerequisites) + { + if (prerequisiteJobId == request.JobId) + { + throw new ArgumentException($"Job '{request.JobId}' cannot depend on itself.", nameof(requests)); + } + + if (!requestsById.ContainsKey(prerequisiteJobId)) + { + throw new ArgumentException( + $"Prerequisite job '{prerequisiteJobId}' for job '{request.JobId}' is not included in the batch.", + nameof(requests)); + } + } + } + + var visiting = new HashSet(); + var visited = new HashSet(); + foreach (var request in requests) + { + visit(request.JobId); + } + + void visit(Guid jobId) + { + if (visited.Contains(jobId)) + { + return; + } + + if (!visiting.Add(jobId)) + { + throw new ArgumentException("Job dependency graphs cannot contain cycles.", nameof(requests)); + } + + foreach (var prerequisiteJobId in requestsById[jobId].PrerequisiteJobIds ?? []) + { + visit(prerequisiteJobId); + } + + visiting.Remove(jobId); + visited.Add(jobId); + } + } + private static async ValueTask WriteNullableAsync( NpgsqlBinaryImporter importer, string? value, diff --git a/src/Sheddueller.Postgres/Internal/Operations/PostgresJobInspectionOperation.cs b/src/Sheddueller.Postgres/Internal/Operations/PostgresJobInspectionOperation.cs index a9f1599..0f27bca 100644 --- a/src/Sheddueller.Postgres/Internal/Operations/PostgresJobInspectionOperation.cs +++ b/src/Sheddueller.Postgres/Internal/Operations/PostgresJobInspectionOperation.cs @@ -305,6 +305,8 @@ await CreateSummaryAsync(context, connection, row, cancellationToken).ConfigureA { Invocation = await ReadInvocationAsync(context, connection, row, cancellationToken).ConfigureAwait(false), RetryCloneJobIds = await ReadRetryCloneJobIdsAsync(context, connection, jobId, cancellationToken).ConfigureAwait(false), + PrerequisiteJobIds = await ReadPrerequisiteJobIdsAsync(context, connection, jobId, cancellationToken).ConfigureAwait(false), + DependentJobIds = await ReadDependentJobIdsAsync(context, connection, jobId, cancellationToken).ConfigureAwait(false), }; } @@ -335,6 +337,15 @@ public static async ValueTask GetQueuePositionAsync( return new JobQueuePosition(jobId, JobQueuePositionKind.Claimed, Position: null, "Job is currently claimed."); } + if (await HasUnfinishedPrerequisitesAsync(context, connection, jobId, cancellationToken).ConfigureAwait(false)) + { + return new JobQueuePosition( + jobId, + JobQueuePositionKind.WaitingForDependencies, + Position: null, + "Job is waiting for prerequisite jobs to become terminal."); + } + var now = await ReadCurrentTimestampAsync(connection, cancellationToken).ConfigureAwait(false); if (row.NotBeforeUtc is { } notBeforeUtc && notBeforeUtc > now) { @@ -649,6 +660,17 @@ private static async ValueTask> Read { var positions = new Dictionary(rows.Count); var readyQueuedRows = new List(); + var queuedJobIds = rows + .Where(static row => row.State == JobState.Queued) + .Select(static row => row.JobId) + .Distinct() + .ToArray(); + var dependencyBlockedJobIds = await ReadDependencyBlockedJobIdsAsync( + context, + connection, + queuedJobIds, + cancellationToken) + .ConfigureAwait(false); DateTimeOffset? resolvedNowUtc = null; foreach (var row in rows) @@ -669,6 +691,16 @@ private static async ValueTask> Read break; case JobState.Queued: + if (dependencyBlockedJobIds.Contains(row.JobId)) + { + positions[row.JobId] = new JobQueuePosition( + row.JobId, + JobQueuePositionKind.WaitingForDependencies, + Position: null, + "Job is waiting for prerequisite jobs to become terminal."); + break; + } + resolvedNowUtc ??= nowUtc ?? await ReadCurrentTimestampAsync(connection, cancellationToken).ConfigureAwait(false); if (row.NotBeforeUtc is { } notBeforeUtc && notBeforeUtc > resolvedNowUtc.Value) { @@ -730,6 +762,13 @@ with claimable as ( from {context.Names.Jobs} job where job.state = 'Queued' and (job.not_before_utc is null or job.not_before_utc <= @now_utc) + and not exists ( + select 1 + from {context.Names.JobDependencies} dependency + join {context.Names.Jobs} prerequisite on prerequisite.job_id = dependency.prerequisite_job_id + where dependency.job_id = job.job_id + and prerequisite.state not in ('Completed', 'Failed', 'Canceled') + ) and not exists ( select 1 from {context.Names.JobConcurrencyGroups} job_group @@ -801,6 +840,13 @@ with claimable as ( from {context.Names.Jobs} job where job.state = 'Queued' and (job.not_before_utc is null or job.not_before_utc <= transaction_timestamp()) + and not exists ( + select 1 + from {context.Names.JobDependencies} dependency + join {context.Names.Jobs} prerequisite on prerequisite.job_id = dependency.prerequisite_job_id + where dependency.job_id = job.job_id + and prerequisite.state not in ('Completed', 'Failed', 'Canceled') + ) and not exists ( select 1 from {context.Names.JobConcurrencyGroups} job_group @@ -1079,6 +1125,113 @@ select job_id return jobIds; } + private static async ValueTask> ReadPrerequisiteJobIdsAsync( + PostgresOperationContext context, + NpgsqlConnection connection, + Guid jobId, + CancellationToken cancellationToken) + { + await using var command = connection.CreateCommand(); + command.CommandText = + $""" + select prerequisite_job_id + from {context.Names.JobDependencies} + where job_id = @job_id + order by prerequisite_job_id; + """; + command.Parameters.AddWithValue("job_id", jobId); + + return await ReadJobIdsAsync(command, cancellationToken).ConfigureAwait(false); + } + + private static async ValueTask> ReadDependentJobIdsAsync( + PostgresOperationContext context, + NpgsqlConnection connection, + Guid jobId, + CancellationToken cancellationToken) + { + await using var command = connection.CreateCommand(); + command.CommandText = + $""" + select job_id + from {context.Names.JobDependencies} + where prerequisite_job_id = @job_id + order by job_id; + """; + command.Parameters.AddWithValue("job_id", jobId); + + return await ReadJobIdsAsync(command, cancellationToken).ConfigureAwait(false); + } + + private static async ValueTask> ReadJobIdsAsync( + NpgsqlCommand command, + CancellationToken cancellationToken) + { + var jobIds = new List(); + await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); + while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) + { + jobIds.Add(reader.GetGuid(0)); + } + + return jobIds; + } + + private static async ValueTask HasUnfinishedPrerequisitesAsync( + PostgresOperationContext context, + NpgsqlConnection connection, + Guid jobId, + CancellationToken cancellationToken) + { + await using var command = connection.CreateCommand(); + command.CommandText = + $""" + select exists ( + select 1 + from {context.Names.JobDependencies} dependency + join {context.Names.Jobs} prerequisite on prerequisite.job_id = dependency.prerequisite_job_id + where dependency.job_id = @job_id + and prerequisite.state not in ('Completed', 'Failed', 'Canceled') + ); + """; + command.Parameters.AddWithValue("job_id", jobId); + + return (bool)(await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false) + ?? throw new InvalidOperationException("PostgreSQL did not return a dependency status.")); + } + + private static async ValueTask> ReadDependencyBlockedJobIdsAsync( + PostgresOperationContext context, + NpgsqlConnection connection, + Guid[] jobIds, + CancellationToken cancellationToken) + { + if (jobIds.Length == 0) + { + return []; + } + + await using var command = connection.CreateCommand(); + command.CommandText = + $""" + select distinct dependency.job_id + from {context.Names.JobDependencies} dependency + join {context.Names.Jobs} prerequisite on prerequisite.job_id = dependency.prerequisite_job_id + where dependency.job_id = any(@job_ids) + and prerequisite.state not in ('Completed', 'Failed', 'Canceled'); + """; + command.Parameters.AddWithValue("job_ids", jobIds); + + var blocked = new HashSet(); + await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); + while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) + { + blocked.Add(reader.GetGuid(0)); + } + + return blocked; + } + private static async ValueTask ReadCurrentTimestampAsync( NpgsqlConnection connection, CancellationToken cancellationToken) diff --git a/src/Sheddueller.Postgres/Internal/Operations/PostgresJobRetentionOperation.cs b/src/Sheddueller.Postgres/Internal/Operations/PostgresJobRetentionOperation.cs index edd1ec6..fb34591 100644 --- a/src/Sheddueller.Postgres/Internal/Operations/PostgresJobRetentionOperation.cs +++ b/src/Sheddueller.Postgres/Internal/Operations/PostgresJobRetentionOperation.cs @@ -63,11 +63,20 @@ private static async ValueTask DeleteTerminalJobsAsync( $""" with candidates as ( select job_id - from {context.Names.Jobs} + from {context.Names.Jobs} job where - (state = 'Completed' and @completed_before_utc is not null and completed_at_utc < @completed_before_utc) - or (state = 'Failed' and @failed_before_utc is not null and failed_at_utc < @failed_before_utc) - or (state = 'Canceled' and @canceled_before_utc is not null and canceled_at_utc < @canceled_before_utc) + ( + (state = 'Completed' and @completed_before_utc is not null and completed_at_utc < @completed_before_utc) + or (state = 'Failed' and @failed_before_utc is not null and failed_at_utc < @failed_before_utc) + or (state = 'Canceled' and @canceled_before_utc is not null and canceled_at_utc < @canceled_before_utc) + ) + and not exists ( + select 1 + from {context.Names.JobDependencies} dependency + join {context.Names.Jobs} dependent on dependent.job_id = dependency.job_id + where dependency.prerequisite_job_id = job.job_id + and dependent.state not in ('Completed', 'Failed', 'Canceled') + ) order by coalesce(completed_at_utc, failed_at_utc, canceled_at_utc) asc, enqueue_sequence asc limit @batch_size for update skip locked diff --git a/src/Sheddueller.Postgres/Internal/Operations/TryClaimNextJobOperation.cs b/src/Sheddueller.Postgres/Internal/Operations/TryClaimNextJobOperation.cs index 1f7e509..6d89435 100644 --- a/src/Sheddueller.Postgres/Internal/Operations/TryClaimNextJobOperation.cs +++ b/src/Sheddueller.Postgres/Internal/Operations/TryClaimNextJobOperation.cs @@ -116,6 +116,13 @@ select job.job_id from {context.Names.Jobs} job where job.state = 'Queued' and (job.not_before_utc is null or job.not_before_utc <= transaction_timestamp()) + and not exists ( + select 1 + from {context.Names.JobDependencies} dependency + join {context.Names.Jobs} prerequisite on prerequisite.job_id = dependency.prerequisite_job_id + where dependency.job_id = job.job_id + and prerequisite.state not in ('Completed', 'Failed', 'Canceled') + ) and not exists ( select 1 from {context.Names.JobConcurrencyGroups} job_group @@ -163,6 +170,13 @@ select min(rate_blocked.next_claim_at_utc) join {context.Names.ConcurrencyGroups} concurrency_group on concurrency_group.group_key = job_group.group_key where job.state = 'Queued' and (job.not_before_utc is null or job.not_before_utc <= transaction_timestamp()) + and not exists ( + select 1 + from {context.Names.JobDependencies} dependency + join {context.Names.Jobs} prerequisite on prerequisite.job_id = dependency.prerequisite_job_id + where dependency.job_id = job.job_id + and prerequisite.state not in ('Completed', 'Failed', 'Canceled') + ) and concurrency_group.effective_rate_permit_count is not null and concurrency_group.rate_theoretical_arrival_at_utc > clock_timestamp() group by job.job_id diff --git a/src/Sheddueller.Postgres/Internal/PostgresMigrator.cs b/src/Sheddueller.Postgres/Internal/PostgresMigrator.cs index df019d5..6ab1052 100644 --- a/src/Sheddueller.Postgres/Internal/PostgresMigrator.cs +++ b/src/Sheddueller.Postgres/Internal/PostgresMigrator.cs @@ -117,6 +117,13 @@ constraint jobs_schedule_occurrence_kind_check check (schedule_occurrence_kind i primary key (job_id, group_key) ); + create table if not exists {this._names.JobDependencies} ( + job_id uuid not null references {this._names.Jobs}(job_id) on delete cascade, + prerequisite_job_id uuid not null references {this._names.Jobs}(job_id) on delete cascade, + primary key (job_id, prerequisite_job_id), + constraint job_dependencies_not_self_check check (job_id <> prerequisite_job_id) + ); + alter table {this._names.Jobs} add column if not exists job_event_sequence bigint not null default 0, add column if not exists retry_clone_source_job_id uuid null, @@ -457,6 +464,9 @@ constraint settings_setting_key_check check (length(setting_key) > 0) create index if not exists idx_job_concurrency_groups_group_key on {this._names.JobConcurrencyGroups} (group_key); + create index if not exists idx_job_dependencies_prerequisite_job_id + on {this._names.JobDependencies} (prerequisite_job_id); + create index if not exists idx_job_tags_name_value_job_id on {this._names.JobTags} (name, value, job_id); diff --git a/src/Sheddueller.Postgres/Internal/PostgresNames.cs b/src/Sheddueller.Postgres/Internal/PostgresNames.cs index be0a4ac..ed84978 100644 --- a/src/Sheddueller.Postgres/Internal/PostgresNames.cs +++ b/src/Sheddueller.Postgres/Internal/PostgresNames.cs @@ -4,7 +4,7 @@ namespace Sheddueller.Postgres.Internal; internal sealed class PostgresNames { - public const int ExpectedSchemaVersion = 14; + public const int ExpectedSchemaVersion = 15; public const string WakeupChannel = "sheddueller_wakeup"; public const string JobEventChannel = "sheddueller_job_event"; @@ -15,6 +15,7 @@ public PostgresNames(string schemaName) this.SchemaInfo = this.Table("schema_info"); this.Jobs = this.Table("jobs"); this.JobConcurrencyGroups = this.Table("job_concurrency_groups"); + this.JobDependencies = this.Table("job_dependencies"); this.JobTags = this.Table("job_tags"); this.ConcurrencyGroups = this.Table("concurrency_groups"); this.RecurringSchedules = this.Table("recurring_schedules"); @@ -38,6 +39,8 @@ public PostgresNames(string schemaName) public string JobConcurrencyGroups { get; } + public string JobDependencies { get; } + public string JobTags { get; } public string ConcurrencyGroups { get; } diff --git a/src/Sheddueller.Testing/FakeEnqueuedJob.cs b/src/Sheddueller.Testing/FakeEnqueuedJob.cs index d8ced66..e3b533a 100644 --- a/src/Sheddueller.Testing/FakeEnqueuedJob.cs +++ b/src/Sheddueller.Testing/FakeEnqueuedJob.cs @@ -23,7 +23,8 @@ internal FakeEnqueuedJob( IReadOnlyList serializableParameterTypes, IReadOnlyList serializableArguments, SerializedJobPayload serializedArguments, - JobSubmission submission) + JobSubmission submission, + IReadOnlyList prerequisiteJobIds) { this.JobId = jobId; this.EnqueueSequence = enqueueSequence; @@ -38,6 +39,7 @@ internal FakeEnqueuedJob( this.SerializableArguments = ToReadOnlyCollection(serializableArguments); this.SerializedArgumentsStorage = ClonePayload(serializedArguments); this.Submission = submission; + this.PrerequisiteJobIds = ToReadOnlyCollection(prerequisiteJobIds); } /// @@ -105,6 +107,11 @@ internal FakeEnqueuedJob( /// public JobSubmission Submission { get; } + /// + /// Gets the jobs in the same batch that must become terminal before this job may run. + /// + public IReadOnlyList PrerequisiteJobIds { get; } + internal SerializedJobPayload StoredSerializedArguments => this.SerializedArgumentsStorage; private SerializedJobPayload SerializedArgumentsStorage { get; } diff --git a/src/Sheddueller.Testing/FakeJobEnqueuer.cs b/src/Sheddueller.Testing/FakeJobEnqueuer.cs index 3a7b329..b35b9aa 100644 --- a/src/Sheddueller.Testing/FakeJobEnqueuer.cs +++ b/src/Sheddueller.Testing/FakeJobEnqueuer.cs @@ -58,7 +58,7 @@ public async ValueTask EnqueueAsync( lock (this._syncRoot) { - var recordedJob = this.CreateRecordedJob(preparedJob, batchId: null, batchIndex: null); + var recordedJob = this.CreateRecordedJob(preparedJob, batchId: null, batchIndex: null, prerequisiteJobIds: []); this._jobs.Add(recordedJob); return recordedJob.JobId; @@ -75,7 +75,7 @@ public async ValueTask EnqueueAsync( lock (this._syncRoot) { - var recordedJob = this.CreateRecordedJob(preparedJob, batchId: null, batchIndex: null); + var recordedJob = this.CreateRecordedJob(preparedJob, batchId: null, batchIndex: null, prerequisiteJobIds: []); this._jobs.Add(recordedJob); return recordedJob.JobId; @@ -92,7 +92,7 @@ public async ValueTask EnqueueAsync( lock (this._syncRoot) { - var recordedJob = this.CreateRecordedJob(preparedJob, batchId: null, batchIndex: null); + var recordedJob = this.CreateRecordedJob(preparedJob, batchId: null, batchIndex: null, prerequisiteJobIds: []); this._jobs.Add(recordedJob); return recordedJob.JobId; @@ -109,7 +109,7 @@ public async ValueTask EnqueueAsync( lock (this._syncRoot) { - var recordedJob = this.CreateRecordedJob(preparedJob, batchId: null, batchIndex: null); + var recordedJob = this.CreateRecordedJob(preparedJob, batchId: null, batchIndex: null, prerequisiteJobIds: []); this._jobs.Add(recordedJob); return recordedJob.JobId; @@ -126,7 +126,7 @@ public async ValueTask EnqueueAsync( lock (this._syncRoot) { - var recordedJob = this.CreateRecordedJob(preparedJob, batchId: null, batchIndex: null); + var recordedJob = this.CreateRecordedJob(preparedJob, batchId: null, batchIndex: null, prerequisiteJobIds: []); this._jobs.Add(recordedJob); return recordedJob.JobId; @@ -143,7 +143,7 @@ public async ValueTask EnqueueAsync( lock (this._syncRoot) { - var recordedJob = this.CreateRecordedJob(preparedJob, batchId: null, batchIndex: null); + var recordedJob = this.CreateRecordedJob(preparedJob, batchId: null, batchIndex: null, prerequisiteJobIds: []); this._jobs.Add(recordedJob); return recordedJob.JobId; @@ -160,7 +160,7 @@ public async ValueTask EnqueueAsync( lock (this._syncRoot) { - var recordedJob = this.CreateRecordedJob(preparedJob, batchId: null, batchIndex: null); + var recordedJob = this.CreateRecordedJob(preparedJob, batchId: null, batchIndex: null, prerequisiteJobIds: []); this._jobs.Add(recordedJob); return recordedJob.JobId; @@ -177,7 +177,7 @@ public async ValueTask EnqueueAsync( lock (this._syncRoot) { - var recordedJob = this.CreateRecordedJob(preparedJob, batchId: null, batchIndex: null); + var recordedJob = this.CreateRecordedJob(preparedJob, batchId: null, batchIndex: null, prerequisiteJobIds: []); this._jobs.Add(recordedJob); return recordedJob.JobId; @@ -198,7 +198,9 @@ public async ValueTask> EnqueueManyAsync( } var jobSnapshot = jobs.ToArray(); + ValidateDependencyGraph(jobSnapshot); var preparedJobs = new PreparedJob[jobSnapshot.Length]; + var preparedByItem = new Dictionary(ReferenceEqualityComparer.Instance); for (var i = 0; i < jobSnapshot.Length; i++) { @@ -208,6 +210,7 @@ public async ValueTask> EnqueueManyAsync( preparedJobs[i] = await this .PrepareJobAsync(JobExpressionParser.Parse(job.ServiceType, job.Work), job.Submission, cancellationToken) .ConfigureAwait(false); + preparedByItem.Add(job, preparedJobs[i]); } lock (this._syncRoot) @@ -217,7 +220,10 @@ public async ValueTask> EnqueueManyAsync( for (var i = 0; i < preparedJobs.Length; i++) { - var recordedJob = this.CreateRecordedJob(preparedJobs[i], batchId, i); + var prerequisiteJobIds = jobSnapshot[i].Prerequisites + .Select(prerequisite => preparedByItem[prerequisite].JobId) + .ToArray(); + var recordedJob = this.CreateRecordedJob(preparedJobs[i], batchId, i, prerequisiteJobIds); this._jobs.Add(recordedJob); jobIds[i] = recordedJob.JobId; } @@ -352,7 +358,8 @@ private async ValueTask SerializeArgumentsAsync( private FakeEnqueuedJob CreateRecordedJob( PreparedJob preparedJob, Guid? batchId, - int? batchIndex) + int? batchIndex, + IReadOnlyList prerequisiteJobIds) => new( preparedJob.JobId, this._nextEnqueueSequence++, @@ -366,7 +373,63 @@ [.. preparedJob.ParsedJob.MethodParameterTypeNames.Select(TypeNameFormatter.Reso preparedJob.ParsedJob.SerializableParameterTypes, preparedJob.ParsedJob.SerializableArguments, preparedJob.SerializedArguments, - preparedJob.Submission); + preparedJob.Submission, + prerequisiteJobIds); + + private static void ValidateDependencyGraph(IReadOnlyList jobs) + { + var submitted = new HashSet(ReferenceEqualityComparer.Instance); + foreach (var job in jobs) + { + ArgumentNullException.ThrowIfNull(job, nameof(jobs)); + if (!submitted.Add(job)) + { + throw new ArgumentException("A job enqueue item cannot appear more than once in a batch.", nameof(jobs)); + } + } + + var hasDependencies = jobs.Any(static job => job.Prerequisites.Count > 0); + if (hasDependencies && jobs.Any(static job => job.Submission?.IdempotencyKind is not null and not JobIdempotencyKind.None)) + { + throw new ArgumentException("Jobs in a dependency graph cannot use idempotency.", nameof(jobs)); + } + + foreach (var job in jobs) + { + if (job.Prerequisites.Any(prerequisite => !submitted.Contains(prerequisite))) + { + throw new ArgumentException("Every prerequisite job must be included in the same batch.", nameof(jobs)); + } + } + + var visiting = new HashSet(ReferenceEqualityComparer.Instance); + var visited = new HashSet(ReferenceEqualityComparer.Instance); + foreach (var job in jobs) + { + visit(job); + } + + void visit(JobEnqueueItem job) + { + if (visited.Contains(job)) + { + return; + } + + if (!visiting.Add(job)) + { + throw new ArgumentException("Job dependency graphs cannot contain cycles.", nameof(jobs)); + } + + foreach (var prerequisite in job.Prerequisites) + { + visit(prerequisite); + } + + visiting.Remove(job); + visited.Add(job); + } + } private static bool Matches( FakeEnqueuedJob job, diff --git a/src/Sheddueller/Enqueueing/JobEnqueuer.cs b/src/Sheddueller/Enqueueing/JobEnqueuer.cs index f725657..7b7e57a 100644 --- a/src/Sheddueller/Enqueueing/JobEnqueuer.cs +++ b/src/Sheddueller/Enqueueing/JobEnqueuer.cs @@ -78,17 +78,32 @@ public async ValueTask> EnqueueManyAsync( } var jobSnapshot = jobs.ToArray(); + ValidateDependencyGraph(jobSnapshot); + var jobIdsByItem = new Dictionary(ReferenceEqualityComparer.Instance); + foreach (var job in jobSnapshot) + { + ArgumentNullException.ThrowIfNull(job, nameof(jobs)); + jobIdsByItem.Add(job, Guid.NewGuid()); + } + + var hasDependencies = jobSnapshot.Any(static job => job.Prerequisites.Count > 0); + if (hasDependencies && jobSnapshot.Any(static job => job.Submission?.IdempotencyKind is not null and not JobIdempotencyKind.None)) + { + throw new ArgumentException("Jobs in a dependency graph cannot use idempotency.", nameof(jobs)); + } + var requests = new EnqueueJobRequest[jobSnapshot.Length]; var enqueuedAtUtc = timeProvider.GetUtcNow(); for (var i = 0; i < jobSnapshot.Length; i++) { var job = jobSnapshot[i]; - ArgumentNullException.ThrowIfNull(job, nameof(jobs)); requests[i] = await this.CreateRequestAsync( JobExpressionParser.Parse(job.ServiceType, job.Work), job.Submission, + jobIdsByItem[job], enqueuedAtUtc, + [.. job.Prerequisites.Select(prerequisite => jobIdsByItem[prerequisite])], cancellationToken) .ConfigureAwait(false); } @@ -124,7 +139,9 @@ private async ValueTask EnqueueCoreAsync( var request = await this.CreateRequestAsync( parsedJob, submission, + Guid.NewGuid(), timeProvider.GetUtcNow(), + prerequisiteJobIds: [], cancellationToken) .ConfigureAwait(false); var result = await store.EnqueueAsync(request, cancellationToken).ConfigureAwait(false); @@ -144,7 +161,9 @@ private async ValueTask EnqueueCoreAsync( private async ValueTask CreateRequestAsync( ParsedJob parsedTask, JobSubmission? submission, + Guid jobId, DateTimeOffset enqueuedAtUtc, + IReadOnlyList prerequisiteJobIds, CancellationToken cancellationToken) { SubmissionValidator.ValidateIdempotency(submission); @@ -166,7 +185,6 @@ private async ValueTask CreateRequestAsync( _ => null, }; - var jobId = Guid.NewGuid(); var request = new EnqueueJobRequest( jobId, submission?.Priority ?? 0, @@ -186,8 +204,64 @@ private async ValueTask CreateRequestAsync( Tags: tags, InvocationTargetKind: parsedTask.InvocationTargetKind, MethodParameterBindings: parsedTask.MethodParameterBindings, - IdempotencyKey: idempotencyKey); + IdempotencyKey: idempotencyKey, + PrerequisiteJobIds: prerequisiteJobIds); return request; } + + private static void ValidateDependencyGraph(IReadOnlyList jobs) + { + var submitted = new HashSet(ReferenceEqualityComparer.Instance); + foreach (var job in jobs) + { + ArgumentNullException.ThrowIfNull(job, nameof(jobs)); + if (!submitted.Add(job)) + { + throw new ArgumentException("A job enqueue item cannot appear more than once in a batch.", nameof(jobs)); + } + } + + foreach (var job in jobs) + { + foreach (var prerequisite in job.Prerequisites) + { + if (!submitted.Contains(prerequisite)) + { + throw new ArgumentException("Every prerequisite job must be included in the same batch.", nameof(jobs)); + } + } + } + + var visiting = new HashSet(ReferenceEqualityComparer.Instance); + var visited = new HashSet(ReferenceEqualityComparer.Instance); + foreach (var job in jobs) + { + Visit(job, visiting, visited); + } + + static void Visit( + JobEnqueueItem job, + HashSet visiting, + HashSet visited) + { + if (visited.Contains(job)) + { + return; + } + + if (!visiting.Add(job)) + { + throw new ArgumentException("Job dependency graphs cannot contain cycles.", nameof(jobs)); + } + + foreach (var prerequisite in job.Prerequisites) + { + Visit(prerequisite, visiting, visited); + } + + visiting.Remove(job); + visited.Add(job); + } + } } diff --git a/src/Sheddueller/Inspection/Jobs/JobInspectionDetail.cs b/src/Sheddueller/Inspection/Jobs/JobInspectionDetail.cs index 9d9e9e9..3554f8f 100644 --- a/src/Sheddueller/Inspection/Jobs/JobInspectionDetail.cs +++ b/src/Sheddueller/Inspection/Jobs/JobInspectionDetail.cs @@ -19,4 +19,14 @@ public sealed record JobInspectionDetail( /// Jobs cloned from this failed job. /// public IReadOnlyList RetryCloneJobIds { get; init; } = []; + + /// + /// Jobs that must become terminal before this job is claimable. + /// + public IReadOnlyList PrerequisiteJobIds { get; init; } = []; + + /// + /// Jobs that depend on this job becoming terminal. + /// + public IReadOnlyList DependentJobIds { get; init; } = []; } diff --git a/src/Sheddueller/Inspection/Jobs/JobQueuePositionKind.cs b/src/Sheddueller/Inspection/Jobs/JobQueuePositionKind.cs index 5269f4c..1601c6c 100644 --- a/src/Sheddueller/Inspection/Jobs/JobQueuePositionKind.cs +++ b/src/Sheddueller/Inspection/Jobs/JobQueuePositionKind.cs @@ -20,6 +20,11 @@ public enum JobQueuePositionKind /// RetryWaiting, + /// + /// Job is waiting for prerequisite jobs to become terminal. + /// + WaitingForDependencies, + /// /// Job is blocked by concurrency group saturation. /// diff --git a/src/Sheddueller/JobEnqueueItem.cs b/src/Sheddueller/JobEnqueueItem.cs index 9958b96..70c5e26 100644 --- a/src/Sheddueller/JobEnqueueItem.cs +++ b/src/Sheddueller/JobEnqueueItem.cs @@ -7,6 +7,8 @@ namespace Sheddueller; /// public sealed class JobEnqueueItem { + private readonly List _prerequisites = []; + private JobEnqueueItem( Type? serviceType, LambdaExpression work, @@ -23,6 +25,48 @@ private JobEnqueueItem( internal JobSubmission? Submission { get; } + internal IReadOnlyList Prerequisites => this._prerequisites; + + /// + /// Makes this job wait until the supplied prerequisite jobs are terminal. + /// + /// Jobs from the same atomic batch that must become terminal first. + /// This enqueue item. + /// + /// Terminal prerequisites include completed, failed, and canceled jobs. Dependency graphs may contain + /// multiple levels, but every referenced prerequisite must be submitted in the same batch. + /// + public JobEnqueueItem DependsOn(IEnumerable prerequisites) + { + ArgumentNullException.ThrowIfNull(prerequisites); + + foreach (var prerequisite in prerequisites) + { + ArgumentNullException.ThrowIfNull(prerequisite, nameof(prerequisites)); + if (ReferenceEquals(this, prerequisite)) + { + throw new ArgumentException("A job cannot depend on itself.", nameof(prerequisites)); + } + + if (this._prerequisites.Contains(prerequisite)) + { + throw new ArgumentException("A prerequisite job cannot be added more than once.", nameof(prerequisites)); + } + + this._prerequisites.Add(prerequisite); + } + + return this; + } + + /// + /// Makes this job wait until the supplied prerequisite jobs are terminal. + /// + /// Jobs from the same atomic batch that must become terminal first. + /// This enqueue item. + public JobEnqueueItem DependsOn(params JobEnqueueItem[] prerequisites) + => this.DependsOn((IEnumerable)prerequisites); + /// /// Creates a batch item for a Task-returning job method call. /// diff --git a/src/Sheddueller/Storage/EnqueueJobRequest.cs b/src/Sheddueller/Storage/EnqueueJobRequest.cs index c992a84..6739063 100644 --- a/src/Sheddueller/Storage/EnqueueJobRequest.cs +++ b/src/Sheddueller/Storage/EnqueueJobRequest.cs @@ -26,4 +26,5 @@ public sealed record EnqueueJobRequest( ScheduleOccurrenceKind? ScheduleOccurrenceKind = null, JobInvocationTargetKind InvocationTargetKind = JobInvocationTargetKind.Instance, IReadOnlyList? MethodParameterBindings = null, - string? IdempotencyKey = null); + string? IdempotencyKey = null, + IReadOnlyList? PrerequisiteJobIds = null); diff --git a/test/Sheddueller.ProviderContracts/JobRetentionStoreContractTests.cs b/test/Sheddueller.ProviderContracts/JobRetentionStoreContractTests.cs index 856411f..8aca0a9 100644 --- a/test/Sheddueller.ProviderContracts/JobRetentionStoreContractTests.cs +++ b/test/Sheddueller.ProviderContracts/JobRetentionStoreContractTests.cs @@ -85,6 +85,34 @@ public async Task CleanupTerminalJobs_NullCutoff_RetainsThatTerminalState() await AssertRetainedAsync(context.Reader, oldCompleted); } + [Fact] + public async Task CleanupTerminalJobs_TerminalPrerequisiteForActiveDependent_RetainsPrerequisite() + { + await using var context = await this.CreateRetentionContextAsync(); + var old = DateTimeOffset.UtcNow.AddDays(-10); + var prerequisite = Guid.NewGuid(); + var dependent = Guid.NewGuid(); + await context.Store.EnqueueManyAsync([ + CreateRequest(prerequisite, old.AddMinutes(-1), priority: 0), + CreateRequest(dependent, old.AddMinutes(-1), priority: 100, prerequisiteJobIds: [prerequisite]), + ]); + var claimed = await ClaimAsync(context.Store, "complete-node"); + claimed.JobId.ShouldBe(prerequisite); + (await context.Store.MarkCompletedAsync( + new CompleteJobRequest(prerequisite, "complete-node", claimed.LeaseToken, old))).ShouldBeTrue(); + + var result = await context.RetentionStore.CleanupTerminalJobsAsync( + new JobRetentionCleanupRequest( + DateTimeOffset.UtcNow.AddDays(-1), + DateTimeOffset.UtcNow.AddDays(-1), + DateTimeOffset.UtcNow.AddDays(-1), + 20)); + + result.DeletedCount.ShouldBe(0); + await AssertRetainedAsync(context.Reader, prerequisite); + await AssertRetainedAsync(context.Reader, dependent); + } + private static async ValueTask CompleteJobAsync( IJobStore store, DateTimeOffset completedAtUtc) @@ -138,6 +166,23 @@ await store.EnqueueAsync(new EnqueueJobRequest( return jobId; } + private static EnqueueJobRequest CreateRequest( + Guid jobId, + DateTimeOffset enqueuedAtUtc, + int priority, + IReadOnlyList? prerequisiteJobIds = null) => new( + jobId, + priority, + typeof(JobRetentionContractService).AssemblyQualifiedName!, + nameof(JobRetentionContractService.RunAsync), + [typeof(CancellationToken).AssemblyQualifiedName!], + new SerializedJobPayload(SystemTextJsonJobPayloadSerializer.JsonContentType, "[]"u8.ToArray()), + ConcurrencyGroupKeys: [], + enqueuedAtUtc, + NotBeforeUtc: null, + MaxAttempts: 1, + PrerequisiteJobIds: prerequisiteJobIds); + private static async ValueTask ClaimAsync( IJobStore store, string nodeId = "node-1") diff --git a/test/Sheddueller.ProviderContracts/JobStoreContractTests.cs b/test/Sheddueller.ProviderContracts/JobStoreContractTests.cs index 5f7038a..5cfd40c 100644 --- a/test/Sheddueller.ProviderContracts/JobStoreContractTests.cs +++ b/test/Sheddueller.ProviderContracts/JobStoreContractTests.cs @@ -96,6 +96,55 @@ public async Task EnqueueMany_EmptyBatch_ReturnsEmptyWithoutPersistingJob() (await context.Store.TryClaimNextAsync(CreateClaimRequest("node-1"))).ShouldBeOfType(); } + [Fact] + public async Task DependencyGraph_PrerequisitesBecomeTerminal_DependentBecomesClaimable() + { + await using var context = await this.CreateContextAsync(); + var first = Guid.NewGuid(); + var second = Guid.NewGuid(); + var dependent = Guid.NewGuid(); + await context.Store.EnqueueManyAsync([ + CreateRequest(first), + CreateRequest(second), + CreateRequest(dependent, priority: 100, prerequisiteJobIds: [first, second]), + ]); + + var firstClaim = await ClaimAsync(context.Store); + firstClaim.JobId.ShouldBe(first); + (await context.Store.MarkCompletedAsync( + new CompleteJobRequest(first, "node-1", firstClaim.LeaseToken, ContractClock))).ShouldBeTrue(); + + var secondClaim = await ClaimAsync(context.Store); + secondClaim.JobId.ShouldBe(second); + (await context.Store.MarkFailedAsync( + new FailJobRequest( + second, + "node-1", + secondClaim.LeaseToken, + ContractClock, + new JobFailureInfo("TestFailure", "failed", null)))).ShouldBeTrue(); + + (await ClaimAsync(context.Store)).JobId.ShouldBe(dependent); + } + + [Fact] + public async Task DependencyGraph_UnfinishedPrerequisite_InspectionReportsWaitingForDependencies() + { + await using var context = await this.CreateContextAsync(); + var prerequisite = Guid.NewGuid(); + var dependent = Guid.NewGuid(); + await context.Store.EnqueueManyAsync([ + CreateRequest(prerequisite), + CreateRequest(dependent, priority: 100, prerequisiteJobIds: [prerequisite]), + ]); + + var position = await GetInspectionReader(context).GetQueuePositionAsync(dependent); + var detail = await GetInspectionReader(context).GetJobAsync(dependent); + + position.Kind.ShouldBe(JobQueuePositionKind.WaitingForDependencies); + detail.ShouldNotBeNull().PrerequisiteJobIds.ShouldBe([prerequisite]); + } + [Fact] public async Task EnqueueMany_DuplicateJobIdInBatch_ThrowsWithoutPersistingAnyBatchItem() { @@ -896,7 +945,8 @@ protected static EnqueueJobRequest CreateRequest( TimeSpan? retryBaseDelay = null, TimeSpan? retryMaxDelay = null, IReadOnlyList? groupKeys = null, - string? idempotencyKey = null) + string? idempotencyKey = null, + IReadOnlyList? prerequisiteJobIds = null) => new( jobId, priority, @@ -911,7 +961,8 @@ protected static EnqueueJobRequest CreateRequest( retryBackoffKind, retryBaseDelay, retryMaxDelay, - IdempotencyKey: idempotencyKey); + IdempotencyKey: idempotencyKey, + PrerequisiteJobIds: prerequisiteJobIds); protected static UpsertRecurringScheduleRequest CreateSchedule( string scheduleKey, diff --git a/test/Sheddueller.Testing.Tests/FakeJobEnqueuerTests.cs b/test/Sheddueller.Testing.Tests/FakeJobEnqueuerTests.cs index 3803f4b..fedea2e 100644 --- a/test/Sheddueller.Testing.Tests/FakeJobEnqueuerTests.cs +++ b/test/Sheddueller.Testing.Tests/FakeJobEnqueuerTests.cs @@ -125,6 +125,25 @@ public async Task EnqueueMany_MixedJobs_RecordsBatchMetadataAndReturnsIdsInInput fake.EnqueuedJobs[1].EnqueueSequence.ShouldBe(1); } + [Fact] + public async Task EnqueueMany_DependencyGraph_RecordsPrerequisiteJobIds() + { + var fake = new FakeJobEnqueuer(); + var first = JobEnqueueItem.Create( + (service, cancellationToken) => service.HandleAsync(new SamplePayload("first", 1), cancellationToken)); + var second = JobEnqueueItem.Create( + (service, cancellationToken) => service.HandleAsync(new SamplePayload("second", 2), cancellationToken)); + var aggregate = JobEnqueueItem.Create( + cancellationToken => TestJobService.StaticTaskAsync("aggregate", cancellationToken)) + .DependsOn([first, second]); + + var jobIds = await fake.EnqueueManyAsync([first, second, aggregate]); + + fake.EnqueuedJobs[0].PrerequisiteJobIds.ShouldBeEmpty(); + fake.EnqueuedJobs[1].PrerequisiteJobIds.ShouldBeEmpty(); + fake.EnqueuedJobs[2].PrerequisiteJobIds.ShouldBe([jobIds[0], jobIds[1]]); + } + [Fact] public async Task EnqueueMany_InvalidItem_DoesNotRecordAnyJobs() { diff --git a/test/Sheddueller.Tests/JobEnqueuerTests.cs b/test/Sheddueller.Tests/JobEnqueuerTests.cs index 31d6106..09e2659 100644 --- a/test/Sheddueller.Tests/JobEnqueuerTests.cs +++ b/test/Sheddueller.Tests/JobEnqueuerTests.cs @@ -281,6 +281,81 @@ public async Task EnqueueMany_MixedServiceMethods_PersistsJobsAndReturnsIdsInInp thirdRequest.InvocationTargetKind.ShouldBe(JobInvocationTargetKind.Static); } + [Fact] + public async Task EnqueueMany_DependencyGraph_PersistsArbitraryDepthUsingReturnedJobIds() + { + using var provider = CreateProvider(); + var enqueuer = provider.GetRequiredService(); + var store = provider.GetRequiredService(); + var first = JobEnqueueItem.Create( + (service, cancellationToken) => service.HandleStringAsync("first", cancellationToken)); + var second = JobEnqueueItem.Create( + (service, cancellationToken) => service.HandleStringAsync("second", cancellationToken)) + .DependsOn(first); + var third = JobEnqueueItem.Create( + (service, cancellationToken) => service.HandleStringAsync("third", cancellationToken)) + .DependsOn([first, second]); + + var jobIds = await enqueuer.EnqueueManyAsync([third, first, second]); + + store.GetRequest(jobIds[0]).PrerequisiteJobIds.ShouldBe([jobIds[1], jobIds[2]], ignoreOrder: true); + store.GetRequest(jobIds[1]).PrerequisiteJobIds.ShouldBeEmpty(); + store.GetRequest(jobIds[2]).PrerequisiteJobIds.ShouldBe([jobIds[1]]); + } + + [Fact] + public async Task EnqueueMany_DependencyOutsideBatch_ThrowsWithoutPersistingJobs() + { + using var provider = CreateProvider(); + var enqueuer = provider.GetRequiredService(); + var store = provider.GetRequiredService(); + var missing = JobEnqueueItem.Create( + (service, cancellationToken) => service.HandleStringAsync("missing", cancellationToken)); + var dependent = JobEnqueueItem.Create( + (service, cancellationToken) => service.HandleStringAsync("dependent", cancellationToken)) + .DependsOn(missing); + + await Should.ThrowAsync(() => enqueuer.EnqueueManyAsync([dependent]).AsTask()); + + store.EnqueuedRequests.ShouldBeEmpty(); + } + + [Fact] + public async Task EnqueueMany_CyclicDependencyGraph_ThrowsWithoutPersistingJobs() + { + using var provider = CreateProvider(); + var enqueuer = provider.GetRequiredService(); + var store = provider.GetRequiredService(); + var first = JobEnqueueItem.Create( + (service, cancellationToken) => service.HandleStringAsync("first", cancellationToken)); + var second = JobEnqueueItem.Create( + (service, cancellationToken) => service.HandleStringAsync("second", cancellationToken)); + first.DependsOn(second); + second.DependsOn(first); + + await Should.ThrowAsync(() => enqueuer.EnqueueManyAsync([first, second]).AsTask()); + + store.EnqueuedRequests.ShouldBeEmpty(); + } + + [Fact] + public async Task EnqueueMany_IdempotentDependencyGraph_ThrowsWithoutPersistingJobs() + { + using var provider = CreateProvider(); + var enqueuer = provider.GetRequiredService(); + var store = provider.GetRequiredService(); + var first = JobEnqueueItem.Create( + (service, cancellationToken) => service.HandleStringAsync("first", cancellationToken)); + var dependent = JobEnqueueItem.Create( + (service, cancellationToken) => service.HandleStringAsync("dependent", cancellationToken), + new JobSubmission(IdempotencyKind: JobIdempotencyKind.MethodAndArguments)) + .DependsOn(first); + + await Should.ThrowAsync(() => enqueuer.EnqueueManyAsync([first, dependent]).AsTask()); + + store.EnqueuedRequests.ShouldBeEmpty(); + } + [Fact] public async Task EnqueueMany_ProgressAwareMethod_PersistsProgressBinding() {