Make sure jobs are picked up after being split

The migration runner only picks up existing jobs that are failed with
attempts less than the max, or jobs that are stuck in pending or running
for an hour.

We keep the split jobs in the failed state so that they can be picked up
immediately.

lib/gitlab/database/background_migration/batched_job.rb

@@ -69,7 +69,7 @@ module Gitlab
             # In this case, we just lower the batch size so that future calls to this
             # method could eventually split the job if it continues to fail.
             if midpoint >= max_value
-              update!(batch_size: new_batch_size, status: :pending)
+              update!(batch_size: new_batch_size, attempts: 0)
             else
               old_max_value = max_value
 
@@ -77,7 +77,6 @@ module Gitlab
                 batch_size: new_batch_size,
                 max_value: midpoint,
                 attempts: 0,
-                status: :pending,
                 started_at: nil,
                 finished_at: nil,
                 metrics: {}

spec/lib/gitlab/database/background_migration/batched_job_spec.rb

@@ -126,20 +126,23 @@ RSpec.describe Gitlab::Database::BackgroundMigration::BatchedJob, type: :model d
   end
 
   describe '#split_and_retry!' do
-    let!(:job) { create(:batched_background_migration_job, batch_size: 10, min_value: 6, max_value: 15, status: :failed) }
+    let!(:job) { create(:batched_background_migration_job, batch_size: 10, min_value: 6, max_value: 15, status: :failed, attempts: 3) }
 
-    it 'splits the job into two and marks them as pending' do
+    context 'when job can be split' do
+      before do
         allow_next_instance_of(Gitlab::BackgroundMigration::BatchingStrategies::PrimaryKeyBatchingStrategy) do |batch_class|
           allow(batch_class).to receive(:next_batch).with(anything, anything, batch_min_value: 6, batch_size: 5).and_return([6, 10])
         end
+      end
 
+      it 'sets the correct attributes' do
         expect { job.split_and_retry! }.to change { described_class.count }.by(1)
 
         expect(job).to have_attributes(
           min_value: 6,
           max_value: 10,
           batch_size: 5,
-        status: 'pending',
+          status: 'failed',
           attempts: 0,
           started_at: nil,
           finished_at: nil,
@@ -153,7 +156,7 @@ RSpec.describe Gitlab::Database::BackgroundMigration::BatchedJob, type: :model d
           min_value: 11,
           max_value: 15,
           batch_size: 5,
-        status: 'pending',
+          status: 'failed',
           attempts: 0,
           started_at: nil,
           finished_at: nil,
@@ -162,6 +165,13 @@ RSpec.describe Gitlab::Database::BackgroundMigration::BatchedJob, type: :model d
         expect(new_job.created_at).not_to eq(job.created_at)
       end
 
+      it 'splits the jobs into retriable jobs' do
+        migration = job.batched_migration
+
+        expect { job.split_and_retry! }.to change { migration.batched_jobs.retriable.count }.from(0).to(2)
+      end
+    end
+
     context 'when job is not failed' do
       let!(:job) { create(:batched_background_migration_job, status: :succeeded) }
 
@@ -185,11 +195,12 @@ RSpec.describe Gitlab::Database::BackgroundMigration::BatchedJob, type: :model d
         end
       end
 
-      it 'lowers the batch size and marks the job as pending' do
+      it 'lowers the batch size and resets the number of attempts' do
         expect { job.split_and_retry! }.not_to change { described_class.count }
 
         expect(job.batch_size).to eq(5)
-        expect(job.status).to eq('pending')
+        expect(job.attempts).to eq(0)
+        expect(job.status).to eq('failed')
       end
     end
   end