aergoio/aergo

/*
 * @file
 * @copyright defined in aergo/LICENSE.txt
 */

package p2p

import (
    "bytes"
    "time"

    "github.com/aergoio/aergo-lib/log"
    "github.com/aergoio/aergo/v2/internal/enc/base58"
    "github.com/aergoio/aergo/v2/p2p/p2pcommon"
    "github.com/aergoio/aergo/v2/p2p/p2putil"
    "github.com/aergoio/aergo/v2/types"
    "github.com/aergoio/aergo/v2/types/message"
)

// GetTxsReceiver is send p2p getTXsRequest to target peer and receive p2p responses till all requests transactions are received
// syncer actor already dropped wait before.
type GetTxsReceiver struct {
    requestID p2pcommon.MsgID
    logger    *log.Logger

    peer  p2pcommon.RemotePeer
    actor p2pcommon.ActorService
    sm    p2pcommon.SyncManager

    ids    []types.TxID
    hashes [][]byte
    missed [][]byte

    timeout  time.Time
    finished bool
    status   receiverStatus

    inOffset       int
    offset         int
    sent           int
    senderFinished chan interface{}
}

func NewGetTxsReceiver(actor p2pcommon.ActorService, peer p2pcommon.RemotePeer, sm p2pcommon.SyncManager, logger *log.Logger, txIDs []types.TxID, ttl time.Duration) *GetTxsReceiver {
    timeout := time.Now().Add(ttl)
    ids := make([]types.TxID, len(txIDs))
    hashes := make([][]byte, len(txIDs))
    for i := range txIDs {
        ids[i] = txIDs[i]
        hashes[i] = ids[i][:]
    }
    return &GetTxsReceiver{actor: actor, peer: peer, sm: sm, ids: ids, hashes: hashes, timeout: timeout, logger: logger}
}

func (br *GetTxsReceiver) StartGet() {
    // create message data
    req := &types.GetTransactionsRequest{Hashes: br.hashes}
    mo := br.peer.MF().NewMsgRequestOrderWithReceiver(br.ReceiveResp, p2pcommon.GetTXsRequest, req)
    br.peer.SendMessage(mo)
    br.requestID = mo.GetMsgID()
}

// ReceiveResp must be called just in read go routine
func (br *GetTxsReceiver) ReceiveResp(msg p2pcommon.Message, msgBody p2pcommon.MessageBody) (ret bool) {
    // cases in waiting
    //   normal not status => wait
    //   normal status (last response)  => finish
    //   abnormal resp (no following resp expected): hasNext is true => cancel
    //   abnormal resp (following resp expected): hasNext is false, or invalid resp data type (maybe remote peer is totally broken) => cancel finish
    // case in status or status
    ret = true
    switch br.status {
    case receiverStatusWaiting:
        br.handleInWaiting(msg, msgBody)
    case receiverStatusCanceled:
        br.ignoreMsg(msg, msgBody)
        return
    case receiverStatusFinished:
        fallthrough
    default:
        return
    }
    return
}

func (br *GetTxsReceiver) handleInWaiting(msg p2pcommon.Message, msgBody p2pcommon.MessageBody) {
    // consuming request id when timeout, no more resp expected (i.e. hasNext == false ) or malformed body.
    // timeout
    if br.timeout.Before(time.Now()) {
        // silently ignore already status job
        br.finishReceiver()
        return
    }
    // malformed responses means that later responses will be also malformed..
    respBody, ok := msgBody.(types.ResponseMessage)
    if !ok || respBody.GetStatus() != types.ResultStatus_OK {
        br.cancelReceiving(message.RemotePeerFailError, false)
        if respBody.GetStatus() == types.ResultStatus_RESOURCE_EXHAUSTED {
            br.sm.RetryGetTx(br.peer, br.hashes)
        }
        return
    }
    // remote peer response malformed data.
    body, ok := msgBody.(*types.GetTransactionsResponse)
    if !ok || len(body.Txs) == 0 {
        br.cancelReceiving(message.MissingHashError, false)
        return
    }

    // add to Got
    for _, tx := range body.Txs {
        // It also error that response has more blocks than expected(=requested).
        if br.offset >= len(br.hashes) {
            br.cancelReceiving(message.TooManyBlocksError, body.HasNext)
            return
        }
        // missing tx
        for !bytes.Equal(br.hashes[br.offset], tx.Hash) {
            br.logger.Trace().Str("expect", base58.Encode(br.hashes[br.offset])).Str("received", base58.Encode(tx.Hash)).Int("offset", br.offset).Msg("expected hash was missing")
            br.missed = append(br.missed, tx.Hash)
            br.offset++
            if br.offset >= len(br.hashes) {
                br.cancelReceiving(message.UnexpectedBlockError, body.HasNext)
                return
            }
        }
        br.actor.SendRequest(message.MemPoolSvc, &message.MemPoolPut{Tx: tx})
        br.sent++
        br.offset++
    }
    // remote peer hopefully sent last chunk
    if !body.HasNext {
        br.finishReceiver()
    }
    return
}

// cancelReceiving is cancel wait for receiving and send syncer the failure result.
// not all part of response is received, it wait remaining (and useless) response. It is assumed cancelling is not frequently occur
func (br *GetTxsReceiver) cancelReceiving(err error, hasNext bool) {
    br.status = receiverStatusCanceled
    br.logger.Info().Str(p2putil.LogOrgReqID, br.requestID.String()).Err(err).Msg("tx receiver canceled by error")
    // check time again. since negative duration of timer will not fire channel.
    interval := br.timeout.Sub(time.Now())
    if !hasNext || interval <= 0 {
        // if remote peer will not send partial response anymore. it it actually same as finish.
        br.finishReceiver()
    } else {
        // canceling in the middle of responses
        br.senderFinished = make(chan interface{})
        go func() {
            timer := time.NewTimer(interval)
            select {
            case <-timer.C:
                break
            case <-br.senderFinished:
                break
            }
            br.peer.ConsumeRequest(br.requestID)
        }()
    }
}

// finishReceiver is to cancel works, assuming cancellations are not frequently occur
func (br *GetTxsReceiver) finishReceiver() {
    br.status = receiverStatusFinished
    br.peer.ConsumeRequest(br.requestID)
    br.logger.Debug().Int("mpSent", br.sent).Int("missed", len(br.missed)).Str(p2putil.LogOrgReqID, br.requestID.String()).Msg("tx receiver finished")
}

// ignoreMsg is silently ignore following responses, which is not useless anymore.
func (br *GetTxsReceiver) ignoreMsg(msg p2pcommon.Message, msgBody p2pcommon.MessageBody) {
    body, ok := msgBody.(*types.GetTransactionsResponse)
    if !ok {
        return
    }
    if !body.HasNext {
        // really status from remote peer
        select {
        case br.senderFinished <- struct{}{}:
        default:
        }
    }
}