Class DurableJobStore

java.lang.Object

com.kingsrook.qqq.backend.core.jobs.store.DurableJobStore

public class DurableJobStore
extends Object

The durable job store, over the jobs_* tables, on any backend with the CONDITIONAL_UPDATE capability. Every state change that two pods could race on is a compare-and-set through ConditionalUpdateAction: - claim: chunk PENDING/RETRY -> RUNNING, guarded by leaseVersion; - heartbeat, checkpoint, complete, fail, release: guarded by the lease token and status RUNNING (fencing - a reclaimed lease's writes change 0 rows); - reap: RUNNING with an expired lease -> RETRY/FAILED/NEEDS_REVIEW, guarded by leaseVersion; - job status: guarded by the expected current statuses. Lease times come from the database clock. All methods run as the system user (the kernel's own bookkeeping), never as the job's user.

Constructor Summary

Constructors

Constructor	Description
DurableJobStore(DurableJobsMetaData settings)	Constructor

Method Summary

Modifier and Type	Method	Description
void	checkpoint(JobLease lease, String checkpointJson, QBackendTransaction transaction)	Save a mid-chunk checkpoint (e.g., process values after a step).
Optional<JobLease>	claimChunk(QRecord chunk, String podId, int leaseTtlSeconds)	Try to claim one specific claimable chunk.
Optional<JobLease>	claimChunk(QRecord chunk, String podId, int leaseTtlSeconds, QBackendTransaction transaction)	Claim one specific chunk, optionally in a (claiming) transaction.
Optional<JobLease>	claimNextChunkSkipLocked(Integer jobId, String podId, int leaseTtlSeconds)	Claim the job's next claimable chunk (lowest seq) that no other claimer has locked: FOR UPDATE SKIP LOCKED plus the fenced conditional update, in one transaction.
void	completeChunk(JobLease lease, ChunkOutcome outcome, QBackendTransaction transaction)	Mark a chunk DONE with its outcome.
ChunkStatus	failChunk(JobLease lease, String errorMessage, boolean retryable)	A chunk run threw.
QRecord	getChunk(Integer chunkId)	Read a chunk (as the system user).
List<QRecord>	getChunks(Integer jobId)	The chunks of a job, in sequence order.
QRecord	getJob(Integer jobId)	Read a job (as the system user).
DurableJobsMetaData	getSettings()	Getter for settings
boolean	heartbeat(JobLease lease, int leaseTtlSeconds)	Renew a lease.
boolean	heartbeat(JobLease lease, int leaseTtlSeconds, QBackendTransaction transaction)	Renew under the progress transaction, holding the fence until its commit.
void	insertChunks(List<QRecord> chunks, QBackendTransaction transaction)	Insert planned chunks for a job.
Integer	insertJob(QRecord job, QBackendTransaction transaction)	Insert a job row; return its id.
Instant	now()	The database clock.
List<QRecord>	query(String tableName, QQueryFilter filter)	Query a jobs table (as the system user).
List<QRecord>	reapExpired(String reaperPodId)	Reclaim every RUNNING chunk whose lease expired (by the database clock).
void	recordEvent(Integer jobId, Integer chunkId, JobEventType type, String actorUserId, String clientChannel, String pod, String detail)	Add a timeline event.
void	recordRowErrors(List<QRecord> rowErrors, QBackendTransaction transaction)	Record row errors for a chunk (batched), in the chunk's transaction.
List<QRecord>	refreshCounters(Integer jobId)	Recompute a job's row and chunk counters from its chunks; return the chunks.
boolean	releaseChunk(JobLease lease, ChunkStatus to)	Give a chunk back without counting the attempt (graceful drain, pause): RUNNING -> PENDING.
int	setChunkStatuses(Integer jobId, Collection<ChunkStatus> from, ChunkStatus to, Map<String,Serializable> extraValues)	Set chunks of a job from some statuses to another (control actions).
boolean	setJobStatus(Integer jobId, Collection<JobStatus> from, JobStatus to, Map<String,Serializable> extraValues)	Compare-and-set a job's status: from one of the expected statuses to a new one, setting any extra values at the same time.
boolean	setJobStatusWithEvent(Integer jobId, Collection<JobStatus> from, JobStatus to, Map<String,Serializable> extraValues, JobEventType eventType, String detail)	Compare-and-set a job's status and record its timeline event in one transaction, so a reader that sees the new status also sees the event (status first and event after left a window where a finished job had no COMPLETED event yet).
void	updateJob(Integer jobId, Map<String,Serializable> values)	Set values on a job without a status condition (counters, reasons).
boolean	usesSkipLockedClaims()	Whether chunk claims use SELECT ...

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

DurableJobStore

public DurableJobStore(DurableJobsMetaData settings)

Constructor

Method Details

now

public Instant now()
            throws QException

The database clock.

Throws:

QException

insertJob

public Integer insertJob(QRecord job,
 QBackendTransaction transaction)
                  throws QException

Insert a job row; return its id.

Throws:

QException

insertChunks

public void insertChunks(List<QRecord> chunks,
 QBackendTransaction transaction)
                  throws QException

Insert planned chunks for a job.

Throws:

QException

getJob

public QRecord getJob(Integer jobId)
               throws QException

Read a job (as the system user).

Throws:

QException

getChunk

public QRecord getChunk(Integer chunkId)
                 throws QException

Read a chunk (as the system user).

Throws:

QException

query

public List<QRecord> query(String tableName,
 QQueryFilter filter)
                    throws QException

Query a jobs table (as the system user).

Throws:

QException

getChunks

public List<QRecord> getChunks(Integer jobId)
                        throws QException

The chunks of a job, in sequence order.

Throws:

QException

setJobStatus

public boolean setJobStatus(Integer jobId,
 Collection<JobStatus> from,
 JobStatus to,
 Map<String,Serializable> extraValues)
                     throws QException

Compare-and-set a job's status: from one of the expected statuses to a new one, setting any extra values at the same time. True if this caller won.

Throws:

QException

setJobStatusWithEvent

public boolean setJobStatusWithEvent(Integer jobId,
 Collection<JobStatus> from,
 JobStatus to,
 Map<String,Serializable> extraValues,
 JobEventType eventType,
 String detail)
                              throws QException

Compare-and-set a job's status and record its timeline event in one transaction, so a reader that sees the new status also sees the event (status first and event after left a window where a finished job had no COMPLETED event yet). Returns false, writing nothing, when the job was not in one of the from-statuses.

Throws:

QException

updateJob

public void updateJob(Integer jobId,
 Map<String,Serializable> values)
               throws QException

Set values on a job without a status condition (counters, reasons).

Throws:

QException

claimChunk

public Optional<JobLease> claimChunk(QRecord chunk,
 String podId,
 int leaseTtlSeconds)
                              throws QException

Try to claim one specific claimable chunk. Returns the lease if this pod won.

Throws:

QException

usesSkipLockedClaims

public boolean usesSkipLockedClaims()

Whether chunk claims use SELECT ... FOR UPDATE SKIP LOCKED (setting on, and the jobs backend supports it).

claimNextChunkSkipLocked

public Optional<JobLease> claimNextChunkSkipLocked(Integer jobId,
 String podId,
 int leaseTtlSeconds)
                                            throws QException

Claim the job's next claimable chunk (lowest seq) that no other claimer has locked: FOR UPDATE SKIP LOCKED plus the fenced conditional update, in one transaction.

Throws:

QException

claimChunk

public Optional<JobLease> claimChunk(QRecord chunk,
 String podId,
 int leaseTtlSeconds,
 QBackendTransaction transaction)
                              throws QException

Claim one specific chunk, optionally in a (claiming) transaction.

Throws:

QException

heartbeat

public boolean heartbeat(JobLease lease,
 int leaseTtlSeconds)
                  throws QException

Renew a lease. False when the lease is no longer held.

Throws:

QException

heartbeat

public boolean heartbeat(JobLease lease,
 int leaseTtlSeconds,
 QBackendTransaction transaction)
                  throws QException

Renew under the progress transaction, holding the fence until its commit.

Throws:

QException

checkpoint

public void checkpoint(JobLease lease,
 String checkpointJson,
 QBackendTransaction transaction)
                throws QException

Save a mid-chunk checkpoint (e.g., process values after a step). Fenced.

Throws:

QException

completeChunk

public void completeChunk(JobLease lease,
 ChunkOutcome outcome,
 QBackendTransaction transaction)
                   throws QException

Mark a chunk DONE with its outcome. Fenced: throws LeaseLostException when the lease is gone, so a caller holding a transaction rolls it back. Pass the chunk's transaction for exactly-once (SAME_TRANSACTION).

Throws:

QException

failChunk

public ChunkStatus failChunk(JobLease lease,
 String errorMessage,
 boolean retryable)
                      throws QException

A chunk run threw. Retry it while attempts remain and the failure is retryable; otherwise FAILED. Fenced. Returns the new status, or null when the lease was already lost.

Throws:

QException

releaseChunk

public boolean releaseChunk(JobLease lease,
 ChunkStatus to)
                     throws QException

Give a chunk back without counting the attempt (graceful drain, pause): RUNNING -> PENDING. Fenced. True when released.

Throws:

QException

reapExpired

public List<QRecord> reapExpired(String reaperPodId)
                          throws QException

Reclaim every RUNNING chunk whose lease expired (by the database clock). Each goes to RETRY, or FAILED at max attempts, or NEEDS_REVIEW when its job is not idempotent. Returns the reclaimed chunks.

Throws:

QException

setChunkStatuses

public int setChunkStatuses(Integer jobId,
 Collection<ChunkStatus> from,
 ChunkStatus to,
 Map<String,Serializable> extraValues)
                     throws QException

Set chunks of a job from some statuses to another (control actions).

Throws:

QException

recordRowErrors

public void recordRowErrors(List<QRecord> rowErrors,
 QBackendTransaction transaction)
                     throws QException

Record row errors for a chunk (batched), in the chunk's transaction.

Throws:

QException

recordEvent

public void recordEvent(Integer jobId,
 Integer chunkId,
 JobEventType type,
 String actorUserId,
 String clientChannel,
 String pod,
 String detail)

Add a timeline event. Best effort: a failure is logged, never thrown.

refreshCounters

public List<QRecord> refreshCounters(Integer jobId)
                              throws QException

Recompute a job's row and chunk counters from its chunks; return the chunks.

Throws:

QException

getSettings

public DurableJobsMetaData getSettings()

Getter for settings