feat: Plan 6 PR B — score recompute scheduler + live CI

.github/workflows/ci.yml

@@ -144,18 +144,53 @@ jobs:
         with:
           persist-credentials: false
 
+      - name: Set up Go
+        uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff
+        with:
+          go-version-file: go.mod
+          cache: true
+
       - name: Set up Node
         uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020
         with:
           node-version: 22
           cache: npm
           cache-dependency-path: control-plane/package-lock.json
 
+      - name: Build trust-engine
+        working-directory: .
+        run: go build -o /tmp/verilink-trust-engine ./cmd/trust-engine
+
+      - name: Start trust-engine
+        working-directory: .
+        run: |
+          /tmp/verilink-trust-engine -grpc-port 9091 -http-port 8086 &
+          echo $! > /tmp/verilink-trust-engine.pid
+          for i in $(seq 1 30); do
+            if curl -sf http://127.0.0.1:8086/healthz >/dev/null; then
+              echo "trust-engine healthy"
+              exit 0
+            fi
+            sleep 0.5
+          done
+          echo "trust-engine failed to become healthy" >&2
+          exit 1
+
       - name: Install dependencies
         run: npm ci
 
       - name: Integration tests
         env:
           DATABASE_URL: postgresql://verilink:verilink@127.0.0.1:5432/verilink_test
           API_KEY_HMAC_SECRET: test-hmac-secret-for-integration
+          TRUST_ENGINE_ADDR: 127.0.0.1:9091
+          CI: "true"
         run: npm run test:integration
+
+      - name: Stop trust-engine
+        if: always()
+        working-directory: .
+        run: |
+          if [ -f /tmp/verilink-trust-engine.pid ]; then
+            kill "$(cat /tmp/verilink-trust-engine.pid)" || true
+          fi

control-plane/src/__tests__/integration/score-recompute.test.ts

@@ -0,0 +1,244 @@
+/**
+ * Plan 6 PR B — live RunVeriRank score recompute (mandatory in CI).
+ *
+ * Requires TRUST_ENGINE_ADDR. Locally you may leave it unset to skip;
+ * CI / GITHUB_ACTIONS must set it (workflow starts trust-engine).
+ */
+process.env.DATABASE_URL ||=
+  'postgresql://verilink:verilink@127.0.0.1:15432/verilink_test';
+process.env.API_KEY_HMAC_SECRET ||= 'test-hmac-secret-for-integration';
+
+import { describe, it, before, after, beforeEach } from 'node:test';
+import assert from 'node:assert/strict';
+import type pg from 'pg';
+import { setupTestDb, teardownTestDb, resetTestData } from '../../testutil/testDb.js';
+import {
+  seedTenant,
+  seedIssuer,
+  seedSubject,
+  seedBootstrapIssuer,
+  seedApiKey,
+  authHeaders,
+  signAttestationToken,
+} from '../../testutil/seedData.js';
+import { startControlPlane, type ControlPlaneHarness } from '../../testutil/appHarness.js';
+
+// Score modules import config/pool at load time — load them after env defaults above
+// via dynamic import in before() (static ESM imports would evaluate first).
+type RecomputeNow = typeof import('../../domains/graph/scoreComputationService.js').recomputeNow;
+type SchedulerCtor = typeof import('../../domains/graph/recomputeScheduler.js').RecomputeScheduler;
+type SetScheduler = typeof import('../../domains/graph/recomputeScheduler.js').setRecomputeSchedulerForTests;
+
+let recomputeNow: RecomputeNow;
+let RecomputeScheduler: SchedulerCtor;
+let setRecomputeSchedulerForTests: SetScheduler;
+
+const trustEngineAddr = process.env.TRUST_ENGINE_ADDR;
+const inCi = process.env.CI === 'true' || process.env.GITHUB_ACTIONS === 'true';
+const skipLive = !trustEngineAddr && !inCi;
+
+if (!trustEngineAddr && inCi) {
+  throw new Error(
+    'TRUST_ENGINE_ADDR is required for score-recompute integration in CI (start cmd/trust-engine)'
+  );
+}
+
+function delay(ms: number): Promise<void> {
+  return new Promise((r) => setTimeout(r, ms));
+}
+
+async function waitUntil(
+  pred: () => boolean,
+  opts: { timeoutMs?: number; intervalMs?: number } = {}
+): Promise<void> {
+  const timeoutMs = opts.timeoutMs ?? 5000;
+  const intervalMs = opts.intervalMs ?? 20;
+  const deadline = Date.now() + timeoutMs;
+  while (!pred() && Date.now() < deadline) {
+    await delay(intervalMs);
+  }
+}
+
+describe('Score recompute (live RunVeriRank)', { skip: skipLive }, () => {
+  let pool: pg.Pool;
+  let harness: ControlPlaneHarness;
+  let tenantId: string;
+  let apiKey: string;
+
+  before(async () => {
+    const scoreMod = await import('../../domains/graph/scoreComputationService.js');
+    const schedMod = await import('../../domains/graph/recomputeScheduler.js');
+    recomputeNow = scoreMod.recomputeNow;
+    RecomputeScheduler = schedMod.RecomputeScheduler;
+    setRecomputeSchedulerForTests = schedMod.setRecomputeSchedulerForTests;
+
+    pool = await setupTestDb();
+    harness = await startControlPlane();
+  });
+
+  after(async () => {
+    setRecomputeSchedulerForTests(null);
+    await harness.stop();
+    await teardownTestDb(pool);
+  });
+
+  beforeEach(async () => {
+    setRecomputeSchedulerForTests(null);
+    await resetTestData(pool);
+    const tenant = await seedTenant(pool, `tenant-score-${Date.now()}`);
+    tenantId = tenant.id;
+    apiKey = await seedApiKey(pool, tenantId, ['attest:write', 'attest:read']);
+  });
+
+  async function seedGraph() {
+    const issuer = await seedIssuer(pool, tenantId);
+    await seedBootstrapIssuer(pool, issuer.id);
+    const subject = await seedSubject(pool, tenantId);
+    return { issuer, subject };
+  }
+
+  async function submitAttestation(opts: {
+    issuer: Awaited<ReturnType<typeof seedIssuer>>;
+    subjectId: string;
+  }): Promise<{ id: string }> {
+    const token = await signAttestationToken({
+      issuerId: opts.issuer.id,
+      subjectId: opts.subjectId,
+      privateKey: opts.issuer.privateKey,
+      keyId: opts.issuer.keyId,
+    });
+    const resp = await fetch(`${harness.url}/v1/attestations/submit`, {
+      method: 'POST',
+      headers: {
+        'Content-Type': 'application/json',
+        ...authHeaders(apiKey),
+      },
+      body: JSON.stringify({ token }),
+    });
+    const bodyText = await resp.text();
+    assert.equal(resp.status, 201, bodyText);
+    const body = JSON.parse(bodyText) as { ok: boolean; data: { id: string } };
+    return body.data;
+  }
+
+  it('seeds bootstrap + attestation → scores + score.upsert', async () => {
+    const { issuer, subject } = await seedGraph();
+    await submitAttestation({ issuer, subjectId: subject.id });
+
+    const result = await recomputeNow(new Date(), { trustEngineAddr });
+    assert.equal(result.status, 'applied');
+    assert.ok(result.upserts >= 1);
+
+    const { rows: scores } = await pool.query(
+      'SELECT principal_id, score FROM network_scores ORDER BY principal_id'
+    );
+    assert.ok(scores.length >= 1);
+
+    const { rows: events } = await pool.query(
+      `SELECT event_type FROM sync_events WHERE event_type = 'score.upsert'`
+    );
+    assert.ok(events.length >= 1);
+  });
+
+  it('excludes expired attestations from scoring', async () => {
+    const { issuer, subject } = await seedGraph();
+    const submitted = await submitAttestation({ issuer, subjectId: subject.id });
+    await pool.query(
+      `UPDATE attestations SET expires_at = NOW() - INTERVAL '1 hour' WHERE id = $1`,
+      [submitted.id]
+    );
+
+    await recomputeNow(new Date(), { trustEngineAddr });
+    const { rows } = await pool.query(
+      'SELECT principal_id FROM network_scores WHERE principal_id = $1',
+      [subject.id]
+    );
+    assert.equal(rows.length, 0, 'expired attestation must not score the subject');
+  });
+
+  it('deactivated principal does not regain a score', async () => {
+    const { issuer, subject } = await seedGraph();
+    await submitAttestation({ issuer, subjectId: subject.id });
+    let result = await recomputeNow(new Date(), { trustEngineAddr });
+    assert.equal(result.status, 'applied');
+
+    await pool.query(`UPDATE principals SET status = 'deactivated' WHERE id = $1`, [
+      subject.id,
+    ]);
+    result = await recomputeNow(new Date(), { trustEngineAddr });
+    assert.equal(result.status, 'applied');
+
+    const { rows } = await pool.query(
+      'SELECT principal_id FROM network_scores WHERE principal_id = $1',
+      [subject.id]
+    );
+    assert.equal(rows.length, 0);
+  });
+
+  it('removing last bootstrap root clears existing scores + score.delete', async () => {
+    const { issuer, subject } = await seedGraph();
+    await submitAttestation({ issuer, subjectId: subject.id });
+    const first = await recomputeNow(new Date(), { trustEngineAddr });
+    assert.equal(first.status, 'applied');
+    assert.ok(first.upserts >= 1);
+
+    await pool.query('DELETE FROM bootstrap_issuers');
+    const cleared = await recomputeNow(new Date(), { trustEngineAddr });
+    assert.equal(cleared.status, 'cleared_stale');
+    assert.ok(cleared.deletes >= 1);
+
+    const { rows: scores } = await pool.query('SELECT principal_id FROM network_scores');
+    assert.equal(scores.length, 0);
+
+    const { rows: deletes } = await pool.query(
+      `SELECT event_type FROM sync_events WHERE event_type = 'score.delete'`
+    );
+    assert.ok(deletes.length >= 1);
+  });
+
+  it('second submit converges via dirty latch (single-flight)', async () => {
+    const { issuer, subject } = await seedGraph();
+    const subject2 = await seedSubject(pool, tenantId);
+    const calls: number[] = [];
+
+    let resolveGate!: () => void;
+    const gate = new Promise<void>((r) => {
+      resolveGate = r;
+    });
+    let passedGate = false;
+
+    const scheduler = new RecomputeScheduler({
+      debounceMs: 15,
+      intervalMs: 60_000,
+      recompute: async (t) => {
+        if (!passedGate) {
+          await gate;
+          passedGate = true;
+        }
+        calls.push(Date.now());
+        return recomputeNow(t, { trustEngineAddr });
+      },
+    });
+    scheduler.start();
+    setRecomputeSchedulerForTests(scheduler);
+
+    const runP = scheduler.kick();
+    await delay(10);
+    await submitAttestation({ issuer, subjectId: subject.id });
+    await submitAttestation({ issuer, subjectId: subject2.id });
+    scheduler.markDirty();
+    resolveGate();
+    await runP;
+    await waitUntil(() => calls.length >= 2);
+    await scheduler.stop();
+    setRecomputeSchedulerForTests(null);
+
+    assert.ok(
+      calls.length >= 2 && calls.length <= 3,
+      `expected 2–3 coalesced runs, got ${calls.length}`
+    );
+
+    const { rows } = await pool.query('SELECT COUNT(*)::int AS n FROM network_scores');
+    assert.ok((rows[0].n as number) >= 1);
+  });
+});

control-plane/src/db/client.ts

@@ -3,13 +3,27 @@ import pg from 'pg';
 import { config } from '../config.js';
 import { logger } from '../shared/logger.js';
 
-export const pool = new pg.Pool({
-  connectionString: config.database.url,
-  max: config.database.poolMax,
-  idleTimeoutMillis: config.database.poolIdleTimeoutMillis,
-  connectionTimeoutMillis: config.database.poolConnectionTimeoutMillis,
-});
+function createPool(): pg.Pool {
+  const p = new pg.Pool({
+    connectionString: config.database.url,
+    max: config.database.poolMax,
+    idleTimeoutMillis: config.database.poolIdleTimeoutMillis,
+    connectionTimeoutMillis: config.database.poolConnectionTimeoutMillis,
+  });
+  p.on('error', (err) => {
+    logger.error({ err }, 'Unexpected idle client error');
+  });
+  return p;
+}
 
-pool.on('error', (err) => {
-  logger.error({ err }, 'Unexpected idle client error');
-});
+export let pool = createPool();
+
+/** Recreate the singleton after `pool.end()` (multi-suite integration harness). */
+export function recreatePool(): void {
+  if (!pool.ended) {
+    throw new Error(
+      'recreatePool() called while the current pool is still open; call pool.end() first'
+    );
+  }
+  pool = createPool();
+}

control-plane/src/db/transaction.ts

@@ -7,6 +7,7 @@ export { pool };
 /**
  * Execute work inside a transaction. The client is automatically released
  * (even if work throws). Commit happens only if work returns without error.
+ * Uses the live `pool` export so recreatePool() is visible after suite teardown.
  */
 export async function withTransaction<T>(
   work: (client: pg.PoolClient) => Promise<T>

control-plane/src/domains/attestation/attestationService.ts

@@ -9,6 +9,8 @@ import { isV0AllowedForIssuer } from './legacyConfig.js';
 import { verifyAttestation, type KeyCandidate } from '../../grpc/trustEngineClient.js';
 import { AppError, CODES } from '../../shared/errors/AppError.js';
 import { withTransaction } from '../../db/transaction.js';
+import { getRecomputeScheduler } from '../graph/recomputeScheduler.js';
+import { logger } from '../../shared/logger.js';
 
 function sha256hex(input: string): string {
   return createHash('sha256').update(input).digest('hex');
@@ -183,7 +185,7 @@ export async function submitAttestation(opts: {
 
   // 10. Store attestation transactionally with concurrency-safe dedup + observation_id pairing
   try {
-    return await withTransaction(async (client) => {
+    const row = await withTransaction(async (client) => {
       // Dedup check inside transaction
       const existing = await attestationRepo.findByTokenDigest(tokenDigest, client);
       if (existing) {
@@ -230,6 +232,15 @@ export async function submitAttestation(opts: {
         verifiedKeyId: verifyResult.verifiedKeyId!,
       }, client);
     });
+    // Plan 6: enqueue score recompute after durable ingest (outside TX).
+    // Isolate from write-path error mapping so a scheduler throw cannot
+    // turn a committed submit into CONFLICT / failed response.
+    try {
+      getRecomputeScheduler().markDirty();
+    } catch (err) {
+      logger.error({ err }, 'failed to enqueue score recompute after ingest');
+    }
+    return row;
   } catch (err: any) {
     if (err instanceof AppError) throw err;
     if (err.code === '23505') {