Return orphaned catchup ranges to the front of the queue

src/MissionParallelCatchup/parallel_catchup_helm/files/worker.sh

@@ -9,6 +9,9 @@ if [ -z "$FAILED_QUEUE" ]; then echo "FAILED_QUEUE not set"; exit 1; fi
 if [ -z "$SUCCESS_QUEUE" ]; then echo "SUCCESS_QUEUE not set"; exit 1; fi
 if [ -z "$METRICS" ]; then echo "METRICS not set"; exit 1; fi
 if [ -z "$JOB_OWNERS" ]; then echo "JOB_OWNERS not set"; exit 1; fi
+if [ -z "$WORKER_CURRENT" ]; then echo "WORKER_CURRENT not set"; exit 1; fi
+if [ -z "$ATTEMPTS" ]; then echo "ATTEMPTS not set"; exit 1; fi
+if [ -z "$MAX_ATTEMPTS" ]; then echo "MAX_ATTEMPTS not set"; exit 1; fi
 if [ -z "$RELEASE_NAME" ]; then echo "RELEASE_NAME not set"; exit 1; fi
 if [ -z "$POD_NAME" ]; then echo "POD_NAME not set"; exit 1; fi
 
@@ -21,6 +24,48 @@ fi
 SLEEP_INTERVAL=10
 LOG_DIR="/data"
 
+# Hand back a range that a previous incarnation of this pod was holding.
+#
+# If our container was OOMKilled or our pod was evicted mid-catchup, the range
+# we had claimed is still sitting in PROGRESS_QUEUE with nobody working on it.
+# Nothing else can tell that from a range that is progressing normally, so it
+# used to stay there until every other range in the mission had finished and
+# the job monitor saw the whole fleet idle -- which is how a single dead worker
+# added hours to a run (stellar/supercluster#409). We are the one process that
+# knows for certain we are not working on it, so we are the one that returns it.
+#
+# The LREM guard means we only re-enqueue a range we actually removed, so this
+# is a no-op if the range was already recovered some other way. Ranges that keep
+# coming back here are failed rather than retried forever, otherwise a range
+# that reliably OOMs would loop (stellar/supercluster#334).
+#
+# KEYS: progress queue, job queue, failed queue, attempts hash, worker current hash
+# ARGV: job key, pod name, max attempts
+# Returns: 0 if the range was no longer in progress, n > 0 if requeued as
+#          attempt n, -1 if attempts were exhausted and the range was failed
+RECLAIM_LUA='
+redis.call("HDEL", KEYS[5], ARGV[2])
+if redis.call("LREM", KEYS[1], -1, ARGV[1]) ~= 1 then return 0 end
+local n = redis.call("HINCRBY", KEYS[4], ARGV[1], 1)
+if n > tonumber(ARGV[3]) then
+    redis.call("LPUSH", KEYS[3], ARGV[1] .. "|" .. ARGV[2])
+    return -1
+end
+redis.call("LPUSH", KEYS[2], ARGV[1])
+return n'
+
+PREV_JOB=$(redis-cli -h "$REDIS_HOST" -p "$REDIS_PORT" HGET "$WORKER_CURRENT" "$POD_NAME")
+if [ -n "$PREV_JOB" ]; then
+    RECLAIMED=$(redis-cli -h "$REDIS_HOST" -p "$REDIS_PORT" EVAL "$RECLAIM_LUA" 5 \
+        "$PROGRESS_QUEUE" "$JOB_QUEUE" "$FAILED_QUEUE" "$ATTEMPTS" "$WORKER_CURRENT" \
+        "$PREV_JOB" "$POD_NAME" "$MAX_ATTEMPTS")
+    case "$RECLAIMED" in
+        0)  echo "Previously held range $PREV_JOB was already recovered elsewhere" ;;
+        -1) echo "Range $PREV_JOB has now been orphaned more than $MAX_ATTEMPTS times; failing it" ;;
+        *)  echo "Returned orphaned range $PREV_JOB to the front of $JOB_QUEUE as attempt $RECLAIMED" ;;
+    esac
+fi
+
 while true; do
 # Fetch the next job key from the Redis queue.
 # Our ranges are generated in the order we want to run them from left to right, so we always pull from the left
@@ -29,8 +74,16 @@ LMOVE_EXIT_CODE=$?
 
 # Only process a job if the command succeeded AND we got a non-empty job key
 if [ $LMOVE_EXIT_CODE -eq 0 ] && [ -n "$JOB_KEY" ]; then
-    # Register ownership so the monitor knows which worker owns this job
-    redis-cli -h "$REDIS_HOST" -p "$REDIS_PORT" HSET "$JOB_OWNERS" "$JOB_KEY" "$POD_NAME"
+    # Register ownership so the monitor knows which worker owns this job, and
+    # record it against our pod name so that if we die and come back we can tell
+    # which range we abandoned. Both directions are needed: JOB_OWNERS answers
+    # "who has this range", WORKER_CURRENT answers "what was this pod holding".
+    redis-cli -h "$REDIS_HOST" -p "$REDIS_PORT" <<EOF
+MULTI
+HSET "$JOB_OWNERS" "$JOB_KEY" "$POD_NAME"
+HSET "$WORKER_CURRENT" "$POD_NAME" "$JOB_KEY"
+EXEC
+EOF
     if [ $? -ne 0 ]; then
         echo "Error: Failed to register job ownership for $JOB_KEY. Exiting."
         exit 1
@@ -93,6 +146,7 @@ $QUEUE_COMMAND
 LREM "$PROGRESS_QUEUE" -1 "$JOB_KEY"
 SADD "$METRICS" "$JOB_KEY|$core_id|$tx_apply_ms|$DURATION"
 HDEL "$JOB_OWNERS" "$JOB_KEY"
+HDEL "$WORKER_CURRENT" "$POD_NAME"
 EXEC
 EOF
         result=$?

src/MissionParallelCatchup/parallel_catchup_helm/templates/catchup_workers.yaml

@@ -164,4 +164,7 @@ data:
   PROGRESS_QUEUE: "{{ .Values.redis.progress_queue }}"
   METRICS: "{{ .Values.redis.metrics }}"
   JOB_OWNERS: "{{ .Values.redis.job_owners }}"
+  WORKER_CURRENT: "{{ .Values.redis.worker_current }}"
+  ATTEMPTS: "{{ .Values.redis.attempts }}"
+  MAX_ATTEMPTS: "{{ .Values.recovery.max_attempts }}"
   RELEASE_NAME: "{{ .Release.Name }}"

src/MissionParallelCatchup/parallel_catchup_helm/values.yaml

@@ -7,6 +7,12 @@ redis:
   progress_queue: "in_progress"
   metrics: "metrics"
   job_owners: "job_owners"
+  # Reverse of job_owners: which range each pod is currently holding. A worker
+  # that restarts reads its own entry to discover the range it abandoned.
+  worker_current: "worker_current"
+  # Per-range recovery tally, so a range that reliably kills its worker is
+  # failed rather than retried forever (stellar/supercluster#334).
+  attempts: "attempts"
   resources:
     requests:
       cpu: "100m"
@@ -35,6 +41,11 @@ worker:
   historyGetCommandCore003:
     "curl -sf http://history.stellar.org/prd/core-live/core_live_003/{0} -o {1}"
 
+recovery:
+  # How many times a range may be orphaned by a dying worker and handed back to
+  # the queue before it is failed instead.
+  max_attempts: 3
+
 monitor:
   gateway_name: "traefik-gateway-private"
   gateway_namespace: "traefik"