chore: add release pool

cmd/debug_jupv6/main.go

@@ -0,0 +1,62 @@
+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,10 +8,8 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
-	"time"
 
 	"github.com/gagliardetto/solana-go/rpc"
-	"github.com/shopspring/decimal"
 
 	"github.com/samlior/libsam/pkg/shreder"
 )
@@ -78,9 +76,9 @@ func main() {
 		cancel()
 	}()
 	// async read from shreder
-	txCh := make(chan shreder.TxSignal, 1000)
+	txCh := make(chan shreder.TxSignalBatch, 1000)
 	go func() {
-		err := shrederClient.ReadEntriesSync(ctx, txCh)
+		err := shrederClient.ReadSync(ctx, txCh)
 		if err != nil {
 			if !errors.Is(err, context.Canceled) {
 				panic(err)
@@ -92,20 +90,13 @@ func main() {
 		select {
 		case <-ctx.Done():
 			return
-		case tx := <-txCh:
+		case txBatch := <-txCh:
 			//jsonData, _ := json.MarshalIndent(txBatch, "", "  ")
-			if tx.Token0Amount.GreaterThan(decimal.NewFromInt(100)) && (tx.Label == "okxdexroutev2" || tx.Label == "jupiterv6" || tx.Label == "dflow") {
+			for _, tx := range txBatch {
-				fmt.Println(time.Now(), "===============", tx.TxHash,
+				if tx.Label == "okxdexroutev2" || tx.Label == "jupiterv6" || tx.Label == "dflow" {
-					"parse time:", tx.Stats.Done.Sub(tx.Stats.Filter),
+					fmt.Println("===============", tx.TxHash, tx.Label, tx.Event, tx.Token0Address, "token:", tx.Token0Amount, "parse time:", tx.ParseEnd.Sub(tx.ParseStart))
-					"decode time:", tx.Stats.Decoded.Sub(tx.Stats.FEC),
+				}
-					"filter time:", tx.Stats.Filter.Sub(tx.Stats.Decoded),
-					"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

@@ -4,10 +4,13 @@ go 1.25.1
 
 require (
 	github.com/BlockRazorinc/solana-trader-client-go v0.0.0-20250722092120-44561cb37455
+	github.com/gagliardetto/binary v0.8.0
 	github.com/gagliardetto/solana-go v1.12.0
+	github.com/hashicorp/golang-lru/v2 v2.0.7
 	github.com/mr-tron/base58 v1.2.0
 	github.com/near/borsh-go v0.3.2-0.20220516180422-1ff87d108454
 	github.com/panjf2000/ants/v2 v2.11.4
+	github.com/quic-go/quic-go v0.58.0
 	github.com/shopspring/decimal v1.4.0
 	google.golang.org/grpc v1.75.0
 	google.golang.org/protobuf v1.36.10
@@ -19,10 +22,8 @@ require (
 	github.com/blendle/zapdriver v1.3.1 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/fatih/color v1.9.0 // indirect
-	github.com/gagliardetto/binary v0.8.0 // indirect
 	github.com/gagliardetto/treeout v0.1.4 // indirect
 	github.com/google/uuid v1.6.0 // indirect
-	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/compress v1.13.6 // indirect
 	github.com/logrusorgru/aurora v2.0.3+incompatible // indirect
@@ -32,7 +33,6 @@ 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

@@ -88,9 +88,8 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 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/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 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=
@@ -118,6 +117,8 @@ go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
 go.uber.org/goleak v1.1.11 h1:wy28qYRKZgnJTxGxvye5/wgWr1EKjmUDGYox5mGlRlI=
 go.uber.org/goleak v1.1.11/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ=
+go.uber.org/mock v0.5.2 h1:LbtPTcP8A5k9WPXj54PPPbjcI4Y6lhyOZXn+VS7wNko=
+go.uber.org/mock v0.5.2/go.mod h1:wLlUxC2vVTPTaE3UD51E0BGOAElKrILxhVSDYQLld5o=
 go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
 go.uber.org/multierr v1.6.0 h1:y6IPFStTAIT5Ytl7/XYmHvzXQ7S3g/IeZW9hyZ5thw4=
 go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=

pkg/shreder/addresstables.go

@@ -59,6 +59,7 @@ 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
 
 }
@@ -92,11 +93,10 @@ func (at *AddressTables) load(tablePubkey solana.PublicKey) {
 			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 {
+func (at *AddressTables) FillToTx(tx *versionedTransaction, tablePubkey solana.PublicKey, idx []uint8) bool {
 	addresses, ok := at.tables.Get(tablePubkey)
 	if !ok {
 		at.load(tablePubkey)
@@ -112,7 +112,7 @@ func (at *AddressTables) FillToTx(tx FillableTransaction, tablePubkey solana.Pub
 			}
 			return false
 		}
-		tx.FillLookupTable(addresses.addresses[i])
+		tx.Message.StaticAccountKeys = append(tx.Message.StaticAccountKeys, addresses.addresses[i])
 	}
 	return true
 }

pkg/shreder/client.go

@@ -70,7 +70,7 @@ func NewShrederClient(
 
 	poolSize := runtime.NumCPU()*2 + 2
 	logger.Info("creating shreder client", "url", url, "pool_size", poolSize)
-	pool, err := ants.NewPool(poolSize, ants.WithNonblocking(true))
+	pool, err := ants.NewPool(poolSize, ants.WithNonblocking(false))
 	if err != nil {
 		return nil, func() {}, err
 	}
@@ -113,42 +113,7 @@ func (c *Client) Wait() {
 	logger.Debug("shreder client stopped")
 }
 
-func (c *Client) ReadEntriesSync(ctx context.Context, txCh chan<- TxSignal) error {
+func (c *Client) ReadSync(ctx context.Context, txCh chan<- TxSignalBatch) 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
@@ -163,10 +128,14 @@ func (c *Client) ReadSync(ctx context.Context, txCh chan<- TxSignal) error {
 		return err
 	}
 
+	// reboot the pool
+	c.pool.Reboot()
+
 	for {
-		response, err := stream.Recv()
+		var response *SubscribeTransactionsResponse
+		response, err = stream.Recv()
 		if err != nil {
-			return err
+			break
 		}
 
 		if c.enableBlockStats {
@@ -184,11 +153,28 @@ func (c *Client) ReadSync(ctx context.Context, txCh chan<- TxSignal) error {
 		txData := response.Transaction
 
 		err = c.pool.Submit(func() {
-			ParseTransaction(txData, c.tableLoader, txCh, c.enableParseStats)
+			txBatch := ParseTransaction(txData, c.tableLoader, c.enableParseStats)
+			if len(txBatch) == 0 {
+				return
+			}
+
+			for _, tx := range txBatch {
+				tx.Source = "shreder"
+			}
+
+			select {
+			case <-ctx.Done():
+				return
+			case txCh <- txBatch:
+			}
 		})
 		if err != nil {
-			logger.Error("failed to submit transaction: ", "err", err)
+			break
 		}
-
 	}
+
+	// sync waiting for all tasks to complete
+	c.pool.Release()
+
+	return err
 }

pkg/shreder/dflow.go

@@ -232,14 +232,19 @@ func decodeSwap2Params(data []byte) (*dflowSwapParams, error) {
 	return out, nil
 }
 
-func parseDFlowInstruction(tx TransactionGetter, accounts []uint8, data []byte) (*TxSignal, error) {
+func parseDFlowInstruction(tx *versionedTransaction, instructionIndex int) (*TxSignal, error) {
-	if len(data) < 8 {
+	msg := tx.Message
+	if instructionIndex >= len(msg.Instructions) {
+		return nil, fmt.Errorf("instruction index out of bounds")
+	}
+	ix := msg.Instructions[instructionIndex]
+	if len(ix.Data) < 8 {
 		return nil, nil
 	}
 
 	var err error
-	disc := data[:8]
+	disc := ix.Data[:8]
-	payload := data[8:]
+	payload := ix.Data[8:]
 
 	var params *dflowSwapParams
 	switch {
@@ -272,12 +277,12 @@ func parseDFlowInstruction(tx TransactionGetter, accounts []uint8, data []byte)
 	var (
 		srcIdx uint8
 	)
-	if len(accounts) <= 6 {
+	if len(ix.Accounts) <= 6 {
 		return nil, nil
 	}
-	accounts = accounts[5:]
+	accounts := ix.Accounts[5:]
 	for i, acctIdx := range accounts {
-		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
 		}
@@ -290,31 +295,29 @@ func parseDFlowInstruction(tx TransactionGetter, accounts []uint8, data []byte)
 		return nil, nil
 	}
 
-	baseMint, err := tx.GetAccount(accounts[srcIdx]) // getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx]))
+	baseMint, err := 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]))
+	quoteMint, err := 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(),
+		TxHash:             tx.Signatures[0].String(),
-		Maker:              maker.String(),
+		Maker:              tx.Message.StaticAccountKeys[0].String(),
 		Program:            "PumpAMM",
 		Event:              "sell",
 		Token0Address:      baseMint.String(),
 		Token1Address:      wsolMint,
 		Token0Amount:       formatTokenAmount(pump.Amount),
 		Token1Amount:       decimal.Zero,
-		Token0AmountUint64: pump.Amount,
+		Token0AmountUint64: uint64(pump.Amount),
-		Block:              tx.Block(),
 		Token1AmountUint64: 0,
 	}
 	return sig, nil

pkg/shreder/entry.go

@@ -1,334 +0,0 @@
-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,16 +859,21 @@ func pumpSwapSellAtIdx0V2(amount uint64, plan []RoutePlanStepV2) (uint64, int) {
 }
 
 // only decodes inputIdx = 0 container pumpSwap instructions for now
-func parseJupiterV6Instruction(tx TransactionGetter, accounts []uint8, data []byte) (*TxSignal, error) {
+func parseJupiterV6Instruction(tx *versionedTransaction, instructionIndex int) (*TxSignal, error) {
+	msg := tx.Message
+	if instructionIndex >= len(msg.Instructions) {
+		return nil, fmt.Errorf("instruction index out of bounds")
+	}
 
-	if len(data) == 0 {
+	instruction := msg.Instructions[instructionIndex]
+	if len(instruction.Data) == 0 {
 		return nil, fmt.Errorf("data is empty")
 	}
-	if len(data) < 8 {
+	if len(instruction.Data) < 8 {
 		return nil, nil
 	}
 
-	disc := data[:8]
+	disc := instruction.Data[:8]
 
 	var (
 		sourceMint  solana.PublicKey
@@ -881,26 +886,26 @@ func parseJupiterV6Instruction(tx TransactionGetter, accounts []uint8, data []by
 	// route/shared_accounts_* (v1) use different account layouts; we only decode args here.
 	switch {
 	case bytes.Equal(disc, jupiterRouteV2):
-		args, err := decodeJupiterV6RouteV2Arg(data[8:])
+		args, err := decodeJupiterV6RouteV2Arg(instruction.Data[8:])
 		if err != nil {
 			return nil, err
 		}
 		inputAmount, planCount = pumpSwapSellAtIdx0V2(args.In, args.Plan)
 	case bytes.Equal(disc, jupiterSharedAccountsRouteV2):
-		args, err := decodeJupiterV6SharedAccountsRouteV2Arg(data[8:])
+		args, err := decodeJupiterV6SharedAccountsRouteV2Arg(instruction.Data[8:])
 		if err != nil {
 			return nil, err
 		}
 		inputAmount, planCount = pumpSwapSellAtIdx0V2(args.In, args.RoutePlan)
 	case bytes.Equal(disc, jupiterRoute):
-		args, err := decodeJupiterV6RouteArg(data[8:])
+		args, err := decodeJupiterV6RouteArg(instruction.Data[8:])
 		if err != nil {
 			return nil, err
 		}
 		_ = args
 		inputAmount, planCount = pumpSwapSellAtIdx0(args.In, args.Plan)
 	case bytes.Equal(disc, jupiterSharedAccountsRoute):
-		args, err := decodeJupiterV6SharedAccountsRouteArg(data[8:])
+		args, err := decodeJupiterV6SharedAccountsRouteArg(instruction.Data[8:])
 		if err != nil {
 			return nil, err
 		}
@@ -911,8 +916,7 @@ func parseJupiterV6Instruction(tx TransactionGetter, accounts []uint8, data []by
 	}
 	if planCount > 1 {
 		// multiple pumpSwapSell at inputIdx=0? should not happen
-		logger.Warn("pumpSwapSell at inputIdx=0: multiple instances found", "tx", tx.Signatures(), "planCount", planCount)
+		logger.Warn("pumpSwapSell at inputIdx=0: multiple instances found", "tx", tx.Signatures[0].String(), "planCount", planCount)
-		return nil, nil
 	}
 	if inputAmount == 0 {
 		return nil, nil
@@ -920,10 +924,10 @@ func parseJupiterV6Instruction(tx TransactionGetter, accounts []uint8, data []by
 
 	// 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(accounts) < 6 {
+		if len(instruction.Accounts) < 6 {
 			return nil, fmt.Errorf("not enough accounts for jupiter v6 v2 instruction")
 		}
-		sourceMint, err = tx.GetAccount(accounts[3]) //getStaticKey(tx.Message.StaticAccountKeys, int(instruction.Accounts[3]))
+		sourceMint, err = getStaticKey(tx.Message.StaticAccountKeys, int(instruction.Accounts[3]))
 		if err != nil {
 			return nil, err
 		}
@@ -931,12 +935,12 @@ func parseJupiterV6Instruction(tx TransactionGetter, accounts []uint8, data []by
 		var (
 			srcIdx uint8
 		)
-		if len(accounts) <= 9 {
+		if len(instruction.Accounts) <= 9 {
 			return nil, nil
 		}
-		accounts = accounts[8:]
+		accounts := instruction.Accounts[8:]
 		for i, acctIdx := range accounts {
-			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
 			}
@@ -949,14 +953,14 @@ func parseJupiterV6Instruction(tx TransactionGetter, accounts []uint8, data []by
 			return nil, nil
 		}
 
-		baseMint, err := tx.GetAccount(accounts[srcIdx]) // getStaticKey(tx.Message.StaticAccountKeys, int(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 := tx.GetAccount(accounts[srcIdx+1]) // getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx+1]))
+		quoteMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx+1]))
 		if err != nil {
 			return nil, err
 		}
@@ -965,22 +969,22 @@ func parseJupiterV6Instruction(tx TransactionGetter, accounts []uint8, data []by
 		}
 
 	} else if bytes.Equal(disc, jupiterSharedAccountsRoute) {
-		if len(accounts) < 12 {
+		if len(instruction.Accounts) < 12 {
 			return nil, fmt.Errorf("not enough accounts for jupiter v6 jupiterSharedAccountsRoute instruction")
 		}
-		sourceMint, err = tx.GetAccount(accounts[7]) // getStaticKey(tx.Message.StaticAccountKeys, int(instruction.Accounts[7]))
+		sourceMint, err = getStaticKey(tx.Message.StaticAccountKeys, int(instruction.Accounts[7]))
 		if err != nil {
 			return nil, err
 		}
 		var (
 			srcIdx uint8
 		)
-		if len(accounts) <= 12 {
+		if len(instruction.Accounts) <= 12 {
 			return nil, nil
 		}
-		accounts = accounts[11:]
+		accounts := instruction.Accounts[11:]
 		for i, acctIdx := range accounts {
-			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
 			}
@@ -993,7 +997,7 @@ func parseJupiterV6Instruction(tx TransactionGetter, accounts []uint8, data []by
 			return nil, nil
 		}
 
-		baseMint, err := tx.GetAccount(accounts[srcIdx]) //  getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx]))
+		baseMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx]))
 		if err != nil {
 			return nil, err
 		}
@@ -1001,7 +1005,7 @@ func parseJupiterV6Instruction(tx TransactionGetter, accounts []uint8, data []by
 			return nil, nil
 		}
 
-		quoteMint, err := tx.GetAccount(accounts[srcIdx+1]) //  getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx+1]))
+		quoteMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx+1]))
 		if err != nil {
 			return nil, err
 		}
@@ -1009,16 +1013,16 @@ func parseJupiterV6Instruction(tx TransactionGetter, accounts []uint8, data []by
 			return nil, nil
 		}
 	} else {
-		if len(accounts) < 10 {
+		if len(instruction.Accounts) < 10 {
 			return nil, fmt.Errorf("not enough accounts for jupiter v6 jupiterRoute instruction")
 		}
 		var (
 			srcIdx uint8
 		)
 
-		accounts = accounts[9:]
+		accounts := instruction.Accounts[9:]
 		for i, acctIdx := range accounts {
-			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
 			}
@@ -1030,12 +1034,12 @@ func parseJupiterV6Instruction(tx TransactionGetter, accounts []uint8, data []by
 		if srcIdx == 0 || srcIdx+1 >= uint8(len(accounts)) {
 			return nil, nil
 		}
-		sourceMint, err = tx.GetAccount(accounts[srcIdx]) // getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx]))
+		sourceMint, err = 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]))
+		quoteMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx+1]))
 		if err != nil {
 			return nil, err
 		}
@@ -1044,10 +1048,9 @@ func parseJupiterV6Instruction(tx TransactionGetter, accounts []uint8, data []by
 		}
 	}
 
-	maker, _ := tx.GetAccount(0)
 	signal := &TxSignal{
-		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),
@@ -1057,7 +1060,7 @@ func parseJupiterV6Instruction(tx TransactionGetter, accounts []uint8, data []by
 		IsToken2022:        false,
 		IsMayhemMode:       false,
 		ExactSOL:           false,
-		Block:              tx.Block(),
+		Block:              tx.Block,
 		Token0AmountUint64: inputAmount,
 		Token1AmountUint64: 0,
 	}

pkg/shreder/okxonchainlab.go

@@ -245,13 +245,17 @@ type OkxV2SwapScorch struct {
 	Id [16]byte
 }
 
-func parseOkxDexRouteV2Instruction(tx TransactionGetter, accounts []uint8, data []byte) (*TxSignal, error) {
+func parseOkxDexRouteV2Instruction(tx *versionedTransaction, instructionIndex int) (*TxSignal, error) {
-
+	msg := tx.Message
-	if len(data) < 8 {
+	if instructionIndex >= len(msg.Instructions) {
+		return nil, fmt.Errorf("instruction index out of bounds")
+	}
+	ix := msg.Instructions[instructionIndex]
+	if len(ix.Data) < 8 {
 		return nil, nil
 	}
-	disc := data[:8]
+	disc := ix.Data[:8]
-	data = data[8:]
+	data := ix.Data[8:]
 
 	var (
 		args *OkxV2SwapArgs
@@ -284,8 +288,8 @@ func parseOkxDexRouteV2Instruction(tx TransactionGetter, accounts []uint8, data
 	default:
 		return nil, nil
 	}
-	if len(accounts) < 15 {
+	if len(ix.Accounts) < 15 {
-		return nil, fmt.Errorf("invalid account count: %d", len(accounts))
+		return nil, fmt.Errorf("invalid account count: %d", len(ix.Accounts))
 	}
 	var (
 		inputAmount uint64
@@ -299,24 +303,24 @@ func parseOkxDexRouteV2Instruction(tx TransactionGetter, accounts []uint8, data
 		}
 	}
 	if routeCount > 1 {
-		logger.Warn("pumpSwapSell at inputIdx=0: multiple instances found", "tx", tx.Signatures(), "routeCount", routeCount)
+		logger.Warn("pumpSwapSell at inputIdx=0: multiple instances found", "tx", tx.Signatures[0].String(), "routeCount", routeCount)
 		return nil, nil
 	}
 	if inputAmount == 0 {
 		return nil, nil
 	}
 
-	srcMint, err := tx.GetAccount(accounts[3]) //getStaticKey(tx.Message.StaticAccountKeys, int(ix.Accounts[3]))
+	srcMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(ix.Accounts[3]))
 
 	var (
 		srcIdx uint8
 	)
-	if len(accounts) <= 15 {
+	if len(ix.Accounts) <= 15 {
 		return nil, nil
 	}
-	accounts = accounts[14:]
+	accounts := ix.Accounts[14:]
 	for i, acctIdx := range accounts {
-		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
 		}
@@ -329,7 +333,7 @@ func parseOkxDexRouteV2Instruction(tx TransactionGetter, accounts []uint8, data
 		return nil, nil
 	}
 
-	baseMint, err := tx.GetAccount(accounts[srcIdx]) //  getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx]))
+	baseMint, err := getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx]))
 	if err != nil {
 		return nil, err
 	}
@@ -337,18 +341,17 @@ func parseOkxDexRouteV2Instruction(tx TransactionGetter, accounts []uint8, data
 		return nil, nil
 	}
 
-	quoteMint, err := tx.GetAccount(accounts[srcIdx+1]) // getStaticKey(tx.Message.StaticAccountKeys, int(accounts[srcIdx+1]))
+	quoteMint, err := 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{
-		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),
@@ -360,7 +363,6 @@ func parseOkxDexRouteV2Instruction(tx TransactionGetter, accounts []uint8, data
 		IsMayhemMode:       false,
 		ExactSOL:           false,
 		Token0AmountUint64: inputAmount,
-		Block:              tx.Block(),
 		Token1AmountUint64: 0,
 	}, nil
 }

pkg/shreder/tx.go

@@ -52,7 +52,8 @@ type TxSignal struct {
 	Token0AmountUint64 uint64 `json:"-"`
 	Token1AmountUint64 uint64 `json:"-"`
 
-	Stats Stats `json:"-"`
+	ParseStart time.Time `json:"parse_start"`
+	ParseEnd   time.Time `json:"parse_end"`
 }
 
 func (t *TxSignal) Parse() *TxSignal {

pkg/shreder/txparser_test.go

@@ -149,16 +149,16 @@ func TestParseTermBuy(t *testing.T) {
 	}
 
 	client := rpc.New(rpcUrl)
-	txChannel := make(chan TxSignal, 1)
+	signals := ParseTransaction(
-	go func() {
+		getTransaction(t, client, "5Gz1fa4Qhb35bkg9QCMXpxCX5uuNr7WcjcmrwajGZA7kXsvNS9pDnYe12ggWeSqf1nwZbVPob6DkX6fcwbE9ofBR"),
-		ParseTransaction(
+		nil,
-			getTransaction(t, client, "5Gz1fa4Qhb35bkg9QCMXpxCX5uuNr7WcjcmrwajGZA7kXsvNS9pDnYe12ggWeSqf1nwZbVPob6DkX6fcwbE9ofBR"),
+		false,
-			nil, txChannel,
+	)
-			false,
+	if len(signals) != 1 {
-		)
+		t.Fatalf("expected 1 signal, got %d", len(signals))
-	}()
+	}
 
-	signal := <-txChannel
+	signal := signals[0]
 	if signal.Label != "terminal" {
 		t.Fatalf("expected terminal signal, got %s", signal.Label)
 	}
@@ -186,16 +186,16 @@ func TestParseBonkBuy(t *testing.T) {
 	}
 
 	client := rpc.New(rpcUrl)
-	txChannel := make(chan TxSignal, 1)
+	signals := ParseTransaction(
-	go func() {
+		getTransaction(t, client, "3gHF3TA2aA8rpjdmoEs2vA89vrq9J9NnTTUSXHfE6uXcaYP9cJgLtEUjCmsK9EWAyHEg7cEiepehQf4GFv1272jW"),
-		ParseTransaction(
+		nil,
-			getTransaction(t, client, "3gHF3TA2aA8rpjdmoEs2vA89vrq9J9NnTTUSXHfE6uXcaYP9cJgLtEUjCmsK9EWAyHEg7cEiepehQf4GFv1272jW"),
+		false,
-			nil, txChannel,
+	)
-			false,
+	if len(signals) != 1 {
-		)
+		t.Fatalf("expected 1 signal, got %d", len(signals))
-	}()
+	}
 
-	signal := <-txChannel
+	signal := signals[0]
 	if signal.Label != "bonk" {
 		t.Fatalf("expected bonk signal, got %s", signal.Label)
 	}
@@ -223,16 +223,16 @@ func TestParseBonkSell(t *testing.T) {
 	}
 
 	client := rpc.New(rpcUrl)
-	txChannel := make(chan TxSignal, 1)
+	signals := ParseTransaction(
-	go func() {
+		getTransaction(t, client, "3XNi6b3j69SSStqLLRQVH5BNGVfEoFxGCzmpdd5FvrY4kmC8T644WGdEhCH9fAdrxWuR2Mtzgywq8K7qetu5MGyb"),
-		ParseTransaction(
+		nil,
-			getTransaction(t, client, "3XNi6b3j69SSStqLLRQVH5BNGVfEoFxGCzmpdd5FvrY4kmC8T644WGdEhCH9fAdrxWuR2Mtzgywq8K7qetu5MGyb"),
+		false,
-			nil, txChannel,
+	)
-			false,
+	if len(signals) != 1 {
-		)
+		t.Fatalf("expected 1 signal, got %d", len(signals))
-	}()
+	}
 
-	signal := <-txChannel
+	signal := signals[0]
 	if signal.Label != "bonk" {
 		t.Fatalf("expected bonk signal, got %s", signal.Label)
 	}
@@ -260,16 +260,16 @@ func TestParsePhotonBuy(t *testing.T) {
 	}
 
 	client := rpc.New(rpcUrl)
-	txChannel := make(chan TxSignal, 1)
+	signals := ParseTransaction(
-	go func() {
+		getTransaction(t, client, "4DCEcXAWBxagXoUNGhWsJ7qfxq5SuE5BG2cBDBqAY7sCHkBopaMJu33ZnXnFHqzPMmWxVxq6666KRF4hMHVB33Ux"),
-		ParseTransaction(
+		nil,
-			getTransaction(t, client, "4DCEcXAWBxagXoUNGhWsJ7qfxq5SuE5BG2cBDBqAY7sCHkBopaMJu33ZnXnFHqzPMmWxVxq6666KRF4hMHVB33Ux"),
+		false,
-			nil, txChannel,
+	)
-			false,
+	if len(signals) != 1 {
-		)
+		t.Fatalf("expected 1 signal, got %d", len(signals))
-	}()
+	}
 
-	signal := <-txChannel
+	signal := signals[0]
 	if signal.Label != "photon" {
 		t.Fatalf("expected terminal signal, got %s", signal.Label)
 	}