Hello - we're noticing failures of the Auto-Materialize Daemon in our dagster cloud instance; it's been about 5 hours since the last run initiated by the daemon - the error message is below:

dagster._core.errors.DagsterInvariantViolationError: Can only calculate data times for records with a materialization event and an asset_key.
  File "/dagster/dagster/_daemon/daemon.py", line 225, in core_loop
    yield from self.run_iteration(workspace_process_context)
  File "/dagster/dagster/_daemon/asset_daemon.py", line 100, in run_iteration
    run_requests, new_cursor, evaluations = reconcile(
  File "/dagster/dagster/_core/definitions/asset_reconciliation_sensor.py", line 921, in reconcile
    determine_asset_partitions_to_auto_materialize_for_freshness(
  File "/dagster/dagster/_core/definitions/freshness_based_auto_materialize.py", line 171, in determine_asset_partitions_to_auto_materialize_for_freshness
    in_progress_data_time = data_time_resolver.get_in_progress_data_time(
  File "/dagster/dagster/_core/definitions/data_time.py", line 398, in get_in_progress_data_time
    run_data_time = self._get_in_progress_data_time_in_run(
  File "/dagster/dagster/_utils/cached_method.py", line 66, in _cached_method_wrapper
    result = method(self, *args, **kwargs)
  File "/dagster/dagster/_core/definitions/data_time.py", line 377, in _get_in_progress_data_time_in_run
    parent_data_time = self._get_in_progress_data_time_in_run(
  File "/dagster/dagster/_utils/cached_method.py", line 66, in _cached_method_wrapper
    result = method(self, *args, **kwargs)
  File "/dagster/dagster/_core/definitions/data_time.py", line 368, in _get_in_progress_data_time_in_run
    return self.get_current_data_time(asset_key, current_time=current_time)
  File "/dagster/dagster/_core/definitions/data_time.py", line 492, in get_current_data_time
    data_times = set(self.get_data_time_by_key_for_record(latest_record, current_time).values())
  File "/dagster/dagster/_core/definitions/data_time.py", line 470, in get_data_time_by_key_for_record
    raise DagsterInvariantViolationError(

This is a bug, I’m going to try to reproduce it. My first theory is that it has to do with wiping assets, did you wipe any before this occurred

nope - there were not any recent wipings. There have been wipings before, however.

Hi @Leo Qin! We located the root cause of this error, and it will be fixed in next week's release. At a high level, this was a bug that could occur in the following situation: • Asset X has one source asset parent, and one non-source asset parent • There is an asset with a freshness policy downstream of Asset X • Asset X is part of an in-progress run (or maybe the most recent run that it was part of failed) This would help explain why the daemon eventually recovered (as the in progress run completed, or a new run materialized asset X successfully). Does that seem plausible for your scenario?

yes - in our case, the last daemon-initiated run was a failure, and the asset that failed in that run was eventually materialized manually shortly before the daemon recovered. So, that tracks... and the asset in question has one source asset input (observable source asset) and two non-source asset inputs