Split singleton backend work into opt-in worker Deployment (TASK-355)

.github/workflows/tests.yml

@@ -336,13 +336,22 @@ jobs:
             -f k8s/helm/commonly/values.yaml \
             -f k8s/helm/commonly/values-dev.yaml \
             > /tmp/dev-rendered.yaml
+          helm template commonly-dev-worker k8s/helm/commonly \
+            -f k8s/helm/commonly/values.yaml \
+            -f k8s/helm/commonly/values-dev.yaml \
+            --set backend.worker.enabled=true \
+            --set backend.worker.startPaused=false \
+            > /tmp/dev-worker-rendered.yaml
+
+      - name: Verify backend worker render
+        run: bash scripts/verify-backend-worker-chart.sh
 
       - name: Lint rendered dev chart
         run: |
           kubeconform --strict --summary -output text \
             --schema-location "https://raw.githubusercontent.com/datreeio/CRDs-catalog/${CRDS_CATALOG_SHA}/{{.Group}}/{{.ResourceKind}}_{{.ResourceAPIVersion}}.json" \
             --schema-location default \
-            /tmp/dev-rendered.yaml
+            /tmp/dev-rendered.yaml /tmp/dev-worker-rendered.yaml
 
       # The dev render has kubeconform coverage, but staging is otherwise first
       # rendered at deploy time. Parse both values combinations before merge.

backend/__tests__/unit/helm/opsRunnerEnvMirror.test.js

@@ -1,12 +1,14 @@
 /**
  * Drift guard for the ops-runner env mirror.
  *
- * templates/ops/ops-runner-deployment.yaml mirrors the backend container's env
- * + volume mounts VERBATIM (it can't DRY-extract into a helper without changing
+ * templates/ops/ops-runner-deployment.yaml mirrors the backend container's app
+ * dependencies and volume mounts (it can't DRY-extract into a helper without changing
  * backend's byte-for-byte render — the env block has blank lines nindent would
  * re-indent). A "KEEP IN SYNC" banner alone drifts (recorded instances across
  * the chart), so this pins the contract: the two Deployments must declare the
- * SAME env entries — the same (name, source) tuples, where source is a
+ * SAME env entries except RUN_WORKER, which controls the serving process and
+ * is not read by the sleeping ops-runner — the same (name, source) tuples,
+ * where source is a
  * secretKeyRef name+key, a configMapKeyRef name+key, or a literal value. If
  * backend gains/loses an env var, or changes a var's source, and ops-runner
  * isn't updated to match, this fails.
@@ -93,7 +95,14 @@ if (typeof describe === 'function') {
     // ops-runner the same number of times, and vice versa. toEqual on the
     // sorted arrays reports exactly which env entry drifted.
     test('ops-runner declares the same env (name + source) tuples as backend', () => {
-      expect([...opsTuples].sort()).toEqual([...backendTuples].sort());
+      const withoutWorkerGate = (tuple) => !tuple.startsWith('RUN_WORKER\t');
+      expect([...opsTuples].filter(withoutWorkerGate).sort())
+        .toEqual([...backendTuples].filter(withoutWorkerGate).sort());
+      expect(opsTuples.every(withoutWorkerGate)).toBe(true);
+      expect(backendTuples.filter((tuple) => !withoutWorkerGate(tuple))).toEqual([
+        'RUN_WORKER\tvalue:"1"',
+        'RUN_WORKER\tvalue:"0"',
+      ]);
     });
   });
 }

backend/__tests__/unit/server.test.js

@@ -48,6 +48,23 @@ jest.mock('../../config/init-pg-db', () => {
   const initPGDB = mockInitPGDB;
   return initPGDB;
 });
+let mockInitPgRetention = jest.fn();
+jest.mock('../../services/pgRetentionService', () => {
+  const initPgRetention = mockInitPgRetention;
+  return { initPgRetention };
+});
+let mockInitInstallationCleanup = jest.fn();
+jest.mock('../../services/agentInstallationCleanupService', () => {
+  const initInstallationCleanup = mockInitInstallationCleanup;
+  return { initInstallationCleanup };
+});
+let mockInitPgMessageReconcile = jest.fn();
+jest.mock('../../services/pgMessageReconcileService', () => {
+  const initPgMessageReconcile = mockInitPgMessageReconcile;
+  return { initPgMessageReconcile };
+});
+jest.mock('../../services/schedulerService', () => ({ start: jest.fn() }));
+jest.mock('../../services/discordGatewayService', () => ({ start: jest.fn() }));
 
 // Replace pg routes with simple routers
 jest.mock('../../routes/pg-status', () => {
@@ -89,13 +106,17 @@ describe('server pg boot routes', () => {
   beforeEach(() => {
     mockConnectPG = jest.fn().mockResolvedValue(null);
     mockInitPGDB = jest.fn();
+    mockInitPgRetention = jest.fn();
+    mockInitInstallationCleanup = jest.fn();
+    mockInitPgMessageReconcile = jest.fn();
   });
 
   afterEach(() => {
     jest.resetModules();
     jest.clearAllMocks();
     delete process.env.PG_HOST;
     delete process.env.PG_BOOT_RETRY_BASE_DELAY_MS;
+    delete process.env.RUN_WORKER;
   });
 
   it('mounts the real status router even when PG is not configured', async () => {
@@ -161,6 +182,33 @@ describe('server pg boot routes', () => {
     await request(app).get('/api/pg/messages').expect(404);
   });
 
+  it.each([
+    ['web', '0', false],
+    ['worker', '1', true],
+    ['legacy unset', undefined, true],
+  ])('%s boot mounts PG routes and applies the worker cron gate', async (_label, runWorker, shouldStart) => {
+    const priorNodeEnv = process.env.NODE_ENV;
+    process.env.NODE_ENV = 'production';
+    process.env.PG_HOST = 'x';
+    if (runWorker !== undefined) process.env.RUN_WORKER = runWorker;
+    mockConnectPG.mockResolvedValue({});
+    mockInitPGDB.mockResolvedValue(true);
+    try {
+      // This imports the actual server boot composition, including its
+      // createPgBoot onMounted callback. Only the cron starters are mocked.
+      // eslint-disable-next-line global-require, import/no-unresolved, import/extensions
+      const { app } = require('../../server');
+      await new Promise((resolve) => { setTimeout(resolve, 30); });
+      await request(app).get('/api/pg/messages').expect(200);
+      expect(mockInitPgRetention).toHaveBeenCalledTimes(shouldStart ? 1 : 0);
+      expect(mockInitInstallationCleanup).toHaveBeenCalledTimes(shouldStart ? 1 : 0);
+      expect(mockInitPgMessageReconcile).toHaveBeenCalledTimes(shouldStart ? 1 : 0);
+    } finally {
+      if (priorNodeEnv === undefined) delete process.env.NODE_ENV;
+      else process.env.NODE_ENV = priorNodeEnv;
+    }
+  });
+
   it('wires the readiness probe to the route table, not to the boot flag (TASK-168)', () => {
     // The states reachable in this file cannot tell the two apart — when PG is
     // missing, the flag and the route table are both false; when it works, both

backend/__tests__/unit/services/backgroundWorkerStart.test.js

@@ -0,0 +1,34 @@
+const { startBackgroundServices } = require('../../../services/backgroundWorkerStart');
+
+describe('background worker start gate', () => {
+  let scheduler;
+  let discordGateway;
+
+  beforeEach(() => {
+    scheduler = { start: jest.fn() };
+    discordGateway = { start: jest.fn() };
+    jest.spyOn(console, 'log').mockImplementation(() => {});
+  });
+
+  afterEach(() => jest.restoreAllMocks());
+
+  test.each([
+    ['unset', undefined, true],
+    ['on', '1', true],
+    ['off', '0', false],
+  ])('RUN_WORKER %s', (_label, runWorker, shouldStart) => {
+    const env = { NODE_ENV: 'production', DISCORD_BOT_TOKEN: 'token' };
+    if (runWorker !== undefined) env.RUN_WORKER = runWorker;
+
+    expect(startBackgroundServices(env, { scheduler, discordGateway })).toBe(shouldStart);
+    expect(scheduler.start).toHaveBeenCalledTimes(shouldStart ? 1 : 0);
+    expect(discordGateway.start).toHaveBeenCalledTimes(shouldStart ? 1 : 0);
+  });
+
+  test('keeps the test environment and tokenless gateway stopped', () => {
+    expect(startBackgroundServices({ NODE_ENV: 'test', DISCORD_BOT_TOKEN: 'token' }, { scheduler, discordGateway })).toBe(false);
+    expect(startBackgroundServices({ NODE_ENV: 'production' }, { scheduler, discordGateway })).toBe(true);
+    expect(scheduler.start).toHaveBeenCalledTimes(1);
+    expect(discordGateway.start).not.toHaveBeenCalled();
+  });
+});

backend/server.ts

@@ -383,15 +383,11 @@ mongoose.connection.once('open', () => {
 // Start the summarizer scheduler
 const schedulerService = require('./services/schedulerService');
 const discordGatewayService = require('./services/discordGatewayService');
-
-if (process.env.NODE_ENV !== 'test') {
-  console.log('Starting summarizer scheduler...');
-  schedulerService.start();
-
-  if (process.env.DISCORD_BOT_TOKEN) {
-    discordGatewayService.start();
-  }
-}
+const { startBackgroundServices, shouldRunBackgroundServices } = require('./services/backgroundWorkerStart');
+startBackgroundServices(process.env, {
+  scheduler: schedulerService,
+  discordGateway: discordGatewayService,
+});
 
 // PostgreSQL status is mounted unconditionally, and NOT as a placeholder:
 // checkStatus reads the pool and the schema itself, so it answers
@@ -441,7 +437,7 @@ if (process.env.PG_HOST) {
       // Kick off the daily 30-day message retention cron. Kept out
       // of schedulerService.ts on purpose so other tracks can edit
       // that file without stomping on this cron.
-      if (process.env.NODE_ENV !== 'test') {
+      if (shouldRunBackgroundServices(process.env)) {
         try {
           const { initPgRetention } = require('./services/pgRetentionService');
           initPgRetention();

backend/services/backgroundWorkerStart.ts

@@ -0,0 +1,22 @@
+type BackgroundServices = {
+  scheduler: { start: () => void };
+  discordGateway: { start: () => void };
+};
+
+export function startBackgroundServices(
+  env: NodeJS.ProcessEnv,
+  services: BackgroundServices,
+): boolean {
+  if (!shouldRunBackgroundServices(env)) return false;
+
+  console.log('Starting summarizer scheduler...');
+  services.scheduler.start();
+  if (env.DISCORD_BOT_TOKEN) services.discordGateway.start();
+  return true;
+}
+
+export function shouldRunBackgroundServices(env: NodeJS.ProcessEnv): boolean {
+  // Unset keeps single-process installs scheduling until the chart moves the
+  // singleton work into its own Deployment.
+  return env.NODE_ENV !== 'test' && env.RUN_WORKER !== '0';
+}

k8s/helm/commonly/templates/_helpers.tpl

@@ -64,6 +64,17 @@ Backend selector labels
 app: backend
 {{- end }}
 
+{{/* The worker must never match the backend Service selector. */}}
+{{- define "commonly.backendWorker.labels" -}}
+{{ include "commonly.labels" . }}
+app: backend-worker
+{{- end }}
+
+{{- define "commonly.backendWorker.selectorLabels" -}}
+{{ include "commonly.selectorLabels" . }}
+app: backend-worker
+{{- end }}
+
 {{/*
 Frontend labels
 */}}

k8s/helm/commonly/templates/core/backend-deployment.yaml

@@ -1,13 +1,18 @@
-{{- if .Values.backend.enabled }}
+{{- define "commonly.backend.deployment" -}}
 apiVersion: apps/v1
 kind: Deployment
 metadata:
-  name: backend
+  name: {{ if .worker }}backend-worker{{ else }}backend{{ end }}
   namespace: {{ include "commonly.namespace" . }}
   labels:
-    {{- include "commonly.backend.labels" . | nindent 4 }}
+    {{- if .worker }}{{ include "commonly.backendWorker.labels" . | nindent 4 }}{{ else }}{{ include "commonly.backend.labels" . | nindent 4 }}{{ end }}
 spec:
-  replicas: {{ .Values.backend.replicaCount }}
+  replicas: {{ if .worker }}{{ if .Values.backend.worker.startPaused }}0{{ else }}1{{ end }}{{ else }}{{ .Values.backend.replicaCount }}{{ end }}
+  {{- if .worker }}
+  # A rolling surge would briefly run two schedulers and two Discord gateways.
+  strategy:
+    type: Recreate
+  {{- else }}
   # Overlap is safe: the catalog is ephemeral, ObjectStore defaults to Mongo,
   # and request/session state is externalized to Mongo, PostgreSQL, and Redis.
   minReadySeconds: 10
@@ -16,13 +21,14 @@ spec:
     rollingUpdate:
       maxUnavailable: 0
       maxSurge: 1
+  {{- end }}
   selector:
     matchLabels:
-      {{- include "commonly.backend.selectorLabels" . | nindent 6 }}
+      {{- if .worker }}{{ include "commonly.backendWorker.selectorLabels" . | nindent 6 }}{{ else }}{{ include "commonly.backend.selectorLabels" . | nindent 6 }}{{ end }}
   template:
     metadata:
       labels:
-        {{- include "commonly.backend.selectorLabels" . | nindent 8 }}
+        {{- if .worker }}{{ include "commonly.backendWorker.selectorLabels" . | nindent 8 }}{{ else }}{{ include "commonly.backend.selectorLabels" . | nindent 8 }}{{ end }}
     spec:
       serviceAccountName: {{ .Values.rbac.serviceAccount.name }}
       terminationGracePeriodSeconds: 180
@@ -64,6 +70,13 @@ spec:
         # Node Environment
         - name: NODE_ENV
           value: {{ .Values.backend.env.nodeEnv }}
+        {{- if .worker }}
+        - name: RUN_WORKER
+          value: "1"
+        {{- else if .Values.backend.worker.enabled }}
+        - name: RUN_WORKER
+          value: "0"
+        {{- end }}
         - name: PORT
           value: {{ .Values.backend.service.port | quote }}
         - name: HOSTED_MCP_HIDDEN_ENTRIES
@@ -693,11 +706,17 @@ spec:
           periodSeconds: {{ .Values.backend.readinessProbe.periodSeconds }}
         {{- end }}
 
-      {{- if and .Values.backend.nodeSelector (gt (len .Values.backend.nodeSelector) 0) }}
+      {{- if and .worker .Values.backend.worker.nodeSelector (gt (len .Values.backend.worker.nodeSelector) 0) }}
+      nodeSelector:
+        {{- toYaml .Values.backend.worker.nodeSelector | nindent 8 }}
+      {{- else if and (not .worker) .Values.backend.nodeSelector (gt (len .Values.backend.nodeSelector) 0) }}
       nodeSelector:
         {{- toYaml .Values.backend.nodeSelector | nindent 8 }}
       {{- end }}
-      {{- if and .Values.backend.tolerations (gt (len .Values.backend.tolerations) 0) }}
+      {{- if and .worker .Values.backend.worker.tolerations (gt (len .Values.backend.worker.tolerations) 0) }}
+      tolerations:
+        {{- toYaml .Values.backend.worker.tolerations | nindent 8 }}
+      {{- else if and (not .worker) .Values.backend.tolerations (gt (len .Values.backend.tolerations) 0) }}
       tolerations:
         {{- toYaml .Values.backend.tolerations | nindent 8 }}
       {{- end }}
@@ -725,4 +744,7 @@ spec:
         configMap:
           name: skills-catalog
       {{- end }}
+{{- end -}}
+{{- if .Values.backend.enabled }}
+{{ include "commonly.backend.deployment" (merge (dict "worker" false) .) }}
 {{- end }}

k8s/helm/commonly/templates/core/backend-worker-deployment.yaml

@@ -0,0 +1,3 @@
+{{- if and .Values.backend.enabled .Values.backend.worker.enabled }}
+{{ include "commonly.backend.deployment" (merge (dict "worker" true) .) }}
+{{- end }}

k8s/helm/commonly/values-dev.yaml

@@ -34,6 +34,16 @@ agentProvisioning:
 
 backend:
   replicaCount: 1
+  worker:
+    enabled: false
+    # Cron and the Discord gateway must survive spot reclamation.
+    nodeSelector:
+      pool: dev
+    tolerations:
+      - key: "pool"
+        operator: "Equal"
+        value: "dev"
+        effect: "NoSchedule"
   image:
     repository: gcr.io/YOUR_GCP_PROJECT_ID/commonly-backend
     tag: "20260417214615-r4"

k8s/helm/commonly/values.yaml

@@ -88,6 +88,14 @@ agentProvisioning:
 backend:
   enabled: true
   replicaCount: 1
+  worker:
+    # Opt in only when a dedicated worker can replace web's background jobs.
+    enabled: false
+    # Enabling the worker first disables web jobs but keeps worker replicas at
+    # zero. After old web pods terminate, set this false to start the worker.
+    startPaused: true
+    nodeSelector: {}
+    tolerations: []
   image:
     repository: gcr.io/YOUR_GCP_PROJECT_ID/commonly-backend
     tag: latest