diff --git a/server/app/jobs/dj_jobs/run_simulate_data_point.rb b/server/app/jobs/dj_jobs/run_simulate_data_point.rb index 2323a11ac..4574196fb 100644 --- a/server/app/jobs/dj_jobs/run_simulate_data_point.rb +++ b/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,6 +585,20 @@ 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}" @@ -530,6 +606,24 @@ def initialize_worker 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 @@ -537,8 +631,22 @@ def write_lock(lock_file_path) 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 diff --git a/server/app/jobs/dj_jobs/urban_opt.rb b/server/app/jobs/dj_jobs/urban_opt.rb index 68cb5a37b..fcdca2c2d 100644 --- a/server/app/jobs/dj_jobs/urban_opt.rb +++ b/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) diff --git a/server/app/jobs/resque_jobs/run_simulate_data_point.rb b/server/app/jobs/resque_jobs/run_simulate_data_point.rb index bf3219cd7..1c79d6a47 100644 --- a/server/app/jobs/resque_jobs/run_simulate_data_point.rb +++ b/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 diff --git a/server/spec/models/dj_run_simulate_data_point_hardening_spec.rb b/server/spec/models/dj_run_simulate_data_point_hardening_spec.rb new file mode 100644 index 000000000..982f4506a --- /dev/null +++ b/server/spec/models/dj_run_simulate_data_point_hardening_spec.rb @@ -0,0 +1,225 @@ +# ******************************************************************************* +# OpenStudio(R), Copyright (c) Alliance for Sustainable Energy, LLC. +# See also https://openstudio.net/license +# ******************************************************************************* + +require 'rails_helper' +require 'tmpdir' +require 'zip' + +# Regression specs for the DjJobs::RunSimulateDataPoint#initialize_worker/#write_lock +# hardening: a competing worker's abandoned analysis_zip.lock (the holder crashed mid +# download/extract without ever writing analysis_zip.receipt) used to either deadlock a +# second worker for the full initialize_worker_timeout (default 8h) or, worse, let it fall +# through to a silent "return true" after the wait timed out. A corrupt analysis.zip was +# also retried 3x before failing, wasting time on a deterministic failure. +# +# See dj_run_simulation_data_point_spec.rb for the full Capybara/Resque integration +# coverage of this class (including the original 'creates a write lock that is +# threadsafe' example). These specs isolate the pure Ruby file/state control flow the +# same way that example does -- no depends_resque/js Capybara stack -- mirroring the +# non-feature style already used at the bottom of that file (the '#inline_log_lines' +# block) and the sibling regression style in spec/models/analysis_init_spec.rb. +RSpec.describe DjJobs::RunSimulateDataPoint, type: :model do + around do |example| + @tmp_sim_root = Dir.mktmpdir('dj-run-simulate-data-point-hardening') + previous_sim_root_path = APP_CONFIG['sim_root_path'] + APP_CONFIG['sim_root_path'] = @tmp_sim_root + example.run + ensure + APP_CONFIG['sim_root_path'] = previous_sim_root_path + FileUtils.rm_rf(@tmp_sim_root) if @tmp_sim_root + end + + before do + Project.destroy_all + end + + after do + Project.destroy_all + end + + # Bare, unsaved-then-saved Project/Analysis/DataPoint records -- the same lightweight + # construction the existing 'creates a write lock that is threadsafe' example in + # dj_run_simulation_data_point_spec.rb uses, rather than the heavier FactoryBot chain + # (which attaches a real seed_zip and is not needed here). + let(:project) { Project.new.tap(&:save!) } + let(:analysis) { Analysis.new(project_id: project.id).tap(&:save!) } + let(:data_point) { DataPoint.new(analysis_id: analysis.id).tap(&:save!) } + + def build_job + described_class.new(data_point.id) + end + + describe '#initialize_worker with a stale/abandoned analysis_zip.lock' do + it 'returns false promptly (not true) when the lock holder disappears without ever writing a receipt' do + analysis.initialize_worker_timeout = 20 # long enough that a real Timeout::Error cannot be what resolves this call + analysis.save! + + job = build_job + write_lock_file = File.join(job.send(:analysis_dir), 'analysis_zip.lock') + receipt_file = File.join(job.send(:analysis_dir), 'analysis_zip.receipt') + File.write(write_lock_file, 'held by a worker that is about to crash') + + # Simulate the original holder dying mid-extraction: its cleanup (fix #2, write_lock's + # rescue) removes the lock file, but it never got far enough to write a receipt. + remover = Thread.new do + sleep 0.5 + FileUtils.rm_f(write_lock_file) + end + + start = Time.now + result = job.initialize_worker + elapsed = Time.now - start + remover.join + + expect(result).to be false + expect(elapsed).to be < 10 # well under the 20s configured timeout: it did NOT wait it out + expect(File.exist?(receipt_file)).to be false + end + + it 'returns false and deletes the stale lock file when the wait genuinely times out' do + analysis.initialize_worker_timeout = 2 # short timeout so the test itself stays fast + analysis.save! + + job = build_job + write_lock_file = File.join(job.send(:analysis_dir), 'analysis_zip.lock') + receipt_file = File.join(job.send(:analysis_dir), 'analysis_zip.receipt') + # Lock file present the whole time and no one ever removes it or writes a receipt -- + # a genuinely stuck/hung holder. + File.write(write_lock_file, 'held by a worker that is stuck') + + start = Time.now + result = job.initialize_worker + elapsed = Time.now - start + + expect(result).to be false + expect(elapsed).to be < 10 # proves it timed out at ~2s, not the old 28800s default + expect(File.exist?(write_lock_file)).to be(false), 'stale lock must be removed so a future worker does not inherit the same wait' + expect(File.exist?(receipt_file)).to be false + end + + it 'still returns true when the receipt file appears before the lock disappears or the wait times out (non-broken-path regression)' do + analysis.initialize_worker_timeout = 30 + analysis.save! + + job = build_job + write_lock_file = File.join(job.send(:analysis_dir), 'analysis_zip.lock') + receipt_file = File.join(job.send(:analysis_dir), 'analysis_zip.receipt') + File.write(write_lock_file, 'held by a worker that is about to finish successfully') + + creator = Thread.new do + sleep 0.3 + File.write(receipt_file, Time.now.to_s) + end + + result = job.initialize_worker + creator.join + + expect(result).to be true + expect(File.exist?(write_lock_file)).to be(true), 'the successful holder (not the waiter) owns cleaning up the lock file' + end + end + + describe '#write_lock' do + subject(:job) { described_class.allocate } # write_lock touches no @data_point/@sim_logger state, so a full DB-backed instance is unnecessary + + it 'removes the lock file (not just releasing the flock) when the protected block raises' do + Dir.mktmpdir do |dir| + lock_path = File.join(dir, 'analysis_zip.lock') + + expect { job.write_lock(lock_path) { raise 'boom' } }.to raise_error('boom') + expect(File.exist?(lock_path)).to be false + end + end + + it 'leaves the lock file in place when the protected block succeeds (only the flock is released)' do + Dir.mktmpdir do |dir| + lock_path = File.join(dir, 'analysis_zip.lock') + + result = job.write_lock(lock_path) { 'downloaded ok' } + + expect(result).to eq 'downloaded ok' + expect(File.exist?(lock_path)).to be true + end + end + end + + describe '#initialize_worker with a corrupt analysis.zip download' do + # Seam choice: initialize_worker downloads via a plain inline URI.open (open-uri) call -- + # there is no separately-extractable "download" method to isolate, so the narrowest seam + # that still exercises the *real* production retry/rescue/cleanup logic is to stub + # URI.open to serve a local corrupt-zip fixture instead of hitting the network. Every + # other step (write_lock, extract_archive, the Zip::Error/Zlib::Error rescue, + # FileUtils.rm_rf(analysis_dir), and initialize_worker's own outer rescue) runs as the + # real, unmodified production code. + def build_zlib_corrupt_zip(path) + # A default (DEFLATE) entry, not STORED: rubyzip does not verify CRCs on #extract (only + # Analysis#seed_zip_error's own validator does, see spec/models/analysis_init_spec.rb), + # so corrupting a STORED entry's bytes would extract silently. Zeroing bytes inside a + # DEFLATE-compressed stream breaks the zlib inflate call itself. + Zip::OutputStream.open(path) do |zos| + zos.put_next_entry('data/seed.txt') + zos.write('deterministic corrupt-zip payload text ' * 200) + end + content = File.binread(path) + comp_start = content.index('data/seed.txt') + 'data/seed.txt'.length + bytes = content.bytes + 20.times { |k| bytes[comp_start + k] = 0x00 } + File.binwrite(path, bytes.pack('C*')) + path + end + + it 'fails on the first extraction attempt (no retries), reports zip corruption, and cleans up the partial analysis_dir' do + Dir.mktmpdir('corrupt-zip-fixture') do |fixture_dir| + corrupt_zip = build_zlib_corrupt_zip(File.join(fixture_dir, 'analysis.zip')) + + job = build_job + allow(job).to receive(:sleep) # skip the real stagger/backoff sleeps around the download step + allow(URI).to receive(:open) do |*_args, &blk| + File.open(corrupt_zip, 'rb', &blk) + end + # This is the crux assertion: extraction must be attempted exactly once. If the + # fix regresses to the old 3x retry, this expectation raises immediately on the + # 2nd call (RSpec::Mocks::MockExpectationError), failing the example loudly. + expect(job).to receive(:extract_archive).once.and_call_original + + analysis_dir = job.send(:analysis_dir) + result = job.initialize_worker + + expect(result).to be false + errs = job.instance_variable_get(:@intialize_worker_errs).join + expect(errs).to match(/is corrupt and cannot be extracted/) + expect(errs).not_to match(/failed 3 times/) + expect(Dir.exist?(analysis_dir)).to be false + end + end + end + + describe '#initialize_worker invalid initialize_worker_timeout correction (typo fix regression)' do + it 'corrects a non-positive initialize_worker_timeout on the real Analysis instance (not a throwaway class variable) and proceeds without raising' do + analysis.initialize_worker_timeout = 0 # invalid; must be corrected to 28800 before the Timeout.timeout call reads it + analysis.save! + + job = build_job + write_lock_file = File.join(job.send(:analysis_dir), 'analysis_zip.lock') + receipt_file = File.join(job.send(:analysis_dir), 'analysis_zip.receipt') + # A pre-existing lock (no receipt yet) routes initialize_worker through the + # "wait for receipt" branch, whose Timeout.timeout call is exactly the one that used + # to read the never-corrected class variable's stale invalid value. + File.write(write_lock_file, 'held by another worker') + + creator = Thread.new do + sleep 0.3 + File.write(receipt_file, Time.now.to_s) + end + + result = job.initialize_worker + creator.join + + expect(result).to be true + expect(job.instance_variable_get(:@intialize_worker_errs)).to be_empty + expect(job.instance_variable_get(:@data_point).analysis.initialize_worker_timeout).to eq 28800 + end + end +end diff --git a/server/spec/models/resque_run_simulate_data_point_hardening_spec.rb b/server/spec/models/resque_run_simulate_data_point_hardening_spec.rb new file mode 100644 index 000000000..902e67344 --- /dev/null +++ b/server/spec/models/resque_run_simulate_data_point_hardening_spec.rb @@ -0,0 +1,116 @@ +# ******************************************************************************* +# OpenStudio(R), Copyright (c) Alliance for Sustainable Energy, LLC. +# See also https://openstudio.net/license +# ******************************************************************************* + +require 'rails_helper' + +# Regression specs for ResqueJobs::RunSimulateDataPoint.perform's new run_flag guard. +# +# Incident: a job can sit on the Resque :simulations/:requeued list for a long time +# (worker backlog, HPA scale-down, etc). If ops explicitly stopped/quarantined the owning +# analysis in the meantime (Analysis#stop_analysis / #soft_stop_analysis, e.g. a +# permanently corrupt seed zip), the analysis' run_flag flips to false -- but the old +# completed+completed-normal-only skip check did not look at run_flag at all, so a stale +# queued/started datapoint for an already-stopped analysis would still get a fresh +# DjJobs::RunSimulateDataPoint worker dispatched against it, recreating the very +# failure/deadlock the stop was meant to end. +# +# This mirrors the light, non-Capybara/non-depends_resque style already used at the +# bottom of dj_run_simulation_data_point_spec.rb and in spec/models/analysis_init_spec.rb +# (the sibling hardening commit) -- ResqueJobs::RunSimulateDataPoint.perform is plain Ruby +# dispatch-guard logic, not something that needs a live Resque/Redis broker to exercise; +# DjJobs::RunSimulateDataPoint.new/#perform are mocked so the heavy simulation pipeline +# never actually runs. See resque_run_simulation_data_point_spec.rb for the full +# Capybara/depends_resque integration coverage of this class. +RSpec.describe ResqueJobs::RunSimulateDataPoint, type: :model do + before { Project.destroy_all } + + after { Project.destroy_all } + + let(:project) { Project.new.tap(&:save!) } + let(:analysis) { Analysis.new(project_id: project.id).tap(&:save!) } + let(:data_point) { DataPoint.new(analysis_id: analysis.id).tap(&:save!) } + + def set_data_point_status(status:, status_message: '') + data_point.update!(status: status, status_message: status_message) + end + + describe '.perform' do + context 'when the owning analysis was started and then stopped (run_flag == false with a start time)' do + before do + # A stopped analysis was necessarily started first: starting creates its Job + # records (Analysis#start_time reads them), stop_analysis then flips run_flag. + # run_flag alone is NOT a stop marker - it also defaults to false on analyses + # nobody has started yet (see the regression context below). + Job.new(analysis_id: analysis.id, index: 0, start_time: Time.now.utc, status: 'started').save! + analysis.update!(run_flag: false) + end + + %w[started queued na].each do |status| + it "skips dispatching a fresh worker for a stale '#{status}' datapoint " \ + '(the core regression: this used to slip past the completed-only check)' do + set_data_point_status(status: status) + + expect(DjJobs::RunSimulateDataPoint).not_to receive(:new) + + described_class.perform(data_point.id) + end + end + + it 'still skips even if the datapoint had already reached completed/completed normal' do + set_data_point_status(status: 'completed', status_message: 'completed normal') + + expect(DjJobs::RunSimulateDataPoint).not_to receive(:new) + + described_class.perform(data_point.id) + end + + it 'still skips a completed-but-failed datapoint (run_flag wins regardless of status)' do + set_data_point_status(status: 'completed', status_message: 'datapoint failure') + + expect(DjJobs::RunSimulateDataPoint).not_to receive(:new) + + described_class.perform(data_point.id) + end + end + + context 'when the owning analysis was never started (run_flag still default false, no jobs)' do + it 'still dispatches a worker: a datapoint submitted directly against a fresh analysis must run' do + # Regression: run_flag defaults to false, so guarding on run_flag alone skipped + # every datapoint submitted via batch upload + submit_simulation before the + # analysis was started - caught by resque_run_simulation_data_point_spec.rb + # ('runs a datapoint') in the docker CI job. + set_data_point_status(status: 'na') + + expect(analysis.start_time).to be_nil + worker = instance_double(DjJobs::RunSimulateDataPoint, perform: true) + expect(DjJobs::RunSimulateDataPoint).to receive(:new).with(data_point.id, {}).and_return(worker) + + described_class.perform(data_point.id) + end + end + + context 'when the owning analysis has run_flag == true (normal in-flight processing; existing paths preserved)' do + before { analysis.update!(run_flag: true) } + + it 'still dispatches a worker for a not-yet-completed datapoint (existing "run it" path)' do + set_data_point_status(status: 'started') + + worker = instance_double(DjJobs::RunSimulateDataPoint, perform: true) + expect(DjJobs::RunSimulateDataPoint).to receive(:new).with(data_point.id, {}).and_return(worker) + + described_class.perform(data_point.id) + end + + it 'still skips a datapoint that already completed successfully ' \ + '(existing "skip, already succeeded" path -- e.g. requeued after a spot-instance kill)' do + set_data_point_status(status: 'completed', status_message: 'completed normal') + + expect(DjJobs::RunSimulateDataPoint).not_to receive(:new) + + described_class.perform(data_point.id) + end + end + end +end