Preserve staging I/O failure causes (T-615)

2026-07-19 20:28:56 +08:00
parent d1603c52b6
commit df6c243b21
14 changed files with 551 additions and 76 deletions
@@ -33,12 +33,15 @@ var (
 	ErrAppManifestMissing       = errors.New("package app.json is missing")
 	ErrDestinationExists        = errors.New("staging destination already exists")
 	ErrArchiveCorrupt           = errors.New("ZIP archive data is corrupt")
+	ErrStagingOutput            = errors.New("staging output failed")
+	ErrStagingCleanup           = errors.New("staging cleanup failed")
 )

 // Extractor writes only payload/ contents from a pre-verified package ZIP.
 type Extractor struct {
 	limits     Limits
 	durability durabilityFence
+	files      stagingFileOperations
 }

 type ExtractResult struct {
@@ -87,7 +90,11 @@ func NewExtractor(limits Limits) (Extractor, error) {
 	if err := limits.validate(); err != nil {
 		return Extractor{}, err
 	}
-	return Extractor{limits: limits, durability: defaultDurability()}, nil
+	return Extractor{
+		limits:     limits,
+		durability: defaultDurability(),
+		files:      defaultStagingFileOperations(),
+	}, nil
 }

 // ExtractFile requires expectedPackageSize from the verified Catalog package.
@@ -136,6 +143,7 @@ func (extractor Extractor) extractPlan(
 	plan []plannedEntry,
 ) (result ExtractResult, err error) {
 	fence := effectiveDurability(extractor.durability)
+	files := effectiveStagingFileOperations(extractor.files)
 	destinationRoot, entrypointPath, err := planOutputPaths(
 		destination,
 		entrypoint,
@@ -145,19 +153,21 @@ func (extractor Extractor) extractPlan(
 		return ExtractResult{}, err
 	}

-	if err := os.MkdirAll(filepath.Dir(destinationRoot), 0o700); err != nil {
-		return ExtractResult{}, fmt.Errorf("create staging parent: %w", err)
+	if err := files.mkdirAll(filepath.Dir(destinationRoot), 0o700); err != nil {
+		return ExtractResult{}, stagingOutputError("create staging parent", err)
 	}
-	if err := os.Mkdir(destinationRoot, 0o700); err != nil {
+	if err := files.mkdir(destinationRoot, 0o700); err != nil {
 		if os.IsExist(err) {
 			return ExtractResult{}, ErrDestinationExists
 		}
-		return ExtractResult{}, fmt.Errorf("create staging destination: %w", err)
+		return ExtractResult{}, stagingOutputError("create staging destination", err)
 	}
 	complete := false
 	defer func() {
 		if !complete {
-			_ = os.RemoveAll(destinationRoot)
+			if cleanupErr := files.removeAll(destinationRoot); cleanupErr != nil {
+				err = errors.Join(err, stagingCleanupError(cleanupErr))
+			}
 		}
 	}()

@@ -167,31 +177,37 @@ func (extractor Extractor) extractPlan(
 			if entry.outputPath == "" {
 				continue
 			}
-			if err := os.MkdirAll(entry.targetPath, 0o700); err != nil {
-				return ExtractResult{}, fmt.Errorf("create staging directory: %w", err)
+			if err := files.mkdirAll(entry.targetPath, 0o700); err != nil {
+				return ExtractResult{}, stagingOutputError("create staging directory", err)
 			}
 			continue
 		}
-		if err := os.MkdirAll(filepath.Dir(entry.targetPath), 0o700); err != nil {
-			return ExtractResult{}, fmt.Errorf("create staging file parent: %w", err)
+		if err := files.mkdirAll(filepath.Dir(entry.targetPath), 0o700); err != nil {
+			return ExtractResult{}, stagingOutputError("create staging file parent", err)
 		}

 		source, err := entry.file.Open()
 		if err != nil {
-			return ExtractResult{}, fmt.Errorf("%w: open %s: %v", ErrArchiveCorrupt, entry.archivePath, err)
+			return ExtractResult{}, archiveInputError("open", entry.archivePath, err)
 		}
 		mode := os.FileMode(0o600)
 		if entry.file.Mode().Perm()&0o111 != 0 {
 			mode = 0o700
 		}
-		output, err := os.OpenFile(
+		output, err := files.openFile(
 			entry.targetPath,
 			os.O_CREATE|os.O_EXCL|os.O_WRONLY,
 			mode,
 		)
 		if err != nil {
-			source.Close()
-			return ExtractResult{}, fmt.Errorf("create staging file: %w", err)
+			outputErr := stagingOutputError("create staging file", err)
+			if closeErr := source.Close(); closeErr != nil {
+				return ExtractResult{}, errors.Join(
+					outputErr,
+					archiveInputError("close", entry.archivePath, closeErr),
+				)
+			}
+			return ExtractResult{}, outputErr
 		}

 		remaining := extractor.limits.MaxUncompressedBytes - written
@@ -200,39 +216,45 @@ func (extractor Extractor) extractPlan(
 			readLimit++
 		}
 		digest := sha256.New()
+		writer := stagingWriter{writer: output}
 		copied, copyErr := io.Copy(
-			io.MultiWriter(output, digest),
+			io.MultiWriter(&writer, digest),
 			io.LimitReader(source, readLimit),
 		)
 		closeSourceErr := source.Close()
 		if copyErr != nil {
-			_ = output.Close()
-			return ExtractResult{}, fmt.Errorf("%w: read %s: %v", ErrArchiveCorrupt, entry.archivePath, copyErr)
+			primary := archiveInputError("read", entry.archivePath, copyErr)
+			if writer.err != nil {
+				primary = stagingOutputError("write staging file", writer.err)
+			}
+			return ExtractResult{}, joinStagingCloseError(primary, output)
 		}
 		if closeSourceErr != nil {
-			_ = output.Close()
-			return ExtractResult{}, fmt.Errorf("%w: close %s: %v", ErrArchiveCorrupt, entry.archivePath, closeSourceErr)
+			return ExtractResult{}, joinStagingCloseError(
+				archiveInputError("close", entry.archivePath, closeSourceErr),
+				output,
+			)
 		}
 		if copied > remaining {
-			_ = output.Close()
-			return ExtractResult{}, ErrExpandedTooLarge
+			return ExtractResult{}, joinStagingCloseError(ErrExpandedTooLarge, output)
 		}
 		if uint64(copied) != entry.file.UncompressedSize64 {
-			_ = output.Close()
-			return ExtractResult{}, fmt.Errorf(
+			return ExtractResult{}, joinStagingCloseError(fmt.Errorf(
 				"%w: %s expanded to %d bytes, header declares %d",
 				ErrArchiveCorrupt,
 				entry.archivePath,
 				copied,
 				entry.file.UncompressedSize64,
+			), output)
+		}
+		if err := syncFileWithFence(fence, output.osFile(), "staging payload"); err != nil {
+			return ExtractResult{}, joinStagingCloseError(
+				stagingOutputError("sync staging file", err),
+				output,
 			)
 		}
-		if err := syncFileWithFence(fence, output, "staging payload"); err != nil {
-			_ = output.Close()
-			return ExtractResult{}, err
-		}
 		if err := output.Close(); err != nil {
-			return ExtractResult{}, fmt.Errorf("close staging file: %w", err)
+			return ExtractResult{}, stagingOutputError("close staging file", err)
 		}
 		written += copied
 		result.Files++
@@ -243,7 +265,7 @@ func (extractor Extractor) extractPlan(
 		})
 	}
 	if err := syncStagingTree(fence, destinationRoot); err != nil {
-		return ExtractResult{}, err
+		return ExtractResult{}, stagingOutputError("sync staging tree", err)
 	}

 	result.Bytes = written
@@ -252,6 +274,40 @@ func (extractor Extractor) extractPlan(
 	return result, nil
 }

+type stagingWriter struct {
+	writer io.Writer
+	err    error
+}
+
+func (writer *stagingWriter) Write(data []byte) (int, error) {
+	written, err := writer.writer.Write(data)
+	if err != nil {
+		writer.err = err
+	} else if written != len(data) {
+		writer.err = io.ErrShortWrite
+	}
+	return written, err
+}
+
+func archiveInputError(operation, path string, cause error) error {
+	return fmt.Errorf("%w: %s %s: %w", ErrArchiveCorrupt, operation, path, cause)
+}
+
+func stagingOutputError(operation string, cause error) error {
+	return fmt.Errorf("%w: %s: %w", ErrStagingOutput, operation, cause)
+}
+
+func stagingCleanupError(cause error) error {
+	return fmt.Errorf("%w: remove staging: %w", ErrStagingCleanup, cause)
+}
+
+func joinStagingCloseError(primary error, output stagingOutputFile) error {
+	if closeErr := output.Close(); closeErr != nil {
+		return errors.Join(primary, stagingOutputError("close staging file", closeErr))
+	}
+	return primary
+}
+
 func planOutputPaths(
 	destination string,
 	entrypoint string,