feat: implement gzip stream calculation

pkg/apk/implementation.go

@@ -449,6 +449,64 @@ func (a *APK) ResolveWorld(ctx context.Context) (toInstall []*repository.Reposit
 	return
 }
 
+func (a *APK) ResolveAndCalculateWorld(ctx context.Context) ([]*APKResolved, error) {
+
+	a.logger.Infof("synchronizing with desired apk world")
+
+	ctx, span := otel.Tracer("go-apk").Start(ctx, "CalculateWorld")
+	defer span.End()
+
+	allpkgs, _, err := a.ResolveWorld(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("error getting package dependencies: %w", err)
+	}
+
+	// TODO: Consider making this configurable option.
+	jobs := runtime.GOMAXPROCS(0)
+
+	g, gctx := errgroup.WithContext(ctx)
+	g.SetLimit(jobs + 1)
+
+	resolved := make([]*APKResolved, len(allpkgs))
+
+	// A slice of pseudo-promises that get closed when expanded[i] is ready.
+	done := make([]chan struct{}, len(allpkgs))
+	for i := range allpkgs {
+		done[i] = make(chan struct{})
+	}
+
+	// Meanwhile, concurrently fetch and expand all our APKs.
+	// We signal they are ready to be installed by closing done[i].
+	for i, pkg := range allpkgs {
+		i, pkg := i, pkg
+
+		g.Go(func() error {
+			r, err := a.FetchPackage(gctx, pkg)
+			if err != nil {
+				return fmt.Errorf("fetching %s: %w", pkg.Name, err)
+			}
+			res, err := ResolveApk(gctx, r)
+			if err != nil {
+				fmt.Println("ResolveApk resolving %s: %w", pkg.Name, err)
+				return fmt.Errorf("resolving %s: %w", pkg.Name, err)
+			}
+
+			res.Package = pkg
+			resolved[i] = res
+
+			close(done[i])
+
+			return nil
+		})
+	}
+
+	if err := g.Wait(); err != nil {
+		return nil, fmt.Errorf("installing packages: %w", err)
+	}
+
+	return resolved, nil
+}
+
 // FixateWorld force apk's resolver to re-resolve the requested dependencies in /etc/apk/world.
 func (a *APK) FixateWorld(ctx context.Context, sourceDateEpoch *time.Time) error {
 	/*

pkg/apk/installed.go

@@ -191,44 +191,16 @@ func (a *APK) controlValue(controlTarGz io.Reader, want string) ([]string, error
 	defer gz.Close()
 	tr := tar.NewReader(gz)
 
-	values := []string{}
-	for {
-		header, err := tr.Next()
-		if errors.Is(err, io.EOF) {
-			break
-		}
-		if err != nil {
-			return nil, err
-		}
-
-		// ignore .PKGINFO as it is not a script
-		if header.Name != ".PKGINFO" {
-			continue
-		}
-
-		b, err := io.ReadAll(tr)
-		if err != nil {
-			return nil, fmt.Errorf("unable to read .PKGINFO from control tar.gz file: %w", err)
-		}
-		lines := strings.Split(string(b), "\n")
-		for _, line := range lines {
-			parts := strings.Split(line, "=")
-			if len(parts) != 2 {
-				continue
-			}
-			key := strings.TrimSpace(parts[0])
-			if key != want {
-				continue
-			}
-
-			value := strings.TrimSpace(parts[1])
-			values = append(values, value)
+	mapping, err := controlValue(tr, want)
+	if err != nil {
+		return nil, err
+	} else {
+		values, ok := mapping[want]
+		if !ok {
+			return []string{}, nil
 		}
-
-		break
+		return values, nil
 	}
-
-	return values, nil
 }
 
 // updateTriggers insert the triggers into the triggers file

pkg/apk/resolveapk.go

@@ -0,0 +1,151 @@
+// THIS FILE IS AN EXACT DUPLICATE OF STUFF IN ALPINE-GO!
+// Unfortunately, all of that is package-private rather than public,
+// so we are duplicating it here. As soon as we can upstream this entire impl,
+// this duplicate file goes away!
+
+//nolint:all
+package apk
+
+import (
+	"archive/tar"
+	"context"
+	"crypto/sha1"
+	"encoding/hex"
+	"fmt"
+	"hash"
+	"io"
+	"strconv"
+	"strings"
+
+	"github.com/klauspost/compress/gzip"
+	"gitlab.alpinelinux.org/alpine/go/repository"
+
+	"go.opentelemetry.io/otel"
+)
+
+type APKResolved struct {
+	Package *repository.RepositoryPackage
+
+	SignatureSize int
+	SignatureHash []byte
+
+	ControlSize int
+	ControlHash []byte
+
+	DataSize int
+	DataHash []byte
+}
+
+type countingWriter struct {
+	bytesWritten int
+}
+
+func (r *countingWriter) Write(p []byte) (n int, err error) {
+	r.bytesWritten += len(p)
+	return n, nil
+}
+
+func ResolveApk(ctx context.Context, source io.Reader) (*APKResolved, error) {
+	ctx, span := otel.Tracer("go-apk").Start(ctx, "ResolveApk")
+	defer span.End()
+
+	gzipStreamSizes := make([]int, 3)
+	hashes := make([][]byte, 3)
+	maxStreams := 2
+	streamId := 0
+	controlIdx := 0
+	signed := false
+
+	var gzi *gzip.Reader
+	cr := &countingWriter{}
+	tr := io.TeeReader(source, cr)
+	norar := newExpandApkReader(tr)
+
+	for {
+		var h hash.Hash = sha1.New() //nolint:gosec
+
+		hr := io.TeeReader(norar, h)
+
+		var err error
+
+		if gzi == nil {
+			gzi, err = gzip.NewReader(hr)
+		} else {
+			err = gzi.Reset(hr)
+		}
+
+		if err == io.EOF {
+			break
+		} else if err != nil {
+			return nil, fmt.Errorf("creating gzip reader: %w", err)
+		}
+		gzi.Multistream(false)
+
+		if streamId == 0 {
+			tr := tar.NewReader(gzi)
+			hdr, err := tr.Next()
+			if err != nil {
+				return nil, fmt.Errorf("ResolveApk error 1: %v", err)
+			}
+			if strings.HasPrefix(hdr.Name, ".SIGN.") {
+				maxStreams = 3
+				controlIdx = 1
+				signed = true
+			}
+		} else if controlIdx == streamId {
+			mapping, err := controlValue(gzi, "datahash", "size")
+			if err != nil {
+				return nil, fmt.Errorf("reading datahash and size from control: %w", err)
+			}
+
+			if sizes, ok := mapping["size"]; !ok {
+				return nil, fmt.Errorf("reading size from control: %w", err)
+			} else if len(sizes) != 1 {
+				return nil, fmt.Errorf("saw %d size values", len(sizes))
+			} else if size, err := strconv.Atoi(sizes[0]); err != nil {
+				return nil, fmt.Errorf("parsing size from  control: %w", err)
+			} else {
+				gzipStreamSizes[maxStreams-1] = size
+			}
+
+			if datahashes, ok := mapping["datahash"]; !ok {
+				return nil, fmt.Errorf("reading datahash from control: %w", err)
+			} else if len(datahashes) != 1 {
+				return nil, fmt.Errorf("saw %d datahash values", len(datahashes))
+			} else if hash, err := hex.DecodeString(datahashes[0]); err != nil {
+				return nil, fmt.Errorf("reading datahash from control: %w", err)
+			} else {
+				hashes[maxStreams-1] = hash
+			}
+		}
+
+		if streamId <= controlIdx {
+			if _, err := io.Copy(io.Discard, gzi); err != nil {
+				return nil, fmt.Errorf("ResolveApk error 2: %v", err)
+			}
+
+			hashes[streamId] = h.Sum(nil)
+			gzipStreamSizes[streamId] = cr.bytesWritten
+		} else {
+			gzi.Close()
+			break
+		}
+
+		streamId++
+	}
+
+	resolved := &APKResolved{
+		SignatureSize: 0,
+		ControlSize:   gzipStreamSizes[controlIdx],
+		ControlHash:   hashes[controlIdx],
+		DataSize:      gzipStreamSizes[controlIdx+1],
+		DataHash:      hashes[controlIdx+1],
+	}
+
+	if signed {
+		resolved.SignatureSize = gzipStreamSizes[0]
+		resolved.SignatureHash = hashes[0]
+	}
+
+	return resolved, nil
+}

pkg/apk/util.go

@@ -14,6 +14,15 @@
 
 package apk
 
+import (
+	"archive/tar"
+	"errors"
+	"fmt"
+	"io"
+	"slices"
+	"strings"
+)
+
 func uniqify[T comparable](s []T) []T {
 	seen := make(map[T]struct{}, len(s))
 	uniq := make([]T, 0, len(s))
@@ -28,3 +37,51 @@ func uniqify[T comparable](s []T) []T {
 
 	return uniq
 }
+
+func controlValue(controlTar io.Reader, want ...string) (map[string][]string, error) {
+	tr := tar.NewReader(controlTar)
+	mapping := map[string][]string{}
+	for {
+		header, err := tr.Next()
+		if errors.Is(err, io.EOF) {
+			break
+		}
+		if err != nil {
+			return nil, err
+		}
+
+		// ignore .PKGINFO as it is not a script
+		if header.Name != ".PKGINFO" {
+			continue
+		}
+
+		b, err := io.ReadAll(tr)
+		if err != nil {
+			return nil, fmt.Errorf("unable to read .PKGINFO from control tar.gz file: %w", err)
+		}
+		lines := strings.Split(string(b), "\n")
+		for _, line := range lines {
+			parts := strings.Split(line, "=")
+			if len(parts) != 2 {
+				continue
+			}
+			key := strings.TrimSpace(parts[0])
+			if !slices.Contains(want, key) {
+				continue
+			}
+
+			values, ok := mapping[key]
+			if !ok {
+				values = []string{}
+			}
+
+			value := strings.TrimSpace(parts[1])
+			values = append(values, value)
+
+			mapping[key] = values
+		}
+
+		break
+	}
+	return mapping, nil
+}