diff --git a/flow/connectors/kafka/kafka.go b/flow/connectors/kafka/kafka.go
index 89adbbd6a4..cfe4652598 100644
--- a/flow/connectors/kafka/kafka.go
+++ b/flow/connectors/kafka/kafka.go
@@ -169,11 +169,11 @@ func (c *KafkaConnector) createPool(
 	ctx context.Context,
 	script string,
 	flowJobName string,
-	wgErr func(error),
+	queueErr func(error),
 ) (*utils.LPool[[]*kgo.Record], error) {
 	produceCb := func(_ *kgo.Record, err error) {
 		if err != nil {
-			wgErr(err)
+			queueErr(err)
 		}
 	}
 
@@ -205,8 +205,8 @@ func (c *KafkaConnector) SyncRecords(ctx context.Context, req *model.SyncRecords
 	numRecords := atomic.Int64{}
 	tableNameRowsMapping := utils.InitialiseTableRowsMap(req.TableMappings)
 
-	wgCtx, wgErr := context.WithCancelCause(ctx)
-	pool, err := c.createPool(wgCtx, req.Script, req.FlowJobName, wgErr)
+	queueCtx, queueErr := context.WithCancelCause(ctx)
+	pool, err := c.createPool(queueCtx, req.Script, req.FlowJobName, queueErr)
 	if err != nil {
 		return nil, err
 	}
@@ -255,7 +255,7 @@ Loop:
 				lfn := ls.Env.RawGetString("onRecord")
 				fn, ok := lfn.(*lua.LFunction)
 				if !ok {
-					wgErr(fmt.Errorf("script should define `onRecord` as function, not %s", lfn))
+					queueErr(fmt.Errorf("script should define `onRecord` as function, not %s", lfn))
 					return nil
 				}
 
@@ -263,7 +263,7 @@ Loop:
 				ls.Push(pua.LuaRecord.New(ls, record))
 				err := ls.PCall(1, -1, nil)
 				if err != nil {
-					wgErr(fmt.Errorf("script failed: %w", err))
+					queueErr(fmt.Errorf("script failed: %w", err))
 					return nil
 				}
 
@@ -272,7 +272,7 @@ Loop:
 				for i := range args {
 					kr, err := lvalueToKafkaRecord(ls, ls.Get(i-args))
 					if err != nil {
-						wgErr(err)
+						queueErr(err)
 						return nil
 					}
 					if kr != nil {
@@ -289,16 +289,16 @@ Loop:
 				return results
 			})
 
-		case <-wgCtx.Done():
+		case <-queueCtx.Done():
 			break Loop
 		}
 	}
 
 	close(flushLoopDone)
-	if err := pool.Wait(wgCtx); err != nil {
+	if err := pool.Wait(queueCtx); err != nil {
 		return nil, err
 	}
-	if err := c.client.Flush(wgCtx); err != nil {
+	if err := c.client.Flush(queueCtx); err != nil {
 		return nil, fmt.Errorf("[kafka] final flush error: %w", err)
 	}
 
diff --git a/flow/connectors/kafka/qrep.go b/flow/connectors/kafka/qrep.go
index 60725ae842..a856ad1ccf 100644
--- a/flow/connectors/kafka/qrep.go
+++ b/flow/connectors/kafka/qrep.go
@@ -28,8 +28,8 @@ func (c *KafkaConnector) SyncQRepRecords(
 	numRecords := atomic.Int64{}
 	schema := stream.Schema()
 
-	wgCtx, wgErr := context.WithCancelCause(ctx)
-	pool, err := c.createPool(wgCtx, config.Script, config.FlowJobName, wgErr)
+	queueCtx, queueErr := context.WithCancelCause(ctx)
+	pool, err := c.createPool(queueCtx, config.Script, config.FlowJobName, queueErr)
 	if err != nil {
 		return 0, err
 	}
@@ -60,7 +60,7 @@ Loop:
 				lfn := ls.Env.RawGetString("onRecord")
 				fn, ok := lfn.(*lua.LFunction)
 				if !ok {
-					wgErr(fmt.Errorf("script should define `onRecord` as function, not %s", lfn))
+					queueErr(fmt.Errorf("script should define `onRecord` as function, not %s", lfn))
 					return nil
 				}
 
@@ -68,7 +68,7 @@ Loop:
 				ls.Push(pua.LuaRecord.New(ls, record))
 				err := ls.PCall(1, -1, nil)
 				if err != nil {
-					wgErr(fmt.Errorf("script failed: %w", err))
+					queueErr(fmt.Errorf("script failed: %w", err))
 					return nil
 				}
 
@@ -77,7 +77,7 @@ Loop:
 				for i := range args {
 					kr, err := lvalueToKafkaRecord(ls, ls.Get(i-args))
 					if err != nil {
-						wgErr(err)
+						queueErr(err)
 						return nil
 					}
 					if kr != nil {
@@ -92,15 +92,15 @@ Loop:
 				return results
 			})
 
-		case <-wgCtx.Done():
+		case <-queueCtx.Done():
 			break Loop
 		}
 	}
 
-	if err := pool.Wait(wgCtx); err != nil {
+	if err := pool.Wait(queueCtx); err != nil {
 		return 0, err
 	}
-	if err := c.client.Flush(wgCtx); err != nil {
+	if err := c.client.Flush(queueCtx); err != nil {
 		return 0, fmt.Errorf("[kafka] final flush error: %w", err)
 	}
 
diff --git a/flow/connectors/pubsub/pubsub.go b/flow/connectors/pubsub/pubsub.go
index 5cf3f579b7..54031f016d 100644
--- a/flow/connectors/pubsub/pubsub.go
+++ b/flow/connectors/pubsub/pubsub.go
@@ -127,7 +127,7 @@ func (c *PubSubConnector) createPool(
 	flowJobName string,
 	topiccache *topicCache,
 	publish chan<- *pubsub.PublishResult,
-	wgErr func(error),
+	queueErr func(error),
 ) (*utils.LPool[[]PubSubMessage], error) {
 	return utils.LuaPool(func() (*lua.LState, error) {
 		ls, err := utils.LoadScript(ctx, script, func(ls *lua.LState) int {
@@ -163,7 +163,7 @@ func (c *PubSubConnector) createPool(
 				return topicClient, nil
 			})
 			if err != nil {
-				wgErr(err)
+				queueErr(err)
 				return
 			}
 
@@ -228,8 +228,8 @@ func (c *PubSubConnector) SyncRecords(ctx context.Context, req *model.SyncRecord
 	publish := make(chan *pubsub.PublishResult, 32)
 	waitChan := make(chan struct{})
 
-	wgCtx, wgErr := context.WithCancelCause(ctx)
-	pool, err := c.createPool(wgCtx, req.Script, req.FlowJobName, &topiccache, publish, wgErr)
+	queueCtx, queueErr := context.WithCancelCause(ctx)
+	pool, err := c.createPool(queueCtx, req.Script, req.FlowJobName, &topiccache, publish, queueErr)
 	if err != nil {
 		return nil, err
 	}
@@ -238,7 +238,7 @@ func (c *PubSubConnector) SyncRecords(ctx context.Context, req *model.SyncRecord
 	go func() {
 		for curpub := range publish {
 			if _, err := curpub.Get(ctx); err != nil {
-				wgErr(err)
+				queueErr(err)
 				break
 			}
 		}
@@ -285,7 +285,7 @@ Loop:
 				lfn := ls.Env.RawGetString("onRecord")
 				fn, ok := lfn.(*lua.LFunction)
 				if !ok {
-					wgErr(fmt.Errorf("script should define `onRecord` as function, not %s", lfn))
+					queueErr(fmt.Errorf("script should define `onRecord` as function, not %s", lfn))
 					return nil
 				}
 
@@ -293,7 +293,7 @@ Loop:
 				ls.Push(pua.LuaRecord.New(ls, record))
 				err := ls.PCall(1, -1, nil)
 				if err != nil {
-					wgErr(fmt.Errorf("script failed: %w", err))
+					queueErr(fmt.Errorf("script failed: %w", err))
 					return nil
 				}
 
@@ -302,7 +302,7 @@ Loop:
 				for i := range args {
 					msg, err := lvalueToPubSubMessage(ls, ls.Get(i-args))
 					if err != nil {
-						wgErr(err)
+						queueErr(err)
 						return nil
 					}
 					if msg.Message != nil {
@@ -319,20 +319,20 @@ Loop:
 				return results
 			})
 
-		case <-wgCtx.Done():
+		case <-queueCtx.Done():
 			break Loop
 		}
 	}
 
 	close(flushLoopDone)
-	if err := pool.Wait(wgCtx); err != nil {
+	if err := pool.Wait(queueCtx); err != nil {
 		return nil, err
 	}
 	close(publish)
-	topiccache.Stop(wgCtx)
+	topiccache.Stop(queueCtx)
 	select {
-	case <-wgCtx.Done():
-		return nil, wgCtx.Err()
+	case <-queueCtx.Done():
+		return nil, queueCtx.Err()
 	case <-waitChan:
 	}
 
diff --git a/flow/connectors/pubsub/qrep.go b/flow/connectors/pubsub/qrep.go
index c6e1138304..c1f21edc4a 100644
--- a/flow/connectors/pubsub/qrep.go
+++ b/flow/connectors/pubsub/qrep.go
@@ -31,8 +31,8 @@ func (c *PubSubConnector) SyncQRepRecords(
 	publish := make(chan *pubsub.PublishResult, 32)
 	waitChan := make(chan struct{})
 
-	wgCtx, wgErr := context.WithCancelCause(ctx)
-	pool, err := c.createPool(wgCtx, config.Script, config.FlowJobName, &topiccache, publish, wgErr)
+	queueCtx, queueErr := context.WithCancelCause(ctx)
+	pool, err := c.createPool(queueCtx, config.Script, config.FlowJobName, &topiccache, publish, queueErr)
 	if err != nil {
 		return 0, err
 	}
@@ -41,7 +41,7 @@ func (c *PubSubConnector) SyncQRepRecords(
 	go func() {
 		for curpub := range publish {
 			if _, err := curpub.Get(ctx); err != nil {
-				wgErr(err)
+				queueErr(err)
 				break
 			}
 		}
@@ -73,7 +73,7 @@ Loop:
 				lfn := ls.Env.RawGetString("onRecord")
 				fn, ok := lfn.(*lua.LFunction)
 				if !ok {
-					wgErr(fmt.Errorf("script should define `onRecord` as function, not %s", lfn))
+					queueErr(fmt.Errorf("script should define `onRecord` as function, not %s", lfn))
 					return nil
 				}
 
@@ -81,7 +81,7 @@ Loop:
 				ls.Push(pua.LuaRecord.New(ls, record))
 				err := ls.PCall(1, -1, nil)
 				if err != nil {
-					wgErr(fmt.Errorf("script failed: %w", err))
+					queueErr(fmt.Errorf("script failed: %w", err))
 					return nil
 				}
 
@@ -90,7 +90,7 @@ Loop:
 				for i := range args {
 					msg, err := lvalueToPubSubMessage(ls, ls.Get(i-args))
 					if err != nil {
-						wgErr(err)
+						queueErr(err)
 						return nil
 					}
 					if msg.Message != nil {
@@ -105,19 +105,19 @@ Loop:
 				return results
 			})
 
-		case <-wgCtx.Done():
+		case <-queueCtx.Done():
 			break Loop
 		}
 	}
 
-	if err := pool.Wait(wgCtx); err != nil {
+	if err := pool.Wait(queueCtx); err != nil {
 		return 0, err
 	}
 	close(publish)
-	topiccache.Stop(wgCtx)
+	topiccache.Stop(queueCtx)
 	select {
-	case <-wgCtx.Done():
-		return 0, wgCtx.Err()
+	case <-queueCtx.Done():
+		return 0, queueCtx.Err()
 	case <-waitChan:
 	}