Fix agent polling

server/grpc/rpc.go

@@ -56,28 +56,57 @@ func (s *RPC) Next(c context.Context, agentFilter rpc.Filter) (*rpc.Workflow, er
 		log.Debug().Msgf("agent connected: %s: polling", hostname)
 	}
 
-	fn := createFilterFunc(agentFilter)
+	filterFn := createFilterFunc(agentFilter)
 	for {
 		agent, err := s.getAgentFromContext(c)
 		if err != nil {
 			return nil, err
-		} else if agent.NoSchedule {
-			return nil, nil
 		}
 
-		task, err := s.queue.Poll(c, agent.ID, fn)
+		if agent.NoSchedule {
+			time.Sleep(1 * time.Second)
+			continue
+		}
+
+		// poll blocks until a task is available
+		task, err := s.queue.Poll(c, agent.ID, filterFn)
 		if err != nil {
 			return nil, err
-		} else if task == nil {
+		}
+
+		if task == nil {
 			return nil, nil
 		}
 
+		agent, err = s.getAgentFromContext(c)
+		if err != nil {
+			return nil, err
+		}
+
+		// the agent is meanwhile marked as no-schedule, push the task back to the queue
+		if agent.NoSchedule {
+			// we need to mark the task as done, otherwise it will be stuck in the queue
+			if err := s.queue.Error(c, task.ID, errors.New("agent is marked as no-schedule")); err != nil {
+				log.Error().Err(err).Msgf("mark task '%s' error failed", task.ID)
+			}
+
+			task.AgentID = -1
+			task.ID += "1"
+			if err := s.queue.Push(c, task); err != nil {
+				log.Error().Err(err).Msgf("push task '%s' back to queue failed", task.ID)
+			}
+
+			time.Sleep(1 * time.Second)
+			continue
+		}
+
 		if task.ShouldRun() {
 			workflow := new(rpc.Workflow)
 			err = json.Unmarshal(task.Data, workflow)
 			return workflow, err
 		}
 
+		// task should not run, so mark it as done
 		if err := s.Done(c, task.ID, rpc.State{}); err != nil {
 			log.Error().Err(err).Msgf("mark task '%s' done failed", task.ID)
 		}

server/queue/fifo.go

@@ -70,7 +70,7 @@ func (q *fifo) Push(_ context.Context, task *model.Task) error {
 	return nil
 }
 
-// Push pushes an item to the tail of this queue.
+// PushAtOnce pushes items to the tail of this queue.
 func (q *fifo) PushAtOnce(_ context.Context, tasks []*model.Task) error {
 	q.Lock()
 	for _, task := range tasks {
@@ -111,12 +111,12 @@ func (q *fifo) Done(_ context.Context, id string, exitStatus model.StatusValue)
 	return q.finished([]string{id}, exitStatus, nil)
 }
 
-// Error signals that the item is done executing with error.
+// Error signals that the item is done executing with an error.
 func (q *fifo) Error(_ context.Context, id string, err error) error {
 	return q.finished([]string{id}, model.StatusFailure, err)
 }
 
-// ErrorAtOnce signals that the item is done executing with error.
+// ErrorAtOnce signals that the items are done executing with an error.
 func (q *fifo) ErrorAtOnce(_ context.Context, id []string, err error) error {
 	return q.finished(id, model.StatusFailure, err)
 }
@@ -145,7 +145,7 @@ func (q *fifo) Evict(c context.Context, id string) error {
 	return q.EvictAtOnce(c, []string{id})
 }
 
-// Evict removes a pending task from the queue.
+// EvictAtOnce removes multiple pending tasks from the queue.
 func (q *fifo) EvictAtOnce(_ context.Context, ids []string) error {
 	q.Lock()
 	defer q.Unlock()
@@ -200,7 +200,6 @@ func (q *fifo) Info(_ context.Context) InfoT {
 	stats.Stats.Pending = q.pending.Len()
 	stats.Stats.WaitingOnDeps = q.waitingOnDeps.Len()
 	stats.Stats.Running = len(q.running)
-	stats.Stats.Complete = 0 // TODO: implement this
 
 	for e := q.pending.Front(); e != nil; e = e.Next() {
 		task, _ := e.Value.(*model.Task)

server/queue/queue.go

@@ -40,7 +40,6 @@ type InfoT struct {
 		Pending       int `json:"pending_count"`
 		WaitingOnDeps int `json:"waiting_on_deps_count"`
 		Running       int `json:"running_count"`
-		Complete      int `json:"completed_count"`
 	} `json:"stats"`
 	Paused bool `json:"paused"`
 } //	@name InfoT