code: func (e *engineImpl) handlePubSubMessage(c context.Context, msg *pubsub.PubsubMessage) error {
  	logging.Infof(c, "Received PubSub message %q", msg.MessageId)
  
  	// Extract Job and Invocation ID from validated auth_token.
  	var jobID string
  	var invID int64
  	data, err := pubsubAuthToken.Validate(c, msg.Attributes["auth_token"], nil)
  	if err != nil {
  		logging.WithError(err).Errorf(c, "Bad auth_token attribute")
  		return err
  	}
  	jobID = data["job"]
  	if invID, err = strconv.ParseInt(data["inv"], 10, 64); err != nil {
  		logging.WithError(err).Errorf(c, "Could not parse 'inv' %q", data["inv"])
  		return err
  	}
  
  	// Hand the message to the controller.
  	action := fmt.Sprintf("pubsub message %q", msg.MessageId)
  	return e.withController(c, jobID, invID, action, func(c context.Context, ctl *taskController) error {
  		err := ctl.manager.HandleNotification(c, ctl, msg)
  		switch {
  		case err == nil:
  			return nil // success! save the invocation
  		case transient.Tag.In(err) || tq.Retry.In(err):
  			return err // ask for redelivery on transient errors, don't touch the invocation
  		}
  		// On fatal errors, move the invocation to failed state (if not already).
  		if ctl.State().Status != task.StatusFailed {
  			ctl.DebugLog("Fatal error when handling PubSub notification, aborting invocation - %s", err)
  			ctl.State().Status = task.StatusFailed
  		}
  		return nil // need to save the invocation, even on fatal errors
  	})
  }
docs: // handlePubSubMessage routes the pubsub message to the invocation.