diff --git a/.env.example b/.env.example index 21c8af3839..fa37e5a36d 100644 --- a/.env.example +++ b/.env.example @@ -269,6 +269,13 @@ REDIS_URL=redis://redis:6379 # REQUIRED for the self-host review # TRICKLE (MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS) force-admits it once it's waited long enough regardless of # pressure, so sustained load can slow maintenance down but never starve it forever. All defaults are sane for # a small single-node box; every value is optional. +# DRAIN (MAINTENANCE_ADMISSION_DRAIN_AGE_MS, #selfhost-maintenance-self-pin): a second, much shorter age escape +# scoped ONLY to `maintenance_pending_high` -- the aggregate lane-backlog check has no feedback loop back to the +# count as jobs individually age out via the trickle above, so without this a backed-up lane can deny every +# claim for hours even though the trickle exists. The drain lets the OLDEST jobs in that same backlog through +# well before the full trickle ceiling, so the backlog can actually shrink (further bounded by +# QUEUE_BACKGROUND_CONCURRENCY). Still fully blocked by host_load_high -- draining more work onto an overloaded +# box is exactly what that check exists to prevent. # MAINTENANCE_ADMISSION_ENABLED=true # set false/0/off to fully disable this policy (old always-run behavior) # MAINTENANCE_ADMISSION_MAX_LIVE_PENDING=5 # defer maintenance once this many live (webhook/regate) jobs are queued # MAINTENANCE_ADMISSION_MAX_LIVE_AGE_MS=120000 # defer maintenance once the oldest live job has waited this long (2m) @@ -276,6 +283,7 @@ REDIS_URL=redis://redis:6379 # REQUIRED for the self-host review # MAINTENANCE_ADMISSION_MAX_HOST_LOAD=1.5 # defer once 1-min load average per CPU core exceeds this (best-effort; see host-pressure.ts) # MAINTENANCE_ADMISSION_DEFER_MS=180000 # base defer duration on denial, before jitter (3m) # MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS=14400000 # trickle ceiling: force-admit a maintenance job that has waited this long (4h) +# MAINTENANCE_ADMISSION_DRAIN_AGE_MS=600000 # drain ceiling: admit a job despite a backed-up lane once it has waited this long (10m); clamped to MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS # --- 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 38464d30fc..12735e807d 100644 --- a/apps/gittensory-ui/src/lib/selfhost-env-reference.ts +++ b/apps/gittensory-ui/src/lib/selfhost-env-reference.ts @@ -11,11 +11,11 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [ }, { name: "AI_EMBED_API_KEY", - firstReference: "src/server.ts:422", + firstReference: "src/server.ts:423", }, { name: "AI_EMBED_BASE_URL", - firstReference: "src/server.ts:419", + firstReference: "src/server.ts:420", }, { name: "AI_EMBED_MODEL", @@ -43,7 +43,7 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [ }, { name: "BACKUP_ACKNOWLEDGED", - firstReference: "src/server.ts:361", + firstReference: "src/server.ts:362", }, { name: "BROWSER_WS_ENDPOINT", @@ -79,11 +79,11 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [ }, { name: "CRON_INTERVAL_MS", - firstReference: "src/server.ts:865", + firstReference: "src/server.ts:872", }, { name: "DATABASE_PATH", - firstReference: "src/server.ts:244", + firstReference: "src/server.ts:245", }, { name: "DATABASE_URL", @@ -107,11 +107,11 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [ }, { name: "GITHUB_CACHE_TTL_SECONDS", - firstReference: "src/server.ts:490", + firstReference: "src/server.ts:491", }, { name: "GITTENSORY_REPO_CONFIG_DIR", - firstReference: "src/server.ts:278", + firstReference: "src/server.ts:279", }, { name: "GITTENSORY_VERSION", @@ -123,11 +123,11 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [ }, { name: "MAINTENANCE_ADMISSION_ENABLED", - firstReference: "src/selfhost/maintenance-admission.ts:83", + firstReference: "src/selfhost/maintenance-admission.ts:99", }, { name: "MIGRATIONS_DIR", - firstReference: "src/server.ts:374", + firstReference: "src/server.ts:375", }, { name: "OBSERVABILITY_SMOKE_POLL_MS", @@ -187,7 +187,7 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [ }, { name: "ORB_BROKER_URL", - firstReference: "src/server.ts:914", + firstReference: "src/server.ts:921", }, { name: "ORB_COLLECTOR_TOKEN", @@ -203,7 +203,7 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [ }, { name: "ORB_RELAY_MODE", - firstReference: "src/server.ts:916", + firstReference: "src/server.ts:923", }, { name: "OTEL_EXPORTER_OTLP_ENDPOINT", @@ -235,11 +235,11 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [ }, { name: "PGVECTOR_ENABLED", - firstReference: "src/server.ts:224", + firstReference: "src/server.ts:225", }, { name: "PORT", - firstReference: "src/server.ts:664", + firstReference: "src/server.ts:671", }, { name: "PUBLIC_API_ORIGIN", @@ -255,7 +255,7 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [ }, { name: "QDRANT_URL", - firstReference: "src/server.ts:509", + firstReference: "src/server.ts:510", }, { name: "QUEUE_BACKGROUND_CONCURRENCY", @@ -267,7 +267,7 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [ }, { name: "REVIEW_AUDIT_DIR", - firstReference: "src/server.ts:554", + firstReference: "src/server.ts:555", }, { name: "SELFHOST_BUNDLE_ALL", @@ -303,7 +303,7 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [ }, { name: "SETUP_OUTPUT_PATH", - firstReference: "src/server.ts:781", + firstReference: "src/server.ts:788", }, ]; @@ -311,15 +311,15 @@ export const SELFHOST_ENV_REFERENCE_MARKDOWN = [ "| Name | First reference |", "| --- | --- |", "| `AI_COMBINE` | `src/selfhost/ai.ts:930` |", - "| `AI_EMBED_API_KEY` | `src/server.ts:422` |", - "| `AI_EMBED_BASE_URL` | `src/server.ts:419` |", + "| `AI_EMBED_API_KEY` | `src/server.ts:423` |", + "| `AI_EMBED_BASE_URL` | `src/server.ts:420` |", "| `AI_EMBED_MODEL` | `src/selfhost/ai.ts:826` |", "| `AI_ON_MERGE` | `src/selfhost/ai.ts:932` |", "| `AI_PROVIDER` | `src/selfhost/ai-config.ts:43` |", "| `ANTHROPIC_AI_BASE_URL` | `src/selfhost/ai.ts:830` |", "| `ANTHROPIC_AI_MODEL` | `src/selfhost/ai.ts:57` |", "| `ANTHROPIC_API_KEY` | `src/selfhost/ai.ts:829` |", - "| `BACKUP_ACKNOWLEDGED` | `src/server.ts:361` |", + "| `BACKUP_ACKNOWLEDGED` | `src/server.ts:362` |", "| `BROWSER_WS_ENDPOINT` | `src/selfhost/stubs/puppeteer.ts:11` |", "| `CLAUDE_AI_EFFORT` | `src/selfhost/ai.ts:108` |", "| `CLAUDE_AI_MODEL` | `src/selfhost/ai.ts:49` |", @@ -328,19 +328,19 @@ export const SELFHOST_ENV_REFERENCE_MARKDOWN = [ "| `CODEX_AI_MODEL` | `src/selfhost/ai.ts:53` |", "| `CODEX_AI_TIMEOUT_MS` | `src/selfhost/ai.ts:112` |", "| `CODEX_HOME` | `src/selfhost/ai.ts:274` |", - "| `CRON_INTERVAL_MS` | `src/server.ts:865` |", - "| `DATABASE_PATH` | `src/server.ts:244` |", + "| `CRON_INTERVAL_MS` | `src/server.ts:872` |", + "| `DATABASE_PATH` | `src/server.ts:245` |", "| `DATABASE_URL` | `src/selfhost/preflight.ts:201` |", "| `DISCORD_REPO_WEBHOOKS` | `src/selfhost/discord-notify.ts:31` |", "| `DISCORD_WEBHOOK_URL` | `src/selfhost/discord-notify.ts:40` |", "| `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:490` |", - "| `GITTENSORY_REPO_CONFIG_DIR` | `src/server.ts:278` |", + "| `GITHUB_CACHE_TTL_SECONDS` | `src/server.ts:491` |", + "| `GITTENSORY_REPO_CONFIG_DIR` | `src/server.ts:279` |", "| `GITTENSORY_VERSION` | `src/selfhost/health.ts:29` |", "| `HOME` | `src/selfhost/ai.ts:274` |", - "| `MAINTENANCE_ADMISSION_ENABLED` | `src/selfhost/maintenance-admission.ts:83` |", - "| `MIGRATIONS_DIR` | `src/server.ts:374` |", + "| `MAINTENANCE_ADMISSION_ENABLED` | `src/selfhost/maintenance-admission.ts:99` |", + "| `MIGRATIONS_DIR` | `src/server.ts:375` |", "| `OBSERVABILITY_SMOKE_POLL_MS` | `scripts/smoke-observability-traces.mjs:8` |", "| `OBSERVABILITY_SMOKE_TIMEOUT_MS` | `scripts/smoke-observability-traces.mjs:6` |", "| `OLLAMA_AI_API_KEY` | `src/selfhost/ai.ts:823` |", @@ -355,11 +355,11 @@ export const SELFHOST_ENV_REFERENCE_MARKDOWN = [ "| `ORB_AIR_GAP` | `src/selfhost/orb-collector.ts:161` |", "| `ORB_ANONYMIZE` | `src/selfhost/orb-collector.ts:174` |", "| `ORB_APP_ID` | `src/selfhost/orb-collector.ts:59` |", - "| `ORB_BROKER_URL` | `src/server.ts:914` |", + "| `ORB_BROKER_URL` | `src/server.ts:921` |", "| `ORB_COLLECTOR_TOKEN` | `src/selfhost/orb-collector.ts:205` |", "| `ORB_COLLECTOR_URL` | `src/selfhost/orb-collector.ts:172` |", "| `ORB_ENROLLMENT_SECRET` | `src/selfhost/orb-collector.ts:165` |", - "| `ORB_RELAY_MODE` | `src/server.ts:916` |", + "| `ORB_RELAY_MODE` | `src/server.ts:923` |", "| `OTEL_EXPORTER_OTLP_ENDPOINT` | `src/selfhost/otel.ts:47` |", "| `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` | `src/selfhost/otel.ts:45` |", "| `OTEL_SERVICE_ENVIRONMENT` | `src/selfhost/otel.ts:60` |", @@ -367,15 +367,15 @@ 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` |", - "| `PGVECTOR_ENABLED` | `src/server.ts:224` |", - "| `PORT` | `src/server.ts:664` |", + "| `PGVECTOR_ENABLED` | `src/server.ts:225` |", + "| `PORT` | `src/server.ts:671` |", "| `PUBLIC_API_ORIGIN` | `src/selfhost/preflight.ts:192` |", "| `QDRANT_API_KEY` | `src/selfhost/qdrant-vectorize.ts:50` |", "| `QDRANT_DIM` | `src/selfhost/qdrant-vectorize.ts:71` |", - "| `QDRANT_URL` | `src/server.ts:509` |", + "| `QDRANT_URL` | `src/server.ts:510` |", "| `QUEUE_BACKGROUND_CONCURRENCY` | `src/selfhost/queue-common.ts:102` |", "| `REDIS_URL` | `src/selfhost/preflight.ts:144` |", - "| `REVIEW_AUDIT_DIR` | `src/server.ts:554` |", + "| `REVIEW_AUDIT_DIR` | `src/server.ts:555` |", "| `SELFHOST_BUNDLE_ALL` | `scripts/build-selfhost.mjs:13` |", "| `SELFHOST_SERVICE` | `scripts/smoke-observability-traces.mjs:5` |", "| `SELFHOST_SETUP_TOKEN` | `src/selfhost/preflight.ts:186` |", @@ -384,5 +384,5 @@ export const SELFHOST_ENV_REFERENCE_MARKDOWN = [ "| `SENTRY_RELEASE` | `src/selfhost/otel.ts:62` |", "| `SENTRY_SERVER_NAME` | `src/selfhost/sentry.ts:383` |", "| `SENTRY_TRACES_SAMPLE_RATE` | `src/selfhost/sentry.ts:171` |", - "| `SETUP_OUTPUT_PATH` | `src/server.ts:781` |", + "| `SETUP_OUTPUT_PATH` | `src/server.ts:788` |", ].join("\n"); diff --git a/grafana/dashboards/gittensory.json b/grafana/dashboards/gittensory.json index 1a5fdd0646..544f87b633 100644 --- a/grafana/dashboards/gittensory.json +++ b/grafana/dashboards/gittensory.json @@ -2396,6 +2396,33 @@ "refId": "A" } ] + }, + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "custom": { "lineWidth": 2, "fillOpacity": 10 }, + "unit": "ops" + } + }, + "gridPos": { "h": 8, "w": 24, "x": 0, "y": 149 }, + "id": 137, + "options": { + "legend": { "calcs": ["lastNotNull"], "displayMode": "table", "placement": "bottom" }, + "tooltip": { "mode": "multi", "sort": "desc" } + }, + "title": "Maintenance Admission Granted Under Pressure (trickle/drain, #selfhost-maintenance-self-pin)", + "description": "A maintenance job admitted DESPITE active pressure -- the trickle (maxDeferAgeMs) or drain (maintenanceDrainAgeMs) age escapes firing. Healthy when this tracks alongside a high 'Maintenance Queue Pending' (the backlog is actively draining); zero activity here while that gauge stays high for a long stretch means the escapes aren't reaching their age thresholds yet.", + "type": "timeseries", + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "expr": "sum by (reason, job_type) (rate(gittensory_jobs_maintenance_admission_granted_under_pressure_total[5m])) or vector(0)", + "legendFormat": "{{reason}} {{job_type}}", + "refId": "A" + } + ] } ], "refresh": "30s", diff --git a/prometheus/rules/alerts.yml b/prometheus/rules/alerts.yml index a98ac2a529..fc57e83d9e 100644 --- a/prometheus/rules/alerts.yml +++ b/prometheus/rules/alerts.yml @@ -134,11 +134,13 @@ groups: runbook: "Open the Runtime Pressure & Maintenance row. If gittensory_host_load_avg1_per_core is elevated, a co-located CI runner or other host process is starving the app -- see docker-compose.yml's runner isolation guidance." - alert: GittensoryMaintenanceStarved - # Maintenance admission (maintenance-admission.ts) force-admits a maintenance job once it has waited - # MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS (default 4h) regardless of pressure -- so under correct - # operation this should never sit much past that trickle ceiling. A value well beyond it means - # either the trickle isn't triggering (a bug) or the host is so overloaded even the trickle-forced - # job can't be claimed/processed. + # Maintenance admission (maintenance-admission.ts) has TWO age escapes (#selfhost-maintenance-self-pin): + # a short maintenanceDrainAgeMs trickle that lets old jobs through even while `maintenance_pending_high` + # is breached (bounded further by the queue's own background concurrency cap), and the ultimate + # MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS (default 4h) force-admit that applies regardless of pressure -- + # so under correct operation this should never sit much past that outer ceiling, and in practice the + # drain escape should keep it well under it. A value well beyond 4h means either an escape isn't + # triggering (a bug) or the host is so overloaded even a force-admitted job can't be claimed/processed. expr: gittensory_queue_oldest_maintenance_pending_age_seconds > 21600 for: 15m labels: @@ -146,7 +148,7 @@ groups: annotations: summary: "gittensory maintenance work has not run in over 6h" description: "The oldest maintenance-lane queue job has been pending for {{ $value | printf \"%.0f\" }}s, past the default trickle ceiling. Contributor evidence, RAG indexing, drift scans, and similar sweeps are stale." - runbook: "Check gittensory_jobs_maintenance_admission_deferred_by_reason_total for the dominant defer reason, and confirm MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS wasn't raised. Sustained host_load_high suggests the box itself (not just this app) is overloaded." + runbook: "Check gittensory_jobs_maintenance_admission_deferred_by_reason_total for the dominant defer reason and gittensory_jobs_maintenance_admission_granted_under_pressure_total to confirm the escapes are actually firing, and confirm MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS / MAINTENANCE_ADMISSION_DRAIN_AGE_MS weren't raised. Sustained host_load_high suggests the box itself (not just this app) is overloaded -- host_load_high also blocks the drain escape specifically, see maintenance-admission.ts." # ── GitHub API budget / queue admission pressure ────────────────────────── - name: gittensory-github-rate-limits diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 1d8e18a339..67c803d367 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -890,7 +890,15 @@ export async function processJob(env: Env, message: JobMessage): Promise { await executeAgentRun(env, message.runId); return; case "notify-evaluate": { - const deliveries = await evaluateNotificationEvent(env, message.event); + // Legacy payload compat: a row enqueued before the batched-events deploy (#selfhost-maintenance-self-pin) + // still carries the OLD singular `event` field on disk, not `events` -- a rolling deploy can process such + // a row after the new code ships, so normalize both shapes rather than assuming every persisted payload + // already matches the current type (which only the type checker, not the durable queue, enforces). + const legacyMessage = message as unknown as { events?: DetectedNotificationEvent[]; event?: DetectedNotificationEvent }; + const events = Array.isArray(legacyMessage.events) ? legacyMessage.events : legacyMessage.event ? [legacyMessage.event] : []; + const deliveries = ( + await mapWithConcurrency(events, NOTIFY_EVALUATE_EVENT_CONCURRENCY, (event) => evaluateNotificationEvent(env, event)) + ).flat(); await Promise.all( deliveries.map((delivery) => env.JOBS.send({ @@ -4144,6 +4152,29 @@ async function countLiveOpenWithConcurrencyUntil( return confirmedOpenCount; } +// A batched notify-evaluate job (#selfhost-maintenance-self-pin) can carry many events from one webhook (a +// popular newly-opened issue can have dozens of watchers) -- an unbounded Promise.all over all of them would +// let a single job spend as many concurrent DB/eval calls as it likes, bypassing the queue's own +// backgroundConcurrency cap (which defaults to 1) entirely from inside one job's execution. Bounded worker-pool +// fan-out, same shape as GLOBAL_OPEN_ITEM_LIVE_CHECK_CONCURRENCY above. +const NOTIFY_EVALUATE_EVENT_CONCURRENCY = 5; + +async function mapWithConcurrency(items: T[], concurrency: number, mapper: (item: T) => Promise): Promise { + const results: R[] = new Array(items.length); + let nextIndex = 0; + const workerCount = Math.max(1, Math.min(concurrency, items.length || 1)); + await Promise.all( + Array.from({ length: workerCount }, async () => { + while (nextIndex < items.length) { + const index = nextIndex; + nextIndex += 1; + results[index] = await mapper(items[index] as T); + } + }), + ); + return results; +} + /** * Install-wide contributor open-item count, LIVE-VERIFIED (#2562 gate-review follow-up): every OTHER counted * item is confirmed still-open via a live GET before counting toward the cap (mirrors the existing per-repo @@ -5022,10 +5053,8 @@ async function processGitHubWebhook( payload.installation?.id, detectNotificationEvents(eventName, payload), ); - for (const notificationEvent of [ - ...trustedReviewEvents, - ...issueWatchEvents, - ]) { + const notificationEvents = [...trustedReviewEvents, ...issueWatchEvents]; + for (const notificationEvent of notificationEvents) { await recordAuditEvent(env, { eventType: "notification.event_detected", actor: notificationEvent.actorLogin, @@ -5042,10 +5071,16 @@ async function processGitHubWebhook( deeplink: notificationEvent.deeplink, }, }); + } + // Batched (#selfhost-maintenance-self-pin): every event this ONE webhook delivery detected rides in a + // single notify-evaluate job instead of one job per event -- the audit trail above still records each + // event individually, so nothing about observability changes, only how many maintenance-lane rows a + // multi-watcher issue (or a review event landing alongside issue-watch matches) creates. + if (notificationEvents.length > 0) { await env.JOBS.send({ type: "notify-evaluate", requestedBy: "webhook", - event: notificationEvent, + events: notificationEvents, }); } diff --git a/src/selfhost/maintenance-admission.ts b/src/selfhost/maintenance-admission.ts index 03bd0bd3b3..a68b74bd1f 100644 --- a/src/selfhost/maintenance-admission.ts +++ b/src/selfhost/maintenance-admission.ts @@ -14,6 +14,20 @@ // TRICKLE: a maintenance job that has been pending since `maxDeferAgeMs` is force-admitted regardless of // current pressure, so a box under SUSTAINED load can never starve maintenance work forever -- it just runs at // a bounded minimum rate instead of its normal cadence. +// +// DRAIN (#selfhost-maintenance-self-pin): `maintenance_pending_high` alone is a LANE-WIDE aggregate count, with +// no feedback loop back to that count as individual jobs age out via the trickle above -- so once the lane backs +// up past `maxMaintenancePendingCount` and stays there (new maintenance work keeps arriving as fast as, or faster +// than, the trickle drains it), EVERY claim is denied `maintenance_pending_high` until each job independently +// reaches the full `maxDeferAgeMs` (hours later), and the aggregate count never has a chance to fall back under +// the threshold in the meantime -- the backlog is deferred because it's high, and stays high because it's +// deferred. `maintenanceDrainAgeMs` is a second, much shorter age escape scoped ONLY to the +// `maintenance_pending_high` branch: a job that has waited at least this long is admitted despite the lane still +// being over threshold, so the oldest jobs steadily leak through (throttled further by the queue's own +// `backgroundConcurrency` claim cap) and the aggregate count can actually shrink well before the 4h backstop. +// Newly-arrived jobs in the same burst still wait out `maintenanceDrainAgeMs` first, so this is a bounded trickle, +// not a flood -- and it applies to `maintenance_pending_high` alone: `live_pending_high` / `live_job_age_high` / +// `host_load_high` keep blocking maintenance outright, so live-review priority and host-load safety are untouched. import { deterministicJitterMs, parsePositiveIntEnv } from "./queue-common"; // Periodic, repo/contributor-set-wide sweeps -- the heavy, deferrable maintenance lane. Deliberately EXCLUDES @@ -71,6 +85,7 @@ export interface MaintenanceAdmissionConfig { maxHostLoadAvg1PerCore: number; deferMs: number; maxDeferAgeMs: number; + maintenanceDrainAgeMs: number; } const DEFAULT_MAX_LIVE_PENDING_COUNT = 5; @@ -79,6 +94,7 @@ const DEFAULT_MAX_MAINTENANCE_PENDING_COUNT = 15; const DEFAULT_MAX_HOST_LOAD_AVG1_PER_CORE = 1.5; const DEFAULT_DEFER_MS = 3 * 60_000; const DEFAULT_MAX_DEFER_AGE_MS = 4 * 60 * 60_000; +const DEFAULT_MAINTENANCE_DRAIN_AGE_MS = 10 * 60_000; function maintenanceAdmissionEnabled(): boolean { const raw = (process.env.MAINTENANCE_ADMISSION_ENABLED ?? "").trim().toLowerCase(); @@ -96,6 +112,14 @@ function parsePositiveFloatEnv(name: string, fallback: number): number { * ONCE per queue instance (mirrors queueBackgroundConcurrency / queueStartupJitterMs) rather than per job, so * a misconfigured value only warns once at startup instead of on every claim. */ export function resolveMaintenanceAdmissionConfig(): MaintenanceAdmissionConfig { + const maxDeferAgeMs = parsePositiveIntEnv("MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS", { + min: 60_000, + fallback: DEFAULT_MAX_DEFER_AGE_MS, + }); + const requestedDrainAgeMs = parsePositiveIntEnv("MAINTENANCE_ADMISSION_DRAIN_AGE_MS", { + min: 1_000, + fallback: DEFAULT_MAINTENANCE_DRAIN_AGE_MS, + }); return { enabled: maintenanceAdmissionEnabled(), maxLivePendingCount: parsePositiveIntEnv("MAINTENANCE_ADMISSION_MAX_LIVE_PENDING", { @@ -115,10 +139,10 @@ export function resolveMaintenanceAdmissionConfig(): MaintenanceAdmissionConfig DEFAULT_MAX_HOST_LOAD_AVG1_PER_CORE, ), deferMs: parsePositiveIntEnv("MAINTENANCE_ADMISSION_DEFER_MS", { min: 1_000, fallback: DEFAULT_DEFER_MS }), - maxDeferAgeMs: parsePositiveIntEnv("MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS", { - min: 60_000, - fallback: DEFAULT_MAX_DEFER_AGE_MS, - }), + maxDeferAgeMs, + // Never longer than the trickle backstop itself -- a misconfigured drain age above maxDeferAgeMs would be a + // no-op (the trickle would always win first), so clamp it down rather than let it silently do nothing. + maintenanceDrainAgeMs: Math.min(requestedDrainAgeMs, maxDeferAgeMs), }; } @@ -128,6 +152,7 @@ export type MaintenanceAdmissionReason = | "live_pending_high" | "live_job_age_high" | "maintenance_pending_high" + | "maintenance_pending_high_drain" | "host_load_high" | "pressure_clear"; @@ -143,7 +168,12 @@ export interface MaintenanceAdmissionDecision { * re-enqueue (a periodic scheduler re-requesting the same still-pending maintenance need) -- only a truly * fresh need, enqueued after the prior row was fully processed and deleted, starts a new clock. Otherwise a * re-enqueue cadence shorter than `maxDeferAgeMs` would keep re-arming the clock and defeat the trickle - * entirely under sustained pressure. */ + * entirely under sustained pressure. + * + * `maintenance_pending_high` alone gets a SECOND, shorter age escape (`maintenanceDrainAgeMs`, see the module + * comment above) so an aggregate-count block on the whole lane can't self-pin indefinitely -- live-pending, + * live-job-age, and host-load stay hard blocks with no drain, since those signals aren't about the maintenance + * lane's own size and letting maintenance through under THEM would defeat their purpose. */ export function evaluateMaintenanceAdmission( signals: MaintenancePressureSignals, config: MaintenanceAdmissionConfig, @@ -156,15 +186,31 @@ export function evaluateMaintenanceAdmission( if (signals.oldestLivePendingAgeMs !== null && signals.oldestLivePendingAgeMs > config.maxLiveJobAgeMs) { return { admit: false, reason: "live_job_age_high" }; } + const hostLoadHigh = + signals.hostLoadAvg1PerCore !== null && signals.hostLoadAvg1PerCore > config.maxHostLoadAvg1PerCore; if (signals.maintenancePendingCount > config.maxMaintenancePendingCount) { + if (nowMs - pendingSinceMs >= config.maintenanceDrainAgeMs) { + // Host load is re-checked HERE, gating the drain escape specifically: draining more maintenance work onto + // an already CPU-overloaded box is exactly what host_load_high exists to prevent. A job that hasn't hit + // drain age yet is denied `maintenance_pending_high` regardless of host load (unchanged from before this + // escape existed) -- this check only ever changes the outcome for a job the drain would otherwise admit. + if (hostLoadHigh) return { admit: false, reason: "host_load_high" }; + return { admit: true, reason: "maintenance_pending_high_drain" }; + } return { admit: false, reason: "maintenance_pending_high" }; } - if (signals.hostLoadAvg1PerCore !== null && signals.hostLoadAvg1PerCore > config.maxHostLoadAvg1PerCore) { - return { admit: false, reason: "host_load_high" }; - } + if (hostLoadHigh) return { admit: false, reason: "host_load_high" }; return { admit: true, reason: "pressure_clear" }; } +/** Admission reasons that grant a maintenance job despite active pressure -- every reason except the two + * "pressure was never a problem" ones (disabled / pressure_clear). Callers record a dedicated + * granted-under-pressure metric for these, the counterpart to the existing deferred-by-reason metric, so an + * operator can see the bounded trickle/drain actually firing instead of only ever seeing denials. */ +export function isMaintenanceAdmissionGrantedUnderPressure(reason: MaintenanceAdmissionReason): boolean { + return reason === "trickle_max_defer_age" || reason === "maintenance_pending_high_drain"; +} + /** Jittered defer duration for a denied maintenance job -- the base `deferMs` plus up to another `deferMs` of * deterministic jitter (seeded by the job's own identity) so a whole cohort of denied jobs doesn't wake up on * the same tick and immediately re-trip the same pressure check (mirrors rateLimitRetryDelayWithJitter). */ diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts index aec35c39c2..450131f76b 100644 --- a/src/selfhost/pg-queue.ts +++ b/src/selfhost/pg-queue.ts @@ -21,6 +21,8 @@ import { isForegroundJobPriority, jobCoalesceAbsorbedByKey, jobCoalesceKey, + jobCoalesceMergeKeyPrefix, + jobCoalesceMergedPayload, jobCoalesceSupersededKeyPrefix, jobPriority, parsePositiveIntEnv, @@ -124,6 +126,7 @@ async function retryPoolUpdateOrLeaveForReclaim( import { hostLoadAvg1PerCore } from "./host-pressure"; import { evaluateMaintenanceAdmission, + isMaintenanceAdmissionGrantedUnderPressure, isMaintenanceJobType, maintenanceAdmissionDeferMs, resolveMaintenanceAdmissionConfig, @@ -510,6 +513,44 @@ export function createPgQueue( return; } } + // Merge two INCREMENTAL rag-index-repo jobs for the same repo (#selfhost-maintenance-self-pin) into one + // pending row's UNION path set instead of piling up as separate maintenance-lane rows -- mirrors + // sqlite-queue.ts exactly. `absorbedByKey` shares mergeKeyPrefix's exact guard so it's provably non-null + // here (asserted, not defaulted); excluding it is defense-in-depth against a job_key collision, not + // load-bearing, though under Postgres's multi-instance concurrency it's a real (if narrow) race guard. + const mergeKeyPrefix = jobCoalesceMergeKeyPrefix(payload); + if (mergeKeyPrefix) { + const mergeCandidate = ( + await pool.query( + `SELECT id, payload, job_key FROM ${TABLE} + WHERE status='pending' AND job_key IS NOT NULL AND left(job_key, $1)=$2 AND job_key<>$3 + ORDER BY priority DESC, run_after DESC, id LIMIT 1`, + [mergeKeyPrefix.length, mergeKeyPrefix, absorbedByKey as string], + ) + ).rows[0] as { id: string; payload: string; job_key: string } | undefined; + if (mergeCandidate) { + const mergedPayload = jobCoalesceMergedPayload(mergeCandidate.payload, payload); + if (mergedPayload) { + const mergedKey = jobCoalesceKey(mergedPayload); + // Guarded by status='pending' AND job_key= so a concurrent claim or a + // second instance's own merge into this same row between the SELECT and here loses cleanly (rowCount + // 0) instead of silently overwriting whatever the winner just wrote -- multiple self-host instances + // can race this exact SELECT-then-UPDATE (gate finding). Falling through (not returning) on a lost + // race lets the normal supersede/coalesce/insert path below handle this job instead. + const merged = await pool.query( + `UPDATE ${TABLE} + SET payload=$1, run_after=GREATEST(run_after, $2), created_at=$3, priority=GREATEST(priority, $4), job_key=$5, last_error=NULL + WHERE id=$6 AND status='pending' AND job_key=$7`, + [mergedPayload, runAfter, now, priority, mergedKey, mergeCandidate.id, mergeCandidate.job_key], + ); + if (merged.rowCount) { + await recordQueueMetric("gittensory_jobs_coalesced_total"); + kickOne(); + return; + } + } + } + } const supersededKeyPrefix = jobCoalesceSupersededKeyPrefix(payload); if (key && supersededKeyPrefix) { const existing = ( @@ -818,6 +859,16 @@ export function createPgQueue( }), ); } + // Broader force-admitted-under-pressure signal (#selfhost-maintenance-self-pin): covers trickle_max_defer_age + // above PLUS maintenance_pending_high_drain (the new scoped drain escape this PR adds) under one counter, + // so an operator can trend "how often does pressure admission get overridden at all" without needing to + // sum multiple per-reason metrics. + if (isMaintenanceAdmissionGrantedUnderPressure(decision.reason)) { + incr("gittensory_jobs_maintenance_admission_granted_under_pressure_total", { + reason: decision.reason, + job_type: message.type, + }); + } } try { await withReviewSpan( diff --git a/src/selfhost/queue-common.ts b/src/selfhost/queue-common.ts index a7bc828f7c..ffbb3d50f9 100644 --- a/src/selfhost/queue-common.ts +++ b/src/selfhost/queue-common.ts @@ -653,7 +653,7 @@ type CoalesceMessage = { runId?: unknown; deliveryId?: unknown; draftId?: unknown; - event?: { dedupKey?: unknown } | null; + events?: Array<{ dedupKey?: unknown } | null | undefined> | null; logins?: unknown; payload?: GitHubWebhookPayload | null; }; @@ -690,6 +690,42 @@ export function jobCoalesceAbsorbedByKey(payload: string): string | null { return ragIndexFullKey(repo); } +// Mirrors processors.ts's per-PR RAG_REINDEX_MAX_PATHS cap: bounds how large a MERGED incremental job's path +// set can grow across repeated merges while pending under pressure (#selfhost-maintenance-self-pin), so a +// backed-up repo with many small merges can't accumulate one ever-growing row instead of separate ones. +const RAG_INDEX_MERGE_MAX_PATHS = 100; + +/** Repo-scoped key PREFIX matching any OTHER pending incremental (path-scoped) rag-index-repo job for the same + * repo -- distinct from `jobCoalesceAbsorbedByKey` (which targets an existing FULL job's exact key). Only + * non-null for an incoming INCREMENTAL job; a full-repo job is handled by `jobCoalesceSupersededKeyPrefix` + * instead. Used together with `jobCoalesceMergedPayload` at enqueue time so several merge-triggered incremental + * jobs for the same repo, arriving while one is still pending, union their paths into a single row instead of + * piling up as separate maintenance-lane entries. */ +export function jobCoalesceMergeKeyPrefix(payload: string): string | null { + const message = parseCoalesceMessage(payload); + if (message?.type !== "rag-index-repo") return null; + const repo = normalizedRepo(message.repoFullName); + if (!repo || !normalizedPathScope(message.paths)) return null; + return ragIndexRepoKeyPrefix(repo); +} + +/** Union the incoming incremental rag-index-repo job's paths into an already-pending incremental job's paths + * (deduped + sorted, for a stable coalesce key). Returns null when either side isn't a path-scoped rag-index-repo + * message, or when the merged set would exceed RAG_INDEX_MERGE_MAX_PATHS -- the caller then falls through to a + * separate row instead of merging, rather than let one row's path list grow unbounded. */ +export function jobCoalesceMergedPayload(existingPayload: string, incomingPayload: string): string | null { + const existing = parseCoalesceMessage(existingPayload); + const incoming = parseCoalesceMessage(incomingPayload); + if (existing?.type !== "rag-index-repo" || incoming?.type !== "rag-index-repo") return null; + const isStringPath = (entry: unknown): entry is string => typeof entry === "string" && entry.trim().length > 0; + const existingPaths = Array.isArray(existing.paths) ? existing.paths.filter(isStringPath) : []; + const incomingPaths = Array.isArray(incoming.paths) ? incoming.paths.filter(isStringPath) : []; + if (existingPaths.length === 0 || incomingPaths.length === 0) return null; + const merged = [...new Set([...existingPaths, ...incomingPaths])].sort(); + if (merged.length > RAG_INDEX_MERGE_MAX_PATHS) return null; + return JSON.stringify({ ...incoming, paths: merged }); +} + export function jobCoalesceKey(payload: string): string | null { try { const message = parseCoalesceMessage(payload); @@ -809,8 +845,16 @@ export function jobCoalesceKey(payload: string): string | null { return deliveryId ? keyOf(type, deliveryId) : null; } case "notify-evaluate": { - const dedupKey = normalizedId(message.event?.dedupKey); - return dedupKey ? keyOf(type, dedupKey) : null; + // A batched job carries every event from one webhook delivery (#selfhost-maintenance-self-pin) -- + // coalescing keys off the FULL sorted set of dedup keys, so a redelivery of the identical batch still + // coalesces (same events -> same key) while any batch with even one different event gets its own key. + // If ANY event is missing its dedup key (a malformed payload), the whole batch is left uncoalesced + // (null) rather than keying off a partial set that could collide with an unrelated batch and silently + // drop the malformed event's work -- same rule as the other event-id-keyed types above. + if (!Array.isArray(message.events) || message.events.length === 0) return null; + const dedupKeys = message.events.map((event) => normalizedId(event?.dedupKey)); + if (dedupKeys.some((dedupKey) => dedupKey === null)) return null; + return keyOf(type, [...(dedupKeys as string[])].sort().join(",")); } case "submit-draft": { const draftId = normalizedId(message.draftId); diff --git a/src/selfhost/sqlite-queue.ts b/src/selfhost/sqlite-queue.ts index 1317f2beb0..e14409b655 100644 --- a/src/selfhost/sqlite-queue.ts +++ b/src/selfhost/sqlite-queue.ts @@ -22,6 +22,8 @@ import { isForegroundJobPriority, jobCoalesceAbsorbedByKey, jobCoalesceKey, + jobCoalesceMergeKeyPrefix, + jobCoalesceMergedPayload, jobCoalesceSupersededKeyPrefix, jobPriority, parsePositiveIntEnv, @@ -40,6 +42,7 @@ import { import { hostLoadAvg1PerCore } from "./host-pressure"; import { evaluateMaintenanceAdmission, + isMaintenanceAdmissionGrantedUnderPressure, isMaintenanceJobType, maintenanceAdmissionDeferMs, resolveMaintenanceAdmissionConfig, @@ -291,6 +294,38 @@ export function createSqliteQueue( return; } } + // Merge two INCREMENTAL rag-index-repo jobs for the same repo (#selfhost-maintenance-self-pin), e.g. several + // merged PRs touching different files in a burst, into one pending row's UNION path set instead of piling up + // as separate maintenance-lane rows. + const mergeKeyPrefix = jobCoalesceMergeKeyPrefix(payload); + if (mergeKeyPrefix) { + const prefixLength = mergeKeyPrefix.length; + // `absorbedByKey` shares mergeKeyPrefix's exact guard (both require an incoming path-scoped rag-index-repo + // message), so it's provably non-null here -- it's asserted, not defaulted, because we only reach this + // branch once it found no pending FULL job to absorb into; excluding that same key guards against a + // job_key collision, it can never actually match a row here. + const mergeCandidate = driver.query( + `SELECT id, payload FROM ${TABLE} + WHERE status='pending' AND job_key IS NOT NULL AND substr(job_key, 1, ?)=? AND job_key<>? + ORDER BY priority DESC, run_after DESC, id LIMIT 1`, + [prefixLength, mergeKeyPrefix, absorbedByKey as string], + ).rows[0] as { id: number; payload: string } | undefined; + if (mergeCandidate) { + const mergedPayload = jobCoalesceMergedPayload(mergeCandidate.payload, payload); + if (mergedPayload) { + const mergedKey = jobCoalesceKey(mergedPayload); + driver.query( + `UPDATE ${TABLE} + SET payload=?, run_after=max(run_after, ?), created_at=?, priority=max(priority, ?), job_key=?, last_error=NULL + WHERE id=?`, + [mergedPayload, runAfter, now, priority, mergedKey, mergeCandidate.id], + ); + recordQueueMetric(driver, "gittensory_jobs_coalesced_total"); + kickOne(); + return; + } + } + } const supersededKeyPrefix = jobCoalesceSupersededKeyPrefix(payload); if (key && supersededKeyPrefix) { const prefixLength = supersededKeyPrefix.length; @@ -587,6 +622,16 @@ export function createSqliteQueue( }), ); } + // Broader force-admitted-under-pressure signal (#selfhost-maintenance-self-pin): covers trickle_max_defer_age + // above PLUS maintenance_pending_high_drain (the new scoped drain escape this PR adds) under one counter, + // so an operator can trend "how often does pressure admission get overridden at all" without needing to + // sum multiple per-reason metrics. + if (isMaintenanceAdmissionGrantedUnderPressure(decision.reason)) { + incr("gittensory_jobs_maintenance_admission_granted_under_pressure_total", { + reason: decision.reason, + job_type: message.type, + }); + } } try { await withReviewSpan( diff --git a/src/types.ts b/src/types.ts index ddc937aa91..b891a6edf8 100644 --- a/src/types.ts +++ b/src/types.ts @@ -160,9 +160,13 @@ export type JobMessage = runId: string; } | { + // Batched (#selfhost-maintenance-self-pin): every notification event detected from ONE webhook delivery + // (a review event plus any issue-watch matches) rides in a single job, instead of one job per event -- + // that was flooding the maintenance lane with a job per watcher on a popular newly-opened issue. Always + // non-empty at enqueue time (see processors.ts); the processor evaluates every event in the batch. type: "notify-evaluate"; requestedBy: "webhook" | "test"; - event: DetectedNotificationEvent; + events: DetectedNotificationEvent[]; } | { type: "notify-deliver"; diff --git a/test/unit/notifications-service.test.ts b/test/unit/notifications-service.test.ts index cfa19f2c2b..fd85df8a0e 100644 --- a/test/unit/notifications-service.test.ts +++ b/test/unit/notifications-service.test.ts @@ -193,7 +193,7 @@ describe("notification queue wiring", () => { } as unknown as Queue, }); - await processJob(env, { type: "notify-evaluate", requestedBy: "test", event: event() }); + await processJob(env, { type: "notify-evaluate", requestedBy: "test", events: [event()] }); const deliverJob = enqueued.find((message) => message.type === "notify-deliver"); expect(deliverJob?.deliveryId).toBeTruthy(); @@ -204,9 +204,59 @@ describe("notification queue wiring", () => { // A retried evaluate (same event) enqueues no further deliver jobs. const before = enqueued.length; - await processJob(env, { type: "notify-evaluate", requestedBy: "test", event: event() }); + await processJob(env, { type: "notify-evaluate", requestedBy: "test", events: [event()] }); expect(enqueued.length).toBe(before); }); + + it("evaluates every event in a batched notify-evaluate job (#selfhost-maintenance-self-pin)", async () => { + const enqueued: Array<{ type: string; deliveryId?: string }> = []; + const env = createTestEnv({ + JOBS: { + async send(message: { type: string; deliveryId?: string }) { + enqueued.push(message); + }, + } as unknown as Queue, + }); + + await processJob(env, { + type: "notify-evaluate", + requestedBy: "test", + events: [ + event({ recipientLogin: "miner-one", dedupKey: "changes_requested:owner/repo#7:reviewer:t1" }), + event({ recipientLogin: "miner-two", dedupKey: "changes_requested:owner/repo#8:reviewer:t2" }), + ], + }); + + const deliverJobs = enqueued.filter((message) => message.type === "notify-deliver"); + expect(deliverJobs).toHaveLength(2); + const deliveredLogins = await Promise.all( + deliverJobs.map(async (job) => { + await processJob(env, { type: "notify-deliver", requestedBy: "test", deliveryId: job.deliveryId! }); + return true; + }), + ); + expect(deliveredLogins).toEqual([true, true]); + expect(await listNotificationDeliveriesForRecipient(env, "miner-one", { unreadOnly: true })).toHaveLength(1); + expect(await listNotificationDeliveriesForRecipient(env, "miner-two", { unreadOnly: true })).toHaveLength(1); + }); + + it("REGRESSION (gate finding): a legacy pre-upgrade payload (singular `event`, no `events` array) is still evaluated instead of throwing", async () => { + const enqueued: Array<{ type: string; deliveryId?: string }> = []; + const env = createTestEnv({ + JOBS: { + async send(message: { type: string; deliveryId?: string }) { + enqueued.push(message); + }, + } as unknown as Queue, + }); + + // A row enqueued before the batched-events deploy still carries the OLD singular `event` field on disk -- + // cast past the current (events-only) type to simulate a durable payload from before the upgrade. + const legacyMessage = { type: "notify-evaluate", requestedBy: "test", event: event() } as unknown as { type: "notify-evaluate"; requestedBy: "test"; events: DetectedNotificationEvent[] }; + + await expect(processJob(env, legacyMessage)).resolves.not.toThrow(); + expect(enqueued.some((message) => message.type === "notify-deliver")).toBe(true); + }); }); describe("notification repository helpers", () => { diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index a42486f42b..5d9bc799a7 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -15368,9 +15368,10 @@ describe("queue processors", () => { }); expect(JSON.stringify(detected.results[0])).not.toMatch(/trust score|wallet|hotkey|reward estimate|reviewability/i); - const evaluateJob = enqueued.find((message): message is { type: "notify-evaluate"; event: { recipientLogin: string } } => message.type === "notify-evaluate"); + const evaluateJob = enqueued.find((message): message is { type: "notify-evaluate"; events: Array<{ recipientLogin: string }> } => message.type === "notify-evaluate"); expect(evaluateJob).toBeDefined(); - expect(evaluateJob!.event.recipientLogin).toBe("contributor"); + expect(evaluateJob!.events).toHaveLength(1); + expect(evaluateJob!.events[0]!.recipientLogin).toBe("contributor"); }); it("skips changes-requested review notifications from reviewers without repository write permission", async () => { @@ -15626,10 +15627,11 @@ describe("queue processors", () => { }); it("notifies issue-watchers when a new grabbable maintainer-created issue opens (#699 path B)", async () => { - const enqueued: Array<{ type: string; event?: { eventType: string; recipientLogin: string; pullNumber: number } }> = []; + const enqueued: Array<{ type: string; events?: Array<{ eventType: string; recipientLogin: string; pullNumber: number }> }> = []; const env = createTestEnv({ JOBS: { async send(message: { type: string }) { enqueued.push(message); } } as unknown as Queue }); vi.stubGlobal("fetch", async () => new Response("not found", { status: 404 })); // no .gittensory.yml → empty manifest - await upsertIssueWatchSubscription(env, { login: "watcher", repoFullName: "JSONbored/gittensory" }); + await upsertIssueWatchSubscription(env, { login: "watcher-one", repoFullName: "JSONbored/gittensory" }); + await upsertIssueWatchSubscription(env, { login: "watcher-two", repoFullName: "JSONbored/gittensory" }); await upsertIssueWatchSubscription(env, { login: "maintainer", repoFullName: "JSONbored/gittensory" }); // the author — should be skipped await processJob(env, { @@ -15644,12 +15646,17 @@ describe("queue processors", () => { }, }); - const watchEvents = enqueued.filter((m): m is { type: "notify-evaluate"; event: { eventType: string; recipientLogin: string; pullNumber: number } } => m.type === "notify-evaluate" && m.event?.eventType === "issue_watch_match"); - expect(watchEvents.map((m) => m.event.recipientLogin)).toEqual(["watcher"]); // maintainer (author) skipped - expect(watchEvents[0]!.event.pullNumber).toBe(91); + // Batched (#selfhost-maintenance-self-pin): both watcher matches from this ONE webhook delivery ride in a + // SINGLE notify-evaluate job, not one job per watcher -- that fan-out was flooding the self-host maintenance + // lane with a job per watcher on a popular issue. + const evaluateJobs = enqueued.filter((m): m is { type: "notify-evaluate"; events: Array<{ eventType: string; recipientLogin: string; pullNumber: number }> } => m.type === "notify-evaluate"); + expect(evaluateJobs).toHaveLength(1); + const watchEvents = evaluateJobs[0]!.events.filter((event) => event.eventType === "issue_watch_match"); + expect(watchEvents.map((event) => event.recipientLogin).sort()).toEqual(["watcher-one", "watcher-two"]); // maintainer (author) skipped + expect(watchEvents.every((event) => event.pullNumber === 91)).toBe(true); - const detected = await env.DB.prepare("select metadata_json from audit_events where event_type = 'notification.event_detected' and target_key = ?").bind("watcher").first<{ metadata_json: string }>(); - expect(JSON.parse(detected!.metadata_json)).toMatchObject({ eventType: "issue_watch_match", recipientLogin: "watcher", repoFullName: "JSONbored/gittensory" }); + const detected = await env.DB.prepare("select metadata_json from audit_events where event_type = 'notification.event_detected' and target_key = ?").bind("watcher-one").first<{ metadata_json: string }>(); + expect(JSON.parse(detected!.metadata_json)).toMatchObject({ eventType: "issue_watch_match", recipientLogin: "watcher-one", repoFullName: "JSONbored/gittensory" }); }); it("appends issue-side slop findings to the issue advisory only when slop is opted in (#533)", async () => { diff --git a/test/unit/selfhost-maintenance-admission.test.ts b/test/unit/selfhost-maintenance-admission.test.ts index 4bcf4365b3..cdef7d5028 100644 --- a/test/unit/selfhost-maintenance-admission.test.ts +++ b/test/unit/selfhost-maintenance-admission.test.ts @@ -1,11 +1,13 @@ import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { evaluateMaintenanceAdmission, + isMaintenanceAdmissionGrantedUnderPressure, isMaintenanceJobType, maintenanceAdmissionDeferMs, MAINTENANCE_JOB_TYPES, resolveMaintenanceAdmissionConfig, type MaintenanceAdmissionConfig, + type MaintenanceAdmissionReason, type MaintenancePressureSignals, } from "../../src/selfhost/maintenance-admission"; @@ -25,6 +27,7 @@ const CONFIG: MaintenanceAdmissionConfig = { maxHostLoadAvg1PerCore: 1.5, deferMs: 180_000, maxDeferAgeMs: 4 * 60 * 60_000, + maintenanceDrainAgeMs: 600_000, }; describe("isMaintenanceJobType", () => { @@ -121,6 +124,73 @@ describe("evaluateMaintenanceAdmission", () => { expect(decision).toEqual({ admit: false, reason: "maintenance_pending_high" }); }); + it("admits when maintenance pending count is AT (not over) the threshold", () => { + const decision = evaluateMaintenanceAdmission( + { ...CLEAR_SIGNALS, maintenancePendingCount: 15 }, + CONFIG, + now - 1_000, + now, + ); + expect(decision).toEqual({ admit: true, reason: "pressure_clear" }); + }); + + // Regression (#selfhost-maintenance-self-pin): before the drain escape, a lane backed up past threshold denied + // EVERY job -- old or new -- until each individually reached the full maxDeferAgeMs (hours later), so the + // aggregate count never got a chance to fall back under the threshold in the meantime: deferred because high, + // stuck high because deferred. The drain escape lets the OLDEST jobs in that same backlog through well before + // the 4h backstop, so the count can actually shrink. + it("drain-admits a maintenance job under a large backlog once it has waited past the drain age", () => { + const decision = evaluateMaintenanceAdmission( + { ...CLEAR_SIGNALS, maintenancePendingCount: 68 }, + CONFIG, + now - CONFIG.maintenanceDrainAgeMs, + now, + ); + expect(decision).toEqual({ admit: true, reason: "maintenance_pending_high_drain" }); + }); + + it("does not drain-admit a maintenance job that hasn't waited long enough yet", () => { + const decision = evaluateMaintenanceAdmission( + { ...CLEAR_SIGNALS, maintenancePendingCount: 68 }, + CONFIG, + now - (CONFIG.maintenanceDrainAgeMs - 1), + now, + ); + expect(decision).toEqual({ admit: false, reason: "maintenance_pending_high" }); + }); + + it("regression: a large backlog still lets a fresh job wait while an old job in the SAME backlog drains", () => { + // Same maintenancePendingCount (the aggregate never moves within a single evaluation) -- only the + // individual job's own age differs, proving the escape is per-job, not a relaxation of the lane threshold. + const signals: MaintenancePressureSignals = { ...CLEAR_SIGNALS, maintenancePendingCount: 68 }; + const oldJob = evaluateMaintenanceAdmission(signals, CONFIG, now - CONFIG.maintenanceDrainAgeMs, now); + const freshJob = evaluateMaintenanceAdmission(signals, CONFIG, now - 1_000, now); + expect(oldJob).toEqual({ admit: true, reason: "maintenance_pending_high_drain" }); + expect(freshJob).toEqual({ admit: false, reason: "maintenance_pending_high" }); + }); + + it("does not drain-admit when host load is ALSO high -- host_load_high wins over the drain escape", () => { + const decision = evaluateMaintenanceAdmission( + { ...CLEAR_SIGNALS, maintenancePendingCount: 68, hostLoadAvg1PerCore: 99 }, + CONFIG, + now - CONFIG.maintenanceDrainAgeMs, + now, + ); + expect(decision).toEqual({ admit: false, reason: "host_load_high" }); + }); + + it("still reports maintenance_pending_high (not host_load_high) before drain age, even if host load is also high", () => { + // Host load is only consulted INSIDE the drain-eligible branch, so a job that hasn't reached drain age yet + // keeps the original (pre-drain-escape) denial reason regardless of host load. + const decision = evaluateMaintenanceAdmission( + { ...CLEAR_SIGNALS, maintenancePendingCount: 68, hostLoadAvg1PerCore: 99 }, + CONFIG, + now - (CONFIG.maintenanceDrainAgeMs - 1), + now, + ); + expect(decision).toEqual({ admit: false, reason: "maintenance_pending_high" }); + }); + it("defers when host load per core exceeds the threshold", () => { const decision = evaluateMaintenanceAdmission( { ...CLEAR_SIGNALS, hostLoadAvg1PerCore: 1.51 }, @@ -141,6 +211,26 @@ describe("evaluateMaintenanceAdmission", () => { expect(decision.admit).toBe(true); }); + it("admits when host load is AT (not over) the threshold", () => { + const decision = evaluateMaintenanceAdmission( + { ...CLEAR_SIGNALS, hostLoadAvg1PerCore: 1.5 }, + CONFIG, + now - 1_000, + now, + ); + expect(decision).toEqual({ admit: true, reason: "pressure_clear" }); + }); + + it("admits when the oldest live job's age is AT (not over) the threshold", () => { + const decision = evaluateMaintenanceAdmission( + { ...CLEAR_SIGNALS, oldestLivePendingAgeMs: 120_000 }, + CONFIG, + now - 1_000, + now, + ); + expect(decision).toEqual({ admit: true, reason: "pressure_clear" }); + }); + it("force-admits via trickle once pending since exceeds the max defer age, even under pressure", () => { const decision = evaluateMaintenanceAdmission( { ...CLEAR_SIGNALS, livePendingCount: 999, hostLoadAvg1PerCore: 99 }, @@ -170,6 +260,31 @@ describe("evaluateMaintenanceAdmission", () => { ); expect(decision.reason).toBe("live_pending_high"); }); + + it("checks the oldest-live-job age before maintenance-lane pressure (priority order)", () => { + const decision = evaluateMaintenanceAdmission( + { ...CLEAR_SIGNALS, oldestLivePendingAgeMs: 120_001, maintenancePendingCount: 16 }, + CONFIG, + now - 1_000, + now, + ); + expect(decision.reason).toBe("live_job_age_high"); + }); +}); + +describe("isMaintenanceAdmissionGrantedUnderPressure", () => { + it.each([ + ["disabled", false], + ["pressure_clear", false], + ["live_pending_high", false], + ["live_job_age_high", false], + ["maintenance_pending_high", false], + ["host_load_high", false], + ["trickle_max_defer_age", true], + ["maintenance_pending_high_drain", true], + ] satisfies Array<[MaintenanceAdmissionReason, boolean]>)("reports %s as granted-under-pressure=%s", (reason, expected) => { + expect(isMaintenanceAdmissionGrantedUnderPressure(reason)).toBe(expected); + }); }); describe("maintenanceAdmissionDeferMs", () => { @@ -200,6 +315,7 @@ describe("resolveMaintenanceAdmissionConfig", () => { "MAINTENANCE_ADMISSION_MAX_HOST_LOAD", "MAINTENANCE_ADMISSION_DEFER_MS", "MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS", + "MAINTENANCE_ADMISSION_DRAIN_AGE_MS", ] as const; const saved: Record = {}; @@ -226,6 +342,7 @@ describe("resolveMaintenanceAdmissionConfig", () => { maxHostLoadAvg1PerCore: 1.5, deferMs: 180_000, maxDeferAgeMs: 4 * 60 * 60_000, + maintenanceDrainAgeMs: 600_000, }); }); @@ -236,6 +353,7 @@ describe("resolveMaintenanceAdmissionConfig", () => { process.env.MAINTENANCE_ADMISSION_MAX_HOST_LOAD = "2.25"; process.env.MAINTENANCE_ADMISSION_DEFER_MS = "5000"; process.env.MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS = "3600000"; + process.env.MAINTENANCE_ADMISSION_DRAIN_AGE_MS = "60000"; const config = resolveMaintenanceAdmissionConfig(); expect(config.maxLivePendingCount).toBe(10); expect(config.maxLiveJobAgeMs).toBe(60_000); @@ -243,6 +361,19 @@ describe("resolveMaintenanceAdmissionConfig", () => { expect(config.maxHostLoadAvg1PerCore).toBe(2.25); expect(config.deferMs).toBe(5_000); expect(config.maxDeferAgeMs).toBe(3_600_000); + expect(config.maintenanceDrainAgeMs).toBe(60_000); + }); + + it("clamps a drain age above the max defer age down to the max defer age", () => { + process.env.MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS = "3600000"; // 1h + process.env.MAINTENANCE_ADMISSION_DRAIN_AGE_MS = "7200000"; // 2h -- would otherwise never fire + expect(resolveMaintenanceAdmissionConfig().maintenanceDrainAgeMs).toBe(3_600_000); + }); + + it("does not clamp a drain age already below the max defer age", () => { + process.env.MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS = "3600000"; // 1h + process.env.MAINTENANCE_ADMISSION_DRAIN_AGE_MS = "60000"; // 1min + expect(resolveMaintenanceAdmissionConfig().maintenanceDrainAgeMs).toBe(60_000); }); it.each(["0", "false", "off", "no"])("treats MAINTENANCE_ADMISSION_ENABLED=%s as disabled", (value) => { diff --git a/test/unit/selfhost-pg-queue.test.ts b/test/unit/selfhost-pg-queue.test.ts index 801ee5fe53..cb0015a04e 100644 --- a/test/unit/selfhost-pg-queue.test.ts +++ b/test/unit/selfhost-pg-queue.test.ts @@ -455,6 +455,150 @@ describe("createPgQueue (durable #977)", () => { ); }); + // #selfhost-maintenance-self-pin: mirrors selfhost-sqlite-queue.test.ts -- two pending incrementals for the + // same repo merge into one row's union path set instead of piling up as separate maintenance-lane rows. + it("merges a new incremental RAG job into an already-pending incremental for the same repo", async () => { + const m = makePool(); + const q = createPgQueue(m.pool, async () => undefined); + await q.init(); + m.fn.mockResolvedValueOnce({ rows: [], rowCount: 0 }); // absorbedByKey check: no pending FULL job for this repo + m.fn.mockResolvedValueOnce({ + rows: [{ id: "existing-incremental", payload: JSON.stringify({ type: "rag-index-repo", requestedBy: "webhook", repoFullName: "JSONbored/gittensory", paths: ["src/a.ts"] }), job_key: "rag-index-repo:jsonbored/gittensory:sha256:existing" }], + rowCount: 1, + }); // merge-lookup query: an existing pending incremental for this repo + m.fn.mockResolvedValueOnce({ rows: [], rowCount: 1 }); // the guarded UPDATE wins the race — 1 row affected + + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/gittensory", + paths: ["src/b.ts"], + }); + + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("left(job_key, $1)=$2"), + ["rag-index-repo:jsonbored/gittensory:".length, "rag-index-repo:jsonbored/gittensory:", "rag-index-repo:jsonbored/gittensory:full"], + ); + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("SET payload=$1, run_after=GREATEST"), + expect.arrayContaining([ + expect.stringContaining('"paths":["src/a.ts","src/b.ts"]'), + expect.any(Number), + expect.any(Number), + 0, + expect.stringContaining("rag-index-repo:jsonbored/gittensory:sha256:"), + "existing-incremental", + "rag-index-repo:jsonbored/gittensory:sha256:existing", + ]), + ); + expect(m.pool.query).not.toHaveBeenCalledWith( + expect.stringContaining("INSERT INTO _selfhost_jobs (payload"), + expect.arrayContaining([expect.stringContaining('"paths":["src/b.ts"]')]), + ); + }); + + it("REGRESSION (gate finding): a lost merge race (rowCount 0 — another instance claimed/mutated the row first) falls through to a normal insert instead of silently overwriting", async () => { + const m = makePool(); + const q = createPgQueue(m.pool, async () => undefined); + await q.init(); + m.fn.mockResolvedValueOnce({ rows: [], rowCount: 0 }); // absorbedByKey check: no pending FULL job for this repo + m.fn.mockResolvedValueOnce({ + rows: [{ id: "existing-incremental", payload: JSON.stringify({ type: "rag-index-repo", requestedBy: "webhook", repoFullName: "JSONbored/gittensory", paths: ["src/a.ts"] }), job_key: "rag-index-repo:jsonbored/gittensory:sha256:existing" }], + rowCount: 1, + }); // merge-lookup query: an existing pending incremental for this repo + m.fn.mockResolvedValueOnce({ rows: [], rowCount: 0 }); // the guarded UPDATE LOSES the race — another instance already claimed/mutated this row + + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/gittensory", + paths: ["src/b.ts"], + }); + + // Falls through to the normal enqueue path — a fresh INSERT for this job, never a second blind UPDATE + // against the same (already-claimed) row. + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("INSERT INTO"), + expect.arrayContaining([expect.stringContaining('"paths":["src/b.ts"]')]), + ); + }); + + it("does not merge an incremental into an already-pending FULL job for that repo", async () => { + const m = makePool(); + const q = createPgQueue(m.pool, async () => undefined); + await q.init(); + // absorbedByKey's own exact-match query finds the pending full job first, so the merge query never runs. + m.fn.mockResolvedValueOnce({ rows: [{ id: "existing-full" }], rowCount: 1 }); + + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/gittensory", + paths: ["src/a.ts"], + }); + + expect(m.pool.query).not.toHaveBeenCalledWith( + expect.stringContaining("left(job_key, $1)=$2"), + expect.anything(), + ); + }); + + it("does not merge when the merge-lookup query finds no candidate (e.g. a different repo)", async () => { + const m = makePool(); + const q = createPgQueue(m.pool, async () => undefined); + await q.init(); + m.fn.mockResolvedValueOnce({ rows: [], rowCount: 0 }); // absorbedByKey check: no pending FULL job + m.fn.mockResolvedValueOnce({ rows: [], rowCount: 0 }); // merge-lookup query: no pending incremental either + + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/gittensory", + paths: ["src/a.ts"], + }); + + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("left(job_key, $1)=$2"), + [36, "rag-index-repo:jsonbored/gittensory:", "rag-index-repo:jsonbored/gittensory:full"], + ); + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("INSERT INTO _selfhost_jobs (payload"), + expect.arrayContaining([expect.stringContaining('"paths":["src/a.ts"]')]), + ); + }); + + it("falls through to a separate row when merging would exceed the bounded path cap", async () => { + const m = makePool(); + const q = createPgQueue(m.pool, async () => undefined); + await q.init(); + m.fn.mockResolvedValueOnce({ rows: [], rowCount: 0 }); // absorbedByKey check: no pending FULL job + m.fn.mockResolvedValueOnce({ + rows: [{ + id: "existing-at-cap", + payload: JSON.stringify({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/gittensory", + paths: Array.from({ length: 100 }, (_, i) => `src/${i}.ts`), + }), + }], + rowCount: 1, + }); // merge-lookup query: an existing pending incremental already at the cap + + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/gittensory", + paths: ["src/extra.ts"], + }); + + // No merge (would be 101 paths, over the cap) -- falls through to a plain INSERT of its own row. + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("INSERT INTO _selfhost_jobs (payload"), + expect.arrayContaining([expect.stringContaining('"paths":["src/extra.ts"]')]), + ); + }); + it("coalesces recurring maintenance jobs by semantic scope and preserves distinct scopes", async () => { const m = makePool(); const q = createPgQueue(m.pool, async () => undefined); @@ -2044,6 +2188,53 @@ describe("createPgQueue (durable #977)", () => { expect(started).not.toContain("build-contributor-evidence"); }); + // Regression (#selfhost-maintenance-self-pin): mirrors selfhost-sqlite-queue.test.ts exactly -- a large + // backlog (well over threshold) no longer denies EVERY job forever; a job old enough for the drain age gets + // admitted while a fresh job in the SAME backlog still defers. + it("drain-admits an old job in a large backlog once it has waited past the drain age, while a fresh job in the SAME backlog still defers", async () => { + const oldEnv = process.env.MAINTENANCE_ADMISSION_DRAIN_AGE_MS; + process.env.MAINTENANCE_ADMISSION_DRAIN_AGE_MS = "60000"; // 1m (parsePositiveIntEnv floor) + try { + const m = makePool(); + m.setPressureSignals({ maintenance: { cnt: 68, oldest: now } }); // mirrors the reported incident's backlog size + m.enqueueResult({ rows: [], rowCount: 0 }); // empty foreground claim + m.enqueueResult({ rows: [{ ...maintenanceRow, created_at: now - 61_000 }], rowCount: 1 }); // old job: drained + const started: string[] = []; + const q = createPgQueue(m.pool, async (j) => void started.push(typeOf(j))); + await q.drain(); + expect(started).toEqual(["build-contributor-evidence"]); + expect(await renderMetrics()).toContain( + 'gittensory_jobs_maintenance_admission_granted_under_pressure_total{job_type="build-contributor-evidence",reason="maintenance_pending_high_drain"} 1', + ); + } finally { + if (oldEnv === undefined) delete process.env.MAINTENANCE_ADMISSION_DRAIN_AGE_MS; + else process.env.MAINTENANCE_ADMISSION_DRAIN_AGE_MS = oldEnv; + } + }); + + it("does not drain-admit when host load is ALSO high", async () => { + const oldEnv = process.env.MAINTENANCE_ADMISSION_DRAIN_AGE_MS; + process.env.MAINTENANCE_ADMISSION_DRAIN_AGE_MS = "60000"; + vi.mocked(hostLoadAvg1PerCore).mockReturnValue(5); + try { + const m = makePool(); + m.setPressureSignals({ maintenance: { cnt: 68, oldest: now } }); + m.enqueueResult({ rows: [], rowCount: 0 }); + m.enqueueResult({ rows: [{ ...maintenanceRow, created_at: now - 61_000 }], rowCount: 1 }); + const started: string[] = []; + const q = createPgQueue(m.pool, async (j) => void started.push(typeOf(j))); + await q.drain(); + expect(started).not.toContain("build-contributor-evidence"); + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("SET status='pending', run_after=GREATEST"), + expect.arrayContaining([expect.stringContaining("host_load_high")]), + ); + } finally { + if (oldEnv === undefined) delete process.env.MAINTENANCE_ADMISSION_DRAIN_AGE_MS; + else process.env.MAINTENANCE_ADMISSION_DRAIN_AGE_MS = oldEnv; + } + }); + it("defers a maintenance job when host load per core is high", async () => { vi.mocked(hostLoadAvg1PerCore).mockReturnValue(5); const m = makePool(); @@ -2070,15 +2261,26 @@ describe("createPgQueue (durable #977)", () => { const q = createPgQueue(m.pool, async (j) => void started.push(typeOf(j))); await q.drain(); expect(started).toEqual(["build-contributor-evidence"]); - expect(await renderMetrics()).toContain( - 'gittensory_jobs_maintenance_trickle_admitted_by_type_total{job_type="build-contributor-evidence"} 1', - ); + const metrics = await renderMetrics(); + expect(metrics).toContain('gittensory_jobs_maintenance_trickle_admitted_by_type_total{job_type="build-contributor-evidence"} 1'); + expect(metrics).toContain('gittensory_jobs_maintenance_admission_granted_under_pressure_total{job_type="build-contributor-evidence",reason="trickle_max_defer_age"} 1'); } finally { if (oldEnv === undefined) delete process.env.MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS; else process.env.MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS = oldEnv; } }); + it("does not record the granted-under-pressure metric for an ordinary pressure_clear admission", async () => { + const m = makePool(); + m.enqueueResult({ rows: [], rowCount: 0 }); + m.enqueueResult({ rows: [maintenanceRow], rowCount: 1 }); + const started: string[] = []; + const q = createPgQueue(m.pool, async (j) => void started.push(typeOf(j))); + await q.drain(); + expect(started).toEqual(["build-contributor-evidence"]); + expect(await renderMetrics()).not.toContain("gittensory_jobs_maintenance_admission_granted_under_pressure_total"); + }); + it("pressureSignals() surfaces the live and maintenance aggregate reads", async () => { const m = makePool(); m.setPressureSignals({ live: { cnt: 2, oldest: now - 1_000 }, maintenance: { cnt: 4, oldest: now - 2_000 } }); diff --git a/test/unit/selfhost-queue-common.test.ts b/test/unit/selfhost-queue-common.test.ts index 07acc2b4b0..e4319350de 100644 --- a/test/unit/selfhost-queue-common.test.ts +++ b/test/unit/selfhost-queue-common.test.ts @@ -18,6 +18,8 @@ import { isForegroundJobPriority, jobCoalesceAbsorbedByKey, jobCoalesceKey, + jobCoalesceMergeKeyPrefix, + jobCoalesceMergedPayload, jobCoalesceSupersededKeyPrefix, jobPriority, matchesGitHubRateLimitAdmissionTarget, @@ -709,7 +711,11 @@ describe("self-host queue common helpers", () => { expect(jobCoalesceKey(payload({ type: "run-agent", requestedBy: "github_comment", runId: "run-abc123" }))).toBe("run-agent:run-abc123"); expect(jobCoalesceKey(payload({ type: "notify-deliver", requestedBy: "notify-evaluate", deliveryId: "del-77" }))).toBe("notify-deliver:del-77"); expect(jobCoalesceKey(payload({ type: "submit-draft", requestedBy: "api", draftId: "draft-9" }))).toBe("submit-draft:draft-9"); - expect(jobCoalesceKey(payload({ type: "notify-evaluate", requestedBy: "webhook", event: { dedupKey: "review_requested:o/r#3:bob" } }))).toBe("notify-evaluate:review_requested:o/r#3:bob"); + expect( + jobCoalesceKey( + payload({ type: "notify-evaluate", requestedBy: "webhook", events: [{ dedupKey: "review_requested:o/r#3:bob" }] }), + ), + ).toBe("notify-evaluate:review_requested:o/r#3:bob"); // Two DISTINCT invocations have distinct ids → distinct keys, so they never merge. expect(jobCoalesceKey(payload({ type: "run-agent", requestedBy: "github_comment", runId: "run-xyz789" }))).toBe("run-agent:run-xyz789"); // A payload missing its id → null (uncoalesced), never a shared key that could drop a distinct job. @@ -717,7 +723,50 @@ describe("self-host queue common helpers", () => { expect(jobCoalesceKey(payload({ type: "notify-deliver", requestedBy: "test" }))).toBeNull(); expect(jobCoalesceKey(payload({ type: "submit-draft", requestedBy: "test" }))).toBeNull(); expect(jobCoalesceKey(payload({ type: "notify-evaluate", requestedBy: "test" }))).toBeNull(); - expect(jobCoalesceKey(payload({ type: "notify-evaluate", requestedBy: "test", event: {} }))).toBeNull(); + expect(jobCoalesceKey(payload({ type: "notify-evaluate", requestedBy: "test", events: [] }))).toBeNull(); + expect(jobCoalesceKey(payload({ type: "notify-evaluate", requestedBy: "test", events: [{}] }))).toBeNull(); + }); + + it("batches a notify-evaluate job's coalesce key off the FULL sorted set of dedup keys (#selfhost-maintenance-self-pin)", () => { + // Order-independent: the same two events in either order produce the same key, so a redelivery with the + // events reordered still coalesces. + const forward = jobCoalesceKey( + payload({ + type: "notify-evaluate", + requestedBy: "webhook", + events: [{ dedupKey: "review_requested:o/r#3:bob" }, { dedupKey: "issue_watch_match:o/r#9:alice" }], + }), + ); + const reversed = jobCoalesceKey( + payload({ + type: "notify-evaluate", + requestedBy: "webhook", + events: [{ dedupKey: "issue_watch_match:o/r#9:alice" }, { dedupKey: "review_requested:o/r#3:bob" }], + }), + ); + expect(forward).toBe("notify-evaluate:issue_watch_match:o/r#9:alice,review_requested:o/r#3:bob"); + expect(reversed).toBe(forward); + // A batch with even one different event gets a DIFFERENT key -- never silently merges with an unrelated batch. + const differentBatch = jobCoalesceKey( + payload({ + type: "notify-evaluate", + requestedBy: "webhook", + events: [{ dedupKey: "review_requested:o/r#3:bob" }, { dedupKey: "issue_watch_match:o/r#9:carol" }], + }), + ); + expect(differentBatch).not.toBe(forward); + // If ANY event in the batch is missing its dedup key, the whole batch is left uncoalesced (null) rather than + // key off a partial set that could collide with -- and silently drop the malformed event from -- an + // unrelated batch. + expect( + jobCoalesceKey( + payload({ + type: "notify-evaluate", + requestedBy: "webhook", + events: [{ dedupKey: "review_requested:o/r#3:bob" }, {}], + }), + ), + ).toBeNull(); }); it("coalesces recurring maintenance jobs while preserving their semantic scope", () => { @@ -847,6 +896,94 @@ describe("self-host queue common helpers", () => { ); }); + describe("rag-index-repo incremental merge coalescing (#selfhost-maintenance-self-pin)", () => { + const incrementalA = payload({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/Gittensory", + paths: ["src/a.ts"], + }); + const incrementalB = payload({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/Gittensory", + paths: ["src/b.ts"], + }); + const fullJob = payload({ + type: "rag-index-repo", + requestedBy: "schedule", + repoFullName: "JSONbored/Gittensory", + }); + + it("returns the repo-scoped prefix only for an incoming INCREMENTAL job", () => { + expect(jobCoalesceMergeKeyPrefix(incrementalA)).toBe("rag-index-repo:jsonbored/gittensory:"); + expect(jobCoalesceMergeKeyPrefix(fullJob)).toBeNull(); // a full job supersedes instead — see jobCoalesceSupersededKeyPrefix + expect(jobCoalesceMergeKeyPrefix(payload({ type: "notify-evaluate", requestedBy: "test" }))).toBeNull(); + expect(jobCoalesceMergeKeyPrefix(payload({ type: "rag-index-repo", requestedBy: "webhook" }))).toBeNull(); // no repo + }); + + it("unions two incremental jobs' paths, deduped and sorted, into the incoming job's shape", () => { + const merged = jobCoalesceMergedPayload(incrementalA, incrementalB); + expect(merged).not.toBeNull(); + expect(JSON.parse(merged!)).toEqual({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/Gittensory", + paths: ["src/a.ts", "src/b.ts"], + }); + // The merged key coalesces the SAME as a single job enqueued with the union directly. + expect(jobCoalesceKey(merged!)).toBe( + jobCoalesceKey( + payload({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/Gittensory", + paths: ["src/a.ts", "src/b.ts"], + }), + ), + ); + }); + + it("dedupes an overlapping path re-merged from both sides", () => { + const merged = jobCoalesceMergedPayload(incrementalA, incrementalA); + expect(JSON.parse(merged!).paths).toEqual(["src/a.ts"]); + }); + + it("returns null when either side isn't a path-scoped rag-index-repo message", () => { + expect(jobCoalesceMergedPayload(fullJob, incrementalA)).toBeNull(); // existing side has no paths + expect(jobCoalesceMergedPayload(incrementalA, fullJob)).toBeNull(); // incoming side has no paths + expect(jobCoalesceMergedPayload(payload({ type: "notify-evaluate", requestedBy: "test" }), incrementalA)).toBeNull(); + expect(jobCoalesceMergedPayload(incrementalA, payload({ type: "notify-evaluate", requestedBy: "test" }))).toBeNull(); + }); + + it("does not merge past the bounded path cap -- falls back to a separate row instead of unbounded growth", () => { + const nearCapExisting = payload({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/Gittensory", + paths: Array.from({ length: 99 }, (_, i) => `src/${i}.ts`), + }); + const twoMorePaths = payload({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/Gittensory", + paths: ["src/extra-1.ts", "src/extra-2.ts"], + }); + // 99 + 2 unique = 101 > the 100 cap → refuse to merge. + expect(jobCoalesceMergedPayload(nearCapExisting, twoMorePaths)).toBeNull(); + // Exactly at the cap (99 + 1 unique = 100) still merges. + const oneMorePath = payload({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/Gittensory", + paths: ["src/extra-1.ts"], + }); + const merged = jobCoalesceMergedPayload(nearCapExisting, oneMorePath); + expect(merged).not.toBeNull(); + expect(JSON.parse(merged!).paths).toHaveLength(100); + }); + }); + it("keys build-contributor-evidence by login/all, and fanned-out batches by their FIRST login (never one shared key) (#1941)", () => { // A single-login (re-index) job coalesces by login; the scheduled trigger (no login/logins) → the "all" slot. expect(jobCoalesceKey(payload({ type: "build-contributor-evidence", requestedBy: "schedule", login: "Alice" }))).toBe("build-contributor-evidence:alice"); diff --git a/test/unit/selfhost-sqlite-queue.test.ts b/test/unit/selfhost-sqlite-queue.test.ts index bc5c5df7c8..4e6d396ef7 100644 --- a/test/unit/selfhost-sqlite-queue.test.ts +++ b/test/unit/selfhost-sqlite-queue.test.ts @@ -2,7 +2,7 @@ import { DatabaseSync } from "node:sqlite"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { nodeSqliteDriver } from "../../src/selfhost/d1-adapter"; import { createSqliteQueue } from "../../src/selfhost/sqlite-queue"; -import { queueSnapshotFromBinding } from "../../src/selfhost/queue-common"; +import { jobCoalesceKey, queueSnapshotFromBinding } from "../../src/selfhost/queue-common"; import { renderMetrics, resetMetrics } from "../../src/selfhost/metrics"; import { RetryableJobError } from "../../src/queue/retryable"; import { hostLoadAvg1PerCore } from "../../src/selfhost/host-pressure"; @@ -934,12 +934,119 @@ describe("createSqliteQueue (durable #980)", () => { requestedBy: "schedule", repoFullName: "JSONbored/gittensory", }); + // The two incrementals now MERGE into one row before the full job supersedes it (#selfhost-maintenance-self-pin): + // 1 insert (the first incremental) + 2 coalesces (the merge, then the supersede), not 2 inserts + 1 coalesce. expect(q.stats()).toMatchObject({ - gittensory_jobs_enqueued_total: 2, + gittensory_jobs_enqueued_total: 1, + gittensory_jobs_coalesced_total: 2, + }); + }); + + // #selfhost-maintenance-self-pin: several merge-triggered incremental RAG jobs for the SAME repo, arriving + // while one is still pending, union their paths into ONE row instead of piling up as separate maintenance-lane + // entries -- distinct from the absorb/supersede pair above, which only ever involve a FULL job on one side. + it("merges two pending incremental RAG jobs for the same repo into one row's union path set", async () => { + const driver = makeDriver(); + const q = createSqliteQueue(driver, async () => undefined); + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/gittensory", + paths: ["src/a.ts"], + }, { delaySeconds: 60 }); + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/gittensory", + paths: ["src/b.ts"], + }, { delaySeconds: 60 }); + + const rows = driver.query( + "SELECT payload, job_key FROM _selfhost_jobs ORDER BY id", + [], + ).rows as Array<{ payload: string; job_key: string }>; + expect(rows).toHaveLength(1); + expect(JSON.parse(rows[0]!.payload)).toEqual({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/gittensory", + paths: ["src/a.ts", "src/b.ts"], + }); + expect(rows[0]?.job_key).toBe(jobCoalesceKey(rows[0]!.payload)); + expect(q.stats()).toMatchObject({ + gittensory_jobs_enqueued_total: 1, gittensory_jobs_coalesced_total: 1, }); }); + it("does not merge a repo's incremental into an already-pending FULL job for that repo", async () => { + const driver = makeDriver(); + const q = createSqliteQueue(driver, async () => undefined); + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "schedule", + repoFullName: "JSONbored/gittensory", + }, { delaySeconds: 60 }); + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/gittensory", + paths: ["src/a.ts"], + }, { delaySeconds: 1 }); + + // Absorbed by the existing FULL job (unchanged behavior), never merged into a narrower shape. + const rows = driver.query("SELECT payload FROM _selfhost_jobs ORDER BY id", []).rows as Array<{ payload: string }>; + expect(rows).toHaveLength(1); + expect(JSON.parse(rows[0]!.payload)).toEqual({ + type: "rag-index-repo", + requestedBy: "schedule", + repoFullName: "JSONbored/gittensory", + }); + }); + + it("does not merge incrementals across DIFFERENT repos", async () => { + const driver = makeDriver(); + const q = createSqliteQueue(driver, async () => undefined); + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/gittensory", + paths: ["src/a.ts"], + }, { delaySeconds: 60 }); + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/metagraphed", + paths: ["src/b.ts"], + }, { delaySeconds: 60 }); + + const rows = driver.query("SELECT payload FROM _selfhost_jobs ORDER BY id", []).rows as Array<{ payload: string }>; + expect(rows).toHaveLength(2); + }); + + it("falls through to a separate row when merging would exceed the bounded path cap", async () => { + const driver = makeDriver(); + const q = createSqliteQueue(driver, async () => undefined); + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/gittensory", + paths: Array.from({ length: 100 }, (_, i) => `src/${i}.ts`), // already at the cap + }, { delaySeconds: 60 }); + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "webhook", + repoFullName: "JSONbored/gittensory", + paths: ["src/extra.ts"], + }, { delaySeconds: 60 }); + + // No merge (would be 101 paths, over the cap) -- the second send lands as its OWN row instead. + const rows = driver.query("SELECT payload FROM _selfhost_jobs ORDER BY id", []).rows as Array<{ payload: string }>; + expect(rows).toHaveLength(2); + expect(JSON.parse(rows[0]!.payload).paths).toHaveLength(100); + expect(JSON.parse(rows[1]!.payload).paths).toEqual(["src/extra.ts"]); + }); + it("snapshot() reports pending/processing/dead queue depth by job type", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined); @@ -1039,26 +1146,36 @@ describe("createSqliteQueue (durable #980)", () => { [], ).rows as Array<{ payload: string; job_key: string }>; - expect(rows).toHaveLength(6); + // The third rag-index-repo send (paths: ["src/c.ts"]) now MERGES into the same-repo incremental row the + // first two sends already coalesced onto (#selfhost-maintenance-self-pin), instead of becoming its own row -- + // 5 rows, not 6, and one fewer coalesce boundary than before that merge existed. + const mergedRagKey = jobCoalesceKey( + JSON.stringify({ + type: "rag-index-repo", + requestedBy: "schedule", + repoFullName: "JSONbored/gittensory", + paths: ["src/a.ts", "src/b.ts", "src/c.ts"], + }), + ); + expect(rows).toHaveLength(5); expect(rows.map((row) => row.job_key)).toEqual([ "backfill-registered-repos:jsonbored/gittensory:resume:1", "backfill-registered-repos:jsonbored/gittensory:light:1", "generate-weekly-value-report:operator:7", "generate-weekly-value-report:public:7", - "rag-index-repo:jsonbored/gittensory:sha256:170cb2cfb288ab59ba4d35b2633120223c9acc6893fd5baec3465c434ad5bedf", - "rag-index-repo:jsonbored/gittensory:sha256:f4f9970f7a842b1b7b619cbd49f05da577a7d725ff1616ba2de8beed1ae5616f", + mergedRagKey, ]); + expect(JSON.parse(rows[4]!.payload).paths).toEqual(["src/a.ts", "src/b.ts", "src/c.ts"]); expect(rows.map((row) => JSON.parse(row.payload).requestedBy)).toEqual([ "api", "api", "api", "api", "schedule", - "schedule", ]); expect(q.stats()).toMatchObject({ - gittensory_jobs_enqueued_total: 6, - gittensory_jobs_coalesced_total: 3, + gittensory_jobs_enqueued_total: 5, + gittensory_jobs_coalesced_total: 4, }); }); @@ -2260,6 +2377,7 @@ describe("createSqliteQueue (durable #980)", () => { "MAINTENANCE_ADMISSION_MAX_HOST_LOAD", "MAINTENANCE_ADMISSION_DEFER_MS", "MAINTENANCE_ADMISSION_MAX_DEFER_AGE_MS", + "MAINTENANCE_ADMISSION_DRAIN_AGE_MS", ] as const; const saved: Record = {}; @@ -2371,6 +2489,71 @@ describe("createSqliteQueue (durable #980)", () => { expect(row.last_error).toContain("maintenance_pending_high"); }); + // Regression (#selfhost-maintenance-self-pin): the reported incident had a maintenance lane backed up well + // past the threshold with EVERY claim denied `maintenance_pending_high`, and no way for the backlog to shrink + // short of each job individually reaching the 4h trickle. The drain escape lets the OLDEST jobs in that same + // backlog through in a bounded trickle well before that. + it("drain-admits the oldest job in a large backlog once it has waited past the drain age, while a fresh job in the SAME backlog still defers", async () => { + process.env.MAINTENANCE_ADMISSION_DRAIN_AGE_MS = "60000"; // 1m (parsePositiveIntEnv floor) + const driver = makeDriver(); + const started: string[] = []; + const q = createSqliteQueue(driver, async (m) => void started.push(typeOf(m))); + const now = Date.now(); + for (let i = 0; i < 68; i += 1) { // mirrors the reported incident's backlog size, well over the threshold + driver.query( + `INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, is_maintenance) + VALUES (?, 'pending', 0, ?, ?, 0, NULL, 1)`, + [JSON.stringify({ type: "rollup-product-usage", requestedBy: "test" }), now + 3_600_000, now], + ); + } + const staleCreatedAt = now - 61_000; + driver.query( + `INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, is_maintenance) + VALUES (?, 'pending', 0, 0, ?, 0, NULL, 1)`, + [JSON.stringify({ type: "build-contributor-evidence", requestedBy: "schedule" }), staleCreatedAt], + ); + await q.binding.send(msg("notify-evaluate")); + await q.drain(); + expect(started).toContain("build-contributor-evidence"); // old job in the backlog: drained + expect(started).not.toContain("notify-evaluate"); // fresh job in the SAME backlog: still deferred + const freshRow = driver.query( + "SELECT last_error FROM _selfhost_jobs WHERE payload LIKE '%notify-evaluate%'", + [], + ).rows[0] as { last_error: string }; + expect(freshRow.last_error).toContain("maintenance_pending_high"); + expect(await renderMetrics()).toContain( + 'gittensory_jobs_maintenance_admission_granted_under_pressure_total{job_type="build-contributor-evidence",reason="maintenance_pending_high_drain"} 1', + ); + }); + + it("does not drain-admit when host load is ALSO high", async () => { + process.env.MAINTENANCE_ADMISSION_DRAIN_AGE_MS = "60000"; + vi.mocked(hostLoadAvg1PerCore).mockReturnValue(5); + const driver = makeDriver(); + const started: string[] = []; + const q = createSqliteQueue(driver, async (m) => void started.push(typeOf(m))); + const now = Date.now(); + for (let i = 0; i < 16; i += 1) { + driver.query( + `INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, is_maintenance) + VALUES (?, 'pending', 0, ?, ?, 0, NULL, 1)`, + [JSON.stringify({ type: "rollup-product-usage", requestedBy: "test" }), now + 3_600_000, now], + ); + } + driver.query( + `INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, is_maintenance) + VALUES (?, 'pending', 0, 0, ?, 0, NULL, 1)`, + [JSON.stringify({ type: "build-contributor-evidence", requestedBy: "schedule" }), now - 61_000], + ); + await q.drain(); + expect(started).not.toContain("build-contributor-evidence"); + const row = driver.query( + "SELECT last_error FROM _selfhost_jobs WHERE payload LIKE '%build-contributor-evidence%'", + [], + ).rows[0] as { last_error: string }; + expect(row.last_error).toContain("host_load_high"); + }); + it("defers a maintenance job when host load per core exceeds the threshold", async () => { vi.mocked(hostLoadAvg1PerCore).mockReturnValue(5); const driver = makeDriver(); @@ -2412,9 +2595,9 @@ describe("createSqliteQueue (durable #980)", () => { await q.drain(); expect(started).toEqual(["build-contributor-evidence"]); expect(q.stats()).toMatchObject({ gittensory_jobs_maintenance_trickle_admitted_total: 1 }); - expect(await renderMetrics()).toContain( - 'gittensory_jobs_maintenance_trickle_admitted_by_type_total{job_type="build-contributor-evidence"} 1', - ); + const metrics = await renderMetrics(); + expect(metrics).toContain('gittensory_jobs_maintenance_trickle_admitted_by_type_total{job_type="build-contributor-evidence"} 1'); + expect(metrics).toContain('gittensory_jobs_maintenance_admission_granted_under_pressure_total{job_type="build-contributor-evidence",reason="trickle_max_defer_age"} 1'); }); it("does not record a trickle-admitted metric on a normal clear-pressure admission", async () => { @@ -2428,6 +2611,16 @@ describe("createSqliteQueue (durable #980)", () => { expect(await renderMetrics()).not.toContain("gittensory_jobs_maintenance_trickle_admitted"); }); + it("does not record the granted-under-pressure metric for an ordinary pressure_clear admission", async () => { + const driver = makeDriver(); + const started: string[] = []; + const q = createSqliteQueue(driver, async (m) => void started.push(typeOf(m))); + await q.binding.send(msg("build-contributor-evidence")); + await q.drain(); + expect(started).toEqual(["build-contributor-evidence"]); + expect(await renderMetrics()).not.toContain("gittensory_jobs_maintenance_admission_granted_under_pressure_total"); + }); + it("pressureSignals() reports live/maintenance pending counts and oldest ages", async () => { const driver = makeDriver(); const q = createSqliteQueue(driver, async () => undefined);