fix: add cancel context for the parallel build flow (#110)

Signed-off-by: chlins <[email protected]>

Author: chlins

pkg/backend/processor/base.go

@@ -72,25 +72,36 @@ func (b *base) Process(ctx context.Context, builder build.Builder, workDir strin
 	var (
 		idx         atomic.Int64
 		mu          sync.Mutex
-		eg          errgroup.Group
+		eg          *errgroup.Group
 		descriptors []ocispec.Descriptor
 	)
 
+	total := int64(len(matchedPaths))
+	sm := ysmrr.NewSpinnerManager()
+	sm.Start()
+	defer sm.Stop()
+
+	// Initialize errgroup with a context can be canceled.
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	eg, ctx = errgroup.WithContext(ctx)
+
 	// Set default concurrency limit to 1 if not specified.
 	if baseOpts.concurrency > 0 {
 		eg.SetLimit(baseOpts.concurrency)
 	} else {
 		eg.SetLimit(1)
 	}
 
-	total := int64(len(matchedPaths))
-	sm := ysmrr.NewSpinnerManager()
-	sm.Start()
-
 	for _, path := range matchedPaths {
+		if ctx.Err() != nil {
+			break
+		}
+
 		eg.Go(func() error {
 			relPath, err := filepath.Rel(absWorkDir, path)
 			if err != nil {
+				cancel()
 				return err
 			}
 
@@ -100,6 +111,7 @@ func (b *base) Process(ctx context.Context, builder build.Builder, workDir strin
 			desc, err := builder.BuildLayer(ctx, b.mediaType, workDir, path)
 			if err != nil {
 				sp.ErrorWithMessagef("Failed to build blob %s: %v", relPath, err)
+				cancel()
 				return err
 			}
 
@@ -117,8 +129,6 @@ func (b *base) Process(ctx context.Context, builder build.Builder, workDir strin
 		return nil, err
 	}
 
-	sm.Stop()
-
 	sort.Slice(descriptors, func(i int, j int) bool {
 		// Sort by filepath by default.
 		var pathI, pathJ string

pkg/backend/processor/code_test.go

@@ -58,7 +58,7 @@ func (s *codeProcessorSuite) TestName() {
 
 func (s *codeProcessorSuite) TestProcess() {
 	ctx := context.Background()
-	s.mockBuilder.On("BuildLayer", ctx, mock.Anything, mock.Anything, mock.Anything).Return(ocispec.Descriptor{
+	s.mockBuilder.On("BuildLayer", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(ocispec.Descriptor{
 		Digest: godigest.Digest("sha256:1234567890abcdef"),
 		Size:   int64(1024),
 		Annotations: map[string]string{

pkg/backend/processor/doc_test.go

@@ -58,7 +58,7 @@ func (s *docProcessorSuite) TestName() {
 
 func (s *docProcessorSuite) TestProcess() {
 	ctx := context.Background()
-	s.mockBuilder.On("BuildLayer", ctx, mock.Anything, mock.Anything, mock.Anything).Return(ocispec.Descriptor{
+	s.mockBuilder.On("BuildLayer", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(ocispec.Descriptor{
 		Digest: godigest.Digest("sha256:1234567890abcdef"),
 		Size:   int64(1024),
 		Annotations: map[string]string{

pkg/backend/processor/model_config_test.go

@@ -58,7 +58,7 @@ func (s *modelConfigProcessorSuite) TestName() {
 
 func (s *modelConfigProcessorSuite) TestProcess() {
 	ctx := context.Background()
-	s.mockBuilder.On("BuildLayer", ctx, mock.Anything, mock.Anything, mock.Anything).Return(ocispec.Descriptor{
+	s.mockBuilder.On("BuildLayer", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(ocispec.Descriptor{
 		Digest: godigest.Digest("sha256:1234567890abcdef"),
 		Size:   int64(1024),
 		Annotations: map[string]string{

pkg/backend/processor/model_test.go

@@ -58,7 +58,7 @@ func (s *modelProcessorSuite) TestName() {
 
 func (s *modelProcessorSuite) TestProcess() {
 	ctx := context.Background()
-	s.mockBuilder.On("BuildLayer", ctx, mock.Anything, mock.Anything, mock.Anything).Return(ocispec.Descriptor{
+	s.mockBuilder.On("BuildLayer", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(ocispec.Descriptor{
 		Digest: godigest.Digest("sha256:1234567890abcdef"),
 		Size:   int64(1024),
 		Annotations: map[string]string{