feat(HNT-2120): run the agent tick loop with Corpus refresh and metrics

services/crawl-agent/package.json

@@ -15,7 +15,8 @@
     "sentry": "workspace:*",
     "express": "^5.1.0",
     "pubsub": "workspace:*",
-    "redis-state": "workspace:*"
+    "redis-state": "workspace:*",
+    "metrics": "workspace:*"
   },
   "devDependencies": {
     "eslint": "^10.0.0",

services/crawl-agent/src/main.ts

@@ -1,10 +1,20 @@
-// Initialize Sentry first to capture errors from other modules.
+// Initialize Sentry and metrics first to capture from other modules.
 import './sentry-init.js';
-
+import './metrics-init.js';
 import { setTimeout as delay } from 'node:timers/promises';
+import { initCorpusApiClient, type PublisherList } from 'crawl-common';
+import { initPubSubClient, shutdownPubSub } from 'pubsub';
+import { initRedisClient, shutdownRedis } from 'redis-state';
+import { count, shutdownMetrics, timing } from 'metrics';
 import { shutdownSentry, withSentryHandler } from 'sentry';
 import { app, isRunning, setLastTickAt, stopRunning } from './app.js';
 import config from './config.js';
+import {
+  corpusSourceEnabled,
+  fetchLiveArticles,
+  loadPublisherList,
+} from './publisher-list.js';
+import { runTick } from './tick.js';
 
 const server = app.listen(config.port, () => {
   console.log(`crawl-agent listening on port ${config.port}`);
@@ -14,43 +24,73 @@ const server = app.listen(config.port, () => {
 const ac = new AbortController();
 const SHUTDOWN_TIMEOUT_MS = 10_000;
 
+// pages are loaded once at startup; live_articles are refreshed from
+// the Corpus API on an interval when that source is configured.
+let publisherList: PublisherList;
+let lastCorpusRefreshAt = 0;
+
 let shuttingDown = false;
 /**
- * Initiate graceful shutdown: stop the tick loop, close the server,
- * flush Sentry, and force-exit after a timeout. K8s sends SIGTERM
- * before pod termination; a clean shutdown prevents duplicate
- * Pub/Sub message processing and ensures errors reach Sentry.
+ * Initiate graceful shutdown: stop the tick loop, drain Pub/Sub and
+ * Redis, close the server, flush Sentry, and force-exit after a
+ * timeout. K8s sends SIGTERM before pod termination; draining first
+ * flushes pending publishes and reports errors to Sentry.
  */
-function shutdown() {
+async function shutdown() {
   if (shuttingDown) return;
   shuttingDown = true;
   console.log('Shutting down');
   stopRunning();
   ac.abort();
-  server.close(async () => {
-    await shutdownSentry();
-    process.exit(0);
-  });
-  setTimeout(() => {
-    console.error(`Forced exit after ${SHUTDOWN_TIMEOUT_MS}ms timeout`);
-    process.exit(1);
-  }, SHUTDOWN_TIMEOUT_MS).unref();
   process.removeListener('SIGTERM', shutdown);
   process.removeListener('SIGINT', shutdown);
+
+  const forceExit = setTimeout(() => {
+    console.error(`Forced exit after ${SHUTDOWN_TIMEOUT_MS}ms timeout`);
+    process.exit(1);
+  }, SHUTDOWN_TIMEOUT_MS);
+  forceExit.unref();
+
+  let exitCode = 0;
+  // Flush pending publishes, then close Redis. Log failures but keep
+  // going so the server close and Sentry flush always run.
+  try {
+    await shutdownPubSub();
+  } catch (err) {
+    console.error('Error draining Pub/Sub:', err);
+    exitCode = 1;
+  }
+  try {
+    await shutdownRedis();
+  } catch (err) {
+    console.error('Error closing Redis:', err);
+    exitCode = 1;
+  }
+  const serverClosed = new Promise<void>((resolve) =>
+    server.close(() => resolve()),
+  );
+  server.closeAllConnections();
+  await serverClosed;
+  await Promise.all([shutdownMetrics(), shutdownSentry()]);
+  process.exit(exitCode);
 }
 
 process.on('SIGTERM', shutdown);
 process.on('SIGINT', shutdown);
 
 /**
- * Execute a single crawl cycle. Currently only logs the time.
- * Eventually, each tick will check publisher pages and live
- * articles against Redis timestamps, then publish due items
- * to the crawl Pub/Sub queues for worker processing.
+ * Execute a single crawl cycle: enqueue due pages and live articles.
+ * The loop records the tick time separately, so /healthz tracks loop
+ * liveness rather than whether a tick succeeded.
  */
 async function tick() {
-  console.log('tick', new Date().toISOString());
-  setLastTickAt(Date.now());
+  const counts = await runTick(publisherList);
+  count('crawl.tick.enqueued', counts.pages, { kind: 'page' });
+  count('crawl.tick.enqueued', counts.liveArticles, { kind: 'live_article' });
+  console.log(
+    `tick enqueued ${counts.pages} pages, ` +
+      `${counts.liveArticles} live articles`,
+  );
 }
 
 // tick() wrapped to report its errors to Sentry with startedAt context.
@@ -60,24 +100,58 @@ const tickWithSentry = withSentryHandler<{ startedAt: string }>(
 );
 
 /**
- * Run the tick loop at the configured interval until the
- * process is signalled to stop.
+ * Re-read live articles from the Corpus API and swap them into the
+ * publisher list. Throws on failure so the caller keeps the last good
+ * list rather than dropping live articles; a transient Corpus outage
+ * degrades freshness, not availability.
+ */
+async function refreshLiveArticles() {
+  publisherList.live_articles = await fetchLiveArticles();
+  lastCorpusRefreshAt = Date.now();
+  console.log(
+    `Refreshed ${publisherList.live_articles.length} live articles from Corpus`,
+  );
+}
+
+const refreshWithSentry = withSentryHandler<{ startedAt: string }>(
+  ({ startedAt }) => ({ context: { startedAt, kind: 'corpus-refresh' } }),
+  refreshLiveArticles,
+);
+
+/**
+ * Run the tick loop at the configured interval until the process is
+ * signalled to stop.
  */
 async function run() {
   while (isRunning()) {
     const start = Date.now();
     const startedAt = new Date(start).toISOString();
+    if (
+      corpusSourceEnabled() &&
+      start - lastCorpusRefreshAt >= config.corpusRefreshMinutes * 60_000
+    ) {
+      try {
+        await refreshWithSentry({ startedAt });
+      } catch (err) {
+        // Keep the last good list and retry next tick; the wrapper
+        // already reported the failure to Sentry.
+        console.error('Corpus refresh failed; keeping last list:', err);
+      }
+    }
     try {
-      // Wrap tick() with a Sentry emission helper because it's the
-      // outermost frame that bubbles errors up before the catch swallows them.
       await tickWithSentry({ startedAt });
     } catch (err) {
-      // tickWithSentry already captured the error and rethrew so
-      // the loop sees it; swallowing here keeps a single failed
-      // tick from killing the agent.
+      // tickWithSentry already captured the error and rethrew so the
+      // loop sees it; swallowing here keeps a single failed tick from
+      // killing the agent.
       console.error('tick failed:', err);
     }
+    // Mark the loop alive even on a failed tick: /healthz should detect
+    // a wedged loop, not a transient Pub/Sub or Redis outage that a
+    // restart would not fix and that Sentry already surfaces.
+    setLastTickAt(Date.now());
     const elapsed = Date.now() - start;
+    timing('crawl.tick.duration_ms', elapsed);
     const remainingMs = Math.max(0, config.tickIntervalMs - elapsed);
     try {
       await delay(remainingMs, undefined, { signal: ac.signal });
@@ -88,7 +162,59 @@ async function run() {
   }
 }
 
-run().catch((err) => {
-  console.error('run loop crashed:', err);
+/**
+ * Initialize the Pub/Sub and Redis clients, load the publisher list,
+ * then start the tick loop. Any failure here aborts startup. When
+ * crawling is disabled (the dev sandbox) it serves health checks only.
+ */
+async function start() {
+  // The agent is the sole origin of crawl jobs, so skipping the tick
+  // loop here stops the whole pipeline at the source without touching
+  // the workers (they idle with nothing to consume).
+  if (!config.crawlEnabled) {
+    console.log(
+      'crawl-agent: scheduled crawling disabled, serving health only',
+    );
+    return;
+  }
+  initPubSubClient({
+    projectId: config.projectId,
+    apiEndpoint: config.pubsubEmulatorHost,
+    useEmulator: Boolean(config.pubsubEmulatorHost),
+  });
+  initRedisClient({ host: config.redisHost, port: config.redisPort });
+  publisherList = await loadPublisherList(config.publisherListPath);
+  const useCorpus = corpusSourceEnabled();
+  if (useCorpus) {
+    await initCorpusApiClient({
+      ...config.corpusApi,
+      clientName: config.service,
+    });
+    // First load fails fast: a misconfigured or unauthorized Corpus
+    // client should abort startup, not run a degraded agent.
+    await refreshLiveArticles();
+  }
+  const liveSource = useCorpus
+    ? `Corpus: ${config.scheduledSurfaceGuids.join(', ')}`
+    : 'from file';
+  console.log(
+    `Loaded ${publisherList.pages.length} pages, ` +
+      `${publisherList.live_articles.length} live articles (${liveSource})`,
+  );
+  await run();
+}
+
+// Wrap start so a startup failure is captured before exit. Otherwise
+// the rejection is caught below and a crash loop never reaches Sentry.
+const startWithSentry = withSentryHandler<void>(
+  () => ({ context: { kind: 'startup' } }),
+  start,
+);
+
+startWithSentry().catch(async (err) => {
+  console.error('agent startup failed:', err);
+  // Flush the captured event before exit. The process ends here, so no
+  // later shutdown will flush it.
+  await shutdownSentry();
   process.exit(1);
 });

services/crawl-agent/src/metrics-init.ts

@@ -0,0 +1,4 @@
+import { initMetrics } from 'metrics';
+import config from './config.js';
+
+initMetrics({ service: config.service });