Merge pull request #2246 from TroyWind/fit-multiple-piece-into-a-sector

a sector contains multiple deals(pieces), the state of deals incorrect

chain/events/events_called.go

@@ -485,13 +485,15 @@ func (me *messageEvents) checkNewCalls(ts *types.TipSet) (map[triggerID]eventDat
 
 		for tid, matchFns := range me.matchers {
 			var matched bool
+			var once bool
 			for _, matchFn := range matchFns {
-				ok, err := matchFn(msg)
+				matchOne, ok, err := matchFn(msg)
 				if err != nil {
 					log.Errorf("event matcher failed: %s", err)
 					continue
 				}
 				matched = ok
+				once = matchOne
 
 				if matched {
 					break
@@ -500,7 +502,9 @@ func (me *messageEvents) checkNewCalls(ts *types.TipSet) (map[triggerID]eventDat
 
 			if matched {
 				res[tid] = msg
-				break
+				if once {
+					break
+				}
 			}
 		}
 	})
@@ -548,7 +552,7 @@ func (me *messageEvents) messagesForTs(ts *types.TipSet, consume func(*types.Mes
 // `curH`-`ts.Height` = `confidence`
 type MsgHandler func(msg *types.Message, rec *types.MessageReceipt, ts *types.TipSet, curH abi.ChainEpoch) (more bool, err error)
 
-type MsgMatchFunc func(msg *types.Message) (bool, error)
+type MsgMatchFunc func(msg *types.Message) (matchOnce bool, matched bool, err error)
 
 // Called registers a callback which is triggered when a specified method is
 //  called on an actor, or a timeout is reached.

chain/events/events_test.go

@@ -561,9 +561,9 @@ func TestAtChainedConfidenceNull(t *testing.T) {
 	require.Equal(t, false, reverted)
 }
 
-func matchAddrMethod(to address.Address, m abi.MethodNum) func(msg *types.Message) (bool, error) {
-	return func(msg *types.Message) (bool, error) {
-		return to == msg.To && m == msg.Method, nil
+func matchAddrMethod(to address.Address, m abi.MethodNum) func(msg *types.Message) (matchOnce bool, matched bool, err error) {
+	return func(msg *types.Message) (matchOnce bool, matched bool, err error) {
+		return true, to == msg.To && m == msg.Method, nil
 	}
 }
 

chain/events/utils.go

@@ -34,11 +34,11 @@ func (me *messageEvents) CheckMsg(ctx context.Context, smsg types.ChainMsg, hnd
 }
 
 func (me *messageEvents) MatchMsg(inmsg *types.Message) MsgMatchFunc {
-	return func(msg *types.Message) (bool, error) {
+	return func(msg *types.Message) (matchOnce bool, matched bool, err error) {
 		if msg.From == inmsg.From && msg.Nonce == inmsg.Nonce && !inmsg.Equals(msg) {
-			return false, xerrors.Errorf("matching msg %s from %s, nonce %d: got duplicate origin/nonce msg %d", inmsg.Cid(), inmsg.From, inmsg.Nonce, msg.Nonce)
+			return true, false, xerrors.Errorf("matching msg %s from %s, nonce %d: got duplicate origin/nonce msg %d", inmsg.Cid(), inmsg.From, inmsg.Nonce, msg.Nonce)
 		}
 
-		return inmsg.Equals(msg), nil
+		return true, inmsg.Equals(msg), nil
 	}
 }