Documentation
¶
Overview ¶
Package sync orchestrates data synchronization from PeeringDB into the local SQLite database using the ent ORM.
Index ¶
- Variables
- func GetCursor(ctx context.Context, db *sql.DB, objType string) (time.Time, error)
- func GetLastSuccessfulSyncTime(ctx context.Context, db *sql.DB) (time.Time, error)
- func InitStatusTable(ctx context.Context, db *sql.DB) error
- func ReapStaleRunningRows(ctx context.Context, db *sql.DB) (int, error)
- func RecordSyncComplete(ctx context.Context, db *sql.DB, id int64, status Status) error
- func RecordSyncStart(ctx context.Context, db *sql.DB, startedAt time.Time) (int64, error)
- func UpsertCursor(ctx context.Context, db *sql.DB, objType string, lastSyncAt time.Time, ...) error
- type Status
- type Worker
- func (w *Worker) HasCompletedSync() bool
- func (w *Worker) SetRetryBackoffs(backoffs []time.Duration)
- func (w *Worker) StartScheduler(ctx context.Context, interval time.Duration)
- func (w *Worker) Sync(ctx context.Context, mode config.SyncMode) error
- func (w *Worker) SyncWithRetry(ctx context.Context, mode config.SyncMode) error
- type WorkerConfig
Constants ¶
This section is empty.
Variables ¶
var ErrSyncMemoryLimitExceeded = errors.New("sync aborted: memory limit exceeded")
ErrSyncMemoryLimitExceeded is returned by Worker.Sync when runtime.ReadMemStats reports HeapAlloc above WorkerConfig.SyncMemoryLimit after the Phase A fetch pass completes. The sync aborts without opening the ent transaction; the running mutex is released on return and the next scheduled retry proceeds normally after the Phase A scratch batches are reclaimed by GC.
Commit F (Plan 54-03) defense-in-depth against PeeringDB data growth that exceeds the 400 MB bench harness baseline at runtime (e.g. if netixlan doubles between benchmarks and production). Callers detect the sentinel via errors.Is per GO-ERR-2.
Functions ¶
func GetCursor ¶ added in v1.2.0
GetCursor returns the last successful sync timestamp for a type. Returns zero time if no cursor exists or the last sync for the type failed.
func GetLastSuccessfulSyncTime ¶ added in v1.5.0
GetLastSuccessfulSyncTime returns the completion time of the most recent successful sync, or zero time if no successful sync has been recorded.
func InitStatusTable ¶
InitStatusTable creates the sync_status and sync_cursors tables if they don't exist. These are not ent-managed entities; they store operational metadata via raw SQL.
func ReapStaleRunningRows ¶ added in v1.14.0
ReapStaleRunningRows transitions any sync_status rows stuck in "running" state to "failed" with an explanatory error message. Call from startup on the primary machine — a row can get stuck in "running" if a previous process was killed mid-sync (e.g. rolling deploy terminated the primary before the sync commit/rollback path ran). The Worker.running atomic is reset on process start, so no future sync is blocked by these rows; the transition is purely cosmetic so /ui/about and /readyz queries stop seeing phantom "running" syncs that will never complete.
Safe to call concurrently with live sync workers because the Consul lease guarantees only one primary at a time. A legitimate in-flight "running" row would be replaced by its own RecordSyncComplete call (latest write wins); in practice the reap runs BEFORE the first sync worker tick so there's no real overlap window.
Returns the number of rows transitioned.
func RecordSyncComplete ¶
RecordSyncComplete updates the sync status row with results.
func RecordSyncStart ¶
RecordSyncStart inserts a new running sync status row and returns its ID.
Types ¶
type Status ¶ added in v1.2.0
type Status struct {
LastSyncAt time.Time
Duration time.Duration
ObjectCounts map[string]int // type -> count
Status string // "success", "failed", "running"
ErrorMessage string // empty on success
}
Status represents the result of a sync operation.
func GetLastCompletedStatus ¶ added in v1.14.0
GetLastCompletedStatus returns the most recent non-running sync status row (either "success" or "failed"). Used by /readyz to fall back past an in-flight "running" row so the health check reflects the most recent outcome — whether success or failure. Returns nil if no completed sync has ever been recorded.
NOTE: this is the right answer for /readyz (which wants to know "what's the most recent outcome?" and reports unhealthy on failed) but NOT for UI freshness displays (which want "when was the last known-good data?"). UI surfaces should use GetLastSuccessfulStatus instead.
func GetLastStatus ¶ added in v1.2.0
GetLastStatus returns the most recent sync status. Returns nil if no sync has been recorded.
func GetLastSuccessfulStatus ¶ added in v1.14.0
GetLastSuccessfulStatus returns the most recent successful sync status row. This is the right answer for UI surfaces that want to display "when was the last known-good data?" — it skips past any in-flight "running" rows AND any "failed" rows to find the most recent row with status="success". Returns nil if no successful sync has ever been recorded.
Unlike GetLastSuccessfulSyncTime (which returns only the timestamp for ETag seeding), this returns the full Status struct including object counts and duration for display purposes.
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
Worker orchestrates PeeringDB data synchronization.
func NewWorker ¶
func NewWorker(pdbClient *peeringdb.Client, entClient *ent.Client, db *sql.DB, cfg WorkerConfig, logger *slog.Logger) *Worker
NewWorker creates a new sync worker. If cfg.IsPrimary is nil, it defaults to always-primary for backward compatibility (local dev, tests without explicit primary config).
func (*Worker) HasCompletedSync ¶
HasCompletedSync reports whether at least one successful sync has completed. Used for 503 behavior per D-30.
func (*Worker) SetRetryBackoffs ¶
SetRetryBackoffs overrides the default retry backoff durations. Intended for testing.
func (*Worker) StartScheduler ¶
StartScheduler runs the sync scheduler on all instances per D-22. On primary nodes it executes sync cycles; on replicas it waits for promotion. Role changes are detected dynamically at each scheduler wakeup via w.config.IsPrimary(). The scheduler stops when ctx is cancelled per CC-2.
Scheduling anchor: the next sync is scheduled at lastCompletion + interval, not at processStart + N*interval. This matters across restarts — a rolling deploy mid-interval would otherwise delay the next sync by up to a full interval (the ticker would re-anchor on process start). Concretely:
- Fresh DB (no prior successful sync) on a primary: run a full sync immediately, then schedule the next at now+interval.
- Warm start on a primary with a recent lastSync: wait until lastSync+interval; if that is already in the past, the first iteration fires immediately.
- Replica: wake every interval to check for promotion. Matches the heartbeat cadence of the pre-rewrite ticker-based design.
After each cycle — success or failure — the next sync is scheduled at time.Now()+interval. A slower-than-expected sync therefore does NOT shorten the following window, and a failed sync gives PeeringDB a full interval to recover before the next external-facing retry.
func (*Worker) Sync ¶
Sync executes a synchronization from PeeringDB to the local database. It acquires a mutex to prevent concurrent runs and wraps all database writes in a single transaction per D-19.
Sync is an orchestrator. PERF-05 splits it into three phases:
- Phase A (NO TX HELD): HTTP fetch + JSON decode stream into an isolated /tmp SQLite "scratch" database — Go heap stays bounded to one element per StreamAll handler invocation (~5-10 KB) instead of one full []T per type (~35 MB for netixlan).
- Fetch Barrier: scratch DB fully populated; open the real LiteFS tx.
- Phase B (SINGLE REAL TX): drain each scratch table in chunks, decode each chunk to typed Go structs, upsertX into the real ent tables, free the chunk, repeat. Delete stale rows. Commit. D-19 preserved: one ent.Tx wraps every real-DB write.
The scratch DB is unlinked on both success and error via defer. See internal/sync/scratch.go for the scratch DB lifecycle and pragmas. Commit D' — mandatory per Decision #2 because Commit A baseline (535 MiB) and Commit D baseline (613 MiB) both exceeded the 400 MiB gate on production-scale fixtures.
Commit F (Plan 54-03): between the Phase A fetch barrier and the ent.Tx open, Sync calls w.checkMemoryLimit with the current runtime.MemStats.HeapAlloc reading. If SyncMemoryLimit is set and the heap exceeds it, Sync aborts with ErrSyncMemoryLimitExceeded before opening the tx — defense-in-depth against PeeringDB data growth that exceeds the benchmark baseline at runtime.
REFAC-03 line budget is <= 100 — enforced by TestWorkerSync_LineBudget.
Plan 59-05 (VIS-05, D-08/D-09): Sync is the SOLE production call site for privacy.DecisionContext(ctx, privacy.Allow). The bypass is stamped on ctx before any other work — before the w.running CAS, before any otel span Start, before any ent call — so every downstream ent read, upsert, and child goroutine (e.g. runSyncCycle's demotion monitor) inherits allow-all via standard context.WithValue parent-chain lookup. The ent rule-dispatch loop short-circuits at the stored decision, so no per-rule predicate mutation runs on bypass ctx.
TestSyncBypass_SingleCallSite enforces "exactly one production call site" — do NOT add the bypass in SyncWithRetry, runSyncCycle, any upsert helper, or any HTTP path. Tier elevation for non-sync callers goes through internal/privctx.WithTier (TierUsers), a different mechanism.
func (*Worker) SyncWithRetry ¶
SyncWithRetry calls Sync and retries on failure with exponential backoff per D-21.
Rate-limit short-circuit: when the wrapped error is a *peeringdb.RateLimitError, the retry ladder is skipped entirely. PeeringDB's unauthenticated quota is 1 request per distinct query-string per hour, and all three default backoffs (30s, 2m, 8m) fall well inside that window — every retry within the window is guaranteed to 429 again AND consumes another slot against the hourly quota. Returning immediately lets the hourly scheduler retry naturally on its next tick (1h interval ≥ most Retry-After values we've observed).
type WorkerConfig ¶
type WorkerConfig struct {
IncludeDeleted bool
IsPrimary func() bool // live primary detection; nil defaults to always-primary
SyncMode config.SyncMode
// OnSyncComplete is called after a successful sync with per-type object
// counts and the completion timestamp. The timestamp is the same
// value persisted into the sync_status row by recordSuccess, so
// downstream consumers (e.g. the caching middleware ETag setter wired
// in cmd/peeringdb-plus/main.go for PERF-07) stay in lock-step with
// the database without an extra round-trip.
OnSyncComplete func(counts map[string]int, syncTime time.Time)
// SyncMemoryLimit is the peak Go heap ceiling (bytes) checked
// after Phase A fetch completes and before the ent.Tx opens. If
// runtime.MemStats.HeapAlloc exceeds this value, Sync aborts with
// ErrSyncMemoryLimitExceeded. Zero disables the guardrail. Wired
// from config.Config.SyncMemoryLimit by main.go. Commit F default
// is 400 MB (matches the DEBT-03 bench regression gate).
SyncMemoryLimit int64
// HeapWarnBytes is the peak Go heap threshold (bytes) above which
// the end-of-sync-cycle emitter fires slog.Warn("heap threshold
// crossed", ...). The OTel span attr pdbplus.sync.peak_heap_mib is
// attached regardless. Zero disables only the Warn (not the attr).
// Wired from config.Config.HeapWarnBytes by main.go.
//
// SEED-001 escalation signal: sustained breach triggers the
// incremental-sync evaluation path documented in
// .planning/seeds/SEED-001-incremental-sync-evaluation.md.
HeapWarnBytes int64
// RSSWarnBytes is the peak OS RSS threshold (bytes) above which
// the emitter fires slog.Warn. Read from /proc/self/status VmHWM on
// Linux; skipped on other OSes (the RSS attr is then omitted — it
// is not set to zero). Zero disables only the Warn.
RSSWarnBytes int64
}
WorkerConfig holds configuration for the sync worker.
Source Files