Support infinite Task.Spec.retryLimit

Instead, Task.Status.History now only records the limited recent
TaskRecords. The number varies on backend type.  10 for Redis.

Author: everpeace

README.md

@@ -319,8 +319,10 @@ name: "this is just an display name"
 #  redis: 1KB
 payload: |
   You can define any task information in payload
-# retryLimit max value varies on backend type to prevent from overloading backend.
-#  redis: 10
+# retryLimit is the maximum number of retry (negative number means infinite)
+# NOTE: only the limited number of recent task records will be recorded in its status.
+#       so, if you set large value or infinite here, you will loose old task records.
+#       please see the description of status.history field in the next section.
 retryLimit: 3
 # timeoutSeconds is for task handler timeout.
 # If set positive value, task handler timeout for processing this task.
@@ -361,7 +363,12 @@ status:
     # timestamps
     receivedAt: 2020-02-12T20:20:39.350631+09:00
     startedAt: 2020-02-12T20:20:39.351479+09:00
-  # history of processing the task.
+  # history of recent records of processing the task.
+  # the limited number of recent records are recorded in this field.
+  # the value varies on backend types to prvent overloading backends:
+  # - redis: 10 entries
+  # NOTE: so, if you set larger value than this limit in spec.rertryLimit,
+  #       you will loose old task records.
   history:
     # TaskRecord: 
     #   this represents a record of task handler invocation 

pkg/apis/task/task.go

@@ -178,7 +178,7 @@ func (t *Task) IsWorkerLost(defaultTimeoutSeconds int) bool {
 	return t.Status.CurrentRecord.ReceivedAt.Add(timeout).Before(time.Now())
 }
 
-func (t *Task) SetSuccess(payload *string, message *string) error {
+func (t *Task) SetSuccess(payload *string, message *string, historyLengthLimit int) error {
 	if t.Status.Phase != TaskPhaseProcessing {
 		return errors.Errorf("invalid status: actual=%s expected=%s", t.Status.Phase, TaskPhaseProcessing)
 	}
@@ -205,10 +205,14 @@ func (t *Task) SetSuccess(payload *string, message *string) error {
 	}
 	t.Status.History = append(t.Status.History, *current)
 
+	if len(t.Status.History) > historyLengthLimit {
+		t.Status.History = t.Status.History[len(t.Status.History)-historyLengthLimit:]
+	}
+
 	return nil
 }
 
-func (t *Task) RecordFailure(reason TaskResultReason, payload *string, message *string) (bool, error) {
+func (t *Task) RecordFailure(reason TaskResultReason, payload *string, message *string, historyLengthLimit int) (bool, error) {
 	if t.Status.Phase != TaskPhaseProcessing && t.Status.Phase != TaskPhaseReceived {
 		return false, errors.Errorf("invalid status: actual=%s expected=[%s,%s]", t.Status.Phase, TaskPhaseProcessing, TaskPhaseReceived)
 	}
@@ -233,13 +237,16 @@ func (t *Task) RecordFailure(reason TaskResultReason, payload *string, message *
 		t.Status.History = []TaskRecord{}
 	}
 	t.Status.History = append(t.Status.History, *current)
+	if len(t.Status.History) > historyLengthLimit {
+		t.Status.History = t.Status.History[len(t.Status.History)-historyLengthLimit:]
+	}
+
 	t.Status.FailureCount = t.Status.FailureCount + 1
 
 	requeue := true
 	t.Status.Phase = TaskPhasePending
-
 	// no requeue because retry exceeded
-	if t.Status.FailureCount > t.Spec.RetryLimit {
+	if t.Spec.RetryLimit >= 0 && t.Status.FailureCount > t.Spec.RetryLimit {
 		requeue = false
 		t.Status.Phase = TaskPhaseFailed
 	}

pkg/backend/redis/redis_test.go

@@ -89,6 +89,7 @@ var (
 		TimeoutSeconds: 60,
 	}
 	SampleInvalidTaskSpec = task.TaskSpec{
+		Name:           strings.Repeat("a", MaxNameLength+1),
 		Payload:        strings.Repeat("x", PayloadMaxSizeInKB*KB+1),
 		RetryLimit:     100,
 		TimeoutSeconds: 0,
@@ -712,8 +713,8 @@ var _ = Describe("Backend", func() {
 						vErr, ok := err.(*util.ValidationError)
 						Expect(ok).To(Equal(true))
 						Expect(len(vErr.Errors)).To(Equal(2))
+						Expect(vErr.Error()).To(ContainSubstring("TaskSpec.Name max length"))
 						Expect(vErr.Error()).To(ContainSubstring("TaskSpec.Payload max size is"))
-						Expect(vErr.Error()).To(ContainSubstring("TaskSpec.retryLimit max is"))
 					})
 				})
 				When("Spec is valid", func() {

pkg/backend/redis/task.go

@@ -41,7 +41,7 @@ const (
 	KB                 = 1 << 10
 	PayloadMaxSizeInKB = 1
 	MessageMaxSizeInKB = 1
-	RetryLimitMax      = 10
+	HistoryLengthMax   = 10
 	MaxNameLength      = 1024
 )
 
@@ -650,6 +650,7 @@ func (b *Backend) SetSucceeded(ctx context.Context, queueUID, workerUID uuid.UUI
 		err = t.SetSuccess(
 			util.Truncate(resultPayload, PayloadMaxSizeInKB*KB),
 			util.Truncate(message, MessageMaxSizeInKB*KB),
+			HistoryLengthMax,
 		)
 		if err != nil {
 			dlerr := b.invalidMessageDLError(
@@ -791,6 +792,7 @@ func (b *Backend) RecordFailure(ctx context.Context, queueUID, workerUID uuid.UU
 			reason,
 			util.Truncate(resultPayload, PayloadMaxSizeInKB*KB),
 			util.Truncate(message, MessageMaxSizeInKB*KB),
+			HistoryLengthMax,
 		)
 		if err != nil {
 			dlerr := b.invalidMessageDLError(
@@ -924,11 +926,6 @@ func (b *Backend) validateTaskSpec(s task.TaskSpec) error {
 			errors.Errorf("TaskSpec.Payload max size is %d Bytes (actual=%d Bytes)", maxBytes, len(s.Payload)),
 		)
 	}
-	if s.RetryLimit > RetryLimitMax {
-		vErrors = multierror.Append(vErrors,
-			errors.Errorf("TaskSpec.retryLimit max is %d (actual=%d)", RetryLimitMax, s.RetryLimit),
-		)
-	}
 	if vErrors != nil {
 		return (*util.ValidationError)(vErrors)
 	}