From 12233e9bfa9475f2530df67ccae58564468e9703 Mon Sep 17 00:00:00 2001
From: Hermes Agent <hermes@binom.pw>
Date: Fri, 21 Aug 2026 12:17:42 +0300
Subject: [PATCH] =?UTF-8?q?feat:=20retry-=D0=BC=D0=B5=D1=85=D0=B0=D0=BD?=
 =?UTF-8?q?=D0=B8=D0=B7=D0=BC=20(max=203=20=D0=BF=D0=BE=D0=BF=D1=8B=D1=82?=
 =?UTF-8?q?=D0=BA=D0=B8)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- V2__add_retry_count.sql: retry_count int NOT NULL DEFAULT 0
- takeNextJob: new + failed (retry_count < 3) + stale processing (30с)
- markFailed: retry_count = retry_count + 1
- JobProcessor: логирование retry (X/3)
- Heartbeat: не трогать (уже работает, 10с)
---
 src/main/kotlin/pw/binom/mirror/worker/db/Job.kt  |  2 ++
 .../pw/binom/mirror/worker/db/JobRepository.kt    | 15 ++++++++++++---
 .../pw/binom/mirror/worker/worker/JobProcessor.kt |  6 ++++++
 .../db/migration/V2__add_retry_count.sql          |  1 +
 4 files changed, 21 insertions(+), 3 deletions(-)
 create mode 100644 src/main/resources/db/migration/V2__add_retry_count.sql

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;