From: Anderson A. Giacomolli <agiacomolli@gmail.com>
Date: Fri, 6 Nov 2026 15:24:07 +0000 (-0300)
Subject: Start the builds in the order of the messages
X-Git-Url: http://git.giacomolli.uk/onechat/commitdiff_plain/ecf2974df6ab9222978c259fdbb091fc8f8f4369

Start the builds in the order of the messages

The queue starts the node whose context ends first: a leaf ends before
its message and a merge after its last message, and at a tie the merge
starts first. With leaves before all merges, a backlog of long messages
kept back the merges that a batch needs, so the view could not shrink.
The merges wait in a heap by the end of their context, so the next build
is found without a scan. The leaf window of 8 unbuilt leaves and the
limit of 8 builds stay as they are.

Issue: beb9a70
---

diff --git a/queue.go b/queue.go
index 5f2fcb2..c128d61 100644
--- a/queue.go
+++ b/queue.go
@@ -1,6 +1,7 @@
 package main
 
 import (
+	"container/heap"
 	"slices"
 )
 
@@ -18,18 +19,24 @@ const (
 // children are built and their texts do not fit in one node. So the core never
 // scans the tree for work.
 //
-// A leaf can start when fewer than leafWindow leaves before it are unbuilt.
-// Leaves start before merges, oldest first, because a turn waits for the leaves
-// of all earlier messages. Merges start in the order in which they became
-// ready. A failed build waits until the next message.
+// The builds start in the order of the messages: the node whose context ends
+// first starts first. The context of leaf i ends before message i, and the
+// context of a merge ends after its last message. At a tie the merge starts
+// first. So a backlog of leaves does not stop the merges that the view needs to
+// shrink. A leaf can start only when fewer than leafWindow leaves before it are
+// unbuilt. A failed build waits until the next message.
 type queue struct {
 	leaves  []int
-	merges  []nodeKey
+	merges  mergeHeap
 	queued  map[nodeKey]bool
 	running map[nodeKey]bool
 	failed  map[nodeKey]bool
+	retry   []nodeKey
 }
 
+// mergeHeap holds the merges that can start, by the end of their context.
+type mergeHeap []nodeKey
+
 func newQueue() *queue {
 	return &queue{
 		queued:  make(map[nodeKey]bool),
@@ -51,7 +58,7 @@ func (q *queue) AddMerge(key nodeKey) {
 	}
 
 	q.queued[key] = true
-	q.merges = append(q.merges, key)
+	heap.Push(&q.merges, key)
 }
 
 // Next returns the next node that can start, and marks it as running.
@@ -60,17 +67,26 @@ func (q *queue) Next() (nodeKey, bool) {
 		return nodeKey{}, false
 	}
 
-	for _, id := range q.leaves[:min(len(q.leaves), leafWindow)] {
-		key := nodeKey{0, id}
-		if q.startable(key) {
-			q.running[key] = true
-			return key, true
-		}
+	leaf, found := q.nextLeaf()
+	if len(q.merges) > 0 && (!found || nodeEnd(q.merges[0].level, q.merges[0].index) <= leaf.index) {
+		key := heap.Pop(&q.merges).(nodeKey)
+		q.running[key] = true
+
+		return key, true
+	}
+
+	if found {
+		q.running[leaf] = true
 	}
 
-	for _, key := range q.merges {
-		if q.startable(key) {
-			q.running[key] = true
+	return leaf, found
+}
+
+// nextLeaf returns the oldest leaf in the window that can start.
+func (q *queue) nextLeaf() (nodeKey, bool) {
+	for _, id := range q.leaves[:min(len(q.leaves), leafWindow)] {
+		key := nodeKey{0, id}
+		if !q.running[key] && !q.failed[key] {
 			return key, true
 		}
 	}
@@ -78,39 +94,66 @@ func (q *queue) Next() (nodeKey, bool) {
 	return nodeKey{}, false
 }
 
-func (q *queue) startable(key nodeKey) bool {
-	return !q.running[key] && !q.failed[key]
-}
-
 // Done removes a built node from the queue.
 func (q *queue) Done(key nodeKey) {
 	delete(q.running, key)
-	if key.level == 0 {
-		q.leaves = slices.DeleteFunc(q.leaves, func(id int) bool {
-			return id == key.index
-		})
-
+	if key.level > 0 {
+		delete(q.queued, key)
 		return
 	}
 
-	delete(q.queued, key)
-	q.merges = slices.DeleteFunc(q.merges, func(other nodeKey) bool {
-		return other == key
-	})
+	position := slices.Index(q.leaves, key.index)
+	if position >= 0 {
+		q.leaves = slices.Delete(q.leaves, position, position+1)
+	}
 }
 
 // Fail keeps a node in the queue, but it cannot start until Retry.
 func (q *queue) Fail(key nodeKey) {
 	delete(q.running, key)
+	if key.level > 0 {
+		q.retry = append(q.retry, key)
+		return
+	}
+
 	q.failed[key] = true
 }
 
 // Retry lets the failed nodes start again.
 func (q *queue) Retry() {
 	clear(q.failed)
+	for _, key := range q.retry {
+		heap.Push(&q.merges, key)
+	}
+
+	q.retry = q.retry[:0]
 }
 
 // Running returns the number of builds that run.
 func (q *queue) Running() int {
 	return len(q.running)
 }
+
+func (h mergeHeap) Len() int {
+	return len(h)
+}
+
+func (h mergeHeap) Less(a, b int) bool {
+	return nodeEnd(h[a].level, h[a].index) < nodeEnd(h[b].level, h[b].index)
+}
+
+func (h mergeHeap) Swap(a, b int) {
+	h[a], h[b] = h[b], h[a]
+}
+
+func (h *mergeHeap) Push(value any) {
+	*h = append(*h, value.(nodeKey))
+}
+
+func (h *mergeHeap) Pop() any {
+	old := *h
+	key := old[len(old)-1]
+	*h = old[:len(old)-1]
+
+	return key
+}
diff --git a/queue_test.go b/queue_test.go
new file mode 100644
index 0000000..6bafb5d
--- /dev/null
+++ b/queue_test.go
@@ -0,0 +1,49 @@
+package main
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestQueueStartsInMessageOrder(t *testing.T) {
+	queue := newQueue()
+	queue.AddLeaf(4)
+	queue.AddLeaf(5)
+	queue.AddMerge(nodeKey{1, 2})
+	queue.AddMerge(nodeKey{1, 1})
+
+	var order []nodeKey
+	for {
+		key, ready := queue.Next()
+		if !ready {
+			break
+		}
+
+		order = append(order, key)
+	}
+
+	require.Equal(t, []nodeKey{{1, 1}, {0, 4}, {0, 5}, {1, 2}}, order)
+}
+
+func TestQueueRetriesAFailedMergeAtTheNextMessage(t *testing.T) {
+	queue := newQueue()
+	queue.AddMerge(nodeKey{1, 0})
+
+	key, ready := queue.Next()
+	require.True(t, ready)
+
+	queue.Fail(key)
+	_, ready = queue.Next()
+	require.False(t, ready)
+
+	queue.Retry()
+	key, ready = queue.Next()
+	require.True(t, ready)
+	require.Equal(t, nodeKey{1, 0}, key)
+
+	queue.Done(key)
+	queue.AddMerge(nodeKey{1, 0})
+	_, ready = queue.Next()
+	require.True(t, ready)
+}