Harden RunSimulateDataPoint worker against permanent lock deadlock and stale-job re-dispatch

server/app/jobs/dj_jobs/run_simulate_data_point.rb

@@ -14,6 +14,12 @@ class RunSimulateDataPoint
     MAX_INLINE_SDP_LOG_BYTES = 5_000_000
     MAX_INLINE_SDP_LOG_LINES = 5_000
 
+    # Raised (instead of a bare String) when the downloaded analysis.zip cannot be
+    # extracted, so initialize_worker's cleanup can distinguish this deterministic
+    # failure and remove the analysis_dir AFTER write_lock has closed and deleted
+    # the lock file - see the CorruptAnalysisZip rescue in initialize_worker.
+    class CorruptAnalysisZip < StandardError; end
+
     def initialize(data_point_id, options = {})
       @data_point = DataPoint.find(data_point_id)
       @options = options
@@ -199,7 +205,7 @@ def perform
             # check for a valid timeout value
             unless @data_point.analysis.run_workflow_timeout.positive?
               @sim_logger.warn "run_workflow_timeout option: #{@data_point.analysis.run_workflow_timeout} is not valid.  Using 28800s instead."
-              @@data_point.analysis.run_workflow_timeout = 28800
+              @data_point.analysis.run_workflow_timeout = 28800
             end
             Timeout.timeout(@data_point.analysis.run_workflow_timeout) do
               Process.wait(pid)
@@ -401,27 +407,73 @@ def initialize_worker
       # add check for a valid timeout value
       unless @data_point.analysis.initialize_worker_timeout.positive?
         @sim_logger.warn "initialize_worker_timeout option: #{@data_point.analysis.initialize_worker_timeout} is not valid.  Using 28800s instead."
-        @@data_point.analysis.initialize_worker_timeout = 28800
+        @data_point.analysis.initialize_worker_timeout = 28800
       end
       # This block makes this code threadsafe for non-docker deployments, i.e. desktop usage
       if File.exist? write_lock_file
         @sim_logger.info 'write_lock_file exists, checking & waiting for receipt file'
 
-        # wait until receipt file appears then return or error
+        # Wait until receipt file appears, then return or error. Also bail out early if the
+        # lock file itself disappears without a receipt ever showing up: that means the
+        # original holder died (crashed, OOM-killed, evicted, hit a deterministic failure
+        # like a corrupt seed zip) partway through without completing initialization.
+        # Waiting the full initialize_worker_timeout (default 8h) for a receipt that will
+        # never come just wastes a worker slot for no reason.
+        lock_holder_gone = false
+        abandoned_probes = 0
         begin
           Timeout.timeout(@data_point.analysis.initialize_worker_timeout) do
             loop do
               break if File.exist? receipt_file
 
+              unless File.exist? write_lock_file
+                lock_holder_gone = true
+                break
+              end
+
+              # flock is released by the OS when its holder dies, even on SIGKILL/OOM -
+              # deaths write_lock's own rescue cleanup never sees. A lock file whose flock
+              # is acquirable is abandoned; detect that here instead of waiting out the
+              # full initialize_worker_timeout. Require two consecutive positive probes
+              # (a poll interval apart) so a holder's open->flock startup gap can never
+              # read as abandonment.
+              if lock_abandoned?(write_lock_file)
+                abandoned_probes += 1
+                if abandoned_probes >= 2
+                  @sim_logger.error 'write_lock_file exists but its flock is not held; the original holder died without cleaning up. Clearing the abandoned lock and failing this attempt so it can be retried.'
+                  FileUtils.rm_f write_lock_file
+                  lock_holder_gone = true
+                  break
+                end
+              else
+                abandoned_probes = 0
+              end
+
               @sim_logger.info 'waiting for receipt file to appear'
               sleep 3
             end
           end
+        rescue ::Timeout::Error
+          # Only clear the lock if nothing is actually holding it: a holder that is alive
+          # but slow (each of its download/extract steps gets its own
+          # initialize_worker_timeout, so it can legitimately outlive this single wait)
+          # still owns the file, and deleting it would let a third worker re-create the
+          # lock on a new inode and initialize the same analysis_dir concurrently.
+          if lock_abandoned?(write_lock_file)
+            @sim_logger.error "Required analysis objects were not retrieved after #{@data_point.analysis.initialize_worker_timeout} seconds and the lock is not held; clearing the stale lock so a future worker does not wait on it."
+            FileUtils.rm_f write_lock_file
+          else
+            @sim_logger.error "Required analysis objects were not retrieved after #{@data_point.analysis.initialize_worker_timeout} seconds; the lock holder is still alive, leaving its lock in place."
+          end
+          return false
+        end
 
+        if File.exist? receipt_file
           @sim_logger.info 'receipt_file appeared, moving on'
           return true
-        rescue ::Timeout::Error
-          @sim_logger.error "Required analysis objects were not retrieved after #{@data_point.analysis.initialize_worker_timeout} seconds."
+        else
+          @sim_logger.error 'write_lock_file was removed by its holder without a receipt_file appearing; the original holder failed to initialize. Failing this attempt so it can be retried.' if lock_holder_gone
+          return false
         end
       else
         # Try to download the analysis zip, but first lock simultanious threads
@@ -461,6 +513,16 @@ def initialize_worker
               #OpenStudio::Workflow.extract_archive(download_file, analysis_dir)
               extract_archive(download_file, analysis_dir)
             end
+          rescue ::Zip::Error, ::Zlib::Error => e
+            # A corrupt zip is a deterministic failure - retrying cannot succeed (issue #841).
+            # Remove partially extracted files (while still holding the lock) so a later
+            # attempt does not skip-and-reuse them. The lock file itself must survive this
+            # sweep: its fd is still open here, Windows cannot delete an open file, and
+            # FileUtils.rm_rf swallows that failure silently. write_lock's rescue unlocks,
+            # closes, and deletes it on the way out, and initialize_worker's
+            # CorruptAnalysisZip rescue then removes the emptied directory.
+            FileUtils.rm_rf(Dir.glob("#{analysis_dir}/*") - [write_lock_file])
+            raise CorruptAnalysisZip, "Analysis zip for datapoint #{@data_point.id} is corrupt and cannot be extracted: #{e.message}"
           rescue StandardError => e
             retry if extract_count < extract_max_count
             raise "Extraction of the analysis.zip file failed #{extract_max_count} times with error #{e.message}"
@@ -523,22 +585,68 @@ def initialize_worker
 
       @sim_logger.info 'Finished worker initialization'
       return true
+    rescue CorruptAnalysisZip => e
+      # write_lock's rescue has already closed and deleted the lock file (the only fd that
+      # was still open under analysis_dir), and the in-lock sweep removed the extracted
+      # content. Dir.rmdir rather than rm_rf: if another worker has already re-locked and
+      # begun re-populating the directory, rmdir fails harmlessly instead of deleting its
+      # in-progress files.
+      begin
+        Dir.rmdir(analysis_dir)
+      rescue SystemCallError
+        @sim_logger.warn "Could not remove #{analysis_dir} after corrupt-zip cleanup; leaving it for the next worker."
+      end
+      @sim_logger.error "Error in initialize_worker in #{__FILE__} with message #{e.message}; #{e.backtrace.join("\n")}"
+      @intialize_worker_errs << "#{e.message}; #{e.backtrace.first}"
+      return false
     rescue StandardError => e
       @sim_logger.error "Error in initialize_worker in #{__FILE__} with message #{e.message}; #{e.backtrace.join("\n")}"
       @intialize_worker_errs << "#{e.message}; #{e.backtrace.first}"
       return false
     end
 
     # Simple method to write a lock file in order for competing threads to wait before continuing.
+    # True when the lock file exists but no process holds its flock - i.e. the
+    # holder died (SIGKILL/OOM/eviction) without running write_lock's cleanup.
+    # The probe takes and immediately releases the flock; a live holder keeps
+    # LOCK_EX for its whole critical section, so LOCK_NB fails against it.
+    def lock_abandoned?(lock_file_path)
+      File.open(lock_file_path, 'r') do |f|
+        if f.flock(File::LOCK_EX | File::LOCK_NB)
+          f.flock(File::LOCK_UN)
+          true
+        else
+          false
+        end
+      end
+    rescue Errno::ENOENT
+      # Deleted between our existence check and the open: holder is gone.
+      true
+    end
+
     def write_lock(lock_file_path)
       lock_file = File.open(lock_file_path, 'a')
       begin
         lock_file.flock(File::LOCK_EX)
         lock_file << Time.now
 
         yield
-      ensure
+      rescue StandardError
+        # The protected block failed before a receipt file could be written. Delete the lock
+        # file (not just release the flock) so a future attempt does not see a stale lock and
+        # wait up to initialize_worker_timeout (default 8h) for a receipt that will never come.
+        # Unlock and close BEFORE deleting: Windows (the desktop/PAT deployment this lock
+        # exists for) cannot unlink an open file, and FileUtils.rm_f swallows that failure
+        # silently, leaving the stale lock - and the deadlock - in place.
         lock_file.flock(File::LOCK_UN)
+        lock_file.close
+        FileUtils.rm_f lock_file_path
+        raise
+      ensure
+        unless lock_file.closed?
+          lock_file.flock(File::LOCK_UN)
+          lock_file.close
+        end
       end
     end
 
@@ -610,7 +718,7 @@ def upload_file(filename, type, display_name = nil, _content_type = nil)
       # add check for a valid timeout value
       unless @data_point.analysis.upload_results_timeout.positive?
         @sim_logger.warn "upload_results_timeout option: #{@data_point.analysis.upload_results_timeout} is not valid.  Using 28800s instead."
-        @@data_point.analysis.upload_results_timeout = 28800
+        @data_point.analysis.upload_results_timeout = 28800
       end
       begin
         Timeout.timeout(@data_point.analysis.upload_results_timeout) do

server/app/jobs/dj_jobs/urban_opt.rb

@@ -55,7 +55,7 @@ def run_urbanopt(uo_simulation_log, uo_process_log)
         # add check for a valid timeout value
         unless @data_point.analysis.run_workflow_timeout.positive?
           @sim_logger.warn "run_workflow_timeout option: #{@data_point.analysis.run_workflow_timeout} is not valid.  Using 28800s instead."
-          @@data_point.analysis.run_workflow_timeout = 28800
+          @data_point.analysis.run_workflow_timeout = 28800
         end
         Timeout.timeout(@data_point.analysis.run_workflow_timeout) do
           Process.wait(pid)
@@ -222,7 +222,7 @@ def run_urbanopt(uo_simulation_log, uo_process_log)
         # add check for a valid timeout value
         unless @data_point.analysis.run_workflow_timeout.positive?
           @sim_logger.warn "run_workflow_timeout option: #{@data_point.analysis.run_workflow_timeout} is not valid.  Using 28800s instead."
-          @@data_point.analysis.run_workflow_timeout = 28800
+          @data_point.analysis.run_workflow_timeout = 28800
         end
         Timeout.timeout(@data_point.analysis.run_workflow_timeout) do
           Process.wait(pid)

server/app/jobs/resque_jobs/run_simulate_data_point.rb

@@ -22,7 +22,24 @@ def self.perform(data_point_id, options = {})
       # There is a case where that worker completes a successful job, before the requeued DP starts.
       # In that case, we should skip re-running that DP because it was both completed and completed normal already.
       # If its a requeued failed job, then that should still get re-run
-      if !(statuses[:status] == 'completed' && statuses[:status_message] == 'completed normal')
+      #
+      # A job can also sit on the :simulations/:requeued Resque list for a long time (worker
+      # backlog, HPA scale-down, etc). If the owning analysis was explicitly stopped in the
+      # meantime (Analysis#stop_analysis/#soft_stop_analysis, e.g. ops quarantining a batch with
+      # a permanently corrupt seed zip), run_flag is false and this stale payload must not
+      # dispatch a fresh worker download/extract against it - that just recreates the same
+      # failure or deadlock the stop was meant to end.
+      # run_flag defaults to false and only flips true when the analysis is started, so
+      # run_flag alone cannot distinguish "ops stopped this analysis" from "nobody has
+      # started it yet" - and datapoints CAN legitimately be submitted against a
+      # never-started analysis (batch datapoint upload + direct submit_simulation).
+      # start_time comes from the analysis' jobs, which exist iff it was started.
+      analysis_stopped = d.analysis.run_flag == false && !d.analysis.start_time.nil?
+      if analysis_stopped
+        msg = "SKIPPING #{data_point_id} because analysis #{d.analysis_id} has run_flag=false (analysis was stopped)"
+        d.add_to_rails_log(msg)
+        puts msg
+      elsif !(statuses[:status] == 'completed' && statuses[:status_message] == 'completed normal')
         msg = "RUNNING DJ: #{statuses[:status]} and #{statuses[:status_message]}"
         d.add_to_rails_log(msg)
         puts msg