TXM In-memory: add priority queue: STEP 1

common/txmgr/priority_queue.go

@@ -0,0 +1,101 @@
+package txmgr
+
+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
+	*txmgrtypes.PriorityQueue[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,
+](maxUnstarted int) *TxPriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE] {
+	pq := TxPriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]{
+		PriorityQueue: txmgrtypes.NewPriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE](maxUnstarted),
+	}
+
+	return &pq
+}
+
+// AddTx adds a transaction to the queue
+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()
+	defer pq.Unlock()
+
+	heap.Push(pq, tx)
+}
+
+// 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()
+
+	return heap.Pop(pq).(*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 i := pq.FindIndexByID(id); i != -1 {
+		return heap.Remove(pq, i).(*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE])
+	}
+
+	return nil
+}
+
+// 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()
+
+	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.Lock()
+	defer pq.Unlock()
+
+	return pq.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.PriorityQueue.Close()
+}
+
+// Cap returns the capacity of the queue
+func (pq *TxPriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Cap() int {
+	return pq.PriorityQueue.Cap()
+}

common/txmgr/types/priority_queue.go

@@ -0,0 +1,86 @@
+package types
+
+import (
+	feetypes "github.com/smartcontractkit/chainlink/v2/common/fee/types"
+	"github.com/smartcontractkit/chainlink/v2/common/types"
+)
+
+type PriorityQueue[
+	CHAIN_ID types.ID,
+	ADDR, TX_HASH, BLOCK_HASH types.Hashable,
+	R ChainReceipt[TX_HASH, BLOCK_HASH],
+	SEQ types.Sequence,
+	FEE feetypes.Fee,
+] struct {
+	txs       []*Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]
+	idToIndex map[int64]int
+}
+
+// NewPriorityQueue returns a new PriorityQueue instance
+func NewPriorityQueue[
+	CHAIN_ID types.ID,
+	ADDR, TX_HASH, BLOCK_HASH types.Hashable,
+	R ChainReceipt[TX_HASH, BLOCK_HASH],
+	SEQ types.Sequence,
+	FEE feetypes.Fee,
+](maxUnstarted int) *PriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE] {
+	pq := PriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]{
+		txs:       make([]*Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], 0, maxUnstarted),
+		idToIndex: make(map[int64]int),
+	}
+
+	return &pq
+}
+
+// Close clears the queue
+func (pq *PriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Close() {
+	clear(pq.txs)
+	clear(pq.idToIndex)
+}
+
+// FindIndexByID returns the index of the transaction with the given ID
+func (pq *PriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) FindIndexByID(id int64) int {
+	i, ok := pq.idToIndex[id]
+	if !ok {
+		return -1
+	}
+	return i
+}
+
+// Peek returns the next transaction to be processed
+func (pq *PriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Peek() *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 *PriorityQueue[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 *PriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Len() int {
+	return len(pq.txs)
+}
+func (pq *PriorityQueue[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 *PriorityQueue[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] = j
+	pq.idToIndex[pq.txs[j].ID] = i
+}
+func (pq *PriorityQueue[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Push(tx any) {
+	pq.txs = append(pq.txs, tx.(*Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]))
+}
+func (pq *PriorityQueue[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]
+	return tx
+}