mirror of
https://github.com/git-pkgs/proxy.git
synced 2026-11-09 15:49:16 -05:00
6c9505f322A streamed cache hit whose bytes don't match the row's digest now records an integrity failure and deletes the row, queueing its blob for deletion, as a corrupt cached artifact is handled. The next request fetches the document from upstream instead of aborting until the TTL runs out.
247 lines
8.4 KiB
Go
247 lines
8.4 KiB
Go
package database
|
|
|
|
import (
|
|
"database/sql"
|
|
"slices"
|
|
"time"
|
|
)
|
|
|
|
// Kinds of CacheEntry.
|
|
const (
|
|
CacheEntryArtifact = "artifact"
|
|
CacheEntryMetadata = "metadata"
|
|
)
|
|
|
|
// CacheEntry is a cached object size eviction can remove: an artifact fetched
|
|
// from upstream or a metadata document. For an artifact Key is its version
|
|
// PURL and Name its filename; for metadata Key is the ecosystem and Name the
|
|
// cache key. FetchedAt is only read for artifacts, UpdatedAt for metadata
|
|
// not backfilled yet.
|
|
type CacheEntry struct {
|
|
Kind string `db:"kind"`
|
|
Key string `db:"cache_key"`
|
|
Name string `db:"name"`
|
|
StoragePath sql.NullString `db:"storage_path"`
|
|
Size sql.NullInt64 `db:"size"`
|
|
LastAccessedAt sql.NullTime `db:"last_accessed_at"`
|
|
FetchedAt sql.NullTime `db:"fetched_at"`
|
|
UpdatedAt sql.NullTime `db:"updated_at"`
|
|
}
|
|
|
|
// GetEvictableCacheSize returns the bytes max_size limits: cached upstream
|
|
// artifacts and cached metadata documents.
|
|
func (db *DB) GetEvictableCacheSize() (int64, error) {
|
|
var total int64
|
|
err := db.Get(&total, `
|
|
SELECT (SELECT COALESCE(SUM(size), 0) FROM artifacts WHERE storage_path IS NOT NULL AND source <> 'local')
|
|
+ (SELECT COALESCE(SUM(size), 0) FROM metadata_cache)`)
|
|
return total, err
|
|
}
|
|
|
|
// GetLeastRecentlyUsedCacheEntries returns up to limit artifacts and metadata
|
|
// documents, least recently used first. An artifact counts as used when it
|
|
// was fetched until it is first served, so a fresh download does not go
|
|
// before every metadata document. Each kind is read in order on its own and
|
|
// the two are merged here, which keeps the queries simple for both dialects.
|
|
func (db *DB) GetLeastRecentlyUsedCacheEntries(limit int) ([]CacheEntry, error) {
|
|
var artifacts, metadata []CacheEntry
|
|
if err := db.Select(&artifacts, db.Rebind(`
|
|
SELECT 'artifact' AS kind, version_purl AS cache_key, filename AS name,
|
|
storage_path, size, last_accessed_at, fetched_at
|
|
FROM artifacts
|
|
WHERE storage_path IS NOT NULL AND source <> 'local'
|
|
ORDER BY COALESCE(last_accessed_at, fetched_at) ASC NULLS FIRST
|
|
LIMIT ?
|
|
`), limit); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := db.Select(&metadata, db.Rebind(`
|
|
SELECT 'metadata' AS kind, ecosystem AS cache_key, name,
|
|
storage_path, size, last_accessed_at, updated_at
|
|
FROM metadata_cache
|
|
ORDER BY COALESCE(last_accessed_at, updated_at) ASC NULLS FIRST
|
|
LIMIT ?
|
|
`), limit); err != nil {
|
|
return nil, err
|
|
}
|
|
// Rows not backfilled yet fall back to updated_at, which Postgres keeps
|
|
// in local time, so the order is fixed up once it is read as UTC.
|
|
for i := range metadata {
|
|
metadata[i].UpdatedAt = db.writtenAtUTC(metadata[i].UpdatedAt)
|
|
}
|
|
slices.SortStableFunc(metadata, func(a, b CacheEntry) int {
|
|
switch {
|
|
case a.usedAfter(b):
|
|
return 1
|
|
case b.usedAfter(a):
|
|
return -1
|
|
}
|
|
return 0
|
|
})
|
|
|
|
entries := make([]CacheEntry, 0, min(limit, len(artifacts)+len(metadata)))
|
|
for len(entries) < limit && (len(artifacts) > 0 || len(metadata) > 0) {
|
|
if len(metadata) == 0 || (len(artifacts) > 0 && !artifacts[0].usedAfter(metadata[0])) {
|
|
entries, artifacts = append(entries, artifacts[0]), artifacts[1:]
|
|
} else {
|
|
entries, metadata = append(entries, metadata[0]), metadata[1:]
|
|
}
|
|
}
|
|
return entries, nil
|
|
}
|
|
|
|
// lastUsed is when e was last served, or for an artifact not served since,
|
|
// when it was fetched. It is invalid when neither is known.
|
|
func (e CacheEntry) lastUsed() sql.NullTime {
|
|
if e.LastAccessedAt.Valid {
|
|
return e.LastAccessedAt
|
|
}
|
|
if e.FetchedAt.Valid {
|
|
return e.FetchedAt
|
|
}
|
|
return e.UpdatedAt
|
|
}
|
|
|
|
// usedAfter reports whether e was used later than other. Entries with no
|
|
// time known count as the oldest, as NULLS FIRST orders them.
|
|
func (e CacheEntry) usedAfter(other CacheEntry) bool {
|
|
a, b := e.lastUsed(), other.lastUsed()
|
|
if !a.Valid {
|
|
return false
|
|
}
|
|
return !b.Valid || a.Time.After(b.Time)
|
|
}
|
|
|
|
// EvictMetadataCache deletes a metadata cache row if it still points at
|
|
// storagePath and has not been used since lastAccessed, queueing the path in
|
|
// the same transaction so a crash before the blob is deleted cannot orphan it.
|
|
func (db *DB) EvictMetadataCache(ecosystem, name, storagePath string, lastAccessed sql.NullTime) (bool, error) {
|
|
query := `
|
|
DELETE FROM metadata_cache
|
|
WHERE ecosystem = ? AND name = ? AND storage_path = ? AND last_accessed_at IS NULL`
|
|
args := []any{ecosystem, name, storagePath}
|
|
if lastAccessed.Valid {
|
|
query = `
|
|
DELETE FROM metadata_cache
|
|
WHERE ecosystem = ? AND name = ? AND storage_path = ?
|
|
AND (last_accessed_at IS NULL OR last_accessed_at <= ?)`
|
|
args = append(args, lastAccessed.Time)
|
|
}
|
|
tx, err := db.Beginx()
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
res, err := tx.Exec(db.Rebind(query), args...)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if n, err := res.RowsAffected(); err != nil || n == 0 {
|
|
return false, err
|
|
}
|
|
if _, err := tx.Exec(db.Rebind(queuePendingDeleteQuery), storagePath, time.Now().UTC()); err != nil {
|
|
return false, err
|
|
}
|
|
return true, tx.Commit()
|
|
}
|
|
|
|
// DiscardMetadataCache deletes a metadata cache row whose stored document
|
|
// failed its integrity check, if it still points at storagePath with digest,
|
|
// and queues the path for deletion in the same transaction. A row a newer
|
|
// fetch has rewritten is left alone.
|
|
func (db *DB) DiscardMetadataCache(ecosystem, name, storagePath, digest string) error {
|
|
tx, err := db.Beginx()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
res, err := tx.Exec(db.Rebind(`
|
|
DELETE FROM metadata_cache
|
|
WHERE ecosystem = ? AND name = ? AND storage_path = ? AND content_digest = ?`),
|
|
ecosystem, name, storagePath, digest)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if n, err := res.RowsAffected(); err != nil || n == 0 {
|
|
return err
|
|
}
|
|
if _, err := tx.Exec(db.Rebind(queuePendingDeleteQuery), storagePath, time.Now().UTC()); err != nil {
|
|
return err
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
// RecordMetadataHit marks a cached metadata document as served now. Like
|
|
// RecordArtifactHit it is buffered when BatchHits is on.
|
|
func (db *DB) RecordMetadataHit(ecosystem, name string) error {
|
|
now := time.Now()
|
|
if db.hits != nil {
|
|
db.hits.addMetadata(metadataHitKey{ecosystem, name}, now)
|
|
return nil
|
|
}
|
|
_, err := db.Exec(db.Rebind(recordMetadataHitQuery), now.UTC(), now.UTC(), ecosystem, name)
|
|
return err
|
|
}
|
|
|
|
// recordMetadataHitQuery takes the access time twice, then the ecosystem and
|
|
// key. The time only moves forward, as for artifacts.
|
|
const recordMetadataHitQuery = `
|
|
UPDATE metadata_cache
|
|
SET last_accessed_at = CASE WHEN last_accessed_at IS NULL OR last_accessed_at < ? THEN ? ELSE last_accessed_at END
|
|
WHERE ecosystem = ? AND name = ?`
|
|
|
|
// writtenAtUTC returns a metadata_cache updated_at or created_at as the
|
|
// instant it was written. Both were written in local time, and a Postgres
|
|
// TIMESTAMP keeps only the wall clock, which lib/pq reads back as UTC.
|
|
func (db *DB) writtenAtUTC(t sql.NullTime) sql.NullTime {
|
|
if t.Valid && db.dialect == DialectPostgres {
|
|
w := t.Time
|
|
t.Time = time.Date(w.Year(), w.Month(), w.Day(), w.Hour(), w.Minute(), w.Second(), w.Nanosecond(), time.Local)
|
|
}
|
|
if t.Valid {
|
|
t.Time = t.Time.UTC()
|
|
}
|
|
return t
|
|
}
|
|
|
|
// BackfillMetadataLastAccessed sets last_accessed_at from the write time on
|
|
// up to limit rows that have none, in one transaction, and reports how many it
|
|
// set. Rows served in the meantime keep the time of that hit.
|
|
func (db *DB) BackfillMetadataLastAccessed(limit int) (int, error) {
|
|
var rows []struct {
|
|
ID int64 `db:"id"`
|
|
UpdatedAt sql.NullTime `db:"updated_at"`
|
|
CreatedAt sql.NullTime `db:"created_at"`
|
|
}
|
|
query := db.Rebind(`SELECT id, updated_at, created_at FROM metadata_cache WHERE last_accessed_at IS NULL LIMIT ?`)
|
|
if err := db.Select(&rows, query, limit); err != nil {
|
|
return 0, err
|
|
}
|
|
if len(rows) == 0 {
|
|
return 0, nil
|
|
}
|
|
tx, err := db.Beginx()
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
stmt, err := tx.Preparex(db.Rebind(`UPDATE metadata_cache SET last_accessed_at = ? WHERE id = ? AND last_accessed_at IS NULL`))
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
defer func() { _ = stmt.Close() }()
|
|
for _, r := range rows {
|
|
at := db.writtenAtUTC(r.UpdatedAt)
|
|
if !at.Valid {
|
|
at = db.writtenAtUTC(r.CreatedAt)
|
|
}
|
|
if !at.Valid {
|
|
// Nothing to go by; counting it as used now at least ends the loop.
|
|
at = sql.NullTime{Time: time.Now().UTC(), Valid: true}
|
|
}
|
|
if _, err := stmt.Exec(at.Time, r.ID); err != nil {
|
|
return 0, err
|
|
}
|
|
}
|
|
return len(rows), tx.Commit()
|
|
}
|