Can asset sensors initiate a run for each materialization of the observed asset since the last sensor tick? After experimenting with them a bit, it seems like the asset sensor will only initiate a run for the latest materialization of the observed asset. I turned off my asset sensor, materialized the observed asset 3 times, then turned my asset sensor on again and the next tick only launched 1 job run when I was expecting 3 job runs. Is this expected?

hi @Tom Reilly this is expected behavior, but you can create your own custom sensor that implements things as you're expecting. This would look something like

@sensor
def my_sensor(context):
    new_materialization_records = context.instance.get_event_records(
                    EventRecordsFilter(
                        event_type=DagsterEventType.ASSET_MATERIALIZATION,
                        asset_key=SOME_KEY,
                        after_cursor=int(context.cursor) if context.cursor else 0,
                    ),
                    ascending=False,
                )
    # ... do something for each new materialization
    for record in new_materialization_records:
        ...
    # ... get the max record.storage_id from those new records
    ...
    # update the cursor so you know which records you've processed
    context.update_cursor(str(max_storage_id))