AIP-93 has shifted directions a bit; rather than creating a state-store for “Asset watching”, we’re planning on creating a general, async-aware state-store that can be accessed from “anywhere” in Airflow (Scheduler, Triggerer, Workers, etc.)

ProcessState  Model

The ProcessState  model should be relatively straightforward. This model will be accessible via some sort of API/Task SDK. It will almost act as a sort of key-value store. One of the things that’s important to me is making sure that there is uniqueness outside of just specifying a “key”. i.e. some sort of “hash” or "process_name". This model will very closely mimic the Variable  model.

class ProcessState(Base):
    id = ...
    
	# Automatically created
    created_date = ...
    update_date = ...
	
	process_name = ...
    key = ...
    value = ...
	
	# Implement set, update, delete, get methods to store and retrieve state
    def set(): ...
	def update(): ...
	def delete(): ...
    def get(): ...

API/Task SDK

Again, mimicking the Variable  model, the ProcessState  model will be accessible via the API and the Task SDK. This will allow for process state to be update each Airflow "component" (Workers, Triggerer, Scheduler/DAG Processor).