Patch/480 poolmanager errors and 481 remove fatal throws (#482)

* Update main indexer loop to add to failed blocks table on broken parser instead of throwing FTL log. Update poolmanager swap parser to use transfer event in the case that the swap uses different events

* gofumpt

core/tx.go

@@ -428,7 +428,8 @@ func ProcessTx(db *gorm.DB, tx txtypes.MergedTx) (txDBWapper dbTypes.TxDBWrapper
 					config.Log.Error(fmt.Sprintf("[Block: %v] ParseCosmosMessage failed for msg of type '%v'.", tx.TxResponse.Height, msgType), err)
 					config.Log.Error(fmt.Sprint(messageLog))
 					config.Log.Error(tx.TxResponse.TxHash)
-					config.Log.Fatal("Issue parsing a cosmos msg that we DO have a parser for! PLEASE INVESTIGATE")
+					config.Log.Error("Issue parsing a cosmos msg that we DO have a parser for! PLEASE INVESTIGATE")
+					return txDBWapper, txTime, fmt.Errorf("error parsing message we have a parser for: '%v'", msgType)
 				}
 				// if this msg isn't include in our list of those we are explicitly ignoring, do something about it.
 				// we have decided to throw the error back up the call stack, which will prevent any indexing from happening on this block and add this to the failed block table

osmosis/modules/poolmanager/types.go

@@ -1,6 +1,7 @@
 package poolmanager
 
 import (
+	"errors"
 	"fmt"
 	"strconv"
 
@@ -95,31 +96,68 @@ func (sf *WrapperMsgSwapExactAmountIn) HandleMsg(msgType string, msg sdk.Msg, lo
 
 	// The attribute in the log message that shows you the tokens swapped
 	tokensSwappedEvt := txModule.GetEventWithType("token_swapped", log)
-	if tokensSwappedEvt == nil {
-		return &txModule.MessageLogFormatError{MessageType: msgType, Log: fmt.Sprintf("%+v", log)}
-	}
+	transferEvt := txModule.GetEventWithType("transfer", log)
 
-	// The last route in the hops gives the token out denom and pool ID for the final output
-	lastRoute := sf.OsmosisMsgSwapExactAmountIn.Routes[len(sf.OsmosisMsgSwapExactAmountIn.Routes)-1]
-	lastRouteDenom := lastRoute.TokenOutDenom
-	lastRoutePoolID := lastRoute.PoolId
+	switch {
+	case tokensSwappedEvt != nil:
+		if tokensSwappedEvt == nil {
+			return &txModule.MessageLogFormatError{MessageType: msgType, Log: fmt.Sprintf("%+v", log)}
+		}
 
-	tokenOutStr := txModule.GetLastValueForAttribute("tokens_out", tokensSwappedEvt)
-	tokenOutPoolID := txModule.GetLastValueForAttribute("pool_id", tokensSwappedEvt)
+		// The last route in the hops gives the token out denom and pool ID for the final output
+		lastRoute := sf.OsmosisMsgSwapExactAmountIn.Routes[len(sf.OsmosisMsgSwapExactAmountIn.Routes)-1]
+		lastRouteDenom := lastRoute.TokenOutDenom
+		lastRoutePoolID := lastRoute.PoolId
 
-	tokenOut, err := sdk.ParseCoinNormalized(tokenOutStr)
-	if err != nil {
-		return &txModule.MessageLogFormatError{MessageType: msgType, Log: fmt.Sprintf("%+v", log)}
-	}
+		tokenOutStr := txModule.GetLastValueForAttribute("tokens_out", tokensSwappedEvt)
+		tokenOutPoolID := txModule.GetLastValueForAttribute("pool_id", tokensSwappedEvt)
 
-	// Sanity check last route swap
-	if tokenOut.Denom != lastRouteDenom || strconv.FormatUint(lastRoutePoolID, 10) != tokenOutPoolID {
-		return &txModule.MessageLogFormatError{MessageType: msgType, Log: fmt.Sprintf("%+v", log)}
-	}
+		tokenOut, err := sdk.ParseCoinNormalized(tokenOutStr)
+		if err != nil {
+			return &txModule.MessageLogFormatError{MessageType: msgType, Log: fmt.Sprintf("%+v", log)}
+		}
 
-	sf.TokenOut = tokenOut
+		// Sanity check last route swap
+		if tokenOut.Denom != lastRouteDenom || strconv.FormatUint(lastRoutePoolID, 10) != tokenOutPoolID {
+			return &txModule.MessageLogFormatError{MessageType: msgType, Log: fmt.Sprintf("%+v", log)}
+		}
 
-	return err
+		sf.TokenOut = tokenOut
+
+		return nil
+	case transferEvt != nil:
+		transferEvts, err := txModule.ParseTransferEvent(*transferEvt)
+		if err != nil {
+			return err
+		}
+
+		// The last transfer event should contain the final transfer to the sender
+		lastTransferEvt := transferEvts[len(transferEvts)-1]
+
+		if lastTransferEvt.Recipient != sf.Address {
+			return errors.New("transfer event recipient does not match message sender")
+		}
+
+		tokenOut, err := sdk.ParseCoinNormalized(lastTransferEvt.Amount)
+		if err != nil {
+			return err
+		}
+
+		// The last route in the hops gives the token out denom and pool ID for the final output
+		lastRoute := sf.OsmosisMsgSwapExactAmountIn.Routes[len(sf.OsmosisMsgSwapExactAmountIn.Routes)-1]
+		lastRouteDenom := lastRoute.TokenOutDenom
+
+		if tokenOut.Denom != lastRouteDenom {
+			return errors.New("final transfer denom does not match last route denom")
+		}
+
+		sf.TokenOut = tokenOut
+
+		return nil
+
+	default:
+		return errors.New("no processable events for poolmanager MsgSwapExactAmountIn")
+	}
 }
 
 // This code is currently untested since I cannot find a TX execution for this