From bd081f7a2d38b400759c9e961919f0885df5d987 Mon Sep 17 00:00:00 2001 From: Anthony Volk <14987227+anth-volk@users.noreply.github.com> Date: Tue, 27 Oct 2026 17:36:57 +0400 Subject: [PATCH 1/3] Keep a failed US state job's error for two minutes A failed economy job without a breakdown is stored as an error for the cache lifetime of 30 days (`REFORM_IMPACT_TTL_SECONDS`), and every later request gets that stored error back. The app runs each US state on its ACS-local dataset for a nationwide report's congressional districts and offers Retry for a state that failed, so that Retry could never start a new run. A failed US state job (`country_id` `us`, region `state/...`, which a bare state code such as `ca` is normalized to) now keeps its error for two minutes on its own record, as a failed local-area breakdown already does. Until then every request gets the same error, which ends "A new run for this state can be requested 2 minutes after this failure."; after it, the next request finds no record and submits a new job. `_record_breakdown_failure` becomes `_record_retryable_failure`, which takes the retry note, and `_failure_retry_note` chooses the note for a failed job: the breakdown note, the US state note, or none for every other job. `LOCAL_AREA_BREAKDOWN_FAILURE_SECONDS` becomes `RETRYABLE_FAILURE_SECONDS` (still 120), since both kinds of job use it. US national, congressional district and place failures, UK national failures and typed SPM failures are still kept for 30 days. The specification's 502 description, docs/canonical-spm.md and a changelog fragment describe the change. Tests written first. Before the change, 3 failed for the expected reason: - `test_a_failed_us_state_job_is_returned_for_two_minutes_then_resubmitted` for `state/ca` and for the bare `ca`: the error had no retry note, and 121 seconds after the failure the request still got the stored error, with the simulation API's `run` called once; - `TestHandleExecutionState`'s state test: the error had no retry note and was stored without `keep_seconds=120`. The tests that failed US national, district and place jobs are still replayed, or stored without `keep_seconds`, passed before and after. Co-Authored-By: Claude Opus 5.5 --- changelog.d/us-state-failure-retry.changed.md | 1 + docs/canonical-spm.md | 10 +- policyengine_api/openapi_spec.yaml | 2 +- policyengine_api/services/economy_service.py | 85 ++++++++++---- tests/unit/services/test_economy_service.py | 68 +++++++++++ .../test_economy_service_request_plumbing.py | 108 ++++++++++++++++++ 6 files changed, 246 insertions(+), 28 deletions(-) create mode 100644 changelog.d/us-state-failure-retry.changed.md diff --git a/changelog.d/us-state-failure-retry.changed.md b/changelog.d/us-state-failure-retry.changed.md new file mode 100644 index 000000000..981e6c08e --- /dev/null +++ b/changelog.d/us-state-failure-retry.changed.md @@ -0,0 +1 @@ +A failed US state economy job (`region=state/...`) now keeps its error for two minutes instead of 30 days, and the error message says when the state can be requested again; the next request after that submits a new job. Failed US national, congressional district and place jobs, UK national jobs and typed SPM failures are still kept for 30 days. diff --git a/docs/canonical-spm.md b/docs/canonical-spm.md index 30608e013..794127e3a 100644 --- a/docs/canonical-spm.md +++ b/docs/canonical-spm.md @@ -345,8 +345,14 @@ completes without `kind: "local_areas"`, also returns HTTP 502. Its result is not stored. Its error is kept for two minutes on that job's own record, so every request in that time gets the same error, which says when a new breakdown can be requested; then the record expires and the next request -submits a new job. Other jobs' records are not changed. A failed national job -is still stored and returned again for the cache lifetime of 30 days. +submits a new job. Other jobs' records are not changed. A failed US state job +(`country_id` `us` with region `state/...`, or a bare state code) keeps its +error for two minutes in the same way, with a message that says when the +state can be requested again, so a client can retry one state of a +nationwide report's congressional districts. Any other failed job, such as +a national, congressional district or place job, is still stored and +returned again for the cache lifetime of 30 days, and so is a typed SPM +failure. Before cache access or submission, the API validates the selected worker application's `canonical-spm-v1` capability diff --git a/policyengine_api/openapi_spec.yaml b/policyengine_api/openapi_spec.yaml index 31940fd8d..8a5e47419 100644 --- a/policyengine_api/openapi_spec.yaml +++ b/policyengine_api/openapi_spec.yaml @@ -951,7 +951,7 @@ paths: message: type: string 502: - description: The calculation failed, or the simulation API refused a breakdown=local_areas request it cannot run. The message gives the reason. A failed breakdown=local_areas job, or one whose result is not kind local_areas, keeps its error for two minutes only; after that the next request submits a new job. + description: The calculation failed, or the simulation API refused a breakdown=local_areas request it cannot run. The message gives the reason. A failed breakdown=local_areas job, or one whose result is not kind local_areas, keeps its error for two minutes only; after that the next request submits a new job. A failed US state job (country_id us with region state/..., or a bare state code) also keeps its error for two minutes only; failed US national, congressional district and place jobs keep theirs for 30 days. content: application/json: schema: diff --git a/policyengine_api/services/economy_service.py b/policyengine_api/services/economy_service.py index 57f287f6e..2087c0f5a 100644 --- a/policyengine_api/services/economy_service.py +++ b/policyengine_api/services/economy_service.py @@ -115,14 +115,22 @@ class ImpactStatus(Enum): # 400 from a gateway whose routed .py release does not declare breakdowns, and # 422 from an entry deployment that does not know the field. LOCAL_AREA_BREAKDOWN_REFUSAL_STATUS_CODES = frozenset({400, 422}) -# How long a failed breakdown's error is returned before the next request may -# submit a new job. Short, so the app's Retry works soon, but long enough that -# a client retrying on every 502 does not start a run on the large local-area -# dataset each time. -LOCAL_AREA_BREAKDOWN_FAILURE_SECONDS = 120 +# How long the error of a failed local-area breakdown or US state job is +# returned before the next request may submit a new job. Both run on large +# local-area datasets, where a timeout or memory failure may not repeat, and +# the app offers Retry for both: for a UK breakdown, and for each state it runs +# for a US nationwide report's congressional districts. Short, so Retry works +# soon, but long enough that a client retrying on every 502 does not start a +# run on that dataset each time. Every other failed job's error is kept for the +# cache lifetime of 30 days. +RETRYABLE_FAILURE_SECONDS = 120 LOCAL_AREA_BREAKDOWN_RETRY_NOTE = ( "A new local-area breakdown can be requested " - f"{LOCAL_AREA_BREAKDOWN_FAILURE_SECONDS // 60} minutes after this failure." + f"{RETRYABLE_FAILURE_SECONDS // 60} minutes after this failure." +) +US_STATE_RETRY_NOTE = ( + "A new run for this state can be requested " + f"{RETRYABLE_FAILURE_SECONDS // 60} minutes after this failure." ) # The telemetry simulation_kind of a breakdown run. The simulation API's # telemetry envelope forbids extra fields, so the run is marked here rather @@ -1569,8 +1577,11 @@ def _handle_execution_state( raw_result = self._simulation_gateway.get_execution_result(execution) kind_error = self._breakdown_result_kind_error(setup_options, raw_result) if kind_error is not None: - return self._record_breakdown_failure( - setup_options, reform_impact, kind_error + return self._record_retryable_failure( + setup_options, + reform_impact, + kind_error, + retry_note=LOCAL_AREA_BREAKDOWN_RETRY_NOTE, ) result = self._with_policyengine_bundle( result=raw_result, @@ -1604,9 +1615,13 @@ def _handle_execution_state( error_message = ( f"Simulation entrypoint execution failed: {execution.error}" ) - if setup_options.options.get("breakdown") is not None: - return self._record_breakdown_failure( - setup_options, reform_impact, error_message + retry_note = self._failure_retry_note(setup_options) + if retry_note is not None: + return self._record_retryable_failure( + setup_options, + reform_impact, + error_message, + retry_note=retry_note, ) self._set_reform_impact_error( @@ -1655,31 +1670,51 @@ def _breakdown_result_kind_error( f"{LOCAL_AREA_BREAKDOWN!r}; the result was not stored" ) - def _record_breakdown_failure( + @staticmethod + def _failure_retry_note(setup_options: EconomicImpactSetupOptions) -> str | None: + """Return the retry note for a failed job whose error is kept briefly. + + A local-area breakdown job and a US state job (region ``state/...``, + already normalized from a bare state code) get a note; every other + failed job, including US national, congressional district and place + jobs, gets None, and its error is kept for the cache lifetime. + """ + if setup_options.options.get("breakdown") is not None: + return LOCAL_AREA_BREAKDOWN_RETRY_NOTE + if setup_options.country_id == "us" and setup_options.region.startswith( + "state/" + ): + return US_STATE_RETRY_NOTE + return None + + def _record_retryable_failure( self, setup_options: EconomicImpactSetupOptions, reform_impact: CachedReformImpact, reason: str, + *, + retry_note: str, ) -> EconomicImpactResult: - """Keep a failed breakdown job's error for two minutes, then let it go. - - A breakdown runs on the large local-area dataset, where a timeout or - memory failure may not repeat, so its failure is not kept for the usual - 30 days. This job's own record, found by its execution ID, becomes an - error that expires after ``LOCAL_AREA_BREAKDOWN_FAILURE_SECONDS``. - Until then every request gets the same error, which says when to try - again; after it, the next request finds no record and submits a new - job. No other job's record is read or changed, and no submission claim - is held at this point: it was released once the job was handed off. - Failures of requests without a breakdown are stored as before. + """Keep a failed job's error for two minutes, then let it go. + + Used for local-area breakdown and US state jobs, which run on large + local-area datasets, where a timeout or memory failure may not repeat, + so their failures are not kept for the usual 30 days. This job's own + record, found by its execution ID, becomes an error that expires after + ``RETRYABLE_FAILURE_SECONDS``. Until then every request gets the same + error, which ends with ``retry_note`` to say when to try again; after + it, the next request finds no record and submits a new job. No other + job's record is read or changed, and no submission claim is held at + this point: it was released once the job was handed off. Other failed + jobs, and typed SPM failures, are stored as before. """ - message = f"{reason}. {LOCAL_AREA_BREAKDOWN_RETRY_NOTE}" + message = f"{reason}. {retry_note}" if reform_impact.execution_id: self._set_reform_impact_error( setup_options=setup_options, message=message, execution_id=reform_impact.execution_id, - keep_seconds=LOCAL_AREA_BREAKDOWN_FAILURE_SECONDS, + keep_seconds=RETRYABLE_FAILURE_SECONDS, ) return EconomicImpactResult.error(message=message) diff --git a/tests/unit/services/test_economy_service.py b/tests/unit/services/test_economy_service.py index e4b3507a3..6463c4b07 100644 --- a/tests/unit/services/test_economy_service.py +++ b/tests/unit/services/test_economy_service.py @@ -2677,6 +2677,74 @@ def test__given_modal_failure_with_error_message__includes_error_in_message( call_args = mock_reform_impacts_service.set_error_reform_impact.call_args assert "Simulation timed out" in call_args[1]["message"] + @pytest.mark.usefixtures("mock_logger") + def test__given_failed_us_state_job__keeps_its_error_for_two_minutes( + self, + economy_service, + setup_options, + mock_reform_impacts_service, + ): + # Given + state_setup_options = setup_options.model_copy( + update={"region": "state/ca"} + ) + reform_impact = create_mock_reform_impact(status="computing") + execution = ModalSimulationExecution( + job_id=MOCK_EXECUTION_ID, status="failed", error="out of memory" + ) + + # When + result = economy_service._handle_execution_state( + state_setup_options, "failed", reform_impact, execution + ) + + # Then + expected_message = ( + "Simulation entrypoint execution failed: out of memory. " + "A new run for this state can be requested 2 minutes after this " + "failure." + ) + assert result == EconomicImpactResult.error(expected_message) + mock_reform_impacts_service.set_error_reform_impact.assert_called_once_with( + execution_id=MOCK_EXECUTION_ID, + message=expected_message, + keep_seconds=120, + ) + + @pytest.mark.usefixtures("mock_logger") + @pytest.mark.parametrize( + "region", + [MOCK_REGION, "congressional_district/CA-37", "place/NJ-57000"], + ids=["national", "district", "place"], + ) + def test__given_failed_us_job_outside_a_state__keeps_its_error_for_30_days( + self, + economy_service, + setup_options, + mock_reform_impacts_service, + region, + ): + # Given + region_setup_options = setup_options.model_copy(update={"region": region}) + reform_impact = create_mock_reform_impact(status="computing") + execution = ModalSimulationExecution( + job_id=MOCK_EXECUTION_ID, status="failed", error="out of memory" + ) + + # When + result = economy_service._handle_execution_state( + region_setup_options, "failed", reform_impact, execution + ) + + # Then + expected_message = "Simulation entrypoint execution failed: out of memory" + assert result == EconomicImpactResult.error(expected_message) + # Without keep_seconds the error has the cache's 30-day lifetime. + mock_reform_impacts_service.set_error_reform_impact.assert_called_once_with( + execution_id=MOCK_EXECUTION_ID, + message=expected_message, + ) + @pytest.mark.usefixtures("mock_logger") def test__given_modal_running_state__then_returns_computing_result( self, economy_service, setup_options diff --git a/tests/unit/services/test_economy_service_request_plumbing.py b/tests/unit/services/test_economy_service_request_plumbing.py index 7dcfbc80f..9a5894d42 100644 --- a/tests/unit/services/test_economy_service_request_plumbing.py +++ b/tests/unit/services/test_economy_service_request_plumbing.py @@ -430,6 +430,18 @@ def _uk_request(options: dict, region: str = "uk") -> EconomicImpactRequest: ) +def _us_request(region: str) -> EconomicImpactRequest: + return EconomicImpactRequest( + country_id="us", + policy_id=MOCK_POLICY_ID, + baseline_policy_id=MOCK_BASELINE_POLICY_ID, + region=region, + dataset="default", + options={}, + api_version=MOCK_API_VERSION, + ) + + def _submitted_payload_fields(payload: dict) -> dict: """Return a submitted payload without its metadata and telemetry.""" return { @@ -781,6 +793,102 @@ def test_a_failed_national_job_is_still_replayed( assert [impact.status for impact in stored] == ["error"] +US_STATE_RETRY_NOTE = ( + "A new run for this state can be requested 2 minutes after this failure." +) + + +# A bare state code is the legacy spelling of the same region. +@pytest.mark.parametrize("region", ["state/ca", "ca"]) +def test_a_failed_us_state_job_is_returned_for_two_minutes_then_resubmitted( + economy_service, mock_simulation_entrypoint, reform_impact_backend, region +): + mock_simulation_entrypoint.get_execution_id.side_effect = [ + "first-state-job", + "second-state-job", + ] + _finish_jobs_with( + mock_simulation_entrypoint, status="failed", error="out of memory" + ) + request = _us_request(region) + + economy_service.get_economic_impact(request, time_period="2025") + failed = economy_service.get_economic_impact(request, time_period="2025") + reform_impact_backend.advance(60) + within_two_minutes = economy_service.get_economic_impact( + request, time_period="2025" + ) + reform_impact_backend.advance(61) + retried = economy_service.get_economic_impact(request, time_period="2025") + + expected = EconomicImpactResult.error( + f"Simulation entrypoint execution failed: out of memory. {US_STATE_RETRY_NOTE}" + ) + assert failed == expected + # Within two minutes the stored failure is returned without a new job. + assert within_two_minutes == expected + mock_simulation_entrypoint.get_execution_by_id.assert_called_once_with( + "first-state-job" + ) + # After that the record has expired and the next request submits again. + assert retried == EconomicImpactResult.computing() + assert mock_simulation_entrypoint.run.call_count == 2 + stored = economy_service._reform_impacts.get_recent_reform_impacts(5) + assert [ + (impact.region, impact.execution_id, impact.status) for impact in stored + ] == [("state/ca", "second-state-job", "computing")] + + +def test_a_failed_us_national_job_is_still_replayed( + economy_service, mock_simulation_entrypoint, reform_impact_backend +): + _finish_jobs_with( + mock_simulation_entrypoint, status="failed", error="out of memory" + ) + request = _us_request("us") + + economy_service.get_economic_impact(request, time_period="2025") + failed = economy_service.get_economic_impact(request, time_period="2025") + reform_impact_backend.advance(121) + replayed = economy_service.get_economic_impact(request, time_period="2025") + + expected = EconomicImpactResult.error( + "Simulation entrypoint execution failed: out of memory" + ) + assert failed == expected + assert replayed == expected + mock_simulation_entrypoint.run.assert_called_once() + mock_simulation_entrypoint.get_execution_by_id.assert_called_once() + stored = economy_service._reform_impacts.get_recent_reform_impacts(5) + assert [impact.status for impact in stored] == ["error"] + + +def test_a_failed_us_district_job_is_still_replayed( + economy_service, mock_simulation_entrypoint, reform_impact_backend +): + _finish_jobs_with( + mock_simulation_entrypoint, status="failed", error="out of memory" + ) + request = _us_request("congressional_district/CA-37") + + economy_service.get_economic_impact(request, time_period="2025") + failed = economy_service.get_economic_impact(request, time_period="2025") + reform_impact_backend.advance(121) + replayed = economy_service.get_economic_impact(request, time_period="2025") + + expected = EconomicImpactResult.error( + "Simulation entrypoint execution failed: out of memory" + ) + assert failed == expected + assert replayed == expected + mock_simulation_entrypoint.run.assert_called_once() + mock_simulation_entrypoint.get_execution_by_id.assert_called_once() + stored = economy_service._reform_impacts.get_recent_reform_impacts(5) + assert [(impact.region, impact.status) for impact in stored] == [ + ("congressional_district/CA-37", "error") + ] + + @pytest.mark.usefixtures("uk_package_versions") @pytest.mark.parametrize( ("result", "reported_kind"), From b0e44efc6d26419008d5d7b13bed8653a2daba07 Mon Sep 17 00:00:00 2001 From: Anthony Volk <14987227+anth-volk@users.noreply.github.com> Date: Mon, 2 Nov 2026 18:07:39 +0400 Subject: [PATCH 2/3] Check that a failed US state job's reason is logged, and fix a comment's tense A failed US state job's stored error now expires after two minutes, so its log record is what keeps the reason. EconomicImpactResult.error already logs the message at ERROR; the state test now asserts it, so the reason cannot stop being logged unnoticed. The constants' comment said the app offers Retry for each state it runs for a nationwide report. That app change has not shipped, so it now says the app will offer it. Co-Authored-By: Claude Opus 5.5 --- policyengine_api/services/economy_impact_types.py | 4 ++-- .../unit/services/test_economy_service_impact_execution.py | 6 +++++- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/policyengine_api/services/economy_impact_types.py b/policyengine_api/services/economy_impact_types.py index 183fdd811..18ee3fe13 100644 --- a/policyengine_api/services/economy_impact_types.py +++ b/policyengine_api/services/economy_impact_types.py @@ -42,8 +42,8 @@ class ImpactStatus(Enum): LOCAL_AREA_BREAKDOWN_REFUSAL_STATUS_CODES = frozenset({400, 422}) # How long the error of a failed local-area breakdown or US state job is # returned before the next request may submit a new job. Both run on large -# local-area datasets, where a timeout or memory failure may not repeat, and -# the app offers Retry for both: for a UK breakdown, and for each state it runs +# local-area datasets, where a timeout or memory failure may not repeat. The +# app offers Retry for a UK breakdown, and will offer it for each state it runs # for a US nationwide report's congressional districts. Short, so Retry works # soon, but long enough that a client retrying on every 502 does not start a # run on that dataset each time. Every other failed job's error is kept for the diff --git a/tests/unit/services/test_economy_service_impact_execution.py b/tests/unit/services/test_economy_service_impact_execution.py index 72d6ad941..4d5331ed9 100644 --- a/tests/unit/services/test_economy_service_impact_execution.py +++ b/tests/unit/services/test_economy_service_impact_execution.py @@ -240,12 +240,12 @@ def test__given_modal_failure_with_error_message__includes_error_in_message( call_args = mock_reform_impacts_service.set_error_reform_impact.call_args assert "Simulation timed out" in call_args[1]["message"] - @pytest.mark.usefixtures("mock_logger") def test__given_failed_us_state_job__keeps_its_error_for_two_minutes( self, economy_service, setup_options, mock_reform_impacts_service, + mock_logger, ): # Given state_setup_options = setup_options.model_copy( @@ -273,6 +273,10 @@ def test__given_failed_us_state_job__keeps_its_error_for_two_minutes( message=expected_message, keep_seconds=120, ) + # The stored error expires, so the log keeps the failure's reason. + mock_logger.log_struct.assert_any_call( + {"message": expected_message}, severity="ERROR" + ) @pytest.mark.usefixtures("mock_logger") @pytest.mark.parametrize( From f5150e3dcf85f3096fe0dd805bb812b9950d12e1 Mon Sep 17 00:00:00 2001 From: Anthony Volk <14987227+anth-volk@users.noreply.github.com> Date: Mon, 2 Nov 2026 18:24:48 +0400 Subject: [PATCH 3/3] Check the failed state job's log before building the expected result The test compared the result with EconomicImpactResult.error(...), which logs the same message through the mocked logger, before it checked the log, so the log check passed even when the code under test logged nothing. It now checks the log first; with _record_retryable_failure temporarily changed not to log, it fails. Co-Authored-By: Claude Opus 5.5 --- .../services/test_economy_service_impact_execution.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/tests/unit/services/test_economy_service_impact_execution.py b/tests/unit/services/test_economy_service_impact_execution.py index 4d5331ed9..97d425ee0 100644 --- a/tests/unit/services/test_economy_service_impact_execution.py +++ b/tests/unit/services/test_economy_service_impact_execution.py @@ -267,16 +267,17 @@ def test__given_failed_us_state_job__keeps_its_error_for_two_minutes( "A new run for this state can be requested 2 minutes after this " "failure." ) + # The stored error expires, so the log keeps the failure's reason. + # Checked first: building the expected result below logs too. + mock_logger.log_struct.assert_any_call( + {"message": expected_message}, severity="ERROR" + ) assert result == EconomicImpactResult.error(expected_message) mock_reform_impacts_service.set_error_reform_impact.assert_called_once_with( execution_id=MOCK_EXECUTION_ID, message=expected_message, keep_seconds=120, ) - # The stored error expires, so the log keeps the failure's reason. - mock_logger.log_struct.assert_any_call( - {"message": expected_message}, severity="ERROR" - ) @pytest.mark.usefixtures("mock_logger") @pytest.mark.parametrize(