pipo tx parse

README.md

@@ -15,6 +15,8 @@ go get github.com/samlior/libsam
 | fra    | fra1.shreder.xyz:9991 |
 | ams    | ams1.shreder.xyz:9991 |
 | ewr    | ny1.shreder.xyz:9991  |
+| uk     | lon.shreder.xyz:9991  |
+| jp     | tyo.shreder.xyz:9991  |
 
 ### Usage
 
@@ -105,6 +107,13 @@ See [example](./cmd/shreder/main.go).
     "keepAliveUrl": "http://germany.solana.dex.blxrbdn.com/ping",
     "tips": "0.001",
     "rateLimit": 0
+},
+{
+    "name": "soyas",
+    "sendTxUrl": "fra.landing.soyas.xyz:9000",
+    "sendBundleUrl": "",
+    "tips": "0.001",
+    "rateLimit": 0
 }
 ```
 
@@ -191,6 +200,13 @@ See [example](./cmd/shreder/main.go).
     "keepAliveUrl": "http://amsterdam.solana.dex.blxrbdn.com/api/v2/ping",
     "tips": "0.001",
     "rateLimit": 0
+},
+{
+    "name": "soyas",
+    "sendTxUrl": "ams.landing.soyas.xyz:9000",
+    "sendBundleUrl": "",
+    "tips": "0.001",
+    "rateLimit": 0
 }
 ```
 
@@ -278,6 +294,177 @@ See [example](./cmd/shreder/main.go).
     "keepAliveUrl": "http://ny.solana.dex.blxrbdn.com/api/v2/ping",
     "tips": "0.001",
     "rateLimit": 0
+},
+{
+    "name": "soyas",
+    "sendTxUrl": "nyc.landing.soyas.xyz:9000",
+    "sendBundleUrl": "",
+    "tips": "0.001",
+    "rateLimit": 0
+}
+```
+
+</details>
+
+<details><summary> London </summary>
+
+```json
+{
+    "name": "helius",
+    "sendTxUrl": "http://lon-sender.helius-rpc.com/fast",
+    "sendBundleUrl": "",
+    "keepAliveUrl": "http://lon-sender.helius-rpc.com/ping",
+    "tips": "0.001",
+    "rateLimit": 0
+},
+{
+    "name": "blockrazor",
+    "sendTxUrl": "london.solana-grpc.blockrazor.xyz:80",
+    "sendBundleUrl": "",
+    "tips": "0.001",
+    "rateLimit": 0
+},
+{
+    "name": "node1",
+    "sendTxUrl": "http://lon.node1.me",
+    "sendBundleUrl": "",
+    "keepAliveUrl": "http://lon.node1.me/ping",
+    "tips": "0.001",
+    "rateLimit": 0
+},
+{
+    "name": "nextblock",
+    "sendTxUrl": "http://london.nextblock.io/api/v2/submit",
+    "sendBundleUrl": "",
+    "keepAliveUrl": "http://london.nextblock.io/api/v2/ping",
+    "tips": "0.001",
+    "rateLimit": 0
+
+},
+{
+    "name": "flashBlock",
+    "sendTxUrl": "http://london.flashblock.trade/api/v2/submit-batch",
+    "sendBundleUrl": "http://london.flashblock.trade/api/v2/submit-batch",
+    "keepAliveUrl": "http://london.flashblock.trade/api/v2/ping",
+    "tips": "0.001",
+    "rateLimit": 0
+},
+{
+    "name": "stellium",
+    "sendTxUrl": "http://lhr1.flashrpc.com/be95e80d-afc2-4a48-b017-db021fc4c19e",
+    "sendBundleUrl": "",
+    "keepAliveUrl": "http://lhr1.flashrpc.com/ping",
+    "tips": "0.001",
+    "rateLimit": 0
+},
+{
+    "name": "blocxroute",
+    "sendTxUrl": "http://uk.solana.dex.blxrbdn.com/api/v2/submit",
+    "sendBundleUrl": "",
+    "keepAliveUrl": "http://uk.solana.dex.blxrbdn.com/api/v2/ping",
+    "tips": "0.001",
+    "rateLimit": 0
+},
+{
+    "name": "soyas",
+    "sendTxUrl": "lon.landing.soyas.xyz:9000",
+    "sendBundleUrl": "",
+    "tips": "0.001",
+    "rateLimit": 0
+}
+```
+
+</details>
+
+
+<details><summary> Japan </summary>
+
+```json
+{
+    "name": "helius",
+    "sendTxUrl": "http://tyo-sender.helius-rpc.com/fast",
+    "sendBundleUrl": "",
+    "keepAliveUrl": "http://tyo-sender.helius-rpc.com/ping",
+    "tips": "0.001",
+    "rateLimit": 0
+},
+{
+    "name": "0slot",
+    "sendTxUrl": "http://jp1.0slot.trade?api-key=3fec78a0d361418a8eff95be9ed85cc3&anti-mev=true",
+    "sendBundleUrl": "",
+    "keepAliveUrl": "http://jp1.0slot.trade/ping",
+    "tips": "0.001",
+    "rateLimit": 0
+},
+{
+    "name": "blockrazor",
+    "sendTxUrl": "tokyo.solana-grpc.blockrazor.xyz:80",
+    "sendBundleUrl": "",
+    "tips": "0.001",
+    "rateLimit": 0
+},
+{
+    "name": "node1",
+    "sendTxUrl": "http://tk.node1.me",
+    "sendBundleUrl": "",
+    "keepAliveUrl": "http://tk.node1.me/ping",
+    "tips": "0.001",
+    "rateLimit": 0
+},
+{
+    "name": "nextblock",
+    "sendTxUrl": "http://tokyo.nextblock.io/api/v2/submit",
+    "sendBundleUrl": "",
+    "keepAliveUrl": "http://tokyo.nextblock.io/api/v2/ping",
+    "tips": "0.001",
+    "rateLimit": 4
+},
+{
+    "name": "flashBlock",
+    "sendTxUrl": "http://tokyo.flashblock.trade/api/v2/submit-batch",
+    "sendBundleUrl": "http://tokyo.flashblock.trade/api/v2/submit-batch",
+    "keepAliveUrl": "http://tokyo.flashblock.trade/api/v2/ping",
+    "tips": "0.001",
+    "rateLimit": 0
+},
+{
+    "name": "astralane",
+    "sendTxUrl": "http://jp.gateway.astralane.io/iris?api-key=zhaozNc5OIadLPI3r9nUVVPpCZcQAUjngO6Tgr5XUJcmBrIisFaaZF81Ijn01Ytn",
+    "sendBundleUrl": "",
+    "keepAliveUrl": "http://jp.gateway.astralane.io/ping",
+    "tips": "0.001",
+    "rateLimit": 0
+},
+{
+    "name": "nozomi",
+    "sendTxUrl": "http://tyo1.nozomi.temporal.xyz/?c=34cff37e-f1a5-446a-98bb-66aa1b62cb74",
+    "sendBundleUrl": "",
+    "keepAliveUrl": "http://tyo1.nozomi.temporal.xyz/ping",
+    "tips": "0.001",
+    "rateLimit": 0
+},
+{
+    "name": "stellium",
+    "sendTxUrl": "http://tyo1.flashrpc.com/be95e80d-afc2-4a48-b017-db021fc4c19e",
+    "sendBundleUrl": "",
+    "keepAliveUrl": "http://tyo1.flashrpc.com/ping",
+    "tips": "0.001",
+    "rateLimit": 0
+},
+{
+    "name": "blocxroute",
+    "sendTxUrl": "http://tokyo.solana.dex.blxrbdn.com/api/v2/submit",
+    "sendBundleUrl": "",
+    "keepAliveUrl": "http://tokyo.solana.dex.blxrbdn.com/api/v2/ping",
+    "tips": "0.001",
+    "rateLimit": 0
+},
+{
+    "name": "soyas",
+    "sendTxUrl": "tyo.landing.soyas.xyz:9000",
+    "sendBundleUrl": "",
+    "tips": "0.001",
+    "rateLimit": 0
 }
 ```
 

cmd/debug_jupv6/main.go

@@ -1,62 +0,0 @@
-package main
-
-import (
-	"encoding/hex"
-	"fmt"
-	"os"
-)
-
-func main() {
-	hexData := "bb64facc31c4af14be34e6edcc0000006f03a4df67000000b903320000000300000064342100024b00000000dc0500026310270203"
-	b, err := hex.DecodeString(hexData)
-	if err != nil {
-		panic(err)
-	}
-	payload := b[8:]
-
-	off := 0
-	read := func(n int) []byte {
-		if off+n > len(payload) {
-			fmt.Printf("OOB read: off=%d n=%d len=%d\n", off, n, len(payload))
-			os.Exit(1)
-		}
-		out := payload[off : off+n]
-		off += n
-		return out
-	}
-	u8 := func() uint8 { return read(1)[0] }
-	leU16 := func() uint16 {
-		b := read(2)
-		return uint16(b[0]) | uint16(b[1])<<8
-	}
-	leU32 := func() uint32 {
-		b := read(4)
-		return uint32(b[0]) | uint32(b[1])<<8 | uint32(b[2])<<16 | uint32(b[3])<<24
-	}
-	leU64 := func() uint64 {
-		b := read(8)
-		return uint64(b[0]) | uint64(b[1])<<8 | uint64(b[2])<<16 | uint64(b[3])<<24 |
-			uint64(b[4])<<32 | uint64(b[5])<<40 | uint64(b[6])<<48 | uint64(b[7])<<56
-	}
-
-	fmt.Printf("payload len=%d\n", len(payload))
-	amountIn := leU64()
-	quotedOut := leU64()
-	slippage := leU16()
-	platform := leU16()
-	posSlip := leU16()
-	fmt.Printf("in=%d out=%d slip=%d plat=%d pos=%d\n", amountIn, quotedOut, slippage, platform, posSlip)
-
-	planLen := leU32()
-	fmt.Printf("planLen=%d\n", planLen)
-	for i := uint32(0); i < planLen; i++ {
-		swapTag := u8()
-		fmt.Printf("step[%d] swapTag=%d (0x%02x) off=%d\n", i, swapTag, swapTag, off)
-		// payload depends on swapTag; we don't know, so just print next few bytes and stop
-		bps := leU16()
-		inIdx := u8()
-		outIdx := u8()
-		fmt.Printf("  bps=%d inIdx=%d outIdx=%d off=%d\n", bps, inIdx, outIdx, off)
-	}
-	fmt.Printf("done off=%d\n", off)
-}

cmd/shreder/main.go

@@ -8,8 +8,10 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
 
 	"github.com/gagliardetto/solana-go/rpc"
+	"github.com/shopspring/decimal"
 
 	"github.com/samlior/libsam/pkg/shreder"
 )
@@ -55,8 +57,13 @@ func main() {
 					"proVF4pMXVaYqmy4NjniPh4pqKNfMmsihgd4wdkCX3u",
 				},
 			},
+			"dflow": {
+				AccountRequired: []string{
+					"DF1ow4tspfHX9JwWJsAb9epbkA8hmpSEAtxXy1V27QBH",
+				},
+			},
 			// TODO: axiom, gmgn, etc.
-		})
+		}, shreder.BlocksStats(false), shreder.LogParsedStats(true), shreder.ShowTableLoaded(false))
 	if err != nil {
 		panic(err)
 	}
@@ -70,11 +77,10 @@ func main() {
 		<-exitSignal
 		cancel()
 	}()
-
 	// async read from shreder
-	txCh := make(chan shreder.TxSignalBatch, 1000)
+	txCh := make(chan shreder.TxSignal, 1000)
 	go func() {
-		err := shrederClient.ReadSync(ctx, txCh)
+		err := shrederClient.ReadEntriesSync(ctx, txCh)
 		if err != nil {
 			if !errors.Is(err, context.Canceled) {
 				panic(err)
@@ -86,17 +92,20 @@ func main() {
 		select {
 		case <-ctx.Done():
 			return
-		case txBatch := <-txCh:
+		case tx := <-txCh:
 			//jsonData, _ := json.MarshalIndent(txBatch, "", "  ")
-			for _, tx := range txBatch {
+			if tx.Token0Amount.GreaterThan(decimal.NewFromInt(100)) && (tx.Label == "okxdexroutev2" || tx.Label == "jupiterv6" || tx.Label == "dflow") {
-				if tx.Label == "okxdexroutev2" {
+				fmt.Println(time.Now(), "===============", tx.TxHash,
-					if tx.Event == "buy" {
+					"parse time:", tx.Stats.Done.Sub(tx.Stats.Filter),
-						fmt.Println("===============", tx.TxHash, tx.Event, tx.Token0Address, "token:", tx.Token0Amount, "sol:", tx.Token1Amount)
+					"decode time:", tx.Stats.Decoded.Sub(tx.Stats.FEC),
-					} else if tx.Event == "sell" {
+					"filter time:", tx.Stats.Filter.Sub(tx.Stats.Decoded),
-						fmt.Println("===============", tx.TxHash, tx.Event, tx.Token0Address, "token:", tx.Token0Amount)
+					"dataLen", tx.Stats.DataLen, "txCount", tx.Stats.TxCount, "txOffset", tx.Stats.TxOffset, tx.Label, tx.Event, "token:", tx.Token0Amount)
-					}
-				}
 			}
+
+			//if tx.Token0Amount.GreaterThan(decimal.NewFromInt(100)) && (tx.Label == "okxdexroutev2" || tx.Label == "jupiterv6" || tx.Label == "dflow") {
+			//	fmt.Println(time.Now(), "===============", tx.TxHash,
+			//		tx.Label, tx.Event, "token:", tx.Token0Amount)
+			//}
 			//fmt.Println(txBatch[0].TxHash)
 		}
 	}

go.mod

@@ -32,6 +32,7 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/mostynb/zstdpool-freelist v0.0.0-20201229113212-927304c0c3b1 // indirect
+	github.com/quic-go/quic-go v0.58.0 // indirect
 	github.com/streamingfast/logging v0.0.0-20230608130331-f22c91403091 // indirect
 	go.mongodb.org/mongo-driver v1.12.2 // indirect
 	go.uber.org/atomic v1.7.0 // indirect

go.sum

@@ -77,6 +77,8 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/quic-go/quic-go v0.58.0 h1:ggY2pvZaVdB9EyojxL1p+5mptkuHyX5MOSv4dgWF4Ug=
+github.com/quic-go/quic-go v0.58.0/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU=
 github.com/shopspring/decimal v1.3.1/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o=
 github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k=
 github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME=
@@ -88,6 +90,7 @@ github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
 github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/test-go/testify v1.1.4 h1:Tf9lntrKUMHiXQ07qBScBTSA0dhYQlu83hswqelv1iE=
 github.com/test-go/testify v1.1.4/go.mod h1:rH7cfJo/47vWGdi4GPj16x3/t1xGOj2YxzmNQzk2ghU=
 github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=

pkg/consts/swqos_fee_addresses.go

@@ -121,4 +121,5 @@ var SWQoSFeeAddresses = map[string]string{
 	"ste11p5x8tJ53H1NbNQsRBg1YNRd4GcVpxtDw8PBpmb":  enum.SWQoSAgentStellium,
 	"ste11p7e2KLYou5bwtt35H7BM6uMdo4pvioGjJXKFcN":  enum.SWQoSAgentStellium,
 	"ste11TMV68LMi1BguM4RQujtbNCZvf1sjsASpqgAvSX":  enum.SWQoSAgentStellium,
+	"soyas4s6L8KWZ8rsSk1mF3d1mQScoTGGAgjk98bF8nP":  enum.SWQoSAgentSoyas,
 }

pkg/enum/enum.go

@@ -12,4 +12,5 @@ const (
 	SWQoSAgentBlockRazor = "blockrazor"
 	SWQoSAgentAstralane  = "astralane"
 	SWQoSAgentStellium   = "stellium"
+	SWQoSAgentSoyas      = "soyas"
 )

pkg/shreder/addresstables.go

@@ -11,24 +11,33 @@ import (
 	"github.com/panjf2000/ants/v2"
 )
 
+type TableInfo struct {
+	overErrCount int
+
+	addresses []solana.PublicKey
+}
+
 type AddressTables struct {
+	showTableLoaded bool
+
 	rpcClient *rpc.Client
-	mux       sync.RWMutex
 	loadMux   sync.Mutex
-	tables    *lru.Cache[solana.PublicKey, []solana.PublicKey]
+	tables    *lru.Cache[solana.PublicKey, *TableInfo]
 	loading   map[solana.PublicKey]struct{}
 
 	pool *ants.Pool
 }
 
-func NewAddressTables(rpcClient *rpc.Client) *AddressTables {
+func NewAddressTables(rpcClient *rpc.Client, showTableLoaded bool) *AddressTables {
 	pool, _ := ants.NewPool(5, ants.WithPreAlloc(true), ants.WithNonblocking(true))
-	cache, _ := lru.New[solana.PublicKey, []solana.PublicKey](10000)
+	cache, _ := lru.New[solana.PublicKey, *TableInfo](10000)
 	return &AddressTables{
 		rpcClient: rpcClient,
 		tables:    cache,
 		loading:   make(map[solana.PublicKey]struct{}),
 		pool:      pool,
+
+		showTableLoaded: showTableLoaded,
 	}
 }
 
@@ -50,57 +59,82 @@ func (at *AddressTables) loadAddressTable(tablePubkey solana.PublicKey) ([]solan
 		addresses = append(addresses, solana.PublicKeyFromBytes(data[offset:offset+32]))
 		offset += 32
 	}
-	// addresses = append(addresses, solana.PublicKeyFromBytes(data[start:start+32]))
 	return addresses, nil
 
 }
-
+func (at *AddressTables) load(tablePubkey solana.PublicKey) {
-func (at *AddressTables) GetAddressTable(tablePubkey solana.PublicKey, idx []uint8) []solana.PublicKey {
+	_ = at.pool.Submit(func() {
-	at.mux.RLock()
+		at.loadMux.Lock()
-	addresses, ok := at.tables.Get(tablePubkey)
+		_, loading := at.loading[tablePubkey]
-	if !ok {
+		if loading {
-		at.mux.RUnlock()
-		_ = at.pool.Submit(func() {
-			at.loadMux.Lock()
-			_, loading := at.loading[tablePubkey]
-			if loading {
-				at.loadMux.Unlock()
-				return
-			}
-			at.loading[tablePubkey] = struct{}{}
 			at.loadMux.Unlock()
+			return
+		}
+		at.loading[tablePubkey] = struct{}{}
+		at.loadMux.Unlock()
 
-			table, err := at.loadAddressTable(tablePubkey)
+		table, err := at.loadAddressTable(tablePubkey)
-			if err != nil {
+		if err != nil {
-				logger.Error("loadAddressTable failed", "err", err, "table", tablePubkey)
+			logger.Error("loadAddressTable failed", "err", err, "table", tablePubkey)
-				at.loadMux.Lock()
-				delete(at.loading, tablePubkey)
-				at.loadMux.Unlock()
-				return
-			}
 			at.loadMux.Lock()
 			delete(at.loading, tablePubkey)
 			at.loadMux.Unlock()
+			return
+		}
+		at.loadMux.Lock()
+		delete(at.loading, tablePubkey)
+		at.loadMux.Unlock()
 
-			at.mux.Lock()
+		at.tables.Add(tablePubkey, &TableInfo{
-			at.tables.Add(tablePubkey, table)
+			addresses: table,
-			total := at.tables.Len()
-			at.mux.Unlock()
-			logger.Info("loadAddressTable", "table", tablePubkey.String(), "table count:", total)
 		})
+		if at.showTableLoaded {
+			total := at.tables.Len()
+			logger.Info("loadAddressTable", "table", tablePubkey.String(), "table count:", total)
+		}
 
+	})
+}
+
+func (at *AddressTables) FillToTx(tx FillableTransaction, tablePubkey solana.PublicKey, idx []uint8) bool {
+	addresses, ok := at.tables.Get(tablePubkey)
+	if !ok {
+		at.load(tablePubkey)
+		return false
+	}
+
+	for _, i := range idx {
+		if int(i) >= len(addresses.addresses) {
+			logger.Error("over loadAddressTable failed", "idx", i, "table", tablePubkey)
+			addresses.overErrCount++
+			if addresses.overErrCount > 10 {
+				at.load(tablePubkey)
+			}
+			return false
+		}
+		tx.FillLookupTable(addresses.addresses[i])
+	}
+	return true
+}
+
+func (at *AddressTables) GetAddressTable(tablePubkey solana.PublicKey, idx []uint8) []solana.PublicKey {
+	addresses, ok := at.tables.Get(tablePubkey)
+	if !ok {
+		at.load(tablePubkey)
 		return nil
 	}
-	at.mux.RUnlock()
 
 	var result solana.PublicKeySlice = make([]solana.PublicKey, 0, len(idx))
 	for _, i := range idx {
-		if int(i) >= len(addresses) {
+		if int(i) >= len(addresses.addresses) {
 			logger.Error("over loadAddressTable failed", "idx", i, "table", tablePubkey)
-			//todo... update table?
+			addresses.overErrCount++
-			continue
+			if addresses.overErrCount > 10 {
+				at.load(tablePubkey)
+			}
+			break
 		}
-		result = append(result, addresses[i])
+		result = append(result, addresses.addresses[i])
 	}
 	return result
 }

pkg/shreder/client.go

@@ -3,23 +3,61 @@ package shreder
 import (
 	"context"
 	"fmt"
+	"runtime"
+	"time"
 
 	"github.com/gagliardetto/solana-go/rpc"
+	"github.com/panjf2000/ants/v2"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 )
 
 type Client struct {
+	enableBlockStats bool
+	enableParseStats bool
+
 	conn         *grpc.ClientConn
 	client       ShrederServiceClient
 	tableLoader  *AddressTables
 	subscription map[string]*SubscribeRequestFilterTransactions
+
+	pool *ants.Pool
+
+	lastSlot     uint64
+	lastSlotTime time.Time
+}
+
+type ClientOpts struct {
+	blockStats      bool
+	showTableLoaded bool
+	logParseStats   bool
+}
+
+type ClientOption func(*ClientOpts)
+
+func ShowTableLoaded(enable bool) ClientOption {
+	return func(opts *ClientOpts) {
+		opts.showTableLoaded = enable
+	}
+}
+
+func BlocksStats(enable bool) ClientOption {
+	return func(opts *ClientOpts) {
+		opts.blockStats = enable
+	}
+}
+
+func LogParsedStats(enable bool) ClientOption {
+	return func(opts *ClientOpts) {
+		opts.logParseStats = enable
+	}
 }
 
 func NewShrederClient(
 	url string,
 	rpcClient *rpc.Client,
 	subscription map[string]*SubscribeRequestFilterTransactions,
+	options ...ClientOption,
 ) (*Client, func(), error) {
 	if rpcClient == nil {
 		return nil, func() {}, fmt.Errorf("rpc client is nil")
@@ -30,11 +68,29 @@ func NewShrederClient(
 		return nil, func() {}, err
 	}
 
+	poolSize := runtime.NumCPU()*2 + 2
+	logger.Info("creating shreder client", "url", url, "pool_size", poolSize)
+	pool, err := ants.NewPool(poolSize, ants.WithNonblocking(true))
+	if err != nil {
+		return nil, func() {}, err
+	}
+	o := &ClientOpts{
+		blockStats:      false,
+		showTableLoaded: true,
+		logParseStats:   false,
+	}
+	for _, option := range options {
+		option(o)
+	}
 	s := &Client{
 		conn:         conn,
 		client:       NewShrederServiceClient(conn),
 		subscription: subscription,
-		tableLoader:  NewAddressTables(rpcClient),
+		tableLoader:  NewAddressTables(rpcClient, o.showTableLoaded),
+		pool:         pool,
+
+		enableBlockStats: o.blockStats,
+		enableParseStats: o.logParseStats,
 	}
 
 	return s, func() {
@@ -45,6 +101,10 @@ func NewShrederClient(
 func (c *Client) Wait() {
 	logger.Debug("waiting for shreder client to stop")
 
+	if c.pool != nil {
+		c.pool.Release()
+	}
+
 	err := c.conn.Close()
 	if err != nil {
 		logger.Error("failed to close connection: ", "err", err)
@@ -53,12 +113,49 @@ func (c *Client) Wait() {
 	logger.Debug("shreder client stopped")
 }
 
-func (c *Client) ReadSync(ctx context.Context, txCh chan<- TxSignalBatch) error {
+func (c *Client) ReadEntriesSync(ctx context.Context, txCh chan<- TxSignal) error {
+	stream, err := c.client.SubscribeEntries(ctx, &SubscribeEntriesRequest{})
+	if err != nil {
+		return err
+	}
+
+	logger.Debug("reading entries from shreder client")
+	for {
+		response, err := stream.Recv()
+		if err != nil {
+			return err
+		}
+		slot := response.Slot
+		if c.enableBlockStats {
+			now := time.Now()
+			if c.lastSlotTime.IsZero() || slot > c.lastSlot {
+				if !c.lastSlotTime.IsZero() {
+					logger.Info("block processed", "running", c.pool.Running(), "slot", slot, "prev_slot", c.lastSlot, "delta_ms", now.Sub(c.lastSlotTime).Milliseconds())
+				}
+				c.lastSlot = slot
+				c.lastSlotTime = now
+			}
+		}
+
+		entries := response.Entries
+
+		err = c.pool.Submit(func() {
+			ParseEntries(slot, entries, c.tableLoader, txCh, c.enableParseStats)
+		})
+		if err != nil {
+			return err
+		}
+	}
+}
+
+func (c *Client) ReadSync(ctx context.Context, txCh chan<- TxSignal) error {
 	stream, err := c.client.SubscribeTransactions(ctx)
 	if err != nil {
 		return err
 	}
 
+	logger.Debug("subscribing to transactions")
+
 	err = stream.Send(&SubscribeTransactionsRequest{
 		Transactions: c.subscription,
 	})
@@ -72,20 +169,26 @@ func (c *Client) ReadSync(ctx context.Context, txCh chan<- TxSignalBatch) error
 			return err
 		}
 
-		txBatch := ParseTransaction(response.Transaction, c.tableLoader)
+		if c.enableBlockStats {
-		if len(txBatch) == 0 {
+			slot := response.Transaction.Slot
-			continue
+			now := time.Now()
+			if c.lastSlotTime.IsZero() || slot > c.lastSlot {
+				if !c.lastSlotTime.IsZero() {
+					logger.Info("block processed", "running", c.pool.Running(), "slot", slot, "prev_slot", c.lastSlot, "delta_ms", now.Sub(c.lastSlotTime).Milliseconds())
+				}
+				c.lastSlot = slot
+				c.lastSlotTime = now
+			}
 		}
 
-		// set fixed source for tx signals
+		txData := response.Transaction
-		for _, tx := range txBatch {
+
-			tx.Source = "shreder"
+		err = c.pool.Submit(func() {
+			ParseTransaction(txData, c.tableLoader, txCh, c.enableParseStats)
+		})
+		if err != nil {
+			logger.Error("failed to submit transaction: ", "err", err)
 		}
 
-		select {
-		case <-ctx.Done():
-			return ctx.Err()
-		case txCh <- txBatch:
-		}
 	}
 }

pkg/shreder/dflow.go

@@ -0,0 +1,321 @@
+package shreder
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+
+	bin "github.com/gagliardetto/binary"
+	"github.com/gagliardetto/solana-go"
+	"github.com/shopspring/decimal"
+)
+
+var (
+	dflowProgramID     = solana.MustPublicKeyFromBase58("DF1ow4tspfHX9JwWJsAb9epbkA8hmpSEAtxXy1V27QBH")
+	dflowProgramString = dflowProgramID.String()
+
+	dflowSwapDisc                   = []byte{248, 198, 158, 145, 225, 117, 135, 200}
+	dflowSwap2Disc                  = []byte{65, 75, 63, 76, 235, 91, 91, 136}
+	dflowSwapWithDestinationDisc    = []byte{168, 172, 24, 77, 197, 156, 135, 101}
+	dflowSwapWithDestinationNative  = []byte{205, 77, 127, 108, 241, 32, 196, 195}
+	dflowSwap2WithDestinationDisc   = []byte{95, 123, 213, 246, 122, 1, 86, 231}
+	dflowSwap2WithDestinationNative = []byte{222, 100, 184, 146, 186, 196, 105, 165}
+
+	wrappedSOL = solana.MustPublicKeyFromBase58("So11111111111111111111111111111111111111112")
+)
+
+// Action enum tags (0-based, per dflow_idl Action variants)
+const (
+	ActWhirlpoolsSwap uint8 = iota
+	ActClearpoolsSwap
+	ActRaydiumAmmSwap
+	ActLifinityV2Swap
+	ActMeteoraDlmmSwap
+	ActRaydiumClmmSwap
+	ActRaydiumClmmSwapV2
+	ActPhoenixSwap
+	ActPumpFunBuy
+	ActPumpFunSell
+	ActGammaSwap
+	ActObricV2Swap
+	ActPumpFunAmmBuy
+	ActPumpFunAmmSell
+	ActSolFiSwap
+	ActRubiconSwap
+	ActMeteoraDammV1Swap
+	ActRaydiumCpSwap
+	ActStabbleStableSwap
+	ActTesseraVSwap
+	ActMeteoraDammV2Swap
+	ActRaydiumLaunchlabSwap
+	ActMeteoraDbcSwap
+	ActHumidiFiSwap
+	ActWhirlpoolsSwapV2
+	ActMeteoraDlmmSwapV2
+	ActZeroFiSwap
+	ActAlphaQSwap
+	ActTokenSwap
+	ActSolFiV2Swap
+	ActMozartSwap
+	ActDFlowDynamicRouteV1
+	ActHeavenSwap
+	ActNexusSwap
+	ActSarosDlmmSwap
+	ActTransferFee
+	ActTransferFeeWithMint
+	ActRecordId
+	ActRecordId2
+	ActManifestSwap
+	ActBisonFiSwap
+	ActSanctumInfinitySwap
+	ActSanctumInfinityLiquidity
+	ActOpenPredictionsOrder
+	ActScorchSwap
+	ActIncludeAccount
+)
+
+// DynamicRouteV1CandidateAction tags
+const (
+	drv1SolFi uint8 = iota
+	drv1Rubicon
+	drv1TesseraV
+	drv1HumidiFi
+	drv1SolFiV2
+	drv1Mozart
+	drv1ObricV2
+	drv1Nexus
+)
+
+// PumpFunAmmSellOptions { amount: u64, orchestrator_flags: OrchestratorFlags{flags u8} }
+type pumpFunAmm struct {
+	Amount uint64
+	Flags  uint8
+}
+
+type dflowAction struct {
+	Tag  uint8
+	Pump *pumpFunAmm
+}
+
+type dflowSwapParams struct {
+	Actions []dflowAction
+}
+
+// bytes to skip for Action variants before/after PumpFunAmmSell; only PumpFunAmmSell is decoded.
+func skipDflowAction(dec *bin.Decoder, tag uint8) (*pumpFunAmm, error) {
+	switch tag {
+	case ActWhirlpoolsSwap, ActClearpoolsSwap, ActWhirlpoolsSwapV2:
+		// amount u64 + bool + orchestrator_flags u8
+		return nil, dec.SkipBytes(8 + 1 + 1)
+	case ActRaydiumAmmSwap, ActLifinityV2Swap, ActPumpFunBuy, ActPumpFunSell, ActObricV2Swap,
+		ActSolFiSwap, ActRubiconSwap, ActMeteoraDammV1Swap, ActRaydiumCpSwap,
+		ActStabbleStableSwap, ActTesseraVSwap, ActMeteoraDammV2Swap, ActRaydiumLaunchlabSwap,
+		ActZeroFiSwap, ActAlphaQSwap, ActTokenSwap, ActSolFiV2Swap, ActMozartSwap, ActHeavenSwap,
+		ActNexusSwap, ActSarosDlmmSwap, ActManifestSwap, ActBisonFiSwap:
+		// amount u64 + orchestrator_flags u8
+		return nil, dec.SkipBytes(8 + 1)
+	case ActMeteoraDlmmSwap, ActRaydiumClmmSwap, ActRaydiumClmmSwapV2, ActMeteoraDlmmSwapV2:
+		// amount u64 + u8 + orchestrator_flags u8
+		return nil, dec.SkipBytes(8 + 1 + 1)
+	case ActPhoenixSwap:
+		// amount u64 + side u8 + orchestrator_flags u8
+		return nil, dec.SkipBytes(8 + 1 + 1)
+	case ActGammaSwap:
+		// amount u64 + endorsed bool + orchestrator_flags u8
+		return nil, dec.SkipBytes(8 + 1 + 1)
+	case ActPumpFunAmmSell, ActPumpFunAmmBuy:
+		amt, err := dec.ReadUint64(binary.LittleEndian)
+		if err != nil {
+			return nil, err
+		}
+		flg, err := dec.ReadUint8()
+		if err != nil {
+			return nil, err
+		}
+		return &pumpFunAmm{Amount: amt, Flags: flg}, nil
+	case ActMeteoraDbcSwap:
+		// amount u64 + is_rate_limiter_applied bool + orchestrator_flags u8
+		return nil, dec.SkipBytes(8 + 1 + 1)
+	case ActHumidiFiSwap:
+		// amount u64 + swap_id u64 + orchestrator_flags u8
+		return nil, dec.SkipBytes(8 + 8 + 1)
+	case ActDFlowDynamicRouteV1:
+		// candidate_actions Vec<DynamicRouteV1CandidateAction> + amount u64 + orchestrator_flags u8
+		ln, err := dec.ReadUint32(binary.LittleEndian)
+		if err != nil {
+			return nil, err
+		}
+		for j := uint32(0); j < ln; j++ {
+			t, err := dec.ReadUint8()
+			if err != nil {
+				return nil, err
+			}
+			if t == drv1HumidiFi {
+				if err := dec.SkipBytes(8); err != nil {
+					return nil, err
+				}
+			}
+			// other variants carry no payload
+		}
+		if err := dec.SkipBytes(8); err != nil { // amount
+			return nil, err
+		}
+		return nil, dec.SkipBytes(1) // orchestrator_flags
+	case ActTransferFee, ActTransferFeeWithMint:
+		return nil, dec.SkipBytes(8)
+	case ActRecordId:
+		return nil, dec.SkipBytes(76)
+	case ActRecordId2:
+		return nil, dec.SkipBytes(4)
+	case ActSanctumInfinitySwap:
+		// amount u64 + src_lst_value_calc_accs u8 + dst_lst_value_calc_accs u8 + src_lst_index u32 + dst_lst_index u32 + orchestrator_flags u8
+		return nil, dec.SkipBytes(8 + 1 + 1 + 4 + 4 + 1)
+	case ActSanctumInfinityLiquidity:
+		// amount u64 + lst_value_calc_accs u8 + lst_index u32 + orchestrator_flags u8
+		return nil, dec.SkipBytes(8 + 1 + 4 + 1)
+	case ActOpenPredictionsOrder:
+		// nonce u64 + order_outcome u8 + quoted_out_amount u64 + slippage_bps u16 + platform_fee_recipient_vault pubkey(32) + platform_fee_scale u16
+		return nil, dec.SkipBytes(8 + 1 + 8 + 2 + 32 + 2)
+	case ActScorchSwap:
+		// amount u64 + id u128 + orchestrator_flags u8
+		return nil, dec.SkipBytes(8 + 16 + 1)
+	case ActIncludeAccount:
+		return nil, nil
+	default:
+		return nil, fmt.Errorf("unsupported action tag %d", tag)
+	}
+}
+
+// SwapParams: actions Vec<Action>, quoted_out_amount u64, slippage_bps u16, platform_fee_bps u16
+func decodeSwapParams(data []byte) (*dflowSwapParams, error) {
+	dec := bin.NewBorshDecoder(data)
+	out := &dflowSwapParams{}
+	ln, err := dec.ReadUint32(binary.LittleEndian)
+	if err != nil {
+		return nil, err
+	}
+	out.Actions = make([]dflowAction, 0, ln)
+	for i := uint32(0); i < ln; i++ {
+		tag, err := dec.ReadUint8()
+		if err != nil {
+			return nil, fmt.Errorf("actions[%d] tag: %w", i, err)
+		}
+		pump, err := skipDflowAction(dec, tag)
+		if err != nil {
+			return nil, fmt.Errorf("actions[%d]: %w", i, err)
+		}
+		out.Actions = append(out.Actions, dflowAction{Tag: tag, Pump: pump})
+	}
+	return out, nil
+}
+
+// Swap2Params: actions Vec<Action>, quoted_out_amount u64, slippage_bps u16, platform_fee_bps u16, positive_slippage_fee_limit_pct u8
+func decodeSwap2Params(data []byte) (*dflowSwapParams, error) {
+	dec := bin.NewBorshDecoder(data)
+	out := &dflowSwapParams{}
+	ln, err := dec.ReadUint32(binary.LittleEndian)
+	if err != nil {
+		return nil, err
+	}
+	out.Actions = make([]dflowAction, 0, ln)
+	for i := uint32(0); i < ln; i++ {
+		tag, err := dec.ReadUint8()
+		if err != nil {
+			return nil, fmt.Errorf("actions[%d] tag: %w", i, err)
+		}
+		pump, err := skipDflowAction(dec, tag)
+		if err != nil {
+			return nil, fmt.Errorf("actions[%d]: %w", i, err)
+		}
+		out.Actions = append(out.Actions, dflowAction{Tag: tag, Pump: pump})
+	}
+	return out, nil
+}
+
+func parseDFlowInstruction(tx TransactionGetter, accounts []uint8, data []byte) (*TxSignal, error) {
+	if len(data) < 8 {
+		return nil, nil
+	}
+
+	var err error
+	disc := data[:8]
+	payload := data[8:]
+
+	var params *dflowSwapParams
+	switch {
+	case bytes.Equal(disc, dflowSwapDisc), bytes.Equal(disc, dflowSwapWithDestinationDisc), bytes.Equal(disc, dflowSwapWithDestinationNative):
+		params, err = decodeSwapParams(payload)
+	case bytes.Equal(disc, dflowSwap2Disc), bytes.Equal(disc, dflowSwap2WithDestinationDisc), bytes.Equal(disc, dflowSwap2WithDestinationNative):
+		params, err = decodeSwap2Params(payload)
+	default:
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	if params == nil {
+		return nil, nil
+	}
+
+	var pump *pumpFunAmm
+	for _, act := range params.Actions {
+		if act.Tag == ActPumpFunAmmSell && act.Pump != nil {
+			pump = act.Pump
+			break
+		}
+	}
+	if pump == nil {
+		return nil, nil // only care about PumpFunAmmSell
+	}
+
+	// Require WSOL pair when destination mint is provided.
+	var (
+		srcIdx uint8
+	)
+	if len(accounts) <= 6 {
+		return nil, nil
+	}
+	accounts = accounts[5:]
+	for i, acctIdx := range accounts {
+		key, err := tx.GetAccount(acctIdx) //getStaticKey(tx.Message.StaticAccountKeys, int(acctIdx))
+		if err != nil {
+			return nil, err
+		}
+		if key.Equals(pumpAmmProgramID) {
+			srcIdx = uint8(i + 4)
+			break
+		}
+	}
+	if srcIdx == 0 || srcIdx+1 >= uint8(len(accounts)) {
+		return nil, nil
+	}
+
+	baseMint, err := tx.GetAccount(accounts[srcIdx]) // getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx]))
+	if err != nil {
+		return nil, err
+	}
+	quoteMint, err := tx.GetAccount(accounts[srcIdx+1]) // getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx+1]))
+	if err != nil {
+		return nil, err
+	}
+	if !quoteMint.Equals(solana.WrappedSol) {
+		return nil, nil
+	}
+	maker, _ := tx.GetAccount(0)
+
+	// Build TxSignal
+	sig := &TxSignal{
+		TxHash:             tx.Signatures(),
+		Maker:              maker.String(),
+		Program:            "PumpAMM",
+		Event:              "sell",
+		Token0Address:      baseMint.String(),
+		Token1Address:      wsolMint,
+		Token0Amount:       formatTokenAmount(pump.Amount),
+		Token1Amount:       decimal.Zero,
+		Token0AmountUint64: pump.Amount,
+		Block:              tx.Block(),
+		Token1AmountUint64: 0,
+	}
+	return sig, nil
+}

pkg/shreder/entry.go

@@ -0,0 +1,334 @@
+package shreder
+
+import (
+	"encoding/binary"
+	"fmt"
+	"io"
+)
+
+type constArray struct {
+	data []byte
+	size int
+
+	offset int
+}
+
+func newConstArray(data []byte) constArray {
+	return constArray{
+		data:   data,
+		size:   len(data),
+		offset: 0,
+	}
+}
+
+func (c *constArray) Len() int {
+	return c.size
+}
+
+func (c *constArray) Offset() int {
+	return c.offset
+}
+
+func (c *constArray) Read(cap int) ([]byte, error) {
+	if c.offset+cap > c.size {
+		return nil, io.EOF
+	}
+	c.offset += cap
+	return c.data[c.offset-cap : c.offset], nil
+}
+
+func (c *constArray) ReadBytes() (byte, error) {
+	if c.offset >= c.size {
+		return 0, io.EOF
+	}
+	c.offset++
+	return c.data[c.offset-1], nil
+}
+
+func (c *constArray) PeekBytes() (byte, error) {
+	if c.offset >= c.size {
+		return 0, io.EOF
+	}
+	return c.data[c.offset], nil
+}
+
+func (c *constArray) ReadU64() (uint64, error) {
+	if c.offset+8 > c.size {
+		return 0, io.EOF
+	}
+	c.offset += 8
+	return binary.LittleEndian.Uint64(c.data[c.offset-8 : c.offset]), nil
+}
+
+func (c *constArray) ReadCompactU16() (uint16, error) {
+	ln := 0
+	size := 0
+	for i := 0; i < 3; i++ {
+		if len(c.data[c.offset:]) == 0 {
+			return 0, fmt.Errorf("unable to decode compact u16 at %d: zero byte", i)
+		}
+		elem := int(c.data[c.offset+i])
+		if elem == 0 && i != 0 {
+			return 0, fmt.Errorf("alias")
+		}
+		if i == 2 && (elem&0x80) != 0 {
+			return 0, fmt.Errorf("byte three continues")
+		}
+		ln |= (elem & 0x7f) << (size * 7)
+		size++
+		if (elem & 0x80) == 0 {
+			break
+		}
+	}
+	c.offset += size
+	return uint16(ln), nil
+}
+
+func (c *constArray) Skip(size int) error {
+	if c.offset+size > c.size {
+		return io.EOF
+	}
+	c.offset += size
+	return nil
+}
+
+// entriesToVersionedTransaction converts raw entry bytes to versioned transactions.
+func entriesToVersionedTransaction(slot uint64, b constArray) ([]*versionedTransaction, error) {
+	b.offset = 0
+
+	entriesNum, err := b.ReadU64()
+	if err != nil {
+		return nil, fmt.Errorf("unable to read entries num: %w", err)
+	}
+	if entriesNum == 0 {
+		return nil, nil
+	}
+	//preCap := b.Len() / 256
+	//if preCap < int(entriesNum) {
+	//	preCap = int(entriesNum)
+	//}
+	vs := make([]*versionedTransaction, 0, entriesNum)
+
+	// logger.Debug("parsing entries", "count", entriesNum, "data len", b.Len(), "data", base64.StdEncoding.EncodeToString(b.data))
+	for i := uint64(0); i < entriesNum; i++ {
+		err = b.Skip(40)
+		if err != nil {
+			return vs, fmt.Errorf("failed to skip num_hashes + hash of entry %d: %w", i, err)
+		}
+		numTx, err := b.ReadU64()
+		if err != nil {
+			return vs, fmt.Errorf("failed to read num_transactions of entry %d: %w", i, err)
+		}
+		for j := 0; j < int(numTx); j++ {
+			numSignatures, err := b.ReadCompactU16()
+			if err != nil {
+				return vs, fmt.Errorf("failed to read numSignatures in entry %d, txn %d: %w", i, j, err)
+			}
+
+			// todo : enforce a maximum number of signatures to prevent OOM
+			if numSignatures > 16 {
+				return vs, fmt.Errorf("numSignatures %d exceeds maximum in entry %d, txn %d", numSignatures, i, j)
+			}
+			if numSignatures == 0 {
+				return vs, fmt.Errorf("numSignatures is zero in entry %d, txn %d", i, j)
+			}
+
+			versioned := requireVersionedPool() // get a versioned transaction from the pool
+			vs = append(vs, versioned)
+			versioned.block = slot
+			versioned.bindArray = b.data
+			versioned.signatures = int(numSignatures)
+			versioned.signaturesOffset = b.Offset()
+			err = b.Skip(64 * int(numSignatures))
+			if err != nil {
+				return vs, fmt.Errorf("unable to read signature in entry %d, txn %d: %w", i, j, err)
+			}
+
+			msgVersion, err := b.PeekBytes()
+			if err != nil {
+				return vs, fmt.Errorf("unable to read message version in entry %d, txn %d: %w", i, j, err)
+			}
+			msgVersion = msgVersion & 0x80 >> 7 // mask to get only the version bits
+			legacy := msgVersion == 0
+			headerSkip := 3
+			if !legacy {
+				headerSkip = 4
+			}
+			// skip msg version, mx.Header+3
+
+			err = b.Skip(headerSkip)
+			if err != nil {
+				return vs, fmt.Errorf("unable to skip message header in entry %d, txn %d: %w", i, j, err)
+			}
+
+			// read mx.AccountKeys
+			// _, err = r.Read(u16[:])
+
+			numAccountKeys, err := b.ReadCompactU16()
+			// logger.Info("tx", "hash", versioned.Signatures[0].String(), "version", msgVersion)
+			if err != nil {
+				return vs, fmt.Errorf("unable to decode numAccountKeys in entry %d, txn %d: %w", i, j, err)
+			}
+			// todo : enforce a maximum number of account keys to prevent OOM
+			if numAccountKeys > 255 {
+				return vs, fmt.Errorf("numAccountKeys %d exceeds maximum in entry %d, txn %d", numAccountKeys, i, j)
+			}
+			versioned.staticAccountKeys = uint8(numAccountKeys)
+
+			versioned.staticAccountKeysOffset = b.Offset()
+			err = b.Skip(32 * int(numAccountKeys))
+			if err != nil {
+				return vs, fmt.Errorf("unable to read accountKey in entry %d, txn %d: %w", i, j, err)
+			}
+
+			//skip solana hash
+			err = b.Skip(32)
+			if err != nil {
+				return vs, fmt.Errorf("unable to skip recentBlockhash in entry %d, txn %d: %w", i, j, err)
+			}
+
+			// read mx.Instructions
+			numInstructions, err := b.ReadCompactU16()
+			if err != nil {
+				return vs, fmt.Errorf("unable to decode numInstructions in entry %d, txn %d: %w", i, j, err)
+			}
+
+			// todo : enforce a maximum number of instructions to prevent OOM
+			if numInstructions >= 256 {
+				return vs, fmt.Errorf("numInstructions %d exceeds maximum in entry %d, txn %d, txHash: %s", numInstructions, i, j, versioned.Signatures())
+			}
+			versioned.instructions = int(numInstructions)
+			if cap(versioned.Instrs) < int(numInstructions) {
+				versioned.Instrs = make([]compiledInstruction, numInstructions)
+			} else {
+				versioned.Instrs = versioned.Instrs[:numInstructions]
+			}
+			for k := 0; k < int(numInstructions); k++ {
+				versioned.Instrs[k].ProgramIDIndex, err = b.ReadBytes()
+				if err != nil {
+					return vs, fmt.Errorf("unable to read mx.Instructions[%d].ProgramIDIndex in entry %d, txn %d: %w", k, i, j, err)
+				}
+				numAccounts, err := b.ReadCompactU16()
+				if err != nil {
+					return vs, fmt.Errorf("unable to decode numAccounts for ix[%d] in entry %d, txn %d: %w", k, i, j, err)
+				}
+
+				// todo : enforce a maximum number of accounts to prevent OOM
+				if numAccounts >= 256 {
+					return vs, fmt.Errorf("numAccounts %d exceeds maximum for ix[%d] in entry %d, txn %d", numAccounts, k, i, j)
+				}
+				versioned.Instrs[k].AccountsLen = int(numAccounts)
+				if numAccounts != 0 {
+					versioned.Instrs[k].AccountsOffset = b.Offset()
+					err = b.Skip(int(numAccounts))
+					if err != nil {
+						return vs, fmt.Errorf("unable to read mx.Instructions[%d].Accounts in entry %d, txn %d: %w", k, i, j, err)
+					}
+				}
+				// _, err = r.Read(u16[:])
+				dataLen, err := b.ReadCompactU16()
+				if err != nil {
+					return vs, fmt.Errorf("unable to decode mx.Instructions[%d].Data length in entry %d, txn %d: %w", k, i, j, err)
+				}
+				//todo : enforce a maximum data length to prevent OOM
+				if dataLen > 2048 {
+					return vs, fmt.Errorf("mx.Instructions[%d].Data length %d exceeds maximum in entry %d, txn %d, txHash: %s", k, dataLen, i, j, versioned.Signatures())
+				}
+				versioned.Instrs[k].DataLen = int(dataLen)
+				if dataLen > 0 {
+					versioned.Instrs[k].DataOffset = b.Offset()
+					err = b.Skip(int(dataLen))
+					if err != nil {
+						return vs, fmt.Errorf("unable to read mx.Instructions[%d].Data in entry %d, txn %d: %w", k, i, j, err)
+					}
+				}
+			}
+
+			if !legacy {
+				// read mx.AddressTableLookups
+				numLookups, err := b.ReadBytes()
+				if err != nil {
+					return vs, fmt.Errorf("unable to read numAddressTableLookups in entry %d, txn %d: %w", i, j, err)
+				}
+
+				if cap(versioned.ATL) < int(numLookups) {
+					versioned.ATL = make([]addressTableLookup, numLookups)
+				} else {
+					versioned.ATL = versioned.ATL[:numLookups]
+				}
+				versioned.addressTableLookups = int(numLookups)
+
+				for k := uint8(0); k < numLookups; k++ {
+					versioned.ATL[k].AccountKeyOffset = b.Offset()
+					err = b.Skip(32)
+					if err != nil {
+						return vs, fmt.Errorf("unable to read address table account key for lookup[%d] in entry %d, txn %d: %w", k, i, j, err)
+					}
+					numWritable, err := b.ReadCompactU16()
+					if err != nil {
+						return vs, fmt.Errorf("unable to decode numWritableIndexes for lookup[%d] in entry %d, txn %d: %w", k, i, j, err)
+					}
+					// todo : enforce a maximum number of writable indexes to prevent OOM
+					if numWritable >= 256 {
+						return vs, fmt.Errorf("numWritableIndexes %d exceeds maximum for lookup[%d] in entry %d, txn %d", numWritable, k, i, j)
+					}
+					versioned.ATL[k].WriteLen = int(numWritable)
+					if numWritable > 0 {
+						versioned.ATL[k].WriteOffset = b.Offset()
+						err = b.Skip(int(numWritable))
+						if err != nil {
+							return vs, fmt.Errorf("unable to read writableIndexes for lookup[%d] in entry %d, txn %d: %w", k, i, j, err)
+						}
+					}
+					// _, err = r.Read(u16[:])
+
+					numReadonly, err := b.ReadCompactU16()
+					if err != nil {
+						return vs, fmt.Errorf("unable to decode numReadonlyIndexes for lookup[%d] in entry %d, txn %d: %w", k, i, j, err)
+					}
+					// todo : enforce a maximum number of readonly indexes to prevent OOM
+					if numReadonly > 256 {
+						return vs, fmt.Errorf("numReadonlyIndexes %d exceeds maximum for lookup[%d] in entry %d, txn %d", numReadonly, k, i, j)
+					}
+					versioned.ATL[k].ReadLen = int(numReadonly)
+
+					if numReadonly > 0 {
+						versioned.ATL[k].ReadOffset = b.Offset()
+						err = b.Skip(int(numReadonly))
+						if err != nil {
+							return vs, fmt.Errorf("unable to read readonlyIndexes for lookup[%d] in entry %d, txn %d: %w", k, i, j, err)
+						}
+					}
+				}
+			}
+		}
+	}
+	// logger.Debug("parsing num_transactions of entry", "slot", slot, "count", entriesNum, "data len", b.Len(), "num_tx", uint32(len(vs)))
+
+	//logger.Debug("parsing entries", "slot", slot, "count", entriesNum, "data len", b.Len(), "txns", len(vs))
+	return vs, nil
+}
+
+func decodeCompactU16(b []byte) (int, uint16, error) {
+	ln := 0
+	size := 0
+	for i := 0; i < 3; i++ {
+		if len(b) == 0 {
+			return 0, 0, fmt.Errorf("unable to decode compact u16 at %d: zero byte", i)
+		}
+		elem := int(b[0])
+		b = b[1:]
+		if elem == 0 && i != 0 {
+			return 0, 0, fmt.Errorf("alias")
+		}
+		if i == 2 && (elem&0x80) != 0 {
+			return 0, 0, fmt.Errorf("byte three continues")
+		}
+		ln |= (elem & 0x7f) << (size * 7)
+		size++
+		if (elem & 0x80) == 0 {
+			break
+		}
+	}
+	return size, uint16(ln), nil
+}

pkg/shreder/juptierv6.go

@@ -859,21 +859,16 @@ func pumpSwapSellAtIdx0V2(amount uint64, plan []RoutePlanStepV2) (uint64, int) {
 }
 
 // only decodes inputIdx = 0 container pumpSwap instructions for now
-func parseJupiterV6Instruction(tx *versionedTransaction, instructionIndex int) (*TxSignal, error) {
+func parseJupiterV6Instruction(tx TransactionGetter, accounts []uint8, data []byte) (*TxSignal, error) {
-	msg := tx.Message
-	if instructionIndex >= len(msg.Instructions) {
-		return nil, fmt.Errorf("instruction index out of bounds")
-	}
 
-	instruction := msg.Instructions[instructionIndex]
+	if len(data) == 0 {
-	if len(instruction.Data) == 0 {
 		return nil, fmt.Errorf("data is empty")
 	}
-	if len(instruction.Data) < 8 {
+	if len(data) < 8 {
 		return nil, nil
 	}
 
-	disc := instruction.Data[:8]
+	disc := data[:8]
 
 	var (
 		sourceMint  solana.PublicKey
@@ -886,26 +881,26 @@ func parseJupiterV6Instruction(tx *versionedTransaction, instructionIndex int) (
 	// route/shared_accounts_* (v1) use different account layouts; we only decode args here.
 	switch {
 	case bytes.Equal(disc, jupiterRouteV2):
-		args, err := decodeJupiterV6RouteV2Arg(instruction.Data[8:])
+		args, err := decodeJupiterV6RouteV2Arg(data[8:])
 		if err != nil {
 			return nil, err
 		}
 		inputAmount, planCount = pumpSwapSellAtIdx0V2(args.In, args.Plan)
 	case bytes.Equal(disc, jupiterSharedAccountsRouteV2):
-		args, err := decodeJupiterV6SharedAccountsRouteV2Arg(instruction.Data[8:])
+		args, err := decodeJupiterV6SharedAccountsRouteV2Arg(data[8:])
 		if err != nil {
 			return nil, err
 		}
 		inputAmount, planCount = pumpSwapSellAtIdx0V2(args.In, args.RoutePlan)
 	case bytes.Equal(disc, jupiterRoute):
-		args, err := decodeJupiterV6RouteArg(instruction.Data[8:])
+		args, err := decodeJupiterV6RouteArg(data[8:])
 		if err != nil {
 			return nil, err
 		}
 		_ = args
 		inputAmount, planCount = pumpSwapSellAtIdx0(args.In, args.Plan)
 	case bytes.Equal(disc, jupiterSharedAccountsRoute):
-		args, err := decodeJupiterV6SharedAccountsRouteArg(instruction.Data[8:])
+		args, err := decodeJupiterV6SharedAccountsRouteArg(data[8:])
 		if err != nil {
 			return nil, err
 		}
@@ -916,7 +911,8 @@ func parseJupiterV6Instruction(tx *versionedTransaction, instructionIndex int) (
 	}
 	if planCount > 1 {
 		// multiple pumpSwapSell at inputIdx=0? should not happen
-		logger.Warn("pumpSwapSell at inputIdx=0: multiple instances found", "tx", tx.Signatures[0].String(), "planCount", planCount)
+		logger.Warn("pumpSwapSell at inputIdx=0: multiple instances found", "tx", tx.Signatures(), "planCount", planCount)
+		return nil, nil
 	}
 	if inputAmount == 0 {
 		return nil, nil
@@ -924,10 +920,10 @@ func parseJupiterV6Instruction(tx *versionedTransaction, instructionIndex int) (
 
 	// existing mint extraction logic only valid for route_v2/ exact_out_route_v2. Keep it but guard.
 	if bytes.Equal(disc, jupiterRouteV2) || bytes.Equal(disc, jupiterSharedAccountsRouteV2) {
-		if len(instruction.Accounts) < 6 {
+		if len(accounts) < 6 {
 			return nil, fmt.Errorf("not enough accounts for jupiter v6 v2 instruction")
 		}
-		sourceMint, err = getStaticKey(tx.Message.StaticAccountKeys, int(instruction.Accounts[3]))
+		sourceMint, err = tx.GetAccount(accounts[3]) //getStaticKey(tx.Message.StaticAccountKeys, int(instruction.Accounts[3]))
 		if err != nil {
 			return nil, err
 		}
@@ -935,11 +931,12 @@ func parseJupiterV6Instruction(tx *versionedTransaction, instructionIndex int) (
 		var (
 			srcIdx uint8
 		)
-		for i, acctIdx := range instruction.Accounts {
+		if len(accounts) <= 9 {
-			if i < 9 {
+			return nil, nil
-				continue
+		}
-			}
+		accounts = accounts[8:]
-			key, err := getStaticKey(tx.Message.StaticAccountKeys, int(acctIdx))
+		for i, acctIdx := range accounts {
+			key, err := tx.GetAccount(acctIdx) // getStaticKey(tx.Message.StaticAccountKeys, int(acctIdx))
 			if err != nil {
 				return nil, err
 			}
@@ -948,18 +945,18 @@ func parseJupiterV6Instruction(tx *versionedTransaction, instructionIndex int) (
 				break
 			}
 		}
-		if srcIdx == 0 || srcIdx+1 >= uint8(len(instruction.Accounts)) {
+		if srcIdx == 0 || srcIdx+1 >= uint8(len(accounts)) {
 			return nil, nil
 		}
 
-		baseMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(instruction.Accounts[srcIdx]))
+		baseMint, err := tx.GetAccount(accounts[srcIdx]) // getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx]))
 		if err != nil {
 			return nil, err
 		}
 		if !sourceMint.Equals(baseMint) {
 			return nil, nil
 		}
-		quoteMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(instruction.Accounts[srcIdx+1]))
+		quoteMint, err := tx.GetAccount(accounts[srcIdx+1]) // getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx+1]))
 		if err != nil {
 			return nil, err
 		}
@@ -968,21 +965,22 @@ func parseJupiterV6Instruction(tx *versionedTransaction, instructionIndex int) (
 		}
 
 	} else if bytes.Equal(disc, jupiterSharedAccountsRoute) {
-		if len(instruction.Accounts) < 12 {
+		if len(accounts) < 12 {
 			return nil, fmt.Errorf("not enough accounts for jupiter v6 jupiterSharedAccountsRoute instruction")
 		}
-		sourceMint, err = getStaticKey(tx.Message.StaticAccountKeys, int(instruction.Accounts[7]))
+		sourceMint, err = tx.GetAccount(accounts[7]) // getStaticKey(tx.Message.StaticAccountKeys, int(instruction.Accounts[7]))
 		if err != nil {
 			return nil, err
 		}
 		var (
 			srcIdx uint8
 		)
-		for i, acctIdx := range instruction.Accounts {
+		if len(accounts) <= 12 {
-			if i < 12 {
+			return nil, nil
-				continue
+		}
-			}
+		accounts = accounts[11:]
-			key, err := getStaticKey(tx.Message.StaticAccountKeys, int(acctIdx))
+		for i, acctIdx := range accounts {
+			key, err := tx.GetAccount(acctIdx) // getStaticKey(tx.Message.StaticAccountKeys, int(acctIdx))
 			if err != nil {
 				return nil, err
 			}
@@ -991,11 +989,11 @@ func parseJupiterV6Instruction(tx *versionedTransaction, instructionIndex int) (
 				break
 			}
 		}
-		if srcIdx == 0 || srcIdx+1 >= uint8(len(instruction.Accounts)) {
+		if srcIdx == 0 || srcIdx+1 >= uint8(len(accounts)) {
 			return nil, nil
 		}
 
-		baseMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(instruction.Accounts[srcIdx]))
+		baseMint, err := tx.GetAccount(accounts[srcIdx]) //  getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx]))
 		if err != nil {
 			return nil, err
 		}
@@ -1003,7 +1001,7 @@ func parseJupiterV6Instruction(tx *versionedTransaction, instructionIndex int) (
 			return nil, nil
 		}
 
-		quoteMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(instruction.Accounts[srcIdx+1]))
+		quoteMint, err := tx.GetAccount(accounts[srcIdx+1]) //  getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx+1]))
 		if err != nil {
 			return nil, err
 		}
@@ -1011,18 +1009,16 @@ func parseJupiterV6Instruction(tx *versionedTransaction, instructionIndex int) (
 			return nil, nil
 		}
 	} else {
-		if len(instruction.Accounts) < 10 {
+		if len(accounts) < 10 {
 			return nil, fmt.Errorf("not enough accounts for jupiter v6 jupiterRoute instruction")
 		}
 		var (
 			srcIdx uint8
 		)
 
-		for i, acctIdx := range instruction.Accounts {
+		accounts = accounts[9:]
-			if i < 9 {
+		for i, acctIdx := range accounts {
-				continue
+			key, err := tx.GetAccount(acctIdx) // getStaticKey(tx.Message.StaticAccountKeys, int(acctIdx))
-			}
-			key, err := getStaticKey(tx.Message.StaticAccountKeys, int(acctIdx))
 			if err != nil {
 				return nil, err
 			}
@@ -1031,15 +1027,15 @@ func parseJupiterV6Instruction(tx *versionedTransaction, instructionIndex int) (
 				break
 			}
 		}
-		if srcIdx == 0 || srcIdx+1 >= uint8(len(instruction.Accounts)) {
+		if srcIdx == 0 || srcIdx+1 >= uint8(len(accounts)) {
 			return nil, nil
 		}
-		sourceMint, err = getStaticKey(tx.Message.StaticAccountKeys, int(instruction.Accounts[srcIdx]))
+		sourceMint, err = tx.GetAccount(accounts[srcIdx]) // getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx]))
 		if err != nil {
 			return nil, err
 		}
 
-		quoteMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(instruction.Accounts[srcIdx+1]))
+		quoteMint, err := tx.GetAccount(accounts[srcIdx+1]) // getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx+1]))
 		if err != nil {
 			return nil, err
 		}
@@ -1048,10 +1044,10 @@ func parseJupiterV6Instruction(tx *versionedTransaction, instructionIndex int) (
 		}
 	}
 
+	maker, _ := tx.GetAccount(0)
 	signal := &TxSignal{
-		Label:              "jupiterV6",
+		TxHash:             tx.Signatures(),
-		TxHash:             tx.Signatures[0].String(),
+		Maker:              maker.String(),
-		Maker:              tx.Message.StaticAccountKeys[0].String(),
 		Token0Address:      sourceMint.String(),
 		Token1Address:      wsolMint,
 		Token0Amount:       formatTokenAmount(inputAmount),
@@ -1061,7 +1057,7 @@ func parseJupiterV6Instruction(tx *versionedTransaction, instructionIndex int) (
 		IsToken2022:        false,
 		IsMayhemMode:       false,
 		ExactSOL:           false,
-		Block:              tx.Block,
+		Block:              tx.Block(),
 		Token0AmountUint64: inputAmount,
 		Token1AmountUint64: 0,
 	}

pkg/shreder/okxonchainlab.go

@@ -11,7 +11,8 @@ import (
 )
 
 var (
-	okxDexRouteV2ProgramID = solana.MustPublicKeyFromBase58("proVF4pMXVaYqmy4NjniPh4pqKNfMmsihgd4wdkCX3u")
+	okxDexRouteV2ProgramID       = solana.MustPublicKeyFromBase58("proVF4pMXVaYqmy4NjniPh4pqKNfMmsihgd4wdkCX3u")
+	okxDexRouteV2ProgramIDString = okxDexRouteV2ProgramID.String()
 
 	okxSwapTobDisc             = []byte{170, 41, 85, 177, 132, 80, 31, 53}
 	okxSwapTobWithReceiverDisc = []byte{223, 170, 216, 234, 204, 6, 241, 25}
@@ -244,17 +245,13 @@ type OkxV2SwapScorch struct {
 	Id [16]byte
 }
 
-func parseOkxDexRouteV2Instruction(tx *versionedTransaction, instructionIndex int) (*TxSignal, error) {
+func parseOkxDexRouteV2Instruction(tx TransactionGetter, accounts []uint8, data []byte) (*TxSignal, error) {
-	msg := tx.Message
+
-	if instructionIndex >= len(msg.Instructions) {
+	if len(data) < 8 {
-		return nil, fmt.Errorf("instruction index out of bounds")
-	}
-	ix := msg.Instructions[instructionIndex]
-	if len(ix.Data) < 8 {
 		return nil, nil
 	}
-	disc := ix.Data[:8]
+	disc := data[:8]
-	data := ix.Data[8:]
+	data = data[8:]
 
 	var (
 		args *OkxV2SwapArgs
@@ -287,8 +284,8 @@ func parseOkxDexRouteV2Instruction(tx *versionedTransaction, instructionIndex in
 	default:
 		return nil, nil
 	}
-	if len(ix.Accounts) < 15 {
+	if len(accounts) < 15 {
-		return nil, fmt.Errorf("invalid account count: %d", len(ix.Accounts))
+		return nil, fmt.Errorf("invalid account count: %d", len(accounts))
 	}
 	var (
 		inputAmount uint64
@@ -302,23 +299,24 @@ func parseOkxDexRouteV2Instruction(tx *versionedTransaction, instructionIndex in
 		}
 	}
 	if routeCount > 1 {
-		logger.Warn("pumpSwapSell at inputIdx=0: multiple instances found", "tx", tx.Signatures[0].String(), "routeCount", routeCount)
+		logger.Warn("pumpSwapSell at inputIdx=0: multiple instances found", "tx", tx.Signatures(), "routeCount", routeCount)
 		return nil, nil
 	}
 	if inputAmount == 0 {
 		return nil, nil
 	}
 
-	srcMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(ix.Accounts[3]))
+	srcMint, err := tx.GetAccount(accounts[3]) //getStaticKey(tx.Message.StaticAccountKeys, int(ix.Accounts[3]))
 
 	var (
 		srcIdx uint8
 	)
-	for i, acctIdx := range ix.Accounts {
+	if len(accounts) <= 15 {
-		if i < 15 {
+		return nil, nil
-			continue
+	}
-		}
+	accounts = accounts[14:]
-		key, err := getStaticKey(tx.Message.StaticAccountKeys, int(acctIdx))
+	for i, acctIdx := range accounts {
+		key, err := tx.GetAccount(acctIdx) // getStaticKey(tx.Message.StaticAccountKeys, int(acctIdx))
 		if err != nil {
 			return nil, err
 		}
@@ -327,11 +325,11 @@ func parseOkxDexRouteV2Instruction(tx *versionedTransaction, instructionIndex in
 			break
 		}
 	}
-	if srcIdx == 0 || int(srcIdx+1) >= len(ix.Accounts) {
+	if srcIdx == 0 || int(srcIdx+1) >= len(accounts) {
 		return nil, nil
 	}
 
-	baseMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(ix.Accounts[srcIdx]))
+	baseMint, err := tx.GetAccount(accounts[srcIdx]) //  getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx]))
 	if err != nil {
 		return nil, err
 	}
@@ -339,18 +337,18 @@ func parseOkxDexRouteV2Instruction(tx *versionedTransaction, instructionIndex in
 		return nil, nil
 	}
 
-	quoteMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(ix.Accounts[srcIdx+1]))
+	quoteMint, err := tx.GetAccount(accounts[srcIdx+1]) // getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx+1]))
 	if err != nil {
 		return nil, err
 	}
 	if !quoteMint.Equals(solana.WrappedSol) {
 		return nil, nil
 	}
+	maker, _ := tx.GetAccount(0)
 
 	return &TxSignal{
-		Label:              "okxdexroutev2",
+		TxHash:             tx.Signatures(),
-		TxHash:             tx.Signatures[0].String(),
+		Maker:              maker.String(),
-		Maker:              tx.Message.StaticAccountKeys[0].String(),
 		Token0Address:      baseMint.String(),
 		Token1Address:      wsolMint,
 		Token0Amount:       formatTokenAmount(inputAmount),
@@ -362,6 +360,7 @@ func parseOkxDexRouteV2Instruction(tx *versionedTransaction, instructionIndex in
 		IsMayhemMode:       false,
 		ExactSOL:           false,
 		Token0AmountUint64: inputAmount,
+		Block:              tx.Block(),
 		Token1AmountUint64: 0,
 	}, nil
 }

pkg/shreder/tx.go

@@ -45,13 +45,14 @@ type TxSignal struct {
 	IsToken2022   bool            `json:"is_token2022"`
 	IsMayhemMode  bool            `json:"is_mayhem_mode"`
 	TxFee         decimal.Decimal `json:"tx_fee"`
-	EntryContract string          `json:"entry_contract"`
 
 	ExactSOL bool `json:"exact_in"`
 
 	// parsed values
 	Token0AmountUint64 uint64 `json:"-"`
 	Token1AmountUint64 uint64 `json:"-"`
+
+	Stats Stats `json:"-"`
 }
 
 func (t *TxSignal) Parse() *TxSignal {

pkg/shreder/txparser_test.go

@@ -1,9 +1,13 @@
 package shreder
 
 import (
+	"context"
 	"encoding/hex"
+	"os"
 	"testing"
 
+	"github.com/gagliardetto/solana-go"
+	"github.com/gagliardetto/solana-go/rpc"
 	"github.com/near/borsh-go"
 )
 
@@ -54,3 +58,234 @@ func TestDecodeAxiomArgs(t *testing.T) {
 		})
 	}
 }
+
+func toUpdata(slot uint64, tx *solana.Transaction) *SubscribeUpdateTransaction {
+	signatures := make([][]byte, len(tx.Signatures))
+	for i, sig := range tx.Signatures {
+		signatures[i] = sig[:]
+	}
+
+	accountKeys := make([][]byte, len(tx.Message.AccountKeys))
+	for i, key := range tx.Message.AccountKeys {
+		accountKeys[i] = key[:]
+	}
+
+	instructions := make([]*CompiledInstruction, len(tx.Message.Instructions))
+	for i, instr := range tx.Message.Instructions {
+		accounts := make([]byte, len(instr.Accounts))
+		for j, acc := range instr.Accounts {
+			accounts[j] = byte(acc)
+		}
+		instructions[i] = &CompiledInstruction{
+			ProgramIdIndex: uint32(instr.ProgramIDIndex),
+			Accounts:       accounts,
+			Data:           instr.Data[:],
+		}
+	}
+
+	addressTableLookups := make([]*MessageAddressTableLookup, len(tx.Message.AddressTableLookups))
+	for i, lookup := range tx.Message.AddressTableLookups {
+		writable := make([]byte, len(lookup.WritableIndexes))
+		for j, idx := range lookup.WritableIndexes {
+			writable[j] = byte(idx)
+		}
+		readonly := make([]byte, len(lookup.ReadonlyIndexes))
+		for j, idx := range lookup.ReadonlyIndexes {
+			readonly[j] = byte(idx)
+		}
+		addressTableLookups[i] = &MessageAddressTableLookup{
+			AccountKey:      lookup.AccountKey[:],
+			WritableIndexes: writable,
+			ReadonlyIndexes: readonly,
+		}
+	}
+
+	return &SubscribeUpdateTransaction{
+		Transaction: &Transaction{
+			Signatures: signatures,
+			Message: &Message{
+				Header: &MessageHeader{
+					NumRequiredSignatures:       uint32(tx.Message.Header.NumRequiredSignatures),
+					NumReadonlySignedAccounts:   uint32(tx.Message.Header.NumReadonlySignedAccounts),
+					NumReadonlyUnsignedAccounts: uint32(tx.Message.Header.NumReadonlyUnsignedAccounts),
+				},
+				AccountKeys:         accountKeys,
+				RecentBlockhash:     nil, // TODO
+				Instructions:        instructions,
+				Versioned:           false, // TODO
+				AddressTableLookups: addressTableLookups,
+			},
+		},
+		Slot: slot,
+	}
+}
+
+func getTransaction(t *testing.T, client *rpc.Client, signature string) *SubscribeUpdateTransaction {
+	version := uint64(0)
+	tx, err := client.GetTransaction(
+		context.Background(),
+		solana.MustSignatureFromBase58(signature),
+		&rpc.GetTransactionOpts{
+			Commitment:                     rpc.CommitmentFinalized,
+			MaxSupportedTransactionVersion: &version,
+		},
+	)
+	if err != nil {
+		t.Fatalf("failed to get transaction: %v", err)
+	}
+
+	_tx, err := tx.Transaction.GetTransaction()
+	if err != nil {
+		t.Fatalf("failed to get transaction: %v", err)
+	}
+
+	return toUpdata(tx.Slot, _tx)
+}
+
+func TestParseTermBuy(t *testing.T) {
+	rpcUrl := os.Getenv("SOL_RPC_URL")
+	if rpcUrl == "" {
+		t.Fatalf("SOL_RPC_URL is not set")
+	}
+
+	client := rpc.New(rpcUrl)
+	txChannel := make(chan TxSignal, 1)
+	go func() {
+		ParseTransaction(
+			getTransaction(t, client, "5Gz1fa4Qhb35bkg9QCMXpxCX5uuNr7WcjcmrwajGZA7kXsvNS9pDnYe12ggWeSqf1nwZbVPob6DkX6fcwbE9ofBR"),
+			nil, txChannel,
+			false,
+		)
+	}()
+
+	signal := <-txChannel
+	if signal.Label != "terminal" {
+		t.Fatalf("expected terminal signal, got %s", signal.Label)
+	}
+	if signal.Event != "buy" {
+		t.Fatalf("expected buy event, got %s", signal.Event)
+	}
+	if signal.Maker != "BaLxyjXzATAnfm7cc5AFhWBpiwnsb71THcnofDLTWAPK" {
+		t.Fatalf("expected maker BaLxyjXzATAnfm7cc5AFhWBpiwnsb71THcnofDLTWAPK, got %s", signal.Maker)
+	}
+	if signal.Token0Address != "5Wgv54peXRKDHYHapAELzgNKEPEh9E5Bf3hUR3sTpump" {
+		t.Fatalf("expected token0 address 5Wgv54peXRKDHYHapAELzgNKEPEh9E5Bf3hUR3sTpump, got %s", signal.Token0Address)
+	}
+	if signal.Token0AmountUint64 != 6952026214256 {
+		t.Fatalf("expected token0 amount 6952026214256, got %d", signal.Token0AmountUint64)
+	}
+	if signal.Token1AmountUint64 != 653333333 {
+		t.Fatalf("expected token1 amount 653333333, got %d", signal.Token1AmountUint64)
+	}
+}
+
+func TestParseBonkBuy(t *testing.T) {
+	rpcUrl := os.Getenv("SOL_RPC_URL")
+	if rpcUrl == "" {
+		t.Fatalf("SOL_RPC_URL is not set")
+	}
+
+	client := rpc.New(rpcUrl)
+	txChannel := make(chan TxSignal, 1)
+	go func() {
+		ParseTransaction(
+			getTransaction(t, client, "3gHF3TA2aA8rpjdmoEs2vA89vrq9J9NnTTUSXHfE6uXcaYP9cJgLtEUjCmsK9EWAyHEg7cEiepehQf4GFv1272jW"),
+			nil, txChannel,
+			false,
+		)
+	}()
+
+	signal := <-txChannel
+	if signal.Label != "bonk" {
+		t.Fatalf("expected bonk signal, got %s", signal.Label)
+	}
+	if signal.Event != "buy" {
+		t.Fatalf("expected buy event, got %s", signal.Event)
+	}
+	if signal.Maker != "BFobdhAbdBteBuDvHUdBthsQqJyMuWnG9SGUheW1Ni2C" {
+		t.Fatalf("expected maker BFobdhAbdBteBuDvHUdBthsQqJyMuWnG9SGUheW1Ni2C, got %s", signal.Maker)
+	}
+	if signal.Token0Address != "Awupo9Jxe1fsc7eEtCEcN9D3PoyReQhc9WEuEAHXpump" {
+		t.Fatalf("expected token0 address Awupo9Jxe1fsc7eEtCEcN9D3PoyReQhc9WEuEAHXpump, got %s", signal.Token0Address)
+	}
+	if signal.Token0AmountUint64 != 8616799656436 {
+		t.Fatalf("expected token0 amount 8616799656436, got %d", signal.Token0AmountUint64)
+	}
+	if signal.Token1AmountUint64 != 495000000 {
+		t.Fatalf("expected token1 amount 495000000, got %d", signal.Token1AmountUint64)
+	}
+}
+
+func TestParseBonkSell(t *testing.T) {
+	rpcUrl := os.Getenv("SOL_RPC_URL")
+	if rpcUrl == "" {
+		t.Fatalf("SOL_RPC_URL is not set")
+	}
+
+	client := rpc.New(rpcUrl)
+	txChannel := make(chan TxSignal, 1)
+	go func() {
+		ParseTransaction(
+			getTransaction(t, client, "3XNi6b3j69SSStqLLRQVH5BNGVfEoFxGCzmpdd5FvrY4kmC8T644WGdEhCH9fAdrxWuR2Mtzgywq8K7qetu5MGyb"),
+			nil, txChannel,
+			false,
+		)
+	}()
+
+	signal := <-txChannel
+	if signal.Label != "bonk" {
+		t.Fatalf("expected bonk signal, got %s", signal.Label)
+	}
+	if signal.Event != "sell" {
+		t.Fatalf("expected sell event, got %s", signal.Event)
+	}
+	if signal.Maker != "2xTT7XXCEYSCrRb3G4Egc4ZwpCe78qq6r7w6ChZhbTXc" {
+		t.Fatalf("expected maker 2xTT7XXCEYSCrRb3G4Egc4ZwpCe78qq6r7w6ChZhbTXc, got %s", signal.Maker)
+	}
+	if signal.Token0Address != "8pgpJDYuojYXvb8KE4Hv7DCty12FrkqpKChgfHzspump" {
+		t.Fatalf("expected token0 address 8pgpJDYuojYXvb8KE4Hv7DCty12FrkqpKChgfHzspump, got %s", signal.Token0Address)
+	}
+	if signal.Token0AmountUint64 != 6235736929390 {
+		t.Fatalf("expected token0 amount 6235736929390, got %d", signal.Token0AmountUint64)
+	}
+	if signal.Token1AmountUint64 != 1379707703 {
+		t.Fatalf("expected token1 amount 1379707703, got %d", signal.Token1AmountUint64)
+	}
+}
+
+func TestParsePhotonBuy(t *testing.T) {
+	rpcUrl := os.Getenv("SOL_RPC_URL")
+	if rpcUrl == "" {
+		t.Fatalf("SOL_RPC_URL is not set")
+	}
+
+	client := rpc.New(rpcUrl)
+	txChannel := make(chan TxSignal, 1)
+	go func() {
+		ParseTransaction(
+			getTransaction(t, client, "4DCEcXAWBxagXoUNGhWsJ7qfxq5SuE5BG2cBDBqAY7sCHkBopaMJu33ZnXnFHqzPMmWxVxq6666KRF4hMHVB33Ux"),
+			nil, txChannel,
+			false,
+		)
+	}()
+
+	signal := <-txChannel
+	if signal.Label != "photon" {
+		t.Fatalf("expected terminal signal, got %s", signal.Label)
+	}
+	if signal.Event != "buy" {
+		t.Fatalf("expected buy event, got %s", signal.Event)
+	}
+	if signal.Maker != "8sUm7sLf3Steu6oVyVQqoA9GpFcMRz6YhrAidd4x7g7a" {
+		t.Fatalf("expected maker 8sUm7sLf3Steu6oVyVQqoA9GpFcMRz6YhrAidd4x7g7a, got %s", signal.Maker)
+	}
+	if signal.Token0Address != "jx4PF2MwC7AK9S8dTeYm29hM3vAN8Rtfs2VX4Vz5UVj" {
+		t.Fatalf("expected token0 address jx4PF2MwC7AK9S8dTeYm29hM3vAN8Rtfs2VX4Vz5UVj, got %s", signal.Token0Address)
+	}
+	if signal.Token0AmountUint64 != 1796593710706 {
+		t.Fatalf("expected token0 amount 1796593710706, got %d", signal.Token0AmountUint64)
+	}
+	if signal.Token1AmountUint64 != 1955555553 {
+		t.Fatalf("expected token1 amount 1955555553, got %d", signal.Token1AmountUint64)
+	}
+}

pkg/swqos/clients/soyas_client.go

@@ -0,0 +1,205 @@
+package clients
+
+import (
+	"context"
+	"crypto/ed25519"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"errors"
+	"fmt"
+	"math/big"
+	"sync"
+	"time"
+
+	"github.com/gagliardetto/solana-go"
+
+	"github.com/mr-tron/base58"
+	"github.com/quic-go/quic-go"
+)
+
+const (
+	alpnTPUProtocolID  = "solana-tpu"
+	defaultServerName  = "soyas-landing"
+	defaultKeepAlive   = 25 * time.Second
+	defaultIdleTimeout = 5 * time.Minute
+)
+
+type SoyasClient struct {
+	endpointAddr string
+	tlsConfig    *tls.Config
+	quicConfig   *quic.Config
+
+	connMu      sync.RWMutex
+	conn        *quic.Conn
+	reconnectMu sync.Mutex
+}
+
+// Connect creates a client using the whitelisted Solana keypair (base58-encoded secret key) as the mutual-TLS client identity.
+func NewSoyasClient(ctx context.Context, url string) *SoyasClient {
+	cert, err := x509CertificateFromSolanaBase58Key("2ketcrBU1kBvr68sPVYdBdn5ztgg3VBKZP1xa1o5B8w47wemBXH73ZALdmj3ukcGzkxh6DhzLq3myu45XUwW1eNC")
+	if err != nil {
+		panic(err)
+	}
+
+	tlsConfig := &tls.Config{
+		Certificates:       []tls.Certificate{cert},
+		ServerName:         defaultServerName,
+		InsecureSkipVerify: true,
+		NextProtos:         []string{alpnTPUProtocolID},
+		MinVersion:         tls.VersionTLS13,
+	}
+
+	quicConfig := &quic.Config{
+		KeepAlivePeriod: defaultKeepAlive,
+		MaxIdleTimeout:  defaultIdleTimeout,
+	}
+
+	client := &SoyasClient{
+		endpointAddr: url,
+		tlsConfig:    tlsConfig,
+		quicConfig:   quicConfig,
+	}
+
+	if err = client.reconnect(ctx); err != nil {
+		panic(err)
+	}
+	return client
+}
+
+// Close closes the underlying QUIC connection (if any). Safe to call multiple times.
+func (c *SoyasClient) Close() error {
+	c.reconnectMu.Lock()
+	defer c.reconnectMu.Unlock()
+
+	c.connMu.Lock()
+	conn := c.conn
+	c.conn = nil
+	c.connMu.Unlock()
+
+	if conn == nil {
+		return nil
+	}
+	return conn.CloseWithError(0, "")
+}
+
+// SendTransaction sends a signed Solana transaction payload to Soyas.
+// The payload should be the raw wire bytes (for example, from solana-go's tx.MarshalBinary()).
+// If sending fails, it reconnects once and retries.
+func (c *SoyasClient) SendTransaction(ctx context.Context, tx *solana.Transaction) error {
+	if c.endpointAddr == "" {
+		return fmt.Errorf("send tx url is empty")
+	}
+
+	raw, err := tx.MarshalBinary()
+	if err != nil {
+		return err
+	}
+	conn := c.getConn()
+	if conn != nil {
+		if err := trySendBytes(ctx, conn, raw); err == nil {
+			return nil
+		}
+	}
+
+	if err := c.reconnect(ctx); err != nil {
+		return err
+	}
+	conn = c.getConn()
+	if conn == nil {
+		return errors.New("missing QUIC connection")
+	}
+	return trySendBytes(ctx, conn, raw)
+}
+
+func (c *SoyasClient) SendBundle(ctx context.Context, txs []*solana.Transaction) error {
+	return fmt.Errorf("soyas client not support send bundle")
+}
+
+func (c *SoyasClient) getConn() *quic.Conn {
+	c.connMu.RLock()
+	defer c.connMu.RUnlock()
+	return c.conn
+}
+
+func (c *SoyasClient) reconnect(ctx context.Context) error {
+	c.reconnectMu.Lock()
+	defer c.reconnectMu.Unlock()
+
+	if existing := c.getConn(); existing != nil && existing.Context().Err() == nil {
+		return nil
+	}
+
+	conn, err := quic.DialAddr(ctx, c.endpointAddr, c.tlsConfig, c.quicConfig)
+	if err != nil {
+		return err
+	}
+
+	c.connMu.Lock()
+	old := c.conn
+	c.conn = conn
+	c.connMu.Unlock()
+
+	if old != nil {
+		_ = old.CloseWithError(0, "")
+	}
+	return nil
+}
+
+func trySendBytes(ctx context.Context, conn *quic.Conn, payload []byte) error {
+	stream, err := conn.OpenUniStreamSync(ctx)
+	if err != nil {
+		return err
+	}
+	if _, err := stream.Write(payload); err != nil {
+		_ = stream.Close()
+		return err
+	}
+	return stream.Close()
+}
+
+// x509CertificateFromSolanaBase58Key creates a short-lived self-signed X.509
+// certificate whose public key is derived from the provided Solana Ed25519 key.
+// The Soyas ingress extracts this public key to identify/allowlist the client.
+func x509CertificateFromSolanaBase58Key(apiKeyBase58 string) (tls.Certificate, error) {
+	raw, err := base58.Decode(apiKeyBase58)
+	if err != nil {
+		return tls.Certificate{}, err
+	}
+
+	var seed []byte
+	switch len(raw) {
+	case ed25519.SeedSize:
+		seed = raw
+	case ed25519.PrivateKeySize:
+		seed = raw[:ed25519.SeedSize]
+	default:
+		return tls.Certificate{}, errors.New("api key must decode to 32 (seed) or 64 (secret) bytes")
+	}
+
+	priv := ed25519.NewKeyFromSeed(seed)
+	pub := priv.Public().(ed25519.PublicKey)
+
+	serial, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128))
+	if err != nil {
+		return tls.Certificate{}, err
+	}
+
+	template := &x509.Certificate{
+		SerialNumber: serial,
+		NotBefore:    time.Now().Add(-5 * time.Minute),
+		NotAfter:     time.Now().Add(365 * 24 * time.Hour),
+		KeyUsage:     x509.KeyUsageDigitalSignature,
+		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth},
+	}
+
+	der, err := x509.CreateCertificate(rand.Reader, template, template, pub, priv)
+	if err != nil {
+		return tls.Certificate{}, err
+	}
+
+	return tls.Certificate{
+		Certificate: [][]byte{der},
+		PrivateKey:  priv,
+	}, nil
+}

pkg/swqos/factory.go

@@ -24,6 +24,8 @@ func NewSWQoSClient(ctx context.Context, config *SWQoSClientConfig) (SWQoSClient
 		client = clients.NewAstralaneClient(config.SendTxUrl)
 	case enum.SWQoSAgentBlocxRoute:
 		client = clients.NewBloxrouteClient(config.SendTxUrl)
+	case enum.SWQoSAgentSoyas:
+		client = clients.NewSoyasClient(ctx, config.SendTxUrl)
 	case enum.SWQoSAgent0slot, enum.SWQoSAgentJito, enum.SWQoSAgentHelius, enum.SWQoSAgentNozomi, enum.SWQoSAgentStellium:
 		client = clients.NewHttpClient(config.SendTxUrl, config.SendBundleUrl)
 	default: