From 707e380c98141c472168cfc68b92244fda7f7cb1 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 25 Oct 2026 13:12:59 +1100 Subject: [PATCH 1/3] Add per-remote sync_interval and spread GitHub syncs across it --- internal/api/v2/remotes.go | 8 +++ internal/api/v2/remotes_ttl_test.go | 45 +++++++++++++ internal/database/github_sync.go | 25 +++++-- internal/database/github_sync_test.go | 84 ++++++++++++++++++++++++ internal/database/remotes.go | 14 ++-- internal/provider/alpine/syncer.go | 8 +-- internal/provider/alpine/syncer_test.go | 23 +++++++ internal/provider/deb/syncer.go | 8 +-- internal/provider/deb/syncer_test.go | 23 +++++++ internal/provider/rpm/syncer.go | 15 ++--- internal/provider/rpm/syncer_test.go | 23 +++++++ migrations/0009_remote_sync_interval.sql | 1 + pkg/models/remote.go | 31 +++++++++ pkg/models/remote_test.go | 38 +++++++++++ 14 files changed, 315 insertions(+), 31 deletions(-) create mode 100644 migrations/0009_remote_sync_interval.sql diff --git a/internal/api/v2/remotes.go b/internal/api/v2/remotes.go index b5c3a82..036c75a 100644 --- a/internal/api/v2/remotes.go +++ b/internal/api/v2/remotes.go @@ -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 diff --git a/internal/api/v2/remotes_ttl_test.go b/internal/api/v2/remotes_ttl_test.go index 7beb859..ff1eb84 100644 --- a/internal/api/v2/remotes_ttl_test.go +++ b/internal/api/v2/remotes_ttl_test.go @@ -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) + } + } +} diff --git a/internal/database/github_sync.go b/internal/database/github_sync.go index d3498fc..f509ed4 100644 --- a/internal/database/github_sync.go +++ b/internal/database/github_sync.go @@ -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 diff --git a/internal/database/github_sync_test.go b/internal/database/github_sync_test.go index d66dc07..684d667 100644 --- a/internal/database/github_sync_test.go +++ b/internal/database/github_sync_test.go @@ -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) + } + } + } +} diff --git a/internal/database/remotes.go b/internal/database/remotes.go index 44b2eb5..58b5376 100644 --- a/internal/database/remotes.go +++ b/internal/database/remotes.go @@ -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 } diff --git a/internal/provider/alpine/syncer.go b/internal/provider/alpine/syncer.go index d4f3fca..a578826 100644 --- a/internal/provider/alpine/syncer.go +++ b/internal/provider/alpine/syncer.go @@ -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 diff --git a/internal/provider/alpine/syncer_test.go b/internal/provider/alpine/syncer_test.go index 1a6ce96..de882c4 100644 --- a/internal/provider/alpine/syncer_test.go +++ b/internal/provider/alpine/syncer_test.go @@ -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) + } + } +} diff --git a/internal/provider/deb/syncer.go b/internal/provider/deb/syncer.go index b261134..bc34b1a 100644 --- a/internal/provider/deb/syncer.go +++ b/internal/provider/deb/syncer.go @@ -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 diff --git a/internal/provider/deb/syncer_test.go b/internal/provider/deb/syncer_test.go index b3919a5..cc4d0af 100644 --- a/internal/provider/deb/syncer_test.go +++ b/internal/provider/deb/syncer_test.go @@ -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) + } + } +} diff --git a/internal/provider/rpm/syncer.go b/internal/provider/rpm/syncer.go index ba7cd1f..a746e41 100644 --- a/internal/provider/rpm/syncer.go +++ b/internal/provider/rpm/syncer.go @@ -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 diff --git a/internal/provider/rpm/syncer_test.go b/internal/provider/rpm/syncer_test.go index be6d9c9..828a43c 100644 --- a/internal/provider/rpm/syncer_test.go +++ b/internal/provider/rpm/syncer_test.go @@ -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) + } + } +} diff --git a/migrations/0009_remote_sync_interval.sql b/migrations/0009_remote_sync_interval.sql new file mode 100644 index 0000000..39f7bf0 --- /dev/null +++ b/migrations/0009_remote_sync_interval.sql @@ -0,0 +1 @@ +ALTER TABLE remotes ADD COLUMN IF NOT EXISTS sync_interval INTEGER NOT NULL DEFAULT 0; diff --git a/pkg/models/remote.go b/pkg/models/remote.go index fd27aa6..ce171fa 100644 --- a/pkg/models/remote.go +++ b/pkg/models/remote.go @@ -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. diff --git a/pkg/models/remote_test.go b/pkg/models/remote_test.go index 7f757f4..b44071e 100644 --- a/pkg/models/remote_test.go +++ b/pkg/models/remote_test.go @@ -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) + } + } +} -- 2.47.3 From 1582533e5a770dfb48740b2cfaa44329a44027a4 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 25 Oct 2026 13:27:43 +1100 Subject: [PATCH 2/3] Confine phased sync slots to sync_interval remotes 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. --- README.md | 2 +- docs/rpm/remote-github.md | 10 +- internal/database/alpine_github_sync.go | 4 +- internal/database/deb_github_sync.go | 4 +- internal/database/deb_github_sync_test.go | 8 +- internal/database/github_sync.go | 36 ++--- internal/database/github_sync_test.go | 174 ++++++++++++---------- internal/provider/alpine/syncer.go | 14 +- internal/provider/alpine/syncer_test.go | 36 ++++- internal/provider/deb/syncer.go | 14 +- internal/provider/deb/syncer_test.go | 36 ++++- internal/provider/rpm/syncer.go | 14 +- internal/provider/rpm/syncer_test.go | 36 ++++- pkg/models/remote.go | 19 +++ pkg/models/remote_test.go | 70 +++++++++ ui/src/api/types.ts | 1 + 16 files changed, 343 insertions(+), 135 deletions(-) diff --git a/README.md b/README.md index 7ca3d60..f8d2a9b 100644 --- a/README.md +++ b/README.md @@ -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 | diff --git a/docs/rpm/remote-github.md b/docs/rpm/remote-github.md index f079b02..ccb9e69 100644 --- a/docs/rpm/remote-github.md +++ b/docs/rpm/remote-github.md @@ -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 diff --git a/internal/database/alpine_github_sync.go b/internal/database/alpine_github_sync.go index 1378e16..4d3835a 100644 --- a/internal/database/alpine_github_sync.go +++ b/internal/database/alpine_github_sync.go @@ -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 diff --git a/internal/database/deb_github_sync.go b/internal/database/deb_github_sync.go index 3a727fe..94f04db 100644 --- a/internal/database/deb_github_sync.go +++ b/internal/database/deb_github_sync.go @@ -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 diff --git a/internal/database/deb_github_sync_test.go b/internal/database/deb_github_sync_test.go index 1049d3a..a5ac83a 100644 --- a/internal/database/deb_github_sync_test.go +++ b/internal/database/deb_github_sync_test.go @@ -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) } diff --git a/internal/database/github_sync.go b/internal/database/github_sync.go index f509ed4..366ac4f 100644 --- a/internal/database/github_sync.go +++ b/internal/database/github_sync.go @@ -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 diff --git a/internal/database/github_sync_test.go b/internal/database/github_sync_test.go index 684d667..0b15ebf 100644 --- a/internal/database/github_sync_test.go +++ b/internal/database/github_sync_test.go @@ -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") + } + }) } } diff --git a/internal/provider/alpine/syncer.go b/internal/provider/alpine/syncer.go index a578826..1b005ed 100644 --- a/internal/provider/alpine/syncer.go +++ b/internal/provider/alpine/syncer.go @@ -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 diff --git a/internal/provider/alpine/syncer_test.go b/internal/provider/alpine/syncer_test.go index de882c4..e5eff31 100644 --- a/internal/provider/alpine/syncer_test.go +++ b/internal/provider/alpine/syncer_test.go @@ -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") + } +} diff --git a/internal/provider/deb/syncer.go b/internal/provider/deb/syncer.go index bc34b1a..7875174 100644 --- a/internal/provider/deb/syncer.go +++ b/internal/provider/deb/syncer.go @@ -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 diff --git a/internal/provider/deb/syncer_test.go b/internal/provider/deb/syncer_test.go index cc4d0af..210b8c8 100644 --- a/internal/provider/deb/syncer_test.go +++ b/internal/provider/deb/syncer_test.go @@ -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") + } +} diff --git a/internal/provider/rpm/syncer.go b/internal/provider/rpm/syncer.go index a746e41..68dfa66 100644 --- a/internal/provider/rpm/syncer.go +++ b/internal/provider/rpm/syncer.go @@ -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 diff --git a/internal/provider/rpm/syncer_test.go b/internal/provider/rpm/syncer_test.go index 828a43c..418e375 100644 --- a/internal/provider/rpm/syncer_test.go +++ b/internal/provider/rpm/syncer_test.go @@ -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") + } +} diff --git a/pkg/models/remote.go b/pkg/models/remote.go index ce171fa..4cff5e5 100644 --- a/pkg/models/remote.go +++ b/pkg/models/remote.go @@ -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 { diff --git a/pkg/models/remote_test.go b/pkg/models/remote_test.go index b44071e..db5bddc 100644 --- a/pkg/models/remote_test.go +++ b/pkg/models/remote_test.go @@ -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) + } + } + } +} diff --git a/ui/src/api/types.ts b/ui/src/api/types.ts index e3efbd8..8ebdb7a 100644 --- a/ui/src/api/types.ts +++ b/ui/src/api/types.ts @@ -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[]; -- 2.47.3 From 82b1852bc41f5b8369e6220f5b5893a62bb8b603 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 25 Oct 2026 13:57:54 +1100 Subject: [PATCH 3/3] Compute sync slots and the first-sync gate on the DB clock --- internal/database/alpine_github_sync.go | 4 +- internal/database/deb_github_sync.go | 4 +- internal/database/deb_github_sync_test.go | 8 +- internal/database/github_sync.go | 31 ++++--- internal/database/github_sync_test.go | 102 ++++++++++++++-------- internal/provider/alpine/syncer.go | 11 +-- internal/provider/alpine/syncer_test.go | 53 +++++------ internal/provider/deb/syncer.go | 11 +-- internal/provider/deb/syncer_test.go | 53 +++++------ internal/provider/rpm/syncer.go | 11 +-- internal/provider/rpm/syncer_test.go | 53 +++++------ pkg/models/remote.go | 15 ++-- pkg/models/remote_test.go | 61 ++----------- 13 files changed, 179 insertions(+), 238 deletions(-) diff --git a/internal/database/alpine_github_sync.go b/internal/database/alpine_github_sync.go index 4d3835a..c5db9c5 100644 --- a/internal/database/alpine_github_sync.go +++ b/internal/database/alpine_github_sync.go @@ -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 diff --git a/internal/database/deb_github_sync.go b/internal/database/deb_github_sync.go index 94f04db..0903375 100644 --- a/internal/database/deb_github_sync.go +++ b/internal/database/deb_github_sync.go @@ -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 diff --git a/internal/database/deb_github_sync_test.go b/internal/database/deb_github_sync_test.go index a5ac83a..a450686 100644 --- a/internal/database/deb_github_sync_test.go +++ b/internal/database/deb_github_sync_test.go @@ -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) } diff --git a/internal/database/github_sync.go b/internal/database/github_sync.go index 366ac4f..46e7247 100644 --- a/internal/database/github_sync.go +++ b/internal/database/github_sync.go @@ -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 { diff --git a/internal/database/github_sync_test.go b/internal/database/github_sync_test.go index 0b15ebf..730bc74 100644 --- a/internal/database/github_sync_test.go +++ b/internal/database/github_sync_test.go @@ -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) + } } }) } diff --git a/internal/provider/alpine/syncer.go b/internal/provider/alpine/syncer.go index 1b005ed..fd4e91d 100644 --- a/internal/provider/alpine/syncer.go +++ b/internal/provider/alpine/syncer.go @@ -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 diff --git a/internal/provider/alpine/syncer_test.go b/internal/provider/alpine/syncer_test.go index e5eff31..22d8b33 100644 --- a/internal/provider/alpine/syncer_test.go +++ b/internal/provider/alpine/syncer_test.go @@ -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) } } diff --git a/internal/provider/deb/syncer.go b/internal/provider/deb/syncer.go index 7875174..1fe15f1 100644 --- a/internal/provider/deb/syncer.go +++ b/internal/provider/deb/syncer.go @@ -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 diff --git a/internal/provider/deb/syncer_test.go b/internal/provider/deb/syncer_test.go index 210b8c8..87cea8c 100644 --- a/internal/provider/deb/syncer_test.go +++ b/internal/provider/deb/syncer_test.go @@ -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) } } diff --git a/internal/provider/rpm/syncer.go b/internal/provider/rpm/syncer.go index 68dfa66..ab34f9b 100644 --- a/internal/provider/rpm/syncer.go +++ b/internal/provider/rpm/syncer.go @@ -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 diff --git a/internal/provider/rpm/syncer_test.go b/internal/provider/rpm/syncer_test.go index 418e375..2ef0f1e 100644 --- a/internal/provider/rpm/syncer_test.go +++ b/internal/provider/rpm/syncer_test.go @@ -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) } } diff --git a/pkg/models/remote.go b/pkg/models/remote.go index 4cff5e5..9a7c4aa 100644 --- a/pkg/models/remote.go +++ b/pkg/models/remote.go @@ -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 { diff --git a/pkg/models/remote_test.go b/pkg/models/remote_test.go index db5bddc..25dcad5 100644 --- a/pkg/models/remote_test.go +++ b/pkg/models/remote_test.go @@ -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) - } - } - } -} -- 2.47.3