run_status_sensor target/request_job not working as expected

What's the issue?

I need a sensor to monitor job_a, which runs on a schedule. When job_a succeeds, I need the sensor to update the dynamic partitions used for job_b and then submit run requests for all partitions (i.e. not just new ones). The code that I would expect to accomplish this looks roughly like:

from dagster import (
    ...
)

from project.jobs import job_a, job_b

partitions_def = DynamicPartitionsDefinition(name="records")

@run_status_sensor(
    run_status=DagsterRunStatus.SUCCESS,
    default_status=DefaultSensorStatus.RUNNING,
    monitored_jobs=[job_a],
    request_job=job_b,
)
def records_success_callback(context: RunStatusSensorContext) -> None:
    """
    Sensor triggered when job_a succeeds
    """
    all_partitions, partitions_to_delete, partitions_to_add = util_that_gets_partitions(context)
    return SensorResult(
        run_requests=[
            RunRequest(partition_key=partition_key) for partition_key in all_partitions
        ],
        dynamic_partitions_requests=[
            page_partitions_def.build_add_request(partition_keys_to_add),
            page_partitions_def.build_delete_request(partition_keys_to_delete),
        ],
    )

This code results in an error: dagster._core.errors.DagsterCodeLocationLoadError: Failure loading src.project: dagster._core.errors.DagsterInvalidDefinitionError: Duplicate definition found for unresolved job 'job_b'

It appears that importing job_b and passing it as the request_job value results in Dagster trying to define the job twice.

I then tried to instead define the target job in the RunRequest itself, rather than in the @run_status_sensor decorator. This looked like:

from dagster import (
    ...
)

from project.jobs import job_a

partitions_def = DynamicPartitionsDefinition(name="records")

@run_status_sensor(
    run_status=DagsterRunStatus.SUCCESS,
    default_status=DefaultSensorStatus.RUNNING,
    monitored_jobs=[job_a],
)
def records_success_callback(context: RunStatusSensorContext) -> None:
    """
    Sensor triggered when job_a succeeds
    """
    all_partitions, partitions_to_delete, partitions_to_add = util_that_gets_partitions(context)
    return SensorResult(
        run_requests=[
            RunRequest(partition_key=partition_key, job_name="job_b") for partition_key in all_partitions
        ],
        dynamic_partitions_requests=[
            page_partitions_def.build_add_request(partition_keys_to_add),
            page_partitions_def.build_delete_request(partition_keys_to_delete),
        ],
    )

But that gave me this error: Error in sensor records_success_callback: Sensor evaluation function returned a RunRequest for a sensor lacking a specified target (job_name, job, or jobs). Targets can be specified by providing job, jobs, or job_name to the @sensor decorator.

So then, I tried defining job or job_name in the @run_status_sensor and got an unexpected key error. After checking the docs, I figured the error message was more specifically for the @sensor decorator and that I should try returning to request_job instead.

As a gut check, I pulled the definition of job_b into the sensor.py file like so:

from dagster import (
    ...
)

from project.jobs import job_a

partitions_def = DynamicPartitionsDefinition(name="records")

job_b_assets = AssetSelection.assets("asset_a", "asset_b", "asset_c")
job_b = define_asset_job(name="job_b", selection=job_b_assets)

@run_status_sensor(
    run_status=DagsterRunStatus.SUCCESS,
    default_status=DefaultSensorStatus.RUNNING,
    monitored_jobs=[job_a],
    request_job=job_b,
)
def records_success_callback(context: RunStatusSensorContext) -> None:
    """
    Sensor triggered when job_a succeeds
    """
    all_partitions, partitions_to_delete, partitions_to_add = util_that_gets_partitions(context)
    return SensorResult(
        run_requests=[
            RunRequest(partition_key=partition_key) for partition_key in all_partitions
        ],
        dynamic_partitions_requests=[
            page_partitions_def.build_add_request(partition_keys_to_add),
            page_partitions_def.build_delete_request(partition_keys_to_delete),
        ],
    )

and (very much to my surprise) I no longer had the Duplicate definition error. However, this time, the sensor started getting activated every 30 seconds — as in, every 30 seconds, without job_a running, run requests for all partitions on job_b were getting submitted.

I went to bed, woke up this morning, and thought, "what the heck, let's try using request_jobs=[job_b] in the sensor decorator, instead of request_job=job_b". And somehow, that seems to be working. 🎉

I'm still not entirely sure what the heart of this issue is (I'm not even really sure that I'm not just doing something wrong), but I think the problems are:

Defining a job in a separate file shouldn't cause a Duplicate definition error if used as a sensor request_job or request_jobs value
Validation should allow a RunRequest returned by a sensor without a target as long as the target is defined in the RunRequest itself
The request_job should not run from a status sensor if the job defined as the monitored_job hasn't succeeded

If you've read this far, tysm for your time and consideration 🙏

What did you expect to happen?

No response

How to reproduce?

No response

Dagster version

1.9.0

Deployment type

None

Deployment details

No response

Additional information

No response

Message from the maintainers

Impacted by this issue? Give it a 👍! We factor engagement into prioritization. By submitting this issue, you agree to follow Dagster's Code of Conduct.

dagster-io / dagster