diff --git a/core/txpool/fastfeed/feed.go b/core/txpool/fastfeed/feed.go
new file mode 100644
index 0000000000..77a7ec88ba
--- /dev/null
+++ b/core/txpool/fastfeed/feed.go
@@ -0,0 +1,289 @@
+// Copyright 2024 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see .
+
+package fastfeed
+
+import (
+ "sync"
+ "sync/atomic"
+ "time"
+ "unsafe"
+
+ "errors"
+
+ "github.com/ethereum/go-ethereum/common"
+ "github.com/ethereum/go-ethereum/core/types"
+ "github.com/ethereum/go-ethereum/log"
+)
+
+const (
+ // DefaultBufferSize is the default ring buffer capacity (must be power of 2)
+ DefaultBufferSize = 16384
+
+ // MaxReaders is the maximum number of concurrent consumers
+ MaxReaders = 64
+
+ // TxEventSize is the size of a transaction event in bytes
+ TxEventSize = 184 // 32 (hash) + 20 (from) + 20 (to) + 32 (value) + 32 (gasPrice) + 8 (nonce) + 8 (gas) + 4 (type) + 8 (timestamp) + 20 (padding)
+)
+
+// TxEventType represents the type of transaction event
+type TxEventType uint8
+
+const (
+ TxEventAdded TxEventType = iota
+ TxEventRemoved
+ TxEventReplaced
+)
+
+// TxEvent is a fixed-size transaction event optimized for zero-copy access.
+// Layout is designed for CPU cache efficiency and minimal memory access.
+type TxEvent struct {
+ Hash [32]byte // Transaction hash
+ From [20]byte // Sender address
+ To [20]byte // Recipient address (0x0 for contract creation)
+ Value [32]byte // Transfer value
+ GasPrice [32]byte // Gas price or maxFeePerGas for EIP-1559
+ Nonce uint64 // Sender nonce
+ Gas uint64 // Gas limit
+ Type uint8 // Transaction type
+ EventType TxEventType // Event type (added/removed/replaced)
+ Timestamp uint64 // Event timestamp (nanoseconds)
+ _ [6]byte // Padding for alignment
+}
+
+// TxFilter defines filtering criteria for transaction events.
+type TxFilter struct {
+ // Addresses to watch (empty = all addresses)
+ Addresses map[common.Address]struct{}
+
+ // Contract methods to watch (first 4 bytes of calldata)
+ Methods map[[4]byte]struct{}
+
+ // Minimum gas price filter
+ MinGasPrice uint64
+
+ // Transaction types to include
+ Types map[uint8]struct{}
+}
+
+// Matches returns true if the transaction matches the filter.
+func (f *TxFilter) Matches(event *TxEvent) bool {
+ // Check addresses
+ if len(f.Addresses) > 0 {
+ fromAddr := common.BytesToAddress(event.From[:])
+ toAddr := common.BytesToAddress(event.To[:])
+ _, fromMatch := f.Addresses[fromAddr]
+ _, toMatch := f.Addresses[toAddr]
+ if !fromMatch && !toMatch {
+ return false
+ }
+ }
+
+ // Check transaction type
+ if len(f.Types) > 0 {
+ if _, ok := f.Types[event.Type]; !ok {
+ return false
+ }
+ }
+
+ // Check gas price
+ if f.MinGasPrice > 0 {
+ // Simple comparison of first 8 bytes as uint64
+ gasPrice := uint64(event.GasPrice[24])<<56 |
+ uint64(event.GasPrice[25])<<48 |
+ uint64(event.GasPrice[26])<<40 |
+ uint64(event.GasPrice[27])<<32 |
+ uint64(event.GasPrice[28])<<24 |
+ uint64(event.GasPrice[29])<<16 |
+ uint64(event.GasPrice[30])<<8 |
+ uint64(event.GasPrice[31])
+ if gasPrice < f.MinGasPrice {
+ return false
+ }
+ }
+
+ return true
+}
+
+// TxFastFeed is a high-performance transaction event feed using lock-free ring buffers.
+type TxFastFeed struct {
+ ring *RingBuffer
+ mu sync.RWMutex
+ filters map[int]*TxFilter
+ nextID int
+ enabled atomic.Bool
+
+ // Metrics
+ eventsPublished atomic.Uint64
+ eventsDropped atomic.Uint64
+ lastPublish atomic.Int64
+}
+
+// NewTxFastFeed creates a new fast transaction feed.
+func NewTxFastFeed() *TxFastFeed {
+ feed := &TxFastFeed{
+ ring: NewRingBuffer(DefaultBufferSize, MaxReaders),
+ filters: make(map[int]*TxFilter),
+ }
+ feed.enabled.Store(true)
+ return feed
+}
+
+// Publish publishes a transaction event to all subscribers.
+func (f *TxFastFeed) Publish(tx *types.Transaction, eventType TxEventType) {
+ if !f.enabled.Load() {
+ return
+ }
+
+ // Convert transaction to fixed-size event
+ event := f.txToEvent(tx, eventType)
+
+ // Write to ring buffer
+ eventPtr := unsafe.Pointer(&event)
+ if !f.ring.Write(eventPtr) {
+ f.eventsDropped.Add(1)
+ log.Warn("Fast feed buffer full, event dropped", "hash", tx.Hash())
+ return
+ }
+
+ f.eventsPublished.Add(1)
+ f.lastPublish.Store(time.Now().UnixNano())
+}
+
+// Subscribe creates a new subscription with optional filtering.
+func (f *TxFastFeed) Subscribe(filter *TxFilter) (*Subscription, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+
+ if f.nextID >= MaxReaders {
+ return nil, ErrTooManySubscribers
+ }
+
+ id := f.nextID
+ f.nextID++
+
+ if filter != nil {
+ f.filters[id] = filter
+ }
+
+ sub := &Subscription{
+ id: id,
+ feed: f,
+ events: make(chan *TxEvent, 256),
+ quit: make(chan struct{}),
+ }
+
+ // Reset reader position to current
+ f.ring.Reset(id)
+
+ // Start event delivery goroutine
+ go sub.deliver()
+
+ return sub, nil
+}
+
+// txToEvent converts a transaction to a fixed-size event.
+func (f *TxFastFeed) txToEvent(tx *types.Transaction, eventType TxEventType) TxEvent {
+ var event TxEvent
+
+ // Hash
+ copy(event.Hash[:], tx.Hash().Bytes())
+
+ // From (will be filled by caller if available)
+ // We don't compute sender here to avoid expensive ECDSA recovery
+
+ // To
+ if to := tx.To(); to != nil {
+ copy(event.To[:], to.Bytes())
+ }
+
+ // Value
+ if value := tx.Value(); value != nil {
+ copy(event.Value[:], value.Bytes())
+ }
+
+ // Gas price
+ if gasPrice := tx.GasPrice(); gasPrice != nil {
+ copy(event.GasPrice[:], gasPrice.Bytes())
+ }
+
+ // Other fields
+ event.Nonce = tx.Nonce()
+ event.Gas = tx.Gas()
+ event.Type = tx.Type()
+ event.EventType = eventType
+ event.Timestamp = uint64(time.Now().UnixNano())
+
+ return event
+}
+
+// PublishWithSender publishes a transaction event with a known sender.
+func (f *TxFastFeed) PublishWithSender(tx *types.Transaction, from common.Address, eventType TxEventType) {
+ if !f.enabled.Load() {
+ return
+ }
+
+ event := f.txToEvent(tx, eventType)
+ copy(event.From[:], from.Bytes())
+
+ eventPtr := unsafe.Pointer(&event)
+ if !f.ring.Write(eventPtr) {
+ f.eventsDropped.Add(1)
+ log.Warn("Fast feed buffer full, event dropped", "hash", tx.Hash())
+ return
+ }
+
+ f.eventsPublished.Add(1)
+ f.lastPublish.Store(time.Now().UnixNano())
+}
+
+// Enable enables the fast feed.
+func (f *TxFastFeed) Enable() {
+ f.enabled.Store(true)
+}
+
+// Disable disables the fast feed.
+func (f *TxFastFeed) Disable() {
+ f.enabled.Store(false)
+}
+
+// Stats returns feed statistics.
+type FeedStats struct {
+ BufferStats BufferStats
+ EventsPublished uint64
+ EventsDropped uint64
+ LastPublishNs int64
+ Subscribers int
+}
+
+// Stats returns current feed statistics.
+func (f *TxFastFeed) Stats() FeedStats {
+ f.mu.RLock()
+ subscribers := len(f.filters)
+ f.mu.RUnlock()
+
+ return FeedStats{
+ BufferStats: f.ring.Stats(),
+ EventsPublished: f.eventsPublished.Load(),
+ EventsDropped: f.eventsDropped.Load(),
+ LastPublishNs: f.lastPublish.Load(),
+ Subscribers: subscribers,
+ }
+}
+
+var ErrTooManySubscribers = errors.New("too many subscribers")
+
diff --git a/core/txpool/fastfeed/ringbuffer.go b/core/txpool/fastfeed/ringbuffer.go
new file mode 100644
index 0000000000..1f1e82dcf2
--- /dev/null
+++ b/core/txpool/fastfeed/ringbuffer.go
@@ -0,0 +1,215 @@
+// Copyright 2024 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see .
+
+package fastfeed
+
+import (
+ "sync/atomic"
+ "unsafe"
+)
+
+// RingBuffer is a lock-free single-producer, multiple-consumer ring buffer
+// optimized for low-latency transaction propagation.
+type RingBuffer struct {
+ buffer []unsafe.Pointer
+ capacity uint64
+ mask uint64
+
+ // Writer position (single producer)
+ writePos atomic.Uint64
+
+ // Padding to prevent false sharing
+ _ [56]byte
+
+ // Reader positions (multiple consumers)
+ // Each consumer maintains its own read position
+ readPositions []atomic.Uint64
+ maxReaders int
+}
+
+// NewRingBuffer creates a new ring buffer with the given capacity.
+// Capacity must be a power of 2 for efficient masking.
+func NewRingBuffer(capacity int, maxReaders int) *RingBuffer {
+ if capacity&(capacity-1) != 0 {
+ panic("capacity must be a power of 2")
+ }
+
+ rb := &RingBuffer{
+ buffer: make([]unsafe.Pointer, capacity),
+ capacity: uint64(capacity),
+ mask: uint64(capacity - 1),
+ maxReaders: maxReaders,
+ readPositions: make([]atomic.Uint64, maxReaders),
+ }
+
+ return rb
+}
+
+// Write adds a new entry to the ring buffer.
+// Returns false if the buffer is full (slowest reader is too far behind).
+func (rb *RingBuffer) Write(data unsafe.Pointer) bool {
+ writePos := rb.writePos.Load()
+
+ // Check if we would overwrite unread data
+ minReadPos := rb.getMinReadPos()
+ if writePos-minReadPos >= rb.capacity {
+ // Buffer is full, would overwrite unread data
+ return false
+ }
+
+ // Write data
+ idx := writePos & rb.mask
+ atomic.StorePointer(&rb.buffer[idx], data)
+
+ // Advance write position
+ rb.writePos.Store(writePos + 1)
+
+ return true
+}
+
+// Read reads the next entry for the given reader ID.
+// Returns nil if no new data is available.
+func (rb *RingBuffer) Read(readerID int) unsafe.Pointer {
+ if readerID >= rb.maxReaders {
+ return nil
+ }
+
+ readPos := rb.readPositions[readerID].Load()
+ writePos := rb.writePos.Load()
+
+ // Check if data is available
+ if readPos >= writePos {
+ return nil
+ }
+
+ // Check if data hasn't been overwritten
+ if writePos-readPos > rb.capacity {
+ // Data was overwritten, skip to oldest available
+ readPos = writePos - rb.capacity
+ rb.readPositions[readerID].Store(readPos)
+ }
+
+ // Read data
+ idx := readPos & rb.mask
+ data := atomic.LoadPointer(&rb.buffer[idx])
+
+ // Advance read position
+ rb.readPositions[readerID].Store(readPos + 1)
+
+ return data
+}
+
+// Peek reads the next entry without advancing the read position.
+func (rb *RingBuffer) Peek(readerID int) unsafe.Pointer {
+ if readerID >= rb.maxReaders {
+ return nil
+ }
+
+ readPos := rb.readPositions[readerID].Load()
+ writePos := rb.writePos.Load()
+
+ if readPos >= writePos {
+ return nil
+ }
+
+ if writePos-readPos > rb.capacity {
+ return nil
+ }
+
+ idx := readPos & rb.mask
+ return atomic.LoadPointer(&rb.buffer[idx])
+}
+
+// Available returns the number of entries available to read for a given reader.
+func (rb *RingBuffer) Available(readerID int) int {
+ if readerID >= rb.maxReaders {
+ return 0
+ }
+
+ readPos := rb.readPositions[readerID].Load()
+ writePos := rb.writePos.Load()
+
+ if readPos >= writePos {
+ return 0
+ }
+
+ available := writePos - readPos
+ if available > rb.capacity {
+ available = rb.capacity
+ }
+
+ return int(available)
+}
+
+// getMinReadPos returns the minimum read position across all readers.
+func (rb *RingBuffer) getMinReadPos() uint64 {
+ min := rb.writePos.Load()
+
+ for i := 0; i < rb.maxReaders; i++ {
+ pos := rb.readPositions[i].Load()
+ if pos < min {
+ min = pos
+ }
+ }
+
+ return min
+}
+
+// Reset resets a reader's position to the current write position.
+// Useful for catching up when a reader falls too far behind.
+func (rb *RingBuffer) Reset(readerID int) {
+ if readerID < rb.maxReaders {
+ rb.readPositions[readerID].Store(rb.writePos.Load())
+ }
+}
+
+// Stats returns statistics about the ring buffer state.
+type BufferStats struct {
+ Capacity int
+ WritePosition uint64
+ ReadPositions []uint64
+ MinReadPos uint64
+ MaxLag uint64
+}
+
+// Stats returns current buffer statistics.
+func (rb *RingBuffer) Stats() BufferStats {
+ writePos := rb.writePos.Load()
+ readPositions := make([]uint64, rb.maxReaders)
+ minReadPos := writePos
+
+ for i := 0; i < rb.maxReaders; i++ {
+ pos := rb.readPositions[i].Load()
+ readPositions[i] = pos
+ if pos < minReadPos {
+ minReadPos = pos
+ }
+ }
+
+ maxLag := uint64(0)
+ if writePos > minReadPos {
+ maxLag = writePos - minReadPos
+ }
+
+ return BufferStats{
+ Capacity: int(rb.capacity),
+ WritePosition: writePos,
+ ReadPositions: readPositions,
+ MinReadPos: minReadPos,
+ MaxLag: maxLag,
+ }
+}
+
diff --git a/core/txpool/fastfeed/subscription.go b/core/txpool/fastfeed/subscription.go
new file mode 100644
index 0000000000..8a333af7c9
--- /dev/null
+++ b/core/txpool/fastfeed/subscription.go
@@ -0,0 +1,97 @@
+// Copyright 2024 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see .
+
+package fastfeed
+
+import (
+ "errors"
+ "sync"
+)
+
+var (
+ ErrSubscriptionClosed = errors.New("subscription closed")
+)
+
+// Subscription represents a subscription to the transaction fast feed.
+type Subscription struct {
+ id int
+ feed *TxFastFeed
+ events chan *TxEvent
+ quit chan struct{}
+ once sync.Once
+}
+
+// Events returns the channel that delivers transaction events.
+func (s *Subscription) Events() <-chan *TxEvent {
+ return s.events
+}
+
+// Unsubscribe unsubscribes from the feed and releases resources.
+func (s *Subscription) Unsubscribe() {
+ s.once.Do(func() {
+ close(s.quit)
+ close(s.events)
+
+ // Remove filter
+ s.feed.mu.Lock()
+ delete(s.feed.filters, s.id)
+ s.feed.mu.Unlock()
+ })
+}
+
+// deliver reads events from the ring buffer and delivers them to the subscription channel.
+func (s *Subscription) deliver() {
+ for {
+ select {
+ case <-s.quit:
+ return
+ default:
+ }
+
+ // Try to read from ring buffer
+ eventPtr := s.feed.ring.Read(s.id)
+ if eventPtr == nil {
+ // No data available, yield
+ continue
+ }
+
+ // Convert pointer to event
+ event := (*TxEvent)(eventPtr)
+
+ // Apply filter if set
+ s.feed.mu.RLock()
+ filter, hasFilter := s.feed.filters[s.id]
+ s.feed.mu.RUnlock()
+
+ if hasFilter && !filter.Matches(event) {
+ continue
+ }
+
+ // Copy event to avoid data races
+ eventCopy := *event
+
+ // Try to deliver to channel
+ select {
+ case s.events <- &eventCopy:
+ case <-s.quit:
+ return
+ default:
+ // Channel full, skip this event
+ // In production, might want to track skipped events
+ }
+ }
+}
+
diff --git a/core/txpool/txpool.go b/core/txpool/txpool.go
index 437861efca..1bf297fe2a 100644
--- a/core/txpool/txpool.go
+++ b/core/txpool/txpool.go
@@ -25,6 +25,7 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/state"
+ "github.com/ethereum/go-ethereum/core/txpool/fastfeed"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/log"
@@ -74,6 +75,9 @@ type TxPool struct {
term chan struct{} // Termination channel to detect a closed pool
sync chan chan error // Testing / simulator channel to block until internal reset is done
+
+ // Fast feed for low-latency transaction propagation
+ fastFeed *fastfeed.TxFastFeed
}
// New creates a new transaction pool to gather, sort and filter inbound
@@ -101,6 +105,7 @@ func New(gasTip uint64, chain BlockChain, subpools []SubPool) (*TxPool, error) {
quit: make(chan chan error),
term: make(chan struct{}),
sync: make(chan chan error),
+ fastFeed: fastfeed.NewTxFastFeed(),
}
reserver := NewReservationTracker()
for i, subpool := range subpools {