This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 60a240a4511 Stop airflowctl from resending writes after ambiguous
failures (#74382)
60a240a4511 is described below
commit 60a240a45114ec1d7f4583f051a005eabe706473
Author: Shawn Sun <[email protected]>
AuthorDate: Fri Oct 30 10:49:57 2026 -0700
Stop airflowctl from resending writes after ambiguous failures (#74382)
Retry writes only when the server could not have received them.
Warn when a write may have succeeded despite an error.
---
airflow-ctl/docs/cli-and-env-variables-ref.rst | 5 ++
airflow-ctl/src/airflowctl/api/client.py | 31 ++++++++--
airflow-ctl/src/airflowctl/ctl/cli_config.py | 22 ++++++-
airflow-ctl/tests/airflow_ctl/api/test_client.py | 69 ++++++++++++++++++++++
.../tests/airflow_ctl/ctl/test_cli_config.py | 46 +++++++++++++++
5 files changed, 168 insertions(+), 5 deletions(-)
diff --git a/airflow-ctl/docs/cli-and-env-variables-ref.rst
b/airflow-ctl/docs/cli-and-env-variables-ref.rst
index 4841a801188..0e242dfdb7a 100644
--- a/airflow-ctl/docs/cli-and-env-variables-ref.rst
+++ b/airflow-ctl/docs/cli-and-env-variables-ref.rst
@@ -70,6 +70,11 @@ Environment Variables
only used if you are using the Airflow API and have not set up
authentication using a different method. The default value is 3.
+ Only read-only calls (``GET``, ``HEAD`` and ``OPTIONS``) are retried after
a server error or
+ a lost response. Calls that change state, such as ``POST``, ``PUT``,
``PATCH`` and ``DELETE``,
+ are retried only when the connection to the server could not be
established, so that a
+ change the server may already have applied is not sent twice.
+
.. envvar:: AIRFLOW_CLI_API_RETRY_WAIT_MIN
The minimum amount of time to wait between API retries in seconds.
diff --git a/airflow-ctl/src/airflowctl/api/client.py
b/airflow-ctl/src/airflowctl/api/client.py
index 5ca17140357..71283b798b4 100644
--- a/airflow-ctl/src/airflowctl/api/client.py
+++ b/airflow-ctl/src/airflowctl/api/client.py
@@ -318,12 +318,35 @@ class BearerAuth(httpx.Auth):
yield request
+_REPLAY_SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
+
+
+def _request_method(error: httpx.HTTPError) -> str | None:
+ """
+ Return the method of the request behind ``error``.
+
+ httpx leaves ``.request`` unset for some errors and reading it then
raises, so return None for those.
+ """
+ try:
+ return error.request.method
+ except RuntimeError:
+ return None
+
+
def _should_retry_api_request(exception: BaseException) -> bool:
- """Determine if an API request should be retried based on the exception
type."""
- if isinstance(exception, httpx.HTTPStatusError):
- return exception.response.status_code >= 500
+ """
+ Determine if an API request can be sent again after it failed.
- return isinstance(exception, httpx.RequestError)
+ A request that never reached the server can always be sent again.
Otherwise the server may
+ already have applied it, so only read-only methods are sent again.
+ """
+ if isinstance(exception, (httpx.ConnectError, httpx.ConnectTimeout,
httpx.PoolTimeout)):
+ return True
+ if not isinstance(exception, (httpx.HTTPStatusError, httpx.RequestError)):
+ return False
+ if isinstance(exception, httpx.HTTPStatusError) and
exception.response.status_code < 500:
+ return False
+ return _request_method(exception) in _REPLAY_SAFE_METHODS
def _get_int_env(name: str, default: int, minimum: int = 0) -> int:
diff --git a/airflow-ctl/src/airflowctl/ctl/cli_config.py
b/airflow-ctl/src/airflowctl/ctl/cli_config.py
index 1f5dd5b2691..1d5212f24a5 100755
--- a/airflow-ctl/src/airflowctl/ctl/cli_config.py
+++ b/airflow-ctl/src/airflowctl/ctl/cli_config.py
@@ -38,7 +38,7 @@ import httpx
import rich
import airflowctl.api.datamodels.generated as generated_datamodels
-from airflowctl.api.client import NEW_API_CLIENT, Client, ClientKind,
provide_api_client
+from airflowctl.api.client import NEW_API_CLIENT, Client, ClientKind,
_request_method, provide_api_client
from airflowctl.api.operations import BaseOperations, ServerResponseError
from airflowctl.ctl.console_formatting import AirflowConsole, is_data_sequence
from airflowctl.ctl.utils.yaml import safe_load
@@ -66,6 +66,22 @@ def lazy_load_command(import_path: str) -> Callable:
return command
+_WRITE_METHODS = frozenset({"POST", "PUT", "PATCH", "DELETE"})
+
+
+def _warn_if_write_may_have_applied(error: httpx.HTTPError) -> None:
+ """Warn that a failed write is not retried because the server may already
have applied it."""
+ if isinstance(error, httpx.HTTPStatusError) and error.response.status_code
< 500:
+ return
+ method = _request_method(error)
+ if method in _WRITE_METHODS:
+ rich.print(
+ f"[yellow]The {method} request failed, but it may already have
been applied on the server. "
+ "Check the server state before retrying.[/yellow]",
+ file=sys.stderr,
+ )
+
+
def safe_call_command(function: Callable, args: Iterable[Arg]) -> None:
if os.getenv("AIRFLOW_CLI_DEBUG_MODE") == "true":
rich.print(
@@ -91,11 +107,13 @@ def safe_call_command(function: Callable, args:
Iterable[Arg]) -> None:
f"[red]Server response error: {e}. "
"Please check if the server is running and the API URL is
correct.[/red]"
)
+ _warn_if_write_may_have_applied(e)
sys.exit(1)
except httpx.ReadTimeout as e:
rich.print(f"[red]Read timeout error: {e}[/red]")
if "timed out" in str(e):
rich.print("[red]Please check if the server is running and the API
ready to accept calls.[/red]")
+ _warn_if_write_may_have_applied(e)
sys.exit(1)
except ServerResponseError as e:
rich.print(f"Server response error: {e}")
@@ -105,6 +123,7 @@ def safe_call_command(function: Callable, args:
Iterable[Arg]) -> None:
"Please check the command and its parameters. "
"If you need help, run the command with --help."
)
+ _warn_if_write_may_have_applied(e)
sys.exit(1)
# Must stay below ``ServerResponseError``, which subclasses it. Responses
the client could not
# turn into a ``ServerResponseError`` -- a 3xx, or a 4xx/5xx whose body is
not JSON -- reach us
@@ -116,6 +135,7 @@ def safe_call_command(function: Callable, args:
Iterable[Arg]) -> None:
"[red]The server answered with a redirect, which airflowctl
does not follow. "
"Please check that the API URL you logged in with points at
the Airflow API server.[/red]"
)
+ _warn_if_write_may_have_applied(e)
sys.exit(1)
diff --git a/airflow-ctl/tests/airflow_ctl/api/test_client.py
b/airflow-ctl/tests/airflow_ctl/api/test_client.py
index 049c384b652..a87616efae3 100644
--- a/airflow-ctl/tests/airflow_ctl/api/test_client.py
+++ b/airflow-ctl/tests/airflow_ctl/api/test_client.py
@@ -34,12 +34,15 @@ import structlog.testing
import tenacity
import time_machine
from httpx import URL
+from tenacity import stop_after_attempt, wait_none
from airflowctl.api.client import (
Client,
ClientKind,
Credentials,
_bounded_get_new_password,
+ _request_method,
+ _should_retry_api_request,
get_client,
get_json_error,
provide_api_client,
@@ -250,6 +253,72 @@ class TestClient:
assert client.base_url == URL(expected_base_url)
+class TestRequestRetries:
+ @pytest.fixture(autouse=True)
+ def configure_retries(self, monkeypatch):
+ monkeypatch.setattr(Client.request.retry, "wait", wait_none())
+ monkeypatch.setattr(Client.request.retry, "stop",
stop_after_attempt(3))
+
+ @pytest.mark.parametrize("method", ["GET", "HEAD", "OPTIONS", "POST",
"PUT", "PATCH", "DELETE", "TRACE"])
+ @pytest.mark.parametrize("status_code", [500, 503])
+ @pytest.mark.parametrize("json_response", [True, False])
+ def test_server_errors_only_replay_safe_methods(self, method, status_code,
json_response):
+ requests = []
+
+ def handle_request(request):
+ requests.append(request)
+ if json_response:
+ return httpx.Response(status_code, json={"detail": "boom"})
+ return httpx.Response(status_code, text="boom")
+
+ with Client(base_url="http://localhost";,
transport=httpx.MockTransport(handle_request)) as client:
+ with pytest.raises(httpx.HTTPStatusError):
+ client.request(method, "/operation")
+
+ assert len(requests) == (3 if method in {"GET", "HEAD", "OPTIONS"}
else 1)
+
+ @pytest.mark.parametrize("method", ["GET", "HEAD", "OPTIONS", "POST",
"PUT", "PATCH", "DELETE", "TRACE"])
+ @pytest.mark.parametrize(
+ ("error_type", "before_send"),
+ [
+ (httpx.ConnectError, True),
+ (httpx.ConnectTimeout, True),
+ (httpx.PoolTimeout, True),
+ (httpx.ReadTimeout, False),
+ (httpx.WriteTimeout, False),
+ (httpx.ReadError, False),
+ (httpx.WriteError, False),
+ (httpx.RemoteProtocolError, False),
+ ],
+ )
+ def test_transport_errors_only_replay_safe_requests(self, method,
error_type, before_send):
+ requests = []
+
+ def handle_request(request):
+ requests.append(request)
+ raise error_type("connection failed", request=request)
+
+ with Client(base_url="http://localhost";,
transport=httpx.MockTransport(handle_request)) as client:
+ with pytest.raises(error_type):
+ client.request(method, "/operation")
+
+ assert len(requests) == (3 if before_send or method in {"GET", "HEAD",
"OPTIONS"} else 1)
+
+ @pytest.mark.parametrize("error", [httpx.ReadTimeout("timed out"),
httpx.ReadError("disconnected")])
+ def test_missing_request_does_not_allow_replay(self, error):
+ assert not _should_retry_api_request(error)
+
+
+class TestRequestMethod:
+ def test_returns_the_method_of_the_attached_request(self):
+ error = httpx.ReadTimeout("timed out", request=httpx.Request("PATCH",
"http://localhost/dags/d";))
+
+ assert _request_method(error) == "PATCH"
+
+ def test_returns_none_when_httpx_attached_no_request(self):
+ assert _request_method(httpx.ReadTimeout("timed out")) is None
+
+
class TestCredentials:
@patch.dict(os.environ, {"AIRFLOW_CLI_TOKEN": "TEST_TOKEN"})
@patch.dict(os.environ, {"AIRFLOW_CLI_ENVIRONMENT": "TEST_SAVE"})
diff --git a/airflow-ctl/tests/airflow_ctl/ctl/test_cli_config.py
b/airflow-ctl/tests/airflow_ctl/ctl/test_cli_config.py
index b41790be1df..a0249295109 100644
--- a/airflow-ctl/tests/airflow_ctl/ctl/test_cli_config.py
+++ b/airflow-ctl/tests/airflow_ctl/ctl/test_cli_config.py
@@ -600,6 +600,52 @@ class TestCommandFactory:
class TestCliConfigMethods:
+ @pytest.mark.parametrize("method", ["GET", "HEAD", "OPTIONS", "POST",
"PUT", "PATCH", "DELETE"])
+ @pytest.mark.parametrize(
+ ("failure", "ambiguous"),
+ [
+ (httpx.ReadTimeout, True),
+ (httpx.ReadError, True),
+ (httpx.RemoteProtocolError, True),
+ ("json-500", True),
+ ("plain-503", True),
+ ("json-400", False),
+ ("plain-401", False),
+ ("redirect", False),
+ ("connection", False),
+ ],
+ )
+ def test_safe_call_command_warns_only_for_ambiguous_writes(self, method,
failure, ambiguous, capsys):
+ request = httpx.Request(method, "http://localhost/operation";)
+ error: Exception
+ if failure == "connection":
+ error = AirflowCtlConnectionException("Connection refused. Is the
API server running?")
+ elif isinstance(failure, str):
+ status_code = {
+ "json-500": 500,
+ "plain-503": 503,
+ "json-400": 400,
+ "plain-401": 401,
+ "redirect": 302,
+ }[failure]
+ response = httpx.Response(status_code, request=request,
json={"detail": "boom"})
+ error_type = ServerResponseError if failure.startswith("json-")
else httpx.HTTPStatusError
+ error = error_type("request failed", request=request,
response=response)
+ else:
+ error = failure("request failed", request=request)
+
+ def raise_error(_args):
+ raise error
+
+ with pytest.raises(SystemExit) as ctx:
+ safe_call_command(raise_error, args=argparse.Namespace())
+
+ assert ctx.value.code == 1
+ stderr = " ".join(capsys.readouterr().err.split())
+ warning_expected = ambiguous and method in {"POST", "PUT", "PATCH",
"DELETE"}
+ assert ("may already have been applied" in stderr) is warning_expected
+ assert ("Check the server state before retrying" in stderr) is
warning_expected
+
@pytest.mark.parametrize(
"raised_exception",
[
