Refactor job run monitoring to use struct with member methods (#3774)

## Changes

Replaced callback builder functions that returned closures capturing
stack pointers with a `jobRunMonitor` struct and a single `onProgress`
method.

The original code had three separate callback builders
(`pullRunIdCallback`, `logDebugCallback`, `logProgressCallback`) that
each returned closures with their own state, leading to pointer
manipulation on the stack and redundant state tracking.

Benefits:
- Single source of truth for run state (`prevState` tracked in one
place)
- Eliminated stack pointer manipulation
- Removed redundant logging (lifecycle state was logged twice)

## Why

I was looking into `libs/cmdio` and how we can refactor it. I suspect we
need to remove or refactor the progress logger to make further
abstraction possible (eg. turn it into a "widget" that is started and
stopped, like a spinner). Refactoring it is not possible without first
changing the only call site, which is this one.

## Tests

Tests pass. Manually performed a job run to

Author: pietern

bundle/run/job.go

@@ -86,66 +86,47 @@ func (r *jobRunner) logFailedTasks(ctx context.Context, runId int64) {
 	}
 }
 
-func pullRunIdCallback(runId *int64) func(info *jobs.Run) {
-	return func(i *jobs.Run) {
-		if *runId == 0 {
-			*runId = i.RunId
-		}
-	}
+// jobRunMonitor tracks state for a single job run and provides callbacks
+// for monitoring progress.
+type jobRunMonitor struct {
+	ctx            context.Context
+	prevState      *jobs.RunState
+	progressLogger *cmdio.Logger
 }
 
-func logDebugCallback(ctx context.Context, runId *int64) func(info *jobs.Run) {
-	var prevState *jobs.RunState
-	return func(i *jobs.Run) {
-		state := i.State
-		if state == nil {
-			return
-		}
-
-		// Log the job run URL as soon as it is available.
-		if prevState == nil {
-			log.Infof(ctx, "Run available at %s", i.RunPageUrl)
-		}
-		if prevState == nil || prevState.LifeCycleState != state.LifeCycleState {
-			log.Infof(ctx, "Run status: %s", i.State.LifeCycleState)
-			prevState = state
-		}
+// onProgress is the single callback that handles all state tracking and logging.
+func (m *jobRunMonitor) onProgress(info *jobs.Run) {
+	state := info.State
+	if state == nil {
+		return
 	}
-}
-
-func logProgressCallback(ctx context.Context, progressLogger *cmdio.Logger) func(info *jobs.Run) {
-	var prevState *jobs.RunState
-	return func(i *jobs.Run) {
-		state := i.State
-		if state == nil {
-			return
-		}
-
-		if prevState == nil {
-			progressLogger.Log(progress.NewJobRunUrlEvent(i.RunPageUrl))
-		}
 
-		if prevState != nil && prevState.LifeCycleState == state.LifeCycleState &&
-			prevState.ResultState == state.ResultState {
-			return
-		} else {
-			prevState = state
-		}
+	// First time we see this run.
+	if m.prevState == nil {
+		log.Infof(m.ctx, "Run available at %s", info.RunPageUrl)
+		m.progressLogger.Log(progress.NewJobRunUrlEvent(info.RunPageUrl))
+	}
 
-		event := &progress.JobProgressEvent{
-			Timestamp: time.Now(),
-			JobId:     i.JobId,
-			RunId:     i.RunId,
-			RunName:   i.RunName,
-			State:     *i.State,
-		}
+	// No state change: do not log.
+	if m.prevState != nil &&
+		m.prevState.LifeCycleState == state.LifeCycleState &&
+		m.prevState.ResultState == state.ResultState {
+		return
+	}
 
-		// log progress events to stderr
-		progressLogger.Log(event)
+	// Capture current state as previous state for next call.
+	m.prevState = state
 
-		// log progress events in using the default logger
-		log.Info(ctx, event.String())
+	// Log progress event both to the terminal (in place or append), and to the logger.
+	event := &progress.JobProgressEvent{
+		Timestamp: time.Now(),
+		JobId:     info.JobId,
+		RunId:     info.RunId,
+		RunName:   info.RunName,
+		State:     *info.State,
 	}
+	m.progressLogger.Log(event)
+	log.Info(m.ctx, event.String())
 }
 
 func (r *jobRunner) Run(ctx context.Context, opts *Options) (output.RunOutput, error) {
@@ -154,8 +135,6 @@ func (r *jobRunner) Run(ctx context.Context, opts *Options) (output.RunOutput, e
 		return nil, fmt.Errorf("job ID is not an integer: %s", r.job.ID)
 	}
 
-	runId := new(int64)
-
 	err = r.convertPythonParams(opts)
 	if err != nil {
 		return nil, err
@@ -172,19 +151,16 @@ func (r *jobRunner) Run(ctx context.Context, opts *Options) (output.RunOutput, e
 
 	w := r.bundle.WorkspaceClient()
 
-	// gets the run id from inside Jobs.RunNowAndWait
-	pullRunId := pullRunIdCallback(runId)
-
-	// callback to log status updates to the universal log destination.
-	// Called on every poll request
-	logDebug := logDebugCallback(ctx, runId)
-
 	// callback to log progress events. Called on every poll request
 	progressLogger, ok := cmdio.FromContext(ctx)
 	if !ok {
 		return nil, errors.New("no progress logger found")
 	}
-	logProgress := logProgressCallback(ctx, progressLogger)
+
+	monitor := &jobRunMonitor{
+		ctx:            ctx,
+		progressLogger: progressLogger,
+	}
 
 	waiter, err := w.Jobs.RunNow(ctx, *req)
 	if err != nil {
@@ -199,13 +175,9 @@ func (r *jobRunner) Run(ctx context.Context, opts *Options) (output.RunOutput, e
 		return nil, err
 	}
 
-	run, err := waiter.OnProgress(func(r *jobs.Run) {
-		pullRunId(r)
-		logDebug(r)
-		logProgress(r)
-	}).GetWithTimeout(jobRunTimeout)
+	run, err := waiter.OnProgress(monitor.onProgress).GetWithTimeout(jobRunTimeout)
 	if err != nil {
-		r.logFailedTasks(ctx, *runId)
+		r.logFailedTasks(ctx, waiter.RunId)
 	}
 	if err != nil {
 		return nil, err
@@ -229,7 +201,7 @@ func (r *jobRunner) Run(ctx context.Context, opts *Options) (output.RunOutput, e
 	// The task completed successfully.
 	case jobs.RunResultStateSuccess:
 		log.Infof(ctx, "Run has completed successfully!")
-		return output.GetJobOutput(ctx, r.bundle.WorkspaceClient(), *runId)
+		return output.GetJobOutput(ctx, r.bundle.WorkspaceClient(), waiter.RunId)
 
 	// The run was stopped after reaching the timeout.
 	case jobs.RunResultStateTimedout: