Confine phased sync slots to sync_interval remotes
ci/woodpecker/pr/pre-commit Pipeline was successful
ci/woodpecker/pr/test Pipeline was successful
ci/woodpecker/pr/build Pipeline was successful

Unset remotes keep the mutable_ttl freshness predicate. Scheduled remotes
compute their slot in Go, skip the create-time prime and wait for their
first slot; a cold-cache prime still runs at once. Document sync_interval.
This commit is contained in:
2026-10-25 13:27:43 +11:00
parent 707e380c98
commit 1582533e5a
16 changed files with 343 additions and 135 deletions
+1 -1
View File
@@ -202,7 +202,7 @@ additive and idempotent for the same reason; a test enforces it.
| `GITHUB_SYNC_RATE` | `1` | `github_rpm` syncer global GitHub request rate (req/s), shared across all remotes. `1`/s = 3600/hr, under an authenticated token's ~5000/hr; unauthenticated (~60/hr) relies on ETag/304 |
| `GITHUB_SYNC_BURST` | `5` | Token-bucket burst for the shared limiter |
| `GITHUB_SYNC_WORKERS` | `3` | Concurrent `github_rpm` scan workers |
| `GITHUB_SYNC_POLL_INTERVAL` | `60` | Base scheduler tick in seconds; per-remote cadence is its `mutable_ttl`, enforced by the DB lease |
| `GITHUB_SYNC_POLL_INTERVAL` | `60` | Base scheduler tick in seconds; per-remote cadence is its `sync_interval` (phase-spread), else `mutable_ttl`, enforced by the DB lease |
| `GITHUB_TOKEN` | | Server-level GitHub PAT (fine-grained or classic), sent as `Authorization: Bearer`. Applies to every GitHub request; per-remote creds override it. See [GitHub authentication](#github-authentication) |
| `GITHUB_APP_ID` | | GitHub App id (App auth mode; mutually exclusive with `GITHUB_TOKEN`) |
| `GITHUB_APP_INSTALLATION_ID` | | GitHub App installation id |
+7 -3
View File
@@ -32,8 +32,12 @@ derived metadata current off the client request path:
The first `dnf` request is served from cache. If a request arrives before the
prime lands, it returns a retryable `503` (with `Retry-After`) rather than
serving an empty repo or blocking on a multi-minute derive.
- **Periodic re-check, driven by `mutable_ttl`.** Each remote is re-checked for
new or changed releases no more often than its `mutable_ttl`. New/changed
- **Periodic re-check, driven by `sync_interval` or `mutable_ttl`.** With
`sync_interval` (seconds) set, a remote re-checks once per interval at a
stable per-remote offset, so remotes on the same interval spread out instead
of syncing together; its first scheduled sync waits for that slot (a
cold-cache request still primes at once). Unset, it re-checks no more often
than its `mutable_ttl`. New/changed
assets are derived incrementally; assets already cached are never re-fetched,
and assets that disappear upstream are pruned.
- **ETag / 304 conditional requests.** The releases-list `ETag` is stored per
@@ -50,7 +54,7 @@ derived metadata current off the client request path:
periodic scan a replica must atomically claim a per-remote lease
(`github_rpm_sync_state`: `last_synced_at`, `etag`, `sync_lease_owner`,
`sync_lease_expires`); only the winner scans. This bounds total GitHub load to
~once per `mutable_ttl` regardless of replica count, and the shared `etag`
~once per `sync_interval` (else `mutable_ttl`) regardless of replica count, and the shared `etag`
lets any replica issue the conditional request.
```hcl
+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) (bool, string, error) {
return db.claimSyncLease(ctx, "github_alpine_sync_state", remoteName, owner, freshness, lease)
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)
}
// 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) (bool, string, error) {
return db.claimSyncLease(ctx, "github_deb_sync_state", remoteName, owner, freshness, lease)
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)
}
// 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)
claimed, etag, err := testDB.ClaimGitHubDebSyncLease(ctx(), name, "replica-1", freshness, lease, time.Time{})
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)
claimed2, _, err := testDB.ClaimGitHubDebSyncLease(ctx(), name, "replica-2", freshness, lease, time.Time{})
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)
claimed3, _, err := testDB.ClaimGitHubDebSyncLease(ctx(), name, "replica-2", freshness, lease, time.Time{})
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)
claimed4, etag4, err := testDB.ClaimGitHubDebSyncLease(ctx(), name, "replica-2", 0, lease, time.Time{})
if err != nil || !claimed4 {
t.Fatalf("prime claim: claimed=%v err=%v", claimed4, err)
}
+13 -23
View File
@@ -3,7 +3,6 @@ package database
import (
"context"
"errors"
"hash/fnv"
"time"
"github.com/jackc/pgx/v5"
@@ -35,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) (bool, string, error) {
return db.claimSyncLease(ctx, "github_rpm_sync_state", remoteName, owner, freshness, lease)
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)
}
// ReleaseGitHubSyncLease records a github_rpm scan outcome and frees the lease.
@@ -48,14 +47,17 @@ func (db *DB) ReleaseGitHubSyncLease(ctx context.Context, remoteName, owner stri
// 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 not
// since its latest slot boundary. Boundaries recur every freshness, offset by a
// stable per-remote phase so remotes sharing a TTL are spread across it rather
// than due together. A zero freshness (prime scans) ignores the recency gate
// but still honours a pending retry's backoff. The returned etag is
// 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
// can short-circuit an unchanged repo.
func (db *DB) claimSyncLease(ctx context.Context, table, remoteName, owner string, freshness, lease time.Duration) (bool, string, error) {
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
}
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))
@@ -63,11 +65,10 @@ func (db *DB) claimSyncLease(ctx context.Context, table, remoteName, owner strin
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 < CASE WHEN $3 <= 0 THEN now()
ELSE to_timestamp(floor((extract(epoch from now())::float8 - $5) / NULLIF($3, 0)) * $3 + $5) END END)
ELSE s.last_synced_at IS NULL OR s.last_synced_at < COALESCE($5::timestamptz, 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(), syncPhase(remoteName, freshness).Seconds())
`, remoteName, owner, freshness.Seconds(), lease.Seconds(), slotArg)
var etag string
if err := row.Scan(&etag); err != nil {
@@ -79,17 +80,6 @@ func (db *DB) claimSyncLease(ctx context.Context, table, remoteName, owner strin
return true, etag, nil
}
// syncPhase is remoteName's stable offset within period, so every replica
// agrees on when each remote is due.
func syncPhase(remoteName string, period time.Duration) time.Duration {
if period <= 0 {
return 0
}
h := fnv.New64a()
_, _ = h.Write([]byte(remoteName))
return time.Duration(h.Sum64() % uint64(period))
}
// 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
+92 -82
View File
@@ -1,7 +1,7 @@
package database
import (
"fmt"
"context"
"testing"
"time"
@@ -31,7 +31,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)
claimed, etag, err := testDB.ClaimGitHubSyncLease(ctx(), name, "replica-1", freshness, lease, time.Time{})
if err != nil || !claimed {
t.Fatalf("replica-1 first claim: claimed=%v err=%v", claimed, err)
}
@@ -40,7 +40,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)
claimed2, _, err := testDB.ClaimGitHubSyncLease(ctx(), name, "replica-2", freshness, lease, time.Time{})
if err != nil {
t.Fatalf("replica-2 claim err: %v", err)
}
@@ -54,7 +54,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)
claimed3, _, err := testDB.ClaimGitHubSyncLease(ctx(), name, "replica-2", freshness, lease, time.Time{})
if err != nil {
t.Fatalf("replica-2 recency claim err: %v", err)
}
@@ -63,7 +63,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)
claimed4, etag4, err := testDB.ClaimGitHubSyncLease(ctx(), name, "replica-2", 0, lease, time.Time{})
if err != nil || !claimed4 {
t.Fatalf("prime claim: claimed=%v err=%v", claimed4, err)
}
@@ -109,7 +109,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)
ok, etag, err := testDB.ClaimGitHubSyncLease(ctx(), name, "r1", f, lease, time.Time{})
if err != nil {
t.Fatalf("claim: %v", err)
}
@@ -221,21 +221,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)
return testDB.ClaimGitHubSyncLease(ctx(), n, "r1", time.Hour, time.Minute, time.Time{})
},
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)
return testDB.ClaimGitHubDebSyncLease(ctx(), n, "r1", time.Hour, time.Minute, time.Time{})
},
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)
return testDB.ClaimGitHubAlpineSyncLease(ctx(), n, "r1", time.Hour, time.Minute, time.Time{})
},
func(n string, res provider.SyncResult) error {
return testDB.ReleaseGitHubAlpineSyncLease(ctx(), n, "r1", res)
@@ -278,85 +278,95 @@ func TestGitHubSyncStateDeletedWithRemote(t *testing.T) {
}
}
func TestSyncPhaseStableAndSpread(t *testing.T) {
const period = 5 * time.Minute
phases := map[time.Duration]bool{}
var lo, hi time.Duration = period, 0
for i := 0; i < 13; i++ {
name := fmt.Sprintf("github-rpm-%02d", i)
p := syncPhase(name, period)
if p != syncPhase(name, period) {
t.Fatalf("%s: phase not stable", name)
type syncTable struct {
table string
pt models.PackageType
claim func(name, owner string, freshness time.Duration, slot time.Time) (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)
return ok, err
}
if p < 0 || p >= period {
t.Fatalf("%s: phase %v outside [0,%v)", name, p, period)
}
phases[p.Truncate(time.Second)] = true
lo, hi = min(lo, p), max(hi, p)
}
if len(phases) != 13 {
t.Fatalf("phases collide: %d distinct of 13", len(phases))
}
if hi-lo < period/2 {
t.Fatalf("phases bunched in %v of %v", hi-lo, period)
}
if syncPhase("x", 0) != 0 {
t.Fatal("zero period must give zero phase")
return []syncTable{
{"github_rpm_sync_state", models.PackageGitHubRPM, wrap(testDB.ClaimGitHubSyncLease)},
{"github_deb_sync_state", models.PackageGitHubDeb, wrap(testDB.ClaimGitHubDebSyncLease)},
{"github_alpine_sync_state", models.PackageGitHubAlpine, wrap(testDB.ClaimGitHubAlpineSyncLease)},
}
}
// TestGitHubSyncLeasePhasedSlots checks a remote is due once per slot: synced
// before its latest boundary (or over a TTL ago) it is claimable by exactly one
// replica; synced after the boundary it is not.
func TestGitHubSyncLeasePhasedSlots(t *testing.T) {
// 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 {
t.Helper()
name := prefix + "-" + tc.table + "-" + time.Now().Format("150405.000000")
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)
}
return name
}
func mustClaim(t *testing.T, tc syncTable, name, owner string, freshness time.Duration, slot time.Time) bool {
t.Helper()
ok, err := tc.claim(name, owner, freshness, slot)
if err != nil {
t.Fatalf("claim: %v", err)
}
return ok
}
// With no slot the claim keeps the freshness-since-last-sync predicate.
func TestSyncLeaseUnsetIntervalFreshness(t *testing.T) {
requireDB(t)
const lease = 15 * time.Minute
ttl := time.Hour
setSynced := func(name string, at time.Time) {
t.Helper()
if _, err := testDB.Pool.Exec(ctx(), `UPDATE github_rpm_sync_state SET last_synced_at = $2, sync_lease_owner = '', sync_lease_expires = NULL WHERE remote_name = $1`, name, at); err != nil {
t.Fatalf("set last_synced_at: %v", err)
}
}
claim := func(name, owner string) bool {
t.Helper()
ok, _, err := testDB.ClaimGitHubSyncLease(ctx(), name, owner, ttl, lease)
if err != nil {
t.Fatalf("claim: %v", err)
}
return ok
}
for i := 0; i < 5; i++ {
name := fmt.Sprintf("gh-slot-%d-%s", i, time.Now().Format("150405.000000"))
seedGitHubRPMRemote(t, name)
if !claim(name, "replica-1") {
t.Fatalf("%s: first claim failed", name)
}
now := time.Now()
phase := syncPhase(name, ttl)
boundary := now.Add(-phase).Truncate(ttl).Add(phase)
setSynced(name, now.Add(-ttl-time.Second))
if !claim(name, "replica-1") {
t.Fatalf("%s: not due after a full TTL", name)
}
if claim(name, "replica-2") {
t.Fatalf("%s: second replica claimed a held lease", name)
}
setSynced(name, boundary.Add(-5*time.Second))
if !claim(name, "replica-2") {
t.Fatalf("%s: not due after crossing its boundary %v", name, boundary)
}
if now.Sub(boundary) > 10*time.Second {
setSynced(name, boundary.Add(5*time.Second))
if claim(name, "replica-1") {
t.Fatalf("%s: re-synced within the same slot", name)
const ttl = time.Hour
for _, tc := range syncTables() {
t.Run(tc.table, func(t *testing.T) {
for _, c := range []struct {
ago time.Duration
freshness time.Duration
due bool
}{
{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 {
t.Errorf("synced %v ago, freshness %v: claimed=%v, want %v", c.ago, c.freshness, got, c.due)
}
}
}
})
}
}
// With a slot the remote is due only when last synced before it, 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)
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")
}
})
}
}
+11 -3
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) (claimed bool, etag string, err error)
ClaimGitHubAlpineSyncLease(ctx context.Context, remoteName, owner string, freshness, lease time.Duration, slot time.Time) (claimed bool, etag string, err error)
ReleaseGitHubAlpineSyncLease(ctx context.Context, remoteName, owner string, res provider.SyncResult) error
}
@@ -143,8 +143,10 @@ func (s *Syncer) schedule(ctx context.Context) {
}
// EnqueuePrime queues an immediate background prime for a freshly created remote.
// A sync_interval remote skips it and syncs in its first slot, so remotes
// created together do not burst.
func (s *Syncer) EnqueuePrime(remote models.Remote) {
if s == nil {
if s == nil || remote.SyncInterval > 0 {
return
}
s.enqueue(remote, true)
@@ -192,11 +194,17 @@ func (s *Syncer) process(ctx context.Context, job syncJob) {
ttl := job.remote.SyncPeriod(defaultSyncFreshness)
freshness := ttl
var slot time.Time
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
}
}
claimed, etag, err := s.store.ClaimGitHubAlpineSyncLease(ctx, job.remote.Name, s.owner, freshness, syncLeaseDuration)
claimed, etag, err := s.store.ClaimGitHubAlpineSyncLease(ctx, job.remote.Name, s.owner, freshness, syncLeaseDuration, slot)
if err != nil {
slog.Error("github_alpine syncer: claim lease", "remote", job.remote.Name, "error", err)
return
+33 -3
View File
@@ -29,6 +29,7 @@ type fakeSyncStore struct {
etags map[string]string
retryAt map[string]time.Time
freshness time.Duration
attempts int
}
func newFakeSyncStore() *fakeSyncStore {
@@ -48,14 +49,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) (bool, string, error) {
func (f *fakeSyncStore) ClaimGitHubAlpineSyncLease(_ context.Context, name, owner string, freshness, lease time.Duration, slot time.Time) (bool, string, error) {
f.mu.Lock()
defer f.mu.Unlock()
f.freshness = freshness
f.attempts++
now := time.Now()
ls, hasLS := f.lastSynced[name]
exp, hasExp := f.leaseExp[name]
freshOK := !hasLS || now.Sub(ls) >= freshness
freshOK := !hasLS || (slot.IsZero() && now.Sub(ls) >= freshness) || (!slot.IsZero() && ls.Before(slot))
if ra, pending := f.retryAt[name]; pending {
freshOK = !now.Before(ra)
}
@@ -225,7 +227,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)
claimed, _, err := store.ClaimGitHubAlpineSyncLease(context.Background(), remote.Name, "replica-1", time.Duration(remote.MutableTTL)*time.Second, syncLeaseDuration, time.Time{})
if err != nil || !claimed {
t.Fatalf("replica-1 claim: claimed=%v err=%v", claimed, err)
}
@@ -386,3 +388,31 @@ func TestSyncerHonoursSyncInterval(t *testing.T) {
}
}
}
// 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) {
store := newFakeSyncStore()
s := newSyncer(store, newTestProvider(), testSyncConfig())
remote := models.Remote{Name: "fresh", SyncInterval: 3600, CreatedAt: time.Now()}
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")
}
}
+11 -3
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) (claimed bool, etag string, err error)
ClaimGitHubDebSyncLease(ctx context.Context, remoteName, owner string, freshness, lease time.Duration, slot time.Time) (claimed bool, etag string, err error)
ReleaseGitHubDebSyncLease(ctx context.Context, remoteName, owner string, res provider.SyncResult) error
}
@@ -143,8 +143,10 @@ func (s *Syncer) schedule(ctx context.Context) {
}
// EnqueuePrime queues an immediate background prime for a freshly created remote.
// A sync_interval remote skips it and syncs in its first slot, so remotes
// created together do not burst.
func (s *Syncer) EnqueuePrime(remote models.Remote) {
if s == nil {
if s == nil || remote.SyncInterval > 0 {
return
}
s.enqueue(remote, true)
@@ -192,11 +194,17 @@ func (s *Syncer) process(ctx context.Context, job syncJob) {
ttl := job.remote.SyncPeriod(defaultSyncFreshness)
freshness := ttl
var slot time.Time
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
}
}
claimed, etag, err := s.store.ClaimGitHubDebSyncLease(ctx, job.remote.Name, s.owner, freshness, syncLeaseDuration)
claimed, etag, err := s.store.ClaimGitHubDebSyncLease(ctx, job.remote.Name, s.owner, freshness, syncLeaseDuration, slot)
if err != nil {
slog.Error("github_deb syncer: claim lease", "remote", job.remote.Name, "error", err)
return
+33 -3
View File
@@ -29,6 +29,7 @@ type fakeSyncStore struct {
etags map[string]string
retryAt map[string]time.Time
freshness time.Duration
attempts int
}
func newFakeSyncStore() *fakeSyncStore {
@@ -48,14 +49,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) (bool, string, error) {
func (f *fakeSyncStore) ClaimGitHubDebSyncLease(_ context.Context, name, owner string, freshness, lease time.Duration, slot time.Time) (bool, string, error) {
f.mu.Lock()
defer f.mu.Unlock()
f.freshness = freshness
f.attempts++
now := time.Now()
ls, hasLS := f.lastSynced[name]
exp, hasExp := f.leaseExp[name]
freshOK := !hasLS || now.Sub(ls) >= freshness
freshOK := !hasLS || (slot.IsZero() && now.Sub(ls) >= freshness) || (!slot.IsZero() && ls.Before(slot))
if ra, pending := f.retryAt[name]; pending {
freshOK = !now.Before(ra)
}
@@ -225,7 +227,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)
claimed, _, err := store.ClaimGitHubDebSyncLease(context.Background(), remote.Name, "replica-1", time.Duration(remote.MutableTTL)*time.Second, syncLeaseDuration, time.Time{})
if err != nil || !claimed {
t.Fatalf("replica-1 claim: claimed=%v err=%v", claimed, err)
}
@@ -386,3 +388,31 @@ func TestSyncerHonoursSyncInterval(t *testing.T) {
}
}
}
// 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) {
store := newFakeSyncStore()
s := newSyncer(store, newTestProvider(), testSyncConfig())
remote := models.Remote{Name: "fresh", SyncInterval: 3600, CreatedAt: time.Now()}
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")
}
}
+11 -3
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) (claimed bool, etag string, err error)
ClaimGitHubSyncLease(ctx context.Context, remoteName, owner string, freshness, lease time.Duration, slot time.Time) (claimed bool, etag string, err error)
ReleaseGitHubSyncLease(ctx context.Context, remoteName, owner string, res provider.SyncResult) error
}
@@ -155,8 +155,10 @@ func (s *Syncer) schedule(ctx context.Context) {
// EnqueuePrime queues an immediate background prime for a freshly created
// remote so its metadata is derived without blocking the create call.
// A sync_interval remote skips it and syncs in its first slot, so remotes
// created together do not burst.
func (s *Syncer) EnqueuePrime(remote models.Remote) {
if s == nil {
if s == nil || remote.SyncInterval > 0 {
return
}
s.enqueue(remote, true)
@@ -207,11 +209,17 @@ func (s *Syncer) process(ctx context.Context, job syncJob) {
ttl := job.remote.SyncPeriod(defaultSyncFreshness)
freshness := ttl
var slot time.Time
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
}
}
claimed, etag, err := s.store.ClaimGitHubSyncLease(ctx, job.remote.Name, s.owner, freshness, syncLeaseDuration)
claimed, etag, err := s.store.ClaimGitHubSyncLease(ctx, job.remote.Name, s.owner, freshness, syncLeaseDuration, slot)
if err != nil {
slog.Error("github_rpm syncer: claim lease", "remote", job.remote.Name, "error", err)
return
+33 -3
View File
@@ -30,6 +30,7 @@ type fakeSyncStore struct {
retryAt map[string]time.Time
lastResult provider.SyncResult
freshness time.Duration
attempts int
}
func newFakeSyncStore() *fakeSyncStore {
@@ -49,14 +50,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) (bool, string, error) {
func (f *fakeSyncStore) ClaimGitHubSyncLease(_ context.Context, name, owner string, freshness, lease time.Duration, slot time.Time) (bool, string, error) {
f.mu.Lock()
defer f.mu.Unlock()
f.freshness = freshness
f.attempts++
now := time.Now()
ls, hasLS := f.lastSynced[name]
exp, hasExp := f.leaseExp[name]
freshOK := !hasLS || now.Sub(ls) >= freshness
freshOK := !hasLS || (slot.IsZero() && now.Sub(ls) >= freshness) || (!slot.IsZero() && ls.Before(slot))
if ra, pending := f.retryAt[name]; pending {
freshOK = !now.Before(ra)
}
@@ -232,7 +234,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)
claimed, _, err := store.ClaimGitHubSyncLease(context.Background(), remote.Name, "replica-1", time.Duration(remote.MutableTTL)*time.Second, syncLeaseDuration, time.Time{})
if err != nil || !claimed {
t.Fatalf("replica-1 claim: claimed=%v err=%v", claimed, err)
}
@@ -493,3 +495,31 @@ func TestSyncerHonoursSyncInterval(t *testing.T) {
}
}
}
// 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) {
store := newFakeSyncStore()
s := newSyncer(store, newTestProvider(), testSyncConfig())
remote := models.Remote{Name: "fresh", SyncInterval: 3600, CreatedAt: time.Now()}
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")
}
}
+19
View File
@@ -2,6 +2,7 @@ package models
import (
"fmt"
"hash/fnv"
"net/url"
"regexp"
"time"
@@ -164,6 +165,24 @@ 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 {
if r.SyncInterval <= 0 {
return time.Time{}
}
interval := time.Duration(r.SyncInterval) * time.Second
phase := syncPhase(r.Name, interval)
return now.Add(-phase).Truncate(interval).Add(phase)
}
func syncPhase(name string, period time.Duration) time.Duration {
h := fnv.New64a()
_, _ = h.Write([]byte(name))
return time.Duration(h.Sum64() % uint64(period))
}
// ValidateSyncInterval rejects a negative sync_interval, or one set on a type
// without a background syncer.
func (r *Remote) ValidateSyncInterval() error {
+70
View File
@@ -2,6 +2,7 @@ package models
import (
"encoding/json"
"fmt"
"strings"
"testing"
"time"
@@ -161,3 +162,72 @@ 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)
}
const period = 5 * time.Minute
phases := map[time.Duration]bool{}
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)
}
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
}
phases[p] = true
lo, hi = min(lo, p), max(hi, p)
}
if len(phases) != 13 || hi-lo < period/2 {
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)
}
}
}
}
+1
View File
@@ -8,6 +8,7 @@ export interface Remote {
immutable_ttl: number;
mutable_ttl: number;
check_mutable: boolean;
sync_interval?: number;
patterns: string[];
blocklist: string[];
mutable_patterns: string[];