From 0e2ee1f10d851a3a5d1ca3e2102e8943ee2e04b3 Mon Sep 17 00:00:00 2001 From: Frank Date: Fri, 16 Oct 2026 19:54:52 +0000 Subject: [PATCH 1/2] Consolidate job definitions into JobDefinition --- docs/architecture.md | 7 +- .../src/components/JobDetail.vue | 4 +- .../DashboardClient/src/contracts.ts | 6 +- .../DashboardClient/src/query.ts | 4 +- .../DashboardJsonSerializerContext.cs | 3 +- ...ameworkCoreJobStorage.DefinitionCatalog.cs | 9 +- .../LinqToDBJobStorage.DefinitionCatalog.cs | 11 +- .../RedisJobStorage.Acquire.cs | 5 +- .../RedisJobStorage.DefinitionCatalog.cs | 8 +- .../RedisJobStorage.Definitions.cs | 2 +- .../RedisJobStorage.Records.cs | 2 +- .../RedisJsonSerializerContext.cs | 3 +- .../Apis/JobDefinitionRecord.cs | 112 ------------------ .../Apis/JobMonitoringDefinitions.cs | 4 +- .../Internals/JobDefinition.cs | 33 +++++- .../Internals/JobSchedulingService.cs | 27 +---- .../Storage/IJobStorage.cs | 3 +- .../InMemoryJobStorage.DefinitionCatalog.cs | 5 +- .../Storage/InMemoryJobStorage.Definitions.cs | 3 +- .../Storage/InMemoryJobStorage.Enqueue.cs | 3 +- .../Storage/JobDefinitionReconciliation.cs | 5 +- .../Storage/JobDefinitionRegistration.cs | 3 +- ...ingleServerJobStorage.DefinitionCatalog.cs | 3 +- .../CapturingJobStorage.cs | 3 +- src/Immediate.Jobs.Testing/JobTestHarness.cs | 3 +- ...DefinitionAcquisitionStorageConformance.cs | 25 ++-- .../DefinitionCatalogStorageConformance.cs | 9 +- .../DefinitionRemovalStorageConformance.cs | 5 +- .../Storage/MonitoringStorageConformance.cs | 3 +- .../Storage/RecurringStorageConformance.cs | 3 +- .../ReplicaStorageConformance.Enqueue.cs | 3 +- .../Storage/ReplicaStorageConformance.cs | 3 +- .../Storage/TagStorageConformance.cs | 7 +- .../DefinitionAcquisitionTests.cs | 2 +- .../Packages/DashboardPackageTests.cs | 5 + .../StorageCapabilityTests.cs | 3 +- .../JobDefinitionSerializationTests.cs | 90 ++++++++++++++ .../RedisConformanceTests.cs | 3 +- 38 files changed, 222 insertions(+), 210 deletions(-) delete mode 100644 src/Immediate.Jobs.Shared/Apis/JobDefinitionRecord.cs create mode 100644 tests/Immediate.Jobs.StorageTests/JobDefinitionSerializationTests.cs diff --git a/docs/architecture.md b/docs/architecture.md index b8b574bb..c2bfa808 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -32,7 +32,7 @@ For each `[Job]` class (see `Templates/Job.sbntxt`), nested in the user's partia current persisted definition on each call, including its acquisition limits. - **`Invoker`**: a singleton `IJobInvoker` that deserializes the payload and runs the handler through the Immediate.Handlers pipeline. -- **`JobDefinition`**: the runtime description (name, queue, attempts, timeout, backoff, cron, +- **`JobDefinition`**: the shared execution, storage, and monitoring description (name, queue, attempts, timeout, backoff, cron, overlap and misfire policies, concurrency, acquisition windows, and normalized routing tags). - **`PayloadJsonContext`**: hand-emitted System.Text.Json metadata for the payload and context types, so serialization never uses reflection. @@ -59,7 +59,10 @@ configuration surface (`ConfigureStorage`, `ConfigureWorkers`, `UseFairQueues`, - **Startup.** Initializes storage and submits the complete definition catalogue and code-defined schedules through `MergeJobDefinitionsListAsync`, even when workers are disabled. `StartAsync` completes this registration before returning. The storage - reconciles only definitions intersecting the server's effective tags. Metadata has no invoker or CLR type. + reconciles only definitions intersecting the server's effective tags. `JobDefinition` serializes queue + settings as `QueueName`, `QueuePriority`, and `QueueConcurrency`. The local invoker, CLR type, queue + object, and computed acquisition limits are excluded from JSON; definitions read from durable storage + have no runtime bindings. Execution uses the locally registered definitions. - **Polling loop.** Each iteration materializes due recurring schedules, builds a `JobAcquisitionRequest` from tag-eligible local definitions, free worker capacity, and queue/job concurrency limits, calls `IJobStorage.AcquireDueJobsAsync`, and writes the claimed records to an unbounded channel. On diff --git a/src/Immediate.Jobs.Dashboard/DashboardClient/src/components/JobDetail.vue b/src/Immediate.Jobs.Dashboard/DashboardClient/src/components/JobDetail.vue index ac0fb179..287b3914 100644 --- a/src/Immediate.Jobs.Dashboard/DashboardClient/src/components/JobDetail.vue +++ b/src/Immediate.Jobs.Dashboard/DashboardClient/src/components/JobDetail.vue @@ -10,7 +10,7 @@ import { canCancelJob, canRetryJob, type JobAcquisitionState, - type JobDefinitionRecord, + type JobDefinition, type JobExecutionRecord, type JobRecord, type JobTelemetryLink, @@ -19,7 +19,7 @@ import { formatAcquisitionStatus, formatDate, formatJobState, formatJson } from const props = withDefaults(defineProps<{ job: JobRecord; - definition?: JobDefinitionRecord | undefined; + definition?: JobDefinition | undefined; acquisition?: JobAcquisitionState | undefined; definitionLoading?: boolean; definitionError?: string | undefined; diff --git a/src/Immediate.Jobs.Dashboard/DashboardClient/src/contracts.ts b/src/Immediate.Jobs.Dashboard/DashboardClient/src/contracts.ts index 0fea7769..5610b7ff 100644 --- a/src/Immediate.Jobs.Dashboard/DashboardClient/src/contracts.ts +++ b/src/Immediate.Jobs.Dashboard/DashboardClient/src/contracts.ts @@ -68,7 +68,7 @@ export interface RecurringJobSchedule { lastRunAt: IsoDateTime | null; } -export interface JobDefinitionRecord { +export interface JobDefinition { name: string; tags: string[]; queueName: string; @@ -149,7 +149,7 @@ export interface JobAcquisitionState { isConcurrencyLimited: boolean; } -export interface JobDefinitionStatus extends JobDefinitionRecord { +export interface JobDefinitionStatus extends JobDefinition { limits: JobAcquisitionLimits; acquisition?: JobAcquisitionState | undefined; } @@ -258,6 +258,6 @@ export type RecurringJobDefinition = Omit; export interface JobMonitoringDefinitions { - jobs: JobDefinitionRecord[]; + jobs: JobDefinition[]; recurring: RecurringJobDefinition[]; } diff --git a/src/Immediate.Jobs.Dashboard/DashboardClient/src/query.ts b/src/Immediate.Jobs.Dashboard/DashboardClient/src/query.ts index b0af1516..df1ba173 100644 --- a/src/Immediate.Jobs.Dashboard/DashboardClient/src/query.ts +++ b/src/Immediate.Jobs.Dashboard/DashboardClient/src/query.ts @@ -23,7 +23,7 @@ import type { DashboardJobPage, DashboardState, JobFilters, - JobDefinitionRecord, + JobDefinition, JobDefinitionStatus, JobMonitoringDefinitions, JobMonitoringSnapshot, @@ -105,7 +105,7 @@ export function useJobDefinitionQuery(jobName: MaybeRefOrGetter(() => { + data: computed(() => { if (snapshot.data.value && !acquisition.value) { return undefined; } diff --git a/src/Immediate.Jobs.Dashboard/DashboardJsonSerializerContext.cs b/src/Immediate.Jobs.Dashboard/DashboardJsonSerializerContext.cs index 3efbb041..00153912 100644 --- a/src/Immediate.Jobs.Dashboard/DashboardJsonSerializerContext.cs +++ b/src/Immediate.Jobs.Dashboard/DashboardJsonSerializerContext.cs @@ -1,11 +1,12 @@ using System.Text.Json.Serialization; using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; namespace Immediate.Jobs.Dashboard; [JsonSourceGenerationOptions(PropertyNamingPolicy = JsonKnownNamingPolicy.CamelCase, UseStringEnumConverter = true)] [JsonSerializable(typeof(JobMonitoringSnapshot))] -[JsonSerializable(typeof(IReadOnlyList))] +[JsonSerializable(typeof(IReadOnlyList))] [JsonSerializable(typeof(JobMonitoringDefinitions))] [JsonSerializable(typeof(DashboardState))] [JsonSerializable(typeof(DashboardJobPage))] diff --git a/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.DefinitionCatalog.cs b/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.DefinitionCatalog.cs index d343f9fe..73353657 100644 --- a/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.DefinitionCatalog.cs +++ b/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.DefinitionCatalog.cs @@ -2,6 +2,7 @@ using System.Data; using System.Globalization; using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; using Immediate.Jobs.Shared.Storage; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Logging; @@ -95,7 +96,7 @@ await context.Set() } /// - public async ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default) + public async ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default) { GetJobDefinitionsAsyncCalled(); cancellationToken.ThrowIfCancellationRequested(); @@ -103,7 +104,7 @@ public async ValueTask> GetJobDefinitionsAsyn return await ReadWithStrategyAsync(ReadJobDefinitionsAsync, cancellationToken); } - private static async Task> ReadJobDefinitionsAsync(TContext context, CancellationToken cancellationToken) + private static async Task> ReadJobDefinitionsAsync(TContext context, CancellationToken cancellationToken) { var definitions = await context.Set() .AsNoTracking() @@ -143,7 +144,7 @@ private async Task EnsureDefinitionCatalogAsync(CancellationToken cancellationTo LastRunAt = entity.LastRunAt, }; - private static ImmediateJobDefinitionMetadataEntity ToEntity(JobDefinitionRecord definition) => new() + private static ImmediateJobDefinitionMetadataEntity ToEntity(JobDefinition definition) => new() { Name = definition.Name, QueueName = definition.QueueName, @@ -165,7 +166,7 @@ private async Task EnsureDefinitionCatalogAsync(CancellationToken cancellationTo Tags = JsonSerializer.Serialize([.. definition.Tags], EntityFrameworkCoreJsonSerializerContext.Default.StringArray), }; - private static JobDefinitionRecord ToRecord(ImmediateJobDefinitionMetadataEntity entity) => new() + private static JobDefinition ToRecord(ImmediateJobDefinitionMetadataEntity entity) => new() { Name = entity.Name, QueueName = entity.QueueName, diff --git a/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.DefinitionCatalog.cs b/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.DefinitionCatalog.cs index 00235d2a..9edc7f92 100644 --- a/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.DefinitionCatalog.cs +++ b/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.DefinitionCatalog.cs @@ -2,6 +2,7 @@ using System.Data; using System.Globalization; using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; using Immediate.Jobs.Shared.Storage; using LinqToDB; using LinqToDB.Async; @@ -68,7 +69,7 @@ private Task RemoveEmptyFairQueueGroupsAsync(DataConnection connection, Can .DeleteAsync(token); /// - public async ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default) + public async ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default) { GetJobDefinitionsAsyncCalled(); cancellationToken.ThrowIfCancellationRequested(); @@ -77,7 +78,7 @@ public async ValueTask> GetJobDefinitionsAsyn return await ReadJobDefinitionsAsync(connection, cancellationToken); } - private async Task> ReadJobDefinitionsAsync(DataConnection connection, CancellationToken cancellationToken) + private async Task> ReadJobDefinitionsAsync(DataConnection connection, CancellationToken cancellationToken) { var definitions = await Definitions(connection).ToListAsync(cancellationToken); return [.. definitions.Select(ToRecord).OrderBy(static definition => definition.Name, StringComparer.OrdinalIgnoreCase)]; @@ -165,7 +166,7 @@ await Recurring(connection).Where(item => item.Name == schedule.Name) private async Task MergeDefinitionMetadataAsync( DataConnection connection, - IReadOnlyList definitions, + IReadOnlyList definitions, IQueryable serverTags, CancellationToken cancellationToken ) @@ -255,7 +256,7 @@ CancellationToken cancellationToken [Sql.Expression(ProviderName.SQLite, "{0} COLLATE BINARY = {1} COLLATE BINARY", ServerSideOnly = true, IsPredicate = true)] private static bool EqualsOrdinal(string left, string right) => throw new InvalidOperationException("This function must be translated to SQL."); - private static ImmediateJobDefinitionMetadataEntity ToEntity(JobDefinitionRecord definition) => new() + private static ImmediateJobDefinitionMetadataEntity ToEntity(JobDefinition definition) => new() { Name = definition.Name, QueueName = definition.QueueName, @@ -277,7 +278,7 @@ CancellationToken cancellationToken Tags = JsonSerializer.Serialize([.. definition.Tags], LinqToDBJsonSerializerContext.Default.StringArray), }; - private static JobDefinitionRecord ToRecord(ImmediateJobDefinitionMetadataEntity entity) => new() + private static JobDefinition ToRecord(ImmediateJobDefinitionMetadataEntity entity) => new() { Name = entity.Name, QueueName = entity.QueueName, diff --git a/src/Immediate.Jobs.Redis/RedisJobStorage.Acquire.cs b/src/Immediate.Jobs.Redis/RedisJobStorage.Acquire.cs index 85a86632..eeb8b2d4 100644 --- a/src/Immediate.Jobs.Redis/RedisJobStorage.Acquire.cs +++ b/src/Immediate.Jobs.Redis/RedisJobStorage.Acquire.cs @@ -1,5 +1,6 @@ using System.Text.Json; using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; using Immediate.Jobs.Shared.Storage; using Microsoft.Extensions.Logging; using StackExchange.Redis; @@ -36,13 +37,13 @@ public async ValueTask> AcquireDueJobsAsync( { cancellationToken.ThrowIfCancellationRequested(); var metadata = await Database.HashGetAsync(DefinitionMetadataKey, definitionKeys).WaitAsync(cancellationToken); - var definitions = new Dictionary(StringComparer.Ordinal); + var definitions = new Dictionary(StringComparer.Ordinal); for (var index = 0; index < names.Count; index++) { if (metadata[index].IsNull) continue; - var definition = JsonSerializer.Deserialize(metadata[index].ToString(), RedisJsonSerializerContext.Default.JobDefinitionRecord) + var definition = JsonSerializer.Deserialize(metadata[index].ToString(), RedisJsonSerializerContext.Default.JobDefinition) ?? throw new ImmediateJobException($"Job definition '{names[index]}' contains invalid data."); definitions.Add(names[index], (metadata[index], definition)); } diff --git a/src/Immediate.Jobs.Redis/RedisJobStorage.DefinitionCatalog.cs b/src/Immediate.Jobs.Redis/RedisJobStorage.DefinitionCatalog.cs index 1639cc6e..41ad6f7c 100644 --- a/src/Immediate.Jobs.Redis/RedisJobStorage.DefinitionCatalog.cs +++ b/src/Immediate.Jobs.Redis/RedisJobStorage.DefinitionCatalog.cs @@ -1,5 +1,5 @@ using System.Text.Json; -using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; using Immediate.Jobs.Shared.Storage; using Microsoft.Extensions.Logging; using StackExchange.Redis; @@ -27,7 +27,7 @@ public async ValueTask MergeJobDefinitionsListAsync(JobDefinitionRegistration re Definitions = changes.Definitions.Select(static definition => new RedisDefinitionMetadata { Name = definition.Name.ToUpperInvariant(), - Metadata = JsonSerializer.Serialize(definition, RedisJsonSerializerContext.Default.JobDefinitionRecord), + Metadata = JsonSerializer.Serialize(definition, RedisJsonSerializerContext.Default.JobDefinition), }).ToList(), RemovedDefinitions = changes.RemovedDefinitionNames.Select(static name => name.ToUpperInvariant()).ToList(), RemovedSchedules = changes.RemovedScheduleNames.Select(NormalizeRecurringName).ToList(), @@ -65,13 +65,13 @@ public async ValueTask MergeJobDefinitionsListAsync(JobDefinitionRegistration re } /// - public async ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default) + public async ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default) { GetJobDefinitionsAsyncCalled(); cancellationToken.ThrowIfCancellationRequested(); await TaskScheduler.Yield(); var metadata = await Database.HashGetAllAsync(DefinitionMetadataKey).WaitAsync(cancellationToken); - return [.. metadata.Select(static entry => JsonSerializer.Deserialize((string)entry.Value!, RedisJsonSerializerContext.Default.JobDefinitionRecord)!) + return [.. metadata.Select(static entry => JsonSerializer.Deserialize((string)entry.Value!, RedisJsonSerializerContext.Default.JobDefinition)!) .OrderBy(static definition => definition.Name, StringComparer.OrdinalIgnoreCase)]; } diff --git a/src/Immediate.Jobs.Redis/RedisJobStorage.Definitions.cs b/src/Immediate.Jobs.Redis/RedisJobStorage.Definitions.cs index c0a280be..d9f3d5e7 100644 --- a/src/Immediate.Jobs.Redis/RedisJobStorage.Definitions.cs +++ b/src/Immediate.Jobs.Redis/RedisJobStorage.Definitions.cs @@ -48,7 +48,7 @@ private async ValueTask EvaluateDefinitionAsync(string jobN var metadata = await Database.HashGetAsync(DefinitionMetadataKey, name).WaitAsync(cancellationToken); if (metadata.IsNull) throw new KeyNotFoundException($"Job definition '{jobName}' was not found."); - var definition = JsonSerializer.Deserialize(metadata.ToString(), RedisJsonSerializerContext.Default.JobDefinitionRecord) + var definition = JsonSerializer.Deserialize(metadata.ToString(), RedisJsonSerializerContext.Default.JobDefinition) ?? throw new ImmediateJobException($"Job definition '{jobName}' contains invalid data."); var limits = definition.AcquisitionLimits; var now = _timeProvider.GetUtcNow(); diff --git a/src/Immediate.Jobs.Redis/RedisJobStorage.Records.cs b/src/Immediate.Jobs.Redis/RedisJobStorage.Records.cs index ed780848..262b0469 100644 --- a/src/Immediate.Jobs.Redis/RedisJobStorage.Records.cs +++ b/src/Immediate.Jobs.Redis/RedisJobStorage.Records.cs @@ -77,7 +77,7 @@ CancellationToken cancellationToken var definition = JsonSerializer.Deserialize( (string)metadata[index]!, - RedisJsonSerializerContext.Default.JobDefinitionRecord + RedisJsonSerializerContext.Default.JobDefinition ) ?? throw new ImmediateJobException($"Job definition '{names[index]}' contains invalid data."); queues[names[index]] = definition.QueueName; } diff --git a/src/Immediate.Jobs.Redis/RedisJsonSerializerContext.cs b/src/Immediate.Jobs.Redis/RedisJsonSerializerContext.cs index cdd17138..91551af3 100644 --- a/src/Immediate.Jobs.Redis/RedisJsonSerializerContext.cs +++ b/src/Immediate.Jobs.Redis/RedisJsonSerializerContext.cs @@ -1,5 +1,6 @@ using System.Text.Json.Serialization; using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; namespace Immediate.Jobs.Redis; @@ -7,6 +8,6 @@ namespace Immediate.Jobs.Redis; [JsonSerializable(typeof(JobRecord))] [JsonSerializable(typeof(RecurringJobSchedule))] [JsonSerializable(typeof(JobServerSnapshot))] -[JsonSerializable(typeof(JobDefinitionRecord))] +[JsonSerializable(typeof(JobDefinition))] [JsonSerializable(typeof(RedisDefinitionCatalogChanges))] internal sealed partial class RedisJsonSerializerContext : JsonSerializerContext; diff --git a/src/Immediate.Jobs.Shared/Apis/JobDefinitionRecord.cs b/src/Immediate.Jobs.Shared/Apis/JobDefinitionRecord.cs deleted file mode 100644 index 28d03ec1..00000000 --- a/src/Immediate.Jobs.Shared/Apis/JobDefinitionRecord.cs +++ /dev/null @@ -1,112 +0,0 @@ -using Immediate.Jobs.Shared.Storage; - -namespace Immediate.Jobs.Shared.Apis; - -/// -/// Persisted, storage-neutral configuration for a job definition. -/// -public sealed record JobDefinitionRecord -{ - /// - /// The stable job name. - /// - public required string Name { get; init; } - - /// - /// The queue used for new invocations. - /// - public string QueueName { get; init; } = "default"; - - /// - /// The dispatch priority of the queue. - /// - public int QueuePriority { get; init; } - - /// - /// The per-server queue concurrency limit. Zero is unbounded. - /// - public int QueueConcurrency { get; init; } - - /// - /// The code-defined recurring expression, or null. - /// - public string? Cron { get; init; } - - /// - /// The recurring time-zone identifier. - /// - public string TimeZone { get; init; } = "UTC"; - - /// - /// The total permitted execution attempts. - /// - public int MaxAttempts { get; init; } = 3; - - /// - /// The execution timeout, or null. - /// - public TimeSpan? Timeout { get; init; } - - /// - /// The shared definition concurrency limit. Zero is unbounded. - /// - public int MaxConcurrency { get; init; } - - /// - /// Maximum acquisitions in the preceding sliding period. Zero disables this limit. - /// - public int SlidingWindowMax { get; init; } - - /// - /// The sliding acquisition period, or null when disabled. - /// - public TimeSpan? SlidingWindowPeriod { get; init; } - - /// - /// Maximum acquisitions in each aligned fixed period. Zero disables this limit. - /// - public int FixedWindowMax { get; init; } - - /// - /// The fixed acquisition period, or null when disabled. - /// - public TimeSpan? FixedWindowPeriod { get; init; } - - /// - /// The parsed limits passed to storage for direct acquisition queries. - /// - [System.Text.Json.Serialization.JsonIgnore] - public JobAcquisitionLimits AcquisitionLimits => new() - { - MaxConcurrency = MaxConcurrency, - SlidingWindowMax = SlidingWindowMax, - SlidingWindowPeriod = SlidingWindowPeriod, - FixedWindowMax = FixedWindowMax, - FixedWindowPeriod = FixedWindowPeriod, - }; - - /// - /// The recurring overlap policy. - /// - public OverlapPolicy OverlapPolicy { get; init; } = OverlapPolicy.Skip; - - /// - /// The recurring misfire policy. - /// - public MisfireHandlingMode MisfireHandlingMode { get; init; } = MisfireHandlingMode.EnqueueOne; - - /// - /// The retry delay algorithm. - /// - public BackoffStrategy Backoff { get; init; } = BackoffStrategy.ExponentialJitter; - - /// - /// The effective job tags. Omitted or empty lists use the default tag. - /// - public IReadOnlyList Tags { get; init => field = JobTags.Normalize(value); } = [JobTags.Default]; - - /// - /// The retry base delay. - /// - public TimeSpan BackoffBase { get; init; } = TimeSpan.FromSeconds(5); -} diff --git a/src/Immediate.Jobs.Shared/Apis/JobMonitoringDefinitions.cs b/src/Immediate.Jobs.Shared/Apis/JobMonitoringDefinitions.cs index a4b67a31..c1067eff 100644 --- a/src/Immediate.Jobs.Shared/Apis/JobMonitoringDefinitions.cs +++ b/src/Immediate.Jobs.Shared/Apis/JobMonitoringDefinitions.cs @@ -1,3 +1,5 @@ +using Immediate.Jobs.Shared.Internals; + namespace Immediate.Jobs.Shared.Apis; /// @@ -8,7 +10,7 @@ public sealed record JobMonitoringDefinitions /// /// All persisted job definitions, irrespective of server tags or local registrations. /// - public required IReadOnlyList Jobs { get; init; } + public required IReadOnlyList Jobs { get; init; } /// /// All code-defined and dynamic recurring schedule definitions. diff --git a/src/Immediate.Jobs.Shared/Internals/JobDefinition.cs b/src/Immediate.Jobs.Shared/Internals/JobDefinition.cs index ce56f02d..ca4dacd9 100644 --- a/src/Immediate.Jobs.Shared/Internals/JobDefinition.cs +++ b/src/Immediate.Jobs.Shared/Internals/JobDefinition.cs @@ -1,12 +1,13 @@ using System.ComponentModel; using System.Diagnostics.CodeAnalysis; +using System.Text.Json.Serialization; using Immediate.Jobs.Shared.Interfaces; using Immediate.Jobs.Shared.Storage; namespace Immediate.Jobs.Shared.Internals; /// -/// Immutable generated execution settings. +/// Immutable job configuration shared by execution, storage, and monitoring. /// [SuppressMessage("Design", "MA0053", Justification = "Inherited by each job to ensure idempotency in registrations")] [EditorBrowsable(EditorBrowsableState.Never)] @@ -21,7 +22,24 @@ public record JobDefinition /// /// The queue used by newly-created invocations. /// - public JobQueueDefinition Queue { get; init; } = JobQueueDefinition.Default; + [AllowNull] + [JsonIgnore] + public JobQueueDefinition Queue { get; init => field = value ?? JobQueueDefinition.Default; } = JobQueueDefinition.Default; + + /// + /// The queue used for new invocations. + /// + public string QueueName { get => Queue.Name; init => Queue = Queue with { Name = value }; } + + /// + /// The dispatch priority of the queue. + /// + public int QueuePriority { get => Queue.Priority; init => Queue = Queue with { Priority = value }; } + + /// + /// The per-server queue concurrency limit. Zero is unbounded. + /// + public int QueueConcurrency { get => Queue.Concurrency; init => Queue = Queue with { Concurrency = value }; } /// /// The stable job name. @@ -29,14 +47,16 @@ public record JobDefinition public required string Name { get; init; } /// - /// The generated invoker. + /// The generated invoker, available only on locally registered definitions. /// - public required IJobInvoker Invoker { get; init; } + [JsonIgnore] + public IJobInvoker? Invoker { get; init; } /// - /// The job CLR type, used only for logging and diagnostics. + /// The local job CLR type, used only for logging and diagnostics. /// - public required Type JobType { get; init; } + [JsonIgnore] + public Type? JobType { get; init; } /// /// The code-defined cron expression, or for a non-recurring job. @@ -86,6 +106,7 @@ public record JobDefinition /// /// The parsed limits passed to storage for direct acquisition queries. /// + [JsonIgnore] public JobAcquisitionLimits AcquisitionLimits => new() { MaxConcurrency = MaxConcurrency, diff --git a/src/Immediate.Jobs.Shared/Internals/JobSchedulingService.cs b/src/Immediate.Jobs.Shared/Internals/JobSchedulingService.cs index c86d67d1..e91f334b 100644 --- a/src/Immediate.Jobs.Shared/Internals/JobSchedulingService.cs +++ b/src/Immediate.Jobs.Shared/Internals/JobSchedulingService.cs @@ -251,29 +251,7 @@ private async Task InitializeAsync(CancellationToken cancellationToken) return; await _storage.InitializeAsync(cancellationToken); - var definitions = _definitions.Values - .Select(static definition => new JobDefinitionRecord - { - Name = definition.Name, - Tags = definition.Tags, - QueueName = definition.Queue.Name, - QueuePriority = definition.Queue.Priority, - QueueConcurrency = definition.Queue.Concurrency, - Cron = definition.Cron, - TimeZone = definition.TimeZone, - MaxAttempts = definition.MaxAttempts, - Timeout = definition.Timeout, - MaxConcurrency = definition.MaxConcurrency, - SlidingWindowMax = definition.SlidingWindowMax, - SlidingWindowPeriod = definition.SlidingWindowPeriod, - FixedWindowMax = definition.FixedWindowMax, - FixedWindowPeriod = definition.FixedWindowPeriod, - OverlapPolicy = definition.OverlapPolicy, - MisfireHandlingMode = definition.MisfireHandlingMode, - Backoff = definition.Backoff, - BackoffBase = definition.BackoffBase, - }) - .ToList(); + var definitions = _definitions.Values.ToList(); var now = _timeProvider.GetUtcNow(); @@ -599,7 +577,8 @@ await _storage.SetExecutionTelemetryAsync( var executionBuffer = new JobExecutionBuffer(); - await definition.Invoker.InvokeAsync( + var invoker = definition.Invoker ?? throw new InvalidOperationException($"Job definition '{definition.Name}' has no runtime invoker."); + await invoker.InvokeAsync( scope.ServiceProvider, new JobExecution { Record = record, Definition = definition, CancellationToken = linkedCts.Token, Buffer = executionBuffer } ); diff --git a/src/Immediate.Jobs.Shared/Storage/IJobStorage.cs b/src/Immediate.Jobs.Shared/Storage/IJobStorage.cs index 7d35e84a..0b6762fa 100644 --- a/src/Immediate.Jobs.Shared/Storage/IJobStorage.cs +++ b/src/Immediate.Jobs.Shared/Storage/IJobStorage.cs @@ -1,4 +1,5 @@ using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; namespace Immediate.Jobs.Shared.Storage; @@ -100,7 +101,7 @@ public interface IJobStorage : IAsyncDisposable /// /// The complete stored definition catalogue. /// - ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default); + ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default); /// /// Inserts a pending or scheduled invocation. diff --git a/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.DefinitionCatalog.cs b/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.DefinitionCatalog.cs index 4c9789c5..b335b29c 100644 --- a/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.DefinitionCatalog.cs +++ b/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.DefinitionCatalog.cs @@ -1,11 +1,12 @@ using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; using Microsoft.Extensions.Logging; namespace Immediate.Jobs.Shared.Storage; public sealed partial class InMemoryJobStorage { - private readonly Dictionary _jobDefinitions = [with(StringComparer.OrdinalIgnoreCase)]; + private readonly Dictionary _jobDefinitions = [with(StringComparer.OrdinalIgnoreCase)]; /// public async ValueTask MergeJobDefinitionsListAsync(JobDefinitionRegistration registration, CancellationToken cancellationToken = default) @@ -66,7 +67,7 @@ public async ValueTask MergeJobDefinitionsListAsync(JobDefinitionRegistration re } /// - public async ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default) + public async ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default) { InMemoryGetJobDefinitionsAsyncCalled(); cancellationToken.ThrowIfCancellationRequested(); diff --git a/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Definitions.cs b/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Definitions.cs index b8c60224..05392254 100644 --- a/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Definitions.cs +++ b/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Definitions.cs @@ -1,4 +1,5 @@ using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; using Microsoft.Extensions.Logging; namespace Immediate.Jobs.Shared.Storage; @@ -67,7 +68,7 @@ public async ValueTask GetJobAcquisitionStateAsync(string j } } - private JobDefinitionRecord GetStoredDefinition(string jobName) => + private JobDefinition GetStoredDefinition(string jobName) => _jobDefinitions.TryGetValue(jobName, out var definition) ? definition : throw new KeyNotFoundException($"Job definition '{jobName}' was not found."); diff --git a/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Enqueue.cs b/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Enqueue.cs index 243bd02b..0fa034ee 100644 --- a/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Enqueue.cs +++ b/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Enqueue.cs @@ -1,5 +1,6 @@ using System.Globalization; using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; using Microsoft.Extensions.Logging; namespace Immediate.Jobs.Shared.Storage; @@ -183,7 +184,7 @@ private JobRecord CreateRecord(JobSubmission job) return job.ToJobRecord(definition.Name); } - internal JobDefinitionRecord GetJobDefinition(string name) + internal JobDefinition GetJobDefinition(string name) { lock (_gate) return _jobDefinitions.TryGetValue(name, out var definition) diff --git a/src/Immediate.Jobs.Shared/Storage/JobDefinitionReconciliation.cs b/src/Immediate.Jobs.Shared/Storage/JobDefinitionReconciliation.cs index 3ffba83a..bb9be2c8 100644 --- a/src/Immediate.Jobs.Shared/Storage/JobDefinitionReconciliation.cs +++ b/src/Immediate.Jobs.Shared/Storage/JobDefinitionReconciliation.cs @@ -1,5 +1,6 @@ using System.ComponentModel; using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; namespace Immediate.Jobs.Shared.Storage; @@ -17,7 +18,7 @@ public sealed record JobDefinitionReconciliation /// /// The definitions to insert or replace. /// - public required IReadOnlyList Definitions { get; init; } + public required IReadOnlyList Definitions { get; init; } /// /// The names of obsolete definitions in this server's scope. @@ -51,7 +52,7 @@ public sealed record JobDefinitionReconciliation /// public static JobDefinitionReconciliation Create( JobDefinitionRegistration registration, - IReadOnlyList existingDefinitions, + IReadOnlyList existingDefinitions, IReadOnlyList existingSchedules ) { diff --git a/src/Immediate.Jobs.Shared/Storage/JobDefinitionRegistration.cs b/src/Immediate.Jobs.Shared/Storage/JobDefinitionRegistration.cs index 1e07f90e..0779d23b 100644 --- a/src/Immediate.Jobs.Shared/Storage/JobDefinitionRegistration.cs +++ b/src/Immediate.Jobs.Shared/Storage/JobDefinitionRegistration.cs @@ -1,4 +1,5 @@ using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; namespace Immediate.Jobs.Shared.Storage; @@ -16,7 +17,7 @@ public sealed record JobDefinitionRegistration /// /// All locally registered definitions, including those outside the server's scope. /// - public required IReadOnlyList Definitions { get; init; } + public required IReadOnlyList Definitions { get; init; } /// /// The complete list of code-defined recurring schedules for these definitions. diff --git a/src/Immediate.Jobs.Shared/Storage/SingleServerJobStorage.DefinitionCatalog.cs b/src/Immediate.Jobs.Shared/Storage/SingleServerJobStorage.DefinitionCatalog.cs index bd04aec7..295d0218 100644 --- a/src/Immediate.Jobs.Shared/Storage/SingleServerJobStorage.DefinitionCatalog.cs +++ b/src/Immediate.Jobs.Shared/Storage/SingleServerJobStorage.DefinitionCatalog.cs @@ -1,4 +1,5 @@ using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; using Microsoft.Extensions.Logging; namespace Immediate.Jobs.Shared.Storage; @@ -49,7 +50,7 @@ await PrimaryStorage.MergeJobDefinitionsListAsync( } /// - public async ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default) + public async ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default) { SingleServerGetJobDefinitionsAsyncCalled(); cancellationToken.ThrowIfCancellationRequested(); diff --git a/src/Immediate.Jobs.Testing/CapturingJobStorage.cs b/src/Immediate.Jobs.Testing/CapturingJobStorage.cs index 93230357..f776691e 100644 --- a/src/Immediate.Jobs.Testing/CapturingJobStorage.cs +++ b/src/Immediate.Jobs.Testing/CapturingJobStorage.cs @@ -1,5 +1,6 @@ using System.Diagnostics.CodeAnalysis; using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; using Immediate.Jobs.Shared.Storage; namespace Immediate.Jobs.Testing; @@ -161,7 +162,7 @@ public virtual async ValueTask MergeJobDefinitionsListAsync(JobDefinitionRegistr } /// - public virtual ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default) => + public virtual ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default) => _inner.GetJobDefinitionsAsync(cancellationToken); /// diff --git a/src/Immediate.Jobs.Testing/JobTestHarness.cs b/src/Immediate.Jobs.Testing/JobTestHarness.cs index 0cd70a7f..9a72d448 100644 --- a/src/Immediate.Jobs.Testing/JobTestHarness.cs +++ b/src/Immediate.Jobs.Testing/JobTestHarness.cs @@ -510,7 +510,8 @@ public async ValueTask RunThroughPipelineAsync( await using var scope = Services.CreateAsyncScope(); - await definition.Invoker.InvokeAsync( + var invoker = definition.Invoker ?? throw new InvalidOperationException($"Job definition '{definition.Name}' has no runtime invoker."); + await invoker.InvokeAsync( scope.ServiceProvider, new JobExecution { Record = record, Definition = definition, CancellationToken = cancellationToken } ); diff --git a/src/Immediate.Jobs.Testing/Storage/DefinitionAcquisitionStorageConformance.cs b/src/Immediate.Jobs.Testing/Storage/DefinitionAcquisitionStorageConformance.cs index b88dfe93..092946fa 100644 --- a/src/Immediate.Jobs.Testing/Storage/DefinitionAcquisitionStorageConformance.cs +++ b/src/Immediate.Jobs.Testing/Storage/DefinitionAcquisitionStorageConformance.cs @@ -1,5 +1,6 @@ using System.Globalization; using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; using Immediate.Jobs.Shared.Storage; using Microsoft.Extensions.Time.Testing; @@ -71,7 +72,7 @@ private static async ValueTask CaseInsensitiveControlsAsync(IJobStorage storage, { const string CanonicalName = "Émail.Σ"; const string Alias = "ÉMAIL.Σ"; - var definition = new JobDefinitionRecord + var definition = new JobDefinition { Name = CanonicalName, MaxConcurrency = 1, @@ -96,7 +97,7 @@ private static async ValueTask CaseInsensitiveControlsAsync(IJobStorage storage, private static async ValueTask MonitoringAsync(IJobStorage storage, FakeTimeProvider clock, CancellationToken token) { - var definition = new JobDefinitionRecord { Name = Name }; + var definition = new JobDefinition { Name = Name }; await storage.MergeJobDefinitionsListAsync(new() { Definitions = [definition, new() { Name = "other-job" }], @@ -141,7 +142,7 @@ private static async ValueTask WindowCountsAsync(IJobStorage storage, FakeTimePr var now = clock.GetUtcNow(); var start = new DateTimeOffset(now.UtcTicks - now.UtcTicks % period.Ticks, TimeSpan.Zero) + period; clock.Advance(start.AddSeconds(10) - now); - var definition = new JobDefinitionRecord + var definition = new JobDefinition { Name = Name, MaxConcurrency = 2, @@ -215,7 +216,7 @@ await storage.MergeJobDefinitionsListAsync(new() private static async ValueTask SlidingAsync(IJobStorage storage, FakeTimeProvider clock, CancellationToken token) { var period = TimeSpan.FromSeconds(1) + TimeSpan.FromMilliseconds(3); - var definition = new JobDefinitionRecord + var definition = new JobDefinition { Name = Name, SlidingWindowMax = 1, @@ -242,7 +243,7 @@ await storage.MergeJobDefinitionsListAsync(new() private static async ValueTask RollingAsync(IJobStorage storage, FakeTimeProvider clock, CancellationToken token) { - var definition = new JobDefinitionRecord + var definition = new JobDefinition { Name = Name, SlidingWindowMax = 2, @@ -273,7 +274,7 @@ private static async ValueTask FixedAsync(IJobStorage storage, FakeTimeProvider var now = clock.GetUtcNow(); var boundary = new DateTimeOffset(now.UtcTicks - now.UtcTicks % period.Ticks, TimeSpan.Zero) + period; clock.Advance(boundary - now - TimeSpan.FromMilliseconds(1)); - var definition = new JobDefinitionRecord + var definition = new JobDefinition { Name = Name, FixedWindowMax = 1, @@ -293,7 +294,7 @@ await storage.MergeJobDefinitionsListAsync(new() private static async ValueTask CombinedAsync(IJobStorage storage, FakeTimeProvider clock, CancellationToken token) { - var definition = new JobDefinitionRecord + var definition = new JobDefinition { Name = Name, SlidingWindowMax = 1, @@ -319,7 +320,7 @@ await storage.MergeJobDefinitionsListAsync(new() private static async ValueTask ConcurrencyAsync(IJobStorage storage, FakeTimeProvider clock, CancellationToken token) { - var definition = new JobDefinitionRecord + var definition = new JobDefinition { Name = Name, MaxConcurrency = 1, @@ -342,7 +343,7 @@ await storage.MergeJobDefinitionsListAsync(new() private static async ValueTask RenewAsync(IJobStorage storage, FakeTimeProvider clock, CancellationToken token) { - var definition = new JobDefinitionRecord + var definition = new JobDefinition { Name = Name, MaxConcurrency = 1, @@ -363,7 +364,7 @@ await storage.MergeJobDefinitionsListAsync(new() private static async ValueTask ContentionAsync(IJobStorage storage, FakeTimeProvider clock, CancellationToken token) { - var definition = new JobDefinitionRecord + var definition = new JobDefinition { Name = Name, MaxConcurrency = 3, @@ -382,7 +383,7 @@ await storage.MergeJobDefinitionsListAsync(new() private static async ValueTask FairAsync(IJobStorage storage, FakeTimeProvider clock, CancellationToken token) { - var definition = new JobDefinitionRecord + var definition = new JobDefinition { Name = Name, SlidingWindowMax = 1, @@ -402,7 +403,7 @@ await storage.MergeJobDefinitionsListAsync(new() private static async ValueTask RetryAsync(IJobStorage storage, FakeTimeProvider clock, CancellationToken token) { - var definition = new JobDefinitionRecord + var definition = new JobDefinition { Name = Name, SlidingWindowMax = 1, diff --git a/src/Immediate.Jobs.Testing/Storage/DefinitionCatalogStorageConformance.cs b/src/Immediate.Jobs.Testing/Storage/DefinitionCatalogStorageConformance.cs index 794a5ca4..dac951b0 100644 --- a/src/Immediate.Jobs.Testing/Storage/DefinitionCatalogStorageConformance.cs +++ b/src/Immediate.Jobs.Testing/Storage/DefinitionCatalogStorageConformance.cs @@ -1,4 +1,5 @@ using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; using Immediate.Jobs.Shared.Storage; using Microsoft.Extensions.Time.Testing; @@ -91,9 +92,9 @@ await ConformanceAssert.ThrowsAsync( ConformanceAssert.Equal(1, (await storage.GetMonitoringSnapshotAsync(token)).Recurring.Count, SubmissionCase, "definition removal retains the existing schedule"); } - private static JobDefinitionRecord Definition(string name) => new() { Name = name }; + private static JobDefinition Definition(string name) => new() { Name = name }; - private static JobDefinitionRegistration Registration(IReadOnlyList definitions) => + private static JobDefinitionRegistration Registration(IReadOnlyList definitions) => new() { Definitions = definitions }; private static async ValueTask RoundTripsConfigurationAsync(IJobStorage storage, FakeTimeProvider timeProvider, CancellationToken token) @@ -133,7 +134,7 @@ await storage.MergeJobDefinitionsListAsync(new() ConformanceAssert.Equal(Values(Definition("Configured")), Values(stored), ConfigurationCase, "an alias update must replace every field, clear nullable values, and preserve the stored name"); - static object Values(JobDefinitionRecord value) => + static object Values(JobDefinition value) => (value.Name, value.QueueName, value.QueuePriority, value.QueueConcurrency, value.Cron, value.TimeZone, value.MaxAttempts, value.Timeout, value.MaxConcurrency, value.OverlapPolicy, value.MisfireHandlingMode, value.Backoff, value.BackoffBase, value.SlidingWindowMax, value.SlidingWindowPeriod, @@ -162,7 +163,7 @@ await ConformanceAssert.ThrowsAsync(() => storage.MergeJobDef ConformanceAssert.Equal(0, (await storage.GetJobDefinitionsAsync(token)).Count, CaseNameCase, "an empty catalogue must remove aliases as one identity"); } - private static RecurringJobSchedule Schedule(JobDefinitionRecord definition, DateTimeOffset next) => new() + private static RecurringJobSchedule Schedule(JobDefinition definition, DateTimeOffset next) => new() { Name = definition.Name, JobName = definition.Name, diff --git a/src/Immediate.Jobs.Testing/Storage/DefinitionRemovalStorageConformance.cs b/src/Immediate.Jobs.Testing/Storage/DefinitionRemovalStorageConformance.cs index 0213ebe2..4a2fff0e 100644 --- a/src/Immediate.Jobs.Testing/Storage/DefinitionRemovalStorageConformance.cs +++ b/src/Immediate.Jobs.Testing/Storage/DefinitionRemovalStorageConformance.cs @@ -1,4 +1,5 @@ using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; using Immediate.Jobs.Shared.Storage; using Microsoft.Extensions.Time.Testing; @@ -40,8 +41,8 @@ internal static class DefinitionRemovalStorageConformance private static async ValueTask CancelsRemovedAsync(IJobStorage storage, FakeTimeProvider clock, CancellationToken token) { var now = clock.GetUtcNow(); - var definition = new JobDefinitionRecord { Name = "Résumé", Tags = ["alpha"] }; - var other = new JobDefinitionRecord { Name = "Other", Tags = ["beta"] }; + var definition = new JobDefinition { Name = "Résumé", Tags = ["alpha"] }; + var other = new JobDefinition { Name = "Other", Tags = ["beta"] }; await storage.MergeJobDefinitionsListAsync(new() { Definitions = [definition, other], ServerTags = ["alpha", "beta"] }, token); await storage.EnqueueAsync(Job("history", definition.Name, now), token); var history = (await storage.AcquireDueJobsAsync(Request(definition.Name), token)).Single(); diff --git a/src/Immediate.Jobs.Testing/Storage/MonitoringStorageConformance.cs b/src/Immediate.Jobs.Testing/Storage/MonitoringStorageConformance.cs index ffc8e29c..1df714a4 100644 --- a/src/Immediate.Jobs.Testing/Storage/MonitoringStorageConformance.cs +++ b/src/Immediate.Jobs.Testing/Storage/MonitoringStorageConformance.cs @@ -1,4 +1,5 @@ using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; using Immediate.Jobs.Shared.Storage; using Microsoft.Extensions.Time.Testing; @@ -18,7 +19,7 @@ internal static class MonitoringStorageConformance private static async ValueTask BulkAsync(IJobStorage storage, FakeTimeProvider clock, CancellationToken token) { var now = clock.GetUtcNow(); - JobDefinitionRecord[] definitions = + JobDefinition[] definitions = [ new() { diff --git a/src/Immediate.Jobs.Testing/Storage/RecurringStorageConformance.cs b/src/Immediate.Jobs.Testing/Storage/RecurringStorageConformance.cs index 27f97faa..ac4b115a 100644 --- a/src/Immediate.Jobs.Testing/Storage/RecurringStorageConformance.cs +++ b/src/Immediate.Jobs.Testing/Storage/RecurringStorageConformance.cs @@ -1,5 +1,6 @@ using System.Globalization; using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; using Immediate.Jobs.Shared.Storage; using Microsoft.Extensions.Time.Testing; @@ -203,7 +204,7 @@ CancellationToken cancellationToken await storage.MergeJobDefinitionsListAsync( new() { - Definitions = supplied.Select(static schedule => new JobDefinitionRecord + Definitions = supplied.Select(static schedule => new JobDefinition { Name = schedule.JobName, QueueName = schedule.QueueName ?? "default", diff --git a/src/Immediate.Jobs.Testing/Storage/ReplicaStorageConformance.Enqueue.cs b/src/Immediate.Jobs.Testing/Storage/ReplicaStorageConformance.Enqueue.cs index a349a935..b12bc1ca 100644 --- a/src/Immediate.Jobs.Testing/Storage/ReplicaStorageConformance.Enqueue.cs +++ b/src/Immediate.Jobs.Testing/Storage/ReplicaStorageConformance.Enqueue.cs @@ -1,4 +1,5 @@ using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; using Immediate.Jobs.Shared.Storage; using Immediate.Jobs.Testing.Storage; using Microsoft.Extensions.DependencyInjection; @@ -136,7 +137,7 @@ public ValueTask MergeJobDefinitionsListAsync(JobDefinitionRegistration registra storage.MergeJobDefinitionsListAsync(registration, cancellationToken); /// - public ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default) => + public ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default) => storage.GetJobDefinitionsAsync(cancellationToken); /// diff --git a/src/Immediate.Jobs.Testing/Storage/ReplicaStorageConformance.cs b/src/Immediate.Jobs.Testing/Storage/ReplicaStorageConformance.cs index e4ef2976..13c8ca12 100644 --- a/src/Immediate.Jobs.Testing/Storage/ReplicaStorageConformance.cs +++ b/src/Immediate.Jobs.Testing/Storage/ReplicaStorageConformance.cs @@ -1,5 +1,6 @@ using System.Globalization; using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; using Immediate.Jobs.Shared.Storage; using Immediate.Jobs.Testing.Storage; using Microsoft.Extensions.Time.Testing; @@ -39,7 +40,7 @@ internal static partial class ReplicaStorageConformance private static async ValueTask RestoresCronDefinitionsAsync(IJobStorage storage, FakeTimeProvider clock, CancellationToken token) { var now = clock.GetUtcNow(); - var definition = new JobDefinitionRecord { Name = JobName, QueueName = QueueName, Cron = "* * * * *" }; + var definition = new JobDefinition { Name = JobName, QueueName = QueueName, Cron = "* * * * *" }; var schedule = new RecurringJobSchedule { Name = JobName, diff --git a/src/Immediate.Jobs.Testing/Storage/TagStorageConformance.cs b/src/Immediate.Jobs.Testing/Storage/TagStorageConformance.cs index 61aa332d..2dd28720 100644 --- a/src/Immediate.Jobs.Testing/Storage/TagStorageConformance.cs +++ b/src/Immediate.Jobs.Testing/Storage/TagStorageConformance.cs @@ -1,4 +1,5 @@ using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; using Immediate.Jobs.Shared.Storage; using Microsoft.Extensions.Time.Testing; @@ -23,9 +24,9 @@ internal static class TagStorageConformance new(CaseNameCase, StorageCapabilities.Queue, ReconcilesCaseInsensitiveNamesAsync), ]; - private static JobDefinitionRecord Definition(string name, params string[] tags) => new() { Name = name, Tags = tags }; + private static JobDefinition Definition(string name, params string[] tags) => new() { Name = name, Tags = tags }; - private static JobDefinitionRegistration Registration(IReadOnlyList definitions, params string[] tags) => + private static JobDefinitionRegistration Registration(IReadOnlyList definitions, params string[] tags) => new() { Definitions = definitions, ServerTags = tags }; private static async ValueTask ReconcilesCaseInsensitiveNamesAsync(IJobStorage storage, FakeTimeProvider timeProvider, CancellationToken token) @@ -111,7 +112,7 @@ private static async ValueTask RetagsAsync(IJobStorage storage, FakeTimeProvider ConformanceAssert.Equal(0, (await storage.GetJobDefinitionsAsync(token)).Count, RetagCase, "Unicode tag case variants must manage the same scope"); } - private static RecurringJobSchedule Schedule(JobDefinitionRecord definition, DateTimeOffset next) => new() + private static RecurringJobSchedule Schedule(JobDefinition definition, DateTimeOffset next) => new() { Name = definition.Name, JobName = definition.Name, diff --git a/tests/Immediate.Jobs.FunctionalTests/DefinitionAcquisitionTests.cs b/tests/Immediate.Jobs.FunctionalTests/DefinitionAcquisitionTests.cs index 1bed7b06..c95c78ba 100644 --- a/tests/Immediate.Jobs.FunctionalTests/DefinitionAcquisitionTests.cs +++ b/tests/Immediate.Jobs.FunctionalTests/DefinitionAcquisitionTests.cs @@ -238,7 +238,7 @@ private sealed class DefinitionCatalogTrackingStorage(TimeProvider timeProvider) { public int CatalogReads { get; private set; } - public override ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default) + public override ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default) { CatalogReads++; return base.GetJobDefinitionsAsync(cancellationToken); diff --git a/tests/Immediate.Jobs.FunctionalTests/Packages/DashboardPackageTests.cs b/tests/Immediate.Jobs.FunctionalTests/Packages/DashboardPackageTests.cs index 67cd1241..ca3dfe52 100644 --- a/tests/Immediate.Jobs.FunctionalTests/Packages/DashboardPackageTests.cs +++ b/tests/Immediate.Jobs.FunctionalTests/Packages/DashboardPackageTests.cs @@ -148,6 +148,11 @@ await storage.AcquireDueJobsAsync(new() var definition = Assert.Single(document.RootElement.GetProperty("jobs").EnumerateArray()); Assert.Equal("limited-acquisition", definition.GetProperty("name").GetString()); Assert.Equal(1, definition.GetProperty("maxConcurrency").GetInt32()); + Assert.Equal("default", definition.GetProperty("queueName").GetString()); + Assert.False(definition.TryGetProperty("queue", out _)); + Assert.False(definition.TryGetProperty("invoker", out _)); + Assert.False(definition.TryGetProperty("jobType", out _)); + Assert.False(definition.TryGetProperty("acquisitionLimits", out _)); Assert.False(definition.TryGetProperty("acquisition", out _)); } else diff --git a/tests/Immediate.Jobs.FunctionalTests/StorageCapabilityTests.cs b/tests/Immediate.Jobs.FunctionalTests/StorageCapabilityTests.cs index 6a5ceafb..62741d19 100644 --- a/tests/Immediate.Jobs.FunctionalTests/StorageCapabilityTests.cs +++ b/tests/Immediate.Jobs.FunctionalTests/StorageCapabilityTests.cs @@ -1,6 +1,7 @@ using System.Globalization; using Immediate.Jobs.Shared.Apis; using Immediate.Jobs.Shared.Interfaces; +using Immediate.Jobs.Shared.Internals; using Immediate.Jobs.Shared.Storage; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Time.Testing; @@ -227,7 +228,7 @@ public ValueTask ResumeRecurringAsync(string name, CancellationToken cancellatio public ValueTask MergeJobDefinitionsListAsync(JobDefinitionRegistration registration, CancellationToken cancellationToken = default) => _inner.MergeJobDefinitionsListAsync(registration, cancellationToken); - public ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default) => + public ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default) => _inner.GetJobDefinitionsAsync(cancellationToken); public ValueTask> GetDueRecurringAsync(DateTimeOffset now, int batchSize, IReadOnlyList jobNames, CancellationToken cancellationToken = default) => diff --git a/tests/Immediate.Jobs.StorageTests/JobDefinitionSerializationTests.cs b/tests/Immediate.Jobs.StorageTests/JobDefinitionSerializationTests.cs new file mode 100644 index 00000000..3111461e --- /dev/null +++ b/tests/Immediate.Jobs.StorageTests/JobDefinitionSerializationTests.cs @@ -0,0 +1,90 @@ +using System.Text.Json; +using Immediate.Jobs.Redis; +using Immediate.Jobs.Shared.Interfaces; +using Immediate.Jobs.Shared.Internals; + +namespace Immediate.Jobs.StorageTests; + +public sealed class JobDefinitionSerializationTests +{ + [Fact] + public void LocalDefinitionsRoundTripWithoutRuntimeBindings() + { + var definition = new LocalDefinition + { + Name = "configured", + Tags = ["email"], + Queue = new() { Name = "priority", Priority = 9, Concurrency = 2 }, + Invoker = new NoOpInvoker(), + JobType = typeof(JobDefinitionSerializationTests), + MaxConcurrency = 3, + SlidingWindowMax = 5, + SlidingWindowPeriod = TimeSpan.FromSeconds(17), + }; + var metadata = RedisJsonSerializerContext.Default.JobDefinition; + var json = JsonSerializer.Serialize(definition, metadata); + using var document = JsonDocument.Parse(json); + var root = document.RootElement; + Assert.Equal("priority", root.GetProperty("QueueName").GetString()); + Assert.Equal(9, root.GetProperty("QueuePriority").GetInt32()); + Assert.Equal(2, root.GetProperty("QueueConcurrency").GetInt32()); + Assert.False(root.TryGetProperty("Queue", out _)); + Assert.False(root.TryGetProperty("Invoker", out _)); + Assert.False(root.TryGetProperty("JobType", out _)); + Assert.False(root.TryGetProperty("AcquisitionLimits", out _)); + + var stored = Assert.IsType(JsonSerializer.Deserialize(json, metadata)); + Assert.Equal(definition.Queue, stored.Queue); + Assert.Equal(definition.AcquisitionLimits, stored.AcquisitionLimits); + Assert.Equal(definition.Tags, stored.Tags); + Assert.Null(stored.Invoker); + Assert.Null(stored.JobType); + Assert.Equal(json, JsonSerializer.Serialize(stored, metadata)); + } + + [Fact] + public void ExistingFlatMetadataRestoresQueueSettings() + { + const string Json = """ + { + "Name": "existing", + "QueueConcurrency": 2, + "QueuePriority": 9, + "QueueName": "priority", + "Tags": [], + "Cron": null, + "TimeZone": "UTC", + "MaxAttempts": 3, + "Timeout": null, + "MaxConcurrency": 0, + "SlidingWindowMax": 0, + "SlidingWindowPeriod": null, + "FixedWindowMax": 0, + "FixedWindowPeriod": null, + "OverlapPolicy": 0, + "MisfireHandlingMode": 1, + "Backoff": 2, + "BackoffBase": "00:00:05" + } + """; + var definition = Assert.IsType(JsonSerializer.Deserialize(Json, RedisJsonSerializerContext.Default.JobDefinition)); + Assert.Equal(new() { Name = "priority", Priority = 9, Concurrency = 2 }, definition.Queue); + Assert.Equal([JobTags.Default], definition.Tags); + Assert.Equal(3, definition.MaxAttempts); + Assert.Equal("UTC", definition.TimeZone); + Assert.Null(definition.Invoker); + Assert.Null(definition.JobType); + + var updated = definition with { QueueName = "changed", QueueConcurrency = 4 }; + Assert.Equal(new() { Name = "changed", Priority = 9, Concurrency = 4 }, updated.Queue); + Assert.Equal("priority", definition.QueueName); + Assert.Equal(2, definition.QueueConcurrency); + } + + private sealed record LocalDefinition : JobDefinition; + + private sealed class NoOpInvoker : IJobInvoker + { + public ValueTask InvokeAsync(IServiceProvider scopedServices, JobExecution execution) => ValueTask.CompletedTask; + } +} diff --git a/tests/Immediate.Jobs.StorageTests/RedisConformanceTests.cs b/tests/Immediate.Jobs.StorageTests/RedisConformanceTests.cs index fd32bc9b..5c632605 100644 --- a/tests/Immediate.Jobs.StorageTests/RedisConformanceTests.cs +++ b/tests/Immediate.Jobs.StorageTests/RedisConformanceTests.cs @@ -1,5 +1,6 @@ using Immediate.Jobs.Redis; using Immediate.Jobs.Shared.Apis; +using Immediate.Jobs.Shared.Internals; using Immediate.Jobs.Shared.Storage; using Immediate.Jobs.Testing.Storage; using Microsoft.Extensions.DependencyInjection; @@ -107,7 +108,7 @@ private static async ValueTask LoadJobs(IServiceProvider serviceProvider, IReadO await storage.MergeJobDefinitionsListAsync(new() { Definitions = jobs.DistinctBy(static job => job.JobName, StringComparer.OrdinalIgnoreCase) - .Select(static job => new JobDefinitionRecord { Name = job.JobName, QueueName = job.QueueName ?? "default" }).ToList(), + .Select(static job => new JobDefinition { Name = job.JobName, QueueName = job.QueueName ?? "default" }).ToList(), }); foreach (var job in jobs) await storage.EnqueueAsync(new() From 763b18cb4adc415fb2ee3e4df56f59e694db7c51 Mon Sep 17 00:00:00 2001 From: Frank Date: Fri, 16 Oct 2026 20:12:25 +0000 Subject: [PATCH 2/2] Use nested queue settings on job definitions --- docs/architecture.md | 7 +-- .../src/components/DefinitionTable.vue | 4 +- .../src/components/JobDetail.vue | 6 +-- .../DashboardClient/src/contracts.ts | 10 ++-- .../src/views/DefinitionsView.vue | 2 +- .../DashboardClient/tests/components.test.ts | 2 +- .../tests/definition-actions.test.ts | 2 +- .../DashboardClient/tests/fixtures.ts | 6 +-- ...ameworkCoreJobStorage.DefinitionCatalog.cs | 10 ++-- .../LinqToDBJobStorage.DefinitionCatalog.cs | 10 ++-- .../RedisJobStorage.Acquire.cs | 3 +- .../RedisJobStorage.DefinitionCatalog.cs | 2 +- .../RedisJobStorage.Definitions.cs | 3 +- .../RedisJobStorage.Records.cs | 7 +-- .../RedisJsonSerializerContext.cs | 25 ++++++++- src/Immediate.Jobs.Redis/RedisScripts.cs | 11 ++-- .../Internals/JobDefinition.cs | 16 ------ .../InMemoryJobStorage.DefinitionCatalog.cs | 2 +- .../DefinitionCatalogStorageConformance.cs | 24 ++++----- .../DefinitionRemovalStorageConformance.cs | 2 +- .../Storage/GraphStorageConformance.cs | 2 +- .../Storage/MonitoringStorageConformance.cs | 2 +- .../Storage/QueueStorageConformance.cs | 14 ++--- .../Storage/RecurringStorageConformance.cs | 12 ++--- .../ReplicaStorageConformance.Enqueue.cs | 2 +- .../Storage/ReplicaStorageConformance.cs | 14 ++--- .../Storage/TagStorageConformance.cs | 12 ++--- .../DefinitionAcquisitionTests.cs | 10 ++-- .../MonitoringApiTests.cs | 2 +- .../Packages/DashboardPackageTests.cs | 16 +++--- .../PersistedQueueSchedulerTests.cs | 5 +- .../JobDefinitionSerializationTests.cs | 18 ++++--- .../RedisConformanceTests.cs | 53 ++++++++++++++++++- 33 files changed, 186 insertions(+), 130 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index c2bfa808..79f90dfb 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -59,9 +59,10 @@ configuration surface (`ConfigureStorage`, `ConfigureWorkers`, `UseFairQueues`, - **Startup.** Initializes storage and submits the complete definition catalogue and code-defined schedules through `MergeJobDefinitionsListAsync`, even when workers are disabled. `StartAsync` completes this registration before returning. The storage - reconciles only definitions intersecting the server's effective tags. `JobDefinition` serializes queue - settings as `QueueName`, `QueuePriority`, and `QueueConcurrency`. The local invoker, CLR type, queue - object, and computed acquisition limits are excluded from JSON; definitions read from durable storage + reconciles only definitions intersecting the server's effective tags. `JobDefinition` serializes its + queue settings through the nested `Queue` object. SQL storage maps that object to the existing queue + columns, and Redis reads older flat metadata at its storage boundary. The local invoker, CLR type, + and computed acquisition limits are excluded from JSON; definitions read from durable storage have no runtime bindings. Execution uses the locally registered definitions. - **Polling loop.** Each iteration materializes due recurring schedules, builds a `JobAcquisitionRequest` from tag-eligible local definitions, free worker capacity, and queue/job concurrency limits, calls diff --git a/src/Immediate.Jobs.Dashboard/DashboardClient/src/components/DefinitionTable.vue b/src/Immediate.Jobs.Dashboard/DashboardClient/src/components/DefinitionTable.vue index 2be7f1a5..b0729576 100644 --- a/src/Immediate.Jobs.Dashboard/DashboardClient/src/components/DefinitionTable.vue +++ b/src/Immediate.Jobs.Dashboard/DashboardClient/src/components/DefinitionTable.vue @@ -41,7 +41,7 @@ defineEmits<{ - + {{ definition.name }} @@ -61,7 +61,7 @@ defineEmits<{ - {{ definition.queueName }} + {{ definition.queue.name }} {{ tag }}
{{ definition.cron }}{{ definition.timeZone }}
diff --git a/src/Immediate.Jobs.Dashboard/DashboardClient/src/components/JobDetail.vue b/src/Immediate.Jobs.Dashboard/DashboardClient/src/components/JobDetail.vue index 287b3914..d048c802 100644 --- a/src/Immediate.Jobs.Dashboard/DashboardClient/src/components/JobDetail.vue +++ b/src/Immediate.Jobs.Dashboard/DashboardClient/src/components/JobDetail.vue @@ -402,9 +402,9 @@ function retryButtonLabel(job: JobRecord, pending: boolean): string {

Definition