diff --git a/src/main/kotlin/pw/binom/mirror/worker/db/Job.kt b/src/main/kotlin/pw/binom/mirror/worker/db/Job.kt index 58cec44..b51b269 100644 --- a/src/main/kotlin/pw/binom/mirror/worker/db/Job.kt +++ b/src/main/kotlin/pw/binom/mirror/worker/db/Job.kt @@ -26,6 +26,7 @@ data class Job( val createdAt: Instant, val updatedAt: Instant, val takenAt: Instant? = null, + val retryCount: Int = 0, ) { companion object { const val STATUS_NEW = "new" @@ -33,5 +34,6 @@ data class Job( const val STATUS_DONE = "done" const val STATUS_FAILED = "failed" const val STATUS_CANCELLED = "cancelled" + const val MAX_RETRIES = 3 } } diff --git a/src/main/kotlin/pw/binom/mirror/worker/db/JobRepository.kt b/src/main/kotlin/pw/binom/mirror/worker/db/JobRepository.kt index 4df7465..d6743ed 100644 --- a/src/main/kotlin/pw/binom/mirror/worker/db/JobRepository.kt +++ b/src/main/kotlin/pw/binom/mirror/worker/db/JobRepository.kt @@ -71,7 +71,7 @@ open class JobRepository( jdbcTemplate.update( """ UPDATE media_mirror.jobs - SET status = 'failed', error = ?, updated_at = now() + SET status = 'failed', error = ?, retry_count = retry_count + 1, updated_at = now() WHERE id = ? """.trimIndent(), error.take(4000), @@ -110,8 +110,16 @@ open class JobRepository( val jobs = jdbcTemplate.query( """ SELECT * FROM media_mirror.jobs - WHERE status = 'new' - ORDER BY created_at + WHERE ( + status = 'new' + OR (status = 'failed' AND retry_count < 3) + OR (status = 'processing' AND taken_at < now() - interval '30 seconds') + ) + ORDER BY + CASE WHEN status = 'new' THEN 0 + WHEN status = 'failed' THEN 1 + ELSE 2 END, + created_at LIMIT 1 FOR UPDATE SKIP LOCKED """.trimIndent(), @@ -152,6 +160,7 @@ open class JobRepository( createdAt = rs.getTimestamp("created_at").toInstant(), updatedAt = rs.getTimestamp("updated_at").toInstant(), takenAt = rs.getTimestamp("taken_at")?.toInstant(), + retryCount = rs.getInt("retry_count"), ) } } diff --git a/src/main/kotlin/pw/binom/mirror/worker/worker/JobProcessor.kt b/src/main/kotlin/pw/binom/mirror/worker/worker/JobProcessor.kt index 42b1bc0..fe25e74 100644 --- a/src/main/kotlin/pw/binom/mirror/worker/worker/JobProcessor.kt +++ b/src/main/kotlin/pw/binom/mirror/worker/worker/JobProcessor.kt @@ -58,6 +58,12 @@ class JobProcessor( throw e } catch (e: Throwable) { logger.warn("Job {} finished with error: {}", job.id, e.message) + val currentRetries = jobRepository.findById(job.id)?.retryCount ?: 0 + if (currentRetries >= Job.MAX_RETRIES) { + logger.warn("Job {} exceeded max retries ({}), permanently failed", job.id, currentRetries) + } else { + logger.info("Job {} failed (retry {}/{}), will be retried", job.id, currentRetries, Job.MAX_RETRIES) + } jobRepository.markFailed(job.id, e.message ?: e.toString()) } finally { heartbeatJob.cancel() diff --git a/src/main/resources/db/migration/V2__add_retry_count.sql b/src/main/resources/db/migration/V2__add_retry_count.sql new file mode 100644 index 0000000..d4c34ea --- /dev/null +++ b/src/main/resources/db/migration/V2__add_retry_count.sql @@ -0,0 +1 @@ +ALTER TABLE media_mirror.jobs ADD COLUMN IF NOT EXISTS retry_count int NOT NULL DEFAULT 0;