Drain local workers by working off remaining jobs on SIGTERM

Local workers handle jobs that require local filesystem access (package,
droplet, and buildpack uploads). When a local worker receives SIGTERM
during drain, the default delayed_job handler calls stop(), which exits
after the current job but leaves remaining queued jobs unprocessed until
the worker restarts.

This adds LocalWorkerDrainPlugin, which replaces the SIGTERM handler for
local workers only. Instead of stopping, the worker sets exit_on_complete
so it finishes all remaining jobs in the queue before exiting, avoiding
dangling PROCESSING_UPLOAD/PENDING states across restarts.

The plugin is a no-op for generic workers and named clock queues, which
retain the default SIGTERM behaviour.

Author: svkrieger

lib/delayed_job/local_worker_drain_plugin.rb

@@ -0,0 +1,13 @@
+# Replaces the default SIGTERM handler for local workers to drain all remaining jobs before exiting.
+class LocalWorkerDrainPlugin < Delayed::Plugin
+  callbacks do |lifecycle|
+    lifecycle.before(:execute) do |worker|
+      trap('TERM') do
+        Thread.new { worker.say 'Draining: will exit after finishing remaining jobs' }
+        worker.class.exit_on_complete = true
+      end
+    end
+  end
+end
+
+Delayed::Worker.plugins << LocalWorkerDrainPlugin

lib/tasks/jobs.rake

@@ -26,6 +26,7 @@ namespace :jobs do
   desc 'Start a delayed_job worker that works on jobs that require access to local resources.'
 
   task :local, [:name] => :environment do |_t, args|
+    require 'delayed_job/local_worker_drain_plugin'
     puts RUBY_DESCRIPTION
     queue = VCAP::CloudController::Jobs::Queues.local(RakeConfig.config).to_s
     args.with_defaults(name: queue)

spec/unit/lib/delayed_job/local_worker_drain_plugin_spec.rb

@@ -0,0 +1,41 @@
+require 'spec_helper'
+require 'delayed_job/local_worker_drain_plugin'
+
+RSpec.describe LocalWorkerDrainPlugin do
+  let(:worker) { Delayed::Worker.new }
+
+  before do
+    Delayed::Worker.exit_on_complete = false
+    allow(worker).to receive(:reload!)
+    allow(worker).to receive(:say)
+  end
+
+  after do
+    Delayed::Worker.exit_on_complete = false
+    Delayed::Worker.plugins.delete(LocalWorkerDrainPlugin)
+  end
+
+  describe 'TERM signal handling' do
+    before do
+      Delayed::Worker.sleep_delay = 0
+    end
+
+    after { Delayed::Worker.sleep_delay = Delayed::Worker::DEFAULT_SLEEP_DELAY }
+
+    it 'works off all remaining jobs in the queue before exiting' do
+      work_off_calls = Queue.new
+      allow(worker).to receive(:work_off) do
+        work_off_calls.push(:called)
+        work_off_calls.size < 3 ? [1, 0] : [0, 0]
+      end
+
+      worker_thread = Thread.new { worker.start }
+      work_off_calls.pop # wait until worker has started
+      Process.kill('TERM', Process.pid)
+      worker_thread.join(5)
+
+      expect(worker_thread.alive?).to be(false)
+      expect(work_off_calls.size).to eq(3)
+    end
+  end
+end