diff --git a/internal/processor/blobs.go b/internal/processor/blobs.go index 5fb8f1785..dfbf19bfc 100644 --- a/internal/processor/blobs.go +++ b/internal/processor/blobs.go @@ -9,6 +9,7 @@ import ( "errors" "fmt" "io" + "math" "net/http" "strconv" "time" @@ -16,6 +17,7 @@ import ( "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 } diff --git a/internal/processor/manifests.go b/internal/processor/manifests.go index 536ef07f0..d3566b3aa 100644 --- a/internal/processor/manifests.go +++ b/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 } diff --git a/internal/tasks/manifests.go b/internal/tasks/manifests.go index 804578af7..d12ec767a 100644 --- a/internal/tasks/manifests.go +++ b/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,