From c3536761af918bdc1b2b0d9ad8d34f58e0d7f5d6 Mon Sep 17 00:00:00 2001 From: thloyi Date: Wed, 28 Jan 2026 14:55:47 +0800 Subject: [PATCH] fix entry parse --- cmd/shreder/main.go | 2 +- pkg/shreder/client.go | 6 +- pkg/shreder/entry.go | 25 ++++--- pkg/shreder/entry_test.go | 30 +++++++++ pkg/shreder/program_dlmm.go | 2 +- pkg/shreder/txparser.go | 8 ++- pkg/shreder/txparser_test.go | 126 ++++++++++++++++++++++++++--------- 7 files changed, 147 insertions(+), 52 deletions(-) create mode 100644 pkg/shreder/entry_test.go diff --git a/cmd/shreder/main.go b/cmd/shreder/main.go index d4285c5..0f245c8 100644 --- a/cmd/shreder/main.go +++ b/cmd/shreder/main.go @@ -78,7 +78,7 @@ func main() { // async read from shreder 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) diff --git a/pkg/shreder/client.go b/pkg/shreder/client.go index c1b1e01..e25fd1f 100644 --- a/pkg/shreder/client.go +++ b/pkg/shreder/client.go @@ -141,10 +141,8 @@ func (c *Client) ReadEntriesSync(ctx context.Context, txCh chan<- TxSignal) erro } } - entries := response.Entries - err = c.pool.Submit(func() { - ParseTransactionForEntries(ctx, slot, bytes.NewReader(entries), c.tableLoader, txCh) + ParseTransactionForEntries(ctx, slot, bytes.NewReader(response.Entries), c.tableLoader, txCh) }) if err != nil && errors.Is(err, ants.ErrPoolOverload) { logger.Warn("task pool is full") @@ -192,7 +190,7 @@ func (c *Client) ReadSync(ctx context.Context, txCh chan<- TxSignal) error { txData := response.Transaction err := c.pool.Submit(func() { - ParseTransactionForSubscribe(ctx, txData, c.tableLoader, txCh) + ParseTransactionForSubscribe(ctx, txData, c.tableLoader, txCh, nil) }) if err != nil && errors.Is(err, ants.ErrPoolOverload) { logger.Warn("task pool is full") diff --git a/pkg/shreder/entry.go b/pkg/shreder/entry.go index 7330f33..bdc8295 100644 --- a/pkg/shreder/entry.go +++ b/pkg/shreder/entry.go @@ -29,9 +29,8 @@ func (wr *wrapperReader) Skip(n int) error { func (wr *wrapperReader) ReadCompactU16() (uint16, error) { ln := 0 size := 0 - + var buf [1]byte for i := 0; i < 3; i++ { - var buf [1]byte _, err := io.ReadFull(wr, buf[:]) if err != nil { return 0, fmt.Errorf("unable to decode compact u16 at %d: %w", i, err) @@ -61,11 +60,11 @@ func (wr *wrapperReader) ReadByte() (uint8, error) { return buf[0], nil } -func ResizeSlice[T any](slice []T, newSize int) { +func ResizeSlice[T any](slice []T, newSize int) []T { if cap(slice) < newSize { slice = append(slice, make([]T, newSize-len(slice))...) } - slice = slice[:newSize] + return slice[:newSize] } // entriesToVersionedTransaction converts raw entry bytes to versioned transactions. @@ -84,7 +83,7 @@ func entriesToVersionedTransaction(slot uint64, data io.Reader, callback func(tx // return nil, nil //} if entriesNum > 2048 { - return fmt.Errorf("entries num is too large: %d > %d", entriesNum, 1024) + return fmt.Errorf("entries num is too large: %d > %d", entriesNum, 2048) } for i := uint64(0); i < entriesNum; i++ { @@ -113,7 +112,7 @@ func entriesToVersionedTransaction(slot uint64, data io.Reader, callback func(tx versioned := VersionedTransaction{} versioned.Block = slot versioned.Time = time.Now() - ResizeSlice(versioned.Signatures, int(numSignatures)) + versioned.Signatures = ResizeSlice(versioned.Signatures, int(numSignatures)) for k := 0; k < int(numSignatures); k++ { _, err = io.ReadFull(b, versioned.Signatures[k][:]) if err != nil { @@ -151,7 +150,7 @@ func entriesToVersionedTransaction(slot uint64, data io.Reader, callback func(tx return fmt.Errorf("numAccountKeys %d exceeds maximum in entry %d, txn %d", numAccountKeys, i, j) } - ResizeSlice(versioned.StaticAccountKeys, int(numAccountKeys)) + versioned.StaticAccountKeys = ResizeSlice(versioned.StaticAccountKeys, int(numAccountKeys)) for k := 0; k < int(numAccountKeys); k++ { _, err = io.ReadFull(b, versioned.StaticAccountKeys[k][:]) @@ -176,7 +175,7 @@ func entriesToVersionedTransaction(slot uint64, data io.Reader, callback func(tx if numInstructions >= 256 { return fmt.Errorf("numInstructions %d exceeds maximum in entry %d, txn %d, txHash: %s", numInstructions, i, j, versioned.GetSignature()) } - ResizeSlice(versioned.Instructions, int(numInstructions)) + versioned.Instructions = ResizeSlice(versioned.Instructions, int(numInstructions)) for k := 0; k < int(numInstructions); k++ { versioned.Instructions[k].ProgramIDIndex, err = b.ReadByte() if err != nil { @@ -191,7 +190,7 @@ func entriesToVersionedTransaction(slot uint64, data io.Reader, callback func(tx if numAccounts >= 256 { return fmt.Errorf("numAccounts %d exceeds maximum for ix[%d] in entry %d, txn %d", numAccounts, k, i, j) } - ResizeSlice(versioned.Instructions[k].Accounts, int(numAccounts)) + versioned.Instructions[k].Accounts = ResizeSlice(versioned.Instructions[k].Accounts, int(numAccounts)) //.AccountsLen = int(numAccounts) if numAccounts != 0 { @@ -208,7 +207,7 @@ func entriesToVersionedTransaction(slot uint64, data io.Reader, callback func(tx if dataLen > 2048 { return fmt.Errorf("mx.Instructions[%d].Data length %d exceeds maximum in entry %d, txn %d, txHash: %s", k, dataLen, i, j, versioned.GetSignature()) } - ResizeSlice(versioned.Instructions[k].Accounts, int(numAccounts)) + versioned.Instructions[k].Data = ResizeSlice(versioned.Instructions[k].Data, int(dataLen)) if dataLen > 0 { _, err = io.ReadFull(b, versioned.Instructions[k].Data) if err != nil { @@ -226,7 +225,7 @@ func entriesToVersionedTransaction(slot uint64, data io.Reader, callback func(tx if numLookups >= 32 { return fmt.Errorf("numLookups %d exceeds maximum in entry %d, txn %d", numLookups, i, j) } - ResizeSlice(versioned.AddressTableLookups, int(numLookups)) + versioned.AddressTableLookups = ResizeSlice(versioned.AddressTableLookups, int(numLookups)) for k := uint8(0); k < numLookups; k++ { _, err = io.ReadFull(b, versioned.AddressTableLookups[k].AccountKey[:]) if err != nil { @@ -240,7 +239,7 @@ func entriesToVersionedTransaction(slot uint64, data io.Reader, callback func(tx if numWritable >= 256 { return fmt.Errorf("numWritableIndexes %d exceeds maximum for lookup[%d] in entry %d, txn %d", numWritable, k, i, j) } - ResizeSlice(versioned.AddressTableLookups[k].WritableIndexes, int(numWritable)) + versioned.AddressTableLookups[k].WritableIndexes = ResizeSlice(versioned.AddressTableLookups[k].WritableIndexes, int(numWritable)) if numWritable > 0 { _, err = io.ReadFull(b, versioned.AddressTableLookups[k].WritableIndexes) if err != nil { @@ -256,7 +255,7 @@ func entriesToVersionedTransaction(slot uint64, data io.Reader, callback func(tx if numReadonly > 256 { return fmt.Errorf("numReadonlyIndexes %d exceeds maximum for lookup[%d] in entry %d, txn %d", numReadonly, k, i, j) } - ResizeSlice(versioned.AddressTableLookups[k].ReadonlyIndexes, int(numReadonly)) + versioned.AddressTableLookups[k].ReadonlyIndexes = ResizeSlice(versioned.AddressTableLookups[k].ReadonlyIndexes, int(numReadonly)) if numReadonly > 0 { _, err = io.ReadFull(b, versioned.AddressTableLookups[k].ReadonlyIndexes) if err != nil { diff --git a/pkg/shreder/entry_test.go b/pkg/shreder/entry_test.go new file mode 100644 index 0000000..d7f50b7 --- /dev/null +++ b/pkg/shreder/entry_test.go @@ -0,0 +1,30 @@ +package shreder + +import ( + "bytes" + "encoding/base64" + "testing" +) + +func TestDecodeEntry(t *testing.T) { + tests := []struct { + name string + data string + }{ + { + name: "TestDecodeEntry1", + data: "BgAAAAAAAADFCQAAAAAAABa3RhozUFDf4JW2I2fuaRUnS2jsksCNpvGoOCvKgSOXAQAAAAAAAAABMhXYcIO0FsOv+exTIC11M9XGWTL0HEn8NGWa5bUiWYHh3eCIKm7gmHNm++o04EB7a1YQm0auEH5lKtoxPuvCDoABAAYJKqrFxl/dG3sl/ET53EdOUY7GkME7PDLzYjrnjXECoJxSx/dHprrpooHmYGfnxMLcjOfV6ynWLJpdaaX3veAU0HJby+7Nj0f6IJgK9M/rtAyoL5FBP7e2TBAbNB7nNuodAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAADBkZv5SEXMv/srbpyw5vnvIzlu8X3EmssQ5s6QAAAAAxKoBKOldPhYiqlAcWFqesHs3NUwQjqC3kbRW3H7qM2DLf6u1L3pki7WzF9mgGLkFfLAkd0+v4B5sTfmMw4WIFTtjzvYd5D6GsWgxoFbrya06c75FbKVGasJZ2fGjzorLrhu5lE80vnhcE09VCUTOaWQ+aAp+uNUW/dgm/9Ojhie4glXf1oNciuf82NedY9xhuipRtE8GnjHbli1VNH6M4DBAAFAsXcAAAEAAkDECcAAAAAAAAFBwAGBwgCAQOMAxwJXZZWmbxzVQAAAABbvRzmF3krR2xVmRwnzf2JeU+fEzVrq8nJJAX18AeWgwAAAAAALdh5AAAAAAAAEi7////4AAAAAGlkfEEAAAAAaWR8QQAAAAAALcKzAAAAAAAAF8wNAAAAWrXF+hEGOoH4/IgPBLRNlnwy7XIaFhKwlUiRfXcp9dbfGvFiHlVY3wRKN84r3tCsJ2zEilNS1s4w7PDFr80TIp70TgTOwdlFbPDqR3oE1rYrzwj84FtD4vTlzf5Lb8jVUxLU5GW7ZYRKTiFDnxymoVg/OoD5jyHgzNXVxYjbW+X0H0729i9wQ8sRM9p9657ocK0lBQwWLnBRkzNvJO+CE8CFL3TMWkPJw9WXLmkBNT5CTFwQyCSEAy5J71XXNJRbH7AwF3rqpcpAlHGXQzmIhU3WAzqfqXzCFoSpH2ZgYWgr/Q+Msprylig2/NajTl1TNn1PbPqLAdEngizvXgzUjPlSEw0AAABbvRzmF3krR2xVmRwnzf2JeU+fEzVrq8nJJAX18AeWgwChDwAAAAAAAHpIeKeU5QUvIhET9QXDNbU1sNu1UZGiclEVaXh+lRTqAQAAAAAAAAABFdDD4PHwP1VOtJCmJQQ/K+MD8X6PhOoM3Jt99HR6SJLSLfq6ZUsjil1yVsijWgNJiChwIRapmsYvxrucGDYbC4ABAAYJKqrFxl/dG3sl/ET53EdOUY7GkME7PDLzYjrnjXECoJxyW8vuzY9H+iCYCvTP67QMqC+RQT+3tkwQGzQe5zbqHfwfzxCF7RlfsUB3A3x42YehLopADLOuNtdbgtPOifOcAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAADBkZv5SEXMv/srbpyw5vnvIzlu8X3EmssQ5s6QAAAAAxKoBKOldPhYiqlAcWFqesHs3NUwQjqC3kbRW3H7qM2DLf6u1L3pki7WzF9mgGLkFfLAkd0+v4B5sTfmMw4WIFTtjzvYd5D6GsWgxoFbrya06c75FbKVGasJZ2fGjzorLrhu5lE80vnhcE09VCUTOaWQ+aAp+uNUW/dgm/9Ojhie4glXf1oNciuf82NedY9xhuipRtE8GnjHbli1VNH6M4DBAAFAsjcAAAEAAkDECcAAAAAAAAFBwAGBwgBAgOMAxwJXZZWmbxzVQAAAAB2+oUVi/FO3ncIf+OuRy9mIT9uovW0Ecst5HJ5SZD6XAAAAAAG8vwiAAAAAAADL5P////4AAAAAGlkfEEAAAAAaWR8QQAAAAAG8vzJAAAAAAADL6kNAAAAJm8WUjlxEejZ93dsaGw1Nx6RghTnNEWFbWjsVDmckpRv2z1oEuAZhgxQXW1500c+YQsAAMX7hHPSL/frWNo4S/RkrT1R/ZS4+InOFIoCGM/iHJOdy+hlYMloqICCNG5Zxfut6ygCpT0n9TX44lQxLAzva/xI8ZshVuN7aqIhxLBaoX9tJYjqS4HjMsbeqiMk6JZefzw2rhD/y675OLNdfDtAOFse814kbQimRFADLMHhSjdeyCSEAy5J71XXNJRbH7AwF3rqpcpAlHGXQzmIhU3WAzqfqXzCFoSpH2ZgYWgr/Q+Msprylig2/NajTl1TNn1PbPqLAdEngizvXgzUjPlSEw0AAAB2+oUVi/FO3ncIf+OuRy9mIT9uovW0Ecst5HJ5SZD6XAD1AQAAAAAAANvS5ucXWXc9QVH9Cq4c4LAGP490izdpUgcUufMycFG4AQAAAAAAAAACNw8RJiLgw2ExpFsGCRLWd1SlZ3l5fxzpclbVJkl9LSA56bjKwtOwRqWImPV8EYaL54WddYglsO0mkGB7tyZwAhZ/+McWh3V+BCHwfM7B5aNHE0OVJnVmTVQJzhdmsum3qSGJqvfZ07T0BEHODnazNOgcmKIeNlxz+HMnJvsaqAiAAgAOG/JJS5mGOGsRH0ssnpy9nYXUMg5nfZUcTJlPhozDahk3jWnWUY1v/4lpQP1C4Co7PK7v51+bVQsqz8u1fDl6rl0Ws6msFbah4llAUolNWci7rt+zbrKeEFBwHdglI4/Su5F3taZcZApPgR+uhuHsmRO7kcuKHuoFQS7xa3736a1VKCm1MRMYJh+fdA+vOpORxn3Z8L1uE6vLOiYioufyk+u0OZFWOPF28JuwCKM7yOOnXk5/Y6X+LfW2k2Jnw7UsDd41iSQ5EUjJrbPiDdooUdVylKv6acxNFd0pZdiTUZDKsuokmOXc8ndSZ+omuY8Pnif0oQ//Mzd53/qaixz9agcj6HopLl/IIEmSMf+KTzYWClgIplUc68POTRddZdr9R0uJB7ADVhAVx8UzZhEj0+x6euFjgrTD5wy1ZGtMpvLJtvHpMHNMg0mXbwCZToudtr6xSVa1E51fnCppnaA8tb2j17sSflitwSymj4NDfsLhw/mCDek+WPkXiikY3ar3tH0kUMBynorkKwHnnOWeB/U24cbtrbZzarHj4brTyGlbAwZGb+UhFzL/7K26csOb57yM5bvF9xJrLEObOkAAAACMlyWPTiSJ8bs9ECkUjg2DC1oTmdr/EIQEjnvY2+n4WQabiFf+q4GE+2h/Y0YYwDXaxDncGus7VZig8AAAAAABAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAG3fbh12Whk9nL4UbO63msHLSF7V9bN5E6jPWFfv8AqQrxw0MhiMo6Y1E1oToYlRrOvSnmrC2uZ//bBtdBbnMoDBTe/IJexnaUJQgYu2VAZfQpjTFW1XG01PgJDBjpqGOJC6ZE/h9VqhnxHNLS7BTTIztuCkvq7vcraYWOIeFw1kUJtvSdN61HR5CSPKiGJMYwRl5NH01w56WjlyyVs2Mi/4ODgYuo+ijDzTttXpP5+rjwl5vDchWsxbJGh3uow8nlSnCVKIOfYcC5uGB5iRwTkhbkenG2L7c77HIWlFh0Xm1lkEO6pSoYG1hEI56NVC0f3tzWgc5+cdB1KdrJbiatQSRuzH14/oHkF3OkaWVBmTeSOgdkR5ffbz61FEJgEMsMNf+pBVqOVo2o97wHVhUnTPHJLKQfQACcUWqkFMJ8cM0FVHWEPtifKtl6Hc4Vb+jAtwo68/cMMNi5Lp5fMAq2Cg0ABQKg9wMADQAJAxAnAAAAAAAADgYAAgAPEBEAEAMAAhIMAgAAAEZ4v8kBAAAAEQECARETFQMAFA8VAgQFBhYHEREQDhcTCBgZGhgz5oWkAX+DrUZ4v8kBAAAA9HFQJGIFAAARAwIAAAEJDgYBCQEPEBEAExcDARQPFQkKBQYWBxEREA4XEwgYCwwZGhlmBj0SAdrr6gfhM/4BAAAA7aKejigGAAAAEQMJAQEBCQCJEwAAAAAAAA9mmuSO/pn1vVCwatBduahHAtWqOYsQckX7LVHGUWxEAQAAAAAAAAAB/08+MAvbV+UW3ONET6GCATN/0j6lr+zFAgtoUs3O9iGKsATjGupdFAfOuIz22ZUo9EhUYulerniPErJf1GwvDoABAAYJKqrFxl/dG3sl/ET53EdOUY7GkME7PDLzYjrnjXECoJwAPKMtldCc34a+8OJeXzB6SLmvDZnGJ1l8UI6v0zK583Jby+7Nj0f6IJgK9M/rtAyoL5FBP7e2TBAbNB7nNuodAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAADBkZv5SEXMv/srbpyw5vnvIzlu8X3EmssQ5s6QAAAAAxKoBKOldPhYiqlAcWFqesHs3NUwQjqC3kbRW3H7qM2DLf6u1L3pki7WzF9mgGLkFfLAkd0+v4B5sTfmMw4WIFTtjzvYd5D6GsWgxoFbrya06c75FbKVGasJZ2fGjzorLrhu5lE80vnhcE09VCUTOaWQ+aAp+uNUW/dgm/9Ojhie4glXf1oNciuf82NedY9xhuipRtE8GnjHbli1VNH6M4DBAAFAtncAAAEAAkDECcAAAAAAAAFBwAGBwgCAQOMAxwJXZZWmbxzVQAAAAApmslIdCp5nSehZJx2A1smV3rQ62WFpa4qaR0x8u6QxAAAAAACJM7+AAAAAAAA7MX////4AAAAAGlkfEEAAAAAaWR8QQAAAAACIRozAAAAAAAA5KoNAAAAEvs0gfitwlofdBEVF7Lt1lfN70g1GT3CdkPWiSgqNbDQbHzLR5E42IQjGe03az8gQegzC6d37HZlRZH/kJ22CZ0PM0dacJgXOMnuEsFIK7SXMgVDanpmZUtCy9Um74f1qdtNvu5ibFkPaKBjL/s5aWnhuQGYaoic/GYMRcWCbqjzKjyADMqIu8pvFtCR+xAqy1nWR5QFDyhatHN2WV6zW4I2Fz1ig+BjCNI8dzE8/PFRksFWEAEL5qLbcIJvDFxsjFOfxytrVXBYzNRn8eCuCKRC0cXFZpG/RtDkDWZgYWgr/Q+Msprylig2/NajTl1TNn1PbPqLAdEngizvXgzUjPlSEw0AAAApmslIdCp5nSehZJx2A1smV3rQ62WFpa4qaR0x8u6QxADpAwAAAAAAAEhcAJ18XU/9Gt8vnjmOC6hTZ4ZKZvmM+NF5mrgM9ffWAQAAAAAAAAABTL5A15gYd/UMJbcGOV2Ct6S7FSBoc2n66Zz6KL0UjdXW7A+zKwz9NgFLbipLVLeGe/p0Cckl9o1dbZ2kvqWUA4ABAAILTwBflcdWwqKPHy3t7aWLaS7OOgzcBUjNyNGjoAmrjJG2bDJOP9v/YR1WdvZFYzgoQx7dSp0XSn+/xVTHMZPeDakBaM20dllxS9FbwoHcAl4+5ayVmW750JcePad73nI5J3BukDE6OqL6buYdDLNet1XTPDVz9z/H4eW0SJdrUaY4roH+3oSwXKMJgUdMj2FNEXn0H06SonFdsNHP5afvsxOaXm74YlA/OiRkxhuwQ9M6IZwPF6HMfUX9J60KWzvUGU8ganVs499OZpNFuQK66ByjevaCa5CLSOWrX6xvXZKuQZuW9cEpZoJ6C0vX00Ct0mcs+8qvF5mFqLoYAvuXK407KiAcJcCKo34WRtNyN6bwxCCPVBnpPuz8JujdjnUOAwZGb+UhFzL/7K26csOb57yM5bvF9xJrLEObOkAAAADb8e/+4cdOK0o6VEHrgrX2LVzaqsoFoWn3PrAFnklFslLjHbkCnl1jpNNz1kAX0wyy369qIOkA9xcwzz2YZmryAwkABQJg2AMACQAJAx8CAAAAAAAACjUACwwNDissLS4vDxAwMRESExUBFhcYGTMUMhobHB0CAx4EHyAhIjQUMiMkJSYFBicoKSoHCFUKFAEAAAADAdgJBgIRExoGAgIAAA4DAgAABgIgIikGAgIAAA4DAgAABgIgIi8GAgIAAA4DAgAAAAUGBAIHCQYDBQYABQIEBwEBBwYHBwIIBwUGCAMHAzz5ttVrusK5h7tt/2RTSLpqafgA6KGY3eCUIL45wcZFCgKSA7ClqAoJr6YIBAUIBwa1tgtenKHzNpGxKsf+ZGPnHLf6WVam69e0+R+9WtPI9xzJZgnX2drb3N7f4OEB3W6sDIXL5UyNDw/eNdnWLD6HO2QfptI9yWG18bDGQ8YqDVFTVFVWWFlaW5KTlJUBV6oJAAAAAAAAAiT+oNqZdb0ugGzuxs1/Ak4VhBDLYWiUBIZQ51J7H2kAAAAAAAAAAA==", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + data, _ := base64.StdEncoding.DecodeString(tt.data) + err := entriesToVersionedTransaction(0, bytes.NewReader(data), func(versioned VersionedTransaction) {}) + if err != nil { + t.Errorf("decodeEntry() error = %v", err) + return + } + }) + } +} diff --git a/pkg/shreder/program_dlmm.go b/pkg/shreder/program_dlmm.go index 77b4988..7b1238b 100644 --- a/pkg/shreder/program_dlmm.go +++ b/pkg/shreder/program_dlmm.go @@ -104,7 +104,7 @@ func parseDlmmInstruction(tx VersionedTransaction, instructionIndex int) (TxSign return nil, fmt.Errorf("data is empty") } if len(instruction.Accounts) < 13 { - return nil, fmt.Errorf("accounts too short") + return nil, nil // fmt.Errorf("accounts too short") } disc := instruction.Data[:8] diff --git a/pkg/shreder/txparser.go b/pkg/shreder/txparser.go index 0a595c5..726a800 100644 --- a/pkg/shreder/txparser.go +++ b/pkg/shreder/txparser.go @@ -68,13 +68,15 @@ var ( } ) -func ParseTransactionForSubscribe(ctx context.Context, update *SubscribeUpdateTransaction, loader *AddressTables, parsed chan<- TxSignal) { +func ParseTransactionForSubscribe(ctx context.Context, update *SubscribeUpdateTransaction, loader *AddressTables, parsed chan<- TxSignal, done chan<- struct{}) { versioned, err := toVersionedTransaction(update) if err != nil { logger.Debug("txparser: failed to convert to versioned transaction", "error", err) + close(done) return } - go ParseTransaction(ctx, versioned, loader, parsed) + ParseTransaction(ctx, versioned, loader, parsed) + close(done) } var VoteProgram = solana.MustPublicKeyFromBase58("Vote111111111111111111111111111111111111111") @@ -147,7 +149,7 @@ func ParseTransaction(ctx context.Context, versioned VersionedTransaction, loade txRes, err := handler.Func(versioned, i) if err != nil { if !strings.HasPrefix(err.Error(), "account index") { - logger.Debug("txparser: failed to parse", "label", handler.Label, "instruction", err, "tx_hash", versioned.Signatures[0].String()) + logger.Debug("txparser: failed to parse", "label", handler.Label, "err", err, "tx_hash", versioned.Signatures[0].String()) } continue } diff --git a/pkg/shreder/txparser_test.go b/pkg/shreder/txparser_test.go index 3954b7e..510c2f1 100644 --- a/pkg/shreder/txparser_test.go +++ b/pkg/shreder/txparser_test.go @@ -149,11 +149,21 @@ func TestParseTermBuy(t *testing.T) { } client := rpc.New(rpcUrl) - signals := ParseTransaction( - getTransaction(t, client, "5Gz1fa4Qhb35bkg9QCMXpxCX5uuNr7WcjcmrwajGZA7kXsvNS9pDnYe12ggWeSqf1nwZbVPob6DkX6fcwbE9ofBR"), - nil, - false, - ) + ch := make(chan TxSignal) + closed := make(chan struct{}) + go func() { + ParseTransactionForSubscribe( + context.Background(), + getTransaction(t, client, "5Gz1fa4Qhb35bkg9QCMXpxCX5uuNr7WcjcmrwajGZA7kXsvNS9pDnYe12ggWeSqf1nwZbVPob6DkX6fcwbE9ofBR"), + nil, + ch, + closed, + ) + }() + signals := make([]TxSignal, 0) + for signal := range ch { + signals = append(signals, signal) + } if len(signals) != 1 { t.Fatalf("expected 1 signal, got %d", len(signals)) } @@ -186,11 +196,23 @@ func TestParseBonkBuy(t *testing.T) { } client := rpc.New(rpcUrl) - signals := ParseTransaction( - getTransaction(t, client, "3gHF3TA2aA8rpjdmoEs2vA89vrq9J9NnTTUSXHfE6uXcaYP9cJgLtEUjCmsK9EWAyHEg7cEiepehQf4GFv1272jW"), - nil, - false, - ) + + ch := make(chan TxSignal) + closed := make(chan struct{}) + go func() { + ParseTransactionForSubscribe( + context.Background(), + getTransaction(t, client, "3gHF3TA2aA8rpjdmoEs2vA89vrq9J9NnTTUSXHfE6uXcaYP9cJgLtEUjCmsK9EWAyHEg7cEiepehQf4GFv1272jW"), + nil, + ch, + closed, + ) + }() + signals := make([]TxSignal, 0) + for signal := range ch { + signals = append(signals, signal) + } + if len(signals) != 1 { t.Fatalf("expected 1 signal, got %d", len(signals)) } @@ -223,11 +245,22 @@ func TestParseBonkSell(t *testing.T) { } client := rpc.New(rpcUrl) - signals := ParseTransaction( - getTransaction(t, client, "3XNi6b3j69SSStqLLRQVH5BNGVfEoFxGCzmpdd5FvrY4kmC8T644WGdEhCH9fAdrxWuR2Mtzgywq8K7qetu5MGyb"), - nil, - false, - ) + ch := make(chan TxSignal) + closed := make(chan struct{}) + go func() { + ParseTransactionForSubscribe( + context.Background(), + getTransaction(t, client, "3XNi6b3j69SSStqLLRQVH5BNGVfEoFxGCzmpdd5FvrY4kmC8T644WGdEhCH9fAdrxWuR2Mtzgywq8K7qetu5MGyb"), + nil, + ch, + closed, + ) + }() + signals := make([]TxSignal, 0) + for signal := range ch { + signals = append(signals, signal) + } + if len(signals) != 1 { t.Fatalf("expected 1 signal, got %d", len(signals)) } @@ -260,11 +293,23 @@ func TestParsePhotonBuy(t *testing.T) { } client := rpc.New(rpcUrl) - signals := ParseTransaction( - getTransaction(t, client, "4DCEcXAWBxagXoUNGhWsJ7qfxq5SuE5BG2cBDBqAY7sCHkBopaMJu33ZnXnFHqzPMmWxVxq6666KRF4hMHVB33Ux"), - nil, - false, - ) + + ch := make(chan TxSignal) + closed := make(chan struct{}) + go func() { + ParseTransactionForSubscribe( + context.Background(), + getTransaction(t, client, "4DCEcXAWBxagXoUNGhWsJ7qfxq5SuE5BG2cBDBqAY7sCHkBopaMJu33ZnXnFHqzPMmWxVxq6666KRF4hMHVB33Ux"), + nil, + ch, + closed, + ) + }() + signals := make([]TxSignal, 0) + for signal := range ch { + signals = append(signals, signal) + } + if len(signals) != 1 { t.Fatalf("expected 1 signal, got %d", len(signals)) } @@ -297,11 +342,22 @@ func TestParseJupiterV6PumpFunBuy(t *testing.T) { } client := rpc.New(rpcUrl) - signals := ParseTransaction( - getTransaction(t, client, "4QF5whXwjx234fMXeH3HrJCy5knFJmKPtgbXys8xKGz1pZypqPvXBr4BoAqXfYn8jLL4HXPY1pcvxCCW1XREFNxd"), - nil, - false, - ) + ch := make(chan TxSignal) + closed := make(chan struct{}) + go func() { + ParseTransactionForSubscribe( + context.Background(), + getTransaction(t, client, "4QF5whXwjx234fMXeH3HrJCy5knFJmKPtgbXys8xKGz1pZypqPvXBr4BoAqXfYn8jLL4HXPY1pcvxCCW1XREFNxd"), + nil, + ch, + closed, + ) + }() + signals := make([]TxSignal, 0) + for signal := range ch { + signals = append(signals, signal) + } + if len(signals) != 1 { t.Fatalf("expected 1 signal, got %d", len(signals)) } @@ -337,11 +393,21 @@ func TestParseJupiterV6PumpFunSell(t *testing.T) { } client := rpc.New(rpcUrl) - signals := ParseTransaction( - getTransaction(t, client, "yCnE7ZA8dqB5iAZtwpSN2ar5HXh3gBjgaG2xtnwXDPFyHAm5XFU8642uTZTH5A2iPQ6G9hrj5eEPAJiWrfe38gM"), - nil, - false, - ) + ch := make(chan TxSignal) + closed := make(chan struct{}) + go func() { + ParseTransactionForSubscribe( + context.Background(), + getTransaction(t, client, "yCnE7ZA8dqB5iAZtwpSN2ar5HXh3gBjgaG2xtnwXDPFyHAm5XFU8642uTZTH5A2iPQ6G9hrj5eEPAJiWrfe38gM"), + nil, + ch, + closed, + ) + }() + signals := make([]TxSignal, 0) + for signal := range ch { + signals = append(signals, signal) + } if len(signals) != 1 { t.Fatalf("expected 1 signal, got %d", len(signals)) }