Function remoteTerminateJobManager

pkg/jobcontroller/jobcontroller.go:989–1050

(ctx context.Context, job *waveobj.Job)

}

func remoteTerminateJobManager(ctx context.Context, job *waveobj.Job) error {
	log.Printf("[job:%s] terminating job manager", job.OID)

	shouldWrite := jobTerminationMessageWritten.TestAndSet(job.OID, true, func(val bool, exists bool) bool {
		return !exists || !val
	})
	if shouldWrite {
		resetTerminalState(ctx, job.AttachedBlockId)
		writeMutedMessageToTerminal(job.AttachedBlockId, "[shell terminated]")
	}

	if job.JobManagerStatus == JobManagerStatus_Done {
		log.Printf("[job:%s] job manager already marked as done, skipping termination", job.OID)
		return nil
	}

	bareRpc := wshclient.GetBareRpcClient()
	terminateData := wshrpc.CommandRemoteTerminateJobManagerData{
		JobId:             job.OID,
		JobManagerPid:     job.JobManagerPid,
		JobManagerStartTs: job.JobManagerStartTs,
	}

	rpcOpts := &wshrpc.RpcOpts{
		Route:   wshutil.MakeConnectionRouteId(job.Connection),
		Timeout: 5000,
	}

	err := wshclient.RemoteTerminateJobManagerCommand(bareRpc, terminateData, rpcOpts)
	if err != nil {
		log.Printf("[job:%s] error terminating job manager: %v", job.OID, err)
		return fmt.Errorf("failed to terminate job manager: %w", err)
	}

	var updatedJob *waveobj.Job
	updateErr := wstore.DBUpdateFn(ctx, job.OID, func(job *waveobj.Job) {
		job.JobManagerStatus = JobManagerStatus_Done
		job.JobManagerDoneReason = JobDoneReason_Terminated
		job.TerminateOnReconnect = false
		if !job.StreamDone {
			job.StreamDone = true
			job.StreamError = "job manager terminated"
		}
		updatedJob = job
	})
	if updateErr != nil {
		log.Printf("[job:%s] error updating job status after termination: %v", job.OID, updateErr)
	} else {
		sendBlockJobStatusEventByJob(ctx, updatedJob)
	}

	telemetry.GoRecordTEventWrap(&telemetrydata.TEvent{
		Event: "job:done",
		Props: telemetrydata.TEventProps{
			JobDoneReason: JobDoneReason_Terminated,
			JobKind:       job.JobKind,
		},
	})