Deadline callback migration 0094 shallow-encodes nested callback kwargs → scheduler CrashLoopBackOff (KeyError('__var'))

[seanmuth]

Apache Airflow version

3.2.0+ (root cause is migration 0094, airflow_version = "3.2.0"; buggy code is still present on main)

What happened

On a deployment using Deadline Alerts with an AsyncCallback whose kwargs contain a nested dict, the scheduler enters CrashLoopBackOff. Every scheduler loop crashes while deserializing an associated callback row in the deadline-processing query:

File ".../airflow/jobs/scheduler_job_runner.py", line 1808, in _run_scheduler_loop
    for deadline in session.scalars( ... selectinload(Deadline.callback) ... )
File ".../airflow/utils/sqlalchemy.py", line 221, in process_result_value
    return BaseSerialization.deserialize(value)
File ".../airflow/serialization/serialized_objects.py", line 632, in deserialize
    return {k: cls.deserialize(v) for k, v in var.items()}
File ".../airflow/serialization/serialized_objects.py", line 629, in deserialize
    var = encoded_var[Encoding.VAR]
KeyError: <Encoding.VAR: '__var'>

The crash happens before the scheduler heartbeats, so it presents as failing liveness probes / restarts with no OOM and no other logged exception — easy to misdiagnose as a probe or DB problem.

Root cause

Migration 0094_3_2_0_replace_deadline_inline_callback_with_fkey.py (revision e812941398f4) moves the old inline deadline callback into the callback table. callback.data is an ExtendedJSON column, whose read path runs BaseSerialization.deserialize (utils/sqlalchemy.py), which requires every nested dict to be wrapped as {"__type": "dict", "__var": {...}}.

But the migration hand-builds callback.data and only wraps the top level, embedding the old callback's kwargs raw:

-- _upgrade_postgresql(): raw nested kwargs pulled from the old inline (SDK-serde) callback
COALESCE(NULLIF(d.callback::jsonb->'__data__'->'kwargs', 'null'::jsonb), '{}'::jsonb) AS cb_kwargs
...
INSERT INTO callback (... data ...)
SELECT ...
  json_build_object(
      '__var', json_build_object(
          'path', b.cb_path,
          'kwargs', b.cb_kwargs,      -- <-- raw nested dict, NOT extended-serialized
          'prefix', :prefix,
          'dag_id', b.dag_id),
      '__type', 'dict')::json

(_upgrade_mysql_sqlite() builds the same shape in Python.) The insert is done via Core (callback_table.insert() / sa.text), which also bypasses ExtendedJSON.process_bind_param, so BaseSerialization.serialize never runs.

Result: callback.data = {"__type":"dict","__var":{"path":..., "kwargs": {<bare nested object>}, ...}}. On read, deserialize recurses into the outer DAT.DICT, reaches the bare kwargs (or a nested dict inside it such as metric tags), executes var = encoded_var[Encoding.VAR], and throws KeyError('__var').

Trigger: callback kwargs containing a nested dict. Callbacks with only flat/primitive kwargs survive because the single top-level wrap is enough.

What you think should happen instead

The migration should produce the same encoding BaseSerialization.serialize would — i.e. recursively wrap nested dicts (kwargs, and any dict within it) as {"__type":"dict","__var":{...}} — so the resulting callback.data round-trips through BaseSerialization.deserialize.

Because 0094 has already shipped (3.2.0) and run on live deployments, a forward repair migration is also needed to re-encode already-corrupted callback.data rows (deployments that ran 0094 with any nested-dict callback kwargs currently have a latent scheduler crash that fires when such a deadline becomes due).

How to reproduce

Minimal, self-contained demonstration of the serialize/deserialize invariant (no Airflow import needed): a correct BaseSerialization.serialize({"path": "...", "kwargs": {"tags": {"a": "b"}}}) wraps kwargs/tags and round-trips, whereas the migration's shallow-wrapped shape {"__type":"dict","__var":{"kwargs":{"tags":{...}}}} raises KeyError('__var') in deserialize.

End-to-end: on 3.1.x create a deadline whose AsyncCallback kwargs contain a nested dict; upgrade across 0094; then run the scheduler (or SELECT the callback.data) — it is stored with unwrapped nested kwargs and the scheduler crashes.

Operating System

Linux (containerized)

Versions of Apache Airflow Providers

N/A (core deadline alerts + serialization)

Deployment

Other

Anything else?

Workaround to unblock a crashing scheduler (neutralize the corrupt rows so the WHERE ~missed deadline query skips them):

UPDATE deadline d SET missed = true
FROM callback c
WHERE c.id = d.callback_id
  AND d.missed = false
  AND jsonb_path_exists(c.data, '$.**.__var.* ? (@.type() == "object" && !exists(@.__type))');

Are you willing to submit PR?

Yes.

---

🤖 Filed with Claude Code (model: Claude Opus 4.8, claude-opus-4-8[1m]).

[dstandish]

@ferruzzi

[dstandish]

Here's a mitigation that can be used after the fact:

BEGIN;

CREATE OR REPLACE FUNCTION pg_temp.repair_node(node jsonb) RETURNS jsonb AS $$
DECLARE k text; v jsonb; out jsonb := '{}'::jsonb;
BEGIN
  IF jsonb_typeof(node) = 'object' THEN
    IF node ? '__type' AND node ? '__var' THEN
      IF node->>'__type' = 'dict' AND jsonb_typeof(node->'__var') = 'object' THEN
        FOR k, v IN SELECT * FROM jsonb_each(node->'__var') LOOP
          out := out || jsonb_build_object(k, pg_temp.repair_node(v));
        END LOOP;
        RETURN jsonb_build_object('__type','dict','__var',out);
      END IF;
      RETURN node;
    ELSE
      FOR k, v IN SELECT * FROM jsonb_each(node) LOOP
        out := out || jsonb_build_object(k, pg_temp.repair_node(v));
      END LOOP;
      RETURN jsonb_build_object('__type','dict','__var',out);
    END IF;
  ELSIF jsonb_typeof(node) = 'array' THEN
    RETURN (SELECT jsonb_agg(pg_temp.repair_node(e)) FROM jsonb_array_elements(node) e);
  END IF;
  RETURN node;
END;
$$ LANGUAGE plpgsql;

UPDATE callback c
SET data = pg_temp.repair_node(c.data::jsonb)
WHERE jsonb_path_exists(c.data::jsonb, '$.**.__var.* ? (@.type()=="object" && !exists(@.__type))');

-- verify: expect 0
SELECT count(*) FROM callback c
WHERE jsonb_path_exists(c.data::jsonb, '$.**.__var.* ? (@.type()=="object" && !exists(@.__type))');

COMMIT;

[seanmuth]

PR #69985 opened.

Per discussion with @dstandish, the fix hotfixes migration 0094 in place so it correctly extended-serializes nested callback kwargs on future upgrades (both the Postgres CTE and the MySQL/SQLite paths, with the downgrade paths decoding back to raw to preserve the round-trip).

Repairing already-corrupted callback.data rows (deployments that have already run 0094 with nested-dict callback kwargs) is intentionally left to end-users / managed services rather than shipping a data-repair migration. Affected rows can be found with:

SELECT count(*) FROM callback c
WHERE jsonb_path_exists(c.data::jsonb, '$.**.__var.* ? (@.type()=="object" && !exists(@.__type))');

and re-encoded so nested dicts are wrapped as {"__type": "dict", "__var": {...}}.

---

🤖 Posted with Claude Code (model: Claude Opus 4.8, claude-opus-4-8[1m]).

[seanmuth]

Here's a cleaner db-agnostic repair script to run in python, claude generated and not tested by me, just FYI, but applies the same fix over the data that the (fixed) migration would do:

import json
from sqlalchemy import create_engine, text

def repair(node):
    if isinstance(node, dict):
        if "__type" in node and "__var" in node:
            if node["__type"] == "dict" and isinstance(node["__var"], dict):
                return {"__type": "dict", "__var": {k: repair(v) for k, v in node["__var"].items()}}
            return node
        return {"__type": "dict", "__var": {k: repair(v) for k, v in node.items()}}
    if isinstance(node, list):
        return [repair(v) for v in node]
    return node

eng = create_engine("mysql://…")   # or postgresql://…
with eng.begin() as c:
    rows = c.execute(text("SELECT id, data FROM callback")).mappings().all()
    for r in rows:
        data = r["data"] if isinstance(r["data"], (dict, list)) else json.loads(r["data"])
        fixed = repair(data)
        if fixed != data:
            c.execute(text("UPDATE callback SET data = :d WHERE id = :i"),
                      {"d": json.dumps(fixed), "i": r["id"]})