Skip to content

internal/ethapi: unify relay endpoints and rate limit tx submissions #2203

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
manav2401 wants to merge 3 commits into
base: master
Choose a base branch
from
Open

internal/ethapi: unify relay endpoints and rate limit tx submissions #2203

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
308827a
internal/ethapi: unify preconf and private tx endpoints
manav2401 Apr 29, 2026
1f87562
limit concurrent tasks
manav2401 Apr 29, 2026
cf62604
fix tests
manav2401 Apr 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 36 additions & 9 deletions internal/ethapi/api.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2171,7 +2171,7 @@

// SendRawTransaction will add the signed transaction to the transaction pool.
// The sender is responsible for signing the transaction and using the correct nonce.
func (api *TransactionAPI) SendRawTransaction(ctx context.Context, input hexutil.Bytes) (common.Hash, error) {

Check failure on line 2174 in internal/ethapi/api.go

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this method to reduce its Cognitive Complexity from 19 to the 15 allowed. 


See more on https://sonarcloud.io/project/issues?id=0xPolygon_bor&issues=AZ3ZSgcmoRMsEZyP1yMX&open=AZ3ZSgcmoRMsEZyP1yMX&pullRequest=2203
tx := new(types.Transaction)
if err := tx.UnmarshalBinary(input); err != nil {
return common.Hash{}, err
Expand All @@ -2189,13 +2189,37 @@
}
}

isPrivateTxEnabled := api.b.PrivateTxEnabled()
if isPrivateTxEnabled {
// Track the tx hash to ensure it is not gossiped in public
api.b.RecordPrivateTx(tx.Hash())
}

hash, err := SubmitTransaction(ctx, api.b, tx)
if err != nil {
// Purge tx from private tx tracker if submission failed. Don't purge
// if `ErrAlreadyKnown` is being returned.
if isPrivateTxEnabled && !errors.Is(err, txpool.ErrAlreadyKnown) {
api.b.PurgePrivateTx(tx.Hash())
}
return hash, err
}

// If preconf is enabled, submit tx directly to BP
// If preconf is enabled, submit it to BPs
if api.b.PreconfEnabled() {
// Preconf processing mostly happens in background so don't float the error back to user
if err := api.b.SubmitTxForPreconf(tx); err != nil {
log.Error("Transaction accepted locally but submission for preconf failed", "err", err)
log.Error("Transaction accepted locally but submission for preconf failed", "hash", tx.Hash(), "err", err)
}
}

// Submit the private transaction directly to BPs
if isPrivateTxEnabled {
// Return an error here to inform user that private tx submission failed as it is critical.
// Note that it will be retried in background.
if err := api.b.SubmitPrivateTx(tx); err != nil {
log.Error("Private tx accepted locally but submission to bp failed", "hash", tx.Hash(), "err", err)
return tx.Hash(), fmt.Errorf("tx accepted locally, submission failed. reason: %w", err)
}
}

Expand Down Expand Up @@ -2300,7 +2324,8 @@
}

// SendRawTransactionForPreconf will accept a preconf transaction from relay if enabled. It will
// offer a soft inclusion confirmation if the transaction is accepted into the pending pool.
// offer a soft inclusion confirmation if the transaction is accepted into the pending pool. Note
// that this is an internal endpoint used by relay to submit transactions to block producers.
func (api *TransactionAPI) SendRawTransactionForPreconf(ctx context.Context, input hexutil.Bytes) (map[string]interface{}, error) {
if !api.b.AcceptPreconfTxs() {
return nil, errors.New("preconf transactions are not accepted on this node")
Expand Down Expand Up @@ -2355,10 +2380,12 @@
api.b.RecordPrivateTx(tx.Hash())

hash, err := SubmitTransaction(ctx, api.b, tx)
// Purge tx from private tx tracker if submission failed. Don't purge
// if `ErrAlreadyKnown` is being returned.
if err != nil && !errors.Is(err, txpool.ErrAlreadyKnown) {
api.b.PurgePrivateTx(tx.Hash())
if err != nil {
// Purge tx from private tx tracker if submission failed. Don't purge
// if `ErrAlreadyKnown` is being returned.
if !errors.Is(err, txpool.ErrAlreadyKnown) {
api.b.PurgePrivateTx(tx.Hash())
}
return hash, err
}

Expand All @@ -2367,8 +2394,8 @@
// Return an error here to inform user that private tx submission failed as it is critical.
// Note that it will be retried in background.
if err := api.b.SubmitPrivateTx(tx); err != nil {
log.Error("Private tx accepted locally but submission to bp failed", "err", err)
return tx.Hash(), fmt.Errorf("private tx accepted locally, submission failed. reason: %w", err)
log.Error("Private tx accepted locally but submission to bp failed", "hash", tx.Hash(), "err", err)
return tx.Hash(), fmt.Errorf("tx accepted locally, submission failed. reason: %w", err)
}
}

Expand Down
214 changes: 203 additions & 11 deletions internal/ethapi/api_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4802,7 +4802,7 @@ func TestSendRawTransactionPrivate(t *testing.T) {

hash, err := api.SendRawTransactionPrivate(context.Background(), raw)
require.Error(t, err)
require.Contains(t, err.Error(), "private tx accepted locally, submission failed")
require.Contains(t, err.Error(), "tx accepted locally, submission failed")
require.Contains(t, err.Error(), "relay down")
require.Equal(t, tx.Hash(), hash, "hash should be returned even on SubmitPrivateTx failure")
})
Expand Down Expand Up @@ -4942,24 +4942,28 @@ func TestTxPoolAPI_TxStatus(t *testing.T) {
func TestSendRawTransaction_PreconfPath(t *testing.T) {
t.Parallel()

t.Run("preconf disabled, SubmitTxForPreconf not called", func(t *testing.T) {
t.Run("neither flag enabled, no relay calls", func(t *testing.T) {
t.Parallel()
genesis := &core.Genesis{Config: params.TestChainConfig, Alloc: types.GenesisAlloc{}}
b := newTestBackend(t, 0, genesis, ethash.NewFaker(), nil)
// preconfEnabled defaults to false
// preconfEnabled and privateTxEnabled both default to false

var preconfCount atomic.Int32
b.submitTxForPreconfFn = func(tx *types.Transaction) error {
preconfCount.Add(1)
return nil
}
var preconfCount, submitPrivateCount, recordCount, purgeCount atomic.Int32
b.submitTxForPreconfFn = func(*types.Transaction) error { preconfCount.Add(1); return nil }
b.submitPrivateTxFn = func(*types.Transaction) error { submitPrivateCount.Add(1); return nil }
b.recordPrivateTxFn = func(common.Hash) { recordCount.Add(1) }
b.purgePrivateTxFn = func(common.Hash) { purgeCount.Add(1) }

api := NewTransactionAPI(b, new(AddrLocker))
raw, _ := makeSelfSignedRaw(t, api, b.acc.Address)
raw, tx := makeSelfSignedRaw(t, api, b.acc.Address)

_, err := api.SendRawTransaction(context.Background(), raw)
require.NoError(t, err)
hash, err := api.SendRawTransaction(context.Background(), raw)
require.NoError(t, err, "endpoint must work for non-relay nodes")
require.Equal(t, tx.Hash(), hash)
require.Equal(t, int32(0), preconfCount.Load(), "SubmitTxForPreconf should NOT be called when preconf is disabled")
require.Equal(t, int32(0), submitPrivateCount.Load(), "SubmitPrivateTx should NOT be called when private is disabled")
require.Equal(t, int32(0), recordCount.Load(), "RecordPrivateTx should NOT be called when private is disabled")
require.Equal(t, int32(0), purgeCount.Load(), "PurgePrivateTx should NOT be called when private is disabled")
})

t.Run("preconf enabled, SubmitTxForPreconf called", func(t *testing.T) {
Expand Down Expand Up @@ -5003,6 +5007,194 @@ func TestSendRawTransaction_PreconfPath(t *testing.T) {
require.Equal(t, tx.Hash(), hash)
})
}

func TestSendRawTransaction_Unified(t *testing.T) {
t.Parallel()

t.Run("private only, success path", func(t *testing.T) {
t.Parallel()
genesis := &core.Genesis{Config: params.TestChainConfig, Alloc: types.GenesisAlloc{}}
b := newTestBackend(t, 0, genesis, ethash.NewFaker(), nil)
b.privateTxEnabled = true

var recordCount, submitPrivateCount, preconfCount atomic.Int32
b.recordPrivateTxFn = func(common.Hash) { recordCount.Add(1) }
b.submitPrivateTxFn = func(*types.Transaction) error { submitPrivateCount.Add(1); return nil }
b.submitTxForPreconfFn = func(*types.Transaction) error { preconfCount.Add(1); return nil }

api := NewTransactionAPI(b, new(AddrLocker))
raw, tx := makeSelfSignedRaw(t, api, b.acc.Address)

hash, err := api.SendRawTransaction(context.Background(), raw)
require.NoError(t, err)
require.Equal(t, tx.Hash(), hash)
require.Equal(t, int32(1), recordCount.Load(), "RecordPrivateTx should be called once")
require.Equal(t, int32(1), submitPrivateCount.Load(), "SubmitPrivateTx should be called once")
require.Equal(t, int32(0), preconfCount.Load(), "SubmitTxForPreconf should not be called when preconf is disabled")
})

t.Run("private only, SubmitPrivateTx fails returns wrapped error", func(t *testing.T) {
t.Parallel()
genesis := &core.Genesis{Config: params.TestChainConfig, Alloc: types.GenesisAlloc{}}
b := newTestBackend(t, 0, genesis, ethash.NewFaker(), nil)
b.privateTxEnabled = true
b.submitPrivateTxFn = func(*types.Transaction) error { return errors.New("relay down") }

api := NewTransactionAPI(b, new(AddrLocker))
raw, tx := makeSelfSignedRaw(t, api, b.acc.Address)

hash, err := api.SendRawTransaction(context.Background(), raw)
require.Error(t, err)
require.Contains(t, err.Error(), "tx accepted locally, submission failed")
require.Contains(t, err.Error(), "relay down")
require.Equal(t, tx.Hash(), hash, "hash should be returned even on SubmitPrivateTx failure")
})

t.Run("private only, SubmitTransaction fails (non-AlreadyKnown) purges record", func(t *testing.T) {
t.Parallel()
genesis := &core.Genesis{Config: params.TestChainConfig, Alloc: types.GenesisAlloc{}}
b := newTestBackend(t, 0, genesis, ethash.NewFaker(), nil)
b.privateTxEnabled = true
b.sendTxErr = errors.New("pool full")

var recordCount, purgeCount, submitPrivateCount, preconfCount atomic.Int32
b.recordPrivateTxFn = func(common.Hash) { recordCount.Add(1) }
b.purgePrivateTxFn = func(common.Hash) { purgeCount.Add(1) }
b.submitPrivateTxFn = func(*types.Transaction) error { submitPrivateCount.Add(1); return nil }
b.submitTxForPreconfFn = func(*types.Transaction) error { preconfCount.Add(1); return nil }

api := NewTransactionAPI(b, new(AddrLocker))
raw, _ := makeSelfSignedRaw(t, api, b.acc.Address)

_, err := api.SendRawTransaction(context.Background(), raw)
require.Error(t, err)
require.Contains(t, err.Error(), "pool full")
// Record-before-submit invariant: record count == 1 even though
// SendTx failed proves RecordPrivateTx ran before SubmitTransaction.
require.Equal(t, int32(1), recordCount.Load(), "RecordPrivateTx should be called before SubmitTransaction")
require.Equal(t, int32(1), purgeCount.Load(), "PurgePrivateTx should be called on non-AlreadyKnown failure")
require.Equal(t, int32(0), submitPrivateCount.Load(), "SubmitPrivateTx should not be called when local pool rejects")
require.Equal(t, int32(0), preconfCount.Load(), "SubmitTxForPreconf should not be called when local pool rejects")
})

t.Run("private only, ErrAlreadyKnown short-circuits BP submit no purge", func(t *testing.T) {
t.Parallel()
genesis := &core.Genesis{Config: params.TestChainConfig, Alloc: types.GenesisAlloc{}}
b := newTestBackend(t, 0, genesis, ethash.NewFaker(), nil)
b.privateTxEnabled = true
b.sendTxErr = txpool.ErrAlreadyKnown

var recordCount, purgeCount, submitPrivateCount, preconfCount atomic.Int32
b.recordPrivateTxFn = func(common.Hash) { recordCount.Add(1) }
b.purgePrivateTxFn = func(common.Hash) { purgeCount.Add(1) }
b.submitPrivateTxFn = func(*types.Transaction) error { submitPrivateCount.Add(1); return nil }
b.submitTxForPreconfFn = func(*types.Transaction) error { preconfCount.Add(1); return nil }

api := NewTransactionAPI(b, new(AddrLocker))
raw, _ := makeSelfSignedRaw(t, api, b.acc.Address)

_, err := api.SendRawTransaction(context.Background(), raw)
require.ErrorIs(t, err, txpool.ErrAlreadyKnown)
require.Equal(t, int32(1), recordCount.Load(), "RecordPrivateTx should still be called")
require.Equal(t, int32(0), purgeCount.Load(), "PurgePrivateTx should NOT be called for ErrAlreadyKnown")
require.Equal(t, int32(0), submitPrivateCount.Load(), "SubmitPrivateTx should NOT be called on ErrAlreadyKnown")
require.Equal(t, int32(0), preconfCount.Load(), "SubmitTxForPreconf should NOT be called on ErrAlreadyKnown")
})

t.Run("both flags on, both submits called", func(t *testing.T) {
t.Parallel()
genesis := &core.Genesis{Config: params.TestChainConfig, Alloc: types.GenesisAlloc{}}
b := newTestBackend(t, 0, genesis, ethash.NewFaker(), nil)
b.preconfEnabled = true
b.privateTxEnabled = true

var preconfCount, submitPrivateCount atomic.Int32
b.submitTxForPreconfFn = func(*types.Transaction) error { preconfCount.Add(1); return nil }
b.submitPrivateTxFn = func(*types.Transaction) error { submitPrivateCount.Add(1); return nil }

api := NewTransactionAPI(b, new(AddrLocker))
raw, tx := makeSelfSignedRaw(t, api, b.acc.Address)

hash, err := api.SendRawTransaction(context.Background(), raw)
require.NoError(t, err)
require.Equal(t, tx.Hash(), hash)
require.Equal(t, int32(1), preconfCount.Load(), "SubmitTxForPreconf should be called once")
require.Equal(t, int32(1), submitPrivateCount.Load(), "SubmitPrivateTx should be called once")
})

t.Run("both flags on, preconf fails private succeeds no error to user", func(t *testing.T) {
t.Parallel()
genesis := &core.Genesis{Config: params.TestChainConfig, Alloc: types.GenesisAlloc{}}
b := newTestBackend(t, 0, genesis, ethash.NewFaker(), nil)
b.preconfEnabled = true
b.privateTxEnabled = true

var preconfCount, submitPrivateCount atomic.Int32
b.submitTxForPreconfFn = func(*types.Transaction) error {
preconfCount.Add(1)
return errors.New("preconf relay down")
}
b.submitPrivateTxFn = func(*types.Transaction) error { submitPrivateCount.Add(1); return nil }

api := NewTransactionAPI(b, new(AddrLocker))
raw, tx := makeSelfSignedRaw(t, api, b.acc.Address)

hash, err := api.SendRawTransaction(context.Background(), raw)
require.NoError(t, err, "preconf failures should be swallowed when private succeeds")
require.Equal(t, tx.Hash(), hash)
require.Equal(t, int32(1), preconfCount.Load())
require.Equal(t, int32(1), submitPrivateCount.Load(), "private submit should still run after preconf failure")
})

t.Run("both flags on, preconf succeeds private fails returns wrapped error", func(t *testing.T) {
t.Parallel()
genesis := &core.Genesis{Config: params.TestChainConfig, Alloc: types.GenesisAlloc{}}
b := newTestBackend(t, 0, genesis, ethash.NewFaker(), nil)
b.preconfEnabled = true
b.privateTxEnabled = true

var preconfCount, submitPrivateCount atomic.Int32
b.submitTxForPreconfFn = func(*types.Transaction) error { preconfCount.Add(1); return nil }
b.submitPrivateTxFn = func(*types.Transaction) error {
submitPrivateCount.Add(1)
return errors.New("private relay down")
}

api := NewTransactionAPI(b, new(AddrLocker))
raw, tx := makeSelfSignedRaw(t, api, b.acc.Address)

hash, err := api.SendRawTransaction(context.Background(), raw)
require.Error(t, err)
require.Contains(t, err.Error(), "tx accepted locally, submission failed")
require.Contains(t, err.Error(), "private relay down")
require.Equal(t, tx.Hash(), hash)
require.Equal(t, int32(1), preconfCount.Load(), "preconf submit should run before private submit")
require.Equal(t, int32(1), submitPrivateCount.Load())
})

t.Run("both flags on, ErrAlreadyKnown short-circuits both BP submits", func(t *testing.T) {
t.Parallel()
genesis := &core.Genesis{Config: params.TestChainConfig, Alloc: types.GenesisAlloc{}}
b := newTestBackend(t, 0, genesis, ethash.NewFaker(), nil)
b.preconfEnabled = true
b.privateTxEnabled = true
b.sendTxErr = txpool.ErrAlreadyKnown

var preconfCount, submitPrivateCount, purgeCount atomic.Int32
b.submitTxForPreconfFn = func(*types.Transaction) error { preconfCount.Add(1); return nil }
b.submitPrivateTxFn = func(*types.Transaction) error { submitPrivateCount.Add(1); return nil }
b.purgePrivateTxFn = func(common.Hash) { purgeCount.Add(1) }

api := NewTransactionAPI(b, new(AddrLocker))
raw, _ := makeSelfSignedRaw(t, api, b.acc.Address)

_, err := api.SendRawTransaction(context.Background(), raw)
require.ErrorIs(t, err, txpool.ErrAlreadyKnown)
require.Equal(t, int32(0), preconfCount.Load(), "SubmitTxForPreconf should NOT be called on ErrAlreadyKnown")
require.Equal(t, int32(0), submitPrivateCount.Load(), "SubmitPrivateTx should NOT be called on ErrAlreadyKnown")
require.Equal(t, int32(0), purgeCount.Load(), "PurgePrivateTx should NOT be called for ErrAlreadyKnown")
})
}
func (b *testBackend) ProtocolVersion() uint {
return 69 // ETH69
}
Expand Down
Loading