cancel pull context when we have already synced a partition

PeerDB-io · Dec 11, 2023 · 4c3bae3 · 4c3bae3
1 parent 73cbb66
commit 4c3bae3
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/flow/activities/flowable.go b/flow/activities/flowable.go
@@ -554,7 +554,8 @@ func (a *FlowableActivity) replicateQRepPartition(ctx context.Context,
 		return fmt.Errorf("failed to update start time for partition: %w", err)
 	}
 
-	srcConn, err := connectors.GetQRepPullConnector(ctx, config.SourcePeer)
+	pullCtx, pullCancel := context.WithCancel(ctx)
+	srcConn, err := connectors.GetQRepPullConnector(pullCtx, config.SourcePeer)
 	if err != nil {
 		return fmt.Errorf("failed to get qrep source connector: %w", err)
 	}
@@ -632,6 +633,7 @@ func (a *FlowableActivity) replicateQRepPartition(ctx context.Context,
 	}
 
 	if res == 0 {
+		pullCancel()
 		log.WithFields(log.Fields{
 			"flowName": config.FlowJobName,
 		}).Infof("no records to push for partition %s\n", partition.PartitionId)