[snapshot] close stream when fail to get srcConn in errgroup (#2409)

if pull connector fails to initialize due to some reason and sync is
already stuck on `chan receive`, then partClone workflow will hang
indefinitely. fixed by closing stream explicitly before pull side
errgroup goroutine returns.

flow/activities/flowable.go

@@ -47,6 +47,10 @@ type FlowableActivity struct {
 	OtelManager *otel_metrics.OtelManager
 }
 
+type StreamCloser interface {
+	Close(error)
+}
+
 func (a *FlowableActivity) CheckConnection(
 	ctx context.Context,
 	config *protos.SetupInput,

flow/activities/flowable_core.go

@@ -387,7 +387,7 @@ func (a *FlowableActivity) getPostgresPeerConfigs(ctx context.Context) ([]*proto
 }
 
 // replicateQRepPartition replicates a QRepPartition from the source to the destination.
-func replicateQRepPartition[TRead any, TWrite any, TSync connectors.QRepSyncConnectorCore, TPull connectors.QRepPullConnectorCore](
+func replicateQRepPartition[TRead any, TWrite StreamCloser, TSync connectors.QRepSyncConnectorCore, TPull connectors.QRepPullConnectorCore](
 	ctx context.Context,
 	a *FlowableActivity,
 	config *protos.QRepConfig,
@@ -440,6 +440,7 @@ func replicateQRepPartition[TRead any, TWrite any, TSync connectors.QRepSyncConn
 		srcConn, err := connectors.GetByNameAs[TPull](ctx, config.Env, a.CatalogPool, config.SourceName)
 		if err != nil {
 			a.Alerter.LogFlowError(ctx, config.FlowJobName, err)
+			stream.Close(err)
 			return fmt.Errorf("failed to get qrep source connector: %w", err)
 		}
 		defer connectors.CloseConnector(ctx, srcConn)