Compare commits
2 Commits
b82b7d9b0e
...
4c0abc5c34
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4c0abc5c34 | ||
|
|
d9aea3e8d7 |
@@ -61,7 +61,7 @@ func main() {
|
||||
},
|
||||
},
|
||||
// TODO: axiom, gmgn, etc.
|
||||
})
|
||||
}, shreder.BlocksStats(false), shreder.LogParsedStats(true), shreder.ShowTableLoaded(false))
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
@@ -75,7 +75,6 @@ func main() {
|
||||
<-exitSignal
|
||||
cancel()
|
||||
}()
|
||||
|
||||
// async read from shreder
|
||||
txCh := make(chan shreder.TxSignalBatch, 1000)
|
||||
go func() {
|
||||
@@ -94,8 +93,8 @@ func main() {
|
||||
case txBatch := <-txCh:
|
||||
//jsonData, _ := json.MarshalIndent(txBatch, "", " ")
|
||||
for _, tx := range txBatch {
|
||||
if tx.Label == "dflow" {
|
||||
fmt.Println("===============", tx.TxHash, tx.Event, tx.Token0Address, "token:", tx.Token0Amount)
|
||||
if tx.Label == "okxdexroutev2" || tx.Label == "jupiterv6" || tx.Label == "dflow" {
|
||||
fmt.Println("===============", tx.TxHash, tx.Label, tx.Event, tx.Token0Address, "token:", tx.Token0Amount, "parse time:", tx.ParseEnd.Sub(tx.ParseStart))
|
||||
}
|
||||
}
|
||||
//fmt.Println(txBatch[0].TxHash)
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -54,12 +63,7 @@ func (at *AddressTables) loadAddressTable(tablePubkey solana.PublicKey) ([]solan
|
||||
return addresses, nil
|
||||
|
||||
}
|
||||
|
||||
func (at *AddressTables) GetAddressTable(tablePubkey solana.PublicKey, idx []uint8) []solana.PublicKey {
|
||||
at.mux.RLock()
|
||||
addresses, ok := at.tables.Get(tablePubkey)
|
||||
if !ok {
|
||||
at.mux.RUnlock()
|
||||
func (at *AddressTables) load(tablePubkey solana.PublicKey) {
|
||||
_ = at.pool.Submit(func() {
|
||||
at.loadMux.Lock()
|
||||
_, loading := at.loading[tablePubkey]
|
||||
@@ -82,25 +86,55 @@ func (at *AddressTables) GetAddressTable(tablePubkey solana.PublicKey, idx []uin
|
||||
delete(at.loading, tablePubkey)
|
||||
at.loadMux.Unlock()
|
||||
|
||||
at.mux.Lock()
|
||||
at.tables.Add(tablePubkey, table)
|
||||
total := at.tables.Len()
|
||||
at.mux.Unlock()
|
||||
logger.Info("loadAddressTable", "table", tablePubkey.String(), "table count:", total)
|
||||
at.tables.Add(tablePubkey, &TableInfo{
|
||||
addresses: table,
|
||||
})
|
||||
if at.showTableLoaded {
|
||||
total := at.tables.Len()
|
||||
logger.Info("loadAddressTable", "table", tablePubkey.String(), "table count:", total)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func (at *AddressTables) FillToTx(tx *versionedTransaction, 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.Message.StaticAccountKeys = append(tx.Message.StaticAccountKeys, 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?
|
||||
continue
|
||||
addresses.overErrCount++
|
||||
if addresses.overErrCount > 10 {
|
||||
at.load(tablePubkey)
|
||||
}
|
||||
result = append(result, addresses[i])
|
||||
break
|
||||
}
|
||||
result = append(result, addresses.addresses[i])
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
@@ -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(false))
|
||||
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)
|
||||
@@ -59,6 +119,8 @@ func (c *Client) ReadSync(ctx context.Context, txCh chan<- TxSignalBatch) error
|
||||
return err
|
||||
}
|
||||
|
||||
logger.Debug("subscribing to transactions")
|
||||
|
||||
err = stream.Send(&SubscribeTransactionsRequest{
|
||||
Transactions: c.subscription,
|
||||
})
|
||||
@@ -72,20 +134,38 @@ func (c *Client) ReadSync(ctx context.Context, txCh chan<- TxSignalBatch) error
|
||||
return err
|
||||
}
|
||||
|
||||
txBatch := ParseTransaction(response.Transaction, c.tableLoader)
|
||||
if len(txBatch) == 0 {
|
||||
continue
|
||||
if c.enableBlockStats {
|
||||
slot := response.Transaction.Slot
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
txData := response.Transaction
|
||||
|
||||
err = c.pool.Submit(func() {
|
||||
txBatch := ParseTransaction(txData, c.tableLoader, c.enableParseStats)
|
||||
if len(txBatch) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
// set fixed source for tx signals
|
||||
for _, tx := range txBatch {
|
||||
tx.Source = "shreder"
|
||||
}
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
return
|
||||
case txCh <- txBatch:
|
||||
}
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
|
||||
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}
|
||||
@@ -276,10 +277,11 @@ func parseDFlowInstruction(tx *versionedTransaction, instructionIndex int) (*TxS
|
||||
var (
|
||||
srcIdx uint8
|
||||
)
|
||||
for i, acctIdx := range ix.Accounts {
|
||||
if i < 6 {
|
||||
continue
|
||||
if len(ix.Accounts) <= 6 {
|
||||
return nil, nil
|
||||
}
|
||||
accounts := ix.Accounts[5:]
|
||||
for i, acctIdx := range accounts {
|
||||
key, err := getStaticKey(tx.Message.StaticAccountKeys, int(acctIdx))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -289,15 +291,15 @@ func parseDFlowInstruction(tx *versionedTransaction, instructionIndex int) (*TxS
|
||||
break
|
||||
}
|
||||
}
|
||||
if srcIdx == 0 || srcIdx+1 >= uint8(len(ix.Accounts)) {
|
||||
if srcIdx == 0 || srcIdx+1 >= uint8(len(accounts)) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
baseMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(ix.Accounts[srcIdx]))
|
||||
baseMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx]))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
quoteMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(ix.Accounts[srcIdx+1]))
|
||||
quoteMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx+1]))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -307,7 +309,6 @@ func parseDFlowInstruction(tx *versionedTransaction, instructionIndex int) (*TxS
|
||||
|
||||
// Build TxSignal
|
||||
sig := &TxSignal{
|
||||
Label: "dflow",
|
||||
TxHash: tx.Signatures[0].String(),
|
||||
Maker: tx.Message.StaticAccountKeys[0].String(),
|
||||
Program: "PumpAMM",
|
||||
|
||||
@@ -935,10 +935,11 @@ func parseJupiterV6Instruction(tx *versionedTransaction, instructionIndex int) (
|
||||
var (
|
||||
srcIdx uint8
|
||||
)
|
||||
for i, acctIdx := range instruction.Accounts {
|
||||
if i < 9 {
|
||||
continue
|
||||
if len(instruction.Accounts) <= 9 {
|
||||
return nil, nil
|
||||
}
|
||||
accounts := instruction.Accounts[8:]
|
||||
for i, acctIdx := range accounts {
|
||||
key, err := getStaticKey(tx.Message.StaticAccountKeys, int(acctIdx))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -948,18 +949,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 := 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 := getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx+1]))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -978,10 +979,11 @@ func parseJupiterV6Instruction(tx *versionedTransaction, instructionIndex int) (
|
||||
var (
|
||||
srcIdx uint8
|
||||
)
|
||||
for i, acctIdx := range instruction.Accounts {
|
||||
if i < 12 {
|
||||
continue
|
||||
if len(instruction.Accounts) <= 12 {
|
||||
return nil, nil
|
||||
}
|
||||
accounts := instruction.Accounts[11:]
|
||||
for i, acctIdx := range accounts {
|
||||
key, err := getStaticKey(tx.Message.StaticAccountKeys, int(acctIdx))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -991,11 +993,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 := getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx]))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -1003,7 +1005,7 @@ func parseJupiterV6Instruction(tx *versionedTransaction, instructionIndex int) (
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
quoteMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(instruction.Accounts[srcIdx+1]))
|
||||
quoteMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx+1]))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -1018,10 +1020,8 @@ func parseJupiterV6Instruction(tx *versionedTransaction, instructionIndex int) (
|
||||
srcIdx uint8
|
||||
)
|
||||
|
||||
for i, acctIdx := range instruction.Accounts {
|
||||
if i < 9 {
|
||||
continue
|
||||
}
|
||||
accounts := instruction.Accounts[9:]
|
||||
for i, acctIdx := range accounts {
|
||||
key, err := getStaticKey(tx.Message.StaticAccountKeys, int(acctIdx))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -1031,15 +1031,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 = 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 := getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx+1]))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -1049,7 +1049,6 @@ func parseJupiterV6Instruction(tx *versionedTransaction, instructionIndex int) (
|
||||
}
|
||||
|
||||
signal := &TxSignal{
|
||||
Label: "jupiterV6",
|
||||
TxHash: tx.Signatures[0].String(),
|
||||
Maker: tx.Message.StaticAccountKeys[0].String(),
|
||||
Token0Address: sourceMint.String(),
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
|
||||
var (
|
||||
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}
|
||||
@@ -314,10 +315,11 @@ func parseOkxDexRouteV2Instruction(tx *versionedTransaction, instructionIndex in
|
||||
var (
|
||||
srcIdx uint8
|
||||
)
|
||||
for i, acctIdx := range ix.Accounts {
|
||||
if i < 15 {
|
||||
continue
|
||||
if len(ix.Accounts) <= 15 {
|
||||
return nil, nil
|
||||
}
|
||||
accounts := ix.Accounts[14:]
|
||||
for i, acctIdx := range accounts {
|
||||
key, err := getStaticKey(tx.Message.StaticAccountKeys, int(acctIdx))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -327,11 +329,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 := getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx]))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -339,7 +341,7 @@ func parseOkxDexRouteV2Instruction(tx *versionedTransaction, instructionIndex in
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
quoteMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(ix.Accounts[srcIdx+1]))
|
||||
quoteMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx+1]))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -348,7 +350,6 @@ func parseOkxDexRouteV2Instruction(tx *versionedTransaction, instructionIndex in
|
||||
}
|
||||
|
||||
return &TxSignal{
|
||||
Label: "okxdexroutev2",
|
||||
TxHash: tx.Signatures[0].String(),
|
||||
Maker: tx.Message.StaticAccountKeys[0].String(),
|
||||
Token0Address: baseMint.String(),
|
||||
|
||||
@@ -52,6 +52,9 @@ type TxSignal struct {
|
||||
// parsed values
|
||||
Token0AmountUint64 uint64 `json:"-"`
|
||||
Token1AmountUint64 uint64 `json:"-"`
|
||||
|
||||
ParseStart time.Time `json:"parse_start"`
|
||||
ParseEnd time.Time `json:"parse_end"`
|
||||
}
|
||||
|
||||
func (t *TxSignal) Parse() *TxSignal {
|
||||
|
||||
@@ -6,6 +6,8 @@ import (
|
||||
"fmt"
|
||||
"math/big"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/gagliardetto/solana-go"
|
||||
"github.com/mr-tron/base58"
|
||||
@@ -21,30 +23,39 @@ const (
|
||||
// program ids
|
||||
var (
|
||||
pumpProgramID = solana.MustPublicKeyFromBase58("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P")
|
||||
|
||||
pumpProgramIDString = pumpProgramID.String()
|
||||
// has no sell function with pump and pump.amm program
|
||||
azczProgramID = solana.MustPublicKeyFromBase58("AzcZqCRUQgKEg5FTAgY7JacATABEYCEfMbjXEzspLYFB")
|
||||
azczProgramIDString = azczProgramID.String()
|
||||
|
||||
// only buy function with pump program
|
||||
f5tfProgramID = solana.MustPublicKeyFromBase58("F5tfvbLog9VdGUPqBDTT8rgXvTTcq7e5UiGnupL1zvBq")
|
||||
|
||||
f5tfProgramIDString = f5tfProgramID.String()
|
||||
// only pump.fun function
|
||||
photonProgramID = solana.MustPublicKeyFromBase58("BSfD6SHZigAfDWSjzD5Q41jw8LmKwtmjskPH9XW1mrRW")
|
||||
photonProgramIDString = photonProgramID.String()
|
||||
|
||||
pumpAmmProgramID = solana.MustPublicKeyFromBase58("pAMMBay6oceH9fJKBRHGP5D4bD4sWpmSwMn52FMfXEA")
|
||||
pumpAmmProgramIDString = pumpAmmProgramID.String()
|
||||
|
||||
boboProgramID = solana.MustPublicKeyFromBase58("BobogA5N2KN2GG4XN3E3rNNRw3L8H1QPXp7QLxGrNHGM")
|
||||
boboProgramIDString = boboProgramID.String()
|
||||
|
||||
qtkvProgramID = solana.MustPublicKeyFromBase58("qtkvapJEvRWWrB7i5K6RaA1kvq5x3qmMKZ98ad71XQ7")
|
||||
qtkvProgramIDString = qtkvProgramID.String()
|
||||
|
||||
// only buy function with pump program
|
||||
fjszProgramID = solana.MustPublicKeyFromBase58("FJsZbftBqRLfF7uqUKpm4s2goDr6xsQ5Q3mN7AFJB6hK")
|
||||
fjszProgramIDString = fjszProgramID.String()
|
||||
|
||||
flasProgramID = solana.MustPublicKeyFromBase58("FLASHX8DrLbgeR8FcfNV1F5krxYcYMUdBkrP1EPBtxB9")
|
||||
flasProgramIDString = flasProgramID.String()
|
||||
|
||||
terminalProgramID = solana.MustPublicKeyFromBase58("term9YPb9mzAsABaqN71A4xdbxHmpBNZavpBiQKZzN3")
|
||||
terminalProgramIDString = terminalProgramID.String()
|
||||
|
||||
jupiterV6ProgramID = solana.MustPublicKeyFromBase58("JUP6LkbZbjS1jKKwapdHNy74zcZ3tLUZoi5QNyVTaV4")
|
||||
jupiterV6ProgramIDString = jupiterV6ProgramID.String()
|
||||
)
|
||||
|
||||
type AccountNotFoundError struct {
|
||||
@@ -118,6 +129,7 @@ type versionedTransaction struct {
|
||||
Signatures []solana.Signature
|
||||
Message versionedMessage
|
||||
Block uint64
|
||||
Time time.Time
|
||||
}
|
||||
|
||||
type pumpExtendedSellArgs struct {
|
||||
@@ -192,15 +204,72 @@ type fjszBuyArgs struct {
|
||||
TokenAmount uint64
|
||||
}
|
||||
|
||||
var (
|
||||
versionedPool = sync.Pool{}
|
||||
|
||||
accIdxPool = sync.Pool{}
|
||||
)
|
||||
|
||||
func requireAccIdxSlice() []uint8 {
|
||||
v := accIdxPool.Get()
|
||||
if v == nil {
|
||||
return make([]uint8, 0, 16)
|
||||
}
|
||||
return v.([]uint8)
|
||||
}
|
||||
|
||||
func releaseAccIdxSlice(s []uint8) {
|
||||
if s == nil {
|
||||
return
|
||||
}
|
||||
s = s[:0]
|
||||
accIdxPool.Put(s)
|
||||
}
|
||||
|
||||
func requireVersionedPool() *versionedTransaction {
|
||||
v := versionedPool.Get()
|
||||
if v == nil {
|
||||
return &versionedTransaction{
|
||||
Signatures: make([]solana.Signature, 0, 10),
|
||||
Message: versionedMessage{
|
||||
StaticAccountKeys: make([]solana.PublicKey, 0, 256),
|
||||
Instructions: make([]compiledInstruction, 0, 16),
|
||||
AddressTableLookups: make([]addressTableLookup, 0, 10),
|
||||
},
|
||||
}
|
||||
}
|
||||
return v.(*versionedTransaction)
|
||||
}
|
||||
|
||||
func releaseVersionedPool(v *versionedTransaction) {
|
||||
if v == nil {
|
||||
return
|
||||
}
|
||||
for i := range v.Message.Instructions {
|
||||
releaseAccIdxSlice(v.Message.Instructions[i].Accounts)
|
||||
}
|
||||
for i := range v.Message.AddressTableLookups {
|
||||
releaseAccIdxSlice(v.Message.AddressTableLookups[i].WritableIndexes)
|
||||
releaseAccIdxSlice(v.Message.AddressTableLookups[i].ReadonlyIndexes)
|
||||
}
|
||||
versionedPool.Put(v)
|
||||
}
|
||||
|
||||
// ParseTransaction mirrors the Rust parse_transaction entry point.
|
||||
func ParseTransaction(update *SubscribeUpdateTransaction, loader *AddressTables) []*TxSignal {
|
||||
func ParseTransaction(update *SubscribeUpdateTransaction, loader *AddressTables, stats bool) []*TxSignal {
|
||||
var now time.Time
|
||||
if stats {
|
||||
now = time.Now()
|
||||
}
|
||||
versioned, err := toVersionedTransaction(update)
|
||||
if err != nil || versioned == nil || len(versioned.Signatures) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
defer func() {
|
||||
releaseVersionedPool(versioned)
|
||||
}()
|
||||
txHash := versioned.Signatures[0]
|
||||
staticKeys := versioned.Message.StaticAccountKeys
|
||||
// staticKeys := versioned.Message.StaticAccountKeys
|
||||
instructions := versioned.Message.Instructions
|
||||
|
||||
if loader != nil && len(versioned.Message.AddressTableLookups) > 0 {
|
||||
@@ -209,85 +278,81 @@ func ParseTransaction(update *SubscribeUpdateTransaction, loader *AddressTables)
|
||||
if len(lookup.WritableIndexes) == 0 {
|
||||
continue
|
||||
}
|
||||
accounts := loader.GetAddressTable(lookup.AccountKey, lookup.WritableIndexes)
|
||||
if len(accounts) != len(lookup.WritableIndexes) {
|
||||
lookupTableOk = false
|
||||
lookupTableOk = loader.FillToTx(versioned, lookup.AccountKey, lookup.WritableIndexes)
|
||||
if !lookupTableOk {
|
||||
break
|
||||
}
|
||||
staticKeys = append(staticKeys, accounts...)
|
||||
|
||||
}
|
||||
if lookupTableOk {
|
||||
for _, lookup := range versioned.Message.AddressTableLookups {
|
||||
if len(lookup.ReadonlyIndexes) == 0 {
|
||||
continue
|
||||
}
|
||||
accounts := loader.GetAddressTable(lookup.AccountKey, lookup.ReadonlyIndexes)
|
||||
if len(accounts) != len(lookup.ReadonlyIndexes) {
|
||||
lookupTableOk = loader.FillToTx(versioned, lookup.AccountKey, lookup.ReadonlyIndexes)
|
||||
if !lookupTableOk {
|
||||
break
|
||||
}
|
||||
staticKeys = append(staticKeys, accounts...)
|
||||
}
|
||||
}
|
||||
versioned.Message.StaticAccountKeys = staticKeys
|
||||
// versioned.Message.StaticAccountKeys = staticKeys
|
||||
}
|
||||
|
||||
var parsed []*TxSignal
|
||||
var parsed []*TxSignal = make([]*TxSignal, 0, 3)
|
||||
|
||||
for i := range instructions {
|
||||
inst := instructions[i]
|
||||
if int(inst.ProgramIDIndex) >= len(staticKeys) {
|
||||
if int(inst.ProgramIDIndex) >= len(versioned.Message.StaticAccountKeys) {
|
||||
continue
|
||||
}
|
||||
|
||||
programID := staticKeys[inst.ProgramIDIndex]
|
||||
programID := versioned.Message.StaticAccountKeys[inst.ProgramIDIndex]
|
||||
switch programID {
|
||||
case pumpProgramID:
|
||||
txRes, err := parsePumpInstruction(versioned, i)
|
||||
parsed = appendParsed(parsed, txRes, err, txHash, "pump", pumpProgramID.String())
|
||||
parsed = appendParsed(now, parsed, txRes, err, txHash, "pump", pumpProgramIDString)
|
||||
case azczProgramID:
|
||||
txRes, err := parseAzczInstruction(versioned, i)
|
||||
parsed = appendParsed(parsed, txRes, err, txHash, "azcz", azczProgramID.String())
|
||||
parsed = appendParsed(now, parsed, txRes, err, txHash, "azcz", azczProgramIDString)
|
||||
case f5tfProgramID:
|
||||
txRes, err := parseF5tfInstruction(versioned, i)
|
||||
parsed = appendParsed(parsed, txRes, err, txHash, "f5tf", f5tfProgramID.String())
|
||||
parsed = appendParsed(now, parsed, txRes, err, txHash, "f5tf", f5tfProgramIDString)
|
||||
case flasProgramID:
|
||||
txRes, err := parseFlasInstruction(versioned, i)
|
||||
parsed = appendParsed(parsed, txRes, err, txHash, "flas", flasProgramID.String())
|
||||
parsed = appendParsed(now, parsed, txRes, err, txHash, "flas", flasProgramIDString)
|
||||
case photonProgramID:
|
||||
txRes, err := parsePhotonInstruction(versioned, i)
|
||||
parsed = appendParsed(parsed, txRes, err, txHash, "photon", photonProgramID.String())
|
||||
parsed = appendParsed(now, parsed, txRes, err, txHash, "photon", photonProgramIDString)
|
||||
case pumpAmmProgramID:
|
||||
txRes, err := parsePumpAmmInstruction(versioned, i)
|
||||
parsed = appendParsed(parsed, txRes, err, txHash, "pumpamm", pumpAmmProgramID.String())
|
||||
parsed = appendParsed(now, parsed, txRes, err, txHash, "pumpamm", pumpAmmProgramIDString)
|
||||
case boboProgramID:
|
||||
txRes, err := parseBoboInstruction(versioned, i)
|
||||
parsed = appendParsed(parsed, txRes, err, txHash, "bobo", boboProgramID.String())
|
||||
parsed = appendParsed(now, parsed, txRes, err, txHash, "bobo", boboProgramIDString)
|
||||
case qtkvProgramID:
|
||||
txRes, err := parseQtkvInstruction(versioned, i)
|
||||
parsed = appendParsed(parsed, txRes, err, txHash, "qtkv", qtkvProgramID.String())
|
||||
parsed = appendParsed(now, parsed, txRes, err, txHash, "qtkv", qtkvProgramIDString)
|
||||
case fjszProgramID:
|
||||
txRes, err := parseFjszInstruction(versioned, i)
|
||||
parsed = appendParsed(parsed, txRes, err, txHash, "fjsz", fjszProgramID.String())
|
||||
parsed = appendParsed(now, parsed, txRes, err, txHash, "fjsz", fjszProgramIDString)
|
||||
case terminalProgramID:
|
||||
txRes, err := parseTermInstruction(versioned, i)
|
||||
parsed = appendParsed(parsed, txRes, err, txHash, "terminal", terminalProgramID.String())
|
||||
parsed = appendParsed(now, parsed, txRes, err, txHash, "terminal", terminalProgramIDString)
|
||||
case jupiterV6ProgramID:
|
||||
txRes, err := parseJupiterV6Instruction(versioned, i)
|
||||
parsed = appendParsed(parsed, txRes, err, txHash, "jupiterv6", jupiterV6ProgramID.String())
|
||||
parsed = appendParsed(now, parsed, txRes, err, txHash, "jupiterv6", jupiterV6ProgramIDString)
|
||||
case okxDexRouteV2ProgramID:
|
||||
txRes, err := parseOkxDexRouteV2Instruction(versioned, i)
|
||||
parsed = appendParsed(parsed, txRes, err, txHash, "okxdexroutev2", okxDexRouteV2ProgramID.String())
|
||||
parsed = appendParsed(now, parsed, txRes, err, txHash, "okxdexroutev2", okxDexRouteV2ProgramIDString)
|
||||
case dflowProgramID:
|
||||
txRes, err := parseDFlowInstruction(versioned, i)
|
||||
parsed = appendParsed(parsed, txRes, err, txHash, "dflow", dflowProgramID.String())
|
||||
parsed = appendParsed(now, parsed, txRes, err, txHash, "dflow", dflowProgramString)
|
||||
}
|
||||
}
|
||||
|
||||
return parsed
|
||||
}
|
||||
|
||||
func appendParsed(list []*TxSignal, parsed *TxSignal, err error, txHash [64]byte, label string, entryContract string) []*TxSignal {
|
||||
func appendParsed(start time.Time, list []*TxSignal, parsed *TxSignal, err error, txHash [64]byte, label string, entryContract string) []*TxSignal {
|
||||
if err != nil {
|
||||
if !strings.HasPrefix(err.Error(), "account index") {
|
||||
logger.Debug("txparser: failed to parse", "label", label, "instruction", err, "tx_hash", base58.Encode(txHash[:]))
|
||||
@@ -296,6 +361,11 @@ func appendParsed(list []*TxSignal, parsed *TxSignal, err error, txHash [64]byte
|
||||
}
|
||||
if parsed != nil {
|
||||
parsed.EntryContract = entryContract
|
||||
parsed.Label = label
|
||||
if !start.IsZero() {
|
||||
parsed.ParseEnd = time.Now()
|
||||
parsed.ParseStart = start
|
||||
}
|
||||
list = append(list, parsed)
|
||||
}
|
||||
return list
|
||||
@@ -308,47 +378,42 @@ func toVersionedTransaction(update *SubscribeUpdateTransaction) (*versionedTrans
|
||||
|
||||
protoTx := update.Transaction
|
||||
msg := protoTx.Message
|
||||
|
||||
signatures := make([]solana.Signature, len(protoTx.Signatures))
|
||||
for i, rawSig := range protoTx.Signatures {
|
||||
signatures[i] = solana.SignatureFromBytes(rawSig)
|
||||
versioned := requireVersionedPool()
|
||||
versioned.Signatures = versioned.Signatures[:0]
|
||||
for _, rawSig := range protoTx.Signatures {
|
||||
versioned.Signatures = append(versioned.Signatures, solana.SignatureFromBytes(rawSig))
|
||||
}
|
||||
|
||||
staticKeys := make([]solana.PublicKey, len(msg.AccountKeys))
|
||||
for i, key := range msg.AccountKeys {
|
||||
staticKeys[i] = solana.PublicKeyFromBytes(key)
|
||||
versioned.Message.StaticAccountKeys = versioned.Message.StaticAccountKeys[:0]
|
||||
for _, key := range msg.AccountKeys {
|
||||
versioned.Message.StaticAccountKeys = append(versioned.Message.StaticAccountKeys, solana.PublicKeyFromBytes(key))
|
||||
}
|
||||
|
||||
instructions := make([]compiledInstruction, len(msg.Instructions))
|
||||
for i, instr := range msg.Instructions {
|
||||
accounts := append([]uint8(nil), instr.Accounts...)
|
||||
instructions[i] = compiledInstruction{
|
||||
versioned.Message.Instructions = versioned.Message.Instructions[:0]
|
||||
for _, instr := range msg.Instructions {
|
||||
accounts := requireAccIdxSlice()
|
||||
accounts = append(accounts, instr.Accounts...)
|
||||
versioned.Message.Instructions = append(versioned.Message.Instructions,
|
||||
compiledInstruction{
|
||||
ProgramIDIndex: uint8(instr.ProgramIdIndex),
|
||||
Accounts: accounts,
|
||||
Data: instr.Data,
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
lookups := make([]addressTableLookup, len(msg.AddressTableLookups))
|
||||
for i, lookup := range msg.AddressTableLookups {
|
||||
writable := append([]uint8(nil), lookup.WritableIndexes...)
|
||||
readonly := append([]uint8(nil), lookup.ReadonlyIndexes...)
|
||||
lookups[i] = addressTableLookup{
|
||||
versioned.Message.AddressTableLookups = versioned.Message.AddressTableLookups[:0]
|
||||
for _, lookup := range msg.AddressTableLookups {
|
||||
writable := requireAccIdxSlice()
|
||||
writable = append(writable, lookup.WritableIndexes...)
|
||||
readonly := requireAccIdxSlice()
|
||||
readonly = append(readonly, lookup.ReadonlyIndexes...)
|
||||
versioned.Message.AddressTableLookups = append(versioned.Message.AddressTableLookups, addressTableLookup{
|
||||
AccountKey: solana.PublicKeyFromBytes(lookup.AccountKey),
|
||||
WritableIndexes: writable,
|
||||
ReadonlyIndexes: readonly,
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
return &versionedTransaction{
|
||||
Signatures: signatures,
|
||||
Message: versionedMessage{
|
||||
StaticAccountKeys: staticKeys,
|
||||
Instructions: instructions,
|
||||
AddressTableLookups: lookups,
|
||||
},
|
||||
Block: update.GetSlot(),
|
||||
}, nil
|
||||
versioned.Block = update.GetSlot()
|
||||
return versioned, nil
|
||||
}
|
||||
|
||||
func formatTokenAmount(amount uint64) decimal.Decimal {
|
||||
|
||||
Reference in New Issue
Block a user