DEEP-11107 Handle context done in case of vuln db download and layer …

…processing

pkg/fanal/artifact/image/image.go

@@ -135,6 +135,12 @@ func (a Artifact) Inspect(ctx context.Context) (types.ArtifactReference, error)
 	}
 
 	if err = a.inspect(ctx, missingImageKey, missingLayers, baseDiffIDs, layerKeyMap, configFile); err != nil {
+		// clean up cache to handle potentially corrupted data, due to abrupt termination in case of context deadline execeeded
+		if err == context.DeadlineExceeded {
+			if cErr := a.cache.DeleteBlobs(missingLayers); cErr != nil {
+				err = xerrors.Errorf(";unable to clean up layer cache %s;%s", cErr, err)
+			}
+		}
 		return types.ArtifactReference{}, xerrors.Errorf("analyze error: %w", err)
 	}
 
@@ -217,32 +223,47 @@ func (a Artifact) inspect(ctx context.Context, missingImage string, layerKeys, b
 
 	var osFound types.OS
 	p := parallel.NewPipeline(a.artifactOption.Parallel, false, layerKeys, func(ctx context.Context, layerKey string) (any, error) {
-		layer := layerKeyMap[layerKey]
-
-		// If it is a base layer, secret scanning should not be performed.
-		var disabledAnalyzers []analyzer.Type
-		if slices.Contains(baseDiffIDs, layer.DiffID) {
-			disabledAnalyzers = append(disabledAnalyzers, analyzer.TypeSecret)
+		done := make(chan error, 1)
+
+		go func() {
+			layer := layerKeyMap[layerKey]
+
+			// If it is a base layer, secret scanning should not be performed.
+			var disabledAnalyzers []analyzer.Type
+			if slices.Contains(baseDiffIDs, layer.DiffID) {
+				disabledAnalyzers = append(disabledAnalyzers, analyzer.TypeSecret)
+			}
+
+			layerInfo, err := a.inspectLayer(ctx, layer, disabledAnalyzers)
+			if err != nil {
+				done <- xerrors.Errorf("failed to analyze layer (%s): %w", layer.DiffID, err)
+			}
+			if err = a.cache.PutBlob(layerKey, layerInfo); err != nil {
+				done <- xerrors.Errorf("failed to store layer: %s in cache: %w", layerKey, err)
+			}
+			if lo.IsNotEmpty(layerInfo.OS) {
+				osFound = layerInfo.OS
+			}
+
+			done <- nil
+		}()
+
+		// handle context done
+		select {
+		case err := <-done:
+			if err != nil {
+				return nil, err
+			}
+		case <-ctx.Done():
+			return nil, ctx.Err()
 		}
 
-		layerInfo, err := a.inspectLayer(ctx, layer, disabledAnalyzers)
-		if err != nil {
-			return nil, xerrors.Errorf("failed to analyze layer (%s): %w", layer.DiffID, err)
-		}
-		if err = a.cache.PutBlob(layerKey, layerInfo); err != nil {
-			return nil, xerrors.Errorf("failed to store layer: %s in cache: %w", layerKey, err)
-		}
-		if lo.IsNotEmpty(layerInfo.OS) {
-			osFound = layerInfo.OS
-		}
 		return nil, nil
-
 	}, nil)
 
 	if err := p.Do(ctx); err != nil {
 		return xerrors.Errorf("pipeline error: %w", err)
 	}
-
 	if missingImage != "" {
 		if err := a.inspectConfig(ctx, missingImage, osFound, configFile); err != nil {
 			return xerrors.Errorf("unable to analyze config: %w", err)

pkg/oci/artifact.go

@@ -182,9 +182,22 @@ func (a *Artifact) download(ctx context.Context, layer v1.Layer, fileName, dir s
 		_ = os.RemoveAll(tempDir)
 	}()
 
+	done := make(chan error, 1)
+
 	// Download the layer content into a temporal file
-	if _, err = io.Copy(f, pr); err != nil {
-		return xerrors.Errorf("copy error: %w", err)
+	go func() {
+		_, err := io.Copy(f, pr)
+		done <- err
+	}()
+
+	// handle context done
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case err := <-done:
+		if err != nil {
+			return xerrors.Errorf("copy error: %w", err)
+		}
 	}
 
 	// Decompress the downloaded file if it is compressed and copy it into the dst