Function ReconnectJobsForConn

pkg/jobcontroller/jobcontroller.go:1171–1202

(ctx context.Context, connName string)

}

func ReconnectJobsForConn(ctx context.Context, connName string) error {
	isConnected, err := conncontroller.IsConnected(connName)
	if err != nil {
		return fmt.Errorf("error checking connection status: %w", err)
	}
	if !isConnected {
		return fmt.Errorf("connection %q is not connected", connName)
	}

	allJobs, err := wstore.DBGetAllObjsByType[*waveobj.Job](ctx, waveobj.OType_Job)
	if err != nil {
		return fmt.Errorf("failed to get jobs: %w", err)
	}

	var jobsToReconnect []*waveobj.Job
	for _, job := range allJobs {
		if job.Connection == connName && isJobManagerRunning(job) {
			jobsToReconnect = append(jobsToReconnect, job)
		}
	}

	log.Printf("[conn:%s] found %d jobs to reconnect", connName, len(jobsToReconnect))

	for _, job := range jobsToReconnect {
		err = ReconnectJob(ctx, job.OID, nil)
		if err != nil {
			log.Printf("[job:%s] error reconnecting: %v", job.OID, err)
		}
	}

	return nil
}

func restartStreaming(ctx context.Context, jobId string, knownConnected bool, rtOpts *waveobj.RuntimeOpts) error {
	job, err := wstore.DBMustGet[*waveobj.Job](ctx, jobId)