insert([ 'id' => $batchId, 'type' => $type, 'total' => $total, 'status' => $total > 0 ? 'processing' : 'completed', ]); } /** * Atomically increments completed or failed for the batch and reports * whether this call finished it. Locks the row for the duration of the * transaction so concurrent workers incrementing the same batch can't * both observe completion. * * @return bool True if this call completed the batch (completed + failed = total). */ public function increment(string $batchId, bool $rowFailed): bool { $this->db->transStart(); $sql = $this->db->table($this->table) ->where('id', $batchId) ->getCompiledSelect(); $batch = $this->db->query($sql . ' FOR UPDATE')->getRowArray(); $column = $rowFailed ? 'failed' : 'completed'; $this->db->table($this->table) ->where('id', $batchId) ->set($column, $column . ' + 1', false) ->update(); $completed = ($batch['completed'] ?? 0) + ($rowFailed ? 0 : 1); $failed = ($batch['failed'] ?? 0) + ($rowFailed ? 1 : 0); $total = $batch['total'] ?? 0; $isComplete = ($completed + $failed) >= $total; if ($isComplete) { $this->db->table($this->table) ->where('id', $batchId) ->update(['status' => $failed > 0 ? 'partial' : 'completed']); } $this->db->transComplete(); return $isComplete; } /** * Deletes batches finished more than $retentionDays ago, but only once no * row referencing their batch_id remains in queue_jobs or * queue_jobs_failed (e.g. a failed row kept alive by a longer failed-job * retention window). 0 means keep until manual purge. */ public function purgeFinished(int $retentionDays): void { if ($retentionDays <= 0) { return; } $cutoff = date('Y-m-d H:i:s', strtotime("-{$retentionDays} days")); $candidates = $this->db->table($this->table) ->select('id') ->whereIn('status', ['completed', 'partial']) ->where('updated_at <=', $cutoff) ->get() ->getResultArray(); foreach ($candidates as $batch) { $batchId = $batch['id']; $stillReferenced = $this->db->table('queue_jobs')->where('batch_id', $batchId)->countAllResults() > 0 || $this->db->table('queue_jobs_failed')->where('batch_id', $batchId)->countAllResults() > 0; if (!$stillReferenced) { $this->db->table($this->table)->where('id', $batchId)->delete(); model(ImportBatchContext::class)->deleteForBatch($batchId); } } } }