Compute sync slots and the first-sync gate on the DB clock
ci/woodpecker/pr/pre-commit Pipeline was successful
ci/woodpecker/pr/build Pipeline was successful
ci/woodpecker/pr/test Pipeline was successful

This commit is contained in:
2026-10-25 13:57:54 +11:00
parent 1582533e5a
commit 82b1852bc4
13 changed files with 179 additions and 238 deletions
+2 -2
View File
@@ -30,8 +30,8 @@ func (db *DB) ListGitHubAlpineRemotes(ctx context.Context) ([]models.Remote, err
// 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, slot time.Time) (bool, string, error) {
return db.claimSyncLease(ctx, "github_alpine_sync_state", remoteName, owner, freshness, lease, slot)
func (db *DB) ClaimGitHubAlpineSyncLease(ctx context.Context, remoteName, owner string, freshness, lease, interval, phase time.Duration) (bool, string, error) {
return db.claimSyncLease(ctx, "github_alpine_sync_state", remoteName, owner, freshness, lease, interval, phase)
}
// ReleaseGitHubAlpineSyncLease records a github_alpine scan outcome and frees the
+2 -2
View File
@@ -30,8 +30,8 @@ func (db *DB) ListGitHubDebRemotes(ctx context.Context) ([]models.Remote, error)
// 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, slot time.Time) (bool, string, error) {
return db.claimSyncLease(ctx, "github_deb_sync_state", remoteName, owner, freshness, lease, slot)
func (db *DB) ClaimGitHubDebSyncLease(ctx context.Context, remoteName, owner string, freshness, lease, interval, phase time.Duration) (bool, string, error) {
return db.claimSyncLease(ctx, "github_deb_sync_state", remoteName, owner, freshness, lease, interval, phase)
}
// ReleaseGitHubDebSyncLease records a github_deb scan outcome and frees the
+4 -4
View File
@@ -29,7 +29,7 @@ func TestGitHubDebSyncLease(t *testing.T) {
const lease = 15 * time.Minute
freshness := time.Hour
claimed, etag, err := testDB.ClaimGitHubDebSyncLease(ctx(), name, "replica-1", freshness, lease, time.Time{})
claimed, etag, err := testDB.ClaimGitHubDebSyncLease(ctx(), name, "replica-1", freshness, lease, 0, 0)
if err != nil || !claimed {
t.Fatalf("replica-1 first claim: claimed=%v err=%v", claimed, err)
}
@@ -37,7 +37,7 @@ func TestGitHubDebSyncLease(t *testing.T) {
t.Fatalf("initial etag should be empty, got %q", etag)
}
claimed2, _, err := testDB.ClaimGitHubDebSyncLease(ctx(), name, "replica-2", freshness, lease, time.Time{})
claimed2, _, err := testDB.ClaimGitHubDebSyncLease(ctx(), name, "replica-2", freshness, lease, 0, 0)
if err != nil {
t.Fatalf("replica-2 claim err: %v", err)
}
@@ -49,7 +49,7 @@ func TestGitHubDebSyncLease(t *testing.T) {
t.Fatalf("release: %v", err)
}
claimed3, _, err := testDB.ClaimGitHubDebSyncLease(ctx(), name, "replica-2", freshness, lease, time.Time{})
claimed3, _, err := testDB.ClaimGitHubDebSyncLease(ctx(), name, "replica-2", freshness, lease, 0, 0)
if err != nil {
t.Fatalf("replica-2 recency claim err: %v", err)
}
@@ -57,7 +57,7 @@ func TestGitHubDebSyncLease(t *testing.T) {
t.Fatal("periodic claim succeeded inside the freshness window")
}
claimed4, etag4, err := testDB.ClaimGitHubDebSyncLease(ctx(), name, "replica-2", 0, lease, time.Time{})
claimed4, etag4, err := testDB.ClaimGitHubDebSyncLease(ctx(), name, "replica-2", 0, lease, 0, 0)
if err != nil || !claimed4 {
t.Fatalf("prime claim: claimed=%v err=%v", claimed4, err)
}
+17 -14
View File
@@ -34,8 +34,8 @@ func (db *DB) ListGitHubRPMRemotes(ctx context.Context) ([]models.Remote, error)
// 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, slot time.Time) (bool, string, error) {
return db.claimSyncLease(ctx, "github_rpm_sync_state", remoteName, owner, freshness, lease, slot)
func (db *DB) ClaimGitHubSyncLease(ctx context.Context, remoteName, owner string, freshness, lease, interval, phase time.Duration) (bool, string, error) {
return db.claimSyncLease(ctx, "github_rpm_sync_state", remoteName, owner, freshness, lease, interval, phase)
}
// ReleaseGitHubSyncLease records a github_rpm scan outcome and frees the lease.
@@ -48,27 +48,30 @@ func (db *DB) ReleaseGitHubSyncLease(ctx context.Context, remoteName, owner stri
// (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
// last synced before slot (when set) or longer than freshness ago (otherwise).
// A zero freshness with no slot (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
// last synced before the current slot (interval set) or longer than freshness
// ago (otherwise). With interval set, slots start at phase past each multiple
// of interval and a remote created inside the current slot waits for the next.
// Slots are computed from the DB clock so replicas agree whatever their own
// clocks say. A zero freshness and interval (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, slot time.Time) (bool, string, error) {
var slotArg *time.Time
if !slot.IsZero() {
slotArg = &slot
}
func (db *DB) claimSyncLease(ctx context.Context, table, remoteName, owner string, freshness, lease, interval, phase time.Duration) (bool, string, error) {
const slot = `to_timestamp(floor((extract(epoch FROM now())::float8 - $6::float8) / NULLIF($5::float8, 0)) * $5::float8 + $6::float8)`
row := db.Pool.QueryRow(ctx, `
INSERT INTO `+table+` AS s (remote_name, sync_lease_owner, sync_lease_expires)
VALUES ($1, $2, now() + make_interval(secs => $4))
SELECT r.name, $2, now() + make_interval(secs => $4)
FROM remotes r
WHERE r.name = $1
AND ($5::float8 <= 0 OR COALESCE(r.created_at, '-infinity') < `+slot+`)
ON CONFLICT (remote_name) DO UPDATE
SET sync_lease_owner = $2,
sync_lease_expires = now() + make_interval(secs => $4)
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 < COALESCE($5::timestamptz, now() - make_interval(secs => $3)) END)
ELSE s.last_synced_at IS NULL OR s.last_synced_at < COALESCE(`+slot+`, 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(), slotArg)
`, remoteName, owner, freshness.Seconds(), lease.Seconds(), interval.Seconds(), phase.Seconds())
var etag string
if err := row.Scan(&etag); err != nil {
+66 -36
View File
@@ -2,6 +2,9 @@ package database
import (
"context"
"fmt"
"math"
"sync/atomic"
"testing"
"time"
@@ -31,7 +34,7 @@ func TestGitHubSyncLease(t *testing.T) {
freshness := time.Hour
// First claim on a never-synced remote wins; etag starts empty.
claimed, etag, err := testDB.ClaimGitHubSyncLease(ctx(), name, "replica-1", freshness, lease, time.Time{})
claimed, etag, err := testDB.ClaimGitHubSyncLease(ctx(), name, "replica-1", freshness, lease, 0, 0)
if err != nil || !claimed {
t.Fatalf("replica-1 first claim: claimed=%v err=%v", claimed, err)
}
@@ -40,7 +43,7 @@ func TestGitHubSyncLease(t *testing.T) {
}
// A second replica cannot claim while the lease is held.
claimed2, _, err := testDB.ClaimGitHubSyncLease(ctx(), name, "replica-2", freshness, lease, time.Time{})
claimed2, _, err := testDB.ClaimGitHubSyncLease(ctx(), name, "replica-2", freshness, lease, 0, 0)
if err != nil {
t.Fatalf("replica-2 claim err: %v", err)
}
@@ -54,7 +57,7 @@ func TestGitHubSyncLease(t *testing.T) {
}
// A periodic re-claim inside the freshness window is blocked by recency.
claimed3, _, err := testDB.ClaimGitHubSyncLease(ctx(), name, "replica-2", freshness, lease, time.Time{})
claimed3, _, err := testDB.ClaimGitHubSyncLease(ctx(), name, "replica-2", freshness, lease, 0, 0)
if err != nil {
t.Fatalf("replica-2 recency claim err: %v", err)
}
@@ -63,7 +66,7 @@ func TestGitHubSyncLease(t *testing.T) {
}
// A prime (freshness 0) bypasses recency and reads the persisted etag.
claimed4, etag4, err := testDB.ClaimGitHubSyncLease(ctx(), name, "replica-2", 0, lease, time.Time{})
claimed4, etag4, err := testDB.ClaimGitHubSyncLease(ctx(), name, "replica-2", 0, lease, 0, 0)
if err != nil || !claimed4 {
t.Fatalf("prime claim: claimed=%v err=%v", claimed4, err)
}
@@ -109,7 +112,7 @@ func TestGitHubSyncLeaseFailureBackoff(t *testing.T) {
freshness := time.Hour
claim := func(f time.Duration) (bool, string) {
t.Helper()
ok, etag, err := testDB.ClaimGitHubSyncLease(ctx(), name, "r1", f, lease, time.Time{})
ok, etag, err := testDB.ClaimGitHubSyncLease(ctx(), name, "r1", f, lease, 0, 0)
if err != nil {
t.Fatalf("claim: %v", err)
}
@@ -221,21 +224,21 @@ func TestGitHubSyncStateDeletedWithRemote(t *testing.T) {
}{
{"github_rpm_sync_state", models.PackageGitHubRPM,
func(n string) (bool, string, error) {
return testDB.ClaimGitHubSyncLease(ctx(), n, "r1", time.Hour, time.Minute, time.Time{})
return testDB.ClaimGitHubSyncLease(ctx(), n, "r1", time.Hour, time.Minute, 0, 0)
},
func(n string, res provider.SyncResult) error {
return testDB.ReleaseGitHubSyncLease(ctx(), n, "r1", res)
}},
{"github_deb_sync_state", models.PackageGitHubDeb,
func(n string) (bool, string, error) {
return testDB.ClaimGitHubDebSyncLease(ctx(), n, "r1", time.Hour, time.Minute, time.Time{})
return testDB.ClaimGitHubDebSyncLease(ctx(), n, "r1", time.Hour, time.Minute, 0, 0)
},
func(n string, res provider.SyncResult) error {
return testDB.ReleaseGitHubDebSyncLease(ctx(), n, "r1", res)
}},
{"github_alpine_sync_state", models.PackageGitHubAlpine,
func(n string) (bool, string, error) {
return testDB.ClaimGitHubAlpineSyncLease(ctx(), n, "r1", time.Hour, time.Minute, time.Time{})
return testDB.ClaimGitHubAlpineSyncLease(ctx(), n, "r1", time.Hour, time.Minute, 0, 0)
},
func(n string, res provider.SyncResult) error {
return testDB.ReleaseGitHubAlpineSyncLease(ctx(), n, "r1", res)
@@ -281,13 +284,13 @@ func TestGitHubSyncStateDeletedWithRemote(t *testing.T) {
type syncTable struct {
table string
pt models.PackageType
claim func(name, owner string, freshness time.Duration, slot time.Time) (bool, error)
claim func(name, owner string, freshness, interval, phase time.Duration) (bool, error)
}
func syncTables() []syncTable {
wrap := func(fn func(context.Context, string, string, time.Duration, time.Duration, time.Time) (bool, string, error)) func(string, string, time.Duration, time.Time) (bool, error) {
return func(name, owner string, freshness time.Duration, slot time.Time) (bool, error) {
ok, _, err := fn(ctx(), name, owner, freshness, 15*time.Minute, slot)
wrap := func(fn func(context.Context, string, string, time.Duration, time.Duration, time.Duration, time.Duration) (bool, string, error)) func(string, string, time.Duration, time.Duration, time.Duration) (bool, error) {
return func(name, owner string, freshness, interval, phase time.Duration) (bool, error) {
ok, _, err := fn(ctx(), name, owner, freshness, 15*time.Minute, interval, phase)
return ok, err
}
}
@@ -298,32 +301,43 @@ func syncTables() []syncTable {
}
}
// seedSynced creates a remote of tc's type with a released lease last synced at.
func seedSynced(t *testing.T, tc syncTable, prefix string, at time.Time) string {
var syncSeq atomic.Int64
// seedSynced creates a remote of tc's type created createdAgo before the DB
// clock and, when syncedAgo >= 0, a released lease last synced syncedAgo before it.
func seedSynced(t *testing.T, tc syncTable, createdAgo, syncedAgo time.Duration) string {
t.Helper()
name := prefix + "-" + tc.table + "-" + time.Now().Format("150405.000000")
name := fmt.Sprintf("sync-%s-%d", tc.table, syncSeq.Add(1))
if err := testDB.CreateRemote(ctx(), &models.Remote{
Name: name, PackageType: tc.pt, RepoType: models.RepoTypeRemote,
BaseURL: "https://api.github.com/repos/acme/tools", ReleasesRemote: "github",
}); err != nil {
t.Fatalf("create remote: %v", err)
}
if _, err := testDB.Pool.Exec(ctx(), `INSERT INTO `+tc.table+` (remote_name, last_synced_at) VALUES ($1, $2)`, name, at); err != nil {
t.Fatalf("seed sync state: %v", err)
if _, err := testDB.Pool.Exec(ctx(), `UPDATE remotes SET created_at = now() - make_interval(secs => $2) WHERE name = $1`, name, createdAgo.Seconds()); err != nil {
t.Fatalf("seed created_at: %v", err)
}
if syncedAgo >= 0 {
if _, err := testDB.Pool.Exec(ctx(), `INSERT INTO `+tc.table+` (remote_name, last_synced_at) VALUES ($1, now() - make_interval(secs => $2))`, name, syncedAgo.Seconds()); err != nil {
t.Fatalf("seed sync state: %v", err)
}
}
return name
}
func mustClaim(t *testing.T, tc syncTable, name, owner string, freshness time.Duration, slot time.Time) bool {
func mustClaim(t *testing.T, tc syncTable, name, owner string, freshness, interval, phase time.Duration) bool {
t.Helper()
ok, err := tc.claim(name, owner, freshness, slot)
ok, err := tc.claim(name, owner, freshness, interval, phase)
if err != nil {
t.Fatalf("claim: %v", err)
}
return ok
}
// With no slot the claim keeps the freshness-since-last-sync predicate.
const neverSynced = -1
// With no interval the claim keeps the freshness-since-last-sync predicate and
// a new remote is due at once.
func TestSyncLeaseUnsetIntervalFreshness(t *testing.T) {
requireDB(t)
const ttl = time.Hour
@@ -334,13 +348,14 @@ func TestSyncLeaseUnsetIntervalFreshness(t *testing.T) {
freshness time.Duration
due bool
}{
{neverSynced, ttl, true},
{ttl + time.Minute, ttl, true},
{ttl - time.Minute, ttl, false},
{time.Second, ttl, false},
{time.Second, 0, true},
} {
name := seedSynced(t, tc, "fresh", time.Now().Add(-c.ago))
if got := mustClaim(t, tc, name, "r1", c.freshness, time.Time{}); got != c.due {
name := seedSynced(t, tc, time.Second, c.ago)
if got := mustClaim(t, tc, name, "r1", c.freshness, 0, 0); got != c.due {
t.Errorf("synced %v ago, freshness %v: claimed=%v, want %v", c.ago, c.freshness, got, c.due)
}
}
@@ -348,24 +363,39 @@ func TestSyncLeaseUnsetIntervalFreshness(t *testing.T) {
}
}
// With a slot the remote is due only when last synced before it, regardless of
// freshness, and still one replica at a time.
// With an interval the slot comes from the DB clock: a remote is due only when
// created and last synced before the current slot, regardless of freshness,
// and still one replica at a time.
func TestSyncLeaseSlot(t *testing.T) {
requireDB(t)
slot := time.Date(2026, 3, 1, 10, 17, 42, 0, time.UTC)
var epoch float64
if err := testDB.Pool.QueryRow(ctx(), `SELECT extract(epoch FROM now())::float8`).Scan(&epoch); err != nil {
t.Fatalf("db now: %v", err)
}
const interval = time.Hour
// Pick the phase that puts the current slot's start 30m before the DB now.
phase := time.Duration(math.Mod(epoch-1800, interval.Seconds()) * float64(time.Second))
const day = 24 * time.Hour
for _, tc := range syncTables() {
t.Run(tc.table, func(t *testing.T) {
name := seedSynced(t, tc, "slot", slot.Add(time.Second))
if mustClaim(t, tc, name, "r1", 0, slot) {
t.Fatal("claimed though synced inside the slot")
}
name = seedSynced(t, tc, "slot", slot.Add(-time.Second))
if !mustClaim(t, tc, name, "r1", time.Hour*24*365*10, slot) {
t.Fatal("not due though last synced before the slot")
}
if mustClaim(t, tc, name, "r2", 0, slot) {
t.Fatal("second replica claimed a held lease")
for _, c := range []struct {
desc string
createdAgo, ago time.Duration
freshness time.Duration
due bool
}{
{"created inside the slot", 10 * time.Second, neverSynced, 0, false},
{"created before the slot", 35 * time.Minute, neverSynced, 0, true},
{"synced inside the slot", day, 25 * time.Minute, 0, false},
{"synced before the slot", day, 35 * time.Minute, 10 * 365 * day, true},
} {
name := seedSynced(t, tc, c.createdAgo, c.ago)
if got := mustClaim(t, tc, name, "r1", c.freshness, interval, phase); got != c.due {
t.Errorf("%s: claimed=%v, want %v", c.desc, got, c.due)
}
if c.due && mustClaim(t, tc, name, "r2", 0, interval, phase) {
t.Errorf("%s: second replica claimed a held lease", c.desc)
}
}
})
}
+4 -7
View File
@@ -27,7 +27,7 @@ const (
type SyncStore interface {
provider.RemoteMetadataStore
ListGitHubAlpineRemotes(ctx context.Context) ([]models.Remote, error)
ClaimGitHubAlpineSyncLease(ctx context.Context, remoteName, owner string, freshness, lease time.Duration, slot time.Time) (claimed bool, etag string, err error)
ClaimGitHubAlpineSyncLease(ctx context.Context, remoteName, owner string, freshness, lease, interval, phase time.Duration) (claimed bool, etag string, err error)
ReleaseGitHubAlpineSyncLease(ctx context.Context, remoteName, owner string, res provider.SyncResult) error
}
@@ -194,17 +194,14 @@ func (s *Syncer) process(ctx context.Context, job syncJob) {
ttl := job.remote.SyncPeriod(defaultSyncFreshness)
freshness := ttl
var slot time.Time
var interval, phase time.Duration
if job.prime {
freshness = 0
} else {
slot = job.remote.SyncSlot(time.Now())
if !slot.IsZero() && !job.remote.CreatedAt.Before(slot) {
return // a scheduled first sync waits for the remote's first slot
}
interval, phase = job.remote.SyncSchedule()
}
claimed, etag, err := s.store.ClaimGitHubAlpineSyncLease(ctx, job.remote.Name, s.owner, freshness, syncLeaseDuration, slot)
claimed, etag, err := s.store.ClaimGitHubAlpineSyncLease(ctx, job.remote.Name, s.owner, freshness, syncLeaseDuration, interval, phase)
if err != nil {
slog.Error("github_alpine syncer: claim lease", "remote", job.remote.Name, "error", err)
return
+20 -33
View File
@@ -29,6 +29,8 @@ type fakeSyncStore struct {
etags map[string]string
retryAt map[string]time.Time
freshness time.Duration
interval time.Duration
phase time.Duration
attempts int
}
@@ -49,15 +51,15 @@ func (f *fakeSyncStore) ListGitHubAlpineRemotes(_ context.Context) ([]models.Rem
return append([]models.Remote(nil), f.remotes...), nil
}
func (f *fakeSyncStore) ClaimGitHubAlpineSyncLease(_ context.Context, name, owner string, freshness, lease time.Duration, slot time.Time) (bool, string, error) {
func (f *fakeSyncStore) ClaimGitHubAlpineSyncLease(_ context.Context, name, owner string, freshness, lease, interval, phase time.Duration) (bool, string, error) {
f.mu.Lock()
defer f.mu.Unlock()
f.freshness = freshness
f.freshness, f.interval, f.phase = freshness, interval, phase
f.attempts++
now := time.Now()
ls, hasLS := f.lastSynced[name]
exp, hasExp := f.leaseExp[name]
freshOK := !hasLS || (slot.IsZero() && now.Sub(ls) >= freshness) || (!slot.IsZero() && ls.Before(slot))
freshOK := !hasLS || now.Sub(ls) >= freshness
if ra, pending := f.retryAt[name]; pending {
freshOK = !now.Before(ra)
}
@@ -227,7 +229,7 @@ func TestSyncerLeasePreventsSecondReplica(t *testing.T) {
s := newSyncer(store, p, testSyncConfig())
remote := fx.remote()
claimed, _, err := store.ClaimGitHubAlpineSyncLease(context.Background(), remote.Name, "replica-1", time.Duration(remote.MutableTTL)*time.Second, syncLeaseDuration, time.Time{})
claimed, _, err := store.ClaimGitHubAlpineSyncLease(context.Background(), remote.Name, "replica-1", time.Duration(remote.MutableTTL)*time.Second, syncLeaseDuration, 0, 0)
if err != nil || !claimed {
t.Fatalf("replica-1 claim: claimed=%v err=%v", claimed, err)
}
@@ -368,51 +370,36 @@ func TestSyncerFailedAssetRetry(t *testing.T) {
}
}
// A periodic check claims with sync_interval as its freshness when set, and
// falls back to mutable_ttl when unset.
// A periodic check claims with sync_interval as its freshness and passes the
// remote's slot schedule; unset, it falls back to mutable_ttl with no schedule.
func TestSyncerHonoursSyncInterval(t *testing.T) {
for _, tc := range []struct {
interval int
want time.Duration
}{
{0, 90 * time.Second},
{7200, 2 * time.Hour},
} {
for _, interval := range []int{0, 7200} {
store := newFakeSyncStore()
s := newSyncer(store, newTestProvider(), testSyncConfig())
store.lastSynced["sched"] = time.Now()
remote := models.Remote{Name: "sched", MutableTTL: 90, SyncInterval: tc.interval}
remote := models.Remote{Name: "sched", MutableTTL: 90, SyncInterval: interval}
wantInterval, wantPhase := remote.SyncSchedule()
s.process(context.Background(), syncJob{remote: remote})
if store.freshness != tc.want {
t.Errorf("sync_interval=%d: claim freshness = %v, want %v", tc.interval, store.freshness, tc.want)
if want := remote.SyncPeriod(0); store.freshness != want || store.interval != wantInterval || store.phase != wantPhase {
t.Errorf("sync_interval=%d: claim freshness/interval/phase = %v/%v/%v, want %v/%v/%v",
interval, store.freshness, store.interval, store.phase, want, wantInterval, wantPhase)
}
}
}
// A scheduled remote's first sync waits for its first slot, and create-time
// primes are skipped, so remotes created together spread out; a cold-cache
// prime still claims at once.
func TestSyncerFirstSyncWaitsForSlot(t *testing.T) {
// Create-time primes are skipped for a scheduled remote so remotes created
// together spread out; a cold-cache prime still claims with no schedule.
func TestSyncerScheduledPrime(t *testing.T) {
store := newFakeSyncStore()
s := newSyncer(store, newTestProvider(), testSyncConfig())
remote := models.Remote{Name: "fresh", SyncInterval: 3600, CreatedAt: time.Now()}
remote := models.Remote{Name: "fresh", SyncInterval: 3600}
s.process(context.Background(), syncJob{remote: remote})
if store.attempts != 0 {
t.Fatal("scheduled first sync claimed before its first slot")
}
s.EnqueuePrime(remote)
if len(s.jobs) != 0 {
t.Fatal("create-time prime enqueued for a sync_interval remote")
}
s.process(context.Background(), syncJob{remote: remote, prime: true})
if store.attempts != 1 {
t.Fatal("cold-cache prime did not claim")
}
remote.CreatedAt = time.Now().Add(-2 * time.Hour)
s.process(context.Background(), syncJob{remote: remote})
if store.attempts != 2 {
t.Fatal("remote created before its slot did not claim")
if store.attempts != 1 || store.freshness != 0 || store.interval != 0 {
t.Fatalf("cold-cache prime: attempts=%d freshness=%v interval=%v, want 1/0/0", store.attempts, store.freshness, store.interval)
}
}
+4 -7
View File
@@ -27,7 +27,7 @@ const (
type SyncStore interface {
provider.RemoteMetadataStore
ListGitHubDebRemotes(ctx context.Context) ([]models.Remote, error)
ClaimGitHubDebSyncLease(ctx context.Context, remoteName, owner string, freshness, lease time.Duration, slot time.Time) (claimed bool, etag string, err error)
ClaimGitHubDebSyncLease(ctx context.Context, remoteName, owner string, freshness, lease, interval, phase time.Duration) (claimed bool, etag string, err error)
ReleaseGitHubDebSyncLease(ctx context.Context, remoteName, owner string, res provider.SyncResult) error
}
@@ -194,17 +194,14 @@ func (s *Syncer) process(ctx context.Context, job syncJob) {
ttl := job.remote.SyncPeriod(defaultSyncFreshness)
freshness := ttl
var slot time.Time
var interval, phase time.Duration
if job.prime {
freshness = 0
} else {
slot = job.remote.SyncSlot(time.Now())
if !slot.IsZero() && !job.remote.CreatedAt.Before(slot) {
return // a scheduled first sync waits for the remote's first slot
}
interval, phase = job.remote.SyncSchedule()
}
claimed, etag, err := s.store.ClaimGitHubDebSyncLease(ctx, job.remote.Name, s.owner, freshness, syncLeaseDuration, slot)
claimed, etag, err := s.store.ClaimGitHubDebSyncLease(ctx, job.remote.Name, s.owner, freshness, syncLeaseDuration, interval, phase)
if err != nil {
slog.Error("github_deb syncer: claim lease", "remote", job.remote.Name, "error", err)
return
+20 -33
View File
@@ -29,6 +29,8 @@ type fakeSyncStore struct {
etags map[string]string
retryAt map[string]time.Time
freshness time.Duration
interval time.Duration
phase time.Duration
attempts int
}
@@ -49,15 +51,15 @@ func (f *fakeSyncStore) ListGitHubDebRemotes(_ context.Context) ([]models.Remote
return append([]models.Remote(nil), f.remotes...), nil
}
func (f *fakeSyncStore) ClaimGitHubDebSyncLease(_ context.Context, name, owner string, freshness, lease time.Duration, slot time.Time) (bool, string, error) {
func (f *fakeSyncStore) ClaimGitHubDebSyncLease(_ context.Context, name, owner string, freshness, lease, interval, phase time.Duration) (bool, string, error) {
f.mu.Lock()
defer f.mu.Unlock()
f.freshness = freshness
f.freshness, f.interval, f.phase = freshness, interval, phase
f.attempts++
now := time.Now()
ls, hasLS := f.lastSynced[name]
exp, hasExp := f.leaseExp[name]
freshOK := !hasLS || (slot.IsZero() && now.Sub(ls) >= freshness) || (!slot.IsZero() && ls.Before(slot))
freshOK := !hasLS || now.Sub(ls) >= freshness
if ra, pending := f.retryAt[name]; pending {
freshOK = !now.Before(ra)
}
@@ -227,7 +229,7 @@ func TestSyncerLeasePreventsSecondReplica(t *testing.T) {
s := newSyncer(store, p, testSyncConfig())
remote := fx.remote()
claimed, _, err := store.ClaimGitHubDebSyncLease(context.Background(), remote.Name, "replica-1", time.Duration(remote.MutableTTL)*time.Second, syncLeaseDuration, time.Time{})
claimed, _, err := store.ClaimGitHubDebSyncLease(context.Background(), remote.Name, "replica-1", time.Duration(remote.MutableTTL)*time.Second, syncLeaseDuration, 0, 0)
if err != nil || !claimed {
t.Fatalf("replica-1 claim: claimed=%v err=%v", claimed, err)
}
@@ -368,51 +370,36 @@ func TestSyncerFailedAssetRetry(t *testing.T) {
}
}
// A periodic check claims with sync_interval as its freshness when set, and
// falls back to mutable_ttl when unset.
// A periodic check claims with sync_interval as its freshness and passes the
// remote's slot schedule; unset, it falls back to mutable_ttl with no schedule.
func TestSyncerHonoursSyncInterval(t *testing.T) {
for _, tc := range []struct {
interval int
want time.Duration
}{
{0, 90 * time.Second},
{7200, 2 * time.Hour},
} {
for _, interval := range []int{0, 7200} {
store := newFakeSyncStore()
s := newSyncer(store, newTestProvider(), testSyncConfig())
store.lastSynced["sched"] = time.Now()
remote := models.Remote{Name: "sched", MutableTTL: 90, SyncInterval: tc.interval}
remote := models.Remote{Name: "sched", MutableTTL: 90, SyncInterval: interval}
wantInterval, wantPhase := remote.SyncSchedule()
s.process(context.Background(), syncJob{remote: remote})
if store.freshness != tc.want {
t.Errorf("sync_interval=%d: claim freshness = %v, want %v", tc.interval, store.freshness, tc.want)
if want := remote.SyncPeriod(0); store.freshness != want || store.interval != wantInterval || store.phase != wantPhase {
t.Errorf("sync_interval=%d: claim freshness/interval/phase = %v/%v/%v, want %v/%v/%v",
interval, store.freshness, store.interval, store.phase, want, wantInterval, wantPhase)
}
}
}
// A scheduled remote's first sync waits for its first slot, and create-time
// primes are skipped, so remotes created together spread out; a cold-cache
// prime still claims at once.
func TestSyncerFirstSyncWaitsForSlot(t *testing.T) {
// Create-time primes are skipped for a scheduled remote so remotes created
// together spread out; a cold-cache prime still claims with no schedule.
func TestSyncerScheduledPrime(t *testing.T) {
store := newFakeSyncStore()
s := newSyncer(store, newTestProvider(), testSyncConfig())
remote := models.Remote{Name: "fresh", SyncInterval: 3600, CreatedAt: time.Now()}
remote := models.Remote{Name: "fresh", SyncInterval: 3600}
s.process(context.Background(), syncJob{remote: remote})
if store.attempts != 0 {
t.Fatal("scheduled first sync claimed before its first slot")
}
s.EnqueuePrime(remote)
if len(s.jobs) != 0 {
t.Fatal("create-time prime enqueued for a sync_interval remote")
}
s.process(context.Background(), syncJob{remote: remote, prime: true})
if store.attempts != 1 {
t.Fatal("cold-cache prime did not claim")
}
remote.CreatedAt = time.Now().Add(-2 * time.Hour)
s.process(context.Background(), syncJob{remote: remote})
if store.attempts != 2 {
t.Fatal("remote created before its slot did not claim")
if store.attempts != 1 || store.freshness != 0 || store.interval != 0 {
t.Fatalf("cold-cache prime: attempts=%d freshness=%v interval=%v, want 1/0/0", store.attempts, store.freshness, store.interval)
}
}
+4 -7
View File
@@ -34,7 +34,7 @@ const (
type SyncStore interface {
provider.RemoteMetadataStore
ListGitHubRPMRemotes(ctx context.Context) ([]models.Remote, error)
ClaimGitHubSyncLease(ctx context.Context, remoteName, owner string, freshness, lease time.Duration, slot time.Time) (claimed bool, etag string, err error)
ClaimGitHubSyncLease(ctx context.Context, remoteName, owner string, freshness, lease, interval, phase time.Duration) (claimed bool, etag string, err error)
ReleaseGitHubSyncLease(ctx context.Context, remoteName, owner string, res provider.SyncResult) error
}
@@ -209,17 +209,14 @@ func (s *Syncer) process(ctx context.Context, job syncJob) {
ttl := job.remote.SyncPeriod(defaultSyncFreshness)
freshness := ttl
var slot time.Time
var interval, phase time.Duration
if job.prime {
freshness = 0 // prime ignores the recency gate but still respects a live lease
} else {
slot = job.remote.SyncSlot(time.Now())
if !slot.IsZero() && !job.remote.CreatedAt.Before(slot) {
return // a scheduled first sync waits for the remote's first slot
}
interval, phase = job.remote.SyncSchedule()
}
claimed, etag, err := s.store.ClaimGitHubSyncLease(ctx, job.remote.Name, s.owner, freshness, syncLeaseDuration, slot)
claimed, etag, err := s.store.ClaimGitHubSyncLease(ctx, job.remote.Name, s.owner, freshness, syncLeaseDuration, interval, phase)
if err != nil {
slog.Error("github_rpm syncer: claim lease", "remote", job.remote.Name, "error", err)
return
+20 -33
View File
@@ -30,6 +30,8 @@ type fakeSyncStore struct {
retryAt map[string]time.Time
lastResult provider.SyncResult
freshness time.Duration
interval time.Duration
phase time.Duration
attempts int
}
@@ -50,15 +52,15 @@ func (f *fakeSyncStore) ListGitHubRPMRemotes(_ context.Context) ([]models.Remote
return append([]models.Remote(nil), f.remotes...), nil
}
func (f *fakeSyncStore) ClaimGitHubSyncLease(_ context.Context, name, owner string, freshness, lease time.Duration, slot time.Time) (bool, string, error) {
func (f *fakeSyncStore) ClaimGitHubSyncLease(_ context.Context, name, owner string, freshness, lease, interval, phase time.Duration) (bool, string, error) {
f.mu.Lock()
defer f.mu.Unlock()
f.freshness = freshness
f.freshness, f.interval, f.phase = freshness, interval, phase
f.attempts++
now := time.Now()
ls, hasLS := f.lastSynced[name]
exp, hasExp := f.leaseExp[name]
freshOK := !hasLS || (slot.IsZero() && now.Sub(ls) >= freshness) || (!slot.IsZero() && ls.Before(slot))
freshOK := !hasLS || now.Sub(ls) >= freshness
if ra, pending := f.retryAt[name]; pending {
freshOK = !now.Before(ra)
}
@@ -234,7 +236,7 @@ func TestSyncerLeasePreventsSecondReplica(t *testing.T) {
remote := fx.remote()
// Replica 1 holds the lease.
claimed, _, err := store.ClaimGitHubSyncLease(context.Background(), remote.Name, "replica-1", time.Duration(remote.MutableTTL)*time.Second, syncLeaseDuration, time.Time{})
claimed, _, err := store.ClaimGitHubSyncLease(context.Background(), remote.Name, "replica-1", time.Duration(remote.MutableTTL)*time.Second, syncLeaseDuration, 0, 0)
if err != nil || !claimed {
t.Fatalf("replica-1 claim: claimed=%v err=%v", claimed, err)
}
@@ -475,51 +477,36 @@ func TestSyncerFailedAssetRetry(t *testing.T) {
}
}
// A periodic check claims with sync_interval as its freshness when set, and
// falls back to mutable_ttl when unset.
// A periodic check claims with sync_interval as its freshness and passes the
// remote's slot schedule; unset, it falls back to mutable_ttl with no schedule.
func TestSyncerHonoursSyncInterval(t *testing.T) {
for _, tc := range []struct {
interval int
want time.Duration
}{
{0, 90 * time.Second},
{7200, 2 * time.Hour},
} {
for _, interval := range []int{0, 7200} {
store := newFakeSyncStore()
s := newSyncer(store, newTestProvider(), testSyncConfig())
store.lastSynced["sched"] = time.Now()
remote := models.Remote{Name: "sched", MutableTTL: 90, SyncInterval: tc.interval}
remote := models.Remote{Name: "sched", MutableTTL: 90, SyncInterval: interval}
wantInterval, wantPhase := remote.SyncSchedule()
s.process(context.Background(), syncJob{remote: remote})
if store.freshness != tc.want {
t.Errorf("sync_interval=%d: claim freshness = %v, want %v", tc.interval, store.freshness, tc.want)
if want := remote.SyncPeriod(0); store.freshness != want || store.interval != wantInterval || store.phase != wantPhase {
t.Errorf("sync_interval=%d: claim freshness/interval/phase = %v/%v/%v, want %v/%v/%v",
interval, store.freshness, store.interval, store.phase, want, wantInterval, wantPhase)
}
}
}
// A scheduled remote's first sync waits for its first slot, and create-time
// primes are skipped, so remotes created together spread out; a cold-cache
// prime still claims at once.
func TestSyncerFirstSyncWaitsForSlot(t *testing.T) {
// Create-time primes are skipped for a scheduled remote so remotes created
// together spread out; a cold-cache prime still claims with no schedule.
func TestSyncerScheduledPrime(t *testing.T) {
store := newFakeSyncStore()
s := newSyncer(store, newTestProvider(), testSyncConfig())
remote := models.Remote{Name: "fresh", SyncInterval: 3600, CreatedAt: time.Now()}
remote := models.Remote{Name: "fresh", SyncInterval: 3600}
s.process(context.Background(), syncJob{remote: remote})
if store.attempts != 0 {
t.Fatal("scheduled first sync claimed before its first slot")
}
s.EnqueuePrime(remote)
if len(s.jobs) != 0 {
t.Fatal("create-time prime enqueued for a sync_interval remote")
}
s.process(context.Background(), syncJob{remote: remote, prime: true})
if store.attempts != 1 {
t.Fatal("cold-cache prime did not claim")
}
remote.CreatedAt = time.Now().Add(-2 * time.Hour)
s.process(context.Background(), syncJob{remote: remote})
if store.attempts != 2 {
t.Fatal("remote created before its slot did not claim")
if store.attempts != 1 || store.freshness != 0 || store.interval != 0 {
t.Fatalf("cold-cache prime: attempts=%d freshness=%v interval=%v, want 1/0/0", store.attempts, store.freshness, store.interval)
}
}
+7 -8
View File
@@ -165,16 +165,15 @@ func (r Remote) SyncPeriod(def time.Duration) time.Duration {
return def
}
// SyncSlot is the start of the sync_interval slot containing now; zero when
// sync_interval is unset. Slots are offset by a stable per-remote phase so
// remotes sharing an interval spread across it, and every replica agrees.
func (r Remote) SyncSlot(now time.Time) time.Time {
// SyncSchedule is the sync_interval and the stable per-remote phase offsetting
// its slots, so remotes sharing an interval spread across it; both zero when
// sync_interval is unset.
func (r Remote) SyncSchedule() (interval, phase time.Duration) {
if r.SyncInterval <= 0 {
return time.Time{}
return 0, 0
}
interval := time.Duration(r.SyncInterval) * time.Second
phase := syncPhase(r.Name, interval)
return now.Add(-phase).Truncate(interval).Add(phase)
interval = time.Duration(r.SyncInterval) * time.Second
return interval, syncPhase(r.Name, interval)
}
func syncPhase(name string, period time.Duration) time.Duration {
+9 -52
View File
@@ -163,10 +163,9 @@ func TestValidateSyncInterval(t *testing.T) {
}
}
func TestSyncSlot(t *testing.T) {
now := time.Date(2026, 3, 1, 10, 17, 42, 0, time.UTC)
if s := (Remote{Name: "x", MutableTTL: 60}).SyncSlot(now); !s.IsZero() {
t.Fatalf("unset sync_interval slot = %v, want zero", s)
func TestSyncSchedule(t *testing.T) {
if i, p := (Remote{Name: "x", MutableTTL: 60}).SyncSchedule(); i != 0 || p != 0 {
t.Fatalf("unset sync_interval schedule = %v/%v, want 0/0", i, p)
}
const period = 5 * time.Minute
@@ -174,17 +173,14 @@ func TestSyncSlot(t *testing.T) {
var lo, hi time.Duration = period, 0
for i := 0; i < 13; i++ {
r := Remote{Name: fmt.Sprintf("github-rpm-%02d", i), SyncInterval: int(period.Seconds())}
s := r.SyncSlot(now)
if s.After(now) || !now.Before(s.Add(period)) {
t.Fatalf("%s: slot %v does not contain %v", r.Name, s, now)
interval, p := r.SyncSchedule()
if interval != period || p < 0 || p >= period {
t.Fatalf("%s: schedule %v/%v outside [0, %v)", r.Name, interval, p, period)
}
if r.SyncSlot(now.Add(-time.Hour)).Add(time.Hour) != s {
t.Fatalf("%s: phase not stable over time", r.Name)
}
p := s.Sub(now.Truncate(period)).Truncate(time.Second)
if p < 0 {
p += period
if _, again := r.SyncSchedule(); again != p {
t.Fatalf("%s: phase not stable", r.Name)
}
p = p.Truncate(time.Second)
phases[p] = true
lo, hi = min(lo, p), max(hi, p)
}
@@ -192,42 +188,3 @@ func TestSyncSlot(t *testing.T) {
t.Fatalf("phases bunched: %d distinct, spread %v of %v", len(phases), hi-lo, period)
}
}
// After a sync_interval change the next sync lands within one new interval, no
// gap exceeds an interval, and syncs settle at one per new slot.
func TestSyncSlotIntervalChange(t *testing.T) {
const tick = time.Minute
start := time.Date(2026, 3, 1, 0, 0, 0, 0, time.UTC)
for _, c := range []struct{ from, to int }{{300, 3600}, {3600, 300}, {3600, 5400}} {
for i := 0; i < 20; i++ {
r := Remote{Name: fmt.Sprintf("chg-%d", i), SyncInterval: c.from}
change := start.Add(6*time.Hour + time.Duration(i)*7*time.Minute)
var last time.Time
for now := start; now.Before(change); now = now.Add(tick) {
if last.IsZero() || last.Before(r.SyncSlot(now)) {
last = now
}
}
r.SyncInterval = c.to
next := change
for !last.Before(r.SyncSlot(next)) {
next = next.Add(tick)
}
if wait := next.Sub(change); wait > time.Duration(c.to)*time.Second+tick {
t.Errorf("%s %v: next sync %v after change exceeds the new interval", r.Name, c, wait)
}
if gap := next.Sub(last); gap > time.Duration(max(c.from, c.to))*time.Second+tick {
t.Errorf("%s %v: gap %v exceeds an interval", r.Name, c, gap)
}
syncs, end := 0, change.Add(6*time.Hour)
for now := change; now.Before(end); now = now.Add(tick) {
if last.Before(r.SyncSlot(now)) {
last, syncs = now, syncs+1
}
}
if limit := int(6*time.Hour/(time.Duration(c.to)*time.Second)) + 1; syncs > limit {
t.Errorf("%s %v: %d syncs in 6h after change, want <= %d", r.Name, c, syncs, limit)
}
}
}
}