Add per-remote sync_interval and spread GitHub syncs across it
This commit is contained in:
@@ -104,6 +104,10 @@ func (h *RemotesHandler) create(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
if err := remote.ValidateSyncInterval(); err != nil {
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
if err := h.db.CreateRemote(r.Context(), &remote); err != nil {
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
@@ -144,6 +148,10 @@ func (h *RemotesHandler) update(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
if err := remote.ValidateSyncInterval(); err != nil {
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
if err := h.db.UpdateRemote(r.Context(), &remote); err != nil {
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
|
||||
@@ -55,3 +55,48 @@ func TestMutableTTLOmittedVsZero(t *testing.T) {
|
||||
t.Errorf("explicit 0 stored as %d", got)
|
||||
}
|
||||
}
|
||||
|
||||
// sync_interval round-trips through the API and DB on github_* remotes and is
|
||||
// rejected when negative or on a type without a background syncer.
|
||||
func TestSyncIntervalAPI(t *testing.T) {
|
||||
if testDSN == "" {
|
||||
t.Skip("Docker unavailable")
|
||||
}
|
||||
db, err := database.New(testDSN)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
h := NewRemotesHandler(db, nil, nil).Routes()
|
||||
interval := func(name string) int {
|
||||
t.Helper()
|
||||
r, err := db.GetRemote(context.Background(), name)
|
||||
if err != nil {
|
||||
t.Fatalf("get %s: %v", name, err)
|
||||
}
|
||||
return r.SyncInterval
|
||||
}
|
||||
const gh = `"package_type":"github_deb","base_url":"https://api.github.com/repos/acme/tools","releases_remote":"github"`
|
||||
|
||||
if c := do(t, h, "POST", "/", `{"name":"si-gh",`+gh+`}`); c != 201 {
|
||||
t.Fatalf("create unset = %d", c)
|
||||
}
|
||||
if got := interval("si-gh"); got != 0 {
|
||||
t.Fatalf("unset sync_interval stored as %d", got)
|
||||
}
|
||||
if c := do(t, h, "PUT", "/si-gh", `{`+gh+`,"sync_interval":3600}`); c != 200 {
|
||||
t.Fatalf("update = %d", c)
|
||||
}
|
||||
if got := interval("si-gh"); got != 3600 {
|
||||
t.Fatalf("sync_interval = %d, want 3600", got)
|
||||
}
|
||||
|
||||
for _, body := range []string{
|
||||
`{"name":"si-neg",` + gh + `,"sync_interval":-1}`,
|
||||
`{"name":"si-rpm","package_type":"rpm","base_url":"https://example.com/repo","sync_interval":60}`,
|
||||
} {
|
||||
if c := do(t, h, "POST", "/", body); c != 400 {
|
||||
t.Errorf("%s = %d, want 400", body, c)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@ package database
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"hash/fnv"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
@@ -47,9 +48,11 @@ 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 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
|
||||
// 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
|
||||
// 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) {
|
||||
@@ -60,10 +63,11 @@ 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 < now() - make_interval(secs => $3) END)
|
||||
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)
|
||||
AND (s.sync_lease_expires IS NULL OR s.sync_lease_expires < now())
|
||||
RETURNING s.etag
|
||||
`, remoteName, owner, freshness.Seconds(), lease.Seconds())
|
||||
`, remoteName, owner, freshness.Seconds(), lease.Seconds(), syncPhase(remoteName, freshness).Seconds())
|
||||
|
||||
var etag string
|
||||
if err := row.Scan(&etag); err != nil {
|
||||
@@ -75,6 +79,17 @@ 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
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -276,3 +277,86 @@ 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)
|
||||
}
|
||||
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")
|
||||
}
|
||||
}
|
||||
|
||||
// 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) {
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -9,7 +9,7 @@ import (
|
||||
)
|
||||
|
||||
const remoteCols = `name, package_type, repo_type, base_url, mirrorlist, mirror_strategy, description, username, password,
|
||||
immutable_ttl, mutable_ttl, check_mutable,
|
||||
immutable_ttl, mutable_ttl, check_mutable, sync_interval,
|
||||
patterns, blocklist, mutable_patterns, immutable_patterns,
|
||||
ban_tags_enabled, ban_tags,
|
||||
quarantine_enabled, quarantine_days, stale_on_error,
|
||||
@@ -29,7 +29,7 @@ func normalizeMirrorStrategy(s string) string {
|
||||
func scanRemote(scanner interface{ Scan(...any) error }, r *models.Remote) error {
|
||||
return scanner.Scan(
|
||||
&r.Name, &r.PackageType, &r.RepoType, &r.BaseURL, &r.Mirrorlist, &r.MirrorStrategy, &r.Description, &r.Username, &r.Password,
|
||||
&r.ImmutableTTL, &r.MutableTTL, &r.CheckMutable,
|
||||
&r.ImmutableTTL, &r.MutableTTL, &r.CheckMutable, &r.SyncInterval,
|
||||
&r.Patterns, &r.Blocklist, &r.MutablePatterns, &r.ImmutablePatterns,
|
||||
&r.BanTagsEnabled, &r.BanTags,
|
||||
&r.QuarantineEnabled, &r.QuarantineDays, &r.StaleOnError,
|
||||
@@ -76,8 +76,8 @@ func (db *DB) CreateRemote(ctx context.Context, r *models.Remote) error {
|
||||
quarantine_enabled, quarantine_days, stale_on_error,
|
||||
releases_remote, managed_by,
|
||||
upstream_dial_timeout, upstream_tls_timeout, upstream_response_header_timeout,
|
||||
mirror_strategy
|
||||
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26)
|
||||
mirror_strategy, sync_interval
|
||||
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27)
|
||||
`,
|
||||
r.Name, r.PackageType, r.RepoType, r.BaseURL, r.Mirrorlist, r.Description, r.Username, r.Password,
|
||||
r.ImmutableTTL, r.MutableTTL, r.CheckMutable,
|
||||
@@ -86,7 +86,7 @@ func (db *DB) CreateRemote(ctx context.Context, r *models.Remote) error {
|
||||
r.QuarantineEnabled, r.QuarantineDays, r.StaleOnError,
|
||||
r.ReleasesRemote, r.ManagedBy,
|
||||
r.UpstreamDialTimeout, r.UpstreamTLSTimeout, r.UpstreamResponseHeaderTimeout,
|
||||
normalizeMirrorStrategy(r.MirrorStrategy),
|
||||
normalizeMirrorStrategy(r.MirrorStrategy), r.SyncInterval,
|
||||
)
|
||||
return err
|
||||
}
|
||||
@@ -101,7 +101,7 @@ func (db *DB) UpdateRemote(ctx context.Context, r *models.Remote) error {
|
||||
quarantine_enabled=$17, quarantine_days=$18, stale_on_error=$19,
|
||||
releases_remote=$20, managed_by=$21,
|
||||
upstream_dial_timeout=$22, upstream_tls_timeout=$23, upstream_response_header_timeout=$24,
|
||||
mirror_strategy=$26,
|
||||
mirror_strategy=$26, sync_interval=$27,
|
||||
updated_at=NOW()
|
||||
WHERE name=$1
|
||||
`,
|
||||
@@ -113,7 +113,7 @@ func (db *DB) UpdateRemote(ctx context.Context, r *models.Remote) error {
|
||||
r.ReleasesRemote, r.ManagedBy,
|
||||
r.UpstreamDialTimeout, r.UpstreamTLSTimeout, r.UpstreamResponseHeaderTimeout,
|
||||
r.Mirrorlist,
|
||||
normalizeMirrorStrategy(r.MirrorStrategy),
|
||||
normalizeMirrorStrategy(r.MirrorStrategy), r.SyncInterval,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -129,7 +129,8 @@ func (s *Syncer) Run(ctx context.Context) {
|
||||
}
|
||||
|
||||
// schedule enqueues a periodic check for every github_alpine remote. The DB lease
|
||||
// enforces the per-remote mutable_ttl cadence and cross-replica coordination.
|
||||
// enforces the per-remote sync_interval (else mutable_ttl) cadence and
|
||||
// cross-replica coordination.
|
||||
func (s *Syncer) schedule(ctx context.Context) {
|
||||
remotes, err := s.store.ListGitHubAlpineRemotes(ctx)
|
||||
if err != nil {
|
||||
@@ -189,10 +190,7 @@ func (s *Syncer) process(ctx context.Context, job syncJob) {
|
||||
s.mu.Unlock()
|
||||
}()
|
||||
|
||||
ttl := time.Duration(job.remote.MutableTTL) * time.Second
|
||||
if ttl <= 0 {
|
||||
ttl = defaultSyncFreshness
|
||||
}
|
||||
ttl := job.remote.SyncPeriod(defaultSyncFreshness)
|
||||
freshness := ttl
|
||||
if job.prime {
|
||||
freshness = 0
|
||||
|
||||
@@ -28,6 +28,7 @@ type fakeSyncStore struct {
|
||||
lastSynced map[string]time.Time
|
||||
etags map[string]string
|
||||
retryAt map[string]time.Time
|
||||
freshness time.Duration
|
||||
}
|
||||
|
||||
func newFakeSyncStore() *fakeSyncStore {
|
||||
@@ -50,6 +51,7 @@ func (f *fakeSyncStore) ListGitHubAlpineRemotes(_ context.Context) ([]models.Rem
|
||||
func (f *fakeSyncStore) ClaimGitHubAlpineSyncLease(_ context.Context, name, owner string, freshness, lease time.Duration) (bool, string, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.freshness = freshness
|
||||
now := time.Now()
|
||||
ls, hasLS := f.lastSynced[name]
|
||||
exp, hasExp := f.leaseExp[name]
|
||||
@@ -363,3 +365,24 @@ 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.
|
||||
func TestSyncerHonoursSyncInterval(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
interval int
|
||||
want time.Duration
|
||||
}{
|
||||
{0, 90 * time.Second},
|
||||
{7200, 2 * time.Hour},
|
||||
} {
|
||||
store := newFakeSyncStore()
|
||||
s := newSyncer(store, newTestProvider(), testSyncConfig())
|
||||
store.lastSynced["sched"] = time.Now()
|
||||
remote := models.Remote{Name: "sched", MutableTTL: 90, SyncInterval: tc.interval}
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -129,7 +129,8 @@ func (s *Syncer) Run(ctx context.Context) {
|
||||
}
|
||||
|
||||
// schedule enqueues a periodic check for every github_deb remote. The DB lease
|
||||
// enforces the per-remote mutable_ttl cadence and cross-replica coordination.
|
||||
// enforces the per-remote sync_interval (else mutable_ttl) cadence and
|
||||
// cross-replica coordination.
|
||||
func (s *Syncer) schedule(ctx context.Context) {
|
||||
remotes, err := s.store.ListGitHubDebRemotes(ctx)
|
||||
if err != nil {
|
||||
@@ -189,10 +190,7 @@ func (s *Syncer) process(ctx context.Context, job syncJob) {
|
||||
s.mu.Unlock()
|
||||
}()
|
||||
|
||||
ttl := time.Duration(job.remote.MutableTTL) * time.Second
|
||||
if ttl <= 0 {
|
||||
ttl = defaultSyncFreshness
|
||||
}
|
||||
ttl := job.remote.SyncPeriod(defaultSyncFreshness)
|
||||
freshness := ttl
|
||||
if job.prime {
|
||||
freshness = 0
|
||||
|
||||
@@ -28,6 +28,7 @@ type fakeSyncStore struct {
|
||||
lastSynced map[string]time.Time
|
||||
etags map[string]string
|
||||
retryAt map[string]time.Time
|
||||
freshness time.Duration
|
||||
}
|
||||
|
||||
func newFakeSyncStore() *fakeSyncStore {
|
||||
@@ -50,6 +51,7 @@ func (f *fakeSyncStore) ListGitHubDebRemotes(_ context.Context) ([]models.Remote
|
||||
func (f *fakeSyncStore) ClaimGitHubDebSyncLease(_ context.Context, name, owner string, freshness, lease time.Duration) (bool, string, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.freshness = freshness
|
||||
now := time.Now()
|
||||
ls, hasLS := f.lastSynced[name]
|
||||
exp, hasExp := f.leaseExp[name]
|
||||
@@ -363,3 +365,24 @@ 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.
|
||||
func TestSyncerHonoursSyncInterval(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
interval int
|
||||
want time.Duration
|
||||
}{
|
||||
{0, 90 * time.Second},
|
||||
{7200, 2 * time.Hour},
|
||||
} {
|
||||
store := newFakeSyncStore()
|
||||
s := newSyncer(store, newTestProvider(), testSyncConfig())
|
||||
store.lastSynced["sched"] = time.Now()
|
||||
remote := models.Remote{Name: "sched", MutableTTL: 90, SyncInterval: tc.interval}
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -21,7 +21,7 @@ const (
|
||||
// scan never loses its lease, while a crashed replica's lease still expires.
|
||||
syncLeaseDuration = 15 * time.Minute
|
||||
// defaultSyncFreshness is the periodic re-check interval used when a remote's
|
||||
// mutable_ttl is unset.
|
||||
// sync_interval and mutable_ttl are unset.
|
||||
defaultSyncFreshness = 5 * time.Minute
|
||||
// jobQueueDepth bounds the pending work queue; enqueues past it are dropped
|
||||
// (a later poll re-enqueues), never blocking the caller.
|
||||
@@ -43,7 +43,7 @@ type SyncConfig struct {
|
||||
RatePerSec float64 // global GitHub request rate (req/s)
|
||||
Burst int // token-bucket burst
|
||||
Workers int // concurrent scan workers
|
||||
PollInterval time.Duration // base scheduler tick; per-remote cadence is mutable_ttl
|
||||
PollInterval time.Duration // base scheduler tick; per-remote cadence is sync_interval or mutable_ttl
|
||||
}
|
||||
|
||||
type syncJob struct {
|
||||
@@ -139,9 +139,9 @@ func (s *Syncer) Run(ctx context.Context) {
|
||||
}
|
||||
|
||||
// schedule enqueues a periodic check for every github_rpm remote. The DB lease
|
||||
// (claimed in the worker) enforces the per-remote mutable_ttl cadence and cross
|
||||
// replica coordination, so enqueuing every tick is cheap: a not-yet-due remote
|
||||
// simply fails to claim and is skipped.
|
||||
// (claimed in the worker) enforces the per-remote sync_interval (else
|
||||
// mutable_ttl) cadence and cross replica coordination, so enqueuing every tick
|
||||
// is cheap: a not-yet-due remote simply fails to claim and is skipped.
|
||||
func (s *Syncer) schedule(ctx context.Context) {
|
||||
remotes, err := s.store.ListGitHubRPMRemotes(ctx)
|
||||
if err != nil {
|
||||
@@ -205,10 +205,7 @@ func (s *Syncer) process(ctx context.Context, job syncJob) {
|
||||
s.mu.Unlock()
|
||||
}()
|
||||
|
||||
ttl := time.Duration(job.remote.MutableTTL) * time.Second
|
||||
if ttl <= 0 {
|
||||
ttl = defaultSyncFreshness
|
||||
}
|
||||
ttl := job.remote.SyncPeriod(defaultSyncFreshness)
|
||||
freshness := ttl
|
||||
if job.prime {
|
||||
freshness = 0 // prime ignores the recency gate but still respects a live lease
|
||||
|
||||
@@ -29,6 +29,7 @@ type fakeSyncStore struct {
|
||||
etags map[string]string
|
||||
retryAt map[string]time.Time
|
||||
lastResult provider.SyncResult
|
||||
freshness time.Duration
|
||||
}
|
||||
|
||||
func newFakeSyncStore() *fakeSyncStore {
|
||||
@@ -51,6 +52,7 @@ func (f *fakeSyncStore) ListGitHubRPMRemotes(_ context.Context) ([]models.Remote
|
||||
func (f *fakeSyncStore) ClaimGitHubSyncLease(_ context.Context, name, owner string, freshness, lease time.Duration) (bool, string, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.freshness = freshness
|
||||
now := time.Now()
|
||||
ls, hasLS := f.lastSynced[name]
|
||||
exp, hasExp := f.leaseExp[name]
|
||||
@@ -470,3 +472,24 @@ 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.
|
||||
func TestSyncerHonoursSyncInterval(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
interval int
|
||||
want time.Duration
|
||||
}{
|
||||
{0, 90 * time.Second},
|
||||
{7200, 2 * time.Hour},
|
||||
} {
|
||||
store := newFakeSyncStore()
|
||||
s := newSyncer(store, newTestProvider(), testSyncConfig())
|
||||
store.lastSynced["sched"] = time.Now()
|
||||
remote := models.Remote{Name: "sched", MutableTTL: 90, SyncInterval: tc.interval}
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
ALTER TABLE remotes ADD COLUMN IF NOT EXISTS sync_interval INTEGER NOT NULL DEFAULT 0;
|
||||
@@ -55,6 +55,9 @@ type Remote struct {
|
||||
ImmutableTTL int `json:"immutable_ttl"`
|
||||
MutableTTL int `json:"mutable_ttl"`
|
||||
CheckMutable bool `json:"check_mutable"`
|
||||
// SyncInterval (seconds) sets how often a github_* remote rescans releases
|
||||
// and rebuilds its index; 0 falls back to mutable_ttl.
|
||||
SyncInterval int `json:"sync_interval,omitempty"`
|
||||
|
||||
// Upstream HTTP timeouts in seconds. 0 means use the server default.
|
||||
UpstreamDialTimeout int `json:"upstream_dial_timeout,omitempty"`
|
||||
@@ -149,6 +152,34 @@ func (r *Remote) ValidateMirrorStrategy() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// SyncPeriod is the background sync cadence: sync_interval, else mutable_ttl,
|
||||
// else def.
|
||||
func (r Remote) SyncPeriod(def time.Duration) time.Duration {
|
||||
if r.SyncInterval > 0 {
|
||||
return time.Duration(r.SyncInterval) * time.Second
|
||||
}
|
||||
if r.MutableTTL > 0 {
|
||||
return time.Duration(r.MutableTTL) * time.Second
|
||||
}
|
||||
return def
|
||||
}
|
||||
|
||||
// ValidateSyncInterval rejects a negative sync_interval, or one set on a type
|
||||
// without a background syncer.
|
||||
func (r *Remote) ValidateSyncInterval() error {
|
||||
if r.SyncInterval < 0 {
|
||||
return fmt.Errorf("sync_interval must be >= 0, got %d", r.SyncInterval)
|
||||
}
|
||||
switch r.PackageType {
|
||||
case PackageGitHubRPM, PackageGitHubDeb, PackageGitHubAlpine:
|
||||
default:
|
||||
if r.SyncInterval > 0 {
|
||||
return fmt.Errorf("sync_interval is only valid on github_rpm, github_deb and github_alpine remotes")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ValidatePatterns ensures every configured regex compiles. Storing an
|
||||
// invalid pattern would otherwise be silently dropped at match time, which
|
||||
// for the blocklist is a fail-open: a mistyped deny rule becomes a no-op.
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestRemote_ValidatePatterns(t *testing.T) {
|
||||
@@ -123,3 +124,40 @@ func TestValidateMirrorlist(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSyncPeriod(t *testing.T) {
|
||||
const def = 5 * time.Minute
|
||||
for _, tc := range []struct {
|
||||
interval, ttl int
|
||||
want time.Duration
|
||||
}{
|
||||
{0, 0, def},
|
||||
{0, 600, 10 * time.Minute},
|
||||
{3600, 600, time.Hour},
|
||||
} {
|
||||
r := Remote{SyncInterval: tc.interval, MutableTTL: tc.ttl}
|
||||
if got := r.SyncPeriod(def); got != tc.want {
|
||||
t.Errorf("SyncPeriod(interval=%d, ttl=%d) = %v, want %v", tc.interval, tc.ttl, got, tc.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestValidateSyncInterval(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
pt PackageType
|
||||
interval int
|
||||
ok bool
|
||||
}{
|
||||
{PackageGitHubRPM, 3600, true},
|
||||
{PackageGitHubDeb, 3600, true},
|
||||
{PackageGitHubAlpine, 3600, true},
|
||||
{PackageGitHubRPM, -1, false},
|
||||
{PackageRPM, 0, true},
|
||||
{PackageRPM, 60, false},
|
||||
} {
|
||||
r := Remote{PackageType: tc.pt, SyncInterval: tc.interval}
|
||||
if err := r.ValidateSyncInterval(); (err == nil) != tc.ok {
|
||||
t.Errorf("%s interval=%d: err=%v, want ok=%v", tc.pt, tc.interval, err, tc.ok)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user