[BCI-2151] Refactor prom reporter db API

common/txmgr/types/tx_store.go

@@ -5,6 +5,8 @@ import (
 	"math/big"
 	"time"
 
+	"gopkg.in/guregu/null.v4"
+
 	"github.com/google/uuid"
 
 	feetypes "github.com/smartcontractkit/chainlink/v2/common/fee/types"
@@ -64,6 +66,7 @@ type TransactionStore[
 	FEE feetypes.Fee,
 ] interface {
 	CountUnconfirmedTransactions(ctx context.Context, fromAddress ADDR, chainID CHAIN_ID) (count uint32, err error)
+	CountAllUnconfirmedTransactions(ctx context.Context, chainID CHAIN_ID) (count uint32, err error)
 	CountUnstartedTransactions(ctx context.Context, fromAddress ADDR, chainID CHAIN_ID) (count uint32, err error)
 	CreateTransaction(ctx context.Context, txRequest TxRequest[ADDR, TX_HASH], chainID CHAIN_ID) (tx Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], err error)
 	DeleteInProgressAttempt(ctx context.Context, attempt TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) error
@@ -79,6 +82,9 @@ type TransactionStore[
 	FindTxWithSequence(ctx context.Context, fromAddress ADDR, seq SEQ) (etx *Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], err error)
 	FindNextUnstartedTransactionFromAddress(ctx context.Context, etx *Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], fromAddress ADDR, chainID CHAIN_ID) error
 	FindTransactionsConfirmedInBlockRange(ctx context.Context, highBlockNumber, lowBlockNumber int64, chainID CHAIN_ID) (etxs []*Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], err error)
+	FindMinUnconfirmedBroadcastTime(ctx context.Context, chainID CHAIN_ID) (null.Time, error)
+	FindEarliestUnconfirmedTxBlock(ctx context.Context, chainID CHAIN_ID) (null.Int, error)
+	GetPipelineRunStats(ctx context.Context) (taskRunsQueued int, runsQueued int, err error)
 	GetTxInProgress(ctx context.Context, fromAddress ADDR) (etx *Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], err error)
 	GetInProgressTxAttempts(ctx context.Context, address ADDR, chainID CHAIN_ID) (attempts []TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], err error)
 	HasInProgressTransaction(ctx context.Context, account ADDR, chainID CHAIN_ID) (exists bool, err error)

core/chains/evm/txmgr/evm_tx_store.go

@@ -10,6 +10,8 @@ import (
 	"strings"
 	"time"
 
+	"go.uber.org/multierr"
+
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/google/uuid"
 	"github.com/jackc/pgconn"
@@ -1107,6 +1109,77 @@ ORDER BY nonce ASC
 	return etxs, pkgerrors.Wrap(err, "FindTransactionsConfirmedInBlockRange failed")
 }
 
+func (o *evmTxStore) FindMinUnconfirmedBroadcastTime(ctx context.Context, chainID *big.Int) (broadcastAt nullv4.Time, err error) {
+	var cancel context.CancelFunc
+	ctx, cancel = o.mergeContexts(ctx)
+	defer cancel()
+	qq := o.q.WithOpts(pg.WithParentCtx(ctx))
+	err = qq.Transaction(func(tx pg.Queryer) error {
+		if err = qq.QueryRowContext(ctx, `SELECT min(initial_broadcast_at) FROM evm.txes WHERE state = 'unconfirmed' AND evm_chain_id = $1`, chainID.String()).Scan(&broadcastAt); err != nil {
+			return fmt.Errorf("failed to query for unconfirmed eth_tx count: %w", err)
+		}
+		return nil
+	}, pg.OptReadOnlyTx())
+	return broadcastAt, err
+}
+
+func (o *evmTxStore) FindEarliestUnconfirmedTxBlock(ctx context.Context, chainID *big.Int) (earliestUnconfirmedTxBlock nullv4.Int, err error) {
+	var cancel context.CancelFunc
+	ctx, cancel = o.mergeContexts(ctx)
+	defer cancel()
+	qq := o.q.WithOpts(pg.WithParentCtx(ctx))
+	err = qq.Transaction(func(tx pg.Queryer) error {
+		err = qq.QueryRowContext(ctx, `
+SELECT MIN(broadcast_before_block_num) FROM evm.tx_attempts
+JOIN evm.txes ON evm.txes.id = evm.tx_attempts.eth_tx_id
+WHERE evm.txes.state = 'unconfirmed'
+AND evm_chain_id = $1`, chainID.String()).Scan(&earliestUnconfirmedTxBlock)
+		if err != nil {
+			return fmt.Errorf("failed to query for earliest unconfirmed tx block: %w", err)
+		}
+		return nil
+	}, pg.OptReadOnlyTx())
+	return earliestUnconfirmedTxBlock, err
+}
+
+func (o *evmTxStore) GetPipelineRunStats(ctx context.Context) (taskRunsQueued int, runsQueued int, err error) {
+	var cancel context.CancelFunc
+	ctx, cancel = o.mergeContexts(ctx)
+	defer cancel()
+	qq := o.q.WithOpts(pg.WithParentCtx(ctx))
+	var rows *sql.Rows
+	err = qq.Transaction(func(tx pg.Queryer) error {
+		rows, err = qq.QueryContext(ctx, `SELECT pipeline_run_id FROM pipeline_task_runs WHERE finished_at IS NULL`)
+		if err != nil {
+			return fmt.Errorf("failed to query for pipeline_run_id: %w", err)
+		}
+		if rows.Err() != nil {
+			return fmt.Errorf("failed to query for pipeline_run_id: %w", rows.Err())
+		}
+		return nil
+	}, pg.OptReadOnlyTx())
+
+	defer func() {
+		err = multierr.Combine(err, rows.Close())
+	}()
+
+	taskRunsQueued = 0
+	pipelineRunsQueuedSet := make(map[int32]struct{})
+	for rows.Next() {
+		var pipelineRunID int32
+		if err = rows.Scan(&pipelineRunID); err != nil {
+			return 0, 0, fmt.Errorf("unexpected error scanning row: %w", err)
+		}
+		taskRunsQueued++
+		pipelineRunsQueuedSet[pipelineRunID] = struct{}{}
+	}
+	if err = rows.Err(); err != nil {
+		return 0, 0, err
+	}
+	runsQueued = len(pipelineRunsQueuedSet)
+	return taskRunsQueued, runsQueued, err
+}
+
 func saveAttemptWithNewState(ctx context.Context, q pg.Queryer, logger logger.Logger, attempt TxAttempt, broadcastAt time.Time) error {
 	var dbAttempt DbEthTxAttempt
 	dbAttempt.FromTxAttempt(&attempt)
@@ -1639,6 +1712,20 @@ func (o *evmTxStore) CountUnconfirmedTransactions(ctx context.Context, fromAddre
 	return o.countTransactionsWithState(ctx, fromAddress, txmgr.TxUnconfirmed, chainID)
 }
 
+// CountAllUnconfirmedTransactions returns the number of unconfirmed transactions for all fromAddresses
+func (o *evmTxStore) CountAllUnconfirmedTransactions(ctx context.Context, chainID *big.Int) (count uint32, err error) {
+	var cancel context.CancelFunc
+	ctx, cancel = o.mergeContexts(ctx)
+	defer cancel()
+	qq := o.q.WithOpts(pg.WithParentCtx(ctx))
+	err = qq.Get(&count, `SELECT count(*) FROM evm.txes WHERE state = $1 AND evm_chain_id = $2`,
+		txmgr.TxUnconfirmed, chainID.String())
+	if err != nil {
+		return 0, fmt.Errorf("failed to CountAllUnconfirmedTransactions: %w", err)
+	}
+	return count, nil
+}
+
 // CountUnstartedTransactions returns the number of unconfirmed transactions
 func (o *evmTxStore) CountUnstartedTransactions(ctx context.Context, fromAddress common.Address, chainID *big.Int) (count uint32, err error) {
 	return o.countTransactionsWithState(ctx, fromAddress, txmgr.TxUnstarted, chainID)

core/chains/evm/txmgr/evm_tx_store_test.go

@@ -822,6 +822,90 @@ func TestORM_FindTransactionsConfirmedInBlockRange(t *testing.T) {
 	})
 }
 
+func TestORM_FindMinUnconfirmedBroadcastTime(t *testing.T) {
+	t.Parallel()
+
+	db := pgtest.NewSqlxDB(t)
+	cfg := newTestChainScopedConfig(t)
+	txStore := cltest.NewTestTxStore(t, db, cfg.Database())
+	ethKeyStore := cltest.NewKeyStore(t, db, cfg.Database()).Eth()
+	ethClient := evmtest.NewEthClientMockWithDefaultChain(t)
+	_, fromAddress := cltest.MustInsertRandomKeyReturningState(t, ethKeyStore)
+
+	t.Run("no unconfirmed eth txes", func(t *testing.T) {
+		broadcastAt, err := txStore.FindMinUnconfirmedBroadcastTime(testutils.Context(t), ethClient.ConfiguredChainID())
+		require.NoError(t, err)
+		require.False(t, broadcastAt.Valid)
+	})
+
+	t.Run("verify broadcast time", func(t *testing.T) {
+		tx := cltest.MustInsertUnconfirmedEthTx(t, txStore, 123, fromAddress)
+		broadcastAt, err := txStore.FindMinUnconfirmedBroadcastTime(testutils.Context(t), ethClient.ConfiguredChainID())
+		require.NoError(t, err)
+		require.True(t, broadcastAt.Ptr().Equal(*tx.BroadcastAt))
+	})
+}
+
+func TestORM_FindEarliestUnconfirmedTxBlock(t *testing.T) {
+	t.Parallel()
+
+	db := pgtest.NewSqlxDB(t)
+	cfg := newTestChainScopedConfig(t)
+	txStore := cltest.NewTestTxStore(t, db, cfg.Database())
+	ethKeyStore := cltest.NewKeyStore(t, db, cfg.Database()).Eth()
+	ethClient := evmtest.NewEthClientMockWithDefaultChain(t)
+	_, fromAddress := cltest.MustInsertRandomKeyReturningState(t, ethKeyStore)
+	_, fromAddress2 := cltest.MustInsertRandomKeyReturningState(t, ethKeyStore)
+
+	t.Run("no earliest unconfirmed tx block", func(t *testing.T) {
+		earliestBlock, err := txStore.FindEarliestUnconfirmedTxBlock(testutils.Context(t), ethClient.ConfiguredChainID())
+		require.NoError(t, err)
+		require.False(t, earliestBlock.Valid)
+	})
+
+	t.Run("verify earliest unconfirmed tx block", func(t *testing.T) {
+		var blockNum int64 = 2
+		tx := mustInsertConfirmedMissingReceiptEthTxWithLegacyAttempt(t, txStore, 123, blockNum, time.Now(), fromAddress)
+		_ = mustInsertConfirmedMissingReceiptEthTxWithLegacyAttempt(t, txStore, 123, blockNum, time.Now().Add(time.Minute), fromAddress2)
+		err := txStore.UpdateTxsUnconfirmed(testutils.Context(t), []int64{tx.ID})
+		require.NoError(t, err)
+
+		earliestBlock, err := txStore.FindEarliestUnconfirmedTxBlock(testutils.Context(t), ethClient.ConfiguredChainID())
+		require.NoError(t, err)
+		require.True(t, earliestBlock.Valid)
+		require.Equal(t, blockNum, earliestBlock.Int64)
+	})
+}
+
+func TestORM_GetPipelineRunStats(t *testing.T) {
+	t.Parallel()
+
+	db := pgtest.NewSqlxDB(t)
+	cfg := newTestChainScopedConfig(t)
+	txStore := cltest.NewTestTxStore(t, db, cfg.Database())
+
+	t.Run("no pipeline run tasks", func(t *testing.T) {
+		taskRunsQueued, runsQueued, err := txStore.GetPipelineRunStats(testutils.Context(t))
+		require.NoError(t, err)
+		require.Equal(t, 0, taskRunsQueued)
+		require.Equal(t, 0, runsQueued)
+	})
+
+	t.Run("queued pipeline run tasks", func(t *testing.T) {
+		pgtest.MustExec(t, db, `SET CONSTRAINTS pipeline_task_runs_pipeline_run_id_fkey DEFERRED`)
+
+		numRuns := int64(5)
+		for id := int64(0); id < numRuns; id++ {
+			cltest.MustInsertUnfinishedPipelineTaskRun(t, db, id)
+		}
+
+		taskRunsQueued, runsQueued, err := txStore.GetPipelineRunStats(testutils.Context(t))
+		require.NoError(t, err)
+		require.Equal(t, int(numRuns), taskRunsQueued)
+		require.Equal(t, int(numRuns), runsQueued)
+	})
+}
+
 func TestORM_SaveInsufficientEthAttempt(t *testing.T) {
 	t.Parallel()
 
@@ -1411,6 +1495,27 @@ func TestORM_CountUnconfirmedTransactions(t *testing.T) {
 	assert.Equal(t, int(count), 3)
 }
 
+func TestORM_CountAllUnconfirmedTransactions(t *testing.T) {
+	t.Parallel()
+
+	db := pgtest.NewSqlxDB(t)
+	cfg := configtest.NewGeneralConfig(t, nil)
+	txStore := cltest.NewTestTxStore(t, db, cfg.Database())
+	ethKeyStore := cltest.NewKeyStore(t, db, cfg.Database()).Eth()
+
+	_, fromAddress1 := cltest.MustInsertRandomKey(t, ethKeyStore)
+	_, fromAddress2 := cltest.MustInsertRandomKey(t, ethKeyStore)
+	_, fromAddress3 := cltest.MustInsertRandomKey(t, ethKeyStore)
+
+	cltest.MustInsertUnconfirmedEthTxWithBroadcastLegacyAttempt(t, txStore, 0, fromAddress1)
+	cltest.MustInsertUnconfirmedEthTxWithBroadcastLegacyAttempt(t, txStore, 1, fromAddress2)
+	cltest.MustInsertUnconfirmedEthTxWithBroadcastLegacyAttempt(t, txStore, 2, fromAddress3)
+
+	count, err := txStore.CountAllUnconfirmedTransactions(testutils.Context(t), &cltest.FixtureChainID)
+	require.NoError(t, err)
+	assert.Equal(t, int(count), 3)
+}
+
 func TestORM_CountUnstartedTransactions(t *testing.T) {
 	t.Parallel()