feat(ql3): add opaque cluster environment bundle delivery

2026-09-22 19:29:13 +08:00 · 2026-08-24 19:31:03 +08:00
parent cf2c0ec7b3
commit 4abf125ce9
36 changed files with 1682 additions and 382 deletions
@@ -77,9 +77,7 @@ function unavailable(): TaskDefinitionUnavailableError {
  return new TaskDefinitionUnavailableError();
 }

-function taskDefinitionRecord(
-  row: TaskDefinitionRow,
-): TaskDefinitionRecord {
+function taskDefinitionRecord(row: TaskDefinitionRow): TaskDefinitionRecord {
  try {
    const description = row.description;
    if (description !== null && typeof description !== 'string') {
@@ -128,6 +126,9 @@ function executionPlanJson(
  return Object.freeze({
    command: revision.command,
    environment: revision.environment,
+    ...(revision.environmentBundleRef === undefined
+      ? {}
+      : { environmentBundleRef: revision.environmentBundleRef }),
    ...(revision.workingDirectory === undefined
      ? {}
      : { workingDirectory: revision.workingDirectory }),
@@ -140,7 +141,9 @@ function executionPlanJson(
  });
 }

-function executionRevision(row: TaskDefinitionRow): ClusterTaskExecutionRevision {
+function executionRevision(
+  row: TaskDefinitionRow,
+): ClusterTaskExecutionRevision {
  try {
    const plan = postgresRequiredJsonObject(row.planJson, unavailable);
    const keys = Object.keys(plan);
@@ -149,9 +152,14 @@ function executionRevision(row: TaskDefinitionRow): ClusterTaskExecutionRevision
      !keys.includes('environment') ||
      keys.some(
        (key) =>
-          !['command', 'environment', 'placement', 'timeoutMs', 'workingDirectory'].includes(
-            key,
-          ),
+          ![
+            'command',
+            'environment',
+            'environmentBundleRef',
+            'placement',
+            'timeoutMs',
+            'workingDirectory',
+          ].includes(key),
      )
    ) {
      throw unavailable();
@@ -176,6 +184,9 @@ function executionRevision(row: TaskDefinitionRow): ClusterTaskExecutionRevision
      command: plan.command as ClusterTaskExecutionRevision['command'],
      environment:
        plan.environment as ClusterTaskExecutionRevision['environment'],
+      ...(plan.environmentBundleRef === undefined
+        ? {}
+        : { environmentBundleRef: plan.environmentBundleRef as string }),
      ...(plan.workingDirectory === undefined
        ? {}
        : { workingDirectory: plan.workingDirectory as string }),
@@ -185,10 +196,9 @@ function executionRevision(row: TaskDefinitionRow): ClusterTaskExecutionRevision
      ...(plan.placement === undefined
        ? {}
        : {
-            placement:
-              plan.placement as unknown as NonNullable<
-                ClusterTaskExecutionRevision['placement']
-              >,
+            placement: plan.placement as unknown as NonNullable<
+              ClusterTaskExecutionRevision['placement']
+            >,
          }),
      contentDigest: postgresRequiredString(row.contentDigest, unavailable),
      createdAtMs: postgresRequiredInteger(row.createdAtMs, unavailable),
@@ -319,7 +329,6 @@ export class PostgresTaskDefinitionSource implements TaskDefinitionSource {
    }
  }

-
  async findCurrentTaskDefinition(
    projectId: string,
    taskId: string,
@@ -422,7 +431,9 @@ export class PostgresTaskExecutionRevisionSource
    readonly sourceRevision: number;
  }): Promise<ClusterTaskExecutionRevision | null> {
    if (!identity || typeof identity !== 'object' || Array.isArray(identity)) {
-      throw new TypeError('Cluster Task execution revision identity is invalid');
+      throw new TypeError(
+        'Cluster Task execution revision identity is invalid',
+      );
    }
    assertTaskDefinitionIdentifier(identity.projectId, 'projectId');
    assertTaskDefinitionIdentifier(identity.taskId, 'taskId');
@@ -447,8 +458,7 @@ export class PostgresTaskDefinitionRepository
 {
  constructor(
    pool: PostgresPool,
-    private readonly semanticRegistry: TaskSpecSemanticRegistry =
-      createBuiltInTaskSpecSemanticRegistry(),
+    private readonly semanticRegistry: TaskSpecSemanticRegistry = createBuiltInTaskSpecSemanticRegistry(),
  ) {
    super(pool);
  }
@@ -483,10 +493,7 @@ export class PostgresTaskDefinitionRepository
      command.kind === 'command' &&
      command.spec.schema === BUILT_IN_COMMAND_TASK_SPEC_SCHEMA
        ? compileClusterCommandTaskDefinition(
-            createTaskDefinitionRecord(
-              command,
-              command.occurredAtMs,
-            ),
+            createTaskDefinitionRecord(command, command.occurredAtMs),
            this.semanticRegistry,
          )
        : null;
@@ -560,10 +567,7 @@ export class PostgresTaskDefinitionRepository
           WHERE id = $1`,
          [command.projectId],
        );
-        if (
-          project.rows.length !== 1 ||
-          project.rows[0]?.status !== 'active'
-        ) {
+        if (project.rows.length !== 1 || project.rows[0]?.status !== 'active') {
          throw new TaskDefinitionConflictError();
        }

@@ -600,8 +604,9 @@ export class PostgresTaskDefinitionRepository
          unavailable,
        );
        if (
-          (created ? command.expectedRevision !== null :
-            currentRevision !== command.expectedRevision) ||
+          (created
+            ? command.expectedRevision !== null
+            : currentRevision !== command.expectedRevision) ||
          command.occurredAtMs < previousUpdatedAtMs
        ) {
          throw new TaskDefinitionConflictError();
@@ -27,7 +27,8 @@ function text(row: Row, key: string): string {

 function integer(row: Row, key: string): number {
  const raw = row[key];
-  const value = typeof raw === 'string' && /^\d+$/.test(raw) ? Number(raw) : raw;
+  const value =
+    typeof raw === 'string' && /^\d+$/.test(raw) ? Number(raw) : raw;
  if (typeof value !== 'number' || !Number.isSafeInteger(value)) {
    throw new RemoteWorkerSecretDeliveryUnavailableError();
  }
@@ -44,10 +45,19 @@ function executionRevision(row: Row): ClusterTaskExecutionRevision {
  if (
    !keys.includes('command') ||
    !keys.includes('environment') ||
-    keys.some((key) =>
-      !['command', 'environment', 'placement', 'timeoutMs', 'workingDirectory']
-        .includes(key))
-  ) throw new RemoteWorkerSecretDeliveryUnavailableError();
+    keys.some(
+      (key) =>
+        ![
+          'command',
+          'environment',
+          'environmentBundleRef',
+          'placement',
+          'timeoutMs',
+          'workingDirectory',
+        ].includes(key),
+    )
+  )
+    throw new RemoteWorkerSecretDeliveryUnavailableError();
  return normalizeClusterTaskExecutionRevision({
    projectId: text(row, 'revisionProjectId'),
    taskId: text(row, 'revisionTaskId'),
@@ -57,7 +67,11 @@ function executionRevision(row: Row): ClusterTaskExecutionRevision {
    executorType: text(row, 'revisionExecutorType') as 'remote_worker',
    planSchema: text(row, 'planSchema') as 'qinglong/command-execution@v1',
    command: value.command as ClusterTaskExecutionRevision['command'],
-    environment: value.environment as ClusterTaskExecutionRevision['environment'],
+    environment:
+      value.environment as ClusterTaskExecutionRevision['environment'],
+    ...(value.environmentBundleRef === undefined
+      ? {}
+      : { environmentBundleRef: value.environmentBundleRef as string }),
    ...(value.workingDirectory === undefined
      ? {}
      : { workingDirectory: value.workingDirectory as string }),
@@ -66,7 +80,11 @@ function executionRevision(row: Row): ClusterTaskExecutionRevision {
      : { timeoutMs: value.timeoutMs as number }),
    ...(value.placement === undefined
      ? {}
-      : { placement: value.placement as NonNullable<ClusterTaskExecutionRevision['placement']> }),
+      : {
+          placement: value.placement as NonNullable<
+            ClusterTaskExecutionRevision['placement']
+          >,
+        }),
    contentDigest: text(row, 'revisionContentDigest'),
    createdAtMs: integer(row, 'revisionCreatedAtMs'),
  });
@@ -80,7 +98,8 @@ async function begin(client: PostgresClient): Promise<void> {
 }

 export class PostgresRemoteWorkerSecretDeliveryAuthorityRepository
-  implements RemoteWorkerSecretDeliveryAuthorityRepository {
+  implements RemoteWorkerSecretDeliveryAuthorityRepository
+{
  constructor(private readonly pool: PostgresPool) {
    if (!pool || typeof pool.connect !== 'function') {
      throw new TypeError('PostgreSQL remote Secret delivery pool is invalid');
@@ -156,7 +175,9 @@ export class PostgresRemoteWorkerSecretDeliveryAuthorityRepository
      }
      const row = result.rows[0];
      if (!row) {
-        throw new RemoteWorkerSecretDeliveryFenceRejectedError('authority_mismatch');
+        throw new RemoteWorkerSecretDeliveryFenceRejectedError(
+          'authority_mismatch',
+        );
      }
      const observedAtMs = integer(row, 'observedAtMs');
      const tokenDigest = digestRunDispatchLeaseToken(command.leaseToken);
@@ -192,25 +213,38 @@ export class PostgresRemoteWorkerSecretDeliveryAuthorityRepository
        row.leaseOfferId === command.offerId &&
        integer(row, 'leaseExpiresAtMs') > observedAtMs;
      if (!matches) {
-        throw new RemoteWorkerSecretDeliveryFenceRejectedError('authority_mismatch');
+        throw new RemoteWorkerSecretDeliveryFenceRejectedError(
+          'authority_mismatch',
+        );
      }
      let revision: ClusterTaskExecutionRevision;
      try {
        revision = executionRevision(row);
      } catch (error) {
-        if (error instanceof RemoteWorkerSecretDeliveryUnavailableError) throw error;
+        if (error instanceof RemoteWorkerSecretDeliveryUnavailableError)
+          throw error;
        throw new RemoteWorkerSecretDeliveryUnavailableError();
      }
      const expectedRefs = Object.freeze([
-        ...new Set(revision.environment.flatMap((binding) =>
-          binding.kind === 'secret' ? [binding.secretRef] : [])),
+        ...new Set(
+          revision.environment.flatMap((binding) =>
+            binding.kind === 'secret' ? [binding.secretRef] : [],
+          ),
+        ),
      ]);
+      const expectedEnvironmentBundleRefs = Object.freeze(
+        revision.environmentBundleRef === undefined
+          ? []
+          : [revision.environmentBundleRef],
+      );
      if (
        revision.projectId !== command.projectId ||
        revision.taskId !== command.taskId ||
        revision.taskRevision !== command.taskRevision ||
        revision.contentDigest !== command.executionDigest ||
-        JSON.stringify(expectedRefs) !== JSON.stringify(command.secretRefs)
+        JSON.stringify(expectedRefs) !== JSON.stringify(command.secretRefs) ||
+        JSON.stringify(expectedEnvironmentBundleRefs) !==
+          JSON.stringify(command.environmentBundleRefs)
      ) {
        throw new RemoteWorkerSecretDeliveryFenceRejectedError(
          'secret_scope_mismatch',
@@ -230,15 +264,21 @@ export class PostgresRemoteWorkerSecretDeliveryAuthorityRepository
        leaseGeneration: command.leaseGeneration,
        leaseVersion: command.expectedLeaseVersion,
        secretRefs: expectedRefs,
+        environmentBundleRefs: expectedEnvironmentBundleRefs,
      });
      await client.query('COMMIT');
      return authority;
    } catch (error) {
-      try { await client.query('ROLLBACK'); } catch { /* preserve root */ }
+      try {
+        await client.query('ROLLBACK');
+      } catch {
+        /* preserve root */
+      }
      if (
        error instanceof RemoteWorkerSecretDeliveryFenceRejectedError ||
        error instanceof RemoteWorkerSecretDeliveryUnavailableError
-      ) throw error;
+      )
+        throw error;
      throw new RemoteWorkerSecretDeliveryUnavailableError();
    } finally {
      client.release();
@@ -17,6 +17,7 @@ const {
 const {
  compileClusterCommandTaskDefinition,
 } = require('@qinglong/runtime-core/cluster-execution-revision');
+const { createSecretRef } = require('@qinglong/runtime-core/secret-reference');
 const {
  TaskDefinitionAdministrationAuthorizationFenceConflictError,
  TaskDefinitionAdministrationMutationConflictError,
@@ -447,6 +448,34 @@ test('publishes TaskDefinition atomically and replays the exact mutation', async
  );
 });

+test('persists only the pinned environment bundle reference in an execution plan', async () => {
+  const environmentBundleRef = createSecretRef({
+    projectId: TASK_COMMAND.projectId,
+    name: 'legacy-env-bundle',
+    version: 7,
+  });
+  const command = {
+    ...TASK_COMMAND,
+    taskId: 'task-bundle-00001',
+    mutationId: '123e4567-e89b-42d3-a456-426614174021',
+    spec: {
+      ...TASK_COMMAND.spec,
+      config: { ...TASK_COMMAND.spec.config, environmentBundleRef },
+    },
+  };
+  const fixture = appendPool('task');
+  await new PostgresTaskDefinitionRepository(
+    fixture.pool,
+  ).appendTaskDefinitionRevision(command);
+  const insert = fixture.queries.find(({ text }) =>
+    text.includes('INSERT INTO "ql3"."task_execution_revisions"'),
+  );
+  const plan = JSON.parse(insert.values[7]);
+  assert.equal(plan.environmentBundleRef, environmentBundleRef);
+  assert.deepEqual(plan.environment, []);
+  assert.equal(JSON.stringify(plan).includes('LEGACY_ENV_NAME'), false);
+});
+
 test('runs TaskDefinition transaction hooks for create and replay before COMMIT', async () => {
  const createdFixture = appendPool('task');
  const createdHook = [];
@@ -18,12 +18,21 @@ const SOURCE_DIGEST = 'a'.repeat(64);
 const TASK_REVISION = `qltd:v1:1:${SOURCE_DIGEST}`;
 const LEASE_TOKEN = 'worker_generated_lease_capability_0000000000000001';
 const SECRET_REF = createSecretRef({ projectId: 'project-1', name: 'token' });
+const ENVIRONMENT_BUNDLE_REF = createSecretRef({
+  projectId: 'project-1',
+  name: 'legacy-env-bundle',
+  version: 4,
+});

 function revision() {
  return createClusterTaskExecutionRevision({
-    projectId: 'project-1', taskId: 'task-1', taskRevision: TASK_REVISION,
-    sourceRevision: 1, sourceContentDigest: SOURCE_DIGEST,
-    executorType: 'remote_worker', planSchema: 'qinglong/command-execution@v1',
+    projectId: 'project-1',
+    taskId: 'task-1',
+    taskRevision: TASK_REVISION,
+    sourceRevision: 1,
+    sourceContentDigest: SOURCE_DIGEST,
+    executorType: 'remote_worker',
+    planSchema: 'qinglong/command-execution@v1',
    command: { kind: 'argv', file: '/bin/true', args: [] },
    environment: [{ name: 'TOKEN', kind: 'secret', secretRef: SECRET_REF }],
    createdAtMs: 1,
@@ -32,40 +41,74 @@ function revision() {

 function command(executionDigest, overrides = {}) {
  return {
-    workerId: 'edge-1', workerSessionId: SESSION_ID, workerGeneration: 2,
-    runId: 'run-1', attemptId: 'attempt-1', projectId: 'project-1',
-    taskId: 'task-1', taskRevision: TASK_REVISION, executionDigest,
-    offerId: 'offer-1', leaseGeneration: 3, leaseToken: LEASE_TOKEN,
-    expectedLeaseVersion: 4, secretRefs: [SECRET_REF], ...overrides,
+    workerId: 'edge-1',
+    workerSessionId: SESSION_ID,
+    workerGeneration: 2,
+    runId: 'run-1',
+    attemptId: 'attempt-1',
+    projectId: 'project-1',
+    taskId: 'task-1',
+    taskRevision: TASK_REVISION,
+    executionDigest,
+    offerId: 'offer-1',
+    leaseGeneration: 3,
+    leaseToken: LEASE_TOKEN,
+    expectedLeaseVersion: 4,
+    secretRefs: [SECRET_REF],
+    environmentBundleRefs: [],
+    ...overrides,
  };
 }

 function authorityRow(plan, overrides = {}) {
  return {
-    observedAtMs: '1000', runId: 'run-1', runProjectId: 'project-1',
-    runTaskId: 'task-1', runTaskRevision: TASK_REVISION,
-    runStatus: 'dispatching', executionOwner: 'runtime',
-    cancelRequestedAtMs: null, attemptStatus: 'starting',
-    attemptExecutorType: 'remote_worker', attemptWorkerId: 'edge-1',
-    attemptWorkerSessionId: SESSION_ID, attemptWorkerGeneration: 2,
+    observedAtMs: '1000',
+    runId: 'run-1',
+    runProjectId: 'project-1',
+    runTaskId: 'task-1',
+    runTaskRevision: TASK_REVISION,
+    runStatus: 'dispatching',
+    executionOwner: 'runtime',
+    cancelRequestedAtMs: null,
+    attemptStatus: 'starting',
+    attemptExecutorType: 'remote_worker',
+    attemptWorkerId: 'edge-1',
+    attemptWorkerSessionId: SESSION_ID,
+    attemptWorkerGeneration: 2,
    attemptLeaseTokenDigest: digestRunDispatchLeaseToken(LEASE_TOKEN),
-    attemptLeaseGeneration: 3, attemptLeaseVersion: 4,
-    attemptOfferId: 'offer-1', sessionId: SESSION_ID, sessionGeneration: 2,
-    sessionStatus: 'online', sessionExpiresAtMs: '5000',
-    leaseRunId: 'run-1', leaseStatus: 'leased', leaseVersion: 4,
-    leaseGeneration: 3, leaseWorkerId: 'edge-1',
-    leaseWorkerSessionId: SESSION_ID, leaseWorkerGeneration: 2,
+    attemptLeaseGeneration: 3,
+    attemptLeaseVersion: 4,
+    attemptOfferId: 'offer-1',
+    sessionId: SESSION_ID,
+    sessionGeneration: 2,
+    sessionStatus: 'online',
+    sessionExpiresAtMs: '5000',
+    leaseRunId: 'run-1',
+    leaseStatus: 'leased',
+    leaseVersion: 4,
+    leaseGeneration: 3,
+    leaseWorkerId: 'edge-1',
+    leaseWorkerSessionId: SESSION_ID,
+    leaseWorkerGeneration: 2,
    leaseTokenDigest: digestRunDispatchLeaseToken(LEASE_TOKEN),
-    leaseOfferId: 'offer-1', leaseExpiresAtMs: '5000',
-    revisionProjectId: plan.projectId, revisionTaskId: plan.taskId,
+    leaseOfferId: 'offer-1',
+    leaseExpiresAtMs: '5000',
+    revisionProjectId: plan.projectId,
+    revisionTaskId: plan.taskId,
    sourceRevision: plan.sourceRevision,
    revisionTaskRevision: plan.taskRevision,
    sourceContentDigest: plan.sourceContentDigest,
-    revisionExecutorType: plan.executorType, planSchema: plan.planSchema,
+    revisionExecutorType: plan.executorType,
+    planSchema: plan.planSchema,
    planJson: {
      command: plan.command,
      environment: plan.environment,
-      ...(plan.workingDirectory === undefined ? {} : { workingDirectory: plan.workingDirectory }),
+      ...(plan.environmentBundleRef === undefined
+        ? {}
+        : { environmentBundleRef: plan.environmentBundleRef }),
+      ...(plan.workingDirectory === undefined
+        ? {}
+        : { workingDirectory: plan.workingDirectory }),
      ...(plan.timeoutMs === undefined ? {} : { timeoutMs: plan.timeoutMs }),
      ...(plan.placement === undefined ? {} : { placement: plan.placement }),
    },
@@ -84,11 +127,15 @@ function fixture(row) {
      if (sql.includes('FROM observation')) return { rows: row ? [row] : [] };
      return { rows: [] };
    },
-    release() { released += 1; },
+    release() {
+      released += 1;
+    },
  };
  return {
    repository: new PostgresRemoteWorkerSecretDeliveryAuthorityRepository({
-      async connect() { return client; },
+      async connect() {
+        return client;
+      },
    }),
    queries,
    released: () => released,
@@ -100,9 +147,13 @@ test('authorizes exact Session, Lease and immutable execution revision fences',
  const { repository, queries, released } = fixture(authorityRow(plan));
  const result = await repository.authorize(command(plan.contentDigest));
  assert.deepEqual(result.secretRefs, [SECRET_REF]);
+  assert.deepEqual(result.environmentBundleRefs, []);
  assert.equal(result.executionDigest, plan.contentDigest);
  assert.equal('leaseToken' in result, false);
-  assert.equal(queries.some(({ sql }) => sql.includes('pg_advisory_xact_lock')), true);
+  assert.equal(
+    queries.some(({ sql }) => sql.includes('pg_advisory_xact_lock')),
+    true,
+  );
  assert.equal(queries.at(-1).sql, 'COMMIT');
  assert.equal(released(), 1);
 });
@@ -129,9 +180,11 @@ test('rejects partial Secret scope and execution digest drift', async () => {
  const extra = createSecretRef({ projectId: 'project-1', name: 'other' });
  const partial = fixture(authorityRow(plan));
  await assert.rejects(
-    partial.repository.authorize(command(plan.contentDigest, {
-      secretRefs: [extra],
-    })),
+    partial.repository.authorize(
+      command(plan.contentDigest, {
+        secretRefs: [extra],
+      }),
+    ),
    /secret_scope_mismatch/,
  );
  const digestDrift = fixture(authorityRow(plan));
@@ -140,3 +193,29 @@ test('rejects partial Secret scope and execution digest drift', async () => {
    /secret_scope_mismatch/,
  );
 });
+
+test('authorizes a bundle-only execution without exposing environment names', async () => {
+  const plan = createClusterTaskExecutionRevision({
+    projectId: 'project-1',
+    taskId: 'task-1',
+    taskRevision: TASK_REVISION,
+    sourceRevision: 1,
+    sourceContentDigest: SOURCE_DIGEST,
+    executorType: 'remote_worker',
+    planSchema: 'qinglong/command-execution@v1',
+    command: { kind: 'argv', file: '/bin/true', args: [] },
+    environment: [],
+    environmentBundleRef: ENVIRONMENT_BUNDLE_REF,
+    createdAtMs: 1,
+  });
+  const { repository } = fixture(authorityRow(plan));
+  const result = await repository.authorize(
+    command(plan.contentDigest, {
+      secretRefs: [],
+      environmentBundleRefs: [ENVIRONMENT_BUNDLE_REF],
+    }),
+  );
+  assert.deepEqual(result.secretRefs, []);
+  assert.deepEqual(result.environmentBundleRefs, [ENVIRONMENT_BUNDLE_REF]);
+  assert.equal(JSON.stringify(result).includes('LEGACY_VALUE'), false);
+});