From e6b9189f369675d6adbf2eee009d1837a373d4bf Mon Sep 17 00:00:00 2001 From: Alex Chapin Date: Wed, 8 Jul 2026 01:37:08 -0400 Subject: [PATCH 1/3] Harden RunSimulateDataPoint worker against permanent lock deadlock and stale-job re-dispatch Root cause of 2026-07-07/08 production incident: 50+ analyses with corrupted seed zips caused `write_lock` (dj_jobs/run_simulate_data_point.rb) to release its flock on failure but never delete the lock file, and never write a receipt. The waiter loop only polled for the receipt file and never rechecked the lock, so every subsequent worker pod for the same analysis blocked forever on `initialize_worker`, waiting out the full 8h `initialize_worker_timeout` and then (due to a second bug) falling through to `return true` anyway -- reporting fake success and letting `perform` proceed to simulate against a directory that was never populated. Fleet-wide this produced ~9,988 permanently-zombied worker pods (99.6% of the tier). Fixes: - `write_lock`: rescue any exception in the protected block, delete the lock file before re-raising (previously only released the flock, leaving a file a future worker could stat as "still locked"). Also closes the file handle in `ensure` (minor fd leak). - `initialize_worker`'s receipt-wait loop: detect `lock_holder_gone` (lock file disappears without a receipt appearing -- the crash/failure signature) and return `false` instead of looping to timeout. On genuine `Timeout::Error`, explicitly delete the stale lock file and return `false` instead of falling through to `return true`. - Add a `rescue ::Zip::Error, ::Zlib::Error` branch ahead of the generic retry rescue in the zip-extraction path: fails fast on deterministic corruption (no futile 3x retry), removes the partial `analysis_dir`. Worker-side counterpart to the web-node fail-fast added for seed.zip validation in #841/#844 (a different, once-per-analysis code path; this is the per-datapoint, per-worker-pod download+extract path). - Fix `@@data_point` -> `@data_point` (uninitialized class variable vs the actual instance variable) in the invalid-timeout fallback branches: 3 occurrences in `dj_jobs/run_simulate_data_point.rb`, 2 more in `dj_jobs/urban_opt.rb` (included into the same class). Bug present since ebc2361b (2019), also confirmed on origin/develop. Previously any invalid timeout value raised NameError instead of applying the 28800s default. - `resque_jobs/run_simulate_data_point.rb#perform`: add a narrow `analysis.run_flag == false` guard (checked ahead of the existing completed-status guard) to skip stale queue payloads for analyses that were explicitly stopped/quarantined via `stop_analysis`/ `soft_stop_analysis`. Deliberately not a blanket `status == 'completed'` check, which would break the admin manual-retry workflow (`requeue`/ `requeue_started` do not reset status before re-enqueueing). Adds `server/spec/models/dj_run_simulate_data_point_hardening_spec.rb` (7 examples) and `server/spec/models/resque_run_simulate_data_point_hardening_spec.rb` (7 examples), all executed against a real local Mongo/Redis and passing. Each also has a companion "would fail against the pre-fix code" check to confirm discriminating power. --- .../jobs/dj_jobs/run_simulate_data_point.rb | 42 +++- server/app/jobs/dj_jobs/urban_opt.rb | 4 +- .../resque_jobs/run_simulate_data_point.rb | 14 +- ..._run_simulate_data_point_hardening_spec.rb | 225 ++++++++++++++++++ ..._run_simulate_data_point_hardening_spec.rb | 93 ++++++++ 5 files changed, 369 insertions(+), 9 deletions(-) create mode 100644 server/spec/models/dj_run_simulate_data_point_hardening_spec.rb create mode 100644 server/spec/models/resque_run_simulate_data_point_hardening_spec.rb 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..82ec80853 100644 --- a/server/app/jobs/dj_jobs/run_simulate_data_point.rb +++ b/server/app/jobs/dj_jobs/run_simulate_data_point.rb @@ -199,7 +199,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 +401,45 @@ 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 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 + @sim_logger.info 'waiting for receipt file to appear' sleep 3 end end + rescue ::Timeout::Error + @sim_logger.error "Required analysis objects were not retrieved after #{@data_point.analysis.initialize_worker_timeout} seconds; clearing the stale lock so a future worker does not wait on it." + FileUtils.rm_f write_lock_file + 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 +479,11 @@ 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 any partially extracted files so a later attempt does not skip-and-reuse them. + FileUtils.rm_rf(analysis_dir) + raise "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}" @@ -537,8 +560,15 @@ def write_lock(lock_file_path) lock_file << Time.now yield + 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. + FileUtils.rm_f lock_file_path + raise ensure lock_file.flock(File::LOCK_UN) + lock_file.close end end @@ -610,7 +640,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..96308dd67 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,19 @@ 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. + analysis_stopped = d.analysis.run_flag == false + 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..56a53ddf4 --- /dev/null +++ b/server/spec/models/resque_run_simulate_data_point_hardening_spec.rb @@ -0,0 +1,93 @@ +# ******************************************************************************* +# 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 has run_flag == false (stopped/quarantined by ops)' do + before { analysis.update!(run_flag: false) } + + %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 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 From 42f4517b811fc2dd07bfa3b5641bd7e5dd615bbf Mon Sep 17 00:00:00 2001 From: brianlball Date: Thu, 16 Jul 2026 14:07:08 -0400 Subject: [PATCH 2/3] Close the lock fd before deleting it: Windows cannot unlink open files The deadlock fix deleted files whose handles were still open, which works on Linux but is a silent no-op on Windows (FileUtils.rm_f/rm_rf swallow the EACCES) - and desktop/PAT-on-Windows is the deployment write_lock's own comment says it exists for. Found by running the new hardening specs on Windows: 12/14, with the write_lock-raise and corrupt-zip-cleanup examples failing exactly this way. - write_lock rescue: unlock + close the fd, then rm_f; ensure only closes if the rescue has not already done so - corrupt-zip rescue (runs inside write_lock, lock fd open): sweep the extracted content but leave the lock file for write_lock to delete; raise typed CorruptAnalysisZip (same message) and let initialize_worker's new rescue rmdir the emptied analysis_dir after the lock is gone. Dir.rmdir, not rm_rf: if another worker already re-locked and started re-populating the dir, rmdir fails harmlessly instead of deleting its in-progress files All 14 hardening specs pass on Windows (were 12/14) against a local mongo; message and behavior assertions unchanged for Linux. Co-Authored-By: Claude Fable 5 --- .../jobs/dj_jobs/run_simulate_data_point.rb | 42 ++++++++++++++++--- 1 file changed, 37 insertions(+), 5 deletions(-) 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 82ec80853..afe377de8 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 @@ -481,9 +487,14 @@ def initialize_worker end rescue ::Zip::Error, ::Zlib::Error => e # A corrupt zip is a deterministic failure - retrying cannot succeed (issue #841). - # Remove any partially extracted files so a later attempt does not skip-and-reuse them. - FileUtils.rm_rf(analysis_dir) - raise "Analysis zip for datapoint #{@data_point.id} is corrupt and cannot be extracted: #{e.message}" + # 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}" @@ -546,6 +557,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}" @@ -564,11 +589,18 @@ def write_lock(lock_file_path) # 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 - lock_file.flock(File::LOCK_UN) - lock_file.close + unless lock_file.closed? + lock_file.flock(File::LOCK_UN) + lock_file.close + end end end From 8314b3429d9761b415ebdd811feba54c16839ac3 Mon Sep 17 00:00:00 2001 From: brianlball Date: Thu, 16 Jul 2026 19:27:32 -0400 Subject: [PATCH 3/3] Fix run_flag guard skipping never-started analyses; probe flock before clearing locks Two fixes surfaced by the merge-ref CI run (develop now includes the re-enabled run_simulation feature specs) and the Copilot review: 1. run_flag defaults to false and only flips true when an analysis is started, so guarding on run_flag alone skipped every datapoint submitted directly against a fresh analysis (batch upload + submit_simulation - the resque feature spec's flow, and a real API path). Require a start_time (jobs exist iff the analysis was started) before treating run_flag=false as "stopped". Hardening spec's stopped context now creates the Job a genuinely stopped analysis would have, plus a regression example for the never-started case. 2. The timeout path deleted the lock file unconditionally, but a holder that is alive-but-slow can legitimately outlive a single waiter timeout (each of its steps gets its own initialize_worker_timeout); deleting a live holder's lock lets a third worker initialize the same analysis_dir concurrently. Probe the flock (LOCK_EX|LOCK_NB) and only clear locks nobody holds. The same probe in the wait loop (two consecutive positives, a poll apart) now detects SIGKILL/OOM-killed holders immediately - the incident's failure mode, which the rescue-based cleanup can never see - instead of after the full 8h. Verified: hardening specs 15/15 on Windows + Linux-container; resque feature spec 2/2 and dj 7/7 against the full docker-compose stack on the rebased branch. Co-Authored-By: Claude Fable 5 --- .../jobs/dj_jobs/run_simulate_data_point.rb | 50 ++++++++++++++++++- .../resque_jobs/run_simulate_data_point.rb | 7 ++- ..._run_simulate_data_point_hardening_spec.rb | 27 +++++++++- 3 files changed, 79 insertions(+), 5 deletions(-) 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 afe377de8..4574196fb 100644 --- a/server/app/jobs/dj_jobs/run_simulate_data_point.rb +++ b/server/app/jobs/dj_jobs/run_simulate_data_point.rb @@ -420,6 +420,7 @@ def initialize_worker # 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 @@ -430,13 +431,40 @@ def initialize_worker 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 - @sim_logger.error "Required analysis objects were not retrieved after #{@data_point.analysis.initialize_worker_timeout} seconds; clearing the stale lock so a future worker does not wait on it." - FileUtils.rm_f write_lock_file + # 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 @@ -578,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 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 96308dd67..1c79d6a47 100644 --- a/server/app/jobs/resque_jobs/run_simulate_data_point.rb +++ b/server/app/jobs/resque_jobs/run_simulate_data_point.rb @@ -29,7 +29,12 @@ def self.perform(data_point_id, options = {}) # 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. - analysis_stopped = d.analysis.run_flag == false + # 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) 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 index 56a53ddf4..902e67344 100644 --- a/server/spec/models/resque_run_simulate_data_point_hardening_spec.rb +++ b/server/spec/models/resque_run_simulate_data_point_hardening_spec.rb @@ -37,8 +37,15 @@ def set_data_point_status(status:, status_message: '') end describe '.perform' do - context 'when the owning analysis has run_flag == false (stopped/quarantined by ops)' do - before { analysis.update!(run_flag: false) } + 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 " \ @@ -68,6 +75,22 @@ def set_data_point_status(status:, status_message: '') 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) }