Fix open issues in TTP classic operation logic that were found as a part of full history verification (#5641)

* Format error message to include more info

* Address suble difference in behavior for LiquidityPoolWithdraw operation

* Make transaction index and operation index ot be 1-indexed

Author: karthikiyer56

ingest/processors/token_transfer/helpers.go

@@ -166,3 +166,11 @@ func getImpactedLiquidityPoolEntriesFromOperation(tx ingest.LedgerTransaction, o
 
 	return entries, nil
 }
+
+func formatError(err error, tx ingest.LedgerTransaction, opIndex uint32, op xdr.Operation) error {
+	if err == nil {
+		return nil
+	}
+	return fmt.Errorf("failed to process token transfer events for ledgerSequence: %v, txHash: %v, operationIndex: %v, operationType: %v. error: %w",
+		tx.Ledger.LedgerSequence(), tx.Hash.HexString(), opIndex, op.Body.Type.String(), err)
+}

ingest/processors/token_transfer/token_transfer_event.go

@@ -84,12 +84,20 @@ func NewFeeEvent(meta *EventMeta, from string, amount string, token *assetProto.
 }
 
 func NewEventMetaFromTx(tx ingest.LedgerTransaction, operationIndex *uint32, contractAddress string) *EventMeta {
+	// The input operationIndex is 0-indexed.
+	// As per SEP-35, the OperationIndex in the output proto should be 1-indexed.
+	// Make that conversion here
+	var outputOpIndex *uint32
+	if operationIndex != nil {
+		temp := *operationIndex + 1
+		outputOpIndex = &temp
+	}
 	return &EventMeta{
 		LedgerSequence:   tx.Ledger.LedgerSequence(),
 		ClosedAt:         timestamppb.New(tx.Ledger.ClosedAt()),
 		TxHash:           tx.Hash.HexString(),
-		TransactionIndex: tx.Index - 1,
-		OperationIndex:   operationIndex,
+		TransactionIndex: tx.Index, // The index in ingest.LedgerTransaction is already 1-indexed
+		OperationIndex:   outputOpIndex,
 		ContractAddress:  contractAddress,
 	}
 }

ingest/processors/token_transfer/token_transfer_event.pb.go

@@ -70,7 +70,7 @@ func (p *EventsProcessor) EventsFromLedger(lcm xdr.LedgerCloseMeta) ([]*TokenTra
 		}
 		txEvents, err = p.EventsFromTransaction(tx)
 		if err != nil {
-			return nil, fmt.Errorf("error processing token transfer events from transaction: %w", err)
+			return nil, err
 		}
 		events = append(events, txEvents...)
 	}
@@ -105,8 +105,7 @@ func (p *EventsProcessor) EventsFromTransaction(tx ingest.LedgerTransaction) ([]
 		// Process the operation and collect events
 		opEvents, err := p.EventsFromOperation(tx, uint32(i), op, opResult)
 		if err != nil {
-			return nil,
-				fmt.Errorf("error processing token transfer events from operation, index: %d, %s: %w", i, op.Body.Type.String(), err)
+			return nil, err
 		}
 
 		events = append(events, opEvents...)
@@ -167,7 +166,7 @@ func (p *EventsProcessor) EventsFromOperation(tx ingest.LedgerTransaction, opInd
 	}
 
 	if err != nil {
-		return nil, err
+		return nil, formatError(err, tx, opIndex, op)
 	}
 
 	// DO not run this reconciliation check for ledgers with protocol version >= 8
@@ -178,7 +177,7 @@ func (p *EventsProcessor) EventsFromOperation(tx ingest.LedgerTransaction, opInd
 	// Run reconciliation for all operations except InvokeHostFunction
 	reconciliationEvent, err := p.generateXlmReconciliationEvents(tx, opIndex, op, events)
 	if err != nil {
-		return nil, fmt.Errorf("error generating reconciliation events: %w", err)
+		return nil, formatError(fmt.Errorf("error generating reconciliation events: %w", err), tx, opIndex, op)
 	}
 
 	if reconciliationEvent != nil {
@@ -189,7 +188,7 @@ func (p *EventsProcessor) EventsFromOperation(tx ingest.LedgerTransaction, opInd
 			// If it is a burn, put the burn event at the end of the list of other events for this operation
 			events = append(events, reconciliationEvent)
 		} else {
-			return nil, fmt.Errorf("invalid reconciliation event type: %v. reconciliation event type can be only mint or burn", reconciliationEvent.GetEventType())
+			return nil, formatError(fmt.Errorf("invalid reconciliation event type: %v. reconciliation event type can be only mint or burn", reconciliationEvent.GetEventType()), tx, opIndex, op)
 		}
 	}
 
@@ -210,16 +209,15 @@ Those 2 will call the underlying proto function for clawback
 */
 func (p *EventsProcessor) mintOrBurnOrTransferEvent(tx ingest.LedgerTransaction, opIndex *uint32, asset xdr.Asset, from addressWrapper, to addressWrapper, amt string) (*TokenTransferEvent, error) {
 	var fromAddress, toAddress string
-	// no need to have a separate flag for transferEvent. if neither burn nor mint, then it is regular transfer
-	var isMintEvent, isBurnEvent bool
+	var isFromIssuer, isToIssuer bool
 
 	assetIssuerAccountId, _ := asset.GetIssuerAccountId()
 
 	// Checking 'from' address
 	if from.account != nil {
 		fromAddress = protoAddressFromAccount(*from.account)
 		if !asset.IsNative() && assetIssuerAccountId.Equals(from.account.ToAccountId()) {
-			isMintEvent = true
+			isFromIssuer = true
 		}
 	} else if from.liquidityPoolId != nil {
 		fromAddress = lpIdToStrkey(*from.liquidityPoolId)
@@ -231,7 +229,7 @@ func (p *EventsProcessor) mintOrBurnOrTransferEvent(tx ingest.LedgerTransaction,
 	if to.account != nil {
 		toAddress = protoAddressFromAccount(*to.account)
 		if !asset.IsNative() && assetIssuerAccountId.Equals(to.account.ToAccountId()) {
-			isBurnEvent = true
+			isToIssuer = true
 		}
 	} else if to.liquidityPoolId != nil {
 		toAddress = lpIdToStrkey(*to.liquidityPoolId)
@@ -245,18 +243,18 @@ func (p *EventsProcessor) mintOrBurnOrTransferEvent(tx ingest.LedgerTransaction,
 	// This means that the payment is a wierd one, where the src == dest AND in addition, the src/dest address is the issuer of the asset
 	// Check this section out in CAP-67 https://github.com/stellar/stellar-protocol/blob/master/core/cap-0067.md#payment
 	// We need to issue a TRANSFER event for this.
-	// Keep in mind though that this wont show up in opeartionMeta as a balance change
+	// Keep in mind though that this wont show up in operationMeta as a balance change
 	// This has happened in ledgerSequence: 4522126 on pubnet
-	if isMintEvent && isBurnEvent {
+	if isFromIssuer && isToIssuer {
 		return NewTransferEvent(meta, fromAddress, toAddress, amt, protoAsset), nil
-	} else if isMintEvent {
+	} else if isFromIssuer {
 
 		// Check for Mint Event
 		if toAddress == "" {
 			return nil, NewEventError("mint event error: to address is nil")
 		}
 		return NewMintEvent(meta, toAddress, amt, protoAsset), nil
-	} else if isBurnEvent {
+	} else if isToIssuer {
 
 		// Check for Burn Event
 		if fromAddress == "" {
@@ -265,6 +263,12 @@ func (p *EventsProcessor) mintOrBurnOrTransferEvent(tx ingest.LedgerTransaction,
 		return NewBurnEvent(meta, fromAddress, amt, protoAsset), nil
 	}
 
+	if fromAddress == "" {
+		return nil, NewEventError("transfer event error: from address is nil")
+	}
+	if toAddress == "" {
+		return nil, NewEventError("transfer event error: to address is nil")
+	}
 	// Create transfer event
 	return NewTransferEvent(meta, fromAddress, toAddress, amt, protoAsset), nil
 }
@@ -598,11 +602,11 @@ func (p *EventsProcessor) liquidityPoolDepositEvents(tx ingest.LedgerTransaction
 	amtA, amtB := delta.amountChangeForAssetA, delta.amountChangeForAssetB
 	if amtA <= 0 {
 		return nil,
-			fmt.Errorf("deposited amount (%v) for asset: %v, cannot be zero or negative in LiquidityPool: %v", amtA, assetA.String(), lpIdToStrkey(lpId))
+			fmt.Errorf("deposited amount (%v) for asset: %v, cannot be zero or negative for LiquidityPool: %v", amtA, assetA.String(), lpIdToStrkey(lpId))
 	}
 	if amtB <= 0 {
 		return nil,
-			fmt.Errorf("deposited amount (%v) for asset: %v, cannot be zero or negative in LiquidityPool: %v", amtB, assetB.String(), lpIdToStrkey(lpId))
+			fmt.Errorf("deposited amount (%v) for asset: %v, cannot be zero or negative for LiquidityPool: %v", amtB, assetB.String(), lpIdToStrkey(lpId))
 	}
 
 	opSrcAcc := operationSourceAccount(tx, op)
@@ -639,13 +643,19 @@ func (p *EventsProcessor) liquidityPoolWithdrawEvents(tx ingest.LedgerTransactio
 	lpId := delta.liquidityPoolId
 	assetA, assetB := delta.assetA, delta.assetB
 	amtA, amtB := delta.amountChangeForAssetA, delta.amountChangeForAssetB
-	if amtA <= 0 {
+	/*
+		This is slightly different from the LpDeposit operation check. In LpDeposit, if amt <=0: then error
+		However, for LpWithdraw, the check is if amt < 0: then error.
+		This is because, the rounding on withdraw could result in nothing being withdrawn
+		Refer https://github.com/sisuresh/stellar-protocol/blob/unified/core/cap-0038.md#price-bounds-for-liquiditypoolwithdrawop
+	*/
+	if amtA < 0 {
 		return nil,
-			fmt.Errorf("deposited amount (%v) for asset: %v, cannot be zero or negative in LiquidityPool: %v", amtA, assetA.String(), lpIdToStrkey(lpId))
+			fmt.Errorf("withdrawn amount (%v) for asset: %v, cannot be negative for LiquidityPool: %v", amtA, assetA.String(), lpIdToStrkey(lpId))
 	}
-	if amtB <= 0 {
+	if amtB < 0 {
 		return nil,
-			fmt.Errorf("deposited amount (%v) for asset: %v, cannot be zero or negative in LiquidityPool: %v", amtB, assetB.String(), lpIdToStrkey(lpId))
+			fmt.Errorf("withdrawn amount (%v) for asset: %v, cannot be negative for LiquidityPool: %v", amtB, assetB.String(), lpIdToStrkey(lpId))
 	}
 
 	opSrcAcc := operationSourceAccount(tx, op)

ingest/processors/token_transfer/token_transfer_processor.go

@@ -12,9 +12,9 @@ message EventMeta {
   uint32 ledger_sequence = 1;
   google.protobuf.Timestamp closed_at = 2;
   string tx_hash = 3;
-  // Index of transaction within this ledger. 0-indexed
+  // Index of transaction within this ledger. This is 1-indexed as per SEP-35
   uint32 transaction_index = 4;
-  // Index of operation within the transaction, if it is not a fee event. 0-indexed
+  // Index of operation within the transaction, if it is not a fee event. This is 1-indexed as per SEP-35
   optional uint32 operation_index = 5;
   /* For a classic operation, this contract_address is the ContractID as derived from asset
      For a smart contract event, this contractId is the id of the contract emiting the core event