Clear out old instances from partial message management (#851)

* Clear out old instances from partial message management

Once a decision about an instance is received, remove all states related
to the older instances from partial message manager and chain exchange.

Part of #792

* Remove sleep from asserting epoch finalisation

Author: masih

Diff for: chainexchange/pubsub.go

@@ -321,8 +321,16 @@ func (p *PubSubChainExchange) cacheAsWantedChain(ctx context.Context, cmsg Messa
 func (p *PubSubChainExchange) RemoveChainsByInstance(_ context.Context, instance uint64) error {
 	p.mu.Lock()
 	defer p.mu.Unlock()
-	delete(p.chainsWanted, instance)
-	delete(p.chainsDiscovered, instance)
+	for i := range p.chainsWanted {
+		if i < instance {
+			delete(p.chainsWanted, i)
+		}
+	}
+	for i := range p.chainsDiscovered {
+		if i < instance {
+			delete(p.chainsDiscovered, i)
+		}
+	}
 	return nil
 }
 

Diff for: f3_test.go

@@ -541,7 +541,6 @@ func (e *testEnv) waitForEpochFinalized(epoch int64) {
 				//       here and reduce the clock advance to give messages a chance of being
 				//       delivered in time. See:
 				//         - https://github.com/filecoin-project/go-f3/issues/818
-				time.Sleep(20 * time.Millisecond)
 				for _, nd := range e.nodes {
 					if nd.f3 == nil || !nd.f3.IsRunning() {
 						continue

Diff for: host.go

@@ -788,7 +788,9 @@ func (h *gpbftHost) ReceiveDecision(decision *gpbft.Justification) (time.Time, e
 	log.Infow("reached a decision", "instance", decision.Vote.Instance,
 		"ecHeadEpoch", decision.Vote.Value.Head().Epoch)
 	if decision.Vote.Instance > 0 {
-		h.pmCache.RemoveGroupsLessThan(decision.Vote.Instance - 1)
+		oldInstance := decision.Vote.Instance - 1
+		h.pmCache.RemoveGroupsLessThan(oldInstance)
+		h.pmm.RemoveMessagesBeforeInstance(context.Background(), oldInstance)
 	}
 	cert, err := h.saveDecision(decision)
 	if err != nil {

Diff for: partial_msg.go

@@ -45,6 +45,8 @@ type partialMessageManager struct {
 	pendingDiscoveredChains chan *discoveredChain
 	// pendingChainBroadcasts is a channel of chains that are pending to be broadcasted.
 	pendingChainBroadcasts chan chainexchange.Message
+	// pendingInstanceRemoval is a channel of instances that are pending to be removed.
+	pendingInstanceRemoval chan uint64
 	// rebroadcastInterval is the interval at which chains are re-broadcasted.
 	rebroadcastInterval time.Duration
 
@@ -58,6 +60,7 @@ func newPartialMessageManager(progress gpbft.Progress, ps *pubsub.PubSub, m *man
 		pendingDiscoveredChains: make(chan *discoveredChain, 100),           // TODO: parameterize buffer size.
 		pendingPartialMessages:  make(chan *PartiallyValidatedMessage, 100), // TODO: parameterize buffer size.
 		pendingChainBroadcasts:  make(chan chainexchange.Message, 100),      // TODO: parameterize buffer size.
+		pendingInstanceRemoval:  make(chan uint64, 10),
 		rebroadcastInterval:     m.ChainExchange.RebroadcastInterval,
 	}
 	var err error
@@ -150,6 +153,23 @@ func (pmm *partialMessageManager) Start(ctx context.Context) (<-chan *PartiallyV
 				// TODO: Add equivocation metrics: check if the message is different and if so
 				//       increment the equivocations counter tagged by phase.
 				//       See: https://github.com/filecoin-project/go-f3/issues/812
+			case instance, ok := <-pmm.pendingInstanceRemoval:
+				if !ok {
+					return
+				}
+				for i := range pmm.pmByInstance {
+					if i < instance {
+						delete(pmm.pmByInstance, i)
+					}
+				}
+				for i := range pmm.pmkByInstanceByChainKey {
+					if i < instance {
+						delete(pmm.pmkByInstanceByChainKey, i)
+					}
+				}
+				if err := pmm.chainex.RemoveChainsByInstance(ctx, instance); err != nil {
+					log.Errorw("Failed to remove chains by instance form chainexchange.", "instance", instance, "error", err)
+				}
 			}
 		}
 	}()
@@ -366,6 +386,16 @@ func inferJustificationVoteValue(pgmsg *PartialGMessage) {
 	}
 }
 
+func (pmm *partialMessageManager) RemoveMessagesBeforeInstance(ctx context.Context, instance uint64) {
+	select {
+	case <-ctx.Done():
+		return
+	case pmm.pendingInstanceRemoval <- instance:
+	default:
+		log.Warnw("Dropped instance removal request as partial message manager is too slow.", "instance", instance)
+	}
+}
+
 func (pmm *partialMessageManager) Shutdown(ctx context.Context) error {
 	if pmm.stop != nil {
 		pmm.stop()