[errors] de-specialize Transient in favor of Tags.

scheduler/appengine/engine/engine.go

 // Copyright 2015 The LUCI Authors. All rights reserved.
 // Use of this source code is governed under the Apache License, Version 2.0
 // that can be found in the LICENSE file.
 
 // Package engine implements the core logic of the scheduler service.
 package engine
 
 import (
         "bytes"
         "encoding/json"
@@ skipping 14 matching lines @@
         ds "github.com/luci/gae/service/datastore"
         "github.com/luci/gae/service/info"
         mc "github.com/luci/gae/service/memcache"
         tq "github.com/luci/gae/service/taskqueue"
 
         "github.com/luci/luci-go/common/clock"
         "github.com/luci/luci-go/common/data/rand/mathrand"
         "github.com/luci/luci-go/common/data/stringset"
         "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/logging"
+        "github.com/luci/luci-go/common/retry"
         "github.com/luci/luci-go/server/auth"
         "github.com/luci/luci-go/server/auth/identity"
         "github.com/luci/luci-go/server/auth/signing"
         "github.com/luci/luci-go/server/tokens"
 
         "github.com/luci/luci-go/scheduler/appengine/catalog"
         "github.com/luci/luci-go/scheduler/appengine/schedule"
         "github.com/luci/luci-go/scheduler/appengine/task"
 )
 
@@ skipping 509 matching lines @@
         // Check the global cache.
         switch _, err := mc.GetKey(c, key); {
         case err == nil:
                 e.lock.Lock()
                 defer e.lock.Unlock()
                 e.doneFlags[key] = true
                 return nil
         case err == mc.ErrCacheMiss:
                 break
         default:
-                return errors.WrapTransient(err)
+                return retry.Tag.Apply(err)
         }
 
         // Do it.
         if err := cb(); err != nil {
                 return err
         }
 
         // Store in the global cache. Ignore errors, it's not a big deal.
         item := mc.NewItem(c, key)
         item.SetValue([]byte("ok"))
         item.SetExpiration(24 * time.Hour)
         if err := mc.Set(c, item); err != nil {
                 logging.Warningf(c, "Failed to write item to memcache - %s", err)
         }
 
         // Store in the local cache.
         e.lock.Lock()
         defer e.lock.Unlock()
         e.doneFlags[key] = true
         return nil
 }
 
 func (e *engineImpl) GetAllProjects(c context.Context) ([]string, error) {
         q := ds.NewQuery("Job").
                 Eq("Enabled", true).
                 Project("ProjectID").
                 Distinct(true)
         entities := []Job{}
         if err := ds.GetAll(c, q, &entities); err != nil {
-                return nil, errors.WrapTransient(err)
+                return nil, retry.Tag.Apply(err)
         }
         // Filter out duplicates, sort.
         projects := stringset.New(len(entities))
         for _, ent := range entities {
                 projects.Add(ent.ProjectID)
         }
         out := projects.ToSlice()
         sort.Strings(out)
         return out, nil
 }
 
 func (e *engineImpl) GetAllJobs(c context.Context) ([]*Job, error) {
         q := ds.NewQuery("Job").Eq("Enabled", true)
         return e.queryEnabledJobs(c, q)
 }
 
 func (e *engineImpl) GetProjectJobs(c context.Context, projectID string) ([]*Job, error) {
         q := ds.NewQuery("Job").Eq("Enabled", true).Eq("ProjectID", projectID)
         return e.queryEnabledJobs(c, q)
 }
 
 func (e *engineImpl) queryEnabledJobs(c context.Context, q *ds.Query) ([]*Job, error) {
         entities := []*Job{}
         if err := ds.GetAll(c, q, &entities); err != nil {
-                return nil, errors.WrapTransient(err)
+                return nil, retry.Tag.Apply(err)
         }
         // Non-ancestor query used, need to recheck filters.
         filtered := make([]*Job, 0, len(entities))
         for _, job := range entities {
                 if job.Enabled {
                         filtered = append(filtered, job)
                 }
         }
         return filtered, nil
 }
 
 func (e *engineImpl) GetJob(c context.Context, jobID string) (*Job, error) {
         job := &Job{JobID: jobID}
         switch err := ds.Get(c, job); {
         case err == nil:
                 return job, nil
         case err == ds.ErrNoSuchEntity:
                 return nil, nil
         default:
-                return nil, errors.WrapTransient(err)
+                return nil, retry.Tag.Apply(err)
         }
 }
 
 func (e *engineImpl) ListInvocations(c context.Context, jobID string, pageSize int, cursor string) ([]*Invocation, string, error) {
         if pageSize == 0 || pageSize > 500 {
                 pageSize = 500
         }
 
         // Deserialize the cursor.
         var cursorObj ds.Cursor
@@ skipping 23 matching lines @@
                         return nil
                 }
                 c, err := getCursor()
                 if err != nil {
                         return err
                 }
                 newCursor = c.String()
                 return ds.Stop
         })
         if err != nil {
-                return nil, "", errors.WrapTransient(err)
+                return nil, "", retry.Tag.Apply(err)
         }
         return out, newCursor, nil
 }
 
 func (e *engineImpl) GetInvocation(c context.Context, jobID string, invID int64) (*Invocation, error) {
         inv := &Invocation{
                 ID:     invID,
                 JobKey: ds.NewKey(c, "Job", jobID, 0, nil),
         }
         switch err := ds.Get(c, inv); {
         case err == nil:
                 return inv, nil
         case err == ds.ErrNoSuchEntity:
                 return nil, nil
         default:
-                return nil, errors.WrapTransient(err)
+                return nil, retry.Tag.Apply(err)
         }
 }
 
 func (e *engineImpl) GetInvocationsByNonce(c context.Context, invNonce int64) ([]*Invocation, error) {
         q := ds.NewQuery("Invocation").Eq("InvocationNonce", invNonce)
         entities := []*Invocation{}
         if err := ds.GetAll(c, q, &entities); err != nil {
-                return nil, errors.WrapTransient(err)
+                return nil, retry.Tag.Apply(err)
         }
         return entities, nil
 }
 
 func (e *engineImpl) UpdateProjectJobs(c context.Context, projectID string, defs []catalog.Definition) error {
         // JobID -> *Job map.
         existing, err := e.getProjectJobs(c, projectID)
         if err != nil {
                 return err
         }
@@ skipping 34 matching lines @@
                 go func(i int, jobID string) {
                         disableErrs.Assign(i, e.disableJob(c, jobID))
                         wg.Done()
                 }(i, jobID)
         }
 
         wg.Wait()
         if updateErrs.Get() == nil && disableErrs.Get() == nil {
                 return nil
         }
-        return errors.WrapTransient(errors.NewMultiError(updateErrs.Get(), disableErrs.Get()))
+        return retry.Tag.Apply(errors.NewMultiError(updateErrs.Get(), disableErrs.Get()))
 }
 
 func (e *engineImpl) ResetAllJobsOnDevServer(c context.Context) error {
         if !info.IsDevAppServer(c) {
                 return errors.New("ResetAllJobsOnDevServer must not be used in production")
         }
         q := ds.NewQuery("Job").Eq("Enabled", true)
         keys := []*ds.Key{}
         if err := ds.GetAll(c, q, &keys); err != nil {
-                return errors.WrapTransient(err)
+                return retry.Tag.Apply(err)
         }
         wg := sync.WaitGroup{}
         errs := errors.NewLazyMultiError(len(keys))
         for i, key := range keys {
                 wg.Add(1)
                 go func(i int, key *ds.Key) {
                         errs.Assign(i, e.resetJobOnDevServer(c, key.StringID()))
                         wg.Done()
                 }(i, key)
         }
         wg.Wait()
-        return errors.WrapTransient(errs.Get())
+        return retry.Tag.Apply(errs.Get())
 }
 
 // getProjectJobs fetches from ds all enabled jobs belonging to a given
 // project.
 func (e *engineImpl) getProjectJobs(c context.Context, projectID string) (map[string]*Job, error) {
         q := ds.NewQuery("Job").
                 Eq("Enabled", true).
                 Eq("ProjectID", projectID)
         entities := []*Job{}
         if err := ds.GetAll(c, q, &entities); err != nil {
-                return nil, errors.WrapTransient(err)
+                return nil, retry.Tag.Apply(err)
         }
         out := make(map[string]*Job, len(entities))
         for _, job := range entities {
                 if job.Enabled && job.ProjectID == projectID {
                         out[job.JobID] = job
                 }
         }
         return out, nil
 }
 
@@ skipping 25 matching lines @@
                         logging.Warningf(c, "Retrying transaction...")
                 }
                 stored := Job{JobID: jobID}
                 err := ds.Get(c, &stored)
                 if err != nil && err != ds.ErrNoSuchEntity {
                         return err
                 }
                 modified := stored
                 err = txn(c, &modified, err == ds.ErrNoSuchEntity)
                 if err != nil && err != errSkipPut {
-                        fatal = !errors.IsTransient(err)
+                        fatal = !retry.Tag.In(err)
                         return err
                 }
                 if err != errSkipPut && !modified.isEqual(&stored) {
                         return ds.Put(c, &modified)
                 }
                 return nil
         }, &defaultTransactionOptions)
         if err != nil {
                 logging.Errorf(c, "Job transaction failed: %s", err)
                 if fatal {
                         return err
                 }
                 // By now err is already transient (since 'fatal' is false) or it is commit
                 // error (i.e. produced by RunInTransaction itself, not by its callback).
                 // Need to wrap commit errors too.
-                return errors.WrapTransient(err)
+                return retry.Tag.Apply(err)
         }
         if attempt > 1 {
                 logging.Infof(c, "Committed on %d attempt", attempt)
         }
         return nil
 }
 
 // rollSM is called under transaction to perform a single state machine
 // transition. It sets up StateMachine instance, calls the callback, mutates
 // job.State in place (with a new state) and enqueues all emitted actions to
 // task queues.
 func (e *engineImpl) rollSM(c context.Context, job *Job, cb func(*StateMachine) error) error {
         sched, err := job.parseSchedule()
         if err != nil {
                 return fmt.Errorf("bad schedule %q - %s", job.effectiveSchedule(), err)
         }
         now := clock.Now(c).UTC()
         rnd := mathrand.Get(c)
         sm := StateMachine{
                 State:    job.State,
                 Now:      now,
                 Schedule: sched,
                 Nonce:    func() int64 { return rnd.Int63() + 1 },
                 Context:  c,
         }
         // All errors returned by state machine transition changes are transient.
         // Fatal errors (when we have them) should be reflected as a state changing
         // into "BROKEN" state.
         if err := cb(&sm); err != nil {
-                return errors.WrapTransient(err)
+                return retry.Tag.Apply(err)
         }
         if len(sm.Actions) != 0 {
                 if err := e.enqueueJobActions(c, job.JobID, sm.Actions); err != nil {
                         return err
                 }
         }
         if sm.State.State != job.State.State {
                 logging.Infof(c, "%s -> %s", job.State.State, sm.State.State)
         }
         job.State = sm.State
@@ skipping 69 matching lines @@
         i := 0
         for queueName, tasks := range qs {
                 wg.Add(1)
                 go func(i int, queueName string, tasks []*tq.Task) {
                         errs.Assign(i, tq.Add(c, queueName, tasks...))
                         wg.Done()
                 }(i, queueName, tasks)
                 i++
         }
         wg.Wait()
-        return errors.WrapTransient(errs.Get())
+        return retry.Tag.Apply(errs.Get())
 }
 
 // enqueueInvTimers submits all timers emitted by an invocation manager by
 // adding corresponding tasks to the task queue. See ExecuteSerializedAction for
 // place where these actions are interpreted.
 func (e *engineImpl) enqueueInvTimers(c context.Context, inv *Invocation, timers []invocationTimer) error {
         tasks := make([]*tq.Task, len(timers))
         for i, timer := range timers {
                 payload, err := json.Marshal(actionTaskPayload{
                         JobID:    inv.JobKey.StringID(),
                         InvID:    inv.ID,
                         InvTimer: &timer,
                 })
                 if err != nil {
                         return err
                 }
                 tasks[i] = &tq.Task{
                         Path:    e.TimersQueuePath,
                         ETA:     clock.Now(c).Add(timer.Delay),
                         Payload: payload,
                 }
         }
-        return errors.WrapTransient(tq.Add(c, e.TimersQueueName, tasks...))
+        return retry.Tag.Apply(tq.Add(c, e.TimersQueueName, tasks...))
 }
 
 func (e *engineImpl) ExecuteSerializedAction(c context.Context, action []byte, retryCount int) error {
         payload := actionTaskPayload{}
         if err := json.Unmarshal(action, &payload); err != nil {
                 return err
         }
         if payload.InvID == 0 {
                 return e.executeJobAction(c, &payload, retryCount)
         }
@@ skipping 272 matching lines @@
                 Started:  now,
                 Finished: now,
                 Status:   task.StatusOverrun,
         }
         if runningInvID == 0 {
                 inv.debugLog(c, "New invocation should be starting now, but previous one is still starting")
         } else {
                 inv.debugLog(c, "New invocation should be starting now, but previous one is still running: %d", runningInvID)
         }
         inv.debugLog(c, "Total overruns thus far: %d", overruns)
-        return errors.WrapTransient(ds.Put(c, &inv))
+        return retry.Tag.Apply(ds.Put(c, &inv))
 }
 
 // invocationTimerTick is called via Task Queue to handle AddTimer callbacks.
 //
 // See also handlePubSubMessage, it is quite similar.
 func (e *engineImpl) invocationTimerTick(c context.Context, jobID string, invID int64, timer *invocationTimer) error {
         c = logging.SetField(c, "JobID", jobID)
         c = logging.SetField(c, "InvID", invID)
 
         logging.Infof(c, "Handling invocation timer %q", timer.Name)
@@ skipping 16 matching lines @@
         ctl, err := e.controllerForInvocation(c, inv)
         if err != nil {
                 logging.Errorf(c, "Cannot get controller - %s", err)
                 return err
         }
 
         // Hand the message to the TaskManager.
         err = ctl.manager.HandleTimer(c, ctl, timer.Name, timer.Payload)
         if err != nil {
                 logging.Errorf(c, "Error when handling the timer - %s", err)
-                if !errors.IsTransient(err) && ctl.State().Status != task.StatusFailed {
+                if !retry.Tag.In(err) && ctl.State().Status != task.StatusFailed {
                         ctl.DebugLog("Fatal error when handling timer, aborting invocation - %s", err)
                         ctl.State().Status = task.StatusFailed
                 }
         }
 
         // Save anyway, to preserve the invocation log.
         saveErr := ctl.Save(c)
         if saveErr != nil {
                 logging.Errorf(c, "Error when saving the invocation - %s", saveErr)
         }
 
         // Retry the delivery if at least one error is transient. HandleTimer must be
         // idempotent.
         switch {
         case err == nil && saveErr == nil:
                 return nil
-        case errors.IsTransient(saveErr):
+        case retry.Tag.In(err):
                 return saveErr
         default:
                 return err // transient or fatal
         }
 }
 
 // startInvocation is called via task queue to start running a job. This call
 // may be retried by task queue service.
 func (e *engineImpl) startInvocation(c context.Context, jobID string, invocationNonce int64,
         triggeredBy identity.Identity, retryCount int) error {
@@ skipping 129 matching lines @@
         // In either case, invocation never ends up in StatusStarting state.
         err = ctl.manager.LaunchTask(c, ctl)
         if ctl.State().Status == task.StatusStarting {
                 ctl.State().Status = task.StatusFailed
                 if err == nil {
                         err = fmt.Errorf("LaunchTask didn't move invocation out of StatusStarting")
                 }
         }
 
         // Give up retrying on transient errors after some number of attempts.
-        if errors.IsTransient(err) && retryCount+1 >= invocationRetryLimit {
+        if retry.Tag.In(err) && retryCount+1 >= invocationRetryLimit {
                 err = fmt.Errorf("Too many retries, giving up (original error - %s)", err)
         }
 
         // If asked to retry the invocation (by returning a transient error), do not
         // touch Job entity when saving the current (failed) invocation. That way Job
         // stays in "QUEUED" state (indicating it's queued for a new invocation).
-        retryInvocation := errors.IsTransient(err)
-        if saveErr := ctl.saveImpl(c, !retryInvocation); saveErr != nil {
+        if saveErr := ctl.saveImpl(c, !retry.Tag.In(err)); saveErr != nil {
                 logging.Errorf(c, "Failed to save invocation state - %s", saveErr)
                 if err == nil {
                         err = saveErr
                 }
         }
 
         // Returning transient error here causes the task queue to retry the task.
         if err != nil {
                 logging.WithError(err).Errorf(c, "Invocation failed to start")
         }
@@ skipping 139 matching lines @@
         _, sub := e.genTopicAndSubNames(c, taskManagerName, publisher)
         msg, ack, err := pullSubcription(c, sub, "")
         if err != nil {
                 return err
         }
         if msg == nil {
                 logging.Infof(c, "No new PubSub messages")
                 return nil
         }
         err = e.handlePubSubMessage(c, msg)
-        if err == nil || !errors.IsTransient(err) {
+        if err == nil || !retry.Tag.In(err) {
                 ack() // ack only on success of fatal errors (to stop redelivery)
         }
         return err
 }
 
 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
@@ skipping 30 matching lines @@
         ctl, err := e.controllerForInvocation(c, inv)
         if err != nil {
                 logging.Errorf(c, "Cannot get controller - %s", err)
                 return err
         }
 
         // Hand the message to the TaskManager.
         err = ctl.manager.HandleNotification(c, ctl, msg)
         if err != nil {
                 logging.Errorf(c, "Error when handling the message - %s", err)
-                if !errors.IsTransient(err) && ctl.State().Status != task.StatusFailed {
+                if !retry.Tag.In(err) && ctl.State().Status != task.StatusFailed {
                         ctl.DebugLog("Fatal error when handling PubSub notification, aborting invocation - %s", err)
                         ctl.State().Status = task.StatusFailed
                 }
         }
 
         // Save anyway, to preserve the invocation log.
         saveErr := ctl.Save(c)
         if saveErr != nil {
                 logging.Errorf(c, "Error when saving the invocation - %s", saveErr)
         }
 
         // Retry the delivery if at least one error is transient. HandleNotification
         // must be idempotent.
         switch {
         case err == nil && saveErr == nil:
                 return nil
-        case errors.IsTransient(saveErr):
+        case retry.Tag.In(saveErr):
                 return saveErr
         default:
                 return err // transient or fatal
         }
 }
 
 ////////////////////////////////////////////////////////////////////////////////
 // TaskController.
 
 type taskController struct {
@@ skipping 80 matching lines @@
 }
 
 // Save is part of task.Controller interface.
 func (ctl *taskController) Save(ctx context.Context) error {
         return ctl.saveImpl(ctx, true)
 }
 
 // errUpdateConflict means Invocation is being modified by two TaskController's
 // concurrently. It should not be happening often. If it happens, task queue
 // call is retried to rerun the two-part transaction from scratch.
-var errUpdateConflict = errors.WrapTransient(errors.New("concurrent modifications of single Invocation"))
+var errUpdateConflict = errors.New("concurrent modifications of single Invocation", retry.Tag)
 
 // saveImpl uploads updated Invocation to the datastore. If updateJob is true,
 // it will also roll corresponding state machine forward.
 func (ctl *taskController) saveImpl(ctx context.Context, updateJob bool) (err error) {
         // Mutate copy in case transaction below fails. Also unpacks ctl.state back
         // into the entity (reverse of 'populateState').
         saving := ctl.saved
         saving.Status = ctl.state.Status
         saving.TaskData = append([]byte(nil), ctl.state.TaskData...)
         saving.ViewURL = ctl.state.ViewURL
@@ skipping 35 matching lines @@
                 // expect it to be.
                 mostRecent := Invocation{
                         ID:     saving.ID,
                         JobKey: saving.JobKey,
                 }
                 switch err := ds.Get(c, &mostRecent); {
                 case err == ds.ErrNoSuchEntity: // should not happen
                         logging.Errorf(c, "Invocation is suddenly gone")
                         return errors.New("invocation is suddenly gone")
                 case err != nil:
-                        return errors.WrapTransient(err)
+                        return retry.Tag.Apply(err)
                 }
 
                 // Make sure no one touched it while we were handling the invocation.
                 if saving.MutationsCount != mostRecent.MutationsCount+1 {
                         logging.Errorf(c, "Invocation was modified by someone else while we were handling it")
                         return errUpdateConflict
                 }
 
                 // Store the invocation entity and schedule invocation timers regardless of
                 // the current state of the Job entity. The table of all invocations is
@@ skipping 36 matching lines @@
                 }
                 if hasFinished {
                         return ctl.eng.rollSM(c, job, func(sm *StateMachine) error {
                                 sm.OnInvocationDone(saving.ID)
                                 return nil
                         })
                 }
                 return nil
         })
 }