Merge pull request #6417 from filecoin-project/fix/broadcast-current-…

…message

fix: send the last gpbft message for each new participant

pkg/vf3/f3.go

@@ -163,20 +163,28 @@ func (fff *F3) runSigningLoop(ctx context.Context) {
 
 	msgCh := fff.inner.MessagesToSign()
 
-loop:
+	var mb *gpbft.MessageBuilder
+	alreadyParticipated := make(map[uint64]struct{})
 	for ctx.Err() == nil {
 		select {
 		case <-ctx.Done():
 			return
-		case mb, ok := <-msgCh:
-			if !ok {
-				continue loop
+		case <-fff.leaser.notifyParticipation:
+			if mb == nil {
+				continue
 			}
-			participants := fff.leaser.getParticipantsByInstance(mb.Payload.Instance)
-			for _, id := range participants {
-				if err := participateOnce(ctx, mb, id); err != nil {
-					log.Errorf("while participating for miner f0%d: %+v", id, err)
-				}
+		case mb = <-msgCh: // never closed
+			clear(alreadyParticipated)
+		}
+
+		participants := fff.leaser.getParticipantsByInstance(mb.Payload.Instance)
+		for _, id := range participants {
+			if _, ok := alreadyParticipated[id]; ok {
+				continue
+			} else if err := participateOnce(ctx, mb, id); err != nil {
+				log.Errorf("while participating for miner f0%d: %+v", id, err)
+			} else {
+				alreadyParticipated[id] = struct{}{}
 			}
 		}
 	}

pkg/vf3/participation_lease.go

@@ -22,6 +22,8 @@ type leaser struct {
 	issuer               peer.ID
 	status               f3Status
 	maxLeasableInstances uint64
+	// Signals that a lease was created and/or updated.
+	notifyParticipation chan struct{}
 }
 
 func newParticipationLeaser(nodeID peer.ID, status f3Status, maxLeasedInstances uint64) *leaser {
@@ -30,6 +32,7 @@ func newParticipationLeaser(nodeID peer.ID, status f3Status, maxLeasedInstances
 		issuer:               nodeID,
 		status:               status,
 		maxLeasableInstances: maxLeasedInstances,
+		notifyParticipation:  make(chan struct{}, 1),
 	}
 }
 
@@ -98,6 +101,10 @@ func (l *leaser) participate(ticket types.F3ParticipationTicket) (types.F3Partic
 		return types.F3ParticipationLease{}, types.ErrF3ParticipationTicketStartBeforeExisting
 	}
 	l.leases[newLease.MinerID] = newLease
+	select {
+	case l.notifyParticipation <- struct{}{}:
+	default:
+	}
 	return newLease, nil
 }