-
+
{{ 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 ac0fb179..d048c802 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;
@@ -402,9 +402,9 @@ function retryButtonLabel(job: JobRecord, pending: boolean): string {
Definition
- - Queue
{{ definition.queueName }}
- - Queue priority
- {{ definition.queuePriority }}
- - Queue concurrency
- {{ definition.queueConcurrency > 0 ? definition.queueConcurrency : 'No limit' }}
+ - Queue
{{ definition.queue.name }}
+ - Queue priority
- {{ definition.queue.priority }}
+ - Queue concurrency
- {{ definition.queue.concurrency > 0 ? definition.queue.concurrency : 'No limit' }}
- Tags
- {{ tag }}
- Schedule
{{ definition.cron }}On demand
- Time zone
- {{ definition.timeZone }}
diff --git a/src/Immediate.Jobs.Dashboard/DashboardClient/src/contracts.ts b/src/Immediate.Jobs.Dashboard/DashboardClient/src/contracts.ts
index 0fea7769..5c7eaf51 100644
--- a/src/Immediate.Jobs.Dashboard/DashboardClient/src/contracts.ts
+++ b/src/Immediate.Jobs.Dashboard/DashboardClient/src/contracts.ts
@@ -68,12 +68,16 @@ export interface RecurringJobSchedule {
lastRunAt: IsoDateTime | null;
}
-export interface JobDefinitionRecord {
+export interface JobQueueDefinition {
+ name: string;
+ priority: number;
+ concurrency: number;
+}
+
+export interface JobDefinition {
name: string;
tags: string[];
- queueName: string;
- queuePriority: number;
- queueConcurrency: number;
+ queue: JobQueueDefinition;
cron: string | null;
timeZone: string;
maxAttempts: number;
@@ -149,7 +153,7 @@ export interface JobAcquisitionState {
isConcurrencyLimited: boolean;
}
-export interface JobDefinitionStatus extends JobDefinitionRecord {
+export interface JobDefinitionStatus extends JobDefinition {
limits: JobAcquisitionLimits;
acquisition?: JobAcquisitionState | undefined;
}
@@ -258,6 +262,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/DashboardClient/src/views/DefinitionsView.vue b/src/Immediate.Jobs.Dashboard/DashboardClient/src/views/DefinitionsView.vue
index 41bd36fa..3be7df93 100644
--- a/src/Immediate.Jobs.Dashboard/DashboardClient/src/views/DefinitionsView.vue
+++ b/src/Immediate.Jobs.Dashboard/DashboardClient/src/views/DefinitionsView.vue
@@ -16,7 +16,7 @@ const search = ref('');
const filteredDefinitions = computed(() => {
const term = search.value.trim().toLowerCase();
return term
- ? definitions.value.filter(definition => [definition.name, definition.queueName, ...definition.tags]
+ ? definitions.value.filter(definition => [definition.name, definition.queue.name, ...definition.tags]
.some(value => value.toLowerCase().includes(term)))
: definitions.value;
});
diff --git a/src/Immediate.Jobs.Dashboard/DashboardClient/tests/components.test.ts b/src/Immediate.Jobs.Dashboard/DashboardClient/tests/components.test.ts
index 6a9a6c3b..b4e63fc9 100644
--- a/src/Immediate.Jobs.Dashboard/DashboardClient/tests/components.test.ts
+++ b/src/Immediate.Jobs.Dashboard/DashboardClient/tests/components.test.ts
@@ -80,7 +80,7 @@ describe('dashboard components', () => {
const wrapper = mount(DefinitionTable, {
props: { definitions: [pausedDefinition, {
...storedDefinition,
- name: 'Unbounded', queueName: 'default',
+ name: 'Unbounded', queue: { ...storedDefinition.queue, name: 'default' },
limits: { slidingWindowMax: 0, slidingWindowPeriod: null, fixedWindowMax: 0, fixedWindowPeriod: null, maxConcurrency: 0 },
acquisition: { jobName: 'Unbounded', isPaused: false, acquisitionStatus: 'Ready', nextEligibleAt: null, activeCount: 3, slidingWindowCount: 0, fixedWindowCount: 0, isConcurrencyLimited: false },
}] },
diff --git a/src/Immediate.Jobs.Dashboard/DashboardClient/tests/definition-actions.test.ts b/src/Immediate.Jobs.Dashboard/DashboardClient/tests/definition-actions.test.ts
index b0a8b160..ec1c0d31 100644
--- a/src/Immediate.Jobs.Dashboard/DashboardClient/tests/definition-actions.test.ts
+++ b/src/Immediate.Jobs.Dashboard/DashboardClient/tests/definition-actions.test.ts
@@ -135,7 +135,7 @@ describe('definition actions', () => {
});
it('filters definitions immediately by name, queue, and tags without refetching', async () => {
- const invoice = { ...storedDefinition, name: 'Invoice', queueName: 'Finance', tags: ['Billing'] };
+ const invoice = { ...storedDefinition, name: 'Invoice', queue: { ...storedDefinition.queue, name: 'Finance' }, tags: ['Billing'] };
getDefinitionsMock.mockResolvedValue({ jobs: [pausedDefinition, invoice], recurring: [] });
getOverviewMock.mockResolvedValue({ ...snapshot(), definitionStatuses: [pausedDefinition.acquisition, { ...invoice.acquisition, jobName: invoice.name }] });
const { wrapper, queryClient } = await mountView();
diff --git a/src/Immediate.Jobs.Dashboard/DashboardClient/tests/fixtures.ts b/src/Immediate.Jobs.Dashboard/DashboardClient/tests/fixtures.ts
index 500a6800..e7027475 100644
--- a/src/Immediate.Jobs.Dashboard/DashboardClient/tests/fixtures.ts
+++ b/src/Immediate.Jobs.Dashboard/DashboardClient/tests/fixtures.ts
@@ -61,9 +61,7 @@ export const workflowGraph: BatchGraph = {
export const storedDefinition: JobDefinitionStatus & { acquisition: JobAcquisitionState } = {
name: 'RemoteEmail',
tags: ['email', 'default'],
- queueName: 'priority',
- queuePriority: 5,
- queueConcurrency: 2,
+ queue: { name: 'priority', priority: 5, concurrency: 2 },
cron: '15 * * * *',
timeZone: 'UTC',
maxAttempts: 7,
@@ -84,7 +82,7 @@ export const storedDefinition: JobDefinitionStatus & { acquisition: JobAcquisiti
export const pausedDefinition: JobDefinitionStatus & { acquisition: JobAcquisitionState } = {
...storedDefinition,
name: 'SendGreeting',
- queueName: 'default',
+ queue: { ...storedDefinition.queue, name: 'default' },
cron: null,
timeout: null,
slidingWindowMax: 10,
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..ec840d3b 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,12 +144,12 @@ 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,
- QueuePriority = definition.QueuePriority,
- QueueConcurrency = definition.QueueConcurrency,
+ QueueName = definition.Queue.Name,
+ QueuePriority = definition.Queue.Priority,
+ QueueConcurrency = definition.Queue.Concurrency,
Cron = definition.Cron,
TimeZone = definition.TimeZone,
MaxAttempts = definition.MaxAttempts,
@@ -165,12 +166,10 @@ 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,
- QueuePriority = entity.QueuePriority,
- QueueConcurrency = entity.QueueConcurrency,
+ Queue = new() { Name = entity.QueueName, Priority = entity.QueuePriority, Concurrency = entity.QueueConcurrency },
Cron = entity.Cron,
TimeZone = entity.TimeZone,
MaxAttempts = entity.MaxAttempts,
diff --git a/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.DefinitionCatalog.cs b/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.DefinitionCatalog.cs
index 00235d2a..559559b9 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,12 +256,12 @@ 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,
- QueuePriority = definition.QueuePriority,
- QueueConcurrency = definition.QueueConcurrency,
+ QueueName = definition.Queue.Name,
+ QueuePriority = definition.Queue.Priority,
+ QueueConcurrency = definition.Queue.Concurrency,
Cron = definition.Cron,
TimeZone = definition.TimeZone,
MaxAttempts = definition.MaxAttempts,
@@ -277,12 +278,10 @@ 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,
- QueuePriority = entity.QueuePriority,
- QueueConcurrency = entity.QueueConcurrency,
+ Queue = new() { Name = entity.QueueName, Priority = entity.QueuePriority, Concurrency = entity.QueueConcurrency },
Cron = entity.Cron,
TimeZone = entity.TimeZone,
MaxAttempts = entity.MaxAttempts,
diff --git a/src/Immediate.Jobs.Redis/RedisJobStorage.Acquire.cs b/src/Immediate.Jobs.Redis/RedisJobStorage.Acquire.cs
index 85a86632..1654011c 100644
--- a/src/Immediate.Jobs.Redis/RedisJobStorage.Acquire.cs
+++ b/src/Immediate.Jobs.Redis/RedisJobStorage.Acquire.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;
@@ -36,13 +36,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 = RedisJsonSerializerContext.DeserializeJobDefinition(metadata[index].ToString())
?? 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..d98f365e 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 => RedisJsonSerializerContext.DeserializeJobDefinition((string)entry.Value!)!)
.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..4e0a9d05 100644
--- a/src/Immediate.Jobs.Redis/RedisJobStorage.Definitions.cs
+++ b/src/Immediate.Jobs.Redis/RedisJobStorage.Definitions.cs
@@ -1,4 +1,3 @@
-using System.Text.Json;
using Microsoft.Extensions.Logging;
using Immediate.Jobs.Shared.Storage;
using StackExchange.Redis;
@@ -48,7 +47,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 = RedisJsonSerializerContext.DeserializeJobDefinition(metadata.ToString())
?? 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..bf17d14d 100644
--- a/src/Immediate.Jobs.Redis/RedisJobStorage.Records.cs
+++ b/src/Immediate.Jobs.Redis/RedisJobStorage.Records.cs
@@ -75,11 +75,8 @@ CancellationToken cancellationToken
continue;
}
- var definition = JsonSerializer.Deserialize(
- (string)metadata[index]!,
- RedisJsonSerializerContext.Default.JobDefinitionRecord
- ) ?? throw new ImmediateJobException($"Job definition '{names[index]}' contains invalid data.");
- queues[names[index]] = definition.QueueName;
+ var definition = RedisJsonSerializerContext.DeserializeJobDefinition((string)metadata[index]!) ?? throw new ImmediateJobException($"Job definition '{names[index]}' contains invalid data.");
+ queues[names[index]] = definition.Queue.Name;
}
return queues;
diff --git a/src/Immediate.Jobs.Redis/RedisJsonSerializerContext.cs b/src/Immediate.Jobs.Redis/RedisJsonSerializerContext.cs
index cdd17138..aab236e3 100644
--- a/src/Immediate.Jobs.Redis/RedisJsonSerializerContext.cs
+++ b/src/Immediate.Jobs.Redis/RedisJsonSerializerContext.cs
@@ -1,5 +1,7 @@
+using System.Text.Json;
using System.Text.Json.Serialization;
using Immediate.Jobs.Shared.Apis;
+using Immediate.Jobs.Shared.Internals;
namespace Immediate.Jobs.Redis;
@@ -7,6 +9,28 @@ 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;
+internal sealed partial class RedisJsonSerializerContext : JsonSerializerContext
+{
+ internal static JobDefinition? DeserializeJobDefinition(string json)
+ {
+ using var document = JsonDocument.Parse(json);
+ var definition = document.RootElement.Deserialize(Default.JobDefinition);
+ if (definition is null || document.RootElement.TryGetProperty("Queue", out _))
+ return definition;
+
+ // Read catalogues written before definitions stored their queue as a nested object.
+ return document.RootElement.TryGetProperty("QueueName", out var name)
+ ? definition with
+ {
+ Queue = new()
+ {
+ Name = name.GetString()!,
+ Priority = document.RootElement.TryGetProperty("QueuePriority", out var priority) ? priority.GetInt32() : 0,
+ Concurrency = document.RootElement.TryGetProperty("QueueConcurrency", out var concurrency) ? concurrency.GetInt32() : 0,
+ },
+ }
+ : definition;
+ }
+}
diff --git a/src/Immediate.Jobs.Redis/RedisScripts.cs b/src/Immediate.Jobs.Redis/RedisScripts.cs
index 91e3f1d5..aaec6833 100644
--- a/src/Immediate.Jobs.Redis/RedisScripts.cs
+++ b/src/Immediate.Jobs.Redis/RedisScripts.cs
@@ -4,6 +4,9 @@ internal static class RedisScripts
{
private const string FairQueueFunctions =
"""
+ local function queueName(definition)
+ return definition.Queue and definition.Queue.Name or definition.QueueName
+ end
local function fairDueKey(root, jobName, group)
if group == '' then return root .. 'fair:ungrouped:' .. jobName end
return root .. 'fair:due:' .. string.len(jobName) .. ':' .. jobName .. ':' .. group
@@ -27,13 +30,13 @@ local function changeActive(root, jobName, group, delta, id, lease)
end
local function queueForJob(root, jobName)
local metadata = redis.call('HGET', root .. 'definition-metadata', jobName)
- return metadata and cjson.decode(metadata).QueueName or nil
+ return metadata and queueName(cjson.decode(metadata)) or nil
end
local function removeClearedQueueCursor(root, queue, group)
if group == '' or not queue then return end
local definitions = redis.call('HGETALL', root .. 'definition-metadata')
for index = 1, #definitions, 2 do
- if cjson.decode(definitions[index + 1]).QueueName == queue then
+ if queueName(cjson.decode(definitions[index + 1])) == queue then
local name = definitions[index]
if redis.call('HEXISTS', root .. 'fair:active:' .. name, group) == 1 then return end
if redis.call('ZCARD', fairDueKey(root, name, group)) > 0 then return end
@@ -305,7 +308,7 @@ local function claim(queue, candidate)
if fair then
for index = 1, #definitions, 2 do
local name = definitions[index]
- if cjson.decode(definitions[index + 1]).QueueName == queue.name then
+ if queueName(cjson.decode(definitions[index + 1])) == queue.name then
for _, group in ipairs(redis.call('SMEMBERS', root .. 'fair:groups:' .. name)) do groups[group] = true end
for _, id in ipairs(redis.call('ZRANGEBYSCORE', root .. 'active:job:' .. name, '(' .. nowScore, '+inf')) do
local group = redis.call('HGET', root .. 'job:' .. id, 'group') or ''
@@ -649,7 +652,7 @@ local function addRecurringDue(root, dueKey, jobName, score, member)
local previousQueues = {}
for _, definition in ipairs(changes.Definitions) do
local previous = redis.call('HGET', KEYS[1], definition.Name)
- if previous then previousQueues[cjson.decode(previous).QueueName] = true end
+ if previous then previousQueues[queueName(cjson.decode(previous))] = true end
redis.call('HSET', KEYS[1], definition.Name, definition.Metadata)
end
for queue in pairs(previousQueues) do
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..469dc334 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,8 @@ public record JobDefinition
///
/// The queue used by newly-created invocations.
///
- public JobQueueDefinition Queue { get; init; } = JobQueueDefinition.Default;
+ [AllowNull]
+ public JobQueueDefinition Queue { get; init => field = value ?? JobQueueDefinition.Default; } = JobQueueDefinition.Default;
///
/// The stable job name.
@@ -29,14 +31,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 +90,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..666e94d7 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();
@@ -76,7 +77,7 @@ public async ValueTask> GetJobDefinitionsAsyn
}
private string? GetQueueName(string jobName) =>
- _jobDefinitions.TryGetValue(jobName, out var definition) ? definition.QueueName : null;
+ _jobDefinitions.TryGetValue(jobName, out var definition) ? definition.Queue.Name : null;
private JobRecord ReadRecord(JobRecord job) => job with { QueueName = GetQueueName(job.JobName) };
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..bf6ff0f1 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;
@@ -25,7 +26,7 @@ internal static class DefinitionCatalogStorageConformance
private static async ValueTask UsesCurrentDefinitionAsync(IJobStorage storage, FakeTimeProvider clock, CancellationToken token)
{
- var definition = Definition("Résumé") with { QueueName = "priority" };
+ var definition = Definition("Résumé") with { Queue = new() { Name = "priority" } };
await storage.MergeJobDefinitionsListAsync(Registration([definition]), token);
var job = new JobSubmission
{
@@ -39,7 +40,7 @@ private static async ValueTask UsesCurrentDefinitionAsync(IJobStorage storage, F
};
await storage.EnqueueAsync(job, token);
var persisted = (await storage.QueryJobsAsync(new() { JobHandle = job.JobHandle }, token)).Single();
- ConformanceAssert.Equal(new JobRecord { JobHandle = job.JobHandle, JobName = definition.Name, QueueName = definition.QueueName, Payload = job.Payload, GroupId = job.GroupId, State = job.State, DueAt = job.DueAt, CreatedAt = job.CreatedAt }, persisted,
+ ConformanceAssert.Equal(new JobRecord { JobHandle = job.JobHandle, JobName = definition.Name, QueueName = definition.Queue.Name, Payload = job.Payload, GroupId = job.GroupId, State = job.State, DueAt = job.DueAt, CreatedAt = job.CreatedAt }, persisted,
SubmissionCase, "reads must expose the canonical name and current queue while preserving invocation data");
var schedule = new RecurringJobSchedule
@@ -53,13 +54,13 @@ private static async ValueTask UsesCurrentDefinitionAsync(IJobStorage storage, F
};
await storage.UpsertRecurringAsync(schedule, token);
var persistedSchedule = (await MonitoringTestQueries.GetSchedulesAsync(storage, token)).Single();
- ConformanceAssert.Equal(schedule with { JobName = definition.Name, QueueName = definition.QueueName }, persistedSchedule,
+ ConformanceAssert.Equal(schedule with { JobName = definition.Name, QueueName = definition.Queue.Name }, persistedSchedule,
SubmissionCase, "dynamic schedules must use the persisted definition without a caller-supplied queue");
- await storage.MergeJobDefinitionsListAsync(Registration([definition with { QueueName = "PRIORITY" }]), token);
+ await storage.MergeJobDefinitionsListAsync(Registration([definition with { Queue = definition.Queue with { Name = "PRIORITY" } }]), token);
ConformanceAssert.Equal("PRIORITY", (await storage.QueryJobsAsync(new() { JobHandle = job.JobHandle }, token)).Single().QueueName,
SubmissionCase, "case-only queue changes must update persisted routing even on case-insensitive databases");
- await storage.MergeJobDefinitionsListAsync(Registration([definition with { QueueName = "changed" }]), token);
+ await storage.MergeJobDefinitionsListAsync(Registration([definition with { Queue = definition.Queue with { Name = "changed" } }]), token);
await storage.EnqueueAsync(job with { JobHandle = JobHandle.FromString("changed-queue"), State = JobState.WaitingForTrigger }, token);
ConformanceAssert.Equal("changed", (await storage.QueryJobsAsync(new() { State = JobState.WaitingForTrigger }, token)).Single().QueueName,
SubmissionCase, "later submissions must use the updated definition");
@@ -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)
@@ -101,9 +102,7 @@ private static async ValueTask RoundTripsConfigurationAsync(IJobStorage storage,
var definition = Definition("Configured") with
{
Tags = [JobTags.Default, "configuration"],
- QueueName = "priority",
- QueuePriority = 9,
- QueueConcurrency = 2,
+ Queue = new() { Name = "priority", Priority = 9, Concurrency = 2 },
Cron = "15 * * * *",
TimeZone = "America/Chicago",
MaxAttempts = 7,
@@ -133,8 +132,8 @@ 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) =>
- (value.Name, value.QueueName, value.QueuePriority, value.QueueConcurrency, value.Cron, value.TimeZone,
+ static object Values(JobDefinition value) =>
+ (value.Name, value.Queue.Name, value.Queue.Priority, value.Queue.Concurrency, value.Cron, value.TimeZone,
value.MaxAttempts, value.Timeout, value.MaxConcurrency, value.OverlapPolicy, value.MisfireHandlingMode,
value.Backoff, value.BackoffBase, value.SlidingWindowMax, value.SlidingWindowPeriod,
value.FixedWindowMax, value.FixedWindowPeriod);
@@ -142,7 +141,7 @@ static object Values(JobDefinitionRecord value) =>
private static async ValueTask ReconcilesCaseInsensitiveNamesAsync(IJobStorage storage, FakeTimeProvider timeProvider, CancellationToken token)
{
- var original = new[] { Definition("Invoice") with { QueueName = "priority", QueuePriority = 9, QueueConcurrency = 2, Timeout = TimeSpan.FromMinutes(3), MaxConcurrency = 4, SlidingWindowMax = 3, SlidingWindowPeriod = TimeSpan.FromSeconds(5) + TimeSpan.FromTicks(3), FixedWindowMax = 12, FixedWindowPeriod = TimeSpan.FromMinutes(1), BackoffBase = TimeSpan.FromSeconds(13) }, Definition("Résumé"), Definition("Δοκιμή"), Definition("測試"), Definition(new string('a', 256)) };
+ var original = new[] { Definition("Invoice") with { Queue = new() { Name = "priority", Priority = 9, Concurrency = 2 }, Timeout = TimeSpan.FromMinutes(3), MaxConcurrency = 4, SlidingWindowMax = 3, SlidingWindowPeriod = TimeSpan.FromSeconds(5) + TimeSpan.FromTicks(3), FixedWindowMax = 12, FixedWindowPeriod = TimeSpan.FromMinutes(1), BackoffBase = TimeSpan.FromSeconds(13) }, Definition("Résumé"), Definition("Δοκιμή"), Definition("測試"), Definition(new string('a', 256)) };
await storage.MergeJobDefinitionsListAsync(Registration(original), token);
var aliases = original.Zip(["INVOICE", "RéSUMé", "Δοκιμή", "測試", new string('A', 256)],
static (definition, name) => definition with { Name = name, MaxAttempts = 9 }).ToList();
@@ -150,7 +149,7 @@ private static async ValueTask ReconcilesCaseInsensitiveNamesAsync(IJobStorage s
var definitions = await storage.GetJobDefinitionsAsync(token);
var invoice = definitions.Single(static definition => string.Equals(definition.Name, "Invoice", StringComparison.Ordinal));
ConformanceAssert.Equal(("priority", 9, 2, TimeSpan.FromMinutes(3), 4, TimeSpan.FromSeconds(13)),
- (invoice.QueueName, invoice.QueuePriority, invoice.QueueConcurrency, invoice.Timeout, invoice.MaxConcurrency, invoice.BackoffBase), CaseNameCase, "non-runtime metadata must round trip");
+ (invoice.Queue.Name, invoice.Queue.Priority, invoice.Queue.Concurrency, invoice.Timeout, invoice.MaxConcurrency, invoice.BackoffBase), CaseNameCase, "non-runtime metadata must round trip");
ConformanceAssert.Equal(original[0].AcquisitionLimits, invoice.AcquisitionLimits, CaseNameCase, "remote monitoring and resume must use the persisted acquisition limits");
ConformanceAssert.Equal(5, definitions.Count, CaseNameCase, "ASCII case variants must update the same definitions, preserving Unicode characters and maximum-length names");
ConformanceAssert.True(definitions.All(static definition => definition.MaxAttempts == 9), CaseNameCase, "updates must reach case-insensitive identities");
@@ -162,11 +161,11 @@ 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,
- QueueName = definition.QueueName,
+ QueueName = definition.Queue.Name,
Cron = definition.Cron!,
TimeZone = definition.TimeZone,
IsCodeDefined = true,
@@ -187,7 +186,7 @@ private static async ValueTask PreservesRecurringAsync(IJobStorage storage, Fake
{
JobHandle = JobHandle.FromString("retained-cron-invocation"),
JobName = ordinary.Name,
- QueueName = ordinary.QueueName,
+ QueueName = ordinary.Queue.Name,
State = JobState.Pending,
Payload = "{}",
CreatedAt = now,
diff --git a/src/Immediate.Jobs.Testing/Storage/DefinitionRemovalStorageConformance.cs b/src/Immediate.Jobs.Testing/Storage/DefinitionRemovalStorageConformance.cs
index 0213ebe2..dbb5e718 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();
@@ -108,7 +109,7 @@ await graph.EnqueueBatchAsync(new() { BatchHandle = batch, CreatedAt = now, Stat
[parent, removedChild, keptChild],
[new() { ChildJobHandle = removedChild.JobHandle, ParentJobHandle = parent.JobHandle, Delay = TimeSpan.Zero, Trigger = ContinuationTrigger.Success },
new() { ChildJobHandle = keptChild.JobHandle, ParentJobHandle = parent.JobHandle, Delay = TimeSpan.Zero, Trigger = ContinuationTrigger.Complete }], token);
- await storage.MergeJobDefinitionsListAsync(new() { Definitions = [new() { Name = "Kept", QueueName = "changed" }] }, token);
+ await storage.MergeJobDefinitionsListAsync(new() { Definitions = [new() { Name = "Kept", Queue = new() { Name = "changed" } }] }, token);
ConformanceAssert.Equal(JobState.Cancelled, (await storage.GetJobStatusAsync(removedChild.JobHandle, token))!.State, GraphCase, "a removed child is canceled before dependency propagation");
ConformanceAssert.Equal(JobState.Pending, (await storage.GetJobStatusAsync(keptChild.JobHandle, token))!.State, GraphCase, "completion-triggered work for retained definitions is released");
ConformanceAssert.Equal("changed", (await storage.QueryJobsAsync(new() { JobHandle = keptChild.JobHandle }, token)).Single().QueueName,
diff --git a/src/Immediate.Jobs.Testing/Storage/GraphStorageConformance.cs b/src/Immediate.Jobs.Testing/Storage/GraphStorageConformance.cs
index 091e459d..8bf18eb8 100644
--- a/src/Immediate.Jobs.Testing/Storage/GraphStorageConformance.cs
+++ b/src/Immediate.Jobs.Testing/Storage/GraphStorageConformance.cs
@@ -48,7 +48,7 @@ private static async ValueTask ResolvesQueuesAtomicallyAsync(IJobStorage storage
var graph = GetGraph(storage, RoutingName);
await storage.MergeJobDefinitionsListAsync(new()
{
- Definitions = [new() { Name = "Routed", QueueName = "priority" }],
+ Definitions = [new() { Name = "Routed", Queue = new() { Name = "priority" } }],
}, token);
var batch = CreateBatch("routing-batch", 1);
var parent = CreateJob("routing-parent", batch.BatchHandle) with { JobName = "ROUTED" };
diff --git a/src/Immediate.Jobs.Testing/Storage/MonitoringStorageConformance.cs b/src/Immediate.Jobs.Testing/Storage/MonitoringStorageConformance.cs
index ffc8e29c..88cbf982 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()
{
@@ -75,7 +76,7 @@ private static async ValueTask MetadataAsync(IJobStorage storage, FakeTimeProvid
await storage.MergeJobDefinitionsListAsync(new()
{
ServerTags = ["remote"],
- Definitions = [new() { Name = "Remote", Tags = ["remote"], QueueName = "priority" }],
+ Definitions = [new() { Name = "Remote", Tags = ["remote"], Queue = new() { Name = "priority" } }],
}, token);
var schedule = new RecurringJobSchedule
{
diff --git a/src/Immediate.Jobs.Testing/Storage/QueueStorageConformance.cs b/src/Immediate.Jobs.Testing/Storage/QueueStorageConformance.cs
index 84f55d64..fc4bb4af 100644
--- a/src/Immediate.Jobs.Testing/Storage/QueueStorageConformance.cs
+++ b/src/Immediate.Jobs.Testing/Storage/QueueStorageConformance.cs
@@ -77,7 +77,7 @@ await storage.MergeJobDefinitionsListAsync(new()
{
Definitions =
[
- new() { Name = "RoundTrip.Job", QueueName = "queue-roundtrip" },
+ new() { Name = "RoundTrip.Job", Queue = new() { Name = "queue-roundtrip" } },
],
}, cancellationToken);
@@ -170,9 +170,9 @@ await storage.MergeJobDefinitionsListAsync(new()
{
Definitions =
[
- new() { Name = "low-job", QueueName = "low" },
- new() { Name = "limited-job", QueueName = "high" },
- new() { Name = "other-job", QueueName = "high" },
+ new() { Name = "low-job", Queue = new() { Name = "low" } },
+ new() { Name = "limited-job", Queue = new() { Name = "high" } },
+ new() { Name = "other-job", Queue = new() { Name = "high" } },
],
}, cancellationToken);
@@ -527,9 +527,9 @@ await storage.MergeJobDefinitionsListAsync(new()
{
Definitions =
[
- new() { Name = "Email.Receipt", QueueName = "priority" },
- new() { Name = "Email.Other", QueueName = "priority" },
- new() { Name = "Cleanup.Job", QueueName = "ordinary" },
+ new() { Name = "Email.Receipt", Queue = new() { Name = "priority" } },
+ new() { Name = "Email.Other", Queue = new() { Name = "priority" } },
+ new() { Name = "Cleanup.Job", Queue = new() { Name = "ordinary" } },
],
}, cancellationToken);
diff --git a/src/Immediate.Jobs.Testing/Storage/RecurringStorageConformance.cs b/src/Immediate.Jobs.Testing/Storage/RecurringStorageConformance.cs
index 27f97faa..c5bc4fc4 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;
@@ -86,7 +87,7 @@ await ConformanceAssert.ThrowsAsync(() => storage.RemoveR
await storage.PauseRecurringAsync(alias, token);
await storage.MergeJobDefinitionsListAsync(new()
{
- Definitions = [new() { Name = alias, QueueName = QueueName, Cron = code.Cron, TimeZone = code.TimeZone }],
+ Definitions = [new() { Name = alias, Queue = new() { Name = QueueName }, Cron = code.Cron, TimeZone = code.TimeZone }],
RecurringSchedules = [code with { Name = alias, JobName = alias }],
}, token);
var reconciled = await GetScheduleAsync(storage, name, CaseName, token);
@@ -115,7 +116,7 @@ CancellationToken cancellationToken
{
await storage.MergeJobDefinitionsListAsync(new()
{
- Definitions = [new() { Name = JobName, QueueName = QueueName }],
+ Definitions = [new() { Name = JobName, Queue = new() { Name = QueueName } }],
}, cancellationToken);
var now = timeProvider.GetUtcNow();
@@ -203,10 +204,10 @@ 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",
+ Queue = new() { Name = schedule.QueueName ?? "default" },
Cron = schedule.Cron,
TimeZone = schedule.TimeZone,
}).ToList(),
@@ -269,7 +270,7 @@ CancellationToken cancellationToken
{
await storage.MergeJobDefinitionsListAsync(new()
{
- Definitions = [new() { Name = JobName, QueueName = QueueName }],
+ Definitions = [new() { Name = JobName, Queue = new() { Name = QueueName } }],
}, cancellationToken);
var now = timeProvider.GetUtcNow();
@@ -331,7 +332,7 @@ CancellationToken cancellationToken
var now = timeProvider.GetUtcNow();
await storage.MergeJobDefinitionsListAsync(new()
{
- Definitions = [new() { Name = JobName, QueueName = QueueName }],
+ Definitions = [new() { Name = JobName, Queue = new() { Name = QueueName } }],
}, cancellationToken);
var parent = new JobSubmission
{
@@ -521,7 +522,7 @@ CancellationToken cancellationToken
{
await storage.MergeJobDefinitionsListAsync(new()
{
- Definitions = [new() { Name = JobName, QueueName = QueueName }],
+ Definitions = [new() { Name = JobName, Queue = new() { Name = QueueName } }],
}, cancellationToken);
var now = timeProvider.GetUtcNow();
diff --git a/src/Immediate.Jobs.Testing/Storage/ReplicaStorageConformance.Enqueue.cs b/src/Immediate.Jobs.Testing/Storage/ReplicaStorageConformance.Enqueue.cs
index a349a935..a86fd1dd 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;
@@ -30,7 +31,7 @@ CancellationToken token
await writer.InitializeAsync(token);
await writer.MergeJobDefinitionsListAsync(new()
{
- Definitions = [new() { Name = JobName, QueueName = QueueName }],
+ Definitions = [new() { Name = JobName, Queue = new() { Name = QueueName } }],
}, token);
var job = Job("enqueue-race", clock.GetUtcNow());
@@ -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..a95055b1 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, Queue = new() { Name = QueueName }, Cron = "* * * * *" };
var schedule = new RecurringJobSchedule
{
Name = JobName,
@@ -122,7 +123,7 @@ await storage.MergeJobDefinitionsListAsync(new()
{
Definitions =
[
- new() { Name = JobName, QueueName = QueueName },
+ new() { Name = JobName, Queue = new() { Name = QueueName } },
],
}, cancellationToken);
@@ -191,7 +192,7 @@ await storage.MergeJobDefinitionsListAsync(new()
{
Definitions =
[
- new() { Name = JobName, QueueName = QueueName },
+ new() { Name = JobName, Queue = new() { Name = QueueName } },
],
}, cancellationToken);
@@ -230,7 +231,7 @@ await storage.MergeJobDefinitionsListAsync(new()
{
Definitions =
[
- new() { Name = JobName, QueueName = QueueName },
+ new() { Name = JobName, Queue = new() { Name = QueueName } },
],
}, cancellationToken);
@@ -305,7 +306,7 @@ await storage.MergeJobDefinitionsListAsync(new()
{
Definitions =
[
- new() { Name = JobName, QueueName = QueueName },
+ new() { Name = JobName, Queue = new() { Name = QueueName } },
],
}, cancellationToken);
@@ -346,7 +347,7 @@ await storage.MergeJobDefinitionsListAsync(new()
{
Definitions =
[
- new() { Name = JobName, QueueName = QueueName },
+ new() { Name = JobName, Queue = new() { Name = QueueName } },
],
}, cancellationToken);
@@ -419,7 +420,7 @@ await storage.MergeJobDefinitionsListAsync(new()
{
Definitions =
[
- new() { Name = JobName, QueueName = QueueName },
+ new() { Name = JobName, Queue = new() { Name = QueueName } },
],
}, cancellationToken);
diff --git a/src/Immediate.Jobs.Testing/Storage/TagStorageConformance.cs b/src/Immediate.Jobs.Testing/Storage/TagStorageConformance.cs
index 61aa332d..01af22b0 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)
@@ -59,9 +60,7 @@ private static async ValueTask ReconcilesScopesAsync(IJobStorage storage, FakeTi
await storage.MergeJobDefinitionsListAsync(Registration([ordinary, other, shared], "email"), token);
var updated = ordinary with
{
- QueueName = "priority",
- QueuePriority = 9,
- QueueConcurrency = 2,
+ Queue = ordinary.Queue with { Name = "priority", Priority = 9, Concurrency = 2 },
MaxAttempts = 8,
MaxConcurrency = 4,
Timeout = TimeSpan.FromMinutes(3),
@@ -78,9 +77,9 @@ private static async ValueTask ReconcilesScopesAsync(IJobStorage storage, FakeTi
ConformanceAssert.SequenceEqual(["email", "ordinary"], definitions.Select(static definition => definition.Name), ScopeCase, "complete-list reconciliation must remove scoped omissions and preserve other scopes");
var actual = definitions.Single(definition => string.Equals(definition.Name, ordinary.Name, StringComparison.Ordinal));
ConformanceAssert.Equal(
- (updated.Name, updated.QueueName, updated.QueuePriority, updated.QueueConcurrency, updated.Cron, updated.TimeZone,
+ (updated.Name, updated.Queue.Name, updated.Queue.Priority, updated.Queue.Concurrency, updated.Cron, updated.TimeZone,
updated.MaxAttempts, updated.MaxConcurrency, updated.Timeout, updated.OverlapPolicy, updated.MisfireHandlingMode, updated.Backoff, updated.BackoffBase),
- (actual.Name, actual.QueueName, actual.QueuePriority, actual.QueueConcurrency, actual.Cron, actual.TimeZone,
+ (actual.Name, actual.Queue.Name, actual.Queue.Priority, actual.Queue.Concurrency, actual.Cron, actual.TimeZone,
actual.MaxAttempts, actual.MaxConcurrency, actual.Timeout, actual.OverlapPolicy, actual.MisfireHandlingMode, actual.Backoff, actual.BackoffBase),
ScopeCase, "all non-runtime metadata must round trip and repeated startup must be idempotent");
ConformanceAssert.SequenceEqual(["default"], actual.Tags, ScopeCase, "empty job tags must persist as default");
@@ -111,11 +110,11 @@ 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,
- QueueName = definition.QueueName,
+ QueueName = definition.Queue.Name,
Cron = definition.Cron!,
TimeZone = definition.TimeZone,
IsCodeDefined = true,
@@ -140,7 +139,7 @@ private static async ValueTask PreservesRecurringAsync(IJobStorage storage, Fake
{
JobHandle = JobHandle.FromString("retained-cron-invocation"),
JobName = ordinary.Name,
- QueueName = ordinary.QueueName,
+ QueueName = ordinary.Queue.Name,
State = JobState.Pending,
Payload = "{}",
CreatedAt = now,
diff --git a/tests/Immediate.Jobs.FunctionalTests/DefinitionAcquisitionTests.cs b/tests/Immediate.Jobs.FunctionalTests/DefinitionAcquisitionTests.cs
index 1bed7b06..646f2e83 100644
--- a/tests/Immediate.Jobs.FunctionalTests/DefinitionAcquisitionTests.cs
+++ b/tests/Immediate.Jobs.FunctionalTests/DefinitionAcquisitionTests.cs
@@ -105,7 +105,7 @@ public async Task SchedulerAndMonitorControlsUseCurrentPersistedLimitsAndConcurr
var original = Assert.Single(await harness.Storage.GetJobDefinitionsAsync(token));
var definition = original with
{
- QueueName = "stored-queue",
+ Queue = original.Queue with { Name = "stored-queue" },
MaxAttempts = 7,
SlidingWindowMax = 3,
SlidingWindowPeriod = TimeSpan.FromSeconds(5),
@@ -114,14 +114,14 @@ public async Task SchedulerAndMonitorControlsUseCurrentPersistedLimitsAndConcurr
};
await harness.Storage.MergeJobDefinitionsListAsync(new() { Definitions = [definition] }, token);
var first = await scheduler.EnqueueAsync(new(), token);
- Assert.Equal(definition.QueueName, (await harness.GetJobAsync(first, token)).QueueName);
+ Assert.Equal(definition.Queue.Name, (await harness.GetJobAsync(first, token)).QueueName);
await scheduler.EnqueueAsync(new(), token);
var acquired = await harness.Storage.AcquireDueJobsAsync(new()
{
WorkerId = "monitor-worker",
Lease = TimeSpan.FromSeconds(30),
BatchSize = 2,
- Queues = [new() { QueueName = definition.QueueName, Capacity = 2, JobCapacities = new Dictionary(StringComparer.OrdinalIgnoreCase) { [definition.Name] = 2 } }],
+ Queues = [new() { QueueName = definition.Queue.Name, Capacity = 2, JobCapacities = new Dictionary(StringComparer.OrdinalIgnoreCase) { [definition.Name] = 2 } }],
}, token);
Assert.Equal(2, acquired.Count);
await scheduler.PauseJobAsync(token);
@@ -180,7 +180,7 @@ await harness.Storage.AcquireDueJobsAsync(new()
WorkerId = "monitor-worker",
Lease = TimeSpan.FromSeconds(30),
BatchSize = 1,
- Queues = [new() { QueueName = definition.QueueName, Capacity = 1, JobCapacities = new Dictionary(StringComparer.Ordinal) { [definition.Name] = 1 } }],
+ Queues = [new() { QueueName = definition.Queue.Name, Capacity = 1, JobCapacities = new Dictionary(StringComparer.Ordinal) { [definition.Name] = 1 } }],
}, token);
await scheduler.PauseJobAsync(token);
await harness.Services.GetRequiredService().DrainAsync(token);
@@ -204,7 +204,7 @@ public async Task MonitorUsesStoredInformationForJobsAndRecurringTriggersWithout
await harness.DrainAsync(token);
await harness.Storage.MergeJobDefinitionsListAsync(new()
{
- Definitions = [new() { Name = "Remote-Job", QueueName = "current-queue", MaxAttempts = 7 }],
+ Definitions = [new() { Name = "Remote-Job", Queue = new() { Name = "current-queue" }, MaxAttempts = 7 }],
}, token);
await harness.Storage.UpsertRecurringAsync(new()
{
@@ -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/MonitoringApiTests.cs b/tests/Immediate.Jobs.FunctionalTests/MonitoringApiTests.cs
index b9026246..a88d1472 100644
--- a/tests/Immediate.Jobs.FunctionalTests/MonitoringApiTests.cs
+++ b/tests/Immediate.Jobs.FunctionalTests/MonitoringApiTests.cs
@@ -28,7 +28,7 @@ public async Task ManualRecurringTriggerChecksConfigurationAndCancellationStatus
var token = TestContext.Current.CancellationToken;
var clock = new FakeTimeProvider(DateTimeOffset.UnixEpoch);
using var storage = new MonitoringStorage(clock);
- await storage.MergeJobDefinitionsListAsync(new() { Definitions = [new() { Name = "Remote", QueueName = "current" }] }, token);
+ await storage.MergeJobDefinitionsListAsync(new() { Definitions = [new() { Name = "Remote", Queue = new() { Name = "current" } }] }, token);
await storage.UpsertRecurringAsync(new()
{
Name = "Résumé",
diff --git a/tests/Immediate.Jobs.FunctionalTests/Packages/DashboardPackageTests.cs b/tests/Immediate.Jobs.FunctionalTests/Packages/DashboardPackageTests.cs
index 67cd1241..45cda8b9 100644
--- a/tests/Immediate.Jobs.FunctionalTests/Packages/DashboardPackageTests.cs
+++ b/tests/Immediate.Jobs.FunctionalTests/Packages/DashboardPackageTests.cs
@@ -48,7 +48,7 @@ await storage.EnqueueAsync(new()
WorkerId = "dashboard-worker",
Lease = TimeSpan.FromSeconds(30),
BatchSize = 1,
- Queues = [new() { QueueName = definition.QueueName, Capacity = 1, JobCapacities = new Dictionary(StringComparer.Ordinal) { [definition.Name] = 2 } }],
+ Queues = [new() { QueueName = definition.Queue.Name, Capacity = 1, JobCapacities = new Dictionary(StringComparer.Ordinal) { [definition.Name] = 2 } }],
};
Assert.Single(await storage.AcquireDueJobsAsync(request, token));
var pendingHandle = JobHandle.FromString("definition-action-pending");
@@ -148,6 +148,13 @@ 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("queue").GetProperty("name").GetString());
+ Assert.False(definition.TryGetProperty("queueName", out _));
+ Assert.False(definition.TryGetProperty("queuePriority", out _));
+ Assert.False(definition.TryGetProperty("queueConcurrency", 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
@@ -190,7 +197,7 @@ await storage.MergeJobDefinitionsListAsync(new()
{
Name = "remote-email",
Tags = ["email"],
- QueueName = "priority",
+ Queue = new() { Name = "priority" },
MaxAttempts = 7,
SlidingWindowMax = 1,
SlidingWindowPeriod = TimeSpan.FromMinutes(1),
@@ -221,7 +228,7 @@ await storage.HeartbeatAsync(new()
Assert.Equal(["ordinary", "remote-email"], definitions.RootElement.GetProperty("jobs").EnumerateArray().Select(static item => item.GetProperty("name").GetString()));
var remote = definitions.RootElement.GetProperty("jobs")[1];
Assert.Equal("email", remote.GetProperty("tags")[0].GetString());
- Assert.Equal("priority", remote.GetProperty("queueName").GetString());
+ Assert.Equal("priority", remote.GetProperty("queue").GetProperty("name").GetString());
Assert.Equal(7, remote.GetProperty("maxAttempts").GetInt32());
var monitor = app.Services.GetRequiredService();
var remoteStatus = (await monitor.GetDefinitionsAsync(token)).Jobs.Single(definition => string.Equals(definition.Name, "remote-email", StringComparison.OrdinalIgnoreCase));
@@ -232,7 +239,7 @@ [new JobRecord
{
JobHandle = JobHandle.FromString("remote-definition-active"),
JobName = remoteStatus.Name,
- QueueName = remoteStatus.QueueName,
+ QueueName = remoteStatus.Queue.Name,
Payload = "{}",
State = JobState.Pending,
CreatedAt = clock.GetUtcNow(),
@@ -243,7 +250,7 @@ [new JobRecord
WorkerId = "remote-worker",
Lease = TimeSpan.FromMinutes(5),
BatchSize = 1,
- Queues = [new() { QueueName = remoteStatus.QueueName, Capacity = 1, JobCapacities = new Dictionary(StringComparer.Ordinal) { [remoteStatus.Name] = 1 } }],
+ Queues = [new() { QueueName = remoteStatus.Queue.Name, Capacity = 1, JobCapacities = new Dictionary(StringComparer.Ordinal) { [remoteStatus.Name] = 1 } }],
}, token));
foreach (var action in new[] { "pause", "resume" })
{
diff --git a/tests/Immediate.Jobs.FunctionalTests/PersistedQueueSchedulerTests.cs b/tests/Immediate.Jobs.FunctionalTests/PersistedQueueSchedulerTests.cs
index 94bda179..f9c4e15d 100644
--- a/tests/Immediate.Jobs.FunctionalTests/PersistedQueueSchedulerTests.cs
+++ b/tests/Immediate.Jobs.FunctionalTests/PersistedQueueSchedulerTests.cs
@@ -73,9 +73,10 @@ public async Task DynamicRecurringScheduleUsesPersistedQueue()
await harness.DrainAsync(token);
await using var scope = harness.Services.CreateAsyncScope();
var scheduler = scope.ServiceProvider.GetRequiredService();
+ var definition = (await harness.Storage.GetJobDefinitionsAsync(token)).Single(item => string.Equals(item.Name, scheduler.JobName, StringComparison.OrdinalIgnoreCase));
await harness.Storage.MergeJobDefinitionsListAsync(new()
{
- Definitions = [(await harness.Storage.GetJobDefinitionsAsync(token)).Single(definition => string.Equals(definition.Name, scheduler.JobName, StringComparison.OrdinalIgnoreCase)) with { QueueName = "recurring-queue" }],
+ Definitions = [definition with { Queue = definition.Queue with { Name = "recurring-queue" } }],
}, token);
await scheduler.AddOrUpdateRecurringAsync("dynamic-queue", "* * * * *", "UTC", token);
Assert.Equal("recurring-queue", (await harness.Storage.GetMonitoringDefinitionsAsync(token)).Recurring.Single(static schedule => string.Equals(schedule.Name, "dynamic-queue", StringComparison.Ordinal)).QueueName);
@@ -96,7 +97,7 @@ private static async ValueTask ChangeQueueAsync(JobTestHarness harness, string j
await harness.Storage.MergeJobDefinitionsListAsync(new()
{
Definitions = (await harness.Storage.GetJobDefinitionsAsync(token)).Select(definition => string.Equals(definition.Name, jobName, StringComparison.OrdinalIgnoreCase)
- ? definition with { Name = jobName, QueueName = queueName } : definition).ToList(),
+ ? definition with { Name = jobName, Queue = definition.Queue with { Name = queueName } } : definition).ToList(),
RecurringSchedules = (await harness.Storage.GetMonitoringDefinitionsAsync(token)).Recurring.Where(static definition => definition.IsCodeDefined)
.Select(definition => Immediate.Jobs.Shared.Apis.RecurringJobSchedule.FromDefinition(definition, snapshot.Recurring.Single(status => string.Equals(status.Name, definition.Name, StringComparison.Ordinal)))).ToList(),
}, token);
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..90575ef2
--- /dev/null
+++ b/tests/Immediate.Jobs.StorageTests/JobDefinitionSerializationTests.cs
@@ -0,0 +1,92 @@
+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("Queue").GetProperty("Name").GetString());
+ Assert.Equal(9, root.GetProperty("Queue").GetProperty("Priority").GetInt32());
+ Assert.Equal(2, root.GetProperty("Queue").GetProperty("Concurrency").GetInt32());
+ Assert.False(root.TryGetProperty("QueueName", out _));
+ Assert.False(root.TryGetProperty("QueuePriority", out _));
+ Assert.False(root.TryGetProperty("QueueConcurrency", 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(RedisJsonSerializerContext.DeserializeJobDefinition(Json));
+ 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 { Queue = definition.Queue with { Name = "changed", Concurrency = 4 } };
+ Assert.Equal(new() { Name = "changed", Priority = 9, Concurrency = 4 }, updated.Queue);
+ Assert.Equal("priority", definition.Queue.Name);
+ Assert.Equal(2, definition.Queue.Concurrency);
+ }
+
+ 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..d117f708 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;
@@ -31,6 +32,54 @@ public async Task RedisConforms(JobStorageConformanceTestCase testCase)
await testCase.RunAsync(fixture.Services, TestContext.Current.CancellationToken);
}
+
+ [Fact]
+ public async Task LegacyQueueMetadataSupportsAcquisitionAndCatalogueUpdates()
+ {
+ var token = TestContext.Current.CancellationToken;
+ await using var fixture = await RedisConformanceFixture.CreateAsync(
+ redis.Container.GetConnectionString(),
+ new() { Jobs = [], Batches = [], Edges = [], RecurringSchedules = [] }
+ );
+ await fixture.SetDefinitionMetadataAsync("legacy", """
+ {"Name":"legacy","QueueName":"old-queue","QueuePriority":9,"QueueConcurrency":2,"Tags":["default"]}
+ """);
+ var storage = fixture.Services.GetRequiredService();
+ var definition = Assert.Single(await storage.GetJobDefinitionsAsync(token));
+ Assert.Equal(new() { Name = "old-queue", Priority = 9, Concurrency = 2 }, definition.Queue);
+ var now = fixture.Services.GetRequiredService().GetUtcNow();
+ foreach (var queue in new[] { "old-queue", "new-queue" })
+ {
+ if (string.Equals(queue, "new-queue", StringComparison.Ordinal))
+ {
+ definition = definition with { Queue = definition.Queue with { Name = queue } };
+ await storage.MergeJobDefinitionsListAsync(new() { Definitions = [definition] }, token);
+ }
+
+ var handle = JobHandle.FromString(queue);
+ await storage.EnqueueAsync(new()
+ {
+ JobHandle = handle,
+ JobName = definition.Name,
+ GroupId = "tenant",
+ Payload = "{}",
+ State = JobState.Pending,
+ DueAt = now,
+ CreatedAt = now,
+ }, token);
+ var acquired = Assert.Single(await storage.AcquireDueJobsAsync(new()
+ {
+ WorkerId = "worker",
+ Lease = TimeSpan.FromMinutes(1),
+ BatchSize = 1,
+ Queues = [new() { QueueName = queue, Capacity = 1, JobCapacities = new Dictionary(StringComparer.Ordinal) { [definition.Name] = 1 } }],
+ FairQueues = new() { ConcurrencyShareThreshold = 0.1, MinInflightForNoisy = 30, GroupRoundRobin = true },
+ }, token));
+ Assert.Equal(handle, acquired.JobHandle);
+ Assert.Equal(queue, acquired.QueueName);
+ await storage.CompleteAsync(handle, acquired.Attempt, "worker", token);
+ }
+ }
}
file sealed class RedisConformanceFixture : IAsyncDisposable
@@ -51,6 +100,9 @@ ServiceProvider services
internal IServiceProvider Services { get; }
+ internal Task SetDefinitionMetadataAsync(string name, string json) =>
+ _connection.GetDatabase().HashSetAsync($"{{{_keyPrefix}}}:definition-metadata", name.ToUpperInvariant(), json);
+
internal static async ValueTask CreateAsync(
string connectionString,
PersistedJobState persistedJobState
@@ -107,7 +159,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, Queue = new() { Name = job.QueueName ?? "default" } }).ToList(),
});
foreach (var job in jobs)
await storage.EnqueueAsync(new()
|