Retry failed GitHub release scans with backoff (#133)
ci/woodpecker/tag/docker Pipeline was successful
ci/woodpecker/tag/docker Pipeline was successful
Releasing a GitHub sync lease always advanced `last_synced_at`, even after a failed scan (e.g. a rate-limit 403). A remote that failed once waited a full `mutable_ttl` before retrying, and a cold remote kept returning 503 until then. - record scan outcomes in one shared lease helper for github_rpm/deb/alpine - keep `last_synced_at` and the ETag on failure; retry from 60s with exponential backoff, capped at min(10m, ttl/4) - honour `Retry-After` / `X-RateLimit-Reset`, clamped to `mutable_ttl` - add `sync_failures` / `next_retry_at` columns (migration 0002) Reviewed-on: #133 Co-authored-by: unkin-agent <unkin-agent@unkin.net> Co-committed-by: unkin-agent <unkin-agent@unkin.net>
This commit was merged in pull request #133.
This commit is contained in:
@@ -2,11 +2,9 @@ package database
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
|
||||
"git.unkin.net/unkin/artifactapi/internal/provider"
|
||||
"git.unkin.net/unkin/artifactapi/pkg/models"
|
||||
)
|
||||
|
||||
@@ -30,42 +28,14 @@ func (db *DB) ListGitHubAlpineRemotes(ctx context.Context) ([]models.Remote, err
|
||||
return remotes, rows.Err()
|
||||
}
|
||||
|
||||
// ClaimGitHubAlpineSyncLease atomically claims the per-remote sync lease. It
|
||||
// succeeds only when the remote is due (never synced, or synced longer than
|
||||
// freshness ago) and no live lease is held by another replica. A zero freshness
|
||||
// (prime scans) ignores the recency gate. The returned etag is the stored
|
||||
// releases-list ETag, shared across replicas.
|
||||
// ClaimGitHubAlpineSyncLease atomically claims the per-remote github_alpine sync
|
||||
// lease. See claimSyncLease.
|
||||
func (db *DB) ClaimGitHubAlpineSyncLease(ctx context.Context, remoteName, owner string, freshness, lease time.Duration) (bool, string, error) {
|
||||
row := db.Pool.QueryRow(ctx, `
|
||||
INSERT INTO github_alpine_sync_state AS s (remote_name, sync_lease_owner, sync_lease_expires)
|
||||
VALUES ($1, $2, now() + make_interval(secs => $4))
|
||||
ON CONFLICT (remote_name) DO UPDATE
|
||||
SET sync_lease_owner = $2,
|
||||
sync_lease_expires = now() + make_interval(secs => $4)
|
||||
WHERE (s.last_synced_at IS NULL OR s.last_synced_at < now() - make_interval(secs => $3))
|
||||
AND (s.sync_lease_expires IS NULL OR s.sync_lease_expires < now())
|
||||
RETURNING s.etag
|
||||
`, remoteName, owner, freshness.Seconds(), lease.Seconds())
|
||||
|
||||
var etag string
|
||||
if err := row.Scan(&etag); err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return false, "", nil
|
||||
}
|
||||
return false, "", err
|
||||
}
|
||||
return true, etag, nil
|
||||
return db.claimSyncLease(ctx, "github_alpine_sync_state", remoteName, owner, freshness, lease)
|
||||
}
|
||||
|
||||
// ReleaseGitHubAlpineSyncLease records the completed scan and frees the lease.
|
||||
// Only the owning replica may release; last_synced_at advances so the next poll
|
||||
// waits a full freshness window, and etag is persisted for the next conditional
|
||||
// request.
|
||||
func (db *DB) ReleaseGitHubAlpineSyncLease(ctx context.Context, remoteName, owner, etag string, syncedAt time.Time) error {
|
||||
_, err := db.Pool.Exec(ctx, `
|
||||
UPDATE github_alpine_sync_state
|
||||
SET last_synced_at = $3, etag = $4, sync_lease_owner = '', sync_lease_expires = NULL
|
||||
WHERE remote_name = $1 AND sync_lease_owner = $2
|
||||
`, remoteName, owner, syncedAt, etag)
|
||||
return err
|
||||
// ReleaseGitHubAlpineSyncLease records a github_alpine scan outcome and frees the
|
||||
// lease. See releaseSyncLease.
|
||||
func (db *DB) ReleaseGitHubAlpineSyncLease(ctx context.Context, remoteName, owner string, res provider.SyncResult) error {
|
||||
return db.releaseSyncLease(ctx, "github_alpine_sync_state", remoteName, owner, res)
|
||||
}
|
||||
|
||||
@@ -2,11 +2,9 @@ package database
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
|
||||
"git.unkin.net/unkin/artifactapi/internal/provider"
|
||||
"git.unkin.net/unkin/artifactapi/pkg/models"
|
||||
)
|
||||
|
||||
@@ -30,41 +28,14 @@ func (db *DB) ListGitHubDebRemotes(ctx context.Context) ([]models.Remote, error)
|
||||
return remotes, rows.Err()
|
||||
}
|
||||
|
||||
// ClaimGitHubDebSyncLease atomically claims the per-remote sync lease. It
|
||||
// succeeds only when the remote is due (never synced, or synced longer than
|
||||
// freshness ago) and no live lease is held by another replica. A zero freshness
|
||||
// (prime scans) ignores the recency gate. The returned etag is the stored
|
||||
// releases-list ETag, shared across replicas.
|
||||
// ClaimGitHubDebSyncLease atomically claims the per-remote github_deb sync
|
||||
// lease. See claimSyncLease.
|
||||
func (db *DB) ClaimGitHubDebSyncLease(ctx context.Context, remoteName, owner string, freshness, lease time.Duration) (bool, string, error) {
|
||||
row := db.Pool.QueryRow(ctx, `
|
||||
INSERT INTO github_deb_sync_state AS s (remote_name, sync_lease_owner, sync_lease_expires)
|
||||
VALUES ($1, $2, now() + make_interval(secs => $4))
|
||||
ON CONFLICT (remote_name) DO UPDATE
|
||||
SET sync_lease_owner = $2,
|
||||
sync_lease_expires = now() + make_interval(secs => $4)
|
||||
WHERE (s.last_synced_at IS NULL OR s.last_synced_at < now() - make_interval(secs => $3))
|
||||
AND (s.sync_lease_expires IS NULL OR s.sync_lease_expires < now())
|
||||
RETURNING s.etag
|
||||
`, remoteName, owner, freshness.Seconds(), lease.Seconds())
|
||||
|
||||
var etag string
|
||||
if err := row.Scan(&etag); err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return false, "", nil
|
||||
}
|
||||
return false, "", err
|
||||
}
|
||||
return true, etag, nil
|
||||
return db.claimSyncLease(ctx, "github_deb_sync_state", remoteName, owner, freshness, lease)
|
||||
}
|
||||
|
||||
// ReleaseGitHubDebSyncLease records the completed scan and frees the lease. Only
|
||||
// the owning replica may release; last_synced_at advances so the next poll waits
|
||||
// a full freshness window, and etag is persisted for the next conditional request.
|
||||
func (db *DB) ReleaseGitHubDebSyncLease(ctx context.Context, remoteName, owner, etag string, syncedAt time.Time) error {
|
||||
_, err := db.Pool.Exec(ctx, `
|
||||
UPDATE github_deb_sync_state
|
||||
SET last_synced_at = $3, etag = $4, sync_lease_owner = '', sync_lease_expires = NULL
|
||||
WHERE remote_name = $1 AND sync_lease_owner = $2
|
||||
`, remoteName, owner, syncedAt, etag)
|
||||
return err
|
||||
// ReleaseGitHubDebSyncLease records a github_deb scan outcome and frees the
|
||||
// lease. See releaseSyncLease.
|
||||
func (db *DB) ReleaseGitHubDebSyncLease(ctx context.Context, remoteName, owner string, res provider.SyncResult) error {
|
||||
return db.releaseSyncLease(ctx, "github_deb_sync_state", remoteName, owner, res)
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.unkin.net/unkin/artifactapi/internal/provider"
|
||||
"git.unkin.net/unkin/artifactapi/pkg/models"
|
||||
)
|
||||
|
||||
@@ -44,7 +45,7 @@ func TestGitHubDebSyncLease(t *testing.T) {
|
||||
t.Fatal("replica-2 claimed while replica-1 holds the lease")
|
||||
}
|
||||
|
||||
if err := testDB.ReleaseGitHubDebSyncLease(ctx(), name, "replica-1", `"etag-1"`, time.Now()); err != nil {
|
||||
if err := testDB.ReleaseGitHubDebSyncLease(ctx(), name, "replica-1", provider.SyncResult{Etag: `"etag-1"`}); err != nil {
|
||||
t.Fatalf("release: %v", err)
|
||||
}
|
||||
|
||||
|
||||
@@ -7,6 +7,8 @@ import (
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
|
||||
"git.unkin.net/unkin/artifactapi/internal/provider"
|
||||
|
||||
"git.unkin.net/unkin/artifactapi/pkg/models"
|
||||
)
|
||||
|
||||
@@ -30,22 +32,35 @@ func (db *DB) ListGitHubRPMRemotes(ctx context.Context) ([]models.Remote, error)
|
||||
return remotes, rows.Err()
|
||||
}
|
||||
|
||||
// ClaimGitHubSyncLease atomically claims the per-remote sync lease. It succeeds
|
||||
// (claimed=true) only when the remote is due — never synced, or synced longer
|
||||
// than freshness ago — and no live lease is held by another replica. This bounds
|
||||
// total GitHub load to roughly one scan per freshness window regardless of how
|
||||
// many replicas poll. The returned etag is the stored releases-list ETag, shared
|
||||
// across replicas so a conditional request can short-circuit an unchanged repo.
|
||||
// A zero freshness (used for prime scans) ignores the recency gate and claims
|
||||
// whenever no live lease is held.
|
||||
// ClaimGitHubSyncLease atomically claims the per-remote github_rpm sync lease.
|
||||
// See claimSyncLease.
|
||||
func (db *DB) ClaimGitHubSyncLease(ctx context.Context, remoteName, owner string, freshness, lease time.Duration) (bool, string, error) {
|
||||
return db.claimSyncLease(ctx, "github_rpm_sync_state", remoteName, owner, freshness, lease)
|
||||
}
|
||||
|
||||
// ReleaseGitHubSyncLease records a github_rpm scan outcome and frees the lease.
|
||||
// See releaseSyncLease.
|
||||
func (db *DB) ReleaseGitHubSyncLease(ctx context.Context, remoteName, owner string, res provider.SyncResult) error {
|
||||
return db.releaseSyncLease(ctx, "github_rpm_sync_state", remoteName, owner, res)
|
||||
}
|
||||
|
||||
// claimSyncLease atomically claims a per-remote sync lease in table. It succeeds
|
||||
// (claimed=true) only when the remote is due and no live lease is held by
|
||||
// another replica. Due means: a pending retry after a failed scan has reached
|
||||
// next_retry_at, or, with no retry pending, the remote was never synced or was
|
||||
// synced longer than freshness ago. A zero freshness (prime scans) ignores the
|
||||
// recency gate but still honours a pending retry's backoff. The returned etag is
|
||||
// the stored releases-list ETag, shared across replicas so a conditional request
|
||||
// can short-circuit an unchanged repo.
|
||||
func (db *DB) claimSyncLease(ctx context.Context, table, remoteName, owner string, freshness, lease time.Duration) (bool, string, error) {
|
||||
row := db.Pool.QueryRow(ctx, `
|
||||
INSERT INTO github_rpm_sync_state AS s (remote_name, sync_lease_owner, sync_lease_expires)
|
||||
INSERT INTO `+table+` AS s (remote_name, sync_lease_owner, sync_lease_expires)
|
||||
VALUES ($1, $2, now() + make_interval(secs => $4))
|
||||
ON CONFLICT (remote_name) DO UPDATE
|
||||
SET sync_lease_owner = $2,
|
||||
sync_lease_expires = now() + make_interval(secs => $4)
|
||||
WHERE (s.last_synced_at IS NULL OR s.last_synced_at < now() - make_interval(secs => $3))
|
||||
WHERE (CASE WHEN s.next_retry_at IS NOT NULL THEN s.next_retry_at <= now()
|
||||
ELSE s.last_synced_at IS NULL OR s.last_synced_at < now() - make_interval(secs => $3) END)
|
||||
AND (s.sync_lease_expires IS NULL OR s.sync_lease_expires < now())
|
||||
RETURNING s.etag
|
||||
`, remoteName, owner, freshness.Seconds(), lease.Seconds())
|
||||
@@ -60,14 +75,27 @@ func (db *DB) ClaimGitHubSyncLease(ctx context.Context, remoteName, owner string
|
||||
return true, etag, nil
|
||||
}
|
||||
|
||||
// ReleaseGitHubSyncLease records the completed scan and frees the lease. Only the
|
||||
// owning replica may release; last_synced_at advances so the next poll waits a
|
||||
// full freshness window, and etag is persisted for the next conditional request.
|
||||
func (db *DB) ReleaseGitHubSyncLease(ctx context.Context, remoteName, owner, etag string, syncedAt time.Time) error {
|
||||
// releaseSyncLease records a scan outcome and frees the lease; only the owning
|
||||
// replica may release. Success advances last_synced_at, persists the etag and
|
||||
// clears any retry. Failure leaves last_synced_at and etag untouched (the last
|
||||
// good metadata keeps serving) and schedules next_retry_at with exponential
|
||||
// backoff from the consecutive-failure count, never before res.RetryAt.
|
||||
func (db *DB) releaseSyncLease(ctx context.Context, table, remoteName, owner string, res provider.SyncResult) error {
|
||||
var retryAt *time.Time
|
||||
if !res.RetryAt.IsZero() {
|
||||
retryAt = &res.RetryAt
|
||||
}
|
||||
_, err := db.Pool.Exec(ctx, `
|
||||
UPDATE github_rpm_sync_state
|
||||
SET last_synced_at = $3, etag = $4, sync_lease_owner = '', sync_lease_expires = NULL
|
||||
UPDATE `+table+`
|
||||
SET last_synced_at = CASE WHEN $3 THEN last_synced_at ELSE now() END,
|
||||
etag = CASE WHEN $3 THEN etag ELSE $4 END,
|
||||
sync_failures = CASE WHEN $3 THEN sync_failures + 1 ELSE 0 END,
|
||||
next_retry_at = CASE WHEN $3 THEN GREATEST(
|
||||
now() + make_interval(secs => LEAST($5 * power(2, LEAST(sync_failures, 20)), $6)),
|
||||
$7::timestamptz)
|
||||
END,
|
||||
sync_lease_owner = '', sync_lease_expires = NULL
|
||||
WHERE remote_name = $1 AND sync_lease_owner = $2
|
||||
`, remoteName, owner, syncedAt, etag)
|
||||
`, remoteName, owner, res.Failed, res.Etag, res.Backoff.Seconds(), res.MaxBackoff.Seconds(), retryAt)
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.unkin.net/unkin/artifactapi/internal/provider"
|
||||
"git.unkin.net/unkin/artifactapi/pkg/models"
|
||||
)
|
||||
|
||||
@@ -47,7 +48,7 @@ func TestGitHubSyncLease(t *testing.T) {
|
||||
}
|
||||
|
||||
// Replica 1 finishes: record the sync and persist an etag.
|
||||
if err := testDB.ReleaseGitHubSyncLease(ctx(), name, "replica-1", `"etag-1"`, time.Now()); err != nil {
|
||||
if err := testDB.ReleaseGitHubSyncLease(ctx(), name, "replica-1", provider.SyncResult{Etag: `"etag-1"`}); err != nil {
|
||||
t.Fatalf("release: %v", err)
|
||||
}
|
||||
|
||||
@@ -93,3 +94,116 @@ func TestListGitHubRPMRemotes(t *testing.T) {
|
||||
t.Fatalf("seeded remote %q not returned", name)
|
||||
}
|
||||
}
|
||||
|
||||
// A failed scan must not count as a sync: the etag and last_synced_at stay put,
|
||||
// every claim (prime included) is held off until the backoff elapses, the delay
|
||||
// doubles per consecutive failure up to the cap, an upstream retry hint pushes
|
||||
// it later, and a success clears the retry state.
|
||||
func TestGitHubSyncLeaseFailureBackoff(t *testing.T) {
|
||||
requireDB(t)
|
||||
name := "gh-retry-" + time.Now().Format("150405.000000")
|
||||
seedGitHubRPMRemote(t, name)
|
||||
|
||||
const lease = 15 * time.Minute
|
||||
freshness := time.Hour
|
||||
claim := func(f time.Duration) (bool, string) {
|
||||
t.Helper()
|
||||
ok, etag, err := testDB.ClaimGitHubSyncLease(ctx(), name, "r1", f, lease)
|
||||
if err != nil {
|
||||
t.Fatalf("claim: %v", err)
|
||||
}
|
||||
return ok, etag
|
||||
}
|
||||
release := func(res provider.SyncResult) {
|
||||
t.Helper()
|
||||
if err := testDB.ReleaseGitHubSyncLease(ctx(), name, "r1", res); err != nil {
|
||||
t.Fatalf("release: %v", err)
|
||||
}
|
||||
}
|
||||
state := func() (failures int, retryIn time.Duration, synced *time.Time, etag string) {
|
||||
t.Helper()
|
||||
var next *time.Time
|
||||
var dbNow time.Time
|
||||
if err := testDB.Pool.QueryRow(ctx(), `SELECT sync_failures, next_retry_at, last_synced_at, etag, now() FROM github_rpm_sync_state WHERE remote_name = $1`, name).
|
||||
Scan(&failures, &next, &synced, &etag, &dbNow); err != nil {
|
||||
t.Fatalf("read state: %v", err)
|
||||
}
|
||||
if next != nil {
|
||||
retryIn = next.Sub(dbNow)
|
||||
}
|
||||
return
|
||||
}
|
||||
elapse := func() {
|
||||
t.Helper()
|
||||
if _, err := testDB.Pool.Exec(ctx(), `UPDATE github_rpm_sync_state SET next_retry_at = now() - interval '1 second' WHERE remote_name = $1`, name); err != nil {
|
||||
t.Fatalf("elapse backoff: %v", err)
|
||||
}
|
||||
}
|
||||
failed := provider.SyncResult{Failed: true, Etag: `"ignored"`, Backoff: time.Minute, MaxBackoff: 3 * time.Minute}
|
||||
|
||||
// Last good sync.
|
||||
if ok, _ := claim(freshness); !ok {
|
||||
t.Fatal("first claim failed")
|
||||
}
|
||||
release(provider.SyncResult{Etag: `"good"`})
|
||||
_, _, goodSynced, _ := state()
|
||||
|
||||
// First failure: 60s backoff, sync time and etag untouched.
|
||||
if ok, _ := claim(0); !ok {
|
||||
t.Fatal("prime claim failed")
|
||||
}
|
||||
release(failed)
|
||||
n, in, synced, etag := state()
|
||||
if n != 1 || in < 55*time.Second || in > 65*time.Second {
|
||||
t.Fatalf("after 1st failure: failures=%d retry in %v, want 1 and ~60s", n, in)
|
||||
}
|
||||
if etag != `"good"` || synced == nil || !synced.Equal(*goodSynced) {
|
||||
t.Fatalf("failed scan overwrote sync state: etag=%q synced=%v", etag, synced)
|
||||
}
|
||||
if ok, _ := claim(0); ok {
|
||||
t.Fatal("prime claimed inside the retry backoff")
|
||||
}
|
||||
|
||||
// Backoff elapsed: claimable despite being inside the freshness window,
|
||||
// and the second failure doubles the delay.
|
||||
elapse()
|
||||
if ok, _ := claim(freshness); !ok {
|
||||
t.Fatal("periodic claim refused after backoff elapsed")
|
||||
}
|
||||
release(failed)
|
||||
if n, in, _, _ := state(); n != 2 || in < 115*time.Second || in > 125*time.Second {
|
||||
t.Fatalf("after 2nd failure: failures=%d retry in %v, want 2 and ~120s", n, in)
|
||||
}
|
||||
|
||||
// Third failure caps at MaxBackoff (3m, not 4m).
|
||||
elapse()
|
||||
claim(freshness)
|
||||
release(failed)
|
||||
if n, in, _, _ := state(); n != 3 || in < 175*time.Second || in > 185*time.Second {
|
||||
t.Fatalf("after 3rd failure: failures=%d retry in %v, want 3 and ~180s", n, in)
|
||||
}
|
||||
|
||||
// An upstream rate-limit reset later than the backoff wins.
|
||||
elapse()
|
||||
claim(freshness)
|
||||
hinted := failed
|
||||
hinted.RetryAt = time.Now().Add(20 * time.Minute)
|
||||
release(hinted)
|
||||
if _, in, _, _ := state(); in < 19*time.Minute || in > 21*time.Minute {
|
||||
t.Fatalf("retry hint ignored: retry in %v, want ~20m", in)
|
||||
}
|
||||
|
||||
// Recovery: success persists the new etag and clears the retry, so the
|
||||
// normal freshness window applies again.
|
||||
elapse()
|
||||
if ok, etag := claim(freshness); !ok || etag != `"good"` {
|
||||
t.Fatalf("recovery claim ok=%v etag=%q, want last good etag", ok, etag)
|
||||
}
|
||||
release(provider.SyncResult{Etag: `"new"`})
|
||||
if n, in, _, etag := state(); n != 0 || in != 0 || etag != `"new"` {
|
||||
t.Fatalf("after success: failures=%d retry in %v etag=%q", n, in, etag)
|
||||
}
|
||||
if ok, _ := claim(freshness); ok {
|
||||
t.Fatal("claimed inside freshness window after a successful sync")
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user