mirror of
https://github.com/git-pkgs/proxy.git
synced 2026-11-09 07:39:19 -05:00
64162c77d8
A detached fetch waiting on another instance's lease ran the wait under the request context, so a client leaving cancelled every caller joined to it. The wait now runs past the client, bounded by the holder's fetch and the lease expiry. When the holder's fetch failed, each waiting instance took the lease in turn and repeated it, so against a hanging upstream the k-th instance waited about k timeouts. The holder now marks the lease failed. The instances waiting on it then serve their stale metadata copy if they have one, and otherwise fetch at the same time.
94 lines
3.5 KiB
Go
94 lines
3.5 KiB
Go
package database
|
|
|
|
import (
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
)
|
|
|
|
// Fetch leases let instances sharing a Postgres database fetch each cache
|
|
// miss from upstream once. The instance that fetches holds the lease for the
|
|
// miss's key and renews it until the result is stored; the others wait for
|
|
// the stored result. Expiry uses the database clock, so instances whose
|
|
// clocks disagree still agree on when a lease has run out. SQLite is
|
|
// single-instance, so these are only used on Postgres.
|
|
|
|
// AcquireFetchLease takes the lease on key for owner unless another owner
|
|
// holds one that has neither expired nor failed, and reports whether it did.
|
|
func (db *DB) AcquireFetchLease(key, owner string, ttl time.Duration) (bool, error) {
|
|
res, err := db.Exec(`
|
|
INSERT INTO fetch_leases (key, owner, expires_at)
|
|
VALUES ($1, $2, now() + $3 * interval '1 millisecond')
|
|
ON CONFLICT (key) DO UPDATE SET owner = excluded.owner, expires_at = excluded.expires_at, failed = FALSE
|
|
WHERE fetch_leases.expires_at < now() OR fetch_leases.failed`, key, owner, ttl.Milliseconds())
|
|
if err != nil {
|
|
return false, fmt.Errorf("acquiring fetch lease: %w", err)
|
|
}
|
|
n, err := res.RowsAffected()
|
|
if err != nil {
|
|
return false, fmt.Errorf("acquiring fetch lease: %w", err)
|
|
}
|
|
return n > 0, nil
|
|
}
|
|
|
|
// RenewFetchLease extends owner's lease on key by ttl from now, and reports
|
|
// whether owner still held it.
|
|
func (db *DB) RenewFetchLease(key, owner string, ttl time.Duration) (bool, error) {
|
|
res, err := db.Exec(`
|
|
UPDATE fetch_leases SET expires_at = now() + $3 * interval '1 millisecond'
|
|
WHERE key = $1 AND owner = $2`, key, owner, ttl.Milliseconds())
|
|
if err != nil {
|
|
return false, fmt.Errorf("renewing fetch lease: %w", err)
|
|
}
|
|
n, err := res.RowsAffected()
|
|
if err != nil {
|
|
return false, fmt.Errorf("renewing fetch lease: %w", err)
|
|
}
|
|
return n > 0, nil
|
|
}
|
|
|
|
// FetchLeaseHolder returns the owner of the lease on key, and whether its
|
|
// fetch failed. ok is false when no lease is held or it has expired.
|
|
func (db *DB) FetchLeaseHolder(key string) (owner string, failed, ok bool, err error) {
|
|
err = db.QueryRow(`
|
|
SELECT owner, failed FROM fetch_leases
|
|
WHERE key = $1 AND expires_at >= now()`, key).Scan(&owner, &failed)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return "", false, false, nil
|
|
}
|
|
if err != nil {
|
|
return "", false, false, fmt.Errorf("reading fetch lease: %w", err)
|
|
}
|
|
return owner, failed, true, nil
|
|
}
|
|
|
|
// FailFetchLease marks owner's lease on key as failed. Instances waiting on
|
|
// it stop waiting, and the next AcquireFetchLease takes it at once.
|
|
func (db *DB) FailFetchLease(key, owner string) error {
|
|
if _, err := db.Exec(`UPDATE fetch_leases SET failed = TRUE WHERE key = $1 AND owner = $2`, key, owner); err != nil {
|
|
return fmt.Errorf("failing fetch lease: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ReleaseFetchLease drops owner's lease on key.
|
|
func (db *DB) ReleaseFetchLease(key, owner string) error {
|
|
if _, err := db.Exec(`DELETE FROM fetch_leases WHERE key = $1 AND owner = $2`, key, owner); err != nil {
|
|
return fmt.Errorf("releasing fetch lease: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// DeleteExpiredFetchLeases removes leases that expired more than grace ago,
|
|
// left by instances that stopped while fetching.
|
|
func (db *DB) DeleteExpiredFetchLeases(grace time.Duration) (int64, error) {
|
|
if db.dialect != DialectPostgres {
|
|
return 0, nil
|
|
}
|
|
res, err := db.Exec(`DELETE FROM fetch_leases WHERE expires_at < now() - $1 * interval '1 millisecond'`, grace.Milliseconds())
|
|
if err != nil {
|
|
return 0, fmt.Errorf("deleting expired fetch leases: %w", err)
|
|
}
|
|
return res.RowsAffected()
|
|
}
|