fix: chunk rank_packages() apply to avoid Sequin replication slot growth [CM-1374]

backend/src/osspckgs/migrations/V1787733800__rank_packages_chunked_apply.sql

@@ -0,0 +1,153 @@
+-- rank_packages() applied its ranking to all of `packages` in one UPDATE (9.66M
+-- rows), which under REPLICA IDENTITY FULL blew up the Sequin replication slot.
+-- rank_packages_chunked() stages the same ranking into an UNLOGGED table, then
+-- applies it in committed keyset chunks so the slot advances continuously.
+-- rank_packages() is left in place until the new worker deploys.
+
+CREATE UNLOGGED TABLE IF NOT EXISTS staging.package_rank (
+    package_id        bigint PRIMARY KEY,
+    impact            numeric(10, 4),
+    is_critical       bool NOT NULL,
+    rank_in_ecosystem int  NOT NULL
+);
+
+CREATE OR REPLACE PROCEDURE rank_packages_chunked(
+    coverage_cutoff numeric DEFAULT 0.90,
+    ecosystems      text[]  DEFAULT NULL,
+    chunk_size      int     DEFAULT 25000,
+    INOUT applied_rows int  DEFAULT 0
+)
+LANGUAGE plpgsql AS $$
+DECLARE
+    effective_ecosystems text[];
+    staged_count          int;
+    batch_rows            int;
+    cursor_id             bigint := 0;
+BEGIN
+    SET LOCAL max_parallel_workers_per_gather = 4;
+
+    IF chunk_size IS NULL OR chunk_size <= 0 THEN
+        RAISE EXCEPTION 'rank_packages_chunked: chunk_size must be a positive integer, got %', chunk_size;
+    END IF;
+
+    -- Session-level: survives the internal COMMITs below
+    IF NOT pg_try_advisory_lock(hashtextextended('rank_packages_chunked', 0)) THEN
+        RAISE EXCEPTION 'rank_packages_chunked: another execution is already in progress';
+    END IF;
+
+    applied_rows := 0;
+
+    IF ecosystems IS NULL THEN
+        SELECT ARRAY_AGG(DISTINCT ecosystem)
+          INTO effective_ecosystems
+          FROM packages;
+    ELSE
+        effective_ecosystems := ecosystems;
+    END IF;
+
+    TRUNCATE staging.package_rank;
+
+    -- Scoring CTE chain, unchanged from rank_packages() (V1783123201).
+    INSERT INTO staging.package_rank (package_id, impact, is_critical, rank_in_ecosystem)
+    WITH base AS (
+        SELECT
+            id,
+            ecosystem,
+            COALESCE(downloads_last_30d,         0) AS downloads,
+            COALESCE(dependent_count,            0) AS direct_dependents,
+            COALESCE(transitive_dependent_count, 0) AS transitive_dependents,
+            COALESCE(sonatype_popularity_score,  0) AS sonatype_popularity,
+            SUM(COALESCE(downloads_last_30d,         0)) OVER (PARTITION BY ecosystem) AS ecosystem_total_downloads,
+            SUM(COALESCE(dependent_count,            0)) OVER (PARTITION BY ecosystem) AS ecosystem_total_direct_dependents,
+            SUM(COALESCE(transitive_dependent_count, 0)) OVER (PARTITION BY ecosystem) AS ecosystem_total_transitive_dependents,
+            SUM(COALESCE(sonatype_popularity_score,  0)) OVER (PARTITION BY ecosystem) AS ecosystem_total_sonatype
+        FROM packages
+        WHERE ecosystem = ANY(effective_ecosystems)
+    ),
+    walked AS (
+        SELECT
+            id,
+            ecosystem,
+            SUM(signal_value) OVER coverage_window / ecosystem_signal_total::numeric                  AS cumulative_share_inclusive,
+            (SUM(signal_value) OVER coverage_window - signal_value) / ecosystem_signal_total::numeric AS cumulative_share_exclusive
+        FROM base
+        CROSS JOIN LATERAL (VALUES
+            ('downloads',             downloads,             ecosystem_total_downloads),
+            ('direct_dependents',     direct_dependents,     ecosystem_total_direct_dependents),
+            ('transitive_dependents', transitive_dependents, ecosystem_total_transitive_dependents),
+            ('sonatype_popularity',   sonatype_popularity,   ecosystem_total_sonatype)
+        ) AS signal(signal_name, signal_value, ecosystem_signal_total)
+        WHERE ecosystem_signal_total > 0
+        WINDOW coverage_window AS (
+            PARTITION BY ecosystem, signal_name
+            ORDER BY signal_value DESC, id
+            ROWS UNBOUNDED PRECEDING
+        )
+    ),
+    combined AS (
+        SELECT
+            id,
+            ecosystem,
+            AVG(1.0 - cumulative_share_inclusive)::numeric(10, 4) AS new_impact,
+            BOOL_OR(cumulative_share_exclusive < coverage_cutoff)  AS new_is_critical
+        FROM walked
+        GROUP BY id, ecosystem
+    ),
+    final AS (
+        SELECT
+            combined.id,
+            combined.new_impact,
+            combined.new_is_critical OR (spotlight.package_id IS NOT NULL) AS new_is_critical,
+            ROW_NUMBER() OVER (
+                PARTITION BY combined.ecosystem
+                ORDER BY combined.new_impact DESC NULLS LAST, combined.id
+            ) AS new_rank_in_ecosystem
+        FROM combined
+        LEFT JOIN package_criticality_spotlight spotlight ON spotlight.package_id = combined.id
+    )
+    SELECT id, new_impact, new_is_critical, new_rank_in_ecosystem::int
+      FROM final;
+
+    GET DIAGNOSTICS staged_count = ROW_COUNT;
+
+    IF staged_count = 0 THEN
+        RAISE EXCEPTION 'rank_packages_chunked: computed 0 rows, refusing to apply an empty ranking';
+    END IF;
+
+    ANALYZE staging.package_rank;
+
+    COMMIT;
+
+    LOOP
+        WITH batch AS (
+            SELECT package_id, impact, is_critical, rank_in_ecosystem
+              FROM staging.package_rank
+             WHERE package_id > cursor_id
+             ORDER BY package_id
+             LIMIT chunk_size
+        ),
+        updated AS (
+            UPDATE packages p
+               SET impact            = b.impact,
+                   is_critical       = b.is_critical,
+                   rank_in_ecosystem = b.rank_in_ecosystem,
+                   last_rank_pass_at = NOW(),
+                   last_synced_at    = NOW()
+              FROM batch b
+             WHERE p.id = b.package_id
+            RETURNING p.id
+        )
+        SELECT COUNT(*), COALESCE(MAX(b.package_id), cursor_id)
+          INTO batch_rows, cursor_id
+          FROM batch b;
+
+        applied_rows := applied_rows + batch_rows;
+
+        COMMIT;
+
+        EXIT WHEN batch_rows < chunk_size;
+    END LOOP;
+
+    PERFORM pg_advisory_unlock(hashtextextended('rank_packages_chunked', 0));
+END;
+$$;

services/apps/packages_worker/src/criticality/activities.ts

@@ -3,7 +3,7 @@ import { Context } from '@temporalio/activity'
 import { createIngestJob, findPendingJobByKind, markJobStatus } from '@crowd/data-access-layer'
 import { getServiceChildLogger } from '@crowd/logging'
 
-import { getPackagesDb } from '../db'
+import { getPackagesDb, getPackagesDbConnection } from '../db'
 
 import { buildGraph, computePageRank } from './graph'
 import { loadDirectEdges, mergeCentralityScores } from './queries'
@@ -65,7 +65,7 @@ export async function criticalityComputePageRank(
   return { ecosystem, nodeCount: graph.N, edgeCount, iterations, durationMs: Date.now() - start }
 }
 
-export async function rankPackages(): Promise<{ scoredRows: number; rankedRows: number }> {
+export async function rankPackages(): Promise<{ appliedRows: number }> {
   const qx = await getPackagesDb()
 
   // On retry, a pending row from the prior attempt may already exist — reuse it.
@@ -75,15 +75,24 @@ export async function rankPackages(): Promise<{ scoredRows: number; rankedRows:
     (await createIngestJob(qx, 'ranking', 'ranking', null))
   try {
     await markJobStatus(qx, jobId, 'merging')
-    const [result] = await qx.select(`SELECT * FROM rank_packages()`)
-    const scoredRows = Number(result.scored_rows ?? 0)
-    const rankedRows = Number(result.ranked_rows ?? 0)
+    // Dedicated connection, killed in `finally` so the procedure's advisory lock
+    // always releases, even mid-run — a recycled pooled connection would keep it held.
+    const db = await getPackagesDbConnection()
+    const conn = await db.connect()
+    let result
+    try {
+      await conn.none(`SET statement_timeout = '75min'`)
+      ;[result] = await conn.query(`CALL rank_packages_chunked(0.90, NULL, 25000, 0)`)
+    } finally {
+      conn.done(true)
+    }
+    const appliedRows = Number(result.applied_rows ?? 0)
     await markJobStatus(qx, jobId, 'done', {
-      rowCountPg: scoredRows,
-      tableRowCounts: { scored: scoredRows, ranked: rankedRows },
+      rowCountPg: appliedRows,
+      tableRowCounts: { applied: appliedRows },
       finishedAt: new Date(),
     })
-    return { scoredRows, rankedRows }
+    return { appliedRows }
   } catch (err) {
     await markJobStatus(qx, jobId, 'failed', {
       errorMessage: (err as Error).message,

services/apps/packages_worker/src/criticality/run-impact.ts

@@ -1,13 +1,14 @@
 #!/usr/bin/env tsx
 
 /**
- * Trigger rank_packages() on demand.
+ * Trigger rank_packages_chunked() on demand.
  *
  * Usage (from services/apps/packages_worker):
  *   pnpm run:impact
  *   pnpm run:impact --cutoff 0.90
  *   pnpm run:impact --ecosystems npm,go
  *   pnpm run:impact --cutoff 0.85 --ecosystems npm
+ *   pnpm run:impact --chunk 5000
  */
 import { getPackagesDb } from '../db'
 
@@ -31,24 +32,26 @@ function parseListArg(flag: string): string[] | null {
 }
 
 const cutoff = parseArg('--cutoff', 0.9)
+const chunk = parseArg('--chunk', 25000)
 const ecosystems = parseListArg('--ecosystems')
 
 async function main() {
-  console.log(`Running rank_packages()`)
+  console.log(`Running rank_packages_chunked()`)
   console.log(`  cutoff    : ${cutoff}`)
+  console.log(`  chunk     : ${chunk}`)
   console.log(`  ecosystems: ${ecosystems ? ecosystems.join(', ') : 'all'}\n`)
 
   const qx = await getPackagesDb()
   const t = Date.now()
 
-  const [result] = await qx.select(`SELECT * FROM rank_packages($/cutoff/, $/ecosystems/)`, {
-    cutoff,
-    ecosystems,
-  })
+  const [result] = await qx.select(
+    `CALL rank_packages_chunked($/cutoff/, $/ecosystems/, $/chunk/, 0)`,
+    { cutoff, ecosystems, chunk },
+  )
 
   const elapsed = ((Date.now() - t) / 1000).toFixed(1)
   console.log(`Done in ${elapsed}s`)
-  console.log(`  processed_rows: ${result.processed_rows?.toLocaleString()}`)
+  console.log(`  applied_rows: ${result.applied_rows?.toLocaleString()}`)
 
   process.exit(0)
 }

services/apps/packages_worker/src/criticality/workflow.ts

@@ -3,7 +3,7 @@ import { proxyActivities } from '@temporalio/workflow'
 import type * as critActivities from './activities'
 
 const { rankPackages } = proxyActivities<typeof critActivities>({
-  startToCloseTimeout: '30 minutes',
+  startToCloseTimeout: '90 minutes',
   retry: { maximumAttempts: 2 },
 })