perf(server): keep the pending queue sorted instead of re-sorting per dispatch (RIG-1180)

server/queue/fifo.go

@@ -73,11 +73,12 @@ func NewMemoryQueue(ctx context.Context) Queue {
 	return q
 }
 
-// PushAtOnce pushes multiple tasks to the tail of this queue.
+// PushAtOnce pushes multiple tasks to this queue, each into its creation-order
+// position among the tasks already pending.
 func (q *fifo) PushAtOnce(_ context.Context, tasks []*model.Task) error {
 	q.Lock()
 	for _, task := range tasks {
-		q.pending.PushBack(task)
+		q.pushPending(task)
 	}
 	q.Unlock()
 	return nil
@@ -287,10 +288,26 @@ func (q *fifo) process() {
 }
 
 func (q *fifo) filterWaiting() {
-	// resubmits all waiting tasks to pending, deps may have cleared
+	// Resubmit all waiting tasks to pending; deps may have cleared. Both lists
+	// are ordered by taskOrderLess — waitingOnDeps is rebuilt below by scanning
+	// pending in order — so this is a merge of two sorted lists, walked once,
+	// rather than a search per task. A drained task from an older pipeline
+	// therefore costs no more than the newer tail it sorts ahead of.
+	at := q.pending.Front()
 	for element := q.waitingOnDeps.Front(); element != nil; element = element.Next() {
 		task, _ := element.Value.(*model.Task)
-		q.pending.PushBack(task)
+		for at != nil {
+			other, _ := at.Value.(*model.Task)
+			if taskOrderLess(task, other) {
+				break
+			}
+			at = at.Next()
+		}
+		if at == nil {
+			q.pending.PushBack(task)
+			continue
+		}
+		q.pending.InsertBefore(task, at)
 	}
 
 	// rebuild waitingDeps
@@ -315,7 +332,7 @@ func (q *fifo) assignToWorker() (*list.Element, *worker) {
 	var bestWorker *worker
 	var bestScore int
 
-	for _, element := range q.pendingByCreation() {
+	for element := q.pending.Front(); element != nil; element = element.Next() {
 		task, _ := element.Value.(*model.Task)
 		log.Debug().Msgf("queue: trying to assign task: %v with deps %v", task.ID, task.Dependencies)
 
@@ -342,35 +359,73 @@ func (q *fifo) assignToWorker() (*list.Element, *worker) {
 	return nil, nil
 }
 
-// pendingByCreation returns the pending tasks' list elements ordered by
-// taskOrderLess (creation time, then workflow name). Dispatch walks this order
-// so a workflow whose dependencies have cleared keeps its creation-order
-// priority instead of being overtaken by tasks from pipelines created later,
-// regardless of the order tasks were appended to the pending list. The sort is
-// stable, so elements the comparator treats as equal keep their relative order.
+// pushPending inserts a task into the pending list, keeping the list ordered by
+// taskOrderLess (creation time, then workflow name). The sort key is immutable
+// once a task is queued, so maintaining the order at insertion time is
+// equivalent to — and replaces — re-sorting the whole pending set on every
+// dispatch. Dispatch therefore walks the list directly, and a workflow whose
+// dependencies have cleared keeps its creation-order priority instead of being
+// overtaken by tasks from pipelines created later.
+//
+// A task is inserted after the tasks it compares equal to, matching the
+// append-then-stable-sort order this replaces. Expects the queue to be locked
+// by the caller.
 //
-// This orders the post-filterWaiting pending set (assignToWorker runs right
-// after filterWaiting on each process tick), where pending never simultaneously
-// holds a task and a task that depends on it, so reordering cannot dispatch a
-// dependent before its dependency. Expects the queue to be locked by the caller.
-func (q *fifo) pendingByCreation() []*list.Element {
-	elements := make([]*list.Element, 0, q.pending.Len())
+// The scan runs back-to-front, which is one step for the common case of a new
+// pipeline's batch appended to the tail. A task belonging ahead of the whole
+// list is that direction's worst case, so it is taken by a front check first.
+// The check is on a strict less-than, so it never reorders equals.
+func (q *fifo) pushPending(task *model.Task) {
+	if front := q.pending.Front(); front == nil {
+		q.pending.PushFront(task)
+		return
+	} else if other, _ := front.Value.(*model.Task); taskOrderLess(task, other) {
+		q.pending.PushFront(task)
+		return
+	}
+
+	for element := q.pending.Back(); element != nil; element = element.Prev() {
+		other, _ := element.Value.(*model.Task)
+		if !taskOrderLess(task, other) {
+			q.pending.InsertAfter(task, element)
+			return
+		}
+	}
+	// Unreachable: the front-check above returns when the task sorts before the
+	// head, so the back-to-front scan always finds an insertion point. Kept as a
+	// defensive fallback.
+	q.pending.PushFront(task)
+}
+
+// pushPendingFront inserts a task ahead of the tasks it compares equal to,
+// otherwise keeping the taskOrderLess order pushPending maintains. A resubmitted
+// expired task has already been dispatched once, so it keeps the head position
+// among its equals that a plain list push-front used to give it.
+// Expects the queue to be locked by the caller.
+//
+// The scan runs front-to-back, the good end for a resubmitted task from an
+// older pipeline. A task sorting past the whole list is taken by a back check
+// first, on a strict less-than so equals are never reordered.
+func (q *fifo) pushPendingFront(task *model.Task) {
+	if back := q.pending.Back(); back == nil {
+		q.pending.PushBack(task)
+		return
+	} else if other, _ := back.Value.(*model.Task); taskOrderLess(other, task) {
+		q.pending.PushBack(task)
+		return
+	}
+
 	for element := q.pending.Front(); element != nil; element = element.Next() {
-		elements = append(elements, element)
-	}
-	slices.SortStableFunc(elements, func(a, b *list.Element) int {
-		taskA, _ := a.Value.(*model.Task)
-		taskB, _ := b.Value.(*model.Task)
-		switch {
-		case taskOrderLess(taskA, taskB):
-			return -1
-		case taskOrderLess(taskB, taskA):
-			return 1
-		default:
-			return 0
+		other, _ := element.Value.(*model.Task)
+		if !taskOrderLess(other, task) {
+			q.pending.InsertBefore(task, element)
+			return
 		}
-	})
-	return elements
+	}
+	// Unreachable: the back-check above returns when the task sorts after the
+	// tail, so the front-to-back scan always finds an insertion point. Kept as a
+	// defensive fallback.
+	q.pending.PushBack(task)
 }
 
 // canRunConcurrent reports whether the given task may currently start without
@@ -454,7 +509,7 @@ func (q *fifo) resubmitExpiredPipelines() {
 		if time.Now().After(taskState.deadline) {
 			log.Info().Msgf("queue: resubmitting expired task %s", taskID)
 			taskState.error = ErrTaskExpired
-			q.pending.PushFront(taskState.item)
+			q.pushPendingFront(taskState.item)
 			delete(q.running, taskID)
 			close(taskState.done)
 		}

server/queue/fifo_test.go

@@ -15,6 +15,7 @@
 package queue
 
 import (
+	"container/list"
 	"context"
 	"errors"
 	"fmt"
@@ -1620,3 +1621,162 @@ func TestFifoFairDispatch(t *testing.T) {
 		assert.Len(t, info.Running, 0)
 	})
 }
+
+func TestFifoMultiDispatchTickOrder(t *testing.T) {
+	ctx, cancel, q := setupTestQueue(t)
+	defer cancel(nil)
+
+	// A single process tick dispatches in a loop, so several tasks leave the
+	// pending list in one pass. Only the earliest-Created tasks may go, however
+	// many workers are waiting: with fewer workers than tasks, the tail of the
+	// creation order must stay pending. Tasks are pushed newest-first, and in
+	// separate batches, so a pending list that is not kept in creation order
+	// dispatches the wrong ones.
+	pushes := [][]*model.Task{
+		{{ID: "50", Created: 500}, {ID: "40", Created: 400}},
+		{{ID: "30", Created: 300}},
+		{{ID: "20", Created: 200}, {ID: "10", Created: 100}},
+	}
+	for _, batch := range pushes {
+		assert.NoError(t, q.PushAtOnce(ctx, batch))
+	}
+
+	// two workers, five pending tasks: exactly the two earliest dispatch.
+	results := make(chan *model.Task, 2)
+	for agentID := int64(1); agentID <= 2; agentID++ {
+		go func() {
+			task, _ := q.Poll(ctx, agentID, filterFnTrue)
+			results <- task
+		}()
+	}
+
+	dispatched := make([]string, 0, 2)
+	for range 2 {
+		select {
+		case task := <-results:
+			assert.NotNil(t, task)
+			dispatched = append(dispatched, task.ID)
+		case <-time.After(time.Second):
+			t.Fatal("timeout waiting for dispatched tasks")
+		}
+	}
+	assert.ElementsMatch(t, []string{"10", "20"}, dispatched,
+		"the two earliest-Created tasks must be the ones dispatched in the tick")
+
+	waitForProcess()
+	info := q.Info(ctx)
+	pending := make([]string, 0, len(info.Pending))
+	for _, task := range info.Pending {
+		pending = append(pending, task.ID)
+	}
+	assert.Equal(t, []string{"30", "40", "50"}, pending,
+		"the remaining tasks stay pending in creation order")
+
+	for _, id := range dispatched {
+		assert.NoError(t, q.Done(ctx, id, model.StatusSuccess))
+	}
+}
+
+func TestFifoPendingInsertOrder(t *testing.T) {
+	// The two insert helpers differ only in where they place a task among the
+	// ones it compares equal to: a normal push goes after them (the order the
+	// append-then-stable-sort it replaces produced), while a resubmitted expired
+	// task goes ahead of them, keeping the retry priority a list push-front used
+	// to give it. Asserted directly on the list: through the queue the two are
+	// indistinguishable whenever the comparator already separates the tasks.
+	pendingIDs := func(q *fifo) []string {
+		ids := make([]string, 0, q.pending.Len())
+		for element := q.pending.Front(); element != nil; element = element.Next() {
+			task, _ := element.Value.(*model.Task)
+			ids = append(ids, task.ID)
+		}
+		return ids
+	}
+
+	q := &fifo{pending: list.New()}
+	q.pushPending(&model.Task{ID: "b1", Created: 200})
+	q.pushPending(&model.Task{ID: "a", Created: 100})
+	q.pushPending(&model.Task{ID: "c", Created: 300})
+	q.pushPending(&model.Task{ID: "b2", Created: 200})
+	assert.Equal(t, []string{"a", "b1", "b2", "c"}, pendingIDs(q),
+		"a push is ordered by Created and lands after the tasks it ties with")
+
+	q.pushPendingFront(&model.Task{ID: "retry", Created: 200})
+	assert.Equal(t, []string{"a", "retry", "b1", "b2", "c"}, pendingIDs(q),
+		"a resubmitted task is ordered by Created but lands ahead of its ties")
+
+	// pushPendingFront's empty-list branch: its first call above went into a
+	// populated list, so cover the fresh-list case too (symmetric with the
+	// pushPending empty-list push at the top).
+	empty := &fifo{pending: list.New()}
+	empty.pushPendingFront(&model.Task{ID: "only", Created: 100})
+	assert.Equal(t, []string{"only"}, pendingIDs(empty),
+		"pushPendingFront into an empty list seeds the list")
+
+	// Both helpers must also handle the ends of the list, not just the middle.
+	q.pushPendingFront(&model.Task{ID: "first", Created: 50})
+	q.pushPending(&model.Task{ID: "last", Created: 400})
+	assert.Equal(t, []string{"first", "a", "retry", "b1", "b2", "c", "last"}, pendingIDs(q))
+}
+
+func TestFifoFilterWaitingDrainOrder(t *testing.T) {
+	// filterWaiting drains dependency-cleared tasks back into pending on every
+	// tick. They must land in creation order among the tasks already there —
+	// a task from an older pipeline goes ahead of the newer tail, which is the
+	// whole point of fair dispatch. Asserted on the list rather than through a
+	// dispatch, so the invariant has a guard of its own.
+	// A single drained task that sorts to the very front stops the cursor on the
+	// first comparison — the simplest merge case.
+	q := &fifo{pending: list.New(), waitingOnDeps: list.New(), running: map[string]*entry{}}
+	q.pending.PushBack(&model.Task{ID: "new1", Created: 300})
+	q.pending.PushBack(&model.Task{ID: "new2", Created: 400})
+	q.waitingOnDeps.PushBack(&model.Task{ID: "old", Created: 100})
+
+	q.filterWaiting()
+
+	pendingIDs := func(q *fifo) []string {
+		ids := make([]string, 0, q.pending.Len())
+		for element := q.pending.Front(); element != nil; element = element.Next() {
+			task, _ := element.Value.(*model.Task)
+			ids = append(ids, task.ID)
+		}
+		return ids
+	}
+	assert.Equal(t, []string{"old", "new1", "new2"}, pendingIDs(q),
+		"a drained task keeps its creation-order priority over newer pending tasks")
+
+	// Multiple drained tasks interleaving into the middle and tail of a
+	// multi-element pending list: exercises the forward cursor advancing across
+	// iterations and the at==nil tail-PushBack branch. waitingOnDeps must be
+	// ascending (the invariant filterWaiting's rebuild maintains).
+	q2 := &fifo{pending: list.New(), waitingOnDeps: list.New(), running: map[string]*entry{}}
+	for _, id := range []struct {
+		name    string
+		created int64
+	}{{"p100", 100}, {"p300", 300}, {"p500", 500}} {
+		q2.pending.PushBack(&model.Task{ID: id.name, Created: id.created})
+	}
+	for _, id := range []struct {
+		name    string
+		created int64
+	}{{"w200", 200}, {"w400", 400}, {"w600", 600}} {
+		q2.waitingOnDeps.PushBack(&model.Task{ID: id.name, Created: id.created})
+	}
+
+	q2.filterWaiting()
+
+	assert.Equal(t, []string{"p100", "w200", "p300", "w400", "p500", "w600"}, pendingIDs(q2),
+		"drained tasks merge into creation order, advancing the cursor through the middle and appending at the tail (w600)")
+
+	// A drained task whose Created ties a pending task lands AFTER the equal-key
+	// pending task, matching the append-then-stable-sort semantics the merge
+	// replaces (taskOrderLess is a strict less-than on Created, then Name).
+	q3 := &fifo{pending: list.New(), waitingOnDeps: list.New(), running: map[string]*entry{}}
+	q3.pending.PushBack(&model.Task{ID: "pending-200", Created: 200, Name: "a"})
+	q3.waitingOnDeps.PushBack(&model.Task{ID: "drained-200", Created: 200, Name: "z"})
+
+	q3.filterWaiting()
+
+	assert.Equal(t, []string{"pending-200", "drained-200"}, pendingIDs(q3),
+		"a drained task tying a pending task's Created lands after it")
+}

server/queue/persistent.go

@@ -58,7 +58,8 @@ func isTerminalWorkflowState(state model.StatusValue) bool {
 	}
 }
 
-// PushAtOnce pushes multiple tasks to the tail of this queue.
+// PushAtOnce persists multiple tasks and pushes them to this queue, each into
+// its creation-order position among the tasks already pending.
 func (q *persistentQueue) PushAtOnce(c context.Context, tasks []*model.Task) error {
 	// TODO: invent store.NewSession who return context including a session and make TaskInsert & TaskDelete use it
 	for _, task := range tasks {