TXM In-memory: add priority queue: STEP 1

.changeset/seven-rice-end.md

@@ -0,0 +1,5 @@
+---
+"chainlink": patch
+---
+
+adds an internal queue package which is part of refactor work

common/internal/queues/priority_heap.go

@@ -0,0 +1,92 @@
+package queues
+
+import (
+	feetypes "github.com/smartcontractkit/chainlink/v2/common/fee/types"
+	txmgrtypes "github.com/smartcontractkit/chainlink/v2/common/txmgr/types"
+	"github.com/smartcontractkit/chainlink/v2/common/types"
+)
+
+// priorityHeap is a priority queue of transactions prioritized by creation time. The oldest transaction is at the front of the queue.
+// It implements the heap interface in the container/heap package.
+type priorityHeap[
+	CHAIN_ID types.ID,
+	ADDR, TX_HASH, BLOCK_HASH types.Hashable,
+	R txmgrtypes.ChainReceipt[TX_HASH, BLOCK_HASH],
+	SEQ types.Sequence,
+	FEE feetypes.Fee,
+] struct {
+	txs       []*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]
+	idToIndex map[int64]int
+}
+
+// newPriorityHeap returns a new priorityHeap instance
+func NewPriorityHeap[
+	CHAIN_ID types.ID,
+	ADDR, TX_HASH, BLOCK_HASH types.Hashable,
+	R txmgrtypes.ChainReceipt[TX_HASH, BLOCK_HASH],
+	SEQ types.Sequence,
+	FEE feetypes.Fee,
+](capacity int) *priorityHeap[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE] {
+	if capacity <= 0 {
+		panic("priority_heap: capacity must be greater than 0")
+	}
+
+	pq := priorityHeap[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]{
+		txs:       make([]*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], 0, capacity),
+		idToIndex: make(map[int64]int),
+	}
+
+	return &pq
+}
+
+// Close clears the queue
+func (pq *priorityHeap[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Close() {
+	pq.txs = nil
+	pq.idToIndex = nil
+}
+
+// FindIndexByID returns the index of the transaction with the given ID
+func (pq *priorityHeap[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) FindIndexByID(id int64) (int, bool) {
+	i, ok := pq.idToIndex[id]
+	return i, ok
+}
+
+// Peek returns the next transaction to be processed
+func (pq *priorityHeap[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Peek() *txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE] {
+	if len(pq.txs) == 0 {
+		return nil
+	}
+	return pq.txs[0]
+}
+
+// Cap returns the capacity of the queue
+func (pq *priorityHeap[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Cap() int {
+	return cap(pq.txs)
+}
+
+// Len, Less, Swap, Push, and Pop methods implement the heap interface
+func (pq *priorityHeap[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Len() int {
+	return len(pq.txs)
+}
+func (pq *priorityHeap[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Less(i, j int) bool {
+	// We want Pop to give us the oldest, not newest, transaction based on creation time
+	return pq.txs[i].CreatedAt.Before(pq.txs[j].CreatedAt)
+}
+func (pq *priorityHeap[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Swap(i, j int) {
+	pq.txs[i], pq.txs[j] = pq.txs[j], pq.txs[i]
+	pq.idToIndex[pq.txs[i].ID] = i
+	pq.idToIndex[pq.txs[j].ID] = j
+}
+func (pq *priorityHeap[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Push(tx any) {
+	pq.txs = append(pq.txs, tx.(*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]))
+	pq.idToIndex[tx.(*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]).ID] = len(pq.txs) - 1
+}
+func (pq *priorityHeap[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Pop() any {
+	old := pq.txs
+	n := len(old)
+	tx := old[n-1]
+	old[n-1] = nil // avoid memory leak
+	pq.txs = old[0 : n-1]
+	delete(pq.idToIndex, tx.ID)
+	return tx
+}

common/internal/queues/tx_priority_queue.go

@@ -0,0 +1,133 @@
+package queues
+
+import (
+	"container/heap"
+	"sync"
+
+	feetypes "github.com/smartcontractkit/chainlink/v2/common/fee/types"
+	txmgrtypes "github.com/smartcontractkit/chainlink/v2/common/txmgr/types"
+	"github.com/smartcontractkit/chainlink/v2/common/types"
+)
+
+// TxPriorityQueue is a priority queue of transactions prioritized by creation time. The oldest transaction is at the front of the queue.
+type TxPriorityQueue[
+	CHAIN_ID types.ID,
+	ADDR, TX_HASH, BLOCK_HASH types.Hashable,
+	R txmgrtypes.ChainReceipt[TX_HASH, BLOCK_HASH],
+	SEQ types.Sequence,
+	FEE feetypes.Fee,
+] struct {
+	sync.RWMutex
+	ph *priorityHeap[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]
+}
+
+// NewTxPriorityQueue returns a new txPriorityQueue instance
+func NewTxPriorityQueue[
+	CHAIN_ID types.ID,
+	ADDR, TX_HASH, BLOCK_HASH types.Hashable,
+	R txmgrtypes.ChainReceipt[TX_HASH, BLOCK_HASH],
+	SEQ types.Sequence,
+	FEE feetypes.Fee,
+](capacity int) *TxPriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE] {
+	return &TxPriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]{
+		ph: NewPriorityHeap[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE](capacity),
+	}
+}
+
+// AddTx adds a transaction to the queue.
+// If the queue is full, the oldest transaction is removed.
+func (pq *TxPriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) AddTx(tx *txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) {
+	pq.Lock()
+
+	if pq.ph.Len() == pq.ph.Cap() {
+		heap.Pop(pq.ph)
+	}
+	heap.Push(pq.ph, tx)
+
+	pq.Unlock()
+}
+
+// RemoveNextTx removes the next transaction to be processed from the queue
+func (pq *TxPriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) RemoveNextTx() *txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE] {
+	pq.Lock()
+	defer pq.Unlock()
+
+	if pq.ph.Len() == 0 {
+		return nil
+	}
+
+	return heap.Pop(pq.ph).(*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE])
+}
+
+// RemoveTxByID removes the transaction with the given ID from the queue
+func (pq *TxPriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) RemoveTxByID(id int64) *txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE] {
+	pq.Lock()
+	defer pq.Unlock()
+
+	if pq.ph.Len() == 0 {
+		return nil
+	}
+
+	return pq._removeTxByID(id)
+}
+
+// PruneByTxIDs removes the transactions with the given IDs from the queue
+func (pq *TxPriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) PruneByTxIDs(ids []int64) []txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE] {
+	pq.Lock()
+	defer pq.Unlock()
+
+	if pq.ph.Len() == 0 {
+		return nil
+	}
+
+	removed := []txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]{}
+	for _, id := range ids {
+		if tx := pq._removeTxByID(id); tx != nil {
+			removed = append(removed, *tx)
+		}
+	}
+
+	return removed
+}
+
+// PeekNextTx returns the next transaction to be processed without removing it from the queue
+func (pq *TxPriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) PeekNextTx() *txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE] {
+	pq.RLock()
+	defer pq.RUnlock()
+
+	return pq.ph.Peek()
+}
+
+// Close clears the queue
+func (pq *TxPriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Close() {
+	pq.Lock()
+	defer pq.Unlock()
+
+	pq.ph.Close()
+}
+
+// Cap returns the capacity of the queue
+func (pq *TxPriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Cap() int {
+	pq.RLock()
+	defer pq.RUnlock()
+
+	return pq.ph.Cap()
+}
+
+// Len returns the length of the queue
+func (pq *TxPriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Len() int {
+	pq.RLock()
+	defer pq.RUnlock()
+
+	return pq.ph.Len()
+}
+
+// _removeTxByID removes the transaction with the given ID from the queue.
+// This method assumes that the caller holds the lock.
+func (pq *TxPriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) _removeTxByID(id int64) *txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE] {
+	if i, ok := pq.ph.FindIndexByID(id); ok {
+		return heap.Remove(pq.ph, i).(*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE])
+	}
+
+	return nil
+}