From e69cc0fb6e205c0cc5c42ec8bde719f0dad52279 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Sat, 4 Jul 2026 00:19:32 -0700 Subject: [PATCH 1/2] feat(selfhost): add per-installation GitHub-fetch concurrency admission QUEUE_BACKGROUND_CONCURRENCY caps how many background jobs run AT ALL, globally -- it has no notion of WHICH installation those jobs belong to, so once an operator raises it above the default of 1, one installation's background sweep/backfill can claim every available background slot at once and starve every other installation's background work, even though GitHub's rate-limit admission is nowhere near exhausted for either installation. Adds a third, per-installation claim-time admission check (alongside GitHub rate-limit admission and maintenance-lane admission), checked only for background jobs that make GitHub calls -- never for foreground live-PR work (github-webhook, agent-regate-pr; the latter is exempted by its actual claim-priority, not by job type, since a live regate must never be deferred by this policy regardless of how the rate-limit-budget classifier sees it). A denied job is deferred with jitter, never dropped, mirroring the existing maintenance-admission mechanism's shape. Deliberately in-process (a plain per-installation in-flight counter), not DB-backed: the queue's existing active/activeBackground counters are already per-process scalars with no cross-process aggregation, and this mirrors that same, already-supported single-process-per-deployment topology rather than introducing a new coordination mechanism. --- .env.example | 13 + .../src/lib/selfhost-env-reference.ts | 5 + .../installation-concurrency-admission.ts | 109 +++++++++ src/selfhost/metrics.ts | 2 + src/selfhost/pg-queue.ts | 75 ++++++ src/selfhost/queue-common.ts | 14 ++ src/selfhost/sqlite-queue.ts | 69 ++++++ ...installation-concurrency-admission.test.ts | 230 ++++++++++++++++++ test/unit/selfhost-pg-queue.test.ts | 138 +++++++++++ test/unit/selfhost-sqlite-queue.test.ts | 133 ++++++++++ 10 files changed, 788 insertions(+) create mode 100644 src/selfhost/installation-concurrency-admission.ts create mode 100644 test/unit/selfhost-installation-concurrency-admission.test.ts diff --git a/.env.example b/.env.example index 630573cb5b..9d756784c7 100644 --- a/.env.example +++ b/.env.example @@ -323,6 +323,19 @@ REDIS_URL=redis://redis:6379 # REQUIRED for the self-host review # # released job re-attempting GitHub at once and immediately # # re-tripping the same rate-limit bucket it was deferred for +# --- Per-installation GitHub-fetch concurrency (#selfhost-installation-concurrency) --- +# QUEUE_BACKGROUND_CONCURRENCY caps how many background jobs run AT ALL, globally -- it has no notion of WHICH +# installation those jobs belong to, so raising it above the default of 1 lets one installation's background +# sweep/backfill claim every available background slot at once, starving every OTHER installation's background +# work even though neither is anywhere near GitHub-rate-limit exhaustion. This is a THIRD, per-installation +# claim-time admission check (alongside GitHub rate-limit admission and maintenance-lane admission above), +# checked only for background jobs that make GitHub calls -- never for foreground live-PR work +# (github-webhook/agent-regate-pr). A denied job is deferred with jitter, never dropped. Per instance, per +# process (same single-process-per-deployment model as QUEUE_CONCURRENCY/QUEUE_BACKGROUND_CONCURRENCY above). +# GITHUB_INSTALLATION_CONCURRENCY_ENABLED=true # set false/0/off to fully disable this check +# GITHUB_INSTALLATION_CONCURRENCY_LIMIT=2 # max concurrent GitHub-fetching background jobs per installation +# GITHUB_INSTALLATION_CONCURRENCY_DEFER_MS=15000 # base defer duration on denial, before jitter (15s) + # --- Caddy HTTPS terminator (#1203; requires --profile caddy) --- # DOMAIN=gittensory.example.com # fully-qualified domain; Caddy auto-obtains a Let's Encrypt cert diff --git a/apps/gittensory-ui/src/lib/selfhost-env-reference.ts b/apps/gittensory-ui/src/lib/selfhost-env-reference.ts index 6efabb2cf5..5540d01197 100644 --- a/apps/gittensory-ui/src/lib/selfhost-env-reference.ts +++ b/apps/gittensory-ui/src/lib/selfhost-env-reference.ts @@ -113,6 +113,10 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [ name: "GITHUB_CACHE_TTL_SECONDS", firstReference: "src/server.ts:508", }, + { + name: "GITHUB_INSTALLATION_CONCURRENCY_ENABLED", + firstReference: "src/selfhost/installation-concurrency-admission.ts:34", + }, { name: "GITTENSORY_REPO_CONFIG_DIR", firstReference: "src/server.ts:288", @@ -345,6 +349,7 @@ export const SELFHOST_ENV_REFERENCE_MARKDOWN = [ "| `GITHUB_APP_ID` | `src/selfhost/orb-collector.ts:59` |", "| `GITHUB_APP_PRIVATE_KEY` | `src/selfhost/orb-collector.ts:166` |", "| `GITHUB_CACHE_TTL_SECONDS` | `src/server.ts:508` |", + "| `GITHUB_INSTALLATION_CONCURRENCY_ENABLED` | `src/selfhost/installation-concurrency-admission.ts:34` |", "| `GITTENSORY_REPO_CONFIG_DIR` | `src/server.ts:288` |", "| `GITTENSORY_VERSION` | `src/selfhost/otel.ts:62` |", "| `HOME` | `src/selfhost/ai.ts:302` |", diff --git a/src/selfhost/installation-concurrency-admission.ts b/src/selfhost/installation-concurrency-admission.ts new file mode 100644 index 0000000000..14e409a4d9 --- /dev/null +++ b/src/selfhost/installation-concurrency-admission.ts @@ -0,0 +1,109 @@ +// Per-installation GitHub-fetch concurrency admission (#selfhost-installation-concurrency). The queue's own +// QUEUE_BACKGROUND_CONCURRENCY caps how many background jobs run AT ALL, globally -- it has no notion of WHICH +// installation those jobs belong to, so once an operator raises that cap above its default of 1, one +// installation's background sweep/backfill can claim every available background slot at once and starve every +// OTHER installation's background work, even though GitHub's rate-limit admission (queue-common.ts) is nowhere +// near exhausted for either installation. This module adds an ORTHOGONAL signal, checked at claim time +// alongside GitHub rate-limit admission and maintenance-lane pressure admission: is THIS installation already +// running its share of concurrent GitHub-fetching background jobs right now? A denied job is pushed back to +// 'pending' with a jittered future run_after, same as the other two admission layers -- never dropped. +// +// Deliberately in-process, not DB-backed: the queue's existing `active`/`activeBackground` counters (pg-queue.ts +// / sqlite-queue.ts) are already per-process scalars with no cross-process aggregation, and maintenance- +// admission.ts's own hostLoadAvg1PerCore() is inherently per-box too -- single-process-per-deployment is already +// the supported topology for the whole admission system (the SQLite backend structurally cannot share state +// across processes at all). A DB-backed live COUNT(*) query would need a new indexed installation column on +// every job row just to answer a question this in-process tracker answers for free in that topology. +// +// Deliberately NEVER applied to foreground jobs (github-webhook, agent-regate-pr): this policy only ever runs +// for a job where isGitHubBudgetBackgroundJob() is true, mirroring exactly how maintenance-admission.ts's +// evaluateMaintenanceAdmission is only invoked for a background-priority job -- "reserve headroom for live PR +// work" is satisfied structurally, not via a headroom calculation. +import { deterministicJitterMs, parsePositiveIntEnv } from "./queue-common"; + +const DEFAULT_MAX_CONCURRENT_PER_INSTALLATION = 2; +const DEFAULT_DEFER_MS = 15_000; + +export interface InstallationConcurrencyConfig { + enabled: boolean; + maxConcurrentPerInstallation: number; + deferMs: number; +} + +function installationConcurrencyEnabled(): boolean { + const raw = (process.env.GITHUB_INSTALLATION_CONCURRENCY_ENABLED ?? "").trim().toLowerCase(); + return raw !== "0" && raw !== "false" && raw !== "off" && raw !== "no"; +} + +/** Reads every GITHUB_INSTALLATION_CONCURRENCY_* knob from process.env, each with a sane, protective default. + * Resolved ONCE per queue instance (mirrors resolveMaintenanceAdmissionConfig) rather than per job. */ +export function resolveInstallationConcurrencyConfig(): InstallationConcurrencyConfig { + return { + enabled: installationConcurrencyEnabled(), + maxConcurrentPerInstallation: parsePositiveIntEnv("GITHUB_INSTALLATION_CONCURRENCY_LIMIT", { + min: 1, + fallback: DEFAULT_MAX_CONCURRENT_PER_INSTALLATION, + }), + deferMs: parsePositiveIntEnv("GITHUB_INSTALLATION_CONCURRENCY_DEFER_MS", { + min: 1_000, + fallback: DEFAULT_DEFER_MS, + }), + }; +} + +export type InstallationConcurrencyReason = "disabled" | "concurrency_high" | "clear"; + +export interface InstallationConcurrencyDecision { + admit: boolean; + reason: InstallationConcurrencyReason; +} + +/** PURE policy decision: is this installation allowed one more concurrent GitHub-budget-background job right + * now? `currentInFlightCount` is the caller's own live read of the InstallationConcurrencyTracker below for + * this exact admission key, taken immediately before this call. */ +export function evaluateInstallationConcurrencyAdmission( + config: InstallationConcurrencyConfig, + currentInFlightCount: number, +): InstallationConcurrencyDecision { + if (!config.enabled) return { admit: true, reason: "disabled" }; + if (currentInFlightCount >= config.maxConcurrentPerInstallation) { + return { admit: false, reason: "concurrency_high" }; + } + return { admit: true, reason: "clear" }; +} + +/** Jittered defer duration for a denied background job -- the base `deferMs` plus up to another `deferMs` of + * deterministic jitter (seeded by the job's own identity) so a cohort of denied jobs for the same installation + * doesn't wake up on the same tick and immediately re-trip this same check (mirrors + * maintenanceAdmissionDeferMs). Its own, shorter default (15s vs. maintenance's 3min) reflects that a + * background-fetch burst for one installation settles on the order of seconds, not minutes. */ +export function installationConcurrencyDeferMs(config: InstallationConcurrencyConfig, jitterSeed: string): number { + return config.deferMs + deterministicJitterMs(jitterSeed, config.deferMs); +} + +/** The ONLY stateful piece in this module -- a plain in-process in-flight counter keyed by GitHub rate-limit + * admission key (installation:). Constructed once per queue backend at module scope, mirroring how + * `active`/`activeBackground` are module-scope scalars in pg-queue.ts/sqlite-queue.ts -- never exported as a + * shared singleton, so it can only be mutated from the claim path that owns it. */ +export class InstallationConcurrencyTracker { + private readonly counts = new Map(); + + currentCount(admissionKey: string): number { + return this.counts.get(admissionKey) ?? 0; + } + + increment(admissionKey: string): void { + this.counts.set(admissionKey, this.currentCount(admissionKey) + 1); + } + + /** Floors at 0 and deletes the key once it reaches 0, so a busy deployment with many distinct installations + * never grows this Map unboundedly with stale zero entries. */ + decrement(admissionKey: string): void { + const next = Math.max(0, this.currentCount(admissionKey) - 1); + if (next === 0) { + this.counts.delete(admissionKey); + } else { + this.counts.set(admissionKey, next); + } + } +} diff --git a/src/selfhost/metrics.ts b/src/selfhost/metrics.ts index da595624e0..eefac79186 100644 --- a/src/selfhost/metrics.ts +++ b/src/selfhost/metrics.ts @@ -77,6 +77,8 @@ const DEFAULT_METRIC_META: readonly (readonly [string, MetricMeta])[] = [ ["gittensory_jobs_rate_limit_budget_deferred_total", { help: "Jobs deferred by rate-limit budget checks.", type: "counter" }], ["gittensory_jobs_rate_limited_by_type_total", { help: "Jobs rate-limited by job type.", type: "counter" }], ["gittensory_jobs_maintenance_admission_deferred_by_reason_total", { help: "Maintenance jobs deferred by reason.", type: "counter" }], + ["gittensory_jobs_installation_concurrency_deferred_total", { help: "Background jobs deferred by per-installation GitHub-fetch concurrency admission.", type: "counter" }], + ["gittensory_jobs_installation_concurrency_deferred_by_reason_total", { help: "Per-installation GitHub-fetch concurrency deferrals by reason and job type.", type: "counter" }], ["gittensory_jobs_dead_letter_revived_total", { help: "Dead-letter jobs revived for retry.", type: "counter" }], ["gittensory_jobs_foreground_liveness_released_total", { help: "Foreground-priority jobs force-released from a stale deferral by the liveness sweep.", type: "counter" }], ["gittensory_jobs_foreground_liveness_released_by_reason_total", { help: "Foreground liveness releases by reason (age vs rate_limit_cleared).", type: "counter" }], diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts index 50057990b0..3893dfccb0 100644 --- a/src/selfhost/pg-queue.ts +++ b/src/selfhost/pg-queue.ts @@ -18,6 +18,7 @@ import { githubRateLimitMetricContext, githubRateLimitRetryDelayMs, buildSelfHostQueueSnapshot, + installationConcurrencyKeyForJob, isForegroundJobPriority, jobCoalesceAbsorbedByKey, jobCoalesceKey, @@ -150,6 +151,12 @@ import { selectForegroundDeferralsToRelease, type ForegroundLivenessConfig, } from "./foreground-liveness"; +import { + evaluateInstallationConcurrencyAdmission, + installationConcurrencyDeferMs, + resolveInstallationConcurrencyConfig, + InstallationConcurrencyTracker, +} from "./installation-concurrency-admission"; import type { JobMessage } from "../types"; const TABLE = "_selfhost_jobs"; @@ -244,6 +251,11 @@ interface JobRow { priority: number | string; created_at: number | string; backgroundSlotReserved?: boolean; + // #selfhost-installation-concurrency: set only when this job was ADMITTED-AND-COUNTED against a specific + // installation's in-flight tracker (see the admission block right before the dispatch try/finally below) -- + // stamped here so the shared finally can release the SAME key, mirroring backgroundSlotReserved's own + // admit-time-stamp / release-in-finally shape. + installationConcurrencyKey?: string; } export interface PgQueueOptions { @@ -286,6 +298,8 @@ export function createPgQueue( let foregroundLivenessTimer: ReturnType | null = null; const maintenanceAdmissionConfig: MaintenanceAdmissionConfig = resolveMaintenanceAdmissionConfig(); const foregroundLivenessConfig: ForegroundLivenessConfig = resolveForegroundLivenessConfig(); + const installationConcurrencyConfig = resolveInstallationConcurrencyConfig(); + const installationConcurrencyTracker = new InstallationConcurrencyTracker(); async function init(): Promise { await pool.query(DDL); @@ -1163,6 +1177,66 @@ export function createPgQueue( }); } } + // Per-installation GitHub-fetch concurrency admission (#selfhost-installation-concurrency), the last-mile + // gate: only reached by a job that already passed rate-limit admission and (if applicable) maintenance- + // lane admission above, immediately before it actually claims a dispatch slot. Explicitly excludes + // foreground-priority jobs (mirrors the maintenance-admission guard above) -- isGitHubBudgetBackgroundJob + // (which installationConcurrencyKeyForJob is built on) answers "does this job draw GitHub rate-limit + // BUDGET", which is also true for a live (non-sweep, non-manual) agent-regate-pr job; that job is still + // FOREGROUND priority and must never be deferred by this policy, so the exclusion is a separate, explicit + // check here rather than folded into the key resolver itself. installationConcurrencyKey is null for + // background jobs whose payload carries no resolvable installationId too -- those fall through unaffected. + const installationConcurrencyKey = isForegroundJobPriority(Number(job.priority)) + ? null + : installationConcurrencyKeyForJob(message); + if (installationConcurrencyKey) { + const decision = evaluateInstallationConcurrencyAdmission( + installationConcurrencyConfig, + installationConcurrencyTracker.currentCount(installationConcurrencyKey), + ); + if (!decision.admit) { + await withReviewSpan( + "selfhost.queue.installation_concurrency_deferred", + { "job.type": message.type, "queue.backend": "postgres", "installation_concurrency.reason": decision.reason }, + async () => { + const now = Date.now(); + const retryAfter = now + installationConcurrencyDeferMs( + installationConcurrencyConfig, + `${job.job_key ?? ""}:${job.id}:${job.payload}`, + ); + const update = await retryPoolUpdateOrLeaveForReclaim( + () => + pool.query( + `UPDATE ${TABLE} SET status='pending', run_after=GREATEST(run_after, $1), last_error=COALESCE(last_error, $2) WHERE id=$3`, + [retryAfter, `installation concurrency admission deferred: ${decision.reason}`, job.id], + ), + job.id, + "selfhost_queue_pg_connection_lost_on_installation_concurrency_defer", + ); + if (update?.rowCount) { + await recordQueueMetric("gittensory_jobs_installation_concurrency_deferred_total"); + incr("gittensory_jobs_installation_concurrency_deferred_by_reason_total", { + reason: decision.reason, + job_type: message.type, + }); + console.warn( + JSON.stringify({ + level: "warn", + event: "selfhost_queue_installation_concurrency_deferred", + jobType: message.type, + reason: decision.reason, + retry_after_ms: Math.max(0, retryAfter - now), + }), + ); + } + }, + { parentTraceParent: jobTraceParent }, + ); + return true; + } + installationConcurrencyTracker.increment(installationConcurrencyKey); + job.installationConcurrencyKey = installationConcurrencyKey; + } try { await withReviewSpan( "selfhost.queue.job", @@ -1319,6 +1393,7 @@ export function createPgQueue( activeJobIds.delete(job.id); if (job.backgroundSlotReserved) activeBackground = Math.max(0, activeBackground - 1); + if (job.installationConcurrencyKey) installationConcurrencyTracker.decrement(job.installationConcurrencyKey); } } diff --git a/src/selfhost/queue-common.ts b/src/selfhost/queue-common.ts index 7f45e6c9d9..f26b865e96 100644 --- a/src/selfhost/queue-common.ts +++ b/src/selfhost/queue-common.ts @@ -402,6 +402,20 @@ export function githubRateLimitAdmissionKeyForJob(message: JobMessage): GitHubRa : null; } +// #selfhost-installation-concurrency: the admission key a per-installation concurrency limiter should track +// THIS job under, or null when the job either makes no GitHub calls isGitHubBudgetBackgroundJob cares about, or +// carries no resolvable installationId. Reusing githubRateLimitAdmissionKeyForJob (rather than inventing a +// second key function) keeps the rate-limit-admission key and the concurrency-admission key for the same job +// always identical by construction. NOTE: isGitHubBudgetBackgroundJob is true for a live (non-sweep, non-manual) +// agent-regate-pr job too, since that job DOES draw GitHub rate-limit budget under this key -- but that job is +// still FOREGROUND priority. This function deliberately does NOT filter foreground jobs out itself (it answers +// "what key would this job's GitHub calls draw against", not "should a background-only policy apply to it") -- +// the caller (pg-queue.ts/sqlite-queue.ts) is responsible for its own `!isForegroundJobPriority(...)` guard +// before ever calling this, exactly mirroring how isMaintenanceJobType is similarly guarded at its own call site. +export function installationConcurrencyKeyForJob(message: JobMessage): GitHubRateLimitAdmissionKey | null { + return isGitHubBudgetBackgroundJob(message) ? githubRateLimitAdmissionKeyForJob(message) : null; +} + export type GitHubRateLimitAdmissionKind = "background" | "webhook"; export type GitHubRateLimitAdmissionTarget = { diff --git a/src/selfhost/sqlite-queue.ts b/src/selfhost/sqlite-queue.ts index adfd40d832..a0440879c3 100644 --- a/src/selfhost/sqlite-queue.ts +++ b/src/selfhost/sqlite-queue.ts @@ -19,6 +19,7 @@ import { githubRateLimitMetricContext, githubRateLimitRetryDelayMs, buildSelfHostQueueSnapshot, + installationConcurrencyKeyForJob, isForegroundJobPriority, jobCoalesceAbsorbedByKey, jobCoalesceKey, @@ -51,6 +52,12 @@ import { type MaintenanceAdmissionConfig, type MaintenancePressureSignals, } from "./maintenance-admission"; +import { + evaluateInstallationConcurrencyAdmission, + installationConcurrencyDeferMs, + resolveInstallationConcurrencyConfig, + InstallationConcurrencyTracker, +} from "./installation-concurrency-admission"; import { AGENT_REGATE_PR_JOB_KEY_PREFIX, backlogRepoCandidatesFromJobKeys, @@ -157,6 +164,10 @@ interface JobRow { priority: number; created_at: number; backgroundSlotReserved?: boolean; + // #selfhost-installation-concurrency: set only when this job was ADMITTED-AND-COUNTED against a specific + // installation's in-flight tracker -- stamped at admission time so the shared finally can release the SAME + // key, mirroring backgroundSlotReserved's own admit-time-stamp / release-in-finally shape. + installationConcurrencyKey?: string; } export interface SqliteQueueOptions { @@ -274,6 +285,8 @@ export function createSqliteQueue( ); const maintenanceAdmissionConfig: MaintenanceAdmissionConfig = resolveMaintenanceAdmissionConfig(); const foregroundLivenessConfig: ForegroundLivenessConfig = resolveForegroundLivenessConfig(); + const installationConcurrencyConfig = resolveInstallationConcurrencyConfig(); + const installationConcurrencyTracker = new InstallationConcurrencyTracker(); // Recover jobs a crashed previous run left mid-flight → make them claimable again. const recovered = recoverProcessingJobs(driver); if (recovered) { @@ -896,6 +909,61 @@ export function createSqliteQueue( }); } } + // Per-installation GitHub-fetch concurrency admission (#selfhost-installation-concurrency), the last-mile + // gate: only reached by a job that already passed rate-limit admission and (if applicable) maintenance- + // lane admission above, immediately before it actually claims a dispatch slot. Explicitly excludes + // foreground-priority jobs (mirrors the maintenance-admission guard above) -- isGitHubBudgetBackgroundJob + // (which installationConcurrencyKeyForJob is built on) answers "does this job draw GitHub rate-limit + // BUDGET", which is also true for a live (non-sweep, non-manual) agent-regate-pr job; that job is still + // FOREGROUND priority and must never be deferred by this policy, so the exclusion is a separate, explicit + // check here rather than folded into the key resolver itself. installationConcurrencyKey is null for + // background jobs whose payload carries no resolvable installationId too -- those fall through unaffected. + const installationConcurrencyKey = isForegroundJobPriority(job.priority) + ? null + : installationConcurrencyKeyForJob(message); + if (installationConcurrencyKey) { + const decision = evaluateInstallationConcurrencyAdmission( + installationConcurrencyConfig, + installationConcurrencyTracker.currentCount(installationConcurrencyKey), + ); + if (!decision.admit) { + await withReviewSpan( + "selfhost.queue.installation_concurrency_deferred", + { "job.type": message.type, "queue.backend": "sqlite", "installation_concurrency.reason": decision.reason }, + async () => { + const now = Date.now(); + const retryAfter = now + installationConcurrencyDeferMs( + installationConcurrencyConfig, + `${job.job_key ?? ""}:${job.id}:${job.payload}`, + ); + const { changes } = driver.query( + `UPDATE ${TABLE} SET status='pending', run_after=max(run_after, ?), last_error=coalesce(last_error, ?) WHERE id=?`, + [retryAfter, `installation concurrency admission deferred: ${decision.reason}`, job.id], + ); + if (changes) { + recordQueueMetric(driver, "gittensory_jobs_installation_concurrency_deferred_total"); + incr("gittensory_jobs_installation_concurrency_deferred_by_reason_total", { + reason: decision.reason, + job_type: message.type, + }); + console.warn( + JSON.stringify({ + level: "warn", + event: "selfhost_queue_installation_concurrency_deferred", + jobType: message.type, + reason: decision.reason, + retry_after_ms: Math.max(0, retryAfter - now), + }), + ); + } + }, + { parentTraceParent: jobTraceParent }, + ); + return true; + } + installationConcurrencyTracker.increment(installationConcurrencyKey); + job.installationConcurrencyKey = installationConcurrencyKey; + } try { await withReviewSpan( "selfhost.queue.job", @@ -1015,6 +1083,7 @@ export function createSqliteQueue( activeJobIds.delete(job.id); if (job.backgroundSlotReserved) activeBackground = Math.max(0, activeBackground - 1); + if (job.installationConcurrencyKey) installationConcurrencyTracker.decrement(job.installationConcurrencyKey); } } diff --git a/test/unit/selfhost-installation-concurrency-admission.test.ts b/test/unit/selfhost-installation-concurrency-admission.test.ts new file mode 100644 index 0000000000..9359b8e4de --- /dev/null +++ b/test/unit/selfhost-installation-concurrency-admission.test.ts @@ -0,0 +1,230 @@ +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { + evaluateInstallationConcurrencyAdmission, + installationConcurrencyDeferMs, + InstallationConcurrencyTracker, + resolveInstallationConcurrencyConfig, + type InstallationConcurrencyConfig, +} from "../../src/selfhost/installation-concurrency-admission"; +import { installationConcurrencyKeyForJob } from "../../src/selfhost/queue-common"; +import type { JobMessage } from "../../src/types"; + +const CONFIG: InstallationConcurrencyConfig = { + enabled: true, + maxConcurrentPerInstallation: 2, + deferMs: 15_000, +}; + +describe("resolveInstallationConcurrencyConfig", () => { + const envKeys = [ + "GITHUB_INSTALLATION_CONCURRENCY_ENABLED", + "GITHUB_INSTALLATION_CONCURRENCY_LIMIT", + "GITHUB_INSTALLATION_CONCURRENCY_DEFER_MS", + ] as const; + const saved: Record = {}; + + beforeEach(() => { + for (const key of envKeys) { + saved[key] = process.env[key]; + delete process.env[key]; + } + }); + + afterEach(() => { + for (const key of envKeys) { + if (saved[key] === undefined) delete process.env[key]; + else process.env[key] = saved[key]; + } + }); + + it("returns protective defaults with no env overrides", () => { + expect(resolveInstallationConcurrencyConfig()).toEqual({ + enabled: true, + maxConcurrentPerInstallation: 2, + deferMs: 15_000, + }); + }); + + it("reads every knob from the environment when set", () => { + process.env.GITHUB_INSTALLATION_CONCURRENCY_LIMIT = "5"; + process.env.GITHUB_INSTALLATION_CONCURRENCY_DEFER_MS = "30000"; + const config = resolveInstallationConcurrencyConfig(); + expect(config.maxConcurrentPerInstallation).toBe(5); + expect(config.deferMs).toBe(30_000); + }); + + it("falls back to the default limit on an invalid (non-numeric) value", () => { + process.env.GITHUB_INSTALLATION_CONCURRENCY_LIMIT = "not-a-number"; + expect(resolveInstallationConcurrencyConfig().maxConcurrentPerInstallation).toBe(2); + }); + + it("falls back to the default limit below the min (1)", () => { + process.env.GITHUB_INSTALLATION_CONCURRENCY_LIMIT = "0"; + expect(resolveInstallationConcurrencyConfig().maxConcurrentPerInstallation).toBe(2); + }); + + it.each(["0", "false", "off", "no"])("treats GITHUB_INSTALLATION_CONCURRENCY_ENABLED=%s as disabled", (value) => { + process.env.GITHUB_INSTALLATION_CONCURRENCY_ENABLED = value; + expect(resolveInstallationConcurrencyConfig().enabled).toBe(false); + }); + + it.each(["1", "true", "on", "yes", "anything-else"])( + "treats GITHUB_INSTALLATION_CONCURRENCY_ENABLED=%s as enabled", + (value) => { + process.env.GITHUB_INSTALLATION_CONCURRENCY_ENABLED = value; + expect(resolveInstallationConcurrencyConfig().enabled).toBe(true); + }, + ); +}); + +describe("evaluateInstallationConcurrencyAdmission", () => { + it("admits unconditionally when disabled, even at/above the limit", () => { + const decision = evaluateInstallationConcurrencyAdmission({ ...CONFIG, enabled: false }, 99); + expect(decision).toEqual({ admit: true, reason: "disabled" }); + }); + + it("admits when the in-flight count is below the limit", () => { + expect(evaluateInstallationConcurrencyAdmission(CONFIG, 0)).toEqual({ admit: true, reason: "clear" }); + expect(evaluateInstallationConcurrencyAdmission(CONFIG, 1)).toEqual({ admit: true, reason: "clear" }); + }); + + it("denies exactly AT the limit (>=, not >)", () => { + expect(evaluateInstallationConcurrencyAdmission(CONFIG, 2)).toEqual({ admit: false, reason: "concurrency_high" }); + }); + + it("denies above the limit", () => { + expect(evaluateInstallationConcurrencyAdmission(CONFIG, 5)).toEqual({ admit: false, reason: "concurrency_high" }); + }); +}); + +describe("installationConcurrencyDeferMs", () => { + it("is deterministic for the same seed", () => { + expect(installationConcurrencyDeferMs(CONFIG, "seed-a")).toBe(installationConcurrencyDeferMs(CONFIG, "seed-a")); + }); + + it("varies for different seeds", () => { + const values = new Set([ + installationConcurrencyDeferMs(CONFIG, "seed-a"), + installationConcurrencyDeferMs(CONFIG, "seed-b"), + installationConcurrencyDeferMs(CONFIG, "seed-c"), + ]); + expect(values.size).toBeGreaterThan(1); + }); + + it("is always >= config.deferMs (jitter only ever adds)", () => { + for (const seed of ["a", "b", "c", "d", "e"]) { + expect(installationConcurrencyDeferMs(CONFIG, seed)).toBeGreaterThanOrEqual(CONFIG.deferMs); + } + }); +}); + +describe("InstallationConcurrencyTracker", () => { + it("starts at 0 for an unknown key", () => { + const tracker = new InstallationConcurrencyTracker(); + expect(tracker.currentCount("installation:1")).toBe(0); + }); + + it("increments and decrements round-trip to 0", () => { + const tracker = new InstallationConcurrencyTracker(); + tracker.increment("installation:1"); + tracker.increment("installation:1"); + expect(tracker.currentCount("installation:1")).toBe(2); + tracker.decrement("installation:1"); + expect(tracker.currentCount("installation:1")).toBe(1); + tracker.decrement("installation:1"); + expect(tracker.currentCount("installation:1")).toBe(0); + }); + + it("never goes negative (floors at 0 on a decrement past 0)", () => { + const tracker = new InstallationConcurrencyTracker(); + tracker.decrement("installation:1"); + expect(tracker.currentCount("installation:1")).toBe(0); + tracker.increment("installation:1"); + tracker.decrement("installation:1"); + tracker.decrement("installation:1"); + expect(tracker.currentCount("installation:1")).toBe(0); + }); + + it("tracks independent keys without interference", () => { + const tracker = new InstallationConcurrencyTracker(); + tracker.increment("installation:1"); + tracker.increment("installation:1"); + tracker.increment("installation:2"); + expect(tracker.currentCount("installation:1")).toBe(2); + expect(tracker.currentCount("installation:2")).toBe(1); + tracker.decrement("installation:1"); + expect(tracker.currentCount("installation:1")).toBe(1); + expect(tracker.currentCount("installation:2")).toBe(1); + }); + + it("does not grow unboundedly across many increment/decrement cycles on the same key", () => { + const tracker = new InstallationConcurrencyTracker(); + for (let i = 0; i < 50; i += 1) { + tracker.increment("installation:1"); + tracker.decrement("installation:1"); + } + expect(tracker.currentCount("installation:1")).toBe(0); + // Internal Map size is not exposed publicly; the public contract (currentCount reads back to 0 after every + // cycle) is what the decrement-deletes-the-zero-entry implementation is FOR -- this proves the observable + // behavior a caller actually depends on, without reaching into a private field. + }); +}); + +describe("installationConcurrencyKeyForJob", () => { + const foregroundWebhook: JobMessage = { + type: "github-webhook", + deliveryId: "d1", + eventName: "pull_request", + payload: { installation: { id: 42 } }, + } as unknown as JobMessage; + + const foregroundRegate: JobMessage = { + type: "agent-regate-pr", + deliveryId: "d2", + repoFullName: "owner/repo", + pullNumber: 1, + installationId: 42, + } as unknown as JobMessage; + + const scheduledSweep: JobMessage = { + type: "agent-regate-sweep", + installationId: 42, + } as unknown as JobMessage; + + const backfillRepoSegment: JobMessage = { + type: "backfill-repo-segment", + installationId: 42, + } as unknown as JobMessage; + + const noInstallationIdType: JobMessage = { + type: "backfill-registered-repos", + repoFullName: "owner/repo", + } as unknown as JobMessage; + + it("returns null for a foreground github-webhook job (never a GitHub-budget-background job)", () => { + expect(installationConcurrencyKeyForJob(foregroundWebhook)).toBeNull(); + }); + + // REGRESSION (caught while writing this test): isGitHubBudgetBackgroundJob is true for a live (non-sweep, + // non-manual) agent-regate-pr job too -- it DOES draw GitHub rate-limit budget under this key -- so this + // pure resolver correctly returns a non-null key here. The "foreground jobs are never gated by this policy" + // guarantee lives at the pg-queue.ts/sqlite-queue.ts call site (an explicit !isForegroundJobPriority(...) + // guard before this function is ever called), NOT inside this key resolver -- see the queue backend tests. + it("returns the admission key for a foreground agent-regate-pr job (it DOES draw budget under this key -- foreground exclusion happens at the call site, not here)", () => { + expect(installationConcurrencyKeyForJob(foregroundRegate)).toBe("installation:42"); + }); + + it("returns the admission key for a GITHUB_BUDGET_BACKGROUND_TYPES job carrying installationId", () => { + expect(installationConcurrencyKeyForJob(scheduledSweep)).toBe("installation:42"); + expect(installationConcurrencyKeyForJob(backfillRepoSegment)).toBe("installation:42"); + }); + + it("returns null for a GITHUB_BUDGET_BACKGROUND_TYPES job whose payload carries no installationId", () => { + expect(installationConcurrencyKeyForJob(noInstallationIdType)).toBeNull(); + }); + + it("returns null for an unrelated job type not in GITHUB_BUDGET_BACKGROUND_TYPES", () => { + const other: JobMessage = { type: "notify-deliver" } as unknown as JobMessage; + expect(installationConcurrencyKeyForJob(other)).toBeNull(); + }); +}); diff --git a/test/unit/selfhost-pg-queue.test.ts b/test/unit/selfhost-pg-queue.test.ts index 75174437a7..1cfd215de6 100644 --- a/test/unit/selfhost-pg-queue.test.ts +++ b/test/unit/selfhost-pg-queue.test.ts @@ -3131,4 +3131,142 @@ describe("createPgQueue (durable #977)", () => { expect(signals.oldestLiveRunnableAgeMs).toBeNull(); }); }); + + describe("installation-concurrency admission (#selfhost-installation-concurrency)", () => { + const oldLimit = process.env.GITHUB_INSTALLATION_CONCURRENCY_LIMIT; + + afterEach(() => { + if (oldLimit === undefined) delete process.env.GITHUB_INSTALLATION_CONCURRENCY_LIMIT; + else process.env.GITHUB_INSTALLATION_CONCURRENCY_LIMIT = oldLimit; + }); + + // backfill-repo-segment (not agent-regate-sweep) is used as the background fixture throughout: unlike every + // other GITHUB_BUDGET_BACKGROUND_TYPES member, agent-regate-sweep's OWN row priority (8, PRIORITY_BY_TYPE) + // equals FOREGROUND_QUEUE_PRIORITY_FLOOR, so it is ALSO foreground-priority and therefore already exempt from + // this policy via the isForegroundJobPriority guard -- a genuinely background-priority (0) type is needed to + // actually exercise the limiter. + + it("a second concurrent background job for the SAME installation is deferred at the limit", async () => { + process.env.GITHUB_INSTALLATION_CONCURRENCY_LIMIT = "1"; + const m = makePool(); + m.enqueueJob("1", { type: "backfill-repo-segment", installationId: 42 }, 0, "backfill:42:a"); + // No job_key on the deferred row (a raw/legacy shape) -- exercises the `job.job_key ?? ""` jitter-seed + // fallback's nullish arm. + m.enqueueJob("2", { type: "backfill-repo-segment", installationId: 42 }, 0, null); + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + let started = 0; + const q = createPgQueue( + m.pool, + async () => { + started++; + await gate; + }, + { concurrency: 2, backgroundConcurrency: 2, pollIntervalMs: 100_000 }, + ); + await q.init(); + try { + q.start(); + for (let i = 0; i < 20 && started < 1; i += 1) await new Promise((r) => setTimeout(r, 10)); + // Give the second job's own pump loop a chance to claim and be evaluated too, while the first is still + // gated open -- only ONE of the two should ever have reached consume() at this point. + await new Promise((r) => setTimeout(r, 30)); + expect(started).toBe(1); + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("SET status='pending', run_after=GREATEST"), + expect.arrayContaining([expect.stringContaining("installation concurrency admission deferred: concurrency_high")]), + ); + expect(await renderMetrics()).toContain( + 'gittensory_jobs_installation_concurrency_deferred_by_reason_total{job_type="backfill-repo-segment",reason="concurrency_high"} 1', + ); + } finally { + release(); + await q.stop(); + } + }); + + it("a background job for a DIFFERENT installation is admitted concurrently with one already at its own limit", async () => { + process.env.GITHUB_INSTALLATION_CONCURRENCY_LIMIT = "1"; + const m = makePool(); + m.enqueueJob("1", { type: "backfill-repo-segment", installationId: 42 }, 0, "backfill:42:a"); + m.enqueueJob("2", { type: "backfill-repo-segment", installationId: 99 }, 0, "backfill:99:a"); + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + let concurrent = 0; + let maxConcurrent = 0; + const q = createPgQueue( + m.pool, + async () => { + concurrent++; + maxConcurrent = Math.max(maxConcurrent, concurrent); + await gate; + concurrent--; + }, + { concurrency: 2, backgroundConcurrency: 2, pollIntervalMs: 100_000 }, + ); + await q.init(); + try { + q.start(); + for (let i = 0; i < 20 && maxConcurrent < 2; i += 1) await new Promise((r) => setTimeout(r, 10)); + expect(maxConcurrent).toBe(2); + } finally { + release(); + await q.stop(); + } + }); + + it("never defers a foreground agent-regate-pr job regardless of installation in-flight count", async () => { + process.env.GITHUB_INSTALLATION_CONCURRENCY_LIMIT = "1"; + const m = makePool(); + m.enqueueJob("1", { type: "backfill-repo-segment", installationId: 42 }, 0, "backfill:42:a"); + // A real agent-regate-pr claim row carries priority 9 (AGENT_REGATE_PRIORITY) -- enqueueJob's fixed shape + // omits `priority` entirely, which would misrepresent this as background-priority (undefined/NaN reads as + // NOT foreground), defeating the exact thing this test verifies. enqueueResult lets the row be explicit. + m.enqueueResult({ + rows: [{ id: "2", payload: JSON.stringify(regateJob(42, 1630)), attempts: 0, job_key: "agent-regate-pr:jsonbored/gittensory#1630", priority: 9 }], + rowCount: 1, + }); + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const seen: string[] = []; + const q = createPgQueue( + m.pool, + async (j) => { + seen.push(typeOf(j)); + if (typeOf(j) === "backfill-repo-segment") await gate; + }, + { concurrency: 2, backgroundConcurrency: 2, pollIntervalMs: 100_000 }, + ); + await q.init(); + try { + q.start(); + for (let i = 0; i < 20 && seen.length < 2; i += 1) await new Promise((r) => setTimeout(r, 10)); + expect(seen).toContain("agent-regate-pr"); + } finally { + release(); + await q.stop(); + } + }); + + it("the tracker decrements on completion, so a subsequent job for the same installation is admitted again", async () => { + process.env.GITHUB_INSTALLATION_CONCURRENCY_LIMIT = "1"; + const m = makePool(); + m.enqueueJob("1", { type: "backfill-repo-segment", installationId: 42 }, 0, "backfill:42:a"); + const seen: string[] = []; + const q = createPgQueue(m.pool, async (j) => void seen.push(typeOf(j)), { backgroundConcurrency: 1 }); + await q.init(); + await q.drain(); + expect(seen).toEqual(["backfill-repo-segment"]); + + m.enqueueJob("2", { type: "backfill-repo-segment", installationId: 42 }, 0, "backfill:42:b"); + await q.drain(); + expect(seen).toEqual(["backfill-repo-segment", "backfill-repo-segment"]); + }); + }); }); diff --git a/test/unit/selfhost-sqlite-queue.test.ts b/test/unit/selfhost-sqlite-queue.test.ts index 79d648a48b..24f8727e34 100644 --- a/test/unit/selfhost-sqlite-queue.test.ts +++ b/test/unit/selfhost-sqlite-queue.test.ts @@ -3538,4 +3538,137 @@ describe("createSqliteQueue (durable #980)", () => { expect(await renderMetrics()).not.toContain("gittensory_jobs_maintenance_admission_deferred_by_reason_total"); }); }); + + describe("installation-concurrency admission (#selfhost-installation-concurrency)", () => { + const oldLimit = process.env.GITHUB_INSTALLATION_CONCURRENCY_LIMIT; + + afterEach(() => { + if (oldLimit === undefined) delete process.env.GITHUB_INSTALLATION_CONCURRENCY_LIMIT; + else process.env.GITHUB_INSTALLATION_CONCURRENCY_LIMIT = oldLimit; + }); + + // backfill-repo-segment (not agent-regate-sweep) is used as the background fixture throughout: unlike every + // other GITHUB_BUDGET_BACKGROUND_TYPES member, agent-regate-sweep's OWN row priority (8, PRIORITY_BY_TYPE) + // equals FOREGROUND_QUEUE_PRIORITY_FLOOR, so it is ALSO foreground-priority and therefore already exempt from + // this policy via the isForegroundJobPriority guard -- a genuinely background-priority (0) type is needed to + // actually exercise the limiter. q.binding.send(...) computes real priority via jobPriority(), so (unlike a + // hand-built mock row) every job here carries an authentic priority value. + + it("a second concurrent background job for the SAME installation is deferred at the limit", async () => { + process.env.GITHUB_INSTALLATION_CONCURRENCY_LIMIT = "1"; + const driver = makeDriver(); + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + let started = 0; + const q = createSqliteQueue( + driver, + async () => { + started++; + await gate; + }, + { concurrency: 2, backgroundConcurrency: 2, pollIntervalMs: 100_000 }, + ); + await q.binding.send({ type: "backfill-repo-segment", installationId: 42, repoFullName: "owner/a" } as unknown as JobMessage); + // The second (to-be-deferred) row is inserted directly with job_key=NULL (a raw/legacy shape) -- + // q.binding.send() would compute a real jobCoalesceKey for this type, which would never exercise the + // `job.job_key ?? ""` jitter-seed fallback's nullish arm. + driver.query( + `INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, is_maintenance) + VALUES (?, 'pending', 0, 0, ?, 0, NULL, 0)`, + [JSON.stringify({ type: "backfill-repo-segment", installationId: 42, repoFullName: "owner/b" }), Date.now()], + ); + try { + q.start(); + for (let i = 0; i < 20 && started < 1; i += 1) await new Promise((r) => setTimeout(r, 10)); + await new Promise((r) => setTimeout(r, 30)); + expect(started).toBe(1); + const row = driver.query( + "SELECT last_error FROM _selfhost_jobs WHERE status='pending' AND payload LIKE '%backfill-repo-segment%'", + [], + ).rows[0] as { last_error: string } | undefined; + expect(row?.last_error).toContain("installation concurrency admission deferred: concurrency_high"); + expect(await renderMetrics()).toContain( + 'gittensory_jobs_installation_concurrency_deferred_by_reason_total{job_type="backfill-repo-segment",reason="concurrency_high"} 1', + ); + } finally { + release(); + await q.stop(); + } + }); + + it("a background job for a DIFFERENT installation is admitted concurrently with one already at its own limit", async () => { + process.env.GITHUB_INSTALLATION_CONCURRENCY_LIMIT = "1"; + const driver = makeDriver(); + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + let concurrent = 0; + let maxConcurrent = 0; + const q = createSqliteQueue( + driver, + async () => { + concurrent++; + maxConcurrent = Math.max(maxConcurrent, concurrent); + await gate; + concurrent--; + }, + { concurrency: 2, backgroundConcurrency: 2, pollIntervalMs: 100_000 }, + ); + await q.binding.send({ type: "backfill-repo-segment", installationId: 42, repoFullName: "owner/a" } as unknown as JobMessage); + await q.binding.send({ type: "backfill-repo-segment", installationId: 99, repoFullName: "owner/b" } as unknown as JobMessage); + try { + q.start(); + for (let i = 0; i < 20 && maxConcurrent < 2; i += 1) await new Promise((r) => setTimeout(r, 10)); + expect(maxConcurrent).toBe(2); + } finally { + release(); + await q.stop(); + } + }); + + it("never defers a foreground agent-regate-pr job regardless of installation in-flight count", async () => { + process.env.GITHUB_INSTALLATION_CONCURRENCY_LIMIT = "1"; + const driver = makeDriver(); + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const seen: string[] = []; + const q = createSqliteQueue( + driver, + async (j) => { + seen.push(typeOf(j)); + if (typeOf(j) === "backfill-repo-segment") await gate; + }, + { concurrency: 2, backgroundConcurrency: 2, pollIntervalMs: 100_000 }, + ); + await q.binding.send({ type: "backfill-repo-segment", installationId: 42, repoFullName: "owner/a" } as unknown as JobMessage); + await q.binding.send(regateJob(42, 1630)); + try { + q.start(); + for (let i = 0; i < 20 && seen.length < 2; i += 1) await new Promise((r) => setTimeout(r, 10)); + expect(seen).toContain("agent-regate-pr"); + } finally { + release(); + await q.stop(); + } + }); + + it("the tracker decrements on completion, so a subsequent job for the same installation is admitted again", async () => { + process.env.GITHUB_INSTALLATION_CONCURRENCY_LIMIT = "1"; + const driver = makeDriver(); + const seen: string[] = []; + const q = createSqliteQueue(driver, async (m) => void seen.push(typeOf(m)), { backgroundConcurrency: 1 }); + await q.binding.send({ type: "backfill-repo-segment", installationId: 42, repoFullName: "owner/a" } as unknown as JobMessage); + await q.drain(); + expect(seen).toEqual(["backfill-repo-segment"]); + + await q.binding.send({ type: "backfill-repo-segment", installationId: 42, repoFullName: "owner/b" } as unknown as JobMessage); + await q.drain(); + expect(seen).toEqual(["backfill-repo-segment", "backfill-repo-segment"]); + }); + }); }); From 206838172caf0c96417643cdac58cbf88df10ace Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Sat, 4 Jul 2026 01:18:00 -0700 Subject: [PATCH 2/2] fix(selfhost): expose all 3 concurrency-admission env knobs, cover the raced-defer branch The env-reference generator only recognized envString(container, "NAME") calls, so every env var read via the widely-used parsePositiveIntEnv("NAME", opts) helper -- including this PR's own GITHUB_INSTALLATION_CONCURRENCY_LIMIT and _DEFER_MS -- was silently missing from the operator-facing self-host env reference. Teach the generator to also recognize that call shape. Also add the sqlite-queue installation-concurrency defer path's missing "UPDATE changed no rows" branch test, mirroring the existing maintenance-admission test for the identical UPDATE shape. --- .../src/lib/selfhost-env-reference.ts | 80 +++++++++++++++++++ scripts/gen-selfhost-env-reference.mjs | 17 ++++ .../selfhost-env-reference-script.test.ts | 2 + test/unit/selfhost-sqlite-queue.test.ts | 46 +++++++++++ 4 files changed, 145 insertions(+) diff --git a/apps/gittensory-ui/src/lib/selfhost-env-reference.ts b/apps/gittensory-ui/src/lib/selfhost-env-reference.ts index 5540d01197..074306180a 100644 --- a/apps/gittensory-ui/src/lib/selfhost-env-reference.ts +++ b/apps/gittensory-ui/src/lib/selfhost-env-reference.ts @@ -97,10 +97,22 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [ name: "DISCORD_WEBHOOK_URL", firstReference: "src/services/notify-discord.ts:78", }, + { + name: "FOREGROUND_LIVENESS_CHECK_INTERVAL_MS", + firstReference: "src/selfhost/foreground-liveness.ts:52", + }, { name: "FOREGROUND_LIVENESS_ENABLED", firstReference: "src/selfhost/foreground-liveness.ts:41", }, + { + name: "FOREGROUND_LIVENESS_MAX_DEFER_MS", + firstReference: "src/selfhost/foreground-liveness.ts:51", + }, + { + name: "FOREGROUND_LIVENESS_MAX_RELEASE_PER_SWEEP", + firstReference: "src/selfhost/foreground-liveness.ts:53", + }, { name: "GITHUB_APP_ID", firstReference: "src/selfhost/orb-collector.ts:59", @@ -113,10 +125,18 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [ name: "GITHUB_CACHE_TTL_SECONDS", firstReference: "src/server.ts:508", }, + { + name: "GITHUB_INSTALLATION_CONCURRENCY_DEFER_MS", + firstReference: "src/selfhost/installation-concurrency-admission.ts:47", + }, { name: "GITHUB_INSTALLATION_CONCURRENCY_ENABLED", firstReference: "src/selfhost/installation-concurrency-admission.ts:34", }, + { + name: "GITHUB_INSTALLATION_CONCURRENCY_LIMIT", + firstReference: "src/selfhost/installation-concurrency-admission.ts:43", + }, { name: "GITTENSORY_REPO_CONFIG_DIR", firstReference: "src/server.ts:288", @@ -129,10 +149,38 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [ name: "HOME", firstReference: "src/selfhost/ai.ts:302", }, + { + name: "MAINTENANCE_ADMISSION_DEFER_MS", + firstReference: "src/selfhost/maintenance-admission.ts:171", + }, + { + name: "MAINTENANCE_ADMISSION_DRAIN_AGE_MS", + firstReference: "src/selfhost/maintenance-admission.ts:145", + }, { name: "MAINTENANCE_ADMISSION_ENABLED", firstReference: "src/selfhost/maintenance-admission.ts:126", }, + { + name: "MAINTENANCE_ADMISSION_MAX_BACKLOG_CONVERGENCE_PENDING", + firstReference: "src/selfhost/maintenance-admission.ts:167", + }, + { + name: "MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS", + firstReference: "src/selfhost/maintenance-admission.ts:141", + }, + { + name: "MAINTENANCE_ADMISSION_MAX_LIVE_AGE_MS", + firstReference: "src/selfhost/maintenance-admission.ts:155", + }, + { + name: "MAINTENANCE_ADMISSION_MAX_LIVE_PENDING", + firstReference: "src/selfhost/maintenance-admission.ts:151", + }, + { + name: "MAINTENANCE_ADMISSION_MAX_PENDING", + firstReference: "src/selfhost/maintenance-admission.ts:159", + }, { name: "MIGRATIONS_DIR", firstReference: "src/server.ts:392", @@ -241,6 +289,10 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [ name: "OTEL_TRACES_SAMPLER_ARG", firstReference: "src/selfhost/otel.ts:76", }, + { + name: "PGPOOL_MAX", + firstReference: "src/selfhost/queue-common.ts:710", + }, { name: "PGVECTOR_ENABLED", firstReference: "src/server.ts:229", @@ -269,6 +321,18 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [ name: "QUEUE_BACKGROUND_CONCURRENCY", firstReference: "src/selfhost/queue-common.ts:130", }, + { + name: "QUEUE_CONCURRENCY", + firstReference: "src/selfhost/pg-queue.ts:285", + }, + { + name: "QUEUE_DEAD_LETTER_AUTO_RETRY_MAX_EXTRA_ATTEMPTS", + firstReference: "src/selfhost/queue-common.ts:718", + }, + { + name: "QUEUE_STARTUP_JITTER_MIN_JOBS", + firstReference: "src/selfhost/queue-common.ts:699", + }, { name: "REDIS_URL", firstReference: "src/selfhost/preflight.ts:144", @@ -345,15 +409,27 @@ export const SELFHOST_ENV_REFERENCE_MARKDOWN = [ "| `DATABASE_URL` | `src/selfhost/preflight.ts:201` |", "| `DISCORD_REPO_WEBHOOKS` | `src/services/notify-discord.ts:41` |", "| `DISCORD_WEBHOOK_URL` | `src/services/notify-discord.ts:78` |", + "| `FOREGROUND_LIVENESS_CHECK_INTERVAL_MS` | `src/selfhost/foreground-liveness.ts:52` |", "| `FOREGROUND_LIVENESS_ENABLED` | `src/selfhost/foreground-liveness.ts:41` |", + "| `FOREGROUND_LIVENESS_MAX_DEFER_MS` | `src/selfhost/foreground-liveness.ts:51` |", + "| `FOREGROUND_LIVENESS_MAX_RELEASE_PER_SWEEP` | `src/selfhost/foreground-liveness.ts:53` |", "| `GITHUB_APP_ID` | `src/selfhost/orb-collector.ts:59` |", "| `GITHUB_APP_PRIVATE_KEY` | `src/selfhost/orb-collector.ts:166` |", "| `GITHUB_CACHE_TTL_SECONDS` | `src/server.ts:508` |", + "| `GITHUB_INSTALLATION_CONCURRENCY_DEFER_MS` | `src/selfhost/installation-concurrency-admission.ts:47` |", "| `GITHUB_INSTALLATION_CONCURRENCY_ENABLED` | `src/selfhost/installation-concurrency-admission.ts:34` |", + "| `GITHUB_INSTALLATION_CONCURRENCY_LIMIT` | `src/selfhost/installation-concurrency-admission.ts:43` |", "| `GITTENSORY_REPO_CONFIG_DIR` | `src/server.ts:288` |", "| `GITTENSORY_VERSION` | `src/selfhost/otel.ts:62` |", "| `HOME` | `src/selfhost/ai.ts:302` |", + "| `MAINTENANCE_ADMISSION_DEFER_MS` | `src/selfhost/maintenance-admission.ts:171` |", + "| `MAINTENANCE_ADMISSION_DRAIN_AGE_MS` | `src/selfhost/maintenance-admission.ts:145` |", "| `MAINTENANCE_ADMISSION_ENABLED` | `src/selfhost/maintenance-admission.ts:126` |", + "| `MAINTENANCE_ADMISSION_MAX_BACKLOG_CONVERGENCE_PENDING` | `src/selfhost/maintenance-admission.ts:167` |", + "| `MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS` | `src/selfhost/maintenance-admission.ts:141` |", + "| `MAINTENANCE_ADMISSION_MAX_LIVE_AGE_MS` | `src/selfhost/maintenance-admission.ts:155` |", + "| `MAINTENANCE_ADMISSION_MAX_LIVE_PENDING` | `src/selfhost/maintenance-admission.ts:151` |", + "| `MAINTENANCE_ADMISSION_MAX_PENDING` | `src/selfhost/maintenance-admission.ts:159` |", "| `MIGRATIONS_DIR` | `src/server.ts:392` |", "| `OBSERVABILITY_SMOKE_POLL_MS` | `scripts/smoke-observability-traces.mjs:8` |", "| `OBSERVABILITY_SMOKE_TIMEOUT_MS` | `scripts/smoke-observability-traces.mjs:6` |", @@ -381,6 +457,7 @@ export const SELFHOST_ENV_REFERENCE_MARKDOWN = [ "| `OTEL_TRACES_EXPORTER` | `src/selfhost/otel.ts:40` |", "| `OTEL_TRACES_SAMPLER` | `src/selfhost/otel.ts:74` |", "| `OTEL_TRACES_SAMPLER_ARG` | `src/selfhost/otel.ts:76` |", + "| `PGPOOL_MAX` | `src/selfhost/queue-common.ts:710` |", "| `PGVECTOR_ENABLED` | `src/server.ts:229` |", "| `PORT` | `src/server.ts:715` |", "| `PUBLIC_API_ORIGIN` | `src/selfhost/preflight.ts:192` |", @@ -388,6 +465,9 @@ export const SELFHOST_ENV_REFERENCE_MARKDOWN = [ "| `QDRANT_DIM` | `src/selfhost/qdrant-vectorize.ts:71` |", "| `QDRANT_URL` | `src/server.ts:527` |", "| `QUEUE_BACKGROUND_CONCURRENCY` | `src/selfhost/queue-common.ts:130` |", + "| `QUEUE_CONCURRENCY` | `src/selfhost/pg-queue.ts:285` |", + "| `QUEUE_DEAD_LETTER_AUTO_RETRY_MAX_EXTRA_ATTEMPTS` | `src/selfhost/queue-common.ts:718` |", + "| `QUEUE_STARTUP_JITTER_MIN_JOBS` | `src/selfhost/queue-common.ts:699` |", "| `REDIS_URL` | `src/selfhost/preflight.ts:144` |", "| `REVIEW_AUDIT_DIR` | `src/server.ts:572` |", "| `SELFHOST_BUNDLE_ALL` | `scripts/build-selfhost.mjs:13` |", diff --git a/scripts/gen-selfhost-env-reference.mjs b/scripts/gen-selfhost-env-reference.mjs index 20114e275b..76c8406007 100644 --- a/scripts/gen-selfhost-env-reference.mjs +++ b/scripts/gen-selfhost-env-reference.mjs @@ -72,6 +72,8 @@ function collectEnvReads(source, fileName) { } } else if (ts.isCallExpression(node) && isStaticEnvHelperCall(node)) { addRead(node.arguments[1].text, node.arguments[1]); + } else if (ts.isCallExpression(node) && isProcessEnvNameHelperCall(node)) { + addRead(node.arguments[0].text, node.arguments[0]); } ts.forEachChild(node, visit); }; @@ -89,6 +91,21 @@ function isStaticEnvHelperCall(node) { ); } +// Some self-host helpers read `process.env` internally by name rather than taking an env container argument -- +// e.g. `parsePositiveIntEnv("QUEUE_CONCURRENCY", { min: 1, fallback: 4 })`. Recognized separately from +// isStaticEnvHelperCall above (envString) because these take the var NAME as arg[0], not arg[1] after a +// container. +const PROCESS_ENV_NAME_HELPERS = new Set(["parsePositiveIntEnv"]); + +function isProcessEnvNameHelperCall(node) { + return ( + ts.isIdentifier(node.expression) && + PROCESS_ENV_NAME_HELPERS.has(node.expression.text) && + node.arguments.length >= 1 && + ts.isStringLiteralLike(node.arguments[0]) + ); +} + function bindingElementName(element) { const candidate = element.propertyName ?? element.name; if (ts.isIdentifier(candidate) || ts.isStringLiteralLike(candidate)) return candidate.text; diff --git a/test/unit/selfhost-env-reference-script.test.ts b/test/unit/selfhost-env-reference-script.test.ts index 23f3957a49..1e57cf62ab 100644 --- a/test/unit/selfhost-env-reference-script.test.ts +++ b/test/unit/selfhost-env-reference-script.test.ts @@ -52,6 +52,7 @@ function fixtureRoot(): string { "const serviceOnly = process.env.SERVICE_ONLY;", "const helperOnly = envString(env, 'SERVICE_HELPER_ONLY');", "const casted = (env as unknown as Record).CASTED_ONLY;", + "const parsedInt = parsePositiveIntEnv('PARSED_INT_ONLY', { min: 1, fallback: 4 });", "", ].join("\n"), ); @@ -73,6 +74,7 @@ describe("gen-selfhost-env-reference (#2081)", () => { { name: "OBJECT_ALIASED", firstReference: "src/selfhost/a.ts:9" }, { name: "OBJECT_BRACKET", firstReference: "src/selfhost/a.ts:8" }, { name: "OBJECT_DESTRUCTURED", firstReference: "src/selfhost/a.ts:9" }, + { name: "PARSED_INT_ONLY", firstReference: "src/services/notify-discord.ts:4" }, { name: "SECOND", firstReference: "src/selfhost/a.ts:2" }, { name: "SERVER_ONLY", firstReference: "src/server.ts:1" }, { name: "SERVICE_HELPER_ONLY", firstReference: "src/services/notify-discord.ts:2" }, diff --git a/test/unit/selfhost-sqlite-queue.test.ts b/test/unit/selfhost-sqlite-queue.test.ts index 24f8727e34..4796a10da2 100644 --- a/test/unit/selfhost-sqlite-queue.test.ts +++ b/test/unit/selfhost-sqlite-queue.test.ts @@ -3670,5 +3670,51 @@ describe("createSqliteQueue (durable #980)", () => { await q.drain(); expect(seen).toEqual(["backfill-repo-segment", "backfill-repo-segment"]); }); + + it("skips the installation-concurrency-deferred metric when the defer update changes no rows", async () => { + process.env.GITHUB_INSTALLATION_CONCURRENCY_LIMIT = "1"; + const base = makeDriver(); + // The row raced out from under this UPDATE (already claimed/mutated by another path) -- mirrors the + // maintenance-admission "changes no rows" test above, which intercepts the identical UPDATE shape. + const driver = { + exec: base.exec.bind(base), + query: vi.fn((sql: string, params: unknown[]) => { + if (sql.includes("SET status='pending', run_after=max(run_after, ?), last_error=coalesce(last_error, ?)")) { + return { rows: [], changes: 0 }; + } + return base.query(sql, params); + }), + } as ReturnType; + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + let started = 0; + const q = createSqliteQueue( + driver, + async () => { + started++; + await gate; + }, + { concurrency: 2, backgroundConcurrency: 2, pollIntervalMs: 100_000 }, + ); + await q.binding.send({ type: "backfill-repo-segment", installationId: 42, repoFullName: "owner/a" } as unknown as JobMessage); + driver.query( + `INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, is_maintenance) + VALUES (?, 'pending', 0, 0, ?, 0, NULL, 0)`, + [JSON.stringify({ type: "backfill-repo-segment", installationId: 42, repoFullName: "owner/b" }), Date.now()], + ); + try { + q.start(); + for (let i = 0; i < 20 && started < 1; i += 1) await new Promise((r) => setTimeout(r, 10)); + await new Promise((r) => setTimeout(r, 30)); + expect(started).toBe(1); + expect(await renderMetrics()).not.toContain("gittensory_jobs_installation_concurrency_deferred_total"); + expect(await renderMetrics()).not.toContain("gittensory_jobs_installation_concurrency_deferred_by_reason_total"); + } finally { + release(); + await q.stop(); + } + }); }); });