Azure Function消费计划冷启动无法触发Service Bus消息处理及解决办法

使用无服务器消费计划的Azure Function无法自动启动处理Service Bus主题中的新消息，除非手动访问Azure门户中的该函数，或者最近连接过函数日志。在最后一条消息处理完成后等待约半小时，下一条消息就不会被处理，必须手动触发才会恢复。

已尝试将函数从Node 16升级到18，Azure Function版本从3升级到4，本地运行未发现错误或崩溃情况。

运行环境 #

• @azure/functions 4.5.0
• Node 18.10.0
• @azure/service-bus 7.9.0

Host.json配置 #

{
    "version": "2.0",
    "extensionBundle": {
        "id": "Microsoft.Azure.Functions.ExtensionBundle",
        "version": "[4.*, 5.0.0)"
    }
}

核心代码 #

index.ts #

import { app, InvocationContext } from '@azure/functions';
import { handleActionEvent } from './src/actionEventHandler';

process.env.AUTH0_CLIENT_ID = process.env.AUTH_AUTH0_CLIENT_ID;
process.env.AUTH0_CLIENT_SECRET = process.env.AUTH_AUTH0_CLIENT_SECRET;

export async function serviceBusTopicTrigger(
    sbMsg: unknown,
    context: InvocationContext
): Promise<void> {
    context.log('MessageId =', context.triggerMetadata.messageId);
    try {
        await handleActionEvent(sbMsg);
    } catch (err) {
        context.log(err);
    }
}

app.serviceBusTopic('serviceBusTopicTrigger', {
    topicName: '%TOPIC_NAME%',
    subscriptionName: '%SUBSCRIPTION_NAME%',
    connection: 'ServiceBusConnString',
    handler: serviceBusTopicTrigger
});

actionEventHandler.ts（核心逻辑） #

export const handleActionEvent = async (
    eventBody: any
): Promise<void> => {
    try {
        let policyId: number | null;
        const event = eventBody.event;
        switch (event) {
            case POLICY_ACTION_EVENT.PolicyChanged:
                if (eventBody.action === POLICY_ACTION.deleted) {
                    policyId = null;
                } else {
                    policyId = eventBody.record.id;
                }
                break;
            case POLICY_ACTION_EVENT.PolicyTicketChanged:
                policyId = eventBody.record.policy_id;
                break;
            case POLICY_ACTION_EVENT.PolicyTicketRuleChanged:
                return;
            case POLICY_ACTION_EVENT.PolicyRiskCalculationTriggered:
                await calculatePolicyRisk(eventBody);
                return;
            default:
                logger.log(
                    LOG_LEVELS.DEBUG,
                    `Debug: Unhandled event ${event} was encountered`
                );
                logger.log(LOG_LEVELS.DEBUG, eventBody.record);
                return;
        }
        const orgId = eventBody.record?.organization_id ?? eventBody.orgId;

        if (eventBody?.projectIds.length) {
            for (const projId of eventBody.projectIds) {
                await scheduleRiskCalculation(
                    orgId,
                    policyId,
                    eventBody.timeOfAction,
                    projId
                );
            }
        } else {
            await scheduleRiskCalculation(
                orgId,
                policyId,
                eventBody.timeOfAction,
                getProjectId(eventBody)
            );
        }
    } catch (err) {
        logCatchError('handlePolicyActionEvent', err.message);
        logger.log(LOG_LEVELS.DEBUG, 'Debug: Event body with issue is:');
        logger.log(LOG_LEVELS.DEBUG, eventBody);
    }
};

async function scheduleRiskCalculation(
    orgId,
    policyId,
    timeOfAction,
    projectId = null
) {
    const accessHeaders = {
        Authorization: `Bearer ${await getAccessToken()}`,
        ['Content-Type']: 'application/json'
    };
    const result = await httpRequest2(
        HTTP_VERBS.POST,
        `${process.env.RESOURCE_HOST}:${process.env.RESOURCE_PORT}/api/v3/schedule-process`,
        { orgId },
        accessHeaders,
        undefined,
        {
            type: SCHEDULED_PROCESS_TYPES.POLICY_RISK_CALCULATION,
            scheduled_date: timeOfAction,
            context: {
                project_id: projectId,
                policy_id: policyId
            }
        }
    );
    if (result.status === 'rejected') {
        throw new Error(
            'Failed to successfully schedule policy risk calculation'
        );
    }
}

async function calculatePolicyRisk(calcBody) {
    const policyId = calcBody.policyId;
    const projectId = getProjectId(calcBody);
    const orgId = calcBody.orgId;
    try {
        const policyRiskTrendBody = {
            policy_id: policyId,
            project_id: null
        };
        if (projectId) {
            policyRiskTrendBody.project_id = projectId;
        }

        const { risk } = await httpRequest2(
            HTTP_VERBS.POST,
            `${process.env.RESOURCE_HOST}:${process.env.RESOURCE_PORT}/api/v3/policies/risk-trends?orgId=${orgId}`,
            undefined,
            undefined,
            undefined,
            policyRiskTrendBody
        );
        logger.log(
            LOG_LEVELS.INFO,
            `The policy risk for id ${policyId} and project id ${projectId} is ${risk}`
        );
    } catch (err) {
        logCatchError('calculatePolicyRisk', err.message);
    }
}

function getProjectId(eventBody) {
    if (eventBody?.record?.project_id) {
        return eventBody?.record?.project_id;
    }
    if (eventBody?.projectId) {
        return eventBody.projectId;
    }
    return null;
}

此前ServiceBusConnString配置为Service Bus的URL（如testservicebus.servicebus.windows.net），该配置在函数未休眠时可用，但函数休眠后冷启动失效。将其切换为Service Bus的主连接字符串后，经过4次以上冷启动测试均正常工作。

内容的提问来源于stack exchange，提问作者TomMakes