Cloud Task 处理器内部多 Promise 失败的处理及重试方案咨询

核心改造思路 #

• 为每个异步步骤添加独立的错误标记，可精准定位失败环节
• 所有操作实现幂等性，多次重试不会产生数据错乱
• 收集全量执行结果，仅存在未完成操作时触发Cloud Task自动重试，已完成的操作重试时直接跳过

改造后代码 #

const cors = require('cors')({origin: true});
const admin = require('firebase-admin');
admin.initializeApp({ ... });

// 封装带步骤标记的错误
class StepError extends Error {
  constructor(step, originalError) {
    super(`Step ${step} failed: ${originalError.message}`);
    this.step = step;
    this.originalError = originalError;
  }
}

// 1. 幂等执行fan-out更新
async function runFanoutUpdate(rootRef, updates, postId) {
  try {
    // 先校验post是否还存在，不存在说明已经执行过该步骤
    const postSnap = await rootRef.child(`posts/${postId}`).get();
    if (!postSnap.exists()) return { success: true, step: 'fanout' };
    await rootRef.update(updates);
    return { success: true, step: 'fanout' };
  } catch (err) {
    return { success: false, step: 'fanout', error: new StepError('fanout', err) };
  }
}

// 2. 幂等执行用户postCt递减
async function runUserPostCtDecrement(userRef, postId) {
  try {
    // 校验post是否还存在，不存在说明已经执行过删除流程，无需再减
    const postSnap = await admin.database().ref(`posts/${postId}`).get();
    if (!postSnap.exists()) return { success: true, step: 'userPostCt' };
    // 用transaction保证递减的原子性，就算重复执行也只会在post存在时减一次
    await userRef.transaction(current => {
      if (current === null) return 0;
      return current - 1;
    });
    return { success: true, step: 'userPostCt' };
  } catch (err) {
    return { success: false, step: 'userPostCt', error: new StepError('userPostCt', err) };
  }
}

// 3. 幂等执行全局postCt递减
async function runTotalPostCtDecrement(totalRef, postId) {
  try {
    const postSnap = await admin.database().ref(`posts/${postId}`).get();
    if (!postSnap.exists()) return { success: true, step: 'totalPostCt' };
    await totalRef.transaction(current => {
      if (current === null) return 0;
      return current - 1;
    });
    return { success: true, step: 'totalPostCt' };
  } catch (err) {
    return { success: false, step: 'totalPostCt', error: new StepError('totalPostCt', err) };
  }
}

// 4. 幂等删除视频文件
async function deleteVideo(bucket, videoPath) {
  try {
    const file = bucket.file(videoPath);
    const [exists] = await file.exists();
    if (!exists) return { success: true, step: 'deleteVideo' };
    await file.delete();
    return { success: true, step: 'deleteVideo' };
  } catch (err) {
    return { success: false, step: 'deleteVideo', error: new StepError('deleteVideo', err) };
  }
}

// 5. 幂等删除缩略图文件
async function deleteThumbnail(bucket, thumbnailPath) {
  try {
    const file = bucket.file(thumbnailPath);
    const [exists] = await file.exists();
    if (!exists) return { success: true, step: 'deleteThumbnail' };
    await file.delete();
    return { success: true, step: 'deleteThumbnail' };
  } catch (err) {
    return { success: false, step: 'deleteThumbnail', error: new StepError('deleteThumbnail', err) };
  }
}

exports.fireCloudTaskAtSpecificTime = functions.https.onRequest((request, response) => {
  return cors(request, response, async () => {
    try {
      const userId = request.body.userId;
      const postId = request.body.postId;
      const videoId = request.body.videoId;
      const thumbnailId = request.body.thumbnailId;

      const updates = {};
      updates[`/posts/${postId}`] = null;
      updates[`/user-postIds/${userId}/${postId}`] = null;

      const userPostCtRef = admin.database().ref('users').child(userId).child('postCt');
      const totalPostCtRef = admin.database().ref('total-posts-count').child('postCt');
      const videoPath = `videos/${userId}/${videoId}`;
      const thumbnailPath = `thumbnails/${userId}/${thumbnailId}`;
      const bucket = admin.storage().bucket();

      const postSnap = await admin.database().ref('posts').child(postId).get();
      if (!postSnap.exists()) {
        return response.status(200).send('Ok - Post No Longer Exists');
      }

      // 按顺序执行所有步骤，收集执行结果
      const results = [];
      results.push(await runFanoutUpdate(admin.database().ref(), updates, postId));
      results.push(await runUserPostCtDecrement(userPostCtRef, postId));
      results.push(await runTotalPostCtDecrement(totalPostCtRef, postId));
      results.push(await deleteVideo(bucket, videoPath));
      results.push(await deleteThumbnail(bucket, thumbnailPath));

      // 筛选失败的步骤
      const failedSteps = results.filter(res => !res.success);
      if (failedSteps.length > 0) {
        // 打印失败步骤方便排查
        console.error('Failed steps: ', failedSteps.map(f => f.step));
        // 返回非2xx状态码触发Cloud Task重试
        return response.status(500).send(`Failed steps: ${failedSteps.map(f => f.step).join(',')}`);
      }

      return response.status(200).send('Ok - Everything Deleted');
    } catch (err) {
      console.error('Global error: ', err);
      return response.status(500).send('Task failed');
    }
  });
});

关键逻辑说明 #

• 错误定位：每个步骤抛出的错误都携带step字段，日志里可以直接看到失败的环节，不需要排查堆栈
• 幂等性保证：每个步骤执行前先校验前置条件，已经完成的操作不会重复执行，就算多次重试也不会出现计数器多减、重复删文件的问题
• 重试控制：仅当有步骤执行失败时才返回非2xx状态码，触发Cloud Task的自动重试，你可以在Cloud Task控制台配置重试次数、间隔，避免无限重试
• 无全有全无问题：单个步骤失败不会终止其他步骤的执行，能完成的操作都会先完成，只重试失败的部分

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