NexJob.Postgres 5.7.0

NexJob.Postgres

PostgreSQL storage provider for NexJob — the background job scheduler for .NET 8+.

Backed by Npgsql and Dapper, this provider uses SELECT FOR UPDATE SKIP LOCKED for atomic, concurrency-safe job fetching across multiple distributed worker nodes.

Installation

dotnet add package NexJob.Postgres

Quick Start

Register PostgreSQL storage in your dependency injection container before calling AddNexJob():

using NexJob;
using NexJob.Postgres;

var builder = WebApplication.CreateBuilder(args);

// 1. Register PostgreSQL storage
builder.Services.AddNexJobPostgres(
    builder.Configuration.GetConnectionString("NexJobConnection")!);

// 2. Register NexJob core services
builder.Services.AddNexJob(options =>
{
    options.Workers = 10;
    options.Queues = ["default", "critical"];
});

Features

• Atomic Job Dispatching: Leverages PostgreSQL FOR UPDATE SKIP LOCKED to ensure zero lock contention and no double-execution across multiple server instances.
• High-Throughput Batch Processing: Dynamic batch dequeue (FetchBatchAsync) and vectorized batch acknowledgment (AcknowledgeBatchAsync) with native array operations (WHERE id = ANY(@Ids)).
• Automatic Schema Migration: Applies database migrations automatically on startup using PostgreSQL advisory locks (pg_advisory_lock), ensuring safe concurrent migrations in multi-node clusters.
• Idempotency Enforcement: Supports strict and conditional deduplication policies (DuplicatePolicy) indexed on idempotency_key.
• Runtime Settings Store: Persists dynamic queue throttling and runtime configuration in nexjob_settings.
• Dashboard Read Replica Support: Offload read-heavy dashboard and telemetry queries to a secondary database replica.

High-Throughput Batch Processing

For heavy ingestion workloads, enable batch processing to eliminate per-job WAL transaction log flushes:

builder.Services.AddNexJobPostgres(connectionString);

builder.Services.AddNexJob(options =>
{
    // Workers dynamically batch fetches to keep all idle slots busy (LIMIT availableWorkers)
    options.Workers = 30;
    options.PollingInterval = TimeSpan.FromMilliseconds(20);

    // Commit successful jobs in asynchronous batches via WHERE id = ANY(@Ids)
    options.EnableBatchAcknowledgment = true;
});

Read Replica Configuration

Reusing an NpgsqlDataSource

If your application already has an NpgsqlDataSource (for example one shared with EF Core or configured with custom type mappings), pass it instead of a connection string:

builder.Services.AddNexJobPostgres(existingDataSource);

If your database topology includes a PostgreSQL read replica, you can isolate dashboard queries from the write path using UseDashboardReadReplica:

builder.Services.AddNexJobPostgres(primaryConnectionString);

builder.Services.AddNexJob(options =>
{
    options.Workers = 10;
})
.UseDashboardReadReplica(readReplicaConnectionString);

Database Tables

The provider automatically manages the following tables:

Connection String Example

{
  "ConnectionStrings": {
    "NexJobConnection": "Host=localhost;Port=5432;Database=nexjob;Username=postgres;Password=postgres;Include Error Detail=true"
  }
}

Connection pool

NexJob shares your database with other applications. A node needs about Workers + 5 connections, so set Maximum Pool Size in the connection string to about Workers + 10 and multiply by the number of nodes. NexJob logs the pool size it finds at startup and warns when it is smaller than Workers. See Database connections and pool sizing.