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.)

StateVariable  Model

The StateVariable  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 “owner". This model would something like this.

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


API/Task SDK

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

Other Considerations

Naming has been a challenge thus far. Some options have included State, StateVariable, ProcessState, Watermark, and more. For now, we're going with StateVariable.