code: func (ctl *Controller) LaunchJob(c context.Context, j *JobConfig) (JobID, error) {
  	disp := ctl.tq()
  
  	if err := j.Validate(); err != nil {
  		return 0, errors.Annotate(err, "bad job config").Err()
  	}
  	if _, err := ctl.getFactory(j.Mapper); err != nil {
  		return 0, errors.Annotate(err, "bad job config").Err()
  	}
  
  	// Prepare and store the job entity, generate its key. Launch a tq task that
  	// subdivides the key space and launches individual shards. We do it
  	// asynchronously since this can be potentially slow (for large number of
  	// shards).
  	var job Job
  	err := runTxn(c, func(c context.Context) error {
  		now := clock.Now(c).UTC()
  		job = Job{
  			Config:  *j,
  			State:   State_STARTING,
  			Created: now,
  			Updated: now,
  		}
  		if err := datastore.Put(c, &job); err != nil {
  			return errors.Annotate(err, "failed to store Job entity").Tag(transient.Tag).Err()
  		}
  		return disp.AddTask(c, &tq.Task{
  			Title: fmt.Sprintf("split:job-%d", job.ID),
  			Payload: &tasks.SplitAndLaunch{
  				JobId: int64(job.ID),
  			},
  		})
  	})
  	if err != nil {
  		return 0, err
  	}
  	return job.ID, nil
  }
docs: // LaunchJob launches a new mapping job, returning its ID (that can be used to
  // control it or query its status).
  //
  // Launches a datastore transaction inside.