prophet/prophet.go

@@ -2,10 +2,18 @@ package prophet
 
 import (
 	"fmt"
+	"log/slog"
 
 	lru "github.com/hashicorp/golang-lru/v2"
 )
 
+// queueBufferSize sets the default size for incoming queues, i.e. the number
+// of futures waiting to be executed and the number of future results waiting
+// to be verified.
+// Trying to add more items to the queue than this size will drop the new
+// items.
+var queueBufferSize = 100
+
 // P is the main prophet process. Use [New] to create a new P.
 type P struct {
 	futures   *q[Future]
@@ -28,8 +36,8 @@ func New() (*P, error) {
 	}
 
 	return &P{
-		futures:       newQ[Future](),
-		proposals:     newQ[FutureResult](),
+		futures:       newQ[Future](queueBufferSize),
+		proposals:     newQ[FutureResult](queueBufferSize),
 		resultsWriter: resultsWriter,
 		votesWriter:   votesWriter,
 	}, nil
@@ -88,16 +96,18 @@ type q[T any] struct {
 	ch chan T
 }
 
-func newQ[T any]() *q[T] {
+func newQ[T any](buffer int) *q[T] {
 	return &q[T]{
-		ch: make(chan T),
+		ch: make(chan T, buffer),
 	}
 }
 
 func (q *q[T]) Add(item T) {
-	go func() {
-		q.ch <- item
-	}()
+	select {
+	case q.ch <- item:
+	default:
+		slog.Warn("q.Add: queue is full, dropped item", "item", item)
+	}
 }
 
 func (q *q[T]) Read() <-chan T {