25707305714

Committed 12 May 2026 01:09AM UTC coverage: 91.725% (-0.01%) from 91.736%

Build # 25707305714

Build Type

push

github

Committed by

web-flow

Commit Message

Delete old versions of all lambda bots (#9175)

* Delete all old lambda versions

Signed-off-by: Matt Long <ml@medplum.com>

* Add the new worker files

Signed-off-by: Matt Long <ml@medplum.com>

* Skip preemptive alias scanning

Since Medplum does not use aliases, don't bother scanning
for them. Aliased versions are protected from deletes, so
optimisitcally try to delete and handle errors gracefully.

Signed-off-by: Matt Long <ml@medplum.com>

* Add lambda:ListFunctions permission

Signed-off-by: Matt Long <ml@medplum.com>

* Test coverage

Signed-off-by: Matt Long <ml@medplum.com>

---------

Signed-off-by: Matt Long <ml@medplum.com>

Coverage Stats

18971 of 21650 branches covered (87.63%)

Branch coverage included in aggregate %.

121 of 127 new or added lines in 5 files covered. (95.28%)

34337 of 36467 relevant lines covered (94.16%)

13268.49 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.16

/packages/server/src/workers/lambda-cleaner.ts

// SPDX-FileCopyrightText: Copyright Orangebot, Inc. and Medplum contributors
// SPDX-License-Identifier: Apache-2.0
import type { LambdaClient } from '@aws-sdk/client-lambda';
import { ListFunctionsCommand } from '@aws-sdk/client-lambda';
import type { WithId } from '@medplum/core';
import { EMPTY } from '@medplum/core';
import type { AsyncJob, Parameters, ParametersParameter } from '@medplum/fhirtypes';
import type { Job, QueueBaseOptions } from 'bullmq';
import { Queue, Worker } from 'bullmq';
import type { DeleteOldLambdaVersionStats } from '../cloud/aws/lambda';
import { createLambdaClient, deleteOldLambdaVersions } from '../cloud/aws/lambda';
import { AsyncJobExecutor } from '../fhir/operations/utils/asyncjobexecutor';
import { getShardSystemRepo } from '../fhir/repo';
import { PLACEHOLDER_SHARD_ID } from '../fhir/sharding';
import { globalLogger } from '../logger';
import type { WorkerInitializer, WorkerInitializerOptions } from './utils';
import { addVerboseQueueLogging, getBullmqRedisConnectionOptions, getWorkerBullmqConfig, queueRegistry } from './utils';

export interface LambdaCleanerOptions {
  readonly nameRegex: string;
  readonly keepLatest?: number;
  readonly deleteConcurrency?: number;
  readonly dryRun?: boolean;
}

interface ResolvedLambdaCleanerOptions extends LambdaCleanerOptions {
  readonly keepLatest: number;
  readonly deleteConcurrency: number;
  readonly dryRun: boolean;
}

const DEFAULT_KEEP_LATEST = 1;
const DEFAULT_DELETE_CONCURRENCY = 5;
const DEFAULT_DRY_RUN = true;

export interface LambdaCleanerJobData {
  readonly asyncJob: WithId<AsyncJob>;
  readonly options: LambdaCleanerOptions;
}

export interface LambdaCleanerSummary extends DeleteOldLambdaVersionStats {
  readonly options: LambdaCleanerOptions;
  functionsScanned: number;
  functionsMatched: number;
  durationMs: number;
}

export const LambdaCleanerQueueName = 'LambdaCleanerQueue';

export const initLambdaCleanerWorker: WorkerInitializer = (config, options?: WorkerInitializerOptions) => {
  const defaultOptions: QueueBaseOptions = {
    connection: getBullmqRedisConnectionOptions(config),
  };

  const queue = new Queue<LambdaCleanerJobData>(LambdaCleanerQueueName, {
    ...defaultOptions,
    defaultJobOptions: { attempts: 1 },
  });

  let worker: Worker<LambdaCleanerJobData> | undefined;
  if (options?.workerEnabled !== false) {
    const workerConfig = getWorkerBullmqConfig(config, 'lambda-cleaner');
    worker = new Worker<LambdaCleanerJobData>(LambdaCleanerQueueName, (job) => lambdaCleanerJobProcessor(job), {
      ...defaultOptions,
      concurrency: 1,
      ...workerConfig,
    });
    addVerboseQueueLogging<LambdaCleanerJobData>(queue, worker, (job) => ({
      asyncJob: `AsyncJob/${job.data.asyncJob.id}`,
      nameRegex: job.data.options.nameRegex,
      dryRun: job.data.options.dryRun,
    }));
  }

  return { queue, worker, name: LambdaCleanerQueueName };
};

export function getLambdaCleanerQueue(): Queue<LambdaCleanerJobData> | undefined {
  return queueRegistry.get(LambdaCleanerQueueName);
}

export async function addLambdaCleanerJobData(jobData: LambdaCleanerJobData): Promise<Job<LambdaCleanerJobData>> {
  const queue = getLambdaCleanerQueue();
  if (!queue) {
    throw new Error(`Job queue ${LambdaCleanerQueueName} not available`);
  }
  return queue.add('LambdaCleanerJob', jobData);
}

export async function lambdaCleanerJobProcessor(job: Job<LambdaCleanerJobData>): Promise<WithId<AsyncJob>> {
  const systemRepo = getShardSystemRepo(PLACEHOLDER_SHARD_ID);
  const exec = new AsyncJobExecutor(systemRepo, job.data.asyncJob);
  return exec.startAsync(async () => {
    const summary = await execLambdaCleanerJob(job.data.options);
    return formatSummary(summary);
  });
}

export async function execLambdaCleanerJob(
  inputOptions: LambdaCleanerOptions,
  client?: LambdaClient
): Promise<LambdaCleanerSummary> {
  const options: ResolvedLambdaCleanerOptions = {
    nameRegex: inputOptions.nameRegex,
    keepLatest: inputOptions.keepLatest ?? DEFAULT_KEEP_LATEST,
    deleteConcurrency: inputOptions.deleteConcurrency ?? DEFAULT_DELETE_CONCURRENCY,
    dryRun: inputOptions.dryRun ?? DEFAULT_DRY_RUN,
  };
  const lambdaClient = client ?? createLambdaClient();
  const startTime = Date.now();
  const summary: LambdaCleanerSummary = {
    options,
    functionsScanned: 0,
    functionsMatched: 0,
    functionsWithDeleteCandidates: 0,
    publishedVersionsScanned: 0,
    versionsPlanned: 0,
    versionsDeleted: 0,
    versionsNotFound: 0,
    versionsHasAlias: 0,
    durationMs: 0,
  };

  const nameRegex = new RegExp(options.nameRegex);
  let marker: string | undefined;
  do {
    const response = await lambdaClient.send(new ListFunctionsCommand({ Marker: marker }));
    marker = response.NextMarker;

    for (const lambdaFunction of response.Functions ?? EMPTY) {
      const functionName = lambdaFunction.FunctionName;
      if (functionName) {
        summary.functionsScanned++;
        if (nameRegex.test(functionName)) {
          summary.functionsMatched++;
          await deleteOldLambdaVersions(lambdaClient, functionName, options, summary);
        }
      }
    }
  } while (marker);

  summary.durationMs = Date.now() - startTime;
  globalLogger.info('Lambda cleaner completed', summary);
  return summary;
}

function formatSummary(summary: LambdaCleanerSummary): Parameters {
  const { options, ...stats } = summary;
  const parameters: ParametersParameter[] = [
    { name: 'options.nameRegex', valueString: options.nameRegex },
    { name: 'options.keepLatest', valueInteger: options.keepLatest },
    { name: 'options.deleteConcurrency', valueInteger: options.deleteConcurrency },
    { name: 'options.dryRun', valueBoolean: options.dryRun },
    { name: 'functionsScanned', valueInteger: stats.functionsScanned },
    { name: 'functionsMatched', valueInteger: stats.functionsMatched },
    { name: 'functionsWithDeleteCandidates', valueInteger: stats.functionsWithDeleteCandidates },
    { name: 'publishedVersionsScanned', valueInteger: stats.publishedVersionsScanned },
    { name: 'versionsPlanned', valueInteger: stats.versionsPlanned },
    { name: 'versionsDeleted', valueInteger: stats.versionsDeleted },
    { name: 'versionsNotFound', valueInteger: stats.versionsNotFound },
    { name: 'versionsHasAlias', valueInteger: stats.versionsHasAlias },
    { name: 'durationMs', valueQuantity: { value: stats.durationMs, code: 'ms' } },
  ];

  return {
    resourceType: 'Parameters',
    parameter: parameters,
  };
}

1	// SPDX-FileCopyrightText: Copyright Orangebot, Inc. and Medplum contributors
2	// SPDX-License-Identifier: Apache-2.0
3	import type { LambdaClient } from '@aws-sdk/client-lambda';
4	import { ListFunctionsCommand } from '@aws-sdk/client-lambda';
5	import type { WithId } from '@medplum/core';
6	import { EMPTY } from '@medplum/core';
7	import type { AsyncJob, Parameters, ParametersParameter } from '@medplum/fhirtypes';
8	import type { Job, QueueBaseOptions } from 'bullmq';
9	import { Queue, Worker } from 'bullmq';
10	import type { DeleteOldLambdaVersionStats } from '../cloud/aws/lambda';
11	import { createLambdaClient, deleteOldLambdaVersions } from '../cloud/aws/lambda';
12	import { AsyncJobExecutor } from '../fhir/operations/utils/asyncjobexecutor';
13	import { getShardSystemRepo } from '../fhir/repo';
14	import { PLACEHOLDER_SHARD_ID } from '../fhir/sharding';
15	import { globalLogger } from '../logger';
16	import type { WorkerInitializer, WorkerInitializerOptions } from './utils';
17	import { addVerboseQueueLogging, getBullmqRedisConnectionOptions, getWorkerBullmqConfig, queueRegistry } from './utils';
18
19	export interface LambdaCleanerOptions {
20	readonly nameRegex: string;
21	readonly keepLatest?: number;
22	readonly deleteConcurrency?: number;
23	readonly dryRun?: boolean;
24	}
25
26	interface ResolvedLambdaCleanerOptions extends LambdaCleanerOptions {
27	readonly keepLatest: number;
28	readonly deleteConcurrency: number;
29	readonly dryRun: boolean;
30	}
31
32	const DEFAULT_KEEP_LATEST = 1;	190✔
33	const DEFAULT_DELETE_CONCURRENCY = 5;	190✔
34	const DEFAULT_DRY_RUN = true;	190✔
35
36	export interface LambdaCleanerJobData {
37	readonly asyncJob: WithId<AsyncJob>;
38	readonly options: LambdaCleanerOptions;
39	}
40
41	export interface LambdaCleanerSummary extends DeleteOldLambdaVersionStats {
42	readonly options: LambdaCleanerOptions;
43	functionsScanned: number;
44	functionsMatched: number;
45	durationMs: number;
46	}
47
48	export const LambdaCleanerQueueName = 'LambdaCleanerQueue';	190✔
49
50	export const initLambdaCleanerWorker: WorkerInitializer = (config, options?: WorkerInitializerOptions) => {	190✔
51	const defaultOptions: QueueBaseOptions = {	248✔
52	connection: getBullmqRedisConnectionOptions(config),
53	};
54
55	const queue = new Queue<LambdaCleanerJobData>(LambdaCleanerQueueName, {	248✔
56	...defaultOptions,
57	defaultJobOptions: { attempts: 1 },
58	});
59
60	let worker: Worker<LambdaCleanerJobData> \| undefined;
61	if (options?.workerEnabled !== false) {	248✔
62	const workerConfig = getWorkerBullmqConfig(config, 'lambda-cleaner');	245✔
63	worker = new Worker<LambdaCleanerJobData>(LambdaCleanerQueueName, (job) => lambdaCleanerJobProcessor(job), {	245✔
64	...defaultOptions,
65	concurrency: 1,
66	...workerConfig,
67	});
68	addVerboseQueueLogging<LambdaCleanerJobData>(queue, worker, (job) => ({	245✔
69	asyncJob: `AsyncJob/${job.data.asyncJob.id}`,
70	nameRegex: job.data.options.nameRegex,
71	dryRun: job.data.options.dryRun,
72	}));
73	}
74
75	return { queue, worker, name: LambdaCleanerQueueName };	248✔
76	};
77
78	export function getLambdaCleanerQueue(): Queue<LambdaCleanerJobData> \| undefined {
79	return queueRegistry.get(LambdaCleanerQueueName);	3✔
80	}
81
82	export async function addLambdaCleanerJobData(jobData: LambdaCleanerJobData): Promise<Job<LambdaCleanerJobData>> {
83	const queue = getLambdaCleanerQueue();	1✔
84	if (!queue) {	1!
NEW 85	throw new Error(`Job queue ${LambdaCleanerQueueName} not available`);	×
86	}
87	return queue.add('LambdaCleanerJob', jobData);	1✔
88	}
89
90	export async function lambdaCleanerJobProcessor(job: Job<LambdaCleanerJobData>): Promise<WithId<AsyncJob>> {
91	const systemRepo = getShardSystemRepo(PLACEHOLDER_SHARD_ID);	1✔
92	const exec = new AsyncJobExecutor(systemRepo, job.data.asyncJob);	1✔
93	return exec.startAsync(async () => {	1✔
94	const summary = await execLambdaCleanerJob(job.data.options);	1✔
95	return formatSummary(summary);	1✔
96	});
97	}
98
99	export async function execLambdaCleanerJob(
100	inputOptions: LambdaCleanerOptions,
101	client?: LambdaClient
102	): Promise<LambdaCleanerSummary> {
103	const options: ResolvedLambdaCleanerOptions = {	2✔
104	nameRegex: inputOptions.nameRegex,
105	keepLatest: inputOptions.keepLatest ?? DEFAULT_KEEP_LATEST,	2!
106	deleteConcurrency: inputOptions.deleteConcurrency ?? DEFAULT_DELETE_CONCURRENCY,	3✔
107	dryRun: inputOptions.dryRun ?? DEFAULT_DRY_RUN,	2!
108	};
109	const lambdaClient = client ?? createLambdaClient();	2✔
110	const startTime = Date.now();	2✔
111	const summary: LambdaCleanerSummary = {	2✔
112	options,
113	functionsScanned: 0,
114	functionsMatched: 0,
115	functionsWithDeleteCandidates: 0,
116	publishedVersionsScanned: 0,
117	versionsPlanned: 0,
118	versionsDeleted: 0,
119	versionsNotFound: 0,
120	versionsHasAlias: 0,
121	durationMs: 0,
122	};
123
124	const nameRegex = new RegExp(options.nameRegex);	2✔
125	let marker: string \| undefined;
126	do {	2✔
127	const response = await lambdaClient.send(new ListFunctionsCommand({ Marker: marker }));	2✔
128	marker = response.NextMarker;	2✔
129
130	for (const lambdaFunction of response.Functions ?? EMPTY) {	2!
131	const functionName = lambdaFunction.FunctionName;	4✔
132	if (functionName) {	4!
133	summary.functionsScanned++;	4✔
134	if (nameRegex.test(functionName)) {	4✔
135	summary.functionsMatched++;	3✔
136	await deleteOldLambdaVersions(lambdaClient, functionName, options, summary);	3✔
137	}
138	}
139	}
140	} while (marker);
141
142	summary.durationMs = Date.now() - startTime;	2✔
143	globalLogger.info('Lambda cleaner completed', summary);	2✔
144	return summary;	2✔
145	}
146
147	function formatSummary(summary: LambdaCleanerSummary): Parameters {
148	const { options, ...stats } = summary;	1✔
149	const parameters: ParametersParameter[] = [	1✔
150	{ name: 'options.nameRegex', valueString: options.nameRegex },
151	{ name: 'options.keepLatest', valueInteger: options.keepLatest },
152	{ name: 'options.deleteConcurrency', valueInteger: options.deleteConcurrency },
153	{ name: 'options.dryRun', valueBoolean: options.dryRun },
154	{ name: 'functionsScanned', valueInteger: stats.functionsScanned },
155	{ name: 'functionsMatched', valueInteger: stats.functionsMatched },
156	{ name: 'functionsWithDeleteCandidates', valueInteger: stats.functionsWithDeleteCandidates },
157	{ name: 'publishedVersionsScanned', valueInteger: stats.publishedVersionsScanned },
158	{ name: 'versionsPlanned', valueInteger: stats.versionsPlanned },
159	{ name: 'versionsDeleted', valueInteger: stats.versionsDeleted },
160	{ name: 'versionsNotFound', valueInteger: stats.versionsNotFound },
161	{ name: 'versionsHasAlias', valueInteger: stats.versionsHasAlias },
162	{ name: 'durationMs', valueQuantity: { value: stats.durationMs, code: 'ms' } },
163	];
164
165	return {	1✔
166	resourceType: 'Parameters',
167	parameter: parameters,
168	};
169	}

medplum / medplum / 25707305714

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous