code: func (ctl *Controller) fanOutShardsHandler(c context.Context, payload proto.Message) error {
  	msg := payload.(*tasks.FanOutShards)
  
  	// Make sure the job is still present. If it is aborted, we still need to
  	// launch the shards, so they notice they are being aborted. We could try
  	// to abort all shards right here and now, but it basically means implementing
  	// an alternative shard abort flow. Seems simpler just to let the regular flow
  	// to proceed.
  	job, err := getJobInState(c, JobID(msg.JobId), State_RUNNING, State_ABORTING)
  	if err != nil || job == nil {
  		return errors.Annotate(err, "in FanOutShards").Err()
  	}
  
  	// Grab the list of shards created in SplitAndLaunch. It must exist at this
  	// point, since the job is in Running state.
  	shardIDs, err := job.fetchShardIDs(c)
  	if err != nil {
  		return errors.Annotate(err, "in FanOutShards").Err()
  	}
  
  	// Enqueue a bunch of named ProcessShard tasks (one per shard) to actually
  	// launch shard processing. This is idempotent operation, so if FanOutShards
  	// crashes midway and later retried, nothing bad happens.
  	tsks := make([]*tq.Task, len(shardIDs))
  	for idx, sid := range shardIDs {
  		tsks[idx] = makeProcessShardTask(job.ID, sid, 0, true)
  	}
  	return ctl.tq().AddTask(c, tsks...)
  }
docs: // fanOutShardsHandler fetches a list of shards from the job and launches
  // named ProcessShard tasks, one per shard.