Add async asset store accessors for async tasks and watcher triggers

[jason810496]

• related: Add async task store accessors for async tasks #68232
• related: Add async task store accessors (aget/aset) for use in async tasks #68214

Why

While reviewing #71135 and #71387, I realized that we should support the async methods for AssetStateStoreAccessor following the pattern of TaskStateStoreAccessor introduced in #68232.

How

• Split each sync method into a _build_*_message helper plus a shared _extract_get_response, then have the sync method send() and the new async method await asend() the same message. No behaviour change on the sync path.
• adelete/aclear await BaseStoreBackend.adelete/aclear, which already exist on the backend base class, and keep the DB-ref-first ordering of their sync counterparts.
• Mirror the four methods on the AssetStateStoreAccessors facade. That is the object TriggerRunner injects as trigger_instance.asset_state_store, so it is what watcher triggers actually call.

Known gaps (inherited from #68232)

• BaseStoreBackend has no async serialize/deserialize variant, so aget/aset still call the sync ref (de)serialization on the event-loop thread. With a real-I/O worker backend that can stall the loop despite the async API. Tracked in Async state store accessors block the event loop on sync ref (de)serialization #72129.
• adelete/aclear delete the DB ref first, then await backend.adelete/aclear, which providers/common/io raised NotImplementedError. Follow-up in Implement async state store methods for object storage backend #72131.

---

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

• Yes, with help of Claude Code Opus 5 following the guidelines

[jroachgolf84]

@jason810496 - this is great, this is something I had on my radar when I originally ported in the Accessor for use with BaseEventTrigger's. Does this change the methods that a Trigger author would use to interact with the state store? i.e. aget rather than get?

[jroachgolf84]

This was my original PR: #68900

[jroachgolf84]

@jason810496, I think we'd need to allow for the async methods to be used from the Triggerer? Similar to what I did in my PR above?

cc: @amoghrajesh

[jason810496]

Does this change the methods that a Trigger author would use to interact with the state store? i.e. aget rather than get?

Yes, but they can be done in the follow-up, since there're only two triggers using the Asset Store so far.
But they're kind of using in incorrectly (e.g. calling directly with the sync method within the async trigger.run method or using asyncio.to_thread.

[jason810496]

Just head up that we need to get this one in recently as the downstream Iceberg provider change just merged.

Additionally, I just realized we need a compatible layer for the aget method on provider side, see #72173 (review) for more context.

[amoghrajesh]

LGTM some nits mostly.

[jason810496]

Thanks for the review.

[github-actions]

Backport failed to create: v3-3-test. View the failure log Run details

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
❌	v3-3-test

You can attempt to backport this manually by running:

cherry_picker 2aa581e v3-3-test

This should apply the commit to the v3-3-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.

After you have resolved the conflicts, you can continue the backport process by running:

cherry_picker --continue

If you don't have cherry-picker installed, see the installation guide.

[jroachgolf84]

I'm going to follow up with the PR to allow these to be used on the Triggerer.