Remove internal retry from AddTask

service/matching/tasklist/task_list_manager.go

@@ -34,7 +34,6 @@ import (
 
 	"github.com/uber/cadence/client/matching"
 	"github.com/uber/cadence/common"
-	"github.com/uber/cadence/common/backoff"
 	"github.com/uber/cadence/common/cache"
 	"github.com/uber/cadence/common/clock"
 	"github.com/uber/cadence/common/cluster"
@@ -54,11 +53,15 @@ const (
 	// Time budget for empty task to propagate through the function stack and be returned to
 	// pollForActivityTask or pollForDecisionTask handler.
 	returnEmptyTaskTimeBudget = time.Second
+	// maxSyncMatchWaitTime is the max amount of time that we are willing to wait for a sync match to happen
+	maxSyncMatchWaitTime = 200 * time.Millisecond
 )
 
 var (
 	// ErrNoTasks is exported temporarily for integration test
 	ErrNoTasks                      = errors.New("no tasks")
+	ErrRemoteSyncMatchFailed        = &types.RemoteSyncMatchedError{Message: "remote sync match failed"}
+	ErrTooManyOutstandingTasks      = &types.ServiceBusyError{Message: "Too many outstanding appends to the TaskList"}
 	_stickyPollerUnavailableError   = &types.StickyWorkerUnavailableError{Message: "sticky worker is unavailable, please use non-sticky task list."}
 	persistenceOperationRetryPolicy = common.CreatePersistenceRetryPolicy()
 	taskListActivityTypeTag         = metrics.TaskListTypeTag("activity")
@@ -120,13 +123,6 @@ type (
 	}
 )
 
-const (
-	// maxSyncMatchWaitTime is the max amount of time that we are willing to wait for a sync match to happen
-	maxSyncMatchWaitTime = 200 * time.Millisecond
-)
-
-var errRemoteSyncMatchFailed = &types.RemoteSyncMatchedError{Message: "remote sync match failed"}
-
 func NewManager(
 	domainCache cache.DomainCache,
 	logger log.Logger,
@@ -250,7 +246,6 @@ func (c *taskListManagerImpl) handleErr(err error) error {
 // be written to database and later asynchronously matched with a poller
 func (c *taskListManagerImpl) AddTask(ctx context.Context, params AddTaskParams) (bool, error) {
 	c.startWG.Wait()
-
 	if c.shouldReload() {
 		c.Stop()
 		return false, errShutdown
@@ -259,57 +254,39 @@ func (c *taskListManagerImpl) AddTask(ctx context.Context, params AddTaskParams)
 		// request sent by history service
 		c.liveness.MarkAlive()
 	}
-	var syncMatch bool
-	_, err := c.executeWithRetry(func() (interface{}, error) {
-		if err := ctx.Err(); err != nil {
-			return nil, err
-		}
-
-		domainEntry, err := c.domainCache.GetDomainByID(params.TaskInfo.DomainID)
-		if err != nil {
-			return nil, err
-		}
-
-		isForwarded := params.ForwardedFrom != ""
-
-		if _, err := domainEntry.IsActiveIn(c.clusterMetadata.GetCurrentClusterName()); err != nil {
-			// standby task, only persist when task is not forwarded from child partition
-			syncMatch = false
-			if isForwarded {
-				return &persistence.CreateTasksResponse{}, errRemoteSyncMatchFailed
-			}
-
-			r, err := c.taskWriter.appendTask(params.TaskInfo)
-			return r, err
-		}
+	domainEntry, err := c.domainCache.GetDomainByID(params.TaskInfo.DomainID)
+	if err != nil {
+		return false, err
+	}
 
+	if isActive, err := domainEntry.IsActiveIn(c.clusterMetadata.GetCurrentClusterName()); err == nil && isActive {
 		isolationGroup, err := c.getIsolationGroupForTask(ctx, params.TaskInfo)
 		if err != nil {
 			return false, err
 		}
 		// active task, try sync match first
-		syncMatch, err = c.trySyncMatch(ctx, params, isolationGroup)
+		syncMatch, err := c.trySyncMatch(ctx, params, isolationGroup)
 		if syncMatch {
-			return &persistence.CreateTasksResponse{}, err
-		}
-		if params.ActivityTaskDispatchInfo != nil {
-			return false, errRemoteSyncMatchFailed
+			return syncMatch, err
 		}
+	}
 
-		if isForwarded {
-			// forwarded from child partition - only do sync match
-			// child partition will persist the task when sync match fails
-			return &persistence.CreateTasksResponse{}, errRemoteSyncMatchFailed
-		}
+	if params.ActivityTaskDispatchInfo != nil {
+		return false, ErrRemoteSyncMatchFailed
+	}
 
-		return c.taskWriter.appendTask(params.TaskInfo)
-	})
+	if params.ForwardedFrom != "" {
+		// forwarded from child partition - only do sync match
+		// child partition will persist the task when sync match fails
+		return false, ErrRemoteSyncMatchFailed
+	}
 
-	if err == nil && !syncMatch {
+	_, err = c.taskWriter.appendTask(ctx, params.TaskInfo)
+	if err == nil {
 		c.taskReader.Signal()
 	}
 
-	return syncMatch, err
+	return false, c.handleErr(err)
 }
 
 // DispatchTask dispatches a task to a poller. When there are no pollers to pick
@@ -358,7 +335,7 @@ func (c *taskListManagerImpl) getTask(ctx context.Context, maxDispatchPerSecond
 	// reached, instead of emptyTask, context timeout error is returned to the frontend by the rpc stack,
 	// which counts against our SLO. By shortening the timeout by a very small amount, the emptyTask can be
 	// returned to the handler before a context timeout error is generated.
-	childCtx, cancel := c.newChildContext(ctx, c.config.LongPollExpirationInterval(), returnEmptyTaskTimeBudget)
+	childCtx, cancel := newChildContext(ctx, c.config.LongPollExpirationInterval(), returnEmptyTaskTimeBudget)
 	defer cancel()
 
 	isolationGroup := IsolationGroupFromContext(ctx)
@@ -494,24 +471,6 @@ func (c *taskListManagerImpl) TaskListID() *Identifier {
 	return c.taskListID
 }
 
-// Retry operation on transient error. On rangeID update by another process calls c.Stop().
-func (c *taskListManagerImpl) executeWithRetry(
-	operation func() (interface{}, error),
-) (result interface{}, err error) {
-
-	op := func() error {
-		result, err = operation()
-		return err
-	}
-
-	throttleRetry := backoff.NewThrottleRetry(
-		backoff.WithRetryPolicy(persistenceOperationRetryPolicy),
-		backoff.WithRetryableError(persistence.IsTransientError),
-	)
-	err = c.handleErr(throttleRetry.Do(context.Background(), op))
-	return
-}
-
 func (c *taskListManagerImpl) trySyncMatch(ctx context.Context, params AddTaskParams, isolationGroup string) (bool, error) {
 	task := newInternalTask(params.TaskInfo, nil, params.Source, params.ForwardedFrom, true, params.ActivityTaskDispatchInfo, isolationGroup)
 	childCtx := ctx
@@ -523,7 +482,7 @@ func (c *taskListManagerImpl) trySyncMatch(ctx context.Context, params AddTaskPa
 	if !task.IsForwarded() {
 		// when task is forwarded from another matching host, we trust the context as is
 		// otherwise, we override to limit the amount of time we can block on sync match
-		childCtx, cancel = c.newChildContext(ctx, waitTime, time.Second)
+		childCtx, cancel = newChildContext(ctx, waitTime, time.Second)
 	}
 	var matched bool
 	var err error
@@ -542,7 +501,7 @@ func (c *taskListManagerImpl) trySyncMatch(ctx context.Context, params AddTaskPa
 // method to create child context when childContext cannot use
 // all of parent's deadline but instead there is a need to leave
 // some time for parent to do some post-work
-func (c *taskListManagerImpl) newChildContext(
+func newChildContext(
 	parent context.Context,
 	timeout time.Duration,
 	tailroom time.Duration,
@@ -649,10 +608,6 @@ func getTaskListTypeTag(taskListType int) metrics.Tag {
 	}
 }
 
-func createServiceBusyError(msg string) *types.ServiceBusyError {
-	return &types.ServiceBusyError{Message: msg}
-}
-
 func rangeIDToTaskIDBlock(rangeID, rangeSize int64) taskIDBlock {
 	return taskIDBlock{
 		start: (rangeID-1)*rangeSize + 1,

service/matching/tasklist/task_reader.go

@@ -355,10 +355,10 @@ func (tr *taskReader) completeTask(task *persistence.TaskInfo, err error) {
 		// Note that RecordTaskStarted only fails after retrying for a long time, so a single task will not be
 		// re-written to persistence frequently.
 		op := func() error {
-			_, err := tr.taskWriter.appendTask(task)
+			_, err := tr.taskWriter.appendTask(tr.cancelCtx, task)
 			return err
 		}
-		err = tr.throttleRetry.Do(context.Background(), op)
+		err = tr.throttleRetry.Do(tr.cancelCtx, op)
 		if err != nil {
 			// OK, we also failed to write to persistence.
 			// This should only happen in very extreme cases where persistence is completely down.

service/matching/tasklist/task_writer.go

@@ -119,7 +119,7 @@ func (w *taskWriter) isStopped() bool {
 	return atomic.LoadInt64(&w.stopped) == 1
 }
 
-func (w *taskWriter) appendTask(taskInfo *persistence.TaskInfo) (*persistence.CreateTasksResponse, error) {
+func (w *taskWriter) appendTask(ctx context.Context, taskInfo *persistence.TaskInfo) (*persistence.CreateTasksResponse, error) {
 	if w.isStopped() {
 		return nil, errShutdown
 	}
@@ -140,8 +140,8 @@ func (w *taskWriter) appendTask(taskInfo *persistence.TaskInfo) (*persistence.Cr
 			// it to cassandra, just bail out and fail this request
 			return nil, errShutdown
 		}
-	default: // channel is full, throttle
-		return nil, createServiceBusyError("Too many outstanding appends to the TaskList")
+	case <-ctx.Done(): // channel is full, throttle
+		return nil, ErrTooManyOutstandingTasks
 	}
 }