Unbreak stopping unloadable schedules/sensors (#7446)

Summary:
The new stop mutations assume that the ExternalSensor can be loaded, which is not the case for unloadable schedules and sensors. Getting this right requires threading the selector ID all the way through into the mutation path (at least in the unloadable path, but might as well do it everywhere for consistency)

Will add a test at the graphql python level as well.

integration_tests/test_suites/celery-k8s-integration-test-suite/test_daemon_scheduler.py

@@ -50,7 +50,9 @@ def test_execute_schedule_on_celery_k8s(  # pylint: disable=redefined-outer-name
 
         finally:
             dagster_instance_for_daemon.stop_schedule(
-                reoriginated_schedule.get_external_origin_id(), reoriginated_schedule
+                reoriginated_schedule.get_external_origin_id(),
+                reoriginated_schedule.selector_id,
+                reoriginated_schedule,
             )
 
         last_run = schedule_runs[0]

js_modules/dagit/packages/core/src/instigation/InstigationUtils.tsx

@@ -42,6 +42,7 @@ export const RUN_STATUS_FRAGMENT = gql`
 export const INSTIGATION_STATE_FRAGMENT = gql`
   fragment InstigationStateFragment on InstigationState {
     id
+    selectorId
     name
     instigationType
     status

js_modules/dagit/packages/core/src/instigation/Unloadable.tsx

@@ -133,7 +133,7 @@ const UnloadableScheduleInfo = () => (
 );
 
 const SensorStateRow = ({sensorState}: {sensorState: InstigationStateFragment}) => {
-  const {id, name, status, repositoryOrigin, ticks} = sensorState;
+  const {id, selectorId, name, status, repositoryOrigin, ticks} = sensorState;
 
   const [stopSensor, {loading: toggleOffInFlight}] = useMutation<StopSensor>(STOP_SENSOR_MUTATION, {
     onCompleted: displaySensorMutationErrors,
@@ -150,7 +150,7 @@ const SensorStateRow = ({sensorState}: {sensorState: InstigationStateFragment})
           'If you turn it off, you will not be able to turn it back on from ' +
           'the currently loaded workspace.',
       });
-      stopSensor({variables: {jobOriginId: id}});
+      stopSensor({variables: {jobOriginId: id, jobSelectorId: selectorId}});
     }
   };
 
@@ -206,7 +206,7 @@ const ScheduleStateRow: React.FC<{
   );
   const [showRepositoryOrigin, setShowRepositoryOrigin] = React.useState(false);
   const confirm = useConfirmation();
-  const {id, name, ticks, status, repositoryOrigin, typeSpecificData} = scheduleState;
+  const {id, selectorId, name, ticks, status, repositoryOrigin, typeSpecificData} = scheduleState;
   const latestTick = ticks.length > 0 ? ticks[0] : null;
   const cronSchedule =
     typeSpecificData && typeSpecificData.__typename === 'ScheduleData'
@@ -221,7 +221,7 @@ const ScheduleStateRow: React.FC<{
           'If you turn it off, you will not be able to turn it back on from ' +
           'the currently loaded workspace.',
       });
-      stopSchedule({variables: {scheduleOriginId: id}});
+      stopSchedule({variables: {scheduleOriginId: id, scheduleSelectorId: selectorId}});
     }
   };
 

js_modules/dagit/packages/core/src/schedules/ScheduleMutations.tsx

@@ -27,8 +27,11 @@ export const START_SCHEDULE_MUTATION = gql`
 `;
 
 export const STOP_SCHEDULE_MUTATION = gql`
-  mutation StopSchedule($scheduleOriginId: String!) {
-    stopRunningSchedule(scheduleOriginId: $scheduleOriginId) {
+  mutation StopSchedule($scheduleOriginId: String!, $scheduleSelectorId: String!) {
+    stopRunningSchedule(
+      scheduleOriginId: $scheduleOriginId
+      scheduleSelectorId: $scheduleSelectorId
+    ) {
       __typename
       ... on ScheduleStateResult {
         scheduleState {

js_modules/dagit/packages/core/src/schedules/ScheduleSwitch.tsx

@@ -25,7 +25,7 @@ interface Props {
 export const ScheduleSwitch: React.FC<Props> = (props) => {
   const {repoAddress, schedule, size = 'large'} = props;
   const {name, scheduleState} = schedule;
-  const {status, id} = scheduleState;
+  const {status, id, selectorId} = scheduleState;
 
   const {canStartSchedule, canStopRunningSchedule} = usePermissions();
 
@@ -50,7 +50,7 @@ export const ScheduleSwitch: React.FC<Props> = (props) => {
   const onStatusChange = () => {
     if (status === InstigationStatus.RUNNING) {
       stopSchedule({
-        variables: {scheduleOriginId: id},
+        variables: {scheduleOriginId: id, scheduleSelectorId: selectorId},
       });
     } else {
       startSchedule({
@@ -102,6 +102,7 @@ export const SCHEDULE_SWITCH_FRAGMENT = gql`
     cronSchedule
     scheduleState {
       id
+      selectorId
       status
     }
   }

js_modules/dagit/packages/core/src/sensors/SensorMutations.tsx

@@ -26,8 +26,8 @@ export const START_SENSOR_MUTATION = gql`
 `;
 
 export const STOP_SENSOR_MUTATION = gql`
-  mutation StopSensor($jobOriginId: String!) {
-    stopSensor(jobOriginId: $jobOriginId) {
+  mutation StopSensor($jobOriginId: String!, $jobSelectorId: String!) {
+    stopSensor(jobOriginId: $jobOriginId, jobSelectorId: $jobSelectorId) {
       __typename
       ... on StopSensorMutationResult {
         instigationState {

js_modules/dagit/packages/core/src/sensors/SensorSwitch.tsx

@@ -27,7 +27,7 @@ export const SensorSwitch: React.FC<Props> = (props) => {
   const {canStartSensor, canStopSensor} = usePermissions();
 
   const {jobOriginId, name, sensorState} = sensor;
-  const {status} = sensorState;
+  const {status, selectorId} = sensorState;
   const sensorSelector = {
     ...repoAddressToSelector(repoAddress),
     sensorName: name,
@@ -43,7 +43,7 @@ export const SensorSwitch: React.FC<Props> = (props) => {
 
   const onChangeSwitch = () => {
     if (status === InstigationStatus.RUNNING) {
-      stopSensor({variables: {jobOriginId}});
+      stopSensor({variables: {jobOriginId, jobSelectorId: selectorId}});
     } else {
       startSensor({variables: {sensorSelector}});
     }
@@ -92,6 +92,7 @@ export const SENSOR_SWITCH_FRAGMENT = gql`
     name
     sensorState {
       id
+      selectorId
       status
     }
   }