Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions docs/architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -59,7 +59,11 @@ 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 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
`IJobStorage.AcquireDueJobsAsync`, and writes the claimed records to an unbounded channel. On
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ defineEmits<{
<tbody>
<tr v-for="definition in definitions" :key="definition.name">
<td>
<RouterLink :to="{ name: 'jobs', query: { search: definition.name, queue: definition.queueName } }">
<RouterLink :to="{ name: 'jobs', query: { search: definition.name, queue: definition.queue.name } }">
<code>{{ definition.name }}</code>
</RouterLink>
</td>
Expand All @@ -61,7 +61,7 @@ defineEmits<{
</button>
</div>
</td>
<td>{{ definition.queueName }}</td>
<td>{{ definition.queue.name }}</td>
<td><span v-for="tag in definition.tags" :key="tag" class="tag">{{ tag }}</span></td>
<td>
<div v-if="definition.cron" class="schedule-cell"><code>{{ definition.cron }}</code><span>{{ definition.timeZone }}</span></div>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ import {
canCancelJob,
canRetryJob,
type JobAcquisitionState,
type JobDefinitionRecord,
type JobDefinition,
type JobExecutionRecord,
type JobRecord,
type JobTelemetryLink,
Expand All @@ -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;
Expand Down Expand Up @@ -402,9 +402,9 @@ function retryButtonLabel(job: JobRecord, pending: boolean): string {
<h3>Definition</h3>
<template v-if="definition">
<dl class="detail-list">
<div><dt>Queue</dt><dd><code>{{ definition.queueName }}</code></dd></div>
<div><dt>Queue priority</dt><dd>{{ definition.queuePriority }}</dd></div>
<div><dt>Queue concurrency</dt><dd>{{ definition.queueConcurrency > 0 ? definition.queueConcurrency : 'No limit' }}</dd></div>
<div><dt>Queue</dt><dd><code>{{ definition.queue.name }}</code></dd></div>
<div><dt>Queue priority</dt><dd>{{ definition.queue.priority }}</dd></div>
<div><dt>Queue concurrency</dt><dd>{{ definition.queue.concurrency > 0 ? definition.queue.concurrency : 'No limit' }}</dd></div>
<div><dt>Tags</dt><dd><span v-for="tag in definition.tags" :key="tag" class="tag">{{ tag }}</span></dd></div>
<div><dt>Schedule</dt><dd><code v-if="definition.cron">{{ definition.cron }}</code><span v-else>On demand</span></dd></div>
<div><dt>Time zone</dt><dd>{{ definition.timeZone }}</dd></div>
Expand Down
16 changes: 10 additions & 6 deletions src/Immediate.Jobs.Dashboard/DashboardClient/src/contracts.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -149,7 +153,7 @@ export interface JobAcquisitionState {
isConcurrencyLimited: boolean;
}

export interface JobDefinitionStatus extends JobDefinitionRecord {
export interface JobDefinitionStatus extends JobDefinition {
limits: JobAcquisitionLimits;
acquisition?: JobAcquisitionState | undefined;
}
Expand Down Expand Up @@ -258,6 +262,6 @@ export type RecurringJobDefinition = Omit<RecurringJobSchedule, 'isPaused' | 'ca
export type RecurringJobStatus = Pick<RecurringJobSchedule, 'name' | 'isPaused' | 'cancellationReason' | 'nextRunAt' | 'lastRunAt'>;

export interface JobMonitoringDefinitions {
jobs: JobDefinitionRecord[];
jobs: JobDefinition[];
recurring: RecurringJobDefinition[];
}
4 changes: 2 additions & 2 deletions src/Immediate.Jobs.Dashboard/DashboardClient/src/query.ts
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ import type {
DashboardJobPage,
DashboardState,
JobFilters,
JobDefinitionRecord,
JobDefinition,
JobDefinitionStatus,
JobMonitoringDefinitions,
JobMonitoringSnapshot,
Expand Down Expand Up @@ -105,7 +105,7 @@ export function useJobDefinitionQuery(jobName: MaybeRefOrGetter<string | undefin
status.jobName.toUpperCase() === name.value));

return {
data: computed<JobDefinitionRecord | undefined>(() => {
data: computed<JobDefinition | undefined>(() => {
if (snapshot.data.value && !acquisition.value) {
return undefined;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
});
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 },
}] },
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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,
Expand Down
Original file line number Diff line number Diff line change
@@ -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<JobDefinitionRecord>))]
[JsonSerializable(typeof(IReadOnlyList<JobDefinition>))]
[JsonSerializable(typeof(JobMonitoringDefinitions))]
[JsonSerializable(typeof(DashboardState))]
[JsonSerializable(typeof(DashboardJobPage))]
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -95,15 +96,15 @@ await context.Set<ImmediateFairQueueGroupEntity>()
}

/// <inheritdoc />
public async ValueTask<IReadOnlyList<JobDefinitionRecord>> GetJobDefinitionsAsync(CancellationToken cancellationToken = default)
public async ValueTask<IReadOnlyList<JobDefinition>> GetJobDefinitionsAsync(CancellationToken cancellationToken = default)
{
GetJobDefinitionsAsyncCalled();
cancellationToken.ThrowIfCancellationRequested();
await TaskScheduler.Yield();
return await ReadWithStrategyAsync(ReadJobDefinitionsAsync, cancellationToken);
}

private static async Task<IReadOnlyList<JobDefinitionRecord>> ReadJobDefinitionsAsync(TContext context, CancellationToken cancellationToken)
private static async Task<IReadOnlyList<JobDefinition>> ReadJobDefinitionsAsync(TContext context, CancellationToken cancellationToken)
{
var definitions = await context.Set<ImmediateJobDefinitionMetadataEntity>()
.AsNoTracking()
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand Down
21 changes: 10 additions & 11 deletions src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.DefinitionCatalog.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -68,7 +69,7 @@ private Task<int> RemoveEmptyFairQueueGroupsAsync(DataConnection connection, Can
.DeleteAsync(token);

/// <inheritdoc />
public async ValueTask<IReadOnlyList<JobDefinitionRecord>> GetJobDefinitionsAsync(CancellationToken cancellationToken = default)
public async ValueTask<IReadOnlyList<JobDefinition>> GetJobDefinitionsAsync(CancellationToken cancellationToken = default)
{
GetJobDefinitionsAsyncCalled();
cancellationToken.ThrowIfCancellationRequested();
Expand All @@ -77,7 +78,7 @@ public async ValueTask<IReadOnlyList<JobDefinitionRecord>> GetJobDefinitionsAsyn
return await ReadJobDefinitionsAsync(connection, cancellationToken);
}

private async Task<IReadOnlyList<JobDefinitionRecord>> ReadJobDefinitionsAsync(DataConnection connection, CancellationToken cancellationToken)
private async Task<IReadOnlyList<JobDefinition>> ReadJobDefinitionsAsync(DataConnection connection, CancellationToken cancellationToken)
{
var definitions = await Definitions(connection).ToListAsync(cancellationToken);
return [.. definitions.Select(ToRecord).OrderBy(static definition => definition.Name, StringComparer.OrdinalIgnoreCase)];
Expand Down Expand Up @@ -165,7 +166,7 @@ await Recurring(connection).Where(item => item.Name == schedule.Name)

private async Task MergeDefinitionMetadataAsync(
DataConnection connection,
IReadOnlyList<JobDefinitionRecord> definitions,
IReadOnlyList<JobDefinition> definitions,
IQueryable<string> serverTags,
CancellationToken cancellationToken
)
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand Down
6 changes: 3 additions & 3 deletions src/Immediate.Jobs.Redis/RedisJobStorage.Acquire.cs
Original file line number Diff line number Diff line change
@@ -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;
Expand Down Expand Up @@ -36,13 +36,13 @@ public async ValueTask<IReadOnlyList<JobRecord>> AcquireDueJobsAsync(
{
cancellationToken.ThrowIfCancellationRequested();
var metadata = await Database.HashGetAsync(DefinitionMetadataKey, definitionKeys).WaitAsync(cancellationToken);
var definitions = new Dictionary<string, (RedisValue Metadata, JobDefinitionRecord Definition)>(StringComparer.Ordinal);
var definitions = new Dictionary<string, (RedisValue Metadata, JobDefinition Definition)>(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));
}
Expand Down
8 changes: 4 additions & 4 deletions src/Immediate.Jobs.Redis/RedisJobStorage.DefinitionCatalog.cs
Original file line number Diff line number Diff line change
@@ -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;
Expand Down Expand Up @@ -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(),
Expand Down Expand Up @@ -65,13 +65,13 @@ public async ValueTask MergeJobDefinitionsListAsync(JobDefinitionRegistration re
}

/// <inheritdoc />
public async ValueTask<IReadOnlyList<JobDefinitionRecord>> GetJobDefinitionsAsync(CancellationToken cancellationToken = default)
public async ValueTask<IReadOnlyList<JobDefinition>> 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)];
}

Expand Down
3 changes: 1 addition & 2 deletions src/Immediate.Jobs.Redis/RedisJobStorage.Definitions.cs
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
using System.Text.Json;
using Microsoft.Extensions.Logging;
using Immediate.Jobs.Shared.Storage;
using StackExchange.Redis;
Expand Down Expand Up @@ -48,7 +47,7 @@ private async ValueTask<JobAcquisitionState> 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();
Expand Down
7 changes: 2 additions & 5 deletions src/Immediate.Jobs.Redis/RedisJobStorage.Records.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Loading
Loading