Let a watcher trigger keep state when several assets watch it

[1fanwang]

Rationale for this change

Two teams watch the same orders feed. Their triggers are identical, so Airflow keeps one trigger row serving both, on purpose: two pollers would hammer the source. It gets one accessor per asset, but the watcher example in our docs assumes only one:

last_seen = self.asset_state_store.get("last_seen_id", default=0)

It raises ValueError: Task has 2 concrete inlets and outlets — use context['asset_state_store'][MY_ASSET] to specify which, and neither team's pipeline runs. A trigger cannot take that advice: it has no context, and never learns its assets, since it is rebuilt from serialize() kwargs that name none.

What changes are included in this PR?

len() and iteration on AssetStateStoreAccessors, so a trigger can ask how many assets it serves and reach each.

seen = [a.get("last_seen_id") for a in self.asset_state_store]
last_seen = min((v for v in seen if v is not None), default=0)  # oldest: a partial
...                            # write repeats a record rather than skipping one
for a in self.asset_state_store:
    a.set("last_seen_id", new_id)

Are these changes tested?

Live, on Postgres with api-server, dag-processor, scheduler and triggerer as separate processes. Two assets watch the docs' PollEventsTrigger unmodified, polling a file; both land on one trigger row.

Before and after

--- on main: the trigger cannot start ---
[error] Trigger ID 1 exited with error Task has 2 concrete inlets and outlets
[error] Trigger exited without sending an event. Dependent tasks will be failed.
$ grep -c "exited with error Task has 2 concrete" triggerer.log
29        # in 59s; with this change, 0

# both runs: echo record-1 > $AIRFLOW_HOME/feed.txt && sleep 30, then
$ psql -c "SELECT a.name, count(e.id) FROM asset a
           LEFT JOIN asset_event e ON e.asset_id=a.id GROUP BY a.name"
 orders_api    | 0 -> 1   # asset_state_store holds "record-1" for each
 orders_mirror | 0 -> 1
$ psql -c "SELECT dag_id, state FROM dag_run"
 (none) -> docs_team_a | running, docs_team_b | running

Six unit tests cover len and iteration; all fail on main.

Are there any user-facing changes?

One asset behaves as before; several can now keep state at all. get/set/delete/clear still raise for several.

#71354's aget/aset/adelete/aclear hit the same _single_accessor and raise here too; the two compose. First consumer: #71751.

[1fanwang]

cc @vincbeck related to #71387, thanks!

[vincbeck]

Hey @1fanwang, sorry but I have second thoughts on this one. Could you explain in human english the change here? Sorry but I find it difficult recently to read these AI generated description. Can you really explain what this change does? What does it allow? A before your PR vs after your PR would help

Second thought

[1fanwang]

Hey @1fanwang, sorry but I have second thoughts on this one. Could you explain in human english the change here? Sorry but I find it difficult recently to read these AI generated description. Can you really explain what this change does? What does it allow? A before your PR vs after your PR would help

hey @vincbeck my bad - we do NOT need this PR, and thanks for raising this!

I tried to walk through this again and repro, it proved me wrong and this only breaks in a case that doesn't really come up in practice: a single DAG waiting on two assets whose watchers are configured in the exact same way - declared in the same DAG file

feed_us = Asset("feed_us", watchers=[AssetWatcher(name="w", trigger=T(source="orders"))])
feed_eu = Asset("feed_eu", watchers=[AssetWatcher(name="w", trigger=T(source="orders"))])

with DAG("combined_report", schedule=[feed_us, feed_eu]):

Airflow dedupes triggers within a single file parse, so both assets land on one trigger row, and the trigger gets two accessors:

 7 | feed_eu      <- one trigger row, two assets
 7 | feed_us

The watcher example in our docs assumes one accessor, so it raises and the triggerer restarts it in a loop:

[error] Trigger ID 7 exited with error Task has 2 concrete inlets and outlets
[error] Trigger exited without sending an event. Dependent tasks will be failed.
$ grep -c "concrete inlets and outlets" triggerer.log
16

Split across two DAG files it doesn't happen at all. Closing this and #71751.