Add self healing to ValidateExistingBlob and validateManifest

internal/processor/blobs.go

@@ -9,13 +9,15 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math"
 	"net/http"
 	"strconv"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/sapcc/go-bits/logg"
 	"go.podman.io/image/v5/manifest"
+	"go.podman.io/image/v5/types"
 	"go.xyrillian.de/gg/gsql"
 	. "go.xyrillian.de/gg/option"
 
@@ -34,6 +36,28 @@ func (p *Processor) ValidateExistingBlob(ctx context.Context, account models.Red
 	}
 
 	readCloser, _, err := p.sd.ReadBlobForValidation(ctx, account, blob.StorageID)
+	// If we cannot find the blob and the account is a replication from somewhere else try to get it from there
+	if errors.Is(err, keppel.NotFoundInStorageError{}) && (account.ExternalPeerURL != "" || account.UpstreamPeerHostName != "") {
+		var blobSize int64
+		if blob.SizeBytes < math.MaxInt64 {
+			blobSize = int64(blob.SizeBytes)
+		} else {
+			return fmt.Errorf("blob %s has strange size", blob.Digest)
+		}
+		layerInfo := manifest.LayerInfo{
+			BlobInfo: types.BlobInfo{
+				Digest:    blob.Digest,
+				Size:      blobSize,
+				MediaType: blob.MediaType,
+			},
+		}
+		_, err = p.FindBlobOrInsertUnbackedBlob(ctx, layerInfo, account.Name)
+		if err == nil {
+			return nil
+		} else {
+			return fmt.Errorf("blob could not be found while validating and replication failed: %w", err)
+		}
+	}
 	if err != nil {
 		return err
 	}

internal/processor/manifests.go

@@ -177,8 +177,15 @@ func (p *Processor) ValidateAndStoreManifest(ctx context.Context, account models
 }
 
 // ValidateExistingManifest validates the given manifest that already exists in the DB.
-func (p *Processor) ValidateExistingManifest(ctx context.Context, account models.ReducedAccount, repo models.ReducedRepository, manifest *models.Manifest) error {
+func (p *Processor) ValidateExistingManifest(ctx context.Context, account models.ReducedAccount, repo models.ReducedRepository, manifest *models.Manifest, tagPolicies []keppel.TagPolicy, actx keppel.AuditContext) error {
 	manifestBytes, err := p.sd.ReadManifestForValidation(ctx, account, repo.Name, manifest.Digest)
+	// If we cannot find the manifest and the account is a replication from somewhere else try to get it from there
+	if errors.Is(err, keppel.NotFoundInStorageError{}) && (account.ExternalPeerURL != "" || account.UpstreamPeerHostName != "") {
+		manifest, manifestBytes, err = p.ReplicateManifest(ctx, account, repo, models.ManifestReference{Digest: manifest.Digest}, tagPolicies, actx)
+		if err != nil {
+			return fmt.Errorf("manifest could not be found while validating and replication failed: %w", err)
+		}
+	}
 	if err != nil {
 		return err
 	}

internal/tasks/manifests.go

@@ -80,18 +80,27 @@ func (j *Janitor) validateManifest(ctx context.Context, manifest models.Manifest
 	if err != nil {
 		return fmt.Errorf("cannot find repo %d for manifest %s: %w", manifest.RepositoryID, manifest.Digest, err)
 	}
-	account, err := keppel.FindReducedAccount(ctx, j.db, repo.AccountName)
+	account, err := keppel.FindAccount(ctx, j.db, repo.AccountName)
 	if err != nil {
 		return fmt.Errorf("cannot find account for manifest %s/%s: %w", repo.FullName(), manifest.Digest, err)
 	}
+	tagPolicies, err := keppel.ParseTagPolicies(account.TagPoliciesJSON)
+	if err != nil {
+		return err
+	}
 
 	// if the validation succeeds, these fields will be committed
 	nextValidationAt := j.timeNow().Add(j.addJitter(models.ManifestValidationInterval))
 	manifest.NextValidationAt = nextValidationAt
 	manifest.ValidationErrorMessage = ""
 
+	actx := keppel.AuditContext{
+		UserIdentity: janitorUserIdentity{TaskName: "manifest-validation"},
+		Request:      janitorDummyRequest,
+	}
+
 	// perform validation
-	err = j.processor().ValidateExistingManifest(ctx, account, repo.Reduced(), &manifest)
+	err = j.processor().ValidateExistingManifest(ctx, account.Reduced(), repo.Reduced(), &manifest, tagPolicies, actx)
 	if err != nil {
 		// on failure, log error message and schedule next validation sooner than usual
 		_, updateErr := j.db.Exec(validateManifestFinishQuery,