diff --git a/docs/architecture.md b/docs/architecture.md
index b7a15f57..4b0bbb0b 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -32,7 +32,7 @@ For each `[Job]` class (see `Templates/Job.sbntxt`), nested in the user's partia
- **`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,
- overlap and misfire policies, concurrency).
+ overlap and misfire policies, and concurrency).
- **`PayloadJsonContext`**: hand-emitted System.Text.Json metadata for the payload and context
types, so serialization never uses reflection.
@@ -44,8 +44,11 @@ configuration surface (`ConfigureStorage`, `ConfigureWorkers`, `UseFairQueues`,
`JobSchedulingService` (`Internals/JobSchedulingService.cs`) is one hosted service per process:
+- **Startup.** Initializes storage and submits the complete definition catalogue and code-defined
+ schedules through `MergeJobDefinitionsListAsync`, even when workers are disabled. The storage
+ reconciles the complete catalogue. Metadata has no invoker or CLR type.
- **Polling loop.** Each iteration materializes due recurring schedules, builds a
- `JobAcquisitionRequest` from free worker capacity and queue/job concurrency limits, calls
+ `JobAcquisitionRequest` from local definitions, free worker capacity, and queue/job concurrency limits, calls
`IJobStorage.AcquireDueJobsAsync`, and writes the claimed records to an unbounded channel. On
`PurgeInterval` it also purges job and batch history. It then waits `PollingInterval`.
- **Workers.** `WorkerCount` tasks read the channel and run `ExecuteJobAsync`: a new DI scope, a
diff --git a/docs/conventions.md b/docs/conventions.md
index 72a9d636..a08594bc 100644
--- a/docs/conventions.md
+++ b/docs/conventions.md
@@ -47,10 +47,10 @@ assembly's `LibraryEventIds`:
| Assembly | Range in use |
| --- | --- |
-| `Immediate.Jobs.Shared` | 11000–11093 (scheduler, in-memory, single-server) |
-| `Immediate.Jobs.EntityFrameworkCore` | 11500–11542 |
-| `Immediate.Jobs.LinqToDB` | 11600–11642 |
-| `Immediate.Jobs.Redis` | 11700–11728 |
+| `Immediate.Jobs.Shared` | 11000–11097 (scheduler, in-memory, single-server) |
+| `Immediate.Jobs.EntityFrameworkCore` | 11500–11544 |
+| `Immediate.Jobs.LinqToDB` | 11600–11644 |
+| `Immediate.Jobs.Redis` | 11700–11730 |
Append new ids after the current maximum of the assembly's range; never reuse or renumber an id,
because users filter on them. `EventName` is `Immediate.Jobs.{Assembly}.{Name}`.
diff --git a/docs/recurring-jobs.md b/docs/recurring-jobs.md
index 4b1b55e3..2ce7c223 100644
--- a/docs/recurring-jobs.md
+++ b/docs/recurring-jobs.md
@@ -17,9 +17,10 @@ A `RecurringJobSchedule` has a unique `Name`, the `JobName` it materializes, a `
`TimeZoneInfo.FindSystemTimeZoneById`. The analyzers validate code-defined expressions at compile
time (`IJOB0007`).
-At startup `JobSchedulingService.InitializeAsync` calls `MergeRecurringSchedulesListAsync` with every
-code-defined schedule. Storage replaces the code-defined set (removing schedules for deleted jobs)
-and leaves dynamic schedules untouched.
+At startup `JobSchedulingService.InitializeAsync` submits all definitions and their code-defined schedules
+through `MergeJobDefinitionsListAsync`. Storage atomically reconciles the catalogue, removing code-defined schedules when a definition disappears or loses its cron expression. It preserves
+dynamic schedules. Unchanged cron/time zone retains the next occurrence; updates
+preserve the pause state and last occurrence.
## Materialization
diff --git a/docs/storage.md b/docs/storage.md
index 2862b734..0e36a61a 100644
--- a/docs/storage.md
+++ b/docs/storage.md
@@ -5,7 +5,7 @@
Two interfaces in `src/Immediate.Jobs.Shared/Storage`:
- **`IJobStorage`**: queueing, acquisition (including fair-queue policies), leases, completion and
- failure, recurring schedules, wait-for-trigger updates and triggers, queries, monitoring snapshots,
+ failure, definition catalogues, recurring schedules, wait-for-trigger updates and triggers, queries, monitoring snapshots,
heartbeats, purging. Every provider implements all of it.
- **`IJobGraphStorage : IJobStorage`**: batches, continuations, dynamic additions from running jobs,
batch triggers, batch reads and purging, plus the two single-server recovery reads
@@ -30,6 +30,45 @@ monitoring snapshots and health-check data.
is preserved through settlement and dynamic additions, and `BatchState.WaitingForTrigger` is
non-terminal.
+## Definition reconciliation
+
+`MergeJobDefinitionsListAsync(JobDefinitionRegistration)` receives the complete local definition list,
+and all code-defined schedules. It inserts or updates supplied definitions and removes stored
+definitions absent from that list. An empty list clears the catalogue.
+Definition deletion never cascades to invocations, running leases, or execution history.
+
+Definition names are case-insensitive identities and retain their readable spelling. In-memory
+and single-server use `OrdinalIgnoreCase`; SQL Server inherits database collation; PostgreSQL uses
+a nondeterministic ICU collation (`und-u-ks-level2`); SQLite uses `NOCASE` (ASCII-only database
+comparison). EF tracking uses an ordinal case-insensitive key comparer. EF applications with custom
+provider conventions can pass `definitionNameCollation` to `AddImmediateJobs`.
+Definition names must not have leading or trailing whitespace: the job analyzer reports `IJOB0008`,
+and storage rejects invalid manually supplied names before modifying the catalogue.
+
+Relational providers store definition configuration in individual columns in
+`immediate_job_definition_metadata`; the name key is the only persisted definition name.
+`TimeoutTicks` and `BackoffBaseTicks` store exact .NET ticks (100 nanoseconds), and policy enums
+use small integers. Redis continues storing serialized definition records.
+
+Concurrent startup snapshots commit atomically. In-memory uses its gate; EF Core takes an early write
+lock on a singleton catalogue row. LinqToDB uses a serializable transaction and merges supplied
+records directly into each table on SQL Server and PostgreSQL 15 or later. Matched updates preserve
+canonical names and recurring progress using target values. SQL Server removes obsolete rows in the
+merge; PostgreSQL uses a separate set-based delete in the same transaction because PostgreSQL 15 has
+no WHEN NOT MATCHED BY SOURCE clause. Only SQLite reads a snapshot for client-side reconciliation; it uses transactional upserts and deletes because
+it does not support SQL `MERGE`. Serialization conflicts retry the complete LinqToDB snapshot.
+Redis compares a version and applies metadata and
+schedule changes in one Lua script, recomputing after a competing startup. Single-server delegates
+metadata reads and reconciliation to durable storage, then refreshes the primary's code-defined
+schedules. Pause and unchanged schedule progress survive startup.
+
+Each application's list is authoritative for the catalogue. Concurrent applications with inconsistent
+lists can remove or replace each other's definitions; the last successful reconciliation wins.
+
+Custom providers must implement the catalogue operations. Relational users must add the definition metadata table
+through their normal schema update process; EF Core also requires the singleton catalogue table. The
+bootstrap helpers do not upgrade existing production databases.
+
## Providers
| Provider | Location | Concurrency idiom |
@@ -51,7 +90,8 @@ back a job transition.
Tables: `immediate_jobs`, `immediate_job_executions`, `immediate_job_continuations`,
`immediate_job_batches`, `immediate_recurring_jobs`, `immediate_job_servers`, and
-`immediate_fair_queue_groups`.
+`immediate_fair_queue_groups`, and `immediate_job_definition_metadata`. EF Core also uses
+`immediate_job_definition_catalog` to serialize catalogue updates.
- **EF Core**: applications call `modelBuilder.AddImmediateJobs()` and own their migrations. A schema
change is a breaking change for every EF Core user, and needs a migration note in the pull request.
@@ -75,7 +115,9 @@ may build keys from the root):
| `leases` | sorted set | Active job ids by lease expiry. |
| `executions:index:{id}` / `executions:data:{id}` | sorted set / hash | Execution history. |
| `recurring:*` | various | Schedules, due index, names, materialization de-duplication. |
-| `server:{worker}` / `servers` | hash / sorted set | Heartbeats. |
+| `server:{worker}` / `servers` | hash / sorted set | Scheduler heartbeats. |
+| `definition-metadata` | hash | All definition metadata keyed by stable job name. |
+| `definition-catalog-version` | string | Version used to atomically reconcile a complete catalogue. |
| `fair:*` | various | Fair-queue indexes, active counts, and cursors; see [fair queues](fair-queues.md). |
The payload lives only inside `record`. Lua must not re-encode it with `cjson` (key order and escaping
diff --git a/docs/testing.md b/docs/testing.md
index 70c625ea..e6f11002 100644
--- a/docs/testing.md
+++ b/docs/testing.md
@@ -11,9 +11,8 @@
**Guidance for adding tests:**
-- **Storage behaviour belongs in the conformance suite**, not in provider-specific tests, so every
- provider is held to it. Write a provider-specific test only for behaviour that exists in that
- provider alone.
+- **Every storage behaviour test belongs in the conformance suite**, so every provider is held to
+ the same contract. Do not write standalone or provider-specific storage tests.
- **Functional tests cover only what the layer above storage adds**, for example a scheduler turning
`TryTriggerAsync`'s `false` into an exception, or argument validation. Don't re-assert storage
semantics there.
@@ -25,7 +24,7 @@
`src/Immediate.Jobs.Testing/Storage` ships in the `Immediate.Jobs.Testing` package, so custom
providers can run it too.
-- **Case files.** `QueueStorageConformance`, `RecurringStorageConformance`,
+- **Case files.** `QueueStorageConformance`, `DefinitionCatalogStorageConformance`, `RecurringStorageConformance`,
`FairQueueStorageConformance`, `TriggerStorageConformance`, `GraphStorageConformance`, and
`ReplicaStorageConformance`.
- **Case shape.** Each case is `new(Name, RequiredCapabilities, Scenario[, PersistedJobState])`. Names
diff --git a/readme.md b/readme.md
index 5b83c072..88ab4103 100644
--- a/readme.md
+++ b/readme.md
@@ -52,6 +52,15 @@ public sealed class SignupService(SendWelcomeEmail.Scheduler welcomeEmail)
Register the generated handlers and jobs methods in `Program.cs`, in that order. For an assembly named `MyApp`, these
are `AddMyAppHandlers()` and `AddMyAppJobs()`.
+## Persisted definitions
+
+Startup provides the complete local catalogue to storage, which inserts, updates, and removes job
+definitions and code-defined schedules. The dashboard shows every stored definition, including those
+absent from its local registrations.
+
+Job names match case-insensitively. Explicit job names must not have leading or trailing whitespace; the analyzer reports `IJOB0008`
+for invalid names. Storage also rejects such names in manually supplied definition catalogues.
+
## Scheduling and handles
Generated schedulers use one method name for relative and absolute scheduling. Pass a `TimeSpan` for a delay or a
diff --git a/samples/DistributedAspire/DistributedAspire.Shared/Migrations/20261009144316_DefinitionCatalog.Designer.cs b/samples/DistributedAspire/DistributedAspire.Shared/Migrations/20261009144316_DefinitionCatalog.Designer.cs
new file mode 100644
index 00000000..81663e7a
--- /dev/null
+++ b/samples/DistributedAspire/DistributedAspire.Shared/Migrations/20261009144316_DefinitionCatalog.Designer.cs
@@ -0,0 +1,489 @@
+//
+using System;
+using Immediate.Jobs.DistributedAspire.Shared.Data;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Infrastructure;
+using Microsoft.EntityFrameworkCore.Migrations;
+using Microsoft.EntityFrameworkCore.Storage.ValueConversion;
+using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata;
+
+#nullable disable
+
+namespace Immediate.Jobs.DistributedAspire.Shared.Migrations
+{
+ [DbContext(typeof(JobsDbContext))]
+ [Migration("20261009144316_DefinitionCatalog")]
+ partial class DefinitionCatalog
+ {
+ ///
+ protected override void BuildTargetModel(ModelBuilder modelBuilder)
+ {
+#pragma warning disable 612, 618
+ modelBuilder
+ .HasAnnotation("Npgsql:CollationDefinition:immediate_jobs_case_insensitive", "und-u-ks-level2,und-u-ks-level2,icu,False")
+ .HasAnnotation("ProductVersion", "10.0.12")
+ .HasAnnotation("Relational:MaxIdentifierLength", 63);
+
+ NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder);
+
+ modelBuilder.Entity("Immediate.Jobs.DistributedAspire.Shared.Data.BatchEntity", b =>
+ {
+ b.Property("Id")
+ .ValueGeneratedOnAdd()
+ .HasColumnType("integer")
+ .HasColumnName("Id");
+
+ NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id"));
+
+ b.Property("CreatedOn")
+ .HasColumnType("timestamp with time zone")
+ .HasColumnName("CreatedOn");
+
+ b.Property("Data")
+ .HasColumnType("uuid")
+ .HasColumnName("Data");
+
+ b.Property("ModifiedOn")
+ .HasColumnType("timestamp with time zone");
+
+ b.Property("Version")
+ .IsConcurrencyToken()
+ .HasColumnType("bigint")
+ .HasColumnName("Version");
+
+ b.HasKey("Id");
+
+ b.HasIndex("Id")
+ .IsUnique();
+
+ b.ToTable("BatchableRows");
+ });
+
+ modelBuilder.Entity("Immediate.Jobs.EntityFrameworkCore.ImmediateFairQueueGroupEntity", b =>
+ {
+ b.Property("QueueName")
+ .HasMaxLength(256)
+ .HasColumnType("character varying(256)");
+
+ b.Property("GroupId")
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)");
+
+ b.Property("ConcurrencyStamp")
+ .IsConcurrencyToken()
+ .HasColumnType("uuid");
+
+ b.Property("LastServedSequence")
+ .HasColumnType("bigint");
+
+ b.HasKey("QueueName", "GroupId");
+
+ b.ToTable("immediate_fair_queue_groups", (string)null);
+ });
+
+ modelBuilder.Entity("Immediate.Jobs.EntityFrameworkCore.ImmediateJobBatchEntity", b =>
+ {
+ b.Property("Id")
+ .HasMaxLength(256)
+ .HasColumnType("character varying(256)");
+
+ b.Property("CancelledCount")
+ .HasColumnType("integer");
+
+ b.Property("CompletedAt")
+ .HasColumnType("bigint");
+
+ b.Property("ConcurrencyStamp")
+ .IsConcurrencyToken()
+ .HasColumnType("uuid");
+
+ b.Property("CreatedAt")
+ .HasColumnType("bigint");
+
+ b.Property("FailedCount")
+ .HasColumnType("integer");
+
+ b.Property("PendingCount")
+ .HasColumnType("integer");
+
+ b.Property("SkippedCount")
+ .HasColumnType("integer");
+
+ b.Property("StartedAt")
+ .HasColumnType("bigint");
+
+ b.Property("State")
+ .HasColumnType("smallint");
+
+ b.Property("SucceededCount")
+ .HasColumnType("integer");
+
+ b.Property("TotalJobs")
+ .HasColumnType("integer");
+
+ b.HasKey("Id");
+
+ b.HasIndex("State", "CompletedAt");
+
+ b.ToTable("immediate_job_batches", (string)null);
+ });
+
+ modelBuilder.Entity("Immediate.Jobs.EntityFrameworkCore.ImmediateJobContinuationEntity", b =>
+ {
+ b.Property("ChildJobHandle")
+ .HasMaxLength(256)
+ .HasColumnType("character varying(256)");
+
+ b.Property("ParentKind")
+ .HasColumnType("smallint");
+
+ b.Property("ParentId")
+ .HasMaxLength(256)
+ .HasColumnType("character varying(256)");
+
+ b.Property("Delay")
+ .HasMaxLength(32)
+ .HasColumnType("bigint");
+
+ b.Property("ParentOutcome")
+ .HasColumnType("smallint");
+
+ b.Property("Trigger")
+ .HasColumnType("smallint");
+
+ b.HasKey("ChildJobHandle", "ParentKind", "ParentId");
+
+ b.HasIndex("ParentKind", "ParentId");
+
+ b.ToTable("immediate_job_continuations", (string)null);
+ });
+
+ modelBuilder.Entity("Immediate.Jobs.EntityFrameworkCore.ImmediateJobDefinitionCatalogEntity", b =>
+ {
+ b.Property("Id")
+ .HasColumnType("integer");
+
+ b.Property("ConcurrencyStamp")
+ .HasColumnType("uuid");
+
+ b.HasKey("Id");
+
+ b.ToTable("immediate_job_definition_catalog", (string)null);
+ });
+
+ modelBuilder.Entity("Immediate.Jobs.EntityFrameworkCore.ImmediateJobDefinitionMetadataEntity", b =>
+ {
+ b.Property("Name")
+ .HasMaxLength(256)
+ .HasColumnType("character varying(256)")
+ .UseCollation("immediate_jobs_case_insensitive");
+
+ b.Property("Backoff")
+ .HasColumnType("smallint");
+
+ b.Property("BackoffBaseTicks")
+ .HasColumnType("bigint");
+
+ b.Property("Cron")
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)");
+
+ b.Property("MaxAttempts")
+ .HasColumnType("integer");
+
+ b.Property("MaxConcurrency")
+ .HasColumnType("integer");
+
+ b.Property("MisfireHandlingMode")
+ .HasColumnType("smallint");
+
+ b.Property("OverlapPolicy")
+ .HasColumnType("smallint");
+
+ b.Property("QueueConcurrency")
+ .HasColumnType("integer");
+
+ b.Property("QueueName")
+ .IsRequired()
+ .HasMaxLength(256)
+ .HasColumnType("character varying(256)");
+
+ b.Property("QueuePriority")
+ .HasColumnType("integer");
+
+ b.Property("TimeZone")
+ .IsRequired()
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)");
+
+ b.Property("TimeoutTicks")
+ .HasColumnType("bigint");
+
+ b.HasKey("Name");
+
+ b.ToTable("immediate_job_definition_metadata", (string)null);
+ });
+
+ modelBuilder.Entity("Immediate.Jobs.EntityFrameworkCore.ImmediateJobEntity", b =>
+ {
+ b.Property("Id")
+ .HasMaxLength(256)
+ .HasColumnType("character varying(256)");
+
+ b.Property("Attempt")
+ .HasColumnType("integer");
+
+ b.Property("BatchHandle")
+ .HasMaxLength(256)
+ .HasColumnType("character varying(256)");
+
+ b.Property("CompletedAt")
+ .HasColumnType("bigint");
+
+ b.Property("ConcurrencyStamp")
+ .IsConcurrencyToken()
+ .HasColumnType("uuid");
+
+ b.Property("Context")
+ .HasColumnType("text");
+
+ b.Property("CreatedAt")
+ .HasColumnType("bigint");
+
+ b.Property("DueAt")
+ .HasColumnType("bigint");
+
+ b.Property("ExecutionSpanId")
+ .HasMaxLength(16)
+ .HasColumnType("character varying(16)");
+
+ b.Property("ExecutionStartedAt")
+ .HasColumnType("bigint");
+
+ b.Property("ExecutionTraceId")
+ .HasMaxLength(32)
+ .HasColumnType("character varying(32)");
+
+ b.Property("FailedDependencies")
+ .HasColumnType("integer");
+
+ b.Property("GroupId")
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)");
+
+ b.Property("JobName")
+ .IsRequired()
+ .HasMaxLength(256)
+ .HasColumnType("character varying(256)")
+ .UseCollation("immediate_jobs_case_insensitive");
+
+ b.Property("LastError")
+ .HasColumnType("text");
+
+ b.Property("LeaseExpiresAt")
+ .HasColumnType("bigint");
+
+ b.Property("Payload")
+ .IsRequired()
+ .HasColumnType("text");
+
+ b.Property("QueueName")
+ .IsRequired()
+ .ValueGeneratedOnAdd()
+ .HasMaxLength(256)
+ .HasColumnType("character varying(256)")
+ .HasDefaultValue("default");
+
+ b.Property("RecurringKey")
+ .HasMaxLength(512)
+ .HasColumnType("character varying(512)");
+
+ b.Property("RemainingDependencies")
+ .HasColumnType("integer");
+
+ b.Property("State")
+ .HasColumnType("smallint");
+
+ b.Property("TraceParent")
+ .HasMaxLength(256)
+ .HasColumnType("character varying(256)");
+
+ b.Property("TraceState")
+ .HasColumnType("text");
+
+ b.Property("WorkerId")
+ .HasMaxLength(256)
+ .HasColumnType("character varying(256)");
+
+ b.HasKey("Id");
+
+ b.HasIndex("BatchHandle");
+
+ b.HasIndex("RecurringKey")
+ .IsUnique();
+
+ b.HasIndex("State", "CreatedAt");
+
+ b.HasIndex("State", "DueAt");
+
+ b.HasIndex("QueueName", "State", "GroupId");
+
+ b.HasIndex("QueueName", "State", "DueAt", "CreatedAt");
+
+ b.ToTable("immediate_jobs", (string)null);
+ });
+
+ modelBuilder.Entity("Immediate.Jobs.EntityFrameworkCore.ImmediateJobExecutionEntity", b =>
+ {
+ b.Property("JobHandle")
+ .HasMaxLength(256)
+ .HasColumnType("character varying(256)");
+
+ b.Property("Attempt")
+ .HasColumnType("integer");
+
+ b.Property("AcquiredAt")
+ .HasColumnType("bigint");
+
+ b.Property("CompletedAt")
+ .HasColumnType("bigint");
+
+ b.Property("Error")
+ .HasColumnType("text");
+
+ b.Property("ExecutionSpanId")
+ .HasMaxLength(16)
+ .HasColumnType("character varying(16)");
+
+ b.Property("ExecutionStartedAt")
+ .HasColumnType("bigint");
+
+ b.Property("ExecutionTraceId")
+ .HasMaxLength(32)
+ .HasColumnType("character varying(32)");
+
+ b.Property("IsSynthetic")
+ .ValueGeneratedOnAdd()
+ .HasColumnType("boolean")
+ .HasDefaultValue(false);
+
+ b.Property("State")
+ .HasColumnType("smallint");
+
+ b.Property("WorkerId")
+ .HasMaxLength(256)
+ .HasColumnType("character varying(256)");
+
+ b.HasKey("JobHandle", "Attempt");
+
+ b.ToTable("immediate_job_executions", (string)null);
+ });
+
+ modelBuilder.Entity("Immediate.Jobs.EntityFrameworkCore.ImmediateJobServerEntity", b =>
+ {
+ b.Property("WorkerId")
+ .HasMaxLength(256)
+ .HasColumnType("character varying(256)");
+
+ b.Property("ActiveWorkers")
+ .HasColumnType("integer");
+
+ b.Property("Details")
+ .IsRequired()
+ .HasColumnType("text");
+
+ b.Property("ExpiresAt")
+ .HasColumnType("bigint");
+
+ b.Property("LastHeartbeat")
+ .HasColumnType("bigint");
+
+ b.Property("MaxWorkers")
+ .HasColumnType("integer");
+
+ b.HasKey("WorkerId");
+
+ b.HasIndex("ExpiresAt");
+
+ b.ToTable("immediate_job_servers", (string)null);
+ });
+
+ modelBuilder.Entity("Immediate.Jobs.EntityFrameworkCore.ImmediateRecurringJobEntity", b =>
+ {
+ b.Property("Name")
+ .HasMaxLength(256)
+ .HasColumnType("character varying(256)");
+
+ b.Property("ConcurrencyStamp")
+ .IsConcurrencyToken()
+ .HasColumnType("uuid");
+
+ b.Property("Cron")
+ .IsRequired()
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)");
+
+ b.Property("IsCodeDefined")
+ .HasColumnType("boolean");
+
+ b.Property("IsPaused")
+ .HasColumnType("boolean");
+
+ b.Property("JobName")
+ .IsRequired()
+ .HasMaxLength(256)
+ .HasColumnType("character varying(256)")
+ .UseCollation("immediate_jobs_case_insensitive");
+
+ b.Property("LastRunAt")
+ .HasColumnType("bigint");
+
+ b.Property("NextRunAt")
+ .HasColumnType("bigint");
+
+ b.Property("QueueName")
+ .IsRequired()
+ .HasMaxLength(256)
+ .HasColumnType("character varying(256)");
+
+ b.Property("TimeZone")
+ .IsRequired()
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)");
+
+ b.HasKey("Name");
+
+ b.HasIndex("IsPaused", "NextRunAt");
+
+ b.ToTable("immediate_recurring_jobs", (string)null);
+ });
+
+ modelBuilder.Entity("Immediate.Jobs.EntityFrameworkCore.ImmediateJobContinuationEntity", b =>
+ {
+ b.HasOne("Immediate.Jobs.EntityFrameworkCore.ImmediateJobEntity", null)
+ .WithMany()
+ .HasForeignKey("ChildJobHandle")
+ .OnDelete(DeleteBehavior.Cascade)
+ .IsRequired();
+ });
+
+ modelBuilder.Entity("Immediate.Jobs.EntityFrameworkCore.ImmediateJobEntity", b =>
+ {
+ b.HasOne("Immediate.Jobs.EntityFrameworkCore.ImmediateJobBatchEntity", null)
+ .WithMany()
+ .HasForeignKey("BatchHandle")
+ .OnDelete(DeleteBehavior.Cascade);
+ });
+
+ modelBuilder.Entity("Immediate.Jobs.EntityFrameworkCore.ImmediateJobExecutionEntity", b =>
+ {
+ b.HasOne("Immediate.Jobs.EntityFrameworkCore.ImmediateJobEntity", null)
+ .WithMany()
+ .HasForeignKey("JobHandle")
+ .OnDelete(DeleteBehavior.Cascade)
+ .IsRequired();
+ });
+#pragma warning restore 612, 618
+ }
+ }
+}
diff --git a/samples/DistributedAspire/DistributedAspire.Shared/Migrations/20261009144316_DefinitionCatalog.cs b/samples/DistributedAspire/DistributedAspire.Shared/Migrations/20261009144316_DefinitionCatalog.cs
new file mode 100644
index 00000000..349022ba
--- /dev/null
+++ b/samples/DistributedAspire/DistributedAspire.Shared/Migrations/20261009144316_DefinitionCatalog.cs
@@ -0,0 +1,78 @@
+using System.Diagnostics.CodeAnalysis;
+using Microsoft.EntityFrameworkCore.Migrations;
+
+#nullable disable
+
+namespace Immediate.Jobs.DistributedAspire.Shared.Migrations;
+
+///
+[SuppressMessage("Design", "CA1062:Validate arguments of public methods")]
+public sealed partial class DefinitionCatalog : Migration
+{
+ ///
+ protected override void Up(MigrationBuilder migrationBuilder)
+ {
+ migrationBuilder.AlterDatabase()
+ .Annotation("Npgsql:CollationDefinition:immediate_jobs_case_insensitive", "und-u-ks-level2,und-u-ks-level2,icu,False");
+
+ migrationBuilder.AlterColumn(
+ name: "JobName", table: "immediate_jobs", type: "character varying(256)",
+ maxLength: 256, nullable: false, collation: "immediate_jobs_case_insensitive",
+ oldClrType: typeof(string), oldType: "character varying(256)", oldMaxLength: 256);
+
+ migrationBuilder.AlterColumn(
+ name: "JobName", table: "immediate_recurring_jobs", type: "character varying(256)",
+ maxLength: 256, nullable: false, collation: "immediate_jobs_case_insensitive",
+ oldClrType: typeof(string), oldType: "character varying(256)", oldMaxLength: 256);
+
+ migrationBuilder.CreateTable(
+ name: "immediate_job_definition_catalog",
+ columns: table => new
+ {
+ Id = table.Column(type: "integer", nullable: false),
+ ConcurrencyStamp = table.Column(type: "uuid", nullable: false),
+ },
+ constraints: table => table.PrimaryKey("PK_immediate_job_definition_catalog", x => x.Id));
+
+ migrationBuilder.CreateTable(
+ name: "immediate_job_definition_metadata",
+ columns: table => new
+ {
+ Name = table.Column(type: "character varying(256)", maxLength: 256, nullable: false, collation: "immediate_jobs_case_insensitive"),
+ QueueName = table.Column(type: "character varying(256)", maxLength: 256, nullable: false),
+ QueuePriority = table.Column(type: "integer", nullable: false),
+ QueueConcurrency = table.Column(type: "integer", nullable: false),
+ Cron = table.Column(type: "character varying(128)", maxLength: 128, nullable: true),
+ TimeZone = table.Column(type: "character varying(128)", maxLength: 128, nullable: false),
+ MaxAttempts = table.Column(type: "integer", nullable: false),
+ TimeoutTicks = table.Column(type: "bigint", nullable: true),
+ MaxConcurrency = table.Column(type: "integer", nullable: false),
+ OverlapPolicy = table.Column(type: "smallint", nullable: false),
+ MisfireHandlingMode = table.Column(type: "smallint", nullable: false),
+ Backoff = table.Column(type: "smallint", nullable: false),
+ BackoffBaseTicks = table.Column(type: "bigint", nullable: false),
+ },
+ constraints: table => table.PrimaryKey("PK_immediate_job_definition_metadata", x => x.Name));
+ }
+
+ ///
+ protected override void Down(MigrationBuilder migrationBuilder)
+ {
+ migrationBuilder.AlterColumn(
+ name: "JobName", table: "immediate_jobs", type: "character varying(256)",
+ maxLength: 256, nullable: false,
+ oldClrType: typeof(string), oldType: "character varying(256)", oldMaxLength: 256,
+ oldCollation: "immediate_jobs_case_insensitive");
+
+ migrationBuilder.AlterColumn(
+ name: "JobName", table: "immediate_recurring_jobs", type: "character varying(256)",
+ maxLength: 256, nullable: false,
+ oldClrType: typeof(string), oldType: "character varying(256)", oldMaxLength: 256,
+ oldCollation: "immediate_jobs_case_insensitive");
+
+ migrationBuilder.DropTable(name: "immediate_job_definition_catalog");
+ migrationBuilder.DropTable(name: "immediate_job_definition_metadata");
+ migrationBuilder.AlterDatabase()
+ .OldAnnotation("Npgsql:CollationDefinition:immediate_jobs_case_insensitive", "und-u-ks-level2,und-u-ks-level2,icu,False");
+ }
+}
diff --git a/samples/DistributedAspire/DistributedAspire.Shared/Migrations/JobsDbContextModelSnapshot.cs b/samples/DistributedAspire/DistributedAspire.Shared/Migrations/JobsDbContextModelSnapshot.cs
index 0f8aea7b..b61e76c0 100644
--- a/samples/DistributedAspire/DistributedAspire.Shared/Migrations/JobsDbContextModelSnapshot.cs
+++ b/samples/DistributedAspire/DistributedAspire.Shared/Migrations/JobsDbContextModelSnapshot.cs
@@ -17,6 +17,7 @@ protected override void BuildModel(ModelBuilder modelBuilder)
{
#pragma warning disable 612, 618
modelBuilder
+ .HasAnnotation("Npgsql:CollationDefinition:immediate_jobs_case_insensitive", "und-u-ks-level2,und-u-ks-level2,icu,False")
.HasAnnotation("ProductVersion", "10.0.12")
.HasAnnotation("Relational:MaxIdentifierLength", 63);
@@ -154,6 +155,72 @@ protected override void BuildModel(ModelBuilder modelBuilder)
b.ToTable("immediate_job_continuations", (string)null);
});
+ modelBuilder.Entity("Immediate.Jobs.EntityFrameworkCore.ImmediateJobDefinitionCatalogEntity", b =>
+ {
+ b.Property("Id")
+ .HasColumnType("integer");
+
+ b.Property("ConcurrencyStamp")
+ .HasColumnType("uuid");
+
+ b.HasKey("Id");
+
+ b.ToTable("immediate_job_definition_catalog", (string)null);
+ });
+
+ modelBuilder.Entity("Immediate.Jobs.EntityFrameworkCore.ImmediateJobDefinitionMetadataEntity", b =>
+ {
+ b.Property("Name")
+ .HasMaxLength(256)
+ .HasColumnType("character varying(256)")
+ .UseCollation("immediate_jobs_case_insensitive");
+
+ b.Property("Backoff")
+ .HasColumnType("smallint");
+
+ b.Property("BackoffBaseTicks")
+ .HasColumnType("bigint");
+
+ b.Property("Cron")
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)");
+
+ b.Property("MaxAttempts")
+ .HasColumnType("integer");
+
+ b.Property("MaxConcurrency")
+ .HasColumnType("integer");
+
+ b.Property("MisfireHandlingMode")
+ .HasColumnType("smallint");
+
+ b.Property("OverlapPolicy")
+ .HasColumnType("smallint");
+
+ b.Property("QueueConcurrency")
+ .HasColumnType("integer");
+
+ b.Property("QueueName")
+ .IsRequired()
+ .HasMaxLength(256)
+ .HasColumnType("character varying(256)");
+
+ b.Property("QueuePriority")
+ .HasColumnType("integer");
+
+ b.Property("TimeZone")
+ .IsRequired()
+ .HasMaxLength(128)
+ .HasColumnType("character varying(128)");
+
+ b.Property("TimeoutTicks")
+ .HasColumnType("bigint");
+
+ b.HasKey("Name");
+
+ b.ToTable("immediate_job_definition_metadata", (string)null);
+ });
+
modelBuilder.Entity("Immediate.Jobs.EntityFrameworkCore.ImmediateJobEntity", b =>
{
b.Property("Id")
@@ -204,7 +271,8 @@ protected override void BuildModel(ModelBuilder modelBuilder)
b.Property("JobName")
.IsRequired()
.HasMaxLength(256)
- .HasColumnType("character varying(256)");
+ .HasColumnType("character varying(256)")
+ .UseCollation("immediate_jobs_case_insensitive");
b.Property("LastError")
.HasColumnType("text");
@@ -361,7 +429,8 @@ protected override void BuildModel(ModelBuilder modelBuilder)
b.Property("JobName")
.IsRequired()
.HasMaxLength(256)
- .HasColumnType("character varying(256)");
+ .HasColumnType("character varying(256)")
+ .UseCollation("immediate_jobs_case_insensitive");
b.Property("LastRunAt")
.HasColumnType("bigint");
diff --git a/src/Common/Utility.cs b/src/Common/Utility.cs
index ea3fd1a2..6ff1413c 100644
--- a/src/Common/Utility.cs
+++ b/src/Common/Utility.cs
@@ -31,6 +31,9 @@ public bool HasNameContent()
public string AsCSharpLiteral() =>
SymbolDisplay.FormatLiteral(str, quote: true);
+ public bool HasLeadingOrTrailingWhitespace() =>
+ str.Length > 0 && (char.IsWhiteSpace(str[0]) || char.IsWhiteSpace(str[^1]));
+
public string EscapeIdentifier() =>
SyntaxFacts.GetKeywordKind(str) == SyntaxKind.None
&& SyntaxFacts.GetContextualKeywordKind(str) == SyntaxKind.None
diff --git a/src/Immediate.Jobs.Analyzers/DuplicateElementsAnalyzer.cs b/src/Immediate.Jobs.Analyzers/DuplicateElementsAnalyzer.cs
index eb3c4516..f20ceab8 100644
--- a/src/Immediate.Jobs.Analyzers/DuplicateElementsAnalyzer.cs
+++ b/src/Immediate.Jobs.Analyzers/DuplicateElementsAnalyzer.cs
@@ -157,7 +157,7 @@ Lock @lock
private static void AnalyzeDuplicateJobNames(CompilationAnalysisContext context, List jobs)
{
- foreach (var jobGroup in jobs.GroupBy(x => x.JobName, StringComparer.Ordinal).Where(g => g.Skip(1).Any()))
+ foreach (var jobGroup in jobs.GroupBy(x => x.JobName, StringComparer.OrdinalIgnoreCase).Where(g => g.Skip(1).Any()))
{
var classes = string.Join(", ", jobGroup.Select(l => l.ClassName));
diff --git a/src/Immediate.Jobs.Analyzers/JobClassAnalyzer.cs b/src/Immediate.Jobs.Analyzers/JobClassAnalyzer.cs
index cfa50532..d3776496 100644
--- a/src/Immediate.Jobs.Analyzers/JobClassAnalyzer.cs
+++ b/src/Immediate.Jobs.Analyzers/JobClassAnalyzer.cs
@@ -53,7 +53,7 @@ public sealed class JobClassAnalyzer : DiagnosticAnalyzer
category: "ImmediateJobs",
defaultSeverity: DiagnosticSeverity.Error,
isEnabledByDefault: true,
- description: "A job must have a name that can identify it in storage.",
+ description: "A job name must contain a letter or digit and must not have leading or trailing whitespace.",
customTags: [WellKnownDiagnosticTags.NotConfigurable]
);
@@ -130,7 +130,8 @@ private static void AnalyzeJobName(SymbolAnalysisContext context, AttributeData
{
context.CancellationToken.ThrowIfCancellationRequested();
- if (jobAttribute.GetJobName(className: context.Symbol.Name).HasNameContent())
+ var name = jobAttribute.GetJobName(className: context.Symbol.Name);
+ if (name.HasNameContent() && !name.HasLeadingOrTrailingWhitespace())
return;
var explicitName = jobAttribute.NamedArguments.GetStringValue("Name");
@@ -142,7 +143,9 @@ private static void AnalyzeJobName(SymbolAnalysisContext context, AttributeData
context.Symbol.Name,
explicitName is null
? $"a name cannot be derived from the class name `{context.Symbol.Name}`; rename the class or set `Name`"
- : "`Name` must contain at least one letter or digit"
+ : !name.HasNameContent()
+ ? "`Name` must contain at least one letter or digit"
+ : "`Name` must not have leading or trailing whitespace"
)
);
}
diff --git a/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.Acquire.cs b/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.Acquire.cs
index eb073d27..ab92344f 100644
--- a/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.Acquire.cs
+++ b/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.Acquire.cs
@@ -30,7 +30,7 @@ public async ValueTask> AcquireDueJobsAsync(
if (queueCapacity <= 0)
continue;
- var jobCapacities = queue.JobCapacities.ToDictionary(static pair => pair.Key, static pair => pair.Value, StringComparer.Ordinal);
+ var jobCapacities = queue.JobCapacities.ToDictionary(static pair => pair.Key, static pair => pair.Value, StringComparer.OrdinalIgnoreCase);
while (queueCapacity > 0)
{
var eligibleNames = jobCapacities.Where(static pair => pair.Value > 0).Select(static pair => pair.Key).ToList();
@@ -54,7 +54,7 @@ public async ValueTask> AcquireDueJobsAsync(
break;
var selected = new List(candidates.Count);
- var selectionCapacities = new Dictionary(jobCapacities, StringComparer.Ordinal);
+ var selectionCapacities = new Dictionary(jobCapacities, StringComparer.OrdinalIgnoreCase);
foreach (var candidate in candidates)
{
if (selectionCapacities[candidate.JobName] <= 0)
diff --git a/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.DefinitionCatalog.cs b/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.DefinitionCatalog.cs
new file mode 100644
index 00000000..58139e7f
--- /dev/null
+++ b/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.DefinitionCatalog.cs
@@ -0,0 +1,155 @@
+using Immediate.Jobs.Shared.Apis;
+using Immediate.Jobs.Shared.Storage;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Logging;
+
+namespace Immediate.Jobs.EntityFrameworkCore;
+
+internal sealed partial class EntityFrameworkCoreJobStorage
+ where TContext : DbContext
+{
+ ///
+ public async ValueTask MergeJobDefinitionsListAsync(JobDefinitionRegistration registration, CancellationToken cancellationToken = default)
+ {
+ MergeJobDefinitionsListAsyncCalled();
+ cancellationToken.ThrowIfCancellationRequested();
+ await TaskScheduler.Yield();
+ // Validate before taking a write lock or creating the catalogue sentinel.
+ JobDefinitionReconciliation.Create(registration, [], []);
+ await RetryConcurrencyAsync(async token =>
+ {
+ await EnsureDefinitionCatalogAsync(token);
+ await using var context = await contextFactory.CreateDbContextAsync(token);
+ await using var transaction = await context.Database.BeginTransactionAsync(token);
+ // A singleton row serializes the complete read/merge/write across application instances.
+ var stamp = Guid.NewGuid();
+ await context.Set().Where(item => item.Id == 1)
+ .ExecuteUpdateAsync(update => update.SetProperty(item => item.ConcurrencyStamp, stamp), token);
+ var definitions = await context.Set()
+ .ToDictionaryAsync(static item => item.Name, StringComparer.OrdinalIgnoreCase, token);
+ var schedules = await context.Set()
+ .ToDictionaryAsync(static item => item.Name, StringComparer.Ordinal, token);
+ var changes = JobDefinitionReconciliation.Create(registration,
+ definitions.Values.Select(ToRecord).ToList(),
+ schedules.Values.Select(ToDefinitionSchedule).ToList());
+ foreach (var name in changes.RemovedDefinitionNames)
+ context.Remove(definitions[name]);
+ foreach (var definition in changes.Definitions)
+ {
+ var entity = ToEntity(definition);
+ if (definitions.TryGetValue(definition.Name, out var current))
+ context.Entry(current).CurrentValues.SetValues(entity);
+ else
+ context.Add(entity);
+ }
+
+ foreach (var name in changes.RemovedScheduleNames)
+ context.Remove(schedules[name]);
+ foreach (var schedule in changes.Schedules)
+ {
+ if (schedules.TryGetValue(schedule.Name, out var current))
+ context.Entry(current).CurrentValues.SetValues(ToEntity(schedule));
+ else
+ context.Add(ToEntity(schedule));
+ }
+
+ await context.SaveChangesAsync(token);
+ await transaction.CommitAsync(token);
+ }, cancellationToken);
+ }
+
+ ///
+ public async ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default)
+ {
+ GetJobDefinitionsAsyncCalled();
+ cancellationToken.ThrowIfCancellationRequested();
+ await TaskScheduler.Yield();
+ return await ReadWithStrategyAsync(async (context, token) =>
+ {
+ var definitions = await context.Set().AsNoTracking().ToListAsync(token);
+ return (IReadOnlyList)[.. definitions.Select(ToRecord)
+ .OrderBy(static definition => definition.Name, StringComparer.OrdinalIgnoreCase)];
+ }, cancellationToken);
+ }
+
+ private async Task EnsureDefinitionCatalogAsync(CancellationToken cancellationToken)
+ {
+ await using var context = await contextFactory.CreateDbContextAsync(cancellationToken);
+ if (await context.Set().AnyAsync(item => item.Id == 1, cancellationToken))
+ return;
+ context.Add(new ImmediateJobDefinitionCatalogEntity { Id = 1, ConcurrencyStamp = Guid.NewGuid() });
+ try
+ {
+ await context.SaveChangesAsync(cancellationToken);
+ }
+ catch (DbUpdateException)
+ {
+ await using var verification = await contextFactory.CreateDbContextAsync(cancellationToken);
+ if (!await verification.Set().AnyAsync(item => item.Id == 1, cancellationToken))
+ throw;
+ }
+ }
+
+ private static RecurringJobSchedule ToDefinitionSchedule(ImmediateRecurringJobEntity entity) => new()
+ {
+ Name = entity.Name,
+ JobName = entity.JobName,
+ QueueName = entity.QueueName,
+ Cron = entity.Cron,
+ TimeZone = entity.TimeZone,
+ IsCodeDefined = entity.IsCodeDefined,
+ IsPaused = entity.IsPaused,
+ NextRunAt = entity.NextRunAt,
+ LastRunAt = entity.LastRunAt,
+ };
+
+ private static ImmediateJobDefinitionMetadataEntity ToEntity(JobDefinitionRecord definition) => new()
+ {
+ Name = definition.Name,
+ QueueName = definition.QueueName,
+ QueuePriority = definition.QueuePriority,
+ QueueConcurrency = definition.QueueConcurrency,
+ Cron = definition.Cron,
+ TimeZone = definition.TimeZone,
+ MaxAttempts = definition.MaxAttempts,
+ TimeoutTicks = definition.Timeout?.Ticks,
+ MaxConcurrency = definition.MaxConcurrency,
+ OverlapPolicy = definition.OverlapPolicy,
+ MisfireHandlingMode = definition.MisfireHandlingMode,
+ Backoff = definition.Backoff,
+ BackoffBaseTicks = definition.BackoffBase.Ticks,
+ };
+
+ private static JobDefinitionRecord ToRecord(ImmediateJobDefinitionMetadataEntity entity) => new()
+ {
+ Name = entity.Name,
+ QueueName = entity.QueueName,
+ QueuePriority = entity.QueuePriority,
+ QueueConcurrency = entity.QueueConcurrency,
+ Cron = entity.Cron,
+ TimeZone = entity.TimeZone,
+ MaxAttempts = entity.MaxAttempts,
+ Timeout = entity.TimeoutTicks is { } ticks ? TimeSpan.FromTicks(ticks) : null,
+ MaxConcurrency = entity.MaxConcurrency,
+ OverlapPolicy = entity.OverlapPolicy,
+ MisfireHandlingMode = entity.MisfireHandlingMode,
+ Backoff = entity.Backoff,
+ BackoffBase = TimeSpan.FromTicks(entity.BackoffBaseTicks),
+ };
+
+ [LoggerMessage(
+ EventId = LibraryEventIds.MergeJobDefinitionsListAsyncCalled,
+ EventName = "Immediate.Jobs.EntityFrameworkCore.MergeJobDefinitionsListAsyncCalled",
+ Level = LogLevel.Debug,
+ Message = "MergeJobDefinitionsListAsyncCalled"
+ )]
+ private partial void MergeJobDefinitionsListAsyncCalled();
+
+ [LoggerMessage(
+ EventId = LibraryEventIds.GetJobDefinitionsAsyncCalled,
+ EventName = "Immediate.Jobs.EntityFrameworkCore.GetJobDefinitionsAsyncCalled",
+ Level = LogLevel.Debug,
+ Message = "GetJobDefinitionsAsyncCalled"
+ )]
+ private partial void GetJobDefinitionsAsyncCalled();
+}
diff --git a/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.Query.cs b/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.Query.cs
index 0aa5e5cc..ef31c2ad 100644
--- a/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.Query.cs
+++ b/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.Query.cs
@@ -87,7 +87,9 @@ public async ValueTask> QueryJobsAsync(JobQuery query,
var search = query.Search.ToUpperInvariant();
// The parameterless form is intentionally used because relational providers translate it to SQL.
#pragma warning disable CA1304, CA1311, CA1862, MA0011
- jobs = jobs.Where(job => job.JobName.ToUpper().Contains(search));
+ jobs = context.Database.ProviderName!.Contains("Npgsql", StringComparison.Ordinal)
+ ? jobs.Where(job => EF.Functions.Collate(job.JobName, "und-x-icu").ToUpper().Contains(search))
+ : jobs.Where(job => job.JobName.ToUpper().Contains(search));
#pragma warning restore CA1304, CA1311, CA1862, MA0011
}
diff --git a/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.Recurring.cs b/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.Recurring.cs
index 6add4f88..b5af6a64 100644
--- a/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.Recurring.cs
+++ b/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.Recurring.cs
@@ -7,61 +7,6 @@ namespace Immediate.Jobs.EntityFrameworkCore;
internal sealed partial class EntityFrameworkCoreJobStorage
where TContext : DbContext
{
- ///
- public async ValueTask MergeRecurringSchedulesListAsync(
- IReadOnlyList schedules,
- CancellationToken cancellationToken = default
- )
- {
- MergeRecurringSchedulesListAsyncCalled();
- cancellationToken.ThrowIfCancellationRequested();
- await TaskScheduler.Yield();
- await using var strategyContext = await contextFactory.CreateDbContextAsync(cancellationToken);
- var strategy = strategyContext.Database.CreateExecutionStrategy();
- await strategy.ExecuteAsync(async operationCancellationToken =>
- {
- await using var context = await contextFactory.CreateDbContextAsync(operationCancellationToken);
- await using var transaction = await context.Database.BeginTransactionAsync(operationCancellationToken);
- var existing = await context.Set()
- .ToDictionaryAsync(schedule => schedule.Name, StringComparer.Ordinal, operationCancellationToken);
-
- foreach (var schedule in schedules)
- {
- if (!existing.Remove(schedule.Name, out var entity))
- {
- _ = context.Add(ToEntity(schedule));
- continue;
- }
-
- entity.NextRunAt =
- string.Equals(entity.Cron, schedule.Cron, StringComparison.Ordinal)
- && string.Equals(entity.TimeZone, schedule.TimeZone, StringComparison.Ordinal)
- ? entity.NextRunAt
- : schedule.NextRunAt;
-
- entity.JobName = schedule.JobName;
- entity.QueueName = schedule.QueueName;
- entity.Cron = schedule.Cron;
- entity.TimeZone = schedule.TimeZone;
- entity.IsCodeDefined = true;
- entity.ConcurrencyStamp = Guid.NewGuid();
- }
-
- if (existing.Count != 0)
- {
- var toRemove = existing
- .Where(kvp => kvp.Value.IsCodeDefined)
- .Select(kvp => kvp.Value)
- .ToList();
-
- context.RemoveRange(toRemove);
- }
-
- await context.SaveChangesAsync(operationCancellationToken);
- await transaction.CommitAsync(operationCancellationToken);
- }, cancellationToken);
- }
-
///
public async ValueTask UpsertRecurringAsync(RecurringJobSchedule schedule, CancellationToken cancellationToken = default)
{
@@ -315,14 +260,6 @@ CancellationToken cancellationToken
}
}
- [LoggerMessage(
- EventId = LibraryEventIds.MergeRecurringSchedulesListAsyncCalled,
- EventName = "Immediate.Jobs.EntityFrameworkCore.MergeRecurringSchedulesListAsyncCalled",
- Level = LogLevel.Debug,
- Message = "MergeRecurringSchedulesListAsync called"
- )]
- private partial void MergeRecurringSchedulesListAsyncCalled();
-
[LoggerMessage(
EventId = LibraryEventIds.UpsertRecurringAsyncCalled,
EventName = "Immediate.Jobs.EntityFrameworkCore.UpsertRecurringAsyncCalled",
diff --git a/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.Trigger.cs b/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.Trigger.cs
index 487cafeb..0b5e73f5 100644
--- a/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.Trigger.cs
+++ b/src/Immediate.Jobs.EntityFrameworkCore/EntityFrameworkCoreJobStorage.Trigger.cs
@@ -124,7 +124,7 @@ CancellationToken cancellationToken
var job = await context.Set()
.SingleOrDefaultAsync(item => item.Id == jobHandle.Value, cancellationToken)
?? throw new KeyNotFoundException($"Job '{jobHandle}' was not found.");
- if (!string.Equals(job.JobName, expectedJobName, StringComparison.Ordinal))
+ if (!string.Equals(job.JobName, expectedJobName, StringComparison.OrdinalIgnoreCase))
throw new ImmediateJobException($"Job '{jobHandle}' is not a '{expectedJobName}' job.");
return job;
diff --git a/src/Immediate.Jobs.EntityFrameworkCore/ImmediateJobsModelBuilderExtensions.cs b/src/Immediate.Jobs.EntityFrameworkCore/ImmediateJobsModelBuilderExtensions.cs
index 28297bff..fc26ddf2 100644
--- a/src/Immediate.Jobs.EntityFrameworkCore/ImmediateJobsModelBuilderExtensions.cs
+++ b/src/Immediate.Jobs.EntityFrameworkCore/ImmediateJobsModelBuilderExtensions.cs
@@ -1,18 +1,32 @@
using Immediate.Jobs.Shared.Apis;
using Immediate.Jobs.Shared.Internals;
using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.ChangeTracking;
using Microsoft.EntityFrameworkCore.Metadata.Builders;
namespace Immediate.Jobs.EntityFrameworkCore;
-/// Adds the Immediate.Jobs persistence model to an application DbContext.
+///
+/// Adds the Immediate.Jobs persistence model to an application DbContext.
+///
public static class ImmediateJobsModelBuilderExtensions
{
- /// Configures the entities required by the Immediate.Jobs EF Core storage provider.
- /// The model builder to configure.
- /// The database schema for the Immediate.Jobs tables, or for the provider default.
- /// The configured model builder.
- public static ModelBuilder AddImmediateJobs(this ModelBuilder modelBuilder, string? schema = null)
+ ///
+ /// Configures the entities required by the Immediate.Jobs EF Core storage provider.
+ ///
+ ///
+ /// The model builder to configure.
+ ///
+ ///
+ /// The database schema for the Immediate.Jobs tables, or for the provider default.
+ ///
+ ///
+ /// An optional case-insensitive collation for definition names; defaults to the supported provider's case-insensitive collation.
+ ///
+ ///
+ /// The configured model builder.
+ ///
+ public static ModelBuilder AddImmediateJobs(this ModelBuilder modelBuilder, string? schema = null, string? definitionNameCollation = null)
{
ArgumentNullException.ThrowIfNull(modelBuilder);
ConfigureBatches(modelBuilder.Entity(), schema);
@@ -22,6 +36,36 @@ public static ModelBuilder AddImmediateJobs(this ModelBuilder modelBuilder, stri
ConfigureContinuations(modelBuilder.Entity(), schema);
ConfigureRecurring(modelBuilder.Entity(), schema);
ConfigureServers(modelBuilder.Entity(), schema);
+ var definitions = modelBuilder.Entity();
+ definitions.ToTable("immediate_job_definition_metadata", schema);
+ definitions.HasKey(definition => definition.Name);
+ var isPostgreSql = modelBuilder.Model.FindAnnotation("Npgsql:ValueGenerationStrategy") is not null;
+ var postgresCollation = schema is null ? "immediate_jobs_case_insensitive" : schema + "_jobs_ci";
+ definitionNameCollation ??= isPostgreSql ? postgresCollation
+ : modelBuilder.Model.FindAnnotation("SqlServer:ValueGenerationStrategy") is not null ? null : "NOCASE";
+ if (isPostgreSql && string.Equals(definitionNameCollation, postgresCollation, StringComparison.Ordinal))
+ {
+ // Npgsql's public collation metadata is stored as this annotation (EF Core 8, 10, and 11).
+ modelBuilder.HasAnnotation("Npgsql:CollationDefinition:" + postgresCollation, "und-u-ks-level2,und-u-ks-level2,icu,False");
+ }
+
+ modelBuilder.Entity().Property(job => job.JobName).UseCollation(definitionNameCollation);
+ modelBuilder.Entity().Property(schedule => schedule.JobName).UseCollation(definitionNameCollation);
+ var name = definitions.Property(definition => definition.Name).HasMaxLength(256).UseCollation(definitionNameCollation);
+ name.Metadata.SetValueComparer(new ValueComparer(
+ (left, right) => string.Equals(left, right, StringComparison.OrdinalIgnoreCase),
+ value => StringComparer.OrdinalIgnoreCase.GetHashCode(value),
+ value => value));
+ definitions.Property(definition => definition.QueueName).HasMaxLength(256).IsRequired();
+ definitions.Property(definition => definition.Cron).HasMaxLength(128);
+ definitions.Property(definition => definition.TimeZone).HasMaxLength(128).IsRequired();
+ definitions.Property(definition => definition.OverlapPolicy).HasConversion();
+ definitions.Property(definition => definition.MisfireHandlingMode).HasConversion();
+ definitions.Property(definition => definition.Backoff).HasConversion();
+ var catalogue = modelBuilder.Entity();
+ catalogue.ToTable("immediate_job_definition_catalog", schema);
+ catalogue.HasKey(item => item.Id);
+ catalogue.Property(item => item.Id).ValueGeneratedNever();
return modelBuilder;
}
@@ -310,3 +354,30 @@ internal sealed class ImmediateJobServerEntity
public int MaxWorkers { get; set; }
public string Details { get; set; } = null!;
}
+
+internal sealed class ImmediateJobDefinitionMetadataEntity
+{
+ public string Name { get; set; } = null!;
+ public string QueueName { get; set; } = null!;
+ public int QueuePriority { get; set; }
+ public int QueueConcurrency { get; set; }
+ public string? Cron { get; set; }
+ public string TimeZone { get; set; } = null!;
+ public int MaxAttempts { get; set; }
+ public long? TimeoutTicks { get; set; }
+ public int MaxConcurrency { get; set; }
+ public OverlapPolicy OverlapPolicy { get; set; }
+ public MisfireHandlingMode MisfireHandlingMode { get; set; }
+ public BackoffStrategy Backoff { get; set; }
+ public long BackoffBaseTicks { get; set; }
+}
+
+///
+/// A singleton row updated at the start of catalogue reconciliation to acquire a database write lock.
+/// The lock serializes complete definition and recurring-schedule snapshots across servers until commit.
+///
+internal sealed class ImmediateJobDefinitionCatalogEntity
+{
+ public int Id { get; set; }
+ public Guid ConcurrencyStamp { get; set; }
+}
diff --git a/src/Immediate.Jobs.EntityFrameworkCore/LibraryEventIds.cs b/src/Immediate.Jobs.EntityFrameworkCore/LibraryEventIds.cs
index 0e44c77a..08852cfc 100644
--- a/src/Immediate.Jobs.EntityFrameworkCore/LibraryEventIds.cs
+++ b/src/Immediate.Jobs.EntityFrameworkCore/LibraryEventIds.cs
@@ -16,7 +16,6 @@ internal static class LibraryEventIds
public const int CompleteWithContinuationsAsyncCalled = 11511;
public const int AddBatchJobAsyncCalled = 11512;
public const int FailAsyncCalled = 11513;
- public const int MergeRecurringSchedulesListAsyncCalled = 11514;
public const int UpsertRecurringAsyncCalled = 11515;
public const int RemoveRecurringAsyncCalled = 11516;
public const int PauseRecurringAsyncCalled = 11517;
@@ -45,4 +44,6 @@ internal static class LibraryEventIds
public const int UpdatePayloadAsyncCalled = 11540;
public const int TryTriggerAsyncCalled = 11541;
public const int TryTriggerBatchAsyncCalled = 11542;
+ public const int MergeJobDefinitionsListAsyncCalled = 11543;
+ public const int GetJobDefinitionsAsyncCalled = 11544;
}
diff --git a/src/Immediate.Jobs.Generators/ImmediateJobsGenerator.Transform.cs b/src/Immediate.Jobs.Generators/ImmediateJobsGenerator.Transform.cs
index 839f3218..bd247328 100644
--- a/src/Immediate.Jobs.Generators/ImmediateJobsGenerator.Transform.cs
+++ b/src/Immediate.Jobs.Generators/ImmediateJobsGenerator.Transform.cs
@@ -31,7 +31,7 @@ CancellationToken cancellationToken
var attribute = context.Attributes[0];
var jobName = attribute.GetJobName(className: symbol.Name);
- if (!jobName.HasNameContent())
+ if (!jobName.HasNameContent() || jobName.HasLeadingOrTrailingWhitespace())
return null;
var arguments = attribute.NamedArguments;
@@ -122,7 +122,6 @@ when queueSymbol.GetAttributes().QueueDefinitionAttribute is { } queueDefinition
return null;
var tags = handlerAttribute?.NamedArguments.GetStringArray("Tags");
-
var extractors = attributes.GetContextExtractors().ToList();
if (extractors.Any(e => !PayloadValidation.CanSerializeToJson(e.ContextType, reportError: null)))
diff --git a/src/Immediate.Jobs.LinqToDB/LibraryEventIds.cs b/src/Immediate.Jobs.LinqToDB/LibraryEventIds.cs
index 7fb539f5..82708f72 100644
--- a/src/Immediate.Jobs.LinqToDB/LibraryEventIds.cs
+++ b/src/Immediate.Jobs.LinqToDB/LibraryEventIds.cs
@@ -17,7 +17,6 @@ internal static class LibraryEventIds
public const int CompleteWithContinuationsAsyncCalled = 11612;
public const int AddBatchJobAsyncCalled = 11613;
public const int FailAsyncCalled = 11614;
- public const int MergeRecurringSchedulesListAsyncCalled = 11615;
public const int UpsertRecurringAsyncCalled = 11616;
public const int RemoveRecurringAsyncCalled = 11617;
public const int PauseRecurringAsyncCalled = 11618;
@@ -45,4 +44,6 @@ internal static class LibraryEventIds
public const int UpdatePayloadAsyncCalled = 11640;
public const int TryTriggerAsyncCalled = 11641;
public const int TryTriggerBatchAsyncCalled = 11642;
+ public const int MergeJobDefinitionsListAsyncCalled = 11643;
+ public const int GetJobDefinitionsAsyncCalled = 11644;
}
diff --git a/src/Immediate.Jobs.LinqToDB/LinqToDBEntities.cs b/src/Immediate.Jobs.LinqToDB/LinqToDBEntities.cs
index 4be7261c..45b0ac04 100644
--- a/src/Immediate.Jobs.LinqToDB/LinqToDBEntities.cs
+++ b/src/Immediate.Jobs.LinqToDB/LinqToDBEntities.cs
@@ -56,6 +56,8 @@ internal sealed class ImmediateJobEntity
[Column(Length = 256, CanBeNull = false)]
public string QueueName { get; set; } = JobQueueDefinition.DefaultName;
[Column(Length = 256, CanBeNull = false)]
+ [Column(Configuration = ProviderName.PostgreSQL, DbType = "varchar(256) COLLATE \"immediate_jobs_case_insensitive\"", CanBeNull = false)]
+ [Column(Configuration = ProviderName.SQLite, DbType = "TEXT COLLATE NOCASE", CanBeNull = false)]
public string JobName { get; set; } = null!;
[Column(Length = 128, CanBeNull = true)]
public string? GroupId { get; set; }
@@ -164,6 +166,8 @@ internal sealed class ImmediateRecurringJobEntity
[PrimaryKey, Column(Length = 256, CanBeNull = false)]
public string Name { get; set; } = null!;
[Column(Length = 256, CanBeNull = false)]
+ [Column(Configuration = ProviderName.PostgreSQL, DbType = "varchar(256) COLLATE \"immediate_jobs_case_insensitive\"", CanBeNull = false)]
+ [Column(Configuration = ProviderName.SQLite, DbType = "TEXT COLLATE NOCASE", CanBeNull = false)]
public string JobName { get; set; } = null!;
[Column(Length = 256, CanBeNull = false)]
public string QueueName { get; set; } = null!;
@@ -199,3 +203,36 @@ internal sealed class ImmediateJobServerEntity
[Column(DataType = DataType.Text)]
public string Details { get; set; } = null!;
}
+
+[Table(Name = "immediate_job_definition_metadata")]
+internal sealed class ImmediateJobDefinitionMetadataEntity
+{
+ [PrimaryKey, Column(Length = 256, CanBeNull = false)]
+ [Column(Configuration = ProviderName.PostgreSQL, DbType = "varchar(256) COLLATE \"immediate_jobs_case_insensitive\"", CanBeNull = false)]
+ [Column(Configuration = ProviderName.SQLite, DbType = "TEXT COLLATE NOCASE", CanBeNull = false)]
+ public string Name { get; set; } = null!;
+ [Column(Length = 256, CanBeNull = false)]
+ public string QueueName { get; set; } = null!;
+ [Column(DataType = DataType.Int32)]
+ public int QueuePriority { get; set; }
+ [Column(DataType = DataType.Int32)]
+ public int QueueConcurrency { get; set; }
+ [Column(Length = 128, CanBeNull = true)]
+ public string? Cron { get; set; }
+ [Column(Length = 128, CanBeNull = false)]
+ public string TimeZone { get; set; } = null!;
+ [Column(DataType = DataType.Int32)]
+ public int MaxAttempts { get; set; }
+ [Column(DataType = DataType.Int64, CanBeNull = true)]
+ public long? TimeoutTicks { get; set; }
+ [Column(DataType = DataType.Int32)]
+ public int MaxConcurrency { get; set; }
+ [Column(DataType = DataType.Int16)]
+ public OverlapPolicy OverlapPolicy { get; set; }
+ [Column(DataType = DataType.Int16)]
+ public MisfireHandlingMode MisfireHandlingMode { get; set; }
+ [Column(DataType = DataType.Int16)]
+ public BackoffStrategy Backoff { get; set; }
+ [Column(DataType = DataType.Int64)]
+ public long BackoffBaseTicks { get; set; }
+}
diff --git a/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Acquire.cs b/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Acquire.cs
index 9b5f2b90..6ea387d8 100644
--- a/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Acquire.cs
+++ b/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Acquire.cs
@@ -31,7 +31,7 @@ public async ValueTask> AcquireDueJobsAsync(
if (queueCapacity <= 0)
continue;
- var jobCapacities = queue.JobCapacities.ToDictionary(static pair => pair.Key, static pair => pair.Value, StringComparer.Ordinal);
+ var jobCapacities = queue.JobCapacities.ToDictionary(static pair => pair.Key, static pair => pair.Value, StringComparer.OrdinalIgnoreCase);
while (queueCapacity > 0)
{
var eligibleNames = jobCapacities.Where(static pair => pair.Value > 0).Select(static pair => pair.Key).ToList();
@@ -52,7 +52,7 @@ public async ValueTask> AcquireDueJobsAsync(
if (candidates.Count == 0)
break;
- var selectionCapacities = new Dictionary(jobCapacities, StringComparer.Ordinal);
+ var selectionCapacities = new Dictionary(jobCapacities, StringComparer.OrdinalIgnoreCase);
var selected = candidates.Where(candidate => selectionCapacities[candidate.JobName]-- > 0).ToList();
var claimed = await AcquireCandidatesAsync(selected, request.WorkerId, request.Lease, now, cancellationToken);
foreach (var job in claimed)
diff --git a/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Concurrency.cs b/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Concurrency.cs
index a248b0f7..c7ad6787 100644
--- a/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Concurrency.cs
+++ b/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Concurrency.cs
@@ -1,9 +1,11 @@
+using System.Data;
using System.Data.Common;
using Immediate.Jobs.Shared.Apis;
using Immediate.Jobs.Shared.Storage;
using LinqToDB;
using LinqToDB.Async;
using LinqToDB.Data;
+using LinqToDB.DataProvider.SqlServer;
namespace Immediate.Jobs.LinqToDB;
@@ -107,7 +109,8 @@ await PropagateTerminalAsync(
private async ValueTask RetryConcurrencyAsync(
Func operation,
CancellationToken cancellationToken,
- int maxAttempts = MaxConcurrencyAttempts
+ int maxAttempts = MaxConcurrencyAttempts,
+ IsolationLevel isolationLevel = IsolationLevel.Unspecified
)
{
var concurrencyAttempt = 0;
@@ -115,7 +118,10 @@ private async ValueTask RetryConcurrencyAsync(
{
await using var scope = contextScope.GetScope(out var connection);
- _ = await connection.BeginTransactionAsync(cancellationToken);
+ if (isolationLevel == IsolationLevel.Unspecified)
+ await connection.BeginTransactionAsync(cancellationToken);
+ else
+ await connection.BeginTransactionAsync(isolationLevel, cancellationToken);
try
{
await operation(connection);
@@ -137,6 +143,18 @@ private async ValueTask RetryConcurrencyAsync(
await connection.RollbackTransactionAsync(cancellationToken);
await DelayConcurrencyRetryAsync(cancellationToken);
}
+ catch (DbException exception) when (
+ isolationLevel == IsolationLevel.Serializable
+ && (exception.SqlState is "40001" or "40P01" || SqlServerTransientExceptionDetector.ShouldRetryOn(exception))
+ && ++concurrencyAttempt < maxAttempts
+ )
+ {
+ // A failed serializable commit may already have completed the underlying transaction.
+ if (connection.Transaction?.Connection is not null)
+ await connection.RollbackTransactionAsync(cancellationToken);
+ // Serializable catalogues can deadlock or fail validation; retry the entire snapshot.
+ await DelayConcurrencyRetryAsync(cancellationToken);
+ }
catch (SyntheticExecutionInsertFailedException exception)
{
await connection.RollbackTransactionAsync(cancellationToken);
@@ -160,7 +178,8 @@ private async ValueTask RetryConcurrencyAsync(
}
catch
{
- await connection.RollbackTransactionAsync(cancellationToken);
+ if (connection.Transaction?.Connection is not null)
+ await connection.RollbackTransactionAsync(cancellationToken);
throw;
}
}
diff --git a/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Database.cs b/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Database.cs
index 56249933..c009bacc 100644
--- a/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Database.cs
+++ b/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Database.cs
@@ -27,6 +27,9 @@ private ITable Recurring(DataConnection connection)
private ITable Servers(DataConnection connection) =>
WithSchema(connection.GetTable());
+ private ITable Definitions(DataConnection connection) =>
+ WithSchema(connection.GetTable());
+
private ITable WithSchema(ITable table)
where TTable : notnull => _schema is null ? table : table.SchemaName(_schema);
diff --git a/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.DefinitionCatalog.cs b/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.DefinitionCatalog.cs
new file mode 100644
index 00000000..d87509d3
--- /dev/null
+++ b/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.DefinitionCatalog.cs
@@ -0,0 +1,233 @@
+using System.Data;
+using Immediate.Jobs.Shared.Apis;
+using Immediate.Jobs.Shared.Storage;
+using LinqToDB;
+using LinqToDB.Async;
+using LinqToDB.Data;
+using Microsoft.Extensions.Logging;
+
+namespace Immediate.Jobs.LinqToDB;
+
+internal sealed partial class LinqToDBJobStorage
+ where T : DataConnection
+{
+ ///
+ public async ValueTask MergeJobDefinitionsListAsync(JobDefinitionRegistration registration, CancellationToken cancellationToken = default)
+ {
+ MergeJobDefinitionsListAsyncCalled();
+ cancellationToken.ThrowIfCancellationRequested();
+ await TaskScheduler.Yield();
+ ValidateRegistration(registration);
+ await RetryConcurrencyAsync(async connection =>
+ {
+ if (connection.DataProvider.Name.Contains("SQLite", StringComparison.OrdinalIgnoreCase))
+ {
+ await MergeSQLiteDefinitionsAsync(connection, registration, cancellationToken);
+ return;
+ }
+
+ await MergeDefinitionMetadataAsync(connection, registration.Definitions, cancellationToken);
+ await MergeRecurringDefinitionsAsync(connection, registration.RecurringSchedules, cancellationToken);
+ }, cancellationToken, isolationLevel: IsolationLevel.Serializable);
+ }
+
+ ///
+ public async ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default)
+ {
+ GetJobDefinitionsAsyncCalled();
+ cancellationToken.ThrowIfCancellationRequested();
+ await TaskScheduler.Yield();
+ await using var scope = contextScope.GetScope(out var connection);
+ var definitions = await Definitions(connection).ToListAsync(cancellationToken);
+ return [.. definitions.Select(static item => ToRecord(item))
+ .OrderBy(static definition => definition.Name, StringComparer.OrdinalIgnoreCase)];
+ }
+
+ private static void ValidateRegistration(JobDefinitionRegistration registration)
+ {
+ ArgumentNullException.ThrowIfNull(registration);
+ ArgumentNullException.ThrowIfNull(registration.Definitions, nameof(registration));
+ ArgumentNullException.ThrowIfNull(registration.RecurringSchedules, nameof(registration));
+ var definitions = registration.Definitions.ToDictionary(static definition => definition.Name, StringComparer.OrdinalIgnoreCase);
+ foreach (var definition in definitions.Values)
+ {
+ ArgumentException.ThrowIfNullOrWhiteSpace(definition.Name, nameof(registration));
+ if (char.IsWhiteSpace(definition.Name[0]) || char.IsWhiteSpace(definition.Name[^1]))
+ throw new ArgumentException("Job definition names must not have leading or trailing whitespace.", nameof(registration));
+ }
+
+ var names = new HashSet(StringComparer.OrdinalIgnoreCase);
+ foreach (var schedule in registration.RecurringSchedules)
+ {
+ if (!definitions.TryGetValue(schedule.JobName, out var definition)
+ || !schedule.IsCodeDefined
+ || !string.Equals(schedule.Name, definition.Name, StringComparison.OrdinalIgnoreCase)
+ || !string.Equals(schedule.Cron, definition.Cron, StringComparison.Ordinal)
+ || !string.Equals(schedule.TimeZone, definition.TimeZone, StringComparison.Ordinal)
+ || !string.Equals(schedule.QueueName, definition.QueueName, StringComparison.Ordinal)
+ || !names.Add(schedule.Name))
+ throw new ArgumentException("Code-defined schedules must match their supplied job definitions.", nameof(registration));
+ }
+
+ if (definitions.Values.Any(definition => definition.Cron is not null && !names.Contains(definition.Name)))
+ throw new ArgumentException("Every managed cron definition must supply its code-defined schedule.", nameof(registration));
+ }
+
+ private async Task MergeSQLiteDefinitionsAsync(
+ DataConnection connection,
+ JobDefinitionRegistration registration,
+ CancellationToken cancellationToken
+ )
+ {
+ // SQLite has no SQL MERGE, so reconcile its stored snapshot in the client.
+ var definitions = await Definitions(connection).ToListAsync(cancellationToken);
+ var schedules = await Recurring(connection).ToListAsync(cancellationToken);
+ var changes = JobDefinitionReconciliation.Create(registration,
+ definitions.Select(ToRecord).ToList(),
+ schedules.Select(ToRecord).ToList());
+ await Definitions(connection).Where(item => item.Name.In(changes.RemovedDefinitionNames)).DeleteAsync(cancellationToken);
+ foreach (var definition in changes.Definitions)
+ await connection.InsertOrReplaceAsync(ToEntity(definition), schemaName: _schema, token: cancellationToken);
+
+ await Recurring(connection).Where(item => item.Name.In(changes.RemovedScheduleNames)).DeleteAsync(cancellationToken);
+ foreach (var schedule in changes.Schedules)
+ await connection.InsertOrReplaceAsync(ToEntity(schedule), schemaName: _schema, token: cancellationToken);
+ }
+
+ private async Task MergeDefinitionMetadataAsync(
+ DataConnection connection,
+ IReadOnlyList definitions,
+ CancellationToken cancellationToken
+ )
+ {
+ var supplied = definitions.Select(ToEntity).ToList().AsQueryable(connection);
+ var merge = Definitions(connection).Merge()
+ .Using(supplied)
+ .On(static (dst, src) => dst.Name == src.Name)
+ .UpdateWhenMatched(static (dst, src) => new ImmediateJobDefinitionMetadataEntity
+ {
+ QueueName = src.QueueName,
+ QueuePriority = src.QueuePriority,
+ QueueConcurrency = src.QueueConcurrency,
+ Cron = src.Cron,
+ TimeZone = src.TimeZone,
+ MaxAttempts = src.MaxAttempts,
+ TimeoutTicks = src.TimeoutTicks,
+ MaxConcurrency = src.MaxConcurrency,
+ OverlapPolicy = src.OverlapPolicy,
+ MisfireHandlingMode = src.MisfireHandlingMode,
+ Backoff = src.Backoff,
+ BackoffBaseTicks = src.BackoffBaseTicks,
+ })
+ .InsertWhenNotMatched();
+ if (connection.DataProvider.Name.Contains("SqlServer", StringComparison.OrdinalIgnoreCase))
+ await merge.DeleteWhenNotMatchedBySource().MergeAsync(cancellationToken);
+ else
+ {
+ await merge.MergeAsync(cancellationToken);
+ // PostgreSQL 15 has no WHEN NOT MATCHED BY SOURCE; remove obsolete rows in the same transaction.
+ await Definitions(connection).Where(dst => !supplied.Any(src => dst.Name == src.Name)).DeleteAsync(cancellationToken);
+ }
+ }
+
+ private async Task MergeRecurringDefinitionsAsync(
+ DataConnection connection,
+ IReadOnlyList schedules,
+ CancellationToken cancellationToken
+ )
+ {
+ var incoming = schedules.Select(ToEntity).ToList().AsQueryable(connection);
+ var supplied = incoming.Join(Definitions(connection),
+ static schedule => schedule.JobName,
+ static definition => definition.Name,
+ static (schedule, definition) => new ImmediateRecurringJobEntity
+ {
+ Name = definition.Name,
+ JobName = definition.Name,
+ QueueName = schedule.QueueName,
+ Cron = schedule.Cron,
+ TimeZone = schedule.TimeZone,
+ IsCodeDefined = true,
+ IsPaused = schedule.IsPaused,
+ NextRunAt = schedule.NextRunAt,
+ LastRunAt = schedule.LastRunAt,
+ ConcurrencyStamp = schedule.ConcurrencyStamp,
+ });
+ var merge = Recurring(connection).Merge()
+ .Using(supplied)
+ .On(static (dst, src) => dst.Name == src.Name)
+ .UpdateWhenMatched(static (dst, src) => new ImmediateRecurringJobEntity
+ {
+ JobName = src.JobName,
+ QueueName = src.QueueName,
+ Cron = src.Cron,
+ TimeZone = src.TimeZone,
+ IsCodeDefined = true,
+ NextRunAt = EqualsOrdinal(dst.Cron, src.Cron) && EqualsOrdinal(dst.TimeZone, src.TimeZone)
+ ? dst.NextRunAt : src.NextRunAt,
+ ConcurrencyStamp = src.ConcurrencyStamp,
+ })
+ .InsertWhenNotMatched();
+ if (connection.DataProvider.Name.Contains("SqlServer", StringComparison.OrdinalIgnoreCase))
+ await merge.DeleteWhenNotMatchedBySourceAnd(static dst => dst.IsCodeDefined).MergeAsync(cancellationToken);
+ else
+ {
+ await merge.MergeAsync(cancellationToken);
+ await Recurring(connection).Where(dst => dst.IsCodeDefined && !supplied.Any(src => dst.Name == src.Name)).DeleteAsync(cancellationToken);
+ }
+ }
+
+ [Sql.Expression(ProviderName.SqlServer, "({0} COLLATE Latin1_General_100_BIN2 = {1} COLLATE Latin1_General_100_BIN2 AND DATALENGTH({0}) = DATALENGTH({1}))", ServerSideOnly = true, IsPredicate = true)]
+ [Sql.Expression(ProviderName.PostgreSQL, "{0} = {1}", 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()
+ {
+ Name = definition.Name,
+ QueueName = definition.QueueName,
+ QueuePriority = definition.QueuePriority,
+ QueueConcurrency = definition.QueueConcurrency,
+ Cron = definition.Cron,
+ TimeZone = definition.TimeZone,
+ MaxAttempts = definition.MaxAttempts,
+ TimeoutTicks = definition.Timeout?.Ticks,
+ MaxConcurrency = definition.MaxConcurrency,
+ OverlapPolicy = definition.OverlapPolicy,
+ MisfireHandlingMode = definition.MisfireHandlingMode,
+ Backoff = definition.Backoff,
+ BackoffBaseTicks = definition.BackoffBase.Ticks,
+ };
+
+ private static JobDefinitionRecord ToRecord(ImmediateJobDefinitionMetadataEntity entity) => new()
+ {
+ Name = entity.Name,
+ QueueName = entity.QueueName,
+ QueuePriority = entity.QueuePriority,
+ QueueConcurrency = entity.QueueConcurrency,
+ Cron = entity.Cron,
+ TimeZone = entity.TimeZone,
+ MaxAttempts = entity.MaxAttempts,
+ Timeout = entity.TimeoutTicks is { } ticks ? TimeSpan.FromTicks(ticks) : null,
+ MaxConcurrency = entity.MaxConcurrency,
+ OverlapPolicy = entity.OverlapPolicy,
+ MisfireHandlingMode = entity.MisfireHandlingMode,
+ Backoff = entity.Backoff,
+ BackoffBase = TimeSpan.FromTicks(entity.BackoffBaseTicks),
+ };
+
+ [LoggerMessage(
+ EventId = LibraryEventIds.MergeJobDefinitionsListAsyncCalled,
+ EventName = "Immediate.Jobs.LinqToDB.MergeJobDefinitionsListAsyncCalled",
+ Level = LogLevel.Debug,
+ Message = "MergeJobDefinitionsListAsyncCalled"
+ )]
+ private partial void MergeJobDefinitionsListAsyncCalled();
+
+ [LoggerMessage(
+ EventId = LibraryEventIds.GetJobDefinitionsAsyncCalled,
+ EventName = "Immediate.Jobs.LinqToDB.GetJobDefinitionsAsyncCalled",
+ Level = LogLevel.Debug,
+ Message = "GetJobDefinitionsAsyncCalled"
+ )]
+ private partial void GetJobDefinitionsAsyncCalled();
+}
diff --git a/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Query.cs b/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Query.cs
index 12a5af73..a99a30e5 100644
--- a/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Query.cs
+++ b/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Query.cs
@@ -102,7 +102,9 @@ public async ValueTask> QueryJobsAsync(
{
var search = query.Search.ToUpperInvariant();
#pragma warning disable CA1304, CA1311, CA1862, MA0011
- jobs = jobs.Where(job => job.JobName.ToUpper().Contains(search));
+ jobs = connection.DataProvider.Name.Contains("PostgreSQL", StringComparison.OrdinalIgnoreCase)
+ ? jobs.Where(job => Sql.Collate(job.JobName, "und-x-icu").ToUpper().Contains(search))
+ : jobs.Where(job => job.JobName.ToUpper().Contains(search));
#pragma warning restore CA1304, CA1311, CA1862, MA0011
}
diff --git a/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Recurring.cs b/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Recurring.cs
index 55f7dbfb..00e9ff2e 100644
--- a/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Recurring.cs
+++ b/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Recurring.cs
@@ -10,68 +10,6 @@ namespace Immediate.Jobs.LinqToDB;
internal sealed partial class LinqToDBJobStorage
where T : DataConnection
{
- ///
- public async ValueTask MergeRecurringSchedulesListAsync(
- IReadOnlyList schedules,
- CancellationToken cancellationToken = default
- )
- {
- MergeRecurringSchedulesListAsyncCalled();
- cancellationToken.ThrowIfCancellationRequested();
- await TaskScheduler.Yield();
-
- await using var scope = contextScope.GetScope(out var connection);
- await using var transaction = await connection.BeginTransactionAsync(cancellationToken);
-
- var existing = await Recurring(connection)
- .ToDictionaryAsync(r => r.Name, StringComparer.Ordinal, cancellationToken);
-
- foreach (var schedule in schedules)
- {
- var entity = ToEntity(schedule);
-
- if (!existing.TryGetValue(schedule.Name, out var current))
- {
- await connection.InsertAsync(entity, schemaName: _schema, token: cancellationToken);
- continue;
- }
-
- existing.Remove(schedule.Name);
-
- var oldStamp = current.ConcurrencyStamp;
-
- current.NextRunAt =
- string.Equals(current.Cron, schedule.Cron, StringComparison.Ordinal)
- && string.Equals(current.TimeZone, schedule.TimeZone, StringComparison.Ordinal)
- ? current.NextRunAt
- : schedule.NextRunAt;
-
- current.JobName = schedule.JobName;
- current.QueueName = schedule.QueueName;
- current.Cron = schedule.Cron;
- current.TimeZone = schedule.TimeZone;
- current.IsCodeDefined = true;
- current.ConcurrencyStamp = Guid.NewGuid();
-
- if (!await UpdateRecurringAsync(connection, current, oldStamp, cancellationToken))
- throw new ImmediateJobException("Failure saving updated schedule.");
- }
-
- if (existing.Count != 0)
- {
- var toRemove = existing
- .Where(kvp => kvp.Value.IsCodeDefined)
- .Select(kvp => kvp.Key)
- .ToList();
-
- await Recurring(connection)
- .Where(r => r.Name.In(toRemove))
- .DeleteAsync(cancellationToken);
- }
-
- await transaction.CommitAsync(cancellationToken);
- }
-
///
public async ValueTask UpsertRecurringAsync(
RecurringJobSchedule schedule,
@@ -314,14 +252,6 @@ CancellationToken cancellationToken
return updated != 0;
}
- [LoggerMessage(
- EventId = LibraryEventIds.MergeRecurringSchedulesListAsyncCalled,
- EventName = "Immediate.Jobs.LinqToDB.MergeRecurringSchedulesListAsyncCalled",
- Level = LogLevel.Debug,
- Message = "MergeRecurringSchedulesListAsync called"
- )]
- private partial void MergeRecurringSchedulesListAsyncCalled();
-
[LoggerMessage(
EventId = LibraryEventIds.UpsertRecurringAsyncCalled,
EventName = "Immediate.Jobs.LinqToDB.UpsertRecurringAsyncCalled",
diff --git a/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Trigger.cs b/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Trigger.cs
index ae956c4a..8532465f 100644
--- a/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Trigger.cs
+++ b/src/Immediate.Jobs.LinqToDB/LinqToDBJobStorage.Trigger.cs
@@ -131,7 +131,7 @@ CancellationToken cancellationToken
{
var job = await Jobs(connection).SingleOrDefaultAsync(item => item.Id == jobHandle.Value, cancellationToken)
?? throw new KeyNotFoundException($"Job '{jobHandle}' was not found.");
- if (!string.Equals(job.JobName, expectedJobName, StringComparison.Ordinal))
+ if (!string.Equals(job.JobName, expectedJobName, StringComparison.OrdinalIgnoreCase))
throw new ImmediateJobException($"Job '{jobHandle}' is not a '{expectedJobName}' job.");
return job;
diff --git a/src/Immediate.Jobs.LinqToDB/LinqToDBSchemaExtensions.cs b/src/Immediate.Jobs.LinqToDB/LinqToDBSchemaExtensions.cs
index 2bdbac6a..c2b445de 100644
--- a/src/Immediate.Jobs.LinqToDB/LinqToDBSchemaExtensions.cs
+++ b/src/Immediate.Jobs.LinqToDB/LinqToDBSchemaExtensions.cs
@@ -3,15 +3,29 @@
namespace Immediate.Jobs.LinqToDB;
-/// Explicit bootstrap helpers for a fresh Immediate.Jobs schema.
+///
+/// Explicit bootstrap helpers for a fresh Immediate.Jobs schema.
+///
public static class LinqToDBSchemaExtensions
{
- /// Creates the Immediate.Jobs tables and indexes when they do not already exist.
- /// This helper bootstraps fresh storage only; it does not perform production schema upgrades.
- /// A LinqToDB Data Connection.
- /// The database schema to create objects in, or for the provider default.
- /// A token that can cancel the operation.
- /// A task that represents the asynchronous schema creation operation.
+ ///
+ /// Creates the Immediate.Jobs tables and indexes when they do not already exist.
+ ///
+ ///
+ /// This helper bootstraps fresh storage only; it does not perform production schema upgrades.
+ ///
+ ///
+ /// A LinqToDB Data Connection.
+ ///
+ ///
+ /// The database schema to create objects in, or for the provider default.
+ ///
+ ///
+ /// A token that can cancel the operation.
+ ///
+ ///
+ /// A task that represents the asynchronous schema creation operation.
+ ///
public static async Task CreateImmediateJobsSchemaAsync(
this TContext context,
string? schema = null,
@@ -30,6 +44,13 @@ public static async Task CreateImmediateJobsSchemaAsync(
if (schema is not null)
await CreateSchemaAsync(context, provider, schema, cancellationToken);
+ await CreateDefinitionNameCollationAsync(context, provider, cancellationToken);
+ await context.CreateTableAsync(
+ schemaName: schema,
+ tableOptions: TableOptions.CreateIfNotExists,
+ token: cancellationToken
+ );
+
await context.CreateTableAsync(
schemaName: schema,
tableOptions: TableOptions.CreateIfNotExists,
@@ -88,6 +109,32 @@ internal static void ValidateSchema(string? schema)
}
}
+ private static Task CreateDefinitionNameCollationAsync(
+ DataConnection connection,
+ string provider,
+ CancellationToken cancellationToken
+ )
+ {
+ if (provider.Contains("PostgreSQL", StringComparison.OrdinalIgnoreCase))
+ {
+ return connection.ExecuteAsync("""
+ DO $collation$
+ BEGIN
+ IF NOT EXISTS (SELECT 1 FROM pg_collation WHERE collname = 'immediate_jobs_case_insensitive') THEN
+ CREATE COLLATION "immediate_jobs_case_insensitive" (provider = icu, locale = 'und-u-ks-level2', deterministic = false);
+ END IF;
+ EXCEPTION WHEN duplicate_object THEN NULL;
+ END $collation$;
+ """, cancellationToken);
+ }
+
+ if (provider.Contains("SqlServer", StringComparison.OrdinalIgnoreCase)
+ || provider.Contains("SQLite", StringComparison.OrdinalIgnoreCase))
+ return Task.CompletedTask;
+
+ throw new NotSupportedException($"Immediate.Jobs schema bootstrap does not support provider '{provider}'.");
+ }
+
private static Task CreateSchemaAsync(
DataConnection connection,
string provider,
diff --git a/src/Immediate.Jobs.LinqToDB/readme.md b/src/Immediate.Jobs.LinqToDB/readme.md
index 21f97a89..11d8cb51 100644
--- a/src/Immediate.Jobs.LinqToDB/readme.md
+++ b/src/Immediate.Jobs.LinqToDB/readme.md
@@ -8,6 +8,10 @@ LinqToDB storage for [Immediate.Jobs](https://www.nuget.org/packages/Immediate.J
SQLite, and SQL Server. The adapter supports durable jobs, recurring schedules, batches, continuations, execution
history, multiple worker processes, and fair queues between tenant groups.
+PostgreSQL storage requires PostgreSQL 15 or later. Definition reconciliation uses SQL `MERGE`;
+configure LinqToDB with a PostgreSQL 15+ dialect. The adapter assumes this requirement is met and
+does not validate the PostgreSQL version at runtime.
+
## Installation
Install the core scheduler, this adapter, and the ADO.NET driver for your database:
diff --git a/src/Immediate.Jobs.Redis/LibraryEventIds.cs b/src/Immediate.Jobs.Redis/LibraryEventIds.cs
index 207426eb..363f4865 100644
--- a/src/Immediate.Jobs.Redis/LibraryEventIds.cs
+++ b/src/Immediate.Jobs.Redis/LibraryEventIds.cs
@@ -20,7 +20,6 @@ internal static class LibraryEventIds
public const int PurgeJobsAsyncCalled = 11715;
public const int HeartbeatAsyncCalled = 11716;
public const int IsHealthyAsyncCalled = 11717;
- public const int MergeRecurringSchedulesListAsyncCalled = 11718;
public const int UpsertRecurringAsyncCalled = 11719;
public const int RemoveRecurringAsyncCalled = 11720;
public const int PauseRecurringAsyncCalled = 11721;
@@ -31,4 +30,6 @@ internal static class LibraryEventIds
public const int MaterializeRecurringAsyncCalledWithDependencies = 11726;
public const int UpdatePayloadAsyncCalled = 11727;
public const int TryTriggerAsyncCalled = 11728;
+ public const int MergeJobDefinitionsListAsyncCalled = 11729;
+ public const int GetJobDefinitionsAsyncCalled = 11730;
}
diff --git a/src/Immediate.Jobs.Redis/RedisJobStorage.Acquire.cs b/src/Immediate.Jobs.Redis/RedisJobStorage.Acquire.cs
index c6cc6618..38ceb6ef 100644
--- a/src/Immediate.Jobs.Redis/RedisJobStorage.Acquire.cs
+++ b/src/Immediate.Jobs.Redis/RedisJobStorage.Acquire.cs
@@ -49,7 +49,7 @@ public async ValueTask> AcquireDueJobsAsync(
values.Add(queue.JobCapacities.Count);
foreach (var capacity in queue.JobCapacities)
{
- values.Add(capacity.Key);
+ values.Add(capacity.Key.ToUpperInvariant());
values.Add(Math.Max(0, capacity.Value));
}
}
diff --git a/src/Immediate.Jobs.Redis/RedisJobStorage.DefinitionCatalog.cs b/src/Immediate.Jobs.Redis/RedisJobStorage.DefinitionCatalog.cs
new file mode 100644
index 00000000..d940f349
--- /dev/null
+++ b/src/Immediate.Jobs.Redis/RedisJobStorage.DefinitionCatalog.cs
@@ -0,0 +1,109 @@
+using System.Text.Json;
+using Immediate.Jobs.Shared.Apis;
+using Immediate.Jobs.Shared.Storage;
+using Microsoft.Extensions.Logging;
+
+namespace Immediate.Jobs.Redis;
+
+internal sealed partial class RedisJobStorage
+{
+ ///
+ public async ValueTask MergeJobDefinitionsListAsync(JobDefinitionRegistration registration, CancellationToken cancellationToken = default)
+ {
+ MergeJobDefinitionsListAsyncCalled();
+ cancellationToken.ThrowIfCancellationRequested();
+ await TaskScheduler.Yield();
+ JobDefinitionReconciliation.Create(registration, [], []);
+ while (true)
+ {
+ cancellationToken.ThrowIfCancellationRequested();
+ var version = await Database.StringGetAsync(DefinitionCatalogVersionKey).WaitAsync(cancellationToken);
+ var definitions = await GetJobDefinitionsAsync(cancellationToken);
+ var schedules = await ReadAllRecurringAsync(cancellationToken);
+ var changes = JobDefinitionReconciliation.Create(registration, definitions, schedules);
+ var payload = new RedisDefinitionCatalogChanges
+ {
+ Definitions = changes.Definitions.Select(static definition => new RedisDefinitionMetadata
+ {
+ Name = definition.Name.ToUpperInvariant(),
+ Metadata = JsonSerializer.Serialize(definition, RedisJsonSerializerContext.Default.JobDefinitionRecord),
+ }).ToList(),
+ RemovedDefinitions = changes.RemovedDefinitionNames.Select(static name => name.ToUpperInvariant()).ToList(),
+ RemovedSchedules = changes.RemovedScheduleNames,
+ Schedules = changes.Schedules.Select(static schedule => new RedisDefinitionSchedule
+ {
+ Name = schedule.Name,
+ Record = JsonSerializer.Serialize(schedule, RedisJsonSerializerContext.Default.RecurringJobSchedule),
+ Cron = schedule.Cron,
+ TimeZone = schedule.TimeZone,
+ Next = Ticks(schedule.NextRunAt),
+ Last = NullableTicks(schedule.LastRunAt),
+ DueScore = Score(schedule.NextRunAt),
+ DueMember = RecurringDueMember(schedule.NextRunAt, schedule.Name),
+ Paused = schedule.IsPaused ? "1" : "0",
+ }).ToList(),
+ };
+ var result = await EvaluateInt64Async(RedisScripts.MergeDefinitionCatalog,
+ [DefinitionMetadataKey, DefinitionCatalogVersionKey, RecurringNamesKey, RecurringDueKey],
+ [version.IsNull ? "0" : version, _root, JsonSerializer.Serialize(payload, RedisJsonSerializerContext.Default.RedisDefinitionCatalogChanges)],
+ cancellationToken);
+ if (result == 1)
+ return;
+ // A competing startup committed. Recompute the entire scoped change from its new snapshot.
+ }
+ }
+
+ ///
+ 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)!)
+ .OrderBy(static definition => definition.Name, StringComparer.OrdinalIgnoreCase)];
+ }
+
+ [LoggerMessage(
+ EventId = LibraryEventIds.MergeJobDefinitionsListAsyncCalled,
+ EventName = "Immediate.Jobs.Redis.MergeJobDefinitionsListAsyncCalled",
+ Level = LogLevel.Debug,
+ Message = "MergeJobDefinitionsListAsyncCalled"
+ )]
+ private partial void MergeJobDefinitionsListAsyncCalled();
+
+ [LoggerMessage(
+ EventId = LibraryEventIds.GetJobDefinitionsAsyncCalled,
+ EventName = "Immediate.Jobs.Redis.GetJobDefinitionsAsyncCalled",
+ Level = LogLevel.Debug,
+ Message = "GetJobDefinitionsAsyncCalled"
+ )]
+ private partial void GetJobDefinitionsAsyncCalled();
+}
+
+internal sealed record RedisDefinitionMetadata
+{
+ public required string Name { get; init; }
+ public required string Metadata { get; init; }
+}
+
+internal sealed record RedisDefinitionSchedule
+{
+ public required string Name { get; init; }
+ public required string Record { get; init; }
+ public required string Cron { get; init; }
+ public required string TimeZone { get; init; }
+ public required string Next { get; init; }
+ public required string Last { get; init; }
+ public required long DueScore { get; init; }
+ public required string DueMember { get; init; }
+ public required string Paused { get; init; }
+}
+
+internal sealed record RedisDefinitionCatalogChanges
+{
+ public required IReadOnlyList Definitions { get; init; }
+ public required IReadOnlyList RemovedDefinitions { get; init; }
+ public required IReadOnlyList Schedules { get; init; }
+ public required IReadOnlyList RemovedSchedules { get; init; }
+}
diff --git a/src/Immediate.Jobs.Redis/RedisJobStorage.Enqueue.cs b/src/Immediate.Jobs.Redis/RedisJobStorage.Enqueue.cs
index 3a418b64..5582a91a 100644
--- a/src/Immediate.Jobs.Redis/RedisJobStorage.Enqueue.cs
+++ b/src/Immediate.Jobs.Redis/RedisJobStorage.Enqueue.cs
@@ -39,7 +39,7 @@ private RedisValue[] CreateEnqueueArguments(JobRecord job) =>
job.ExecutionSpanId ?? "",
NullableTicks(job.ExecutionStartedAt),
job.QueueName,
- job.JobName,
+ job.JobName.ToUpperInvariant(),
Score(job.CreatedAt),
job.JobHandle.Value,
Score(job.DueAt),
diff --git a/src/Immediate.Jobs.Redis/RedisJobStorage.Keys.cs b/src/Immediate.Jobs.Redis/RedisJobStorage.Keys.cs
index 946eb217..47116206 100644
--- a/src/Immediate.Jobs.Redis/RedisJobStorage.Keys.cs
+++ b/src/Immediate.Jobs.Redis/RedisJobStorage.Keys.cs
@@ -16,6 +16,10 @@ internal sealed partial class RedisJobStorage
private RedisKey RecurringDedupeKey => _root + "recurring:dedupe";
+ private RedisKey DefinitionMetadataKey => _root + "definition-metadata";
+
+ private RedisKey DefinitionCatalogVersionKey => _root + "definition-catalog-version";
+
private RedisKey ServersKey => _root + "servers";
private RedisKey JobKey(JobHandle id) => _root + "job:" + id.Value;
diff --git a/src/Immediate.Jobs.Redis/RedisJobStorage.Query.cs b/src/Immediate.Jobs.Redis/RedisJobStorage.Query.cs
index b8953dd7..b245de33 100644
--- a/src/Immediate.Jobs.Redis/RedisJobStorage.Query.cs
+++ b/src/Immediate.Jobs.Redis/RedisJobStorage.Query.cs
@@ -121,7 +121,7 @@ public async ValueTask> QueryNonCompletedJobsAsync(
var jobs = await ReadJobsAsync(ids, cancellationToken);
return
[
- .. jobs.Where(job => string.Equals(job.JobName, jobName, StringComparison.Ordinal)
+ .. jobs.Where(job => string.Equals(job.JobName, jobName, StringComparison.OrdinalIgnoreCase)
&& job.State is (JobState.AwaitingContinuation or JobState.WaitingForTrigger
or JobState.Scheduled or JobState.Pending or JobState.Active)),
];
@@ -338,7 +338,7 @@ private static bool MatchesQuery(JobRecord job, JobQuery query) =>
(query.State is not { } state || job.State == state) &&
(query.CreatedBefore is not { } createdBefore || job.CreatedAt < createdBefore) &&
(string.IsNullOrWhiteSpace(query.QueueName) || string.Equals(job.QueueName, query.QueueName, StringComparison.Ordinal)) &&
- (string.IsNullOrWhiteSpace(query.JobName) || string.Equals(job.JobName, query.JobName, StringComparison.Ordinal)) &&
+ (string.IsNullOrWhiteSpace(query.JobName) || string.Equals(job.JobName, query.JobName, StringComparison.OrdinalIgnoreCase)) &&
(string.IsNullOrWhiteSpace(query.Search) ||
job.JobName.Contains(query.Search, StringComparison.OrdinalIgnoreCase));
diff --git a/src/Immediate.Jobs.Redis/RedisJobStorage.Recurring.cs b/src/Immediate.Jobs.Redis/RedisJobStorage.Recurring.cs
index 6112317e..bf2cde4f 100644
--- a/src/Immediate.Jobs.Redis/RedisJobStorage.Recurring.cs
+++ b/src/Immediate.Jobs.Redis/RedisJobStorage.Recurring.cs
@@ -10,32 +10,34 @@ internal sealed partial class RedisJobStorage
private static readonly RedisValue[] RecurringMutableFields = ["record", "paused", "next", "last"];
///
- public async ValueTask MergeRecurringSchedulesListAsync(
- IReadOnlyList schedules,
+ public async ValueTask UpsertRecurringAsync(
+ RecurringJobSchedule schedule,
CancellationToken cancellationToken = default
)
{
- MergeRecurringSchedulesListAsyncCalled();
+ UpsertRecurringAsyncCalled(schedule.Name);
cancellationToken.ThrowIfCancellationRequested();
await TaskScheduler.Yield();
- foreach (var schedule in schedules)
- await UpsertRecurringAsync(schedule, preserveUnchangedNextRun: true, cancellationToken);
-
- await RemoveObsoleteCodeDefinedRecurringAsync(
- schedules.Select(static schedule => schedule.Name).ToList(),
+ var result = await EvaluateInt64Async(
+ RedisScripts.UpsertRecurring,
+ [RecurringKey(schedule.Name), RecurringNamesKey, RecurringDueKey],
+ [
+ JsonSerializer.Serialize(schedule, RedisJsonSerializerContext.Default.RecurringJobSchedule),
+ schedule.IsCodeDefined ? 1 : 0,
+ schedule.IsPaused ? 1 : 0,
+ Ticks(schedule.NextRunAt),
+ NullableTicks(schedule.LastRunAt),
+ schedule.Name,
+ Score(schedule.NextRunAt),
+ RecurringDueMember(schedule.NextRunAt, schedule.Name),
+ schedule.Cron,
+ schedule.TimeZone,
+ ],
cancellationToken
);
- }
-
- ///
- public async ValueTask UpsertRecurringAsync(
- RecurringJobSchedule schedule,
- CancellationToken cancellationToken = default
- )
- {
- UpsertRecurringAsyncCalled(schedule.Name);
- await UpsertRecurringAsync(schedule, preserveUnchangedNextRun: false, cancellationToken);
+ if (result < 0)
+ throw new ImmediateJobException("Code-defined recurring schedules cannot be replaced by dynamic schedules.");
}
///
@@ -168,57 +170,6 @@ public async ValueTask MaterializeRecurringAsync(
return result == 1;
}
- private async ValueTask UpsertRecurringAsync(
- RecurringJobSchedule schedule,
- bool preserveUnchangedNextRun,
- CancellationToken cancellationToken
- )
- {
- cancellationToken.ThrowIfCancellationRequested();
- await TaskScheduler.Yield();
-
- var result = await EvaluateInt64Async(
- RedisScripts.UpsertRecurring,
- [RecurringKey(schedule.Name), RecurringNamesKey, RecurringDueKey],
- [
- JsonSerializer.Serialize(schedule, RedisJsonSerializerContext.Default.RecurringJobSchedule),
- schedule.IsCodeDefined ? 1 : 0,
- schedule.IsPaused ? 1 : 0,
- Ticks(schedule.NextRunAt),
- NullableTicks(schedule.LastRunAt),
- schedule.Name,
- Score(schedule.NextRunAt),
- RecurringDueMember(schedule.NextRunAt, schedule.Name),
- schedule.Cron,
- schedule.TimeZone,
- preserveUnchangedNextRun ? 1 : 0,
- ],
- cancellationToken
- );
- if (result < 0)
- throw new ImmediateJobException("Code-defined recurring schedules cannot be replaced by dynamic schedules.");
- }
-
- private async ValueTask RemoveObsoleteCodeDefinedRecurringAsync(
- List activeScheduleNames,
- CancellationToken cancellationToken = default
- )
- {
- var values = new RedisValue[activeScheduleNames.Count + 1];
- values[0] = _root;
-
- var index = 1;
- foreach (var name in activeScheduleNames)
- values[index++] = name;
-
- _ = await EvaluateInt64Async(
- RedisScripts.RemoveObsoleteRecurring,
- [RecurringNamesKey, RecurringDueKey],
- values,
- cancellationToken
- );
- }
-
private async ValueTask SetRecurringPausedAsync(
string name,
bool isPaused,
@@ -300,7 +251,7 @@ DateTimeOffset now
job.ExecutionSpanId ?? "",
NullableTicks(job.ExecutionStartedAt),
job.QueueName,
- job.JobName,
+ job.JobName.ToUpperInvariant(),
Score(job.CreatedAt),
Ticks(nextRunAt),
Score(nextRunAt),
@@ -328,14 +279,6 @@ private static void ValidateMaterializedJob(JobRecord job)
private static string RecurringDueMember(DateTimeOffset nextRunAt, string name) => $"{Ticks(nextRunAt)}|{name}";
- [LoggerMessage(
- EventId = LibraryEventIds.MergeRecurringSchedulesListAsyncCalled,
- EventName = "Immediate.Jobs.Redis.MergeRecurringSchedulesListAsyncCalled",
- Level = LogLevel.Debug,
- Message = "MergeRecurringSchedulesListAsync called"
- )]
- private partial void MergeRecurringSchedulesListAsyncCalled();
-
[LoggerMessage(
EventId = LibraryEventIds.UpsertRecurringAsyncCalled,
EventName = "Immediate.Jobs.Redis.UpsertRecurringAsyncCalled",
diff --git a/src/Immediate.Jobs.Redis/RedisJobStorage.Trigger.cs b/src/Immediate.Jobs.Redis/RedisJobStorage.Trigger.cs
index 6ea74626..c135f2ef 100644
--- a/src/Immediate.Jobs.Redis/RedisJobStorage.Trigger.cs
+++ b/src/Immediate.Jobs.Redis/RedisJobStorage.Trigger.cs
@@ -23,7 +23,7 @@ public async ValueTask UpdatePayloadAsync(
.WaitAsync(cancellationToken);
if (values[0].IsNull)
throw new KeyNotFoundException($"Job '{jobHandle}' was not found.");
- if (!string.Equals(values[2], expectedJobName, StringComparison.Ordinal))
+ if (!string.Equals(values[2], expectedJobName, StringComparison.OrdinalIgnoreCase))
throw new ImmediateJobException($"Job '{jobHandle}' is not a '{expectedJobName}' job.");
if ((int)values[1] != (int)JobState.WaitingForTrigger)
throw new ImmediateJobException($"Job '{jobHandle}' is not waiting for a trigger.");
@@ -58,7 +58,7 @@ public async ValueTask TryTriggerAsync(
var result = await EvaluateInt64Async(
RedisScripts.Trigger,
[JobKey(jobHandle)],
- [expectedJobName, Ticks(dueAt), Score(dueAt), Score(now), _root, jobHandle.Value],
+ [expectedJobName.ToUpperInvariant(), Ticks(dueAt), Score(dueAt), Score(now), _root, jobHandle.Value],
cancellationToken
);
if (result == 0)
diff --git a/src/Immediate.Jobs.Redis/RedisJsonSerializerContext.cs b/src/Immediate.Jobs.Redis/RedisJsonSerializerContext.cs
index cdece3bb..464c7354 100644
--- a/src/Immediate.Jobs.Redis/RedisJsonSerializerContext.cs
+++ b/src/Immediate.Jobs.Redis/RedisJsonSerializerContext.cs
@@ -6,4 +6,6 @@ namespace Immediate.Jobs.Redis;
[JsonSerializable(typeof(JobRecord))]
[JsonSerializable(typeof(RecurringJobSchedule))]
[JsonSerializable(typeof(JobServerSnapshot))]
+[JsonSerializable(typeof(JobDefinitionRecord))]
+[JsonSerializable(typeof(RedisDefinitionCatalogChanges))]
internal sealed partial class RedisJsonSerializerContext : JsonSerializerContext;
diff --git a/src/Immediate.Jobs.Redis/RedisScripts.cs b/src/Immediate.Jobs.Redis/RedisScripts.cs
index bae19061..49080c24 100644
--- a/src/Immediate.Jobs.Redis/RedisScripts.cs
+++ b/src/Immediate.Jobs.Redis/RedisScripts.cs
@@ -550,6 +550,46 @@ elseif jobCapacity and jobCapacity > 0 then
return 1
""";
+ internal const string MergeDefinitionCatalog =
+ """
+ if (redis.call('GET', KEYS[2]) or '0') ~= ARGV[1] then return 0 end
+ local changes = cjson.decode(ARGV[3])
+ for _, name in ipairs(changes.RemovedDefinitions) do redis.call('HDEL', KEYS[1], name) end
+ for _, definition in ipairs(changes.Definitions) do
+ redis.call('HSET', KEYS[1], definition.Name, definition.Metadata)
+ end
+ for _, name in ipairs(changes.RemovedSchedules) do
+ local key = ARGV[2] .. 'recurring:' .. name
+ if redis.call('HGET', key, 'code') == '1' then
+ local dueMember = redis.call('HGET', key, 'dueMember')
+ if dueMember then redis.call('ZREM', KEYS[4], dueMember) end
+ redis.call('DEL', key)
+ redis.call('SREM', KEYS[3], name)
+ end
+ end
+ for _, schedule in ipairs(changes.Schedules) do
+ local key = ARGV[2] .. 'recurring:' .. schedule.Name
+ local paused, last, next = schedule.Paused, schedule.Last, schedule.Next
+ local dueScore, dueMember = schedule.DueScore, schedule.DueMember
+ local current = redis.call('HMGET', key, 'paused', 'last', 'next', 'dueScore', 'dueMember', 'cron', 'timeZone')
+ paused = current[1] or paused
+ last = current[2] or last
+ if current[5] then redis.call('ZREM', KEYS[4], current[5]) end
+ if current[6] == schedule.Cron and current[7] == schedule.TimeZone then
+ next = current[3] or next
+ dueScore = current[4] or dueScore
+ dueMember = current[5] or dueMember
+ end
+ redis.call('HSET', key, 'record', schedule.Record, 'code', '1', 'paused', paused,
+ 'next', next, 'last', last, 'dueScore', dueScore, 'dueMember', dueMember,
+ 'cron', schedule.Cron, 'timeZone', schedule.TimeZone)
+ redis.call('SADD', KEYS[3], schedule.Name)
+ if paused ~= '1' then redis.call('ZADD', KEYS[4], dueScore, dueMember) end
+ end
+ redis.call('INCR', KEYS[2])
+ return 1
+ """;
+
internal const string UpsertRecurring =
"""
local exists = redis.call('EXISTS', KEYS[1]) == 1
@@ -559,25 +599,12 @@ elseif jobCapacity and jobCapacity > 0 then
local dueScore = ARGV[7]
local dueMember = ARGV[8]
if exists then
- local current = redis.call('HMGET', KEYS[1], 'code', 'paused', 'last', 'dueMember', 'cron', 'timeZone', 'record', 'next', 'dueScore')
+ local current = redis.call('HMGET', KEYS[1], 'code', 'paused', 'last', 'dueMember')
if current[1] == '1' and ARGV[2] == '0' then return -1 end
paused = current[2] or paused
last = current[3] or last
if current[4] then redis.call('ZREM', KEYS[3], current[4]) end
- if ARGV[11] == '1' then
- local currentCron = current[5]
- local currentTimeZone = current[6]
- if (not currentCron or not currentTimeZone) and current[7] then
- local record = cjson.decode(current[7])
- currentCron = record.Cron
- currentTimeZone = record.TimeZone
- end
- if currentCron == ARGV[9] and currentTimeZone == ARGV[10] then
- next = current[8] or next
- dueScore = current[9] or dueScore
- dueMember = current[4] or dueMember
- end
- end
+
end
redis.call('HSET', KEYS[1],
'record', ARGV[1], 'code', ARGV[2], 'paused', paused,
@@ -592,25 +619,6 @@ elseif jobCapacity and jobCapacity > 0 then
return 1
""";
- internal const string RemoveObsoleteRecurring =
- """
- local active = {}
- for index = 2, #ARGV do active[ARGV[index]] = true end
- local names = redis.call('SMEMBERS', KEYS[1])
- local removed = 0
- for _, name in ipairs(names) do
- local key = ARGV[1] .. 'recurring:' .. name
- if redis.call('HGET', key, 'code') == '1' and not active[name] then
- local dueMember = redis.call('HGET', key, 'dueMember')
- redis.call('DEL', key)
- redis.call('SREM', KEYS[1], name)
- if dueMember then redis.call('ZREM', KEYS[2], dueMember) end
- removed = removed + 1
- end
- end
- return removed
- """;
-
internal const string RemoveRecurring =
"""
if redis.call('EXISTS', KEYS[1]) == 0 then return 0 end
diff --git a/src/Immediate.Jobs.Shared/Apis/JobDefinitionRecord.cs b/src/Immediate.Jobs.Shared/Apis/JobDefinitionRecord.cs
new file mode 100644
index 00000000..b2173d7d
--- /dev/null
+++ b/src/Immediate.Jobs.Shared/Apis/JobDefinitionRecord.cs
@@ -0,0 +1,72 @@
+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 per-server definition concurrency limit. Zero is unbounded.
+ ///
+ public int MaxConcurrency { get; init; }
+
+ ///
+ /// 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 retry base delay.
+ ///
+ public TimeSpan BackoffBase { get; init; } = TimeSpan.FromSeconds(5);
+}
diff --git a/src/Immediate.Jobs.Shared/Internals/JobSchedulingService.cs b/src/Immediate.Jobs.Shared/Internals/JobSchedulingService.cs
index cac55bc9..a5450158 100644
--- a/src/Immediate.Jobs.Shared/Internals/JobSchedulingService.cs
+++ b/src/Immediate.Jobs.Shared/Internals/JobSchedulingService.cs
@@ -44,7 +44,7 @@ private readonly List<
> _queuesByPriority;
private readonly ConcurrentDictionary _queueReservations = new(StringComparer.Ordinal);
- private readonly ConcurrentDictionary _jobReservations = new(StringComparer.Ordinal);
+ private readonly ConcurrentDictionary _jobReservations = new(StringComparer.OrdinalIgnoreCase);
private readonly ConcurrentDictionary _openLeases = [];
private readonly CancellationTokenSource _workerCancellation = new();
private readonly string _workerId = string.Create(CultureInfo.InvariantCulture, $"{Environment.MachineName}:{Environment.ProcessId}:{DateTimeOffset.UtcNow.Ticks}");
@@ -112,9 +112,9 @@ ILogger logger
#pragma warning disable CA1851 // `definitions` is backed by a list
_definitions = definitions
- .ToDictionary(x => x.Name, StringComparer.Ordinal);
+ .ToDictionary(x => x.Name, StringComparer.OrdinalIgnoreCase);
- _queuesByPriority = definitions
+ _queuesByPriority = _definitions.Values
.GroupBy(d => d.Queue)
.GroupBy(
g => g.Key.Priority,
@@ -257,8 +257,27 @@ private async Task InitializeAsync(CancellationToken cancellationToken)
})
.ToList();
- await _storage.MergeRecurringSchedulesListAsync(
- schedules,
+ await _storage.MergeJobDefinitionsListAsync(
+ new()
+ {
+ Definitions = _definitions.Values.Select(static definition => new JobDefinitionRecord
+ {
+ Name = definition.Name,
+ 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,
+ OverlapPolicy = definition.OverlapPolicy,
+ MisfireHandlingMode = definition.MisfireHandlingMode,
+ Backoff = definition.Backoff,
+ BackoffBase = definition.BackoffBase,
+ }).ToList(),
+ RecurringSchedules = schedules,
+ },
cancellationToken
);
}
@@ -663,7 +682,7 @@ await _storage.FailAsync(
Capacity = GetJobAcquisitionCapacity(definition, capacity),
})
.Where(item => item.Capacity > 0)
- .ToDictionary(item => item.Name, item => item.Capacity, StringComparer.Ordinal);
+ .ToDictionary(item => item.Name, item => item.Capacity, StringComparer.OrdinalIgnoreCase);
if (jobCapacities.Count == 0)
continue;
diff --git a/src/Immediate.Jobs.Shared/JobAttribute.cs b/src/Immediate.Jobs.Shared/JobAttribute.cs
index 0af14d74..cbbf1998 100644
--- a/src/Immediate.Jobs.Shared/JobAttribute.cs
+++ b/src/Immediate.Jobs.Shared/JobAttribute.cs
@@ -11,6 +11,9 @@ public sealed class JobAttribute : Attribute
///
/// The explicit persisted job name, or to derive it.
///
+ ///
+ /// Names must not have leading or trailing whitespace.
+ ///
///
/// The explicit persisted job name, or to derive it.
///
diff --git a/src/Immediate.Jobs.Shared/LibraryEventIds.cs b/src/Immediate.Jobs.Shared/LibraryEventIds.cs
index f8a51a72..e7317738 100644
--- a/src/Immediate.Jobs.Shared/LibraryEventIds.cs
+++ b/src/Immediate.Jobs.Shared/LibraryEventIds.cs
@@ -27,7 +27,6 @@ internal static class LibraryEventIds
public const int InMemoryCompleteWithContinuationsAsyncCalled = 11023;
public const int InMemoryAddBatchJobAsyncCalled = 11024;
public const int InMemoryFailAsyncCalled = 11025;
- public const int InMemoryMergeRecurringSchedulesListAsyncCalled = 11026;
public const int InMemoryUpsertRecurringAsyncCalled = 11027;
public const int InMemoryRemoveRecurringAsyncCalled = 11028;
public const int InMemoryPauseRecurringAsyncCalled = 11029;
@@ -63,7 +62,6 @@ internal static class LibraryEventIds
public const int SingleServerCompleteWithContinuationsAsyncCalled = 11059;
public const int SingleServerAddBatchJobAsyncCalled = 11060;
public const int SingleServerFailAsyncCalled = 11061;
- public const int SingleServerMergeRecurringSchedulesListAsyncCalled = 11062;
public const int SingleServerUpsertRecurringAsyncCalled = 11063;
public const int SingleServerRemoveRecurringAsyncCalled = 11064;
public const int SingleServerPauseRecurringAsyncCalled = 11065;
@@ -95,4 +93,8 @@ internal static class LibraryEventIds
public const int SingleServerUpdatePayloadAsyncCalled = 11091;
public const int SingleServerTryTriggerAsyncCalled = 11092;
public const int SingleServerTryTriggerBatchAsyncCalled = 11093;
+ public const int InMemoryMergeJobDefinitionsListAsyncCalled = 11094;
+ public const int InMemoryGetJobDefinitionsAsyncCalled = 11095;
+ public const int SingleServerMergeJobDefinitionsListAsyncCalled = 11096;
+ public const int SingleServerGetJobDefinitionsAsyncCalled = 11097;
}
diff --git a/src/Immediate.Jobs.Shared/Storage/IJobStorage.cs b/src/Immediate.Jobs.Shared/Storage/IJobStorage.cs
index 2e2ad657..ea8675a4 100644
--- a/src/Immediate.Jobs.Shared/Storage/IJobStorage.cs
+++ b/src/Immediate.Jobs.Shared/Storage/IJobStorage.cs
@@ -20,6 +20,36 @@ public interface IJobStorage : IAsyncDisposable
///
ValueTask InitializeAsync(CancellationToken cancellationToken = default);
+ ///
+ /// Atomically reconciles definitions and code-defined schedules against the complete application catalogue.
+ ///
+ ///
+ /// The complete local definition catalogue, and code-defined schedules.
+ ///
+ ///
+ /// A token that can cancel the operation.
+ ///
+ ///
+ /// A value task representing the complete reconciliation.
+ ///
+ ///
+ /// Supplied definitions are inserted or updated; stored definitions omitted from the supplied list
+ /// are removed. Dynamic schedules, and invocation history are preserved.
+ /// Each server's list must be authoritative for the catalogue. Repeated calls are idempotent.
+ ///
+ ValueTask MergeJobDefinitionsListAsync(JobDefinitionRegistration registration, CancellationToken cancellationToken = default);
+
+ ///
+ /// Returns every persisted definition, ordered by ordinal, case-insensitive name, without local registration filtering.
+ ///
+ ///
+ /// A token that can cancel the operation.
+ ///
+ ///
+ /// The complete stored definition catalogue.
+ ///
+ ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default);
+
///
/// Inserts a pending or scheduled invocation.
///
@@ -414,31 +444,10 @@ ValueTask PurgeJobsAsync(
///
/// A token that can cancel the storage operation.
///
- /// when the provider is reachable; otherwise, .
- ///
- ValueTask IsHealthyAsync(CancellationToken cancellationToken = default);
-
- ///
- /// Resets the list of code-defined recurring job schedules to the provided list.
- ///
- ///
- /// The complete list of code-defined recurring job schedules.
- ///
- ///
- /// A token that can cancel the storage operation.
- ///
///
- /// A value task that represents the asynchronous merge.
+ /// when the provider is reachable; otherwise, .
///
- ///
- /// This method should be called exactly once per app start, after to reset the list of code-defined cron jobs to the
- /// currently compiled list.
- ///
- ValueTask MergeRecurringSchedulesListAsync(
- IReadOnlyList schedules,
- CancellationToken cancellationToken = default
- );
+ ValueTask IsHealthyAsync(CancellationToken cancellationToken = default);
///
/// Creates or updates a recurring schedule.
diff --git a/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Acquire.cs b/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Acquire.cs
index 1e9d8fa0..f10b23e8 100644
--- a/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Acquire.cs
+++ b/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Acquire.cs
@@ -59,7 +59,7 @@ public async ValueTask> AcquireDueJobsAsync(
if (queueCapacity <= 0)
continue;
- var jobCapacities = queue.JobCapacities.ToDictionary(static pair => pair.Key, static pair => pair.Value, StringComparer.Ordinal);
+ var jobCapacities = queue.JobCapacities.ToDictionary(static pair => pair.Key, static pair => pair.Value, StringComparer.OrdinalIgnoreCase);
if (!HasEligibleGroupedJob(queue.QueueName, jobCapacities, now))
{
AcquireQueueInExistingOrder(
@@ -96,7 +96,7 @@ private List AcquireInExistingOrder(JobAcquisitionRequest request, Da
if (queueCapacity <= 0)
continue;
- var jobCapacities = queue.JobCapacities.ToDictionary(static pair => pair.Key, static pair => pair.Value, StringComparer.Ordinal);
+ var jobCapacities = queue.JobCapacities.ToDictionary(static pair => pair.Key, static pair => pair.Value, StringComparer.OrdinalIgnoreCase);
AcquireQueueInExistingOrder(
request,
queue.QueueName,
diff --git a/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.DefinitionCatalog.cs b/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.DefinitionCatalog.cs
new file mode 100644
index 00000000..5333a156
--- /dev/null
+++ b/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.DefinitionCatalog.cs
@@ -0,0 +1,55 @@
+using Immediate.Jobs.Shared.Apis;
+using Microsoft.Extensions.Logging;
+
+namespace Immediate.Jobs.Shared.Storage;
+
+public sealed partial class InMemoryJobStorage
+{
+ private readonly Dictionary _jobDefinitions = [with(StringComparer.OrdinalIgnoreCase)];
+
+ ///
+ public async ValueTask MergeJobDefinitionsListAsync(JobDefinitionRegistration registration, CancellationToken cancellationToken = default)
+ {
+ InMemoryMergeJobDefinitionsListAsyncCalled();
+ cancellationToken.ThrowIfCancellationRequested();
+ await TaskScheduler.Yield();
+ lock (_gate)
+ {
+ var changes = JobDefinitionReconciliation.Create(registration, [.. _jobDefinitions.Values], [.. _recurring.Values]);
+ foreach (var name in changes.RemovedDefinitionNames)
+ _jobDefinitions.Remove(name);
+ foreach (var definition in changes.Definitions)
+ _jobDefinitions[definition.Name] = definition;
+ foreach (var name in changes.RemovedScheduleNames)
+ _recurring.Remove(name);
+ foreach (var schedule in changes.Schedules)
+ _recurring[schedule.Name] = schedule;
+ }
+ }
+
+ ///
+ public async ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default)
+ {
+ InMemoryGetJobDefinitionsAsyncCalled();
+ cancellationToken.ThrowIfCancellationRequested();
+ await TaskScheduler.Yield();
+ lock (_gate)
+ return [.. _jobDefinitions.Values.OrderBy(static definition => definition.Name, StringComparer.OrdinalIgnoreCase)];
+ }
+
+ [LoggerMessage(
+ EventId = LibraryEventIds.InMemoryMergeJobDefinitionsListAsyncCalled,
+ EventName = "Immediate.Jobs.Shared.InMemoryMergeJobDefinitionsListAsyncCalled",
+ Level = LogLevel.Debug,
+ Message = "InMemoryMergeJobDefinitionsListAsyncCalled"
+ )]
+ private partial void InMemoryMergeJobDefinitionsListAsyncCalled();
+
+ [LoggerMessage(
+ EventId = LibraryEventIds.InMemoryGetJobDefinitionsAsyncCalled,
+ EventName = "Immediate.Jobs.Shared.InMemoryGetJobDefinitionsAsyncCalled",
+ Level = LogLevel.Debug,
+ Message = "InMemoryGetJobDefinitionsAsyncCalled"
+ )]
+ private partial void InMemoryGetJobDefinitionsAsyncCalled();
+}
diff --git a/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Query.cs b/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Query.cs
index 95d05938..fcf55047 100644
--- a/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Query.cs
+++ b/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Query.cs
@@ -45,7 +45,7 @@ public async ValueTask> QueryJobsAsync(JobQuery query,
if (!string.IsNullOrWhiteSpace(query.QueueName))
jobs = jobs.Where(x => string.Equals(x.QueueName, query.QueueName, StringComparison.Ordinal));
if (!string.IsNullOrWhiteSpace(query.JobName))
- jobs = jobs.Where(x => string.Equals(x.JobName, query.JobName, StringComparison.Ordinal));
+ jobs = jobs.Where(x => string.Equals(x.JobName, query.JobName, StringComparison.OrdinalIgnoreCase));
if (query.CreatedBefore is { } createdBefore)
jobs = jobs.Where(x => x.CreatedAt < createdBefore);
@@ -72,7 +72,7 @@ public async ValueTask> QueryNonCompletedJobsAsync(stri
lock (_gate)
{
return _jobs.Values
- .Where(j => string.Equals(j.JobName, jobName, StringComparison.Ordinal))
+ .Where(j => string.Equals(j.JobName, jobName, StringComparison.OrdinalIgnoreCase))
.Where(
j => j.State is
JobState.AwaitingContinuation
diff --git a/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Recurring.cs b/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Recurring.cs
index 54e9e407..ef1cb491 100644
--- a/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Recurring.cs
+++ b/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Recurring.cs
@@ -6,47 +6,6 @@ namespace Immediate.Jobs.Shared.Storage;
public sealed partial class InMemoryJobStorage
{
- ///
- public async ValueTask MergeRecurringSchedulesListAsync(
- IReadOnlyList schedules,
- CancellationToken cancellationToken = default
- )
- {
- MergeRecurringSchedulesListAsyncCalled();
- cancellationToken.ThrowIfCancellationRequested();
- await TaskScheduler.Yield();
-
- lock (_gate)
- {
- var existingStaticDefinitions = _recurring
- .Where(kvp => kvp.Value.IsCodeDefined)
- .ToDictionary(StringComparer.Ordinal);
-
- foreach (var schedule in schedules)
- {
- ref var current = ref CollectionsMarshal.GetValueRefOrAddDefault(_recurring, schedule.Name, out _);
-
- current = current switch
- {
- { } existing when
- string.Equals(existing.Cron, schedule.Cron, StringComparison.Ordinal)
- && string.Equals(existing.TimeZone, schedule.TimeZone, StringComparison.Ordinal) =>
- existing with { JobName = schedule.JobName, QueueName = schedule.QueueName, IsCodeDefined = true },
-
- { } existing =>
- schedule with { IsPaused = existing.IsPaused, LastRunAt = existing.LastRunAt },
-
- _ => schedule,
- };
-
- existingStaticDefinitions.Remove(schedule.Name);
- }
-
- foreach (var s in existingStaticDefinitions)
- _recurring.Remove(s.Key);
- }
- }
-
///
public async ValueTask UpsertRecurringAsync(RecurringJobSchedule schedule, CancellationToken cancellationToken = default)
{
@@ -179,14 +138,6 @@ private ValueTask SetRecurringPausedAsync(string name, bool isPaused, Cancellati
return ValueTask.CompletedTask;
}
- [LoggerMessage(
- EventId = LibraryEventIds.InMemoryMergeRecurringSchedulesListAsyncCalled,
- EventName = "Immediate.Jobs.Shared.MergeRecurringSchedulesListAsyncCalled",
- Level = LogLevel.Debug,
- Message = "MergeRecurringSchedulesListAsync called"
- )]
- private partial void MergeRecurringSchedulesListAsyncCalled();
-
[LoggerMessage(
EventId = LibraryEventIds.InMemoryUpsertRecurringAsyncCalled,
EventName = "Immediate.Jobs.Shared.UpsertRecurringAsyncCalled",
diff --git a/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Trigger.cs b/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Trigger.cs
index 3d3ef200..3f3ba714 100644
--- a/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Trigger.cs
+++ b/src/Immediate.Jobs.Shared/Storage/InMemoryJobStorage.Trigger.cs
@@ -84,7 +84,7 @@ private JobRecord GetTriggerable(JobHandle jobHandle, string expectedJobName)
{
if (!_jobs.TryGetValue(jobHandle, out var job))
throw new KeyNotFoundException($"Job '{jobHandle}' was not found.");
- if (!string.Equals(job.JobName, expectedJobName, StringComparison.Ordinal))
+ if (!string.Equals(job.JobName, expectedJobName, StringComparison.OrdinalIgnoreCase))
throw new ImmediateJobException($"Job '{jobHandle}' is not a '{expectedJobName}' job.");
return job;
diff --git a/src/Immediate.Jobs.Shared/Storage/JobDefinitionReconciliation.cs b/src/Immediate.Jobs.Shared/Storage/JobDefinitionReconciliation.cs
new file mode 100644
index 00000000..9e5c284e
--- /dev/null
+++ b/src/Immediate.Jobs.Shared/Storage/JobDefinitionReconciliation.cs
@@ -0,0 +1,122 @@
+using System.ComponentModel;
+using Immediate.Jobs.Shared.Apis;
+
+namespace Immediate.Jobs.Shared.Storage;
+
+///
+/// Storage-neutral changes computed inside a provider's atomic startup operation.
+///
+[EditorBrowsable(EditorBrowsableState.Never)]
+public sealed record JobDefinitionReconciliation
+{
+ ///
+ /// The definitions to insert or replace.
+ ///
+ public required IReadOnlyList Definitions { get; init; }
+
+ ///
+ /// The names of obsolete definitions in the catalogue.
+ ///
+ public required IReadOnlyList RemovedDefinitionNames { get; init; }
+
+ ///
+ /// The merged code-defined schedules, preserving pause and unchanged progress.
+ ///
+ public required IReadOnlyList Schedules { get; init; }
+
+ ///
+ /// The names of obsolete code-defined schedules in the catalogue.
+ ///
+ public required IReadOnlyList RemovedScheduleNames { get; init; }
+
+ ///
+ /// Validates a complete catalogue and computes changes against the provider's locked snapshot.
+ ///
+ ///
+ /// The application's complete catalogue and code-defined schedules.
+ ///
+ ///
+ /// The stored definitions before reconciliation.
+ ///
+ ///
+ /// The stored recurring schedules before this reconciliation.
+ ///
+ ///
+ /// The changes to commit atomically.
+ ///
+ public static JobDefinitionReconciliation Create(
+ JobDefinitionRegistration registration,
+ IReadOnlyList existingDefinitions,
+ IReadOnlyList existingSchedules
+ )
+ {
+ ArgumentNullException.ThrowIfNull(registration);
+ ArgumentNullException.ThrowIfNull(existingDefinitions);
+ ArgumentNullException.ThrowIfNull(existingSchedules);
+ ArgumentNullException.ThrowIfNull(registration.Definitions, nameof(registration));
+ ArgumentNullException.ThrowIfNull(registration.RecurringSchedules, nameof(registration));
+ var supplied = registration.Definitions.ToDictionary(static definition => definition.Name, StringComparer.OrdinalIgnoreCase);
+ foreach (var definition in supplied.Values)
+ {
+ ArgumentException.ThrowIfNullOrWhiteSpace(definition.Name, nameof(registration));
+ if (char.IsWhiteSpace(definition.Name[0]) || char.IsWhiteSpace(definition.Name[^1]))
+ throw new ArgumentException("Job definition names must not have leading or trailing whitespace.", nameof(registration));
+ }
+
+ var canonicalNames = existingDefinitions.ToDictionary(static definition => definition.Name, static definition => definition.Name, StringComparer.OrdinalIgnoreCase);
+ var desired = supplied.Values
+ .Select(definition => canonicalNames.TryGetValue(definition.Name, out var name) ? definition with { Name = name } : definition)
+ .ToDictionary(static definition => definition.Name, StringComparer.OrdinalIgnoreCase);
+ var managedNames = existingDefinitions
+ .Select(static definition => definition.Name)
+ .ToHashSet(StringComparer.OrdinalIgnoreCase);
+ var removedDefinitions = managedNames.Where(name => !desired.ContainsKey(name)).ToList();
+ managedNames.UnionWith(existingSchedules.Where(static schedule => schedule.IsCodeDefined).Select(static schedule => schedule.JobName));
+ managedNames.UnionWith(desired.Keys);
+
+ var existing = existingSchedules.ToDictionary(static schedule => schedule.Name, StringComparer.Ordinal);
+ var schedules = new List();
+ var scheduleNames = new HashSet(StringComparer.OrdinalIgnoreCase);
+ foreach (var suppliedSchedule in registration.RecurringSchedules)
+ {
+ var schedule = suppliedSchedule;
+ if (!supplied.TryGetValue(schedule.JobName, out var definition)
+ || !schedule.IsCodeDefined
+ || !string.Equals(schedule.Name, definition.Name, StringComparison.OrdinalIgnoreCase)
+ || !string.Equals(schedule.Cron, definition.Cron, StringComparison.Ordinal)
+ || !string.Equals(schedule.TimeZone, definition.TimeZone, StringComparison.Ordinal)
+ || !string.Equals(schedule.QueueName, definition.QueueName, StringComparison.Ordinal)
+ || !scheduleNames.Add(schedule.Name))
+ throw new ArgumentException("Code-defined schedules must match their supplied job definitions.", nameof(registration));
+ if (!desired.TryGetValue(schedule.JobName, out var managedDefinition))
+ continue;
+ schedule = schedule with { Name = managedDefinition.Name, JobName = managedDefinition.Name };
+ if (existing.TryGetValue(schedule.Name, out var current))
+ {
+ schedules.Add(schedule with
+ {
+ IsPaused = current.IsPaused,
+ LastRunAt = current.LastRunAt,
+ NextRunAt = string.Equals(current.Cron, schedule.Cron, StringComparison.Ordinal)
+ && string.Equals(current.TimeZone, schedule.TimeZone, StringComparison.Ordinal)
+ ? current.NextRunAt : schedule.NextRunAt,
+ });
+ }
+ else
+ schedules.Add(schedule);
+ }
+
+ if (desired.Values.Any(definition => definition.Cron is not null && !scheduleNames.Contains(definition.Name)))
+ throw new ArgumentException("Every managed cron definition must supply its code-defined schedule.", nameof(registration));
+ var desiredScheduleNames = schedules.Select(static schedule => schedule.Name).ToHashSet(StringComparer.OrdinalIgnoreCase);
+ return new()
+ {
+ Definitions = [.. desired.Values],
+ RemovedDefinitionNames = removedDefinitions,
+ Schedules = schedules,
+ RemovedScheduleNames = existingSchedules
+ .Where(schedule => schedule.IsCodeDefined && managedNames.Contains(schedule.JobName) && !desiredScheduleNames.Contains(schedule.Name))
+ .Select(static schedule => schedule.Name).ToList(),
+ };
+ }
+}
diff --git a/src/Immediate.Jobs.Shared/Storage/JobDefinitionRegistration.cs b/src/Immediate.Jobs.Shared/Storage/JobDefinitionRegistration.cs
new file mode 100644
index 00000000..cf05c75d
--- /dev/null
+++ b/src/Immediate.Jobs.Shared/Storage/JobDefinitionRegistration.cs
@@ -0,0 +1,19 @@
+using Immediate.Jobs.Shared.Apis;
+
+namespace Immediate.Jobs.Shared.Storage;
+
+///
+/// The complete application startup catalogue.
+///
+public sealed record JobDefinitionRegistration
+{
+ ///
+ /// All locally registered definitions, provided at startup.
+ ///
+ public required IReadOnlyList Definitions { get; init; }
+
+ ///
+ /// The complete list of code-defined recurring schedules for these definitions.
+ ///
+ public IReadOnlyList RecurringSchedules { get; init; } = [];
+}
diff --git a/src/Immediate.Jobs.Shared/Storage/SingleServerJobStorage.DefinitionCatalog.cs b/src/Immediate.Jobs.Shared/Storage/SingleServerJobStorage.DefinitionCatalog.cs
new file mode 100644
index 00000000..44810a03
--- /dev/null
+++ b/src/Immediate.Jobs.Shared/Storage/SingleServerJobStorage.DefinitionCatalog.cs
@@ -0,0 +1,61 @@
+using Immediate.Jobs.Shared.Apis;
+using Microsoft.Extensions.Logging;
+
+namespace Immediate.Jobs.Shared.Storage;
+
+internal sealed partial class SingleServerJobStorage
+{
+ ///
+ public async ValueTask MergeJobDefinitionsListAsync(JobDefinitionRegistration registration, CancellationToken cancellationToken = default)
+ {
+ SingleServerMergeJobDefinitionsListAsyncCalled();
+ cancellationToken.ThrowIfCancellationRequested();
+ await TaskScheduler.Yield();
+ await EnsureInitializedAsync(cancellationToken);
+ await _recurringMaterialization.WaitAsync(cancellationToken);
+ try
+ {
+ await DurableStorage.MergeJobDefinitionsListAsync(registration, cancellationToken);
+ // Refresh the entire durable catalogue of schedules, from durable storage.
+ var snapshot = await DurableStorage.GetMonitoringSnapshotAsync(cancellationToken);
+ await PrimaryStorage.MergeJobDefinitionsListAsync(
+ new()
+ {
+ Definitions = await DurableStorage.GetJobDefinitionsAsync(cancellationToken),
+ RecurringSchedules = snapshot.Recurring.Where(static schedule => schedule.IsCodeDefined).ToList(),
+ },
+ cancellationToken
+ );
+ }
+ finally
+ {
+ _recurringMaterialization.Release();
+ }
+ }
+
+ ///
+ public async ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default)
+ {
+ SingleServerGetJobDefinitionsAsyncCalled();
+ cancellationToken.ThrowIfCancellationRequested();
+ await TaskScheduler.Yield();
+ await EnsureInitializedAsync(cancellationToken);
+ return await DurableStorage.GetJobDefinitionsAsync(cancellationToken);
+ }
+
+ [LoggerMessage(
+ EventId = LibraryEventIds.SingleServerMergeJobDefinitionsListAsyncCalled,
+ EventName = "Immediate.Jobs.Shared.SingleServerMergeJobDefinitionsListAsyncCalled",
+ Level = LogLevel.Debug,
+ Message = "SingleServerMergeJobDefinitionsListAsyncCalled"
+ )]
+ private partial void SingleServerMergeJobDefinitionsListAsyncCalled();
+
+ [LoggerMessage(
+ EventId = LibraryEventIds.SingleServerGetJobDefinitionsAsyncCalled,
+ EventName = "Immediate.Jobs.Shared.SingleServerGetJobDefinitionsAsyncCalled",
+ Level = LogLevel.Debug,
+ Message = "SingleServerGetJobDefinitionsAsyncCalled"
+ )]
+ private partial void SingleServerGetJobDefinitionsAsyncCalled();
+}
diff --git a/src/Immediate.Jobs.Shared/Storage/SingleServerJobStorage.Recurring.cs b/src/Immediate.Jobs.Shared/Storage/SingleServerJobStorage.Recurring.cs
index fdf8827f..636649c9 100644
--- a/src/Immediate.Jobs.Shared/Storage/SingleServerJobStorage.Recurring.cs
+++ b/src/Immediate.Jobs.Shared/Storage/SingleServerJobStorage.Recurring.cs
@@ -7,19 +7,6 @@ internal sealed partial class SingleServerJobStorage
{
private readonly SemaphoreSlim _recurringMaterialization = new(1, 1);
- ///
- public async ValueTask MergeRecurringSchedulesListAsync(
- IReadOnlyList schedules,
- CancellationToken cancellationToken = default
- )
- {
- SingleServerMergeRecurringSchedulesListAsyncCalled(schedules.Count);
- await TaskScheduler.Yield();
- await EnsureInitializedAsync(cancellationToken);
- await DurableStorage.MergeRecurringSchedulesListAsync(schedules, cancellationToken);
- await PrimaryStorage.MergeRecurringSchedulesListAsync(schedules, cancellationToken);
- }
-
///
public async ValueTask UpsertRecurringAsync(RecurringJobSchedule schedule, CancellationToken cancellationToken = default)
{
@@ -111,14 +98,6 @@ public async ValueTask MaterializeRecurringAsync(
}
}
- [LoggerMessage(
- EventId = LibraryEventIds.SingleServerMergeRecurringSchedulesListAsyncCalled,
- EventName = "Immediate.Jobs.Shared.SingleServerMergeRecurringSchedulesListAsyncCalled",
- Level = LogLevel.Debug,
- Message = "Single-server storage MergeRecurringSchedulesListAsync called (Schedules={Schedules})"
- )]
- private partial void SingleServerMergeRecurringSchedulesListAsyncCalled(int schedules);
-
[LoggerMessage(
EventId = LibraryEventIds.SingleServerUpsertRecurringAsyncCalled,
EventName = "Immediate.Jobs.Shared.SingleServerUpsertRecurringAsyncCalled",
diff --git a/src/Immediate.Jobs.Testing/CapturingJobStorage.cs b/src/Immediate.Jobs.Testing/CapturingJobStorage.cs
index 5b2d558d..de85c054 100644
--- a/src/Immediate.Jobs.Testing/CapturingJobStorage.cs
+++ b/src/Immediate.Jobs.Testing/CapturingJobStorage.cs
@@ -28,45 +28,67 @@ public class CapturingJobStorage(TimeProvider timeProvider) :
private readonly List _recurringOperations = [];
private readonly List _recurringMaterializations = [];
- /// Gets snapshots of all job records submitted by schedulers.
+ ///
+ /// Gets snapshots of all job records submitted by schedulers.
+ ///
public IReadOnlyList Jobs { get { lock (_gate) return [.. _jobs]; } }
- /// Gets snapshots of continuation enqueue operations.
+ ///
+ /// Gets snapshots of continuation enqueue operations.
+ ///
public IReadOnlyList Continuations { get { lock (_gate) return [.. _continuations]; } }
- /// Gets snapshots of committed batch enqueue operations.
+ ///
+ /// Gets snapshots of committed batch enqueue operations.
+ ///
public IReadOnlyList Batches { get { lock (_gate) return [.. _batches]; } }
- /// Gets snapshots of jobs dynamically added to running batches.
+ ///
+ /// Gets snapshots of jobs dynamically added to running batches.
+ ///
public IReadOnlyList BatchJobs { get { lock (_gate) return [.. _batchJobs]; } }
- /// Gets snapshots of continuations buffered by running jobs.
+ ///
+ /// Gets snapshots of continuations buffered by running jobs.
+ ///
public IReadOnlyList DynamicContinuations { get { lock (_gate) return [.. _dynamicContinuations]; } }
- /// Gets recurring schedules submitted for creation or update.
+ ///
+ /// Gets recurring schedules submitted for creation or update.
+ ///
public IReadOnlyDictionary RecurringSchedules { get { lock (_gate) return _recurringSchedules.ToDictionary(StringComparer.Ordinal); } }
- /// Gets recurring schedule mutation calls in call order.
+ ///
+ /// Gets recurring schedule mutation calls in call order.
+ ///
public IReadOnlyList RecurringOperations { get { lock (_gate) return [.. _recurringOperations]; } }
- /// Gets recurring occurrence materialization attempts.
+ ///
+ /// Gets recurring occurrence materialization attempts.
+ ///
public IReadOnlyList RecurringMaterializations { get { lock (_gate) return [.. _recurringMaterializations]; } }
- /// Returns the captured job with the supplied identifier, or .
+ ///
+ /// Returns the captured job with the supplied identifier, or .
+ ///
public JobRecord? FindJob(JobHandle jobHandle)
{
lock (_gate)
return _jobs.LastOrDefault(job => job.JobHandle == jobHandle);
}
- /// Returns the captured batch with the supplied identifier, or .
+ ///
+ /// Returns the captured batch with the supplied identifier, or .
+ ///
public BatchCapture? FindBatch(BatchHandle batchHandle)
{
lock (_gate)
return _batches.LastOrDefault(batch => batch.Batch.BatchHandle == batchHandle);
}
- /// Clears captured inputs without changing persisted in-memory state.
+ ///
+ /// Clears captured inputs without changing persisted in-memory state.
+ ///
public void Clear()
{
lock (_gate)
@@ -117,6 +139,24 @@ IReadOnlyList recurringSchedules
///
public virtual async ValueTask InitializeAsync(CancellationToken cancellationToken = default) => await _inner.InitializeAsync(cancellationToken);
+ ///
+ public virtual async ValueTask MergeJobDefinitionsListAsync(JobDefinitionRegistration registration, CancellationToken cancellationToken = default)
+ {
+ ArgumentNullException.ThrowIfNull(registration);
+ await _inner.MergeJobDefinitionsListAsync(registration, cancellationToken);
+ lock (_gate)
+ {
+ foreach (var schedule in registration.RecurringSchedules)
+ {
+ _recurringSchedules[schedule.Name] = schedule;
+ }
+ }
+ }
+
+ ///
+ public virtual ValueTask> GetJobDefinitionsAsync(CancellationToken cancellationToken = default) =>
+ _inner.GetJobDefinitionsAsync(cancellationToken);
+
///
public virtual async ValueTask EnqueueAsync(JobRecord job, CancellationToken cancellationToken = default)
{
@@ -176,23 +216,6 @@ public virtual async ValueTask CompleteWithContinuationsAsync(JobHandle jobHandl
await _inner.CompleteWithContinuationsAsync(jobHandle, executionNumber, workerId, additions, cancellationToken);
}
- ///
- public virtual async ValueTask MergeRecurringSchedulesListAsync(
- IReadOnlyList schedules,
- CancellationToken cancellationToken = default
- )
- {
- ArgumentNullException.ThrowIfNull(schedules);
-
- await _inner.MergeRecurringSchedulesListAsync(schedules, cancellationToken);
-
- lock (_gate)
- {
- foreach (var rs in schedules)
- _recurringSchedules[rs.Name] = rs;
- }
- }
-
///
public virtual async ValueTask UpsertRecurringAsync(RecurringJobSchedule schedule, CancellationToken cancellationToken = default)
{
@@ -297,12 +320,20 @@ public virtual async ValueTask ResumeRecurringAsync(string name, CancellationTok
///
public virtual async ValueTask PurgeBatchesAsync(TimeSpan batchSucceededRetention, TimeSpan batchFailedRetention, CancellationToken cancellationToken = default) => await _inner.PurgeBatchesAsync(batchSucceededRetention, batchFailedRetention, cancellationToken);
///
- /// Not supported: capturing storage cannot be used as a single-server durable store.
- /// Always thrown.
+ ///
+ /// Not supported: capturing storage cannot be used as a single-server durable store.
+ ///
+ ///
+ /// Always thrown.
+ ///
public virtual async ValueTask> AcquireJobsAsync(IReadOnlyCollection jobs, string workerId, TimeSpan lease, CancellationToken cancellationToken = default) => await _inner.AcquireJobsAsync(jobs, workerId, lease, cancellationToken);
///
- /// Not supported: capturing storage cannot be used as a single-server durable store.
- /// Always thrown.
+ ///
+ /// Not supported: capturing storage cannot be used as a single-server durable store.
+ ///
+ ///
+ /// Always thrown.
+ ///
public virtual async ValueTask> GetIncomingEdgesAsync(IReadOnlyCollection childJobs, CancellationToken cancellationToken = default) => await _inner.GetIncomingEdgesAsync(childJobs, cancellationToken);
///
public virtual async ValueTask HeartbeatAsync(JobServerSnapshot server, CancellationToken cancellationToken = default) => await _inner.HeartbeatAsync(server, cancellationToken);
@@ -315,35 +346,57 @@ private void Capture(JobRecord job)
}
}
-/// A captured continuation enqueue operation.
+///
+/// A captured continuation enqueue operation.
+///
public sealed record ContinuationCapture(JobRecord Job, IReadOnlyList Edges);
-/// A captured atomic batch enqueue operation.
+///
+/// A captured atomic batch enqueue operation.
+///
public sealed record BatchCapture(BatchRecord Batch, IReadOnlyList Jobs, IReadOnlyList Edges);
-/// A captured dynamic batch-member operation.
+///
+/// A captured dynamic batch-member operation.
+///
public sealed record BatchJobCapture(JobHandle CurrentJobHandle, int ExecutionNumber, JobRecord Job, ContinuationOptions Options);
-/// A captured set of continuations flushed when a running job completed.
+///
+/// A captured set of continuations flushed when a running job completed.
+///
public sealed record DynamicContinuationCapture(JobHandle JobHandle, int ExecutionNumber, string WorkerId, IReadOnlyList Additions);
-/// The kind of recurring schedule mutation that was captured.
+///
+/// The kind of recurring schedule mutation that was captured.
+///
public enum RecurringOperation
{
- /// A schedule was created or updated.
+ ///
+ /// A schedule was created or updated.
+ ///
Upsert,
- /// A schedule was removed.
+ ///
+ /// A schedule was removed.
+ ///
Remove,
- /// A schedule was paused.
+ ///
+ /// A schedule was paused.
+ ///
Pause,
- /// A schedule was resumed.
+ ///
+ /// A schedule was resumed.
+ ///
Resume,
}
-/// A captured recurring schedule mutation.
+///
+/// A captured recurring schedule mutation.
+///
public sealed record RecurringOperationCapture(RecurringOperation Operation, string Name);
-/// A captured recurring occurrence materialization attempt.
+///
+/// A captured recurring occurrence materialization attempt.
+///
public sealed record RecurringMaterializationCapture(
RecurringJobSchedule Schedule,
JobRecord Job,
diff --git a/src/Immediate.Jobs.Testing/Storage/DefinitionCatalogStorageConformance.cs b/src/Immediate.Jobs.Testing/Storage/DefinitionCatalogStorageConformance.cs
new file mode 100644
index 00000000..8d55f566
--- /dev/null
+++ b/src/Immediate.Jobs.Testing/Storage/DefinitionCatalogStorageConformance.cs
@@ -0,0 +1,181 @@
+using Immediate.Jobs.Shared.Apis;
+using Immediate.Jobs.Shared.Storage;
+using Microsoft.Extensions.Time.Testing;
+
+namespace Immediate.Jobs.Testing.Storage;
+
+internal static class DefinitionCatalogStorageConformance
+{
+ private const string RecurringCase = "Definitions.Recurring.PreservesProgressAndInvocations";
+ private const string ConcurrentCase = "Definitions.Catalogue.SerializesConcurrentCompleteLists";
+ private const string InvalidCase = "Definitions.Catalogue.RejectsInvalidSnapshotsWithoutPartialChanges";
+ private const string CaseNameCase = "Definitions.Catalogue.ReconcilesCaseInsensitiveNames";
+ private const string ConfigurationCase = "Definitions.Catalogue.RoundTripsAndUpdatesConfiguration";
+
+ internal static IReadOnlyList Cases { get; } =
+ [
+ new(RecurringCase, StorageCapabilities.Queue, PreservesRecurringAsync),
+ new(ConcurrentCase, StorageCapabilities.Queue, ReconcilesConcurrentlyAsync),
+ new(InvalidCase, StorageCapabilities.Queue, RejectsInvalidAsync),
+ new(CaseNameCase, StorageCapabilities.Queue, ReconcilesCaseInsensitiveNamesAsync),
+ new(ConfigurationCase, StorageCapabilities.Queue, RoundTripsConfigurationAsync),
+ ];
+
+ private static JobDefinitionRecord Definition(string name) => new() { Name = name };
+
+ private static JobDefinitionRegistration Registration(IReadOnlyList