pkg-proxy/internal/handler/metadata_stored.go

package handler

import (
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"hash"
	"io"
	"net/http"
	"time"

	"github.com/git-pkgs/proxy/internal/database"
	"github.com/git-pkgs/proxy/internal/metrics"
)

// A metadata document can be rewritten from storage instead of from memory:
// storedMetadataDocument streams it from upstream into storage, and the
// rewrite reads it back as often as it needs. Nothing holds the whole
// document, so its size is not limited by metadata_max_size or by memory.

// errMetadataNotStored is returned when metadata is not cached in storage, so
// a document has to be read into memory.
var errMetadataNotStored = errors.New("metadata is not cached in storage")

// errStoredMetadataChanged is returned when a stored document no longer has
// the digest its cache row records: a fetch replaced it while it was read.
var errStoredMetadataChanged = errors.New("stored metadata changed while it was read")

// storedMetadataDocument returns the cache row of a metadata document stored
// uncompressed with its digest: the stored copy while it is within its TTL,
// or else the document fetched from upstream and stored, or the stored copy
// when upstream fails. The document is streamed from upstream into storage
// without being held in memory, so a fetch takes no metadata_max_size limit.
// Concurrent calls for one document share a fetch, and instances sharing the
// database take turns as for other metadata.
//
// validate, when not nil, checks a fetched document before it replaces the
// stored one; a document it rejects is not stored, and the stored copy is
// served as when upstream fails. It does not check a stored copy within its
// TTL.
//
// It returns errMetadataNotStored when the proxy doesn't cache metadata or
// storing the document failed, and the caller reads it into memory instead.
func (p *Proxy) storedMetadataDocument(ctx context.Context, ecosystem, cacheKey, upstreamURL, accept string, validate func(io.Reader) error) (*database.MetadataCacheEntry, error) {
	res := p.storedMetadata(ctx, ecosystem, cacheKey, upstreamURL, accept, validate, "")
	return res.entry, res.err
}

// storedMetadata implements storedMetadataDocument, and reports in the
// result whether the row was read from cache or served stale. A row with
// the digest gone, one whose stored copy was missing or out of date when
// read, is not served or revalidated, within its TTL or not, and the
// document is fetched again.
func (p *Proxy) storedMetadata(ctx context.Context, ecosystem, cacheKey, upstreamURL, accept string, validate func(io.Reader) error, gone string) metadataResult {
	if !p.CacheMetadata || p.DB == nil || p.Storage == nil || p.MetadataTTL <= 0 {
		return metadataResult{err: errMetadataNotStored}
	}
	if containsPathTraversal(cacheKey) {
		return metadataResult{err: fmt.Errorf("invalid cache key: %q", cacheKey)}
	}
	if entry, fresh := p.storedMetadataEntryState(ctx, ecosystem, cacheKey, gone); fresh {
		metrics.RecordMetadataCacheHit(ecosystem, int(entry.Size.Int64))
		p.recordMetadataRead(entry)
		return metadataResult{entry: entry, fromCache: true}
	}
	p.recordMetadataCacheMiss(ecosystem)

	key := metadataCoalesceKey(ecosystem, cacheKey, upstreamURL, accept, "", false) + "\x00stored"
	res, _ := p.coalesceMetadata(ctx, key, func(fetchCtx context.Context) metadataResult {
		entry, fresh := p.storedMetadataEntryState(fetchCtx, ecosystem, cacheKey, gone)
		if fresh {
			return metadataResult{entry: entry}
		}
		release, outcome, err := p.awaitFetchLease(fetchCtx, metrics.CacheKindMetadata, key, func() bool {
			entry, fresh = p.storedMetadataEntryState(fetchCtx, ecosystem, cacheKey, gone)
			return fresh
		})
		if err != nil {
			return metadataResult{err: err}
		}
		switch outcome {
		case leaseStored:
			return metadataResult{entry: entry}
		case leaseFailed:
			// Serve the stored copy, as the holder did, rather than trying
			// the failing upstream again.
			p.reloadUpstreamHolds()
			if usableStoredMetadata(entry) {
				return p.staleStoredMetadata(ecosystem, cacheKey, entry, errFetchFailedElsewhere)
			}
		case leaseFetch:
		}
		res := metadataResult{err: errSharedFetchAbandoned}
		defer func() { release(res.err != nil || res.stale) }()
		res = p.fetchStoredMetadata(fetchCtx, ecosystem, cacheKey, upstreamURL, accept, validate, entry)
		return res
	})
	return res
}

// storedMetadataEntryState reads the cache row of a document and reports
// whether it is within its TTL and usable by storedMetadataDocument. A row
// with the digest gone is returned as nil.
func (p *Proxy) storedMetadataEntryState(ctx context.Context, ecosystem, cacheKey, gone string) (*database.MetadataCacheEntry, bool) {
	entry, _ := p.DB.GetMetadataCache(ecosystem, cacheKey)
	entry = p.migrateLegacyMetadataEntry(ctx, ecosystem, cacheKey, entry)
	if gone != "" && entry != nil && entry.ContentDigest.String == gone {
		return nil, false
	}
	fresh := usableStoredMetadata(entry) && entry.FetchedAt.Valid && time.Since(entry.FetchedAt.Time) < p.MetadataTTL
	return entry, fresh
}

// usableStoredMetadata reports whether entry describes a document stored
// uncompressed with its digest, which a rewrite can read back and check.
func usableStoredMetadata(entry *database.MetadataCacheEntry) bool {
	return entry != nil && entry.ContentDigest.Valid && entry.ContentEncoding.String == ""
}

// fetchStoredMetadata fetches a document from upstream into storage,
// revalidating entry when there is one, and falls back to entry when
// upstream fails or validate rejects what it sent. A failure to store is
// returned as it is, for the caller to read the document into memory.
func (p *Proxy) fetchStoredMetadata(ctx context.Context, ecosystem, cacheKey, upstreamURL, accept string, validate func(io.Reader) error, entry *database.MetadataCacheEntry) metadataResult {
	revalidate := entry
	if !usableStoredMetadata(entry) {
		revalidate = nil
	}
	stored, err := p.storeUpstreamMetadata(ctx, ecosystem, cacheKey, upstreamURL, accept, validate, revalidate)
	if errors.Is(err, errStale304) {
		stored, err = p.storeUpstreamMetadata(ctx, ecosystem, cacheKey, upstreamURL, accept, validate, nil)
	}
	if err == nil {
		return metadataResult{entry: stored}
	}
	// As in fetchMetadataFromUpstream, any upstream failure, a 404 included,
	// falls back to the stored copy.
	if !usableStoredMetadata(entry) || errors.Is(err, errMetadataNotStored) {
		return metadataResult{err: err}
	}
	return p.staleStoredMetadata(ecosystem, cacheKey, entry, err)
}

// staleStoredMetadata falls back to the stored document entry describes after
// the upstream fetch failed with err.
func (p *Proxy) staleStoredMetadata(ecosystem, cacheKey string, entry *database.MetadataCacheEntry, err error) metadataResult {
	p.Logger.Warn("upstream metadata fetch failed, serving cached copy",
		"ecosystem", ecosystem, "key", cacheKey, "error", err)
	entry = p.currentMetadataEntry(ecosystem, cacheKey, entry)
	if !usableStoredMetadata(entry) {
		return metadataResult{err: fmt.Errorf("upstream failed and no usable cached metadata: %w", err)}
	}
	return metadataResult{entry: entry, stale: true, upstreamErr: err}
}

// storeUpstreamMetadata fetches a document and streams it into storage, and
// returns its new cache row. On a 304 for entry it only moves the row's
// fetched_at on; errStale304 means the stored copy is gone or the row was
// replaced meanwhile, and the caller should fetch without validators.
//
// With validate, the document is staged under a path of its own and checked
// there first, so one validate rejects never replaces the stored copy.
func (p *Proxy) storeUpstreamMetadata(ctx context.Context, ecosystem, cacheKey, upstreamURL, accept string, validate func(io.Reader) error, entry *database.MetadataCacheEntry) (*database.MetadataCacheEntry, error) {
	resp, err := p.requestUpstreamMetadata(ctx, upstreamURL, entry, accept, "")
	if err != nil {
		return nil, err
	}
	defer func() { _ = resp.Body.Close() }()

	if resp.StatusCode == http.StatusNotModified && entry != nil {
		if ok, err := p.Storage.Exists(ctx, entry.StoragePath); err != nil || !ok {
			return nil, errStale304
		}
		refreshed, err := p.DB.RefreshMetadataCache(ecosystem, cacheKey, entry.ContentDigest.String, time.Now())
		if err != nil || !refreshed {
			return nil, errStale304
		}
		return p.currentMetadataEntry(ecosystem, cacheKey, entry), nil
	}
	if err := upstreamMetadataStatus(resp); err != nil {
		return nil, err
	}

	meta := &upstreamMetadata{
		contentType: resp.Header.Get(headerContentType),
		etag:        resp.Header.Get(headerETag),
	}
	if meta.contentType == "" {
		meta.contentType = contentTypeJSON
	}
	if lm := resp.Header.Get(headerLastModified); lm != "" {
		meta.lastModified, _ = http.ParseTime(lm)
	}
	limit := p.MetadataStreamMaxSize
	if limit <= 0 {
		limit = defaultMetadataStreamMaxSize
	}
	// Go decompresses the body, since no Accept-Encoding was set, so it is
	// stored uncompressed.
	var body io.Reader = &limitedBody{r: upstreamBodyReader{resp.Body}, left: limit}
	if validate != nil {
		staged, err := p.stageMetadata(ctx, ecosystem, cacheKey, body, validate)
		if err != nil {
			return nil, err
		}
		defer func() { _ = staged.Close() }()
		body = staged
	}
	if _, err := p.storeMetadataBlob(ctx, ecosystem, cacheKey, metadataStoragePath(ecosystem, cacheKey), body, meta); err != nil {
		return nil, notStored("", err)
	}
	stored, err := p.DB.GetMetadataCache(ecosystem, cacheKey)
	if err != nil || !usableStoredMetadata(stored) {
		return nil, fmt.Errorf("reading the stored metadata row: %w", errors.Join(err, errMetadataNotStored))
	}
	return stored, nil
}

// stageMetadata stores body under a staging path of its own, checks it with
// validate, and returns it opened for reading, to be stored in place of the
// cached copy. The staged copy is deleted when the reader is closed. A
// failure to store it is returned as errMetadataNotStored.
func (p *Proxy) stageMetadata(ctx context.Context, ecosystem, cacheKey string, body io.Reader, validate func(io.Reader) error) (io.ReadCloser, error) {
	path := metadataStoragePath(ecosystem, cacheKey) + ".staged-" + rand.Text()
	discard := func() {
		if err := p.Storage.Delete(context.WithoutCancel(ctx), path); err != nil {
			p.Logger.Warn("failed to delete staged metadata", "ecosystem", ecosystem, "key", cacheKey, "error", err)
		}
	}
	if _, _, err := p.Storage.Store(ctx, path, body); err != nil {
		discard()
		return nil, notStored("staging: ", err)
	}
	if err := readStored(func() (io.ReadCloser, error) { return p.Storage.Open(ctx, path) }, validate); err != nil {
		discard()
		return nil, err
	}
	staged, err := p.Storage.Open(ctx, path)
	if err != nil {
		discard()
		return nil, fmt.Errorf("%w: reading staged copy: %w", errMetadataNotStored, err)
	}
	return &stagedReader{ReadCloser: staged, discard: discard}, nil
}

// limitedBody fails with ErrMetadataTooLarge, as an upstream failure, once
// r has sent more than left bytes.
type limitedBody struct {
	r    io.Reader
	left int64
}

func (l *limitedBody) Read(b []byte) (int, error) {
	n, err := l.r.Read(b)
	if l.left -= int64(n); l.left < 0 {
		return 0, &upstreamBodyError{ErrMetadataTooLarge}
	}
	return n, err
}

// notStored returns a failure to store a document as errMetadataNotStored,
// unless reading it from upstream failed.
func notStored(what string, err error) error {
	if body := (*upstreamBodyError)(nil); errors.As(err, &body) {
		return body
	}
	return fmt.Errorf("%w: %s%w", errMetadataNotStored, what, err)
}

// stagedReader deletes the staged copy it reads when closed.
type stagedReader struct {
	io.ReadCloser
	discard func()
}

func (s *stagedReader) Close() error {
	err := s.ReadCloser.Close()
	s.discard()
	return err
}

// openStoredMetadata opens the document entry describes. Reading it to the
// end fails with errStoredMetadataChanged when its bytes don't have the
// digest entry records.
func (p *Proxy) openStoredMetadata(ctx context.Context, entry *database.MetadataCacheEntry) (io.ReadCloser, error) {
	r, err := p.Storage.Open(ctx, entry.StoragePath)
	if err != nil {
		return nil, err
	}
	return &digestReader{r: r, h: sha256.New(), want: entry.ContentDigest.String}, nil
}

// digestReader checks at the end of its reader that what it read has the
// digest want, in the form the metadata cache records it.
type digestReader struct {
	r    io.ReadCloser
	h    hash.Hash
	want string
}

func (d *digestReader) Read(b []byte) (int, error) {
	n, err := d.r.Read(b)
	d.h.Write(b[:n])
	if errors.Is(err, io.EOF) && "sha256:"+hex.EncodeToString(d.h.Sum(nil)) != d.want {
		return n, errStoredMetadataChanged
	}
	return n, err
}

func (d *digestReader) Close() error { return d.r.Close() }