fix: Fix Transform hang in CLI sync (#2001)

When CLI runs sync_v3 and a transformer is in the spec, if the
transformer errors at transform time, the CLI can hang forever.

This fix makes sure that the `.Transform()` function always returns
after an error.

Tested 20 times:

```bash
Loading spec(s) from cmd/testdata/transformer-errors.yml
Starting sync for: test (cloudquery/[email protected]) -> [test (cloudquery/[email protected])]
Error: failed to sync v3 source test: rpc error: code = Internal desc = failing at the transformer stage according to spec requirements
exit status 1
Loading spec(s) from cmd/testdata/transformer-errors.yml
Starting sync for: test (cloudquery/[email protected]) -> [test (cloudquery/[email protected])]
Error: failed to sync v3 source test: rpc error: code = Internal desc = failing at the transformer stage according to spec requirements
exit status 1
Loading spec(s) from cmd/testdata/transformer-errors.yml
Starting sync for: test (cloudquery/[email protected]) -> [test (cloudquery/[email protected])]
Error: failed to sync v3 source test: rpc error: code = Internal desc = failing at the transformer stage according to spec requirements
exit status 1
Loading spec(s) from cmd/testdata/transformer-errors.yml
Starting sync for: test (cloudquery/[email protected]) -> [test (cloudquery/[email protected])]
Error: failed to sync v3 source test: rpc error: code = Internal desc = failing at the transformer stage according to spec requirements
exit status 1
Loading spec(s) from cmd/testdata/transformer-errors.yml
Starting sync for: test (cloudquery/[email protected]) -> [test (cloudquery/[email protected])]
Error: failed to sync v3 source test: rpc error: code = Internal desc = failing at the transformer stage according to spec requirements
exit status 1
Loading spec(s) from cmd/testdata/transformer-errors.yml
Starting sync for: test (cloudquery/[email protected]) -> [test (cloudquery/[email protected])]
Error: failed to sync v3 source test: rpc error: code = Internal desc = failing at the transformer stage according to spec requirements
exit status 1
Loading spec(s) from cmd/testdata/transformer-errors.yml
Starting sync for: test (cloudquery/[email protected]) -> [test (cloudquery/[email protected])]
Error: failed to sync v3 source test: EOF
exit status 1
Loading spec(s) from cmd/testdata/transformer-errors.yml
Starting sync for: test (cloudquery/[email protected]) -> [test (cloudquery/[email protected])]
Error: failed to sync v3 source test: rpc error: code = Internal desc = failing at the transformer stage according to spec requirements
exit status 1
Loading spec(s) from cmd/testdata/transformer-errors.yml
Starting sync for: test (cloudquery/[email protected]) -> [test (cloudquery/[email protected])]
Error: failed to sync v3 source test: rpc error: code = Internal desc = failing at the transformer stage according to spec requirements
exit status 1
Loading spec(s) from cmd/testdata/transformer-errors.yml
Starting sync for: test (cloudquery/[email protected]) -> [test (cloudquery/[email protected])]
Error: failed to sync v3 source test: rpc error: code = Internal desc = failing at the transformer stage according to spec requirements
exit status 1
Loading spec(s) from cmd/testdata/transformer-errors.yml
Starting sync for: test (cloudquery/[email protected]) -> [test (cloudquery/[email protected])]
Error: failed to sync v3 source test: rpc error: code = Internal desc = failing at the transformer stage according to spec requirements
exit status 1
Loading spec(s) from cmd/testdata/transformer-errors.yml
Starting sync for: test (cloudquery/[email protected]) -> [test (cloudquery/[email protected])]
Error: failed to sync v3 source test: rpc error: code = Internal desc = failing at the transformer stage according to spec requirements
exit status 1
Loading spec(s) from cmd/testdata/transformer-errors.yml
Starting sync for: test (cloudquery/[email protected]) -> [test (cloudquery/[email protected])]
Error: failed to sync v3 source test: rpc error: code = Internal desc = failing at the transformer stage according to spec requirements
exit status 1
Loading spec(s) from cmd/testdata/transformer-errors.yml
Starting sync for: test (cloudquery/[email protected]) -> [test (cloudquery/[email protected])]
Error: failed to sync v3 source test: rpc error: code = Internal desc = failing at the transformer stage according to spec requirements
exit status 1
Loading spec(s) from cmd/testdata/transformer-errors.yml
Starting sync for: test (cloudquery/[email protected]) -> [test (cloudquery/[email protected])]
Error: failed to sync v3 source test: rpc error: code = Internal desc = failing at the transformer stage according to spec requirements
exit status 1
Loading spec(s) from cmd/testdata/transformer-errors.yml
Starting sync for: test (cloudquery/[email protected]) -> [test (cloudquery/[email protected])]
Error: failed to sync v3 source test: rpc error: code = Internal desc = failing at the transformer stage according to spec requirements
exit status 1
Loading spec(s) from cmd/testdata/transformer-errors.yml
Starting sync for: test (cloudquery/[email protected]) -> [test (cloudquery/[email protected])]
Error: failed to sync v3 source test: rpc error: code = Internal desc = failing at the transformer stage according to spec requirements
exit status 1
Loading spec(s) from cmd/testdata/transformer-errors.yml
Starting sync for: test (cloudquery/[email protected]) -> [test (cloudquery/[email protected])]
Error: failed to sync v3 source test: rpc error: code = Internal desc = failing at the transformer stage according to spec requirements
exit status 1
Loading spec(s) from cmd/testdata/transformer-errors.yml
Starting sync for: test (cloudquery/[email protected]) -> [test (cloudquery/[email protected])]
Error: failed to sync v3 source test: rpc error: code = Internal desc = failing at the transformer stage according to spec requirements
exit status 1
Loading spec(s) from cmd/testdata/transformer-errors.yml
Starting sync for: test (cloudquery/[email protected]) -> [test (cloudquery/[email protected])]
Error: failed to sync v3 source test: rpc error: code = Internal desc = failing at the transformer stage according to spec requirements
exit status 1
```

Author: marianogappa

internal/servers/plugin/v3/plugin.go

@@ -402,23 +402,21 @@ func (s *Server) Write(stream pb.Plugin_WriteServer) error {
 
 func (s *Server) Transform(stream pb.Plugin_TransformServer) error {
 	var (
-		recvRecords       = make(chan arrow.Record)
-		sendRecords       = make(chan arrow.Record)
-		pluginStopsWriter = make(chan struct{})
-		doneReading       = false
-		ctx               = stream.Context()
-		eg, gctx          = errgroup.WithContext(ctx)
+		recvRecords = make(chan arrow.Record)
+		sendRecords = make(chan arrow.Record)
+		ctx         = stream.Context()
+		eg, gctx    = errgroup.WithContext(ctx)
 	)
 
 	// Run the plugin's transform with both channels.
 	//
 	// When the plugin is done, it must return with either an error or nil.
 	// The plugin must not close either channel.
 	eg.Go(func() error {
-		err := s.Plugin.Transform(gctx, recvRecords, sendRecords)
-		close(pluginStopsWriter)
-		doneReading = true
-		return err
+		if err := s.Plugin.Transform(gctx, recvRecords, sendRecords); err != nil {
+			return status.Error(codes.Internal, err.Error())
+		}
+		return nil
 	})
 
 	// Write transformed records from transformer to destination.
@@ -429,21 +427,16 @@ func (s *Server) Transform(stream pb.Plugin_TransformServer) error {
 	// The reading never closes the writer, because it's up to the Plugin to decide when to finish
 	// writing, regardless of if the reading finished.
 	eg.Go(func() error {
-		for {
-			select {
-			case record := <-sendRecords:
-				recordBytes, err := pb.RecordToBytes(record)
-				if err != nil {
-					return status.Errorf(codes.Internal, "failed to convert record to bytes: %v", err)
-				}
-
-				if err := stream.Send(&pb.Transform_Response{Record: recordBytes}); err != nil {
-					return fmt.Errorf("error sending response: %w", err)
-				}
-			case <-pluginStopsWriter:
-				return nil
+		for record := range sendRecords {
+			recordBytes, err := pb.RecordToBytes(record)
+			if err != nil {
+				return status.Errorf(codes.Internal, "failed to convert record to bytes: %v", err)
+			}
+			if err := stream.Send(&pb.Transform_Response{Record: recordBytes}); err != nil {
+				return status.Errorf(codes.Internal, "error sending response: %v", err)
 			}
 		}
+		return nil
 	})
 
 	// Read records from source to transformer
@@ -463,18 +456,23 @@ func (s *Server) Transform(stream pb.Plugin_TransformServer) error {
 			}
 			if err != nil {
 				close(recvRecords)
-				return fmt.Errorf("Error receiving request: %v", err)
-			}
-			if doneReading {
-				return nil
+				return status.Errorf(codes.Internal, "Error receiving request: %v", err)
 			}
 			record, err := pb.NewRecordFromBytes(req.Record)
 			if err != nil {
 				close(recvRecords)
 				return status.Errorf(codes.InvalidArgument, "failed to create record: %v", err)
 			}
 
-			recvRecords <- record
+			select {
+			case recvRecords <- record:
+			case <-gctx.Done():
+				close(recvRecords)
+				return status.Errorf(codes.Canceled, "context done: %v", gctx.Err())
+			case <-ctx.Done():
+				close(recvRecords)
+				return status.Errorf(codes.Canceled, "context done: %v", ctx.Err())
+			}
 		}
 	})
 

plugin/plugin_transformer.go

@@ -12,7 +12,9 @@ type TransformerClient interface {
 }
 
 func (p *Plugin) Transform(ctx context.Context, recvRecords <-chan arrow.Record, sendRecords chan<- arrow.Record) error {
-	return p.client.Transform(ctx, recvRecords, sendRecords)
+	err := p.client.Transform(ctx, recvRecords, sendRecords)
+	close(sendRecords)
+	return err
 }
 func (p *Plugin) TransformSchema(ctx context.Context, old *arrow.Schema) (*arrow.Schema, error) {
 	return p.client.TransformSchema(ctx, old)