Merge branch 'develop' into buyback

CHANGELOG.md

@@ -31,7 +31,7 @@
 
 - [11453](https://github.com/vegaprotocol/vega/issues/11453) - Fix margin for fully collateralised markets.
 - [11462](https://github.com/vegaprotocol/vega/issues/11462) - Update the time weighted notional when publishing live game scores.
-
+- [11474](https://github.com/vegaprotocol/vega/issues/11474) - Fail `checkTx` downstream for delayed transactions so they don't get included in more than one block. 
 
 ## 0.77.0
 

core/processor/abci.go

@@ -202,6 +202,7 @@ type TxCache interface {
 	NewDelayedTransaction(ctx context.Context, delayed [][]byte, height uint64) []byte
 	IsDelayRequired(marketID string) bool
 	IsDelayRequiredAnyMarket() bool
+	IsTxInCache(tx []byte) bool
 }
 
 type App struct {
@@ -1584,6 +1585,12 @@ func (app *App) removeOldCheckpoints() error {
 func (app *App) OnCheckTxSpam(tx abci.Tx) tmtypes.ResponseCheckTx {
 	resp := tmtypes.ResponseCheckTx{}
 
+	if app.txCache.IsTxInCache(tx.Hash()) {
+		resp.Code = blockchain.AbciSpamError
+		resp.Data = []byte("delayed transaction already included in a block")
+		return resp
+	}
+
 	// verify proof of work and replay
 	if !app.nilPow {
 		if err := app.pow.CheckTx(tx); err != nil {

core/txcache/cache.go

@@ -17,8 +17,10 @@ package txcache
 
 import (
 	"context"
+	"encoding/hex"
 	"fmt"
 	"sort"
+	"sync"
 
 	"code.vegaprotocol.io/vega/core/nodewallets"
 	"code.vegaprotocol.io/vega/core/txn"
@@ -27,13 +29,16 @@ import (
 	"code.vegaprotocol.io/vega/libs/proto"
 	commandspb "code.vegaprotocol.io/vega/protos/vega/commands/v1"
 	snapshotpb "code.vegaprotocol.io/vega/protos/vega/snapshot/v1"
+
+	"github.com/cometbft/cometbft/crypto/tmhash"
 )
 
 func NewTxCache(commander *nodewallets.Commander) *TxCache {
 	return &TxCache{
 		commander:             commander,
 		marketToDelayRequired: map[string]bool{},
 		heightToTxs:           map[uint64][][]byte{},
+		cachedTxs:             map[string]struct{}{},
 	}
 }
 
@@ -44,6 +49,34 @@ type TxCache struct {
 	numBlocksToDelay uint64
 	// no need to include is snapshot - is updated when markets are created/updated/loaded from snapshot
 	marketToDelayRequired map[string]bool
+	// map of transactions that have been picked up for delay
+	cachedTxs map[string]struct{}
+	lock      sync.RWMutex
+}
+
+func (t *TxCache) IsTxInCache(txHash []byte) bool {
+	t.lock.RLock()
+	defer t.lock.RUnlock()
+	_, ok := t.cachedTxs[hex.EncodeToString(txHash)]
+	return ok
+}
+
+func (t *TxCache) removeHeightFromCache(height uint64) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	if txs, ok := t.heightToTxs[height]; ok {
+		for _, tx := range txs {
+			delete(t.cachedTxs, hex.EncodeToString(tmhash.Sum(tx)))
+		}
+	}
+}
+
+func (t *TxCache) addHeightToCache(txs [][]byte) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	for _, tx := range txs {
+		t.cachedTxs[hex.EncodeToString(tmhash.Sum(tx))] = struct{}{}
+	}
 }
 
 // MarketDelayRequiredUpdated is called when the market configuration is created/updated with support for
@@ -84,9 +117,11 @@ func (t *TxCache) NewDelayedTransaction(ctx context.Context, delayed [][]byte, c
 
 func (t *TxCache) SetRawTxs(rtx [][]byte, height uint64) {
 	if rtx == nil {
+		t.removeHeightFromCache(height)
 		delete(t.heightToTxs, height)
 	} else {
 		t.heightToTxs[height] = rtx
+		t.addHeightToCache(rtx)
 	}
 }
 
@@ -139,6 +174,7 @@ func (t *TxCache) LoadState(_ context.Context, p *types.Payload) ([]types.StateP
 		t.heightToTxs = map[uint64][][]byte{}
 		for _, tx := range data.TxCache.Txs {
 			t.heightToTxs[tx.Height] = tx.Tx
+			t.addHeightToCache(tx.Tx)
 		}
 		return nil, nil
 	default: