When materializing multiple sling assets, the assets materialize, but the steps fail

[nicklausroach]

What's the issue?

When I try to materialize multiple sling assets at once, the assets materialize successfully, but the steps fail on the following error:

dagster._check.functions.CheckError: Invariant failed.
  File "/Users/nicklausroach/Projects/internal/python_modules/product-operations/.venv/lib/python3.11/site-packages/dagster/_core/execution/plan/execute_plan.py", line 245, in dagster_event_sequence_for_step
    yield from check.generator(step_events)
  File "/Users/nicklausroach/Projects/internal/python_modules/product-operations/.venv/lib/python3.11/site-packages/dagster/_core/execution/plan/execute_step.py", line 501, in core_dagster_event_sequence_for_step
    for user_event in _step_output_error_checked_user_event_sequence(
  File "/Users/nicklausroach/Projects/internal/python_modules/product-operations/.venv/lib/python3.11/site-packages/dagster/_core/execution/plan/execute_step.py", line 184, in _step_output_error_checked_user_event_sequence
    for user_event in user_event_sequence:
  File "/Users/nicklausroach/Projects/internal/python_modules/product-operations/.venv/lib/python3.11/site-packages/dagster/_core/execution/plan/execute_step.py", line 88, in _process_asset_results_to_events
    for user_event in user_event_sequence:
  File "/Users/nicklausroach/Projects/internal/python_modules/product-operations/.venv/lib/python3.11/site-packages/dagster/_core/execution/plan/compute.py", line 200, in execute_core_compute
    step_context.job_def.asset_layer.node_output_handle_for_asset(asset_key).output_name
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/nicklausroach/Projects/internal/python_modules/product-operations/.venv/lib/python3.11/site-packages/dagster/_core/definitions/asset_layer.py", line 152, in node_output_handle_for_asset
    check.invariant(len(matching_handles) == 1)
  File "/Users/nicklausroach/Projects/internal/python_modules/product-operations/.venv/lib/python3.11/site-packages/dagster/_check/functions.py", line 1683, in invariant
    raise CheckError("Invariant failed.")

However, when I run these assets individually, they both run fine and the steps succeed.

I've found that by going into the source code of the replicate() function and iterating over the SlingEventIterator (or converting it to a list, forcing iteration) prior to returning the iterator itself...

event_iterator = SlingEventIterator(
            self._replicate(
                context=context,
                replication_config=replication_config_dict,
                dagster_sling_translator=dagster_sling_translator,
                debug=debug,
            ),
            sling_cli=self,
            replication_config=replication_config_dict,
            context=context,
        )
        list(event_iterator) # <-- force iteration of the Iterator
        return event_iterator

...I can run both assets at the same time and the steps succeed. Based on this I'm guessing the root of the problem lies in the SlingEventIterator and how it is handling the events, but I'm having trouble finding a true root cause.

What did you expect to happen?

Run multiple sling replication assets successfully

How to reproduce?

Using the latest versions of dagster, sling, and the dagster sling integration, create two sling assets that will run as separate steps, materialize them in the same run

Dagster version

1.9.9

Deployment type

Dagster Cloud

Deployment details

No response

Additional information

No response

Message from the maintainers

Impacted by this issue? Give it a 👍! We factor engagement into prioritization.