Merge pull request #224 from meshplus/fix/rollback-ibtp

fix(*): recover pool index on pier restart

Author: jzhe886

internal/exchanger/handler.go

@@ -88,6 +88,16 @@ func (ex *Exchanger) applyReceipt(wIbtp *model.WrappedIBTP, entry logrus.FieldLo
 
 func (ex *Exchanger) applyInterchain(wIbtp *model.WrappedIBTP, entry logrus.FieldLogger) {
 	ibtp := wIbtp.Ibtp
+	_, ok := ex.serviceMeta[ibtp.To]
+	if !ok {
+		ex.serviceMeta[ibtp.To] = &pb.Interchain{
+			ID:                      ibtp.To,
+			InterchainCounter:       make(map[string]uint64),
+			ReceiptCounter:          make(map[string]uint64),
+			SourceInterchainCounter: make(map[string]uint64),
+			SourceReceiptCounter:    make(map[string]uint64),
+		}
+	}
 	index := ex.serviceMeta[ibtp.To].SourceInterchainCounter[ibtp.From]
 	if index >= ibtp.Index {
 		entry.Infof("Ignore ibtp, expected %d", index+1)

internal/exchanger/receipt.go

@@ -8,16 +8,7 @@ import (
 )
 
 func (ex *Exchanger) feedIBTPReceipt(receipt *model.WrappedIBTP) {
-	var pool *Pool
-	servicePair := fmt.Sprintf("%s-%s", receipt.Ibtp.From, receipt.Ibtp.To)
-	act, loaded := ex.receipts.Load(servicePair)
-	if !loaded {
-		ex.logger.Infof("new pool for service pair %s at begin index %d", servicePair, ex.serviceMeta[receipt.Ibtp.From].ReceiptCounter[receipt.Ibtp.To]+1)
-		pool = NewPool(ex.serviceMeta[receipt.Ibtp.From].ReceiptCounter[receipt.Ibtp.To] + 1)
-		ex.receipts.Store(servicePair, pool)
-	} else {
-		pool = act.(*Pool)
-	}
+	pool, loaded := ex.loadPool(receipt.Ibtp.From, receipt.Ibtp.To)
 	pool.feed(receipt)
 
 	if !loaded {
@@ -70,3 +61,17 @@ func (ex *Exchanger) feedIBTPReceipt(receipt *model.WrappedIBTP) {
 		}(pool)
 	}
 }
+
+func (ex *Exchanger) loadPool(from, to string) (*Pool, bool) {
+	servicePair := fmt.Sprintf("%s-%s", from, to)
+	beginIndex := ex.serviceMeta[from].ReceiptCounter[to] + 1
+	if ex.callbackMeta[servicePair]+1 > beginIndex {
+		beginIndex = ex.callbackMeta[servicePair] + 1
+	}
+
+	ex.logger.Infof("new pool for service pair %s at begin index %d", servicePair, beginIndex)
+	pool := NewPool(beginIndex)
+	act, loaded := ex.receipts.LoadOrStore(servicePair, pool)
+
+	return act.(*Pool), loaded
+}