diff --git a/src/content/docs/Immediate.Jobs/api-reference.md b/src/content/docs/Immediate.Jobs/api-reference.md index cf0488e..6357bfa 100644 --- a/src/content/docs/Immediate.Jobs/api-reference.md +++ b/src/content/docs/Immediate.Jobs/api-reference.md @@ -1,12 +1,25 @@ --- title: API reference -description: Application-facing Immediate.Jobs attributes, schedulers, options, monitoring, providers and testing contracts. +description: Public APIs for defining, scheduling, monitoring, managing, storing and testing jobs. order: 16 group: Reference --- -This reference groups the supported application surface. Generated scheduler methods are shown on -their public base contracts even though application code normally uses `YourJob.Scheduler`. +This page lists the public APIs most applications use. Generated scheduler methods appear on their +public base contracts, although application code normally calls `YourJob.Scheduler`. + +## Namespaces + +| Namespace | Contains | +| ---------------------------------- | ----------------------------------------------------------------- | +| `Immediate.Jobs.Shared` | Job declarations, handles, schedulers, batches and configuration. | +| `Immediate.Jobs.Shared.Interfaces` | Scheduling, recurring jobs, monitoring, serialization and IDs. | +| `Immediate.Jobs.Shared.Apis` | `JobMonitor` and the records returned by monitoring calls. | +| `Immediate.Jobs.Shared.Storage` | Contracts for custom storage providers. | + +Storage-provider extensions use `Immediate.Jobs.EntityFrameworkCore`, `Immediate.Jobs.LinqToDB`, +and `Immediate.Jobs.Redis`. The generated `AddXxxJobs` and `RecurringJobs` types are placed in the +application project's `RootNamespace`. ## Declaration attributes and enums @@ -68,20 +81,16 @@ sealed record BatchHandle string Id { get; } } -public abstract class JobContextExtractor +public abstract class JobContextExtractor { public abstract string Key { get; } -} - -public abstract class JobContextExtractor : JobContextExtractor -{ public abstract TContext? Capture(); public abstract void Restore(TContext context); } ``` `IIdGenerator.CreateId(IdKind kind)` creates `Job` and `Batch` IDs. The default returns a GUID in -the `N` format. `ImmediateJobsBuilder.UseIdGenerator()` replaces it with a singleton, +the `N` format. `IImmediateJobsBuilder.UseIdGenerator()` replaces it with a singleton, thread-safe generator; see [Custom identifiers](/docs/Immediate.Jobs/enqueueing-and-scheduling#custom-identifiers) for a Snowflake example. @@ -103,12 +112,12 @@ interface IJobScheduler Every generated `JobScheduler` additionally exposes: ```csharp -JobHandle AddToBatch(JobBatch batch, TPayload payload, TimeSpan? delay = null); +JobHandle AddToBatch(Batch batch, TPayload payload, TimeSpan? delay = null); JobHandle AddToBatchInGroup( - JobBatch batch, TPayload payload, string? groupId, TimeSpan? delay = null); -JobHandle AddToBatchAt(JobBatch batch, TPayload payload, DateTimeOffset runAt); + Batch batch, TPayload payload, string? groupId, TimeSpan? delay = null); +JobHandle AddToBatchAt(Batch batch, TPayload payload, DateTimeOffset runAt); JobHandle AddToBatchAt( - JobBatch batch, TPayload payload, DateTimeOffset runAt, string? groupId); + Batch batch, TPayload payload, DateTimeOffset runAt, string? groupId); ValueTask ScheduleAfterAsync( JobHandle parent, TPayload payload, @@ -161,19 +170,25 @@ interface IRecurringJobScheduler : IRecurringJobTrigger ValueTask RemoveRecurringAsync(string name, CancellationToken token = default); } -public sealed class JobBatch : IAsyncDisposable +// Generated in the application's root namespace for all payloadless jobs. +sealed class RecurringJobs +{ + ValueTask TriggerNowAsync(string jobName, CancellationToken token = default); +} + +public sealed class Batch : IAsyncDisposable { public string Id { get; } public ValueTask CommitAsync(CancellationToken token = default); } -interface IJobBatchScheduler +interface IBatchScheduler { ValueTask CancelAsync(BatchHandle handle, CancellationToken token = default); - JobBatch Begin(); - JobBatch Begin(BatchHandle after, ContinuationTrigger on = ContinuationTrigger.Success); + Batch Begin(); + Batch Begin(BatchHandle after, ContinuationTrigger on = ContinuationTrigger.Success); ValueTask RunAsync( - Func body, CancellationToken token = default); + Func body, CancellationToken token = default); } ``` @@ -181,11 +196,32 @@ interface IJobBatchScheduler ## Runtime configuration -`AddXxxJobs(Action? configure = null, params ... tags)` returns -`ImmediateJobsBuilder`. +`AddXxxJobs(params ... tags)` returns `IImmediateJobsBuilder`. The interface exposes: + +```csharp +IServiceCollection Services { get; } + +IImmediateJobsBuilder ConfigureWorkers(Action configure); +IImmediateJobsBuilder ConfigureWorkers( + Action> configure); +IImmediateJobsBuilder DisableWorkers(); + +IImmediateJobsBuilder UseFairQueues(); +IImmediateJobsBuilder UseFairQueues( + Action> configure); + +IImmediateJobsBuilder ConfigureStorage( + Action configure); +IImmediateJobsBuilder UseIdGenerator(); +IImmediateJobsBuilder AddHealthCheck( + string name = "immediate-jobs", + HealthStatus? failureStatus = null, + IEnumerable? tags = null); +``` | `ImmediateJobsOptions` member | Default | | ------------------------------------------------ | --------------------------------------------------: | +| `IsJobSchedulingServiceEnabled` | `true` | | `MaxParallelJobs` | `Math.Clamp(Environment.ProcessorCount * 4, 8, 32)` | | `AcquisitionBatchSize` | `32` | | `PollingInterval` | 1 second | @@ -194,40 +230,51 @@ interface IJobBatchScheduler | `SucceededRetention` / `BatchSucceededRetention` | 24 hours | | `FailedRetention` / `BatchFailedRetention` | 7 days | | `PurgeInterval` | 1 hour | -| `StorageMode` | `InMemory` when no storage provider is selected | -Fluent methods are `UseInMemory()`, `UseStorage(factory)`, `UseSingleServer()`, -`UseSingleServer(factory)`, `UseDistributed()` and `UseFairQueues(configure)`. `FairQueueOptions` -has `ConcurrencyShareThreshold = 0.10`, `MinInflightForNoisy = 30`, and -`GroupRoundRobin = true`. Builder extensions are `UseIdGenerator()` and -`AddHealthCheck(name = "immediate-jobs", failureStatus = null, tags = null)`. +`IImmediateJobsStorageBuilder` exposes `Services`, `UseInMemory()`, `UseStorage(factory)`, +`UseStorage()`, `UseSingleServer()`, `UseSingleServer(factory)`, `UseDistributed()`, +and `UseDistributed(factory)`. Call `ConfigureStorage` exactly once and select a provider. A +durable provider uses single-server mode unless you select a mode explicitly. Redis always uses +distributed mode. Provider extensions can use `Services` to add their dependencies. -`ImmediateJobsOptions.StorageMode` starts as `SingleServer`, which is the default topology when a -durable factory does not select another mode. During registration, however, no selected storage -factory causes Jobs to call `UseInMemory()`; the effective no-provider mode is therefore -`InMemory`. +`DisableWorkers()` sets `IsJobSchedulingServiceEnabled` to `false`. The hosted worker exits without +initializing storage or executing jobs. Registration, schedulers and storage remain available. + +`FairQueueOptions` defaults to `Enabled = false`, `ConcurrencyShareThreshold = 0.10`, +`MinInflightForNoisy = 30`, and `GroupRoundRobin = true`. `UseFairQueues` sets `Enabled` to `true`. +The `OptionsBuilder` overloads support configuration binding. Jobs validates +`ImmediateJobsOptions` and `FairQueueOptions` at startup. ## Serialization and telemetry -`IJobSerializer` exposes generic `Serialize`/`Deserialize` pairs both with and without a generated -`JsonTypeInfo` factory. `SystemTextJsonJobSerializer` uses web defaults and exposes `Options`. -Generated jobs always call the metadata-factory overload. `JobTelemetry.ActivitySource` and -`JobTelemetry.Meter` are the public OpenTelemetry entry points. +`IJobSerializer` exposes generic `Serialize` and `Deserialize` overloads with or without generated +`JsonTypeInfo`. `SystemTextJsonJobSerializer` uses web defaults and exposes `Options`. Generated +jobs use the overloads with generated JSON metadata. The activity source and meter are both named +`Immediate.Jobs`. + +## Monitoring and management -## Monitoring +`JobMonitor` is the scoped service for reading status and managing stored jobs, batches and +recurring schedules. `IJobMonitor` contains only the read methods and resolves to the same scoped +instance. Use the interface when a component does not need management commands or when a test +needs a simple replacement. ```csharp interface IJobMonitor { + ValueTask GetSnapshotAsync(CancellationToken token = default); + ValueTask> QueryJobsAsync( + JobQuery query, CancellationToken token = default); + ValueTask> QueryExecutionsAsync( + JobExecutionQuery query, CancellationToken token = default); ValueTask GetJobAsync(string jobId, CancellationToken token = default); -} - -interface IJobBatchMonitor -{ - ValueTask GetStatusAsync(string batchId, CancellationToken token = default); - ValueTask> QueryMembersAsync( + ValueTask?> QueryBatchesAsync( + BatchQuery query, CancellationToken token = default); + ValueTask GetBatchAsync(string batchId, CancellationToken token = default); + ValueTask?> QueryBatchMembersAsync( string batchId, BatchMemberQuery query, CancellationToken token = default); - ValueTask GetGraphAsync(string batchId, CancellationToken token = default); + ValueTask GetBatchGraphAsync( + string batchId, CancellationToken token = default); } sealed record BatchStatus( @@ -252,72 +299,95 @@ sealed record JobExecutionRecord string? Error { get; init; } bool IsSynthetic { get; init; } } +``` -sealed record JobExecutionQuery -{ - const int MaximumTake = 1000; - void Validate(); - string JobId { get; init; } - int? Attempt { get; init; } - int Skip { get; init; } - int Take { get; init; } // 100 -} +The concrete `JobMonitor` also exposes these management methods: + +```csharp +ValueTask CancelJobAsync(string jobId, CancellationToken token = default); +ValueTask RetryJobAsync(string jobId, CancellationToken token = default); +ValueTask CancelBatchAsync(string batchId, CancellationToken token = default); +ValueTask DeleteBatchAsync(string batchId, CancellationToken token = default); +ValueTask PauseRecurringAsync(string name, CancellationToken token = default); +ValueTask ResumeRecurringAsync(string name, CancellationToken token = default); +ValueTask TriggerRecurringAsync(string name, CancellationToken token = default); ``` -`BatchMemberQuery` and `JobBatchQuery` contain optional state, `Skip`, and `Take = 100`. +Query objects enforce these rules: + +- `JobQuery` can filter by ID, state, queue name, job name or search text for job names. +- `JobExecutionQuery` requires a job ID. An optional attempt number must be positive. +- `BatchQuery` and `BatchMemberQuery` can filter by state. +- IDs and text filters cannot be blank. +- `Skip` must be zero or greater. `Take` must be from 1 through 1,000 and defaults to 100. + `JobStatus`, `BatchStatus`, `BatchMemberStatus`, `BatchGraph`, `BatchGraphNode` and -`BatchGraphEdge` are immutable monitoring records. `FractionSettled` includes every terminal -outcome, including `Skipped`. `IJobStorage.QueryJobExecutionsAsync` returns retained executions -newest first unless `JobExecutionQuery.Attempt` selects an exact one. `IsSynthetic` marks a -best-effort record reconstructed from the owning `JobRecord` when a separate execution entry is -unavailable. +`BatchGraphEdge` are read-only monitoring records. `FractionSettled` counts every finished result, +including `Skipped`. `QueryExecutionsAsync` returns saved attempts newest first unless +`JobExecutionQuery.Attempt` selects one. `IsSynthetic` is `true` when Jobs rebuilt execution data +from the owning `JobRecord` because no separate execution record was available. + +`GetSnapshotAsync` reports the features supported by the current storage provider. `GetJobAsync` +includes the current job definition's `MaxAttempts` when that definition is available. Batch reads +return `null` when storage does not support graphs. `GetBatchAsync` and `GetBatchGraphAsync` also +return `null` when the batch does not exist. + +`CancelJobAsync` cancels a job that has not finished. `RetryJobAsync` retries a failed job or runs +a scheduled job now. Batch commands require graph storage. Recurring commands require recurring +storage and use the saved schedule name. Blank IDs and names are rejected. ## Dashboard ```csharp -IServiceCollection AddImmediateJobsDashboard( - this IServiceCollection services, - Action? configure = null); +IImmediateJobsDashboardBuilder AddImmediateJobsDashboard( + this IImmediateJobsBuilder builder); +IImmediateJobsDashboardBuilder ConfigureDashboard( + Action configure); +IImmediateJobsDashboardBuilder ConfigureDashboard( + Action> configure); +IImmediateJobsDashboardBuilder AddTelemetryLink( + string label, + JobTelemetryLinkKind kind, + Func createUrl); RouteGroupBuilder MapImmediateJobsDashboard( - this IEndpointRouteBuilder endpoints, - Action? configure = null); + this IEndpointRouteBuilder endpoints); RouteGroupBuilder MapImmediateJobsDashboard( - this IEndpointRouteBuilder endpoints, string prefix, - Action? configure = null); + this IEndpointRouteBuilder endpoints, string prefix); ``` -Call `AddImmediateJobsDashboard` before building the application. It registers the dashboard's -generated Immediate.Apis handlers and Immediate.Validations behavior. `MapImmediateJobsDashboard` -also accepts an optional configuration callback, but service registration is the preferred -configuration point. `ImmediateJobsDashboardOptions.UpdateInterval` defaults to two seconds. -`AllowInAnyEnvironment()`, `RequireAuthorization(string policy)` and -`AddTelemetryLink(string label, JobTelemetryLinkKind kind, -Func createUrl)` return the same options object. Without an -authorization policy, dashboard endpoints are restricted to the `Development` environment by -default; `AllowInAnyEnvironment()` explicitly disables that restriction. A configured -authorization policy remains authoritative. Link kinds are `Trace` and `Logs`. +Chain `AddImmediateJobsDashboard` from the jobs registration before building the application. +`ConfigureDashboard` sets options. `MapImmediateJobsDashboard` only selects the default or custom +path. Jobs validates the settings when the host starts. + +`ImmediateJobsDashboardOptions.UpdateInterval` defaults to two seconds. +`RestrictToDevelopmentEnvironment` defaults to `true`, and `AuthorizationPolicy` defaults to +`null`. A named policy replaces the environment check. With no policy, setting the restriction to +`false` makes the dashboard available in every environment. Link kinds are `Trace` and `Logs`. `JobTelemetryLinkContext.Execution` is `null` for a job-level link and contains the exact `JobExecutionRecord` for an execution-level link. ## Provider registration ```csharp -ImmediateJobsOptions UseEntityFrameworkCore(); +IImmediateJobsStorageBuilder UseEntityFrameworkCore(); ModelBuilder AddImmediateJobs(string? schema = null); -ImmediateJobsOptions UseLinqToDB(DataOptions dataOptions, string? schema = null); -Task CreateImmediateJobsSchemaAsync( - this DataOptions dataOptions, string? schema = null, - CancellationToken token = default); +IImmediateJobsStorageBuilder UseLinqToDB(string? schema = null); +Task CreateImmediateJobsSchemaAsync( + this TContext context, string? schema = null, + CancellationToken cancellationToken = default); -ImmediateJobsOptions UseRedis( - string configuration, Action? configure = null); -ImmediateJobsOptions UseRedis( - IConnectionMultiplexer connection, Action? configure = null); +IImmediateJobsRedisBuilder UseRedis(); +IImmediateJobsRedisBuilder ConfigureRedis( + Action configure); +IImmediateJobsRedisBuilder ConfigureRedis( + Action> configure); ``` -`RedisJobStorageOptions` exposes `Database = -1` and `KeyPrefix = "immediate-jobs"`. +`UseLinqToDB` requires a registered `DataConnection` type. The schema helper extends that +connection type. `UseRedis` requires a registered `IConnectionMultiplexer` and selects distributed +mode. `RedisJobStorageOptions` exposes `Database = -1` and `KeyPrefix = "immediate-jobs"`. ## NodaTime @@ -343,23 +413,41 @@ it exposes `Services`, `Storage`, `TimeProvider`, and `Batches`. Operations are `AssertContinuationReleasedAfterAsync`, `AssertCascadeSkippedAsync`, `AssertCascadeCancelledAsync`, and `RunThroughPipelineAsync`. +`JobStorageConformanceSuite.GetCases(StorageCapabilities)` returns an independent +`JobStorageConformanceTestCase` for each selected storage behavior. The suite works with any test +framework. Each case exposes `Name`, `RequiredCapabilities`, and +`RunAsync(IServiceProvider, CancellationToken)`. It checks the storage registration and reported +features before running. Tests that depend on time require a `FakeTimeProvider` registered as +`TimeProvider`. `JobStorageConformanceSuite.AllCasesByName` is a case-insensitive map of every known +case by name. + ## Custom storage contracts -| Interface | Atomic responsibilities | -| ---------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------- | -| `IJobStorage` | Initialize; enqueue; lease/acquire/renew; persist/query execution history; complete/fail; status; cancel/retry/delete/purge; heartbeat and health. | -| `IRecurringJobStorage` | Upsert/remove/pause/resume schedules; identify due rows; uniquely materialize each occurrence; reconcile obsolete code-defined schedules. | -| `IJobGraphStorage` | Atomically enqueue batches/edges; settle and release/skip dependencies; add mid-run members; monitor/cancel/delete/purge graphs. | -| `IJobStorageReplica` | Restore durable records and mirror explicit acquisitions for the single-server wrapper. | - -`StorageCapabilities` flags are `Queue`, `Recurring` and `Graph`; call -`storage.GetCapabilities()` to detect the optional interfaces. Low-level `JobRecord`, acquisition, -definition and graph persistence records are provider contracts, not application scheduling APIs. -`IJobStorage.RetryAsync` accepts `Failed` and `Scheduled`: a scheduled invocation is moved to -`Pending` immediately while retaining its attempt count and latest failure details. -`IJobStorage.CancelAsync` accepts any non-terminal state, while `DeleteAsync` accepts terminal -states only. Worker-owned telemetry, renewal, completion and failure are fenced by job ID, -execution number and worker ID; graph expansion is fenced by job ID and execution number. An -expired or cancelled attempt therefore cannot mutate a newer durable state. Custom providers must -implement `QueryJobExecutionsAsync(JobExecutionQuery, ...)` and retain execution rows for the -lifetime of their owning job or batch. +| Interface | Purpose | +| ------------------------- | -------------------------------------------------------------------------------------------------- | +| `IJobStorage` | Store, claim, renew, finish, retry, cancel, delete and query jobs; report worker health. | +| `IRecurringJobStorage` | Store schedules, find due runs, create each run once and remove obsolete code-defined schedules. | +| `IJobGraphStorage` | Store and update batches and continuations; add jobs while a batch runs; query and manage batches. | +| `IFairQueueStorage` | Share available work fairly across groups. | +| `IJobStorageReplica` | Claim the exact job IDs selected by the single-server in-memory queue. | +| `IJobGraphStorageReplica` | Load incoming continuation links when a single-server worker starts. | + +`StorageCapabilities` flags are `Queue`, `Recurring`, `Graph`, `FairQueues`, and `Replica`. Call +`storage.GetCapabilities()` to report these features. `Replica` represents only +`IJobStorageReplica`. Single-server storage also requires `IJobGraphStorageReplica`, +`IRecurringJobStorage`, and `IJobGraphStorage`. + +`JobRecord` and the other storage record types are for provider authors, not ordinary scheduling +or monitoring code. + +`IJobStorage.RetryAsync` accepts `Failed` and `Scheduled`. A scheduled job moves to `Pending` +immediately without changing its attempt count or latest failure details. + +`IJobStorage.CancelAsync` accepts jobs that have not finished. `DeleteAsync` accepts only final +states. Storage must reject updates from an older attempt after its lease expires or it is +cancelled. Worker updates match the job ID, execution number and worker ID. Graph changes match the +job ID and execution number. + +Custom providers must implement +`QueryJobExecutionsAsync(JobExecutionQuery, ...)` and keep execution records until the owning job +or batch is deleted or purged. diff --git a/src/content/docs/Immediate.Jobs/batches-and-continuations.md b/src/content/docs/Immediate.Jobs/batches-and-continuations.md index d45a50d..5c98d4c 100644 --- a/src/content/docs/Immediate.Jobs/batches-and-continuations.md +++ b/src/content/docs/Immediate.Jobs/batches-and-continuations.md @@ -1,6 +1,6 @@ --- title: Batches and continuations -description: Build atomic job graphs, chains, fan-out/fan-in and dynamically expanded workflows. +description: Create batches, continuations, parallel branches and workflows that add jobs while running. order: 7 group: Guides --- @@ -9,19 +9,17 @@ group: Guides import { Callout } from '$lib/components/docs'; -Batches persist jobs and their dependency edges atomically. They require a graph-capable -provider; Redis exposes queue and recurring capabilities only. Resolve the scoped -`IJobBatchScheduler` from DI, normally through constructor injection alongside the generated job -schedulers. +Batches save jobs and their dependencies in one operation. They require storage with graph +support, which Redis does not provide. Inject the scoped `IBatchScheduler` alongside the generated +job schedulers. -## Atomic workflow graph +## Create a batch -The constructor below makes every receiver explicit: `batches` is the runtime batch scheduler; -the other parameters are nested scheduler types generated for their corresponding job classes. +Inject `IBatchScheduler` and the generated scheduler for each job in the workflow: ```csharp public sealed class ImportWorkflow( - IJobBatchScheduler batches, + IBatchScheduler batches, ImportData.Scheduler import, BuildIndex.Scheduler index, NotifyOwner.Scheduler notify, @@ -65,9 +63,9 @@ public sealed class ImportWorkflow( } ``` -Within an open batch, `AddToBatch` and `AddToBatchAt` only buffer records. Continuations built from -their handles remain in the same buffer. `CommitAsync` performs one atomic graph write and returns -a `BatchHandle`; nothing becomes visible before commit. +Within an open batch, `AddToBatch` and `AddToBatchAt` keep jobs in memory. Continuations created +from their handles stay in the same batch. `CommitAsync` saves the entire batch in one operation +and returns a `BatchHandle`. Nothing is visible before the commit. `Begin()` returns the in-memory buffer shown above. Always dispose it: disposal without commit abandons the buffer. A batch can commit only once and cannot be modified after commit. As an @@ -81,16 +79,15 @@ BatchHandle handle = await workflow.StartAsync(importId, cancellationToken); await batches.CancelAsync(handle, cancellationToken); ``` -This includes scheduled, active and continuation-waiting members, and the aggregate batch becomes -`Cancelled` after its members settle. Cancelling an active member records cancellation durably but -does not forcibly stop handler code already running in process; stale worker completion is fenced -from changing the terminal result. +This includes scheduled, active and continuation-waiting jobs. The batch becomes `Cancelled` after +every job reaches a final state. Cancelling an active job saves the cancellation but does not +forcibly stop handler code that is already running. If that code finishes later, it cannot +overwrite the cancelled result. -Failures before `CommitAsync` begins write nothing. Once commit begins, however, the batch is -closed even when the call throws, and a transport failure can leave the durable outcome unknown: -storage may have committed the graph before the caller lost the response. Do not retry the same -`JobBatch`; an operation that rebuilds and commits another batch needs application-level -idempotency or duplicate tracking. +A failure before `CommitAsync` begins saves nothing. Once the commit begins, the batch closes even +if the call throws. If the storage connection fails during the commit, the caller may not know +whether the batch was saved. Do not reuse the same `Batch`. If you create another batch, guard +against running the work twice. Batch members can carry the same fair-queue group IDs as ordinary scheduled work: @@ -102,9 +99,9 @@ var grouped = import.AddToBatchInGroup(batch, new(importId), tenantId); var groupedAt = import.AddToBatchAt(batch, new(importId), runAt, tenantId); ``` -`AddToBatchInGroup` also accepts an optional delay. Whitespace group IDs are normalized to no -group, the 128-character limit still applies, and the configured provider must support fair -acquisition for the group to affect dispatch order. +`AddToBatchInGroup` also accepts an optional delay. A blank group ID means no group, and group IDs +cannot exceed 128 characters. The group changes scheduling order only when the storage provider +supports fair queues. ## Chains, fan-out and fan-in @@ -163,8 +160,8 @@ work relates to the current job's existing continuations: | `BesideContinuations` | Current batch | Unchanged; the new job forms a parallel branch. | | `BeforeContinuations` (default) | Current batch | They also wait for the new job, creating an additive dependency. | -The `BeforeContinuations` splice keeps each existing dependency on the current job and adds a -dependency on the new job. Existing continuations therefore wait for both jobs. +With `BeforeContinuations`, each existing follow-up job waits for both the current job and the new +job. @@ -174,7 +171,9 @@ except for detached scheduling, the current job must belong to a batch. `IJOB001 -Monitor a graph through `IJobBatchMonitor.GetStatusAsync`, `QueryMembersAsync` and `GetGraphAsync`. -`BatchStatus` counts succeeded, failed, cancelled and skipped members separately; a batch can +Use the scoped `JobMonitor` to read a graph. Call `GetBatchAsync`, `QueryBatchMembersAsync`, or +`GetBatchGraphAsync`. These methods return `null` when storage does not support graphs. +`BatchStatus` counts succeeded, failed, cancelled and skipped members separately. A batch can succeed when every executed member succeeded even if conditional branches were skipped. The -dashboard exposes the same progress and workflow states alongside batch cancel/delete operations. +concrete monitor also provides `CancelBatchAsync` for jobs that have not finished and +`DeleteBatchAsync` for a completed batch. The dashboard offers the same actions. diff --git a/src/content/docs/Immediate.Jobs/choosing-storage.md b/src/content/docs/Immediate.Jobs/choosing-storage.md index 66af560..f41de2c 100644 --- a/src/content/docs/Immediate.Jobs/choosing-storage.md +++ b/src/content/docs/Immediate.Jobs/choosing-storage.md @@ -1,53 +1,62 @@ --- title: Choosing storage -description: Choose an Immediate.Jobs topology and provider by durability, scale and capability. +description: Choose storage by durability, worker count and supported job features. order: 10 group: Guides --- -Storage choice has two dimensions: the provider holds records; the topology decides whether memory -or that provider is authoritative. +Choose both a storage provider and a mode. The provider stores job data. The mode controls whether +workers coordinate through memory or through the provider. -| Topology | Authority | Processes | Durability | Use for | -| -------------- | --------------------------------------- | ----------: | ------------------------ | ----------------------------------------- | -| `InMemory` | Process memory | One | None | Unit tests, local demos, disposable work. | -| `SingleServer` | Memory with synchronous durable replica | Exactly one | Durable restart recovery | Low-latency single-instance services. | -| `Distributed` | Durable provider | One or more | Durable coordination | Scale-out and high availability. | +| Mode | Where jobs are coordinated | Worker processes | Survives restart | Use for | +| -------------- | -------------------------------- | ---------------- | ---------------- | ----------------------------------------- | +| `InMemory` | Current process | One | No | Unit tests, local demos, disposable work. | +| `SingleServer` | Memory backed by durable storage | Exactly one | Yes | Low-latency single-instance services. | +| `Distributed` | Storage provider | One or more | Yes | Scale-out and high availability. | -Calling a durable provider selects single-server mode unless you explicitly call -`UseDistributed()`. `UseRedis` always selects distributed mode. Never point two processes at the -same single-server replica: each believes its private memory is authoritative and drift detection -will fail. +A durable SQL provider uses single-server mode unless you call `UseDistributed()`. Redis always +uses distributed mode. Do not connect two scheduler processes to the same single-server storage; +the mode expects exactly one process and fails when it detects another. -## Capability matrix +## Supported features -| Provider | Queue | Recurring | Graph | Fair groups | Topologies | +| Provider | Queue | Recurring | Graph | Fair groups | Modes | | ------------ | :---: | :-------: | :---: | :---------: | -------------------------- | | In-memory | ✓ | ✓ | ✓ | ✓ | In-memory only | | EF Core SQL | ✓ | ✓ | ✓ | ✓ | Single-server, distributed | | LinqToDB SQL | ✓ | ✓ | ✓ | ✓ | Single-server, distributed | | Redis | ✓ | ✓ | — | — | Distributed | -Queue capability includes ordinary scheduling, execution history and job monitoring. Recurring -adds durable schedule reconciliation/materialization. Graph adds atomic batches, dependencies, -continuations and batch monitoring. The dashboard hides or returns 404 for unsupported graph -views. +Queue support includes scheduling, execution history and job monitoring. Recurring support stores +schedules and creates runs when they are due. Graph support adds batches, dependencies, +continuations and batch monitoring. The dashboard hides graph views when storage does not support +them. ## Tradeoffs -- In-memory is fastest and deterministic, but a restart loses pending jobs and history. -- Single-server acquires from memory and writes every transition to a full-capability SQL replica. - Startup restores the durable snapshot. It cannot provide multi-process failover. -- Distributed SQL coordinates leases, recurring schedules, graph transitions and fair-group - cursors in the database and is the full-featured scale-out option. +- In-memory is fast and predictable in tests, but a restart loses pending jobs and history. +- Single-server selects work in memory and writes every change to SQL. It restores that state after + a restart but cannot fail over to another process. +- Distributed SQL coordinates workers through the database. It supports multiple processes and + all job features. - Redis offers efficient distributed queues and recurring work, but not batches, continuations or - fair-group acquisition. + fair queues. ## A custom provider -Implement `IJobStorage` for queue capability. Add `IRecurringJobStorage` and/or `IJobGraphStorage` -only when their atomicity contracts are honored. Implement `IJobStorageReplica` as well to qualify -for single-server mode. Providers must initialize idempotently, claim due work atomically, enforce -worker ownership and leases, make recurring materialization unique, paginate monitoring, tolerate -repeated async disposal, and make graph commit/release/cascade transitions atomic. See the compact -contract map in [API reference](/docs/Immediate.Jobs/api-reference#custom-storage-contracts). +Implement `IJobStorage` to support queues. Add `IRecurringJobStorage`, `IJobGraphStorage`, and +`IFairQueueStorage` only for features the provider supports. + +Single-server storage needs two extra interfaces for restart recovery. `IJobStorageReplica` +claims the exact job IDs selected by the in-memory queue. `IJobGraphStorageReplica` loads incoming +continuation links at startup. A provider must implement both interfaces, plus recurring and graph +support, to use single-server mode. + +Starting or disposing the provider more than once must be safe. It must save each claim, recurring +run and graph change in one operation so workers cannot create duplicates or overwrite each other. +It must also enforce leases and worker ownership, and return monitoring results in pages. + +Run the `JobStorageConformanceSuite` from `Immediate.Jobs.Testing` with the same service +registration an application would use. Select the tests that match the provider's features. See +[Testing jobs](/docs/Immediate.Jobs/testing-jobs#test-a-storage-provider) and the contract summary +in [API reference](/docs/Immediate.Jobs/api-reference#custom-storage-contracts). diff --git a/src/content/docs/Immediate.Jobs/configuring-storage-providers.md b/src/content/docs/Immediate.Jobs/configuring-storage-providers.md index 0b2da3b..986f0c1 100644 --- a/src/content/docs/Immediate.Jobs/configuring-storage-providers.md +++ b/src/content/docs/Immediate.Jobs/configuring-storage-providers.md @@ -1,6 +1,6 @@ --- title: Configuring storage providers -description: Configure in-memory, EF Core, LinqToDB and Redis storage and own their schemas correctly. +description: Configure built-in storage providers and manage their database schemas. order: 11 group: Guides --- @@ -12,11 +12,12 @@ group: Guides ## In-memory ```csharp -builder.Services.AddMyAppJobs(options => options.UseInMemory()); +builder.Services.AddMyAppJobs() + .ConfigureStorage(storage => storage.UseInMemory()); ``` -This is also the default when no storage is selected. It is non-durable and single-node but -implements recurring, graph and fair-queue behavior for development and tests. +Select in-memory storage explicitly. It keeps data in one process and loses it on restart, but it +supports every job feature. Use it for development and tests. ## Entity Framework Core @@ -39,8 +40,10 @@ builder.Services.AddDbContextFactory(db => // db.UseSqlite(jobsConnectionString); // SQLite // db.UseSqlServer(jobsConnectionString); // SQL Server -builder.Services.AddMyAppJobs(options => - options.UseEntityFrameworkCore()); +builder.Services.AddMyAppJobs() + .ConfigureStorage(storage => storage + .UseEntityFrameworkCore() + .UseSingleServer()); public sealed class AppDbContext(DbContextOptions options) : DbContext(options) { @@ -90,22 +93,36 @@ dotnet add package Immediate.Jobs.LinqToDB --prerelease ``` ```csharp +using LinqToDB; +using LinqToDB.Data; +using LinqToDB.Extensions.DependencyInjection; + var dataOptions = new DataOptions().UsePostgreSQL(connectionString); // new DataOptions().UseSQLite(connectionString); // new DataOptions().UseSqlServer(connectionString); -await dataOptions.CreateImmediateJobsSchemaAsync( - schema: "background", // must be null for SQLite - CancellationToken.None -); +builder.Services.AddLinqToDBContext(() => dataOptions); + +await using (var connection = new JobsDataConnection(dataOptions)) +{ + await connection.CreateImmediateJobsSchemaAsync( + schema: "background", // must be null for SQLite + CancellationToken.None + ); +} + +builder.Services.AddMyAppJobs() + .ConfigureStorage(storage => storage + .UseLinqToDB(schema: "background") + .UseSingleServer()); -builder.Services.AddMyAppJobs(options => - options.UseLinqToDB(dataOptions, schema: "background")); +public sealed class JobsDataConnection(DataOptions options) : DataConnection(options); ``` -The application owns `DataOptions`, the matching ADO.NET driver and schema lifecycle. The helper -supports SQLite (without a named schema), PostgreSQL and SQL Server and creates the tables and -indexes for a fresh database. +Register the `DataConnection` type with dependency injection. Jobs resolves it when storage work +starts. The application owns `DataOptions`, the matching ADO.NET driver and the database schema. +The helper supports SQLite (without a named schema), PostgreSQL and SQL Server. It creates the +tables and indexes for a new database. ## Redis @@ -113,31 +130,44 @@ indexes for a fresh database. dotnet add package Immediate.Jobs.Redis --prerelease ``` -Pass a configuration string when Jobs should own the connection: +Register an `IConnectionMultiplexer`, then select Redis storage: ```csharp -builder.Services.AddMyAppJobs(options => options.UseRedis( - "localhost:6379", - redis => - { - redis.Database = 1; - redis.KeyPrefix = "billing-jobs"; - } -)); +using StackExchange.Redis; + +builder.Services.AddSingleton(_ => + ConnectionMultiplexer.Connect("localhost:6379")); + +builder.Services.AddMyAppJobs() + .ConfigureStorage(storage => storage + .UseRedis() + .ConfigureRedis(redis => + { + redis.Database = 1; + redis.KeyPrefix = "billing-jobs"; + })); ``` -Or pass an application-owned `IConnectionMultiplexer`; the provider will not dispose it. The -configuration-string overload owns and disposes its connection. `Database` defaults to `-1` -(server default), and `KeyPrefix` defaults to `immediate-jobs`. Prefixes cannot contain braces -because the provider adds its own Redis Cluster hash tag for atomic Lua operations. +The provider uses the registered connection and does not dispose it. The dependency injection +container disposes the connection in this example because it creates the singleton. If you +register an existing instance, its owner must dispose it. `Database` defaults to `-1` (server +default), and `KeyPrefix` defaults to `immediate-jobs`. The prefix cannot contain `{` or `}` because +Jobs uses those characters internally. Jobs validates these options at startup. + +`ConfigureRedis` also accepts an `OptionsBuilder` action. Use it when you +need configuration binding. Redis always selects distributed mode and supports queue plus recurring capabilities. It does not support graph workflows or fair queues. - +Call `ConfigureStorage` exactly once. With EF Core or LinqToDB, choose `UseSingleServer()` for one +scheduler process or `UseDistributed()` for more than one. Jobs defaults to single-server mode +when neither is selected. Redis always uses distributed mode. + + -Storage initialization is idempotent provider startup, not schema creation. Keep every -Immediate.Jobs provider package at the same preview revision as the core package, and create test -databases from the current EF model or `CreateImmediateJobsSchemaAsync` helper. +Starting Jobs does not create or update a database schema. Keep every Immediate.Jobs provider +package at the same preview version as the core package. Create test databases from the current EF +model or with `CreateImmediateJobsSchemaAsync`. diff --git a/src/content/docs/Immediate.Jobs/dashboard-and-monitoring.md b/src/content/docs/Immediate.Jobs/dashboard-and-monitoring.md index a4ed6f0..32748ca 100644 --- a/src/content/docs/Immediate.Jobs/dashboard-and-monitoring.md +++ b/src/content/docs/Immediate.Jobs/dashboard-and-monitoring.md @@ -1,20 +1,15 @@ --- title: Dashboard and monitoring -description: Secure the embedded dashboard and use its HTTP and programmatic monitoring APIs. +description: Secure the dashboard and monitor or manage jobs through its UI, API and JobMonitor. order: 13 group: Guides --- - - ```bash dotnet add package Immediate.Jobs.Dashboard --prerelease ``` -Register the dashboard's generated API handlers before building the application, then map the -embedded UI and JSON/SSE API: +Configure and register the dashboard before building the application. Then map its UI and API: ```csharp using Immediate.Jobs.Dashboard; @@ -22,73 +17,79 @@ using Immediate.Jobs.Dashboard; var traceExplorer = new Uri("https://traces.example/"); var logExplorer = new Uri("https://logs.example/"); -builder.Services.AddImmediateJobsDashboard(options => -{ - _ = options.RequireAuthorization("operations"); - _ = options.AddTelemetryLink( +builder.Services.AddMyAppJobs() + .ConfigureStorage(storage => storage.UseInMemory()) + .AddImmediateJobsDashboard() + .ConfigureDashboard(options => options.AuthorizationPolicy = "operations") + .AddTelemetryLink( "View execution trace", JobTelemetryLinkKind.Trace, context => context.Execution?.ExecutionTraceId is { } traceId ? new(traceExplorer, $"trace/{traceId}") : null - ); - _ = options.AddTelemetryLink( + ) + .AddTelemetryLink( "View execution logs", JobTelemetryLinkKind.Logs, context => context.Execution is { } execution ? new(logExplorer, $"search?jobId={Uri.EscapeDataString(context.Job.Id)}&attempt={execution.Attempt}") : null - ); - _ = options.AddTelemetryLink( + ) + .AddTelemetryLink( "View all retry logs", JobTelemetryLinkKind.Logs, context => context.Execution is null ? new(logExplorer, $"search?jobId={Uri.EscapeDataString(context.Job.Id)}") : null ); -}); var app = builder.Build(); app.MapImmediateJobsDashboard("/jobs"); ``` -Without `RequireAuthorization`, every dashboard endpoint is development-only by default and -returns 403 in other environments. For a trusted custom development environment, explicitly -disable this restriction when mapping the dashboard: +Chain `AddImmediateJobsDashboard` from the generated jobs registration. Set dashboard options with +`ConfigureDashboard`; the mapping call only selects the URL path. `ConfigureDashboard` also accepts +an `OptionsBuilder` action for configuration binding. Jobs validates +the settings when the host starts. + +By default, every dashboard endpoint is limited to the `Development` environment and returns 403 +elsewhere. Setting `AuthorizationPolicy` uses that policy instead of the environment check. For a +trusted custom development environment, you can remove the default restriction without setting a +policy: ```csharp -app.MapImmediateJobsDashboard("/jobs", options => - _ = options.AllowInAnyEnvironment() -); +builder.Services.AddMyAppJobs() + .ConfigureStorage(storage => storage.UseInMemory()) + .AddImmediateJobsDashboard() + .ConfigureDashboard(options => options.RestrictToDevelopmentEnvironment = false); + +var app = builder.Build(); +app.MapImmediateJobsDashboard("/jobs"); ``` -Treat the dashboard as an administrative surface: it exposes payloads, errors, identifiers and -mutations. Prefer `RequireAuthorization` whenever the dashboard is available outside a trusted -development environment. A named policy applies to UI assets and APIs together and remains -authoritative if `AllowInAnyEnvironment` is also configured. +Treat the dashboard as an administrative tool because it exposes job inputs, failures, IDs and +actions that change job state. Set `AuthorizationPolicy` whenever the dashboard is available +outside a trusted development environment. The policy protects both the UI and API. -The UI shows queue/state totals, including skipped work, recent history, jobs and details, -recurring schedules, scheduler servers, batches and workflow graphs. Graph views appear only for -graph-capable storage. +The UI shows queue and state totals, recent history, job details, recurring schedules, scheduler +servers and batches. It also shows workflow graphs when storage supports them. ## Dashboard UI ### Inspect jobs -The Jobs view lists durable invocations and their current state. Select a job to open its dedicated -detail route with payload data and a newest-first, collapsible timeline of retained executions. -Each execution records its attempt, state, worker, acquisition/start/completion times, trace/span -identifiers and full failure text. Failed jobs offer **Retry**; scheduled first attempts and delayed -retries offer **Run now**, which moves the existing invocation to `Pending` without changing its -attempt or failure history. Every non-terminal job offers a confirmation-backed **Cancel** action. +The Jobs view lists saved jobs and their current state. Select a job to see its payload and saved +attempts, newest first. Each attempt includes its state, worker, timing, trace IDs and failure text. +Failed jobs offer **Retry**. Scheduled jobs and delayed retries offer **Run now**, which moves the +same job to `Pending` without changing its attempt count or failure history. Jobs that have not +finished offer **Cancel** with a confirmation step. ### Follow batch workflows -The Batches view visualizes the jobs in a batch and the continuations between them. Progress and -workflow nodes distinguish skipped conditional branches from explicitly cancelled work. Select a -node to inspect that job without losing the surrounding workflow context. Executing batches offer -a confirmation-backed **Cancel** action from their workflow details. +The Batches view shows the jobs in a batch and their dependencies. It distinguishes skipped +branches from cancelled work. Select a job without leaving the workflow view. A running batch +offers **Cancel** with a confirmation step.
- A batch workflow graph with continuation relationships and job details available in place. + A batch workflow showing each job and its dependencies.
## Telemetry links -`AddTelemetryLink` adds application-defined destinations to job and execution details. The -dashboard evaluates its synchronous URL factory through the job-level or exact-execution telemetry -endpoint and supplies a `JobTelemetryLinkContext`. +`AddTelemetryLink` adds application-defined links to job and execution details. For each link, the +dashboard calls your URL function with a `JobTelemetryLinkContext`. | Argument | Purpose | | ----------- | ----------------------------------------------------------------------------------------------------- | @@ -126,15 +126,14 @@ endpoint and supplies a `JobTelemetryLinkContext`. | `kind` | `Trace` or `Logs`; controls how the dashboard identifies the link. | | `createUrl` | Builds the destination from `context.Job` and optional `context.Execution`; return `null` to hide it. | -For a job-level request, `context.Execution` is `null` and the execution fields on `context.Job` -describe the latest attempt. For an exact-execution request, `context.Execution` contains the -selected retained record; `context.Job.Attempt`, `ExecutionTraceId`, `ExecutionSpanId` and -`ExecutionStartedAt` also represent that same attempt. Use `Execution` for per-attempt trace/log -links, and use `Job.Id` when a destination should search across every retry. +For a job link, `context.Execution` is `null` and the execution fields on `context.Job` describe +the latest attempt. For an attempt link, `context.Execution` and the execution fields on +`context.Job` both describe the selected attempt. Use `Execution` for links to one attempt. Use +`Job.Id` when a destination should search across every retry. -Factories may return HTTP(S) or dashboard-relative URLs. Other absolute URI schemes are rejected -when the endpoint evaluates the link. Return `null` before an execution trace exists or whenever a -destination does not apply to the current record. +The URL function may return HTTP(S) or dashboard-relative URLs. Other absolute URI schemes are +rejected. Return `null` before an execution trace exists or whenever a destination does not apply +to the current record. ## HTTP endpoints @@ -142,50 +141,57 @@ All paths below are relative to the mapped prefix. | Method and path | Purpose | | -------------------------------------------------------------------- | ------------------------------------------------------------------------ | -| `GET /api/overview` | Monitoring snapshot and storage capabilities. | +| `GET /api/overview` | Current counts and supported storage features. | | `GET /api/jobs` | Filter by `state`, `queue`, `search`; `skip`; `take` 1–200 (default 50). | -| `GET /api/jobs/{jobId}` | Latest durable record. | -| `GET /api/jobs/{jobId}/executions` | Retained attempts newest first; `skip`; `take` 1–200 (default 50). | -| `GET /api/jobs/{jobId}/telemetry-links` | Configured job-level trace/log destinations. | -| `GET /api/jobs/{jobId}/executions/{executionNumber}/telemetry-links` | Configured destinations for one retained execution. | -| `POST /api/jobs/{jobId}/cancel` | Cancel non-terminal work. | +| `GET /api/jobs/{jobId}` | Latest saved record. | +| `GET /api/jobs/{jobId}/executions` | Saved attempts newest first; `skip`; `take` 1–200 (default 50). | +| `GET /api/jobs/{jobId}/telemetry-links` | Configured trace and log links for a job. | +| `GET /api/jobs/{jobId}/executions/{executionNumber}/telemetry-links` | Configured links for one saved attempt. | +| `POST /api/jobs/{jobId}/cancel` | Cancel a job that has not finished. | | `POST /api/jobs/{jobId}/retry` | Retry failed work or run scheduled work now. | | `GET /api/recurring` | Recurring schedules. | -| `POST /api/recurring/{name}/trigger` | Materialize an immediate invocation. | +| `POST /api/recurring/{name}/trigger` | Start one run now. | | `POST /api/recurring/{name}/pause` / `resume` | Change schedule state. | -| `GET /api/servers` | Worker heartbeat snapshots. | +| `GET /api/servers` | Recently active workers. | | `GET /api/batches` | Filter by `state`, `skip`, `take` 1–500 (default 100). | -| `GET /api/batches/{id}` | Aggregate status. | -| `GET /api/batches/{id}/members` | Filtered/paged member status. | -| `GET /api/batches/{id}/graph` | Dependency graph. | -| `POST /api/batches/{id}/cancel` | Cascade-cancel unsettled members. | -| `DELETE /api/batches/{id}` | Delete a terminal graph. | +| `GET /api/batches/{id}` | Batch status. | +| `GET /api/batches/{id}/members` | Filter and page through jobs in a batch. | +| `GET /api/batches/{id}/graph` | Jobs and dependencies in a batch. | +| `POST /api/batches/{id}/cancel` | Cancel jobs in a batch that have not finished. | +| `DELETE /api/batches/{id}` | Delete a completed batch. | | `GET /api/events` | SSE `state` snapshots at `UpdateInterval`. | | `GET /api/batches/{id}/stream` | SSE `status` and `graph` events on change. | -Successful cancel, retry, pause, resume and batch mutations return `204`; recurring trigger returns -`202`. A request returns `400` Validation Problem Details for invalid route or paging values, `404` -when its job, execution, batch or recurring schedule does not exist (or the provider lacks the -required capability), and `409` when a mutation targets a resource whose lifecycle state does not -allow the operation. +Actions return these status codes: -SSE sends `retry: 3000`, disables proxy buffering and ends when the request is aborted. It is a -poll-backed live view, not a durable event log; clients must refresh after reconnecting. +| Status | Meaning | +| ------ | --------------------------------------------------------------------------- | +| `202` | A recurring run was accepted. | +| `204` | A cancel, retry, pause, resume or batch action succeeded. | +| `400` | A route or paging value is invalid. | +| `404` | The item does not exist, or storage does not support the requested feature. | +| `409` | The item's current state does not allow the action. | -## Programmatic monitoring +Live endpoints use server-sent events (SSE). They send `retry: 3000`, disable proxy buffering and +close when the request is aborted. The server checks for changes on a timer instead of keeping an +event log, so clients should reload after reconnecting. -Inject scoped `IJobMonitor` and call `GetJobAsync`. With a graph provider, inject -`IJobBatchMonitor` and call `GetStatusAsync`, `QueryMembersAsync`, or `GetGraphAsync`. These are -read-only contracts suitable for application status endpoints. Custom operational views can query -newest-first attempts through `IJobStorage.QueryJobExecutionsAsync`; the execution history remains -with its owning job or batch until that aggregate is deleted or purged. +## Use JobMonitor in code -Monitoring snapshots include only scheduler servers whose last heartbeat is at most two minutes -old. SQL providers prune stale server rows on later heartbeats, while Redis expires their hashes. +Inject the scoped `JobMonitor` for custom status pages and administrative endpoints. It reads +snapshots, jobs, saved attempts, batches, batch members and workflow graphs. It can also cancel or +retry jobs, cancel or delete batches, and pause, resume or trigger recurring schedules. +The dashboard uses this same service. - +`IJobMonitor` exposes only the read methods. Use it when code needs no management commands or a test +needs a simple replacement. Do not use `IJobStorage` for application monitoring or management. It +is for storage providers and the Jobs runtime. -The dashboard calls storage query APIs directly. Apply paging and authorization to any custom -monitoring endpoint too; payload and exception data can contain business-sensitive values. +`QueryExecutionsAsync` returns attempts newest first. Execution history remains with its job or +batch until that record is deleted or purged. Batch reads return `null` when the provider does not +support graphs, so non-batch monitoring still works with Redis and other queue-only providers. - +Monitoring snapshots include only scheduler servers whose last heartbeat is at most two minutes +old. SQL providers remove stale server rows on later heartbeats, while Redis expires them. Apply +paging and authorization to custom endpoints because payload and failure data can contain +business-sensitive values. diff --git a/src/content/docs/Immediate.Jobs/diagnostics.md b/src/content/docs/Immediate.Jobs/diagnostics.md index bb9fb94..1840374 100644 --- a/src/content/docs/Immediate.Jobs/diagnostics.md +++ b/src/content/docs/Immediate.Jobs/diagnostics.md @@ -1,13 +1,12 @@ --- title: Diagnostics -description: Current Immediate.Jobs analyzer errors and warnings, their locations and remediations. +description: Look up Immediate.Jobs analyzer errors, warnings and common runtime failures. order: 18 group: Diagnostics --- -Immediate.Jobs currently uses one zero-padded diagnostic sequence. The analyzer package exposes -the following IDs; method-shape, `partial` and other handler diagnostics come from -Immediate.Handlers separately. +The Immediate.Jobs analyzer reports the IDs below. Immediate.Handlers separately reports invalid +handler methods, missing `partial` modifiers and other handler problems. | ID | Severity | Trigger and location | Fix | | ---------- | -------- | ------------------------------------------------------------------------------------------------------------ | ----------------------------------------------------------------------------------------------- | @@ -26,25 +25,31 @@ Immediate.Handlers separately. | `IJOB0013` | Error | Payload graph cannot receive generated JSON metadata; offending request member/type. | Use supported concrete values, one-dimensional arrays, `List` or `Dictionary`. | | `IJOB0014` | Error | Context graph cannot receive generated JSON metadata; offending context member/type. | Apply the same AOT-safe shape rules as a job payload. | | `IJOB0015` | Warning | `AddToBatchAsync(JobDetails, ..., ContinuationOptions.Detached)`; the `Detached` argument. | Use `ScheduleAfter` for detached work or a batch-joining option. | +| `IJOB0016` | Warning | A `[QueueDefinition]` has no job assigned to it; queue type. | Remove the unused definition or attach a job with `[UsesQueue]`. | ## Related runtime failures -Some facts depend on runtime values or durable state and cannot be diagnosed at compile time: +Some problems depend on settings or stored data, so an analyzer cannot catch them: - duplicate context extractor keys throw `ImmediateJobException` while capturing; - negative delays and over-128-character group IDs throw argument exceptions; -- a non-positive dashboard update interval throws `InvalidOperationException`, and mapping the - dashboard without first calling `AddImmediateJobsDashboard` also throws; -- invalid dynamic cron/time zones fail when adding/updating the schedule; -- malformed persisted recurring schedules are logged individually without blocking other - schedules or ordinary job acquisition; +- mapping the dashboard without first calling `AddImmediateJobsDashboard` throws + `InvalidOperationException`; +- an invalid dynamic cron expression or time zone fails when the schedule is added or updated; +- Jobs logs and skips a malformed stored recurring schedule without blocking other schedules or + queued jobs; - graph operations on Redis or another queue-only provider throw `NotSupportedException`; - fair acquisition on Redis throws `NotSupportedException` when `UseFairQueues` is enabled; -- single-server mode rejects providers without replica, recurring and graph capabilities and - detects multiple-process replica drift; -- unknown stored job names fail terminally because no generated definition can execute them; -- unknown context slices are logged and skipped so rolling deployments can continue; -- dashboard route/paging validation returns HTTP 400 Validation Problem Details; +- omitting `ConfigureStorage` fails validation when the host starts; +- conflicting storage selections or a second `ConfigureStorage` call throw `ImmediateJobException` + during registration; +- invalid runtime, fair-queue, dashboard, or Redis options fail validation when the host starts; +- single-server mode requires `IJobStorageReplica`, `IJobGraphStorageReplica`, recurring, and graph + support; +- single-server mode stops if it detects another scheduler process using the same durable storage; +- an unknown stored job name fails because no generated job definition can run it; +- Jobs logs and skips unknown context data so rolling deployments can continue; +- invalid dashboard route or paging values return HTTP 400 with validation details; - dashboard mutations return HTTP 404 for an unknown job, batch or recurring schedule; - retry/delete/cancel operations reject incompatible lifecycle states with `ImmediateJobException` (HTTP 409 in the dashboard). diff --git a/src/content/docs/Immediate.Jobs/enqueueing-and-scheduling.md b/src/content/docs/Immediate.Jobs/enqueueing-and-scheduling.md index 3b5bb7d..4011c3e 100644 --- a/src/content/docs/Immediate.Jobs/enqueueing-and-scheduling.md +++ b/src/content/docs/Immediate.Jobs/enqueueing-and-scheduling.md @@ -41,9 +41,9 @@ await scheduler.ScheduleAtAsync(payload, shipAt, tenantId, cancellationToken); ``` Whitespace is normalized to no group. Group IDs longer than 128 characters are rejected. A -non-empty group is persisted even without `options.UseFairQueues()`; in that case it does not -affect order and the worker logs one warning. Enabling fair acquisition requires a supporting -provider; Redis rejects fair acquisition. +non-empty group is still stored when `UseFairQueues()` was not called on the registration builder, +but it does not affect order and the worker logs one warning. Fair scheduling requires a provider +that supports it; Redis does not. Because schedulers are scoped, a singleton worker creates a scope for each unit of work: @@ -66,7 +66,8 @@ Immediate.Jobs creates job and batch IDs before writing them to storage. The def platform uses Snowflake, ULID or another globally unique string format: ```csharp -builder.Services.AddMyAppJobs(options => options.UseInMemory()) +builder.Services.AddMyAppJobs() + .ConfigureStorage(storage => storage.UseInMemory()) .UseIdGenerator(); // ISnowflakeService is supplied and registered by your chosen Snowflake implementation. @@ -89,7 +90,7 @@ public sealed class SnowflakeIdGenerator(ISnowflakeService snowflakes) : IIdGene `UseIdGenerator()` registers `TGenerator` as a singleton, so its implementation and dependencies must be thread-safe. With distributed storage, configure Snowflake worker/node IDs so separate application instances cannot generate the same value. `IdKind` lets the generator -distinguish individual job invocations—including recurring occurrences—from atomic batches. +distinguish job runs, including recurring runs, from batches. Treat the result as opaque even when your generator adds a readable prefix. Applications should store and compare `JobHandle.Id` or `BatchHandle.Id`, not parse business meaning from their format. @@ -109,9 +110,9 @@ await scheduler.CancelAsync(handle, cancellationToken); Cancellation immediately persists `Cancelled`, including for scheduled, pending, continuation- waiting and active work. If a worker already owns the invocation, its in-process handler is not -forcibly interrupted; attempt fencing prevents that worker's later completion or failure from -overwriting the cancelled record. Cancelling an unknown handle fails as not found, and cancelling -a terminal invocation fails with `ImmediateJobException`. +forcibly interrupted. Storage rejects that worker's later completion or failure, so it cannot +overwrite the cancelled record. Cancelling an unknown handle fails as not found. Cancelling a +finished invocation fails with `ImmediateJobException`. The token passed to `CancelAsync` cancels the storage operation only. Likewise, cancellation tokens on scheduling calls do not become future execution-cancellation tokens. diff --git a/src/content/docs/Immediate.Jobs/how-it-works.md b/src/content/docs/Immediate.Jobs/how-it-works.md index 99c3f15..8523a1e 100644 --- a/src/content/docs/Immediate.Jobs/how-it-works.md +++ b/src/content/docs/Immediate.Jobs/how-it-works.md @@ -1,6 +1,6 @@ --- title: How it works -description: Follow a job from Roslyn discovery through generated code, durable acquisition and scoped execution. +description: See how Immediate.Jobs finds, saves and runs a job. order: 17 group: Reference --- @@ -9,69 +9,71 @@ group: Reference The incremental generator discovers classes with `[Job]` and a valid Immediate.Handlers `[Handler]`. Analyzers validate names, queues, cron and execution settings, the exact -`HandleAsync` shape, context extractor contracts, and whether payload/context types can receive -source-generated JSON metadata. +`HandleAsync` shape, context extractors and whether generated JSON can represent the payload and +context types. For each job it emits `IJ...g.cs` containing: - a scoped nested `Scheduler` deriving from `JobScheduler`; -- an internal singleton `Invoker` that deserializes, restores context and enters the generated - Immediate.Handlers pipeline; +- an internal singleton `Invoker` that restores saved data and calls the Immediate.Handlers + pipeline; - a singleton `JobDefinition` factory with stable name, queue and execution policy; - a generated `JsonSerializerContext`/resolver for payload and context types; - registrations for the scheduler, invoker, extractors and definition. -At assembly level, `IJ.ServiceCollectionExtensions.g.cs` contains `AddXxxJobs`. It calls the -runtime registration once, registers queue definitions, and conditionally adds jobs selected by -tags. The assembly identifier and tags follow the same conventions as the other platform +At assembly level, `IJ.ServiceCollectionExtensions.g.cs` contains `AddXxxJobs` and the generated +`RecurringJobs` service. `AddXxxJobs` registers jobs and returns `IImmediateJobsBuilder`. +`RecurringJobs` can trigger payloadless jobs by name. Both types are placed in the project's +`RootNamespace`. Calling the registration method again does not duplicate jobs, queues or the +hosted worker. The assembly identifier and tags follow the same conventions as the other platform generators. ## Enqueue data flow 1. Application code resolves the scoped generated scheduler. -2. The scheduler captures the current trace link and opted-in context extractors. -3. It serializes payload/context with generated metadata, generates an ID, and builds a record with - stable job/queue names, due time and optional group/batch data. -4. Storage persists that record (or a batch buffers it until atomic commit). +2. The scheduler captures the current trace link and any selected context values. +3. It serializes the payload and context, creates an ID, and builds a record with the job name, + queue, due time and optional group or batch data. +4. Storage saves that record. An open batch holds it in memory until commit. 5. The scheduler returns a `JobHandle`; it does not wait for execution. ## Worker data flow -The hosted service initializes storage and recurring definitions, then builds acquisition requests -from queue priority plus node, queue and job capacity. Storage atomically changes eligible due work -to `Active`, assigns a worker/lease, increments attempts and creates a retained execution record. -Distributed providers coordinate this in the shared backend; single-server mode acquires from -memory and mirrors ownership to its durable replica. +The hosted service prepares storage and recurring schedules. It asks storage for due jobs based on +queue priority and the available capacity for the worker, queue and job. Storage marks each +selected job `Active`, assigns its worker and lease, increments its attempt number and saves an +execution record. It makes those changes in one operation. -For each acquired record the worker creates a consumer activity and logging scope, starts lease -renewal and a linked timeout token, then creates a fresh async DI scope. The generated invoker -deserializes context and payload, restores known slices, assigns `JobDetails` and resolves the -generated handler. The call enters Immediate.Handlers behaviors and ends at the private job method. +Distributed mode coordinates workers through shared storage. Single-server mode selects jobs in +memory and copies each claim to durable storage. -Success atomically closes the execution and records completion plus any buffered mid-execution -continuations. Failure closes the attempt with its full exception and either schedules a retry or -leaves a terminal failure. Telemetry, renewal and terminal updates carry the acquired attempt and -worker ID so stale owners cannot mutate a reacquired or explicitly cancelled job. Settling a graph -node releases eligible children or marks unselected branches `Skipped`. Release and recursive -skipping are committed in the same transaction as the parent's terminal transition. +For each selected job, the worker starts tracing, logging, lease renewal and its timeout. It also +creates a new dependency injection scope. The generated invoker restores the payload and context, +sets `JobDetails`, resolves the handler and runs its Immediate.Handlers behaviors. -## Recurring materialization +On success, Jobs closes the attempt and saves any follow-up jobs that the handler buffered. On +failure, it saves the full exception and either schedules a retry or leaves the job failed. Every +update includes the attempt number and worker ID. An old worker therefore cannot overwrite a job +that another worker acquired or a user cancelled. -Code-defined schedules are reconciled at startup. The recurring loop asks storage for due -schedules; the provider atomically materializes a uniquely keyed occurrence and advances its next -run. This keeps multiple distributed nodes from creating the same occurrence. Overlap policy is -evaluated against active occurrences of the same schedule; `Skip` persists a terminal skipped -occurrence so monitoring retains the scheduling decision. +When a batch job finishes, storage starts eligible child jobs and marks other branches `Skipped`. +It saves these changes together with the parent's result. + +## Recurring runs + +At startup, Jobs updates the code-defined schedules in storage. It then checks for schedules that +are due. Storage creates one job for each due time and advances the schedule in the same operation, +which prevents two workers from creating the same run. If the overlap policy is `Skip`, Jobs still +saves a skipped run so monitoring shows what happened. ## Generated JSON, trimming and Native AOT Schedulers and invokers call `IJobSerializer` overloads that receive generated -`JsonTypeInfo`. The generated resolver covers the payload graph and every opted-in context -type, so trimming does not need to preserve reflection-discovered constructors or properties. -Resolved payload metadata is cached per payload type for subsequent serialize/deserialize calls. -Unsupported shapes fail at compile time. The Native AOT sample publishes the same generated path; -custom serializers must honor the metadata overloads to retain this property. - -The runtime itself does not scan assemblies or use a service locator to discover jobs. Durable -job names, queue names, extractor keys and serialized contracts are nevertheless versioned data: -deploy changes compatibly or drain/transform durable records before removing them. +`JsonTypeInfo`. This generated information describes each supported payload and selected context +type, so a trimmed application does not need to keep constructors or properties found through +reflection. Jobs caches the information for later calls. Unsupported types fail at compile time. +The Native AOT sample uses the same path. Custom serializers must use these overloads too. + +The runtime does not scan assemblies to find jobs. Stored job names, queue names, context keys and +serialized data can outlive a deployment. Keep changes compatible, or drain or migrate old records +before removing their definitions. diff --git a/src/content/docs/Immediate.Jobs/introduction.md b/src/content/docs/Immediate.Jobs/introduction.md index 0139df4..4ffa109 100644 --- a/src/content/docs/Immediate.Jobs/introduction.md +++ b/src/content/docs/Immediate.Jobs/introduction.md @@ -34,8 +34,8 @@ main package in the project that declares the handlers: dotnet add package Immediate.Jobs --prerelease ``` -Choose a durable provider before production; in-memory storage is the automatic default when no -provider is selected. +Select storage during registration. Use in-memory storage for development and tests. Choose a +durable provider before production. ## Your first job @@ -63,11 +63,12 @@ Register handlers and jobs, then inject the generated scoped scheduler: ```csharp title="Program.cs" builder.Services.AddMyAppHandlers(); -builder.Services.AddMyAppJobs(options => options.UseInMemory()); +builder.Services.AddMyAppJobs() + .ConfigureStorage(storage => storage.UseInMemory()); ``` -The injected type is `SendWelcomeEmail.Scheduler`. The returned `JobHandle` is an opaque -identifier for monitoring and continuations—not evidence that the job completed. +The injected type is `SendWelcomeEmail.Scheduler`. The returned `JobHandle` identifies the saved +job for monitoring and continuations. It does not mean the job has finished. @@ -83,10 +84,10 @@ possible. Immediate.Jobs does not include a transactional outbox. - - - + + + - - + + diff --git a/src/content/docs/Immediate.Jobs/nodatime.md b/src/content/docs/Immediate.Jobs/nodatime.md index ef81daf..f7b6520 100644 --- a/src/content/docs/Immediate.Jobs/nodatime.md +++ b/src/content/docs/Immediate.Jobs/nodatime.md @@ -24,7 +24,8 @@ Register the integration with Jobs: using Immediate.Jobs.NodaTime; builder.Services.AddMyAppHandlers(); -builder.Services.AddMyAppJobs(options => options.UseInMemory()); +builder.Services.AddMyAppJobs() + .ConfigureStorage(storage => storage.UseInMemory()); builder.Services.AddImmediateJobsNodaTime(); ``` diff --git a/src/content/docs/Immediate.Jobs/observability-and-health.md b/src/content/docs/Immediate.Jobs/observability-and-health.md index 4f046d2..59bc0fe 100644 --- a/src/content/docs/Immediate.Jobs/observability-and-health.md +++ b/src/content/docs/Immediate.Jobs/observability-and-health.md @@ -1,6 +1,6 @@ --- title: Observability and health -description: Export Immediate.Jobs traces, metrics and structured logs, and register scheduler health checks. +description: Export traces, metrics and logs, and add scheduler health checks. order: 14 group: Guides --- @@ -14,12 +14,11 @@ builder.Services.AddOpenTelemetry() ``` Each execution creates a consumer activity named `job {job.name}` with `job.name`, `job.queue`, -`job.id` and `job.attempt`. Enqueue trace context is persisted and linked to the execution rather -than used as its parent, so asynchronous work remains causally visible without pretending to be a -single synchronous span. Every acquired attempt retains its trace/span IDs, timing, worker, outcome -and full failure text for the lifetime of its owning job or batch. The latest values remain on -`JobRecord` as the latest-execution projection, while the dashboard can build links for an exact -`JobExecutionRecord`. +`job.id` and `job.attempt`. Jobs saves the trace context from the scheduling call and links it to +the later execution trace. It does not make the scheduling call the parent because the job runs +asynchronously. Each attempt stores its trace and span IDs, timing, worker, outcome and full failure +text until its job or batch is deleted. `JobRecord` holds the latest attempt. The dashboard uses +`JobExecutionRecord` when it needs one specific attempt. ## Metrics @@ -33,27 +32,41 @@ and full failure text for the lifetime of its owning job or batch. The latest va | `queue.depth` | Observable gauge | none | | `workers.active` | Observable gauge | none | -The gauges are local runtime observations, not authoritative cluster totals. Use provider -monitoring snapshots for durable/cluster state. Alert on growing queue depth, exhausted failures, -retries and stale server heartbeats; interpret duration by job name and outcome. +The gauges describe the current process, not every worker. Use provider monitoring snapshots for +stored totals across workers. Alert on growing queue depth, final failures, retries and stale server +heartbeats. Compare duration by job name and outcome. ## Structured logs Worker logs carry the scope properties `JobName`, `QueueName`, `JobId` and `Attempt`. Events cover scheduler iteration failure, shutdown-drain timeout, unhandled worker errors, completion, retry, -attempt exhaustion and disabled optional capabilities. Include scopes in the logging exporter to -make these fields queryable. +attempt exhaustion and features that storage does not support. Include scopes in your logging +output when you want to search these fields. ## Health checks ```csharp -builder.Services.AddMyAppJobs(options => options.UseEntityFrameworkCore()) +builder.Services.AddMyAppJobs() + .ConfigureStorage(storage => storage + .UseEntityFrameworkCore() + .UseDistributed()) .AddHealthCheck(name: "my-app-jobs", tags: ["ready"]); -app.MapHealthChecks("/health/ready"); +app.MapHealthChecks("/health/ready", new HealthCheckOptions +{ + Predicate = registration => registration.Tags.Contains("ready"), + ResultStatusCodes = + { + [HealthStatus.Degraded] = StatusCodes.Status503ServiceUnavailable, + [HealthStatus.Unhealthy] = StatusCodes.Status503ServiceUnavailable, + }, +}); ``` -The check combines scheduler liveness with provider connectivity. Choose a `failureStatus` when -degraded versus unhealthy behavior matters to orchestration. The Aspire sample uses the same -OpenTelemetry sources, health registration and dashboard telemetry-link hooks; Immediate.Jobs -does not require Aspire and does not ship an Aspire-specific runtime package. +The check covers both the worker and its storage connection. Filter the readiness endpoint by the +tag passed to `AddHealthCheck`. It reports `Degraded` until the worker starts, so map that status to +HTTP 503 when the application should not receive traffic during startup. No extra options +registration is needed; the check and worker use the same validated settings. + +The Aspire sample shows the same tracing, metrics, health-check and dashboard-link setup. Aspire is +optional; Immediate.Jobs does not ship a separate Aspire runtime package. diff --git a/src/content/docs/Immediate.Jobs/queues-and-fairness.md b/src/content/docs/Immediate.Jobs/queues-and-fairness.md index c5c389b..1a8ea92 100644 --- a/src/content/docs/Immediate.Jobs/queues-and-fairness.md +++ b/src/content/docs/Immediate.Jobs/queues-and-fairness.md @@ -1,6 +1,6 @@ --- title: Queues and fairness -description: Define queues and combine priority, concurrency and fair-group scheduling. +description: Define queues, limit concurrency, set priority and share capacity across groups. order: 6 group: Guides --- @@ -33,15 +33,14 @@ on one scheduler node; zero is unbounded. Node-wide `MaxParallelJobs` and job-le Enable fairness globally and put a tenant/customer key on each scheduled invocation: ```csharp -builder.Services.AddMyAppJobs(options => -{ - options.UseFairQueues(fair => +builder.Services.AddMyAppJobs() + .UseFairQueues(options => options.Configure(fair => { fair.ConcurrencyShareThreshold = 0.10; fair.MinInflightForNoisy = 30; fair.GroupRoundRobin = true; - }); -}); + })) + .ConfigureStorage(storage => storage.UseInMemory()); await welcomeEmail.EnqueueAsync( new(userId, "v2"), @@ -50,16 +49,16 @@ await welcomeEmail.EnqueueAsync( ); ``` -Round-robin interleaves due groups. A group becomes noisy only after it has at least -`MinInflightForNoisy` active jobs and exceeds `ConcurrencyShareThreshold` of the queue's effective -capacity; quieter groups are then preferred. Ungrouped jobs remain eligible. Fairness affects -acquisition order, not durable priority or a job's retry policy. +Round-robin alternates between groups that have work ready. A group becomes noisy after it reaches +`MinInflightForNoisy` active jobs and uses more than `ConcurrencyShareThreshold` of the queue's +capacity. Jobs then favors quieter groups. Jobs without a group remain eligible. Fairness changes +which ready job runs next; it does not change priority or retry rules. -| Provider/topology | Fair groups | -| --------------------- | ---------------------------------------------------------------- | -| In-memory | Supported | -| EF Core / LinqToDB | Supported | -| Redis | Not supported; grouped acquisition is rejected | -| Single-server wrapper | Supported when its durable replica has full graph/fair semantics | +| Storage and mode | Fair groups | +| ------------------ | --------------------------------------------------------------- | +| In-memory | Supported | +| EF Core / LinqToDB | Supported | +| Redis | Not supported; enabling fair queues causes an error | +| Single-server | Supported when its durable provider meets the mode requirements | Queue and group names are persisted. Renaming either does not rename already-persisted work. diff --git a/src/content/docs/Immediate.Jobs/recurring-jobs.md b/src/content/docs/Immediate.Jobs/recurring-jobs.md index 3e7ffcb..bd4340b 100644 --- a/src/content/docs/Immediate.Jobs/recurring-jobs.md +++ b/src/content/docs/Immediate.Jobs/recurring-jobs.md @@ -1,6 +1,6 @@ --- title: Recurring jobs -description: Define and manage recurring schedules with cron expressions, time zones and overlap policies. +description: Create recurring schedules, run them on demand and control overlapping runs. order: 4 group: Guides --- @@ -8,7 +8,7 @@ group: Guides Recurring jobs are payloadless. A code-defined schedule lives on `[Job]`: ```csharp -[Handler, Job(Cron = "0 */5 * * * *", TimeZone = "Europe/Vienna")] +[Handler, Job(Name = "cleanup-sessions", Cron = "0 */5 * * * *", TimeZone = "Europe/Vienna")] public sealed partial class CleanupSessionsJob(AppDbContext db) { private ValueTask HandleAsync(EmptyJobRequest request, CancellationToken cancellationToken) => @@ -16,6 +16,22 @@ public sealed partial class CleanupSessionsJob(AppDbContext db) } ``` +Use the generated `RecurringJobs` service when code needs to run a payloadless job by name instead +of using its scheduler type: + +```csharp +public sealed class NamedJobOperations(RecurringJobs recurringJobs) +{ + public ValueTask RunNowAsync(CancellationToken cancellationToken) => + recurringJobs.TriggerNowAsync("cleanup-sessions", cancellationToken); +} +``` + +`TriggerNowAsync` matches `[Job(Name = ...)]` exactly and is case-sensitive. It throws +`ImmediateJobException` for an unknown name or a job excluded by registration tags. Only +payloadless jobs are available. The method returns after saving the new run but does not return its +`JobHandle`; use the typed scheduler when the caller needs that handle. + Cron expressions accept five fields (minute precision), six fields (seconds first), or the case-insensitive macros `@yearly`/`@annually`, `@monthly`, `@weekly`, `@daily`/`@midnight`, `@hourly`, `@every_minute` and `@every_second`. Time zones are IANA identifiers and default to @@ -34,14 +50,14 @@ public sealed class CleanupOperations(CleanupSessionsJob.Scheduler scheduler) } ``` -At startup the hosted service upserts every code-defined schedule and removes obsolete -code-defined rows. Dynamic rows are left alone. This reconciliation means a deploy can change a -cron expression, but two versions of an application should not intentionally define different -schedules under the same name. +At startup, Jobs saves or updates every code-defined schedule when storage supports recurring +jobs. It removes old code-defined schedules but leaves dynamic schedules alone. A queue-only +provider skips this work. This lets a deployment change a cron expression. Do not run two versions +of an application that define different schedules under the same name. -When the persisted cron expression and time zone are unchanged, reconciliation preserves its -stored `NextRunAt`, including an occurrence that became due while the application was stopped. A -changed cron expression or time zone recomputes the next occurrence from the current time. +If the saved cron expression and time zone are unchanged, Jobs keeps `NextRunAt`, including a run +that became due while the application was stopped. If either setting changes, Jobs calculates the +next run from the current time. ## Dynamic schedules @@ -73,26 +89,48 @@ public sealed class TenantScheduleManager(TenantCleanupJob.Scheduler tenantClean } ``` -`AddOrUpdateRecurringAsync` is durable and idempotently replaces the named dynamic schedule. -`TriggerNowAsync` creates an immediate invocation without moving the next cron occurrence. The -dashboard can trigger, pause and resume existing schedules. +`AddOrUpdateRecurringAsync` saves the schedule and replaces an existing schedule with the same +name. `TriggerNowAsync` starts a run now without moving the next cron occurrence. The dashboard can +also trigger, pause and resume schedules. + +## Manage stored schedules + +Use `JobMonitor` to pause, resume or trigger a stored schedule by its schedule name: + +```csharp +public sealed class RecurringScheduleOperations(JobMonitor jobs) +{ + public ValueTask PauseAsync(string name, CancellationToken cancellationToken) => + jobs.PauseRecurringAsync(name, cancellationToken); + + public ValueTask ResumeAsync(string name, CancellationToken cancellationToken) => + jobs.ResumeRecurringAsync(name, cancellationToken); + + public ValueTask RunNowAsync(string name, CancellationToken cancellationToken) => + jobs.TriggerRecurringAsync(name, cancellationToken); +} +``` + +`JobMonitor.TriggerRecurringAsync` takes a schedule name. `RecurringJobs.TriggerNowAsync` takes the +job name from `[Job(Name = ...)]`. These names can differ: a dynamic schedule named +`tenant-42-cleanup` can run the job named `tenant-cleanup`. Both methods start a run without moving +the next cron occurrence. ## Overlap policy -| Policy | When the previous occurrence is still active | -| ------------ | -------------------------------------------------------------------------------- | -| `Skip` | Persist the occurrence as terminal `Skipped` history without executing it. | -| `Queue` | Materialize every occurrence but admit only one invocation of the job at a time. | -| `Concurrent` | Allow both invocations to execute, subject to other concurrency limits. | +| Policy | When the previous run is still active | +| ------------ | ---------------------------------------------------------------------- | +| `Skip` | Record the new run as `Skipped` without executing it. | +| `Queue` | Create every run, but execute only one instance of this job at a time. | +| `Concurrent` | Allow runs to overlap, subject to other concurrency limits. | -Materialization is coordinated in durable storage, so `Recurring` capability is required. Redis -and the SQL providers support it; graph support is unrelated. A malformed persisted schedule is -logged and skipped for that pass without blocking other recurring schedules or ordinary queued -jobs. +Recurring schedules need storage with recurring support so multiple workers do not create the same +run. Redis and the SQL providers support them; graph support is unrelated. Jobs logs and skips a +malformed stored schedule without blocking other schedules or queued jobs. ## NodaTime Install `Immediate.Jobs.NodaTime` to configure NodaTime payload serialization and use `Duration`, `Instant` and `DateTimeZone` scheduling overloads. See the dedicated -[NodaTime guide](/docs/Immediate.Jobs/nodatime) for registration, examples and the complete package -surface. +[NodaTime guide](/docs/Immediate.Jobs/nodatime) for registration, examples and the full API +reference. diff --git a/src/content/docs/Immediate.Jobs/registration-and-hosting.md b/src/content/docs/Immediate.Jobs/registration-and-hosting.md index 5afc576..0ab6efb 100644 --- a/src/content/docs/Immediate.Jobs/registration-and-hosting.md +++ b/src/content/docs/Immediate.Jobs/registration-and-hosting.md @@ -1,6 +1,6 @@ --- title: Registration and hosting -description: Register handlers, behaviors, generated jobs, storage and the hosted worker correctly. +description: Register handlers, jobs, storage and the worker service. order: 9 group: Guides --- @@ -11,24 +11,62 @@ also adds each selected handler's behavior dependencies: ```csharp title="Program.cs" builder.Services.AddMyAppHandlers(); -builder.Services.AddMyAppJobs(options => -{ - options.UseEntityFrameworkCore(); // single-server mode by default - options.MaxParallelJobs = 16; - options.PollingInterval = TimeSpan.FromSeconds(1); -}).AddHealthCheck(); +builder.Services.AddMyAppJobs() + .ConfigureWorkers(options => + { + options.MaxParallelJobs = 16; + options.PollingInterval = TimeSpan.FromSeconds(1); + }) + .ConfigureStorage(storage => storage + .UseEntityFrameworkCore() + .UseSingleServer()) + .AddHealthCheck(); ``` `MyApp` is the shared [assembly identifier](/docs/concepts/assembly-identifier). `AddMyAppJobs` -registers each selected job's scheduler, invoker, context extractors and definition, all queue -definitions, then adds the runtime once. It does **not** register the generated Immediate.Handlers -handlers; without `AddMyAppHandlers`, enqueue succeeds but execution fails when the worker cannot -resolve the handler or its behaviors. +accepts optional tags and returns `IImmediateJobsBuilder`. Use that builder to configure worker +settings, fair queues, storage and health checks. -Generated job schedulers, `IJobBatchScheduler`, `IJobMonitor` and `IJobBatchMonitor` are scoped. -Definitions, queue definitions, invokers, storage, serializer, ID generator, options and the worker -service are singleton. Every execution creates its own scope for extractors, behaviors, handler -and dependencies. +The generated method lives in the project's `RootNamespace`, matching Immediate.Handlers. Import +that namespace (for example, `using MyApp;`) when startup code is outside it, including a top-level +`Program.cs`. + +Registration adds the selected jobs, every queue definition and the generated `RecurringJobs` +service. Calling the method again does not duplicate jobs or add another worker. + +`AddMyAppJobs` does **not** register the generated Immediate.Handlers handlers; without +`AddMyAppHandlers`, enqueue succeeds but execution fails when the worker cannot resolve the handler +or its behaviors. + +The main service lifetimes are: + +| Lifetime | Services | +| --------- | ---------------------------------------------------------------------------------------------------------------------- | +| Scoped | Generated job schedulers, `IBatchScheduler`, `JobMonitor` and its read-only `IJobMonitor` interface. | +| Singleton | Job and queue definitions, generated invokers, `RecurringJobs`, storage, serializer, ID generator, options and worker. | + +Each job run gets a new scope for its context extractors, behaviors, handler and dependencies. + +## Fluent configuration + +`ConfigureWorkers` accepts either a direct options action or an +`OptionsBuilder` action. Use the second form to bind configuration. +`UseFairQueues` has the same binding option. Call `ConfigureStorage` exactly once: + +```csharp +builder.Services.AddMyAppJobs() + .ConfigureWorkers(options => options.BindConfiguration("ImmediateJobs")) + .UseFairQueues(options => options.BindConfiguration("ImmediateJobs:FairQueues")) + .ConfigureStorage(storage => storage + .UseEntityFrameworkCore() + .UseDistributed()) + .AddHealthCheck(tags: ["ready"]); +``` + +Jobs validates these options when the host starts. Use `UseInMemory()` for in-memory storage. A +durable provider defaults to single-server mode when neither `UseSingleServer` nor +`UseDistributed` is selected. In production, choose a mode explicitly so the registration shows +whether one or several scheduler processes may run. ## Tagged registration @@ -36,23 +74,27 @@ Jobs participate in the shared `[Handler(Tags = [...])]` model: ```csharp builder.Services.AddMyAppHandlers(tags: ["fulfillment"]); -builder.Services.AddMyAppJobs(tags: ["fulfillment"]); +builder.Services.AddMyAppJobs(tags: ["fulfillment"]) + .ConfigureStorage(storage => storage.UseInMemory()); ``` With no tags, all jobs are registered. With tags, an untagged job is always included and a tagged -job is included when any requested tag matches. Pass the same host slice to `AddMyAppHandlers` so -the selected job definitions have matching generated handlers. Queue definitions are -assembly-wide and are registered regardless of selected job tags. +job is included when any requested tag matches. Pass the same tags to `AddMyAppHandlers` so every +selected job has its generated handler. Queue definitions are registered for the whole assembly, +regardless of the selected job tags. -## Hosted-service lifecycle +## Worker startup and shutdown -The worker starts with the host. It initializes storage, reconciles recurring definitions, -recovers eligible work, polls/acquires jobs, renews leases, emits heartbeats and periodically -purges history. At shutdown it stops acquisition, cancels active work and waits up to -`ShutdownTimeout` (30 seconds by default). +The worker starts with the host. It prepares storage, restores saved work, starts jobs when they +are due, renews their leases, reports its health and removes old history. At shutdown, it stops +taking new work, cancels active jobs and waits up to `ShutdownTimeout` (30 seconds by default). -Provider/schema initialization happens during worker startup. Your application still owns its -database schema or bootstrap as described in +The storage provider starts with the worker, but the application still creates and updates its +database schema as described in [Configuring storage providers](/docs/Immediate.Jobs/configuring-storage-providers). Start the entire `IHost` in console and worker-service applications; merely building the service provider does not run jobs. + +Call `DisableWorkers()` when an application should enqueue or display jobs but a separate process +will execute them. Job registration and storage remain available, but the hosted worker exits +without preparing storage or starting jobs. The application must still call `ConfigureStorage`. diff --git a/src/content/docs/Immediate.Jobs/testing-jobs.md b/src/content/docs/Immediate.Jobs/testing-jobs.md index 64106d2..32d6797 100644 --- a/src/content/docs/Immediate.Jobs/testing-jobs.md +++ b/src/content/docs/Immediate.Jobs/testing-jobs.md @@ -1,6 +1,6 @@ --- title: Testing jobs -description: Test scheduling and execution deterministically with fake time, draining, captures and workflow assertions. +description: Test scheduling and execution with a controllable clock, captured calls and workflow assertions. order: 15 group: Guides --- @@ -11,14 +11,14 @@ dotnet add package Immediate.Jobs.Testing --prerelease ## Execute with fake time -`JobTestHarness` builds an in-memory, full-capability scheduler around a +`JobTestHarness` builds an in-memory scheduler with every feature and a controllable `FakeTimeProvider`. Register the generated jobs and their dependencies: ```csharp await using var harness = new JobTestHarness(services => { services.AddMyAppHandlers(); - services.AddMyAppJobs(options => options.UseInMemory()); + services.AddMyAppJobs(); services.AddSingleton(); }); @@ -41,19 +41,22 @@ await harness.AdvanceTimeAndDrainAsync(TimeSpan.FromMinutes(10), cancellationTok Assert.Equal(JobState.Succeeded, (await harness.GetJobAsync(handle, cancellationToken)).State); ``` -`DrainAsync` runs everything due without wall-clock sleeps. `AdvanceTimeAndDrainAsync` accepts a -`TimeSpan` or absolute `DateTimeOffset`. `QueryJobsAsync` and `GetJobAsync` inspect durable records; -`AssertEnqueuedAsync` validates state and deserializes the payload. The harness also -exposes `Storage`, `TimeProvider`, `Services` and `Batches`. +`DrainAsync` runs every due job immediately. `AdvanceTimeAndDrainAsync` moves the clock by a +`TimeSpan` or to a `DateTimeOffset`, then runs due jobs. `QueryJobsAsync` and `GetJobAsync` read saved +job records. `AssertEnqueuedAsync` checks the state and payload. The harness also exposes +`Storage`, `TimeProvider`, `Services` and `Batches`. + +Register generated jobs in the callback, but do not call `ConfigureStorage`. The harness installs +its own in-memory provider and fake clock after application registrations. For graphs, use `AssertBatchCommittedAtomicallyAsync`, `AssertContinuationReleasedAfterAsync` and `AssertCascadeSkippedAsync`. -Call `RunThroughPipelineAsync` when a test already has a record/payload and needs to execute -its generated invoker through the real DI/behavior pipeline. +Call `RunThroughPipelineAsync` when a test already has a record and payload and needs to +run the generated handler with its registered behaviors and dependencies. ## Capture scheduling only -Use `CaptureOnlyJobScheduler` when the subject should decide _what_ to schedule but no +Use `CaptureOnlyJobScheduler` when a test only needs to verify what was scheduled and no worker should run: ```csharp @@ -80,6 +83,58 @@ run time, group ID and generated handle. `CaptureOnlyRecurringJobScheduler` reco add/update/remove/trigger operations for payloadless dynamic schedules. Override the capture scheduler's ID creation when stable IDs make assertions clearer. -Testing helpers throw `JobTestAssertionException` with job/batch-specific mismatch details. They -exercise the same generated JSON metadata and storage state machine as production without relying -on wall-clock delays. +Failed checks throw `JobTestAssertionException` with details about the job or batch. The helpers +use the same serialization and execution paths as production while the fake clock avoids delays. + +## Test a storage provider + +Storage-provider authors can run the shared behavior tests with the same registration an +application would use. Choose the `StorageCapabilities` flags that match the provider and expose +each returned case separately to the test runner: + +```csharp +using Immediate.Jobs.Shared.Storage; +using Immediate.Jobs.Testing; +using Microsoft.Extensions.Time.Testing; + +private const StorageCapabilities Capabilities = + StorageCapabilities.Queue | + StorageCapabilities.Recurring; + +public static TheoryData Cases => + [.. JobStorageConformanceSuite.GetCases(Capabilities)]; + +[Theory] +[MemberData(nameof(Cases))] +public async Task StorageConforms(JobStorageConformanceTestCase testCase) +{ + await using var fixture = await AcmeStorageFixture.CreateAsync(); + await testCase.RunAsync(fixture.Services); +} +``` + +For each case: + +- create a fresh service provider that resolves exactly one `IJobStorage`; +- register `FakeTimeProvider` as `TimeProvider`; +- use separate data, such as a unique database, schema or key prefix. + +`FakeTimeProvider` comes from `Microsoft.Extensions.Time.Testing` in the +`Microsoft.Extensions.TimeProvider.Testing` package. `GetCases` always includes queue tests. It +adds recurring, graph, fair-queue and replica tests based on `StorageCapabilities`, and verifies +the provider reports those features before each test runs. + +`AllCasesByName` provides a case-insensitive lookup of every known case. A test runner can use it +when it stores a case name and later needs the matching `JobStorageConformanceTestCase`. + +The `Replica` flag covers `IJobStorageReplica`; `IJobGraphStorageReplica` has no separate flag. For +a provider that supports single-server mode, also run the relevant cases through its +single-server registration. This tests both replica interfaces. + +The graph tests verify that an older execution cannot add work after a newer attempt starts. They +also cover continuations added after a parent finishes and jobs added with a batch ID that does not +match the running batch. A rejected addition must not leave partial data behind. + +The suite does not choose a test framework or database tooling. Keep separate provider tests for +migrations, database-specific behavior, Redis key layout and scripts, connection ownership and +backend failures. Dispose the service provider before deleting its test data. diff --git a/src/content/docs/concepts/tags.md b/src/content/docs/concepts/tags.md index 0db877f..5588232 100644 --- a/src/content/docs/concepts/tags.md +++ b/src/content/docs/concepts/tags.md @@ -43,9 +43,10 @@ Filtering happens where you register, not where you declare: ```csharp title="Program.cs" // Worker host — background handlers and their services only -services.AddTodoHandlers(tags: "worker"); +services.AddTodoHandlers(tags: ["worker", "fulfillment"]); services.AddTodoServices("worker"); -services.AddTodoJobs(tags: ["fulfillment"]); +services.AddTodoJobs(tags: ["fulfillment"]) + .ConfigureStorage(storage => storage.UseInMemory()); // Web host — HTTP endpoints only app.MapTodoEndpoints(tags: "web"); @@ -97,17 +98,18 @@ app.MapTodoEndpoints(tags: ["web"]); app.MapTodoEndpoints("/v1", "web"); ``` -`AddXxxJobs` takes its optional options delegate before `tags`. It reads the job's -Immediate.Handlers `Tags` value; use the same filter for Jobs and Handlers so every selected job -has a generated handler at execution: +`AddXxxJobs` accepts `tags` and returns `IImmediateJobsBuilder` for other settings. It reads the same +Immediate.Handlers `Tags` value. Pass the same tags to both registration methods so every +registered job also has its handler: ```csharp services.AddTodoHandlers(tags: ["fulfillment"]); -services.AddTodoJobs(options => options.UseInMemory(), tags: ["fulfillment"]); +services.AddTodoJobs(tags: ["fulfillment"]) + .ConfigureStorage(storage => storage.UseInMemory()); ``` -`AddXxxJobs` does not replace `AddXxxHandlers`. Job queue definitions are assembly-wide and remain -registered even when job tags filter which job definitions and schedulers are added. +`AddXxxJobs` does not replace `AddXxxHandlers`. Queue definitions are registered for the whole +assembly even when tags limit which jobs are added. ## Where to go next