klokka-postgres/com.eventslooped.klokka.postgres/PostgresJobStore

PostgresJobStore

class PostgresJobStore(dataSource: DataSource, config: PostgresStoreConfig = PostgresStoreConfig()) : JobStore, PushCapableStore, QueryableStore

The reference JobStore: Postgres on plain JDBC, no driver classes and no ORM. Every contract rule is enforced by single SQL statements against database time:

claim and dueSchedules are one FOR UPDATE SKIP LOCKED claim each: atomic across concurrent callers, due-ness decided by now(), fences bumped in the same statement.
transition and completeFire are compare-and-set UPDATEs; a stale state or fence changes nothing.
The uniqueKey rule is a partial unique index, not application logic.
sweep compiles the declarative RetentionPolicy into bulk DELETEs.

Instants are stored with Postgres's microsecond precision; sub-microsecond components of caller-supplied times are truncated on write and in comparisons.

Constructors

PostgresJobStore

constructor(dataSource: DataSource, config: PostgresStoreConfig = PostgresStoreConfig())

Functions

claim

open suspend override fun claim(queues: List<QueueName>, kinds: Set<String>, limit: Int, lease: Duration, worker: WorkerId): List<ClaimedJob>

completeFire

open suspend override fun completeFire(scheduleId: String, fence: Long, runs: List<NewJob>, nextFireAt: Instant?): List<JobId>?

countsByStatus

open suspend override fun countsByStatus(): Map<JobStatus, Long>

dueSchedules

open suspend override fun dueSchedules(ids: Set<String>, limit: Int, lease: Duration, worker: WorkerId): List<ScheduleFire>

enqueue

open suspend override fun enqueue(jobs: List<NewJob>): List<JobId>

getJob

open suspend override fun getJob(id: JobId): JobDetails?

heartbeat

open suspend override fun heartbeat(ids: List<JobId>, worker: WorkerId, extend: Duration)

listJobs

open suspend override fun listJobs(query: JobQuery): List<JobSummary>

migrate

suspend fun migrate()

Applies the bundled migrations now, regardless of PostgresStoreConfig.autoMigrate. Idempotent and safe under concurrent callers (advisory-locked). With autoMigrate on, the store calls this lazily before its first statement.

sweep

open suspend override fun sweep(retention: RetentionPolicy)

transition

open suspend override fun transition(id: JobId, from: JobState, to: JobState, fence: Long?, error: JobError?): Boolean

upsertSchedules

open suspend override fun upsertSchedules(schedules: List<ScheduleSpec>)

wakeups

open override fun wakeups(): Flow<Unit>

LISTEN/NOTIFY push wake-ups, fed by the V002 schema triggers on every write that makes a job claimable. Delivery is at COMMIT, so a wake-up never precedes the row it announces; batch inserts collapse to one notification.