LowCodeHub.Jobs 0.2.0

LowCodeHub.Jobs

Durable, typed background jobs for ASP.NET Core (.NET 11): retries with backoff, delayed / recurring / continuation scheduling, cooperative cancellation, checkpoints, durable progress streaming over SSE, an embedded admin dashboard, and multi-pod coordination. Storage: SQL Server, PostgreSQL or in-memory.

dotnet add package LowCodeHub.Jobs

Quick start

// 1. A job = a request record + a handler.
public sealed record SendWelcomeEmail(string To) : IJobRequest<bool>
{
    public static JobMetadata Metadata { get; } = JobMetadataBuilder.For("emails.welcome.v1");
}

public sealed class SendWelcomeEmailHandler(IEmailSender email) : IJobHandler<SendWelcomeEmail, bool>
{
    public async Task<bool> HandleAsync(SendWelcomeEmail job, IJobContext ctx, CancellationToken ct)
    {
        await email.SendAsync(job.To, "Welcome!", ct);
        return true;
    }
}

// 2. Register (one storage provider + your handlers).
builder.Services.AddJobs(jobs => jobs
    .WithInMemory()                               // or .WithSqlServer(cs) / .WithPostgreSql(cs)
    .AddHandlersFromAssemblyContaining<Program>());

// 3. Enqueue from anywhere.
app.MapPost("/users/{email}/welcome", async (string email, IJobScheduler jobs, CancellationToken ct) =>
{
    var job = await jobs.EnqueueAsync(new SendWelcomeEmail(email), ct);
    return TypedResults.Accepted($"/jobs/{job.JobId}", job);
});

WithInMemory() needs no infrastructure (state is lost on restart, single instance only). For production switch to SQL Server or PostgreSQL — nothing else changes.

Defining jobs

• JobMetadata.Name is the stable wire name stored in the database — version it ("exports.users.v1") and never reuse it for a different payload.
• IJobRequest<TResult> / IJobHandler<TRequest, TResult> for plain jobs; IJobRequest<TResult, TProgress> / IJobHandler<TRequest, TResult, TProgress> adds a typed ctx.ReportAsync(progress).
• Handlers are scoped services resolved per attempt, so constructor injection works as usual.
• Register handlers with AddHandler<THandler>() (types inferred), AddHandlersFromAssemblyContaining<T>(), or the explicit AddHandler<TRequest, TResult, THandler>().

public static JobMetadata Metadata { get; } = JobMetadataBuilder.For("exports.users.v1")
    .WithDisplayName("Export users")
    .WithMaxAttempts(5)                          // total attempts incl. the first (default: Worker.DefaultMaxRetries = 3)
    .WithExecutionTimeout(TimeSpan.FromMinutes(10))
    .WithBackoff(5, 30, 120);                    // seconds between retries (default: exponential with jitter)

Failure semantics:

Other exceptions store only the exception type in ErrorMessage ("The job handler failed (HttpRequestException)."), because ErrorMessage is returned by IJobStatusReader and often shown to end users; the full exception is logged. Set Worker.IncludeExceptionDetails = true to persist messages.

A handler that ignores its cancellation token past Worker.TimeoutGracePeriod keeps its worker slot and lease until it returns, so it never runs concurrently with its own retry.

Handler context

IJobContext gives JobId, Attempt, EnqueuedAt, CorrelationId (trace id) and:

Every event is persisted in order (payload limit EventLog.MaxPayloadBytes, default 16 KB — larger payloads make the call throw).

Scheduling

await jobs.EnqueueAsync(request, ct);                                  // run now
await jobs.ScheduleAsync(request, TimeSpan.FromMinutes(30), ct);       // delay
await jobs.ScheduleAsync(request, runAt: tomorrow9am, ct);             // at a time
await jobs.ContinueWithAsync(parentJobId, followUp, ct);               // after the parent succeeds

await jobs.EnqueueAsync(request, new ScheduleOptions
{
    IdempotencyKey = $"invoice:{invoiceId}",   // duplicate enqueues return the existing job
    IdempotencyTtl = TimeSpan.FromDays(1),     // default: the result retention
    MaxAttempts = 5,
    ResultTtlOverride = TimeSpan.FromDays(7),
}, ct);

Outcome<Success> cancelled = await jobs.CancelAsync(jobId, ct);   // see "Outcomes" below

• Idempotency is per job type: same key + same payload returns the existing job; same key + different payload throws JobIdempotencyConflictException.
• Continuations run once the parent succeeds. If the parent ends Failed or Cancelled (or is deleted), waiting continuations — and their own continuations — are Cancelled with an explanatory ErrorMessage. Continuing a parent that already failed throws.

Recurring jobs

await jobs.ScheduleRecurringAsync("nightly-export", CronSchedules.Daily("02:00"),
    new ExportUsers(tenantId), timeZoneId: "Europe/Amsterdam", ct);
await jobs.RemoveRecurringAsync("nightly-export", ct);   // Outcome<Success>

Or declare it on a System job and it is registered automatically by the recurring scheduler:

public sealed record NightlyCleanup(int KeepDays = 30) : IJobRequest<bool>
{
    public static JobMetadata Metadata { get; } = JobMetadataBuilder.For("system.cleanup.v1")
        .AsSystem()
        .WithCron("0 2 * * *", timeZoneId: "UTC");    // request must be constructible with defaults
}

ScheduleRecurringAsync (and the automatic registration) is idempotent: calling it on every start is safe. An existing schedule only changes when job type, payload, cron or time zone differ; its next fire time is recomputed only when cron/time zone change; and a schedule disabled in the dashboard stays disabled.

Missed occurrences (downtime, long leader fail-over) are handled per tick, never replayed as a backlog:

Override per job type with JobMetadataBuilder.WithMisfirePolicy(...). A schedule with a bad cron or time zone is logged and skipped without blocking other schedules. Schedules whose job type has no handler on the scheduling pod are still enqueued and run on a worker that registers the handler.

Tracking jobs

// Poll
app.MapGet("/jobs/{id:guid}", async (Guid id, IJobStatusReader reader, CancellationToken ct) =>
    await reader.GetAsync(id, ct) is { } job ? TypedResults.Ok(job) : TypedResults.NotFound());

// Stream (Server-Sent Events, resumable with Last-Event-ID)
app.MapGet("/jobs/{id:guid}/events", (Guid id, HttpContext http, IJobStreamService streams, CancellationToken ct) =>
    TypedResults.ServerSentEvents(streams.StreamAsync(id, http.Request.Headers["Last-Event-ID"].ToString(), ct)));

JobDescriptor: JobId, JobType, Status, Attempt, EnqueuedAt, StartedAt, CompletedAt, ResultJson, ErrorMessage. These endpoints are yours — protect them like any other endpoint (Jobs exposes no public routes).

Statuses: Queued → Running → Succeeded | Failed | Cancelled; Scheduled = waiting for a delay, a retry backoff, or a continuation parent. Retries show as Scheduled; Failed is terminal.

Stream items are SseItem<JobEventEnvelope> whose SSE event: is lifecycle, progress, log, chunk, metric or heartbeat; the event id is the per-job sequence. Lifecycle events are written on every state change after enqueue (running, scheduled for a retry, queued after a re-queue, succeeded, failed, cancelled) and carry the error message for failures. The stream replays everything after Last-Event-ID, then tails live; it completes after the terminal lifecycle event and immediately for unknown or deleted jobs. Heartbeats carry the job's current status.

const es = new EventSource(`/jobs/${id}/events`);
es.addEventListener('progress', e => bar.value = JSON.parse(e.data).progress.percent);
es.addEventListener('lifecycle', e => {
  const { status, errorMessage } = JSON.parse(e.data).lifecycle;   // status is the numeric JobStatus
  if (status >= 3) es.close();                                     // Succeeded / Failed / Cancelled
});

Tracing: EnqueueAsync records the caller's W3C trace context; the job.execute span (ActivitySource LowCodeHub.Jobs) continues that trace, and ctx.CorrelationId is its trace id.

Storage

.WithSqlServer(builder.Configuration.GetConnectionString("Jobs")!)
.WithPostgreSql(o =>
{
    o.ConnectionString = builder.Configuration.GetConnectionString("Jobs")!;
    o.ApplySchemaOnStartup = true;    // dev/simple deployments: run the embedded scripts at startup
})

Exactly one provider must be registered. Each registers a health check (jobs-sql / jobs-pg).

Applying the schema yourself

The idempotent scripts are embedded resources (SqlJobs.SqlServerResources / SqlJobs.PostgreSqlResources).

With LowCodeHub.Migration.SqlServer / .PostgreSql:

builder.Services.AddSqlServerMigrations(m => m.AddTarget<IJobsScriptScanner>(o =>
{
    o.ConnectionString = builder.Configuration.GetConnectionString("Jobs")!;
    o.Directories = ["LowCodeHub.Jobs.Repositories.SqlServer.Scripts."];   // PostgreSQL: ...PostgreSql.Scripts.
}));

Always filter by directory — the assembly contains scripts for both providers. With EF Core, run them from an application migration:

protected override void Up(MigrationBuilder migrationBuilder)
{
    foreach (var resource in SqlJobs.SqlServerResources)
        migrationBuilder.Sql(SqlJobs.ReadFromResource(resource));   // GO separators removed
}

Admin dashboard

builder.Services.AddJobsDashboard(o => o.Password = builder.Configuration["Jobs:DashboardPassword"]!);
// or reuse your app's auth instead of the built-in login:
// builder.Services.AddJobsDashboard(o => o.AuthorizationPolicy = "Admin");

app.UseJobsDashboard();   // UI at /jobs-ui, API at /jobs-ui/api

Security model:

• Every API endpoint requires authentication — the built-in session cookie or your AuthorizationPolicy ("" = any authenticated user). Startup fails without a password of ≥ 12 characters or a policy.
• Built-in sessions are HMAC-signed with SessionSigningKey or a PBKDF2-derived key (never the raw password), so they work across pods and restarts; changing the password ends existing sessions. The cookie is HttpOnly, SameSite=Strict, scoped to the dashboard path, and Secure on HTTPS (CookieSecurePolicy).
• Failed logins are limited per client IP (MaxFailedLogins = 5 per LoginLockout = 15 min → HTTP 429).
• Mutating requests must come from the dashboard's own origin (Origin/Referer; add others via AllowedOrigins). Requests with neither header are accepted only when authenticated via an Authorization header.

Features: status counts, filtered job list, job detail with attempts/events/result/error, cancel (queued/scheduled/running), retry (failed or cancelled → re-queued with the job type's attempt budget), delete (any non-running job, with its events, attempts, checkpoints and idempotency key), recurring schedules (enable/disable, trigger now), worker and leader status.

IJobAdminRepository (scoped) exposes the same operations for your own tooling.

Outcomes

Expected failures of the scheduler and admin operations are returned as Outcome<Success> instead of bool (LowCodeHub.Outcomes is a dependency of this package): IJobScheduler.CancelAsync / RemoveRecurringAsync and IJobAdminRepository.RetryAsync / DeleteJobAsync / SetRecurringEnabledAsync / TriggerRecurringNowAsync.

app.MapPost("/admin/jobs/{id:guid}/cancel", async (Guid id, IJobScheduler jobs, CancellationToken ct) =>
    (await jobs.CancelAsync(id, ct)).ToHttpResult());      // 204, or problem+json with errorCode

var outcome = await admin.RetryAsync(id, additionalAttempts: 2, ct);
switch (outcome)
{
    case Success:
        break;
    case ProblemDetails problem when problem.HasErrorCode(JobErrorCodes.NotFound):
        return TypedResults.NotFound();
    case ProblemDetails problem:
        return problem.ToHttpResult();                      // Jobs.InvalidState -> 409
}

Handler control flow (JobPermanentFailureException, lease loss) and transport/storage failures still throw.

Breaking changes. IJobScheduler.CancelAsync / RemoveRecurringAsync and IJobAdminRepository.RetryAsync / DeleteJobAsync / SetRecurringEnabledAsync / TriggerRecurringNowAsync now return Outcome<Success> instead of bool. CancelAsync on an already finished or cancelled job now returns 409 Jobs.InvalidState (it was an idempotent true).

Deployment topologies

builder.Services.AddJobs(jobs => jobs
    .WithSqlServer(cs)
    .WithRedisCoordination(o => o.ConnectionString = redisCs)   // or .WithDatabaseCoordination()
    .AddHandlersFromAssemblyContaining<Program>());

• WithRedisCoordination probes Redis while registering services; if unreachable it falls back to the database lock (SQL Server sp_getapplock, PostgreSQL advisory lock) and logs a warning at startup. Pass o.ConnectionMultiplexer to reuse an existing connection.
• Without coordination a no-op lock is used (warning logged): correct thanks to compare-and-swap writes, but every pod runs the scheduler and janitor.
• ConfigureCoordination(c => …) tunes LeaseTtl (30 s), RenewInterval (10 s), AcquireRetryInterval (5 s), KeyPrefix.

Workers renew their job lease every Worker.LeaseRenewalInterval; transient storage errors are retried while the lease is still valid. A crashed worker's jobs are reclaimed after Worker.LeaseDuration. On shutdown in-flight jobs get Worker.DrainTimeout to finish, then are cancelled and re-queued; DrainTimeout + TimeoutGracePeriod must be shorter than HostOptions.ShutdownTimeout (validated at startup).

Configuration

Bind from configuration and/or a lambda (the lambda wins):

builder.Services.AddJobs(builder.Configuration, jobs => jobs.WithSqlServer(cs).AddHandlersFromAssemblyContaining<Program>());
// or: jobs.Configure(builder.Configuration.GetSection("Jobs")).Configure(o => o.Worker.MaxParallelism = 16)

{
  "Jobs": {
    "Worker": {
      "Enabled": true,
      "BatchSize": 32,
      "MaxParallelism": 8,
      "PollInterval": "00:00:01",
      "LeaseDuration": "00:05:00",
      "LeaseRenewalInterval": "00:01:00",
      "DefaultExecutionTimeout": "00:30:00",
      "TimeoutGracePeriod": "00:00:05",
      "DefaultMaxRetries": 3,
      "BaseRetryDelay": "00:00:05",
      "MaxRetryDelay": "00:30:00",
      "RetryJitter": 0.2,
      "ResultTtl": "1.00:00:00",
      "DrainOnShutdown": true,
      "DrainTimeout": "00:00:20",
      "CancellationPollInterval": "00:00:02",
      "IncludeExceptionDetails": false
    },
    "Recurring": {
      "Enabled": true,
      "TickInterval": "00:01:00",
      "LookAhead": "00:02:00",
      "MisfireThreshold": "00:15:00",
      "MisfirePolicy": "Skip"
    },
    "Janitor": {
      "Enabled": true,
      "SweepInterval": "00:05:00",
      "BatchSize": 500,
      "MaxBatchesPerSweep": 100
    },
    "EventLog": { "MaxPayloadBytes": 16384 },
    "Stream": { "PollInterval": "00:00:00.250", "BatchSize": 200, "HeartbeatInterval": "00:00:15" }
  }
}

All options are validated at startup (ValidateOnStart). ResultTtl controls how long finished jobs (and their events, attempts and checkpoints) are kept; the janitor deletes them in batches until nothing is expired or MaxBatchesPerSweep is reached.

Health & metrics

• jobs-worker (tags jobs, liveness): Healthy, or Healthy with "disabled" on client-only nodes; Degraded while storage calls keep failing (a restart would not help); Unhealthy when the worker loop stopped or its heartbeat is stale (> max(60 s, 10 × PollInterval)).
• jobs-sql / jobs-pg (tags jobs, db): database reachability.
• Meter LowCodeHub.Jobs: jobs.queue_depth, jobs.scheduled_pending, jobs.running_now (gauges), jobs.execution.total and jobs.execution.duration_ms (tags job_type, outcome = succeeded|failed|cancelled|retrying|abandoned), jobs.retry.scheduled.
• ActivitySource LowCodeHub.Jobs: job.enqueue (producer) and job.execute (consumer) spans.

Testing handlers

Handlers are plain classes — call HandleAsync with a fake IJobContext. For integration tests use AddJobs(b => b.WithInMemory().AddHandler<THandler>()) with a fake TimeProvider registered before AddJobs.