Reserve transportation reclassification before enqueue

Author: Freika

CHANGELOG.md

@@ -8,6 +8,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ### Fixed
 
+- Reserve transportation reclassification while its worker is queued, preventing duplicate runs from repeated requests and allowing retry after enqueue failure.
+
 - Prometheus metrics from web and Sidekiq no longer produce duplicate series when label order differs (#3769).
 - KMZ imports no longer repeatedly re-import their own archive and fill the import queue and storage (#3766).
 

app/controllers/tracks/recalculations_controller.rb

@@ -8,7 +8,9 @@ class Tracks::RecalculationsController < ApplicationController
   def create
     status = Tracks::TransportationRecalculationStatus.new(current_user.id)
 
-    if status.in_progress?
+    enqueued = status.enqueue { TransportationModes::UserReclassifyJob.perform_later(current_user.id) }
+
+    if !enqueued
       respond_to do |format|
         format.turbo_stream do
           render turbo_stream: stream_flash(
@@ -22,7 +24,6 @@ def create
         end
       end
     else
-      TransportationModes::UserReclassifyJob.perform_later(current_user.id)
       respond_to do |format|
         format.turbo_stream do
           render turbo_stream: stream_flash(

app/services/tracks/transportation_recalculation_status.rb

@@ -27,6 +27,23 @@ def data
       Rails.cache.read(cache_key) || { 'status' => 'idle' }
     end
 
+    def enqueue
+      User.with_advisory_lock("transportation_recalculation:#{user_id}") do
+        return false if in_progress?
+
+        begin
+          raise IOError, 'Reclassification status could not be reserved' unless start(total_tracks: 0)
+
+          job = yield
+          raise ActiveJob::EnqueueError, 'Reclassification could not be queued' unless job.try(:successfully_enqueued?)
+        rescue StandardError => e
+          fail(e.message)
+          raise
+        end
+        true
+      end
+    end
+
     def start(total_tracks:)
       reset_counter
       Rails.cache.write(
@@ -49,10 +66,12 @@ def increment_processed!
       # hash would resurrect the short-lived 'completed' entry with CACHE_TTL.
       return data['processed_tracks'].to_i if current_status == 'completed'
 
-      count = Rails.cache.increment(counter_key, 1, expires_in: CACHE_TTL) || 1
       total = data['total_tracks']
+      return data['processed_tracks'].to_i if total&.zero?
+
+      count = Rails.cache.increment(counter_key, 1, expires_in: CACHE_TTL) || 1
       update_progress(processed_tracks: count, total_tracks: total)
-      complete if total && count >= total
+      complete if total&.positive? && count >= total
       count
     end
 

spec/regressions/queued_transport_reclassification_spec.rb

@@ -0,0 +1,84 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe 'Queued transportation reclassification', type: :request do
+  let(:user) { create(:user) }
+  let(:progress) { Tracks::TransportationRecalculationStatus.new(user.id) }
+
+  before do
+    sign_in user
+    clear_enqueued_jobs
+  end
+
+  def request_reclassification
+    post tracks_recalculation_path, headers: { 'Accept' => 'text/vnd.turbo-stream.html' }
+  end
+
+  it 'enqueues once while the first fan-out is still waiting in the queue' do
+    request_reclassification
+    request_reclassification
+
+    expect(TransportationModes::UserReclassifyJob).to have_been_enqueued.exactly(:once).with(user.id)
+    expect(progress.in_progress?).to be true
+    expect(progress.data).to include('status' => 'processing', 'total_tracks' => 0, 'processed_tracks' => 0)
+    expect(response.body).to include('already running')
+  end
+
+  it 'keeps a waiting reservation when an older per-track job reports late progress' do
+    request_reclassification
+    progress.increment_processed!
+    request_reclassification
+
+    expect(progress.current_status).to eq('processing')
+    expect(TransportationModes::UserReclassifyJob).to have_been_enqueued.exactly(:once).with(user.id)
+  end
+
+  it 'does not write late track progress into a waiting reservation' do
+    request_reclassification
+    before_progress = progress.data
+    expect(progress).not_to receive(:update_progress)
+
+    expect(progress.increment_processed!).to eq(0)
+    expect(progress.data).to eq(before_progress)
+  end
+
+  it 'does not enqueue work when its reservation cannot be stored' do
+    allow(Rails.cache).to receive(:write).and_call_original
+    allow(Rails.cache).to receive(:write).with(progress.cache_key, anything, expires_in: anything).and_return(false)
+
+    expect { request_reclassification }.to raise_error(IOError)
+    expect(TransportationModes::UserReclassifyJob).not_to have_been_enqueued
+  end
+
+  it 'lets a completed run be requested again' do
+    request_reclassification
+    progress.start(total_tracks: 1)
+    progress.complete
+    request_reclassification
+
+    expect(TransportationModes::UserReclassifyJob).to have_been_enqueued.exactly(:twice).with(user.id)
+  end
+
+  it 'releases a reservation after the enqueue adapter rejects the job' do
+    adapter = ActiveJob::Base.queue_adapter
+    allow(adapter).to receive(:enqueue).and_raise(ActiveJob::EnqueueError, 'synthetic enqueue failure')
+
+    expect { request_reclassification }.to raise_error(ActiveJob::EnqueueError)
+    expect(progress.in_progress?).to be false
+    allow(adapter).to receive(:enqueue).and_call_original
+    sign_in user
+    request_reclassification
+
+    expect(TransportationModes::UserReclassifyJob).to have_been_enqueued.exactly(:once).with(user.id)
+  end
+
+  it 'turns the queued reservation into ordinary progress when the worker starts' do
+    request_reclassification
+    TransportationModes::UserReclassifyJob.perform_now(user.id)
+
+    expect(progress.current_status).to eq('completed')
+    request_reclassification
+    expect(TransportationModes::UserReclassifyJob).to have_been_enqueued.exactly(:twice).with(user.id)
+  end
+end