From ea8a6657729ea89f9d8fb0d8e8ff1d3a0276d0b0 Mon Sep 17 00:00:00 2001 From: Gijs van Dam Date: Tue, 3 Mar 2026 12:06:22 +0100 Subject: [PATCH] queue: add BackpressureQueue[T] with Random Early Drop Add a generic BackpressureQueue that uses a DropPredicate to proactively shed load before the queue is completely full. Two predicate types are provided: - DropCheckFunc: length-only drop decision (func(queueLen int) bool) - DropPredicate[T]: item-aware drop decision RandomEarlyDrop returns a DropCheckFunc since RED only considers queue depth. The AsDropPredicate helper adapts it to DropPredicate[T] for use with BackpressureQueue. In addition to the blocking Enqueue/Dequeue methods, the queue exposes TryEnqueue (non-blocking send with drop check), Len, ReceiveChan, and Close. These are needed by the actor package's BackpressureMailbox which uses BackpressureQueue as its core buffer while implementing the Mailbox interface's select-based iteration and lifecycle methods. Property-based tests using pgregory.net/rapid verify queue invariants (capacity bounds, FIFO ordering, model consistency) across randomized enqueue/dequeue sequences with RED enabled. --- queue/back_pressure.go | 266 ++++++++++++++++++++ queue/back_pressure_test.go | 488 ++++++++++++++++++++++++++++++++++++ queue/go.mod | 19 +- queue/go.sum | 23 ++ 4 files changed, 794 insertions(+), 2 deletions(-) create mode 100644 queue/back_pressure.go create mode 100644 queue/back_pressure_test.go diff --git a/queue/back_pressure.go b/queue/back_pressure.go new file mode 100644 index 000000000..570fb83c2 --- /dev/null +++ b/queue/back_pressure.go @@ -0,0 +1,266 @@ +package queue + +import ( + "context" + "errors" + "math/rand" + "sync" + "sync/atomic" + + "github.com/lightningnetwork/lnd/fn/v2" +) + +// ErrQueueClosed is returned by Enqueue/TryEnqueue when the queue has already +// been closed. +var ErrQueueClosed = errors.New("queue closed") + +// DropCheckFunc decides whether to drop an item based solely on the current +// queue depth. This is the natural return type for length-only strategies such +// as RandomEarlyDrop. +type DropCheckFunc func(queueLen int) bool + +// DropPredicate decides whether to drop an item based on the current queue +// depth and the item itself. It returns true to drop, false to enqueue. Use +// this when the drop decision depends on the item itself; for length-only +// checks prefer DropCheckFunc. +type DropPredicate[T any] func(queueLen int, item T) bool + +// AsDropPredicate adapts a length-only DropCheckFunc into a DropPredicate[T], +// ignoring the item. +func AsDropPredicate[T any](f DropCheckFunc) DropPredicate[T] { + return func(queueLen int, _ T) bool { + return f(queueLen) + } +} + +// ErrItemDropped is returned by Enqueue when the item is dropped by the +// DropPredicate. This can happen before the queue is actually full (e.g. with +// RED-style early drops). +var ErrItemDropped = errors.New("item dropped by drop predicate") + +// ErrNegativeMinThreshold is returned by RandomEarlyDrop when minThreshold +// is negative. +var ErrNegativeMinThreshold = errors.New( + "queue: minThreshold must be >= 0", +) + +// ErrInvalidThresholdOrder is returned by RandomEarlyDrop when maxThreshold +// is not strictly greater than minThreshold. +var ErrInvalidThresholdOrder = errors.New( + "queue: maxThreshold must be > minThreshold", +) + +// BackpressureQueue is a generic, fixed-capacity queue with predicate-based +// drop behavior. When full, it uses the DropPredicate to perform early drops +// (e.g., RED-style). +type BackpressureQueue[T any] struct { + ch chan T + dropPredicate DropPredicate[T] + + closed atomic.Bool + closeOnce sync.Once +} + +// NewBackpressureQueue creates a new BackpressureQueue with the given capacity +// and drop predicate. Panics if capacity <= 0 or predicate is nil. +func NewBackpressureQueue[T any](capacity int, + predicate DropPredicate[T]) *BackpressureQueue[T] { + + if capacity <= 0 { + panic("queue: NewBackpressureQueue requires capacity > 0") + } + if predicate == nil { + panic("queue: NewBackpressureQueue requires " + + "a non-nil predicate") + } + + return &BackpressureQueue[T]{ + ch: make(chan T, capacity), + dropPredicate: predicate, + } +} + +// Enqueue attempts to add an item to the queue, respecting context +// cancellation. Returns ErrItemDropped if dropped, or context error if ctx is +// done before enqueue. Otherwise, `nil` is returned on success. +func (q *BackpressureQueue[T]) Enqueue(ctx context.Context, item T) error { + if q.closed.Load() { + return ErrQueueClosed + } + + // Consult the drop predicate based on the current queue length. + // + // NOTE: There is a TOCTOU gap here — the queue length snapshot can + // become stale between this check and the channel send below if + // there are multiple concurrent writers. This is acceptable because + // RED is inherently probabilistic and approximate; a slightly + // outdated length does not compromise correctness. + if q.dropPredicate(len(q.ch), item) { + return ErrItemDropped + } + + // If the predicate decides not to drop, attempt to enqueue the item. + select { + case q.ch <- item: + return nil + + default: + // Channel is full, and the predicate decided not to drop. We + // must block until space is available or context is cancelled. + select { + case q.ch <- item: + return nil + + case <-ctx.Done(): + return ctx.Err() + } + } +} + +// TryEnqueue attempts to add an item to the queue without blocking. Returns +// true if successfully enqueued, false if the drop predicate rejected the item +// or the queue is at capacity. +func (q *BackpressureQueue[T]) TryEnqueue(item T) bool { + if q.closed.Load() { + return false + } + + if q.dropPredicate(len(q.ch), item) { + return false + } + + select { + case q.ch <- item: + return true + default: + return false + } +} + +// Dequeue retrieves the next item from the queue, blocking until available or +// context done. Returns the item or an error if ctx is done before an item is +// available. +func (q *BackpressureQueue[T]) Dequeue(ctx context.Context) fn.Result[T] { + select { + + case item, ok := <-q.ch: + if !ok { + return fn.Err[T](ErrQueueClosed) + } + return fn.Ok(item) + + case <-ctx.Done(): + return fn.Err[T](ctx.Err()) + } +} + +// Len returns the current number of items buffered in the queue. +func (q *BackpressureQueue[T]) Len() int { + return len(q.ch) +} + +// ReceiveChan returns the receive-only end of the internal channel, allowing +// callers to select on it alongside other channels (e.g., context.Done). +func (q *BackpressureQueue[T]) ReceiveChan() <-chan T { + return q.ch +} + +// Close closes the internal channel. It is safe to call multiple times; +// only the first call has any effect. After Close, no more items can be +// enqueued. Remaining items can still be received via ReceiveChan. +func (q *BackpressureQueue[T]) Close() { + q.closeOnce.Do(func() { + q.closed.Store(true) + close(q.ch) + }) +} + +// redConfig holds configuration for RandomEarlyDrop. +type redConfig struct { + // randSrc returns a float64 in [0.0, 1.0). It must be safe for + // concurrent use if the returned DropCheckFunc will be called from + // multiple goroutines. The default (math/rand.Float64) is safe since + // Go 1.20+. + randSrc func() float64 +} + +// REDOption is a functional option for configuring RandomEarlyDrop. +type REDOption func(*redConfig) + +// WithRandSource provides a custom random number source (a function that +// returns a float64 between 0.0 and 1.0). +func WithRandSource(src func() float64) REDOption { + return func(cfg *redConfig) { + cfg.randSrc = src + } +} + +// RandomEarlyDrop returns a DropCheckFunc that implements Random Early +// Detection (RED), inspired by TCP-RED queue management. +// +// RED prevents sudden buffer overflows by proactively dropping packets before +// the queue is full. It establishes two thresholds: +// +// 1. minThreshold: queue length below which no drops occur. +// 2. maxThreshold: queue length at or above which all items are dropped. +// +// Between these points, the drop probability p increases linearly: +// +// p = (queueLen - minThreshold) / (maxThreshold - minThreshold) +// +// For example, with minThreshold=15 and maxThreshold=35: +// - At queueLen=15, p=0.0 (0% drop chance) +// - At queueLen=25, p=0.5 (50% drop chance) +// - At queueLen=35, p=1.0 (100% drop chance) +// +// This smooth ramp helps avoid tail-drop spikes, smooths queue occupancy, +// and gives early back-pressure signals to senders. +func RandomEarlyDrop(minThreshold, maxThreshold int, + opts ...REDOption) (DropCheckFunc, error) { + + if minThreshold < 0 { + return nil, ErrNegativeMinThreshold + } + if maxThreshold <= minThreshold { + return nil, ErrInvalidThresholdOrder + } + + cfg := redConfig{ + randSrc: rand.Float64, + } + + for _, opt := range opts { + opt(&cfg) + } + if cfg.randSrc == nil { + cfg.randSrc = rand.Float64 + } + + // Precompute the denominator for the linear drop probability + // scaling. Since minThreshold < maxThreshold is enforced above, + // this is always positive. + denominator := float64(maxThreshold - minThreshold) + + dropFn := func(queueLen int) bool { + // If the queue is below the minimum threshold, then we never + // drop. + if queueLen < minThreshold { + return false + } + + // If the queue is at or above the maximum threshold, then we + // always drop. + if queueLen >= maxThreshold { + return true + } + + // If we're in the middle, then we implement linear scaling of + // the drop probability based on our thresholds. At this point, + // minThreshold <= queueLen < maxThreshold. + p := float64(queueLen-minThreshold) / denominator + + return cfg.randSrc() < p + } + + return dropFn, nil +} diff --git a/queue/back_pressure_test.go b/queue/back_pressure_test.go new file mode 100644 index 000000000..c122a9f03 --- /dev/null +++ b/queue/back_pressure_test.go @@ -0,0 +1,488 @@ +package queue + +import ( + "context" + "errors" + "math/rand" + "testing" + "time" + + "github.com/stretchr/testify/require" + "pgregory.net/rapid" +) + +// queueMachine is the generic state machine logic for testing +// BackpressureQueue. T must be comparable for use in assertions. +type queueMachine[T comparable] struct { + tb rapid.TB + + capacity int + + queue *BackpressureQueue[T] + + modelQueue []T + + dropPredicate DropPredicate[T] + + itemGenerator *rapid.Generator[T] +} + +// Enqueue is a state machine action. It enqueues an item and updates the model. +func (m *queueMachine[T]) Enqueue(t *rapid.T) { + item := m.itemGenerator.Draw(t, "item") + + err := m.queue.Enqueue(context.Background(), item) + + actualDrop := false + if errors.Is(err, ErrItemDropped) { + actualDrop = true + } else if err != nil { + // If Enqueue with background context returns an error other than + // ErrItemDropped, it's unexpected. + m.tb.Fatalf("Enqueue with background context returned "+ + "unexpected error: %v", err) + } + + if !actualDrop { + // If the item was not dropped, it must have been enqueued. Add + // it to the model. The modelQueue should not exceed capacity. + // This is also checked in Check(). + m.modelQueue = append(m.modelQueue, item) + } +} + +// Dequeue is a state machine action. It dequeues an item and updates the model. +func (m *queueMachine[T]) Dequeue(t *rapid.T) { + if len(m.modelQueue) == 0 { + // If the model is empty, the actual queue channel should also + // be empty. + require.Zero( + m.tb, len(m.queue.ch), "actual queue channel not "+ + "empty when model is empty", + ) + + // Attempting to dequeue from an empty queue should block. We + // verify this by trying to dequeue with a very short timeout. + ctx, cancel := context.WithTimeout( + context.Background(), 5*time.Millisecond, + ) + defer cancel() + + result := m.queue.Dequeue(ctx) + require.True( + m.tb, result.IsErr(), "dequeue "+ + "should return error on empty queue with timeout", + ) + require.ErrorIs( + m.tb, result.Err(), + context.DeadlineExceeded, "dequeue should "+ + "block on empty queue", + ) + + return + } + + // The model is not empty, so we expect to dequeue an item. + expectedItem := m.modelQueue[0] + m.modelQueue = m.modelQueue[1:] + + // Perform the dequeue operation, this should succeed. + result := m.queue.Dequeue(context.Background()) + actualItem, err := result.Unpack() + require.NoError(t, err) + require.Equal( + m.tb, expectedItem, actualItem, "dequeued item does not "+ + "match model (FIFO violation or model error)", + ) +} + +// Check is called by rapid after each action to verify invariants. +func (m *queueMachine[T]) Check(t *rapid.T) { + // Invariant 1: The length of the internal channel must not exceed + // capacity. + require.LessOrEqual( + m.tb, len(m.queue.ch), m.capacity, + "queue channel length exceeds capacity", + ) + + // Invariant 2: The length of our model queue must match the length of + // the actual queue's channel. + require.Equal( + m.tb, len(m.modelQueue), len(m.queue.ch), + "model queue length mismatch with actual queue channel length", + ) +} + +// intQueueMachine is a concrete wrapper for queueMachine[int] for rapid. +type intQueueMachine struct { + *queueMachine[int] +} + +// NewIntQueueMachine creates a new queueMachine specialized for int items. +func NewIntQueueMachine(rt *rapid.T) *intQueueMachine { + // Draw from the rapid distribution for the made params of our queue. + capacity := rapid.IntRange(1, 50).Draw(rt, "capacity") + minThreshold := rapid.IntRange( + 0, capacity-1, + ).Draw(rt, "minThreshold") + maxThreshold := rapid.IntRange( + minThreshold+1, capacity, + ).Draw(rt, "maxThreshold") + + // Draw a seed for this machine's local RNG using rapid. This makes the + // predicate's randomness part of rapid's generated test case. + machineSeed := rapid.Int64().Draw(rt, "machine_rng_seed") + localRngFixed := rand.New(rand.NewSource(machineSeed)) + + rt.Logf("NewIntQueueMachine: capacity=%d, minT=%d, maxT=%d, "+ + "machineSeed=%d", capacity, minThreshold, maxThreshold, + machineSeed) + + redCheck, err := RandomEarlyDrop( + minThreshold, maxThreshold, + WithRandSource(localRngFixed.Float64), + ) + require.NoError(rt, err) + predicate := AsDropPredicate[int](redCheck) + + q := NewBackpressureQueue(capacity, predicate) + + return &intQueueMachine{ + queueMachine: &queueMachine[int]{ + tb: rt, + capacity: capacity, + queue: q, + modelQueue: make([]int, 0, capacity), + dropPredicate: predicate, + itemGenerator: rapid.IntRange(-1000, 1000), + }, + } +} + +// Enqueue forwards the call to the generic queueMachine. +func (m *intQueueMachine) Enqueue(t *rapid.T) { m.queueMachine.Enqueue(t) } + +// Dequeue forwards the call to the generic queueMachine. +func (m *intQueueMachine) Dequeue(t *rapid.T) { m.queueMachine.Dequeue(t) } + +// Check forwards the call to the generic queueMachine. +func (m *intQueueMachine) Check(t *rapid.T) { m.queueMachine.Check(t) } + +// TestBackpressureQueueRapidInt is the main property-based test for +// BackpressureQueue using the IntQueueMachine state machine. +func TestBackpressureQueueRapidInt(t *testing.T) { + rapid.Check(t, func(rt *rapid.T) { + // Initialize the state machine instance within the property + // function. NewIntQueueMachine expects *rapid.T, which rt is. + machine := NewIntQueueMachine(rt) + + // Generate the actions map from the machine's methods. Rapid + // will randomly call the methods, and then use the `Check` + // method to verify invariants. + rt.Repeat(rapid.StateMachineActions(machine)) + }) +} + +// TestBackpressureQueueEnqueueCancellation tests that Enqueue respects context +// cancellation when it would otherwise block. +func TestBackpressureQueueEnqueueCancellation(t *testing.T) { + rapid.Check(t, func(rt *rapid.T) { + capacity := rapid.IntRange(1, 20).Draw(rt, "capacity") + + // Use a predicate that never drops when full, to force blocking + // behavior. + q := NewBackpressureQueue(capacity, + func(_ int, _ int) bool { return false }, + ) + + // Fill the queue to its capacity. The predicate always returns + // false, so no drops expected. + for i := range capacity { + err := q.Enqueue(context.Background(), i) + require.NoError( + rt, err, "enqueue failed during setup: %v", err, + ) + } + require.Equal( + rt, capacity, len(q.ch), "queue "+ + "should be full after setup", + ) + + // Attempt to enqueue one more item with an immediately cancelled + // context. + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err := q.Enqueue(ctx, 999) + require.Error( + rt, err, "enqueue should have "+ + "returned an error for cancelled context", + ) + require.ErrorIs( + rt, err, context.Canceled, + "error should be context.Canceled", + ) + + // Ensure the queue state (length) is unchanged. + require.Equal( + rt, capacity, len(q.ch), "queue length changed "+ + "after cancelled enqueue attempt", + ) + }) +} + +// TestBackpressureQueueDequeueCancellation tests that Dequeue respects context +// cancellation when the queue is empty and it would otherwise block. +func TestBackpressureQueueDequeueCancellation(t *testing.T) { + rapid.Check(t, func(rt *rapid.T) { + capacity := rapid.IntRange(1, 20).Draw(rt, "capacity") + + // The predicate doesn't matter much here as the queue will be + // empty. Use a never-drop predicate for simplicity. + q := NewBackpressureQueue(capacity, + func(_ int, _ int) bool { return false }, + ) + + require.Zero( + rt, len(q.ch), "queue should be empty initially for "+ + "Dequeue cancellation test", + ) + + // Attempt to dequeue from the empty queue with an immediately + // cancelled context. + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + result := q.Dequeue(ctx) + require.ErrorIs( + rt, result.UnwrapRightOr(nil), + context.Canceled, + "error should be context.Canceled", + ) + }) +} + +// TestBackpressureQueueComposedPredicate demonstrates testing with a composed +// predicate. This is a scenario-based test rather than a full property-based +// state machine. +func TestBackpressureQueueComposedPredicate(t *testing.T) { + capacity := 10 + minThresh, maxThresh := 3, 7 + + // Use a deterministic random source for this specific test case to + // ensure predictable behavior of RandomEarlyDrop. + const testSeed = int64(12345) + localRng := rand.New(rand.NewSource(testSeed)) + + redCheck, err := RandomEarlyDrop( + minThresh, maxThresh, WithRandSource(localRng.Float64), + ) + require.NoError(t, err) + + // Next, we'll define a custom predicate: drop items with value 42. + customValuePredicate := func(_ int, item int) bool { + return item == 42 + } + + // We'll also make a composed predicate: drop if RED says so OR if item + // is 42. + composedPredicate := func(queueLen int, item int) bool { + isRedDrop := redCheck(queueLen) + isCustomDrop := customValuePredicate(queueLen, item) + return isRedDrop || isCustomDrop + } + + q := NewBackpressureQueue(capacity, composedPredicate) + + // Scenario 1: Enqueue item 42 when queue length is between min/max + // thresholds. As we're below the max threshold, we shouldn't drop + // anything. + for i := range minThresh { + // All items aren't 42, and queue is not full enough for RED to + // drop. + err := q.Enqueue(context.Background(), i) + require.NoErrorf(t, err, "enqueue S1 setup "+ + "item %d (qLen before: %d) should not be dropped. "+ + "Predicate was redCheck(%d) || customPred(%d,%d)", + i, len(q.ch)-1, len(q.ch)-1, len(q.ch)-1, i) + + } + + currentLen := len(q.ch) + require.Equal(t, minThresh, currentLen, "queue length after S1 setup") + + // Enqueue item 42. customValuePredicate is true, so composedPredicate + // is true. Item 42 should be dropped regardless of what redCheck + // decides. + err = q.Enqueue(context.Background(), 42) + require.ErrorIs( + t, err, ErrItemDropped, + "item 42 should have been dropped by composed predicate", + ) + require.Equal( + t, currentLen, len(q.ch), "queue length should not change "+ + "after dropping 42", + ) + + // Re-create the main SUT queue with the composedPredicate. We will + // manually fill its channel to capacity to bypass Enqueue logic for + // setup. + q = NewBackpressureQueue(capacity, composedPredicate) + for i := range capacity { + q.ch <- i + } + require.Equal( + t, capacity, len(q.ch), "queue manually filled to capacity "+ + "for S2 test", + ) + + err = q.Enqueue(context.Background(), 100) + + // Expect drop because queue is full (len=capacity), so + // redCheck(capacity) is true. customValuePredicate(capacity, 100) + // is false. Thus, composedPredicate should be true. + require.ErrorIs( + t, err, ErrItemDropped, + "item 100 should be dropped (due to RED part "+ + "of composed predicate) when queue full", + ) + require.Equal( + t, capacity, len(q.ch), "queue length should not change "+ + "after dropping 100", + ) +} + +// TestBackpressureQueueTryEnqueue verifies non-blocking enqueue with drop +// predicate checks. +func TestBackpressureQueueTryEnqueue(t *testing.T) { + t.Parallel() + + const capacity = 5 + const dropThreshold = 3 + + alwaysDropAboveThreshold := DropPredicate[int]( + func(queueLen int, _ int) bool { + return queueLen >= dropThreshold + }, + ) + + q := NewBackpressureQueue(capacity, alwaysDropAboveThreshold) + + // Fill up to the drop threshold — all should succeed. + for i := range dropThreshold { + ok := q.TryEnqueue(i) + require.True(t, ok, "TryEnqueue(%d) should succeed", i) + } + + require.Equal(t, dropThreshold, q.Len()) + + // Next TryEnqueue should be dropped by predicate. + ok := q.TryEnqueue(99) + require.False(t, ok, "should be dropped at threshold") + require.Equal(t, dropThreshold, q.Len()) + + // With a never-drop predicate, fill to capacity and verify TryEnqueue + // returns false when the channel is full. + q2 := NewBackpressureQueue(capacity, + func(_ int, _ int) bool { return false }, + ) + for i := range capacity { + ok := q2.TryEnqueue(i) + require.True(t, ok, "TryEnqueue(%d) should succeed", i) + } + ok = q2.TryEnqueue(999) + require.False(t, ok, "TryEnqueue should fail when channel is full") +} + +// TestBackpressureQueueLenAndReceiveChan verifies Len and ReceiveChan. +func TestBackpressureQueueLenAndReceiveChan(t *testing.T) { + t.Parallel() + + neverDrop := DropPredicate[int](func(_ int, _ int) bool { + return false + }) + q := NewBackpressureQueue(10, neverDrop) + + require.Equal(t, 0, q.Len()) + + for i := range 3 { + require.NoError(t, q.Enqueue(context.Background(), i)) + } + require.Equal(t, 3, q.Len()) + + // ReceiveChan should yield the items. + ch := q.ReceiveChan() + val := <-ch + require.Equal(t, 0, val) + require.Equal(t, 2, q.Len()) +} + +// TestBackpressureQueueClose verifies that Close shuts down the channel. +func TestBackpressureQueueClose(t *testing.T) { + t.Parallel() + + neverDrop := DropPredicate[int](func(_ int, _ int) bool { + return false + }) + q := NewBackpressureQueue(10, neverDrop) + + for i := range 3 { + require.NoError(t, q.Enqueue(context.Background(), i)) + } + + q.Close() + + // Remaining items should still be readable. + var items []int + for v := range q.ReceiveChan() { + items = append(items, v) + } + require.Equal(t, []int{0, 1, 2}, items) +} + +// TestBackpressureQueueDoubleClose verifies that calling Close twice does not +// panic. +func TestBackpressureQueueDoubleClose(t *testing.T) { + t.Parallel() + + neverDrop := DropPredicate[int](func(_ int, _ int) bool { + return false + }) + q := NewBackpressureQueue(5, neverDrop) + + q.Close() + q.Close() // must not panic +} + +// TestBackpressureQueueEnqueueAfterClose verifies that Enqueue returns +// ErrQueueClosed after the queue has been closed. +func TestBackpressureQueueEnqueueAfterClose(t *testing.T) { + t.Parallel() + + neverDrop := DropPredicate[int](func(_ int, _ int) bool { + return false + }) + q := NewBackpressureQueue(5, neverDrop) + + require.NoError(t, q.Enqueue(context.Background(), 1)) + q.Close() + + err := q.Enqueue(context.Background(), 2) + require.ErrorIs(t, err, ErrQueueClosed) +} + +// TestBackpressureQueueTryEnqueueAfterClose verifies that TryEnqueue returns +// false after the queue has been closed. +func TestBackpressureQueueTryEnqueueAfterClose(t *testing.T) { + t.Parallel() + + neverDrop := DropPredicate[int](func(_ int, _ int) bool { + return false + }) + q := NewBackpressureQueue(5, neverDrop) + + q.Close() + + ok := q.TryEnqueue(1) + require.False(t, ok, "TryEnqueue after Close should return false") +} diff --git a/queue/go.mod b/queue/go.mod index 672610ad6..aab97704d 100644 --- a/queue/go.mod +++ b/queue/go.mod @@ -1,7 +1,22 @@ module github.com/lightningnetwork/lnd/queue -require github.com/lightningnetwork/lnd/ticker v1.0.0 +go 1.25.5 + +require ( + github.com/lightningnetwork/lnd/fn/v2 v2.0.8 + github.com/lightningnetwork/lnd/ticker v1.0.0 + github.com/stretchr/testify v1.8.1 + pgregory.net/rapid v1.2.0 +) + +require ( + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + golang.org/x/exp v0.0.0-20231226003508-02704c960a9b // indirect + golang.org/x/sync v0.7.0 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) replace github.com/lightningnetwork/lnd/ticker v1.0.0 => ../ticker -go 1.25.5 +replace github.com/lightningnetwork/lnd/fn/v2 => ../fn diff --git a/queue/go.sum b/queue/go.sum index e69de29bb..8368c6325 100644 --- a/queue/go.sum +++ b/queue/go.sum @@ -0,0 +1,23 @@ +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= +github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= +github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= +github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk= +github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= +golang.org/x/exp v0.0.0-20231226003508-02704c960a9b h1:kLiC65FbiHWFAOu+lxwNPujcsl8VYyTYYEZnsOO1WK4= +golang.org/x/exp v0.0.0-20231226003508-02704c960a9b/go.mod h1:iRJReGqOEeBhDZGkGbynYwcHlctCvnjTYIamk7uXpHI= +golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M= +golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +pgregory.net/rapid v1.2.0 h1:keKAYRcjm+e1F0oAuU5F5+YPAWcyxNNRK2wud503Gnk= +pgregory.net/rapid v1.2.0/go.mod h1:PY5XlDGj0+V1FCq0o192FdRhpKHGTRIWBgqjDBTrq04=