[worker] fix possible task scheduling deadlock

pool.go

@@ -12,6 +12,7 @@ type TaskFn func()
 // Pool represents worker pool.
 type Pool struct {
 	conf                        Config
+	wg                          sync.WaitGroup
 	tasks                       chan TaskFn
 	quit                        chan struct{}
 	mu                          sync.Mutex
@@ -96,7 +97,7 @@ func (p *Pool) add(t TaskFn) error {
 		return ErrPoolFull
 	}
 
-	left := time.Since(started)
+	left := time.Since(started) - p.conf.TaskScheduleTimeout
 	if left <= 0 {
 		return ErrScheduleTimeout
 	}

pool_test.go

@@ -193,17 +193,21 @@ func TestPoolCloseAfterWorkerTask(t *testing.T) {
 		UnstoppableWorkers: 0,
 		ExtraWorkerTTL:     time.Minute,
 	})
+	done := make(chan struct{})
 	err := pool.Add(TaskFn(func() {
-		time.Sleep(20 * time.Millisecond)
+		select {
+		case <-time.After(10 * time.Millisecond):
+			close(done)
+		}
 	}))
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	time.Sleep(10 * time.Millisecond)
-
 	pool.Shutdown()
-	time.Sleep(500 * time.Millisecond)
+	select {
+	case <-done:
+	}
 }
 
 func TestPoolWithAdditionalWorkersClose(t *testing.T) {

worker.go

@@ -56,19 +56,16 @@ func newAdditionalWorker(tasks <-chan TaskFn, quit <-chan struct{}, params *work
 }
 
 func (w *additionalWorker) run(t TaskFn) {
-	select {
-	case <-w.quit:
-		return
-	default:
-	}
+	// Execute the task anyway.
+	// But next we can easily stop the worker.
 
 	ticker := time.NewTicker(w.conf.ttl)
 	defer ticker.Stop()
 
 	w.conf.onExtraWorkerSpawned()
 	defer w.conf.onExtraWorkerFinished()
 
-	t()
+	w.runTask(t)
 
 	for {
 		select {