From bbbbc3316b26fbb3ded2937a21b9989f437ce3f9 Mon Sep 17 00:00:00 2001
From: Kaushik Iska <iska.kaushik@gmail.com>
Date: Tue, 19 Dec 2023 14:58:42 -0500
Subject: [PATCH] Cancel PullCtx when sync fails

---
 flow/activities/flowable.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/flow/activities/flowable.go b/flow/activities/flowable.go
index 69628c28d5..1c5a27cf88 100644
--- a/flow/activities/flowable.go
+++ b/flow/activities/flowable.go
@@ -631,6 +631,7 @@ func (a *FlowableActivity) replicateQRepPartition(ctx context.Context,
 
 	rowsSynced, err := dstConn.SyncQRepRecords(config, partition, stream)
 	if err != nil {
+		pullCancel()
 		return fmt.Errorf("failed to sync records: %w", err)
 	}