[1fanwang]

A short metadata database hiccup makes a connection or variable look missing. The lookup treats the database error like any other backend failure, moves on to the next backend, and ends with "isn't defined" or "does not exist". Tasks get the same answer: the server reports the failed lookup as not found, and the worker reads any server error as not found too. With the secrets cache enabled, the miss is cached, so the key keeps looking missing after the database recovers. A variable read with a default quietly gets the default.

What changes:

• A database error no longer counts as "missing". The lookup still asks the remaining backends and returns the secret if one has it. If none does, the database error is raised. An access-denied answer still stops the search.
• A lookup that failed on the database is not cached.
• The metadata backend no longer retries. The retry rolled back the database session between attempts, and callers can now pass their own session (Allow Variable.get to reuse the caller's database session #71968), so that rollback could discard their work.
• In tasks, only an explicit not-found answer from the server means missing. Other server errors are raised after the remaining backends are tried, including under dag.test(), where the server error reaches the task directly rather than as a message. Failures in other secrets backends still end in not found, as before.

Behaviour change: when the database fails and no backend has the key, reading a variable with a default raises the error instead of returning the default, in tasks and on the server.

Testing Done

Lookups against the metadata database

A real SQLite metadata database, migrated with airflow db migrate and seeded with a variable and a connection, with the secrets cache enabled. A second connection holds an exclusive lock, so every read fails with "database is locked", then releases it. The later-backend line puts the environment variables backend after the metastore (in the default chain the metastore is last).

$ AIRFLOW__SECRETS__USE_CACHE=True uv run --project airflow-core python e2e_locked_db.py

Reproducer source: e2e_locked_db.py

"""Drive Variable/Connection lookups against a real SQLite metadata DB while another connection holds a lock.

No mocks on the lookup path: the metastore query hits a real ``database is locked`` OperationalError.
"""

from __future__ import annotations

import os
import sqlite3
import time
from unittest import mock

from sqlalchemy import delete

from airflow.models import Connection, Variable
from airflow.sdk import SecretCache
from airflow.secrets.environment_variables import EnvironmentVariablesBackend
from airflow.secrets.metastore import MetastoreBackend
from airflow.settings import engine
from airflow.utils.session import create_session

DB = engine.url.database


def attempt(label, func):
    started = time.monotonic()
    try:
        result = func()
    except Exception as error:
        message = str(error).splitlines()[0]
        result = f"raised {type(error).__module__}.{type(error).__name__}: {message}"
    else:
        result = f"returned {result!r}"
    print(f"{label}: {result} ({time.monotonic() - started:.1f}s)")


def main() -> None:
    with create_session() as session:
        session.execute(delete(Variable).where(Variable.key.in_(["api_key", "region"])))
        session.execute(delete(Connection).where(Connection.conn_id == "warehouse"))
    Variable.set("api_key", "s3cr3t")
    Variable.set("region", "from-metastore")
    with create_session() as session:
        session.add(Connection(conn_id="warehouse", conn_type="postgres", host="db.internal"))

    SecretCache.init()
    print(f"db={os.path.basename(DB)} cache_enabled={SecretCache._cache is not None}")

    engine.dispose()  # close Airflow's pooled connections so the lock below can be taken
    lock = sqlite3.connect(DB, isolation_level=None)
    # The metadata DB runs in WAL mode, where readers ignore BEGIN EXCLUSIVE; exclusive locking mode plus a
    # write makes every other connection get "database is locked" until this connection closes.
    lock.execute("PRAGMA locking_mode=EXCLUSIVE")
    lock.execute("BEGIN EXCLUSIVE")
    lock.execute("UPDATE variable SET val = val WHERE 0")
    print("-- another connection holds an exclusive lock on the metadata DB")
    attempt("Variable.get('api_key')", lambda: Variable.get("api_key"))
    attempt(
        "Variable.get('api_key', default_var='default')", lambda: Variable.get("api_key", default_var="default")
    )
    attempt(
        "Connection.get_connection_from_secrets('warehouse')",
        lambda: Connection.get_connection_from_secrets("warehouse").host,
    )
    with (
        mock.patch.dict(os.environ, {"AIRFLOW_VAR_REGION": "from-env-backend"}),
        mock.patch(
            "airflow.models.variable.ensure_secrets_loaded",
            return_value=[MetastoreBackend(), EnvironmentVariablesBackend()],
        ),
    ):
        attempt(
            "metastore first, env backend after: Variable.get_variable_from_secrets('region')",
            lambda: Variable.get_variable_from_secrets("region"),
        )
    lock.execute("ROLLBACK")
    lock.close()

    print("-- lock released")
    attempt("Variable.get('api_key')", lambda: Variable.get("api_key"))
    attempt(
        "Connection.get_connection_from_secrets('warehouse')",
        lambda: Connection.get_connection_from_secrets("warehouse").host,
    )


if __name__ == "__main__":
    main()

Before this change:

-- another connection holds an exclusive lock on the metadata DB
Variable.get('api_key'): raised builtins.KeyError: 'Variable api_key does not exist.' (17.4s)
Variable.get('api_key', default_var='default'): returned 'default' (0.0s)
Connection.get_connection_from_secrets('warehouse'): raised airflow.sdk.exceptions.AirflowNotFoundException: The conn_id `warehouse` isn't defined (17.1s)
metastore first, env backend after: Variable.get_variable_from_secrets('region'): returned 'from-env-backend' (16.5s)
-- lock released
Variable.get('api_key'): raised builtins.KeyError: 'Variable api_key does not exist.' (0.0s)
Connection.get_connection_from_secrets('warehouse'): returned 'db.internal' (0.0s)

After:

-- another connection holds an exclusive lock on the metadata DB
Variable.get('api_key'): raised sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) database is locked (5.6s)
Variable.get('api_key', default_var='default'): raised sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) database is locked (5.2s)
Connection.get_connection_from_secrets('warehouse'): raised sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) database is locked (5.2s)
metastore first, env backend after: Variable.get_variable_from_secrets('region'): returned 'from-env-backend' (5.3s)
-- lock released
Variable.get('api_key'): returned 's3cr3t' (0.0s)
Connection.get_connection_from_secrets('warehouse'): returned 'db.internal' (0.0s)

Lookups from a task

A task run with dag.test() reads a variable and a connection while the matching table is renamed away, so the server's query fails with a real "no such table" error. dag.test() runs the task in-process: task, supervisor, Execution API, then the same SQLite database.

$ AIRFLOW__CORE__LOAD_EXAMPLES=False uv run --project airflow-core python e2e_worker_path.py

Reproducer source: e2e_worker_path.py

"""Read a variable and a connection from a task, through the worker path, while a metadata table is unavailable.

dag.test() runs the task under the in-process supervisor: the task asks the supervisor, the supervisor calls the
Execution API app, and the API reads the real SQLite metadata DB. The table is renamed away for the duration of
the read, so the database answers with a real "no such table" error.
"""

from __future__ import annotations

import sqlite3
import sys
import time
from datetime import datetime

from sqlalchemy import delete

from airflow.models import Connection as DbConnection, Variable as DbVariable
from airflow.sdk import DAG, task
from airflow.settings import engine
from airflow.utils.session import create_session

DB = engine.url.database


def attempt(label, func):
    started = time.monotonic()
    try:
        result = func()
    except Exception as error:
        message = str(error).splitlines()[0]
        result = f"raised {type(error).__module__}.{type(error).__name__}: {message}"
    else:
        result = f"returned {result!r}"
    print(f"{label}: {result} ({time.monotonic() - started:.1f}s)", file=sys.__stdout__, flush=True)


def rename(table: str, *, online: bool) -> None:
    source, target = (f"{table}_offline", table) if online else (table, f"{table}_offline")
    con = sqlite3.connect(DB, isolation_level=None, timeout=30)
    con.execute(f"ALTER TABLE {source} RENAME TO {target}")
    con.close()


with DAG("secrets_worker_path", schedule=None, start_date=datetime(2026, 1, 1), catchup=False) as dag:

    @task
    def read_secrets():
        from airflow.sdk import Connection, Variable

        attempt("tables online: Variable.get('api_key')", lambda: Variable.get("api_key"))
        rename("variable", online=False)
        try:
            attempt("variable table offline: Variable.get('api_key')", lambda: Variable.get("api_key"))
            attempt(
                "variable table offline: Variable.get('api_key', default='fallback')",
                lambda: Variable.get("api_key", default="fallback"),
            )
        finally:
            rename("variable", online=True)
        rename("connection", online=False)
        try:
            attempt(
                "connection table offline: Connection.get('warehouse')",
                lambda: Connection.get("warehouse").host,
            )
        finally:
            rename("connection", online=True)
        attempt("tables back: Variable.get('api_key')", lambda: Variable.get("api_key"))
        attempt("tables back: Connection.get('warehouse')", lambda: Connection.get("warehouse").host)

    read_secrets()


if __name__ == "__main__":
    with create_session() as session:
        session.execute(delete(DbVariable).where(DbVariable.key == "api_key"))
        session.execute(delete(DbConnection).where(DbConnection.conn_id == "warehouse"))
    DbVariable.set("api_key", "s3cr3t")
    with create_session() as session:
        session.add(DbConnection(conn_id="warehouse", conn_type="postgres", host="db.internal"))
    print(f"db={DB.rsplit('/', 1)[-1]}", file=sys.__stdout__, flush=True)
    dag.test()

Before this change:

tables online: Variable.get('api_key'): returned 's3cr3t' (0.0s)
variable table offline: Variable.get('api_key'): raised airflow.sdk.exceptions.AirflowRuntimeError: VARIABLE_NOT_FOUND: {'message': 'Variable api_key not found'} (0.5s)
variable table offline: Variable.get('api_key', default='fallback'): returned 'fallback' (1.4s)
connection table offline: Connection.get('warehouse'): raised airflow.sdk.exceptions.AirflowNotFoundException: The conn_id `warehouse` isn't defined (0.5s)
tables back: Variable.get('api_key'): returned 's3cr3t' (0.0s)
tables back: Connection.get('warehouse'): returned 'db.internal' (0.0s)

After:

tables online: Variable.get('api_key'): returned 's3cr3t' (0.0s)
variable table offline: Variable.get('api_key'): raised airflow.sdk.exceptions.AirflowRuntimeError: API_SERVER_ERROR: {'status_code': 500, 'message': 'Server returned error', 'detail': {'reason': 'Database error', 'statement': 'hidden', 'orig_error': 'hidden', 'message': 'Serious error when handling your request. Check logs for more details - you will find it in api server when you look for ID bXIRHrOx'}} (0.1s)
variable table offline: Variable.get('api_key', default='fallback'): raised airflow.sdk.exceptions.AirflowRuntimeError: API_SERVER_ERROR: {'status_code': 500, 'message': 'Server returned error', 'detail': {'reason': 'Database error', 'statement': 'hidden', 'orig_error': 'hidden', 'message': 'Serious error when handling your request. Check logs for more details - you will find it in api server when you look for ID 9yfDJZJg'}} (0.0s)
connection table offline: Connection.get('warehouse'): raised airflow.sdk.exceptions.AirflowRuntimeError: API_SERVER_ERROR: {'status_code': 500, 'message': 'Server returned error', 'detail': {'reason': 'Database error', 'statement': 'hidden', 'orig_error': 'hidden', 'message': 'Serious error when handling your request. Check logs for more details - you will find it in api server when you look for ID 2P74pkFz'}} (0.0s)
tables back: Variable.get('api_key'): returned 's3cr3t' (0.0s)
tables back: Connection.get('warehouse'): returned 'db.internal' (0.0s)

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: GitHub Copilot CLI (Auto model selection) following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[ferruzzi]

First, I'm sorry this sat over a month without any reviews, we really got swarmed with AI-generated contributions and didn't (still don't, TBH) have a good way to filter them out.

That said, nice troubleshooting. A permanent 404 for a secret that exists is a real bug, and the None caching makes it worse. But I think the retry is a bandaid rather than the fix. Once you exhaust max_db_retries, you still get the same 404 and it is still cached for cache_ttl_seconds.

The root cause looks like it is that both callers accept a database failure as missing data. get_connection_from_secrets already distinguishes one class of error:

except AirflowSecretsBackendAccessDenied:
    # Authoritative deny — must NOT fall through to a less-restrictive backend.
    raise
except Exception:
    ...fall through...

...and then, after the loop, raise AirflowNotFoundException(f"The conn_id {conn_id} isn't defined"). (DBAPIError, StaleDataError) belongs in that carve-out for exactly the same reason: it isn't an answer about whether the secret exists.

Here's what I'd suggest: keep trying the remaining backends since a metastore blip shouldn't stop Vault from answering, but track that a backend errored. Then, if nothing is found, surface a database error rather than not-found, and skip the save_variable(key, None) in that case.

One more reason to prefer that over the decorator: retry_db_transaction calls session.rollback(), and MetastoreBackend.get_variable takes a session argument that callers will be supplying once #71968 lands. That means it can roll back a transaction it doesn't own, which is the problem #71968 exists to avoid. Fixing the classification sidesteps the session question entirely.

Happy to be argued out of it if you think the retry is the better solution. It's an improvement either way, just not one that closes the 404 bug.