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 InitialObjectCounts(ctx context.Context, client *ent.Client) (map[string]int64, 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 StepOrder() []string
- 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 sync timestamp for a type. Returns zero time only if no cursor row exists for the type.
v1.18.3: dropped the prior `AND last_status = 'success'` filter. The cursor is a high-water-mark timestamp; failure observability lives in the separate sync_status table. Coupling cursor reads to last_status caused subtle "all cursors zero after a failed cycle" surprises that could trigger expensive re-fetches (and was load-bearing in the v1.18.2 bootstrap regression). The last_status column is preserved for stored data compatibility and any future dashboard queries.
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 InitialObjectCounts ¶ added in v1.18.0
InitialObjectCounts runs a one-shot Count(ctx) against each of the 13 PeeringDB entity tables and returns the result keyed by PeeringDB type name. The keys match those produced by syncSteps() so the same atomic cache can be primed by either the startup path (this helper) or the OnSyncComplete callback.
Implements OBS-01 D-01: synchronous startup population so the pdbplus_data_type_count gauge reports correct values within 30s of process start instead of holding zeros until the first sync cycle completes (~15 min default).
Cost: ~1-2s on a primed LiteFS DB (13 sequential SQLite COUNT(*) queries against indexed tables). Replicas already cold-sync the DB in 5-45s; the added latency is noise on top of hydration.
Errors are returned wrapped with the type name so an operator can see which table failed; partial results are NOT returned — a single failure aborts the whole call to keep the contract simple. The caller chooses whether to fail-fast or proceed with stale zeros.
Note: counts include all rows regardless of status (matching the existing OnSyncComplete cache contract — "raw upserted-row count from the latest sync cycle"). Phase 68 tombstones (status="deleted") are rows the dashboard wants to see in "Total Objects" until tombstone GC ships (SEED-004 dormant). If a future requirement wants live-only counts, that's a separate metric.
Tier elevation: ctx is stamped with TierUsers via privctx.WithTier before each Count call. Without this, Poc.Policy() (the only entity with a row-level Privacy policy) would filter visible!="Public" rows out of this counter while the OnSyncComplete writer — which runs under privacy.DecisionContext(ctx, privacy.Allow) — counts every row. The cross-writer disagreement specifically on POC produced the `pdbplus_data_type_count{type="poc"}` 2x/0.5x oscillation that was visible on the Grafana "Object Counts Over Time" panel: replicas (which only ever run InitialObjectCounts) held the public-only count `P` while the primary's cache flipped between `T ≈ 2P` (just after a full sync) and tiny incremental deltas, and `max by(type)` across the 8-instance fleet alternated between `T` and `P` accordingly. See .planning/debug/poc-count-doubling-halving.md for the full analysis.
We use privctx.WithTier(ctx, TierUsers) rather than privacy.DecisionContext(ctx, privacy.Allow) deliberately — the internal/sync/bypass_audit_test.go invariant restricts privacy.Allow to exactly one production call site (Worker.Sync). TierUsers is the documented mechanism for "non-sync tier elevation" (see internal/sync/bypass_audit_test.go:208 and internal/privctx/privctx.go godoc) and produces the same effect on Poc.Policy without diluting the bypass audit.
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 {
IsPrimary func() bool // live primary detection; nil defaults to always-primary
SyncMode config.SyncMode
// OnSyncComplete is called after a successful sync with the worker's
// ctx 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.
//
// Quick task 260427-ojm: the per-cycle upsert-count map (the old
// `counts map[string]int` arg) was removed. It was the wrong value to
// feed into the pdbplus_data_type_count gauge cache — for incremental
// syncs it was a delta, and for Poc it never agreed with the
// privacy-filtered Count(ctx) used at startup ("doubling-halving").
// Consumers should run pdbsync.InitialObjectCounts(ctx, client) on
// the supplied ctx if they want live row counts, or query
// sync_status (which still persists the raw upsert deltas) if they
// want cycle telemetry.
OnSyncComplete func(ctx context.Context, 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_bytes 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
// FKBackfillMaxRequestsPerCycle caps the number of underlying HTTP
// requests issued by FK-backfill per sync cycle. v1.18.5 semantic
// shift: the previous cap counted ROWS (per-row in 260428-2zl,
// nominally per-row but batched in 260428-5xt — a weak circuit
// breaker once batching collapsed N rows into 1 request). The cap
// now directly bounds upstream HTTP traffic, the surface protected
// by upstream's API_THROTTLE_REPEATED_REQUEST and our local rate
// limiter. Default 20 (PDBPLUS_FK_BACKFILL_MAX_REQUESTS_PER_CYCLE)
// — at 1 req/sec auth, ≈20s of upstream pressure max per cycle.
// 0 disables backfill entirely (drop-on-miss behavior, preserved
// as an operator escape-hatch).
FKBackfillMaxRequestsPerCycle int
// FKBackfillTimeout is the per-cycle wall-clock budget for FK
// backfill HTTP activity. v1.18.3: added because backfill calls
// happen inside the sync transaction; without a deadline a cascade
// of slow / rate-limited backfills could hold the tx open for tens
// of minutes, stalling LiteFS replication. After the deadline,
// fkBackfillParent short-circuits to drop-on-miss so the rest of
// the sync (bulk fetches + upserts) can commit and the next cycle
// picks up where we left off. Default 5 minutes
// (PDBPLUS_FK_BACKFILL_TIMEOUT). Zero or negative disables the
// deadline (only the cap applies).
FKBackfillTimeout time.Duration
}
WorkerConfig holds configuration for the sync worker.
Source Files