refactor(worker): implement immediate retry in same worker

manager.go

@@ -112,9 +112,7 @@ func (m *Manager[T]) next() {
 	m.logger.Debug("got task", "id", task.GetID())
 	go func() {
 		defer func() {
-			if task.GetState() == StateWaitingRetry {
-				m.queue.Push(task)
-			}
+			// Execute() handles all retries in the same worker, task is completed when it returns
 			m.workers.Put(worker)
 			m.next()
 		}()

worker.go

@@ -16,12 +16,6 @@ type Worker[T Task] struct {
 
 // Execute executes the task
 func (w Worker[T]) Execute(task T) {
-	if isRetry(task) {
-		task.SetState(StateBeforeRetry)
-		if hook, ok := Task(task).(OnBeforeRetry); ok {
-			hook.OnBeforeRetry()
-		}
-	}
 	onError := func(err error) {
 		task.SetErr(err)
 		if errors.Is(err, context.Canceled) {
@@ -37,23 +31,41 @@ func (w Worker[T]) Execute(task T) {
 			task.SetState(StateFailed)
 		}
 	}
-	defer func() {
-		if err := recover(); err != nil {
-			log.Printf("error [%s] while run task [%s],stack trace:\n%s", err, task.GetID(), getCurrentGoroutineStack())
-			onError(NewErr(fmt.Sprintf("panic: %v", err)))
+	// Retry immediately in the same worker until success or max retry exhausted
+	for {
+		func() {
+			defer func() {
+				if err := recover(); err != nil {
+					log.Printf("error [%s] while run task [%s],stack trace:\n%s", err, task.GetID(), getCurrentGoroutineStack())
+					onError(NewErr(fmt.Sprintf("panic: %v", err)))
+				}
+			}()
+
+			if isRetry(task) {
+				task.SetState(StateBeforeRetry)
+				if hook, ok := Task(task).(OnBeforeRetry); ok {
+					hook.OnBeforeRetry()
+				}
+			}
+
+			task.SetState(StateRunning)
+			err := task.Run()
+			if err != nil {
+				onError(err)
+				return
+			}
+
+			task.SetState(StateSucceeded)
+			if onSucceeded, ok := Task(task).(OnSucceeded); ok {
+				onSucceeded.OnSucceeded()
+			}
+			task.SetErr(nil)
+		}()
+
+		if task.GetState() != StateWaitingRetry {
+			return
 		}
-	}()
-	task.SetState(StateRunning)
-	err := task.Run()
-	if err != nil {
-		onError(err)
-		return
-	}
-	task.SetState(StateSucceeded)
-	if onSucceeded, ok := Task(task).(OnSucceeded); ok {
-		onSucceeded.OnSucceeded()
 	}
-	task.SetErr(nil)
 }
 
 // WorkerPool is the pool of workers