speed up snailmail apply

Signed-off-by: Polina Simonenko <rabarbrablad@gmail.com>

Author: rabarbra

internal/app/apk_verify.go

@@ -52,14 +52,12 @@ func VerifyAPKClient(ctx context.Context, repository, runner, image string) (bui
 	if !digestPinnedImage(image) {
 		return buildgraph.RepositoryManifest{}, 0, errors.New("a digest-pinned Alpine verification image is required")
 	}
-	installed := 0
-	for _, verification := range manifest.VerificationCases {
-		if err := verifyAPKCase(ctx, runner, image, snapshot, verification); err != nil {
-			return buildgraph.RepositoryManifest{}, 0, err
-		}
-		installed++
+	if err := verifyCases(ctx, manifest.VerificationCases, func(caseCtx context.Context, verification domain.VerificationCase) error {
+		return verifyAPKCase(caseCtx, runner, image, snapshot, verification)
+	}); err != nil {
+		return buildgraph.RepositoryManifest{}, 0, err
 	}
-	return manifest, installed, nil
+	return manifest, len(manifest.VerificationCases), nil
 }
 
 func verifyAPKCase(ctx context.Context, runner, image, snapshot string, verification domain.VerificationCase) error {

internal/app/deb_endpoint.go

@@ -94,10 +94,10 @@ func VerifyDebClientEndpointAccess(ctx context.Context, root string, access host
 	// only reachable with host networking. A public endpoint uses the runner's
 	// default network and stays isolated from the host.
 	hostNetwork := isLoopbackHost(endpoint.Hostname())
-	for _, verification := range verificationCases {
-		if err := verifyDebEndpointCase(ctx, runner, image, access.Endpoint, trust, hostNetwork, workspaceBytes, manifest, verification); err != nil {
-			return buildgraph.RepositoryManifest{}, 0, err
-		}
+	if err := verifyCases(ctx, verificationCases, func(caseCtx context.Context, verification domain.VerificationCase) error {
+		return verifyDebEndpointCase(caseCtx, runner, image, access.Endpoint, trust, hostNetwork, workspaceBytes, manifest, verification)
+	}); err != nil {
+		return buildgraph.RepositoryManifest{}, 0, err
 	}
 	return manifest, len(manifest.VerificationCases), nil
 }

internal/app/rpm_verify.go

@@ -55,14 +55,12 @@ func VerifyRPMClient(ctx context.Context, repository, runner, image string) (bui
 	if !digestPinnedImage(image) {
 		return buildgraph.RepositoryManifest{}, 0, errors.New("a digest-pinned RPM verification image is required")
 	}
-	installed := 0
-	for _, verification := range manifest.VerificationCases {
-		if err := verifyRPMCase(ctx, runner, image, snapshot, verification); err != nil {
-			return buildgraph.RepositoryManifest{}, 0, err
-		}
-		installed++
+	if err := verifyCases(ctx, manifest.VerificationCases, func(caseCtx context.Context, verification domain.VerificationCase) error {
+		return verifyRPMCase(caseCtx, runner, image, snapshot, verification)
+	}); err != nil {
+		return buildgraph.RepositoryManifest{}, 0, err
 	}
-	return manifest, installed, nil
+	return manifest, len(manifest.VerificationCases), nil
 }
 
 func verifyRPMCase(ctx context.Context, runner, image, snapshot string, verification domain.VerificationCase) error {

internal/app/verify_parallel.go

@@ -0,0 +1,67 @@
+package app
+
+import (
+	"context"
+	"sync"
+
+	"github.com/shellcell/snailmail/internal/domain"
+)
+
+// maxConcurrentVerifications bounds how many client containers run at once.
+//
+// Each case is one container installing one package version, and they are
+// independent: every one mounts the same repository read-only and writes
+// nothing a sibling reads. Running them one at a time made publication cost
+// grow with the number of versions a repository holds, which is the number that
+// only ever goes up. The bound is small on purpose — verifying a foreign
+// architecture runs under emulation, and oversubscribing a two-core runner with
+// emulated containers spends more time context-switching than it saves.
+const maxConcurrentVerifications = 4
+
+// verifyCases runs one verification per case and reports the earliest failing
+// case in the order the manifest lists them.
+//
+// It deliberately does not stop at the first failure. Cases acquire their slot
+// in whatever order the scheduler hands them out, so cancelling the rest would
+// leave an earlier case unrun — and a case that never ran cannot be known to
+// have failed. The publication would then blame whichever container happened to
+// finish first, and the same defect would be reported differently on every run.
+// A refused publication is the rare path and is already going to be repeated;
+// naming the same case every time is worth more there than finishing sooner.
+func verifyCases(ctx context.Context, cases []domain.VerificationCase, verify func(context.Context, domain.VerificationCase) error) error {
+	if len(cases) == 0 {
+		return nil
+	}
+	if len(cases) == 1 {
+		return verify(ctx, cases[0])
+	}
+	limit := maxConcurrentVerifications
+	if len(cases) < limit {
+		limit = len(cases)
+	}
+	slots := make(chan struct{}, limit)
+	failures := make([]error, len(cases))
+	var wait sync.WaitGroup
+	for index, verification := range cases {
+		wait.Add(1)
+		go func(index int, verification domain.VerificationCase) {
+			defer wait.Done()
+			slots <- struct{}{}
+			defer func() { <-slots }()
+			// A cancelled parent still stops the run; that cancellation is the
+			// caller giving up, not a verdict about any case.
+			if err := ctx.Err(); err != nil {
+				failures[index] = err
+				return
+			}
+			failures[index] = verify(ctx, verification)
+		}(index, verification)
+	}
+	wait.Wait()
+	for _, err := range failures {
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}

internal/app/verify_parallel_test.go

@@ -0,0 +1,109 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/shellcell/snailmail/internal/domain"
+)
+
+func cases(count int) []domain.VerificationCase {
+	made := make([]domain.VerificationCase, 0, count)
+	for index := range count {
+		made = append(made, domain.VerificationCase{Package: fmt.Sprintf("p%d", index), Version: "1"})
+	}
+	return made
+}
+
+func TestVerifyCasesRunsEveryCase(t *testing.T) {
+	var mutex sync.Mutex
+	seen := map[string]bool{}
+	if err := verifyCases(context.Background(), cases(9), func(_ context.Context, verification domain.VerificationCase) error {
+		mutex.Lock()
+		defer mutex.Unlock()
+		if seen[verification.Package] {
+			t.Errorf("case %s was verified twice", verification.Package)
+		}
+		seen[verification.Package] = true
+		return nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if len(seen) != 9 {
+		t.Fatalf("verified %d cases, want 9", len(seen))
+	}
+}
+
+// A failed publication must name the same case every time, or the same defect
+// reads as a different problem on each run.
+func TestVerifyCasesReportsTheFirstFailureInOrder(t *testing.T) {
+	for range 20 {
+		err := verifyCases(context.Background(), cases(8), func(_ context.Context, verification domain.VerificationCase) error {
+			if verification.Package == "p3" || verification.Package == "p6" {
+				return errors.New(verification.Package + " failed")
+			}
+			return nil
+		})
+		if err == nil || err.Error() != "p3 failed" {
+			t.Fatalf("got %v, want the earliest failing case p3", err)
+		}
+	}
+}
+
+// Every case runs even once one has failed. That is what makes the reported
+// case deterministic: a case skipped because a sibling failed first cannot be
+// compared against the ones that did run, and the earliest failure would depend
+// on which container won the race for a slot.
+func TestVerifyCasesRunsEveryCaseEvenAfterAFailure(t *testing.T) {
+	var started atomic.Int64
+	err := verifyCases(context.Background(), cases(32), func(_ context.Context, verification domain.VerificationCase) error {
+		started.Add(1)
+		if verification.Package == "p0" {
+			return errors.New("refused")
+		}
+		return nil
+	})
+	if err == nil || err.Error() != "refused" {
+		t.Fatalf("got %v, want the failing case reported", err)
+	}
+	if count := started.Load(); count != 32 {
+		t.Fatalf("ran %d of 32 cases; skipping any makes the reported failure depend on scheduling", count)
+	}
+}
+
+// Cancelling the parent must not read as every case having passed.
+func TestVerifyCasesHonoursACancelledContext(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if err := verifyCases(ctx, cases(4), func(context.Context, domain.VerificationCase) error {
+		return nil
+	}); !errors.Is(err, context.Canceled) {
+		t.Fatalf("got %v, want context.Canceled", err)
+	}
+}
+
+// The bound is what keeps an emulated architecture from oversubscribing a small
+// runner, so it has to actually hold.
+func TestVerifyCasesRespectsItsConcurrencyBound(t *testing.T) {
+	var running, peak atomic.Int64
+	if err := verifyCases(context.Background(), cases(32), func(context.Context, domain.VerificationCase) error {
+		current := running.Add(1)
+		defer running.Add(-1)
+		for {
+			was := peak.Load()
+			if current <= was || peak.CompareAndSwap(was, current) {
+				break
+			}
+		}
+		return nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if peak.Load() > maxConcurrentVerifications {
+		t.Fatalf("%d cases ran at once, above the bound of %d", peak.Load(), maxConcurrentVerifications)
+	}
+}

signer/apkrsa/apkrsa_test.go

@@ -67,7 +67,13 @@ func TestVerifyRejectsWhatItShould(t *testing.T) {
 		},
 		"altered signature": func() error {
 			altered := response
-			altered.Content = append(append([]byte(nil), response.Content[:len(response.Content)-1]...), 0x00)
+			// Flip a bit rather than write a fixed byte. Setting the last byte
+			// to zero leaves the signature untouched whenever it already ended
+			// in zero, which an RSA signature does about once in 256 runs — and
+			// an unaltered signature verifying is the test passing by accident,
+			// reported as a failure.
+			altered.Content = append([]byte(nil), response.Content...)
+			altered.Content[len(altered.Content)-1] ^= 0x01
 			return VerifyResponse(request, altered, publicDER(t, generated), generated.Identity.Fingerprint)
 		},
 		"another key": func() error {